From 101d45eed0f901d854354f33d05ade717261d95d Mon Sep 17 00:00:00 2001 From: Makisuo Date: Tue, 15 Sep 2026 00:57:28 +0200 Subject: [PATCH 1/2] fix(ai): keep a pass alive through tool errors, file close-outs as partials MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Found running the single agent end to end against a local stack. A tool that failed ended the whole pass. Maple's tool handlers declared `MapleToolFailure` as the tool's failure type, and the engine ends the run on a declared failure — that is how an approval-gated proposal becomes the turn's last word — so a rejected `run_sql` on the first call killed the investigation before it had read anything. Tool errors are now returned as the call's text and the model rewrites the call; only the approval gate still fails the run. A close-out's report landed as `diagnosed`. Whatever the close-out files is a partial by construction, so `SubmitDiagnosisRequest` carries `partial` and the service routes it to the inconclusive writer: low confidence, no severity, no `diagnosed_at`, no issue-side writes. A model stream that stalls now fails after two minutes without a chunk. One did, mid-sentence, on the second run; the engine's ten-minute rail never interrupted it, and the investigation sat until the 15-minute stale sweep marked it failed with nothing filed. Failing the stream hands the run to the close-out instead. Empty text deltas are dropped from the session log. One provider streamed one per reasoning token, which put two thousand empty events into a Durable Object's SQLite in a minute — and the log is what every reconnect replays. --- apps/ai/src/chat/events.test.ts | 4 ++ apps/ai/src/chat/events.ts | 3 + apps/ai/src/chat/run.ts | 3 + apps/ai/src/chat/tools.ts | 3 + apps/ai/src/chat/turn-runner.ts | 13 +++- apps/ai/src/mcp/tools/llm-tools.test.ts | 38 +++++++++++- apps/ai/src/mcp/tools/llm-tools.ts | 23 +++---- apps/ai/src/platform/genai-spans.ts | 22 +++++++ .../errors/InvestigationService.test.ts | 19 ++++++ .../services/errors/InvestigationService.ts | 62 ++++++++++++------- packages/domain/src/http/investigations.ts | 5 ++ 11 files changed, 159 insertions(+), 36 deletions(-) diff --git a/apps/ai/src/chat/events.test.ts b/apps/ai/src/chat/events.test.ts index d05eb4d49..6f5d342ba 100644 --- a/apps/ai/src/chat/events.test.ts +++ b/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( diff --git a/apps/ai/src/chat/events.ts b/apps/ai/src/chat/events.ts index 1fb30f2e3..0681953d5 100644 --- a/apps/ai/src/chat/events.ts +++ b/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 [ diff --git a/apps/ai/src/chat/run.ts b/apps/ai/src/chat/run.ts index d382294eb..ba15f3aa1 100644 --- a/apps/ai/src/chat/run.ts +++ b/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 @@ -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])) diff --git a/apps/ai/src/chat/tools.ts b/apps/ai/src/chat/tools.ts index 8a343965a..226752b56 100644 --- a/apps/ai/src/chat/tools.ts +++ b/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))), diff --git a/apps/ai/src/chat/turn-runner.ts b/apps/ai/src/chat/turn-runner.ts index f3744b674..73f5a028c 100644 --- a/apps/ai/src/chat/turn-runner.ts +++ b/apps/ai/src/chat/turn-runner.ts @@ -291,7 +291,11 @@ 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 | undefined - const run = (turn: { readonly text: string; readonly history: ReadonlyArray }) => + const run = (turn: { + readonly text: string + readonly history: ReadonlyArray + readonly closeOut?: boolean + }) => runChatTurn({ sessionId: input.sessionId, messageId: input.messageId, @@ -299,6 +303,7 @@ export const runChatSessionTurn = async (input: RunChatSessionTurnInput): Promis 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) diff --git a/apps/ai/src/mcp/tools/llm-tools.test.ts b/apps/ai/src/mcp/tools/llm-tools.test.ts index fbfc72e60..12a7eeb8a 100644 --- a/apps/ai/src/mcp/tools/llm-tools.test.ts +++ b/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") }) diff --git a/apps/ai/src/mcp/tools/llm-tools.ts b/apps/ai/src/mcp/tools/llm-tools.ts index 801afab09..bb539bc54 100644 --- a/apps/ai/src/mcp/tools/llm-tools.ts +++ b/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()( "@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.", ) diff --git a/apps/ai/src/platform/genai-spans.ts b/apps/ai/src/platform/genai-spans.ts index 050e1483f..46b9e628f 100644 --- a/apps/ai/src/platform/genai-spans.ts +++ b/apps/ai/src/platform/genai-spans.ts @@ -20,6 +20,7 @@ import { } from "@maple/domain/gen-ai" import { Effect, Option, Predicate, Stream } from "effect" import type { Tracer } from "effect" +import * as AiError from "effect/unstable/ai/AiError" import type * as LanguageModel from "effect/unstable/ai/LanguageModel" import type * as Prompt from "effect/unstable/ai/Prompt" import * as Telemetry from "effect/unstable/ai/Telemetry" @@ -337,6 +338,23 @@ const modelCallTransformer = } } +/** + * A model stream that produces nothing for this long is dead. Failing it hands the run to the + * close-out; left alone, a stalled stream sat past the engine's duration rail until the + * 15-minute stale sweep marked the investigation failed with nothing filed (seen 2026-09-14). + * Generous, because a reasoning burst before the first token can run long. + */ +const MODEL_STREAM_IDLE_TIMEOUT = "2 minutes" + +const idleTimeout = (): AiError.AiError => + new AiError.AiError({ + module: "Maple", + method: "streamText", + reason: new AiError.UnknownError({ + description: `Model stream produced nothing for ${MODEL_STREAM_IDLE_TIMEOUT}`, + }), + }) + /** * Build a provider's language model so that every `streamText` call — the only call Maple makes, and * the only one effect-agent makes — annotates its own span. @@ -372,6 +390,10 @@ export const instrumentLanguageModel = ( }) : Effect.void, ), + Stream.timeoutOrElse({ + duration: MODEL_STREAM_IDLE_TIMEOUT, + orElse: () => Stream.fail(idleTimeout()), + }), Stream.provideService( Telemetry.CurrentSpanTransformer, modelCallTransformer(telemetry, timing), diff --git a/packages/backend/src/services/errors/InvestigationService.test.ts b/packages/backend/src/services/errors/InvestigationService.test.ts index ed4bc6abb..619063c3d 100644 --- a/packages/backend/src/services/errors/InvestigationService.test.ts +++ b/packages/backend/src/services/errors/InvestigationService.test.ts @@ -172,6 +172,25 @@ describe("InvestigationService", () => { }).pipe(Effect.provide(makeLayer())), ) + it.effect("files a close-out's partial report as inconclusive, at low confidence", () => + Effect.gen(function* () { + const service = yield* InvestigationService + const created = yield* service.createInvestigation(ORG, null, freeformRequest("cut short")) + + const partial = yield* service.submitDiagnosis( + ORG, + created.id, + new SubmitDiagnosisRequest({ report: sampleReport(), model: "test-model", partial: true }), + ) + assert.strictEqual(partial.status, "inconclusive") + assert.strictEqual(partial.confidence, "low") + // The hub shows the incident's own severity; a partial assessed none. + assert.isNull(partial.severity) + assert.isNull(partial.diagnosedAt) + assert.strictEqual(partial.report?.suspectedCause, sampleReport().suspectedCause) + }).pipe(Effect.provide(makeLayer())), + ) + it.effect("submit_diagnosis records the turn's tokens on the row without metering them", () => { // The chat-session runner meters that turn in full, keyed on the turn. A second // meter here billed the same tokens twice; billing here *instead* lost the charge diff --git a/packages/backend/src/services/errors/InvestigationService.ts b/packages/backend/src/services/errors/InvestigationService.ts index 9db05c1e8..ac966c14a 100644 --- a/packages/backend/src/services/errors/InvestigationService.ts +++ b/packages/backend/src/services/errors/InvestigationService.ts @@ -3,7 +3,6 @@ import { randomUUID } from "node:crypto" import { type AiTriageIncidentKind, AiTriageResult, - type InvestigationConfidence, InvestigationCreateRequest, InvestigationDataCorruptionError, InvestigationDocument, @@ -26,7 +25,11 @@ import { investigations, type InvestigationRow } from "@maple/db" import { WorkerEnvironment } from "@maple/infra/worker-runtime" import { and, desc, eq, isNull, lt, sql } from "drizzle-orm" import { Clock, Context, Effect, Layer, Option, Schema } from "effect" -import { applyDiagnosisWrites, subjectTypeOf } from "@maple/backend/services/errors/apply-diagnosis" +import { + applyDiagnosisWrites, + applyInconclusiveWrites, + subjectTypeOf, +} from "@maple/backend/services/errors/apply-diagnosis" import { startInvestigationTurn } from "@maple/backend/services/errors/investigation-start" import { STALE_MS, @@ -603,26 +606,41 @@ export class InvestigationService extends Context.Service model: Schema.optionalKey(Schema.String), inputTokens: Schema.optionalKey(Schema.Number), outputTokens: Schema.optionalKey(Schema.Number), + /** + * The report is a partial: filed by the close-out turn after the pass itself ended + * without one. It lands as `inconclusive`, never as a diagnosis. + */ + partial: Schema.optionalKey(Schema.Boolean), }) {} // Errors From e58a0462eda422d71d3a94731a3610338e01ac5e Mon Sep 17 00:00:00 2001 From: Makisuo Date: Tue, 15 Sep 2026 01:30:41 +0200 Subject: [PATCH 2/2] fix(ai): keep a chat turn's Durable Object alive with a heartbeat alarm MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit An outbound fetch never keeps a Durable Object in memory, even while the response body 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 does not: running end to end locally, an unobserved pass went silent about two minutes in, with no error and no close-out, and sat until the 15-minute stale sweep marked it failed. Every alert-triggered investigation is unobserved. While a turn holds the slot, the object re-arms its alarm every 30 seconds; the alarm is the event that prevents eviction. When the alarm lands on an activation that holds the claim but is not running the turn — the object was evicted anyway, by a deploy — it releases the slot with a terminal event instead of waiting for the watchdog. --- apps/ai/src/chat/ChatSession.test.ts | 41 ++++++++++++++ apps/ai/src/chat/ChatSession.ts | 65 +++++++++++++++++++--- apps/ai/src/chat/ChatSessionObject.test.ts | 4 +- apps/ai/test/chat/fake-do-state.ts | 16 +++++- 4 files changed, 115 insertions(+), 11 deletions(-) diff --git a/apps/ai/src/chat/ChatSession.test.ts b/apps/ai/src/chat/ChatSession.test.ts index 817d5f9cf..e9601198b 100644 --- a/apps/ai/src/chat/ChatSession.test.ts +++ b/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): Promise => { diff --git a/apps/ai/src/chat/ChatSession.ts b/apps/ai/src/chat/ChatSession.ts index 435740dc6..e4c02c55f 100644 --- a/apps/ai/src/chat/ChatSession.ts +++ b/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 + } waitUntil(promise: Promise): 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, @@ -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 = { : never } +/** The RPC surface plus the heartbeat alarm, which alchemy's bridge dispatches as the object's `alarm`. */ +type ChatSessionObjectApi = EffectRpc & { readonly alarm: () => Effect.Effect } + /** * 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 + 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 ->()("ChatSession", { transferredFrom: "api" }) {} +export class ChatSessionObject extends Cloudflare.DurableObject()( + "ChatSession", + { transferredFrom: "api" }, +) {} /** The activation, as the layer the host Worker provides. */ // `` pinned: the activation's requirements are all `DurableObjectServices`, diff --git a/apps/ai/src/chat/ChatSessionObject.test.ts b/apps/ai/src/chat/ChatSessionObject.test.ts index 9d6ffadb0..dc4ecaffe 100644 --- a/apps/ai/src/chat/ChatSessionObject.test.ts +++ b/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) }), diff --git a/apps/ai/test/chat/fake-do-state.ts b/apps/ai/test/chat/fake-do-state.ts index f95c9b2d5..a37cd93d7 100644 --- a/apps/ai/test/chat/fake-do-state.ts +++ b/apps/ai/test/chat/fake-do-state.ts @@ -13,7 +13,12 @@ import { DatabaseSync } from "node:sqlite" /** Everything the DO under test reads off its state, and nothing more. */ export interface FakeDurableObjectState { - readonly storage: { readonly sql: SqlStorage } + readonly storage: { + readonly sql: SqlStorage + readonly setAlarm: (scheduledTime: number) => Promise + } + /** Every alarm time the object asked for, in order. */ + readonly alarms: Array /** Collected rather than awaited, so a test can drive the turn itself. */ readonly waitUntil: (promise: Promise) => void readonly pending: Array> @@ -22,6 +27,7 @@ export interface FakeDurableObjectState { export const makeFakeDurableObjectState = (): FakeDurableObjectState => { const db = new DatabaseSync(":memory:") const pending: Array> = [] + const alarms: Array = [] const sql = { exec: (statement: string, ...bindings: ReadonlyArray) => { @@ -43,7 +49,13 @@ export const makeFakeDurableObjectState = (): FakeDurableObjectState => { } return { - storage: { sql: sql as SqlStorage }, + storage: { + sql: sql as SqlStorage, + setAlarm: async (scheduledTime) => { + alarms.push(scheduledTime) + }, + }, + alarms, waitUntil: (promise) => { // Swallow rejections here the way the runtime does; a test that cares awaits `pending`. pending.push(promise.catch(() => undefined))