fix(cli): name the safe retry when rig stream emit times out (#2619)

A timed-out or unreadable rig stream emit can still land late. Its error now names the retry that can't add a second item: with --id, repeat the same emit with that --id (the daemon keeps one item per stream item ID) or check first with rig stream show <id>; without --id, check with rig stream list --source <recorded seat> --since <five minutes before the send>, using the seat the daemon records, and pass a stable --id on later emits. queue create is unchanged; its existing --id already makes a retry safe, and the openrig-user skill now says so for both in one paragraph (skill digest updated). No new refusal, error code or setting.

packages/cli/src/commands/stream.ts

@@ -1,6 +1,8 @@
 import { Command } from "commander";
-import { DaemonClient } from "../client.js";
+import { DaemonClient, DaemonResponseError, DaemonTimeoutError } from "../client.js";
 import { getDaemonStatus, getDaemonUrl , daemonStatusGuard} from "../daemon-lifecycle.js";
+import { readOpenRigEnv } from "../openrig-compat.js";
+import { shellQuote } from "../cross-host-executor.js";
 import { realDeps } from "./daemon.js";
 import type { StatusDeps } from "./status.js";
 import { streamOutcome } from "./stream-outcome.js";
@@ -42,6 +44,21 @@ function printResult(json: boolean, body: unknown, status: number): void {
   if (status >= 400) process.exitCode = status >= 500 ? 2 : 1;
 }
 
+/**
+ * Added to an emit whose reply came too late or unreadable: the item may still have landed, and the
+ * daemon keeps one item per stream item ID, so only a retry with the same ID can't add a second one.
+ */
+function emitRetryGuidance(id: string | undefined, source: string, sentAt: number): string {
+  if (id !== undefined) {
+    return `The item may still land. To retry safely, repeat the same emit with --id ${shellQuote(id)}; the daemon keeps one item per ID. To check first: rig stream show ${shellQuote(id)}.`;
+  }
+  // The daemon records the seat the request came from (its seat env), not --source, when the two differ.
+  const recorded = readOpenRigEnv("OPENRIG_SESSION_NAME", "RIGGED_SESSION_NAME")?.trim() || source;
+  // Five minutes before the send, for a daemon whose clock runs behind this one.
+  const since = new Date(sentAt - 5 * 60_000).toISOString();
+  return `The item may still land, and this emit had no --id, so emitting again can add a second item. Check first: rig stream list --source ${shellQuote(recorded)} --since ${since}. Pass a stable --id on emits so a later retry is safe.`;
+}
+
 function isWatchedStreamItem(value: unknown): value is WatchedStreamItem {
   if (!value || typeof value !== "object") return false;
   const item = value as Record<string, unknown>;
@@ -132,6 +149,7 @@ To keep emits passive: rig config set stream.intake off`)
       const deps = getDeps();
       const tags = opts.hintTags ? opts.hintTags.split(",").map((s) => s.trim()).filter(Boolean) : null;
       await withClient(deps, async (client) => {
+        const sentAt = Date.now();
         const res = await client.post<Record<string, unknown>>("/api/stream/emit", {
           streamItemId: opts.id,
           sourceSession: opts.source,
@@ -142,6 +160,12 @@ To keep emits passive: rig config set stream.intake off`)
           hintUrgency: opts.hintUrgency ?? null,
           hintTags: tags,
           interrupt: opts.interrupt ?? false,
+        }).catch((error: unknown): never => {
+          // The shared renderer reads both as an UNKNOWN outcome; name the retry that can't duplicate.
+          if (error instanceof DaemonTimeoutError || error instanceof DaemonResponseError) {
+            error.message += ` ${emitRetryGuidance(opts.id, opts.source, sentAt)}`;
+          }
+          throw error;
         });
         printResult(opts.json ?? false, res.data, res.status);
       });

packages/cli/test/stream.test.ts

@@ -126,6 +126,49 @@ describe("rig stream CLI", () => {
     expect((call!.body as { streamItemId: string }).streamItemId).toBe("stream-fixed-id");
   });
 
+  it("an emit whose reply is late or unreadable names the retry that can't add a second item", async () => {
+    const { runProgram } = await import("../src/cli-error.js");
+    const { DaemonConnectionError, DaemonResponseError, DaemonTimeoutError } = await import("../src/client.js");
+    const run = async (error: Error, extra: string[]) => {
+      const deps: StreamDeps = {
+        lifecycleDeps: {} as StreamDeps["lifecycleDeps"],
+        clientFactory: () => ({ post: vi.fn(async () => { throw error; }) }) as unknown as ReturnType<StreamDeps["clientFactory"]>,
+      };
+      const lines: string[] = [];
+      let code = 0;
+      await runProgram(createProgram({ streamDeps: deps }),
+        ["node", "rig", "stream", "emit", "--source", "alice@rig", "--body", "x", ...extra],
+        { out: (l) => lines.push(l), err: (l) => lines.push(l), exit: (c) => { code = c; } });
+      return { text: lines.join("\n"), code };
+    };
+    const now = vi.spyOn(Date, "now").mockReturnValue(Date.parse("2026-11-06T20:00:00.000Z"));
+    vi.stubEnv("OPENRIG_SESSION_NAME", "");
+    vi.stubEnv("RIGGED_SESSION_NAME", "");
+    try {
+      const withId = await run(new DaemonTimeoutError("Request to http://localhost:7433/api/stream/emit timed out after 5000ms"), ["--id", "obs-1"]);
+      expect(withId.text).toContain("repeat the same emit with --id 'obs-1'; the daemon keeps one item per ID");
+      expect(withId.text).toContain("rig stream show 'obs-1'");
+      expect(withId.code).toBe(1);
+
+      const withoutId = await run(new DaemonTimeoutError("Request to http://localhost:7433/api/stream/emit timed out after 5000ms"), []);
+      expect(withoutId.text).toContain("this emit had no --id, so emitting again can add a second item");
+      expect(withoutId.text).toContain("rig stream list --source 'alice@rig' --since 2026-11-06T19:55:00.000Z");
+      expect(withoutId.text).toContain("Pass a stable --id on emits so a later retry is safe.");
+
+      // The daemon records the seat env's session, not --source, so the check names that seat.
+      vi.stubEnv("OPENRIG_SESSION_NAME", "bob@rig");
+      const unreadable = await run(new DaemonResponseError(502, ""), []);
+      expect(unreadable.text).toContain("rig stream list --source 'bob@rig' --since 2026-11-06T19:55:00.000Z");
+
+      // A refused connection delivered nothing: no "may still land".
+      const refused = await run(new DaemonConnectionError("connect ECONNREFUSED 127.0.0.1:7433", "ECONNREFUSED"), []);
+      expect(refused.text).not.toContain("may still land");
+    } finally {
+      now.mockRestore();
+      vi.unstubAllEnvs();
+    }
+  });
+
   it("list GETs /api/stream/list with limit + sourceSession query", async () => {
     const { deps, calls } = makeDeps({
       routes: { "GET /api/stream/list?sourceSession=bob%40rig&limit=20": { status: 200, data: [] } },

packages/daemon/assets/plugins/openrig-core/skills/openrig-user/SKILL.md

@@ -101,6 +101,10 @@ when the command exits 0. A non-zero exit means no row was confirmed, even when
 indeterminate answer repeats the request ID so you can reconcile by it. Don't merge stderr into stdout and keep the
 first `qitem-…` you find there; a refused create prints its request ID first, too.
 
+A timed-out write can still land late. To retry a `create` or `rig stream emit` safely, repeat it unchanged with the
+same `--id` (create's request ID, or the emit's own), since the daemon keeps one row or item per ID; an emit sent
+without `--id` has no ID to reuse, so look for it with `rig stream list --source <seat>` before emitting again.
+
 Read back (`rig queue show <id> --full --json`, or `rig queue transitions <id>`) when the answer doesn't show what you
 need, for example:
 

scripts/skill-edge-digests.generated.json

@@ -89,7 +89,7 @@
       "openrig-operating-model/templates/SOP.md": "1eaf0daf4f9bbf0715e6a5aaf0f29f030af5fe8b8a10232488ee97c507afa6dc",
       "openrig-operating-model/templates/SPEC.md": "f126536daa59cea3f1cfb0898d83c7cdf99f6392f6f155ea64fe55187dd0f5f1",
       "openrig-skills/SKILL.md": "bdd6594ae1108455103239d39694c865cc40d3f7d9f24cd3566a18397df270da",
-      "openrig-user/SKILL.md": "01cb169050a3d4f51ad7b263772c6ed58f62683df820efeac1a88faadf9cd086",
+      "openrig-user/SKILL.md": "0776fe4e197d347614f72eec817f881debfcae372c4a1f912f2e0fbbf91c67da",
       "orienting-to-an-inherited-seat/CHANGELOG.md": "4d0970e75e5c15b00a48d6a7150376d646257777264fd1b9a191165a8d5a6e84",
       "orienting-to-an-inherited-seat/SKILL.md": "252fed402f19027056bbba241df3305be88de3175b67dc8fe051143bb0d4412b",
       "queue-handoff/SKILL.md": "96251f6830006bb9d8ede2e5826ab936df4a9fb91090187a23ccaf860ab76336",