mirror of
https://github.com/mvschwarz/openrig.git
synced 2026-11-13 18:49:34 -10:00
Compare commits
5
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
0107d6e788 | ||
|
|
5a52230e95 | ||
|
|
50f58dfaca | ||
|
|
f6ba977092 | ||
|
|
12ca636227 |
No files matched your search
@@ -14,6 +14,7 @@ import type { Command } from "commander";
|
||||
import { InvalidArgumentError } from "commander";
|
||||
import { DaemonConnectionError, DaemonResponseError, DaemonTimeoutError } from "./client.js";
|
||||
import { blockedConnectionGuidance, isBlockedConnectionCode } from "./daemon-reachability.js";
|
||||
import { resetSenderSession } from "./sender-identity.js";
|
||||
|
||||
/**
|
||||
* Commander option parser — a positive integer (>= 1). Rejects negative, zero,
|
||||
@@ -175,9 +176,14 @@ export async function runProgram(program: Command, argv: string[], io: RunProgra
|
||||
};
|
||||
suppressErr(program);
|
||||
|
||||
// Each run starts clean: an earlier in-process run's failure isn't this run's status, and its sender isn't this run's.
|
||||
process.exitCode = undefined;
|
||||
resetSenderSession();
|
||||
try {
|
||||
await program.parseAsync(argv);
|
||||
return 0;
|
||||
// An action's own failure sets process.exitCode. It's returned, not passed to process.exit(), which can drop output
|
||||
// still being written to a pipe; the process ends with process.exitCode.
|
||||
return typeof process.exitCode === "number" ? process.exitCode : 0;
|
||||
} catch (e) {
|
||||
if (isCleanCommanderExit(e)) return 0;
|
||||
// Response-integrity: render daemon transport failures honestly (3-part, both
|
||||
|
||||
@@ -6,6 +6,7 @@ import { ConfigStore } from "./config-store.js";
|
||||
import { getOpenRigHome, readOpenRigEnv, readOpenRigUrl } from "./openrig-compat.js";
|
||||
import { fetchWithTimeout, FetchTimeoutError } from "./fetch-with-timeout.js";
|
||||
import { FETCH_HEADERS_TIMEOUT_MS, nodeHttpFetch } from "./long-request-fetch.js";
|
||||
import { resolveSenderSession, type BindingCheck, type TmuxIdentityReader } from "./sender-identity.js";
|
||||
|
||||
export function terminalAuthHeaders(): Record<string, string> {
|
||||
const token = resolveTerminalToken();
|
||||
@@ -35,8 +36,10 @@ export const OCCUPANT_GENERATION_HEADER = "X-OpenRig-Occupant-Generation";
|
||||
* chokepoint: `DaemonClient.fetch` stamps this on every request; callers never set it by hand. A client built for one
|
||||
* authenticated caller (`sessionName`) carries that caller's seat instead of the process env.
|
||||
*/
|
||||
export function senderIdentityHeaders(originSelfHostId?: string, sessionName?: string): Record<string, string> {
|
||||
const session = (sessionName ?? readOpenRigEnv("OPENRIG_SESSION_NAME", "RIGGED_SESSION_NAME"))?.trim();
|
||||
export function senderIdentityHeaders(
|
||||
originSelfHostId?: string, sessionName?: string, tmuxReader?: TmuxIdentityReader, isBound?: BindingCheck,
|
||||
): Record<string, string> {
|
||||
const session = (sessionName ?? resolveSenderSession(tmuxReader, isBound))?.trim();
|
||||
if (!session) return {};
|
||||
// Preserve qualified senders; append only a locally derived instance id.
|
||||
const value =
|
||||
|
||||
@@ -17,6 +17,7 @@ import { omittedReadField, readView } from "../read-view.js";
|
||||
import { buildWriteReceipt, compactOutputEnabled, type ReceiptContext } from "../queue-receipt.js";
|
||||
import { TRANSITIONS_SHOWN, compactShow, compactTransitions, showEmptyNotice, transitionsNotice } from "../queue-read-compact.js";
|
||||
import { codeResultOf, planItemArtifact, type FileRead, type SharedRoot } from "../queue-artifact.js";
|
||||
import { resolveSenderSession } from "../sender-identity.js";
|
||||
|
||||
/**
|
||||
* `rig queue` — coordination primitive L3/inbox/outbox commands (PL-004 Phase A).
|
||||
@@ -694,6 +695,167 @@ async function printQueueItemResult(
|
||||
await printWriteResult(client, json, body, status, receipt && remoteHost ? { ...receipt, remoteHost } : receipt, remoteHost !== undefined);
|
||||
}
|
||||
|
||||
const UPDATE_READ_BACK_FIELDS = [
|
||||
"state", "closureReason", "closureTarget", "closureRequiredAt", "resolution", "blockedOn", "summary", "evidenceRef",
|
||||
] as const;
|
||||
|
||||
interface QueueTransitionReadBack {
|
||||
transitionId: number;
|
||||
transitionNote: string | null;
|
||||
actorSession: string;
|
||||
}
|
||||
|
||||
function queueUpdateFieldMatches(field: (typeof UPDATE_READ_BACK_FIELDS)[number], requested: unknown, actual: unknown): boolean {
|
||||
if (field === "closureRequiredAt" && typeof requested === "string" && typeof actual === "string") {
|
||||
const requestedTime = Date.parse(requested);
|
||||
const actualTime = Date.parse(actual);
|
||||
return Number.isFinite(requestedTime) && Number.isFinite(actualTime) && requestedTime === actualTime;
|
||||
}
|
||||
if (field === "resolution" && typeof requested === "string" && typeof actual === "string") {
|
||||
return actual === requested || actual.endsWith(`\n${requested}`);
|
||||
}
|
||||
return Object.is(actual, requested);
|
||||
}
|
||||
|
||||
/** Whether a recorded note is this update's note as the daemon records it (QueueRepository.update): as written, or with
|
||||
* only what the daemon adds for this request, the reopen prefix, the summary suffix, or a revised deadline's receipt
|
||||
* naming the new time and the accepting actor. Another writer's note that merely starts the same way isn't. */
|
||||
function noteRecordedAs(actual: string, note: string, requested: Record<string, unknown>, actor: string, deadline: unknown): boolean {
|
||||
const bases = [note];
|
||||
if (requested.reopen === true) bases.push(`reopen acknowledged: ${note}`);
|
||||
if (typeof requested.summary === "string" && requested.summary.trim() && !note.includes("(summary updated)")) {
|
||||
bases.push(`${note} (summary updated)`);
|
||||
}
|
||||
if (requested.closureRequiredAt === undefined) return bases.includes(actual);
|
||||
// The receipt's earlier time isn't known here: it's one line between the fixed parts.
|
||||
const tail = ` -> ${String(deadline)}; accepted by ${actor}`;
|
||||
return bases.some((base) => {
|
||||
const head = `${base}\nclosure deadline re-estimated: `;
|
||||
if (!actual.startsWith(head) || !actual.endsWith(tail) || actual.length <= head.length + tail.length) return false;
|
||||
return !actual.slice(head.length, actual.length - tail.length).includes("\n");
|
||||
});
|
||||
}
|
||||
|
||||
/** Who the daemon records as this write's actor: the sender header as sent, else the actor the body declares. Empty when
|
||||
* neither is known. A daemon drops a sender's host suffix only when it's that daemon's own id, so the bare form is
|
||||
* this write's only where the selected daemon answers with that id; nothing else (an absent --host included) shows it. */
|
||||
async function updateActors(client: Pick<DaemonClient, "senderIdentity" | "get">, bodyActor: unknown): Promise<string[]> {
|
||||
const { session } = await client.senderIdentity();
|
||||
if (session) {
|
||||
const parts = session.split("@");
|
||||
if (parts.length !== 3) return [session];
|
||||
let daemonId: unknown;
|
||||
try {
|
||||
daemonId = (await client.get<{ selfHostId?: unknown }>("/healthz", { timeoutMs: 1_000 })).data?.selfHostId;
|
||||
} catch { /* Unanswered: the suffix stays unproven, so only the full actor counts. */ }
|
||||
return daemonId === parts[2] ? [session, `${parts[0]}@${parts[1]}`] : [session];
|
||||
}
|
||||
return typeof bodyActor === "string" && bodyActor.trim() ? [bodyActor.trim()] : [];
|
||||
}
|
||||
|
||||
async function readQueueTransitionCursor(
|
||||
client: Pick<DaemonClient, "get">,
|
||||
qitemId: string,
|
||||
options: { headers?: Record<string, string>; timeoutMs?: number },
|
||||
): Promise<{ ok: true; latestTransitionId: number } | { ok: false; detail: string }> {
|
||||
let response: DaemonResponse<unknown>;
|
||||
try {
|
||||
response = await client.get<unknown>(`/api/queue/${encodeURIComponent(qitemId)}/transitions`, options);
|
||||
} catch (error) {
|
||||
return { ok: false, detail: `canonical transition preflight did not answer: ${(error as Error).message}` };
|
||||
}
|
||||
if (response.status !== 200 || !Array.isArray(response.data)) {
|
||||
return { ok: false, detail: `canonical transition preflight returned HTTP ${response.status} without a transition list` };
|
||||
}
|
||||
const transitionIds = response.data.flatMap((entry) => {
|
||||
if (!entry || typeof entry !== "object") return [];
|
||||
const transitionId = (entry as { transitionId?: unknown }).transitionId;
|
||||
return typeof transitionId === "number" && Number.isSafeInteger(transitionId) ? [transitionId] : [];
|
||||
});
|
||||
if (transitionIds.length !== response.data.length) {
|
||||
return { ok: false, detail: "canonical transition preflight returned a malformed transition list" };
|
||||
}
|
||||
return { ok: true, latestTransitionId: Math.max(0, ...transitionIds) };
|
||||
}
|
||||
|
||||
async function readBackQueueUpdate(
|
||||
client: Pick<DaemonClient, "get">,
|
||||
qitemId: string,
|
||||
requested: Record<string, unknown>,
|
||||
options: { headers?: Record<string, string>; timeoutMs?: number },
|
||||
transitionFloor?: number,
|
||||
/** Why the transition cursor wasn't taken before the write, when it wasn't. */
|
||||
floorUnknown?: string,
|
||||
/** Who the daemon records as this write's actor (updateActors). */
|
||||
actors: string[] = [],
|
||||
): Promise<{ ok: true; row: Record<string, unknown> } | { ok: false; detail: string }> {
|
||||
let response: DaemonResponse<Record<string, unknown>>;
|
||||
try {
|
||||
response = await client.get<Record<string, unknown>>(`/api/queue/${encodeURIComponent(qitemId)}`, options);
|
||||
} catch (error) {
|
||||
return { ok: false, detail: `canonical read-back did not answer: ${(error as Error).message}` };
|
||||
}
|
||||
if (response.status !== 200 || !response.data || response.data.qitemId !== qitemId) {
|
||||
return { ok: false, detail: `canonical read-back returned HTTP ${response.status} without row ${qitemId}` };
|
||||
}
|
||||
const mismatches = UPDATE_READ_BACK_FIELDS.flatMap((field) => {
|
||||
if (requested[field] === undefined) return [];
|
||||
return queueUpdateFieldMatches(field, requested[field], response.data[field])
|
||||
? []
|
||||
: [`${field}: requested ${JSON.stringify(requested[field])}, read ${JSON.stringify(response.data[field])}`];
|
||||
});
|
||||
if (mismatches.length > 0) {
|
||||
return { ok: false, detail: `canonical read-back mismatch (${mismatches.join("; ")})` };
|
||||
}
|
||||
|
||||
const requestedNote = typeof requested.transitionNote === "string" && requested.transitionNote.trim().length > 0
|
||||
? requested.transitionNote
|
||||
: null;
|
||||
if (requestedNote !== null) {
|
||||
// Without the cursor, a transition carrying the note can't be shown to be this write's: never accept an older one.
|
||||
if (transitionFloor === undefined) {
|
||||
return { ok: false, detail: `the requested note can't be shown to be new, because the transition read before the write failed (${floorUnknown ?? "no cursor"})` };
|
||||
}
|
||||
if (actors.length === 0) {
|
||||
return { ok: false, detail: "the requested note can't be tied to this write, because the actor it was sent as isn't known" };
|
||||
}
|
||||
let transitions: DaemonResponse<unknown>;
|
||||
try {
|
||||
transitions = await client.get<unknown>(`/api/queue/${encodeURIComponent(qitemId)}/transitions`, options);
|
||||
} catch (error) {
|
||||
return { ok: false, detail: `canonical transition read-back did not answer: ${(error as Error).message}` };
|
||||
}
|
||||
if (transitions.status !== 200 || !Array.isArray(transitions.data)) {
|
||||
return { ok: false, detail: `canonical transition read-back returned HTTP ${transitions.status} without a transition list` };
|
||||
}
|
||||
const matching = transitions.data.some((entry) => {
|
||||
if (!entry || typeof entry !== "object") return false;
|
||||
const transition = entry as Partial<QueueTransitionReadBack>;
|
||||
return typeof transition.transitionId === "number"
|
||||
&& transition.transitionId > transitionFloor
|
||||
&& typeof transition.transitionNote === "string"
|
||||
&& typeof transition.actorSession === "string"
|
||||
&& actors.includes(transition.actorSession)
|
||||
&& noteRecordedAs(transition.transitionNote, requestedNote, requested, transition.actorSession, response.data.closureRequiredAt);
|
||||
});
|
||||
if (!matching) {
|
||||
return {
|
||||
ok: false,
|
||||
detail: `no newer canonical transition carried requested note ${JSON.stringify(requestedNote)} from ${actors.map((actor) => JSON.stringify(actor)).join(" or ")}`,
|
||||
};
|
||||
}
|
||||
}
|
||||
return { ok: true, row: response.data };
|
||||
}
|
||||
|
||||
function emitQueueUpdateReadBackFailure(qitemId: string, detail: string, json: boolean): void {
|
||||
const message = `Queue update response was not accepted as success: ${detail}. The write may have committed, but its requested state was not verified.`;
|
||||
const action = `Run rig queue show ${shellQuote(qitemId)} --full --json against the same daemon before retrying.`;
|
||||
if (json) console.error(JSON.stringify({ error: "queue_update_readback_failed", message, action }));
|
||||
else console.error(`${message}\n${action}`);
|
||||
process.exitCode = 1;
|
||||
}
|
||||
|
||||
// OPR.0.4.3.03 — `rig queue show` body preview.
|
||||
//
|
||||
// Default `show` renders a BOUNDED body preview instead of dumping the whole
|
||||
@@ -874,21 +1036,21 @@ function emitBodyResolveError(err: Error & { fact?: string; consequence?: string
|
||||
}
|
||||
|
||||
function resolveCurrentSession(explicit: string | undefined, optionName: string): string | undefined {
|
||||
const session = explicit ?? readOpenRigEnv("OPENRIG_SESSION_NAME", "RIGGED_SESSION_NAME");
|
||||
const session = explicit ?? resolveSenderSession();
|
||||
if (session) return session;
|
||||
|
||||
console.error(`--${optionName} is required when OPENRIG_SESSION_NAME is not set`);
|
||||
console.error(`--${optionName} is required when OPENRIG_SESSION_NAME is not set and reconciled tmux pane metadata does not resolve a session`);
|
||||
process.exitCode = 1;
|
||||
return undefined;
|
||||
}
|
||||
|
||||
/** For a verb whose actor is the seat's own identity (the X-OpenRig-Session header the CLI sets from OPENRIG_SESSION_NAME),
|
||||
/** For a verb whose actor is the seat's own identity (the X-OpenRig-Session header the CLI sets from managed identity),
|
||||
* where the named flag is deprecated and ignored: the error names what actually sets the actor, never the flag. */
|
||||
function requireSeatSession(ignoredFlag: string): string | undefined {
|
||||
const session = readOpenRigEnv("OPENRIG_SESSION_NAME", "RIGGED_SESSION_NAME");
|
||||
const session = resolveSenderSession();
|
||||
if (session) return session;
|
||||
|
||||
console.error(`OPENRIG_SESSION_NAME is not set, so there's no seat to act as. Run this from a managed seat, or set ` +
|
||||
console.error(`OPENRIG_SESSION_NAME is not set, so there's no seat to act as, and reconciled tmux pane metadata does not resolve one. Run this from a managed seat, or set ` +
|
||||
`OPENRIG_SESSION_NAME to the seat's session name (for example dev-impl@my-rig). --${ignoredFlag} is ignored: the ` +
|
||||
"actor is the seat's own identity.");
|
||||
process.exitCode = 1;
|
||||
@@ -912,8 +1074,8 @@ const UPDATE_STATE_WRITE_FLAGS = [
|
||||
*/
|
||||
function resolveUpdateActor(source: string | undefined, opts: Record<string, unknown>): { actorSession?: string } | false {
|
||||
const refuse = (message: string): false => { console.error(message); process.exitCode = 1; return false; };
|
||||
// Trimmed like the DaemonClient's header (senderIdentityHeaders): a blank env is no seat, so the two never disagree.
|
||||
const seat = readOpenRigEnv("OPENRIG_SESSION_NAME", "RIGGED_SESSION_NAME")?.trim() || undefined;
|
||||
// The same resolver the DaemonClient header uses, so the preflight and transport can never disagree.
|
||||
const seat = resolveSenderSession();
|
||||
if (source !== undefined) {
|
||||
const trimmed = source.trim();
|
||||
if (parseSessionName(trimmed).kind !== "external") {
|
||||
@@ -933,10 +1095,10 @@ function resolveUpdateActor(source: string | undefined, opts: Record<string, unk
|
||||
if (seat) return {};
|
||||
const noteOnly = opts.note !== undefined && UPDATE_STATE_WRITE_FLAGS.every(([key]) => opts[key] === undefined);
|
||||
return refuse(noteOnly
|
||||
? "notes need a seat; an external routine can pass --source <name>@external. OPENRIG_SESSION_NAME is not set, so " +
|
||||
"there's no seat to act as: run this from a managed seat, or add --source <name>@external to record the note as " +
|
||||
? "notes need a seat; an external routine can pass --source <name>@external. OPENRIG_SESSION_NAME is not set and reconciled " +
|
||||
"tmux pane metadata does not resolve a seat: run this from a managed seat, or add --source <name>@external to record the note as " +
|
||||
"that external routine (claimed, never a seat). --actor is ignored: the actor is the seat's own identity."
|
||||
: "OPENRIG_SESSION_NAME is not set, so there's no seat to act as. Run this from a managed seat, or set " +
|
||||
: "OPENRIG_SESSION_NAME is not set, so there's no seat to act as, and reconciled tmux pane metadata does not resolve one. Run this from a managed seat, or set " +
|
||||
"OPENRIG_SESSION_NAME to the seat's session name (for example dev-impl@my-rig). --actor is ignored: the actor is " +
|
||||
"the seat's own identity. Changing a row needs a seat, and notes need a seat; an external routine can pass " +
|
||||
"--source <name>@external for a note only. A person answering an ask addressed to them (a host pairing approval, " +
|
||||
@@ -1166,7 +1328,7 @@ export function queueCommand(depsOverride?: QueueDeps): Command {
|
||||
}
|
||||
// A managed seat keeps transport-derived identity; an external caller can
|
||||
// name the existing claimed:v1 body actor without forging a transport header.
|
||||
const managedSource = readOpenRigEnv("OPENRIG_SESSION_NAME", "RIGGED_SESSION_NAME");
|
||||
const managedSource = resolveSenderSession();
|
||||
const source = resolveCurrentSession(managedSource ?? opts.source, "source");
|
||||
if (!source) return;
|
||||
const deps = getDeps();
|
||||
@@ -1468,7 +1630,7 @@ For a new cross-host transfer, prefer rig queue handoff --host <registered-host>
|
||||
process.exitCode = 1;
|
||||
return;
|
||||
}
|
||||
const res = await target.client.post<unknown>(`/api/queue/${encodeURIComponent(qitemId)}/update`, {
|
||||
const requested = {
|
||||
state: opts.state,
|
||||
reopen: opts.reopen,
|
||||
closureReason: opts.closureReason,
|
||||
@@ -1482,7 +1644,29 @@ For a new cross-host transfer, prefer rig queue handoff --host <registered-host>
|
||||
evidenceRef: opts.evidenceRef,
|
||||
transitionNote: opts.note,
|
||||
...actor,
|
||||
}, target.options);
|
||||
};
|
||||
let transitionFloor: number | undefined;
|
||||
let floorUnknown: string | undefined;
|
||||
if (typeof requested.transitionNote === "string" && requested.transitionNote.trim().length > 0) {
|
||||
// The cursor proves the note is new afterwards. Without it the write still goes; the receipt says what's unproven.
|
||||
const cursor = await readQueueTransitionCursor(target.client, qitemId, target.options);
|
||||
if (cursor.ok) transitionFloor = cursor.latestTransitionId;
|
||||
else floorUnknown = cursor.detail;
|
||||
}
|
||||
const res = await target.client.post<unknown>(`/api/queue/${encodeURIComponent(qitemId)}/update`, requested, target.options);
|
||||
if (res.status < 400) {
|
||||
const actors = await updateActors(target.client, requested.actorSession);
|
||||
const verified = await readBackQueueUpdate(target.client, qitemId, requested, target.options, transitionFloor, floorUnknown, actors);
|
||||
if (!verified.ok) {
|
||||
emitQueueUpdateReadBackFailure(qitemId, verified.detail, opts.json ?? false);
|
||||
return;
|
||||
}
|
||||
// The canonical row's fields win; the write's own advisories (a park with no wake) come only with its response.
|
||||
const advisories = (res.data as { advisories?: unknown } | null)?.advisories;
|
||||
const row = Array.isArray(advisories) ? { ...verified.row, advisories } : verified.row;
|
||||
await printQueueItemResult(target.client, qitemId, opts.json ?? false, row, res.status, writeReceipt(opts.full, { verb: "update" }), target.remote);
|
||||
return;
|
||||
}
|
||||
await printQueueItemResult(target.client, qitemId, opts.json ?? false, res.data, res.status, writeReceipt(opts.full, { verb: "update" }), target.remote);
|
||||
}, remoteHostOf(opts.host) !== undefined, remoteHostOf(opts.host));
|
||||
});
|
||||
|
||||
@@ -0,0 +1,22 @@
|
||||
import { readFileSync } from "node:fs";
|
||||
import { createRequire } from "node:module";
|
||||
import { join } from "node:path";
|
||||
import type Database from "better-sqlite3";
|
||||
import { ConfigStore } from "./config-store.js";
|
||||
import { getOpenRigHome } from "./openrig-compat.js";
|
||||
|
||||
/** This home's database, opened read-only without requiring the local daemon. better-sqlite3 loads only when this is
|
||||
* called, so importing this module costs a command nothing. Throws when there's no database. */
|
||||
export function openLocalDbReadonly(): Database.Database {
|
||||
const configured = new ConfigStore().resolveWithSource("db.path");
|
||||
let dbPath = configured.value as string;
|
||||
// An explicit configured DB wins. Otherwise retain the last launch's --db selection.
|
||||
if (configured.source === "default") {
|
||||
try {
|
||||
const state = JSON.parse(readFileSync(join(getOpenRigHome(), "daemon.json"), "utf8"));
|
||||
if (typeof state.db === "string" && state.db.length > 0) dbPath = state.db;
|
||||
} catch { /* No launch record: use the configured default, never search other homes. */ }
|
||||
}
|
||||
const Sqlite = createRequire(import.meta.url)("better-sqlite3") as typeof Database;
|
||||
return new Sqlite(dbPath, { readonly: true, fileMustExist: true, timeout: 100 });
|
||||
}
|
||||
@@ -1,24 +1,12 @@
|
||||
import Database from "better-sqlite3";
|
||||
import { readFileSync } from "node:fs";
|
||||
import { join } from "node:path";
|
||||
import { ConfigStore } from "./config-store.js";
|
||||
import { getOpenRigHome } from "./openrig-compat.js";
|
||||
import type Database from "better-sqlite3";
|
||||
import { validateHostRegistry } from "./host-registry.js";
|
||||
import { openLocalDbReadonly } from "./local-db.js";
|
||||
|
||||
/** Read the boot-minted identity without requiring the local daemon or minting a replacement. */
|
||||
export function readLocalOrigin(): string | undefined {
|
||||
let db: Database.Database | undefined;
|
||||
try {
|
||||
const configured = new ConfigStore().resolveWithSource("db.path");
|
||||
let dbPath = configured.value as string;
|
||||
// An explicit configured DB wins. Otherwise retain the last launch's --db selection.
|
||||
if (configured.source === "default") {
|
||||
try {
|
||||
const state = JSON.parse(readFileSync(join(getOpenRigHome(), "daemon.json"), "utf8"));
|
||||
if (typeof state.db === "string" && state.db.length > 0) dbPath = state.db;
|
||||
} catch { /* No launch record: use the configured default, never search other homes. */ }
|
||||
}
|
||||
db = new Database(dbPath, { readonly: true, fileMustExist: true, timeout: 100 });
|
||||
db = openLocalDbReadonly();
|
||||
const row = db.prepare("SELECT host_id FROM self_host_identity WHERE singleton = 1").get() as { host_id?: unknown } | undefined;
|
||||
const id = row?.host_id;
|
||||
if (typeof id !== "string" || id === "localhost") return undefined;
|
||||
|
||||
@@ -1,20 +1,90 @@
|
||||
import { readOpenRigEnv } from "./openrig-compat.js";
|
||||
import { execFileSync } from "node:child_process";
|
||||
import type Database from "better-sqlite3";
|
||||
import { openLocalDbReadonly } from "./local-db.js";
|
||||
|
||||
export type TmuxIdentityReader = (pane: string, option: string) => string | undefined;
|
||||
|
||||
/** A busy or wedged tmux server must not hold every queue request: past this, the pane's seat is unresolved. */
|
||||
export const TMUX_IDENTITY_TIMEOUT_MS = 1_000;
|
||||
|
||||
const readTmuxIdentity: TmuxIdentityReader = (pane, option) => {
|
||||
try {
|
||||
const value = execFileSync("tmux", ["show-option", "-v", "-t", pane, option], {
|
||||
encoding: "utf8",
|
||||
stdio: ["ignore", "pipe", "pipe"],
|
||||
timeout: TMUX_IDENTITY_TIMEOUT_MS,
|
||||
}).trim();
|
||||
return value || undefined;
|
||||
} catch {
|
||||
return undefined;
|
||||
}
|
||||
};
|
||||
|
||||
/** Whether the daemon still binds a node to a tmux session. Claim records the binding beside the session's metadata;
|
||||
* unclaim deletes it in its own transaction even when clearing the metadata fails. */
|
||||
export type BindingCheck = (nodeId: string, tmuxSession: string) => boolean;
|
||||
|
||||
export function bindingIsCurrent(db: Database.Database, nodeId: string, tmuxSession: string): boolean {
|
||||
return db.prepare("SELECT 1 FROM bindings WHERE node_id = ? AND tmux_session = ?").get(nodeId, tmuxSession) !== undefined;
|
||||
}
|
||||
|
||||
/** The binding as this home's database records it; no database, or no answer, isn't evidence of a binding. */
|
||||
const localBindingIsCurrent: BindingCheck = (nodeId, tmuxSession) => {
|
||||
let db: Database.Database | undefined;
|
||||
try {
|
||||
db = openLocalDbReadonly();
|
||||
return bindingIsCurrent(db, nodeId, tmuxSession);
|
||||
} catch {
|
||||
return false;
|
||||
} finally {
|
||||
db?.close();
|
||||
}
|
||||
};
|
||||
|
||||
// The pane's seat, read once per run (runProgram resets it), so a command's preflight and every request it sends share one
|
||||
// value and one tmux read. Keyed by what produced it, so a different pane or reader reads again.
|
||||
let paneSeat: { pane: string; tmuxReader: TmuxIdentityReader; isBound: BindingCheck; session: string | undefined } | undefined;
|
||||
|
||||
/** Forget the pane seat read for the previous run. */
|
||||
export function resetSenderSession(): void {
|
||||
paneSeat = undefined;
|
||||
}
|
||||
|
||||
/**
|
||||
* The seat identity a CLI dispatch is attributed to — the SAME value the
|
||||
* DaemonClient stamps as `X-OpenRig-Session` (the P18 chokepoint), from which
|
||||
* the daemon DERIVES the actor / rendered `From:`. `--from` is deprecated and
|
||||
* ignored: origin is the seat env, never a caller-supplied body claim. Returns
|
||||
* `undefined` when the seat is unresolvable — the caller then delivers-and-labels
|
||||
* with `SENDER_FALLBACK` (never a refusal, never a forged actor).
|
||||
* the daemon DERIVES the actor / rendered `From:`. The process seat env wins;
|
||||
* a reconciled managed pane's `@rigged_session_name` repairs a live process
|
||||
* that predates its env binding, but only while the daemon still binds that
|
||||
* session's node to it: metadata left behind on a released session isn't a
|
||||
* current seat. Caller-supplied body claims never become a seat. Returns
|
||||
* `undefined` when neither managed source resolves.
|
||||
*/
|
||||
export function resolveSenderSession(): string | undefined {
|
||||
return readOpenRigEnv("OPENRIG_SESSION_NAME", "RIGGED_SESSION_NAME");
|
||||
export function resolveSenderSession(
|
||||
tmuxReader: TmuxIdentityReader = readTmuxIdentity,
|
||||
isBound: BindingCheck = localBindingIsCurrent,
|
||||
): string | undefined {
|
||||
const env = readOpenRigEnv("OPENRIG_SESSION_NAME", "RIGGED_SESSION_NAME")?.trim();
|
||||
if (env) return env;
|
||||
const pane = process.env.TMUX_PANE?.trim();
|
||||
if (!pane) return undefined;
|
||||
if (paneSeat?.pane === pane && paneSeat.tmuxReader === tmuxReader && paneSeat.isBound === isBound) return paneSeat.session;
|
||||
const session = boundPaneSeat(pane, tmuxReader, isBound);
|
||||
paneSeat = { pane, tmuxReader, isBound, session };
|
||||
return session;
|
||||
}
|
||||
|
||||
function boundPaneSeat(pane: string, tmuxReader: TmuxIdentityReader, isBound: BindingCheck): string | undefined {
|
||||
const session = tmuxReader(pane, "@rigged_session_name")?.trim();
|
||||
if (!session) return undefined;
|
||||
const nodeId = tmuxReader(pane, "@rigged_node_id")?.trim();
|
||||
return nodeId && isBound(nodeId, session) ? session : undefined;
|
||||
}
|
||||
|
||||
/**
|
||||
* The session-less fall-open marker for an unattributable CLI dispatch. P18 (deletion atom) REVERSES A1's
|
||||
* seat-boundary refusal: an env-less `rig send`/`broadcast` now DELIVERS carrying this HONEST label rather
|
||||
* seat-boundary refusal: an unresolved `rig send`/`broadcast` now DELIVERS carrying this HONEST label rather
|
||||
* than refusing, because the daemon half delivers-and-labels the header-absent write (no downstream 401) —
|
||||
* unverified is labelled unknown, never laundered to verified.
|
||||
*
|
||||
|
||||
@@ -513,6 +513,17 @@ describe("PUBLIC bin ownership (guard finding 1 — the wrapper is the real fron
|
||||
|
||||
describe("script-safety integration (the compiled front door)", () => {
|
||||
const cliEntry = join(__dirname, "..", "dist", "index.js");
|
||||
it.skipIf(!existsSync(cliEntry))("an unresolved queue actor stays nonzero through a pipe with pipefail", () => {
|
||||
const result = spawnSync("/bin/bash", ["--noprofile", "--norc", "-o", "pipefail", "-c",
|
||||
'env -u OPENRIG_SESSION_NAME -u RIGGED_SESSION_NAME -u TMUX_PANE "$NODE" "$CLI" queue update qitem-pipe-proof --state in-progress --json 2>&1 | cat'], {
|
||||
timeout: 5000,
|
||||
encoding: "utf-8",
|
||||
env: { ...process.env, NODE: process.execPath, CLI: cliEntry },
|
||||
});
|
||||
expect(result.status).toBe(1);
|
||||
expect(result.stdout).toContain("OPENRIG_SESSION_NAME is not set");
|
||||
});
|
||||
|
||||
it.skipIf(!existsSync(cliEntry))("`echo x | rig` returns usage fast — no hang (exit code preserves today's clean-help baseline)", () => {
|
||||
const started = Date.now();
|
||||
const result = spawnSync(process.execPath, [cliEntry], {
|
||||
|
||||
@@ -92,12 +92,17 @@ describe("cross-host queue writes wait for the daemon's answer", () => {
|
||||
}
|
||||
|
||||
it("a create that gets no answer in time reads as unknown, never as a failure, with the ID to reconcile by", async () => {
|
||||
await expect(run(["create", "--destination", "dev@rig-b@vps-b", "--body", "b", "--summary", "s"],
|
||||
[new DaemonTimeoutError("Request to http://localhost:7433/api/queue/create timed out after 15000ms")])).rejects.toThrow();
|
||||
let thrown: unknown;
|
||||
await run(["create", "--destination", "dev@rig-b@vps-b", "--body", "b", "--summary", "s"],
|
||||
[new DaemonTimeoutError("Request to http://localhost:7433/api/queue/create timed out after 15000ms")]).catch((e: unknown) => { thrown = e; });
|
||||
const all = stderr.join("\n");
|
||||
expect(all).toContain("queue request got no answer in time (routing to host 'vps-b')");
|
||||
expect(all).not.toContain("queue transport failure");
|
||||
expect(all).toContain("The write outcome is INDETERMINATE");
|
||||
expect(all).toMatch(/repeat the same create with --id 'qitem-\d{14}-[0-9a-f]{16}'/);
|
||||
// The final timeout error carries the same-ID instruction itself, not only the request line printed before the write.
|
||||
const id = /Queue create request ID: (qitem-\d{14}-[0-9a-f]{16})/.exec(all)?.[1];
|
||||
expect(id).toBeDefined();
|
||||
expect(thrown).toBeInstanceOf(DaemonTimeoutError);
|
||||
expect((thrown as Error).message).toContain(`repeat the same create with --id '${id}' and unchanged source, destination, body and options`);
|
||||
});
|
||||
});
|
||||
@@ -52,6 +52,7 @@ describe("queue verbs on an existing row with --host (#2637)", () => {
|
||||
let local: Awaited<ReturnType<typeof serve>>;
|
||||
let remote: Awaited<ReturnType<typeof serve>>;
|
||||
let remoteAnswer: Answer;
|
||||
let remoteRow: Record<string, unknown>;
|
||||
let home: string;
|
||||
let out: string[];
|
||||
let err: string[];
|
||||
@@ -79,11 +80,22 @@ describe("queue verbs on an existing row with --host (#2637)", () => {
|
||||
? { status: 200, data: { healthy: true, selfHostId: "vm-a" } }
|
||||
: { status: 500, data: { error: "the selected daemon must not be asked for the row" } });
|
||||
localPort = Number(new URL(local.url).port);
|
||||
remoteRow = { ...ROW };
|
||||
const remoteTransitions: Array<{ transitionId: number; state: string; transitionNote: string | null }> = [
|
||||
{ transitionId: 1, state: "pending", transitionNote: null },
|
||||
];
|
||||
remoteAnswer = (req) => {
|
||||
if (req.path === "/healthz") return { status: 200, data: { healthy: true, selfHostId: "devbox-id" } };
|
||||
if (req.path.endsWith("/transitions")) return { status: 200, data: [{ transitionId: 1, state: "pending" }] };
|
||||
if (req.path.endsWith("/transitions")) return { status: 200, data: remoteTransitions };
|
||||
if (req.path.startsWith("/api/queue/list")) return { status: 200, data: [ROW] };
|
||||
return { status: 200, data: ROW };
|
||||
if (req.method === "POST" && req.path.endsWith("/update")) {
|
||||
remoteRow = { ...remoteRow, ...(req.body as Record<string, unknown>) };
|
||||
const transitionNote = (req.body as { transitionNote?: unknown }).transitionNote;
|
||||
if (typeof transitionNote === "string") {
|
||||
remoteTransitions.push({ transitionId: remoteTransitions.length + 1, state: String(remoteRow.state), transitionNote });
|
||||
}
|
||||
}
|
||||
return { status: 200, data: remoteRow };
|
||||
};
|
||||
remote = await serve((req) => remoteAnswer(req));
|
||||
hosts = [
|
||||
@@ -121,7 +133,7 @@ describe("queue verbs on an existing row with --host (#2637)", () => {
|
||||
it("transitions --host reads the remote log", async () => {
|
||||
await run("transitions", "qitem-remote-1", "--host", "devbox", "--json", "--full");
|
||||
expect(remoteQueueCalls().map((c) => c.path)).toEqual(["/api/queue/qitem-remote-1/transitions"]);
|
||||
expect(JSON.parse(out.join("\n"))).toEqual([{ transitionId: 1, state: "pending" }]);
|
||||
expect(JSON.parse(out.join("\n"))).toEqual([{ transitionId: 1, state: "pending", transitionNote: null }]);
|
||||
});
|
||||
|
||||
it("list --host lists the remote rows, its filters kept and no rig derived from this host's seat", async () => {
|
||||
@@ -143,7 +155,7 @@ describe("queue verbs on an existing row with --host (#2637)", () => {
|
||||
|
||||
it("update --host writes on the remote daemon, the actor stamped with this host's id", async () => {
|
||||
await run("update", "qitem-remote-1", "--host", "devbox", "--state", "done", "--closure-reason", "no-follow-on", "--note", "settled from vm-a", "--json");
|
||||
const [call] = remoteQueueCalls();
|
||||
const call = remoteQueueCalls().find((entry) => entry.method === "POST");
|
||||
expect(call).toMatchObject({ method: "POST", path: "/api/queue/qitem-remote-1/update" });
|
||||
expect(call!.headers["x-openrig-session"]).toBe("dev@fabrica@vm-a");
|
||||
expect(call!.headers.authorization).toBe("Bearer s3cret");
|
||||
@@ -208,7 +220,7 @@ describe("queue verbs on an existing row with --host (#2637)", () => {
|
||||
it("a self-id learned at the entry's current URL routes to that entry", async () => {
|
||||
bind({ observedUrl: remote.url });
|
||||
await run("update", "qitem-remote-1", "--host", "devbox-self-id", "--state", "done", "--json");
|
||||
expect(remoteQueueCalls().map((c) => c.path)).toEqual(["/api/queue/qitem-remote-1/update"]);
|
||||
expect(remoteQueueCalls().map((c) => c.path)).toEqual(["/api/queue/qitem-remote-1/update", "/api/queue/qitem-remote-1"]);
|
||||
expect(process.exitCode).toBeUndefined();
|
||||
});
|
||||
|
||||
@@ -263,7 +275,9 @@ describe("queue verbs on an existing row with --host (#2637)", () => {
|
||||
const { compactOutputEnabled } = await import("../src/queue-receipt.js");
|
||||
expect(compactOutputEnabled()).toBe(true);
|
||||
// The remote daemon's own id equals the alias: the receipt must still call the row remote.
|
||||
remoteAnswer = (req) => req.path === "/healthz" ? { status: 200, data: { selfHostId: "devbox" } } : { status: 200, data: ROW };
|
||||
remoteAnswer = (req) => req.path === "/healthz"
|
||||
? { status: 200, data: { selfHostId: "devbox" } }
|
||||
: { status: 200, data: { ...ROW, state: "in-progress" } };
|
||||
await run("update", "qitem-remote-1", "--host", "devbox", "--state", "in-progress", "--json");
|
||||
const receipt = JSON.parse(out.join("\n"));
|
||||
expect(receipt.host).toBe("devbox");
|
||||
@@ -274,13 +288,15 @@ describe("queue verbs on an existing row with --host (#2637)", () => {
|
||||
local.close();
|
||||
local = await serve((req) => req.path === "/healthz"
|
||||
? { status: 200, data: { selfHostId: "vm-a" } }
|
||||
: { status: 200, data: req.path.startsWith("/api/queue/list") ? [] : ROW });
|
||||
: { status: 200, data: req.path.startsWith("/api/queue/list") ? [] : { ...ROW, state: "done" } });
|
||||
localPort = Number(new URL(local.url).port);
|
||||
await run("show", "qitem-remote-1", "--json");
|
||||
await run("list", "--json");
|
||||
await run("update", "qitem-remote-1", "--state", "done", "--json");
|
||||
expect(remote.seen).toEqual([]);
|
||||
expect(localQueueCalls().map((c) => c.path.split("?")[0])).toEqual(["/api/queue/qitem-remote-1", "/api/queue/list", "/api/queue/qitem-remote-1/update"]);
|
||||
expect(localQueueCalls().map((c) => c.path.split("?")[0])).toEqual([
|
||||
"/api/queue/qitem-remote-1", "/api/queue/list", "/api/queue/qitem-remote-1/update", "/api/queue/qitem-remote-1",
|
||||
]);
|
||||
// The local list still derives the rig from the seat, and no sender is stamped with a host id.
|
||||
expect(new URL(localQueueCalls()[1]!.path, "http://x").searchParams.get("rig")).toBe("fabrica");
|
||||
expect(localQueueCalls()[2]!.headers["x-openrig-session"]).toBe("dev@fabrica");
|
||||
|
||||
@@ -68,8 +68,11 @@ for (const json of [false, true]) {
|
||||
expect(body.hint).toBe(`Queue item "${id}" was not found on daemon "selected-host".\n`
|
||||
+ `If it came from another host, run rig host list; replace <daemon-url> with its registered daemon URL and run: OPENRIG_URL='<daemon-url>' rig queue show ${shellQuote(id)} --full --json\n`
|
||||
+ `Or name that host: rig queue show ${shellQuote(id)} --host <host-id> --full --json`);
|
||||
expect(calls).toHaveLength(2);
|
||||
expect(calls[1]).toEqual({ path: "/healthz", options: { timeoutMs: 1_000 } });
|
||||
// A note's update first reads the row's transitions (read-only), so its read-back can prove the note is new.
|
||||
const preflight = verb === "update" ? 1 : 0;
|
||||
expect(calls).toHaveLength(2 + preflight);
|
||||
if (preflight) expect(calls[0]!.path).toBe(`/api/queue/${id}/transitions`);
|
||||
expect(calls[1 + preflight]).toEqual({ path: "/healthz", options: { timeoutMs: 1_000 } });
|
||||
expect(calls.some(c => c.path.includes("/hosts"))).toBe(false);
|
||||
});
|
||||
});
|
||||
@@ -106,6 +109,8 @@ it("quotes the supplied ID without guessing a host or suggesting a write replay"
|
||||
const body = JSON.parse(output.join("\n"));
|
||||
expect(body.hint).toContain(`rig queue show ${shellQuote(supplied)} --full --json`);
|
||||
expect(body.hint).not.toContain("queue update");
|
||||
expect(calls[0]?.path).toBe(`/api/queue/${encodeURIComponent(supplied)}/update`);
|
||||
expect(calls.slice(0, 2).map((c) => c.path)).toEqual([
|
||||
`/api/queue/${encodeURIComponent(supplied)}/transitions`, `/api/queue/${encodeURIComponent(supplied)}/update`,
|
||||
]);
|
||||
expect(process.exitCode).toBe(1);
|
||||
});
|
||||
@@ -42,6 +42,8 @@ describe("an agent's pane task: woken, claimed and closed (#165)", () => {
|
||||
return { status: response.status, data: await response.json() };
|
||||
};
|
||||
const client = {
|
||||
// The sender it stamps, as the real client reports it: the update's read-back checks the recorded actor against it.
|
||||
senderIdentity: async () => ({ session: process.env.OPENRIG_SESSION_NAME, originUnknown: false }),
|
||||
get: async (url: string) => request(url, { method: "GET" }),
|
||||
post: async (url: string, body: unknown) => request(url, { method: "POST", body: JSON.stringify(body) }),
|
||||
};
|
||||
|
||||
@@ -0,0 +1,300 @@
|
||||
import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { Command } from "commander";
|
||||
import type Database from "better-sqlite3";
|
||||
import { createDb } from "../../daemon/src/db/connection.js";
|
||||
import { migrate } from "../../daemon/src/db/migrate.js";
|
||||
import { ALL_MIGRATIONS } from "../../daemon/src/db/all-migrations.js";
|
||||
import { EventBus } from "../../daemon/src/domain/event-bus.js";
|
||||
import { OutboxHandler } from "../../daemon/src/domain/outbox-handler.js";
|
||||
import { QueueRepository } from "../../daemon/src/domain/queue-repository.js";
|
||||
import { runProgram } from "../src/cli-error.js";
|
||||
import { queueCommand } from "../src/commands/queue.js";
|
||||
import { STATE_FILE } from "../src/daemon-lifecycle.js";
|
||||
|
||||
type UpdateTransform = (body: Record<string, unknown>) => Record<string, unknown> | null;
|
||||
/** The nth read of the row's transitions (0 is the cursor read before the write): a response, or a throw for no answer. */
|
||||
type TransitionsRead = (call: number) => { status: number; data: unknown } | undefined;
|
||||
|
||||
let database: Database.Database;
|
||||
let repo: QueueRepository;
|
||||
|
||||
/** The selected daemon as the CLI sees it: the sender header it stamps, and the id the daemon's /healthz answers with. */
|
||||
interface Endpoint { sender?: string; selfHostId?: string }
|
||||
|
||||
async function runUpdate(qitemId: string, args: string[], transform: UpdateTransform = (body) => body, transitionsRead?: TransitionsRead,
|
||||
endpoint: Endpoint = {}) {
|
||||
let transitionReads = 0;
|
||||
let posts = 0;
|
||||
let exits = 0;
|
||||
const output: string[] = [];
|
||||
const errors: string[] = [];
|
||||
const program = new Command();
|
||||
program.addCommand(queueCommand({
|
||||
lifecycleDeps: {
|
||||
spawn: () => ({ pid: 1, unref() {} }),
|
||||
fetch: async () => ({ ok: true }),
|
||||
kill: () => true,
|
||||
readFile: (file: string) => file === STATE_FILE
|
||||
? JSON.stringify({ pid: 1, port: 7433, host: "127.0.0.1", db: "test", startedAt: "test" })
|
||||
: null,
|
||||
writeFile() {},
|
||||
removeFile() {},
|
||||
exists: (file: string) => file === STATE_FILE,
|
||||
mkdirp() {},
|
||||
openForAppend: () => 3,
|
||||
isProcessAlive: () => true,
|
||||
},
|
||||
clientFactory: () => ({
|
||||
senderIdentity: async () => ({ session: endpoint.sender ?? "reviewer@fixture", originUnknown: false }),
|
||||
get: async (route: string) => route === "/healthz"
|
||||
? { status: 200, data: { selfHostId: endpoint.selfHostId } }
|
||||
: route.endsWith("/transitions")
|
||||
? transitionsRead?.(transitionReads++) ?? { status: 200, data: repo.listTransitions(qitemId) }
|
||||
: { status: 200, data: repo.getById(qitemId) },
|
||||
post: async (_route: string, rawBody: unknown) => {
|
||||
posts += 1;
|
||||
const body = transform(structuredClone(rawBody as Record<string, unknown>));
|
||||
return {
|
||||
status: 200,
|
||||
data: body === null
|
||||
? repo.getById(qitemId)
|
||||
: repo.update({ ...body, qitemId, actorSession: "reviewer@fixture" }),
|
||||
};
|
||||
},
|
||||
}) as never,
|
||||
}));
|
||||
// No exit status is reset here: each run's status is runProgram's to start clean.
|
||||
const originalLog = console.log;
|
||||
const originalError = console.error;
|
||||
console.log = (...values: unknown[]) => output.push(values.join(" "));
|
||||
console.error = (...values: unknown[]) => errors.push(values.join(" "));
|
||||
try {
|
||||
const code = await runProgram(program, ["node", "rig", "queue", "update", qitemId, ...args, "--json"], {
|
||||
out: (value) => output.push(value),
|
||||
err: (value) => errors.push(value),
|
||||
exit: () => { exits += 1; },
|
||||
});
|
||||
return { code, output, errors, row: repo.getByIdOrThrow(qitemId), transitions: repo.listTransitions(qitemId), posts, exits };
|
||||
} finally {
|
||||
console.log = originalLog;
|
||||
console.error = originalError;
|
||||
}
|
||||
}
|
||||
|
||||
async function claimedRow(): Promise<string> {
|
||||
const row = await repo.create({
|
||||
sourceSession: "reviewer@fixture",
|
||||
destinationSession: "reviewer@fixture",
|
||||
body: "work",
|
||||
nudge: false,
|
||||
});
|
||||
repo.claim({ qitemId: row.qitemId, destinationSession: "reviewer@fixture" });
|
||||
return row.qitemId;
|
||||
}
|
||||
|
||||
describe("queue update canonical read-back against daemon domain semantics", () => {
|
||||
beforeAll(() => {
|
||||
database = createDb();
|
||||
migrate(database, ALL_MIGRATIONS);
|
||||
const bus = new EventBus(database);
|
||||
repo = new QueueRepository(database, bus, {
|
||||
validateRig: () => true,
|
||||
loadHumanRegistry: () => ({ ok: true, entities: [] }),
|
||||
});
|
||||
repo.attachOutbox(new OutboxHandler(database));
|
||||
});
|
||||
|
||||
beforeEach(() => {
|
||||
vi.stubEnv("OPENRIG_SESSION_NAME", "reviewer@fixture");
|
||||
process.exitCode = undefined;
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
vi.unstubAllEnvs();
|
||||
process.exitCode = undefined;
|
||||
});
|
||||
|
||||
afterAll(() => database.close());
|
||||
|
||||
it("accepts a closing resolution appended after an existing fallback resolution", async () => {
|
||||
const row = await repo.create({
|
||||
sourceSession: "reviewer@fixture",
|
||||
destinationSession: "worker@fixture",
|
||||
body: "work",
|
||||
nudge: false,
|
||||
});
|
||||
repo.routeToFallback(row.qitemId, "reviewer@fixture", "worker unavailable");
|
||||
repo.claim({ qitemId: row.qitemId, destinationSession: "reviewer@fixture" });
|
||||
|
||||
const result = await runUpdate(row.qitemId, [
|
||||
"--state", "done", "--closure-reason", "no-follow-on", "--resolution", "Handled by reviewer",
|
||||
]);
|
||||
|
||||
expect(result.code, [...result.errors, ...result.output].join("\n")).toBe(0);
|
||||
expect(result.row.resolution).toBe("fallback: worker unavailable\nHandled by reviewer");
|
||||
});
|
||||
|
||||
it("accepts close, reopen, and reclose while preserving each resolution line", async () => {
|
||||
const qitemId = await claimedRow();
|
||||
const first = await runUpdate(qitemId, [
|
||||
"--state", "done", "--closure-reason", "no-follow-on", "--resolution", "first outcome",
|
||||
]);
|
||||
const reopened = await runUpdate(qitemId, ["--state", "pending", "--reopen", "--note", "repair requested"]);
|
||||
const second = await runUpdate(qitemId, [
|
||||
"--state", "done", "--closure-reason", "no-follow-on", "--resolution", "second outcome",
|
||||
]);
|
||||
|
||||
expect([first.code, reopened.code, second.code], [first.errors, reopened.errors, second.errors].flat().join("\n"))
|
||||
.toEqual([0, 0, 0]);
|
||||
expect(second.row.resolution).toBe("first outcome\nsecond outcome");
|
||||
});
|
||||
|
||||
it.each([
|
||||
"2026-11-14T10:00:00Z",
|
||||
"2026-11-14T12:00:00+02:00",
|
||||
])("accepts equivalent deadline spelling %s after daemon normalization", async (deadline) => {
|
||||
const qitemId = await claimedRow();
|
||||
const result = await runUpdate(qitemId, [
|
||||
"--closure-required-at", deadline, "--note", "Accepted revised deadline",
|
||||
]);
|
||||
|
||||
expect(result.code, result.errors.join("\n")).toBe(0);
|
||||
expect(result.row.closureRequiredAt).toBe("2026-11-14T10:00:00.000Z");
|
||||
});
|
||||
|
||||
it("rejects a genuinely different stored deadline", async () => {
|
||||
const qitemId = await claimedRow();
|
||||
repo.update({
|
||||
qitemId,
|
||||
actorSession: "reviewer@fixture",
|
||||
closureRequiredAt: "2026-11-14T09:00:00Z",
|
||||
transitionNote: "initial deadline",
|
||||
});
|
||||
const result = await runUpdate(qitemId, [
|
||||
"--closure-required-at", "2026-11-14T10:00:00Z", "--note", "Accepted revised deadline",
|
||||
], () => null);
|
||||
|
||||
expect(result.code).toBe(1);
|
||||
expect(result.errors.join("\n")).toContain("closureRequiredAt");
|
||||
});
|
||||
|
||||
it("rejects a close whose requested resolution was dropped", async () => {
|
||||
const qitemId = await claimedRow();
|
||||
const result = await runUpdate(qitemId, [
|
||||
"--state", "done", "--closure-reason", "no-follow-on", "--resolution", "must persist",
|
||||
], (body) => {
|
||||
delete body.resolution;
|
||||
return body;
|
||||
});
|
||||
|
||||
expect(result.code).toBe(1);
|
||||
expect(result.row.state).toBe("done");
|
||||
expect(result.row.resolution).toBeNull();
|
||||
expect(result.errors.join("\n")).toContain("resolution");
|
||||
});
|
||||
|
||||
it("requires a newer canonical transition carrying the requested note", async () => {
|
||||
const qitemId = await claimedRow();
|
||||
const before = repo.listTransitions(qitemId).length;
|
||||
const result = await runUpdate(qitemId, ["--note", "must persist"], () => null);
|
||||
|
||||
expect(result.code).toBe(1);
|
||||
expect(result.transitions).toHaveLength(before);
|
||||
expect(result.errors.join("\n")).toContain("no newer canonical transition carried requested note");
|
||||
});
|
||||
|
||||
// The cursor read before the write only proves a note is new afterwards. When it fails, the write is still sent and
|
||||
// persists; the receipt says the note can't be shown to be new and exits non-zero, even with an older transition
|
||||
// already carrying the same note.
|
||||
it.each([
|
||||
["doesn't answer", () => { throw new Error("socket hang up (fixture)"); }],
|
||||
["answers HTTP 503", () => ({ status: 503, data: { error: "busy" } })],
|
||||
["returns a malformed list", () => ({ status: 200, data: [{ transitionId: "seven" }] })],
|
||||
] as const)("sends a note-bearing update when the cursor read %s, and says the note is unproven", async (_label, fail) => {
|
||||
const qitemId = await claimedRow();
|
||||
repo.update({ qitemId, actorSession: "reviewer@fixture", transitionNote: "must persist" }); // an older matching note
|
||||
const before = repo.listTransitions(qitemId).length;
|
||||
const result = await runUpdate(qitemId, ["--note", "must persist", "--summary", "kept summary"], undefined,
|
||||
(call) => call === 0 ? fail() : undefined);
|
||||
|
||||
expect(result.posts).toBe(1);
|
||||
expect(result.transitions).toHaveLength(before + 1);
|
||||
expect(result.transitions.at(-1)?.transitionNote).toMatch(/^must persist/);
|
||||
expect(result.row.summary).toBe("kept summary");
|
||||
expect(result.code).toBe(1);
|
||||
const said = result.errors.join("\n");
|
||||
expect(said).toContain("queue_update_readback_failed");
|
||||
expect(said).toContain("can't be shown to be new");
|
||||
expect(said).not.toContain("queue_update_preflight_failed");
|
||||
});
|
||||
|
||||
it("gives each run its own exit status: a failed update, then a successful one in the same process", async () => {
|
||||
const qitemId = await claimedRow();
|
||||
const failed = await runUpdate(qitemId, ["--note", "dropped"], () => null);
|
||||
const succeeded = await runUpdate(qitemId, ["--note", "kept"]);
|
||||
|
||||
expect(failed.code).toBe(1);
|
||||
expect(succeeded.code, succeeded.errors.join("\n")).toBe(0);
|
||||
// The action's failure is returned, never passed to process.exit(), which can cut off piped output.
|
||||
expect([failed.exits, succeeded.exits]).toEqual([0, 0]);
|
||||
});
|
||||
|
||||
// A note is this write's only when this write's actor recorded it, as written or with what the daemon adds for it.
|
||||
it.each([
|
||||
["another writer's identical note", "other@fixture", "must persist"],
|
||||
["a lookalike note that starts the same way", "reviewer@fixture", "must persist (correction: still blocked)"],
|
||||
["a lookalike note on a new line", "reviewer@fixture", "must persist\nbut still blocked"],
|
||||
] as const)("doesn't take %s for a dropped note", async (_label, actor, note) => {
|
||||
const qitemId = await claimedRow();
|
||||
const result = await runUpdate(qitemId, ["--note", "must persist"], () => {
|
||||
repo.update({ qitemId, actorSession: actor, transitionNote: note });
|
||||
return null;
|
||||
});
|
||||
|
||||
expect(result.transitions.at(-1)?.transitionNote).toBe(note);
|
||||
expect(result.code).toBe(1);
|
||||
const said = JSON.parse(result.errors.join("\n")) as { error: string; message: string };
|
||||
expect(said.error).toBe("queue_update_readback_failed");
|
||||
expect(said.message).toContain('no newer canonical transition carried requested note "must persist" from "reviewer@fixture"');
|
||||
});
|
||||
|
||||
// A seat on host-a writing to a daemon at a direct endpoint sends reviewer@fixture@host-a. Only a daemon whose own id is
|
||||
// host-a drops that suffix; any other daemon keeps it, and its own reviewer@fixture is a different seat.
|
||||
it("doesn't take the selected daemon's own same-named seat for a dropped note's writer", async () => {
|
||||
const qitemId = await claimedRow();
|
||||
const result = await runUpdate(qitemId, ["--note", "must persist"], () => {
|
||||
repo.update({ qitemId, actorSession: "reviewer@fixture", transitionNote: "must persist" });
|
||||
return null;
|
||||
}, undefined, { sender: "reviewer@fixture@host-a", selfHostId: "host-b" });
|
||||
|
||||
expect(result.code).toBe(1);
|
||||
const said = JSON.parse(result.errors.join("\n")) as { error: string; message: string };
|
||||
expect(said.message).toContain('from "reviewer@fixture@host-a"');
|
||||
expect(said.message).not.toContain('or "reviewer@fixture"');
|
||||
});
|
||||
|
||||
it("accepts the bare actor when the selected daemon answers with the sender's own host id", async () => {
|
||||
const qitemId = await claimedRow();
|
||||
const result = await runUpdate(qitemId, ["--note", "must persist"], undefined, undefined,
|
||||
{ sender: "reviewer@fixture@host-a", selfHostId: "host-a" });
|
||||
|
||||
expect(result.code, result.errors.join("\n")).toBe(0);
|
||||
expect(result.transitions.at(-1)).toMatchObject({ actorSession: "reviewer@fixture", transitionNote: "must persist" });
|
||||
});
|
||||
|
||||
it("accepts the summary suffix the daemon adds to a note", async () => {
|
||||
const qitemId = await claimedRow();
|
||||
const result = await runUpdate(qitemId, ["--note", "must persist", "--summary", "new subject"]);
|
||||
|
||||
expect(result.code, result.errors.join("\n")).toBe(0);
|
||||
expect(result.transitions.at(-1)?.transitionNote).toBe("must persist (summary updated)");
|
||||
});
|
||||
|
||||
it("accepts a note only after its canonical transition appears", async () => {
|
||||
const qitemId = await claimedRow();
|
||||
const result = await runUpdate(qitemId, ["--note", "must persist"]);
|
||||
|
||||
expect(result.code, result.errors.join("\n")).toBe(0);
|
||||
expect(result.transitions.at(-1)?.transitionNote).toBe("must persist");
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,117 @@
|
||||
// A verified `rig queue update` prints the canonical row it read back, and keeps what only the write's response carries:
|
||||
// the park advisory (#2739). A real `queueCommand` talks over loopback HTTP to the real queueRoutes and QueueRepository,
|
||||
// so the POST and the GET that reads it back are the daemon's own.
|
||||
import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { serve } from "@hono/node-server";
|
||||
import { Hono } from "hono";
|
||||
import type Database from "better-sqlite3";
|
||||
import { createDb } from "../../daemon/src/db/connection.js";
|
||||
import { migrate } from "../../daemon/src/db/migrate.js";
|
||||
import { ALL_MIGRATIONS } from "../../daemon/src/db/all-migrations.js";
|
||||
import { EventBus } from "../../daemon/src/domain/event-bus.js";
|
||||
import { QueueRepository } from "../../daemon/src/domain/queue-repository.js";
|
||||
import { queueRoutes } from "../../daemon/src/routes/queue.js";
|
||||
import { DaemonClient } from "../src/client.js";
|
||||
import { queueCommand } from "../src/commands/queue.js";
|
||||
import { STATE_FILE } from "../src/daemon-lifecycle.js";
|
||||
|
||||
let db: Database.Database;
|
||||
let repo: QueueRepository;
|
||||
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
||||
let server: any;
|
||||
let port = 0;
|
||||
|
||||
beforeAll(async () => {
|
||||
db = createDb();
|
||||
migrate(db, ALL_MIGRATIONS);
|
||||
const bus = new EventBus(db);
|
||||
repo = new QueueRepository(db, bus, { validateRig: () => true });
|
||||
const app = new Hono();
|
||||
app.use("*", async (c, next) => { c.set("eventBus" as never, bus); c.set("queueRepo" as never, repo); await next(); });
|
||||
app.route("/api/queue", queueRoutes());
|
||||
await new Promise<void>((resolve) => {
|
||||
server = serve({ fetch: app.fetch, hostname: "127.0.0.1", port: 0 }, (info: { port: number }) => { port = info.port; resolve(); });
|
||||
});
|
||||
});
|
||||
|
||||
afterAll(async () => {
|
||||
if (server) await new Promise<void>((resolve) => server.close(() => resolve()));
|
||||
db?.close();
|
||||
});
|
||||
|
||||
beforeEach(() => {
|
||||
// A seat, on this host's own daemon: the CLI stamps it as the sender. Ambient endpoints would win over the state file.
|
||||
vi.stubEnv("OPENRIG_SESSION_NAME", "dev-x@rig");
|
||||
vi.stubEnv("TMUX_PANE", "");
|
||||
for (const key of ["OPENRIG_URL", "RIGGED_URL", "OPENRIG_PORT"]) vi.stubEnv(key, "");
|
||||
});
|
||||
afterEach(() => vi.unstubAllEnvs());
|
||||
|
||||
async function runCli(args: string[]): Promise<{ out: string; err: string; exit: number | undefined }> {
|
||||
const out: string[] = [];
|
||||
const err: string[] = [];
|
||||
const ol = console.log, oe = console.error;
|
||||
console.log = (...a: unknown[]) => { out.push(a.join(" ")); };
|
||||
console.error = (...a: unknown[]) => { err.push(a.join(" ")); };
|
||||
process.exitCode = undefined;
|
||||
try {
|
||||
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
||||
await queueCommand({
|
||||
lifecycleDeps: {
|
||||
spawn: () => ({ pid: 1, unref() {} }), fetch: async () => ({ ok: true }), kill: () => true,
|
||||
readFile: (p: string) => (p === STATE_FILE ? JSON.stringify({ pid: 1, port, host: "127.0.0.1", db: "x", startedAt: "x" }) : null),
|
||||
writeFile() {}, removeFile() {}, exists: (p: string) => p === STATE_FILE, mkdirp() {}, openForAppend: () => 3, isProcessAlive: () => true,
|
||||
},
|
||||
clientFactory: (url: string) => new DaemonClient(url),
|
||||
} as any).parseAsync(["node", "rig", ...args]);
|
||||
} finally {
|
||||
console.log = ol;
|
||||
console.error = oe;
|
||||
}
|
||||
const exit = process.exitCode;
|
||||
process.exitCode = undefined;
|
||||
return { out: out.join("\n"), err: err.join("\n"), exit };
|
||||
}
|
||||
|
||||
async function claimedRow(): Promise<string> {
|
||||
const row = await repo.create({ sourceSession: "orch@rig", destinationSession: "dev-x@rig", body: "work", nudge: false });
|
||||
repo.claim({ qitemId: row.qitemId, destinationSession: "dev-x@rig" });
|
||||
return row.qitemId;
|
||||
}
|
||||
|
||||
const park = (id: string) => ["update", id, "--state", "blocked", "--blocked-on", "external:vendor-fix", "--note", "waiting on the vendor"];
|
||||
|
||||
describe("a verified queue update keeps the write's park advisory", () => {
|
||||
it.each([
|
||||
["the compact JSON receipt", "true", ["--json"]],
|
||||
["the full JSON row", "false", ["--json"]],
|
||||
["the full row with --full", "true", ["--full", "--json"]],
|
||||
] as const)("in %s", async (_label, compact, flags) => {
|
||||
vi.stubEnv("OPENRIG_OUTPUT_COMPACT", compact);
|
||||
const id = await claimedRow();
|
||||
const { out, err, exit } = await runCli([...park(id), ...flags]);
|
||||
|
||||
expect(exit, err).toBeFalsy();
|
||||
const printed = JSON.parse(out) as { state?: string; advisories?: Array<{ code?: string }> };
|
||||
expect(printed.advisories).toEqual([expect.objectContaining({ code: "park_without_wake" })]);
|
||||
expect(repo.getByIdOrThrow(id)).toMatchObject({ state: "blocked", blockedOn: "external:vendor-fix" });
|
||||
});
|
||||
|
||||
it("in the compact text receipt", async () => {
|
||||
vi.stubEnv("OPENRIG_OUTPUT_COMPACT", "true");
|
||||
const id = await claimedRow();
|
||||
const { out, err, exit } = await runCli(park(id));
|
||||
|
||||
expect(exit, err).toBeFalsy();
|
||||
expect(out).toMatch(/advisory: No wake is set for this park/);
|
||||
});
|
||||
|
||||
it("and a park with a wake carries none", async () => {
|
||||
vi.stubEnv("OPENRIG_OUTPUT_COMPACT", "true");
|
||||
const id = await claimedRow();
|
||||
const { out, err, exit } = await runCli([...park(id), "--wake-after", "10m", "--json"]);
|
||||
|
||||
expect(exit, err).toBeFalsy();
|
||||
expect((JSON.parse(out) as { advisories?: unknown }).advisories).toBeUndefined();
|
||||
});
|
||||
});
|
||||
@@ -49,6 +49,8 @@ function row(over: Row = {}, waiting: Row = {}): Row {
|
||||
}
|
||||
|
||||
function makeDeps(routes: Record<string, { status: number; data: unknown } | Array<{ status: number; data: unknown }>>): QueueDeps {
|
||||
const transitions: Array<{ transitionId: number; transitionNote: string | null; actorSession: string }> =
|
||||
[{ transitionId: 1, transitionNote: null, actorSession: "stub@rig" }];
|
||||
const pick = (key: string, fallback: { status: number; data: unknown }) => {
|
||||
const r = routes[key];
|
||||
return (Array.isArray(r) ? r.shift() : r) ?? fallback;
|
||||
@@ -56,8 +58,27 @@ function makeDeps(routes: Record<string, { status: number; data: unknown } | Arr
|
||||
return {
|
||||
lifecycleDeps: {} as QueueDeps["lifecycleDeps"],
|
||||
clientFactory: () => ({
|
||||
get: vi.fn(async (p: string) => pick(`GET ${p}`, { status: 200, data: {} })),
|
||||
post: vi.fn(async (p: string) => pick(`POST ${p}`, { status: 500, data: { error: "unrouted" } })),
|
||||
// The sender this stub stamps; the transitions it records carry it as their actor, as the daemon does.
|
||||
senderIdentity: async () => ({ session: "stub@rig", originUnknown: false }),
|
||||
get: vi.fn(async (p: string) => {
|
||||
const explicit = routes[`GET ${p}`];
|
||||
if (explicit) return Array.isArray(explicit) ? explicit.shift()! : explicit;
|
||||
if (p.endsWith("/transitions")) return { status: 200, data: transitions };
|
||||
const update = routes[`POST ${p}/update`];
|
||||
if (update) {
|
||||
const response = Array.isArray(update) ? update[0] : update;
|
||||
return { status: 200, data: response?.data ?? {} };
|
||||
}
|
||||
return { status: 200, data: {} };
|
||||
}),
|
||||
post: vi.fn(async (p: string, body: unknown) => {
|
||||
const response = pick(`POST ${p}`, { status: 500, data: { error: "unrouted" } });
|
||||
const transitionNote = body && typeof body === "object" ? (body as { transitionNote?: unknown }).transitionNote : undefined;
|
||||
if (response.status < 400 && typeof transitionNote === "string") {
|
||||
transitions.push({ transitionId: transitions.length + 1, transitionNote, actorSession: "stub@rig" });
|
||||
}
|
||||
return response;
|
||||
}),
|
||||
}) as unknown as ReturnType<QueueDeps["clientFactory"]>,
|
||||
};
|
||||
}
|
||||
|
||||
@@ -31,6 +31,8 @@ interface StubResponse {
|
||||
data: unknown;
|
||||
}
|
||||
|
||||
const STUB_SENDER = "stub@rig";
|
||||
|
||||
function makeDeps(opts?: {
|
||||
routes?: Record<string, StubResponse | StubResponse[]>;
|
||||
}): { deps: QueueDeps; calls: Array<{ method: string; path: string; body?: unknown }> } {
|
||||
@@ -41,10 +43,37 @@ function makeDeps(opts?: {
|
||||
deps: {
|
||||
lifecycleDeps: {} as QueueDeps["lifecycleDeps"],
|
||||
clientFactory: () => ({
|
||||
// The sender this stub stamps; the transitions it synthesizes record it as their actor, as the daemon does.
|
||||
senderIdentity: async () => ({ session: STUB_SENDER, originUnknown: false }),
|
||||
get: vi.fn(async (path: string) => {
|
||||
calls.push({ method: "GET", path });
|
||||
const route = routes[`GET ${path}`];
|
||||
return (Array.isArray(route) ? route.shift() : route) ?? { status: 200, data: {} };
|
||||
if (route) return Array.isArray(route) ? route.shift() : route;
|
||||
const transitionMatch = /^\/api\/queue\/([^/]+)\/transitions$/.exec(path);
|
||||
if (transitionMatch) {
|
||||
const qitemId = decodeURIComponent(transitionMatch[1]!);
|
||||
const update = [...calls].reverse().find((call) => call.method === "POST"
|
||||
&& call.path === `/api/queue/${encodeURIComponent(qitemId)}/update`);
|
||||
const transitionNote = update?.body && typeof update.body === "object"
|
||||
? (update.body as { transitionNote?: unknown }).transitionNote
|
||||
: undefined;
|
||||
return { status: 200, data: [
|
||||
{ transitionId: 1, transitionNote: null, actorSession: STUB_SENDER },
|
||||
...(typeof transitionNote === "string" ? [{ transitionId: 2, transitionNote, actorSession: STUB_SENDER }] : []),
|
||||
] };
|
||||
}
|
||||
const match = /^\/api\/queue\/([^/]+)$/.exec(path);
|
||||
if (match) {
|
||||
const qitemId = decodeURIComponent(match[1]!);
|
||||
const update = [...calls].reverse().find((call) => call.method === "POST"
|
||||
&& call.path === `/api/queue/${encodeURIComponent(qitemId)}/update`);
|
||||
if (update?.body && typeof update.body === "object") {
|
||||
const body = Object.fromEntries(Object.entries(update.body as Record<string, unknown>)
|
||||
.filter(([key, value]) => value !== undefined && !["transitionNote", "actorSession", "reopen"].includes(key)));
|
||||
return { status: 200, data: { qitemId, ...body } };
|
||||
}
|
||||
}
|
||||
return { status: 200, data: {} };
|
||||
}),
|
||||
getText: vi.fn(async (path: string) => {
|
||||
calls.push({ method: "GET", path });
|
||||
@@ -83,6 +112,7 @@ describe("rig queue CLI", () => {
|
||||
// AMBIENT OPENRIG_SESSION_NAME (which masked the env-less break in a managed runner). Per-test stubs
|
||||
// that need a specific seat override this.
|
||||
vi.stubEnv("OPENRIG_SESSION_NAME", "seat@rig");
|
||||
vi.stubEnv("TMUX_PANE", "");
|
||||
logs = [];
|
||||
errors = [];
|
||||
vi.spyOn(console, "log").mockImplementation((...args) => logs.push(args.join(" ")));
|
||||
@@ -1203,6 +1233,37 @@ describe("rig queue CLI", () => {
|
||||
evidenceRef: undefined,
|
||||
transitionNote: "audit bytes",
|
||||
});
|
||||
expect(calls.some((c) => c.method === "GET" && c.path === "/api/queue/qitem-note")).toBe(true);
|
||||
});
|
||||
|
||||
it("update reports success only after canonical read-back matches the requested state", async () => {
|
||||
const row = { qitemId: "qitem-verified", state: "in-progress", summary: "working" };
|
||||
const { deps, calls } = makeDeps({ routes: {
|
||||
"POST /api/queue/qitem-verified/update": { status: 200, data: { qitemId: "qitem-verified", state: "in-progress" } },
|
||||
"GET /api/queue/qitem-verified": { status: 200, data: row },
|
||||
} });
|
||||
await createProgram({ queueDeps: deps }).parseAsync([
|
||||
"node", "rig", "queue", "update", "qitem-verified", "--state", "in-progress", "--summary", "working", "--json",
|
||||
]);
|
||||
expect(calls.map((call) => `${call.method} ${call.path}`)).toEqual([
|
||||
"POST /api/queue/qitem-verified/update",
|
||||
"GET /api/queue/qitem-verified",
|
||||
]);
|
||||
expect(JSON.parse(logs.at(-1)!)).toMatchObject(row);
|
||||
expect(process.exitCode).toBeUndefined();
|
||||
});
|
||||
|
||||
it("update fails closed when canonical read-back does not contain the requested state", async () => {
|
||||
const { deps } = makeDeps({ routes: {
|
||||
"POST /api/queue/qitem-mismatch/update": { status: 200, data: { qitemId: "qitem-mismatch", state: "in-progress" } },
|
||||
"GET /api/queue/qitem-mismatch": { status: 200, data: { qitemId: "qitem-mismatch", state: "pending" } },
|
||||
} });
|
||||
await createProgram({ queueDeps: deps }).parseAsync([
|
||||
"node", "rig", "queue", "update", "qitem-mismatch", "--state", "in-progress", "--json",
|
||||
]);
|
||||
expect(process.exitCode).toBe(1);
|
||||
expect(JSON.parse(errors.at(-1)!)).toMatchObject({ error: "queue_update_readback_failed" });
|
||||
expect(logs).toEqual([]);
|
||||
});
|
||||
|
||||
it("update forwards an explicitly accepted closure deadline without a state change", async () => {
|
||||
|
||||
@@ -62,12 +62,12 @@ function depsThrowing(err: Error): { queueDeps: QueueDeps } {
|
||||
async function runCli(args: string[], deps?: Parameters<typeof createProgram>[0]) {
|
||||
const out: string[] = [];
|
||||
const err: string[] = [];
|
||||
let exitCode = 0;
|
||||
const program = createProgram(deps);
|
||||
await runProgram(program, ["node", "rig", ...args], {
|
||||
// The run's status is what runProgram returns; it calls exit only on its own error paths, with the same code.
|
||||
const exitCode = await runProgram(program, ["node", "rig", ...args], {
|
||||
out: (l) => out.push(l),
|
||||
err: (l) => err.push(l),
|
||||
exit: (c) => { exitCode = c; },
|
||||
exit: () => undefined,
|
||||
});
|
||||
return { out: out.join("\n"), err: err.join("\n"), exitCode };
|
||||
}
|
||||
@@ -122,6 +122,17 @@ describe("render response integrity — 3-part error + honest exit, never daemon
|
||||
beforeEach(() => { vi.stubEnv("OPENRIG_SESSION_NAME", "harness@rig"); process.exitCode = undefined; });
|
||||
afterEach(() => { vi.unstubAllEnvs(); });
|
||||
|
||||
it("an unresolved queue actor stays nonzero when a wrapper pipes stdout", async () => {
|
||||
vi.stubEnv("OPENRIG_SESSION_NAME", "");
|
||||
vi.stubEnv("RIGGED_SESSION_NAME", "");
|
||||
vi.stubEnv("TMUX_PANE", "");
|
||||
const { exitCode } = await runCli(
|
||||
["queue", "update", "qitem-x", "--state", "in-progress", "--json"],
|
||||
depsThrowing(new Error("daemon must not be contacted")),
|
||||
);
|
||||
expect(exitCode).toBe(1);
|
||||
});
|
||||
|
||||
it("bad-response renders fact/consequence/action JSON with nonzero exit (json)", async () => {
|
||||
const err = new DaemonResponseError(502, "<html>502 Bad Gateway</html>");
|
||||
const { out, exitCode } = await runCli(
|
||||
|
||||
@@ -241,6 +241,8 @@ describe("Send CLI", () => {
|
||||
// (Hermetic-gate default is env-UNSET, so without this env-less delivery would render the unknown
|
||||
// marker.) Nested describes that stub their own envs re-run after this; the block afterEach restores.
|
||||
vi.stubEnv("OPENRIG_SESSION_NAME", "sender@my-rig");
|
||||
// A runner inside tmux would otherwise let the managed-pane fallback recover the runner's own seat.
|
||||
vi.stubEnv("TMUX_PANE", "");
|
||||
vi.stubEnv("RIGGED_SESSION_NAME", "");
|
||||
});
|
||||
afterEach(() => { vi.unstubAllEnvs(); });
|
||||
|
||||
@@ -0,0 +1,121 @@
|
||||
import { afterEach, describe, expect, it, vi } from "vitest";
|
||||
import { Command } from "commander";
|
||||
import { createDb } from "../../daemon/src/db/connection.js";
|
||||
import { migrate } from "../../daemon/src/db/migrate.js";
|
||||
import { ALL_MIGRATIONS } from "../../daemon/src/db/all-migrations.js";
|
||||
import { createTestApp } from "../../daemon/test/helpers/test-app.js";
|
||||
import { runProgram } from "../src/cli-error.js";
|
||||
import { senderIdentityHeaders } from "../src/client.js";
|
||||
import { bindingIsCurrent, resetSenderSession, resolveSenderSession, TMUX_IDENTITY_TIMEOUT_MS } from "../src/sender-identity.js";
|
||||
|
||||
const execFileSync = vi.hoisted(() => vi.fn());
|
||||
vi.mock("node:child_process", async (original) => ({ ...(await original<typeof import("node:child_process")>()), execFileSync }));
|
||||
|
||||
afterEach(() => {
|
||||
vi.unstubAllEnvs();
|
||||
execFileSync.mockReset();
|
||||
resetSenderSession();
|
||||
});
|
||||
|
||||
function outsideASeatEnv(pane: string): void {
|
||||
vi.stubEnv("OPENRIG_SESSION_NAME", "");
|
||||
vi.stubEnv("RIGGED_SESSION_NAME", "");
|
||||
vi.stubEnv("TMUX_PANE", pane);
|
||||
}
|
||||
|
||||
describe("managed sender identity recovery", () => {
|
||||
it("uses reconciled pane metadata when the live process missed the seat env", () => {
|
||||
outsideASeatEnv("%149");
|
||||
const metadata: Record<string, string> = { "@rigged_session_name": "producer-lead@studio", "@rigged_node_id": "node-1" };
|
||||
const read = (pane: string, option: string) => {
|
||||
expect(pane).toBe("%149");
|
||||
return metadata[option];
|
||||
};
|
||||
const bound = (nodeId: string, session: string) => nodeId === "node-1" && session === "producer-lead@studio";
|
||||
expect(resolveSenderSession(read, bound)).toBe("producer-lead@studio");
|
||||
expect(senderIdentityHeaders(undefined, undefined, read, bound)).toEqual({
|
||||
"X-OpenRig-Session": "producer-lead@studio",
|
||||
});
|
||||
});
|
||||
|
||||
it("keeps the process seat env authoritative over pane metadata", () => {
|
||||
vi.stubEnv("OPENRIG_SESSION_NAME", "dev-coder@studio");
|
||||
vi.stubEnv("TMUX_PANE", "%149");
|
||||
expect(resolveSenderSession(() => "producer-lead@studio", () => true)).toBe("dev-coder@studio");
|
||||
});
|
||||
|
||||
it("bounds the tmux read, and a read that times out leaves the seat unresolved", () => {
|
||||
outsideASeatEnv("%149");
|
||||
// The reader catches what the call throws, so the options are checked here, not inside the mock.
|
||||
execFileSync.mockImplementationOnce(() => {
|
||||
throw Object.assign(new Error("spawnSync tmux ETIMEDOUT"), { code: "ETIMEDOUT" });
|
||||
});
|
||||
expect(resolveSenderSession()).toBeUndefined();
|
||||
expect(execFileSync).toHaveBeenCalledTimes(1);
|
||||
expect(execFileSync.mock.calls[0]![2]).toMatchObject({ timeout: TMUX_IDENTITY_TIMEOUT_MS });
|
||||
expect(TMUX_IDENTITY_TIMEOUT_MS).toBeLessThanOrEqual(2_000);
|
||||
});
|
||||
|
||||
it("stays unresolved when neither env nor managed pane metadata exists", () => {
|
||||
outsideASeatEnv("%404");
|
||||
expect(resolveSenderSession(() => undefined, () => true)).toBeUndefined();
|
||||
});
|
||||
|
||||
it("reads the pane once per run: a command's preflight and its requests share it, and the next run reads again", async () => {
|
||||
outsideASeatEnv("%149");
|
||||
execFileSync.mockImplementation((_tmux: string, args: string[]) => args.at(-1) === "@rigged_session_name" ? "producer-lead@studio\n" : "node-1\n");
|
||||
const bound = () => true;
|
||||
const run = () => runProgram(new Command().action(() => {
|
||||
expect(resolveSenderSession(undefined, bound)).toBe("producer-lead@studio"); // the preflight
|
||||
expect(senderIdentityHeaders(undefined, undefined, undefined, bound)).toEqual({ "X-OpenRig-Session": "producer-lead@studio" });
|
||||
expect(senderIdentityHeaders("origin-a", undefined, undefined, bound)).toEqual({ "X-OpenRig-Session": "producer-lead@studio@origin-a" });
|
||||
}), ["node", "rig"], { exit: () => undefined });
|
||||
expect(await run()).toBe(0);
|
||||
expect(execFileSync).toHaveBeenCalledTimes(2); // the seat and its node, once
|
||||
expect(await run()).toBe(0);
|
||||
expect(execFileSync).toHaveBeenCalledTimes(4);
|
||||
});
|
||||
});
|
||||
|
||||
describe("a pane's seat holds only while the daemon binds it", () => {
|
||||
// A real claim sets the session's metadata and the daemon's binding. A real unclaim deletes the binding in its own
|
||||
// transaction even when clearing the metadata fails, which leaves the pane naming its old seat.
|
||||
async function claimedPane() {
|
||||
const db = createDb();
|
||||
migrate(db, ALL_MIGRATIONS);
|
||||
const setup = createTestApp(db);
|
||||
const metadata = new Map<string, string>();
|
||||
vi.mocked(setup.tmuxAdapter.setSessionOption).mockImplementation(async (_session: string, key: string, value: string) => {
|
||||
metadata.set(key, value);
|
||||
return { ok: true as const };
|
||||
});
|
||||
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 });
|
||||
if (!claim.ok) throw new Error(claim.error);
|
||||
const read = (_pane: string, option: string) => metadata.get(option) || undefined;
|
||||
const bound = (nodeId: string, session: string) => bindingIsCurrent(db, nodeId, session);
|
||||
return { db, setup, claim, read, bound };
|
||||
}
|
||||
|
||||
it("reads a claimed session the daemon binds as its seat", async () => {
|
||||
outsideASeatEnv("%1");
|
||||
const { db, read, bound } = await claimedPane();
|
||||
expect(resolveSenderSession(read, bound)).toBe("manual-adoption");
|
||||
db.close();
|
||||
});
|
||||
|
||||
it("reads a released session as unknown when clearing its metadata failed", async () => {
|
||||
outsideASeatEnv("%1");
|
||||
const { db, setup, claim, read, bound } = await claimedPane();
|
||||
vi.mocked(setup.tmuxAdapter.setSessionOption).mockImplementation(async () => { throw new Error("tmux: no server"); });
|
||||
expect((await setup.rigLifecycleService.unclaimSession(claim.sessionId)).ok).toBe(true);
|
||||
expect(read("%1", "@rigged_session_name")).toBe("manual-adoption");
|
||||
expect(resolveSenderSession(read, bound)).toBeUndefined();
|
||||
db.close();
|
||||
});
|
||||
});
|
||||
@@ -95,6 +95,17 @@ rig queue handoff <qitem-id> \
|
||||
A queue write (`create`, `claim`, `update`, `handoff`, `handoff-and-complete`) answers with the row it wrote; a
|
||||
handoff answers with the closed row and its successor. When that answer shows what you need to establish, use it.
|
||||
|
||||
`queue update` is stricter: before it reports success, the CLI reads the row back from the canonical queue and checks
|
||||
every requested row field using daemon semantics (including normalized deadlines and appended closing resolutions).
|
||||
When `--note` is present, it first records the transition cursor and then requires a newer canonical transition,
|
||||
recorded under the actor the write was sent as, that carries those note bytes as written or with only what the daemon
|
||||
adds for that request (the reopen prefix, the summary suffix, a revised deadline's receipt). Another writer's note, or
|
||||
a note that only starts the same way, doesn't count. If that cursor can't be read, the write is still sent, and the
|
||||
note can't be shown to be new: the command says so and exits non-zero. A failed post-write check exits non-zero and
|
||||
says the write may have committed. Reconcile with `rig queue show <id> --full --json` before retrying. Preserve that
|
||||
exit status in wrappers, and use `set -o pipefail` when piping its output so a downstream formatter cannot turn
|
||||
failure into success.
|
||||
|
||||
`create` also prints `Queue create request ID: <id> (not proof of persistence)` on stderr *before* it sends the write,
|
||||
so an interrupted create can be reconciled. That ID names a request, not a row. The row is the `qitemId` on stdout
|
||||
when the command exits 0. A non-zero exit means no row was confirmed, even when stdout carries a `qitemId`: an
|
||||
|
||||
@@ -93,7 +93,7 @@
|
||||
"openrig-operating-model/templates/SOP.md": "1eaf0daf4f9bbf0715e6a5aaf0f29f030af5fe8b8a10232488ee97c507afa6dc",
|
||||
"openrig-operating-model/templates/SPEC.md": "f126536daa59cea3f1cfb0898d83c7cdf99f6392f6f155ea64fe55187dd0f5f1",
|
||||
"openrig-skills/SKILL.md": "6d7bf933ac40531f11757a31a93d26dcbda617f1b85f48a21865b3a5c5e2ddaa",
|
||||
"openrig-user/SKILL.md": "6176761014c9ecd61675cb47d4de991de6dd126a92be7e06c3ef31aa0573f82e",
|
||||
"openrig-user/SKILL.md": "ef22c218348bdf06d0f56d9de86a3ac8a5f954eae161a212d1e1010a19e464b9",
|
||||
"orienting-to-an-inherited-seat/CHANGELOG.md": "a91f117e88034d8bae9b723916235e858fbc151983cc7c32784f5d4ba186274a",
|
||||
"orienting-to-an-inherited-seat/SKILL.md": "278c554b3bb4c481f2928499f02442d947ffd4751d60f830610b652e24cb0468",
|
||||
"queue-handoff/SKILL.md": "96251f6830006bb9d8ede2e5826ab936df4a9fb91090187a23ccaf860ab76336",
|
||||
|
||||
Reference in new issue
Block a user