From 1d62bdcd351e925e977ff4eaf8d1de0dcb57568b Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Thu, 8 Oct 2026 15:43:09 -0700 Subject: [PATCH] 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 | 11 ++-- gateway/session_kanban.py | 4 ++ gateway/session_managed_worker.py | 23 +++++--- gateway/session_surface.py | 19 ++++++ tests/gateway/test_managed_turn_contract.py | 26 +++++++++ tests/gateway/test_surface_admission.py | 65 +++++++++++++++++++++ 6 files changed, 134 insertions(+), 14 deletions(-) diff --git a/gateway/run_turn_runner.py b/gateway/run_turn_runner.py index 75adaeff19..0b1840c63a 100644 --- a/gateway/run_turn_runner.py +++ b/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. diff --git a/gateway/session_kanban.py b/gateway/session_kanban.py index cab0d8d3b1..46fc124188 100644 --- a/gateway/session_kanban.py +++ b/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) diff --git a/gateway/session_managed_worker.py b/gateway/session_managed_worker.py index 79ae75853a..d371243acd 100644 --- a/gateway/session_managed_worker.py +++ b/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 diff --git a/gateway/session_surface.py b/gateway/session_surface.py index 13e47c0789..7e57e5daf4 100644 --- a/gateway/session_surface.py +++ b/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) diff --git a/tests/gateway/test_managed_turn_contract.py b/tests/gateway/test_managed_turn_contract.py index 0736defbae..9d90ef3570 100644 --- a/tests/gateway/test_managed_turn_contract.py +++ b/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 diff --git a/tests/gateway/test_surface_admission.py b/tests/gateway/test_surface_admission.py index d13da83a92..a2fd986514 100644 --- a/tests/gateway/test_surface_admission.py +++ b/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