Persist write recovery and add nonblocking agent connections

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`.

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 <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));

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<typeof agentConnectionEnvelope> | 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");

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) {

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,13 +319,16 @@ function transportFor(runtime: CliRuntime, apiUrl: string, token?: string): Tran
   return loggingTransport(base, loggerOf(runtime));
 }
 
+const writeRecoveries = new WeakMap<CliRuntime, WriteRecovery>();
+
 function clientFor(runtime: CliRuntime, args: ParsedArgs, apiUrl: string, token?: string): ApiClient {
   return new ApiClient({
     transport: transportFor(runtime, apiUrl, token),
     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<ReturnType<typeof resolveConfig>>,
   connection: AgentConnectionConfig,
   timeoutMs: number,
+  returnPending = false,
 ): Promise<AgentConnection> {
   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"

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<string, { idempotency_key: string; created_at: string }>;
       updated_at?: string;
     }