Gabriel Vidal
36fccc17ee
A guest spawn runs in exactly one project they hold chat on, from its directory, on the default account, the host sidecar and the claude harness, with an allowed model and under their monthly budget, whatever the composer sent. It carries a scope (dontAsk + edits only under the project and /tmp, credential reads denied, a system prompt naming the guest and their level, no remote control) and pings the owner. In a shared conversation every turn is prefixed with an <ai-agent-from/> tag the parser turns into the item's author. Tests: test_guests.py (19). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
220 lines
11 KiB
Python
220 lines
11 KiB
Python
"""main's run routing across runners: a new run picks the account's online
|
|
worker (or the composer's pick), a continue follows the session to its worker,
|
|
the terminal-live guard, interrupt routing and per-runner exit baselines.
|
|
|
|
Imports ``main`` against throwaway paths (no startup event → no threads)."""
|
|
|
|
import os
|
|
import pathlib
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
|
|
TMP = pathlib.Path(tempfile.mkdtemp())
|
|
for k, v in {"DB_PATH": TMP / "x.db", "TRANSCRIPTS_DIR": TMP / "t",
|
|
"WORKSPACE": TMP, "REPO_DIR": TMP, "ARCHIVE_DIR": TMP / "arch",
|
|
"META_PATH": TMP / "meta.json", "WORKERS_PATH": TMP / "w.json",
|
|
"WORKERS_TRANSCRIPTS_DIR": TMP / "mirror",
|
|
"CRON_PATH": TMP / "cron.json", "SIDECAR_TOKEN": "",
|
|
"PI_TRANSCRIPTS_DIR": ""}.items():
|
|
os.environ[k] = str(v)
|
|
|
|
from fastapi import HTTPException # noqa: E402
|
|
|
|
import accounts # noqa: E402
|
|
import main # noqa: E402
|
|
import workers # noqa: E402
|
|
from core.state import _record_worker_files # noqa: E402
|
|
from cron.routes import CronCreateBody, CronPatchBody, cron_create, cron_delete, cron_update # noqa: E402
|
|
from cron.service import _fire_cron_job # noqa: E402
|
|
from runs.routes import InterruptBody, interrupt_conversation # noqa: E402
|
|
from runs.service import _send_message # noqa: E402
|
|
from runs.sidecar_client import _sidecar_live_sids # noqa: E402
|
|
from users.caller import OWNER # noqa: E402
|
|
|
|
DEPS = main.app.state.ai
|
|
from _fakeworker import FakeWorker # noqa: E402
|
|
|
|
SID = "11111111-2222-3333-4444-555555555555"
|
|
|
|
|
|
class Routing(unittest.TestCase):
|
|
def setUp(self):
|
|
self.fw = FakeWorker(TMP / "remote", TMP / "state")
|
|
self.w = workers.pair(DEPS.worker_store, self.fw.pairing_string("work"),
|
|
account=None, name="mac", hub_name="t",
|
|
resolve_login=lambda _l: None,
|
|
valid_account=accounts.valid,
|
|
auto_route=True)
|
|
|
|
def tearDown(self):
|
|
self.fw.stop()
|
|
DEPS.worker_store.delete(self.w["id"])
|
|
|
|
def test_new_run_on_the_account_goes_to_its_worker(self):
|
|
out = _send_message(DEPS, "hi", projects=["orus-monorepo"])
|
|
path, body = self.fw.calls[-1]
|
|
self.assertEqual(path, "/spawn")
|
|
self.assertNotIn("account", body) # the worker's own login
|
|
self.assertIsNone(body["cwd"]) # the worker's own default
|
|
m = DEPS.meta_store.get(out["sessionId"])
|
|
self.assertEqual((m["worker"], m["account"]), (self.w["id"], "work"))
|
|
|
|
def test_without_auto_route_the_lab_keeps_the_account(self):
|
|
DEPS.worker_store.patch(self.w["id"], {"autoRoute": False})
|
|
# auto: the worker is online but opt-out ⇒ the host sidecar
|
|
# (unconfigured here ⇒ 503), never the worker
|
|
with self.assertRaises(HTTPException) as cm:
|
|
_send_message(DEPS, "hi", projects=["orus-monorepo"])
|
|
self.assertEqual(cm.exception.status_code, 503)
|
|
self.assertFalse([c for c in self.fw.calls if c[0] == "/spawn"])
|
|
# an explicit pick still lands there
|
|
out = _send_message(DEPS, "hi", worker=self.w["id"])
|
|
self.assertEqual(self.fw.calls[-1][0], "/spawn")
|
|
self.assertEqual(DEPS.meta_store.get(out["sessionId"])["worker"],
|
|
self.w["id"])
|
|
|
|
def test_local_pick_and_offline_fallback(self):
|
|
DEPS.worker_store.set_status(self.w["id"], status="offline")
|
|
# auto: offline worker ⇒ the host sidecar (unconfigured here ⇒ 503)
|
|
with self.assertRaises(HTTPException) as cm:
|
|
_send_message(DEPS, "hi", projects=["orus-monorepo"])
|
|
self.assertEqual(cm.exception.status_code, 503)
|
|
# an explicit pick of an offline worker is refused, not rerouted
|
|
with self.assertRaises(HTTPException) as cm:
|
|
_send_message(DEPS, "hi", worker=self.w["id"])
|
|
self.assertEqual(cm.exception.detail["kind"], "worker_offline")
|
|
self.assertEqual(self.fw.calls, [])
|
|
|
|
def test_explicit_worker_bills_its_account(self):
|
|
out = _send_message(DEPS, "hi", account="personal", worker=self.w["id"])
|
|
self.assertEqual(DEPS.meta_store.get(out["sessionId"])["account"],
|
|
"work")
|
|
|
|
def test_pi_never_goes_to_a_worker(self):
|
|
with self.assertRaises(HTTPException) as cm:
|
|
_send_message(DEPS, "hi", harness="pi", worker=self.w["id"])
|
|
self.assertEqual(cm.exception.detail["kind"], "worker_harness")
|
|
|
|
def test_continue_follows_the_session_and_guards_a_live_terminal(self):
|
|
DEPS.meta_store.update(SID, {"workerSource": self.w["id"],
|
|
"accountSource": "work"})
|
|
mirror = TMP / "mirror" / "-Users-g-repo" / f"{SID}.jsonl"
|
|
mirror.parent.mkdir(parents=True, exist_ok=True)
|
|
mirror.write_text("{}\n") # grew just now
|
|
with self.assertRaises(HTTPException) as cm:
|
|
_send_message(DEPS, "more", session_id=SID, cwd="/Users/g/repo")
|
|
self.assertEqual(cm.exception.detail["kind"], "terminal_live")
|
|
self.assertEqual(self.fw.calls, [])
|
|
# confirmed takeover goes through, to the worker, force-resume
|
|
_send_message(DEPS, "more", session_id=SID, cwd="/Users/g/repo",
|
|
takeover=True)
|
|
path, body = self.fw.calls[-1]
|
|
self.assertEqual((path, body["force"], body["cwd"]),
|
|
("/resume", True, "/Users/g/repo"))
|
|
self.assertEqual(DEPS.meta_store.get(SID)["worker"], self.w["id"])
|
|
# an old transcript isn't guarded
|
|
os.utime(mirror, (time.time() - 3600, time.time() - 3600))
|
|
_send_message(DEPS, "again", session_id=SID, cwd="/Users/g/repo")
|
|
# …nor is our own live run on the worker
|
|
mirror.write_text("{}\n{}\n")
|
|
self.fw.live = {SID}
|
|
_send_message(DEPS, "again", session_id=SID, cwd="/Users/g/repo")
|
|
self.assertEqual([c[0] for c in self.fw.calls],
|
|
["/resume", "/resume", "/resume"])
|
|
|
|
def test_growth_from_our_own_finished_run_is_not_a_terminal(self):
|
|
sid = "22222222-2222-3333-4444-555555555555"
|
|
mirror = TMP / "mirror" / "-Users-g-repo" / f"{sid}.jsonl"
|
|
mirror.parent.mkdir(parents=True, exist_ok=True)
|
|
mirror.write_text("{}\n")
|
|
now = time.time()
|
|
os.utime(mirror, (now - 10, now - 10))
|
|
stamp = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(now - 8))
|
|
DEPS.meta_store.update(sid, {"worker": self.w["id"], "state": "finished",
|
|
"exitedAt": stamp})
|
|
_send_message(DEPS, "next", session_id=sid, cwd="/Users/g/repo")
|
|
self.assertEqual(self.fw.calls[-1][0], "/resume")
|
|
# …but a write after our run exited is a terminal's
|
|
DEPS.meta_store.update(sid, {"state": "finished"})
|
|
os.utime(mirror, (now, now))
|
|
with self.assertRaises(HTTPException) as cm:
|
|
_send_message(DEPS, "next", session_id=sid, cwd="/Users/g/repo")
|
|
self.assertEqual(cm.exception.detail["kind"], "terminal_live")
|
|
|
|
def test_continue_into_a_live_run_goes_through_its_inbox(self):
|
|
DEPS.meta_store.update(SID, {"worker": self.w["id"], "state": "running"})
|
|
self.fw.live = {SID}
|
|
self.fw.inbox = {SID}
|
|
out = _send_message(DEPS, "Up?", session_id=SID, cwd="/Users/g/repo")
|
|
self.assertEqual(out["delivered"], "inbox")
|
|
self.assertEqual([c[0] for c in self.fw.calls], ["/message"])
|
|
self.assertEqual(self.fw.calls[-1][1]["prompt"], "Up?")
|
|
self.assertEqual(DEPS.meta_store.get(SID)["state"], "running")
|
|
# the live run can't take it (409) ⇒ the force-resume as before
|
|
self.fw.inbox = set()
|
|
_send_message(DEPS, "Up?", session_id=SID, cwd="/Users/g/repo")
|
|
path, body = self.fw.calls[-1]
|
|
self.assertEqual((path, body["force"]), ("/resume", True))
|
|
# nothing live (404) ⇒ a plain resume, still force (harmless)
|
|
self.fw.live = set()
|
|
_send_message(DEPS, "Up?", session_id=SID, cwd="/Users/g/repo")
|
|
self.assertEqual(self.fw.calls[-1][0], "/resume")
|
|
|
|
def test_interrupt_routes_to_the_worker(self):
|
|
DEPS.meta_store.update(SID, {"worker": self.w["id"]})
|
|
interrupt_conversation(DEPS, OWNER, InterruptBody(sessionId=SID))
|
|
self.assertEqual(self.fw.calls[-1][0], "/interrupt")
|
|
|
|
def test_mirror_stamps_worker_and_account(self):
|
|
sid = "99999999-2222-3333-4444-555555555555"
|
|
_record_worker_files(DEPS, self.w, [f"-Users-g-repo/{sid}.jsonl",
|
|
f"-Users-g-repo/{sid}/subagents/a.jsonl"])
|
|
m = DEPS.meta_store.get(sid)
|
|
self.assertEqual((m["workerSource"], m["accountSource"]),
|
|
(self.w["id"], "work"))
|
|
self.assertEqual(accounts.resolve({}, m), "work")
|
|
|
|
def test_cron_job_runs_on_its_worker_and_never_auto_routes(self):
|
|
agents = TMP / ".claude" / "agents"
|
|
agents.mkdir(parents=True, exist_ok=True)
|
|
# account: work would auto-route to the Mac — an unpinned job stays home
|
|
(agents / "pr.md").write_text("---\naccount: work\n---\nwatch PRs\n")
|
|
job = cron_create(DEPS, CronCreateBody(
|
|
name="pr", schedule="0 9 * * *",
|
|
promptFile=".claude/agents/pr.md"))
|
|
self.assertIsNone(job["worker"])
|
|
with self.assertRaises(HTTPException) as cm: # lab sidecar: 503 here
|
|
_fire_cron_job(DEPS, DEPS.cron_store.get("pr"))
|
|
self.assertEqual(cm.exception.status_code, 503)
|
|
self.assertEqual(self.fw.calls, [])
|
|
# pinned to the worker: the prompt text goes there
|
|
job = cron_update(DEPS, "pr", CronPatchBody(worker=self.w["id"]))
|
|
self.assertEqual(job["worker"], self.w["id"])
|
|
out = _fire_cron_job(DEPS, DEPS.cron_store.get("pr"))
|
|
path, body = self.fw.calls[-1]
|
|
self.assertEqual((path, body["prompt"]), ("/spawn", "watch PRs"))
|
|
m = DEPS.meta_store.get(out["sessionId"])
|
|
self.assertEqual((m["worker"], m["cron"]["id"]), (self.w["id"], "pr"))
|
|
# offline ⇒ refused, not rerouted to the lab
|
|
DEPS.worker_store.set_status(self.w["id"], status="offline")
|
|
with self.assertRaises(HTTPException) as cm:
|
|
_fire_cron_job(DEPS, DEPS.cron_store.get("pr"))
|
|
self.assertEqual(cm.exception.detail["kind"], "worker_offline")
|
|
# unknown ids are rejected up front; "" moves it back to the lab
|
|
with self.assertRaises(HTTPException):
|
|
cron_update(DEPS, "pr", CronPatchBody(worker="nope"))
|
|
job = cron_update(DEPS, "pr", CronPatchBody(worker=""))
|
|
self.assertIsNone(job["worker"])
|
|
cron_delete(DEPS, "pr")
|
|
|
|
def test_live_sids_per_runner(self):
|
|
self.fw.live = {"a", "b"}
|
|
self.assertEqual(_sidecar_live_sids(DEPS, self.w["id"]), {"a", "b"})
|
|
DEPS.worker_store.set_status(self.w["id"], status="offline")
|
|
self.assertIsNone(_sidecar_live_sids(DEPS, self.w["id"]))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|