Something went wrong. Try again.
small gleam coding and (not yet) persistent agent daemon with a detachable cli forked from okami.mom/albedo
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650"""Python agent API: spawn, family identity, refusals, progress, explicit mail,agents.get, reading messages across sessions, and listing and searching them."""
import jsonimport reimport timeimport 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', modelsassert all('/' in model for model in models), modelsassert 'beta/fixture-beta' in models, modelstry: 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.selfprint('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()