From 3b1411d444612c291f0dea8e55dc71c0d2ad8152 Mon Sep 17 00:00:00 2001 From: Filipe Forattini Date: Fri, 4 Sep 2026 18:14:12 -0300 Subject: [PATCH] fix(session): close turns left open by a process that died Only one run exists per session at a time, so an assistant message still open when a fresh run starts was left by a run that no longer exists. Until now nothing closed it, and the TUI reads open as "a turn is in progress": every message typed afterwards was stamped QUEUED, across restarts, with nothing running behind it. A session that survived one OOM looked jammed forever. Close them on the way in, count them as an intervention so a week of these says the OOM came back, and stop the badge from calling an open message a queue when the session is idle. Claude-Session: https://claude.ai/code/session_014XwJPDhq1ahcm3rd454WzQ --- .changeset/orphan-turns.md | 8 +++++ packages/redcode/src/session/guard-log.ts | 2 +- packages/redcode/src/session/orphan.ts | 23 ++++++++++++++ packages/redcode/src/session/prompt.ts | 26 +++++++++++++++ packages/redcode/test/session/orphan.test.ts | 33 ++++++++++++++++++++ packages/redcode/test/session/prompt.test.ts | 32 +++++++++++++++++++ packages/tui/src/routes/session/index.tsx | 5 +++ 7 files changed, 128 insertions(+), 1 deletion(-) create mode 100644 .changeset/orphan-turns.md create mode 100644 packages/redcode/src/session/orphan.ts create mode 100644 packages/redcode/test/session/orphan.test.ts diff --git a/.changeset/orphan-turns.md b/.changeset/orphan-turns.md new file mode 100644 index 000000000000..8bba08862894 --- /dev/null +++ b/.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. diff --git a/packages/redcode/src/session/guard-log.ts b/packages/redcode/src/session/guard-log.ts index c2050763aea2..84529e7b3dd5 100644 --- a/packages/redcode/src/session/guard-log.ts +++ b/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 { diff --git a/packages/redcode/src/session/orphan.ts b/packages/redcode/src/session/orphan.ts new file mode 100644 index 000000000000..e3dd7af6d450 --- /dev/null +++ b/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" diff --git a/packages/redcode/src/session/prompt.ts b/packages/redcode/src/session/prompt.ts index 1259ca5e92b2..01c9ecd639eb 100644 --- a/packages/redcode/src/session/prompt.ts +++ b/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) diff --git a/packages/redcode/test/session/orphan.test.ts b/packages/redcode/test/session/orphan.test.ts new file mode 100644 index 000000000000..ba53344845ac --- /dev/null +++ b/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[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([]) + }) +}) diff --git a/packages/redcode/test/session/prompt.test.ts b/packages/redcode/test/session/prompt.test.ts index f64adb7ad569..a14258acdc12 100644 --- a/packages/redcode/test/session/prompt.test.ts +++ b/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) diff --git a/packages/tui/src/routes/session/index.tsx b/packages/tui/src/routes/session/index.tsx index 5440967ad3e9..cb776334d924 100644 --- a/packages/tui/src/routes/session/index.tsx +++ b/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,