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