"""Python agent API: spawn, family identity, refusals, progress, explicit mail, agents.get, reading messages across sessions, and listing and searching them.""" import json import re import time import unittest from harness import Albedo, Provider, exclusive, latest_user, python, text LEAD = """try: await agents.self.spawn('guess', name='guess', model='alpha/fixture-guessed') except RuntimeError as error: print('UNLISTED', error) kid = await agents.self.spawn('count to three', name='scout', model='alpha/fixture-alpha') models = await agents.models() assert models[0] == 'alpha/fixture-alpha', models assert all('/' in model for model in models), models assert 'beta/fixture-beta' in models, models try: await agents.self.spawn('again', name='scout2', model='') except TypeError as error: print('EMPTY_MODEL', error) print('SPAWNED', kid.name, kid.depth, kid.parent.name, 'KID=' + kid.id) """ CHILD = """me = agents.self print('ME', me.name, me.depth, me.parent.name) print('PROGRESS', await agents.progress('counting')) print('Café Ünïcode') try: await me.parent.cancel() except AgentsError as error: print('REFUSED', error) page = await me.parent.messages() print('PARENT_READ', 'spawn a scout' in page.content) up = await agents.get('parent') print('GOT', up.id == me.parent.id, up.depth, up.parent, up.closed) try: await me.parent.delete() except AgentsError as error: print('ASK', error) receipt = await mail.submit('parent', 'three') print('SENT', receipt.status, receipt.name) """ READ = """kid = (await agents.self.children())[0] page = await kid.messages() print('TASK_SEEN', 'count to three' in page.content) hits = await kid.search_messages('COUNT TO THREE') seq = hits.rows[0]['seq'] row = await kid.messages(seq=seq, limit=200) print('FOUND', hits.count, row.content.startswith(f'[row #{seq}]'), 'count to three' in row.content) """ PEEK = """async def paged_sessions(query='', cwd=None): found = [] while True: page = await agents.sessions(query, cwd=cwd, offset=len(found)) if not page: return found found.extend(page) kid = await agents.get({kid!r}) print('PEEK', kid.name, kid.depth, kid.closed) page = await kid.messages() print('PEEK_READ', 'count to three' in page.content) try: await kid.cancel() except AgentsError as error: print('PEEK_REFUSED', error) listed = await paged_sessions() print('LISTED', {kid!r} in [s.id for s in listed], agents.self.id in [s.id for s in listed]) child_session = next(s for s in listed if s.id == {kid!r}) scouts = await paged_sessions('SCOUT', cwd=child_session.cwd) scout = next(s for s in scouts if s.id == {kid!r}) print('BY_NAME', scout.name, scout.depth, scout.model, scout.cwd == child_session.cwd) print('ELSEWHERE', await paged_sessions(cwd='/nowhere')) talked = await paged_sessions('COUNT TO THREE') said = next(s for s in talked if s.id == {kid!r}) hit = said.matches[0] row = await said.messages(seq=hit['seq'], limit=200) print('SEARCHED', agents.self.id in [s.id for s in talked], 'count to three' in hit['preview'], 'count to three' in row.content) print('UNQUERIED', listed[0].matches) print('UNICODE', {kid!r} in [s.id for s in await paged_sessions('CAFÉ ÜNÏCODE')]) print('NOTHING', await paged_sessions('never ' + 'said anywhere')) try: await agents.get('nobody') except AgentsError as error: print('PEEK_MISSING', error) """ RESOURCES = """await commands.rename('worker-box') job = run('sleep', '47') mine = (await resources.query(session='self')).sessions[0] sleeper = [p for p in mine.processes if p.name == 'sleep'] print('MINE', len(sleeper), sleeper[0].job_id == job.id, mine.session == agents.self.id, hasattr(agents.self, 'resources')) print('TOTAL', mine.process_count >= 2, mine.total.process_count == mine.process_count, mine.memory_bytes > 0) found = await resources.query('sleep 47') by_session_name = await resources.query('worker-box') by_named_target = await resources.query(session='worker-box') print('QUERY', [s.session for s in found.sessions] == [agents.self.id], found.process_count, by_session_name.session_count == 1, by_named_target.session_count == 1) here = await resources.query(process='SLEEP', cwd=sleeper[0].cwd, session=agents.self) print('CWD', here.process_count, here.sessions[0].session == agents.self.id) print('ELSEWHERE', (await resources.query('sleep', cwd='/nowhere')).session_count) await job.stop() print('GONE', len([p for p in (await resources.query(session=agents.self.id)).processes if p.name == 'sleep'])) """ def wait_for(predicate, timeout=60): deadline = time.monotonic() + timeout while time.monotonic() < deadline: result = predicate() if result: return result time.sleep(0.05) raise AssertionError(f"agent Python turn never arrived within {timeout}s") class AgentsPythonTests(unittest.TestCase): def test_python_spawn_identity_refusals_and_explicit_answer(self): def script(request): last = request["messages"][-1] user = latest_user(request) peek = re.search(r"peek ([0-9a-f]+)", user) if last.get("role") == "user" and peek: return python(PEEK.format(kid=peek.group(1))) if last.get("role") == "user" and user == "spawn a scout": return python(LEAD) if last.get("role") == "user" and 'kind="task"' in user: return python(CHILD) if last.get("role") == "user" and 'kind="message"' in user: return python(READ) return text("ok") provider = Provider(script) providers = { name: { "baseUrl": provider.url + f"/{name}/v1", "apiKey": "key", "model": f"fixture-{name}", "protocol": "chat_completions", } for name in ("alpha", "beta") } try: with Albedo(provider, providers=providers) as app: lead = app.session(name="lead") app.prompt(lead, "spawn a scout").close() def observed(): records = [] for item in provider.requests: messages = item["request"]["messages"] records.append( { "user": latest_user(item["request"]), "context": " ".join( str(m.get("content", "")) for m in messages ), "tool": " ".join( str(m.get("content", "")) for m in messages if m.get("role") == "tool" ), } ) return records spawned = wait_for( lambda: next( (r["tool"] for r in observed() if "SPAWNED" in r["tool"]), None ) ) self.assertIn("SPAWNED scout 1", spawned) self.assertIn("UNLISTED call await agents.models() first", spawned) self.assertIn("EMPTY_MODEL", spawned) child = wait_for( lambda: next( (r["tool"] for r in observed() if "SENT" in r["tool"]), None ) ) self.assertIn("ME scout 1", child) self.assertIn("PROGRESS True", child) self.assertIn("REFUSED", child) self.assertIn("own children", child) self.assertIn("PARENT_READ True", child) self.assertIn("GOT True 0 None False", child) self.assertIn("ASK", child) self.assertIn("only the user deletes", child) self.assertTrue( "SENT delivered lead" in child or "SENT queued lead" in child ) task = next(r for r in observed() if 'kind="task"' in r["user"]) self.assertIn('You are child agent "scout"', task["context"]) wait_for( lambda: any( 'kind="message"' in r["user"] and "three" in r["user"] for r in observed() ) ) read = wait_for( lambda: next( (r["tool"] for r in observed() if "FOUND" in r["tool"]), None ) ) self.assertIn("TASK_SEEN True", read) self.assertRegex(read, r"FOUND [1-9]\d* True True") match = re.search(r"KID=([0-9a-f]+)", spawned) assert match is not None kid = match.group(1) outsider = app.session() app.prompt(outsider, f"peek {kid}").close() app.idle(outsider, timeout=60) delivered = [record["tool"] for record in observed() if record["tool"]] peeked = next( (record for record in delivered if "PEEK_MISSING" in record), None ) assert peeked is not None, ( "PEEK turn finished without its final marker; delivered tools: " + repr(delivered) ) self.assertIn("PEEK scout 1 False", peeked) self.assertIn("PEEK_READ True", peeked) self.assertIn( "PEEK_REFUSED you can only switch, cancel or close", peeked ) self.assertIn("no agent named 'nobody'", peeked) self.assertIn("LISTED True True", peeked) self.assertIn("BY_NAME scout 1 fixture-alpha True", peeked) self.assertIn("ELSEWHERE []", peeked) self.assertIn("SEARCHED False True True", peeked) self.assertIn("UNQUERIED []", peeked) self.assertIn("UNICODE True", peeked) self.assertIn("NOTHING []", peeked) app.idle(lead) time.sleep(1) self.assertFalse( any('kind="unreviewed"' in r["user"] for r in observed()) ) finally: provider.close() @exclusive def test_agent_creation_uses_requested_supported_efforts(self): def script(request): if latest_user(request) == "set agent efforts": return python( "models = await agents.models()\n" "child = await agents.self.spawn(\n" " 'child task', name='high-child', model=models[0], effort=' HIGH '\n" ")\n" "peer = await agents.start(\n" " 'peer task', 'low-peer', models[0], effort=' LOW '\n" ")\n" "default_child = await agents.self.spawn(\n" " 'default task', name='default-child', model=models[0]\n" ")\n" "try:\n" " await agents.self.spawn(\n" " 'bad task', name='bad-child', model=models[0], effort='bogus'\n" " )\n" "except AgentsError as error:\n" " print('EFFORT_REJECTED', error)\n" "try:\n" " await agents.start(\n" " 'bad peer task', 'bad-peer', models[0],\n" " effort='bogus',\n" " )\n" "except AgentsError as error:\n" " print('PEER_EFFORT_REJECTED', error)\n" "try:\n" " await agents.self.spawn(\n" " 'blank task', name='blank-child', model=models[0],\n" " effort=' ',\n" " )\n" "except TypeError as error:\n" " print('BLANK_SPAWN_REJECTED', error)\n" "try:\n" " await agents.start(\n" " 'blank peer task', 'blank-peer', models[0],\n" " effort=' ',\n" " )\n" "except TypeError as error:\n" " print('BLANK_START_REJECTED', error)\n" "assert not await agents.sessions(query='bad-peer')\n" "print('BAD_PEER_NOT_CREATED')\n" "print('EFFORT_CREATED', child.name, peer.name, default_child.name, " "sorted(agent.name for agent in await agents.self.children()))" ) return text("done") provider = Provider(script) profile = f"effort-{provider.route}" providers = { profile: { "extension": "openai", "baseUrl": provider.url, "apiKey": "key", "model": "o3-mini", "protocol": "chat_completions", } } def prepare(app): (app.home / "models.json").write_text( json.dumps( { "openai": { "api": provider.url, "models": { "o3-mini": { "id": "o3-mini", "reasoning_options": [ { "type": "effort", "values": ["low", "medium", "high"], } ], } }, } } ) ) try: with Albedo(provider, providers=providers, prepare=prepare) as app: lead = app.session() app.prompt(lead, "set agent efforts").close() app.idle(lead, timeout=60) def efforts_seen(): return { item["request"].get("reasoning_effort") for item in provider.requests } def efforts_for_task(task): return { item["request"].get("reasoning_effort") for item in provider.requests if any( message.get("role") == "user" and 'kind="task"' in str(message.get("content", "")) and task in str(message.get("content", "")) for message in item["request"]["messages"] ) } # The catalog lists efforts but no default, so the default # child omits the control and the provider decides. wait_for( lambda: ( {"high", "low"} <= efforts_seen() and efforts_for_task("default task") == {None} ), timeout=60, ) def outputs(): return [ str(message.get("content", "")) for item in provider.requests for message in item["request"]["messages"] if message.get("role") == "tool" ] output = wait_for( lambda: next( (value for value in outputs() if "EFFORT_CREATED" in value), None, ), timeout=60, ) self.assertIn( "EFFORT_CREATED high-child low-peer default-child", output ) self.assertIn("['default-child', 'high-child']", output) self.assertEqual(efforts_for_task("default task"), {None}) self.assertIn( "EFFORT_REJECTED unsupported effort bogus; available: low, medium, high", output, ) self.assertIn( "PEER_EFFORT_REJECTED unsupported effort bogus; available: low, medium, high", output, ) self.assertIn( "BLANK_SPAWN_REJECTED effort must be a non-empty str", output ) self.assertIn( "BLANK_START_REJECTED effort must be a non-empty str", output ) self.assertIn("BAD_PEER_NOT_CREATED", output) finally: provider.close() def test_spawn_verifier_gates_parent_mail_and_stopping_until_command_exits_zero( self, ): lead_code = ( "models = await agents.models()\n" "buddy = await agents.self.spawn('wait', name='buddy', model=models[0])\n" "try:\n" " await agents.self.spawn('bad', name='bad', model=models[0], verifier=' ')\n" "except TypeError as error:\n" " print('EMPTY_VERIFIER', error)\n" "worker = await agents.self.spawn(\n" " 'create gate-passed',\n" " name='worker',\n" " model=models[0],\n" " verifier='test -f gate-passed && grep -q ready gate-passed || { echo MISSING_GATE >&2; exit 1; }',\n" ")\n" "print('SPAWNED_VERIFIED', worker.name, buddy.name)\n" ) worker_first = ( "peer_receipt = await mail.submit('buddy', 'hello sibling')\n" "print('PEER_MAIL', peer_receipt.status)\n" "try:\n" " await mail.submit('parent', 'premature done')\n" "except PermissionError as error:\n" " print('GATE_BLOCKED', error)\n" ) worker_fix = ( "q = await mail.request('parent', 'need clarification')\n" "print('QUESTION_MAIL', q.status)\n" "b = await mail.submit('parent', 'unavoidable issue', blocker=True)\n" "print('BLOCKER_MAIL', b.status)\n" "files.write('gate-passed', 'ready\\n')\n" "receipt = await mail.submit('parent', 'verified done')\n" "print('GATE_PASSED', receipt.status)\n" ) def script(request): last = request["messages"][-1] user = latest_user(request) if last.get("role") == "user" and user == "verify worker": return python(lead_code) if ( last.get("role") == "user" and 'kind="task"' in user and "create gate-passed" in user ): return python(worker_first) if last.get("role") == "tool" and "GATE_BLOCKED" in str( last.get("content", "") ): return text("stopping without fixing yet") if last.get("role") == "user" and "MISSING_GATE" in user: return python(worker_fix) if last.get("role") == "user" and 'kind="task"' in user and "wait" in user: return python("await mail.submit('parent', 'buddy done')") return text("ok") provider = Provider(script) try: with Albedo(provider) as app: lead = app.session() app.prompt(lead, "verify worker").close() def observed(): records = [] for item in provider.requests: messages = item["request"]["messages"] records.append( { "user": latest_user(item["request"]), "tool": " ".join( str(m.get("content", "")) for m in messages if m.get("role") == "tool" ), } ) return records try: spawned = wait_for( lambda: next( ( r["tool"] for r in observed() if "SPAWNED_VERIFIED" in r["tool"] ), None, ), ) self.assertIn("EMPTY_VERIFIER", spawned) self.assertIn("SPAWNED_VERIFIED worker buddy", spawned) blocked = wait_for( lambda: next( ( r["tool"] for r in observed() if "GATE_BLOCKED" in r["tool"] ), None, ), ) self.assertIn("PEER_MAIL", blocked) self.assertIn("MISSING_GATE", blocked) passed = wait_for( lambda: next( ( r["tool"] for r in observed() if "GATE_PASSED" in r["tool"] ), None, ), ) self.assertIn("QUESTION_MAIL", passed) self.assertIn("BLOCKER_MAIL", passed) self.assertIn("GATE_PASSED", passed) wait_for( lambda: any( 'from="worker"' in r["user"] and "verified done" in r["user"] for r in observed() ), ) self.assertFalse( any("premature done" in r["user"] for r in observed()) ) except Exception: print("DEBUG_OBSERVED:", json.dumps(observed(), indent=2)) raise finally: provider.close() def test_parent_clears_a_failing_verifier_and_only_the_parent_may(self): lead_code = ( "models = await agents.models()\n" "worker = await agents.self.spawn('finish up', name='worker', model=models[0], verifier='echo NEVER_PASSES >&2; exit 1')\n" "buddy = await agents.self.spawn('wait', name='buddy', model=models[0])\n" "print('SPAWNED_GATED', worker.id, buddy.id)\n" ) clear_code = ( "worker = await agents.get('worker')\n" "try:\n" " await worker.set_verifier(' ')\n" "except TypeError as error:\n" " print('EMPTY_REFUSED', error)\n" "print('CLEARED', await worker.set_verifier(None))\n" "await mail.submit('worker', 'retry now')\n" ) worker_code = ( "try:\n" " await mail.submit('parent', 'too early')\n" "except PermissionError as error:\n" " print('GATE_BLOCKED', error)\n" ) worker_retry = ( "receipt = await mail.submit('parent', 'done after clear')\n" "print('SENT_AFTER_CLEAR', receipt.status)\n" ) buddy_code = ( "peer = await agents.get('worker')\n" "try:\n" " await peer.set_verifier('true')\n" "except AgentsError as error:\n" " print('PEER_REFUSED', error)\n" ) def script(request): last = request["messages"][-1] user = latest_user(request) tool = str(last.get("content", "")) if last.get("role") == "user" and user == "gate worker": return python(lead_code) if last.get("role") == "tool" and "SPAWNED_GATED" in tool: return text("spawned") if last.get("role") == "user" and user == "now clear it": return python(clear_code) if last.get("role") == "user" and 'kind="task"' in user: return python(worker_code if "finish up" in user else buddy_code) if last.get("role") == "tool" and "GATE_BLOCKED" in tool: return text("waiting for my parent") if ( last.get("role") == "user" and "retry now" in user and "cleared your verifier" in json.dumps(request["messages"]) ): return python(worker_retry) return text("ok") provider = Provider(script) try: with Albedo(provider) as app: lead = app.session() app.prompt(lead, "gate worker").close() def tools(): return " ".join( str(m.get("content", "")) for item in provider.requests for m in item["request"]["messages"] if m.get("role") == "tool" ) blocked = wait_for(lambda: "GATE_BLOCKED" in tools() and tools()) self.assertIn("NEVER_PASSES", blocked) refused = wait_for(lambda: "PEER_REFUSED" in tools() and tools()) self.assertIn("only a child's parent can change its verifier", refused) app.idle(lead, timeout=60) spawned = re.search(r"SPAWNED_GATED (\w+)", tools()) assert spawned, tools() worker_id = spawned.group(1) app.idle(worker_id, timeout=60) app.prompt(lead, "now clear it").close() cleared = wait_for(lambda: "SENT_AFTER_CLEAR" in tools() and tools()) self.assertIn("EMPTY_REFUSED", cleared) self.assertIn("CLEARED True", cleared) self.assertNotIn( "too early", " ".join( latest_user(item["request"]) for item in provider.requests ), ) finally: provider.close() def test_resources_attribute_a_sessions_processes_to_its_jobs(self): def script(request): if request["messages"][-1].get("role") == "user": return python(RESOURCES) return text("ok") provider = Provider(script) try: with Albedo(provider) as app: session = app.session() app.prompt(session, "look at your processes").close() app.idle(session, timeout=60) tool = " ".join( str(m.get("content", "")) for item in provider.requests for m in item["request"]["messages"] if m.get("role") == "tool" ) self.assertIn("MINE 1 True True False", tool) self.assertIn("TOTAL True True True", tool) self.assertRegex(tool, r"QUERY True [1-9] True True") self.assertIn("CWD 1 True", tool) self.assertIn("ELSEWHERE 0", tool) self.assertIn("GONE 0", tool) finally: provider.close()