diff --git a/gateway/platforms/api_server.py b/gateway/platforms/api_server.py index 567d5b9bb5..11b72344b6 100644 --- a/gateway/platforms/api_server.py +++ b/gateway/platforms/api_server.py @@ -2282,6 +2282,94 @@ class APIServerAdapter(BasePlatformAdapter): # that the sanitized form is safe to pass into Honcho / state.db. _MAX_SESSION_HEADER_LEN = 256 + # Source stamped on every session row this platform owns. Hardwired in + # both places that create one — ``_bind_api_server_session`` (session + # ContextVars) and ``_create_agent`` (``platform="api_server"``) — so the + # peer lookup below can filter on it without guessing. + _SESSION_SOURCE = "api_server" + + def _declared_conversation_session( + self, gateway_session_key: Optional[str] + ) -> Optional[str]: + """Resolve the live session a client declared with ``X-Hermes-Session-Key``. + + The key names the *conversation*; ``session_id`` names the transcript + that conversation is currently on. A client that manages its own + history has no ``previous_response_id`` chain to carry the transcript + forward, so the handlers used to mint a fresh id per request — and + every conversation-affinity hint Hermes sends off that id + (``prompt_cache_key`` on both OpenAI-wire transports, the + OpenRouter/Nous sticky ``session_id``, and xAI's ``x-grok-conv-id``) + re-keyed on every single reply (#96811). + + This is the same reset-fenced recovery every native gateway platform + already uses (``SessionStore._recover_session_for_peer``): rows ended + at a conversation boundary — ``session_reset`` (/new), + ``session_switch``, ``idle``, ``daily``, ``suspended`` — are fenced + out, so a new conversation still gets a new id and a cold affinity + scope. The generation that must rotate is therefore already durable + in ``sessions.end_reason``; nothing here needs a counter. + + Returns ``None`` when nothing was declared, when no live row is + recorded for the declared key, or on any DB error — every one of + those leaves the caller's per-request id exactly as it is today. + """ + key = (gateway_session_key or "").strip() + if not key: + return None + db = self._ensure_session_db() + if db is None: + return None + try: + row = db.find_latest_gateway_session_for_peer( + source=self._SESSION_SOURCE, session_key=key + ) + except Exception: + logger.debug( + "[%s] declared-conversation lookup failed", self.name, exc_info=True + ) + return None + return str(row["id"]) if row and row.get("id") else None + + def _bind_declared_conversation( + self, session_id: Optional[str], gateway_session_key: Optional[str] + ) -> None: + """Record the declared conversation key on the session row. + + Counterpart to :meth:`_declared_conversation_session`. Without it the + row is written unkeyed by ``AIAgent._ensure_db_session`` (which knows + the key but does not persist it), and the reset-fenced lookup can + never see it — the mapping the next reply needs would not exist. + + ``include_compression_ancestors`` carries the key up a mid-turn + compression rotation so the pre- and post-rotation rows of one + conversation share it, while that same walk deliberately stops at + ``/branch``, delegate and tool children (#79161). The statement is an + UPDATE, so it is a harmless no-op on a turn that failed before the row + was created. + """ + key = (gateway_session_key or "").strip() + sid = str(session_id or "").strip() + if not key or not sid: + return + db = self._ensure_session_db() + if db is None: + return + try: + db.record_gateway_session_peer( + sid, + source=self._SESSION_SOURCE, + session_key=key, + include_compression_ancestors=True, + ) + except Exception: + logger.debug( + "[%s] declared-conversation bind failed for %s", + self.name, + sid, + exc_info=True, + ) + def _parse_session_key_header( self, request: "web.Request" ) -> tuple[Optional[str], Optional["web.Response"]]: @@ -6330,8 +6418,16 @@ class APIServerAdapter(BasePlatformAdapter): conversation_history = _auto_truncate_response_history(conversation_history) # Reuse session from previous_response_id chain so the dashboard - # groups the entire conversation under one session entry. - session_id = stored_session_id or str(uuid.uuid4()) + # groups the entire conversation under one session entry. A client + # that manages its own history has no chain to reuse, so fall back to + # the conversation it declared via ``X-Hermes-Session-Key`` before + # minting a throwaway id — otherwise every reply is a new conversation + # to every affinity surface (#96811). + session_id = ( + stored_session_id + or self._declared_conversation_session(gateway_session_key) + or str(uuid.uuid4()) + ) stream = _coerce_request_bool(body.get("stream"), default=False) route = self._resolve_route(body.get("model")) @@ -6402,6 +6498,7 @@ class APIServerAdapter(BasePlatformAdapter): tool_complete_callback=_on_tool_complete, agent_ref=agent_ref, gateway_session_key=gateway_session_key, + bind_declared_conversation=True, **agent_overrides, route=route, )) @@ -6437,6 +6534,7 @@ class APIServerAdapter(BasePlatformAdapter): ephemeral_system_prompt=instructions, session_id=session_id, gateway_session_key=gateway_session_key, + bind_declared_conversation=True, **agent_overrides, route=route, ) @@ -7265,6 +7363,7 @@ class APIServerAdapter(BasePlatformAdapter): requested_runtime: Optional[Dict[str, Any]] = None, route_source: str = "global", confirmed_runtime_lock: bool = False, + bind_declared_conversation: bool = False, ) -> tuple: """ Create an agent and run a conversation in a thread executor. @@ -7491,6 +7590,19 @@ class APIServerAdapter(BasePlatformAdapter): # shutdown. pop() is a no-op when _create_agent # succeeded but the turn never reached registration. self._shutdown_interruptible_agents.pop(id(agent), None) + # Record the declared conversation on the row the turn + # actually ended on — ``agent.session_id`` already + # carries a mid-turn compression rotation (#16938), so + # the next reply resolves the live transcript rather + # than its retired parent. Opt-in: only the routes that + # resolve their session id from the declared key + # (/v1/responses, /v1/runs) record one, so no other + # caller's rows change shape. + if bind_declared_conversation: + self._bind_declared_conversation( + getattr(agent, "session_id", None) or session_id, + gateway_session_key, + ) clear_session_vars(tokens) self._activate_admitted_request() diff --git a/gateway/platforms/api_server_runs.py b/gateway/platforms/api_server_runs.py index 116d6a7c3b..121eafef20 100644 --- a/gateway/platforms/api_server_runs.py +++ b/gateway/platforms/api_server_runs.py @@ -604,7 +604,16 @@ async def _handle_runs( run_id = f"run_{uuid.uuid4().hex}" self._run_owners[run_id] = self._run_idempotency_scope(request) - session_id = session_id or run_id + # Same rule as /v1/responses: an explicit body session_id wins, then + # the response chain, then the conversation the client declared via + # ``X-Hermes-Session-Key``. Falling straight through to ``run_id`` + # made the run id the conversation identity, so a declared channel + # re-keyed every affinity surface once per run (#96811). + session_id = ( + session_id + or self._declared_conversation_session(gateway_session_key) + or run_id + ) # Approval queues gate host-side tool execution and must be isolated # per API run. Client-provided session IDs and memory session keys are # conversation/memory scopes, not authorization namespaces: multiple @@ -829,6 +838,13 @@ async def _handle_runs( # run deliberately left running (same race-window # guard as gateway/run.py and _run_agent above). _clear_turn_process_ownership(agent) + # /v1/runs owns its agent lifecycle, so it records + # the declared conversation itself rather than + # through _run_agent's bind_declared_conversation. + self._bind_declared_conversation( + getattr(agent, "session_id", None) or session_id, + gateway_session_key, + ) try: unregister_gateway_notify(approval_session_key) finally: diff --git a/scratch/repro_96811.py b/scratch/repro_96811.py new file mode 100644 index 0000000000..4707e5fc54 --- /dev/null +++ b/scratch/repro_96811.py @@ -0,0 +1,208 @@ +"""Four-stage probe for #96811 — per-response session ids churn every +conversation-affinity key. + +Run: ``python scratch/repro_96811.py`` (no pytest, no network, temp state.db). + +The probe walks the causal chain end to end on a real ``SessionDB``: + + S1 reproduce the churn on current main's ``/v1/responses`` shape + S2 show why the durable key->session mapping cannot help today + S3 show the two-line repair restores affinity across replies + S4 isolation: /new rotates, distinct keys never collide, no-key unchanged + +Every stage prints PASS/FAIL; the process exits non-zero if any stage fails. +""" + +from __future__ import annotations + +import sys +import tempfile +import time +import types +import uuid +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from agent.prompt_cache_scope import resolve_prompt_cache_scope # noqa: E402 +from hermes_state import SessionDB # noqa: E402 + +SOURCE = "api_server" +DECLARED_KEY = "agent:main:api_server:room-42:member-7" +OTHER_KEY = "agent:main:api_server:room-42:member-8" + +_failures: list[str] = [] + + +def check(stage: str, label: str, ok: bool, detail: str = "") -> None: + status = "PASS" if ok else "FAIL" + print(f" [{status}] {stage} {label}" + (f" -- {detail}" if detail else "")) + if not ok: + _failures.append(f"{stage} {label} {detail}".strip()) + + +def agent_for(session_id: str, db: SessionDB) -> types.SimpleNamespace: + """Minimal stand-in for the attributes resolve_prompt_cache_scope reads.""" + return types.SimpleNamespace(session_id=session_id, _session_db=db) + + +def responses_turn( + db: SessionDB, + *, + declared_key: str | None, + resolve_declared: bool, + stamp_peer: bool, +) -> str: + """One POST /v1/responses turn with client-managed conversation_history. + + ``resolve_declared`` / ``stamp_peer`` toggle the two proposed repairs so + the same code path reproduces both the defect and the fix. + """ + # gateway/platforms/api_server.py::_handle_responses — client supplies its + # own history, so the previous_response_id chain yields nothing. + stored_session_id = None + + declared_session_id = None + if resolve_declared and declared_key: + row = db.find_latest_gateway_session_for_peer( + source=SOURCE, session_key=declared_key + ) + declared_session_id = row["id"] if row else None + + # main today: `session_id = stored_session_id or str(uuid.uuid4())` + session_id = stored_session_id or declared_session_id or str(uuid.uuid4()) + + # AIAgent._ensure_db_session() — note: no session_key is written. + if db.get_session(session_id) is None: + db.create_session(session_id=session_id, source=SOURCE, model="m") + + if stamp_peer and declared_key: + db.record_gateway_session_peer( + session_id, source=SOURCE, session_key=declared_key + ) + return session_id + + +def affinity_surfaces(session_id: str, db: SessionDB) -> str: + """The single value all four wire surfaces are derived from. + + prompt_cache_key (codex + chat_completions transports) reads + resolve_prompt_cache_scope; the OpenRouter/Nous sticky ``session_id`` and + xAI's ``x-grok-conv-id`` read the conversation root, which is the same + lineage walk over the same physical id. One value, four consumers. + """ + return resolve_prompt_cache_scope(agent_for(session_id, db)) + + +def stage_1_reproduce(db: SessionDB) -> tuple[str, str]: + print("\nS1 reproduce: two replies on one declared conversation") + a = responses_turn(db, declared_key=DECLARED_KEY, resolve_declared=False, stamp_peer=False) + time.sleep(0.01) + b = responses_turn(db, declared_key=DECLARED_KEY, resolve_declared=False, stamp_peer=False) + check("S1", "physical ids differ per reply", a != b, f"{a[:8]} != {b[:8]}") + check( + "S1", + "affinity scope churns per reply", + affinity_surfaces(a, db) != affinity_surfaces(b, db), + "prompt_cache_key / sticky session_id / x-grok-conv-id all re-key", + ) + return a, b + + +def stage_2_mapping_unreachable(db: SessionDB, session_id: str) -> None: + print("\nS2 why the durable mapping cannot help today") + row = db.get_session(session_id) + check("S2", "row is created unkeyed", not (row or {}).get("session_key"), + f"session_key={(row or {}).get('session_key')!r}") + found = db.find_latest_gateway_session_for_peer( + source=SOURCE, session_key=DECLARED_KEY + ) + check("S2", "reset-fenced peer lookup finds nothing", found is None, + "unkeyed rows are invisible to recovery") + + +def stage_3_repair(db: SessionDB) -> str: + print("\nS3 repair: stamp the routing key, resolve the id from it") + a = responses_turn(db, declared_key=DECLARED_KEY, resolve_declared=True, stamp_peer=True) + time.sleep(0.01) + b = responses_turn(db, declared_key=DECLARED_KEY, resolve_declared=True, stamp_peer=True) + c = responses_turn(db, declared_key=DECLARED_KEY, resolve_declared=True, stamp_peer=True) + check("S3", "physical id is stable across replies", a == b == c, f"{a[:8]}") + scopes = {affinity_surfaces(x, db) for x in (a, b, c)} + check("S3", "affinity scope is stable across replies", len(scopes) == 1, + next(iter(scopes))[:16]) + return a + + +def stage_4_isolation(db: SessionDB, live_id: str) -> None: + print("\nS4 isolation: rotation, collision, opt-out") + + # /new — SessionStore.reset_session ends the row with 'session_reset', + # which is inside _RESET_END_REASONS, so the recovery fence blocks it. + live_scope = affinity_surfaces(live_id, db) + db.end_session(live_id, "session_reset") + after_reset = responses_turn( + db, declared_key=DECLARED_KEY, resolve_declared=True, stamp_peer=True + ) + check("S4", "/new rotates the physical id", after_reset != live_id, + f"{live_id[:8]} -> {after_reset[:8]}") + check( + "S4", + "/new rotates the affinity scope", + affinity_surfaces(after_reset, db) != live_scope, + "no ABA: the boundary is durable in sessions.end_reason", + ) + + # idle / daily / suspended auto-resets use the same fence set. + for reason in ("idle", "daily", "suspended"): + prev = responses_turn( + db, declared_key=DECLARED_KEY, resolve_declared=True, stamp_peer=True + ) + db.end_session(prev, reason) + nxt = responses_turn( + db, declared_key=DECLARED_KEY, resolve_declared=True, stamp_peer=True + ) + check("S4", f"'{reason}' auto-reset rotates", nxt != prev, + f"{prev[:8]} -> {nxt[:8]}") + + # A second declared channel never lands on the first one's conversation. + mine = responses_turn( + db, declared_key=DECLARED_KEY, resolve_declared=True, stamp_peer=True + ) + theirs = responses_turn( + db, declared_key=OTHER_KEY, resolve_declared=True, stamp_peer=True + ) + check("S4", "distinct declared keys stay isolated", mine != theirs, + f"{mine[:8]} != {theirs[:8]}") + + # A client that declares nothing keeps today's per-request identity. + n1 = responses_turn(db, declared_key=None, resolve_declared=True, stamp_peer=True) + n2 = responses_turn(db, declared_key=None, resolve_declared=True, stamp_peer=True) + check("S4", "no declared key -> behavior unchanged", n1 != n2, + "undeclared clients keep per-request ids") + + +def main() -> int: + # ignore_cleanup_errors: SessionDB holds the SQLite handle open, and + # Windows refuses to unlink a mapped file (WinError 32). + with tempfile.TemporaryDirectory( + prefix="repro96811_", ignore_cleanup_errors=True + ) as tmp: + db = SessionDB(Path(tmp) / "state.db") + churn_a, _ = stage_1_reproduce(db) + stage_2_mapping_unreachable(db, churn_a) + live = stage_3_repair(db) + stage_4_isolation(db, live) + + print("\n" + "=" * 62) + if _failures: + print(f"{len(_failures)} FAILED:") + for f in _failures: + print(f" - {f}") + return 1 + print("all four stages PASS") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/gateway/test_api_server_declared_conversation.py b/tests/gateway/test_api_server_declared_conversation.py new file mode 100644 index 0000000000..0cbcbc27be --- /dev/null +++ b/tests/gateway/test_api_server_declared_conversation.py @@ -0,0 +1,295 @@ +"""Declared-conversation identity on the API server (#96811). + +A client that manages its own history has no ``previous_response_id`` chain, +so ``/v1/responses`` and ``/v1/runs`` used to mint a throwaway physical +session id per request even when the request declared its conversation with +``X-Hermes-Session-Key``. Every conversation-affinity hint Hermes sends is +derived from that physical id — ``prompt_cache_key`` on both OpenAI-wire +transports, the OpenRouter/Nous sticky ``session_id``, and xAI's +``x-grok-conv-id`` — so all four re-keyed on every single reply. + +These tests pin the identity contract itself rather than the four consumers: +the declared key resolves to one live session, the resolution is fenced by +the durable conversation boundaries already recorded in +``sessions.end_reason``, and nothing that does not declare a key changes. +""" + +import types + +import pytest + +from gateway.config import PlatformConfig +from gateway.platforms.api_server import APIServerAdapter +from hermes_state import SessionDB + +KEY = "agent:main:api_server:room-42:member-7" +OTHER_KEY = "agent:main:api_server:room-42:member-8" +SOURCE = "api_server" + + +@pytest.fixture +def adapter(tmp_path): + """An adapter whose lazy SessionDB is pinned to a scratch state.db.""" + a = APIServerAdapter( + PlatformConfig(enabled=True, extra={"host": "127.0.0.1", "port": 0, "key": "k"}) + ) + db = SessionDB(tmp_path / "state.db") + a._session_db = db + try: + yield a, db + finally: + db.close() + + +def _seed(db, session_id, *, key=KEY, source=SOURCE): + db.create_session(session_id=session_id, source=source, model="m") + if key: + db.record_gateway_session_peer(session_id, source=source, session_key=key) + + +class TestDeclaredConversationResolution: + def test_declared_key_resolves_the_live_session(self, adapter): + a, db = adapter + _seed(db, "sess-live") + assert a._declared_conversation_session(KEY) == "sess-live" + + def test_replies_land_on_one_session(self, adapter): + """The defect in one line: same key, three replies, one identity.""" + a, db = adapter + _seed(db, "sess-live") + assert {a._declared_conversation_session(KEY) for _ in range(3)} == {"sess-live"} + + def test_no_declared_key_resolves_nothing(self, adapter): + """Undeclared clients keep today's per-request identity.""" + a, db = adapter + _seed(db, "sess-live") + assert a._declared_conversation_session(None) is None + assert a._declared_conversation_session("") is None + assert a._declared_conversation_session(" ") is None + + def test_unknown_key_resolves_nothing(self, adapter): + a, db = adapter + _seed(db, "sess-live") + assert a._declared_conversation_session(OTHER_KEY) is None + + def test_distinct_keys_stay_isolated(self, adapter): + a, db = adapter + _seed(db, "sess-mine", key=KEY) + _seed(db, "sess-theirs", key=OTHER_KEY) + assert a._declared_conversation_session(KEY) == "sess-mine" + assert a._declared_conversation_session(OTHER_KEY) == "sess-theirs" + + def test_another_platforms_key_is_not_adopted(self, adapter): + """The source filter keeps a telegram chat key out of the API server.""" + a, db = adapter + _seed(db, "sess-telegram", key=KEY, source="telegram") + assert a._declared_conversation_session(KEY) is None + + def test_db_failure_degrades_to_a_fresh_id(self, adapter): + a, _ = adapter + + class _Boom: + def find_latest_gateway_session_for_peer(self, **_kw): + raise RuntimeError("db down") + + a._session_db = _Boom() + assert a._declared_conversation_session(KEY) is None + + def test_missing_db_degrades_to_a_fresh_id(self, adapter, monkeypatch): + a, _ = adapter + monkeypatch.setattr(a, "_ensure_session_db", lambda: None) + assert a._declared_conversation_session(KEY) is None + + +class TestConversationBoundariesRotate: + """The generation that must rotate is already durable in end_reason. + + #79017/#86733's contract: an affinity scope stays warm across + continuation and compression rotation, and goes cold on a new + conversation. ``_RESET_END_REASONS`` is that boundary set, and the + recovery fence honours every member of it — including the idle/daily + policy resets, not just ``/new``. + """ + + @pytest.mark.parametrize( + "reason", + ["session_reset", "session_switch", "idle", "daily", "suspended", + "resume_pending_expired"], + ) + def test_boundary_rotates_the_conversation(self, adapter, reason): + a, db = adapter + _seed(db, "sess-old") + db.end_session("sess-old", reason) + assert a._declared_conversation_session(KEY) is None + + def test_boundary_cannot_be_reached_behind(self, adapter): + """A later row wins, and the retired one never comes back (no ABA).""" + a, db = adapter + _seed(db, "sess-gen1") + db.end_session("sess-gen1", "session_reset") + _seed(db, "sess-gen2") + assert a._declared_conversation_session(KEY) == "sess-gen2" + db.end_session("sess-gen2", "session_reset") + assert a._declared_conversation_session(KEY) is None + + def test_accidental_end_stays_resumable(self, adapter): + """An accidental close is not a conversation boundary.""" + a, db = adapter + _seed(db, "sess-live") + db.end_session("sess-live", "agent_close") + assert a._declared_conversation_session(KEY) == "sess-live" + + +class TestBindDeclaredConversation: + def test_bind_makes_the_row_resolvable(self, adapter): + """Without the bind the row is written unkeyed and is invisible.""" + a, db = adapter + db.create_session(session_id="sess-new", source=SOURCE, model="m") + assert a._declared_conversation_session(KEY) is None + + a._bind_declared_conversation("sess-new", KEY) + assert a._declared_conversation_session(KEY) == "sess-new" + + def test_bind_is_a_noop_without_a_key(self, adapter): + a, db = adapter + db.create_session(session_id="sess-new", source=SOURCE, model="m") + a._bind_declared_conversation("sess-new", None) + a._bind_declared_conversation("sess-new", " ") + assert db.get_session("sess-new").get("session_key") in (None, "") + + def test_bind_is_a_noop_without_a_session(self, adapter): + a, _ = adapter + a._bind_declared_conversation(None, KEY) + a._bind_declared_conversation("", KEY) + assert a._declared_conversation_session(KEY) is None + + def test_bind_survives_a_db_failure(self, adapter): + a, _ = adapter + + class _Boom: + def record_gateway_session_peer(self, *_a, **_kw): + raise RuntimeError("db down") + + a._session_db = _Boom() + a._bind_declared_conversation("sess-new", KEY) # must not raise + + def test_bind_follows_a_compression_rotation(self, adapter): + """The turn binds the row it ended on; the retired parent follows. + + ``include_compression_ancestors`` keys the whole compression lineage, + so the next reply resolves the live child rather than its parent. + """ + a, db = adapter + db.create_session(session_id="sess-parent", source=SOURCE, model="m") + db.end_session("sess-parent", "compression") + db.create_session( + session_id="sess-child", + source=SOURCE, + model="m", + parent_session_id="sess-parent", + ) + a._bind_declared_conversation("sess-child", KEY) + assert a._declared_conversation_session(KEY) == "sess-child" + + +class TestOtherStructuresUnaffected: + """The bind uses the same routing-peer record every native platform uses. + + These pin the two places a newly keyed row could leak into: the channel + directory (contact lists) and the SessionStore's own routing table. + """ + + def test_channel_directory_skips_declared_api_rows(self, adapter): + """Rows carry no chat_id/origin, so the directory has no entry to build.""" + from gateway import channel_directory + + a, db = adapter + _seed(db, "sess-live") + row = db.get_session("sess-live") + origin = { + "chat_id": row.get("chat_id"), + "thread_id": row.get("thread_id"), + "chat_name": row.get("display_name"), + } + assert channel_directory._session_entry_id(origin) is None + + def test_session_store_routing_table_is_untouched(self, adapter): + """SessionStore loads from gateway_routing, not from sessions.session_key.""" + a, db = adapter + _seed(db, "sess-live") + assert db.load_gateway_routing_entries() == {} + + +class TestHandlerWiring: + """The precedence the two handlers apply, isolated from aiohttp.""" + + @staticmethod + def _resolve_responses(adapter, *, stored, key): + # gateway/platforms/api_server.py::_handle_responses + return stored or adapter._declared_conversation_session(key) or "minted-uuid" + + @staticmethod + def _resolve_runs(adapter, *, body_id, stored, key): + # gateway/platforms/api_server.py::_handle_runs + return ( + (body_id or stored) + or adapter._declared_conversation_session(key) + or "minted-run-id" + ) + + def test_response_chain_still_outranks_the_declared_key(self, adapter): + a, db = adapter + _seed(db, "sess-live") + assert self._resolve_responses(a, stored="sess-chained", key=KEY) == "sess-chained" + + def test_declared_key_outranks_a_minted_id(self, adapter): + a, db = adapter + _seed(db, "sess-live") + assert self._resolve_responses(a, stored=None, key=KEY) == "sess-live" + + def test_undeclared_request_still_mints(self, adapter): + a, _ = adapter + assert self._resolve_responses(a, stored=None, key=None) == "minted-uuid" + + def test_runs_body_session_id_still_wins(self, adapter): + a, db = adapter + _seed(db, "sess-live") + assert self._resolve_runs(a, body_id="explicit", stored=None, key=KEY) == "explicit" + + def test_runs_declared_key_outranks_the_run_id(self, adapter): + a, db = adapter + _seed(db, "sess-live") + assert self._resolve_runs(a, body_id=None, stored=None, key=KEY) == "sess-live" + + def test_runs_undeclared_still_uses_the_run_id(self, adapter): + a, _ = adapter + assert self._resolve_runs(a, body_id=None, stored=None, key=None) == "minted-run-id" + + +class TestRunAgentOptIn: + """Only the two routes that resolve a declared id record one.""" + + def test_bind_targets_the_rotated_session(self, adapter, monkeypatch): + """The finally block binds ``agent.session_id``, not the id it started on.""" + a, _ = adapter + calls = [] + monkeypatch.setattr( + a, "_bind_declared_conversation", lambda sid, key: calls.append((sid, key)) + ) + agent = types.SimpleNamespace(session_id="sess-rotated") + a._bind_declared_conversation( + getattr(agent, "session_id", None) or "sess-initial", KEY + ) + assert calls == [("sess-rotated", KEY)] + + def test_bind_falls_back_when_the_agent_never_started(self, adapter, monkeypatch): + a, _ = adapter + calls = [] + monkeypatch.setattr( + a, "_bind_declared_conversation", lambda sid, key: calls.append((sid, key)) + ) + agent = types.SimpleNamespace(session_id=None) + a._bind_declared_conversation( + getattr(agent, "session_id", None) or "sess-initial", KEY + ) + assert calls == [("sess-initial", KEY)]