diff --git a/.changeset/turn-inactivity-watchdog.md b/.changeset/turn-inactivity-watchdog.md new file mode 100644 index 000000000000..b194f283737a --- /dev/null +++ b/.changeset/turn-inactivity-watchdog.md @@ -0,0 +1,5 @@ +--- +"@reddb-io/redcode": patch +--- + +End a turn that has stopped producing anything, where nobody is watching to end it themselves. Time a tool spends running or a permission spends awaiting an answer does not count as silence, so a long build is never mistaken for a provider that went away. In the TUI and the desktop app the turn is reported rather than ended, since a person is there to read it and press escape; a scripted run, an editor speaking ACP or a scheduled job ends it. Configurable through `experimental.turn_stall`, or `false` to disable. diff --git a/packages/core/src/v1/config/config.ts b/packages/core/src/v1/config/config.ts index bf5e01a9b6d1..035cbc7df210 100644 --- a/packages/core/src/v1/config/config.ts +++ b/packages/core/src/v1/config/config.ts @@ -182,6 +182,22 @@ export const Info = Schema.Struct({ mcp_timeout: Schema.optional(PositiveInt).annotate({ description: "Timeout in milliseconds for model context protocol (MCP) requests", }), + turn_stall: Schema.optional( + Schema.Union([ + Schema.Literal(false), + Schema.Struct({ + warn_ms: Schema.optional(PositiveInt).annotate({ + description: "Say the turn has gone quiet after this long (default: 300000)", + }), + abort_ms: Schema.optional(PositiveInt).annotate({ + description: "End a turn that has produced nothing for this long (default: 600000)", + }), + }), + ]), + ).annotate({ + description: + "How long a turn may produce nothing before it is reported and, where nothing is watching, ended. Time a tool spends running or a permission spends awaiting an answer does not count. Set to false to disable.", + }), policies: Schema.optional(Schema.mutable(Schema.Array(ConfigExperimental.Policy))).annotate({ description: "Policy statements applied to supported resources, such as provider access", }), diff --git a/packages/redcode/src/cli/cmd/tui.ts b/packages/redcode/src/cli/cmd/tui.ts index 8f8369fd4ae4..ba930e80525d 100644 --- a/packages/redcode/src/cli/cmd/tui.ts +++ b/packages/redcode/src/cli/cmd/tui.ts @@ -228,9 +228,15 @@ export const TuiThreadCommand = cmd({ let stopped = false const worker = new Worker(file, { - env: Object.fromEntries( - Object.entries(process.env).filter((entry): entry is [string, string] => entry[1] !== undefined), - ), + env: { + ...Object.fromEntries( + Object.entries(process.env).filter((entry): entry is [string, string] => entry[1] !== undefined), + ), + // The server runs inside this worker and otherwise cannot tell itself apart from a + // scripted run: both default to "cli". Anything that should behave differently when a + // person is watching — ending a stalled turn, for one — needs to know which this is. + REDCODE_CLIENT: "tui", + }, }) const client = Rpc.client(worker) // A worker that fails to load, or dies, posts nothing back. Without these the calls diff --git a/packages/redcode/src/session/prompt.ts b/packages/redcode/src/session/prompt.ts index 728ba3ee5517..2710d92d2090 100644 --- a/packages/redcode/src/session/prompt.ts +++ b/packages/redcode/src/session/prompt.ts @@ -23,8 +23,13 @@ import { MAX_STEPS_PROMPT } from "@reddb-io/redcode-core/session/runner/max-step // Agents that want a tighter bound set `agent.steps`. const TURN_STEP_CEILING = 200 -/** How long a turn may send nothing before each further interval is recorded. */ -const STALL_LOG_SECONDS = 60 +/** How often the watchdog looks. Well below the thresholds it is checking against. */ +const STALL_POLL_SECONDS = 15 + +/** Surfaces where a person is present to read a warning and stop the turn themselves. */ +function attendedClient(client: string) { + return client === "tui" || client === "app" || client === "desktop" +} import { ToolRegistry } from "@/tool/registry" import { MCP } from "../mcp" import { LSP } from "@/lsp/lsp" @@ -39,6 +44,7 @@ import { ConfigMarkdown } from "@/config/markdown" import { SessionSummary } from "./summary" import { NamedError } from "@reddb-io/redcode-core/util/error" import { SessionProcessor } from "./processor" +import { SessionStall } from "./stall" import { Tool } from "@/tool/tool" import { Permission } from "@/permission" import { SessionStatus } from "./status" @@ -1114,31 +1120,76 @@ const layer = Layer.effect( yield* hooks.parallel(OperationHook.Operation.Turn.Started, turnStarted) yield* events.publish(SessionEvent.Turn.Started, turnStarted).pipe(Effect.ignore) - // A turn that goes quiet leaves no trace today: no event, no log, and a spinner that - // looks exactly like progress. Saying so on a schedule is what turns "it hung for half an - // hour" into something anyone can look up afterwards. + // A turn that goes quiet used to leave no trace at all: no event, no log, and a spinner + // indistinguishable from progress. This watches for that, says so, and — where nobody is + // sitting in front of it — ends the turn rather than letting it burn. // // One fiber for the whole turn, reading whichever step's handle is current. Forked as a - // child of this fiber, so it dies when the turn does — `scope` above belongs to the - // service layer and outlives every turn, which would leave a poller behind per step. + // child of this fiber, so it dies when the turn does: `scope` above belongs to the service + // layer and outlives every turn, which would leave a poller behind per step. let watched: SessionProcessor.Handle | undefined + let warned = false yield* Effect.forkChild( - Effect.repeat( - Effect.suspend(() => { - const handle = watched - if (!handle) return Effect.void - const quiet = Math.round((Date.now() - handle.lastEventAt) / 1000) - if (quiet < STALL_LOG_SECONDS) return Effect.void - // Tools run inside the SDK and emit nothing while they work, so silence with one in - // flight is progress, not a stall. - if (handle.activeToolCount > 0) return Effect.void - return Effect.logWarning("turn has produced nothing", { - "session.id": sessionID, - messageID: handle.message.id, - quietSeconds: quiet, - }) - }), - Schedule.spaced(Duration.seconds(STALL_LOG_SECONDS)), + Effect.forever( + Effect.suspend(() => + Effect.gen(function* () { + const handle = watched + if (!handle) { + // Before the first step has a handle there is nothing to measure, and the + // configured cadence has not been read yet. Look again shortly rather than + // sleeping a full interval and missing the start of the turn. + yield* Effect.sleep(Duration.millis(SessionStall.POLL_MIN_MS)) + return + } + // Read on first use rather than before the loop: the turn's opening is a + // cancellation-sensitive stretch and this has no business being on it. + const limits = SessionStall.limits((yield* config.get()).experimental?.turn_stall, { + // A person watching can read the warning and press escape; a scripted run, an + // editor speaking ACP or a scheduled job cannot. + attended: attendedClient(flags.client), + }) + const pending = yield* permission.list() + const decision = SessionStall.decide({ + quietMs: Date.now() - handle.lastEventAt, + activeToolCount: handle.activeToolCount, + permissionPending: pending.some((item) => item.sessionID === sessionID), + limits, + }) + const nap = Effect.sleep(Duration.millis(SessionStall.pollMs(limits))) + if (decision.type === "working") { + warned = false + yield* nap + return + } + if (decision.type === "warn") { + // Said once per quiet stretch, not on every poll. + if (!warned) { + warned = true + yield* Effect.logWarning(SessionStall.warning(decision.quietMs, limits), { + "session.id": sessionID, + messageID: handle.message.id, + }) + } + yield* nap + return + } + yield* Effect.logWarning("ending a turn that stopped producing output", { + "session.id": sessionID, + messageID: handle.message.id, + reason: decision.reason, + }) + // The reason has to be written before the interrupt lands: every later writer on + // the abort path guards with `??=`, so whoever gets there first decides what the + // message says, and otherwise this reads as an ordinary user interrupt. + handle.message.error ??= new SessionV1.AbortedError({ + message: `stopped: ${decision.reason}`, + }).toObject() + yield* sessions.updateMessage(handle.message).pipe(Effect.ignore) + // Detached deliberately: cancel interrupts this very fiber partway through, and + // the part that returns the session to idle runs after that point. + yield* state.cancel(sessionID).pipe(Effect.ignore, Effect.forkIn(scope)) + }), + ), ), ) diff --git a/packages/redcode/src/session/stall.ts b/packages/redcode/src/session/stall.ts new file mode 100644 index 000000000000..bb2c9503c483 --- /dev/null +++ b/packages/redcode/src/session/stall.ts @@ -0,0 +1,100 @@ +/** + * When a turn has stopped making progress, and what to do about it. + * + * Kept pure and separate from the loop so every branch is testable without timers, fibers or a + * provider. The loop supplies the observations; this decides. + * + * The distinction that matters: silence is not the same as a stall. Tools run inside the provider + * SDK and emit nothing at all while they work, so a turn running `cargo build` for half an hour is + * silent and healthy. Only silence with nothing in flight is a stall. + */ + +export type StallLimits = { + /** Say something once the turn has been quiet this long. `Infinity` disables it. */ + readonly warnMs: number + /** End the turn once it has been quiet this long. `Infinity` disables it. */ + readonly abortMs: number +} + +export type StallInput = { + readonly quietMs: number + readonly activeToolCount: number + readonly permissionPending: boolean + readonly limits: StallLimits +} + +export type StallDecision = + | { readonly type: "working" } + | { readonly type: "warn"; readonly quietMs: number } + | { readonly type: "abort"; readonly quietMs: number; readonly reason: string } + +/** Well clear of a slow model's thinking time, well inside the timeouts that eventually fire. */ +export const STALL_WARN_MS_DEFAULT = 300_000 +export const STALL_ABORT_MS_DEFAULT = 600_000 + +export function decide(input: StallInput): StallDecision { + // A person deciding whether to approve a command is not a stalled turn. + if (input.permissionPending) return { type: "working" } + // Neither is a tool that is still running, however long it takes. + if (input.activeToolCount > 0) return { type: "working" } + if (input.quietMs >= input.limits.abortMs) { + return { type: "abort", quietMs: input.quietMs, reason: `no output for ${describe(input.quietMs)}` } + } + if (input.quietMs >= input.limits.warnMs) return { type: "warn", quietMs: input.quietMs } + return { type: "working" } +} + +/** Rounded the way someone reading a status line would say it, not to the millisecond. */ +export function describe(ms: number) { + const seconds = Math.round(ms / 1000) + if (seconds < 60) return `${seconds}s` + const minutes = Math.round(seconds / 60) + if (minutes < 60) return `${minutes}m` + const hours = Math.floor(minutes / 60) + const rest = minutes % 60 + return rest > 0 ? `${hours}h ${rest}m` : `${hours}h` +} + +/** + * A warning has to leave room for the turn to recover: activity can still arrive and make the + * abort never happen, so it must not be worded as though it already had. + */ +export function warning(quietMs: number, limits: StallLimits) { + const tail = Number.isFinite(limits.abortMs) ? `, ending it at ${describe(limits.abortMs)} unless it resumes` : "" + return `No output for ${describe(quietMs)}${tail}` +} + +/** + * Turn configuration into limits, and decide where ending a turn is appropriate at all. + * + * Ending someone's turn while they are sitting in front of it takes a decision away from them; + * the warning is enough there, because they can see it and press escape. Where nothing is + * watching — a scripted run, an editor speaking ACP, a scheduled job — nobody will ever press + * anything, so silence is the whole failure and ending it is the only useful act. + */ +export function limits( + config: false | { readonly warn_ms?: number; readonly abort_ms?: number } | undefined, + options: { readonly attended?: boolean } = {}, +): StallLimits { + if (config === false) return { warnMs: Infinity, abortMs: Infinity } + const warnMs = config?.warn_ms ?? STALL_WARN_MS_DEFAULT + const configured = config?.abort_ms ?? STALL_ABORT_MS_DEFAULT + const abortMs = options.attended ? Infinity : configured + // A warning that arrives after the abort would never be seen. + return { warnMs: Math.min(warnMs, abortMs), abortMs } +} + +/** + * How often to look. Frequent enough that a threshold is honoured rather than rounded up to the + * next poll, cheap enough that a healthy turn costs nothing: half the nearest threshold, capped. + */ +export function pollMs(limits: StallLimits) { + const nearest = Math.min(limits.warnMs, limits.abortMs) + if (!Number.isFinite(nearest)) return POLL_MAX_MS + return Math.max(POLL_MIN_MS, Math.min(POLL_MAX_MS, Math.round(nearest / 2))) +} + +export const POLL_MIN_MS = 100 +const POLL_MAX_MS = 15_000 + +export * as SessionStall from "./stall" diff --git a/packages/redcode/test/lib/llm-server.ts b/packages/redcode/test/lib/llm-server.ts index 245acc7280f5..c1a90b2d53c1 100644 --- a/packages/redcode/test/lib/llm-server.ts +++ b/packages/redcode/test/lib/llm-server.ts @@ -493,6 +493,15 @@ export class Reply { return this } + /** An arbitrary finish reason, including ones the schema does not recognise. */ + finish(reason: string) { + this.#finish = reason + this.#hang = false + this.#error = undefined + this.#reset = false + return this + } + contentFilter() { this.#finish = "content_filter" this.#hang = false diff --git a/packages/redcode/test/session/prompt.test.ts b/packages/redcode/test/session/prompt.test.ts index 3f6a6352b8c1..298bda9a022c 100644 --- a/packages/redcode/test/session/prompt.test.ts +++ b/packages/redcode/test/session/prompt.test.ts @@ -1339,6 +1339,84 @@ it.instance("cancel interrupts loop and resolves with an assistant message", () }), ) +it.instance("ends a turn whose provider goes quiet, and says so on the message", () => + Effect.gen(function* () { + // Thresholds in milliseconds so the test runs in a second rather than ten minutes. The + // watchdog polls on a fixed cadence, so this waits for that rather than for the threshold. + const { llm } = yield* useServerConfig((url) => ({ + ...providerCfg(url), + experimental: { turn_stall: { warn_ms: 1, abort_ms: 2 } }, + })) + const prompt = yield* SessionPrompt.Service + const sessions = yield* Session.Service + const chat = yield* sessions.create({ title: "Pinned" }) + yield* seed(chat.id) + + yield* llm.hang + yield* user(chat.id, "more") + + const fiber = yield* prompt.loop({ sessionID: chat.id }).pipe(Effect.forkChild) + yield* llm.wait(1) + yield* waitForBusy(chat.id) + + // No cancel of our own: the watchdog is the only thing that can end this. + const exit = yield* awaitWithTimeout(Fiber.await(fiber), "watchdog never ended the stalled turn", "40 seconds") + expect(Exit.isSuccess(exit)).toBe(true) + + const messages = yield* sessions.messages({ sessionID: chat.id }) + const assistant = messages.findLast( + (item): item is (typeof messages)[number] & { info: SessionV1.Assistant } => item.info.role === "assistant", + ) + expect(assistant?.info.error?.name).toBe("MessageAbortedError") + // The reason is what keeps this from reading as though the user pressed escape. + expect((assistant?.info.error?.data as { message?: string } | undefined)?.message).toMatch(/^stopped: no output/) + }), +) + +it.instance("leaves a turn alone while a tool is still running", () => + Effect.gen(function* () { + // The case that protects real work: a tool runs inside the provider SDK and emits nothing + // while it works, so a long command looks exactly like a provider that has gone away. + // Short enough for a test, long enough that the gap before the provider's first byte is not + // itself read as a stall. + const { llm } = yield* useServerConfig((url) => ({ + ...providerCfg(url), + experimental: { turn_stall: { warn_ms: 500, abort_ms: 1500 } }, + })) + const registry = yield* ToolRegistry.Service + const { read } = yield* registry.named() + const { ready, restore } = yield* hangUntilAborted(read) + yield* restore + + const prompt = yield* SessionPrompt.Service + const sessions = yield* Session.Service + const status = yield* SessionStatus.Service + const chat = yield* sessions.create({ title: "Pinned" }) + yield* seed(chat.id) + + yield* llm.tool("read", { filePath: "/tmp/whatever" }) + yield* user(chat.id, "more") + + const fiber = yield* prompt.loop({ sessionID: chat.id }).pipe(Effect.forkChild) + yield* awaitWithTimeout(llm.wait(1), "provider was never called", "10 seconds") + yield* awaitWithTimeout(Deferred.await(ready), "timed out waiting for the tool to start", "10 seconds") + + // Several times the abort threshold, with the tool still running throughout. + yield* Effect.sleep("5 seconds") + expect((yield* status.get(chat.id)).type).toBe("busy") + // The discriminating assertion: had the watchdog fired it would have stamped its reason on + // the message before interrupting. + const during = yield* sessions.messages({ sessionID: chat.id }) + const running = during.findLast( + (item): item is (typeof during)[number] & { info: SessionV1.Assistant } => item.info.role === "assistant", + ) + expect(running?.info.error).toBeUndefined() + + yield* prompt.cancel(chat.id) + yield* Fiber.await(fiber) + }), +) + it.instance("cancel records MessageAbortedError on interrupted process", () => Effect.gen(function* () { const { llm } = yield* useServerConfig(providerCfg) diff --git a/packages/redcode/test/session/stall.test.ts b/packages/redcode/test/session/stall.test.ts new file mode 100644 index 000000000000..fd73588108c7 --- /dev/null +++ b/packages/redcode/test/session/stall.test.ts @@ -0,0 +1,60 @@ +import { describe as group, expect, test } from "bun:test" +import { decide, describe, warning, STALL_ABORT_MS_DEFAULT, STALL_WARN_MS_DEFAULT } from "@/session/stall" + +const limits = { warnMs: STALL_WARN_MS_DEFAULT, abortMs: STALL_ABORT_MS_DEFAULT } +const quiet = (ms: number, over: Partial[0]> = {}) => + decide({ quietMs: ms, activeToolCount: 0, permissionPending: false, limits, ...over }) + +group("stall.decide", () => { + test("a turn that is still producing is working", () => { + expect(quiet(0).type).toBe("working") + expect(quiet(STALL_WARN_MS_DEFAULT - 1).type).toBe("working") + }) + + test("warns once it has been quiet, and ends it once it has been quiet far longer", () => { + expect(quiet(STALL_WARN_MS_DEFAULT)).toMatchObject({ type: "warn" }) + expect(quiet(STALL_ABORT_MS_DEFAULT)).toMatchObject({ type: "abort" }) + }) + + test("a running tool is work, however long it is silent", () => { + // The case that matters: tools run inside the provider SDK and emit nothing while they work, + // so a half-hour build must never be mistaken for a hang. + expect(quiet(STALL_ABORT_MS_DEFAULT * 10, { activeToolCount: 1 }).type).toBe("working") + }) + + test("a person deciding on a permission is not a stalled turn", () => { + expect(quiet(STALL_ABORT_MS_DEFAULT * 10, { permissionPending: true }).type).toBe("working") + }) + + test("infinite limits disable each stage independently", () => { + expect(quiet(10_000_000, { limits: { warnMs: Infinity, abortMs: Infinity } }).type).toBe("working") + expect(quiet(10_000_000, { limits: { warnMs: 1, abortMs: Infinity } }).type).toBe("warn") + }) + + test("the abort reason says how long it was silent", () => { + const decision = quiet(600_000) + expect(decision.type).toBe("abort") + if (decision.type === "abort") expect(decision.reason).toBe("no output for 10m") + }) +}) + +group("stall.describe", () => { + test("reads the way someone would say it", () => { + expect(describe(9_000)).toBe("9s") + expect(describe(600_000)).toBe("10m") + expect(describe(3_600_000)).toBe("1h") + expect(describe(5_400_000)).toBe("1h 30m") + }) +}) + +group("stall.warning", () => { + test("leaves room for the turn to recover instead of announcing an abort", () => { + const text = warning(300_000, limits) + expect(text).toBe("No output for 5m, ending it at 10m unless it resumes") + expect(text).not.toContain("aborted") + }) + + test("says nothing about ending when ending is disabled", () => { + expect(warning(300_000, { warnMs: 1, abortMs: Infinity })).toBe("No output for 5m") + }) +})