From 877bc53669daaffc730135736a3306cb9990d41b Mon Sep 17 00:00:00 2001 From: NIKHIL Date: Sun, 18 Oct 2026 00:45:45 +0530 Subject: [PATCH] fix(daemon): preserve current custody on stale unclaim (#1379) Preserve current custody on a stale unclaim. - Unclaim re-checks the latest session and current binding inside its detach transaction. If a replacement appears during the unguarded tmux cleanup, it returns 409 without changing database custody or emitting a detach event. The message names the session and says OpenRig's tmux metadata may already have been cleared. - A retry on a seat that is already fully released keeps main's result, including persisting and notifying `session.detached` on every call. - Unclaim and release run under typing protection through a custody-only lease. The lease grants no typing, input or launch authority. - A release without `--delete` captures its targets before waiting, releases the unchanged ones, and reports a target that changed through the existing partial result. Already-detached claimed seats stay in `released`. - A release with `--delete` validates the whole captured set before any effect. Membership and session ids are re-read in its deletion transaction, so a seat added or reclaimed during cleanup keeps the rig. --- .../src/domain/rig-lifecycle-service.ts | 242 ++++++--- .../daemon/src/domain/seat-delivery-guard.ts | 65 ++- packages/daemon/src/routes/sessions.ts | 2 +- .../test/unclaim-ownership-native.test.ts | 143 +++++ .../daemon/test/unclaim-ownership.test.ts | 507 ++++++++++++++++++ 5 files changed, 878 insertions(+), 81 deletions(-) create mode 100644 packages/daemon/test/unclaim-ownership-native.test.ts create mode 100644 packages/daemon/test/unclaim-ownership.test.ts diff --git a/packages/daemon/src/domain/rig-lifecycle-service.ts b/packages/daemon/src/domain/rig-lifecycle-service.ts index 9cc119d88..1f56ff6f5 100644 --- a/packages/daemon/src/domain/rig-lifecycle-service.ts +++ b/packages/daemon/src/domain/rig-lifecycle-service.ts @@ -6,6 +6,7 @@ import type { EventBus } from "./event-bus.js"; import type { TmuxAdapter } from "../adapters/tmux.js"; import type { QueueRepository } from "./queue-repository.js"; import { findOtherSessionOwner } from "./session-owner.js"; +import { DeliveryGuardError } from "./seat-delivery-guard.js"; type ClaimedSessionRow = { session_id: string; @@ -34,6 +35,8 @@ type RigReleaseRow = { session_id: string | null; session_name: string | null; session_origin: string | null; + session_status: string | null; + tmux_session: string | null; }; type FallbackValidationFailure = { @@ -53,7 +56,7 @@ export type UnclaimSessionResult = } | { ok: false; - code: "session_not_found" | "session_ambiguous"; + code: "session_not_found" | "session_ambiguous" | "session_not_current"; error: string; }; @@ -167,6 +170,7 @@ export class RigLifecycleService { this.tmuxAdapter = deps.tmuxAdapter ?? null; } + /** Release a current claimed session without killing its external process. */ async unclaimSession(sessionRef: string): Promise { const exactId = this.db.prepare(` SELECT s.id AS session_id, s.session_name, n.id AS node_id, n.rig_id, n.logical_id, b.tmux_session @@ -203,8 +207,40 @@ export class RigLifecycleService { const session = matches[0]!; + const guard = this.tmuxAdapter?.deliveryGuard; + const success: UnclaimSessionResult = { + ok: true, rigId: session.rig_id, nodeId: session.node_id, logicalId: session.logical_id, + sessionId: session.session_id, sessionName: session.session_name, + }; + const isReleased = () => { + const latest = this.db.prepare( + "SELECT status FROM sessions WHERE node_id = ? ORDER BY created_at DESC, id DESC LIMIT 1", + ).get(session.node_id) as { status: string } | undefined; + return !this.sessionRegistry.getBindingForNode(session.node_id) && latest?.status === "detached"; + }; + const releasedRetry = isReleased(); + if (!releasedRetry && guard && !guard.ownsCustody(session.node_id)) { + return guard.custody([session.node_id], () => this.unclaimSession(session.session_id)); + } + /** Read fresh custody both before metadata cleanup and at the detach commit. */ + const isCurrent = () => { + const latest = this.db.prepare( + "SELECT id FROM sessions WHERE node_id = ? ORDER BY created_at DESC, id DESC LIMIT 1", + ).get(session.node_id) as { id: string } | undefined; + const binding = this.sessionRegistry.getBindingForNode(session.node_id); + return latest?.id === session.session_id + && (!binding?.tmuxSession || binding.tmuxSession === session.session_name); + }; + if (!releasedRetry && !isCurrent()) { + return { + ok: false, + code: "session_not_current", + error: `Claimed session '${session.session_id}' no longer owns this seat. No metadata or custody was changed.`, + }; + } + // Best-effort: clear OpenRig-owned tmux metadata from the adopted session. - if (this.tmuxAdapter && session.tmux_session) { + if (!releasedRetry && this.tmuxAdapter && session.tmux_session) { const keys = [ "@rigged_node_id", "@rigged_session_name", @@ -224,9 +260,12 @@ export class RigLifecycleService { let persistedSeq = 0; let persistedAt = ""; const tx = this.db.transaction(() => { + if (releasedRetry ? !isReleased() : !isCurrent()) return false; this.sessionRegistry.markDetached(session.session_id); - this.sessionRegistry.clearBinding(session.node_id); - this.discoveryRepo.releaseClaimByNodeId(session.node_id); + if (!releasedRetry) { + this.sessionRegistry.clearBinding(session.node_id); + this.discoveryRepo.releaseClaimByNodeId(session.node_id); + } const event = this.eventBus.persistWithinTransaction({ type: "session.detached", rigId: session.rig_id, @@ -235,8 +274,15 @@ export class RigLifecycleService { }); persistedSeq = event.seq; persistedAt = event.createdAt; + return true; }); - tx(); + if (!tx()) { + return { + ok: false, + code: "session_not_current", + error: `Claimed session '${session.session_id}' no longer owns this seat. Database custody was not changed. OpenRig tmux metadata on '${session.tmux_session ?? session.session_name}' may already have been cleared.`, + }; + } this.eventBus.notifySubscribers({ type: "session.detached", @@ -247,30 +293,27 @@ export class RigLifecycleService { createdAt: persistedAt, }); - return { - ok: true, - rigId: session.rig_id, - nodeId: session.node_id, - logicalId: session.logical_id, - sessionId: session.session_id, - sessionName: session.session_name, - }; + return success; } + /** Release adopted seats under their lifecycle leases, optionally deleting the rig. */ async releaseRig(rigId: string, opts?: { delete?: boolean }): Promise { const rig = this.rigRepo.getRig(rigId); if (!rig) { return { ok: false, code: "rig_not_found", error: `Rig '${rigId}' not found.` }; } - const rows = this.db.prepare(` + const readRows = () => this.db.prepare(` SELECT n.id AS node_id, n.logical_id, s.id AS session_id, s.session_name, - s.origin AS session_origin + s.origin AS session_origin, + s.status AS session_status, + b.tmux_session FROM nodes n + LEFT JOIN bindings b ON b.node_id = n.id LEFT JOIN sessions s ON s.id = ( SELECT s2.id FROM sessions s2 @@ -281,6 +324,7 @@ export class RigLifecycleService { WHERE n.rig_id = ? ORDER BY n.logical_id `).all(rigId) as RigReleaseRow[]; + const rows = readRows(); const launchedLogicalIds = rows .filter((row) => row.session_id && row.session_origin === "launched") @@ -294,73 +338,119 @@ export class RigLifecycleService { }; } - const released: Array<{ - nodeId: string; - logicalId: string; - sessionId: string; - sessionName: string; - }> = []; - const failed: Array<{ - nodeId: string; - logicalId: string; - sessionId: string | null; - sessionName: string | null; - error: string; - }> = []; - - for (const row of rows) { - if (!row.session_id || row.session_origin !== "claimed") continue; - const result = await this.unclaimSession(row.session_id); - if (result.ok) { - released.push({ - nodeId: result.nodeId, - logicalId: result.logicalId, - sessionId: result.sessionId, - sessionName: result.sessionName, - }); - } else { - failed.push({ - nodeId: row.node_id, - logicalId: row.logical_id, - sessionId: row.session_id, - sessionName: row.session_name, - error: result.error, - }); + const targets = opts?.delete ? rows : rows.filter(row => row.session_origin === "claimed"); + const leaseTargets = opts?.delete ? targets : targets.filter(row => row.session_status !== "detached" || row.tmux_session !== null); + const changed = new Map(); + const matches = (target: RigReleaseRow, row?: RigReleaseRow) => !!row + && row.session_id === target.session_id && row.session_name === target.session_name + && row.session_origin === target.session_origin && row.logical_id === target.logical_id + && row.session_status === target.session_status && row.tmux_session === target.tmux_session; + const release = async (): Promise => { + const current = new Map(readRows().map(row => [row.node_id, row])); + for (const target of targets) { + if (!matches(target, current.get(target.node_id))) changed.set(target.node_id, "Release target changed while waiting; no cleanup performed for this seat."); + } + if (opts?.delete && (current.size !== rows.length || changed.size > 0)) { + throw new DeliveryGuardError("guard_target_changed", "Release target changed while waiting; no cleanup performed."); } - } - let deleted = false; - if (failed.length === 0 && opts?.delete) { - let persistedSeq = 0; - let persistedAt = ""; - const tx = this.db.transaction(() => { - const event = this.eventBus.persistWithinTransaction({ - type: "rig.deleted", - rigId, + const released: Array<{ + nodeId: string; + logicalId: string; + sessionId: string; + sessionName: string; + }> = []; + const failed: Array<{ + nodeId: string; + logicalId: string; + sessionId: string | null; + sessionName: string | null; + error: string; + }> = []; + + for (const row of targets) { + if (!row.session_id || row.session_origin !== "claimed") continue; + // Unleased detached seats may be reclaimed while another seat is cleaned. + if (!matches(row, readRows().find(current => current.node_id === row.node_id))) { + changed.set(row.node_id, "Release target changed; no cleanup performed for this seat."); + } + const error = changed.get(row.node_id); + if (error) { + failed.push({ nodeId: row.node_id, logicalId: row.logical_id, sessionId: row.session_id, sessionName: row.session_name, error }); + continue; + } + const result = await this.unclaimSession(row.session_id).catch((error: unknown) => { + if (!(error instanceof DeliveryGuardError)) throw error; + return { ok: false as const, error: error.message }; }); - persistedSeq = event.seq; - persistedAt = event.createdAt; - this.rigRepo.deleteRig(rigId); - }); - tx(); + if (result.ok) { + released.push({ + nodeId: result.nodeId, + logicalId: result.logicalId, + sessionId: result.sessionId, + sessionName: result.sessionName, + }); + } else { + failed.push({ + nodeId: row.node_id, + logicalId: row.logical_id, + sessionId: row.session_id, + sessionName: row.session_name, + error: result.error, + }); + } + } - this.eventBus.notifySubscribers({ - type: "rig.deleted", + let deleted = false; + if (failed.length === 0 && opts?.delete) { + let persistedSeq = 0; + let persistedAt = ""; + const tx = this.db.transaction(() => { + const current = readRows(); + const conflicts = current.filter(row => !rows.some(captured => captured.node_id === row.node_id && captured.session_id === row.session_id)); + const missing = rows.filter(captured => !current.some(row => row.node_id === captured.node_id)); + if (conflicts.length > 0 || missing.length > 0) { + for (const row of [...conflicts, ...missing]) { + const captured = rows.find(target => target.node_id === row.node_id); + failed.push({ nodeId: row.node_id, logicalId: row.logical_id, + sessionId: captured?.session_id ?? null, sessionName: captured?.session_name ?? null, + error: "Rig membership or session changed during cleanup; rig was not deleted.", + }); + } + return false; + } + const event = this.eventBus.persistWithinTransaction({ + type: "rig.deleted", + rigId, + }); + persistedSeq = event.seq; + persistedAt = event.createdAt; + this.rigRepo.deleteRig(rigId); + return true; + }); + if (tx()) { + this.eventBus.notifySubscribers({ + type: "rig.deleted", + rigId, + seq: persistedSeq, + createdAt: persistedAt, + }); + deleted = true; + } + } + + return { + ok: true, + status: failed.length > 0 ? "partial" : "ok", rigId, - seq: persistedSeq, - createdAt: persistedAt, - }); - deleted = true; - } - - return { - ok: true, - status: failed.length > 0 ? "partial" : "ok", - rigId, - deleted, - released, - failed, + deleted, + released, + failed, + }; }; + const guard = this.tmuxAdapter?.deliveryGuard; + return guard ? guard.custody(leaseTargets.map(row => row.node_id), release, + opts?.delete ? undefined : (nodeId, error) => { changed.set(nodeId, error.message); }) : release(); } async removeNode( diff --git a/packages/daemon/src/domain/seat-delivery-guard.ts b/packages/daemon/src/domain/seat-delivery-guard.ts index 6688b3486..4b44f5a77 100644 --- a/packages/daemon/src/domain/seat-delivery-guard.ts +++ b/packages/daemon/src/domain/seat-delivery-guard.ts @@ -12,7 +12,8 @@ export interface GuardTarget { interface Lease { target: GuardTarget; active: boolean; - origin: "automatic" | "human"; + serialOwner?: Lease; + origin: "automatic" | "human" | "custody"; lifecycle?: boolean; } @@ -118,7 +119,7 @@ export class SeatDeliveryGuard { const bound = this.target(name); const inherited = this.scope.getStore()?.get(bound.nodeId); if (inherited?.active && inherited.target.nodeId === bound.nodeId) { - this.assertCurrent(name, inherited); + this.checkInput(name); return fn(); } return this.serial(bound.nodeId, async () => { @@ -130,6 +131,7 @@ export class SeatDeliveryGuard { throw new DeliveryGuardError("typing_guard_enabled", "Automatic input is paused for this seat. Disable its typing guard explicitly before this writing operation."); } const lease: Lease = { target: bound, active: true, origin: "automatic" }; + lease.serialOwner = lease; try { return await this.scope.run(new Map([...(this.scope.getStore() ?? []), [lease.target.nodeId, lease]]), fn); } finally { lease.active = false; } }); @@ -142,6 +144,57 @@ export class SeatDeliveryGuard { return true; } + /** Custody cleanup can join a writing lifecycle lease without granting input permission. */ + ownsCustody(nodeId: string): boolean { + const lease = this.scope.getStore()?.get(nodeId); + if (!lease?.active || (lease.origin !== "custody" && !lease.lifecycle)) return false; + this.assertCurrent(nodeId, lease); + return true; + } + + /** Serialize metadata/custody cleanup even when typing is paused. Capture every + * target before waiting, then validate all of them before the first effect. */ + async custody(nodeIds: string[], fn: () => Promise, onTargetChanged?: (nodeId: string, error: DeliveryGuardError) => void): Promise { + const targets = [...new Set(nodeIds)].sort().map(id => this.target(id)); + const excluded = new Set(); + const validate = (target: GuardTarget, inherited?: Lease): boolean => { + if (excluded.has(target.nodeId)) return false; + try { + if (inherited) this.assertCurrent(target.nodeId, inherited); + if (!this.same(target, this.target(target.nodeId))) { + throw new DeliveryGuardError("guard_target_changed", "Custody target changed while waiting; no cleanup performed for this seat."); + } + return true; + } catch (error) { + if (!onTargetChanged || !(error instanceof DeliveryGuardError)) throw error; + excluded.add(target.nodeId); + onTargetChanged(target.nodeId, error); + return false; + } + }; + const acquire = async (index: number): Promise => { + const target = targets[index]; + if (!target) { for (const captured of targets) validate(captured); return fn(); } + const inherited = this.scope.getStore()?.get(target.nodeId); + const run = async () => { + const lease: Lease = { target, active: true, origin: "custody" }; + lease.serialOwner = inherited?.active && inherited.serialOwner?.active ? inherited.serialOwner : lease; + try { + return await this.scope.run(new Map([...(this.scope.getStore() ?? []), [target.nodeId, lease]]), () => acquire(index + 1)); + } finally { lease.active = false; } + }; + if (inherited?.active && inherited.serialOwner?.active) { + if (!validate(target, inherited)) return acquire(index + 1); + return inherited.origin === "custody" || inherited.lifecycle ? acquire(index + 1) : run(); + } + return this.serial(target.nodeId, async () => { + if (!validate(target)) return acquire(index + 1); + return run(); + }); + }; + return acquire(0); + } + /** Multi-seat restore takes leases in stable order before any rig mutation. * Nested per-seat launch joins these leases; it must not reacquire them. */ async lifecycle(nodeIds: string[], fn: () => Promise): Promise { @@ -178,6 +231,9 @@ export class SeatDeliveryGuard { const lease = this.scope.getStore()?.get(target.nodeId); if (!lease) throw new DeliveryGuardError("guard_lease_required", "Input requires an active operation lease."); this.assertCurrent(name, lease); + if (lease.origin === "custody") { + throw new DeliveryGuardError("guard_lease_required", "Custody cleanup does not authorize input or writing lifecycle operations."); + } } /** Reconciliation is a synchronous DB transaction, not a nested input operation. @@ -200,14 +256,15 @@ export class SeatDeliveryGuard { async input(name: string, fn: () => Promise): Promise { const target = this.target(name); const lease = this.scope.getStore()?.get(target.nodeId); - if (lease?.active) { this.assertCurrent(name, lease); return fn(); } + if (lease?.active) { this.checkInput(name); return fn(); } return this.operation(name, fn); } /** Internal broker path only; never an option accepted by the send HTTP route. */ async humanInput(name: string, fn: () => Promise): Promise { const target = this.target(name); - const lease: Lease = { target, active: true, origin: "human" }; + const inherited = this.scope.getStore()?.get(target.nodeId); + const lease: Lease = { target, active: true, serialOwner: inherited?.active ? inherited.serialOwner : undefined, origin: "human" }; this.humanLeases.add(lease); try { return await this.scope.run(new Map([...(this.scope.getStore() ?? []), [lease.target.nodeId, lease]]), fn); } finally { lease.active = false; this.humanLeases.delete(lease); } diff --git a/packages/daemon/src/routes/sessions.ts b/packages/daemon/src/routes/sessions.ts index 039e8a37b..3fd06f838 100644 --- a/packages/daemon/src/routes/sessions.ts +++ b/packages/daemon/src/routes/sessions.ts @@ -665,7 +665,7 @@ sessionAdminRoutes.post("/:sessionRef/unclaim", terminalAuthGuard(), async (c) = const result = await rigLifecycleService.unclaimSession(sessionRef); if (!result.ok) { - const status = result.code === "session_ambiguous" ? 409 : 404; + const status = result.code === "session_not_found" ? 404 : 409; return c.json(result, status); } diff --git a/packages/daemon/test/unclaim-ownership-native.test.ts b/packages/daemon/test/unclaim-ownership-native.test.ts new file mode 100644 index 000000000..aa206da6f --- /dev/null +++ b/packages/daemon/test/unclaim-ownership-native.test.ts @@ -0,0 +1,143 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { execFile, execFileSync } from "node:child_process"; +import { promisify } from "node:util"; +import type Database from "better-sqlite3"; +import { createDb } from "../src/db/connection.js"; +import { migrate } from "../src/db/migrate.js"; +import { ALL_MIGRATIONS } from "../src/db/all-migrations.js"; +import { TmuxAdapter } from "../src/adapters/tmux.js"; +import { SeatDeliveryGuard, resolveGuardTarget } from "../src/domain/seat-delivery-guard.js"; +import { createTestApp } from "./helpers/test-app.js"; + +let nativeAvailable = process.platform !== "win32"; +try { execFileSync("tmux", ["-V"], { timeout: 2000, stdio: "ignore" }); } catch { nativeAvailable = false; } + +describe.skipIf(!nativeAvailable)("claimed-session custody with native tmux", () => { + let db: Database.Database; + let setup: ReturnType; + let run: (...args: string[]) => Promise; + + beforeEach(async () => { + const socket = `openrig-unclaim-test-${process.pid}-${Date.now()}`; + const execute = promisify(execFile); + run = async (...args) => (await execute("tmux", ["-L", socket, "-f", "/dev/null", ...args], { timeout: 5000 })).stdout; + db = createDb(); + migrate(db, ALL_MIGRATIONS); + const adapter = new TmuxAdapter(async () => { throw new Error("expected argv transport"); }, undefined, + async argv => run(...argv.slice(1))); + adapter.deliveryGuard = new SeatDeliveryGuard(db, target => resolveGuardTarget(db, target)); + setup = createTestApp(db, { tmux: adapter }); + await run("new-session", "-d", "-s", "manual-adoption", "cat"); + }); + afterEach(async () => { + await run("kill-server").catch(() => {}); + db.close(); + }); + + async function claimSeat() { + const rig = setup.rigRepo.createRig("native-adoption"); + const node = setup.rigRepo.addNode(rig.id, "external.helper", { runtime: "terminal" }); + const pane = (await run("display-message", "-p", "-t", "=manual-adoption:", "#{pane_id}")).trim(); + const discovered = setup.discoveryRepo.upsertDiscoveredSession({ + tmuxSession: "manual-adoption", tmuxPane: pane, cwd: "/tmp", + activeCommand: "cat", runtimeHint: "terminal", confidence: "high", + }); + const bind = () => setup.claimService.bind({ discoveredId: discovered.id, rigId: rig.id, logicalId: node.logicalId }); + const first = await bind(); + expect(first.ok).toBe(true); + if (!first.ok) throw new Error(first.error); + return { node, discovered, first, bind }; + } + + async function livePanePid(): Promise { + const reading = (await run("display-message", "-p", "-t", "=manual-adoption:", "#{pane_pid}")).trim(); + expect(reading).toMatch(/^[1-9]\d*$/); + const pid = Number(reading); + expect(Number.isSafeInteger(pid)).toBe(true); + expect(() => process.kill(pid, 0)).not.toThrow(); + return pid; + } + + it("preserves the physical session and its new metadata after an old-ID retry", async () => { + const { node, first, bind } = await claimSeat(); + expect((await setup.rigLifecycleService.unclaimSession(first.sessionId)).ok).toBe(true); + expect((await bind()).ok).toBe(true); + const before = await run("show-options", "-qv", "-t", "=manual-adoption:", "@rigged_node_id"); + expect(before.trim()).toBe(node.id); + const pid = await livePanePid(); + const response = await setup.app.request(`/api/sessions/${first.sessionId}/unclaim`, { method: "POST" }); + expect(await run("show-options", "-qv", "-t", "=manual-adoption:", "@rigged_node_id")).toBe(before); + expect(await livePanePid()).toBe(pid); + expect(() => process.kill(pid, 0)).not.toThrow(); + expect(response.status).toBe(409); + expect(setup.sessionRegistry.getBindingForNode(node.id)?.tmuxSession).toBe("manual-adoption"); + }); + + it("ordinary unclaim clears only managed metadata and leaves the external process alive", async () => { + const { node, discovered, first } = await claimSeat(); + expect((await run("show-options", "-qv", "-t", "=manual-adoption:", "@rigged_node_id")).trim()).toBe(node.id); + const pid = await livePanePid(); + const response = await setup.app.request(`/api/sessions/${first.sessionId}/unclaim`, { method: "POST" }); + expect(response.status).toBe(200); + expect((await run("show-options", "-qv", "-t", "=manual-adoption:", "@rigged_node_id")).trim()).toBe(""); + expect(await livePanePid()).toBe(pid); + expect(() => process.kill(pid, 0)).not.toThrow(); + expect(setup.sessionRegistry.getBindingForNode(node.id)).toBeNull(); + expect(setup.discoveryRepo.getDiscoveredSession(discovered.id)?.status).toBe("active"); + }); + + it("keeps successor metadata through a real claim lease and a queued stale unclaim", async () => { + const { node, discovered, first, bind } = await claimSeat(); + expect((await setup.rigLifecycleService.unclaimSession(first.sessionId)).ok).toBe(true); + const pid = await livePanePid(); + let entered!: () => void, finish!: () => void; + const paused = new Promise(resolve => { entered = resolve; }); + const resume = new Promise(resolve => { finish = resolve; }); + const adapter = setup.tmuxAdapter; + const setOption = adapter.setSessionOption.bind(adapter); + let waiting = true; + vi.spyOn(adapter, "setSessionOption").mockImplementation(async (name, key, value) => { + const result = await setOption(name, key, value); + if (waiting && key === "@rigged_node_id" && value === node.id) { + waiting = false; + entered(); + await resume; + } + return result; + }); + const claim = bind(); + await paused; + expect((await run("show-options", "-qv", "-t", "=manual-adoption:", "@rigged_node_id")).trim()).toBe(node.id); + const binding = setup.sessionRegistry.getBindingForNode(node.id); + const discovery = setup.discoveryRepo.getDiscoveredSession(discovered.id); + const detached = db.prepare("SELECT count(*) AS n FROM events WHERE type = 'session.detached'").get(); + const notify = vi.spyOn(setup.eventBus, "notifySubscribers"); + let completed = false; + const request = setup.app.request(`/api/sessions/${first.sessionId}/unclaim`, { method: "POST" }) + .then(response => { completed = true; return response; }); + await new Promise(resolve => setImmediate(resolve)); + const completedDuringClaim = completed; + finish(); + const successor = await claim; + const response = await request; + expect(successor.ok).toBe(true); + if (!successor.ok) throw new Error(successor.error); + const rig = setup.rigRepo.getRig(node.rigId)!; + const expected: Record = { + "@rigged_node_id": node.id, "@rigged_session_name": "manual-adoption", + "@rigged_rig_id": node.rigId, "@rigged_rig_name": rig.rig.name, "@rigged_logical_id": node.logicalId, + }; + for (const [key, value] of Object.entries(expected)) { + expect((await run("show-options", "-qv", "-t", "=manual-adoption:", key)).trim()).toBe(value); + } + expect(response.status).toBe(409); + expect(completedDuringClaim).toBe(false); + expect(setup.sessionRegistry.getBindingForNode(node.id)).toEqual(binding); + expect(setup.discoveryRepo.getDiscoveredSession(discovered.id)).toEqual(discovery); + expect(db.prepare("SELECT status FROM sessions WHERE id = ?").get(successor.sessionId)).toEqual({ status: "running" }); + expect(db.prepare("SELECT count(*) AS n FROM events WHERE type = 'session.detached'").get()).toEqual(detached); + expect(notify).not.toHaveBeenCalledWith(expect.objectContaining({ type: "session.detached" })); + expect(await livePanePid()).toBe(pid); + expect(() => process.kill(pid, 0)).not.toThrow(); + }); +}); diff --git a/packages/daemon/test/unclaim-ownership.test.ts b/packages/daemon/test/unclaim-ownership.test.ts new file mode 100644 index 000000000..47979994d --- /dev/null +++ b/packages/daemon/test/unclaim-ownership.test.ts @@ -0,0 +1,507 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import type Database from "better-sqlite3"; +import { createTestApp } from "./helpers/test-app.js"; +import { createDb } from "../src/db/connection.js"; +import { migrate } from "../src/db/migrate.js"; +import { ALL_MIGRATIONS } from "../src/db/all-migrations.js"; +import { SeatDeliveryGuard, resolveGuardTarget } from "../src/domain/seat-delivery-guard.js"; + +describe("unclaim preserves a successor's custody", () => { + let db: Database.Database; + let setup: ReturnType; + + beforeEach(() => { + db = createDb(); + migrate(db, ALL_MIGRATIONS); + setup = createTestApp(db); + setup.tmuxAdapter.deliveryGuard = new SeatDeliveryGuard(db, target => resolveGuardTarget(db, target)); + }); + afterEach(() => { db.close(); }); + + async function claimedSeat() { + const rig = setup.rigRepo.createRig("adoption-fixture"); + const node = setup.rigRepo.addNode(rig.id, "external.helper", { runtime: "terminal" }); + const discovered = setup.discoveryRepo.upsertDiscoveredSession({ + tmuxSession: "manual-adoption", tmuxPane: "%1", cwd: "/tmp", + activeCommand: "bash", runtimeHint: "terminal", confidence: "high", + }); + const claim = await setup.claimService.bind({ discoveredId: discovered.id, rigId: rig.id, logicalId: node.logicalId }); + expect(claim.ok).toBe(true); + if (!claim.ok) throw new Error(claim.error); + return { rig, node, discovered, claim }; + } + + function barrier() { + let resolve!: () => void; + const promise = new Promise(done => { resolve = done; }); + return { promise, resolve }; + } + + it.each([true, false])("does not let a replayed old session ID release a newly reclaimed seat (guard=%s)", async guarded => { + const { rig, node, discovered, claim } = await claimedSeat(); + const first = await setup.app.request(`/api/sessions/${claim.sessionId}/unclaim`, { method: "POST" }); + expect(first.status).toBe(200); + const successor = await setup.claimService.bind({ discoveredId: discovered.id, rigId: rig.id, logicalId: node.logicalId }); + expect(successor.ok).toBe(true); + const binding = setup.sessionRegistry.getBindingForNode(node.id); + const discovery = setup.discoveryRepo.getDiscoveredSession(discovered.id); + if (!guarded) setup.tmuxAdapter.deliveryGuard = undefined; + vi.mocked(setup.tmuxAdapter.setSessionOption).mockClear(); + + const replay = await setup.app.request(`/api/sessions/${claim.sessionId}/unclaim`, { method: "POST" }); + + expect(setup.sessionRegistry.getBindingForNode(node.id)).toEqual(binding); + expect(setup.discoveryRepo.getDiscoveredSession(discovered.id)).toEqual(discovery); + expect(setup.tmuxAdapter.setSessionOption).not.toHaveBeenCalled(); + expect(replay.status).toBe(409); + }); + + it("does not let an old claimed session ID detach a launched successor", async () => { + const { rig, node, claim } = await claimedSeat(); + expect((await setup.rigLifecycleService.unclaimSession(claim.sessionId)).ok).toBe(true); + const launched = await setup.nodeLauncher.launchNode(rig.id, node.logicalId); + expect(launched.ok).toBe(true); + const binding = setup.sessionRegistry.getBindingForNode(node.id); + vi.mocked(setup.tmuxAdapter.setSessionOption).mockClear(); + + const res = await setup.app.request(`/api/sessions/${claim.sessionId}/unclaim`, { method: "POST" }); + + expect(setup.sessionRegistry.getBindingForNode(node.id)).toEqual(binding); + expect(setup.tmuxAdapter.setSessionOption).not.toHaveBeenCalled(); + expect(res.status).toBe(409); + }); + + it.each(["unclaim", "release"] as const)("serializes %s metadata cleanup with a concurrent rebind", async operation => { + const { rig, node, discovered, claim } = await claimedSeat(); + const entered = barrier(); + const finish = barrier(); + vi.mocked(setup.tmuxAdapter.setSessionOption).mockImplementationOnce(async () => { + entered.resolve(); + await finish.promise; + return { ok: true }; + }); + const request = setup.app.request(operation === "unclaim" + ? `/api/sessions/${claim.sessionId}/unclaim` + : `/api/rigs/${rig.id}/release`, { method: "POST" }); + await entered.promise; + let rebindEntered = false; + const guard = setup.tmuxAdapter.deliveryGuard!; + const rebind = guard.lifecycle([node.id], async () => { + rebindEntered = true; + // This is the same binding replacement performed by a lifecycle operation. + const next = setup.sessionRegistry.registerClaimedSession(node.id, "replacement-adoption"); + setup.sessionRegistry.updateBinding(node.id, { tmuxSession: next.sessionName, tmuxPane: "%2" }); + setup.discoveryRepo.markClaimed(discovered.id, node.id); + guard.rebindLifecycle(node.id); + return setup.sessionRegistry.getBindingForNode(node.id); + }).then(binding => ({ binding, error: null }), error => ({ binding: null, error })); + // An unguarded releaser allows this operation to enter before its cleanup ends. + await new Promise(resolve => setImmediate(resolve)); + const enteredDuringCleanup = rebindEntered; + finish.resolve(); + await request; + const rebound = await rebind; + expect(enteredDuringCleanup).toBe(false); + expect(rebound.error).toMatchObject({ code: "guard_target_changed" }); + expect(setup.sessionRegistry.getBindingForNode(node.id)).toBeNull(); + const successor = await setup.claimService.bind({ discoveredId: discovered.id, rigId: rig.id, logicalId: node.logicalId }); + expect(successor.ok).toBe(true); + expect(setup.sessionRegistry.getBindingForNode(node.id)?.tmuxSession).toBe("manual-adoption"); + expect(setup.discoveryRepo.getDiscoveredSession(discovered.id)?.status).toBe("claimed"); + }); + + it("rejects an unclaim whose target changes while it waits for a lifecycle lease", async () => { + const { node, claim } = await claimedSeat(); + const entered = barrier(); + const finish = barrier(); + const guard = setup.tmuxAdapter.deliveryGuard!; + const lifecycle = guard.lifecycle([node.id], async () => { + entered.resolve(); + await finish.promise; + setup.sessionRegistry.registerClaimedSession(node.id, "replacement-adoption"); + setup.sessionRegistry.updateBinding(node.id, { tmuxSession: "replacement-adoption", tmuxPane: "%2" }); + guard.rebindLifecycle(node.id); + }); + await entered.promise; + vi.mocked(setup.tmuxAdapter.setSessionOption).mockClear(); + const request = setup.app.request(`/api/sessions/${claim.sessionId}/unclaim`, { method: "POST" }); + await new Promise(resolve => setImmediate(resolve)); + finish.resolve(); + await lifecycle; + const binding = setup.sessionRegistry.getBindingForNode(node.id); + const response = await request; + expect(response.status).toBe(409); + expect(setup.sessionRegistry.getBindingForNode(node.id)).toEqual(binding); + expect(setup.tmuxAdapter.setSessionOption).not.toHaveBeenCalled(); + }); + + it.each(["claim", "launch", "binding"] as const)("preserves a concurrent %s during unguarded metadata cleanup", async replacement => { + const { rig, node, discovered, claim } = await claimedSeat(); + setup.tmuxAdapter.deliveryGuard = undefined; + const entered = barrier(); + const finish = barrier(); + vi.mocked(setup.tmuxAdapter.setSessionOption).mockImplementationOnce(async () => { + entered.resolve(); + await finish.promise; + return { ok: true }; + }); + const request = setup.app.request(`/api/sessions/${claim.sessionId}/unclaim`, { method: "POST" }); + await entered.promise; + if (replacement === "binding") { + setup.sessionRegistry.updateBinding(node.id, { tmuxSession: "replacement-adoption", tmuxPane: "%2" }); + } else { + expect((await setup.rigLifecycleService.unclaimSession(claim.sessionId)).ok).toBe(true); + const successor = replacement === "claim" + ? await setup.claimService.bind({ discoveredId: discovered.id, rigId: rig.id, logicalId: node.logicalId }) + : await setup.nodeLauncher.launchNode(rig.id, node.logicalId); + expect(successor.ok).toBe(true); + } + const binding = setup.sessionRegistry.getBindingForNode(node.id); + const discovery = setup.discoveryRepo.getDiscoveredSession(discovered.id); + const sessions = db.prepare("SELECT * FROM sessions WHERE node_id = ? ORDER BY id").all(node.id); + const events = db.prepare("SELECT * FROM events ORDER BY seq").all(); + finish.resolve(); + const response = await request; + expect(response.status).toBe(409); + expect(await response.json()).toMatchObject({ + code: "session_not_current", + error: `Claimed session '${claim.sessionId}' no longer owns this seat. Database custody was not changed. OpenRig tmux metadata on 'manual-adoption' may already have been cleared.`, + }); + expect(setup.sessionRegistry.getBindingForNode(node.id)).toEqual(binding); + expect(setup.discoveryRepo.getDiscoveredSession(discovered.id)).toEqual(discovery); + expect(db.prepare("SELECT * FROM sessions WHERE node_id = ? ORDER BY id").all(node.id)).toEqual(sessions); + expect(db.prepare("SELECT * FROM events ORDER BY seq").all()).toEqual(events); + }); + + it.each(["unclaim", "release", "delete"] as const)("allows %s custody cleanup with typing protection enabled", async operation => { + const { rig, node, discovered, claim } = await claimedSeat(); + await setup.tmuxAdapter.deliveryGuard!.set(node.id, true, "operator", "fixture pause"); + vi.mocked(setup.tmuxAdapter.setSessionOption).mockClear(); + const response = await setup.app.request(operation === "unclaim" + ? `/api/sessions/${claim.sessionId}/unclaim` : `/api/rigs/${rig.id}/release`, { + method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify({ delete: operation === "delete" }), + }); + expect(response.status).toBe(200); + expect(setup.sessionRegistry.getBindingForNode(node.id)).toBeNull(); + expect(setup.discoveryRepo.getDiscoveredSession(discovered.id)?.status).toBe("active"); + expect(setup.tmuxAdapter.killSession).not.toHaveBeenCalled(); + if (operation !== "delete") expect(setup.tmuxAdapter.deliveryGuard!.preference(node.id).desired).toBe(true); + }); + + it("keeps an old-ID retry harmless after a newer claim was released", async () => { + const { rig, node, discovered, claim } = await claimedSeat(); + const original = await setup.rigLifecycleService.unclaimSession(claim.sessionId); + const next = await setup.claimService.bind({ discoveredId: discovered.id, rigId: rig.id, logicalId: node.logicalId }); + if (!next.ok) throw new Error(next.error); + await setup.rigLifecycleService.unclaimSession(next.sessionId); + const events = db.prepare("SELECT * FROM events ORDER BY seq").all(); + const notify = vi.spyOn(setup.eventBus, "notifySubscribers"); + vi.mocked(setup.tmuxAdapter.setSessionOption).mockClear(); + const response = await setup.app.request(`/api/sessions/${claim.sessionId}/unclaim`, { method: "POST" }); + expect(response.status).toBe(200); + expect(await response.json()).toEqual(original); + expect(setup.sessionRegistry.getBindingForNode(node.id)).toBeNull(); + expect(setup.tmuxAdapter.setSessionOption).not.toHaveBeenCalled(); + const second = await setup.app.request(`/api/sessions/${claim.sessionId}/unclaim`, { method: "POST" }); + expect(await second.json()).toEqual(original); + expect(db.prepare("SELECT * FROM events ORDER BY seq").all()).toHaveLength(events.length + 2); + expect(notify).toHaveBeenCalledTimes(2); + expect(notify).toHaveBeenCalledWith(expect.objectContaining({ type: "session.detached", nodeId: node.id, sessionName: "manual-adoption" })); + }); + + it.each([false, true])("does not grant input or launch permission through a custody lease (protected=%s)", async protectedSeat => { + const { rig, node } = await claimedSeat(); + const guard = setup.tmuxAdapter.deliveryGuard!; + if (protectedSeat) await guard.set(node.id, true, "operator", "fixture pause"); + const write = vi.fn(async () => {}); + await guard.custody([node.id], async () => { + expect(guard.ownsCustody(node.id)).toBe(true); + expect(guard.ownsLifecycle(node.id)).toBe(false); + expect(() => guard.checkInput(node.id)).toThrow("Custody cleanup does not authorize"); + await expect(guard.input(node.id, write)).rejects.toMatchObject({ code: "guard_lease_required" }); + await expect(guard.operation(node.id, write)).rejects.toMatchObject({ code: "guard_lease_required" }); + await expect(guard.lifecycle([node.id], write)).rejects.toMatchObject({ code: "guard_lease_required" }); + await expect(setup.nodeLauncher.launchNode(rig.id, node.logicalId)).rejects.toMatchObject({ code: "guard_lease_required" }); + }); + expect(write).not.toHaveBeenCalled(); + expect(setup.tmuxAdapter.createSession).not.toHaveBeenCalled(); + if (protectedSeat) await expect(guard.operation(node.id, write)).rejects.toMatchObject({ code: "typing_guard_enabled" }); + else await guard.operation(node.id, write); + expect(write).toHaveBeenCalledTimes(protectedSeat ? 0 : 1); + }); + + it.each(["operation", "humanInput"] as const)("scopes nested custody cleanup inside %s", async origin => { + const { node, claim } = await claimedSeat(); + const guard = setup.tmuxAdapter.deliveryGuard!; + const write = vi.fn(async () => {}); + await guard[origin](node.id, async () => { + await guard.custody([node.id], async () => { + expect(guard.ownsCustody(node.id)).toBe(true); + await expect(guard.input(node.id, write)).rejects.toMatchObject({ code: "guard_lease_required" }); + expect((await setup.rigLifecycleService.unclaimSession(claim.sessionId)).ok).toBe(true); + }); + }); + expect(write).not.toHaveBeenCalled(); + expect(setup.sessionRegistry.getBindingForNode(node.id)).toBeNull(); + }); + + it("serializes human-scoped cleanup with an existing automatic operation", async () => { + const { node, claim } = await claimedSeat(); + const guard = setup.tmuxAdapter.deliveryGuard!; + const binding = setup.sessionRegistry.getBindingForNode(node.id); + const entered = barrier(), finish = barrier(); + const busy = guard.operation(node.id, async () => { entered.resolve(); await finish.promise; }); + await entered.promise; + let completed = false; + const cleanup = guard.humanInput(node.id, () => setup.rigLifecycleService.unclaimSession(claim.sessionId)) + .then(result => { completed = true; return result; }); + await new Promise(resolve => setImmediate(resolve)); + const completedWhileBusy = completed; + const bindingWhileBusy = setup.sessionRegistry.getBindingForNode(node.id); + finish.resolve(); + await busy; + expect((await cleanup).ok).toBe(true); + expect(completedWhileBusy).toBe(false); + expect(bindingWhileBusy).toEqual(binding); + expect(setup.sessionRegistry.getBindingForNode(node.id)).toBeNull(); + }); + + it("joins the serial lock when a human scope is nested inside its owner", async () => { + const { node, claim } = await claimedSeat(); + const guard = setup.tmuxAdapter.deliveryGuard!; + const result = await guard.operation(node.id, () => guard.humanInput(node.id, + () => setup.rigLifecycleService.unclaimSession(claim.sessionId))); + expect(result.ok).toBe(true); + expect(setup.sessionRegistry.getBindingForNode(node.id)).toBeNull(); + }); + + it("does not retain a finished parent's serial ownership in detached human cleanup", async () => { + const { node, claim } = await claimedSeat(); + const guard = setup.tmuxAdapter.deliveryGuard!; + const humanEntered = barrier(), resumeHuman = barrier(), busyEntered = barrier(), finishBusy = barrier(); + let cleanup!: Promise; + let completed = false; + await guard.operation(node.id, async () => { + cleanup = guard.humanInput(node.id, async () => { + humanEntered.resolve(); await resumeHuman.promise; + return setup.rigLifecycleService.unclaimSession(claim.sessionId); + }).then(result => { completed = true; return result; }); + await humanEntered.promise; + }); + const busy = guard.operation(node.id, async () => { busyEntered.resolve(); await finishBusy.promise; }); + await busyEntered.promise; + const binding = setup.sessionRegistry.getBindingForNode(node.id); + resumeHuman.resolve(); + await new Promise(resolve => setImmediate(resolve)); + const completedWhileBusy = completed; + const bindingWhileBusy = setup.sessionRegistry.getBindingForNode(node.id); + finishBusy.resolve(); + await busy; + expect(await cleanup).toMatchObject({ ok: true }); + expect(completedWhileBusy).toBe(false); + expect(bindingWhileBusy).toEqual(binding); + }); + + it.each([[false, false], [false, true], [true, false], [true, true]])("does not wait on an unaffected seat unless deleting (delete=%s, previouslyClaimed=%s)", async (deleting, previouslyClaimed) => { + const { rig } = await claimedSeat(); + const unaffected = setup.rigRepo.addNode(rig.id, "external.empty", { runtime: "terminal" }); + let detachedSessionId: string | undefined; + if (previouslyClaimed) { + const discovered = setup.discoveryRepo.upsertDiscoveredSession({ + tmuxSession: "manual-empty", tmuxPane: "%2", cwd: "/tmp", activeCommand: "bash", runtimeHint: "terminal", confidence: "high", + }); + const claim = await setup.claimService.bind({ discoveredId: discovered.id, rigId: rig.id, logicalId: unaffected.logicalId }); + if (!claim.ok) throw new Error(claim.error); + detachedSessionId = claim.sessionId; + await setup.rigLifecycleService.unclaimSession(claim.sessionId); + } + const entered = barrier(), finish = barrier(); + const busy = setup.tmuxAdapter.deliveryGuard!.lifecycle([unaffected.id], async () => { + entered.resolve(); await finish.promise; + }); + await entered.promise; + const notify = vi.spyOn(setup.eventBus, "notifySubscribers"); + let completed = false; + const request = setup.app.request(`/api/rigs/${rig.id}/release`, { + method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify({ delete: deleting }), + }).then(response => { completed = true; return response; }); + await new Promise(resolve => setImmediate(resolve)); + const completedWhileBusy = completed; + finish.resolve(); + await busy; + const response = await request; + expect(response.status).toBe(200); + expect(completedWhileBusy).toBe(!deleting); + if (previouslyClaimed) { + expect(await response.json()).toMatchObject({ released: expect.arrayContaining([ + expect.objectContaining({ nodeId: unaffected.id, sessionId: detachedSessionId, sessionName: "manual-empty" }), + ]) }); + expect(notify).toHaveBeenCalledWith(expect.objectContaining({ type: "session.detached", nodeId: unaffected.id, sessionName: "manual-empty" })); + } + }); + + it.each([[false, "guard", false], [true, "guard", false], [false, "session", false], [true, "session", false], [false, "guard", true], [true, "guard", true]] as const)("captures every release target before waiting (delete=%s, check=%s, detached=%s)", async (deleting, check, detached) => { + const first = await claimedSeat(); + const node = setup.rigRepo.addNode(first.rig.id, "external.second", { runtime: "terminal" }); + const discovered = setup.discoveryRepo.upsertDiscoveredSession({ + tmuxSession: "manual-second", tmuxPane: "%2", cwd: "/tmp", activeCommand: "bash", runtimeHint: "terminal", confidence: "high", + }); + const claim = await setup.claimService.bind({ discoveredId: discovered.id, rigId: first.rig.id, logicalId: node.logicalId }); + if (!claim.ok) throw new Error(claim.error); + const guard = setup.tmuxAdapter.deliveryGuard!; + if (check === "session") { + // Isolate the service's session-ID check from the guard's occupant check. + const target = guard.target.bind(guard); + vi.spyOn(guard, "target").mockImplementation(name => ({ ...target(name), occupant: null })); + } + const [blocked, changed] = [first, { rig: first.rig, node, discovered, claim }].sort((a, b) => a.node.id.localeCompare(b.node.id)); + if (detached) expect((await setup.rigLifecycleService.unclaimSession(changed!.claim.sessionId)).ok).toBe(true); + const entered = barrier(), finish = barrier(); + const busy = setup.tmuxAdapter.deliveryGuard!.lifecycle([blocked!.node.id], async () => { + entered.resolve(); await finish.promise; + }); + await entered.promise; + const request = setup.app.request(`/api/rigs/${first.rig.id}/release`, { + method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify({ delete: deleting }), + }); + await new Promise(resolve => setImmediate(resolve)); + expect((await setup.rigLifecycleService.unclaimSession(changed!.claim.sessionId)).ok).toBe(true); + expect((await setup.claimService.bind({ discoveredId: changed!.discovered.id, rigId: first.rig.id, logicalId: changed!.node.logicalId })).ok).toBe(true); + const bindings = db.prepare("SELECT * FROM bindings ORDER BY id").all(); + const discovery = setup.discoveryRepo.getDiscoveredSession(changed!.discovered.id); + vi.mocked(setup.tmuxAdapter.setSessionOption).mockClear(); + finish.resolve(); + await busy; + const response = await request; + expect(response.status).toBe(deleting ? 409 : 207); + if (deleting) { + expect(db.prepare("SELECT * FROM bindings ORDER BY id").all()).toEqual(bindings); + expect(setup.tmuxAdapter.setSessionOption).not.toHaveBeenCalled(); + } else { + expect(await response.json()).toMatchObject({ status: "partial", deleted: false, + released: [expect.objectContaining({ nodeId: blocked!.node.id, sessionId: blocked!.claim.sessionId })], + failed: [expect.objectContaining({ nodeId: changed!.node.id, sessionId: changed!.claim.sessionId })], + }); + expect(setup.sessionRegistry.getBindingForNode(blocked!.node.id)).toBeNull(); + expect(setup.sessionRegistry.getBindingForNode(changed!.node.id)?.tmuxSession).toBe(changed!.discovered.tmuxSession); + expect(vi.mocked(setup.tmuxAdapter.setSessionOption).mock.calls.every(([name]) => name === blocked!.discovered.tmuxSession)).toBe(true); + } + expect(setup.discoveryRepo.getDiscoveredSession(changed!.discovered.id)).toEqual(discovery); + expect(setup.rigRepo.getRig(first.rig.id)).not.toBeNull(); + }); + + it("keeps a repeat of an already released current ID harmless", async () => { + const { node, claim } = await claimedSeat(); + expect((await setup.rigLifecycleService.unclaimSession(claim.sessionId)).ok).toBe(true); + vi.mocked(setup.tmuxAdapter.setSessionOption).mockClear(); + const response = await setup.app.request(`/api/sessions/${claim.sessionId}/unclaim`, { method: "POST" }); + expect(response.status).toBe(200); + expect(setup.sessionRegistry.getBindingForNode(node.id)).toBeNull(); + expect(setup.tmuxAdapter.setSessionOption).not.toHaveBeenCalled(); + }); + + it("reports a later custody refusal after preserving an earlier release effect", async () => { + const { rig, node } = await claimedSeat(); + const later = setup.rigRepo.addNode(rig.id, "external.second", { runtime: "terminal" }); + const discovered = setup.discoveryRepo.upsertDiscoveredSession({ + tmuxSession: "manual-second", tmuxPane: "%2", cwd: "/tmp", activeCommand: "bash", runtimeHint: "terminal", confidence: "high", + }); + const claim = await setup.claimService.bind({ discoveredId: discovered.id, rigId: rig.id, logicalId: later.logicalId }); + if (!claim.ok) throw new Error(claim.error); + vi.mocked(setup.tmuxAdapter.setSessionOption).mockImplementation(async (name, key) => { + if (name === "manual-adoption" && key === "@rigged_logical_id") { + // A transport identity change after the full captured-set check. + setup.sessionRegistry.updateBinding(later.id, { tmuxPane: "%99" }); + } + return { ok: true }; + }); + const response = await setup.app.request(`/api/rigs/${rig.id}/release`, { method: "POST" }); + expect(response.status).toBe(207); + expect(await response.json()).toMatchObject({ status: "partial", deleted: false, + released: [expect.objectContaining({ nodeId: node.id })], + failed: [expect.objectContaining({ nodeId: later.id, sessionId: claim.sessionId })], + }); + expect(setup.sessionRegistry.getBindingForNode(node.id)).toBeNull(); + expect(setup.sessionRegistry.getBindingForNode(later.id)?.tmuxPane).toBe("%99"); + expect(setup.discoveryRepo.getDiscoveredSession(discovered.id)?.status).toBe("claimed"); + }); + + it("does not delete a seat added and claimed during cleanup", async () => { + const { rig, node } = await claimedSeat(); + const entered = barrier(), finish = barrier(); + vi.mocked(setup.tmuxAdapter.setSessionOption).mockImplementationOnce(async () => { + entered.resolve(); await finish.promise; return { ok: true }; + }); + const request = setup.app.request(`/api/rigs/${rig.id}/release`, { + method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify({ delete: true }), + }); + await entered.promise; + const added = setup.rigRepo.addNode(rig.id, "external.new", { runtime: "terminal" }); + const discovered = setup.discoveryRepo.upsertDiscoveredSession({ + tmuxSession: "manual-new", tmuxPane: "%2", cwd: "/tmp", activeCommand: "bash", runtimeHint: "terminal", confidence: "high", + }); + expect((await setup.claimService.bind({ discoveredId: discovered.id, rigId: rig.id, logicalId: added.logicalId })).ok).toBe(true); + const binding = setup.sessionRegistry.getBindingForNode(added.id); + const discovery = setup.discoveryRepo.getDiscoveredSession(discovered.id); + finish.resolve(); + const response = await request; + expect(response.status).toBe(207); + expect(await response.json()).toMatchObject({ status: "partial", deleted: false, + released: [expect.objectContaining({ nodeId: node.id })], + failed: [expect.objectContaining({ nodeId: added.id, sessionId: null })], + }); + expect(setup.rigRepo.getRig(rig.id)).not.toBeNull(); + expect(setup.sessionRegistry.getBindingForNode(node.id)).toBeNull(); + expect(setup.sessionRegistry.getBindingForNode(added.id)).toEqual(binding); + expect(setup.discoveryRepo.getDiscoveredSession(discovered.id)).toEqual(discovery); + expect(db.prepare("SELECT count(*) AS n FROM events WHERE node_id = ? AND type = 'session.detached'").get(added.id)).toEqual({ n: 0 }); + }); + + it("holds earlier seats through a whole-rig release and deletion", async () => { + const { rig, node } = await claimedSeat(); + const secondNode = setup.rigRepo.addNode(rig.id, "external.second", { runtime: "terminal" }); + const discovered = setup.discoveryRepo.upsertDiscoveredSession({ + tmuxSession: "manual-second", tmuxPane: "%2", cwd: "/tmp", + activeCommand: "bash", runtimeHint: "terminal", confidence: "high", + }); + expect((await setup.claimService.bind({ discoveredId: discovered.id, rigId: rig.id, logicalId: secondNode.logicalId })).ok).toBe(true); + const entered = barrier(); + const finish = barrier(); + vi.mocked(setup.tmuxAdapter.setSessionOption).mockImplementation(async (sessionName, key, value) => { + if (sessionName === "manual-second" && key === "@rigged_node_id" && value === "") { + entered.resolve(); + await finish.promise; + } + return { ok: true }; + }); + const release = setup.app.request(`/api/rigs/${rig.id}/release`, { + method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify({ delete: true }), + }); + await entered.promise; + let launchEntered = false; + const launch = setup.tmuxAdapter.deliveryGuard!.lifecycle([node.id], async () => { + launchEntered = true; + return setup.nodeLauncher.launchNode(rig.id, node.logicalId); + }).then(result => ({ result, error: null }), error => ({ result: null, error })); + await new Promise(resolve => setImmediate(resolve)); + const launchedDuringRelease = launchEntered; + finish.resolve(); + const response = await release; + const outcome = await launch; + expect(launchedDuringRelease).toBe(false); + expect(launchEntered).toBe(false); + expect(outcome.error).toMatchObject({ code: "guard_target_unknown" }); + expect(response.status).toBe(200); + expect(setup.rigRepo.getRig(rig.id)).toBeNull(); + expect(setup.tmuxAdapter.createSession).not.toHaveBeenCalled(); + }); + + it("still releases the current claim without killing its tmux session", async () => { + const { node, discovered, claim } = await claimedSeat(); + const res = await setup.app.request(`/api/sessions/${claim.sessionId}/unclaim`, { method: "POST" }); + expect(res.status).toBe(200); + expect(setup.sessionRegistry.getBindingForNode(node.id)).toBeNull(); + expect(setup.discoveryRepo.getDiscoveredSession(discovered.id)?.status).toBe("active"); + expect(setup.tmuxAdapter.killSession).not.toHaveBeenCalled(); + }); +});