diff --git a/docs/engineering/contextual_knowledge_evolution.md b/docs/engineering/contextual_knowledge_evolution.md index 7ca19113b..5ec1ba973 100644 --- a/docs/engineering/contextual_knowledge_evolution.md +++ b/docs/engineering/contextual_knowledge_evolution.md @@ -98,6 +98,13 @@ Raw actor-scoped diagnostic text remains in its authoritative diagnostic carrier. Situation revisions retain the producing request identity so an invisible sidecar change can be traced to the turn that authored it. +Treat the situation, retained observations, and omission state as one coherent +carrier snapshot at read boundaries. After a mutation or reset, read back the +whole projection rather than combining a new situation revision with +turn-start observation state. An authoritative read may legitimately reduce +counts or clear content; stale asynchronous responses must not repopulate what +that read superseded. + Carrier visibility, contribution, and destructive authority are different. An accepted participant may contribute turns without thereby acquiring the right to reset the owner's transcript and shared situation. Treat erase, @@ -147,7 +154,12 @@ As observed on 29 July 2026, related capabilities are distributed across: - inspectable conversation-situation text and bounded exact observations on chat-history sessions, used as provisional per-conversation context rather - than publication or canonical knowledge; + than publication or canonical knowledge; the conversation UI exposes the + text, revision provenance, retention state, and exact retained observations; +- an authority-bound, bounded `chat_history_get_segments` projection that + carries the same situation and observations for telemetry consumers, while + the compact conversation locator reports only availability and freshness + metadata plus the route to that carrier-bearing read; - base concept and text relations in `concept_service.py`, `concept_relation_service.py`, and `text_value_service.py`; - actor- and organisation-visible assertion deltas in diff --git a/src/backend/integrations/internal_mcp/catalogue.py b/src/backend/integrations/internal_mcp/catalogue.py index c61087bcc..41a00213f 100644 --- a/src/backend/integrations/internal_mcp/catalogue.py +++ b/src/backend/integrations/internal_mcp/catalogue.py @@ -16125,6 +16125,64 @@ def _turn_execution_namespace_coverage_report(**kwargs): ) +_DELEGATED_TELEMETRY_STDIO_MAX_RESPONSE_CHARS_DEFAULT = 100_000 +_DELEGATED_TELEMETRY_STDIO_MAX_RESPONSE_CHARS_MIN = 10_000 +_DELEGATED_TELEMETRY_PAGE_CONTEXT_FIELDS = ( + "request_id", + "history_location", + "read_delegation", + "provenance", + "session_id", + "chat_session_id", + "history_owner_user_id", + "requested_user_id", + "namespace", + "access_mode", + "identifier_binding", +) + + +def _delegated_telemetry_json_default(value: Any) -> Any: + """Return a stable JSON representation for persisted BSON/Python values.""" + + if isinstance(value, datetime): + normalised = value + if normalised.tzinfo is None: + normalised = normalised.replace(tzinfo=timezone.utc) + return normalised.isoformat() + if isinstance(value, (set, frozenset)): + return sorted(value, key=lambda item: str(item)) + return str(value) + + +def _delegated_telemetry_stdio_max_response_chars() -> int: + """Mirror the stdio text guard without importing its catalogue consumer.""" + + raw_value = os.getenv("VON_MCP_STDIO_MAX_RESPONSE_CHARS") + if raw_value: + try: + parsed = int(raw_value) + except ValueError: + parsed = _DELEGATED_TELEMETRY_STDIO_MAX_RESPONSE_CHARS_DEFAULT + return max(_DELEGATED_TELEMETRY_STDIO_MAX_RESPONSE_CHARS_MIN, parsed) + return _DELEGATED_TELEMETRY_STDIO_MAX_RESPONSE_CHARS_DEFAULT + + +def _delegated_telemetry_stdio_response_chars(payload: Mapping[str, Any]) -> int: + """Measure the exact JSON text length checked by the stdio response guard.""" + + import json + + return len( + json.dumps( + dict(payload), + indent=2, + ensure_ascii=True, + default=_delegated_telemetry_json_default, + ) + ) + + def _bounded_delegated_telemetry_payload( payload: Mapping[str, Any], *, @@ -16162,39 +16220,85 @@ def _bounded_delegated_telemetry_payload( sort_keys=True, separators=(",", ":"), ensure_ascii=True, + default=_delegated_telemetry_json_default, ) + canonical_payload = json.loads(canonical_json) 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", + response_limit = _delegated_telemetry_stdio_max_response_chars() + + page_context: dict[str, Any] = { + "success": True, "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: + for field_name in _DELEGATED_TELEMETRY_PAGE_CONTEXT_FIELDS: + if field_name in canonical_payload or field_name in { + "request_id", + "history_location", + "read_delegation", + "provenance", + }: + page_context[field_name] = canonical_payload.get(field_name) + + def _page_for_end(end: int) -> dict[str, Any]: return { - **dict(payload), + "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 ""), + } + + def _paged_response_for_end(end: int) -> dict[str, Any]: + return { + **page_context, + "bounded_read": _page_for_end(end), + } + + if preserve_inline_below_limit and offset == 0 and total_chars <= limit: + inline_page = _page_for_end(total_chars) + inline_payload = { + **canonical_payload, "bounded_read": { - key: value for key, value in page.items() if key != "json_chunk" + key: value for key, value in inline_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, - } + if _delegated_telemetry_stdio_response_chars(inline_payload) <= response_limit: + return inline_payload + + requested_end = min(total_chars, offset + limit) + requested_page = _paged_response_for_end(requested_end) + if _delegated_telemetry_stdio_response_chars(requested_page) <= response_limit: + return requested_page + + low = min(offset, total_chars) + high = requested_end + while low < high: + candidate_end = (low + high + 1) // 2 + candidate_page = _paged_response_for_end(candidate_end) + if _delegated_telemetry_stdio_response_chars(candidate_page) <= response_limit: + low = candidate_end + else: + high = candidate_end - 1 + + if low <= offset and offset < total_chars: + return { + **page_context, + "success": False, + "error_code": "telemetry_page_envelope_too_large", + "message": ( + "Authority-bound telemetry page metadata exceeds the configured " + "stdio response guard." + ), + "response_guard_chars": response_limit, + } + return _paged_response_for_end(low) def _chat_history_get_segments(**kwargs): @@ -16258,6 +16362,7 @@ def _chat_history_get_segments(**kwargs): include_debug=include_debug, history_tail_limit=effective_kwargs.get("history_tail_limit"), return_meta=True, + include_conversation_state=True, ) except chat_history_service.ChatHistoryServiceError as exc: return make_error_response( @@ -16274,6 +16379,26 @@ def _chat_history_get_segments(**kwargs): else: segments, meta = result, {"history_truncated": False} + meta_mapping = meta if isinstance(meta, Mapping) else {} + raw_situation = meta_mapping.get("conversation_situation") + conversation_situation = ( + dict(raw_situation) if isinstance(raw_situation, Mapping) else None + ) + raw_observations = meta_mapping.get("conversation_observations") + conversation_observations = ( + list(raw_observations) if isinstance(raw_observations, list) else [] + ) + raw_observation_state = meta_mapping.get("conversation_observation_state") + if isinstance(raw_observation_state, Mapping): + conversation_observation_state = dict(raw_observation_state) + else: + conversation_observation_state = { + "schema_version": "conversation_observation_state.v1", + "retained_count": len(conversation_observations), + "total_count": len(conversation_observations), + "omitted_count": 0, + "retention_limit": chat_history_service.CONVERSATION_OBSERVATION_MAX_ITEMS, + } payload = { "success": True, "session_id": access.get("session_id"), @@ -16285,17 +16410,42 @@ def _chat_history_get_segments(**kwargs): "identifier_binding": access.get("identifier_binding"), "segments": segments, "segment_count": len(segments) if isinstance(segments, list) else 0, - "history_truncated": bool( - meta.get("history_truncated") if isinstance(meta, Mapping) else False - ), + "history_truncated": bool(meta_mapping.get("history_truncated", False)), + "conversation_situation": conversation_situation, + "conversation_observations": conversation_observations, + "conversation_observation_state": conversation_observation_state, } if authorisation.get("delegated"): payload["read_delegation"] = authorisation.get("read_delegation") - return _with_rag_provenance( + provenanced_payload = _with_rag_provenance( payload=payload, item_kind="chat_history_segments", source_system="mongo.chat_history", ) + bounded_payload = _bounded_delegated_telemetry_payload( + provenanced_payload, + arguments=effective_kwargs, + delegated=bool(authorisation.get("delegated")), + artifact_kind="chat_history_segments", + preserve_inline_below_limit=True, + ) + if authorisation.get("delegated"): + for field_name in ( + "session_id", + "chat_session_id", + "history_owner_user_id", + "requested_user_id", + "namespace", + "access_mode", + "identifier_binding", + "read_delegation", + ): + if field_name in provenanced_payload: + bounded_payload.setdefault( + field_name, + provenanced_payload[field_name], + ) + return bounded_payload def _chat_history_get_debug_entry(**kwargs): @@ -35704,6 +35854,8 @@ def _build_default_catalogue_diagnostics_and_research_definitions() -> ( "history_tail_limit": (int, type(None)), "include_debug": (bool,), "include_legacy": (bool,), + "offset": (int,), + "limit": (int,), }, allow_unknown=True, description=( @@ -35714,9 +35866,10 @@ def _build_default_catalogue_diagnostics_and_research_definitions() -> ( output_schema=None, category="read", description=( - "Fetch the stored conversation transcript in segment form, with optional embedded " - "assistant llm_debug_data and history_location locators. Prefer conversation_ref " - "over raw session_id on model-driven paths." + "Fetch the bounded stored conversation carrier: transcript segments, inspectable " + "situation text, exact observations, and optional assistant llm_debug_data with " + "history_location locators. Prefer conversation_ref over raw session_id on " + "model-driven paths; use offset/limit when a delegated response is paged." ), ), MethodDefinition( @@ -35771,9 +35924,10 @@ def _build_default_catalogue_diagnostics_and_research_definitions() -> ( output_schema=None, category="read", description=( - "Build the compact conversation telemetry locator so agents can fetch the same " - "stored conversation/turn diagnostics without pasting large JSON blobs. Prefer " - "conversation_ref over raw session_id on model-driven paths." + "Build compact conversation telemetry and carrier-freshness metadata, with an " + "authority-bound descriptor for fetching the full conversation carrier through " + "chat_history_get_segments. Prefer conversation_ref over raw session_id on " + "model-driven paths." ), ), MethodDefinition( diff --git a/src/backend/mcp_server/vontology_mcp.json b/src/backend/mcp_server/vontology_mcp.json index 630b42817..52d6f1b2a 100644 --- a/src/backend/mcp_server/vontology_mcp.json +++ b/src/backend/mcp_server/vontology_mcp.json @@ -228,7 +228,7 @@ }, { "name": "chat_history_get_segments", - "description": "Fetch the stored conversation transcript in segment form, with optional embedded assistant llm_debug_data and history_location locators. Prefer conversation_ref over raw session_id on model-driven paths.", + "description": "Fetch the bounded stored conversation carrier: transcript segments, inspectable situation text, exact observations, and optional assistant llm_debug_data with history_location locators. Prefer conversation_ref over raw session_id on model-driven paths; use offset/limit when a delegated response is paged.", "inputSchema": { "type": "object", "properties": { @@ -277,6 +277,12 @@ }, "include_legacy": { "type": "boolean" + }, + "offset": { + "type": "integer" + }, + "limit": { + "type": "integer" } }, "required": [], @@ -751,7 +757,7 @@ }, { "name": "conversation_telemetry_get_locator", - "description": "Build the compact conversation telemetry locator so agents can fetch the same stored conversation/turn diagnostics without pasting large JSON blobs. Prefer conversation_ref over raw session_id on model-driven paths.", + "description": "Build compact conversation telemetry and carrier-freshness metadata, with an authority-bound descriptor for fetching the full conversation carrier through chat_history_get_segments. Prefer conversation_ref over raw session_id on model-driven paths.", "inputSchema": { "type": "object", "properties": { diff --git a/src/backend/server/routes/generate_route_support.py b/src/backend/server/routes/generate_route_support.py index db9f28e33..67f8459e1 100644 --- a/src/backend/server/routes/generate_route_support.py +++ b/src/backend/server/routes/generate_route_support.py @@ -433,6 +433,9 @@ def _build_generate_success_body( llm_debug_info: Mapping[str, Any], display_elements_contract: Mapping[str, Any] | None, rag_trace: Any, + conversation_situation: Mapping[str, Any] | None = None, + conversation_observations: Sequence[Mapping[str, Any]] | None = None, + conversation_observation_state: Mapping[str, Any] | None = None, ) -> dict[str, Any]: return { "success": success, @@ -461,6 +464,21 @@ def _build_generate_success_body( ), "display_elements": display_elements_contract, "rag_trace": rag_trace, + "conversation_situation": ( + dict(conversation_situation) + if isinstance(conversation_situation, Mapping) + else None + ), + "conversation_observations": [ + dict(observation) + for observation in (conversation_observations or ()) + if isinstance(observation, Mapping) + ], + "conversation_observation_state": ( + dict(conversation_observation_state) + if isinstance(conversation_observation_state, Mapping) + else None + ), } diff --git a/src/backend/server/routes/von_routes.py b/src/backend/server/routes/von_routes.py index e2ea9b934..cacf29b9b 100644 --- a/src/backend/server/routes/von_routes.py +++ b/src/backend/server/routes/von_routes.py @@ -9286,6 +9286,85 @@ def _persist_conversation_situation_fail_soft( ) +def _read_conversation_carrier_after_persistence_fail_soft( + *, + user_id: str | None, + session_id: str, + namespace: str | None, + fallback_situation: Mapping[str, Any] | None, + fallback_observations: list[dict[str, Any]], + fallback_observation_state: Mapping[str, Any], + request_id: str | None = None, +) -> tuple[dict[str, Any] | None, list[dict[str, Any]], dict[str, Any]]: + """Read one canonical carrier snapshot, preserving the turn-start snapshot on failure.""" + + fallback_snapshot = ( + dict(fallback_situation) if isinstance(fallback_situation, Mapping) else None, + [ + dict(observation) + for observation in fallback_observations + if isinstance(observation, Mapping) + ], + dict(fallback_observation_state), + ) + if not isinstance(user_id, str) or not user_id.strip(): + return fallback_snapshot + + try: + session_state = chat_history_service.get_chat_history_session_state( + user_id=user_id.strip(), + session_id=session_id, + namespace=namespace, + include_history=False, + ) + if not isinstance(session_state, Mapping): + raise RuntimeError("canonical conversation session state was not found") + + conversation_situation, _, _ = ( + _normalise_conversation_situation_descriptor( + session_state.get("conversation_situation") + ) + ) + observations = ( + [ + dict(observation) + for observation in session_state.get( + "conversation_observations", [] + ) + if isinstance(observation, Mapping) + ] + if isinstance(session_state.get("conversation_observations"), list) + else [] + ) + raw_observation_state = session_state.get( + "conversation_observation_state" + ) + observation_state = ( + dict(raw_observation_state) + if isinstance(raw_observation_state, Mapping) + else { + "schema_version": "conversation_observation_state.v1", + "retained_count": len(observations), + "total_count": len(observations), + "omitted_count": 0, + "retention_limit": ( + chat_history_service.CONVERSATION_OBSERVATION_MAX_ITEMS + ), + } + ) + return conversation_situation, observations, observation_state + except Exception as exc: + _safe_app_log( + "warning", + "Conversation carrier read-back unavailable for session_id=%s " + "request_id=%s: %s", + session_id, + request_id, + exc, + ) + return fallback_snapshot + + def _namespace_is_org_scoped(namespace: str | None) -> bool: return ( isinstance(namespace, str) @@ -12819,6 +12898,19 @@ def _refresh_llm_debug_timing_payload(debug_payload: dict[str, Any]) -> None: updated_by=user_concept_id, request_id=request_id, ) + ( + response_conversation_situation, + response_conversation_observations, + response_conversation_observation_state, + ) = _read_conversation_carrier_after_persistence_fail_soft( + user_id=history_user_id, + session_id=session_id, + namespace=history_namespace, + fallback_situation=conversation_situation_descriptor, + fallback_observations=conversation_observations, + fallback_observation_state=conversation_observation_state, + request_id=request_id, + ) if progress_updates_enabled: final_progress_payload = { @@ -12868,6 +12960,11 @@ def _refresh_llm_debug_timing_payload(debug_payload: dict[str, Any]) -> None: llm_debug_info=llm_debug_info, display_elements_contract=display_elements_contract, rag_trace=rag_trace, + conversation_situation=response_conversation_situation, + conversation_observations=response_conversation_observations, + conversation_observation_state=( + response_conversation_observation_state + ), ) ) if adaptive_success: diff --git a/src/backend/services/browser_test_auth_service.py b/src/backend/services/browser_test_auth_service.py index fa76382e5..5d3adc117 100644 --- a/src/backend/services/browser_test_auth_service.py +++ b/src/backend/services/browser_test_auth_service.py @@ -655,6 +655,27 @@ def _chat_fixture_specs() -> Iterable[dict[str, Any]]: { "session_id": "browser-fixture-user-view-state", "session_name": "Authenticated user-view fixture sanity check", + "conversation_situation": { + "text": ( + "The user and Von are validating the authenticated browser " + "experience. Their shared goal is to keep saved conversations, " + "unread messages, organisation context, and the conversation " + "situation inspectable at desktop and mobile widths. Refreshing " + "this browser-test fixture must not duplicate its durable records." + ), + "source_request_id": "browser_user_view.v1:situation", + }, + "conversation_observations": ( + { + "observation_id": "browser_user_view.v1:fixture-ready", + "kind": "fixture_state", + "observed_at_utc": _BROWSER_TEST_FIXTURE_STABLE_SEEDED_AT, + "capability_name": "authenticated_browser_user_view", + "terminal_status": "ready", + "outcome_finality": "fixture", + "changed": False, + }, + ), "history": ( { "fixture_entry_id": "user-view-fixture-user-1", @@ -793,7 +814,11 @@ def _ensure_fixture_chat_session( with bypass_access_control(): doc = coll.find_one( {"user_id": user_concept_id, "session_id": session_id}, - {"history.fixture_entry_id": 1}, + { + "history.fixture_entry_id": 1, + "conversation_situation": 1, + "conversation_observations.observation_id": 1, + }, ) existing_ids: set[str] = set() @@ -822,10 +847,62 @@ def _ensure_fixture_chat_session( ) created_entries += 1 + situation_updated = False + situation_spec = spec.get("conversation_situation") + if isinstance(situation_spec, Mapping): + situation_text = str(situation_spec.get("text") or "").strip() + existing_situation = ( + doc.get("conversation_situation") if isinstance(doc, Mapping) else None + ) + existing_text = ( + str(existing_situation.get("text") or "").strip() + if isinstance(existing_situation, Mapping) + else "" + ) + existing_revision = ( + existing_situation.get("revision") + if isinstance(existing_situation, Mapping) + and isinstance(existing_situation.get("revision"), int) + and not isinstance(existing_situation.get("revision"), bool) + else 0 + ) + if situation_text and situation_text != existing_text: + result = chat_history_service.set_chat_history_conversation_situation( + user_id=user_concept_id, + session_id=session_id, + text=situation_text, + expected_revision=max(0, existing_revision), + source="browser_test_fixture", + updated_by=VON_SYSTEM_ID, + namespace=namespace, + include_legacy=False, + source_request_id=str( + situation_spec.get("source_request_id") + or f"{_BROWSER_TEST_FIXTURE_ID}:situation" + ), + ) + situation_updated = bool(result.get("updated")) + + observations_added = 0 + for observation in spec.get("conversation_observations") or (): + if not isinstance(observation, dict): + continue + result = chat_history_service.append_chat_history_conversation_observation( + user_id=user_concept_id, + session_id=session_id, + observation=observation, + namespace=namespace, + include_legacy=False, + ) + if result.get("updated"): + observations_added += 1 + return { "session_id": session_id, "session_name": session_name, "created_entries": created_entries, + "situation_updated": situation_updated, + "observations_added": observations_added, } diff --git a/src/backend/services/chat_history_service.py b/src/backend/services/chat_history_service.py index d9c61bfb8..f16ab096e 100644 --- a/src/backend/services/chat_history_service.py +++ b/src/backend/services/chat_history_service.py @@ -1606,53 +1606,15 @@ def get_chat_history_session_state( if not isinstance(doc, dict): return None - try: - conversation_situation = _conversation_situation_from_doc(doc) - except ChatHistoryServiceError as exc: - # A corrupt optional carrier must not make the canonical transcript - # unavailable. Writes remain strict and CAS will not overwrite the - # malformed field because its revision cannot match. - logger.warning( - "Ignoring malformed conversation situation for session_id=%s: %s", - session_id, - exc, - ) - conversation_situation = None history = ( _normalise_chat_history_entries(doc.get("history", [])) if include_history else [] ) - observations = _conversation_observations_from_doc(doc) - raw_observations = doc.get("conversation_observations") - raw_retained_count = ( - len(raw_observations) if isinstance(raw_observations, list) else 0 - ) - raw_observation_total = doc.get("conversation_observation_total") - observation_total = ( - raw_observation_total - if isinstance(raw_observation_total, int) - and not isinstance(raw_observation_total, bool) - and raw_observation_total >= 0 - else raw_retained_count - ) - observation_total = max( - observation_total, - raw_retained_count, - len(observations), - ) state = { "session_id": session_id, "history": history, - "conversation_situation": conversation_situation, - "conversation_observations": observations, - "conversation_observation_state": { - "schema_version": "conversation_observation_state.v1", - "retained_count": len(observations), - "total_count": observation_total, - "omitted_count": max(0, observation_total - len(observations)), - "retention_limit": CONVERSATION_OBSERVATION_MAX_ITEMS, - }, + **_conversation_state_from_doc(doc), } if ( include_history @@ -1989,6 +1951,50 @@ def _conversation_observations_from_doc( return observations +def _conversation_state_from_doc(doc: Dict[str, Any]) -> Dict[str, Any]: + """Return bounded conversation state without making an optional carrier brittle.""" + + try: + conversation_situation = _conversation_situation_from_doc(doc) + except ChatHistoryServiceError as exc: + logger.warning( + "Ignoring malformed conversation situation for session_id=%s: %s", + doc.get("session_id"), + exc, + ) + conversation_situation = None + + observations = _conversation_observations_from_doc(doc) + raw_observations = doc.get("conversation_observations") + raw_retained_count = ( + len(raw_observations) if isinstance(raw_observations, list) else 0 + ) + raw_observation_total = doc.get("conversation_observation_total") + observation_total = ( + raw_observation_total + if isinstance(raw_observation_total, int) + and not isinstance(raw_observation_total, bool) + and raw_observation_total >= 0 + else raw_retained_count + ) + observation_total = max( + observation_total, + raw_retained_count, + len(observations), + ) + return { + "conversation_situation": conversation_situation, + "conversation_observations": observations, + "conversation_observation_state": { + "schema_version": "conversation_observation_state.v1", + "retained_count": len(observations), + "total_count": observation_total, + "omitted_count": max(0, observation_total - len(observations)), + "retention_limit": CONVERSATION_OBSERVATION_MAX_ITEMS, + }, + } + + def append_chat_history_conversation_observation( *, user_id: str, @@ -2113,11 +2119,12 @@ def append_chat_history_conversation_observation( include_legacy=include_legacy, include_history=False, ) - observations = ( + raw_observations = ( current_state.get("conversation_observations") if isinstance(current_state, dict) else [] ) + observations = raw_observations if isinstance(raw_observations, list) else [] duplicate = any( isinstance(item, dict) and item.get("observation_id") == observation_id for item in observations @@ -2301,6 +2308,84 @@ def _hydrate_chat_history_entries(history: Any) -> List[Dict[str, Any]]: return _normalise_chat_history_entries(history, hydrate_blob_refs=True) +def _conversation_locator_state_from_doc(doc: Dict[str, Any]) -> Dict[str, Any]: + """Return compact carrier availability/freshness without exposing its content.""" + + raw_situation_state = doc.get("conversation_situation_state") + raw_situation = doc.get("conversation_situation") + if isinstance(raw_situation_state, dict): + situation_available = raw_situation_state.get("available") is True + raw_revision = raw_situation_state.get("revision") + raw_updated_at = raw_situation_state.get("updated_at") + else: + situation_available = bool( + isinstance(raw_situation, dict) + and isinstance(raw_situation.get("text"), str) + and raw_situation.get("text", "").strip() + ) + raw_revision = ( + raw_situation.get("revision") if isinstance(raw_situation, dict) else None + ) + raw_updated_at = ( + raw_situation.get("updated_at") if isinstance(raw_situation, dict) else None + ) + situation_revision = ( + raw_revision + if isinstance(raw_revision, int) + and not isinstance(raw_revision, bool) + and raw_revision >= 0 + else None + ) + situation_updated_at = _coerce_datetime(raw_updated_at) + + raw_observation_state = doc.get("conversation_observation_state") + raw_observations = doc.get("conversation_observations") + fallback_retained_count = ( + len(raw_observations) if isinstance(raw_observations, list) else 0 + ) + retained_candidate = ( + raw_observation_state.get("retained_count") + if isinstance(raw_observation_state, dict) + else fallback_retained_count + ) + retained_count = ( + retained_candidate + if isinstance(retained_candidate, int) + and not isinstance(retained_candidate, bool) + and retained_candidate >= 0 + else fallback_retained_count + ) + total_candidate = ( + raw_observation_state.get("total_count") + if isinstance(raw_observation_state, dict) + else doc.get("conversation_observation_total") + ) + total_count = ( + total_candidate + if isinstance(total_candidate, int) + and not isinstance(total_candidate, bool) + and total_candidate >= 0 + else retained_count + ) + total_count = max(total_count, retained_count) + return { + "conversation_situation_state": { + "available": situation_available, + "revision": situation_revision, + "updated_at": ( + situation_updated_at.isoformat() if situation_updated_at else None + ), + }, + "conversation_observation_state": { + "schema_version": "conversation_observation_state.v1", + "retained_count": retained_count, + "total_count": total_count, + "omitted_count": max(0, total_count - retained_count), + "retention_limit": CONVERSATION_OBSERVATION_MAX_ITEMS, + }, + } + + def get_chat_history_telemetry_locator_projection( user_id: str, session_id: str, @@ -2357,6 +2442,39 @@ def get_chat_history_telemetry_locator_projection( "created_at": 1, "updated_at": 1, "history": compact_history_projection, + "conversation_situation_state": { + "available": { + "$cond": [ + { + "$eq": [ + {"$type": "$conversation_situation.text"}, + "string", + ] + }, + { + "$gt": [ + {"$strLenCP": "$conversation_situation.text"}, + 0, + ] + }, + False, + ] + }, + "revision": "$conversation_situation.revision", + "updated_at": "$conversation_situation.updated_at", + }, + "conversation_observation_state": { + "retained_count": { + "$size": { + "$cond": [ + {"$isArray": "$conversation_observations"}, + "$conversation_observations", + [], + ] + } + }, + "total_count": "$conversation_observation_total", + }, } try: @@ -2386,15 +2504,25 @@ def get_chat_history_telemetry_locator_projection( ) else: # Test doubles and older collection adapters may not expose aggregate. + fallback_projection = { + key: value + for key, value in metadata_projection.items() + if key + not in { + "history", + "conversation_situation_state", + "conversation_observation_state", + } + } | { + "history": 1, + "conversation_situation": 1, + "conversation_observations": 1, + "conversation_observation_total": 1, + } doc = _read_find_one( chat_history_coll, query, - { - key: value - for key, value in metadata_projection.items() - if key != "history" - } - | {"history": 1}, + fallback_projection, operation=( "get_chat_history_telemetry_locator_projection.find_fallback" ), @@ -2402,7 +2530,14 @@ def get_chat_history_telemetry_locator_projection( if doc is not None: break _record_chat_history_read_success() - return dict(doc) if isinstance(doc, dict) else None + if not isinstance(doc, dict): + return None + payload = dict(doc) + payload.update(_conversation_locator_state_from_doc(payload)) + payload.pop("conversation_situation", None) + payload.pop("conversation_observations", None) + payload.pop("conversation_observation_total", None) + return payload except PyMongoError as exc: _record_chat_history_read_failure( "get_chat_history_telemetry_locator_projection", @@ -2430,12 +2565,14 @@ def get_chat_history_segments( history_tail_limit: Optional[int] = None, return_meta: bool = False, hydrate_blob_refs: bool = False, + include_conversation_state: bool = False, ) -> List[List[Dict[str, Any]]] | tuple[List[List[Dict[str, Any]]], Dict[str, Any]]: """ Return chat history split into segments separated by reset markers. Retrieves history from the requested session only. - When return_meta is True, returns (segments, {"history_truncated": bool}). + When return_meta is True, returns ``(segments, metadata)``. Callers may opt + into bounded conversation state in that metadata without another DB read. """ if not user_id: raise ChatHistoryServiceError("user_id is required.") @@ -2452,7 +2589,16 @@ def get_chat_history_segments( history_offset = 0 history_length = None doc = None - projection = None + projection = ( + { + "history": 1, + "conversation_situation": 1, + "conversation_observations": 1, + "conversation_observation_total": 1, + } + if include_conversation_state + else None + ) read_queries = _build_chat_history_session_read_queries( user_id=user_id, session_id=session_id, @@ -2464,17 +2610,24 @@ def get_chat_history_segments( if hasattr(chat_history_coll, "aggregate") and callable( getattr(chat_history_coll, "aggregate") ): + tail_projection = { + "history": _history_tail_projection_expr( + history_tail_limit=history_tail_limit, + include_debug=include_debug, + ), + "history_length": {"$size": _history_array_expr()}, + } + if include_conversation_state: + tail_projection.update( + { + "conversation_situation": 1, + "conversation_observations": 1, + "conversation_observation_total": 1, + } + ) pipeline = [ {"$match": query}, - { - "$project": { - "history": _history_tail_projection_expr( - history_tail_limit=history_tail_limit, - include_debug=include_debug, - ), - "history_length": {"$size": _history_array_expr()}, - } - }, + {"$project": tail_projection}, ] try: doc = next( @@ -2513,7 +2666,10 @@ def get_chat_history_segments( break if not doc: _record_chat_history_read_success() - return ([], {"history_truncated": False}) if return_meta else [] + empty_meta = {"history_truncated": False} + if include_conversation_state: + empty_meta.update(_conversation_state_from_doc({})) + return ([], empty_meta) if return_meta else [] history = _normalise_chat_history_entries( doc.get("history") or [], @@ -2521,7 +2677,10 @@ def get_chat_history_segments( ) if not isinstance(history, list) or not history: _record_chat_history_read_success() - return ([], {"history_truncated": False}) if return_meta else [] + empty_history_meta = {"history_truncated": False} + if include_conversation_state: + empty_history_meta.update(_conversation_state_from_doc(doc)) + return ([], empty_history_meta) if return_meta else [] if history_length is None: history_length_raw = doc.get("history_length") if isinstance(history_length_raw, int): @@ -2565,7 +2724,10 @@ def get_chat_history_segments( result = _chunk_history_segments(segments, segment_size) _record_chat_history_read_success() if return_meta: - return result, {"history_truncated": history_truncated} + result_meta = {"history_truncated": history_truncated} + if include_conversation_state: + result_meta.update(_conversation_state_from_doc(doc)) + return result, result_meta return result except PyMongoError as e: _record_chat_history_read_failure("get_chat_history_segments", e) diff --git a/src/backend/services/conversation_telemetry_locator_service.py b/src/backend/services/conversation_telemetry_locator_service.py index 3944f1c43..d7759c28f 100644 --- a/src/backend/services/conversation_telemetry_locator_service.py +++ b/src/backend/services/conversation_telemetry_locator_service.py @@ -425,6 +425,30 @@ def _build_bound_conversation_args(*, include_debug: bool | None = None) -> dict user_id=user_id_value, organisation_concept_id=resolved_org_id, ) + raw_situation_state = session_projection.get("conversation_situation_state") + conversation_situation_state = ( + dict(raw_situation_state) + if isinstance(raw_situation_state, Mapping) + else { + "available": False, + "revision": None, + "updated_at": None, + } + ) + raw_observation_state = session_projection.get("conversation_observation_state") + conversation_observation_state = ( + dict(raw_observation_state) + if isinstance(raw_observation_state, Mapping) + else { + "schema_version": "conversation_observation_state.v1", + "retained_count": 0, + "total_count": 0, + "omitted_count": 0, + "retention_limit": ( + chat_history_service.CONVERSATION_OBSERVATION_MAX_ITEMS + ), + } + ) return { "schema_version": CONVERSATION_LLM_TELEMETRY_LOCATOR_SCHEMA_VERSION, @@ -432,6 +456,8 @@ def _build_bound_conversation_args(*, include_debug: bool | None = None) -> dict "session_id": session_id_value, "session_name": _safe_str(session_projection.get("session_name")), "namespace_context": namespace_context, + "conversation_situation_state": conversation_situation_state, + "conversation_observation_state": conversation_observation_state, "metadata": metadata, "mcp_access": { "conversation_telemetry_get_locator": _build_tool_call_descriptor( @@ -449,8 +475,9 @@ def _build_bound_conversation_args(*, include_debug: bool | None = None) -> dict "history_tail_limit": 100, }, purpose=( - "Fetch a bounded transcript projection; use each turn's exact " - "debug-entry descriptor for persisted debug payloads." + "Fetch the bounded conversation carrier: transcript segments, " + "inspectable situation text, and exact observations. Use each " + "turn's debug-entry descriptor for persisted debug payloads." ), ), }, diff --git a/src/frontend/web/von_interface/static/js/chatTab.js b/src/frontend/web/von_interface/static/js/chatTab.js index 7ccb0c25f..9a28965d3 100644 --- a/src/frontend/web/von_interface/static/js/chatTab.js +++ b/src/frontend/web/von_interface/static/js/chatTab.js @@ -111,6 +111,7 @@ const CONVERSATION_INFO_EXPORT_SCHEMA_VERSION = 'conversation_info_export.v1'; const CONVERSATION_TELEMETRY_ACCESS_SCHEMA_VERSION = 'conversation_telemetry_access.v1'; const CONVERSATION_LLM_TELEMETRY_SCHEMA_VERSION = 'conversation_llm_telemetry.v1'; const CONVERSATION_LLM_TELEMETRY_LOCATOR_SCHEMA_VERSION = 'conversation_llm_telemetry_locator.v1'; +const CONVERSATION_SITUATION_EXPORT_SCHEMA_VERSION = 'conversation_situation_export.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'; @@ -148,6 +149,8 @@ const historyMetricsState = { lastHealthy: null }; const sessionHistoryCache = new Map(); +const conversationSituationStateBySession = new Map(); +let conversationSituationRefreshRequestCounter = 0; let loadingChatSessionId = null; const SESSION_TABS_REFRESH_COOLDOWN_MS = 15_000; let lastSessionTabsRefreshMs = 0; @@ -3990,6 +3993,682 @@ function createClientRequestId() { return `req-${Date.now()}-${Math.random().toString(16).slice(2)}`; } +function _normaliseConversationSituationInteger(value, fallback = 0) { + return Number.isInteger(value) && value >= 0 ? value : fallback; +} + +function _normaliseConversationSituationDescriptor(value) { + if (!value || typeof value !== 'object' || Array.isArray(value)) { + return null; + } + const descriptor = { + text: (typeof value.text === 'string' && value.text.trim()) ? value.text : null, + revision: _normaliseConversationSituationInteger(value.revision), + source: (typeof value.source === 'string' && value.source.trim()) ? value.source.trim() : null, + updated_by: (typeof value.updated_by === 'string' && value.updated_by.trim()) + ? value.updated_by.trim() + : null, + updated_at: (typeof value.updated_at === 'string' && value.updated_at.trim()) + ? value.updated_at.trim() + : null + }; + if (typeof value.source_request_id === 'string' && value.source_request_id.trim()) { + descriptor.source_request_id = value.source_request_id.trim(); + } + return descriptor; +} + +function _normaliseConversationObservation(value) { + if (!value || typeof value !== 'object' || Array.isArray(value)) { + return null; + } + const observation = {}; + Object.entries(value).forEach(([key, rawValue]) => { + if ( + typeof rawValue === 'string' + || typeof rawValue === 'boolean' + || (typeof rawValue === 'number' && Number.isFinite(rawValue)) + || rawValue === null + ) { + observation[key] = rawValue; + } + }); + const observationId = typeof observation.observation_id === 'string' + ? observation.observation_id.trim() + : ''; + const kind = typeof observation.kind === 'string' ? observation.kind.trim() : ''; + if (!observationId || !kind) { + return null; + } + observation.observation_id = observationId; + observation.kind = kind; + return observation; +} + +function _normaliseConversationObservationState(value, retainedCount) { + const fallbackRetained = _normaliseConversationSituationInteger(retainedCount); + const source = value && typeof value === 'object' && !Array.isArray(value) ? value : {}; + const retained = _normaliseConversationSituationInteger( + source.retained_count, + fallbackRetained + ); + const total = Math.max( + retained, + _normaliseConversationSituationInteger(source.total_count, retained) + ); + return { + schema_version: ( + typeof source.schema_version === 'string' + && source.schema_version.trim() + ) + ? source.schema_version.trim() + : 'conversation_observation_state.v1', + retained_count: retained, + total_count: total, + omitted_count: Math.max( + 0, + _normaliseConversationSituationInteger( + source.omitted_count, + total - retained + ) + ), + retention_limit: _normaliseConversationSituationInteger( + source.retention_limit, + Math.max(retained, 12) + ) + }; +} + +function _conversationSituationPayloadHasCarrierState(payload) { + if (!payload || typeof payload !== 'object') { + return false; + } + return Object.prototype.hasOwnProperty.call(payload, 'conversation_situation') + || Object.prototype.hasOwnProperty.call(payload, 'conversation_observations') + || Object.prototype.hasOwnProperty.call(payload, 'conversation_observation_state'); +} + +function _conversationSituationPayloadHasCompleteCarrierState(payload) { + return _conversationSituationPayloadHasCarrierState(payload) + && Object.prototype.hasOwnProperty.call(payload, 'conversation_situation') + && Object.prototype.hasOwnProperty.call(payload, 'conversation_observations') + && Object.prototype.hasOwnProperty.call(payload, 'conversation_observation_state'); +} + +function _acceptConversationSituationPayload(payload, sessionId, options = {}) { + const sid = normaliseHistorySessionId(sessionId); + if (!sid || !_conversationSituationPayloadHasCarrierState(payload)) { + return null; + } + const authoritativeSnapshot = options.authoritativeSnapshot === true; + + const incomingSituation = _normaliseConversationSituationDescriptor( + payload.conversation_situation + ); + const incomingObservations = Array.isArray(payload.conversation_observations) + ? payload.conversation_observations + .map(_normaliseConversationObservation) + .filter(Boolean) + : []; + const incomingObservationState = _normaliseConversationObservationState( + payload.conversation_observation_state, + incomingObservations.length + ); + const existing = conversationSituationStateBySession.get(sid); + const existingRevision = _normaliseConversationSituationInteger( + existing?.situation?.revision + ); + const incomingRevision = _normaliseConversationSituationInteger( + incomingSituation?.revision + ); + const existingObservationTotal = _normaliseConversationSituationInteger( + existing?.observationState?.total_count + ); + const canonicalSituationRevision = Number.isInteger( + existing?.canonicalSituationRevision + ) + ? existing.canonicalSituationRevision + : null; + const incomingPredatesCanonicalSituation = ( + !authoritativeSnapshot + && canonicalSituationRevision !== null + && incomingRevision < canonicalSituationRevision + ); + const useIncomingSituation = ( + authoritativeSnapshot + || !existing + || incomingRevision >= existingRevision + ); + const useIncomingObservations = ( + !incomingPredatesCanonicalSituation + && ( + authoritativeSnapshot + || !existing + || incomingObservationState.total_count >= existingObservationTotal + ) + ); + + const accepted = { + sessionId: sid, + // Asynchronous generate responses merge their monotonic dimensions + // independently so a slow response cannot regress either half of the + // carrier. Guarded canonical history reads replace the whole snapshot, + // including deliberate resets to a lower revision or total. + situation: useIncomingSituation + ? incomingSituation + : existing.situation, + observations: useIncomingObservations + ? incomingObservations + : existing.observations, + observationState: useIncomingObservations + ? incomingObservationState + : existing.observationState, + canonicalSituationRevision: authoritativeSnapshot + ? incomingRevision + : canonicalSituationRevision, + availability: 'ready', + notice: typeof options.notice === 'string' ? options.notice : '', + loadedAtMs: Date.now() + }; + conversationSituationStateBySession.set(sid, accepted); + if (sid === activeChatSessionId) { + _renderConversationSituationForActiveSession(); + } + return accepted; +} + +function _markConversationSituationLoading(sessionId, message = 'Loading the shared situation…') { + const sid = normaliseHistorySessionId(sessionId); + if (!sid) { + return null; + } + const existing = conversationSituationStateBySession.get(sid); + const nextState = existing + ? { + ...existing, + availability: 'refreshing', + notice: message + } + : { + sessionId: sid, + situation: null, + observations: [], + observationState: _normaliseConversationObservationState(null, 0), + availability: 'loading', + notice: message, + loadedAtMs: null + }; + conversationSituationStateBySession.set(sid, nextState); + if (sid === activeChatSessionId) { + _renderConversationSituationForActiveSession(); + } + return nextState; +} + +function _markConversationSituationUnavailable(sessionId, message) { + const sid = normaliseHistorySessionId(sessionId); + if (!sid) { + return null; + } + const existing = conversationSituationStateBySession.get(sid); + const hasRetainedState = Boolean( + existing?.situation + || (Array.isArray(existing?.observations) && existing.observations.length > 0) + ); + const nextState = { + sessionId: sid, + situation: existing?.situation || null, + observations: Array.isArray(existing?.observations) ? existing.observations : [], + observationState: existing?.observationState + || _normaliseConversationObservationState(null, 0), + canonicalSituationRevision: Number.isInteger( + existing?.canonicalSituationRevision + ) + ? existing.canonicalSituationRevision + : null, + availability: hasRetainedState ? 'stale' : 'unavailable', + notice: message || 'The shared situation is temporarily unavailable.', + loadedAtMs: existing?.loadedAtMs || null + }; + conversationSituationStateBySession.set(sid, nextState); + if (sid === activeChatSessionId) { + _renderConversationSituationForActiveSession(); + } + return nextState; +} + +function _setConversationSituationPanelOpen(isOpen) { + const panel = document.getElementById('conversationSituationPanel'); + const toggle = document.getElementById('conversationSituationToggleBtn'); + const open = Boolean(isOpen); + if (panel) { + panel.classList.toggle('hidden', !open); + } + if (toggle) { + toggle.setAttribute('aria-expanded', String(open)); + } +} + +function _humaniseConversationSituationField(value) { + const cleaned = String(value || '').trim().replace(/[_-]+/g, ' '); + if (!cleaned) { + return 'Observation'; + } + return cleaned.charAt(0).toUpperCase() + cleaned.slice(1); +} + +function _formatConversationSituationValue(key, value) { + if (typeof value === 'boolean') { + return value ? 'Yes' : 'No'; + } + if (value === null || value === undefined || value === '') { + return '—'; + } + if ( + typeof value === 'string' + && ( + key.endsWith('_at') + || key.endsWith('_at_utc') + || key === 'updated_at' + || key === 'completed_at' + ) + ) { + return formatAbsoluteTimestamp(value) || value; + } + return String(value); +} + +function _appendConversationSituationDefinitionRow(list, label, value, key = '') { + if (!list || value === null || value === undefined || value === '') { + return; + } + const term = document.createElement('dt'); + term.textContent = label; + const description = document.createElement('dd'); + description.textContent = _formatConversationSituationValue(key, value); + if (typeof value === 'string' && description.textContent !== value) { + description.title = value; + } + list.appendChild(term); + list.appendChild(description); +} + +function _renderConversationObservation(observation) { + const card = document.createElement('article'); + card.className = 'chat-situation-observation'; + + const title = document.createElement('h4'); + title.className = 'chat-situation-observation-title'; + title.textContent = _humaniseConversationSituationField(observation.kind); + card.appendChild(title); + + const fields = document.createElement('dl'); + fields.className = 'chat-situation-observation-fields'; + const preferredKeys = [ + 'observation_id', + 'observed_at_utc', + 'terminal_status', + 'effect_status', + 'outcome_finality', + 'final_state', + 'capability_name', + 'workflow_id', + 'execution_id', + 'instance_id', + 'request_id', + 'effect_id', + 'call_id', + 'completed_at', + 'execution_trace_id', + 'changed', + 'failure_detail_available' + ]; + const excludedKeys = new Set(['schema_version', 'kind']); + const orderedKeys = [ + ...preferredKeys.filter(key => Object.prototype.hasOwnProperty.call(observation, key)), + ...Object.keys(observation) + .filter(key => !preferredKeys.includes(key) && !excludedKeys.has(key)) + .sort() + ]; + orderedKeys.forEach((key) => { + _appendConversationSituationDefinitionRow( + fields, + _humaniseConversationSituationField(key), + observation[key], + key + ); + }); + card.appendChild(fields); + return card; +} + +function _renderConversationSituationForActiveSession() { + const body = document.getElementById('conversationSituationBody'); + const status = document.getElementById('conversationSituationStatus'); + const badge = document.getElementById('conversationSituationBadge'); + const refreshButton = document.getElementById('conversationSituationRefreshBtn'); + const copyButton = document.getElementById('conversationSituationCopyBtn'); + if (!body) { + return; + } + body.textContent = ''; + + const sid = normaliseHistorySessionId(activeChatSessionId); + const state = sid ? conversationSituationStateBySession.get(sid) : null; + const revision = _normaliseConversationSituationInteger(state?.situation?.revision); + const retainedCount = _normaliseConversationSituationInteger( + state?.observationState?.retained_count, + state?.observations?.length || 0 + ); + const hasCarrierContent = Boolean(state?.situation || retainedCount > 0); + const isBusy = Boolean( + sid + && (!state || ['loading', 'refreshing'].includes(state.availability)) + ); + + body.setAttribute('aria-busy', String(isBusy)); + if (status) { + if (!sid) { + status.textContent = ''; + } else if (!state || state.availability === 'loading') { + status.textContent = state?.notice || 'Loading the shared situation…'; + } else if (state.availability === 'refreshing') { + status.textContent = state.notice || 'Refreshing the shared situation…'; + } else if ( + state.availability === 'stale' + || state.availability === 'unavailable' + ) { + status.textContent = ( + state.notice + || 'The shared situation is temporarily unavailable.' + ); + } else { + status.textContent = state.notice || ''; + } + } + + if (badge) { + if (hasCarrierContent) { + badge.textContent = revision > 0 ? `r${revision}` : String(retainedCount); + badge.classList.remove('hidden'); + badge.title = revision > 0 + ? `Conversation situation revision ${revision}` + : `${retainedCount} retained observations`; + } else { + badge.textContent = ''; + badge.classList.add('hidden'); + badge.removeAttribute('title'); + } + } + if (refreshButton) { + refreshButton.disabled = !sid || ['loading', 'refreshing'].includes(state?.availability); + } + if (copyButton) { + copyButton.disabled = !sid || !state || state.availability === 'loading'; + } + + if (!sid) { + const empty = document.createElement('p'); + empty.className = 'chat-situation-empty'; + empty.textContent = 'Select or start a conversation to inspect its shared situation.'; + body.appendChild(empty); + return; + } + + if (!state || state.availability === 'loading') { + const loading = document.createElement('p'); + loading.className = 'chat-situation-notice'; + loading.textContent = state?.notice || 'Loading the shared situation…'; + body.appendChild(loading); + return; + } + + if (state.availability === 'refreshing') { + const refreshing = document.createElement('p'); + refreshing.className = 'chat-situation-notice'; + refreshing.textContent = state.notice || 'Refreshing the shared situation…'; + body.appendChild(refreshing); + } else if (state.availability === 'stale' || state.availability === 'unavailable') { + const warning = document.createElement('p'); + warning.className = 'chat-situation-notice is-warning'; + warning.textContent = state.notice || 'The shared situation is temporarily unavailable.'; + body.appendChild(warning); + } + + const situation = state.situation; + if (situation?.text) { + const textSection = document.createElement('section'); + const label = document.createElement('h4'); + label.className = 'chat-situation-section-label'; + label.textContent = 'Current description'; + const text = document.createElement('pre'); + text.className = 'chat-situation-text'; + text.textContent = situation.text; + textSection.appendChild(label); + textSection.appendChild(text); + body.appendChild(textSection); + } else { + const empty = document.createElement('p'); + empty.className = 'chat-situation-empty'; + empty.textContent = revision > 0 + ? 'This conversation currently has no situation description; the prior description was cleared.' + : 'No shared situation description has been recorded for this conversation yet.'; + body.appendChild(empty); + } + + if (situation) { + const metadataSection = document.createElement('section'); + const label = document.createElement('h4'); + label.className = 'chat-situation-section-label'; + label.textContent = 'Revision and provenance'; + const metadata = document.createElement('dl'); + metadata.className = 'chat-situation-metadata'; + _appendConversationSituationDefinitionRow( + metadata, + 'Revision', + situation.revision, + 'revision' + ); + _appendConversationSituationDefinitionRow(metadata, 'Source', situation.source, 'source'); + _appendConversationSituationDefinitionRow( + metadata, + 'Updated by', + situation.updated_by, + 'updated_by' + ); + _appendConversationSituationDefinitionRow( + metadata, + 'Updated', + situation.updated_at, + 'updated_at' + ); + _appendConversationSituationDefinitionRow( + metadata, + 'Source request', + situation.source_request_id, + 'source_request_id' + ); + metadataSection.appendChild(label); + metadataSection.appendChild(metadata); + body.appendChild(metadataSection); + } + + const observations = Array.isArray(state.observations) ? state.observations : []; + const observationState = state.observationState + || _normaliseConversationObservationState(null, observations.length); + const details = document.createElement('details'); + details.className = 'chat-situation-observations'; + const summary = document.createElement('summary'); + summary.textContent = `Recent exact observations (${observationState.retained_count})`; + const summaryMeta = document.createElement('span'); + summaryMeta.className = 'chat-situation-observation-summary-meta'; + summaryMeta.textContent = `${observationState.total_count} total · limit ${observationState.retention_limit}`; + summary.appendChild(summaryMeta); + details.appendChild(summary); + + const observationList = document.createElement('div'); + observationList.className = 'chat-situation-observation-list'; + if (observationState.omitted_count > 0) { + const omission = document.createElement('p'); + omission.className = 'chat-situation-observation-omission'; + omission.textContent = `${observationState.omitted_count} older observation${observationState.omitted_count === 1 ? ' was' : 's were'} omitted by retention.`; + observationList.appendChild(omission); + } + if (observations.length === 0) { + const empty = document.createElement('p'); + empty.className = 'chat-situation-empty'; + empty.textContent = 'No exact observations are currently retained.'; + observationList.appendChild(empty); + } else { + observations.forEach(observation => { + observationList.appendChild(_renderConversationObservation(observation)); + }); + } + details.appendChild(observationList); + body.appendChild(details); +} + +function _buildConversationSituationExportPayload(sessionId = activeChatSessionId) { + const sid = normaliseHistorySessionId(sessionId); + const state = sid ? conversationSituationStateBySession.get(sid) : null; + return { + schema_version: CONVERSATION_SITUATION_EXPORT_SCHEMA_VERSION, + generated_at_utc: new Date().toISOString(), + session_id: sid, + session_name: sid === activeChatSessionId ? activeChatSessionName : null, + availability: state?.availability || 'unavailable', + conversation_situation: state?.situation || null, + conversation_observations: state?.observations || [], + conversation_observation_state: state?.observationState || null + }; +} + +async function _copyActiveConversationSituation(button) { + if (!(button instanceof HTMLButtonElement)) { + return false; + } + const payload = _buildConversationSituationExportPayload(); + return copyJsonTextWithButtonFeedback( + button, + JSON.stringify(payload, null, 2), + { fallbackLabel: 'Copy JSON' } + ); +} + +async function refreshConversationSituationForSession(sessionId = activeChatSessionId) { + const sid = normaliseHistorySessionId(sessionId); + if (!sid) { + _renderConversationSituationForActiveSession(); + return false; + } + + const requestId = ++conversationSituationRefreshRequestCounter; + _markConversationSituationLoading(sid, 'Refreshing the shared situation…'); + try { + await _ensureInviteSessionContext(); + const userContext = getUserContext(); + const params = new URLSearchParams({ + session_id: sid, + segments: '1', + segment_size: '1', + tail_limit: '1', + include_debug: '0' + }); + if (userContext?.user_id) { + params.append('user_id', userContext.user_id); + } + const response = await fetch(`/von/history?${params.toString()}`, { + headers: buildChatFetchHeaders() + }); + const data = await response.json().catch(() => ({})); + if (requestId !== conversationSituationRefreshRequestCounter) { + return false; + } + if (!response.ok || data?.degraded === true || data?.authenticated === false) { + const message = data?.authenticated === false + ? 'Sign in to inspect this conversation situation.' + : describeHistoryUnavailable(data); + _markConversationSituationUnavailable(sid, message); + return false; + } + if (!_conversationSituationPayloadHasCompleteCarrierState(data)) { + _markConversationSituationUnavailable( + sid, + 'The server did not return conversation situation state; try refreshing again.' + ); + return false; + } + return Boolean(_acceptConversationSituationPayload(data, sid, { + authoritativeSnapshot: true, + notice: 'Conversation situation refreshed.' + })); + } catch (error) { + if (requestId !== conversationSituationRefreshRequestCounter) { + return false; + } + _markConversationSituationUnavailable( + sid, + 'The shared situation could not be refreshed; try again.' + ); + console.warn('[chatTab] Unable to refresh conversation situation:', error); + return false; + } +} + +function _clearConversationSituationState(options = {}) { + conversationSituationRefreshRequestCounter += 1; + conversationSituationStateBySession.clear(); + if (options.closePanel === true) { + _setConversationSituationPanelOpen(false); + } + _renderConversationSituationForActiveSession(); +} + +function initializeConversationSituationPanel() { + const toggle = document.getElementById('conversationSituationToggleBtn'); + const closeButton = document.getElementById('conversationSituationCloseBtn'); + const refreshButton = document.getElementById('conversationSituationRefreshBtn'); + const copyButton = document.getElementById('conversationSituationCopyBtn'); + if (!toggle || toggle.dataset.situationPanelWired === 'true') { + _renderConversationSituationForActiveSession(); + return; + } + toggle.dataset.situationPanelWired = 'true'; + toggle.addEventListener('click', () => { + const willOpen = toggle.getAttribute('aria-expanded') !== 'true'; + _setConversationSituationPanelOpen(willOpen); + _renderConversationSituationForActiveSession(); + if (willOpen && activeChatSessionId) { + void refreshConversationSituationForSession(activeChatSessionId); + } + if (willOpen) { + document.getElementById('conversationSituationPanel')?.focus({ + preventScroll: true + }); + } + }); + closeButton?.addEventListener('click', () => { + _setConversationSituationPanelOpen(false); + toggle.focus(); + }); + refreshButton?.addEventListener('click', () => { + void refreshConversationSituationForSession(activeChatSessionId); + }); + copyButton?.addEventListener('click', () => { + void _copyActiveConversationSituation(copyButton); + }); + document.addEventListener('keydown', (event) => { + if ( + event.key === 'Escape' + && toggle.getAttribute('aria-expanded') === 'true' + ) { + _setConversationSituationPanelOpen(false); + toggle.focus(); + } + }); + _renderConversationSituationForActiveSession(); +} + function updateSessionHistoryCache(sessionId, history, meta = {}) { const sid = String(sessionId || '').trim(); if (!sid || !Array.isArray(history)) { @@ -21622,6 +22301,18 @@ function setActiveChatSession(sessionId, sessionName) { if (previousSessionId !== activeChatSessionId) { synchroniseLlmExecutionContext({ reason: 'conversation_session_changed' }); + if (activeChatSessionId) { + if (conversationSituationStateBySession.has(activeChatSessionId)) { + _renderConversationSituationForActiveSession(); + } else { + _markConversationSituationLoading( + activeChatSessionId, + 'Loading the selected conversation situation…' + ); + } + } else { + _renderConversationSituationForActiveSession(); + } } if (activeChatSessionId) { @@ -24094,6 +24785,23 @@ async function createChatSession(sessionName) { } setActiveChatSession(data?.session_id, data?.session_name); + _acceptConversationSituationPayload( + _conversationSituationPayloadHasCarrierState(data) + ? data + : { + conversation_situation: null, + conversation_observations: [], + conversation_observation_state: { + schema_version: 'conversation_observation_state.v1', + retained_count: 0, + total_count: 0, + omitted_count: 0, + retention_limit: 12 + } + }, + data?.session_id, + { authoritativeSnapshot: true } + ); syncActiveChatSessionThinkingState(); const nowIso = new Date().toISOString(); @@ -24963,6 +25671,7 @@ async function loadChatHistory(options = {}) { if (data?.degraded === true) { const degradedMessage = describeHistoryUnavailable(data); + _markConversationSituationUnavailable(targetSessionId, degradedMessage); setHistoryLoadState({ sessionId: targetSessionId, message: degradedMessage, @@ -24986,6 +25695,18 @@ async function loadChatHistory(options = {}) { return false; } + if (response.ok && _conversationSituationPayloadHasCompleteCarrierState(data)) { + _acceptConversationSituationPayload(data, targetSessionId, { + authoritativeSnapshot: true, + notice: 'Conversation situation loaded.' + }); + } else if (response.ok) { + _markConversationSituationUnavailable( + targetSessionId, + 'The server did not return conversation situation state; try refreshing again.' + ); + } + if (response.ok && Array.isArray(data?.history)) { const historyMessages = filterRecentPair ? selectRecentChatPair(data.history) @@ -25029,6 +25750,7 @@ async function loadChatHistory(options = {}) { const fallbackMessage = (typeof data?.error === 'string' && data.error.trim()) ? data.error.trim() : 'Conversation history request failed; please retry.'; + _markConversationSituationUnavailable(targetSessionId, fallbackMessage); setHistoryLoadState({ sessionId: targetSessionId, message: fallbackMessage, @@ -25068,6 +25790,7 @@ async function loadChatHistory(options = {}) { } const targetSessionId = normaliseHistorySessionId(requestedSessionId || activeChatSessionId); const fallbackMessage = 'Conversation history request failed; please retry.'; + _markConversationSituationUnavailable(targetSessionId, fallbackMessage); setHistoryLoadState({ sessionId: targetSessionId, message: fallbackMessage, @@ -28827,6 +29550,7 @@ async function handleOrgSwitchForChatTab(_detail) { lastRenderedSessionCount = 0; chatSessionMetadataOpenKey = null; _clearChatSessionMetadata(); + _clearConversationSituationState({ closePanel: true }); // JVNAUTOSCI-1011: Clear the active session and chat display on org switch // to prevent showing data from the previous org @@ -28903,6 +29627,7 @@ function handleAuthStatusChangeForChatTab(detail) { lastRenderedSessionCount = 0; chatSessionMetadataOpenKey = null; _clearChatSessionMetadata(); + _clearConversationSituationState({ closePanel: true }); if (container) { renderChatSessionTabsPlaceholder(detail?.authenticated ? 'loading' : 'unauthenticated'); @@ -29180,6 +29905,7 @@ export function initializeChatTab() { initializeLlmDebugPopup(); bindDiagnosticsExportShortcut(); initializeHistoryControls(); + initializeConversationSituationPanel(); updateHistoryBanner(); setupChatTabContextMenu(); @@ -31259,6 +31985,7 @@ async function handleSendPrompt(options = {}) { if (!claimForegroundDelivery()) { return false; } + _acceptConversationSituationPayload(data, request.sessionId); const responsePresenterState = resolveResponsePresenterState(data); const responseChannels = responsePresenterState.presenterChannels; const screenText = responsePresenterState.screenText; @@ -33879,7 +34606,11 @@ function buildConversationTelemetryAccessPayload({ locatorPayload = null } = {}) 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.', + ( + 'Call chat_history_get_segments for the bounded conversation-carrier ' + + 'projection, including the shared situation, retained exact observations, ' + + 'observation state, and transcript segments.' + ), 'Use each returned turn descriptor to fetch the exact bounded debug or diagnostics artefact.' ] : [ @@ -34270,6 +35001,21 @@ export function __testOnly_buildWorkflowMonitorLocatorPayload() { export async function __testOnly_loadChatHistory(options = {}) { return loadChatHistory(options); } +export function __testOnly_initializeConversationSituationPanel() { + initializeConversationSituationPanel(); +} +export function __testOnly_applyConversationSituationPayload(payload, sessionId, options = {}) { + return _acceptConversationSituationPayload(payload, sessionId, options); +} +export function __testOnly_buildConversationSituationExportPayload(sessionId = null) { + return _buildConversationSituationExportPayload(sessionId || activeChatSessionId); +} +export async function __testOnly_refreshConversationSituation(sessionId = null) { + return refreshConversationSituationForSession(sessionId || activeChatSessionId); +} +export function __testOnly_resetConversationSituationState() { + _clearConversationSituationState({ closePanel: true }); +} export async function __testOnly_refreshChatSessionTabs() { return refreshChatSessionTabs(); } @@ -34351,6 +35097,7 @@ export function __testOnly_resetChatRequestState() { activeChatSessionId = null; activeChatSessionName = null; activeChatSessionOwnerId = null; + _clearConversationSituationState({ closePanel: true }); renderChatTaskQueuePanel(); updateSendButtonForCurrentChatState(); } 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 3200bc8f2..b531691de 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 @@ -6,6 +6,11 @@ import { __testOnly_buildWorkflowMonitorExportPayload, __testOnly_buildWorkflowMonitorLocatorPayload, __testOnly_loadChatHistory, + __testOnly_initializeConversationSituationPanel, + __testOnly_applyConversationSituationPayload, + __testOnly_buildConversationSituationExportPayload, + __testOnly_refreshConversationSituation, + __testOnly_resetConversationSituationState, __testOnly_refreshChatSessionTabs, __testOnly_refreshAvailableWorkflowDefinitions, __testOnly_refreshWorkflowCapabilityIndexStatus, @@ -1715,6 +1720,426 @@ describe('workflow monitor active snapshot degradation handling', () => { }); }); +describe('conversation situation affordance', () => { + const situationPayload = (overrides = {}) => ({ + history: [], + segments_returned: 0, + total_segments: 0, + has_more_history: false, + conversation_situation: { + text: 'The user and Von are planning together.', + revision: 3, + source: 'adaptive_turn', + updated_by: '#V#user', + updated_at: '2026-07-29T09:00:00+00:00', + source_request_id: 'request-situation-3' + }, + conversation_observations: [ + { + schema_version: 'conversation_observation.v1', + observation_id: 'effect:paper-representation', + kind: 'durable_effect_terminal', + observed_at_utc: '2026-07-29T09:01:00Z', + capability_name: 'represent_paper', + terminal_status: 'completed', + changed: true + } + ], + conversation_observation_state: { + schema_version: 'conversation_observation_state.v1', + retained_count: 1, + total_count: 4, + omitted_count: 3, + retention_limit: 12 + }, + ...overrides + }); + + beforeEach(() => { + const { getUserContext, postJson } = require('../apiService.js'); + getUserContext.mockReturnValue({ + user_id: '#V#user', + org_id: '#V#org', + language: 'en-NZ', + gmail_profile: null + }); + postJson.mockResolvedValue({ status: 'updated' }); + document.body.innerHTML = ` + + +
Old transcript
+ +
+ `; + __testOnly_resetHistoryUiState(); + __testOnly_resetConversationSituationState(); + __testOnly_setActiveChatSession('session-1', 'Session 1'); + __testOnly_initializeConversationSituationPanel(); + }); + + afterEach(() => { + delete global.fetch; + __testOnly_resetHistoryUiState(); + __testOnly_resetConversationSituationState(); + __testOnly_setActiveChatSession(null, null); + }); + + test('shows inspectable situation, provenance, retention, and safe exact observations even with an empty transcript', async () => { + let historyFetchCount = 0; + let resolvePanelRefresh; + global.fetch = jest.fn((url) => { + if (typeof url === 'string' && url.startsWith('/von/api/session/context')) { + return Promise.resolve({ + ok: true, + json: async () => ({ + user_id: '#V#user', + organisation_id: '#V#org', + namespace: '#V#user@org' + }) + }); + } + if (typeof url === 'string' && url.startsWith('/von/history?')) { + historyFetchCount += 1; + const historyResponse = { + ok: true, + status: 200, + json: async () => situationPayload() + }; + if (historyFetchCount > 1) { + return new Promise((resolve) => { + resolvePanelRefresh = () => resolve(historyResponse); + }); + } + return Promise.resolve(historyResponse); + } + return Promise.resolve({ ok: true, json: async () => ({}) }); + }); + + const loaded = await __testOnly_loadChatHistory({ segments: 1 }); + document.getElementById('conversationSituationToggleBtn').click(); + await Promise.resolve(); + await Promise.resolve(); + + const panel = document.getElementById('conversationSituationPanel'); + const body = document.getElementById('conversationSituationBody'); + const panelText = panel.textContent; + expect(loaded).toBe(true); + expect(panel.classList.contains('hidden')).toBe(false); + expect(document.getElementById('conversationSituationToggleBtn').getAttribute('aria-expanded')).toBe('true'); + expect(document.getElementById('conversationSituationBadge').textContent).toBe('r3'); + expect(panelText).toContain('The user and Von are planning together.'); + expect(panel.querySelector('img')).toBeNull(); + expect(panel.hasAttribute('aria-live')).toBe(false); + expect(document.getElementById('conversationSituationStatus').textContent).toBe( + 'Refreshing the shared situation…' + ); + expect(document.getElementById('conversationSituationStatus').textContent).not.toContain( + 'planning setTimeout(resolve, 0)); + expect(typeof resolvePanelRefresh).toBe('function'); + resolvePanelRefresh(); + await new Promise(resolve => setTimeout(resolve, 0)); + expect(document.getElementById('conversationSituationStatus').textContent).toBe( + 'Conversation situation refreshed.' + ); + expect(body.getAttribute('aria-busy')).toBe('false'); + }); + + test('replaces the visible carrier immediately when the active session changes', () => { + __testOnly_applyConversationSituationPayload(situationPayload(), 'session-1'); + expect(document.getElementById('conversationSituationBody').textContent).toContain( + 'planning { + __testOnly_applyConversationSituationPayload(situationPayload(), 'session-1'); + __testOnly_applyConversationSituationPayload(situationPayload({ + conversation_situation: { + text: 'A newer situation with an older observation snapshot.', + revision: 4, + source: 'adaptive_turn', + updated_by: '#V#user', + updated_at: '2026-07-29T09:05:00+00:00' + }, + conversation_observations: [], + conversation_observation_state: { + retained_count: 0, + total_count: 2, + omitted_count: 2, + retention_limit: 12 + } + }), 'session-1'); + __testOnly_applyConversationSituationPayload(situationPayload({ + conversation_situation: { + text: 'A stale situation attached to newer observations.', + revision: 2, + source: 'adaptive_turn', + updated_by: '#V#user', + updated_at: '2026-07-29T09:06:00+00:00' + }, + conversation_observation_state: { + retained_count: 1, + total_count: 5, + omitted_count: 4, + retention_limit: 12 + } + }), 'session-1'); + + const exported = __testOnly_buildConversationSituationExportPayload('session-1'); + expect(exported.conversation_situation.text).toBe( + 'A newer situation with an older observation snapshot.' + ); + expect(exported.conversation_situation.revision).toBe(4); + expect(exported.conversation_observation_state.total_count).toBe(5); + expect(exported.conversation_observations).toHaveLength(1); + }); + + test('a canonical history read replaces the carrier after a reset lowers its revision and total', async () => { + __testOnly_applyConversationSituationPayload(situationPayload(), 'session-1'); + const resetPayload = situationPayload({ + conversation_situation: { + text: null, + revision: 1, + source: 'conversation_reset', + updated_by: '#V#user', + updated_at: '2026-07-29T09:10:00+00:00', + source_request_id: 'request-reset-1' + }, + conversation_observations: [], + conversation_observation_state: { + retained_count: 0, + total_count: 0, + omitted_count: 0, + retention_limit: 12 + } + }); + global.fetch = jest.fn((url) => { + if (typeof url === 'string' && url.startsWith('/von/api/session/context')) { + return Promise.resolve({ + ok: true, + json: async () => ({ + user_id: '#V#user', + organisation_id: '#V#org', + namespace: '#V#user@org' + }) + }); + } + if (typeof url === 'string' && url.startsWith('/von/history?')) { + return Promise.resolve({ + ok: true, + status: 200, + json: async () => resetPayload + }); + } + return Promise.resolve({ ok: true, json: async () => ({}) }); + }); + + const loaded = await __testOnly_loadChatHistory({ segments: 1 }); + const exported = __testOnly_buildConversationSituationExportPayload('session-1'); + const body = document.getElementById('conversationSituationBody'); + + expect(loaded).toBe(true); + expect(exported.availability).toBe('ready'); + expect(exported.conversation_situation.text).toBeNull(); + expect(exported.conversation_situation.revision).toBe(1); + expect(exported.conversation_situation.source).toBe('conversation_reset'); + expect(exported.conversation_observation_state.total_count).toBe(0); + expect(exported.conversation_observations).toEqual([]); + expect(body.textContent).toContain('the prior description was cleared'); + expect(body.textContent).toContain('No exact observations are currently retained.'); + expect(body.textContent).not.toContain('represent_paper'); + expect(body.getAttribute('aria-busy')).toBe('false'); + }); + + test('a late pre-reset generate payload cannot repopulate a canonical reset', () => { + __testOnly_applyConversationSituationPayload(situationPayload({ + conversation_situation: { + text: null, + revision: 4, + source: 'conversation_reset', + updated_by: '#V#user', + updated_at: '2026-07-29T09:10:00+00:00', + source_request_id: 'request-reset-4' + }, + conversation_observations: [], + conversation_observation_state: { + retained_count: 0, + total_count: 0, + omitted_count: 0, + retention_limit: 12 + } + }), 'session-1', { authoritativeSnapshot: true }); + + __testOnly_applyConversationSituationPayload(situationPayload({ + conversation_situation: { + text: 'The pre-reset situation.', + revision: 3, + source: 'adaptive_turn', + updated_by: '#V#user', + updated_at: '2026-07-29T09:09:00+00:00', + source_request_id: 'request-situation-3' + }, + conversation_observation_state: { + retained_count: 1, + total_count: 5, + omitted_count: 4, + retention_limit: 12 + } + }), 'session-1'); + + const exported = __testOnly_buildConversationSituationExportPayload('session-1'); + expect(exported.conversation_situation.text).toBeNull(); + expect(exported.conversation_situation.revision).toBe(4); + expect(exported.conversation_situation.source).toBe('conversation_reset'); + expect(exported.conversation_observation_state.total_count).toBe(0); + expect(exported.conversation_observations).toEqual([]); + }); + + test('a successful refresh without carrier fields exits refreshing as typed unavailable state', async () => { + global.fetch = jest.fn((url) => { + if (typeof url === 'string' && url.startsWith('/von/api/session/context')) { + return Promise.resolve({ + ok: true, + json: async () => ({ + user_id: '#V#user', + organisation_id: '#V#org', + namespace: '#V#user@org' + }) + }); + } + if (typeof url === 'string' && url.startsWith('/von/history?')) { + return Promise.resolve({ + ok: true, + status: 200, + json: async () => ({ + history: [], + segments_returned: 0, + total_segments: 0, + has_more_history: false, + conversation_situation: null + }) + }); + } + return Promise.resolve({ ok: true, json: async () => ({}) }); + }); + + const refreshed = await __testOnly_refreshConversationSituation('session-1'); + const exported = __testOnly_buildConversationSituationExportPayload('session-1'); + const refreshButton = document.getElementById('conversationSituationRefreshBtn'); + const body = document.getElementById('conversationSituationBody'); + + expect(refreshed).toBe(false); + expect(exported.availability).toBe('unavailable'); + expect(refreshButton.disabled).toBe(false); + expect(body.getAttribute('aria-busy')).toBe('false'); + expect(body.textContent).toContain('did not return conversation situation state'); + expect(document.getElementById('conversationSituationStatus').textContent).toBe( + 'The server did not return conversation situation state; try refreshing again.' + ); + }); + + test('refreshes only the bounded carrier projection and copies the exact cached state', async () => { + __testOnly_applyConversationSituationPayload(situationPayload({ + conversation_situation: { + text: 'A superseded local snapshot.', + revision: 8, + source: 'adaptive_turn', + updated_by: '#V#user', + updated_at: '2026-07-29T09:04:00+00:00' + }, + conversation_observation_state: { + retained_count: 1, + total_count: 8, + omitted_count: 7, + retention_limit: 12 + } + }), 'session-1'); + const updatedPayload = situationPayload({ + conversation_situation: { + text: 'The refreshed shared situation.', + revision: 4, + source: 'adaptive_turn', + updated_by: '#V#user', + updated_at: '2026-07-29T09:05:00+00:00', + source_request_id: 'request-situation-4' + } + }); + global.fetch = jest.fn((url) => { + if (typeof url === 'string' && url.startsWith('/von/api/session/context')) { + return Promise.resolve({ + ok: true, + json: async () => ({ + user_id: '#V#user', + organisation_id: '#V#org', + namespace: '#V#user@org' + }) + }); + } + if (typeof url === 'string' && url.startsWith('/von/history?')) { + return Promise.resolve({ + ok: true, + status: 200, + json: async () => updatedPayload + }); + } + return Promise.resolve({ ok: true, json: async () => ({}) }); + }); + + const refreshed = await __testOnly_refreshConversationSituation('session-1'); + const historyUrl = global.fetch.mock.calls + .map(([url]) => url) + .find(url => typeof url === 'string' && url.startsWith('/von/history?')); + const exported = __testOnly_buildConversationSituationExportPayload('session-1'); + + expect(refreshed).toBe(true); + expect(historyUrl).toContain('segment_size=1'); + expect(historyUrl).toContain('tail_limit=1'); + expect(historyUrl).toContain('include_debug=0'); + expect(exported.schema_version).toBe('conversation_situation_export.v1'); + expect(exported.conversation_situation.text).toBe('The refreshed shared situation.'); + expect(exported.conversation_situation.revision).toBe(4); + expect(exported.conversation_observation_state.total_count).toBe(4); + expect(exported.conversation_observation_state.omitted_count).toBe(3); + expect(document.getElementById('conversationSituationStatus').textContent).toBe( + 'Conversation situation refreshed.' + ); + expect(document.getElementById('conversationSituationStatus').textContent).not.toContain( + 'The refreshed shared situation.' + ); + }); +}); + describe('loadChatHistory degraded handling', () => { beforeEach(() => { const { getUserContext, postJson } = require('../apiService.js'); diff --git a/src/frontend/web/von_interface/static/styles.css b/src/frontend/web/von_interface/static/styles.css index 41746abdf..ecd9173ab 100644 --- a/src/frontend/web/von_interface/static/styles.css +++ b/src/frontend/web/von_interface/static/styles.css @@ -7347,6 +7347,259 @@ body[data-active-tab="settingsTab"] .tab-content-area { display: none; } +.chat-situation-toggle { + min-height: 32px; +} + +.chat-situation-toggle[aria-expanded="true"] { + background-color: #007bff; + color: #fff; +} + +.chat-situation-toggle-label { + font-size: 13px; + line-height: 1; +} + +.chat-situation-badge { + min-width: 20px; + padding: 2px 6px; + border-radius: 999px; + background: #dbeafe; + color: #1e40af; + font-size: 10px; + font-weight: 700; + line-height: 1.2; + font-variant-numeric: tabular-nums; +} + +.chat-situation-toggle[aria-expanded="true"] .chat-situation-badge { + background: #fff; + color: #1d4ed8; +} + +.chat-situation-panel { + position: fixed; + top: clamp(72px, 10vh, 110px); + right: 18px; + bottom: 64px; + z-index: 1300; + width: min(720px, calc(100vw - 36px)); + margin: 0; + padding: 16px 18px; + box-sizing: border-box; + border: 1px solid #bfdbfe; + border-radius: 14px; + background: linear-gradient(145deg, #f8fbff 0%, #eff6ff 100%); + box-shadow: 0 18px 52px rgba(15, 23, 42, 0.24); + color: #0f172a; + overflow-y: auto; + overscroll-behavior: contain; +} + +.chat-situation-header { + display: flex; + align-items: flex-start; + justify-content: space-between; + gap: 16px; +} + +.chat-situation-eyebrow { + margin-bottom: 3px; + color: #1d4ed8; + font-size: 11px; + font-weight: 700; + letter-spacing: 0.06em; + text-transform: uppercase; +} + +.chat-situation-title { + margin: 0; + font-size: 20px; + line-height: 1.25; +} + +.chat-situation-actions { + display: flex; + align-items: center; + justify-content: flex-end; + gap: 7px; + flex-wrap: wrap; +} + +.chat-situation-explanation { + max-width: 88ch; + margin: 10px 0 14px; + color: #475569; + font-size: 12px; + line-height: 1.45; +} + +.chat-situation-body { + display: flex; + flex-direction: column; + gap: 12px; +} + +.chat-situation-notice, +.chat-situation-empty { + margin: 0; + padding: 9px 11px; + border-radius: 8px; + background: rgba(255, 255, 255, 0.78); + color: #475569; + font-size: 13px; + line-height: 1.45; +} + +.chat-situation-notice.is-warning { + border: 1px solid #fbbf24; + background: #fffbeb; + color: #92400e; +} + +.chat-situation-text { + margin: 0; + padding: 13px 14px; + border: 1px solid #dbeafe; + border-radius: 10px; + background: #fff; + color: #172554; + font-family: inherit; + font-size: 14px; + line-height: 1.55; + overflow-wrap: anywhere; + white-space: pre-wrap; +} + +.chat-situation-section-label { + margin: 0 0 6px; + color: #334155; + font-size: 12px; + font-weight: 700; +} + +.chat-situation-metadata, +.chat-situation-observation-fields { + display: grid; + grid-template-columns: minmax(110px, max-content) minmax(0, 1fr); + gap: 5px 12px; + margin: 0; + padding: 10px 12px; + border: 1px solid #dbeafe; + border-radius: 9px; + background: rgba(255, 255, 255, 0.72); + font-size: 12px; +} + +.chat-situation-metadata dt, +.chat-situation-observation-fields dt { + color: #64748b; + font-weight: 600; +} + +.chat-situation-metadata dd, +.chat-situation-observation-fields dd { + min-width: 0; + margin: 0; + color: #1e293b; + overflow-wrap: anywhere; +} + +.chat-situation-observations { + border: 1px solid #bfdbfe; + border-radius: 10px; + background: rgba(255, 255, 255, 0.62); +} + +.chat-situation-observations > summary { + padding: 10px 12px; + cursor: pointer; + color: #1e3a8a; + font-size: 13px; + font-weight: 700; +} + +.chat-situation-observation-summary-meta { + margin-left: 5px; + color: #64748b; + font-size: 11px; + font-weight: 500; +} + +.chat-situation-observation-list { + display: flex; + flex-direction: column; + gap: 9px; + padding: 0 10px 10px; +} + +.chat-situation-observation { + padding: 10px 11px; + border: 1px solid #e2e8f0; + border-radius: 9px; + background: #fff; +} + +.chat-situation-observation-title { + margin: 0 0 7px; + color: #1e293b; + font-size: 13px; + font-weight: 700; +} + +.chat-situation-observation-omission { + margin: 0; + color: #92400e; + font-size: 12px; +} + +@media (max-width: 640px) { + .chat-header { + align-items: flex-start; + flex-direction: column; + } + + .chat-header .main-title { + width: 100%; + text-align: left; + } + + .chat-header-controls { + width: 100%; + justify-content: flex-start; + } + + .chat-situation-panel { + top: 56px; + right: 10px; + bottom: 52px; + left: 10px; + width: auto; + padding: 13px; + } + + .chat-situation-header { + flex-direction: column; + } + + .chat-situation-actions { + width: 100%; + justify-content: flex-start; + } + + .chat-situation-metadata, + .chat-situation-observation-fields { + grid-template-columns: 1fr; + gap: 2px; + } + + .chat-situation-metadata dd, + .chat-situation-observation-fields dd { + margin-bottom: 5px; + } +} + .workflow-status-panel { margin: 12px 0 18px; padding: 12px 14px; diff --git a/src/frontend/web/von_interface/templates/chat_tab.html b/src/frontend/web/von_interface/templates/chat_tab.html index 64d19cffb..295e28981 100644 --- a/src/frontend/web/von_interface/templates/chat_tab.html +++ b/src/frontend/web/von_interface/templates/chat_tab.html @@ -5,6 +5,12 @@

Talk to Von Neumarkt

+ + + +
+
+

+ This is the conversation's provisional, revisable account of the shared situation. It is not canonical + domain knowledge or an authority grant. +

+ +
+ +
diff --git a/tests/backend/test_browser_test_auth_service.py b/tests/backend/test_browser_test_auth_service.py index cd5f2428e..528bc66b4 100644 --- a/tests/backend/test_browser_test_auth_service.py +++ b/tests/backend/test_browser_test_auth_service.py @@ -234,6 +234,8 @@ def test_ensure_fixture_chat_session_skips_duplicate_history_entries(monkeypatch fake_coll = _FakeChatHistoryCollection() added_entries: list[str] = [] skip_rag_indexing_flags: list[bool | None] = [] + situation_updates: list[dict] = [] + observation_updates: list[str] = [] def _fake_create_chat_session( *, @@ -282,6 +284,29 @@ def _fake_add_message_to_history( added_entries.append(message["fixture_entry_id"]) skip_rag_indexing_flags.append(skip_rag_indexing) + def _fake_set_conversation_situation(**kwargs): + doc = fake_coll.docs[(kwargs["user_id"], kwargs["session_id"])] + revision = kwargs["expected_revision"] + 1 + doc["conversation_situation"] = { + "text": kwargs["text"], + "revision": revision, + "source": kwargs["source"], + "updated_by": kwargs["updated_by"], + "source_request_id": kwargs["source_request_id"], + } + situation_updates.append(dict(kwargs)) + return {"updated": True, "current_revision": revision} + + def _fake_append_conversation_observation(**kwargs): + doc = fake_coll.docs[(kwargs["user_id"], kwargs["session_id"])] + observations = doc.setdefault("conversation_observations", []) + observation_id = kwargs["observation"]["observation_id"] + if any(item.get("observation_id") == observation_id for item in observations): + return {"updated": False, "duplicate": True} + observations.append(dict(kwargs["observation"])) + observation_updates.append(observation_id) + return {"updated": True, "duplicate": False} + monkeypatch.setattr( service.chat_history_service, "create_chat_session", @@ -297,10 +322,31 @@ def _fake_add_message_to_history( "add_message_to_history", _fake_add_message_to_history, ) + monkeypatch.setattr( + service.chat_history_service, + "set_chat_history_conversation_situation", + _fake_set_conversation_situation, + ) + monkeypatch.setattr( + service.chat_history_service, + "append_chat_history_conversation_observation", + _fake_append_conversation_observation, + ) spec = { "session_id": "browser-fixture-user-view-state", "session_name": "Authenticated user-view fixture sanity check", + "conversation_situation": { + "text": "The fixture situation is inspectable.", + "source_request_id": "browser-fixture-request", + }, + "conversation_observations": ( + { + "observation_id": "browser-fixture-ready", + "kind": "fixture_state", + "terminal_status": "ready", + }, + ), "history": ( { "fixture_entry_id": "entry-1", @@ -333,15 +379,27 @@ def _fake_add_message_to_history( ) assert first["created_entries"] == 2 + assert first["situation_updated"] is True + assert first["observations_added"] == 1 assert second["created_entries"] == 0 + assert second["situation_updated"] is False + assert second["observations_added"] == 0 assert added_entries == ["entry-1", "entry-2"] assert skip_rag_indexing_flags == [True, True] + assert len(situation_updates) == 1 + assert situation_updates[0]["expected_revision"] == 0 + assert situation_updates[0]["include_legacy"] is False + assert observation_updates == ["browser-fixture-ready"] stored_doc = fake_coll.docs[ ("#V#zhan_von_witbrock", "browser-fixture-user-view-state") ] assert stored_doc["origin_kind"] == "browser_test_fixture" assert stored_doc["is_agent_created"] is True assert stored_doc["test_artifact_kind"] == "browser_test_fixture_chat_session" + assert stored_doc["conversation_situation"]["text"] == ( + "The fixture situation is inspectable." + ) + assert stored_doc["conversation_observations"][0]["terminal_status"] == "ready" def test_login_browser_test_user_handles_existing_counterpart_name_variants( diff --git a/tests/backend/test_chat_history_service_segments.py b/tests/backend/test_chat_history_service_segments.py index 6d10135e7..5d39f7680 100644 --- a/tests/backend/test_chat_history_service_segments.py +++ b/tests/backend/test_chat_history_service_segments.py @@ -409,9 +409,11 @@ def test_get_chat_history_telemetry_locator_projection_keeps_debug_payloads_out( class _CompactProjectionCollection: def __init__(self): self.pipeline = None + self.aggregate_calls = 0 def aggregate(self, pipeline): self.pipeline = pipeline + self.aggregate_calls += 1 return iter( [ { @@ -419,6 +421,17 @@ def aggregate(self, pipeline): "session_id": "s1", "session_name": "Compact locator", "namespace": "#V#u@org", + "conversation_situation_state": { + "available": True, + "revision": 4, + "updated_at": datetime( + 2026, 7, 29, 9, 0, tzinfo=timezone.utc + ), + }, + "conversation_observation_state": { + "retained_count": 2, + "total_count": 5, + }, "history": [ { "role": "assistant", @@ -455,10 +468,107 @@ def aggregate(self, pipeline): } assert collection.pipeline is not None history_projection = collection.pipeline[-1]["$project"]["history"] - projected_debug = history_projection["$map"]["in"]["llm_debug_data"][ - "$cond" - ][1] + projected_debug = history_projection["$map"]["in"]["llm_debug_data"]["$cond"][1] assert set(projected_debug) == {"request_id", "turn_id", "timestamp_utc"} + assert result["conversation_situation_state"] == { + "available": True, + "revision": 4, + "updated_at": "2026-07-29T09:00:00+00:00", + } + assert result["conversation_observation_state"] == { + "schema_version": "conversation_observation_state.v1", + "retained_count": 2, + "total_count": 5, + "omitted_count": 3, + "retention_limit": 12, + } + assert "conversation_situation" not in result + assert "conversation_observations" not in result + compact_projection = collection.pipeline[-1]["$project"] + assert "conversation_situation_state" in compact_projection + assert "conversation_observation_state" in compact_projection + assert "conversation_situation" not in compact_projection + assert "conversation_observations" not in compact_projection + assert collection.aggregate_calls == 1 + + +def test_get_chat_history_segments_can_return_carried_state_with_empty_history_in_one_read( + monkeypatch, +): + from src.backend.services import chat_history_service + + situation = { + "text": "Waiting for a durable workflow outcome.", + "revision": 2, + "source": "adaptive_turn", + "updated_by": "#V#u", + "updated_at": datetime(2026, 7, 29, 9, 0, tzinfo=timezone.utc), + } + observations = [ + { + "schema_version": "conversation_observation.v1", + "observation_id": "late-effect-1", + "kind": "late_terminal_effect", + "terminal_status": "completed", + } + ] + + class _EmptyHistoryStateCollection: + def __init__(self): + self.aggregate_calls = 0 + self.pipeline = None + + def aggregate(self, pipeline, **_kwargs): + self.aggregate_calls += 1 + self.pipeline = pipeline + return iter( + [ + { + "history": [], + "history_length": 0, + "conversation_situation": situation, + "conversation_observations": observations, + "conversation_observation_total": 3, + } + ] + ) + + def find_one(self, *_args, **_kwargs): + raise AssertionError("state must not require a second DB read") + + collection = _EmptyHistoryStateCollection() + monkeypatch.setattr( + chat_history_service, + "get_chat_history_collection_service", + lambda **_kwargs: collection, + ) + + result = chat_history_service.get_chat_history_segments( + "#V#u", + "s1", + history_tail_limit=100, + return_meta=True, + include_conversation_state=True, + ) + + assert isinstance(result, tuple) + segments, meta = result + assert segments == [] + assert meta["conversation_situation"]["text"] == situation["text"] + assert meta["conversation_situation"]["revision"] == 2 + assert meta["conversation_observations"] == observations + assert meta["conversation_observation_state"] == { + "schema_version": "conversation_observation_state.v1", + "retained_count": 1, + "total_count": 3, + "omitted_count": 2, + "retention_limit": 12, + } + projected = collection.pipeline[-1]["$project"] + assert projected["conversation_situation"] == 1 + assert projected["conversation_observations"] == 1 + assert projected["conversation_observation_total"] == 1 + assert collection.aggregate_calls == 1 def test_get_chat_history_segments_keeps_debug_blob_refs_compact_by_default( diff --git a/tests/backend/test_conversation_telemetry_locator_service.py b/tests/backend/test_conversation_telemetry_locator_service.py index ad178b6d8..5a272419a 100644 --- a/tests/backend/test_conversation_telemetry_locator_service.py +++ b/tests/backend/test_conversation_telemetry_locator_service.py @@ -71,6 +71,20 @@ def test_build_conversation_locator_prefers_exact_namespace_and_preserves_org_sc "namespace": "#V#u@org", "organisation_concept_id": "#V#org", "session_name": "Org Session", + "conversation_situation": { + "text": "Inspectable carrier text must not enter the locator.", + "revision": 7, + "source": "adaptive_turn", + "updated_by": "#V#u", + "updated_at": "2026-07-29T10:00:00Z", + }, + "conversation_observations": [ + { + "observation_id": "effect-1", + "kind": "late_terminal_effect", + } + ], + "conversation_observation_total": 4, "history": [ { "role": "assistant", @@ -108,6 +122,20 @@ def test_build_conversation_locator_prefers_exact_namespace_and_preserves_org_sc assert payload["metadata"]["total_turns"] == 1 assert payload["metadata"]["transcript_turn_count"] == 1 assert payload["turns"][0]["request_id"] == "req-123" + assert payload["conversation_situation_state"] == { + "available": True, + "revision": 7, + "updated_at": "2026-07-29T10:00:00+00:00", + } + assert payload["conversation_observation_state"] == { + "schema_version": "conversation_observation_state.v1", + "retained_count": 1, + "total_count": 4, + "omitted_count": 3, + "retention_limit": 12, + } + assert "conversation_situation" not in payload + assert "conversation_observations" not in payload debug_args = payload["turns"][0]["mcp_access"]["chat_history_get_debug_entry"][ "arguments" ] @@ -132,3 +160,54 @@ def test_build_conversation_locator_prefers_exact_namespace_and_preserves_org_sc ) assert verified_conversation_ref["success"] is True assert verified_conversation_ref["chat_session_id"] == "s1" + segments_descriptor = payload["mcp_access"]["chat_history_get_segments"] + assert "conversation carrier" in segments_descriptor["purpose"] + + +def test_build_conversation_locator_keeps_carrier_freshness_when_history_is_empty( + monkeypatch, +): + from src.backend.services import chat_history_service + from src.backend.services.conversation_telemetry_locator_service import ( + build_conversation_llm_telemetry_locator, + ) + + calls = [] + monkeypatch.setattr( + chat_history_service, + "get_chat_history_telemetry_locator_projection", + lambda *args, **kwargs: ( + calls.append((args, kwargs)) + or { + "session_id": "empty-session", + "namespace": "#V#u@org", + "organisation_concept_id": "#V#org", + "history": [], + "conversation_situation_state": { + "available": True, + "revision": 2, + "updated_at": "2026-07-29T11:00:00+00:00", + }, + "conversation_observation_state": { + "schema_version": "conversation_observation_state.v1", + "retained_count": 0, + "total_count": 1, + "omitted_count": 1, + "retention_limit": 12, + }, + } + ), + ) + + payload = build_conversation_llm_telemetry_locator( + user_id="#V#u", + session_id="empty-session", + namespace="#V#u@org", + organisation_concept_id="#V#org", + ) + + assert len(calls) == 1 + assert payload["turns"] == [] + assert payload["metadata"]["transcript_turn_count"] == 0 + assert payload["conversation_situation_state"]["revision"] == 2 + assert payload["conversation_observation_state"]["total_count"] == 1 diff --git a/tests/backend/test_internal_mcp_locator_tools.py b/tests/backend/test_internal_mcp_locator_tools.py index d1c15aca4..7bc4cabea 100644 --- a/tests/backend/test_internal_mcp_locator_tools.py +++ b/tests/backend/test_internal_mcp_locator_tools.py @@ -50,21 +50,62 @@ def test_chat_history_get_segments_returns_provenanced_payload(monkeypatch): }, ) + situation = { + "text": "The shared situation is inspectable.", + "revision": 3, + "source": "adaptive_turn", + "updated_by": "#V#user", + "updated_at": "2026-07-29T10:00:00+00:00", + } + observations = [ + { + "schema_version": "conversation_observation.v1", + "observation_id": "effect-1", + "kind": "late_terminal_effect", + } + ] + observation_state = { + "schema_version": "conversation_observation_state.v1", + "retained_count": 1, + "total_count": 2, + "omitted_count": 1, + "retention_limit": 12, + } + + def get_segments(*_args, **kwargs): + assert kwargs["include_conversation_state"] is True + return ( + [ + { + "segment_index": 0, + "history": [{"role": "assistant", "content": "Done"}], + } + ], + { + "history_truncated": False, + "conversation_situation": situation, + "conversation_observations": observations, + "conversation_observation_state": observation_state, + }, + ) + monkeypatch.setattr( "src.backend.services.chat_history_service.get_chat_history_segments", - lambda *args, **kwargs: ( - [{"segment_index": 0, "history": [{"role": "assistant", "content": "Done"}]}], - {"history_truncated": False}, - ), + get_segments, ) - result = cat._chat_history_get_segments(session_id="chat-1", namespace="#V#user@org") + result = cat._chat_history_get_segments( + session_id="chat-1", namespace="#V#user@org" + ) assert result["success"] is True assert result["session_id"] == "chat-1" assert result["chat_session_id"] == "chat-1" assert result["segment_count"] == 1 assert result["history_truncated"] is False + assert result["conversation_situation"] == situation + assert result["conversation_observations"] == observations + assert result["conversation_observation_state"] == observation_state assert result["identifier_binding"]["mode"] == "raw_parameters" assert result["provenance"]["item_kind"] == "chat_history_segments" diff --git a/tests/backend/test_telemetry_read_delegation_stdio.py b/tests/backend/test_telemetry_read_delegation_stdio.py index a4691d1c0..127427038 100644 --- a/tests/backend/test_telemetry_read_delegation_stdio.py +++ b/tests/backend/test_telemetry_read_delegation_stdio.py @@ -1,28 +1,35 @@ from __future__ import annotations import asyncio +import hashlib import json from datetime import datetime, timedelta, timezone from typing import Any, cast from src.backend.services.conversation_scope_binding_service import ( + build_conversation_scope_binding, build_history_location_binding, build_turn_telemetry_binding, ) -def _stdio_payload(tool_name: str, arguments: dict[str, Any]) -> dict[str, Any]: +def _stdio_text_and_payload( + tool_name: str, + arguments: dict[str, Any], +) -> tuple[str, dict[str, Any]]: from src.backend.mcp_server import mcp_stdio_server - async def run() -> dict[str, Any]: + async def run() -> str: 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 cast(str, getattr(first, "text")) + + text = asyncio.run(run()) + return text, cast(dict[str, Any], json.loads(text)) - return asyncio.run(run()) + +def _stdio_payload(tool_name: str, arguments: dict[str, Any]) -> dict[str, Any]: + return _stdio_text_and_payload(tool_name, arguments)[1] def test_real_stdio_handler_path_accepts_exact_actor_bound_reads(monkeypatch) -> None: @@ -121,6 +128,316 @@ def test_real_stdio_handler_path_accepts_exact_actor_bound_reads(monkeypatch) -> assert diagnostics_result["source_system"] == "mongo.chat_history" +def test_actor_bound_conversation_carrier_is_paged_and_locator_stays_compact( + monkeypatch, +) -> None: + conversation_ref = build_conversation_scope_binding( + chat_session_id="chat-shared-carrier", + history_owner_user_id="#V#owner", + read_namespace="#V#owner@org", + organisation_concept_id="#V#org", + delegated_actor_user_id="#V#invitee", + delegated_actor_namespace="#V#invitee@org", + ) + situation = { + "text": "Inspectable shared situation", + "revision": 5, + "source": "adaptive_turn", + "updated_by": "#V#invitee", + "updated_at": "2026-07-29T10:00:00+00:00", + } + observation_state = { + "schema_version": "conversation_observation_state.v1", + "retained_count": 1, + "total_count": 3, + "omitted_count": 2, + "retention_limit": 12, + } + observations = [ + { + "schema_version": "conversation_observation.v1", + "observation_id": "late-1", + "kind": "late_terminal_effect", + } + ] + message_timestamp = datetime(2026, 7, 29, 9, 59, tzinfo=timezone.utc) + captured_segments: dict[str, Any] = {} + captured_locator: dict[str, Any] = {} + + 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#owner", + ) + monkeypatch.setattr( + "src.backend.services.shared_conversation_service.get_accepted_invite_for_user_session", + lambda **_kwargs: {"organisation_concept_id": "#V#org"}, + ) + monkeypatch.setattr( + "src.backend.integrations.internal_mcp.catalogue._is_user_member_of_organisation", + lambda **_kwargs: True, + ) + 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#owner" + and session_id == "chat-shared-carrier" + and namespace == "#V#owner@org" + ), + ) + + def get_segments(*_args, **kwargs): + captured_segments.update(kwargs) + return ( + [ + [ + { + "role": "assistant", + "content": "x" * 80_000, + "timestamp": message_timestamp, + } + ] + ], + { + "history_truncated": False, + "conversation_situation": situation, + "conversation_observations": observations, + "conversation_observation_state": observation_state, + }, + ) + + monkeypatch.setattr( + "src.backend.services.chat_history_service.get_chat_history_segments", + get_segments, + ) + + def build_locator(**kwargs): + captured_locator.update(kwargs) + return { + "schema_version": "conversation_llm_telemetry_locator.v1", + "generated_at_utc": "2026-07-29T10:01:00Z", + "session_id": "chat-shared-carrier", + "namespace_context": { + "namespace": "#V#owner@org", + "user_id": "#V#owner", + "org_id": "#V#org", + }, + "conversation_situation_state": { + "available": True, + "revision": 5, + "updated_at": "2026-07-29T10:00:00+00:00", + }, + "conversation_observation_state": observation_state, + "metadata": {"total_turns": 0}, + "mcp_access": {}, + "turns": [], + } + + monkeypatch.setattr( + "src.backend.services.conversation_telemetry_locator_service.build_conversation_llm_telemetry_locator", + build_locator, + ) + + authority_args = { + "conversation_ref": conversation_ref, + "namespace": "#V#invitee@org", + "user_concept_id": "#V#invitee", + "organisation_concept_id": "#V#org", + } + carrier = _stdio_payload( + "chat_history_get_segments", + {**authority_args, "limit": 20_000}, + ) + carrier_pages = [carrier] + while carrier_pages[-1]["bounded_read"]["has_more"]: + next_offset = carrier_pages[-1]["bounded_read"]["next_offset"] + assert isinstance(next_offset, int) + assert len(carrier_pages) < 20 + carrier_pages.append( + _stdio_payload( + "chat_history_get_segments", + { + **authority_args, + "offset": next_offset, + "limit": 20_000, + }, + ) + ) + locator = _stdio_payload( + "conversation_telemetry_get_locator", + authority_args, + ) + + assert carrier["success"] is True + assert "segments" not in carrier + assert carrier["bounded_read"]["returned_chars"] == 20_000 + assert carrier["bounded_read"]["has_more"] is True + assert carrier["history_owner_user_id"] == "#V#owner" + assert carrier["requested_user_id"] == "#V#invitee" + assert carrier["namespace"] == "#V#owner@org" + assert carrier["identifier_binding"]["validation_status"] == "verified" + assert carrier["read_delegation"]["delegated_actor_user_id"] == "#V#invitee" + assert captured_segments["namespace"] == "#V#owner@org" + assert captured_segments["include_conversation_state"] is True + + page_metadata = [page["bounded_read"] for page in carrier_pages] + assert len({page["sha256"] for page in page_metadata}) == 1 + assert len({page["total_chars"] for page in page_metadata}) == 1 + assert [page["offset"] for page in page_metadata] == [ + index * 20_000 for index in range(len(page_metadata)) + ] + canonical_json = "".join(page["json_chunk"] for page in page_metadata) + assert len(canonical_json) == page_metadata[0]["total_chars"] + assert ( + hashlib.sha256(canonical_json.encode("utf-8")).hexdigest() + == (page_metadata[0]["sha256"]) + ) + reconstructed_carrier = json.loads(canonical_json) + assert reconstructed_carrier["conversation_situation"] == situation + assert reconstructed_carrier["conversation_observations"] == observations + assert reconstructed_carrier["conversation_observation_state"] == ( + observation_state + ) + assert reconstructed_carrier["history_owner_user_id"] == "#V#owner" + assert reconstructed_carrier["requested_user_id"] == "#V#invitee" + assert reconstructed_carrier["segments"][0][0]["timestamp"] == ( + message_timestamp.isoformat() + ) + for page in carrier_pages: + assert page["history_owner_user_id"] == "#V#owner" + assert page["requested_user_id"] == "#V#invitee" + assert page["namespace"] == "#V#owner@org" + assert page["read_delegation"]["delegated_actor_user_id"] == "#V#invitee" + + assert locator["conversation_situation_state"]["revision"] == 5 + assert locator["conversation_observation_state"] == observation_state + assert "conversation_situation" not in locator + assert "conversation_observations" not in locator + assert locator["history_owner_user_id"] == "#V#owner" + assert locator["requested_user_id"] == "#V#invitee" + assert locator["read_delegation"]["delegated_actor_user_id"] == "#V#invitee" + assert captured_locator["user_id"] == "#V#owner" + assert captured_locator["requested_user_id"] == "#V#invitee" + + +def test_escape_heavy_carrier_pages_remain_under_real_stdio_guard( + monkeypatch, +) -> None: + monkeypatch.setenv("VON_MCP_STDIO_MAX_RESPONSE_CHARS", "10000") + conversation_ref = build_conversation_scope_binding( + chat_session_id="chat-escape-heavy-carrier", + history_owner_user_id="#V#actor", + read_namespace="#V#actor@org", + organisation_concept_id="#V#org", + delegated_actor_user_id="#V#actor", + delegated_actor_namespace="#V#actor@org", + ) + message_timestamp = datetime(2026, 7, 29, 12, 34, 56, tzinfo=timezone.utc) + escape_heavy_content = '\\"' * 6_000 + + 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", + ) + monkeypatch.setattr( + "src.backend.integrations.internal_mcp.catalogue._is_user_member_of_organisation", + lambda **_kwargs: True, + ) + 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" + and session_id == "chat-escape-heavy-carrier" + and namespace == "#V#actor@org" + ), + ) + monkeypatch.setattr( + "src.backend.services.chat_history_service.get_chat_history_segments", + lambda *_args, **_kwargs: ( + [ + [ + { + "role": "assistant", + "content": escape_heavy_content, + "timestamp": message_timestamp, + } + ] + ], + { + "history_truncated": False, + "conversation_situation": None, + "conversation_observations": [], + "conversation_observation_state": { + "schema_version": "conversation_observation_state.v1", + "retained_count": 0, + "total_count": 0, + "omitted_count": 0, + "retention_limit": 12, + }, + }, + ), + ) + + authority_args = { + "conversation_ref": conversation_ref, + "namespace": "#V#actor@org", + "user_concept_id": "#V#actor", + "organisation_concept_id": "#V#org", + } + response_text, first_page = _stdio_text_and_payload( + "chat_history_get_segments", + {**authority_args, "limit": 20_000}, + ) + response_texts = [response_text] + pages = [first_page] + while pages[-1]["bounded_read"]["has_more"]: + next_offset = pages[-1]["bounded_read"]["next_offset"] + assert isinstance(next_offset, int) + assert next_offset > pages[-1]["bounded_read"]["offset"] + assert len(pages) < 30 + response_text, page = _stdio_text_and_payload( + "chat_history_get_segments", + { + **authority_args, + "offset": next_offset, + "limit": 20_000, + }, + ) + response_texts.append(response_text) + pages.append(page) + + assert all(len(text) <= 10_000 for text in response_texts) + assert all(page.get("error_code") != "payload_too_large" for page in pages) + assert pages[0]["bounded_read"]["returned_chars"] < 20_000 + page_metadata = [page["bounded_read"] for page in pages] + assert len({page["sha256"] for page in page_metadata}) == 1 + assert len({page["total_chars"] for page in page_metadata}) == 1 + canonical_json = "".join(page["json_chunk"] for page in page_metadata) + assert len(canonical_json) == page_metadata[0]["total_chars"] + assert ( + hashlib.sha256(canonical_json.encode("utf-8")).hexdigest() + == (page_metadata[0]["sha256"]) + ) + reconstructed = json.loads(canonical_json) + assert reconstructed["segments"][0][0]["content"] == escape_heavy_content + assert reconstructed["segments"][0][0]["timestamp"] == ( + message_timestamp.isoformat() + ) + for page in pages: + assert page["history_owner_user_id"] == "#V#actor" + assert page["requested_user_id"] == "#V#actor" + assert page["namespace"] == "#V#actor@org" + assert page["identifier_binding"]["validation_status"] == "verified" + assert page["read_delegation"]["delegated_actor_user_id"] == "#V#actor" + + 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", diff --git a/tests/backend/test_von_generate_conversation_session_override.py b/tests/backend/test_von_generate_conversation_session_override.py index f59dfad0a..6ee8a7ba9 100644 --- a/tests/backend/test_von_generate_conversation_session_override.py +++ b/tests/backend/test_von_generate_conversation_session_override.py @@ -264,6 +264,7 @@ def test_generate_carries_and_updates_inspectable_conversation_situation( events: list[tuple[str, object]] = [] adaptive_calls: list[dict[str, Any]] = [] set_calls: list[dict[str, Any]] = [] + state_calls: list[dict[str, Any]] = [] existing_situation = { "text": "We are deciding how to represent the paper.", @@ -279,16 +280,60 @@ def test_generate_carries_and_updates_inspectable_conversation_situation( "summary": "The paper representation completed.", } ] + observation_state = { + "schema_version": "conversation_observation_state.v1", + "retained_count": 1, + "total_count": 1, + "omitted_count": 0, + "retention_limit": 24, + } + canonical_situation = { + "text": "We are representing the paper; its canonical identity is now known.", + "revision": 5, + "source": "adaptive_turn", + "updated_by": "#V#user", + "updated_at": "2026-07-29T09:00:00+00:00", + "source_request_id": "request-from-turn", + } + canonical_observations = [ + *observations, + { + "observation_id": "identity:paper", + "kind": "canonical_read_back", + "summary": "The paper identity is available.", + }, + ] + canonical_observation_state = { + "schema_version": "conversation_observation_state.v1", + "retained_count": 2, + "total_count": 2, + "omitted_count": 0, + "retention_limit": 24, + } + + def _session_state(**kwargs: Any) -> dict[str, Any]: + state_calls.append(dict(kwargs)) + if len(state_calls) == 1: + return { + "session_id": kwargs["session_id"], + "history": [{"role": "user", "content": "Earlier context"}], + "conversation_situation": existing_situation, + "conversation_observations": observations, + "conversation_observation_state": observation_state, + } + canonical_situation["source_request_id"] = adaptive_calls[0]["turn_id"] + return { + "session_id": kwargs["session_id"], + "history": [], + "conversation_situation": canonical_situation, + "conversation_observations": canonical_observations, + "conversation_observation_state": canonical_observation_state, + } monkeypatch.setattr( von_routes.chat_history_service, "get_chat_history_session_state", - lambda **kwargs: { - "session_id": kwargs["session_id"], - "history": [{"role": "user", "content": "Earlier context"}], - "conversation_situation": existing_situation, - "conversation_observations": observations, - }, + _session_state, ) def _add_message(**kwargs: Any) -> None: @@ -310,12 +355,13 @@ def _adaptive(**kwargs: Any) -> AdaptiveTurnResult: def _set_situation(**kwargs: Any) -> dict[str, Any]: set_calls.append(dict(kwargs)) events.append(("situation", kwargs["text"])) + next_revision = kwargs["expected_revision"] + 1 return { "updated": True, "matched": True, "conflict": False, "expected_revision": kwargs["expected_revision"], - "current_revision": kwargs["expected_revision"] + 1, + "current_revision": next_revision, "session_id": kwargs["session_id"], } @@ -336,6 +382,13 @@ def _set_situation(**kwargs: Any) -> dict[str, Any]: ) assert response.status_code == 200, response.get_json() + body = response.get_json() + assert body["conversation_situation"] == canonical_situation + assert body["conversation_observations"] == canonical_observations + assert ( + body["conversation_observation_state"] + == canonical_observation_state + ) assert adaptive_calls[0]["conversation_id"] == "session-situation" assert ( adaptive_calls[0]["conversation_situation"] @@ -360,6 +413,19 @@ def _set_situation(**kwargs: Any) -> dict[str, Any]: "source_request_id": adaptive_calls[0]["turn_id"], } ] + assert state_calls == [ + { + "user_id": "#V#user", + "session_id": "session-situation", + "namespace": "#V#user@org", + }, + { + "user_id": "#V#user", + "session_id": "session-situation", + "namespace": "#V#user@org", + "include_history": False, + }, + ] assistant_event_index = next( index for index, event in enumerate(events) @@ -378,21 +444,69 @@ def test_generate_returns_answer_when_situation_revision_conflicts(monkeypatch): history_calls: list[dict[str, object]] = [] app = _make_app(monkeypatch, history_calls) + state_calls: list[dict[str, Any]] = [] + initial_observations = [ + { + "observation_id": "turn:start", + "kind": "turn_state", + "summary": "The turn started from revision 2.", + } + ] + concurrent_observations = [ + { + "observation_id": "turn:concurrent", + "kind": "turn_state", + "summary": "Another turn updated the shared situation.", + } + ] + concurrent_observation_state = { + "schema_version": "conversation_observation_state.v1", + "retained_count": 1, + "total_count": 3, + "omitted_count": 2, + "retention_limit": 24, + } - monkeypatch.setattr( - von_routes.chat_history_service, - "get_chat_history_session_state", - lambda **kwargs: { + def _session_state(**kwargs: Any) -> dict[str, Any]: + state_calls.append(dict(kwargs)) + if len(state_calls) == 1: + return { + "session_id": kwargs["session_id"], + "history": [], + "conversation_situation": { + "text": "Loaded situation", + "revision": 2, + "source": "adaptive_turn", + "updated_by": "#V#user", + "updated_at": "2026-07-29T08:00:00+00:00", + }, + "conversation_observations": initial_observations, + "conversation_observation_state": { + "schema_version": "conversation_observation_state.v1", + "retained_count": 1, + "total_count": 1, + "omitted_count": 0, + "retention_limit": 24, + }, + } + return { "session_id": kwargs["session_id"], "history": [], "conversation_situation": { - "text": "Loaded situation", - "revision": 2, + "text": "Concurrently updated situation", + "revision": 3, "source": "adaptive_turn", - "updated_by": "#V#user", - "updated_at": "2026-07-29T08:00:00+00:00", + "updated_by": "#V#other", + "updated_at": "2026-07-29T08:01:00+00:00", }, - }, + "conversation_observations": concurrent_observations, + "conversation_observation_state": concurrent_observation_state, + } + + monkeypatch.setattr( + von_routes.chat_history_service, + "get_chat_history_session_state", + _session_state, ) monkeypatch.setattr( von_routes, @@ -427,7 +541,134 @@ def test_generate_returns_answer_when_situation_revision_conflicts(monkeypatch): ) assert response.status_code == 200 - assert response.get_json()["response"] == "useful answer" + body = response.get_json() + assert body["response"] == "useful answer" + assert body["conversation_situation"]["text"] == ( + "Concurrently updated situation" + ) + assert body["conversation_observations"] == concurrent_observations + assert ( + body["conversation_observation_state"] + == concurrent_observation_state + ) + assert state_calls == [ + { + "user_id": "#V#user", + "session_id": "session-conflict", + "namespace": "#V#user@org", + }, + { + "user_id": "#V#user", + "session_id": "session-conflict", + "namespace": "#V#user@org", + "include_history": False, + }, + ] + + +def test_generate_preserves_turn_start_carrier_when_read_back_fails(monkeypatch): + from src.backend.server.routes import von_routes + + history_calls: list[dict[str, object]] = [] + app = _make_app(monkeypatch, history_calls) + state_calls: list[dict[str, Any]] = [] + initial_situation = { + "text": "Loaded situation", + "revision": 7, + "source": "adaptive_turn", + "updated_by": "#V#user", + "updated_at": "2026-07-29T08:00:00+00:00", + } + initial_observations = [ + { + "observation_id": "turn:start", + "kind": "turn_state", + "summary": "This belongs to the loaded situation.", + } + ] + initial_observation_state = { + "schema_version": "conversation_observation_state.v1", + "retained_count": 1, + "total_count": 4, + "omitted_count": 3, + "retention_limit": 24, + } + + def _session_state(**kwargs: Any) -> dict[str, Any]: + state_calls.append(dict(kwargs)) + if len(state_calls) == 1: + return { + "session_id": kwargs["session_id"], + "history": [], + "conversation_situation": initial_situation, + "conversation_observations": initial_observations, + "conversation_observation_state": initial_observation_state, + } + raise RuntimeError("read-back unavailable") + + monkeypatch.setattr( + von_routes.chat_history_service, + "get_chat_history_session_state", + _session_state, + ) + monkeypatch.setattr( + von_routes, + "execute_adaptive_turn", + lambda **_kwargs: AdaptiveTurnResult( + response_text="useful answer despite read failure", + extra_messages=(), + tool_invocations=(), + aux_llm_calls=(), + conversation_situation="A newly proposed situation", + ), + ) + monkeypatch.setattr( + von_routes.chat_history_service, + "set_chat_history_conversation_situation", + lambda **kwargs: { + "updated": True, + "matched": True, + "conflict": False, + "expected_revision": kwargs["expected_revision"], + "current_revision": 8, + "session_id": kwargs["session_id"], + "conversation_situation": { + "text": kwargs["text"], + "revision": 8, + "source": "adaptive_turn", + "updated_by": "#V#user", + "updated_at": "2026-07-29T08:02:00+00:00", + }, + }, + ) + + response = app.test_client().post( + "/von/generate", + json={ + "prompt": "Continue", + "conversation_session_id": "session-read-failure", + }, + ) + + assert response.status_code == 200 + body = response.get_json() + assert body["response"] == "useful answer despite read failure" + assert body["conversation_situation"] == initial_situation + assert body["conversation_observations"] == initial_observations + assert body["conversation_observation_state"] == initial_observation_state + assert state_calls == [ + { + "user_id": "#V#user", + "session_id": "session-read-failure", + "namespace": "#V#user@org", + }, + { + "user_id": "#V#user", + "session_id": "session-read-failure", + "namespace": "#V#user@org", + "include_history": False, + }, + ] def test_shared_generate_uses_owner_situation_and_owner_storage_copy(monkeypatch): @@ -489,7 +730,13 @@ def _invitee_history(*args: Any, **kwargs: Any) -> list[dict[str, Any]]: "user_id": "#V#owner", "session_id": "shared-session", "namespace": "#V#owner@org", - } + }, + { + "user_id": "#V#owner", + "session_id": "shared-session", + "namespace": "#V#owner@org", + "include_history": False, + }, ] assert invitee_history_calls == [ ("#V#user", "shared-session", {"namespace": "#V#user@org"}) diff --git a/tests/frontend/chatTabConversationLlmTelemetryCopy.test.js b/tests/frontend/chatTabConversationLlmTelemetryCopy.test.js index 6654cc2f8..62fdbb5fc 100644 --- a/tests/frontend/chatTabConversationLlmTelemetryCopy.test.js +++ b/tests/frontend/chatTabConversationLlmTelemetryCopy.test.js @@ -233,6 +233,9 @@ describe('chat conversation info copy control', () => { 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.agent_instructions.steps).toContain( + 'Call chat_history_get_segments for the bounded conversation-carrier projection, including the shared situation, retained exact observations, observation state, and transcript segments.' + ); expect(payload.retrieval_status).toBe('server_delegation_available'); }); });