From ce363c00c5ceb9f481ac6a56bf86c92d04d09f60 Mon Sep 17 00:00:00 2001 From: witbrock Date: Mon, 27 Jul 2026 11:17:53 +0100 Subject: [PATCH] JVNAUTOSCI-2600 preserve late effect finality --- .../integrations/internal_mcp/catalogue.py | 175 ++--- .../integrations/internal_mcp/gateway.py | 64 +- .../integrations/internal_mcp/transport.py | 212 +++++- src/backend/mcp_server/mcp_stdio_server.py | 1 + src/backend/server/routes/von_routes.py | 21 +- src/backend/services/adaptive_turn_service.py | 624 ++++++++++++++-- .../services/turn_execution_record_service.py | 207 +++++- tests/backend/test_adaptive_turn_service.py | 682 +++++++++++++++++- .../test_internal_mcp_transport_deadlines.py | 288 +++++++- ...est_mcp_create_concepts_extent_batching.py | 181 +++++ ...exposes_turn_execution_diagnostics_tool.py | 23 + ...cution_record_service_execution_summary.py | 157 ++++ .../test_von_generate_workflow_instances.py | 58 +- 13 files changed, 2522 insertions(+), 171 deletions(-) create mode 100644 tests/backend/test_mcp_create_concepts_extent_batching.py diff --git a/src/backend/integrations/internal_mcp/catalogue.py b/src/backend/integrations/internal_mcp/catalogue.py index 634ab70a9..93b7f2554 100644 --- a/src/backend/integrations/internal_mcp/catalogue.py +++ b/src/backend/integrations/internal_mcp/catalogue.py @@ -877,6 +877,9 @@ def _create_concepts(**kwargs): from ...services.create_concepts_parent_resolution_service import ( resolve_parent_for_create_concepts, ) + from ...services.relationship_extent_index_service import ( + defer_relationship_extent_index_sync, + ) parent_id: str | None = kwargs.get("parent_id") concepts = kwargs.get("concepts", []) @@ -1092,95 +1095,101 @@ def _create_concepts(**kwargs): ) results = [] - for concept_data in concepts: - # This is both the batch boundary and the safe cancellation point after - # the preceding concept's complete logical write bundle. - raise_if_internal_mcp_cancelled() - if not isinstance(concept_data, dict): - results.append({"error": "Concept must be an object", "data": concept_data}) - continue + with defer_relationship_extent_index_sync(): + for concept_data in concepts: + # This is both the batch boundary and the safe cancellation point after + # the preceding concept's complete logical write bundle. + raise_if_internal_mcp_cancelled() + if not isinstance(concept_data, dict): + results.append( + {"error": "Concept must be an object", "data": concept_data} + ) + continue - name = concept_data.get("name") - kind = (concept_data.get("kind") or "type").strip().lower() + name = concept_data.get("name") + kind = (concept_data.get("kind") or "type").strip().lower() - if not name: - results.append( - {"error": "Concept missing required 'name' field", "data": concept_data} - ) - continue + if not name: + results.append( + { + "error": "Concept missing required 'name' field", + "data": concept_data, + } + ) + continue - # Map kind to create_as_instance parameter - if kind == "individual": - kind = "instance" + # Map kind to create_as_instance parameter + if kind == "individual": + kind = "instance" - if kind == "predicate": - create_as_instance = True - parent_id_for_concept = PREDICATE_TYPE_ID - else: - create_as_instance = kind == "instance" - parent_id_for_concept = validated_parent_id - - duplicate_match = find_existing_concept_for_create_concepts( - concept_name=str(name), - kind=kind, - parent_id_for_concept=parent_id_for_concept, - preferred_language="en-NZ", - allow_duplicate_instances=allow_duplicate_instances, - duplicate_resolution_mode=( - _CREATE_CONCEPTS_DUPLICATE_RESOLUTION_CANONICAL_ID_ONLY - if canonical_id_only - else None - ), - ) - if duplicate_match is not None: - result = build_duplicate_prevented_create_concepts_result( - requested_name=str(name), - requested_kind=kind, - existing_concept_id=duplicate_match.existing_concept_id, - guard_scope=duplicate_match.guard_scope, - match_source=duplicate_match.match_source, + if kind == "predicate": + create_as_instance = True + parent_id_for_concept = PREDICATE_TYPE_ID + else: + create_as_instance = kind == "instance" + parent_id_for_concept = validated_parent_id + + duplicate_match = find_existing_concept_for_create_concepts( + concept_name=str(name), + kind=kind, + parent_id_for_concept=parent_id_for_concept, + preferred_language="en-NZ", + allow_duplicate_instances=allow_duplicate_instances, + duplicate_resolution_mode=( + _CREATE_CONCEPTS_DUPLICATE_RESOLUTION_CANONICAL_ID_ONLY + if canonical_id_only + else None + ), ) - result["concept_id"] = duplicate_match.existing_concept_id - results.append(result) - continue + if duplicate_match is not None: + result = build_duplicate_prevented_create_concepts_result( + requested_name=str(name), + requested_kind=kind, + existing_concept_id=duplicate_match.existing_concept_id, + guard_scope=duplicate_match.guard_scope, + match_source=duplicate_match.match_source, + ) + result["concept_id"] = duplicate_match.existing_concept_id + results.append(result) + continue - # Duplicate preflights are read-only and may consume the entire - # transport budget. Never start an insert after cancellation. - raise_if_internal_mcp_cancelled() - result = create_vontology_concept( - parent_id=parent_id_for_concept, - new_concept_name=name, - create_as_instance=create_as_instance, - description=concept_data.get("description"), - notes=concept_data.get("notes"), - created_by_concept_id=actor_user_id, - organisation_concept_id=actor_org_id, - event_namespace=namespace, - visibility_scope_mode=scope_mode, - ) - # Enrich result with the requested name for traceability and surface - # the canonical created concept_id at a stable top-level key so UI - # summaries can reliably name what was created. - result["requested_name"] = name - result["requested_kind"] = kind - concept_id_value = result.get("concept_id") - if not isinstance(concept_id_value, str) or not concept_id_value.strip(): - nested_concept = result.get("concept") - if isinstance(nested_concept, dict): - nested_id = nested_concept.get("concept_id") - if isinstance(nested_id, str) and nested_id.strip(): - concept_id_value = nested_id - if not isinstance(concept_id_value, str) or not concept_id_value.strip(): - canonical_id = result.get("canonical_concept_id") - if isinstance(canonical_id, str) and canonical_id.strip(): - concept_id_value = canonical_id - if not isinstance(concept_id_value, str) or not concept_id_value.strip(): - existing_id = result.get("existing_concept_id") - if isinstance(existing_id, str) and existing_id.strip(): - concept_id_value = existing_id - if isinstance(concept_id_value, str) and concept_id_value.strip(): - result["concept_id"] = concept_id_value.strip() - results.append(result) + # Duplicate preflights are read-only and may consume the entire + # transport budget. Never start an insert after cancellation. + raise_if_internal_mcp_cancelled() + result = create_vontology_concept( + parent_id=parent_id_for_concept, + new_concept_name=name, + create_as_instance=create_as_instance, + description=concept_data.get("description"), + notes=concept_data.get("notes"), + created_by_concept_id=actor_user_id, + organisation_concept_id=actor_org_id, + event_namespace=namespace, + visibility_scope_mode=scope_mode, + ) + # Enrich result with the requested name for traceability and surface + # the canonical created concept_id at a stable top-level key so UI + # summaries can reliably name what was created. + result["requested_name"] = name + result["requested_kind"] = kind + concept_id_value = result.get("concept_id") + if not isinstance(concept_id_value, str) or not concept_id_value.strip(): + nested_concept = result.get("concept") + if isinstance(nested_concept, dict): + nested_id = nested_concept.get("concept_id") + if isinstance(nested_id, str) and nested_id.strip(): + concept_id_value = nested_id + if not isinstance(concept_id_value, str) or not concept_id_value.strip(): + canonical_id = result.get("canonical_concept_id") + if isinstance(canonical_id, str) and canonical_id.strip(): + concept_id_value = canonical_id + if not isinstance(concept_id_value, str) or not concept_id_value.strip(): + existing_id = result.get("existing_concept_id") + if isinstance(existing_id, str) and existing_id.strip(): + concept_id_value = existing_id + if isinstance(concept_id_value, str) and concept_id_value.strip(): + result["concept_id"] = concept_id_value.strip() + results.append(result) # Count different outcome types for summary successful = sum(1 for r in results if isinstance(r, dict) and r.get("success")) diff --git a/src/backend/integrations/internal_mcp/gateway.py b/src/backend/integrations/internal_mcp/gateway.py index 6bb12e888..bc68052b5 100644 --- a/src/backend/integrations/internal_mcp/gateway.py +++ b/src/backend/integrations/internal_mcp/gateway.py @@ -15,7 +15,11 @@ normalise_payload_aliases, validate_payload, ) -from .transport import InternalMCPTransport, TransportResult +from .transport import ( + InternalMCPTransport, + LateCompletionObserver, + TransportResult, +) logger = logging.getLogger(__name__) @@ -339,6 +343,7 @@ def invoke( payload: Optional[MutableMapping[str, Any]] = None, *, deadline_monotonic: float | None = None, + late_completion_observer: LateCompletionObserver | None = None, ) -> TransportResult: if not self._enabled: raise GatewayDisabledError("Internal MCP gateway is disabled.") @@ -382,6 +387,62 @@ def invoke( definition.category, hard_timeout_sec=float(timeout or self._transport.read_timeout_sec), ) + observed_late_completion = late_completion_observer + if late_completion_observer is not None: + + def _observe_validated_late_completion( + observation: Dict[str, Any], + ) -> None: + enriched = dict(observation) + result_payload = enriched.get("payload") + if enriched.get("outcome") != "late_success": + validation = "handler_error" + valid: bool | None = None + validation_error = None + elif enriched.get("payload_truncated") is True: + validation = "indeterminate_truncated" + valid = None + validation_error = "Late handler payload was truncated." + elif definition.output_schema is None: + validation = "not_configured" + valid = None + validation_error = None + elif not isinstance(result_payload, MutableMapping): + validation = "invalid" + valid = False + validation_error = ( + "Output schema provided but late handler returned " + "a non-mapping payload." + ) + elif result_payload.get("success") is False: + validation = "standard_error_response" + valid = True + validation_error = None + else: + try: + valid, validation_errors = validate_payload( + definition.output_schema, + result_payload, + ) + except Exception as exc: + valid = False + validation_errors = [type(exc).__name__] + validation = "valid" if valid else "invalid" + validation_error = ( + None + if valid + else "; ".join(validation_errors)[:2_000] + ) + enriched.update( + { + "output_schema_validation": validation, + "output_schema_valid": valid, + "output_schema_error": validation_error, + } + ) + late_completion_observer(enriched) + + observed_late_completion = _observe_validated_late_completion try: from src.backend.security.access_control import ( get_effective_organisation_concept_id, @@ -460,6 +521,7 @@ def invoke( advisory_timeout_sec=advisory_timeout, deadline_monotonic=deadline_monotonic, log_tag=self._log_tag, + late_completion_observer=observed_late_completion, ) finally: _PREEXISTING_ACTOR_CONTEXT.reset(preexisting_actor_token) diff --git a/src/backend/integrations/internal_mcp/transport.py b/src/backend/integrations/internal_mcp/transport.py index f617ab027..846356fd8 100644 --- a/src/backend/integrations/internal_mcp/transport.py +++ b/src/backend/integrations/internal_mcp/transport.py @@ -10,8 +10,10 @@ * the advisory budget is telemetry only and never changes a successful result; * the hard deadline is terminal for the current turn and returns a typed MCP - timeout payload. A handler that later completes is diagnostics-only and its - payload is discarded rather than allowed to rewrite the turn outcome. + timeout payload. A handler that later completes cannot rewrite that outcome. + Reads and calls without an observer discard the late payload; an explicitly + observed handler-started write may emit one bounded out-of-band completion + observation. The pool and its queue are both bounded. Python cannot forcibly stop an arbitrary running thread, so handlers may also use the cooperative cancellation @@ -22,6 +24,8 @@ from __future__ import annotations +import hashlib +import json import logging import os import queue @@ -31,7 +35,8 @@ from collections import deque from contextvars import Context, ContextVar, copy_context from dataclasses import dataclass, field -from typing import Any, Callable, Dict +from datetime import datetime, timezone +from typing import Any, Callable, Dict, Mapping logger = logging.getLogger(__name__) @@ -42,6 +47,18 @@ _DEFAULT_HANDLER_WORKER_COUNT = 8 _DEFAULT_HANDLER_QUEUE_CAPACITY = 32 _LATE_COMPLETION_HISTORY_LIMIT = 50 +_LATE_COMPLETION_MAX_PAYLOAD_CHARS = 32_000 +_LATE_COMPLETION_MAX_RECEIPT_FIELD_CHARS = 4_000 +_LATE_COMPLETION_RECEIPT_FIELDS = ( + "success", + "status", + "effect_status", + "changed", + "error", + "error_code", + "mutation_outcome", + "partial_failures", +) def _positive_float_env(name: str, default: float) -> float: @@ -60,6 +77,58 @@ def _positive_int_env(name: str, default: int) -> int: return value if value > 0 else int(default) +def _utc_now_iso() -> str: + return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") + + +def _bounded_late_completion_value(value: Any) -> tuple[Any, bool]: + """Return a small JSON-compatible handler-result projection.""" + + try: + serialised = json.dumps( + value, + default=str, + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + ) + except Exception as exc: + return { + "_truncated": True, + "_type": type(value).__name__, + "_serialisation_error": type(exc).__name__, + }, True + if len(serialised) <= _LATE_COMPLETION_MAX_PAYLOAD_CHARS: + return json.loads(serialised), False + + projection: dict[str, Any] = { + "_truncated": True, + "_original_char_count": len(serialised), + "_sha256": hashlib.sha256(serialised.encode("utf-8")).hexdigest(), + } + if isinstance(value, Mapping): + for key in _LATE_COMPLETION_RECEIPT_FIELDS: + if key not in value: + continue + try: + field_text = json.dumps( + value.get(key), + default=str, + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + ) + except Exception: + continue + if len(field_text) > _LATE_COMPLETION_MAX_RECEIPT_FIELD_CHARS: + projection[key] = ( + field_text[:_LATE_COMPLETION_MAX_RECEIPT_FIELD_CHARS] + "..." + ) + else: + projection[key] = json.loads(field_text) + return projection, True + + @dataclass(frozen=True) class InternalMCPExecutionScope: """Context available to a cooperatively cancellable MCP handler.""" @@ -127,6 +196,7 @@ class TransportResult: handler_elapsed_ms: float | None = None transport_overhead_ms: float | None = None timeout_phase: str | None = None + late_result_policy: str = "discard_from_turn" @property def timed_out(self) -> bool: @@ -148,10 +218,13 @@ def telemetry_metadata(self) -> Dict[str, Any]: "handler_elapsed_ms": self.handler_elapsed_ms, "transport_overhead_ms": self.transport_overhead_ms, "timeout_phase": self.timeout_phase, - "late_result_policy": "discard_from_turn", + "late_result_policy": self.late_result_policy, } +LateCompletionObserver = Callable[[Dict[str, Any]], None] + + @dataclass class _HandlerTask: execution_id: str @@ -162,6 +235,8 @@ class _HandlerTask: deadline_monotonic: float submitted_at: float on_late_completion: Callable[["_HandlerTask"], None] + category: str = "read" + late_completion_observer: LateCompletionObserver | None = None cancellation_event: threading.Event = field(default_factory=threading.Event) done_event: threading.Event = field(default_factory=threading.Event) lock: threading.Lock = field(default_factory=threading.Lock) @@ -172,6 +247,7 @@ class _HandlerTask: exception: BaseException | None = None terminal_returned: bool = False cancelled_before_start: bool = False + late_completion_recorded: bool = False def _invoke(self) -> Any: scope = InternalMCPExecutionScope( @@ -416,11 +492,68 @@ def _timing_snapshot( return queue_duration_ms, handler_duration_ms, handler_elapsed_ms def _record_late_completion(self, task: _HandlerTask) -> None: + with task.lock: + if task.late_completion_recorded: + return + task.late_completion_recorded = True terminal_at = task.completed_at or time.perf_counter() queue_ms, handler_ms, _ = self._timing_snapshot( task, terminal_at=terminal_at, ) + is_observable_write = ( + str(task.category or "").strip().lower() == "write" + and task.late_completion_observer is not None + ) + observer_notified = False + observer_error_type: str | None = None + if is_observable_write: + bounded_payload = None + payload_truncated = False + error_text = None + if task.exception is None: + bounded_payload, payload_truncated = ( + _bounded_late_completion_value(task.result) + ) + else: + bounded_error, payload_truncated = ( + _bounded_late_completion_value(str(task.exception)) + ) + error_text = str(bounded_error) + observation = { + "schema_version": "internal_mcp_late_completion.v1", + "execution_id": task.execution_id, + "method_name": task.method_name, + "category": "write", + "outcome": ( + "late_error" if task.exception is not None else "late_success" + ), + "observed_at_utc": _utc_now_iso(), + "queue_duration_ms": queue_ms, + "handler_duration_ms": handler_ms, + "payload": bounded_payload, + "payload_truncated": payload_truncated, + "error_type": ( + type(task.exception).__name__ + if task.exception is not None + else None + ), + "error": error_text, + "output_schema_validation": "not_checked", + "output_schema_valid": None, + "output_schema_error": None, + } + try: + task.late_completion_observer(observation) + observer_notified = True + except BaseException as exc: + observer_error_type = type(exc).__name__ + logger.exception( + "[mcp_transport] late-completion observer failed for %s " + "(execution_id=%s)", + task.method_name, + task.execution_id, + ) with self._diagnostics_lock: self._late_completion_count += 1 self._late_completions.append( @@ -432,11 +565,15 @@ def _record_late_completion(self, task: _HandlerTask) -> None: ), "queue_duration_ms": queue_ms, "handler_duration_ms": handler_ms, - "payload_discarded": True, + "payload_discarded": not observer_notified, + "observer_notified": observer_notified, + "observer_error_type": observer_error_type, } ) logger.warning( - "[mcp_transport] discarded late %s for %s (execution_id=%s, handler=%.2fms)", + "[mcp_transport] %s late %s for %s " + "(execution_id=%s, handler=%.2fms)", + "observed" if observer_notified else "discarded", "error" if task.exception is not None else "result", task.method_name, task.execution_id, @@ -454,9 +591,10 @@ def _timeout_payload( timeout_phase: str, queue_duration_ms: float, handler_elapsed_ms: float | None, + late_result_policy: str = "discard_from_turn", ) -> Dict[str, Any]: is_write = str(category or "").strip().lower() == "write" - outcome_unknown = is_write and timeout_phase != "pre_dispatch" + outcome_unknown = is_write and timeout_phase == "handler" payload: Dict[str, Any] = { "success": False, "status": "timed_out", @@ -480,7 +618,7 @@ def _timeout_payload( "cancellation_requested": True, "handler_isolation": "bounded_worker_pool", "outcome_finality": "terminal_for_turn", - "late_result_policy": "discard_from_turn", + "late_result_policy": late_result_policy, } if outcome_unknown: payload["mutation_outcome"] = "unknown" @@ -488,6 +626,8 @@ def _timeout_payload( {"action_type": "inspect_operation_state_before_retry"} ] else: + if is_write: + payload["mutation_outcome"] = "not_started" payload["recovery_affordances"] = [ {"action_type": "bounded_retry"}, {"action_type": "choose_alternate_represented_path"}, @@ -505,27 +645,26 @@ def _saturation_payload( advisory_timeout_sec: float, ) -> Dict[str, Any]: is_write = str(category or "").strip().lower() == "write" - return { + payload: Dict[str, Any] = { "success": False, "status": "failed", "error": "The bounded internal MCP handler pool is saturated.", "error_code": "internal_mcp_handler_pool_saturated", "error_type": "capacity_exhausted", - "retryable": not is_write, + "retryable": True, "execution_id": execution_id, "timeout_seconds": timeout_sec, "advisory_timeout_seconds": advisory_timeout_sec, "handler_isolation": "bounded_worker_pool", "outcome_finality": "terminal_for_turn", - "recovery_affordances": ( - [{"action_type": "inspect_operation_state_before_retry"}] - if is_write - else [ - {"action_type": "bounded_retry"}, - {"action_type": "choose_alternate_represented_path"}, - ] - ), + "recovery_affordances": [ + {"action_type": "bounded_retry"}, + {"action_type": "choose_alternate_represented_path"}, + ], } + if is_write: + payload["mutation_outcome"] = "not_started" + return payload def execute( self, @@ -538,14 +677,18 @@ def execute( advisory_timeout_sec: float | None = None, deadline_monotonic: float | None = None, log_tag: str = "[mcp_gateway]", + late_completion_observer: LateCompletionObserver | None = None, ) -> TransportResult: """Execute a handler within a bounded hard deadline. The returned result is immutable for the current turn. If the handler - ignores cooperative cancellation and completes late, only bounded - aggregate diagnostics are updated; the late payload is never surfaced. - A caller may provide an absolute monotonic deadline to shorten, but - never extend, the method's configured hard timeout. + ignores cooperative cancellation and completes late, the terminal + result remains unchanged. Reads and calls without an observer discard + the late payload. A dispatched write with an observer emits exactly + one bounded observation out of band after the handler starts. A write + cancelled while still queued reports ``not_started`` and cannot promise + an observation. A caller may provide an absolute monotonic deadline to + shorten, but never extend, the method's configured hard timeout. """ configured_hard_timeout_sec = float( @@ -576,6 +719,15 @@ def execute( advisory_sec = min(advisory_sec, hard_timeout_sec) execution_id = f"mcp_{uuid.uuid4().hex}" handler_deadline_monotonic = submitted_monotonic + hard_timeout_sec + observe_late_write = ( + str(category or "").strip().lower() == "write" + and late_completion_observer is not None + ) + dispatched_late_result_policy = ( + "observe_out_of_band" + if observe_late_write + else "discard_from_turn" + ) if hard_timeout_sec <= 0.0: with self._diagnostics_lock: @@ -614,6 +766,10 @@ def execute( deadline_monotonic=handler_deadline_monotonic, submitted_at=submitted_at, on_late_completion=self._record_late_completion, + category=category, + late_completion_observer=( + late_completion_observer if observe_late_write else None + ), ) logger.info( "%s invoking %s (advisory=%.1fs, hard_deadline=%.1fs, execution_id=%s)", @@ -663,6 +819,11 @@ def execute( else: task.terminal_returned = True task.cancellation_event.set() + if task.started_at is None: + # A queued task will observe cancellation before invoking + # its handler. Remove the observer while holding the task + # lock so no future worker path can imply otherwise. + task.late_completion_observer = None task_completed = False record_already_completed_late = bool( task.done_event.is_set() @@ -685,6 +846,11 @@ def execute( with self._diagnostics_lock: self._timeout_count += 1 timeout_phase = "handler" if started_at is not None else "queue" + late_result_policy = ( + dispatched_late_result_policy + if timeout_phase == "handler" + else "discard_from_turn" + ) transport_overhead_ms = max( 0.0, duration_ms - queue_ms - (handler_elapsed_ms or 0.0), @@ -709,6 +875,7 @@ def execute( timeout_phase=timeout_phase, queue_duration_ms=queue_ms, handler_elapsed_ms=handler_elapsed_ms, + late_result_policy=late_result_policy, ) return TransportResult( payload=timeout_payload, @@ -723,6 +890,7 @@ def execute( handler_elapsed_ms=handler_elapsed_ms, transport_overhead_ms=transport_overhead_ms, timeout_phase=timeout_phase, + late_result_policy=late_result_policy, ) if exception is not None: diff --git a/src/backend/mcp_server/mcp_stdio_server.py b/src/backend/mcp_server/mcp_stdio_server.py index b375febbc..e80ebc80d 100644 --- a/src/backend/mcp_server/mcp_stdio_server.py +++ b/src/backend/mcp_server/mcp_stdio_server.py @@ -410,6 +410,7 @@ def _bind_imports(module_name: str, names: list[str]) -> None: "_chat_history_get_debug_entry", "_chat_history_get_segments", "_conversation_telemetry_get_locator", + "_mongo_query_diagnostics_report", "_turn_execution_get", "_turn_execution_get_diagnostics", "_turn_execution_get_live_progress", diff --git a/src/backend/server/routes/von_routes.py b/src/backend/server/routes/von_routes.py index d203e7879..33b7b08b4 100644 --- a/src/backend/server/routes/von_routes.py +++ b/src/backend/server/routes/von_routes.py @@ -10991,6 +10991,9 @@ def _progress_update(info: dict[str, Any]) -> None: ) adaptive_success = adaptive_terminal_status == "completed" response_text = adaptive_turn_result.response_text + effect_finality_fallback = bool( + adaptive_turn_result.effect_finality_fallback + ) tool_messages = [dict(msg) for msg in adaptive_turn_result.extra_messages] tool_invocations = list(adaptive_turn_result.tool_invocations) auxiliary_llm_calls.extend(adaptive_turn_result.aux_llm_calls) @@ -11023,7 +11026,15 @@ def _progress_update(info: dict[str, Any]) -> None: ) rag_trace["tool_results_included_in_prompt"] = bool(tool_messages) - presenter_channels = _extract_presenter_channels(response_text) + presenter_channels = ( + { + "screen": response_text, + "spoken": response_text, + "format": "effect_finality_fallback_v1", + } + if presenter_mode_requested and effect_finality_fallback + else _extract_presenter_channels(response_text) + ) current_turn_messages = [{"role": "user", "content": prompt_text}] if tool_messages: current_turn_messages.extend(tool_messages) @@ -11053,7 +11064,7 @@ def _progress_update(info: dict[str, Any]) -> None: default=True ) - if presenter_mode_requested: + if presenter_mode_requested and not effect_finality_fallback: 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( @@ -11497,6 +11508,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: + screen_backfill_status = "skipped" + screen_backfill_suppression_reason = "effect_finality_literal" elif not needs_screen_backfill: screen_backfill_status = "skipped" screen_backfill_suppression_reason = "not_required" @@ -11818,6 +11832,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: + spoken_backfill_status = "skipped" + spoken_backfill_suppression_reason = "effect_finality_literal" 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 d987972dd..56eb2e36b 100644 --- a/src/backend/services/adaptive_turn_service.py +++ b/src/backend/services/adaptive_turn_service.py @@ -14,6 +14,7 @@ import json import os import re +import threading import time from collections.abc import Mapping, MutableMapping, Sequence from concurrent.futures import ThreadPoolExecutor @@ -56,6 +57,10 @@ } _DEFAULT_TURN_BUDGET_SECONDS = 180.0 _DEFAULT_FINAL_RESERVE_SECONDS = 30.0 +# The answer checkpoint is an experimental allocation seam, disabled unless a +# candidate environment or caller supplies a value. It is not Von's theory of +# how much thought a task deserves. +_DEFAULT_FINAL_ANSWER_RESERVE_SECONDS = 0.0 _DEFAULT_OUTER_TOOL_WORKERS = 8 _MODEL_EVIDENCE_INDEX_MAX_BYTES = 24_000 _MODEL_TOOL_RESULT_BATCH_MAX_BYTES = 24_000 @@ -76,6 +81,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 def _positive_float_env(name: str, default: float) -> float: @@ -94,6 +100,14 @@ def _positive_int_env(name: str, default: int) -> int: return value if value > 0 else int(default) +def _non_negative_float_env(name: str, default: float) -> float: + try: + value = float(os.getenv(name, str(default))) + except (TypeError, ValueError): + return float(default) + return value if value >= 0.0 else float(default) + + def _json_bytes(value: Any) -> bytes: return json.dumps( value, @@ -663,7 +677,17 @@ def _ordered_unique_evidence_views( if not isinstance(item, Mapping): continue evidence_id = item.get("evidence_id") - if not isinstance(evidence_id, str) or not evidence_id.strip(): + is_index_page = ( + item.get("schema_version") + == "adaptive_turn_evidence_index_page.v1" + ) + if ( + not is_index_page + and ( + not isinstance(evidence_id, str) + or not evidence_id.strip() + ) + ): continue view = dict(item) digest = hashlib.sha256(_json_bytes(view)).hexdigest() @@ -818,6 +842,8 @@ def _final_synthesis_context( context: Sequence[Mapping[str, Any]], evidence_index: Sequence[Mapping[str, Any]], evidence_views: Sequence[Mapping[str, Any]] = (), + *, + draft_text: str | None = None, ) -> list[dict[str, Any]]: """Build a fresh synthesis request with usable, bounded tool evidence.""" @@ -827,6 +853,14 @@ def _final_synthesis_context( if isinstance(item, Mapping) and str(item.get("role") or "").strip().lower() != "tool" ] + clean_draft = draft_text.strip() if isinstance(draft_text, str) else "" + if clean_draft: + fresh_context.append( + { + "role": "assistant", + "content": clean_draft, + } + ) if evidence_index or evidence_views: evidence_context = _bounded_evidence_context_message( evidence_index, @@ -1024,6 +1058,7 @@ def _scope_message( *, delegated_count: int, final_synthesis: bool, + answer_only: bool = False, ) -> str: actor = scope.user_concept_id or "unauthenticated" organisation = scope.organisation_concept_id or "none" @@ -1048,7 +1083,14 @@ def _scope_message( "identifiers and delegated reads to inspect canonical state before " "claiming that a representation persisted." ) - if final_synthesis: + if answer_only: + message += ( + "\n- The answer-only checkpoint has been reached. Answer now from the " + "bounded evidence already present in the request. Do not request " + "tools or new external capabilities. State material limitations " + "instead of filling evidence gaps." + ) + elif final_synthesis: message += ( "\n- The research deadline has ended. Answer now from the evidence " "already obtained. You may list or hydrate existing evidence, but " @@ -1531,6 +1573,28 @@ def _effect_status( return "failed" if bool(getattr(transport_result, "timed_out", False)) else "succeeded" +def _late_effect_observation_state( + observation: Mapping[str, Any], +) -> tuple[str, bool | None]: + """Interpret a validated late handler observation without claiming read-back.""" + + if observation.get("outcome") != "late_success": + return "indeterminate", None + if observation.get("output_schema_valid") is False: + return "indeterminate", None + if observation.get("output_schema_validation") == "indeterminate_truncated": + return "indeterminate", None + payload = observation.get("payload") + effect_status = _effect_status(payload, transport_result=None) + changed = ( + payload.get("changed") + if isinstance(payload, Mapping) + and isinstance(payload.get("changed"), bool) + else (False if effect_status == "failed" else None) + ) + return effect_status, changed + + def _error_payload(code: str, message: str, *, retryable: bool = False) -> dict[str, Any]: return { "success": False, @@ -1583,6 +1647,7 @@ def execute_adaptive_turn( turn_id: str | None = None, turn_budget_seconds: float | None = None, final_synthesis_reserve_seconds: float | None = None, + final_answer_reserve_seconds: float | None = None, clock: Any = time.monotonic, ) -> AdaptiveTurnResult: """Run one ordinary turn without a selector, master workflow, critic, or gate.""" @@ -1604,6 +1669,14 @@ def execute_adaptive_turn( _DEFAULT_FINAL_RESERVE_SECONDS, ) ) + requested_final_answer_reserve = float( + final_answer_reserve_seconds + if final_answer_reserve_seconds is not None + else _non_negative_float_env( + "VON_ADAPTIVE_TURN_FINAL_ANSWER_RESERVE_SEC", + _DEFAULT_FINAL_ANSWER_RESERVE_SECONDS, + ) + ) if turn_budget <= 0.0: raise ValueError("turn_budget_seconds must be positive") if not 0.0 < final_reserve < turn_budget: @@ -1611,8 +1684,15 @@ def execute_adaptive_turn( "final_synthesis_reserve_seconds must be positive and smaller than " "turn_budget_seconds" ) + if requested_final_answer_reserve < 0.0: + raise ValueError("final_answer_reserve_seconds must be non-negative") + final_answer_reserve = min( + requested_final_answer_reserve, + final_reserve, + ) turn_deadline = started + turn_budget research_deadline = turn_deadline - final_reserve + final_answer_deadline = turn_deadline - final_answer_reserve scope = TrustedTurnScope( user_concept_id=user_concept_id, @@ -1653,18 +1733,151 @@ def execute_adaptive_turn( seen_request_digests: set[str] = set() last_partial_text = "" final_synthesis = False + answer_only = False + final_context_base: list[dict[str, Any]] | None = None terminal_status = "completed" evidence_views: list[dict[str, Any]] = [] seen_evidence_view_digests: set[str] = set() + effect_state_lock = threading.RLock() + effect_states: dict[str, dict[str, Any]] = {} + effect_state_generation = 0 + last_partial_effect_generation = 0 + + def remember_effect_state( + effect_id: str, + *, + phase: int, + effect_status: str, + changed: bool | None, + execution_id: str | None = None, + late_observation: Mapping[str, Any] | None = None, + evidence_id: str | None = None, + ) -> None: + nonlocal effect_state_generation + with effect_state_lock: + existing = effect_states.get(effect_id) + if existing is not None and int(existing.get("phase") or 0) > phase: + return + state = { + "phase": phase, + "effect_status": effect_status, + "changed": changed, + "execution_id": execution_id, + "evidence_id": evidence_id, + } + if late_observation is not None: + state["late_observation"] = { + key: late_observation.get(key) + for key in ( + "schema_version", + "execution_id", + "method_name", + "outcome", + "observed_at_utc", + "output_schema_validation", + "output_schema_valid", + "output_schema_error", + "payload_truncated", + ) + if key in late_observation + } + effect_states[effect_id] = state + effect_state_generation += 1 + + def late_effect_observer( + *, + effect_id: str, + call_id: str, + capability_name: str, + ): + def observe(observation: Mapping[str, Any]) -> None: + observed = dict(observation) + effect_status, changed = _late_effect_observation_state(observed) + payload = observed.get("payload") + evidence_value = ( + payload + if observed.get("outcome") == "late_success" + else { + "success": False, + "error_code": "late_effect_handler_error", + "error": observed.get("error"), + "mutation_outcome": "unknown", + } + ) + envelope = evidence_store.record( + capability_name, + call_id, + evidence_value, + provenance={ + "namespace": scope.namespace, + "user_concept_id": scope.user_concept_id, + "organisation_concept_id": scope.organisation_concept_id, + "effect_id": effect_id, + "effect_status": effect_status, + "late_completion": True, + "execution_id": observed.get("execution_id"), + "output_schema_validation": observed.get( + "output_schema_validation" + ), + }, + status=effect_status, + ) + execution_id = str(observed.get("execution_id") or "").strip() or None + remember_effect_state( + effect_id, + phase=1, + effect_status=effect_status, + changed=changed, + execution_id=execution_id, + late_observation=observed, + evidence_id=envelope.evidence_id, + ) + if not turn_id or not execution_id: + return + from src.backend.services.turn_execution_record_service import ( + submit_late_effect_observation, + ) + + submit_late_effect_observation( + request_id=turn_id, + effect_id=effect_id, + execution_id=execution_id, + observation={ + **observed, + "call_id": call_id, + "capability_name": capability_name, + "effect_status": effect_status, + "changed": changed, + }, + user_id=scope.user_concept_id, + namespace=scope.namespace, + org_id=scope.organisation_concept_id, + ) + + return observe def retain_model_evidence_views(results: Sequence[ToolResult]) -> None: for result in results: - if result.tool_name not in {_INVOKE_TOOL_NAME, _EVIDENCE_TOOL_NAME}: + if result.tool_name not in { + _INVOKE_TOOL_NAME, + _EVIDENCE_TOOL_NAME, + _EVIDENCE_INDEX_TOOL_NAME, + }: continue if not isinstance(result.output, Mapping): continue evidence_id = result.output.get("evidence_id") - if not isinstance(evidence_id, str) or not evidence_id.strip(): + is_index_page = ( + result.output.get("schema_version") + == "adaptive_turn_evidence_index_page.v1" + ) + if ( + not is_index_page + and ( + not isinstance(evidence_id, str) + or not evidence_id.strip() + ) + ): continue view = dict(result.output) digest = hashlib.sha256(_json_bytes(view)).hexdigest() @@ -1674,6 +1887,89 @@ def retain_model_evidence_views(results: Sequence[ToolResult]) -> None: evidence_views.append(view) def finish(text: str, *, status: str = "completed") -> AdaptiveTurnResult: + with effect_state_lock: + effect_snapshot = { + effect_id: dict(state) + for effect_id, state in effect_states.items() + } + reconciled_invocations: list[dict[str, Any]] = [] + for raw_invocation in tool_invocations: + invocation = dict(raw_invocation) + effect_id = invocation.get("effect_id") + state = ( + effect_snapshot.get(effect_id) + if isinstance(effect_id, str) + else None + ) + if isinstance(state, Mapping): + invocation["effect_status"] = state.get("effect_status") + invocation["changed"] = state.get("changed") + if state.get("execution_id"): + invocation["execution_id"] = state.get("execution_id") + if state.get("evidence_id"): + invocation["late_evidence_id"] = state.get("evidence_id") + if isinstance(state.get("late_observation"), Mapping): + invocation["late_completion"] = dict( + state["late_observation"] + ) + reconciled_invocations.append(invocation) + + relevant_effects = [ + state + for state in effect_snapshot.values() + 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 + ) + ) + ] + effect_finality_fallback = status != "completed" and bool(relevant_effects) + if 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", + ) + } + 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" + ) + 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." + ) + 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, + } + ) evidence_index = _compact_evidence_index(evidence_store.index()) aux_calls.append( { @@ -1687,7 +1983,7 @@ def finish(text: str, *, status: str = "completed") -> AdaptiveTurnResult: return AdaptiveTurnResult( response_text=text, extra_messages=tuple(extra_messages), - tool_invocations=tuple(tool_invocations), + tool_invocations=tuple(reconciled_invocations), aux_llm_calls=tuple(aux_calls), llm_calls=tuple(llm_calls), llm_usage=( @@ -1700,20 +1996,104 @@ 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, + ) + + def synthesis_draft_text() -> str: + """Retain only a draft based on the latest observed effect state.""" + + with effect_state_lock: + if last_partial_effect_generation != effect_state_generation: + return "" + return last_partial_text + + def enter_final_synthesis() -> None: + nonlocal final_synthesis + nonlocal answer_only + nonlocal continuation + nonlocal pending_results + nonlocal current_context + nonlocal final_context_base + if final_context_base is None: + final_context_base = [ + dict(item) + for item in current_context + if isinstance(item, Mapping) + and str(item.get("role") or "").strip().lower() != "tool" + ] + final_synthesis = True + answer_only = False + continuation = None + pending_results = [] + current_context = _final_synthesis_context( + final_context_base, + evidence_store.index(), + evidence_views, + draft_text=synthesis_draft_text(), + ) + + def enter_answer_only(reason: str) -> None: + nonlocal final_synthesis + nonlocal answer_only + nonlocal continuation + nonlocal pending_results + nonlocal current_context + nonlocal final_context_base + if answer_only: + return + if final_context_base is None: + final_context_base = [ + dict(item) + for item in current_context + if isinstance(item, Mapping) + and str(item.get("role") or "").strip().lower() != "tool" + ] + final_synthesis = True + answer_only = True + continuation = None + pending_results = [] + current_context = _final_synthesis_context( + final_context_base, + evidence_store.index(), + evidence_views, + draft_text=synthesis_draft_text(), + ) + aux_calls.append( + { + "type": "adaptive_turn_final_answer_reserve_entered", + "schema_version": ( + "adaptive_turn_final_answer_reserve_entered.v1" + ), + "reason": reason, + "requested_answer_reserve_seconds": ( + requested_final_answer_reserve + ), + "effective_answer_reserve_seconds": final_answer_reserve, + "final_synthesis_reserve_seconds": final_reserve, + "evidence_capable_reserve_seconds": ( + final_reserve - final_answer_reserve + ), + "reserve_clamped": ( + final_answer_reserve + < requested_final_answer_reserve + ), + "remaining_ms": max( + 0.0, + (turn_deadline - clock()) * 1000.0, + ), + } ) while True: _check_cancellation(progress_tracker) now = clock() if not final_synthesis and now >= research_deadline: - final_synthesis = True - continuation = None - pending_results = [] - current_context = _final_synthesis_context( - current_context, - evidence_store.index(), - evidence_views, - ) + if now >= final_answer_deadline: + enter_answer_only("direct_final_entry") + else: + enter_final_synthesis() + elif final_synthesis and not answer_only and now >= final_answer_deadline: + enter_answer_only("deadline_reached") if now >= turn_deadline: terminal_status = "turn_deadline_exceeded" text = last_partial_text.strip() or ( @@ -1723,12 +2103,18 @@ def finish(text: str, *, status: str = "completed") -> AdaptiveTurnResult: return finish(text, status=terminal_status) stage = "final_synthesis" if final_synthesis else "adaptive_research" + mode = ( + "answer_only" + if answer_only + else ("evidence_capable" if final_synthesis else "research") + ) _emit( progress_tracker, { "status": "thinking", "stage": stage, "phase": stage, + "mode": mode, "phase_label": ( "Generating response" if final_synthesis else "Researching" ), @@ -1750,13 +2136,31 @@ def finish(text: str, *, status: str = "completed") -> AdaptiveTurnResult: ) seen_request_digests.add(request_digest) request_started = clock() - stage_deadline = turn_deadline if final_synthesis else research_deadline + stage_deadline = ( + turn_deadline + if answer_only + else ( + final_answer_deadline + if final_synthesis + else research_deadline + ) + ) effective_params = dict(model_parameters or {}) effective_params["request_timeout_seconds"] = max( 0.001, stage_deadline - request_started, ) - request_tools = final_synthesis_tools if final_synthesis else available_tools + with effect_state_lock: + request_effect_generation = effect_state_generation + request_tools = ( + [] + if answer_only + else ( + final_synthesis_tools + if final_synthesis + else available_tools + ) + ) try: response: LLMResponse = llm_client.generate_with_tools( "" if continuation is not None else prompt, @@ -1767,6 +2171,7 @@ def finish(text: str, *, status: str = "completed") -> AdaptiveTurnResult: scope, delegated_count=len(delegated_names), final_synthesis=final_synthesis, + answer_only=answer_only, ), llm_params=effective_params, **( @@ -1779,6 +2184,21 @@ def finish(text: str, *, status: str = "completed") -> AdaptiveTurnResult: ), ) except StructuredToolContextLimitError as exc: + if final_synthesis and not answer_only: + aux_calls.append( + { + "type": "adaptive_turn_context_limit_recovery", + "before_digest": request_digest, + "after_digest": None, + "before_bytes": request_size, + "after_bytes": None, + "changed": True, + "reason": "answer_without_tools", + "error": str(exc), + } + ) + enter_answer_only("evidence_call_failed") + continue fresh_context = _compact_context_after_limit( prompt=prompt, context=current_context, @@ -1847,7 +2267,11 @@ def finish(text: str, *, status: str = "completed") -> AdaptiveTurnResult: { "type": "adaptive_turn_model_call", "stage": stage, + "mode": mode, "model": model, + "request_timeout_seconds": effective_params[ + "request_timeout_seconds" + ], "duration_ms": max( 0.0, (call_failed_at - request_started) * 1000.0, @@ -1858,15 +2282,14 @@ def finish(text: str, *, status: str = "completed") -> AdaptiveTurnResult: "error_class": type(exc).__name__, } ) + if final_synthesis and not answer_only: + enter_answer_only("evidence_call_failed") + continue if not final_synthesis and call_failed_at >= research_deadline: - final_synthesis = True - continuation = None - pending_results = [] - current_context = _final_synthesis_context( - current_context, - evidence_store.index(), - evidence_views, - ) + if call_failed_at >= final_answer_deadline: + enter_answer_only("direct_final_entry") + else: + enter_final_synthesis() aux_calls.append( { "type": "adaptive_turn_research_deadline_recovery", @@ -1893,7 +2316,11 @@ def finish(text: str, *, status: str = "completed") -> AdaptiveTurnResult: { "type": "adaptive_turn_model_call", "stage": stage, + "mode": mode, "model": response.model or model, + "request_timeout_seconds": effective_params[ + "request_timeout_seconds" + ], "duration_ms": max( 0.0, (response_received_at - request_started) * 1000.0, @@ -1910,24 +2337,28 @@ def finish(text: str, *, status: str = "completed") -> AdaptiveTurnResult: "type": "adaptive_turn_late_model_result", "schema_version": "adaptive_turn_late_model_result.v1", "stage": stage, + "mode": mode, "late_result_policy": "discard_from_terminal_result", } ) - if final_synthesis or response_received_at >= turn_deadline: + if ( + final_synthesis + and not answer_only + and response_received_at < turn_deadline + ): + enter_answer_only("late_evidence_result") + continue + if answer_only or response_received_at >= turn_deadline: terminal_status = "turn_deadline_exceeded" text = last_partial_text.strip() or ( "I could not produce a useful response before this turn's " "elapsed-time deadline." ) return finish(text, status=terminal_status) - final_synthesis = True - continuation = None - pending_results = [] - current_context = _final_synthesis_context( - current_context, - evidence_store.index(), - evidence_views, - ) + if response_received_at >= final_answer_deadline: + enter_answer_only("direct_final_entry") + else: + enter_final_synthesis() continue call_duration_ms = max( @@ -1939,7 +2370,11 @@ def finish(text: str, *, status: str = "completed") -> AdaptiveTurnResult: { "type": "adaptive_turn_model_call", "stage": stage, + "mode": mode, "model": response.model or model, + "request_timeout_seconds": effective_params[ + "request_timeout_seconds" + ], "duration_ms": call_duration_ms, "usage": dict(response.usage) if response.usage else None, "status": "completed", @@ -1951,16 +2386,13 @@ def finish(text: str, *, status: str = "completed") -> AdaptiveTurnResult: ) if response.text_response.strip(): last_partial_text = response.text_response.strip() + last_partial_effect_generation = request_effect_generation calls = list(response.tool_calls) if not final_synthesis and clock() >= research_deadline and calls: - final_synthesis = True - continuation = None - pending_results = [] - current_context = _final_synthesis_context( - current_context, - evidence_store.index(), - evidence_views, - ) + if clock() >= final_answer_deadline: + enter_answer_only("direct_final_entry") + else: + enter_final_synthesis() aux_calls.append( { "type": "adaptive_turn_research_deadline_recovery", @@ -1970,24 +2402,55 @@ def finish(text: str, *, status: str = "completed") -> AdaptiveTurnResult: } ) continue + if ( + final_synthesis + and not answer_only + and clock() >= final_answer_deadline + and calls + ): + aux_calls.append( + { + "type": "adaptive_turn_expired_final_evidence_calls", + "schema_version": ( + "adaptive_turn_expired_final_evidence_calls.v1" + ), + "discarded_tool_call_count": len(calls), + } + ) + enter_answer_only("deadline_reached") + continue if not calls: if response.text_response.strip(): return finish(response.text_response.strip()) + if final_synthesis and not answer_only: + enter_answer_only("evidence_call_failed") + continue terminal_status = "model_non_answer" text = last_partial_text or ( "The model returned neither an answer nor a capability request." ) return finish(text, status=terminal_status) + if answer_only and calls: + terminal_status = "final_synthesis_protocol_error" + text = last_partial_text or ( + "The model requested a tool during the answer-only interval." + ) + return finish(text, status=terminal_status) if final_synthesis and any( call.tool_name not in {_EVIDENCE_INDEX_TOOL_NAME, _EVIDENCE_TOOL_NAME} for call in calls ): - terminal_status = "final_synthesis_protocol_error" - text = last_partial_text or ( - "The model requested a new external capability after the research " - "deadline." + aux_calls.append( + { + "type": "adaptive_turn_final_synthesis_protocol_recovery", + "schema_version": ( + "adaptive_turn_final_synthesis_protocol_recovery.v1" + ), + "discarded_tool_call_count": len(calls), + } ) - return finish(text, status=terminal_status) + enter_answer_only("evidence_call_failed") + continue continuation = response.continuation batch_results: list[ToolResult | None] = [None] * len(calls) @@ -2206,7 +2669,7 @@ def invoke_and_contain( int, tuple[Any, Any, dict[str, Any], str, bool], ]: - index, _call, canonical_name, arguments, is_effect = item + index, call, canonical_name, arguments, is_effect = item assert gateway is not None definition = gateway.get_method_definition(canonical_name) subject_argument = ( @@ -2244,6 +2707,15 @@ def invoke_and_contain( is_effect, ) try: + effect_identifier = ( + _effect_id( + turn_id=turn_id, + call_id=call.call_id, + capability_name=canonical_name, + ) + if is_effect + else None + ) with override_current_actor( scope.user_concept_id, scope.organisation_concept_id, @@ -2252,13 +2724,34 @@ def invoke_and_contain( canonical_name, arguments, deadline_monotonic=research_deadline, + late_completion_observer=( + late_effect_observer( + effect_id=effect_identifier, + call_id=call.call_id, + capability_name=canonical_name, + ) + if effect_identifier is not None + else None + ), ) raw_payload = transport_result.payload except SchemaValidationError as exc: + output_invalid = exc.stage == "output_schema" raw_payload = _error_payload( - "capability_arguments_invalid", + ( + "effect_output_invalid" + if output_invalid and is_effect + else ( + "capability_output_invalid" + if output_invalid + else "capability_arguments_invalid" + ) + ), str(exc), + retryable=output_invalid and not is_effect, ) + if output_invalid and is_effect: + raw_payload["mutation_outcome"] = "unknown" transport_result = None except Exception as exc: # noqa: BLE001 raw_payload = _error_payload( @@ -2395,6 +2888,23 @@ def invoke_and_contain( and isinstance(raw_payload.get("changed"), bool) else (False if effect_status == "failed" else None) ) + remember_effect_state( + effect_identifier, + phase=0, + effect_status=effect_status, + changed=changed, + execution_id=( + str( + getattr( + transport_result, + "execution_id", + "", + ) + or "" + ).strip() + or None + ), + ) envelope_payload.update( { "effect_id": effect_identifier, @@ -2477,11 +2987,15 @@ def invoke_and_contain( "action": ( "fresh_final_synthesis_with_pageable_evidence_index" if not final_synthesis - else "terminal_bounded_non_success" + else ( + "answer_from_bounded_evidence" + if not answer_only + else "terminal_bounded_non_success" + ) ), } ) - if final_synthesis: + if answer_only: return finish( last_partial_text or ( @@ -2491,14 +3005,10 @@ def invoke_and_contain( ), status="tool_result_batch_context_limit", ) - final_synthesis = True - continuation = None - pending_results = [] - current_context = _final_synthesis_context( - current_context, - evidence_store.index(), - evidence_views, - ) + if final_synthesis: + enter_answer_only("evidence_call_failed") + else: + enter_final_synthesis() continue pending_results = bounded_results retain_model_evidence_views(pending_results) diff --git a/src/backend/services/turn_execution_record_service.py b/src/backend/services/turn_execution_record_service.py index 55cd3bd2e..56ad9ec98 100644 --- a/src/backend/services/turn_execution_record_service.py +++ b/src/backend/services/turn_execution_record_service.py @@ -13,6 +13,7 @@ import re import threading import time +from concurrent.futures import ThreadPoolExecutor from datetime import datetime, timezone from typing import Any, Mapping, Sequence, cast @@ -80,6 +81,7 @@ TURN_EXECUTION_RECORD_SCHEMA_VERSION = "turn_execution_record.v1" TURN_EXECUTION_CORRECTNESS_SCHEMA_VERSION = "turn_execution_correctness.v1" +LATE_EFFECT_OBSERVATION_SCHEMA_VERSION = "late_effect_observation.v1" WORKFLOW_ROUTING_DIAGNOSTICS_SCHEMA_VERSION = "workflow_routing_diagnostics.v1" FINAL_ANSWER_SYNTHESIS_TELEMETRY_SCHEMA_VERSION = "final_answer_synthesis_telemetry.v1" TOOL_EVIDENCE_PROJECTION_REACHABILITY_SCHEMA_VERSION = ( @@ -120,6 +122,16 @@ _FINAL_ANSWER_PUBLIC_PROJECTION_MAX_ENTRIES = 16 _FINAL_ANSWER_PUBLIC_PROJECTION_MAX_FIELD_ENTRIES = 32 _FINAL_ANSWER_PUBLIC_PROJECTION_MAX_IDENTIFIERS = 32 +_LATE_EFFECT_OBSERVATION_MAX_ID_CHARS = 512 +_LATE_EFFECT_APPEND_WORKERS = 2 +_LATE_EFFECT_APPEND_MAX_PENDING = 32 +_LATE_EFFECT_APPEND_EXECUTOR = ThreadPoolExecutor( + max_workers=_LATE_EFFECT_APPEND_WORKERS, + thread_name_prefix="late-effect-observation", +) +_LATE_EFFECT_APPEND_SLOTS = threading.BoundedSemaphore( + _LATE_EFFECT_APPEND_WORKERS + _LATE_EFFECT_APPEND_MAX_PENDING +) _FINAL_ANSWER_PROJECTION_SECRET_KEY_PARTS = ( "access_token", "api_key", @@ -5199,10 +5211,18 @@ def _summarise_tool_invocations( timing_value = _safe_float(invocation.get(timing_key)) if timing_value is not None: serialised_invocation[timing_key] = timing_value - for identifier_key in ("call_id", "execution_id", "timeout_phase"): + for identifier_key in ( + "call_id", + "execution_id", + "timeout_phase", + "effect_id", + "effect_status", + ): identifier_value = _safe_str(invocation.get(identifier_key)) if identifier_value: serialised_invocation[identifier_key] = identifier_value + if isinstance(invocation.get("changed"), bool): + serialised_invocation["changed"] = invocation.get("changed") if isinstance(invocation.get("advisory_budget_exceeded"), bool): serialised_invocation["advisory_budget_exceeded"] = invocation.get( "advisory_budget_exceeded" @@ -10451,6 +10471,10 @@ def upsert_turn_execution_record_projection( now = _now_utc() payload = ensure_turn_execution_record_execution_correctness(record) + # Late observations arrive independently after the ordinary projection was + # assembled. They are append-only and must not be replaced by a later + # whole-turn snapshot containing no, or stale, late-observation state. + payload.pop("late_effect_observations", None) payload["request_id"] = request_id if _safe_str(user_id): payload.setdefault("user_id", _safe_str(user_id)) @@ -10509,6 +10533,187 @@ def upsert_turn_execution_record_projection( } +def append_late_effect_observation( + *, + request_id: Any, + effect_id: Any, + execution_id: Any, + observation: Mapping[str, Any], + user_id: Any = None, + session_id: Any = None, + namespace: Any = None, + org_id: Any = None, +) -> dict[str, Any]: + """Append one bounded, identity-bearing late-effect observation. + + The entry is additive to the ordinary turn projection: a later full-record + ``$set`` upsert does not remove it. ``$addToSet`` also makes an exact retry + of the same observation harmless. The caller remains responsible for + interpreting a handler receipt versus canonical state. + + Entries are individually bounded, but the array has no arbitrary count + ceiling. Replacing this atomic idempotent append with ``$push/$slice`` + would admit duplicates, while a read-trim-write sequence would lose + concurrent observations. A per-turn retention limit should therefore use + an atomic pipeline or a separately indexed collection if observed growth + demonstrates the need. + """ + + clean_request_id = _safe_str(request_id) + clean_effect_id = _safe_str(effect_id) + clean_execution_id = _safe_str(execution_id) + identities = { + "request_id": clean_request_id, + "effect_id": clean_effect_id, + "execution_id": clean_execution_id, + } + for field_name, value in identities.items(): + if not value: + return {"updated": False, "reason": f"missing_{field_name}"} + if len(value) > _LATE_EFFECT_OBSERVATION_MAX_ID_CHARS: + return {"updated": False, "reason": f"invalid_{field_name}"} + if not isinstance(observation, Mapping): + return {"updated": False, "reason": "invalid_observation"} + + coll = get_turn_execution_records_collection() + if coll is None: + return { + "updated": False, + "reason": "collection_unavailable", + "request_id": clean_request_id, + } + + bounded_observation = _compact_final_answer_projection_payload( + dict(observation), + max_depth=4, + max_items=32, + max_string_chars=2_000, + ) + if not isinstance(bounded_observation, Mapping): + bounded_observation = {"value": bounded_observation} + bounded_observation = dict(bounded_observation) + + observed_at_utc = ( + _safe_str(bounded_observation.get("observed_at_utc")) + or _iso_utc(_now_utc()) + ) + observation_identity = "\0".join( + (clean_request_id, clean_effect_id, clean_execution_id) + ) + observation_id = hashlib.sha256( + observation_identity.encode("utf-8") + ).hexdigest() + entry = dict(bounded_observation) + entry.update( + { + "schema_version": LATE_EFFECT_OBSERVATION_SCHEMA_VERSION, + "observation_id": observation_id, + "effect_id": clean_effect_id, + "execution_id": clean_execution_id, + "observed_at_utc": observed_at_utc, + } + ) + + now = _now_utc() + insert_payload: dict[str, Any] = { + "request_id": clean_request_id, + "schema_version": TURN_EXECUTION_RECORD_SCHEMA_VERSION, + "created_at_utc": _iso_utc(now), + "updated_at_utc": _iso_utc(now), + "inserted_at": now, + } + for field_name, raw_value in ( + ("user_id", user_id), + ("session_id", session_id), + ("namespace", namespace), + ("org_id", org_id), + ): + clean_value = _safe_str(raw_value) + if clean_value: + insert_payload[field_name] = clean_value + + try: + result = _turn_execution_update_one( + coll, + {"request_id": clean_request_id}, + { + "$addToSet": {"late_effect_observations": entry}, + "$setOnInsert": insert_payload, + }, + operation="append_late_effect_observation.update_one", + detail=clean_execution_id, + upsert=True, + ) + appended = bool(getattr(result, "modified_count", 0) > 0) + inserted = getattr(result, "upserted_id", None) is not None + matched = bool(getattr(result, "matched_count", 0) > 0) + return { + "updated": appended or inserted, + "appended": appended or inserted, + "duplicate": matched and not appended, + "inserted": inserted, + "matched": matched, + "request_id": clean_request_id, + "effect_id": clean_effect_id, + "execution_id": clean_execution_id, + "observation_id": observation_id, + } + except PyMongoError as exc: + logger.warning( + "Failed to append late effect observation for request_id=%s " + "effect_id=%s execution_id=%s: %s", + clean_request_id, + clean_effect_id, + clean_execution_id, + exc, + ) + return { + "updated": False, + "reason": "mongo_error", + "request_id": clean_request_id, + "effect_id": clean_effect_id, + "execution_id": clean_execution_id, + } + + +def submit_late_effect_observation(**kwargs: Any) -> bool: + """Queue a durable append without occupying an MCP handler worker.""" + + if not _LATE_EFFECT_APPEND_SLOTS.acquire(blocking=False): + logger.warning( + "Late-effect observation queue is full; request_id=%s effect_id=%s", + _safe_str(kwargs.get("request_id")), + _safe_str(kwargs.get("effect_id")), + ) + return False + + def _append() -> None: + try: + append_late_effect_observation(**kwargs) + except Exception: + logger.exception( + "Late-effect observation append failed; request_id=%s " + "effect_id=%s", + _safe_str(kwargs.get("request_id")), + _safe_str(kwargs.get("effect_id")), + ) + finally: + _LATE_EFFECT_APPEND_SLOTS.release() + + try: + _LATE_EFFECT_APPEND_EXECUTOR.submit(_append) + except RuntimeError: + _LATE_EFFECT_APPEND_SLOTS.release() + logger.warning( + "Late-effect observation executor is unavailable; request_id=%s " + "effect_id=%s", + _safe_str(kwargs.get("request_id")), + _safe_str(kwargs.get("effect_id")), + ) + return False + return True + + def get_latest_turn_execution_record_projection( *, session_id: str | None, diff --git a/tests/backend/test_adaptive_turn_service.py b/tests/backend/test_adaptive_turn_service.py index 131b81d19..f2d62663a 100644 --- a/tests/backend/test_adaptive_turn_service.py +++ b/tests/backend/test_adaptive_turn_service.py @@ -2,6 +2,7 @@ import json import time +from threading import Event from typing import Any import pytest @@ -12,7 +13,10 @@ MethodDefinition, ) from src.backend.integrations.internal_mcp.schemas import Schema -from src.backend.integrations.internal_mcp.transport import InternalMCPTransport +from src.backend.integrations.internal_mcp.transport import ( + InternalMCPTransport, + internal_mcp_cancellation_requested, +) from src.backend.languagemodels.structured_tool_calling.types import ( LLMContinuation, LLMResponse, @@ -203,6 +207,7 @@ def _effect_gateway( handler: Any, *, write_timeout_sec: float = 1.0, + effect_output_schema: Schema | None = None, ) -> InternalMCPGateway: catalogue = MethodCatalogue() catalogue.register( @@ -223,6 +228,7 @@ def _effect_gateway( name=name, handler=lambda _name=name, **kwargs: handler(_name, kwargs), input_schema=Schema(allow_unknown=True), + output_schema=effect_output_schema, category="write", ordinary_turn_effect=True, ordinary_turn_mutation_subject_argument=subject_argument, @@ -318,6 +324,26 @@ def generate_with_tools( return super().generate_with_tools(prompt, available_tools, **kwargs) +class _TimedSequenceClient(_SequenceClient): + def __init__( + self, + clock: _ManualClock, + *timed_responses: tuple[float, Any], + ) -> None: + super().__init__(*(item[1] for item in timed_responses)) + self.clock = clock + self.response_times = [item[0] for item in timed_responses] + + def generate_with_tools( + self, + prompt: str, + available_tools: list[Any], + **kwargs: Any, + ) -> LLMResponse: + self.clock.now = self.response_times[len(self.calls)] + return super().generate_with_tools(prompt, available_tools, **kwargs) + + class _HydrationDeadlineClient: def __init__( self, @@ -738,6 +764,332 @@ def handler(name: str, _arguments: dict[str, Any]) -> dict[str, Any]: assert result.tool_invocations[0]["changed"] is True +def test_terminal_model_failure_cannot_claim_no_change_after_effect() -> None: + client = _SequenceClient( + LLMResponse( + text_response="I have not changed anything.", + tool_calls=[ + ToolCall( + tool_name="turn_invoke_capability", + call_id="effect-before-model-failure", + payload={ + "name": "create_concepts", + "arguments": { + "concepts": [{"name": "A real represented concept"}] + }, + }, + ) + ], + ), + TimeoutError("final model call failed"), + ) + + result = execute_adaptive_turn( + gateway=_effect_gateway( + lambda _name, _arguments: { + "success": True, + "effect_status": "succeeded", + "changed": True, + } + ), + prompt="Represent this concept.", + context=[], + llm_client=client, + model="test-model", + user_namespace="#V#person@org", + user_concept_id="#V#person", + org_concept_id="#V#org", + turn_id="effect-before-model-failure", + turn_budget_seconds=10, + final_synthesis_reserve_seconds=2, + ) + + 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 "I have not changed anything" not in result.response_text + assert result.tool_invocations[0]["effect_status"] == "succeeded" + assert result.tool_invocations[0]["changed"] is True + + +def test_post_handler_output_validation_failure_is_indeterminate() -> None: + committed: list[str] = [] + + def handler(_name: str, _arguments: dict[str, Any]) -> dict[str, Any]: + committed.append("#V#committed_before_invalid_receipt") + return {"success": True} + + client = _SequenceClient( + LLMResponse( + text_response="I have not changed anything.", + tool_calls=[ + ToolCall( + tool_name="turn_invoke_capability", + call_id="invalid-receipt-after-commit", + payload={ + "name": "create_concepts", + "arguments": { + "concepts": [{"name": "Committed before invalid receipt"}] + }, + }, + ) + ], + ), + TimeoutError("final model call failed"), + ) + + result = execute_adaptive_turn( + gateway=_effect_gateway( + handler, + effect_output_schema=Schema( + required={"success": bool, "changed": bool}, + allow_unknown=True, + ), + ), + prompt="Represent this concept.", + context=[], + llm_client=client, + model="test-model", + user_namespace="#V#person@org", + user_concept_id="#V#person", + org_concept_id="#V#org", + turn_id="invalid-receipt-after-commit", + turn_budget_seconds=10, + final_synthesis_reserve_seconds=2, + ) + + 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.tool_invocations[0]["effect_status"] == "indeterminate" + assert result.tool_invocations[0]["changed"] is None + + +@pytest.mark.parametrize( + ("answer_reserve", "effect_finished_at", "expected_tool_names"), + [ + ( + 0.0, + 8.1, + {"turn_list_evidence", "turn_read_evidence"}, + ), + (2.0, 8.5, set()), + ], + ids=["evidence-capable-final", "answer-only-final"], +) +def test_effect_removes_false_draft_from_fresh_final_context( + answer_reserve: float, + effect_finished_at: float, + expected_tool_names: set[str], +) -> None: + started = time.monotonic() + clock = _ManualClock(started) + + def handler(_name: str, _arguments: dict[str, Any]) -> dict[str, Any]: + clock.now = started + effect_finished_at + return { + "success": True, + "effect_status": "succeeded", + "changed": True, + } + + false_draft = "I have not changed anything." + client = _SequenceClient( + LLMResponse( + text_response=false_draft, + tool_calls=[ + ToolCall( + tool_name="turn_invoke_capability", + call_id=f"effect-before-{answer_reserve}", + payload={ + "name": "create_concepts", + "arguments": { + "concepts": [{"name": "A represented concept"}] + }, + }, + ) + ], + ), + LLMResponse(text_response="The represented concept was created."), + ) + + result = execute_adaptive_turn( + gateway=_effect_gateway(handler), + prompt="Represent this concept.", + context=[], + llm_client=client, + model="test-model", + user_namespace="#V#person@org", + user_concept_id="#V#person", + org_concept_id="#V#org", + turn_id=f"effect-draft-{answer_reserve}", + turn_budget_seconds=10, + final_synthesis_reserve_seconds=2, + final_answer_reserve_seconds=answer_reserve, + clock=clock, + ) + + assert result.response_text == "The represented concept was created." + assert len(client.calls) == 2 + final_call = client.calls[1] + assert { + tool.name for tool in final_call["available_tools"] + } == expected_tool_names + assert not any( + item.get("role") == "assistant" and item.get("content") == false_draft + for item in final_call["context"] + ) + assert false_draft not in json.dumps(final_call["context"]) + + +def test_post_effect_draft_remains_available_to_answer_only_recovery() -> None: + started = time.monotonic() + clock = _ManualClock(started) + useful_draft = "USEFUL POST-EFFECT DRAFT" + + def handler(_name: str, _arguments: dict[str, Any]) -> dict[str, Any]: + clock.now = started + 8.1 + return { + "success": True, + "effect_status": "succeeded", + "changed": True, + } + + client = _SequenceClient( + LLMResponse( + text_response="I have not changed anything.", + tool_calls=[ + ToolCall( + tool_name="turn_invoke_capability", + call_id="effect-before-useful-draft", + payload={ + "name": "create_concepts", + "arguments": { + "concepts": [{"name": "A represented concept"}] + }, + }, + ) + ], + ), + LLMResponse( + text_response=useful_draft, + tool_calls=[ + ToolCall( + tool_name="turn_list_evidence", + call_id="list-after-effect", + payload={}, + ) + ], + ), + TimeoutError("evidence-capable synthesis failed"), + LLMResponse(text_response="Recovered from the useful draft."), + ) + + result = execute_adaptive_turn( + gateway=_effect_gateway(handler), + prompt="Represent this concept.", + context=[], + llm_client=client, + model="test-model", + user_namespace="#V#person@org", + user_concept_id="#V#person", + org_concept_id="#V#org", + turn_id="post-effect-draft", + turn_budget_seconds=10, + final_synthesis_reserve_seconds=2, + final_answer_reserve_seconds=1, + clock=clock, + ) + + assert result.response_text == "Recovered from the useful draft." + answer_only_call = client.calls[-1] + assert answer_only_call["available_tools"] == [] + assert any( + item.get("role") == "assistant" and item.get("content") == useful_draft + for item in answer_only_call["context"] + ) + + +def test_late_effect_completion_is_persisted_without_rewriting_terminal_result( + monkeypatch: pytest.MonkeyPatch, +) -> None: + from src.backend.services import turn_execution_record_service + + release_handler = Event() + observation_persisted = Event() + persisted: list[dict[str, Any]] = [] + + def handler(_name: str, _arguments: dict[str, Any]) -> dict[str, Any]: + while not internal_mcp_cancellation_requested(): + time.sleep(0.001) + assert release_handler.wait(timeout=1.0) + return { + "success": True, + "effect_status": "succeeded", + "changed": True, + "created_ids": ["#V#late_real_concept"], + } + + def append_observation(**kwargs: Any) -> dict[str, Any]: + persisted.append(dict(kwargs)) + observation_persisted.set() + return {"updated": True} + + monkeypatch.setattr( + turn_execution_record_service, + "append_late_effect_observation", + append_observation, + ) + client = _SequenceClient( + LLMResponse( + text_response="", + tool_calls=[ + ToolCall( + tool_name="turn_invoke_capability", + call_id="late-effect", + payload={ + "name": "create_concepts", + "arguments": { + "concepts": [{"name": "A late real concept"}] + }, + }, + ) + ], + ), + TimeoutError("model failed after the effect timeout"), + ) + + result = execute_adaptive_turn( + gateway=_effect_gateway(handler, write_timeout_sec=0.02), + prompt="Represent this concept.", + context=[], + llm_client=client, + model="test-model", + user_namespace="#V#person@org", + user_concept_id="#V#person", + org_concept_id="#V#org", + turn_id="late-effect-request", + turn_budget_seconds=10, + final_synthesis_reserve_seconds=2, + ) + + assert result.effect_finality_fallback is True + assert result.tool_invocations[0]["effect_status"] == "indeterminate" + assert "1 indeterminate" in result.response_text + + release_handler.set() + assert observation_persisted.wait(timeout=1.0) + assert len(persisted) == 1 + durable = persisted[0] + assert durable["request_id"] == "late-effect-request" + assert durable["effect_id"] == result.tool_invocations[0]["effect_id"] + assert durable["observation"]["effect_status"] == "succeeded" + assert durable["observation"]["changed"] is True + # The already returned turn remains an honest point-in-time snapshot. + assert result.tool_invocations[0]["effect_status"] == "indeterminate" + + @pytest.mark.parametrize( "relationships", [ @@ -2196,6 +2548,7 @@ def test_research_deadline_failure_preserves_final_synthesis_reserve() -> None: assert { tool.name for tool in client.calls[1]["available_tools"] } == {"turn_list_evidence", "turn_read_evidence"} + assert result.llm_calls[1]["mode"] == "evidence_capable" recovery = next( item for item in result.aux_llm_calls @@ -2204,6 +2557,333 @@ def test_research_deadline_failure_preserves_final_synthesis_reserve() -> None: assert recovery["action"] == "fresh_final_synthesis_from_available_evidence" +def test_final_answer_checkpoint_preempts_repeated_evidence_cycles() -> None: + clock = _ManualClock() + client = _TimedSequenceClient( + clock, + (6.0, TimeoutError("research deadline")), + ( + 6.4, + LLMResponse( + text_response="A useful draft based on the evidence page.", + tool_calls=[ + ToolCall( + tool_name="turn_list_evidence", + call_id="list-before-checkpoint", + payload={"offset": 0, "limit": 20}, + ) + ] + ), + ), + ( + 8.0, + LLMResponse( + text_response="", + tool_calls=[ + ToolCall( + tool_name="turn_list_evidence", + call_id="list-at-checkpoint", + payload={"offset": 0, "limit": 20}, + ) + ] + ), + ), + (8.2, LLMResponse(text_response="The protected final answer.")), + ) + + result = execute_adaptive_turn( + gateway=None, + prompt="Research, then answer.", + context=[], + llm_client=client, + model="test-model", + turn_id="turn-answer-checkpoint", + turn_budget_seconds=10, + final_synthesis_reserve_seconds=4, + final_answer_reserve_seconds=2, + clock=clock, + ) + + assert result.response_text == "The protected final answer." + assert [call["mode"] for call in result.llm_calls] == [ + "research", + "evidence_capable", + "evidence_capable", + "answer_only", + ] + assert [tool.name for tool in client.calls[1]["available_tools"]] == [ + "turn_read_evidence", + "turn_list_evidence", + ] + assert client.calls[-1]["available_tools"] == [] + assert client.calls[-1]["llm_params"]["request_timeout_seconds"] == 2.0 + final_context = client.calls[-1]["context"] + assert final_context[0] == { + "role": "assistant", + "content": "A useful draft based on the evidence page.", + } + evidence_contexts = [ + json.loads(item["content"]) + for item in final_context + if item.get("role") == "user" + and isinstance(item.get("content"), str) + and item["content"].startswith("{") + ] + assert len(evidence_contexts) == 1 + assert evidence_contexts[0]["evidence_views"][0]["schema_version"] == ( + "adaptive_turn_evidence_index_page.v1" + ) + assert not any( + item.get("call_id") == "list-at-checkpoint" + for item in result.tool_invocations + ) + checkpoint = next( + item + for item in result.aux_llm_calls + if item.get("type") == "adaptive_turn_final_answer_reserve_entered" + ) + assert checkpoint["reason"] == "late_evidence_result" + assert checkpoint["effective_answer_reserve_seconds"] == 2.0 + assert checkpoint["reserve_clamped"] is False + + +def test_final_evidence_failure_recovers_to_answer_only() -> None: + clock = _ManualClock() + client = _TimedSequenceClient( + clock, + (6.0, TimeoutError("research deadline")), + (6.5, TimeoutError("evidence-capable final call failed")), + (6.6, LLMResponse(text_response="Answered from retained evidence.")), + ) + + result = execute_adaptive_turn( + gateway=None, + prompt="Answer despite a final evidence-call failure.", + context=[], + llm_client=client, + model="test-model", + turn_id="turn-final-evidence-failure", + turn_budget_seconds=10, + final_synthesis_reserve_seconds=4, + final_answer_reserve_seconds=2, + clock=clock, + ) + + assert result.response_text == "Answered from retained evidence." + assert client.calls[1]["available_tools"] + assert client.calls[2]["available_tools"] == [] + assert result.llm_calls[1]["status"] == "failed" + checkpoint = next( + item + for item in result.aux_llm_calls + if item.get("type") == "adaptive_turn_final_answer_reserve_entered" + ) + assert checkpoint["reason"] == "evidence_call_failed" + assert checkpoint["remaining_ms"] == 3_500.0 + + +def test_final_evidence_overrun_records_actual_remaining_answer_time() -> None: + clock = _ManualClock() + client = _TimedSequenceClient( + clock, + (6.0, TimeoutError("research deadline")), + ( + 9.5, + LLMResponse( + text_response="", + tool_calls=[ + ToolCall( + tool_name="turn_list_evidence", + call_id="late-list", + payload={"offset": 0, "limit": 20}, + ) + ], + ), + ), + (9.6, LLMResponse(text_response="Answered in the actual time left.")), + ) + + result = execute_adaptive_turn( + gateway=None, + prompt="Finish from the available evidence.", + context=[], + llm_client=client, + model="test-model", + turn_id="turn-final-evidence-overrun", + turn_budget_seconds=10, + final_synthesis_reserve_seconds=4, + final_answer_reserve_seconds=2, + clock=clock, + ) + + assert result.response_text == "Answered in the actual time left." + assert client.calls[-1]["llm_params"]["request_timeout_seconds"] == 0.5 + checkpoint = next( + item + for item in result.aux_llm_calls + if item.get("type") == "adaptive_turn_final_answer_reserve_entered" + ) + assert checkpoint["reason"] == "late_evidence_result" + assert checkpoint["remaining_ms"] == 500.0 + assert result.llm_calls[1]["status"] == "late_result_discarded" + + +def test_small_final_reserve_is_a_recorded_answer_only_checkpoint() -> None: + clock = _ManualClock() + client = _TimedSequenceClient( + clock, + (8.0, TimeoutError("research model deadline")), + (8.5, LLMResponse(text_response="A bounded answer.")), + ) + + result = execute_adaptive_turn( + gateway=None, + prompt="Answer within the caller's small reserve.", + context=[], + llm_client=client, + model="test-model", + turn_id="turn-small-final-reserve", + turn_budget_seconds=10, + final_synthesis_reserve_seconds=2, + final_answer_reserve_seconds=20, + clock=clock, + ) + + assert result.response_text == "A bounded answer." + assert len(client.calls) == 2 + assert client.calls[1]["available_tools"] == [] + assert client.calls[1]["llm_params"]["request_timeout_seconds"] == 2.0 + checkpoint = next( + item + for item in result.aux_llm_calls + if item.get("type") == "adaptive_turn_final_answer_reserve_entered" + ) + assert checkpoint["reason"] == "direct_final_entry" + assert checkpoint["requested_answer_reserve_seconds"] == 20.0 + assert checkpoint["effective_answer_reserve_seconds"] == 2.0 + assert checkpoint["evidence_capable_reserve_seconds"] == 0.0 + assert checkpoint["reserve_clamped"] is True + + +def test_answer_checkpoint_can_be_enabled_from_candidate_environment( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv( + "VON_ADAPTIVE_TURN_FINAL_ANSWER_RESERVE_SEC", + "1.5", + ) + clock = _ManualClock() + client = _TimedSequenceClient( + clock, + (8.0, TimeoutError("research model deadline")), + (8.5, LLMResponse(text_response="A late evidence-capable draft.")), + (8.6, LLMResponse(text_response="The candidate answer.")), + ) + + result = execute_adaptive_turn( + gateway=None, + prompt="Exercise the candidate allocation seam.", + context=[], + llm_client=client, + model="test-model", + turn_id="turn-env-answer-checkpoint", + turn_budget_seconds=10, + final_synthesis_reserve_seconds=2, + clock=clock, + ) + + assert result.response_text == "The candidate answer." + assert client.calls[1]["llm_params"]["request_timeout_seconds"] == 0.5 + assert { + tool.name for tool in client.calls[1]["available_tools"] + } == {"turn_list_evidence", "turn_read_evidence"} + assert client.calls[2]["available_tools"] == [] + checkpoint = next( + item + for item in result.aux_llm_calls + if item.get("type") == "adaptive_turn_final_answer_reserve_entered" + ) + assert checkpoint["effective_answer_reserve_seconds"] == 1.5 + assert checkpoint["reserve_clamped"] is False + + +def test_final_evidence_tools_remain_available_before_answer_checkpoint() -> None: + clock = _ManualClock() + client = _TimedSequenceClient( + clock, + (6.0, TimeoutError("research deadline")), + ( + 6.5, + LLMResponse( + text_response="", + tool_calls=[ + ToolCall( + tool_name="turn_list_evidence", + call_id="list-final-evidence", + payload={"offset": 0, "limit": 20}, + ) + ] + ), + ), + (6.6, LLMResponse(text_response="Answered before the checkpoint.")), + ) + + result = execute_adaptive_turn( + gateway=None, + prompt="Use final evidence only if it helps.", + context=[], + llm_client=client, + model="test-model", + turn_id="turn-final-evidence-window", + turn_budget_seconds=10, + final_synthesis_reserve_seconds=4, + final_answer_reserve_seconds=2, + clock=clock, + ) + + assert result.response_text == "Answered before the checkpoint." + assert { + tool.name for tool in client.calls[1]["available_tools"] + } == {"turn_list_evidence", "turn_read_evidence"} + assert result.tool_invocations[0]["call_id"] == "list-final-evidence" + assert not any( + item.get("type") == "adaptive_turn_final_answer_reserve_entered" + for item in result.aux_llm_calls + ) + + +def test_answer_checkpoint_can_be_disabled_for_an_adaptive_caller() -> None: + clock = _ManualClock() + client = _TimedSequenceClient( + clock, + (6.0, TimeoutError("research deadline")), + (9.0, LLMResponse(text_response="The caller retained adaptive time.")), + ) + + result = execute_adaptive_turn( + gateway=None, + prompt="Use the final interval adaptively.", + context=[], + llm_client=client, + model="test-model", + turn_id="turn-disabled-answer-checkpoint", + turn_budget_seconds=10, + final_synthesis_reserve_seconds=4, + final_answer_reserve_seconds=0, + clock=clock, + ) + + assert result.response_text == "The caller retained adaptive time." + assert { + tool.name for tool in client.calls[1]["available_tools"] + } == {"turn_list_evidence", "turn_read_evidence"} + assert client.calls[1]["llm_params"]["request_timeout_seconds"] == 4.0 + assert not any( + item.get("type") == "adaptive_turn_final_answer_reserve_entered" + for item in result.aux_llm_calls + ) + + def test_model_result_returned_after_turn_deadline_is_discarded() -> None: clock = _ManualClock() client = _LateResponseClient( diff --git a/tests/backend/test_internal_mcp_transport_deadlines.py b/tests/backend/test_internal_mcp_transport_deadlines.py index d34847f7c..d92b3d379 100644 --- a/tests/backend/test_internal_mcp_transport_deadlines.py +++ b/tests/backend/test_internal_mcp_transport_deadlines.py @@ -25,6 +25,7 @@ def _gateway_for( handler, transport: InternalMCPTransport, category: str = "read", + output_schema: Schema | None = None, ) -> InternalMCPGateway: catalogue = MethodCatalogue() catalogue.register( @@ -32,7 +33,7 @@ def _gateway_for( name=method_name, handler=handler, input_schema=Schema(required={}, optional={}, allow_unknown=False), - output_schema=None, + output_schema=output_schema, category=category, ) ) @@ -284,7 +285,7 @@ def test_expired_write_deadline_reports_that_no_mutation_started() -> None: assert result.timeout_phase == "pre_dispatch" assert result.payload["error_code"] == "tool_timeout" assert result.payload["retryable"] is True - assert "mutation_outcome" not in result.payload + assert result.payload["mutation_outcome"] == "not_started" assert handler_called.is_set() is False assert executor.diagnostics()["submitted_count"] == 0 @@ -350,8 +351,280 @@ def test_write_timeout_is_explicitly_indeterminate_and_not_retryable() -> None: ] +def test_dispatched_write_reports_one_bounded_late_completion_observation() -> None: + observations = [] + observation_seen = Event() + transport = InternalMCPTransport( + write_timeout_sec=0.03, + write_advisory_timeout_sec=0.01, + ) + + def _handler(): + while not internal_mcp_cancellation_requested(): + time.sleep(0.002) + return { + "success": True, + "effect_status": "succeeded", + "changed": True, + "created_ids": ["#V#late_created"], + "api_key": "must-not-leave-the-transport-worker", + "large_detail": "x" * 100_000, + } + + def _observe(observation) -> None: + observations.append(observation) + observation_seen.set() + + gateway = _gateway_for( + method_name="synthetic_observed_slow_write", + handler=_handler, + transport=transport, + category="write", + ) + + result = gateway.invoke( + "synthetic_observed_slow_write", + {}, + late_completion_observer=_observe, + ) + + assert result.outcome == "timed_out" + assert result.payload["mutation_outcome"] == "unknown" + assert result.payload["late_result_policy"] == "observe_out_of_band" + assert result.telemetry_metadata()["late_result_policy"] == "observe_out_of_band" + assert observation_seen.wait(timeout=1.0) + assert len(observations) == 1 + observation = observations[0] + assert observation["execution_id"] == result.execution_id + assert observation["method_name"] == "synthetic_observed_slow_write" + assert observation["category"] == "write" + assert observation["outcome"] == "late_success" + assert observation["payload"]["effect_status"] == "succeeded" + assert observation["payload"]["changed"] is True + assert observation["payload"]["_truncated"] is True + assert observation["payload_truncated"] is True + assert observation["output_schema_validation"] == "indeterminate_truncated" + assert observation["output_schema_valid"] is None + + time.sleep(0.02) + assert len(observations) == 1 + diagnostics = transport.get_diagnostics() + assert diagnostics["late_completion_count"] == 1 + assert diagnostics["late_completions"][0]["observer_notified"] is True + assert diagnostics["late_completions"][0]["payload_discarded"] is False + + +def test_queued_write_timeout_reports_not_started_and_never_observes() -> None: + blocker_started = Event() + release_blocker = Event() + write_handler_called = Event() + observations = [] + executor = _BoundedHandlerExecutor(worker_count=1, queue_capacity=1) + transport = InternalMCPTransport( + read_timeout_sec=0.5, + write_timeout_sec=0.03, + read_advisory_timeout_sec=0.1, + write_advisory_timeout_sec=0.01, + handler_executor=executor, + ) + + def _blocker(): + blocker_started.set() + assert release_blocker.wait(timeout=1.0) + return {"success": True} + + blocker_gateway = _gateway_for( + method_name="synthetic_queue_blocker", + handler=_blocker, + transport=transport, + ) + blocker_results = [] + blocker_thread = Thread( + target=lambda: blocker_results.append( + blocker_gateway.invoke("synthetic_queue_blocker", {}) + ) + ) + blocker_thread.start() + assert blocker_started.wait(timeout=1.0) + + def _queued_write(): + write_handler_called.set() + return {"success": True, "changed": True} + + write_gateway = _gateway_for( + method_name="synthetic_queued_observed_write", + handler=_queued_write, + transport=transport, + category="write", + ) + result = write_gateway.invoke( + "synthetic_queued_observed_write", + {}, + late_completion_observer=observations.append, + ) + + assert result.outcome == "timed_out" + assert result.timeout_phase == "queue" + assert result.payload["timeout_phase"] == "queue" + assert result.payload["error_code"] == "tool_timeout" + assert result.payload["retryable"] is True + assert result.payload["mutation_outcome"] == "not_started" + assert result.payload["late_result_policy"] == "discard_from_turn" + assert result.telemetry_metadata()["late_result_policy"] == "discard_from_turn" + assert write_handler_called.is_set() is False + assert observations == [] + + release_blocker.set() + blocker_thread.join(timeout=1.0) + assert blocker_thread.is_alive() is False + assert blocker_results and blocker_results[0].outcome == "completed" + + completion_deadline = time.monotonic() + 1.0 + while ( + executor.diagnostics()["completed_count"] < 2 + and time.monotonic() < completion_deadline + ): + time.sleep(0.002) + + assert executor.diagnostics()["completed_count"] == 2 + assert write_handler_called.is_set() is False + assert observations == [] + assert transport.get_diagnostics()["late_completion_count"] == 0 + + +def test_gateway_marks_invalid_late_write_output_as_indeterminate_evidence() -> None: + observations = [] + observation_seen = Event() + transport = InternalMCPTransport( + write_timeout_sec=0.03, + write_advisory_timeout_sec=0.01, + ) + + def _handler(): + while not internal_mcp_cancellation_requested(): + time.sleep(0.002) + return {"success": True, "effect_status": 7} + + def _observe(observation) -> None: + observations.append(observation) + observation_seen.set() + + gateway = _gateway_for( + method_name="synthetic_invalid_late_write", + handler=_handler, + transport=transport, + category="write", + output_schema=Schema( + required={"success": bool, "effect_status": str}, + optional={}, + allow_unknown=False, + ), + ) + result = gateway.invoke( + "synthetic_invalid_late_write", + {}, + late_completion_observer=_observe, + ) + + assert result.outcome == "timed_out" + assert result.payload["mutation_outcome"] == "unknown" + assert observation_seen.wait(timeout=1.0) + assert len(observations) == 1 + assert observations[0]["output_schema_validation"] == "invalid" + assert observations[0]["output_schema_valid"] is False + assert "effect_status" in observations[0]["output_schema_error"] + + +def test_read_late_completion_is_discarded_even_when_observer_is_supplied() -> None: + observations = [] + transport = InternalMCPTransport( + read_timeout_sec=0.03, + read_advisory_timeout_sec=0.01, + ) + + def _handler(): + while not internal_mcp_cancellation_requested(): + time.sleep(0.002) + return {"success": True, "late_payload": "discard me"} + + gateway = _gateway_for( + method_name="synthetic_observer_ignored_for_read", + handler=_handler, + transport=transport, + ) + result = gateway.invoke( + "synthetic_observer_ignored_for_read", + {}, + late_completion_observer=observations.append, + ) + + assert result.outcome == "timed_out" + assert result.payload["late_result_policy"] == "discard_from_turn" + deadline = time.monotonic() + 1.0 + diagnostics = transport.get_diagnostics() + while diagnostics["late_completion_count"] < 1 and time.monotonic() < deadline: + time.sleep(0.005) + diagnostics = transport.get_diagnostics() + assert observations == [] + assert diagnostics["late_completions"][0]["payload_discarded"] is True + assert diagnostics["late_completions"][0]["observer_notified"] is False + + +def test_late_completion_observer_failure_does_not_kill_bounded_worker() -> None: + executor = _BoundedHandlerExecutor(worker_count=1, queue_capacity=2) + transport = InternalMCPTransport( + write_timeout_sec=0.03, + write_advisory_timeout_sec=0.01, + handler_executor=executor, + ) + + def _late_handler(): + while not internal_mcp_cancellation_requested(): + time.sleep(0.002) + return {"success": True, "changed": True} + + gateway = _gateway_for( + method_name="synthetic_observer_failure_write", + handler=_late_handler, + transport=transport, + category="write", + ) + + result = gateway.invoke( + "synthetic_observer_failure_write", + {}, + late_completion_observer=lambda _observation: (_ for _ in ()).throw( + RuntimeError("observer failed") + ), + ) + assert result.outcome == "timed_out" + + deadline = time.monotonic() + 1.0 + diagnostics = transport.get_diagnostics() + while diagnostics["late_completion_count"] < 1 and time.monotonic() < deadline: + time.sleep(0.005) + diagnostics = transport.get_diagnostics() + assert diagnostics["late_completions"][0]["observer_error_type"] == "RuntimeError" + assert diagnostics["late_completions"][0]["payload_discarded"] is True + + follow_up_gateway = _gateway_for( + method_name="synthetic_write_after_observer_failure", + handler=lambda: {"success": True, "changed": True}, + transport=transport, + category="write", + ) + follow_up = follow_up_gateway.invoke( + "synthetic_write_after_observer_failure", + {}, + ) + assert follow_up.outcome == "completed" + assert follow_up.payload == {"success": True, "changed": True} + assert executor.diagnostics()["completed_count"] >= 2 + + def test_handler_pool_rejects_excess_work_instead_of_growing_unbounded() -> None: handler_started = Event() + rejected_write_started = Event() release_handler = Event() executor = _BoundedHandlerExecutor(worker_count=1, queue_capacity=1) transport = InternalMCPTransport( @@ -392,11 +665,20 @@ def _handler(): ): time.sleep(0.002) - rejected = gateway.invoke("synthetic_bounded_pool_read", {}) + rejected_write_gateway = _gateway_for( + method_name="synthetic_rejected_pool_write", + handler=lambda: rejected_write_started.set(), + transport=transport, + category="write", + ) + rejected = rejected_write_gateway.invoke("synthetic_rejected_pool_write", {}) assert rejected.outcome == "saturated" assert rejected.payload["error_code"] == "internal_mcp_handler_pool_saturated" assert rejected.payload["retryable"] is True + assert rejected.payload["mutation_outcome"] == "not_started" + assert rejected.payload["recovery_affordances"][0]["action_type"] == "bounded_retry" + assert rejected_write_started.is_set() is False assert executor.diagnostics()["worker_count"] == 1 assert executor.diagnostics()["queue_capacity"] == 1 assert executor.diagnostics()["rejected_count"] == 1 diff --git a/tests/backend/test_mcp_create_concepts_extent_batching.py b/tests/backend/test_mcp_create_concepts_extent_batching.py new file mode 100644 index 000000000..2aea16d6a --- /dev/null +++ b/tests/backend/test_mcp_create_concepts_extent_batching.py @@ -0,0 +1,181 @@ +from __future__ import annotations + +import pytest + +from src.backend.integrations.internal_mcp.catalogue import _create_concepts +from src.backend.services import ( + create_concepts_duplicate_guard_service, + create_concepts_parent_resolution_service, + relationship_extent_index_service, + workflow_event_integration_service, +) +from src.backend.services.create_concepts_parent_resolution_service import ( + ParentResolutionResult, +) +from src.backend.vontology import utils_vontology + + +def test_create_concepts_coalesces_relationship_extent_sync_for_the_batch( + monkeypatch, +) -> None: + parent_id = "#V#bounded_parent" + single_syncs: list[str] = [] + batch_syncs: list[list[str]] = [] + + monkeypatch.setattr( + create_concepts_parent_resolution_service, + "resolve_parent_for_create_concepts", + lambda requested_parent_id: ParentResolutionResult( + requested_parent_id=requested_parent_id, + canonical_parent_id=parent_id, + resolved_parent_id=parent_id, + fallback_used=False, + fallback_candidates_checked=(), + fallback_selected_parent_id=None, + ), + ) + monkeypatch.setattr( + create_concepts_duplicate_guard_service, + "find_existing_concept_for_create_concepts", + lambda **_kwargs: None, + ) + monkeypatch.setattr( + workflow_event_integration_service, + "resolve_event_actor_context", + lambda **_kwargs: (None, None), + ) + monkeypatch.setattr( + relationship_extent_index_service, + "_sync_relationship_extent_index_for_concept_id_now", + lambda source_id: single_syncs.append(source_id) or {"success": True}, + ) + monkeypatch.setattr( + relationship_extent_index_service, + "_sync_relationship_extent_index_for_concept_ids_now", + lambda source_ids: batch_syncs.append(list(source_ids)) or {"success": True}, + ) + + def _fake_create_vontology_concept( + *, + parent_id: str, + new_concept_name: str, + **_kwargs, + ) -> dict: + concept_id = f"#V#{new_concept_name.lower().replace(' ', '_')}" + relationship_extent_index_service.sync_relationship_extent_index_for_concept_id( + parent_id + ) + relationship_extent_index_service.sync_relationship_extent_index_for_concept_id( + concept_id + ) + return { + "success": True, + "concept": {"concept_id": concept_id}, + "canonical_concept_id": concept_id, + } + + monkeypatch.setattr( + utils_vontology, + "create_vontology_concept", + _fake_create_vontology_concept, + ) + + result = _create_concepts( + parent_id=parent_id, + concepts=[ + {"name": "First bounded concept", "kind": "type"}, + {"name": "Second bounded concept", "kind": "type"}, + ], + ) + + assert result["successful"] == 2 + assert single_syncs == [] + assert batch_syncs == [ + [ + "#V#bounded_parent", + "#V#first_bounded_concept", + "#V#second_bounded_concept", + ] + ] + + +def test_create_concepts_flushes_completed_extent_syncs_when_batch_stops( + monkeypatch, +) -> None: + parent_id = "#V#bounded_parent" + created_names: list[str] = [] + batch_syncs: list[list[str]] = [] + + monkeypatch.setattr( + create_concepts_parent_resolution_service, + "resolve_parent_for_create_concepts", + lambda requested_parent_id: ParentResolutionResult( + requested_parent_id=requested_parent_id, + canonical_parent_id=parent_id, + resolved_parent_id=parent_id, + fallback_used=False, + fallback_candidates_checked=(), + fallback_selected_parent_id=None, + ), + ) + monkeypatch.setattr( + create_concepts_duplicate_guard_service, + "find_existing_concept_for_create_concepts", + lambda **_kwargs: None, + ) + monkeypatch.setattr( + workflow_event_integration_service, + "resolve_event_actor_context", + lambda **_kwargs: (None, None), + ) + monkeypatch.setattr( + relationship_extent_index_service, + "_sync_relationship_extent_index_for_concept_ids_now", + lambda source_ids: batch_syncs.append(list(source_ids)) or {"success": True}, + ) + + def _fake_create_vontology_concept( + *, + parent_id: str, + new_concept_name: str, + **_kwargs, + ) -> dict: + created_names.append(new_concept_name) + if new_concept_name == "Second bounded concept": + raise RuntimeError("simulated canonical write failure") + concept_id = f"#V#{new_concept_name.lower().replace(' ', '_')}" + relationship_extent_index_service.sync_relationship_extent_index_for_concept_id( + parent_id + ) + relationship_extent_index_service.sync_relationship_extent_index_for_concept_id( + concept_id + ) + return { + "success": True, + "concept": {"concept_id": concept_id}, + "canonical_concept_id": concept_id, + } + + monkeypatch.setattr( + utils_vontology, + "create_vontology_concept", + _fake_create_vontology_concept, + ) + + with pytest.raises(RuntimeError, match="simulated canonical write failure"): + _create_concepts( + parent_id=parent_id, + concepts=[ + {"name": "First bounded concept", "kind": "type"}, + {"name": "Second bounded concept", "kind": "type"}, + {"name": "Third bounded concept", "kind": "type"}, + ], + ) + + assert created_names == [ + "First bounded concept", + "Second bounded concept", + ] + assert batch_syncs == [ + ["#V#bounded_parent", "#V#first_bounded_concept"] + ] diff --git a/tests/backend/test_mcp_stdio_server_exposes_turn_execution_diagnostics_tool.py b/tests/backend/test_mcp_stdio_server_exposes_turn_execution_diagnostics_tool.py index d15e49017..21d9dcdbc 100644 --- a/tests/backend/test_mcp_stdio_server_exposes_turn_execution_diagnostics_tool.py +++ b/tests/backend/test_mcp_stdio_server_exposes_turn_execution_diagnostics_tool.py @@ -1,3 +1,4 @@ +import asyncio import json from pathlib import Path @@ -13,6 +14,27 @@ def test_mcp_stdio_server_has_turn_execution_diagnostics_handler() -> None: assert "conversation_telemetry_get_locator" in mcp_stdio_server._TOOL_HANDLERS assert "turn_execution_get_live_progress" in mcp_stdio_server._TOOL_HANDLERS assert "workflow_list_use_episodes" in mcp_stdio_server._TOOL_HANDLERS + assert "mongo_query_diagnostics_report" in mcp_stdio_server._TOOL_HANDLERS + + +def test_mongo_query_diagnostics_stdio_routes_the_runtime_binding(monkeypatch) -> None: + from src.backend.mcp_server import mcp_stdio_server + + monkeypatch.setattr( + mcp_stdio_server, + "_mongo_query_diagnostics_report", + lambda **_kwargs: {"success": True, "source": "stdio_runtime_binding"}, + ) + + result = asyncio.run( + mcp_stdio_server.call_tool( + "mongo_query_diagnostics_report", + {"allow_operator_diagnostics": True}, + ) + ) + payload = json.loads(result[0].text) + + assert payload == {"success": True, "source": "stdio_runtime_binding"} def test_vontology_mcp_manifest_includes_turn_execution_diagnostics_tool() -> None: @@ -34,3 +56,4 @@ def test_vontology_mcp_manifest_includes_turn_execution_diagnostics_tool() -> No assert "conversation_telemetry_get_locator" in names assert "turn_execution_get_live_progress" in names assert "workflow_list_use_episodes" in names + assert "mongo_query_diagnostics_report" in names 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 c366e4296..f537a72ce 100644 --- a/tests/backend/test_turn_execution_record_service_execution_summary.py +++ b/tests/backend/test_turn_execution_record_service_execution_summary.py @@ -7,6 +7,7 @@ ) from src.backend.services.turn_execution_record_service import ( TURN_EXECUTION_CORRECTNESS_SCHEMA_VERSION, + append_late_effect_observation, build_turn_execution_correctness_summary, build_turn_execution_record, build_workflow_routing_diagnostics, @@ -137,6 +138,162 @@ def find_one(self, query, **_kwargs): assert invocation["transport"] == transport_metadata +def test_late_effect_observation_is_bounded_idempotent_and_survives_full_upsert( + monkeypatch, +) -> None: + class _Collection: + document = None + + def update_one(self, _query, update, **_kwargs): + existed = self.document is not None + if self.document is None: + self.document = dict(update.get("$setOnInsert") or {}) + modified = False + if "$set" in update: + for key, value in update["$set"].items(): + if self.document.get(key) != value: + modified = True + self.document[key] = value + for key, value in (update.get("$addToSet") or {}).items(): + values = self.document.setdefault(key, []) + if value not in values: + values.append(value) + modified = True + return SimpleNamespace( + modified_count=1 if modified and existed else 0, + matched_count=1 if existed else 0, + upserted_id=None if existed else "late-observation-1", + ) + + def find_one(self, query, **_kwargs): + if not self.document: + return None + for field_name, expected in query.items(): + if self.document.get(field_name) != expected: + return None + return dict(self.document) + + collection = _Collection() + import src.backend.services.turn_execution_record_service as record_service + + monkeypatch.setattr( + record_service, + "get_turn_execution_records_collection", + lambda: collection, + ) + late_observation = { + "schema_version": "internal_mcp_late_completion.v1", + "execution_id": "mcp_late_effect_1", + "method_name": "synthetic_effect", + "category": "write", + "outcome": "late_success", + "observed_at_utc": "2026-07-27T10:00:00Z", + "queue_duration_ms": 1.0, + "handler_duration_ms": 22_000.0, + "payload": { + "success": True, + "effect_status": "succeeded", + "changed": True, + "created_ids": ["#V#late_created"], + "api_key": "must-not-be-persisted", + "large_detail": "x" * 20_000, + }, + "payload_truncated": False, + "payload_redacted": False, + "error_type": None, + "error": None, + } + + first = append_late_effect_observation( + request_id="req-late-effect", + effect_id="effect-late-1", + execution_id="mcp_late_effect_1", + observation=late_observation, + user_id="#V#user", + session_id="session-late-effect", + namespace="#V#user@org", + org_id="#V#org", + ) + duplicate = append_late_effect_observation( + request_id="req-late-effect", + effect_id="effect-late-1", + execution_id="mcp_late_effect_1", + observation=late_observation, + user_id="#V#user", + session_id="session-late-effect", + namespace="#V#user@org", + org_id="#V#org", + ) + assert first["appended"] is True + assert duplicate["appended"] is False + assert duplicate["duplicate"] is True + assert collection.document is not None + assert len(collection.document["late_effect_observations"]) == 1 + stored_observation = collection.document["late_effect_observations"][0] + assert stored_observation["effect_id"] == "effect-late-1" + assert stored_observation["execution_id"] == "mcp_late_effect_1" + assert stored_observation["payload"]["effect_status"] == "succeeded" + assert stored_observation["payload"]["api_key"] == "[redacted]" + assert len(stored_observation["payload"]["large_detail"]) == 2_003 + assert len(json.dumps(stored_observation)) < 64_000 + assert "source_observation_sha256" not in stored_observation + assert "storage_truncated" not in stored_observation + + full_record = build_turn_execution_record( + request_id="req-late-effect", + session_id="session-late-effect", + namespace="#V#user@org", + user_id="#V#user", + org_id="#V#org", + prompt_text="Apply the bounded effect.", + response_text="The turn later completed.", + interaction_timestamp_utc="2026-07-27T10:00:01Z", + tool_invocations=[ + { + "tool": "synthetic_effect", + "status": "timeout", + "effect_id": "effect-late-1", + "effect_status": "indeterminate", + "changed": False, + "execution_id": "mcp_late_effect_1", + "effective_payload": { + "success": False, + "error_code": "tool_timeout_outcome_unknown", + }, + } + ], + ) + # Even an accidentally stale whole-turn snapshot must not replace the + # independently appended observation. + full_record["late_effect_observations"] = [] + upsert_turn_execution_record_projection( + record=full_record, + user_id="#V#user", + session_id="session-late-effect", + namespace="#V#user@org", + org_id="#V#org", + ) + readback = get_turn_execution_record_projection( + request_id="req-late-effect", + namespace="#V#user@org", + ) + + assert readback is not None + assert len(readback["late_effect_observations"]) == 1 + assert ( + readback["late_effect_observations"][0]["observation_id"] + == first["observation_id"] + ) + invocation = readback["execution"]["tool_invocations"][0] + assert invocation["effect_status"] == "indeterminate" + assert invocation["changed"] is False + assert ( + readback["late_effect_observations"][0]["payload"]["effect_status"] + == "succeeded" + ) + assert readback["late_effect_observations"][0]["payload"]["changed"] is True + + def test_projection_field_telemetry_preserves_bounded_collection_row_index() -> None: entries = _normalise_projection_field_entries( [ diff --git a/tests/backend/test_von_generate_workflow_instances.py b/tests/backend/test_von_generate_workflow_instances.py index d98de2b74..7eda05203 100644 --- a/tests/backend/test_von_generate_workflow_instances.py +++ b/tests/backend/test_von_generate_workflow_instances.py @@ -26,6 +26,8 @@ def app(monkeypatch: pytest.MonkeyPatch) -> Flask: "render_plan": None, "extra_messages": (), "tool_invocations": (), + "terminal_status": "completed", + "effect_finality_fallback": False, } def _execute_adaptive_turn(**kwargs: Any) -> AdaptiveTurnResult: @@ -45,7 +47,10 @@ def _execute_adaptive_turn(**kwargs: Any) -> AdaptiveTurnResult: llm_usage={"total_tokens": 7}, duration_ms=2.0, render_plan=adaptive_state["render_plan"], - terminal_status="completed", + terminal_status=adaptive_state["terminal_status"], + effect_finality_fallback=adaptive_state[ + "effect_finality_fallback" + ], ) def _retired_outer_controller_called(*_args: Any, **_kwargs: Any) -> None: @@ -364,6 +369,57 @@ def test_presenter_mode_projects_tagged_adaptive_answer(app: Flask) -> None: } +def test_effect_finality_fallback_is_presented_literally_without_model_backfill( + app: Flask, + monkeypatch: pytest.MonkeyPatch, +) -> None: + from src.backend.server.routes import von_routes + + fallback = ( + "That turn did not finish cleanly. An effect remains indeterminate; " + "inspect represented state before retrying it." + ) + adaptive_state = app.config["_ADAPTIVE_TURN_STATE"] + adaptive_state["response_text"] = fallback + adaptive_state["terminal_status"] = "model_error" + adaptive_state["effect_finality_fallback"] = True + + def _unexpected_backfill(*_args: Any, **_kwargs: Any) -> str: + raise AssertionError("effect-finality fallback must remain literal") + + monkeypatch.setattr( + von_routes, + "_invoke_presenter_screen_backfill_prompt", + _unexpected_backfill, + ) + monkeypatch.setattr( + von_routes, + "_llm_generate_spoken_backfill", + _unexpected_backfill, + ) + + response = app.test_client().post( + "/von/generate", + json={"prompt": "Present the result.", "presenter_mode": True}, + ) + + assert response.status_code == 200 + payload = response.get_json() + assert payload["success"] is False + assert payload["response"] == fallback + assert payload["response_channels"] == { + "spoken": fallback, + "screen": fallback, + "format": "effect_finality_fallback_v1", + } + assert ( + payload["llm_debug"]["screen_backfill_second_pass_attempted"] is False + ) + assert ( + payload["llm_debug"]["spoken_backfill_second_pass_attempted"] is False + ) + + def test_presenter_channel_parser_ignores_tags_inside_fenced_blocks() -> None: from src.backend.server.routes.von_routes import _extract_presenter_channels