diff --git a/docs/workspace-recycle.md b/docs/workspace-recycle.md new file mode 100644 index 000000000..a4fc398c0 --- /dev/null +++ b/docs/workspace-recycle.md @@ -0,0 +1,110 @@ +# 工作区会话回收 + +`botmux workspace-recycle` 在外部工作区回收成功后，关闭事先确认属于该工作区的 Botmux 会话。发现依据为会话自身的 `workingDir`；群绑定、路径不存在、Bot 名称和标题都不单独授权关闭。跨 Bot 的写入由各自在线 daemon 执行，CLI 没有离线改库或全局删除兜底。 + +## 操作入口 + +先安装包含本功能的 Botmux 到工作区之外，并让涉及的 daemon 使用同版本。不要从即将被删除的 checkout 运行收尾命令。数据目录也必须在目标工作区之外。 + +```sh +# 只读。也可对已经不存在的目录生成存量候选清单。 +botmux workspace-recycle discover --workspace /work/tasks/example + +# 回收前：目录必须仍然存在。operationId 由外部通用 Hook 持久化并复用。 +botmux workspace-recycle prepare \ + --workspace /work/tasks/example --operation reclaim-example-001 + +# 外部系统执行它自己获准的工作区回收，保存真实结果。 +# 只有回收成功且旧目录确实消失后，才通知成功。 +botmux workspace-recycle finish \ + --operation reclaim-example-001 --event end-result-example-001 --outcome succeeded + +# 回收失败：不关闭任何会话，终止该次操作。 +botmux workspace-recycle finish \ + --operation reclaim-example-001 --event end-result-example-001 --outcome failed + +# 纯回读，包含各会话的结果、资源快照和失败原因。 +botmux workspace-recycle status --operation reclaim-example-001 +``` + +同一次操作的终态事件不能从失败改成成功，也不能换 `eventId`。失败回收后的新一轮使用新的 `operationId`。准备失败时，外部 Hook 应停止工作区删除；修复原因后重新准备，身份或目标集合变更时创建新的操作。相同成功事件可以重放，已验证关闭的会话只回读。 + +主机 CLI 会从受管进程上下文识别当前会话；脱离会话的 Hook 可在 `prepare` 显式传 `--initiator `。它必须属于发现的目标集合。准备时允许该发起会话仍在运行，其他忙碌会话会阻止准备成功。成功回收后： + +1. 先关闭并验证其他目标；任何失败、残留或覆盖缺口都会保留发起会话。 +2. 将发起会话的交接写入工作区之外的持久日志，先返回 `pending` 回执。 +3. daemon 在后续检查点等待该会话空闲、无未结束 turn/队列、无新输入，并再次检查其他目标的关闭状态及目标集合。 +4. 条件满足才最后关闭发起会话。等待超过 15 分钟、路径/输入/worker 代次变化等情况返回具体 blocker。不会以忙碌强杀解决这些 blocker。 + +成功回收后的关闭会把 `workspaceRetirement` 与 `closed` 状态一起持久化。该会话永久退役：CLI、卡片、API 和会话群自动续聊均拒绝恢复，会话群也不会因恢复被拒绝而隐式新建会话。重建同名目录或修改旧行的工作目录不能解除退役；继续工作需在有效工作区显式创建新会话。已关闭历史、资源残留的关闭记录也保留退役标记，失败回收不会添加标记。 + +daemon 重启会恢复已持久化的延迟交接；如果重启改变了 worker 或输入证据，会安全停止，不能把重新启动的执行当成原空闲会话。普通部分失败不自动重试，操作员可检查原因后重放同一成功事件。`pending` 不是资源释放成功，调用方须用 `status` 获取最终回读。 + +## 通用生命周期 Hook 适配 + +Botmux 不调用工作区删除程序，也不往 Agent Task 核心植入 Botmux 逻辑。外部生命周期系统显式配置两次命令调用，将它自己的事件映射成下面的中立 JSON。当前接口可独立验证；接入某个通用 Hook 实现之后，还需要验证真实事件时序与失败传播。 + +```json +{ + "protocol": "botmux.workspace-recycle.v1", + "phase": "before-reclaim", + "operationId": "reclaim-example-001", + "workspacePath": "/work/tasks/example", + "initiatorSessionId": "exact-botmux-session-id" +} +``` + +```json +{ + "protocol": "botmux.workspace-recycle.v1", + "phase": "after-reclaim", + "operationId": "reclaim-example-001", + "eventId": "durable-end-result-example-001", + "outcome": "succeeded" +} +``` + +两个事件均交给同一个显式入口： + +```sh +/opt/botmux/bin/botmux workspace-recycle hook --event-file - < /durable/events/event.json +``` + +配置方应保证： + +- before 事件在删除前执行，准备失败会阻止删除；after 事件读取实际回收结果。 +- operationId、原事件和命令回执保存在工作区之外；同一次失败的网络回执先回读再重放。 +- 两个阶段使用相同主机与 `SESSION_DATA_DIR`，覆盖所有相关 Bot。daemon 不在线、版本缺少此接口、鉴权失败或任意 store 无法读取均作为失败处理。 +- 若另有群工作目录改绑 Hook，应将群绑定指向实际归档结果；绑定变化不替代运行中会话的关闭。新出现且不在本次计划中的会话只报告，不能偷偷扩充关闭集合。 +- 所有 Hook 默认是显式配置的命令调用；安装本功能不会扫描并关闭历史工作区。历史候选只能只读展示，实际清理需要另外确认精确目标和适用流程。 + +## 结果与资源证据 + +返回 JSON 中保留 exact `sessionId`、`larkAppId`、`chatId`、`rootMessageId`、scope、工作目录原值与规范化路径、身份指纹，以及 `session.workingDir` 关联依据。目录包含关系使用路径段边界，`/work/a-other` 不属于 `/work/a`。存活祖先目录的符号链接会被规范化；悬空或被改变的别名不能被猜测为原工作区。 + +证据位于 `/workspace-recycle//`：`operation.json` 为协调记录，按 Bot/session 键散列命名的 JSON 为各 daemon 的持久关闭记录。写入使用锁、临时文件、原子 rename 与 fsync；日志不包含 prompt、原始 transcript、附件或凭据。标准关闭继续保留历史会话记录，并沿用原有临时资源清理行为。 + +每个目标分别记录 `before` / `after`：活跃注册、worker 端口、具有出生身份的 worker/CLI/可发现子进程、RSS、FD、Linux inotify 实例与 watch 数，以及持久后端探测结果。PID 消失和 PID 被复用都会与原进程仍存活区分；不会对复用 PID 或共享进程补发终止信号。 + +| 状态 | 含义 | +| --- | --- | +| `prepared` | 精确目标已持久化，尚未关闭 | +| `deferred` / 聚合 `pending` | 当前会话已持久交接，尚未验证关闭 | +| `closed` | durable closed、退出活跃注册，所观测的所属资源释放验证通过 | +| `closed_with_residual` | 本地会话关闭，但资源仍存在、远端残留或无法完成资源证明 | +| `blocked` / 聚合 `partial` | 有明确失败/覆盖缺口；成功的兄弟目标不会被回滚或重复关闭 | +| `aborted` | 工作区回收失败，未触发关闭 | + +退出码：`0` 已准备/已验证/已终止；`1` 部分失败或 blocker；`2` 参数/事件/读取错误；`3` 延迟交接待回读。调用者必须同时检查 `status`，不能把 `0` 一概解释为资源释放。 + +Linux 上的 inotify 配额是每用户的资源限制，不能将实例数与 Botmux 会话数等同。单次资源前后对比不能证明历史故障根因。当前资源观测覆盖采样时可归属的进程和已冻结的持久后端；不宣称枚举任意已脱离进程树的外部进程。其他平台无法验证子进程集合时会保留残留说明，远端 Mojo/Riff 的拒绝与隔离残留直接沿用标准关闭结果。 + +## 验证 + +```sh +bun run build +bun run test -- test/workspace-recycle.test.ts \ + test/workspace-recycle.integration.test.ts test/workspace-recycle-ipc.test.ts +``` + +测试使用临时目录和自建子进程。覆盖多 Bot/多群、目录边界与别名、外部会话排除、忙碌输入与排空锁、失败回收、部分失败恢复、关闭回执丢失、残留、当前会话最后退出及重复创建/回收。真实进程夹具创建自己的 HTTP 监听和 watcher，走标准关闭后验证 PID/注册/历史；不会连接真实飞书会话或关闭开发机现有 worker。 diff --git a/src/cli.ts b/src/cli.ts index 9d5def21f..fb5db4c21 100644 --- a/src/cli.ts +++ b/src/cli.ts @@ -6195,6 +6195,8 @@ async function cmdResume(): Promise { console.error('❌ 该静默定时轮次未创建话题，隐藏会话只保留审计记录，不能 resume。'); } else if (errCode === 'resume_cancelled') { console.error('❌ 恢复过程中会话被关闭，本次 resume 已取消。'); + } else if (errCode === 'workspace_retired') { + console.error('❌ 该会话的工作区已回收，无法恢复。请在有效工作区创建新会话。'); } else { console.error(`❌ 恢复失败: ${errCode}`); } @@ -6331,6 +6333,8 @@ botmux v${getVersion()} — IM ↔ AI 编程 CLI 桥接 在宿主终端注册、查看或清除 desktop device 凭证（AI CLI 会话内拒绝） actor current --json 返回当前 BotMux turn 的已验证企业用户名，不暴露 open_id/邮箱；脱离当前进程树时拒绝 + workspace-recycle discover|prepare|finish|status|hook + 按精确工作区发现、回收会话并读取资源验证结果 mojo-containment list|revoke 查看 / 显式撤销无法自证静止的 mojo containment handle（设备隔离 blocker 的可审计操作员出口；revoke 需 --yes，存活证据需 --force） @@ -14893,6 +14897,11 @@ switch (command) { process.exitCode = await runMojoContainmentCommand(process.argv.slice(3)); break; } + case 'workspace-recycle': { + const { runWorkspaceRecycleCommand } = await import('./cli/workspace-recycle.js'); + process.exitCode = await runWorkspaceRecycleCommand(process.argv.slice(3)); + break; + } case 'list': case 'ls': await cmdList(); break; case '__zmx-attach-managed': cmdManagedZmxAttach(process.argv.slice(3)); break; diff --git a/src/cli/workspace-recycle.ts b/src/cli/workspace-recycle.ts new file mode 100644 index 000000000..d1eec13f8 --- /dev/null +++ b/src/cli/workspace-recycle.ts @@ -0,0 +1,79 @@ +import { readFileSync } from 'node:fs'; +import { WorkspaceRecycler } from '../services/workspace-recycle.js'; +import { resolveBotmuxDataDir } from '../core/data-dir.js'; +import { resolveSessionContext } from '../core/session-marker.js'; +import { WORKSPACE_RECYCLE_PROTOCOL } from '../core/workspace-recycle-model.js'; + +const USAGE = `botmux workspace-recycle discover --workspace +botmux workspace-recycle prepare --workspace --operation [--initiator ] +botmux workspace-recycle finish --operation --event --outcome succeeded|failed +botmux workspace-recycle status --operation +botmux workspace-recycle hook --event-file + +discover is read-only (including missing legacy workspaces). prepare captures +exact targets while the directory exists. finish requires a successful generic +reclamation event AND a missing old directory. A failed event closes nothing. +The optional hook consumes botmux.workspace-recycle.v1 JSON from a generic +lifecycle adapter; Botmux must be installed outside the reclaimed directory. +JSON output; exit 0 = verified/prepared, 1 = partial/blocked, 2 = invalid input, +3 = durable initiator handoff pending (poll status). No force/global-delete mode.`; + +export async function runWorkspaceRecycleCommand(args: string[]): Promise { + if (args.length === 0 || args.includes('--help')) { console.log(USAGE); return 0; } + try { + const [action, ...rest] = args; + const options = new Map(); + for (let i = 0; i < rest.length; i += 2) { + if (!rest[i].startsWith('--') || !rest[i + 1] || rest[i + 1].startsWith('--') || options.has(rest[i])) throw new Error('invalid_or_duplicate_option'); + options.set(rest[i], rest[i + 1]); + } + const allowed: Record = { + discover: ['--workspace'], prepare: ['--workspace', '--operation', '--initiator'], + finish: ['--operation', '--event', '--outcome'], status: ['--operation'], hook: ['--event-file'], + }; + if (!allowed[action] || [...options.keys()].some(key => !allowed[action].includes(key))) throw new Error('unknown_recycle_action_or_option'); + const required = (key: string): string => { + const value = options.get(key); + if (!value) throw new Error(`required:${key}`); + return value; + }; + const dataDir = resolveBotmuxDataDir(); + const recycler = new WorkspaceRecycler({ dataDir }); + const initiator = (workspace: string): string | undefined => { + const explicit = options.get('--initiator'); + if (explicit) return explicit; + const current = resolveSessionContext(dataDir, process.env.BOTMUX_SESSION_ID)?.sessionId; + return current && recycler.discover(workspace).targets.some(target => target.sessionId === current) ? current : undefined; + }; + let result: unknown; + if (action === 'discover') { + const discovery = recycler.discover(required('--workspace')); + result = { ok: discovery.errors.length === 0, dryRun: true, ...discovery }; + } else if (action === 'prepare') { + result = await recycler.prepare(required('--operation'), required('--workspace'), initiator(required('--workspace'))); + } else if (action === 'status') { + result = recycler.status(required('--operation')); + } else if (action === 'finish') { + const outcome = required('--outcome'); + if (outcome !== 'succeeded' && outcome !== 'failed') throw new Error('invalid_recycle_outcome'); + result = await recycler.finish(required('--operation'), { eventId: required('--event'), outcome }); + } else { + const eventFile = required('--event-file'); + const event = JSON.parse(readFileSync(eventFile === '-' ? 0 : eventFile, 'utf8')); + if (event.protocol !== WORKSPACE_RECYCLE_PROTOCOL || typeof event.operationId !== 'string') throw new Error('invalid_hook_event'); + if (event.phase === 'before-reclaim') { + if (typeof event.workspacePath !== 'string' || (event.initiatorSessionId !== undefined && typeof event.initiatorSessionId !== 'string')) throw new Error('invalid_hook_workspace'); + result = await recycler.prepare(event.operationId, event.workspacePath, event.initiatorSessionId ?? initiator(event.workspacePath)); + } else if (event.phase === 'after-reclaim') { + if (typeof event.eventId !== 'string' || !['succeeded', 'failed'].includes(event.outcome)) throw new Error('invalid_hook_outcome'); + result = await recycler.finish(event.operationId, { eventId: event.eventId, outcome: event.outcome }); + } else throw new Error('invalid_hook_phase'); + } + console.log(JSON.stringify(result, null, 2)); + const outcome = result as { ok: boolean; status?: string }; + return outcome.status === 'pending' ? 3 : outcome.ok ? 0 : 1; + } catch (error) { + console.log(JSON.stringify({ ok: false, error: String(error) })); + return 2; + } +} diff --git a/src/core/command-handler.ts b/src/core/command-handler.ts index 7fb01a6b8..0ce135bb4 100644 --- a/src/core/command-handler.ts +++ b/src/core/command-handler.ts @@ -3382,6 +3382,8 @@ export async function handleCommand( await sessionReply(rootId, t('card.action.resume_deferred_unmaterialized', undefined, loc)); } else if (result.error === 'resume_cancelled') { await sessionReply(rootId, t('card.action.resume_cancelled', undefined, loc)); + } else if (result.error === 'workspace_retired') { + await sessionReply(rootId, t('card.action.resume_workspace_retired', undefined, loc)); } else { await sessionReply(rootId, t('cmd.adopt.resume_not_found', undefined, loc)); } diff --git a/src/core/dashboard-ipc-server.ts b/src/core/dashboard-ipc-server.ts index a302968bd..e443cdf49 100644 --- a/src/core/dashboard-ipc-server.ts +++ b/src/core/dashboard-ipc-server.ts @@ -162,6 +162,9 @@ import { DEFAULT_SESSION_OWNER_REMINDER } from './session-owner-reminder.js'; import { updateSessionOwnerReminderConfig } from '../services/session-owner-reminder-config-store.js'; import { sendSessionOwnerThreadNotification } from '../services/session-owner-notification.js'; import { matchesExpectedSessionLocateScope, type SessionLocateExpectedScope } from './session-locate-guard.js'; +import { WorkspaceRecycleRuntime, type RecycleAction } from './workspace-recycle-runtime.js'; +import { validateRecycleRequest } from './workspace-recycle-journal.js'; +import { isSessionLifecycleInFlight, hasPendingOrdinaryImDelivery } from './worker-pool.js'; import { buildTerminalUrl } from './terminal-url.js'; import { dashboardEventBus } from './dashboard-events.js'; import { validateWorkingDir } from './working-dir.js'; @@ -1212,6 +1215,38 @@ ipcRoute('GET', '/api/sessions', (_req, res) => { jsonRes(res, 200, { sessions: composeDashboardSessionRows({ includeTokenUsage: false }) }); }); +// This route deliberately has NO session-capability/public allowlist entry. +// Recycling multiple sessions requires the existing trusted-host HMAC. +const workspaceRecycleRuntime = new WorkspaceRecycleRuntime({ + appId: () => cachedLarkAppId, + dataDir: () => config.session.dataDir, + getSession: id => { + sessionStore.listSessionsStrict(); + return sessionStore.getOwnedSession(id); + }, + getRuntime: findActiveBySessionId, + allSessions: () => sessionStore.loadAllSessionsStrict(config.session.dataDir), + close: closeSession, + retireClosed: (id, workspaceRetirement) => sessionStore.closeSession(id, { workspaceRetirement }), + lifecycleBusy: ds => isSessionTransferring(ds) || isSessionLifecycleInFlight(ds) || hasPendingOrdinaryImDelivery(ds), + closeResidual: session => mojoCloseResidualForRow(session)?.reason, + onError: error => logger.warn(`[workspace-recycle] deferred recovery failed: ${String(error)}`), +}); + +ipcRoute('POST', '/api/workspace-recycle/:action', async (req, res, params) => { + if (!ipcHmacAuthorized(req)) return jsonRes(res, 401, { ok: false, error: 'unauthorized' }); + if (!['prepare', 'close', 'defer', 'abort'].includes(params.action)) return jsonRes(res, 400, { ok: false, error: 'invalid_recycle_action' }); + const body = await readJsonBody(req); + try { validateRecycleRequest(body); } + catch (error) { return jsonRes(res, 400, { ok: false, error: String(error) }); } + try { + const result = await workspaceRecycleRuntime.perform(params.action as RecycleAction, body); + return jsonRes(res, result.status === 'deferred' ? 202 : result.ok ? 200 : 409, result); + } catch (error) { + return jsonRes(res, 409, { ok: false, error: String(error) }); + } +}); + ipcRoute('GET', '/api/sessions/:sessionId', (_req, res, params) => { const ds = findActiveBySessionId(params.sessionId); if (ds) return jsonRes(res, 200, { session: composeRowFromActive(ds) }); @@ -7590,6 +7625,12 @@ export function startIpcServer(opts: { log: (m) => logger.warn(`[dashboard-ipc] ${m}`), }).then((port) => { boundPort = port; + // Restored current-session handoffs wait for the normal session restore + // barrier. The controller never depends on the removed workspace/process. + workspaceRecycleRuntime.start(); + void (opts.ready ?? Promise.resolve()).then(() => workspaceRecycleRuntime.recoverDeferred()) + .catch(error => logger.warn(`[workspace-recycle] deferred recovery failed: ${String(error)}`)); + server.once('close', () => workspaceRecycleRuntime.stop()); return { port, close: () => new Promise(r => server.close(() => r())), diff --git a/src/core/session-manager.ts b/src/core/session-manager.ts index 454860ff4..c31cd29a4 100644 --- a/src/core/session-manager.ts +++ b/src/core/session-manager.ts @@ -2110,7 +2110,7 @@ export async function restoreActiveSessions( // Registration itself is CAS, so a fresh runtime occupant always wins over // every startup candidate regardless of this disk ordering. const active = sessions - .filter(s => s.status === 'active') + .filter(s => s.status === 'active' && !s.workspaceRetirement) // Idempotency quarantine (at-most-once): a session the boot reconcile just // terminalized as `dispatch_unknown` (or dropped as a pre-dispatch reserved // orphan) must NOT be re-attached — the poller now sees it `failed`, so @@ -3127,9 +3127,10 @@ export async function resumeSession( sessionId: string, activeSessions: Map, ): Promise<{ ok: true; ds: DaemonSession } -| { ok: false; error: 'not_found' | 'not_closed' | 'anchor_occupied' | 'adopt_unsupported' | 'deferred_unmaterialized' | 'resume_cancelled'; activeSessionId?: string }> { +| { ok: false; error: 'not_found' | 'not_closed' | 'anchor_occupied' | 'adopt_unsupported' | 'deferred_unmaterialized' | 'resume_cancelled' | 'workspace_retired'; activeSessionId?: string }> { let session = sessionStore.getSession(sessionId); if (!session) return { ok: false, error: 'not_found' }; + if (session.workspaceRetirement) return { ok: false, error: 'workspace_retired' }; if (session.status !== 'closed') return { ok: false, error: 'not_closed' }; // Plan B: a VC meeting agent is an ordinary chat-scope session, so a closed one @@ -3164,6 +3165,7 @@ export async function resumeSession( // durable row and keep the already-active creator as first owner. const latest = sessionStore.getSession(sessionId); if (!latest) return { ok: false as const, error: 'not_found' as const }; + if (latest.workspaceRetirement) return { ok: false as const, error: 'workspace_retired' as const }; if (latest.status !== 'closed') return { ok: false as const, error: 'not_closed' as const }; if (latest.deferredScheduleRun && !readDeferredTopicBinding(config.session.dataDir, latest.sessionId)) { diff --git a/src/core/worker-pool.ts b/src/core/worker-pool.ts index c28570638..f6be19dfb 100644 --- a/src/core/worker-pool.ts +++ b/src/core/worker-pool.ts @@ -6700,7 +6700,7 @@ export async function closeSessionForBackgroundCleanup( export async function closeSession( sessionId: string, - opts?: { awaitWorkerExit?: boolean }, + opts?: { awaitWorkerExit?: boolean; workspaceRetirement?: Session['workspaceRetirement'] }, ): Promise { // `awaitWorkerExit` (default true): whether to block on the worker process // actually exiting before returning. A busy CLI wedges in node-pty teardown @@ -6830,6 +6830,7 @@ export async function closeSession( // earlier — one layer up. sessionStore.closeSession(sessionId, { cleanupBridgeMarkers: !hadLiveWorker, + ...(opts?.workspaceRetirement ? { workspaceRetirement: opts.workspaceRetirement } : {}), ...(prepared.parkMojoLineage ? { parkMojoLineage: prepared.parkMojoLineage } : {}), // Park a LOCAL residual so an idempotent re-close still reports it — the // journal (its runtime home) is wiped by this same transaction. @@ -6882,6 +6883,7 @@ export async function closeSession( // SUCCESSFUL save, and skipped when the two are the same object anyway, so // the runtime view cannot end up carrying a park the disk does not have. if (after && after !== ds.session) { + ds.session.workspaceRetirement = after.workspaceRetirement; ds.session.mojoCloseJournal = after.mojoCloseJournal; if (clearMojoLineage || prepared.parkMojoLineage) { ds.session.riffParentTaskId = after.riffParentTaskId; @@ -7238,7 +7240,7 @@ function removeInactiveRegistration( key: string, ds: DaemonSession, ): boolean { - if (ds.session.status === 'active') return false; + if (ds.session.status === 'active' && !ds.session.workspaceRetirement) return false; // Only remove our exact stale object. A newer session may already own the // same routing key and must never be evicted by this continuation. if (map.get(key) === ds) map.delete(key); @@ -7549,6 +7551,15 @@ type OrdinaryImDelivery = { * this in-memory delivery without retry would permanently drop the message. */ const pendingOrdinaryImDeliveries = new Map(); +/** A cached idle screen does not prove that a just-admitted message has been + * committed by the worker. Automatic workspace recycling must preserve it. */ +export function hasPendingOrdinaryImDelivery(ds: DaemonSession): boolean { + for (const delivery of pendingOrdinaryImDeliveries.values()) { + if (delivery.ds === ds) return true; + } + return false; +} + function ordinaryImDeliveryKey(ds: DaemonSession, turnId: string, workerGeneration: number): string { return `${ds.session.sessionId}:${workerGeneration}:${turnId}`; } diff --git a/src/core/workspace-recycle-journal.ts b/src/core/workspace-recycle-journal.ts new file mode 100644 index 000000000..3da04fb76 --- /dev/null +++ b/src/core/workspace-recycle-journal.ts @@ -0,0 +1,90 @@ +import { mkdirSync, readFileSync } from 'node:fs'; +import { createHash } from 'node:crypto'; +import { join, dirname } from 'node:path'; +import { atomicWriteFileSync } from '../utils/atomic-write.js'; +import { canonicalWorkspacePath, pathInside, type WorkspaceIdentity } from './workspace-recycle-path.js'; +import { WORKSPACE_RECYCLE_PROTOCOL, type WorkspaceSessionTarget } from './workspace-recycle-model.js'; +import type { RecycleResources } from './workspace-recycle-resources.js'; +import type { CloseSessionResult } from './worker-pool.js'; + +export interface RecyclePeer { larkAppId: string; sessionId: string } +export interface RecycleRequest { + protocol: typeof WORKSPACE_RECYCLE_PROTOCOL; + operationId: string; + workspace: WorkspaceIdentity; + target: WorkspaceSessionTarget; + peers: RecyclePeer[]; + initiator?: RecyclePeer; +} + +export interface RecycleJournal extends RecycleRequest { + phase: 'prepared' | 'closing' | 'deferred' | 'closed' | 'blocked' | 'closed_with_residual' | 'aborted'; + inputStamp: string; + before: RecycleResources; + after?: RecycleResources; + closeResult?: CloseSessionResult; + blockers: string[]; + preparedAt: string; + updatedAt: string; + deferredUntil?: number; +} + +export function recycleKey(peer: RecyclePeer): string { + return createHash('sha256').update(JSON.stringify([peer.larkAppId, peer.sessionId])).digest('hex'); +} + +export function validateRecycleOperationId(value: string): void { + if (typeof value !== 'string' || !/^[a-zA-Z0-9][a-zA-Z0-9._-]{0,127}$/.test(value)) throw new Error('invalid_operation_id'); +} + +export function recycleJournalPath(dataDir: string, operationId: string, peer: RecyclePeer): string { + validateRecycleOperationId(operationId); + return join(dataDir, 'workspace-recycle', operationId, `${recycleKey(peer)}.json`); +} + +export function assertRecycleStateOutsideWorkspace(dataDir: string, workspace: WorkspaceIdentity): void { + if (pathInside(workspace.canonicalPath, canonicalWorkspacePath(dataDir))) throw new Error('recycle_state_inside_workspace'); +} + +export function persistRecycleJson(path: string, value: unknown): void { + mkdirSync(dirname(path), { recursive: true, mode: 0o700 }); + atomicWriteFileSync(path, `${JSON.stringify(value, null, 2)}\n`, { durable: true, mode: 0o600, followTargetSymlink: false }); +} + +export function readRecycleJournal(path: string): RecycleJournal | undefined { + let parsed: RecycleJournal; + try { parsed = JSON.parse(readFileSync(path, 'utf8')); } + catch (error) { + if ((error as NodeJS.ErrnoException).code === 'ENOENT') return undefined; + throw error; + } + validateRecycleRequest(parsed); + if (!parsed.before || !Array.isArray(parsed.before.processes) || typeof parsed.inputStamp !== 'string' + || !Array.isArray(parsed.blockers) || !['prepared', 'closing', 'deferred', 'closed', 'blocked', 'closed_with_residual', 'aborted'].includes(parsed.phase)) { + throw new Error('invalid_recycle_journal'); + } + return parsed; +} + +export function validateRecycleRequest(value: unknown): asserts value is RecycleRequest { + const r = value as RecycleRequest; + if (!r || r.protocol !== WORKSPACE_RECYCLE_PROTOCOL) throw new Error('unsupported_recycle_protocol'); + validateRecycleOperationId(r.operationId); + if (!r.workspace || typeof r.workspace.device !== 'string' || typeof r.workspace.inode !== 'string') throw new Error('invalid_workspace_identity'); + canonicalWorkspacePath(r.workspace.path); + canonicalWorkspacePath(r.workspace.canonicalPath); + const t = r.target; + if (!t || ['sessionId', 'larkAppId', 'chatId', 'rootMessageId', 'createdAt', 'workingDir', 'canonicalWorkingDir'].some(key => typeof t[key as keyof WorkspaceSessionTarget] !== 'string') + || !/^[a-f0-9]{64}$/.test(t.fingerprint) || t.evidence !== 'session.workingDir' || !['chat', 'thread'].includes(t.scope)) throw new Error('invalid_recycle_target'); + if (!pathInside(r.workspace.canonicalPath, t.canonicalWorkingDir)) throw new Error('target_outside_workspace'); + if (!Array.isArray(r.peers) || r.peers.length > 10_000 || r.peers.some(p => !p || typeof p.sessionId !== 'string' || typeof p.larkAppId !== 'string') + || new Set(r.peers.map(recycleKey)).size !== r.peers.length + || !r.peers.some(p => recycleKey(p) === recycleKey(t))) throw new Error('invalid_recycle_peers'); + if (r.initiator && !r.peers.some(p => recycleKey(p) === recycleKey(r.initiator!))) throw new Error('initiator_not_in_plan'); +} + +export function sameRecycleRequest(a: RecycleRequest, b: RecycleRequest): boolean { + return a.operationId === b.operationId && JSON.stringify(a.workspace) === JSON.stringify(b.workspace) + && a.target.fingerprint === b.target.fingerprint && recycleKey(a.target) === recycleKey(b.target) + && JSON.stringify(a.peers) === JSON.stringify(b.peers) && JSON.stringify(a.initiator) === JSON.stringify(b.initiator); +} diff --git a/src/core/workspace-recycle-model.ts b/src/core/workspace-recycle-model.ts new file mode 100644 index 000000000..1b3f6fd43 --- /dev/null +++ b/src/core/workspace-recycle-model.ts @@ -0,0 +1,87 @@ +import { createHash } from 'node:crypto'; +import type { Session } from '../types.js'; +import { canonicalWorkspacePath, pathInside, type WorkspaceIdentity } from './workspace-recycle-path.js'; + +export const WORKSPACE_RECYCLE_PROTOCOL = 'botmux.workspace-recycle.v1'; + +/** Only identity/ownership fields. No prompts, credentials, or transcript. */ +export interface WorkspaceSessionTarget { + sessionId: string; + larkAppId: string; + chatId: string; + rootMessageId: string; + scope: 'chat' | 'thread'; + createdAt: string; + workingDir: string; + canonicalWorkingDir: string; + backendType?: Session['backendType']; + persistentBackendTarget?: Session['persistentBackendTarget']; + cliSessionId?: string; + fingerprint: string; + evidence: 'session.workingDir'; +} + +export function targetFingerprint(session: Session): string { + return createHash('sha256').update(JSON.stringify([ + session.sessionId, session.larkAppId, session.chatId, session.rootMessageId, + session.scope ?? 'thread', session.createdAt, session.workingDir, + session.backendType ?? null, session.persistentBackendTarget ?? null, + session.cliSessionId ?? null, session.adoptedFrom ?? null, + session.existingAppServerEndpoint ?? null, + ])).digest('hex'); +} + +export function workspaceTarget(session: Session, workspace: Pick): WorkspaceSessionTarget | undefined { + if (!session.workingDir) return undefined; + const canonicalWorkingDir = canonicalWorkspacePath(session.workingDir); + if (!pathInside(workspace.canonicalPath, canonicalWorkingDir)) return undefined; + if (!session.larkAppId || !session.sessionId || !session.chatId || !session.createdAt) { + throw new Error('session_owner_or_identity_missing'); + } + return { + sessionId: session.sessionId, larkAppId: session.larkAppId, + chatId: session.chatId, rootMessageId: session.rootMessageId, + scope: session.scope ?? 'thread', createdAt: session.createdAt, + workingDir: session.workingDir, canonicalWorkingDir, + backendType: session.backendType, + persistentBackendTarget: session.persistentBackendTarget, + cliSessionId: session.cliSessionId, + fingerprint: targetFingerprint(session), evidence: 'session.workingDir', + }; +} + +export interface WorkspaceDiscovery { + protocol: typeof WORKSPACE_RECYCLE_PROTOCOL; + workspace: { path: string; canonicalPath: string }; + targets: WorkspaceSessionTarget[]; + excluded: Array<{ sessionId: string; larkAppId?: string; reason: string }>; + errors: Array<{ sessionId: string; error: string }>; +} + +export function discoverWorkspaceSessions(sessions: Session[], path: string): WorkspaceDiscovery { + const workspace = { path, canonicalPath: canonicalWorkspacePath(path) }; + const result: WorkspaceDiscovery = { protocol: WORKSPACE_RECYCLE_PROTOCOL, workspace, targets: [], excluded: [], errors: [] }; + const counts = new Map(); + for (const session of sessions) counts.set(session.sessionId, (counts.get(session.sessionId) ?? 0) + 1); + for (const session of sessions) { + try { + const target = workspaceTarget(session, workspace); + if (!target) continue; + if (counts.get(session.sessionId)! > 1) { + result.errors.push({ sessionId: session.sessionId, error: 'ambiguous_session_owner' }); + } else if (session.adoptedFrom || session.existingAppServerEndpoint) { + result.excluded.push({ sessionId: session.sessionId, larkAppId: session.larkAppId, reason: 'external_or_shared_session' }); + } else if (session.status === 'closed') { + result.excluded.push({ sessionId: session.sessionId, larkAppId: session.larkAppId, reason: 'already_closed' }); + } else { + result.targets.push(target); + } + } catch (error) { + // A failed ownership read is a coverage gap, even if the row would have + // turned out unrelated. Never turn an unreadable path into an empty set. + result.errors.push({ sessionId: session.sessionId, error: String(error) }); + } + } + result.targets.sort((a, b) => a.larkAppId.localeCompare(b.larkAppId) || a.sessionId.localeCompare(b.sessionId)); + return result; +} diff --git a/src/core/workspace-recycle-path.ts b/src/core/workspace-recycle-path.ts new file mode 100644 index 000000000..04b0f2a9d --- /dev/null +++ b/src/core/workspace-recycle-path.ts @@ -0,0 +1,63 @@ +import { lstatSync, realpathSync, statSync } from 'node:fs'; +import { basename, dirname, isAbsolute, join, parse, relative, resolve, sep } from 'node:path'; + +/** Resolve surviving ancestors too: /home -> /data/home still matters after + * the task directory itself has been removed. Permission/loop errors are NOT + * equivalent to a missing directory. */ +export function canonicalWorkspacePath(value: string): string { + if (typeof value !== 'string' || !isAbsolute(value) || value.includes('\0')) { + throw new Error('workspace_path_must_be_absolute'); + } + let cursor = resolve(value); + const suffix: string[] = []; + for (;;) { + try { return join(realpathSync(cursor), ...suffix.reverse()); } + catch (error) { + if ((error as NodeJS.ErrnoException).code !== 'ENOENT') throw error; + // A dangling symlink has unknown ownership, not an ordinary missing tail. + try { + if (lstatSync(cursor).isSymbolicLink()) throw new Error('dangling_workspace_alias'); + } catch (statError) { + if ((statError as NodeJS.ErrnoException).code !== 'ENOENT') throw statError; + } + const parent = dirname(cursor); + if (parent === cursor) throw error; + suffix.push(basename(cursor)); + cursor = parent; + } + } +} + +export function pathInside(root: string, candidate: string): boolean { + const child = relative(root, candidate); + return child === '' || (child !== '..' && !child.startsWith(`..${sep}`) && !isAbsolute(child)); +} + +export interface WorkspaceIdentity { + path: string; + canonicalPath: string; + device: string; + inode: string; +} + +export function captureWorkspace(path: string): WorkspaceIdentity { + const canonicalPath = canonicalWorkspacePath(path); + if (canonicalPath === parse(canonicalPath).root) throw new Error('filesystem_root_is_not_a_workspace'); + const stat = statSync(canonicalPath, { bigint: true }); + if (!stat.isDirectory()) throw new Error('workspace_is_not_a_directory'); + return { path: resolve(path), canonicalPath, device: String(stat.dev), inode: String(stat.ino) }; +} + +/** Removal is an additional check on a successful lifecycle event, never the + * authority to discover/close arbitrary old sessions. */ +export function assertWorkspaceRemoved(workspace: WorkspaceIdentity): void { + for (const path of new Set([workspace.path, workspace.canonicalPath])) { + try { + lstatSync(path); + } catch (error) { + if ((error as NodeJS.ErrnoException).code === 'ENOENT') continue; + throw error; + } + throw new Error('workspace_still_exists_or_path_reused'); + } +} diff --git a/src/core/workspace-recycle-resources.ts b/src/core/workspace-recycle-resources.ts new file mode 100644 index 000000000..a1ee8ebc9 --- /dev/null +++ b/src/core/workspace-recycle-resources.ts @@ -0,0 +1,117 @@ +import { readdirSync, readFileSync, readlinkSync } from 'node:fs'; +import type { Session } from '../types.js'; +import type { DaemonSession } from './types.js'; +import { readDurableProcessIdentity, isLinuxZombie } from '../utils/process-identity.js'; +import { isProcessAlive } from './session-liveness.js'; +import { parseProcessStat } from './resource-monitor/procfs.js'; +import { isSuspendableBackendType, probePersistentBackendTarget, resolvePersistentBackendTarget } from './persistent-backend.js'; + +export interface RecycleProcess { + pid: number; + identity?: string; + source: 'worker' | 'cli' | 'descendant'; + state: 'alive' | 'gone' | 'unknown' | 'reused'; + rssBytes?: number; + fileDescriptors?: number; + inotifyInstances?: number; + inotifyWatches?: number; +} + +export interface RecycleResources { + registered: boolean; + workerPort?: number; + processes: RecycleProcess[]; + backing: { type?: Session['backendType']; state: 'missing' | 'exists' | 'unknown' | 'not_applicable' | 'remote' }; + errors: string[]; +} + +export function sampleRecycleProcess(pid: number, source: RecycleProcess['source'], expectedIdentity?: string): RecycleProcess { + const identity = readDurableProcessIdentity(pid); + const state = !isProcessAlive(pid) || isLinuxZombie(pid) ? 'gone' + : !identity ? 'unknown' : expectedIdentity && expectedIdentity !== identity ? 'reused' : 'alive'; + const result: RecycleProcess = { pid, identity: expectedIdentity ?? identity, source, state }; + if (state !== 'alive' || process.platform !== 'linux') return result; + try { + const rss = readFileSync(`/proc/${pid}/status`, 'utf8').match(/^VmRSS:\s+(\d+)\s+kB/m); + result.rssBytes = rss ? Number(rss[1]) * 1024 : undefined; + const fds = readdirSync(`/proc/${pid}/fd`); + result.fileDescriptors = fds.length; + let instances = 0; + let watches = 0; + for (const fd of fds) { + try { + if (!readlinkSync(`/proc/${pid}/fd/${fd}`).includes('inotify')) continue; + instances++; + watches += readFileSync(`/proc/${pid}/fdinfo/${fd}`, 'utf8').split('\n').filter(line => line.startsWith('inotify ')).length; + } catch { /* descriptors can close during a sample */ } + } + result.inotifyInstances = instances; + result.inotifyWatches = watches; + } catch { /* metrics unavailable does not prove process absence */ } + return result; +} + +/** Read-only attribution. Nothing here sends a signal other than kill(0), + * probes a shared process for termination, or deletes an artifact. */ +export function captureRecycleResources(session: Session, ds?: DaemonSession): RecycleResources { + const processes: RecycleProcess[] = []; + const errors: string[] = []; + const workerPid = ds?.worker?.pid ?? session.pid; + if (workerPid) processes.push(sampleRecycleProcess(workerPid, 'worker')); + const attestation = ds?.localProcessAttestation; + if (attestation?.cliPid && !processes.some(p => p.pid === attestation.cliPid)) { + const cli = sampleRecycleProcess(attestation.cliPid, 'cli'); + // Attestation is bound to this worker generation. Refuse a reused CLI PID. + if (attestation.cliProcStart && cli.identity?.endsWith(`:${attestation.cliProcStart}`) === false + && cli.identity !== attestation.cliProcStart) errors.push('cli_process_identity_changed'); + else processes.push(cli); + } + if (process.platform === 'linux') { + try { + const rows = readdirSync('/proc').filter(name => /^\d+$/.test(name)).flatMap(name => { + try { const row = parseProcessStat(readFileSync(`/proc/${name}/stat`, 'utf8')); return row ? [row] : []; } + catch { return []; } + }); + const owned = new Set(processes.filter(p => p.state === 'alive').map(p => p.pid)); + let changed = true; + while (changed) { + changed = false; + for (const row of rows) { + if (!owned.has(row.pid) && owned.has(row.ppid)) { + owned.add(row.pid); + processes.push(sampleRecycleProcess(row.pid, 'descendant')); + changed = true; + } + } + } + } catch { errors.push('process_descendants_unavailable'); } + } else if (processes.some(p => p.state === 'alive')) { + errors.push('process_descendants_unavailable_on_platform'); + } + const type = ds?.initConfig?.backendType ?? session.backendType; + let state: RecycleResources['backing']['state'] = 'not_applicable'; + if (isSuspendableBackendType(type)) { + try { state = probePersistentBackendTarget(resolvePersistentBackendTarget(type, session.sessionId, session.persistentBackendTarget)); } + catch { state = 'unknown'; } + } else if (type === 'mojo' || type === 'riff') state = 'remote'; + else if (!type) state = 'unknown'; + return { registered: !!ds, workerPort: ds?.workerPort ?? undefined, processes, backing: { type, state }, errors }; +} + +export function rereadRecycleResources(before: RecycleResources, session: Session, ds?: DaemonSession): RecycleResources { + // A stored PID can already belong to an unrelated process after close. + // Only a live runtime may introduce new owned processes at readback. + const after = captureRecycleResources({ ...session, pid: undefined }, ds); + const observed = before.processes.map(p => sampleRecycleProcess(p.pid, p.source, p.identity)); + for (const p of after.processes) if (!observed.some(old => old.pid === p.pid)) observed.push(p); + return { ...after, processes: observed, errors: [...new Set([...before.errors, ...after.errors])] }; +} + +export function resourceResiduals(resources: RecycleResources): string[] { + return [ + ...(resources.registered ? ['active_registration_remains'] : []), + ...resources.processes.filter(p => p.state === 'alive' || p.state === 'unknown').map(p => `${p.source}_${p.state}:${p.pid}`), + ...(['exists', 'unknown'].includes(resources.backing.state) ? [`backing_${resources.backing.state}`] : []), + ...resources.errors, + ]; +} diff --git a/src/core/workspace-recycle-runtime.ts b/src/core/workspace-recycle-runtime.ts new file mode 100644 index 000000000..9e2ce4dd1 --- /dev/null +++ b/src/core/workspace-recycle-runtime.ts @@ -0,0 +1,315 @@ +import { mkdirSync, readdirSync } from 'node:fs'; +import { dirname, join } from 'node:path'; +import { createHash } from 'node:crypto'; +import type { Session } from '../types.js'; +import type { DaemonSession } from './types.js'; +import type { CloseSessionResult } from './worker-pool.js'; +import { withFileLock } from '../utils/file-lock.js'; +import { tryWithBotTurnMutation } from './bot-turn-mutation-gate.js'; +import { protectedSessionMutationReasons } from './session-mutation-guard.js'; +import { hasPendingSessionTurns } from './session-turn-queue.js'; +import { assertWorkspaceRemoved, canonicalWorkspacePath, captureWorkspace } from './workspace-recycle-path.js'; +import { discoverWorkspaceSessions, targetFingerprint } from './workspace-recycle-model.js'; +import { + assertRecycleStateOutsideWorkspace, persistRecycleJson, readRecycleJournal, + recycleJournalPath, recycleKey, sameRecycleRequest, validateRecycleRequest, + type RecycleJournal, type RecycleRequest, +} from './workspace-recycle-journal.js'; +import { + captureRecycleResources, rereadRecycleResources, resourceResiduals, + type RecycleResources, +} from './workspace-recycle-resources.js'; + +export interface WorkspaceRecycleRuntimeDeps { + appId: () => string; + dataDir: () => string; + getSession: (id: string) => Session | undefined; + getRuntime: (id: string) => DaemonSession | undefined; + allSessions: () => Session[]; + close: (id: string, opts: { workspaceRetirement: NonNullable }) => Promise; + /** Synchronous durable re-close only: attach retirement to a row that was + * independently closed after prepare, without retrying process teardown. */ + retireClosed: (id: string, retirement: NonNullable) => void; + lifecycleBusy: (ds: DaemonSession) => boolean; + closeResidual: (session: Session) => string | undefined; + capture?: typeof captureRecycleResources; + reread?: typeof rereadRecycleResources; + now?: () => number; + onError?: (error: unknown) => void; +} + +export type RecycleAction = 'prepare' | 'close' | 'defer' | 'abort'; +export interface RecycleTargetResult { + ok: boolean; + status: RecycleJournal['phase']; + journal: RecycleJournal; +} + +function inputStamp(session: Session, ds?: DaemonSession): string { + // Status/output observations deliberately do not count as NEW INPUT. + return createHash('sha256').update(JSON.stringify([ + session.lastMessageAt ?? null, ds?.lastMessageAt ?? null, + ds?.streamCardTurnGeneration ?? null, ds?.workerGeneration ?? null, ds?.currentTurnId ?? null, + ])).digest('hex'); +} + +export function recycleBusyReasons(session: Session, ds?: DaemonSession): string[] { + const reasons: string[] = protectedSessionMutationReasons(ds ?? session); + if (hasPendingSessionTurns(session.sessionId)) reasons.push('session_turn_pending'); + if (session.ordinaryTurnRecovery && ['running', 'backoff', 'dispatching'].includes(session.ordinaryTurnRecovery.status)) reasons.push('ordinary_turn_unsettled'); + if (!ds) { + if (session.pid) reasons.push('unregistered_worker'); + return reasons; + } + if (ds.worker && ds.lastScreenStatus !== 'idle') reasons.push('worker_not_idle'); + if (ds.managedTurnOrigin?.turnId) reasons.push('managed_turn_not_terminal'); + if (ds.pendingRepo || ds.pendingRepoCommitInFlight) reasons.push('repository_setup'); + if (ds.pendingPrompt || ds.pendingRawInput || ds.pendingFollowUpInput + || ds.pendingFollowUps?.length || ds.pendingAttachments?.length) reasons.push('buffered_input'); + if (ds.pendingWaitPromises?.size) reasons.push('pending_reply'); + if (ds.idempotentAsyncTurns?.size || [...(ds.asyncTriggerResults?.values() ?? [])].some(t => t.status === 'pending')) reasons.push('async_trigger_pending'); + if (ds.tuiPromptProcessing || ds.stuckWarningProcessing) reasons.push('interactive_input_pending'); + return [...new Set(reasons)]; +} + +/** Host-only per-target close, under the SAME admission/drain gate used by + * ordinary close. All decisions are made after earlier inbound turns drain. + * Journals live in the daemon data directory, never in the retiring cwd. */ +export class WorkspaceRecycleRuntime { + private timer?: ReturnType; + private stopped = false; + constructor(private readonly deps: WorkspaceRecycleRuntimeDeps) {} + + private now(): number { return (this.deps.now ?? Date.now)(); } + private path(request: RecycleRequest): string { + return recycleJournalPath(this.deps.dataDir(), request.operationId, request.target); + } + private save(journal: RecycleJournal): void { + journal.updatedAt = new Date(this.now()).toISOString(); + persistRecycleJson(this.path(journal), journal); + } + private result(journal: RecycleJournal): RecycleTargetResult { + return { ok: journal.blockers.length === 0 && ['prepared', 'closed'].includes(journal.phase), status: journal.phase, journal }; + } + private block(journal: RecycleJournal, blockers: string[]): RecycleTargetResult { + journal.phase = 'blocked'; + journal.blockers = blockers; + this.save(journal); + return this.result(journal); + } + private isInitiator(request: RecycleRequest): boolean { + return !!request.initiator && recycleKey(request.initiator) === recycleKey(request.target); + } + + private retirement(request: RecycleRequest): NonNullable { + return { operationId: request.operationId, workspacePath: request.workspace.canonicalPath, + retiredAt: new Date(this.now()).toISOString() }; + } + + private checkIdentity(request: RecycleRequest, session: Session, ds?: DaemonSession): string[] { + const blockers: string[] = []; + if (session.larkAppId !== this.deps.appId() || targetFingerprint(session) !== request.target.fingerprint) blockers.push('session_identity_or_location_changed'); + if (ds && (ds.larkAppId !== request.target.larkAppId || ds.chatId !== request.target.chatId + || (ds.workingDir && ds.workingDir !== session.workingDir))) blockers.push('runtime_identity_or_location_changed'); + if (session.adoptedFrom || session.existingAppServerEndpoint || ds?.adoptedFrom || ds?.initConfig?.adoptMode) blockers.push('external_or_shared_session'); + try { + if (canonicalWorkspacePath(session.workingDir!) !== request.target.canonicalWorkingDir) blockers.push('workspace_alias_changed'); + } catch { blockers.push('workspace_alias_unverifiable'); } + return blockers; + } + + private peersClosed(journal: RecycleJournal, sessions: Session[]): boolean { + return journal.peers.filter(peer => recycleKey(peer) !== recycleKey(journal.target)).every(peer => { + const other = readRecycleJournal(recycleJournalPath(this.deps.dataDir(), journal.operationId, peer)); + const copies = sessions.filter(session => session.sessionId === peer.sessionId); + return other?.phase === 'closed' && other.blockers.length === 0 + && JSON.stringify(other.workspace) === JSON.stringify(journal.workspace) + && JSON.stringify(other.peers) === JSON.stringify(journal.peers) + && copies.length === 1 && copies[0].status === 'closed' && !!copies[0].workspaceRetirement + && targetFingerprint(copies[0]) === other.target.fingerprint; + }); + } + + async perform(action: RecycleAction | 'drain', request: RecycleRequest): Promise { + validateRecycleRequest(request); + if (request.target.larkAppId !== this.deps.appId()) throw new Error('wrong_owner_daemon'); + assertRecycleStateOutsideWorkspace(this.deps.dataDir(), request.workspace); + const path = this.path(request); + mkdirSync(dirname(path), { recursive: true, mode: 0o700 }); + return withFileLock(path, async () => { + const previous = readRecycleJournal(path); + if (previous && !sameRecycleRequest(previous, request)) throw new Error('recycle_operation_conflict'); + if (action === 'abort') { + if (!previous) throw new Error('recycle_not_prepared'); + if (!['prepared', 'blocked', 'deferred', 'aborted'].includes(previous.phase) || previous.closeResult) throw new Error('recycle_already_started'); + previous.phase = 'aborted'; previous.blockers = []; + this.save(previous); + return this.result(previous); + } + if (previous?.phase === 'aborted') throw new Error('recycle_aborted'); + if (action !== 'prepare' && !previous) throw new Error('recycle_not_prepared'); + + const guarded = await tryWithBotTurnMutation(this.deps.appId(), 1_000, async () => { + try { + const session = this.deps.getSession(request.target.sessionId); + const ds = this.deps.getRuntime(request.target.sessionId); + if (!session) throw new Error('session_owner_record_missing'); + if (action === 'prepare') { + // Recheck preflight readiness without approving a changed input stamp. + if (previous && !['prepared', 'blocked'].includes(previous.phase)) return this.result(previous); + if (previous && previous.inputStamp !== inputStamp(session, ds)) return this.block(previous, ['new_input_or_worker_generation_after_prepare']); + if (JSON.stringify(captureWorkspace(request.workspace.path)) !== JSON.stringify(request.workspace)) throw new Error('workspace_identity_changed'); + const blockers = this.checkIdentity(request, session, ds); + const busy = recycleBusyReasons(session, ds); + if (ds && this.deps.lifecycleBusy(ds)) busy.push('session_lifecycle_in_flight'); + if (!this.isInitiator(request)) blockers.push(...busy); + if (session.status !== 'active') blockers.push('session_already_closed_before_prepare'); + const journal: RecycleJournal = { + ...request, phase: 'prepared', inputStamp: previous?.inputStamp ?? inputStamp(session, ds), + before: previous?.before ?? (this.deps.capture ?? captureRecycleResources)(session, ds), + blockers, preparedAt: previous?.preparedAt ?? new Date(this.now()).toISOString(), updatedAt: '', + }; + this.save(journal); + return this.result(journal); + } + + const journal = previous!; + assertWorkspaceRemoved(journal.workspace); + const changed = this.checkIdentity(journal, session, ds); + if (changed.length) return this.block(journal, changed); + if (session.status === 'closed') return this.verify(journal, session, ds); + if (journal.phase === 'closed' || journal.phase === 'closed_with_residual') return this.block(journal, ['session_reactivated_after_close']); + if (journal.inputStamp !== inputStamp(session, ds)) return this.block(journal, ['new_input_or_worker_generation_after_prepare']); + if (this.isInitiator(journal)) { + // The caller may have exited minutes ago. Re-discover coverage before + // cashing in its deferred handoff; never hide a newly-created peer. + const allSessions = this.deps.allSessions(); + if (!this.peersClosed(journal, allSessions)) return this.block(journal, ['initiator_waiting_for_other_targets']); + const discovery = discoverWorkspaceSessions(allSessions, journal.workspace.canonicalPath); + const planned = new Set(journal.peers.map(recycleKey)); + if (discovery.errors.length || discovery.targets.some(t => !planned.has(recycleKey(t)))) return this.block(journal, ['workspace_coverage_changed']); + } + if (action === 'defer') { + if (!this.isInitiator(journal)) throw new Error('only_initiator_can_be_deferred'); + journal.deferredUntil ??= this.now() + 15 * 60_000; + journal.phase = 'deferred'; journal.blockers = []; + this.save(journal); + // Always return before closing the caller, even if its screen is + // momentarily idle while the hook subprocess is still receiving ACK. + this.scheduleRecovery(); + return this.result(journal); + } + const busy = recycleBusyReasons(session, ds); + if (ds && this.deps.lifecycleBusy(ds)) busy.push('session_lifecycle_in_flight'); + if (busy.length) { + if (action === 'drain' && this.isInitiator(journal)) { + journal.deferredUntil ??= this.now() + 15 * 60_000; + if (this.now() >= journal.deferredUntil) return this.block(journal, ['initiator_idle_deadline_exceeded', ...busy]); + journal.phase = 'deferred'; journal.blockers = busy; + this.save(journal); // ACK means durable handoff, never completed close. + this.scheduleRecovery(); + return this.result(journal); + } + return this.block(journal, busy); + } + // Capture late descendants before the standard close can detach them. + const latest = (this.deps.capture ?? captureRecycleResources)(session, ds); + const seen = new Set(journal.before.processes.map(p => `${p.pid}:${p.identity}`)); + journal.before.processes.push(...latest.processes.filter(p => !seen.has(`${p.pid}:${p.identity}`))); + journal.phase = 'closing'; journal.blockers = []; + this.save(journal); // A crash after here recovers by readback, not a blind kill. + try { journal.closeResult = await this.deps.close(session.sessionId, { workspaceRetirement: this.retirement(journal) }); } + catch (error) { + const fresh = this.deps.getSession(session.sessionId); + if (fresh?.status === 'closed') { + journal.blockers = [`close_exception:${String(error)}`]; + return this.verify(journal, fresh, this.deps.getRuntime(session.sessionId)); + } + return this.block(journal, [`close_exception:${String(error)}`]); + } + const fresh = this.deps.getSession(session.sessionId); + if (!fresh) return this.block(journal, ['session_record_missing_after_close']); + if (!journal.closeResult.ok) return this.block(journal, [`close_refused:${journal.closeResult.error}`]); + return this.verify(journal, fresh, this.deps.getRuntime(session.sessionId)); + } catch (error) { + if (previous) return this.block(previous, [`recycle_revalidation_error:${String(error)}`]); + throw error; + } + }); + if (guarded.acquired) return guarded.value; + if (previous) return this.block(previous, ['inbound_admission_drain_timeout']); + throw new Error('inbound_admission_drain_timeout'); + }, { maxWaitMs: 1_500 }); + } + + private async verify(journal: RecycleJournal, session: Session, ds?: DaemonSession): Promise { + if (session.status === 'closed' && !session.workspaceRetirement) { + this.deps.retireClosed(session.sessionId, this.retirement(journal)); + session = this.deps.getSession(session.sessionId) ?? session; + } + journal.after = (this.deps.reread ?? rereadRecycleResources)(journal.before, session, ds); + // A standard close fence can be acknowledged just BEFORE process exit. + // Give already-closing owned processes a bounded read-only release window. + const deadline = Date.now() + 2_000; + while (journal.after.processes.some(p => p.state === 'alive') && Date.now() < deadline) { + await new Promise(resolve => setTimeout(resolve, 50)); + journal.after = (this.deps.reread ?? rereadRecycleResources)(journal.before, session, this.deps.getRuntime(session.sessionId)); + } + const residual = this.deps.closeResidual(session); + const blockers = resourceResiduals(journal.after); + if (session.status !== 'closed') blockers.push('durable_status_not_closed'); + if (!session.workspaceRetirement) blockers.push('durable_retirement_missing'); + if (residual) blockers.push(residual); + if (journal.closeResult?.ok && journal.closeResult.outcome === 'closed_with_residual') blockers.push(journal.closeResult.residual.reason); + // Remote cancellation is proved by the standard close's durable result. + // A crash that lost that receipt must not guess cancellation from a null id. + if (journal.after.backing.state === 'remote' && !journal.closeResult?.ok) blockers.push('remote_close_receipt_missing'); + journal.blockers = [...new Set([...journal.blockers.filter(s => s.startsWith('close_exception:')), ...blockers])]; + journal.phase = session.status !== 'closed' ? 'blocked' : journal.blockers.length ? 'closed_with_residual' : 'closed'; + this.save(journal); + return this.result(journal); + } + + /** Only explicitly handed-off initiators resume automatically. Ordinary + * partial failures remain inspectable until an operator replays commit. */ + async recoverDeferred(): Promise { + if (this.stopped) return; + const root = join(this.deps.dataDir(), 'workspace-recycle'); + let operations: string[]; + try { operations = readdirSync(root); } + catch (error) { if ((error as NodeJS.ErrnoException).code === 'ENOENT') return; throw error; } + let pending = false; + for (const operation of operations) { + let files: string[]; + try { files = readdirSync(join(root, operation)); } catch { continue; } + for (const file of files.filter(file => /^[a-f0-9]{64}\.json$/.test(file))) { + try { + const journal = readRecycleJournal(join(root, operation, file)); + if (journal?.phase !== 'deferred' || journal.target.larkAppId !== this.deps.appId()) continue; + const result = await this.perform('drain', journal); + pending ||= result.status === 'deferred'; + } catch (error) { + // An unreadable journal must not hide other independent handoffs. + this.deps.onError?.(error); + } + } + } + if (pending) this.scheduleRecovery(); + } + + scheduleRecovery(): void { + if (this.stopped || this.timer) return; + this.timer = setTimeout(() => { + this.timer = undefined; + void this.recoverDeferred().catch(error => this.deps.onError?.(error)); + }, 2_000); + this.timer.unref(); + } + stop(): void { + this.stopped = true; + if (this.timer) clearTimeout(this.timer); + this.timer = undefined; + } + + start(): void { this.stopped = false; } +} diff --git a/src/daemon.ts b/src/daemon.ts index 7e88ac68f..660560145 100644 --- a/src/daemon.ts +++ b/src/daemon.ts @@ -18131,7 +18131,7 @@ async function handleNewTopicAdmitted(data: any, ctx: RoutingContext): Promise = { 'card.action.resume_adopt_unsupported': '⚠️ Adopted sessions cannot be resumed.', 'card.action.resume_deferred_unmaterialized': '⚠️ This silent scheduled run never created a topic. Its hidden session is audit-only and cannot be resumed.', 'card.action.resume_cancelled': '⚠️ The session was closed while resume was committing; resume was cancelled.', + 'card.action.resume_workspace_retired': 'This session’s workspace has been reclaimed and the session cannot be resumed. Create a new session in a valid workspace.', 'card.action.disconnected': '⏏ Disconnected. The original CLI is untouched.', 'card.voice.toast_wait': '🔊 Generating a voice summary, hang tight…', 'card.voice.toast_already': '🔊 A voice summary for this reply is already on the way.', diff --git a/src/i18n/zh.ts b/src/i18n/zh.ts index 68aad120f..149094556 100644 --- a/src/i18n/zh.ts +++ b/src/i18n/zh.ts @@ -946,6 +946,7 @@ export const messages: Record = { 'card.action.resume_adopt_unsupported': '⚠️ adopt 接管会话不支持 resume。', 'card.action.resume_deferred_unmaterialized': '⚠️ 该静默定时轮次未创建话题，隐藏会话只保留审计记录，无法恢复。', 'card.action.resume_cancelled': '⚠️ 恢复过程中会话被关闭，本次恢复已取消。', + 'card.action.resume_workspace_retired': '该会话的工作区已回收，无法恢复。请在有效工作区创建新会话。', 'card.action.disconnected': '⏏ 已断开，原 CLI 会话不受影响', 'card.voice.toast_wait': '🔊 正在生成语音总结，请耐心等待…', 'card.voice.toast_already': '🔊 这条已经在生成语音啦，请稍候', diff --git a/src/im/lark/card-handler.ts b/src/im/lark/card-handler.ts index 363bbb1ed..c30e4b9e0 100644 --- a/src/im/lark/card-handler.ts +++ b/src/im/lark/card-handler.ts @@ -3029,6 +3029,8 @@ export async function handleCardAction(data: CardActionData, deps: CardHandlerDe await sessionReply(rootId, t('card.action.resume_deferred_unmaterialized', undefined, locDsResume)); } else if (result.error === 'resume_cancelled') { await sessionReply(rootId, t('card.action.resume_cancelled', undefined, locDsResume)); + } else if (result.error === 'workspace_retired') { + await sessionReply(rootId, t('card.action.resume_workspace_retired', undefined, locDsResume)); } } } diff --git a/src/services/session-commands.ts b/src/services/session-commands.ts index d98330439..2b6117b55 100644 --- a/src/services/session-commands.ts +++ b/src/services/session-commands.ts @@ -26,6 +26,7 @@ import type { LarkAttachment, Session } from '../types.js'; /** Close inputs only the owning daemon can produce. */ type DaemonCloseOptions = { + workspaceRetirement?: Session['workspaceRetirement']; /** * Close-time token snapshot, sampled by the caller BEFORE the write (the * daemon's transcript scan, never run under the store lock). Omit to leave @@ -156,6 +157,10 @@ function applyClose(row: Session, command: SessionCloseCommand, now: Date): Sess // The materialised images are cleaned up AFTER the row commits, so the list // is handed back before it is dropped from the row. const released: SessionRowReleased = {}; + if (command.workspaceRetirement && !row.workspaceRetirement) { + row.workspaceRetirement = { ...command.workspaceRetirement }; + changed = true; + } if (!alreadyClosed) { if (row.dashboardAttachments?.length) { diff --git a/src/services/session-store.ts b/src/services/session-store.ts index 606d25471..0ac545594 100644 --- a/src/services/session-store.ts +++ b/src/services/session-store.ts @@ -501,10 +501,13 @@ function listStoreRefs(dataDir: string, opts: { strict?: boolean } = {}): StoreF /** All [key, value] entries of one store. Throws on an unreadable store; * callers decide skip-vs-propagate (capability errors always propagate). */ -function readStoreEntries(ref: StoreFileRef): [string, Session][] { +function readStoreEntries(ref: StoreFileRef, strict = false): [string, Session][] { if (ref.kind === 'json') { const parsed = JSON.parse(readFileSync(ref.path, 'utf-8')) as unknown; - if (!parsed || typeof parsed !== 'object') return []; + if (!parsed || typeof parsed !== 'object' || (strict && Array.isArray(parsed))) { + if (strict) throw new Error(`Invalid session store: ${ref.path}`); + return []; + } return Object.entries(parsed as Record); } const db = openDbForRead(ref.path); @@ -512,7 +515,8 @@ function readStoreEntries(ref: StoreFileRef): [string, Session][] { const rows = db.prepare('SELECT session_id, row FROM sessions').all() as { session_id: string; row: string }[]; const entries: [string, Session][] = []; for (const r of rows) { - try { entries.push([r.session_id, JSON.parse(r.row) as Session]); } catch { /* skip unparseable row */ } + try { entries.push([r.session_id, JSON.parse(r.row) as Session]); } + catch (error) { if (strict) throw error; /* display readers skip unparseable rows */ } } return entries; } finally { @@ -1662,6 +1666,12 @@ function persistRow(session: Session): void { const existing = ownStore.selectRow.get(session.sessionId) as { row: string } | undefined; if (existing) { const durable = JSON.parse(existing.row) as Session; + if (durable.workspaceRetirement) { + if (session.status !== 'closed') throw new Error('workspace_retired'); + // Late whole-row writers may hold a pre-close copy. Retirement cannot + // be erased by a stale metadata write or by changing workingDir. + session = { ...session, workspaceRetirement: durable.workspaceRetirement }; + } if (durable.cliInstanceBinding) { if (session.cliInstanceBinding && JSON.stringify(session.cliInstanceBinding) !== JSON.stringify(durable.cliInstanceBinding)) { throw new Error('Codex instance binding is immutable'); @@ -2106,6 +2116,7 @@ export function closeSession( sessionId: string, opts: { cleanupBridgeMarkers?: boolean; + workspaceRetirement?: Session['workspaceRetirement']; clearRiffParentTaskId?: boolean; /** * Park an uncancellable mojo lineage as PART of this transaction. @@ -2164,6 +2175,7 @@ export function closeSession( type: 'close', ...(tokenUsage !== undefined ? { tokenUsage } : {}), clearMojoCloseJournal: true, + ...(opts.workspaceRetirement ? { workspaceRetirement: opts.workspaceRetirement } : {}), ...(opts.parkMojoLineage ? { parkMojoLineage: opts.parkMojoLineage } : {}), ...(opts.parkLocalResidual ? { parkLocalResidual: opts.parkLocalResidual } : {}), ...(opts.clearRiffParentTaskId ? { clearRiffParentTaskId: true } : {}), @@ -2207,10 +2219,11 @@ export function closeSession( export function reactivateClosedSession( sessionId: string, ): { ok: true; session: Session } -| { ok: false; error: 'not_found' | 'not_closed' } { +| { ok: false; error: 'not_found' | 'not_closed' | 'workspace_retired' } { loadForWrite(); const session = sessions.get(sessionId); if (!session) return { ok: false, error: 'not_found' }; + if (session.workspaceRetirement) return { ok: false, error: 'workspace_retired' }; if (session.status !== 'closed') return { ok: false, error: 'not_closed' }; // Durable first (see closeSession): the reactivated row is committed before @@ -2580,6 +2593,27 @@ export function loadAllSessionsSnapshot(options: { return out; } +/** Destructive lifecycle discovery must account for every store, including + * corrupt stores and duplicate owners. Unlike the display snapshot this does + * not skip errors or collapse two copies of the same session id. Pure reader. */ +export function loadAllSessionsStrict(dataDir = config.session.dataDir): Session[] { + const result: Session[] = []; + for (const ref of listStoreRefs(dataDir, { strict: true })) { + for (const [key, raw] of readStoreEntries(ref, true)) { + const session = raw as Session; + if (!session || typeof session !== 'object' || Array.isArray(session) + || session.sessionId !== key || !['active', 'closed'].includes(session.status)) { + throw new Error(`Invalid session identity in ${ref.path}`); + } + if (ref.appId && session.larkAppId && session.larkAppId !== ref.appId) { + throw new Error(`Session owner mismatch in ${ref.path}`); + } + result.push({ ...session, larkAppId: ref.appId ?? session.larkAppId }); + } + } + return result; +} + /** * Unlocked point-read of one row straight from disk, bypassing this process's * in-memory cache: the owning per-bot store first, then the legacy store. diff --git a/src/services/workspace-recycle.ts b/src/services/workspace-recycle.ts new file mode 100644 index 000000000..19aa78dc5 --- /dev/null +++ b/src/services/workspace-recycle.ts @@ -0,0 +1,230 @@ +import { mkdirSync, readFileSync } from 'node:fs'; +import { dirname, join } from 'node:path'; +import { withFileLock } from '../utils/file-lock.js'; +import { loadAllSessionsStrict } from './session-store.js'; +import { listOnlineDaemons, type OnlineDaemonInfo } from '../utils/daemon-discovery.js'; +import { fetchDaemonIpc } from '../core/daemon-ipc-auth.js'; +import { captureWorkspace, assertWorkspaceRemoved, type WorkspaceIdentity } from '../core/workspace-recycle-path.js'; +import { discoverWorkspaceSessions, targetFingerprint, WORKSPACE_RECYCLE_PROTOCOL, type WorkspaceDiscovery, type WorkspaceSessionTarget } from '../core/workspace-recycle-model.js'; +import { + assertRecycleStateOutsideWorkspace, persistRecycleJson, readRecycleJournal, recycleJournalPath, + recycleKey, validateRecycleOperationId, type RecyclePeer, type RecycleRequest, type RecycleJournal, +} from '../core/workspace-recycle-journal.js'; +import type { RecycleAction, RecycleTargetResult } from '../core/workspace-recycle-runtime.js'; +import type { Session } from '../types.js'; + +export interface WorkspaceRecycleOperation { + protocol: typeof WORKSPACE_RECYCLE_PROTOCOL; + operationId: string; + workspace: WorkspaceIdentity; + initiator?: RecyclePeer; + targets: WorkspaceSessionTarget[]; + excluded: WorkspaceDiscovery['excluded']; + phase: 'preparing' | 'prepared' | 'committing' | 'pending' | 'partial' | 'closed' | 'aborted'; + prepared: boolean; + errors: Array<{ sessionId?: string; error: string }>; + lifecycle?: { eventId: string; outcome: 'succeeded' | 'failed' }; + createdAt: string; + updatedAt: string; +} + +export interface WorkspaceRecycleStatus { + ok: boolean; + status: WorkspaceRecycleOperation['phase']; + operation: WorkspaceRecycleOperation; + results: Array; + errors: WorkspaceRecycleOperation['errors']; + evidencePath: string; +} + +export interface WorkspaceRecycleDeps { + dataDir: string; + sessions?: () => Session[]; + daemons?: () => OnlineDaemonInfo[]; + call?: (daemon: OnlineDaemonInfo, action: RecycleAction, request: RecycleRequest) => Promise; +} + +/** This coordinator does not own a session store and never closes a row on + * disk as a fallback. Only each target's authenticated live owner may close. */ +export class WorkspaceRecycler { + constructor(private readonly deps: WorkspaceRecycleDeps) {} + + discover(path: string): WorkspaceDiscovery { + return discoverWorkspaceSessions(this.sessions(), path); + } + private sessions(): Session[] { return (this.deps.sessions ?? (() => loadAllSessionsStrict(this.deps.dataDir)))(); } + private path(operationId: string): string { + validateRecycleOperationId(operationId); + return join(this.deps.dataDir, 'workspace-recycle', operationId, 'operation.json'); + } + private save(operation: WorkspaceRecycleOperation): void { + operation.updatedAt = new Date().toISOString(); + persistRecycleJson(this.path(operation.operationId), operation); + } + private load(operationId: string): WorkspaceRecycleOperation { + const operation = JSON.parse(readFileSync(this.path(operationId), 'utf8')) as WorkspaceRecycleOperation; + if (operation.protocol !== WORKSPACE_RECYCLE_PROTOCOL || operation.operationId !== operationId + || !Array.isArray(operation.targets) || !Array.isArray(operation.errors)) throw new Error('invalid_recycle_operation'); + assertRecycleStateOutsideWorkspace(this.deps.dataDir, operation.workspace); + return operation; + } + private request(operation: WorkspaceRecycleOperation, target: WorkspaceSessionTarget): RecycleRequest { + return { + protocol: WORKSPACE_RECYCLE_PROTOCOL, operationId: operation.operationId, + workspace: operation.workspace, target, + peers: operation.targets.map(({ larkAppId, sessionId }) => ({ larkAppId, sessionId })), + initiator: operation.initiator, + }; + } + + private coverageErrors(operation: WorkspaceRecycleOperation): WorkspaceRecycleOperation['errors'] { + try { + const after = this.discover(operation.workspace.canonicalPath); + const planned = new Set(operation.targets.map(recycleKey)); + return [ + ...after.errors, + ...after.targets.filter(target => !planned.has(recycleKey(target))) + .map(target => ({ sessionId: target.sessionId, error: 'unplanned_active_session' })), + ]; + } catch (error) { return [{ error: `workspace_coverage_unavailable:${String(error)}` }]; } + } + + private async call(action: RecycleAction, request: RecycleRequest): Promise { + const daemons = (this.deps.daemons ?? (() => listOnlineDaemons(this.deps.dataDir)))(); + const matches = daemons.filter(d => d.larkAppId === request.target.larkAppId); + if (matches.length !== 1 || !Number.isSafeInteger(matches[0].ipcPort) + || matches[0].ipcPort < 1 || matches[0].ipcPort > 65_535) throw new Error('owner_daemon_missing_or_ambiguous'); + if (this.deps.call) return this.deps.call(matches[0], action, request); + const response = await fetchDaemonIpc(matches[0].ipcPort, `/api/workspace-recycle/${action}`, { + method: 'POST', headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify(request), signal: AbortSignal.timeout(60_000), + }); + const result = await response.json() as RecycleTargetResult & { error?: string }; + if (!result.journal || result.journal.target.sessionId !== request.target.sessionId + || result.journal.target.larkAppId !== request.target.larkAppId || result.journal.operationId !== request.operationId) { + throw new Error(`recycle_owner_response_unavailable:${response.status}:${result.error ?? 'invalid_result'}`); + } + return result; + } + + async prepare(operationId: string, path: string, initiatorSessionId?: string): Promise { + const workspace = captureWorkspace(path); + assertRecycleStateOutsideWorkspace(this.deps.dataDir, workspace); + const file = this.path(operationId); + mkdirSync(dirname(file), { recursive: true, mode: 0o700 }); + return withFileLock(file, async () => { + let prior: WorkspaceRecycleOperation | undefined; + try { prior = this.load(operationId); } + catch (error) { if ((error as NodeJS.ErrnoException).code !== 'ENOENT') throw error; } + if (prior) { + if (JSON.stringify(prior.workspace) !== JSON.stringify(workspace) + || (initiatorSessionId && prior.initiator?.sessionId !== initiatorSessionId)) throw new Error('recycle_operation_conflict'); + if (!['preparing', 'prepared'].includes(prior.phase)) return this.status(operationId); + } + const discovered = this.discover(path); + if (discovered.errors.length) throw new Error(`workspace_discovery_incomplete:${JSON.stringify(discovered.errors)}`); + const initiator = discovered.targets.find(t => t.sessionId === initiatorSessionId); + if (initiatorSessionId && !initiator) throw new Error('initiator_not_in_plan'); + const operation: WorkspaceRecycleOperation = prior ?? { + protocol: WORKSPACE_RECYCLE_PROTOCOL, operationId, workspace, + targets: discovered.targets, excluded: discovered.excluded, + initiator: initiator ? { larkAppId: initiator.larkAppId, sessionId: initiator.sessionId } : undefined, + phase: 'preparing', prepared: false, errors: [], createdAt: new Date().toISOString(), updatedAt: '', + }; + operation.errors = []; + this.save(operation); + for (const target of operation.targets) { + try { + const result = await this.call('prepare', this.request(operation, target)); + if (!result.ok) operation.errors.push({ sessionId: target.sessionId, error: result.journal.blockers.join(',') }); + } catch (error) { + operation.errors.push({ sessionId: target.sessionId, error: String(error) }); + } + this.save(operation); + } + // Discovery may have raced a newly-created row while other bots prepared. + const after = this.discover(path); + const planned = new Set(operation.targets.map(recycleKey)); + for (const target of after.targets) if (!planned.has(recycleKey(target))) operation.errors.push({ sessionId: target.sessionId, error: 'new_session_during_prepare' }); + operation.errors.push(...after.errors); + operation.prepared = operation.errors.length === 0; + operation.phase = operation.prepared ? 'prepared' : 'preparing'; + this.save(operation); + return this.status(operationId); + }, { maxWaitMs: 1_500 }); + } + + /** Equivalent generic lifecycle event. A failed end NEVER closes anything. + * The caller supplies the same operationId stamped by its pre-end hook. */ + async finish(operationId: string, event: { eventId: string; outcome: 'succeeded' | 'failed' }): Promise { + if (!event.eventId || !['succeeded', 'failed'].includes(event.outcome)) throw new Error('invalid_recycle_completion_event'); + return withFileLock(this.path(operationId), async () => { + const operation = this.load(operationId); + if (operation.lifecycle && JSON.stringify(operation.lifecycle) !== JSON.stringify(event)) throw new Error('recycle_event_conflict'); + if (operation.phase === 'aborted') return this.status(operationId); + if (event.outcome === 'succeeded' && !operation.prepared) throw new Error('recycle_preflight_not_ready'); + if (event.outcome === 'succeeded') assertWorkspaceRemoved(operation.workspace); + operation.lifecycle = event; + operation.errors = []; + operation.phase = event.outcome === 'failed' ? 'aborted' : 'committing'; + if (event.outcome === 'succeeded') operation.errors.push(...this.coverageErrors(operation)); + this.save(operation); // durable before the first owner receives a close + const targets = [...operation.targets].sort((a, b) => Number(a.sessionId === operation.initiator?.sessionId) - Number(b.sessionId === operation.initiator?.sessionId)); + for (const target of targets) { + const isInitiator = target.sessionId === operation.initiator?.sessionId; + // Check again BEFORE handing off the last session: new rows can arrive + // while its peers close. An incomplete scan must keep the closer alive. + if (event.outcome === 'succeeded' && isInitiator) operation.errors.push(...this.coverageErrors(operation)); + if (event.outcome === 'succeeded' && isInitiator && operation.errors.length) { + operation.errors.push({ sessionId: target.sessionId, error: 'initiator_waiting_for_other_targets' }); + continue; + } + try { + const result = await this.call(event.outcome === 'failed' ? 'abort' : isInitiator ? 'defer' : 'close', this.request(operation, target)); + if (!result.ok && !['aborted', 'deferred'].includes(result.status)) operation.errors.push({ sessionId: target.sessionId, error: result.journal.blockers.join(',') }); + } catch (error) { + operation.errors.push({ sessionId: target.sessionId, error: String(error) }); + } + this.save(operation); + } + if (event.outcome === 'succeeded') { + operation.errors.push(...this.coverageErrors(operation)); + this.save(operation); + const status = this.status(operationId); + operation.phase = status.status; + } + this.save(operation); + return this.status(operationId); + }, { maxWaitMs: 1_500 }); + } + + /** Pure readback; deferred initiator completion can happen after CLI exit. */ + status(operationId: string): WorkspaceRecycleStatus { + const operation = this.load(operationId); + const results = operation.targets.map(target => readRecycleJournal(recycleJournalPath(this.deps.dataDir, operationId, target)) ?? { target, phase: 'missing' as const }); + let status = operation.phase; + const errors = [...operation.errors]; + if (operation.lifecycle?.outcome === 'succeeded') { + errors.push(...this.coverageErrors(operation)); + try { + assertWorkspaceRemoved(operation.workspace); + const sessions = this.sessions(); + for (const result of results.filter(result => result.phase === 'closed')) { + const matches = sessions.filter(s => s.sessionId === result.target.sessionId); + if (matches.length !== 1 || matches[0].status !== 'closed' || !matches[0].workspaceRetirement || targetFingerprint(matches[0]) !== result.target.fingerprint) { + errors.push({ sessionId: result.target.sessionId, error: 'durable_close_readback_changed_or_missing' }); + } + } + } catch (error) { errors.push({ error: `final_readback_unavailable:${String(error)}` }); } + for (const result of results) { + if (!['closed', 'deferred'].includes(result.phase)) errors.push({ sessionId: result.target.sessionId, error: result.phase === 'missing' ? 'target_journal_missing' : result.blockers.join(',') || result.phase }); + } + status = errors.length ? 'partial' : results.some(r => r.phase === 'deferred') ? 'pending' : 'closed'; + } + return { + ok: errors.length === 0 && (status === 'closed' || status === 'prepared' || status === 'aborted'), + status, operation: { ...operation, phase: status }, results, errors, + evidencePath: this.path(operationId), + }; + } +} diff --git a/src/types.ts b/src/types.ts index 9779f0d4e..ee651eb37 100644 --- a/src/types.ts +++ b/src/types.ts @@ -413,6 +413,13 @@ export interface Session { * a bot's own output. */ lastHumanMessageAt?: string; closedAt?: string; + /** Permanent retirement after successful workspace reclamation. Written with + * the closed row; generic resume must never reuse this conversation's cwd. */ + workspaceRetirement?: { + operationId: string; + workspacePath: string; + retiredAt: string; + }; /** Last cumulative token usage persisted at close time. Dashboard list * reads this durable snapshot without rescanning historical transcripts. */ tokenUsage?: SessionTokenUsageSnapshot | null; diff --git a/test/fixtures/workspace-recycle-worker.ts b/test/fixtures/workspace-recycle-worker.ts new file mode 100644 index 000000000..ccb340dbc --- /dev/null +++ b/test/fixtures/workspace-recycle-worker.ts @@ -0,0 +1,18 @@ +import { watch } from 'node:fs'; +import { createServer } from 'node:http'; + +// Only this test-owned worker, socket and watcher are destroyed. Readiness is +// published after EVERY resource and the close handler has been installed. +const watcher = watch(process.argv[2], () => {}); +const timer = setInterval(() => {}, 60_000); +const server = createServer((_req, res) => res.end('isolated-workspace-worker')); +process.on('message', message => { + if ((message as { type?: string })?.type !== 'close') return; + watcher.close(); clearInterval(timer); + server.close(() => process.exit(0)); +}); +server.listen(0, '127.0.0.1', () => { + const address = server.address(); + if (!address || typeof address === 'string') throw new Error('worker_port_missing'); + process.send?.({ type: 'fixture_ready', pid: process.pid, port: address.port }); +}); diff --git a/test/session-resume.test.ts b/test/session-resume.test.ts index 62ae2c6e4..f0ec9d6c0 100644 --- a/test/session-resume.test.ts +++ b/test/session-resume.test.ts @@ -183,6 +183,7 @@ import { restoreUsageLimitRuntimeState, setActiveSessionSafe, setActiveSessionIfActive, + withActiveSessionKeyLock, } from '../src/core/worker-pool.js'; import { TmuxBackend } from '../src/adapters/backend/tmux-backend.js'; import * as sessionStore from '../src/services/session-store.js'; @@ -224,6 +225,19 @@ function makeClosedSession(overrides: Partial { + it('rechecks retirement after waiting for the active-session lock', async () => { + const closed = makeClosedSession(); + const map = new Map(); + vi.mocked(withActiveSessionKeyLock).mockImplementationOnce(async (_map, _key, action) => { + sessionStore.closeSession(closed.sessionId, { + workspaceRetirement: { operationId: 'recycle-lock', workspacePath: '/removed', retiredAt: new Date().toISOString() }, + }); + return action(); + }); + expect(await resumeSession(closed.sessionId, map)).toEqual({ ok: false, error: 'workspace_retired' }); + expect(sessionStore.getSession(closed.sessionId)?.status).toBe('closed'); + expect(map.size).toBe(0); + }); describe('error branches', () => { it('returns not_found for an unknown session id', async () => { const r = await resumeSession('no-such-id', new Map()); diff --git a/test/session-store.test.ts b/test/session-store.test.ts index 3704f78fb..0a64d3a2c 100644 --- a/test/session-store.test.ts +++ b/test/session-store.test.ts @@ -673,6 +673,34 @@ describe('closeSession()', () => { expect(getSession(session.sessionId)?.riffParentTaskId).toBeUndefined(); }); + it('commits workspace retirement with closed state and refuses stale writers that resurrect or erase it', () => { + const session = createSession('chat1', 'root1', 'Retired workspace'); + const stale = { ...session }; + const retirement = { operationId: 'recycle-1', workspacePath: '/removed', retiredAt: new Date().toISOString() }; + closeSession(session.sessionId, { workspaceRetirement: retirement }); + init(); + expect(getSession(session.sessionId)).toMatchObject({ status: 'closed', workspaceRetirement: retirement }); + expect(reactivateClosedSession(session.sessionId)).toEqual({ ok: false, error: 'workspace_retired' }); + expect(() => updateSession({ ...stale, workingDir: tempDir })).toThrow('workspace_retired'); + updateSession({ ...stale, status: 'closed', title: 'Late metadata' }); + init(); + expect(getSession(session.sessionId)).toMatchObject({ status: 'closed', title: 'Late metadata', workspaceRetirement: retirement }); + }); + + it('leaves no retirement in memory or on disk if the atomic close write fails', () => { + const session = createSession('chat1', 'root1', 'Retirement write failure'); + __testOnly_setBeforeRowPersist(() => { throw new Error('retirement write failure'); }); + expect(() => closeSession(session.sessionId, { + workspaceRetirement: { operationId: 'recycle-1', workspacePath: '/removed', retiredAt: new Date().toISOString() }, + })).toThrow('retirement write failure'); + expect(session.status).toBe('active'); + expect(session.workspaceRetirement).toBeUndefined(); + __testOnly_setBeforeRowPersist(undefined); + init(); + expect(getSession(session.sessionId)?.status).toBe('active'); + expect(getSession(session.sessionId)?.workspaceRetirement).toBeUndefined(); + }); + it('restores Riff close state in memory when the atomic save fails', () => { const session = createSession('chat1', 'root1', 'Close Riff Save Failure'); session.backendType = 'riff'; diff --git a/test/workspace-recycle-ipc.test.ts b/test/workspace-recycle-ipc.test.ts new file mode 100644 index 000000000..ffffcfc6b --- /dev/null +++ b/test/workspace-recycle-ipc.test.ts @@ -0,0 +1,59 @@ +import { mkdtempSync, mkdirSync, rmSync } from 'node:fs'; +import { join } from 'node:path'; +import { tmpdir } from 'node:os'; +import { afterEach, describe, expect, it } from 'vitest'; +import { config } from '../src/config.js'; +import { setIpcAuthSecret, setLarkAppId, startIpcServer, type IpcServerHandle } from '../src/core/dashboard-ipc-server.js'; +import { fetchDaemonIpc } from '../src/core/daemon-ipc-auth.js'; +import { loopbackFetch } from '../src/core/loopback-fetch.js'; +import * as store from '../src/services/session-store.js'; +import * as pool from '../src/core/worker-pool.js'; +import { captureWorkspace } from '../src/core/workspace-recycle-path.js'; +import { workspaceTarget, WORKSPACE_RECYCLE_PROTOCOL } from '../src/core/workspace-recycle-model.js'; +import type { RecycleRequest } from '../src/core/workspace-recycle-journal.js'; + +let server: IpcServerHandle | undefined; +const roots: string[] = []; +const secret = 'isolated-workspace-recycle-ipc-secret'; +const originalDataDir = config.session.dataDir; +afterEach(async () => { + await server?.close(); server = undefined; + store.init(); config.session.dataDir = originalDataDir; + setIpcAuthSecret(null); setLarkAppId(''); pool.setActiveSessionsRegistry(new Map()); + for (const root of roots.splice(0)) rmSync(root, { recursive: true, force: true }); +}); + +describe('workspace recycle host IPC', () => { + it('requires host HMAC, routes exact owner, and completes through the actual endpoint', async () => { + const root = mkdtempSync(join(tmpdir(), 'botmux-recycle-ipc-')); roots.push(root); + const workspacePath = join(root, 'work'); mkdirSync(workspacePath); + const dataDir = join(root, 'data'); mkdirSync(dataDir); + config.session.dataDir = dataDir; store.init('fixture-app'); setLarkAppId('fixture-app'); + setIpcAuthSecret(secret); pool.setActiveSessionsRegistry(new Map()); + server = await startIpcServer({ port: 0, host: '127.0.0.1', authRequired: true }); + const session = store.createSession('fixture-chat', 'fixture-thread', 'fixture', 'group'); + Object.assign(session, { larkAppId: 'fixture-app', workingDir: workspacePath, backendType: 'pty' }); + store.updateSession(session); + const workspace = captureWorkspace(workspacePath); + const target = workspaceTarget(session, workspace)!; + const request: RecycleRequest = { protocol: WORKSPACE_RECYCLE_PROTOCOL, operationId: 'ipc-op', workspace, target, peers: [{ sessionId: session.sessionId, larkAppId: 'fixture-app' }] }; + const unsigned = await loopbackFetch(`http://127.0.0.1:${server.port}/api/workspace-recycle/prepare`, { + method: 'POST', body: JSON.stringify({ ...request, originCapability: 'not-a-host-capability' }), headers: { 'Content-Type': 'application/json' }, + }); + expect(unsigned.status).toBe(401); + const post = (action: string, body: unknown) => fetchDaemonIpc(server!.port, `/api/workspace-recycle/${action}`, { + method: 'POST', body: JSON.stringify(body), headers: { 'Content-Type': 'application/json' }, signal: AbortSignal.timeout(10_000), + }, secret); + const wrongOwner = await post('prepare', { ...request, target: { ...target, larkAppId: 'other' }, peers: [{ sessionId: session.sessionId, larkAppId: 'other' }] }); + expect(wrongOwner.status).toBe(409); + const prepared = await post('prepare', request); + expect(prepared.status).toBe(200); + expect(await prepared.json()).toMatchObject({ ok: true, status: 'prepared' }); + expect((await post('close', request)).status).toBe(409); // root still exists + rmSync(workspacePath, { recursive: true }); + const result = await post('close', request); + expect(result.status).toBe(200); + expect(await result.json()).toMatchObject({ ok: true, status: 'closed', journal: { after: { registered: false } } }); + expect(store.readSessionRowFromDisk(session.sessionId, 'fixture-app', dataDir)?.status).toBe('closed'); + }); +}); diff --git a/test/workspace-recycle-session-group.test.ts b/test/workspace-recycle-session-group.test.ts new file mode 100644 index 000000000..edcefb7c1 --- /dev/null +++ b/test/workspace-recycle-session-group.test.ts @@ -0,0 +1,77 @@ +import { afterEach, expect, it, vi } from 'vitest'; +import { mkdirSync, rmSync } from 'node:fs'; +import { join } from 'node:path'; + +const mocks = vi.hoisted(() => { + const dataDir = `${process.env.TMPDIR ?? '/tmp'}/botmux-recycle-group-${process.pid}`; + process.env.SESSION_DATA_DIR = dataDir; + process.env.BOTS_CONFIG = `${dataDir}/bots.json`; + delete process.env.BOTMUX_SESSION_ID; + delete process.env.BOTMUX_LARK_APP_ID; + return { dataDir, reply: vi.fn(async () => 'om_reply'), send: vi.fn(async () => 'om_sent'), fork: vi.fn() }; +}); +vi.mock('@larksuiteoapi/node-sdk', () => ({ Client: class {} })); +vi.mock('../src/im/lark/client.js', async () => ({ + ...await vi.importActual('../src/im/lark/client.js'), + replyMessage: mocks.reply, sendMessage: mocks.send, + getChatMode: vi.fn(async () => 'group'), getChatModeStrict: vi.fn(async () => 'group'), + getChatNameAndMode: vi.fn(async () => ({ name: null, mode: 'group' })), +})); +vi.mock('../src/core/worker-pool.js', async () => ({ + ...await vi.importActual('../src/core/worker-pool.js'), forkWorker: mocks.fork, +})); + +import { registerBot } from '../src/bot-registry.js'; +import * as store from '../src/services/session-store.js'; +import * as pool from '../src/core/worker-pool.js'; +import { initSessionGroups, registerSessionGroup, getSessionGroup } from '../src/services/session-groups-store.js'; +import { WorkspaceRecycleRuntime } from '../src/core/workspace-recycle-runtime.js'; +import { WorkspaceRecycler } from '../src/services/workspace-recycle.js'; +import { __testOnly_activeSessions as active, __testOnly_handleNewTopic as handleNewTopic } from '../src/daemon.js'; +import { type DaemonSession, activeSessionKey } from '../src/core/types.js'; + +const APP = 'app-recycle-group'; +const CHAT = 'oc-recycle-group'; +let controller: WorkspaceRecycleRuntime | undefined; +afterEach(() => { + controller?.stop(); active.clear(); pool.setActiveSessionsRegistry(new Map()); + store.init(); initSessionGroups(APP); + rmSync(mocks.dataDir, { recursive: true, force: true }); +}); + +it('finish(succeeded) keeps the real durable row retired through session-group automatic resume, without fresh fallback', async () => { + const workspace = join(mocks.dataDir, 'workspace'); mkdirSync(workspace, { recursive: true }); + const bot = registerBot({ larkAppId: APP, larkAppSecret: 'fixture', cliId: 'claude-code', allowedUsers: ['ou_fixture'] }); + bot.resolvedAllowedUsers = ['ou_fixture']; + store.init(APP); initSessionGroups(APP); pool.setActiveSessionsRegistry(active); + const session = store.createSession(CHAT, 'om_original', 'recycle group fixture', 'group', 'chat'); + Object.assign(session, { larkAppId: APP, workingDir: workspace, backendType: 'pty' }); store.updateSession(session); + const ds = { session, larkAppId: APP, chatId: CHAT, chatType: 'group', scope: 'chat', workingDir: workspace, + worker: null, workerPort: null, workerToken: null, lastMessageAt: 1, lastScreenStatus: 'idle' } as DaemonSession; + active.set(activeSessionKey(ds), ds); + registerSessionGroup(CHAT, { ownerOpenId: 'ou_fixture', lastSessionId: session.sessionId, createdAt: 1, lastActiveAt: 1 }); + controller = new WorkspaceRecycleRuntime({ + appId: () => APP, dataDir: () => mocks.dataDir, getSession: store.getOwnedSession, + getRuntime: pool.findActiveBySessionId, allSessions: () => store.loadAllSessionsStrict(mocks.dataDir), + close: pool.closeSession, retireClosed: (id, workspaceRetirement) => store.closeSession(id, { workspaceRetirement }), + lifecycleBusy: () => false, closeResidual: () => undefined, + }); + const recycler = new WorkspaceRecycler({ dataDir: mocks.dataDir, daemons: () => [{ larkAppId: APP, ipcPort: 12345 }], + call: (_daemon, action, request) => controller!.perform(action, request) }); + expect((await recycler.prepare('retire-group', workspace)).ok).toBe(true); + rmSync(workspace, { recursive: true }); + expect((await recycler.finish('retire-group', { eventId: 'end-group', outcome: 'succeeded' })).status).toBe('closed'); + // Reload both stores to prove that rejection does not depend on runtime memory. + store.init(APP); initSessionGroups(APP); + mocks.reply.mockClear(); mocks.send.mockClear(); + await handleNewTopic({ + sender: { sender_id: { open_id: 'ou_fixture' }, sender_type: 'user' }, + message: { message_id: 'om_continue', chat_id: CHAT, message_type: 'text', content: JSON.stringify({ text: 'continue' }), create_time: String(Date.now()) }, + }, { chatId: CHAT, messageId: 'om_continue', chatType: 'group', scope: 'chat', anchor: CHAT, larkAppId: APP }); + expect(store.loadAllSessionsStrict(mocks.dataDir)).toMatchObject([{ sessionId: session.sessionId, status: 'closed', + workspaceRetirement: { operationId: 'retire-group', workspacePath: workspace } }]); + expect(active.size).toBe(0); + expect(mocks.fork).not.toHaveBeenCalled(); + expect(getSessionGroup(CHAT)?.lastSessionId).toBe(session.sessionId); + expect([...mocks.reply.mock.calls, ...mocks.send.mock.calls].flat().join(' ')).toMatch(/工作区已回收|workspace has been reclaimed/); +}); diff --git a/test/workspace-recycle.integration.test.ts b/test/workspace-recycle.integration.test.ts new file mode 100644 index 000000000..a3df8a339 --- /dev/null +++ b/test/workspace-recycle.integration.test.ts @@ -0,0 +1,114 @@ +import { mkdtempSync, mkdirSync, rmSync, writeFileSync, readFileSync } from 'node:fs'; +import { join } from 'node:path'; +import { tmpdir } from 'node:os'; +import { once } from 'node:events'; +import type { ChildProcess } from 'node:child_process'; +import { afterEach, describe, expect, it } from 'vitest'; +import { spawnTsScript } from './helpers/ts-runner.js'; +import { config } from '../src/config.js'; +import * as store from '../src/services/session-store.js'; +import * as pool from '../src/core/worker-pool.js'; +import { activeSessionKey, type DaemonSession } from '../src/core/types.js'; +import { WorkspaceRecycleRuntime } from '../src/core/workspace-recycle-runtime.js'; +import { WorkspaceRecycler } from '../src/services/workspace-recycle.js'; +import { sampleRecycleProcess } from '../src/core/workspace-recycle-resources.js'; +import { resumeSession } from '../src/core/session-manager.js'; + +const dirs: string[] = []; +const children: ChildProcess[] = []; +const controllers: WorkspaceRecycleRuntime[] = []; +const originalDataDir = config.session.dataDir; +afterEach(async () => { + for (const controller of controllers.splice(0)) controller.stop(); + for (const child of children.splice(0)) { + if (child.exitCode === null && child.signalCode === null) { + const exited = once(child, 'exit'); child.kill('SIGTERM'); await exited; + } + } + pool.setActiveSessionsRegistry(new Map()); + store.init(); config.session.dataDir = originalDataDir; + for (const dir of dirs.splice(0)) rmSync(dir, { recursive: true, force: true }); +}); + +async function startWorker(workspace: string): Promise<{ worker: ChildProcess; pid: number; port: number }> { + const worker = spawnTsScript(join(process.cwd(), 'test/fixtures/workspace-recycle-worker.ts'), [workspace], { + cwd: process.cwd(), stdio: ['ignore', 'pipe', 'pipe', 'ipc'], + }); + children.push(worker); + const ready = await new Promise<{ pid: number; port: number }>((resolve, reject) => { + let stderr = ''; + worker.stderr?.on('data', data => { stderr += data.toString(); }); + const timeout = setTimeout(() => reject(new Error(`fixture_ready_timeout:${stderr}`)), 10_000); + worker.once('exit', code => { clearTimeout(timeout); reject(new Error(`fixture_exited_before_ready:${code}:${stderr}`)); }); + worker.on('message', message => { + const event = message as { type: string; pid: number; port: number }; + if (event.type === 'fixture_ready') { clearTimeout(timeout); resolve(event); } + }); + }); + return { worker, ...ready }; +} + +describe('isolated real process and standard close', () => { + it('releases worker sockets/inotify resources and preserves closed history over three rounds', async () => { + const root = mkdtempSync(join(tmpdir(), 'botmux-recycle-real-')); dirs.push(root); + const dataDir = join(root, 'data'); mkdirSync(dataDir); + config.session.dataDir = dataDir; store.init('app-recycle-fixture'); + const active = new Map(); pool.setActiveSessionsRegistry(active); + const controller = new WorkspaceRecycleRuntime({ + appId: () => 'app-recycle-fixture', dataDir: () => dataDir, + getSession: id => store.getOwnedSession(id), getRuntime: pool.findActiveBySessionId, + allSessions: () => store.loadAllSessionsStrict(dataDir), close: pool.closeSession, + retireClosed: (id, workspaceRetirement) => store.closeSession(id, { workspaceRetirement }), + lifecycleBusy: ds => pool.isSessionLifecycleInFlight(ds) || pool.isSessionTransferring(ds), + closeResidual: session => pool.mojoCloseResidualForRow(session)?.reason, + }); controllers.push(controller); + const recycler = new WorkspaceRecycler({ + dataDir, daemons: () => [{ larkAppId: 'app-recycle-fixture', ipcPort: 12345 }], + call: (_daemon, action, request) => controller.perform(action, request), + }); + const evidence: unknown[] = []; + for (let round = 0; round < 3; round++) { + const workspace = join(root, `workspace-${round}`); mkdirSync(workspace); + const transcript = join(root, `retained-transcript-${round}.jsonl`); writeFileSync(transcript, 'retained\n'); + const { worker, pid, port } = await startWorker(workspace); + const session = store.createSession(`fixture-chat-${round}`, `fixture-thread-${round}`, 'isolated recycle fixture', 'group'); + Object.assign(session, { larkAppId: 'app-recycle-fixture', workingDir: workspace, backendType: 'pty', pid }); + store.updateSession(session); + const ds = { + session, worker, workerReady: true, workerPort: port, workerToken: 'fixture-token', + workerGeneration: 1, larkAppId: 'app-recycle-fixture', chatId: session.chatId, + chatType: 'group', scope: 'thread', workingDir: workspace, spawnedAt: Date.now(), + cliVersion: 'fixture', lastMessageAt: 1, lastScreenStatus: 'idle', hasHistory: true, + initConfig: { backendType: 'pty' }, + } as DaemonSession; + active.set(activeSessionKey(ds), ds); + const before = sampleRecycleProcess(pid, 'worker'); + expect(before.state).toBe('alive'); + if (process.platform === 'linux') expect(before.inotifyInstances).toBeGreaterThan(0); + const operationId = `round-${round}`; + expect((await recycler.prepare(operationId, workspace)).ok).toBe(true); + rmSync(workspace, { recursive: true }); + const result = await recycler.finish(operationId, { eventId: `end-${round}`, outcome: 'succeeded' }); + expect(result.status).toBe('closed'); + expect(store.getOwnedSession(session.sessionId)?.status).toBe('closed'); + expect(active.size).toBe(0); + store.init('app-recycle-fixture'); + expect(await resumeSession(session.sessionId, active)).toEqual({ ok: false, error: 'workspace_retired' }); + expect(store.getOwnedSession(session.sessionId)?.status).toBe('closed'); + expect(active.size).toBe(0); + mkdirSync(workspace); // Even a newly created directory cannot revive the old owner. + expect(await resumeSession(session.sessionId, active)).toEqual({ ok: false, error: 'workspace_retired' }); + rmSync(workspace, { recursive: true }); + expect(sampleRecycleProcess(pid, 'worker', before.identity).state).toBe('gone'); + expect(readFileSync(transcript, 'utf8')).toBe('retained\n'); + expect((await recycler.finish(operationId, { eventId: `end-${round}`, outcome: 'succeeded' })).status).toBe('closed'); + evidence.push({ round, sessionId: session.sessionId, larkAppId: session.larkAppId, chatId: session.chatId, workspace, before, result, + resumeResult: 'workspace_retired', activeAfterResume: active.size, durableRow: store.getOwnedSession(session.sessionId) }); + } + expect(store.listSessionsStrict().filter(s => s.status === 'active')).toHaveLength(0); + expect(store.listSessionsStrict().filter(s => s.status === 'closed')).toHaveLength(3); + // Optional evidence path is supplied only by this task's isolated validation + // run. Normal CI leaves no artifact outside the disposable fixture. + if (process.env.BOTMUX_RECYCLE_TEST_EVIDENCE) writeFileSync(process.env.BOTMUX_RECYCLE_TEST_EVIDENCE, JSON.stringify(evidence, null, 2)); + }, 30_000); +}); diff --git a/test/workspace-recycle.test.ts b/test/workspace-recycle.test.ts new file mode 100644 index 000000000..eeff07395 --- /dev/null +++ b/test/workspace-recycle.test.ts @@ -0,0 +1,378 @@ +import { mkdtempSync, mkdirSync, rmSync, symlinkSync, writeFileSync, readFileSync } from 'node:fs'; +import { join } from 'node:path'; +import { tmpdir } from 'node:os'; +import { DatabaseSync } from 'node:sqlite'; +import { runWorkspaceRecycleCommand } from '../src/cli/workspace-recycle.js'; +import { afterEach, describe, expect, it, vi } from 'vitest'; +import type { Session } from '../src/types.js'; +import type { DaemonSession } from '../src/core/types.js'; +import { WorkspaceRecycler } from '../src/services/workspace-recycle.js'; +import { WorkspaceRecycleRuntime } from '../src/core/workspace-recycle-runtime.js'; +import { canonicalWorkspacePath, captureWorkspace, pathInside } from '../src/core/workspace-recycle-path.js'; +import { discoverWorkspaceSessions, WORKSPACE_RECYCLE_PROTOCOL, workspaceTarget } from '../src/core/workspace-recycle-model.js'; +import { readRecycleJournal, recycleJournalPath, type RecycleRequest } from '../src/core/workspace-recycle-journal.js'; +import { rereadRecycleResources, resourceResiduals, type RecycleResources } from '../src/core/workspace-recycle-resources.js'; +import { withBotTurnAdmission } from '../src/core/bot-turn-mutation-gate.js'; +import { loadAllSessionsStrict } from '../src/services/session-store.js'; + +const roots: string[] = []; +const controllers: WorkspaceRecycleRuntime[] = []; +afterEach(() => { + for (const controller of controllers.splice(0)) controller.stop(); + for (const root of roots.splice(0)) rmSync(root, { recursive: true, force: true }); + vi.restoreAllMocks(); + vi.unstubAllEnvs(); +}); + +function fixture() { + const root = mkdtempSync(join(tmpdir(), 'botmux-recycle-')); + roots.push(root); + const workspace = join(root, 'work'); + const dataDir = join(root, 'data'); + mkdirSync(workspace); mkdirSync(dataDir); + const sessions = new Map(); + const live = new Map(); + const runtime = new Map(); + const closed: string[] = []; + const failures = new Set(); + const residuals = new Set(); + const resources = (session: Session, ds?: DaemonSession): RecycleResources => ({ + registered: !!ds, processes: [], backing: { type: 'pty', state: 'not_applicable' }, errors: [], + }); + function add(id: string, app = 'app-a', cwd = workspace, chat = 'chat-one') { + const session: Session = { + sessionId: id, larkAppId: app, chatId: chat, rootMessageId: `thread-${id}`, + title: id, status: 'active', scope: 'thread', createdAt: '2026-01-01T00:00:00.000Z', + workingDir: cwd, backendType: 'pty', + }; + sessions.set(id, session); + live.set(id, { + session, larkAppId: app, chatId: chat, scope: 'thread', worker: null, + workerPort: null, workerToken: null, workingDir: cwd, lastMessageAt: 1, + lastScreenStatus: 'idle', + } as DaemonSession); + if (!runtime.has(app)) { + const controller = new WorkspaceRecycleRuntime({ + appId: () => app, dataDir: () => dataDir, + getSession: sid => { const s = sessions.get(sid); return s?.larkAppId === app ? s : undefined; }, + getRuntime: sid => live.get(sid), lifecycleBusy: () => false, + allSessions: () => [...sessions.values()], + closeResidual: s => residuals.has(s.sessionId) ? 'mojo_lineage_quarantined' : undefined, + capture: resources, reread: (_before, s, ds) => resources(s, ds), + retireClosed: (sid, retirement) => { sessions.get(sid)!.workspaceRetirement ??= retirement; }, + close: async (sid, opts) => { + if (failures.has(sid)) throw new Error('injected_close_failure'); + closed.push(sid); sessions.get(sid)!.status = 'closed'; live.delete(sid); + sessions.get(sid)!.workspaceRetirement = opts.workspaceRetirement; + return residuals.has(sid) + ? { ok: true, outcome: 'closed_with_residual', known: true, alreadyClosed: false, residual: { reason: 'mojo_lineage_quarantined', taskId: 'remote-fixture' } } + : { ok: true, outcome: 'closed', known: true, alreadyClosed: false }; + }, + }); + runtime.set(app, controller); controllers.push(controller); + } + return session; + } + const recycler = new WorkspaceRecycler({ + dataDir, sessions: () => [...sessions.values()], + daemons: () => [...runtime.keys()].map(larkAppId => ({ larkAppId, ipcPort: 12345 })), + call: async (daemon, action, request) => runtime.get(daemon.larkAppId)!.perform(action, request), + }); + return { root, workspace, dataDir, sessions, live, runtime, closed, failures, residuals, add, recycler }; +} + +const success = { eventId: 'end-result-1', outcome: 'succeeded' as const }; + +describe('workspace ownership discovery', () => { + it('selects all bots/chats and nested paths, rejects prefix neighbours and same-chat unrelated rows', () => { + const f = fixture(); + f.add('root'); f.add('nested', 'app-b', join(f.workspace, 'repo'), 'chat-two'); + f.add('neighbour', 'app-a', `${f.workspace}-other`); + f.add('same-chat', 'app-c', join(f.root, 'unrelated')); + const shared = f.add('shared'); shared.adoptedFrom = { source: 'tmux', tmuxTarget: 'user:1', cwd: f.workspace }; + f.add('old').status = 'closed'; + const discovered = f.recycler.discover(f.workspace); + expect(discovered.targets.map(t => t.sessionId)).toEqual(['root', 'nested']); + expect(discovered.targets[1]).toMatchObject({ larkAppId: 'app-b', chatId: 'chat-two', workingDir: join(f.workspace, 'repo'), evidence: 'session.workingDir' }); + expect(discovered.excluded.map(t => t.reason)).toEqual(['external_or_shared_session', 'already_closed']); + expect(pathInside(f.workspace, `${f.workspace}-other`)).toBe(false); + }); + + it('resolves existing ancestor aliases after removal and detects changed symlink targets', async () => { + const f = fixture(); + const alias = join(f.root, 'alias'); + symlinkSync(f.root, alias, 'dir'); + const raw = join(alias, 'work', 'repo'); + f.add('aliased', 'app-a', raw); + expect(canonicalWorkspacePath(raw)).toBe(join(f.workspace, 'repo')); + expect((await f.recycler.prepare('op-alias', f.workspace)).ok).toBe(true); + rmSync(f.workspace, { recursive: true }); + expect((await f.recycler.finish('op-alias', success)).status).toBe('closed'); + expect(f.sessions.get('aliased')?.status).toBe('closed'); + }); + + it('reports duplicate owners and malformed stores as coverage errors', () => { + const f = fixture(); const a = f.add('duplicate'); + const result = discoverWorkspaceSessions([a, { ...a, larkAppId: 'app-b' }], f.workspace); + expect(result.targets).toEqual([]); + expect(result.errors).toHaveLength(2); + writeFileSync(join(f.dataDir, 'sessions-app-a.json'), 'null'); + expect(() => loadAllSessionsStrict(f.dataDir)).toThrow('Invalid session store'); + writeFileSync(join(f.dataDir, 'sessions-app-a.json'), JSON.stringify({ duplicate: a })); + writeFileSync(join(f.dataDir, 'sessions-app-b.json'), '{broken'); + expect(() => loadAllSessionsStrict(f.dataDir)).toThrow(); + }); + + it('refuses malformed SQLite rows instead of silently dropping coverage', () => { + const f = fixture(); + const dir = join(f.dataDir, 'session-stores', 'app-a'); mkdirSync(dir, { recursive: true }); + const db = new DatabaseSync(join(dir, 'sessions.db')); + db.exec('CREATE TABLE sessions (session_id TEXT PRIMARY KEY, row TEXT NOT NULL)'); + db.prepare('INSERT INTO sessions VALUES (?, ?)').run('broken', '{bad json'); db.close(); + expect(() => loadAllSessionsStrict(f.dataDir)).toThrow(); + }); + + it('allows missing-workspace dry-run but refuses prepare without a live directory', async () => { + const f = fixture(); f.add('legacy'); rmSync(f.workspace, { recursive: true }); + expect(f.recycler.discover(f.workspace).targets).toHaveLength(1); + await expect(f.recycler.prepare('legacy-op', f.workspace)).rejects.toThrow(); + expect(f.closed).toEqual([]); + }); +}); + +describe('recycle lifecycle and recovery', () => { + it('closes exact targets across bots, keeps history, and replays without another close', async () => { + const f = fixture(); f.add('one'); f.add('two', 'app-b'); f.add('unrelated', 'app-b', `${f.workspace}-other`); + expect((await f.recycler.prepare('op', f.workspace)).status).toBe('prepared'); + rmSync(f.workspace, { recursive: true }); + expect((await f.recycler.finish('op', success)).status).toBe('closed'); + expect((await f.recycler.finish('op', success)).status).toBe('closed'); + expect(f.closed).toEqual(['one', 'two']); + expect(f.sessions.get('one')?.title).toBe('one'); + expect(f.sessions.get('unrelated')?.status).toBe('active'); + }); + + it('does not close after a failed recycle or when the root still exists/reappears', async () => { + const f = fixture(); f.add('one'); await f.recycler.prepare('op', f.workspace); + await expect(f.recycler.finish('op', success)).rejects.toThrow('workspace_still_exists'); + const failed = await f.recycler.finish('op', { eventId: 'failed', outcome: 'failed' }); + expect(failed.status).toBe('aborted'); expect(f.closed).toEqual([]); + expect(f.sessions.get('one')?.workspaceRetirement).toBeUndefined(); + await expect(f.recycler.finish('op', success)).rejects.toThrow('recycle_event_conflict'); + }); + + it('rechecks busy preflight on replay while keeping changed input unapproved', async () => { + const f = fixture(); f.add('one'); + f.live.get('one')!.pendingWaitPromises = new Set([Promise.resolve()]); + expect((await f.recycler.prepare('op', f.workspace)).ok).toBe(false); + f.live.get('one')!.pendingWaitPromises.clear(); + expect((await f.recycler.prepare('op', f.workspace)).ok).toBe(true); + f.live.get('one')!.lastMessageAt = 2; + const retry = await f.recycler.prepare('op', f.workspace); + expect(retry.ok).toBe(false); + expect(retry.errors[0].error).toContain('new_input_or_worker_generation'); + expect(f.closed).toEqual([]); + }); + + it('rejects an explicit initiator outside the exact target plan', async () => { + const f = fixture(); f.add('one'); + await expect(f.recycler.prepare('op', f.workspace, 'wrong-id')).rejects.toThrow('initiator_not_in_plan'); + expect(f.closed).toEqual([]); + }); + + it('records partial failure and resumes only the remaining exact session', async () => { + const f = fixture(); f.add('one'); f.add('two', 'app-b'); f.failures.add('two'); + await f.recycler.prepare('op', f.workspace); rmSync(f.workspace, { recursive: true }); + const partial = await f.recycler.finish('op', success); + expect(partial.status).toBe('partial'); expect(f.closed).toEqual(['one']); + f.failures.delete('two'); + const recovered = await f.recycler.finish('op', success); + expect(recovered.status).toBe('closed'); expect(f.closed).toEqual(['one', 'two']); + }); + + it('retains closed_with_residual on retries and never calls it overall success', async () => { + const f = fixture(); f.add('remote'); f.residuals.add('remote'); + await f.recycler.prepare('op', f.workspace); rmSync(f.workspace, { recursive: true }); + const first = await f.recycler.finish('op', success); + expect(first.status).toBe('partial'); expect(first.ok).toBe(false); + expect(first.results[0].phase).toBe('closed_with_residual'); + expect((await f.recycler.finish('op', success)).results[0].phase).toBe('closed_with_residual'); + expect(f.closed).toEqual(['remote']); + }); + + it('protects a busy session before reclamation and refuses to discard new input after preparation', async () => { + const f = fixture(); f.add('busy'); f.live.get('busy')!.pendingPrompt = 'in-flight'; + expect((await f.recycler.prepare('op-busy', f.workspace)).ok).toBe(false); + await expect(f.recycler.finish('op-busy', success)).rejects.toThrow('preflight_not_ready'); + f.live.get('busy')!.pendingPrompt = undefined; + await f.recycler.prepare('op-idle', f.workspace); + f.live.get('busy')!.lastMessageAt++; + rmSync(f.workspace, { recursive: true }); + const changed = await f.recycler.finish('op-idle', success); + expect(changed.status).toBe('partial'); expect(f.closed).toEqual([]); + expect(changed.errors.some(e => e.error.includes('new_input'))).toBe(true); + }); + + it('refuses a transferred or repinned session and does not act on a replacement owner', async () => { + const f = fixture(); const s = f.add('moved'); await f.recycler.prepare('op', f.workspace); + s.workingDir = join(f.root, 'elsewhere'); + rmSync(f.workspace, { recursive: true }); + expect((await f.recycler.finish('op', success)).status).toBe('partial'); + expect(f.closed).toEqual([]); + }); + + it('waits for already-admitted input before taking its closing decision', async () => { + const f = fixture(); f.add('race'); await f.recycler.prepare('op', f.workspace); + rmSync(f.workspace, { recursive: true }); + let release!: () => void; + const held = new Promise(r => { release = r; }); + let admitted!: () => void; + const ready = new Promise(r => { admitted = r; }); + const turn = withBotTurnAdmission('app-a', async () => { admitted(); await held; f.live.get('race')!.pendingPrompt = 'new-message'; }); + await ready; + const closing = f.recycler.finish('op', success); + release(); await turn; + expect((await closing).status).toBe('partial'); expect(f.closed).toEqual([]); + }); + + it('hands off the initiator durably, closes it last when idle, and keeps all peers explicit', async () => { + const f = fixture(); f.add('initiator'); f.add('peer', 'app-b'); + f.live.get('initiator')!.pendingPrompt = 'finishing turn'; + expect((await f.recycler.prepare('op', f.workspace, 'initiator')).ok).toBe(true); + rmSync(f.workspace, { recursive: true }); + const result = await f.recycler.finish('op', success); + expect(result.status).toBe('pending'); expect(result.ok).toBe(false); + expect(f.closed).toEqual(['peer']); + const journal = readRecycleJournal(recycleJournalPath(f.dataDir, 'op', { larkAppId: 'app-a', sessionId: 'initiator' })); + expect(journal?.phase).toBe('deferred'); + await f.runtime.get('app-a')!.recoverDeferred(); + expect(f.closed).toEqual(['peer']); + f.live.get('initiator')!.pendingPrompt = undefined; + await f.runtime.get('app-a')!.recoverDeferred(); + expect(f.closed).toEqual(['peer', 'initiator']); + expect(f.recycler.status('op').status).toBe('closed'); + }); + + it('does not hand off the initiator if a peer fails or an unplanned session arrives', async () => { + const f = fixture(); f.add('initiator'); f.add('peer', 'app-b'); + await f.recycler.prepare('op', f.workspace, 'initiator'); + f.add('new', 'app-b'); rmSync(f.workspace, { recursive: true }); + const result = await f.recycler.finish('op', success); + expect(result.status).toBe('partial'); + expect(f.closed).toEqual(['peer']); expect(f.live.has('initiator')).toBe(true); + expect(f.sessions.get('new')?.status).toBe('active'); + }); + + it('rechecks new peers and reactivated peers before deferred initiator exit', async () => { + const f = fixture(); f.add('initiator'); f.add('peer', 'app-b'); + await f.recycler.prepare('op', f.workspace, 'initiator'); + rmSync(f.workspace, { recursive: true }); + expect((await f.recycler.finish('op', success)).status).toBe('pending'); + f.sessions.get('peer')!.status = 'active'; + await f.runtime.get('app-a')!.recoverDeferred(); + expect(f.closed).toEqual(['peer']); + expect(f.recycler.status('op').status).toBe('partial'); + }); + + it('recovers a lost local close receipt from durable closed state without killing twice', async () => { + const f = fixture(); f.add('one'); await f.recycler.prepare('op', f.workspace); + rmSync(f.workspace, { recursive: true }); + await f.recycler.finish('op', success); + const path = recycleJournalPath(f.dataDir, 'op', { larkAppId: 'app-a', sessionId: 'one' }); + const journal = readRecycleJournal(path)!; + journal.phase = 'closing'; delete journal.closeResult; delete journal.after; + writeFileSync(path, JSON.stringify(journal)); + expect((await f.recycler.finish('op', success)).status).toBe('closed'); + expect(f.closed).toEqual(['one']); + }); + + it('retires a target closed independently after prepare without repeating process close', async () => { + const f = fixture(); f.add('one'); await f.recycler.prepare('op', f.workspace); + f.sessions.get('one')!.status = 'closed'; f.live.delete('one'); + rmSync(f.workspace, { recursive: true }); + expect((await f.recycler.finish('op', success)).status).toBe('closed'); + expect(f.sessions.get('one')?.workspaceRetirement).toMatchObject({ operationId: 'op', workspacePath: f.workspace }); + expect(f.closed).toEqual([]); + }); + + it('keeps a reused path or changed alias from authorizing any stale close', async () => { + const f = fixture(); + const alias = join(f.root, 'alias'); symlinkSync(f.workspace, alias, 'dir'); + f.add('one', 'app-a', alias); await f.recycler.prepare('op', f.workspace); + const other = join(f.root, 'other'); mkdirSync(other); + rmSync(alias); symlinkSync(other, alias, 'dir'); + rmSync(f.workspace, { recursive: true }); + expect((await f.recycler.finish('op', success)).status).toBe('partial'); + expect(f.closed).toEqual([]); + }); + + it('returns partial on missing owners; never writes session files as a fallback', async () => { + const f = fixture(); f.add('offline'); + const before = JSON.stringify([...f.sessions.values()]); + f.runtime.clear(); + expect((await f.recycler.prepare('op', f.workspace)).ok).toBe(false); + expect(JSON.stringify([...f.sessions.values()])).toBe(before); + }); + + it('fails closed for state inside the workspace, changed operation identity, and missing prepare', async () => { + const f = fixture(); const s = f.add('one'); + const inside = new WorkspaceRecycler({ dataDir: f.workspace, sessions: () => [s] }); + await expect(inside.prepare('bad', f.workspace)).rejects.toThrow('recycle_state_inside_workspace'); + const workspace = captureWorkspace(f.workspace); + const request: RecycleRequest = { protocol: WORKSPACE_RECYCLE_PROTOCOL, operationId: 'no-prepare', workspace, target: workspaceTarget(s, workspace)!, peers: [{ sessionId: s.sessionId, larkAppId: s.larkAppId! }] }; + await expect(f.runtime.get('app-a')!.perform('close', request)).rejects.toThrow('recycle_not_prepared'); + await f.recycler.prepare('op', f.workspace); + const other = join(f.root, 'other'); mkdirSync(other); + await expect(f.recycler.prepare('op', other)).rejects.toThrow('recycle_operation_conflict'); + }); + + it('keeps active registrations bounded over repeated create/recycle rounds', async () => { + const f = fixture(); + for (let round = 0; round < 8; round++) { + if (round) mkdirSync(f.workspace); + f.add(`round-${round}-a`); f.add(`round-${round}-b`, 'app-b'); + expect((await f.recycler.prepare(`op-${round}`, f.workspace)).ok).toBe(true); + rmSync(f.workspace, { recursive: true }); + expect((await f.recycler.finish(`op-${round}`, { ...success, eventId: `event-${round}` })).status).toBe('closed'); + expect(f.live.size).toBe(0); + } + expect(f.sessions.size).toBe(16); // retained historical records + }); +}); + +describe('generic lifecycle hook command', () => { + it('accepts equivalent before/after events and rejects conflicting or malformed events', async () => { + const f = fixture(); + vi.stubEnv('SESSION_DATA_DIR', f.dataDir); + vi.stubEnv('BOTMUX_SESSION_ID', ''); + const output = vi.spyOn(console, 'log').mockImplementation(() => {}); + const eventPath = join(f.root, 'event.json'); + const run = async (event: unknown) => { + writeFileSync(eventPath, JSON.stringify(event)); + return runWorkspaceRecycleCommand(['hook', '--event-file', eventPath]); + }; + const event = { protocol: WORKSPACE_RECYCLE_PROTOCOL, operationId: 'hook-op' }; + expect(await run({ ...event, phase: 'before-reclaim', workspacePath: f.workspace })).toBe(0); + expect(JSON.parse(output.mock.calls.at(-1)![0]).status).toBe('prepared'); + expect(await run({ ...event, phase: 'after-reclaim', eventId: 'end-1', outcome: 'succeeded' })).toBe(2); + rmSync(f.workspace, { recursive: true }); + expect(await run({ ...event, phase: 'after-reclaim', eventId: 'end-1', outcome: 'succeeded' })).toBe(0); + expect(JSON.parse(output.mock.calls.at(-1)![0]).status).toBe('closed'); + expect(await run({ ...event, phase: 'after-reclaim', eventId: 'end-2', outcome: 'failed' })).toBe(2); + expect(await run(null)).toBe(2); + expect(await runWorkspaceRecycleCommand(['finish', '--operation', 'hook-op', '--operation', 'duplicate'])).toBe(2); + }); +}); + +describe('durable process readback', () => { + it('does not reattribute a reused stored PID or its children to the closed session', () => { + const f = fixture(); const session = f.add('closed'); session.status = 'closed'; session.pid = process.pid; + const before: RecycleResources = { + registered: true, processes: [{ pid: process.pid, identity: 'previous-process-birth', source: 'worker', state: 'alive' }], + backing: { type: 'pty', state: 'not_applicable' }, errors: [], + }; + const after = rereadRecycleResources(before, session); + expect(after.processes).toEqual([{ pid: process.pid, identity: 'previous-process-birth', source: 'worker', state: 'reused' }]); + expect(resourceResiduals(after)).toEqual([]); + }); +});