mirror of
https://github.com/0xWheatyz/handler.git
synced 2026-08-31 04:06:23 +00:00
fix(control,ui): close review findings - heartbeat starvation, resume race, stale UI writes
Fixes from the post-migration code review (3 major, 4 minor): - worker: heartbeat between every drained command so a long queue can't starve proof-of-life into a false reap; worker_stale_after default 60s -> 300s (one slow sync/login command must not look like a crash) - repository.create_run: enforces one running run per agent atomically (agent-row FOR UPDATE on Postgres; SQLite's single writer suffices) - two workers claiming resumes for the same agent can no longer both launch claude on one session; resume surfaces the loss loudly - headless._settle: upload the final session archive BEFORE marking the run finished - a resume claimed the instant a run leaves 'running' materializes from session_archives, and the old order let it race an incomplete archive into needless context re-injection (found as a test flake, real in production) - store.tsx: generation token drops in-flight loadRun writes after the user switches runs (run A's events/log/checkmark no longer land on run B), plus id-keyed dedup on event appends from overlapping polls - credsync: credential files written 0600 from the first byte - headless: seq counter locked (reader thread + supervisor both emit events); proc.stdout closed after reader join - login: submit pins to the latest CLAIMED login_start (a still-running one previously pinned to the wrong worker) Suite 296 green (new: create_run conflict coverage); reaper tests track the new staleness default.
This commit is contained in:
@@ -211,6 +211,10 @@ export function DashboardProvider({
|
||||
const [events, setEvents] = useState<AgentEvent[]>([]);
|
||||
const eventsRef = useRef<AgentEvent[]>([]);
|
||||
eventsRef.current = events;
|
||||
/* Bumped on every selectRun: an in-flight loadRun that resolves after the user picked
|
||||
* a different run must drop its writes, or run A's checkmark/log/events land on run
|
||||
* B's panel (the 5s tick keeps polls in flight constantly). */
|
||||
const runGenRef = useRef(0);
|
||||
const [approvals, setApprovals] = useState<Approval[]>([]);
|
||||
const [hosts, setHosts] = useState<Host[]>([]);
|
||||
const [commands, setCommands] = useState<Command[]>([]);
|
||||
@@ -270,12 +274,15 @@ export function DashboardProvider({
|
||||
|
||||
const loadRun = useCallback(async (projectId: string, name: string) => {
|
||||
const path = `/projects/${encodeURIComponent(projectId)}/agents/${encodeURIComponent(name)}`;
|
||||
const gen = runGenRef.current;
|
||||
const stale = () => gen !== runGenRef.current;
|
||||
try {
|
||||
const cm = await clientRef.current.api<Checkmark>(`${path}/checkmark`);
|
||||
if (stale()) return;
|
||||
setCheckmark(cm);
|
||||
setCheckmarkMissing(false);
|
||||
} catch (e) {
|
||||
if (e instanceof AuthError) return;
|
||||
if (e instanceof AuthError || stale()) return;
|
||||
if ((e as ApiError).status === 404) {
|
||||
setCheckmark(null);
|
||||
setCheckmarkMissing(true);
|
||||
@@ -285,9 +292,10 @@ export function DashboardProvider({
|
||||
const entries = await clientRef.current.api<LogEntry[]>(
|
||||
`${path}/log?limit=${LOG_LIMIT}&offset=${logOffsetRef.current}`,
|
||||
);
|
||||
if (stale()) return;
|
||||
setLog(entries);
|
||||
} catch (e) {
|
||||
swallow(e);
|
||||
if (!stale()) swallow(e);
|
||||
}
|
||||
try {
|
||||
// Cursor poll: only events newer than what we already hold come back.
|
||||
@@ -296,9 +304,16 @@ export function DashboardProvider({
|
||||
const fresh = await clientRef.current.api<AgentEvent[]>(
|
||||
`${path}/events?after_id=${after}&limit=500`,
|
||||
);
|
||||
if (fresh.length) setEvents((prev) => [...prev, ...fresh]);
|
||||
if (stale() || fresh.length === 0) return;
|
||||
// Dedup by id at append time: overlapping polls for the same run can both fetch
|
||||
// from the same cursor; only genuinely-new rows may append.
|
||||
setEvents((prev) => {
|
||||
const last = prev.length ? prev[prev.length - 1].id : 0;
|
||||
const add = fresh.filter((e) => e.id > last);
|
||||
return add.length ? [...prev, ...add] : prev;
|
||||
});
|
||||
} catch (e) {
|
||||
swallow(e);
|
||||
if (!stale()) swallow(e);
|
||||
}
|
||||
}, []);
|
||||
|
||||
@@ -419,6 +434,7 @@ export function DashboardProvider({
|
||||
|
||||
const selectRun = useCallback(
|
||||
(projectId: string, name: string) => {
|
||||
runGenRef.current += 1; // invalidate any in-flight loadRun for the previous run
|
||||
setSelectedRun({ projectId, name });
|
||||
setLogOffset(0);
|
||||
logOffsetRef.current = 0;
|
||||
|
||||
@@ -50,7 +50,7 @@ def enqueue_login_submit(body: LoginSubmitIn, conn: Connection = Depends(db_conn
|
||||
nothing to paste into. No prior login_start leaves the pin empty (single-worker
|
||||
deployments behave exactly as before).
|
||||
"""
|
||||
started = repo.get_latest_finished_command(conn, "login_start")
|
||||
started = repo.get_latest_claimed_command(conn, "login_start")
|
||||
return repo.enqueue_command(
|
||||
conn,
|
||||
"login_submit",
|
||||
|
||||
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
@@ -1 +1 @@
|
||||
<!DOCTYPE html><html lang="en"><head><meta charSet="utf-8"/><meta name="viewport" content="width=device-width, initial-scale=1"/><link rel="stylesheet" href="/_next/static/css/bbd6ee1e7265be74.css" data-precedence="next"/><link rel="preload" as="script" fetchPriority="low" href="/_next/static/chunks/webpack-29957205745576aa.js"/><script src="/_next/static/chunks/fd9d1056-e47440ca368281f5.js" async=""></script><script src="/_next/static/chunks/117-d9d1ed00b6332a85.js" async=""></script><script src="/_next/static/chunks/main-app-cb8ae0c3c2bb8b85.js" async=""></script><script src="/_next/static/chunks/app/page-0637b8cf149a88ec.js" async=""></script><title>Handler · Claude Activity</title><meta name="description" content="Monitor and manage Claude Code agents across projects."/><link rel="icon" href="/icon.svg?bab509b45a421e43" type="image/svg+xml" sizes="any"/><script src="/_next/static/chunks/polyfills-42372ed130431b0a.js" noModule=""></script></head><body><div class="gate"><form class="gate-card"><div class="gate-brand"><span class="logo" style="width:26px;height:26px;border-radius:7px"></span>Claude Monitor</div><p class="muted" style="font-size:var(--text-sm);margin:0">Paste your API token to continue. Management actions require the admin token; read-only views work with the plain auth token.</p><input class="input" type="password" autoComplete="current-password" placeholder="API token" autofocus="" value=""/><button class="btn btn-primary" type="submit">Continue</button></form></div><script src="/_next/static/chunks/webpack-29957205745576aa.js" async=""></script><script>(self.__next_f=self.__next_f||[]).push([0]);self.__next_f.push([2,null])</script><script>self.__next_f.push([1,"1:HL[\"/_next/static/css/bbd6ee1e7265be74.css\",\"style\"]\n"])</script><script>self.__next_f.push([1,"2:I[2846,[],\"\"]\n4:I[9107,[],\"ClientPageRoot\"]\n5:I[9859,[\"931\",\"static/chunks/app/page-0637b8cf149a88ec.js\"],\"default\",1]\n6:I[4707,[],\"\"]\n7:I[6423,[],\"\"]\n9:I[1060,[],\"\"]\na:[]\n0:[\"$\",\"$L2\",null,{\"buildId\":\"ga21jhjKhYsutf8F3vo_-\",\"assetPrefix\":\"\",\"urlParts\":[\"\",\"\"],\"initialTree\":[\"\",{\"children\":[\"__PAGE__\",{}]},\"$undefined\",\"$undefined\",true],\"initialSeedData\":[\"\",{\"children\":[\"__PAGE__\",{},[[\"$L3\",[\"$\",\"$L4\",null,{\"props\":{\"params\":{},\"searchParams\":{}},\"Component\":\"$5\"}],null],null],null]},[[[[\"$\",\"link\",\"0\",{\"rel\":\"stylesheet\",\"href\":\"/_next/static/css/bbd6ee1e7265be74.css\",\"precedence\":\"next\",\"crossOrigin\":\"$undefined\"}]],[\"$\",\"html\",null,{\"lang\":\"en\",\"children\":[\"$\",\"body\",null,{\"children\":[\"$\",\"$L6\",null,{\"parallelRouterKey\":\"children\",\"segmentPath\":[\"children\"],\"error\":\"$undefined\",\"errorStyles\":\"$undefined\",\"errorScripts\":\"$undefined\",\"template\":[\"$\",\"$L7\",null,{}],\"templateStyles\":\"$undefined\",\"templateScripts\":\"$undefined\",\"notFound\":[[\"$\",\"title\",null,{\"children\":\"404: This page could not be found.\"}],[\"$\",\"div\",null,{\"style\":{\"fontFamily\":\"system-ui,\\\"Segoe UI\\\",Roboto,Helvetica,Arial,sans-serif,\\\"Apple Color Emoji\\\",\\\"Segoe UI Emoji\\\"\",\"height\":\"100vh\",\"textAlign\":\"center\",\"display\":\"flex\",\"flexDirection\":\"column\",\"alignItems\":\"center\",\"justifyContent\":\"center\"},\"children\":[\"$\",\"div\",null,{\"children\":[[\"$\",\"style\",null,{\"dangerouslySetInnerHTML\":{\"__html\":\"body{color:#000;background:#fff;margin:0}.next-error-h1{border-right:1px solid rgba(0,0,0,.3)}@media (prefers-color-scheme:dark){body{color:#fff;background:#000}.next-error-h1{border-right:1px solid rgba(255,255,255,.3)}}\"}}],[\"$\",\"h1\",null,{\"className\":\"next-error-h1\",\"style\":{\"display\":\"inline-block\",\"margin\":\"0 20px 0 0\",\"padding\":\"0 23px 0 0\",\"fontSize\":24,\"fontWeight\":500,\"verticalAlign\":\"top\",\"lineHeight\":\"49px\"},\"children\":\"404\"}],[\"$\",\"div\",null,{\"style\":{\"display\":\"inline-block\"},\"children\":[\"$\",\"h2\",null,{\"style\":{\"fontSize\":14,\"fontWeight\":400,\"lineHeight\":\"49px\",\"margin\":0},\"children\":\"This page could not be found.\"}]}]]}]}]],\"notFoundStyl"])</script><script>self.__next_f.push([1,"es\":[]}]}]}]],null],null],\"couldBeIntercepted\":false,\"initialHead\":[null,\"$L8\"],\"globalErrorComponent\":\"$9\",\"missingSlots\":\"$Wa\"}]\n"])</script><script>self.__next_f.push([1,"8:[[\"$\",\"meta\",\"0\",{\"name\":\"viewport\",\"content\":\"width=device-width, initial-scale=1\"}],[\"$\",\"meta\",\"1\",{\"charSet\":\"utf-8\"}],[\"$\",\"title\",\"2\",{\"children\":\"Handler · Claude Activity\"}],[\"$\",\"meta\",\"3\",{\"name\":\"description\",\"content\":\"Monitor and manage Claude Code agents across projects.\"}],[\"$\",\"link\",\"4\",{\"rel\":\"icon\",\"href\":\"/icon.svg?bab509b45a421e43\",\"type\":\"image/svg+xml\",\"sizes\":\"any\"}]]\n3:null\n"])</script></body></html>
|
||||
<!DOCTYPE html><html lang="en"><head><meta charSet="utf-8"/><meta name="viewport" content="width=device-width, initial-scale=1"/><link rel="stylesheet" href="/_next/static/css/bbd6ee1e7265be74.css" data-precedence="next"/><link rel="preload" as="script" fetchPriority="low" href="/_next/static/chunks/webpack-29957205745576aa.js"/><script src="/_next/static/chunks/fd9d1056-e47440ca368281f5.js" async=""></script><script src="/_next/static/chunks/117-d9d1ed00b6332a85.js" async=""></script><script src="/_next/static/chunks/main-app-cb8ae0c3c2bb8b85.js" async=""></script><script src="/_next/static/chunks/app/page-2ff318d75a2e3799.js" async=""></script><title>Handler · Claude Activity</title><meta name="description" content="Monitor and manage Claude Code agents across projects."/><link rel="icon" href="/icon.svg?bab509b45a421e43" type="image/svg+xml" sizes="any"/><script src="/_next/static/chunks/polyfills-42372ed130431b0a.js" noModule=""></script></head><body><div class="gate"><form class="gate-card"><div class="gate-brand"><span class="logo" style="width:26px;height:26px;border-radius:7px"></span>Claude Monitor</div><p class="muted" style="font-size:var(--text-sm);margin:0">Paste your API token to continue. Management actions require the admin token; read-only views work with the plain auth token.</p><input class="input" type="password" autoComplete="current-password" placeholder="API token" autofocus="" value=""/><button class="btn btn-primary" type="submit">Continue</button></form></div><script src="/_next/static/chunks/webpack-29957205745576aa.js" async=""></script><script>(self.__next_f=self.__next_f||[]).push([0]);self.__next_f.push([2,null])</script><script>self.__next_f.push([1,"1:HL[\"/_next/static/css/bbd6ee1e7265be74.css\",\"style\"]\n"])</script><script>self.__next_f.push([1,"2:I[2846,[],\"\"]\n4:I[9107,[],\"ClientPageRoot\"]\n5:I[9859,[\"931\",\"static/chunks/app/page-2ff318d75a2e3799.js\"],\"default\",1]\n6:I[4707,[],\"\"]\n7:I[6423,[],\"\"]\n9:I[1060,[],\"\"]\na:[]\n0:[\"$\",\"$L2\",null,{\"buildId\":\"zI_TsbchBb39diXA-dnfw\",\"assetPrefix\":\"\",\"urlParts\":[\"\",\"\"],\"initialTree\":[\"\",{\"children\":[\"__PAGE__\",{}]},\"$undefined\",\"$undefined\",true],\"initialSeedData\":[\"\",{\"children\":[\"__PAGE__\",{},[[\"$L3\",[\"$\",\"$L4\",null,{\"props\":{\"params\":{},\"searchParams\":{}},\"Component\":\"$5\"}],null],null],null]},[[[[\"$\",\"link\",\"0\",{\"rel\":\"stylesheet\",\"href\":\"/_next/static/css/bbd6ee1e7265be74.css\",\"precedence\":\"next\",\"crossOrigin\":\"$undefined\"}]],[\"$\",\"html\",null,{\"lang\":\"en\",\"children\":[\"$\",\"body\",null,{\"children\":[\"$\",\"$L6\",null,{\"parallelRouterKey\":\"children\",\"segmentPath\":[\"children\"],\"error\":\"$undefined\",\"errorStyles\":\"$undefined\",\"errorScripts\":\"$undefined\",\"template\":[\"$\",\"$L7\",null,{}],\"templateStyles\":\"$undefined\",\"templateScripts\":\"$undefined\",\"notFound\":[[\"$\",\"title\",null,{\"children\":\"404: This page could not be found.\"}],[\"$\",\"div\",null,{\"style\":{\"fontFamily\":\"system-ui,\\\"Segoe UI\\\",Roboto,Helvetica,Arial,sans-serif,\\\"Apple Color Emoji\\\",\\\"Segoe UI Emoji\\\"\",\"height\":\"100vh\",\"textAlign\":\"center\",\"display\":\"flex\",\"flexDirection\":\"column\",\"alignItems\":\"center\",\"justifyContent\":\"center\"},\"children\":[\"$\",\"div\",null,{\"children\":[[\"$\",\"style\",null,{\"dangerouslySetInnerHTML\":{\"__html\":\"body{color:#000;background:#fff;margin:0}.next-error-h1{border-right:1px solid rgba(0,0,0,.3)}@media (prefers-color-scheme:dark){body{color:#fff;background:#000}.next-error-h1{border-right:1px solid rgba(255,255,255,.3)}}\"}}],[\"$\",\"h1\",null,{\"className\":\"next-error-h1\",\"style\":{\"display\":\"inline-block\",\"margin\":\"0 20px 0 0\",\"padding\":\"0 23px 0 0\",\"fontSize\":24,\"fontWeight\":500,\"verticalAlign\":\"top\",\"lineHeight\":\"49px\"},\"children\":\"404\"}],[\"$\",\"div\",null,{\"style\":{\"display\":\"inline-block\"},\"children\":[\"$\",\"h2\",null,{\"style\":{\"fontSize\":14,\"fontWeight\":400,\"lineHeight\":\"49px\",\"margin\":0},\"children\":\"This page could not be found.\"}]}]]}]}]],\"notFoundStyl"])</script><script>self.__next_f.push([1,"es\":[]}]}]}]],null],null],\"couldBeIntercepted\":false,\"initialHead\":[null,\"$L8\"],\"globalErrorComponent\":\"$9\",\"missingSlots\":\"$Wa\"}]\n"])</script><script>self.__next_f.push([1,"8:[[\"$\",\"meta\",\"0\",{\"name\":\"viewport\",\"content\":\"width=device-width, initial-scale=1\"}],[\"$\",\"meta\",\"1\",{\"charSet\":\"utf-8\"}],[\"$\",\"title\",\"2\",{\"children\":\"Handler · Claude Activity\"}],[\"$\",\"meta\",\"3\",{\"name\":\"description\",\"content\":\"Monitor and manage Claude Code agents across projects.\"}],[\"$\",\"link\",\"4\",{\"rel\":\"icon\",\"href\":\"/icon.svg?bab509b45a421e43\",\"type\":\"image/svg+xml\",\"sizes\":\"any\"}]]\n3:null\n"])</script></body></html>
|
||||
@@ -1,7 +1,7 @@
|
||||
2:I[9107,[],"ClientPageRoot"]
|
||||
3:I[9859,["931","static/chunks/app/page-0637b8cf149a88ec.js"],"default",1]
|
||||
3:I[9859,["931","static/chunks/app/page-2ff318d75a2e3799.js"],"default",1]
|
||||
4:I[4707,[],""]
|
||||
5:I[6423,[],""]
|
||||
0:["ga21jhjKhYsutf8F3vo_-",[[["",{"children":["__PAGE__",{}]},"$undefined","$undefined",true],["",{"children":["__PAGE__",{},[["$L1",["$","$L2",null,{"props":{"params":{},"searchParams":{}},"Component":"$3"}],null],null],null]},[[[["$","link","0",{"rel":"stylesheet","href":"/_next/static/css/bbd6ee1e7265be74.css","precedence":"next","crossOrigin":"$undefined"}]],["$","html",null,{"lang":"en","children":["$","body",null,{"children":["$","$L4",null,{"parallelRouterKey":"children","segmentPath":["children"],"error":"$undefined","errorStyles":"$undefined","errorScripts":"$undefined","template":["$","$L5",null,{}],"templateStyles":"$undefined","templateScripts":"$undefined","notFound":[["$","title",null,{"children":"404: This page could not be found."}],["$","div",null,{"style":{"fontFamily":"system-ui,\"Segoe UI\",Roboto,Helvetica,Arial,sans-serif,\"Apple Color Emoji\",\"Segoe UI Emoji\"","height":"100vh","textAlign":"center","display":"flex","flexDirection":"column","alignItems":"center","justifyContent":"center"},"children":["$","div",null,{"children":[["$","style",null,{"dangerouslySetInnerHTML":{"__html":"body{color:#000;background:#fff;margin:0}.next-error-h1{border-right:1px solid rgba(0,0,0,.3)}@media (prefers-color-scheme:dark){body{color:#fff;background:#000}.next-error-h1{border-right:1px solid rgba(255,255,255,.3)}}"}}],["$","h1",null,{"className":"next-error-h1","style":{"display":"inline-block","margin":"0 20px 0 0","padding":"0 23px 0 0","fontSize":24,"fontWeight":500,"verticalAlign":"top","lineHeight":"49px"},"children":"404"}],["$","div",null,{"style":{"display":"inline-block"},"children":["$","h2",null,{"style":{"fontSize":14,"fontWeight":400,"lineHeight":"49px","margin":0},"children":"This page could not be found."}]}]]}]}]],"notFoundStyles":[]}]}]}]],null],null],["$L6",null]]]]
|
||||
0:["zI_TsbchBb39diXA-dnfw",[[["",{"children":["__PAGE__",{}]},"$undefined","$undefined",true],["",{"children":["__PAGE__",{},[["$L1",["$","$L2",null,{"props":{"params":{},"searchParams":{}},"Component":"$3"}],null],null],null]},[[[["$","link","0",{"rel":"stylesheet","href":"/_next/static/css/bbd6ee1e7265be74.css","precedence":"next","crossOrigin":"$undefined"}]],["$","html",null,{"lang":"en","children":["$","body",null,{"children":["$","$L4",null,{"parallelRouterKey":"children","segmentPath":["children"],"error":"$undefined","errorStyles":"$undefined","errorScripts":"$undefined","template":["$","$L5",null,{}],"templateStyles":"$undefined","templateScripts":"$undefined","notFound":[["$","title",null,{"children":"404: This page could not be found."}],["$","div",null,{"style":{"fontFamily":"system-ui,\"Segoe UI\",Roboto,Helvetica,Arial,sans-serif,\"Apple Color Emoji\",\"Segoe UI Emoji\"","height":"100vh","textAlign":"center","display":"flex","flexDirection":"column","alignItems":"center","justifyContent":"center"},"children":["$","div",null,{"children":[["$","style",null,{"dangerouslySetInnerHTML":{"__html":"body{color:#000;background:#fff;margin:0}.next-error-h1{border-right:1px solid rgba(0,0,0,.3)}@media (prefers-color-scheme:dark){body{color:#fff;background:#000}.next-error-h1{border-right:1px solid rgba(255,255,255,.3)}}"}}],["$","h1",null,{"className":"next-error-h1","style":{"display":"inline-block","margin":"0 20px 0 0","padding":"0 23px 0 0","fontSize":24,"fontWeight":500,"verticalAlign":"top","lineHeight":"49px"},"children":"404"}],["$","div",null,{"style":{"display":"inline-block"},"children":["$","h2",null,{"style":{"fontSize":14,"fontWeight":400,"lineHeight":"49px","margin":0},"children":"This page could not be found."}]}]]}]}]],"notFoundStyles":[]}]}]}]],null],null],["$L6",null]]]]
|
||||
6:[["$","meta","0",{"name":"viewport","content":"width=device-width, initial-scale=1"}],["$","meta","1",{"charSet":"utf-8"}],["$","title","2",{"children":"Handler · Claude Activity"}],["$","meta","3",{"name":"description","content":"Monitor and manage Claude Code agents across projects."}],["$","link","4",{"rel":"icon","href":"/icon.svg?bab509b45a421e43","type":"image/svg+xml","sizes":"any"}]]
|
||||
1:null
|
||||
|
||||
@@ -58,8 +58,10 @@ class Settings(BaseSettings):
|
||||
# start a run are left queued (for another worker) while all slots are busy.
|
||||
max_concurrent_runs: int = 4
|
||||
# Heartbeats older than this many seconds mark a worker dead; the reaper flips its
|
||||
# running runs (and their agents) to ``crashed``.
|
||||
worker_stale_after: float = 60.0
|
||||
# running runs (and their agents) to ``crashed``. Generous by design: one slow
|
||||
# synchronous command (a large clone, the interactive login flow) can hold a worker
|
||||
# between heartbeats for a minute or more, and a false reap is worse than a slow one.
|
||||
worker_stale_after: float = 300.0
|
||||
# Per-run spend cap passed as ``--max-budget-usd``. 0 disables the flag.
|
||||
run_budget_usd: float = 0.0
|
||||
# Refuse to upload a claude session archive larger than this (a runaway sidecar dir
|
||||
|
||||
@@ -106,10 +106,16 @@ def _merge_claude_json(path: str, incoming: str) -> None:
|
||||
for key in _ACCOUNT_KEYS:
|
||||
if key in new_data:
|
||||
merged[key] = new_data[key]
|
||||
_write_private(path, json.dumps(merged, indent=2))
|
||||
|
||||
|
||||
def _write_private(path: str, content: str) -> None:
|
||||
"""Atomic write with 0600 from the first byte — these files carry OAuth tokens."""
|
||||
os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
|
||||
tmp = f"{path}.credsync.tmp"
|
||||
with open(tmp, "w") as fh:
|
||||
json.dump(merged, fh, indent=2)
|
||||
fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
|
||||
with os.fdopen(fd, "w") as fh:
|
||||
fh.write(content)
|
||||
os.replace(tmp, path)
|
||||
|
||||
|
||||
@@ -151,11 +157,7 @@ def refresh() -> str | None:
|
||||
if rel == ".claude.json":
|
||||
_merge_claude_json(path, content)
|
||||
else:
|
||||
os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
|
||||
tmp = f"{path}.credsync.tmp"
|
||||
with open(tmp, "w") as fh:
|
||||
fh.write(content)
|
||||
os.replace(tmp, path)
|
||||
_write_private(path, content)
|
||||
_state.seen_updated_at = row["updated_at"]
|
||||
_state.last_fingerprint = fingerprint()
|
||||
return "materialized"
|
||||
|
||||
@@ -198,6 +198,7 @@ class RunSupervisor:
|
||||
self.archive_interval = archive_interval
|
||||
self.on_exit = on_exit # worker's slot-release callback
|
||||
self._seq = 0
|
||||
self._seq_lock = threading.Lock() # reader thread + supervisor both emit events
|
||||
self._result_payload: dict | None = None
|
||||
self._canceled = False
|
||||
self.thread: threading.Thread | None = None
|
||||
@@ -211,13 +212,15 @@ class RunSupervisor:
|
||||
# ---------------------------------------------------------------- internals
|
||||
|
||||
def _insert_event(self, etype: str, payload: dict) -> None:
|
||||
self._seq += 1
|
||||
with self._seq_lock:
|
||||
self._seq += 1
|
||||
seq = self._seq
|
||||
with connection() as conn:
|
||||
repo.insert_agent_event(
|
||||
conn,
|
||||
self.agent["id"],
|
||||
self.run["id"],
|
||||
seq=self._seq,
|
||||
seq=seq,
|
||||
type=etype,
|
||||
payload=payload,
|
||||
session_id=self.run["session_id"],
|
||||
@@ -299,13 +302,22 @@ class RunSupervisor:
|
||||
last_archive = now
|
||||
proc.wait()
|
||||
reader.join(timeout=30.0)
|
||||
if proc.stdout is not None:
|
||||
proc.stdout.close()
|
||||
stderr_file.seek(0)
|
||||
stderr_tail = stderr_file.read()[-4000:].decode("utf-8", "replace")
|
||||
stderr_file.close()
|
||||
self._settle(exit_code=proc.returncode, stderr_tail=stderr_tail)
|
||||
|
||||
def _settle(self, exit_code: int | None, stderr_tail: str) -> None:
|
||||
"""Reconcile run + agent status once the process is gone, then final-archive."""
|
||||
"""Final-archive, then reconcile run + agent status once the process is gone.
|
||||
|
||||
Order matters: a resume can be claimed the moment the run leaves ``running``, and
|
||||
it materializes from ``session_archives`` — so the archive upload must land
|
||||
BEFORE the run is marked finished, or a fast resume races an incomplete archive
|
||||
and needlessly degrades to context re-injection.
|
||||
"""
|
||||
self._upload_archive()
|
||||
result = self._result_payload
|
||||
clean = (
|
||||
exit_code == 0
|
||||
@@ -341,7 +353,6 @@ class RunSupervisor:
|
||||
self._insert_event("worker", detail)
|
||||
except Exception: # noqa: BLE001 - never let bookkeeping raise out of the thread
|
||||
pass
|
||||
self._upload_archive()
|
||||
if self.on_exit is not None:
|
||||
try:
|
||||
self.on_exit(self)
|
||||
|
||||
@@ -268,14 +268,20 @@ def resume(agent: dict, answer: str, worker_id: str | None = None) -> tuple[bool
|
||||
elif not transcript.exists():
|
||||
return _resume_reinjected(agent, answer, settings_path, env, worker_id)
|
||||
|
||||
run = headless.launch(
|
||||
agent,
|
||||
kind="resume",
|
||||
prompt=answer,
|
||||
settings_path=settings_path,
|
||||
env=env,
|
||||
worker_id=worker_id,
|
||||
)
|
||||
try:
|
||||
run = headless.launch(
|
||||
agent,
|
||||
kind="resume",
|
||||
prompt=answer,
|
||||
settings_path=settings_path,
|
||||
env=env,
|
||||
worker_id=worker_id,
|
||||
)
|
||||
except repo.RunConflictError:
|
||||
# Another worker won the race for this resume (two queued resume commands, or a
|
||||
# concurrent spawn) — losing loudly here beats two claude processes corrupting
|
||||
# one session transcript.
|
||||
return False, "another worker is already running this agent's session"
|
||||
return True, f"headless resume run {run['id']} started for session {agent['session_id']}"
|
||||
|
||||
|
||||
@@ -302,14 +308,17 @@ def _resume_reinjected(
|
||||
if entry.get("summary"):
|
||||
parts.append(f"Earlier log: {entry['summary']}")
|
||||
parts.append(f"The operator's answer/instruction: {answer}")
|
||||
run = headless.launch(
|
||||
agent,
|
||||
kind="spawn", # a genuinely new session (new UUID) — --resume has nothing to load
|
||||
prompt="\n\n".join(parts),
|
||||
settings_path=settings_path,
|
||||
env=env,
|
||||
worker_id=worker_id,
|
||||
)
|
||||
try:
|
||||
run = headless.launch(
|
||||
agent,
|
||||
kind="spawn", # a genuinely new session (new UUID) — --resume has nothing to load
|
||||
prompt="\n\n".join(parts),
|
||||
settings_path=settings_path,
|
||||
env=env,
|
||||
worker_id=worker_id,
|
||||
)
|
||||
except repo.RunConflictError:
|
||||
return False, "another worker is already running this agent's session"
|
||||
with connection() as conn:
|
||||
repo.insert_agent_event(
|
||||
conn,
|
||||
|
||||
@@ -413,6 +413,12 @@ def drain(worker_id: str, limit: int | None = None) -> int:
|
||||
break
|
||||
_run_one(command)
|
||||
processed += 1
|
||||
# Heartbeat between commands: a long queue must not starve the proof-of-life
|
||||
# (a peer's reaper would otherwise mark this worker's live runs crashed).
|
||||
try:
|
||||
heartbeat(worker_id)
|
||||
except Exception: # noqa: BLE001 - bookkeeping must not stop the drain
|
||||
pass
|
||||
return processed
|
||||
|
||||
|
||||
|
||||
@@ -733,10 +733,35 @@ def delete_worker(conn: Connection, worker_id: str) -> bool:
|
||||
return result.rowcount > 0
|
||||
|
||||
|
||||
class RunConflictError(Exception):
|
||||
"""The agent already has a ``running`` run — a second concurrent claude process on
|
||||
one session would corrupt its transcript."""
|
||||
|
||||
|
||||
def create_run(
|
||||
conn: Connection, agent_id: int, session_id: str, worker_id: str, kind: str
|
||||
) -> dict:
|
||||
"""Open an ``agent_runs`` row for a launching headless invocation."""
|
||||
"""Open an ``agent_runs`` row for a launching headless invocation.
|
||||
|
||||
Enforces **one running run per agent** atomically: two workers that both claimed a
|
||||
resume for the same agent must not both launch. The agent row is locked first on
|
||||
Postgres (``FOR UPDATE``) so concurrent transactions serialize; SQLite's single
|
||||
writer gives the same guarantee for free. Raises :class:`RunConflictError` if a
|
||||
running run already exists.
|
||||
"""
|
||||
lock = select(agents.c.id).where(agents.c.id == agent_id)
|
||||
if conn.dialect.name == "postgresql":
|
||||
lock = lock.with_for_update()
|
||||
conn.execute(lock)
|
||||
existing = conn.execute(
|
||||
select(agent_runs.c.id)
|
||||
.where(agent_runs.c.agent_id == agent_id, agent_runs.c.status == "running")
|
||||
.limit(1)
|
||||
).first()
|
||||
if existing is not None:
|
||||
raise RunConflictError(
|
||||
f"agent {agent_id} already has running run {existing[0]}"
|
||||
)
|
||||
result = conn.execute(
|
||||
agent_runs.insert().values(
|
||||
agent_id=agent_id,
|
||||
@@ -902,12 +927,13 @@ def get_runtime_secret(conn: Connection, key: str) -> dict | None:
|
||||
return _row_to_dict(row)
|
||||
|
||||
|
||||
def get_latest_finished_command(conn: Connection, type: str) -> dict | None:
|
||||
"""The most recent ``done`` command of a type (login pinning reads login_start's
|
||||
``claimed_by`` to route login_submit to the same worker container)."""
|
||||
def get_latest_claimed_command(conn: Connection, type: str) -> dict | None:
|
||||
"""The most recent command of a type that some worker has claimed (running or
|
||||
finished). Login pinning reads login_start's ``claimed_by`` to route login_submit to
|
||||
the same worker container — including while the login_start is still in flight."""
|
||||
row = conn.execute(
|
||||
select(commands)
|
||||
.where(commands.c.type == type, commands.c.status == "done")
|
||||
.where(commands.c.type == type, commands.c.claimed_by.is_not(None))
|
||||
.order_by(commands.c.id.desc())
|
||||
.limit(1)
|
||||
).first()
|
||||
|
||||
@@ -36,7 +36,7 @@ def _hb(worker_id, age_seconds=0.0):
|
||||
|
||||
def test_reaper_marks_stale_workers_runs_crashed(env):
|
||||
agent, run = _seed_run("w-dead")
|
||||
_hb("w-dead", age_seconds=120) # stale: default worker_stale_after is 60s
|
||||
_hb("w-dead", age_seconds=600) # stale: default worker_stale_after is 300s
|
||||
|
||||
assert worker.reap_dead_workers() == 1
|
||||
|
||||
@@ -64,7 +64,7 @@ def test_reaper_preserves_paused_agent_status(env):
|
||||
"""paused_for_input still accurately describes what the agent needs — only agents
|
||||
stuck in 'working' get flipped to crashed."""
|
||||
agent, run = _seed_run("w-dead2", agent_name="paused", agent_status="paused_for_input")
|
||||
_hb("w-dead2", age_seconds=120)
|
||||
_hb("w-dead2", age_seconds=600)
|
||||
|
||||
assert worker.reap_dead_workers() == 1
|
||||
with get_engine().begin() as conn:
|
||||
@@ -74,7 +74,7 @@ def test_reaper_preserves_paused_agent_status(env):
|
||||
|
||||
def test_reaper_idempotent_against_races(env):
|
||||
agent, run = _seed_run("w-dead3", agent_name="raced")
|
||||
_hb("w-dead3", age_seconds=120)
|
||||
_hb("w-dead3", age_seconds=600)
|
||||
assert worker.reap_dead_workers() == 1
|
||||
# A second reaper (or the same one next pass) finds nothing left to settle.
|
||||
assert worker.reap_dead_workers() == 0
|
||||
|
||||
@@ -55,17 +55,29 @@ def test_cancel_request_roundtrip(conn):
|
||||
def test_list_running_runs_scoped_by_worker(conn):
|
||||
agent = _agent(conn)
|
||||
r1 = repo.create_run(conn, agent["id"], "s1", "worker-a", "spawn")
|
||||
r2 = repo.create_run(conn, agent["id"], "s2", "worker-b", "spawn")
|
||||
repo.finish_run(conn, r1["id"], "completed")
|
||||
r2 = repo.create_run(conn, agent["id"], "s2", "worker-b", "spawn")
|
||||
running = repo.list_running_runs(conn)
|
||||
assert [r["id"] for r in running] == [r2["id"]]
|
||||
assert repo.list_running_runs(conn, worker_id="worker-a") == []
|
||||
assert [r["id"] for r in repo.list_running_runs(conn, worker_id="worker-b")] == [r2["id"]]
|
||||
|
||||
|
||||
def test_create_run_refuses_concurrent_run_for_agent(conn):
|
||||
"""One running run per agent, atomically — two workers racing a resume must not both
|
||||
launch a claude process on the same session."""
|
||||
import pytest
|
||||
|
||||
agent = _agent(conn)
|
||||
repo.create_run(conn, agent["id"], "s1", "worker-a", "spawn")
|
||||
with pytest.raises(repo.RunConflictError):
|
||||
repo.create_run(conn, agent["id"], "s1", "worker-b", "resume")
|
||||
|
||||
|
||||
def test_latest_run_and_agent_session(conn):
|
||||
agent = _agent(conn)
|
||||
repo.create_run(conn, agent["id"], "s1", "w", "spawn")
|
||||
first = repo.create_run(conn, agent["id"], "s1", "w", "spawn")
|
||||
repo.finish_run(conn, first["id"], "completed")
|
||||
latest = repo.create_run(conn, agent["id"], "s1", "w", "resume")
|
||||
assert repo.get_latest_run(conn, agent["id"])["id"] == latest["id"]
|
||||
|
||||
|
||||
@@ -22,22 +22,26 @@ def headless_env(env, monkeypatch):
|
||||
config.get_settings.cache_clear()
|
||||
|
||||
|
||||
def _seed(conn_count_running_for=None):
|
||||
def _seed(extra_agents=("b",)):
|
||||
with get_engine().begin() as conn:
|
||||
repo.create_project(conn, "p", "/tmp/p")
|
||||
agent = repo.create_agent(conn, "p", "a", "/tmp/p/a")
|
||||
for name in extra_agents:
|
||||
repo.create_agent(conn, "p", name, f"/tmp/p/{name}")
|
||||
return agent
|
||||
|
||||
|
||||
def _running_run(agent_id, worker_id):
|
||||
def _running_run(agent_name, worker_id):
|
||||
with get_engine().begin() as conn:
|
||||
return repo.create_run(conn, agent_id, f"sid-{worker_id}", worker_id, "spawn")
|
||||
agent = repo.get_agent_by_name(conn, "p", agent_name)
|
||||
return repo.create_run(conn, agent["id"], f"sid-{agent_name}", worker_id, "spawn")
|
||||
|
||||
|
||||
def test_full_worker_skips_run_commands_but_processes_others(headless_env, monkeypatch):
|
||||
agent = _seed()
|
||||
_running_run(agent["id"], "w-full")
|
||||
_running_run(agent["id"], "w-full") # 2 running == MAX_CONCURRENT_RUNS
|
||||
# 2 running (one per agent — one running run per agent) == MAX_CONCURRENT_RUNS
|
||||
_running_run("a", "w-full")
|
||||
_running_run("b", "w-full")
|
||||
|
||||
spawned = {}
|
||||
monkeypatch.setattr(
|
||||
@@ -66,9 +70,9 @@ def test_full_worker_skips_run_commands_but_processes_others(headless_env, monke
|
||||
|
||||
|
||||
def test_slot_frees_when_run_finishes(headless_env, monkeypatch):
|
||||
agent = _seed()
|
||||
run1 = _running_run(agent["id"], "w1")
|
||||
_running_run(agent["id"], "w1")
|
||||
_seed()
|
||||
run1 = _running_run("a", "w1")
|
||||
_running_run("b", "w1")
|
||||
assert worker._full_slot_exclusions("w1") == worker._RUN_COMMANDS
|
||||
|
||||
with get_engine().begin() as conn:
|
||||
|
||||
Reference in New Issue
Block a user