From c24d8fd336c9c4e303a6098d654deea5f21de10f Mon Sep 17 00:00:00 2001 From: witbrock Date: Tue, 28 Jul 2026 16:24:43 +0100 Subject: [PATCH] Reconcile late durable workflow outcomes --- .../integrations/internal_mcp/catalogue.py | 67 +++- .../turn_execution_diagnostics_service.py | 143 ++++++++- .../services/turn_execution_record_service.py | 298 +++++++++++++++++- .../workflows/durable/instance_manager.py | 52 +++ tests/backend/test_durable_workflow_system.py | 42 +++ ...g_turn_execution_records_mcp_read_tools.py | 171 ++++++++++ .../test_subworkflow_budget_diagnosability.py | 23 ++ ...cution_record_service_execution_summary.py | 140 ++++++++ 8 files changed, 923 insertions(+), 13 deletions(-) diff --git a/src/backend/integrations/internal_mcp/catalogue.py b/src/backend/integrations/internal_mcp/catalogue.py index ff5e5d4e8..11e9ea886 100644 --- a/src/backend/integrations/internal_mcp/catalogue.py +++ b/src/backend/integrations/internal_mcp/catalogue.py @@ -11437,6 +11437,61 @@ def _summarise_turn_execution_tool_invocations( entry["tool"] = tool_name if status_text is not None: entry["status"] = status_text + for field_name in ( + "call_id", + "execution_id", + "effect_id", + "effect_status", + "error_code", + "mutation_outcome", + "outcome_finality", + "capability_kind", + "execution_method", + "represented_workflow_id", + "workflow_id", + "instance_id", + "durable_submission_status", + "final_status", + ): + value = raw_entry.get(field_name) + if isinstance(value, str) and value.strip(): + entry[field_name] = value.strip() + evidence = raw_entry.get("evidence") + if isinstance(evidence, Mapping): + evidence_provenance = ( + evidence.get("provenance") + if isinstance(evidence.get("provenance"), Mapping) + else {} + ) + for field_name in ( + "execution_id", + "effect_id", + "effect_status", + "error_code", + "mutation_outcome", + "outcome_finality", + "workflow_id", + "instance_id", + "durable_submission_status", + "final_status", + ): + if field_name in entry: + continue + value = evidence.get(field_name) + if isinstance(value, str) and value.strip(): + entry[field_name] = value.strip() + for field_name in ( + "capability_kind", + "execution_method", + "represented_workflow_id", + ): + if field_name in entry: + continue + value = evidence_provenance.get(field_name) + if isinstance(value, str) and value.strip(): + entry[field_name] = value.strip() + if isinstance(raw_entry.get("changed"), bool): + entry["changed"] = raw_entry.get("changed") summary.append(entry) if len(summary) >= max_items: break @@ -11635,6 +11690,10 @@ def _extract_turn_execution_tool_invocation_summary( if isinstance(existing_summary, list): return _summarise_turn_execution_tool_invocations(existing_summary) + direct_invocations = item.get("tool_invocations") + if isinstance(direct_invocations, list): + return _summarise_turn_execution_tool_invocations(direct_invocations) + execution_raw = item.get("execution") execution = execution_raw if isinstance(execution_raw, Mapping) else {} return _summarise_turn_execution_tool_invocations(execution.get("tool_invocations")) @@ -23200,7 +23259,13 @@ def _rag_get_item(**kwargs): execution_summary: Mapping[str, Any] = ( execution_summary_raw if isinstance(execution_summary_raw, Mapping) else {} ) - tool_invocations = execution_payload.get("tool_invocations") + direct_tool_invocations = doc.get("tool_invocations") + tool_invocations = ( + direct_tool_invocations + if isinstance(direct_tool_invocations, list) + and direct_tool_invocations + else execution_payload.get("tool_invocations") + ) discovery_payload_raw = workflow_routing_diagnostics.get("discovery") discovery_payload: Mapping[str, Any] = ( discovery_payload_raw if isinstance(discovery_payload_raw, Mapping) else {} diff --git a/src/backend/services/turn_execution_diagnostics_service.py b/src/backend/services/turn_execution_diagnostics_service.py index 61c723873..6ed9fc145 100644 --- a/src/backend/services/turn_execution_diagnostics_service.py +++ b/src/backend/services/turn_execution_diagnostics_service.py @@ -524,6 +524,119 @@ def _normalise_tool_history(tool_history: Any) -> list[dict[str, Any]]: return [] +def _summarise_tool_invocation_history( + tool_invocations: Any, +) -> list[dict[str, Any]]: + """Build a bounded diagnostic history from ordinary-turn invocations.""" + + if not isinstance(tool_invocations, list): + return [] + summary: list[dict[str, Any]] = [] + for raw_entry in tool_invocations: + if not isinstance(raw_entry, Mapping): + continue + tool_name = ( + _safe_str(raw_entry.get("tool")) + or _safe_str(raw_entry.get("tool_name")) + or _safe_str(raw_entry.get("method")) + ) + if not tool_name: + continue + entry: dict[str, Any] = {"tool": tool_name} + for key in ( + "status", + "call_id", + "effect_id", + "effect_status", + "error_code", + "execution_id", + "capability_kind", + "execution_method", + "represented_workflow_id", + "workflow_id", + "instance_id", + "durable_submission_status", + "mutation_outcome", + "outcome_finality", + "final_status", + "result_summary", + ): + value = raw_entry.get(key) + if isinstance(value, str) and value.strip(): + entry[key] = value.strip() + if isinstance(raw_entry.get("changed"), bool): + entry["changed"] = raw_entry.get("changed") + transport = raw_entry.get("transport") + if isinstance(transport, Mapping): + entry["transport"] = { + key: transport.get(key) + for key in ( + "schema_version", + "execution_id", + "outcome", + "duration_ms", + "timeout_sec", + "advisory_timeout_sec", + "queue_duration_ms", + "handler_duration_ms", + "handler_elapsed_ms", + "timeout_phase", + "late_result_policy", + ) + if key in transport + } + if "execution_id" not in entry: + execution_id = _safe_str(transport.get("execution_id")) + if execution_id: + entry["execution_id"] = execution_id + evidence = raw_entry.get("evidence") + if isinstance(evidence, Mapping): + for key in ( + "effect_status", + "error_code", + "instance_id", + "workflow_id", + "durable_submission_status", + "mutation_outcome", + "outcome_finality", + "final_status", + ): + if key in entry: + continue + value = evidence.get(key) + if isinstance(value, str) and value.strip(): + entry[key] = value.strip() + summary.append(entry) + return summary + + +def _fallback_tool_invocation_history( + *, + execution: Mapping[str, Any] | None, + turn_record: Mapping[str, Any] | None, + llm_debug: Mapping[str, Any] | None, +) -> list[dict[str, Any]]: + candidates = ( + execution.get("tool_invocations") + if isinstance(execution, Mapping) + else None, + turn_record.get("tool_invocations") + if isinstance(turn_record, Mapping) + else None, + llm_debug.get("tool_invocations") + if isinstance(llm_debug, Mapping) + else None, + llm_debug.get("invocations") + if isinstance(llm_debug, Mapping) + else None, + ) + for candidate in candidates: + summary = _summarise_tool_invocation_history(candidate) + if summary: + return summary + return [] + + def _derive_tool_counts(tool_history: Sequence[Mapping[str, Any]]) -> dict[str, int]: total = 0 success = 0 @@ -534,7 +647,7 @@ def _derive_tool_counts(tool_history: Sequence[Mapping[str, Any]]) -> dict[str, continue total += 1 status = (_safe_str(entry.get("status")) or "").lower() - if status in {"completed", "complete", "success", "succeeded"}: + if status in {"ok", "completed", "complete", "success", "succeeded"}: success += 1 ended += 1 elif status in { @@ -544,6 +657,11 @@ def _derive_tool_counts(tool_history: Sequence[Mapping[str, Any]]) -> dict[str, "blocked", "cancelled", "canceled", + "partial", + "indeterminate", + "not_started", + "timeout", + "timed_out", }: failure += 1 ended += 1 @@ -1087,8 +1205,10 @@ def _build_fallback_turn_execution_diagnostics( ) ) - tool_history = _normalise_tool_history( - execution.get("tool_invocations") if execution else None + tool_history = _fallback_tool_invocation_history( + execution=execution, + turn_record=turn_record, + llm_debug=llm_debug_mapping, ) tool_counts = _derive_tool_counts(tool_history) @@ -1344,16 +1464,19 @@ def _normalise_embedded_diagnostics_payload( ) tool_history = _normalise_tool_history(payload.get("tool_history")) - tool_invocations = execution.get("tool_invocations") if execution else None - if not tool_history and isinstance(tool_invocations, list): - tool_history = _normalise_tool_history(tool_invocations) - payload["tool_history"] = tool_history - else: - payload["tool_history"] = tool_history + hydrated_tool_history = False + if not tool_history: + tool_history = _fallback_tool_invocation_history( + execution=execution, + turn_record=turn_record, + llm_debug=llm_debug_mapping, + ) + hydrated_tool_history = bool(tool_history) + payload["tool_history"] = tool_history tool_counts = _derive_tool_counts(tool_history) for key, value in tool_counts.items(): - if not isinstance(payload.get(key), int): + if hydrated_tool_history or not isinstance(payload.get(key), int): payload[key] = value _attach_minimal_turn_timing_trace( diff --git a/src/backend/services/turn_execution_record_service.py b/src/backend/services/turn_execution_record_service.py index 6942d58e3..a4294cbf8 100644 --- a/src/backend/services/turn_execution_record_service.py +++ b/src/backend/services/turn_execution_record_service.py @@ -11146,8 +11146,16 @@ def _debug_list(key: str) -> list[Any] | None: response_text=None, interaction_timestamp_utc=debug.get("interaction_timestamp_utc"), workflow_discovery=_debug_mapping("workflow_discovery"), - workflow_routing=_debug_mapping("workflow_routing"), - tool_invocations=_debug_list("invocations"), + workflow_routing=( + _debug_mapping("workflow_routing") + or infer_turn_execution_workflow_routing_from_debug( + llm_debug=debug + ) + ), + tool_invocations=( + _debug_list("tool_invocations") + or _debug_list("invocations") + ), turn_execution_diagnostics=_debug_mapping("turn_execution_diagnostics"), aux_llm_calls=_debug_list("aux_llm_calls"), llm_calls=_debug_list("llm_calls"), @@ -11600,6 +11608,256 @@ def record_effect_observation_phase( } +def reconcile_durable_workflow_terminal_effect( + *, + request_id: Any, + instance_id: Any, + workflow_id: Any, + terminal_status: Any, + final_state: Any = None, + completed_at: Any = None, + error: Any = None, + error_step: Any = None, + execution_trace_id: Any = None, + user_id: Any = None, + namespace: Any = None, + org_id: Any = None, +) -> dict[str, Any]: + """Reconcile one canonical durable terminal state with its timed-out turn. + + A represented workflow may be durably submitted before the ordinary-turn + transport reaches its hard deadline. The immutable turn receipt must stay + partial at that decision boundary, while the actor-scoped effect journal + later records the canonical workflow terminal state. This function only + observes and records that state; it never executes or retries the workflow. + """ + + clean_request_id = _safe_str(request_id) + clean_instance_id = _safe_str(instance_id) + clean_workflow_id = _safe_str(workflow_id) + raw_terminal_status = getattr(terminal_status, "value", terminal_status) + clean_terminal_status = (_safe_str(raw_terminal_status) or "").lower() + if not clean_request_id: + return {"updated": False, "reason": "missing_request_id"} + if not clean_instance_id: + return { + "updated": False, + "reason": "missing_instance_id", + "request_id": clean_request_id, + } + if not clean_workflow_id: + return { + "updated": False, + "reason": "missing_workflow_id", + "request_id": clean_request_id, + "instance_id": clean_instance_id, + } + if clean_terminal_status not in {"completed", "failed", "cancelled"}: + return { + "updated": False, + "reason": "workflow_not_terminal", + "request_id": clean_request_id, + "instance_id": clean_instance_id, + } + + actor_scope: dict[str, str] = {} + for field_name, raw_value in ( + ("namespace", namespace), + ("user_id", user_id), + ("org_id", org_id), + ): + clean_value = _safe_str(raw_value) + if clean_value: + actor_scope[field_name] = clean_value + if not actor_scope: + return { + "updated": False, + "reason": "missing_actor_scope", + "request_id": clean_request_id, + "instance_id": clean_instance_id, + } + + coll = get_turn_execution_records_collection() + if coll is None: + return { + "updated": False, + "reason": "collection_unavailable", + "request_id": clean_request_id, + "instance_id": clean_instance_id, + } + try: + record = _turn_execution_find_one( + coll, + {"request_id": clean_request_id, **actor_scope}, + projection={ + "_id": 0, + "effect_observation_journal": 1, + }, + operation="reconcile_durable_workflow_terminal_effect.find_record", + detail=clean_instance_id, + ) + except PyMongoError as exc: + logger.warning( + "Failed to load effect journal for durable terminal reconciliation " + "request_id=%s instance_id=%s: %s", + clean_request_id, + clean_instance_id, + exc, + ) + return { + "updated": False, + "reason": "mongo_error", + "request_id": clean_request_id, + "instance_id": clean_instance_id, + } + if not isinstance(record, Mapping): + return { + "updated": False, + "reason": "turn_execution_record_not_found", + "request_id": clean_request_id, + "instance_id": clean_instance_id, + } + + journal = record.get("effect_observation_journal") + if not isinstance(journal, Mapping): + return { + "updated": False, + "reason": "effect_observation_journal_not_found", + "request_id": clean_request_id, + "instance_id": clean_instance_id, + } + + matched_effect_id: str | None = None + matched_entry: Mapping[str, Any] | None = None + matched_turn_terminal: Mapping[str, Any] | None = None + matched_receipt: Mapping[str, Any] | None = None + for raw_effect_id, raw_entry in journal.items(): + if not isinstance(raw_effect_id, str) or not isinstance(raw_entry, Mapping): + continue + turn_terminal = raw_entry.get("turn_terminal") + if not isinstance(turn_terminal, Mapping): + continue + receipt = turn_terminal.get("receipt") + if not isinstance(receipt, Mapping): + continue + if _safe_str(receipt.get("instance_id")) != clean_instance_id: + continue + receipt_workflow_id = _safe_str(receipt.get("workflow_id")) + if receipt_workflow_id and receipt_workflow_id != clean_workflow_id: + continue + matched_effect_id = raw_effect_id + matched_entry = raw_entry + matched_turn_terminal = turn_terminal + matched_receipt = receipt + break + + if ( + matched_effect_id is None + or matched_entry is None + or matched_turn_terminal is None + or matched_receipt is None + ): + return { + "updated": False, + "reason": "durable_submission_receipt_not_found", + "request_id": clean_request_id, + "instance_id": clean_instance_id, + } + + identity = ( + matched_entry.get("identity") + if isinstance(matched_entry.get("identity"), Mapping) + else {} + ) + transport = ( + matched_turn_terminal.get("transport") + if isinstance(matched_turn_terminal.get("transport"), Mapping) + else {} + ) + execution_id = ( + _safe_str(matched_receipt.get("execution_id")) + or _safe_str(transport.get("execution_id")) + ) + changed = ( + matched_receipt.get("changed") + if isinstance(matched_receipt.get("changed"), bool) + else None + ) + completed = clean_terminal_status == "completed" + effect_status = ( + "succeeded" + if completed + else ("partial" if changed is True else "failed") + ) + mutation_outcome = ( + "succeeded" + if completed + else ("partial" if changed is True else "failed") + ) + completed_at_value = ( + _iso_utc(completed_at) + if isinstance(completed_at, datetime) + else _safe_str(completed_at) + ) + canonical_receipt = { + "schema_version": "durable_workflow_terminal_receipt.v1", + "success": completed, + "status": clean_terminal_status, + "effect_status": effect_status, + "changed": changed, + "mutation_outcome": mutation_outcome, + "outcome_finality": "canonical_durable_terminal", + "workflow_id": clean_workflow_id, + "instance_id": clean_instance_id, + "final_status": clean_terminal_status, + "final_state": _safe_str(final_state), + "completed_at": completed_at_value, + "error": _safe_str(error), + "error_step": _safe_str(error_step), + "execution_trace_id": _safe_str(execution_trace_id), + } + canonical_receipt = { + key: value for key, value in canonical_receipt.items() if value is not None + } + observation = { + "schema_version": "durable_workflow_terminal_observation.v1", + "call_id": _safe_str(identity.get("call_id")) + or _safe_str(matched_turn_terminal.get("call_id")), + "capability_name": _safe_str(identity.get("capability_name")) + or _safe_str(matched_turn_terminal.get("capability_name")), + "execution_id": execution_id, + "method_name": "workflow_execute", + "outcome": "late_success", + "observed_at_utc": _iso_utc(_now_utc()), + "output_schema_validation": "canonical_durable_instance", + "output_schema_valid": True, + "payload_truncated": False, + "effect_status": effect_status, + "changed": changed, + "payload": canonical_receipt, + } + observation = { + key: value for key, value in observation.items() if value is not None + } + outcome = record_effect_observation_phase( + request_id=clean_request_id, + effect_id=matched_effect_id, + phase="late_terminal", + observation=observation, + user_id=user_id, + namespace=namespace, + org_id=org_id, + ) + return { + **outcome, + "instance_id": clean_instance_id, + "workflow_id": clean_workflow_id, + "terminal_status": clean_terminal_status, + "effect_status": effect_status, + "execution_id": execution_id, + } + + def append_late_effect_observation( *, request_id: Any, @@ -12577,6 +12835,42 @@ def infer_turn_execution_workflow_routing_from_debug( for raw_invocation in tool_invocations: if not isinstance(raw_invocation, Mapping): continue + evidence = ( + raw_invocation.get("evidence") + if isinstance(raw_invocation.get("evidence"), Mapping) + else {} + ) + provenance = ( + evidence.get("provenance") + if isinstance(evidence.get("provenance"), Mapping) + else {} + ) + capability_kind = ( + _safe_str(raw_invocation.get("capability_kind")) + or _safe_str(provenance.get("capability_kind")) + ) + execution_method = ( + _safe_str(raw_invocation.get("execution_method")) + or _safe_str(provenance.get("execution_method")) + ) + represented_workflow_id = ( + _safe_str(raw_invocation.get("represented_workflow_id")) + or _safe_str(raw_invocation.get("workflow_id")) + or _safe_str(evidence.get("workflow_id")) + or _safe_str(provenance.get("represented_workflow_id")) + ) + if represented_workflow_id and ( + capability_kind == "represented_workflow" + or execution_method == "workflow_execute" + or str(raw_invocation.get("tool") or "").startswith( + "represented_workflow_" + ) + ): + return { + "workflow_id": represented_workflow_id, + "verdict": "represented_workflow_execution", + "source": "synthesised_from_tool_invocation", + } if _is_write_tool(_safe_str(raw_invocation.get("name"))): has_write_tools = True break diff --git a/src/backend/workflows/durable/instance_manager.py b/src/backend/workflows/durable/instance_manager.py index fbdb4e3f7..770c32c6a 100644 --- a/src/backend/workflows/durable/instance_manager.py +++ b/src/backend/workflows/durable/instance_manager.py @@ -2064,6 +2064,55 @@ def _emit_episode_evaluation_terminal_event( exc, ) + @staticmethod + def _reconcile_conversation_turn_terminal_effect( + instance: WorkflowInstance, + ) -> None: + """Best-effort canonical read-back for a timed-out ordinary turn.""" + + if ( + instance.source_event_type != "conversation_turn" + or not instance.source_event_id + ): + return + try: + from ...services.turn_execution_record_service import ( + reconcile_durable_workflow_terminal_effect, + ) + + outcome = reconcile_durable_workflow_terminal_effect( + request_id=instance.source_event_id, + instance_id=instance.instance_id, + workflow_id=instance.workflow_id, + terminal_status=instance.status, + final_state=instance.current_state, + completed_at=instance.completed_at, + error=instance.error, + error_step=instance.error_step, + execution_trace_id=instance.execution_trace_id, + user_id=instance.user_id, + namespace=instance.namespace, + org_id=instance.org_id, + ) + reason = str(outcome.get("reason") or "") + if not bool(outcome.get("updated") or outcome.get("duplicate")) and reason not in { + "turn_execution_record_not_found", + "effect_observation_journal_not_found", + "durable_submission_receipt_not_found", + }: + logger.warning( + "[durable_workflow] Turn-effect terminal reconciliation " + "not acknowledged for %s: %s", + instance.instance_id, + outcome, + ) + except Exception: + logger.exception( + "[durable_workflow] Turn-effect terminal reconciliation failed " + "for %s", + instance.instance_id, + ) + def mark_completed( self, instance_id: str, @@ -2149,6 +2198,7 @@ def mark_completed( ), ) self._broadcast_instance(completed_instance) + self._reconcile_conversation_turn_terminal_effect(completed_instance) self._emit_episode_evaluation_terminal_event( instance=completed_instance, terminal_status=WorkflowInstanceStatus.COMPLETED.value, @@ -2282,6 +2332,7 @@ def mark_failed( ), ) self._broadcast_instance(failed_instance) + self._reconcile_conversation_turn_terminal_effect(failed_instance) self._emit_episode_evaluation_terminal_event( instance=failed_instance, terminal_status=WorkflowInstanceStatus.FAILED.value, @@ -2347,6 +2398,7 @@ def mark_cancelled(self, instance_id: str) -> bool: progress_updated_at=now, ) self._broadcast_instance(cancelled_instance) + self._reconcile_conversation_turn_terminal_effect(cancelled_instance) self._emit_episode_evaluation_terminal_event( instance=cancelled_instance, terminal_status=WorkflowInstanceStatus.CANCELLED.value, diff --git a/tests/backend/test_durable_workflow_system.py b/tests/backend/test_durable_workflow_system.py index 7a8cecc40..ba4d28727 100644 --- a/tests/backend/test_durable_workflow_system.py +++ b/tests/backend/test_durable_workflow_system.py @@ -600,6 +600,48 @@ def test_mark_completed_emits_episode_evaluation_terminal_event( assert launches[0]["final_state"] == "end_state" assert launches[0]["instance"].instance_id == instance_id + def test_mark_completed_reconciles_conversation_turn_effect( + self, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + manager = WorkflowInstanceManager() + reconciliations: list[dict[str, Any]] = [] + monkeypatch.setattr( + "src.backend.services.turn_execution_record_service.reconcile_durable_workflow_terminal_effect", + lambda **kwargs: ( + reconciliations.append(dict(kwargs)) + or {"updated": True} + ), + ) + instance_id = manager.create_instance( + "#V#test_workflow", + user_id="#V#user", + org_id="#V#org", + namespace="#V#user@org", + source_event_type="conversation_turn", + source_event_id="request-durable-terminal", + ) + + success = manager.mark_completed( + instance_id, + outputs={"result": "done"}, + final_state="#V#completed_state", + execution_trace_id="trace-durable-terminal", + ) + + assert success is True + assert len(reconciliations) == 1 + reconciliation = reconciliations[0] + assert reconciliation["request_id"] == "request-durable-terminal" + assert reconciliation["instance_id"] == instance_id + assert reconciliation["workflow_id"] == "#V#test_workflow" + assert reconciliation["terminal_status"] == WorkflowInstanceStatus.COMPLETED + assert reconciliation["final_state"] == "#V#completed_state" + assert reconciliation["execution_trace_id"] == "trace-durable-terminal" + assert reconciliation["user_id"] == "#V#user" + assert reconciliation["namespace"] == "#V#user@org" + assert reconciliation["org_id"] == "#V#org" + def test_mark_failed_updates_status(self) -> None: """mark_failed() should set status to FAILED with error.""" manager = WorkflowInstanceManager() diff --git a/tests/backend/test_rag_turn_execution_records_mcp_read_tools.py b/tests/backend/test_rag_turn_execution_records_mcp_read_tools.py index 64ffc5a02..f0ee03c06 100644 --- a/tests/backend/test_rag_turn_execution_records_mcp_read_tools.py +++ b/tests/backend/test_rag_turn_execution_records_mcp_read_tools.py @@ -1802,6 +1802,177 @@ def test_turn_execution_get_diagnostics_reconstructs_from_projection(monkeypatch ) +def test_turn_execution_get_prefers_identity_bearing_top_level_invocations( + monkeypatch, +): + from src.backend.integrations.internal_mcp import catalogue as cat + + turn_docs = [ + { + "request_id": "req-durable-identities", + "session_id": "chat-durable-identities", + "namespace": "#V#user@org", + "user_id": "#V#user", + "org_id": "#V#org", + "tool_invocations": [ + { + "tool": "represented_workflow_test", + "status": "error", + "call_id": "call-durable-identities", + "execution_id": "mcp-durable-identities", + "effect_id": "effect-durable-identities", + "effect_status": "partial", + "error_code": "tool_timeout_after_durable_submission", + "workflow_id": "#V#paper_workflow", + "evidence": { + "instance_id": "instance-durable-identities", + "durable_submission_status": "pending", + "provenance": { + "capability_kind": "represented_workflow", + "execution_method": "workflow_execute", + "represented_workflow_id": "#V#paper_workflow", + }, + }, + } + ], + "execution": { + "tool_invocations": [ + { + "tool": "represented_workflow_test", + "status": "timeout", + } + ] + }, + } + ] + monkeypatch.setattr( + "src.backend.db.connection_manager.get_db", + lambda: _DB({"turn_execution_records": _TurnExecutionCollection(turn_docs)}), + ) + + result = cat._turn_execution_get( + namespace="#V#user@org", + request_id="req-durable-identities", + ) + + assert result["success"] is True + assert result["tool_invocation_summary"] == [ + { + "tool": "represented_workflow_test", + "status": "error", + "call_id": "call-durable-identities", + "execution_id": "mcp-durable-identities", + "effect_id": "effect-durable-identities", + "effect_status": "partial", + "error_code": "tool_timeout_after_durable_submission", + "capability_kind": "represented_workflow", + "execution_method": "workflow_execute", + "represented_workflow_id": "#V#paper_workflow", + "workflow_id": "#V#paper_workflow", + "instance_id": "instance-durable-identities", + "durable_submission_status": "pending", + } + ] + + +def test_turn_execution_get_diagnostics_hydrates_empty_progress_tool_history( + monkeypatch, +): + from src.backend.integrations.internal_mcp import catalogue as cat + + chat_docs = [ + { + "user_id": "#V#user", + "session_id": "chat-receipt-1", + "namespace": "#V#user@org", + "organisation_concept_id": "#V#org", + "history": [ + { + "role": "assistant", + "content": "Inspect the durable workflow instance before retrying.", + "llm_debug_data": { + "request_id": "req-receipt-1", + "tool_invocations": [ + { + "tool": "turn_capabilities", + "status": "ok", + "call_id": "call-capabilities", + }, + { + "tool": "represented_workflow_test", + "status": "error", + "call_id": "call-workflow", + "effect_id": "effect-workflow", + "effect_status": "partial", + "error_code": ( + "tool_timeout_after_durable_submission" + ), + "execution_id": "mcp-workflow", + "workflow_id": "#V#paper_workflow", + "instance_id": "instance-workflow", + "transport": { + "outcome": "timed_out", + "timeout_phase": "handler", + }, + }, + { + "tool": "turn_read_evidence", + "status": "ok", + "call_id": "call-read", + }, + ], + "turn_execution_diagnostics": { + "request_id": "req-receipt-1", + "generated_at_utc": "2026-07-28T15:02:31Z", + "tool_history": [], + "tool_call_count": 0, + "tool_call_start_count": 0, + "tool_call_end_count": 0, + "tool_success_count": 0, + "tool_failure_count": 0, + "tool_pending_count": 0, + "progress_events": [], + "activity_history": [], + "phase_history": [], + "stage_diagnostics": [], + }, + }, + } + ], + } + ] + monkeypatch.setattr( + "src.backend.services.turn_execution_diagnostics_service.get_chat_history_collection_service", + lambda read_only=True: _DiagnosticsChatHistoryCollection(chat_docs), + ) + monkeypatch.setattr( + "src.backend.services.turn_execution_diagnostics_service.get_turn_execution_records_collection", + lambda: None, + ) + + result = cat._turn_execution_get_diagnostics( + namespace="#V#user@org", + request_id="req-receipt-1", + ) + + assert result["success"] is True + assert result["tool_call_count"] == 3 + assert result["tool_call_start_count"] == 3 + assert result["tool_call_end_count"] == 3 + assert result["tool_success_count"] == 2 + assert result["tool_failure_count"] == 1 + workflow_call = result["tool_history"][1] + assert workflow_call["tool"] == "represented_workflow_test" + assert workflow_call["effect_status"] == "partial" + assert workflow_call["error_code"] == ( + "tool_timeout_after_durable_submission" + ) + assert workflow_call["execution_id"] == "mcp-workflow" + assert workflow_call["workflow_id"] == "#V#paper_workflow" + assert workflow_call["instance_id"] == "instance-workflow" + assert "effective_arguments" not in workflow_call + + def test_turn_execution_list_filters_by_session_id(monkeypatch): from src.backend.integrations.internal_mcp import catalogue as cat diff --git a/tests/backend/test_subworkflow_budget_diagnosability.py b/tests/backend/test_subworkflow_budget_diagnosability.py index 96ecdf6e6..ec80624f4 100644 --- a/tests/backend/test_subworkflow_budget_diagnosability.py +++ b/tests/backend/test_subworkflow_budget_diagnosability.py @@ -149,6 +149,19 @@ def _fake_upsert(*, record, user_id=None, session_id=None, namespace=None, org_i prompt_text="Show me the last 5 email messages", llm_debug_info={ "aux_llm_calls": [{"type": "workflow_continuation_decision"}], + "tool_invocations": [ + { + "tool": "represented_workflow_test", + "status": "error", + "error_code": "tool_timeout_after_durable_submission", + "effect_status": "partial", + "mutation_outcome": "partial", + "capability_kind": "represented_workflow", + "execution_method": "workflow_execute", + "workflow_id": "#V#paper_workflow", + "instance_id": "instance-timeout-1", + } + ], }, progress_snapshot={"phase": "recovery_decision"}, ) @@ -173,6 +186,16 @@ def _fake_upsert(*, record, user_id=None, session_id=None, namespace=None, org_i assert ledger_projection["receipt"]["cause_code"] == ( "represented_critic_receipt_missing" ) + invocation = record["execution"]["tool_invocations"][0] + assert invocation["tool"] == "represented_workflow_test" + assert invocation["status"] == "partial" + assert invocation["error_code"] == ( + "tool_timeout_after_durable_submission" + ) + assert invocation["instance_id"] == "instance-timeout-1" + assert record["workflow_selection"]["selected_workflow_id"] == ( + "#V#paper_workflow" + ) def test_persist_failed_turn_execution_record_requires_request_id(): diff --git a/tests/backend/test_turn_execution_record_service_execution_summary.py b/tests/backend/test_turn_execution_record_service_execution_summary.py index ce627f7d6..4d0a7a2ba 100644 --- a/tests/backend/test_turn_execution_record_service_execution_summary.py +++ b/tests/backend/test_turn_execution_record_service_execution_summary.py @@ -14,7 +14,9 @@ build_turn_execution_record, build_workflow_routing_diagnostics, get_turn_execution_record_projection, + infer_turn_execution_workflow_routing_from_debug, project_final_answer_tool_evidence, + reconcile_durable_workflow_terminal_effect, record_effect_observation_phase, upsert_turn_execution_record_projection, _classify_tool_invocation_status, @@ -77,6 +79,33 @@ def test_tool_receipt_targets_require_explicit_generic_target_fields() -> None: ) +def test_workflow_routing_reconstruction_prefers_represented_invocation() -> None: + routing = infer_turn_execution_workflow_routing_from_debug( + llm_debug={ + "tool_invocations": [ + { + "tool": "represented_workflow_abc123", + "status": "error", + "evidence": { + "workflow_id": "#V#paper_workflow", + "provenance": { + "capability_kind": "represented_workflow", + "execution_method": "workflow_execute", + "represented_workflow_id": "#V#paper_workflow", + }, + }, + } + ] + } + ) + + assert routing == { + "workflow_id": "#V#paper_workflow", + "verdict": "represented_workflow_execution", + "source": "synthesised_from_tool_invocation", + } + + def _build_effect_projection_record( request_id: str, tool_invocations: list[dict], @@ -1031,6 +1060,117 @@ def test_effect_observation_journal_is_actor_scoped_idempotent_and_survives_upse ) +def test_durable_workflow_terminal_reconciles_timed_out_turn_effect( + monkeypatch, +) -> None: + import mongomock + import src.backend.services.turn_execution_record_service as record_service + + collection = mongomock.MongoClient().von_test.turn_execution_records + collection.create_index("request_id", unique=True) + monkeypatch.setattr( + record_service, + "get_turn_execution_records_collection", + lambda: collection, + ) + monkeypatch.setattr( + record_service, + "_turn_execution_mongo_comment", + lambda *_args, **_kwargs: None, + ) + scope = { + "user_id": "#V#user", + "namespace": "#V#user@org", + "org_id": "#V#org", + } + record_effect_observation_phase( + request_id="req-durable-late", + effect_id="effect_durable_late", + phase="dispatch_intent", + observation={ + "call_id": "call-durable-late", + "capability_name": "represented_workflow_test", + "dispatch_state": "intent_recorded", + }, + **scope, + ) + record_effect_observation_phase( + request_id="req-durable-late", + effect_id="effect_durable_late", + phase="turn_terminal", + observation={ + "call_id": "call-durable-late", + "capability_name": "represented_workflow_test", + "effect_status": "partial", + "changed": True, + "transport": { + "execution_id": "mcp-durable-late", + "outcome": "timed_out", + "late_result_policy": "observe_out_of_band", + }, + "receipt": { + "success": False, + "error_code": "tool_timeout_after_durable_submission", + "effect_status": "partial", + "mutation_outcome": "partial", + "changed": True, + "workflow_id": "#V#durable_workflow", + "instance_id": "instance-durable-late", + "execution_id": "mcp-durable-late", + }, + }, + **scope, + ) + + outcome = reconcile_durable_workflow_terminal_effect( + request_id="req-durable-late", + instance_id="instance-durable-late", + workflow_id="#V#durable_workflow", + terminal_status="completed", + final_state="#V#completed_state", + completed_at="2026-07-28T15:03:45Z", + execution_trace_id="trace-durable-late", + **scope, + ) + duplicate = reconcile_durable_workflow_terminal_effect( + request_id="req-durable-late", + instance_id="instance-durable-late", + workflow_id="#V#durable_workflow", + terminal_status="completed", + final_state="#V#completed_state", + completed_at="2026-07-28T15:03:45Z", + execution_trace_id="trace-durable-late", + **scope, + ) + + assert outcome["updated"] is True + assert outcome["effect_status"] == "succeeded" + assert duplicate["updated"] is False + assert duplicate["duplicate"] is True + stored = collection.find_one({"request_id": "req-durable-late"}) + late_terminal = stored["effect_observation_journal"][ + "effect_durable_late" + ]["late_terminal"] + assert late_terminal["outcome"] == "late_success" + assert late_terminal["effect_status"] == "succeeded" + assert late_terminal["changed"] is True + assert late_terminal["payload"] == { + "schema_version": "durable_workflow_terminal_receipt.v1", + "success": True, + "status": "completed", + "effect_status": "succeeded", + "changed": True, + "mutation_outcome": "succeeded", + "outcome_finality": "canonical_durable_terminal", + "workflow_id": "#V#durable_workflow", + "instance_id": "instance-durable-late", + "final_status": "completed", + "final_state": "#V#completed_state", + "completed_at": "2026-07-28T15:03:45Z", + "execution_trace_id": "trace-durable-late", + } + + def test_effect_observation_journal_refuses_cross_actor_request_collision( monkeypatch, ) -> None: