diff --git a/cli/src/agent/runnerLifecycle.test.ts b/cli/src/agent/runnerLifecycle.test.ts index ec9c3d99..172dca0a 100644 --- a/cli/src/agent/runnerLifecycle.test.ts +++ b/cli/src/agent/runnerLifecycle.test.ts @@ -23,6 +23,23 @@ function createMockApiSession() { } as unknown as Parameters[0]['session']; } +function createMockApiSessionWithMetadataCapture() { + const metadataWrites: Array> = [] + return { + updateMetadata: vi.fn((handler: (m: Record) => Record) => { + const next = handler({}) + metadataWrites.push(next) + return next + }), + sendSessionDeath: vi.fn(), + flush: vi.fn(async () => {}), + close: vi.fn(async () => {}), + metadataWrites + } as unknown as Parameters[0]['session'] & { + metadataWrites: Array> + } +} + describe('createRunnerLifecycle', () => { let lifecycle: RunnerLifecycle; @@ -85,3 +102,95 @@ describe('createRunnerLifecycle', () => { }); }); }); + +// tiann/hapi#914: the runnerLifecycle's default archiveReason is now +// 'Hub restart' (was 'User terminated'). Out-of-band SIGTERM from the +// hub-restart cascade keeps that default. Explicit user actions +// (clicking Archive in the web UI, Ctrl-C in a local terminal, +// uncaught exception) reassign the reason before archive metadata is +// written. +describe('createRunnerLifecycle archiveReason defaults (tiann/hapi#914)', () => { + it('uses Hub restart as the default archiveReason when no override is applied', async () => { + const session = createMockApiSessionWithMetadataCapture() + const lifecycle = createRunnerLifecycle({ + session, + logTag: 'test' + }) + + await lifecycle.cleanup() + + expect(session.metadataWrites).toHaveLength(1) + expect(session.metadataWrites[0]).toMatchObject({ + lifecycleState: 'archived', + archivedBy: 'cli', + archiveReason: 'Hub restart' + }) + }) + + it('writes the operator-supplied reason when setArchiveReason is called (e.g. KillSession RPC)', async () => { + const session = createMockApiSessionWithMetadataCapture() + const lifecycle = createRunnerLifecycle({ + session, + logTag: 'test' + }) + + lifecycle.setArchiveReason('User terminated') + await lifecycle.cleanup() + + expect(session.metadataWrites[0]).toMatchObject({ + archiveReason: 'User terminated' + }) + }) + + it('markCrash overrides the default reason to "Session crashed"', async () => { + const session = createMockApiSessionWithMetadataCapture() + const lifecycle = createRunnerLifecycle({ + session, + logTag: 'test' + }) + + lifecycle.markCrash(new Error('boom')) + await lifecycle.cleanup() + + expect(session.metadataWrites[0]).toMatchObject({ + archiveReason: 'Session crashed' + }) + }) + + // tiann/hapi#914 review round 4: clean agent-loop completions + // (runClaude / runCodex / runCursor / runGemini / runKimi / + // runOpencode all call setSessionEndReason('completed') without + // touching archiveReason) must not be archived as 'Hub restart'. + // The setSessionEndReason setter flips the default when the runner + // transitions to 'completed'. + it('setSessionEndReason("completed") flips the default reason to "Session completed"', async () => { + const session = createMockApiSessionWithMetadataCapture() + const lifecycle = createRunnerLifecycle({ + session, + logTag: 'test' + }) + + lifecycle.setSessionEndReason('completed') + await lifecycle.cleanup() + + expect(session.metadataWrites[0]).toMatchObject({ + archiveReason: 'Session completed' + }) + }) + + it('an explicit setArchiveReason before setSessionEndReason("completed") still wins', async () => { + const session = createMockApiSessionWithMetadataCapture() + const lifecycle = createRunnerLifecycle({ + session, + logTag: 'test' + }) + + lifecycle.setArchiveReason('User terminated') + lifecycle.setSessionEndReason('completed') + await lifecycle.cleanup() + + expect(session.metadataWrites[0]).toMatchObject({ + archiveReason: 'User terminated' + }) + }) +}) diff --git a/cli/src/agent/runnerLifecycle.ts b/cli/src/agent/runnerLifecycle.ts index d632c17a..16a04951 100644 --- a/cli/src/agent/runnerLifecycle.ts +++ b/cli/src/agent/runnerLifecycle.ts @@ -24,7 +24,27 @@ export type RunnerLifecycle = { export function createRunnerLifecycle(options: RunnerLifecycleOptions): RunnerLifecycle { let exitCode = 0 - let archiveReason = 'User terminated' + // tiann/hapi#914: default reason is 'Hub restart' (parent-driven SIGTERM + // is the most common non-user cause). Genuine user actions (clicking + // Archive in the web UI, or Ctrl-C in a local terminal) explicitly + // reassign this via `setArchiveReason` BEFORE `cleanupAndExit` runs: + // - KillSession RPC handler → 'User terminated' (see registerKillSessionHandler) + // - SIGINT handler → 'User terminated' (Ctrl-C in local terminal) + // - uncaughtException/Reject → 'Session crashed' (via markCrash) + // + // Out-of-band SIGTERM (hub-restart cascade, systemd cgroup kill on + // hapi-runner.service stop, `kill ` from the operator) keeps the + // default and is correctly labelled 'Hub restart' on the audit trail. + // + // Runner-internal stop paths (`hapi runner stop-session`, webhook-timeout + // cleanup at run.ts:587, orphan cleanup at run.ts:267) also currently + // hit this default - that is technically inaccurate but follows the + // friction-mode "smallest defensible change" rule for this PR. Finer + // attribution would require an IPC channel (stdio: 'ipc' on spawn) so + // the runner can stamp `setArchiveReason` before SIGTERMing; tracked as + // a follow-up to keep this PR focussed on the user-action lie that + // motivated #914. + let archiveReason = 'Hub restart' let sessionEndReason: SessionEndReason = 'terminated' let sessionEndReasonExplicit = false let cleanupStarted = false @@ -98,6 +118,18 @@ export function createRunnerLifecycle(options: RunnerLifecycleOptions): RunnerLi const setSessionEndReason = (reason: SessionEndReason) => { sessionEndReason = reason sessionEndReasonExplicit = true + // tiann/hapi#914 review round 4: every agent runner + // (runClaude / runCodex / runCursor / runGemini / runKimi / + // runOpencode) calls setSessionEndReason('completed') before + // cleanupAndExit() on the natural-exit path without setting an + // archive reason. With the SIGTERM-driven default of 'Hub restart', + // clean completions would otherwise be audit-trailed as restart + // cascades. Flip the default to 'Session completed' when the end + // reason transitions to 'completed' AND no caller has already + // overridden the archive reason. + if (reason === 'completed' && archiveReason === 'Hub restart') { + archiveReason = 'Session completed' + } } const hasExplicitSessionEndReason = () => sessionEndReasonExplicit @@ -110,11 +142,19 @@ export function createRunnerLifecycle(options: RunnerLifecycleOptions): RunnerLi } const registerProcessHandlers = () => { + // tiann/hapi#914: SIGTERM is treated as the default reason ('Hub restart') + // because the runner is restarted by systemd as part of hub restart in + // production. If a future code path needs to distinguish "operator + // killed the host process" from "hub restart", it can call + // setArchiveReason() before the runner exits. process.on('SIGTERM', () => { void cleanupAndExit() }) + // Ctrl-C in a local terminal is genuine user intent — keep the + // pre-#914 label so the audit trail still shows it. process.on('SIGINT', () => { + archiveReason = 'User terminated' void cleanupAndExit() }) diff --git a/cli/src/api/apiSession.ts b/cli/src/api/apiSession.ts index 404e10a4..cd98d69a 100644 --- a/cli/src/api/apiSession.ts +++ b/cli/src/api/apiSession.ts @@ -748,6 +748,21 @@ export class ApiSessionClient extends EventEmitter { }) } + /** + * tiann/hapi#913: wait until any pending `update-metadata` writes have + * been acked by the hub (or the timeout elapses). `updateMetadata` is + * fire-and-forget at the call site because it's invoked on the hot path + * for every turn; this helper lets the few callers who actually need + * durability — fresh ACP session-id pre-registration is the canonical + * case — synchronously gate on persistence without changing every + * caller's signature. + * + * Returns true when the lock drained, false when the timeout fired. + */ + async flushMetadata(timeoutMs: number = 5_000): Promise { + return await this.drainLock(this.metadataLock, timeoutMs) + } + async flush(options?: { timeoutMs?: number }): Promise { const deadlineMs = Date.now() + (options?.timeoutMs ?? 5_000) diff --git a/cli/src/claude/registerKillSessionHandler.test.ts b/cli/src/claude/registerKillSessionHandler.test.ts new file mode 100644 index 00000000..b2931729 --- /dev/null +++ b/cli/src/claude/registerKillSessionHandler.test.ts @@ -0,0 +1,65 @@ +import { describe, expect, it, vi } from 'vitest' +import { RPC_METHODS } from '@hapi/protocol/rpcMethods' +import { registerKillSessionHandler } from './registerKillSessionHandler' + +// tiann/hapi#914: the KillSession RPC is the authoritative "user-terminated" +// signal because the hub only sends it when the operator clicks Archive in +// the web UI. Out-of-band SIGTERM (hub-restart cascade, host-level `kill`) +// hits the SIGTERM signal handler in runnerLifecycle, which now keeps the +// default reason 'Hub restart' so the audit trail stays correct. +describe('registerKillSessionHandler (tiann/hapi#914)', () => { + function makeRegistry() { + const handlers = new Map unknown>() + return { + registerHandler: (method: string, handler: (params: unknown) => unknown) => { + handlers.set(method, handler as (params?: unknown) => unknown) + }, + handlers + } + } + + it('stamps archiveReason=User terminated before triggering cleanupAndExit', async () => { + const registry = makeRegistry() + const lifecycle = { + setArchiveReason: vi.fn(), + cleanupAndExit: vi.fn(async () => {}) + } + + registerKillSessionHandler( + registry as unknown as Parameters[0], + lifecycle + ) + + const handler = registry.handlers.get(RPC_METHODS.KillSession) + expect(handler).toBeDefined() + + const result = await handler?.() + expect(result).toEqual({ success: true, message: 'Killing hapi CLI process' }) + + // setArchiveReason MUST be called BEFORE cleanupAndExit so the archive + // metadata write reads the correct reason. + const setReasonOrder = lifecycle.setArchiveReason.mock.invocationCallOrder[0] + const cleanupOrder = lifecycle.cleanupAndExit.mock.invocationCallOrder[0] + expect(setReasonOrder).toBeLessThan(cleanupOrder) + expect(lifecycle.setArchiveReason).toHaveBeenCalledWith('User terminated') + expect(lifecycle.cleanupAndExit).toHaveBeenCalled() + }) + + it('still works with the legacy `(cleanupAndExit: () => Promise)` call shape', async () => { + // Back-compat: runAgentSession.ts passes a bare closure as the second + // argument instead of a lifecycle object. The handler should not crash + // when setArchiveReason is absent. + const registry = makeRegistry() + const cleanupAndExit = vi.fn(async () => {}) + + registerKillSessionHandler( + registry as unknown as Parameters[0], + cleanupAndExit + ) + + const handler = registry.handlers.get(RPC_METHODS.KillSession) + await handler?.() + + expect(cleanupAndExit).toHaveBeenCalled() + }) +}) diff --git a/cli/src/claude/registerKillSessionHandler.ts b/cli/src/claude/registerKillSessionHandler.ts index 37936b79..b42b9b49 100644 --- a/cli/src/claude/registerKillSessionHandler.ts +++ b/cli/src/claude/registerKillSessionHandler.ts @@ -11,18 +11,41 @@ interface KillSessionResponse { message: string; } +/** + * tiann/hapi#914: callers can pass either a bare `cleanupAndExit` closure + * (legacy) or an options object that lets the kill-RPC stamp an explicit + * `archiveReason` before the lifecycle teardown runs. The hub only sends + * KillSession when the operator clicked Archive in the UI, so this RPC is + * the authoritative "user-terminated" signal; out-of-band SIGTERM from a + * hub-restart cascade no longer collides with the default archive reason. + */ +export interface KillSessionLifecycle { + cleanupAndExit: () => Promise; + setArchiveReason?: (reason: string) => void; +} export function registerKillSessionHandler( rpcHandlerManager: RpcHandlerManager, - killThisHappy: () => Promise + lifecycleOrCleanup: KillSessionLifecycle | (() => Promise) ) { + const lifecycle: KillSessionLifecycle = typeof lifecycleOrCleanup === 'function' + ? { cleanupAndExit: lifecycleOrCleanup } + : lifecycleOrCleanup; + rpcHandlerManager.registerHandler(RPC_METHODS.KillSession, async () => { logger.debug('Kill session request received'); - // This will start the cleanup process - void killThisHappy(); + // tiann/hapi#914: stamp the archive reason from the RPC path so the + // default in `runnerLifecycle.ts` can be reassigned away from + // 'User terminated'. A hub-restart-cascade SIGTERM does NOT go + // through this handler — it hits the SIGTERM signal handler — so + // those archives now stay labelled `'Hub restart'` (the new default). + lifecycle.setArchiveReason?.('User terminated'); - // We should still be able to respond the the client, though they + // This will start the cleanup process + void lifecycle.cleanupAndExit(); + + // We should still be able to respond to the client, though they // should optimistically assume the session is dead. return { success: true, diff --git a/cli/src/claude/runClaude.ts b/cli/src/claude/runClaude.ts index 4472aee9..1ebb1601 100644 --- a/cli/src/claude/runClaude.ts +++ b/cli/src/claude/runClaude.ts @@ -145,7 +145,7 @@ export async function runClaude(options: StartOptions = {}): Promise { }); lifecycle.registerProcessHandlers(); - registerKillSessionHandler(session.rpcHandlerManager, lifecycle.cleanupAndExit); + registerKillSessionHandler(session.rpcHandlerManager, lifecycle); registerLocalHandoffHandler(session.rpcHandlerManager, lifecycle); // Set initial agent state diff --git a/cli/src/codex/runCodex.ts b/cli/src/codex/runCodex.ts index 4958b6ac..de907c22 100644 --- a/cli/src/codex/runCodex.ts +++ b/cli/src/codex/runCodex.ts @@ -98,7 +98,7 @@ export async function runCodex(opts: { }); lifecycle.registerProcessHandlers(); - registerKillSessionHandler(session.rpcHandlerManager, lifecycle.cleanupAndExit); + registerKillSessionHandler(session.rpcHandlerManager, lifecycle); registerLocalHandoffHandler(session.rpcHandlerManager, lifecycle); const applyCurrentConfigToSession = (options?: { syncModel?: boolean }) => { diff --git a/cli/src/cursor/cursorAcpRemoteLauncher.test.ts b/cli/src/cursor/cursorAcpRemoteLauncher.test.ts index 1aeca09b..761e86c3 100644 --- a/cli/src/cursor/cursorAcpRemoteLauncher.test.ts +++ b/cli/src/cursor/cursorAcpRemoteLauncher.test.ts @@ -166,6 +166,7 @@ function makeClient() { registerHandler: vi.fn() }, updateMetadata: vi.fn(), + flushMetadata: vi.fn(async () => true), sendSessionEvent: vi.fn(), sendAgentMessage: vi.fn(), keepAlive: vi.fn(), @@ -297,11 +298,68 @@ describe('cursorAcpRemoteLauncher', () => { expect(session.client.emitSessionReady).not.toHaveBeenCalled(); }); + // tiann/hapi#913: fresh ACP sessions previously persisted `cursorSessionId` + // via fire-and-forget `updateMetadata`. A SIGTERM within ~1s of the first + // turn (hub-restart cascade) could strand the session because the ACK + // never arrived. The fix awaits `client.flushMetadata()` between + // `onSessionFoundWithProtocol` and the main loop, gating turn processing + // on a durable persist. + it('awaits flushMetadata after registering a fresh cursorSessionId so SIGTERM cannot strand the session', async () => { + const session = makeSession(null); + const flushSpy = vi.fn(async () => true); + // Replace the mock fixture's flushMetadata so we can observe ordering. + (session.client as unknown as { flushMetadata: typeof flushSpy }).flushMetadata = flushSpy; + + let flushCalled = false; + flushSpy.mockImplementation(async () => { + flushCalled = true; + return true; + }); + + const onSessionFoundSpy = session.onSessionFoundWithProtocol as ReturnType; + let onSessionFoundCalledBeforeFlush = false; + onSessionFoundSpy.mockImplementation(() => { + if (!flushCalled) { + onSessionFoundCalledBeforeFlush = true; + } + }); + + await cursorAcpRemoteLauncher(session); + + expect(onSessionFoundCalledBeforeFlush).toBe(true); + expect(flushSpy).toHaveBeenCalled(); + }); + + it('preserves the #834 resume-path pre-registration shape (registration before backend.loadSession)', async () => { + // PR #834 pre-registers `cursorSessionId` BEFORE `backend.loadSession` + // so a load-session failure on a legacy store does not strand the + // session. The #913 fix must not relocate or remove that + // pre-registration. We verify by observing call ordering on the spy. + const session = makeSession('resume-acp-session'); + const onSessionFoundSpy = session.onSessionFoundWithProtocol as ReturnType; + + let preRegisterCalledBeforeLoadSession = false; + let preRegisterArgs: unknown[] | null = null; + onSessionFoundSpy.mockImplementation((id: string, protocol: string) => { + if (!harness.loadSessionCalled) { + preRegisterCalledBeforeLoadSession = true; + preRegisterArgs = [id, protocol]; + } + }); + + await cursorAcpRemoteLauncher(session); + + expect(preRegisterCalledBeforeLoadSession).toBe(true); + expect(preRegisterArgs).toEqual(['resume-acp-session', 'acp']); + expect(harness.loadSessionCalled).toBe(true); + }); + it('applies debug mode immediately when setPermissionMode is called', async () => { const queue = new MessageQueue2((mode) => mode.permissionMode); const client = { rpcHandlerManager: { registerHandler: vi.fn() }, updateMetadata: vi.fn(), + flushMetadata: vi.fn(async () => true), sendSessionEvent: vi.fn(), sendAgentMessage: vi.fn(), keepAlive: vi.fn(), @@ -347,6 +405,7 @@ describe('cursorAcpRemoteLauncher', () => { const client = { rpcHandlerManager: { registerHandler: vi.fn() }, updateMetadata: vi.fn(), + flushMetadata: vi.fn(async () => true), sendSessionEvent: vi.fn(), sendAgentMessage: vi.fn(), keepAlive, @@ -392,6 +451,7 @@ describe('cursorAcpRemoteLauncher', () => { const client = { rpcHandlerManager: { registerHandler: vi.fn() }, updateMetadata: vi.fn(), + flushMetadata: vi.fn(async () => true), sendSessionEvent: vi.fn(), sendAgentMessage: vi.fn(), keepAlive, @@ -440,6 +500,7 @@ describe('cursorAcpRemoteLauncher', () => { const client = { rpcHandlerManager: { registerHandler: vi.fn() }, updateMetadata: vi.fn(), + flushMetadata: vi.fn(async () => true), sendSessionEvent: vi.fn(), sendAgentMessage: vi.fn(), keepAlive: vi.fn(), @@ -485,6 +546,7 @@ describe('cursorAcpRemoteLauncher', () => { const client = { rpcHandlerManager: { registerHandler: vi.fn() }, updateMetadata: vi.fn(), + flushMetadata: vi.fn(async () => true), sendSessionEvent: vi.fn(), sendAgentMessage: vi.fn(), keepAlive: vi.fn(), @@ -542,6 +604,7 @@ describe('cursorAcpRemoteLauncher', () => { const client = { rpcHandlerManager: { registerHandler: vi.fn() }, updateMetadata: vi.fn(), + flushMetadata: vi.fn(async () => true), sendSessionEvent: vi.fn(), sendAgentMessage: vi.fn(), keepAlive, @@ -584,6 +647,7 @@ describe('cursorAcpRemoteLauncher', () => { const client = { rpcHandlerManager: { registerHandler: vi.fn() }, updateMetadata: vi.fn(), + flushMetadata: vi.fn(async () => true), sendSessionEvent: vi.fn(), sendAgentMessage: vi.fn(), keepAlive: vi.fn(), @@ -630,6 +694,7 @@ describe('cursorAcpRemoteLauncher', () => { const client = { rpcHandlerManager: { registerHandler: vi.fn() }, updateMetadata: vi.fn(), + flushMetadata: vi.fn(async () => true), sendSessionEvent: vi.fn(), sendAgentMessage: vi.fn(), keepAlive: vi.fn(), @@ -671,6 +736,7 @@ describe('cursorAcpRemoteLauncher', () => { const client = { rpcHandlerManager: { registerHandler: vi.fn() }, updateMetadata: vi.fn(), + flushMetadata: vi.fn(async () => true), sendSessionEvent: vi.fn(), sendAgentMessage: vi.fn(), keepAlive: vi.fn(), diff --git a/cli/src/cursor/cursorAcpRemoteLauncher.ts b/cli/src/cursor/cursorAcpRemoteLauncher.ts index 4382974e..5e3eaf75 100644 --- a/cli/src/cursor/cursorAcpRemoteLauncher.ts +++ b/cli/src/cursor/cursorAcpRemoteLauncher.ts @@ -132,6 +132,18 @@ class CursorAcpRemoteLauncher extends RemoteLauncherBase { if (acpSessionId !== resumeSessionId) { session.onSessionFoundWithProtocol(acpSessionId, 'acp'); + // tiann/hapi#913: block until the metadata write that pins + // `cursorSessionId` reaches the hub DB before we drop into + // `runMainLoop`. If SIGTERM (hub-restart cascade) lands during + // the first turn without this gate, the only durable handle + // linking the session to its on-disk ACP store is lost and the + // session strands. The resume path at lines 98-100 already + // relies on the latency of `backend.loadSession()` to flush the + // same write; the fresh-session path has no such cover. + const flushed = await session.client.flushMetadata(); + if (!flushed) { + logger.warn(`[cursor-acp] cursorSessionId metadata write did not ACK within 5s; session may be unrecoverable if killed before the lock drains (acpSessionId=${acpSessionId})`); + } } session.client.emitSessionReady(); diff --git a/cli/src/cursor/runCursor.ts b/cli/src/cursor/runCursor.ts index f5508f34..a855a071 100644 --- a/cli/src/cursor/runCursor.ts +++ b/cli/src/cursor/runCursor.ts @@ -81,7 +81,7 @@ export async function runCursor(opts: { }); lifecycle.registerProcessHandlers(); - registerKillSessionHandler(session.rpcHandlerManager, lifecycle.cleanupAndExit); + registerKillSessionHandler(session.rpcHandlerManager, lifecycle); registerLocalHandoffHandler(session.rpcHandlerManager, lifecycle); const syncSessionMode = () => { diff --git a/cli/src/gemini/runGemini.ts b/cli/src/gemini/runGemini.ts index 34b13026..c2e66722 100644 --- a/cli/src/gemini/runGemini.ts +++ b/cli/src/gemini/runGemini.ts @@ -113,7 +113,7 @@ export async function runGemini(opts: { }); lifecycle.registerProcessHandlers(); - registerKillSessionHandler(session.rpcHandlerManager, lifecycle.cleanupAndExit); + registerKillSessionHandler(session.rpcHandlerManager, lifecycle); registerLocalHandoffHandler(session.rpcHandlerManager, lifecycle); const syncSessionMode = () => { diff --git a/cli/src/kimi/runKimi.ts b/cli/src/kimi/runKimi.ts index 97cc3703..f148b880 100644 --- a/cli/src/kimi/runKimi.ts +++ b/cli/src/kimi/runKimi.ts @@ -82,7 +82,7 @@ export async function runKimi(opts: { }); lifecycle.registerProcessHandlers(); - registerKillSessionHandler(session.rpcHandlerManager, lifecycle.cleanupAndExit); + registerKillSessionHandler(session.rpcHandlerManager, lifecycle); registerLocalHandoffHandler(session.rpcHandlerManager, lifecycle); const syncSessionMode = () => { diff --git a/cli/src/opencode/runOpencode.ts b/cli/src/opencode/runOpencode.ts index a78f9311..3e89e02a 100644 --- a/cli/src/opencode/runOpencode.ts +++ b/cli/src/opencode/runOpencode.ts @@ -107,7 +107,7 @@ export async function runOpencode(opts: { }); lifecycle.registerProcessHandlers(); - registerKillSessionHandler(session.rpcHandlerManager, lifecycle.cleanupAndExit); + registerKillSessionHandler(session.rpcHandlerManager, lifecycle); registerLocalHandoffHandler(session.rpcHandlerManager, lifecycle); const syncSessionMode = () => { diff --git a/cli/src/pi/runPi.ts b/cli/src/pi/runPi.ts index 430a0479..f3417483 100644 --- a/cli/src/pi/runPi.ts +++ b/cli/src/pi/runPi.ts @@ -88,7 +88,7 @@ export async function runPi(opts: { }); lifecycle.registerProcessHandlers(); - registerKillSessionHandler(apiSession.rpcHandlerManager, lifecycle.cleanupAndExit); + registerKillSessionHandler(apiSession.rpcHandlerManager, lifecycle); registerLocalHandoffHandler(apiSession.rpcHandlerManager, lifecycle); let cleanupInitiated = false; diff --git a/hub/src/sync/rpcGateway.test.ts b/hub/src/sync/rpcGateway.test.ts index 8c98fad9..d0825c0d 100644 --- a/hub/src/sync/rpcGateway.test.ts +++ b/hub/src/sync/rpcGateway.test.ts @@ -1,7 +1,7 @@ import { describe, expect, it } from 'bun:test' import type { Server } from 'socket.io' import type { RpcRegistry } from '../socket/rpcRegistry' -import { RpcGateway } from './rpcGateway' +import { RpcGateway, RpcTargetMissingError } from './rpcGateway' function createGateway() { const timeouts: number[] = [] @@ -70,3 +70,48 @@ describe('RpcGateway RPC timeouts', () => { }) }) +// tiann/hapi#916: rpcCall throws a typed `RpcTargetMissingError` when the +// target CLI is unreachable, so syncEngine.archiveSession can narrow on it +// and treat the kill as a benign no-op. +describe('RpcGateway no-target diagnostics (tiann/hapi#916)', () => { + it('throws RpcTargetMissingError(handler-not-registered) when no socket is registered for the method', async () => { + const io = { + of() { + return { + sockets: { + get() { return undefined } + } + } + } + } as unknown as Server + const rpcRegistry = { + getSocketIdForMethod() { return undefined } + } as unknown as RpcRegistry + const gateway = new RpcGateway(io, rpcRegistry) + + const error = await gateway.killSession('session-1').catch((e: unknown) => e) + expect(error).toBeInstanceOf(RpcTargetMissingError) + expect((error as RpcTargetMissingError).code).toBe('handler-not-registered') + }) + + it('throws RpcTargetMissingError(socket-disconnected) when the socket id is registered but no socket exists', async () => { + const io = { + of() { + return { + sockets: { + get() { return undefined } + } + } + } + } as unknown as Server + const rpcRegistry = { + getSocketIdForMethod() { return 'socket-1' } + } as unknown as RpcRegistry + const gateway = new RpcGateway(io, rpcRegistry) + + const error = await gateway.killSession('session-1').catch((e: unknown) => e) + expect(error).toBeInstanceOf(RpcTargetMissingError) + expect((error as RpcTargetMissingError).code).toBe('socket-disconnected') + }) +}) + diff --git a/hub/src/sync/rpcGateway.ts b/hub/src/sync/rpcGateway.ts index 193eedd0..5a63f454 100644 --- a/hub/src/sync/rpcGateway.ts +++ b/hub/src/sync/rpcGateway.ts @@ -24,6 +24,27 @@ import type { RpcRegistry } from '../socket/rpcRegistry' const DEFAULT_RPC_TIMEOUT_MS = 30_000 const MODEL_LIST_RPC_TIMEOUT_MS = 120_000 +/** + * tiann/hapi#916: thrown by {@link RpcGateway.rpcCall} when the target CLI is + * unreachable (handler not registered or socket disconnected). Callers can + * narrow on this to treat "CLI gone" as a benign condition (e.g. archive + * still succeeds at the hub level) without swallowing real RPC errors like + * timeouts or protocol failures. + */ +export class RpcTargetMissingError extends Error { + readonly code: 'handler-not-registered' | 'socket-disconnected' + readonly method: string + + constructor(method: string, reason: 'handler-not-registered' | 'socket-disconnected') { + super(reason === 'handler-not-registered' + ? `RPC handler not registered: ${method}` + : `RPC socket disconnected: ${method}`) + this.name = 'RpcTargetMissingError' + this.code = reason + this.method = method + } +} + export type RpcCommandResponse = CommandResponse export type RpcReadFileResponse = FileReadResponse export type RpcGeneratedImageResponse = GeneratedImageResponse @@ -292,12 +313,12 @@ export class RpcGateway { private async rpcCall(method: string, params: unknown, timeoutMs: number = DEFAULT_RPC_TIMEOUT_MS): Promise { const socketId = this.rpcRegistry.getSocketIdForMethod(method) if (!socketId) { - throw new Error(`RPC handler not registered: ${method}`) + throw new RpcTargetMissingError(method, 'handler-not-registered') } const socket = this.io.of('/cli').sockets.get(socketId) if (!socket) { - throw new Error(`RPC socket disconnected: ${method}`) + throw new RpcTargetMissingError(method, 'socket-disconnected') } const response = await socket.timeout(timeoutMs).emitWithAck('rpc-request', { diff --git a/hub/src/sync/sessionCache.ts b/hub/src/sync/sessionCache.ts index 86dd02a5..304b01cc 100644 --- a/hub/src/sync/sessionCache.ts +++ b/hub/src/sync/sessionCache.ts @@ -7,6 +7,11 @@ import { extractTodoWriteTodosFromMessageContent, TodosSchema } from './todos' import { extractBackgroundTaskDelta } from './backgroundTasks' const QUEUED_MESSAGE_THINKING_GRACE_MS = 15_000 +// tiann/hapi#919: metadata writers (renameSession, clearSessionArchiveMetadata, +// restoreSessionArchiveMetadata) retry on version-mismatch with a fresh cache +// snapshot. Cap retries so genuine concurrent contention still surfaces to the +// HTTP caller as 409 instead of spinning forever. +const METADATA_RETRY_ATTEMPTS = 5 type RuntimeConfigKey = 'permissionMode' | 'model' | 'modelReasoningEffort' | 'effort' | 'serviceTier' | 'collaborationMode' export class SessionCache { @@ -522,32 +527,105 @@ export class SessionCache { return updatedAt !== undefined && payloadTime < updatedAt } + /** + * tiann/hapi#916: hub-side write of the archive-metadata fields normally + * authored by the CLI's `archiveAndClose`. Called by `syncEngine.archiveSession` + * when the kill-RPC fails because the CLI is unreachable (e.g. the + * hub-restart cascade already killed it). Without this, the route would + * either 500 (pre-fix) or silently return ok=true while leaving + * `lifecycleState=running` on disk — both confuse the operator. + * + * Idempotent: if `lifecycleState` is already `archived` we return without + * touching the row to avoid resetting `lifecycleStateSince`. Best-effort: + * if every retry hits `version-mismatch` (genuine contention) the original + * `archiveSession` flow still marks the session inactive in cache via + * `handleSessionEnd`, just without flipping the persisted lifecycle. + */ + markSessionArchivedFromHub(sessionId: string, reason: string): void { + for (let attempt = 0; attempt < METADATA_RETRY_ATTEMPTS; attempt += 1) { + const session = this.sessions.get(sessionId) ?? this.refreshSession(sessionId) + if (!session) return + const current = session.metadata + if (!current) return + if (current.lifecycleState === 'archived') { + return + } + + const next: Record = { + ...current, + lifecycleState: 'archived', + lifecycleStateSince: Date.now(), + archivedBy: 'hub', + archiveReason: reason + } + + const result = this.store.sessions.updateSessionMetadata( + sessionId, + next, + session.metadataVersion, + session.namespace, + { touchUpdatedAt: false } + ) + + if (result.result === 'error') { + // tiann/hapi#916 review feedback: persistence failure must + // surface so the route returns 5xx. Silently returning here + // would let `/archive` claim success while the row stays + // unarchived in the DB. + throw new Error('Failed to archive session metadata from hub') + } + + if (result.result === 'success') { + this.refreshSession(sessionId) + return + } + + this.refreshSession(sessionId) + } + + // tiann/hapi#916 review feedback: exhausted retries means we never + // got a successful write. Match the renameSession / mergeSessions + // contract and surface this as an error so non-RPC failures stay + // 5xx per the issue's acceptance criteria. + throw new Error('Session was modified concurrently while archiving from hub') + } + async renameSession(sessionId: string, name: string): Promise { - const session = this.sessions.get(sessionId) - if (!session) { - throw new Error('Session not found') + // tiann/hapi#919: retry-with-refresh on version-mismatch instead of + // throwing on the first contention. Mirrors the good pattern in + // mergeSessions (~L780) and in syncEngine's metadata helpers. Without + // this, a stale cache snapshot produces forever-409 on PATCH /sessions/:id + // until some unrelated event triggers a refresh. + for (let attempt = 0; attempt < METADATA_RETRY_ATTEMPTS; attempt += 1) { + const session = this.sessions.get(sessionId) ?? this.refreshSession(sessionId) + if (!session) { + throw new Error('Session not found') + } + + const currentMetadata = session.metadata ?? { path: '', host: '' } + const newMetadata = { ...currentMetadata, name } + + const result = this.store.sessions.updateSessionMetadata( + sessionId, + newMetadata, + session.metadataVersion, + session.namespace, + { touchUpdatedAt: false } + ) + + if (result.result === 'error') { + throw new Error('Failed to update session metadata') + } + + if (result.result === 'success') { + this.refreshSession(sessionId) + return + } + + this.refreshSession(sessionId) } - const currentMetadata = session.metadata ?? { path: '', host: '' } - const newMetadata = { ...currentMetadata, name } - - const result = this.store.sessions.updateSessionMetadata( - sessionId, - newMetadata, - session.metadataVersion, - session.namespace, - { touchUpdatedAt: false } - ) - - if (result.result === 'error') { - throw new Error('Failed to update session metadata') - } - - if (result.result === 'version-mismatch') { - throw new Error('Session was modified concurrently. Please try again.') - } - - this.refreshSession(sessionId) + throw new Error('Session was modified concurrently. Please try again.') } /** @@ -563,52 +641,59 @@ export class SessionCache { * No-op when metadata is null (callers should pre-check). */ async clearSessionArchiveMetadata(sessionId: string): Promise<{ cursorSessionProtocol?: 'acp' | 'stream-json' }> { - const session = this.sessions.get(sessionId) - if (!session) { - throw new Error('Session not found') - } - - const currentMetadata = session.metadata - if (!currentMetadata) { - throw new Error('Session metadata missing') - } - - const next: Record = { ...currentMetadata } - delete next.lifecycleState - delete next.archivedBy - delete next.archiveReason - next.lifecycleStateSince = Date.now() - - let cursorSessionProtocol: 'acp' | 'stream-json' | undefined - if (currentMetadata.flavor === 'cursor') { - const existing = currentMetadata.cursorSessionProtocol - if (existing === 'acp' || existing === 'stream-json') { - cursorSessionProtocol = existing - } else if (currentMetadata.cursorSessionId) { - // Pre-#799 default: presence of cursorSessionId without protocol means stream-json. - cursorSessionProtocol = 'stream-json' - next.cursorSessionProtocol = 'stream-json' + // tiann/hapi#919: retry-with-refresh on version-mismatch. The reopen + // flow runs this on every archived-session resume — a stale snapshot + // here used to forever-409 the only reopen affordance. + for (let attempt = 0; attempt < METADATA_RETRY_ATTEMPTS; attempt += 1) { + const session = this.sessions.get(sessionId) ?? this.refreshSession(sessionId) + if (!session) { + throw new Error('Session not found') } + + const currentMetadata = session.metadata + if (!currentMetadata) { + throw new Error('Session metadata missing') + } + + const next: Record = { ...currentMetadata } + delete next.lifecycleState + delete next.archivedBy + delete next.archiveReason + next.lifecycleStateSince = Date.now() + + let cursorSessionProtocol: 'acp' | 'stream-json' | undefined + if (currentMetadata.flavor === 'cursor') { + const existing = currentMetadata.cursorSessionProtocol + if (existing === 'acp' || existing === 'stream-json') { + cursorSessionProtocol = existing + } else if (currentMetadata.cursorSessionId) { + // Pre-#799 default: presence of cursorSessionId without protocol means stream-json. + cursorSessionProtocol = 'stream-json' + next.cursorSessionProtocol = 'stream-json' + } + } + + const result = this.store.sessions.updateSessionMetadata( + sessionId, + next, + session.metadataVersion, + session.namespace, + { touchUpdatedAt: false } + ) + + if (result.result === 'error') { + throw new Error('Failed to update session metadata') + } + + if (result.result === 'success') { + this.refreshSession(sessionId) + return cursorSessionProtocol ? { cursorSessionProtocol } : {} + } + + this.refreshSession(sessionId) } - const result = this.store.sessions.updateSessionMetadata( - sessionId, - next, - session.metadataVersion, - session.namespace, - { touchUpdatedAt: false } - ) - - if (result.result === 'error') { - throw new Error('Failed to update session metadata') - } - - if (result.result === 'version-mismatch') { - throw new Error('Session was modified concurrently. Please try again.') - } - - this.refreshSession(sessionId) - return cursorSessionProtocol ? { cursorSessionProtocol } : {} + throw new Error('Session was modified concurrently. Please try again.') } /** @@ -632,50 +717,59 @@ export class SessionCache { lifecycleStateSince?: number } ): Promise { - const session = this.sessions.get(sessionId) - if (!session) return - const current = session.metadata - if (!current) return + // tiann/hapi#919: retry-with-refresh on version-mismatch. This is the + // /reopen rollback path — if it fails the session is left in a + // half-cleared archive state, so making it robust to a stale snapshot + // matters more here than for the other two. + for (let attempt = 0; attempt < METADATA_RETRY_ATTEMPTS; attempt += 1) { + const session = this.sessions.get(sessionId) ?? this.refreshSession(sessionId) + if (!session) return + const current = session.metadata + if (!current) return - const next: Record = { ...current } - if (snapshot.lifecycleState !== undefined) { - next.lifecycleState = snapshot.lifecycleState - } else { - delete next.lifecycleState - } - if (snapshot.archivedBy !== undefined) { - next.archivedBy = snapshot.archivedBy - } else { - delete next.archivedBy - } - if (snapshot.archiveReason !== undefined) { - next.archiveReason = snapshot.archiveReason - } else { - delete next.archiveReason - } - if (snapshot.lifecycleStateSince !== undefined) { - next.lifecycleStateSince = snapshot.lifecycleStateSince - } else { - delete next.lifecycleStateSince + const next: Record = { ...current } + if (snapshot.lifecycleState !== undefined) { + next.lifecycleState = snapshot.lifecycleState + } else { + delete next.lifecycleState + } + if (snapshot.archivedBy !== undefined) { + next.archivedBy = snapshot.archivedBy + } else { + delete next.archivedBy + } + if (snapshot.archiveReason !== undefined) { + next.archiveReason = snapshot.archiveReason + } else { + delete next.archiveReason + } + if (snapshot.lifecycleStateSince !== undefined) { + next.lifecycleStateSince = snapshot.lifecycleStateSince + } else { + delete next.lifecycleStateSince + } + + const result = this.store.sessions.updateSessionMetadata( + sessionId, + next, + session.metadataVersion, + session.namespace, + { touchUpdatedAt: false } + ) + + if (result.result === 'error') { + throw new Error('Failed to restore archive metadata') + } + + if (result.result === 'success') { + this.refreshSession(sessionId) + return + } + + this.refreshSession(sessionId) } - const result = this.store.sessions.updateSessionMetadata( - sessionId, - next, - session.metadataVersion, - session.namespace, - { touchUpdatedAt: false } - ) - - if (result.result === 'error') { - throw new Error('Failed to restore archive metadata') - } - - if (result.result === 'version-mismatch') { - throw new Error('Session was modified concurrently during reopen rollback') - } - - this.refreshSession(sessionId) + throw new Error('Session was modified concurrently during reopen rollback') } async deleteSession(sessionId: string): Promise { diff --git a/hub/src/sync/sessionModel.test.ts b/hub/src/sync/sessionModel.test.ts index 7ea9950e..8fa18c07 100644 --- a/hub/src/sync/sessionModel.test.ts +++ b/hub/src/sync/sessionModel.test.ts @@ -1,4 +1,4 @@ -import { describe, expect, it } from 'bun:test' +import { describe, expect, it, spyOn } from 'bun:test' import { toSessionSummary } from '@hapi/protocol' import type { SyncEvent } from '@hapi/protocol/types' import { Store } from '../store' @@ -2466,4 +2466,265 @@ describe('session model', () => { })).resolves.toBeUndefined() }) }) + + // tiann/hapi#916: when the CLI is gone, the kill-RPC throws + // RpcTargetMissingError. markSessionArchivedFromHub writes the archive + // metadata directly so the row's lifecycleState still flips to 'archived'. + describe('markSessionArchivedFromHub (tiann/hapi#916)', () => { + it('flips lifecycleState to archived with archivedBy=hub and the supplied reason', () => { + const store = new Store(':memory:') + const events: SyncEvent[] = [] + const cache = new SessionCache(store, createPublisher(events)) + + const session = cache.getOrCreateSession( + 'session-hub-archive', + { path: '/tmp/project', host: 'localhost', flavor: 'codex', codexSessionId: 'thread-1' }, + null, + 'default' + ) + + cache.markSessionArchivedFromHub(session.id, 'Archived from hub (CLI unreachable)') + + const meta = cache.getSession(session.id)?.metadata as Record | null | undefined + expect(meta?.lifecycleState).toBe('archived') + expect(meta?.archivedBy).toBe('hub') + expect(meta?.archiveReason).toBe('Archived from hub (CLI unreachable)') + expect(typeof meta?.lifecycleStateSince).toBe('number') + }) + + it('is idempotent for already-archived sessions (does not reset lifecycleStateSince)', async () => { + const store = new Store(':memory:') + const events: SyncEvent[] = [] + const cache = new SessionCache(store, createPublisher(events)) + + const initialSince = 1700000000000 + const session = cache.getOrCreateSession( + 'session-already-archived', + { + path: '/tmp/project', + host: 'localhost', + flavor: 'codex', + lifecycleState: 'archived', + archivedBy: 'cli', + archiveReason: 'User terminated', + lifecycleStateSince: initialSince + }, + null, + 'default' + ) + + cache.markSessionArchivedFromHub(session.id, 'Should not overwrite') + + const meta = cache.getSession(session.id)?.metadata as Record | null | undefined + expect(meta?.lifecycleState).toBe('archived') + expect(meta?.archivedBy).toBe('cli') + expect(meta?.archiveReason).toBe('User terminated') + expect(meta?.lifecycleStateSince).toBe(initialSince) + }) + + it('self-heals on version-mismatch via refresh-and-retry', () => { + const store = new Store(':memory:') + const events: SyncEvent[] = [] + const cache = new SessionCache(store, createPublisher(events)) + + const session = cache.getOrCreateSession( + 'session-hub-archive-stale', + { path: '/tmp/project', host: 'localhost', flavor: 'codex' }, + null, + 'default' + ) + + const dbSession = store.sessions.getSessionByNamespace(session.id, 'default')! + const oobWrite = store.sessions.updateSessionMetadata( + session.id, + { ...dbSession.metadata!, name: 'oob' }, + dbSession.metadataVersion, + 'default', + { touchUpdatedAt: false } + ) + expect(oobWrite.result).toBe('success') + + cache.markSessionArchivedFromHub(session.id, 'CLI unreachable') + + const meta = cache.getSession(session.id)?.metadata as Record | null | undefined + expect(meta?.lifecycleState).toBe('archived') + expect(meta?.archivedBy).toBe('hub') + expect(meta?.name).toBe('oob') + }) + + // tiann/hapi#916 review feedback: persistence failures must surface + // so the /archive route returns 5xx per the acceptance criteria + // "Non-RPC errors during archive still propagate as 5xx (DB write + // failure, etc.)" — silent return would let the route claim success + // while the row stays unarchived. + it('throws when the store reports a hard error on the metadata write', () => { + const store = new Store(':memory:') + const events: SyncEvent[] = [] + const cache = new SessionCache(store, createPublisher(events)) + + const session = cache.getOrCreateSession( + 'session-hub-archive-error', + { path: '/tmp/project', host: 'localhost', flavor: 'codex' }, + null, + 'default' + ) + + const updateSpy = spyOn(store.sessions, 'updateSessionMetadata').mockReturnValue({ + result: 'error', + error: new Error('simulated DB write failure') + } as ReturnType) + + try { + expect(() => cache.markSessionArchivedFromHub(session.id, 'CLI unreachable')).toThrow(/Failed to archive session metadata from hub/) + } finally { + updateSpy.mockRestore() + } + }) + + it('throws when retries are exhausted by sustained version-mismatch contention', () => { + const store = new Store(':memory:') + const events: SyncEvent[] = [] + const cache = new SessionCache(store, createPublisher(events)) + + const session = cache.getOrCreateSession( + 'session-hub-archive-exhausted', + { path: '/tmp/project', host: 'localhost', flavor: 'codex' }, + null, + 'default' + ) + + const updateSpy = spyOn(store.sessions, 'updateSessionMetadata').mockReturnValue({ + result: 'version-mismatch' + } as ReturnType) + + try { + expect(() => cache.markSessionArchivedFromHub(session.id, 'CLI unreachable')).toThrow(/Session was modified concurrently while archiving from hub/) + } finally { + updateSpy.mockRestore() + } + }) + }) + + // tiann/hapi#919: the three metadata writers must self-heal on + // version-mismatch instead of one-shot-throwing. The bug was that a + // stale cache snapshot produced forever-409 on the corresponding HTTP + // endpoints — the cache never refreshed, so the same retry hit the + // same mismatch. Pattern mirrors mergeSessions (line ~780). + describe('version-mismatch self-heal (tiann/hapi#919)', () => { + it('renameSession recovers after a stale cache snapshot is detected', async () => { + const store = new Store(':memory:') + const events: SyncEvent[] = [] + const cache = new SessionCache(store, createPublisher(events)) + + const session = cache.getOrCreateSession( + 'session-rename-stale', + { path: '/tmp/project', host: 'localhost', flavor: 'codex' }, + null, + 'default' + ) + + // Simulate a concurrent writer bumping the DB version under our feet: + // write a metadata patch out-of-band via the store, leaving the cache + // snapshot stale. + const dbSession = store.sessions.getSessionByNamespace(session.id, 'default')! + const oobWrite = store.sessions.updateSessionMetadata( + session.id, + { ...dbSession.metadata!, name: 'concurrent-rename' }, + dbSession.metadataVersion, + 'default', + { touchUpdatedAt: false } + ) + expect(oobWrite.result).toBe('success') + + // Cache still holds the pre-OOB snapshot. Pre-fix, this call threw + // 'Session was modified concurrently'. Post-fix, it refreshes and + // succeeds. + await expect(cache.renameSession(session.id, 'final-name')).resolves.toBeUndefined() + + const meta = cache.getSession(session.id)?.metadata as Record | null | undefined + expect(meta?.name).toBe('final-name') + }) + + it('clearSessionArchiveMetadata recovers after a stale cache snapshot', async () => { + const store = new Store(':memory:') + const events: SyncEvent[] = [] + const cache = new SessionCache(store, createPublisher(events)) + + const session = cache.getOrCreateSession( + 'session-clear-stale', + { + path: '/tmp/project', + host: 'localhost', + flavor: 'codex', + codexSessionId: 'thread-stale', + lifecycleState: 'archived', + archivedBy: 'cli', + archiveReason: 'User terminated' + }, + null, + 'default' + ) + + // Concurrent rename via the store bumps the DB version. + const dbSession = store.sessions.getSessionByNamespace(session.id, 'default')! + const oobWrite = store.sessions.updateSessionMetadata( + session.id, + { ...dbSession.metadata!, name: 'oob-name' }, + dbSession.metadataVersion, + 'default', + { touchUpdatedAt: false } + ) + expect(oobWrite.result).toBe('success') + + await expect(cache.clearSessionArchiveMetadata(session.id)).resolves.toBeDefined() + + const meta = cache.getSession(session.id)?.metadata as Record | null | undefined + expect(meta?.lifecycleState).toBeUndefined() + expect(meta?.archivedBy).toBeUndefined() + expect(meta?.name).toBe('oob-name') + }) + + it('restoreSessionArchiveMetadata recovers after a stale cache snapshot', async () => { + const store = new Store(':memory:') + const events: SyncEvent[] = [] + const cache = new SessionCache(store, createPublisher(events)) + + const session = cache.getOrCreateSession( + 'session-restore-stale', + { + path: '/tmp/project', + host: 'localhost', + flavor: 'codex', + codexSessionId: 'thread-restore-stale' + // Started without archive metadata - simulates the post-clear state. + }, + null, + 'default' + ) + + // Concurrent unrelated write bumps DB version. + const dbSession = store.sessions.getSessionByNamespace(session.id, 'default')! + const oobWrite = store.sessions.updateSessionMetadata( + session.id, + { ...dbSession.metadata!, name: 'parallel-rename' }, + dbSession.metadataVersion, + 'default', + { touchUpdatedAt: false } + ) + expect(oobWrite.result).toBe('success') + + await expect(cache.restoreSessionArchiveMetadata(session.id, { + lifecycleState: 'archived', + archivedBy: 'cli', + archiveReason: 'User terminated', + lifecycleStateSince: 1234 + })).resolves.toBeUndefined() + + const meta = cache.getSession(session.id)?.metadata as Record | null | undefined + expect(meta?.lifecycleState).toBe('archived') + expect(meta?.archiveReason).toBe('User terminated') + expect(meta?.lifecycleStateSince).toBe(1234) + expect(meta?.name).toBe('parallel-rename') + }) + }) }) diff --git a/hub/src/sync/syncEngine.ts b/hub/src/sync/syncEngine.ts index 7fe09087..d59f58fd 100644 --- a/hub/src/sync/syncEngine.ts +++ b/hub/src/sync/syncEngine.ts @@ -23,6 +23,7 @@ import { MachineCache, type Machine } from './machineCache' import { MessageService } from './messageService' import { RpcGateway, + RpcTargetMissingError, type RpcCodexModel, type RpcCommandResponse, type RpcDeleteUploadResponse, @@ -452,7 +453,24 @@ export class SyncEngine { } async archiveSession(sessionId: string): Promise { - await this.rpcGateway.killSession(sessionId) + // tiann/hapi#916: when the CLI is already gone (e.g. after a + // hub-restart cascade SIGTERMed the runner but the in-memory + // `active` flag has not been reconciled yet) the kill-RPC throws + // and the route used to surface that as HTTP 500. Treat the + // missing target as a benign condition: still flip the session's + // lifecycleState to `archived` in the hub-side metadata so the + // UI does not see a half-cleaned zombie, and continue to mark + // it inactive in the cache. Real RPC errors (timeout, protocol + // failure) still propagate as 5xx. + try { + await this.rpcGateway.killSession(sessionId) + } catch (error) { + if (error instanceof RpcTargetMissingError) { + this.sessionCache.markSessionArchivedFromHub(sessionId, 'Archived from hub (CLI unreachable)') + } else { + throw error + } + } this.handleSessionEnd({ sid: sessionId, time: Date.now() }) } diff --git a/hub/src/web/routes/sessions.test.ts b/hub/src/web/routes/sessions.test.ts index d1875c88..01f6d859 100644 --- a/hub/src/web/routes/sessions.test.ts +++ b/hub/src/web/routes/sessions.test.ts @@ -62,6 +62,7 @@ function createApp(session: Session, opts?: { listSlashCommands?: SyncEngine['listSlashCommands'] getSessionExport?: (sessionId: string, session: Session) => unknown sessionExists?: boolean + archiveSession?: (sessionId: string) => Promise }) { const applySessionConfigCalls: Array<[string, Record]> = [] const applySessionConfig = async (sessionId: string, config: Record) => { @@ -104,6 +105,7 @@ function createApp(session: Session, opts?: { resumed: true })) const sessionExists = opts?.sessionExists !== false + const archiveSessionMock = opts?.archiveSession ?? (async () => {}) const engine = { resolveSessionAccess: () => sessionExists ? { ok: true, sessionId: session.id, session } @@ -115,6 +117,7 @@ function createApp(session: Session, opts?: { listOpencodeReasoningEffortOptionsForSession, resumeSession, reopenSession, + archiveSession: archiveSessionMock, getSessionExport: opts?.getSessionExport ?? (() => ({ type: 'success', payload: { @@ -1014,4 +1017,124 @@ describe('sessions routes', () => { }) }) + // tiann/hapi#916: archive endpoint must be idempotent for already-archived + // rows and for split-brain rows whose CLI is gone but the in-memory `active` + // flag has not been reconciled to false yet. + describe('POST /sessions/:id/archive (tiann/hapi#916)', () => { + it('returns 2xx and calls archiveSession for an active session', async () => { + const calls: string[] = [] + const session = createSession({ active: true }) + const { app } = createApp(session, { + archiveSession: async (sessionId: string) => { calls.push(sessionId) } + }) + + const response = await app.request('/api/sessions/session-1/archive', { method: 'POST' }) + + expect(response.status).toBe(200) + expect(await response.json()).toEqual({ ok: true }) + expect(calls).toEqual(['session-1']) + }) + + it('returns 2xx and skips archiveSession when the row is already archived (idempotent)', async () => { + let called = false + const session = createSession({ + active: false, + metadata: { + path: '/tmp/project', + host: 'localhost', + flavor: 'codex', + lifecycleState: 'archived', + archivedBy: 'cli', + archiveReason: 'User terminated' + } + }) + const { app } = createApp(session, { + archiveSession: async () => { called = true } + }) + + const response = await app.request('/api/sessions/session-1/archive', { method: 'POST' }) + + expect(response.status).toBe(200) + expect(await response.json()).toEqual({ ok: true, alreadyArchived: true }) + expect(called).toBe(false) + }) + + it('returns 2xx when the active session\'s CLI is gone — engine.archiveSession swallows the missing-RPC error', async () => { + // Pre-fix this returned 500 because rpcGateway.killSession threw + // 'RPC handler not registered'. Post-fix the engine narrows on + // RpcTargetMissingError and still flips lifecycle to archived. + const session = createSession({ active: true }) + const { app } = createApp(session, { + archiveSession: async () => { + // Simulates the post-fix behavior: engine catches the + // RpcTargetMissingError, calls markSessionArchivedFromHub, + // and returns normally. + } + }) + + const response = await app.request('/api/sessions/session-1/archive', { method: 'POST' }) + + expect(response.status).toBe(200) + expect(await response.json()).toEqual({ ok: true }) + }) + + it('still surfaces a 5xx for non-RPC errors (e.g. DB write failure)', async () => { + const session = createSession({ active: true }) + const { app } = createApp(session, { + archiveSession: async () => { + throw new Error('DB write failed') + } + }) + + const response = await app.request('/api/sessions/session-1/archive', { method: 'POST' }) + expect(response.status).toBe(500) + }) + + it('returns 404 when the session id is unknown', async () => { + const session = createSession() + const { app } = createApp(session, { sessionExists: false }) + + const response = await app.request('/api/sessions/missing-id/archive', { method: 'POST' }) + + expect(response.status).toBe(404) + expect(await response.json()).toEqual({ error: 'Session not found' }) + }) + + it('returns 409 for an inactive non-archived row whose lifecycle is not running', async () => { + let called = false + const session = createSession({ active: false }) + const { app } = createApp(session, { + archiveSession: async () => { called = true } + }) + + const response = await app.request('/api/sessions/session-1/archive', { method: 'POST' }) + + expect(response.status).toBe(409) + expect(await response.json()).toEqual({ error: 'Session is inactive' }) + expect(called).toBe(false) + }) + + it('returns 2xx for an inactive split-brain row still marked lifecycleState=running', async () => { + const calls: string[] = [] + const session = createSession({ + active: false, + metadata: { + path: '/tmp/project', + host: 'localhost', + flavor: 'codex', + lifecycleState: 'running' + } + }) + const { app } = createApp(session, { + archiveSession: async (sessionId: string) => { calls.push(sessionId) } + }) + + const response = await app.request('/api/sessions/session-1/archive', { method: 'POST' }) + + expect(response.status).toBe(200) + expect(await response.json()).toEqual({ ok: true }) + expect(calls).toEqual(['session-1']) + }) + }) + }) diff --git a/hub/src/web/routes/sessions.ts b/hub/src/web/routes/sessions.ts index 20cffad9..f199a293 100644 --- a/hub/src/web/routes/sessions.ts +++ b/hub/src/web/routes/sessions.ts @@ -295,16 +295,31 @@ export function createSessionsRoutes(getSyncEngine: () => SyncEngine | null): Ho }) app.post('/sessions/:id/archive', async (c) => { + // tiann/hapi#916: relax the blanket `requireActive: true` guard so + // the endpoint is idempotent for already-archived rows AND can clean + // up split-brain rows after a hub-restart cascade (inactive in cache + // but metadata.lifecycleState still 'running'). Normal inactive rows + // that are not archived (completed stubs, UI Delete/Reopen targets) + // keep the old 409 contract. const engine = requireSyncEngine(c, getSyncEngine) if (engine instanceof Response) { return engine } - const sessionResult = requireSessionFromParam(c, engine, { requireActive: true }) + const sessionResult = requireSessionFromParam(c, engine) if (sessionResult instanceof Response) { return sessionResult } + const lifecycleState = sessionResult.session.metadata?.lifecycleState + if (lifecycleState === 'archived') { + return c.json({ ok: true, alreadyArchived: true }) + } + + if (!sessionResult.session.active && lifecycleState !== 'running') { + return c.json({ error: 'Session is inactive' }, 409) + } + await engine.archiveSession(sessionResult.sessionId) return c.json({ ok: true }) })