fix(gateway): read a new session's prefill_messages_file from its own config

Gateway-owned sessions (in-process turns and managed workers alike) got
the runner's prefill examples, loaded once when the daemon started from
the launch profile: a prefill_messages_file path or file edited after
that never reached a new session. On pre-merge main every `hermes chat`
process read the file once at startup from its own profile.

session_prefill() resolves the path from the session policy's
creation-time config, relative to the session's own profile home. The
file is read by the creation attempt's commit (commit_local_session),
before its write transaction; on the async create paths that runs in
tracked_write's worker thread, not on the owner's event loop. The
snapshot is carried with that creation attempt. It is held in the owning
authority's memory only when that attempt's commit creates the session.
An overlapping retry or a conflicting create never replaces it, and a
refused or failed create (for example a retry of a deleted session) holds
nothing. Other sessions cannot evict it, an edit after creation reaches
only new sessions, and deleting the session releases it. A session this
owner did not create (restored after a gateway restart, or a new branch)
reads the file once on first use and keeps it; main's CLI likewise
re-read it on every start. The in-process agent build and
construct_inputs for managed workers share the snapshot, so both send the
same examples. Bypass sessions still never load the profile's file. The
HERMES_PREFILL_MESSAGES_FILE env precedence is unchanged.

Related open PR, not used: #124630 (re-reads the active profile on every
agent build for multiplexed messaging turns, against the pre-split
layout; does not read the session's frozen config).

Refs #134889 (CFG-7)
Refs #139523

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"),

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):

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)

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."""

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):

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

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()