diff --git a/.changeset/aux-deadlines.md b/.changeset/aux-deadlines.md new file mode 100644 index 000000000000..429c0be931a6 --- /dev/null +++ b/.changeset/aux-deadlines.md @@ -0,0 +1,8 @@ +--- +"@reddb-io/redcode": patch +"@reddb-io/redcode-core": patch +--- + +Bound the model calls a turn makes that are not the turn itself + +Naming a session and compacting the conversation both call a provider outside the step loop, where the turn's inactivity watchdog cannot see them: one runs before any step handle exists, the other creates a processor of its own. A provider that stopped answering during either held the turn open with nothing on screen and no error. Both now give up — naming after two minutes, compacting after ten — and say so. A session keeping its default name is a far smaller loss than a turn that never starts. Configurable via `experimental.aux_timeout`. diff --git a/packages/core/src/v1/config/config.ts b/packages/core/src/v1/config/config.ts index bd364b765197..6ee4d533937a 100644 --- a/packages/core/src/v1/config/config.ts +++ b/packages/core/src/v1/config/config.ts @@ -182,6 +182,10 @@ export const Info = Schema.Struct({ mcp_timeout: Schema.optional(PositiveInt).annotate({ description: "Timeout in milliseconds for model context protocol (MCP) requests", }), + aux_timeout: Schema.optional(Schema.Union([Schema.Literal(false), PositiveInt])).annotate({ + description: + "Milliseconds the calls around a turn - naming the session, compacting the conversation - may wait for a provider before being given up on (defaults: 120000 and 600000). Set to false to remove the bound.", + }), turn_steps: Schema.optional( Schema.Union([ Schema.Literal(false), diff --git a/packages/redcode/src/session/aux-deadline.ts b/packages/redcode/src/session/aux-deadline.ts new file mode 100644 index 000000000000..c42505ca1ded --- /dev/null +++ b/packages/redcode/src/session/aux-deadline.ts @@ -0,0 +1,34 @@ +/** + * Bounds for the model calls a turn makes that are not the turn itself. + * + * Naming the session, and compacting it when the context fills, both call a provider outside the + * step loop. Neither is covered by the turn's inactivity watchdog — the watchdog reads a step + * handle, and these either have none or have one of their own — so a provider that stops answering + * during either of them holds the turn open with nothing on screen and no error. + * + * Neither is the work the user asked for, so both can fail without the turn failing: a session + * keeps its default name, and a compaction that did not happen is reported as itself. + */ + +/** Naming a session is one short request against a small model. */ +export const TITLE_MS = 120_000 + +/** Compacting reads the whole conversation back, so it is allowed to take real time. */ +export const COMPACTION_MS = 600_000 + +export type Call = "title" | "compaction" + +const DEFAULTS: Record = { title: TITLE_MS, compaction: COMPACTION_MS } + +export function deadlineMs(call: Call, configured?: number | false): number | undefined { + if (configured === false) return undefined + if (configured === undefined) return DEFAULTS[call] + return configured > 0 ? configured : undefined +} + +export function message(call: Call, ms: number) { + const what = call === "title" ? "Naming the session" : "Compacting the conversation" + return `${what} got no answer from the provider within ${Math.round(ms / 1000)}s and was given up on.` +} + +export * as AuxDeadline from "./aux-deadline" diff --git a/packages/redcode/src/session/compaction.ts b/packages/redcode/src/session/compaction.ts index 8635a6b99432..b5a542cb8cd4 100644 --- a/packages/redcode/src/session/compaction.ts +++ b/packages/redcode/src/session/compaction.ts @@ -7,13 +7,14 @@ import { Provider } from "@/provider/provider" import { MessageV2 } from "./message-v2" import { Token } from "@/util/token" import { SessionProcessor } from "./processor" +import { AuxDeadline } from "./aux-deadline" import { Agent } from "@/agent/agent" import { SessionEvent } from "@reddb-io/redcode-core/session/event" import { Plugin } from "@/plugin" import { Config } from "@/config/config" import { NotFoundError } from "@/storage/storage" -import { DateTime, Effect, Layer, Context } from "effect" +import { DateTime, Duration, Effect, Layer, Context } from "effect" import { InstanceState } from "@/effect/instance-state" import { isOverflow as overflow, usable } from "./overflow" import { serviceUse } from "@reddb-io/redcode-core/effect/service-use" @@ -450,6 +451,10 @@ const layer = Layer.effect( sessionID: input.sessionID, model, }) + // The turn's watchdog reads the step handle, and this processor is not it, so a provider + // that stops answering here holds the turn open with nothing to show. A compaction that did + // not happen is reported as itself rather than as silence. + const compactionMs = AuxDeadline.deadlineMs("compaction", (yield* config.get()).experimental?.aux_timeout) const result = yield* processor.process({ user: userMessage, agent, @@ -473,7 +478,25 @@ const layer = Layer.effect( }, ], model, - }) + }).pipe( + compactionMs === undefined + ? (self) => self + : Effect.timeoutOrElse({ + duration: Duration.millis(compactionMs), + orElse: () => + Effect.gen(function* () { + yield* Effect.logWarning(AuxDeadline.message("compaction", compactionMs), { + "session.id": input.sessionID, + }) + processor.message.error = new SessionV1.ContextOverflowError({ + message: AuxDeadline.message("compaction", compactionMs), + }).toObject() + processor.message.finish = "error" + yield* session.updateMessage(processor.message) + return "stop" as const + }), + }), + ) if (result === "compact") { processor.message.error = new SessionV1.ContextOverflowError({ diff --git a/packages/redcode/src/session/prompt.ts b/packages/redcode/src/session/prompt.ts index 2f06badd9584..a22e41a727d1 100644 --- a/packages/redcode/src/session/prompt.ts +++ b/packages/redcode/src/session/prompt.ts @@ -41,6 +41,7 @@ import { SessionSummary } from "./summary" import { NamedError } from "@reddb-io/redcode-core/util/error" import { SessionProcessor } from "./processor" import { StepBudget } from "./step-budget" +import { AuxDeadline } from "./aux-deadline" import { SessionStall } from "./stall" import { Tool } from "@/tool/tool" import { Permission } from "@/permission" @@ -239,6 +240,7 @@ const layer = Layer.effect( const msgs = onlySubtasks ? [{ role: "user" as const, content: subtasks.map((p) => p.prompt).join("\n") }] : yield* MessageV2.toModelMessagesEffect(context, mdl) + const titleMs = AuxDeadline.deadlineMs("title", (yield* config.get()).experimental?.aux_timeout) const text = yield* llm .stream({ agent: ag, @@ -256,6 +258,18 @@ const layer = Layer.effect( Stream.map((e) => e.text), Stream.mkString, Effect.orDie, + // Naming the session happens inside the turn loop, so a small model that stops answering + // holds up the work the user actually asked for. A session keeping its default name is a + // far smaller loss than a turn that never starts. + titleMs === undefined + ? (self) => self + : Effect.timeoutOrElse({ + duration: Duration.millis(titleMs), + orElse: () => + Effect.logWarning(AuxDeadline.message("title", titleMs), { + "session.id": input.session.id, + }).pipe(Effect.as("")), + }), ) const cleaned = text .replace(/[\s\S]*?<\/think>\s*/g, "") diff --git a/packages/redcode/test/lib/llm-server.ts b/packages/redcode/test/lib/llm-server.ts index c1a90b2d53c1..1c288ecf9294 100644 --- a/packages/redcode/test/lib/llm-server.ts +++ b/packages/redcode/test/lib/llm-server.ts @@ -632,6 +632,8 @@ namespace TestLLMServer { readonly fail: (message?: unknown) => Effect.Effect readonly error: (status: number, body: unknown) => Effect.Effect readonly hang: Effect.Effect + /** Answer every "name this session" request with silence, the way a wedged small model does. */ + readonly hangTitles: Effect.Effect readonly hold: (value: string, wait: PromiseLike) => Effect.Effect readonly reset: Effect.Effect readonly hits: Effect.Effect @@ -651,6 +653,7 @@ export class TestLLMServer extends Context.Service { + titlesHang = true + }), toolHang: Effect.fn("TestLLMServer.toolHang")(function* (name: string, input: unknown) { queue(reply().pendingTool(name, input).hang().item()) }), @@ -767,6 +775,7 @@ export class TestLLMServer extends Context.Service { hits = [] + titlesHang = false list = [] waits = [] misses = [] diff --git a/packages/redcode/test/session/aux-deadline.test.ts b/packages/redcode/test/session/aux-deadline.test.ts new file mode 100644 index 000000000000..878a812102d0 --- /dev/null +++ b/packages/redcode/test/session/aux-deadline.test.ts @@ -0,0 +1,22 @@ +import { describe, expect, test } from "bun:test" +import { COMPACTION_MS, deadlineMs, message, TITLE_MS } from "@/session/aux-deadline" + +describe("deadlines for the calls around a turn", () => { + test("bounds both, and gives compacting the longer rope", () => { + // Naming is one short request; compacting reads the whole conversation back. + expect(deadlineMs("title")).toBe(TITLE_MS) + expect(deadlineMs("compaction")).toBe(COMPACTION_MS) + expect(COMPACTION_MS).toBeGreaterThan(TITLE_MS) + }) + + test("configuration overrides, and false or zero removes the bound", () => { + expect(deadlineMs("title", 5_000)).toBe(5_000) + expect(deadlineMs("title", false)).toBeUndefined() + expect(deadlineMs("compaction", 0)).toBeUndefined() + }) + + test("says which call gave up, and for how long it waited", () => { + expect(message("title", 120_000)).toContain("Naming the session") + expect(message("compaction", 600_000)).toContain("600s") + }) +}) diff --git a/packages/redcode/test/session/prompt.test.ts b/packages/redcode/test/session/prompt.test.ts index a0eae437d2fb..b061bfaf8065 100644 --- a/packages/redcode/test/session/prompt.test.ts +++ b/packages/redcode/test/session/prompt.test.ts @@ -1531,6 +1531,39 @@ it.instance("corrects a model that repeats itself, then ends the turn if nothing 60_000, ) +it.instance("does not let naming the session hold up the turn", () => + Effect.gen(function* () { + // Naming happens inside the turn loop against a small model, and it is not covered by the + // turn's watchdog, so a provider that stops answering there used to hold up the work the user + // actually asked for with nothing on screen. + const { llm } = yield* useServerConfig((url) => ({ + ...providerCfg(url), + experimental: { aux_timeout: 500 }, + })) + const prompt = yield* SessionPrompt.Service + const sessions = yield* Session.Service + // The name is only generated for a session still carrying its default one. + const title = `New session - ${new Date().toISOString()}` + const chat = yield* sessions.create({ title }) + + yield* llm.hangTitles + yield* prompt.prompt({ + sessionID: chat.id, + agent: "build", + noReply: true, + parts: [{ type: "text", text: "say something" }], + }) + yield* llm.text("done") + + const result = yield* awaitWithTimeout(prompt.loop({ sessionID: chat.id }), "the turn never finished", "20 seconds") + + // The turn produced its answer; only the name was given up on. + expect(result.parts).toContainEqual(expect.objectContaining({ type: "text", text: "done" })) + expect((yield* sessions.get(chat.id)).title).toBe(title) + }), + 60_000, +) + it.instance("cancel records MessageAbortedError on interrupted process", () => Effect.gen(function* () { const { llm } = yield* useServerConfig(providerCfg) diff --git a/turbo.json b/turbo.json index 01c1e0702377..3137edf003f7 100644 --- a/turbo.json +++ b/turbo.json @@ -5,7 +5,10 @@ "tasks": { "typecheck": {}, "build": { - "dependsOn": [], + // Topological, not arbitrary: the CLI build bundles the app, which imports sources the SDK + // build generates. An empty `dependsOn` disabled that ordering, so the two raced and the + // loser read a file that did not exist yet. + "dependsOn": ["^build"], "outputs": ["dist/**"] }, "test": {