diff --git a/.changeset/agent-profile-json-schema.md b/.changeset/agent-profile-json-schema.md new file mode 100644 index 0000000..706f8be --- /dev/null +++ b/.changeset/agent-profile-json-schema.md @@ -0,0 +1,5 @@ +--- +"@tangle-network/agent-interface": minor +--- + +Export a model-facing JSON Schema generated from the canonical `AgentProfile` validator. diff --git a/.changeset/agent-profile-routing-identity.md b/.changeset/agent-profile-routing-identity.md new file mode 100644 index 0000000..db9955b --- /dev/null +++ b/.changeset/agent-profile-routing-identity.md @@ -0,0 +1,5 @@ +--- +"@tangle-network/agent-interface": patch +--- + +Document and test that authored harness and model-routing preferences participate in canonical profile identity while execution receipts separately bind overrides. diff --git a/.changeset/cli-bridge-exact-profile.md b/.changeset/cli-bridge-exact-profile.md new file mode 100644 index 0000000..4d7eae3 --- /dev/null +++ b/.changeset/cli-bridge-exact-profile.md @@ -0,0 +1,5 @@ +--- +"@tangle-network/agent-provider-cli-bridge": patch +--- + +Keep the task message separate from the exact `AgentProfile`, and forward the profile's reasoning effort to cli-bridge. diff --git a/.changeset/cli-bridge-steerable-sessions.md b/.changeset/cli-bridge-steerable-sessions.md new file mode 100644 index 0000000..1a12261 --- /dev/null +++ b/.changeset/cli-bridge-steerable-sessions.md @@ -0,0 +1,5 @@ +--- +"@tangle-network/agent-provider-cli-bridge": minor +--- + +Expose cli-bridge durable runs through `AgentEnvironment.dispatch()` and `AgentEnvironment.session()` with exact identity, cursor replay, usage-preserving results, continuation, and terminal-confirmed cancellation. diff --git a/packages/agent-interface/CHANGELOG.md b/packages/agent-interface/CHANGELOG.md index 19f45e2..8a7c2e1 100644 --- a/packages/agent-interface/CHANGELOG.md +++ b/packages/agent-interface/CHANGELOG.md @@ -177,7 +177,7 @@ ### Minor Changes -- afe552d: Add optional `AgentProfile.harness` — a typed, executor-overridable preferred execution harness (`HarnessType`). Formalizes the `profile.harness` runtimes already read untyped; identity stays harness-agnostic (the leaderboard `harness × model` axis and per-worker supervisor routing still override it), and it becomes a first-class lever an improvement loop can optimize. +- afe552d: Add optional `AgentProfile.harness` — a typed, executor-overridable preferred execution harness (`HarnessType`). Formalizes the `profile.harness` runtimes already read untyped; the authored preference participates in canonical profile identity while execution receipts separately bind any override, and it becomes a first-class lever an improvement loop can optimize. ## 0.19.0 diff --git a/packages/agent-interface/src/agent-execution-preparation.test.ts b/packages/agent-interface/src/agent-execution-preparation.test.ts index a4a06e0..d2437b8 100644 --- a/packages/agent-interface/src/agent-execution-preparation.test.ts +++ b/packages/agent-interface/src/agent-execution-preparation.test.ts @@ -498,6 +498,91 @@ describe("profile materialization leaves", () => { }); describe("AgentExecutionPreparationReceipt", () => { + it("binds authored routing preferences and separately receipts execution overrides", () => { + const authoredProfile: AgentProfile = { + name: "worker", + harness: "pi", + model: { + default: "tangle/glm-5.2", + provider: "tangle-router", + }, + }; + const harnessChanged: AgentProfile = { + ...authoredProfile, + harness: "codex", + }; + const modelChanged: AgentProfile = { + ...authoredProfile, + model: { + ...authoredProfile.model, + default: "anthropic/claude-opus-4-1", + }, + }; + const providerChanged: AgentProfile = { + ...authoredProfile, + model: { + ...authoredProfile.model, + provider: "anthropic", + }, + }; + const authoredDigests = [ + authoredProfile, + harnessChanged, + modelChanged, + providerChanged, + ].map(canonicalAgentProfileDigest); + expect(new Set(authoredDigests)).toHaveLength(4); + + const effectiveProfile: AgentProfile = { + ...authoredProfile, + harness: "codex", + model: { + default: "anthropic/claude-opus-4-1", + provider: "anthropic", + }, + }; + const overriddenAxes = new Set([ + "harness", + "modelDefault", + "modelProvider", + ]); + const receipt = buildReceipt({ + authoredProfile, + effectiveProfile, + axisResults: coverage(authoredProfile).map((result) => + overriddenAxes.has(result.axis) + ? { + ...result, + disposition: "overridden" as const, + reason: "execution request selected a different route", + } + : result, + ), + }); + + expect(receipt.authoredProfileDigest).toBe( + canonicalAgentProfileDigest(authoredProfile), + ); + expect(receipt.effectiveProfileDigest).toBe( + canonicalAgentProfileDigest(effectiveProfile), + ); + expect(receipt.authoredProfileDigest).not.toBe( + receipt.effectiveProfileDigest, + ); + expect(receipt).toMatchObject({ + harness: "codex", + resolvedModel: { + requested: "anthropic/claude-opus-4-1", + provider: "anthropic", + }, + }); + expect( + receipt.axisResults + .filter((result) => result.disposition === "overridden") + .map((result) => result.axis), + ).toEqual(["modelDefault", "modelProvider", "harness"]); + }); + it("builds from a sealed lease and validates only after execution binding", () => { const receipt = buildReceipt(); diff --git a/packages/agent-interface/src/agent-execution-preparation.ts b/packages/agent-interface/src/agent-execution-preparation.ts index 95efba1..97c8981 100644 --- a/packages/agent-interface/src/agent-execution-preparation.ts +++ b/packages/agent-interface/src/agent-execution-preparation.ts @@ -118,7 +118,9 @@ export interface AgentExecutionPreparationReceipt { schemaVersion: 1; preparationId: string; requestDigest: Sha256Digest; + /** Canonical identity before executor overrides. */ authoredProfileDigest: Sha256Digest; + /** Canonical identity after executor overrides. */ effectiveProfileDigest: Sha256Digest; backend: string; harness: HarnessType; diff --git a/packages/agent-interface/src/agent-profile.ts b/packages/agent-interface/src/agent-profile.ts index 57af33a..a86a503 100644 --- a/packages/agent-interface/src/agent-profile.ts +++ b/packages/agent-interface/src/agent-profile.ts @@ -366,13 +366,12 @@ export interface AgentProfile { * Preferred execution harness for this profile — the coding-CLI runtime that * materializes and runs it (`claude-code`, `codex`, `opencode`, `pi`, …). * - * This is an optional PREFERENCE, not part of profile IDENTITY: the same - * profile still runs on any harness, and an executor MAY override it (e.g. the - * leaderboard's `harness × model` axis sweeps a profile across every harness, - * and a supervisor spawning a worker may pick a harness per subtask). When - * unset, the caller/executor chooses. Formalizes what runtimes already read as - * `profile.harness`; making it typed also lets an improvement loop optimize - * harness routing as a first-class lever. + * This optional authored preference participates in canonical profile + * identity. An executor MAY override it for one run; the effective profile and + * execution receipt then bind that override without changing what was + * authored. When unset, the caller/executor chooses. Formalizes what runtimes + * already read as `profile.harness`; making it typed also lets an improvement + * loop optimize harness routing as a first-class lever. */ harness?: HarnessType; permissions?: Record; diff --git a/packages/agent-interface/src/harness.ts b/packages/agent-interface/src/harness.ts index 698a069..5890cf1 100644 --- a/packages/agent-interface/src/harness.ts +++ b/packages/agent-interface/src/harness.ts @@ -3,13 +3,12 @@ import { z } from "zod"; /** * The execution runner for an agent — WHICH runtime materializes and runs an `AgentProfile`. * - * Harness is an EXECUTION concern, not part of profile IDENTITY: the same `AgentProfile` - * (prompt/model/skills/tools/mcp/subagents) runs on any harness. A profile MAY carry an optional - * `harness` PREFERENCE (`AgentProfile.harness`), but the caller/executor can always override it per - * run — the leaderboard's `harness × model` axis sweeps one profile across every harness, and a - * supervisor may pick a harness per spawned worker. This is the single shared enum every layer - * references instead of keeping its own copy (session control, the profile materializer, the - * cli-bridge backends, VB profile specs). + * This enum describes an EXECUTION concern. When an `AgentProfile` authors a `harness` + * preference, that field participates in its canonical identity like every other parsed profile + * field. A caller or executor may override the preference for one run, but the execution receipt + * preserves the authored identity, effective identity, and selected harness separately. This is + * the single shared enum every layer references instead of keeping its own copy (session control, + * the profile materializer, the cli-bridge backends, VB profile specs). * * `cli-base` is the router-backed mode — a plain multi-turn router call (a reviewer, a cheap judge, * a one-shot) with no full coding-agent harness. The rest are full agentic harnesses run in a diff --git a/packages/agent-interface/src/profile-harness.test.ts b/packages/agent-interface/src/profile-harness.test.ts index 363ea3d..37b0642 100644 --- a/packages/agent-interface/src/profile-harness.test.ts +++ b/packages/agent-interface/src/profile-harness.test.ts @@ -1,4 +1,5 @@ import { describe, expect, it } from "vitest"; +import { canonicalAgentProfileDigest } from "./agent-execution-preparation.js"; import type { AgentProfile } from "./agent-profile.js"; import { agentProfileSchema } from "./profile-schema.js"; import type { HarnessType } from "./harness.js"; @@ -14,7 +15,7 @@ describe("AgentProfile.harness (optional overridable preference)", () => { expect(parsed.harness).toBe("codex"); }); - it("is optional — a profile without harness still parses (harness-agnostic identity)", () => { + it("is optional — a profile without a preference still parses", () => { const parsed = agentProfileSchema.parse({ name: "w" }); expect(parsed.harness).toBeUndefined(); }); @@ -26,12 +27,15 @@ describe("AgentProfile.harness (optional overridable preference)", () => { expect(() => agentProfileSchema.parse({ harness: "kimi" })).toThrow(); }); - it("does not constrain identity — the same profile is valid with any harness swapped in", () => { + it("accepts every known runner while binding each preference into authored identity", () => { const base: AgentProfile = { name: "w", prompt: { systemPrompt: "do the task" } }; + const digests = []; for (const harness of ["claude-code", "opencode", "pi", "cli-base"] as const) { const parsed = agentProfileSchema.parse({ ...base, harness }); expect(parsed.harness).toBe(harness); expect(parsed.prompt).toEqual(base.prompt); + digests.push(canonicalAgentProfileDigest(parsed)); } + expect(new Set(digests)).toHaveLength(4); }); }); diff --git a/packages/agent-interface/src/profile-schema.test.ts b/packages/agent-interface/src/profile-schema.test.ts index 1e006cd..ecc69c8 100644 --- a/packages/agent-interface/src/profile-schema.test.ts +++ b/packages/agent-interface/src/profile-schema.test.ts @@ -1,11 +1,15 @@ import { describe, expect, it } from "vitest"; +import { z } from "zod"; import { REASONING_EFFORTS, + type AgentProfile, type AgentProfileMcpServer, } from "./agent-profile.js"; +import { harnessTypeSchema } from "./harness.js"; import { validateAgentProfileSecurity } from "./profile-security.js"; import { agentProfileDiffSchema, + agentProfileJsonSchema, agentProfileSchema, capabilitySchema, reasoningEffortSchema, @@ -366,6 +370,180 @@ describe("agentProfileSchema", () => { }); }); +describe("agentProfileJsonSchema", () => { + it("describes the complete authored profile without encoded record-key constraints", () => { + const properties = agentProfileJsonSchema.properties as Record< + string, + Record + >; + + expect(properties.prompt).toMatchObject({ + type: "object", + properties: { + systemPrompt: { type: "string" }, + instructions: { type: "array", items: { type: "string" } }, + }, + }); + expect(properties.model).toMatchObject({ + type: "object", + properties: { + default: { type: "string" }, + provider: { type: "string" }, + }, + }); + expect(properties.harness).toEqual({ + type: "string", + enum: harnessTypeSchema.options, + }); + expect(properties.extensions).toMatchObject({ + type: "object", + additionalProperties: { + anyOf: [{ type: "object", additionalProperties: {} }, {}], + }, + }); + expect(properties.tools).toMatchObject({ + type: "object", + additionalProperties: { type: "boolean" }, + }); + expect(properties.permissions).toMatchObject({ + type: "object", + additionalProperties: { + anyOf: [ + { type: "string", enum: ["allow", "deny", "ask"] }, + { + type: "object", + additionalProperties: { + type: "string", + enum: ["allow", "deny", "ask"], + }, + }, + ], + }, + }); + + const serialized = JSON.stringify(agentProfileJsonSchema); + expect(serialized).not.toContain("^u(?:[0-9a-f]{4})*$"); + expect(serialized).not.toContain('"propertyNames"'); + expect(serialized).not.toContain('"$schema"'); + }); + + it("admits one ordinary complete profile through both published contracts", () => { + const profile: AgentProfile = { + name: "research-worker", + description: "Investigate one question and preserve evidence.", + version: "1.0.0", + tags: ["research", "technical"], + prompt: { + systemPrompt: "Investigate the supplied task.", + instructions: ["Record evidence.", "State uncertainty."], + }, + model: { + default: "router/frontier", + small: "router/fast", + provider: "router", + reasoningEffort: "high", + metadata: { routing: { latencyClass: "interactive" } }, + }, + harness: "pi", + permissions: { + shell: "allow", + network: { read: "allow", write: "ask" }, + }, + tools: { read_file: true, write_file: false }, + mcp: { + knowledge: { + transport: "http", + url: "https://mcp.example.com", + headers: { + Authorization: { + kind: "secret-ref", + key: "MCP_AUTH", + format: "bearer", + }, + }, + }, + }, + connections: [ + { + connectionId: "github-primary", + capabilities: ["repo.read"], + alias: "source", + }, + ], + subagents: { + reviewer: { + description: "Review evidence.", + prompt: "Find unsupported claims.", + model: "router/frontier", + tools: { read_file: true }, + permissions: { shell: "deny" }, + maxSteps: 4, + metadata: { focus: { citations: true } }, + }, + }, + resources: { + files: [ + { + path: "AGENTS.md", + resource: { + kind: "inline", + name: "instructions", + content: "Preserve primary evidence.", + }, + }, + ], + skills: [ + { + kind: "inline", + name: "source-review", + content: "Check every claim against its source.", + }, + ], + instructions: "Follow the supplied research protocol.", + failOnError: true, + }, + hooks: { + beforeRun: [ + { + command: "prepare", + timeoutMs: 1_000, + blocking: true, + matcher: "research", + env: { + MODE: { kind: "public", value: "read-only" }, + }, + }, + ], + }, + modes: { + review: { + description: "Audit a draft.", + model: "router/frontier", + prompt: "Check the draft.", + tools: { read_file: true }, + permissions: { shell: "deny" }, + metadata: { severity: "strict" }, + }, + }, + confidential: { + tee: "tdx", + attestationNonce: "public-nonce", + sealed: true, + attestationRefresh: true, + }, + metadata: { owner: { team: "discovery" } }, + extensions: { + provider: { session: { durable: true } }, + }, + }; + + const modelInputSchema = z.fromJSONSchema(agentProfileJsonSchema); + + expect(modelInputSchema.safeParse(profile).success).toBe(true); + expect(agentProfileSchema.safeParse(profile).success).toBe(true); + }); +}); + describe("profile container schemas", () => { it("rejects unknown diff and capability fields", () => { expect( diff --git a/packages/agent-interface/src/profile-schema.ts b/packages/agent-interface/src/profile-schema.ts index beaa0f4..9544fe1 100644 --- a/packages/agent-interface/src/profile-schema.ts +++ b/packages/agent-interface/src/profile-schema.ts @@ -404,6 +404,59 @@ export const agentProfileSchema = z validateNestedRecordKeys(profile, context, [], new Set()); }); +const encodedRecordKeyPattern = "^u(?:[0-9a-f]{4})*$"; + +function isEncodedRecordKeyPropertyNames(value: unknown): boolean { + if (value === null || typeof value !== "object" || Array.isArray(value)) { + return false; + } + const propertyNames = value as Record; + return ( + propertyNames.type === "string" && + propertyNames.pattern === encodedRecordKeyPattern + ); +} + +function removeModelInputSchemaArtifacts(value: unknown): unknown { + if (Array.isArray(value)) { + return value.map(removeModelInputSchemaArtifacts); + } + if (value === null || typeof value !== "object") { + return value; + } + + const result: Record = {}; + for (const [key, entry] of Object.entries(value)) { + if (key === "$schema") { + continue; + } + if (key === "propertyNames" && isEncodedRecordKeyPropertyNames(entry)) { + continue; + } + result[key] = removeModelInputSchemaArtifacts(entry); + } + return result; +} + +/** + * Model-facing JSON Schema for an authored {@link AgentProfile}. + * + * This is generated from {@link agentProfileSchema}'s input contract rather + * than restating the profile shape. Zod sees encoded record keys at its + * preprocessing boundary, so the generated `propertyNames` constraints for + * that internal encoding are removed before exposing the schema to callers. + * The root dialect declaration is also omitted because tool APIs embed this + * value as a subschema rather than serving it as a standalone document. + * Runtime admission remains the canonical Zod validator above. + */ +export const agentProfileJsonSchema = removeModelInputSchemaArtifacts( + z.toJSONSchema(agentProfileSchema, { + target: "draft-2020-12", + io: "input", + unrepresentable: "any", + }), +) as z.core.JSONSchema.JSONSchema; + function validateNestedRecordKeys( value: unknown, context: z.RefinementCtx, diff --git a/packages/agent-provider-cli-bridge/README.md b/packages/agent-provider-cli-bridge/README.md index 424ef0e..adac17b 100644 --- a/packages/agent-provider-cli-bridge/README.md +++ b/packages/agent-provider-cli-bridge/README.md @@ -17,6 +17,22 @@ const environment = await provider.create({ model: { default: 'gpt-5' }, }, }) + +const reference = await environment.dispatch({ + prompt: 'inspect the repository', + sessionId: 'research-session', + executionId: 'research-turn-1', +}) +const session = environment.session(reference.id) + +for await (const event of session.events({ since: '0' })) { + // Sequence-numbered bridge events, including normalized usage and result data. +} + +await session.prompt({ + prompt: 'change direction using the evidence already collected', + executionId: 'research-turn-2', +}) ``` The bridge model is selected from run data in this order: the turn, the provider default, or the profile's `harness` plus `model.default`. @@ -24,7 +40,13 @@ Execution fails before network use when none is present. Passing the same `sessionId` on later turns continues the same CLI conversation. `executionId` gives a turn stable bridge identity, and `lastEventId` reattaches after a reader failure. -Stopping a reader or destroying the environment cancels every active bridge run and waits for terminal confirmation. +`dispatch()` starts a bridge-owned durable run and returns after detaching its HTTP reader. +The returned `AgentSession` exposes status, cursor-based event replay, the terminal result, continuation, and cancellation. +Cancellation returns only after cli-bridge confirms the run is terminal. +When `dispatch()` receives no `sessionId`, it creates one from that turn's stable run id and returns it. +Replay and result reads fail loudly after cli-bridge's configured replay retention expires. +Stopping a `session.events()` reader detaches only that replay observer. +Stopping a direct `environment.stream()` reader or destroying the environment cancels its active bridge runs and waits for terminal confirmation. Response headers and streamed bodies have no transport timeout by default. For unattended runs, set `headersTimeoutMs`, `bodyTimeoutMs`, or an `AbortSignal` so an unresponsive bridge cannot wait forever. diff --git a/packages/agent-provider-cli-bridge/src/index.test.ts b/packages/agent-provider-cli-bridge/src/index.test.ts index 4637b61..31c4643 100644 --- a/packages/agent-provider-cli-bridge/src/index.test.ts +++ b/packages/agent-provider-cli-bridge/src/index.test.ts @@ -1,9 +1,449 @@ import { createServer } from "node:http"; +import type { AgentProfile } from "@tangle-network/agent-interface"; import type { AgentEnvironment } from "@tangle-network/agent-interface/environment-provider"; import { describe, expect, it } from "vitest"; import { createCliBridgeProvider } from "./index.js"; describe("createCliBridgeProvider", () => { + it("rejects a named profile before network use", async () => { + let called = false; + const provider = createCliBridgeProvider({ + baseUrl: "http://bridge.local", + fetch: async () => { + called = true; + return new Response(); + }, + }); + + await expect(provider.create({ profile: "profile-id" })).rejects.toThrow( + /requires an inline AgentProfile/, + ); + expect(called).toBe(false); + }); + + it("keeps profile authority separate from the task and forwards it unchanged", async () => { + let body: Record | undefined; + const profile: AgentProfile = { + name: "scientist", + harness: "pi", + model: { + provider: "tangle-router", + default: "glm-5.2", + reasoningEffort: "xhigh", + }, + prompt: { systemPrompt: "Use this system prompt exactly once." }, + mcp: { + coordination: { + transport: "http", + url: "http://127.0.0.1:4444/mcp", + }, + }, + }; + const expectedProfile = structuredClone(profile); + const provider = createCliBridgeProvider({ + baseUrl: "http://bridge.local", + fetch: async (_url, init) => { + body = JSON.parse(String(init?.body)); + return new Response( + 'data: {"choices":[{"delta":{"content":"done"},"finish_reason":"stop"}]}\n\ndata: [DONE]\n\n', + { status: 200, headers: { "content-type": "text/event-stream" } }, + ); + }, + }); + const environment = await provider.create({ profile }); + await expect(environment.session?.("missing").status()).resolves.toBeNull(); + profile.prompt!.systemPrompt = "Caller mutation must not cross intake."; + profile.model!.default = "different-model"; + + await consumeTurn(environment, { + prompt: "run the task", + sessionId: "profile-session", + turnId: "profile-turn", + executionId: "profile-run", + }); + + expect(body).toMatchObject({ + model: "pi/tangle-router/glm-5.2", + effort: "xhigh", + messages: [{ role: "user", content: "run the task" }], + }); + expect(body?.agent_profile).toEqual(expectedProfile); + }); + + it("maps durable dispatch, replay, result, and continuation into one exact session", async () => { + const profile: AgentProfile = { + name: "research-leader", + harness: "pi", + model: { + provider: "tangle-router", + default: "glm-5.2", + reasoningEffort: "high", + }, + prompt: { systemPrompt: "Lead the research." }, + }; + const requests: Array<{ + body: Record; + cursor: string | null; + }> = []; + const statuses = new Map(); + let dispatchReaderDetached = false; + const provider = createCliBridgeProvider({ + baseUrl: "http://bridge.local", + fetch: async (url, init) => { + const target = String(url); + if (init?.method === "GET") { + const runId = decodeURIComponent(target.split("/").at(-1)?.split("?")[0] ?? ""); + const status = statuses.get(runId) ?? "running"; + return runResponse(runId, status, status === "done"); + } + const body = JSON.parse(String(init?.body)) as Record; + const runId = String(body.run_id); + const cursor = new Headers(init?.headers).get("last-event-id"); + requests.push({ body, cursor }); + const headers = { + "content-type": "text/event-stream", + "x-run-id": runId, + "x-run-request-digest": `digest-${runId}`, + }; + if (body.stream === false) { + return Response.json( + { + choices: [{ + message: { role: "assistant", content: `complete-${runId}` }, + finish_reason: "stop", + }], + usage: { + prompt_tokens: 11, + completion_tokens: 7, + total_tokens: 18, + reasoning_tokens: 3, + cost: 0.04, + }, + }, + { + headers: { + "x-run-id": runId, + "x-run-request-digest": `digest-${runId}`, + }, + }, + ); + } + if (cursor !== null) { + statuses.set(runId, "done"); + return new Response( + [ + `id: 2\ndata: {"choices":[{"delta":{"content":"replayed-${runId}"},"finish_reason":"stop"}],"usage":{"prompt_tokens":11,"completion_tokens":7,"total_tokens":18,"reasoning_tokens":3,"cost":0.04}}\n\n`, + "data: [DONE]\n\n", + ].join(""), + { status: 200, headers }, + ); + } + if (runId === "run-2") { + statuses.set(runId, "done"); + return new Response( + `id: 1\ndata: {"choices":[{"delta":{"content":"continued"},"finish_reason":"stop"}],"usage":{"prompt_tokens":5,"completion_tokens":2,"cost":0.01}}\n\ndata: [DONE]\n\n`, + { status: 200, headers }, + ); + } + statuses.set(runId, "running"); + return new Response( + new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode(": connected\n\n")); + }, + cancel() { + dispatchReaderDetached = true; + }, + }), + { status: 200, headers }, + ); + }, + }); + const environment = await provider.create({ profile }); + + const reference = await environment.dispatch?.({ + prompt: "initial task", + sessionId: "research-session", + turnId: "turn-1", + executionId: "run-1", + }); + + expect(reference).toEqual({ + id: "research-session", + provider: "cli-bridge", + metadata: { + runId: "run-1", + requestDigest: "digest-run-1", + }, + }); + expect(dispatchReaderDetached).toBe(true); + const session = environment.session?.(reference!.id); + await expect(session?.status()).resolves.toBe("running"); + const replayed = []; + for await (const event of session!.events({ since: "1" })) replayed.push(event); + expect(replayed.map((event) => event.type)).toEqual([ + "usage", + "message.part.updated", + "result", + ]); + expect(replayed[0]?.usage).toEqual({ + inputTokens: 11, + outputTokens: 7, + totalTokens: 18, + reasoningTokens: 3, + cost: 0.04, + }); + expect(replayed.at(-1)).toMatchObject({ + id: "2", + data: { + finalText: "complete-run-1", + status: "completed", + }, + }); + await expect(session?.result()).resolves.toMatchObject({ + text: "complete-run-1", + success: true, + sessionId: "research-session", + usage: { + inputTokens: 11, + outputTokens: 7, + totalTokens: 18, + reasoningTokens: 3, + cost: 0.04, + }, + metadata: { + runId: "run-1", + status: "done", + requestDigest: "digest-run-1", + }, + }); + await expect(session?.prompt({ + prompt: "new direction", + turnId: "turn-2", + executionId: "run-2", + })).resolves.toMatchObject({ + text: "continued", + success: true, + sessionId: "research-session", + usage: { + inputTokens: 5, + outputTokens: 2, + cost: 0.01, + }, + metadata: { runId: "run-2", status: "done" }, + }); + await expect(session?.prompt({ + prompt: "wrong conversation", + sessionId: "other-session", + })).rejects.toThrow(/cannot prompt session/); + + const wireBodies = requests + .filter(({ body }) => body.stream !== false) + .map(({ body }) => body); + expect(wireBodies).toEqual([ + expect.objectContaining({ + run_id: "run-1", + session_id: "research-session", + agent_profile: profile, + messages: [{ role: "user", content: "initial task" }], + }), + expect.objectContaining({ + run_id: "run-1", + session_id: "research-session", + agent_profile: profile, + messages: [{ role: "user", content: "initial task" }], + }), + expect.objectContaining({ + run_id: "run-1", + session_id: "research-session", + agent_profile: profile, + messages: [{ role: "user", content: "initial task" }], + }), + expect.objectContaining({ + run_id: "run-2", + session_id: "research-session", + agent_profile: profile, + messages: [{ role: "user", content: "new direction" }], + }), + ]); + expect(requests.filter(({ body }) => body.stream !== false).map(({ cursor }) => cursor)) + .toEqual([null, "1", "0", null]); + for (const { body } of requests) { + expect(body.messages).not.toEqual( + expect.arrayContaining([{ role: "system", content: "Lead the research." }]), + ); + } + expect(provider.capabilities()).toMatchObject({ + streaming: { detach: true, replay: true }, + sessions: { continue: true }, + }); + }); + + it("waits for terminal proof when cancelling a dispatched session", async () => { + const requested: string[] = []; + let getCalls = 0; + const provider = createCliBridgeProvider({ + baseUrl: "http://bridge.local", + defaultModel: "runner/model", + fetch: async (url, init) => { + const target = String(url); + requested.push(target); + if (target.endsWith("/cancel")) { + return cancelResponse("cancel-run", "running", false, 202); + } + if (init?.method === "GET") { + getCalls += 1; + return runResponse( + "cancel-run", + getCalls === 1 ? "running" : "cancelled", + getCalls > 1, + ); + } + return new Response( + new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode(": connected\n\n")); + }, + }), + { + status: 200, + headers: { + "content-type": "text/event-stream", + "x-run-id": "cancel-run", + "x-run-request-digest": "cancel-digest", + }, + }, + ); + }, + }); + const environment = await provider.create({ profile: { name: "worker" } }); + const reference = await environment.dispatch?.({ + prompt: "long task", + sessionId: "cancel-session", + executionId: "cancel-run", + }); + + await environment.session?.(reference!.id).cancel(); + + expect(requested).toContain( + "http://bridge.local/v1/runs/cancel-run/cancel", + ); + expect(requested).toContain( + "http://bridge.local/v1/runs/cancel-run?wait_ms=30000", + ); + expect(getCalls).toBe(2); + }); + + it("rejects replay when the bridge changes a bound request digest", async () => { + const provider = createCliBridgeProvider({ + baseUrl: "http://bridge.local", + defaultModel: "runner/model", + fetch: async (_url, init) => { + const body = JSON.parse(String(init?.body)) as Record; + const runId = String(body.run_id); + const replay = new Headers(init?.headers).has("last-event-id"); + return new Response( + replay + ? 'data: {"choices":[{"delta":{"content":"wrong"},"finish_reason":"stop"}]}\n\ndata: [DONE]\n\n' + : new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode(": connected\n\n")); + }, + }), + { + status: 200, + headers: { + "content-type": "text/event-stream", + "x-run-id": runId, + "x-run-request-digest": replay ? "changed-digest" : "original-digest", + }, + }, + ); + }, + }); + const environment = await provider.create({ profile: { name: "worker" } }); + const reference = await environment.dispatch?.({ + prompt: "task", + sessionId: "digest-session", + executionId: "digest-run", + }); + + await expect( + consumeEvents(environment.session!(reference!.id).events({ since: "0" })), + ).rejects.toThrow(/changed request digest/); + }); + + it("keeps concurrent continuation results bound to their own run identity", async () => { + const statuses = new Map(); + let releaseFirst: (() => void) | undefined; + const provider = createCliBridgeProvider({ + baseUrl: "http://bridge.local", + defaultModel: "runner/model", + fetch: async (url, init) => { + if (init?.method === "GET") { + const runId = decodeURIComponent( + String(url).split("/").at(-1)?.split("?")[0] ?? "", + ); + const status = statuses.get(runId) ?? "running"; + return runResponse(runId, status, status === "done"); + } + const body = JSON.parse(String(init?.body)) as Record; + const runId = String(body.run_id); + statuses.set(runId, "running"); + const headers = { + "content-type": "text/event-stream", + "x-run-id": runId, + "x-run-request-digest": `digest-${runId}`, + }; + if (runId === "concurrent-a") { + return new Response( + new ReadableStream({ + start(controller) { + releaseFirst = () => { + statuses.set(runId, "done"); + controller.enqueue( + new TextEncoder().encode( + 'data: {"choices":[{"delta":{"content":"first"},"finish_reason":"stop"}]}\n\ndata: [DONE]\n\n', + ), + ); + controller.close(); + }; + }, + }), + { status: 200, headers }, + ); + } + statuses.set(runId, "done"); + return new Response( + 'data: {"choices":[{"delta":{"content":"second"},"finish_reason":"stop"}]}\n\ndata: [DONE]\n\n', + { status: 200, headers }, + ); + }, + }); + const environment = await provider.create({ profile: { name: "worker" } }); + const session = environment.session!("concurrent-session"); + const first = session.prompt({ + prompt: "first", + executionId: "concurrent-a", + }); + while (!releaseFirst) await Promise.resolve(); + const second = await session.prompt({ + prompt: "second", + executionId: "concurrent-b", + }); + releaseFirst(); + const firstResult = await first; + + expect(firstResult).toMatchObject({ + text: "first", + metadata: { runId: "concurrent-a", status: "done" }, + }); + expect(second).toMatchObject({ + text: "second", + metadata: { runId: "concurrent-b", status: "done" }, + }); + }); + it("streams canonical text, tool, usage, and result events", async () => { let body: Record | undefined; const provider = createCliBridgeProvider({ @@ -182,16 +622,28 @@ describe("createCliBridgeProvider", () => { it("enforces a configured response-header timeout", async () => { let delayedResponse: ReturnType | undefined; const server = createServer((request, response) => { + const runId = decodeURIComponent( + request.url?.split("/")[3]?.split("?")[0] ?? "", + ); if (request.url?.endsWith("/cancel")) { response.writeHead(200, { "content-type": "application/json" }); response.end( - '{"cancelled":true,"cancel_requested":true,"terminal":true,"run":{"id":"timeout-run","status":"cancelled","terminal":true}}', + JSON.stringify({ + cancelled: true, + cancel_requested: true, + terminal: true, + run: { id: runId, status: "cancelled", terminal: true }, + }), ); return; } if (request.url?.startsWith("/v1/runs/")) { response.writeHead(200, { "content-type": "application/json" }); - response.end('{"id":"timeout-run","status":"running","terminal":false}'); + response.end(JSON.stringify({ + id: runId, + status: "running", + terminal: false, + })); return; } delayedResponse = setTimeout(() => { @@ -227,16 +679,28 @@ describe("createCliBridgeProvider", () => { it("enforces a configured response-body idle timeout", async () => { let delayedBody: ReturnType | undefined; const server = createServer((request, response) => { + const runId = decodeURIComponent( + request.url?.split("/")[3]?.split("?")[0] ?? "", + ); if (request.url?.endsWith("/cancel")) { response.writeHead(200, { "content-type": "application/json" }); response.end( - '{"cancelled":true,"cancel_requested":true,"terminal":true,"run":{"id":"timeout-run","status":"cancelled","terminal":true}}', + JSON.stringify({ + cancelled: true, + cancel_requested: true, + terminal: true, + run: { id: runId, status: "cancelled", terminal: true }, + }), ); return; } if (request.url?.startsWith("/v1/runs/")) { response.writeHead(200, { "content-type": "application/json" }); - response.end('{"id":"timeout-run","status":"running","terminal":false}'); + response.end(JSON.stringify({ + id: runId, + status: "running", + terminal: false, + })); return; } response.writeHead(200, { "content-type": "text/event-stream" }); diff --git a/packages/agent-provider-cli-bridge/src/index.ts b/packages/agent-provider-cli-bridge/src/index.ts index 5868211..0766e6a 100644 --- a/packages/agent-provider-cli-bridge/src/index.ts +++ b/packages/agent-provider-cli-bridge/src/index.ts @@ -4,17 +4,22 @@ import type { AgentEnvironmentCapabilities, AgentEnvironmentEvent, AgentEnvironmentProvider, + AgentSession, + AgentSessionRef, + AgentSessionStatus, AgentProfileRef, AgentTurnInput, + AgentTurnResult, CreateAgentEnvironmentInput, } from "@tangle-network/agent-interface/environment-provider"; -import type { - AgentProfile, - InputPart, - MessagePartUpdatedEvent, - TextPart, - TokenUsage, - ToolPart, +import { + type AgentProfile, + type InputPart, + type MessagePartUpdatedEvent, + snapshotAgentProfile, + type TextPart, + type TokenUsage, + type ToolPart, } from "@tangle-network/agent-interface"; import { Agent, fetch as undiciFetch } from "undici"; @@ -50,9 +55,19 @@ export function createCliBridgeProvider(options: CliBridgeProviderOptions): Agen name, capabilities: () => options.capabilities ?? defaultCliBridgeCapabilities(), async create(input) { + if (typeof input.profile === "string") { + throw new Error( + `createCliBridgeProvider requires an inline AgentProfile; named profile "${input.profile}" is unsupported`, + ); + } + const environmentInput: CreateAgentEnvironmentInput = { + ...input, + profile: snapshotAgentProfile(input.profile), + }; const transport = createTransport(options); const environmentId = input.idempotencyKey ?? crypto.randomUUID(); const runs = new Map(); + const sessions = new Map(); const readers = new Set(); let destroyed = false; let closePromise: Promise | undefined; @@ -60,13 +75,20 @@ export function createCliBridgeProvider(options: CliBridgeProviderOptions): Agen turn: AgentTurnInput, ): AsyncIterable { if (destroyed) throw new Error("cli-bridge environment is destroyed"); - yield* streamTrackedCliBridgeTurn( + const prepared = prepareCliBridgeRun( options, - input, + environmentInput, turn, - transport, environmentId, + false, + ); + yield* streamTrackedCliBridgeTurn( + options, + environmentInput, + prepared, + transport, runs, + sessions, readers, ); }; @@ -76,6 +98,39 @@ export function createCliBridgeProvider(options: CliBridgeProviderOptions): Agen ...(input.name ? { name: input.name } : {}), status: async () => (destroyed ? "stopped" : "running"), stream, + async dispatch(turn: AgentTurnInput): Promise { + if (destroyed) throw new Error("cli-bridge environment is destroyed"); + const prepared = prepareCliBridgeRun( + options, + environmentInput, + turn, + environmentId, + true, + ); + return dispatchCliBridgeTurn( + options, + environmentInput, + prepared, + transport, + name, + runs, + sessions, + ); + }, + session(id: string): AgentSession { + return createCliBridgeSession({ + id, + providerName: name, + options, + environmentInput, + environmentId, + transport, + runs, + sessions, + readers, + isDestroyed: () => destroyed, + }); + }, placement: async () => ({ kind: options.defaultExecution?.kind === "sandbox" ? "sandbox" : "local", providerMetadata: { baseUrl: options.baseUrl }, @@ -106,6 +161,7 @@ export function createCliBridgeProvider(options: CliBridgeProviderOptions): Agen ); } await transport.close(); + sessions.clear(); })(); closePromise = attempt; try { @@ -124,31 +180,117 @@ export function createCliBridgeProvider(options: CliBridgeProviderOptions): Agen interface CliBridgeRun { readonly id: string; + readonly sessionId?: string; + readonly turnId: string; + readonly requestBody: string; readonly readers: Set; + requestDigest?: string; cancellation?: Promise; } +interface CliBridgeSessionState { + readonly id: string; + current: CliBridgeRun; +} + interface CliBridgeRunSnapshot { readonly id: string; readonly status: "running" | "done" | "error" | "cancelled"; readonly terminal: boolean; } -async function* streamTrackedCliBridgeTurn( +interface PreparedCliBridgeRun { + readonly run: CliBridgeRun; + readonly turn: AgentTurnInput; +} + +function prepareCliBridgeRun( options: CliBridgeProviderOptions, environmentInput: CreateAgentEnvironmentInput, originalTurn: AgentTurnInput, - transport: CliBridgeTransport, environmentId: string, + requireSession: boolean, +): PreparedCliBridgeRun { + const turnId = originalTurn.turnId ?? crypto.randomUUID(); + const runId = cliBridgeRunId(environmentId, originalTurn, turnId); + const sessionId = originalTurn.sessionId ?? (requireSession ? runId : undefined); + const turn = { + ...originalTurn, + turnId, + ...(sessionId ? { sessionId } : {}), + }; + return { + turn, + run: { + id: runId, + ...(sessionId ? { sessionId } : {}), + turnId, + requestBody: JSON.stringify( + toChatCompletionsBody(options, environmentInput, turn, runId), + ), + readers: new Set(), + }, + }; +} + +function bindCliBridgeRun( + prepared: CliBridgeRun, runs: Map, +): CliBridgeRun | undefined { + const previous = runs.get(prepared.id); + runs.set(prepared.id, prepared); + return previous; +} + +function restoreCliBridgeRun( + run: CliBridgeRun, + previous: CliBridgeRun | undefined, + runs: Map, +): void { + if (runs.get(run.id) !== run) return; + if (previous) { + runs.set(run.id, previous); + } else { + runs.delete(run.id); + } +} + +function bindCliBridgeSession( + run: CliBridgeRun, + sessions: Map, +): CliBridgeRun | undefined { + if (!run.sessionId) return undefined; + const previous = sessions.get(run.sessionId)?.current; + sessions.set(run.sessionId, { id: run.sessionId, current: run }); + return previous; +} + +function restoreCliBridgeSession( + run: CliBridgeRun, + previous: CliBridgeRun | undefined, + sessions: Map, +): void { + if (!run.sessionId || sessions.get(run.sessionId)?.current !== run) return; + if (previous) { + sessions.set(run.sessionId, { id: run.sessionId, current: previous }); + } else { + sessions.delete(run.sessionId); + } +} + +async function* streamTrackedCliBridgeTurn( + options: CliBridgeProviderOptions, + environmentInput: CreateAgentEnvironmentInput, + prepared: PreparedCliBridgeRun, + transport: CliBridgeTransport, + runs: Map, + sessions: Map, readers: Set, ): AsyncIterable { + const originalTurn = prepared.turn; if (originalTurn.detach) { throw new Error("cli-bridge provider does not support detached turns"); } - const sessionId = originalTurn.sessionId; - const turnId = originalTurn.turnId ?? crypto.randomUUID(); - const runId = cliBridgeRunId(environmentId, originalTurn, turnId); const controller = new AbortController(); const signals = [ originalTurn.signal, @@ -157,19 +299,13 @@ async function* streamTrackedCliBridgeTurn( ].filter((signal): signal is AbortSignal => signal !== undefined); const turn = { ...originalTurn, - turnId, ...(signals.length > 0 ? { signal: AbortSignal.any(signals) } : {}), }; - const requestBody = JSON.stringify( - toChatCompletionsBody(options, environmentInput, turn, runId), - ); - const run = runs.get(runId) ?? { - id: runId, - readers: new Set(), - }; + const run = prepared.run; + const previousRun = bindCliBridgeRun(run, runs); + const previousSessionRun = bindCliBridgeSession(run, sessions); run.readers.add(controller); readers.add(controller); - runs.set(runId, run); let drained = false; let threw = false; @@ -177,11 +313,12 @@ async function* streamTrackedCliBridgeTurn( for await (const event of streamCliBridgeTurn( options, turn, - requestBody, + run.requestBody, transport, - runId, + run.id, originalTurn.lastEventId, signals.length > 0 ? AbortSignal.any(signals) : undefined, + (response) => captureCliBridgeRunIdentity(response, run, false), )) { yield event; } @@ -190,7 +327,8 @@ async function* streamTrackedCliBridgeTurn( } catch (error) { threw = true; if (error instanceof CliBridgeRequestRejectedError) { - if (runs.get(run.id) === run) runs.delete(run.id); + restoreCliBridgeRun(run, previousRun, runs); + restoreCliBridgeSession(run, previousSessionRun, sessions); throw error; } let snapshot: CliBridgeRunSnapshot | null | undefined; @@ -218,6 +356,405 @@ async function* streamTrackedCliBridgeTurn( } } +async function dispatchCliBridgeTurn( + options: CliBridgeProviderOptions, + environmentInput: CreateAgentEnvironmentInput, + prepared: PreparedCliBridgeRun, + transport: CliBridgeTransport, + providerName: string, + runs: Map, + sessions: Map, +): Promise { + const run = prepared.run; + const previousRun = bindCliBridgeRun(run, runs); + const previousSessionRun = bindCliBridgeSession(run, sessions); + const signals = [ + prepared.turn.signal, + environmentInput.signal, + ].filter((signal): signal is AbortSignal => signal !== undefined); + const signal = signals.length > 0 ? AbortSignal.any(signals) : undefined; + let accepted = false; + let responseBody: AsyncIterable | undefined; + try { + const response = await transport.fetch( + `${trimSlash(options.baseUrl)}/v1/chat/completions`, + { + method: "POST", + headers: { + ...requestHeaders(options), + accept: "text/event-stream", + ...(run.sessionId ? { "x-session-id": run.sessionId } : {}), + }, + body: run.requestBody, + ...(signal ? { signal } : {}), + }, + ); + if (!response.ok) { + let detail = "request rejected"; + try { + detail = await response.text(); + } catch { + // The HTTP status already proves this request was rejected. + } + throw new CliBridgeRequestRejectedError(response.status, detail); + } + accepted = true; + if (!response.body) throw new Error("cli-bridge response body is empty"); + responseBody = response.body; + captureCliBridgeRunIdentity(response, run, true); + await detachCliBridgeReader(responseBody); + responseBody = undefined; + return { + id: run.sessionId!, + provider: providerName, + metadata: { + runId: run.id, + requestDigest: run.requestDigest, + }, + }; + } catch (error) { + let failure = error; + if (responseBody) { + try { + await detachCliBridgeReader(responseBody); + } catch (detachError) { + failure = new AggregateError( + [failure, detachError], + `cli-bridge dispatch "${run.id}" failed and its reader did not detach`, + ); + } + } + if (failure instanceof CliBridgeRequestRejectedError) { + restoreCliBridgeRun(run, previousRun, runs); + restoreCliBridgeSession(run, previousSessionRun, sessions); + throw failure; + } + if (accepted || signal?.aborted) { + try { + await cancelCliBridgeRun(options, transport, run); + if (runs.get(run.id) === run) runs.delete(run.id); + } catch (cancellationError) { + throw new AggregateError( + [failure, cancellationError], + `cli-bridge dispatch "${run.id}" failed and cancellation was not confirmed`, + ); + } + } else { + restoreCliBridgeRun(run, previousRun, runs); + } + restoreCliBridgeSession(run, previousSessionRun, sessions); + throw failure; + } +} + +interface CreateCliBridgeSessionArgs { + readonly id: string; + readonly providerName: string; + readonly options: CliBridgeProviderOptions; + readonly environmentInput: CreateAgentEnvironmentInput; + readonly environmentId: string; + readonly transport: CliBridgeTransport; + readonly runs: Map; + readonly sessions: Map; + readonly readers: Set; + readonly isDestroyed: () => boolean; +} + +function createCliBridgeSession(args: CreateCliBridgeSessionArgs): AgentSession { + const currentRun = (): CliBridgeRun | undefined => { + if (args.isDestroyed()) throw new Error("cli-bridge environment is destroyed"); + return args.sessions.get(args.id)?.current; + }; + const requireCurrentRun = (): CliBridgeRun => { + const run = currentRun(); + if (!run) throw new Error(`cli-bridge session "${args.id}" has no run`); + return run; + }; + + return { + id: args.id, + async status(): Promise { + const run = currentRun(); + if (!run) return null; + const snapshot = await getCliBridgeRun(args.options, args.transport, run.id); + if (!snapshot) return null; + if (snapshot.terminal && args.runs.get(run.id) === run) { + args.runs.delete(run.id); + } + return agentSessionStatusFromRun(snapshot); + }, + async *events(options): AsyncIterable { + const run = requireCurrentRun(); + yield* streamCliBridgeSessionEvents( + args.options, + args.environmentInput, + run, + args.transport, + args.runs, + args.readers, + options, + ); + }, + async result(): Promise { + const run = requireCurrentRun(); + return collectCliBridgeTurnResult( + streamCliBridgeSessionEvents( + args.options, + args.environmentInput, + run, + args.transport, + args.runs, + args.readers, + { since: "0" }, + ), + run, + args.options, + args.transport, + ); + }, + async prompt(input: AgentTurnInput): Promise { + if (input.sessionId && input.sessionId !== args.id) { + throw new Error( + `cli-bridge session "${args.id}" cannot prompt session "${input.sessionId}"`, + ); + } + const prepared = prepareCliBridgeRun( + args.options, + args.environmentInput, + { + ...input, + sessionId: args.id, + }, + args.environmentId, + false, + ); + const result = await collectCliBridgeTurnResult( + streamTrackedCliBridgeTurn( + args.options, + args.environmentInput, + prepared, + args.transport, + args.runs, + args.sessions, + args.readers, + ), + prepared.run, + args.options, + args.transport, + ); + return { ...result, sessionId: args.id }; + }, + async cancel(): Promise { + const run = requireCurrentRun(); + await cancelCliBridgeRun(args.options, args.transport, run); + if (args.runs.get(run.id) === run) args.runs.delete(run.id); + }, + }; +} + +async function* streamCliBridgeSessionEvents( + options: CliBridgeProviderOptions, + environmentInput: CreateAgentEnvironmentInput, + run: CliBridgeRun, + transport: CliBridgeTransport, + runs: Map, + readers: Set, + eventOptions?: { since?: string; signal?: AbortSignal }, +): AsyncIterable { + const controller = new AbortController(); + const signals = [ + eventOptions?.signal, + environmentInput.signal, + controller.signal, + ].filter((signal): signal is AbortSignal => signal !== undefined); + const signal = signals.length > 0 ? AbortSignal.any(signals) : undefined; + run.readers.add(controller); + readers.add(controller); + let drained = false; + try { + yield* streamCliBridgeTurn( + options, + { + ...(run.sessionId ? { sessionId: run.sessionId } : {}), + turnId: run.turnId, + }, + run.requestBody, + transport, + run.id, + eventOptions?.since ?? "0", + signal, + (response) => captureCliBridgeRunIdentity(response, run, false), + ); + drained = true; + if (runs.get(run.id) === run) runs.delete(run.id); + } finally { + if (!drained) { + controller.abort( + new DOMException("cli-bridge session event reader detached", "AbortError"), + ); + } + run.readers.delete(controller); + readers.delete(controller); + } +} + +async function collectCliBridgeTurnResult( + source: AsyncIterable, + run: CliBridgeRun, + options: CliBridgeProviderOptions, + transport: CliBridgeTransport, +): Promise { + const events: AgentEnvironmentEvent[] = []; + let text = ""; + let usage: TokenUsage | undefined; + let streamError: unknown; + try { + for await (const event of source) { + events.push(event); + const finalText = event.data.finalText; + if (typeof finalText === "string") { + text = finalText; + } else if (typeof event.data.delta === "string") { + text += event.data.delta; + } + usage = addTokenUsage(usage, event.usage); + } + } catch (error) { + streamError = error; + } + + const snapshot = await getCliBridgeRun(options, transport, run.id); + if (!snapshot) { + if (streamError) throw streamError; + throw new Error(`cli-bridge lost run "${run.id}" before reading its result`); + } + if (!snapshot.terminal) { + if (streamError) throw streamError; + throw new Error(`cli-bridge run "${run.id}" has no terminal result`); + } + const success = snapshot.status === "done" && streamError === undefined; + return { + text, + success, + ...(!success + ? { + error: streamError instanceof Error + ? streamError.message + : `cli-bridge run ended ${snapshot.status}`, + } + : {}), + ...(run.sessionId ? { sessionId: run.sessionId } : {}), + ...(usage ? { usage } : {}), + metadata: { + runId: run.id, + status: snapshot.status, + ...(run.requestDigest ? { requestDigest: run.requestDigest } : {}), + }, + events, + }; +} + +function agentSessionStatusFromRun( + snapshot: CliBridgeRunSnapshot, +): AgentSessionStatus { + if (snapshot.status === "done") return "completed"; + if (snapshot.status === "error") return "failed"; + if (snapshot.status === "cancelled") return "cancelled"; + return "running"; +} + +function addTokenUsage( + current: TokenUsage | undefined, + next: TokenUsage | undefined, +): TokenUsage | undefined { + if (!next) return current; + if (!current) return { ...next }; + return { + inputTokens: current.inputTokens + next.inputTokens, + outputTokens: current.outputTokens + next.outputTokens, + ...(current.totalTokens !== undefined || next.totalTokens !== undefined + ? { totalTokens: (current.totalTokens ?? 0) + (next.totalTokens ?? 0) } + : {}), + ...(current.cacheReadInputTokens !== undefined || + next.cacheReadInputTokens !== undefined + ? { + cacheReadInputTokens: + (current.cacheReadInputTokens ?? 0) + + (next.cacheReadInputTokens ?? 0), + } + : {}), + ...(current.cacheCreationInputTokens !== undefined || + next.cacheCreationInputTokens !== undefined + ? { + cacheCreationInputTokens: + (current.cacheCreationInputTokens ?? 0) + + (next.cacheCreationInputTokens ?? 0), + } + : {}), + ...(current.reasoningTokens !== undefined || next.reasoningTokens !== undefined + ? { + reasoningTokens: + (current.reasoningTokens ?? 0) + (next.reasoningTokens ?? 0), + } + : {}), + ...(current.cost !== undefined || next.cost !== undefined + ? { cost: (current.cost ?? 0) + (next.cost ?? 0) } + : {}), + }; +} + +function captureCliBridgeRunIdentity( + response: CliBridgeResponse, + run: CliBridgeRun, + required: boolean, +): void { + const responseRunId = response.headers.get("x-run-id"); + const requestDigest = response.headers.get("x-run-request-digest"); + if (responseRunId !== null && responseRunId !== run.id) { + throw new Error( + `cli-bridge accepted run "${responseRunId}" for requested run "${run.id}"`, + ); + } + if ((required || run.requestDigest !== undefined) && responseRunId === null) { + throw new Error("cli-bridge response omitted X-Run-Id"); + } + if (required && requestDigest === null) { + throw new Error("cli-bridge dispatch response omitted X-Run-Request-Digest"); + } + if ( + requestDigest !== null && + run.requestDigest !== undefined && + requestDigest !== run.requestDigest + ) { + throw new Error( + `cli-bridge changed request digest for run "${run.id}" from "${run.requestDigest}" to "${requestDigest}"`, + ); + } + if (run.requestDigest !== undefined && requestDigest === null) { + throw new Error( + `cli-bridge replay response omitted X-Run-Request-Digest for run "${run.id}"`, + ); + } + if (requestDigest !== null) run.requestDigest = requestDigest; +} + +async function detachCliBridgeReader( + body: AsyncIterable, +): Promise { + const cancellable = body as AsyncIterable & { + cancel?: () => Promise; + }; + if (cancellable.cancel) { + await cancellable.cancel(); + return; + } + const iterator = body[Symbol.asyncIterator](); + if (!iterator.return) { + throw new Error("cli-bridge response body cannot detach its reader"); + } + await iterator.return(); +} + async function cancelCliBridgeRun( options: CliBridgeProviderOptions, transport: CliBridgeTransport, @@ -237,6 +774,7 @@ async function cancelCliBridgeRun( throw new Error(`cli-bridge cancel ${response.status}: ${await response.text()}`); } let snapshot: CliBridgeRunSnapshot | null = cancelSnapshot(await response.text()); + assertCliBridgeRunSnapshotIdentity(snapshot, run.id); const waitBudgetMs = options.cancelWaitMs ?? 30_000; const deadline = Date.now() + waitBudgetMs; while (!snapshot.terminal) { @@ -287,7 +825,20 @@ async function getCliBridgeRun( if (!response.ok) { throw new Error(`cli-bridge run status ${response.status}: ${await response.text()}`); } - return runSnapshot(await response.text()); + const snapshot = runSnapshot(await response.text()); + assertCliBridgeRunSnapshotIdentity(snapshot, runId); + return snapshot; +} + +function assertCliBridgeRunSnapshotIdentity( + snapshot: CliBridgeRunSnapshot, + expectedRunId: string, +): void { + if (snapshot.id !== expectedRunId) { + throw new Error( + `cli-bridge returned run "${snapshot.id}" for requested run "${expectedRunId}"`, + ); + } } async function* streamCliBridgeTurn( @@ -298,6 +849,7 @@ async function* streamCliBridgeTurn( runId: string, lastEventId?: string, signal?: AbortSignal, + onAccepted?: (response: CliBridgeResponse) => void, ): AsyncIterable { const response = await transport.fetch(`${trimSlash(options.baseUrl)}/v1/chat/completions`, { method: "POST", @@ -319,6 +871,7 @@ async function* streamCliBridgeTurn( } throw new CliBridgeRequestRejectedError(response.status, detail); } + onAccepted?.(response); if (!response.body) throw new Error("cli-bridge response body is empty"); let text = ""; @@ -326,6 +879,7 @@ async function* streamCliBridgeTurn( const messageId = turn.turnId ?? `${sessionId}:assistant`; const emittedToolCalls = new Set(); let completed = false; + let sawUsage = false; let terminalCursor: string | undefined; for await (const frame of parseSse(response.body)) { if (frame.data === "[DONE]") continue; @@ -347,6 +901,7 @@ async function* streamCliBridgeTurn( const nextUsage = usageFromOpenAi(parsed.usage); const frameEvents: AgentEnvironmentEvent[] = []; if (nextUsage) { + sawUsage = true; frameEvents.push({ type: "usage", data: {}, usage: nextUsage }); } if (chunk) { @@ -429,7 +984,11 @@ async function* streamCliBridgeTurn( requestBody, transport, signal, + onAccepted, ); + if (result.usage && !sawUsage) { + yield { type: "usage", data: {}, usage: result.usage }; + } yield { type: "result", data: { @@ -454,7 +1013,12 @@ async function readFullCliBridgeResult( requestBody: string, transport: CliBridgeTransport, signal?: AbortSignal, -): Promise<{ text: string; finishReason: string }> { + onAccepted?: (response: CliBridgeResponse) => void, +): Promise<{ + text: string; + finishReason: string; + usage?: TokenUsage; +}> { const body = safeJson(requestBody); if (!body) throw new Error("cli-bridge replay request is not valid JSON"); const response = await transport.fetch(`${trimSlash(options.baseUrl)}/v1/chat/completions`, { @@ -466,6 +1030,7 @@ async function readFullCliBridgeResult( if (!response.ok) { throw new Error(`cli-bridge replay result ${response.status}: ${await response.text()}`); } + onAccepted?.(response); const parsed = safeJson(await response.text()); if (parsed?.error && typeof parsed.error === "object") { const error = parsed.error as Record; @@ -487,7 +1052,12 @@ async function readFullCliBridgeResult( if (choice.finish_reason === "error" || choice.finish_reason === "timeout") { throw new Error(`cli-bridge replay result ended ${choice.finish_reason}`); } - return { text: message.content, finishReason: choice.finish_reason }; + const usage = usageFromOpenAi(parsed?.usage); + return { + text: message.content, + finishReason: choice.finish_reason, + ...(usage ? { usage } : {}), + }; } interface CliBridgeTransport { @@ -506,6 +1076,9 @@ interface CliBridgeResponse { readonly ok: boolean; readonly status: number; readonly body: AsyncIterable | null; + readonly headers: { + get(name: string): string | null; + }; text(): Promise; } @@ -548,12 +1121,15 @@ function toChatCompletionsBody( const profile = inlineProfile(environmentInput.profile); return { model: resolveBridgeModel(options, environmentInput, turn, profile), - messages: messagesFromTurn(turn, profile), + messages: messagesFromTurn(turn), stream: true, ...(turn.sessionId ? { session_id: turn.sessionId } : {}), run_id: runId, ...(options.defaultMode ? { mode: options.defaultMode } : {}), ...(profile ? { agent_profile: profile } : {}), + ...(profile?.model?.reasoningEffort + ? { effort: profile.model.reasoningEffort } + : {}), ...(environmentInput.env ? { env: environmentInput.env } : {}), ...(environmentInput.workspace?.cwd ? { cwd: environmentInput.workspace.cwd } : {}), ...(executionFromInput(options, environmentInput) ? { execution: executionFromInput(options, environmentInput) } : {}), @@ -587,12 +1163,8 @@ function resolveBridgeModel( return `${harness}/${model}`; } -function messagesFromTurn(turn: AgentTurnInput, profile: AgentProfile | undefined): Array> { - const messages: Array> = []; - const systemPrompt = profile?.prompt?.systemPrompt; - if (systemPrompt) messages.push({ role: "system", content: systemPrompt }); - messages.push({ role: "user", content: contentFromTurn(turn) }); - return messages; +function messagesFromTurn(turn: AgentTurnInput): Array> { + return [{ role: "user", content: contentFromTurn(turn) }]; } function contentFromTurn(turn: AgentTurnInput): string | InputPart[] { @@ -808,7 +1380,7 @@ export function defaultCliBridgeCapabilities(): AgentEnvironmentCapabilities { runtimeUpdate: false, validation: false, }, - streaming: { live: true, replay: true, detach: false, turnIdempotency: true }, + streaming: { live: true, replay: true, detach: true, turnIdempotency: true }, sessions: { continue: true, list: false, messages: false }, workspace: { read: false, write: false, exec: false, git: false, upload: false, download: false }, branching: { checkpoint: false, fork: false },