fix(ai): keep a chat turn's Durable Object alive with a heartbeat alarm

apps/ai/src/chat/ChatSession.test.ts

@@ -373,6 +373,47 @@ describe("ChatSession turn mutex", () => {
 	})
 })
 
+describe("ChatSession turn heartbeat", () => {
+	/**
+	 * An outbound fetch never keeps a Durable Object alive, so an autonomous turn nobody was watching
+	 * was evicted about two minutes in. The alarm is the incoming event that prevents it.
+	 */
+	it("arms the alarm when a turn starts, and re-arms it while that turn runs", () => {
+		const { session, state } = makeSession()
+		session.beginTurn({ sessionId: "org_test:tab", messageId: "u1", text: "hi", tenant: TENANT })
+		assert.lengthOf(state.alarms, 1)
+
+		session.alarm()
+		assert.lengthOf(state.alarms, 2)
+		assert.isTrue(session.running())
+	})
+
+	it("stops re-arming once no turn holds the slot", () => {
+		const { session, state, turnId } = makeSession()
+		session.beginTurn({ sessionId: "org_test:tab", messageId: "u1", text: "hi", tenant: TENANT })
+		session.endTurn(turnId()!)
+
+		session.alarm()
+		assert.lengthOf(state.alarms, 1)
+	})
+
+	/** A fresh activation holds the claim but not the fiber: the object was evicted mid-turn. */
+	it("releases a slot whose turn did not survive an eviction", () => {
+		const { session, state, turnId } = makeSession()
+		session.beginTurn({ sessionId: "org_test:tab", messageId: "u1", text: "hi", tenant: TENANT })
+		const orphaned = turnId()!
+
+		const revived = new ChatSession(state, {})
+		revived.alarm()
+
+		assert.isFalse(revived.running())
+		const last = revived.since(0).at(-1)
+		assert.strictEqual(last?.type, "turn-end")
+		assert.strictEqual(last?.type === "turn-end" ? last.messageId : undefined, orphaned)
+		assert.strictEqual(last?.type === "turn-end" ? last.reason : undefined, "error")
+	})
+})
+
 describe("ChatSession.subscribe", () => {
 	/** Read the whole subscription, which ends at `turn-end`. */
 	const drain = async (stream: ReadableStream<Uint8Array>): Promise<string> => {

apps/ai/src/chat/ChatSession.ts

@@ -47,9 +47,12 @@ import {
 } from "@maple/domain/chat-session"
 import { type ChatSessionStub } from "@maple/domain/chat-session-stub"
 
-/** What the class reads off its Durable Object state: the SQLite handle and the object's own `waitUntil`. */
+/** What the class reads off its Durable Object state: SQLite, the alarm, and the object's own `waitUntil`. */
 interface ChatSessionState {
-	readonly storage: { readonly sql: SqlStorage }
+	readonly storage: {
+		readonly sql: SqlStorage
+		setAlarm(scheduledTime: number): Promise<void>
+	}
 	waitUntil(promise: Promise<unknown>): void
 }
 
@@ -118,6 +121,16 @@ const RETRY_HINT = "retry: 1000\n\n"
 const TURN_STALE_MS = 15 * 60 * 1000
 const CHAT_TURN_FAILED = "Maple couldn't complete this response."
 
+/**
+ * How often a running turn re-arms the object's alarm.
+ *
+ * An outbound `fetch` never keeps a Durable Object alive, even while the response streams, and an
+ * object with no incoming request or event for 70-140 seconds is evicted. A chat turn survives
+ * because the open page holds a subscription; an autonomous investigation nobody is watching was
+ * evicted about two minutes in, mid-run (seen 2026-09-15). The alarm is the event that prevents it.
+ */
+const TURN_HEARTBEAT_MS = 30 * 1000
+
 export class ChatSession {
 	private readonly sql: SqlStorage
 
@@ -132,6 +145,12 @@ export class ChatSession {
 	 */
 	private waiters = new Set<() => void>()
 
+	/**
+	 * The turn this activation is actually running. SQL says which turn holds the slot; only this
+	 * says its fiber still exists — an evicted object comes back with the claim and without the turn.
+	 */
+	private liveTurn: string | undefined
+
 	constructor(
 		private readonly ctx: ChatSessionState,
 		private readonly env: Record<string, unknown>,
@@ -364,8 +383,11 @@ export class ChatSession {
 			turnId,
 		)
 		this.append({ type: "user-message", id: input.messageId, text: input.text })
+		this.liveTurn = turnId
+		this.armHeartbeat()
 		// `waitUntil` on the DO's own context: the turn is now this object's work, and it outlives
-		// whatever request asked for it.
+		// whatever request asked for it. `waitUntil` alone does not keep the object in memory — the
+		// heartbeat alarm does.
 		this.ctx.waitUntil(this.runTurn(input.sessionId, turnId, input.tenant))
 		return { cursor, messageId: input.messageId }
 	}
@@ -412,6 +434,28 @@ export class ChatSession {
 		return this.isRunning()
 	}
 
+	/**
+	 * The heartbeat. Re-arms while this activation runs the turn that holds the slot.
+	 *
+	 * A slot held by a turn this activation is not running means the object was evicted mid-turn (a
+	 * deploy, or eviction before the heartbeat existed): the fiber is gone, so the slot is released
+	 * with a terminal event now rather than when the 15-minute watchdog expires it.
+	 */
+	alarm(): void {
+		const messageId = this.runningTurn()
+		if (messageId === undefined) return
+		if (messageId !== null && this.liveTurn !== messageId) {
+			this.clearRunning()
+			this.append({ type: "turn-end", messageId, reason: "error", error: CHAT_TURN_FAILED })
+			return
+		}
+		this.armHeartbeat()
+	}
+
+	private armHeartbeat(): void {
+		this.ctx.waitUntil(this.ctx.storage.setAlarm(Date.now() + TURN_HEARTBEAT_MS).catch(() => undefined))
+	}
+
 	/**
 	 * Drive one turn to completion, appending events as they are produced.
 	 *
@@ -439,6 +483,7 @@ export class ChatSession {
 				})
 			}
 		} finally {
+			if (this.liveTurn === messageId) this.liveTurn = undefined
 			this.endTurn(messageId)
 		}
 	}
@@ -643,6 +688,9 @@ type EffectRpc<Stub> = {
 		: never
 }
 
+/** The RPC surface plus the heartbeat alarm, which alchemy's bridge dispatches as the object's `alarm`. */
+type ChatSessionObjectApi = EffectRpc<ChatSessionStub> & { readonly alarm: () => Effect.Effect<void> }
+
 /**
  * The session's methods, one Effect each. alchemy runs the Effect per RPC call and hands its value
  * back as-is — a `ReadableStream` included, which Workers RPC carries by reference — so
@@ -660,7 +708,8 @@ export const chatSessionRpc = (session: ChatSession) =>
 		holdsTurn: (messageId) => Effect.sync(() => session.holdsTurn(messageId)),
 		endTurn: (messageId) => Effect.sync(() => session.endTurn(messageId)),
 		abort: () => Effect.sync(() => session.abort()),
-	}) satisfies EffectRpc<ChatSessionStub>
+		alarm: () => Effect.sync(() => session.alarm()),
+	}) satisfies ChatSessionObjectApi
 
 /**
  * One activation, in alchemy's two phases: the outer Effect resolves the state and env (it also
@@ -690,10 +739,10 @@ export const activateChatSession = Effect.map(
  * The props-carrying class form is what makes room for that: the single-argument overload takes an
  * implementation and no props, so the implementation moves to `ChatSessionLive` below.
  */
-export class ChatSessionObject extends Cloudflare.DurableObject<
-	ChatSessionObject,
-	EffectRpc<ChatSessionStub>
->()("ChatSession", { transferredFrom: "api" }) {}
+export class ChatSessionObject extends Cloudflare.DurableObject<ChatSessionObject, ChatSessionObjectApi>()(
+	"ChatSession",
+	{ transferredFrom: "api" },
+) {}
 
 /** The activation, as the layer the host Worker provides. */
 // `<never>` pinned: the activation's requirements are all `DurableObjectServices`,

apps/ai/src/chat/ChatSessionObject.test.ts

@@ -75,7 +75,9 @@ describe("the ChatSession Durable Object on alchemy's form", () => {
 			assert.isDefined(begun)
 			assert.strictEqual(yield* rpc.running(), true)
 			// The turn was scheduled on the object's own context, not the caller's.
-			assert.strictEqual(state.pending.length, 1)
+			// Two pieces of object work: the turn and the heartbeat alarm that keeps the object alive.
+			assert.strictEqual(state.pending.length, 2)
+			assert.lengthOf(state.alarms, 1)
 			yield* rpc.abort()
 			assert.strictEqual(yield* rpc.running(), false)
 		}),

apps/ai/src/chat/events.test.ts

@@ -33,6 +33,10 @@ describe("toChatEvents", () => {
 		])
 	})
 
+	it("drops an empty text delta rather than logging it", () => {
+		assert.deepEqual(toChatEvents(event("TextDelta", { text: "" }), base), [])
+	})
+
 	it("announces a declared call with its arguments", () => {
 		assert.deepEqual(
 			toChatEvents(

apps/ai/src/chat/events.ts

@@ -98,6 +98,9 @@ export const toChatEvents = (
 		case "RunStarted":
 			return [tagged(context, { type: "turn-start", messageId: context.messageId })]
 		case "TextDelta":
+			// Some providers stream an empty delta per reasoning token; a run wrote two thousand of
+			// them into the session log in a minute, and the log is what every reconnect replays.
+			if (event.text === "") return []
 			return [tagged(context, { type: "text-delta", messageId: context.messageId, text: event.text })]
 		case "ToolCallDeclared":
 			return [

apps/ai/src/chat/run.ts

@@ -92,6 +92,8 @@ export interface ChatRunInput {
 	readonly toolExecutor: McpToolExecutorApi
 	readonly model: ResolvedModel
 	readonly submitDiagnosis: SubmitDiagnosis
+	/** This run is an autonomous pass's close-out: a report it files is a partial. */
+	readonly closeOut?: boolean
 	/** The message the user just sent, which is this run's input. */
 	readonly text: string
 	readonly history: ReadonlyArray<ChatMessage>
@@ -127,6 +129,7 @@ export const runChatTurn = (input: ChatRunInput) => {
 		input.submitDiagnosis,
 		input.usage,
 		input.model.name,
+		input.closeOut === true,
 	)
 
 	const toolkit = Toolkit.merge(maple.toolkit, ...(completion === undefined ? [] : [completion.toolkit]))

apps/ai/src/chat/tools.ts

@@ -126,6 +126,8 @@ export const buildDiagnosisCompletion = (
 	submitDiagnosis: SubmitDiagnosis,
 	usage: RunUsage,
 	modelName: string,
+	/** This run is the close-out: whatever it files is a partial, and lands as `inconclusive`. */
+	partial = false,
 ) => {
 	const investigationId = investigationForSession(sessionId)
 	if (investigationId === undefined) return undefined
@@ -143,6 +145,7 @@ export const buildDiagnosisCompletion = (
 						model: modelName,
 						inputTokens: usage.input,
 						outputTokens: usage.output,
+						...(partial ? { partial: true } : undefined),
 					}),
 				).pipe(
 					Effect.tap(() => Effect.sync(() => (submitted = true))),

apps/ai/src/chat/turn-runner.ts

@@ -291,14 +291,19 @@ export const runChatSessionTurn = async (input: RunChatSessionTurnInput): Promis
 		// in prose or died on a model error gets one close-out turn first, so its terminal is held
 		// back until the outcome is known.
 		let held: Extract<ChatTurnEvent, { readonly type: "turn-end" }> | undefined
-		const run = (turn: { readonly text: string; readonly history: ReadonlyArray<ChatMessage> }) =>
+		const run = (turn: {
+			readonly text: string
+			readonly history: ReadonlyArray<ChatMessage>
+			readonly closeOut?: boolean
+		}) =>
 			runChatTurn({
 				sessionId: input.sessionId,
 				messageId: input.messageId,
 				tenant,
 				toolExecutor,
 				model,
 				submitDiagnosis: investigations.submitDiagnosis,
+				...(turn.closeOut === true ? { closeOut: true } : undefined),
 				text: turn.text,
 				history: turn.history,
 				...(compaction === undefined ? undefined : { compaction }),
@@ -349,7 +354,11 @@ export const runChatSessionTurn = async (input: RunChatSessionTurnInput): Promis
 		if (!submitted && holdsTurn()) {
 			held = undefined
 			const closeOut = yield* recoverAutonomousFailure(
-				run({ text: CLOSE_OUT_PROMPT, history: withToolTranscript(input.session.history()) }),
+				run({
+					text: CLOSE_OUT_PROMPT,
+					history: withToolTranscript(input.session.history()),
+					closeOut: true,
+				}),
 			)
 			submitted = closeOut.submittedDiagnosis
 			yield* Effect.annotateCurrentSpan("maple.investigation.closed_out", submitted)

apps/ai/src/mcp/tools/llm-tools.test.ts

@@ -40,6 +40,38 @@ const handlerFor = (executor: McpToolExecutorApi, name: string) => {
 }
 
 describe("buildMapleToolkit", () => {
+	/**
+	 * A tool error is the call's answer, not the run's end. The engine ends a run on a declared
+	 * failure — right for a proposal, wrong for a rejected query the model can rewrite.
+	 */
+	it("answers a failed tool with its message instead of failing the run", async () => {
+		const executor: McpToolExecutorApi = {
+			execute: () =>
+				Effect.succeed({
+					isError: true,
+					content: [
+						{ type: "text" as const, text: "Tool failed: SQL rejected (MissingOrgFilter)" },
+					],
+				}),
+		}
+		const result = await Effect.runPromise(
+			Effect.result(handlerFor(executor, "run_sql")({ sql: "select 1" }, {} as never)),
+		)
+		assert.isTrue(Result.isSuccess(result))
+		assert.include(Result.isSuccess(result) ? result.success : "", "MissingOrgFilter")
+	})
+
+	it("answers a tool that died with a summary instead of failing the run", async () => {
+		const executor: McpToolExecutorApi = {
+			execute: () => Effect.die(new Error("connection reset")),
+		}
+		const result = await Effect.runPromise(
+			Effect.result(handlerFor(executor, "list_services")({ limit: 10 }, {} as never)),
+		)
+		assert.isTrue(Result.isSuccess(result))
+		assert.include(Result.isSuccess(result) ? result.success : "", "Tool failed")
+	})
+
 	it("records a gated tool's description on its span as the model saw it", async () => {
 		const { executor } = countingExecutor()
 		const handler = buildMapleToolkit(executor, TENANT, { gate: () => true }).handlers.list_services
@@ -66,7 +98,11 @@ describe("buildMapleToolkit", () => {
 			assert.isTrue(Result.isSuccess(result), `attempt ${attempt + 1} should have run`)
 		}
 
-		assert.isTrue(Result.isFailure(await call()))
+		// Answered, not failed: a declared failure would end the run, and a model repeating itself
+		// needs to be told, not stopped.
+		const fourth = await call()
+		assert.isTrue(Result.isSuccess(fourth))
+		assert.include(Result.isSuccess(fourth) ? fourth.success : "", "already been called")
 		assert.equal(dispatched(), 3, "the fourth call must not reach the executor")
 	})
 

apps/ai/src/mcp/tools/llm-tools.ts

@@ -82,10 +82,12 @@ export interface BuildMapleToolsOptions {
 }
 
 /**
- * A tool's failure as the model sees it: one line of text, never a cause.
+ * A tool call the run must stop on: an approval-gated mutation the model proposed.
  *
- * Declared rather than thrown so the runtime records a tool failure the model can route around,
- * instead of the failure ending the run.
+ * The engine ends the run on a declared failure (that is how a proposal becomes the turn's last
+ * word and reaches the approval card), so this is NOT how an ordinary tool error is reported.
+ * Those are handed back as the call's text — see `dispatch` below — so the model can route
+ * around them; every one of them used to end the whole pass.
  */
 export class MapleToolFailure extends Schema.TaggedError<MapleToolFailure>()(
 	"@maple/api/mcp/MapleToolFailure",
@@ -155,21 +157,20 @@ export const buildMapleToolkit = (
 	const handlers = Object.fromEntries(
 		definitions.map((definition) => {
 			const gated = options.gate?.(definition.name) ?? false
+			// A tool that fails — a rejected query, an unknown tool, a tenant error — answers with its
+			// message as an ordinary result. A declared failure ends the run, which is right for a
+			// proposal and wrong for a bad SQL statement the model can simply rewrite.
 			const dispatch = (params: unknown) =>
 				executor.execute(tenant, definition.name, params, options.surface ?? "chat").pipe(
-					Effect.flatMap((result) =>
-						result.isError
-							? fail(toolResultText(result))
-							: Effect.succeed(toolResultText(result)),
+					Effect.map((result) => toolResultText(result)),
+					Effect.catchCause((cause) =>
+						Effect.succeed(`Tool failed: ${summarizeToolFailure(cause)}`),
 					),
-					// A tool that fails outright (unknown tool, tenant error) must not kill the run —
-					// hand the model the message and let it route around.
-					Effect.catchCause((cause) => fail(`Tool failed: ${summarizeToolFailure(cause)}`)),
 				)
 			const handle = (params: unknown) => {
 				if (gated) return fail(`${definition.name} requires user approval and was not executed.`)
 				if (repeats(dispatched, definition.name, params) > IDENTICAL_CALL_LIMIT) {
-					return fail(
+					return Effect.succeed(
 						`${definition.name} has already been called ${IDENTICAL_CALL_LIMIT} times with these ` +
 							"exact arguments in this turn. Read the result you already have, or call it differently.",
 					)