fix(gateway): carry the committed surface into managed worker turns

Canonical admission commits surface_v1 (HUD / live voice / voice_turn) and the
owner binds it for in-process execution, but the managed-worker bootstrap
dropped it and the child called run_conversation without the agent's one-turn
inputs. A voice-live delegation reached the model without the spoken contract
or its voice_context, and a committed voice turn ran on the main model instead
of auxiliary.voice_chat, while both receipts completed.

- _bootstrap: the committed surface rides the per-turn hydrated request as
  turn_v1.surface_v1 (the closed BOOTSTRAP_FIELDS set is unchanged).
- worker_turn_scope: re-validates it child-side with the admission validator
  (session_surface.restore_surface -> admit_surface round-trip; anything
  admission could not have committed is invalid_managed_worker_bootstrap) and
  binds surface_turn_scope beside finite_turn_scope.
- session_surface.arm_surface_turn: ONE helper sets agent._voice_turn_pending
  and agent._gateway_turn_context_notes; run_turn_runner (in-process) and
  session_kanban.run_worker_turns (child) both call it, unconditionally, so a
  reused agent never replays a stale note or route.

Probe (real daemon, real managed child, loopback model, fake HOME), turns
voice-live+voice_context / voice_turn / plain:
  r6-base managed: VOICE note=False ctx=False; SPOKEN model=main-model
  fixed managed:   VOICE note=True ctx=True;   SPOKEN model=voice-model;
                   PLAIN main-model, no note (worker_executions=3)
  in-process control identical to fixed managed on both trees.

Tests (red on r6-base sources, green after):
- test_surface_admission::test_managed_worker_turn_carries_the_committed_surface_to_the_model
  r6-base: AssertionError: VOICE_LIVE_TURN_NOTE ... in 'VOICE_TURN'
- test_managed_turn_contract::test_committed_surface_crosses_the_bootstrap_validated_and_one_turn_only
  r6-base: AttributeError: 'types.SimpleNamespace' object has no attribute '_voice_turn_pending'
Adjacent: 47 files, 253 tests passed, 0 failed. scripts/check ok.

Review: https://github.com/NousResearch/hermes-agent/pull/106742#pullrequestreview-5461749257

gateway/run_turn_runner.py

@@ -385,14 +385,11 @@ class TurnRunner(GatewayTurnProgressMixin, GatewaySessionAgentMixin):
         agent.notice_clear_callback = None  # sends can't be retracted
         agent.event_callback = ctx._event_callback_sync
         agent.reasoning_config, agent.service_tier = reasoning_config, runner._service_tier
-        from gateway.session_surface import surface_turn_note, surface_voice_turn
-        # auxiliary.voice_chat route: a voice note, or a canonical admission committed as a voice turn.
-        agent._voice_turn_pending = ctx.voice_turn or surface_voice_turn()
         self._merge_turn_request_overrides(agent, turn_route)
-        # Must-deliver notes for THIS turn ride the current user message (api_content sidecar), never
-        # the system prompt. Assigned unconditionally so a reused agent never replays a stale note.
-        agent._gateway_turn_context_notes = "\n\n".join(
-            note for note in (*runner._consume_pending_turn_sidecar_notes(ctx.session_key), surface_turn_note(agent)) if note)
+        from gateway.session_surface import arm_surface_turn
+        # auxiliary.voice_chat route (a voice note, or an admission committed as a voice turn) and
+        # this turn's must-deliver notes; a managed worker arms its agent through the same helper.
+        arm_surface_turn(agent, runner._consume_pending_turn_sidecar_notes(ctx.session_key), voice_turn=ctx.voice_turn)
         agent.background_review_callback, bg_release = self._make_bg_review_callbacks()
         # Register the release hook on the adapter so base.py's finally block fires it after the
         # main response is delivered.

gateway/session_kanban.py

@@ -202,7 +202,11 @@ def _watch_submitter(db, run_id):
 
 def run_worker_turns(agent, frame, history):
     from gateway.session_managed_worker import worker_turn_scope
+    from gateway.session_surface import arm_surface_turn
     with worker_turn_scope(frame):
+        # The agent consumes explicit one-turn attributes, not the ContextVar: arm them exactly as
+        # the in-process runner does (voice route marker + surface note on the user message).
+        arm_surface_turn(agent)
         return _run_worker_turns(agent, frame, history)
 
 

gateway/session_managed_worker.py

@@ -60,10 +60,12 @@ def _bootstrap(authority, ref, row, policy, scope):
                 terminal[path[1]] = value
     live = authority.sessions[ref.session_id]
     # This turn's facts ride the per-turn hydrated request (the bootstrap field set is closed):
-    # the admission's one-shot flags and the route's YOLO as of now, never the frozen launch flag.
+    # the admission's one-shot flags, the route's YOLO as of now (never the frozen launch flag) and
+    # the committed surface (HUD / live voice / voice turn) when admission recorded one.
     request = dict(json.loads(policy.request_json), turn_v1={
         'finite': row['payload'].get('finite', False), 'unattended': row['payload'].get('unattended') is True,
-        'yolo': _session_yolo(authority, live.route, policy)})
+        'yolo': _session_yolo(authority, live.route, policy),
+        **({'surface_v1': row['payload']['surface_v1']} if row['payload'].get('surface_v1') else {})})
     from gateway.session_worker_construct import construct_inputs
     request['construct_v1'] = construct_inputs(authority, policy, live.source.chat_id)
     hydrated = replace(policy, config_json=json.dumps(_worker_config(authority, policy)), request_json=json.dumps(request),
@@ -109,20 +111,27 @@ def _session_yolo(authority, route, policy):
 @contextmanager
 def worker_turn_scope(frame):
     """Child side of ``turn_v1``: bind what the in-process turn binds on the owner
-    (execute_finite_admission, the route's YOLO), so ``chat -q``/``-z`` never park a prompt and
-    the session's current YOLO governs this child. Frames without it bind nothing."""
+    (execute_finite_admission, the route's YOLO, the committed surface), so ``chat -q``/``-z``
+    never park a prompt, the session's current YOLO governs this child and a HUD / live-voice /
+    voice turn reaches the model as it does in process. Frames without it bind nothing."""
     from gateway.session_finite import finite_turn_scope
+    from gateway.session_surface import restore_surface, surface_turn_scope
     turn = json.loads(frame['policy'].get('request_json') or '{}').get('turn_v1')
     if turn is None:
         yield
         return
-    if (not isinstance(turn, dict) or set(turn) != {'finite', 'unattended', 'yolo'}
-            or any(type(v) is not bool for v in turn.values()) or (turn['unattended'] and not turn['finite'])):
+    flags = {k: v for k, v in turn.items() if k != 'surface_v1'} if isinstance(turn, dict) else None
+    if (flags is None or set(flags) != {'finite', 'unattended', 'yolo'}
+            or any(type(v) is not bool for v in flags.values()) or (turn['unattended'] and not turn['finite'])):
         raise ValueError('invalid_managed_worker_bootstrap')
+    try:
+        surface = restore_surface(turn['surface_v1']) if 'surface_v1' in turn else None
+    except RuntimeStoreError as exc:
+        raise ValueError('invalid_managed_worker_bootstrap') from exc
     if turn['yolo']:
         from tools.approval import enable_session_yolo
         enable_session_yolo(frame['route'])
-    with finite_turn_scope(turn['finite'], turn['unattended']):
+    with finite_turn_scope(turn['finite'], turn['unattended']), surface_turn_scope(surface):
         yield
 
 

gateway/session_surface.py

@@ -45,6 +45,15 @@ def admit_surface(params):
     return {'surface_v1': committed} if committed else {}
 
 
+def restore_surface(committed):
+    """A committed ``surface_v1`` arriving across a process boundary (managed worker), re-checked by
+    the admission validator: only an object admission itself could have committed is accepted."""
+    if (not isinstance(committed, dict) or not committed or set(committed) - set(_SUBMIT_FIELDS)
+            or admit_surface(committed).get('surface_v1', {}) != committed):
+        raise RuntimeStoreError('invalid_params')
+    return committed
+
+
 def _hud_note(committed, valid_tool_names):
     from agent.prompt_builder import hud_surface_note
     return hud_surface_note(valid_tool_names)
@@ -91,3 +100,13 @@ def surface_turn_note(agent):
     if not committed:
         return ''
     return surface_note(committed, getattr(agent, 'valid_tool_names', None))
+
+
+def arm_surface_turn(agent, notes=(), voice_turn=False):
+    """Project the executing admission's surface onto the agent's one-turn inputs, the same way in
+    process and in a managed worker: the ``auxiliary.voice_chat`` marker and the must-deliver notes
+    that ride the current user message (never the system prompt). Assigned unconditionally so a
+    reused agent never replays a stale note or route."""
+    agent._voice_turn_pending = bool(voice_turn) or surface_voice_turn()
+    agent._gateway_turn_context_notes = '\n\n'.join(
+        note for note in (*notes, surface_turn_note(agent)) if note)

tests/gateway/test_managed_turn_contract.py

@@ -107,6 +107,32 @@ def test_current_yolo_revocation_overrides_frozen_launch(tmp_path, monkeypatch):
     clear_session('managed-contract')
 
 
+
+def test_committed_surface_crosses_the_bootstrap_validated_and_one_turn_only(tmp_path, monkeypatch):
+    from agent.managed_worker import validate_bootstrap
+    from gateway.session_kanban import run_worker_turns
+    from tools.voice_live import VOICE_LIVE_TURN_NOTE
+    frame, (authority, ref, row, policy, scope) = assignment(tmp_path, monkeypatch)
+    surface = {'surface': 'voice-live', 'voice_context': 'User: UNIQUE_PRIOR_CONTEXT', 'voice_turn': True}
+    from gateway.session_managed_worker import _bootstrap
+    voiced = validate_bootstrap(json.loads(json.dumps(_bootstrap(
+        authority, ref, {'payload': {**row['payload'], 'surface_v1': surface}}, policy, scope))))
+    seen = []
+    agent = SimpleNamespace(valid_tool_names=set(), run_conversation=lambda text, **kw: seen.append(
+        (agent._voice_turn_pending, agent._gateway_turn_context_notes)) or {'final_response': 'ok'})
+    run_worker_turns(agent, voiced, [])
+    run_worker_turns(agent, frame, [])  # the same (reused) agent's next, plain turn
+    (marker, notes), plain = seen
+    assert marker is True and notes.startswith(VOICE_LIVE_TURN_NOTE) and 'UNIQUE_PRIOR_CONTEXT' in notes
+    assert plain == (False, '')
+    # The child re-checks the object with the admission validator: no smuggled model input.
+    request = json.loads(voiced['policy']['request_json'])
+    request['turn_v1']['surface_v1'] = {'voice_context': 'User: smuggled'}
+    forged = {**voiced, 'policy': {**voiced['policy'], 'request_json': json.dumps(request)}}
+    with pytest.raises(ValueError, match='invalid_managed_worker_bootstrap'):
+        run_worker_turns(agent, forged, [])
+    assert len(seen) == 2
+
 @pytest.mark.asyncio
 async def test_stop_during_environment_preparation_prevents_bootstrap(tmp_path, monkeypatch):
     from gateway import session_managed_worker as managed

tests/gateway/test_surface_admission.py

@@ -179,3 +179,68 @@ def test_real_daemon_accepts_hud_and_voice_live_submits_and_notes_the_model(tmp_
     assert 'HUD mode' not in users['hud'] and 'HUD mode' not in users['window']
     assert all(request['messages'][0]['role'] == 'system' for request in model.requests)
     assert len({request['messages'][0]['content'] for request in model.requests}) == 1, 'frozen system prefix must not vary per surface'
+
+
+def test_managed_worker_turn_carries_the_committed_surface_to_the_model(tmp_path, monkeypatch):
+    """The same committed surface reaches the model from a managed child (N2): live-voice note and
+    spoken context on the user message, ``voice_turn`` on ``auxiliary.voice_chat``, and the following
+    typed turn carries neither."""
+    from tools.voice_live import VOICE_LIVE_TURN_NOTE
+
+    root = Path(__file__).resolve().parents[2]
+    home, user = tmp_path / 'state', tmp_path / 'user'
+    home.mkdir(mode=0o700)
+    user.mkdir()
+    model = ThreadingHTTPServer(('127.0.0.1', 0), Model)
+    model.requests = []
+    model.blocked, model.release = threading.Event(), threading.Event()
+    thread = threading.Thread(target=model.serve_forever, daemon=True)
+    thread.start()
+    base = f'http://127.0.0.1:{model.server_port}/v1'
+    (home / 'config.yaml').write_text(json.dumps({
+        'gateway': {'multiplex_profiles': False, 'managed_workers': True},
+        'model': {'provider': 'custom', 'default': 'main-model', 'base_url': base},
+        'auxiliary': {'title_generation': {'enabled': False}, 'voice_chat': {
+            'provider': 'custom', 'model': 'voice-model', 'base_url': base, 'api_key': 'loopback-only'}},
+        'platform_toolsets': {'cli': []},
+    }))
+    env = child_env() | dict(HOME=str(user), USERPROFILE=str(user), HERMES_HOME=str(home),
+        PYTHONPATH=str(root), OPENAI_API_KEY='loopback-only', OPENAI_BASE_URL=base, PYTHONUNBUFFERED='1')
+
+    async def drive(desc):
+        async with websocket(home, desc) as ws:
+            created = await rpc(ws, 'session.create', request_id='managed-surface', source='cli', cwd=str(home),
+                                model='main-model', provider='custom', base_url=base, api_key='loopback-only',
+                                toolsets=[], ignore_rules=True)
+            assert 'result' in created, created
+            for name, extra in (('voice', {'surface': 'voice-live', 'voice_context': 'User: UNIQUE_PRIOR_CONTEXT'}),
+                                ('spoken', {'voice_turn': True}), ('plain', {})):
+                reply = await rpc(ws, 'prompt.submit', session_id=created['result']['session_id'],
+                                  submission_id=name, text=name.upper() + '_TURN', **extra)
+                assert reply.get('result', {}).get('status') == 'queued', reply
+                await asyncio.to_thread(_wait_for, lambda: _admissions(home).get(name) == 'terminal')
+
+    try:
+        with daemon(root, home, env, barrier=False) as (proc, desc):
+            asyncio.run(drive(desc))
+    finally:
+        model.release.set()
+        model.shutdown()
+        model.server_close()
+        thread.join(timeout=5)
+
+    with closing(sqlite3.connect(f'file:{home / "state.db"}?mode=ro', uri=True)) as db:
+        assert db.execute('SELECT COUNT(*) FROM worker_executions').fetchone()[0] == 3, 'every turn ran in a child'
+    turns = {}
+    for request in model.requests:
+        content = next(m['content'] for m in reversed(request['messages']) if m['role'] == 'user')
+        text = content if isinstance(content, str) else ''.join(p.get('text', '') for p in content)
+        name = next((n for n in ('voice', 'spoken', 'plain') if n.upper() + '_TURN' in text), None)
+        if name:
+            turns[name] = (request['model'], text)
+    assert set(turns) == {'voice', 'spoken', 'plain'}, turns
+    assert VOICE_LIVE_TURN_NOTE in turns['voice'][1] and 'UNIQUE_PRIOR_CONTEXT' in turns['voice'][1]
+    assert turns['spoken'][0] == 'voice-model', 'a committed voice turn runs on auxiliary.voice_chat'
+    assert turns['plain'] == ('main-model', 'PLAIN_TURN'), 'the following typed turn carries no surface'
+    # The voice route rewrites only its own request's model identity; main-model turns share one prefix.
+    assert len({r['messages'][0]['content'] for r in model.requests if r['model'] == 'main-model'}) == 1