mirror of
https://github.com/mvschwarz/openrig.git
synced 2026-11-07 05:56:21 -10:00
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.
This commit is contained in:
1 parent
465ac18af6
commit
877bc53669
5 files changed
+878
-81
No files matched your search
@@ -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<UnclaimSessionResult> {
|
||||
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<ReleaseRigResult> {
|
||||
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<string, string>();
|
||||
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<ReleaseRigResult> => {
|
||||
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(
|
||||
|
||||
@@ -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<T>(nodeIds: string[], fn: () => Promise<T>, onTargetChanged?: (nodeId: string, error: DeliveryGuardError) => void): Promise<T> {
|
||||
const targets = [...new Set(nodeIds)].sort().map(id => this.target(id));
|
||||
const excluded = new Set<string>();
|
||||
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<T> => {
|
||||
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<T>(nodeIds: string[], fn: () => Promise<T>): Promise<T> {
|
||||
@@ -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<T>(name: string, fn: () => Promise<T>): Promise<T> {
|
||||
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<T>(name: string, fn: () => Promise<T>): Promise<T> {
|
||||
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); }
|
||||
|
||||
@@ -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);
|
||||
}
|
||||
|
||||
|
||||
@@ -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<typeof createTestApp>;
|
||||
let run: (...args: string[]) => Promise<string>;
|
||||
|
||||
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<number> {
|
||||
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<void>(resolve => { entered = resolve; });
|
||||
const resume = new Promise<void>(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<void>(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<string, string> = {
|
||||
"@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();
|
||||
});
|
||||
});
|
||||
@@ -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<typeof createTestApp>;
|
||||
|
||||
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<void>(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<void>(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<void>(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<void>(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<unknown>;
|
||||
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<void>(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<void>(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<void>(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<void>(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();
|
||||
});
|
||||
});
|
||||
Reference in new issue
Block a user