feat(session): end a turn that has stopped producing anything

.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.

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",
       }),

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<typeof rpc>(worker)
       // A worker that fails to load, or dies, posts nothing back. Without these the calls

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))
+              }),
+            ),
           ),
         )
 

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"

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

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)