agent/prompt_cache_scope.py: durable per-(source,session_key) monotonic generation; list _RESET_END_REASONS; note never garbage-collected. tests/agent/test_studio_bridge_affinity.py: build AIAgent before db.create_session; docstring corrected.
This commit is contained in:
joaomarcos
+14
-31
@@ -145,40 +145,23 @@ def _agent_source(
|
||||
|
||||
|
||||
def _conversation_generation(session_key: str, source: str, session_db: Any) -> str:
|
||||
"""Return the generation marker for *session_key*'s current conversation.
|
||||
"""Return the durable generation for *session_key*'s current conversation.
|
||||
|
||||
The declared key is a per-CHAT identifier and deliberately outlives any
|
||||
single conversation on it — ``reset_session()`` mints a fresh physical id
|
||||
on ``/new`` but keeps the key, and the idle/daily/suspended policy resets
|
||||
do the same. Hashing the key alone would therefore map the conversation
|
||||
before a reset and the one after it onto ONE affinity scope, violating the
|
||||
#79017/#86733 contract (warm across compression rotation, cold across a
|
||||
new conversation).
|
||||
The declared key names a chat and deliberately survives `/new` and policy
|
||||
resets. Hashing it alone would therefore reuse one affinity scope across
|
||||
distinct conversations, violating the #79017/#86733 contract: warm across
|
||||
compression, cold across a conversation boundary.
|
||||
|
||||
The generation that must rotate is already durable: every one of those
|
||||
boundaries closes the outgoing row with an ``_RESET_END_REASONS``
|
||||
end_reason, and ``SessionDB.latest_conversation_boundary`` reports the most
|
||||
recent one. Qualifying the key with it gives a carrier that is
|
||||
``SessionDB.latest_conversation_boundary`` reads the monotonic
|
||||
``conversation_generations`` counter for ``(source, session_key)``. The
|
||||
counter advances in the same transaction that records an
|
||||
``_RESET_END_REASONS`` boundary. It is independent of prunable session rows
|
||||
and wall-clock time, so deletion, bulk pruning, and clock rollback cannot
|
||||
reissue an old generation. Compression continues the current conversation
|
||||
and does not advance it.
|
||||
|
||||
- stable across a host's per-response physical ids (no boundary is written
|
||||
when nothing was reset, so every reply hashes the same value), and
|
||||
- rotating on every conversation replacement, ``/new`` and the policy
|
||||
auto-resets alike.
|
||||
|
||||
The marker is a durable counter kept outside prunable session history
|
||||
(``conversation_generations``, advanced in the same transaction that writes
|
||||
each boundary). Deriving it from the session rows instead — an aggregate
|
||||
over ``_RESET_END_REASONS`` boundaries — cannot prove non-reuse: deleting or
|
||||
pruning an ended row makes the aggregate return a pair it already emitted,
|
||||
handing a new conversation a retired affinity identity. It is also
|
||||
wall-clock-free, so a backwards NTP correction cannot reorder it.
|
||||
|
||||
No counter is introduced anywhere: the marker is read from state the
|
||||
reset paths already write, and it is read on the memoized resolution path,
|
||||
not per API call.
|
||||
|
||||
Returns ``""`` when the key has never been reset, when the DB does not
|
||||
expose the lookup, or when it reports nothing.
|
||||
This lookup runs on the memoized resolution path, not once per API call.
|
||||
Return ``""`` when the key has never reset or the DB exposes no generation.
|
||||
"""
|
||||
reader = getattr(session_db, "latest_conversation_boundary", None)
|
||||
if not callable(reader):
|
||||
|
||||
@@ -19,9 +19,8 @@ The shape reproduced here:
|
||||
1. ``groupRuntimeSessionId(room, profile, name)`` mints
|
||||
``gc_run_<room>_<profile>_<name>`` truncated to 96 characters plus a fresh
|
||||
UUID4 hex — a NEW physical id per reply;
|
||||
2. the bridge writes the row itself before the agent runs, so the row already
|
||||
exists when the scope first resolves, and the row's ``source`` — not
|
||||
``agent.platform`` — is the identity the peer queries match on;
|
||||
2. the agent is constructed before the row exists (same as Studio's pool);
|
||||
the bridge then writes the row, so the resolution reads a landed row;
|
||||
3. ``AIAgent(...)`` is constructed with ``platform`` / ``session_id`` /
|
||||
``session_db`` and no routing identity of any kind.
|
||||
|
||||
@@ -112,15 +111,24 @@ def _reply_scope(
|
||||
name: str = "Reviewer",
|
||||
row_source: str = BRIDGE_ROW_SOURCE,
|
||||
) -> str:
|
||||
"""One Studio reply, in the bridge's own order, returning its scope."""
|
||||
"""One Studio reply, in the bridge's own order, returning its scope.
|
||||
|
||||
Mirrors the real Studio construction order: ``AIAgent(...)`` is built
|
||||
first with a fresh physical id, then the bridge persists the row, then
|
||||
the scope is resolved against the now-landed row. Constructing the
|
||||
agent before persistence is the part the test exists to pin — the
|
||||
resolution path must read the row at resolve time, not assume it was
|
||||
already there when the agent was instantiated.
|
||||
"""
|
||||
session_id = _runtime_session_id(room=room, name=name)
|
||||
# The bridge writes the row before the agent runs.
|
||||
# AIAgent first — same order as the Studio bridge pool.
|
||||
agent = _bridge_agent(session_db, session_id, declared_key=declared_key)
|
||||
# Then the bridge writes the row, which is when the scope's row reads
|
||||
# finally have something to find.
|
||||
session_db.create_session(
|
||||
session_id=session_id, source=row_source, model="test/model"
|
||||
)
|
||||
return resolve_prompt_cache_scope(
|
||||
_bridge_agent(session_db, session_id, declared_key=declared_key)
|
||||
)
|
||||
return resolve_prompt_cache_scope(agent)
|
||||
|
||||
|
||||
class TestStudioBridgeAffinity:
|
||||
@@ -203,23 +211,21 @@ class TestStudioBridgeAffinity:
|
||||
conversation = _reply_scope(db, declared_key=key)
|
||||
|
||||
child_id = _runtime_session_id()
|
||||
child = _bridge_agent(db, child_id, declared_key=key)
|
||||
db.create_session(session_id=child_id, source="tool", model="test/model")
|
||||
child = resolve_prompt_cache_scope(
|
||||
_bridge_agent(db, child_id, declared_key=key)
|
||||
)
|
||||
|
||||
assert child == child_id
|
||||
assert child != conversation
|
||||
assert resolve_prompt_cache_scope(child) == child_id
|
||||
assert child_id != conversation
|
||||
|
||||
def test_a_background_review_fork_never_borrows_the_declaration(self, db):
|
||||
"""Background review clones the live runtime, declared key included."""
|
||||
key = _bridge_session_key()
|
||||
session_id = _runtime_session_id()
|
||||
agent = _bridge_agent(db, session_id, declared_key=key)
|
||||
agent._persist_disabled = True
|
||||
db.create_session(
|
||||
session_id=session_id, source=BRIDGE_ROW_SOURCE, model="test/model"
|
||||
)
|
||||
agent = _bridge_agent(db, session_id, declared_key=key)
|
||||
agent._persist_disabled = True
|
||||
|
||||
assert resolve_prompt_cache_scope(agent) == session_id
|
||||
|
||||
@@ -229,12 +235,11 @@ class TestUndeclaredBridgeIsUnchanged:
|
||||
|
||||
def test_the_undeclared_shape_resolves_to_the_physical_id(self, db):
|
||||
session_id = _runtime_session_id()
|
||||
agent = _bridge_agent(db, session_id, declared_key=None)
|
||||
db.create_session(
|
||||
session_id=session_id, source=BRIDGE_ROW_SOURCE, model="test/model"
|
||||
)
|
||||
|
||||
agent = _bridge_agent(db, session_id, declared_key=None)
|
||||
|
||||
assert resolve_prompt_cache_scope(agent) == session_id
|
||||
|
||||
def test_an_undeclared_compression_rotation_still_walks_its_lineage(self, db):
|
||||
@@ -248,7 +253,6 @@ class TestUndeclaredBridgeIsUnchanged:
|
||||
model="test/model",
|
||||
parent_session_id=root,
|
||||
)
|
||||
|
||||
agent = _bridge_agent(db, rotated, declared_key=None)
|
||||
|
||||
assert resolve_prompt_cache_scope(agent) == root
|
||||
|
||||
Reference in New Issue
Block a user