From 564d0aa26b08a89b6464e2babba49c066d1442c8 Mon Sep 17 00:00:00 2001 From: witbrock Date: Sun, 14 Jun 2026 13:59:13 -0400 Subject: [PATCH] JVNAUTOSCI-2517 Restore stable live LLM exchange/call IDs The live Thinking card groups a prepared/sent/received/completed/failed LLM lifecycle (and each fallback attempt) by call_id / llm_exchange_id, with von_routes preserving those IDs from progress updates and chatTab.js grouping on them. The orchestrator producer side was missing, so live LLM rows fell back to fragile composite keys (test_orchestrator_live_llm_call_ids was red: the _build_live_llm_exchange_id / _build_live_llm_call_id helpers were gone and _build_live_llm_progress_payload no longer accepted the IDs). Restore the producer contract: - _build_live_llm_exchange_id: deterministic per-exchange id (stage, workflow stage, prepared timestamp, request shape) prefixed "llm-". - _build_live_llm_call_id: per-attempt id "{exchange}:attempt:{n}". - _build_live_llm_progress_payload: accept and emit llm_exchange_id/call_id. - Thread a single exchange id through both _run_llm_with_fallbacks and _run_llm_with_tools_fallbacks: the prepared emit carries it directly, and attempt_meta (spread into every per-attempt lifecycle event) carries the exchange id plus the attempt's call id. Adds a production-wiring test proving the orchestrator emits one shared exchange id with distinct per-attempt call ids through the real fallback path, alongside the restored helper/payload unit contract. Co-Authored-By: Claude Opus 4.8 --- .../integrations/internal_mcp/orchestrator.py | 98 +++++++++++++++++ ...t_orchestrator_model_fallback_execution.py | 102 ++++++++++++++++++ 2 files changed, 200 insertions(+) diff --git a/src/backend/integrations/internal_mcp/orchestrator.py b/src/backend/integrations/internal_mcp/orchestrator.py index e5846022b..ad7260482 100644 --- a/src/backend/integrations/internal_mcp/orchestrator.py +++ b/src/backend/integrations/internal_mcp/orchestrator.py @@ -16784,6 +16784,16 @@ def _run_llm_with_fallbacks( context_telemetry=context_telemetry, ) request_prepared_at_utc = self._utc_now_iso() + # JVNAUTOSCI-2517: one stable exchange id per prepared request; each + # fallback attempt gets a derived call id so the live Thinking card + # groups the whole lifecycle as a single exchange. + live_llm_exchange_id = self._build_live_llm_exchange_id( + stage=stage, + workflow_stage_id=workflow_stage_id, + prepared_at_utc=request_prepared_at_utc, + request_telemetry=request_telemetry, + ) + prepared_call_id = self._build_live_llm_call_id(live_llm_exchange_id, 1) if callable(emit_progress): emit_progress( { @@ -16793,6 +16803,8 @@ def _run_llm_with_fallbacks( **self._build_live_llm_progress_payload( request_telemetry=request_telemetry, request_state="prepared", + llm_exchange_id=live_llm_exchange_id, + call_id=prepared_call_id, prepared_at_utc=request_prepared_at_utc, ), } @@ -16836,6 +16848,13 @@ def _run_llm_with_fallbacks( "model_source": candidate.source, "requested_model": default_model, "effective_model": model_name, + # JVNAUTOSCI-2517: carry the stable exchange/attempt identity on + # every per-attempt live lifecycle event (sent/received/ + # completed/cancelled/failed) so the Thinking card groups them. + "llm_exchange_id": live_llm_exchange_id, + "call_id": self._build_live_llm_call_id( + live_llm_exchange_id, attempt_no + ), } if provider: attempt_meta["provider"] = provider @@ -17594,18 +17613,78 @@ def _build_llm_request_telemetry( def _utc_now_iso() -> str: return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") + @staticmethod + def _build_live_llm_exchange_id( + *, + stage: str | None, + workflow_stage_id: str | None, + prepared_at_utc: str | None, + request_telemetry: Mapping[str, Any] | None, + ) -> str: + """Build a stable identity for one live LLM exchange (JVNAUTOSCI-2517). + + The same prepared request (same stage, workflow stage, prepared + timestamp, and request shape) always yields the same id so the live + Thinking card can group a prepared/sent/received/completed/failed + lifecycle -- and every fallback attempt -- as one exchange instead of + relying on fragile composite fallback keys. + """ + + prompt_hash: str | None = None + context_message_count: Any = None + tool_count: Any = None + if isinstance(request_telemetry, Mapping): + prompt_capture = request_telemetry.get("prompt") + prompt_text = ( + prompt_capture.get("text") + if isinstance(prompt_capture, Mapping) + and isinstance(prompt_capture.get("text"), str) + else None + ) + if prompt_text: + prompt_hash = hashlib.sha256(prompt_text.encode("utf-8")).hexdigest() + context_message_count = request_telemetry.get("context_message_count") + tool_count = request_telemetry.get("tool_count") + seed = { + "stage": stage, + "workflow_stage_id": workflow_stage_id, + "prepared_at_utc": prepared_at_utc, + "prompt_sha256": prompt_hash, + "context_message_count": context_message_count, + "tool_count": tool_count, + } + rendered = json.dumps(seed, sort_keys=True, separators=(",", ":"), default=str) + return f"llm-{hashlib.sha256(rendered.encode('utf-8')).hexdigest()[:16]}" + + @staticmethod + def _build_live_llm_call_id( + exchange_id: str, + attempt_no: int | None, + ) -> str: + """Build a per-attempt call id within an exchange (JVNAUTOSCI-2517).""" + + if isinstance(attempt_no, int) and attempt_no > 0: + return f"{exchange_id}:attempt:{attempt_no}" + return exchange_id + @classmethod def _build_live_llm_progress_payload( cls, *, request_telemetry: Mapping[str, Any] | None, request_state: str | None, + llm_exchange_id: str | None = None, + call_id: str | None = None, prepared_at_utc: str | None = None, sent_at_utc: str | None = None, first_output_at_utc: str | None = None, response: Any = None, ) -> dict[str, Any]: payload: dict[str, Any] = {} + if isinstance(llm_exchange_id, str) and llm_exchange_id.strip(): + payload["llm_exchange_id"] = llm_exchange_id.strip() + if isinstance(call_id, str) and call_id.strip(): + payload["call_id"] = call_id.strip() if isinstance(request_telemetry, Mapping) and request_telemetry: payload["llm_request"] = dict(request_telemetry) if isinstance(request_state, str) and request_state.strip(): @@ -17773,6 +17852,16 @@ def _run_llm_with_tools_fallbacks( context_telemetry=context_telemetry, ) request_prepared_at_utc = self._utc_now_iso() + # JVNAUTOSCI-2517: one stable exchange id per prepared request; each + # fallback attempt gets a derived call id so the live Thinking card + # groups the whole lifecycle as a single exchange. + live_llm_exchange_id = self._build_live_llm_exchange_id( + stage=stage, + workflow_stage_id=workflow_stage_id, + prepared_at_utc=request_prepared_at_utc, + request_telemetry=request_telemetry, + ) + prepared_call_id = self._build_live_llm_call_id(live_llm_exchange_id, 1) if callable(emit_progress): emit_progress( { @@ -17782,6 +17871,8 @@ def _run_llm_with_tools_fallbacks( **self._build_live_llm_progress_payload( request_telemetry=request_telemetry, request_state="prepared", + llm_exchange_id=live_llm_exchange_id, + call_id=prepared_call_id, prepared_at_utc=request_prepared_at_utc, ), } @@ -17822,6 +17913,13 @@ def _run_llm_with_tools_fallbacks( "model_source": candidate.source, "requested_model": default_model, "effective_model": model_name, + # JVNAUTOSCI-2517: carry the stable exchange/attempt identity on + # every per-attempt live lifecycle event (sent/received/ + # completed/cancelled/failed) so the Thinking card groups them. + "llm_exchange_id": live_llm_exchange_id, + "call_id": self._build_live_llm_call_id( + live_llm_exchange_id, attempt_no + ), } if provider: attempt_meta["provider"] = provider diff --git a/tests/backend/test_orchestrator_model_fallback_execution.py b/tests/backend/test_orchestrator_model_fallback_execution.py index 088a726af..fbd6a828e 100644 --- a/tests/backend/test_orchestrator_model_fallback_execution.py +++ b/tests/backend/test_orchestrator_model_fallback_execution.py @@ -345,6 +345,108 @@ def _record_llm_call(**payload: Any) -> None: assert recorded_calls[0]["note"] == "candidate reachability probe failed; trying fallback" +def test_run_llm_with_fallbacks_emits_stable_live_llm_exchange_identity( + monkeypatch, +) -> None: + # JVNAUTOSCI-2517: the live Thinking card groups a prepared/sent/received/ + # completed lifecycle (and every fallback attempt) by a stable exchange id + # plus per-attempt call id. Prove the orchestrator actually emits them. + orchestrator = _bare_orchestrator() + ollama_candidate = _ModelCandidate( + provider="ollama", + model="granite3.3:2b", + raw="ollama:granite3.3:2b", + source="policy", + host="http://localhost:11434", + ) + openai_candidate = _ModelCandidate( + provider="openai", + model="gpt-5.2-chat-latest", + raw="openai:gpt-5.2-chat-latest", + source="policy", + ) + monkeypatch.setattr( + orchestrator, + "_stage_model_candidates", + lambda **_kwargs: [ollama_candidate, openai_candidate], + ) + + def _create_client_for_candidate( + candidate: _ModelCandidate, + **_kwargs: Any, + ) -> tuple[Any, str, Mapping[str, Any]]: + if candidate.provider == "ollama": + return ( + _FailingClient(), + "granite3.3:2b", + {"provider": "ollama", "model": "granite3.3:2b", "host": None}, + ) + return ( + _SuccessfulClient(), + "gpt-5.2-chat-latest", + {"provider": "openai", "model": "gpt-5.2-chat-latest", "host": None}, + ) + + monkeypatch.setattr( + orchestrator, + "_create_client_for_candidate", + _create_client_for_candidate, + ) + monkeypatch.setattr( + orchestrator, + "_probe_model_candidate_reachability", + lambda **_kwargs: None, + ) + monkeypatch.setattr( + orchestrator, + "_invoke_with_llm_heartbeat", + lambda *, call, **_kwargs: call(), + ) + + progress_events: list[dict[str, Any]] = [] + orchestrator._run_llm_with_fallbacks( + stage="buttonify", + prompt="Return quick-reply options", + context=[], + default_client=object(), + default_model="gpt-5.2-chat-latest", + policy_state=_policy_state(), + registry_snapshot=None, + user_concept_id=None, + org_concept_id=None, + llm_calls_log=[], + aux_log=[], + record_llm_call=lambda **_payload: None, + emit_progress=lambda payload: progress_events.append(dict(payload)), + workflow_stage_id="tool_plan", + ) + + prepared = next( + e for e in progress_events if e.get("status") == "llm_request_prepared" + ) + exchange_id = prepared["llm_exchange_id"] + assert exchange_id.startswith("llm-") + assert prepared["call_id"] == f"{exchange_id}:attempt:1" + + # Every live LLM lifecycle event shares the one exchange id, and each + # attempt's call id is derived from its fallback attempt number. + lifecycle = [e for e in progress_events if e.get("llm_exchange_id")] + assert len(lifecycle) >= 3 + assert {e["llm_exchange_id"] for e in lifecycle} == {exchange_id} + for event in lifecycle: + attempt_no = event.get("fallback_attempt_no") + if isinstance(attempt_no, int) and attempt_no > 0: + assert event["call_id"] == f"{exchange_id}:attempt:{attempt_no}" + + # The successful second attempt is reported under attempt-2's call id. + completed = next( + e + for e in progress_events + if e.get("status") == "llm_call_end" and e.get("success") is True + ) + assert completed["call_id"] == f"{exchange_id}:attempt:2" + + def test_run_llm_with_fallbacks_marks_policy_primary_active_llm_selection( monkeypatch, ) -> None: