diff --git a/agent/managed_worker.py b/agent/managed_worker.py index 2eaec31292..23c412b191 100644 --- a/agent/managed_worker.py +++ b/agent/managed_worker.py @@ -23,6 +23,9 @@ import threading logger = logging.getLogger(__name__) MAX_FRAME = 4 * 1024 * 1024 +# Toolsets a managed worker turn never offers (subtracted last, so the tool_search bridge and a +# restored tools[] pin drop them too): delegate_task's child has no owner registration here. +WORKER_DISABLED_TOOLSETS = ('delegation',) BOOTSTRAP_FIELDS = {'version', 'home', 'scope', 'policy', 'api_key', 'text', 'route', 'user_id', 'chat_id', 'turn_author', 'safe_mode', 'ignore_user_config'} # The bounded run_conversation fields a turn receipt keeps (outcome, exit reason, tokens, cost, @@ -313,6 +316,9 @@ def execute(frame, channel): agent = _construct_agent(frame, AIAgent, model=policy.model, provider=policy.provider, base_url=policy.base_url, api_key=frame['api_key'], session_db=store, session_id=scope['session_id'], enabled_toolsets=list(policy.toolsets), max_iterations=policy.max_turns, + # A delegated child needs its own owner-registered execution, which this single-turn + # worker cannot reserve; never offer a tool whose only outcome here is a refusal. + disabled_toolsets=list(WORKER_DISABLED_TOOLSETS), reasoning_config=policy.reasoning_config, # ``tool`` / ``oneshot`` are storage labels for a CLI-surface agent (session_policy.SURFACES). platform=policy.platform if policy.source in ('tool', 'oneshot') else policy.source, diff --git a/tests/gateway/test_managed_worker_no_delegation.py b/tests/gateway/test_managed_worker_no_delegation.py new file mode 100644 index 0000000000..d185eb5fc6 --- /dev/null +++ b/tests/gateway/test_managed_worker_no_delegation.py @@ -0,0 +1,83 @@ +"""A managed worker turn never offers ``delegate_task``: its child would need an owner-registered +execution this single-turn worker cannot reserve, so the tool's only outcome would be a refusal +(real daemon, real managed worker, loopback model).""" +import asyncio +from contextlib import closing +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +import json +import os +from pathlib import Path +import sqlite3 +import threading + +import pytest + +from tests.gateway.fixtures.local_recovery_probe import daemon, rpc, websocket + + +class Model(BaseHTTPRequestHandler): + def log_message(self, *args): + pass + + def do_POST(self): + body = json.loads(self.rfile.read(int(self.headers['Content-Length']))) + if body.get('messages'): + self.server.offered.append(json.dumps(body.get('tools') or [])) + message = {'role': 'assistant', 'content': 'WORKER_DONE'} + frame = {'id': 'tools', 'model': 'm', 'choices': [{'index': 0, 'delta': message, 'finish_reason': 'stop'}], + 'usage': {'prompt_tokens': 1, 'completion_tokens': 1, 'total_tokens': 2}} + if body.get('stream'): + payload, kind = ('data: ' + json.dumps(frame) + '\n\ndata: [DONE]\n\n').encode(), 'text/event-stream' + else: + frame['choices'][0]['message'] = frame['choices'][0].pop('delta') + payload, kind = json.dumps(frame).encode(), 'application/json' + self.send_response(200) + self.send_header('Content-Type', kind) + self.send_header('Content-Length', str(len(payload))) + self.end_headers() + self.wfile.write(payload) + + +@pytest.mark.platforms("linux") +def test_managed_worker_turn_does_not_offer_delegate_task(tmp_path): + root = Path(__file__).resolve().parents[2] + home, user = tmp_path / 'state', tmp_path / 'user' + home.mkdir(mode=0o700) + user.mkdir() + peer = ThreadingHTTPServer(('127.0.0.1', 0), Model) + peer.offered = [] + threading.Thread(target=peer.serve_forever, daemon=True).start() + url = f'http://127.0.0.1:{peer.server_port}/v1' + (home / 'config.yaml').write_text(json.dumps({ + 'gateway': {'multiplex_profiles': False, 'managed_workers': True}, + 'model': {'provider': 'custom', 'default': 'm', 'base_url': url}, + 'auxiliary': {'title_generation': {'enabled': False}}, + 'platform_toolsets': {'cli': ['delegation', 'todo']}})) + env = {k: os.environ[k] for k in ('PATH', 'LANG', 'TZ') if k in os.environ} + env.update(HOME=str(user), USERPROFILE=str(user), HERMES_HOME=str(home), PYTHONPATH=str(root), + OPENAI_API_KEY='loopback-only', OPENAI_BASE_URL=url) + + def rows(sql): + with closing(sqlite3.connect(f'file:{home / "state.db"}?mode=ro', uri=True)) as db: + return db.execute(sql).fetchall() + + async def exercise(desc): + async with websocket(home, desc) as ws: + session = await rpc(ws, 'session.create', request_id='tools', source='cli', cwd=str(home), model='m', + provider='custom', base_url=url, api_key='loopback-only', + toolsets=['delegation', 'todo'], ignore_rules=True) + sid = session['result']['session_id'] + submitted = await rpc(ws, 'prompt.submit', session_id=sid, input_id='run', text='hello') + assert 'result' in submitted, submitted + async with asyncio.timeout(90): + while rows("SELECT status FROM session_admissions WHERE request_id='run'") != [('terminal',)]: + await asyncio.sleep(.1) + try: + with daemon(root, home, env, barrier=False) as (_owner, desc): + asyncio.run(exercise(desc)) + finally: + peer.shutdown() + peer.server_close() + assert peer.offered, 'the managed worker never reached the model' + assert all('todo_list' in tools for tools in peer.offered), peer.offered # the turn kept its other tools + assert not any('delegate_task' in tools for tools in peer.offered), peer.offered