"""Python agent API: spawn, family identity, refusals, progress, explicit mail,
agents.get, reading messages across sessions, and listing and searching them."""
import re
import time
import unittest
from harness import Albedo, Provider, python, text
LEAD = """models = await agents.models()
assert 'beta/fixture-beta' in models, models
kid = await agents.self.spawn('count to three', name='scout', model=models[0])
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 = """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 agents.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 agents.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 agents.sessions(cwd='/nowhere'))
talked = await agents.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 agents.sessions('CAFÉ ÜNÏCODE')])
print('NOTHING', await agents.sessions('never ' + 'said anywhere'))
try:
await agents.get('nobody')
except AgentsError as error:
print('PEEK_MISSING', error)
"""
def user_text(request):
return next(
(
item.get("content", "")
for item in reversed(request["messages"])
if item.get("role") == "user"
),
"",
)
def wait_for(predicate):
deadline = time.monotonic() + 60
while time.monotonic() < deadline:
result = predicate()
if result:
return result
time.sleep(0.1)
raise AssertionError("agent Python turn never arrived")
class AgentsPythonTests(unittest.TestCase):
def test_python_spawn_identity_refusals_and_explicit_answer(self):
def script(request):
last = request["messages"][-1]
user = user_text(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 "spawn a scout" in user:
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()
app.prompt(lead, "spawn a scout").close()
def observed():
records = []
for item in provider.requests:
messages = item["request"]["messages"]
last = messages[-1]
records.append(
{
"user": user_text(item["request"]),
"context": " ".join(
str(m.get("content", "")) for m in messages
),
"tool": str(last.get("content", ""))
if last.get("role") == "tool"
else "",
}
)
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("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 spawn a scout" in child
or "SENT queued spawn a scout" 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()
peeked = wait_for(
lambda: next(
(r["tool"] for r in observed() if "PEEK_MISSING" in r["tool"]),
None,
)
)
self.assertIn("PEEK scout 1 False", peeked)
self.assertIn("PEEK_READ True", peeked)
self.assertIn("PEEK_REFUSED you can only 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()