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>
289 lines
14 KiB
Python
289 lines
14 KiB
Python
"""The one send path behind both composers: spawn, resume, runner and account routing."""
|
|
|
|
import datetime
|
|
import os
|
|
import pathlib
|
|
import time
|
|
import uuid as uuidlib
|
|
|
|
from fastapi import HTTPException
|
|
|
|
import accounts as accounts_mod
|
|
from conversations.catalog import _session_conv_map
|
|
from core.config import TRANSCRIPTS_DIR, WORKERS_SOURCE_DIR
|
|
from core.state import AppState
|
|
from runs.sidecar_client import (
|
|
_sidecar_busy,
|
|
_sidecar_call,
|
|
_sidecar_live_sids,
|
|
_valid_sid,
|
|
)
|
|
|
|
# A worker session whose transcript grew this recently with no worker-launched
|
|
# run behind it is probably still open in a terminal on that machine.
|
|
TERMINAL_LIVE_S = float(os.environ.get("TERMINAL_LIVE_S", "60"))
|
|
|
|
|
|
def _session_worker(deps: AppState, sid: str) -> str | None:
|
|
"""The worker a session lives on: the one the hub launched it on
|
|
(``meta.worker``), else the one the mirror imported it from
|
|
(``meta.workerSource``). None = the host sidecar's."""
|
|
m = deps.meta_store.peek(sid)
|
|
return m.get("worker") or m.get("workerSource")
|
|
|
|
|
|
def _mirror_transcript(sid: str) -> pathlib.Path | None:
|
|
if not WORKERS_SOURCE_DIR or not WORKERS_SOURCE_DIR.is_dir():
|
|
return None
|
|
return next(WORKERS_SOURCE_DIR.glob(f"*/{sid}.jsonl"), None)
|
|
|
|
|
|
def _terminal_live_guard(deps: AppState, sid: str, worker: str) -> None:
|
|
"""Refuse (409 ``terminal_live``) to resume a worker session a terminal on
|
|
that machine may still hold: two writers on one transcript interleave its
|
|
parent chain. The worker can't see the terminal's process, so it is a
|
|
heuristic — the mirror copy grew in the last TERMINAL_LIVE_S and the worker
|
|
has no run of its own for the session. The UI turns it into a confirm."""
|
|
ap = _mirror_transcript(sid)
|
|
if ap is None:
|
|
return
|
|
try:
|
|
age = time.time() - ap.stat().st_mtime
|
|
except OSError:
|
|
return
|
|
if age > TERMINAL_LIVE_S:
|
|
return
|
|
live = _sidecar_live_sids(deps, worker) or set()
|
|
if sid in live:
|
|
return # our own run — a normal force-resume
|
|
# Growth our own last run explains: it wrote the file, then the exit
|
|
# watcher stamped exitedAt (or hasn't ticked yet and it still reads
|
|
# running). Only writes *after* that are a terminal's.
|
|
m = deps.meta_store.peek(sid)
|
|
if m.get("worker") and m.get("state") == "running":
|
|
return
|
|
exited = m.get("exitedAt")
|
|
if exited:
|
|
try:
|
|
ex = datetime.datetime.strptime(exited, "%Y-%m-%dT%H:%M:%SZ") \
|
|
.replace(tzinfo=datetime.timezone.utc).timestamp()
|
|
if ap.stat().st_mtime <= ex + 5:
|
|
return
|
|
except (ValueError, OSError):
|
|
pass
|
|
w = deps.worker_store.get(worker) or {}
|
|
raise HTTPException(409, {
|
|
"message": f"this session was active in a terminal on "
|
|
f"{w.get('name') or 'the worker'} {int(age)}s ago",
|
|
"kind": "terminal_live", "ageSeconds": int(age),
|
|
"hint": "close it there first, or continue anyway — two writers on "
|
|
"one session garble its history"})
|
|
|
|
|
|
# The explicit "run it on the lab's own sidecar" runner pick.
|
|
LOCAL_RUNNER = "local"
|
|
|
|
|
|
def _pick_worker(deps: AppState, choice: str | None, account: str, harness: str) -> str | None:
|
|
"""Which runner a *new* run goes to: an explicit worker id (must be
|
|
online — a run silently landing elsewhere is the worse surprise), "local"
|
|
for the host sidecar, unset ⇒ the account's online worker, else local.
|
|
Workers run claude only (no pi runner there)."""
|
|
c = (choice or "").strip()
|
|
if c == LOCAL_RUNNER:
|
|
return None
|
|
if c:
|
|
w = deps.worker_store.get(c)
|
|
if not w:
|
|
raise HTTPException(400, {"message": f"unknown worker: {c!r}",
|
|
"kind": "worker_unknown"})
|
|
if harness != "claude":
|
|
raise HTTPException(400, {"message": "workers run claude only",
|
|
"kind": "worker_harness"})
|
|
if not deps.worker_store.online(c):
|
|
raise HTTPException(409, {
|
|
"message": f"worker {w.get('name') or c} is offline",
|
|
"kind": "worker_offline",
|
|
"hint": "wake the machine, or send it to the lab instead"})
|
|
return c
|
|
if harness != "claude":
|
|
return None
|
|
w = deps.worker_store.for_account(account)
|
|
return w["id"] if w else None
|
|
|
|
|
|
def _session_account(deps: AppState, sid: str, cwd: str | None = None) -> str:
|
|
"""The account an existing session runs on — the one a resume/fork must
|
|
launch on, since its transcript lives only in that account's ``projects/``.
|
|
|
|
Same order as ``accounts.resolve``, fed from the metadata stamp and the
|
|
stored summary's cwd (falling back to the cwd the caller sent, for a
|
|
session whose transcript hasn't been parsed yet)."""
|
|
conv = _session_conv_map(deps, ).get(sid)
|
|
summary = deps.store.get_summary(str(TRANSCRIPTS_DIR / conv)) if conv else None
|
|
return accounts_mod.resolve(summary or {"cwd": cwd}, deps.meta_store.peek(sid))
|
|
|
|
|
|
def _send_message(deps: AppState, prompt: str, *, session_id: str | None = None,
|
|
model: str | None = None, cwd: str | None = None,
|
|
harness: str | None = None,
|
|
thinking: bool | None = None,
|
|
effort: str | None = None,
|
|
account: str | None = None,
|
|
projects: list[str] | None = None,
|
|
worker: str | None = None,
|
|
takeover: bool = False,
|
|
scope: dict | None = None,
|
|
stamp: dict | None = None) -> dict:
|
|
"""The one send path behind both composers.
|
|
|
|
With no ``session_id`` this starts a new conversation: we mint the UUID here
|
|
so the frontend can start watching for the transcript to sync (~2s) and
|
|
redirect to it without waiting for the run. With one, it continues that
|
|
session — and a continue *always* takes the prompt: if a run is still live
|
|
(or parked on a ScheduleWakeup/Monitor), the sidecar stops it first (SIGINT
|
|
the group, wait for it to die) and then launches the ``--resume`` with the
|
|
new turn, so sending into a "running"/"paused" conversation restarts it on
|
|
the new message. Either way the actual agent process (`claude -p` or the pi
|
|
runner, per ``harness``) is launched by the host sidecar (the container
|
|
can't reach the host's authenticated CLIs), and we stamp ``running``
|
|
immediately so the UI flips before the first turn has synced.
|
|
|
|
**Account.** A new claude run bills ``account`` when given (the composer's
|
|
chip), else the account owning one of ``projects`` (orus-monorepo → work),
|
|
else the default one; a non-default account also starts in its own repo
|
|
unless ``cwd`` says otherwise. A continue always keeps the account the
|
|
session started on — whatever the caller asks. The choice is stamped as
|
|
``meta.account`` next to ``harness``. pi runs aren't a Claude login: they
|
|
bill the default account.
|
|
|
|
**Worker.** A new run goes to ``worker`` when the composer picked one
|
|
("local" = the host sidecar), else to the account's online worker, else
|
|
the host (``_pick_worker``); a worker run bills the worker's own login.
|
|
A continue goes where the session lives (``_session_worker``) — never
|
|
elsewhere, its transcript isn't anywhere else — and a session a terminal
|
|
may still hold needs ``takeover`` (``_terminal_live_guard``). Stamped as
|
|
``meta.worker``.
|
|
|
|
**Scope.** ``scope`` (a guest's run — users/runs.py) goes to the sidecar,
|
|
which launches the run with those permissions; ``stamp`` is merged into
|
|
a new conversation's metadata (``owner``, ``projects``). A follow-up into a
|
|
live run (the inbox) keeps whatever scope that run was launched with."""
|
|
prompt = (prompt or "").strip()
|
|
if not prompt:
|
|
raise HTTPException(400, "empty prompt")
|
|
if session_id:
|
|
sid = _valid_sid(session_id)
|
|
# A conversation resumes on the harness it started on — and with the
|
|
# thinking setting it last ran with — unless the caller overrides. Both
|
|
# were stamped into the metadata sidecar by the spawn.
|
|
prev = deps.meta_store.get(sid)
|
|
harness = (harness or prev.get("harness") or "claude").lower()
|
|
if thinking is None:
|
|
thinking = prev.get("thinking")
|
|
if effort is None:
|
|
effort = prev.get("effort")
|
|
account = (_session_account(deps, sid, cwd) if harness == "claude"
|
|
else accounts_mod.DEFAULT)
|
|
run_on = _session_worker(deps, sid) if harness == "claude" else None
|
|
if run_on:
|
|
w = deps.worker_store.get(run_on)
|
|
if w and not deps.worker_store.online(run_on):
|
|
raise HTTPException(409, {
|
|
"message": f"this session lives on {w.get('name') or run_on}"
|
|
", which is offline",
|
|
"kind": "worker_offline",
|
|
"hint": "wake the machine — a session can't move runners"})
|
|
if not takeover:
|
|
_terminal_live_guard(deps, sid, run_on)
|
|
if not cwd: # --resume is cwd-scoped: the session's own, on the Mac
|
|
conv = _session_conv_map(deps, ).get(sid)
|
|
summ = deps.store.get_summary(str(TRANSCRIPTS_DIR / conv)) \
|
|
if conv else None
|
|
cwd = (summ or {}).get("cwd")
|
|
# A run that is still alive takes the follow-up through its inbox
|
|
# socket (sidecar /message): its background work — a build, a
|
|
# subagent, a Monitor on a form — survives, and the model reads the
|
|
# message between tool calls, or at once when it is idle-waiting.
|
|
# Only when nothing is live (404: a plain --resume) or the live run
|
|
# can't take it (409 inbox_unavailable: a pi run, an older CLI) does
|
|
# the message restart the session the old way, below. A model /
|
|
# thinking / effort switch doesn't apply through the inbox — the run
|
|
# keeps what it was launched with; Stop it first to switch.
|
|
if harness == "claude":
|
|
try:
|
|
out = _sidecar_call(deps, "/message", {"sessionId": sid,
|
|
"prompt": prompt},
|
|
worker=run_on)
|
|
except HTTPException as e:
|
|
if e.status_code not in (404, 409):
|
|
raise
|
|
else:
|
|
deps.meta_store.update(sid, {"state": "running"})
|
|
deps.hub.publish({"type": "meta", "id": sid})
|
|
return {"sessionId": sid, "pid": out.get("pid"),
|
|
"delivered": out.get("delivered") or "inbox"}
|
|
# A force-resume swaps the session's process; shield the swap from the
|
|
# exit watcher so the old run's vanish isn't stamped "finished" over
|
|
# the "running" written below.
|
|
_sidecar_busy.add(sid)
|
|
try:
|
|
out = _sidecar_call(deps, "/resume", {"sessionId": sid, "prompt": prompt,
|
|
"model": model, "cwd": cwd,
|
|
"harness": harness,
|
|
"thinking": thinking,
|
|
"effort": effort,
|
|
"account": account,
|
|
"scope": scope,
|
|
"force": True}, worker=run_on)
|
|
if out.get("reused"):
|
|
# force=True makes the sidecar stop a live run instead of
|
|
# answering `reused`; only a sidecar predating the flag still
|
|
# can. The turn was dropped either way — a 409, not a success,
|
|
# so the composer says so instead of waiting on an answer that
|
|
# isn't coming.
|
|
deps.meta_store.update(sid, {"state": "running"})
|
|
deps.hub.publish({"type": "meta", "id": sid})
|
|
raise HTTPException(409, "this session is still running and "
|
|
"the sidecar is too old to stop it — "
|
|
"restart the claude-sidecar service")
|
|
# Stamp running (and re-stamp harness + thinking) before lifting the
|
|
# shield — the UI flips before the transcript syncs. A None thinking
|
|
# is dropped by the merge, so the session keeps what it had.
|
|
deps.meta_store.update(sid, {"state": "running", "harness": harness,
|
|
"thinking": thinking, "effort": effort,
|
|
"account": account, "worker": run_on})
|
|
finally:
|
|
_sidecar_busy.discard(sid)
|
|
else:
|
|
sid = str(uuidlib.uuid4())
|
|
harness = (harness or "claude").lower()
|
|
if account and not accounts_mod.valid(account):
|
|
raise HTTPException(400, f"unknown account: {account!r}")
|
|
if harness == "claude":
|
|
account = (accounts_mod.valid(account)
|
|
or accounts_mod.account_for_projects(projects)
|
|
or accounts_mod.DEFAULT)
|
|
else:
|
|
account = accounts_mod.DEFAULT
|
|
run_on = _pick_worker(deps, worker, account, harness)
|
|
if run_on:
|
|
# The worker's own login bills it; its own default cwd (the repo
|
|
# its installer was pointed at) applies unless the caller chose.
|
|
account = (deps.worker_store.get(run_on) or {}).get("account") or account
|
|
elif harness == "claude":
|
|
cwd = cwd or accounts_mod.sidecar_payload(account)["cwd"]
|
|
out = _sidecar_call(deps, "/spawn", {"prompt": prompt, "sessionId": sid,
|
|
"model": model, "harness": harness,
|
|
"thinking": thinking, "effort": effort,
|
|
"account": account, "cwd": cwd,
|
|
"scope": scope},
|
|
worker=run_on)
|
|
# A fresh id can't be in the exit watcher's baseline — no shield needed.
|
|
deps.meta_store.update(sid, {"state": "running", "harness": harness,
|
|
"thinking": thinking, "effort": effort,
|
|
"account": account, "worker": run_on,
|
|
**(stamp or {})})
|
|
deps.hub.publish({"type": "meta", "id": sid})
|
|
return {"sessionId": sid, "pid": out.get("pid")}
|