diff --git a/apps/ai/src/chat/budgets.ts b/apps/ai/src/chat/budgets.ts index 04ccf948ad..32a405b1c2 100644 --- a/apps/ai/src/chat/budgets.ts +++ b/apps/ai/src/chat/budgets.ts @@ -31,15 +31,16 @@ export const TOOL_CONCURRENCY = 4 /** * Consecutive tool-call *failures* that mean the model is stuck rather than working. * - * Three, because two is a plausible retry — a model reissuing a call after a transient tool failure - * is behaving correctly — and the third consecutive failure says nothing in the results is being - * read. It becomes the policy's `repeatedFailureLimit`, and a success resets the count. + * Five, because tool errors are returned to the model: rewriting a rejected query two or three times + * is the model working, and one batch can fail `TOOL_CONCURRENCY` calls at once. Five in a row says + * nothing in the results is being read. It becomes the policy's `repeatedFailureLimit`, and a + * success resets the count. * * Note what it does *not* bound: identical calls that keep succeeding. That guard lives with the * tool handlers, as `IDENTICAL_CALL_LIMIT` in `@/mcp/tools/llm-tools`, because refusing there is a - * declared failure the model can read — and three of those trip this limit, which stops the run. + * returned failure the model can read, and enough of those trip this limit, which stops the run. */ -export const REPEATED_TOOL_CALLS = 3 +export const REPEATED_TOOL_CALLS = 5 /** * Wall clock one run may take. diff --git a/apps/ai/src/chat/events.test.ts b/apps/ai/src/chat/events.test.ts index e8e50b9e10..d5103a31a6 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 e0242d90d3..90342b613b 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-tool-failures.test.ts b/apps/ai/src/chat/run-tool-failures.test.ts new file mode 100644 index 0000000000..2e7d349309 --- /dev/null +++ b/apps/ai/src/chat/run-tool-failures.test.ts @@ -0,0 +1,136 @@ +/** + * What a failed tool call does to a whole run, through the real engine and a scripted model. + * + * The handler tests pin each tool's `failureMode`; this pins what the engine does with it. A rejected + * query must reach the model as a failed result it can rewrite, and a gated proposal must still end + * the run. + */ +import { OrgId, UserId } from "@maple/domain" +import { Effect, Exit, Layer, Schema, Stream } from "effect" +import { LanguageModel } from "effect/unstable/ai" +import type { Prompt, Response } from "effect/unstable/ai" +import * as AiModel from "effect/unstable/ai/Model" +import { assert, describe, it } from "vitest" +import type { TenantContext } from "@maple/backend/services/auth/tenant-context" +import type { McpToolExecutorApi } from "../mcp/dispatcher" +import type { ResolvedModel } from "../platform/Llm" +import type { ChatTurnEvent } from "./events" +import { runChatTurn } from "./run" +import { makeRunUsage } from "./tools" + +const TENANT: TenantContext = { + orgId: Schema.decodeSync(OrgId)("org_test"), + userId: Schema.decodeSync(UserId)("user_test"), + roles: [], + authMode: "self_hosted", +} + +const USAGE = { inputTokens: { total: 1, uncached: 1 }, outputTokens: { total: 1, text: 1 } } + +const callTool = (name: string, params: unknown): ReadonlyArray => [ + { type: "tool-call", id: "call-1", name, params }, + { type: "finish", reason: "tool-calls", usage: USAGE }, +] + +const answer = (text: string): ReadonlyArray => [ + { type: "text-start", id: "text-1" }, + { type: "text-delta", id: "text-1", delta: text }, + { type: "text-end", id: "text-1" }, + { type: "finish", reason: "stop", usage: USAGE }, +] + +/** A model that plays `script` one response per call, recording every prompt it was sent. */ +const scriptedModel = (script: ReadonlyArray>) => { + const prompts: Array = [] + const respond = (options: LanguageModel.ProviderOptions) => { + prompts.push(options.prompt) + return script[prompts.length - 1] ?? answer("(script ran out)") + } + const model: ResolvedModel = { + provider: "openrouter", + name: "scripted", + limits: { context: 100_000, output: 4_000 }, + layer: AiModel.make( + "openrouter", + "scripted", + Layer.effect( + LanguageModel.LanguageModel, + LanguageModel.make({ + generateText: (options) => Effect.succeed([...respond(options)]), + streamText: (options) => Stream.suspend(() => Stream.fromIterable(respond(options))), + }), + ), + ), + } + return { model, prompts } +} + +const run = (model: ResolvedModel, executor: McpToolExecutorApi) => { + const events: Array = [] + const effect = runChatTurn({ + sessionId: "org_test:tab-1", + messageId: "msg-1", + tenant: TENANT, + toolExecutor: executor, + model, + submitDiagnosis: () => Effect.die("no investigation in a default session"), + text: "how many services?", + history: [], + usage: makeRunUsage(), + holdsTurn: () => true, + append: (event) => events.push(event), + }) + return { effect, events } +} + +const rejectingExecutor: McpToolExecutorApi = { + execute: () => + Effect.succeed({ + isError: true, + content: [{ type: "text" as const, text: "SQL rejected (MissingOrgFilter)" }], + }), +} + +describe("runChatTurn tool failures", () => { + it("hands a rejected query back to the model and finishes the run", async () => { + const { model, prompts } = scriptedModel([ + callTool("run_sql", { sql: "select 1" }), + answer("Rewrote it."), + ]) + const { effect, events } = run(model, rejectingExecutor) + + const exit = await Effect.runPromiseExit(effect) + + assert.isTrue(Exit.isSuccess(exit), `run failed: ${String(exit)}`) + assert.lengthOf(prompts, 2, "the model is called again after the failed tool") + const results = prompts[1]!.content.flatMap((message) => + message.role === "tool" ? message.content : [], + ) + assert.deepInclude( + results.map( + (part) => part.type === "tool-result" && { name: part.name, isFailure: part.isFailure }, + ), + { name: "run_sql", isFailure: true }, + ) + assert.deepInclude( + events.map( + (event) => event.type === "tool-result" && { output: event.output, isError: event.isError }, + ), + { output: "SQL rejected (MissingOrgFilter)", isError: true }, + ) + assert.deepEqual(events.at(-1), { type: "turn-end", messageId: "msg-1", reason: "stop" }) + }) + + it("still ends the run on a gated proposal", async () => { + const { model, prompts } = scriptedModel([ + callTool("create_dashboard", { name: "Checkout" }), + answer("should never be asked"), + ]) + const { effect } = run(model, rejectingExecutor) + + const exit = await Effect.runPromiseExit(effect) + + assert.isTrue(Exit.isFailure(exit)) + assert.lengthOf(prompts, 1, "a proposal is the turn's last word") + }) +}) diff --git a/apps/ai/src/chat/run.ts b/apps/ai/src/chat/run.ts index d382294ebe..ba15f3aa1d 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 8a343965ae..226752b562 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 f3744b6740..73f5a028c4 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 fbfc72e60f..b5f9c8219a 100644 --- a/apps/ai/src/mcp/tools/llm-tools.test.ts +++ b/apps/ai/src/mcp/tools/llm-tools.test.ts @@ -3,7 +3,7 @@ * * The old turn loop stopped a model that reissued an identical tool batch until its turns ran out. * The engine has no equivalent — its `repeatedFailureLimit` counts consecutive tool *failures*, - * which is a different thing — so the guard lives with the handlers, where refusing is a declared + * which is a different thing — so the guard lives with the handlers, where refusing is a returned * tool failure the model can read rather than an authorization denial that ends the turn. */ import { OrgId, UserId } from "@maple/domain" @@ -40,6 +40,52 @@ const handlerFor = (executor: McpToolExecutorApi, name: string) => { } describe("buildMapleToolkit", () => { + /** + * A tool error is the call's answer, not the run's end: `"return"` hands it to the model. Only a + * gated tool propagates, because a proposal must be the turn's last word. + */ + it("returns ordinary tool failures to the model and propagates a gated one", () => { + const { executor } = countingExecutor() + const { toolkit } = buildMapleToolkit(executor, TENANT, { + include: (name) => name === "list_services" || name === "create_dashboard", + gate: (name) => name === "create_dashboard", + }) + + assert.equal(toolkit.tools.list_services?.failureMode, "return") + assert.equal(toolkit.tools.create_dashboard?.failureMode, "error") + }) + + it("fails a tool that reported an error with its message", 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.isFailure(result)) + assert.equal( + Result.isFailure(result) ? result.failure.message : "", + "Tool failed: SQL rejected (MissingOrgFilter)", + ) + }) + + it("fails a tool that died with a summary, never the cause", 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.isFailure(result)) + assert.include(Result.isFailure(result) ? result.failure.message : "", "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 +112,10 @@ describe("buildMapleToolkit", () => { assert.isTrue(Result.isSuccess(result), `attempt ${attempt + 1} should have run`) } - assert.isTrue(Result.isFailure(await call())) + // A returned failure: the model repeating itself is told, and the policy stops it if it persists. + const fourth = await call() + assert.isTrue(Result.isFailure(fourth)) + assert.include(Result.isFailure(fourth) ? fourth.failure.message : "", "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 801afab09a..5dfb712ba5 100644 --- a/apps/ai/src/mcp/tools/llm-tools.ts +++ b/apps/ai/src/mcp/tools/llm-tools.ts @@ -84,8 +84,9 @@ export interface BuildMapleToolsOptions { /** * A tool's failure as the model sees it: one line of text, never a cause. * - * Declared rather than thrown so the runtime records a tool failure the model can route around, - * instead of the failure ending the run. + * Where it goes depends on the tool's `failureMode` (see `buildMapleToolkit`): an ordinary tool + * returns it to the model, which rewrites the call; a gated tool propagates it and ends the run, + * which is how a proposal becomes the turn's last word and reaches the approval card. */ export class MapleToolFailure extends Schema.TaggedError()( "@maple/api/mcp/MapleToolFailure", @@ -103,7 +104,7 @@ const fail = (message: string) => Effect.fail(new MapleToolFailure({ message })) * it already has, and left alone it will spend every turn it owns doing that. * * It refuses rather than denying authorization, because the two end differently: a refusal is a - * declared tool failure the model can read and route around, and a third consecutive one trips the + * returned tool failure the model can read and route around, and enough consecutive ones trip the * policy's own `repeatedFailureLimit`, which stops the run. A host authorization denial ends the * run outright, and a user watching a chat turn would see an error instead of an answer. */ @@ -146,6 +147,8 @@ export const buildMapleToolkit = ( parameters: toInputSchema(definition.schema), success: Schema.String, failure: MapleToolFailure, + // A proposal must end the run; any other failure goes back to the model as the call's result. + failureMode: gated ? "error" : "return", }) }) const toolkit = Toolkit.make(...tools) @@ -157,14 +160,14 @@ export const buildMapleToolkit = ( const gated = options.gate?.(definition.name) ?? false const dispatch = (params: unknown) => executor.execute(tenant, definition.name, params, options.surface ?? "chat").pipe( + // A tool that dies (unknown tool, tenant error) fails like one that reported an error. + // Caught before the `flatMap`, so a reported error is not wrapped a second time. + Effect.catchCause((cause) => fail(`Tool failed: ${summarizeToolFailure(cause)}`)), Effect.flatMap((result) => result.isError ? fail(toolResultText(result)) : Effect.succeed(toolResultText(result)), ), - // 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.`) diff --git a/apps/ai/src/platform/genai-spans.ts b/apps/ai/src/platform/genai-spans.ts index 050e1483f1..46b9e628f8 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 ed4bc6abbf..619063c3d4 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 9db05c1e89..ac966c14a8 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