Fix 9 problems from the log (http, db, perf)

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<Raster> {
   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);
     }

packages/core/src/db/migrations.ts

@@ -1415,6 +1415,21 @@ ALTER TABLE dreams ADD COLUMN memory_target INTEGER;
 -- Dreams run on Opus by default now: saved settings that still carry the old default follow.
 UPDATE settings SET value = json_set(value, '$.dreaming.model', 'claude-opus-5-5')
 WHERE key = 'memory' AND json_valid(value) AND json_extract(value, '$.dreaming.model') = 'sonnet';
+`,
+  },
+  {
+    id: 87,
+    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)
+);
 `,
   },
 ];

packages/core/src/runner/runner.ts

@@ -85,6 +85,7 @@ import {
   setConversationState,
   titleFromContent,
   updateMessage,
+  wholeBlocks,
 } from "../services/conversations";
 import { startQueued, takeQueued } from "../services/messageQueue";
 import {
@@ -951,9 +952,12 @@ export function recoverInterruptedRuns(): void {
       "SELECT id, content, blocks FROM messages WHERE run_id = ? AND role = 'assistant'",
       r.id,
     )) {
-      const blocks = withdrawOpenBlocks(parseJson<Message["blocks"]>(m.blocks, []), "Godmode restarted before this was asked.");
+      const blocks = withdrawOpenBlocks(wholeBlocks(m.id, parseJson<Message["blocks"]>(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);
+      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"));
@@ -1398,10 +1402,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<string, string> | 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<string, unknown> {
@@ -1454,9 +1460,36 @@ 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<string, string>();
+  for (const { redacted: b } of view) if (b.type === "tool_use" && b.image) shots.set(b.id, b.image);
+  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;
+    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);
+    }
+    return true;
+  });
+  if (saved) live.images = shots;
 }
 
 /** The run's blocks as they are stored and shown, secrets masked. */
@@ -1495,7 +1528,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));
@@ -1773,16 +1806,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<Outcome> {
@@ -2244,7 +2298,8 @@ async function runClaude(job: Job, agent: Agent, res: Resources): Promise<Outcom
       for (const child of delegatedBy(job.runId)) await cancelRun(child, "Stopped (the run that delegated this waits for Claude's usage limit)").catch(() => {});
       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();

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<string> = 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<string, unknown>) : {};
+  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,12 @@ export function assess(run: Pick<WatchedRun, "blocks" | "lastOutputAt" | "starte
   if (pending && WAITING_TOOLS.has(pending.name)) return null;
   const background = tools.find((b) => 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 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
     ? `while this step ran: ${stepLabel(pending)}`
     : background

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<MessageRow>("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<MessageBlock[]>(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) };
 }
 
@@ -644,16 +647,40 @@ export async function deleteConversation(id: string): Promise<void> {
 /* 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<ImageRow>("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[] {
-  return all<MessageRow>("SELECT * FROM messages WHERE conversation_id = ? ORDER BY created_at ASC, rowid ASC", conversationId).map(
-    toMessage,
-  );
+  return whole(all<MessageRow>("SELECT * FROM messages WHERE conversation_id = ? ORDER BY created_at ASC, rowid ASC", conversationId).map(toMessage));
 }
 
 export function getMessage(id: string): Message {
   const row = get<MessageRow>("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(
@@ -693,10 +720,14 @@ 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);
   });
   const message = getMessage(id);
   if (opts.emit !== false) bus.emit({ type: "message.updated", message });

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.