Local resource

Temporal

_docs/TEMPORAL.md

Temporal

The launcher selects ProcessConversationChunk for normal runs and ProcessSplitConversationReport for oversized eligible Daily reports.

Workflow

flowchart TD
  A["run_id + normalized FlowRunRequest"] --> P["payload-size preflight"]
  P -->|within inline budget| B["ProcessConversationChunk.run"]
  P -->|eligible Daily exceeds budget| S["persist deterministic split manifest"]
  S --> W["ProcessSplitConversationReport"]
  W --> X["bounded child workflows by conversation/message window"]
  X --> Y["recompose exact request + prompt budget gate"]
  Y --> H
  B --> C["effective media policy"]
  C --> D{"messages with enabled media?"}
  D -->|yes| E["fan out media activities concurrently to task queues"]
  D -->|no| F["run_flow_pipeline activity"]
  E --> G["media_processed identities, including activity_failed items"]
  G --> F
  F --> H["Recipe runner"]
  H --> I["FlowRunResult"]

Auto-split is enabled only when REPORT_TEMPORAL_AUTOSPLIT_ENABLED=true, the flow is allowlisted, and the normalized payload has conversations. V1 allows only daily-summary-and-notifications-report. The preflight measures the actual workflow envelope and final activity request. It chooses split mode when either the workflow input exceeds TEMPORAL_WORKFLOW_INPUT_SAFE_BYTES or the activity request exceeds REPORT_PARTITION_TARGET_BYTES.

The planner encodes canonical UTF-8 JSON and greedily keeps complete conversations together. A conversation that cannot fit alone is divided only at deterministic half-open message boundaries. Every partition records byte counts, original message offsets and SHA-256 digests. Media counts use the effective recipe/override policy, so text-only links or disabled media do not artificially force tiny partitions. The reducer rejects gaps, overlaps, duplicates, reordered content or digest changes, then reconstructs the exact normalized request before building one primary/shadow recipe request. It never returns a partial report and does not shard model generation. An indivisible request that remains below the real Temporal activity budget falls back to the inline workflow instead of failing only because the conservative partition target is lower.

Media activities are scheduled in deterministic bounded waves (fan-out with asyncio.gather) and their outcomes are fanned in as media_processed before run_flow_pipeline. V2 uses (chat_id, message_id) internally and explicit identity entries at workflow/activity/artifact boundaries; legacy unambiguous histories retain their message-id dict shape. MEDIA_MAX_ITEMS_PER_CHUNK (default 50) limits each concurrency wave; it does not discard later media. When one conversation is split, child workflows carry its original message-start offset so orden_chat remains identical to inline execution. Equal message ids in different chats schedule independent activities and retain both results. A duplicate complete identity or an activity result whose echoed identity differs from its input fails closed; no reducer uses last-write-wins.

Broker-launched runs derive a stable UUIDv5 workflow id from tenant_id plus request_id. A repeated Kafka delivery uses Temporal USE_EXISTING while the workflow is running and REJECT_DUPLICATE after it closes, so it returns the same run instead of executing media again. HTTP launches without an idempotency key keep independent random UUIDs.

Before workflow start, CIS projects repeated WAHA identity metadata into the canonical identity fields and removes per-message raw_info. Results larger than TEMPORAL_MEDIA_ACTIVITY_RESULT_SAFE_BYTES are reduced to the prompt projection; a sanitized audit copy can be stored as a media_processed artifact. Before run_flow_pipeline, the combined request is reduced through standard/strict/minimal projections until it fits TEMPORAL_ACTIVITY_INPUT_SAFE_BYTES. If the request itself cannot fit, the workflow returns temporal_activity_input_too_large without scheduling an invalid activity. Raising Temporal/gRPC limits is not the primary remedy.

The broker lifecycle is part of the same safety boundary. For allowlisted Daily runs, BROKER_SPLIT_RESULT_TIMEOUT_SECONDS is a non-terminal polling window: while Temporal is running, CIS publishes nothing, leaves the offset uncommitted and relies on idempotent supervisor redelivery. Kafka's max poll interval must exceed that window, and batch/result-publishing workers must enable BROKER_KAFKA_RESULT_REFS_ENABLED.

For flow-run contract v3, memo also carries wardian_intent_version, wardian_intent_alg, wardian_intent_key_id, and wardian_intent_digest. The launcher validates these fields after USE_EXISTING returns, not only before start, so concurrent consumers cannot join a workflow created for another intent. Pre-v3 histories have no such memo and are governed by FLOW_INTENT_LEGACY_RUN_POLICY during the drain window. Temporal's configured execution timeout is the actual terminal deadline.

Payload-size version 2 is also flag-gated. Roll out the new image to every worker with REPORT_TEMPORAL_AUTOSPLIT_ENABLED=false; only after old workers are gone may the flag be enabled. Rollback disables new v2 launches but keeps new workers alive until existing v2/split histories drain. An existing allowlisted Daily discovered on broker redelivery keeps non-terminal split polling during that drain even though the launch flag is already off.

Media identity follows the same replay rule. New launchers snapshot media_identity_version=2, and new split manifests use wardian-report-split.v2. Missing workflow fields and v1 manifests select the legacy branch deterministically. Rollback stops new v2 launches but keeps the v2-capable workers running until existing inline and split histories drain.

All Temporal worker entrypoints pass an explicit SDK graceful shutdown timeout. On SIGTERM/SIGINT the worker stops polling and drains the task already in process. New containers wait at the shared quiesce startup gate before connecting and polling, so they cannot create new activity during the replacement barrier. The deploy keeps the old workers alive until Temporal Visibility returns stable zero for every running workflow type; inability to prove zero stops all CIS consumers fail-closed.

CP Prompt Lab Pre-model Workflow

ProcessPromptLabSnapshot is a separate dev-only workflow. It is available to the Prompt Lab service only when both DEV_MODE=true and ENABLE_CIS_PROMPT_LAB=true; it does not replace or modify ProcessConversationChunk.

flowchart TD
  A["reviewed CP envelope"] --> B["force initial-ingestion-report-all-media"]
  B --> C["enable all seven media policies"]
  C --> D["bounded media fan-out and fan-in"]
  D --> E["prepare_prompt_lab_snapshot_inputs"]
  E --> F["conversation documents + derived media + initial-report request preview"]
  F --> G["return inline or through bounded transport artifact"]
  G --> H["append-only Prompt Lab repository"]
  H --> I["stop with report_model_calls=0"]

The workflow runs the production normalization, media activities, media hydration, and conversation builder. It deliberately does not call run_flow_pipeline, the recipe runner, or the Initial Report model. Optional media failures produce a usable degraded snapshot with bounded issue codes; a preparation failure produces a terminal failed snapshot.

Workflow IDs are prompt-lab-prepare-{preparation_id} and use WorkflowIDReusePolicy.REJECT_DUPLICATE. A deterministic bootstrap retry attaches to the existing active or completed workflow instead of starting media processing again. Before the launch, CIS writes a non-sensitive durable recovery descriptor and then its separately deletable private recovery-input object. This order cannot strand sensitive input when descriptor publication fails. Reconciliation first resumes the deterministic workflow by ID. Only when Temporal returns NOT_FOUND does it use the private recovery input to rebuild the context artifact and start the missing workflow. This avoids re-uploading a large request after the workflow already completed. It then deletes the descriptor and input after terminal snapshot publication and temporary-artifact cleanup. This loop continues during the API process lifetime, so a workflow that completes after a client polling timeout is published without requiring a restart or a second media run. Prompt Lab retries transient context/prepared-artifact object-store operations in-process using the report-split artifact retry budget. If those attempts are exhausted, both the request path and reconciliation keep the snapshot pending and retain its recovery input; a later reconciliation pass retries instead of publishing a terminal failed snapshot. Recoveries run independently, so one still-active workflow does not block completed workflows from finalizing and releasing their transport artifacts. S3-backed recovery and finalization scans run outside the API event loop and use a five-second idle interval.

Before workflow launch, the normalized request, prompt content, model parameters, monitored-age rendering flag, and versioned user-prompt renderer are pinned in the private preparation-context object. The workflow input carries only a deterministic media projection. It omits non-media messages and non-processing metadata while retaining each media message's original per-chat ordinal. If reusable media-insight bindings would exceed the workflow input blob budget, the projection is transported as deterministic gzip/base64 JSON. Only if that compressed representation still exceeds the safe budget are the optional bindings omitted and the media processed from its original URL. Activity retries never reread mutable prompt or model configuration and fail closed on an incompatible renderer. Existing inline workflow inputs remain replay-compatible. Client waits are bounded by CIS_PROMPT_LAB_RESULT_POLL_TIMEOUT_SECONDS without cancelling the durable workflow. The client observes status with describe() and only reads the result after a terminal state, so an HTTP polling timeout cannot propagate a cancellation request to Temporal. The workflow itself is bounded by CIS_PROMPT_LAB_WORKFLOW_EXECUTION_TIMEOUT_SECONDS, including time spent waiting for an activity poller. Temporal input and activity payloads use the existing safe byte limits. If the normalized request plus derived media does not fit the final text-activity input budget, the workflow omits the inline request and passes its private preparation-context reference instead. That temporary object pins both the normalized request and prompt-rendering context; the text activity validates the reference and hydrates the request. Existing inline inputs and older context objects remain readable. If prepared inputs do not fit the activity result budget, the text activity writes a bounded transport object below flow-artifacts/prompt_lab_preparation/prompt_lab_preparation/; the caller validates its bucket, prefix, kind, schema, and preparation ID, but retains it until the dedicated Prompt Lab snapshot has been published durably. The service then acknowledges the preparation and deletes the transport object with bounded artifact I/O retries and backoff. Startup reconciliation performs that acknowledgement for a terminal snapshot left behind by an interrupted process. Cleanup failure after retry exhaustion is fail-closed. Consequently, Temporal Prompt Lab startup requires the Flow Artifact S3 endpoint, credentials, and bucket even when its durable repository uses the local filesystem. This transport object is not the durable Prompt Lab dataset; the dedicated Prompt Lab repository becomes authoritative only after durable publication. When recovery starts a genuinely missing workflow, it retains the replacement context object it derived. Resuming an existing workflow reuses the workflow's original context reference without another upload. Terminal acknowledgement or failed cleanup deletes the known references only after the workflow no longer needs them. The workflow returns its original context reference even on terminal failure so failed cleanup can release the exact pinned object. Terminal workflow failures are converted to their sanitized original error before prepared-input validation so a secondary schema error cannot hide the actionable failure.

The orchestration worker registers ProcessPromptLabSnapshot, the text worker registers prepare_prompt_lab_snapshot_inputs, and the existing audio, vision, document, video, and text workers process the seven enabled media types. Before rolling back worker code, stop new bootstrap submissions and let active Prompt Lab workflow histories drain. Disabling the feature gate prevents new service launches but does not cancel an already-started Temporal workflow.

Task Queues

Queue	Worker	Activities
`orchestration`	`workers/orchestration_worker.py`	`ProcessConversationChunk`, `ProcessSplitConversationReport`, `ProcessReportPartition`, `ProcessPromptLabSnapshot`
`text-processing`	`workers/text_worker.py`	`run_flow_pipeline`, `process_link`, `prepare_prompt_lab_snapshot_inputs`, split manifest/hydrate/store activities
`report-split-processing`	`workers/report_split_worker.py`	retryable split prepare plus one-shot model reducer; Activity concurrency defaults to 1
`audio-processing`	`workers/audio_worker.py`	`transcribe_audio`
`vision-processing`	`workers/vision_worker.py`	`analyze_image` for static images
`document-processing`	`workers/document_worker.py`	`process_document`
`video-processing`	`workers/video_worker.py`	`process_video`

The split worker uses a 2400-second cooperative shutdown budget by default. SIGTERM/SIGINT trigger Worker.shutdown() and Docker waits the same interval, so a deployment can finish the one-attempt reducer before replacing the container. Fail-closed operations send SIGTERM to brokers and Temporal workers; the SDK stops polling and drains already-started Activities within that budget. Partition hydrate is also idempotent: its three sequential artifact reads use one in-process attempt inside the two-minute Activity window, while Temporal retries the complete Activity up to three times. This avoids stacking three full internal retry budgets under a single start-to-close timeout.

Media Routing

Message type	Activity	Queue	Output
`audio`	`transcribe_audio`	`audio-processing`	transcription + model trace
`image`	`analyze_image`	`vision-processing`	visual JSON description + model trace
`sticker`, `gif`	`process_video`	`video-processing`	static vision result or animated whole-video analysis
`video`	`process_video`	`video-processing`	whole-video summary, audio transcription, timeline, trace
`document`	`process_document`	`document-processing`	text excerpt, embedded image summaries, traces
text with links	`process_link`	`text-processing`	fetched metadata summaries

Step Contracts

run_flow_pipeline input:

{
  "run_id": "uuid",
  "request": {"flow": "...", "recipe": "...", "payload": {}},
  "media_processed": {}
}

Output is FlowRunResult serialized as JSON.

Timeouts And Retries

Current workflow constants:

Media activities: 5 minutes, maximum 2 attempts, 10-second heartbeat timeout.
Final flow activity: 4 minutes, maximum 2 attempts.
Split artifact load/hydrate/store activities: 2 minutes, maximum 3 attempts.
Split prepare: REPORT_SPLIT_PREPARE_ACTIVITY_TIMEOUT_SECONDS (default 900),

maximum 3 attempts. It performs bounded parallel artifact hydration, recomposition, merge/gates and content-addressed checkpoint writes.

Split reducer: REPORT_REDUCE_ACTIVITY_TIMEOUT_SECONDS (default 2100), one

Temporal attempt because its primary/shadow calls are not replay-safe. Startup checks this against the explicit SDK retries, JSON corrective retry, primary and shadow calls plus two artifact-read waves, final write and compute reserve.

Eligible Daily workflow execution: REPORT_WORKFLOW_EXECUTION_TIMEOUT_SECONDS

(default 43200). Startup verifies it covers configured child/media waves and retry budgets with headroom.

Split launcher preflight is bounded to 900 seconds. Signed intermediate

artifact references and input media TTL checks cover this bound before the complete workflow deadline begins.

Before persisting a viable split, effective query-signed/tokenized media URLs

must cover that deadline plus REPORT_SPLIT_MEDIA_URL_EXPIRY_HEADROOM_SECONDS (default 300), plus the 900-second preflight bound. A second pre-start check after artifact preparation requires the full deadline again, so slow object-store writes cannot consume the URL lifetime silently.

Activities must be idempotent. They receive explicit input and return explicit output. Every media Activity heartbeats immediately and then every 2 seconds. Temporal cancellation therefore reaches an in-flight HTTP download and cancels its async task; downloader and processor context managers must finish cleanup before the cancelled Activity exits. Heartbeats run on a dedicated thread so synchronous PDF/DOCX/PIL phases cannot starve the 10-second timeout by blocking the worker event loop. CancelledError is propagated and is never converted to an expected media degradation. Fixture-based replay coverage verifies, without downloading a Temporal test server, that the Activity option change replays histories created by the previous workflow code on Temporal SDK 1.25 without a patch marker. Rollout media workers before orchestration; rollback must retain heartbeat-capable workers until affected histories drain. The video activity defaults to OpenRouter Gemini and preserves the legacy frame/audio backend behind an explicit per-run override. Animated image media is converted at 1 FPS inside the Activity; media bytes never enter Temporal history.

Failure Behavior

Optional media activity exceptions become status="activity_failed" items in

media_processed after retries. The workflow continues and the conversation builder renders [MEDIA_ERROR] for that message.

Audio inputs with no audio stream return status="activity_failed" and

error_code="no_audio_track" as a successful Activity result. No Groq call or Temporal retry occurs; the report continues in degraded mode. Technical ffprobe failures raise audio_probe_failed:* and retain the normal retry behavior.

Animated image and video frame failures are partial when another frame or the

video audio is still usable. If no usable media output remains, only that media item degrades.

OpenRouter video/provider errors are re-raised for Temporal retry and, after

exhausting retries, degrade only that media item. Only a deterministic inline-size limit selects the legacy backend automatically.

AI JSON failures in run_flow_pipeline fail the run.
The launcher rejects workflow inputs above

TEMPORAL_WORKFLOW_INPUT_SAFE_BYTES. Each media activity input and the final flow activity input are checked against TEMPORAL_ACTIVITY_INPUT_SAFE_BYTES; the final fan-in is compacted before it fails with temporal_activity_input_too_large.

Shadow model failures do not fail the run.
Split guardrails fail explicitly with stable report_* errors for an

indivisible oversized item, partition-count/total-size limits, corrupt artifacts, recomposition mismatch or prompt-budget overflow. No content is silently dropped to make a report fit.

A viable split with a signed media URL fails before Temporal start with

report_split_media_url_ttl_insufficient when its known expiry cannot cover the run, or report_split_media_url_expiry_unverifiable when signing markers exist without one unambiguous parseable expiry. URL/query values are never logged. Recipe-excluded media and requests which safely fall back inline are not subject to this split-only check.

Privacy

Temporal history can persist payloads. media_url accepts only remote http/https transport references or the explicit wardian://media-omitted-by-recipe placeholder; embedded data URIs and local paths fail validation before workflow start. Do not put raw media bytes in activity arguments or results. Temporary files must live in /tmp/wardian/ and be owned by a context manager or deleted in finally. Model request/response debug and source media URLs are removed before an oversized media audit artifact is written.

media_processed audit artifacts are operationally temporary. Release this CIS change together with the WAHA alpha object-store lifecycle rule that expires the flow-artifacts/media_processed/ prefix after 7 days; do not enable derived media artifact externalization without that retention guardrail.

Split execution additionally persists bounded transport artifacts under flow-artifacts/report_split/: normalized base/partition JSON, derived media text permitted by the active privacy profile, prepared recomposed-request/media checkpoints, and the internal final result reference. It never stores raw media bytes. These artifacts are not product memory and must not be reused by later runs. The feature flag must remain off until the object store enforces an equal-or-shorter lifecycle than REPORT_SPLIT_ARTIFACT_RETENTION_DAYS (default 8) over the complete prefix, including failed/partial runs, with private access and encryption at rest. The setting declares the required retention; WAHA/CP object-store deployment enforces it. With v2 reference emission enabled, CIS refuses to start unless that retention outlives the maximum signed-reference TTL plus clock skew. Keys below this scoped prefix are clock-independent and content-addressed, so an Activity retry across midnight returns the same ref. Tenantless POST /analyze runs retain an empty external tenant memo while their internal v2 transport artifacts use the reserved cis-internal-tenantless scope plus request/run binding. Workflow payloads carry that value separately as artifact_tenant_id; media processing receives only the application tenant_id, which remains empty for tenantless runs.

Cancellation and workflow-id fencing

Flow cancellation addresses the same deterministic workflow id used by broker idempotency. A running execution receives Temporal cancellation, which propagates to child workflows. Every workflow Activity uses WAIT_CANCELLATION_COMPLETED, and workflow handlers re-raise Temporal cancellation instead of converting it to a failed result. The non-idempotent report reducer heartbeats while its provider operation is active so cooperative cancellation stops that operation before artifact cleanup and the receipt. If no execution exists, CIS starts the terminal CanceledFlowFence workflow with REJECT_DUPLICATE so a publish/cancel race cannot create work after the cancellation receipt. A redelivery or API reconciliation that observes Temporal status CANCELED restores repository state canceled even after the worker's in-memory repository was restarted.

Cancellation is cooperative and provider abort is best effort. A receipt never promises that a remote model provider deleted an already accepted request. Repository state canceled is terminal: late local completion or failure callbacks cannot replace it.

During rolling deployment, install the new orchestration workflow type on every orchestration worker before enabling the cancellation consumer. Disabling the feature stops new command processing but does not revert terminal workflow ids.