From a5b469ee1229941ae8946aed1b0ee9028e209e9b Mon Sep 17 00:00:00 2001 From: witbrock Date: Tue, 28 Jul 2026 16:53:54 +0100 Subject: [PATCH] Add actor-scoped telemetry read delegation --- .../workflow_mcp_capability_matrix.md | 18 +- .../integrations/internal_mcp/catalogue.py | 661 ++++++++++++++++-- src/backend/mcp_server/vontology_mcp.json | 46 +- src/backend/server/routes/von_routes.py | 312 ++++++++- src/backend/services/chat_history_service.py | 117 ++++ .../conversation_scope_binding_service.py | 418 ++++++++++- .../conversation_telemetry_locator_service.py | 199 +++++- .../turn_execution_diagnostics_service.py | 170 ++++- .../web/von_interface/static/js/chatTab.js | 350 +++++----- .../static/js/test/chatTab.test.js | 31 +- .../test_chat_history_service_segments.py | 60 ++ ...test_conversation_scope_binding_service.py | 115 +++ ...test_internal_mcp_locator_tools_gateway.py | 12 +- ...mcp_stdio_server_exposes_workflow_tools.py | 5 +- .../test_telemetry_read_delegation_stdio.py | 409 +++++++++++ ...ecution_diagnostics_workflow_trace_refs.py | 67 ++ ..._von_history_telemetry_locator_endpoint.py | 110 +++ ...hatTabConversationLlmTelemetryCopy.test.js | 69 +- .../chatTabLlmDebugWorkflowExecution.test.js | 117 +++- .../chatTabThinkingDiagnosticsCopy.test.js | 31 + 20 files changed, 2951 insertions(+), 366 deletions(-) create mode 100644 tests/backend/test_telemetry_read_delegation_stdio.py diff --git a/docs/engineering/workflow_mcp_capability_matrix.md b/docs/engineering/workflow_mcp_capability_matrix.md index 9cc232603..63c31e251 100644 --- a/docs/engineering/workflow_mcp_capability_matrix.md +++ b/docs/engineering/workflow_mcp_capability_matrix.md @@ -10,7 +10,7 @@ available. It prevents ambiguous "missing tool" behaviour across MCP surfaces. | Surface | Intended usage | | --- | --- | | `internal_mcp_gateway` | Canonical workflow management and introspection surface used by Von internals. | -| `vontology_mcp_stdio_server` | Vontology concept/text tools for external IDE agents. Durable `workflow_*` tools are exposed directly; chat introspection tools remain internal-only. | +| `vontology_mcp_stdio_server` | Vontology concept/text tools for external IDE agents. Durable `workflow_*` tools are exposed directly. Selected exact telemetry reads accept short-lived actor-bound delegations; other chat introspection remains internal-only. | | `vonrag_mcp_stdio_server` | Focused RAG/search tools only. | ## Tool Availability @@ -37,6 +37,11 @@ available. It prevents ambiguous "missing tool" behaviour across MCP surfaces. | `workflow_delete_schedule` | Yes | Yes | No | | `workflow_trigger_schedule` | Yes | Yes | No | | `workflow_mcp_health_check` | Yes | Yes | No | +| `conversation_telemetry_get_locator` | Yes | Yes (signed actor delegation or operator) | No | +| `chat_history_get_segments` | Yes | Yes (signed actor delegation or operator) | No | +| `chat_history_get_debug_entry` | Yes | Yes (signed actor delegation or operator) | No | +| `turn_execution_get_diagnostics` | Yes | Yes (signed actor delegation or operator) | No | +| `turn_execution_get_live_progress` | Yes | Yes (signed actor delegation or operator) | No | | `chat_get_prompt_context` | Yes | No | No | | `chat_introspect` | Yes | No | No | | `settings_get_public` | Yes | No | No | @@ -45,6 +50,16 @@ available. It prevents ambiguous "missing tool" behaviour across MCP surfaces. - `vontology_mcp_stdio_server` exposes direct durable `workflow_*` calls and can also access workflow behaviour indirectly via `von_chat_run` (tool orchestration pathway). +- Browser telemetry-copy surfaces issue short-lived references bound to the + authenticated actor, audience, exact tool, organisation/namespace, and exact + conversation, history entry, or request. The stdio server verifies those + claims and canonical ownership before returning a bounded read. +- Raw session, history, request, user, organisation, or namespace fields are + locators only and do not grant telemetry authority. Tampered, expired, + cross-actor, cross-tool, or cross-target references fail closed. +- The delegation does not grant global list, workflow trace, critic, + experiment, mutation, or control-plane authority. Those surfaces retain + their existing operator or global-admin requirements. - `vonrag_mcp_stdio_server` is intentionally limited to RAG/search tools. ## Diagnostics Contract @@ -110,4 +125,3 @@ Suggested rollback sequence: 2. Confirm diagnostics stream is healthy (`workflow_metadata_validation_events` present). 3. If workflows still degrade, set `VON_WORKFLOW_METADATA_VALIDATION_MODE=off`. 4. After remediation, restore to `enforce`. - diff --git a/src/backend/integrations/internal_mcp/catalogue.py b/src/backend/integrations/internal_mcp/catalogue.py index 11e9ea886..cee33b962 100644 --- a/src/backend/integrations/internal_mcp/catalogue.py +++ b/src/backend/integrations/internal_mcp/catalogue.py @@ -15378,19 +15378,121 @@ def _turn_execution_namespace_coverage_report(**kwargs): ) +def _bounded_delegated_telemetry_payload( + payload: Mapping[str, Any], + *, + arguments: Mapping[str, Any], + delegated: bool, + artifact_kind: str, + preserve_inline_below_limit: bool, +) -> dict[str, Any]: + """Keep bearer-delegated stdio reads below the MCP text response guard.""" + + if not delegated: + return dict(payload) + + import hashlib + import json + + default_limit = 60_000 + max_limit = 75_000 + raw_offset = arguments.get("offset", 0) + raw_limit = arguments.get("limit", default_limit) + offset = raw_offset if isinstance(raw_offset, int) and not isinstance(raw_offset, bool) else 0 + limit = raw_limit if isinstance(raw_limit, int) and not isinstance(raw_limit, bool) else default_limit + offset = max(0, offset) + limit = min(max_limit, max(1, limit)) + canonical_json = json.dumps( + dict(payload), + sort_keys=True, + separators=(",", ":"), + ensure_ascii=True, + ) + total_chars = len(canonical_json) + digest = hashlib.sha256(canonical_json.encode("utf-8")).hexdigest() + end = min(total_chars, offset + limit) + page = { + "schema_version": "bounded_telemetry_json_page.v1", + "artifact_kind": artifact_kind, + "encoding": "canonical_json_ascii", + "sha256": digest, + "offset": offset, + "limit": limit, + "returned_chars": max(0, end - offset), + "total_chars": total_chars, + "has_more": end < total_chars, + "next_offset": end if end < total_chars else None, + "json_chunk": canonical_json[offset:end] if offset <= total_chars else "", + } + if ( + preserve_inline_below_limit + and offset == 0 + and total_chars <= limit + ): + return { + **dict(payload), + "bounded_read": { + key: value for key, value in page.items() if key != "json_chunk" + }, + } + return { + "success": True, + "artifact_kind": artifact_kind, + "request_id": payload.get("request_id"), + "history_location": payload.get("history_location"), + "read_delegation": payload.get("read_delegation"), + "provenance": payload.get("provenance"), + "bounded_read": page, + } + + def _chat_history_get_segments(**kwargs): - if denial := _internal_mcp_actor_scoped_read_denial("chat history"): - return denial + authorisation = _authorise_internal_mcp_telemetry_read( + kwargs, + tool_name="chat_history_get_segments", + surface="chat history", + reference_argument="conversation_ref", + operator_control_plane=False, + ) + if not authorisation.get("success"): + return authorisation + effective_kwargs = dict(authorisation["payload"]) + if authorisation.get("delegated"): + effective_kwargs["include_debug"] = False + requested_segment_size = effective_kwargs.get("segment_size") + effective_kwargs["segment_size"] = min( + 20, + ( + requested_segment_size + if isinstance(requested_segment_size, int) + and not isinstance(requested_segment_size, bool) + and requested_segment_size > 0 + else 20 + ), + ) + requested_tail_limit = effective_kwargs.get("history_tail_limit") + effective_kwargs["history_tail_limit"] = min( + 100, + ( + requested_tail_limit + if isinstance(requested_tail_limit, int) + and not isinstance(requested_tail_limit, bool) + and requested_tail_limit > 0 + else 100 + ), + ) from ...services import chat_history_service - access = _resolve_chat_history_read_target(kwargs) + access = _resolve_chat_history_read_target(effective_kwargs) if not isinstance(access, dict) or not access.get("success", False): return access - include_debug = kwargs.get("include_debug", True) + include_debug = effective_kwargs.get("include_debug", True) if not isinstance(include_debug, bool): include_debug = bool(include_debug) - include_legacy = kwargs.get("include_legacy", access.get("include_legacy", True)) + include_legacy = effective_kwargs.get( + "include_legacy", access.get("include_legacy", True) + ) if not isinstance(include_legacy, bool): include_legacy = bool(include_legacy) @@ -15401,9 +15503,9 @@ def _chat_history_get_segments(**kwargs): include_locations=True, namespace=_clean_optional_string(access.get("read_namespace")), include_legacy=include_legacy, - segment_size=kwargs.get("segment_size"), + segment_size=effective_kwargs.get("segment_size"), include_debug=include_debug, - history_tail_limit=kwargs.get("history_tail_limit"), + history_tail_limit=effective_kwargs.get("history_tail_limit"), return_meta=True, ) except chat_history_service.ChatHistoryServiceError as exc: @@ -15436,6 +15538,8 @@ def _chat_history_get_segments(**kwargs): meta.get("history_truncated") if isinstance(meta, Mapping) else False ), } + if authorisation.get("delegated"): + payload["read_delegation"] = authorisation.get("read_delegation") return _with_rag_provenance( payload=payload, item_kind="chat_history_segments", @@ -15444,15 +15548,23 @@ def _chat_history_get_segments(**kwargs): def _chat_history_get_debug_entry(**kwargs): - if denial := _internal_mcp_actor_scoped_read_denial("chat history"): - return denial + authorisation = _authorise_internal_mcp_telemetry_read( + kwargs, + tool_name="chat_history_get_debug_entry", + surface="chat history", + reference_argument="history_location_ref", + operator_control_plane=False, + ) + if not authorisation.get("success"): + return authorisation + effective_kwargs = dict(authorisation["payload"]) from ...services import chat_history_service - access = _resolve_chat_history_read_target(kwargs) + access = _resolve_chat_history_read_target(effective_kwargs) if not isinstance(access, dict) or not access.get("success", False): return access - history_index = kwargs.get("history_index") + history_index = effective_kwargs.get("history_index") if not isinstance(history_index, int): history_index = access.get("history_index") if not isinstance(history_index, int) or history_index < 0: @@ -15466,7 +15578,9 @@ def _chat_history_get_debug_entry(**kwargs): ], ) - include_legacy = kwargs.get("include_legacy", access.get("include_legacy", True)) + include_legacy = effective_kwargs.get( + "include_legacy", access.get("include_legacy", True) + ) if not isinstance(include_legacy, bool): include_legacy = bool(include_legacy) @@ -15519,21 +15633,38 @@ def _chat_history_get_debug_entry(**kwargs): "identifier_binding": access.get("identifier_binding"), "llm_debug_data": debug_data, } - return _with_rag_provenance( + if authorisation.get("delegated"): + payload["read_delegation"] = authorisation.get("read_delegation") + result = _with_rag_provenance( payload=payload, item_kind="chat_history_debug_entry", source_system="mongo.chat_history", ) + return _bounded_delegated_telemetry_payload( + result, + arguments=effective_kwargs, + delegated=bool(authorisation.get("delegated")), + artifact_kind="chat_history_debug_entry", + preserve_inline_below_limit=True, + ) def _conversation_telemetry_get_locator(**kwargs): - if denial := _internal_mcp_actor_scoped_read_denial("conversation telemetry"): - return denial + authorisation = _authorise_internal_mcp_telemetry_read( + kwargs, + tool_name="conversation_telemetry_get_locator", + surface="conversation telemetry", + reference_argument="conversation_ref", + operator_control_plane=False, + ) + if not authorisation.get("success"): + return authorisation + effective_kwargs = dict(authorisation["payload"]) from ...services.conversation_telemetry_locator_service import ( build_conversation_llm_telemetry_locator, ) - access = _resolve_chat_history_read_target(kwargs) + access = _resolve_chat_history_read_target(effective_kwargs) if not isinstance(access, dict) or not access.get("success", False): return access @@ -15552,7 +15683,9 @@ def _conversation_telemetry_get_locator(**kwargs): access.get("organisation_concept_id") ), include_legacy=bool( - kwargs.get("include_legacy", access.get("include_legacy", True)) + effective_kwargs.get( + "include_legacy", access.get("include_legacy", True) + ) ), ) except Exception as exc: @@ -15569,6 +15702,8 @@ def _conversation_telemetry_get_locator(**kwargs): payload["requested_user_id"] = access.get("requested_user_id") payload["access_mode"] = access.get("access_mode") payload["identifier_binding"] = access.get("identifier_binding") + if authorisation.get("delegated"): + payload["read_delegation"] = authorisation.get("read_delegation") return _with_rag_provenance( payload=payload, item_kind="conversation_telemetry_locator", @@ -15577,13 +15712,21 @@ def _conversation_telemetry_get_locator(**kwargs): def _turn_execution_get_live_progress(**kwargs): - if denial := _internal_mcp_operator_control_plane_denial("turn telemetry"): - return denial + authorisation = _authorise_internal_mcp_telemetry_read( + kwargs, + tool_name="turn_execution_get_live_progress", + surface="turn telemetry", + reference_argument="turn_telemetry_ref", + operator_control_plane=True, + ) + if not authorisation.get("success"): + return authorisation + effective_kwargs = dict(authorisation["payload"]) from ...services.turn_execution_live_progress_service import ( get_turn_execution_live_progress_payload, ) - request_id = kwargs.get("request_id") + request_id = effective_kwargs.get("request_id") if not isinstance(request_id, str) or not request_id.strip(): return make_error_response( "missing_parameter", @@ -15594,18 +15737,32 @@ def _turn_execution_get_live_progress(**kwargs): payload = get_turn_execution_live_progress_payload( request_id=request_id.strip(), - namespace=_clean_optional_string(kwargs.get("namespace")), + namespace=_clean_optional_string(effective_kwargs.get("namespace")), user_concept_id=_normalise_optional_concept_id( - kwargs.get("user_concept_id") - or kwargs.get("acting_user_concept_id") - or kwargs.get("actor_user_id") - ), - window_session_id=_clean_optional_string(kwargs.get("window_session_id")), - anonymous_session_id=_clean_optional_string(kwargs.get("anonymous_session_id")), - scope_key=_clean_optional_string(kwargs.get("scope_key")), - section=_clean_optional_string(kwargs.get("section")), - limit=kwargs.get("limit"), - offset=kwargs.get("offset"), + effective_kwargs.get("user_concept_id") + or effective_kwargs.get("acting_user_concept_id") + or effective_kwargs.get("actor_user_id") + ), + window_session_id=( + None + if authorisation.get("delegated") + else _clean_optional_string(effective_kwargs.get("window_session_id")) + ), + anonymous_session_id=( + None + if authorisation.get("delegated") + else _clean_optional_string( + effective_kwargs.get("anonymous_session_id") + ) + ), + scope_key=( + None + if authorisation.get("delegated") + else _clean_optional_string(effective_kwargs.get("scope_key")) + ), + section=_clean_optional_string(effective_kwargs.get("section")), + limit=effective_kwargs.get("limit"), + offset=effective_kwargs.get("offset"), ) if not isinstance(payload, dict): return make_error_response( @@ -15617,6 +15774,83 @@ def _turn_execution_get_live_progress(**kwargs): "Provide user_concept_id, namespace, or the window_session_id used by the browser", ], ) + if authorisation.get("delegated"): + delegated_actor = _normalise_optional_concept_id( + effective_kwargs.get("user_concept_id") + ) + expected_scope_key = ( + f"user:{delegated_actor}" if delegated_actor else None + ) + resolved_scope_key = _clean_optional_string( + payload.get("resolved_scope_key") + ) + if expected_scope_key and resolved_scope_key != expected_scope_key: + return make_error_response( + "READ_DELEGATION_CANONICAL_TARGET_MISMATCH", + "Canonical live progress does not match the signed actor scope", + details={ + "field_mismatches": [ + { + "field": "resolved_scope_key", + "read_back": resolved_scope_key, + "bound": expected_scope_key, + } + ], + "identifier_binding": authorisation.get( + "identifier_binding" + ), + }, + ) + from ...services.conversation_scope_binding_service import ( + build_turn_telemetry_binding, + ) + + payload["read_delegation"] = authorisation.get("read_delegation") + payload["mcp_access"] = { + "turn_execution_get_live_progress": { + "tool_name": "turn_execution_get_live_progress", + "arguments": { + "request_id": request_id.strip(), + "turn_telemetry_ref": build_turn_telemetry_binding( + request_id=request_id.strip(), + delegated_actor_user_id=str( + effective_kwargs["user_concept_id"] + ), + permitted_tool="turn_execution_get_live_progress", + chat_session_id=_clean_optional_string( + effective_kwargs.get("session_id") + ), + history_owner_user_id=_normalise_optional_concept_id( + effective_kwargs.get( + "_delegated_history_owner_user_id" + ) + ), + read_namespace=_clean_optional_string( + effective_kwargs.get("_delegated_read_namespace") + ), + delegated_actor_namespace=_clean_optional_string( + effective_kwargs.get("namespace") + ), + organisation_concept_id=_normalise_optional_concept_id( + effective_kwargs.get("organisation_concept_id") + ), + ), + "namespace": _clean_optional_string( + effective_kwargs.get("namespace") + ), + "user_concept_id": _normalise_optional_concept_id( + effective_kwargs.get("user_concept_id") + ), + "organisation_concept_id": _normalise_optional_concept_id( + effective_kwargs.get("organisation_concept_id") + ), + }, + "purpose": ( + "Fetch the bounded live progress snapshot; pass section, " + "limit, and offset only for explicit detail hydration." + ), + } + } return _with_rag_provenance( payload=payload, item_kind="turn_live_progress", @@ -15650,14 +15884,22 @@ def _turn_execution_get(**kwargs): def _turn_execution_get_diagnostics(**kwargs): - if denial := _internal_mcp_operator_control_plane_denial("turn telemetry"): - return denial + authorisation = _authorise_internal_mcp_telemetry_read( + kwargs, + tool_name="turn_execution_get_diagnostics", + surface="turn telemetry", + reference_argument="turn_telemetry_ref", + operator_control_plane=True, + ) + if not authorisation.get("success"): + return authorisation + effective_kwargs = dict(authorisation["payload"]) from ...services.turn_execution_diagnostics_service import ( TurnExecutionDiagnosticsServiceError, get_turn_execution_diagnostics_payload, ) - request_id = kwargs.get("request_id") + request_id = effective_kwargs.get("request_id") if not isinstance(request_id, str) or not request_id.strip(): return make_error_response( "missing_parameter", @@ -15670,7 +15912,41 @@ def _turn_execution_get_diagnostics(**kwargs): try: payload = get_turn_execution_diagnostics_payload( request_id=request_id_value, - namespace=kwargs.get("namespace"), + namespace=( + effective_kwargs.get("_delegated_read_namespace") + if authorisation.get("delegated") + else effective_kwargs.get("namespace") + ), + delegated_actor_user_id=_normalise_optional_concept_id( + effective_kwargs.get("user_concept_id") + ), + delegated_actor_namespace=_clean_optional_string( + effective_kwargs.get("namespace") + ), + chat_session_id=( + _clean_optional_string(effective_kwargs.get("session_id")) + if authorisation.get("delegated") + else None + ), + history_index=( + effective_kwargs.get("history_index") + if authorisation.get("delegated") + else None + ), + history_owner_user_id=( + _normalise_optional_concept_id( + effective_kwargs.get("_delegated_history_owner_user_id") + ) + if authorisation.get("delegated") + else None + ), + organisation_concept_id=( + _normalise_optional_concept_id( + effective_kwargs.get("organisation_concept_id") + ) + if authorisation.get("delegated") + else None + ), ) except TurnExecutionDiagnosticsServiceError as exc: return make_error_response( @@ -15678,7 +15954,7 @@ def _turn_execution_get_diagnostics(**kwargs): str(exc), details={ "request_id": request_id_value, - "namespace": kwargs.get("namespace"), + "namespace": effective_kwargs.get("namespace"), }, ) @@ -15688,22 +15964,81 @@ def _turn_execution_get_diagnostics(**kwargs): f"Turn execution diagnostics {request_id_value} not found", details={ "request_id": request_id_value, - "namespace": kwargs.get("namespace"), + "namespace": effective_kwargs.get("namespace"), }, suggestions=["Check the request_id and namespace"], ) + if authorisation.get("delegated"): + canonical_comparisons: tuple[tuple[str, Any, Any], ...] = ( + ("request_id", payload.get("request_id"), request_id_value), + ( + "chat_session_id", + _clean_optional_string(payload.get("chat_session_id")), + _clean_optional_string(effective_kwargs.get("session_id")), + ), + ( + "history_owner_user_id", + _normalise_optional_concept_id( + payload.get("derived_user_concept_id") + ), + _normalise_optional_concept_id( + effective_kwargs.get("_delegated_history_owner_user_id") + ), + ), + ( + "read_namespace", + _clean_optional_string(payload.get("namespace")), + _clean_optional_string( + effective_kwargs.get("_delegated_read_namespace") + ), + ), + ( + "organisation_concept_id", + _normalise_optional_concept_id( + payload.get("derived_organisation_concept_id") + ), + _normalise_optional_concept_id( + effective_kwargs.get("organisation_concept_id") + ), + ), + ) + canonical_mismatches = [ + {"field": field, "read_back": actual, "bound": expected} + for field, actual, expected in canonical_comparisons + if actual is not None and expected is not None and actual != expected + ] + if canonical_mismatches: + return make_error_response( + "READ_DELEGATION_CANONICAL_TARGET_MISMATCH", + "Canonical turn diagnostics do not match the signed delegation target", + details={ + "field_mismatches": canonical_mismatches, + "identifier_binding": authorisation.get( + "identifier_binding" + ), + }, + ) + payload["read_delegation"] = authorisation.get("read_delegation") + payload["item_kind"] = "turn_execution_diagnostics" payload["source_system"] = ( "mongo.turn_execution_records" if payload.get("diagnostics_source") == "mongo.turn_execution_records" else "mongo.chat_history" ) - return _with_rag_provenance( + result = _with_rag_provenance( payload=payload, item_kind="turn_execution_diagnostics_item", source_system=str(payload["source_system"]), ) + return _bounded_delegated_telemetry_payload( + result, + arguments=effective_kwargs, + delegated=bool(authorisation.get("delegated")), + artifact_kind="turn_execution_diagnostics", + preserve_inline_below_limit=True, + ) def _failure_case_intake_collect(**kwargs): @@ -18480,6 +18815,200 @@ def _internal_mcp_actor_scoped_read_denial( ) +def _authorise_internal_mcp_telemetry_read( + payload: Mapping[str, Any], + *, + tool_name: str, + surface: str, + reference_argument: str, + operator_control_plane: bool, +) -> dict[str, Any]: + """Resolve a narrow signed stdio read delegation without actor elevation.""" + + ordinary_denial = ( + _internal_mcp_operator_control_plane_denial(surface) + if operator_control_plane + else _internal_mcp_actor_scoped_read_denial(surface) + ) + if ordinary_denial is None: + return { + "success": True, + "payload": dict(payload), + "delegated": False, + "read_delegation": None, + } + + reference = payload.get(reference_argument) + if not isinstance(reference, Mapping): + return ordinary_denial + if ( + reference_argument == "conversation_ref" + and reference.get("kind") == "von_conversation_ref" + ): + # Public locator wrappers are convenient identifiers, not proof of + # actor authority. Preserve the ordinary actor-provenance denial + # instead of treating an unsigned wrapper as a malformed signature. + return ordinary_denial + + from ...services.conversation_scope_binding_service import ( + TELEMETRY_READ_DELEGATION_AUDIENCE, + verify_conversation_scope_binding, + verify_history_location_binding, + verify_turn_telemetry_binding, + ) + + claimed_actor = _normalise_optional_concept_id(payload.get("user_concept_id")) + if reference_argument == "conversation_ref": + verified = verify_conversation_scope_binding( + reference, + require_read_delegation=True, + expected_audience=TELEMETRY_READ_DELEGATION_AUDIENCE, + expected_tool=tool_name, + expected_actor_user_id=claimed_actor, + ) + elif reference_argument == "history_location_ref": + verified = verify_history_location_binding( + reference, + require_read_delegation=True, + expected_audience=TELEMETRY_READ_DELEGATION_AUDIENCE, + expected_tool=tool_name, + expected_actor_user_id=claimed_actor, + ) + elif reference_argument == "turn_telemetry_ref": + verified = verify_turn_telemetry_binding( + reference, + expected_audience=TELEMETRY_READ_DELEGATION_AUDIENCE, + expected_tool=tool_name, + expected_actor_user_id=claimed_actor, + ) + else: + return make_error_response( + "READ_DELEGATION_REFERENCE_UNSUPPORTED", + "Unsupported telemetry read delegation reference", + details={"reference_argument": reference_argument}, + ) + + if not verified.get("success"): + return make_error_response( + str(verified.get("error_code") or "INVALID_CONTEXT_BINDING"), + str(verified.get("error_message") or "Telemetry read delegation is invalid"), + details={ + "identifier_binding": dict( + verified.get("identifier_binding") or {} + ) + }, + ) + + effective = dict(payload) + actor_user_id = _normalise_optional_concept_id( + verified.get("delegated_actor_user_id") + ) + actor_namespace = _clean_optional_string( + verified.get("delegated_actor_namespace") + ) + organisation_concept_id = _normalise_optional_concept_id( + verified.get("organisation_concept_id") + ) + bound_session_id = _clean_optional_string(verified.get("chat_session_id")) + bound_request_id = _clean_optional_string(verified.get("request_id")) + bound_history_index = verified.get("history_index") + + comparisons: tuple[tuple[str, Any, Any], ...] = ( + ("user_concept_id", claimed_actor, actor_user_id), + ( + "namespace", + _clean_optional_string(payload.get("namespace")), + actor_namespace, + ), + ( + "organisation_concept_id", + _normalise_optional_concept_id( + payload.get("organisation_concept_id") + ), + organisation_concept_id, + ), + ( + "session_id", + _clean_optional_string( + payload.get("session_id") + or payload.get("conversation_session_id") + ), + bound_session_id, + ), + ( + "request_id", + _clean_optional_string(payload.get("request_id")), + bound_request_id, + ), + ( + "history_index", + ( + payload.get("history_index") + if isinstance(payload.get("history_index"), int) + else None + ), + ( + bound_history_index + if isinstance(bound_history_index, int) + else None + ), + ), + ) + mismatches = [ + { + "field": field, + "provided": provided, + "bound": bound, + } + for field, provided, bound in comparisons + if provided is not None and bound is not None and provided != bound + ] + if mismatches: + identifier_binding = dict(verified.get("identifier_binding") or {}) + identifier_binding["read_delegation_validation_status"] = "target_mismatch" + return make_error_response( + "INVALID_CONTEXT_BINDING", + "Telemetry read arguments do not match the signed delegation target", + details={ + "reason_code": "READ_DELEGATION_TARGET_MISMATCH", + "field_mismatches": mismatches, + "identifier_binding": identifier_binding, + }, + ) + + if actor_user_id: + effective["user_concept_id"] = actor_user_id + if actor_namespace: + effective["namespace"] = actor_namespace + if organisation_concept_id: + effective["organisation_concept_id"] = organisation_concept_id + if bound_session_id: + effective["session_id"] = bound_session_id + if bound_request_id: + effective["request_id"] = bound_request_id + if isinstance(bound_history_index, int): + effective["history_index"] = bound_history_index + effective["_verified_read_delegation"] = dict( + verified.get("read_delegation") or {} + ) + effective["_verified_identifier_binding"] = dict( + verified.get("identifier_binding") or {} + ) + effective["_delegated_history_owner_user_id"] = _normalise_optional_concept_id( + verified.get("history_owner_user_id") + ) + effective["_delegated_read_namespace"] = _clean_optional_string( + verified.get("read_namespace") + ) + return { + "success": True, + "payload": effective, + "delegated": True, + "read_delegation": effective["_verified_read_delegation"], + "identifier_binding": effective["_verified_identifier_binding"], + } + + def _workflow_actor_scope_error_response(exc: Exception) -> dict[str, Any]: reason = str(getattr(exc, "reason", "") or "workflow_actor_scope_mismatch") mismatch_fields = getattr(exc, "mismatch_fields", ()) @@ -30593,6 +31122,9 @@ def _binding_error_result( bound_org = _normalise_optional_concept_id( bound_conversation_context.get("organisation_concept_id") ) + bound_read_namespace = _clean_optional_string( + bound_conversation_context.get("read_namespace") + ) if bound_owner and bound_owner != owner_user_id: return _binding_error_result( message=( @@ -30625,6 +31157,23 @@ def _binding_error_result( "chat_session_id": session_id, }, ) + if ( + bound_read_namespace + and owner_namespace + and bound_read_namespace != owner_namespace + ): + return _binding_error_result( + message=( + "Bound conversation reference namespace does not match " + "the resolved authorised read namespace" + ), + identifier_binding_payload=identifier_binding, + details={ + "bound_read_namespace": bound_read_namespace, + "resolved_read_namespace": owner_namespace, + "chat_session_id": session_id, + }, + ) return { "success": True, "session_id": session_id, @@ -30693,6 +31242,9 @@ def _binding_error_result( bound_org = _normalise_optional_concept_id( bound_conversation_context.get("organisation_concept_id") ) + bound_read_namespace = _clean_optional_string( + bound_conversation_context.get("read_namespace") + ) if bound_owner and bound_owner != read_user_id: return _binding_error_result( message=( @@ -30725,6 +31277,23 @@ def _binding_error_result( "chat_session_id": session_id, }, ) + if ( + bound_read_namespace + and read_namespace + and bound_read_namespace != read_namespace + ): + return _binding_error_result( + message=( + "Bound conversation reference namespace does not match " + "the resolved authorised read namespace" + ), + identifier_binding_payload=identifier_binding, + details={ + "bound_read_namespace": bound_read_namespace, + "resolved_read_namespace": read_namespace, + "chat_session_id": session_id, + }, + ) return { "success": True, @@ -34280,6 +34849,8 @@ def _build_default_catalogue_diagnostics_and_research_definitions() -> List[ "user_concept_id": (str, type(None)), "organisation_concept_id": (str, type(None)), "include_legacy": (bool,), + "offset": (int,), + "limit": (int,), }, allow_unknown=True, description=( @@ -34327,8 +34898,11 @@ def _build_default_catalogue_diagnostics_and_research_definitions() -> List[ input_schema=Schema( required={"request_id": str}, optional={ + "turn_telemetry_ref": (dict,), "namespace": (str, type(None)), "user_concept_id": (str, type(None)), + "organisation_concept_id": (str, type(None)), + "session_id": (str, type(None)), "window_session_id": (str, type(None)), "anonymous_session_id": (str, type(None)), "scope_key": (str, type(None)), @@ -35133,10 +35707,19 @@ def _build_default_catalogue_diagnostics_and_research_definitions() -> List[ input_schema=Schema( required={"request_id": str}, optional={ + "turn_telemetry_ref": (dict,), "namespace": (str, type(None)), + "user_concept_id": (str, type(None)), + "organisation_concept_id": (str, type(None)), + "offset": (int,), + "limit": (int,), }, allow_unknown=True, - description="Get one full turn diagnostics payload by request_id.", + description=( + "Get one exact turn diagnostics payload by request_id. External " + "stdio callers use a server-issued turn_telemetry_ref; large " + "payloads return canonical JSON pages controlled by offset/limit." + ), ), output_schema=None, category="read", diff --git a/src/backend/mcp_server/vontology_mcp.json b/src/backend/mcp_server/vontology_mcp.json index d48d7b599..7d665d5ad 100644 --- a/src/backend/mcp_server/vontology_mcp.json +++ b/src/backend/mcp_server/vontology_mcp.json @@ -202,6 +202,12 @@ "null" ] }, + "offset": { + "type": "integer" + }, + "limit": { + "type": "integer" + }, "include_legacy": { "type": "boolean" } @@ -5754,18 +5760,40 @@ "request_id": { "type": "string" }, + "turn_telemetry_ref": { + "type": "object", + "additionalProperties": true + }, "namespace": { "type": [ "string", "null" ] + }, + "user_concept_id": { + "type": [ + "string", + "null" + ] + }, + "organisation_concept_id": { + "type": [ + "string", + "null" + ] + }, + "offset": { + "type": "integer" + }, + "limit": { + "type": "integer" } }, "required": [ "request_id" ], "additionalProperties": true, - "description": "Get one full turn diagnostics payload by request_id." + "description": "Get one exact turn diagnostics payload by request_id. External stdio callers use a server-issued turn_telemetry_ref; large payloads return canonical JSON pages controlled by offset/limit." } }, { @@ -5777,6 +5805,10 @@ "request_id": { "type": "string" }, + "turn_telemetry_ref": { + "type": "object", + "additionalProperties": true + }, "namespace": { "type": [ "string", @@ -5789,6 +5821,18 @@ "null" ] }, + "organisation_concept_id": { + "type": [ + "string", + "null" + ] + }, + "session_id": { + "type": [ + "string", + "null" + ] + }, "window_session_id": { "type": [ "string", diff --git a/src/backend/server/routes/von_routes.py b/src/backend/server/routes/von_routes.py index 09a22f279..a129c14eb 100644 --- a/src/backend/server/routes/von_routes.py +++ b/src/backend/server/routes/von_routes.py @@ -3650,9 +3650,12 @@ def _build_turn_execution_mcp_access( namespace: str | None, history_owner_user_id: str | None, organisation_concept_id: str | None, + delegated_actor_user_id: str | None, + delegated_actor_namespace: str | None, ) -> dict[str, Any]: from ...services.conversation_scope_binding_service import ( build_conversation_scope_binding, + build_turn_telemetry_binding, ) def _tool_call_descriptor( @@ -3674,28 +3677,49 @@ def _tool_call_descriptor( access: dict[str, Any] = { "turn_execution_get_diagnostics": _tool_call_descriptor( "turn_execution_get_diagnostics", - {"request_id": request_id, "namespace": namespace}, + { + "request_id": request_id, + "turn_telemetry_ref": ( + build_turn_telemetry_binding( + request_id=request_id, + delegated_actor_user_id=delegated_actor_user_id, + permitted_tool="turn_execution_get_diagnostics", + chat_session_id=session_id, + history_owner_user_id=history_owner_user_id, + read_namespace=namespace, + delegated_actor_namespace=delegated_actor_namespace, + organisation_concept_id=organisation_concept_id, + ) + if delegated_actor_user_id + else None + ), + "namespace": namespace, + "user_concept_id": delegated_actor_user_id, + "organisation_concept_id": organisation_concept_id, + }, purpose="Fetch the persisted full turn-execution diagnostics payload.", ), - "turn_execution_get": _tool_call_descriptor( - "turn_execution_get", - {"request_id": request_id, "namespace": namespace}, - purpose="Fetch the projected turn-execution record for this request.", - ), } - if isinstance(session_id, str) and session_id.strip(): + if ( + isinstance(session_id, str) + and session_id.strip() + and delegated_actor_user_id + ): conversation_ref = build_conversation_scope_binding( chat_session_id=session_id, history_owner_user_id=history_owner_user_id, read_namespace=namespace, organisation_concept_id=organisation_concept_id, + delegated_actor_user_id=delegated_actor_user_id, + delegated_actor_namespace=delegated_actor_namespace, ) access["conversation_telemetry_get_locator"] = _tool_call_descriptor( "conversation_telemetry_get_locator", { "conversation_ref": conversation_ref, "namespace": namespace, + "user_concept_id": delegated_actor_user_id, "organisation_concept_id": organisation_concept_id, }, purpose="Fetch the compact conversation locator for the surrounding chat session.", @@ -3705,10 +3729,16 @@ def _tool_call_descriptor( { "conversation_ref": conversation_ref, "namespace": namespace, - "include_debug": True, + "user_concept_id": delegated_actor_user_id, + "include_debug": False, + "segment_size": 20, + "history_tail_limit": 100, "organisation_concept_id": organisation_concept_id, }, - purpose="Fetch the stored transcript segments and embedded debug payloads for this session.", + purpose=( + "Fetch a bounded transcript projection; use the exact debug-entry " + "descriptor for persisted debug payloads." + ), ) return access @@ -12347,6 +12377,8 @@ def _coerce_spoken_text(text: object) -> str | None: namespace=user_namespace, history_owner_user_id=history_user_id, organisation_concept_id=org_concept_id, + delegated_actor_user_id=user_concept_id, + delegated_actor_namespace=user_namespace, ) with turn_timing_recorder.span( stage_id="response_finalising", @@ -13253,6 +13285,268 @@ def history_telemetry_locator(): return jsonify({"error": str(e)}), 500 +@von_bp.route("/history/turn_telemetry_access", methods=["GET"]) +def history_turn_telemetry_access(): + """Issue fresh actor-bound MCP read delegations for one exact turn.""" + + from ...services.conversation_telemetry_locator_service import ( + build_turn_telemetry_mcp_access_locator, + ) + from ...services.turn_execution_diagnostics_service import ( + get_turn_execution_diagnostics_payload, + ) + from ...services.turn_execution_live_progress_service import ( + get_turn_execution_live_progress_payload, + ) + + try: + from ...security.access_control import get_effective_user_concept_id + + user_concept_id = get_effective_user_concept_id() + except Exception: + user_concept_id = session.get("user_concept_id") + user_concept_id = _normalise_concept_id(user_concept_id) + if not user_concept_id: + return jsonify({"error": "Not authenticated"}), 401 + + request_id = _progress_str(request.args.get("request_id")) + if not request_id: + return jsonify({"error": "request_id required"}), 400 + requested_session_id = _progress_str( + request.args.get("session_id") or session.get("session_id") + ) + requested_history_index = request.args.get("history_index", type=int) + + window_session_id = request.headers.get("X-Von-Window-Session") + effective = get_effective_context( + window_session_id, dict(session), user_concept_id + ) + actor_namespace, organisation_concept_id = _resolve_history_request_scope_hints( + user_concept_id=user_concept_id, + effective_context=effective, + ) + + owner_user_id = user_concept_id + owner_namespace = actor_namespace + shared_invite: Mapping[str, Any] | None = None + if requested_session_id: + resolved_owner, resolved_invite = _resolve_shared_conversation_owner( + user_concept_id=user_concept_id, + session_id=requested_session_id, + ) + if resolved_owner: + owner_user_id = resolved_owner + shared_invite = ( + resolved_invite if isinstance(resolved_invite, Mapping) else None + ) + elif not chat_history_service.has_chat_history_session( + user_concept_id, + requested_session_id, + namespace=actor_namespace, + ) and not chat_history_service.has_chat_history_session( + user_concept_id, + requested_session_id, + namespace=None, + ): + return jsonify({"error": "Not authorised for conversation"}), 403 + + shared_org_id = ( + _normalise_concept_id(shared_invite.get("organisation_concept_id")) + if shared_invite + else None + ) + organisation_concept_id = shared_org_id or organisation_concept_id + owner_namespace = ( + _derive_namespace_for_user_org( + owner_user_id, organisation_concept_id + ) + or chat_history_service.resolve_chat_history_namespace(owner_user_id) + ) + + diagnostics: Mapping[str, Any] | None = None + if requested_session_id and requested_history_index is not None: + try: + compact_debug = chat_history_service.get_chat_history_debug_entry( + user_id=owner_user_id, + session_id=requested_session_id, + history_index=requested_history_index, + namespace=owner_namespace or actor_namespace, + include_legacy=True, + hydrate_blob_refs=False, + ) + except Exception as exc: + current_app.logger.warning( + "Turn telemetry delegation exact history lookup failed: %s", + exc, + ) + compact_debug = None + compact_request_id = ( + _progress_str(compact_debug.get("request_id")) + if isinstance(compact_debug, Mapping) + else None + ) + if compact_request_id and compact_request_id != request_id: + return jsonify({"error": "request_id does not belong to history_index"}), 403 + if compact_request_id == request_id: + # The authenticated exact history location is sufficient to issue a + # read-only descriptor. Avoid hydrating the full diagnostics body + # merely to mint a short-lived locator for that same entry. + diagnostics = { + "success": True, + "request_id": request_id, + "chat_session_id": requested_session_id, + "history_location": { + "session_id": requested_session_id, + "history_index": requested_history_index, + }, + "derived_user_concept_id": owner_user_id, + "derived_organisation_concept_id": organisation_concept_id, + "namespace": owner_namespace or actor_namespace, + } + + if diagnostics is None: + try: + diagnostics = get_turn_execution_diagnostics_payload( + request_id=request_id, + namespace=owner_namespace or actor_namespace, + delegated_actor_user_id=user_concept_id, + delegated_actor_namespace=actor_namespace, + chat_session_id=requested_session_id, + history_index=requested_history_index, + history_owner_user_id=owner_user_id, + organisation_concept_id=organisation_concept_id, + ) + except Exception as exc: + current_app.logger.warning( + "Turn telemetry delegation diagnostics lookup failed: %s", + exc, + ) + diagnostics = None + diagnostics_session_id = ( + _progress_str(diagnostics.get("chat_session_id")) + if isinstance(diagnostics, Mapping) + else None + ) + diagnostics_owner_user_id = ( + _normalise_concept_id(diagnostics.get("derived_user_concept_id")) + if isinstance(diagnostics, Mapping) + else None + ) + diagnostics_namespace = ( + _progress_str(diagnostics.get("namespace")) + if isinstance(diagnostics, Mapping) + else None + ) + diagnostics_org_id = ( + _normalise_concept_id(diagnostics.get("derived_organisation_concept_id")) + if isinstance(diagnostics, Mapping) + else None + ) + diagnostics_history_location = ( + diagnostics.get("history_location") + if isinstance(diagnostics, Mapping) + and isinstance(diagnostics.get("history_location"), Mapping) + else {} + ) + diagnostics_history_index = diagnostics_history_location.get("history_index") + if not isinstance(diagnostics_history_index, int): + diagnostics_history_index = None + + if ( + requested_session_id + and diagnostics_session_id + and requested_session_id != diagnostics_session_id + ): + return jsonify({"error": "request_id does not belong to session_id"}), 403 + if ( + requested_history_index is not None + and diagnostics_history_index is not None + and requested_history_index != diagnostics_history_index + ): + return jsonify({"error": "request_id does not belong to history_index"}), 403 + + resolved_session_id = diagnostics_session_id or requested_session_id + resolved_history_index = ( + diagnostics_history_index + if diagnostics_history_index is not None + else requested_history_index + ) + if diagnostics_owner_user_id and diagnostics_owner_user_id != owner_user_id: + if not resolved_session_id: + return jsonify({"error": "Not authorised for turn"}), 403 + resolved_owner, resolved_invite = _resolve_shared_conversation_owner( + user_concept_id=user_concept_id, + session_id=resolved_session_id, + ) + if resolved_owner != diagnostics_owner_user_id: + return jsonify({"error": "Not authorised for turn"}), 403 + owner_user_id = diagnostics_owner_user_id + if isinstance(resolved_invite, Mapping): + organisation_concept_id = ( + _normalise_concept_id( + resolved_invite.get("organisation_concept_id") + ) + or organisation_concept_id + ) + elif diagnostics_owner_user_id: + owner_user_id = diagnostics_owner_user_id + + diagnostics_available = ( + isinstance(diagnostics, Mapping) and diagnostics.get("success") is True + ) + live_progress: Mapping[str, Any] | None = None + if not diagnostics_available: + live_progress = get_turn_execution_live_progress_payload( + request_id=request_id, + namespace=actor_namespace, + user_concept_id=user_concept_id, + window_session_id=window_session_id, + ) + if not isinstance(live_progress, Mapping) and not diagnostics_available: + return jsonify({"error": "Turn telemetry not found"}), 404 + if isinstance(live_progress, Mapping): + live_session_id = _progress_str( + live_progress.get("session_id") + or live_progress.get("chat_session_id") + or live_progress.get("conversation_session_id") + ) + if ( + requested_session_id + and live_session_id + and requested_session_id != live_session_id + ): + return ( + jsonify({"error": "request_id does not belong to session_id"}), + 403, + ) + if not diagnostics_available: + resolved_session_id = live_session_id or requested_session_id + owner_user_id = user_concept_id + owner_namespace = actor_namespace + + owner_namespace = ( + diagnostics_namespace + or owner_namespace + or _derive_namespace_for_user_org( + owner_user_id, diagnostics_org_id or organisation_concept_id + ) + ) + organisation_concept_id = diagnostics_org_id or organisation_concept_id + locator = build_turn_telemetry_mcp_access_locator( + request_id=request_id, + delegated_actor_user_id=user_concept_id, + delegated_actor_namespace=actor_namespace, + history_owner_user_id=owner_user_id, + read_namespace=owner_namespace, + organisation_concept_id=organisation_concept_id, + chat_session_id=resolved_session_id, + history_index=resolved_history_index, + include_diagnostics=diagnostics_available, + include_live_progress=isinstance(live_progress, Mapping), + ) + return jsonify(locator), 200 + + @von_bp.route("/diagnostics/export", methods=["POST"]) def export_diagnostics_snapshot(): """Write a sanitised diagnostics snapshot to data/diagnostic_latest.json. diff --git a/src/backend/services/chat_history_service.py b/src/backend/services/chat_history_service.py index 9b2bb72ca..7c1b991bc 100644 --- a/src/backend/services/chat_history_service.py +++ b/src/backend/services/chat_history_service.py @@ -1428,6 +1428,123 @@ def _hydrate_chat_history_entries(history: Any) -> List[Dict[str, Any]]: return _normalise_chat_history_entries(history, hydrate_blob_refs=True) +def get_chat_history_telemetry_locator_projection( + user_id: str, + session_id: str, + *, + namespace: Optional[str] = None, + include_legacy: bool = True, +) -> Optional[Dict[str, Any]]: + """Return only session and per-entry fields needed to build telemetry locators.""" + + if not user_id: + raise ChatHistoryServiceError("user_id is required.") + if not session_id: + raise ChatHistoryServiceError("session_id is required.") + + chat_history_coll = get_chat_history_collection_service(read_only=True) + if chat_history_coll is None: + raise ChatHistoryServiceError("Could not connect to chat history collection.") + + _guard_chat_history_read("get_chat_history_telemetry_locator_projection") + + compact_history_projection = { + "$map": { + "input": {"$ifNull": ["$history", []]}, + "as": "entry", + "in": { + "role": "$$entry.role", + "sender": "$$entry.sender", + "turn_id": "$$entry.turn_id", + "id": "$$entry.id", + "message_id": "$$entry.message_id", + "timestamp": "$$entry.timestamp", + "created_at": "$$entry.created_at", + "llm_debug_data": { + "$cond": [ + {"$eq": [{"$type": "$$entry.llm_debug_data"}, "object"]}, + { + "request_id": "$$entry.llm_debug_data.request_id", + "turn_id": "$$entry.llm_debug_data.turn_id", + "timestamp_utc": "$$entry.llm_debug_data.timestamp_utc", + }, + None, + ] + }, + }, + } + } + metadata_projection = { + "_id": 0, + "user_id": 1, + "session_id": 1, + "session_name": 1, + "namespace": 1, + "organisation_concept_id": 1, + "created_at": 1, + "updated_at": 1, + "history": compact_history_projection, + } + + try: + doc = None + for query in _build_chat_history_session_read_queries( + user_id=user_id, + session_id=session_id, + namespace=namespace, + include_legacy=include_legacy, + ): + if hasattr(chat_history_coll, "aggregate") and callable( + getattr(chat_history_coll, "aggregate") + ): + doc = next( + _read_aggregate( + chat_history_coll, + [ + {"$match": query}, + {"$limit": 1}, + {"$project": metadata_projection}, + ], + operation=( + "get_chat_history_telemetry_locator_projection.aggregate" + ), + ), + None, + ) + else: + # Test doubles and older collection adapters may not expose aggregate. + doc = _read_find_one( + chat_history_coll, + query, + { + key: value + for key, value in metadata_projection.items() + if key != "history" + } + | {"history": 1}, + operation=( + "get_chat_history_telemetry_locator_projection.find_fallback" + ), + ) + if doc is not None: + break + _record_chat_history_read_success() + return dict(doc) if isinstance(doc, dict) else None + except PyMongoError as exc: + _record_chat_history_read_failure( + "get_chat_history_telemetry_locator_projection", + exc, + ) + logger.error( + "Error retrieving compact telemetry locator projection: %s", + exc, + exc_info=True, + ) + raise ChatHistoryServiceError( + f"Could not retrieve telemetry locator projection: {exc}" + ) from exc + + def get_chat_history_segments( user_id: str, session_id: str, diff --git a/src/backend/services/conversation_scope_binding_service.py b/src/backend/services/conversation_scope_binding_service.py index 6110e9cb2..3ed1d7545 100644 --- a/src/backend/services/conversation_scope_binding_service.py +++ b/src/backend/services/conversation_scope_binding_service.py @@ -6,14 +6,21 @@ import hmac import json import os -from typing import Any, Mapping +import secrets +from datetime import datetime, timedelta, timezone +from typing import Any, Mapping, Sequence CONVERSATION_SCOPE_BINDING_SCHEMA_VERSION = "conversation_scope_binding.v1" HISTORY_LOCATION_BINDING_SCHEMA_VERSION = "history_location_binding.v1" +TURN_TELEMETRY_BINDING_SCHEMA_VERSION = "turn_telemetry_binding.v1" +TELEMETRY_READ_DELEGATION_AUDIENCE = "vontology_stdio_mcp" _CONVERSATION_SCOPE_KIND = "conversation_scope" _HISTORY_LOCATION_KIND = "history_location" +_TURN_TELEMETRY_KIND = "turn_telemetry" _DEFAULT_BINDING_SECRET = "von-dev-secret-key-change-in-production" +_DEFAULT_READ_DELEGATION_TTL_SECONDS = 15 * 60 +_MAX_READ_DELEGATION_TTL_SECONDS = 60 * 60 def _safe_str(value: Any) -> str | None: @@ -32,6 +39,82 @@ def _safe_int(value: Any) -> int | None: return None +def _safe_string_list(value: Any) -> list[str]: + if isinstance(value, str): + candidates: Sequence[Any] = (value,) + elif isinstance(value, Sequence) and not isinstance( + value, (bytes, bytearray, str) + ): + candidates = value + else: + return [] + return sorted( + { + cleaned + for item in candidates + if (cleaned := _safe_str(item)) is not None + } + ) + + +def _normalise_utc_datetime(value: datetime | None = None) -> datetime: + resolved = value or datetime.now(timezone.utc) + if resolved.tzinfo is None: + resolved = resolved.replace(tzinfo=timezone.utc) + return resolved.astimezone(timezone.utc) + + +def _utc_iso(value: datetime) -> str: + return _normalise_utc_datetime(value).isoformat().replace("+00:00", "Z") + + +def _parse_utc(value: Any) -> datetime | None: + cleaned = _safe_str(value) + if not cleaned: + return None + candidate = f"{cleaned[:-1]}+00:00" if cleaned.endswith("Z") else cleaned + try: + parsed = datetime.fromisoformat(candidate) + except Exception: + return None + return _normalise_utc_datetime(parsed) + + +def _read_delegation_fields( + *, + delegated_actor_user_id: str | None, + permitted_tools: Sequence[str], + audience: str, + ttl_seconds: int | None, + issued_at: datetime | None, + nonce: str | None, +) -> dict[str, Any]: + actor = _safe_str(delegated_actor_user_id) + tools = _safe_string_list(permitted_tools) + audience_value = _safe_str(audience) + if not actor: + raise ValueError("delegated_actor_user_id is required") + if not tools: + raise ValueError("at least one permitted telemetry read tool is required") + if not audience_value: + raise ValueError("audience is required") + + ttl = _safe_int(ttl_seconds) + if ttl is None: + ttl = _DEFAULT_READ_DELEGATION_TTL_SECONDS + ttl = min(_MAX_READ_DELEGATION_TTL_SECONDS, max(1, ttl)) + issued = _normalise_utc_datetime(issued_at) + expires = issued + timedelta(seconds=ttl) + return { + "delegated_actor_user_id": actor, + "audience": audience_value, + "permitted_tools": tools, + "issued_at_utc": _utc_iso(issued), + "expires_at_utc": _utc_iso(expires), + "nonce": _safe_str(nonce) or secrets.token_urlsafe(12), + } + + def _canonical_json(value: Mapping[str, Any]) -> str: return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True) @@ -170,12 +253,131 @@ def _verify_signed_binding( } +def _verify_read_delegation_claims( + *, + payload: Mapping[str, Any], + identifier_binding: Mapping[str, Any], + require_read_delegation: bool, + expected_audience: str | None, + expected_tool: str | None, + expected_actor_user_id: str | None, + now: datetime | None, +) -> dict[str, Any]: + audience = _safe_str(payload.get("audience")) + actor = _safe_str(payload.get("delegated_actor_user_id")) + permitted_tools = _safe_string_list(payload.get("permitted_tools")) + issued_at = _parse_utc(payload.get("issued_at_utc")) + expires_at = _parse_utc(payload.get("expires_at_utc")) + nonce = _safe_str(payload.get("nonce")) + has_complete_delegation = bool( + audience + and actor + and permitted_tools + and issued_at + and expires_at + and nonce + ) + + enriched_binding = { + **dict(identifier_binding), + "delegation_audience": audience, + "delegated_actor_user_id": actor, + "permitted_tools": permitted_tools, + "issued_at_utc": _utc_iso(issued_at) if issued_at else None, + "expires_at_utc": _utc_iso(expires_at) if expires_at else None, + } + if not has_complete_delegation: + if not require_read_delegation: + return { + "success": True, + "read_delegation": None, + "identifier_binding": enriched_binding, + } + enriched_binding["validation_status"] = "read_delegation_claims_missing" + return { + "success": False, + "error_code": "READ_DELEGATION_REQUIRED", + "error_message": ( + "The signed reference does not carry a complete external MCP " + "read delegation" + ), + "identifier_binding": enriched_binding, + } + + current_time = _normalise_utc_datetime(now) + if expires_at <= issued_at: + enriched_binding["validation_status"] = "invalid_delegation_window" + return { + "success": False, + "error_code": "INVALID_CONTEXT_BINDING", + "error_message": "Read delegation expiry is not after its issue time", + "identifier_binding": enriched_binding, + } + if current_time >= expires_at: + enriched_binding["validation_status"] = "read_delegation_expired" + return { + "success": False, + "error_code": "READ_DELEGATION_EXPIRED", + "error_message": "Read delegation has expired", + "identifier_binding": enriched_binding, + } + if expected_audience and audience != _safe_str(expected_audience): + enriched_binding["validation_status"] = "read_delegation_audience_mismatch" + return { + "success": False, + "error_code": "READ_DELEGATION_AUDIENCE_MISMATCH", + "error_message": "Read delegation audience does not match this MCP surface", + "identifier_binding": enriched_binding, + } + if expected_tool and _safe_str(expected_tool) not in permitted_tools: + enriched_binding["validation_status"] = "read_delegation_tool_mismatch" + return { + "success": False, + "error_code": "READ_DELEGATION_TOOL_MISMATCH", + "error_message": "Read delegation does not permit this MCP tool", + "identifier_binding": enriched_binding, + } + expected_actor = _safe_str(expected_actor_user_id) + if expected_actor and actor != expected_actor: + enriched_binding["validation_status"] = "read_delegation_actor_mismatch" + return { + "success": False, + "error_code": "READ_DELEGATION_ACTOR_MISMATCH", + "error_message": "Read delegation is bound to a different actor", + "identifier_binding": enriched_binding, + } + + enriched_binding["read_delegation_validation_status"] = "verified" + return { + "success": True, + "read_delegation": { + "binding_id": enriched_binding.get("binding_id"), + "audience": audience, + "delegated_actor_user_id": actor, + "permitted_tools": permitted_tools, + "issued_at_utc": _utc_iso(issued_at), + "expires_at_utc": _utc_iso(expires_at), + }, + "identifier_binding": enriched_binding, + } + + def build_conversation_scope_binding( *, chat_session_id: str, history_owner_user_id: str | None = None, read_namespace: str | None = None, organisation_concept_id: str | None = None, + delegated_actor_user_id: str | None = None, + delegated_actor_namespace: str | None = None, + permitted_tools: Sequence[str] = ( + "conversation_telemetry_get_locator", + "chat_history_get_segments", + ), + audience: str = TELEMETRY_READ_DELEGATION_AUDIENCE, + ttl_seconds: int | None = None, + issued_at: datetime | None = None, + nonce: str | None = None, ) -> dict[str, Any]: chat_session_id_value = _safe_str(chat_session_id) if not chat_session_id_value: @@ -189,11 +391,33 @@ def build_conversation_scope_binding( "history_owner_user_id": _safe_str(history_owner_user_id), "read_namespace": _safe_str(read_namespace), "organisation_concept_id": _safe_str(organisation_concept_id), + "delegated_actor_namespace": ( + _safe_str(delegated_actor_namespace) or _safe_str(read_namespace) + ), + **_read_delegation_fields( + delegated_actor_user_id=( + _safe_str(delegated_actor_user_id) + or _safe_str(history_owner_user_id) + ), + permitted_tools=permitted_tools, + audience=audience, + ttl_seconds=ttl_seconds, + issued_at=issued_at, + nonce=nonce, + ), }, ) -def verify_conversation_scope_binding(binding: Any) -> dict[str, Any]: +def verify_conversation_scope_binding( + binding: Any, + *, + require_read_delegation: bool = False, + expected_audience: str | None = None, + expected_tool: str | None = None, + expected_actor_user_id: str | None = None, + now: datetime | None = None, +) -> dict[str, Any]: verified = _verify_signed_binding( binding, schema_version=CONVERSATION_SCOPE_BINDING_SCHEMA_VERSION, @@ -214,7 +438,19 @@ def verify_conversation_scope_binding(binding: Any) -> dict[str, Any]: "identifier_binding": identifier_binding, } - identifier_binding = dict(verified["identifier_binding"]) + delegation = _verify_read_delegation_claims( + payload=payload, + identifier_binding=verified["identifier_binding"], + require_read_delegation=require_read_delegation, + expected_audience=expected_audience, + expected_tool=expected_tool, + expected_actor_user_id=expected_actor_user_id, + now=now, + ) + if not delegation.get("success"): + return delegation + + identifier_binding = dict(delegation["identifier_binding"]) identifier_binding["chat_session_id_source"] = "conversation_ref" return { "success": True, @@ -222,6 +458,13 @@ def verify_conversation_scope_binding(binding: Any) -> dict[str, Any]: "history_owner_user_id": _safe_str(payload.get("history_owner_user_id")), "read_namespace": _safe_str(payload.get("read_namespace")), "organisation_concept_id": _safe_str(payload.get("organisation_concept_id")), + "delegated_actor_user_id": _safe_str( + payload.get("delegated_actor_user_id") + ), + "delegated_actor_namespace": _safe_str( + payload.get("delegated_actor_namespace") + ), + "read_delegation": delegation.get("read_delegation"), "identifier_binding": identifier_binding, } @@ -233,6 +476,13 @@ def build_history_location_binding( history_owner_user_id: str | None = None, read_namespace: str | None = None, organisation_concept_id: str | None = None, + delegated_actor_user_id: str | None = None, + delegated_actor_namespace: str | None = None, + permitted_tools: Sequence[str] = ("chat_history_get_debug_entry",), + audience: str = TELEMETRY_READ_DELEGATION_AUDIENCE, + ttl_seconds: int | None = None, + issued_at: datetime | None = None, + nonce: str | None = None, ) -> dict[str, Any]: chat_session_id_value = _safe_str(chat_session_id) history_index_value = _safe_int(history_index) @@ -250,11 +500,33 @@ def build_history_location_binding( "history_owner_user_id": _safe_str(history_owner_user_id), "read_namespace": _safe_str(read_namespace), "organisation_concept_id": _safe_str(organisation_concept_id), + "delegated_actor_namespace": ( + _safe_str(delegated_actor_namespace) or _safe_str(read_namespace) + ), + **_read_delegation_fields( + delegated_actor_user_id=( + _safe_str(delegated_actor_user_id) + or _safe_str(history_owner_user_id) + ), + permitted_tools=permitted_tools, + audience=audience, + ttl_seconds=ttl_seconds, + issued_at=issued_at, + nonce=nonce, + ), }, ) -def verify_history_location_binding(binding: Any) -> dict[str, Any]: +def verify_history_location_binding( + binding: Any, + *, + require_read_delegation: bool = False, + expected_audience: str | None = None, + expected_tool: str | None = None, + expected_actor_user_id: str | None = None, + now: datetime | None = None, +) -> dict[str, Any]: verified = _verify_signed_binding( binding, schema_version=HISTORY_LOCATION_BINDING_SCHEMA_VERSION, @@ -278,7 +550,19 @@ def verify_history_location_binding(binding: Any) -> dict[str, Any]: "identifier_binding": identifier_binding, } - identifier_binding = dict(verified["identifier_binding"]) + delegation = _verify_read_delegation_claims( + payload=payload, + identifier_binding=verified["identifier_binding"], + require_read_delegation=require_read_delegation, + expected_audience=expected_audience, + expected_tool=expected_tool, + expected_actor_user_id=expected_actor_user_id, + now=now, + ) + if not delegation.get("success"): + return delegation + + identifier_binding = dict(delegation["identifier_binding"]) identifier_binding["chat_session_id_source"] = "history_location_ref" identifier_binding["history_index_source"] = "history_location_ref" return { @@ -288,5 +572,129 @@ def verify_history_location_binding(binding: Any) -> dict[str, Any]: "history_owner_user_id": _safe_str(payload.get("history_owner_user_id")), "read_namespace": _safe_str(payload.get("read_namespace")), "organisation_concept_id": _safe_str(payload.get("organisation_concept_id")), + "delegated_actor_user_id": _safe_str( + payload.get("delegated_actor_user_id") + ), + "delegated_actor_namespace": _safe_str( + payload.get("delegated_actor_namespace") + ), + "read_delegation": delegation.get("read_delegation"), "identifier_binding": identifier_binding, } + + +def build_turn_telemetry_binding( + *, + request_id: str, + delegated_actor_user_id: str, + permitted_tool: str, + chat_session_id: str | None = None, + history_index: int | None = None, + history_owner_user_id: str | None = None, + read_namespace: str | None = None, + delegated_actor_namespace: str | None = None, + organisation_concept_id: str | None = None, + audience: str = TELEMETRY_READ_DELEGATION_AUDIENCE, + ttl_seconds: int | None = None, + issued_at: datetime | None = None, + nonce: str | None = None, +) -> dict[str, Any]: + request_id_value = _safe_str(request_id) + tool_value = _safe_str(permitted_tool) + if not request_id_value: + raise ValueError("request_id is required") + if not tool_value: + raise ValueError("permitted_tool is required") + history_index_value = ( + _safe_int(history_index) if history_index is not None else None + ) + if history_index_value is not None and history_index_value < 0: + raise ValueError("history_index must be a non-negative integer") + + return _build_signed_binding( + schema_version=TURN_TELEMETRY_BINDING_SCHEMA_VERSION, + binding_kind=_TURN_TELEMETRY_KIND, + fields={ + "request_id": request_id_value, + "chat_session_id": _safe_str(chat_session_id), + "history_index": history_index_value, + "history_owner_user_id": _safe_str(history_owner_user_id), + "read_namespace": _safe_str(read_namespace), + "delegated_actor_namespace": ( + _safe_str(delegated_actor_namespace) or _safe_str(read_namespace) + ), + "organisation_concept_id": _safe_str(organisation_concept_id), + **_read_delegation_fields( + delegated_actor_user_id=delegated_actor_user_id, + permitted_tools=(tool_value,), + audience=audience, + ttl_seconds=ttl_seconds, + issued_at=issued_at, + nonce=nonce, + ), + }, + ) + + +def verify_turn_telemetry_binding( + binding: Any, + *, + expected_tool: str, + expected_audience: str = TELEMETRY_READ_DELEGATION_AUDIENCE, + expected_actor_user_id: str | None = None, + now: datetime | None = None, +) -> dict[str, Any]: + verified = _verify_signed_binding( + binding, + schema_version=TURN_TELEMETRY_BINDING_SCHEMA_VERSION, + binding_kind=_TURN_TELEMETRY_KIND, + ) + if not verified.get("success"): + return verified + + payload = dict(verified["payload"]) + request_id = _safe_str(payload.get("request_id")) + if not request_id: + identifier_binding = dict(verified["identifier_binding"]) + identifier_binding["validation_status"] = "missing_request_id" + return { + "success": False, + "error_code": "INVALID_CONTEXT_BINDING", + "error_message": "Bound turn telemetry reference is missing request_id", + "identifier_binding": identifier_binding, + } + + delegation = _verify_read_delegation_claims( + payload=payload, + identifier_binding=verified["identifier_binding"], + require_read_delegation=True, + expected_audience=expected_audience, + expected_tool=expected_tool, + expected_actor_user_id=expected_actor_user_id, + now=now, + ) + if not delegation.get("success"): + return delegation + + history_index = ( + _safe_int(payload.get("history_index")) + if payload.get("history_index") is not None + else None + ) + return { + "success": True, + "request_id": request_id, + "chat_session_id": _safe_str(payload.get("chat_session_id")), + "history_index": history_index, + "history_owner_user_id": _safe_str(payload.get("history_owner_user_id")), + "read_namespace": _safe_str(payload.get("read_namespace")), + "organisation_concept_id": _safe_str(payload.get("organisation_concept_id")), + "delegated_actor_user_id": _safe_str( + payload.get("delegated_actor_user_id") + ), + "delegated_actor_namespace": _safe_str( + payload.get("delegated_actor_namespace") + ), + "read_delegation": delegation.get("read_delegation"), + "identifier_binding": delegation["identifier_binding"], + } diff --git a/src/backend/services/conversation_telemetry_locator_service.py b/src/backend/services/conversation_telemetry_locator_service.py index 2e04747cc..3944f1c43 100644 --- a/src/backend/services/conversation_telemetry_locator_service.py +++ b/src/backend/services/conversation_telemetry_locator_service.py @@ -9,11 +9,13 @@ from .conversation_scope_binding_service import ( build_conversation_scope_binding, build_history_location_binding, + build_turn_telemetry_binding, ) CONVERSATION_LLM_TELEMETRY_LOCATOR_SCHEMA_VERSION = ( "conversation_llm_telemetry_locator.v1" ) +TURN_TELEMETRY_MCP_ACCESS_SCHEMA_VERSION = "turn_telemetry_mcp_access.v1" def _safe_str(value: Any) -> str | None: @@ -103,6 +105,138 @@ def _build_namespace_context( } +def build_turn_telemetry_mcp_access_locator( + *, + request_id: str, + delegated_actor_user_id: str, + delegated_actor_namespace: str | None, + history_owner_user_id: str, + read_namespace: str | None, + organisation_concept_id: str | None, + chat_session_id: str | None = None, + history_index: int | None = None, + include_diagnostics: bool = True, + include_live_progress: bool = True, +) -> dict[str, Any]: + """Issue fresh exact-tool read delegations for one authenticated turn.""" + + request_id_value = _safe_str(request_id) + actor_user_id = _safe_str(delegated_actor_user_id) + owner_user_id = _safe_str(history_owner_user_id) + session_id_value = _safe_str(chat_session_id) + history_index_value = ( + _safe_int(history_index) if history_index is not None else None + ) + if not request_id_value: + raise ValueError("request_id is required") + if not actor_user_id: + raise ValueError("delegated_actor_user_id is required") + if not owner_user_id: + raise ValueError("history_owner_user_id is required") + if history_index_value is not None and history_index_value < 0: + raise ValueError("history_index must be a non-negative integer") + + def turn_descriptor(tool_name: str, purpose: str) -> dict[str, Any]: + return _build_tool_call_descriptor( + tool_name, + { + "request_id": request_id_value, + "turn_telemetry_ref": build_turn_telemetry_binding( + request_id=request_id_value, + delegated_actor_user_id=actor_user_id, + permitted_tool=tool_name, + chat_session_id=session_id_value, + history_index=history_index_value, + history_owner_user_id=owner_user_id, + read_namespace=read_namespace, + delegated_actor_namespace=delegated_actor_namespace, + organisation_concept_id=organisation_concept_id, + ), + "namespace": delegated_actor_namespace, + "user_concept_id": actor_user_id, + "organisation_concept_id": organisation_concept_id, + }, + purpose=purpose, + ) + + mcp_access: dict[str, Any] = {} + if include_diagnostics: + mcp_access["turn_execution_get_diagnostics"] = turn_descriptor( + "turn_execution_get_diagnostics", + "Fetch the exact persisted diagnostics for this turn.", + ) + if include_live_progress: + mcp_access["turn_execution_get_live_progress"] = turn_descriptor( + "turn_execution_get_live_progress", + ( + "Fetch the bounded live progress snapshot; pass section, limit, " + "and offset only for explicit detail hydration." + ), + ) + if session_id_value: + conversation_ref = build_conversation_scope_binding( + chat_session_id=session_id_value, + history_owner_user_id=owner_user_id, + read_namespace=read_namespace, + organisation_concept_id=organisation_concept_id, + delegated_actor_user_id=actor_user_id, + delegated_actor_namespace=delegated_actor_namespace, + ) + mcp_access["conversation_telemetry_get_locator"] = ( + _build_tool_call_descriptor( + "conversation_telemetry_get_locator", + { + "conversation_ref": conversation_ref, + "namespace": delegated_actor_namespace, + "user_concept_id": actor_user_id, + "organisation_concept_id": organisation_concept_id, + }, + purpose="Fetch a fresh compact locator for the surrounding conversation.", + ) + ) + if session_id_value and history_index_value is not None: + mcp_access["chat_history_get_debug_entry"] = _build_tool_call_descriptor( + "chat_history_get_debug_entry", + { + "history_location_ref": build_history_location_binding( + chat_session_id=session_id_value, + history_index=history_index_value, + history_owner_user_id=owner_user_id, + read_namespace=read_namespace, + organisation_concept_id=organisation_concept_id, + delegated_actor_user_id=actor_user_id, + delegated_actor_namespace=delegated_actor_namespace, + ), + "namespace": delegated_actor_namespace, + "user_concept_id": actor_user_id, + "organisation_concept_id": organisation_concept_id, + }, + purpose="Fetch the exact bounded stored llm_debug_data for this turn.", + ) + + return { + "schema_version": TURN_TELEMETRY_MCP_ACCESS_SCHEMA_VERSION, + "generated_at_utc": _now_utc_iso(), + "request_id": request_id_value, + "chat_session_id": session_id_value, + "history_location": ( + { + "session_id": session_id_value, + "history_index": history_index_value, + } + if session_id_value and history_index_value is not None + else None + ), + "namespace_context": _build_namespace_context( + namespace=delegated_actor_namespace, + user_id=actor_user_id, + organisation_concept_id=organisation_concept_id, + ), + "history_owner_user_id": owner_user_id, + "mcp_access": mcp_access, + } + + def build_conversation_llm_telemetry_locator( *, user_id: str, @@ -122,16 +256,20 @@ def build_conversation_llm_telemetry_locator( if not user_id_value: raise chat_history_service.ChatHistoryServiceError("user_id is required.") - session_summary = chat_history_service.get_chat_history_session_summary( - user_id_value, - session_id_value, - namespace=namespace, - include_legacy=include_legacy, - summary_mode="light", - ) or {} - resolved_namespace = _safe_str(namespace) or _safe_str(session_summary.get("namespace")) + session_projection = ( + chat_history_service.get_chat_history_telemetry_locator_projection( + user_id_value, + session_id_value, + namespace=namespace, + include_legacy=include_legacy, + ) + or {} + ) + resolved_namespace = _safe_str(namespace) or _safe_str( + session_projection.get("namespace") + ) resolved_org_id = _safe_str(organisation_concept_id) or _safe_str( - session_summary.get("organisation_concept_id") + session_projection.get("organisation_concept_id") ) requested_user_id_value = _safe_str(requested_user_id) or user_id_value requested_namespace_value = ( @@ -142,6 +280,8 @@ def build_conversation_llm_telemetry_locator( history_owner_user_id=user_id_value, read_namespace=resolved_namespace, organisation_concept_id=resolved_org_id, + delegated_actor_user_id=requested_user_id_value, + delegated_actor_namespace=requested_namespace_value, ) def _build_bound_conversation_args(*, include_debug: bool | None = None) -> dict[str, Any]: @@ -155,12 +295,7 @@ def _build_bound_conversation_args(*, include_debug: bool | None = None) -> dict arguments["include_debug"] = include_debug return arguments - history = chat_history_service.get_chat_history( - user_id_value, - session_id_value, - namespace=resolved_namespace, - include_legacy=include_legacy, - ) + history = session_projection.get("history") if not isinstance(history, list): history = [] @@ -223,6 +358,8 @@ def _build_bound_conversation_args(*, include_debug: bool | None = None) -> dict history_owner_user_id=user_id_value, read_namespace=resolved_namespace, organisation_concept_id=resolved_org_id, + delegated_actor_user_id=requested_user_id_value, + delegated_actor_namespace=requested_namespace_value, ), "namespace": requested_namespace_value, "user_concept_id": requested_user_id_value, @@ -239,7 +376,20 @@ def _build_bound_conversation_args(*, include_debug: bool | None = None) -> dict "turn_execution_get_diagnostics", { "request_id": request_id, - "namespace": resolved_namespace, + "turn_telemetry_ref": build_turn_telemetry_binding( + request_id=request_id, + delegated_actor_user_id=requested_user_id_value, + permitted_tool="turn_execution_get_diagnostics", + chat_session_id=session_id_value, + history_index=history_index, + history_owner_user_id=user_id_value, + read_namespace=resolved_namespace, + delegated_actor_namespace=requested_namespace_value, + organisation_concept_id=resolved_org_id, + ), + "namespace": requested_namespace_value, + "user_concept_id": requested_user_id_value, + "organisation_concept_id": resolved_org_id, }, purpose=( "Fetch the persisted full turn-execution diagnostics payload for this assistant turn." @@ -280,7 +430,7 @@ def _build_bound_conversation_args(*, include_debug: bool | None = None) -> dict "schema_version": CONVERSATION_LLM_TELEMETRY_LOCATOR_SCHEMA_VERSION, "generated_at_utc": _now_utc_iso(), "session_id": session_id_value, - "session_name": _safe_str(session_summary.get("session_name")), + "session_name": _safe_str(session_projection.get("session_name")), "namespace_context": namespace_context, "metadata": metadata, "mcp_access": { @@ -293,16 +443,15 @@ def _build_bound_conversation_args(*, include_debug: bool | None = None) -> dict ), "chat_history_get_segments": _build_tool_call_descriptor( "chat_history_get_segments", - _build_bound_conversation_args(include_debug=True), - purpose="Fetch the stored conversation transcript segments and embedded debug payloads.", - ), - "turn_execution_list": _build_tool_call_descriptor( - "turn_execution_list", { - "session_id": session_id_value, - "namespace": resolved_namespace, + **_build_bound_conversation_args(include_debug=False), + "segment_size": 20, + "history_tail_limit": 100, }, - purpose="List turn-execution projections for this conversation.", + purpose=( + "Fetch a bounded transcript projection; use each turn's exact " + "debug-entry descriptor for persisted debug payloads." + ), ), }, "turns": turns, diff --git a/src/backend/services/turn_execution_diagnostics_service.py b/src/backend/services/turn_execution_diagnostics_service.py index 6ed9fc145..c0c4f9820 100644 --- a/src/backend/services/turn_execution_diagnostics_service.py +++ b/src/backend/services/turn_execution_diagnostics_service.py @@ -7,10 +7,15 @@ from datetime import datetime, timezone from typing import Any, Mapping, Sequence -from .chat_history_service import get_chat_history_collection_service +from .chat_history_service import ( + ChatHistoryServiceError, + get_chat_history_collection_service, + get_chat_history_debug_entry, +) from .conversation_scope_binding_service import ( build_conversation_scope_binding, build_history_location_binding, + build_turn_telemetry_binding, ) from .debug_payload_store import hydrate_debug_payload_blob_refs from .turn_decision_attribution_service import build_turn_decision_attribution @@ -215,6 +220,58 @@ def _resolve_history_context( } +def _resolve_exact_history_context( + *, + request_id: str, + namespace: str | None, + chat_session_id: str, + history_index: int, + history_owner_user_id: str, + organisation_concept_id: str | None, +) -> dict[str, Any] | None: + """Resolve one signed history location without a request-id-wide scan.""" + + try: + llm_debug = get_chat_history_debug_entry( + user_id=history_owner_user_id, + session_id=chat_session_id, + history_index=history_index, + namespace=namespace, + include_legacy=True, + hydrate_blob_refs=True, + ) + except ChatHistoryServiceError as exc: + raise TurnExecutionDiagnosticsServiceError( + "Could not query the exact delegated chat-history entry " + f"for request_id={request_id}: {exc}" + ) from exc + + if not isinstance(llm_debug, Mapping): + return None + if _safe_str(llm_debug.get("request_id")) != request_id: + return None + + hydrated = hydrate_debug_payload_blob_refs(llm_debug, fail_soft=True) + target_llm_debug = ( + dict(hydrated.payload) + if isinstance(hydrated.payload, Mapping) + else dict(llm_debug) + ) + return { + "user_id": history_owner_user_id, + "session_id": chat_session_id, + "namespace": namespace, + "org_id": organisation_concept_id, + "target_index": history_index, + "target_message": { + "role": "assistant", + "llm_debug_data": target_llm_debug, + }, + "target_llm_debug": target_llm_debug, + "prompt_text": _extract_prompt_text_from_debug(target_llm_debug), + } + + def _load_turn_execution_record( *, request_id: str, @@ -437,38 +494,53 @@ def _build_turn_diagnostics_mcp_access( history_index: int | None, history_owner_user_id: str | None, organisation_concept_id: str | None, + delegated_actor_user_id: str | None, + delegated_actor_namespace: str | None, payload: Mapping[str, Any] | None, ) -> dict[str, Any]: + delegated_actor = delegated_actor_user_id or history_owner_user_id access: dict[str, Any] = { "turn_execution_get_diagnostics": _build_tool_call_descriptor( "turn_execution_get_diagnostics", { "request_id": request_id, + "turn_telemetry_ref": ( + build_turn_telemetry_binding( + request_id=request_id, + delegated_actor_user_id=delegated_actor, + permitted_tool="turn_execution_get_diagnostics", + chat_session_id=session_id, + history_index=history_index, + history_owner_user_id=history_owner_user_id, + read_namespace=namespace, + delegated_actor_namespace=delegated_actor_namespace, + organisation_concept_id=organisation_concept_id, + ) + if delegated_actor + else None + ), "namespace": namespace, + "user_concept_id": delegated_actor, + "organisation_concept_id": organisation_concept_id, }, purpose="Fetch the persisted full turn-execution diagnostics payload.", ), - "turn_execution_get": _build_tool_call_descriptor( - "turn_execution_get", - { - "request_id": request_id, - "namespace": namespace, - }, - purpose="Fetch the projected turn-execution record for this request.", - ), } - if session_id: + if session_id and delegated_actor: conversation_ref = build_conversation_scope_binding( chat_session_id=session_id, history_owner_user_id=history_owner_user_id, read_namespace=namespace, organisation_concept_id=organisation_concept_id, + delegated_actor_user_id=delegated_actor, + delegated_actor_namespace=delegated_actor_namespace, ) access["conversation_telemetry_get_locator"] = _build_tool_call_descriptor( "conversation_telemetry_get_locator", { "conversation_ref": conversation_ref, "namespace": namespace, + "user_concept_id": delegated_actor, "organisation_concept_id": organisation_concept_id, }, purpose="Fetch the compact conversation locator for the surrounding chat session.", @@ -478,12 +550,18 @@ def _build_turn_diagnostics_mcp_access( { "conversation_ref": conversation_ref, "namespace": namespace, - "include_debug": True, + "user_concept_id": delegated_actor, + "include_debug": False, + "segment_size": 20, + "history_tail_limit": 100, "organisation_concept_id": organisation_concept_id, }, - purpose="Fetch the stored transcript segments and embedded debug payloads for this session.", + purpose=( + "Fetch a bounded transcript projection; use the exact debug-entry " + "descriptor for persisted debug payloads." + ), ) - if session_id and history_index is not None: + if session_id and history_index is not None and delegated_actor: access["chat_history_get_debug_entry"] = _build_tool_call_descriptor( "chat_history_get_debug_entry", { @@ -493,27 +571,31 @@ def _build_turn_diagnostics_mcp_access( history_owner_user_id=history_owner_user_id, read_namespace=namespace, organisation_concept_id=organisation_concept_id, + delegated_actor_user_id=delegated_actor, + delegated_actor_namespace=delegated_actor_namespace, ), "namespace": namespace, + "user_concept_id": delegated_actor, "organisation_concept_id": organisation_concept_id, }, purpose="Fetch the exact stored llm_debug_data entry for this assistant turn.", ) - workflow_traces = _extract_workflow_execution_trace_refs(payload) - if workflow_traces: - access["workflow_execution_traces"] = workflow_traces - primary_trace = next( - ( - trace - for trace in workflow_traces - if trace.get("trace_role") == "selected_workflow" - ), - workflow_traces[0], - ) - primary_access = primary_trace.get("mcp_access") - if isinstance(primary_access, Mapping): - access["workflow_get_execution_trace"] = dict(primary_access) + if delegated_actor_user_id is None: + workflow_traces = _extract_workflow_execution_trace_refs(payload) + if workflow_traces: + access["workflow_execution_traces"] = workflow_traces + primary_trace = next( + ( + trace + for trace in workflow_traces + if trace.get("trace_role") == "selected_workflow" + ), + workflow_traces[0], + ) + primary_access = primary_trace.get("mcp_access") + if isinstance(primary_access, Mapping): + access["workflow_get_execution_trace"] = dict(primary_access) return access @@ -1533,15 +1615,39 @@ def get_turn_execution_diagnostics_payload( *, request_id: str, namespace: str | None = None, + delegated_actor_user_id: str | None = None, + delegated_actor_namespace: str | None = None, + chat_session_id: str | None = None, + history_index: int | None = None, + history_owner_user_id: str | None = None, + organisation_concept_id: str | None = None, ) -> dict[str, Any] | None: request_id_value = _safe_str(request_id) if not request_id_value: return None - history_context = _resolve_history_context( - request_id=request_id_value, - namespace=_safe_str(namespace), - ) + exact_session_id = _safe_str(chat_session_id) + exact_owner_user_id = _safe_str(history_owner_user_id) + if ( + exact_session_id + and exact_owner_user_id + and isinstance(history_index, int) + and not isinstance(history_index, bool) + and history_index >= 0 + ): + history_context = _resolve_exact_history_context( + request_id=request_id_value, + namespace=_safe_str(namespace), + chat_session_id=exact_session_id, + history_index=history_index, + history_owner_user_id=exact_owner_user_id, + organisation_concept_id=_safe_str(organisation_concept_id), + ) + else: + history_context = _resolve_history_context( + request_id=request_id_value, + namespace=_safe_str(namespace), + ) turn_record = _load_turn_execution_record( request_id=request_id_value, namespace=_safe_str(namespace) @@ -1677,6 +1783,8 @@ def get_turn_execution_diagnostics_payload( organisation_concept_id=_safe_str( payload.get("derived_organisation_concept_id") ), + delegated_actor_user_id=_safe_str(delegated_actor_user_id), + delegated_actor_namespace=_safe_str(delegated_actor_namespace), payload=payload, ) payload["decision_attribution"] = build_turn_decision_attribution( diff --git a/src/frontend/web/von_interface/static/js/chatTab.js b/src/frontend/web/von_interface/static/js/chatTab.js index aa53bdc44..7ccb0c25f 100644 --- a/src/frontend/web/von_interface/static/js/chatTab.js +++ b/src/frontend/web/von_interface/static/js/chatTab.js @@ -112,6 +112,7 @@ const CONVERSATION_TELEMETRY_ACCESS_SCHEMA_VERSION = 'conversation_telemetry_acc const CONVERSATION_LLM_TELEMETRY_SCHEMA_VERSION = 'conversation_llm_telemetry.v1'; const CONVERSATION_LLM_TELEMETRY_LOCATOR_SCHEMA_VERSION = 'conversation_llm_telemetry_locator.v1'; const CONVERSATION_TELEMETRY_LOCATOR_FETCH_TIMEOUT_MS = 4000; +const TURN_TELEMETRY_MCP_ACCESS_SCHEMA_VERSION = 'turn_telemetry_mcp_access.v1'; const TURN_TELEMETRY_LOCATOR_SCHEMA_VERSION = 'turn_telemetry_locator.v1'; const TURN_LIVE_PROGRESS_LOCATOR_SCHEMA_VERSION = 'turn_live_progress_locator.v1'; const WORKFLOW_USE_EPISODES_LOCATOR_SCHEMA_VERSION = 'workflow_use_episodes_locator.v1'; @@ -29956,7 +29957,23 @@ async function copyActiveThinkingDiagnostics(button = null, requestOverride = nu if (!request) { return false; } - const payload = buildThinkingDiagnosticsLocatorPayload(request); + const requestId = ( + typeof request.clientRequestId === 'string' + ? request.clientRequestId.trim() + : '' + ) || ( + typeof request.latestProgress?.request_id === 'string' + ? request.latestProgress.request_id.trim() + : '' + ); + const turnAccess = await fetchTurnTelemetryMcpAccess({ + requestId, + sessionId: activeChatSessionId || request.sessionId || null, + includeLiveProgress: true + }); + const payload = buildThinkingDiagnosticsLocatorPayload(request, { + mcpAccess: turnAccess?.mcp_access || null + }); if (!payload) { showToast('No diagnostic reference is available yet.', 'info'); return false; @@ -32711,7 +32728,29 @@ async function buildLlmDebugClipboardJsonForTurn(turnId) { return null; } - const debugData = await hydrateTurnHistoryLocationFromConversationLocator(turnId, debugDataRaw); + let debugData = await hydrateTurnHistoryLocationFromConversationLocator(turnId, debugDataRaw); + const requestId = resolveConversationTelemetryRequestId(debugData); + const historyLocation = cloneConversationHistoryLocation( + debugData?.turn_execution_diagnostics?.history_location + || debugData?.history_location + ); + const turnAccess = await fetchTurnTelemetryMcpAccess({ + requestId, + sessionId: historyLocation?.session_id || activeChatSessionId || null, + historyIndex: Number.isInteger(historyLocation?.history_index) + ? historyLocation.history_index + : null, + includeLiveProgress: false + }); + if (turnAccess?.mcp_access && typeof turnAccess.mcp_access === 'object') { + debugData = { + ...debugData, + telemetry_locator_mcp_access: turnAccess.mcp_access, + history_location: cloneConversationHistoryLocation( + turnAccess.history_location || historyLocation + ) + }; + } const metadata = buildLlmDebugMetadata(debugData); const workflowExecutionTelemetry = _getWorkflowExecutionTelemetry(debugData); const copyPayload = buildLlmDebugLocatorPayload({ @@ -33132,51 +33171,7 @@ function buildChatHistoryAccessArgs({ sessionId, historyIndex = undefined } = {} }; } -function buildConversationTelemetrySessionMcpAccess( - sessionId, - namespaceContext = buildConversationTelemetryNamespaceContext() -) { - const cleanSessionId = typeof sessionId === 'string' && sessionId.trim() - ? sessionId.trim() - : null; - if (!cleanSessionId) { - return {}; - } - - return { - conversation_telemetry_get_locator: buildMcpToolAccess( - 'conversation_telemetry_get_locator', - { - session_id: cleanSessionId, - namespace: namespaceContext.namespace || null, - user_concept_id: namespaceContext.user_id || null, - organisation_concept_id: namespaceContext.org_id || null - }, - 'Fetch the authoritative server-side conversation telemetry locator.' - ), - chat_history_get_segments: buildMcpToolAccess( - 'chat_history_get_segments', - { - session_id: cleanSessionId, - namespace: namespaceContext.namespace || null, - user_concept_id: namespaceContext.user_id || null, - organisation_concept_id: namespaceContext.org_id || null, - include_debug: true - }, - 'Fetch the stored transcript segments and embedded debug payloads for this session.' - ), - turn_execution_list: buildMcpToolAccess( - 'turn_execution_list', - { - session_id: cleanSessionId, - namespace: namespaceContext.namespace || null - }, - 'List turn-execution records for this conversation.' - ) - }; -} - -function buildThinkingDiagnosticsLocatorPayload(request) { +function buildThinkingDiagnosticsLocatorPayload(request, { mcpAccess = null } = {}) { if (!request || typeof request !== 'object') { return null; } @@ -33190,6 +33185,23 @@ function buildThinkingDiagnosticsLocatorPayload(request) { if (!requestId) { return null; } + const executableMcpAccess = ( + mcpAccess + && typeof mcpAccess === 'object' + ) + ? Object.fromEntries( + Object.entries(mcpAccess).filter(([toolName, descriptor]) => ( + [ + 'turn_execution_get_live_progress', + 'turn_execution_get_diagnostics', + 'conversation_telemetry_get_locator', + 'chat_history_get_debug_entry' + ].includes(toolName) + && descriptor + && typeof descriptor === 'object' + )) + ) + : {}; return { schema_version: TURN_LIVE_PROGRESS_LOCATOR_SCHEMA_VERSION, @@ -33205,26 +33217,10 @@ function buildThinkingDiagnosticsLocatorPayload(request) { ? Math.max(0, Math.round(Number(latestProgress.elapsed_ms))) : null } : null, - mcp_access: { - turn_execution_get_live_progress: buildMcpToolAccess( - 'turn_execution_get_live_progress', - { - request_id: requestId, - namespace: namespaceContext.namespace || null, - user_concept_id: namespaceContext.user_id || null, - window_session_id: getWindowSessionId() - }, - 'Fetch the bounded live progress snapshot for this active turn; pass section, limit, and offset only when explicit detail hydration is needed.' - ), - turn_execution_get_diagnostics: buildMcpToolAccess( - 'turn_execution_get_diagnostics', - { - request_id: requestId, - namespace: namespaceContext.namespace || null - }, - 'Fetch the persisted post-turn diagnostics once the request has completed.' - ) - } + mcp_access: executableMcpAccess, + retrieval_status: Object.keys(executableMcpAccess).length > 0 + ? 'server_delegation_available' + : 'server_delegation_unavailable' }; } @@ -33411,7 +33407,7 @@ function buildLlmDebugLocatorDiagnosticSummary({ debugData, turnExecutionDiagnos ); } -function buildLlmDebugLocatorPayload({ turnId, debugData, metadata, workflowExecutionTelemetry }) { +function buildLlmDebugLocatorPayload({ turnId, debugData, metadata }) { if (!debugData || typeof debugData !== 'object') { return null; } @@ -33431,14 +33427,29 @@ function buildLlmDebugLocatorPayload({ turnId, debugData, metadata, workflowExec turnExecutionDiagnostics?.history_location || debugData.history_location ); const sessionId = historyLocation?.session_id || activeChatSessionId || null; - const historyIndex = Number.isInteger(historyLocation?.history_index) - ? historyLocation.history_index - : null; - const diagnosticSummary = buildLlmDebugLocatorDiagnosticSummary({ debugData, turnExecutionDiagnostics }); + const authoritativeMcpAccess = ( + debugData.telemetry_locator_mcp_access + && typeof debugData.telemetry_locator_mcp_access === 'object' + ) + ? debugData.telemetry_locator_mcp_access + : null; + const executableMcpAccess = authoritativeMcpAccess + ? Object.fromEntries( + Object.entries(authoritativeMcpAccess).filter(([toolName, descriptor]) => ( + [ + 'chat_history_get_debug_entry', + 'conversation_telemetry_get_locator', + 'turn_execution_get_diagnostics' + ].includes(toolName) + && descriptor + && typeof descriptor === 'object' + )) + ) + : {}; const payload = { schema_version: TURN_TELEMETRY_LOCATOR_SCHEMA_VERSION, @@ -33460,84 +33471,12 @@ function buildLlmDebugLocatorPayload({ turnId, debugData, metadata, workflowExec turnExecutionDiagnostics }) ), - mcp_access: {} + mcp_access: executableMcpAccess, + retrieval_status: Object.keys(executableMcpAccess).length > 0 + ? 'server_delegation_available' + : 'server_delegation_unavailable' }; - if (requestId) { - payload.mcp_access.turn_execution_get_diagnostics = buildMcpToolAccess( - 'turn_execution_get_diagnostics', - { - request_id: requestId, - namespace: namespaceContext.namespace || null - }, - 'Fetch the persisted turn-execution diagnostics for this turn.' - ); - } - if (sessionId && historyIndex !== null) { - payload.mcp_access.chat_history_get_debug_entry = buildMcpToolAccess( - 'chat_history_get_debug_entry', - buildChatHistoryAccessArgs({ - sessionId, - historyIndex - }), - 'Fetch the exact stored llm_debug_data entry for this turn.' - ); - } - if (sessionId) { - payload.mcp_access.conversation_telemetry_get_locator = buildMcpToolAccess( - 'conversation_telemetry_get_locator', - { - session_id: sessionId, - namespace: namespaceContext.namespace || null, - user_concept_id: namespaceContext.user_id || null, - organisation_concept_id: namespaceContext.org_id || null - }, - 'Fetch the compact conversation locator for the surrounding session.' - ); - } - const workflowExecutionTraces = Array.isArray(workflowExecutionTelemetry?.traces) - ? workflowExecutionTelemetry.traces - .map((trace) => { - if (!trace || typeof trace !== 'object') { - return null; - } - const executionId = typeof trace.execution_id === 'string' - ? trace.execution_id.trim() - : ''; - const instanceId = typeof trace.instance_id === 'string' - ? trace.instance_id.trim() - : ''; - if (!executionId && !instanceId) { - return null; - } - const traceRole = workflowExecutionTelemetry.primaryTrace === trace - ? 'selected_workflow' - : 'auxiliary_workflow'; - return { - execution_id: executionId || null, - instance_id: instanceId || null, - workflow_id: typeof trace.workflow_id === 'string' ? trace.workflow_id : null, - trace_role: traceRole, - mcp_access: buildMcpToolAccess( - 'workflow_get_execution_trace', - { - execution_id: executionId || null, - instance_id: instanceId || null - }, - 'Fetch the durable workflow execution trace referenced by this turn.' - ) - }; - }) - .filter(Boolean) - : []; - if (workflowExecutionTraces.length > 0) { - payload.mcp_access.workflow_execution_traces = workflowExecutionTraces; - const primaryTrace = workflowExecutionTraces.find((trace) => trace.trace_role === 'selected_workflow'); - if (primaryTrace?.mcp_access) { - payload.mcp_access.workflow_get_execution_trace = primaryTrace.mcp_access; - } - } - return payload; } @@ -33607,6 +33546,50 @@ async function fetchConversationTelemetryLocatorPayload(sessionId) { } } +async function fetchTurnTelemetryMcpAccess({ + requestId, + sessionId = null, + historyIndex = null, + includeLiveProgress = true +} = {}) { + const cleanRequestId = typeof requestId === 'string' ? requestId.trim() : ''; + if (!cleanRequestId || typeof fetch !== 'function') { + return null; + } + + try { + const params = new URLSearchParams({ + request_id: cleanRequestId, + include_live_progress: includeLiveProgress ? 'true' : 'false' + }); + const cleanSessionId = typeof sessionId === 'string' ? sessionId.trim() : ''; + if (cleanSessionId) { + params.set('session_id', cleanSessionId); + } + if (Number.isInteger(historyIndex) && historyIndex >= 0) { + params.set('history_index', String(historyIndex)); + } + const response = await fetchWithTimeout( + `/von/history/turn_telemetry_access?${params.toString()}`, + { + headers: buildChatFetchHeaders(), + timeoutMs: CONVERSATION_TELEMETRY_LOCATOR_FETCH_TIMEOUT_MS + } + ); + const body = await response.json(); + if (!response.ok || !body || typeof body !== 'object') { + return null; + } + if (body.schema_version !== TURN_TELEMETRY_MCP_ACCESS_SCHEMA_VERSION) { + return null; + } + return body; + } catch (error) { + console.warn('[chatTab] Failed to fetch server turn telemetry delegation:', error); + return null; + } +} + function findConversationTelemetryLocatorTurn(payload, { turnId = null, requestId = null, historyLocation = null } = {}) { const turns = Array.isArray(payload?.turns) ? payload.turns : []; const cleanRequestId = typeof requestId === 'string' ? requestId.trim() : ''; @@ -33651,10 +33634,6 @@ async function hydrateTurnHistoryLocationFromConversationLocator(turnId, debugDa debugData.turn_execution_diagnostics?.history_location || debugData.history_location ); const requestId = resolveConversationTelemetryRequestId(debugData); - if (existingHistoryLocation && requestId) { - return debugData; - } - const sessionId = ( (typeof existingHistoryLocation?.session_id === 'string' && existingHistoryLocation.session_id.trim()) || (typeof activeChatSessionId === 'string' ? activeChatSessionId.trim() : '') @@ -33690,6 +33669,9 @@ async function hydrateTurnHistoryLocationFromConversationLocator(turnId, debugDa if (matchedRequestId && !resolveConversationTelemetryRequestId(merged)) { merged.request_id = matchedRequestId; } + if (matchedTurn?.mcp_access && typeof matchedTurn.mcp_access === 'object') { + merged.telemetry_locator_mcp_access = matchedTurn.mcp_access; + } if (debugData.turn_execution_diagnostics && typeof debugData.turn_execution_diagnostics === 'object') { merged.turn_execution_diagnostics = { ...debugData.turn_execution_diagnostics, @@ -33758,29 +33740,9 @@ function buildConversationLlmTelemetryLocatorPayload() { : null, history_location: historyLocation, request_id: requestId, - mcp_access: { - chat_history_get_debug_entry: buildMcpToolAccess( - 'chat_history_get_debug_entry', - buildChatHistoryAccessArgs({ - sessionId: historyLocation?.session_id || activeChatSessionId || null, - historyIndex: Number.isInteger(historyLocation?.history_index) - ? historyLocation.history_index - : undefined - }), - 'Fetch the exact stored llm_debug_data for this turn.' - ) - } + mcp_access: {}, + retrieval_status: 'server_delegation_unavailable' }; - if (requestId) { - turnPayload.mcp_access.turn_execution_get_diagnostics = buildMcpToolAccess( - 'turn_execution_get_diagnostics', - { - request_id: requestId, - namespace: buildConversationTelemetryNamespaceContext().namespace || null - }, - 'Fetch the persisted turn-execution diagnostics for this turn.' - ); - } if (unavailableLocatorFields.length > 0) { turnPayload.unavailable_locator_fields = unavailableLocatorFields; } @@ -33805,7 +33767,8 @@ function buildConversationLlmTelemetryLocatorPayload() { turns_with_unavailable_locator_fields_count: turnsWithUnavailableLocatorFieldsCount, ordering: 'timestamp_then_turn_id' }, - mcp_access: buildConversationTelemetrySessionMcpAccess(sessionId, namespaceContext), + mcp_access: {}, + retrieval_status: 'server_delegation_unavailable', turns }; } @@ -33848,6 +33811,24 @@ function buildConversationTelemetryAccessPayload({ locatorPayload = null } = {}) org_id: basePayload.namespace_context.org_id || null } : buildConversationTelemetryNamespaceContext(); + const authoritativeMcpAccess = ( + authoritativeLocator?.mcp_access + && typeof authoritativeLocator.mcp_access === 'object' + ) + ? Object.fromEntries( + Object.entries(authoritativeLocator.mcp_access).filter(([toolName, descriptor]) => ( + [ + 'conversation_telemetry_get_locator', + 'chat_history_get_segments' + ].includes(toolName) + && descriptor + && typeof descriptor === 'object' + )) + ) + : {}; + const retrievalStatus = Object.keys(authoritativeMcpAccess).length > 0 + ? 'server_delegation_available' + : 'server_delegation_unavailable'; return { schema_version: CONVERSATION_TELEMETRY_ACCESS_SCHEMA_VERSION, @@ -33888,20 +33869,25 @@ function buildConversationTelemetryAccessPayload({ locatorPayload = null } = {}) locator_schema_version: authoritativeLocator?.schema_version || null, locator_generated_at_utc: authoritativeLocator?.generated_at_utc || null }, - mcp_access: buildConversationTelemetrySessionMcpAccess(sessionId, namespaceContext), + mcp_access: authoritativeMcpAccess, + retrieval_status: retrievalStatus, agent_instructions: { summary: ( 'This payload is an MCP access envelope for the conversation, not the ' + 'authoritative per-turn locator.' ), - steps: [ - 'Call conversation_telemetry_get_locator first to fetch the authoritative per-turn locator for this session.', - 'Call chat_history_get_segments with include_debug=true to retrieve transcript segments and embedded llm_debug payloads.', - 'Call turn_execution_list to inspect persisted turn-execution records and request_id coverage.' - ], + steps: retrievalStatus === 'server_delegation_available' + ? [ + 'Call conversation_telemetry_get_locator first to fetch the authoritative per-turn locator for this session.', + 'Call chat_history_get_segments for the bounded transcript projection.', + 'Use each returned turn descriptor to fetch the exact bounded debug or diagnostics artefact.' + ] + : [ + 'Copy the conversation info again after the server locator becomes available.' + ], notes: [ 'Treat the server locator as authoritative when history_location or request_id matters.', - 'Use the session_id and namespace_context in this payload to scope those MCP calls.' + 'If retrieval_status is server_delegation_unavailable, no raw-ID fallback is authorised; copy again after the server locator is available.' ] } }; diff --git a/src/frontend/web/von_interface/static/js/test/chatTab.test.js b/src/frontend/web/von_interface/static/js/test/chatTab.test.js index 69a99a911..3200bc8f2 100644 --- a/src/frontend/web/von_interface/static/js/test/chatTab.test.js +++ b/src/frontend/web/von_interface/static/js/test/chatTab.test.js @@ -6290,16 +6290,14 @@ describe('thinking card toggle accessibility', () => { retained.copyButton.click(); await Promise.resolve(); + await new Promise((resolve) => setTimeout(resolve, 0)); expect(writeText).toHaveBeenCalledTimes(1); expect(JSON.parse(writeText.mock.calls[0][0])).toEqual(expect.objectContaining({ schema_version: 'turn_live_progress_locator.v1', request_id: expect.any(String), - mcp_access: expect.objectContaining({ - turn_execution_get_live_progress: expect.objectContaining({ - tool_name: 'turn_execution_get_live_progress' - }) - }) + mcp_access: {}, + retrieval_status: 'server_delegation_unavailable' })); retained.toggleButton.click(); @@ -7045,11 +7043,10 @@ describe('copy diagnostics button visibility on preserved finished card', () => expect(copiedPayload.schema_version).toBe('turn_live_progress_locator.v1'); expect(copiedPayload.request_id).toBe('request-1458'); expect(copiedPayload.thinking_card_mode).toBeUndefined(); - expect(copiedPayload.mcp_access).toEqual(expect.objectContaining({ - turn_execution_get_live_progress: expect.objectContaining({ - tool_name: 'turn_execution_get_live_progress' - }) - })); + expect(copiedPayload.mcp_access).toEqual({}); + expect(copiedPayload.retrieval_status).toBe( + 'server_delegation_unavailable' + ); expect(copyBtn.textContent).toBe('✓ Copied'); expect(copyBtn.classList.contains('copy-json-copied')).toBe(true); expect(copyBtn.classList.contains('success-feedback')).toBe(false); @@ -9431,10 +9428,10 @@ describe('conversation LLM telemetry clipboard export', () => { }, request_id: 'req-1684-a' }); - expect(payload.turns[0].mcp_access).toEqual(expect.objectContaining({ - chat_history_get_debug_entry: expect.any(Object), - turn_execution_get_diagnostics: expect.any(Object) - })); + expect(payload.turns[0].mcp_access).toEqual({}); + expect(payload.turns[0].retrieval_status).toBe( + 'server_delegation_unavailable' + ); expect(payload.turns[1]).toMatchObject({ sequence: 2, turn_id: 'assistant-1743760860000', @@ -9784,7 +9781,11 @@ describe('conversation LLM telemetry clipboard export', () => { expect(copiedPayload.metadata.authoritative_locator_available).toBe(true); expect(copiedPayload.metadata.access_payload_source).toBe('authoritative_locator'); expect(copiedPayload.turns).toBeUndefined(); - expect(copiedPayload.agent_instructions.steps).toHaveLength(3); + expect(copiedPayload.mcp_access).toEqual({}); + expect(copiedPayload.retrieval_status).toBe( + 'server_delegation_unavailable' + ); + expect(copiedPayload.agent_instructions.steps).toHaveLength(1); expect(copiedPayload).toEqual(expect.objectContaining({ session_id: accessPayload.session_id, session_name: accessPayload.session_name, diff --git a/tests/backend/test_chat_history_service_segments.py b/tests/backend/test_chat_history_service_segments.py index 20a854adf..6d10135e7 100644 --- a/tests/backend/test_chat_history_service_segments.py +++ b/tests/backend/test_chat_history_service_segments.py @@ -401,6 +401,66 @@ def test_get_chat_history_debug_entry_hydrates_blob_refs(monkeypatch): assert debug == original_debug +def test_get_chat_history_telemetry_locator_projection_keeps_debug_payloads_out( + monkeypatch, +): + from src.backend.services import chat_history_service + + class _CompactProjectionCollection: + def __init__(self): + self.pipeline = None + + def aggregate(self, pipeline): + self.pipeline = pipeline + return iter( + [ + { + "user_id": "#V#u", + "session_id": "s1", + "session_name": "Compact locator", + "namespace": "#V#u@org", + "history": [ + { + "role": "assistant", + "timestamp": "2026-07-28T00:00:00Z", + "llm_debug_data": { + "request_id": "req-compact-locator", + "turn_id": "turn-compact-locator", + "timestamp_utc": "2026-07-28T00:00:00Z", + }, + } + ], + } + ] + ) + + collection = _CompactProjectionCollection() + monkeypatch.setattr( + chat_history_service, + "get_chat_history_collection_service", + lambda **_kwargs: collection, + ) + + result = chat_history_service.get_chat_history_telemetry_locator_projection( + "#V#u", + "s1", + namespace="#V#u@org", + ) + + assert result is not None + assert result["history"][0]["llm_debug_data"] == { + "request_id": "req-compact-locator", + "turn_id": "turn-compact-locator", + "timestamp_utc": "2026-07-28T00:00:00Z", + } + assert collection.pipeline is not None + history_projection = collection.pipeline[-1]["$project"]["history"] + projected_debug = history_projection["$map"]["in"]["llm_debug_data"][ + "$cond" + ][1] + assert set(projected_debug) == {"request_id", "turn_id", "timestamp_utc"} + + def test_get_chat_history_segments_keeps_debug_blob_refs_compact_by_default( monkeypatch, ): diff --git a/tests/backend/test_conversation_scope_binding_service.py b/tests/backend/test_conversation_scope_binding_service.py index ba9d755f1..3ff6ae010 100644 --- a/tests/backend/test_conversation_scope_binding_service.py +++ b/tests/backend/test_conversation_scope_binding_service.py @@ -1,8 +1,13 @@ +from datetime import datetime, timedelta, timezone + from src.backend.services.conversation_scope_binding_service import ( + TELEMETRY_READ_DELEGATION_AUDIENCE, build_conversation_scope_binding, build_history_location_binding, + build_turn_telemetry_binding, verify_conversation_scope_binding, verify_history_location_binding, + verify_turn_telemetry_binding, ) @@ -59,3 +64,113 @@ def test_tampered_conversation_scope_binding_fails_closed() -> None: assert verified["success"] is False assert verified["error_code"] == "INVALID_CONTEXT_BINDING" assert verified["identifier_binding"]["validation_status"] == "binding_id_mismatch" + + +def test_external_conversation_read_delegation_binds_audience_tool_and_actor() -> None: + binding = build_conversation_scope_binding( + chat_session_id="chat-2609-1", + history_owner_user_id="#V#owner", + read_namespace="#V#owner@org", + delegated_actor_user_id="#V#actor_a", + delegated_actor_namespace="#V#actor_a@org", + organisation_concept_id="#V#org", + permitted_tools=("conversation_telemetry_get_locator",), + ) + + verified = verify_conversation_scope_binding( + binding, + require_read_delegation=True, + expected_audience=TELEMETRY_READ_DELEGATION_AUDIENCE, + expected_tool="conversation_telemetry_get_locator", + expected_actor_user_id="#V#actor_a", + ) + + assert verified["success"] is True + assert verified["delegated_actor_user_id"] == "#V#actor_a" + assert verified["delegated_actor_namespace"] == "#V#actor_a@org" + assert verified["read_delegation"]["permitted_tools"] == [ + "conversation_telemetry_get_locator" + ] + assert verified["identifier_binding"]["read_delegation_validation_status"] == ( + "verified" + ) + + wrong_tool = verify_conversation_scope_binding( + binding, + require_read_delegation=True, + expected_audience=TELEMETRY_READ_DELEGATION_AUDIENCE, + expected_tool="chat_history_get_debug_entry", + expected_actor_user_id="#V#actor_a", + ) + wrong_actor = verify_conversation_scope_binding( + binding, + require_read_delegation=True, + expected_audience=TELEMETRY_READ_DELEGATION_AUDIENCE, + expected_tool="conversation_telemetry_get_locator", + expected_actor_user_id="#V#actor_b", + ) + wrong_audience = verify_conversation_scope_binding( + binding, + require_read_delegation=True, + expected_audience="another_mcp_surface", + expected_tool="conversation_telemetry_get_locator", + expected_actor_user_id="#V#actor_a", + ) + + assert wrong_tool["error_code"] == "READ_DELEGATION_TOOL_MISMATCH" + assert wrong_actor["error_code"] == "READ_DELEGATION_ACTOR_MISMATCH" + assert wrong_audience["error_code"] == "READ_DELEGATION_AUDIENCE_MISMATCH" + + +def test_expired_read_delegation_fails_closed() -> None: + issued_at = datetime(2026, 7, 28, 12, 0, tzinfo=timezone.utc) + binding = build_history_location_binding( + chat_session_id="chat-2609-expired", + history_index=4, + history_owner_user_id="#V#actor_a", + delegated_actor_user_id="#V#actor_a", + ttl_seconds=30, + issued_at=issued_at, + ) + + verified = verify_history_location_binding( + binding, + require_read_delegation=True, + expected_audience=TELEMETRY_READ_DELEGATION_AUDIENCE, + expected_tool="chat_history_get_debug_entry", + expected_actor_user_id="#V#actor_a", + now=issued_at + timedelta(seconds=31), + ) + + assert verified["success"] is False + assert verified["error_code"] == "READ_DELEGATION_EXPIRED" + + +def test_turn_telemetry_delegation_binds_exact_request_and_tool() -> None: + binding = build_turn_telemetry_binding( + request_id="req-2609-1", + delegated_actor_user_id="#V#actor_a", + permitted_tool="turn_execution_get_diagnostics", + chat_session_id="chat-2609-2", + history_index=7, + history_owner_user_id="#V#actor_a", + read_namespace="#V#actor_a@org", + organisation_concept_id="#V#org", + ) + + verified = verify_turn_telemetry_binding( + binding, + expected_tool="turn_execution_get_diagnostics", + expected_actor_user_id="#V#actor_a", + ) + wrong_tool = verify_turn_telemetry_binding( + binding, + expected_tool="turn_execution_get_live_progress", + expected_actor_user_id="#V#actor_a", + ) + + assert verified["success"] is True + assert verified["request_id"] == "req-2609-1" + assert verified["chat_session_id"] == "chat-2609-2" + assert verified["history_index"] == 7 + assert wrong_tool["error_code"] == "READ_DELEGATION_TOOL_MISMATCH" diff --git a/tests/backend/test_internal_mcp_locator_tools_gateway.py b/tests/backend/test_internal_mcp_locator_tools_gateway.py index 93ca68908..c7200f546 100644 --- a/tests/backend/test_internal_mcp_locator_tools_gateway.py +++ b/tests/backend/test_internal_mcp_locator_tools_gateway.py @@ -107,7 +107,7 @@ def test_chat_history_get_segments_gateway_accepts_bound_conversation_ref( _assert_schema_conformance(gateway, "chat_history_get_segments", payload) -def test_chat_history_get_segments_gateway_accepts_emitted_von_conversation_ref( +def test_chat_history_get_segments_gateway_rejects_emitted_von_conversation_ref_as_authority( monkeypatch, ) -> None: gateway = _build_gateway() @@ -145,14 +145,8 @@ def get_segments(*args, **kwargs): {"conversation_ref": _emitted_von_conversation_ref()}, ).payload - assert payload.get("success") is True - assert payload.get("session_id") == "ff9be41d-28f8-4864-ab0b-8c201e3152d0" - assert payload.get("identifier_binding", {}).get("mode") == ( - "public_conversation_ref" - ) - assert payload.get("identifier_binding", {}).get("validation_status") == ( - "normalised" - ) + assert payload.get("success") is False + assert payload.get("error_code") == "authenticated_actor_context_required" _assert_schema_conformance(gateway, "chat_history_get_segments", payload) diff --git a/tests/backend/test_mcp_stdio_server_exposes_workflow_tools.py b/tests/backend/test_mcp_stdio_server_exposes_workflow_tools.py index bea95f023..d8648a5c6 100644 --- a/tests/backend/test_mcp_stdio_server_exposes_workflow_tools.py +++ b/tests/backend/test_mcp_stdio_server_exposes_workflow_tools.py @@ -324,7 +324,10 @@ async def _runner(): assert all(payload["success"] is False for payload in payloads) assert { payload["error_code"] for payload in payloads - } == {"workflow_global_admin_authority_required"} + } == { + "authenticated_actor_context_required", + "workflow_global_admin_authority_required", + } def test_raw_stdio_sensitive_workflow_handlers_bind_untrusted_actor_source(monkeypatch): diff --git a/tests/backend/test_telemetry_read_delegation_stdio.py b/tests/backend/test_telemetry_read_delegation_stdio.py new file mode 100644 index 000000000..a4691d1c0 --- /dev/null +++ b/tests/backend/test_telemetry_read_delegation_stdio.py @@ -0,0 +1,409 @@ +from __future__ import annotations + +import asyncio +import json +from datetime import datetime, timedelta, timezone +from typing import Any, cast + +from src.backend.services.conversation_scope_binding_service import ( + build_history_location_binding, + build_turn_telemetry_binding, +) + + +def _stdio_payload(tool_name: str, arguments: dict[str, Any]) -> dict[str, Any]: + from src.backend.mcp_server import mcp_stdio_server + + async def run() -> dict[str, Any]: + result = await mcp_stdio_server.call_tool(tool_name, arguments) + first = cast(list[Any], result)[0] + return cast( + dict[str, Any], + json.loads(cast(str, getattr(first, "text"))), + ) + + return asyncio.run(run()) + + +def test_real_stdio_handler_path_accepts_exact_actor_bound_reads(monkeypatch) -> None: + history_ref = build_history_location_binding( + chat_session_id="chat-2609-stdio", + history_index=2, + history_owner_user_id="#V#actor_a", + read_namespace="#V#actor_a@org", + organisation_concept_id="#V#org", + delegated_actor_user_id="#V#actor_a", + delegated_actor_namespace="#V#actor_a@org", + ) + turn_ref = build_turn_telemetry_binding( + request_id="req-2609-stdio", + delegated_actor_user_id="#V#actor_a", + permitted_tool="turn_execution_get_diagnostics", + chat_session_id="chat-2609-stdio", + history_index=2, + history_owner_user_id="#V#actor_a", + read_namespace="#V#actor_a@org", + delegated_actor_namespace="#V#actor_a@org", + organisation_concept_id="#V#org", + ) + + monkeypatch.setattr( + "src.backend.services.workflow_event_integration_service.resolve_event_actor_context", + lambda user_id=None, org_id=None, namespace=None: (user_id, org_id), + ) + monkeypatch.setattr( + "src.backend.services.shared_conversation_service.resolve_conversation_owner", + lambda session_id: "#V#actor_a", + ) + monkeypatch.setattr( + "src.backend.services.chat_history_service.has_chat_history_session", + lambda user_id, session_id, namespace=None, include_legacy=True: ( + user_id == "#V#actor_a" + and session_id == "chat-2609-stdio" + and namespace == "#V#actor_a@org" + ), + ) + stored_debug = { + "request_id": "req-2609-stdio", + "diagnostic": "exact stored entry", + } + monkeypatch.setattr( + "src.backend.services.chat_history_service.get_chat_history_debug_entry", + lambda **_kwargs: stored_debug, + ) + monkeypatch.setattr( + "src.backend.services.turn_execution_diagnostics_service.get_turn_execution_diagnostics_payload", + lambda **_kwargs: { + "success": True, + "schema_version": "turn_execution_diagnostics.v1", + "request_id": "req-2609-stdio", + "chat_session_id": "chat-2609-stdio", + "history_location": { + "session_id": "chat-2609-stdio", + "history_index": 2, + }, + "derived_user_concept_id": "#V#actor_a", + "derived_organisation_concept_id": "#V#org", + "namespace": "#V#actor_a@org", + "diagnostics_source": "chat_history.llm_debug_data.turn_execution_diagnostics", + }, + ) + + debug_result = _stdio_payload( + "chat_history_get_debug_entry", + { + "history_location_ref": history_ref, + "namespace": "#V#actor_a@org", + "user_concept_id": "#V#actor_a", + "organisation_concept_id": "#V#org", + }, + ) + diagnostics_result = _stdio_payload( + "turn_execution_get_diagnostics", + { + "request_id": "req-2609-stdio", + "turn_telemetry_ref": turn_ref, + "namespace": "#V#actor_a@org", + "user_concept_id": "#V#actor_a", + "organisation_concept_id": "#V#org", + }, + ) + + assert debug_result["success"] is True + assert debug_result["llm_debug_data"] == stored_debug + assert debug_result["read_delegation"]["delegated_actor_user_id"] == "#V#actor_a" + assert debug_result["provenance"]["source_system"] == "mongo.chat_history" + assert diagnostics_result["success"] is True + assert diagnostics_result["request_id"] == "req-2609-stdio" + assert diagnostics_result["read_delegation"]["delegated_actor_user_id"] == ( + "#V#actor_a" + ) + assert diagnostics_result["source_system"] == "mongo.chat_history" + + +def test_stdio_raw_ids_retain_original_denials_and_actor_b_fails_closed() -> None: + actor_a_ref = build_turn_telemetry_binding( + request_id="req-2609-a", + delegated_actor_user_id="#V#actor_a", + permitted_tool="turn_execution_get_diagnostics", + read_namespace="#V#actor_a@org", + delegated_actor_namespace="#V#actor_a@org", + organisation_concept_id="#V#org", + ) + + raw_debug = _stdio_payload( + "chat_history_get_debug_entry", + { + "session_id": "chat-guessed", + "history_index": 2, + "namespace": "#V#actor_a@org", + "user_concept_id": "#V#actor_a", + }, + ) + raw_diagnostics = _stdio_payload( + "turn_execution_get_diagnostics", + { + "request_id": "req-guessed", + "namespace": "#V#actor_a@org", + "user_concept_id": "#V#actor_a", + }, + ) + actor_b = _stdio_payload( + "turn_execution_get_diagnostics", + { + "request_id": "req-2609-a", + "turn_telemetry_ref": actor_a_ref, + "namespace": "#V#actor_b@org", + "user_concept_id": "#V#actor_b", + "organisation_concept_id": "#V#org", + }, + ) + + assert raw_debug["error_code"] == "authenticated_actor_context_required" + assert raw_diagnostics["error_code"] == ( + "workflow_global_admin_authority_required" + ) + assert actor_b["error_code"] == "READ_DELEGATION_ACTOR_MISMATCH" + + +def test_stdio_delegation_tamper_expiry_tool_and_target_mismatches_fail_closed() -> None: + issued_at = datetime.now(timezone.utc) - timedelta(minutes=5) + expired_ref = build_turn_telemetry_binding( + request_id="req-expired-2609", + delegated_actor_user_id="#V#actor_a", + permitted_tool="turn_execution_get_diagnostics", + issued_at=issued_at, + ttl_seconds=30, + ) + diagnostics_ref = build_turn_telemetry_binding( + request_id="req-bound-2609", + delegated_actor_user_id="#V#actor_a", + permitted_tool="turn_execution_get_diagnostics", + chat_session_id="chat-bound-2609", + history_owner_user_id="#V#actor_a", + read_namespace="#V#actor_a@org", + delegated_actor_namespace="#V#actor_a@org", + organisation_concept_id="#V#org", + ) + tampered_ref = dict(diagnostics_ref) + tampered_ref["request_id"] = "req-tampered-2609" + + expired = _stdio_payload( + "turn_execution_get_diagnostics", + { + "request_id": "req-expired-2609", + "turn_telemetry_ref": expired_ref, + "user_concept_id": "#V#actor_a", + }, + ) + tampered = _stdio_payload( + "turn_execution_get_diagnostics", + { + "request_id": "req-tampered-2609", + "turn_telemetry_ref": tampered_ref, + "user_concept_id": "#V#actor_a", + }, + ) + wrong_tool = _stdio_payload( + "turn_execution_get_live_progress", + { + "request_id": "req-bound-2609", + "turn_telemetry_ref": diagnostics_ref, + "user_concept_id": "#V#actor_a", + }, + ) + wrong_session = _stdio_payload( + "turn_execution_get_diagnostics", + { + "request_id": "req-bound-2609", + "session_id": "chat-other-2609", + "turn_telemetry_ref": diagnostics_ref, + "user_concept_id": "#V#actor_a", + }, + ) + wrong_namespace = _stdio_payload( + "turn_execution_get_diagnostics", + { + "request_id": "req-bound-2609", + "turn_telemetry_ref": diagnostics_ref, + "namespace": "#V#actor_a@other_org", + "user_concept_id": "#V#actor_a", + "organisation_concept_id": "#V#org", + }, + ) + + assert expired["error_code"] == "READ_DELEGATION_EXPIRED" + assert tampered["error_code"] == "INVALID_CONTEXT_BINDING" + assert wrong_tool["error_code"] == "READ_DELEGATION_TOOL_MISMATCH" + assert wrong_session["error_code"] == "INVALID_CONTEXT_BINDING" + assert wrong_session["error_details"]["reason_code"] == ( + "READ_DELEGATION_TARGET_MISMATCH" + ) + assert wrong_namespace["error_code"] == "INVALID_CONTEXT_BINDING" + + +def test_stdio_large_debug_payload_is_paginated(monkeypatch) -> None: + history_ref = build_history_location_binding( + chat_session_id="chat-2609-large", + history_index=5, + history_owner_user_id="#V#actor_a", + read_namespace="#V#actor_a@org", + delegated_actor_user_id="#V#actor_a", + ) + monkeypatch.setattr( + "src.backend.services.workflow_event_integration_service.resolve_event_actor_context", + lambda user_id=None, org_id=None, namespace=None: (user_id, org_id), + ) + monkeypatch.setattr( + "src.backend.services.shared_conversation_service.resolve_conversation_owner", + lambda session_id: "#V#actor_a", + ) + monkeypatch.setattr( + "src.backend.services.chat_history_service.has_chat_history_session", + lambda *_args, **_kwargs: True, + ) + monkeypatch.setattr( + "src.backend.services.chat_history_service.get_chat_history_debug_entry", + lambda **_kwargs: {"large": "x" * 150_000}, + ) + + first_page = _stdio_payload( + "chat_history_get_debug_entry", + { + "history_location_ref": history_ref, + "namespace": "#V#actor_a@org", + "user_concept_id": "#V#actor_a", + "limit": 20_000, + }, + ) + + assert first_page["success"] is True + assert "llm_debug_data" not in first_page + assert first_page["bounded_read"]["returned_chars"] == 20_000 + assert first_page["bounded_read"]["has_more"] is True + assert first_page["bounded_read"]["next_offset"] == 20_000 + assert len(first_page["bounded_read"]["json_chunk"]) == 20_000 + + +def test_stdio_live_progress_delegation_cannot_select_another_scope( + monkeypatch, +) -> None: + turn_ref = build_turn_telemetry_binding( + request_id="req-live-2609", + delegated_actor_user_id="#V#actor_a", + permitted_tool="turn_execution_get_live_progress", + read_namespace="#V#actor_a@org", + delegated_actor_namespace="#V#actor_a@org", + organisation_concept_id="#V#org", + ) + captured: dict[str, Any] = {} + + def get_live_progress(**kwargs): + captured.update(kwargs) + return { + "request_id": "req-live-2609", + "resolved_scope_key": "user:#V#actor_a", + "status": "running", + } + + monkeypatch.setattr( + "src.backend.services.turn_execution_live_progress_service.get_turn_execution_live_progress_payload", + get_live_progress, + ) + + result = _stdio_payload( + "turn_execution_get_live_progress", + { + "request_id": "req-live-2609", + "turn_telemetry_ref": turn_ref, + "namespace": "#V#actor_a@org", + "user_concept_id": "#V#actor_a", + "organisation_concept_id": "#V#org", + "scope_key": "user:#V#actor_b", + "window_session_id": "actor-b-window", + "anonymous_session_id": "actor-b-session", + }, + ) + + assert result["status"] == "running" + assert result["read_delegation"]["delegated_actor_user_id"] == "#V#actor_a" + assert captured["scope_key"] is None + assert captured["window_session_id"] is None + assert captured["anonymous_session_id"] is None + + monkeypatch.setattr( + "src.backend.services.turn_execution_live_progress_service.get_turn_execution_live_progress_payload", + lambda **_kwargs: { + "request_id": "req-live-2609", + "resolved_scope_key": "user:#V#actor_b", + "status": "running", + }, + ) + mismatched = _stdio_payload( + "turn_execution_get_live_progress", + { + "request_id": "req-live-2609", + "turn_telemetry_ref": turn_ref, + "namespace": "#V#actor_a@org", + "user_concept_id": "#V#actor_a", + "organisation_concept_id": "#V#org", + }, + ) + + assert mismatched["error_code"] == ( + "READ_DELEGATION_CANONICAL_TARGET_MISMATCH" + ) + + +def test_stdio_diagnostics_delegation_reads_owner_namespace_for_invitee( + monkeypatch, +) -> None: + turn_ref = build_turn_telemetry_binding( + request_id="req-shared-2609", + delegated_actor_user_id="#V#invitee", + permitted_tool="turn_execution_get_diagnostics", + chat_session_id="chat-shared-2609", + history_index=9, + history_owner_user_id="#V#owner", + read_namespace="#V#owner@org", + delegated_actor_namespace="#V#invitee@org", + organisation_concept_id="#V#org", + ) + captured: dict[str, Any] = {} + + def get_diagnostics(**kwargs): + captured.update(kwargs) + return { + "success": True, + "request_id": "req-shared-2609", + "chat_session_id": "chat-shared-2609", + "history_location": { + "session_id": "chat-shared-2609", + "history_index": 9, + }, + "derived_user_concept_id": "#V#owner", + "derived_organisation_concept_id": "#V#org", + "namespace": "#V#owner@org", + "diagnostics_source": "mongo.turn_execution_records", + } + + monkeypatch.setattr( + "src.backend.services.turn_execution_diagnostics_service.get_turn_execution_diagnostics_payload", + get_diagnostics, + ) + + result = _stdio_payload( + "turn_execution_get_diagnostics", + { + "request_id": "req-shared-2609", + "turn_telemetry_ref": turn_ref, + "namespace": "#V#invitee@org", + "user_concept_id": "#V#invitee", + "organisation_concept_id": "#V#org", + }, + ) + + assert result["success"] is True + assert captured["namespace"] == "#V#owner@org" + assert captured["delegated_actor_user_id"] == "#V#invitee" + assert captured["delegated_actor_namespace"] == "#V#invitee@org" diff --git a/tests/backend/test_turn_execution_diagnostics_workflow_trace_refs.py b/tests/backend/test_turn_execution_diagnostics_workflow_trace_refs.py index 34210bbe2..009cf2490 100644 --- a/tests/backend/test_turn_execution_diagnostics_workflow_trace_refs.py +++ b/tests/backend/test_turn_execution_diagnostics_workflow_trace_refs.py @@ -229,6 +229,73 @@ def test_observational_diagnostics_do_not_synthesise_retired_workflow_identity( assert payload["phase_history"] == embedded_diagnostics["phase_history"] +def test_exact_delegated_history_location_avoids_request_wide_history_scan( + monkeypatch, +) -> None: + captured: dict[str, object] = {} + + def exact_debug_lookup(**kwargs): + captured.update(kwargs) + return { + "request_id": "req-exact-2609", + "prompt_text": "Inspect this exact turn.", + "turn_execution_diagnostics": { + "schema_version": "turn_execution_diagnostics.v1", + "request_id": "req-exact-2609", + "latest_progress": { + "status": "completed", + "phase": "completed", + }, + }, + } + + monkeypatch.setattr( + diagnostics_service, + "get_chat_history_debug_entry", + exact_debug_lookup, + ) + monkeypatch.setattr( + diagnostics_service, + "_resolve_history_context", + lambda **_kwargs: (_ for _ in ()).throw( + AssertionError("request-wide history lookup must not run") + ), + ) + monkeypatch.setattr( + diagnostics_service, + "_load_turn_execution_record", + lambda **_kwargs: None, + ) + + payload = get_turn_execution_diagnostics_payload( + request_id="req-exact-2609", + namespace="#V#actor_a@org", + delegated_actor_user_id="#V#actor_a", + delegated_actor_namespace="#V#actor_a@org", + chat_session_id="session-exact-2609", + history_index=7, + history_owner_user_id="#V#actor_a", + organisation_concept_id="#V#org", + ) + + assert payload is not None + assert payload["chat_session_id"] == "session-exact-2609" + assert payload["history_location"] == { + "session_id": "session-exact-2609", + "history_index": 7, + } + assert payload["derived_user_concept_id"] == "#V#actor_a" + assert payload["derived_organisation_concept_id"] == "#V#org" + assert captured == { + "user_id": "#V#actor_a", + "session_id": "session-exact-2609", + "history_index": 7, + "namespace": "#V#actor_a@org", + "include_legacy": True, + "hydrate_blob_refs": True, + } + + def test_legacy_diagnostics_still_reconstruct_historical_workflow_identity( monkeypatch, ) -> None: diff --git a/tests/backend/test_von_history_telemetry_locator_endpoint.py b/tests/backend/test_von_history_telemetry_locator_endpoint.py index 071b65531..73450e190 100644 --- a/tests/backend/test_von_history_telemetry_locator_endpoint.py +++ b/tests/backend/test_von_history_telemetry_locator_endpoint.py @@ -2,6 +2,11 @@ from flask import Flask +from src.backend.services.conversation_scope_binding_service import ( + verify_history_location_binding, + verify_turn_telemetry_binding, +) + def test_history_telemetry_locator_uses_org_hint_to_upgrade_bare_namespace( monkeypatch, @@ -69,3 +74,108 @@ def test_history_telemetry_locator_uses_org_hint_to_upgrade_bare_namespace( assert captured["requested_namespace"] == "#V#test_user@org" assert captured["requested_user_id"] == "#V#test_user" assert captured["organisation_concept_id"] == "#V#org" + + +def test_history_turn_telemetry_access_issues_exact_actor_bound_descriptors( + monkeypatch, +): + import src.backend.server.routes.von_routes as von_routes + + monkeypatch.setattr( + "src.backend.security.access_control.get_effective_user_concept_id", + lambda: "#V#actor_a", + ) + monkeypatch.setattr( + von_routes, + "get_effective_context", + lambda *_args, **_kwargs: {"namespace": "#V#actor_a@org"}, + ) + monkeypatch.setattr( + von_routes, + "_resolve_history_request_scope_hints", + lambda **_kwargs: ("#V#actor_a@org", "#V#org"), + ) + monkeypatch.setattr( + von_routes, + "_resolve_shared_conversation_owner", + lambda **_kwargs: ("#V#actor_a", None), + ) + monkeypatch.setattr( + von_routes.chat_history_service, + "get_chat_history_debug_entry", + lambda **_kwargs: {"request_id": "req-route-2609"}, + ) + monkeypatch.setattr( + "src.backend.services.turn_execution_diagnostics_service.get_turn_execution_diagnostics_payload", + lambda **_kwargs: (_ for _ in ()).throw( + AssertionError("exact history locator must avoid full diagnostics hydration") + ), + ) + + app = Flask(__name__) + app.secret_key = "test-secret" + app.register_blueprint(von_routes.von_bp, url_prefix="/von") + response = app.test_client().get( + "/von/history/turn_telemetry_access", + query_string={ + "request_id": "req-route-2609", + "session_id": "session-route-2609", + "history_index": 3, + "include_live_progress": "false", + }, + headers={"X-Von-Window-Session": "ws-route-2609"}, + ) + + assert response.status_code == 200 + body = response.get_json() + assert body["schema_version"] == "turn_telemetry_mcp_access.v1" + assert set(body["mcp_access"]) == { + "chat_history_get_debug_entry", + "conversation_telemetry_get_locator", + "turn_execution_get_diagnostics", + } + + diagnostics_arguments = body["mcp_access"][ + "turn_execution_get_diagnostics" + ]["arguments"] + with app.app_context(): + verified_turn = verify_turn_telemetry_binding( + diagnostics_arguments["turn_telemetry_ref"], + expected_tool="turn_execution_get_diagnostics", + expected_actor_user_id="#V#actor_a", + ) + assert verified_turn["success"] is True + assert verified_turn["request_id"] == "req-route-2609" + + debug_arguments = body["mcp_access"]["chat_history_get_debug_entry"][ + "arguments" + ] + with app.app_context(): + verified_history = verify_history_location_binding( + debug_arguments["history_location_ref"], + require_read_delegation=True, + expected_tool="chat_history_get_debug_entry", + expected_actor_user_id="#V#actor_a", + ) + assert verified_history["success"] is True + assert verified_history["history_index"] == 3 + + +def test_history_turn_telemetry_access_requires_authenticated_actor(monkeypatch): + import src.backend.server.routes.von_routes as von_routes + + monkeypatch.setattr( + "src.backend.security.access_control.get_effective_user_concept_id", + lambda: None, + ) + + app = Flask(__name__) + app.secret_key = "test-secret" + app.register_blueprint(von_routes.von_bp, url_prefix="/von") + response = app.test_client().get( + "/von/history/turn_telemetry_access", + query_string={"request_id": "req-route-2609"}, + ) + + assert response.status_code == 401 + assert response.get_json()["error"] == "Not authenticated" diff --git a/tests/frontend/chatTabConversationLlmTelemetryCopy.test.js b/tests/frontend/chatTabConversationLlmTelemetryCopy.test.js index c2c6c3ff6..6654cc2f8 100644 --- a/tests/frontend/chatTabConversationLlmTelemetryCopy.test.js +++ b/tests/frontend/chatTabConversationLlmTelemetryCopy.test.js @@ -4,6 +4,7 @@ const chatTabModulePath = '../../src/frontend/web/von_interface/static/js/chatTa jest.mock('../../src/frontend/web/von_interface/static/js/apiService.js', () => ({ annotateTurn: jest.fn(), + fetchWithTimeout: jest.fn(), getUserContext: jest.fn(), getWindowSessionId: jest.fn(() => 'test-window-session'), postJson: jest.fn(), @@ -113,11 +114,10 @@ describe('chat conversation info copy control', () => { authoritative_locator_available: false, access_payload_source: 'local_context_summary' })); - expect(copiedPayload.mcp_access).toEqual(expect.objectContaining({ - conversation_telemetry_get_locator: expect.any(Object), - chat_history_get_segments: expect.any(Object), - turn_execution_list: expect.any(Object) - })); + expect(copiedPayload.mcp_access).toEqual({}); + expect(copiedPayload.retrieval_status).toBe( + 'server_delegation_unavailable' + ); expect(copiedPayload.turns).toBeUndefined(); expect(copiedPayload.agent_instructions).toEqual(expect.objectContaining({ summary: expect.any(String), @@ -175,15 +175,64 @@ describe('chat conversation info copy control', () => { authoritative_locator_available: false, access_payload_source: 'local_context_summary' })); - expect(copiedPayload.mcp_access).toEqual(expect.objectContaining({ - conversation_telemetry_get_locator: expect.any(Object), - chat_history_get_segments: expect.any(Object), - turn_execution_list: expect.any(Object) - })); + expect(copiedPayload.mcp_access).toEqual({}); + expect(copiedPayload.retrieval_status).toBe( + 'server_delegation_unavailable' + ); expect(copiedPayload.turns).toBeUndefined(); expect(showToast).toHaveBeenCalledWith( 'Copied conversation info JSON (server locator unavailable; local summary only).', 'info' ); }); + + test('copies only server-issued conversation read delegations', async () => { + const chatTab = require(chatTabModulePath); + const { fetchWithTimeout } = require('../../src/frontend/web/von_interface/static/js/apiService.js'); + const button = document.getElementById('copyConversationInfoJsonBtn'); + const signedAccess = { + conversation_telemetry_get_locator: { + tool_name: 'conversation_telemetry_get_locator', + arguments: { + conversation_ref: { signature: 'signed-conversation-ref' } + } + }, + chat_history_get_segments: { + tool_name: 'chat_history_get_segments', + arguments: { + conversation_ref: { signature: 'signed-segments-ref' } + } + } + }; + + global.fetch = jest.fn(); + fetchWithTimeout.mockResolvedValue({ + ok: true, + json: async () => ({ + schema_version: 'conversation_llm_telemetry_locator.v1', + generated_at_utc: '2026-07-28T12:00:00Z', + session_id: 'session-signed', + session_name: 'Signed Session', + namespace_context: {}, + metadata: { total_turns: 0 }, + mcp_access: { + ...signedAccess, + turn_execution_list: { + tool_name: 'turn_execution_list', + arguments: {} + } + }, + turns: [] + }) + }); + chatTab.__testOnly_setActiveChatSession('session-signed', 'Signed Session'); + + const copied = await chatTab.__testOnly_copyConversationInfoToClipboard(button); + + expect(copied).toBe(true); + const payload = JSON.parse(navigator.clipboard.writeText.mock.calls[0][0]); + expect(payload.mcp_access).toEqual(signedAccess); + expect(payload.mcp_access.turn_execution_list).toBeUndefined(); + expect(payload.retrieval_status).toBe('server_delegation_available'); + }); }); diff --git a/tests/frontend/chatTabLlmDebugWorkflowExecution.test.js b/tests/frontend/chatTabLlmDebugWorkflowExecution.test.js index 94ac44333..744096d33 100644 --- a/tests/frontend/chatTabLlmDebugWorkflowExecution.test.js +++ b/tests/frontend/chatTabLlmDebugWorkflowExecution.test.js @@ -109,7 +109,7 @@ describe('LLM debug popup workflow execution hook', () => { }); global.fetch = jest.fn(); const { fetchWithTimeout } = require('../../src/frontend/web/von_interface/static/js/apiService.js'); - fetchWithTimeout.mockResolvedValue({ + fetchWithTimeout.mockResolvedValueOnce({ ok: true, json: async () => ({ schema_version: 'conversation_llm_telemetry_locator.v1', @@ -125,6 +125,34 @@ describe('LLM debug popup workflow execution hook', () => { } ] }) + }).mockResolvedValueOnce({ + ok: true, + json: async () => ({ + schema_version: 'turn_telemetry_mcp_access.v1', + history_location: { + session_id: 'session-history-ref', + history_index: 4 + }, + mcp_access: { + chat_history_get_debug_entry: { + tool_name: 'chat_history_get_debug_entry', + arguments: { + history_location_ref: { + signature: 'signed-history-ref' + } + } + }, + turn_execution_get_diagnostics: { + tool_name: 'turn_execution_get_diagnostics', + arguments: { + request_id: 'req-history-ref', + turn_telemetry_ref: { + signature: 'signed-turn-ref' + } + } + } + } + }) }); setLlmDebugDataForTurn('assistant-history-ref', { @@ -162,6 +190,13 @@ describe('LLM debug popup workflow execution hook', () => { chat_history_get_debug_entry: expect.any(Object), turn_execution_get_diagnostics: expect.any(Object) })); + expect(copiedPayload.retrieval_status).toBe( + 'server_delegation_available' + ); + const [turnAccessUrl] = fetchWithTimeout.mock.calls[1]; + expect( + new URL(turnAccessUrl, 'https://example.test').pathname + ).toBe('/von/history/turn_telemetry_access'); expect(copiedPayload.llm_debug_data).toBeUndefined(); expect(copiedPayload.messages).toBeUndefined(); expect(copiedPayload.response).toBeUndefined(); @@ -351,9 +386,10 @@ describe('LLM debug popup workflow execution hook', () => { } } }); - expect(payload.mcp_access).toEqual(expect.objectContaining({ - turn_execution_get_diagnostics: expect.any(Object) - })); + expect(payload.mcp_access).toEqual({}); + expect(payload.retrieval_status).toBe( + 'server_delegation_unavailable' + ); expect(payload.model).toBeUndefined(); expect(payload.messages).toBeUndefined(); }); @@ -462,22 +498,11 @@ describe('LLM debug popup workflow execution hook', () => { const popup = document.getElementById('chatLlmDebugPopup'); const payload = JSON.parse(popup.dataset.currentDebugData || '{}'); - expect(payload.mcp_access.workflow_get_execution_trace).toEqual(expect.objectContaining({ - tool_name: 'workflow_get_execution_trace', - arguments: expect.objectContaining({ - execution_id: 'exec-selected' - }) - })); - expect(payload.mcp_access.workflow_execution_traces).toEqual([ - expect.objectContaining({ - workflow_id: '#V#chat_narration_workflow', - trace_role: 'auxiliary_workflow' - }), - expect.objectContaining({ - workflow_id: '#V#arxiv_paper_representation_workflow', - trace_role: 'selected_workflow' - }) - ]); + expect(payload.mcp_access.workflow_get_execution_trace).toBeUndefined(); + expect(payload.mcp_access.workflow_execution_traces).toBeUndefined(); + expect(payload.retrieval_status).toBe( + 'server_delegation_unavailable' + ); }); test('copy payload treats workflow-use episode instance as selected-workflow evidence', async () => { @@ -513,23 +538,11 @@ describe('LLM debug popup workflow execution hook', () => { const popup = document.getElementById('chatLlmDebugPopup'); const payload = JSON.parse(popup.dataset.currentDebugData || '{}'); - expect(payload.mcp_access.workflow_get_execution_trace).toEqual(expect.objectContaining({ - tool_name: 'workflow_get_execution_trace', - arguments: expect.objectContaining({ - instance_id: 'wf-instance-selected' - }) - })); - expect(payload.mcp_access.workflow_execution_traces).toEqual([ - expect.objectContaining({ - workflow_id: '#V#chat_narration_workflow', - trace_role: 'auxiliary_workflow' - }), - expect.objectContaining({ - workflow_id: '#V#arxiv_paper_representation_workflow', - instance_id: 'wf-instance-selected', - trace_role: 'selected_workflow' - }) - ]); + expect(payload.mcp_access.workflow_get_execution_trace).toBeUndefined(); + expect(payload.mcp_access.workflow_execution_traces).toBeUndefined(); + expect(payload.retrieval_status).toBe( + 'server_delegation_unavailable' + ); }); test('hydrates popup locator history_location from the server conversation locator when missing locally', async () => { @@ -558,6 +571,33 @@ describe('LLM debug popup workflow execution hook', () => { history_location: { session_id: 'session-1718', history_index: 17 + }, + mcp_access: { + chat_history_get_debug_entry: { + tool_name: 'chat_history_get_debug_entry', + arguments: { + history_location_ref: { + signature: 'signed-debug-ref' + } + } + }, + conversation_telemetry_get_locator: { + tool_name: 'conversation_telemetry_get_locator', + arguments: { + conversation_ref: { + signature: 'signed-conversation-ref' + } + } + }, + turn_execution_get_diagnostics: { + tool_name: 'turn_execution_get_diagnostics', + arguments: { + request_id: 'req-1718', + turn_telemetry_ref: { + signature: 'signed-turn-ref' + } + } + } } } ] @@ -595,5 +635,8 @@ describe('LLM debug popup workflow execution hook', () => { conversation_telemetry_get_locator: expect.any(Object), turn_execution_get_diagnostics: expect.any(Object) })); + expect(payload.retrieval_status).toBe( + 'server_delegation_available' + ); }); }); diff --git a/tests/frontend/chatTabThinkingDiagnosticsCopy.test.js b/tests/frontend/chatTabThinkingDiagnosticsCopy.test.js index 0c3f8c41d..ea5d6980f 100644 --- a/tests/frontend/chatTabThinkingDiagnosticsCopy.test.js +++ b/tests/frontend/chatTabThinkingDiagnosticsCopy.test.js @@ -4,6 +4,7 @@ const chatTabModulePath = '../../src/frontend/web/von_interface/static/js/chatTa jest.mock('../../src/frontend/web/von_interface/static/js/apiService.js', () => ({ annotateTurn: jest.fn(), + fetchWithTimeout: jest.fn(), getUserContext: jest.fn(), getWindowSessionId: jest.fn(() => 'test-window-session'), postJson: jest.fn(), @@ -44,6 +45,7 @@ jest.mock('../../src/frontend/web/von_interface/static/js/utils/toast.js', () => describe('chat thinking diagnostics copy control', () => { beforeEach(() => { jest.resetModules(); + global.fetch = undefined; document.body.innerHTML = `