diff --git a/app/src/main/db/ddl.ts b/app/src/main/db/ddl.ts index 91a711f..76f9e16 100644 --- a/app/src/main/db/ddl.ts +++ b/app/src/main/db/ddl.ts @@ -87,7 +87,11 @@ export const SCHEMA_DDL = ` source_id TEXT, prompt TEXT NOT NULL, background INTEGER NOT NULL, - fired_at INTEGER NOT NULL + fired_at INTEGER NOT NULL, + outcome TEXT, + finished_at INTEGER, + failure_reason TEXT, + failure_category TEXT ); CREATE INDEX IF NOT EXISTS wakes_agent_fired ON wakes (agent_id, fired_at); CREATE TABLE IF NOT EXISTS recent_notifications ( diff --git a/app/src/main/db/migrate.test.ts b/app/src/main/db/migrate.test.ts index 712780d..ad1585b 100644 --- a/app/src/main/db/migrate.test.ts +++ b/app/src/main/db/migrate.test.ts @@ -38,6 +38,7 @@ describe("db migrations", () => { expect(columns(db, "wakes")).toContain("source_id"); // v4 expect(columns(db, "agents")).toContain("last_turn_at"); // v5 expect(columns(db, "schedules")).toContain("timezone"); // v6 + expect(columns(db, "wakes")).toContain("outcome"); // v7 // Whole new tables ride the DDL — no migration, no version bump (§6.3). expect(columns(db, "recent_notifications")).toContain("at"); }); @@ -141,6 +142,33 @@ describe("db migrations", () => { expect(row.timezone).toBeNull(); // the scheduler backfills this at its next boot }); + test("v7 adds the wake outcome columns to an existing (v6) DB, preserving rows as NULL", () => { + const db = new Database(":memory:"); + const m = wrap(db); + // A pre-v7 wakes table (through source_id) with a recorded fire in it, stamped at v6. + db.exec(`CREATE TABLE wakes ( + id TEXT PRIMARY KEY, agent_id TEXT NOT NULL, source_kind TEXT NOT NULL, + source_id TEXT, prompt TEXT NOT NULL, background INTEGER NOT NULL, + fired_at INTEGER NOT NULL + );`); + db.exec( + `INSERT INTO wakes (id, agent_id, source_kind, source_id, prompt, background, fired_at) + VALUES ('w1', 'a1', 'cron', 's1', 'run', 1, 1000)`, + ); + db.exec("PRAGMA user_version = 6"); + db.exec(SCHEMA_DDL); + migrate(m, { fresh: false }); + + expect(userVersion(m)).toBe(SCHEMA_VERSION); + for (const c of ["outcome", "finished_at", "failure_reason", "failure_category"]) { + expect(columns(db, "wakes")).toContain(c); + } + const row = db.query("SELECT * FROM wakes WHERE id = 'w1'").get() as Record; + expect(row.source_id).toBe("s1"); // data survived + expect(row.outcome).toBeNull(); // pre-v7 rows have no known outcome + expect(row.finished_at).toBeNull(); + }); + test("refuses to open a DB stamped newer than this build (downgrade guard)", () => { const db = new Database(":memory:"); const m = wrap(db); diff --git a/app/src/main/db/migrate.ts b/app/src/main/db/migrate.ts index 9388a55..15f00e8 100644 --- a/app/src/main/db/migrate.ts +++ b/app/src/main/db/migrate.ts @@ -34,7 +34,7 @@ export interface MigrationDb { } /** Bump on every schema change, with a matching entry in MIGRATIONS. */ -export const SCHEMA_VERSION = 6; +export const SCHEMA_VERSION = 7; const MIGRATIONS: Record void> = { // v2 — headless turn limit: per-agent unattended-turn counter + on/off toggle. @@ -63,6 +63,14 @@ const MIGRATIONS: Record void> = { 6: (db) => { addColumnIfMissing(db, "schedules", "timezone", "TEXT"); }, + // v7 — per-wake outcome for the Monitor tab's History (§12.2): how the run ended, + // when, and why it failed. All nullable: pre-v7 rows read NULL (no outcome known). + 7: (db) => { + addColumnIfMissing(db, "wakes", "outcome", "TEXT"); + addColumnIfMissing(db, "wakes", "finished_at", "INTEGER"); + addColumnIfMissing(db, "wakes", "failure_reason", "TEXT"); + addColumnIfMissing(db, "wakes", "failure_category", "TEXT"); + }, }; export function userVersion(db: MigrationDb): number { diff --git a/app/src/main/db/schema.ts b/app/src/main/db/schema.ts index 7973cbf..762f4e5 100644 --- a/app/src/main/db/schema.ts +++ b/app/src/main/db/schema.ts @@ -126,7 +126,17 @@ export const wakes = sqliteTable( prompt: text("prompt").notNull(), /** True if delivered headlessly (no live interactive session); false if warm via the channel. */ background: integer("background", { mode: "boolean" }).notNull(), + /** When the run/turn actually started — a row exists only for a wake that did. */ firedAt: integer("fired_at").notNull(), + /** `WakeOutcome`: "running" | "succeeded" | "failed" | "stopped". Nullable: rows + * written before v7 read NULL. */ + outcome: text("outcome"), + /** When `outcome` settled; NULL while running (and on pre-v7 rows). */ + finishedAt: integer("finished_at"), + /** `WakeFailureReason` (see `@shared/schedule`), set only on a failed run. */ + failureReason: text("failure_reason"), + /** `WakeFailureCategory` classified from the failure text, when recognized. */ + failureCategory: text("failure_category"), }, (t) => [index("wakes_agent_fired").on(t.agentId, t.firedAt)], ); diff --git a/app/src/main/services/analytics/analytics.test.ts b/app/src/main/services/analytics/analytics.test.ts index 03564d7..91e53a5 100644 --- a/app/src/main/services/analytics/analytics.test.ts +++ b/app/src/main/services/analytics/analytics.test.ts @@ -293,6 +293,11 @@ const DIAG: FeedbackDiagnostics = { agents_active_24h: 2, schedules_enabled: 0, monitors_enabled: 0, + wakes_7d: 0, + wakes_failed_7d: 0, + wakes_stopped_7d: 0, + wakes_failed_api_7d: 0, + wakes_top_failure_7d: null, broker_status: "connected", broker_authorized: true, broker_portfolio_age_sec: 5, diff --git a/app/src/main/services/feedback/diagnostics.test.ts b/app/src/main/services/feedback/diagnostics.test.ts index 08bbe28..80b8b2b 100644 --- a/app/src/main/services/feedback/diagnostics.test.ts +++ b/app/src/main/services/feedback/diagnostics.test.ts @@ -35,6 +35,7 @@ const VALUES: DiagnosticsValues = { ], crons: 4, monitors: 1, + wakes: { total: 12, failed: 3, stopped: 1, apiErrors: 2, topFailureCategory: "billing" }, brokerStatus: "connected", brokerAuthorized: true, portfolioFetchedAt: NOW - 12_500, @@ -63,6 +64,11 @@ describe("buildDiagnostics", () => { agents_active_24h: 1, schedules_enabled: 4, monitors_enabled: 1, + wakes_7d: 12, + wakes_failed_7d: 3, + wakes_stopped_7d: 1, + wakes_failed_api_7d: 2, + wakes_top_failure_7d: "billing", broker_status: "connected", broker_authorized: true, broker_portfolio_age_sec: 12, diff --git a/app/src/main/services/feedback/diagnostics.ts b/app/src/main/services/feedback/diagnostics.ts index 3b6207c..de623ff 100644 --- a/app/src/main/services/feedback/diagnostics.ts +++ b/app/src/main/services/feedback/diagnostics.ts @@ -3,12 +3,18 @@ import type { Agent } from "@shared/agent"; import type { BrokerConnectionStatus } from "@shared/broker"; import type { FeedbackDiagnostics } from "@shared/feedback"; import type { AppSettings } from "@shared/settings"; +import type { WakeStats } from "../scheduler"; + +/** The wake window the diagnostics block summarizes. */ +export const WAKE_STATS_WINDOW_MS = 7 * 86_400_000; export interface DiagnosticsValues { agents: Agent[]; /** Enabled cron schedules / monitors across every agent. */ crons: number; monitors: number; + /** Wake outcomes over the last `WAKE_STATS_WINDOW_MS`, every agent. */ + wakes: WakeStats; brokerStatus: BrokerConnectionStatus; brokerAuthorized: boolean; /** `fetchedAt` of the cached portfolio snapshot, or null before the first fetch. */ @@ -61,6 +67,12 @@ export function buildDiagnostics(v: DiagnosticsValues, now = Date.now()): Feedba schedules_enabled: v.crons, monitors_enabled: v.monitors, + wakes_7d: v.wakes.total, + wakes_failed_7d: v.wakes.failed, + wakes_stopped_7d: v.wakes.stopped, + wakes_failed_api_7d: v.wakes.apiErrors, + wakes_top_failure_7d: v.wakes.topFailureCategory, + broker_status: v.brokerStatus, broker_authorized: v.brokerAuthorized, broker_portfolio_age_sec: diff --git a/app/src/main/services/harness/claude.ts b/app/src/main/services/harness/claude.ts index bd9e91d..f87088b 100644 --- a/app/src/main/services/harness/claude.ts +++ b/app/src/main/services/harness/claude.ts @@ -21,7 +21,7 @@ const execFileAsync = promisify(execFile); /** * The agent-dir `.claude/settings.json` that wires Claude Code's order gate: the * PreToolUse hook on the money-moving tools (→ `approval-gate.sh`, the approval card), - * the PostToolUse order-result capture, and the Notification/Stop status hooks, plus + * the PostToolUse order-result capture, and the Notification/Stop/StopFailure status hooks, plus * the allowlist for reads and cosmetic writes. The matcher and allowlist derive from * the classification table in `@shared/robinhood-tools` — the single place the gated * set is maintained. Generated IN CODE (not copied from the template): the @@ -75,6 +75,16 @@ const CLAUDE_SETTINGS_JSON = `${JSON.stringify( ], }, ], + // Fires INSTEAD of Stop when the turn ends in an API error (billing, auth, rate + // limit, server/connection failure). Same forwarder: the host reads + // `hook_event_name` + `error` and settles the outstanding wake as failed (§12.2). + StopFailure: [ + { + hooks: [ + { type: "command", command: "$CLAUDE_PROJECT_DIR/.claude/hooks/status-notify.sh" }, + ], + }, + ], }, }, null, diff --git a/app/src/main/services/local-api/index.ts b/app/src/main/services/local-api/index.ts index 9e931f2..b76c080 100644 --- a/app/src/main/services/local-api/index.ts +++ b/app/src/main/services/local-api/index.ts @@ -6,6 +6,7 @@ import type { AgentRegistry } from "../agents/registry"; import type { ApprovalService } from "../approvals"; import type { BrokerService } from "../broker"; import type { Scheduler } from "../scheduler"; +import { categoryForStopFailure } from "../scheduler/wake/failure-category"; import type { WakeTransport } from "../scheduler/wake/types"; import type { StatusArbiter } from "../status/arbiter"; @@ -233,13 +234,20 @@ export class LocalApiServer { // which is exactly what "last active" should track — a user message produces // neither, so typing at an agent never moves its timestamp (§12.6). Codex // fires Stop only (it has no Notification event); claude fires both. - if (event === "Notification" || event === "Stop") { + // StopFailure (claude-only) fires INSTEAD of Stop when the turn ends in an API + // error — same "turn is over" bookkeeping as Stop, but the outstanding wake failed. + const turnEnded = event === "Stop" || event === "StopFailure"; + if (event === "Notification" || turnEnded) { this.registry.markAgentTurn(agentId); } if (event === "Notification") { this.deps.arbiter.setNeedsInput(agentId, true); - } else if (event === "Stop") { + } else if (turnEnded) { this.deps.arbiter.setNeedsInput(agentId, false); + // A wake delivered into this session has now been consumed: settle its History + // row (§12.2). No-op when no wake is outstanding (a plain user turn). + if (event === "Stop") this.wake?.onTurnEnded(agentId); + else this.wake?.onTurnFailed(agentId, categoryForStopFailure(String(body?.error ?? ""))); // Session capture is claude-only: for codex, `lastSessionId` is the // app-server THREAD id (minted by thread/start and adopted from the TUI, // §13) — codex's hook `session_id` is not verified to match it, and an diff --git a/app/src/main/services/local-api/status-route.test.ts b/app/src/main/services/local-api/status-route.test.ts new file mode 100644 index 0000000..f7d1d1a --- /dev/null +++ b/app/src/main/services/local-api/status-route.test.ts @@ -0,0 +1,85 @@ +import { afterAll, beforeAll, beforeEach, describe, expect, test } from "bun:test"; +import type { AgentRegistry } from "../agents/registry"; +import type { WakeTransport } from "../scheduler/wake/types"; +import { LocalApiServer } from "./index"; + +/** Every side effect the status route can have, in call order. */ +const calls: string[] = []; + +const registry = { + get: (id: string) => (id === "a1" ? ({ id: "a1", harness: "claude" } as never) : undefined), + markAgentTurn: (id: string) => calls.push(`turn:${id}`), + setLastSessionId: (id: string, sid: string) => calls.push(`session:${id}:${sid}`), +} as unknown as AgentRegistry; +const arbiter = { + setNeedsInput: (id: string, on: boolean) => calls.push(`needsInput:${id}:${on}`), +}; +const wake = { + onTurnEnded: (id: string) => calls.push(`ended:${id}`), + onTurnFailed: (id: string, category: string) => calls.push(`failed:${id}:${category}`), +} as unknown as WakeTransport; + +// biome-ignore lint/suspicious/noExplicitAny: stubs stand in for the real services +const server = new LocalApiServer({ registry, arbiter } as any); +server.setWake(wake); +let base = ""; +const hdrs = { + "x-opentrade-token": "", + "x-opentrade-agent": "a1", + "content-type": "application/json", +}; + +beforeAll(async () => { + await server.start(); + base = `http://127.0.0.1:${server.port}`; + hdrs["x-opentrade-token"] = server.token; +}); +afterAll(() => server.stop()); +beforeEach(() => { + calls.length = 0; +}); + +/** POST a Claude Code hook payload as the agent's status hook script would. */ +async function hook(body: Record): Promise { + const res = await fetch(`${base}/hook/status`, { + method: "POST", + headers: hdrs, + body: JSON.stringify(body), + }); + return res.status; +} + +describe("/hook/status", () => { + test("Notification: needs-input on, last-active stamped, no wake settlement", async () => { + expect(await hook({ hook_event_name: "Notification", session_id: "s1" })).toBe(200); + expect(calls).toEqual(["turn:a1", "needsInput:a1:true"]); + }); + + test("Stop: clears needs-input, settles the wake succeeded, captures the session", async () => { + expect(await hook({ hook_event_name: "Stop", session_id: "s1" })).toBe(200); + expect(calls).toEqual(["turn:a1", "needsInput:a1:false", "ended:a1", "session:a1:s1"]); + }); + + test("StopFailure: same turn-ended bookkeeping, wake settles failed with the mapped category", async () => { + // The payload's `error` field is Claude Code's enum; the route maps it (billing_error → billing). + expect( + await hook({ hook_event_name: "StopFailure", error: "billing_error", session_id: "s1" }), + ).toBe(200); + expect(calls).toEqual(["turn:a1", "needsInput:a1:false", "failed:a1:billing", "session:a1:s1"]); + }); + + test("StopFailure without an error field still settles the wake (category other)", async () => { + expect(await hook({ hook_event_name: "StopFailure" })).toBe(200); + expect(calls).toContain("failed:a1:other"); + }); + + test("an unknown agent is rejected before any bookkeeping", async () => { + const res = await fetch(`${base}/hook/status`, { + method: "POST", + headers: { ...hdrs, "x-opentrade-agent": "nope" }, + body: JSON.stringify({ hook_event_name: "Stop" }), + }); + expect(res.status).toBe(200); // the route answers ok to the hook regardless… + expect(calls).toEqual([]); // …but touches nothing for an agent it doesn't know + }); +}); diff --git a/app/src/main/services/scheduler/index.ts b/app/src/main/services/scheduler/index.ts index a24c043..38aaf58 100644 --- a/app/src/main/services/scheduler/index.ts +++ b/app/src/main/services/scheduler/index.ts @@ -1,3 +1,4 @@ +import type { WakeFailureCategory } from "@shared/analytics"; import { firstLine } from "@shared/notify"; import type { CronCreateInput, @@ -6,7 +7,7 @@ import type { Schedule, Wake, } from "@shared/schedule"; -import { and, desc, eq, isNull } from "drizzle-orm"; +import { and, desc, eq, gte, isNull } from "drizzle-orm"; import { nanoid } from "nanoid"; import type { Db } from "../../db/client"; import { @@ -23,16 +24,28 @@ import { buildAgentEnv } from "../terminal/env"; import { CronTimer } from "./cron-timer"; import { MonitorRunner } from "./monitor-runner"; import { systemTimeZone } from "./system-timezone"; -import type { WakeTransport } from "./wake/types"; +import type { PendingWake, WakeResult, WakeTransport } from "./wake/types"; /** * Durable autonomy scheduler, owned by the always-on backend host. Arms cron * timers and supervises monitor children that survive the GUI closing (unlike - * Claude Code's session-scoped CronCreate/Monitor). When a trigger fires it - * records a wake in the Run History feed and hands a wake to the `WakeTransport`, which delivers - * it either warm (a `claude/channel` inject into the live PTY) or cold (a headless - * `claude --resume -p` run) — the Scheduler doesn't care which. + * Claude Code's session-scoped CronCreate/Monitor). When a trigger fires it hands a + * wake to the `WakeTransport`, which delivers it either warm (a `claude/channel` inject + * into the live PTY) or cold (a headless `claude --resume -p` run) — the Scheduler + * doesn't care which. The coordinator calls back (`wakeStarted` / `wakeFinished`) when + * the wake actually runs, and THAT is when the History row is written and settled. */ +/** Wake outcome counts over a window, for the feedback diagnostics block (§12.8). */ +export interface WakeStats { + total: number; + failed: number; + stopped: number; + /** Failures whose reason is `api_error` (an API-error turn reported by StopFailure). */ + apiErrors: number; + /** The most frequent failure category in the window, or null when nothing failed. */ + topFailureCategory: WakeFailureCategory | null; +} + export class Scheduler { private cron = new CronTimer(); private runners = new Map(); @@ -54,6 +67,17 @@ export class Scheduler { .set({ timezone: bootZone }) .where(isNull(schedulesTable.timezone)) .run(); + // Any wake still `running` at boot is orphaned: this host is the table's only writer + // and nothing in flight survives a restart — a clean quit SIGTERMs headless children + // and clears their markers before an exit handler can settle the row, a crash settles + // nothing, and a warm wake's `liveWakes` die with the process. Mark them `stopped` so + // History doesn't spin forever and `wakeStats` can count them. No `wake_finished` + // event: the settle time is unknown. + this.db + .update(wakesTable) + .set({ outcome: "stopped", finishedAt: Date.now() }) + .where(eq(wakesTable.outcome, "running")) + .run(); for (const row of this.db.select().from(schedulesTable).all()) { if (!row.enabled) continue; // Self-heal a genuinely-orphaned row (agent no longer exists at all) by hard-deleting @@ -319,6 +343,27 @@ export class Scheduler { .map(rowToMonitor); } + /** ALL of this agent's crons and monitors, **retired rows included** — for the Monitor + * tab, whose History resolves a fire's trigger after the trigger is gone (the point of + * retire-not-delete). The Active list filters `enabled` itself; the MCP-facing + * `listCron`/`listMonitors` keep hiding retired rows. */ + listTriggers(agentId: string): { schedules: Schedule[]; monitors: Monitor[] } { + return { + schedules: this.db + .select() + .from(schedulesTable) + .where(eq(schedulesTable.agentId, agentId)) + .all() + .map(rowToSchedule), + monitors: this.db + .select() + .from(monitorsTable) + .where(eq(monitorsTable.agentId, agentId)) + .all() + .map(rowToMonitor), + }; + } + // ---- wake history ---- /** This agent's recorded wakes, newest first, for the Run History pane. */ @@ -434,13 +479,15 @@ export class Scheduler { } /** - * Record the fire in the Monitor tab and hand the wake to the coordinator. The - * coordinator owns routing (interactive via the channel / headless via `-p`) and - * per-agent queueing, so a fire is fire-and-forget here — never blocks the timer/monitor. + * Hand the wake to the coordinator. The coordinator owns routing (interactive via the + * channel / headless via `-p`) and per-agent queueing, so a fire is fire-and-forget here + * — never blocks the timer/monitor. Nothing is recorded yet: the History row is written + * when the wake actually starts (`wakeStarted`), so a wake the coordinator later drops + * (turn budget spent while queued, agent gone broken) leaves no row. * Returns whether the fire actually happened: `false` means the agent is paused (broken / * out of turns) and the wake was skipped, so the caller must NOT consume the schedule * (advance last-fired, retire a one-shot) — see the callers in `armCron` / `start()`. - * `sourceId` is the originating schedule/monitor id, stored on the wake (with `sourceKind`) + * `sourceId` is the originating schedule/monitor id, carried on the wake (with `sourceKind`) * so history can resolve the timer's details even after it's retired — every caller * (`armCron`, `start()` catch-up, `startMonitor`'s trigger) has it in scope. */ @@ -457,44 +504,126 @@ export class Scheduler { // notification + history row on every cron tick / monitor trigger while nothing runs. // Re-checked live each fire, so a reset / toggle / GUI-open resumes with no re-arm. if (this.wake.wouldDropWake(agentId)) return false; - // How this wake will be delivered: a live interactive session (the channel) takes it - // warm; anything else (offline/headless) routes to a background `-p` run. The - // coordinator decides this synchronously off the same execution state, so reading it - // here — before enqueue — captures the routing the wake will get. - const background = this.registry.executionStateOf(agentId) !== "interactive"; - // The wake row is the Monitor tab's record of this fire; the `scheduler:changed` - // emit below re-queries it (and the upcoming schedules) — no separate bus event needed. + // The id minted here becomes the History row's id once the coordinator starts the + // wake (`wakeStarted`), so a fire is traceable end to end. + this.wake.enqueue(agentId, { id: nanoid(), agentId, prompt, sourceKind, sourceId }); + // Surface the updated last/next-fire times in the Monitor tab live. + bus.emitEvent("scheduler:changed", { agentId }); + return true; + } + + // ---- wake outcome recording (SchedulerControl, called back by the coordinator) ---- + + /** + * The coordinator actually started the wake — a headless child spawned (`background`) + * or the live session accepted it. Only now does the fire exist in History: the row is + * written `running`, the user is notified, and the agent's "last active" moves. + */ + wakeStarted(wake: PendingWake, background: boolean): void { + const agent = this.registry.get(wake.agentId); + // Same guard as `fire()`: an agent archived while this wake sat queued gets no row + // (the strategy short-circuits its run anyway) and no "now running" notification. + if (!agent || agent.archivedAt !== null) return; this.db .insert(wakesTable) .values({ - id: nanoid(), - agentId, - sourceKind, - sourceId, - prompt, + id: wake.id, + agentId: wake.agentId, + sourceKind: wake.sourceKind, + sourceId: wake.sourceId, + prompt: wake.prompt, background, firedAt: Date.now(), + outcome: "running", }) .run(); analytics.track("schedule_fired", { - source: sourceKind, + source: wake.sourceKind, path: background ? "headless" : "warm", }); // Stamp `last_turn_at` at wake START, not just at the Stop hook: a run that dies // mid-flight (API error — Stop never fires) still moves the tray's "last active". - this.registry.markAgentTurn(agentId); - this.wake.enqueue(agentId, prompt); - // Surface the new wake + updated last/next-fire times in the Monitor tab live. - bus.emitEvent("scheduler:changed", { agentId }); + this.registry.markAgentTurn(wake.agentId); + // The new row is the Monitor tab's record of this fire; the emit re-queries it. + bus.emitEvent("scheduler:changed", { agentId: wake.agentId }); // Notify the user their agent just started working (the launcher shows it only // while OpenTrade is unfocused — see §12.4). bus.emitEvent("notify", { kind: "wake", - title: `${agent.name} — ${sourceKind === "cron" ? "Scheduled run" : "Monitor fired"}`, - body: `${agent.name} is now running: ${firstLine(prompt)}`, - agentId, + title: `${agent.name} — ${wake.sourceKind === "cron" ? "Scheduled run" : "Monitor fired"}`, + body: `${agent.name} is now running: ${firstLine(wake.prompt)}`, + agentId: wake.agentId, }); - return true; + } + + /** The started wake settled: stamp its outcome (+ failure detail) and finish time, and + * track it — this is the one place that knows the final outcome on either path. */ + wakeFinished(wake: PendingWake, result: WakeResult): void { + const finishedAt = Date.now(); + // One guarded UPDATE: only a row still `running` settles, so a wake that never + // started (agent archived meanwhile) or was already settled is a no-op — exactly-once + // holds at the DB layer too, never a double `wake_finished`. RETURNING hands back the + // two facts the event needs that `PendingWake` doesn't carry. + const row = this.db + .update(wakesTable) + .set({ + outcome: result.outcome, + finishedAt, + failureReason: result.failureReason ?? null, + failureCategory: result.failureCategory ?? null, + }) + .where(and(eq(wakesTable.id, wake.id), eq(wakesTable.outcome, "running"))) + .returning({ background: wakesTable.background, firedAt: wakesTable.firedAt }) + .get(); + if (!row) return; + analytics.track("wake_finished", { + source: wake.sourceKind, + path: row.background ? "headless" : "warm", + outcome: result.outcome, + duration_ms: Math.max(0, finishedAt - row.firedAt), + ...(result.failureReason ? { failure_reason: result.failureReason } : {}), + ...(result.failureCategory ? { failure_category: result.failureCategory } : {}), + }); + bus.emitEvent("scheduler:changed", { agentId: wake.agentId }); + } + + /** + * Outcome counts across ALL agents for wakes started since `sinceMs` — the feedback + * form's "did autonomy work" block (§12.8). Counts and one category, never prompts. + */ + wakeStats(sinceMs: number): WakeStats { + const rows = this.db + .select({ + outcome: wakesTable.outcome, + failureReason: wakesTable.failureReason, + failureCategory: wakesTable.failureCategory, + }) + .from(wakesTable) + .where(gte(wakesTable.firedAt, sinceMs)) + .all(); + let failed = 0; + let stopped = 0; + let apiErrors = 0; + const byCategory = new Map(); + for (const r of rows) { + if (r.outcome === "failed") { + failed += 1; + if (r.failureReason === "api_error") apiErrors += 1; + if (r.failureCategory) { + byCategory.set(r.failureCategory, (byCategory.get(r.failureCategory) ?? 0) + 1); + } + } else if (r.outcome === "stopped") { + stopped += 1; + } + } + const top = [...byCategory.entries()].sort((a, b) => b[1] - a[1])[0]?.[0] ?? null; + return { + total: rows.length, + failed, + stopped, + apiErrors, + topFailureCategory: top as WakeFailureCategory | null, + }; } /** @@ -565,5 +694,9 @@ function rowToWake(row: typeof wakesTable.$inferSelect): Wake { prompt: row.prompt, background: row.background, firedAt: row.firedAt, + outcome: row.outcome as Wake["outcome"], + finishedAt: row.finishedAt, + failureReason: row.failureReason as Wake["failureReason"], + failureCategory: row.failureCategory as Wake["failureCategory"], }; } diff --git a/app/src/main/services/scheduler/scheduler.test.ts b/app/src/main/services/scheduler/scheduler.test.ts index 7e86ade..826a540 100644 --- a/app/src/main/services/scheduler/scheduler.test.ts +++ b/app/src/main/services/scheduler/scheduler.test.ts @@ -10,7 +10,7 @@ import type { AgentRegistry } from "../agents/registry"; import { bus } from "../event-bus"; import type { LocalApiServer } from "../local-api"; import { Scheduler } from "./index"; -import type { WakeTransport } from "./wake/types"; +import type { PendingWake, WakeTransport } from "./wake/types"; function memDb(): Db { const sqlite = new Database(":memory:"); @@ -27,7 +27,8 @@ function memDb(): Db { CREATE TABLE wakes ( id TEXT PRIMARY KEY, agent_id TEXT NOT NULL, source_kind TEXT NOT NULL, source_id TEXT, prompt TEXT NOT NULL, background INTEGER NOT NULL, - fired_at INTEGER NOT NULL); + fired_at INTEGER NOT NULL, outcome TEXT, finished_at INTEGER, + failure_reason TEXT, failure_category TEXT); `); return drizzle(sqlite, { schema }) as unknown as Db; } @@ -47,11 +48,20 @@ const AGENT: Agent = { archivedAt: null, }; -/** Standard test stubs: a no-op wake transport (fires never dropped), a registry that - * only knows AGENT, and a dummy localApi. */ +/** Standard test stubs: a wake transport that "starts" every wake immediately as a + * background run (so a fire lands in the wakes table synchronously, the way the + * coordinator's `wakeStarted` callback does in prod), a registry that only knows AGENT, + * and a dummy localApi. `bind` wires the transport to the scheduler once built. */ function stdDeps() { + let scheduler: Scheduler | undefined; + const wakes: PendingWake[] = []; // every wake handed to the transport, in order const wake: WakeTransport = { - enqueue: () => {}, + enqueue: (_agentId, w) => { + wakes.push(w); + scheduler?.wakeStarted(w, true); + }, + onTurnEnded: () => {}, + onTurnFailed: () => {}, awaitPoll: async () => null, onInteractiveUp: () => {}, onInteractiveDown: () => {}, @@ -66,14 +76,23 @@ function stdDeps() { markAgentTurn: () => {}, } as unknown as AgentRegistry; const localApi = { port: 12345, token: "tok" } as unknown as LocalApiServer; - return { wake, registry, localApi }; + const bind = (s: Scheduler) => { + scheduler = s; + return s; + }; + return { wake, wakes, registry, localApi, bind }; } /** Build a Scheduler over a caller-supplied db so the test can inspect raw rows * (e.g. assert a retired row survives). */ function makeSchedulerOn(db: Db): Scheduler { - const { wake, registry, localApi } = stdDeps(); - return new Scheduler(db, wake, registry, localApi); + return makeSchedulerWithWakes(db).s; +} + +/** As `makeSchedulerOn`, also exposing the wakes the transport received. */ +function makeSchedulerWithWakes(db: Db): { s: Scheduler; wakes: PendingWake[] } { + const { wake, wakes, registry, localApi, bind } = stdDeps(); + return { s: bind(new Scheduler(db, wake, registry, localApi)), wakes }; } function makeScheduler() { @@ -185,6 +204,8 @@ describe("Scheduler CRUD", () => { .run(); const wake: WakeTransport = { enqueue: () => {}, + onTurnEnded: () => {}, + onTurnFailed: () => {}, awaitPoll: async () => null, onInteractiveUp: () => {}, onInteractiveDown: () => {}, @@ -230,6 +251,8 @@ describe("Scheduler CRUD", () => { enqueue: () => { enqueued += 1; }, + onTurnEnded: () => {}, + onTurnFailed: () => {}, awaitPoll: async () => null, onInteractiveUp: () => {}, onInteractiveDown: () => {}, @@ -299,6 +322,8 @@ describe("Scheduler CRUD", () => { enqueue: () => { enqueued += 1; }, + onTurnEnded: () => {}, + onTurnFailed: () => {}, awaitPoll: async () => null, onInteractiveUp: () => {}, onInteractiveDown: () => {}, @@ -382,6 +407,155 @@ describe("Scheduler CRUD", () => { expect(wakes[0].sourceId).toBe("cron1"); // links wake → (now retired) schedule }); + test("wakeStarted writes the row `running`; wakeFinished settles it once with the failure detail", () => { + const db = memDb(); + db.insert(schema.schedules) + .values({ + id: "cron1", + agentId: "agent1", + cronExpr: "0 9 * * *", + prompt: "p", + recurring: false, + enabled: true, + nextFireAt: 1, // past → start() catch-up fires it + lastFiredAt: null, + createdAt: 1, + }) + .run(); + const { s, wakes: handed } = makeSchedulerWithWakes(db); + scheduler = s; + scheduler.start(); + + expect(handed).toHaveLength(1); + let row = db.select().from(schema.wakes).get()!; + expect(row.id).toBe(handed[0].id); // the fire-time id IS the History row's id + expect(row.outcome).toBe("running"); + expect(row.finishedAt).toBeNull(); + + scheduler.wakeFinished(handed[0], { + outcome: "failed", + failureReason: "resume_fail", + failureCategory: "billing", + }); + row = db.select().from(schema.wakes).get()!; + expect(row.outcome).toBe("failed"); + expect(row.failureReason).toBe("resume_fail"); + expect(row.failureCategory).toBe("billing"); + expect(row.finishedAt).not.toBeNull(); + expect(db.select().from(schema.wakes).all()).toHaveLength(1); // updated, not re-inserted + + // Settling is exactly-once at the DB layer: a second settle for the same wake is a no-op. + scheduler.wakeFinished(handed[0], { outcome: "succeeded" }); + row = db.select().from(schema.wakes).get()!; + expect(row.outcome).toBe("failed"); + expect(row.failureCategory).toBe("billing"); + }); + + test("start() marks wakes left `running` by the previous host as stopped", () => { + const db = memDb(); + const base = { + agentId: "agent1", + sourceKind: "cron", + sourceId: null, + prompt: "p", + background: true, + firedAt: 1000, + finishedAt: null, + failureReason: null, + failureCategory: null, + } as const; + db.insert(schema.wakes) + .values({ ...base, id: "orphan", outcome: "running" }) + .run(); + db.insert(schema.wakes) + .values({ ...base, id: "done", outcome: "succeeded", finishedAt: 2000 }) + .run(); + scheduler = makeSchedulerOn(db); + scheduler.start(); + + const rows = Object.fromEntries( + db + .select() + .from(schema.wakes) + .all() + .map((r) => [r.id, r]), + ); + expect(rows.orphan.outcome).toBe("stopped"); + expect(rows.orphan.finishedAt).not.toBeNull(); + expect(rows.done.outcome).toBe("succeeded"); // settled rows are untouched + expect(rows.done.finishedAt).toBe(2000); + }); + + test("wakeStats counts outcomes in the window across agents, with the top failure category", () => { + const db = memDb(); + scheduler = makeSchedulerOn(db); + const now = Date.now(); + const row = (id: string, agentId: string, firedAt: number, extra: Record) => + db + .insert(schema.wakes) + .values({ + id, + agentId, + sourceKind: "cron", + sourceId: null, + prompt: "p", + background: true, + firedAt, + outcome: null, + finishedAt: null, + failureReason: null, + failureCategory: null, + ...extra, + }) + .run(); + row("w1", "agent1", now - 1000, { outcome: "succeeded" }); + row("w2", "agent1", now - 2000, { + outcome: "failed", + failureReason: "resume_fail", + failureCategory: "billing", + }); + row("w3", "agent2", now - 3000, { + outcome: "failed", + failureReason: "api_error", + failureCategory: "billing", + }); + row("w4", "agent2", now - 4000, { + outcome: "failed", + failureReason: "api_error", + failureCategory: "network", + }); + row("w5", "agent1", now - 5000, { outcome: "stopped" }); + row("w6", "agent1", now - 6000, { outcome: "running" }); + row("old", "agent1", now - 10 * 86_400_000, { outcome: "failed", failureCategory: "auth" }); // outside + + expect(scheduler.wakeStats(now - 7 * 86_400_000)).toEqual({ + total: 6, + failed: 3, + stopped: 1, + apiErrors: 2, + topFailureCategory: "billing", + }); + expect(scheduler.wakeStats(now).topFailureCategory).toBeNull(); // empty window + }); + + test("listTriggers returns the agent's retired crons/monitors too (History resolves them)", () => { + scheduler = makeScheduler(); + const cron = scheduler.createCron("agent1", { + cron: "30 9 * * 1-5", + prompt: "p", + recurring: true, + }); + const mon = scheduler.createMonitor("agent1", { command: "sleep 5" }); + scheduler.deleteCron("agent1", cron.id); + scheduler.stopMonitor("agent1", mon.id); + + expect(scheduler.listCron("agent1")).toEqual([]); // MCP-facing lists hide retired rows + expect(scheduler.listMonitors("agent1")).toEqual([]); + const all = scheduler.listTriggers("agent1"); + expect(all.schedules.map((s) => [s.id, s.enabled])).toEqual([[cron.id, false]]); + expect(all.monitors.map((m) => [m.id, m.enabled])).toEqual([[mon.id, false]]); + }); + test("a fired monitor wake records source_id + source_kind linking back to its monitor", async () => { const db = memDb(); scheduler = makeSchedulerOn(db); diff --git a/app/src/main/services/scheduler/wake/codex-strategy.ts b/app/src/main/services/scheduler/wake/codex-strategy.ts index 129762e..60e1eb5 100644 --- a/app/src/main/services/scheduler/wake/codex-strategy.ts +++ b/app/src/main/services/scheduler/wake/codex-strategy.ts @@ -1,5 +1,4 @@ import { basename } from "node:path"; -import type { WakeFailureCategory } from "@shared/analytics"; import { hostLog } from "../../../host/log"; import type { AgentRegistry } from "../../agents/registry"; import { analytics } from "../../analytics"; @@ -12,7 +11,7 @@ import { import { classifyWakeFailure } from "./failure-category"; import { formatWakePrompt } from "./prompt"; import { clearSpawnMarker, writeSpawnMarker } from "./spawn-marker"; -import type { HeadlessExitReason, HeadlessWakeStrategy } from "./types"; +import type { HeadlessExit, HeadlessWakeStrategy } from "./types"; /** * The codex headless transport: instead of spawning a one-shot CLI child, a wake @@ -31,7 +30,7 @@ export class CodexHeadlessStrategy implements HeadlessWakeStrategy { private manager: CodexAppServerManager, ) {} - run(agentId: string, prompt: string, onExit: (reason: HeadlessExitReason) => void): void { + run(agentId: string, prompt: string, onExit: HeadlessExit): void { const agent = this.registry.get(agentId); if (!agent || agent.archivedAt !== null) { onExit("ok"); @@ -46,17 +45,13 @@ export class CodexHeadlessStrategy implements HeadlessWakeStrategy { const startedAt = Date.now(); let settled = false; - const settle = (reason: HeadlessExitReason, failureCategory?: WakeFailureCategory) => { + const settle: HeadlessExit = (reason, failureCategory) => { if (settled) return; settled = true; this.active.delete(agentId); clearSpawnMarker(agentId); - analytics.track("headless_run_finished", { - result: reason === "ok" ? "ok" : reason === "resumeFail" ? "resume_fail" : "spawn_fail", - duration_ms: Math.max(0, Date.now() - startedAt), - ...(failureCategory ? { failure_category: failureCategory } : {}), - }); - onExit(reason); + // The outcome is tracked once the coordinator settles the wake (`wake_finished`, §12.2). + onExit(reason, failureCategory); }; void (async () => { @@ -148,7 +143,7 @@ export class HarnessRoutingHeadlessStrategy implements HeadlessWakeStrategy { return this.registry.get(agentId)?.harness === "codex" ? this.codex : this.claude; } - run(agentId: string, prompt: string, onExit: (reason: HeadlessExitReason) => void): void { + run(agentId: string, prompt: string, onExit: HeadlessExit): void { this.pick(agentId).run(agentId, prompt, onExit); } diff --git a/app/src/main/services/scheduler/wake/coordinator.test.ts b/app/src/main/services/scheduler/wake/coordinator.test.ts index c5217b1..3f7fc96 100644 --- a/app/src/main/services/scheduler/wake/coordinator.test.ts +++ b/app/src/main/services/scheduler/wake/coordinator.test.ts @@ -1,8 +1,16 @@ import { describe, expect, test } from "bun:test"; import type { ExecutionState } from "@shared/agent"; +import type { WakeFailureCategory } from "@shared/analytics"; import type { AgentRegistry } from "../../agents/registry"; import { WakeCoordinator } from "./coordinator"; -import type { HeadlessExitReason, HeadlessWakeStrategy } from "./types"; +import type { + HeadlessExit, + HeadlessExitReason, + HeadlessWakeStrategy, + PendingWake, + SchedulerControl, + WakeResult, +} from "./types"; const tick = () => new Promise((r) => setTimeout(r, 0)); const wait = (ms: number) => new Promise((r) => setTimeout(r, ms)); @@ -37,14 +45,14 @@ class FakeRegistry { class FakeHeadless implements HeadlessWakeStrategy { calls: string[] = []; stops = 0; - private exits: Array<(reason: HeadlessExitReason) => void> = []; - run(id: string, prompt: string, onExit: (reason: HeadlessExitReason) => void): void { + private exits: HeadlessExit[] = []; + run(id: string, prompt: string, onExit: HeadlessExit): void { this.calls.push(`${id}:${prompt}`); this.exits.push(onExit); } /** Simulate the active `-p` child exiting with the given outcome. */ - finishNext(reason: HeadlessExitReason = "ok"): void { - this.exits.shift()?.(reason); + finishNext(reason: HeadlessExitReason = "ok", failureCategory?: WakeFailureCategory): void { + this.exits.shift()?.(reason, failureCategory); } stop(): boolean { this.stops++; @@ -64,6 +72,26 @@ function make(maxHeadlessRunMs = 10_000, maxHeadlessTurns = 20, featureEnabled = return { reg, headless, coord }; } +let wakeSeq = 0; +/** A queued wake carrying `prompt` (the scheduler mints these at fire time). */ +function w(prompt: string): PendingWake { + wakeSeq += 1; + return { id: `wake${wakeSeq}`, agentId: "a", prompt, sourceKind: "cron", sourceId: "s1" }; +} + +/** A recording `SchedulerControl`: what the coordinator reported started/finished. */ +function makeRecorder() { + const started: Array<{ prompt: string; background: boolean }> = []; + const finished: Array<{ wakeId: string; result: WakeResult }> = []; + const sched: SchedulerControl = { + disarmAgent: () => {}, + rearmAgent: () => {}, + wakeStarted: (wake, background) => started.push({ prompt: wake.prompt, background }), + wakeFinished: (wake, result) => finished.push({ wakeId: wake.id, result }), + }; + return { sched, started, finished }; +} + /** Start a `/wake-stream` poll; returns the promise + its abort controller. */ function poll(c: WakeCoordinator, id: string, holdMs = 10_000) { const ac = new AbortController(); @@ -73,7 +101,7 @@ function poll(c: WakeCoordinator, id: string, holdMs = 10_000) { describe("WakeCoordinator — headless transport (ported)", () => { test("offline agent runs headless, completion-gated on exit", () => { const { reg, headless, coord } = make(); - coord.enqueue("a", "p1"); + coord.enqueue("a", w("p1")); expect(headless.calls).toEqual(["a:p1"]); expect(reg.executionStateOf("a")).toBe("headless"); headless.finishNext("ok"); @@ -82,8 +110,8 @@ describe("WakeCoordinator — headless transport (ported)", () => { test("a second headless wake queues behind the active run, then drains in order", () => { const { headless, coord } = make(); - coord.enqueue("b", "p1"); - coord.enqueue("b", "p2"); + coord.enqueue("b", w("p1")); + coord.enqueue("b", w("p2")); expect(headless.calls).toEqual(["b:p1"]); // p2 queued, held at the head until exit headless.finishNext("ok"); expect(headless.calls).toEqual(["b:p1", "b:p2"]); @@ -92,7 +120,7 @@ describe("WakeCoordinator — headless transport (ported)", () => { test("never serves a poll while a headless run holds the agent", async () => { const { headless, coord } = make(); - coord.enqueue("f", "p1"); // offline → headless run + coord.enqueue("f", w("p1")); // offline → headless run expect(headless.calls).toEqual(["f:p1"]); const ac = new AbortController(); expect(await coord.awaitPoll("f", ac.signal, 20)).toBeNull(); // channel inert under -p @@ -102,7 +130,7 @@ describe("WakeCoordinator — headless transport (ported)", () => { test("a headless run is killed by the max-runtime timer", async () => { const { headless, coord } = make(20); // tiny max-runtime - coord.enqueue("x", "p1"); + coord.enqueue("x", w("p1")); expect(headless.calls).toEqual(["x:p1"]); await wait(40); // kill timer fires → SIGTERM the child expect(headless.stops).toBe(1); @@ -113,7 +141,7 @@ describe("WakeCoordinator — interactive transport (channel)", () => { test("a wake queued before any poll is handed to the next poll", async () => { const { headless, coord } = make(); coord.onInteractiveUp("d"); - coord.enqueue("d", "p1"); + coord.enqueue("d", w("p1")); expect(headless.calls).toEqual([]); // interactive, no poll yet → queued, never headless const { p } = poll(coord, "d"); expect(await p).toBe("p1"); // handed off from the queue head @@ -124,7 +152,7 @@ describe("WakeCoordinator — interactive transport (channel)", () => { coord.onInteractiveUp("c"); const { p } = poll(coord, "c"); await tick(); - coord.enqueue("c", "p1"); // mid-turn or not, the channel accepts the push + coord.enqueue("c", w("p1")); // mid-turn or not, the channel accepts the push expect(await p).toBe("p1"); expect(headless.calls).toEqual([]); // never headless while interactive }); @@ -132,8 +160,8 @@ describe("WakeCoordinator — interactive transport (channel)", () => { test("two wakes fired back-to-back are handed to successive polls in order", async () => { const { coord } = make(); coord.onInteractiveUp("t"); - coord.enqueue("t", "p1"); - coord.enqueue("t", "p2"); // both queued (no poll parked yet) + coord.enqueue("t", w("p1")); + coord.enqueue("t", w("p2")); // both queued (no poll parked yet) const { p: p1 } = poll(coord, "t"); expect(await p1).toBe("p1"); const { p: p2 } = poll(coord, "t"); @@ -143,7 +171,7 @@ describe("WakeCoordinator — interactive transport (channel)", () => { test("an undelivered head re-routes to headless when the PTY dies before handoff", () => { const { headless, coord } = make(); coord.onInteractiveUp("m"); - coord.enqueue("m", "p1"); // interactive, no poll → queued + coord.enqueue("m", w("p1")); // interactive, no poll → queued expect(headless.calls).toEqual([]); coord.onInteractiveDown("m"); // PTY dies (crash / GUI quit) before any handoff expect(headless.calls).toEqual(["m:p1"]); // re-routed to the -p transport @@ -170,7 +198,7 @@ describe("WakeCoordinator — broken / resume-fail", () => { test("a broken agent drops its queued wakes and is never served", async () => { const { reg, headless, coord } = make(); reg.setExecutionState("h", "broken"); // seed from a boot-time reconcile - coord.enqueue("h", "p1"); + coord.enqueue("h", w("p1")); expect(headless.calls).toEqual([]); expect(reg.executionStateOf("h")).toBe("broken"); const ac = new AbortController(); @@ -180,12 +208,12 @@ describe("WakeCoordinator — broken / resume-fail", () => { test("broken only after 3 consecutive resume-fails; each drops its own wake", () => { const { reg, headless, coord } = make(); for (let i = 1; i <= 2; i++) { - coord.enqueue("r", `p${i}`); + coord.enqueue("r", w(`p${i}`)); expect(reg.executionStateOf("r")).toBe("headless"); headless.finishNext("resumeFail"); // drops the wake, increments the streak expect(reg.executionStateOf("r")).toBe("offline"); // not broken yet } - coord.enqueue("r", "p3"); + coord.enqueue("r", w("p3")); headless.finishNext("resumeFail"); // 3rd in a row expect(reg.executionStateOf("r")).toBe("broken"); expect(headless.calls).toEqual(["r:p1", "r:p2", "r:p3"]); // each ran once, then dropped @@ -193,19 +221,19 @@ describe("WakeCoordinator — broken / resume-fail", () => { test("a clean exit resets the resume-fail streak", () => { const { reg, headless, coord } = make(); - coord.enqueue("s", "p1"); + coord.enqueue("s", w("p1")); headless.finishNext("resumeFail"); // streak = 1 - coord.enqueue("s", "p2"); + coord.enqueue("s", w("p2")); headless.finishNext("ok"); // streak reset to 0 - coord.enqueue("s", "p3"); + coord.enqueue("s", w("p3")); headless.finishNext("resumeFail"); // streak = 1 again, NOT 3 expect(reg.executionStateOf("s")).toBe("offline"); }); test("a spawn error is one-strike broken and drops the queue", () => { const { reg, headless, coord } = make(); - coord.enqueue("z", "p1"); - coord.enqueue("z", "p2"); // queued behind the active run + coord.enqueue("z", w("p1")); + coord.enqueue("z", w("p2")); // queued behind the active run headless.finishNext("spawnFail"); expect(reg.executionStateOf("z")).toBe("broken"); expect(headless.calls).toEqual(["z:p1"]); // p2 dropped, never ran @@ -214,7 +242,7 @@ describe("WakeCoordinator — broken / resume-fail", () => { test("restart (onInteractiveUp) clears broken back to interactive", () => { const { reg, coord } = make(); reg.setExecutionState("w", "broken"); - coord.enqueue("w", "p1"); // creates the writer, seeded broken + coord.enqueue("w", w("p1")); // creates the writer, seeded broken expect(reg.executionStateOf("w")).toBe("broken"); coord.onInteractiveUp("w"); // manual Restart spawns a fresh PTY expect(reg.executionStateOf("w")).toBe("interactive"); @@ -223,6 +251,7 @@ describe("WakeCoordinator — broken / resume-fail", () => { test("going broken disarms the agent's scheduling; recovering re-arms it", () => { const { headless, coord } = make(); const sched = { + ...makeRecorder().sched, disarmed: [] as string[], rearmed: [] as string[], disarmAgent(id: string) { @@ -235,7 +264,7 @@ describe("WakeCoordinator — broken / resume-fail", () => { coord.setScheduler(sched); for (let i = 1; i <= 3; i++) { - coord.enqueue("b", `p${i}`); + coord.enqueue("b", w(`p${i}`)); headless.finishNext("resumeFail"); } expect(sched.disarmed).toEqual(["b"]); // paused exactly once, on the broken transition @@ -248,8 +277,8 @@ describe("WakeCoordinator — broken / resume-fail", () => { test("a spawn-fail broken also disarms scheduling", () => { const { coord, headless } = make(); const disarmed: string[] = []; - coord.setScheduler({ disarmAgent: (id) => disarmed.push(id), rearmAgent: () => {} }); - coord.enqueue("s", "p1"); + coord.setScheduler({ ...makeRecorder().sched, disarmAgent: (id) => disarmed.push(id) }); + coord.enqueue("s", w("p1")); headless.finishNext("spawnFail"); // one-strike broken expect(disarmed).toEqual(["s"]); }); @@ -259,12 +288,12 @@ describe("WakeCoordinator — headless turn limit", () => { test("each headless run consumes one turn; the run past the budget is dropped", () => { const { reg, headless, coord } = make(10_000, 2); reg.budgets.set("a", { turnLimitEnabled: true, headlessTurnsUsed: 0 }); - coord.enqueue("a", "p1"); + coord.enqueue("a", w("p1")); headless.finishNext("ok"); - coord.enqueue("a", "p2"); + coord.enqueue("a", w("p2")); headless.finishNext("ok"); expect(reg.budgets.get("a")!.headlessTurnsUsed).toBe(2); - coord.enqueue("a", "p3"); // budget spent → dropped, never spawned + coord.enqueue("a", w("p3")); // budget spent → dropped, never spawned expect(headless.calls).toEqual(["a:p1", "a:p2"]); expect(reg.executionStateOf("a")).toBe("offline"); // stays OFFLINE, not headless }); @@ -272,8 +301,8 @@ describe("WakeCoordinator — headless turn limit", () => { test("an exhausted budget also gates queued wakes draining after the active run", () => { const { reg, headless, coord } = make(10_000, 1); reg.budgets.set("q", { turnLimitEnabled: true, headlessTurnsUsed: 0 }); - coord.enqueue("q", "p1"); // consumes the only turn - coord.enqueue("q", "p2"); // queued behind the active run + coord.enqueue("q", w("p1")); // consumes the only turn + coord.enqueue("q", w("p2")); // queued behind the active run headless.finishNext("ok"); // drain → gate trips → p2 dropped expect(headless.calls).toEqual(["q:p1"]); expect(reg.executionStateOf("q")).toBe("offline"); @@ -282,10 +311,10 @@ describe("WakeCoordinator — headless turn limit", () => { test("a reset (the turn-limit button's Reset control) re-opens the budget", () => { const { reg, headless, coord } = make(10_000, 1); reg.budgets.set("v", { turnLimitEnabled: true, headlessTurnsUsed: 1 }); // spent - coord.enqueue("v", "p1"); + coord.enqueue("v", w("p1")); expect(headless.calls).toEqual([]); // gated reg.budgets.get("v")!.headlessTurnsUsed = 0; // = registry.resetHeadlessTurns (agents.resetTurnLimit) - coord.enqueue("v", "p2"); + coord.enqueue("v", w("p2")); expect(headless.calls).toEqual(["v:p2"]); headless.finishNext("ok"); }); @@ -293,7 +322,7 @@ describe("WakeCoordinator — headless turn limit", () => { test("a disabled per-agent toggle bypasses the limit", () => { const { reg, headless, coord } = make(10_000, 1); reg.budgets.set("d", { turnLimitEnabled: false, headlessTurnsUsed: 99 }); - coord.enqueue("d", "p1"); + coord.enqueue("d", w("p1")); expect(headless.calls).toEqual(["d:p1"]); headless.finishNext("ok"); expect(reg.budgets.get("d")!.headlessTurnsUsed).toBe(100); // still counted, never gated @@ -303,7 +332,7 @@ describe("WakeCoordinator — headless turn limit", () => { const { reg, headless, coord } = make(10_000, 1); reg.budgets.set("i", { turnLimitEnabled: true, headlessTurnsUsed: 5 }); // way past the limit coord.onInteractiveUp("i"); - coord.enqueue("i", "p1"); + coord.enqueue("i", w("p1")); const { p } = poll(coord, "i"); expect(await p).toBe("p1"); // delivered via the channel regardless of the budget expect(headless.calls).toEqual([]); @@ -330,8 +359,8 @@ describe("WakeCoordinator — headless turn limit", () => { test("the global feature switch off: never gated (but still counts — no freeze)", () => { const { reg, headless, coord } = make(10_000, 1, /* featureEnabled */ false); reg.budgets.set("g", { turnLimitEnabled: true, headlessTurnsUsed: 5 }); // past the limit - coord.enqueue("g", "p1"); - coord.enqueue("g", "p2"); // queued behind the active run + coord.enqueue("g", w("p1")); + coord.enqueue("g", w("p2")); // queued behind the active run expect(headless.calls).toEqual(["g:p1"]); // runs despite budget being spent headless.finishNext("ok"); expect(headless.calls).toEqual(["g:p1", "g:p2"]); // and drains the next, no gate @@ -345,8 +374,8 @@ describe("WakeCoordinator — headless turn limit", () => { describe("WakeCoordinator — stop", () => { test("stop() clears pending and ends an in-flight headless run", () => { const { reg, headless, coord } = make(); - coord.enqueue("i", "p1"); // headless run in flight - coord.enqueue("i", "p2"); // queued + coord.enqueue("i", w("p1")); // headless run in flight + coord.enqueue("i", w("p2")); // queued expect(coord.stop("i")).toBe(true); expect(headless.stops).toBe(1); headless.finishNext("ok"); // the SIGTERM'd child exits (treated as a deliberate stop) @@ -357,7 +386,7 @@ describe("WakeCoordinator — stop", () => { test("stop() on an interactive agent clears the queue and reports no headless run", async () => { const { headless, coord } = make(); coord.onInteractiveUp("j"); - coord.enqueue("j", "p1"); // queued (no poll) + coord.enqueue("j", w("p1")); // queued (no poll) expect(coord.stop("j")).toBe(false); // nothing headless to stop const { p } = poll(coord, "j", 10); expect(await p).toBeNull(); // queue cleared → a fresh poll parks, then the hold elapses @@ -387,7 +416,7 @@ describe("WakeCoordinator — push transport (codex interactive)", () => { const { headless, coord } = make(); const { push, delivered, settle } = makePush(); coord.onInteractiveUp("a", push); - coord.enqueue("a", "w1"); + coord.enqueue("a", w("w1")); expect(delivered).toEqual(["w1"]); // Not acked yet — a PTY drop now must re-route the (still-queued) head. settle(true); @@ -399,8 +428,8 @@ describe("WakeCoordinator — push transport (codex interactive)", () => { const { coord } = make(); const { push, delivered, settle } = makePush(); coord.onInteractiveUp("a", push); - coord.enqueue("a", "w1"); - coord.enqueue("a", "w2"); + coord.enqueue("a", w("w1")); + coord.enqueue("a", w("w2")); expect(delivered).toEqual(["w1"]); // one in flight at a time settle(true); await tick(); @@ -412,7 +441,7 @@ describe("WakeCoordinator — push transport (codex interactive)", () => { const { coord } = make(); const { push, delivered, settle } = makePush(); coord.onInteractiveUp("a", push); - coord.enqueue("a", "w1"); + coord.enqueue("a", w("w1")); settle(false); await tick(); expect(delivered).toEqual(["w1"]); // not retried yet (5s backoff) @@ -425,7 +454,7 @@ describe("WakeCoordinator — push transport (codex interactive)", () => { const { headless, coord } = make(); const { push, settle } = makePush(); coord.onInteractiveUp("a", push); - coord.enqueue("a", "w1"); + coord.enqueue("a", w("w1")); coord.onInteractiveDown("a"); // TUI died before the ack expect(headless.calls).toEqual(["a:w1"]); // head re-routed, not lost settle(true); // late ack from the dead session must not double-deliver @@ -438,7 +467,7 @@ describe("WakeCoordinator — push transport (codex interactive)", () => { const { push, delivered, settle } = makePush(); coord.onInteractiveUp("a", push); const { p, ac } = poll(coord, "a", 50); - coord.enqueue("a", "w1"); + coord.enqueue("a", w("w1")); expect(delivered).toEqual(["w1"]); // push got it… settle(true); expect(await p).toBeNull(); // …the poll parked inertly and timed out empty @@ -449,7 +478,7 @@ describe("WakeCoordinator — push transport (codex interactive)", () => { const { coord } = make(); coord.onInteractiveUp("a"); // no push: channel transport const { p } = poll(coord, "a"); - coord.enqueue("a", "w1"); + coord.enqueue("a", w("w1")); expect(await p).toBe("w1"); }); @@ -457,7 +486,7 @@ describe("WakeCoordinator — push transport (codex interactive)", () => { const { coord } = make(); const first = makePush(); coord.onInteractiveUp("a", first.push); - coord.enqueue("a", "w1"); + coord.enqueue("a", w("w1")); expect(first.delivered).toEqual(["w1"]); // in flight on the first push, not yet acked // A respawn-while-interactive installs a FRESH push before the first one settled @@ -474,7 +503,192 @@ describe("WakeCoordinator — push transport (codex interactive)", () => { second.settle(true); await tick(); // A follow-up wake still delivers, proving the queue isn't wedged. - coord.enqueue("a", "w2"); + coord.enqueue("a", w("w2")); expect(second.delivered).toEqual(["w1", "w2"]); }); }); + +describe("WakeCoordinator — History recording (wakeStarted / wakeFinished)", () => { + test("a headless run is recorded when it starts and settled succeeded on a clean exit", () => { + const { headless, coord } = make(); + const { sched, started, finished } = makeRecorder(); + coord.setScheduler(sched); + const wake = w("p1"); + coord.enqueue("a", wake); + expect(started).toEqual([{ prompt: "p1", background: true }]); + expect(finished).toEqual([]); // still running + headless.finishNext("ok"); + expect(finished).toEqual([{ wakeId: wake.id, result: { outcome: "succeeded" } }]); + }); + + test("a resume failure settles failed with its reason + classified category", () => { + const { headless, coord } = make(); + const { sched, finished } = makeRecorder(); + coord.setScheduler(sched); + const wake = w("p1"); + coord.enqueue("a", wake); + headless.finishNext("resumeFail", "billing"); + expect(finished).toEqual([ + { + wakeId: wake.id, + result: { outcome: "failed", failureReason: "resume_fail", failureCategory: "billing" }, + }, + ]); + }); + + test("a spawn failure settles failed (spawn_fail) before the agent goes broken", () => { + const { headless, coord } = make(); + const { sched, finished } = makeRecorder(); + coord.setScheduler(sched); + const wake = w("p1"); + coord.enqueue("a", wake); + headless.finishNext("spawnFail"); + expect(finished[0]).toEqual({ + wakeId: wake.id, + result: { outcome: "failed", failureReason: "spawn_fail", failureCategory: undefined }, + }); + }); + + test("a user Stop mid-run settles the head as stopped", () => { + const { headless, coord } = make(); + const { sched, finished } = makeRecorder(); + coord.setScheduler(sched); + const wake = w("p1"); + coord.enqueue("a", wake); + expect(coord.stop("a")).toBe(true); + headless.finishNext("ok"); // the SIGTERM'd child's exit + expect(finished).toEqual([{ wakeId: wake.id, result: { outcome: "stopped" } }]); + }); + + test("a wake queued behind a run is recorded only when ITS run starts", () => { + const { headless, coord } = make(); + const { sched, started } = makeRecorder(); + coord.setScheduler(sched); + coord.enqueue("a", w("p1")); + coord.enqueue("a", w("p2")); // queued behind the active run + expect(started.map((s) => s.prompt)).toEqual(["p1"]); + headless.finishNext("ok"); + expect(started.map((s) => s.prompt)).toEqual(["p1", "p2"]); + }); + + test("a wake dropped for an exhausted turn budget is never recorded", () => { + const { reg, headless, coord } = make(10_000, 1); + reg.budgets.set("a", { turnLimitEnabled: true, headlessTurnsUsed: 0 }); + const { sched, started } = makeRecorder(); + coord.setScheduler(sched); + coord.enqueue("a", w("p1")); // consumes the only turn + headless.finishNext("ok"); + coord.enqueue("a", w("p2")); // budget spent → dropped, never spawned + expect(started.map((s) => s.prompt)).toEqual(["p1"]); + expect(headless.calls).toEqual(["a:p1"]); + }); + + test("a warm (channel) wake is recorded on handoff and settled succeeded by the Stop hook", async () => { + const { coord } = make(); + const { sched, started, finished } = makeRecorder(); + coord.setScheduler(sched); + coord.onInteractiveUp("a"); + const { p } = poll(coord, "a"); + const wake = w("p1"); + coord.enqueue("a", wake); + expect(await p).toBe("p1"); + expect(started).toEqual([{ prompt: "p1", background: false }]); + expect(finished).toEqual([]); // the turn hasn't ended + coord.onTurnEnded("a"); + expect(finished).toEqual([{ wakeId: wake.id, result: { outcome: "succeeded" } }]); + coord.onTurnEnded("a"); // a later user turn: nothing outstanding, no double settle + expect(finished).toHaveLength(1); + }); + + test("a warm wake whose session goes away before its turn ends settles stopped", async () => { + const { headless, coord } = make(); + const { sched, finished } = makeRecorder(); + coord.setScheduler(sched); + coord.onInteractiveUp("a"); + const { p } = poll(coord, "a"); + const wake = w("p1"); + coord.enqueue("a", wake); + await p; + coord.onInteractiveDown("a"); + expect(finished).toEqual([{ wakeId: wake.id, result: { outcome: "stopped" } }]); + expect(headless.calls).toEqual([]); // already handed off; nothing re-routes + }); + + test("StopFailure settles an outstanding warm wake as failed with the hook's category", async () => { + const { coord } = make(); + const { sched, finished } = makeRecorder(); + coord.setScheduler(sched); + coord.onInteractiveUp("a"); + const { p } = poll(coord, "a"); + const wake = w("p1"); + coord.enqueue("a", wake); + await p; + coord.onTurnFailed("a", "billing"); + expect(finished).toEqual([ + { + wakeId: wake.id, + result: { outcome: "failed", failureReason: "api_error", failureCategory: "billing" }, + }, + ]); + coord.onTurnEnded("a"); // nothing outstanding any more + expect(finished).toHaveLength(1); + }); + + test("StopFailure during a headless run makes its otherwise-clean exit settle failed", () => { + const { headless, coord } = make(); + const { sched, finished } = makeRecorder(); + coord.setScheduler(sched); + const wake = w("p1"); + coord.enqueue("a", wake); + coord.onTurnFailed("a", "rate_limit"); // the hook lands before the child exits + expect(finished).toEqual([]); // held until exit + headless.finishNext("ok"); // past the fast-fail window, exit code alone says "ok" + expect(finished).toEqual([ + { + wakeId: wake.id, + result: { outcome: "failed", failureReason: "api_error", failureCategory: "rate_limit" }, + }, + ]); + // The held failure belongs to that child only: the next run starts clean. + coord.enqueue("a", w("p2")); + headless.finishNext("ok"); + expect(finished[1].result).toEqual({ outcome: "succeeded" }); + }); + + test("StopFailure's category wins over a fast resume-fail's stderr guess", () => { + const { headless, coord } = make(); + const { sched, finished } = makeRecorder(); + coord.setScheduler(sched); + const wake = w("p1"); + coord.enqueue("a", wake); + coord.onTurnFailed("a", "auth"); + headless.finishNext("resumeFail", "other"); + expect(finished).toEqual([ + { + wakeId: wake.id, + result: { outcome: "failed", failureReason: "resume_fail", failureCategory: "auth" }, + }, + ]); + }); + + test("StopFailure with nothing outstanding is a no-op", () => { + const { coord } = make(); + const { sched, finished } = makeRecorder(); + coord.setScheduler(sched); + coord.onTurnFailed("a", "billing"); // a user's own turn failed; no wake involved + expect(finished).toEqual([]); + }); + + test("a push (codex) wake is recorded once the app-server acks it", async () => { + const { coord } = make(); + const { sched, started } = makeRecorder(); + coord.setScheduler(sched); + const settles: Array<(ok: boolean) => void> = []; + coord.onInteractiveUp("a", () => new Promise((r) => settles.push(r))); + coord.enqueue("a", w("p1")); + expect(started).toEqual([]); // in flight, not yet accepted + settles.shift()?.(true); + await tick(); + expect(started).toEqual([{ prompt: "p1", background: false }]); + }); +}); diff --git a/app/src/main/services/scheduler/wake/coordinator.ts b/app/src/main/services/scheduler/wake/coordinator.ts index 8cc6379..236a01b 100644 --- a/app/src/main/services/scheduler/wake/coordinator.ts +++ b/app/src/main/services/scheduler/wake/coordinator.ts @@ -1,4 +1,5 @@ import type { ExecutionState } from "@shared/agent"; +import type { WakeFailureCategory } from "@shared/analytics"; import { DEFAULT_SETTINGS } from "@shared/settings"; import { hostLog } from "../../../host/log"; import type { AgentRegistry } from "../../agents/registry"; @@ -8,7 +9,9 @@ import type { HeadlessExitReason, HeadlessWakeStrategy, InteractivePush, + PendingWake, SchedulerControl, + WakeResult, WakeTransport, } from "./types"; @@ -75,12 +78,29 @@ interface AgentWriterDeps { * agent's crons + monitors (see {@link SchedulerControl}). No-op until a scheduler binds. */ onBroken: (id: string) => void; onUnbroken: (id: string) => void; + /** History recording (§12.2): a row is written when a wake actually STARTS (never at + * enqueue — a wake that is dropped or never drained leaves no row) and settled exactly + * once when it ends. No-ops until a scheduler binds. */ + recordStarted: (wake: PendingWake, background: boolean) => void; + recordFinished: (wake: PendingWake, result: WakeResult) => void; } class AgentWriter { private state: WriterState = "OFFLINE"; /** The one wake queue (FIFO). Advanced on handoff (interactive) or on exit (headless). */ - private pending: string[] = []; + private pending: PendingWake[] = []; + /** Warm wakes delivered into the live session whose turn hasn't ended yet. Settled + * `succeeded` by the Stop hook (`onTurnEnded`), `failed` by the StopFailure hook + * (`onTurnFailed`), or `stopped` if the session goes away. */ + private liveWakes: PendingWake[] = []; + /** A StopFailure that fired for the active `-p` child (`onTurnFailed` while + * HEADLESS_RUNNING). The hook lands BEFORE the child exits (the hook script curls the + * host in the foreground and the route settles synchronously), so it's held here and + * applied in `headlessExited` — an otherwise-clean exit settles `failed`, not + * `succeeded`. Cleared on every exit and before every spawn. Known limit: if the host + * took longer than the hook's 5 s curl timeout to answer, the hook is lost and the + * wake reads `succeeded`; a hook arriving after the exit would stamp the NEXT child. */ + private headApiError?: WakeFailureCategory; /** A currently-parked `/wake-stream` long-poll (one poller per agent), or undefined. */ private interactivePoll?: (prompt: string | null) => void; /** Non-channel interactive delivery (codex app-server push). While set, the parked @@ -104,6 +124,8 @@ class AgentWriter { private readonly turnLimitFeatureEnabled: () => boolean; private readonly onBroken: (id: string) => void; private readonly onUnbroken: (id: string) => void; + private readonly recordStarted: AgentWriterDeps["recordStarted"]; + private readonly recordFinished: AgentWriterDeps["recordFinished"]; constructor( private id: string, @@ -116,6 +138,8 @@ class AgentWriter { this.turnLimitFeatureEnabled = deps.turnLimitFeatureEnabled; this.onBroken = deps.onBroken; this.onUnbroken = deps.onUnbroken; + this.recordStarted = deps.recordStarted; + this.recordFinished = deps.recordFinished; // Seed BROKEN from a boot-time spawn-marker reconcile: single-writer crash recovery // sets `executionState = broken` directly, before this coordinator exists. (The // scheduler's own boot sweep skips arming a broken agent, so no disarm is needed @@ -126,18 +150,18 @@ class AgentWriter { // ---- producer / consumer ---- /** A wake was produced (cron/monitor fire). Route by state. */ - enqueue(prompt: string): void { + enqueue(wake: PendingWake): void { switch (this.state) { case "OFFLINE": - this.pending.push(prompt); + this.pending.push(wake); this.startHeadless(); break; case "INTERACTIVE_RUNNING": - this.pending.push(prompt); + this.pending.push(wake); this.serveInteractive(); // hand to a parked poll if one's waiting break; case "HEADLESS_RUNNING": - this.pending.push(prompt); // drains when the active child exits + this.pending.push(wake); // drains when the active child exits break; case "BROKEN": // Unresumable; drop. A recurring cron re-fires after a manual Restart. @@ -186,6 +210,8 @@ class AgentWriter { this.push = undefined; this.pushInFlight = false; this.clearPushRetry(); + // A warm wake whose turn hadn't ended was cut off with the session. + this.settleLive({ outcome: "stopped" }); // The live writer is gone; the head + any queued wakes re-route to the `-p` transport. this.transition("OFFLINE"); this.drain(); @@ -197,6 +223,7 @@ class AgentWriter { * interactive session itself is torn down by TerminalService, not here. */ stop(): boolean { this.pending = []; + this.settleLive({ outcome: "stopped" }); if (this.interactivePoll) { const poll = this.interactivePoll; this.interactivePoll = undefined; @@ -231,7 +258,8 @@ class AgentWriter { if (!this.interactivePoll || this.pending.length === 0) return; const poll = this.interactivePoll; const head = this.pending.shift()!; - poll(head); // resolves the parked /wake-stream long-poll; finish() clears the slot + this.deliveredLive(head); + poll(head.prompt); // resolves the parked /wake-stream long-poll; finish() clears the slot } /** Push-mode delivery: one in-flight push at a time; advance-on-ack; a failed @@ -242,13 +270,13 @@ class AgentWriter { if (!push) return; const head = this.pending[0]; this.pushInFlight = true; - push(head).then( + push(head.prompt).then( (ok) => this.pushSettled(push, head, ok), () => this.pushSettled(push, head, false), ); } - private pushSettled(push: InteractivePush, head: string, ok: boolean): void { + private pushSettled(push: InteractivePush, head: PendingWake, ok: boolean): void { // A newer push replaced this one (respawn-while-interactive installed a fresh push // via onInteractiveUp): the current in-flight state belongs to THAT push, so a stale // settle must not clear its `pushInFlight` (which would let a duplicate delivery @@ -259,7 +287,10 @@ class AgentWriter { // the head then belongs to whatever transport took over; don't touch it here. if (this.state !== "INTERACTIVE_RUNNING") return; if (ok) { - if (this.pending[0] === head) this.pending.shift(); + if (this.pending[0] === head) { + this.pending.shift(); + this.deliveredLive(head); // the app-server accepted the turn + } this.serveInteractive(); // deliver the next queued wake, if any return; } @@ -294,6 +325,7 @@ class AgentWriter { } this.transition("HEADLESS_RUNNING"); this.armKillTimer(); + this.headApiError = undefined; // belongs to the previous child, if ever set // Always count the run (no freeze while the feature is off — the count is reset // wholesale when the feature is re-enabled, so there's nothing to preserve). The // pause NOTIFICATION only makes sense when the feature + the agent's switch are on @@ -317,7 +349,10 @@ class AgentWriter { }); } const head = this.pending[0]; // kept at the head until exit (no lost wake on crash) - this.headless.run(this.id, head, (reason) => this.headlessExited(reason)); + this.recordStarted(head, true); + this.headless.run(this.id, head.prompt, (reason, failureCategory) => + this.headlessExited(head, reason, failureCategory), + ); } /** True when the turn-limit feature is on globally, the agent's own switch is on, and @@ -342,22 +377,43 @@ class AgentWriter { return this.turnBudgetExhausted(); } - private headlessExited(reason: HeadlessExitReason): void { + /** The `-p` child for `head` ended. Settles its History row, then routes by reason. */ + private headlessExited( + head: PendingWake, + reason: HeadlessExitReason, + failureCategory?: WakeFailureCategory, + ): void { this.clearKillTimer(); + // A StopFailure hook that fired for this child (see `headApiError`): the turn ended + // in an API error, whatever the exit code says. Its category is the authoritative + // one — the hook's structured `error` beats a stderr-tail guess. + const apiError = this.headApiError; + this.headApiError = undefined; if (this.stopping) { // A deliberate user Stop killed the child — don't count it as a resume failure. this.stopping = false; + this.recordFinished(head, { outcome: "stopped" }); this.transition("OFFLINE"); this.drain(); // in case a fresh wake arrived during the stop window return; } if (reason === "spawnFail") { // A spawn error is a config fault, not a flaky session: one-strike broken. + this.recordFinished(head, { + outcome: "failed", + failureReason: "spawn_fail", + failureCategory, + }); this.pending = []; this.transition("BROKEN"); return; } if (reason === "resumeFail") { + this.recordFinished(head, { + outcome: "failed", + failureReason: "resume_fail", + failureCategory: apiError ?? failureCategory, + }); this.pending.shift(); // drop the failed wake this.resumeFailCount += 1; if (this.resumeFailCount >= MAX_RESUME_FAILS) { @@ -370,12 +426,49 @@ class AgentWriter { return; } // ok (clean exit, or the max-runtime backstop): complete the head, drain the next. + // An API-error turn exits "ok" too (past the fast-fail window) — StopFailure is what + // tells them apart. Routing is unchanged either way: the wake is consumed, not retried. + this.recordFinished( + head, + apiError + ? { outcome: "failed", failureReason: "api_error", failureCategory: apiError } + : { outcome: "succeeded" }, + ); this.resumeFailCount = 0; this.pending.shift(); this.transition("OFFLINE"); this.drain(); } + // ---- history recording ---- + + /** A warm wake entered the live session: record it started and hold it until the + * turn ends (`onTurnEnded`) or the session goes away (`onInteractiveDown` / `stop`). */ + private deliveredLive(wake: PendingWake): void { + this.recordStarted(wake, false); + this.liveWakes.push(wake); + } + + /** The Stop hook fired: the session finished a turn, so every warm wake delivered so + * far has been consumed — settle them all. (A user message typed before the wake's + * turn ends fires Stop too and settles early; accepted — see docs/TODO.md.) */ + onTurnEnded(): void { + this.settleLive({ outcome: "succeeded" }); + } + + /** The StopFailure hook fired (instead of Stop): the turn died on an API error. Warm + * wakes settle `failed` now; a running `-p` child's failure is held until its exit. */ + onTurnFailed(failureCategory: WakeFailureCategory): void { + this.settleLive({ outcome: "failed", failureReason: "api_error", failureCategory }); + if (this.state === "HEADLESS_RUNNING") this.headApiError = failureCategory; + } + + private settleLive(result: WakeResult): void { + const live = this.liveWakes; + this.liveWakes = []; + for (const w of live) this.recordFinished(w, result); + } + private armKillTimer(): void { this.clearKillTimer(); this.headlessKillTimer = setTimeout(() => { @@ -461,14 +554,25 @@ export class WakeCoordinator implements WakeTransport { turnLimitFeatureEnabled: this.turnLimitFeatureEnabled, onBroken: (aid) => this.scheduler?.disarmAgent(aid), onUnbroken: (aid) => this.scheduler?.rearmAgent(aid), + recordStarted: (wake, background) => this.scheduler?.wakeStarted(wake, background), + recordFinished: (wake, result) => this.scheduler?.wakeFinished(wake, result), }); this.writers.set(id, w); } return w; } - enqueue(agentId: string, prompt: string): void { - this.writer(agentId).enqueue(prompt); + enqueue(agentId: string, wake: PendingWake): void { + this.writer(agentId).enqueue(wake); + } + + onTurnEnded(agentId: string): void { + // No writer ⇒ no warm wake was ever delivered; a user-turn Stop is a no-op. + this.writers.get(agentId)?.onTurnEnded(); + } + + onTurnFailed(agentId: string, failureCategory: WakeFailureCategory): void { + this.writers.get(agentId)?.onTurnFailed(failureCategory); } wouldDropWake(agentId: string): boolean { diff --git a/app/src/main/services/scheduler/wake/failure-category.test.ts b/app/src/main/services/scheduler/wake/failure-category.test.ts index fa1f9f4..c45004d 100644 --- a/app/src/main/services/scheduler/wake/failure-category.test.ts +++ b/app/src/main/services/scheduler/wake/failure-category.test.ts @@ -1,5 +1,5 @@ import { describe, expect, test } from "bun:test"; -import { classifyWakeFailure } from "./failure-category"; +import { categoryForStopFailure, classifyWakeFailure } from "./failure-category"; describe("classifyWakeFailure", () => { test("unresumable-session lines from both harnesses", () => { @@ -46,26 +46,83 @@ describe("classifyWakeFailure", () => { test("unrecognized text fails closed to other", () => { expect(classifyWakeFailure("")).toBe("other"); expect(classifyWakeFailure("segmentation fault")).toBe("other"); - expect(classifyWakeFailure("TypeError: fetch failed")).toBe("other"); - expect(classifyWakeFailure("spawn claude ENOENT")).toBe("other"); + expect(classifyWakeFailure("spawn claude ENOENT")).toBe("other"); // a missing binary, not the network }); test("digit substrings and near-miss words don't fake a category", () => { // Status codes are digit-bounded: durations, request ids, and byte counts in the // 2000-char tail are full of 401/402/429 substrings that are not HTTP statuses. - expect(classifyWakeFailure("Request timed out after 40200ms")).toBe("other"); + expect(classifyWakeFailure("Request timed out after 40200ms")).toBe("network"); // the 402 inside 40200 is not billing expect(classifyWakeFailure("stream error: retrying in 4013ms")).toBe("other"); + // The 429 inside the request id is not a rate limit; the 500 IS a server error. expect(classifyWakeFailure("request_id: req_011CR4291abc failed with status 500")).toBe( - "other", + "network", ); // "logging"/"dialog" must not read as "login". expect(classifyWakeFailure("Error while logging to stderr file")).toBe("other"); expect(classifyWakeFailure("error dialog initialization failed")).toBe("other"); }); + test("network: transport codes, fetch failures, claude's connection wording, 5xx", () => { + expect(classifyWakeFailure("TypeError: fetch failed")).toBe("network"); + expect(classifyWakeFailure("Error: connect ECONNREFUSED 127.0.0.1:443")).toBe("network"); + expect(classifyWakeFailure("getaddrinfo ENOTFOUND api.anthropic.com")).toBe("network"); + expect( + classifyWakeFailure( + "API Error: Connection refused — a firewall or proxy may be blocking it (ConnectionRefused)", + ), + ).toBe("network"); + expect(classifyWakeFailure("API Error: 503 overloaded_error")).toBe("rate_limit"); // overloaded wins + expect(classifyWakeFailure("API Error: 502 Bad Gateway")).toBe("network"); + expect(classifyWakeFailure("API Error: 522 origin connection time-out")).toBe("network"); // Cloudflare edge + expect(classifyWakeFailure("request timed out after 60000ms")).toBe("network"); + // Word-bounded: identifiers and near-miss words containing the patterns don't count. + expect(classifyWakeFailure("TIMEOUT_MS=300 budget exceeded")).toBe("other"); + expect(classifyWakeFailure("HOMECONNRESET=1")).toBe("other"); + expect(classifyWakeFailure("waited 500s for approval")).toBe("other"); + // Digit-bounded AND not a duration: "15020ms" and "500ms" are not statuses. + expect(classifyWakeFailure("retrying in 15020ms")).toBe("other"); + expect(classifyWakeFailure("retrying in 500ms")).toBe("other"); + expect(classifyWakeFailure("took 503 ms")).toBe("other"); + }); + test("the fatal line wins over preceding noise lines in a multi-line tail", () => { const tail = "[debug] logging initialized\n[warn] retrying in 4013ms\nAPI Error: 429 rate_limit_error"; expect(classifyWakeFailure(tail)).toBe("rate_limit"); }); }); + +describe("categoryForStopFailure", () => { + test("maps Claude Code's StopFailure `error` enum onto the coarse category", () => { + expect(categoryForStopFailure("billing_error")).toBe("billing"); + expect(categoryForStopFailure("rate_limit")).toBe("rate_limit"); + expect(categoryForStopFailure("overloaded")).toBe("rate_limit"); + for (const e of [ + "authentication_failed", + "oauth_org_not_allowed", + "account_on_hold", + "verification_required", + "cloud_credential_error", + ]) { + expect(categoryForStopFailure(e)).toBe("auth"); + } + }); + + test("server_error (a 5xx or a connection failure) reads as `network`", () => { + expect(categoryForStopFailure("server_error")).toBe("network"); + }); + + test("everything the category can't express reads as `other`", () => { + for (const e of [ + "invalid_request", + "model_not_found", + "max_output_tokens", + "unknown", + "", // hook payload without an error field + "some_future_value", + ]) { + expect(categoryForStopFailure(e)).toBe("other"); + } + }); +}); diff --git a/app/src/main/services/scheduler/wake/failure-category.ts b/app/src/main/services/scheduler/wake/failure-category.ts index a2ea4f1..23ae767 100644 --- a/app/src/main/services/scheduler/wake/failure-category.ts +++ b/app/src/main/services/scheduler/wake/failure-category.ts @@ -1,8 +1,35 @@ import type { WakeFailureCategory } from "@shared/analytics"; +/** + * Map Claude Code's `StopFailure` hook `error` value (the turn ended in an API error) + * onto the same coarse category. The hook's enum is finer than ours: the credential + * states all read as `auth`, throttling/overload as `rate_limit`, `server_error` (a 5xx + * OR a connection failure — Claude doesn't distinguish) as `network`, and what's left — + * bad requests, unknown model, output cap, unknown — as `other`. + */ +export function categoryForStopFailure(error: string): WakeFailureCategory { + switch (error) { + case "billing_error": + return "billing"; + case "rate_limit": + case "overloaded": + return "rate_limit"; + case "server_error": + return "network"; + case "authentication_failed": + case "oauth_org_not_allowed": + case "account_on_hold": + case "verification_required": + case "cloud_credential_error": + return "auth"; + default: + return "other"; + } +} + /** * Classify a failed headless run's error text into the coarse `WakeFailureCategory` - * that ships on `headless_run_finished`. The input — the claude CLI's stderr tail or + * that ships on `wake_finished`. The input — the claude CLI's stderr tail or * a codex turn error — never leaves the machine (it lands in the host log); only the * returned category is tracked, so the patterns here can afford to be broad. * @@ -39,5 +66,18 @@ export function classifyWakeFailure(text: string): WakeFailureCategory { ) { return "auth"; } + // The API never answered: undici/Node transport codes (word-bounded both sides), + // "fetch failed", claude's own "Connection refused/reset" phrasing, a whole-word + // "timed out"/"timeout" (not TIMEOUT_MS or "timeouts"), and 5xx statuses incl. the + // Cloudflare 520–524 origin errors the API's edge emits. The status is digit-bounded + // like the codes above AND must not be a duration ("500ms", "500s"), which is far more + // common in a stderr tail than a bare 5xx. + if ( + /fetch failed|connection (refused|reset|closed|error)|socket hang up|network error|\bE(CONNREFUSED|CONNRESET|NOTFOUND|AI_AGAIN|TIMEDOUT|HOSTUNREACH|NETUNREACH)\b|UND_ERR_|\btimed? ?out\b|(? boolean = () => true, + private subscriptionAuthEnabled: () => boolean = () => true, ) {} /** EC1 "Stop task" / max-runtime kill: SIGTERM the running headless child; its exit @@ -83,7 +82,7 @@ export class HeadlessRunStrategy implements HeadlessWakeStrategy { this.children.clear(); } - run(agentId: string, prompt: string, onExit: (reason: HeadlessExitReason) => void): void { + run(agentId: string, prompt: string, onExit: HeadlessExit): void { const agent = this.registry.get(agentId); // Archived/missing agent — nothing to run; report a clean completion so the // coordinator releases the head and returns to OFFLINE. @@ -116,7 +115,7 @@ export class HeadlessRunStrategy implements HeadlessWakeStrategy { // needs no prefix: it self-identifies as ``. const startedAt = Date.now(); const wakePrompt = formatWakePrompt(prompt, startedAt); - const stripEnvKeys = this.useSubscriptionAuth() ? harness.subscriptionAuthStrip : []; + const stripEnvKeys = this.subscriptionAuthEnabled() ? harness.subscriptionAuthStrip : []; // This strategy is the CLI-child transport; the routing strategy only sends it // harnesses whose headless transport IS a CLI child (i.e. defines headlessArgs). if (!harness.headlessArgs) { @@ -152,17 +151,14 @@ export class HeadlessRunStrategy implements HeadlessWakeStrategy { // Report the outcome exactly once (error and exit can't both meaningfully fire). let settled = false; - const settle = (reason: HeadlessExitReason, failureCategory?: WakeFailureCategory) => { + const settle: HeadlessExit = (reason, failureCategory) => { if (settled) return; settled = true; this.children.delete(agentId); clearSpawnMarker(agentId); - analytics.track("headless_run_finished", { - result: reason === "ok" ? "ok" : reason === "resumeFail" ? "resume_fail" : "spawn_fail", - duration_ms: Math.max(0, Date.now() - startedAt), - ...(failureCategory ? { failure_category: failureCategory } : {}), - }); - onExit(reason); + // The outcome is tracked once the coordinator settles the wake (`wake_finished`, + // §12.2) — it sees the StopFailure hook too, which this exit site can't. + onExit(reason, failureCategory); }; child.on("error", (err) => { diff --git a/app/src/main/services/scheduler/wake/types.ts b/app/src/main/services/scheduler/wake/types.ts index 77d26e3..6f2265e 100644 --- a/app/src/main/services/scheduler/wake/types.ts +++ b/app/src/main/services/scheduler/wake/types.ts @@ -1,3 +1,6 @@ +import type { WakeFailureCategory } from "@shared/analytics"; +import type { WakeFailureReason, WakeOutcome } from "@shared/schedule"; + /** * The wake-delivery seam. `Scheduler` enqueues via `enqueue`; the `/wake-stream` * long-poll consumes via `awaitPoll`; `TerminalService` reports PTY up/down. The @@ -7,8 +10,18 @@ */ export interface WakeTransport { /** Enqueue a wake for an agent. Drained via the channel (a live PTY exists) or a - * headless `-p` child (none). Never blocks, never throws. */ - enqueue(agentId: string, prompt: string): void; + * headless `-p` child (none). Never blocks, never throws. Nothing is recorded here: + * the history row is written by the coordinator the moment the wake actually + * starts (`SchedulerControl.wakeStarted`), so a wake that never runs leaves no row. */ + enqueue(agentId: string, wake: PendingWake): void; + /** The agent's turn ended (the Stop status hook, §6.7). Settles every warm wake + * delivered into the live session as `succeeded`. No-op when none is outstanding. */ + onTurnEnded(agentId: string): void; + /** The agent's turn ended in an API error (Claude Code's `StopFailure` hook, which + * fires INSTEAD of Stop). Settles every outstanding warm wake as `failed`; for a + * headless run still in flight, the failure is held so the child's exit settles + * `failed` rather than `succeeded`. No-op when nothing is outstanding. */ + onTurnFailed(agentId: string, failureCategory: WakeFailureCategory): void; /** Would a wake for this agent be dropped rather than delivered (session BROKEN, or * out of background turns)? The Scheduler checks this to skip firing a paused agent — * no wake notification, history row, or enqueue — instead of firing then dropping. */ @@ -47,8 +60,38 @@ export interface SchedulerControl { disarmAgent(agentId: string): void; /** Re-arm this agent's still-enabled crons + monitors (Restart / recovery). */ rearmAgent(agentId: string): void; + /** The wake actually started: a headless child spawned, or the live session accepted + * it. Writes the History row (`outcome = running`) + the wake notification. */ + wakeStarted(wake: PendingWake, background: boolean): void; + /** The started wake settled. Called exactly once per `wakeStarted`. */ + wakeFinished(wake: PendingWake, result: WakeResult): void; +} + +/** A wake produced by the scheduler, carried through the coordinator's queue. `id` is + * minted at fire time and becomes the History row's id once the wake starts. */ +export interface PendingWake { + id: string; + agentId: string; + prompt: string; + sourceKind: "cron" | "monitor"; + /** The originating schedule/monitor id (joined with `sourceKind`). */ + sourceId: string; } +/** How a started wake settled (see `WakeOutcome` in `@shared/schedule`). */ +export interface WakeResult { + outcome: Exclude; + failureReason?: WakeFailureReason; + failureCategory?: WakeFailureCategory; +} + +/** A headless strategy's exit report: how the child ended and, for a failure, the + * category classified from its error text (when one was recognized). */ +export type HeadlessExit = ( + reason: HeadlessExitReason, + failureCategory?: WakeFailureCategory, +) => void; + /** * Interactive wake delivery for a non-channel harness (codex): deliver the RAW wake * prompt into the live session (the implementation prefixes/format as needed). @@ -68,7 +111,7 @@ export interface HeadlessWakeStrategy { /** Spawn the headless child for the head wake. Reports its terminal outcome via * `onExit` (called exactly once). Never blocks — the run is fire-and-forget; the * coordinator owns the max-runtime kill timer. */ - run(agentId: string, prompt: string, onExit: (reason: HeadlessExitReason) => void): void; + run(agentId: string, prompt: string, onExit: HeadlessExit): void; /** SIGTERM the active headless run for an agent, if any. Returns whether one died. */ stop(agentId: string): boolean; /** SIGTERM every live headless child + clear its marker (clean host shutdown). */ diff --git a/app/src/main/trpc/routers/feedback.ts b/app/src/main/trpc/routers/feedback.ts index 46a8cf4..9e87a66 100644 --- a/app/src/main/trpc/routers/feedback.ts +++ b/app/src/main/trpc/routers/feedback.ts @@ -1,6 +1,10 @@ import { FeedbackInput } from "@shared/feedback"; import { analytics } from "../../services/analytics"; -import { buildDiagnostics, cliVersionOf } from "../../services/feedback/diagnostics"; +import { + buildDiagnostics, + cliVersionOf, + WAKE_STATS_WINDOW_MS, +} from "../../services/feedback/diagnostics"; import { harnessFor } from "../../services/harness"; import { buildAgentEnv } from "../../services/terminal/env"; import { publicProcedure, router } from "../trpc"; @@ -22,6 +26,7 @@ export const feedbackRouter = router({ agents: ctx.registry.list(), crons: ctx.scheduler.listAllCron().length, monitors: ctx.scheduler.listAllMonitors().length, + wakes: ctx.scheduler.wakeStats(Date.now() - WAKE_STATS_WINDOW_MS), brokerStatus: ctx.broker.getStatus(), brokerAuthorized: ctx.broker.isAuthorized(), portfolioFetchedAt: ctx.broker.getCachedPortfolio()?.fetchedAt ?? null, diff --git a/app/src/main/trpc/routers/schedule.ts b/app/src/main/trpc/routers/schedule.ts index 2f06d61..a1afe8b 100644 --- a/app/src/main/trpc/routers/schedule.ts +++ b/app/src/main/trpc/routers/schedule.ts @@ -11,10 +11,11 @@ export const scheduleRouter = router({ monitors: ctx.scheduler.listAllMonitors(), })), - /** One agent's upcoming schedules/monitors + recorded wakes, for the Run History pane. */ + /** One agent's schedules/monitors (**retired included** — the panel filters `enabled` + * for its Active list and resolves History rows' triggers from the same arrays) + + * its recorded wakes, for the Monitor tab. */ forAgent: publicProcedure.input(z.object({ agentId: z.string() })).query(({ ctx, input }) => ({ - schedules: ctx.scheduler.listCron(input.agentId), - monitors: ctx.scheduler.listMonitors(input.agentId), + ...ctx.scheduler.listTriggers(input.agentId), wakes: ctx.scheduler.listWakes(input.agentId), })), diff --git a/app/src/renderer/components/panels/Monitor.tsx b/app/src/renderer/components/panels/Monitor.tsx index 269a26f..214d01f 100644 --- a/app/src/renderer/components/panels/Monitor.tsx +++ b/app/src/renderer/components/panels/Monitor.tsx @@ -1,5 +1,13 @@ +import type { WakeFailureCategory } from "@shared/analytics"; import type { Monitor, Schedule, Wake } from "@shared/schedule"; -import { ChevronRight, Clock, type LucideIcon, Radio } from "lucide-react"; +import { + ChevronRight, + CircleAlert, + Clock, + LoaderCircle, + type LucideIcon, + Radio, +} from "lucide-react"; import { useState } from "react"; import { useMonitor } from "../../hooks/useSchedules"; import { ago, cronZone, dateTime, describeCron, until } from "../../lib/format"; @@ -17,13 +25,16 @@ import { Badge } from "../ui/badge"; /** This agent's scheduled runs and the wakes they've fired. */ export function MonitorPanel() { const agentId = useUIStore((s) => s.selectedAgentId) ?? undefined; + // `schedules`/`monitors` include RETIRED rows so History can still resolve a fire's + // trigger; Active shows only the live ones. const { schedules, monitors, wakes } = useMonitor(agentId); const [historyOpen, setHistoryOpen] = useState(true); // Soonest-first so the next wake leads; never-scheduled (null) sinks to the bottom. - const upcomingCrons = [...schedules].sort( - (a, b) => (a.nextFireAt ?? Infinity) - (b.nextFireAt ?? Infinity), - ); - const hasUpcoming = upcomingCrons.length > 0 || monitors.length > 0; + const upcomingCrons = schedules + .filter((s) => s.enabled) + .sort((a, b) => (a.nextFireAt ?? Infinity) - (b.nextFireAt ?? Infinity)); + const liveMonitors = monitors.filter((m) => m.enabled); + const hasUpcoming = upcomingCrons.length > 0 || liveMonitors.length > 0; if (!agentId) { return

Select an agent.

; @@ -37,7 +48,7 @@ export function MonitorPanel() { {upcomingCrons.map((s) => ( ))} - {monitors.map((m) => ( + {liveMonitors.map((m) => ( ))} @@ -60,7 +71,7 @@ export function MonitorPanel() { (wakes.length > 0 ? (
{wakes.map((w) => ( - + ))}
) : ( @@ -257,35 +268,146 @@ function DetailGrid({ rows }: { rows: DetailRow[] }) { ); } -/** A recorded wake: which trigger fired, its prompt, and how long ago. */ -function WakeRow({ wake }: { wake: Wake }) { +/** Placeholder for a detail that can't be resolved. */ +const DASH =

—

; + +/** + * A recorded wake: which trigger kind fired, how it went (the state icon), and how long + * ago. Expands to the fire's details and the timer/monitor that fired it, looked up by + * `sourceId` in the agent's (retired-inclusive) `schedules`/`monitors` — a retired + * trigger still resolves (§12.2 retire-not-delete). Title stays "Timer fired" / + * "Monitor fired" so the list scans by kind; the trigger's own name is in the body. + */ +function WakeRow({ + wake, + schedules, + monitors, +}: { + wake: Wake; + schedules: Schedule[]; + monitors: Monitor[]; +}) { + const [open, setOpen] = useState(false); const isMonitor = wake.sourceKind === "monitor"; + const marker = wakeMarker(wake); + // Undefined for pre-link rows (no sourceId) or a source hard-deleted by the boot + // orphan self-heal — rendered as a dash below. + const schedule = isMonitor ? undefined : schedules.find((s) => s.id === wake.sourceId); + const monitor = isMonitor ? monitors.find((m) => m.id === wake.sourceId) : undefined; return ( -
- - {isMonitor ? ( - - ) : ( - - )} - -
-
- - {wake.sourceKind === "monitor" ? "Monitor fired" : "Timer fired"} - - {wake.background && ( - - Background - +
+ + {open && ( +
+ + {/* The trigger itself — just its prompt (timer) or command (monitor); the + cadence/run timestamps live on the Active row, not here. */} + {isMonitor ? ( + <> + {monitor?.description && ( + +

{monitor.description}

+ + )} + + {monitor ? {monitor.command} : DASH} + + + ) : ( + + {schedule ? {schedule.prompt} : DASH} + )}
- {ago(wake.firedAt)} -
+ )}
); } +/** + * The row's leading icon reflects the wake's state, not just its kind: an orange spinner + * while running, a red alert when failed, an orange alert when stopped. Only a settled + * success (or a pre-v7 row with no outcome) keeps the timer/monitor kind icon. + */ +function wakeMarker(wake: Wake): { icon: LucideIcon; tone: string } { + switch (wake.outcome) { + case "running": + return { icon: LoaderCircle, tone: "text-orange-400 animate-spin" }; + case "failed": + return { icon: CircleAlert, tone: "text-destructive" }; + case "stopped": + return { icon: CircleAlert, tone: "text-orange-400" }; + default: + return wake.sourceKind === "monitor" + ? { icon: Radio, tone: "text-emerald-400" } + : { icon: Clock, tone: "text-sky-400" }; + } +} + +/** Hint labels for a recognized failure category. `other` (unclassified text) is + * deliberately absent — it reads as no hint, the same as a run with no error text. */ +const FAILURE_CATEGORY: Partial> = { + unknown_session: "Session not found", + billing: "Billing", + rate_limit: "Rate limit", + auth: "Authentication", + network: "Network", +}; + +/** The Outcome row: the state, with the classified failure category as its hint (the + * resume-vs-spawn reason is stored but deliberately not shown). */ +function outcomeRow(wake: Wake): DetailRow { + switch (wake.outcome) { + case "succeeded": + return ["Outcome", "Succeeded", null]; + case "failed": + return [ + "Outcome", + + Failed + , + (wake.failureCategory && FAILURE_CATEGORY[wake.failureCategory]) || null, + ]; + case "stopped": + return [ + "Outcome", + + Stopped + , + null, + ]; + case "running": + return [ + "Outcome", + + Running + , + null, + ]; + default: + return ["Outcome", "—", null]; // pre-v7 rows: no outcome was recorded + } +} + /** First non-empty line of a multi-line prompt, for compact one-line row titles. */ function firstLine(text: string): string { return ( diff --git a/app/src/renderer/components/settings/TelemetryOptOutDialog.tsx b/app/src/renderer/components/settings/TelemetryOptOutDialog.tsx index 3e3d90f..18b3afd 100644 --- a/app/src/renderer/components/settings/TelemetryOptOutDialog.tsx +++ b/app/src/renderer/components/settings/TelemetryOptOutDialog.tsx @@ -22,7 +22,7 @@ export const OPENTRADE_DISCORD_URL = "https://discord.gg/F63YFPRtq"; * `notification_clicked`, `broker_connected|broker_connect_failed`, `feedback_sent` * (usage only — the feedback message itself is not telemetry, §12.8). * 2. agents — `agent_created|archived|restarted`, `terminal_session_started|respawned`, - * `schedule_created|fired`, `headless_run_finished`, `agent_marked_broken`, + * `schedule_created|fired`, `wake_finished`, `agent_marked_broken`, * `turn_limit_reached`, and the categorical `order_gate_prompted|decided` + * `order_submit_resolved`. * 3. errors — `app_error`. diff --git a/app/src/renderer/hooks/useSchedules.ts b/app/src/renderer/hooks/useSchedules.ts index 362c61b..7916d15 100644 --- a/app/src/renderer/hooks/useSchedules.ts +++ b/app/src/renderer/hooks/useSchedules.ts @@ -13,8 +13,10 @@ export function useSchedules() { } /** - * One agent's upcoming schedules/monitors + recorded wakes for the Monitor tab, - * live via the same `scheduler:changed` subscription. No-ops until an agent is selected. + * One agent's schedules/monitors (retired included — the panel filters `enabled` for + * Active and resolves History rows' triggers from the same arrays) + recorded wakes for + * the Monitor tab, live via the same `scheduler:changed` subscription. No-ops until an + * agent is selected. */ export function useMonitor(agentId?: string) { const utils = trpc.useUtils(); diff --git a/app/src/shared/analytics.ts b/app/src/shared/analytics.ts index bb4acb4..c89dd2a 100644 --- a/app/src/shared/analytics.ts +++ b/app/src/shared/analytics.ts @@ -59,16 +59,20 @@ const orderType = z.enum(["market", "limit", "other"]); const orderKind = z.enum(["place", "cancel", "exercise"]); /** - * Why a failed headless wake run failed, classified from the claude CLI's stderr tail - * (or the codex app-server's turn error) at the exit site. Only the category ships — - * the text it was derived from stays in the local host log. `other` is the fail-closed - * bucket for anything the classifier doesn't recognize. + * Why a wake failed, classified from the claude CLI's stderr tail (or the codex + * app-server's turn error) at the exit site, or mapped from Claude Code's StopFailure + * hook `error` value. Only the category ships (on `wake_finished`) — the text it was + * derived from stays in the local host log. `other` is the fail-closed bucket for + * anything the classifier doesn't recognize. */ export const WakeFailureCategory = z.enum([ "auth", "billing", "rate_limit", "unknown_session", + /** The API couldn't be reached or answered: connection refused/reset, DNS, timeout, + * or a server-side 5xx (Claude Code's StopFailure reports both as `server_error`). */ + "network", "other", ]); export type WakeFailureCategory = z.infer; @@ -207,13 +211,20 @@ export const TELEMETRY_EVENTS = { source: z.enum(["cron", "monitor"]), path: z.enum(["warm", "headless"]), }), - headless_run_finished: z.strictObject({ - result: z.enum(["ok", "resume_fail", "spawn_fail"]), + /** A started wake settled (§12.2): the outcome for BOTH delivery paths, emitted where + * the History row is settled — the one place that also sees the StopFailure hook. + * Replaces the former `headless_run_finished` (child-exit view of a headless run + * only, which read `ok` for a late API-error turn); `path = headless` is its + * successor series. `failure_category` is present only when there was error text or + * a hook error value to classify — spawn errors and pre-delivery interrupts carry + * none. Without it a resume-fail streak that breaks an agent is undiagnosable from + * telemetry (the text itself is local-only). */ + wake_finished: z.strictObject({ + source: z.enum(["cron", "monitor"]), + path: z.enum(["warm", "headless"]), + outcome: z.enum(["succeeded", "failed", "stopped"]), duration_ms: z.number().int().nonnegative(), - /** Present only on failed runs whose exit site has error text to classify (a - * claude resume-fail's stderr tail, a codex turn error) — spawn errors and - * pre-delivery interrupts carry none. Without it a resume-fail streak that - * breaks an agent is undiagnosable from telemetry (the text is local-only). */ + failure_reason: z.enum(["resume_fail", "spawn_fail", "api_error"]).optional(), failure_category: WakeFailureCategory.optional(), }), agent_marked_broken: z.strictObject({}), diff --git a/app/src/shared/feedback.ts b/app/src/shared/feedback.ts index bbe6057..467d871 100644 --- a/app/src/shared/feedback.ts +++ b/app/src/shared/feedback.ts @@ -1,5 +1,6 @@ import { z } from "zod"; import { ApprovalMode } from "./agent"; +import { WakeFailureCategory } from "./analytics"; import { BrokerConnectionStatus } from "./broker"; /** @@ -67,6 +68,16 @@ export const FeedbackDiagnostics = z.strictObject({ schedules_enabled: count, monitors_enabled: count, + // wakes (§12.2) — the last 7 days, counts + one category, never prompts + /** Wakes that actually started (a History row exists) in the window. */ + wakes_7d: count, + wakes_failed_7d: count, + wakes_stopped_7d: count, + /** Failed wakes whose turn ended in an API error (Claude Code's StopFailure). */ + wakes_failed_api_7d: count, + /** The most frequent failure category in the window; null when nothing failed. */ + wakes_top_failure_7d: WakeFailureCategory.nullable(), + // broker — status only broker_status: BrokerConnectionStatus, broker_authorized: z.boolean(), diff --git a/app/src/shared/schedule.ts b/app/src/shared/schedule.ts index 6375d64..7f382a4 100644 --- a/app/src/shared/schedule.ts +++ b/app/src/shared/schedule.ts @@ -1,4 +1,5 @@ import { z } from "zod"; +import { WakeFailureCategory } from "./analytics"; /** * Durable autonomy primitives owned by the backend scheduler. These mirror Claude @@ -40,6 +41,20 @@ export const Monitor = z.object({ }); export type Monitor = z.infer; +/** + * How a recorded wake ended. A row is written only when the run/turn actually starts + * (`running`), then settles exactly once: `succeeded` (the headless child exited, or the + * warm turn's Stop hook fired), `failed` (the child couldn't resume/spawn — see + * `WakeFailureReason`), or `stopped` (a user Stop / the live session went away mid-turn). + */ +export const WakeOutcome = z.enum(["running", "succeeded", "failed", "stopped"]); +export type WakeOutcome = z.infer; + +/** Why a wake failed: the session couldn't be resumed, the child never spawned, or the + * turn itself ended in an API error (Claude Code's `StopFailure` hook — warm or headless). */ +export const WakeFailureReason = z.enum(["resume_fail", "spawn_fail", "api_error"]); +export type WakeFailureReason = z.infer; + /** One recorded autonomy wake — a cron firing or a monitor trigger. */ export const Wake = z.object({ id: z.string(), @@ -52,6 +67,14 @@ export const Wake = z.object({ /** Delivered headlessly (no live interactive session) vs warm via the channel. */ background: z.boolean(), firedAt: z.number(), + /** Null on rows recorded before outcomes existed (pre-v7). */ + outcome: WakeOutcome.nullable(), + /** When the outcome settled; null while `running` and on pre-v7 rows. */ + finishedAt: z.number().nullable(), + /** Set only when `outcome` is `failed`. */ + failureReason: WakeFailureReason.nullable(), + /** Coarse classification of the failure text, when one was recognized. */ + failureCategory: WakeFailureCategory.nullable(), }); export type Wake = z.infer; diff --git a/resources/hooks/status-notify.sh b/resources/hooks/status-notify.sh index 5930b2c..5db557d 100755 --- a/resources/hooks/status-notify.sh +++ b/resources/hooks/status-notify.sh @@ -1,9 +1,11 @@ #!/bin/bash -# OpenTrade status hook (Notification / Stop). +# OpenTrade status hook (Notification / Stop / StopFailure). # # Forwards the Claude Code hook payload to the app's local server, which dispatches # on hook_event_name: Notification → needs-input; Stop → clears needs-input + captures -# session_id for the Resume button. Fire-and-forget with a short timeout so it never +# session_id for the Resume button; StopFailure (fires instead of Stop on an API-error +# turn) → the same, plus the outstanding wake is recorded as failed with the payload's +# error category. Fire-and-forget with a short timeout so it never # delays Claude Code; always exits 0. INPUT=$(cat)