diff --git a/gateway/session_execution.py b/gateway/session_execution.py index e0522b9c42..3277a93230 100644 --- a/gateway/session_execution.py +++ b/gateway/session_execution.py @@ -154,14 +154,17 @@ class GatewaySessionAgentMixin: ctx = self._ctx runner = self._runner src = ctx.source + from gateway.session_authorities import active_authority from gateway.session_policy import policy_for_source + from gateway.session_worker_construct import session_prefill policy = policy_for_source(runner, src) return ctx.AIAgent( model=turn_route["model"], **turn_route["runtime"], **_checkpoint_agent_kwargs(ctx.user_config), max_iterations=max_iterations, quiet_mode=not bool(policy and policy.verbose), verbose_logging=False, enabled_toolsets=ctx.enabled_toolsets, disabled_toolsets=ctx.disabled_toolsets, ephemeral_system_prompt=combined_ephemeral or None, - prefill_messages=runner._prefill_messages or None, + prefill_messages=(session_prefill(active_authority(runner), policy, src.chat_id) if policy + else runner._prefill_messages) or None, reasoning_config=reasoning_config, service_tier=runner._service_tier, request_overrides=turn_route.get("request_overrides"), providers_allowed=pr.get("only"), providers_ignored=pr.get("ignore"), providers_order=pr.get("order"), diff --git a/gateway/session_local.py b/gateway/session_local.py index e8c314b11b..5e57cdcd09 100644 --- a/gateway/session_local.py +++ b/gateway/session_local.py @@ -269,8 +269,11 @@ def stage_local_session(authority, actor, params, *, trusted_policy=None, truste # naive cutoff, and one aware entry aborts the whole iteration. now = _now() entry = SessionEntry(route, sid, now, now, origin=source, platform=Platform.LOCAL) - return {'profile_id': authority.profile_id, 'principal_id': actor.subject, 'request_id': request_id, - 'session_id': sid, 'route': route, 'entry': entry.to_dict(), 'policy': asdict(policy)} + # Prefill is read now, at creation (main's CLI: at startup), and held only if this attempt commits. + from gateway.session_worker_construct import stage_session_prefill + return stage_session_prefill(authority, policy, { + 'profile_id': authority.profile_id, 'principal_id': actor.subject, 'request_id': request_id, + 'session_id': sid, 'route': route, 'entry': entry.to_dict(), 'policy': asdict(policy)}) def publish_local_policy(authority, session_id): diff --git a/gateway/session_mutations.py b/gateway/session_mutations.py index eb9d324524..e81db1797c 100644 --- a/gateway/session_mutations.py +++ b/gateway/session_mutations.py @@ -313,3 +313,5 @@ def retire_live_sessions(authority, session_ids): evict(candidate.route) from gateway.session_local_mcp import release_editor_mcp release_editor_mcp(authority, session_ids) + from gateway.session_worker_construct import release_session_prefill + release_session_prefill(authority, session_ids) diff --git a/gateway/session_worker_construct.py b/gateway/session_worker_construct.py index 14c51d58be..3e863a2a00 100644 --- a/gateway/session_worker_construct.py +++ b/gateway/session_worker_construct.py @@ -6,6 +6,7 @@ worker must never reload the profile), so the owner resolves them per turn, mode and they ride the hydrated request beside ``turn_v1``. The child validates the closed shape. """ import json +from pathlib import Path KEYS = frozenset({'fallback_model', 'ephemeral_system_prompt', 'prefill_messages'}) @@ -16,7 +17,7 @@ def construct_inputs(authority, policy, chat_id): with owner_scope(authority): return {'fallback_model': _fallback_chain(authority, policy), 'ephemeral_system_prompt': _ephemeral_prompt(authority, policy, chat_id), - 'prefill_messages': _prefill(authority, policy)} + 'prefill_messages': session_prefill(authority, policy, chat_id)} def _fallback_chain(authority, policy): @@ -45,14 +46,6 @@ def _ephemeral_prompt(authority, policy, chat_id): return '\n\n'.join(p for p in (configured, policy.skills_prompt) if p) or None -def _prefill(authority, policy): - """The runner's prefill examples, exactly as the in-process constructor passes them. A bypass - session never loads the profile's prefill file (and the safe policy drops prefill anyway).""" - if policy.ignore_user_config: - return None - return getattr(authority.runner, '_prefill_messages', None) or None - - def session_routing(authority, policy): """The session's ``provider_routing``: its own profile's, as of its creation (main's CLI read it once at startup). Never the runner's daemon-start copy.""" @@ -60,6 +53,79 @@ def session_routing(authority, policy): return routing if isinstance(routing, dict) else {} +def session_prefill(authority, policy, session_id): + """The session's prefill examples, for the in-process turn and a managed worker alike: the + snapshot ``stage_session_prefill`` took when the session was created (main's CLI read the file + once at startup). A session this owner did not create (restored after a restart, a branch) + reads it on first use and keeps that. A bypass session never loads it.""" + if policy.ignore_user_config: + return None + held = _held_prefill(authority) + if session_id not in held: + held.setdefault(session_id, (_read_prefill(authority, policy),)) + return held[session_id][0] + + +class StagedReceipt(dict): + """A creation receipt plus owner memory ``commit_local_session`` publishes only from the + transaction that creates the session (never persisted: the receipt itself is the dict).""" + on_create: '_PrefillPublication | None' = None + + +class _PrefillPublication: + def __init__(self, authority, policy, session_id): + # A box per attempt: ownership is identity, even when two attempts read the same contents. + self.authority, self.policy, self.session_id, self.box = authority, policy, session_id, None + + def read(self): + """This attempt's snapshot, read by its commit before the transaction (off the owner loop).""" + self.box = (_read_prefill(self.authority, self.policy),) + + def publish(self): + _held_prefill(self.authority)[self.session_id] = self.box + + def withdraw(self): + held = _held_prefill(self.authority) + if held.get(self.session_id) is self.box: + held.pop(self.session_id, None) + + +def stage_session_prefill(authority, policy, receipt): + """Stage a new session's prefill snapshot: the ``prefill_messages_file`` of its config, relative + to its own profile home, read by this attempt's commit (``commit_local_session``, which the async + creates run in ``tracked_write``'s worker thread), never on the owner loop. The snapshot is held + in the authority's memory only if this attempt's commit creates the session: an overlapping + retry or conflicting create never replaces the winner's, and a refused or failed create leaves + nothing. Never evicted by other sessions; ``release_session_prefill`` drops it with the session.""" + staged = StagedReceipt(receipt) + if not policy.ignore_user_config: + staged.on_create = _PrefillPublication(authority, policy, receipt['session_id']) + return staged + + +def release_session_prefill(authority, session_ids): + held = getattr(authority, '_local_prefill', None) or {} + for sid in session_ids: + held.pop(str(sid), None) + + +def _held_prefill(authority): + held = getattr(authority, '_local_prefill', None) + if held is None: + held = authority._local_prefill = {} + return held + + +def _read_prefill(authority, policy): + from hermes_cli.cli_config_load import _load_prefill_messages, _resolve_prefill_messages_file + home = Path(str(authority.profile_id)) + if not home.is_absolute(): + from hermes_constants import get_hermes_home + home = get_hermes_home() + path = _resolve_prefill_messages_file(policy.config(authority, secrets=False)) + return _load_prefill_messages(path, home) or None + + def construct_kwargs(frame): """Child side: the validated ``AIAgent`` keyword arguments; a frame without the object (older owner) constructs exactly as before.""" diff --git a/hermes_state_local.py b/hermes_state_local.py index bee2b8bdbc..dbfc6b0712 100644 --- a/hermes_state_local.py +++ b/hermes_state_local.py @@ -10,7 +10,12 @@ API_DECLARED_PREFIX = 'gateway.api.conversation.v1.' def commit_local_session(db, *, epoch, receipt): - """Reserve the row, route and immutable policy together, before publication.""" + """Reserve the row, route and immutable policy together, before publication. A staged receipt's + ``on_create`` (owner memory, never stored) is published by the transaction that creates the row + and withdrawn if that transaction fails: an exact retry, a refused or a losing attempt adds none.""" + on_create = getattr(receipt, 'on_create', None) + if on_create is not None: + on_create.read() # its file read: in the caller's thread, outside the write transaction encoded = _json(receipt) receipt = json.loads(encoded) sid, route = receipt['session_id'], receipt['route'] @@ -53,8 +58,15 @@ def commit_local_session(db, *, epoch, receipt): conn.execute("INSERT INTO gateway_routing(scope,session_key,entry_json,updated_at) VALUES('',?,?,?)", (route, _json(entry), started)) conn.execute('INSERT INTO state_meta(key,value) VALUES(?,?)', (key, encoded)) + if on_create is not None: + on_create.publish() # before the row is visible, so no reader lazy-loads around it return receipt - return db._execute_write(write) + try: + return db._execute_write(write) + except BaseException: + if on_create is not None: + on_create.withdraw() # only this attempt's own snapshot, never another winner's + raise def client_request(request_json, derived=None): diff --git a/tests/gateway/test_managed_worker_construct.py b/tests/gateway/test_managed_worker_construct.py index 989a6c3fbb..894b1574dd 100644 --- a/tests/gateway/test_managed_worker_construct.py +++ b/tests/gateway/test_managed_worker_construct.py @@ -74,9 +74,10 @@ def _serve(cls): return server -def run_pair(tmp_path, extra_config, turn, **create_params): +def run_pair(tmp_path, extra_config, turn, *, before_create=None, **create_params): """Boot one ordinary daemon; create an in-process and a managed session from the same config; - run ``turn(ws, sid, name, settle, servers)`` for each and return its results by session.""" + run ``turn(ws, sid, name, settle, servers)`` for each and return its results by session. + ``before_create(home)`` edits the profile after the daemon started, before each create.""" root = Path(__file__).resolve().parents[2] home, user = tmp_path / 'state', tmp_path / 'user' home.mkdir(mode=0o700) @@ -112,6 +113,8 @@ def run_pair(tmp_path, extra_config, turn, **create_params): async def exercise(desc): async with websocket(home, desc) as ws: for name in ('inproc', 'managed'): + if before_create is not None: + before_create(home) if name == 'managed': edited = json.loads((home / 'config.yaml').read_text()) edited['gateway']['managed_workers'] = True @@ -174,8 +177,37 @@ def test_managed_turn_sends_configured_instruction_and_prefill(tmp_path): assert found == {'system_mark': [True], 'prefill': ['PREFILL_Q_MARK', 'PREFILL_A_MARK'], 'stored': False}, (name, results) +@pytest.mark.platforms("linux") +def test_new_sessions_send_the_profiles_current_prefill(tmp_path): + """CFG-7: a ``prefill_messages_file`` path and contents edited after the daemon started reach the + next new session, in process and managed, as a new ``hermes chat`` process read them on main.""" + def pair(home, name, mark): + (home / name).write_text(json.dumps([{'role': 'user', 'content': f'PREFILL_{mark}_Q'}, + {'role': 'assistant', 'content': f'PREFILL_{mark}_A'}])) + + def extra(home): + pair(home, 'boot.json', 'BOOT') + return {'prefill_messages_file': 'boot.json'} + + def edit(home): + pair(home, 'edited.json', 'EDITED') + config = json.loads((home / 'config.yaml').read_text()) + config['prefill_messages_file'] = 'edited.json' + (home / 'config.yaml').write_text(json.dumps(config)) + + async def turn(ws, sid, name, settle, servers): + start = len(servers.primary.requests) + assert (await settle(ws, sid, name + '-plain', 'PLAIN_TURN'))['final_response'] == 'PRIMARY_OK' + [request] = servers.primary.requests[start:] + return {'prefill': [m['content'] for m in request['messages'] if 'PREFILL_' in str(m.get('content'))]} + results = run_pair(tmp_path, extra, turn, before_create=edit) + for name in ('inproc', 'managed'): + assert results[name]['prefill'] == ['PREFILL_EDITED_Q', 'PREFILL_EDITED_A'], (name, results) + + def test_construct_inputs_follow_the_session_mode(tmp_path, monkeypatch): - """Ordinary: the owner's per-turn profile refresh, configured prompt and prefill; config-only: + """Ordinary: the owner's per-turn profile refresh, configured prompt and the session's own prefill + (none configured here; never the runner's daemon-start copy); config-only: the frozen snapshot's chain and launch skills only, never the profile; safe mode: nothing. The child accepts only the closed shape.""" from gateway.session_local import _bypass_policy @@ -192,7 +224,7 @@ def test_construct_inputs_follow_the_session_mode(tmp_path, monkeypatch): config_only = _bypass_policy({**launch, 'ignore_user_config': True}, private_secrets={}) safe = _bypass_policy({**launch, 'safe_mode': True}, private_secrets={}) assert construct_inputs(authority, ordinary, 'c') == { - 'fallback_model': chain, 'ephemeral_system_prompt': 'SYS', 'prefill_messages': runner._prefill_messages} + 'fallback_model': chain, 'ephemeral_system_prompt': 'SYS', 'prefill_messages': None} assert profile_reads == ['fallback', 'prompt'] empty = {'fallback_model': None, 'ephemeral_system_prompt': None, 'prefill_messages': None} assert construct_inputs(authority, config_only, 'c') == empty diff --git a/tests/gateway/test_session_config_loaders.py b/tests/gateway/test_session_config_loaders.py index 6e784b5c16..8e90c402db 100644 --- a/tests/gateway/test_session_config_loaders.py +++ b/tests/gateway/test_session_config_loaders.py @@ -1,12 +1,18 @@ -"""A gateway-owned session runs with ``provider_routing`` from its own -profile's config as of its creation, as a new ``hermes chat`` process read it once at startup on -pre-merge main (CFG-6). The daemon-start snapshot on the runner never reaches it; a managed -worker already builds from the same frozen config.""" +"""A gateway-owned session runs with ``provider_routing`` and ``prefill_messages_file`` from its own +profile's config as of its creation, as a new ``hermes chat`` process read them once at startup on +pre-merge main (CFG-6, CFG-7). The daemon-start snapshot on the runner never reaches it, and the +in-process turn and a managed worker get the same values.""" +import json from types import SimpleNamespace import pytest LAUNCH = {'model': 'm', 'provider': 'custom', 'base_url': 'http://127.0.0.1:9/v1'} +STALE = [{'role': 'user', 'content': 'DAEMON_START'}] + + +def _prefill(home, name, mark): + (home / name).write_text(json.dumps([{'role': 'user', 'content': mark}])) class _Built(Exception): @@ -42,6 +48,97 @@ def test_in_process_turn_routes_with_the_sessions_own_provider_routing(tmp_path, assert caught.value.args[0] == {'only': ['Fresh'], 'sort': 'price'} +def test_each_new_session_reads_its_profiles_prefill_once_and_workers_match(tmp_path, monkeypatch): + """The prefill path and file contents a session gets are its own profile's when it was created; a + later edit reaches the next new session and never an existing one (main: one read per CLI + process). Bypass sessions never load the profile's file.""" + from gateway.session_local import _bypass_policy + from gateway.session_policy import build_policy + from gateway.session_worker_construct import construct_inputs + monkeypatch.delenv('HERMES_PREFILL_MESSAGES_FILE', raising=False) + _prefill(tmp_path, 'prefill.json', 'FIRST') + runner = SimpleNamespace(config=None, _prefill_messages=STALE, _refresh_fallback_model=lambda: None, + _get_system_prompt_for_channel=lambda *a: None) + authority = SimpleNamespace(runner=runner, profile_id=str(tmp_path)) + launch = {'cwd': str(tmp_path), **LAUNCH} + first = build_policy(launch, {'prefill_messages_file': 'prefill.json'}) + assert construct_inputs(authority, first, 'sid-1')['prefill_messages'] == [{'role': 'user', 'content': 'FIRST'}] + from gateway.session_worker_construct import session_prefill + _prefill(tmp_path, 'prefill.json', 'EDITED') + _prefill(tmp_path, 'moved.json', 'MOVED') + # The open session keeps what it read (no per-turn re-read); the in-process helper agrees. + assert construct_inputs(authority, first, 'sid-1')['prefill_messages'][0]['content'] == 'FIRST' + assert session_prefill(authority, first, 'sid-1')[0]['content'] == 'FIRST' + # New sessions read the file as it is now, at the path their own config names. + assert session_prefill(authority, first, 'sid-2')[0]['content'] == 'EDITED' + moved = build_policy(launch, {'agent': {'prefill_messages_file': str(tmp_path / 'moved.json')}}) + assert construct_inputs(authority, moved, 'sid-3')['prefill_messages'][0]['content'] == 'MOVED' + assert session_prefill(authority, build_policy(launch, {}), 'sid-4') is None + config_only = _bypass_policy({**launch, 'ignore_user_config': True}, private_secrets={}) + assert session_prefill(authority, config_only, 'sid-5') is None + + +def _in_process_prefill(authority, policy, monkeypatch, chat_id): + """The prefill an in-process agent (re)build hands ``AIAgent`` (``_build_fresh_agent``).""" + from gateway import session_policy + from gateway.session_execution import GatewaySessionAgentMixin + monkeypatch.setattr(session_policy, 'policy_for_source', lambda runner, source: policy) + built = {} + turn = GatewaySessionAgentMixin.__new__(GatewaySessionAgentMixin) + turn._runner = SimpleNamespace(session_authority=authority, _prefill_messages=STALE, _service_tier=None, + _session_db=None, _refresh_fallback_model=lambda: None) + source = SimpleNamespace(chat_id=chat_id, user_id='u', user_id_alt=None, user_name=None, chat_name=None, + chat_type='dm', thread_id=None) + turn._ctx = SimpleNamespace(source=source, user_config={}, enabled_toolsets=[], disabled_toolsets=None, + session_id=chat_id, session_key='k', AIAgent=lambda **kw: built.update(kw)) + turn._build_fresh_agent({'model': 'm', 'runtime': {}}, 'local', None, 1, None, {}, False) + return built['prefill_messages'] + + +def test_an_open_sessions_prefill_survives_cache_pressure_and_file_edits(tmp_path, monkeypatch): + """However many other sessions read prefill meanwhile, an open session keeps the examples it + started with on every later construction: a managed worker's and an in-process rebuild's.""" + from gateway.session_policy import build_policy + from gateway.session_worker_construct import construct_inputs, session_prefill + monkeypatch.delenv('HERMES_PREFILL_MESSAGES_FILE', raising=False) + _prefill(tmp_path, 'prefill.json', 'ORIGINAL') + runner = SimpleNamespace(config=None, _prefill_messages=STALE, _refresh_fallback_model=lambda: None, + _get_system_prompt_for_channel=lambda *a: None) + authority = SimpleNamespace(runner=runner, profile_id=str(tmp_path)) + policy = build_policy({'cwd': str(tmp_path), **LAUNCH}, {'prefill_messages_file': 'prefill.json'}) + assert session_prefill(authority, policy, 'open')[0]['content'] == 'ORIGINAL' + _prefill(tmp_path, 'prefill.json', 'EDITED') + for n in range(600): # past any bounded acceleration cache + session_prefill(authority, policy, f'other-{n}') + assert construct_inputs(authority, policy, 'open')['prefill_messages'][0]['content'] == 'ORIGINAL' + assert _in_process_prefill(authority, policy, monkeypatch, 'open')[0]['content'] == 'ORIGINAL' + + +@pytest.mark.asyncio +async def test_prefill_is_read_when_the_session_is_created_and_dropped_with_it(tmp_path, monkeypatch): + """An edit between ``session.create`` and the first turn never reaches the session (main's CLI + read the file at startup); deleting the session releases its snapshot.""" + from gateway.session_mutations import retire_live_sessions + from gateway.session_worker_construct import construct_inputs + from tests.gateway.test_local_create_env_refs import _owner + monkeypatch.delenv('HERMES_PREFILL_MESSAGES_FILE', raising=False) + home = tmp_path / 'home' + home.mkdir() + (home / 'config.yaml').write_text(CONFIG + 'prefill_messages_file: prefill.json\n') + _prefill(home, 'prefill.json', 'AT_CREATE') + owner, connection = await _owner(home, monkeypatch) + try: + ref, policy = _create(owner, connection, home, 'prefill') + _prefill(home, 'prefill.json', 'BEFORE_FIRST_TURN') + assert construct_inputs(owner, policy, ref.session_id)['prefill_messages'][0]['content'] == 'AT_CREATE' + assert _in_process_prefill(owner, policy, monkeypatch, ref.session_id)[0]['content'] == 'AT_CREATE' + retire_live_sessions(owner, [ref.session_id]) + assert ref.session_id not in owner._local_prefill + finally: + await connection.close() + owner.db.close() + + def test_bg_side_agent_uses_the_sessions_provider_routing(tmp_path): """``/bg`` through its handler: the side agent it constructs routes with the session's ``provider_routing``, not the runner's daemon-start copy.""" @@ -99,3 +196,163 @@ def _create(owner, connection, home, request_id): ref = create_local_session(owner, connection.actor, {'request_id': request_id, 'source': 'cli', 'cwd': str(home), 'toolsets': []}) return ref, restore_policy(local_receipt(owner.db, ref.session_id)['policy']) + + +async def _prefill_owner(tmp_path, monkeypatch): + from tests.gateway.test_local_create_env_refs import _owner + monkeypatch.delenv('HERMES_PREFILL_MESSAGES_FILE', raising=False) + home = tmp_path / 'home' + home.mkdir() + (home / 'config.yaml').write_text(CONFIG + 'prefill_messages_file: prefill.json\n') + _prefill(home, 'prefill.json', 'FIRST') + owner, connection = await _owner(home, monkeypatch) + return home, owner, connection + + +def _stage(owner, connection, cwd, request_id): + from gateway.session_local import stage_local_session + return stage_local_session(owner, connection.actor, {'request_id': request_id, 'source': 'cli', + 'cwd': str(cwd), 'toolsets': []}) + + +def _commit(owner, staged): + """The off-loop commit and publication every async create runs after staging.""" + from gateway.session_local_recovery import restore_local_session + from hermes_state_local import commit_local_session + commit_local_session(owner.db, epoch=owner.epoch, receipt=staged) + return restore_local_session(owner, staged['session_id']) + + +@pytest.mark.asyncio +async def test_the_prefill_file_is_read_by_the_off_loop_commit_not_while_staging(tmp_path, monkeypatch): + """``session.create`` stages on the owner loop and commits through ``tracked_write`` (a worker thread): + the prefill file, possibly on a slow mount, is read by this attempt's commit, never while staging.""" + import threading + from functools import partial + from gateway.session_local_recovery import restore_local_session + from gateway.session_runtime_workers import tracked_write + from hermes_cli import cli_config_load + from hermes_state_local import commit_local_session + home, owner, connection = await _prefill_owner(tmp_path, monkeypatch) + readers, real = [], cli_config_load._load_prefill_messages + monkeypatch.setattr(cli_config_load, '_load_prefill_messages', + lambda *args: readers.append(threading.get_ident()) or real(*args)) + try: + staged = _stage(owner, connection, home, 'slow-mount') + assert readers == [] + await tracked_write(owner, partial(commit_local_session, owner.db, epoch=owner.epoch, receipt=staged)) + assert len(readers) == 1 and readers[0] != threading.get_ident() + restore_local_session(owner, staged['session_id']) + assert _seen(owner, monkeypatch, staged['session_id']) == 'FIRST' and len(readers) == 1 + finally: + await connection.close() + owner.db.close() + + +def _seen(owner, monkeypatch, sid): + """The prefill a managed worker and an in-process build of ``sid`` get (they must agree).""" + from gateway.session_policy import restore_policy + from gateway.session_worker_construct import construct_inputs + from hermes_state_local import local_receipt + policy = restore_policy(local_receipt(owner.db, sid)['policy']) + worker = construct_inputs(owner, policy, sid)['prefill_messages'][0]['content'] + assert _in_process_prefill(owner, policy, monkeypatch, sid)[0]['content'] == worker + return worker + + +@pytest.mark.asyncio +@pytest.mark.parametrize('winner', [0, 1]) +async def test_overlapping_exact_creates_keep_the_committed_attempts_prefill(tmp_path, monkeypatch, winner): + """Two in-flight creates of one identity (a retry before the first commit lands), the file + edited between their commits: the session gets the contents the committing attempt read, + whichever staged last, and only its snapshot is held.""" + home, owner, connection = await _prefill_owner(tmp_path, monkeypatch) + try: + staged = [_stage(owner, connection, home, 'overlap'), _stage(owner, connection, home, 'overlap')] + ordered = staged if winner == 0 else staged[::-1] + refs = [_commit(owner, ordered[0])] # both staged before either commits + _prefill(home, 'prefill.json', 'SECOND') + refs.append(_commit(owner, ordered[1])) # the losing retry reads SECOND, never held + assert refs[0] == refs[1] + _prefill(home, 'prefill.json', 'LATER') + assert _seen(owner, monkeypatch, refs[0].session_id) == 'FIRST' + assert list(owner._local_prefill) == [refs[0].session_id] + finally: + await connection.close() + owner.db.close() + + +@pytest.mark.asyncio +async def test_a_conflicting_losing_create_never_replaces_the_winners_prefill(tmp_path, monkeypatch): + from hermes_state_runtime import RuntimeStoreError + home, owner, connection = await _prefill_owner(tmp_path, monkeypatch) + try: + winner = _stage(owner, connection, home, 'conflict') + loser = _stage(owner, connection, tmp_path, 'conflict') # same identity, different request + ref = _commit(owner, winner) + _prefill(home, 'prefill.json', 'LOSER') + with pytest.raises(RuntimeStoreError) as refused: + _commit(owner, loser) + assert refused.value.reason == 'invalid_params' + assert _seen(owner, monkeypatch, ref.session_id) == 'FIRST' + assert list(owner._local_prefill) == [ref.session_id] + finally: + await connection.close() + owner.db.close() + + +@pytest.mark.asyncio +async def test_a_create_refused_by_a_retirement_tombstone_holds_no_prefill(tmp_path, monkeypatch): + """Retrying the create of a deleted session is refused at commit (``not_found``); the attempt + must not leave a snapshot behind for a session that cannot exist.""" + from gateway.session_mutations import mutate_session + from hermes_state_runtime import RuntimeStoreError + home, owner, connection = await _prefill_owner(tmp_path, monkeypatch) + owner.runner.session_store._db = owner.db # one routing store: the delete's live guard requires it + try: + ref, _policy = _create(owner, connection, home, 'gone') + await mutate_session(owner, connection.actor, ref, dict( + session_id=ref.session_id, request_id='delete', expected_revision=0, expected_generation=0, + operation='delete', payload={})) + assert ref.session_id not in owner._local_prefill + with pytest.raises(RuntimeStoreError) as refused: + _create(owner, connection, home, 'gone') + assert refused.value.reason == 'not_found' + assert ref.session_id not in owner._local_prefill + finally: + await connection.close() + owner.db.close() + + +@pytest.mark.asyncio +async def test_retried_restored_and_branched_sessions_load_prefill_once_and_keep_it(tmp_path, monkeypatch): + """A sequential exact retry keeps the original snapshot; a session restored by an owner that + holds nothing for it (a restart) and a new branch read the file on first use, then keep it.""" + from gateway.session_local_recovery import restore_local_session + from gateway.session_mutations import mutate_session + home, owner, connection = await _prefill_owner(tmp_path, monkeypatch) + try: + ref, _policy = _create(owner, connection, home, 'kept') + _prefill(home, 'prefill.json', 'RETRY') + assert _create(owner, connection, home, 'kept')[0] == ref + assert _seen(owner, monkeypatch, ref.session_id) == 'FIRST' + + owner._local_prefill.clear() # a restarted owner: nothing held for its restored sessions + restore_local_session(owner, ref.session_id) + _prefill(home, 'prefill.json', 'RESTORED') + assert _seen(owner, monkeypatch, ref.session_id) == 'RESTORED' + _prefill(home, 'prefill.json', 'AFTER_RESTORE') + assert _seen(owner, monkeypatch, ref.session_id) == 'RESTORED' + + result = await mutate_session(owner, connection.actor, ref, dict( + session_id=ref.session_id, request_id='branch', expected_revision=0, expected_generation=0, + operation='branch', payload={})) + child = result['branched_session_id'] + _prefill(home, 'prefill.json', 'BRANCHED') + assert _seen(owner, monkeypatch, child) == 'BRANCHED' + _prefill(home, 'prefill.json', 'AFTER_BRANCH') + assert _seen(owner, monkeypatch, child) == 'BRANCHED' + assert _seen(owner, monkeypatch, ref.session_id) == 'RESTORED' + finally: + await connection.close() + owner.db.close()