From 30518dc79fe7c88f9ec7736816a2f020e3a31411 Mon Sep 17 00:00:00 2001 From: Brian Anglin Date: Thu, 3 Sep 2026 15:37:17 -0700 Subject: [PATCH] Resume chats that were mid-turn when Kanna shut down MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Quitting Kanna, or letting an update restart it, cancelled every running turn and left the work there. Coming back meant finding each chat and nudging it by hand, and a long agentic turn that was most of the way through a task lost all of it. Shutdown now marks each in-flight chat before cancelling it, and the next boot restarts those turns. The marker is persisted (`resumePending`, set by a new `turn_resume_pending_set` turn event) rather than inferred from `lastTurnStartedAt > lastTurnEndedAt`: shutdown cancels the turn like any other cancel, so the timestamps alone can't tell "we exited" from "the user pressed stop", and inferring it would make every chat ever killed with `kill -9` — including months-old ones — start running again on upgrade. The resume prompt is wire-only, so the transcript shows the interrupted turn picking back up rather than a user message nobody typed. It leans on the harness session surviving: resumed by token, it still holds the original prompt and everything the turn did before it died, so "carry on, and re-check whatever you had in flight" is all that has to be said. A chat with no session token isn't resumed at all — a bare "carry on" into an empty session is worse than leaving the turn interrupted — and when the session is gone entirely, startTurnForChat's existing recovery rebuilds context from our transcript first. Everything a user-initiated cancel does still happens on shutdown, so a chat that never gets resumed reads exactly as it does today. Chats deleted or archived while the turn ran are skipped, and the marker is cleared before the attempt, so a resume that fails isn't retried on every boot from here to eternity: one shutdown earns one resume. 🌸 Shipped with Kanna — https://kanna.sh Co-Authored-By: Kanna Kanna-Agent: claude/opus[1m] --- CLAUDE.md | 5 ++ src/server/agent.test.ts | 138 ++++++++++++++++++++++++++++++++ src/server/agent.ts | 81 +++++++++++++++++++ src/server/event-store.test.ts | 29 +++++++ src/server/event-store.ts | 30 +++++++ src/server/events.ts | 19 +++++ src/server/resume-turns.test.ts | 119 +++++++++++++++++++++++++++ src/server/resume-turns.ts | 68 ++++++++++++++++ src/server/server.ts | 27 ++++++- 9 files changed, 513 insertions(+), 3 deletions(-) create mode 100644 src/server/resume-turns.test.ts create mode 100644 src/server/resume-turns.ts diff --git a/CLAUDE.md b/CLAUDE.md index d19ff319e..19ff9d6f3 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -41,6 +41,11 @@ React client (src/client) - Provider adapters normalize three different wire protocols into `HarnessEvent`s (`harness-types.ts`). Claude runs through the Agent SDK in `agent.ts` directly; codex/cursor/pi produce `HarnessTurn`s. +- Shutdown cancels every in-flight turn and marks its chat `resumePending` + (`agent.interruptForShutdown`); the next boot restarts those turns with a + wire-only "carry on" prompt (`resume-turns.ts`). A user-initiated cancel + never sets the marker, and the marker is cleared before the attempt, so one + shutdown earns one resume. - Transcripts are append-only JSONL per chat (`transcripts/.jsonl`) with a small LRU cache in the EventStore. `debugRaw` (raw provider JSON) is stamped only on `system_init` — the one entry with a raw JSON view. Tool diff --git a/src/server/agent.test.ts b/src/server/agent.test.ts index d20719134..c1585eccc 100644 --- a/src/server/agent.test.ts +++ b/src/server/agent.test.ts @@ -10,6 +10,7 @@ import { normalizeClaudeContextUsage, normalizeClaudeStreamMessage, normalizeClaudeUsageSnapshot, + RESUME_AFTER_RESTART_MESSAGE, } from "./agent" import type { HarnessTurn } from "./harness-types" import type { ChatAttachment, TranscriptEntry } from "../shared/types" @@ -2373,6 +2374,132 @@ describe("session restore on lost native session", () => { }) }) +describe("AgentCoordinator restart resume", () => { + test("shutdown cancels running turns and marks their chats for resume", async () => { + const events = new AsyncEventQueue() + const fakeCodexManager = { + async startSession() {}, + async startTurn(): Promise { + return { + provider: "codex", + stream: events, + interrupt: async () => {}, + close: () => events.close(), + } + }, + } + + const store = createFakeStore() + const coordinator = new AgentCoordinator({ + store: store as never, + onStateChange: () => {}, + codexManager: fakeCodexManager as never, + }) + + await coordinator.send({ + type: "chat.send", + chatId: "chat-1", + provider: "codex", + content: "long running task", + }) + await waitFor(() => coordinator.activeTurns.has("chat-1")) + + await coordinator.interruptForShutdown() + + expect(store.chat.resumePending).toBe(true) + expect(coordinator.activeTurns.size).toBe(0) + // The turn is still cancelled like any other, so a chat that never gets + // resumed reads exactly as it does today. + expect(store.messages.some((entry) => entry.kind === "interrupted")).toBe(true) + }) + + test("a user-initiated cancel leaves no resume marker", async () => { + const events = new AsyncEventQueue() + const fakeCodexManager = { + async startSession() {}, + async startTurn(): Promise { + return { + provider: "codex", + stream: events, + interrupt: async () => {}, + close: () => events.close(), + } + }, + } + + const store = createFakeStore() + const coordinator = new AgentCoordinator({ + store: store as never, + onStateChange: () => {}, + codexManager: fakeCodexManager as never, + }) + + await coordinator.send({ + type: "chat.send", + chatId: "chat-1", + provider: "codex", + content: "long running task", + }) + await waitFor(() => coordinator.activeTurns.has("chat-1")) + + await coordinator.cancel("chat-1") + + expect(store.chat.resumePending).toBeUndefined() + }) + + test("resuming an interrupted turn sends a wire-only continuation, not a user prompt", async () => { + const events = new AsyncEventQueue() + const prompts: string[] = [] + const store = createFakeStore() + store.chat.provider = "claude" + store.chat.sessionToken = "session-1" + store.chat.resumePending = true + store.chat.lastModel = "opus" + + const coordinator = new AgentCoordinator({ + store: store as never, + onStateChange: () => {}, + checkSessionArtifact: () => "present" as SessionArtifactStatus, + startClaudeSession: async () => ({ + provider: "claude", + stream: events, + getAccountInfo: async () => null, + interrupt: async () => {}, + close: () => {}, + setModel: async () => {}, + setPermissionMode: async () => {}, + sendPrompt: async (content: string) => { + prompts.push(content) + }, + }), + }) + + expect(await coordinator.resumeInterruptedTurn("chat-1")).toBe(true) + + expect(prompts).toEqual([RESUME_AFTER_RESTART_MESSAGE]) + // Nobody typed anything, so nothing lands in the transcript as if they had. + expect(store.messages.some((entry) => entry.kind === "user_prompt")).toBe(false) + expect(coordinator.activeTurns.has("chat-1")).toBe(true) + }) + + test("does not resume a chat with no session to resume into", async () => { + const store = createFakeStore() + store.chat.provider = "claude" + store.chat.resumePending = true + + const coordinator = new AgentCoordinator({ + store: store as never, + onStateChange: () => {}, + startClaudeSession: async () => { + throw new Error("Should not start a session") + }, + }) + + expect(await coordinator.resumeInterruptedTurn("chat-1")).toBe(false) + expect(store.messages).toEqual([]) + }) +}) + function createFakeChat(id: string, projectId: string, title = "New Chat") { return { id, @@ -2383,6 +2510,9 @@ function createFakeChat(id: string, projectId: string, title = "New Chat") { autoPlan: false, sessionToken: null as string | null, pendingForkSessionToken: null as string | null, + resumePending: undefined as boolean | undefined, + lastModel: undefined as string | undefined, + deletedAt: undefined as number | undefined, } } @@ -2441,6 +2571,14 @@ function createFakeStore(options?: { throw new Error("Did not expect turn failure") }, async recordTurnCancelled() {}, + async setTurnResumePending(chatId: string, pending: boolean) { + const target = requireChat(chatId) + if (pending) { + target.resumePending = true + } else { + delete target.resumePending + } + }, async setSessionToken(chatId: string, sessionToken: string | null) { requireChat(chatId).sessionToken = sessionToken }, diff --git a/src/server/agent.ts b/src/server/agent.ts index e0b005038..ff4da586e 100644 --- a/src/server/agent.ts +++ b/src/server/agent.ts @@ -233,6 +233,18 @@ const STEERED_MESSAGE_PREFIX = ` The user would like to inform you of something while you continue to work. Acknowledge receipt immediately with a text response, then continue with the task at hand, incorporating the user's feedback if needed. ` +/** + * Wire-only prompt that restarts a turn Kanna killed by shutting down (see + * `AgentCoordinator.resumeInterruptedTurn`). The harness session carries the + * original prompt and the work already done, so this only has to say what + * happened and warn that the last thing it was doing may be half-finished — + * the process died mid-tool-call as often as not. + */ +export const RESUME_AFTER_RESTART_MESSAGE = ` +Kanna restarted while you were working on this, so your process was stopped mid-task and is now back up. Continue the task you were on from where it left off. +Whatever you were doing last may not have completed — re-check the state of any file you were editing and any command you had running before assuming it finished. +` + interface SendMessageOptions { provider?: AgentProvider model?: string @@ -1833,6 +1845,75 @@ export class AgentCoordinator { return scanned?.path ? { name: scanned.name, path: scanned.path } : undefined } + /** + * Cancel every in-flight turn because Kanna itself is going down, marking + * each chat so the next boot picks the work back up (`resumeInterruptedTurn`). + * + * Everything a user-initiated cancel does still happens — the harness is + * interrupted, the pending tool call is discarded, the transcript gets its + * `interrupted` entry — so a chat that never gets resumed reads exactly as it + * does today. The marker is written first: a shutdown that dies partway + * through leaves a chat resumable-but-not-cancelled, which the resume pass + * handles, rather than cancelled-but-forgotten, which it can't. + */ + async interruptForShutdown() { + for (const chatId of [...this.activeTurns.keys()]) { + try { + await this.store.setTurnResumePending(chatId, true) + } catch { + // Best effort — a chat we can't mark still gets cancelled cleanly. + } + await this.cancel(chatId) + } + } + + /** + * Restart a turn that the previous process cut short by shutting down. + * + * The prompt is wire-only (`appendUserPrompt: false`), so the transcript + * shows the interrupted turn picking back up rather than a user message + * nobody typed. Resuming leans on the harness session having survived: it + * holds the original prompt and everything the turn did before it died, so + * "carry on" is all that has to be said. When the session is gone, + * `startTurnForChat`'s own recovery notices and rebuilds the context from our + * transcript first (`prepareSessionRestore`), which is exactly what's wanted. + * + * Returns whether a turn was actually started. + */ + async resumeInterruptedTurn(chatId: string) { + const chat = this.store.getChat(chatId) + if (!chat || chat.deletedAt) return false + if (!chat.provider) return false + if (this.activeTurns.has(chatId)) return false + // No session to resume into means the harness never got far enough to have + // context worth continuing; a bare "carry on" would be sent into an empty + // session, so leave the chat interrupted instead. + if (!chat.sessionToken && !chat.pendingForkSessionToken) return false + + // Everything the chat record remembers about how the turn was running: + // the model it actually ran with plus the two persisted modes. Reasoning + // effort and fast mode are picked in the composer and never stored server + // side, so the resumed turn falls back to the provider defaults for those. + const settings = this.getProviderSettings(chat.provider, { + model: chat.lastModel, + planMode: chat.planMode, + autoPlan: chat.autoPlan, + }) + await this.startTurnForChat({ + chatId, + provider: chat.provider, + content: RESUME_AFTER_RESTART_MESSAGE, + attachments: [], + model: settings.model, + effort: settings.effort, + serviceTier: settings.serviceTier, + planMode: settings.planMode, + autoPlan: settings.autoPlan, + appendUserPrompt: false, + }) + return true + } + async forkChat(chatId: string) { const chat = this.store.requireChat(chatId) if (this.activeTurns.has(chatId) || this.drainingStreams.has(chatId)) { diff --git a/src/server/event-store.test.ts b/src/server/event-store.test.ts index 320e43c68..4136b3ea6 100644 --- a/src/server/event-store.test.ts +++ b/src/server/event-store.test.ts @@ -839,6 +839,35 @@ describe("EventStore", () => { expect(store.getMessages(forked.id)).toEqual(store.getMessages(source.id)) }) + test("the resume marker survives a restart and a compaction", async () => { + const dataDir = await createTempDataDir() + const store = new EventStore(dataDir) + await store.initialize() + + const project = await store.openProject("/tmp/project") + const chat = await store.createChat(project.id) + await store.recordTurnStarted(chat.id) + await store.setTurnResumePending(chat.id, true) + // Shutdown cancels the turn like any other cancel; the marker is what tells + // the next boot the difference. + await store.recordTurnCancelled(chat.id) + expect(store.requireChat(chat.id).resumePending).toBe(true) + expect(store.requireChat(chat.id).lastTurnOutcome).toBe("cancelled") + + const reloaded = new EventStore(dataDir) + await reloaded.initialize() + expect(reloaded.requireChat(chat.id).resumePending).toBe(true) + + // Cleared by the boot that acts on it, and the clear sticks the same way. + await reloaded.setTurnResumePending(chat.id, false) + await reloaded.compact() + expect(reloaded.requireChat(chat.id).resumePending).toBeUndefined() + + const afterCompaction = new EventStore(dataDir) + await afterCompaction.initialize() + expect(afterCompaction.requireChat(chat.id).resumePending).toBeUndefined() + }) + test("lastAgentMessageAt tracks agent entries mid-turn, ignoring user prompts", async () => { const dataDir = await createTempDataDir() const store = new EventStore(dataDir) diff --git a/src/server/event-store.ts b/src/server/event-store.ts index 1bf572d81..64f06a96c 100644 --- a/src/server/event-store.ts +++ b/src/server/event-store.ts @@ -210,6 +210,7 @@ function getReplayEventPriority(event: StoreEvent) { case "pending_fork_session_token_set": return 6 case "turn_cancelled": + case "turn_resume_pending_set": return 7 case "turn_finished": case "turn_failed": @@ -861,6 +862,16 @@ export class EventStore { chat.lastTurnEndedAt = event.timestamp break } + case "turn_resume_pending_set": { + const chat = this.state.chatsById.get(event.chatId) + if (!chat) break + if (event.pending) { + chat.resumePending = true + } else { + delete chat.resumePending + } + break + } case "session_token_set": { const chat = this.state.chatsById.get(event.chatId) if (!chat) break @@ -1826,6 +1837,25 @@ export class EventStore { this.onTurnEnded?.(chatId) } + /** + * Flag (or clear) a chat whose turn Kanna cut short by shutting down, so the + * next process can pick it back up. Deliberately does not touch `updatedAt`: + * it's bookkeeping about the process, not activity in the chat, and bumping + * it would shuffle the sidebar on every boot. + */ + async setTurnResumePending(chatId: string, pending: boolean) { + const chat = this.requireChat(chatId) + if (Boolean(chat.resumePending) === pending) return + const event: TurnEvent = { + v: STORE_VERSION, + type: "turn_resume_pending_set", + timestamp: Date.now(), + chatId, + pending, + } + await this.append(this.turnsLogPath, event) + } + async setSessionToken(chatId: string, sessionToken: string | null) { const chat = this.requireChat(chatId) if (chat.sessionToken === sessionToken) return diff --git a/src/server/events.ts b/src/server/events.ts index 5743bc219..59187b101 100644 --- a/src/server/events.ts +++ b/src/server/events.ts @@ -103,6 +103,18 @@ export interface ChatRecord { */ lastAgentMessagePreviewAt?: number lastTurnOutcome: "success" | "failed" | "cancelled" | null + /** + * Set when a turn was cut short because Kanna itself went down, and cleared + * by the next boot's resume pass. A turn the *user* stopped never gets this, + * which is the whole point: it's how the next process tells "this chat was + * mid-task when we exited" from "this chat was stopped on purpose". + * + * Persisted rather than derived from `lastTurnStartedAt > lastTurnEndedAt`: + * shutdown cancels the turn like any other cancel, so the timestamps alone + * can't distinguish the two, and inferring it would make every chat killed + * by an old `kill -9` resume out of nowhere on upgrade. + */ + resumePending?: boolean /** * Files this chat has changed, unioned across its turns and measured by * diffing worktree snapshots at each turn boundary (see `TurnFileTracker`). @@ -356,6 +368,13 @@ export type TurnEvent = timestamp: number chatId: string } + | { + v: 2 + type: "turn_resume_pending_set" + timestamp: number + chatId: string + pending: boolean + } | { v: 2 type: "session_token_set" diff --git a/src/server/resume-turns.test.ts b/src/server/resume-turns.test.ts new file mode 100644 index 000000000..46e136bf3 --- /dev/null +++ b/src/server/resume-turns.test.ts @@ -0,0 +1,119 @@ +import { describe, expect, test } from "bun:test" +import { resumeInterruptedTurns } from "./resume-turns" +import type { ChatRecord, StoreState } from "./events" + +function createChat(id: string, overrides?: Partial): ChatRecord { + return { + id, + projectId: "project-1", + title: id, + createdAt: 1, + updatedAt: 1, + unread: false, + provider: "claude", + planMode: false, + autoPlan: false, + sessionToken: "session-1", + lastTurnOutcome: null, + ...overrides, + } +} + +function createDeps(chats: ChatRecord[], options?: { resume?: (chatId: string) => Promise }) { + const state = { + projectsById: new Map(), + projectIdsByPath: new Map(), + chatsById: new Map(chats.map((chat) => [chat.id, chat])), + queuedMessagesByChatId: new Map(), + } as StoreState + const cleared: string[] = [] + const attempted: string[] = [] + const errors: Array<{ chatId: string; error: unknown }> = [] + return { + cleared, + attempted, + errors, + deps: { + store: { + state, + setTurnResumePending: async (chatId: string, pending: boolean) => { + if (!pending) cleared.push(chatId) + delete state.chatsById.get(chatId)!.resumePending + }, + }, + agent: { + resumeInterruptedTurn: async (chatId: string) => { + attempted.push(chatId) + return await (options?.resume?.(chatId) ?? Promise.resolve(true)) + }, + }, + onError: (chatId: string, error: unknown) => { + errors.push({ chatId, error }) + }, + }, + } +} + +describe("resumeInterruptedTurns", () => { + test("resumes only the chats the last shutdown marked", async () => { + const { deps, attempted } = createDeps([ + createChat("marked", { resumePending: true }), + createChat("idle"), + // A turn that never ended but was never marked either — a hard crash, or + // a log from before resume existed. Left alone on purpose. + createChat("crashed", { lastTurnStartedAt: 10 }), + ]) + + expect(await resumeInterruptedTurns(deps)).toEqual(["marked"]) + expect(attempted).toEqual(["marked"]) + }) + + test("skips chats the user deleted or archived while the turn was running", async () => { + const { deps, attempted } = createDeps([ + createChat("deleted", { resumePending: true, deletedAt: 5 }), + createChat("archived", { resumePending: true, archivedAt: 5 }), + ]) + + expect(await resumeInterruptedTurns(deps)).toEqual([]) + expect(attempted).toEqual([]) + }) + + test("resumes in the order the interrupted turns started", async () => { + const { deps, attempted } = createDeps([ + createChat("second", { resumePending: true, lastTurnStartedAt: 200 }), + createChat("first", { resumePending: true, lastTurnStartedAt: 100 }), + createChat("third", { resumePending: true, lastTurnStartedAt: 300 }), + ]) + + await resumeInterruptedTurns(deps) + expect(attempted).toEqual(["first", "second", "third"]) + }) + + test("clears the marker even when the resume fails, so it is not retried every boot", async () => { + const { deps, cleared, errors } = createDeps( + [ + createChat("broken", { resumePending: true, lastTurnStartedAt: 100 }), + createChat("fine", { resumePending: true, lastTurnStartedAt: 200 }), + ], + { resume: async (chatId) => { + if (chatId === "broken") throw new Error("harness is gone") + return true + } } + ) + + expect(await resumeInterruptedTurns(deps)).toEqual(["fine"]) + expect(cleared).toEqual(["broken", "fine"]) + expect(errors).toHaveLength(1) + expect(errors[0]?.chatId).toBe("broken") + // A failure on one chat must not strand the ones behind it. + expect(deps.store.state.chatsById.get("fine")?.resumePending).toBeUndefined() + }) + + test("reports a chat that could not be resumed without counting it as resumed", async () => { + const { deps } = createDeps([createChat("no-session", { resumePending: true })], { + resume: async () => false, + }) + + expect(await resumeInterruptedTurns(deps)).toEqual([]) + }) +}) diff --git a/src/server/resume-turns.ts b/src/server/resume-turns.ts new file mode 100644 index 000000000..d53765560 --- /dev/null +++ b/src/server/resume-turns.ts @@ -0,0 +1,68 @@ +import type { ChatRecord, StoreState } from "./events" + +/** + * Boot-time counterpart to `AgentCoordinator.interruptForShutdown`: pick back + * up every chat whose turn the previous process cut short by exiting. + * + * Only chats the shutdown actually marked are considered — a crash that never + * ran the shutdown path leaves no marker and no resume, which is deliberate. + * Inferring "in progress" from `lastTurnStartedAt > lastTurnEndedAt` instead + * would make every chat ever killed with `kill -9`, including ones from months + * ago, start running again on the next upgrade. + */ +export interface ResumeInterruptedTurnsDeps { + store: { + state: StoreState + setTurnResumePending: (chatId: string, pending: boolean) => Promise + } + agent: { + resumeInterruptedTurn: (chatId: string) => Promise + } + onError?: (chatId: string, error: unknown) => void +} + +function isResumable(chat: ChatRecord) { + if (!chat.resumePending) return false + // Deleted is self-explanatory; archived means the user put the chat away + // between the turn starting and the shutdown, and starting an agent in it + // behind their back is worse than leaving the turn interrupted. + return !chat.deletedAt && !chat.archivedAt +} + +/** + * Resumes each marked chat and returns the ids that actually started a turn. + * + * Sequential rather than parallel: every resume spawns a harness process, and + * a machine that went down with six chats running should not try to bring six + * of them up in the same instant as the rest of boot. + */ +export async function resumeInterruptedTurns(deps: ResumeInterruptedTurnsDeps): Promise { + const pending = [...deps.store.state.chatsById.values()] + .filter(isResumable) + // Oldest interruption first, so the resumes go out in the order the turns + // originally started rather than in map order. + .sort((left, right) => (left.lastTurnStartedAt ?? left.updatedAt) - (right.lastTurnStartedAt ?? right.updatedAt)) + + const resumed: string[] = [] + for (const chat of pending) { + // Cleared *before* the attempt: a chat that can't be resumed (no session + // to resume into, harness fails to start) must not be retried on every + // boot from here to eternity. One shutdown earns one resume attempt. + try { + await deps.store.setTurnResumePending(chat.id, false) + } catch (error) { + deps.onError?.(chat.id, error) + continue + } + + try { + if (await deps.agent.resumeInterruptedTurn(chat.id)) { + resumed.push(chat.id) + } + } catch (error) { + deps.onError?.(chat.id, error) + } + } + + return resumed +} diff --git a/src/server/server.ts b/src/server/server.ts index 0caeaa67e..3fefb1fda 100644 --- a/src/server/server.ts +++ b/src/server/server.ts @@ -24,6 +24,7 @@ import { DiffStore } from "./diff-store" import { WorktreeProbe } from "./worktree-probe" import { TurnFileTracker } from "./worktree-snapshot" import { backfillTouchedFileBases } from "./touched-file-backfill" +import { resumeInterruptedTurns } from "./resume-turns" import { discoverProjects, type DiscoveredProject } from "./discovery" import { KeybindingsManager } from "./keybindings" import { clearGitHubRepoCache } from "./github" @@ -346,7 +347,27 @@ export async function startKannaServer(options: StartKannaServerOptions = {}) { await router.broadcastSnapshots() } } + // Chats that were mid-turn when Kanna last exited pick up where they left + // off. Not awaited — each resume starts a harness process, and boot should + // not wait on them; chained onto the GC sweep so a chat about to be archived + // or deleted for staleness isn't resumed on its way out. void runStartupGc() + .then(() => resumeInterruptedTurns({ + store, + agent, + onError: (chatId, error) => { + console.warn(`${LOG_PREFIX} could not resume chat ${chatId} after restart:`, error) + }, + })) + .then(async (resumedChatIds) => { + if (resumedChatIds.length > 0) { + console.log(`${LOG_PREFIX} resumed ${resumedChatIds.length} chat(s) interrupted by the last shutdown`) + await router.broadcastSnapshots() + } + }) + .catch((error) => { + console.warn(`${LOG_PREFIX} resuming interrupted chats failed:`, error) + }) // Then keep sweeping for the lifetime of the (potentially months-long) // process: empties every minute, deletes daily, archives every 6 hours. @@ -661,9 +682,9 @@ export async function startKannaServer(options: StartKannaServerOptions = {}) { clearInterval(staleChatAutoArchiveInterval) clearInterval(staleChatDeleteInterval) worktreeProbe.stop() - for (const chatId of [...agent.activeTurns.keys()]) { - await agent.cancel(chatId) - } + // Cancels every in-flight turn *and* marks its chat, so the next boot + // restarts the work instead of leaving it interrupted (see resume-turns.ts). + await agent.interruptForShutdown() router.dispose() providerAuth.dispose() usageLimits.dispose()