fix(acp): every fresh session snapshot projects the edit mode

A model pick, a confirmed deferred pick, /new, /compress, /model, /undo and
/retry installed a fresh session.resume snapshot and its watermark directly.
A session.info carrying another viewer's mode change still queued behind that
watermark was then skipped, so the editor showed Default while the owner
auto-allowed edits. Every snapshot install (create, load, fork included) now
goes through _adopt, which projects the snapshot's edit policy.

acp_adapter/gateway_commands.py

@@ -119,7 +119,7 @@ async def _rewind(agent, client, session_id, command, arg):
         return t('cli.session.retry_no_user_message' if retry else 'cli.session.undo_no_user_message'), None
     # Retained until the refresh lands: a repeat after a lost refresh replays this rewind instead of
     # deriving a new target from the already-shortened transcript (which would drop the turn before).
-    agent._snapshots[session_id] = await client.rpc('session.resume', session_id=session_id)
+    await agent._adopt(session_id, await client.rpc('session.resume', session_id=session_id))
     agent._mutations.acknowledge(session_id, 'rewind', payload)
     content = (result.get('target_message') or {}).get('content')
     if retry:
@@ -174,7 +174,7 @@ async def _mutate(agent, client, session_id, command, arg):
     if unchanged is not None:
         reply = unchanged(result)
     else:
-        agent._snapshots[session_id] = await client.rpc('session.resume', session_id=session_id)
+        await agent._adopt(session_id, await client.rpc('session.resume', session_id=session_id))
         if operation in {'reset', 'compress'}:
             await _refresh_context(agent, client, session_id, operation)
         reply = {'model': lambda: _model_text('Model switched to', result),

acp_adapter/gateway_server.py

@@ -197,7 +197,8 @@ class GatewayACPAgent(GatewayReplyMixin, acp.Agent):
             await self._notify_unknown(session_id)
 
     async def _adopt(self, session_id, snapshot, resumes=None):
-        """Replace a session's projection with a fresh ``session.resume`` snapshot (reconnect, replay gap).
+        """Install a fresh ``session.resume`` snapshot as a session's projection: every install (create, load,
+        fork, a control's refresh, reconnect, replay gap) goes through here.
 
         Events before its watermark are skipped from here on, so the session state they would have carried
         is projected from the snapshot itself: the editor is never left showing Default while the owner
@@ -279,7 +280,7 @@ class GatewayACPAgent(GatewayReplyMixin, acp.Agent):
             editor={"mcp_servers": [s.model_dump(by_alias=True) for s in mcp_servers or []],
                     "edit_approval_policy": "ask"}, **flags)
         session_id = snapshot["session_id"]
-        self._snapshots[session_id] = snapshot
+        await self._adopt(session_id, snapshot)
         if mcp_servers:
             self._editor_mcp[session_id] = {
                 "mcp_servers": [s.model_dump(by_alias=True) for s in mcp_servers]}
@@ -363,7 +364,7 @@ class GatewayACPAgent(GatewayReplyMixin, acp.Agent):
             # capability, idle only), as the workspace move above, instead of refusing the thread.
             await self._rebind_editor_mcp(client, session_id, resume_params['editor'])
             snapshot = await client.rpc("session.resume", session_id=session_id, **resume_params)
-        self._snapshots[session_id] = snapshot
+        await self._adopt(session_id, snapshot)
         if 'editor' in resume_params:
             self._editor_mcp[session_id] = resume_params['editor']
         from acp_adapter.server import _history_replay_updates
@@ -454,7 +455,7 @@ class GatewayACPAgent(GatewayReplyMixin, acp.Agent):
         child = result['branched_session_id']
         if self._moved(info, cwd):
             await self._move_workspace(client, child, cwd)
-        self._snapshots[child] = await client.rpc('session.resume', session_id=child)
+        await self._adopt(child, await client.rpc('session.resume', session_id=child))
         if payload or session_id in self._editor_mcp:
             self._editor_mcp[child] = payload or self._editor_mcp[session_id]
         self._mutations.acknowledge(session_id, 'branch', payload)
@@ -476,7 +477,7 @@ class GatewayACPAgent(GatewayReplyMixin, acp.Agent):
         if result.get('status') == 'cancelled':
             # Declined: nothing was written; the editor's picker must not show the target as active.
             raise GatewayClientError('model_switch_cancelled')
-        self._snapshots[session_id] = await client.rpc('session.resume', session_id=session_id)
+        await self._adopt(session_id, await client.rpc('session.resume', session_id=session_id))
         self._mutations.acknowledge(session_id, 'model', payload)
         if result.get('status') == 'deferred' and self._conn:
             # A pick during a running turn lands for the next one (#139604); say so, as Desktop does.
@@ -515,7 +516,7 @@ class GatewayACPAgent(GatewayReplyMixin, acp.Agent):
                 await say(model_refusal_text(result, confirmed=True))
                 return
             if session_id in self._snapshots:
-                self._snapshots[session_id] = await client.rpc("session.resume", session_id=session_id)
+                await self._adopt(session_id, await client.rpc("session.resume", session_id=session_id))
             # Confirmed while a turn runs: stored again for the next boundary, said as such.
             await say(deferred_pick_text(result) if result.get("status") == "deferred"
                       else _model_text("Model switched to", result))

tests/acp_adapter/test_gateway_snapshot_adoption.py

@@ -0,0 +1,73 @@
+"""Every fresh ``session.resume`` the editor installs advances the event watermark, so a ``session.info``
+queued behind it (another viewer's mode change) is skipped by the pump. Each such install projects the
+snapshot's own edit policy, so the editor never shows Default while the owner auto-allows edits."""
+import asyncio
+from unittest.mock import AsyncMock
+
+import pytest
+
+from acp_adapter.gateway_commands import run_command
+from acp_adapter.gateway_server import GatewayACPAgent
+
+OWNER = {'edit_approval_policy': 'session'}  # another viewer already chose Don't Ask (seq 11)
+
+
+def _set_model(agent):
+    return agent.set_session_model('p:m', 's')
+
+
+async def _deferred_confirm(agent):
+    async def yes(ask):
+        return True
+    agent._model_confirmer = lambda sid: yes
+    await agent._confirm_deferred_pick('s', 's', {'payload': {'model': 'm'}, 'confirm': 'tok'})
+
+
+def _command(text):
+    command, _, arg = text.partition(' ')
+    return lambda agent: run_command(agent, 's', command, arg)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize('install', [_set_model, _deferred_confirm, _command('new'), _command('compress'),
+                                     _command('model p:m'), _command('undo'), _command('retry')],
+                         ids=['set_session_model', 'deferred_confirm', 'new', 'compress', 'model', 'undo', 'retry'])
+async def test_a_mode_change_queued_behind_a_new_watermark_still_repaints_the_editor(install, monkeypatch):
+    agent = GatewayACPAgent()
+    agent._conn = AsyncMock()
+    agent._snapshots['s'] = {'pending': [], 'replay_epoch': 'e', 'last_sequence': 10,
+                             'execution_generation': 1, 'revision': 1}
+    agent._modes['s'] = 'default'
+
+    async def rpc(method, **params):
+        if method == 'session.resume':
+            return {'session_id': 's', 'pending': [], 'prompts': [], 'replay_epoch': 'e', 'last_sequence': 12,
+                    'execution_generation': 1, 'revision': 3, 'info': dict(OWNER),
+                    'messages': [{'row_id': 1, 'role': 'user', 'content': 'hello'}]}
+        if method == 'session.mutate':
+            return {'session_id': 's', 'operation': params['operation'], 'revision': 3, 'model': 'm',
+                    'provider': 'p', 'rewound_count': 1, 'target_message': {'content': 'hello'}}
+        if method == 'session.stats':
+            return {'context_length': 1000, 'context_tokens': 0}
+        raise AssertionError(method)
+
+    gateway = type('Gateway', (), {})()
+    gateway.events, gateway.rpc = asyncio.Queue(), rpc
+    agent._gateway = gateway
+
+    async def client():
+        return gateway
+    monkeypatch.setattr(agent, '_client', client)
+    from hermes_cli import models
+    monkeypatch.setattr(models, 'parse_model_input', lambda m, d: ('p', 'm'))
+    await install(agent)
+    # The owner's session.info for the other viewer's change (seq 11) is delivered after the resume.
+    await gateway.events.put({'params': {'session_id': 's', 'type': 'session.info', 'replay_epoch': 'e', 'seq': 11,
+                                         'payload': dict(OWNER)}})
+    pump = asyncio.create_task(agent._events())
+    await asyncio.sleep(0.1)
+    pump.cancel()
+    modes = [c.kwargs['update'].current_mode_id for c in agent._conn.session_update.await_args_list
+             if getattr(c.kwargs['update'], 'session_update', None) == 'current_mode_update']
+    assert agent._snapshots['s']['last_sequence'] == 12
+    assert agent._modes['s'] == 'dont_ask' and modes == ['dont_ask']