810 lines
31 KiB
Python
810 lines
31 KiB
Python
"""Host-declared conversation scope on the affinity-key path (issue #96811).
|
|
|
|
A host that mints one physical ``session_id`` per RESPONSE re-keys every
|
|
conversation-affinity hint Hermes sends — ``prompt_cache_key`` on both
|
|
OpenAI-wire transports, the OpenRouter/Nous sticky ``session_id``, and xAI's
|
|
``x-grok-conv-id`` — so the conversation never lands back on the routing
|
|
bucket it warmed. Hermes cannot infer the logical conversation from the id's
|
|
syntax (#79017's failure class), but it does not have to: the host declares
|
|
it through ``gateway_session_key`` (the ``X-Hermes-Session-Key`` /
|
|
``build_session_key`` per-chat key).
|
|
|
|
These tests pin the declaration contract and the two boundaries it must not
|
|
cross: explicit fork children (``/branch``, delegate, tool) and
|
|
background-review forks, which share the parent's chat key but are separate
|
|
conversations under #79161.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from agent.portal_tags import (
|
|
get_affinity_scope,
|
|
reset_affinity_scope,
|
|
reset_conversation_context,
|
|
set_affinity_scope,
|
|
set_conversation_context,
|
|
)
|
|
from agent.prompt_cache_scope import (
|
|
declared_conversation_scope,
|
|
declared_conversation_scope_safe,
|
|
resolve_prompt_cache_scope,
|
|
)
|
|
from agent.transports.codex import _cache_scope_from_session_id, _content_cache_key
|
|
from hermes_state import SessionDB
|
|
|
|
# One room member, two consecutive replies: the Studio group-chat shape
|
|
# (``gc_run_<room>_<profile>_<name>`` truncated to 96 chars + a per-response
|
|
# UUID4 hex) and the ``POST /v1/responses`` shape (a bare ``str(uuid4())``).
|
|
RUN_1 = "gc_run_room7_default_Reviewer_11111111111141118111111111111111"
|
|
RUN_2 = "gc_run_room7_default_Reviewer_22222222222242228222222222222222"
|
|
CHAT_KEY = "agent:main:telegram:group:-100123:456"
|
|
|
|
|
|
@pytest.fixture()
|
|
def db(tmp_path):
|
|
session_db = SessionDB(db_path=tmp_path / "state.db")
|
|
try:
|
|
yield session_db
|
|
finally:
|
|
session_db.close()
|
|
|
|
|
|
def _agent(session_id, session_db=None, key=None):
|
|
return SimpleNamespace(
|
|
session_id=session_id,
|
|
_session_db=session_db,
|
|
_gateway_session_key=key,
|
|
)
|
|
|
|
|
|
def _sticky_key(session_id):
|
|
from providers import get_provider_profile
|
|
|
|
return get_provider_profile("openrouter").build_extra_body(session_id=session_id)[
|
|
"session_id"
|
|
]
|
|
|
|
|
|
def _grok_headers(session_id):
|
|
from providers import get_provider_profile
|
|
|
|
_extra_body, top_level = get_provider_profile("openrouter").build_api_kwargs_extras(
|
|
model="x-ai/grok-4",
|
|
session_id=session_id,
|
|
)
|
|
return top_level["extra_headers"]
|
|
|
|
|
|
class TestDeclaredConversationScope:
|
|
def test_per_response_ids_resolve_to_one_declared_scope(self, db):
|
|
"""THE fix: two replies of one conversation share one scope."""
|
|
db.create_session(RUN_1, source="api_server")
|
|
db.create_session(RUN_2, source="api_server")
|
|
|
|
first = resolve_prompt_cache_scope(_agent(RUN_1, db, CHAT_KEY))
|
|
second = resolve_prompt_cache_scope(_agent(RUN_2, db, CHAT_KEY))
|
|
|
|
assert first == second
|
|
assert first not in (RUN_1, RUN_2)
|
|
|
|
def test_distinct_declarations_stay_isolated(self, db):
|
|
db.create_session(RUN_1, source="api_server")
|
|
other = _agent(RUN_1, db, "agent:main:telegram:group:-100123:999")
|
|
|
|
assert resolve_prompt_cache_scope(
|
|
_agent(RUN_1, db, CHAT_KEY)
|
|
) != resolve_prompt_cache_scope(other)
|
|
|
|
def test_scope_never_carries_the_raw_key(self, db):
|
|
"""The scope leaves the process verbatim (sticky id, x-grok-conv-id).
|
|
|
|
A session id is a Hermes-internal token; a session KEY embeds the
|
|
platform, chat and user identifiers, so it is hashed first.
|
|
"""
|
|
db.create_session(RUN_1, source="api_server")
|
|
scope = resolve_prompt_cache_scope(_agent(RUN_1, db, CHAT_KEY))
|
|
|
|
assert scope.startswith("gwk_")
|
|
assert "telegram" not in scope
|
|
assert "-100123" not in scope
|
|
assert len(scope) <= 64 # provider key budget
|
|
|
|
def test_no_declaration_keeps_lineage_behavior(self, db):
|
|
"""Unchanged for every host that keeps one id per conversation."""
|
|
db.create_session("root-sess", source="webui")
|
|
db.end_session("root-sess", "compression")
|
|
db.create_session("rotated-1", source="webui", parent_session_id="root-sess")
|
|
|
|
assert resolve_prompt_cache_scope(_agent("rotated-1", db)) == "root-sess"
|
|
assert declared_conversation_scope(_agent("rotated-1", db)) is None
|
|
|
|
def test_declaration_outranks_the_lineage_root(self, db):
|
|
"""Both are stable; the declared key is stable across MORE (per-response
|
|
ids), so it wins rather than being a fallback."""
|
|
db.create_session("root-sess", source="webui")
|
|
db.end_session("root-sess", "compression")
|
|
db.create_session("rotated-1", source="webui", parent_session_id="root-sess")
|
|
|
|
scope = resolve_prompt_cache_scope(_agent("rotated-1", db, CHAT_KEY))
|
|
# Same declared conversation reached through a different physical id
|
|
# on the same peer — the property the lineage walk cannot provide.
|
|
db.create_session("rotated-2", source="webui")
|
|
assert scope == resolve_prompt_cache_scope(_agent("rotated-2", db, CHAT_KEY))
|
|
assert scope != "root-sess"
|
|
assert scope.startswith("gwk_")
|
|
|
|
def test_branch_child_ignores_the_shared_chat_key(self, db):
|
|
"""/branch keys off session_id, not the chat key — #79161 isolation."""
|
|
db.create_session("root-sess", source="telegram")
|
|
db.create_session(
|
|
"branch-child",
|
|
source="telegram",
|
|
parent_session_id="root-sess",
|
|
model_config={"_branched_from": "root-sess"},
|
|
)
|
|
|
|
assert (
|
|
resolve_prompt_cache_scope(_agent("branch-child", db, CHAT_KEY))
|
|
== "branch-child"
|
|
)
|
|
assert (
|
|
resolve_prompt_cache_scope(_agent("root-sess", db, CHAT_KEY))
|
|
!= "branch-child"
|
|
)
|
|
|
|
def test_delegate_child_ignores_the_declaration(self, db):
|
|
db.create_session("parent-sess", source="telegram")
|
|
db.create_session(
|
|
"delegate-child",
|
|
source="telegram",
|
|
parent_session_id="parent-sess",
|
|
model_config={"_delegate_from": "parent-sess"},
|
|
)
|
|
|
|
assert (
|
|
resolve_prompt_cache_scope(_agent("delegate-child", db, CHAT_KEY))
|
|
== "delegate-child"
|
|
)
|
|
|
|
def test_tool_child_ignores_the_declaration(self, db):
|
|
db.create_session("parent-sess", source="telegram")
|
|
db.create_session("tool-child", source="tool", parent_session_id="parent-sess")
|
|
|
|
assert (
|
|
resolve_prompt_cache_scope(_agent("tool-child", db, CHAT_KEY))
|
|
== "tool-child"
|
|
)
|
|
|
|
def test_background_review_fork_ignores_the_declaration(self, db):
|
|
"""The review fork clones the live runtime, key included."""
|
|
db.create_session("live-sess", source="telegram")
|
|
agent = _agent("review-fork", db, CHAT_KEY)
|
|
agent._persist_disabled = True
|
|
|
|
assert declared_conversation_scope(agent) is None
|
|
assert resolve_prompt_cache_scope(agent) == "review-fork"
|
|
|
|
def test_fork_check_failure_degrades_to_the_physical_scope(self):
|
|
"""A transient DB error must not merge a fork onto its parent's key."""
|
|
|
|
class BoomDB:
|
|
def is_explicit_fork_child(self, sid):
|
|
raise RuntimeError("db exploded")
|
|
|
|
def get_compression_lineage(self, sid):
|
|
return [sid]
|
|
|
|
agent = _agent("maybe-fork", BoomDB(), CHAT_KEY)
|
|
assert declared_conversation_scope(agent) is None
|
|
assert resolve_prompt_cache_scope(agent) == "maybe-fork"
|
|
|
|
def test_declaration_applies_before_the_row_lands(self, db):
|
|
"""turn_context resolves before _ensure_db_session persists the row."""
|
|
agent = _agent(RUN_1, db, CHAT_KEY)
|
|
assert resolve_prompt_cache_scope(agent).startswith("gwk_")
|
|
|
|
def test_blank_declarations_are_no_declaration(self, db):
|
|
db.create_session(RUN_1, source="api_server")
|
|
for blank in (None, "", " "):
|
|
assert declared_conversation_scope(_agent(RUN_1, db, blank)) is None
|
|
|
|
def test_safe_variant_never_raises(self):
|
|
class ExplodingAgent:
|
|
@property
|
|
def _gateway_session_key(self):
|
|
raise RuntimeError("hostile property")
|
|
|
|
assert declared_conversation_scope_safe(ExplodingAgent()) is None
|
|
assert declared_conversation_scope_safe(
|
|
_agent("sess", None, CHAT_KEY)
|
|
).startswith("gwk_")
|
|
|
|
|
|
class TestOneIdentityReadPerResolution:
|
|
"""The fork verdict and the row's source come from one ``sessions`` read.
|
|
|
|
Resolution is memoized per transcript segment, so this was never on the
|
|
per-API-call hot path (#79017) — but reading the same row twice per
|
|
resolution was one read too many (@teknium1 on #98811), and a ``SessionDB``
|
|
that predates the combined view has to keep the path it had.
|
|
"""
|
|
|
|
def test_the_identity_row_is_read_once(self, db):
|
|
db.create_session(RUN_1, source="api_server")
|
|
reads = []
|
|
real_get_session = db.get_session
|
|
|
|
def counted(session_id):
|
|
reads.append(session_id)
|
|
return real_get_session(session_id)
|
|
|
|
db.get_session = counted
|
|
try:
|
|
scope = declared_conversation_scope(_agent(RUN_1, db, CHAT_KEY))
|
|
finally:
|
|
del db.get_session
|
|
|
|
assert scope.startswith("gwk_")
|
|
assert reads == [RUN_1]
|
|
|
|
def test_a_db_without_the_combined_view_keeps_the_two_call_path(self):
|
|
calls = []
|
|
|
|
class LegacyDB:
|
|
def is_explicit_fork_child(self, sid):
|
|
calls.append(("fork", sid))
|
|
return False
|
|
|
|
def get_session(self, sid):
|
|
calls.append(("row", sid))
|
|
return {"source": "telegram"}
|
|
|
|
def latest_conversation_boundary(self, key, source):
|
|
return None
|
|
|
|
scope = declared_conversation_scope(_agent("sess-legacy", LegacyDB(), CHAT_KEY))
|
|
|
|
assert scope is not None and scope.startswith("gwk_")
|
|
assert calls == [("fork", "sess-legacy"), ("row", "sess-legacy")]
|
|
|
|
def test_the_combined_read_still_fails_closed(self):
|
|
"""A fork must never merge onto its parent's key on a DB failure."""
|
|
|
|
class BoomDB:
|
|
def declared_scope_identity(self, sid):
|
|
raise RuntimeError("db exploded")
|
|
|
|
def get_compression_lineage(self, sid):
|
|
return [sid]
|
|
|
|
agent = _agent("maybe-fork", BoomDB(), CHAT_KEY)
|
|
|
|
assert declared_conversation_scope(agent) is None
|
|
assert resolve_prompt_cache_scope(agent) == "maybe-fork"
|
|
|
|
def test_the_combined_read_still_refuses_a_fork(self, db):
|
|
db.create_session("tool-child", source="tool")
|
|
|
|
assert declared_conversation_scope(_agent("tool-child", db, CHAT_KEY)) is None
|
|
|
|
|
|
class TestPromptCacheKeyStability:
|
|
"""The reported symptom, at the wire layer: one conversation, one key."""
|
|
|
|
INSTRUCTIONS = "You are Reviewer in room7."
|
|
TOOLS = [{"type": "function", "name": "terminal"}]
|
|
|
|
def _key_for(self, agent):
|
|
scope = _cache_scope_from_session_id(resolve_prompt_cache_scope(agent))
|
|
return _content_cache_key(self.INSTRUCTIONS, self.TOOLS, scope)
|
|
|
|
def test_key_survives_a_per_response_id(self, db):
|
|
db.create_session(RUN_1, source="api_server")
|
|
db.create_session(RUN_2, source="api_server")
|
|
|
|
assert self._key_for(_agent(RUN_1, db, CHAT_KEY)) == self._key_for(
|
|
_agent(RUN_2, db, CHAT_KEY)
|
|
)
|
|
|
|
def test_key_still_churns_without_a_declaration(self, db):
|
|
"""Nothing is inferred from the id itself — the #79017 rule holds."""
|
|
db.create_session(RUN_1, source="api_server")
|
|
db.create_session(RUN_2, source="api_server")
|
|
|
|
assert self._key_for(_agent(RUN_1, db)) != self._key_for(_agent(RUN_2, db))
|
|
|
|
def test_codex_transport_key_matches_across_responses(self, db):
|
|
from agent.transports.codex import ResponsesApiTransport
|
|
|
|
db.create_session(RUN_1, source="api_server")
|
|
db.create_session(RUN_2, source="api_server")
|
|
transport = ResponsesApiTransport()
|
|
base = dict(
|
|
model="gpt-5.5",
|
|
messages=[
|
|
{"role": "system", "content": self.INSTRUCTIONS},
|
|
{"role": "user", "content": "hi"},
|
|
],
|
|
tools=[],
|
|
)
|
|
|
|
def key(session_id):
|
|
scope = resolve_prompt_cache_scope(_agent(session_id, db, CHAT_KEY))
|
|
return transport.build_kwargs(
|
|
**base, session_id=session_id, cache_scope_id=scope
|
|
)["prompt_cache_key"]
|
|
|
|
assert key(RUN_1) == key(RUN_2)
|
|
|
|
def test_chat_completions_key_matches_across_responses(self, db):
|
|
from agent.transports.chat_completions import _add_prompt_cache_key
|
|
|
|
db.create_session(RUN_1, source="api_server")
|
|
db.create_session(RUN_2, source="api_server")
|
|
messages = [{"role": "system", "content": self.INSTRUCTIONS}]
|
|
|
|
def key(session_id):
|
|
kwargs: dict = {}
|
|
_add_prompt_cache_key(
|
|
kwargs,
|
|
messages=messages,
|
|
tools=None,
|
|
supports_prompt_cache_key=True,
|
|
session_id=session_id,
|
|
cache_scope_id=resolve_prompt_cache_scope(
|
|
_agent(session_id, db, CHAT_KEY)
|
|
),
|
|
)
|
|
return kwargs["prompt_cache_key"]
|
|
|
|
assert key(RUN_1) == key(RUN_2)
|
|
|
|
def test_transcript_identity_is_not_rewritten(self, db):
|
|
"""#57012: the session header still carries the physical id."""
|
|
from agent.transports.codex import ResponsesApiTransport
|
|
|
|
db.create_session(RUN_1, source="api_server")
|
|
kwargs = ResponsesApiTransport().build_kwargs(
|
|
model="gpt-5.5",
|
|
messages=[{"role": "system", "content": self.INSTRUCTIONS}],
|
|
tools=[],
|
|
session_id=RUN_1,
|
|
cache_scope_id=resolve_prompt_cache_scope(_agent(RUN_1, db, CHAT_KEY)),
|
|
is_codex_backend=True,
|
|
)
|
|
assert kwargs["extra_headers"]["session_id"] == RUN_1
|
|
|
|
|
|
class TestProviderStickyKeys:
|
|
"""OpenRouter / Nous sticky ids and x-grok-conv-id read the same scope."""
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _clean_context(self):
|
|
affinity = set_affinity_scope(None)
|
|
conversation = set_conversation_context(None)
|
|
try:
|
|
yield
|
|
finally:
|
|
reset_conversation_context(conversation)
|
|
reset_affinity_scope(affinity)
|
|
|
|
def test_declared_scope_pins_the_sticky_key(self):
|
|
scope = declared_conversation_scope(_agent(RUN_1, None, CHAT_KEY))
|
|
token = set_affinity_scope(scope)
|
|
try:
|
|
first = _sticky_key(RUN_1)
|
|
second = _sticky_key(RUN_2)
|
|
finally:
|
|
reset_affinity_scope(token)
|
|
|
|
assert first == second == scope
|
|
|
|
def test_without_a_declaration_the_conversation_id_still_wins(self):
|
|
"""Delegate trees keep sharing their parent's sticky key."""
|
|
conversation = set_conversation_context("parent-root")
|
|
try:
|
|
assert get_affinity_scope() is None
|
|
assert _sticky_key("delegate-child") == "parent-root"
|
|
finally:
|
|
reset_conversation_context(conversation)
|
|
|
|
def test_grok_conv_id_follows_the_declared_scope(self):
|
|
scope = declared_conversation_scope(_agent(RUN_1, None, CHAT_KEY))
|
|
token = set_affinity_scope(scope)
|
|
try:
|
|
headers = _grok_headers(RUN_1)
|
|
headers_next = _grok_headers(RUN_2)
|
|
finally:
|
|
reset_affinity_scope(token)
|
|
|
|
assert headers["x-grok-conv-id"] == headers_next["x-grok-conv-id"] == scope
|
|
|
|
def test_nous_sticky_key_follows_the_declared_scope(self):
|
|
from providers import get_provider_profile
|
|
|
|
scope = declared_conversation_scope(_agent(RUN_1, None, CHAT_KEY))
|
|
token = set_affinity_scope(scope)
|
|
try:
|
|
body = get_provider_profile("nous").build_extra_body(session_id=RUN_1)
|
|
body_next = get_provider_profile("nous").build_extra_body(session_id=RUN_2)
|
|
finally:
|
|
reset_affinity_scope(token)
|
|
|
|
assert body["session_id"] == body_next["session_id"] == scope
|
|
|
|
|
|
class TestConversationGenerationRotates:
|
|
"""The declared key must not outlive the conversation it names.
|
|
|
|
``gateway_session_key`` is a per-CHAT identifier: ``reset_session()``
|
|
mints a fresh physical id on ``/new`` and keeps the key, and the
|
|
idle/daily/suspended policy resets do the same. Hashing the key alone
|
|
would map the conversation before a reset and the one after it onto ONE
|
|
affinity scope, violating the #79017/#86733 contract.
|
|
|
|
The generation is read from the boundary those resets already write
|
|
(``_RESET_END_REASONS`` on the outgoing row), so nothing new is persisted
|
|
and the two fences cannot drift.
|
|
"""
|
|
|
|
KEY = "agent:main:telegram:dm:123"
|
|
|
|
def _keyed(self, db, session_id):
|
|
db.create_session(
|
|
session_id=session_id,
|
|
source="telegram",
|
|
session_key=self.KEY,
|
|
)
|
|
return _agent(session_id, db, self.KEY)
|
|
|
|
def test_new_rotates_the_declared_scope(self, db):
|
|
"""The exact reproduction that blocked this PR, now green.
|
|
|
|
``/new`` ends the outgoing row with ``session_reset`` and mints a new
|
|
physical id under the same chat key; before the generation qualifier
|
|
both sides hashed to one ``gwk_`` value.
|
|
"""
|
|
before = self._keyed(db, "sess-A")
|
|
scope_before = resolve_prompt_cache_scope(before)
|
|
|
|
db.end_session("sess-A", "session_reset")
|
|
after = self._keyed(db, "sess-B")
|
|
|
|
assert scope_before.startswith("gwk_")
|
|
assert resolve_prompt_cache_scope(after).startswith("gwk_")
|
|
assert resolve_prompt_cache_scope(after) != scope_before
|
|
|
|
@pytest.mark.parametrize(
|
|
"reason",
|
|
["session_reset", "session_switch", "idle", "daily", "suspended",
|
|
"resume_pending_expired"],
|
|
)
|
|
def test_every_reset_boundary_rotates(self, db, reason):
|
|
"""Policy auto-resets are conversation replacements too.
|
|
|
|
A hand-rolled counter incremented only in ``reset_session()`` would
|
|
leave these on the previous generation; reading the durable boundary
|
|
covers the whole set by construction.
|
|
"""
|
|
first = self._keyed(db, f"sess-{reason}-1")
|
|
scope_first = resolve_prompt_cache_scope(first)
|
|
db.end_session(f"sess-{reason}-1", reason)
|
|
second = self._keyed(db, f"sess-{reason}-2")
|
|
assert resolve_prompt_cache_scope(second) != scope_first
|
|
|
|
def test_generations_never_roll_back(self, db):
|
|
"""Three conversations on one key produce three distinct scopes."""
|
|
scopes = []
|
|
for i in range(3):
|
|
agent = self._keyed(db, f"sess-gen{i}")
|
|
scopes.append(resolve_prompt_cache_scope(agent))
|
|
db.end_session(f"sess-gen{i}", "session_reset")
|
|
assert len(set(scopes)) == 3
|
|
|
|
def test_per_response_ids_still_share_one_scope(self, db):
|
|
"""The whole point of the PR survives the fix.
|
|
|
|
A host that mints one id per RESPONSE writes no boundary, so every
|
|
reply reads the same (empty) generation and lands on one scope.
|
|
"""
|
|
scopes = {
|
|
resolve_prompt_cache_scope(self._keyed(db, f"gc_run_{i}"))
|
|
for i in range(4)
|
|
}
|
|
assert len(scopes) == 1
|
|
assert next(iter(scopes)).startswith("gwk_")
|
|
|
|
def test_an_accidental_end_is_not_a_boundary(self, db):
|
|
"""Only intentional breaks rotate; a crash-close keeps the scope warm."""
|
|
first = self._keyed(db, "sess-live")
|
|
scope_first = resolve_prompt_cache_scope(first)
|
|
db.end_session("sess-live", "agent_close")
|
|
second = self._keyed(db, "sess-resumed")
|
|
assert resolve_prompt_cache_scope(second) == scope_first
|
|
|
|
def test_another_chats_reset_does_not_rotate_this_one(self, db):
|
|
"""The boundary is read per declared key, never globally."""
|
|
mine = self._keyed(db, "sess-mine")
|
|
scope_mine = resolve_prompt_cache_scope(mine)
|
|
|
|
other_key = "agent:main:telegram:dm:999"
|
|
db.create_session(
|
|
session_id="sess-other", source="telegram", session_key=other_key
|
|
)
|
|
db.end_session("sess-other", "session_reset")
|
|
|
|
again = self._keyed(db, "sess-mine-2")
|
|
assert resolve_prompt_cache_scope(again) == scope_mine
|
|
|
|
def test_scope_never_carries_the_raw_key_or_boundary(self, db):
|
|
agent = self._keyed(db, "sess-A")
|
|
db.end_session("sess-A", "session_reset")
|
|
rotated = self._keyed(db, "sess-B")
|
|
scope = resolve_prompt_cache_scope(rotated)
|
|
assert self.KEY not in scope
|
|
assert "telegram" not in scope
|
|
assert scope.startswith("gwk_")
|
|
assert len(scope) == len("gwk_") + 24
|
|
|
|
def test_generation_read_failure_degrades_to_the_physical_scope(self):
|
|
"""Fail closed: an unqualified key would span a /new."""
|
|
|
|
class BoomDB:
|
|
def is_explicit_fork_child(self, sid):
|
|
return False
|
|
|
|
def latest_conversation_boundary(self, key):
|
|
raise RuntimeError("db down")
|
|
|
|
def get_compression_lineage(self, sid):
|
|
return []
|
|
|
|
agent = _agent("sess-A", BoomDB(), self.KEY)
|
|
assert declared_conversation_scope(agent) is None
|
|
|
|
def test_a_db_without_the_lookup_keeps_the_declaration(self, db):
|
|
"""Forward/backward compatible: no boundary API means no boundary."""
|
|
|
|
class LegacyDB:
|
|
def is_explicit_fork_child(self, sid):
|
|
return False
|
|
|
|
def get_compression_lineage(self, sid):
|
|
return []
|
|
|
|
agent = _agent("sess-A", LegacyDB(), self.KEY)
|
|
scope = declared_conversation_scope(agent)
|
|
assert scope is not None and scope.startswith("gwk_")
|
|
|
|
def test_a_backwards_clock_does_not_reuse_a_generation(self, db):
|
|
"""An NTP correction between two resets must not merge them.
|
|
|
|
``MAX(ended_at)`` alone would keep returning the earlier, larger
|
|
timestamp; the boundary COUNT is what separates them.
|
|
"""
|
|
first = self._keyed(db, "sess-clock-1")
|
|
scope_first = resolve_prompt_cache_scope(first)
|
|
db.end_session("sess-clock-1", "session_reset")
|
|
|
|
second = self._keyed(db, "sess-clock-2")
|
|
scope_second = resolve_prompt_cache_scope(second)
|
|
db.end_session("sess-clock-2", "session_reset")
|
|
# The clock went backwards: this boundary lands BEFORE the first one.
|
|
with db._lock:
|
|
db._conn.execute(
|
|
"UPDATE sessions SET ended_at = ("
|
|
" SELECT MIN(ended_at) FROM sessions WHERE ended_at IS NOT NULL"
|
|
") - 60 WHERE id = ?",
|
|
("sess-clock-2",),
|
|
)
|
|
db._conn.commit()
|
|
|
|
third = self._keyed(db, "sess-clock-3")
|
|
scope_third = resolve_prompt_cache_scope(third)
|
|
assert len({scope_first, scope_second, scope_third}) == 3
|
|
|
|
|
|
class TestPeerIdentityIsSourceQualified:
|
|
"""The generation and the carrier use the same identity tuple as recovery.
|
|
|
|
``X-Hermes-Session-Key`` accepts any authenticated caller-supplied string,
|
|
so an API conversation may legally carry the same key as a Telegram row in
|
|
one database. Keying on the string alone let a ``/new`` on that unrelated
|
|
row rotate this conversation's affinity identity, while
|
|
``find_latest_gateway_session_for_peer`` correctly refused to cross the
|
|
same line — the physical identity stayed put while the affinity identity
|
|
moved under it (@andrexibiza on #98811).
|
|
"""
|
|
|
|
KEY = "shared-key-string"
|
|
|
|
def _row(self, db, sid, source):
|
|
db.create_session(session_id=sid, source=source, session_key=self.KEY)
|
|
return SimpleNamespace(
|
|
session_id=sid, _session_db=db, _gateway_session_key=self.KEY,
|
|
platform=source,
|
|
)
|
|
|
|
def test_a_foreign_sources_reset_does_not_rotate_this_conversation(self, db):
|
|
mine = self._row(db, "api-1", "api_server")
|
|
before = resolve_prompt_cache_scope(mine)
|
|
|
|
# Same key string, different platform, reset only over there.
|
|
self._row(db, "tg-1", "telegram")
|
|
db.end_session("tg-1", "session_reset")
|
|
|
|
assert resolve_prompt_cache_scope(self._row(db, "api-2", "api_server")) == before
|
|
|
|
def test_our_own_reset_still_rotates(self, db):
|
|
mine = self._row(db, "api-1", "api_server")
|
|
before = resolve_prompt_cache_scope(mine)
|
|
db.end_session("api-1", "session_reset")
|
|
assert resolve_prompt_cache_scope(self._row(db, "api-2", "api_server")) != before
|
|
|
|
def test_equal_keys_under_different_sources_never_share_a_scope(self, db):
|
|
mine = resolve_prompt_cache_scope(self._row(db, "api-1", "api_server"))
|
|
theirs = resolve_prompt_cache_scope(self._row(db, "tg-1", "telegram"))
|
|
assert mine != theirs
|
|
assert mine.startswith("gwk_") and theirs.startswith("gwk_")
|
|
|
|
def test_the_boundary_read_is_peer_scoped(self, db):
|
|
db.create_session(session_id="tg-1", source="telegram", session_key=self.KEY)
|
|
db.end_session("tg-1", "session_reset")
|
|
assert db.latest_conversation_boundary(self.KEY, "telegram") is not None
|
|
assert db.latest_conversation_boundary(self.KEY, "api_server") is None
|
|
assert db.latest_conversation_boundary(self.KEY, "") is None
|
|
|
|
|
|
class TestGenerationSurvivesPruning:
|
|
"""A generation derived from prunable rows cannot prove non-reuse.
|
|
|
|
`delete_session()` orphans surviving children and deletes the selected
|
|
row, and bulk prune selects ended rows, so an aggregate over
|
|
`_RESET_END_REASONS` boundaries can return a pair it already emitted:
|
|
`(1, T1) -> (2, T2) -> delete boundary B -> (1, T1)`, handing a new
|
|
conversation a retired affinity identity (@andrexibiza on #98811).
|
|
|
|
The counter therefore lives in `conversation_generations`, outside session
|
|
history, and only ever increments.
|
|
"""
|
|
|
|
KEY = "agent:main:telegram:dm:777"
|
|
SOURCE = "telegram"
|
|
|
|
def _keyed(self, db, sid):
|
|
db.create_session(session_id=sid, source=self.SOURCE, session_key=self.KEY)
|
|
return SimpleNamespace(
|
|
session_id=sid, _session_db=db, _gateway_session_key=self.KEY,
|
|
platform=self.SOURCE,
|
|
)
|
|
|
|
def _gen(self, db):
|
|
return db.latest_conversation_boundary(self.KEY, self.SOURCE)
|
|
|
|
def test_deleting_the_newest_boundary_does_not_roll_back(self, db):
|
|
a = self._keyed(db, "s-a")
|
|
scope_a = resolve_prompt_cache_scope(a)
|
|
db.end_session("s-a", "session_reset")
|
|
|
|
b = self._keyed(db, "s-b")
|
|
scope_b = resolve_prompt_cache_scope(b)
|
|
db.end_session("s-b", "session_reset")
|
|
assert self._gen(db) == 2
|
|
|
|
db.delete_session("s-b") # prune the newest boundary
|
|
assert self._gen(db) == 2 # counter is outside that rowset
|
|
|
|
c = self._keyed(db, "s-c")
|
|
scope_c = resolve_prompt_cache_scope(c)
|
|
assert len({scope_a, scope_b, scope_c}) == 3
|
|
|
|
def test_deleting_every_boundary_does_not_roll_back(self, db):
|
|
a = self._keyed(db, "s-a")
|
|
scope_a = resolve_prompt_cache_scope(a)
|
|
db.end_session("s-a", "session_reset")
|
|
db.delete_session("s-a")
|
|
|
|
b = self._keyed(db, "s-b")
|
|
assert resolve_prompt_cache_scope(b) != scope_a
|
|
|
|
def test_a_backwards_clock_then_a_prune_still_cannot_repeat(self, db):
|
|
"""The reviewer's exact shape: (1,T1) -> (2,T1) -> delete -> (1,T1)."""
|
|
a = self._keyed(db, "s-a")
|
|
scope_a = resolve_prompt_cache_scope(a)
|
|
db.end_session("s-a", "session_reset")
|
|
|
|
b = self._keyed(db, "s-b")
|
|
scope_b = resolve_prompt_cache_scope(b)
|
|
db.end_session("s-b", "session_reset")
|
|
# Clock went backwards: this boundary lands before the first one.
|
|
with db._lock:
|
|
db._conn.execute(
|
|
"UPDATE sessions SET ended_at = ("
|
|
" SELECT MIN(ended_at) FROM sessions WHERE ended_at IS NOT NULL"
|
|
") - 60 WHERE id = ?",
|
|
("s-b",),
|
|
)
|
|
db._conn.commit()
|
|
db.delete_session("s-b")
|
|
|
|
c = self._keyed(db, "s-c")
|
|
assert len({scope_a, scope_b, resolve_prompt_cache_scope(c)}) == 3
|
|
|
|
def test_only_real_boundaries_advance_it(self, db):
|
|
self._keyed(db, "s-a")
|
|
db.end_session("s-a", "compression")
|
|
assert self._gen(db) is None
|
|
db.create_session(session_id="s-b", source=self.SOURCE, session_key=self.KEY)
|
|
db.end_session("s-b", "agent_close")
|
|
assert self._gen(db) is None
|
|
|
|
def test_a_repeated_end_does_not_double_count(self, db):
|
|
"""end_session no-ops on an ended row; the first reason wins."""
|
|
self._keyed(db, "s-a")
|
|
db.end_session("s-a", "session_reset")
|
|
db.end_session("s-a", "session_reset")
|
|
db.end_session("s-a", "idle")
|
|
assert self._gen(db) == 1
|
|
|
|
def test_promote_advances_it_too(self, db):
|
|
"""/new and the policy resets promote rather than end_session."""
|
|
self._keyed(db, "s-a")
|
|
db.end_session("s-a", "agent_close")
|
|
assert self._gen(db) is None
|
|
assert db.promote_to_session_reset("s-a", "session_reset") is True
|
|
assert self._gen(db) == 1
|
|
|
|
def test_an_unkeyed_row_advances_nothing(self, db):
|
|
db.create_session(session_id="s-bare", source=self.SOURCE)
|
|
db.end_session("s-bare", "session_reset")
|
|
assert self._gen(db) is None
|
|
|
|
def test_the_counter_is_peer_scoped(self, db):
|
|
self._keyed(db, "s-a")
|
|
db.end_session("s-a", "session_reset")
|
|
assert self._gen(db) == 1
|
|
assert db.latest_conversation_boundary(self.KEY, "api_server") is None
|
|
|
|
|
|
class TestSourceOverrideDomain:
|
|
"""The scope is memoized immediately, so the source must be right first.
|
|
|
|
``_agent_source`` used ``agent.platform`` before the row landed while
|
|
persistence uses ``_session_source_for_agent``, which honors
|
|
``HERMES_SESSION_SOURCE``. Under an override both sides of a ``/new``
|
|
queried the platform domain, missed the boundary stored under the
|
|
override, and hashed the same scope (@andrexibiza on #98811).
|
|
"""
|
|
|
|
KEY = "agent:main:telegram:dm:888"
|
|
|
|
def test_the_pre_row_source_matches_persistence(self, db, monkeypatch):
|
|
from agent.prompt_cache_scope import _agent_source
|
|
|
|
monkeypatch.setenv("HERMES_SESSION_SOURCE", "override-src")
|
|
agent = SimpleNamespace(
|
|
session_id="s-none", _session_db=db, _gateway_session_key=self.KEY,
|
|
platform="telegram",
|
|
)
|
|
assert _agent_source(agent, "", db) == "override-src"
|
|
|
|
def test_new_rotates_under_a_source_override(self, db, monkeypatch):
|
|
monkeypatch.setenv("HERMES_SESSION_SOURCE", "override-src")
|
|
|
|
def keyed(sid):
|
|
db.create_session(
|
|
session_id=sid, source="override-src", session_key=self.KEY
|
|
)
|
|
return SimpleNamespace(
|
|
session_id=sid, _session_db=db,
|
|
_gateway_session_key=self.KEY, platform="telegram",
|
|
)
|
|
|
|
before = resolve_prompt_cache_scope(keyed("s-a"))
|
|
db.end_session("s-a", "session_reset")
|
|
assert resolve_prompt_cache_scope(keyed("s-b")) != before
|