diff --git a/sdk/typescript/README.md b/sdk/typescript/README.md index eb3c9f18e8..8fc9e07816 100644 --- a/sdk/typescript/README.md +++ b/sdk/typescript/README.md @@ -13,6 +13,8 @@ The slice validates the intended public object model: same connection and existing Agent Runtime owner; - `Query` is an ordered async stream with idempotent cancellation, cached final `Result`, and explicit close semantics; +- the same stream reports safe Tool lifecycle facts and permission requests; + `Query.respondPermission()` supports allow once, allow always, or reject; - protocol and process failures use `SdkError`, including outcome certainty. Lifecycle cleanup is bounded. The Windows Host contains descendants in a @@ -26,9 +28,21 @@ existing `agent-runtime::sdk` API. ## Repository usage -Build `bitfun-sdk-host`, then pass its absolute path and one process-lifetime -model configuration while the platform-native package layout is still pending. -The Host path must be explicit in this slice: +Build the private SDK and `bitfun-sdk-host`, then stage that already-built Host +into the local package. This does not install BitFun or publish anything: + +```bash +cargo build -p bitfun-sdk-host-app +pnpm --dir sdk/typescript build +pnpm --dir sdk/typescript stage:host -- ../../target/debug/bitfun-sdk-host.exe +``` + +Use `bitfun-sdk-host` without `.exe` on macOS and Linux. The staging command +copies only the current platform's executable into the package build under +`dist/sdk/typescript/native/-/`. + +The trusted application then supplies one process-lifetime model configuration; +the SDK finds and manages the staged native Host automatically: ```typescript import { AgentClient } from "@bitfun/agent-sdk"; @@ -36,7 +50,6 @@ import { AgentClient } from "@bitfun/agent-sdk"; const apiKey = await trustedSecretStore.read("openai"); await using client = await AgentClient.start({ cwd: process.cwd(), - hostPath: "/absolute/path/to/bitfun-sdk-host", model: { provider: "openai", model: "gpt-5.4", @@ -47,22 +60,37 @@ await using client = await AgentClient.start({ await using query = await client.query({ prompt: "Summarize this repository" }); for await (const item of query) { - if (item.type === "assistant_text_delta") { - process.stdout.write(item.text); + switch (item.type) { + case "assistant_text_delta": + process.stdout.write(item.text); + break; + case "tool_event": + console.log(item.toolName, item.status); + break; + case "permission_request": + await query.respondPermission(item.requestId, { decision: "allow_once" }); + break; } } const result = await query.result(); ``` -This repository-local package is private and unpublished. Node 24.14.1 and Bun -1.4.0 are the locally verified runners for this slice; they are not bundled -executables or a final minimum-version policy. The eventual installable package -must bundle or resolve a matching signed Host; it must not require a separately -installed BitFun CLI. +An explicit absolute `hostPath` remains available as a development override. +The SDK never searches `PATH` or an environment variable for the Host. + +This repository-local package is private and unpublished. Node 24.14.1 is +locally verified for this slice. Bun uses the same ESM build but remains a +release-verification target when a Bun runner is available; neither runtime is +a bundled executable or a final minimum-version policy. `pnpm --dir sdk/typescript pack` +can produce a local tarball containing the staged Host. An application installs +that tarball as an ordinary dependency; it does not install BitFun or a CLI +separately. This PR does not publish the package. A future registry release +still needs platform packages, signing, and release verification. Browser and mobile runtimes cannot launch the local native Host. Custom -functions, permission and user-input callbacks, structured output, usage, -Session resume, Python support, and native package staging remain deferred. +functions, general user-input callbacks, structured output, usage, Session +resume, Python support, platform package publication, signing, and downloads +remain deferred. ## Development @@ -71,6 +99,7 @@ pnpm --dir sdk/typescript test pnpm --dir sdk/typescript type-check pnpm --dir sdk/typescript smoke:node pnpm --dir sdk/typescript smoke:bun +pnpm --dir sdk/typescript smoke:consumer ``` The internal TypeScript wire bindings are generated from the Rust SDK Host diff --git a/sdk/typescript/package.json b/sdk/typescript/package.json index c6dd7d992c..c32c7ab0b5 100644 --- a/sdk/typescript/package.json +++ b/sdk/typescript/package.json @@ -14,13 +14,16 @@ "dist/sdk/typescript/src/*.d.ts", "dist/sdk/typescript/src/*.js", "dist/sdk/typescript/src/internal/*.js", + "dist/sdk/typescript/native/**", "dist/src/crates/adapters/transport/typescript/src/*.js", "README.md" ], "scripts": { "build": "pnpm run generate:wire && tsc -p tsconfig.json", "generate:wire": "node scripts/generate-wire.mjs", + "stage:host": "node scripts/stage-host.mjs", "smoke:bun": "bun test/real-host-smoke.mjs", + "smoke:consumer": "node test/local-package-consumer.mjs", "smoke:node": "node test/real-host-smoke.mjs", "test": "pnpm run build && node --test scripts/*.test.mjs dist/src/crates/adapters/transport/typescript/test/**/*.test.js dist/sdk/typescript/test/**/*.test.js", "type-check": "pnpm run generate:wire && tsc -p tsconfig.json --noEmit" diff --git a/sdk/typescript/scripts/generate-wire.mjs b/sdk/typescript/scripts/generate-wire.mjs index 04b4a9f56b..61b1b498db 100644 --- a/sdk/typescript/scripts/generate-wire.mjs +++ b/sdk/typescript/scripts/generate-wire.mjs @@ -49,6 +49,8 @@ const requiredTypes = [ "HostCapabilities", "InitializeParams", "InitializeResult", + "PermissionRespondParams", + "PermissionRespondResult", "QueryCancelParams", "QueryCancelResult", "QueryEventParams", diff --git a/sdk/typescript/scripts/generated-wire-runtime.test.mjs b/sdk/typescript/scripts/generated-wire-runtime.test.mjs index 784a4ac7a0..59ba8cb9e8 100644 --- a/sdk/typescript/scripts/generated-wire-runtime.test.mjs +++ b/sdk/typescript/scripts/generated-wire-runtime.test.mjs @@ -43,7 +43,7 @@ test("Rust wire export produces executable validators for every type", async () } const initializeResult = { - protocolVersion: 2, + protocolVersion: 3, runtimeVersion: "0.1.0", stability: "not_delivered", capabilities: { @@ -53,10 +53,11 @@ test("Rust wire export produces executable validators for every type", async () queryCancel: true, sessionClose: true, eventStream: true, + toolEvents: true, structuredOutput: false, usage: false, customTools: false, - permissionCallbacks: false, + permissionResponses: true, hooks: false, mcpConfiguration: false, prestartedTransport: false, diff --git a/sdk/typescript/scripts/stage-host.mjs b/sdk/typescript/scripts/stage-host.mjs new file mode 100644 index 0000000000..8e0914cb9f --- /dev/null +++ b/sdk/typescript/scripts/stage-host.mjs @@ -0,0 +1,47 @@ +import { chmod, copyFile, mkdir, stat } from "node:fs/promises"; +import { dirname, resolve } from "node:path"; +import { pathToFileURL } from "node:url"; + +export async function stageHost(source, destination) { + let sourceMetadata; + try { + sourceMetadata = await stat(source); + } catch (cause) { + throw new Error(`Host source was not found: ${source}`, { cause }); + } + if (!sourceMetadata.isFile()) { + throw new Error(`Host source must be a file: ${source}`); + } + + await mkdir(dirname(destination), { recursive: true }); + await copyFile(source, destination); + if (process.platform !== "win32") { + await chmod(destination, 0o755); + } +} + +async function main() { + const [source, ...extra] = process.argv.slice(2); + if (source === undefined || extra.length > 0) { + throw new Error("Usage: pnpm stage:host -- "); + } + + const { packageHostPath } = await import( + "../dist/sdk/typescript/src/internal/host-path.js" + ); + const destination = packageHostPath(process.platform, process.arch); + await stageHost(resolve(source), destination); + process.stdout.write(`Staged BitFun SDK Host at ${destination}\n`); +} + +if ( + process.argv[1] !== undefined && + import.meta.url === pathToFileURL(process.argv[1]).href +) { + try { + await main(); + } catch (error) { + process.stderr.write(`${error instanceof Error ? error.message : String(error)}\n`); + process.exitCode = 1; + } +} diff --git a/sdk/typescript/scripts/stage-host.test.mjs b/sdk/typescript/scripts/stage-host.test.mjs new file mode 100644 index 0000000000..e69c407f49 --- /dev/null +++ b/sdk/typescript/scripts/stage-host.test.mjs @@ -0,0 +1,53 @@ +import assert from "node:assert/strict"; +import { mkdtemp, mkdir, readFile, rm, stat, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import test from "node:test"; + +test("stageHost copies an already-built Host into its package destination", async () => { + const stageHost = await loadStageHost(); + const root = await mkdtemp(join(tmpdir(), "bitfun-sdk-stage-host-")); + const source = join(root, "source-host"); + const destination = join(root, "package", "native", "host"); + const contents = Buffer.from("local-host-fixture\n", "utf8"); + try { + await writeFile(source, contents, { mode: 0o600 }); + + await stageHost(source, destination); + + assert.deepEqual(await readFile(destination), contents); + if (process.platform !== "win32") { + assert.notEqual((await stat(destination)).mode & 0o111, 0); + } + } finally { + await rm(root, { recursive: true, force: true }); + } +}); + +test("stageHost rejects a directory source", async () => { + const stageHost = await loadStageHost(); + const root = await mkdtemp(join(tmpdir(), "bitfun-sdk-stage-host-invalid-")); + try { + const source = join(root, "source-directory"); + await mkdir(source); + await assert.rejects( + stageHost(source, join(root, "destination")), + /Host source must be a file/, + ); + } finally { + await rm(root, { recursive: true, force: true }); + } +}); + +async function loadStageHost() { + try { + const module = await import("./stage-host.mjs"); + assert.equal(typeof module.stageHost, "function"); + return module.stageHost; + } catch (error) { + if (error?.code === "ERR_MODULE_NOT_FOUND") { + assert.fail("stage-host.mjs must export stageHost"); + } + throw error; + } +} diff --git a/sdk/typescript/src/client.ts b/sdk/typescript/src/client.ts index bd8e4d0d33..9f929599f9 100644 --- a/sdk/typescript/src/client.ts +++ b/sdk/typescript/src/client.ts @@ -1,7 +1,6 @@ -import { isAbsolute } from "node:path"; - import type { InitializeResult, QueryStartParams, QueryStartResult } from "./internal/wire/index.js"; import type { JsonRpcConnection } from "./internal/json-rpc.js"; +import { resolveHostPath } from "./internal/host-path.js"; import { SdkError } from "./errors.js"; import { Query } from "./query.js"; import { Session, Sessions } from "./session.js"; @@ -32,16 +31,7 @@ export class AgentClient { static async start(options: AgentClientOptions): Promise { validateModelOptions(options.model as unknown); - const hostPath = options.hostPath; - if (typeof hostPath !== "string" || !isAbsolute(hostPath)) { - throw new SdkError("SDK Host path must be an explicit absolute path", { - code: "invalid_request", - stage: "initialize", - retryable: false, - correlationId: "local:host_validation", - outcomeCertainty: "not_started", - }); - } + const hostPath = resolveHostPath(options.hostPath); const [{ createAgentClient }, { startManagedHost }] = await Promise.all([ import("./internal/client.js"), import("./internal/managed-host.js"), @@ -70,6 +60,14 @@ export class AgentClient { query: initialized.capabilities.query, sessions: initialized.capabilities.sessionCreate, cancellation: initialized.capabilities.queryCancel, + eventStream: initialized.capabilities.eventStream, + toolEvents: initialized.capabilities.toolEvents, + permissionResponses: initialized.capabilities.permissionResponses, + structuredOutput: initialized.capabilities.structuredOutput, + usage: initialized.capabilities.usage, + customTools: initialized.capabilities.customTools, + hooks: initialized.capabilities.hooks, + mcpConfiguration: initialized.capabilities.mcpConfiguration, }); this.sessions = Sessions.forClient( connection, diff --git a/sdk/typescript/src/index.ts b/sdk/typescript/src/index.ts index 54e043a9b8..1e6a0fc558 100644 --- a/sdk/typescript/src/index.ts +++ b/sdk/typescript/src/index.ts @@ -9,6 +9,10 @@ export type { AgentModelProvider, AssistantTextDelta, OutcomeCertainty, + PermissionDecision, + PermissionRequestEvent, + PermissionResponse, + PermissionSource, QueryInput, QueryStreamItem, RecoveryAction, @@ -22,4 +26,5 @@ export type { SessionLifetime, Turn, TurnInput, + ToolEvent, } from "./types.js"; diff --git a/sdk/typescript/src/internal/client.ts b/sdk/typescript/src/internal/client.ts index 5e0dc4ac8e..bcb85c6310 100644 --- a/sdk/typescript/src/internal/client.ts +++ b/sdk/typescript/src/internal/client.ts @@ -4,7 +4,7 @@ import { JsonRpcConnection } from "./json-rpc.js"; import type { HostTransport } from "./transport.js"; import type { InitializeParams, InitializeResult } from "./wire/index.js"; -const PROTOCOL_VERSION = 2; +const PROTOCOL_VERSION = 3; const DEFAULT_INITIALIZE_TIMEOUT_MS = 30_000; export async function createAgentClient( @@ -15,7 +15,10 @@ export async function createAgentClient( const params: InitializeParams = { protocolVersion: PROTOCOL_VERSION, clientInfo: { name: "@bitfun/agent-sdk", version: "0.0.0" }, - capabilities: { serverNotifications: true }, + capabilities: { + serverNotifications: true, + permissionResponses: true, + }, model: { provider: options.model.provider, model: options.model.model, diff --git a/sdk/typescript/src/internal/host-path.ts b/sdk/typescript/src/internal/host-path.ts new file mode 100644 index 0000000000..431ee1f1ee --- /dev/null +++ b/sdk/typescript/src/internal/host-path.ts @@ -0,0 +1,30 @@ +import { isAbsolute } from "node:path"; +import { fileURLToPath } from "node:url"; + +import { SdkError } from "../errors.js"; + +export function resolveHostPath(explicitPath?: string): string { + if (explicitPath === undefined) { + return packageHostPath(process.platform, process.arch); + } + if (typeof explicitPath !== "string" || !isAbsolute(explicitPath)) { + throw new SdkError("SDK Host path must be an explicit absolute path", { + code: "invalid_request", + stage: "initialize", + retryable: false, + correlationId: "local:host_validation", + outcomeCertainty: "not_started", + }); + } + return explicitPath; +} + +export function packageHostPath( + platform: NodeJS.Platform, + arch: NodeJS.Architecture, +): string { + const executable = platform === "win32" ? "bitfun-sdk-host.exe" : "bitfun-sdk-host"; + return fileURLToPath( + new URL(`../../native/${platform}-${arch}/${executable}`, import.meta.url), + ); +} diff --git a/sdk/typescript/src/internal/wire-validation.ts b/sdk/typescript/src/internal/wire-validation.ts index acea726f89..d8838cca9f 100644 --- a/sdk/typescript/src/internal/wire-validation.ts +++ b/sdk/typescript/src/internal/wire-validation.ts @@ -1,6 +1,7 @@ import type { ErrorData, InitializeResult, + PermissionRespondResult, QueryCancelResult, QueryEventParams, QueryResultParams, @@ -12,6 +13,7 @@ import type { import { isErrorData as isGeneratedErrorData, isInitializeResult, + isPermissionRespondResult, isQueryCancelResult, isQueryEventParams, isQueryResultParams, @@ -39,6 +41,8 @@ export function validateResponseResult(method: string, value: unknown): T { return validateQueryStartResult(value) as T; case "query/cancel": return validateQueryCancelResult(value) as T; + case "permission/respond": + return validatePermissionRespondResult(value) as T; case "session/close": return validateSessionCloseResult(value) as T; case "shutdown": @@ -134,6 +138,18 @@ function validateQueryCancelResult(value: unknown): QueryCancelResult { return result; } +function validatePermissionRespondResult(value: unknown): PermissionRespondResult { + const result = validateWireValue( + isPermissionRespondResult, + value, + "permission response result", + ); + if (!isNonEmptyString(result.requestId) || !result.accepted) { + throw new Error("SDK Host permission response result is invalid"); + } + return result; +} + function validateSessionCloseResult(value: unknown): SessionCloseResult { const result = validateWireValue( isSessionCloseResult, diff --git a/sdk/typescript/src/query.ts b/sdk/typescript/src/query.ts index 2fbe466c51..32d7c7d092 100644 --- a/sdk/typescript/src/query.ts +++ b/sdk/typescript/src/query.ts @@ -1,6 +1,8 @@ import type { JsonRpcConnection } from "./internal/json-rpc.js"; import { withTimeout } from "./internal/deadline.js"; import type { + PermissionRespondParams, + PermissionRespondResult, QueryCancelParams, QueryCancelResult, QueryEventParams, @@ -9,7 +11,13 @@ import type { SessionCloseParams, SessionCloseResult, } from "./internal/wire/index.js"; -import type { QueryStreamItem, Result, ResultError, Turn } from "./types.js"; +import type { + PermissionResponse, + QueryStreamItem, + Result, + ResultError, + Turn, +} from "./types.js"; import { isConnectionUnusableError, SdkError } from "./errors.js"; interface QueueWaiter { @@ -33,6 +41,7 @@ export class Query implements AsyncIterable { readonly #ownsSession: boolean; readonly #closeTimeoutMs: number; readonly #closedHandlers = new Set<() => void>(); + readonly #pendingPermissionIds = new Set(); #lastSequence = 0; #terminal = false; #unsubscribe: () => void = () => {}; @@ -101,6 +110,34 @@ export class Query implements AsyncIterable { return this.#cancelPromise; } + async respondPermission( + requestId: string, + response: PermissionResponse, + ): Promise { + if (!this.#pendingPermissionIds.delete(requestId)) { + throw new Error("Permission request is unknown, expired, or already answered"); + } + if (response.decision !== "reject" && response.feedback !== undefined) { + throw new Error("Permission feedback is only valid when rejecting a request"); + } + const params: PermissionRespondParams = { + queryId: this.id, + sessionId: this.turn.sessionId, + turnId: this.turn.id, + operationId: this.operationId, + requestId, + decision: response.decision, + feedback: response.feedback, + }; + const result = await this.#connection.request( + "permission/respond", + params, + ); + if (result.requestId !== requestId || !result.accepted) { + throw new Error("SDK Host answered a different permission request"); + } + } + close(): Promise { this.#closePromise ??= this.#closeQuery(); return this.#closePromise; @@ -151,6 +188,43 @@ export class Query implements AsyncIterable { sequence: params.sequence, text: params.event.text, }); + } else if (params.event.type === "tool_event") { + this.#push({ + type: "tool_event", + queryId: params.queryId, + sessionId: params.sessionId, + turnId: params.turnId, + operationId: params.operationId, + sequence: params.sequence, + toolCallId: params.event.toolCallId, + toolName: params.event.toolName, + status: params.event.status, + ...(params.event.progress === undefined + ? {} + : { progress: params.event.progress }), + ...(params.event.durationMs === undefined + ? {} + : { durationMs: params.event.durationMs }), + }); + } else { + if (this.#pendingPermissionIds.has(params.event.requestId)) { + throw new Error("SDK Host repeated a pending permission request"); + } + this.#pendingPermissionIds.add(params.event.requestId); + this.#push({ + type: "permission_request", + queryId: params.queryId, + sessionId: params.sessionId, + turnId: params.turnId, + operationId: params.operationId, + sequence: params.sequence, + requestId: params.event.requestId, + action: params.event.action, + resources: params.event.resources, + source: params.event.source, + toolCallId: params.event.toolCallId ?? undefined, + responseTimeoutMs: params.event.responseTimeoutMs, + }); } } @@ -244,6 +318,7 @@ export class Query implements AsyncIterable { return; } this.#terminal = true; + this.#pendingPermissionIds.clear(); this.#resolveResult(result); this.#unsubscribe(); while (this.#waiters.length > 0) { @@ -317,6 +392,7 @@ export class Query implements AsyncIterable { return; } this.#terminal = true; + this.#pendingPermissionIds.clear(); this.#failure = error; this.#rejectResult(error); this.#unsubscribe(); @@ -341,9 +417,9 @@ function bufferedItemBytes(item: QueryStreamItem): number { // Result frames have their own connection-level frame bound. Counting their // aggregate output again against the event backlog would reject a valid // Query that has not consumed the same text deltas yet. - return item.type === "assistant_text_delta" - ? Buffer.byteLength(item.text, "utf8") + 256 - : 0; + return item.type === "result" + ? 0 + : Buffer.byteLength(JSON.stringify(item), "utf8") + 128; } function mapResultError(error: QueryResultParams["error"]): ResultError { diff --git a/sdk/typescript/src/types.ts b/sdk/typescript/src/types.ts index 0fad3f955a..ce399907b0 100644 --- a/sdk/typescript/src/types.ts +++ b/sdk/typescript/src/types.ts @@ -13,8 +13,8 @@ export interface AgentModelOptions { export interface AgentClientOptions { cwd: string; - /** Explicit absolute path to the native `bitfun-sdk-host`. */ - hostPath: string; + /** Advanced absolute-path override for the package-local native Host. */ + hostPath?: string; /** Deadline for the SDK Host initialize handshake. */ initializeTimeoutMs?: number; /** Process-lifetime model credentials installed into this Host connection. */ @@ -25,6 +25,14 @@ export interface AgentCapabilities { query: boolean; sessions: boolean; cancellation: boolean; + eventStream: boolean; + toolEvents: boolean; + permissionResponses: boolean; + structuredOutput: boolean; + usage: boolean; + customTools: boolean; + hooks: boolean; + mcpConfiguration: boolean; } export type SdkErrorCode = @@ -107,6 +115,47 @@ export interface AssistantTextDelta { text: string; } +export interface ToolEvent { + type: "tool_event"; + queryId: string; + sessionId: string; + turnId: string; + operationId: string; + sequence: number; + toolCallId: string; + toolName: string; + status: "started" | "progress" | "completed" | "failed" | "cancelled"; + progress?: number; + durationMs?: number; +} + +export interface PermissionSource { + kind: "tool_call" | "provider" | "extension"; + identity: string; +} + +export interface PermissionRequestEvent { + type: "permission_request"; + queryId: string; + sessionId: string; + turnId: string; + operationId: string; + sequence: number; + requestId: string; + action: string; + resources: readonly string[]; + source: PermissionSource; + toolCallId?: string; + responseTimeoutMs: number; +} + +export type PermissionDecision = "allow_once" | "allow_always" | "reject"; + +export interface PermissionResponse { + decision: PermissionDecision; + feedback?: string; +} + export type ResultStatus = "completed" | "failed" | "cancelled"; export interface ResultError extends SdkErrorDetails { @@ -124,4 +173,8 @@ export interface Result { error?: ResultError; } -export type QueryStreamItem = AssistantTextDelta | Result; +export type QueryStreamItem = + | AssistantTextDelta + | ToolEvent + | PermissionRequestEvent + | Result; diff --git a/sdk/typescript/test/client.test.ts b/sdk/typescript/test/client.test.ts index f059ad545e..6cbcd638be 100644 --- a/sdk/typescript/test/client.test.ts +++ b/sdk/typescript/test/client.test.ts @@ -22,11 +22,10 @@ const clientOptions = { }, } satisfies AgentClientOptions; -// @ts-expect-error An explicit native Host path is required until platform packages exist. -const missingHostPathOptions: AgentClientOptions = { +const packageHostOptions = { cwd: "D:/workspace/project", model: clientOptions.model, -}; +} satisfies AgentClientOptions; // @ts-expect-error Query model selection is bound at AgentClient.start. const queryModelOverride: QueryInput = { prompt: "hello", model: "attempted-override" }; @@ -34,9 +33,9 @@ const queryModelOverride: QueryInput = { prompt: "hello", model: "attempted-over const sessionModelOverride: SessionCreateInput = { model: "attempted-override" }; void queryModelOverride; void sessionModelOverride; -void missingHostPathOptions; +void packageHostOptions; -test("a Query streams ordered events and returns the Host terminal Result", async () => { +test("a Query streams tool and permission events before the terminal Result", async () => { const clientToHost = new PassThrough(); const hostToClient = new PassThrough(); const initializeRequests: unknown[] = []; @@ -56,9 +55,12 @@ test("a Query streams ordered events and returns the Host terminal Result", asyn assert.ok(client instanceof AgentClient); assert.equal(initializeRequests.length, 1); assert.deepEqual(initializeRequests[0], { - protocolVersion: 2, + protocolVersion: 3, clientInfo: { name: "@bitfun/agent-sdk", version: "0.0.0" }, - capabilities: { serverNotifications: true }, + capabilities: { + serverNotifications: true, + permissionResponses: true, + }, model: { provider: "openai", model: "fixture-model", @@ -73,19 +75,76 @@ test("a Query streams ordered events and returns the Host terminal Result", asyn assert.equal(query.id, "query-1"); assert.equal(query.operationId, "operation-1"); assert.deepEqual(query.turn, { id: "turn-1", sessionId: "session-1" }); + assert.deepEqual(client.capabilities, { + query: true, + sessions: true, + cancellation: true, + eventStream: true, + toolEvents: true, + permissionResponses: true, + structuredOutput: false, + usage: false, + customTools: false, + hooks: false, + mcpConfiguration: false, + }); const items = []; for await (const item of query) { items.push(item); + if (item.type === "permission_request") { + await query.respondPermission(item.requestId, { decision: "allow_once" }); + await assert.rejects( + query.respondPermission(item.requestId, { decision: "allow_once" }), + /unknown, expired, or already answered/, + ); + } } assert.deepEqual(items, [ { - type: "assistant_text_delta", + type: "tool_event", queryId: "query-1", sessionId: "session-1", turnId: "turn-1", operationId: "operation-1", sequence: 1, + toolCallId: "tool-1", + toolName: "Read", + status: "started", + }, + { + type: "permission_request", + queryId: "query-1", + sessionId: "session-1", + turnId: "turn-1", + operationId: "operation-1", + sequence: 2, + requestId: "permission-1", + action: "read", + resources: ["README.md"], + source: { kind: "tool_call", identity: "Read" }, + toolCallId: "tool-1", + responseTimeoutMs: 120_000, + }, + { + type: "tool_event", + queryId: "query-1", + sessionId: "session-1", + turnId: "turn-1", + operationId: "operation-1", + sequence: 3, + toolCallId: "tool-1", + toolName: "Read", + status: "completed", + durationMs: 12, + }, + { + type: "assistant_text_delta", + queryId: "query-1", + sessionId: "session-1", + turnId: "turn-1", + operationId: "operation-1", + sequence: 4, text: "fixture result", }, { @@ -471,7 +530,7 @@ async function runFixtureHost( jsonrpc: "2.0", id: request.id, result: { - protocolVersion: 2, + protocolVersion: 3, runtimeVersion: "0.2.17", stability: "not_delivered", capabilities: { @@ -481,10 +540,11 @@ async function runFixtureHost( queryCancel: true, sessionClose: true, eventStream: true, + toolEvents: true, structuredOutput: false, usage: false, customTools: false, - permissionCallbacks: false, + permissionResponses: true, hooks: false, mcpConfiguration: false, prestartedTransport: false, @@ -525,6 +585,77 @@ async function runFixtureHost( turnId: "turn-1", operationId: "operation-1", sequence: 1, + event: { + type: "tool_event", + toolCallId: "tool-1", + toolName: "Read", + status: "started", + }, + }, + }); + write(responses, { + jsonrpc: "2.0", + method: "query/event", + params: { + queryId: "query-1", + sessionId: "session-1", + turnId: "turn-1", + operationId: "operation-1", + sequence: 2, + event: { + type: "permission_request", + requestId: "permission-1", + action: "read", + resources: ["README.md"], + source: { kind: "tool_call", identity: "Read" }, + toolCallId: "tool-1", + responseTimeoutMs: 120_000, + }, + }, + }); + continue; + } + if (request.method === "permission/respond") { + assert.deepEqual(request.params, { + queryId: "query-1", + sessionId: "session-1", + turnId: "turn-1", + operationId: "operation-1", + requestId: "permission-1", + decision: "allow_once", + }); + write(responses, { + jsonrpc: "2.0", + id: request.id, + result: { requestId: "permission-1", accepted: true }, + }); + write(responses, { + jsonrpc: "2.0", + method: "query/event", + params: { + queryId: "query-1", + sessionId: "session-1", + turnId: "turn-1", + operationId: "operation-1", + sequence: 3, + event: { + type: "tool_event", + toolCallId: "tool-1", + toolName: "Read", + status: "completed", + durationMs: 12, + }, + }, + }); + write(responses, { + jsonrpc: "2.0", + method: "query/event", + params: { + queryId: "query-1", + sessionId: "session-1", + turnId: "turn-1", + operationId: "operation-1", + sequence: 4, event: { type: "assistant_text_delta", text: "fixture result" }, }, }); @@ -1096,7 +1227,7 @@ function initializeResponse(id: number): unknown { jsonrpc: "2.0", id, result: { - protocolVersion: 2, + protocolVersion: 3, runtimeVersion: "0.2.17", stability: "not_delivered", capabilities: { @@ -1106,10 +1237,11 @@ function initializeResponse(id: number): unknown { queryCancel: true, sessionClose: true, eventStream: true, + toolEvents: true, structuredOutput: false, usage: false, customTools: false, - permissionCallbacks: false, + permissionResponses: true, hooks: false, mcpConfiguration: false, prestartedTransport: false, diff --git a/sdk/typescript/test/fixtures/host.mjs b/sdk/typescript/test/fixtures/host.mjs index 6a56b2a07f..191b14d09b 100644 --- a/sdk/typescript/test/fixtures/host.mjs +++ b/sdk/typescript/test/fixtures/host.mjs @@ -5,7 +5,7 @@ for await (const line of lines) { const request = JSON.parse(line); if (request.method === "initialize") { if ( - request.params?.protocolVersion !== 2 || + request.params?.protocolVersion !== 3 || request.params?.model?.apiKey !== "fixture-secret" ) { throw new Error("Invalid initialize request"); @@ -14,7 +14,7 @@ for await (const line of lines) { jsonrpc: "2.0", id: request.id, result: { - protocolVersion: 2, + protocolVersion: 3, runtimeVersion: "fixture", stability: "not_delivered", capabilities: { @@ -24,10 +24,11 @@ for await (const line of lines) { queryCancel: true, sessionClose: true, eventStream: true, + toolEvents: true, structuredOutput: false, usage: false, customTools: false, - permissionCallbacks: false, + permissionResponses: true, hooks: false, mcpConfiguration: false, prestartedTransport: false, diff --git a/sdk/typescript/test/local-package-consumer.mjs b/sdk/typescript/test/local-package-consumer.mjs new file mode 100644 index 0000000000..d42816dd39 --- /dev/null +++ b/sdk/typescript/test/local-package-consumer.mjs @@ -0,0 +1,139 @@ +import assert from "node:assert/strict"; +import { spawn } from "node:child_process"; +import { mkdir, mkdtemp, rm, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { dirname, join, resolve } from "node:path"; +import { fileURLToPath } from "node:url"; + +const packageRoot = resolve(dirname(fileURLToPath(import.meta.url)), ".."); +const isolatedRoot = await mkdtemp(join(tmpdir(), "bitfun-sdk-consumer-")); +const packedRoot = join(isolatedRoot, "packed"); +const consumerRoot = join(isolatedRoot, "consumer"); +const workspace = join(isolatedRoot, "workspace"); +const userRoot = join(isolatedRoot, "user-root"); +const configRoot = join(isolatedRoot, "config-root"); + +try { + await Promise.all( + [packedRoot, consumerRoot, workspace, userRoot, configRoot].map((path) => + mkdir(path, { recursive: true }), + ), + ); + const packed = await runNpm([ + "pack", + packageRoot, + "--pack-destination", + packedRoot, + "--json", + ], isolatedRoot); + const packResult = JSON.parse(packed.stdout); + assert.equal(Array.isArray(packResult), true); + assert.equal(packResult.length, 1); + const tarball = join(packedRoot, packResult[0].filename); + + await writeFile( + join(consumerRoot, "package.json"), + `${JSON.stringify({ private: true, type: "module" }, null, 2)}\n`, + ); + await runNpm( + ["install", "--ignore-scripts", "--no-audit", "--no-fund", tarball], + consumerRoot, + ); + await writeFile( + join(consumerRoot, "run.mjs"), + ` +import assert from "node:assert/strict"; +import { AgentClient } from "@bitfun/agent-sdk"; + +const client = await AgentClient.start({ + cwd: process.cwd(), + model: { + provider: "openai", + model: "fixture-model", + apiKey: "local-consumer-fixture", + baseUrl: "http://127.0.0.1:9/v1", + }, +}); +try { + assert.equal(client.capabilities.query, true); + assert.equal(client.capabilities.toolEvents, true); + assert.equal(client.capabilities.permissionResponses, true); +} finally { + await client.close(); +} +process.stdout.write("local-package-consumer: PASS\\n"); +`, + ); + + const result = await run(process.execPath, [join(consumerRoot, "run.mjs")], consumerRoot, { + ...process.env, + BITFUN_E2E_STORAGE_GUARD: "1", + BITFUN_E2E_USER_ROOT: userRoot, + BITFUN_E2E_HOME: userRoot, + APPDATA: configRoot, + XDG_CONFIG_HOME: configRoot, + HOME: userRoot, + USERPROFILE: userRoot, + }); + assert.match(result.stdout, /local-package-consumer: PASS/); + process.stdout.write("local-package-consumer-smoke: PASS\n"); +} finally { + await rm(isolatedRoot, { + recursive: true, + force: true, + maxRetries: 20, + retryDelay: 100, + }); +} + +function runNpm(args, cwd) { + if (process.platform !== "win32") { + return run("npm", args, cwd); + } + const npmCli = join( + dirname(process.execPath), + "node_modules", + "npm", + "bin", + "npm-cli.js", + ); + return run(process.execPath, [npmCli, ...args], cwd); +} + +async function run(command, args, cwd, env = process.env) { + const child = spawn(command, args, { + cwd, + env, + shell: false, + windowsHide: true, + stdio: ["ignore", "pipe", "pipe"], + }); + let stdout = ""; + let stderr = ""; + child.stdout.setEncoding("utf8"); + child.stderr.setEncoding("utf8"); + child.stdout.on("data", (chunk) => { + stdout = appendBounded(stdout, chunk); + }); + child.stderr.on("data", (chunk) => { + stderr = appendBounded(stderr, chunk); + }); + const exit = await new Promise((resolveExit, rejectExit) => { + child.once("error", rejectExit); + child.once("exit", (code, signal) => resolveExit({ code, signal })); + }); + if (exit.code !== 0) { + throw new Error( + `${command} failed with ${exit.signal ?? `exit ${String(exit.code)}`}: ${stderr}`, + ); + } + return { stdout, stderr }; +} + +function appendBounded(current, chunk) { + const next = current + chunk; + if (Buffer.byteLength(next, "utf8") > 1024 * 1024) { + throw new Error("local package command output exceeded its size limit"); + } + return next; +} diff --git a/sdk/typescript/test/managed-host.test.ts b/sdk/typescript/test/managed-host.test.ts index 150b18c500..9def63ca73 100644 --- a/sdk/typescript/test/managed-host.test.ts +++ b/sdk/typescript/test/managed-host.test.ts @@ -6,6 +6,7 @@ import test from "node:test"; import { AgentClient, SdkError } from "../src/index.js"; import { createAgentClient } from "../src/internal/client.js"; +import { resolveHostPath } from "../src/internal/host-path.js"; import { forceKillTree, startManagedHost } from "../src/internal/managed-host.js"; import type { AgentClientOptions } from "../src/types.js"; @@ -51,41 +52,19 @@ test("AgentClient.start reports a missing Host before an operation begins", asyn ); }); -test("AgentClient.start rejects missing and relative Host paths before spawning", async (context) => { +test("the Host resolver uses the package-local executable without environment fallback", () => { const previousHostPath = process.env.BITFUN_SDK_HOST_PATH; - delete process.env.BITFUN_SDK_HOST_PATH; + process.env.BITFUN_SDK_HOST_PATH = process.execPath; try { - const cases: Array<{ name: string; options: unknown }> = [ - { - name: "missing path", - options: { cwd: process.cwd(), model }, - }, - { - name: "relative path", - options: { - cwd: dirname(process.execPath), - hostPath: basename(process.execPath), - initializeTimeoutMs: 100, - model, - }, - }, - ]; - - for (const fixture of cases) { - await context.test(fixture.name, async () => { - await assert.rejects( - AgentClient.start(fixture.options as AgentClientOptions), - (error: unknown) => { - assert.ok(error instanceof SdkError); - assert.equal(error.code, "invalid_request"); - assert.equal(error.stage, "initialize"); - assert.equal(error.outcomeCertainty, "not_started"); - assert.doesNotMatch(String(error.stack), /fixture-secret/); - return true; - }, - ); - }); - } + const executableName = process.platform === "win32" ? "bitfun-sdk-host.exe" : "bitfun-sdk-host"; + const expectedHost = fileURLToPath( + new URL( + `../native/${process.platform}-${process.arch}/${executableName}`, + import.meta.url, + ), + ); + assert.equal(resolveHostPath(), expectedHost); + assert.notEqual(resolveHostPath(), process.execPath); } finally { if (previousHostPath === undefined) { delete process.env.BITFUN_SDK_HOST_PATH; @@ -95,6 +74,25 @@ test("AgentClient.start rejects missing and relative Host paths before spawning" } }); +test("AgentClient.start rejects a relative Host override before spawning", async () => { + await assert.rejects( + AgentClient.start({ + cwd: dirname(process.execPath), + hostPath: basename(process.execPath), + initializeTimeoutMs: 100, + model, + }), + (error: unknown) => { + assert.ok(error instanceof SdkError); + assert.equal(error.code, "invalid_request"); + assert.equal(error.stage, "initialize"); + assert.equal(error.outcomeCertainty, "not_started"); + assert.doesNotMatch(String(error.stack), /fixture-secret/); + return true; + }, + ); +}); + test("AgentClient.start rejects invalid model options before spawning a Host", async (context) => { const missingHost = fileURLToPath( new URL("../../../../test/fixtures/missing-host", import.meta.url), diff --git a/sdk/typescript/test/public-contract.test.ts b/sdk/typescript/test/public-contract.test.ts index 531b636ddf..a544bb5abd 100644 --- a/sdk/typescript/test/public-contract.test.ts +++ b/sdk/typescript/test/public-contract.test.ts @@ -41,6 +41,7 @@ test("package files keep internal declarations and wire DTOs private", async () "dist/sdk/typescript/src/*.d.ts", "dist/sdk/typescript/src/*.js", "dist/sdk/typescript/src/internal/*.js", + "dist/sdk/typescript/native/**", "dist/src/crates/adapters/transport/typescript/src/*.js", "README.md", ]); diff --git a/sdk/typescript/test/real-host-smoke.mjs b/sdk/typescript/test/real-host-smoke.mjs index c38d225456..c3bc8b4232 100644 --- a/sdk/typescript/test/real-host-smoke.mjs +++ b/sdk/typescript/test/real-host-smoke.mjs @@ -31,10 +31,6 @@ async function runParent() { ), ); - const hostPath = process.env.BITFUN_SDK_HOST_PATH; - assert.equal(typeof hostPath, "string", "BITFUN_SDK_HOST_PATH is required"); - assert.notEqual(hostPath.length, 0, "BITFUN_SDK_HOST_PATH is required"); - const apiKey = `bitfun-sdk-${randomBytes(24).toString("hex")}`; let requestCount = 0; const requestTraces = []; @@ -79,7 +75,6 @@ async function runParent() { const address = await listenLocalhost(server); const workerEnvironment = { BITFUN_SDK_SMOKE_BASE_URL: `http://127.0.0.1:${String(address.port)}/v1`, - BITFUN_SDK_HOST_PATH: hostPath, BITFUN_SDK_SMOKE_WORKSPACE: workspace, BITFUN_E2E_STORAGE_GUARD: "1", BITFUN_E2E_USER_ROOT: userRoot, @@ -147,7 +142,6 @@ async function runParent() { async function runWorker() { const apiKey = await readApiKeyFromStdin(); const workspace = requiredEnvironment("BITFUN_SDK_SMOKE_WORKSPACE"); - const hostPath = requiredEnvironment("BITFUN_SDK_HOST_PATH"); const baseUrl = requiredEnvironment("BITFUN_SDK_SMOKE_BASE_URL"); const missingHost = join(workspace, "missing-bitfun-sdk-host"); @@ -186,7 +180,6 @@ async function runWorker() { const client = await AgentClient.start({ cwd: workspace, - hostPath, model: validModel, }); process.stdout.write("phase:client_started\n"); diff --git a/src/apps/sdk-host/src/transport.rs b/src/apps/sdk-host/src/transport.rs index e0757efa53..cc23f929a8 100644 --- a/src/apps/sdk-host/src/transport.rs +++ b/src/apps/sdk-host/src/transport.rs @@ -17,8 +17,8 @@ use bitfun_sdk_host::host::{ ConnectionControl, HostOutput, SdkHostConfig, SdkHostConnection, TemporaryModelInstaller, }; use bitfun_sdk_host::protocol::{ - JsonRpcErrorResponse, JsonRpcRequest, RequestId, METHOD_INITIALIZE, METHOD_QUERY_CANCEL, - METHOD_SESSION_CLOSE, METHOD_SHUTDOWN, + JsonRpcErrorResponse, JsonRpcRequest, RequestId, METHOD_INITIALIZE, METHOD_PERMISSION_RESPOND, + METHOD_QUERY_CANCEL, METHOD_SESSION_CLOSE, METHOD_SHUTDOWN, }; #[derive(Debug, Clone)] @@ -278,7 +278,7 @@ where let is_control_request = matches!( request.method.as_str(), - METHOD_QUERY_CANCEL | METHOD_SESSION_CLOSE + METHOD_PERMISSION_RESPOND | METHOD_QUERY_CANCEL | METHOD_SESSION_CLOSE ); let request_set = if is_control_request { &mut control_requests diff --git a/src/apps/sdk-host/tests/stdio_process.rs b/src/apps/sdk-host/tests/stdio_process.rs index a26a081cc9..3bdc8d0223 100644 --- a/src/apps/sdk-host/tests/stdio_process.rs +++ b/src/apps/sdk-host/tests/stdio_process.rs @@ -43,9 +43,12 @@ async fn standalone_sdk_host_negotiates_and_shuts_down_without_cli() { 1, "initialize", json!({ - "protocolVersion": 2, + "protocolVersion": 3, "clientInfo": { "name": "standalone-process-fixture", "version": "0.1.0" }, - "capabilities": { "serverNotifications": true }, + "capabilities": { + "serverNotifications": true, + "permissionResponses": true + }, "model": { "provider": "openai", "model": "fixture-model", @@ -57,7 +60,7 @@ async fn standalone_sdk_host_negotiates_and_shuts_down_without_cli() { .await; let initialized = read_response(&mut stdout, "initialize").await; assert_eq!(initialized["id"], 1); - assert_eq!(initialized["result"]["protocolVersion"], 2); + assert_eq!(initialized["result"]["protocolVersion"], 3); assert!(initialized["result"]["modelId"] .as_str() .is_some_and(|model_id| model_id.starts_with("sdk:openai:"))); diff --git a/src/apps/sdk-host/tests/stdio_transport.rs b/src/apps/sdk-host/tests/stdio_transport.rs index d6f7215729..1987c1291d 100644 --- a/src/apps/sdk-host/tests/stdio_transport.rs +++ b/src/apps/sdk-host/tests/stdio_transport.rs @@ -222,7 +222,7 @@ async fn stdio_transport_serves_initialize_and_shutdown_without_non_protocol_std client_write .write_all( concat!( - "{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"initialize\",\"params\":{\"protocolVersion\":2,\"clientInfo\":{\"name\":\"fixture\",\"version\":\"0.1\"},\"capabilities\":{\"serverNotifications\":true},\"model\":{\"provider\":\"openai\",\"model\":\"fixture-model\",\"apiKey\":\"fixture-secret\"}}}\n", + "{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"initialize\",\"params\":{\"protocolVersion\":3,\"clientInfo\":{\"name\":\"fixture\",\"version\":\"0.1\"},\"capabilities\":{\"serverNotifications\":true,\"permissionResponses\":true},\"model\":{\"provider\":\"openai\",\"model\":\"fixture-model\",\"apiKey\":\"fixture-secret\"}}}\n", "{\"jsonrpc\":\"2.0\",\"id\":2,\"method\":\"shutdown\",\"params\":{}}\n" ) .as_bytes(), @@ -238,7 +238,7 @@ async fn stdio_transport_serves_initialize_and_shutdown_without_non_protocol_std serde_json::from_str(&lines.next_line().await.unwrap().unwrap()).unwrap(); assert_eq!(initialized["id"], 1); - assert_eq!(initialized["result"]["protocolVersion"], 2); + assert_eq!(initialized["result"]["protocolVersion"], 3); assert_eq!(initialized["result"]["modelId"], "sdk:openai:transport"); assert_eq!(shutdown["id"], 2); assert_eq!(shutdown["result"]["accepted"], true); @@ -272,7 +272,7 @@ async fn stdio_transport_executes_json_rpc_notifications_without_replying() { client_write .write_all( concat!( - "{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"initialize\",\"params\":{\"protocolVersion\":2,\"clientInfo\":{\"name\":\"fixture\",\"version\":\"0.1\"},\"capabilities\":{\"serverNotifications\":true},\"model\":{\"provider\":\"openai\",\"model\":\"fixture-model\",\"apiKey\":\"fixture-secret\"}}}\n", + "{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"initialize\",\"params\":{\"protocolVersion\":3,\"clientInfo\":{\"name\":\"fixture\",\"version\":\"0.1\"},\"capabilities\":{\"serverNotifications\":true,\"permissionResponses\":true},\"model\":{\"provider\":\"openai\",\"model\":\"fixture-model\",\"apiKey\":\"fixture-secret\"}}}\n", "{\"jsonrpc\":\"2.0\",\"method\":\"shutdown\",\"params\":{}}\n" ) .as_bytes(), @@ -372,7 +372,7 @@ async fn transport_accepts_input_while_an_owner_call_is_pending_and_bounds_reque client_write .write_all( concat!( - "{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"initialize\",\"params\":{\"protocolVersion\":2,\"clientInfo\":{\"name\":\"fixture\",\"version\":\"0.1\"},\"capabilities\":{\"serverNotifications\":true},\"model\":{\"provider\":\"openai\",\"model\":\"fixture-model\",\"apiKey\":\"fixture-secret\"}}}\n", + "{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"initialize\",\"params\":{\"protocolVersion\":3,\"clientInfo\":{\"name\":\"fixture\",\"version\":\"0.1\"},\"capabilities\":{\"serverNotifications\":true,\"permissionResponses\":true},\"model\":{\"provider\":\"openai\",\"model\":\"fixture-model\",\"apiKey\":\"fixture-secret\"}}}\n", "{\"jsonrpc\":\"2.0\",\"id\":2,\"method\":\"session/create\",\"params\":{}}\n", "{\"jsonrpc\":\"2.0\",\"id\":3,\"method\":\"session/create\",\"params\":{}}\n" ) @@ -444,7 +444,7 @@ async fn shutdown_remains_available_when_the_data_request_budget_is_exhausted() )); client_write .write_all( - b"{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"initialize\",\"params\":{\"protocolVersion\":2,\"clientInfo\":{\"name\":\"fixture\",\"version\":\"0.1\"},\"capabilities\":{\"serverNotifications\":true},\"model\":{\"provider\":\"openai\",\"model\":\"fixture-model\",\"apiKey\":\"fixture-secret\"}}}\n", + b"{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"initialize\",\"params\":{\"protocolVersion\":3,\"clientInfo\":{\"name\":\"fixture\",\"version\":\"0.1\"},\"capabilities\":{\"serverNotifications\":true,\"permissionResponses\":true},\"model\":{\"provider\":\"openai\",\"model\":\"fixture-model\",\"apiKey\":\"fixture-secret\"}}}\n", ) .await .unwrap(); @@ -466,7 +466,24 @@ async fn shutdown_remains_available_when_the_data_request_budget_is_exhausted() .expect("blocking data request must start"); client_write - .write_all(b"{\"jsonrpc\":\"2.0\",\"id\":3,\"method\":\"shutdown\",\"params\":{}}\n") + .write_all( + b"{\"jsonrpc\":\"2.0\",\"id\":3,\"method\":\"permission/respond\",\"params\":{\"queryId\":\"missing-query\",\"sessionId\":\"missing-session\",\"turnId\":\"missing-turn\",\"operationId\":\"missing-operation\",\"requestId\":\"missing-permission\",\"decision\":\"reject\"}}\n", + ) + .await + .unwrap(); + let permission_response: serde_json::Value = serde_json::from_str( + &timeout(Duration::from_secs(1), lines.next_line()) + .await + .expect("permission response must use control capacity") + .unwrap() + .unwrap(), + ) + .unwrap(); + assert_eq!(permission_response["id"], 3); + assert_eq!(permission_response["error"]["data"]["code"], "not_found"); + + client_write + .write_all(b"{\"jsonrpc\":\"2.0\",\"id\":4,\"method\":\"shutdown\",\"params\":{}}\n") .await .unwrap(); client_write.shutdown().await.unwrap(); @@ -479,7 +496,7 @@ async fn shutdown_remains_available_when_the_data_request_budget_is_exhausted() .unwrap(), ) .unwrap(); - assert_eq!(shutdown["id"], 3); + assert_eq!(shutdown["id"], 4); assert_eq!(shutdown["result"]["accepted"], true); task.await.unwrap().unwrap(); } @@ -507,9 +524,9 @@ async fn duplicate_initialize_does_not_abort_an_in_flight_request() { client_write .write_all( concat!( - "{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"initialize\",\"params\":{\"protocolVersion\":2,\"clientInfo\":{\"name\":\"fixture\",\"version\":\"0.1\"},\"capabilities\":{\"serverNotifications\":true},\"model\":{\"provider\":\"openai\",\"model\":\"fixture-model\",\"apiKey\":\"fixture-secret\"}}}\n", + "{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"initialize\",\"params\":{\"protocolVersion\":3,\"clientInfo\":{\"name\":\"fixture\",\"version\":\"0.1\"},\"capabilities\":{\"serverNotifications\":true,\"permissionResponses\":true},\"model\":{\"provider\":\"openai\",\"model\":\"fixture-model\",\"apiKey\":\"fixture-secret\"}}}\n", "{\"jsonrpc\":\"2.0\",\"id\":2,\"method\":\"session/create\",\"params\":{}}\n", - "{\"jsonrpc\":\"2.0\",\"id\":3,\"method\":\"initialize\",\"params\":{\"protocolVersion\":2,\"clientInfo\":{\"name\":\"fixture\",\"version\":\"0.1\"},\"capabilities\":{\"serverNotifications\":true},\"model\":{\"provider\":\"openai\",\"model\":\"fixture-model\",\"apiKey\":\"fixture-secret\"}}}\n" + "{\"jsonrpc\":\"2.0\",\"id\":3,\"method\":\"initialize\",\"params\":{\"protocolVersion\":3,\"clientInfo\":{\"name\":\"fixture\",\"version\":\"0.1\"},\"capabilities\":{\"serverNotifications\":true,\"permissionResponses\":true},\"model\":{\"provider\":\"openai\",\"model\":\"fixture-model\",\"apiKey\":\"fixture-secret\"}}}\n" ) .as_bytes(), ) @@ -567,7 +584,7 @@ async fn connection_eof_cleans_a_session_created_after_its_request_is_aborted() client_write .write_all( concat!( - "{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"initialize\",\"params\":{\"protocolVersion\":2,\"clientInfo\":{\"name\":\"fixture\",\"version\":\"0.1\"},\"capabilities\":{\"serverNotifications\":true},\"model\":{\"provider\":\"openai\",\"model\":\"fixture-model\",\"apiKey\":\"fixture-secret\"}}}\n", + "{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"initialize\",\"params\":{\"protocolVersion\":3,\"clientInfo\":{\"name\":\"fixture\",\"version\":\"0.1\"},\"capabilities\":{\"serverNotifications\":true,\"permissionResponses\":true},\"model\":{\"provider\":\"openai\",\"model\":\"fixture-model\",\"apiKey\":\"fixture-secret\"}}}\n", "{\"jsonrpc\":\"2.0\",\"id\":2,\"method\":\"session/create\",\"params\":{}}\n" ) .as_bytes(), @@ -626,7 +643,7 @@ async fn explicit_shutdown_bounds_request_drain_and_transient_cleanup_together() client_write .write_all( concat!( - "{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"initialize\",\"params\":{\"protocolVersion\":2,\"clientInfo\":{\"name\":\"fixture\",\"version\":\"0.1\"},\"capabilities\":{\"serverNotifications\":true},\"model\":{\"provider\":\"openai\",\"model\":\"fixture-model\",\"apiKey\":\"fixture-secret\"}}}\n", + "{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"initialize\",\"params\":{\"protocolVersion\":3,\"clientInfo\":{\"name\":\"fixture\",\"version\":\"0.1\"},\"capabilities\":{\"serverNotifications\":true,\"permissionResponses\":true},\"model\":{\"provider\":\"openai\",\"model\":\"fixture-model\",\"apiKey\":\"fixture-secret\"}}}\n", "{\"jsonrpc\":\"2.0\",\"id\":2,\"method\":\"session/create\",\"params\":{}}\n" ) .as_bytes(), @@ -682,9 +699,9 @@ async fn requests_before_a_successful_initialize_cannot_cross_the_handshake() { client_write .write_all( concat!( - "{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"initialize\",\"params\":{\"protocolVersion\":999,\"clientInfo\":{\"name\":\"fixture\",\"version\":\"0.1\"},\"capabilities\":{\"serverNotifications\":true},\"model\":{\"provider\":\"openai\",\"model\":\"fixture-model\",\"apiKey\":\"fixture-secret\"}}}\n", + "{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"initialize\",\"params\":{\"protocolVersion\":999,\"clientInfo\":{\"name\":\"fixture\",\"version\":\"0.1\"},\"capabilities\":{\"serverNotifications\":true,\"permissionResponses\":true},\"model\":{\"provider\":\"openai\",\"model\":\"fixture-model\",\"apiKey\":\"fixture-secret\"}}}\n", "{\"jsonrpc\":\"2.0\",\"id\":2,\"method\":\"session/create\",\"params\":{}}\n", - "{\"jsonrpc\":\"2.0\",\"id\":3,\"method\":\"initialize\",\"params\":{\"protocolVersion\":2,\"clientInfo\":{\"name\":\"fixture\",\"version\":\"0.1\"},\"capabilities\":{\"serverNotifications\":true},\"model\":{\"provider\":\"openai\",\"model\":\"fixture-model\",\"apiKey\":\"fixture-secret\"}}}\n" + "{\"jsonrpc\":\"2.0\",\"id\":3,\"method\":\"initialize\",\"params\":{\"protocolVersion\":3,\"clientInfo\":{\"name\":\"fixture\",\"version\":\"0.1\"},\"capabilities\":{\"serverNotifications\":true,\"permissionResponses\":true},\"model\":{\"provider\":\"openai\",\"model\":\"fixture-model\",\"apiKey\":\"fixture-secret\"}}}\n" ) .as_bytes(), ) @@ -703,7 +720,7 @@ async fn requests_before_a_successful_initialize_cannot_cross_the_handshake() { assert_eq!(pre_initialize["id"], 2); assert_eq!(pre_initialize["error"]["data"]["code"], "not_initialized"); assert_eq!(initialized["id"], 3); - assert_eq!(initialized["result"]["protocolVersion"], 2); + assert_eq!(initialized["result"]["protocolVersion"], 3); assert_eq!(initialized["result"]["modelId"], "sdk:openai:transport"); client_write @@ -744,7 +761,7 @@ async fn blocked_output_times_out_and_ends_the_connection() { )); client_write .write_all( - b"{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"initialize\",\"params\":{\"protocolVersion\":2,\"clientInfo\":{\"name\":\"fixture\",\"version\":\"0.1\"},\"capabilities\":{\"serverNotifications\":true},\"model\":{\"provider\":\"openai\",\"model\":\"fixture-model\",\"apiKey\":\"fixture-secret\"}}}\n", + b"{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"initialize\",\"params\":{\"protocolVersion\":3,\"clientInfo\":{\"name\":\"fixture\",\"version\":\"0.1\"},\"capabilities\":{\"serverNotifications\":true,\"permissionResponses\":true},\"model\":{\"provider\":\"openai\",\"model\":\"fixture-model\",\"apiKey\":\"fixture-secret\"}}}\n", ) .await .unwrap(); diff --git a/src/crates/interfaces/sdk-host/src/host.rs b/src/crates/interfaces/sdk-host/src/host.rs index 150e3eb896..cab170696f 100644 --- a/src/crates/interfaces/sdk-host/src/host.rs +++ b/src/crates/interfaces/sdk-host/src/host.rs @@ -9,12 +9,12 @@ use bitfun_agent_runtime::sdk::{ AgentDialogTurnRequest, AgentRuntime, AgentSessionCreateRequest, AgentSessionCreateResult, AgentSubmissionSource, AgentTransientSessionDiscardRequest, AgentTurnCancellationRequest, AgentTurnSettlementRequest, DialogSubmissionPolicy, DialogSubmitOutcome, PermissionReply, - PermissionReplySource, PermissionRequest, PermissionRequestEvent, PortErrorKind, RuntimeError, - AUTO_APPROVE_ASK_CONTEXT_KEY, + PermissionReplySource, PermissionRequest, PermissionRequestEvent, PermissionRequestSourceKind, + PortErrorKind, RuntimeError, AUTO_APPROVE_ASK_CONTEXT_KEY, }; use bitfun_agent_runtime::user_questions::USER_INPUT_AVAILABLE_CONTEXT_KEY; use bitfun_core_types::ErrorCategory; -use bitfun_events::AgenticEvent; +use bitfun_events::{AgenticEvent, ToolEventData}; use futures_util::{stream::FuturesUnordered, FutureExt, StreamExt}; use tokio::sync::{mpsc, oneshot, Mutex, OwnedSemaphorePermit, Semaphore}; use tokio::task::JoinHandle; @@ -24,19 +24,23 @@ use tokio_util::sync::CancellationToken; use crate::protocol::{ ErrorCode, ErrorData, ErrorStage, InitializeParams, InitializeResult, JsonRpcErrorResponse, JsonRpcNotification, JsonRpcRequest, JsonRpcSuccessResponse, OutcomeCertainty, - QueryCancelParams, QueryCancelResult, QueryEvent, QueryEventParams, QueryOutput, - QueryResultError, QueryResultParams, QueryStartParams, QueryStartResult, QueryTerminalStatus, - RecoveryAction, RequestId, SessionCloseParams, SessionCloseResult, SessionCreateParams, - SessionCreateResult, SessionLifetime, ShutdownParams, ShutdownResult, TemporaryModelConfig, - JSON_RPC_VERSION, METHOD_INITIALIZE, METHOD_QUERY_CANCEL, METHOD_QUERY_START, - METHOD_SESSION_CLOSE, METHOD_SESSION_CREATE, METHOD_SHUTDOWN, NOTIFICATION_QUERY_EVENT, - NOTIFICATION_QUERY_RESULT, PROTOCOL_VERSION, + PermissionDecision, PermissionRespondParams, PermissionRespondResult, PermissionSource, + PermissionSourceKind, QueryCancelParams, QueryCancelResult, QueryEvent, QueryEventParams, + QueryOutput, QueryResultError, QueryResultParams, QueryStartParams, QueryStartResult, + QueryTerminalStatus, RecoveryAction, RequestId, SessionCloseParams, SessionCloseResult, + SessionCreateParams, SessionCreateResult, SessionLifetime, ShutdownParams, ShutdownResult, + TemporaryModelConfig, ToolEventStatus, JSON_RPC_VERSION, METHOD_INITIALIZE, + METHOD_PERMISSION_RESPOND, METHOD_QUERY_CANCEL, METHOD_QUERY_START, METHOD_SESSION_CLOSE, + METHOD_SESSION_CREATE, METHOD_SHUTDOWN, NOTIFICATION_QUERY_EVENT, NOTIFICATION_QUERY_RESULT, + PROTOCOL_VERSION, }; const DEFAULT_SESSION_NAME: &str = "BitFun SDK query"; const DEFAULT_AGENT: &str = "agentic"; const DEFAULT_TURN_SETTLEMENT_TIMEOUT_MS: u64 = 5_000; const PERMISSION_REJECTION_TIMEOUT_MS: u64 = 2_000; +const DEFAULT_PERMISSION_RESPONSE_TIMEOUT_MS: u64 = 120_000; +const MAX_PERMISSION_FEEDBACK_BYTES: usize = 4 * 1024; const MAX_SESSION_CLOSE_TIMEOUT_MS: u64 = 30_000; const MAX_QUERY_OUTPUT_WIRE_BYTES: usize = 768 * 1024; @@ -59,6 +63,7 @@ pub struct SdkHostConfig { pub max_in_flight_control_requests: usize, pub max_active_queries: usize, pub max_leased_sessions: usize, + pub permission_response_timeout: Duration, } impl Default for SdkHostConfig { @@ -68,6 +73,9 @@ impl Default for SdkHostConfig { max_in_flight_control_requests: 4, max_active_queries: 16, max_leased_sessions: 64, + permission_response_timeout: Duration::from_millis( + DEFAULT_PERMISSION_RESPONSE_TIMEOUT_MS, + ), } } } @@ -118,6 +126,7 @@ struct ConnectionInner { control_request_budget: Arc, query_budget: Arc, session_budget: Arc, + permission_response_timeout: Duration, shutdown_started: CancellationToken, connection_failed: CancellationToken, } @@ -126,6 +135,7 @@ struct ConnectionInner { struct ConnectionState { initialization: InitializationState, model_id: Option, + permission_responses: bool, shutting_down: bool, cleanup_failed: bool, sessions: HashMap, @@ -181,6 +191,7 @@ struct QueryLease { terminal: AtomicBool, stop_forwarding: CancellationToken, emit_output: bool, + pending_permissions: StdMutex>, _budget: OwnedSemaphorePermit, } @@ -234,6 +245,7 @@ impl SdkHostConnection { )), query_budget: Arc::new(Semaphore::new(config.max_active_queries.max(1))), session_budget: Arc::new(Semaphore::new(config.max_leased_sessions.max(1))), + permission_response_timeout: config.permission_response_timeout, shutdown_started: CancellationToken::new(), connection_failed: CancellationToken::new(), }), @@ -279,7 +291,7 @@ impl SdkHostConnection { } else { let budget = if matches!( request.method.as_str(), - METHOD_QUERY_CANCEL | METHOD_SESSION_CLOSE + METHOD_QUERY_CANCEL | METHOD_PERMISSION_RESPOND | METHOD_SESSION_CLOSE ) { self.inner.control_request_budget.clone() } else { @@ -354,6 +366,7 @@ impl SdkHostConnection { METHOD_SESSION_CREATE => self.handle_session_create(request).await, METHOD_QUERY_START => self.handle_query_start(request).await, METHOD_QUERY_CANCEL => self.handle_query_cancel(request).await, + METHOD_PERMISSION_RESPOND => self.handle_permission_respond(request).await, METHOD_SESSION_CLOSE => self.handle_session_close(request).await, METHOD_SHUTDOWN => { if self @@ -770,6 +783,7 @@ impl SdkHostConnection { .await; return; } + let permission_responses = params.capabilities.permission_responses; let initialization_error = { let mut state = self.inner.state.lock().await; if state.shutting_down { @@ -842,6 +856,7 @@ impl SdkHostConnection { true } else { state.model_id = Some(model_id.clone()); + state.permission_responses = permission_responses; state.initialization = InitializationState::Initialized; false } @@ -859,11 +874,9 @@ impl SdkHostConnection { .await; return; } - self.send_success( - request.id.clone(), - InitializeResult::current(self.inner.runtime_version, model_id), - ) - .await; + let mut result = InitializeResult::current(self.inner.runtime_version, model_id); + result.capabilities.permission_responses = permission_responses; + self.send_success(request.id.clone(), result).await; } async fn handle_session_create(&self, request: JsonRpcRequest) { @@ -1174,6 +1187,7 @@ impl SdkHostConnection { terminal: AtomicBool::new(false), stop_forwarding: CancellationToken::new(), emit_output, + pending_permissions: StdMutex::new(HashMap::new()), _budget: query_budget, }); { @@ -1224,16 +1238,33 @@ impl SdkHostConnection { &lease, ) => { - connection.reject_permission_and_finish(&lease, &request).await; - return; + if !connection + .forward_permission_request(&lease, &request, &mut sequence) + .await + { + return; + } + continue; + } + Ok(PermissionRequestEvent::Replied { request_id, .. }) + | Ok(PermissionRequestEvent::Cancelled { request_id, .. }) => { + if let Some(timeout_cancel) = lease + .pending_permissions + .lock() + .expect("SDK Host pending permission lock poisoned") + .remove(&request_id) + { + timeout_cancel.cancel(); + } + continue; } Ok(_) => continue, Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => { match connection.inner.runtime.pending_permission_requests() { Ok(pending) => { - if let Some(request) = pending + let pending = pending .into_iter() - .find(|request| { + .filter(|request| { permission_request_targets_query( request, connection @@ -1248,9 +1279,36 @@ impl SdkHostConnection { &lease, ) }) + .collect::>(); + let authoritative = pending + .iter() + .map(|request| request.request_id.as_str()) + .collect::>(); { - connection.reject_permission_and_finish(&lease, &request).await; - return; + let mut tracked = lease + .pending_permissions + .lock() + .expect("SDK Host pending permission lock poisoned"); + tracked.retain(|request_id, timeout_cancel| { + let keep = authoritative + .contains(request_id.as_str()); + if !keep { + timeout_cancel.cancel(); + } + keep + }); + } + for request in pending { + if !connection + .forward_permission_request( + &lease, + &request, + &mut sequence, + ) + .await + { + return; + } } continue; } @@ -1322,38 +1380,39 @@ impl SdkHostConnection { let terminal = terminal_fact(&envelope.event, &lease.turn_id, &lease.query_id); if lease.emit_output { if let Some(projected) = project_query_event(&envelope.event) { - let QueryEvent::AssistantTextDelta { text } = &projected; - let output_exceeded = { - let encoded_bytes = json_string_content_bytes(text); - let mut output = lease - .output - .lock() - .expect("SDK Host Query output lock poisoned"); - if encoded_bytes - > MAX_QUERY_OUTPUT_WIRE_BYTES.saturating_sub(output.wire_bytes) - { - true - } else { - output.text.push_str(text); - output.wire_bytes += encoded_bytes; - false + if let QueryEvent::AssistantTextDelta { text } = &projected { + let output_exceeded = { + let encoded_bytes = json_string_content_bytes(text); + let mut output = lease + .output + .lock() + .expect("SDK Host Query output lock poisoned"); + if encoded_bytes + > MAX_QUERY_OUTPUT_WIRE_BYTES.saturating_sub(output.wire_bytes) + { + true + } else { + output.text.push_str(text); + output.wire_bytes += encoded_bytes; + false + } + }; + if output_exceeded { + connection + .cancel_and_finish( + &lease, + QueryResultError::new( + ErrorCode::Overloaded, + false, + None, + &lease.query_id, + "SDK Host Query output exceeded the protocol size limit", + ), + true, + ) + .await; + return; } - }; - if output_exceeded { - connection - .cancel_and_finish( - &lease, - QueryResultError::new( - ErrorCode::Overloaded, - false, - None, - &lease.query_id, - "SDK Host Query output exceeded the protocol size limit", - ), - true, - ) - .await; - return; } sequence += 1; if !connection @@ -1489,6 +1548,174 @@ impl SdkHostConnection { } } + async fn handle_permission_respond(&self, request: JsonRpcRequest) { + let Some(mut params) = self + .parse_params::(&request, ErrorStage::Query) + .await + else { + return; + }; + if params.decision != PermissionDecision::Reject && params.feedback.is_some() { + self.send_invalid_params( + request.id.clone(), + ErrorStage::Query, + "feedback is only valid when rejecting a permission request", + ) + .await; + return; + } + if params + .feedback + .as_ref() + .is_some_and(|feedback| feedback.len() > MAX_PERMISSION_FEEDBACK_BYTES) + { + self.send_invalid_params( + request.id.clone(), + ErrorStage::Query, + "Permission rejection feedback exceeds the size limit", + ) + .await; + return; + } + params.feedback = params + .feedback + .map(|feedback| feedback.trim().to_string()) + .filter(|feedback| !feedback.is_empty()); + let (permission_responses, lease) = { + let state = self.inner.state.lock().await; + ( + state.permission_responses, + state.queries.get(¶ms.query_id).cloned(), + ) + }; + if !permission_responses { + self.send_error( + request.id.clone(), + ErrorCode::CapabilityUnavailable, + ErrorStage::Query, + false, + None, + "permission responses were not negotiated for this connection", + ) + .await; + return; + } + let Some(lease) = lease else { + self.send_error( + request.id.clone(), + ErrorCode::NotFound, + ErrorStage::Query, + false, + None, + "Query is not active on this SDK Host connection", + ) + .await; + return; + }; + if lease.session_id != params.session_id + || lease.turn_id != params.turn_id + || lease.operation_id != params.operation_id + { + self.send_invalid_params( + request.id.clone(), + ErrorStage::Query, + "Permission response identity does not match the owning Query", + ) + .await; + return; + } + let timeout_cancel = lease + .pending_permissions + .lock() + .expect("SDK Host pending permission lock poisoned") + .remove(¶ms.request_id); + let Some(timeout_cancel) = timeout_cancel else { + self.send_error( + request.id.clone(), + ErrorCode::NotFound, + ErrorStage::Query, + false, + None, + "Permission request is unknown, expired, or already answered", + ) + .await; + return; + }; + timeout_cancel.cancel(); + let reply = match params.decision { + PermissionDecision::AllowOnce => PermissionReply::Once, + PermissionDecision::AllowAlways => PermissionReply::Always, + PermissionDecision::Reject => PermissionReply::Reject { + feedback: params.feedback, + }, + }; + match timeout( + Duration::from_millis(PERMISSION_REJECTION_TIMEOUT_MS), + self.inner.runtime.respond_permission_with_source( + ¶ms.request_id, + reply, + PermissionReplySource::User, + ), + ) + .await + { + Ok(Ok(())) => { + self.send_success( + request.id.clone(), + PermissionRespondResult { + request_id: params.request_id, + accepted: true, + }, + ) + .await; + } + Ok(Err(error)) => { + self.cancel_and_finish( + &lease, + query_error_from_runtime( + &lease.query_id, + error, + "SDK Host permission response failed", + ), + true, + ) + .await; + self.send_error( + request.id.clone(), + ErrorCode::Internal, + ErrorStage::Query, + false, + None, + "Permission response failed and the Query was cancelled", + ) + .await; + } + Err(_) => { + self.cancel_and_finish( + &lease, + QueryResultError::new( + ErrorCode::Timeout, + false, + None, + &lease.query_id, + "SDK Host permission response timed out", + ), + true, + ) + .await; + self.send_error( + request.id.clone(), + ErrorCode::Timeout, + ErrorStage::Query, + false, + None, + "Permission response outcome is unknown and the Query was cancelled", + ) + .await; + } + } + } + async fn handle_session_close(&self, request: JsonRpcRequest) { let Some(params) = self .parse_params::(&request, ErrorStage::Session) @@ -1927,6 +2154,15 @@ impl SdkHostConnection { if !lease.finish_once() { return; } + lease.stop_forwarding.cancel(); + for (_, timeout_cancel) in lease + .pending_permissions + .lock() + .expect("SDK Host pending permission lock poisoned") + .drain() + { + timeout_cancel.cancel(); + } let settlement = timeout( Duration::from_millis(DEFAULT_TURN_SETTLEMENT_TIMEOUT_MS + 500), self.inner @@ -2102,6 +2338,145 @@ impl SdkHostConnection { .await; } + async fn forward_permission_request( + &self, + lease: &Arc, + request: &PermissionRequest, + sequence: &mut u64, + ) -> bool { + if !self.inner.state.lock().await.permission_responses { + self.reject_permission_and_finish(lease, request).await; + return false; + } + let timeout_cancel = { + let mut pending = lease + .pending_permissions + .lock() + .expect("SDK Host pending permission lock poisoned"); + if pending.contains_key(&request.request_id) { + return true; + } + let timeout_cancel = CancellationToken::new(); + pending.insert(request.request_id.clone(), timeout_cancel.clone()); + timeout_cancel + }; + *sequence += 1; + let delivered = self + .send_notification( + NOTIFICATION_QUERY_EVENT, + QueryEventParams { + query_id: lease.query_id.clone(), + session_id: lease.session_id.clone(), + turn_id: lease.turn_id.clone(), + operation_id: lease.operation_id.clone(), + sequence: *sequence, + event: QueryEvent::PermissionRequest { + request_id: request.request_id.clone(), + action: request.action.clone(), + resources: request.resources.clone(), + source: PermissionSource { + kind: match request.source.kind { + PermissionRequestSourceKind::ToolCall => { + PermissionSourceKind::ToolCall + } + PermissionRequestSourceKind::Provider => { + PermissionSourceKind::Provider + } + PermissionRequestSourceKind::Extension => { + PermissionSourceKind::Extension + } + }, + identity: request.source.identity.clone(), + }, + tool_call_id: request.tool_call_id.clone(), + response_timeout_ms: duration_ms(self.inner.permission_response_timeout), + }, + }, + ) + .await; + if !delivered { + if let Some(timeout_cancel) = lease + .pending_permissions + .lock() + .expect("SDK Host pending permission lock poisoned") + .remove(&request.request_id) + { + timeout_cancel.cancel(); + } + self.reject_permission_and_finish(lease, request).await; + return false; + } + self.spawn_permission_timeout(lease.clone(), request.request_id.clone(), timeout_cancel); + true + } + + fn spawn_permission_timeout( + &self, + lease: Arc, + request_id: String, + timeout_cancel: CancellationToken, + ) { + let connection = self.clone(); + tokio::spawn(async move { + tokio::select! { + _ = lease.stop_forwarding.cancelled() => return, + _ = timeout_cancel.cancelled() => return, + _ = tokio::time::sleep(connection.inner.permission_response_timeout) => {} + } + let expired = lease + .pending_permissions + .lock() + .expect("SDK Host pending permission lock poisoned") + .remove(&request_id) + .is_some(); + if !expired { + return; + } + let rejection = timeout( + Duration::from_millis(PERMISSION_REJECTION_TIMEOUT_MS), + connection.inner.runtime.respond_permission_with_source( + &request_id, + PermissionReply::Reject { + feedback: Some("SDK permission response timed out".to_string()), + }, + PermissionReplySource::System, + ), + ) + .await; + match rejection { + Ok(Ok(())) => {} + Ok(Err(error)) => { + connection + .cancel_and_finish( + &lease, + query_error_from_runtime( + &lease.query_id, + error, + "SDK Host could not reject an expired permission request", + ), + true, + ) + .await; + } + Err(_) => { + connection + .cancel_and_finish( + &lease, + QueryResultError::new( + ErrorCode::Timeout, + true, + Some(RecoveryAction::RestartHost), + &lease.query_id, + "SDK Host permission timeout rejection did not settle", + ), + true, + ) + .await; + } + } + }); + } + async fn parse_params(&self, request: &JsonRpcRequest, stage: ErrorStage) -> Option where T: serde::de::DeserializeOwned, @@ -2283,7 +2658,8 @@ fn event_turn_id(event: &AgenticEvent) -> Option<&str> { AgenticEvent::DialogTurnCompleted { turn_id, .. } | AgenticEvent::DialogTurnCancelled { turn_id, .. } | AgenticEvent::DialogTurnFailed { turn_id, .. } - | AgenticEvent::TextChunk { turn_id, .. } => Some(turn_id), + | AgenticEvent::TextChunk { turn_id, .. } + | AgenticEvent::ToolEvent { turn_id, .. } => Some(turn_id), _ => None, } } @@ -2297,6 +2673,31 @@ fn project_query_event(event: &AgenticEvent) -> Option { AgenticEvent::TextChunk { text, .. } => { Some(QueryEvent::AssistantTextDelta { text: text.clone() }) } + AgenticEvent::ToolEvent { tool_event, .. } => { + let (status, progress, duration_ms) = match tool_event { + ToolEventData::Started { .. } => (ToolEventStatus::Started, None, None), + ToolEventData::Progress { percentage, .. } => { + (ToolEventStatus::Progress, Some(*percentage), None) + } + ToolEventData::Completed { duration_ms, .. } => { + (ToolEventStatus::Completed, None, Some(*duration_ms)) + } + ToolEventData::Failed { duration_ms, .. } => { + (ToolEventStatus::Failed, None, *duration_ms) + } + ToolEventData::Cancelled { duration_ms, .. } => { + (ToolEventStatus::Cancelled, None, *duration_ms) + } + _ => return None, + }; + Some(QueryEvent::ToolEvent { + tool_call_id: tool_event.tool_id().to_string(), + tool_name: tool_event.effective_tool_name().to_string(), + status, + progress, + duration_ms, + }) + } _ => None, } } diff --git a/src/crates/interfaces/sdk-host/src/protocol.rs b/src/crates/interfaces/sdk-host/src/protocol.rs index 0a6cb3983a..268fb5e350 100644 --- a/src/crates/interfaces/sdk-host/src/protocol.rs +++ b/src/crates/interfaces/sdk-host/src/protocol.rs @@ -6,12 +6,13 @@ use serde::de::DeserializeOwned; use serde::{Deserialize, Serialize}; pub const JSON_RPC_VERSION: &str = "2.0"; -pub const PROTOCOL_VERSION: u32 = 2; +pub const PROTOCOL_VERSION: u32 = 3; pub const METHOD_INITIALIZE: &str = "initialize"; pub const METHOD_SESSION_CREATE: &str = "session/create"; pub const METHOD_QUERY_START: &str = "query/start"; pub const METHOD_QUERY_CANCEL: &str = "query/cancel"; +pub const METHOD_PERMISSION_RESPOND: &str = "permission/respond"; pub const METHOD_SESSION_CLOSE: &str = "session/close"; pub const METHOD_SHUTDOWN: &str = "shutdown"; pub const NOTIFICATION_QUERY_EVENT: &str = "query/event"; @@ -221,6 +222,7 @@ pub struct ClientInfo { #[serde(deny_unknown_fields, rename_all = "camelCase")] pub struct ClientCapabilities { pub server_notifications: bool, + pub permission_responses: bool, } #[derive(Debug, Clone, PartialEq, Eq, Serialize)] @@ -264,10 +266,11 @@ pub struct HostCapabilities { pub query_cancel: bool, pub session_close: bool, pub event_stream: bool, + pub tool_events: bool, pub structured_output: bool, pub usage: bool, pub custom_tools: bool, - pub permission_callbacks: bool, + pub permission_responses: bool, pub hooks: bool, pub mcp_configuration: bool, pub prestarted_transport: bool, @@ -282,10 +285,11 @@ impl HostCapabilities { query_cancel: true, session_close: true, event_stream: true, + tool_events: true, structured_output: false, usage: false, custom_tools: false, - permission_callbacks: false, + permission_responses: true, hooks: false, mcp_configuration: false, prestarted_transport: false, @@ -497,6 +501,55 @@ pub struct QueryCancelResult { pub requested: bool, } +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[cfg_attr(feature = "ts", derive(ts_rs::TS), ts(export))] +#[serde(rename_all = "snake_case")] +pub enum PermissionDecision { + AllowOnce, + AllowAlways, + Reject, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[cfg_attr(feature = "ts", derive(ts_rs::TS), ts(export))] +#[serde(deny_unknown_fields, rename_all = "camelCase")] +pub struct PermissionRespondParams { + pub query_id: String, + pub session_id: String, + pub turn_id: String, + pub operation_id: String, + pub request_id: String, + pub decision: PermissionDecision, + #[cfg_attr(feature = "ts", ts(optional = nullable))] + #[serde(default)] + pub feedback: Option, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[cfg_attr(feature = "ts", derive(ts_rs::TS), ts(export))] +#[serde(rename_all = "camelCase")] +pub struct PermissionRespondResult { + pub request_id: String, + pub accepted: bool, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[cfg_attr(feature = "ts", derive(ts_rs::TS), ts(export))] +#[serde(rename_all = "snake_case")] +pub enum PermissionSourceKind { + ToolCall, + Provider, + Extension, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[cfg_attr(feature = "ts", derive(ts_rs::TS), ts(export))] +#[serde(rename_all = "camelCase")] +pub struct PermissionSource { + pub kind: PermissionSourceKind, + pub identity: String, +} + #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] #[cfg_attr(feature = "ts", derive(ts_rs::TS), ts(export))] #[serde(deny_unknown_fields, rename_all = "camelCase")] @@ -539,11 +592,49 @@ pub struct QueryEventParams { pub event: QueryEvent, } -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] #[cfg_attr(feature = "ts", derive(ts_rs::TS), ts(export))] -#[serde(tag = "type", rename_all = "snake_case")] +#[serde( + tag = "type", + rename_all = "snake_case", + rename_all_fields = "camelCase" +)] pub enum QueryEvent { - AssistantTextDelta { text: String }, + AssistantTextDelta { + text: String, + }, + ToolEvent { + tool_call_id: String, + tool_name: String, + status: ToolEventStatus, + #[cfg_attr(feature = "ts", ts(optional))] + #[serde(skip_serializing_if = "Option::is_none")] + progress: Option, + #[cfg_attr(feature = "ts", ts(optional))] + #[serde(skip_serializing_if = "Option::is_none")] + duration_ms: Option, + }, + PermissionRequest { + request_id: String, + action: String, + resources: Vec, + source: PermissionSource, + #[cfg_attr(feature = "ts", ts(optional))] + #[serde(skip_serializing_if = "Option::is_none")] + tool_call_id: Option, + response_timeout_ms: u64, + }, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[cfg_attr(feature = "ts", derive(ts_rs::TS), ts(export))] +#[serde(rename_all = "snake_case")] +pub enum ToolEventStatus { + Started, + Progress, + Completed, + Failed, + Cancelled, } #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] diff --git a/src/crates/interfaces/sdk-host/tests/host_lifecycle.rs b/src/crates/interfaces/sdk-host/tests/host_lifecycle.rs index fcc070a4e4..da61bef7c0 100644 --- a/src/crates/interfaces/sdk-host/tests/host_lifecycle.rs +++ b/src/crates/interfaces/sdk-host/tests/host_lifecycle.rs @@ -16,7 +16,7 @@ use bitfun_agent_runtime::sdk::{ PermissionRequestSourceKind, PortError, PortErrorKind, PortResult, }; use bitfun_core_types::ErrorCategory; -use bitfun_events::AgenticEvent; +use bitfun_events::{AgenticEvent, ToolEventData, ToolEventIdentity}; use bitfun_runtime_ports::{ ClockPort, PermissionAuditRecord, PermissionAuditStorePort, PermissionGrant, PermissionReplyStorePort, RuntimeServiceCapability, RuntimeServicePort, @@ -45,6 +45,7 @@ struct FakeOwner { queue_dialog: bool, dialog_session_override: Option, output_text: Option, + emit_tool_events: bool, block_dialog_submit: bool, block_agent_resolution: bool, block_first_cancel: bool, @@ -109,6 +110,15 @@ impl FakeOwner { } } + fn with_tool_events(queue: Arc) -> Self { + Self { + queue: Mutex::new(Some(queue)), + emit_terminal: true, + emit_tool_events: true, + ..Self::default() + } + } + fn failing_dialog(queue: Arc) -> Self { Self { queue: Mutex::new(Some(queue)), @@ -377,6 +387,44 @@ impl AgentDialogTurnPort for FakeOwner { }); } let queue = self.queue.lock().unwrap().clone().unwrap(); + if self.emit_tool_events { + queue + .enqueue( + AgenticEvent::ToolEvent { + session_id: request.session_id.clone(), + turn_id: turn_id.clone(), + round_id: "round-fixture".to_string(), + attempt_id: Some("attempt-fixture".to_string()), + attempt_index: Some(0), + tool_event: ToolEventData::Started { + identity: ToolEventIdentity::direct("tool-fixture", "Read"), + params: serde_json::json!({ "path": "must-not-leak.txt" }), + timeout_seconds: None, + }, + }, + None, + ) + .await + .unwrap(); + queue + .enqueue( + AgenticEvent::ToolEvent { + session_id: request.session_id.clone(), + turn_id: turn_id.clone(), + round_id: "round-fixture".to_string(), + attempt_id: Some("attempt-fixture".to_string()), + attempt_index: Some(0), + tool_event: ToolEventData::Progress { + identity: ToolEventIdentity::direct("tool-fixture", "Read"), + message: "must-not-leak-progress".to_string(), + percentage: 50.0, + }, + }, + None, + ) + .await + .unwrap(); + } queue .enqueue( AgenticEvent::TextChunk { @@ -394,6 +442,32 @@ impl AgentDialogTurnPort for FakeOwner { ) .await .unwrap(); + if self.emit_tool_events { + queue + .enqueue( + AgenticEvent::ToolEvent { + session_id: request.session_id.clone(), + turn_id: turn_id.clone(), + round_id: "round-fixture".to_string(), + attempt_id: Some("attempt-fixture".to_string()), + attempt_index: Some(0), + tool_event: ToolEventData::Completed { + identity: ToolEventIdentity::direct("tool-fixture", "Read"), + result: serde_json::json!({ "content": "must-not-leak" }), + result_for_assistant: None, + image_attachments: None, + duration_ms: 12, + queue_wait_ms: None, + preflight_ms: None, + confirmation_wait_ms: None, + execution_ms: Some(12), + }, + }, + None, + ) + .await + .unwrap(); + } if self.emit_terminal { queue .enqueue( @@ -596,6 +670,28 @@ fn blocking_permission_manager() -> Arc { )) } +fn permission_request_fixture(request_id: &str, order: u32, session_id: &str) -> PermissionRequest { + PermissionRequest { + request_id: request_id.to_string(), + round_id: "round-fixture".to_string(), + order, + tool_call_id: Some("tool-fixture".to_string()), + project_path: Some("D:/workspace/project".to_string()), + project_id: "project-fixture".to_string(), + session_id: session_id.to_string(), + agent_id: "agentic".to_string(), + action: "edit".to_string(), + resources: vec!["src/lib.rs".to_string()], + save_resources: Vec::new(), + source: PermissionRequestSource { + kind: PermissionRequestSourceKind::ToolCall, + identity: "edit".to_string(), + }, + delegation: None, + display_metadata: serde_json::Map::new(), + } +} + async fn host_with_query_limit( max_active_queries: usize, ) -> ( @@ -767,6 +863,38 @@ async fn host_with_temporary_model_installer( ) } +async fn host_with_tool_events() -> ( + SdkHostConnection, + Arc, + mpsc::Receiver, +) { + let queue = Arc::new(EventQueue::new(EventQueueConfig::default())); + let owner = Arc::new(FakeOwner::with_tool_events(queue.clone())); + let runtime = AgentRuntimeBuilder::new() + .with_submission_port(owner.clone()) + .with_dialog_turn_port(owner.clone()) + .with_cancellation_port(owner.clone()) + .with_turn_settlement_port(owner.clone()) + .with_session_management_port(owner.clone()) + .with_session_close_port(owner.clone()) + .with_permission_request_manager(permission_manager()) + .with_event_source(AgentEventSource::new(queue)) + .build() + .unwrap(); + let (output, receiver) = mpsc::channel(32); + ( + SdkHostConnection::new( + runtime, + "D:/workspace/project", + output, + SdkHostConfig::default(), + fake_installer(), + ), + owner, + receiver, + ) +} + async fn initialize(host: &SdkHostConnection, output: &mut mpsc::Receiver) { host.handle_request(request(serde_json::json!({ "jsonrpc": "2.0", @@ -775,7 +903,10 @@ async fn initialize(host: &SdkHostConnection, output: &mut mpsc::Receiver JsonRpcRequest { "params": { "protocolVersion": PROTOCOL_VERSION, "clientInfo": { "name": "fixture", "version": "0.1.0" }, - "capabilities": { "serverNotifications": true }, + "capabilities": { + "serverNotifications": true, + "permissionResponses": true + }, "model": { "provider": "openai", "model": "fixture-model", @@ -837,7 +971,10 @@ async fn temporary_model_is_connection_scoped_and_cannot_be_overridden() { "params": { "protocolVersion": PROTOCOL_VERSION, "clientInfo": { "name": "fixture", "version": "0.1.0" }, - "capabilities": { "serverNotifications": true }, + "capabilities": { + "serverNotifications": true, + "permissionResponses": true + }, "model": { "provider": "openai", "model": "fixture-model", @@ -1013,7 +1150,10 @@ async fn initialize_is_required_and_version_mismatch_fails_closed() { "params": { "protocolVersion": 99, "clientInfo": { "name": "fixture", "version": "0.1.0" }, - "capabilities": { "serverNotifications": true }, + "capabilities": { + "serverNotifications": true, + "permissionResponses": true + }, "model": { "provider": "openai", "model": "fixture-model", @@ -1067,6 +1207,51 @@ async fn query_streams_existing_events_and_one_terminal_result() { assert!(output.try_recv().is_err(), "terminal result must be unique"); } +#[tokio::test] +async fn query_projects_safe_tool_activity_without_raw_inputs_or_results() { + let (host, _, mut output) = host_with_tool_events().await; + initialize(&host, &mut output).await; + + host.handle_request(request(serde_json::json!({ + "jsonrpc": "2.0", + "id": "query-tools", + "method": "query/start", + "params": { "prompt": "read a file" } + }))) + .await; + + let accepted = output.recv().await.unwrap(); + let started = output.recv().await.unwrap(); + let progress = output.recv().await.unwrap(); + let text = output.recv().await.unwrap(); + let completed = output.recv().await.unwrap(); + let result = output.recv().await.unwrap(); + + assert_eq!(started["params"]["sequence"], 1); + assert_eq!(started["params"]["event"]["type"], "tool_event"); + assert_eq!(started["params"]["event"]["toolCallId"], "tool-fixture"); + assert_eq!(started["params"]["event"]["toolName"], "Read"); + assert_eq!(started["params"]["event"]["status"], "started"); + assert!(started["params"]["event"].get("params").is_none()); + + assert_eq!(progress["params"]["sequence"], 2); + assert_eq!(progress["params"]["event"]["status"], "progress"); + assert_eq!(progress["params"]["event"]["progress"], 50.0); + assert!(!serde_json::to_string(&progress) + .unwrap() + .contains("must-not-leak-progress")); + assert_eq!(text["params"]["sequence"], 3); + assert_eq!(text["params"]["event"]["type"], "assistant_text_delta"); + assert_eq!(completed["params"]["sequence"], 4); + assert_eq!(completed["params"]["event"]["status"], "completed"); + assert_eq!(completed["params"]["event"]["durationMs"], 12); + assert!(completed["params"]["event"].get("result").is_none()); + + assert_eq!(result["params"]["status"], "completed"); + assert_eq!(result["params"]["queryId"], accepted["result"]["queryId"]); + assert_eq!(result["params"]["output"]["text"], "fixture result"); +} + #[tokio::test] async fn escaped_query_output_fails_before_exceeding_the_wire_budget() { let queue = Arc::new(EventQueue::new(EventQueueConfig::default())); @@ -2521,7 +2706,7 @@ async fn query_start_rejects_if_session_close_finishes_before_reservation() { } #[tokio::test] -async fn permission_without_callback_is_rejected_and_finishes_action_required() { +async fn permission_request_is_streamed_and_can_be_allowed_once() { let queue = Arc::new(EventQueue::new(EventQueueConfig::default())); let owner = Arc::new(FakeOwner::without_terminal(queue.clone())); let permissions = permission_manager(); @@ -2541,7 +2726,10 @@ async fn permission_without_callback_is_rejected_and_finishes_action_required() runtime, "D:/workspace/project", sender, - SdkHostConfig::default(), + SdkHostConfig { + permission_response_timeout: Duration::from_millis(250), + ..SdkHostConfig::default() + }, fake_installer(), ); initialize(&host, &mut output).await; @@ -2560,25 +2748,7 @@ async fn permission_without_callback_is_rejected_and_finishes_action_required() .to_string(); assert_eq!(output.recv().await.unwrap()["method"], "query/event"); - let permission_request = PermissionRequest { - request_id: "permission-fixture".to_string(), - round_id: "round-fixture".to_string(), - order: 0, - tool_call_id: Some("tool-fixture".to_string()), - project_path: Some("D:/workspace/project".to_string()), - project_id: "project-fixture".to_string(), - session_id, - agent_id: "agentic".to_string(), - action: "edit".to_string(), - resources: vec!["src/lib.rs".to_string()], - save_resources: Vec::new(), - source: PermissionRequestSource { - kind: PermissionRequestSourceKind::ToolCall, - identity: "edit".to_string(), - }, - delegation: None, - display_metadata: serde_json::Map::new(), - }; + let permission_request = permission_request_fixture("permission-fixture", 0, &session_id); let unrelated = permissions .register_batch_for_turn( vec![PermissionRequest { @@ -2606,31 +2776,174 @@ async fn permission_without_callback_is_rejected_and_finishes_action_required() )); let pending = permissions - .register_batch_for_turn(vec![permission_request], "turn-fixture") + .register_batch_for_turn(vec![permission_request.clone()], "turn-fixture") .await .unwrap() .pop() .unwrap(); - let result = loop { - let value = output.recv().await.unwrap(); - if value["method"] == "query/result" { - break value; + let permission = output.recv().await.unwrap(); + assert_eq!(permission["method"], "query/event"); + assert_eq!(permission["params"]["event"]["type"], "permission_request"); + assert_eq!( + permission["params"]["event"]["requestId"], + "permission-fixture" + ); + assert_eq!(permission["params"]["event"]["action"], "edit"); + assert_eq!( + permission["params"]["event"]["resources"], + serde_json::json!(["src/lib.rs"]) + ); + assert_eq!(permission["params"]["event"]["source"]["kind"], "tool_call"); + + host.handle_request(request(serde_json::json!({ + "jsonrpc": "2.0", + "id": "permission-response", + "method": "permission/respond", + "params": { + "queryId": accepted["result"]["queryId"], + "sessionId": accepted["result"]["sessionId"], + "turnId": accepted["result"]["turnId"], + "operationId": accepted["result"]["operationId"], + "requestId": "permission-fixture", + "decision": "allow_once" } - }; - assert_eq!(result["params"]["status"], "failed"); - assert_eq!(result["params"]["error"]["data"]["code"], "action_required"); + }))) + .await; + let response = output.recv().await.unwrap(); + assert_eq!(response["result"]["accepted"], true); + assert_eq!(response["result"]["requestId"], "permission-fixture"); + let resolution = pending.wait().await; assert!(matches!( resolution, bitfun_agent_runtime::permission::PermissionWaitOutcome::Replied( - bitfun_agent_runtime::sdk::PermissionReply::Reject { .. } + bitfun_agent_runtime::sdk::PermissionReply::Once ) )); + + host.handle_request(request(serde_json::json!({ + "jsonrpc": "2.0", + "id": "duplicate-permission-response", + "method": "permission/respond", + "params": { + "queryId": accepted["result"]["queryId"], + "sessionId": accepted["result"]["sessionId"], + "turnId": accepted["result"]["turnId"], + "operationId": accepted["result"]["operationId"], + "requestId": "permission-fixture", + "decision": "allow_once" + } + }))) + .await; + assert_eq!( + output.recv().await.unwrap()["error"]["data"]["code"], + "not_found" + ); + + let rejected = permissions + .register_batch_for_turn( + vec![permission_request_fixture( + "permission-rejected", + 1, + accepted["result"]["sessionId"].as_str().unwrap(), + )], + "turn-fixture", + ) + .await + .unwrap() + .pop() + .unwrap(); + assert_eq!( + output.recv().await.unwrap()["params"]["event"]["requestId"], + "permission-rejected" + ); + + host.handle_request(request(serde_json::json!({ + "jsonrpc": "2.0", + "id": "wrong-query-identity", + "method": "permission/respond", + "params": { + "queryId": accepted["result"]["queryId"], + "sessionId": accepted["result"]["sessionId"], + "turnId": accepted["result"]["turnId"], + "operationId": "another-operation", + "requestId": "permission-rejected", + "decision": "reject", + "feedback": "not needed" + } + }))) + .await; + assert_eq!( + output.recv().await.unwrap()["error"]["data"]["code"], + "invalid_request" + ); + + host.handle_request(request(serde_json::json!({ + "jsonrpc": "2.0", + "id": "reject-permission", + "method": "permission/respond", + "params": { + "queryId": accepted["result"]["queryId"], + "sessionId": accepted["result"]["sessionId"], + "turnId": accepted["result"]["turnId"], + "operationId": accepted["result"]["operationId"], + "requestId": "permission-rejected", + "decision": "reject", + "feedback": "not needed" + } + }))) + .await; + assert_eq!(output.recv().await.unwrap()["result"]["accepted"], true); + assert!(matches!( + rejected.wait().await, + bitfun_agent_runtime::permission::PermissionWaitOutcome::Replied( + bitfun_agent_runtime::sdk::PermissionReply::Reject { feedback } + ) if feedback.as_deref() == Some("not needed") + )); + + let expired = permissions + .register_batch_for_turn( + vec![permission_request_fixture( + "permission-expired", + 2, + accepted["result"]["sessionId"].as_str().unwrap(), + )], + "turn-fixture", + ) + .await + .unwrap() + .pop() + .unwrap(); + assert_eq!( + output.recv().await.unwrap()["params"]["event"]["requestId"], + "permission-expired" + ); + assert!(matches!( + tokio::time::timeout(Duration::from_secs(1), expired.wait()) + .await + .expect("permission timeout must settle through the Runtime owner"), + bitfun_agent_runtime::permission::PermissionWaitOutcome::Replied( + bitfun_agent_runtime::sdk::PermissionReply::Reject { feedback } + ) if feedback.as_deref() == Some("SDK permission response timed out") + )); + + host.handle_request(request(serde_json::json!({ + "jsonrpc": "2.0", + "id": "cancel-after-permission", + "method": "query/cancel", + "params": { + "queryId": accepted["result"]["queryId"], + "sessionId": accepted["result"]["sessionId"], + "turnId": accepted["result"]["turnId"], + "operationId": accepted["result"]["operationId"] + } + }))) + .await; } #[tokio::test] -async fn stalled_permission_rejection_is_bounded_and_cancels_the_exact_turn() { +async fn stalled_user_permission_response_is_bounded_and_cancels_the_exact_turn() { let queue = Arc::new(EventQueue::new(EventQueueConfig::default())); let owner = Arc::new(FakeOwner::without_terminal(queue.clone())); let permissions = blocking_permission_manager(); @@ -2667,28 +2980,15 @@ async fn stalled_permission_rejection_is_bounded_and_cancels_the_exact_turn() { .as_str() .unwrap() .to_string(); + assert_eq!(output.recv().await.unwrap()["method"], "query/event"); let _pending = permissions .register_batch_for_turn( - vec![PermissionRequest { - request_id: "permission-stalled".to_string(), - round_id: "round-fixture".to_string(), - order: 0, - tool_call_id: Some("tool-fixture".to_string()), - project_path: Some("D:/workspace/project".to_string()), - project_id: "project-fixture".to_string(), - session_id, - agent_id: "agentic".to_string(), - action: "edit".to_string(), - resources: vec!["src/lib.rs".to_string()], - save_resources: Vec::new(), - source: PermissionRequestSource { - kind: PermissionRequestSourceKind::ToolCall, - identity: "edit".to_string(), - }, - delegation: None, - display_metadata: serde_json::Map::new(), - }], + vec![permission_request_fixture( + "permission-stalled", + 0, + &session_id, + )], "turn-fixture", ) .await @@ -2696,17 +2996,46 @@ async fn stalled_permission_rejection_is_bounded_and_cancels_the_exact_turn() { .pop() .unwrap(); - let result = tokio::time::timeout(Duration::from_secs(5), async { + let permission = output.recv().await.unwrap(); + assert_eq!( + permission["params"]["event"]["requestId"], + "permission-stalled" + ); + host.handle_request(request(serde_json::json!({ + "jsonrpc": "2.0", + "id": "stalled-permission-response", + "method": "permission/respond", + "params": { + "queryId": accepted["result"]["queryId"], + "sessionId": accepted["result"]["sessionId"], + "turnId": accepted["result"]["turnId"], + "operationId": accepted["result"]["operationId"], + "requestId": "permission-stalled", + "decision": "allow_once" + } + }))) + .await; + + let (result, response) = tokio::time::timeout(Duration::from_secs(5), async { + let mut result = None; + let mut response = None; loop { let value = output.recv().await.unwrap(); if value["method"] == "query/result" { - break value; + result = Some(value); + } else if value["id"] == "stalled-permission-response" { + response = Some(value); + } + if result.is_some() && response.is_some() { + break (result.unwrap(), response.unwrap()); } } }) .await .expect("permission rejection must remain bounded"); assert_eq!(result["params"]["error"]["data"]["code"], "timeout"); + assert_eq!(response["error"]["data"]["code"], "timeout"); + assert_eq!(response["error"]["data"]["retryable"], false); assert_eq!( owner.cancel_requests.lock().unwrap()[0].turn_id.as_deref(), Some("turn-fixture") diff --git a/src/crates/interfaces/sdk-host/tests/protocol_contracts.rs b/src/crates/interfaces/sdk-host/tests/protocol_contracts.rs index acb686c849..7c17f8e1ff 100644 --- a/src/crates/interfaces/sdk-host/tests/protocol_contracts.rs +++ b/src/crates/interfaces/sdk-host/tests/protocol_contracts.rs @@ -1,9 +1,10 @@ use bitfun_sdk_host::protocol::{ ErrorCode, ErrorData, ErrorStage, HostCapabilities, InitializeParams, InitializeResult, - JsonRpcErrorResponse, JsonRpcRequest, JsonRpcSuccessResponse, OutcomeCertainty, QueryEvent, - QueryOutput, QueryResultError, QueryResultParams, QueryTerminalStatus, RecoveryAction, - RequestId, SessionLifetime, Stability, TemporaryModelConfig, TemporaryModelProvider, - PROTOCOL_VERSION, + JsonRpcErrorResponse, JsonRpcRequest, JsonRpcSuccessResponse, OutcomeCertainty, + PermissionDecision, PermissionRespondParams, PermissionSource, PermissionSourceKind, + QueryEvent, QueryOutput, QueryResultError, QueryResultParams, QueryTerminalStatus, + RecoveryAction, RequestId, SessionLifetime, Stability, TemporaryModelConfig, + TemporaryModelProvider, ToolEventStatus, PROTOCOL_VERSION, }; #[test] @@ -13,9 +14,12 @@ fn initialize_contract_is_versioned_and_binds_one_temporary_model() { "id": 1, "method": "initialize", "params": { - "protocolVersion": 2, + "protocolVersion": 3, "clientInfo": { "name": "fixture", "version": "0.1.0" }, - "capabilities": { "serverNotifications": true }, + "capabilities": { + "serverNotifications": true, + "permissionResponses": true + }, "model": { "provider": "openai", "model": "fixture-model", @@ -28,9 +32,10 @@ fn initialize_contract_is_versioned_and_binds_one_temporary_model() { let params: InitializeParams = request.params_as().unwrap(); assert_eq!(request.id, Some(RequestId::Number(1))); - assert_eq!(PROTOCOL_VERSION, 2); + assert_eq!(PROTOCOL_VERSION, 3); assert_eq!(params.protocol_version, PROTOCOL_VERSION); assert!(params.capabilities.server_notifications); + assert!(params.capabilities.permission_responses); assert_eq!(params.model.provider, TemporaryModelProvider::Openai); assert_eq!(params.model.model, "fixture-model"); assert_eq!(params.model.api_key, "fixture-secret"); @@ -72,10 +77,11 @@ fn initialize_contract_is_versioned_and_binds_one_temporary_model() { query_cancel: true, session_close: true, event_stream: true, + tool_events: true, structured_output: false, usage: false, custom_tools: false, - permission_callbacks: false, + permission_responses: true, hooks: false, mcp_configuration: false, prestarted_transport: false, @@ -95,6 +101,7 @@ fn current_host_capabilities_are_a_deliberate_subset_of_the_headless_cli_target( assert!(capabilities.query_cancel); assert!(capabilities.session_close); assert!(capabilities.event_stream); + assert!(capabilities.tool_events); assert_eq!( capabilities.session_create_lifetime, @@ -103,7 +110,7 @@ fn current_host_capabilities_are_a_deliberate_subset_of_the_headless_cli_target( assert!(!capabilities.structured_output); assert!(!capabilities.usage); assert!(!capabilities.custom_tools); - assert!(!capabilities.permission_callbacks); + assert!(capabilities.permission_responses); assert!(!capabilities.hooks); assert!(!capabilities.mcp_configuration); assert!(!capabilities.prestarted_transport); @@ -120,6 +127,55 @@ fn query_events_and_terminal_errors_are_closed_protocol_values() { serde_json::json!({ "type": "assistant_text_delta", "text": "hello" }) ); + let tool_event = serde_json::to_value(QueryEvent::ToolEvent { + tool_call_id: "tool-1".to_string(), + tool_name: "Read".to_string(), + status: ToolEventStatus::Started, + progress: None, + duration_ms: None, + }) + .unwrap(); + assert_eq!( + tool_event, + serde_json::json!({ + "type": "tool_event", + "toolCallId": "tool-1", + "toolName": "Read", + "status": "started" + }) + ); + assert!(tool_event.get("params").is_none()); + assert!(tool_event.get("result").is_none()); + + let permission_event = serde_json::to_value(QueryEvent::PermissionRequest { + request_id: "permission-1".to_string(), + action: "edit".to_string(), + resources: vec!["src/lib.rs".to_string()], + source: PermissionSource { + kind: PermissionSourceKind::ToolCall, + identity: "edit".to_string(), + }, + tool_call_id: Some("tool-1".to_string()), + response_timeout_ms: 120_000, + }) + .unwrap(); + assert_eq!(permission_event["type"], "permission_request"); + assert_eq!(permission_event["requestId"], "permission-1"); + assert_eq!(permission_event["source"]["kind"], "tool_call"); + assert_eq!(permission_event["responseTimeoutMs"], 120_000); + + let respond: PermissionRespondParams = serde_json::from_value(serde_json::json!({ + "queryId": "query-1", + "sessionId": "session-1", + "turnId": "turn-1", + "operationId": "operation-1", + "requestId": "permission-1", + "decision": "allow_once" + })) + .unwrap(); + assert_eq!(respond.decision, PermissionDecision::AllowOnce); + assert!(respond.feedback.is_none()); + let result = serde_json::to_value(QueryResultParams { query_id: "query-1".to_string(), session_id: "session-1".to_string(), @@ -239,9 +295,12 @@ fn json_rpc_request_debug_redacts_temporary_model_secret() { "id": 1, "method": "initialize", "params": { - "protocolVersion": 2, + "protocolVersion": 3, "clientInfo": { "name": "fixture", "version": "0.1.0" }, - "capabilities": { "serverNotifications": true }, + "capabilities": { + "serverNotifications": true, + "permissionResponses": true + }, "model": { "provider": "openai", "model": "fixture-model",