From c62b6313fc82daba17b4d0ac0f0f53c12d55ccdb Mon Sep 17 00:00:00 2001 From: pranav Date: Mon, 14 Sep 2026 10:28:23 -0700 Subject: [PATCH] Record wake outcomes and expand Monitor tab history rows Wake history rows are now written when a wake actually starts and settled once when it ends, instead of being written when the timer or monitor fires. The scheduler mints a wake id at fire time and hands the coordinator a PendingWake; the coordinator calls back wakeStarted when the headless child spawns or the live session accepts the wake (channel handoff or codex push ack), and wakeFinished when it settles. The "agent is now running" notification and the schedule_fired event move to the same start moment. A wake the coordinator drops before it runs leaves no row, and an agent archived while its wake sat queued gets no row either. Outcomes: succeeded (headless child exited, or the warm turn's Stop hook fired via a new WakeTransport.onTurnEnded), failed with a reason (resume_fail / spawn_fail / api_error) and the classified failure category, or stopped (user Stop mid-run, or the live session went away before the turn ended). The failure category gains a network value, for connection failures and server errors. Claude Code's StopFailure hook, which fires instead of Stop when a turn ends in an API error, is now registered in the generated agent settings (so existing agents pick it up on their next launch) and handled by the status route: it does the same turn-ended bookkeeping as Stop and settles the outstanding wake as failed with the category mapped from the hook's error value (WakeTransport.onTurnFailed). For a headless run the hook lands before the child exits, so the coordinator holds the failure and an otherwise-clean exit settles failed rather than succeeded. A wake_finished telemetry event, replacing headless_run_finished, is emitted where the row settles, carrying the source, delivery path, outcome, duration, and failure detail for both paths. The feedback form's diagnostics block gains seven-day wake counts: started, failed, stopped, failed on an API error, and the most frequent failure category. Schema v7 adds outcome, finished_at, failure_reason and failure_category to wakes, all nullable; rows from earlier versions read NULL and render as before. Settling is a single guarded update on a running row, so a wake settles at most once, and any row still running at host start is marked stopped, since nothing in flight survives a restart. schedule.forAgent now returns the agent's schedules and monitors with retired rows included (Scheduler.listTriggers), so a history row can resolve its trigger after the trigger is gone; the Monitor tab filters enabled rows for its Active list itself. History rows expand to the fire time, the outcome (with the category as a hint for recognized failures), and the trigger's prompt or command. The row's leading icon reflects the state: a spinner while running, a red alert when failed, an orange alert when stopped. Tests cover the recorded outcomes on the headless, channel and push paths, StopFailure on warm and in-flight headless wakes, the drop-before-start case, the wakeFinished update, listTriggers, wakeStats, the StopFailure error mapping and network classifier patterns, the status route over HTTP, the diagnostics block, and the v7 migration. --- app/src/main/db/ddl.ts | 6 +- app/src/main/db/migrate.test.ts | 28 ++ app/src/main/db/migrate.ts | 10 +- app/src/main/db/schema.ts | 10 + .../main/services/analytics/analytics.test.ts | 5 + .../services/feedback/diagnostics.test.ts | 6 + app/src/main/services/feedback/diagnostics.ts | 12 + app/src/main/services/harness/claude.ts | 12 +- app/src/main/services/local-api/index.ts | 12 +- .../services/local-api/status-route.test.ts | 85 +++++ app/src/main/services/scheduler/index.ts | 195 +++++++++-- .../main/services/scheduler/scheduler.test.ts | 190 ++++++++++- .../services/scheduler/wake/codex-strategy.ts | 17 +- .../scheduler/wake/coordinator.test.ts | 314 +++++++++++++++--- .../services/scheduler/wake/coordinator.ts | 130 +++++++- .../scheduler/wake/failure-category.test.ts | 67 +++- .../scheduler/wake/failure-category.ts | 42 ++- .../scheduler/wake/headless-strategy.ts | 20 +- app/src/main/services/scheduler/wake/types.ts | 49 ++- app/src/main/trpc/routers/feedback.ts | 7 +- app/src/main/trpc/routers/schedule.ts | 7 +- .../renderer/components/panels/Monitor.tsx | 178 ++++++++-- .../settings/TelemetryOptOutDialog.tsx | 2 +- app/src/renderer/hooks/useSchedules.ts | 6 +- app/src/shared/analytics.ts | 31 +- app/src/shared/feedback.ts | 11 + app/src/shared/schedule.ts | 23 ++ resources/hooks/status-notify.sh | 6 +- 28 files changed, 1295 insertions(+), 186 deletions(-) create mode 100644 app/src/main/services/local-api/status-route.test.ts 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)