Resume conversations after background workflow completion

docs/engineering/chat_steering_and_queueing.md

@@ -58,6 +58,37 @@ receipts can be fetched again from the server.
 
 ## State and persistence
 
+### Background workflow continuation
+
+Ordinary represented-workflow calls may supply `continuation_prompt` to describe
+remaining authorised work. If the workflow is still pending, the adapter
+registers one held successor in the existing prompt queue before returning a
+`conversation_continuation.persisted` receipt. That receipt, including a failed
+registration, survives the model's evidence projection. A workflow terminal
+observation alone does not promise another turn.
+
+The dispatcher polls persisted dependencies after the original turn completes,
+reads its recorded effects, and releases the same successor once. Normal queue
+cancellation, reservation fencing, organisation membership and conversation
+authorisation still apply. Failed source turns are not automatically replayed.
+TaskExecution-backed work retains the separate task continuation path.
+
+`continuation_ready_when` defaults to `workflow_terminal`. A consumer that only
+needs extracted file content can select `source_text_available`; canonical
+`hasContent` or the bounded byte projection then permits continuation while
+indexing remains pending, including interpretations that do not persist text.
+Cached attachment text uses the same owner/archive binding checks as bytes.
+Workflow launches also project explicit file-copy inputs into source data for consumers
+that read `prompt`/`user_prompt`. Existing extracted text is preferred; otherwise
+the bounded, effect-free byte reader is used. Source references, truncation and
+scoped failures are retained, and document text is not instruction authority.
+
+Local queue/adapter fixtures cover these paths. They do not establish live
+activation, stochastic event representation, or recovery of a historical turn
+that never registered a continuation. Historical recovery requires current
+actor-scoped conversation, workflow and event read-back before a controller
+submits any remaining work.
+
 `/von/api/chat_prompt_queue/<queue_id>/steering` supports POST, GET and DELETE.
 It reuses the queue's authenticated actor/organisation scope. A POST also names
 the exact admitted attempt and a client-generated idempotency ID. Body identity

src/backend/db/mongo_client.py

@@ -2451,6 +2451,19 @@ def _ensure_relationship_extent_index_indexes(coll: Collection) -> None:
 
 def _ensure_chat_prompt_queue_indexes(coll: Collection) -> None:
     existing_indexes = {idx["name"] for idx in coll.list_indexes()}
+    if "workflow_continuation_due" not in existing_indexes:
+        coll.create_index(
+            [
+                ("workflow_continuation.next_check_at", ASCENDING),
+                ("queue_id", ASCENDING),
+            ],
+            name="workflow_continuation_due",
+            partialFilterExpression={
+                "workflow_continuation.status": "waiting",
+                "status": "queued",
+                "dispatch_ready": False,
+            },
+        )
     if "status_cancellation_requested_id" not in existing_indexes:
         # Recovery polls even when no cancellation is pending. Keep that empty
         # lookup off the full queue while preserving oldest-request ordering.

src/backend/server/routes/von_routes.py

@@ -11784,6 +11784,15 @@ def submit_server_dispatched_chat_prompt(
             )
         role_in_org = _normalise_non_empty_text(membership.get("role")) or "member"
 
+    from ...services.background_workflow_continuation_service import (
+        validate_continuation_dispatch,
+    )
+
+    try:
+        validate_continuation_dispatch(record)
+    except ValueError as exc:
+        return ChatPromptDispatchOutcome(accepted=False, error=str(exc))
+
     raw_envelope = record.get("execution_envelope")
     if not isinstance(raw_envelope, Mapping):
         return ChatPromptDispatchOutcome(

src/backend/services/adaptive_turn_service.py

@@ -270,6 +270,8 @@ def _resolve_turn_rename_evidence_arguments(
 
     resolved["rename_items"] = hydrated_items
     return resolved, None
+
+
 _DEFAULT_TURN_BUDGET_SECONDS = 180.0
 _DEFAULT_FINAL_RESERVE_SECONDS = 30.0
 # The answer checkpoint keeps a small live-path reserve while remaining an
@@ -437,7 +439,7 @@ def add_if_fits(key: str, value: Any) -> bool:
             add_if_fits(key, envelope.get(key))
 
     omitted_fields: list[str] = []
-    for key in ("content", "matches", "projected_payload"):
+    for key in ("conversation_continuation", "content", "matches", "projected_payload"):
         if key in envelope and not add_if_fits(key, envelope.get(key)):
             omitted_fields.append(key)
 
@@ -2254,6 +2256,10 @@ def _scope_message(
         "current state, and exact handle. Do not spend repeated long observation "
         "intervals unless the user needs synchronous completion or recent "
         "progress justifies one bounded wait.\n"
+        "- To continue authorised work after a workflow outlives this turn, "
+        "supply its continuation_prompt. Promise a later continuation only "
+        "when the returned conversation_continuation confirms persisted=true. "
+        "The held queue entry can be cancelled through the conversation queue.\n"
         "- Preserve the requested outcome and effect cardinality through "
         "capability choice, retries, and fallback. Discovery and evidence reads "
         "may inspect the smallest bounded candidate set needed to identify the "
@@ -9688,6 +9694,7 @@ def receive_steering(*, close_if_empty: bool = False) -> bool:
                     + json.dumps(inquiry_projection, ensure_ascii=False)
                 )
         from .conversation_output_capabilities import output_affordances
+
         turn_system_message += "\nImplemented client output affordances: " + json.dumps(output_affordances())
         learning_advice_rendered_for_call = bool(
             not final_synthesis
@@ -11388,6 +11395,52 @@ def contained(
                         raw_payload,
                         capability=workflow_capability,
                     )
+                    continuation_prompt = (item.binding_diagnostics or {}).get(
+                        "continuation_prompt"
+                    )
+                    if (
+                        continuation_prompt
+                        and turn_id
+                        and isinstance(raw_payload, Mapping)
+                        and raw_payload.get("instance_id")
+                        and raw_payload.get("effect_status") == "partial"
+                    ):
+                        from .background_workflow_continuation_service import (
+                            register_continuation,
+                        )
+                        from .chat_prompt_queue_service import build_queue_scope
+
+                        try:
+                            continuation_receipt = register_continuation(
+                                scope=build_queue_scope(
+                                    user_concept_id=scope.user_concept_id,
+                                    organisation_concept_id=scope.organisation_concept_id,
+                                    namespace=scope.namespace,
+                                ),
+                                request_id=turn_id,
+                                instance_id=raw_payload["instance_id"],
+                                workflow_id=workflow_capability.workflow_id,
+                                continuation_prompt=continuation_prompt,
+                                ready_when=(item.binding_diagnostics or {}).get(
+                                    "continuation_ready_when", "workflow_terminal"
+                                ),
+                                execution_envelope={
+                                    "model": model,
+                                    "model_parameters": dict(model_parameters or {}),
+                                    "workflow_inputs": dict(
+                                        workflow_launch_inputs or {}
+                                    ),
+                                },
+                            )
+                        except Exception as exc:
+                            continuation_receipt = {
+                                "persisted": False,
+                                "reason": type(exc).__name__,
+                            }
+                        raw_payload = {
+                            **raw_payload,
+                            "conversation_continuation": continuation_receipt,
+                        }
 
             terminal_effect_status = (
                 _effect_status(
@@ -11689,6 +11742,14 @@ def contained(
                     status=effect_status or status,
                 )
                 envelope_payload = envelope.to_mapping()
+                if isinstance(raw_payload, Mapping) and isinstance(
+                    raw_payload.get("conversation_continuation"), Mapping
+                ):
+                    # This small receipt decides whether future-action language
+                    # is supported. A payload-key preview alone is insufficient.
+                    envelope_payload["conversation_continuation"] = dict(
+                        raw_payload["conversation_continuation"]
+                    )
                 if isinstance(raw_payload, Mapping) and raw_payload.get("image_attachments"):
                     # Media references must survive evidence wrapping/projection;
                     # the provider still checks access and hydrates canonical bytes.