diff --git a/src/backend/integrations/internal_mcp/catalogue.py b/src/backend/integrations/internal_mcp/catalogue.py index b678b34e..75a41dc1 100644 --- a/src/backend/integrations/internal_mcp/catalogue.py +++ b/src/backend/integrations/internal_mcp/catalogue.py @@ -13267,7 +13267,9 @@ def _summarise_effect_observation_journal( for phase_name in ( "dispatch_intent", "turn_terminal", + "current_state_observation", "late_terminal", + "canonical_reconciliation", ): raw_phase = raw_entry.get(phase_name) if not isinstance(raw_phase, Mapping): @@ -13276,13 +13278,29 @@ def _summarise_effect_observation_journal( phase_projection = { key: raw_phase.get(key) for key in ( + "schema_version", "phase", "recorded_at_utc", "dispatch_state", "execution_id", "outcome", + "status", + "verified", "effect_status", "changed", + "initial_effect_status", + "current_outcome_status", + "outcome_resolved", + "reconciliation_basis", + "observation_identity_sha256", + "method_name", + "read_method_name", + "read_call_id", + "receipt_id", + "intent_fingerprint", + "target_concept_ids", + "canonical_scope", + "evidence_id", "output_schema_validation", "output_schema_valid", "payload_truncated", @@ -13326,27 +13344,78 @@ def _summarise_effect_observation_journal( if key in receipt } entry[phase_name] = phase_projection + entry["historical_phases"] = list(available_phases) entry["available_phases"] = available_phases - entry["latest_phase"] = available_phases[-1] if available_phases else None - late_terminal = raw_entry.get("late_terminal") - turn_terminal = raw_entry.get("turn_terminal") + recorded_phases = [ + phase_name + for phase_name in available_phases + if str(raw_entry.get(phase_name, {}).get("recorded_at_utc") or "") + ] + entry["latest_phase"] = ( + max( + recorded_phases, + key=lambda phase_name: str( + raw_entry.get(phase_name, {}).get("recorded_at_utc") or "" + ), + ) + if recorded_phases + else (available_phases[-1] if available_phases else None) + ) + # Keep each immutable receipt visible above. Current outcome is only a + # derived actor-safe projection, with exact canonical read-back taking + # precedence over earlier transport and turn-terminal observations. + current_phase = next( + ( + phase_name + for phase_name in ( + "canonical_reconciliation", + "late_terminal", + "current_state_observation", + "turn_terminal", + ) + if isinstance(raw_entry.get(phase_name), Mapping) + ), + None, + ) + current_raw = raw_entry.get(current_phase) if current_phase else None + current_projection = entry.get(current_phase) if current_phase else None + entry["current_outcome"] = ( + dict(current_projection) + if isinstance(current_projection, Mapping) + else None + ) outcome_resolved = False - if isinstance(late_terminal, Mapping): - late_payload = late_terminal.get("payload") + if current_phase in { + "canonical_reconciliation", + "current_state_observation", + } and isinstance( + current_raw, Mapping + ): + outcome_resolved = ( + current_raw.get("outcome_resolved") + if isinstance(current_raw.get("outcome_resolved"), bool) + else bool( + current_raw.get("effect_status") + not in {None, "indeterminate", "unknown"} + and isinstance(current_raw.get("changed"), bool) + ) + ) + elif current_phase == "late_terminal" and isinstance(current_raw, Mapping): + late_payload = current_raw.get("payload") outcome_resolved = bool( - late_terminal.get("outcome") == "late_success" - and late_terminal.get("effect_status") + current_raw.get("outcome") == "late_success" + and current_raw.get("effect_status") not in {None, "indeterminate", "unknown"} and isinstance(late_payload, Mapping) and late_payload.get("mutation_outcome") != "unknown" ) - elif isinstance(turn_terminal, Mapping): - transport = turn_terminal.get("transport") - receipt = turn_terminal.get("receipt") + elif current_phase == "turn_terminal" and isinstance(current_raw, Mapping): + transport = current_raw.get("transport") + receipt = current_raw.get("receipt") outcome_resolved = bool( isinstance(transport, Mapping) and transport.get("outcome") not in {None, "timed_out"} - and turn_terminal.get("effect_status") + and current_raw.get("effect_status") not in {None, "indeterminate", "unknown"} and isinstance(receipt, Mapping) and receipt.get("mutation_outcome") != "unknown" diff --git a/src/backend/integrations/internal_mcp/orchestrator.py b/src/backend/integrations/internal_mcp/orchestrator.py index e06217ae..397e6fea 100644 --- a/src/backend/integrations/internal_mcp/orchestrator.py +++ b/src/backend/integrations/internal_mcp/orchestrator.py @@ -145,6 +145,7 @@ WORKFLOW_RESULT_ENVELOPE_KEY, WORKFLOW_STEP_RESULT_ENVELOPES_KEY, derive_workflow_terminal_status, + normalise_workflow_termination_code, ) from ...workflows.mcp_tool_bridge import ( apply_namespace_to_mcp_payload, @@ -27490,14 +27491,6 @@ def _record_durable_instance_submission_event( finalise_episode_fn = None build_episode_key_fn = None - def _normalise_error_code(error_value: Any) -> str: - if not isinstance(error_value, str) or not error_value.strip(): - return "terminated" - cleaned_error = error_value.strip() - if ":" in cleaned_error: - return cleaned_error.split(":", 1)[0].strip().lower() or "terminated" - return "terminated" - def _record_episode_telemetry( *, completed: bool, @@ -27718,7 +27711,10 @@ def _record_episode_telemetry( termination_code = "completed" termination_detail = None elif error_value: - termination_code = _normalise_error_code(error_value) + termination_code = normalise_workflow_termination_code( + error_value, + default="terminated", + ) termination_detail = explicit_failure_detail or error_value elif _workflow_final_state_is_failure_like(final_state): termination_code = "failed_terminal_state" diff --git a/src/backend/server/routes/von_routes.py b/src/backend/server/routes/von_routes.py index 04c1e584..bc301947 100644 --- a/src/backend/server/routes/von_routes.py +++ b/src/backend/server/routes/von_routes.py @@ -11674,12 +11674,12 @@ def _progress_update(info: dict[str, Any]) -> None: updated_conversation_situation_text = getattr( adaptive_turn_result, "conversation_situation", None ) - effect_finality_fallback = bool( - adaptive_turn_result.effect_finality_fallback - ) + response_authority = str( + getattr(adaptive_turn_result, "response_authority", "model") or "model" + ).strip() + canonical_outcome_response = response_authority == "canonical_outcome" adaptive_partial_delivery = ( adaptive_terminal_status == "effect_partially_completed" - and not effect_finality_fallback and isinstance(response_text, str) and bool(response_text.strip()) ) @@ -11752,9 +11752,9 @@ def _progress_update(info: dict[str, Any]) -> None: { "screen": response_text, "spoken": response_text, - "format": "effect_finality_fallback_v1", + "format": "effect_outcome_report_v1", } - if presenter_mode_requested and effect_finality_fallback + if presenter_mode_requested and canonical_outcome_response else _extract_presenter_channels(response_text) ) current_turn_messages = [{"role": "user", "content": prompt_text}] @@ -11786,7 +11786,7 @@ def _progress_update(info: dict[str, Any]) -> None: default=True ) - if presenter_mode_requested and not effect_finality_fallback: + if presenter_mode_requested and not canonical_outcome_response: screen_tag_present = _presenter_tag_present(response_text, "screen") screen_backfill_screen_tag_present = screen_tag_present required_screen_json_fence = _extract_required_screen_json_fence( @@ -12230,9 +12230,9 @@ def _tool_messages_include_description_write( if not presenter_mode_requested: screen_backfill_status = "skipped" screen_backfill_suppression_reason = "presenter_mode_disabled" - elif effect_finality_fallback: + elif canonical_outcome_response: screen_backfill_status = "skipped" - screen_backfill_suppression_reason = "effect_finality_literal" + screen_backfill_suppression_reason = "canonical_outcome_report" elif not needs_screen_backfill: screen_backfill_status = "skipped" screen_backfill_suppression_reason = "not_required" @@ -12596,9 +12596,9 @@ def _coerce_spoken_text(text: object) -> str | None: if not presenter_mode_requested: spoken_backfill_status = "skipped" spoken_backfill_suppression_reason = "presenter_mode_disabled" - elif effect_finality_fallback: + elif canonical_outcome_response: spoken_backfill_status = "skipped" - spoken_backfill_suppression_reason = "effect_finality_literal" + spoken_backfill_suppression_reason = "canonical_outcome_report" elif not needs_spoken_backfill: spoken_backfill_status = "skipped" spoken_backfill_suppression_reason = "not_required" diff --git a/src/backend/services/adaptive_turn_service.py b/src/backend/services/adaptive_turn_service.py index 06de1086..c0e332aa 100644 --- a/src/backend/services/adaptive_turn_service.py +++ b/src/backend/services/adaptive_turn_service.py @@ -115,7 +115,7 @@ class AdaptiveTurnResult: render_plan: Mapping[str, Any] | None = None terminal_status: str = "completed" evidence_index: Sequence[Mapping[str, Any]] = () - effect_finality_fallback: bool = False + response_authority: str = "model" conversation_situation: str | None = None @@ -3450,8 +3450,10 @@ def _effect_result_target_ids(raw_payload: Any) -> list[str]: "existing_concept_id", "file_copy_concept_id", "instance_id", + "marker_concept_id", "message_id", "object_id", + "paper_concept_id", "relation_id", "source_concept_id", "source_id", @@ -3464,7 +3466,10 @@ def _effect_result_target_ids(raw_payload: Any) -> list[str]: sequence_fields = { "concept_ids", "created_concept_ids", + "paper_concept_ids", "relation_ids", + "represented_artefact_concept_ids", + "represented_artifact_concept_ids", "source_ids", "target_ids", } @@ -3659,6 +3664,7 @@ def _canonical_effect_readback_receipt(raw_payload: Any) -> dict[str, Any] | Non "predicate", "target", "relationship_present", + "relation_present", "inverse_predicate", "inverse_relationship_present", "publication_context", @@ -4013,53 +4019,6 @@ def _cited_ontology_mutation_claim_conflicts( return conflicts -def _cited_ontology_mutation_conflict_response( - conflicts: Sequence[Mapping[str, Any]], -) -> str: - lines = [ - ( - "I cannot safely present the drafted ontology completion claim because " - "it contradicts the cited durable mutation evidence." - ), - "", - "The following cited mutation is not canonically verified:", - ] - for conflict in conflicts: - source_id = str(conflict.get("source_id") or "").strip() - predicate = str(conflict.get("predicate") or "").strip() - target = str(conflict.get("target") or "").strip() - relation = ( - f"`{source_id} --{predicate}--> {target}`" - if source_id and predicate and target - else f"`{conflict.get('method') or 'ontology mutation'}`" - ) - status = str(conflict.get("effect_status") or "unknown") - changed = conflict.get("changed") - evidence_id = str(conflict.get("evidence_id") or "").strip() - effect_id = str(conflict.get("effect_id") or "").strip() - reason = str(conflict.get("reason") or "") - if reason == "canonical_relation_not_verified": - detail = ( - f"handler status `{status}`, but the exact canonical relation " - "read-back is absent or negative" - ) - else: - detail = f"effect status `{status}` and changed `{str(changed).lower()}`" - handle = evidence_id or effect_id - lines.append(f"- {relation}: {detail}; evidence `{handle}`.") - lines.extend( - ( - "", - ( - "The turn remains partial. Sibling mutations must be assessed from " - "their own terminal receipts and exact canonical read-backs; this " - "response does not infer a successful batch count." - ), - ) - ) - return "\n".join(lines) - - def _canonically_verified_material_effect_ids( tool_invocations: Sequence[Mapping[str, Any]], effect_snapshot: Mapping[str, Mapping[str, Any]], @@ -4140,6 +4099,462 @@ def _canonically_verified_material_effect_ids( return material_effect_ids, verified_effect_ids +def _bounded_outcome_text(value: Any, *, limit: int = 500) -> str | None: + if not isinstance(value, str): + return None + cleaned = " ".join(value.split()).strip() + cleaned = cleaned.replace("`", "'").replace("<", "‹").replace(">", "›") + if not cleaned: + return None + if len(cleaned) <= limit: + return cleaned + return f"{cleaned[: max(0, limit - 3)].rstrip()}..." + + +def _nested_mapping_value(value: Any, path: Sequence[str]) -> Any: + current = value + for key in path: + if not isinstance(current, Mapping): + return None + current = current.get(key) + return current + + +def _typed_effect_failure_fact( + payload: Any, + *, + workflow_event: Mapping[str, Any] | None = None, +) -> dict[str, str] | None: + """Project only declared failure fields, never an arbitrary payload walk.""" + + receipt = payload if isinstance(payload, Mapping) else {} + event = workflow_event if isinstance(workflow_event, Mapping) else {} + error_code = _bounded_outcome_text( + event.get("error_code") or receipt.get("error_code"), + limit=160, + ) + error = _bounded_outcome_text( + event.get("error") + or receipt.get("failure_reason") + or receipt.get("error") + ) + if not error_code and not error: + return None + return { + **({"error_code": error_code} if error_code else {}), + **({"error": error} if error else {}), + } + + +def _effect_failure_fact(state: Mapping[str, Any]) -> tuple[str | None, str | None]: + """Return the bounded typed cause projected at the capability boundary.""" + + failure_fact = state.get("failure_fact") + if not isinstance(failure_fact, Mapping): + return None, None + return ( + _bounded_outcome_text(failure_fact.get("error_code"), limit=160), + _bounded_outcome_text(failure_fact.get("error")), + ) + + +def _effect_scope_fact( + state: Mapping[str, Any], + *, + trusted_scope: TrustedTurnScope, +) -> dict[str, str] | None: + canonical_scope = state.get("canonical_scope") + if isinstance(canonical_scope, Mapping): + mode = str(canonical_scope.get("mode") or "").strip().lower() + concept_id = str(canonical_scope.get("concept_id") or "").strip() + if mode in {"user", "organisation", "global"} and ( + concept_id or mode == "global" + ): + return {"mode": mode, "concept_id": concept_id} + + readback = state.get("canonical_readback") + publication_context = ( + readback.get("publication_context") + if isinstance(readback, Mapping) + else None + ) + if isinstance(publication_context, Mapping): + mode = str(publication_context.get("kind") or "").strip().lower() + concept_id = str(publication_context.get("concept_id") or "").strip() + if mode in {"user", "organisation", "global"} and ( + concept_id or mode == "global" + ): + return {"mode": mode, "concept_id": concept_id} + + if isinstance(readback, Mapping): + scope_mode = str(readback.get("scope_mode") or "").strip().lower() + if readback.get("trusted_scope_identity_sha256") and scope_mode == "user": + concept_id = str(trusted_scope.user_concept_id or "").strip() + return {"mode": "user", "concept_id": concept_id} if concept_id else None + if ( + readback.get("trusted_scope_identity_sha256") + and scope_mode == "organisation" + ): + concept_id = str(trusted_scope.organisation_concept_id or "").strip() + return ( + {"mode": "organisation", "concept_id": concept_id} + if concept_id + else None + ) + + return None + + +def _build_effect_outcome_report( + *, + terminal_status: str, + effect_snapshot: Mapping[str, Mapping[str, Any]], + tool_invocations: Sequence[Mapping[str, Any]], + trusted_scope: TrustedTurnScope, +) -> tuple[str, dict[str, Any]]: + """Render acknowledged effect facts without asking another answer author.""" + + invocation_by_effect_id = { + str(invocation.get("effect_id")): invocation + for invocation in tool_invocations + if isinstance(invocation.get("effect_id"), str) + and str(invocation.get("effect_id")).strip() + } + facts: list[dict[str, Any]] = [] + scope_facts: list[dict[str, str]] = [] + for effect_id, state in effect_snapshot.items(): + if state.get("turn_finality_required") is False: + continue + invocation = invocation_by_effect_id.get(effect_id, {}) + effect_status = str(state.get("effect_status") or "unknown").strip() + if not ( + effect_status in {"failed", "partial", "indeterminate", "not_started"} + or state.get("changed") is True + or (effect_status == "succeeded" and state.get("changed") is None) + or state.get("outcome_resolved") is True + ): + continue + tool_name = str( + invocation.get("capability_display_name") + or state.get("workflow_id") + or invocation.get("tool") + or state.get("execution_method") + or "effect" + ).strip() + target_ids = [ + str(item).strip() + for item in ( + state.get("result_target_ids") + or invocation.get("result_target_ids") + or () + ) + if isinstance(item, str) and item.strip() + ][:8] + evidence = invocation.get("evidence") + evidence_id = ( + str(evidence.get("evidence_id") or "").strip() + if isinstance(evidence, Mapping) + else "" + ) + error_code, error_detail = _effect_failure_fact(state) + if not error_code: + error_code = _bounded_outcome_text(invocation.get("error_code"), limit=160) + canonical_readback = state.get("canonical_readback") + execution_method = str( + invocation.get("execution_method") + or invocation.get("tool") + or state.get("execution_method") + or "" + ).strip() + if execution_method in _ONTOLOGY_RELATION_POSTCONDITION_METHODS: + canonical_readback_verified = ( + _canonical_relation_readback_matches_invocation( + invocation, + canonical_readback, + ) + ) + elif execution_method == "upsert_scoped_assertion": + canonical_readback_verified = ( + _canonical_scoped_assertion_readback_matches_invocation( + invocation, + canonical_readback, + ) + ) + else: + canonical_readback_verified = bool( + isinstance(canonical_readback, Mapping) + and ( + canonical_readback.get("verified") is True + or str(canonical_readback.get("status") or "") + .strip() + .lower() + == "verified" + or canonical_readback.get("assertion_present") is True + ) + ) + effective_arguments = state.get("effective_arguments") + argument_identity: dict[str, str] = {} + if isinstance(effective_arguments, Mapping): + for field in ("concept_id", "source_id", "predicate", "target"): + bounded = _bounded_outcome_text( + effective_arguments.get(field), + limit=240, + ) + if bounded: + argument_identity[field] = bounded + fact = { + "effect_id": effect_id, + "tool": tool_name, + "effect_status": effect_status, + "changed": state.get("changed") if isinstance(state.get("changed"), bool) else None, + "initial_effect_status": state.get("initial_effect_status"), + "current_outcome_status": state.get("current_outcome_status"), + "outcome_resolved": state.get("outcome_resolved") is True, + "reconciliation_status": state.get("reconciliation_status"), + "canonical_readback_present": isinstance( + canonical_readback, Mapping + ), + "canonical_readback_verified": canonical_readback_verified, + "workflow_id": state.get("workflow_id"), + "instance_id": state.get("instance_id"), + "target_ids": target_ids, + "evidence_id": evidence_id or state.get("reconciliation_evidence_id"), + "error_code": error_code, + "error": error_detail, + "recovered_by_effect_id": state.get("recovered_by_effect_id"), + "argument_identity": argument_identity or None, + } + facts.append({key: value for key, value in fact.items() if value is not None}) + scope_fact = _effect_scope_fact(state, trusted_scope=trusted_scope) + if scope_fact and scope_fact not in scope_facts: + scope_facts.append(scope_fact) + + heading = { + "effect_partially_completed": "This turn completed only partially.", + "effect_outcome_indeterminate": ( + "This turn has at least one unresolved effect outcome." + ), + "effect_failed": "This turn did not complete all requested effects.", + "effect_not_started": "At least one requested effect was not started.", + "model_error": "The model did not produce a reliable final answer.", + "model_non_answer": "The model did not produce a usable final answer.", + }.get(terminal_status, "This turn did not finish cleanly.") + lines = ["## Effect outcome report", "", heading] + confirmed = [ + fact + for fact in facts + if fact.get("effect_status") == "succeeded" + or fact.get("outcome_resolved") is True + or fact.get("recovered_by_effect_id") + ] + unresolved = [fact for fact in facts if fact not in confirmed] + + if confirmed: + lines.extend( + ( + "", + "### Verified, observed, recovered, or handler-reported", + ) + ) + for fact in confirmed: + label = f"`{fact['tool']}`" + status = str(fact.get("effect_status") or "unknown") + if fact.get("outcome_resolved") and status != "succeeded": + if fact.get("current_outcome_status") == "target_absent": + detail = ( + "exact read-back found the target absent; the original " + f"attempt receipt remains `{status}`" + ) + else: + detail = ( + "the current target was read back exactly; the original " + f"attempt receipt remains `{status}`" + ) + if fact.get("error_code"): + detail += f" (`{fact['error_code']}`)" + if fact.get("recovered_by_effect_id"): + detail += ( + "; a later exact retry succeeded as effect " + f"`{fact['recovered_by_effect_id']}`" + ) + elif fact.get("reconciliation_status") == "canonically_verified": + detail = "succeeded after exact canonical reconciliation" + elif fact.get("effect_status") == "succeeded": + if fact.get("canonical_readback_verified"): + detail = "succeeded with canonical read-back" + elif fact.get("canonical_readback_present"): + detail = ( + "the handler reported `succeeded`, but the embedded " + "canonical read-back did not verify the outcome" + ) + else: + detail = "reported `succeeded`" + else: + detail = f"receipt `{status}` was recovered by a later effect" + handles = [] + if fact.get("instance_id"): + handles.append(f"instance `{fact['instance_id']}`") + if fact.get("target_ids"): + handles.append( + "target" + ("s" if len(fact["target_ids"]) != 1 else "") + + " " + + ", ".join(f"`{item}`" for item in fact["target_ids"]) + ) + if fact.get("evidence_id"): + handles.append(f"evidence `{fact['evidence_id']}`") + if fact.get("error"): + handles.append(f"cause: {fact['error']}") + suffix = f"; {'; '.join(handles)}" if handles else "" + lines.append(f"- {label}: {detail}{suffix}.") + + if unresolved: + lines.extend(("", "### Unsuccessful or unresolved")) + for fact in unresolved: + label = f"`{fact['tool']}`" + status = str(fact.get("effect_status") or "unknown") + details = [f"status `{status}`"] + if fact.get("changed") is False: + details.append("reported no change") + elif fact.get("changed") is True: + details.append("reported a possible or partial change") + if fact.get("error_code"): + details.append(f"error code `{fact['error_code']}`") + if fact.get("error"): + details.append(str(fact["error"])) + argument_identity = fact.get("argument_identity") + if isinstance(argument_identity, Mapping): + relation_parts = [ + f"{key} `{value}`" + for key, value in argument_identity.items() + ] + if relation_parts: + details.append(", ".join(relation_parts)) + if fact.get("instance_id"): + details.append(f"instance `{fact['instance_id']}`") + if fact.get("evidence_id"): + details.append(f"evidence `{fact['evidence_id']}`") + lines.append(f"- {label}: {'; '.join(details)}.") + + if scope_facts: + lines.extend(("", "### Canonical scope")) + for scope_fact in scope_facts: + mode = scope_fact["mode"] + concept_id = scope_fact.get("concept_id") or "" + if mode == "user": + lines.append( + f"- User scope `{concept_id}`. The organisation-qualified " + f"namespace `{trusted_scope.namespace}` is execution context, " + "not organisation publication." + ) + elif mode == "organisation": + lines.append(f"- Organisation scope `{concept_id}`.") + else: + lines.append("- Global publication scope.") + + if any( + fact.get("effect_status") in {"partial", "indeterminate"} + and not fact.get("outcome_resolved") + for fact in facts + ): + lines.extend( + ( + "", + "Inspect the named current state before retrying an unresolved effect.", + ) + ) + report = { + "schema_version": "adaptive_turn_effect_outcome_report.v1", + "terminal_status": terminal_status, + "facts": facts, + "canonical_scopes": scope_facts, + } + return "\n".join(lines), report + + +def _canonical_scope_from_exact_concept_read( + raw_payload: Any, + *, + trusted_scope: TrustedTurnScope, +) -> dict[str, str] | None: + """Project actor scope only when the exact concept read proves it.""" + + if not isinstance(raw_payload, Mapping): + return None + publication_context = raw_payload.get("publication_context") + if isinstance(publication_context, Mapping): + mode = str(publication_context.get("kind") or "").strip().lower() + concept_id = str(publication_context.get("concept_id") or "").strip() + if mode == "user" and concept_id == trusted_scope.user_concept_id: + return {"mode": "user", "concept_id": concept_id} + if ( + mode == "organisation" + and concept_id == trusted_scope.organisation_concept_id + ): + return {"mode": "organisation", "concept_id": concept_id} + + relationships = raw_payload.get("relationships") + if not isinstance(relationships, Mapping): + return None + user_targets = relationships.get("#V#specific_to_user") + if not isinstance(user_targets, Sequence) or isinstance( + user_targets, (str, bytes, bytearray) + ): + user_targets = relationships.get("specific_to_user") + if isinstance(user_targets, Sequence) and not isinstance( + user_targets, (str, bytes, bytearray) + ): + clean_targets = { + str(item).strip() for item in user_targets if str(item).strip() + } + if trusted_scope.user_concept_id in clean_targets: + return { + "mode": "user", + "concept_id": trusted_scope.user_concept_id, + } + return None + + +def _canonical_reconciliation_observation_identity( + *, + effect_id: str, + basis: str, + method_name: str, + target_ids: Sequence[str], + receipt_id: Any = None, + intent_fingerprint: Any = None, + read_call_id: Any = None, + canonical_scope: Mapping[str, Any] | None = None, + terminal_status: Any = None, +) -> str: + """Identify one exact immutable reconciliation independently of timestamps.""" + + return hashlib.sha256( + _json_bytes( + { + "effect_id": effect_id, + "basis": basis, + "method_name": method_name, + "target_ids": sorted( + str(item).strip() + for item in target_ids + if isinstance(item, str) and item.strip() + ), + "receipt_id": str(receipt_id or "").strip() or None, + "intent_fingerprint": ( + str(intent_fingerprint or "").strip() or None + ), + "read_call_id": str(read_call_id or "").strip() or None, + "canonical_scope": ( + dict(canonical_scope) + if isinstance(canonical_scope, Mapping) + else None + ), + "terminal_status": str(terminal_status or "").strip() or None, + } + ) + ).hexdigest() + + def _effect_status( raw_payload: Any, *, @@ -4485,6 +4900,13 @@ def _build_represented_workflow_execution_event( latest_step = dict(latest_step_raw) if isinstance(latest_step_raw, Mapping) else {} diagnostics_raw = latest_step.get("diagnostics") diagnostics = dict(diagnostics_raw) if isinstance(diagnostics_raw, Mapping) else {} + nested_mcp_result = _nested_mapping_value( + latest_step, + ("output_payload", "mcp_result"), + ) + nested_mcp_result = ( + nested_mcp_result if isinstance(nested_mcp_result, Mapping) else {} + ) progress_evidence = project_nested_workflow_progress_evidence(receipt) progress_facts = ( @@ -4524,14 +4946,25 @@ def _build_represented_workflow_execution_event( else: event_status = "workflow_execution_failed" + error_code = ( + _bounded_outcome_text(nested_mcp_result.get("error_code"), limit=160) + or _bounded_outcome_text(latest_step.get("error_code"), limit=160) + or _bounded_outcome_text( + workflow_execution.get("error_code"), limit=160 + ) + or _bounded_outcome_text(receipt.get("error_code"), limit=160) + ) error = ( - _bounded_workflow_progress_text( + _bounded_outcome_text(nested_mcp_result.get("error"), limit=320) + or _bounded_outcome_text( + nested_mcp_result.get("failure_reason"), limit=320 + ) + or _bounded_outcome_text( workflow_execution.get("failure_reason"), limit=320 ) - or _bounded_workflow_progress_text(receipt.get("failure_reason"), limit=320) - or _bounded_workflow_progress_text(diagnostics.get("error"), limit=320) - or _bounded_workflow_progress_text(workflow_execution.get("error"), limit=320) - or _bounded_workflow_progress_text(receipt.get("error_code"), limit=160) + or _bounded_outcome_text(receipt.get("failure_reason"), limit=320) + or _bounded_outcome_text(diagnostics.get("error"), limit=320) + or _bounded_outcome_text(workflow_execution.get("error"), limit=320) ) recovery_affordances = receipt.get("recovery_affordances") next_action = None @@ -4566,6 +4999,7 @@ def _build_represented_workflow_execution_event( "final_state": workflow_execution.get("current_state") or workflow_instance.get("current_state"), "error": error, + "error_code": error_code, "effect_status": effect_status, "mutation_outcome": receipt.get("mutation_outcome"), "outcome_finality": receipt.get("outcome_finality"), @@ -4613,44 +5047,6 @@ def _usage_add( totals[str(key)] = totals.get(str(key), 0.0) + float(value) -def _can_preserve_pending_durable_response( - text: str, - effects: Sequence[Mapping[str, Any]], -) -> bool: - """Return whether a useful, honest pending-workflow answer can be shown.""" - - cleaned = text.strip() - if not cleaned or not effects: - return False - lowered = cleaned.lower() - if not any( - marker in lowered - for marker in ( - "pending", - "queued", - "running", - "paused", - "partial", - "not yet", - "not complete", - "not finished", - "still processing", - "in progress", - ) - ): - return False - - for effect in effects: - if effect.get("capability_kind") != "represented_workflow": - return False - if effect.get("effect_status") != "partial": - return False - instance_id = str(effect.get("instance_id") or "").strip() - if not instance_id or instance_id not in cleaned: - return False - return True - - def execute_adaptive_turn( *, gateway: InternalMCPGateway | None, @@ -4886,10 +5282,13 @@ def emit_model_call_end(call: Mapping[str, Any]) -> None: seen_evidence_view_digests: set[str] = set() effect_state_lock = threading.RLock() effect_states: dict[str, dict[str, Any]] = {} + exact_read_observations: list[dict[str, Any]] = [] effect_state_generation = 0 last_partial_effect_generation = 0 successful_effect_mutation_generation = 0 terminal_failed_effect_requests: dict[str, tuple[int, str | None]] = {} + indeterminate_effect_requests: dict[str, tuple[str, str | None]] = {} + exact_absence_effect_requests: dict[str, str] = {} recoverable_effect_ids: dict[str, list[str]] = {} def scoped_assertion_recovery_key( @@ -5031,6 +5430,34 @@ def reconcile_successful_recovery( failed_state["recovery_status"] = "succeeded" effect_state_generation += 1 + def reconcile_successful_exact_absence_retry( + *, + recovery_effect_id: str, + request_signature: str | None, + effect_status: str, + ) -> None: + nonlocal effect_state_generation + if effect_status != "succeeded" or not request_signature: + return + prior_effect_id = exact_absence_effect_requests.pop( + request_signature, + None, + ) + if prior_effect_id is None: + return + with effect_state_lock: + prior_state = effect_states.get(prior_effect_id) + if ( + prior_state is None + or prior_state.get("effect_status") != "indeterminate" + or prior_state.get("outcome_resolved") is not True + or prior_state.get("current_outcome_status") != "target_absent" + ): + return + prior_state["recovered_by_effect_id"] = recovery_effect_id + prior_state["recovery_status"] = "succeeded" + effect_state_generation += 1 + def remember_effect_state( effect_id: str, *, @@ -5048,7 +5475,10 @@ def remember_effect_state( execution_method: str | None = None, effective_arguments: Mapping[str, Any] | None = None, original_result: Mapping[str, Any] | None = None, + failure_fact: Mapping[str, Any] | None = None, + effect_request_signature: str | None = None, result_target_ids: Sequence[str] | None = None, + observation_order: int | None = None, ) -> None: nonlocal effect_state_generation with effect_state_lock: @@ -5066,11 +5496,15 @@ def remember_effect_state( "workflow_id": workflow_id, "evidence_id": evidence_id, "execution_method": execution_method, + "effect_request_signature": effect_request_signature, + "observation_order": observation_order, } if effective_arguments is not None: state["effective_arguments"] = dict(effective_arguments) if original_result is not None: state["original_result"] = dict(original_result) + if failure_fact is not None: + state["failure_fact"] = dict(failure_fact) if result_target_ids: state["result_target_ids"] = [ str(item).strip() @@ -5105,7 +5539,10 @@ def remember_effect_state( "execution_method", "effective_arguments", "original_result", + "failure_fact", + "effect_request_signature", "result_target_ids", + "observation_order", ): if not state.get(identity_field) and existing.get(identity_field): state[identity_field] = existing[identity_field] @@ -5119,8 +5556,116 @@ def remember_effect_state( effect_states[effect_id] = state effect_state_generation += 1 + def _record_reconciliation_persistence_failure( + *, + effect_id: str, + reason: Any, + phase: str = "canonical_reconciliation", + ) -> None: + aux_calls.append( + { + "type": "effect_observation_persistence_failure", + "schema_version": "effect_observation_persistence_failure.v1", + "effect_id": effect_id, + "phase": phase, + "reason": str(reason or "not_acknowledged"), + } + ) + + def _matching_exact_current_state_observation( + candidate: Mapping[str, Any], + ) -> dict[str, Any] | None: + if str(candidate.get("execution_method") or "").strip() != "create_concepts": + return None + candidate_targets = { + str(item).strip() + for item in candidate.get("result_target_ids") or () + if isinstance(item, str) and item.strip() + } + targets_derived_from_arguments = False + if not candidate_targets: + arguments = candidate.get("effective_arguments") + concepts = ( + arguments.get("concepts") + if isinstance(arguments, Mapping) + else None + ) + if isinstance(concepts, Sequence) and not isinstance( + concepts, + (str, bytes, bytearray), + ) and len(concepts) == 1: + concept = concepts[0] + concept_id = ( + str(concept.get("concept_id") or "").strip() + if isinstance(concept, Mapping) + else "" + ) + if concept_id: + candidate_targets = {concept_id} + targets_derived_from_arguments = True + candidate_order = candidate.get("observation_order") + if not candidate_targets or not isinstance(candidate_order, int): + return None + arguments = candidate.get("effective_arguments") + expected_scope_mode = ( + str(arguments.get("scope_mode") or "").strip().lower() + if isinstance(arguments, Mapping) + else "" + ) + if expected_scope_mode not in {"", "user", "user_only_default"}: + return None + with effect_state_lock: + observations = [dict(item) for item in exact_read_observations] + for observation in sorted( + observations, + key=lambda item: int(item.get("observation_order") or -1), + ): + if observation.get("capability_name") != "fetch_concept": + continue + observation_order = observation.get("observation_order") + if not isinstance(observation_order, int) or observation_order <= candidate_order: + continue + observed_targets = { + str(item).strip() + for item in observation.get("result_target_ids") or () + if isinstance(item, str) and item.strip() + } + exact_requested_targets = { + str(item).strip() + for item in observation.get("requested_target_ids") or () + if isinstance(item, str) and item.strip() + } + matched_targets = candidate_targets.intersection( + observed_targets, + exact_requested_targets, + ) + target_state = str( + observation.get("target_state") or "present" + ).strip().lower() + canonical_scope = observation.get("canonical_scope") + if len(candidate_targets) != 1: + continue + if target_state == "absent": + if candidate_targets != exact_requested_targets: + continue + matched_targets = set(candidate_targets) + else: + if targets_derived_from_arguments: + continue + if matched_targets != candidate_targets or not ( + isinstance(canonical_scope, Mapping) + and canonical_scope.get("mode") == "user" + and canonical_scope.get("concept_id") == scope.user_concept_id + ): + continue + return { + **observation, + "matched_target_ids": sorted(matched_targets), + } + return None + def reconcile_indeterminate_ontology_effects() -> None: - """Resolve exact ontology postconditions once more before terminal output.""" + """Resolve current state only after its actor-scoped journal write acks.""" nonlocal effect_state_generation with effect_state_lock: @@ -5128,18 +5673,26 @@ def reconcile_indeterminate_ontology_effects() -> None: (effect_id, dict(state)) for effect_id, state in effect_states.items() if state.get("effect_status") == "indeterminate" + and state.get("outcome_resolved") is not True and state.get("turn_finality_required") is not False and isinstance(state.get("original_result"), Mapping) - and state["original_result"].get("outcome_finality") - == "requires_canonical_reconciliation" - and isinstance( - state["original_result"].get("postcondition_reconciliation"), - Mapping, - ) - and state["original_result"]["postcondition_reconciliation"].get( - "schema_version" + and ( + state.get("execution_method") == "create_concepts" + or ( + state["original_result"].get("outcome_finality") + == "requires_canonical_reconciliation" + and isinstance( + state["original_result"].get( + "postcondition_reconciliation" + ), + Mapping, + ) + and state["original_result"][ + "postcondition_reconciliation" + ].get("schema_version") + == "ontology_mutation_postcondition_reconciliation.v1" + ) ) - == "ontology_mutation_postcondition_reconciliation.v1" ] if not candidates: return @@ -5155,74 +5708,258 @@ def reconcile_indeterminate_ontology_effects() -> None: original_result, Mapping ): continue - try: - with override_current_actor( - scope.user_concept_id, - scope.organisation_concept_id, + postcondition = original_result.get("postcondition_reconciliation") + governed_reconciliation_available = bool( + original_result.get("outcome_finality") + == "requires_canonical_reconciliation" + and isinstance(postcondition, Mapping) + and postcondition.get("schema_version") + == "ontology_mutation_postcondition_reconciliation.v1" + ) + reconciliation: Mapping[str, Any] = {} + if governed_reconciliation_available: + try: + with override_current_actor( + scope.user_concept_id, + scope.organisation_concept_id, + ): + reconciliation = reconcile_governed_ontology_postcondition( + method_name=method_name, + arguments=arguments, + original_result=original_result, + ) + except Exception as exc: # noqa: BLE001 + reconciliation = { + "success": False, + "verified": False, + "error_code": "ontology_reconciliation_unavailable", + "exception_type": type(exc).__name__, + } + verified = reconciliation.get("verified") is True + if governed_reconciliation_available: + aux_calls.append( + { + "type": ( + "adaptive_turn_ontology_postcondition_reconciliation" + ), + "schema_version": ( + "adaptive_turn_ontology_postcondition_reconciliation.v1" + ), + "effect_id": effect_id, + "method_name": method_name, + "verified": verified, + "receipt_id": reconciliation.get("receipt_id"), + "error_code": reconciliation.get("error_code"), + } + ) + if verified: + envelope = evidence_store.record( + f"{method_name}_canonical_reconciliation", + f"{effect_id}:canonical_reconciliation", + reconciliation, + provenance={ + "namespace": scope.namespace, + "user_concept_id": scope.user_concept_id, + "organisation_concept_id": scope.organisation_concept_id, + "effect_id": effect_id, + "effect_status": "succeeded", + "reconciliation": True, + }, + status="succeeded", + ) + target_ids = [ + str(item).strip() + for item in reconciliation.get("target_concept_ids") or () + if isinstance(item, str) and str(item).strip() + ] + identity = _canonical_reconciliation_observation_identity( + effect_id=effect_id, + basis="governed_postcondition", + method_name=method_name, + target_ids=target_ids, + receipt_id=reconciliation.get("receipt_id"), + intent_fingerprint=reconciliation.get("intent_fingerprint"), + ) + observation = { + "schema_version": ( + "ontology_mutation_canonical_reconciliation.v1" + ), + "effect_status": "succeeded", + "changed": True, + "current_outcome_status": "succeeded", + "outcome_resolved": True, + "reconciliation_basis": "governed_postcondition", + "observation_identity_sha256": identity, + "method_name": method_name, + "receipt_id": reconciliation.get("receipt_id"), + "intent_fingerprint": reconciliation.get( + "intent_fingerprint" + ), + "target_concept_ids": target_ids, + "evidence_id": envelope.evidence_id, + } + try: + phase_outcome = persist_effect_observation_phase( + effect_id=effect_id, + phase="canonical_reconciliation", + observation=observation, + ) + except Exception as exc: # noqa: BLE001 + _record_reconciliation_persistence_failure( + effect_id=effect_id, + reason=type(exc).__name__, + ) + continue + if not _effect_phase_acknowledged( + phase_outcome, + expected_identity=identity, ): - reconciliation = reconcile_governed_ontology_postcondition( - method_name=method_name, - arguments=arguments, - original_result=original_result, + _record_reconciliation_persistence_failure( + effect_id=effect_id, + reason=phase_outcome.get("reason"), ) - except Exception as exc: # noqa: BLE001 - unresolved remains indeterminate - reconciliation = { - "success": False, - "verified": False, - "error_code": "ontology_reconciliation_unavailable", - "exception_type": type(exc).__name__, - } - verified = reconciliation.get("verified") is True - aux_calls.append( - { - "type": "adaptive_turn_ontology_postcondition_reconciliation", + continue + stored_phase = phase_outcome.get("stored_phase") + stored_evidence_id = ( + stored_phase.get("evidence_id") + if isinstance(stored_phase, Mapping) + else None + ) + canonical_projection = { "schema_version": ( - "adaptive_turn_ontology_postcondition_reconciliation.v1" + "ontology_mutation_canonical_reconciliation.v1" ), - "effect_id": effect_id, + "status": "verified", + "verified": True, "method_name": method_name, - "verified": verified, "receipt_id": reconciliation.get("receipt_id"), - "error_code": reconciliation.get("error_code"), + "intent_fingerprint": reconciliation.get( + "intent_fingerprint" + ), + "target_concept_ids": target_ids, + "evidence_id": stored_evidence_id or envelope.evidence_id, } + with effect_state_lock: + current = effect_states.get(effect_id) + if ( + current is None + or current.get("effect_status") != "indeterminate" + or int(current.get("phase") or 0) > 2 + ): + continue + current.update( + { + "phase": 2, + "initial_effect_status": "indeterminate", + "initial_changed": current.get("changed"), + "effect_status": "succeeded", + "changed": True, + "current_outcome_status": "succeeded", + "outcome_resolved": True, + "recovery_status": "succeeded", + "reconciliation_status": "canonically_verified", + "reconciliation_basis": "governed_postcondition", + "reconciliation_evidence_id": ( + stored_evidence_id or envelope.evidence_id + ), + "canonical_readback": canonical_projection, + "result_target_ids": target_ids + or list(current.get("result_target_ids") or ()), + } + ) + effect_state_generation += 1 + continue + + exact_observation = _matching_exact_current_state_observation(candidate) + if exact_observation is None: + continue + target_ids = list(exact_observation["matched_target_ids"]) + target_state = str( + exact_observation.get("target_state") or "present" + ).strip().lower() + raw_canonical_scope = exact_observation.get("canonical_scope") + canonical_scope = ( + dict(raw_canonical_scope) + if isinstance(raw_canonical_scope, Mapping) + else None + ) + identity = _canonical_reconciliation_observation_identity( + effect_id=effect_id, + basis="later_exact_current_state_read", + method_name=method_name, + target_ids=target_ids, + read_call_id=exact_observation.get("call_id"), + canonical_scope=canonical_scope, ) - if not verified: + observation = { + "schema_version": "ontology_mutation_current_state_observation.v1", + "effect_status": "indeterminate", + "changed": candidate.get("changed"), + "initial_effect_status": "indeterminate", + "current_outcome_status": ( + "target_absent" if target_state == "absent" else "target_observed" + ), + "outcome_resolved": True, + "reconciliation_basis": "later_exact_current_state_read", + "observation_identity_sha256": identity, + "method_name": method_name, + "read_method_name": exact_observation.get("capability_name"), + "read_call_id": exact_observation.get("call_id"), + "target_concept_ids": target_ids, + "evidence_id": exact_observation.get("evidence_id"), + } + if canonical_scope is not None: + observation["canonical_scope"] = canonical_scope + try: + phase_outcome = persist_effect_observation_phase( + effect_id=effect_id, + phase="current_state_observation", + observation=observation, + ) + except Exception as exc: # noqa: BLE001 + _record_reconciliation_persistence_failure( + effect_id=effect_id, + reason=type(exc).__name__, + phase="current_state_observation", + ) continue - envelope = evidence_store.record( - f"{method_name}_canonical_reconciliation", - f"{effect_id}:canonical_reconciliation", - reconciliation, - provenance={ - "namespace": scope.namespace, - "user_concept_id": scope.user_concept_id, - "organisation_concept_id": scope.organisation_concept_id, - "effect_id": effect_id, - "effect_status": "succeeded", - "reconciliation": True, - }, - status="succeeded", + if not _effect_phase_acknowledged( + phase_outcome, + expected_identity=identity, + ): + _record_reconciliation_persistence_failure( + effect_id=effect_id, + reason=phase_outcome.get("reason"), + phase="current_state_observation", + ) + continue + stored_phase = phase_outcome.get("stored_phase") + stored_evidence_id = ( + stored_phase.get("evidence_id") + if isinstance(stored_phase, Mapping) + else None ) - target_ids = [ - str(item).strip() - for item in reconciliation.get("target_concept_ids") or () - if isinstance(item, str) and str(item).strip() - ] canonical_projection = { - "schema_version": "ontology_mutation_canonical_reconciliation.v1", - "status": "verified", + "schema_version": "ontology_mutation_current_state_observation.v1", + "status": ( + "target_absent" if target_state == "absent" else "target_observed" + ), "verified": True, "method_name": method_name, - "receipt_id": reconciliation.get("receipt_id"), - "intent_fingerprint": reconciliation.get("intent_fingerprint"), + "read_method_name": exact_observation.get("capability_name"), + "read_call_id": exact_observation.get("call_id"), "target_concept_ids": target_ids, - "evidence_id": envelope.evidence_id, + "evidence_id": stored_evidence_id + or exact_observation.get("evidence_id"), } + if canonical_scope is not None: + canonical_projection["canonical_scope"] = canonical_scope with effect_state_lock: current = effect_states.get(effect_id) if ( current is None or current.get("effect_status") != "indeterminate" - or int(current.get("phase") or 0) > 2 + or current.get("outcome_resolved") is True ): continue current.update( @@ -5230,53 +5967,40 @@ def reconcile_indeterminate_ontology_effects() -> None: "phase": 2, "initial_effect_status": "indeterminate", "initial_changed": current.get("changed"), - "effect_status": "succeeded", - "changed": True, - "recovery_status": "succeeded", - "reconciliation_status": "canonically_verified", - "reconciliation_evidence_id": envelope.evidence_id, - "canonical_readback": canonical_projection, - "result_target_ids": target_ids - or list(current.get("result_target_ids") or ()), - } - ) - effect_state_generation += 1 - try: - persist_effect_observation_phase( - effect_id=effect_id, - phase="canonical_reconciliation", - observation={ - "schema_version": ( - "ontology_mutation_canonical_reconciliation.v1" + "current_outcome_status": ( + "target_absent" + if target_state == "absent" + else "target_observed" ), - "effect_status": "succeeded", - "changed": True, - "method_name": method_name, - "receipt_id": reconciliation.get("receipt_id"), - "intent_fingerprint": reconciliation.get( - "intent_fingerprint" + "outcome_resolved": True, + "reconciliation_status": "current_state_observed", + "reconciliation_basis": ( + "later_exact_current_state_read" ), - "target_concept_ids": target_ids, - "evidence_id": envelope.evidence_id, - }, - ) - except Exception as exc: # noqa: BLE001 - receipt is already durable - aux_calls.append( - { - "type": "effect_observation_persistence_failure", - "schema_version": ( - "effect_observation_persistence_failure.v1" + "reconciliation_evidence_id": ( + stored_evidence_id + or exact_observation.get("evidence_id") ), - "effect_id": effect_id, - "phase": "canonical_reconciliation", - "reason": type(exc).__name__, + "canonical_readback": canonical_projection, + "result_target_ids": target_ids, } ) + if canonical_scope is not None: + current["canonical_scope"] = canonical_scope + effect_state_generation += 1 + if target_state == "absent": + request_signature = str( + candidate.get("effect_request_signature") or "" + ).strip() + if request_signature: + indeterminate_effect_requests.pop(request_signature, None) + exact_absence_effect_requests[request_signature] = effect_id def reconcile_workflow_instance_readback( *, capability_name: str, payload: Any, + call_id: str, evidence_id: str | None, ) -> None: """Resolve an earlier workflow effect from an exact canonical instance read.""" @@ -5307,27 +6031,90 @@ def reconcile_workflow_instance_readback( else "failed" ) with effect_state_lock: - for state in effect_states.values(): - if state.get("capability_kind") != "represented_workflow": - continue - if state.get("instance_id") != instance_id: - continue - state_workflow_id = str(state.get("workflow_id") or "").strip() - if ( - workflow_id - and state_workflow_id - and workflow_id != state_workflow_id - ): - continue - if ( - state.get("effect_status") == reconciled_status - and int(state.get("phase") or 0) >= 2 - ): + candidates = [ + (effect_id, dict(state)) + for effect_id, state in effect_states.items() + if state.get("capability_kind") == "represented_workflow" + and state.get("instance_id") == instance_id + ] + for effect_id, candidate in candidates: + state_workflow_id = str(candidate.get("workflow_id") or "").strip() + if ( + workflow_id + and state_workflow_id + and workflow_id != state_workflow_id + ): + continue + if ( + candidate.get("effect_status") == reconciled_status + and int(candidate.get("phase") or 0) >= 2 + and candidate.get("outcome_resolved") is True + ): + continue + identity = _canonical_reconciliation_observation_identity( + effect_id=effect_id, + basis="workflow_instance_terminal_read", + method_name="workflow_get_instance", + target_ids=[instance_id], + read_call_id=call_id, + terminal_status=terminal_status, + ) + observation = { + "schema_version": "workflow_instance_canonical_reconciliation.v1", + "effect_status": reconciled_status, + "changed": candidate.get("changed"), + "initial_effect_status": candidate.get("effect_status"), + "current_outcome_status": reconciled_status, + "outcome_resolved": True, + "reconciliation_basis": "workflow_instance_terminal_read", + "observation_identity_sha256": identity, + "method_name": "workflow_get_instance", + "read_call_id": call_id, + "instance_id": instance_id, + "workflow_id": workflow_id or state_workflow_id or None, + "terminal_status": terminal_status, + "evidence_id": evidence_id, + } + try: + phase_outcome = persist_effect_observation_phase( + effect_id=effect_id, + phase="canonical_reconciliation", + observation=observation, + ) + except Exception as exc: # noqa: BLE001 + _record_reconciliation_persistence_failure( + effect_id=effect_id, + reason=type(exc).__name__, + ) + continue + if not _effect_phase_acknowledged( + phase_outcome, + expected_identity=identity, + ): + _record_reconciliation_persistence_failure( + effect_id=effect_id, + reason=phase_outcome.get("reason"), + ) + continue + with effect_state_lock: + state = effect_states.get(effect_id) + if state is None: continue + initial_status = state.get("effect_status") + initial_changed = state.get("changed") state.update( { "phase": 2, + "initial_effect_status": initial_status, + "initial_changed": initial_changed, "effect_status": reconciled_status, + "current_outcome_status": reconciled_status, + "outcome_resolved": True, + "reconciliation_status": "canonically_verified", + "reconciliation_basis": ( + "workflow_instance_terminal_read" + ), + "reconciliation_evidence_id": evidence_id, "canonical_readback": { "capability": capability_name, "instance_id": instance_id, @@ -5368,8 +6155,30 @@ def persist_effect_observation_phase( history_namespace=conversation_history_namespace, ) - def _effect_phase_acknowledged(outcome: Mapping[str, Any]) -> bool: - return bool(outcome.get("updated") or outcome.get("duplicate")) + def _effect_phase_acknowledged( + outcome: Mapping[str, Any], + *, + expected_identity: str | None = None, + ) -> bool: + if outcome.get("updated"): + if expected_identity is None: + return True + stored_phase = outcome.get("stored_phase") + return bool( + isinstance(stored_phase, Mapping) + and stored_phase.get("observation_identity_sha256") + == expected_identity + ) + if not outcome.get("duplicate"): + return False + if expected_identity is None: + return True + stored_phase = outcome.get("stored_phase") + return bool( + isinstance(stored_phase, Mapping) + and stored_phase.get("observation_identity_sha256") + == expected_identity + ) def late_effect_observer( *, @@ -5519,6 +6328,18 @@ def finish(text: str, *, status: str = "completed") -> AdaptiveTurnResult: and state.get("recovery_status") == "succeeded" and state.get("recovered_by_effect_id") ) + and not ( + state.get("effect_status") == "indeterminate" + and state.get("current_outcome_status") == "target_absent" + and state.get("outcome_resolved") is True + and state.get("recovery_status") == "succeeded" + and state.get("recovered_by_effect_id") + ) + ] + unresolved_incomplete_effects = [ + state + for state in incomplete_effects + if state.get("outcome_resolved") is not True ] succeeded_effects = [ state @@ -5527,57 +6348,65 @@ def finish(text: str, *, status: str = "completed") -> AdaptiveTurnResult: and state.get("turn_finality_required") is not False ] ( - material_succeeded_effect_ids, + _material_succeeded_effect_ids, canonically_verified_effect_ids, ) = _canonically_verified_material_effect_ids( tool_invocations, effect_snapshot, ) all_material_successes_verified = bool( - material_succeeded_effect_ids - ) and material_succeeded_effect_ids.issubset(canonically_verified_effect_ids) - mixed_no_change_failures = bool( - incomplete_effects and succeeded_effects - ) and all( - state.get("effect_status") in {"failed", "not_started"} - and state.get("changed") is False - and state.get("recovery_status") != "mismatched" - for state in incomplete_effects + _material_succeeded_effect_ids + ) and _material_succeeded_effect_ids.issubset( + canonically_verified_effect_ids ) - mixed_verified_workflow_fallback = bool( - incomplete_effects and succeeded_effects and all_material_successes_verified - ) and all( - state.get("effect_status") == "failed" - and state.get("capability_kind") == "represented_workflow" - and bool(state.get("instance_id")) - and isinstance(state.get("canonical_readback"), Mapping) - and str(state["canonical_readback"].get("status") or "").strip().lower() - in {"failed", "cancelled", "canceled"} - and state.get("recovery_status") != "mismatched" + bounded_incomplete_effects = all( + state.get("recovery_status") != "mismatched" + and ( + state.get("outcome_resolved") is True + or ( + state.get("effect_status") in {"failed", "not_started"} + and state.get("changed") is False + ) + ) for state in incomplete_effects ) - preservable_mixed_failures = ( - mixed_no_change_failures or mixed_verified_workflow_fallback + changed_non_success_present = any( + state.get("changed") is True for state in incomplete_effects + ) + current_state_observation_present = any( + state.get("reconciliation_status") == "current_state_observed" + for state in incomplete_effects ) if status == "completed" and incomplete_effects: + all_incomplete_statuses = { + str(state.get("effect_status") or "") + for state in incomplete_effects + } incomplete_statuses = { - str(state.get("effect_status") or "") for state in incomplete_effects + str(state.get("effect_status") or "") + for state in unresolved_incomplete_effects } if "indeterminate" in incomplete_statuses: status = "effect_outcome_indeterminate" elif "partial" in incomplete_statuses: status = "effect_partially_completed" - elif preservable_mixed_failures: - # A rejected or otherwise known-no-change attempt must not erase a - # useful answer about sibling effects that did complete. A failed - # durable workflow may also coexist with later material effects - # when its own terminal state and every later changed target were - # read back exactly. The turn remains honestly partial. + elif current_state_observation_present: + status = "effect_partially_completed" + elif ( + succeeded_effects + and bounded_incomplete_effects + and ( + not changed_non_success_present + or all_material_successes_verified + ) + ): status = "effect_partially_completed" - elif "failed" in incomplete_statuses: + elif "failed" in all_incomplete_statuses: status = "effect_failed" - else: + elif "not_started" in all_incomplete_statuses: status = "effect_not_started" + else: + status = "effect_partially_completed" reconciled_invocations: list[dict[str, Any]] = [] for raw_invocation in tool_invocations: invocation = dict(raw_invocation) @@ -5609,12 +6438,19 @@ def finish(text: str, *, status: str = "completed") -> AdaptiveTurnResult: invocation["reconciliation_status"] = state.get( "reconciliation_status" ) - invocation["status"] = "ok" - invocation["mutation_outcome"] = "succeeded" - invocation["outcome_finality"] = "terminal_for_turn" - if invocation.get("error_code"): - invocation["initial_error_code"] = invocation.get("error_code") - invocation.pop("error_code", None) + if ( + state.get("reconciliation_status") + == "canonically_verified" + and state.get("effect_status") == "succeeded" + ): + invocation["status"] = "ok" + invocation["mutation_outcome"] = "succeeded" + invocation["outcome_finality"] = "terminal_for_turn" + if invocation.get("error_code"): + invocation["initial_error_code"] = invocation.get( + "error_code" + ) + invocation.pop("error_code", None) if state.get("initial_effect_status"): invocation["initial_effect_status"] = state.get( "initial_effect_status" @@ -5624,6 +6460,17 @@ def finish(text: str, *, status: str = "completed") -> AdaptiveTurnResult: invocation["reconciliation_evidence_id"] = state.get( "reconciliation_evidence_id" ) + for outcome_field in ( + "current_outcome_status", + "outcome_resolved", + "reconciliation_basis", + "canonical_scope", + ): + if state.get(outcome_field) is not None: + value = state.get(outcome_field) + invocation[outcome_field] = ( + dict(value) if isinstance(value, Mapping) else value + ) if state.get("result_target_ids"): invocation["result_target_ids"] = list( state.get("result_target_ids") or () @@ -5662,92 +6509,8 @@ def finish(text: str, *, status: str = "completed") -> AdaptiveTurnResult: ) if cited_ontology_mutation_claim_conflicts and status == "completed": status = "effect_partially_completed" - preserve_pending_durable_response = ( - model_answer_completed - and status == "effect_partially_completed" - and _can_preserve_pending_durable_response(text, relevant_effects) - and not cited_ontology_mutation_claim_conflicts - ) - preserve_mixed_effect_response = ( - model_answer_completed - and status == "effect_partially_completed" - and preservable_mixed_failures - and not cited_ontology_mutation_claim_conflicts - ) - effect_finality_fallback = bool(cited_ontology_mutation_claim_conflicts) or ( - status != "completed" - and bool(relevant_effects) - and not preserve_pending_durable_response - and not preserve_mixed_effect_response - ) - if preserve_pending_durable_response: - aux_calls.append( - { - "type": "adaptive_turn_pending_effect_response_preserved", - "schema_version": ( - "adaptive_turn_pending_effect_response_preserved.v1" - ), - "terminal_status": status, - "effect_count": len(relevant_effects), - "instance_ids": [ - state.get("instance_id") for state in relevant_effects - ], - } - ) - if preserve_mixed_effect_response: - failed_count = len(incomplete_effects) - succeeded_count = len(succeeded_effects) - if mixed_verified_workflow_fallback: - workflow_attempt_label = "attempt" if failed_count == 1 else "attempts" - workflow_failure_label = ( - "workflow failure" if failed_count == 1 else "workflow failures" - ) - qualification = ( - f"Effect receipts also report {succeeded_count} succeeded and " - f"{failed_count} failed represented-workflow " - f"{workflow_attempt_label}. The {workflow_failure_label} and " - "the later changed targets were read back " - "exactly, so the verified successful result is preserved while " - "the turn remains partial." - ) - else: - qualification = ( - f"Effect receipts also report {succeeded_count} succeeded and " - f"{failed_count} failed or not started with no reported change. " - "The successful result is preserved; the unsuccessful attempts " - "can be inspected or retried independently." - ) - text = ( - f"{text.rstrip()}\n\n{qualification}" if text.strip() else qualification - ) - aux_calls.append( - { - "type": "adaptive_turn_mixed_effect_response_preserved", - "schema_version": ( - "adaptive_turn_mixed_effect_response_preserved.v1" - ), - "terminal_status": status, - "succeeded_count": succeeded_count, - "known_no_change_count": ( - failed_count if mixed_no_change_failures else 0 - ), - "failed_workflow_count": ( - failed_count if mixed_verified_workflow_fallback else 0 - ), - "canonically_verified_succeeded_count": len( - canonically_verified_effect_ids - ), - "preservation_basis": ( - "failed_workflow_and_material_successes_exactly_read_back" - if mixed_verified_workflow_fallback - else "known_no_change_failures" - ), - } - ) + response_authority = "model" if cited_ontology_mutation_claim_conflicts: - text = _cited_ontology_mutation_conflict_response( - cited_ontology_mutation_claim_conflicts - ) aux_calls.append( { "type": "adaptive_turn_cited_ontology_mutation_claim_rejected", @@ -5761,74 +6524,35 @@ def finish(text: str, *, status: str = "completed") -> AdaptiveTurnResult: ], } ) - elif effect_finality_fallback: - status_counts = { - effect_status: sum( - 1 - for state in relevant_effects - if state.get("effect_status") == effect_status - ) - for effect_status in ( - "succeeded", - "partial", - "failed", - "indeterminate", - "not_started", - ) - } - count_text = ", ".join( - f"{count} {effect_status}" - for effect_status, count in status_counts.items() - if count - ) - if not count_text: - count_text = ( - f"{len(relevant_effects)} receipt" - f"{'s' if len(relevant_effects) != 1 else ''} " - "with unresolved status" - ) - unknown_change_effects = [ - state - for state in relevant_effects - if state.get("effect_status") in {"partial", "indeterminate"} - or state.get("changed") is True - or ( - state.get("effect_status") == "succeeded" - and state.get("changed") is None - ) - ] - if unknown_change_effects: - text = ( - "That turn did not finish cleanly. Its effect receipts currently " - f"report {count_text}. A handler receipt is not canonical " - "read-back, so I will not claim that nothing changed. Inspect " - "the represented state before retrying any effect whose outcome " - "is unknown." - ) - elif status_counts["not_started"]: - text = ( - "That turn did not finish cleanly. Its effect receipts currently " - f"report {count_text}. The not-started effect" - f"{'s were' if status_counts['not_started'] != 1 else ' was'} " - "not dispatched and " - f"{'report' if status_counts['not_started'] != 1 else 'reports'} " - "no change; " - f"{'they can' if status_counts['not_started'] != 1 else 'it can'} " - "be retried in a new turn." + if bool(relevant_effects) and ( + status != "completed" or cited_ontology_mutation_claim_conflicts + ): + model_draft, _ = _extract_conversation_situation_sidecar( + text, + current_situation=conversation_situation, + ) + text, outcome_report = _build_effect_outcome_report( + terminal_status=status, + effect_snapshot=effect_snapshot, + tool_invocations=reconciled_invocations, + trusted_scope=scope, + ) + if model_answer_completed and model_draft.strip(): + quoted_draft = "\n".join( + f"> {line}" if line else ">" + for line in model_draft.strip().splitlines() ) - else: text = ( - "That turn did not finish cleanly. Its effect receipts currently " - f"report {count_text} and report no change. Inspect the failure " - "receipt before retrying." + f"{text}\n\n### Model draft (non-authoritative)\n\n" + "This draft is retained for useful context, but its effect, " + "scope, and success claims do not override the report above." + f"\n\n{quoted_draft}" ) + response_authority = "canonical_outcome" aux_calls.append( { - "type": "adaptive_turn_effect_finality_fallback", - "schema_version": ("adaptive_turn_effect_finality_fallback.v1"), - "terminal_status": status, - "effect_count": len(relevant_effects), - "status_counts": status_counts, + "type": "adaptive_turn_effect_outcome_report", + **outcome_report, } ) evidence_index = _compact_evidence_index(evidence_store.index()) @@ -5868,7 +6592,7 @@ def finish(text: str, *, status: str = "completed") -> AdaptiveTurnResult: duration_ms=max(0.0, (clock() - started) * 1000.0), terminal_status=status, evidence_index=tuple(evidence_index), - effect_finality_fallback=effect_finality_fallback, + response_authority=response_authority, conversation_situation=updated_conversation_situation, ) @@ -7018,6 +7742,49 @@ def contained( if effect_request_signature is not None else None ) + prior_indeterminate_request = ( + indeterminate_effect_requests.get(effect_request_signature) + if effect_request_signature is not None + and item.capability_kind != "represented_workflow" + else None + ) + if prior_indeterminate_request is not None: + prior_effect_id, prior_error_code = prior_indeterminate_request + with effect_state_lock: + prior_effect_state = effect_states.get(prior_effect_id) or {} + current_state_observed = ( + prior_effect_state.get("outcome_resolved") is True + ) + return index, contained( + { + **_error_payload( + ( + "effect_request_current_state_already_observed" + if current_state_observed + else "effect_request_reconciliation_required" + ), + ( + "This exact effect request was not repeated because " + "its earlier outcome requires canonical inspection." + if not current_state_observed + else ( + "This exact effect request was not repeated " + "because its target state was already read back." + ) + ), + ), + "status": "not_started", + "mutation_outcome": "not_started", + "outcome_finality": "terminal_for_turn", + "prior_effect_id": prior_effect_id, + "prior_error_code": prior_error_code, + "changed": False, + "recovery_affordances": [ + {"action_type": "inspect_canonical_state_before_retry"}, + {"action_type": "use_typed_recovery"}, + ], + } + ) if ( prior_terminal_failure is not None and prior_terminal_failure[0] == successful_effect_mutation_generation @@ -7333,6 +8100,27 @@ def contained( else None ), ) + elif ( + is_effect + and effect_request_signature is not None + and terminal_effect_status == "indeterminate" + and item.capability_kind != "represented_workflow" + and effect_identifier is not None + and execution_method_name == "create_concepts" + and ( + not isinstance(raw_payload, Mapping) + or raw_payload.get("retryable") is not True + ) + ): + indeterminate_effect_requests[effect_request_signature] = ( + effect_identifier, + ( + str(raw_payload.get("error_code")).strip() + if isinstance(raw_payload, Mapping) + and raw_payload.get("error_code") + else None + ), + ) elif ( is_effect and terminal_effect_status in {"succeeded", "partial"} @@ -7653,6 +8441,14 @@ def contained( canonical_effect_readback ) if is_effect: + failure_fact = _typed_effect_failure_fact( + raw_payload, + workflow_event=selected_workflow_execution_event, + ) + current_effect_request_signature = _effect_request_signature( + canonical_name, + arguments, + ) remember_effect_state( effect_identifier, phase=0, @@ -7687,7 +8483,10 @@ def contained( original_result=( raw_payload if isinstance(raw_payload, Mapping) else None ), + failure_fact=failure_fact, + effect_request_signature=current_effect_request_signature, result_target_ids=result_target_ids, + observation_order=len(tool_invocations), ) remember_recovery_affordance( effect_id=effect_identifier, @@ -7699,6 +8498,11 @@ def contained( arguments=arguments, effect_status=effect_status, ) + reconcile_successful_exact_absence_retry( + recovery_effect_id=effect_identifier, + request_signature=current_effect_request_signature, + effect_status=effect_status, + ) envelope_payload.update( { "effect_id": effect_identifier, @@ -7723,10 +8527,67 @@ def contained( receipt_value = raw_payload.get(receipt_key) if isinstance(receipt_value, (str, int, float, bool)): envelope_payload[receipt_key] = receipt_value + if failure_fact is not None: + envelope_payload["failure_fact"] = dict(failure_fact) else: + requested_target_ids = { + str(arguments.get(field)).strip() + for field in _EXACT_READBACK_ARGUMENT_FIELDS + if isinstance(arguments.get(field), str) + and str(arguments.get(field)).strip() + } + exact_target_ids = requested_target_ids.intersection( + result_target_ids + ) + negative_exact_target = None + if ( + canonical_name == "fetch_concept" + and isinstance(raw_payload, Mapping) + and raw_payload.get("success") is False + and raw_payload.get("error_code") == "concept_not_found" + and len(requested_target_ids) == 1 + ): + error_details = raw_payload.get("error_details") + missing_concept_id = ( + str(error_details.get("concept_id") or "").strip() + if isinstance(error_details, Mapping) + else "" + ) + requested_concept_id = next(iter(requested_target_ids)) + if missing_concept_id == requested_concept_id: + negative_exact_target = requested_concept_id + if (status == "ok" and exact_target_ids) or negative_exact_target: + canonical_scope = ( + _canonical_scope_from_exact_concept_read( + raw_payload, + trusted_scope=scope, + ) + if canonical_name == "fetch_concept" + else None + ) + with effect_state_lock: + exact_read_observations.append( + { + "observation_order": len(tool_invocations), + "capability_name": canonical_name, + "call_id": call.call_id, + "evidence_id": envelope.evidence_id, + "requested_target_ids": sorted( + requested_target_ids + ), + "result_target_ids": list(result_target_ids), + "canonical_scope": canonical_scope, + "target_state": ( + "absent" + if negative_exact_target + else "present" + ), + } + ) reconcile_workflow_instance_readback( capability_name=canonical_name, payload=raw_payload, + call_id=call.call_id, evidence_id=envelope.evidence_id, ) batch_results[index] = ToolResult( @@ -7828,6 +8689,12 @@ def contained( receipt_value = raw_payload.get(receipt_key) if isinstance(receipt_value, (str, int, float, bool)): invocation[receipt_key] = receipt_value + failure_fact = _typed_effect_failure_fact( + raw_payload, + workflow_event=selected_workflow_execution_event, + ) + if failure_fact is not None: + invocation["failure_fact"] = dict(failure_fact) if canonical_effect_readback is not None: invocation["canonical_readback"] = dict( canonical_effect_readback @@ -7978,7 +8845,7 @@ def contained( reconciled_states = { effect_id: dict(state) for effect_id, state in effect_states.items() - if state.get("reconciliation_status") == "canonically_verified" + if state.get("reconciliation_status") } reconciled_results: list[ToolResult] = [] for result in correlated_results: @@ -7993,16 +8860,27 @@ def contained( reconciled_results.append(result) continue output = dict(result.output) - if output.get("error_code"): - output["initial_error_code"] = output.get("error_code") - output.pop("error_code", None) + reconciled_status = str(state.get("reconciliation_status") or "") + if reconciled_status == "canonically_verified": + if output.get("error_code"): + output["initial_error_code"] = output.get("error_code") + output.pop("error_code", None) + output.update( + { + "effect_status": "succeeded", + "changed": True, + "mutation_outcome": "succeeded", + "outcome_finality": "terminal_for_turn", + } + ) output.update( { - "effect_status": "succeeded", - "changed": True, - "mutation_outcome": "succeeded", - "outcome_finality": "terminal_for_turn", - "reconciliation_status": "canonically_verified", + "reconciliation_status": reconciled_status, + "reconciliation_basis": state.get("reconciliation_basis"), + "current_outcome_status": state.get( + "current_outcome_status" + ), + "outcome_resolved": state.get("outcome_resolved") is True, "reconciliation_evidence_id": state.get( "reconciliation_evidence_id" ), @@ -8019,7 +8897,11 @@ def contained( call_id=result.call_id, tool_name=result.tool_name, output=output, - status="ok", + status=( + "ok" + if reconciled_status == "canonically_verified" + else result.status + ), ) ) correlated_results = reconciled_results diff --git a/src/backend/services/turn_execution_record_service.py b/src/backend/services/turn_execution_record_service.py index 8ff9fe4c..9c3aa0d8 100644 --- a/src/backend/services/turn_execution_record_service.py +++ b/src/backend/services/turn_execution_record_service.py @@ -129,6 +129,8 @@ "dispatch_intent", "turn_terminal", "late_terminal", + "current_state_observation", + "canonical_reconciliation", "conversation_projection", } ) @@ -12397,6 +12399,7 @@ def record_effect_observation_phase( if isinstance(effect_entry, Mapping) and isinstance( effect_entry.get(clean_phase), Mapping ): + stored_phase = dict(effect_entry[clean_phase]) return { "updated": False, "duplicate": True, @@ -12405,6 +12408,7 @@ def record_effect_observation_phase( "request_id": clean_request_id, "effect_id": clean_effect_id, "phase": clean_phase, + "stored_phase": stored_phase, } if existing is None: request_id_collision = _turn_execution_find_one( @@ -12468,6 +12472,7 @@ def record_effect_observation_phase( "request_id": clean_request_id, "effect_id": clean_effect_id, "phase": clean_phase, + "stored_phase": dict(phase_entry), } except DuplicateKeyError: return { diff --git a/src/backend/workflows/durable/instance_manager.py b/src/backend/workflows/durable/instance_manager.py index 0c29e0d4..c4ab2b48 100644 --- a/src/backend/workflows/durable/instance_manager.py +++ b/src/backend/workflows/durable/instance_manager.py @@ -31,6 +31,7 @@ hydrate_workflow_payload_blob_refs, is_workflow_payload_blob_ref, ) +from ..execution_contracts import normalise_workflow_termination_code from .authority_snapshot_attestation import ( DURABLE_AUTHORITY_CHECKPOINT_ATTESTATION_FIELD, DURABLE_WORKER_CLAIM_PROVENANCE_SCHEMA_VERSION, @@ -2313,16 +2314,15 @@ def mark_failed( logger.warning( "[durable_workflow] Instance %s failed: %s", instance_id, error ) - reason_code = ( - error.split(":", 1)[0].strip().lower() - if isinstance(error, str) and ":" in error - else "failed" + reason_code = normalise_workflow_termination_code( + error, + default="failed", ) self._record_durable_episode_final( instance=instance_before, completed=False, terminal_stage=error_step or instance_before.current_state or "failed", - termination_code=reason_code or "failed", + termination_code=reason_code, termination_detail=error, final_state=instance_before.current_state, ) diff --git a/src/backend/workflows/execution_contracts.py b/src/backend/workflows/execution_contracts.py index bf8d9307..016e747d 100644 --- a/src/backend/workflows/execution_contracts.py +++ b/src/backend/workflows/execution_contracts.py @@ -7,6 +7,7 @@ from __future__ import annotations +import re from copy import deepcopy from typing import Any, Dict, Mapping, MutableMapping @@ -116,11 +117,38 @@ LAST_CONTROL_SIGNAL_ERROR_KEY = "last_control_signal_error" WORKFLOW_RETURN_PAYLOAD_KEY = "workflow_return_payload" +_WORKFLOW_TERMINATION_CODE_MAX_CHARS = 128 +_WORKFLOW_TERMINATION_CODE_PATTERN = re.compile(r"^[a-z][a-z0-9]*(?:_[a-z0-9]+)*$") + def _normalise_text(value: Any) -> str: return str(value or "").strip() +def normalise_workflow_termination_code( + value: Any, + *, + default: str, +) -> str: + """Return a bounded machine error code without promoting prose to a code. + + Workflow failures may carry either a bare snake-case code or + ``code: detail``. Only the machine-code portion belongs in episode + telemetry; arbitrary error prose remains the termination detail. + """ + + if not isinstance(value, str): + return default + candidate = value.strip().split(":", 1)[0].strip().lower() + if ( + not candidate + or len(candidate) > _WORKFLOW_TERMINATION_CODE_MAX_CHARS + or _WORKFLOW_TERMINATION_CODE_PATTERN.fullmatch(candidate) is None + ): + return default + return candidate + + def normalise_control_signal( value: Any, *, diff --git a/tests/backend/test_adaptive_turn_service.py b/tests/backend/test_adaptive_turn_service.py index f64d403f..55f7836c 100644 --- a/tests/backend/test_adaptive_turn_service.py +++ b/tests/backend/test_adaptive_turn_service.py @@ -77,6 +77,7 @@ def _acknowledge_effect_observation_journal( "updated": True, "duplicate": False, "phase": kwargs.get("phase"), + "stored_phase": dict(kwargs.get("observation") or {}), }, ) @@ -2452,7 +2453,7 @@ def handler(name: str, _arguments: dict[str, Any]) -> dict[str, Any]: assert result.tool_invocations[0]["changed"] is True -def test_finality_fallback_rejects_pre_reconciliation_situation( +def test_canonical_outcome_rejects_pre_reconciliation_situation( monkeypatch: pytest.MonkeyPatch, ) -> None: _stub_same_turn_ontology_delegation(monkeypatch) @@ -2504,8 +2505,9 @@ def test_finality_fallback_rejects_pre_reconciliation_situation( ) assert result.terminal_status == "model_error" - assert result.effect_finality_fallback is True - assert "will not claim that nothing changed" in result.response_text + assert result.response_authority == "canonical_outcome" + assert "## Effect outcome report" in result.response_text + assert "`create_concepts`" in result.response_text assert "I have not changed anything" not in result.response_text assert "von_conversation_situation" not in result.response_text assert result.conversation_situation == current_situation @@ -2612,8 +2614,9 @@ def handler(_name: str, _arguments: dict[str, Any]) -> dict[str, Any]: assert committed == ["#V#committed_before_invalid_receipt"] assert result.terminal_status == "model_error" - assert result.effect_finality_fallback is True - assert "will not claim that nothing changed" in result.response_text + assert result.response_authority == "canonical_outcome" + assert "## Effect outcome report" in result.response_text + assert "`create_concepts`" in result.response_text assert result.tool_invocations[0]["effect_status"] == "indeterminate" assert result.tool_invocations[0]["changed"] is None @@ -2760,9 +2763,9 @@ def persist_phase(**kwargs: Any) -> dict[str, Any]: final_synthesis_reserve_seconds=2, ) - assert result.effect_finality_fallback is True + assert result.response_authority == "canonical_outcome" assert result.tool_invocations[0]["effect_status"] == "indeterminate" - assert "1 indeterminate" in result.response_text + assert "status `indeterminate`" in result.response_text release_handler.set() assert observation_persisted.wait(timeout=1.0) @@ -3030,13 +3033,11 @@ def invoke(method_name: str, *_args: Any, **_kwargs: Any) -> SimpleNamespace: ) assert result.terminal_status == "effect_partially_completed" - assert result.effect_finality_fallback is False + assert result.response_authority == "canonical_outcome" assert "The valid effect completed." in result.response_text - assert "1 succeeded and 1 failed or not started" in result.response_text - assert any( - call.get("type") == "adaptive_turn_mixed_effect_response_preserved" - for call in result.aux_llm_calls - ) + assert "### Model draft (non-authoritative)" in result.response_text + assert "visibility_effect_not_delegated" in result.response_text + assert "`create_concepts`" in result.response_text assert invoked == ["create_concepts"] assert len(result.tool_invocations) == 2 assert len({item["effect_id"] for item in result.tool_invocations}) == 2 @@ -3154,10 +3155,21 @@ def handler(_name: str, arguments: dict[str, Any]) -> dict[str, Any]: ) assert result.terminal_status == "effect_partially_completed" - assert result.effect_finality_fallback is True - assert drafted_claim not in result.response_text - assert "#V#gael_gendron --is_an_instance_of--> #V#student" in result.response_text - assert "effect status `not_started` and changed `false`" in result.response_text + assert result.response_authority == "canonical_outcome" + assert good_effect_id in result.response_text + assert failed_effect_id in result.response_text + assert "Added; forward and inverse verified" in result.response_text + assert result.response_text.index("### Unsuccessful or unresolved") < ( + result.response_text.index("### Model draft (non-authoritative)") + ) + assert "source_id `#V#gael_gendron`" in result.response_text + assert "predicate `is_an_instance_of`" in result.response_text + assert "target `#V#student`" in result.response_text + assert "status `not_started`" in result.response_text + assert "reported no change" in result.response_text + assert "error code `ontology_mutation_target_not_accessible`" in ( + result.response_text + ) rejection = next( item for item in result.aux_llm_calls @@ -3252,10 +3264,11 @@ def test_cited_ontology_success_without_relation_readback_is_rejected( ) assert result.terminal_status == "effect_partially_completed" - assert result.effect_finality_fallback is True - assert "Added and canonically verified" not in result.response_text - assert "handler status `succeeded`" in result.response_text - assert "canonical relation read-back is absent or negative" in ( + assert result.response_authority == "canonical_outcome" + assert "Added and canonically verified" in result.response_text + assert "### Model draft (non-authoritative)" in result.response_text + assert "handler reported `succeeded`" in result.response_text + assert "canonical read-back did not verify the outcome" in ( result.response_text ) rejection = next( @@ -3363,9 +3376,10 @@ def invoke(method_name: str, *_args: Any, **_kwargs: Any) -> SimpleNamespace: assert invoked == ["create_concepts", "general_read"] assert result.terminal_status == "effect_outcome_indeterminate" - assert result.effect_finality_fallback is True + assert result.response_authority == "canonical_outcome" assert "indeterminate" in result.response_text - assert "The first effect needs canonical inspection." not in result.response_text + assert "The first effect needs canonical inspection." in result.response_text + assert "### Model draft (non-authoritative)" in result.response_text assert result.tool_invocations[0]["effect_status"] == "indeterminate" assert result.tool_invocations[1]["result_target_ids"] == ["#V#created_if_present"] blocked = result.tool_invocations[2] @@ -3459,25 +3473,535 @@ def test_exact_terminal_reconciliation_preserves_recovered_ontology_answer( turn_budget_seconds=10, final_synthesis_reserve_seconds=2, ) - - assert result.terminal_status == "completed" - assert result.effect_finality_fallback is False - assert result.response_text == answer - tool_messages = [ + + assert result.terminal_status == "completed" + assert result.response_authority == "model" + assert result.response_text == answer + tool_messages = [ + item + for item in client.calls[1]["context"] + if item.get("role") == "tool" + ] + model_receipt = json.loads(tool_messages[-1]["content"]) + assert model_receipt["effect_status"] == "succeeded" + assert model_receipt["reconciliation_status"] == "canonically_verified" + assert model_receipt["result_target_ids"] == [concept_id] + invocation = result.tool_invocations[0] + assert invocation["effect_status"] == "succeeded" + assert invocation["initial_effect_status"] == "indeterminate" + assert invocation["reconciliation_status"] == "canonically_verified" + assert invocation["canonical_readback"]["verified"] is True + assert invocation["result_target_ids"] == [concept_id] + + +@pytest.mark.parametrize("marker_succeeds", [True, False]) +def test_exact_current_readback_renders_truthful_paper_partial_outcome( + monkeypatch: pytest.MonkeyPatch, + marker_succeeds: bool, +) -> None: + _stub_same_turn_ontology_delegation(monkeypatch) + concept_id = "#V#paper_2512_23333" + marker_id = "#V#paper_2512_23333_processing_marker" + invoked: list[str] = [] + persisted: list[dict[str, Any]] = [] + + monkeypatch.setattr( + "src.backend.services.ontology_mutation_command_service." + "reconcile_governed_ontology_postcondition", + lambda **_kwargs: { + "success": False, + "verified": False, + "method_name": "create_concepts", + }, + ) + + from src.backend.services import turn_execution_record_service + + def persist_phase(**kwargs: Any) -> dict[str, Any]: + persisted.append(dict(kwargs)) + return { + "updated": True, + "duplicate": False, + "phase": kwargs.get("phase"), + "stored_phase": dict(kwargs.get("observation") or {}), + } + + monkeypatch.setattr( + turn_execution_record_service, + "record_effect_observation_phase", + persist_phase, + ) + + def handler(name: str, _arguments: dict[str, Any]) -> dict[str, Any]: + invoked.append(name) + if name == "download_paper": + return { + "success": False, + "effect_status": "failed", + "changed": False, + "error_code": "arxiv_acquisition_unavailable", + "error": ( + "RuntimeError: asyncio lock is bound to a different event loop" + ), + } + if name == "create_concepts": + return { + "success": False, + "effect_status": "indeterminate", + "changed": None, + "mutation_outcome": "unknown", + "outcome_finality": "requires_canonical_reconciliation", + "error_code": "ontology_mutation_postcondition_failed", + "created_concept_ids": [concept_id], + "postcondition_reconciliation": { + "schema_version": ( + "ontology_mutation_postcondition_reconciliation.v1" + ) + }, + } + if name == "record_source_processing_marker": + if not marker_succeeds: + return { + "success": False, + "effect_status": "failed", + "changed": False, + "error_code": "processing_marker_write_failed", + } + return { + "success": True, + "effect_status": "succeeded", + "changed": True, + "marker_concept_id": marker_id, + } + raise AssertionError(name) + + gateway = _effect_gateway(handler) + gateway._catalogue.register( + MethodDefinition( + name="download_paper", + handler=lambda **kwargs: handler("download_paper", kwargs), + input_schema=Schema(allow_unknown=True), + output_schema=Schema(required={"success": bool}, allow_unknown=True), + category="write", + ordinary_turn_effect=True, + ) + ) + gateway._catalogue.register( + MethodDefinition( + name="fetch_concept", + handler=lambda concept_id: { + "success": True, + "concept_id": concept_id, + "publication_context": { + "kind": "user", + "concept_id": "#V#michael_witbrock", + }, + "relationships": { + "#V#specific_to_user": ["#V#michael_witbrock"] + }, + }, + input_schema=Schema(required={"concept_id": str}), + output_schema=Schema(required={"success": bool}, allow_unknown=True), + category="read", + ) + ) + gateway._catalogue.register( + MethodDefinition( + name="record_source_processing_marker", + handler=lambda **kwargs: handler( + "record_source_processing_marker", + kwargs, + ), + input_schema=Schema(allow_unknown=True), + output_schema=Schema(required={"success": bool}, allow_unknown=True), + category="write", + ordinary_turn_effect=True, + ) + ) + for method_name in ( + "download_paper", + "fetch_concept", + "record_source_processing_marker", + ): + gateway.register_metrics_if_missing(method_name) + + client = _SequenceClient( + LLMResponse( + text_response="", + tool_calls=[ + ToolCall( + tool_name="turn_invoke_capability", + call_id="download-paper", + payload={ + "name": "download_paper", + "arguments": {"arxiv_id": "2512.23333"}, + }, + ) + ], + ), + LLMResponse( + text_response="", + tool_calls=[ + ToolCall( + tool_name="turn_invoke_capability", + call_id="create-paper-concept", + payload={ + "name": "create_concepts", + "arguments": { + "concepts": [ + { + "concept_id": concept_id, + "name": "A represented arXiv paper", + } + ] + }, + }, + ) + ], + ), + LLMResponse( + text_response="", + tool_calls=[ + ToolCall( + tool_name="turn_invoke_capability", + call_id="fetch-paper-concept", + payload={ + "name": "fetch_concept", + "arguments": {"concept_id": concept_id}, + }, + ) + ], + ), + LLMResponse( + text_response="", + tool_calls=[ + ToolCall( + tool_name="turn_invoke_capability", + call_id="mark-paper-processed", + payload={ + "name": "record_source_processing_marker", + "arguments": { + "source_system": "arxiv", + "source_item_id": "2512.23333", + }, + }, + ) + ], + ), + LLMResponse( + text_response=( + "The paper and PDF were represented in the University of " + "Auckland Strong AI Lab organisation namespace." + ) + ), + ) + + result = execute_adaptive_turn( + gateway=gateway, + prompt="Represent this arXiv paper and report what actually persisted.", + context=[], + llm_client=client, + model="test-model", + user_namespace="#V#michael_witbrock@uoasail", + user_concept_id="#V#michael_witbrock", + org_concept_id="#V#uoasail", + turn_id="paper-partial-current-state-readback", + turn_budget_seconds=20, + final_synthesis_reserve_seconds=2, + ) + + assert invoked == [ + "download_paper", + "create_concepts", + "record_source_processing_marker", + ] + assert len(client.calls) == 5 + assert result.terminal_status == "effect_partially_completed" + assert result.response_authority == "canonical_outcome" + assert concept_id in result.response_text + if marker_succeeds: + assert marker_id in result.response_text + else: + assert marker_id not in result.response_text + assert "processing_marker_write_failed" in result.response_text + assert "arxiv_acquisition_unavailable" in result.response_text + assert "asyncio lock is bound to a different event loop" in result.response_text + assert "User scope `#V#michael_witbrock`" in result.response_text + assert "organisation namespace" in result.response_text + assert result.response_text.index("User scope `#V#michael_witbrock`") < ( + result.response_text.index("organisation namespace") + ) + assert "### Model draft (non-authoritative)" in result.response_text + create_invocation = next( + item + for item in result.tool_invocations + if item.get("tool") == "create_concepts" + ) + assert create_invocation["effect_status"] == "indeterminate" + assert create_invocation["initial_effect_status"] == "indeterminate" + assert create_invocation["current_outcome_status"] == "target_observed" + assert create_invocation["outcome_resolved"] is True + current_state_phases = [ + item + for item in persisted + if item.get("phase") == "current_state_observation" + ] + assert len(current_state_phases) == 1 + assert current_state_phases[0]["observation"]["target_concept_ids"] == [ + concept_id + ] + + +def test_one_exact_read_does_not_resolve_multi_target_create( + monkeypatch: pytest.MonkeyPatch, +) -> None: + _stub_same_turn_ontology_delegation(monkeypatch) + target_ids = ["#V#multi_target_a", "#V#multi_target_b"] + persisted: list[dict[str, Any]] = [] + monkeypatch.setattr( + "src.backend.services.ontology_mutation_command_service." + "reconcile_governed_ontology_postcondition", + lambda **_kwargs: {"success": False, "verified": False}, + ) + + from src.backend.services import turn_execution_record_service + + def persist_phase(**kwargs: Any) -> dict[str, Any]: + persisted.append(dict(kwargs)) + return {"updated": True, "duplicate": False} + + monkeypatch.setattr( + turn_execution_record_service, + "record_effect_observation_phase", + persist_phase, + ) + + gateway = _effect_gateway( + lambda name, _arguments: ( + { + "success": False, + "effect_status": "indeterminate", + "changed": None, + "mutation_outcome": "unknown", + "outcome_finality": "requires_canonical_reconciliation", + "error_code": "ontology_mutation_postcondition_failed", + "created_concept_ids": target_ids, + "postcondition_reconciliation": { + "schema_version": ( + "ontology_mutation_postcondition_reconciliation.v1" + ) + }, + } + if name == "create_concepts" + else {"success": True} + ) + ) + gateway._catalogue.register( + MethodDefinition( + name="fetch_concept", + handler=lambda concept_id: { + "success": True, + "concept_id": concept_id, + "publication_context": { + "kind": "user", + "concept_id": "#V#user", + }, + }, + input_schema=Schema(required={"concept_id": str}), + output_schema=Schema(required={"success": bool}, allow_unknown=True), + category="read", + ) + ) + gateway.register_metrics_if_missing("fetch_concept") + client = _SequenceClient( + LLMResponse( + text_response="", + tool_calls=[ + ToolCall( + tool_name="turn_invoke_capability", + call_id="create-multiple-targets", + payload={ + "name": "create_concepts", + "arguments": { + "concepts": [ + {"concept_id": target_id, "name": target_id} + for target_id in target_ids + ] + }, + }, + ) + ], + ), + LLMResponse( + text_response="", + tool_calls=[ + ToolCall( + tool_name="turn_invoke_capability", + call_id="read-only-one-target", + payload={ + "name": "fetch_concept", + "arguments": {"concept_id": target_ids[0]}, + }, + ) + ], + ), + LLMResponse(text_response="Both targets were created."), + ) + + result = execute_adaptive_turn( + gateway=gateway, + prompt="Create both targets and verify them.", + context=[], + llm_client=client, + model="test-model", + user_namespace="#V#user@org", + user_concept_id="#V#user", + org_concept_id="#V#org", + turn_id="turn-partial-multi-target-readback", + turn_budget_seconds=20, + final_synthesis_reserve_seconds=2, + ) + + invocation = next( + item + for item in result.tool_invocations + if item.get("tool") == "create_concepts" + ) + assert result.terminal_status == "effect_outcome_indeterminate" + assert invocation["effect_status"] == "indeterminate" + assert "current_outcome_status" not in invocation + assert not any( + item.get("phase") == "current_state_observation" for item in persisted + ) + + +def test_unacknowledged_current_state_observation_does_not_resolve_effect( + monkeypatch: pytest.MonkeyPatch, +) -> None: + _stub_same_turn_ontology_delegation(monkeypatch) + concept_id = "#V#unacknowledged_current_state" + monkeypatch.setattr( + "src.backend.services.ontology_mutation_command_service." + "reconcile_governed_ontology_postcondition", + lambda **_kwargs: {"success": False, "verified": False}, + ) + + from src.backend.services import turn_execution_record_service + + def persist_phase(**kwargs: Any) -> dict[str, Any]: + if kwargs.get("phase") == "current_state_observation": + return { + "updated": False, + "duplicate": False, + "reason": "collection_unavailable", + } + return {"updated": True, "duplicate": False} + + monkeypatch.setattr( + turn_execution_record_service, + "record_effect_observation_phase", + persist_phase, + ) + gateway = _effect_gateway( + lambda name, _arguments: ( + { + "success": False, + "effect_status": "indeterminate", + "changed": None, + "mutation_outcome": "unknown", + "outcome_finality": "requires_canonical_reconciliation", + "error_code": "ontology_mutation_postcondition_failed", + "created_concept_ids": [concept_id], + "postcondition_reconciliation": { + "schema_version": ( + "ontology_mutation_postcondition_reconciliation.v1" + ) + }, + } + if name == "create_concepts" + else {"success": True} + ) + ) + gateway._catalogue.register( + MethodDefinition( + name="fetch_concept", + handler=lambda concept_id: { + "success": True, + "concept_id": concept_id, + "publication_context": { + "kind": "user", + "concept_id": "#V#user", + }, + }, + input_schema=Schema(required={"concept_id": str}), + output_schema=Schema(required={"success": bool}, allow_unknown=True), + category="read", + ) + ) + gateway.register_metrics_if_missing("fetch_concept") + client = _SequenceClient( + LLMResponse( + text_response="", + tool_calls=[ + ToolCall( + tool_name="turn_invoke_capability", + call_id="create-before-unacknowledged-read", + payload={ + "name": "create_concepts", + "arguments": { + "concepts": [ + {"concept_id": concept_id, "name": "Unacknowledged"} + ] + }, + }, + ) + ], + ), + LLMResponse( + text_response="", + tool_calls=[ + ToolCall( + tool_name="turn_invoke_capability", + call_id="read-current-without-journal-ack", + payload={ + "name": "fetch_concept", + "arguments": {"concept_id": concept_id}, + }, + ) + ], + ), + LLMResponse(text_response="The concept exists."), + ) + + result = execute_adaptive_turn( + gateway=gateway, + prompt="Create and inspect this concept.", + context=[], + llm_client=client, + model="test-model", + user_namespace="#V#user@org", + user_concept_id="#V#user", + org_concept_id="#V#org", + turn_id="turn-unacknowledged-current-state", + turn_budget_seconds=20, + final_synthesis_reserve_seconds=2, + ) + + invocation = next( + item + for item in result.tool_invocations + if item.get("tool") == "create_concepts" + ) + assert result.terminal_status == "effect_outcome_indeterminate" + assert invocation["effect_status"] == "indeterminate" + assert "current_outcome_status" not in invocation + failure = next( item - for item in client.calls[1]["context"] - if item.get("role") == "tool" - ] - model_receipt = json.loads(tool_messages[-1]["content"]) - assert model_receipt["effect_status"] == "succeeded" - assert model_receipt["reconciliation_status"] == "canonically_verified" - assert model_receipt["result_target_ids"] == [concept_id] - invocation = result.tool_invocations[0] - assert invocation["effect_status"] == "succeeded" - assert invocation["initial_effect_status"] == "indeterminate" - assert invocation["reconciliation_status"] == "canonically_verified" - assert invocation["canonical_readback"]["verified"] is True - assert invocation["result_target_ids"] == [concept_id] + for item in result.aux_llm_calls + if item.get("type") == "effect_observation_persistence_failure" + and item.get("phase") == "current_state_observation" + ) + assert failure["reason"] == "collection_unavailable" def test_per_method_minimum_admits_sequential_effects_below_hard_cap( @@ -3605,8 +4129,8 @@ def handler(name: str, _arguments: dict[str, Any]) -> dict[str, Any]: assert invoked == [] assert result.terminal_status == "effect_not_started" - assert "1 not_started" in result.response_text - assert "was not dispatched and reports no change" in result.response_text + assert "status `not_started`" in result.response_text + assert "effect_observation_unavailable" in result.response_text assert result.tool_invocations[0]["effect_status"] == "not_started" assert result.tool_invocations[0]["changed"] is False preview = result.tool_invocations[0]["evidence"]["preview"] @@ -6873,8 +7397,9 @@ def handler(_name: str, arguments: dict[str, Any]) -> dict[str, Any]: assert seen_cases == ["succeeded", "partial", "failed", "timeout"] assert result.terminal_status == "effect_outcome_indeterminate" - assert result.effect_finality_fallback is True - assert "The bounded effects were reported truthfully." not in result.response_text + assert result.response_authority == "canonical_outcome" + assert "The bounded effects were reported truthfully." in result.response_text + assert "### Model draft (non-authoritative)" in result.response_text assert [item["effect_status"] for item in result.tool_invocations] == [ "succeeded", "partial", @@ -6949,9 +7474,11 @@ def test_partial_effect_downgrades_nominal_model_completion( ) assert result.terminal_status == "effect_partially_completed" - assert result.effect_finality_fallback is True - assert "1 partial" in result.response_text - assert "Everything was created." not in result.response_text + assert result.response_authority == "canonical_outcome" + assert "status `partial`" in result.response_text + assert "derived_relation_failed" in result.response_text + assert "Everything was created." in result.response_text + assert "### Model draft (non-authoritative)" in result.response_text def test_identity_shaped_targets_are_not_globally_rewritten() -> None: @@ -8743,6 +9270,41 @@ def test_represented_workflow_failure_progress_is_actionable() -> None: assert progress_evidence["facts"][0]["value"] == ("research-description.pdf") +def test_represented_workflow_failure_projects_typed_nested_cause() -> None: + from src.backend.services.adaptive_turn_service import ( + _build_represented_workflow_execution_event, + ) + + event, _ = _build_represented_workflow_execution_event( + payload={ + "final_status": "failed", + "effect_status": "failed", + "workflow_execution": { + "latest_step_result_envelope": { + "output_payload": { + "mcp_result": { + "error_code": "arxiv_acquisition_unavailable", + "error": ( + "RuntimeError: asyncio lock is bound to a " + "different event loop" + ), + "private_payload": "must not be projected", + } + } + } + }, + }, + workflow_id="#V#arxiv_paper_representation_workflow", + workflow_name="ArXiv paper representation workflow", + ) + + assert event["error_code"] == "arxiv_acquisition_unavailable" + assert event["error"] == ( + "RuntimeError: asyncio lock is bound to a different event loop" + ) + assert "private_payload" not in event + + def test_represented_workflow_nonfinite_wait_is_typed_not_started_feedback( monkeypatch, ) -> None: @@ -9052,9 +9614,11 @@ def _read_instance(**kwargs: Any) -> dict[str, Any]: else: assert "canonical_readback" not in workflow_invocation assert result.terminal_status == expected_terminal_status - assert result.effect_finality_fallback is expected_fallback - if expected_fallback: + expected_authority = "canonical_outcome" if expected_fallback else "model" + assert result.response_authority == expected_authority + if expected_authority == "canonical_outcome": assert result.response_text != "The durable work product was verified." + assert "workflow-instance-readback-1" in result.response_text else: assert result.response_text == "The durable work product was verified." @@ -9307,7 +9871,7 @@ def direct_effect(name: str, arguments: dict[str, Any]) -> dict[str, Any]: ) assert result.terminal_status == expected_status - assert result.effect_finality_fallback is expected_fallback + assert result.response_authority == "canonical_outcome" workflow_invocation = next( item for item in result.tool_invocations @@ -9316,21 +9880,11 @@ def direct_effect(name: str, arguments: dict[str, Any]) -> dict[str, Any]: assert workflow_invocation["effect_status"] == "failed" assert workflow_invocation["changed"] is True assert workflow_invocation["canonical_readback"]["status"] == "failed" - if read_back_trip: - assert useful_answer in result.response_text - preservation = next( - item - for item in result.aux_llm_calls - if item.get("type") == "adaptive_turn_mixed_effect_response_preserved" - ) - assert preservation["preservation_basis"] == ( - "failed_workflow_and_material_successes_exactly_read_back" - ) - assert preservation["canonically_verified_succeeded_count"] == 4 - assert preservation["failed_workflow_count"] == 1 - assert preservation["known_no_change_count"] == 0 - else: - assert useful_answer not in result.response_text + assert useful_answer in result.response_text + assert "### Model draft (non-authoritative)" in result.response_text + assert trip_id in result.response_text + assert instance_id in result.response_text + assert "metadata validation failed before domain mutation" in result.response_text def test_failed_workflows_and_recovered_denials_preserve_verified_scoped_results( @@ -9640,9 +10194,11 @@ def persist_scoped_assertion(**kwargs: Any) -> dict[str, Any]: for call in delegation_calls ) assert result.terminal_status == "effect_partially_completed" - assert result.effect_finality_fallback is False - assert "Both research descriptions were durably read back" in (result.response_text) - assert "2 failed represented-workflow attempts" in result.response_text + assert result.response_authority == "canonical_outcome" + assert "Both research descriptions were durably read back" in ( + result.response_text + ) + assert "### Canonical scope" in result.response_text effects = [ invocation for invocation in result.tool_invocations @@ -9683,19 +10239,17 @@ def persist_scoped_assertion(**kwargs: Any) -> dict[str, Any]: invocation["canonical_readback"]["scope_mode"] == "organisation" for invocation in recoveries ) - preservation = next( + reports = [ item for item in result.aux_llm_calls - if item.get("type") == "adaptive_turn_mixed_effect_response_preserved" - ) - assert preservation["preservation_basis"] == ( - "failed_workflow_and_material_successes_exactly_read_back" - ) - assert preservation["failed_workflow_count"] == 2 - assert preservation["canonically_verified_succeeded_count"] == 2 + if item.get("type") == "adaptive_turn_effect_outcome_report" + ] + assert len(reports) == 1 -def test_pending_durable_response_with_exact_handle_is_preserved(monkeypatch) -> None: +def test_pending_durable_response_uses_canonical_report_with_exact_handle( + monkeypatch, +) -> None: from src.backend.services.workflow_turn_capability_service import ( WorkflowTurnCapability, ) @@ -9821,13 +10375,12 @@ def _read_instance(**kwargs: Any) -> dict[str, Any]: assert instance_reads == [{"instance_id": instance_id, "await_terminal": False}] assert result.terminal_status == "effect_partially_completed" - assert result.effect_finality_fallback is False - assert result.response_text == final_text - assert any( - call.get("schema_version") - == "adaptive_turn_pending_effect_response_preserved.v1" - for call in result.aux_llm_calls - ) + assert result.response_authority == "canonical_outcome" + assert result.response_text != final_text + assert final_text in result.response_text + assert "### Model draft (non-authoritative)" in result.response_text + assert instance_id in result.response_text + assert "status `partial`" in result.response_text def test_workflow_instance_readback_does_not_reconcile_unrelated_effect( @@ -9915,7 +10468,7 @@ def _effect_handler(_name: str, _arguments: dict[str, Any]) -> dict[str, Any]: assert effect_invocation["effect_status"] == "partial" assert "canonical_readback" not in effect_invocation assert result.terminal_status == "effect_partially_completed" - assert result.effect_finality_fallback is True + assert result.response_authority == "canonical_outcome" assert result.response_text != "Everything completed." @@ -10020,7 +10573,7 @@ def test_read_only_workflow_not_started_preserves_successful_direct_recovery( assert result.response_text == ("The recent messages were summarised successfully.") assert result.terminal_status == "completed" - assert result.effect_finality_fallback is False + assert result.response_authority == "model" workflow_invocation = next( item for item in result.tool_invocations @@ -10246,6 +10799,295 @@ def handler(name: str, arguments: dict[str, Any]) -> dict[str, Any]: assert result.tool_invocations[1]["changed"] is False +@pytest.mark.parametrize( + ("inspection_target", "concepts"), + [ + ( + None, + [{"concept_id": "#V#unknown_create", "name": "Unknown"}], + ), + ( + "#V#wrong_target_create", + [{"concept_id": "#V#unknown_create", "name": "Unknown"}], + ), + ( + "#V#unknown_create", + [ + {"concept_id": "#V#unknown_create", "name": "Unknown"}, + {"name": "Possibly created without an explicit ID"}, + ], + ), + ], +) +def test_unchanged_indeterminate_create_requires_exact_inspection_before_retry( + monkeypatch: pytest.MonkeyPatch, + inspection_target: str | None, + concepts: list[dict[str, str]], +) -> None: + _stub_same_turn_ontology_delegation(monkeypatch) + handler_calls: list[dict[str, Any]] = [] + + def handler(name: str, arguments: dict[str, Any]) -> dict[str, Any]: + assert name == "create_concepts" + handler_calls.append(arguments) + return { + "success": False, + "effect_status": "indeterminate", + "changed": None, + "error_code": "tool_timeout_outcome_unknown", + "mutation_outcome": "unknown", + } + + effect_arguments = {"concepts": concepts} + responses = [ + LLMResponse( + text_response="", + tool_calls=[ + ToolCall( + tool_name="turn_invoke_capability", + call_id="unknown-create-1", + payload={ + "name": "create_concepts", + "arguments": effect_arguments, + }, + ) + ], + ), + ] + if inspection_target is not None: + responses.append( + LLMResponse( + text_response="", + tool_calls=[ + ToolCall( + tool_name="turn_invoke_capability", + call_id="inspect-wrong-create-target", + payload={ + "name": "fetch_concept", + "arguments": {"concept_id": inspection_target}, + }, + ) + ], + ) + ) + responses.extend( + [ + LLMResponse( + text_response="", + tool_calls=[ + ToolCall( + tool_name="turn_invoke_capability", + call_id="unknown-create-2", + payload={ + "name": "create_concepts", + "arguments": effect_arguments, + }, + ) + ], + ), + LLMResponse(text_response="The create outcome still needs inspection."), + ] + ) + + gateway = _effect_gateway(handler) + gateway._catalogue.register( + MethodDefinition( + name="fetch_concept", + handler=lambda concept_id: { + "success": False, + "error_code": "concept_not_found", + "error_details": { + "concept_id": concept_id, + "status": "not_found", + }, + }, + input_schema=Schema(required={"concept_id": str}), + output_schema=Schema(required={"success": bool}, allow_unknown=True), + category="read", + ) + ) + gateway.register_metrics_if_missing("fetch_concept") + client = _SequenceClient(*responses) + + result = execute_adaptive_turn( + gateway=gateway, + prompt="Create this concept once.", + context=[], + llm_client=client, + model="test-model", + user_namespace="#V#user@org", + user_concept_id="#V#user", + org_concept_id="#V#org", + turn_id="turn-repeat-indeterminate-create", + turn_budget_seconds=20, + final_synthesis_reserve_seconds=2, + ) + + assert len(handler_calls) == 1 + assert len(result.tool_invocations) == (3 if inspection_target else 2) + assert result.tool_invocations[0]["effect_status"] == "indeterminate" + assert "current_outcome_status" not in result.tool_invocations[0] + blocked_invocation = result.tool_invocations[-1] + assert blocked_invocation["error_code"] == ( + "effect_request_reconciliation_required" + ) + assert blocked_invocation["effect_status"] == "not_started" + + +@pytest.mark.parametrize("rich_receipt", [True, False]) +def test_exact_absence_allows_retry_of_indeterminate_create( + monkeypatch: pytest.MonkeyPatch, + rich_receipt: bool, +) -> None: + _stub_same_turn_ontology_delegation(monkeypatch) + concept_id = "#V#absent_then_created" + create_calls = 0 + + monkeypatch.setattr( + "src.backend.services.ontology_mutation_command_service." + "reconcile_governed_ontology_postcondition", + lambda **_kwargs: {"success": False, "verified": False}, + ) + + def handler(name: str, _arguments: dict[str, Any]) -> dict[str, Any]: + nonlocal create_calls + assert name == "create_concepts" + create_calls += 1 + if create_calls == 1: + failure = { + "success": False, + "effect_status": "indeterminate", + "changed": None, + "error_code": ( + "tool_timeout_outcome_unknown" + if rich_receipt + else "effect_outcome_unknown" + ), + "mutation_outcome": "unknown", + } + if rich_receipt: + failure.update( + { + "outcome_finality": "requires_canonical_reconciliation", + "created_concept_ids": [concept_id], + "postcondition_reconciliation": { + "schema_version": ( + "ontology_mutation_postcondition_reconciliation.v1" + ) + }, + } + ) + return failure + return { + "success": True, + "effect_status": "succeeded", + "changed": True, + "created_concept_ids": [concept_id], + } + + gateway = _effect_gateway(handler) + gateway._catalogue.register( + MethodDefinition( + name="fetch_concept", + handler=lambda concept_id: { + "success": False, + "error_code": "concept_not_found", + "error": f"Concept {concept_id} was not found.", + "error_details": { + "concept_id": concept_id, + "status": "not_found", + }, + }, + input_schema=Schema(required={"concept_id": str}), + output_schema=Schema(required={"success": bool}, allow_unknown=True), + category="read", + ) + ) + gateway.register_metrics_if_missing("fetch_concept") + effect_arguments = { + "concepts": [{"concept_id": concept_id, "name": "Absent then created"}] + } + client = _SequenceClient( + LLMResponse( + text_response="", + tool_calls=[ + ToolCall( + tool_name="turn_invoke_capability", + call_id="create-before-absence-read", + payload={ + "name": "create_concepts", + "arguments": effect_arguments, + }, + ) + ], + ), + LLMResponse( + text_response="", + tool_calls=[ + ToolCall( + tool_name="turn_invoke_capability", + call_id="read-exact-absence", + payload={ + "name": "fetch_concept", + "arguments": {"concept_id": concept_id}, + }, + ) + ], + ), + LLMResponse( + text_response="", + tool_calls=[ + ToolCall( + tool_name="turn_invoke_capability", + call_id="create-after-absence-read", + payload={ + "name": "create_concepts", + "arguments": effect_arguments, + }, + ) + ], + ), + LLMResponse(text_response=f"The concept {concept_id} now exists."), + ) + + result = execute_adaptive_turn( + gateway=gateway, + prompt="Create this concept, inspecting an unknown outcome before retry.", + context=[], + llm_client=client, + model="test-model", + user_namespace="#V#user@org", + user_concept_id="#V#user", + org_concept_id="#V#org", + turn_id="turn-retry-after-exact-absence", + turn_budget_seconds=20, + final_synthesis_reserve_seconds=2, + ) + + assert create_calls == 2 + create_invocations = [ + item + for item in result.tool_invocations + if item.get("tool") == "create_concepts" + ] + assert [item["effect_status"] for item in create_invocations] == [ + "indeterminate", + "succeeded", + ] + assert create_invocations[0]["current_outcome_status"] == "target_absent" + assert create_invocations[0]["outcome_resolved"] is True + assert create_invocations[0]["initial_effect_status"] == "indeterminate" + assert create_invocations[0]["recovery_status"] == "succeeded" + assert create_invocations[0]["recovered_by_effect_id"] == _effect_id( + turn_id="turn-retry-after-exact-absence", + call_id="create-after-absence-read", + capability_name="create_concepts", + ) + assert result.terminal_status == "completed" + assert result.response_authority == "model" + assert result.response_text == f"The concept {concept_id} now exists." + + def test_model_call_progress_reports_cumulative_usage_cost_and_exact_identity() -> None: progress_events: list[dict[str, Any]] = [] client = _SequenceClient( diff --git a/tests/backend/test_durable_workflow_system.py b/tests/backend/test_durable_workflow_system.py index b7d9931f..7ff202c8 100644 --- a/tests/backend/test_durable_workflow_system.py +++ b/tests/backend/test_durable_workflow_system.py @@ -1781,10 +1781,30 @@ def test_worker_heartbeat_registry_records_build_identity(self) -> None: assert workers[0]["active_instance_ids"] == ["instance-a"] assert workers[0]["build"]["git_short_commit"] == "abcdef123456" + @pytest.mark.parametrize( + ("error", "expected_code"), + [ + ( + "paper_reference_ingestion_no_items_succeeded", + "paper_reference_ingestion_no_items_succeeded", + ), + ("tool_timeout:Workflow step timed out", "tool_timeout"), + ("Unexpected workflow error while downloading", "failed"), + ("", "failed"), + ], + ) def test_claim_and_terminal_updates_record_workflow_episodes( - self, monkeypatch + self, + monkeypatch, + error: str, + expected_code: str, ) -> None: """Durable lifecycle should emit one start + one final episode record.""" + monkeypatch.delenv("VON_DURABLE_MIN_WORKER_BUILD", raising=False) + monkeypatch.delenv( + "VON_DURABLE_MIN_WORKER_GIT_SHORT_COMMIT", + raising=False, + ) manager = WorkflowInstanceManager() started: list[dict[str, Any]] = [] finalised: list[dict[str, Any]] = [] @@ -1814,7 +1834,7 @@ def test_claim_and_terminal_updates_record_workflow_episodes( success = manager.mark_failed( instance_id, - error="tool_timeout:Workflow step timed out", + error=error, error_step="tool_execution", worker_id="worker-episode-test", claim_token=claimed.claim_token, @@ -1824,7 +1844,8 @@ def test_claim_and_terminal_updates_record_workflow_episodes( assert finalised[0]["workflow_id"] == "#V#test_workflow" assert finalised[0]["completed"] is False assert finalised[0]["terminal_stage"] == "tool_execution" - assert finalised[0]["termination_code"] == "tool_timeout" + assert finalised[0]["termination_code"] == expected_code + assert finalised[0]["termination_detail"] == error def test_find_and_claim_prioritises_active_conversation_workflows(self) -> None: """Active user work is prioritised without reviving the retired controller.""" diff --git a/tests/backend/test_orchestrator_durable_instance_telemetry.py b/tests/backend/test_orchestrator_durable_instance_telemetry.py index 7337b1cc..477aa282 100644 --- a/tests/backend/test_orchestrator_durable_instance_telemetry.py +++ b/tests/backend/test_orchestrator_durable_instance_telemetry.py @@ -519,6 +519,89 @@ def test_execute_workflow_marks_failure_like_terminal_state_as_failed( ) +@pytest.mark.parametrize( + ("error", "final_state", "expected_code", "expected_detail"), + [ + ( + "paper_reference_ingestion_no_items_succeeded", + "failed", + "paper_reference_ingestion_no_items_succeeded", + "paper_reference_ingestion_no_items_succeeded", + ), + ( + "tool_timeout:Workflow step timed out", + "failed", + "tool_timeout", + "tool_timeout:Workflow step timed out", + ), + ( + "Unexpected workflow error while downloading", + "failed", + "terminated", + "Unexpected workflow error while downloading", + ), + ("", "stopped", "terminated", None), + ], +) +def test_execute_workflow_records_only_machine_failure_codes_in_episode_telemetry( + monkeypatch, + error: str, + final_state: str, + expected_code: str, + expected_detail: str | None, +) -> None: + orchestrator = _build_orchestrator() + workflow_id = "#V#machine_failure_code_workflow" + _register_test_workflow(orchestrator, workflow_id=workflow_id) + fake_manager = _FakeWorkflowInstanceManager() + _patch_submit_verified_instance(monkeypatch) + monkeypatch.setattr( + "src.backend.workflows.durable.WorkflowInstanceManager", + lambda: fake_manager, + ) + monkeypatch.setattr( + orchestrator._workflow_executor, + "run", + lambda *_args, **_kwargs: SimpleNamespace( + completed=False, + final_state=final_state, + error=error, + data={}, + ), + ) + input_data = { + "prompt": "Exercise failure telemetry.", + "user_concept_id": "#V#user", + "org_concept_id": "#V#org", + "conversation_session_id": "chat-failure-code", + "turn_id": "turn-failure-code", + "aux_llm_calls": [], + } + + result = orchestrator.execute_workflow( + workflow_id, + data=input_data, + llm_client=object(), + model="test-model", + user_namespace="#V#user@org", + conversation_session_id="chat-failure-code", + turn_id="turn-failure-code", + episode_source="chat_turn_workflow", + ) + + assert result is not None + episode_entries = [ + entry + for entry in input_data["aux_llm_calls"] + if isinstance(entry, dict) and entry.get("type") == "workflow_use_episode" + ] + assert len(episode_entries) == 1 + assert episode_entries[0]["termination_reason"] == { + "code": expected_code, + "detail": expected_detail, + } + + def test_execute_workflow_uses_explicit_failure_detail_for_failure_like_terminal_state( monkeypatch, ) -> None: 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 e7d33744..34e4ca65 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 @@ -1100,6 +1100,68 @@ def test_rag_get_item_supports_turn_execution_records(monkeypatch): "private_detail": "not projected", }, }, + "canonical_reconciliation": { + "schema_version": ( + "ontology_mutation_canonical_reconciliation.v1" + ), + "phase": "canonical_reconciliation", + "status": "verified", + "verified": True, + "effect_status": "succeeded", + "changed": False, + "method_name": "add_relationship", + "receipt_id": "canonical-receipt-9", + "intent_fingerprint": "canonical-intent-9", + "target_concept_ids": ["#V#target_9"], + "evidence_id": "canonical-evidence-9", + "recorded_at_utc": "2026-02-19T01:00:03Z", + "private_detail": "not projected", + }, + }, + "effect_weak_then_late": { + "identity": { + "schema_version": "effect_observation_journal.v1", + "effect_id": "effect_weak_then_late", + "call_id": "call-weak-then-late", + "capability_name": "create_concepts", + }, + "turn_terminal": { + "phase": "turn_terminal", + "effect_status": "indeterminate", + "changed": None, + "recorded_at_utc": "2026-02-19T02:00:01Z", + "transport": { + "execution_id": "mcp-weak-then-late", + "outcome": "timed_out", + }, + "receipt": {"mutation_outcome": "unknown"}, + }, + "current_state_observation": { + "phase": "current_state_observation", + "effect_status": "indeterminate", + "changed": None, + "initial_effect_status": "indeterminate", + "current_outcome_status": "target_observed", + "outcome_resolved": True, + "reconciliation_basis": "later_exact_current_state_read", + "observation_identity_sha256": "weak-observation-identity", + "target_concept_ids": ["#V#weakly_observed_target"], + "evidence_id": "weak-observation-evidence", + "recorded_at_utc": "2026-02-19T02:00:02Z", + }, + "late_terminal": { + "phase": "late_terminal", + "outcome": "late_success", + "effect_status": "succeeded", + "changed": True, + "recorded_at_utc": "2026-02-19T02:00:03Z", + "payload": { + "success": True, + "effect_status": "succeeded", + "changed": True, + "mutation_outcome": "succeeded", + }, + }, }, "effect_unresolved": { "identity": { @@ -1191,13 +1253,44 @@ def test_rag_get_item_supports_turn_execution_records(monkeypatch): }, } ] - assert result["effect_observation_journal_count"] == 4 + assert result["effect_observation_journal_count"] == 5 assert result["effect_observation_journal_truncated"] is False journal_by_id = { item["effect_id"]: item for item in result["effect_observation_journal"] } - assert journal_by_id["effect_2"]["latest_phase"] == "late_terminal" + assert ( + journal_by_id["effect_2"]["latest_phase"] + == "canonical_reconciliation" + ) + assert journal_by_id["effect_2"]["historical_phases"] == [ + "dispatch_intent", + "turn_terminal", + "late_terminal", + "canonical_reconciliation", + ] + assert ( + journal_by_id["effect_2"]["turn_terminal"]["effect_status"] + == "indeterminate" + ) + assert ( + journal_by_id["effect_2"]["late_terminal"]["effect_status"] + == "succeeded" + ) + assert journal_by_id["effect_2"]["current_outcome"] == { + "schema_version": "ontology_mutation_canonical_reconciliation.v1", + "phase": "canonical_reconciliation", + "recorded_at_utc": "2026-02-19T01:00:03Z", + "status": "verified", + "verified": True, + "effect_status": "succeeded", + "changed": False, + "method_name": "add_relationship", + "receipt_id": "canonical-receipt-9", + "intent_fingerprint": "canonical-intent-9", + "target_concept_ids": ["#V#target_9"], + "evidence_id": "canonical-evidence-9", + } assert journal_by_id["effect_2"]["outcome_resolved"] is True assert journal_by_id["effect_2"]["late_terminal"]["receipt"] == { "success": True, @@ -1205,6 +1298,41 @@ def test_rag_get_item_supports_turn_execution_records(monkeypatch): "changed": True, } assert "private_detail" not in json.dumps(journal_by_id["effect_2"]) + weak_then_late = journal_by_id["effect_weak_then_late"] + assert weak_then_late["historical_phases"] == [ + "turn_terminal", + "current_state_observation", + "late_terminal", + ] + assert weak_then_late["latest_phase"] == "late_terminal" + assert weak_then_late["current_outcome"] == { + "phase": "late_terminal", + "recorded_at_utc": "2026-02-19T02:00:03Z", + "outcome": "late_success", + "effect_status": "succeeded", + "changed": True, + "receipt": { + "success": True, + "effect_status": "succeeded", + "changed": True, + "mutation_outcome": "succeeded", + }, + } + assert weak_then_late["outcome_resolved"] is True + assert weak_then_late["turn_terminal"]["effect_status"] == "indeterminate" + assert weak_then_late["current_state_observation"] == { + "phase": "current_state_observation", + "recorded_at_utc": "2026-02-19T02:00:02Z", + "effect_status": "indeterminate", + "changed": None, + "initial_effect_status": "indeterminate", + "current_outcome_status": "target_observed", + "outcome_resolved": True, + "reconciliation_basis": "later_exact_current_state_read", + "observation_identity_sha256": "weak-observation-identity", + "target_concept_ids": ["#V#weakly_observed_target"], + "evidence_id": "weak-observation-evidence", + } assert journal_by_id["effect_unresolved"]["outcome_resolved"] is False assert ( journal_by_id["effect_missing_late_payload"]["outcome_resolved"] 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 c0e78ce5..3c428635 100644 --- a/tests/backend/test_turn_execution_record_service_execution_summary.py +++ b/tests/backend/test_turn_execution_record_service_execution_summary.py @@ -1004,6 +1004,23 @@ def test_effect_observation_journal_is_actor_scoped_idempotent_and_survives_upse }, **scope, ) + current_state_observation = record_effect_observation_phase( + request_id="req-effect-journal", + effect_id="effect_abc123", + phase="current_state_observation", + observation={ + "effect_status": "indeterminate", + "changed": None, + "initial_effect_status": "indeterminate", + "current_outcome_status": "target_observed", + "outcome_resolved": True, + "reconciliation_basis": "later_exact_current_state_read", + "observation_identity_sha256": "weak-current-state-identity", + "target_concept_ids": ["#V#canonical_target"], + "evidence_id": "weak-current-state-evidence", + }, + **scope, + ) late = record_effect_observation_phase( request_id="req-effect-journal", effect_id="effect_abc123", @@ -1018,6 +1035,33 @@ def test_effect_observation_journal_is_actor_scoped_idempotent_and_survives_upse }, **scope, ) + reconciliation = record_effect_observation_phase( + request_id="req-effect-journal", + effect_id="effect_abc123", + phase="canonical_reconciliation", + observation={ + "schema_version": "ontology_mutation_canonical_reconciliation.v1", + "effect_status": "succeeded", + "changed": True, + "receipt_id": "receipt-canonical-1", + "intent_fingerprint": "intent-fingerprint-1", + "target_concept_ids": ["#V#canonical_target"], + "evidence_id": "evidence-canonical-1", + }, + **scope, + ) + duplicate_reconciliation = record_effect_observation_phase( + request_id="req-effect-journal", + effect_id="effect_abc123", + phase="canonical_reconciliation", + observation={ + "schema_version": "ontology_mutation_canonical_reconciliation.v1", + "effect_status": "failed", + "changed": False, + "receipt_id": "different-receipt-must-not-replace", + }, + **scope, + ) duplicate_late = record_effect_observation_phase( request_id="req-effect-journal", effect_id="effect_abc123", @@ -1031,7 +1075,16 @@ def test_effect_observation_journal_is_actor_scoped_idempotent_and_survives_upse ) assert terminal["updated"] is True + assert current_state_observation["updated"] is True assert late["updated"] is True + assert reconciliation["updated"] is True + assert reconciliation["stored_phase"]["receipt_id"] == "receipt-canonical-1" + assert duplicate_reconciliation["updated"] is False + assert duplicate_reconciliation["duplicate"] is True + assert ( + duplicate_reconciliation["stored_phase"]["receipt_id"] + == "receipt-canonical-1" + ) assert duplicate_late["updated"] is False assert duplicate_late["duplicate"] is True stored = collection.find_one({"request_id": "req-effect-journal"}) @@ -1039,7 +1092,18 @@ def test_effect_observation_journal_is_actor_scoped_idempotent_and_survives_upse assert journal["identity"] == identity assert journal["identity"]["call_id"] == "call-1" assert journal["identity"]["capability_name"] == "upsert_text_relation" + assert journal["turn_terminal"]["effect_status"] == "indeterminate" + assert ( + journal["current_state_observation"]["current_outcome_status"] + == "target_observed" + ) assert journal["late_terminal"]["effect_status"] == "succeeded" + assert journal["canonical_reconciliation"]["effect_status"] == "succeeded" + assert reconciliation["stored_phase"] == journal["canonical_reconciliation"] + assert ( + duplicate_reconciliation["stored_phase"] + == journal["canonical_reconciliation"] + ) full_record = build_turn_execution_record( request_id="req-effect-journal", diff --git a/tests/backend/test_von_generate_workflow_instances.py b/tests/backend/test_von_generate_workflow_instances.py index 8b5f6eb3..2dc83c32 100644 --- a/tests/backend/test_von_generate_workflow_instances.py +++ b/tests/backend/test_von_generate_workflow_instances.py @@ -27,7 +27,7 @@ def app(monkeypatch: pytest.MonkeyPatch) -> Flask: "extra_messages": (), "tool_invocations": (), "terminal_status": "completed", - "effect_finality_fallback": False, + "response_authority": "model", } def _execute_adaptive_turn(**kwargs: Any) -> AdaptiveTurnResult: @@ -48,9 +48,7 @@ def _execute_adaptive_turn(**kwargs: Any) -> AdaptiveTurnResult: duration_ms=2.0, render_plan=adaptive_state["render_plan"], terminal_status=adaptive_state["terminal_status"], - effect_finality_fallback=adaptive_state[ - "effect_finality_fallback" - ], + response_authority=adaptive_state["response_authority"], ) def _retired_outer_controller_called(*_args: Any, **_kwargs: Any) -> None: @@ -578,7 +576,7 @@ def test_presenter_mode_recovers_unclosed_terminal_screen_block(app: Flask) -> N assert payload["llm_debug"]["screen_backfill_second_pass_attempted"] is False -def test_effect_finality_fallback_is_presented_literally_without_model_backfill( +def test_canonical_outcome_report_is_presented_literally_without_model_backfill( app: Flask, monkeypatch: pytest.MonkeyPatch, ) -> None: @@ -591,10 +589,10 @@ def test_effect_finality_fallback_is_presented_literally_without_model_backfill( adaptive_state = app.config["_ADAPTIVE_TURN_STATE"] adaptive_state["response_text"] = fallback adaptive_state["terminal_status"] = "model_error" - adaptive_state["effect_finality_fallback"] = True + adaptive_state["response_authority"] = "canonical_outcome" def _unexpected_backfill(*_args: Any, **_kwargs: Any) -> str: - raise AssertionError("effect-finality fallback must remain literal") + raise AssertionError("canonical outcome report must remain literal") monkeypatch.setattr( von_routes, @@ -619,7 +617,7 @@ def _unexpected_backfill(*_args: Any, **_kwargs: Any) -> str: assert payload["response_channels"] == { "spoken": fallback, "screen": fallback, - "format": "effect_finality_fallback_v1", + "format": "effect_outcome_report_v1", } assert ( payload["llm_debug"]["screen_backfill_second_pass_attempted"] is False @@ -627,6 +625,17 @@ def _unexpected_backfill(*_args: Any, **_kwargs: Any) -> str: assert ( payload["llm_debug"]["spoken_backfill_second_pass_attempted"] is False ) + transformations = payload["llm_debug"]["response_transformations"][ + "transformations" + ] + for transform_name in ("screen_backfill", "spoken_backfill"): + event = next( + item + for item in transformations + if item.get("transform_name") == transform_name + ) + assert event["status"] == "skipped" + assert event["suppression_reason"] == "canonical_outcome_report" def test_pending_effect_answer_reaches_presenter_channels(app: Flask) -> None: @@ -638,7 +647,7 @@ def test_pending_effect_answer_reaches_presenter_channels(app: Flask) -> None: f"{instance_id}." ) adaptive_state["terminal_status"] = "effect_partially_completed" - adaptive_state["effect_finality_fallback"] = False + adaptive_state["response_authority"] = "model" response = app.test_client().post( "/von/generate",