From 89ec1e6040ebe937b27b962dc91ed0c8df0754c9 Mon Sep 17 00:00:00 2001 From: Daniel Ehrhardt Date: Sun, 11 Oct 2026 16:08:51 +0200 Subject: [PATCH 1/3] Fix a cut-off phone screenshot, paused runs of runner chats, sleeping runs and silent usage limit exits - A screenshot that arrives cut short (the phone went away meanwhile) is a device error the app can show, not a failed request. - The copy of a run that stands still on a runner is no longer closed, and warned about, on every start. - The watchdog lets a run sleep as long as it asked Claude Code to wake it later (ScheduleWakeup). - A run Claude Code ends without a word right after the usage limit ended another one waits for the reset too. --- packages/core/src/android/screen.ts | 11 ++++++- packages/core/src/runner/runner.ts | 34 ++++++++++++++++++---- packages/core/src/runner/watchdog.ts | 19 ++++++++++-- packages/core/src/services/pauses.ts | 6 +++- packages/core/test/android.test.ts | 17 +++++++++-- packages/core/test/fixtures/fake-claude.ts | 5 ++++ packages/core/test/heartbeats.test.ts | 12 ++++++++ packages/core/test/pause.test.ts | 31 ++++++++++++++++++++ 8 files changed, 122 insertions(+), 13 deletions(-) diff --git a/packages/core/src/android/screen.ts b/packages/core/src/android/screen.ts index 811e03f6..d7dfec96 100644 --- a/packages/core/src/android/screen.ts +++ b/packages/core/src/android/screen.ts @@ -98,6 +98,15 @@ export function parseRawScreencap(buf: Buffer): Raster { return { width, height, data }; } +/** What `screencap | gzip` sent: cut short when the phone or its cable went away meanwhile. */ +export function unpackScreencap(buf: Buffer): Buffer { + try { + return gunzipSync(buf); + } catch { + throw new AdbError(`The phone sent an incomplete picture (${buf.length} bytes)`); + } +} + export async function captureRaw(serial: string, signal?: AbortSignal): Promise { const gzip = gzipWorks.get(serial) !== false; const res = await adb(["exec-out", gzip ? "screencap | gzip -1" : "screencap"], { serial, timeoutMs: 20_000, signal }); @@ -112,7 +121,7 @@ export async function captureRaw(serial: string, signal?: AbortSignal): Promise< if (gzip) { if (buf[0] === 0x1f && buf[1] === 0x8b) { gzipWorks.set(serial, true); - buf = gunzipSync(buf); + buf = unpackScreencap(buf); } else { gzipWorks.set(serial, false); } diff --git a/packages/core/src/runner/runner.ts b/packages/core/src/runner/runner.ts index 3ea58545..554d444c 100644 --- a/packages/core/src/runner/runner.ts +++ b/packages/core/src/runner/runner.ts @@ -1772,16 +1772,37 @@ function bodyOf(job: Job, agent: Agent): string { return continueContext({ reason: resumed.reason, userName: getSettings().general.userName, pausedAt: resumed.pausedAt, messages, answer: resumed.answer }); } -/** The usage limit that ended the run, when that is why it failed and the run can wait for the reset. */ -function limitOf(job: Job, attempt: Attempt): LimitPause | null { - // Checks and dreams come round again by themselves; delegated work reports its failure to the run that waits for it. - if (job.trigger === "check" || job.trigger === "dream" || job.parentRunId) return null; +/** The usage limit Claude last refused a run with on this computer, and when: every run works on the same account. */ +let lastLimit: { pause: LimitPause; at: number } | null = null; +const LIMIT_ECHO_MS = 5 * 60_000; + +/** The usage limit Claude Code named when the run ended. */ +function limitSaid(job: Job, attempt: Attempt): LimitPause | null { const { final, limit, limitLine } = job.acc; if (final?.subtype === "error_max_turns" || final?.subtype === "error_max_budget_usd") return null; // The answer itself only counts while Claude reports a limit as reached: the model may write about limits too. const said = [limitLine ?? "", limit ? (final?.text ?? "") : "", final?.errors.join("\n") ?? "", attempt.stderr, attempt.noise.join("\n")].join("\n"); // Refused requests (429) while Claude reports a limit as reached are that limit, whatever the wording. - return limitReached(limit, final?.apiErrorStatus === 429 && limit ? `${said}\nusage limit reached` : said); + const reached = limitReached(limit, final?.apiErrorStatus === 429 && limit ? `${said}\nusage limit reached` : said); + if (reached) lastLimit = { pause: reached, at: Date.now() }; + return reached; +} + +/** + * Claude Code ended without a word right after the usage limit ended another run: at the limit it can exit before its + * own line about it is out. That limit is this run's too. + */ +function limitUnsaid(job: Job, attempt: Attempt): LimitPause | null { + if (!lastLimit || job.acc.final || !attempt.exitCode || attempt.stderr.trim() || attempt.noise.length) return null; + const until = lastLimit.pause.resumeAt ? Date.parse(lastLimit.pause.resumeAt) : 0; + return Date.now() - lastLimit.at < LIMIT_ECHO_MS && until > Date.now() ? lastLimit.pause : null; +} + +/** The usage limit that ended the run, when that is why it failed and the run can wait for the reset. */ +function limitOf(job: Job, attempt: Attempt): LimitPause | null { + const limit = limitSaid(job, attempt) ?? limitUnsaid(job, attempt); + // Checks and dreams come round again by themselves; delegated work reports its failure to the run that waits for it. + return job.trigger === "check" || job.trigger === "dream" || job.parentRunId ? null : limit; } async function runClaude(job: Job, agent: Agent, res: Resources): Promise { @@ -2243,7 +2264,8 @@ async function runClaude(job: Job, agent: Agent, res: Resources): Promise {}); return { status: "paused", error: null }; } - return { status: "failed", error: describeFailure(job, attempt) }; + const unsaid = limitUnsaid(job, attempt); + return { status: "failed", error: unsaid ? `Claude's ${unsaid.limit} is reached` : describeFailure(job, attempt) }; } finally { try { await logSink.end(); diff --git a/packages/core/src/runner/watchdog.ts b/packages/core/src/runner/watchdog.ts index 415ccce7..b6d19943 100644 --- a/packages/core/src/runner/watchdog.ts +++ b/packages/core/src/runner/watchdog.ts @@ -27,6 +27,19 @@ const KEEP_DAYS = 90; /** Steps that wait on purpose — for another agent, or for the human. */ const WAITING_TOOLS: ReadonlySet = new Set(["mcp__godmode__agent_delegate", "mcp__godmode__ask_human", "mcp__godmode__request_approval"]); +/** Claude Code's own "wake me later": the process stays and is silent until the wakeup calls the model again. */ +const WAKEUP_TOOL = "ScheduleWakeup"; + +/** How long the run sleeps on purpose after this step. */ +function wakeupMs(step: ToolBlock | undefined): number { + if (step?.name !== WAKEUP_TOOL || step.isError) return 0; + const input = step.input && typeof step.input === "object" ? (step.input as Record) : {}; + const seconds = Number(input.delaySeconds); + if (input.stop === true || !Number.isFinite(seconds)) return 0; + // Claude Code keeps the delay between a minute and an hour. + return Math.min(3600, Math.max(60, seconds)) * 1000; +} + export interface Verdict { kind: WatchdogKind; report: string; @@ -92,11 +105,11 @@ export function assess(run: Pick b.task?.status === "running") ?? null; const busy = pending ?? background; - const limit = busy ? Math.max(limits.stallMs * TOOL_FACTOR, Math.min(MIN_TOOL_STALL_MS, limits.stallMs * 10)) : limits.stallMs; - const idle = at - run.lastOutputAt; - if (idle < limit) return null; const done = tools.filter((b) => b.result !== undefined); const last = done[done.length - 1]; + const limit = busy ? Math.max(limits.stallMs * TOOL_FACTOR, Math.min(MIN_TOOL_STALL_MS, limits.stallMs * 10)) : limits.stallMs + wakeupMs(last); + const idle = at - run.lastOutputAt; + if (idle < limit) return null; const where = pending ? `while this step ran: ${stepLabel(pending)}` : background diff --git a/packages/core/src/services/pauses.ts b/packages/core/src/services/pauses.ts index b7382b2a..7033365a 100644 --- a/packages/core/src/services/pauses.ts +++ b/packages/core/src/services/pauses.ts @@ -306,7 +306,11 @@ export function startPauses(): void { started = true; // A pause without its run, or a paused run without its pause (a restored backup): nothing could continue it. exec("DELETE FROM paused_runs WHERE run_id NOT IN (SELECT id FROM runs WHERE status = 'paused')"); - const lost = all<{ id: string }>("SELECT id FROM runs WHERE status = 'paused' AND id NOT IN (SELECT run_id FROM paused_runs)"); + // Not the copy of a run that stands still on a runner: its pause is kept there. + const lost = all<{ id: string }>( + `SELECT id FROM runs WHERE status = 'paused' AND id NOT IN (SELECT run_id FROM paused_runs) + AND conversation_id NOT IN (SELECT id FROM conversations WHERE runner_id IS NOT NULL)`, + ); for (const { id } of lost) void cancelRun(id, "Paused, but what it needed to continue is gone").catch((err) => log.warn(`could not close run ${id}`, err)); if (lost.length) log.warn(`closed ${lost.length} paused run(s) that couldn't be continued`); // An open question belongs to a run that stands still for it, and the other way round. diff --git a/packages/core/test/android.test.ts b/packages/core/test/android.test.ts index eec11da8..b34b69c6 100644 --- a/packages/core/test/android.test.ts +++ b/packages/core/test/android.test.ts @@ -2,6 +2,7 @@ import { afterAll, beforeAll, beforeEach, describe, expect, test } from "bun:tes import { mkdtempSync, readFileSync, realpathSync, rmSync, writeFileSync } from "node:fs"; import { join, resolve } from "node:path"; import { tmpdir } from "node:os"; +import { gzipSync } from "node:zlib"; import type { Agent } from "@godmode/shared"; import { makeAgent, setupEnv, type TestEnv } from "./fixtures/runner-harness"; import * as vault from "../src/vault/vault"; @@ -10,7 +11,7 @@ import { createConversation, getConversationSummary } from "../src/services/conv import { updateSettings } from "../src/services/settings"; import { issueRunToken, revokeRunToken } from "../src/mcp/tokens"; import { createCredential } from "../src/vault/credentials"; -import { connectionOf, parseDevices } from "../src/android/adb"; +import { AdbError, connectionOf, parseDevices } from "../src/android/adb"; import { runAndroidDeviceIds } from "../src/android/assignments"; import { androidStatus, @@ -25,7 +26,7 @@ import { serialOf, updateDevice, } from "../src/android/service"; -import { frameToScreen, parseRawScreencap, parseScreenSize, parseState, parseUiXml, typeCommands, untypable } from "../src/android/screen"; +import { frameToScreen, parseRawScreencap, parseScreenSize, parseState, parseUiXml, typeCommands, unpackScreencap, untypable } from "../src/android/screen"; import { FrameAssembler, avcC, codecString, splitNals, videoSize } from "../src/android/stream"; import { forgetAndroidRun } from "../src/android/tools"; import { buildSystemPrompt } from "../src/runner/prompt"; @@ -119,6 +120,18 @@ describe("parsing", () => { // BGRA becomes RGBA. expect([...parseRawScreencap(raw).data]).toEqual([30, 20, 10, 255, 3, 2, 1, 255]); expect(() => parseRawScreencap(raw.subarray(0, 18))).toThrow(/incomplete/); + + const packed = gzipSync(raw); + expect(unpackScreencap(packed).equals(raw)).toBe(true); + // The phone went away while the picture came: an answer the app can show, not a crash of the request. + let cut: unknown; + try { + unpackScreencap(packed.subarray(0, packed.length - 6)); + } catch (err) { + cut = err; + } + expect(cut).toBeInstanceOf(AdbError); + expect((cut as Error).message).toMatch(/incomplete picture/); }); test("typing: quoting for the phone's shell, %s, Enter and Tab", () => { diff --git a/packages/core/test/fixtures/fake-claude.ts b/packages/core/test/fixtures/fake-claude.ts index cdef40d6..9feab264 100755 --- a/packages/core/test/fixtures/fake-claude.ts +++ b/packages/core/test/fixtures/fake-claude.ts @@ -46,6 +46,8 @@ * answer, else answer "step finished" * LIMIT_HIT while the state dir has a `limit` file (its content: the reset time in unix seconds, may be empty) end * like Claude Code does when a usage limit is reached; afterwards answer "back after the limit" + * LIMIT_SILENT while the state dir has a `limit` file exit with code 1 and without a word, like Claude Code + * sometimes does at a usage limit; afterwards whatever the rest of the prompt asks for * OVERAGE_CRASH report a reached limit like Claude Code does while requests still go through, then crash like CRASH * Dream: … a dream (memory consolidation): rewrites MEMORY.md from the `REMEMBER: ` lines of the activity * digest (+ memory/dream-notes.md), calls the gateway (tools/list, a forbidden tool, memory_dream_report) @@ -782,6 +784,9 @@ if (slash?.[1] === "clear") { textTurn("step finished"); result("step finished"); } +} else if (prompt.includes("LIMIT_SILENT") && existsSync(join(stateDir, "limit"))) { + out(init); + process.exit(1); } else if (prompt.includes("LIMIT_HIT")) { out(init); const limitFile = join(stateDir, "limit"); diff --git a/packages/core/test/heartbeats.test.ts b/packages/core/test/heartbeats.test.ts index 2e648ffa..2d619a7b 100644 --- a/packages/core/test/heartbeats.test.ts +++ b/packages/core/test/heartbeats.test.ts @@ -240,6 +240,18 @@ describe("watchdog", () => { expect(assess({ blocks: building, lastOutputAt: t0, startedAt: t0, waiting: true }, t0 + 5 * 3_600_000, limits)).toBeNull(); }); + test("a run that asked Claude Code to wake it later sleeps that long", () => { + const asleep = [tool("a", "Read", { file_path: "/x" }, "ok"), tool("w", "ScheduleWakeup", { delaySeconds: 1200, reason: "waiting for CI", prompt: "check" }, "Next wakeup scheduled")]; + expect(assess({ blocks: asleep, lastOutputAt: t0, startedAt: t0, waiting: false }, t0 + 21 * 60_000, limits)).toBeNull(); + expect(assess({ blocks: asleep, lastOutputAt: t0, startedAt: t0, waiting: false }, t0 + 39 * 60_000, limits)).toBeNull(); + // The wakeup came and went, and still nothing. + expect(assess({ blocks: asleep, lastOutputAt: t0, startedAt: t0, waiting: false }, t0 + 41 * 60_000, limits)?.kind).toBe("stalled"); + const ended = [tool("w", "ScheduleWakeup", { stop: true }, "Loop ended")]; + expect(assess({ blocks: ended, lastOutputAt: t0, startedAt: t0, waiting: false }, t0 + 21 * 60_000, limits)?.kind).toBe("stalled"); + const moved = [...asleep, tool("b", "Read", { file_path: "/y" }, "ok")]; + expect(assess({ blocks: moved, lastOutputAt: t0, startedAt: t0, waiting: false }, t0 + 21 * 60_000, limits)?.kind).toBe("stalled"); + }); + test("the same step with the same answer again and again goes in circles", () => { const same = (i: number) => tool(`l${i}`, "mcp__browser__browser_click", { index: 3 }, "Nothing happened"); const blocks = [same(1), { type: "text", text: "Trying again" } as MessageBlock, same(2), same(3), same(4)]; diff --git a/packages/core/test/pause.test.ts b/packages/core/test/pause.test.ts index 5fc369dc..e209ce04 100644 --- a/packages/core/test/pause.test.ts +++ b/packages/core/test/pause.test.ts @@ -437,6 +437,21 @@ describe("Claude's usage limit", () => { expect(listNotifications().filter((n) => n.title === "Claude's session limit is reached")).toHaveLength(1); }); + test("a run Claude Code ends without a word right after the limit ended another one waits for the reset too", async () => { + signal("limit", String(inAnHour())); + const first = await startChat({ agentId: agent.id, content: "LIMIT_HIT says so" }); + await paused(first.run.id); + const silent = await startChat({ agentId: agent.id, content: "LIMIT_SILENT then answer" }); + await paused(silent.run.id); + expect(pauseOf(silent.conversation.id)).toMatchObject({ reason: "limit", limit_name: "session limit", auto: 1, resume_at: pauseOf(first.conversation.id)!.resume_at }); + + rmSync(join(env.stateDir, "limit")); + for (const id of [first.run.id, silent.run.id]) due(id); + sweep(); + expect(await waitForRun(first.run.id, 20_000)).toMatchObject({ status: "succeeded", result: "back after the limit" }); + expect((await waitForRun(silent.run.id, 20_000)).status).toBe("succeeded"); + }); + test("a chat archived while it waits for the reset still continues by itself", async () => { signal("limit", String(inAnHour())); const started = await startChat({ agentId: agent.id, content: "LIMIT_HIT then filed away" }); @@ -652,4 +667,20 @@ describe("restart", () => { expect(getRun(lost.run.id).status).toBe("cancelled"); expect(all<{ id: string }>("SELECT id FROM runs WHERE status = 'paused'")).toEqual([]); }); + + test("the copy of a run that stands still on a runner is left alone", async () => { + const remote = await working("SLEEP on a runner"); + await pauseConversation(remote.conversation.id); + await paused(remote.run.id); + // As the copy of a runner's chat looks here: the pause is kept on the runner. + sql("DELETE FROM paused_runs WHERE run_id = ?", remote.run.id); + sql("UPDATE conversations SET runner_id = 'rnr_asleep' WHERE id = ?", remote.conversation.id); + stopPauses(); + startPauses(); + expect(getRun(remote.run.id).status).toBe("paused"); + sql("UPDATE conversations SET runner_id = NULL WHERE id = ?", remote.conversation.id); + stopPauses(); + startPauses(); + expect(getRun(remote.run.id).status).toBe("cancelled"); + }); }); From c6e264f038efaf2f95c4f1c75e97ea1fe70ae859 Mon Sep 17 00:00:00 2001 From: Daniel Ehrhardt Date: Sun, 11 Oct 2026 16:23:38 +0200 Subject: [PATCH 2/3] Save the screenshots of an in-flight message once instead of with every save While a run works its message is saved every few seconds. Its screenshots are most of its size (up to 8 MB) and never change: they are written once to message_images, the blocks are saved without them, and whoever reads the message gets them back. When the run ends or stands still the message is saved whole again. --- packages/core/src/db/migrations.ts | 15 ++++++++ packages/core/src/runner/runner.ts | 39 ++++++++++++++++++--- packages/core/src/services/conversations.ts | 34 ++++++++++++++++-- packages/core/test/run-stream.test.ts | 11 +++++- packages/core/test/runner.test.ts | 20 ++++++++++- 5 files changed, 110 insertions(+), 9 deletions(-) diff --git a/packages/core/src/db/migrations.ts b/packages/core/src/db/migrations.ts index 6a704b0c..bbab7294 100644 --- a/packages/core/src/db/migrations.ts +++ b/packages/core/src/db/migrations.ts @@ -1393,6 +1393,21 @@ ALTER TABLE conversations ADD COLUMN android_device_ids TEXT NOT NULL DEFAULT '[ -- holds their manifest and hooks for reading. contents: what such a plugin brings (JSON, ModContents). ALTER TABLE mods ADD COLUMN source TEXT; ALTER TABLE mods ADD COLUMN contents TEXT; +`, + }, + { + id: 85, + name: "message_images", + sql: /* sql */ ` +-- The screenshots of a message a run is still writing. The message is saved every few seconds while the run works: +-- its megabytes of pictures, which never change, are written once here instead of with every save. When the run +-- ends or stands still the message is saved whole and its rows here go. +CREATE TABLE message_images ( + message_id TEXT NOT NULL REFERENCES messages(id) ON DELETE CASCADE, + block_id TEXT NOT NULL, + image TEXT NOT NULL, + PRIMARY KEY (message_id, block_id) +); `, }, ]; diff --git a/packages/core/src/runner/runner.ts b/packages/core/src/runner/runner.ts index 554d444c..ee2ae6a8 100644 --- a/packages/core/src/runner/runner.ts +++ b/packages/core/src/runner/runner.ts @@ -84,6 +84,7 @@ import { setConversationState, titleFromContent, updateMessage, + wholeBlocks, } from "../services/conversations"; import { startQueued, takeQueued } from "../services/messageQueue"; import { @@ -950,9 +951,10 @@ export function recoverInterruptedRuns(): void { "SELECT id, content, blocks FROM messages WHERE run_id = ? AND role = 'assistant'", r.id, )) { - const blocks = withdrawOpenBlocks(parseJson(m.blocks, []), "Godmode restarted before this was asked."); + const blocks = withdrawOpenBlocks(wholeBlocks(m.id, parseJson(m.blocks, [])), "Godmode restarted before this was asked."); blocks.push({ type: "error", text: INTERRUPTED }); sql("UPDATE messages SET blocks = ? WHERE id = ?", JSON.stringify(blocks), m.id); + sql("DELETE FROM message_images WHERE message_id = ?", m.id); } } for (const agentId of agentIds) safely("reset agent status", () => setAgentStatus(agentId, "idle")); @@ -1397,10 +1399,12 @@ interface LiveView { sent: MessageBlock[]; /** Deltas sent so far. */ seq: number; + /** The screenshots saved apart from the in-flight message, by the id of their step; null before its first save. */ + images: Map | null; } function liveOf(job: Job): LiveView { - return (job.live ??= { stream: newId("str"), epoch: redactionEpoch(), blocks: new WeakMap(), sent: [], seq: 0 }); + return (job.live ??= { stream: newId("str"), epoch: redactionEpoch(), blocks: new WeakMap(), sent: [], seq: 0, images: null }); } function isPlain(v: unknown): v is Record { @@ -1453,9 +1457,34 @@ function liveBlocks(job: Job): LiveBlock[] { }); } -/** For the row that is saved while the run works: each block is serialized once (screenshots are megabytes). */ +/** For the row that is saved while the run works: each block is serialized once, without its screenshot (megabytes). */ function storedJson(b: LiveBlock): string { - return (b.json ??= JSON.stringify(b.redacted)); + return (b.json ??= JSON.stringify(b.redacted.type === "tool_use" && b.redacted.image ? { ...b.redacted, image: undefined } : b.redacted)); +} + +/** + * Save the in-flight message. Its screenshots never change and are most of its size: each is written once, apart from + * the blocks that are written again with every save (message_images). Whoever reads the message gets them back. + */ +function saveInFlight(job: Job, view: LiveBlock[]) { + const live = liveOf(job); + const shots = new Map(); + for (const { redacted: b } of view) if (b.type === "tool_use" && b.image) shots.set(b.id, b.image); + tx(() => { + // Someone else saved the message whole meanwhile (or this is the first save): start over. + const stored = get<{ n: number }>("SELECT COUNT(*) AS n FROM message_images WHERE message_id = ?", job.messageId)?.n ?? 0; + let known = live.images; + if (!known || known.size !== stored) { + if (stored) sql("DELETE FROM message_images WHERE message_id = ?", job.messageId); + known = new Map(); + } + for (const id of known.keys()) if (!shots.has(id)) sql("DELETE FROM message_images WHERE message_id = ? AND block_id = ?", job.messageId, id); + for (const [id, image] of shots) { + if (known.get(id) !== image) sql("INSERT OR REPLACE INTO message_images (message_id, block_id, image) VALUES (?, ?, ?)", job.messageId, id, image); + } + sql("UPDATE messages SET blocks = ? WHERE id = ?", `[${view.map(storedJson).join(",")}]`, job.messageId); + }); + live.images = shots; } /** The run's blocks as they are stored and shown, secrets masked. */ @@ -1494,7 +1523,7 @@ function emitDelta(job: Job) { if (Date.now() - job.lastPersistAt < (job.persistEveryMs ?? persistIntervalMs)) return; job.lastPersistAt = Date.now(); const started = performance.now(); - safely("persist in-flight message", () => sql("UPDATE messages SET blocks = ? WHERE id = ?", `[${view.map(storedJson).join(",")}]`, job.messageId)); + safely("persist in-flight message", () => saveInFlight(job, view)); const ms = performance.now() - started; job.slowestPersistMs = Math.max(job.slowestPersistMs ?? 0, ms); job.persistEveryMs = Math.max(persistIntervalMs, Math.round(ms * persistBackoff)); diff --git a/packages/core/src/services/conversations.ts b/packages/core/src/services/conversations.ts index d4e72243..9e4ae184 100644 --- a/packages/core/src/services/conversations.ts +++ b/packages/core/src/services/conversations.ts @@ -562,16 +562,44 @@ export async function deleteConversation(id: string): Promise { /* Messages */ /* ------------------------------------------------------------------ */ +interface ImageRow { + message_id: string; + block_id: string; + image: string; +} + +/** Put the screenshots a run saved apart from its in-flight message (message_images) back on their steps. */ +function withImages(blocks: MessageBlock[], images: ImageRow[]): MessageBlock[] { + if (!images.length) return blocks; + const byBlock = new Map(images.map((i) => [i.block_id, i.image])); + return blocks.map((b) => (b.type === "tool_use" && !b.image && byBlock.has(b.id) ? { ...b, image: byBlock.get(b.id) } : b)); +} + +/** The blocks of a message as stored, whole: with the screenshots that were saved apart while its run worked. */ +export function wholeBlocks(messageId: string, blocks: MessageBlock[]): MessageBlock[] { + return withImages(blocks, all("SELECT * FROM message_images WHERE message_id = ?", messageId)); +} + export function listMessages(conversationId: string): Message[] { - return all("SELECT * FROM messages WHERE conversation_id = ? ORDER BY created_at ASC, rowid ASC", conversationId).map( + const messages = all("SELECT * FROM messages WHERE conversation_id = ? ORDER BY created_at ASC, rowid ASC", conversationId).map( toMessage, ); + const images = all( + "SELECT i.* FROM message_images i JOIN messages m ON m.id = i.message_id WHERE m.conversation_id = ?", + conversationId, + ); + if (!images.length) return messages; + return messages.map((m) => { + const own = images.filter((i) => i.message_id === m.id); + return own.length ? { ...m, blocks: withImages(m.blocks, own) } : m; + }); } export function getMessage(id: string): Message { const row = get("SELECT * FROM messages WHERE id = ?", id); if (!row) throw notFound("Message"); - return toMessage(row); + const message = toMessage(row); + return { ...message, blocks: wholeBlocks(id, message.blocks) }; } export function addMessage( @@ -616,6 +644,8 @@ export function updateMessage( blocks: patch.blocks === undefined ? undefined : JSON.stringify(patch.blocks), run_id: patch.runId, }); + // Saved whole: what its run kept apart while it worked is in the blocks now. + if (patch.blocks !== undefined) sql("DELETE FROM message_images WHERE message_id = ?", id); const message = getMessage(id); if (opts.emit !== false) bus.emit({ type: "message.updated", message }); return message; diff --git a/packages/core/test/run-stream.test.ts b/packages/core/test/run-stream.test.ts index 23301a4f..071556e6 100644 --- a/packages/core/test/run-stream.test.ts +++ b/packages/core/test/run-stream.test.ts @@ -194,8 +194,14 @@ describe("the in-flight message", () => { expect(shots(state())).toBe(3); const messageId = deltasOf(events, runId)[0]!.messageId; - const saved = JSON.parse(get<{ blocks: string }>("SELECT blocks FROM messages WHERE id = ?", messageId)!.blocks) as MessageBlock[]; + const row = () => JSON.parse(get<{ blocks: string }>("SELECT blocks FROM messages WHERE id = ?", messageId)!.blocks) as MessageBlock[]; + const apart = () => get<{ n: number }>("SELECT COUNT(*) AS n FROM message_images WHERE message_id = ?", messageId)!.n; + // The screenshots are written once, apart from the row that is saved again and again; a reader gets them back. + expect(shots(row())).toBe(0); + expect(apart()).toBe(3); + const saved = getMessage(messageId).blocks; expect(saved).toEqual(state()); + expect(getConversation(started.conversation.id).messages.find((m) => m.id === messageId)!.blocks).toEqual(state()); expect(saved.map((b) => (b.type === "tool_use" ? b.result : b.type === "text" ? b.text : null))).toEqual(["shot 0", "••••••••", "shot 2", "almost there"]); await Bun.write(`${env.stateDir}/inflight-done`, ""); @@ -204,6 +210,9 @@ describe("the in-flight message", () => { expect(run.status).toBe("succeeded"); const message = getMessage(messageId); expect(shots(message.blocks)).toBe(3); + // Saved whole at the end: nothing is kept apart anymore. + expect(shots(row())).toBe(3); + expect(apart()).toBe(0); expect(JSON.stringify(message.blocks)).not.toContain("shot 1"); expect(state()).toEqual(message.blocks); diff --git a/packages/core/test/runner.test.ts b/packages/core/test/runner.test.ts index da3c0a69..7199c5a4 100644 --- a/packages/core/test/runner.test.ts +++ b/packages/core/test/runner.test.ts @@ -5,7 +5,7 @@ import { join } from "node:path"; import type { Agent, ServerEvent } from "@godmode/shared"; import { MAX_INSTRUCTIONS_LENGTH } from "@godmode/shared"; import { argValue, browserRuns, captureEvents, invocations, makeAgent, setupEnv, until, type TestEnv } from "./fixtures/runner-harness"; -import { insert, run as sql } from "../src/db"; +import { all, get, insert, run as sql } from "../src/db"; import { getSettings, updateSettings } from "../src/services/settings"; import { config } from "../src/config"; import { listMissingLogins } from "../src/services/missingLogins"; @@ -545,4 +545,22 @@ describe("recoverInterruptedRuns", () => { const msg = getConversation(conv.id).messages.find((m) => m.id === "msg_stale_test_000001")!; expect(msg.blocks).toEqual([{ type: "error", text: INTERRUPTED }]); }); + + test("the screenshots a cut-off run had saved apart from its message are back in it", () => { + const conv = createConversation({ agentId: agent.id }); + const runId = "run_stale_test_000002"; + const messageId = "msg_stale_test_000002"; + insert("runs", { id: runId, agent_id: agent.id, conversation_id: conv.id, trigger: "chat", status: "running", prompt: "x", created_at: now() }); + const step = { type: "tool_use", id: "toolu_shot", name: "mcp__browser__browser_screenshot", input: {}, result: "Took a screenshot" }; + insert("messages", { id: messageId, conversation_id: conv.id, role: "assistant", content: "", blocks: JSON.stringify([step]), run_id: runId, attachments: "[]", created_at: now() }); + insert("message_images", { message_id: messageId, block_id: "toolu_shot", image: "iVBORw0KGgoAAAA" }); + recoverInterruptedRuns(); + const stored = get<{ blocks: string }>("SELECT blocks FROM messages WHERE id = ?", messageId)!.blocks; + expect(JSON.parse(stored)).toEqual([{ ...step, image: "iVBORw0KGgoAAAA" }, { type: "error", text: INTERRUPTED }]); + expect(all("SELECT 1 FROM message_images WHERE message_id = ?", messageId)).toEqual([]); + // A message that goes takes what was kept apart with it. + insert("message_images", { message_id: messageId, block_id: "toolu_shot", image: "iVBORw0KGgoAAAA" }); + sql("DELETE FROM messages WHERE id = ?", messageId); + expect(all("SELECT 1 FROM message_images WHERE message_id = ?", messageId)).toEqual([]); + }); }); From 01438eb845e9837d8b02bb7d6b6193227a686bf2 Mon Sep 17 00:00:00 2001 From: Daniel Ehrhardt Date: Sun, 11 Oct 2026 16:25:38 +0200 Subject: [PATCH 3/3] In-flight screenshots: page and picture readers, one transaction per save, a sleeping run with a background task - The message page and the picture of a step (phone app) read the screenshots a working run keeps apart. - Saving a message whole and dropping what was kept apart is one transaction; a save for a deleted chat does nothing. - A scheduled wakeup also counts while a background task runs. --- packages/core/src/runner/runner.ts | 14 ++++--- packages/core/src/runner/watchdog.ts | 3 +- packages/core/src/services/conversations.ts | 45 +++++++++++---------- packages/core/test/heartbeats.test.ts | 5 +++ packages/core/test/run-stream.test.ts | 5 ++- 5 files changed, 43 insertions(+), 29 deletions(-) diff --git a/packages/core/src/runner/runner.ts b/packages/core/src/runner/runner.ts index c7e5751e..8f629ad1 100644 --- a/packages/core/src/runner/runner.ts +++ b/packages/core/src/runner/runner.ts @@ -954,8 +954,10 @@ export function recoverInterruptedRuns(): void { )) { const blocks = withdrawOpenBlocks(wholeBlocks(m.id, parseJson(m.blocks, [])), "Godmode restarted before this was asked."); blocks.push({ type: "error", text: INTERRUPTED }); - sql("UPDATE messages SET blocks = ? WHERE id = ?", JSON.stringify(blocks), m.id); - sql("DELETE FROM message_images WHERE message_id = ?", m.id); + tx(() => { + sql("UPDATE messages SET blocks = ? WHERE id = ?", JSON.stringify(blocks), m.id); + sql("DELETE FROM message_images WHERE message_id = ?", m.id); + }); } } for (const agentId of agentIds) safely("reset agent status", () => setAgentStatus(agentId, "idle")); @@ -1471,7 +1473,9 @@ function saveInFlight(job: Job, view: LiveBlock[]) { const live = liveOf(job); const shots = new Map(); for (const { redacted: b } of view) if (b.type === "tool_use" && b.image) shots.set(b.id, b.image); - tx(() => { + const saved = tx(() => { + // The chat was deleted under the run: nothing to save, and no pictures without their message. + if (!sql("UPDATE messages SET blocks = ? WHERE id = ?", `[${view.map(storedJson).join(",")}]`, job.messageId).changes) return false; // Someone else saved the message whole meanwhile (or this is the first save): start over. const stored = get<{ n: number }>("SELECT COUNT(*) AS n FROM message_images WHERE message_id = ?", job.messageId)?.n ?? 0; let known = live.images; @@ -1483,9 +1487,9 @@ function saveInFlight(job: Job, view: LiveBlock[]) { for (const [id, image] of shots) { if (known.get(id) !== image) sql("INSERT OR REPLACE INTO message_images (message_id, block_id, image) VALUES (?, ?, ?)", job.messageId, id, image); } - sql("UPDATE messages SET blocks = ? WHERE id = ?", `[${view.map(storedJson).join(",")}]`, job.messageId); + return true; }); - live.images = shots; + if (saved) live.images = shots; } /** The run's blocks as they are stored and shown, secrets masked. */ diff --git a/packages/core/src/runner/watchdog.ts b/packages/core/src/runner/watchdog.ts index b6d19943..a901a48f 100644 --- a/packages/core/src/runner/watchdog.ts +++ b/packages/core/src/runner/watchdog.ts @@ -107,7 +107,8 @@ export function assess(run: Pick b.result !== undefined); const last = done[done.length - 1]; - const limit = busy ? Math.max(limits.stallMs * TOOL_FACTOR, Math.min(MIN_TOOL_STALL_MS, limits.stallMs * 10)) : limits.stallMs + wakeupMs(last); + const base = busy ? Math.max(limits.stallMs * TOOL_FACTOR, Math.min(MIN_TOOL_STALL_MS, limits.stallMs * 10)) : limits.stallMs; + const limit = pending ? base : Math.max(base, limits.stallMs + wakeupMs(last)); const idle = at - run.lastOutputAt; if (idle < limit) return null; const where = pending diff --git a/packages/core/src/services/conversations.ts b/packages/core/src/services/conversations.ts index a33495b5..b2c5db63 100644 --- a/packages/core/src/services/conversations.ts +++ b/packages/core/src/services/conversations.ts @@ -24,7 +24,7 @@ import type { import type { ComputerTarget, ConversationPatch, ConversationWithMessages, MessagePage, SendMessageInput, SendMessageResult, StartChatResult } from "@godmode/shared"; import { computerTargetLabel, MAX_MESSAGE_ATTACHMENT_BYTES } from "@godmode/shared"; import { liteMessage } from "./liteMessage"; -import { all, bool, get, insert, int, run as sql, update } from "../db"; +import { all, bool, get, insert, int, run as sql, tx, update } from "../db"; import { bus } from "../events/bus"; import { logger } from "../log"; import { badRequest, conflict, newId, notFound, now, parseJson } from "../util"; @@ -444,15 +444,18 @@ export function listMessagePage(conversationId: string, opts: { limit: number; b } else { rows = all("SELECT * FROM messages WHERE conversation_id = ? ORDER BY created_at DESC, rowid DESC LIMIT ?", conversationId, limit + 1); } - return { messages: rows.slice(0, limit).reverse().map(toMessage), hasMore: rows.length > limit }; + return { messages: whole(rows.slice(0, limit).reverse().map(toMessage)), hasMore: rows.length > limit }; } /** The picture a tool returned in a message, decoded, with its type. */ export function messageImage(conversationId: string, messageId: string, toolUseId: string): { data: Uint8Array; mime: string } { const row = get<{ blocks: string }>("SELECT blocks FROM messages WHERE id = ? AND conversation_id = ?", messageId, conversationId); const block = parseJson(row?.blocks ?? "[]", []).find((b) => b.type === "tool_use" && b.id === toolUseId); - if (!block || block.type !== "tool_use" || !block.image) throw notFound("Picture"); - const data = Buffer.from(block.image, "base64"); + if (!block || block.type !== "tool_use") throw notFound("Picture"); + // While its run works the picture is kept apart from the message. + const image = block.image ?? get<{ image: string }>("SELECT image FROM message_images WHERE message_id = ? AND block_id = ?", messageId, toolUseId)?.image; + if (!image) throw notFound("Picture"); + const data = Buffer.from(image, "base64"); return { data: new Uint8Array(data), mime: imageMime(data) }; } @@ -662,19 +665,15 @@ export function wholeBlocks(messageId: string, blocks: MessageBlock[]): MessageB return withImages(blocks, all("SELECT * FROM message_images WHERE message_id = ?", messageId)); } +/** Messages as they are read: with the screenshots their runs keep apart while they work. */ +function whole(messages: Message[]): Message[] { + const apart = new Set(all<{ message_id: string }>("SELECT DISTINCT message_id FROM message_images").map((r) => r.message_id)); + if (!apart.size) return messages; + return messages.map((m) => (apart.has(m.id) ? { ...m, blocks: wholeBlocks(m.id, m.blocks) } : m)); +} + export function listMessages(conversationId: string): Message[] { - const messages = all("SELECT * FROM messages WHERE conversation_id = ? ORDER BY created_at ASC, rowid ASC", conversationId).map( - toMessage, - ); - const images = all( - "SELECT i.* FROM message_images i JOIN messages m ON m.id = i.message_id WHERE m.conversation_id = ?", - conversationId, - ); - if (!images.length) return messages; - return messages.map((m) => { - const own = images.filter((i) => i.message_id === m.id); - return own.length ? { ...m, blocks: withImages(m.blocks, own) } : m; - }); + return whole(all("SELECT * FROM messages WHERE conversation_id = ? ORDER BY created_at ASC, rowid ASC", conversationId).map(toMessage)); } export function getMessage(id: string): Message { @@ -721,13 +720,15 @@ export function updateMessage( ): Message | null { const exists = get<{ id: string }>("SELECT id FROM messages WHERE id = ?", id); if (!exists) return null; - update("messages", id, { - content: patch.content, - blocks: patch.blocks === undefined ? undefined : JSON.stringify(patch.blocks), - run_id: patch.runId, + tx(() => { + update("messages", id, { + content: patch.content, + blocks: patch.blocks === undefined ? undefined : JSON.stringify(patch.blocks), + run_id: patch.runId, + }); + // Saved whole: what its run kept apart while it worked is in the blocks now. + if (patch.blocks !== undefined) sql("DELETE FROM message_images WHERE message_id = ?", id); }); - // Saved whole: what its run kept apart while it worked is in the blocks now. - if (patch.blocks !== undefined) sql("DELETE FROM message_images WHERE message_id = ?", id); const message = getMessage(id); if (opts.emit !== false) bus.emit({ type: "message.updated", message }); return message; diff --git a/packages/core/test/heartbeats.test.ts b/packages/core/test/heartbeats.test.ts index 2d619a7b..e3b194f4 100644 --- a/packages/core/test/heartbeats.test.ts +++ b/packages/core/test/heartbeats.test.ts @@ -248,6 +248,11 @@ describe("watchdog", () => { expect(assess({ blocks: asleep, lastOutputAt: t0, startedAt: t0, waiting: false }, t0 + 41 * 60_000, limits)?.kind).toBe("stalled"); const ended = [tool("w", "ScheduleWakeup", { stop: true }, "Loop ended")]; expect(assess({ blocks: ended, lastOutputAt: t0, startedAt: t0, waiting: false }, t0 + 21 * 60_000, limits)?.kind).toBe("stalled"); + // A background task that is still running doesn't shorten the sleep, and a step that hangs isn't covered by it. + const hour = [tool("w", "ScheduleWakeup", { delaySeconds: 3600, reason: "nightly build", prompt: "check" }, "Next wakeup scheduled")]; + const building = { ...tool("t", "Agent", { prompt: "build" }, "started"), task: { status: "running" } } as MessageBlock; + expect(assess({ blocks: [building, ...hour], lastOutputAt: t0, startedAt: t0, waiting: false }, t0 + 75 * 60_000, limits)).toBeNull(); + expect(assess({ blocks: [...hour, tool("b", "Bash", { command: "sleep 9999" })], lastOutputAt: t0, startedAt: t0, waiting: false }, t0 + 61 * 60_000, limits)?.kind).toBe("stalled"); const moved = [...asleep, tool("b", "Read", { file_path: "/y" }, "ok")]; expect(assess({ blocks: moved, lastOutputAt: t0, startedAt: t0, waiting: false }, t0 + 21 * 60_000, limits)?.kind).toBe("stalled"); }); diff --git a/packages/core/test/run-stream.test.ts b/packages/core/test/run-stream.test.ts index 071556e6..fab6f771 100644 --- a/packages/core/test/run-stream.test.ts +++ b/packages/core/test/run-stream.test.ts @@ -8,7 +8,7 @@ import type { ServerWebSocket } from "bun"; import { applyRunDelta, type Agent, type ClientEvent, type MessageBlock, type RunDelta, type RunDeltaState, type ServerEvent } from "@godmode/shared"; import { captureEvents, makeAgent, setupEnv, until, type TestEnv } from "./fixtures/runner-harness"; import { get } from "../src/db"; -import { getConversation, getMessage, sendMessage, startChat } from "../src/services/conversations"; +import { getConversation, getMessage, listMessagePage, messageImage, sendMessage, startChat } from "../src/services/conversations"; import { __setPersistForTests, findRunLog, waitForRun } from "../src/runner/runner"; import { rememberSecret } from "../src/vault/vault"; import { websocketHandler, type WsData } from "../src/server/ws"; @@ -202,6 +202,9 @@ describe("the in-flight message", () => { const saved = getMessage(messageId).blocks; expect(saved).toEqual(state()); expect(getConversation(started.conversation.id).messages.find((m) => m.id === messageId)!.blocks).toEqual(state()); + expect(listMessagePage(started.conversation.id, { limit: 5 }).messages.find((m) => m.id === messageId)!.blocks).toEqual(state()); + const shot = state().find((b) => b.type === "tool_use" && b.image) as Extract; + expect(Buffer.from(messageImage(started.conversation.id, messageId, shot.id).data).toString("base64")).toBe(shot.image!); expect(saved.map((b) => (b.type === "tool_use" ? b.result : b.type === "text" ? b.text : null))).toEqual(["shot 0", "••••••••", "shot 2", "almost there"]); await Bun.write(`${env.stateDir}/inflight-done`, "");