From 06090c37a96f7fa088fd3f6e657427404cdc9614 Mon Sep 17 00:00:00 2001 From: Kurtis Jantzen Date: Thu, 10 Sep 2026 15:48:50 -0700 Subject: [PATCH] Persist ordinary write retries and add nonblocking agent connections --- README.md | 42 ++++++++++++ src/cli-commands/agent.ts | 3 +- src/cli.test.ts | 84 +++++++++++++++++++++++- src/client.ts | 15 ++++- src/commands.ts | 75 +++++++++++++++++---- src/config.ts | 2 + src/write-recovery.test.ts | 131 +++++++++++++++++++++++++++++++++++++ src/write-recovery.ts | 94 ++++++++++++++++++++++++++ 8 files changed, 430 insertions(+), 16 deletions(-) create mode 100644 src/write-recovery.test.ts create mode 100644 src/write-recovery.ts diff --git a/README.md b/README.md index 5a5cc1d0..92629ae1 100644 --- a/README.md +++ b/README.md @@ -77,6 +77,48 @@ accepted fields (`data.id`, `data.release_id`, `data.operation_id`) and ## Configuration +### Nonblocking agent connection + +`screenrig agent connect --no-wait --print-url` starts or resumes a connection, +reads one server status snapshot, and returns a JSON envelope. While approval is +pending, `data.status` is `pending`, `data.approval_url` contains the dashboard +handoff, and `data.next.command` identifies the resume command. `data.next.argv` +supplies its arguments, preserving the selected config and API origin. Open the handoff +for the intended user, then run `screenrig agent connect --no-wait` again. An +approved connection completes credential collection and activation and returns +`data.status: active`. No credential is included in either result. + +Without `--print-url`, the CLI tries to open the browser and includes the handoff +URL in the pending result only if opening fails. `--no-wait` defaults to a +30-second status-read budget; it does not wait for a person to approve. Normal +`agent connect` retains its existing approval wait of up to 24 hours. `--timeout` +overrides either mode's wait budget. Terminal denial, cancellation, and expiry +remain errors. + +### Recovering writes + +Ordinary application uploads/updates, playlist creates/updates/deletes, screen +mutations other than provisioning, media tag updates/deletes, K/V and comment +writes, feedback submissions, and operation cancellation persist an idempotency +key before sending the request. After an ambiguous network failure or server +error, rerun the same command with unchanged input. The CLI reuses the saved key; +it does not automatically send another request within the failed invocation. +`write_recovery_saved` indicates that recovery state was retained. + +The private config stores only request fingerprints, keys, and timestamps, never +request payloads. Fingerprints include the origin, credential, target, request +body, and revision. Changed requests receive different keys. A completed command +clears its pending state; application acceptance followed by a failed processing +wait retains it so retrying does not create another application. Definite +refusals still require reconciliation; revision checks remain in force. + +Automatic replay stops after 23 hours, before the server's 24-hour replay window +ends. Inspect the resource before explicitly supplying a new `--idempotency-key` +for a reconciled write. Explicit keys remain supported. Pending entries are not +silently evicted; resolve outstanding writes if the 256-entry limit is reached. +Enrollment, generation, media uploads/batches, bundle imports, and browser +handoffs retain their existing specialized recovery behavior. + User-private config lives outside the replaceable plugin directory (`$XDG_CONFIG_HOME/screenrig/config.json`, or `%APPDATA%\screenrig\config.json` on Windows). The default service is `https://api.screenrig.ai`. diff --git a/src/cli-commands/agent.ts b/src/cli-commands/agent.ts index 724fa590..cb9a7467 100644 --- a/src/cli-commands/agent.ts +++ b/src/cli-commands/agent.ts @@ -15,7 +15,8 @@ export function registerAgentCommands(root: Command, bind: CommandActionBinder): addCommandNotes(agent.command("connect").description("Connect this installation with approval") .option("--name ", "Set this agent installation name") .option("--print-url", "Return the browser handoff URL") - .action(bind(handleAgentConnect)), "Approval expires after 24 hours; --timeout defaults to 86400000 ms. Retry agent connect to resume after an interrupted wait."); + .option("--no-wait", "Return pending status instead of waiting for dashboard approval") + .action(bind(handleAgentConnect)), "Approval expires after 24 hours. Default mode waits up to 86400000 ms. --no-wait reads one status snapshot (30000 ms default), returning pending with a resume command or completing an approved connection. --print-url places the handoff URL in the pending result. Retry agent connect to resume after an interrupted wait."); agent.command("status").description("Inspect this agent's connection") .action(bind(handleAgentStatus)); diff --git a/src/cli.test.ts b/src/cli.test.ts index 306c3e41..57509a28 100644 --- a/src/cli.test.ts +++ b/src/cli.test.ts @@ -579,6 +579,86 @@ test("agent connect resumes after its cached approval expiry when approved while await rm(result.configDir, { recursive: true, force: true }); }); +test("agent connect --no-wait returns a pending handoff and resumes the same approved connection", async () => { + const transport = new FakeTransport(); + let sealed: ReturnType | undefined; + const connectionToken = `sac_${"C".repeat(43)}`; + transport.on("POST", "/api/v1/agent-connections", (req) => { + const input = req.body as { recipient_public_key: { kty: "OKP"; crv: "X25519"; x: string } }; + sealed = agentConnectionEnvelope(input.recipient_public_key); + sealed.collection.issuance_expires_at = "2026-08-16T16:00:00.000Z"; + return { + status: 201, + headers: { "cache-control": "private, no-store", "referrer-policy": "no-referrer" }, + body: { + connection_id: sealed.connectionId, + connection_token: connectionToken, + approval_url: `https://dashboard.screenrig.ai/agents/connect/${sealed.connectionId}`, + expires_at: "2026-08-15T17:00:00.000Z", + }, + }; + }); + transport.pushStream(`event: agent.connection\ndata: ${JSON.stringify({ + connection_id: "acn_AAAAAAAAAAAAAAAAAAAAAAAA", + name: "Office Codex", + agent_type: "cli", + status: "approved", + expires_at: "2026-08-15T17:00:00.000Z", + created_at: "2026-08-14T17:00:00.000Z", + })}\n\n`); + transport.on("POST", /\/api\/v1\/agent-connections\/acn_.*\/credential/, (req) => { + assert.equal(req.headers?.authorization, `ScreenRig-Agent-Connect ${connectionToken}`); + return { status: 200, headers: { "cache-control": "private, no-store" }, body: sealed!.collection }; + }); + transport.on("POST", "/api/v1/agents/self/activate", (req) => { + assert.equal(req.headers?.authorization, `Bearer ${sealed!.pendingToken}`); + return { status: 200, headers: { "cache-control": "private, no-store" }, body: { ...sealed!.pendingAgent, state: "active", connected_at: "2026-08-14T17:00:01.000Z" } }; + }); + transport.on("GET", "/api/v1/agents/self", () => ({ + status: 200, + headers: { "cache-control": "private, no-store" }, + body: { agent: { ...sealed!.pendingAgent, state: "active", connected_at: "2026-08-14T17:00:01.000Z" }, connection_ready: true }, + })); + transport.queueStream({ chunks: [`event: agent.connection\ndata: ${JSON.stringify({ + connection_id: "acn_AAAAAAAAAAAAAAAAAAAAAAAA", name: "Office Codex", agent_type: "cli", + status: "pending", expires_at: "2026-08-15T17:00:00.000Z", created_at: "2026-08-14T17:00:00.000Z", + })}\n\n`] }); + const interrupted = await withRuntime(["agent", "connect", "--no-wait", "--print-url"], transport, { openUrl: async () => { throw new Error("print-url must not open browser"); } }); + assert.equal(interrupted.code, 0, interrupted.stdout); + const pending = JSON.parse(interrupted.stdout).data; + assert.equal(pending.status, "pending"); + assert.equal(pending.approval_url, "https://dashboard.screenrig.ai/agents/connect/acn_AAAAAAAAAAAAAAAAAAAAAAAA"); + assert.equal(pending.next.command, "screenrig agent connect --no-wait"); + assert.deepEqual(pending.next.argv, ["agent", "connect", "--no-wait", "--config", + path.join(interrupted.configDir, "screenrig", "config.json"), "--api-url", "https://api.screenrig.ai"]); + assert.equal(interrupted.stderr, ""); + assert.equal(transport.calls.filter(call => call.method === "POST").length, 1); + assert.doesNotMatch(interrupted.stdout, /sr_live_|sac_|ciphertext|nonce|private_jwk/); + const resumedFs = { mkdir, open, rename, rm, chmod, stat, homedir: () => interrupted.configDir, env: { XDG_CONFIG_HOME: interrupted.configDir } }; + const opened: string[] = []; + const resumed = await withRuntime(["agent", "connect", "--no-wait", "--name", "Office Codex"], transport, { + fs: resumedFs, + now: () => new Date("2026-08-16T05:00:00.000Z"), + openUrl: async (url) => { opened.push(url); return true; }, + }); + const result = { ...resumed, configDir: interrupted.configDir }; + assert.equal(transport.calls.filter(call => call.method === "POST" && call.path === "/api/v1/agent-connections").length, 1); + assert.equal(result.code, 0, result.stdout); + assert.deepEqual(opened, [], "approved snapshots must not reopen the approval page"); + assert.equal(JSON.parse(result.stdout).data.status, "active"); + assert.equal(result.stderr, ""); + assert.doesNotMatch(result.stdout, /sr_live_|sac_|ciphertext|nonce|private|authorization/i); + const configPath = path.join(result.configDir, "screenrig", "config.json"); + const config = await readConfigFile(configPath, { + mkdir, open, rename, rm, chmod, stat, homedir: () => result.configDir, env: { XDG_CONFIG_HOME: result.configDir }, + }); + assert.equal(config?.token, sealed?.pendingToken); + assert.equal(config?.agent_id, sealed?.agentId); + assert.equal(config?.agent_connection, undefined); + assert.equal((await stat(configPath)).mode & 0o777, 0o600); + await rm(result.configDir, { recursive: true, force: true }); +}); + test("agent disconnect revokes only this installation and preserves safe disconnected status", async () => { const transport = new FakeTransport(); const active = { ...TEST_AGENT, name: "Office Codex" }; @@ -663,7 +743,7 @@ test("agent connect resumes activation after the pending bearer was durably stor await rm(configDir, { recursive: true, force: true }); }); -test("agent connect clears private and pending bearer state when the dashboard cancels", async () => { +for (const waitFlags of [[], ["--no-wait"]]) test(`agent connect ${waitFlags.join(" ")} clears private state when the dashboard cancels`, async () => { const transport = new FakeTransport(); const connectionId = "acn_CANCELAAAAAAAAAAAAAAAA"; transport.on("POST", "/api/v1/agent-connections", () => ({ @@ -684,7 +764,7 @@ test("agent connect clears private and pending bearer state when the dashboard c expires_at: "2026-08-14T17:10:00.000Z", created_at: "2026-08-14T17:00:00.000Z", })}\n\n`); - const result = await withRuntime(["--json", "agent", "connect"], transport, { openUrl: async () => true }); + const result = await withRuntime(["--json", "agent", "connect", ...waitFlags], transport, { openUrl: async () => true }); assert.equal(result.code, ExitCode.Client); const envelope = JSON.parse(result.stdout) as { error: { code: string; next: { command: string } } }; assert.equal(envelope.error.code, "agent_connection_cancelled"); diff --git a/src/client.ts b/src/client.ts index 61237668..375a19d8 100644 --- a/src/client.ts +++ b/src/client.ts @@ -1,3 +1,4 @@ +import type { WriteRecovery } from "./write-recovery.js"; import { creditsLowWarnings, observeCreditsRemaining, parseCreditsRemainingHeader } from "./credits.js"; import { ExitCode } from "./exit-codes.js"; import { isValidIdempotencyKey, isValidRequestId, newIdempotencyKey, newRequestId } from "./ids.js"; @@ -26,6 +27,7 @@ export interface ApiClientOptions { /** When set, authenticated remaining credits are observed for the envelope warning. */ creditsOwner?: object; logger?: OperationLogger; + writeRecovery?: WriteRecovery; } export class ApiClient { @@ -36,9 +38,13 @@ export class ApiClient { private readonly timeoutMs: number; private readonly creditsOwner?: object; private readonly logger: OperationLogger; + private readonly writeRecovery?: WriteRecovery; + private readonly requestedKey?: string; constructor(options: ApiClientOptions) { this.transport = options.transport; + this.writeRecovery = options.writeRecovery; + this.requestedKey = options.idempotencyKey; this.token = options.token; this.timeoutMs = options.timeoutMs ?? 30_000; this.creditsOwner = options.creditsOwner; @@ -72,7 +78,9 @@ export class ApiClient { if (idempotencyKey !== undefined && !isValidIdempotencyKey(idempotencyKey)) { throw usageError("Invalid per-request idempotency key."); } - const headers = this.headers(idempotent === true, req.headers, idempotencyKey); + const recovery = idempotent === true && idempotencyKey === undefined ? this.writeRecovery : undefined; + const pending = await recovery?.prepare(transportRequest, this.requestedKey); + const headers = this.headers(idempotent === true, req.headers, pending?.key ?? idempotencyKey); const extraType = req.headers?.["content-type"]; const summary = requestSummary(req.body, extraType); const keys = queryKeys(req.query); @@ -97,6 +105,11 @@ export class ApiClient { span.error(err); throw err; } + // Definite refusals need reconciliation, not automatic replay of a stale key. + // Keep ambiguous timeouts and conflicts (which can mean work is in progress). + if (pending && response.status >= 400 && response.status < 500 && ![408, 409].includes(response.status)) { + await recovery!.clear(pending); + } const remaining = this.token ? parseCreditsRemainingHeader(response.headers) : undefined; const requestId = response.headers["x-request-id"] ?? this.requestId; if (response.status >= 400) { diff --git a/src/commands.ts b/src/commands.ts index 4b233ea8..7a9c8ad2 100644 --- a/src/commands.ts +++ b/src/commands.ts @@ -1,3 +1,4 @@ +import { WriteRecovery } from "./write-recovery.js"; import { createHash } from "node:crypto"; import { open, readFile, rename, rm, stat, writeFile } from "node:fs/promises"; import path from "node:path"; @@ -318,6 +319,8 @@ function transportFor(runtime: CliRuntime, apiUrl: string, token?: string): Tran return loggingTransport(base, loggerOf(runtime)); } +const writeRecoveries = new WeakMap(); + function clientFor(runtime: CliRuntime, args: ParsedArgs, apiUrl: string, token?: string): ApiClient { return new ApiClient({ transport: transportFor(runtime, apiUrl, token), @@ -325,6 +328,7 @@ function clientFor(runtime: CliRuntime, args: ParsedArgs, apiUrl: string, token? requestId: flagString(args.flags, "request-id"), idempotencyKey: flagString(args.flags, "idempotency-key"), timeoutMs: flagNumber(args.flags, "timeout"), + writeRecovery: token ? writeRecoveries.get(runtime) : undefined, creditsOwner: runtime, logger: loggerOf(runtime), }); @@ -391,7 +395,33 @@ function commandHandler( : "Create the first agent with unverified contact metadata, then retry the original command.", }); } - return handler(args, runtime, resolved); + // Specialized enrollment, generation, upload/bundle, and handoff flows own + // their recovery. Ordinary mutations share the durable request ledger. + const [group, action] = args.command; + const ordinary = authenticated && resolved.token && ( + ["kv", "comment", "feedback", "operations"].includes(group ?? "") || + (group === "app" && ["upload", "update"].includes(action ?? "")) || + (group === "playlist" && ["create", "update", "delete"].includes(action ?? "")) || + (group === "media" && ["update", "delete"].includes(action ?? "")) || + (group === "screen" && action !== "provision") + ); + const recovery = ordinary ? new WriteRecovery(resolved, runtime) : undefined; + if (recovery) writeRecoveries.set(runtime, recovery); + try { + const result = await handler(args, runtime, resolved); + await recovery?.finish(); + return result; + } catch (error) { + if (recovery?.hasPending && error instanceof CliError) { + throw new CliError(error.problem, error.exitCode, [...error.warnings, { + code: "write_recovery_saved", + message: "The write key is saved locally. After an ambiguous failure, rerun the same command with unchanged input to reuse it. Reconcile explicit refusals or revision conflicts before changing the request.", + }]); + } + throw error; + } finally { + writeRecoveries.delete(runtime); + } }; } @@ -798,6 +828,7 @@ async function waitForAgentConnectionApproval( resolved: Awaited>, connection: AgentConnectionConfig, timeoutMs: number, + returnPending = false, ): Promise { if (!connection.connection_id || !connection.connection_token) throw configError("Pending agent connection authority is incomplete."); const transport = transportFor(runtime, resolved.apiUrl); @@ -834,7 +865,7 @@ async function waitForAgentConnectionApproval( throw configError("Agent connection SSE emitted invalid JSON."); } latest = validateAgentConnectionEvent(decoded, connection.connection_id); - if (latest.status !== "pending") return latest; + if (returnPending || latest.status !== "pending") return latest; } } } catch (err) { @@ -1014,26 +1045,30 @@ async function agentConnect( let pendingToken = current?.token; let pendingAgentId = connection.pending_agent_id; + const noWait = flagBool(args.flags, "no-wait"); let opened = false; let printed = false; if (!(pendingToken && pendingAgentId)) { if (!connection.approval_url || !connection.connection_id || !connection.connection_token) { throw configError("Pending agent connection is incomplete."); } - if (flagBool(args.flags, "print-url")) { - emitAgentApprovalUrl(args, runtime, connection.approval_url); - printed = true; - } else { - opened = await (runtime.openUrl?.(connection.approval_url) ?? Promise.resolve(false)); - if (!opened) { - emitAgentApprovalUrl(args, runtime, connection.approval_url); + const handoff = async () => { + if (flagBool(args.flags, "print-url")) { + if (!noWait) emitAgentApprovalUrl(args, runtime, connection.approval_url!); printed = true; + } else { + opened = await (runtime.openUrl?.(connection.approval_url!) ?? Promise.resolve(false)); + if (!opened) { + if (!noWait) emitAgentApprovalUrl(args, runtime, connection.approval_url!); + printed = true; + } } - } - const timeoutMs = requestedTimeout ?? 86_400_000; + }; + if (!noWait) await handoff(); + const timeoutMs = requestedTimeout ?? (noWait ? 30_000 : 86_400_000); let status: AgentConnection; try { - status = await waitForAgentConnectionApproval(args, runtime, resolved, connection, timeoutMs); + status = await waitForAgentConnectionApproval(args, runtime, resolved, connection, timeoutMs, noWait); } catch (err) { if (err instanceof CliError && ["agent_connection_cancelled", "agent_connection_expired", "agent_connection_invalid"].includes(err.problem.code)) { return clearDefinitivePendingAgentFailure( @@ -1046,6 +1081,22 @@ async function agentConnect( } throw err; } + if (status.status === "pending") { + await handoff(); + const next = { + command: "screenrig agent connect --no-wait", + argv: ["agent", "connect", "--no-wait", "--config", resolved.configPath, "--api-url", resolved.apiUrl], + reason: "Complete dashboard approval, then resume. Use argv to preserve this configuration and API origin.", + }; + return { + envelope: successEnvelope({ status: "pending", connection_id: connection.connection_id, + expires_at: status.expires_at, opened, approval_url_printed: printed, + ...(printed ? { approval_url: connection.approval_url } : {}), next }), + exitCode: ExitCode.Success, + human: humanLines("Agent approval pending", [["status", "pending"], ["connection_id", connection.connection_id], + ["approval_url", printed ? connection.approval_url : undefined], ["next", next.command]]), + }; + } if (status.status === "denied" || status.status === "expired" || status.status === "cancelled") { await clearAgentConnection(runtime, resolved, connection.connection_id, true); const code = status.status === "denied" diff --git a/src/config.ts b/src/config.ts index c7370f72..2d5e7664 100644 --- a/src/config.ts +++ b/src/config.ts @@ -55,6 +55,8 @@ import { chmod, mkdir, open, rename, rm, stat } from "node:fs/promises"; idempotency_key: string; request_hash: string; }; + /** Pending ordinary writes: hashes and keys only, never payloads. */ + pending_writes?: Record; updated_at?: string; } diff --git a/src/write-recovery.test.ts b/src/write-recovery.test.ts new file mode 100644 index 00000000..22ea908f --- /dev/null +++ b/src/write-recovery.test.ts @@ -0,0 +1,131 @@ +import assert from "node:assert/strict"; +import { mkdtemp, readFile, rm, writeFile, stat } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import path from "node:path"; +import { Writable } from "node:stream"; +import { test } from "node:test"; +import { run, processRuntime } from "./main.js"; +import { FakeTransport } from "./transport/fake.js"; +import { networkError } from "./problems.js"; + +const command = ["screen", "update", "scr_TEST", "--name", "Private lobby name", "--if-match", "1"]; +async function fixture(t: { after(fn: () => Promise): void }) { + const directory = await mkdtemp(path.join(tmpdir(), "screenrig-write-recovery-")); + t.after(() => rm(directory, { recursive: true, force: true })); + const config = `${directory}/config.json`; + await writeFile(config, JSON.stringify({ api_url: "https://api.screenrig.ai", token: "test-only-credential" }), { mode: 0o600 }); + const read = async () => JSON.parse(await readFile(config, "utf8")); + async function invoke(transport: FakeTransport, args = command, now = "2026-09-10T12:00:00Z") { + let stdout = "", stderr = ""; + const code = await run({ ...processRuntime(), argv: ["--config", config, ...args], env: {}, transport, + now: () => new Date(now), + stdout: new Writable({ write(chunk, _encoding, done) { stdout += chunk; done(); } }), + stderr: new Writable({ write(chunk, _encoding, done) { stderr += chunk; done(); } }), + }); + return { code, result: JSON.parse(stdout), stderr }; + } + return { directory, config, read, invoke }; +} +const failed = () => new FakeTransport().on("PATCH", "/api/v1/screens/scr_TEST", () => { throw networkError("Response lost"); }); +const success = () => new FakeTransport().on("PATCH", "/api/v1/screens/scr_TEST", () => ({ status: 200, headers: {}, body: { id: "scr_TEST", revision: 2 } })); +const key = (transport: FakeTransport) => transport.calls[0]?.headers?.["idempotency-key"]; + +test("ordinary writes persist before sending, replay after response loss, and clear after success", async (t) => { + const f = await fixture(t); + const first = new FakeTransport().on("PATCH", "/api/v1/screens/scr_TEST", async request => { + const entries = Object.values((await f.read()).pending_writes) as Array<{ idempotency_key: string }>; + assert.equal(entries[0]?.idempotency_key, request.headers?.["idempotency-key"]); + throw networkError("Response lost"); + }); + assert.notEqual((await f.invoke(first)).code, 0); + const pending = JSON.stringify((await f.read()).pending_writes); + assert.doesNotMatch(pending, /Private lobby name|test-only-credential|scr_TEST/); + if (process.platform !== "win32") assert.equal((await stat(f.config)).mode & 0o777, 0o600); + const second = success(); + assert.equal((await f.invoke(second)).code, 0); + assert.equal(key(second), key(first)); + assert.equal((await f.read()).pending_writes, undefined); + const third = success(); + assert.equal((await f.invoke(third)).code, 0); + assert.notEqual(key(third), key(first)); +}); + +test("changed body, revision, origin and credentials do not reuse pending write keys", async (t) => { + const f = await fixture(t); + const transports: FakeTransport[] = []; + for (const args of [command, command.map(x => x === "Private lobby name" ? "Other lobby" : x), command.map(x => x === "1" ? "2" : x), [...command, "--api-url", "https://example.com"]]) { + const transport = failed(); transports.push(transport); + await f.invoke(transport, args); + } + await writeFile(f.config, JSON.stringify({ ...(await f.read()), token: "changed-test-credential" }), { mode: 0o600 }); + const last = failed(); transports.push(last); await f.invoke(last); + assert.equal(new Set(transports.map(key)).size, transports.length); +}); + +test("definitive precondition refusal clears the pending key without retrying", async (t) => { + const f = await fixture(t); + const refused = new FakeTransport().on("PATCH", "/api/v1/screens/scr_TEST", () => ({ status: 412, headers: {}, body: { code: "revision_conflict", detail: "Refetch", status: 412 } })); + assert.notEqual((await f.invoke(refused)).code, 0); + assert.equal(refused.calls.length, 1); + assert.equal((await f.read()).pending_writes, undefined); +}); + +test("expired pending writes stop before network and require explicit reconciliation", async (t) => { + const f = await fixture(t); const first = failed(); await f.invoke(first); + const blocked = success(); + const result = await f.invoke(blocked, command, "2026-09-11T12:00:00Z"); + assert.equal(result.code, 2); + assert.match(result.result.error.detail, /safe replay window/); + assert.equal(blocked.calls.length, 0); + assert.equal((await f.invoke(blocked, [...command, "--idempotency-key", "explicit-reconciled-write"], "2026-09-11T12:00:00Z")).code, 0); + assert.equal(key(blocked), "explicit-reconciled-write"); +}); + +test("invalid recovery state fails closed before mutation", async (t) => { + const f = await fixture(t); + await writeFile(f.config, JSON.stringify({ ...(await f.read()), pending_writes: { broken: {} } }), { mode: 0o600 }); + const transport = success(); + assert.equal((await f.invoke(transport)).code, 12); + assert.equal(transport.calls.length, 0); +}); + +test("concurrent identical unresolved invocations use one persisted key", async (t) => { + const f = await fixture(t); + const first = failed(); const second = failed(); + await Promise.all([f.invoke(first), f.invoke(second)]); + assert.ok(key(first)); + assert.equal(key(first), key(second)); + assert.equal(Object.keys((await f.read()).pending_writes).length, 1); +}); + +test("server errors retain the saved key and expose recovery guidance", async (t) => { + const f = await fixture(t); + const first = new FakeTransport().on("PATCH", "/api/v1/screens/scr_TEST", () => ({ status: 503, headers: {}, body: { detail: "Unavailable", status: 503 } })); + const result = await f.invoke(first); + assert.notEqual(result.code, 0); + assert.equal(result.result.warnings[0].code, "write_recovery_saved"); + const second = success(); await f.invoke(second); + assert.equal(key(first), key(second)); +}); + +test("application acceptance followed by lost operation status replays the same upload", async (t) => { + const f = await fixture(t); + const { mkdir } = await import("node:fs/promises"); + const { memoryBackend } = await import("./transport/fake.js"); + const app = `${f.directory}/app`; await mkdir(app); + await writeFile(`${app}/index.html`, "hello"); + const transport = () => { + const fake = new FakeTransport(); + const backend = memoryBackend(); + fake.on("GET", "/api/v1/capabilities", req => backend.request(req)); + fake.on("POST", "/api/v1/applications", req => backend.request(req)); + fake.on("GET", /\/api\/v1\/operations\//, () => { throw networkError("Status unavailable"); }); + return fake; + }; + const first = transport(); const second = transport(); + assert.notEqual((await f.invoke(first, ["app", "upload", app])).code, 0); + assert.notEqual((await f.invoke(second, ["app", "upload", app])).code, 0); + const upload = (fake: FakeTransport) => fake.calls.find(req => req.method === "POST"); + assert.ok(upload(first)?.headers?.["idempotency-key"]); + assert.equal(upload(first)?.headers?.["idempotency-key"], upload(second)?.headers?.["idempotency-key"]); +}); diff --git a/src/write-recovery.ts b/src/write-recovery.ts new file mode 100644 index 00000000..72989ac3 --- /dev/null +++ b/src/write-recovery.ts @@ -0,0 +1,94 @@ +import { createHash } from "node:crypto"; +import { readConfigFile, withConfigLock, writeConfigAtomic, type ResolvedConfig, type ScreenRigConfig } from "./config.js"; +import { isValidIdempotencyKey, newIdempotencyKey } from "./ids.js"; +import { configError, usageError } from "./problems.js"; +import type { CliRuntime } from "./runtime.js"; +import type { TransportRequest } from "./transport/types.js"; + +interface PendingWrite { fingerprint: string; key: string } +type Ledger = NonNullable; +// Server replay records last 24 hours. Stop earlier rather than silently +// replaying a mutation after the server may have forgotten its key. +const SAFE_REPLAY_MS = 23 * 60 * 60 * 1000; + +function ledger(config: ScreenRigConfig): Ledger { + const value = config.pending_writes === undefined ? {} : config.pending_writes; + if (!value || typeof value !== "object" || Array.isArray(value) || + Object.entries(value).some(([hash, entry]) => !/^[a-f0-9]{64}$/.test(hash) || + !entry || typeof entry.idempotency_key !== "string" || !isValidIdempotencyKey(entry.idempotency_key) || + typeof entry.created_at !== "string" || !Number.isFinite(Date.parse(entry.created_at)))) { + throw configError("Pending write recovery state is invalid; no request was sent."); + } + return value; +} + +/** Per-invocation coordinator. Only hashes, keys and timestamps reach disk. */ +export class WriteRecovery { + private readonly touched = new Map(); + constructor(private readonly resolved: ResolvedConfig, private readonly runtime: CliRuntime) {} + + private async update(work: (config: ScreenRigConfig, pending: Ledger) => T): Promise { + const fs = { ...this.runtime.fs, env: this.runtime.env, homedir: this.runtime.homedir }; + return withConfigLock(this.resolved.configPath, fs, + { sleep: this.runtime.sleep, now: () => this.runtime.now().getTime() }, async () => { + const config = await readConfigFile(this.resolved.configPath, fs); + if (!config || config.token !== this.resolved.token) { + throw configError("Agent credential changed during write recovery; retry with the current installation."); + } + const pending = ledger(config); + const result = work(config, pending); + const { pending_writes: _old, ...rest } = config; + await writeConfigAtomic(this.resolved.configPath, + { ...rest, ...(Object.keys(pending).length ? { pending_writes: pending } : {}) }, fs); + return result; + }); + } + + async prepare(request: Omit & { headers?: Record }, requestedKey?: string): Promise { + const headers = Object.entries(request.headers ?? {}) + .filter(([name]) => !["x-request-id", "idempotency-key"].includes(name.toLowerCase())) + .map(([name, value]) => [name.toLowerCase(), value]).sort(([a], [b]) => a!.localeCompare(b!)); + const body = request.body === undefined ? undefined : request.body instanceof Uint8Array + ? request.body : typeof request.body === "string" ? request.body : JSON.stringify(request.body); + const hash = createHash("sha256").update(JSON.stringify([ + this.resolved.apiUrl, this.resolved.token, request.method, request.path, + Object.entries(request.query ?? {}).sort(), headers, body === undefined ? "absent" : "present", + ])); + if (body !== undefined) hash.update(body); + const fingerprint = hash.digest("hex"); + const pending = await this.update((_config, entries) => { + const existing = entries[fingerprint]; + const reuse = existing && (!requestedKey || requestedKey === existing.idempotency_key); + if (reuse && this.runtime.now().getTime() - Date.parse(existing.created_at) >= SAFE_REPLAY_MS) { + throw usageError("This unresolved write is older than the safe replay window. Inspect the resource before explicitly supplying a new --idempotency-key for a reconciled write."); + } + if (!existing && Object.keys(entries).length >= 256) { + throw configError("Too many unresolved writes. Reconcile pending writes before starting another mutation."); + } + const key = reuse ? existing.idempotency_key : requestedKey ?? newIdempotencyKey(); + entries[fingerprint] = reuse ? existing : { idempotency_key: key, created_at: this.runtime.now().toISOString() }; + return { fingerprint, key }; + }); + this.touched.set(fingerprint, pending); + return pending; + } + + async clear(pending: PendingWrite): Promise { + await this.update((_config, entries) => { + if (entries[pending.fingerprint]?.idempotency_key === pending.key) delete entries[pending.fingerprint]; + }); + this.touched.delete(pending.fingerprint); + } + + get hasPending(): boolean { return this.touched.size > 0; } + + async finish(): Promise { + if (!this.touched.size) return; + await this.update((_config, entries) => { + for (const pending of this.touched.values()) { + if (entries[pending.fingerprint]?.idempotency_key === pending.key) delete entries[pending.fingerprint]; + } + }); + this.touched.clear(); + } +}