From 1ff76d595451f3316b1d84a2351bdb1d7dc1af8d Mon Sep 17 00:00:00 2001 From: Mike Schwarz <171890339+mvschwarz@users.noreply.github.com> Date: Fri, 6 Nov 2026 13:22:01 -0800 Subject: [PATCH] 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 ; without --id, check with rig stream list --source --since , 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 | 26 ++++++++++- packages/cli/test/stream.test.ts | 43 +++++++++++++++++++ .../openrig-core/skills/openrig-user/SKILL.md | 4 ++ scripts/skill-edge-digests.generated.json | 2 +- 4 files changed, 73 insertions(+), 2 deletions(-) diff --git a/packages/cli/src/commands/stream.ts b/packages/cli/src/commands/stream.ts index 2f5c5b0c7..6bc0822cd 100644 --- a/packages/cli/src/commands/stream.ts +++ b/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; @@ -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>("/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); }); diff --git a/packages/cli/test/stream.test.ts b/packages/cli/test/stream.test.ts index ed7ad8809..49aca02fb 100644 --- a/packages/cli/test/stream.test.ts +++ b/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, + }; + 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: [] } }, diff --git a/packages/daemon/assets/plugins/openrig-core/skills/openrig-user/SKILL.md b/packages/daemon/assets/plugins/openrig-core/skills/openrig-user/SKILL.md index 145b24c66..d5729c65e 100644 --- a/packages/daemon/assets/plugins/openrig-core/skills/openrig-user/SKILL.md +++ b/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 ` before emitting again. + Read back (`rig queue show --full --json`, or `rig queue transitions `) when the answer doesn't show what you need, for example: diff --git a/scripts/skill-edge-digests.generated.json b/scripts/skill-edge-digests.generated.json index 386af82ed..15807250a 100644 --- a/scripts/skill-edge-digests.generated.json +++ b/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",