feat(agent): adopt AI SDK v7 — crash safety, context pruning, HITL approvals (#3362)

* docs(adr): propose agent AI SDK v7 adoption plan

* fix(ai): bind analyzeResume through aiService in service test

The test destructured analyzeResume as a named export that does not exist; main was red.

* test(agent): keep pure ai helpers real via spread-actual mock factory

* feat(agent): add run guards, patch version guard, run wall-clock timeout

* feat(agent): validate UI messages at the send boundary

* feat(agent): crash-safe draft-row persistence and server-side cancellation

* feat(agent): reap stale run claims at boot, on send, and on thread open

* feat(agent): fresh-document patch output and tiered context pruning

* feat(ai): shared agent tool contracts and message metadata schema

* feat(agent): add per-thread review-patches setting with update endpoint

* feat(agent): gate resume patches behind hmac-signed tool approval

* feat(agent): merge question answers and approval decisions before run claim

* feat(agent): approval ui with composed auto-send and fixture-driven tests

* feat(agent): usage metadata, tool activity cards, smoother streaming

* feat(agent): tool-call repair, input examples, structured step logging

* chore(i18n): translate new agent workspace strings across all locales

* fix(agent): gate stale-run draft cancellation on winning the claim clear

Snapshot streaming drafts before the conditional clear and skip the flip entirely when another reaper or a replacement run already cleared the claim. Also address review nits in eleven locale catalogs.

* fix(agent): flip reaped drafts only when their snapshotted state is unchanged

* fix(agent): address review findings across run lifecycle, context budget, and approval flow

- bind patches to the revision the model read via signed baseUpdatedAt
- claim the run before consuming a continuation; recorded-but-unexecuted approvals retry as pending continuations
- keep run ownership on stop() until cancellation persists; preserve the claim for the reaper when final persistence fails
- estimate tokens without serializing binary attachments (tokenx) and enforce the budget by dropping oldest whole turns
- mark crash-recovered patch results as snapshot boundaries; strip /data prefixes at execution time
- retry failed continuations without regenerate; mount a single AgentChat; disable response controls on read-only threads; freeze review toggle during runs (client+server)
- accumulate usage across continuations and match the SDK's nested usage shape; label-form token strings; reorderable source label; accessible note field; state-neutral web-search label

* chore(i18n): translate revised agent strings across all locales

* fix(agent): harden baseUpdatedAt validation and address review follow-ups

- bundle tokenx in the server runtime dependencies (e2e boot failure)
- strict ISO schema for baseUpdatedAt plus loud executor rejection of unparseable values
- it-IT source label consistency (Fonte)
- prove penultimate-turn retention in the context pruning test

* chore(deps): exempt tokenx from knip for the externalized server bundle

packages/api/package.json

@@ -5,6 +5,7 @@
 	"private": true,
 	"exports": {
 		"./context": "./src/context.ts",
+		"./features/agent/runs": "./src/features/agent/runs.ts",
 		"./features/flags": "./src/features/flags/index.ts",
 		"./features/resume/export": "./src/features/resume/export.ts",
 		"./features/resume/public-pdf": "./src/features/resume/public-pdf.ts",
@@ -52,10 +53,12 @@
 		"drizzle-zod": "1.0.0-beta.14-a36c63d",
 		"es-toolkit": "^1.51.0",
 		"ioredis": "^6.0.0",
+		"jsonrepair": "^3.15.0",
 		"ollama-ai-provider-v2": "^4.0.1",
 		"react": "^19.2.8",
 		"resumable-stream": "^2.2.12",
 		"sharp": "^0.35.3",
+		"tokenx": "^2.1.0",
 		"ts-pattern": "^5.9.0",
 		"zod": "^4.4.3"
 	},

packages/api/src/features/agent/context.test.ts

@@ -0,0 +1,288 @@
+import type { ModelMessage } from "ai";
+import { describe, expect, it } from "vitest";
+import { estimateTokenCount, pruneAgentModelContext } from "./context";
+
+const BIG_RESUME = { basics: { name: "Alice" }, sections: { summary: { content: "x".repeat(2_000) } } };
+
+function readResumeExchange(callId: string): ModelMessage[] {
+	return [
+		{
+			role: "assistant",
+			content: [{ type: "tool-call", toolCallId: callId, toolName: "read_resume", input: {} }],
+		},
+		{
+			role: "tool",
+			content: [
+				{
+					type: "tool-result",
+					toolCallId: callId,
+					toolName: "read_resume",
+					output: { type: "json", value: { id: "resume-1", data: BIG_RESUME } },
+				},
+			],
+		},
+	] as ModelMessage[];
+}
+
+function patchExchange(callId: string): ModelMessage[] {
+	return [
+		{
+			role: "assistant",
+			content: [{ type: "tool-call", toolCallId: callId, toolName: "apply_resume_patch", input: { title: "Edit" } }],
+		},
+		{
+			role: "tool",
+			content: [
+				{
+					type: "tool-result",
+					toolCallId: callId,
+					toolName: "apply_resume_patch",
+					output: { type: "json", value: { actionId: `action-${callId}`, resume: BIG_RESUME } },
+				},
+			],
+		},
+	] as ModelMessage[];
+}
+
+function user(text: string): ModelMessage {
+	return { role: "user", content: [{ type: "text", text }] };
+}
+
+function messageParts(message: ModelMessage | undefined) {
+	return (message?.content ?? []) as Array<Record<string, unknown>>;
+}
+
+function snapshotValue(message: ModelMessage | undefined) {
+	const part = messageParts(message)[0];
+	return ((part?.output ?? {}) as { value?: Record<string, unknown> }).value ?? {};
+}
+
+describe("pruneAgentModelContext — tier 0 (snapshot supersession)", () => {
+	it("keeps only the last resume snapshot even when under budget", () => {
+		const messages = [user("hi"), ...readResumeExchange("call-1"), ...patchExchange("call-2")];
+
+		const pruned = pruneAgentModelContext(messages, 1_000_000);
+
+		expect(snapshotValue(pruned[2])).not.toHaveProperty("data");
+		expect(snapshotValue(pruned[2]).note).toContain("Superseded resume snapshot");
+		expect(snapshotValue(pruned[4])).toHaveProperty("resume");
+	});
+
+	it("returns the same array reference when there is at most one snapshot", () => {
+		const messages = [user("hi"), ...readResumeExchange("call-1")];
+
+		expect(pruneAgentModelContext(messages, 1_000_000)).toBe(messages);
+	});
+
+	it("treats a crash-recovered synthetic patch result (resume: null) as the surviving snapshot boundary", () => {
+		const synthetic: ModelMessage[] = [
+			{
+				role: "assistant",
+				content: [{ type: "tool-call", toolCallId: "synthetic-1", toolName: "apply_resume_patch", input: {} }],
+			},
+			{
+				role: "tool",
+				content: [
+					{
+						type: "tool-result",
+						toolCallId: "synthetic-1",
+						toolName: "apply_resume_patch",
+						output: { type: "json", value: { actionId: "action-1", resume: null, note: "Re-read the resume" } },
+					},
+				],
+			},
+		];
+
+		const pruned = pruneAgentModelContext([user("hi"), ...readResumeExchange("call-1"), ...synthetic], 1_000_000);
+
+		// The older full read_resume snapshot is superseded; the recovery note survives.
+		expect(snapshotValue(pruned[2])).not.toHaveProperty("data");
+		expect(snapshotValue(pruned[4])).toMatchObject({ note: "Re-read the resume" });
+	});
+
+	it("keeps non-snapshot fields of a superseded patch result (indexes may matter)", () => {
+		const messages = [user("hi"), ...patchExchange("call-1"), ...patchExchange("call-2")];
+
+		const pruned = pruneAgentModelContext(messages, 1_000_000);
+
+		expect(snapshotValue(pruned[2])).toMatchObject({ actionId: "action-call-1" });
+		expect(snapshotValue(pruned[2])).not.toHaveProperty("resume");
+	});
+});
+
+describe("pruneAgentModelContext — tier 1 (reasoning)", () => {
+	it("strips reasoning from all but the last assistant message when over budget", () => {
+		const messages: ModelMessage[] = [
+			user("hi"),
+			{
+				role: "assistant",
+				content: [
+					{ type: "reasoning", text: "r".repeat(400) },
+					{ type: "text", text: "First answer" },
+				],
+			},
+			user("more"),
+			{
+				role: "assistant",
+				content: [
+					{ type: "reasoning", text: "keep me" },
+					{ type: "text", text: "Second answer" },
+				],
+			},
+		];
+
+		const pruned = pruneAgentModelContext(messages, 50);
+
+		const firstAssistant = pruned.find((message) => message.role === "assistant");
+		expect(JSON.stringify(firstAssistant)).not.toContain("rrrr");
+		expect(JSON.stringify(pruned.at(-1))).toContain("keep me");
+	});
+});
+
+describe("pruneAgentModelContext — tier 2 (tool pairs)", () => {
+	it("stubs old pairs together, protecting the surviving snapshot and the last assistant message", () => {
+		const messages = [
+			user("hi"),
+			...patchExchange("call-1"),
+			...patchExchange("call-2"),
+			{
+				role: "assistant",
+				content: [{ type: "text", text: "All done" }],
+			} as ModelMessage,
+		];
+
+		const pruned = pruneAgentModelContext(messages, 100);
+
+		// Oldest pair stubbed on both sides.
+		const firstCall = messageParts(pruned[1])[0];
+		expect(firstCall?.input).toEqual({});
+		expect(snapshotValue(pruned[2]).note).toBeDefined();
+		// Surviving snapshot pair untouched by tier 2 (still carries the resume).
+		expect(snapshotValue(pruned[4])).toHaveProperty("resume");
+	});
+
+	it("never stubs a call that has no result (unresolved question stays intact)", () => {
+		const question = {
+			role: "assistant",
+			content: [
+				{
+					type: "tool-call",
+					toolCallId: "call-q",
+					toolName: "ask_user_question",
+					input: { question: "?".repeat(400) },
+				},
+			],
+		} as ModelMessage;
+
+		const pruned = pruneAgentModelContext([user("hi"), question, user("answer pending")], 20);
+
+		const call = messageParts(pruned[1])[0];
+		expect(call?.input).toEqual({ question: "?".repeat(400) });
+	});
+
+	it("skips messages carrying approval content", () => {
+		const approval = {
+			role: "assistant",
+			content: [
+				{ type: "tool-call", toolCallId: "call-a", toolName: "apply_resume_patch", input: { title: "x".repeat(400) } },
+				{ type: "tool-approval-request", approvalId: "approval-1", toolCallId: "call-a" },
+			],
+		} as ModelMessage;
+		const result = {
+			role: "tool",
+			content: [
+				{
+					type: "tool-result",
+					toolCallId: "call-a",
+					toolName: "apply_resume_patch",
+					output: { type: "json", value: { actionId: "action-a" } },
+				},
+			],
+		} as ModelMessage;
+
+		const pruned = pruneAgentModelContext([user("hi"), approval, result, user("next")], 20);
+
+		const call = messageParts(pruned[1])[0];
+		expect(call?.input).toEqual({ title: "x".repeat(400) });
+	});
+});
+
+describe("pruneAgentModelContext — tier 3 (turn dropping)", () => {
+	function textTurn(userText: string, assistantText: string): ModelMessage[] {
+		return [
+			{ role: "user", content: [{ type: "text", text: userText }] },
+			{ role: "assistant", content: [{ type: "text", text: assistantText }] },
+		];
+	}
+
+	it("drops the oldest complete turns until the budget is met", () => {
+		const messages = [
+			...textTurn("old question ".repeat(50), "old answer ".repeat(50)),
+			...textTurn("middle question ".repeat(50), "middle answer ".repeat(50)),
+			...textTurn("latest question", "latest answer"),
+		];
+
+		const pruned = pruneAgentModelContext(messages, 60);
+
+		expect(JSON.stringify(pruned)).not.toContain("old question");
+		expect(JSON.stringify(pruned)).toContain("latest question");
+		expect(JSON.stringify(pruned)).toContain("latest answer");
+	});
+
+	it("drops tool call/result pairs atomically with their turn", () => {
+		const messages = [
+			...textTurn("intro ".repeat(60), "ok"),
+			{ role: "user", content: [{ type: "text", text: "edit please ".repeat(60) }] } as ModelMessage,
+			...patchExchange("call-old"),
+			...textTurn("follow up ".repeat(30), "done"),
+			...textTurn("latest question", "latest answer"),
+		];
+
+		const pruned = pruneAgentModelContext(messages, 40);
+
+		// The dropped turn takes both the tool call and its result with it — no orphaned side.
+		const text = JSON.stringify(pruned);
+		expect(text).not.toContain("call-old");
+		expect(text).not.toContain('"tool-result"');
+		expect(text).toContain("latest question");
+	});
+
+	it("never drops the final two turns even when still over budget", () => {
+		const messages = [
+			...textTurn("first question ".repeat(100), "first answer ".repeat(100)),
+			...textTurn("second question ".repeat(100), "second answer ".repeat(100)),
+			...textTurn("third question ".repeat(100), "third answer ".repeat(100)),
+		];
+
+		const pruned = pruneAgentModelContext(messages, 10);
+
+		// Only the oldest turn is droppable; the penultimate and final turns must both survive
+		// even though the result is still over budget.
+		expect(pruned).toHaveLength(4);
+		expect(JSON.stringify(pruned)).not.toContain("first question");
+		expect(JSON.stringify(pruned)).toContain("second question");
+		expect(JSON.stringify(pruned)).toContain("third question");
+	});
+});
+
+describe("estimateTokenCount", () => {
+	it("estimates natural text at roughly one token per word", () => {
+		expect(estimateTokenCount("old question ".repeat(50))).toBeGreaterThanOrEqual(90);
+		expect(estimateTokenCount("old question ".repeat(50))).toBeLessThan(120);
+		expect(estimateTokenCount({ a: 1 })).toBeGreaterThan(0);
+	});
+
+	it("treats binary attachment data as opaque bytes instead of serializing it", () => {
+		const bytes = new Uint8Array(1024 * 1024);
+		const message = { role: "user", content: [{ type: "image", image: bytes, mediaType: "image/png" }] };
+
+		const started = performance.now();
+		const estimate = estimateTokenCount(message);
+		const elapsedMs = performance.now() - started;
+
+		// ~bytes/4 tokens, computed without expanding each byte into JSON.
+		expect(estimate).toBeGreaterThan(200_000);
+		expect(estimate).toBeLessThan(300_000);
+		expect(elapsedMs).toBeLessThan(200);
+	});
+});

packages/api/src/features/agent/context.ts

@@ -0,0 +1,220 @@
+import type { ModelMessage } from "ai";
+import { pruneMessages } from "ai";
+import { estimateTokenCount as estimateTextTokenCount } from "tokenx";
+
+// Pure model-context pruning, wired into the agent loop via prepareStep so it runs before every
+// step. Tier 0 always runs (a stale resume snapshot is actively harmful — shifted array indexes);
+// the remaining tiers only fire once the estimated token count exceeds the budget.
+
+const AGENT_CONTEXT_TOKEN_BUDGET = 40_000;
+
+const SNAPSHOT_TOOL_NAMES = new Set(["read_resume", "apply_resume_patch"]);
+const SUPERSEDED_SNAPSHOT_NOTE =
+	"Superseded resume snapshot removed. Base further edits on the resume state in the latest read_resume or apply_resume_patch result.";
+const PRUNED_TOOL_RESULT_NOTE = "Older tool result pruned to fit the context budget.";
+
+type LoosePart = Record<string, unknown> & { type: string };
+
+// Binary attachment data (up to 25MB per file) must never be JSON.stringified — each byte would
+// expand into a numeric-key object entry and repeated estimation would exhaust the heap.
+function isBinary(value: unknown): value is ArrayBufferView | ArrayBuffer {
+	return ArrayBuffer.isView(value) || value instanceof ArrayBuffer;
+}
+
+// Manual walk instead of JSON.stringify: a Buffer's toJSON() runs before any stringify replacer,
+// so serialization-based estimation would expand binary into per-byte entries (hundreds of MB of
+// JSON for one allowed 25MB attachment). String leaves go through tokenx; binary counts as
+// opaque bytes; structural syntax gets a small flat cost.
+function estimateTokens(value: unknown): number {
+	if (value === null || value === undefined) return 1;
+	if (typeof value === "string") return estimateTextTokenCount(value);
+	if (typeof value === "number" || typeof value === "boolean" || typeof value === "bigint") return 2;
+	if (isBinary(value)) return Math.ceil(value.byteLength / 4);
+	if (Array.isArray(value)) {
+		let sum = 1;
+		for (const item of value) sum += estimateTokens(item) + 1;
+		return sum;
+	}
+	if (typeof value === "object") {
+		let sum = 1;
+		for (const [key, nested] of Object.entries(value)) sum += estimateTextTokenCount(key) + 1 + estimateTokens(nested);
+		return sum;
+	}
+	return 2;
+}
+
+export function estimateTokenCount(value: unknown): number {
+	return estimateTokens(value);
+}
+
+function contentParts(message: ModelMessage): LoosePart[] {
+	return Array.isArray(message.content) ? (message.content as unknown as LoosePart[]) : [];
+}
+
+function unwrapToolOutput(output: unknown): unknown {
+	if (output && typeof output === "object" && "type" in output && "value" in output) {
+		return (output as { value: unknown }).value;
+	}
+	return output;
+}
+
+function wrapToolOutput(original: unknown, value: unknown): unknown {
+	if (original && typeof original === "object" && "type" in original && "value" in original) {
+		return { ...(original as Record<string, unknown>), type: "json", value };
+	}
+	return value;
+}
+
+function isSnapshotResultPart(part: LoosePart): boolean {
+	if (part.type !== "tool-result" || typeof part.toolName !== "string" || !SNAPSHOT_TOOL_NAMES.has(part.toolName)) {
+		return false;
+	}
+
+	const value = unwrapToolOutput(part.output);
+	if (!value || typeof value !== "object") return false;
+
+	return part.toolName === "read_resume" ? "data" in value : "resume" in value;
+}
+
+function supersedeSnapshotPart(part: LoosePart): LoosePart {
+	const value = unwrapToolOutput(part.output);
+	const record = value && typeof value === "object" ? (value as Record<string, unknown>) : {};
+	const { data: _data, resume: _resume, ...rest } = record;
+
+	return { ...part, output: wrapToolOutput(part.output, { ...rest, note: SUPERSEDED_SNAPSHOT_NOTE }) };
+}
+
+// Tier 0: exactly one full resume snapshot survives — the last one, positioned where the model
+// last acted. Runs regardless of budget.
+function supersedeStaleSnapshots(messages: ModelMessage[]): ModelMessage[] {
+	const locations: Array<{ messageIndex: number; partIndex: number }> = [];
+
+	for (const [messageIndex, message] of messages.entries()) {
+		for (const [partIndex, part] of contentParts(message).entries()) {
+			if (isSnapshotResultPart(part)) locations.push({ messageIndex, partIndex });
+		}
+	}
+
+	if (locations.length <= 1) return messages;
+
+	const stale = locations.slice(0, -1);
+	const next = [...messages];
+	for (const { messageIndex, partIndex } of stale) {
+		const message = next[messageIndex] as ModelMessage & { content: LoosePart[] };
+		const parts = [...contentParts(message)];
+		// biome-ignore lint/style/noNonNullAssertion: location was collected from this array
+		parts[partIndex] = supersedeSnapshotPart(parts[partIndex]!);
+		next[messageIndex] = { ...message, content: parts } as ModelMessage;
+	}
+
+	return next;
+}
+
+function hasApprovalContent(message: ModelMessage): boolean {
+	return contentParts(message).some(
+		(part) => part.type === "tool-approval-request" || part.type === "tool-approval-response",
+	);
+}
+
+function lastIndexWhere<T>(items: T[], predicate: (item: T) => boolean): number {
+	for (let index = items.length - 1; index >= 0; index--) {
+		// biome-ignore lint/style/noNonNullAssertion: index is in range
+		if (predicate(items[index]!)) return index;
+	}
+	return -1;
+}
+
+// Tier 2: collapse the oldest tool call/result pairs into stubs — always both sides of a pair
+// (several BYOK gateways reject unpaired tool messages), never the protected regions.
+function collapseOldestToolPairs(messages: ModelMessage[], budget: number): ModelMessage[] {
+	const lastAssistantIndex = lastIndexWhere(messages, (message) => message.role === "assistant");
+	const survivingSnapshotCallIds = new Set<string>();
+
+	// The surviving snapshot (last one, by Tier 0) must keep its full pair.
+	for (let index = messages.length - 1; index >= 0; index--) {
+		// biome-ignore lint/style/noNonNullAssertion: index is in range
+		const part = contentParts(messages[index]!).findLast(isSnapshotResultPart);
+		if (part && typeof part.toolCallId === "string") {
+			survivingSnapshotCallIds.add(part.toolCallId);
+			break;
+		}
+	}
+
+	const resolvedCallIds = new Set<string>();
+	for (const message of messages) {
+		for (const part of contentParts(message)) {
+			if (part.type === "tool-result" && typeof part.toolCallId === "string") resolvedCallIds.add(part.toolCallId);
+		}
+	}
+
+	const next = [...messages];
+	const stubbedCallIds = new Set<string>();
+	let changed = false;
+
+	for (const [index, message] of next.entries()) {
+		if (estimateTokenCount(next) <= budget) break;
+		if (index === lastAssistantIndex || hasApprovalContent(message)) continue;
+		if (message.role !== "assistant" && message.role !== "tool") continue;
+
+		const parts = contentParts(message).map((part) => {
+			const toolCallId = typeof part.toolCallId === "string" ? part.toolCallId : null;
+			if (!toolCallId || survivingSnapshotCallIds.has(toolCallId)) return part;
+
+			if (part.type === "tool-call" && resolvedCallIds.has(toolCallId)) {
+				stubbedCallIds.add(toolCallId);
+				return { ...part, input: {} };
+			}
+
+			if (part.type === "tool-result" && stubbedCallIds.has(toolCallId)) {
+				return { ...part, output: wrapToolOutput(part.output, { note: PRUNED_TOOL_RESULT_NOTE }) };
+			}
+
+			return part;
+		});
+
+		if (parts.some((part, partIndex) => part !== contentParts(message)[partIndex])) {
+			next[index] = { ...message, content: parts } as ModelMessage;
+			changed = true;
+		}
+	}
+
+	return changed ? next : messages;
+}
+
+// Tier 3 (last resort): drop the oldest complete turns. A turn is one user message plus every
+// message up to the next user message, so tool call/result and approval request/response pairs
+// are always dropped atomically. The final two turns (latest user turn and the trailing
+// assistant turn, when present) are never dropped.
+function dropOldestTurns(messages: ModelMessage[], budget: number): ModelMessage[] {
+	const turnStarts: number[] = [];
+	for (const [index, message] of messages.entries()) {
+		if (message.role === "user" || turnStarts.length === 0) turnStarts.push(index);
+	}
+	if (turnStarts.length <= 2) return messages;
+
+	const turns = turnStarts.map((start, turnIndex) => messages.slice(start, turnStarts[turnIndex + 1]));
+	let dropCount = 0;
+	while (dropCount < turns.length - 2 && estimateTokenCount(turns.slice(dropCount).flat()) > budget) {
+		dropCount += 1;
+	}
+
+	return dropCount === 0 ? messages : turns.slice(dropCount).flat();
+}
+
+export function pruneAgentModelContext(
+	messages: ModelMessage[],
+	budget: number = AGENT_CONTEXT_TOKEN_BUDGET,
+): ModelMessage[] {
+	let current = supersedeStaleSnapshots(messages);
+	if (estimateTokenCount(current) <= budget) return current;
+
+	// Tier 1: strip reasoning from all but the last assistant message.
+	const withoutReasoning = pruneMessages({ messages: current, reasoning: "before-last-message" });
+	if (estimateTokenCount(withoutReasoning) < estimateTokenCount(current)) current = withoutReasoning;
+	if (estimateTokenCount(current) <= budget) return current;
+
+	current = collapseOldestToolPairs(current, budget);
+	if (estimateTokenCount(current) <= budget) return current;
+
+	return dropOldestTurns(current, budget);
+}

packages/api/src/features/agent/messages-merge.test.ts

@@ -0,0 +1,168 @@
+import type { UIMessage } from "ai";
+import { describe, expect, it } from "vitest";
+import { mergeClientToolResponses } from "./messages-merge";
+
+type LoosePart = Record<string, unknown> & { type: string };
+
+function assistantMessage(parts: LoosePart[]): UIMessage {
+	return { id: "ui-1", role: "assistant", parts: parts as UIMessage["parts"] };
+}
+
+const QUESTION_PENDING: LoosePart = {
+	type: "tool-ask_user_question",
+	toolCallId: "call-q",
+	state: "input-available",
+	input: { question: "Which tone?" },
+};
+
+const APPROVAL_REQUESTED: LoosePart = {
+	type: "tool-apply_resume_patch",
+	toolCallId: "call-p",
+	state: "approval-requested",
+	input: { title: "Edit", operations: [] },
+	approval: { id: "approval-1", signature: "server-signature" },
+};
+
+function approvalResponse(approved: boolean, reason?: string): LoosePart {
+	return {
+		type: "tool-apply_resume_patch",
+		toolCallId: "call-p",
+		state: "approval-responded",
+		input: { title: "Edit", operations: [] },
+		approval: { id: "approval-1", approved, ...(reason ? { reason } : {}), signature: "client-signature" },
+	};
+}
+
+function questionAnswer(output = "Formal"): LoosePart {
+	return { type: "tool-ask_user_question", toolCallId: "call-q", state: "output-available", input: {}, output };
+}
+
+describe("mergeClientToolResponses — questions (legacy behavior)", () => {
+	it("merges an answer into a pending question", () => {
+		const result = mergeClientToolResponses(assistantMessage([QUESTION_PENDING]), assistantMessage([questionAnswer()]));
+
+		expect(result.mergedCount).toBe(1);
+		expect(result.message.parts[0]).toMatchObject({ state: "output-available", output: "Formal" });
+	});
+
+	it("merges an error answer with its errorText", () => {
+		const result = mergeClientToolResponses(
+			assistantMessage([QUESTION_PENDING]),
+			assistantMessage([
+				{ type: "tool-ask_user_question", toolCallId: "call-q", state: "output-error", errorText: "boom" },
+			]),
+		);
+
+		expect(result.message.parts[0]).toMatchObject({ state: "output-error", errorText: "boom" });
+	});
+
+	it("counts a re-submitted answer as already resolved", () => {
+		const answered = assistantMessage([{ ...QUESTION_PENDING, state: "output-available", output: "Formal" }]);
+
+		const result = mergeClientToolResponses(answered, assistantMessage([questionAnswer()]));
+
+		expect(result.mergedCount).toBe(0);
+		expect(result.alreadyResolvedCount).toBe(1);
+	});
+
+	it("returns zero counts when nothing matches", () => {
+		const result = mergeClientToolResponses(assistantMessage([QUESTION_PENDING]), assistantMessage([]));
+
+		expect(result).toMatchObject({ mergedCount: 0, alreadyResolvedCount: 0, conflictingCount: 0 });
+	});
+});
+
+describe("mergeClientToolResponses — approvals", () => {
+	it("approves: flips the stored part but keeps the server-signed request payload", () => {
+		const result = mergeClientToolResponses(
+			assistantMessage([APPROVAL_REQUESTED]),
+			assistantMessage([approvalResponse(true)]),
+		);
+
+		expect(result.mergedCount).toBe(1);
+		expect(result.message.parts[0]).toMatchObject({
+			state: "approval-responded",
+			approval: { id: "approval-1", approved: true, signature: "server-signature" },
+		});
+	});
+
+	it("denies with a reason", () => {
+		const result = mergeClientToolResponses(
+			assistantMessage([APPROVAL_REQUESTED]),
+			assistantMessage([approvalResponse(false, "Wrong section")]),
+		);
+
+		expect(result.message.parts[0]).toMatchObject({
+			state: "approval-responded",
+			approval: { approved: false, reason: "Wrong section", signature: "server-signature" },
+		});
+	});
+
+	it("resubmitting the same decision on a responded-but-unexecuted approval is a pending continuation", () => {
+		const responded = assistantMessage([
+			{ ...APPROVAL_REQUESTED, state: "approval-responded", approval: { id: "approval-1", approved: true } },
+		]);
+
+		const result = mergeClientToolResponses(responded, assistantMessage([approvalResponse(true)]));
+
+		expect(result).toMatchObject({
+			mergedCount: 0,
+			alreadyResolvedCount: 0,
+			pendingContinuationCount: 1,
+			conflictingCount: 0,
+		});
+	});
+
+	it("resubmitting the decision after the approved call executed is already resolved", () => {
+		const executed = assistantMessage([
+			{
+				...APPROVAL_REQUESTED,
+				state: "output-available",
+				output: { actionId: "action-1" },
+				approval: { id: "approval-1", approved: true },
+			},
+		]);
+
+		const result = mergeClientToolResponses(executed, assistantMessage([approvalResponse(true)]));
+
+		expect(result).toMatchObject({ mergedCount: 0, alreadyResolvedCount: 1, pendingContinuationCount: 0 });
+	});
+
+	it("double submit with a conflicting decision is flagged", () => {
+		const responded = assistantMessage([
+			{ ...APPROVAL_REQUESTED, state: "approval-responded", approval: { id: "approval-1", approved: true } },
+		]);
+
+		const result = mergeClientToolResponses(responded, assistantMessage([approvalResponse(false)]));
+
+		expect(result).toMatchObject({ mergedCount: 0, alreadyResolvedCount: 0, conflictingCount: 1 });
+	});
+
+	it("a response for a different approval id does not touch the stored part", () => {
+		const result = mergeClientToolResponses(
+			assistantMessage([APPROVAL_REQUESTED]),
+			assistantMessage([
+				{
+					...approvalResponse(true),
+					approval: { id: "approval-other", approved: true },
+				},
+			]),
+		);
+
+		expect(result.mergedCount).toBe(0);
+		expect(result.message.parts[0]).toMatchObject({ state: "approval-requested" });
+	});
+});
+
+describe("mergeClientToolResponses — mixed", () => {
+	it("handles a question answer and an approval decision in one pass", () => {
+		const result = mergeClientToolResponses(
+			assistantMessage([QUESTION_PENDING, APPROVAL_REQUESTED]),
+			assistantMessage([questionAnswer(), approvalResponse(true)]),
+		);
+
+		expect(result.mergedCount).toBe(2);
+		expect(result.message.parts[0]).toMatchObject({ state: "output-available" });
+		expect(result.message.parts[1]).toMatchObject({ state: "approval-responded" });
+	});
+});

packages/api/src/features/agent/messages-merge.ts

@@ -0,0 +1,145 @@
+import type { UIMessage } from "ai";
+
+// Pure merge of client-authored tool responses into a stored assistant message. Handles both
+// ask_user_question answers and tool-approval responses in one pass (a message can carry both).
+// For approvals only the client's decision fields are copied — the stored, server-signed request
+// payload (approval id + signature) is kept, so a client cannot substitute a forged request.
+
+type AgentToolPart = UIMessage["parts"][number] & {
+	toolCallId?: string;
+	state?: string;
+	output?: unknown;
+	errorText?: string;
+	approval?: {
+		id: string;
+		approved?: boolean;
+		reason?: string;
+		isAutomatic?: boolean;
+		signature?: string;
+	};
+};
+
+type ApprovalDecision = { approved: boolean; reason?: string };
+
+export type MergeClientToolResponsesResult = {
+	message: UIMessage;
+	mergedCount: number;
+	/** Matches on terminal parts (question answered, approval executed or denied). */
+	alreadyResolvedCount: number;
+	/**
+	 * Matches on approval parts that are responded but not yet executed — a continuation run was
+	 * claimed (or persisted) earlier but never completed. Callers should proceed with a run so the
+	 * recorded decision can still execute instead of stranding it.
+	 */
+	pendingContinuationCount: number;
+	conflictingCount: number;
+};
+
+function isAnsweredQuestionPart(part: AgentToolPart): part is AgentToolPart & { toolCallId: string } {
+	return (
+		part.type === "tool-ask_user_question" &&
+		typeof part.toolCallId === "string" &&
+		(part.state === "output-available" || part.state === "output-error")
+	);
+}
+
+function approvalKey(toolCallId: string, approvalId: string) {
+	return `${toolCallId}:${approvalId}`;
+}
+
+export function mergeClientToolResponses(
+	existingMessage: UIMessage,
+	incomingMessage: UIMessage,
+): MergeClientToolResponsesResult {
+	const answeredQuestions = new Map<string, AgentToolPart>();
+	const approvalDecisions = new Map<string, ApprovalDecision>();
+
+	for (const part of incomingMessage.parts as AgentToolPart[]) {
+		if (isAnsweredQuestionPart(part)) {
+			answeredQuestions.set(part.toolCallId, part);
+			continue;
+		}
+
+		if (
+			part.state === "approval-responded" &&
+			typeof part.toolCallId === "string" &&
+			typeof part.approval?.id === "string" &&
+			typeof part.approval.approved === "boolean"
+		) {
+			approvalDecisions.set(approvalKey(part.toolCallId, part.approval.id), {
+				approved: part.approval.approved,
+				...(part.approval.reason ? { reason: part.approval.reason } : {}),
+			});
+		}
+	}
+
+	let mergedCount = 0;
+	let alreadyResolvedCount = 0;
+	let pendingContinuationCount = 0;
+	let conflictingCount = 0;
+
+	const parts = existingMessage.parts.map((part) => {
+		const existingPart = part as AgentToolPart;
+
+		if (existingPart.type === "tool-ask_user_question" && typeof existingPart.toolCallId === "string") {
+			const answer = answeredQuestions.get(existingPart.toolCallId);
+			if (answer) {
+				if (existingPart.state === "input-available") {
+					mergedCount += 1;
+					if (answer.state === "output-error") {
+						return {
+							...part,
+							state: "output-error",
+							errorText: answer.errorText ?? "User answer failed.",
+						} as UIMessage["parts"][number];
+					}
+
+					return { ...part, state: "output-available", output: answer.output } as UIMessage["parts"][number];
+				}
+
+				if (existingPart.state === "output-available" || existingPart.state === "output-error") {
+					alreadyResolvedCount += 1;
+				}
+
+				return part;
+			}
+		}
+
+		if (typeof existingPart.toolCallId === "string" && typeof existingPart.approval?.id === "string") {
+			const decision = approvalDecisions.get(approvalKey(existingPart.toolCallId, existingPart.approval.id));
+			if (decision) {
+				if (existingPart.state === "approval-requested") {
+					mergedCount += 1;
+					return {
+						...part,
+						state: "approval-responded",
+						// Keep the stored (signed) request payload; copy only the decision fields.
+						approval: {
+							...existingPart.approval,
+							approved: decision.approved,
+							...(decision.reason ? { reason: decision.reason } : {}),
+						},
+					} as UIMessage["parts"][number];
+				}
+
+				if (typeof existingPart.approval.approved === "boolean") {
+					if (existingPart.approval.approved !== decision.approved) conflictingCount += 1;
+					else if (existingPart.state === "approval-responded") pendingContinuationCount += 1;
+					else alreadyResolvedCount += 1;
+				}
+
+				return part;
+			}
+		}
+
+		return part;
+	});
+
+	return {
+		message: { ...existingMessage, parts },
+		mergedCount,
+		alreadyResolvedCount,
+		pendingContinuationCount,
+		conflictingCount,
+	};
+}

packages/api/src/features/agent/messages-persistence.test.ts

@@ -0,0 +1,227 @@
+import type { UIMessage } from "ai";
+import { describe, expect, it } from "vitest";
+import { applyStepToUiMessage, upsertAssistantUiMessage, withAccumulatedUsageMetadata } from "./messages-persistence";
+
+function emptyMessage(): UIMessage {
+	return { id: "ui-1", role: "assistant", parts: [] };
+}
+
+describe("applyStepToUiMessage", () => {
+	it("folds text, reasoning, and paired tool call/result content into UI parts", () => {
+		const folded = applyStepToUiMessage(emptyMessage(), {
+			content: [
+				{ type: "reasoning", text: "thinking" },
+				{ type: "tool-call", toolCallId: "call-1", toolName: "apply_resume_patch", input: { title: "Edit" } },
+				{
+					type: "tool-result",
+					toolCallId: "call-1",
+					toolName: "apply_resume_patch",
+					output: { actionId: "action-1" },
+				},
+				{ type: "text", text: "Done." },
+			],
+		});
+
+		expect(folded.parts).toEqual([
+			{ type: "step-start" },
+			{ type: "reasoning", text: "thinking" },
+			{
+				type: "tool-apply_resume_patch",
+				toolCallId: "call-1",
+				state: "output-available",
+				input: { title: "Edit" },
+				output: { actionId: "action-1" },
+			},
+			{ type: "text", text: "Done." },
+		]);
+	});
+
+	it("appends to existing parts instead of replacing them", () => {
+		const first = applyStepToUiMessage(emptyMessage(), { content: [{ type: "text", text: "one" }] });
+		const second = applyStepToUiMessage(first, { content: [{ type: "text", text: "two" }] });
+
+		expect(second.parts.map((part) => part.type)).toEqual(["step-start", "text", "step-start", "text"]);
+	});
+
+	it("marks tool errors as output-error with the error message", () => {
+		const folded = applyStepToUiMessage(emptyMessage(), {
+			content: [
+				{ type: "tool-call", toolCallId: "call-1", toolName: "apply_resume_patch", input: {} },
+				{
+					type: "tool-error",
+					toolCallId: "call-1",
+					toolName: "apply_resume_patch",
+					input: {},
+					error: new Error("The resume changed"),
+				},
+			],
+		});
+
+		expect(folded.parts.at(-1)).toMatchObject({ state: "output-error", errorText: "The resume changed" });
+	});
+
+	it("keeps an unpaired tool result by synthesizing a complete tool part", () => {
+		const folded = applyStepToUiMessage(emptyMessage(), {
+			content: [{ type: "tool-result", toolCallId: "call-9", toolName: "read_resume", input: {}, output: { id: "r" } }],
+		});
+
+		expect(folded.parts.at(-1)).toMatchObject({
+			type: "tool-read_resume",
+			toolCallId: "call-9",
+			state: "output-available",
+		});
+	});
+
+	it("folds dynamic tool calls into dynamic-tool parts", () => {
+		const folded = applyStepToUiMessage(emptyMessage(), {
+			content: [
+				{ type: "tool-call", toolCallId: "call-2", toolName: "web_search", input: { q: "x" }, dynamic: true },
+				{ type: "tool-result", toolCallId: "call-2", toolName: "web_search", output: [], dynamic: true },
+			],
+		});
+
+		expect(folded.parts.at(-1)).toMatchObject({
+			type: "dynamic-tool",
+			toolName: "web_search",
+			state: "output-available",
+		});
+	});
+
+	it("skips sources, files, and approval content", () => {
+		const folded = applyStepToUiMessage(emptyMessage(), {
+			content: [
+				{ type: "source", sourceType: "url", url: "https://example.com" },
+				{ type: "file", file: {} },
+				{ type: "tool-approval-request", approvalId: "a-1" },
+			],
+		});
+
+		expect(folded.parts).toEqual([{ type: "step-start" }]);
+	});
+});
+
+describe("withAccumulatedUsageMetadata", () => {
+	function messageWithUsage(usage: Record<string, unknown>): UIMessage {
+		return { id: "ui-1", role: "assistant", parts: [], metadata: { model: "gpt-5", usage } } as UIMessage;
+	}
+
+	it("sums token counts across a continuation, including nested details", () => {
+		const previous = messageWithUsage({
+			inputTokens: 100,
+			outputTokens: 50,
+			totalTokens: 150,
+			inputTokenDetails: { cacheReadTokens: 40 },
+			outputTokenDetails: { reasoningTokens: 10 },
+		});
+		const next = messageWithUsage({
+			inputTokens: 200,
+			outputTokens: 30,
+			totalTokens: 230,
+			inputTokenDetails: { cacheReadTokens: 60, cacheWriteTokens: 5 },
+		});
+
+		const merged = withAccumulatedUsageMetadata(previous, next) as UIMessage & {
+			metadata: { usage: Record<string, unknown> };
+		};
+
+		expect(merged.metadata.usage).toMatchObject({
+			inputTokens: 300,
+			outputTokens: 80,
+			totalTokens: 380,
+			inputTokenDetails: { cacheReadTokens: 100, cacheWriteTokens: 5 },
+			outputTokenDetails: { reasoningTokens: 10 },
+		});
+	});
+
+	it("returns the next message unchanged when either side has no usage", () => {
+		const next = messageWithUsage({ totalTokens: 42 });
+		const noUsage: UIMessage = { id: "ui-1", role: "assistant", parts: [] };
+
+		expect(withAccumulatedUsageMetadata(noUsage, next)).toBe(next);
+		expect(withAccumulatedUsageMetadata(next, noUsage)).toBe(noUsage);
+	});
+});
+
+type ScriptedDb = {
+	updates: unknown[];
+	inserts: unknown[];
+};
+
+// Minimal scripted stand-in for the drizzle client: each update() consumes the next scripted
+// returning() result; insert() always succeeds. No vi.mock — the database is an injected value.
+function scriptedDatabase(updateResults: Array<Array<{ id: string }>>): ScriptedDb & Record<string, unknown> {
+	const state: ScriptedDb = { updates: [], inserts: [] };
+	let updateCall = 0;
+
+	return {
+		updates: state.updates,
+		inserts: state.inserts,
+		select: () => ({
+			from: () => ({
+				where: async () => [{ maxSequence: 3 }],
+			}),
+		}),
+		update: () => ({
+			set: (value: unknown) => {
+				state.updates.push(value);
+				return {
+					where: () => {
+						const result = updateResults[updateCall] ?? [];
+						updateCall += 1;
+						return Object.assign(Promise.resolve(undefined), {
+							returning: async () => result,
+						});
+					},
+				};
+			},
+		}),
+		insert: () => ({
+			values: (value: unknown) => {
+				state.inserts.push(value);
+				return { returning: async () => [{ id: "inserted-row" }] };
+			},
+		}),
+		delete: () => ({ where: async () => undefined }),
+	};
+}
+
+describe("upsertAssistantUiMessage", () => {
+	const message: UIMessage = { id: "ui-1", role: "assistant", parts: [{ type: "text", text: "hi" }] };
+
+	it("updates by row id when the row exists", async () => {
+		const database = scriptedDatabase([[{ id: "row-1" }], []]);
+
+		const result = await upsertAssistantUiMessage(
+			{ userId: "user-1", threadId: "thread-1", rowId: "row-1", message, status: "streaming" },
+			database as never,
+		);
+
+		expect(result).toEqual({ rowId: "row-1" });
+		expect(database.inserts).toHaveLength(0);
+	});
+
+	it("falls back to matching the stored uiMessage id, then updates in place", async () => {
+		const database = scriptedDatabase([[{ id: "row-2" }]]);
+
+		const result = await upsertAssistantUiMessage(
+			{ userId: "user-1", threadId: "thread-1", message, status: "completed" },
+			database as never,
+		);
+
+		expect(result).toEqual({ rowId: "row-2" });
+		expect(database.inserts).toHaveLength(0);
+	});
+
+	it("inserts a new row only when no existing row matches", async () => {
+		const database = scriptedDatabase([[], []]);
+
+		const result = await upsertAssistantUiMessage(
+			{ userId: "user-1", threadId: "thread-1", rowId: "row-gone", message, status: "completed" },
+			database as never,
+		);
+
+		expect(result).toEqual({ rowId: "inserted-row" });
+		expect(database.inserts).toHaveLength(1);
+		expect(database.inserts[0]).toMatchObject({ role: "assistant", status: "completed", sequence: 4 });
+	});
+});

packages/api/src/features/agent/messages-persistence.ts

@@ -0,0 +1,265 @@
+import type { UIMessage } from "ai";
+import { and, eq, max, sql } from "drizzle-orm";
+import { db } from "@reactive-resume/db/client";
+import * as schema from "@reactive-resume/db/schema";
+
+// Crash-safe incremental persistence for the assistant message of an agent run.
+// A draft row is inserted before the stream starts, folded step-by-step in onStepEnd,
+// and finalized (upserted) with the SDK's authoritative response message in onFinish.
+// All functions take an injectable database so tests can stub queries without
+// touching the send path's positional db mock.
+
+type AgentMessagesDb = Pick<typeof db, "select" | "insert" | "update" | "delete">;
+
+type UiMessagePart = UIMessage["parts"][number];
+
+type StepContentPart = Record<string, unknown> & { type: string };
+
+type AgentStepLike = { content: ReadonlyArray<unknown> };
+
+function toolPartFromCall(part: StepContentPart): UiMessagePart {
+	if (part.dynamic) {
+		return {
+			type: "dynamic-tool",
+			toolName: String(part.toolName),
+			toolCallId: String(part.toolCallId),
+			state: "input-available",
+			input: part.input,
+		} as UiMessagePart;
+	}
+
+	return {
+		type: `tool-${String(part.toolName)}`,
+		toolCallId: String(part.toolCallId),
+		state: "input-available",
+		input: part.input,
+	} as UiMessagePart;
+}
+
+// Pure fold: append one step's content (text, reasoning, tool call/result/error) to a UI message.
+// Sources, files, and approval parts are skipped — the authoritative onFinish message carries them.
+export function applyStepToUiMessage(message: UIMessage, step: AgentStepLike): UIMessage {
+	const parts: UiMessagePart[] = [...message.parts, { type: "step-start" } as UiMessagePart];
+	const toolPartIndexByCallId = new Map<string, number>();
+
+	for (const rawContent of step.content) {
+		if (!rawContent || typeof rawContent !== "object" || typeof (rawContent as { type?: unknown }).type !== "string") {
+			continue;
+		}
+		const content = rawContent as StepContentPart;
+		if (content.type === "text" && typeof content.text === "string" && content.text) {
+			parts.push({ type: "text", text: content.text } as UiMessagePart);
+			continue;
+		}
+
+		if (content.type === "reasoning" && typeof content.text === "string" && content.text) {
+			parts.push({ type: "reasoning", text: content.text } as UiMessagePart);
+			continue;
+		}
+
+		if (content.type === "tool-call" && typeof content.toolCallId === "string") {
+			toolPartIndexByCallId.set(content.toolCallId, parts.length);
+			parts.push(toolPartFromCall(content));
+			continue;
+		}
+
+		if ((content.type === "tool-result" || content.type === "tool-error") && typeof content.toolCallId === "string") {
+			const resolution =
+				content.type === "tool-result"
+					? { state: "output-available", output: content.output }
+					: {
+							state: "output-error",
+							errorText: content.error instanceof Error ? content.error.message : String(content.error),
+						};
+
+			const index = toolPartIndexByCallId.get(content.toolCallId);
+			if (index === undefined) {
+				parts.push({ ...toolPartFromCall(content), ...resolution } as UiMessagePart);
+			} else {
+				parts[index] = { ...(parts[index] as Record<string, unknown>), ...resolution } as UiMessagePart;
+			}
+		}
+	}
+
+	return { ...message, parts };
+}
+
+type UsageDetails = Record<string, number | undefined>;
+type UsageLike = {
+	inputTokens?: number | undefined;
+	outputTokens?: number | undefined;
+	totalTokens?: number | undefined;
+	inputTokenDetails?: UsageDetails | undefined;
+	outputTokenDetails?: UsageDetails | undefined;
+};
+type MessageWithUsage = { metadata?: Record<string, unknown> & { usage?: UsageLike } };
+
+function addCounts(a: number | undefined, b: number | undefined): number | undefined {
+	if (typeof a !== "number" && typeof b !== "number") return undefined;
+	return (a ?? 0) + (b ?? 0);
+}
+
+function addDetails(a: UsageDetails | undefined, b: UsageDetails | undefined): UsageDetails | undefined {
+	if (!a && !b) return undefined;
+	const keys = new Set([...Object.keys(a ?? {}), ...Object.keys(b ?? {})]);
+	const sum: UsageDetails = {};
+	for (const key of keys) {
+		const value = addCounts(a?.[key], b?.[key]);
+		if (value !== undefined) sum[key] = value;
+	}
+	return sum;
+}
+
+// A question/approval continuation streams into the SAME assistant message; the SDK deep-merges
+// metadata but replaces primitive token counts, so the continuation's usage would silently
+// overwrite the pre-halt run's. Sum the previous usage into the final message before persisting.
+export function withAccumulatedUsageMetadata(previous: UIMessage, next: UIMessage): UIMessage {
+	const previousUsage = (previous as MessageWithUsage).metadata?.usage;
+	const nextMetadata = (next as MessageWithUsage).metadata;
+	if (!previousUsage || !nextMetadata?.usage) return next;
+
+	const usage: UsageLike = {
+		inputTokens: addCounts(previousUsage.inputTokens, nextMetadata.usage.inputTokens),
+		outputTokens: addCounts(previousUsage.outputTokens, nextMetadata.usage.outputTokens),
+		totalTokens: addCounts(previousUsage.totalTokens, nextMetadata.usage.totalTokens),
+		inputTokenDetails: addDetails(previousUsage.inputTokenDetails, nextMetadata.usage.inputTokenDetails),
+		outputTokenDetails: addDetails(previousUsage.outputTokenDetails, nextMetadata.usage.outputTokenDetails),
+	};
+
+	return { ...next, metadata: { ...nextMetadata, usage } } as UIMessage;
+}
+
+async function nextMessageSequence(threadId: string, database: AgentMessagesDb) {
+	const [row] = await database
+		.select({ maxSequence: max(schema.agentMessage.sequence) })
+		.from(schema.agentMessage)
+		.where(eq(schema.agentMessage.threadId, threadId));
+
+	return (row?.maxSequence ?? -1) + 1;
+}
+
+async function touchThread(input: { threadId: string; userId: string }, database: AgentMessagesDb) {
+	await database
+		.update(schema.agentThread)
+		.set({ lastMessageAt: new Date() })
+		.where(and(eq(schema.agentThread.id, input.threadId), eq(schema.agentThread.userId, input.userId)));
+}
+
+export async function insertDraftAssistantMessage(
+	input: { userId: string; threadId: string; uiMessageId: string },
+	database: AgentMessagesDb = db,
+) {
+	const sequence = await nextMessageSequence(input.threadId, database);
+	const [row] = await database
+		.insert(schema.agentMessage)
+		.values({
+			userId: input.userId,
+			threadId: input.threadId,
+			role: "assistant",
+			status: "streaming",
+			sequence,
+			uiMessage: { id: input.uiMessageId, role: "assistant", parts: [] },
+		})
+		.returning({ id: schema.agentMessage.id });
+
+	if (!row) throw new Error("AGENT_DRAFT_MESSAGE_CREATE_FAILED");
+
+	await touchThread(input, database);
+
+	return { rowId: row.id, sequence };
+}
+
+// Upsert by row id first, then by the uiMessage's embedded id (a question/approval continuation
+// streams into the SAME uiMessage id as the stored assistant row), then insert as a last resort.
+export async function upsertAssistantUiMessage(
+	input: {
+		userId: string;
+		threadId: string;
+		rowId?: string;
+		message: UIMessage;
+		status: "streaming" | "completed" | "canceled";
+	},
+	database: AgentMessagesDb = db,
+) {
+	const set = {
+		status: input.status,
+		uiMessage: input.message as unknown as Record<string, unknown>,
+	};
+	const isFinal = input.status !== "streaming";
+
+	if (input.rowId) {
+		const updated = await database
+			.update(schema.agentMessage)
+			.set(set)
+			.where(
+				and(
+					eq(schema.agentMessage.id, input.rowId),
+					eq(schema.agentMessage.threadId, input.threadId),
+					eq(schema.agentMessage.userId, input.userId),
+				),
+			)
+			.returning({ id: schema.agentMessage.id });
+
+		if (updated.length === 1) {
+			if (isFinal) await touchThread(input, database);
+			// biome-ignore lint/style/noNonNullAssertion: length checked above
+			return { rowId: updated[0]!.id };
+		}
+	}
+
+	const updatedById = await database
+		.update(schema.agentMessage)
+		.set(set)
+		.where(
+			and(
+				eq(schema.agentMessage.threadId, input.threadId),
+				eq(schema.agentMessage.userId, input.userId),
+				eq(schema.agentMessage.role, "assistant"),
+				sql`${schema.agentMessage.uiMessage}->>'id' = ${input.message.id}`,
+			),
+		)
+		.returning({ id: schema.agentMessage.id });
+
+	if (updatedById.length >= 1) {
+		if (isFinal) await touchThread(input, database);
+		// biome-ignore lint/style/noNonNullAssertion: length checked above
+		return { rowId: updatedById[0]!.id };
+	}
+
+	const sequence = await nextMessageSequence(input.threadId, database);
+	const [inserted] = await database
+		.insert(schema.agentMessage)
+		.values({
+			userId: input.userId,
+			threadId: input.threadId,
+			role: input.message.role,
+			status: input.status,
+			sequence,
+			uiMessage: input.message as unknown as Record<string, unknown>,
+		})
+		.returning({ id: schema.agentMessage.id });
+
+	if (!inserted) throw new Error("AGENT_MESSAGE_CREATE_FAILED");
+
+	await touchThread(input, database);
+
+	return { rowId: inserted.id };
+}
+
+// Removes a draft that never received content (sync failure before the stream produced anything).
+export async function deleteDraftIfEmpty(
+	input: { rowId: string; threadId: string; userId: string },
+	database: AgentMessagesDb = db,
+) {
+	await database
+		.delete(schema.agentMessage)
+		.where(
+			and(
+				eq(schema.agentMessage.id, input.rowId),
+				eq(schema.agentMessage.threadId, input.threadId),
+				eq(schema.agentMessage.userId, input.userId),
+				eq(schema.agentMessage.status, "streaming"),
+				sql`jsonb_array_length(${schema.agentMessage.uiMessage}->'parts') = 0`,
+			),
+		);
+}

packages/api/src/features/agent/messages.ts

@@ -43,6 +43,8 @@ export const messagesRouter = {
 		.input(
 			z.object({
 				threadId: z.string(),
+				// Deprecated and ignored: partial content now persists server-side via the run's
+				// abort path. Kept in the schema for one release so mid-deploy clients still parse.
 				partialMessage: z.custom<UIMessage>(isUiMessage, { message: "Invalid UI message." }).optional(),
 			}),
 		)
@@ -52,7 +54,6 @@ export const messagesRouter = {
 			agentService.messages.stop({
 				userId: context.user.id,
 				threadId: input.threadId,
-				...(input.partialMessage ? { partialMessage: input.partialMessage } : {}),
 			}),
 		),
 

packages/api/src/features/agent/repair.test.ts

@@ -0,0 +1,63 @@
+import { describe, expect, it } from "vitest";
+import { repairAgentPatchToolCallInput, repairAgentToolCall } from "./repair";
+
+const VALID_INPUT = JSON.stringify({
+	title: "Edit",
+	operations: [{ op: "replace", path: "/basics/name", value: "Bob" }],
+});
+
+describe("repairAgentPatchToolCallInput", () => {
+	it("repairs sloppy JSON (single quotes, trailing commas)", () => {
+		const sloppy = `{'title': 'Edit', 'operations': [{'op': 'replace', 'path': '/basics/name', 'value': 'Bob'},]}`;
+
+		const repaired = repairAgentPatchToolCallInput(sloppy);
+
+		expect(repaired).not.toBeNull();
+		expect(JSON.parse(repaired ?? "")).toMatchObject({ title: "Edit" });
+	});
+
+	it("strips /data prefixes from path and from", () => {
+		const input = JSON.stringify({
+			title: "Move",
+			operations: [{ op: "move", path: "/data/basics/name", from: "/data/basics/headline" }],
+		});
+
+		const repaired = JSON.parse(repairAgentPatchToolCallInput(input) ?? "");
+
+		expect(repaired.operations[0]).toEqual({ op: "move", path: "/basics/name", from: "/basics/headline" });
+	});
+
+	it("returns null when the input cannot be made schema-valid", () => {
+		expect(repairAgentPatchToolCallInput(`{"title": "Edit", "operations": []}`)).toBeNull();
+		expect(repairAgentPatchToolCallInput("not even close {{{")).toBeNull();
+	});
+});
+
+describe("repairAgentToolCall", () => {
+	const baseCall = {
+		type: "tool-call" as const,
+		toolCallId: "call-1",
+		toolName: "apply_resume_patch",
+	};
+
+	it("returns a repaired call for fixable apply_resume_patch input", async () => {
+		const result = await repairAgentToolCall({
+			toolCall: {
+				...baseCall,
+				input: `{'title': 'Edit', 'operations': [{'op': 'remove', 'path': '/data/basics/url'}]}`,
+			},
+		} as never);
+
+		expect(result).not.toBeNull();
+		expect(JSON.parse(result?.input ?? "")).toMatchObject({
+			operations: [{ op: "remove", path: "/basics/url" }],
+		});
+	});
+
+	it("returns null for other tools and for already-valid input", async () => {
+		expect(
+			await repairAgentToolCall({ toolCall: { ...baseCall, toolName: "read_resume", input: "{}" } } as never),
+		).toBeNull();
+		expect(await repairAgentToolCall({ toolCall: { ...baseCall, input: VALID_INPUT } } as never)).toBeNull();
+	});
+});

packages/api/src/features/agent/repair.ts

@@ -0,0 +1,56 @@
+import type { ToolCallRepairFunction, ToolSet } from "ai";
+import { jsonrepair } from "jsonrepair";
+import { applyResumePatchInputSchema } from "@reactive-resume/ai/tools/agent-tool-contracts";
+
+// Repairs sloppy apply_resume_patch calls from weaker BYOK models: fix broken JSON with
+// jsonrepair, strip the common `/data` path prefix, then re-validate against the shared schema.
+// Returning null falls back to the SDK's re-ask. Section-shortcut normalization stays in
+// normalizeAgentResumePatchOperations (it needs the resume's section ids at execute time).
+
+function stripDataPrefix(path: unknown): unknown {
+	if (typeof path !== "string") return path;
+	if (path === "/data") return "";
+	return path.startsWith("/data/") ? path.slice("/data".length) : path;
+}
+
+function normalizeRepairedInput(parsed: unknown): unknown {
+	if (!parsed || typeof parsed !== "object" || !Array.isArray((parsed as { operations?: unknown }).operations)) {
+		return parsed;
+	}
+
+	const record = parsed as Record<string, unknown> & { operations: unknown[] };
+	return {
+		...record,
+		operations: record.operations.map((operation) => {
+			if (!operation || typeof operation !== "object") return operation;
+			const op = operation as Record<string, unknown>;
+			return {
+				...op,
+				path: stripDataPrefix(op.path),
+				...("from" in op ? { from: stripDataPrefix(op.from) } : {}),
+			};
+		}),
+	};
+}
+
+// Pure core, exported for tests: returns the repaired stringified input or null.
+export function repairAgentPatchToolCallInput(rawInput: string): string | null {
+	let parsed: unknown;
+	try {
+		parsed = JSON.parse(jsonrepair(rawInput));
+	} catch {
+		return null;
+	}
+
+	const validated = applyResumePatchInputSchema.safeParse(normalizeRepairedInput(parsed));
+	return validated.success ? JSON.stringify(validated.data) : null;
+}
+
+export const repairAgentToolCall: ToolCallRepairFunction<ToolSet> = ({ toolCall }) => {
+	if (toolCall.toolName !== "apply_resume_patch") return Promise.resolve(null);
+
+	const repairedInput = repairAgentPatchToolCallInput(toolCall.input);
+	if (repairedInput === null || repairedInput === toolCall.input) return Promise.resolve(null);
+
+	return Promise.resolve({ ...toolCall, input: repairedInput });
+};

packages/api/src/features/agent/resume.test.ts

@@ -35,4 +35,18 @@ describe("normalizeAgentResumePatchOperations", () => {
 			{ op: "copy", from: "/sections/education/items/0", path: "/customSections/0/items/-" },
 		]);
 	});
+
+	it("strips the /data prefix from path and from at execution time", () => {
+		const result = normalizeAgentResumePatchOperations({ sections: { experience: {} } }, [
+			{ op: "replace", path: "/data/basics/name", value: "Bob" },
+			{ op: "move", path: "/data/basics/headline", from: "/data/basics/label" },
+			{ op: "replace", path: "/data/experience/items/0/description", value: "Combined with section shortcut" },
+		]);
+
+		expect(result).toEqual([
+			{ op: "replace", path: "/basics/name", value: "Bob" },
+			{ op: "move", path: "/basics/headline", from: "/basics/label" },
+			{ op: "replace", path: "/sections/experience/items/0/description", value: "Combined with section shortcut" },
+		]);
+	});
 });

packages/api/src/features/agent/resume.ts

@@ -29,6 +29,15 @@ function decodeJsonPointerSegment(segment: string) {
 	return segment.replace(/~1/g, "/").replace(/~0/g, "~");
 }
 
+// Models frequently prefix paths with /data because read_resume nests the document under `data`.
+// Patch paths are rooted at the document itself, and the root has no `data` key, so stripping is
+// always safe. This runs at execution time — the schema accepts such paths, so the SDK's
+// repairToolCall hook (parse/validation failures only) never sees them.
+function stripDataPrefix(path: string) {
+	if (path === "/data") return "";
+	return path.startsWith("/data/") ? path.slice("/data".length) : path;
+}
+
 function normalizeSectionShortcutPath(data: { sections: Record<string, unknown> }, path: string) {
 	if (!path.startsWith("/") || path.startsWith("/sections/")) return path;
 
@@ -43,12 +52,12 @@ export function normalizeAgentResumePatchOperations(
 	operations: JsonPatchOperation[],
 ): JsonPatchOperation[] {
 	return operations.map((operation) => {
-		const path = normalizeSectionShortcutPath(data, operation.path);
+		const path = normalizeSectionShortcutPath(data, stripDataPrefix(operation.path));
 		const normalized = path === operation.path ? operation : { ...operation, path };
 
 		if (!("from" in normalized)) return normalized;
 
-		const from = normalizeSectionShortcutPath(data, normalized.from);
+		const from = normalizeSectionShortcutPath(data, stripDataPrefix(normalized.from));
 		return from === normalized.from ? normalized : { ...normalized, from };
 	});
 }

packages/api/src/features/agent/runs.test.ts

@@ -0,0 +1,168 @@
+import type { UIMessage } from "ai";
+import { describe, expect, it, vi } from "vitest";
+
+vi.mock("@reactive-resume/db/client", () => ({ db: {} }));
+
+const { appendMissingActionParts, isStaleAgentRun, reapStaleAgentRun, STALE_AGENT_RUN_TTL_MS } = await import("./runs");
+
+const NOW = new Date("2026-08-20T12:00:00.000Z");
+
+function minutesBefore(minutes: number) {
+	return new Date(NOW.getTime() - minutes * 60_000);
+}
+
+describe("isStaleAgentRun", () => {
+	it("is false without an active run", () => {
+		expect(isStaleAgentRun({ activeRunId: null, activeRunStartedAt: null }, NOW)).toBe(false);
+	});
+
+	it("is false while the run is younger than the TTL", () => {
+		expect(isStaleAgentRun({ activeRunId: "run-1", activeRunStartedAt: minutesBefore(14) }, NOW)).toBe(false);
+	});
+
+	it("is true once the run outlives the TTL", () => {
+		expect(isStaleAgentRun({ activeRunId: "run-1", activeRunStartedAt: minutesBefore(16) }, NOW)).toBe(true);
+		expect(STALE_AGENT_RUN_TTL_MS).toBe(15 * 60_000);
+	});
+
+	it("treats a legacy claim without a start timestamp as stale", () => {
+		expect(isStaleAgentRun({ activeRunId: "run-1", activeRunStartedAt: null }, NOW)).toBe(true);
+	});
+});
+
+function buildAction(overrides: Record<string, unknown> = {}) {
+	return {
+		id: "action-1",
+		resumeId: "resume-1",
+		title: "Tighten summary",
+		summary: null,
+		operations: [{ op: "replace" as const, path: "/basics/name", value: "Bob" }],
+		appliedUpdatedAt: NOW,
+		...overrides,
+	};
+}
+
+describe("appendMissingActionParts", () => {
+	const draft: UIMessage = { id: "ui-1", role: "assistant", parts: [{ type: "text", text: "Editing…" }] };
+
+	it("appends a synthetic call/result pair for an action missing from the parts", () => {
+		const patched = appendMissingActionParts(draft, [buildAction()]);
+
+		expect(patched.parts.at(-1)).toMatchObject({
+			type: "tool-apply_resume_patch",
+			toolCallId: "synthetic-action-1",
+			state: "output-available",
+			output: expect.objectContaining({ actionId: "action-1" }),
+		});
+	});
+
+	it("returns the message unchanged when every action is already represented", () => {
+		const withPart: UIMessage = {
+			...draft,
+			parts: [
+				{
+					type: "tool-apply_resume_patch",
+					toolCallId: "call-1",
+					state: "output-available",
+					input: {},
+					output: { actionId: "action-1" },
+				} as UIMessage["parts"][number],
+			],
+		};
+
+		expect(appendMissingActionParts(withPart, [buildAction()])).toBe(withPart);
+	});
+});
+
+type ScriptedReaperDb = {
+	updates: Array<{ set: unknown }>;
+};
+
+function scriptedReaperDatabase(input: {
+	drafts: Array<Record<string, unknown>>;
+	actions: Array<Record<string, unknown>>;
+	clearMatches?: boolean;
+}) {
+	const state: ScriptedReaperDb = { updates: [] };
+	let selectCall = 0;
+
+	return {
+		state,
+		select: () => {
+			const call = selectCall;
+			selectCall += 1;
+			return {
+				from: () => ({
+					where: async () => (call === 0 ? input.drafts : input.actions),
+				}),
+			};
+		},
+		update: () => ({
+			set: (set: unknown) => {
+				state.updates.push({ set });
+				return {
+					where: () =>
+						Object.assign(Promise.resolve(undefined), {
+							returning: async () => ((input.clearMatches ?? true) ? [{ id: "thread-1" }] : []),
+						}),
+				};
+			},
+		}),
+	};
+}
+
+describe("reapStaleAgentRun", () => {
+	it("clears the run claim and flips streaming drafts to canceled with synthetic action parts", async () => {
+		const database = scriptedReaperDatabase({
+			drafts: [
+				{
+					id: "row-1",
+					uiMessage: { id: "ui-1", role: "assistant", parts: [{ type: "text", text: "Editing…" }] },
+				},
+			],
+			actions: [buildAction()],
+		});
+
+		await reapStaleAgentRun(
+			{ threadId: "thread-1", userId: "user-1", runId: "run-1", streamId: "stream-1" },
+			database as never,
+		);
+
+		// First update clears the run claim; second flips the draft.
+		expect(database.state.updates[0]?.set).toMatchObject({ activeRunId: null, activeStreamId: null });
+		expect(database.state.updates[1]?.set).toMatchObject({ status: "canceled" });
+		const uiMessage = (database.state.updates[1]?.set as { uiMessage?: UIMessage } | undefined)?.uiMessage;
+		expect(uiMessage?.parts.at(-1)).toMatchObject({ toolCallId: "synthetic-action-1" });
+	});
+
+	it("only clears the claim when there is no streaming draft", async () => {
+		const database = scriptedReaperDatabase({ drafts: [], actions: [] });
+
+		await reapStaleAgentRun(
+			{ threadId: "thread-1", userId: "user-1", runId: "run-1", streamId: null },
+			database as never,
+		);
+
+		expect(database.state.updates).toHaveLength(1);
+	});
+
+	// Regression: a concurrent request/replica can claim a replacement run (and insert a live
+	// draft) between the stale read and this reap. When the conditional clear matches nothing,
+	// the loser must not flip any draft.
+	it("does not touch drafts when another reaper already cleared or replaced the run", async () => {
+		const database = scriptedReaperDatabase({
+			drafts: [{ id: "row-live", uiMessage: { id: "ui-live", role: "assistant", parts: [] } }],
+			actions: [],
+			clearMatches: false,
+		});
+
+		await reapStaleAgentRun(
+			{ threadId: "thread-1", userId: "user-1", runId: "run-stale", streamId: "stream-stale" },
+			database as never,
+		);
+
+		// Only the (no-op) conditional clear ran; no draft status flip.
+		expect(database.state.updates).toHaveLength(1);
+		expect(database.state.updates[0]?.set).toMatchObject({ activeRunId: null });
+	});
+});

packages/api/src/features/agent/runs.ts

@@ -1,8 +1,27 @@
-import { and, eq, isNull } from "drizzle-orm";
+import type { UIMessage } from "ai";
+import { and, eq, isNotNull, isNull, sql } from "drizzle-orm";
 import { db } from "@reactive-resume/db/client";
 import * as schema from "@reactive-resume/db/schema";
 
 type AgentRunStateDb = Pick<typeof db, "update">;
+type AgentRunReaperDb = Pick<typeof db, "select" | "update">;
+
+// Deliberately TTL-only (no controller-map heuristic) so reaping stays multi-replica-safe. The TTL
+// exceeds the 10-minute run wall clock, so a live run always dies by its own timeout first.
+export const STALE_AGENT_RUN_TTL_MS = 15 * 60_000;
+
+type StaleRunThreadFields = {
+	activeRunId: string | null;
+	activeRunStartedAt: Date | null;
+};
+
+export function isStaleAgentRun(thread: StaleRunThreadFields, now = new Date()) {
+	if (!thread.activeRunId) return false;
+	// A run claim without a start timestamp is a legacy row that can never age out on its own.
+	if (!thread.activeRunStartedAt) return true;
+
+	return now.getTime() - thread.activeRunStartedAt.getTime() > STALE_AGENT_RUN_TTL_MS;
+}
 
 export async function claimActiveAgentRun(
 	input: { threadId: string; userId: string; runId: string; streamId: string },
@@ -27,7 +46,7 @@ export async function clearActiveAgentRunIfCurrent(
 	input: { threadId: string; userId: string; runId: string; streamId: string | null },
 	database: AgentRunStateDb = db,
 ) {
-	await database
+	const cleared = await database
 		.update(schema.agentThread)
 		.set({ activeRunId: null, activeStreamId: null, activeRunStartedAt: null })
 		.where(
@@ -39,5 +58,148 @@ export async function clearActiveAgentRunIfCurrent(
 					? isNull(schema.agentThread.activeStreamId)
 					: eq(schema.agentThread.activeStreamId, input.streamId),
 			),
+		)
+		.returning({ id: schema.agentThread.id });
+
+	return cleared.length === 1;
+}
+
+type ReapableActionRow = Pick<
+	typeof schema.agentAction.$inferSelect,
+	"id" | "resumeId" | "title" | "summary" | "operations" | "appliedUpdatedAt"
+>;
+
+function hasActionPart(message: UIMessage, actionId: string) {
+	return message.parts.some((part) => {
+		const output = (part as { output?: unknown }).output;
+		return (
+			part.type === "tool-apply_resume_patch" &&
+			typeof output === "object" &&
+			output !== null &&
+			(output as { actionId?: unknown }).actionId === actionId
 		);
+	});
+}
+
+// Applied actions missing from a dead draft's parts get a synthetic call/result pair so the
+// replayed history stays provider-valid (providers reject tool results without matching calls).
+export function appendMissingActionParts(message: UIMessage, actions: ReapableActionRow[]): UIMessage {
+	const missing = actions.filter((action) => !hasActionPart(message, action.id));
+	if (missing.length === 0) return message;
+
+	return {
+		...message,
+		parts: [
+			...message.parts,
+			...missing.map(
+				(action) =>
+					({
+						type: "tool-apply_resume_patch",
+						toolCallId: `synthetic-${action.id}`,
+						state: "output-available",
+						input: {
+							title: action.title,
+							...(action.summary ? { summary: action.summary } : {}),
+							operations: action.operations,
+						},
+						output: {
+							actionId: action.id,
+							resumeId: action.resumeId,
+							title: action.title,
+							summary: action.summary,
+							operations: action.operations,
+							appliedUpdatedAt: action.appliedUpdatedAt.toISOString(),
+							// Snapshot boundary: the `resume` key makes context pruning supersede older
+							// full snapshots, and the null value forces a fresh read_resume — the edit
+							// committed, but the post-patch document was lost with the crashed run.
+							resume: null,
+							note: "This edit was applied, but the run was interrupted before the updated resume could be recorded. Re-read the resume before making further edits.",
+						},
+					}) as UIMessage["parts"][number],
+			),
+		],
+	};
+}
+
+// Reap = conditionally clear the run claim, then flip dead "streaming" drafts to canceled with
+// their committed actions represented. Applied actions stay applied — they are real, individually
+// revertable edits; auto-reverting on reap would be worse than the crash.
+export async function reapStaleAgentRun(
+	input: { threadId: string; userId: string; runId: string; streamId: string | null },
+	database: AgentRunReaperDb = db,
+) {
+	// Snapshot drafts BEFORE clearing the claim: while the stale claim still holds, no new run can
+	// start, so every "streaming" draft visible here belongs to the dead run — a draft inserted by
+	// a replacement run claimed after this point is never in the snapshot.
+	const drafts = await database
+		.select()
+		.from(schema.agentMessage)
+		.where(
+			and(
+				eq(schema.agentMessage.threadId, input.threadId),
+				eq(schema.agentMessage.userId, input.userId),
+				eq(schema.agentMessage.status, "streaming"),
+			),
+		);
+
+	// Concurrent reapers (another request or replica) race on this conditional clear; the loser
+	// must not touch drafts — the thread has already moved on under a different run.
+	const cleared = await clearActiveAgentRunIfCurrent(input, database);
+	if (!cleared) return;
+
+	for (const draft of drafts) {
+		const actions = await database
+			.select({
+				id: schema.agentAction.id,
+				resumeId: schema.agentAction.resumeId,
+				title: schema.agentAction.title,
+				summary: schema.agentAction.summary,
+				operations: schema.agentAction.operations,
+				appliedUpdatedAt: schema.agentAction.appliedUpdatedAt,
+			})
+			.from(schema.agentAction)
+			.where(
+				and(
+					eq(schema.agentAction.messageId, draft.id),
+					eq(schema.agentAction.kind, "resume_patch"),
+					eq(schema.agentAction.status, "applied"),
+				),
+			);
+
+		const message = appendMissingActionParts(draft.uiMessage as unknown as UIMessage, actions);
+		await database
+			.update(schema.agentMessage)
+			.set({ status: "canceled", uiMessage: message as unknown as Record<string, unknown> })
+			.where(
+				and(
+					eq(schema.agentMessage.id, draft.id),
+					// A continuation can claim the thread right after our conditional clear and start
+					// writing into this very row. Flip only the exact state we snapshotted — any
+					// concurrent write changes status or uiMessage and makes this a no-op.
+					eq(schema.agentMessage.status, "streaming"),
+					sql`${schema.agentMessage.uiMessage} = ${JSON.stringify(draft.uiMessage)}::jsonb`,
+				),
+			);
+	}
+}
+
+export async function reapStaleAgentRunsAtBoot(database: AgentRunReaperDb = db) {
+	const threads = await database
+		.select({
+			id: schema.agentThread.id,
+			userId: schema.agentThread.userId,
+			activeRunId: schema.agentThread.activeRunId,
+			activeStreamId: schema.agentThread.activeStreamId,
+			activeRunStartedAt: schema.agentThread.activeRunStartedAt,
+		})
+		.from(schema.agentThread)
+		.where(isNotNull(schema.agentThread.activeRunId));
+
+	for (const thread of threads) {
+		if (!isStaleAgentRun(thread) || !thread.activeRunId) continue;
+		await reapStaleAgentRun(
+			{ threadId: thread.id, userId: thread.userId, runId: thread.activeRunId, streamId: thread.activeStreamId },
+			database,
+		);
+	}
 }

packages/api/src/features/agent/service.test.ts

@@ -11,6 +11,13 @@ const dbMock = {
 
 const clearActiveAgentRunIfCurrentMock = vi.fn();
 const claimActiveAgentRunMock = vi.fn();
+const messagesPersistenceMock = {
+	applyStepToUiMessage: vi.fn((message: unknown) => message),
+	insertDraftAssistantMessage: vi.fn(),
+	upsertAssistantUiMessage: vi.fn(),
+	deleteDraftIfEmpty: vi.fn(),
+	withAccumulatedUsageMetadata: vi.fn((_previous: unknown, next: unknown) => next),
+};
 const storageServiceMock = {
 	delete: vi.fn(),
 	write: vi.fn(),
@@ -38,6 +45,7 @@ vi.mock("@reactive-resume/db/schema", () => ({
 		deletedAt: "agent_threads.deleted_at",
 		archivedAt: "agent_threads.archived_at",
 		status: "agent_threads.status",
+		reviewPatches: "agent_threads.review_patches",
 		activeRunId: "agent_threads.active_run_id",
 		activeStreamId: "agent_threads.active_stream_id",
 		activeRunStartedAt: "agent_threads.active_run_started_at",
@@ -96,14 +104,29 @@ vi.mock("drizzle-orm", () => ({
 	sql: () => ({ type: "sql" }),
 }));
 
-vi.mock("ai", () => ({
+// Spread-actual: pure helpers (isStepCount, safeValidateUIMessages, pruneMessages, ...) stay real;
+// only the scripted seams are mocked.
+vi.mock("ai", async (importOriginal) => ({
+	...(await importOriginal<typeof import("ai")>()),
 	convertToModelMessages: vi.fn(),
-	stepCountIs: vi.fn(),
 	ToolLoopAgent: vi.fn(),
 }));
 
-vi.mock("../ai/service", () => ({ getAgentModel: vi.fn() }));
-vi.mock("../ai/credentials", () => ({ assertAgentEnvironment: vi.fn() }));
+// Minimal V4 model stub: the real wrapLanguageModel/addToolInputExamplesMiddleware run against it.
+vi.mock("../ai/service", () => ({
+	getAgentModel: vi.fn(() => ({
+		specificationVersion: "v4",
+		provider: "mock",
+		modelId: "mock-model",
+		supportedUrls: {},
+		doGenerate: vi.fn(),
+		doStream: vi.fn(),
+	})),
+}));
+vi.mock("../ai/credentials", () => ({
+	assertAgentEnvironment: vi.fn(),
+	getAgentToolApprovalSecret: vi.fn(() => "test-approval-secret"),
+}));
 vi.mock("../ai-providers/service", () => ({ aiProvidersService: aiProvidersServiceMock }));
 vi.mock("../resume/service", () => ({ resumeService: resumeServiceMock }));
 vi.mock("../storage/service", () => ({
@@ -118,7 +141,10 @@ vi.mock("./resume", () => ({
 vi.mock("./runs", () => ({
 	claimActiveAgentRun: claimActiveAgentRunMock,
 	clearActiveAgentRunIfCurrent: clearActiveAgentRunIfCurrentMock,
+	isStaleAgentRun: vi.fn(() => false),
+	reapStaleAgentRun: vi.fn(),
 }));
+vi.mock("./messages-persistence", () => messagesPersistenceMock);
 vi.mock("./streams", () => ({
 	agentStreamLifecycle: { create: vi.fn(), resume: vi.fn() },
 }));
@@ -135,6 +161,12 @@ beforeEach(() => {
 	dbMock.transaction.mockImplementation(async <T>(callback: (tx: typeof dbMock) => Promise<T>) => callback(dbMock));
 	clearActiveAgentRunIfCurrentMock.mockReset();
 	claimActiveAgentRunMock.mockReset();
+	for (const mock of Object.values(messagesPersistenceMock)) mock.mockReset();
+	messagesPersistenceMock.applyStepToUiMessage.mockImplementation((message: unknown) => message);
+	messagesPersistenceMock.insertDraftAssistantMessage.mockResolvedValue({ rowId: "draft-row-1", sequence: 1 });
+	messagesPersistenceMock.upsertAssistantUiMessage.mockResolvedValue({ rowId: "draft-row-1" });
+	messagesPersistenceMock.deleteDraftIfEmpty.mockResolvedValue(undefined);
+	messagesPersistenceMock.withAccumulatedUsageMetadata.mockImplementation((_previous: unknown, next: unknown) => next);
 	for (const mock of Object.values(storageServiceMock)) mock.mockReset();
 	for (const mock of Object.values(resumeServiceMock)) mock.mockReset();
 	for (const mock of Object.values(aiProvidersServiceMock)) mock.mockReset();
@@ -149,6 +181,7 @@ function buildArchivedThread(overrides: Record<string, unknown> = {}) {
 		sourceResumeId: null,
 		title: "Archived thread",
 		status: "archived",
+		reviewPatches: false,
 		activeRunId: null,
 		activeStreamId: null,
 		activeRunStartedAt: null,
@@ -398,7 +431,7 @@ describe("agentService.messages.send", () => {
 		]);
 	});
 
-	it("stores snapshotData and applies a valid JSON Patch without a timestamp conflict guard", async () => {
+	it("stores snapshotData and applies a valid JSON Patch guarded by the pre-read timestamp", async () => {
 		const activeThread = buildActiveThread();
 		const persistedMessage = {
 			id: "message-1",
@@ -472,8 +505,13 @@ describe("agentService.messages.send", () => {
 		];
 		const insertValues: unknown[] = [];
 
+		const patchedData = { basics: { customFields: [{ id: "field-1" }] } };
 		resumeServiceMock.getById.mockResolvedValue({ data: beforeData, updatedAt: beforeUpdatedAt });
-		resumeServiceMock.patchInTransaction.mockResolvedValue({ id: "resume-1", updatedAt: patchedUpdatedAt });
+		resumeServiceMock.patchInTransaction.mockResolvedValue({
+			id: "resume-1",
+			updatedAt: patchedUpdatedAt,
+			data: patchedData,
+		});
 		dbMock.insert.mockReturnValue({
 			values: vi.fn((value) => {
 				insertValues.push(value);
@@ -501,6 +539,7 @@ describe("agentService.messages.send", () => {
 			id: "resume-1",
 			userId: "user-1",
 			operations,
+			expectedUpdatedAt: beforeUpdatedAt,
 		});
 		expect(insertValues).toContainEqual(
 			expect.objectContaining({
@@ -510,7 +549,83 @@ describe("agentService.messages.send", () => {
 				appliedUpdatedAt: patchedUpdatedAt,
 			}),
 		);
-		expect(result).toEqual(expect.objectContaining({ actionId: "action-1", resumeId: "resume-1" }));
+		expect(result).toEqual(
+			expect.objectContaining({
+				actionId: "action-1",
+				resumeId: "resume-1",
+				changedPaths: ["/basics/customFields/-"],
+				resume: patchedData,
+			}),
+		);
+	});
+
+	it("rethrows a resume version conflict as a recoverable plain tool error", async () => {
+		const activeThread = buildActiveThread();
+		const persistedMessage = {
+			id: "message-1",
+			userId: "user-1",
+			threadId: "thread-1",
+			role: "user",
+			status: "completed",
+			sequence: 0,
+			uiMessage: { id: "ui-message-1", role: "user", parts: [{ type: "text", text: "Edit" }] },
+		};
+
+		dbMock.select
+			.mockImplementationOnce(() => selectLimitResult([activeThread]))
+			.mockImplementationOnce(() => selectWhereResult([{ maxSequence: -1 }]))
+			.mockImplementationOnce(() => selectWhereResult([{ total: 1 }]))
+			.mockImplementationOnce(() => selectOrderByResult([persistedMessage]));
+		dbMock.insert.mockReturnValue({
+			values: vi.fn(() => ({ returning: vi.fn(async () => [persistedMessage]) })),
+		});
+		dbMock.update.mockReturnValue({ set: vi.fn(() => ({ where: vi.fn(async () => undefined) })) });
+
+		claimActiveAgentRunMock.mockResolvedValue(true);
+		aiProvidersServiceMock.getRunnableById.mockResolvedValue({
+			id: "provider-1",
+			provider: "openai",
+			model: "gpt-5",
+			apiKey: "secret",
+			baseURL: null,
+		});
+		aiProvidersServiceMock.markUsed.mockResolvedValue(undefined);
+
+		const [
+			{ convertToModelMessages, ToolLoopAgent },
+			{ agentStreamLifecycle },
+			{ buildAgentTools },
+			{ streamToEventIterator },
+		] = await Promise.all([import("ai"), import("./streams"), import("./tools"), import("@orpc/server")]);
+		vi.mocked(convertToModelMessages).mockResolvedValue([{ role: "user", content: [{ type: "text", text: "Edit" }] }]);
+		class MockToolLoopAgent {
+			stream = vi.fn(async () => ({ toUIMessageStream: vi.fn(() => new ReadableStream()) }));
+		}
+		vi.mocked(ToolLoopAgent).mockImplementation(MockToolLoopAgent as never);
+		vi.mocked(agentStreamLifecycle.create).mockResolvedValue(new ReadableStream());
+		vi.mocked(streamToEventIterator).mockReturnValue("iterator" as never);
+
+		const { agentService } = await import("./service");
+
+		await agentService.messages.send({
+			threadId: "thread-1",
+			userId: "user-1",
+			// biome-ignore lint/suspicious/noExplicitAny: minimal fixture for unit test
+			message: { id: "ui-message-1", role: "user", parts: [{ type: "text", text: "Edit" }] } as any,
+		});
+
+		resumeServiceMock.getById.mockResolvedValue({ data: {}, updatedAt: new Date("2026-05-01T00:00:00.000Z") });
+		resumeServiceMock.patchInTransaction.mockRejectedValue(new ORPCError("RESUME_VERSION_CONFLICT"));
+
+		const toolConfig = vi.mocked(buildAgentTools).mock.calls.at(-1)?.[0];
+		// biome-ignore lint/suspicious/noExplicitAny: captured mocked tool config has intentionally loose handler types
+		const applying = (toolConfig as any).handlers.applyResumePatch({
+			title: "Edit",
+			operations: [{ op: "replace", path: "/basics/name", value: "Bob" }],
+		});
+
+		await expect(applying).rejects.toThrowError("The resume changed while this edit was being prepared");
+		await expect(applying).rejects.not.toBeInstanceOf(ORPCError);
 	});
 
 	it("persists canonical attachment UI parts, links selected attachments, and appends server-read model parts", async () => {
@@ -784,6 +899,10 @@ describe("agentService.messages.send", () => {
 						type: "tool-ask_user_question",
 						toolCallId: "call-1",
 						state: "output-available",
+						input: {
+							question: "How broadly should I rename?",
+							choices: ["Only change the main resume header name"],
+						},
 						output: "Only change the main resume header name",
 					},
 				],
@@ -809,6 +928,122 @@ describe("agentService.messages.send", () => {
 		expect(convertToModelMessages).toHaveBeenCalledWith([userMessage.uiMessage, answeredAssistantModelInput]);
 	});
 
+	// Regression (defect 8): a question continuation streams into the SAME uiMessage id; onFinish
+	// must upsert the existing assistant row instead of inserting a duplicate row.
+	it("continues the existing assistant row on a question continuation instead of inserting a duplicate", async () => {
+		const activeThread = buildActiveThread();
+		const userMessage = {
+			id: "message-user-1",
+			userId: "user-1",
+			threadId: "thread-1",
+			role: "user",
+			status: "completed",
+			sequence: 0,
+			uiMessage: { id: "ui-user-1", role: "user", parts: [{ type: "text", text: "Change the name" }] },
+		};
+		const question = { question: "How broadly should I rename?", choices: ["Only the header"] };
+		const unansweredAssistantMessage = {
+			id: "message-assistant-1",
+			userId: "user-1",
+			threadId: "thread-1",
+			role: "assistant",
+			status: "completed",
+			sequence: 1,
+			uiMessage: {
+				id: "ui-assistant-1",
+				role: "assistant",
+				parts: [{ type: "tool-ask_user_question", toolCallId: "call-1", state: "input-available", input: question }],
+			},
+		};
+		const answeredAssistantMessage = {
+			...unansweredAssistantMessage,
+			uiMessage: {
+				...unansweredAssistantMessage.uiMessage,
+				parts: [
+					{
+						type: "tool-ask_user_question",
+						toolCallId: "call-1",
+						state: "output-available",
+						input: question,
+						output: "Only the header",
+					},
+				],
+			},
+		};
+
+		dbMock.select
+			.mockImplementationOnce(() => selectLimitResult([activeThread]))
+			.mockImplementationOnce(() => selectOrderByResult([userMessage, unansweredAssistantMessage]))
+			.mockImplementationOnce(() => selectOrderByResult([userMessage, answeredAssistantMessage]));
+		dbMock.update.mockImplementation(() => ({ set: vi.fn(() => ({ where: vi.fn(async () => undefined) })) }));
+
+		claimActiveAgentRunMock.mockResolvedValue(true);
+		aiProvidersServiceMock.getRunnableById.mockResolvedValue({
+			id: "provider-1",
+			provider: "openai",
+			model: "gpt-5",
+			apiKey: "secret",
+			baseURL: null,
+		});
+		aiProvidersServiceMock.markUsed.mockResolvedValue(undefined);
+
+		const [{ convertToModelMessages, ToolLoopAgent }, { agentStreamLifecycle }, { streamToEventIterator }] =
+			await Promise.all([import("ai"), import("./streams"), import("@orpc/server")]);
+		vi.mocked(convertToModelMessages).mockResolvedValue([
+			{ role: "user", content: [{ type: "text", text: "Change the name" }] },
+		]);
+
+		let uiStreamOptions: Record<string, unknown> | undefined;
+		class MockToolLoopAgent {
+			stream = vi.fn(async () => ({
+				toUIMessageStream: vi.fn((options: Record<string, unknown>) => {
+					uiStreamOptions = options;
+					return new ReadableStream();
+				}),
+			}));
+		}
+		vi.mocked(ToolLoopAgent).mockImplementation(MockToolLoopAgent as never);
+		vi.mocked(agentStreamLifecycle.create).mockImplementation((_streamId, makeStream) => {
+			(makeStream as () => unknown)();
+			return Promise.resolve(new ReadableStream());
+		});
+		vi.mocked(streamToEventIterator).mockReturnValue("iterator" as never);
+
+		const { agentService } = await import("./service");
+
+		await agentService.messages.send({
+			threadId: "thread-1",
+			userId: "user-1",
+			// biome-ignore lint/suspicious/noExplicitAny: minimal fixture for unit test
+			message: answeredAssistantMessage.uiMessage as any,
+		});
+
+		const onFinish = uiStreamOptions?.onFinish as (event: Record<string, unknown>) => Promise<void>;
+		const continuedMessage = {
+			...answeredAssistantMessage.uiMessage,
+			parts: [...answeredAssistantMessage.uiMessage.parts, { type: "text", text: "Renamed the header." }],
+		};
+		await onFinish({ responseMessage: continuedMessage, isAborted: false, isContinuation: true, messages: [] });
+
+		expect(messagesPersistenceMock.insertDraftAssistantMessage).not.toHaveBeenCalled();
+		expect(dbMock.insert).not.toHaveBeenCalled();
+		expect(messagesPersistenceMock.upsertAssistantUiMessage).toHaveBeenCalledWith(
+			expect.objectContaining({
+				rowId: "message-assistant-1",
+				status: "completed",
+				message: expect.objectContaining({ id: "ui-assistant-1" }),
+			}),
+		);
+
+		// Usage metadata is attached on the finish part only and carries the provider's model id.
+		const messageMetadata = uiStreamOptions?.messageMetadata as (options: { part: Record<string, unknown> }) => unknown;
+		expect(messageMetadata({ part: { type: "finish", totalUsage: { totalTokens: 42 } } })).toEqual({
+			usage: { totalTokens: 42 },
+			model: "gpt-5",
+		});
+		expect(messageMetadata({ part: { type: "text-delta" } })).toBeUndefined();
+	});
+
 	it("repairs legacy user-answer messages that followed an unresolved ask-user-question tool call", async () => {
 		const activeThread = buildActiveThread();
 		const firstUserMessage = {
@@ -952,6 +1187,30 @@ describe("agentService.messages.send", () => {
 		]);
 	});
 
+	it("rejects malformed UI message parts before claiming a run", async () => {
+		dbMock.select.mockImplementationOnce(() => selectLimitResult([buildActiveThread()]));
+		aiProvidersServiceMock.getRunnableById.mockResolvedValue({
+			id: "provider-1",
+			provider: "openai",
+			model: "gpt-5",
+			apiKey: "secret",
+			baseURL: null,
+		});
+
+		const { agentService } = await import("./service");
+
+		const sending = agentService.messages.send({
+			threadId: "thread-1",
+			userId: "user-1",
+			// biome-ignore lint/suspicious/noExplicitAny: malformed fixture on purpose
+			message: { id: "ui-message-1", role: "user", parts: [{ type: "text" }] } as any,
+		});
+
+		await expect(sending).rejects.toMatchObject({ code: "BAD_REQUEST", message: "Invalid UI message parts." });
+		expect(claimActiveAgentRunMock).not.toHaveBeenCalled();
+		expect(dbMock.insert).not.toHaveBeenCalled();
+	});
+
 	it("rejects malformed attachment IDs before persisting a message", async () => {
 		dbMock.select.mockImplementationOnce(() => selectLimitResult([buildActiveThread()]));
 		aiProvidersServiceMock.getRunnableById.mockResolvedValue({

packages/api/src/features/agent/service.ts

@@ -1,26 +1,49 @@
+import type { ApplyResumePatchInput } from "@reactive-resume/ai/tools/agent-tool-contracts";
 import type { JsonPatchOperation } from "@reactive-resume/resume/patch";
 import type { Locale } from "@reactive-resume/utils/locale";
 import type { FilePart, ImagePart, ModelMessage, TextPart, UIMessage } from "ai";
 import type { getModel } from "../ai/service";
 import { ORPCError } from "@orpc/client";
 import { streamToEventIterator } from "@orpc/server";
-import { convertToModelMessages, stepCountIs, ToolLoopAgent } from "ai";
+import {
+	addToolInputExamplesMiddleware,
+	convertToModelMessages,
+	isStepCount,
+	safeValidateUIMessages,
+	smoothStream,
+	ToolLoopAgent,
+	wrapLanguageModel,
+} from "ai";
 import { and, asc, count, desc, eq, gte, inArray, isNull, max, sql } from "drizzle-orm";
 import { db } from "@reactive-resume/db/client";
 import * as schema from "@reactive-resume/db/schema";
 import { defaultResumeData } from "@reactive-resume/schema/resume/default";
 import { generateId } from "@reactive-resume/utils/string";
-import { assertAgentEnvironment } from "../ai/credentials";
+import { assertAgentEnvironment, getAgentToolApprovalSecret } from "../ai/credentials";
 import { getAgentModel } from "../ai/service";
 import { aiProvidersService } from "../ai-providers/service";
 import { resumeService } from "../resume/service";
 import { getStorageService, inferContentType } from "../storage/service";
+import { pruneAgentModelContext } from "./context";
+import { mergeClientToolResponses } from "./messages-merge";
+import {
+	applyStepToUiMessage,
+	deleteDraftIfEmpty,
+	insertDraftAssistantMessage,
+	upsertAssistantUiMessage,
+	withAccumulatedUsageMetadata,
+} from "./messages-persistence";
+import { repairAgentToolCall } from "./repair";
 import { buildAgentDraftResumeName, buildUniqueAgentDraftSlug, normalizeAgentResumePatchOperations } from "./resume";
-import { claimActiveAgentRun, clearActiveAgentRunIfCurrent } from "./runs";
+import { claimActiveAgentRun, clearActiveAgentRunIfCurrent, isStaleAgentRun, reapStaleAgentRun } from "./runs";
 import { agentStreamLifecycle } from "./streams";
 import { buildAgentInstructions, buildAgentTools } from "./tools";
 
 const MAX_AGENT_STEPS = 30;
+const MAX_AGENT_OUTPUT_TOKENS = 8_192;
+const MAX_AGENT_MODEL_RETRIES = 2;
+const AGENT_STEP_TIMEOUT_MS = 120_000;
+const AGENT_RUN_TIMEOUT_MS = 600_000;
 const MAX_ATTACHMENTS_PER_MESSAGE = 10;
 const MAX_ATTACHMENT_BYTES = 25 * 1024 * 1024;
 const MAX_THREAD_ATTACHMENT_BYTES = 100 * 1024 * 1024;
@@ -39,7 +62,7 @@ const ROLLBACK_CONFLICT_MESSAGE = "The resume changed after this action was appl
 const ROLLED_BACK_MESSAGE = "This patch was rolled back when the resume was restored to an earlier state.";
 
 const activeRunControllers = new Map<string, AbortController>();
-const canceledRunsWithPersistedPartial = new Set<string>();
+const activeRunTimeouts = new Map<string, ReturnType<typeof setTimeout>>();
 
 // Abort reasons MUST be an AbortError: the AI SDK only treats `err.name === "AbortError"`
 // (via isAbortError) as a cancellation. A bare-string reason is treated as a genuine stream
@@ -88,6 +111,7 @@ function toThreadSummary(row: AgentThreadRecord & { resumeName?: string | null;
 		id: row.id,
 		title: row.title,
 		status: row.status,
+		reviewPatches: row.reviewPatches,
 		sourceResumeId: row.sourceResumeId,
 		workingResumeId: row.workingResumeId,
 		aiProviderId: row.aiProviderId,
@@ -193,63 +217,6 @@ type AgentToolPart = UIMessage["parts"][number] & {
 	toolCallId?: string;
 };
 
-type AnsweredAskUserQuestionPart = AgentToolPart & {
-	toolCallId: string;
-};
-
-function isAnsweredAskUserQuestionPart(part: UIMessage["parts"][number]): part is AnsweredAskUserQuestionPart {
-	const toolPart = part as AgentToolPart;
-	return (
-		toolPart.type === "tool-ask_user_question" &&
-		typeof toolPart.toolCallId === "string" &&
-		(toolPart.state === "output-available" || toolPart.state === "output-error")
-	);
-}
-
-function mergeAskUserQuestionOutputs(existingMessage: UIMessage, incomingMessage: UIMessage): UIMessage {
-	const answeredParts = new Map<string, AgentToolPart>();
-
-	for (const part of incomingMessage.parts) {
-		if (isAnsweredAskUserQuestionPart(part)) answeredParts.set(part.toolCallId, part);
-	}
-
-	let didMerge = false;
-	const parts = existingMessage.parts.map((part) => {
-		const existingPart = part as AgentToolPart;
-		if (
-			existingPart.type !== "tool-ask_user_question" ||
-			typeof existingPart.toolCallId !== "string" ||
-			existingPart.state !== "input-available"
-		) {
-			return part;
-		}
-
-		const answeredPart = answeredParts.get(existingPart.toolCallId);
-		if (!answeredPart) return part;
-
-		didMerge = true;
-		if (answeredPart.state === "output-error") {
-			return {
-				...part,
-				state: "output-error",
-				errorText: answeredPart.errorText ?? "User answer failed.",
-			} as UIMessage["parts"][number];
-		}
-
-		return {
-			...part,
-			state: "output-available",
-			output: answeredPart.output,
-		} as UIMessage["parts"][number];
-	});
-
-	if (!didMerge) {
-		throw new ORPCError("BAD_REQUEST", { message: "No matching unanswered user question was found." });
-	}
-
-	return { ...existingMessage, parts };
-}
-
 function getFirstUnansweredAskUserQuestionToolCallId(message: UIMessage) {
 	const part = message.parts.find((part) => {
 		const toolPart = part as AgentToolPart;
@@ -264,7 +231,7 @@ function getFirstUnansweredAskUserQuestionToolCallId(message: UIMessage) {
 }
 
 function answerAskUserQuestionToolCall(message: UIMessage, toolCallId: string, answer: string) {
-	return mergeAskUserQuestionOutputs(message, {
+	const { message: merged } = mergeClientToolResponses(message, {
 		...message,
 		parts: [
 			{
@@ -276,6 +243,8 @@ function answerAskUserQuestionToolCall(message: UIMessage, toolCallId: string, a
 			} as UIMessage["parts"][number],
 		],
 	});
+
+	return merged;
 }
 
 function attachmentLabel(attachment: AgentAttachmentRecord) {
@@ -534,7 +503,25 @@ async function updateAssistantToolResultMessage(input: { userId: string; threadI
 		throw new ORPCError("BAD_REQUEST", { message: "The answered assistant message was not found." });
 	}
 
-	const mergedMessage = mergeAskUserQuestionOutputs(toMessage(existingRow), input.message);
+	const {
+		message: mergedMessage,
+		mergedCount,
+		alreadyResolvedCount,
+		pendingContinuationCount,
+		conflictingCount,
+	} = mergeClientToolResponses(toMessage(existingRow), input.message);
+
+	if (conflictingCount > 0) {
+		throw new ORPCError("BAD_REQUEST", { message: "This approval was already answered with a different decision." });
+	}
+	// A recorded-but-unexecuted approval (pendingContinuationCount) proceeds: a prior continuation
+	// attempt failed after persisting the decision, and this retry is the recovery path.
+	if (mergedCount === 0 && pendingContinuationCount === 0) {
+		if (alreadyResolvedCount > 0) {
+			throw new ORPCError("CONFLICT", { message: "This response was already handled." });
+		}
+		throw new ORPCError("BAD_REQUEST", { message: "No matching unanswered user question was found." });
+	}
 
 	await db
 		.update(schema.agentMessage)
@@ -555,7 +542,7 @@ async function updateAssistantToolResultMessage(input: { userId: string; threadI
 		.set({ lastMessageAt: new Date() })
 		.where(and(eq(schema.agentThread.id, input.threadId), eq(schema.agentThread.userId, input.userId)));
 
-	return mergedMessage;
+	return { message: mergedMessage, rowId: existingRow.id };
 }
 
 async function repairLegacyAskUserQuestionAnswers(
@@ -610,9 +597,16 @@ async function cleanupActiveRun(input: {
 	runId: string;
 	streamId: string;
 	primaryError?: unknown;
+	// When final persistence failed, keep the run claim: the reaper only examines threads with an
+	// active claim, so releasing it here would orphan the "streaming" draft forever. The TTL reap
+	// heals the claim and the draft together.
+	preserveClaimForReaper?: boolean;
 }) {
 	activeRunControllers.delete(input.runId);
-	canceledRunsWithPersistedPartial.delete(input.runId);
+	clearTimeout(activeRunTimeouts.get(input.runId));
+	activeRunTimeouts.delete(input.runId);
+
+	if (input.preserveClaimForReaper) return;
 
 	try {
 		await clearActiveAgentRunIfCurrent(input);
@@ -680,7 +674,7 @@ async function readAttachment(input: { id: string; threadId: string; userId: str
 		filename: attachment.filename,
 		mediaType: attachment.mediaType,
 		size: attachment.size,
-		content: new TextDecoder().decode(stored.data).slice(0, 40_000),
+		content: new TextDecoder().decode(stored.data).slice(0, MAX_ATTACHMENT_TEXT_CHARS),
 	};
 }
 
@@ -688,42 +682,74 @@ async function applyResumePatch(input: {
 	userId: string;
 	threadId: string;
 	resumeId: string;
+	messageId?: string;
 	title: string;
 	summary?: string;
+	baseUpdatedAt?: string;
 	operations: JsonPatchOperation[];
 }) {
 	const before = await resumeService.getById({ id: input.resumeId, userId: input.userId });
+
+	// Bind the patch to the revision the model actually read (and, under review, the revision the
+	// user approved): index-based operations built against an older document could otherwise
+	// silently target different items after a concurrent edit. baseUpdatedAt travels inside the
+	// signed tool input, so an approval cannot be replayed against a changed resume either.
+	if (input.baseUpdatedAt) {
+		const baseTime = new Date(input.baseUpdatedAt).getTime();
+		// An unparseable value must fail loudly rather than silently skip the revision check.
+		if (Number.isNaN(baseTime)) {
+			throw new Error(
+				`baseUpdatedAt is not a valid timestamp. Pass the updatedAt from the read_resume or apply_resume_patch result verbatim (currently ${before.updatedAt.toISOString()}).`,
+			);
+		}
+		if (baseTime !== before.updatedAt.getTime()) {
+			throw new Error(
+				`The resume changed after it was read (its updatedAt is now ${before.updatedAt.toISOString()}). Re-read the resume and rebuild the patch against the current document.`,
+			);
+		}
+	}
+
 	const snapshotData = cloneResumeData(before.data);
 	const operations = normalizeAgentResumePatchOperations(before.data, input.operations);
 
-	const { action, patched } = await db.transaction(async (tx) => {
-		const patched = await resumeService.patchInTransaction(tx, {
-			id: input.resumeId,
-			userId: input.userId,
-			operations,
-		});
-
-		const [action] = await tx
-			.insert(schema.agentAction)
-			.values({
+	const { action, patched } = await db
+		.transaction(async (tx) => {
+			const patched = await resumeService.patchInTransaction(tx, {
+				id: input.resumeId,
 				userId: input.userId,
-				threadId: input.threadId,
-				resumeId: input.resumeId,
-				kind: "resume_patch",
-				status: "applied",
-				title: input.title,
-				...(input.summary !== undefined ? { summary: input.summary } : {}),
 				operations,
-				snapshotData,
-				baseUpdatedAt: before.updatedAt,
-				appliedUpdatedAt: patched.updatedAt,
-			})
-			.returning();
+				expectedUpdatedAt: before.updatedAt,
+			});
 
-		if (!action) throw new Error("AGENT_ACTION_CREATE_FAILED");
+			const [action] = await tx
+				.insert(schema.agentAction)
+				.values({
+					userId: input.userId,
+					threadId: input.threadId,
+					resumeId: input.resumeId,
+					...(input.messageId ? { messageId: input.messageId } : {}),
+					kind: "resume_patch",
+					status: "applied",
+					title: input.title,
+					...(input.summary !== undefined ? { summary: input.summary } : {}),
+					operations,
+					snapshotData,
+					baseUpdatedAt: before.updatedAt,
+					appliedUpdatedAt: patched.updatedAt,
+				})
+				.returning();
 
-		return { action, patched };
-	});
+			if (!action) throw new Error("AGENT_ACTION_CREATE_FAILED");
+
+			return { action, patched };
+		})
+		.catch((error: unknown) => {
+			// Surface the version conflict as a recoverable tool error, not a run-fatal ORPCError.
+			if (error instanceof ORPCError && error.code === "RESUME_VERSION_CONFLICT") {
+				throw new Error("The resume changed while this edit was being prepared. Re-read the resume and retry.");
+			}
+			throw error;
+		});
 
 	await resumeService.notifyResumePatched({
 		resumeId: patched.id,
@@ -738,6 +764,10 @@ async function applyResumePatch(input: {
 		summary: action.summary,
 		operations: action.operations,
 		appliedUpdatedAt: action.appliedUpdatedAt.toISOString(),
+		changedPaths: [...new Set(operations.flatMap((op) => ("from" in op ? [op.path, op.from] : [op.path])))],
+		// Full post-patch document: array indexes may have shifted, so the model must base
+		// further patches on this instead of an earlier read_resume snapshot.
+		resume: patched.data,
 	};
 }
 
@@ -745,6 +775,8 @@ function createAgent(input: {
 	userId: string;
 	threadId: string;
 	resumeId: string;
+	draftRowId?: string;
+	requirePatchApproval?: boolean;
 	provider: {
 		provider: Parameters<typeof getModel>[0]["provider"];
 		model: string;
@@ -753,10 +785,35 @@ function createAgent(input: {
 	};
 	model: ReturnType<typeof getModel>;
 }) {
+	// One greppable JSON line per tool execution.
+	const timedToolHandler =
+		<A extends unknown[], R>(toolName: string, run: (...args: A) => Promise<R>) =>
+		async (...args: A): Promise<R> => {
+			const startedAt = Date.now();
+			let ok = true;
+			try {
+				return await run(...args);
+			} catch (error) {
+				ok = false;
+				throw error;
+			} finally {
+				console.info(
+					JSON.stringify({
+						evt: "agent.tool",
+						threadId: input.threadId,
+						tool: toolName,
+						ok,
+						durationMs: Date.now() - startedAt,
+					}),
+				);
+			}
+		};
+
 	const tools = buildAgentTools({
 		provider: input.provider,
+		options: { requirePatchApproval: !!input.requirePatchApproval },
 		handlers: {
-			readResume: async () => {
+			readResume: timedToolHandler("read_resume", async () => {
 				const resume = await resumeService.getById({ id: input.resumeId, userId: input.userId });
 				return {
 					id: resume.id,
@@ -777,25 +834,56 @@ function createAgent(input: {
 					],
 					data: resume.data,
 				};
-			},
-			readAttachment: (attachmentId) =>
+			}),
+			readAttachment: timedToolHandler("read_attachment", (attachmentId: string) =>
 				readAttachment({ id: attachmentId, threadId: input.threadId, userId: input.userId }),
-			applyResumePatch: ({ title, summary, operations }) =>
-				applyResumePatch({
-					userId: input.userId,
-					threadId: input.threadId,
-					resumeId: input.resumeId,
-					title,
-					...(summary !== undefined ? { summary } : {}),
-					operations,
-				}),
+			),
+			applyResumePatch: timedToolHandler(
+				"apply_resume_patch",
+				({ title, summary, baseUpdatedAt, operations }: ApplyResumePatchInput) =>
+					applyResumePatch({
+						userId: input.userId,
+						threadId: input.threadId,
+						resumeId: input.resumeId,
+						...(input.draftRowId ? { messageId: input.draftRowId } : {}),
+						title,
+						...(summary !== undefined ? { summary } : {}),
+						...(baseUpdatedAt !== undefined ? { baseUpdatedAt } : {}),
+						operations,
+					}),
+			),
 		},
 	});
 
+	const instructionsText = buildAgentInstructions({ hasProviderNativeSearch: "web_search" in tools });
+
 	return new ToolLoopAgent({
-		model: input.model,
-		instructions: buildAgentInstructions({ hasProviderNativeSearch: "web_search" in tools }),
-		stopWhen: stepCountIs(MAX_AGENT_STEPS),
+		// Providers without native inputExamples support get them appended to the tool description.
+		model: wrapLanguageModel({ model: input.model, middleware: addToolInputExamplesMiddleware() }),
+		// The loop re-sends stable instructions every step; on anthropic, prompt caching pays from step 2.
+		instructions:
+			input.provider.provider === "anthropic"
+				? {
+						role: "system",
+						content: instructionsText,
+						providerOptions: { anthropic: { cacheControl: { type: "ephemeral" } } },
+					}
+				: instructionsText,
+		repairToolCall: repairAgentToolCall,
+		stopWhen: isStepCount(MAX_AGENT_STEPS),
+		maxOutputTokens: MAX_AGENT_OUTPUT_TOKENS,
+		maxRetries: MAX_AGENT_MODEL_RETRIES,
+		timeout: { stepMs: AGENT_STEP_TIMEOUT_MS },
+		// HMAC-signs approval requests at issuance and verifies them when replayed on the
+		// continuation run, so a client cannot forge or alter an approval payload. The agent
+		// runtime forwards constructor settings to streamText verbatim; ToolLoopAgentSettings
+		// does not type this key yet, hence the spread-cast.
+		...({ experimental_toolApprovalSecret: getAgentToolApprovalSecret() } as object),
+		// Runs before every loop step, so intra-run growth (N patches → N snapshots) is pruned too.
+		prepareStep: ({ messages }) => {
+			const pruned = pruneAgentModelContext(messages);
+			return pruned === messages ? {} : { messages: pruned };
+		},
 		tools,
 	});
 }
@@ -808,6 +896,7 @@ const threadSummarySelection = {
 	workingResumeId: schema.agentThread.workingResumeId,
 	title: schema.agentThread.title,
 	status: schema.agentThread.status,
+	reviewPatches: schema.agentThread.reviewPatches,
 	activeRunId: schema.agentThread.activeRunId,
 	activeStreamId: schema.agentThread.activeStreamId,
 	activeRunStartedAt: schema.agentThread.activeRunStartedAt,
@@ -931,6 +1020,21 @@ export const agentService = {
 			assertAgentEnvironment();
 
 			const thread = await getThread(input);
+
+			// Heal on open: clear a dead run's claim before reading messages so the client neither
+			// resumes a dead stream nor renders a perpetually "streaming" draft.
+			if (thread.activeRunId && isStaleAgentRun(thread)) {
+				await reapStaleAgentRun({
+					threadId: input.id,
+					userId: input.userId,
+					runId: thread.activeRunId,
+					streamId: thread.activeStreamId,
+				});
+				thread.activeRunId = null;
+				thread.activeStreamId = null;
+				thread.activeRunStartedAt = null;
+			}
+
 			const [messages, actions, attachments, resume] = await Promise.all([
 				listThreadMessages({ threadId: input.id, userId: input.userId }),
 				db
@@ -963,6 +1067,27 @@ export const agentService = {
 			};
 		},
 
+		update: async (input: { id: string; userId: string; reviewPatches: boolean }) => {
+			assertAgentEnvironment();
+
+			const thread = await getThread({ id: input.id, userId: input.userId });
+			// Approval behavior is captured when a run's agent is created; toggling mid-run would
+			// show "review on" while later patches from the same run still auto-apply.
+			if (thread.activeRunId && !isStaleAgentRun(thread)) {
+				throw new ORPCError("CONFLICT", { message: "Review settings cannot change while a run is active." });
+			}
+
+			const [updated] = await db
+				.update(schema.agentThread)
+				.set({ reviewPatches: input.reviewPatches })
+				.where(and(eq(schema.agentThread.id, input.id), eq(schema.agentThread.userId, input.userId)))
+				.returning();
+
+			if (!updated) throw new ORPCError("NOT_FOUND");
+
+			return toThreadSummary(updated);
+		},
+
 		archive: async (input: { id: string; userId: string }) => {
 			assertAgentEnvironment();
 
@@ -1025,7 +1150,16 @@ export const agentService = {
 				throw new ORPCError("CONFLICT", { message: "This thread is archived." });
 			}
 			if (thread.activeRunId) {
-				throw new ORPCError("CONFLICT", { message: "This thread already has an active run." });
+				if (!isStaleAgentRun(thread)) {
+					throw new ORPCError("CONFLICT", { message: "This thread already has an active run." });
+				}
+				// Lazy reap: a dead run's claim heals on the next send instead of CONFLICTing forever.
+				await reapStaleAgentRun({
+					threadId: input.threadId,
+					userId: input.userId,
+					runId: thread.activeRunId,
+					streamId: thread.activeStreamId,
+				});
 			}
 			if (!thread.workingResumeId || !thread.aiProviderId) {
 				throw new ORPCError("BAD_REQUEST", { message: "This thread is read-only." });
@@ -1034,6 +1168,12 @@ export const agentService = {
 				throw new ORPCError("BAD_REQUEST", { message: "Agent messages must be user messages or tool results." });
 			}
 
+			// Deliberately schema-less: provider-echoed tool parts must pass, and replayed history is never re-validated.
+			const validated = await safeValidateUIMessages({ messages: [input.message] });
+			if (!validated.success) {
+				throw new ORPCError("BAD_REQUEST", { message: "Invalid UI message parts." });
+			}
+
 			const [runnableProvider, attachments] = await Promise.all([
 				aiProvidersService.getRunnableById({
 					id: thread.aiProviderId,
@@ -1056,6 +1196,19 @@ export const agentService = {
 				throw new ORPCError("CONFLICT", { message: "This thread already has an active run." });
 			}
 
+			// Whole-run wall clock. Must abort with an AbortError (see abortReason) — never AbortSignal.timeout().
+			activeRunTimeouts.set(
+				runId,
+				setTimeout(() => controller.abort(abortReason("RUN_TIMEOUT")), AGENT_RUN_TIMEOUT_MS),
+			);
+
+			// Row + message the run streams into. A continuation reuses the existing assistant
+			// row (same uiMessage id); a fresh turn inserts a "streaming" draft row below.
+			let draftRowId: string | undefined;
+			let insertedDraft = false;
+			const responseMessageId = generateId();
+			let draftUiMessage: UIMessage = { id: responseMessageId, role: "assistant", parts: [] };
+
 			try {
 				let attachmentsForModel: AgentAttachmentRecord[] = [];
 
@@ -1064,11 +1217,17 @@ export const agentService = {
 						throw new ORPCError("BAD_REQUEST", { message: "Tool result messages cannot include attachments." });
 					}
 
-					await updateAssistantToolResultMessage({
+					// Merge AFTER the exclusive claim: concurrent approve/deny requests serialize on
+					// the claim instead of both persisting, and a merge that is rejected (or any later
+					// setup failure) releases the claim via the catch below. A response persisted by a
+					// failed earlier attempt re-enters as pendingContinuation and still gets its run.
+					const continuation = await updateAssistantToolResultMessage({
 						userId: input.userId,
 						threadId: input.threadId,
 						message: input.message,
 					});
+					draftRowId = continuation.rowId;
+					draftUiMessage = continuation.message;
 				} else {
 					attachmentsForModel = attachments;
 					const sequence = await getNextMessageSequence(input.threadId);
@@ -1109,10 +1268,25 @@ export const agentService = {
 				const messages = messageRows.map(toMessage);
 				const modelMessages = await convertToModelMessages(messages.map(toModelInputMessage));
 				const attachmentModelParts = buildAttachmentModelParts(await readAttachmentModelInputs(attachmentsForModel));
+
+				// Draft row inserted after the replay snapshot (so it is not replayed) and before the
+				// stream starts, so a crash mid-run leaves a resumable record instead of nothing.
+				if (input.message.role === "user") {
+					const draft = await insertDraftAssistantMessage({
+						userId: input.userId,
+						threadId: input.threadId,
+						uiMessageId: responseMessageId,
+					});
+					draftRowId = draft.rowId;
+					insertedDraft = true;
+				}
+
 				const agent = createAgent({
 					userId: input.userId,
 					threadId: input.threadId,
 					resumeId: thread.workingResumeId,
+					...(draftRowId ? { draftRowId } : {}),
+					requirePatchApproval: thread.reviewPatches,
 					provider: {
 						provider: runnableProvider.provider,
 						model: runnableProvider.model,
@@ -1130,25 +1304,58 @@ export const agentService = {
 				const result = await agent.stream({
 					messages: attachModelPartsToLatestUserMessage(modelMessages, attachmentModelParts),
 					abortSignal: controller.signal,
+					experimental_transform: smoothStream({ chunking: "word" }),
+					// Crash-safety: fold each finished step into the draft row so a process death
+					// mid-run loses at most the current step, never the whole transcript.
+					onStepEnd: async (step) => {
+						console.info(
+							JSON.stringify({
+								evt: "agent.step",
+								threadId: input.threadId,
+								runId,
+								step: step.stepNumber,
+								toolNames: step.toolCalls.map((call) => call.toolName),
+								usage: step.usage,
+								finishReason: step.finishReason,
+							}),
+						);
+						try {
+							draftUiMessage = applyStepToUiMessage(draftUiMessage, step);
+							const upserted = await upsertAssistantUiMessage({
+								userId: input.userId,
+								threadId: input.threadId,
+								...(draftRowId ? { rowId: draftRowId } : {}),
+								message: draftUiMessage,
+								status: "streaming",
+							});
+							draftRowId = upserted.rowId;
+						} catch (error) {
+							console.error("[agent] Failed to persist step draft", error);
+						}
+					},
 				});
 
 				return streamToEventIterator(
 					await agentStreamLifecycle.create(streamId, () =>
 						result.toUIMessageStream({
 							originalMessages: messages,
-							generateMessageId: generateId,
+							generateMessageId: () => responseMessageId,
 							sendSources: true,
+							// Round-trips inside the persisted uiMessage jsonb — no migration needed.
+							messageMetadata: ({ part }) =>
+								part.type === "finish" ? { usage: part.totalUsage, model: runnableProvider.model } : undefined,
 							onFinish: async ({ responseMessage, isAborted }) => {
 								let persistError: unknown;
 								try {
-									if (!(isAborted && canceledRunsWithPersistedPartial.has(runId))) {
-										await persistMessage({
-											userId: input.userId,
-											threadId: input.threadId,
-											message: responseMessage,
-											status: isAborted ? "canceled" : "completed",
-										});
-									}
+									await upsertAssistantUiMessage({
+										userId: input.userId,
+										threadId: input.threadId,
+										...(draftRowId ? { rowId: draftRowId } : {}),
+										// A continuation reuses the message; the SDK replaces primitive
+										// metadata, so prior-run usage must be summed back in.
+										message: withAccumulatedUsageMetadata(draftUiMessage, responseMessage),
+										status: isAborted ? "canceled" : "completed",
+									});
 								} catch (error) {
 									persistError = error;
 									throw error;
@@ -1159,6 +1366,7 @@ export const agentService = {
 										runId,
 										streamId,
 										primaryError: persistError,
+										preserveClaimForReaper: !!persistError,
 									});
 								}
 							},
@@ -1167,6 +1375,11 @@ export const agentService = {
 					),
 				);
 			} catch (error) {
+				if (insertedDraft && draftRowId) {
+					await deleteDraftIfEmpty({ rowId: draftRowId, threadId: input.threadId, userId: input.userId }).catch(
+						(cleanupError: unknown) => console.error("[agent] Failed to delete empty draft", cleanupError),
+					);
+				}
 				await cleanupActiveRun({
 					threadId: input.threadId,
 					userId: input.userId,
@@ -1178,47 +1391,34 @@ export const agentService = {
 			}
 		},
 
-		stop: async (input: { userId: string; threadId: string; partialMessage?: UIMessage }) => {
+		// Server-authored cancellation: the abort makes onFinish({isAborted: true}) persist exactly
+		// what the server generated. The deprecated client `partialMessage` is ignored.
+		stop: async (input: { userId: string; threadId: string }) => {
 			assertAgentEnvironment();
 
 			const thread = await getThread({ id: input.threadId, userId: input.userId });
 			const activeRunId = thread.activeRunId;
 			const activeStreamId = thread.activeStreamId;
+			if (!activeRunId) return;
 
-			let persistError: unknown;
-			let cleanupError: unknown;
-			try {
-				if (input.partialMessage) {
-					await persistMessage({
-						userId: input.userId,
-						threadId: input.threadId,
-						message: input.partialMessage,
-						status: "canceled",
-					});
-					if (activeRunId) canceledRunsWithPersistedPartial.add(activeRunId);
-				}
-			} catch (error) {
-				persistError = error;
-			} finally {
-				if (activeRunId) {
-					activeRunControllers.get(activeRunId)?.abort(abortReason("USER_STOPPED"));
-					activeRunControllers.delete(activeRunId);
-					try {
-						await clearActiveAgentRunIfCurrent({
-							threadId: input.threadId,
-							userId: input.userId,
-							runId: activeRunId,
-							streamId: activeStreamId,
-						});
-					} catch (error) {
-						cleanupError = error;
-						if (persistError) console.error("[agent] Failed to clear active run after stop persistence error", error);
-					}
-				}
+			const controller = activeRunControllers.get(activeRunId);
+			if (controller) {
+				// This replica owns the run: abort only. The claim stays until onFinish has
+				// persisted the terminal (canceled) state, so no new run can interleave with a
+				// still-committing tool or write. onFinish's cleanup releases the claim.
+				controller.abort(abortReason("USER_STOPPED"));
+				return;
 			}
 
-			if (persistError) throw persistError;
-			if (cleanupError) throw cleanupError;
+			// No local controller (another replica owns the run, or the process restarted):
+			// best-effort claim release so the user is not stuck. Cross-replica abort signaling
+			// is a documented follow-up; the stale-run reaper covers the leftovers.
+			await clearActiveAgentRunIfCurrent({
+				threadId: input.threadId,
+				userId: input.userId,
+				runId: activeRunId,
+				streamId: activeStreamId,
+			});
 		},
 		resume: async (input: { userId: string; threadId: string }) => {
 			assertAgentEnvironment();

packages/api/src/features/agent/threads.ts

@@ -64,6 +64,24 @@ export const threadsRouter = {
 		.use(mapAgentEnvironmentError)
 		.handler(({ context, input }) => agentService.threads.get({ id: input.id, userId: context.user.id })),
 
+	update: protectedProcedure
+		.route({
+			method: "PATCH",
+			path: "/agent/threads/{id}",
+			tags: ["Agent"],
+			operationId: "updateAgentThread",
+			summary: "Update agent thread settings",
+		})
+		.input(z.object({ id: z.string(), reviewPatches: z.boolean() }))
+		.use(mapAgentEnvironmentError)
+		.handler(({ context, input }) =>
+			agentService.threads.update({
+				id: input.id,
+				userId: context.user.id,
+				reviewPatches: input.reviewPatches,
+			}),
+		),
+
 	archive: protectedProcedure
 		.route({
 			method: "POST",

packages/api/src/features/agent/tools.test.ts

@@ -26,9 +26,15 @@ const handlers = {
 	}),
 };
 
-function buildTools(provider: AIProvider, options?: { model?: string; baseURL?: string }) {
+function buildTools(
+	provider: AIProvider,
+	options?: { model?: string; baseURL?: string; requirePatchApproval?: boolean },
+) {
 	return buildAgentTools({
 		provider: { provider, model: options?.model ?? "gpt-5-mini", apiKey: "test-key", baseURL: options?.baseURL ?? "" },
+		...(options?.requirePatchApproval !== undefined
+			? { options: { requirePatchApproval: options.requirePatchApproval } }
+			: {}),
 		handlers,
 	});
 }
@@ -76,6 +82,14 @@ describe("agent tools", () => {
 		},
 	);
 
+	it("marks apply_resume_patch as needing approval only when review is required", () => {
+		const gated = buildTools("openai-compatible", { requirePatchApproval: true });
+		const open = buildTools("openai-compatible");
+
+		expect(gated.apply_resume_patch).toMatchObject({ needsApproval: true });
+		expect(open.apply_resume_patch?.needsApproval).toBeUndefined();
+	});
+
 	it("keeps instructions explicit about native search availability", () => {
 		expect(buildAgentInstructions({ hasProviderNativeSearch: true })).toContain("Use web_search");
 		expect(buildAgentInstructions({ hasProviderNativeSearch: true })).toContain("user-provided public URLs");
@@ -92,7 +106,7 @@ describe("agent tools", () => {
 		expect(buildAgentInstructions({ hasProviderNativeSearch: false })).toContain(
 			"/customSections/0/items/0/description",
 		);
-		expect(buildAgentInstructions({ hasProviderNativeSearch: false })).toContain("never /data/basics/name or /name");
+		expect(buildAgentInstructions({ hasProviderNativeSearch: false })).toContain("never prefixed with /data");
 		expect(buildAgentInstructions({ hasProviderNativeSearch: false })).toContain("clean Markdown");
 	});
 });

packages/api/src/features/agent/tools.ts

@@ -1,9 +1,13 @@
+import type { ApplyResumePatchInput } from "@reactive-resume/ai/tools/agent-tool-contracts";
 import type { AIProvider } from "@reactive-resume/ai/types";
 import type { ToolSet } from "ai";
 import { createOpenAI } from "@ai-sdk/openai";
 import { tool } from "ai";
 import z from "zod";
-import { jsonPatchOperationSchema } from "@reactive-resume/resume/patch";
+import {
+	applyResumePatchInputSchema,
+	askUserQuestionInputSchema,
+} from "@reactive-resume/ai/tools/agent-tool-contracts";
 import { supportsProviderNativeWebSearch } from "../ai/capabilities";
 
 type AgentProviderConfig = {
@@ -13,16 +17,13 @@ type AgentProviderConfig = {
 	baseURL?: string | null;
 };
 
-const applyResumePatchToolInputSchema = z.object({
-	title: z.string().trim().min(1),
-	summary: z.string().trim().optional(),
-	operations: z.array(jsonPatchOperationSchema).min(1),
-});
-
-type ApplyResumePatchToolInput = z.infer<typeof applyResumePatchToolInputSchema>;
+type ApplyResumePatchToolInput = ApplyResumePatchInput;
 
 type BuildAgentToolsInput = {
 	provider: AgentProviderConfig;
+	options?: {
+		requirePatchApproval?: boolean;
+	};
 	handlers: {
 		readResume: () => Promise<unknown>;
 		readAttachment: (attachmentId: string) => Promise<unknown>;
@@ -51,8 +52,10 @@ function buildProviderNativeAgentTools(provider: AgentProviderConfig): ToolSet {
 }
 
 export function buildAgentInstructions({ hasProviderNativeSearch }: { hasProviderNativeSearch: boolean }) {
+	// The JSON-Pointer conventions live in the read_resume result, the tool descriptions, and the
+	// tool input examples; the instructions keep only a compact reminder to save tokens per step.
 	const baseInstructions =
-		"You are an expert resume-writing agent inside Reactive Resume. Help the user improve the working resume for a target role. Read the resume before editing. Respond to the user in clean Markdown with concise paragraphs, bullets, and bold text when it improves scanability. Apply concise, valid JSON Patch operations when changes are useful. Patch paths are evaluated against the resume data object returned by read_resume, so use paths like /basics/name for the visible name and never /data/basics/name or /name. Built-in sections must use /sections/<sectionId>, for example /sections/experience/items/0/description. Custom sections must use /customSections/<index>, for example /customSections/0/items/0/description, even when their type is experience, education, or another built-in section type. apply_resume_patch cannot rename the resume file/title metadata. Batch related JSON Patch operations into one apply_resume_patch call for each coherent edit instead of making repeated patch calls for the same request. Ask the user a question when a missing preference blocks a high-confidence edit.";
+		"You are an expert resume-writing agent inside Reactive Resume. Help the user improve the working resume for a target role. Read the resume before editing. Respond to the user in clean Markdown with concise paragraphs, bullets, and bold text when it improves scanability. Apply concise, valid JSON Patch operations when changes are useful. Patch paths are rooted at the resume data object returned by read_resume — for example /basics/name, /sections/experience/items/0/description, or /customSections/0/items/0/description — never prefixed with /data. apply_resume_patch cannot rename the resume file/title metadata. Batch related JSON Patch operations into one apply_resume_patch call for each coherent edit instead of making repeated patch calls for the same request. Ask the user a question when a missing preference blocks a high-confidence edit.";
 
 	if (!hasProviderNativeSearch) {
 		return `${baseInstructions} Live web research is unavailable with the selected provider or model. If the user asks you to browse, search the web, fetch a URL, or use current online context, briefly tell them live web research is unavailable with the selected provider/model and ask them to paste or attach the relevant content. Continue normal resume editing using the resume, chat context, and attachments.`;
@@ -67,11 +70,7 @@ export function buildAgentTools(input: BuildAgentToolsInput): ToolSet {
 		ask_user_question: tool({
 			description:
 				"Ask the user a short question when you need a preference, missing fact, or choice before continuing. Provide 2-4 recommended answer choices when possible.",
-			inputSchema: z.object({
-				question: z.string().trim().min(1),
-				choices: z.array(z.string().trim().min(1)).min(1).max(4).optional(),
-				recommendedChoice: z.string().trim().optional(),
-			}),
+			inputSchema: askUserQuestionInputSchema,
 		}),
 		read_resume: tool({
 			description: "Read the current working resume JSON and metadata.",
@@ -86,8 +85,22 @@ export function buildAgentTools(input: BuildAgentToolsInput): ToolSet {
 		}),
 		apply_resume_patch: tool({
 			description:
-				"Apply one cohesive batch of JSON Patch operations to the working resume data immediately. Paths are rooted at resume data; use /basics/name for the visible resume name, not /data/basics/name or /name. This tool cannot rename the resume file/title metadata. The user can restore the draft to the snapshot captured before a patch later.",
-			inputSchema: applyResumePatchToolInputSchema,
+				"Apply one cohesive batch of JSON Patch operations to the working resume data immediately. Paths are rooted at resume data; use /basics/name for the visible resume name, not /data/basics/name or /name. This tool cannot rename the resume file/title metadata. The user can restore the draft to the snapshot captured before a patch later. The result includes the complete post-patch resume; array indexes may have shifted — base further patches on it, never on an earlier read_resume. Always pass baseUpdatedAt: the updatedAt of the read_resume or apply_resume_patch result these operations were built against; the edit is rejected if the resume changed since.",
+			inputSchema: applyResumePatchInputSchema,
+			inputExamples: [
+				{
+					input: {
+						title: "Tighten the summary",
+						baseUpdatedAt: "2026-08-20T10:15:00.000Z",
+						operations: [
+							{ op: "replace", path: "/sections/summary/content", value: "Impact-driven engineer with 8 years…" },
+						],
+					},
+				},
+			],
+			// Static approval gate: when the thread has "Review edits" on, the loop halts with an
+			// approval-requested part instead of executing; the SDK executes after approval.
+			...(input.options?.requirePatchApproval ? { needsApproval: true } : {}),
 			execute: (toolInput) => input.handlers.applyResumePatch(toolInput),
 		}),
 	};

packages/api/src/features/ai/credentials.ts

@@ -84,6 +84,15 @@ export function redactEncryptedCredential(fields: StoredCredentialFields): Redac
 	};
 }
 
+// Domain-separated from the AES key. Deterministic derivation (no new env var) means an approval
+// signature minted when a run halts still verifies at continuation, even across a server restart.
+export function getAgentToolApprovalSecret() {
+	const secret = getEncryptionSecret();
+	if (!secret) throw new Error("AI_CREDENTIAL_ENCRYPTION_UNAVAILABLE");
+
+	return createHash("sha256").update(`${secret}:agent-tool-approval`).digest("hex");
+}
+
 function isCredentialEncryptionConfigured() {
 	return !!getEncryptionSecret();
 }

packages/api/src/features/ai/service.test.ts

@@ -69,7 +69,8 @@ function stubRejectedFetch(error: unknown) {
 	return fetchMock;
 }
 
-const { analyzeResume, testConnection } = await import("./service");
+const { aiService, testConnection } = await import("./service");
+const { analyzeResume } = aiService;
 
 describe("AI provider connection test", () => {
 	it("names the rejected key instead of reporting a transport failure", async () => {