fix(session): close turns left open by a process that died

.changeset/orphan-turns.md

@@ -0,0 +1,8 @@
+---
+"@reddb-io/redcode": patch
+"@reddb-io/redcode-app": patch
+---
+
+Close turns left open by a process that died, and stop calling them a queue
+
+`time.completed` on an assistant message is written by the process running the turn. Killed mid-turn — an OOM, a machine going to sleep — nobody writes it, and the message stays open for the rest of the session's life. The TUI reads an open assistant message as a turn in progress and stamps QUEUED on everything typed after it, across restarts, with nothing running: a session that survived one crash looks jammed forever. A fresh run now closes anything left behind by a run that is gone, records it, and the QUEUED badge requires the session to actually be busy.

packages/redcode/src/session/guard-log.ts

@@ -18,7 +18,7 @@ import type { SessionID } from "./schema"
  *
  * Writing must never be able to break a turn: a guard that cannot be recorded still acts.
  */
-export type Guard = "stall" | "tool_timeout" | "loop" | "steps" | "aux"
+export type Guard = "stall" | "tool_timeout" | "loop" | "steps" | "aux" | "orphan"
 export type Action = "warn" | "correct" | "stop"
 
 export interface Trip {

packages/redcode/src/session/orphan.ts

@@ -0,0 +1,23 @@
+import type { SessionV1 } from "@reddb-io/redcode-core/v1/session"
+
+/**
+ * Turns left behind by a process that died mid-flight.
+ *
+ * `time.completed` on an assistant message is written by the process running the turn. When that
+ * process is killed — an OOM, a machine going to sleep, a crash — nobody writes it, and the
+ * message stays open forever. The TUI reads an open assistant message as "a turn is in progress"
+ * and stamps QUEUED on everything typed after it, so a session that survived an OOM looks jammed
+ * from then on, across restarts, with no way for the user to tell it apart from a real queue.
+ *
+ * A turn interrupted while the process lives finalizes itself on the way out. So an open message
+ * found at the start of a fresh run belongs to a run that is no longer there.
+ */
+export const ORPHAN_MESSAGE = "The process ended before this turn finished."
+
+export function orphans(messages: readonly SessionV1.WithParts[]): SessionV1.Assistant[] {
+  return messages.flatMap((item) =>
+    item.info.role === "assistant" && !item.info.time.completed ? [item.info] : [],
+  )
+}
+
+export * as SessionOrphan from "./orphan"

packages/redcode/src/session/prompt.ts

@@ -43,6 +43,7 @@ import { SessionProcessor } from "./processor"
 import { StepBudget } from "./step-budget"
 import { AuxDeadline } from "./aux-deadline"
 import { SessionGuardLog } from "./guard-log"
+import { SessionOrphan } from "./orphan"
 import { SessionStall } from "./stall"
 import { Tool } from "@/tool/tool"
 import { Permission } from "@/permission"
@@ -1138,6 +1139,31 @@ const layer = Layer.effect(
         let todoContinuations = 0
         const session = yield* sessions.get(sessionID).pipe(Effect.orDie)
 
+        // Only one run exists per session at a time, so an assistant message still open here was
+        // left by a run that is gone — a process that died before it could close it. Left alone it
+        // reads as a turn in progress for the rest of the session's life, and everything typed
+        // after it is stamped QUEUED, across restarts, with nothing running.
+        const abandoned = SessionOrphan.orphans(
+          yield* MessageV2.filterCompactedEffect(sessionID).pipe(Effect.provideService(Database.Service, database)),
+        )
+        for (const message of abandoned) {
+          message.error ??= new SessionV1.AbortedError({ message: SessionOrphan.ORPHAN_MESSAGE }).toObject()
+          message.time.completed = Date.now()
+          yield* sessions.updateMessage(message).pipe(Effect.ignore)
+          yield* guards.record({
+            sessionID,
+            guard: "orphan",
+            action: "stop",
+            detail: SessionOrphan.ORPHAN_MESSAGE,
+          })
+        }
+        if (abandoned.length > 0) {
+          yield* Effect.logWarning("closed turns left behind by a process that died", {
+            "session.id": sessionID,
+            count: abandoned.length,
+          })
+        }
+
         // Turn lifecycle: fire `Turn.Started` once per turn (before any step).
         const turnStarted = { sessionID, timestamp: yield* DateTime.now }
         yield* hooks.parallel(OperationHook.Operation.Turn.Started, turnStarted)

packages/redcode/test/session/orphan.test.ts

@@ -0,0 +1,33 @@
+import { describe, expect, test } from "bun:test"
+import { orphans } from "@/session/orphan"
+
+type Message = Parameters<typeof orphans>[0][number]
+
+const assistant = (id: string, completed?: number) =>
+  ({
+    info: { id, role: "assistant", time: { created: 1, ...(completed ? { completed } : {}) } },
+    parts: [],
+  }) as unknown as Message
+
+const user = (id: string) => ({ info: { id, role: "user", time: { created: 1 } }, parts: [] }) as unknown as Message
+
+describe("turns left behind by a dead process", () => {
+  test("finds an assistant message nobody ever closed", () => {
+    // The process that would have written `completed` was killed; the message stays open, and the
+    // TUI reads open as "in progress" forever.
+    expect(orphans([user("u1"), assistant("a1")]).map((m) => m.id)).toEqual(["a1"] as never)
+  })
+
+  test("leaves finished turns alone", () => {
+    expect(orphans([user("u1"), assistant("a1", 2), user("u2"), assistant("a2", 3)])).toEqual([])
+  })
+
+  test("finds every one of them, not just the last", () => {
+    // Several crashes in a row leave several open messages, and one sweep should end them all.
+    expect(orphans([assistant("a1"), user("u1"), assistant("a2", 5), assistant("a3")]).map((m) => m.id)).toEqual(["a1", "a3"] as never)
+  })
+
+  test("ignores user messages, which never carry a completion", () => {
+    expect(orphans([user("u1"), user("u2")])).toEqual([])
+  })
+})

packages/redcode/test/session/prompt.test.ts

@@ -1605,6 +1605,38 @@ it.instance("writes down that a guard intervened, so the thresholds can be argue
   60_000,
 )
 
+it.instance("closes a turn left open by a process that died, instead of carrying it forever", () =>
+  Effect.gen(function* () {
+    // `time.completed` is written by the process running the turn. Killed mid-turn — an OOM, a
+    // machine asleep — nobody writes it, and the message stays open for the rest of the session's
+    // life: the TUI reads open as "in progress" and stamps QUEUED on everything typed after it,
+    // across restarts, with nothing running.
+    const { llm } = yield* useServerConfig(providerCfg)
+    const prompt = yield* SessionPrompt.Service
+    const sessions = yield* Session.Service
+    const guards = yield* SessionGuardLog.Service
+    const chat = yield* sessions.create({ title: "Pinned" })
+
+    const seeded = yield* seed(chat.id)
+    // Exactly what a killed process leaves behind: an assistant message with no completion.
+    const abandoned = { ...seeded.assistant, time: { created: seeded.assistant.time.created } }
+    yield* sessions.updateMessage(abandoned)
+    expect((yield* sessions.messages({ sessionID: chat.id })).some((m) => m.info.id === abandoned.id)).toBe(true)
+
+    yield* llm.text("carrying on")
+    yield* user(chat.id, "still there?")
+    yield* awaitWithTimeout(prompt.loop({ sessionID: chat.id }), "the turn never finished", "30 seconds")
+
+    const messages = yield* sessions.messages({ sessionID: chat.id })
+    const reaped = messages.find((item) => item.info.id === abandoned.id)
+    expect(reaped?.info.role === "assistant" && reaped.info.time.completed).toBeTruthy()
+    expect((reaped?.info as SessionV1.Assistant).error?.name).toBe("MessageAbortedError")
+    // And it is counted, so a week of these says the OOM came back.
+    expect((yield* guards.summary()).some((row) => row.guard === "orphan")).toBe(true)
+  }),
+  60_000,
+)
+
 it.instance("cancel records MessageAbortedError on interrupted process", () =>
   Effect.gen(function* () {
     const { llm } = yield* useServerConfig(providerCfg)

packages/tui/src/routes/session/index.tsx

@@ -251,6 +251,11 @@ export function Session() {
   const disabled = createMemo(() => permissions().length > 0 || questions().length > 0)
 
   const pending = createMemo(() => {
+    // An open assistant message alone is not a queue. A process killed mid-turn — an OOM, a
+    // machine asleep — never writes `time.completed`, so the message stays open forever and every
+    // later message read as QUEUED, across restarts, with nothing running behind it. The session
+    // has to actually be working for anything to be waiting on it.
+    if ((sync.data.session_status[route.sessionID]?.type ?? "idle") === "idle") return undefined
     const completed = messages().findLastIndex((message) => message.role === "assistant" && message.time.completed)
     const pending = messages().findLastIndex(
       (message, index) => index > completed && message.role === "assistant" && !message.time.completed,