Broker Integration

Media-identity failure boundary

Capabilities v3 advertises wardian.media-identity.v2 input and wardian.media-results.v2 output. The request declares payload.payload.media_identity_contract_version; each media item is identified by the pair conversation.chat.id + message.id. The pair is carried through activities, artifacts and reducers as an explicit identity object.

Legacy requests remain accepted only when media message.id values are unambiguous across chats. legacy_media_identity_ambiguous, duplicate_media_identity, media_identity_v2_invalid and media_identity_result_mismatch are permanent contract failures: publish one terminal result with retryable=false, send the original input to the DLQ, and commit only after publication. No Temporal run or model request is started for an invalid input. Structured log events expose only the error code and envelope identifiers, never conversation content.

Identity-certainty failure boundary

The capabilities manifest advertises identity contracts in preference order: [wardian-core-identity.v2, wardian-core-identity.v1]. Deploy CIS with both before Core starts emitting V2.

The broker validates the V2 side contract both for inline requests and hydrated artifacts. identity_certainty_invalid is permanent: publish the original envelope to the flow-runs DLQ, publish one failed flow result, and commit the offset. Do not launch Temporal, retry a model, or silently coerce the request to V1. The failed result metadata is limited to the sanitized snapshot ACK; raw validation input is not copied into diagnostics.

Successful and ordinary failed V2 runs echo the same ACK from normalized input. Oversized-result compaction preserves it. Redelivery reuses the Core-owned request_id and frozen snapshot_ref; CIS never refreshes identity or invents a new generation.

CIS consumes flow run requests and publishes flow run results.

Topics

Direction	Topic	Contract
consume	content-intelligence.flow-runs.v1 through .v4	FlowRunEnvelope
publish	content-intelligence.flow-results.v1 through .v4	FlowResultEnvelope
publish diagnostics	content-intelligence.flow-diagnostics.v1	FlowDiagnosticEnvelope
publish dev preview	content-intelligence.prompt-lab-results.v1	PromptLabPreviewResultEnvelope
publish invalid input	content-intelligence.flow-runs.dlq.v1	diagnostic object

Flow

sequenceDiagram
  participant Kafka as Kafka
  participant Broker as BrokerWorkerService
  participant Launcher as WorkflowLauncher
  participant Temporal as Temporal
  participant Store as Private Artifact Store

  Kafka->>Broker: FlowRunEnvelope
  Broker->>Broker: validate contract version and payload
  Broker->>Broker: normalize flow run request (exactly once)
  Broker->>Broker: negotiate and attest effective privacy
  alt privacy requirement cannot be met
    Broker->>Kafka: non-retryable v4 rejection; no Temporal start
  else privacy accepted or legacy-unnegotiated
  Broker->>Launcher: start_analysis(normalized FlowRunRequest)
  Launcher->>Launcher: effective payload-size preflight
  alt normal or ineligible request
    Launcher->>Temporal: ProcessConversationChunk
  else oversized eligible Daily
    Launcher->>Store: deterministic manifest + partitions
    Launcher->>Temporal: ProcessSplitConversationReport(manifest_ref)
    Temporal->>Store: partition results + internal final result
  end
  Broker->>Launcher: poll get_run(run_id)
  Launcher-->>Broker: completed or failed
  Broker->>Kafka: FlowResultEnvelope
  Broker->>Kafka: FlowDiagnosticEnvelope when degraded or failed
  end

During a rollout, SIGTERM or the shared quiesce marker stops acquisition of the next Kafka record. A record already handed to BrokerWorkerService is allowed to finish its Temporal wait, publish its result/diagnostic, and commit before the consumer and producer close. BROKER_GRACEFUL_SHUTDOWN_SECONDS and Compose stop grace must cover the largest normal/split result timeout plus 60 seconds. If that relationship or the read-only marker mount cannot be proven from the rendered Compose configuration, deploy aborts before runtime mutation.

When CP Prompt Lab publishes a flow-run envelope with Kafka header x-cis-prompt-lab-mode: preview-only, the broker validates and hydrates the same FlowRunEnvelope as the live flow, including payload_ref, but does not start Temporal or call models. It builds the conversation documents and runtime model request preview, then publishes a PromptLabPreviewResultEnvelope to the prompt-lab result topic. This mode is available only with DEV_MODE=true and ENABLE_CIS_PROMPT_LAB=true.

This broker preview remains the lightweight, no-workflow inspection path. The separate prepared-snapshot action in /dev/cp-prompt-lab deliberately launches one normal workflow with initial-ingestion-report-all-media; it then caches the resulting derived conversation documents locally for prompt iteration. It does not add a Kafka contract, change live flow envelopes, or alter product result contracts.

FlowRunEnvelope

{
  "contract_version": "content-intelligence.flow-runs.v1",
  "request_id": "request-1",
  "tenant_id": "tenant-1",
  "payload": {
    "flow": "daily-summary-and-notifications-report",
    "recipe": "daily-summary-and-notifications-report-text-only",
    "settings": {"output_language": "es"},
    "payload": {
      "source": "broker",
      "monitored": {"id": "minor-1", "name": "Felipe"},
      "metadata": {
        "window_start": "2026-06-15T00:00:00-03:00",
        "window_end": "2026-06-16T00:00:00-03:00"
      },
      "conversations": [
        {
          "chat": {"id": "chat-1", "name": "Sofi", "type": "direct"},
          "messages": [
            {
              "id": "m1",
              "timestamp": "2026-06-15T18:05:00-03:00",
              "sender": {"role": "contact", "name": "Sofi"},
              "type": "text",
              "text": "Hola"
            }
          ]
        }
      ]
    }
  },
  "trace_id": "trace-1"
}

FlowRunEnvelope carries exactly one of payload or payload_ref. payload.payload is the same FlowRunRequest.payload documented in _docs/PAYLOAD_CONTRACTS.md. When payload_ref is present, CIS downloads the referenced JSON artifact, verifies sha256, decompresses gzip, and then uses the same validation path as inline payloads. The optional payload.context.monitored_age_band side contract travels inside that existing request body for either transport. It does not change FlowRunEnvelope, its public version, topic routing or result envelopes. Roll it out consumer-first: deploy CIS support with its flag off, then enable the CIS consumer flag, and only then enable CP emission.

wardian-source-coverage.v2 remains nested at payload.payload.user_stats_metadata.source_coverage; it does not change content-intelligence.flow-runs.v1, topic names, broker headers or result envelopes. Inline and payload_ref requests apply the same planned-pause interval validation before Temporal/media/model execution.

conversation_memory.v2 is another nested Daily side contract under payload.payload.context.conversation_memory; it does not change the flow-run envelope or topic. CP sends it only after selecting the exact literal from capabilities v4. CIS validates the same closed schema for inline and hydrated requests before Temporal, including per-slot digests and dual legacy/v2 equivalence. In enforce mode its conversation_memory_ack is part of output, so inline results, result refs, production-minimal projections, content-trust and output digests all bind the ACK. The report-context seal binds the exact v2 input snapshot independently. Deploy the CP consumer first, then enable CIS observe/enforce and only then CP production. Queued N-1 jobs without the selection remain legacy.

For live flow retries, CIS first checks the deterministic workflow id. If that run already exists, it reuses the run and does not require the referenced artifact to remain available. Prompt Lab preview still hydrates the artifact because it does not start or reconcile a workflow.

Report auto-split does not change FlowRunEnvelope: CP publishes one complete request or one existing external payload_ref. Split manifests and partition refs live only inside CIS under flow-artifacts/report_split/. The feature requires the shared object store even if external Kafka result refs are disabled. A transient split-artifact error is raised without committing the offset; redelivery reuses the stable workflow/run identity. A permanent integrity or limit error completes as a failed run instead of producing a partial report.

New split artifacts use wardian-report-split.v2 with media_identity_version: 2; partition media results are ordered arrays of explicit identities. The runtime continues to read v1 manifests/results as legacy message-id maps, but rejects mixed v1/v2 partitions and conflicting duplicates instead of using last-write-wins.

FlowArtifactRef

Large flow-run payloads and completed flow results are stored in object storage and referenced from Kafka with this shape:

{
  "source": "object_store",
  "bucket": "wardian-flow-artifacts",
  "object_key": "flow-runs/daily/request-1.json.gz",
  "content_type": "application/json",
  "compression": "gzip",
  "size_bytes": 12345,
  "sha256": "hex-encoded-sha256",
  "artifact_kind": "flow_run_payload"
}

artifact_kind is flow_run_payload for FlowRunEnvelope.payload_ref and flow_result for FlowResultEnvelope.result_ref. Permanent artifact failures such as an invalid checksum go to DLQ and also publish a terminal failed FlowResultEnvelope with error: "invalid_flow_run_artifact", empty run_id, and metadata.retryable=false; this lets CP close the published request even though no Temporal run exists. CIS commits only after that result publish succeeds. Transient object-storage failures are retried by raising from the broker loop without committing the Kafka offset.

FlowResultEnvelope

{
  "contract_version": "content-intelligence.flow-results.v1",
  "request_id": "request-1",
  "tenant_id": "tenant-1",
  "run_id": "uuid",
  "status": "completed",
  "result": {},
  "trace_id": "trace-1"
}

When the run fails, status is failed, result is omitted, and error contains the sanitized reason. Optional media failures do not make this envelope failed; they publish a completed result with degraded metadata and placeholders inside media_processed.

Negotiated safe failures

Every v1-v5 FlowResultEnvelope parser accepts an optional additive failure block:

{
  "contract_version": "wardian.safe-failure.v1",
  "reason_code": "processing_failed",
  "safe_message_key": "cis.failure.processing_failed",
  "retry_scope": "unknown",
  "suggested_action": "unknown"
}

The fields are bounded machine keys. suggested_action is exactly one of new_request, not_retryable, or unknown; CIS never emits same_request or recover_identity. The optional diagnostic_ref is an opaque machine key and cannot contain diagnostic text. failure is forbidden on completed envelopes.

Emission has two independent gates. The runtime must use SAFE_FAILURE_ENVELOPE_MODE=emit, and the v2, v4, or v5 input must freeze result_delivery.safe_failure_contract="wardian.safe-failure.v1". off (default) performs no classification. observe classifies and emits structured measurements but serializes the legacy envelope. Missing consumer negotiation also serializes the legacy envelope even in emit mode. Invalid mode values fail settings validation.

Classification is deterministic and never copies the original error text. Request-contract failures map to a new-request decision; policy rejections map to not-retryable; timeouts and unknown processing failures remain unknown. Correctable privacy request failures (stale capabilities, missing privacy request, policy-version mismatch, or unmasked protected input) are request contract failures and therefore require a new request rather than being marked not-retryable. result_transport_failed is emitted only after CIS has local evidence of a failed result publication path (for example, the first Kafka payload exceeded the broker limit and the compact terminal retry can still be published) or an explicit safe diagnostic reference. If the terminal result itself cannot be published, CIS cannot manufacture that evidence inside the missing result.

The main terminal is built once. Best-effort diagnostics published afterward do not mutate its failure, even if a late diagnostic reports a different transport outcome.

Malformed inline v2 payload contracts, including invalid negotiated conversation_memory.v2, are routed through this terminal request-rejection path when request and tenant scope can be recovered; the broker also records the invalid payload in the DLQ and does not leave Control Plane waiting for a result.

media_processed entries may also carry an internal object-store reference when one Temporal activity produced an oversized derived result. The artifact kind is media_processed; source URLs, media bytes, data URLs, and raw model request/response debug are excluded before upload. This nested reference does not replace FlowResultEnvelope.result_ref and does not change the external flow result contract.

Prompt Lab snapshot preparation also uses an internal object-store reference with artifact_kind="prompt_lab_preparation" and the prompt_lab_preparation/ object-key prefix. The preparation activity always returns this bounded ref to Temporal; the Prompt Lab preparer resolves it before returning prepared conversation documents and initial-report request previews. The preparer accepts refs only from the configured artifact bucket and dedicated object-key prefix, then deletes the read-once transport artifact after successful or invalid-payload handoff. Cleanup failure is terminal and explicit so derived plaintext is not silently retained. This ref is internal and does not change the Kafka broker contracts.

For completed envelopes, CIS publishes exactly one of result or result_ref. result, when present, is the FlowRunResult documented in _docs/OUTPUT_CONTRACTS.md. When result_ref is present, CP resolves the artifact and persists the same output it would have read from the inline result. CIS only falls back to a compact failed envelope with error: "kafka_message_too_large" if artifact upload or the compact result-ref publish path fails.

For split execution, Temporal itself receives only a small internal final flow_result ref. The launcher validates and hydrates it before the broker builds the same public FlowResultEnvelope path above. This internal ref is independent of BROKER_KAFKA_RESULT_REFS_ENABLED and never appears as a new field in the public result contract.

FlowDiagnosticEnvelope

content-intelligence.flow-diagnostics.v1 carries operational diagnostics for valid flow runs. It is not a DLQ and is published best-effort after the main result envelope.

{
  "contract_version": "content-intelligence.flow-diagnostics.v1",
  "event_type": "media_activity_failed",
  "severity": "warning",
  "request_id": "request-1",
  "tenant_id": "tenant-1",
  "run_id": "uuid",
  "flow": "daily-summary-and-notifications-report",
  "recipe": "daily-summary-and-notifications-report-all-media",
  "message_id": "m1",
  "media_type": "image",
  "activity": "analyze_image",
  "error_code": "provider_400",
  "error": "provider_400:fileUri parameter rejected data:<redacted>;base64,<redacted>",
  "trace_id": "trace-1"
}

flow_run_failed diagnostics use severity="error" and describe critical run failures that already produced a failed FlowResultEnvelope.

Both error and explicit activity-provided error_code values are sanitized before publication, including bearer/basic credentials, password/database URL assignments and URLs containing userinfo. Media-diagnostic metadata is limited to operational frame/link indices. Media types intentionally disabled by a recipe or request-level settings.media_policy do not emit media_activity_failed: no media activity was attempted, so the omission is not a degradation event.

PromptLabPreviewResultEnvelope

{
  "contract_version": "content-intelligence.prompt-lab-preview.v1",
  "request_id": "request-1",
  "tenant_id": "tenant-1",
  "source_topic": "content-intelligence.flow-runs.daily.v1",
  "status": "completed",
  "preview": {
    "status": "ai_request_preview",
    "conversation_documents": [],
    "requests": []
  },
  "trace_id": "trace-1"
}

When prompt construction fails, status is failed, preview is omitted, and error contains the reason. Preview envelopes can include rendered conversation text and prompts, so this topic is dev-only and should have short retention. When BROKER_KAFKA_RESULT_REFS_ENABLED=true and a completed preview exceeds the inline Kafka limit, CIS stores the preview in FLOW_ARTIFACTS_BUCKET and publishes preview_ref with artifact_kind="prompt_lab_preview". The dev API resolves preview_ref server-side before returning the preview to the browser. If refs are disabled or artifact storage cannot be used for an oversized preview, CIS publishes the compact kafka_message_too_large failed envelope.

Runtime Settings

CI_ENABLE_BATCH_INGRESS=true
CONTENT_INTELLIGENCE_FLOW_RUNS_TOPIC=content-intelligence.flow-runs.v1
CONTENT_INTELLIGENCE_FLOW_RUNS_TOPICS=
CONTENT_INTELLIGENCE_FLOW_RESULTS_TOPIC=content-intelligence.flow-results.v1
CONTENT_INTELLIGENCE_FLOW_DIAGNOSTICS_TOPIC=content-intelligence.flow-diagnostics.v1
CONTENT_INTELLIGENCE_PROMPT_LAB_RESULTS_TOPIC=content-intelligence.prompt-lab-results.v1
CONTENT_INTELLIGENCE_FLOW_RUNS_DLQ_TOPIC=content-intelligence.flow-runs.dlq.v1
CONTENT_INTELLIGENCE_FLOW_RUNS_CONSUMER_GROUP=wardian-cis-flow-runs
BROKER_RESULT_TIMEOUT_SECONDS=900
BROKER_SPLIT_RESULT_TIMEOUT_SECONDS=900
BROKER_KAFKA_MAX_REQUEST_SIZE_BYTES=2097152
BROKER_KAFKA_MAX_POLL_INTERVAL_MS=1800000
BROKER_KAFKA_MAX_POLL_RECORDS=1
BROKER_KAFKA_CONSUMER_REQUEST_TIMEOUT_MS=60000
BROKER_KAFKA_SESSION_TIMEOUT_MS=45000
BROKER_KAFKA_HEARTBEAT_INTERVAL_MS=10000
BROKER_KAFKA_RETRY_BACKOFF_MS=500
BROKER_KAFKA_CONSUMER_RESTART_MAX_BACKOFF_SECONDS=30
BROKER_KAFKA_METADATA_MAX_AGE_MS=300000
BROKER_KAFKA_RESULT_REFS_ENABLED=false
BROKER_KAFKA_INLINE_RESULT_MAX_BYTES=786432
FLOW_ARTIFACTS_BUCKET=wardian-flow-artifacts
FLOW_ARTIFACT_MAX_BYTES=8388608
FLOW_ARTIFACT_MAX_DECODED_BYTES=33554432
FLOW_RESULT_EVIDENCE_ENABLED=false
FLOW_RESULT_EVIDENCE_STORAGE_POLICY_VERIFIED=false
FLOW_RESULT_EVIDENCE_TTL_SECONDS=86400
FLOW_RESULT_EVIDENCE_RETENTION_SECONDS=172800
FLOW_RESULT_EVIDENCE_UPLOAD_GRACE_SECONDS=900
FLOW_RESULT_EVIDENCE_UPLOAD_MAX_ATTEMPTS=3
SAFE_FAILURE_ENVELOPE_MODE=off

CONTENT_INTELLIGENCE_FLOW_RUNS_TOPIC can be a comma-separated list. If CONTENT_INTELLIGENCE_FLOW_RUNS_TOPICS is set, it takes precedence.

Use BROKER_KAFKA_MAX_POLL_INTERVAL_MS and BROKER_KAFKA_MAX_POLL_RECORDS to bound broker consumer concurrency and keep Kafka group membership stable during long flow runs. Alpha runs with one record per poll. Before auto-split is enabled, max poll must be strictly greater than the split polling window; the defaults are 1800000 ms and 900 seconds respectively.

Consumer membership emits structured kafka_consumer_group_joined, kafka_consumer_partitions_revoked, and kafka_consumer_group_rejoined events; the rejoin event includes rebalance_duration_ms. Failures surfaced by aiokafka are classified as kafka_heartbeat_expired, kafka_coordinator_lost, kafka_transport_timeout, kafka_transport_unavailable, or the fallback kafka_consumer_loop_failed. Transient transport, coordinator, and heartbeat failures recreate the consumer in-process with exponential backoff capped at 30 seconds; application failures still terminate for supervisor visibility. kafka_consumer_restart_scheduled records the attempt and delay without message payloads. The intentional broker_split_result_pending path still exits for idempotent supervisor redelivery.

Same ID, same intent (v3)

Sealed report context

Control Plane may include payload.metadata.report_context_seal using wardian-report-context-seal.v1. The digest covers the effective monitored profile allowlist plus identity-certainty snapshot, guidance, feedback, memories, antecedents, processing policy and source coverage, binds the opaque snapshot ID, and includes Daily window or InterWeek period bounds. The profile allowlist excludes DOB, phone, JID and raw profile fields. CIS validates component and aggregate SHA-256 digests before Temporal or model work, records only the version, snapshot ID and digest in the Temporal memo, and echoes the verified seal at flow-result.metadata.report_context_seal.

The seal is optional for N-1 jobs. Once present it fails closed: malformed metadata, an allowlist violation or digest drift produces a terminal non-retryable result without starting or returning a cached workflow. The intent fingerprint remains the authenticated binding for the stable v3 request. It excludes only the verified publication-scoped seal metadata so an observe publication can be retried under enforce; the context seal is the immutable identity of the report context. Referenced payloads are hydrated before legacy cache lookup so their seal cannot bypass validation. The verified digest and snapshot ID are persisted with the in-memory run and Temporal memo; redelivery may reuse a cached result only when that binding matches exactly. Direct HTTP and internal launcher callers execute the same verification before repository claims or Temporal start.

content-intelligence.flow-runs.v3 requires a tenant-scoped, HMAC-SHA256 intent_fingerprint. For v3, CIS authorizes and hydrates payload_ref, validates and normalizes the request, verifies the fingerprint, and only then looks up an idempotent run. Therefore the same logical request deduplicates identically when transported inline or by reference. Signed URL query/fragment volatility is not part of intent, but the media object scheme/host/path and identity-bearing query parameters are. Canonicalization removes only the documented volatile signing parameters (sig, signature, expiry/token/auth/policy keys, x-amz-*, and x-goog-*); it never drops the full query string. URL scheme and hostname are case-normalized, while case-sensitive userinfo credentials are preserved. This signed-URL projection applies only to ConversationMessage.media_url; arbitrary context or metadata fields with the same key remain byte-material.

The fingerprint uses the same compacted request that enters Temporal. WAHA raw_info is first projected into the stable identity fields consumed by the runtime and is then excluded, so operational redelivery metadata cannot create false conflicts. Transport-only payload.metadata.request_id and payload.metadata.delivery_after are removed from that shared runtime request, not only from the digest. CIS-owned settings.privacy_decision is also excluded: it is attached only after producer intent verification and must not change a v3 identity during N/N-1 key rotation. Conversation array order is preserved because it also determines runtime document and prompt order; changing that order is a material intent change. Message order follows the same schema normalization used by the runtime. Canonical JSON preserves runtime key and string composition byte-materially, plus cross-language number rendering (0.000001, not Python-specific 1e-06 at that boundary); object keys are ordered by UTF-16 code units to match the CP/JCS producer. Non-finite values and normalized-key collisions are terminal typed rejections.

The Temporal memo stores fingerprint version, algorithm, key ID, and complete digest. After USE_EXISTING, the launcher describes the selected execution and checks that memo again, closing the race between a broker lookup and workflow start. A mismatch returns content-intelligence.flow-results.v3 with typed error idempotency_intent_conflict, retryable=false, no result/result_ref, and no cached run ID. CIS never executes a model or republishes the old result on that path. Logs contain only version, key ID and a 12-character HMAC prefix. Partial, malformed, or unreadable v3 memo fields also fail closed; only a memo where every v3 fingerprint field is absent can follow the configured legacy-run policy. A rolling-deploy launcher that cannot persist tenant-scoped fingerprint metadata is rejected before workflow start. Contract-validation failures still echo a valid v3 fingerprint and use flow-results.v3 for correlation.

Canonicalization and the coordinated CP/CIS rollout are specified in CP's control-plane/docs/CIS_IDEMPOTENCY_INTENT.md. Canonicalization v1 is immutable; changes after CP begins v3 emission require a new version and a cross-language golden test. CP's producer canonicalizer and golden vectors must match these rules before CIS_FLOW_INTENT_V3_EMIT_ENABLED is enabled.

Requested privacy, effective privacy (v4)

content-intelligence.flow-runs.v4 is the strict per-job privacy contract. CP freezes the capabilities-v2 digest, policy version, requested mode, and input protection mode before publication. CIS hydrates and normalizes the request, then admits it only when that frozen request is compatible with the effective runtime privacy capability.

Requested	Effective CIS assurance	Input protection	Decision
protected	protected	masked	accepted_exact
protected	debug_plaintext	any	reject privacy_requirement_unsatisfied
protected	degraded	any	reject privacy_assurance_degraded
protected	protected	not masked	reject privacy_input_not_masked
debug_plaintext	debug_plaintext/degraded	any	accepted_exact
debug_plaintext	protected	any	accepted_effective_stricter

V4 also rejects a missing privacy request, stale capabilities digest, or policy version mismatch. Rejections publish flow-results.v4 with a typed privacy_decision, are non-retryable, and do not launch Temporal. Completed v4 results repeat the accepted decision whether the result is inline or behind a signed result ref. Logs contain only enum/version/digest and correlation fields; they never log request content, PII, or raw model I/O for this negotiation.

The accepted decision is propagated to text and media activities. Before a worker constructs prompts or model/media clients, it derives its local assurance from its own settings and requires the policy version and effective mode to match the ingress decision. A mismatched pool fails the run with privacy_worker_profile_mismatch; it cannot continue under a weaker global profile. Legacy v1-v3 activities remain compatible because their decision is absent or legacy_unnegotiated.

Although v4 forbids a producer-supplied intent fingerprint, CIS derives an internal, domain-separated digest from the normalized producer-owned request and stores it with the run/Temporal memo. A repeated tenant/request ID must match that digest before CIS can reuse an existing execution. This internal binding is never accepted from CP and is never emitted in flow-results.v4.

V1-v3 remain available for N/N-1 compatibility and are recorded as legacy_unnegotiated; absence of a v4 decision never proves protected handling. CP must persist its frozen selection and the returned decision for retries and manual replays. Observe mode may persist a projection while continuing to emit the legacy wire contract; enforce mode emits v4 and fails closed.

V4 redeliveries check the existing run and its internal intent binding before re-negotiating current capabilities. A matching cached or in-flight run reuses the original decision stored in the run cache and Temporal memo, so a manifest change cannot turn a successful retry into a stale-capability rejection. The new manifest still applies to every new run.

Conversations are data (v5)

content-intelligence.flow-runs.v5 composes the v4 privacy negotiation with the v3 artifact and producer-owned fingerprint semantics, and adds content_trust_policy:

{
  "contract_version": "content-intelligence.flow-runs.v5",
  "content_trust_policy": {
    "contract_version": "wardian.content-trust-policy.v1",
    "policy_version": "conversations-are-data.v1",
    "mode": "observe|enforce"
  }
}

The identical policy is required at payload.settings.content_trust_policy, so it is covered by the v3-compatible intent fingerprint. Inline requests validate that equality immediately; payload_ref requests validate it after authorized hydration and before intent verification or Temporal lookup. V1-v4 forbid either policy field. Direct HTTP and dev-tool requests also reject settings.content_trust_policy; only the verified broker v5 transport may supply it to the runner.

All flow user prompts are canonical untrusted-data envelopes. System prompts state that dialogue, metadata, OCR, transcripts, document/link extraction, video text, memories and prior reports can supply evidence but cannot change instructions or schemas. The model gateway validates each typed step output. When enabled, one independent strict-JSON evaluator checks the final typed output against the complete deduplicated set of bounded user prompts that actually shaped the selected flow output, including guidance, feedback, statistics, memories and prior reports. If that combined evaluator input exceeds its configured bound, the assessment becomes review; CIS never silently drops one of those evidence sources. The evaluator returns no free text or copied snippets.

Completed v5 results use content-intelligence.flow-results.v5, echo the fingerprint and require result.content_trust_assessment whether the result was inline or hydrated from result_ref. Before publication CIS pins assessment mode to the requested policy. Evaluator outage, invalid response, oversize input, or missing evaluator configuration produces review; observe reports that decision without withholding delivery, while enforce fails closed; deterministic supported no-activity outputs are the sole no-evaluator pass. Inline v5 envelopes whose outer policy does not match the policy inside the signed payload publish a terminal v5 invalid_flow_run result (and DLQ evidence) instead of leaving Control Plane waiting for a result that can never execute. Schema-valid evaluator payloads with a pass decision and any non-pass reason code are likewise classified as invalid and downgraded to review. The output hash and typed state are consumer-verifiable. Contracts v1-v4 remain unchanged for N-1 drains. V5 completed results also carry the v4-compatible privacy_decision; the rollout never reinterprets the published v4 contract.

Manual offset commits remain after result publication. A commit lost to a rebalance emits flow_run_commit_lost_after_publish; redelivery reuses the deterministic Temporal workflow id and therefore does not repeat model fan-out.

BROKER_RESULT_TIMEOUT_SECONDS (default 900) bounds ordinary result polling and remains terminal for non-auto-split flows. For an allowlisted Daily while the feature is enabled, BROKER_SPLIT_RESULT_TIMEOUT_SECONDS (default 900) is a non-terminal polling window. If Temporal is still running, CIS logs broker_split_result_pending, publishes no result/diagnostic, leaves the offset uncommitted and exits the consumer so its supervisor restarts it. Redelivery reattaches to the deterministic workflow id and resumes polling without repeating media or model work. The configured Temporal workflow execution timeout is the eventual terminal boundary.

For v1/v2 compatibility traffic, before hydrating either an inline payload or payload_ref, a non-preview redelivery may check the deterministic run id against the repository/Temporal source of truth. A terminal existing legacy run can publish directly without hydrating the original artifact. Contract v3 never uses that shortcut: it must hydrate and verify logical intent first. Signed v2 artifact capabilities are still authorized against tenant/request scope before any lookup to prevent cross-tenant disclosure; that check performs no object-store read. Legacy refs are validated only when hydration is required. Only an unresolved accepted pre-start claim re-enters the launcher: payload_ref is hydrated first so the actual flow selects the polling policy, and Temporal USE_EXISTING creates or joins the deterministic execution. This may repeat deterministic preflight and content-addressed artifact writes before Temporal owns the workflow id, but does not repeat media or model work afterward. Newly launched Daily runs still use the ordinary terminal timeout while auto-split is disabled. The two identifiers are serialized as a JSON pair before UUIDv5 hashing, so separator characters inside either value cannot make distinct requests collide. Transient Temporal start RPC failures remove the non-terminal cache claim; a Kafka redelivery retries the same stable id and attaches to the workflow if the first RPC actually reached Temporal.

For the first deployment of deterministic ids, drain or explicitly cancel pre-release workflows before enabling the new broker. Those older executions used random workflow ids and cannot be discovered from request_id, so an old run and its first deterministic retry could otherwise overlap once during the cutover.

Deploy in compatibility order: first deploy CIS that accepts payload_ref, then enable CP to emit payload_ref; next deploy CP that accepts result_ref, then set BROKER_KAFKA_RESULT_REFS_ENABLED=true for CIS brokers that should emit result_ref.

Auto-split has stricter startup guardrails. A batch-ingress worker requires BROKER_KAFKA_MAX_POLL_INTERVAL_MS to be greater than BROKER_SPLIT_RESULT_TIMEOUT_SECONDS * 1000 and, because it always publishes the terminal result, requires BROKER_KAFKA_RESULT_REFS_ENABLED=true. A separately flagged result publisher has the same result-ref requirement. Enable the feature only after CP result-ref hydration and artifact lifecycle controls are deployed. Deploy the new worker image everywhere with the feature flag off, then enable it in a second phase; this prevents mixed old/new workers from replaying payload-size version 2 with different commands.

For monitored age bands, keep the same consumer-first order: CIS code, CIS flag, CP code, CP emission flag. Rollback is flag-only and does not require changing the public broker envelopes.

Before publishing, Control Plane reads the authenticated GET /v1/internal/capabilities manifest described in CAPABILITIES.md, computes a contract intersection and persists the selected contracts with the job. This pre-publish gate complements, but never replaces, the validation performed when CIS consumes a flow run or CP consumes a result.

Flow-run v2 may add result_delivery.profile. The default/absent value remains legacy full. wardian.cis-result.production-minimal.v1 publishes output plus bounded lineage/usage/degradation summary and a separate evidence descriptor. The evidence upload happens after the durable Temporal result exists, so a Kafka redelivery reuses the same run and content-addressed object without repeating model calls. broker_result_evidence_pending keeps the offset uncommitted during the configured grace. After grace, outage degrades only the evidence descriptor; it never republishes debug collections inline. No second topic or dual result is used.

Validation

• Invalid JSON goes to DLQ with invalid_json.
• Unsupported contract version goes to DLQ with unsupported_contract_version:<value>.
• Pydantic validation failures go to DLQ with invalid_flow_run and details.
• Permanent payload_ref failures go to DLQ and publish a failed flow result

with error: "invalid_flow_run_artifact"; transient object-store failures remain retryable and do not commit the Kafka record.

• Valid payloads use the same normalization path as HTTP requests. Each ingress

(HTTP handler or broker) normalizes exactly once before start_analysis; launchers do not normalize again.

• Repeated valid envelopes with the same tenant_id and request_id return the

original run_id and never create a second Temporal execution.

• Valid runs with optional media failures publish FlowResultEnvelope.status=completed

plus diagnostic events.

• Valid runs with critical workflow failures publish FlowResultEnvelope.status=failed

plus a flow_run_failed diagnostic event.

Cancellation lane

Monitoring close uses a separate consumer group and topic so a long-running flow-run record cannot head-of-line block cancellation:

• commands: content-intelligence.flow-cancellations.v1;
• receipts: content-intelligence.flow-cancellation-receipts.v1;
• worker: python -m workers.cancellation_worker (Docker profile broker).

The command is authenticated before Temporal or object storage access. Signed scope fields with leading or trailing whitespace are rejected rather than normalized after signature verification, and a supplied run_id must equal the deterministic workflow id derived from the signed tenant/request pair. The worker cancels that workflow id, or starts the terminal CanceledFlowFence workflow when the flow has not started. It then deletes every identity-scoped v2 prefix for flow input/result, result evidence, workflow media and report-split artifacts before publishing a signed privacy-safe receipt. Duplicate commands republish the cached receipt; Temporal and object deletion remain idempotent after process restart.

The normal broker calls the Control Plane fence endpoint before artifact hydration, before workflow start, immediately after launcher preflight, before result publication, and again after an externalized result artifact finishes uploading. The post-preflight check repeats identity-scoped cleanup because report-split planning can write artifacts before Temporal starts; the post-upload check routes a concurrent cancellation through the same cleanup before Kafka publication. An unavailable fence endpoint is fail-closed and leaves the source offset uncommitted. A canceled flow publishes no normal result or diagnostic payload. If v2 artifact emission is disabled, cleanup returns partial without requiring an object-store client; the receipt does not claim complete deletion.