From 968ffa3ecf2ed095be1e004fce796ba36e7e217f Mon Sep 17 00:00:00 2001 From: Codex Date: Wed, 15 Jul 2026 05:47:52 +0200 Subject: [PATCH] feat(agentrun): add events long polling --- .agents/skills/unidesk-code-queue/SKILL.md | 9 +- .../references/resources.md | 9 +- scripts/src/agentrun-events-long-poll.test.ts | 185 ++++++++++++++++++ scripts/src/agentrun/config.ts | 1 + scripts/src/agentrun/entry.ts | 9 +- scripts/src/agentrun/render.ts | 11 ++ scripts/src/agentrun/resource-actions.ts | 74 ++++++- scripts/src/agentrun/rest-bridge.ts | 24 ++- scripts/src/agentrun/utils.ts | 2 + 9 files changed, 307 insertions(+), 17 deletions(-) create mode 100644 scripts/src/agentrun-events-long-poll.test.ts diff --git a/.agents/skills/unidesk-code-queue/SKILL.md b/.agents/skills/unidesk-code-queue/SKILL.md index fcf551fb..fdfd19aa 100644 --- a/.agents/skills/unidesk-code-queue/SKILL.md +++ b/.agents/skills/unidesk-code-queue/SKILL.md @@ -14,7 +14,7 @@ bun scripts/cli.ts agentrun get tasks --limit 20 bun scripts/cli.ts agentrun describe task/ bun scripts/cli.ts agentrun describe task/ --input -o json bun scripts/cli.ts agentrun describe aipodspec/ -bun scripts/cli.ts agentrun events run/ --limit 20 +bun scripts/cli.ts agentrun events run/ --after-seq 0 --expect 10 --timeout 120s --limit 20 bun scripts/cli.ts agentrun events run/ --detail-seq -o json bun scripts/cli.ts agentrun logs session/ --tail 120 bun scripts/cli.ts agentrun result run/ --command @@ -53,6 +53,13 @@ AipodSpec/Artificer、task manifest 和 queue 渐进披露见 [references/resour - `unidesk-trans` required skill、runner tools bundle、`unidesk-ssh` SecretRef 和 endpoint presence 缺一项即归为装配失败； - Secret 输出只显示对象、key、presence/fingerprint 与 `valuesPrinted=false`。 - 默认输出保持低噪声表格/摘要；机器消费显式 `-o json|yaml` 或 `--raw`。 +- 主代理观察 active run 时默认使用单次长轮询： + - `--after-seq` 是可继续拉取的增量游标； + - `--expect ` 是等待条件； + - `--timeout ` 默认 `120s`，达到 `expect` 或目标 run/command 终态时提前返回； + - timeout 到期返回当前事件，即使 0 条也明确披露 `timedOut`、`observed` 和 `nextAfterSeq`； + - `--limit` 只控制输出预算，不得承担等待条件； + - CLI 只发起一个 manager-held 请求，禁止本地高频短轮询或忙轮询。 - 读取 task 的可重建输入： - 使用 `agentrun describe task/ --input -o json`； - 该入口只投影 task create 字段、AipodSpec identity、资源引用和脱敏 provenance； diff --git a/.agents/skills/unidesk-code-queue/references/resources.md b/.agents/skills/unidesk-code-queue/references/resources.md index 514937ea..3b6c0f99 100644 --- a/.agents/skills/unidesk-code-queue/references/resources.md +++ b/.agents/skills/unidesk-code-queue/references/resources.md @@ -62,6 +62,13 @@ Artificer 内部执行边界： `agentrun events` 是高频渐进披露入口： +- active run 的推荐观察入口是： + - `bun scripts/cli.ts agentrun events run/ --after-seq --expect 10 --timeout 120s --limit 20`； + - manager 在单个请求内等待 after-seq 后新增事件达到 `expect`，或目标 run/command 进入终态； + - 未达到条件时最多等待 `timeout`，默认 `120s`； + - timeout 的部分返回和空返回都必须明确披露 `timedOut`、`observed` 与 `nextAfterSeq`； + - 后续继续使用返回的 `nextAfterSeq`，不要恢复高频短轮询； + - `limit` 是独立输出预算，不能替代 `expect`； - 默认 human、`-o json` 和 `-o yaml` 最多返回 20 条同构事件摘要，并用一条 lookahead 披露 `hasMore` 与 `nextAfterSeq`； - tool call 只展示 identity、seq/type/status、commandId、工具、命令首行、exit/duration、output bytes/hash、truncation 与 `outputOmitted`； - 默认不得展开 stdout、stderr、skill 正文或完整 payload，也不得通过提高全局 stdout 阈值掩盖 dump； @@ -72,7 +79,7 @@ Artificer 内部执行边界： - durable `resolutionAuthority` 保持原事实或显式脱敏，基于其他字段的非权威判断只进入独立 inference 字段； - 标准 Git URL 移除 userinfo、query 和 fragment，SCP-like 与未知 remote 只披露不可逆 fingerprint； - credential value 和 URL credential 固定脱敏，`valuesPrinted=false`； -- `--detail-seq` 与 `--after-seq`、`--limit`、`--full`、`--raw` 互斥，并在请求 manager 前 fail closed； +- `--detail-seq` 与 `--after-seq`、`--expect`、`--timeout`、`--limit`、`--full`、`--raw` 互斥，并在请求 manager 前 fail closed； - 完整事件继续使用显式 `--full`、`--raw`，或按 `afterSeq=item.seq-1 --limit 1 --full` 下钻； - 该 CLI 只渲染 AgentRun durable facts，不删除事件、不按正文去重，也不改变 Kafka/SSE 事件链。 diff --git a/scripts/src/agentrun-events-long-poll.test.ts b/scripts/src/agentrun-events-long-poll.test.ts new file mode 100644 index 00000000..9b421f2f --- /dev/null +++ b/scripts/src/agentrun-events-long-poll.test.ts @@ -0,0 +1,185 @@ +import { rmSync } from "node:fs"; +import { describe, expect, test } from "bun:test"; + +import { runAgentRunCommand } from "./agentrun"; + +const RUN_ID = "run_long_poll_fixture"; +const agentRunClientYaml = [ + "version: 1", + "kind: AgentRunConfig", + "metadata:", + " name: agentrun", + "manager:", + " baseUrl: __BASE_URL__", + " timeoutMs: 15000", + "auth:", + " env: HWLAB_API_KEY", + " file: /tmp/hwlab-api-key.env", + " header: Authorization", + " scheme: Bearer", + "client:", + " role: render-only", + " transport: direct-http", + " sessionPolicy:", + " tenantId: unidesk", + " projectId: test", + " providerId: NC01", + " backendProfile: codex", + " workspaceRef:", + " kind: opaque", + " executionPolicy:", + " sandbox: workspace-write", + " approval: never", + " timeoutMs: 7200000", + " network: enabled", + " secretScope:", + " allowCredentialEcho: false", +].join("\n"); + +function renderedText(value: Awaited>): string { + if (!("renderedText" in value) || typeof value.renderedText !== "string") throw new Error("expected rendered AgentRun CLI output"); + return value.renderedText; +} + +function event(seq: number): Record { + return { + id: `evt_${seq}`, + seq, + type: "backend_status", + payload: { commandId: "cmd_fixture", phase: `phase-${seq}`, message: `event-${seq}` }, + }; +} + +async function withManager( + handler: (request: Request) => Response | Promise, + action: () => Promise, +): Promise { + const server = Bun.serve({ port: 0, fetch: handler }); + const previousConfig = process.env.AGENTRUN_CLIENT_CONFIG; + const previousKey = process.env.HWLAB_API_KEY; + const configPath = `/tmp/unidesk-agentrun-events-long-poll-${process.pid}-${Date.now()}.yaml`; + process.env.AGENTRUN_CLIENT_CONFIG = configPath; + process.env.HWLAB_API_KEY = "fixture-secret"; + await Bun.write(configPath, agentRunClientYaml.replace("__BASE_URL__", server.url.href.replace(/\/$/u, ""))); + try { + await action(); + } finally { + server.stop(true); + rmSync(configPath, { force: true }); + if (previousConfig === undefined) delete process.env.AGENTRUN_CLIENT_CONFIG; + else process.env.AGENTRUN_CLIENT_CONFIG = previousConfig; + if (previousKey === undefined) delete process.env.HWLAB_API_KEY; + else process.env.HWLAB_API_KEY = previousKey; + } +} + +describe.serial("AgentRun events manager-held long polling", () => { + test("forwards default 120s timeout and returns as soon as expect is reached", async () => { + let requests = 0; + await withManager(async (request) => { + requests += 1; + const url = new URL(request.url); + expect(url.pathname).toBe(`/api/v1/runs/${RUN_ID}/events`); + expect(url.searchParams.get("afterSeq")).toBe("100"); + expect(url.searchParams.get("expect")).toBe("2"); + expect(url.searchParams.get("timeoutMs")).toBe("120000"); + expect(url.searchParams.get("limit")).toBe("21"); + await Promise.all([Bun.sleep(10), Bun.sleep(20)]); + return Response.json({ + ok: true, + data: { + items: [event(101), event(102)], + timedOut: false, + observed: 2, + nextAfterSeq: 102, + completionReason: "expect-reached", + }, + }); + }, async () => { + const result = await runAgentRunCommand(null, ["events", `run/${RUN_ID}`, "--after-seq", "100", "--expect", "2", "-o", "json"]); + expect(result.ok).toBe(true); + const payload = JSON.parse(renderedText(result)) as Record; + expect(payload).toMatchObject({ count: 2, expect: 2, timeoutMs: 120000, timedOut: false, observed: 2, nextAfterSeq: 102, completionReason: "expect-reached" }); + expect(requests).toBe(1); + }); + }); + + test("renders partial and empty timeout responses with an explicit cursor", async () => { + for (const fixture of [ + { items: [event(201)], observed: 1, nextAfterSeq: 201 }, + { items: [], observed: 0, nextAfterSeq: 200 }, + ]) { + await withManager((request) => { + const url = new URL(request.url); + expect(url.searchParams.get("timeoutMs")).toBe("3000"); + return Response.json({ ok: true, data: { ...fixture, timedOut: true, completionReason: "timeout" } }); + }, async () => { + const result = await runAgentRunCommand(null, ["events", `run/${RUN_ID}`, "--after-seq", "200", "--expect", "10", "--timeout", "3s", "-o", "json"]); + const payload = JSON.parse(renderedText(result)) as Record; + expect(payload).toMatchObject({ timedOut: true, observed: fixture.observed, nextAfterSeq: fixture.nextAfterSeq, completionReason: "timeout" }); + }); + } + }); + + test("returns concurrent event arrivals in one manager request", async () => { + let requests = 0; + await withManager(async () => { + requests += 1; + const arrivals: Record[] = []; + await Promise.all([ + Bun.sleep(10).then(() => arrivals.push(event(301))), + Bun.sleep(20).then(() => arrivals.push(event(302))), + ]); + return Response.json({ ok: true, data: { items: arrivals, timedOut: false, observed: 2, nextAfterSeq: 302, completionReason: "expect-reached" } }); + }, async () => { + const result = await runAgentRunCommand(null, ["events", `run/${RUN_ID}`, "--after-seq", "300", "--expect", "2", "--timeout", "1m", "-o", "json"]); + const payload = JSON.parse(renderedText(result)) as { items?: Array<{ seq?: number }> }; + expect(payload.items?.map((item) => item.seq)).toEqual([301, 302]); + expect(requests).toBe(1); + }); + }); + + test("returns immediately when the target is terminal before expect is reached", async () => { + await withManager(() => Response.json({ + ok: true, + data: { + items: [], + timedOut: false, + observed: 0, + nextAfterSeq: 400, + completionReason: "target-completed", + }, + }), async () => { + const result = await runAgentRunCommand(null, ["events", `run/${RUN_ID}`, "--after-seq", "400", "--expect", "10", "--timeout", "120s", "-o", "json"]); + const payload = JSON.parse(renderedText(result)) as Record; + expect(payload).toMatchObject({ timedOut: false, observed: 0, nextAfterSeq: 400, completionReason: "target-completed" }); + }); + }); + + test("rejects invalid duration before request and preserves manager errors", async () => { + let requests = 0; + await withManager(() => { + requests += 1; + return Response.json({ ok: false, failureKind: "manager-unavailable", message: "event notifier unavailable" }, { status: 503 }); + }, async () => { + const invalid = await runAgentRunCommand(null, ["events", `run/${RUN_ID}`, "--expect", "2", "--timeout", "3minutes", "-o", "json"]); + expect(invalid.ok).toBe(false); + expect(renderedText(invalid)).toContain("--timeout must use an integer duration"); + expect(requests).toBe(0); + + const failed = await runAgentRunCommand(null, ["events", `run/${RUN_ID}`, "--expect", "2", "--timeout", "1s", "-o", "json"]); + expect(failed.ok).toBe(false); + expect(renderedText(failed)).toContain("event notifier unavailable"); + expect(requests).toBe(1); + }); + }); + + test("fails visibly when a manager ignores the long-poll response contract", async () => { + await withManager(() => Response.json({ ok: true, data: { items: [], nextAfterSeq: 500 } }), async () => { + const result = await runAgentRunCommand(null, ["events", `run/${RUN_ID}`, "--after-seq", "500", "--expect", "2", "--timeout", "1s", "-o", "json"]); + expect(result.ok).toBe(false); + expect(renderedText(result)).toContain("schema-mismatch"); + expect(renderedText(result)).toContain("must include boolean timedOut and non-negative observed fields"); + }); + }); +}); diff --git a/scripts/src/agentrun/config.ts b/scripts/src/agentrun/config.ts index 8ca1c052..6c1a436c 100644 --- a/scripts/src/agentrun/config.ts +++ b/scripts/src/agentrun/config.ts @@ -371,6 +371,7 @@ export function agentRunQuery(args: string[], names: string[]): string { const params = new URLSearchParams(); const keyMap: Record = { "after-seq": "afterSeq", + "timeout-ms": "timeoutMs", "backend-profile": "backendProfile", "command": "commandId", "command-id": "commandId", diff --git a/scripts/src/agentrun/entry.ts b/scripts/src/agentrun/entry.ts index 3955fcde..71f2f599 100644 --- a/scripts/src/agentrun/entry.ts +++ b/scripts/src/agentrun/entry.ts @@ -59,7 +59,7 @@ export function agentRunHelp(): unknown { "bun scripts/cli.ts agentrun get tasks -o json", "bun scripts/cli.ts agentrun describe task/", "bun scripts/cli.ts agentrun describe task/ --input -o json", - "bun scripts/cli.ts agentrun events run/ --after-seq 0 --limit 100", + "bun scripts/cli.ts agentrun events run/ --after-seq 0 --expect 10 --timeout 120s --limit 100", "bun scripts/cli.ts agentrun events run/ --detail-seq ", "bun scripts/cli.ts agentrun logs session/ --tail 100", "bun scripts/cli.ts agentrun result run/ --command ", @@ -292,9 +292,10 @@ export function agentRunHelpText(args: string[]): string { } if (verb === "events") { return [ - "Usage: bun scripts/cli.ts agentrun events run/ [--after-seq N] [--limit 100] [-o json|yaml] [--full|--raw]", + "Usage: bun scripts/cli.ts agentrun events run/ [--after-seq N] [--expect N [--timeout 120s]] [--limit 100] [-o json|yaml] [--full|--raw]", " bun scripts/cli.ts agentrun events run/ --detail-seq [-o json|yaml]", "", + "--expect waits in one manager-held request until N new events, target completion, or timeout; --timeout defaults to 120s. --limit remains an independent output budget.", "--detail-seq returns one bounded Secret-safe EventDetail projection. Use the separate --after-seq --limit 1 --full|--raw path only for explicit complete disclosure.", ].join("\n"); } @@ -390,7 +391,7 @@ export function agentRunHelpText(args: string[]): string { " bun scripts/cli.ts agentrun get tasks --queue commander --limit 20", " bun scripts/cli.ts agentrun describe task/", " bun scripts/cli.ts agentrun describe run/ --node NC01 --lane nc01-v02", - " bun scripts/cli.ts agentrun events run/ --after-seq 0 --limit 100", + " bun scripts/cli.ts agentrun events run/ --after-seq 0 --expect 10 --timeout 120s --limit 100", " bun scripts/cli.ts agentrun logs session/ --tail 100", "", "Use --raw on a resource command when you need the direct AgentRun REST envelope.", @@ -407,7 +408,7 @@ export function agentRunHelpText(args: string[]): string { " bun scripts/cli.ts agentrun describe task/", " bun scripts/cli.ts agentrun describe task/ --input -o json", " bun scripts/cli.ts agentrun describe run/ --node NC01 --lane nc01-v02", - " bun scripts/cli.ts agentrun events run/ --after-seq 0 --limit 100", + " bun scripts/cli.ts agentrun events run/ --after-seq 0 --expect 10 --timeout 120s --limit 100", " bun scripts/cli.ts agentrun events run/ --detail-seq ", " bun scripts/cli.ts agentrun logs session/ --tail 100", " bun scripts/cli.ts agentrun retry task/ --idempotency-key --reason --dry-run", diff --git a/scripts/src/agentrun/render.ts b/scripts/src/agentrun/render.ts index 452c641c..98a174d9 100644 --- a/scripts/src/agentrun/render.ts +++ b/scripts/src/agentrun/render.ts @@ -360,6 +360,13 @@ export interface EventLikePageDisclosure { hasMore: boolean; nextAfterSeq: number | string | null; detailCommand: string; + wait: { + expect: number; + timeoutMs: number; + timedOut: boolean; + observed: number; + completionReason: string | null; + } | null; } export function renderEventLike(command: string, raw: Record, options: AgentRunResourceOptions, kind: string, items: Record[], sourceName: string, pageDisclosure: EventLikePageDisclosure | null = null): RenderedCliResult { @@ -376,6 +383,7 @@ export function renderEventLike(command: string, raw: Record, o effectiveLimit: pageDisclosure.effectiveLimit, fetchedCount: pageDisclosure.fetchedCount, hasMore: pageDisclosure.hasMore, + ...(pageDisclosure.wait === null ? {} : pageDisclosure.wait), disclosure: { output: "omitted; metadata only", identity: "item.identity; detail --detail-seq item.seq", @@ -415,6 +423,9 @@ export function renderEventLike(command: string, raw: Record, o ])), ]; if (pageDisclosure !== null) lines.push(`Detail: ${pageDisclosure.detailCommand}`); + if (pageDisclosure?.wait !== null && pageDisclosure?.wait !== undefined) { + lines.push(`Wait: expect=${pageDisclosure.wait.expect} observed=${pageDisclosure.wait.observed} timedOut=${String(pageDisclosure.wait.timedOut)} timeoutMs=${pageDisclosure.wait.timeoutMs} nextAfterSeq=${String(nextAfterSeq ?? 0)}${pageDisclosure.wait.completionReason === null ? "" : ` reason=${pageDisclosure.wait.completionReason}`}`); + } if (nextAfterSeq !== undefined && nextAfterSeq !== null) lines.push(`Next: bun scripts/cli.ts ${nextPagedResourceCommand(command, String(nextAfterSeq), options.limit)}`); return renderedCliResult(raw.ok !== false, command, lines.join("\n")); } diff --git a/scripts/src/agentrun/resource-actions.ts b/scripts/src/agentrun/resource-actions.ts index 93931491..ac0458e1 100644 --- a/scripts/src/agentrun/resource-actions.ts +++ b/scripts/src/agentrun/resource-actions.ts @@ -178,6 +178,8 @@ export function parseResourceOptions(args: string[]): AgentRunResourceOptions { commandId: null, sessionId: null, afterSeq: null, + expect: null, + timeoutMs: 120_000, eventDetailSeq: null, tail: null, fullText: false, @@ -196,7 +198,7 @@ export function parseResourceOptions(args: string[]): AgentRunResourceOptions { mdtodoId: null, passthroughArgs: [], }; - const valueFlags = new Set(["-o", "--output", "--limit", "--cursor", "--queue", "--state", "--reader-id", "--task", "--task-id", "--run", "--run-id", "--command", "--command-id", "--session", "--session-id", "--after-seq", "--detail-seq", "--tail", "--reason", "-f", "--file", "--filename", "--aipod", "--idempotency-key", "--node", "--lane", "--target", "--target-workspace", "--repo", "--ref", "--mdtodo-id"]); + const valueFlags = new Set(["-o", "--output", "--limit", "--cursor", "--queue", "--state", "--reader-id", "--task", "--task-id", "--run", "--run-id", "--command", "--command-id", "--session", "--session-id", "--after-seq", "--expect", "--timeout", "--detail-seq", "--tail", "--reason", "-f", "--file", "--filename", "--aipod", "--idempotency-key", "--node", "--lane", "--target", "--target-workspace", "--repo", "--ref", "--mdtodo-id"]); const booleanFlags = new Set(["--full", "--raw", "--debug", "--input", "--unread", "--dry-run", "--full-text", "--prompt-stdin", "--stdin"]); for (let index = 0; index < args.length; index += 1) { const arg = args[index] ?? ""; @@ -243,7 +245,7 @@ export function parseResourceOptions(args: string[]): AgentRunResourceOptions { ); } if (options.eventDetailSeq !== null) { - const conflictingOptions = ["--after-seq", "--limit", "--full", "--raw"] + const conflictingOptions = ["--after-seq", "--expect", "--timeout", "--limit", "--full", "--raw"] .filter((flag) => resourceOptionPresent(args, flag)); if (conflictingOptions.length > 0) { throw new AgentRunRestError( @@ -253,7 +255,7 @@ export function parseResourceOptions(args: string[]): AgentRunResourceOptions { details: { conflictingOptions: ["--detail-seq", ...conflictingOptions], recoveryActions: [ - "Use --detail-seq without --after-seq, --limit, --full, or --raw for the bounded Secret-safe EventDetail projection.", + "Use --detail-seq without --after-seq, --expect, --timeout, --limit, --full, or --raw for the bounded Secret-safe EventDetail projection.", "Use --after-seq --limit 1 --full -o json or --raw only when explicitly requesting the complete durable event.", ], valuesPrinted: false, @@ -262,6 +264,18 @@ export function parseResourceOptions(args: string[]): AgentRunResourceOptions { ); } } + if (resourceOptionPresent(args, "--timeout") && options.expect === null) { + throw new AgentRunRestError( + "validation-failed", + "--timeout requires --expect ; timeout controls the manager-held long poll and is not a client-side polling interval.", + { + details: { + recoveryActions: ["Use --expect [--timeout ] or remove --timeout for an immediate paged read."], + valuesPrinted: false, + }, + }, + ); + } return options; } @@ -278,6 +292,18 @@ function validateResourceOptionsForVerb(verb: AgentRunResourceVerb, options: Age }, ); } + if (options.expect !== null && verb !== "events") { + throw new AgentRunRestError( + "validation-failed", + "--expect and --timeout are supported only by agentrun events run/.", + { + details: { + recoveryActions: ["Use: bun scripts/cli.ts agentrun events run/ --after-seq --expect [--timeout 120s]."], + valuesPrinted: false, + }, + }, + ); + } } function resourceOptionPresent(args: string[], flag: string): boolean { @@ -308,6 +334,8 @@ export function applyResourceOption(options: AgentRunResourceOptions, flag: stri else if (flag === "--command" || flag === "--command-id") options.commandId = requiredValue(value, flag); else if (flag === "--session" || flag === "--session-id") options.sessionId = requiredValue(value, flag); else if (flag === "--after-seq") options.afterSeq = parseNonNegativeInt(value, "--after-seq", 0, Number.MAX_SAFE_INTEGER); + else if (flag === "--expect") options.expect = parsePositiveInt(value, "--expect"); + else if (flag === "--timeout") options.timeoutMs = parseDurationMs(value, "--timeout"); else if (flag === "--detail-seq") options.eventDetailSeq = parsePositiveInt(value, "--detail-seq"); else if (flag === "--tail") options.tail = parseNonNegativeInt(value, "--tail", 100, 1000); else if (flag === "--reason") options.reason = requiredValue(value, flag); @@ -330,6 +358,17 @@ function parsePositiveInt(raw: string | null, flag: string): number { return value; } +export function parseDurationMs(raw: string | null, flag: string): number { + if (raw === null || raw.length === 0) throw new Error(`${flag} requires a duration`); + const match = /^(\d+)(ms|s|m|h)$/u.exec(raw.trim()); + if (match === null) throw new Error(`${flag} must use an integer duration such as 500ms, 120s, 3m, or 1h`); + const amount = Number(match[1]); + const multiplier = match[2] === "ms" ? 1 : match[2] === "s" ? 1_000 : match[2] === "m" ? 60_000 : 3_600_000; + const value = amount * multiplier; + if (!Number.isSafeInteger(value) || value < 1 || value > 86_400_000) throw new Error(`${flag} must be between 1ms and 24h`); + return value; +} + export function parseNonNegativeInt(raw: string | null, flag: string, defaultValue: number, maxValue: number): number { if (raw === null || raw.length === 0) return defaultValue; const value = Number(raw); @@ -607,6 +646,7 @@ export async function resourceEvents(config: UniDeskConfig | null, command: stri const effectiveLimit = options.full || options.raw ? requestedLimit : Math.min(requestedLimit, 20); const requestLimit = options.full || options.raw ? effectiveLimit : Math.min(500, effectiveLimit + 1); const eventArgs = ["events", runId, "--after-seq", String(options.afterSeq ?? 0), "--limit", String(requestLimit)]; + if (options.expect !== null) eventArgs.push("--expect", String(options.expect), "--timeout-ms", String(options.timeoutMs)); const result = await runAgentRunRestCommand(config, "runs", eventArgs); if (options.raw) return renderMachine(command, result, "json", result.ok !== false); if (options.full) return renderMachine(command, result, options.output === "yaml" ? "yaml" : "json", result.ok !== false); @@ -616,7 +656,32 @@ export async function resourceEvents(config: UniDeskConfig | null, command: stri const hasMore = fetchedItems.length > visibleItems.length; const lastVisibleSeq = visibleItems.length === 0 ? null : nonNegativeIntegerOrNull(visibleItems.at(-1)?.seq); const responseNextAfterSeq = nonNegativeIntegerOrNull(record(innerData(result)).nextAfterSeq); - const nextAfterSeq = hasMore ? lastVisibleSeq : responseNextAfterSeq; + const nextAfterSeq = hasMore ? lastVisibleSeq : responseNextAfterSeq ?? options.afterSeq ?? 0; + const waitData = record(innerData(result)); + const observed = nonNegativeIntegerOrNull(waitData.observed); + if (options.expect !== null && (typeof waitData.timedOut !== "boolean" || observed === null)) { + throw new AgentRunRestError( + "schema-mismatch", + "AgentRun manager long-poll response must include boolean timedOut and non-negative observed fields.", + { + details: { + runId, + expect: options.expect, + timeoutMs: options.timeoutMs, + returnedTimedOutType: typeof waitData.timedOut, + returnedObserved: waitData.observed ?? null, + valuesPrinted: false, + }, + }, + ); + } + const waitDisclosure = options.expect === null ? null : { + expect: options.expect, + timeoutMs: options.timeoutMs, + timedOut: waitData.timedOut === true, + observed: observed ?? fetchedItems.length, + completionReason: stringOrNull(waitData.completionReason) ?? stringOrNull(waitData.reason), + }; const targetArgs = [ options.node === null ? null : `--node ${options.node}`, options.lane === null ? null : `--lane ${options.lane}`, @@ -629,6 +694,7 @@ export async function resourceEvents(config: UniDeskConfig | null, command: stri hasMore, nextAfterSeq, detailCommand, + wait: waitDisclosure, }); } diff --git a/scripts/src/agentrun/rest-bridge.ts b/scripts/src/agentrun/rest-bridge.ts index cc260486..b59a5ce4 100644 --- a/scripts/src/agentrun/rest-bridge.ts +++ b/scripts/src/agentrun/rest-bridge.ts @@ -523,7 +523,16 @@ export async function runAgentRunSessionsRest(action: string | undefined, id: st export async function runAgentRunRunsRest(action: string | undefined, id: string | undefined, args: string[]): Promise> { if (action === "create") return await agentRunRestRequest("agentrun runs create", "POST", "/api/v1/runs", await requiredJsonBody(args, "runs create")); if (action === "show" && id) return await agentRunRestRequest("agentrun runs show", "GET", `/api/v1/runs/${encodeURIComponent(id)}`); - if (action === "events" && id) return await agentRunRestRequest("agentrun runs events", "GET", `/api/v1/runs/${encodeURIComponent(id)}/events${agentRunQuery(args, ["after-seq", "limit"])}`); + if (action === "events" && id) { + const timeoutMs = Number(agentRunOption(args, "timeout-ms") ?? "0"); + return await agentRunRestRequest( + "agentrun runs events", + "GET", + `/api/v1/runs/${encodeURIComponent(id)}/events${agentRunQuery(args, ["after-seq", "expect", "timeout-ms", "limit"])}`, + undefined, + Number.isSafeInteger(timeoutMs) && timeoutMs > 0 ? timeoutMs + 5_000 : undefined, + ); + } if (action === "result" && id) return await agentRunRestRequest("agentrun runs result", "GET", `/api/v1/runs/${encodeURIComponent(id)}/result${agentRunQuery(args, ["command-id", "command"])}`); if (action === "cancel" && id) return await agentRunRestRequest("agentrun runs cancel", "POST", `/api/v1/runs/${encodeURIComponent(id)}/cancel`, cancelBodyFromArgs(args)); throw new AgentRunRestError("validation-failed", `unsupported runs command: ${[action, id].filter(Boolean).join(" ") || "(empty)"}`); @@ -1157,14 +1166,15 @@ export async function sessionRunnerJobBody(args: string[], defaults: Record> { +export async function agentRunRestRequest(command: string, method: AgentRunHttpMethod, pathValue: string, body?: unknown, requestTimeoutMs?: number): Promise> { const clientConfig = readAgentRunClientConfig(); - if (activeAgentRunRestTarget !== null) return await agentRunLaneRestRequest(command, method, pathValue, body, clientConfig, activeAgentRunRestTarget); + const effectiveTimeoutMs = Math.max(clientConfig.manager.timeoutMs, requestTimeoutMs ?? 0); + if (activeAgentRunRestTarget !== null) return await agentRunLaneRestRequest(command, method, pathValue, body, clientConfig, activeAgentRunRestTarget, effectiveTimeoutMs); const auth = resolveAgentRunAuth(clientConfig); const bridgeBase = agentRunRestBridgeMetadata(clientConfig, auth, method, pathValue); const startedAt = Date.now(); const controller = new AbortController(); - const timeout = setTimeout(() => controller.abort(), clientConfig.manager.timeoutMs); + const timeout = setTimeout(() => controller.abort(), effectiveTimeoutMs); let response: Response; try { const headers: Record = { @@ -1178,7 +1188,7 @@ export async function agentRunRestRequest(command: string, method: AgentRunHttpM response = await fetch(new URL(pathValue, clientConfig.manager.baseUrl), init); } catch (error) { const timedOut = isAbortLikeError(error); - throw new AgentRunRestError(timedOut ? "agentrun-timeout" : "agentrun-connect-failed", timedOut ? `AgentRun server request timed out for ${method} ${pathValue} after ${clientConfig.manager.timeoutMs}ms` : `AgentRun server connection failed for ${method} ${pathValue}: ${error instanceof Error ? error.message : String(error)}`, { bridge: { ...bridgeBase, elapsedMs: Date.now() - startedAt } }); + throw new AgentRunRestError(timedOut ? "agentrun-timeout" : "agentrun-connect-failed", timedOut ? `AgentRun server request timed out for ${method} ${pathValue} after ${effectiveTimeoutMs}ms` : `AgentRun server connection failed for ${method} ${pathValue}: ${error instanceof Error ? error.message : String(error)}`, { bridge: { ...bridgeBase, elapsedMs: Date.now() - startedAt } }); } finally { clearTimeout(timeout); } @@ -1215,7 +1225,7 @@ export function isAbortLikeError(error: unknown): boolean { return error.name === "AbortError" || /abort|timed out|timeout/iu.test(error.message); } -export async function agentRunLaneRestRequest(command: string, method: AgentRunHttpMethod, pathValue: string, body: unknown, clientConfig: AgentRunClientConfig, target: AgentRunRestTarget): Promise> { +export async function agentRunLaneRestRequest(command: string, method: AgentRunHttpMethod, pathValue: string, body: unknown, clientConfig: AgentRunClientConfig, target: AgentRunRestTarget, requestTimeoutMs = clientConfig.manager.timeoutMs): Promise> { const bridgeBase = agentRunLaneRestBridgeMetadata(clientConfig, target, method, pathValue); const startedAt = Date.now(); const proxyBaseUrl = `http://127.0.0.1:${target.spec.runtime.managerPort}`; @@ -1225,7 +1235,7 @@ export async function agentRunLaneRestRequest(command: string, method: AgentRunH body: body ?? null, baseUrl: proxyBaseUrl, serviceBaseUrl: target.spec.runtime.internalBaseUrl, - timeoutMs: clientConfig.manager.timeoutMs, + timeoutMs: requestTimeoutMs, authEnv: clientConfig.auth.env, header: clientConfig.auth.header, scheme: clientConfig.auth.scheme, diff --git a/scripts/src/agentrun/utils.ts b/scripts/src/agentrun/utils.ts index 06f45530..b22068c7 100644 --- a/scripts/src/agentrun/utils.ts +++ b/scripts/src/agentrun/utils.ts @@ -76,6 +76,8 @@ export interface AgentRunResourceOptions { commandId: string | null; sessionId: string | null; afterSeq: number | null; + expect: number | null; + timeoutMs: number; eventDetailSeq: number | null; tail: number | null; fullText: boolean;