diff --git a/package-lock.json b/package-lock.json index 197776a25..a1c6dc39e 100644 --- a/package-lock.json +++ b/package-lock.json @@ -798,6 +798,10 @@ "resolved": "packages/evals", "link": true }, + "node_modules/@earendil-works/pi-protocol": { + "resolved": "packages/protocol", + "link": true + }, "node_modules/@earendil-works/pi-server": { "resolved": "packages/server", "link": true @@ -5669,6 +5673,21 @@ "dev": true, "license": "MIT" }, + "packages/protocol": { + "name": "@earendil-works/pi-protocol", + "version": "0.83.0", + "license": "MIT", + "dependencies": { + "typebox": "1.3.7" + }, + "devDependencies": { + "shx": "0.4.0", + "vitest": "4.1.9" + }, + "engines": { + "node": ">=22.19.0" + } + }, "packages/server": { "name": "@earendil-works/pi-server", "version": "0.83.0", diff --git a/package.json b/package.json index 74979fc70..107d54bce 100644 --- a/package.json +++ b/package.json @@ -13,8 +13,8 @@ ], "scripts": { "clean": "npm run clean --workspaces", - "build": "cd packages/tui && npm run build && cd ../ai && npm run build && cd ../agent && npm run build && cd ../storage/sqlite-node && npm run build && cd ../../coding-agent && npm run build && cd ../server && npm run build", - "build:offline": "cd packages/tui && npm run build && cd ../ai && npm run build:offline && cd ../agent && npm run build && cd ../storage/sqlite-node && npm run build && cd ../../coding-agent && npm run build && cd ../server && npm run build", + "build": "cd packages/tui && npm run build && cd ../ai && npm run build && cd ../agent && npm run build && cd ../storage/sqlite-node && npm run build && cd ../../protocol && npm run build && cd ../coding-agent && npm run build && cd ../server && npm run build", + "build:offline": "cd packages/tui && npm run build && cd ../ai && npm run build:offline && cd ../agent && npm run build && cd ../storage/sqlite-node && npm run build && cd ../../protocol && npm run build && cd ../coding-agent && npm run build && cd ../server && npm run build", "check": "biome check --write --error-on-warnings . && npm run check:pinned-deps && npm run check:ts-imports && npm run check:shrinkwrap && npm run check:install-lock:coding-agent && tsgo --noEmit && npm run check:browser-smoke", "check:browser-smoke": "node scripts/check-browser-smoke.mjs", "check:pinned-deps": "node scripts/check-pinned-deps.mjs", diff --git a/packages/agent/src/harness/agent-harness.ts b/packages/agent/src/harness/agent-harness.ts index f97d0210e..c4eb3efe7 100644 --- a/packages/agent/src/harness/agent-harness.ts +++ b/packages/agent/src/harness/agent-harness.ts @@ -137,6 +137,8 @@ const SUBSCRIBER_EVENT_TYPE = "*"; type AgentHarnessHandler = (event: any, signal?: AbortSignal) => Promise | any; +type TrackedTaskKind = "operation" | "mutation"; + function normalizeHarnessError(error: unknown, fallbackCode: AgentHarnessError["code"]): AgentHarnessError { if (error instanceof AgentHarnessError) return error; const cause = toError(error); @@ -178,7 +180,7 @@ export class AgentHarness< readonly models: Models; private phase: AgentHarnessPhase = "idle"; private activeAbortController?: AbortController; - private activeOperationPromise?: Promise; + private readonly activeTasks = new Map, TrackedTaskKind>(); private shutdownPromise?: Promise; private isShutdown = false; private pendingSessionWrites: PendingSessionWrite[] = []; @@ -336,19 +338,46 @@ export class AgentHarness< const abortController = new AbortController(); let finish = () => {}; this.activeAbortController = abortController; - this.activeOperationPromise = new Promise((resolve) => { - finish = resolve; - }); + void this.track( + "operation", + () => + new Promise((resolve) => { + finish = resolve; + }), + ); return { signal: abortController.signal, finish: () => { this.activeAbortController = undefined; - this.activeOperationPromise = undefined; finish(); }, }; } + private async track(kind: TrackedTaskKind, operation: () => Promise): Promise { + let settle = () => {}; + const settled = new Promise((resolve) => { + settle = resolve; + }); + this.activeTasks.set(settled, kind); + try { + return await operation(); + } finally { + this.activeTasks.delete(settled); + settle(); + } + } + + private async waitForTasks(kind?: TrackedTaskKind): Promise { + while (true) { + const tasks = [...this.activeTasks].flatMap(([task, taskKind]) => + kind === undefined || kind === taskKind ? [task] : [], + ); + if (tasks.length === 0) return; + await Promise.all(tasks); + } + } + private async resolveToolContext(): Promise { if (typeof this.toolContext === "function") { return await (this.toolContext as () => TContext | Promise)(); @@ -738,15 +767,17 @@ export class AgentHarness< async appendMessage(message: AgentMessage): Promise { this.assertNotShutDown(); - try { - if (this.phase === "idle") { - await this.session.appendMessage(message); - } else { - this.pendingSessionWrites.push({ type: "message", message }); + return this.track("mutation", async () => { + try { + if (this.phase === "idle") { + await this.session.appendMessage(message); + } else { + this.pendingSessionWrites.push({ type: "message", message }); + } + } catch (error) { + throw normalizeHarnessError(error, "session"); } - } catch (error) { - throw normalizeHarnessError(error, "session"); - } + }); } async compact(customInstructions?: string): Promise { @@ -914,18 +945,20 @@ export class AgentHarness< async setModel(model: Model): Promise { this.assertNotShutDown(); - try { - const previousModel = this.model; - if (this.phase === "idle") { - await this.session.appendModelChange(model.provider, model.id); - } else { - this.pendingSessionWrites.push({ type: "model_change", provider: model.provider, modelId: model.id }); + return this.track("mutation", async () => { + try { + const previousModel = this.model; + if (this.phase === "idle") { + await this.session.appendModelChange(model.provider, model.id); + } else { + this.pendingSessionWrites.push({ type: "model_change", provider: model.provider, modelId: model.id }); + } + this.model = model; + await this.emitOwn({ type: "model_update", model, previousModel, source: "set" }); + } catch (error) { + throw normalizeHarnessError(error, "session"); } - this.model = model; - await this.emitOwn({ type: "model_update", model, previousModel, source: "set" }); - } catch (error) { - throw normalizeHarnessError(error, "session"); - } + }); } getThinkingLevel(): ThinkingLevel { @@ -934,18 +967,20 @@ export class AgentHarness< async setThinkingLevel(level: ThinkingLevel): Promise { this.assertNotShutDown(); - try { - const previousLevel = this.thinkingLevel; - if (this.phase === "idle") { - await this.session.appendThinkingLevelChange(level); - } else { - this.pendingSessionWrites.push({ type: "thinking_level_change", thinkingLevel: level }); + return this.track("mutation", async () => { + try { + const previousLevel = this.thinkingLevel; + if (this.phase === "idle") { + await this.session.appendThinkingLevelChange(level); + } else { + this.pendingSessionWrites.push({ type: "thinking_level_change", thinkingLevel: level }); + } + this.thinkingLevel = level; + await this.emitOwn({ type: "thinking_level_update", level, previousLevel }); + } catch (error) { + throw normalizeHarnessError(error, "session"); } - this.thinkingLevel = level; - await this.emitOwn({ type: "thinking_level_update", level, previousLevel }); - } catch (error) { - throw normalizeHarnessError(error, "session"); - } + }); } getTools(): TTool[] { @@ -954,6 +989,10 @@ export class AgentHarness< async setTools(tools: TTool[], activeToolNames?: string[]): Promise { this.assertNotShutDown(); + return this.track("mutation", () => this.applyTools(tools, activeToolNames)); + } + + private async applyTools(tools: TTool[], activeToolNames?: string[]): Promise { try { this.validateUniqueNames( tools.map((tool) => tool.name), @@ -990,6 +1029,10 @@ export class AgentHarness< async setActiveTools(toolNames: string[]): Promise { this.assertNotShutDown(); + return this.track("mutation", () => this.applyActiveTools(toolNames)); + } + + private async applyActiveTools(toolNames: string[]): Promise { try { this.validateToolNames(toolNames); const previousToolNames = [...this.tools.keys()]; @@ -1069,7 +1112,7 @@ export class AgentHarness< this.followUpQueue = []; this.nextTurnQueue = []; this.activeAbortController?.abort(); - this.shutdownPromise = this.waitForIdle(); + this.shutdownPromise = this.waitForTasks(); return this.shutdownPromise; } @@ -1104,7 +1147,7 @@ export class AgentHarness< } async waitForIdle(): Promise { - await this.activeOperationPromise; + await this.waitForTasks("operation"); } subscribe( diff --git a/packages/agent/test/harness/agent-harness.test.ts b/packages/agent/test/harness/agent-harness.test.ts index d678427b0..9b1ec714b 100644 --- a/packages/agent/test/harness/agent-harness.test.ts +++ b/packages/agent/test/harness/agent-harness.test.ts @@ -89,6 +89,25 @@ function createAssistantMessage(text: string): AgentMessage { }; } +class BlockingSessionStorage extends InMemorySessionStorage { + readonly allWritesStarted = deferred(); + readonly releaseWrites = deferred(); + private readonly expectedWrites: number; + private writesStarted = 0; + + constructor(expectedWrites: number) { + super(); + this.expectedWrites = expectedWrites; + } + + override async appendEntry(entry: Parameters[0]): Promise { + this.writesStarted++; + if (this.writesStarted === this.expectedWrites) this.allWritesStarted.resolve(); + await this.releaseWrites.promise; + await super.appendEntry(entry); + } +} + describe("AgentHarness", () => { it("constructs directly and exposes queue modes", () => { const session = new Session(new InMemorySessionStorage()); @@ -265,6 +284,53 @@ describe("AgentHarness", () => { expect(await session.getLeafId()).toBe(originalLeafId); }); + it("does not treat concurrent mutations as active operations", async () => { + const storage = new BlockingSessionStorage(1); + const harness = new AgentHarness({ + models, + session: new Session(storage), + model: getModel("anthropic", "claude-sonnet-4-5"), + }); + const mutation = harness.appendMessage(createUserMessage("concurrent")); + await storage.allWritesStarted.promise; + + const firstSettlement = await Promise.race([ + harness.waitForIdle().then(() => "idle" as const), + new Promise<"mutation-pending">((resolve) => setImmediate(() => resolve("mutation-pending"))), + ]); + + expect(firstSettlement).toBe("idle"); + storage.releaseWrites.resolve(); + await mutation; + }); + + it("awaits concurrent idle session mutations before shutdown resolves", async () => { + const storage = new BlockingSessionStorage(3); + const harness = new AgentHarness({ + models, + session: new Session(storage), + model: getModel("anthropic", "claude-sonnet-4-5"), + }); + const nextModel = getModel("anthropic", "claude-haiku-4-5"); + const mutations = [ + harness.appendMessage(createUserMessage("concurrent")), + harness.setModel(nextModel), + harness.setThinkingLevel("high"), + ]; + await storage.allWritesStarted.promise; + + const shutdown = harness.shutdown(); + const firstSettlement = await Promise.race([ + shutdown.then(() => "shutdown" as const), + new Promise<"writes-pending">((resolve) => setImmediate(() => resolve("writes-pending"))), + ]); + + expect(firstSettlement).toBe("writes-pending"); + storage.releaseWrites.resolve(); + await Promise.all([...mutations, shutdown]); + expect(await storage.getEntries()).toHaveLength(3); + }); + it("shuts down an idle harness without modifying its durable session", async () => { const session = new Session(new InMemorySessionStorage()); await session.appendMessage(createUserMessage("existing")); diff --git a/packages/protocol/CHANGELOG.md b/packages/protocol/CHANGELOG.md new file mode 100644 index 000000000..5a02bb6c1 --- /dev/null +++ b/packages/protocol/CHANGELOG.md @@ -0,0 +1,7 @@ +# Changelog + +## [Unreleased] + +### Added + +- Added transport-neutral CBOR protocol schemas, codecs, and length-prefixed framing for remote pi sessions. diff --git a/packages/protocol/README.md b/packages/protocol/README.md new file mode 100644 index 000000000..428f75203 --- /dev/null +++ b/packages/protocol/README.md @@ -0,0 +1,68 @@ +# @earendil-works/pi-protocol + +Runtime-neutral schemas, types, CBOR encoding, and byte-stream framing for the experimental pi protocol. + +Protocol version `2` uses binary messages with this wire layout: + +1. A four-byte unsigned big-endian payload length. +2. One definite-length CBOR item containing the message. + +The first client message is always `hello`, containing `PROTOCOL_VERSION` and a bearer token. Subsequent messages use correlated request/response envelopes and server event envelopes. Session and server snapshots are authoritative. Progress events are transient UI hints and must not be reduced into authoritative state. + +## Validated message API + +`encodeClientMessage()` and `encodeServerMessage()` validate a message and return a complete framed `Uint8Array`. The incremental decoders accept arbitrary fragmentation or coalescing, so they work with streams, sockets, and custom byte transports. + +```ts +import { + PROTOCOL_VERSION, + createServerMessageDecoder, + encodeClientMessage, + type ClientHello, +} from "@earendil-works/pi-protocol"; + +const hello: ClientHello = { + type: "hello", + version: PROTOCOL_VERSION, + token: bearerToken, +}; + +transport.send(encodeClientMessage(hello)); + +const decoder = createServerMessageDecoder({ maxFrameLength: 1024 * 1024 }); +for (const message of decoder.push(incomingChunk)) { + handleServerMessage(message); +} +decoder.end(); // Call when the byte stream closes to detect truncation. +``` + +`ClientMessageDecoder` and `ServerMessageDecoder` are also available directly. Schema violations, malformed CBOR, and invalid framing throw `ProtocolValidationError`. Validation errors do not retain rejected payloads. + +`parseClientMessage()` and `parseServerMessage()` only validate already-decoded values. They do not parse JSON strings. + +## Transport support + +Every transport carries the same complete bytes: `[uint32-be CBOR length][CBOR payload]`. Transports may split or coalesce those bytes arbitrarily. + +This package does not bundle a transport. Consumers provide a byte-stream transport that preserves byte order and reports stream closure. Custom transports must handle arbitrary frame fragmentation and coalescing. + +All transports are untrusted. Configure matching frame limits and enforce the authentication and access controls appropriate for the transport. + +## Encoding and framing + +`encodeCbor()` and `decodeCbor()` implement the protocol's strict RFC 8949 subset. `encodeFrame()` and `FrameDecoder` handle framing independently of schemas and CBOR. + +The CBOR subset supports: + +- `null` and booleans +- finite numbers, with integers restricted to JavaScript's safe range and non-integers encoded as float64 +- UTF-8 strings +- `Uint8Array` byte strings +- definite-length arrays +- definite-length maps represented by objects with unique string keys + +Undefined object properties are omitted. JSON-valued protocol fields reject CBOR byte strings and non-plain objects. Top-level undefined, undefined array entries, sparse arrays, non-finite or unsafe numbers, tags, indefinite-length items, malformed UTF-8, trailing data, excessive nesting, and oversized values are rejected. + +Default limits are 16 MiB per CBOR payload/frame, 1,000,000 array elements or map entries, and 64 nested item levels. Options can configure these limits. A frame decoder validates the declared length before buffering payload bytes. + +All schemas reject unknown object properties. The protocol is experimental and has no compatibility guarantees. diff --git a/packages/protocol/package.json b/packages/protocol/package.json new file mode 100644 index 000000000..da7189ad0 --- /dev/null +++ b/packages/protocol/package.json @@ -0,0 +1,32 @@ +{ + "name": "@earendil-works/pi-protocol", + "version": "0.83.0", + "description": "Transport-neutral CBOR protocol for remote pi sessions", + "type": "module", + "main": "./dist/index.js", + "types": "./dist/index.d.ts", + "exports": { + ".": { + "types": "./dist/index.d.ts", + "import": "./dist/index.js" + } + }, + "files": ["dist", "README.md"], + "scripts": { + "clean": "shx rm -rf dist", + "build": "tsgo -p tsconfig.build.json", + "test": "vitest --run", + "prepublishOnly": "npm run clean && npm run build" + }, + "keywords": ["pi", "protocol", "cbor", "binary", "framing"], + "author": "Earendil Works", + "license": "MIT", + "repository": { + "type": "git", + "url": "git+https://github.com/earendil-works/pi.git", + "directory": "packages/protocol" + }, + "engines": { "node": ">=22.19.0" }, + "dependencies": { "typebox": "1.3.7" }, + "devDependencies": { "shx": "0.4.0", "vitest": "4.1.9" } +} diff --git a/packages/protocol/src/cbor/decoder.ts b/packages/protocol/src/cbor/decoder.ts new file mode 100644 index 000000000..4a4b580dd --- /dev/null +++ b/packages/protocol/src/cbor/decoder.ts @@ -0,0 +1,168 @@ +import { + CborError, + type CborOptions, + type ResolvedCborOptions, + resolveOptions, + textDecoder, + UINT32_BASE, +} from "./options.ts"; + +class CborReader { + private readonly bytes: Uint8Array; + private offset = 0; + private readonly options: ResolvedCborOptions; + + constructor(bytes: Uint8Array, options: ResolvedCborOptions) { + this.bytes = bytes; + this.options = options; + } + + decode(): unknown { + const value = this.readItem(0); + if (this.offset !== this.bytes.byteLength) throw new CborError("CBOR payload contains trailing data"); + return value; + } + + private readItem(depth: number): unknown { + if (depth > this.options.maxDepth) { + throw new CborError(`CBOR nesting depth exceeds configured limit of ${this.options.maxDepth}`); + } + const initial = this.readByte(); + const majorType = initial >>> 5; + const additionalInformation = initial & 0x1f; + + switch (majorType) { + case 0: + return this.readArgument(additionalInformation); + case 1: { + const value = -1 - this.readArgument(additionalInformation); + if (!Number.isSafeInteger(value)) throw new CborError("Decoded CBOR integer is outside the safe range"); + return value; + } + case 2: { + const length = this.readLength(additionalInformation, "byte string", this.options.maxByteLength); + return new Uint8Array(this.readBytes(length)); + } + case 3: { + const length = this.readLength(additionalInformation, "text string", this.options.maxByteLength); + const bytes = this.readBytes(length); + try { + return textDecoder.decode(bytes); + } catch (_error) { + throw new CborError("CBOR text string contains invalid UTF-8"); + } + } + case 4: { + const length = this.readLength(additionalInformation, "array", this.options.maxContainerLength); + const result: unknown[] = []; + for (let index = 0; index < length; index++) result.push(this.readItem(depth + 1)); + return result; + } + case 5: { + const length = this.readLength(additionalInformation, "map", this.options.maxContainerLength); + const result: Record = {}; + const keys = new Set(); + for (let index = 0; index < length; index++) { + const key = this.readItem(depth + 1); + if (typeof key !== "string") throw new CborError("CBOR map keys must be strings"); + if (keys.has(key)) throw new CborError("CBOR map contains a duplicate key"); + keys.add(key); + Object.defineProperty(result, key, { + configurable: true, + enumerable: true, + value: this.readItem(depth + 1), + writable: true, + }); + } + return result; + } + case 6: + throw new CborError("CBOR tags are not supported"); + case 7: + return this.readSimple(additionalInformation); + default: + throw new CborError("Malformed CBOR major type"); + } + } + + private readSimple(additionalInformation: number): unknown { + switch (additionalInformation) { + case 20: + return false; + case 21: + return true; + case 22: + return null; + case 27: { + const bytes = this.readBytes(8); + const value = new DataView(bytes.buffer, bytes.byteOffset, bytes.byteLength).getFloat64(0, false); + if (!Number.isFinite(value)) throw new CborError("Decoded CBOR number must be finite"); + if (Number.isInteger(value) && !Number.isSafeInteger(value)) { + throw new CborError("Decoded CBOR integer is outside the safe range"); + } + return value; + } + case 31: + throw new CborError("CBOR break marker is not supported"); + default: + throw new CborError("Unsupported CBOR simple value or floating-point width"); + } + } + + private readLength(additionalInformation: number, kind: string, limit: number): number { + if (additionalInformation === 31) throw new CborError(`Indefinite-length CBOR ${kind}s are not supported`); + const length = this.readArgument(additionalInformation); + if (length > limit) throw new CborError(`CBOR ${kind} length exceeds configured limit of ${limit}`); + return length; + } + + private readArgument(additionalInformation: number): number { + if (additionalInformation < 24) return additionalInformation; + switch (additionalInformation) { + case 24: + return this.readByte(); + case 25: { + const bytes = this.readBytes(2); + return bytes[0]! * 0x100 + bytes[1]!; + } + case 26: { + const bytes = this.readBytes(4); + return bytes[0]! * 0x1_000_000 + bytes[1]! * 0x1_0000 + bytes[2]! * 0x100 + bytes[3]!; + } + case 27: { + const high = this.readArgument(26); + const low = this.readArgument(26); + if (high > 0x1f_ffff) throw new CborError("Decoded CBOR integer or length is outside the safe range"); + return high * UINT32_BASE + low; + } + case 31: + throw new CborError("Indefinite-length CBOR items are not supported"); + default: + throw new CborError("Malformed CBOR additional information"); + } + } + + private readByte(): number { + if (this.offset >= this.bytes.byteLength) throw new CborError("Truncated CBOR payload"); + const value = this.bytes[this.offset]!; + this.offset++; + return value; + } + + private readBytes(length: number): Uint8Array { + if (length > this.bytes.byteLength - this.offset) throw new CborError("Truncated CBOR payload"); + const value = this.bytes.subarray(this.offset, this.offset + length); + this.offset += length; + return value; + } +} + +/** Decodes exactly one item from the protocol's strict RFC 8949 subset. */ +export function decodeCbor(bytes: Uint8Array, options?: CborOptions): unknown { + if (!(bytes instanceof Uint8Array)) throw new TypeError("CBOR input must be a Uint8Array"); + const resolved = resolveOptions(options); + if (bytes.byteLength > resolved.maxByteLength) { + throw new CborError(`CBOR byte length exceeds configured limit of ${resolved.maxByteLength}`); + } + return new CborReader(bytes, resolved).decode(); +} diff --git a/packages/protocol/src/cbor/encoder.ts b/packages/protocol/src/cbor/encoder.ts new file mode 100644 index 000000000..892910f7a --- /dev/null +++ b/packages/protocol/src/cbor/encoder.ts @@ -0,0 +1,216 @@ +import { + CborError, + type CborOptions, + MAX_UINT32, + type ResolvedCborOptions, + resolveOptions, + textDecoder, + textEncoder, + UINT32_BASE, +} from "./options.ts"; + +class CborWriter { + private buffer: Uint8Array; + private offset = 0; + private readonly maxByteLength: number; + + constructor(maxByteLength: number) { + this.maxByteLength = maxByteLength; + this.buffer = new Uint8Array(Math.min(256, maxByteLength)); + } + + writeByte(value: number): void { + this.ensureCapacity(1); + this.buffer[this.offset] = value; + this.offset++; + } + + writeBytes(bytes: Uint8Array): void { + this.ensureCapacity(bytes.byteLength); + this.buffer.set(bytes, this.offset); + this.offset += bytes.byteLength; + } + + writeUint16(value: number): void { + this.ensureCapacity(2); + this.buffer[this.offset] = value >>> 8; + this.buffer[this.offset + 1] = value; + this.offset += 2; + } + + writeUint32(value: number): void { + this.ensureCapacity(4); + this.buffer[this.offset] = value >>> 24; + this.buffer[this.offset + 1] = value >>> 16; + this.buffer[this.offset + 2] = value >>> 8; + this.buffer[this.offset + 3] = value; + this.offset += 4; + } + + writeUint64(value: number): void { + const high = Math.floor(value / UINT32_BASE); + const low = value - high * UINT32_BASE; + this.writeUint32(high); + this.writeUint32(low); + } + + writeFloat64(value: number): void { + this.ensureCapacity(9); + this.buffer[this.offset] = 0xfb; + new DataView(this.buffer.buffer).setFloat64(this.offset + 1, value, false); + this.offset += 9; + } + + finish(): Uint8Array { + return this.buffer.slice(0, this.offset); + } + + private ensureCapacity(additionalBytes: number): void { + const required = this.offset + additionalBytes; + if (required > this.maxByteLength) { + throw new CborError(`CBOR byte length exceeds configured limit of ${this.maxByteLength}`); + } + if (required <= this.buffer.byteLength) return; + + let capacity = Math.max(1, this.buffer.byteLength); + while (capacity < required) capacity = Math.min(this.maxByteLength, Math.max(required, capacity * 2)); + const expanded = new Uint8Array(capacity); + expanded.set(this.buffer); + this.buffer = expanded; + } +} + +function writeArgument(writer: CborWriter, majorType: number, value: number): void { + const prefix = majorType << 5; + if (value < 24) { + writer.writeByte(prefix | value); + } else if (value <= 0xff) { + writer.writeByte(prefix | 24); + writer.writeByte(value); + } else if (value <= 0xffff) { + writer.writeByte(prefix | 25); + writer.writeUint16(value); + } else if (value <= MAX_UINT32) { + writer.writeByte(prefix | 26); + writer.writeUint32(value); + } else { + writer.writeByte(prefix | 27); + writer.writeUint64(value); + } +} + +function isPlainObject(value: unknown): value is Record { + if (typeof value !== "object" || value === null) return false; + const prototype = Object.getPrototypeOf(value); + return prototype === Object.prototype || prototype === null; +} + +function encodeText(writer: CborWriter, value: string, options: ResolvedCborOptions): void { + const bytes = textEncoder.encode(value); + if (bytes.byteLength > options.maxByteLength) { + throw new CborError(`CBOR text string length exceeds configured limit of ${options.maxByteLength}`); + } + if (textDecoder.decode(bytes) !== value) + throw new CborError("CBOR text strings must contain valid Unicode scalar values"); + writeArgument(writer, 3, bytes.byteLength); + writer.writeBytes(bytes); +} + +function encodeValue( + writer: CborWriter, + value: unknown, + options: ResolvedCborOptions, + depth: number, + ancestors: Set, +): void { + if (depth > options.maxDepth) + throw new CborError(`CBOR nesting depth exceeds configured limit of ${options.maxDepth}`); + + if (value === null) { + writer.writeByte(0xf6); + return; + } + if (typeof value === "boolean") { + writer.writeByte(value ? 0xf5 : 0xf4); + return; + } + if (typeof value === "number") { + if (!Number.isFinite(value)) throw new CborError("CBOR numbers must be finite"); + if (Number.isInteger(value) && !Object.is(value, -0)) { + if (!Number.isSafeInteger(value)) throw new CborError("CBOR integers must be safe JavaScript integers"); + if (value >= 0) writeArgument(writer, 0, value); + else writeArgument(writer, 1, -1 - value); + } else { + writer.writeFloat64(value); + } + return; + } + if (typeof value === "string") { + encodeText(writer, value, options); + return; + } + if (value instanceof Uint8Array) { + if (value.byteLength > options.maxByteLength) { + throw new CborError(`CBOR byte string length exceeds configured limit of ${options.maxByteLength}`); + } + writeArgument(writer, 2, value.byteLength); + writer.writeBytes(value); + return; + } + if (Array.isArray(value)) { + if (ancestors.has(value)) throw new CborError("CBOR values must not contain cycles"); + if (value.length > options.maxContainerLength) { + throw new CborError(`CBOR array length exceeds configured limit of ${options.maxContainerLength}`); + } + ancestors.add(value); + try { + writeArgument(writer, 4, value.length); + for (let index = 0; index < value.length; index++) { + if (!Object.hasOwn(value, index) || value[index] === undefined) { + throw new CborError("CBOR arrays must not contain holes or undefined values"); + } + encodeValue(writer, value[index], options, depth + 1, ancestors); + } + } finally { + ancestors.delete(value); + } + return; + } + if (isPlainObject(value)) { + if (ancestors.has(value)) throw new CborError("CBOR values must not contain cycles"); + for (const symbol of Object.getOwnPropertySymbols(value)) { + if (Object.prototype.propertyIsEnumerable.call(value, symbol)) { + throw new CborError("CBOR map keys must be strings"); + } + } + const entries: Array = []; + for (const key of Object.keys(value)) { + const entryValue = value[key]; + if (entryValue !== undefined) entries.push([key, entryValue]); + } + if (entries.length > options.maxContainerLength) { + throw new CborError(`CBOR map length exceeds configured limit of ${options.maxContainerLength}`); + } + ancestors.add(value); + try { + writeArgument(writer, 5, entries.length); + for (const [key, entryValue] of entries) { + encodeText(writer, key, options); + encodeValue(writer, entryValue, options, depth + 1, ancestors); + } + } finally { + ancestors.delete(value); + } + return; + } + + throw new CborError(`Unsupported CBOR value type: ${typeof value}`); +} + +/** Encodes the protocol's strict, definite-length RFC 8949 subset. */ +export function encodeCbor(value: unknown, options?: CborOptions): Uint8Array { + const resolved = resolveOptions(options); + const writer = new CborWriter(resolved.maxByteLength); + encodeValue(writer, value, resolved, 0, new Set()); + return writer.finish(); +} diff --git a/packages/protocol/src/cbor/index.ts b/packages/protocol/src/cbor/index.ts new file mode 100644 index 000000000..e1eb92b1e --- /dev/null +++ b/packages/protocol/src/cbor/index.ts @@ -0,0 +1,9 @@ +export { decodeCbor } from "./decoder.ts"; +export { encodeCbor } from "./encoder.ts"; +export { + CborError, + type CborOptions, + DEFAULT_MAX_CBOR_BYTE_LENGTH, + DEFAULT_MAX_CBOR_CONTAINER_LENGTH, + DEFAULT_MAX_CBOR_DEPTH, +} from "./options.ts"; diff --git a/packages/protocol/src/cbor/options.ts b/packages/protocol/src/cbor/options.ts new file mode 100644 index 000000000..a831768f2 --- /dev/null +++ b/packages/protocol/src/cbor/options.ts @@ -0,0 +1,52 @@ +export const UINT32_BASE = 0x1_0000_0000; +export const MAX_UINT32 = 0xffff_ffff; +const MAX_CONFIGURED_DEPTH = 512; + +/** Safe defaults for untrusted protocol payloads. */ +export const DEFAULT_MAX_CBOR_BYTE_LENGTH = 16 * 1024 * 1024; +export const DEFAULT_MAX_CBOR_CONTAINER_LENGTH = 1_000_000; +export const DEFAULT_MAX_CBOR_DEPTH = 64; + +export interface CborOptions { + /** Maximum encoded input/output bytes and maximum byte/text string length. */ + maxByteLength?: number; + /** Maximum number of elements in an array or entries in a map. */ + maxContainerLength?: number; + /** Maximum recursive item depth. */ + maxDepth?: number; +} + +export interface ResolvedCborOptions { + maxByteLength: number; + maxContainerLength: number; + maxDepth: number; +} + +export class CborError extends Error { + constructor(message: string) { + super(message); + this.name = "CborError"; + } +} + +export const textEncoder = new TextEncoder(); +export const textDecoder = new TextDecoder("utf-8", { fatal: true, ignoreBOM: true }); + +function resolveLimit(name: string, value: number, maximum: number): number { + if (!Number.isSafeInteger(value) || value < 0 || value > maximum) { + throw new RangeError(`${name} must be an integer between 0 and ${maximum}`); + } + return value; +} + +export function resolveOptions(options: CborOptions | undefined): ResolvedCborOptions { + return { + maxByteLength: resolveLimit("maxByteLength", options?.maxByteLength ?? DEFAULT_MAX_CBOR_BYTE_LENGTH, MAX_UINT32), + maxContainerLength: resolveLimit( + "maxContainerLength", + options?.maxContainerLength ?? DEFAULT_MAX_CBOR_CONTAINER_LENGTH, + MAX_UINT32, + ), + maxDepth: resolveLimit("maxDepth", options?.maxDepth ?? DEFAULT_MAX_CBOR_DEPTH, MAX_CONFIGURED_DEPTH), + }; +} diff --git a/packages/protocol/src/codec.ts b/packages/protocol/src/codec.ts new file mode 100644 index 000000000..b83ae9853 --- /dev/null +++ b/packages/protocol/src/codec.ts @@ -0,0 +1,172 @@ +import { Check } from "typebox/value"; +import { decodeCbor, encodeCbor } from "./cbor/index.ts"; +import { + assertCompleteFrame, + DEFAULT_MAX_FRAME_LENGTH, + encodeFrame, + FrameDecoder, + type FrameDecoderOptions, +} from "./framing.ts"; +import { + type ClientMessage, + ClientMessageSchema, + PROTOCOL_VERSION, + type ServerMessage, + ServerMessageSchema, +} from "./schemas.ts"; + +export class ProtocolValidationError extends Error { + constructor(message: string, _value?: unknown) { + super(message); + this.name = "ProtocolValidationError"; + } +} + +function isProtocolValue(value: unknown, optionalProperty = false, ancestors = new Set()): boolean { + if (value === undefined) return optionalProperty; + if (value === null || typeof value === "boolean" || typeof value === "number" || typeof value === "string") { + return true; + } + if (typeof value !== "object" || ancestors.has(value)) return false; + ancestors.add(value); + try { + if (Array.isArray(value)) return value.every((item) => isProtocolValue(item, false, ancestors)); + if (Object.getPrototypeOf(value) !== Object.prototype) return false; + return Object.values(value).every((item) => isProtocolValue(item, true, ancestors)); + } finally { + ancestors.delete(value); + } +} + +export function parseClientMessage(value: unknown): ClientMessage { + if (!isProtocolValue(value) || !Check(ClientMessageSchema, value)) { + throw new ProtocolValidationError("Invalid client protocol message"); + } + return value; +} + +export function parseServerMessage(value: unknown): ServerMessage { + if (!isProtocolValue(value) || !Check(ServerMessageSchema, value)) { + throw new ProtocolValidationError("Invalid server protocol message"); + } + return value; +} + +function boundedErrorMessage(error: unknown): string { + if (!(error instanceof Error)) return "Unknown codec error"; + return error.message.length <= 500 ? error.message : `${error.message.slice(0, 497)}...`; +} + +function encodeProtocolMessage( + value: T, + parse: (candidate: unknown) => T, + kind: string, + options?: FrameDecoderOptions, +): Uint8Array { + const validated = parse(value); + try { + const maxFrameLength = options?.maxFrameLength ?? DEFAULT_MAX_FRAME_LENGTH; + const frame = encodeFrame(encodeCbor(validated, { maxByteLength: maxFrameLength })); + assertCompleteFrame(frame, { maxFrameLength }); + return frame; + } catch (error) { + if (error instanceof ProtocolValidationError) throw error; + throw new ProtocolValidationError(`Unable to encode ${kind} protocol message: ${boundedErrorMessage(error)}`); + } +} + +/** Validates and encodes one complete length-prefixed client message. */ +export function encodeClientMessage(message: ClientMessage, options?: FrameDecoderOptions): Uint8Array { + return encodeProtocolMessage(message, parseClientMessage, "client", options); +} + +/** Validates and encodes one complete length-prefixed server message. */ +export function encodeServerMessage(message: ServerMessage, options?: FrameDecoderOptions): Uint8Array { + return encodeProtocolMessage(message, parseServerMessage, "server", options); +} + +class ValidatedMessageDecoder { + private failed = false; + private readonly frames: FrameDecoder; + private readonly kind: string; + private readonly maxFrameLength: number; + private readonly parse: (candidate: unknown) => T; + + constructor(kind: string, parse: (candidate: unknown) => T, options?: FrameDecoderOptions) { + this.frames = new FrameDecoder(options); + this.kind = kind; + this.maxFrameLength = options?.maxFrameLength ?? DEFAULT_MAX_FRAME_LENGTH; + this.parse = parse; + } + + push(chunk: Uint8Array): T[] { + if (this.failed) throw new ProtocolValidationError(`${this.kind} message decoder has failed`); + try { + const messages: T[] = []; + for (const frame of this.frames.push(chunk)) { + messages.push(this.parse(decodeCbor(frame, { maxByteLength: this.maxFrameLength }))); + } + return messages; + } catch (error) { + this.failed = true; + if (error instanceof ProtocolValidationError) throw error; + throw new ProtocolValidationError(`Invalid ${this.kind} protocol frame: ${boundedErrorMessage(error)}`); + } + } + + end(): void { + if (this.failed) throw new ProtocolValidationError(`${this.kind} message decoder has failed`); + try { + this.frames.end(); + } catch (error) { + this.failed = true; + throw new ProtocolValidationError(`Invalid ${this.kind} protocol framing: ${boundedErrorMessage(error)}`); + } + } +} + +/** Incrementally decodes and validates framed client messages. */ +export class ClientMessageDecoder { + private readonly decoder: ValidatedMessageDecoder; + + constructor(options?: FrameDecoderOptions) { + this.decoder = new ValidatedMessageDecoder("client", parseClientMessage, options); + } + + push(chunk: Uint8Array): ClientMessage[] { + return this.decoder.push(chunk); + } + + end(): void { + this.decoder.end(); + } +} + +/** Incrementally decodes and validates framed server messages. */ +export class ServerMessageDecoder { + private readonly decoder: ValidatedMessageDecoder; + + constructor(options?: FrameDecoderOptions) { + this.decoder = new ValidatedMessageDecoder("server", parseServerMessage, options); + } + + push(chunk: Uint8Array): ServerMessage[] { + return this.decoder.push(chunk); + } + + end(): void { + this.decoder.end(); + } +} + +export function createClientMessageDecoder(options?: FrameDecoderOptions): ClientMessageDecoder { + return new ClientMessageDecoder(options); +} + +export function createServerMessageDecoder(options?: FrameDecoderOptions): ServerMessageDecoder { + return new ServerMessageDecoder(options); +} + +export function isSupportedProtocolVersion(version: number): version is typeof PROTOCOL_VERSION { + return Number.isInteger(version) && version === PROTOCOL_VERSION; +} diff --git a/packages/protocol/src/framing.ts b/packages/protocol/src/framing.ts new file mode 100644 index 000000000..6751e286d --- /dev/null +++ b/packages/protocol/src/framing.ts @@ -0,0 +1,165 @@ +const FRAME_HEADER_LENGTH = 4; +const MAX_UINT32 = 0xffff_ffff; +const PAYLOAD_BLOCK_SIZE = 64 * 1024; + +/** Default upper bound for one framed CBOR payload. */ +export const DEFAULT_MAX_FRAME_LENGTH = 16 * 1024 * 1024; + +export interface FrameDecoderOptions { + maxFrameLength?: number; +} + +export class FrameError extends Error { + constructor(message: string) { + super(message); + this.name = "FrameError"; + } +} + +function resolveMaxFrameLength(options: FrameDecoderOptions | undefined): number { + const value = options?.maxFrameLength ?? DEFAULT_MAX_FRAME_LENGTH; + if (!Number.isSafeInteger(value) || value < 0 || value > MAX_UINT32) { + throw new RangeError(`maxFrameLength must be an integer between 0 and ${MAX_UINT32}`); + } + return value; +} + +/** Prefixes a payload with its unsigned 32-bit big-endian byte length. */ +export function encodeFrame(payload: Uint8Array): Uint8Array { + if (!(payload instanceof Uint8Array)) throw new TypeError("Frame payload must be a Uint8Array"); + if (payload.byteLength > MAX_UINT32) throw new RangeError("Frame payload exceeds the unsigned 32-bit length limit"); + const frame = new Uint8Array(FRAME_HEADER_LENGTH + payload.byteLength); + const length = payload.byteLength; + frame[0] = length >>> 24; + frame[1] = length >>> 16; + frame[2] = length >>> 8; + frame[3] = length; + frame.set(payload, FRAME_HEADER_LENGTH); + return frame; +} + +/** Validates that bytes contain exactly one complete frame within the configured limit. */ +export function assertCompleteFrame(frame: Uint8Array, options?: FrameDecoderOptions): void { + if (!(frame instanceof Uint8Array)) throw new TypeError("Frame must be a Uint8Array"); + if (frame.byteLength < FRAME_HEADER_LENGTH) throw new FrameError("Frame does not contain a complete length prefix"); + const length = frame[0]! * 0x1_000_000 + frame[1]! * 0x1_0000 + frame[2]! * 0x100 + frame[3]!; + const maxFrameLength = resolveMaxFrameLength(options); + if (length > maxFrameLength) { + throw new FrameError(`Frame length ${length} exceeds configured limit of ${maxFrameLength}`); + } + if (frame.byteLength !== FRAME_HEADER_LENGTH + length) { + throw new FrameError("Frame must contain exactly one complete payload"); + } +} + +type DecoderState = "open" | "ended" | "failed"; + +/** Incrementally splits arbitrary byte chunks into length-prefixed payloads. */ +export class FrameDecoder { + private readonly header = new Uint8Array(FRAME_HEADER_LENGTH); + private headerLength = 0; + private readonly maxFrameLength: number; + private payloadBlocks: Uint8Array[] = []; + private currentPayloadBlock: Uint8Array | undefined; + private currentPayloadBlockLength = 0; + private expectedPayloadLength: number | undefined; + private payloadLength = 0; + private state: DecoderState = "open"; + + constructor(options?: FrameDecoderOptions) { + this.maxFrameLength = resolveMaxFrameLength(options); + } + + push(chunk: Uint8Array): Uint8Array[] { + if (this.state === "ended") throw new FrameError("Frame decoder has ended"); + if (this.state === "failed") throw new FrameError("Frame decoder has failed"); + if (!(chunk instanceof Uint8Array)) throw new TypeError("Frame chunk must be a Uint8Array"); + + const frames: Uint8Array[] = []; + let chunkOffset = 0; + while (chunkOffset < chunk.byteLength) { + if (this.expectedPayloadLength === undefined) { + const headerBytes = Math.min(FRAME_HEADER_LENGTH - this.headerLength, chunk.byteLength - chunkOffset); + this.header.set(chunk.subarray(chunkOffset, chunkOffset + headerBytes), this.headerLength); + this.headerLength += headerBytes; + chunkOffset += headerBytes; + if (this.headerLength < FRAME_HEADER_LENGTH) continue; + + const frameLength = + this.header[0]! * 0x1_000_000 + this.header[1]! * 0x1_0000 + this.header[2]! * 0x100 + this.header[3]!; + this.headerLength = 0; + if (frameLength > this.maxFrameLength) { + this.fail(`Frame length ${frameLength} exceeds configured limit of ${this.maxFrameLength}`); + } + if (frameLength === 0) { + frames.push(new Uint8Array()); + continue; + } + this.expectedPayloadLength = frameLength; + this.payloadBlocks = []; + this.currentPayloadBlock = undefined; + this.currentPayloadBlockLength = 0; + this.payloadLength = 0; + } + + const expectedPayloadLength = this.expectedPayloadLength; + if (expectedPayloadLength === undefined) continue; + while (chunkOffset < chunk.byteLength && this.payloadLength < expectedPayloadLength) { + let block = this.currentPayloadBlock; + if (!block || this.currentPayloadBlockLength === block.byteLength) { + block = new Uint8Array(Math.min(PAYLOAD_BLOCK_SIZE, expectedPayloadLength - this.payloadLength)); + this.payloadBlocks.push(block); + this.currentPayloadBlock = block; + this.currentPayloadBlockLength = 0; + } + const payloadBytes = Math.min( + block.byteLength - this.currentPayloadBlockLength, + chunk.byteLength - chunkOffset, + ); + block.set(chunk.subarray(chunkOffset, chunkOffset + payloadBytes), this.currentPayloadBlockLength); + this.currentPayloadBlockLength += payloadBytes; + this.payloadLength += payloadBytes; + chunkOffset += payloadBytes; + } + if (this.payloadLength === expectedPayloadLength) { + if (this.payloadBlocks.length === 1) { + frames.push(this.payloadBlocks[0]!); + } else { + const payload = new Uint8Array(expectedPayloadLength); + let offset = 0; + for (const payloadBlock of this.payloadBlocks) { + payload.set(payloadBlock, offset); + offset += payloadBlock.byteLength; + } + frames.push(payload); + } + this.payloadBlocks = []; + this.currentPayloadBlock = undefined; + this.currentPayloadBlockLength = 0; + this.expectedPayloadLength = undefined; + this.payloadLength = 0; + } + } + return frames; + } + + end(): void { + if (this.state === "ended") throw new FrameError("Frame decoder has ended"); + if (this.state === "failed") throw new FrameError("Frame decoder has failed"); + if (this.headerLength !== 0 || this.expectedPayloadLength !== undefined) { + this.fail("Truncated frame at end of stream"); + } + this.state = "ended"; + } + + private fail(message: string): never { + this.state = "failed"; + this.headerLength = 0; + this.payloadBlocks = []; + this.currentPayloadBlock = undefined; + this.currentPayloadBlockLength = 0; + this.expectedPayloadLength = undefined; + this.payloadLength = 0; + throw new FrameError(message); + } +} diff --git a/packages/protocol/src/index.ts b/packages/protocol/src/index.ts new file mode 100644 index 000000000..411641ef8 --- /dev/null +++ b/packages/protocol/src/index.ts @@ -0,0 +1,4 @@ +export * from "./cbor/index.ts"; +export * from "./codec.ts"; +export * from "./framing.ts"; +export * from "./schemas.ts"; diff --git a/packages/protocol/src/schemas.ts b/packages/protocol/src/schemas.ts new file mode 100644 index 000000000..16cde6320 --- /dev/null +++ b/packages/protocol/src/schemas.ts @@ -0,0 +1,412 @@ +import Type, { type Static } from "typebox"; + +export const PROTOCOL_VERSION = 2 as const; + +const IdSchema = Type.String({ minLength: 1 }); +const TimestampSchema = Type.Integer({ minimum: 0 }); +const StrictObject = [0]>(properties: T) => + Type.Object(properties, { additionalProperties: false }); + +export type JsonValue = null | boolean | number | string | JsonValue[] | { [key: string]: JsonValue }; +const JsonValueRecursiveSchema = Type.Cyclic( + { + JsonValue: Type.Union([ + Type.Null(), + Type.Boolean(), + Type.Number(), + Type.String(), + Type.Array(Type.Ref("JsonValue")), + Type.Record(Type.String(), Type.Ref("JsonValue")), + ]), + }, + "JsonValue", +); +export const JsonValueSchema = Type.Unsafe(JsonValueRecursiveSchema); + +export const ThinkingLevelSchema = Type.Union([ + Type.Literal("off"), + Type.Literal("minimal"), + Type.Literal("low"), + Type.Literal("medium"), + Type.Literal("high"), + Type.Literal("xhigh"), + Type.Literal("max"), +]); +export type ThinkingLevel = Static; + +/** Matches AgentHarnessPhase so adapters do not need a second phase vocabulary. */ +export const SessionPhaseSchema = Type.Union([ + Type.Literal("idle"), + Type.Literal("turn"), + Type.Literal("compaction"), + Type.Literal("branch_summary"), + Type.Literal("retry"), +]); +export type SessionPhase = Static; + +export const ModelRefSchema = StrictObject({ + provider: IdSchema, + id: IdSchema, +}); +export type ModelRef = Static; + +export const ModelCostSchema = StrictObject({ + input: Type.Number({ minimum: 0 }), + output: Type.Number({ minimum: 0 }), + cacheRead: Type.Number({ minimum: 0 }), + cacheWrite: Type.Number({ minimum: 0 }), +}); + +export const ModelMetadataSchema = StrictObject({ + provider: IdSchema, + id: IdSchema, + name: Type.String({ minLength: 1 }), + api: IdSchema, + reasoning: Type.Boolean(), + input: Type.Array(Type.Union([Type.Literal("text"), Type.Literal("image")])), + contextWindow: Type.Integer({ minimum: 1 }), + maxTokens: Type.Integer({ minimum: 1 }), + cost: ModelCostSchema, + supportedThinkingLevels: Type.Array(ThinkingLevelSchema, { minItems: 1 }), + authenticated: Type.Boolean(), +}); +export type ModelMetadata = Static; + +export const TextContentSchema = StrictObject({ + type: Type.Literal("text"), + text: Type.String(), +}); +export const ThinkingContentSchema = StrictObject({ + type: Type.Literal("thinking"), + thinking: Type.String(), + redacted: Type.Optional(Type.Boolean()), +}); +export const ImageContentSchema = StrictObject({ + type: Type.Literal("image"), + data: Type.String(), + mimeType: Type.String({ minLength: 1 }), +}); +export const ToolCallContentSchema = StrictObject({ + type: Type.Literal("tool_call"), + toolCallId: IdSchema, + toolName: IdSchema, + input: JsonValueSchema, +}); +export const UserContentSchema = Type.Union([TextContentSchema, ImageContentSchema]); +export const AssistantContentSchema = Type.Union([TextContentSchema, ThinkingContentSchema, ToolCallContentSchema]); +export const ToolContentSchema = Type.Union([TextContentSchema, ImageContentSchema]); +export type TextContent = Static; +export type ThinkingContent = Static; +export type ImageContent = Static; +export type ToolCallContent = Static; + +export const UsageSchema = StrictObject({ + input: Type.Integer({ minimum: 0 }), + output: Type.Integer({ minimum: 0 }), + cacheRead: Type.Integer({ minimum: 0 }), + cacheWrite: Type.Integer({ minimum: 0 }), + reasoning: Type.Optional(Type.Integer({ minimum: 0 })), + totalTokens: Type.Integer({ minimum: 0 }), + cost: StrictObject({ + input: Type.Number({ minimum: 0 }), + output: Type.Number({ minimum: 0 }), + cacheRead: Type.Number({ minimum: 0 }), + cacheWrite: Type.Number({ minimum: 0 }), + total: Type.Number({ minimum: 0 }), + }), +}); +export type Usage = Static; + +export const UserTranscriptItemSchema = StrictObject({ + id: IdSchema, + role: Type.Literal("user"), + content: Type.Array(UserContentSchema), + timestamp: TimestampSchema, +}); +export const AssistantTranscriptItemSchema = StrictObject({ + id: IdSchema, + role: Type.Literal("assistant"), + content: Type.Array(AssistantContentSchema), + status: Type.Union([ + Type.Literal("streaming"), + Type.Literal("complete"), + Type.Literal("error"), + Type.Literal("aborted"), + ]), + model: ModelRefSchema, + responseModel: Type.Optional(Type.String({ minLength: 1 })), + usage: Type.Optional(UsageSchema), + stopReason: Type.Optional( + Type.Union([ + Type.Literal("stop"), + Type.Literal("length"), + Type.Literal("tool_use"), + Type.Literal("error"), + Type.Literal("aborted"), + ]), + ), + errorMessage: Type.Optional(Type.String()), + timestamp: TimestampSchema, +}); +export const ToolTranscriptItemSchema = StrictObject({ + id: IdSchema, + role: Type.Literal("tool"), + toolCallId: IdSchema, + toolName: IdSchema, + input: JsonValueSchema, + content: Type.Array(ToolContentSchema), + details: Type.Optional(JsonValueSchema), + status: Type.Union([Type.Literal("running"), Type.Literal("complete"), Type.Literal("error")]), + isError: Type.Boolean(), + usage: Type.Optional(UsageSchema), + timestamp: TimestampSchema, +}); +export const TranscriptItemSchema = Type.Union([ + UserTranscriptItemSchema, + AssistantTranscriptItemSchema, + ToolTranscriptItemSchema, +]); +export type UserTranscriptItem = Static; +export type AssistantTranscriptItem = Static; +export type ToolTranscriptItem = Static; +export type TranscriptItem = Static; + +/** Normalized incremental activity. Snapshots remain authoritative. */ +export const TranscriptProgressSchema = Type.Union([ + StrictObject({ + type: Type.Literal("item_started"), + item: TranscriptItemSchema, + }), + StrictObject({ + type: Type.Literal("assistant_delta"), + messageId: IdSchema, + contentIndex: Type.Integer({ minimum: 0 }), + kind: Type.Union([Type.Literal("text"), Type.Literal("thinking"), Type.Literal("tool_call")]), + delta: Type.String(), + }), + StrictObject({ + type: Type.Literal("item_updated"), + item: Type.Union([AssistantTranscriptItemSchema, ToolTranscriptItemSchema]), + }), + StrictObject({ + type: Type.Literal("item_finished"), + item: Type.Union([AssistantTranscriptItemSchema, ToolTranscriptItemSchema]), + }), +]); +export type TranscriptProgress = Static; + +const SessionSummaryProperties = { + id: IdSchema, + name: Type.Optional(Type.String()), + cwd: Type.String({ minLength: 1 }), + createdAt: TimestampSchema, + updatedAt: TimestampSchema, + phase: SessionPhaseSchema, + model: ModelRefSchema, + thinkingLevel: ThinkingLevelSchema, + attached: Type.Boolean(), + locked: Type.Boolean(), +} as const; + +export const SessionSummarySchema = StrictObject(SessionSummaryProperties); +export const SessionSnapshotSchema = StrictObject({ + ...SessionSummaryProperties, + revision: Type.Integer({ minimum: 0 }), + transcript: Type.Array(TranscriptItemSchema), + queuedSteer: Type.Array(UserTranscriptItemSchema), + queuedSteerCount: Type.Integer({ minimum: 0 }), +}); +export type SessionSummary = Static; +export type SessionSnapshot = Static; + +export const ServerSnapshotSchema = StrictObject({ + serverId: IdSchema, + protocolVersion: Type.Literal(PROTOCOL_VERSION), + revision: Type.Integer({ minimum: 0 }), + sessions: Type.Array(SessionSummarySchema), + models: Type.Array(ModelMetadataSchema), +}); +export type ServerSnapshot = Static; + +export const ProtocolErrorCodeSchema = Type.Union([ + Type.Literal("auth"), + Type.Literal("version"), + Type.Literal("busy"), + Type.Literal("session_locked"), + Type.Literal("not_found"), + Type.Literal("invalid_request"), +]); +export const ProtocolErrorSchema = StrictObject({ + code: ProtocolErrorCodeSchema, + message: Type.String(), + details: Type.Optional(JsonValueSchema), +}); +export type ProtocolErrorCode = Static; +export type ProtocolError = Static; + +const PromptPayloadProperties = { + sessionId: IdSchema, + text: Type.String(), +} as const; + +export const ListCommandSchema = StrictObject({ command: Type.Literal("list") }); +export const CreateCommandSchema = StrictObject({ + command: Type.Literal("create"), + cwd: Type.Optional(Type.String({ minLength: 1 })), + name: Type.Optional(Type.String()), + model: Type.Optional(ModelRefSchema), + thinkingLevel: Type.Optional(ThinkingLevelSchema), +}); +export const AttachCommandSchema = StrictObject({ command: Type.Literal("attach"), sessionId: IdSchema }); +export const DetachCommandSchema = StrictObject({ command: Type.Literal("detach"), sessionId: IdSchema }); +export const PromptCommandSchema = StrictObject({ command: Type.Literal("prompt"), ...PromptPayloadProperties }); +export const SteerCommandSchema = StrictObject({ command: Type.Literal("steer"), ...PromptPayloadProperties }); +export const AbortCommandSchema = StrictObject({ command: Type.Literal("abort"), sessionId: IdSchema }); +export const SetModelCommandSchema = StrictObject({ + command: Type.Literal("set_model"), + sessionId: IdSchema, + model: ModelRefSchema, +}); +export const SetThinkingCommandSchema = StrictObject({ + command: Type.Literal("set_thinking"), + sessionId: IdSchema, + thinkingLevel: ThinkingLevelSchema, +}); +export const CommandSchema = Type.Union([ + ListCommandSchema, + CreateCommandSchema, + AttachCommandSchema, + DetachCommandSchema, + PromptCommandSchema, + SteerCommandSchema, + AbortCommandSchema, + SetModelCommandSchema, + SetThinkingCommandSchema, +]); +export type Command = Static; +export type CommandName = Command["command"]; + +export const CreateResultSchema = StrictObject({ + command: Type.Literal("create"), + session: SessionSnapshotSchema, +}); +export const AttachResultSchema = StrictObject({ + command: Type.Literal("attach"), + session: SessionSnapshotSchema, +}); +export const PromptResultSchema = StrictObject({ + command: Type.Literal("prompt"), + session: SessionSnapshotSchema, +}); +export const SteerResultSchema = StrictObject({ + command: Type.Literal("steer"), + session: SessionSnapshotSchema, +}); +export const AbortResultSchema = StrictObject({ + command: Type.Literal("abort"), + session: SessionSnapshotSchema, +}); +export const SetModelResultSchema = StrictObject({ + command: Type.Literal("set_model"), + session: SessionSnapshotSchema, +}); +export const SetThinkingResultSchema = StrictObject({ + command: Type.Literal("set_thinking"), + session: SessionSnapshotSchema, +}); + +export const ListResultSchema = StrictObject({ + command: Type.Literal("list"), + sessions: Type.Array(SessionSummarySchema), +}); +export const DetachResultSchema = StrictObject({ + command: Type.Literal("detach"), + sessionId: IdSchema, +}); +export const CommandResultSchema = Type.Union([ + ListResultSchema, + CreateResultSchema, + AttachResultSchema, + DetachResultSchema, + PromptResultSchema, + SteerResultSchema, + AbortResultSchema, + SetModelResultSchema, + SetThinkingResultSchema, +]); +export type CommandResult = Static; + +export type ResultForCommand = TCommand["command"] extends "list" + ? Static + : TCommand["command"] extends "detach" + ? Static + : Extract; + +/** Must be the first frame sent by a client. Version is intentionally an integer, not a coercible string. */ +export const ClientHelloSchema = StrictObject({ + type: Type.Literal("hello"), + version: Type.Integer({ minimum: 0 }), + token: Type.String({ minLength: 1 }), +}); +export type ClientHello = Static; + +export const RequestEnvelopeSchema = StrictObject({ + type: Type.Literal("request"), + id: IdSchema, + request: CommandSchema, +}); +export type RequestEnvelope = Static; +export const ClientMessageSchema = Type.Union([ClientHelloSchema, RequestEnvelopeSchema]); +export type ClientMessage = Static; + +export const ServerEventSchema = Type.Union([ + StrictObject({ type: Type.Literal("server_snapshot"), snapshot: ServerSnapshotSchema }), + StrictObject({ type: Type.Literal("session_snapshot"), snapshot: SessionSnapshotSchema }), + StrictObject({ + type: Type.Literal("session_progress"), + sessionId: IdSchema, + progress: TranscriptProgressSchema, + }), + StrictObject({ type: Type.Literal("session_removed"), sessionId: IdSchema }), +]); +export type ServerEvent = Static; + +export const ServerHelloSchema = StrictObject({ + type: Type.Literal("hello"), + version: Type.Literal(PROTOCOL_VERSION), + connectionId: IdSchema, + snapshot: ServerSnapshotSchema, +}); +export const ServerHelloErrorSchema = StrictObject({ + type: Type.Literal("hello_error"), + error: ProtocolErrorSchema, +}); +export const ResponseEnvelopeSchema = Type.Union([ + StrictObject({ + type: Type.Literal("response"), + id: IdSchema, + ok: Type.Literal(true), + result: CommandResultSchema, + }), + StrictObject({ + type: Type.Literal("response"), + id: IdSchema, + ok: Type.Literal(false), + error: ProtocolErrorSchema, + }), +]); +export const EventEnvelopeSchema = StrictObject({ + type: Type.Literal("event"), + event: ServerEventSchema, +}); +export const ServerMessageSchema = Type.Union([ + ServerHelloSchema, + ServerHelloErrorSchema, + ResponseEnvelopeSchema, + EventEnvelopeSchema, +]); +export type ServerHello = Static; +export type ServerHelloError = Static; +export type ResponseEnvelope = Static; +export type EventEnvelope = Static; +export type ServerMessage = Static; diff --git a/packages/protocol/test/cbor/cbor.test.ts b/packages/protocol/test/cbor/cbor.test.ts new file mode 100644 index 000000000..20da27b2f --- /dev/null +++ b/packages/protocol/test/cbor/cbor.test.ts @@ -0,0 +1,175 @@ +import { describe, expect, test } from "vitest"; +import { + CborError, + DEFAULT_MAX_CBOR_BYTE_LENGTH, + DEFAULT_MAX_CBOR_CONTAINER_LENGTH, + DEFAULT_MAX_CBOR_DEPTH, + decodeCbor, + encodeCbor, +} from "../../src/index.ts"; + +function fromHex(hex: string): Uint8Array { + if (hex.length % 2 !== 0) throw new Error("Hex fixture must contain whole bytes"); + const bytes = new Uint8Array(hex.length / 2); + for (let index = 0; index < bytes.length; index++) + bytes[index] = Number.parseInt(hex.slice(index * 2, index * 2 + 2), 16); + return bytes; +} + +function toHex(bytes: Uint8Array): string { + return Array.from(bytes, (byte) => byte.toString(16).padStart(2, "0")).join(""); +} + +const knownVectors: ReadonlyArray = [ + [null, "f6"], + [false, "f4"], + [true, "f5"], + [0, "00"], + [1, "01"], + [10, "0a"], + [23, "17"], + [24, "1818"], + [25, "1819"], + [100, "1864"], + [1000, "1903e8"], + [1_000_000, "1a000f4240"], + [1_000_000_000_000, "1b000000e8d4a51000"], + [Number.MAX_SAFE_INTEGER, "1b001fffffffffffff"], + [-1, "20"], + [-10, "29"], + [-24, "37"], + [-25, "3818"], + [-100, "3863"], + [-1000, "3903e7"], + [-1_000_000, "3a000f423f"], + [Number.MIN_SAFE_INTEGER, "3b001ffffffffffffe"], + [1.1, "fb3ff199999999999a"], + [-0, "fb8000000000000000"], + [new Uint8Array([1, 2, 3, 4]), "4401020304"], + ["", "60"], + ["IETF", "6449455446"], + ["ü", "62c3bc"], + ["水", "63e6b0b4"], + ["𐅑", "64f0908591"], + [[], "80"], + [[1, 2, 3], "83010203"], + [[1, [2, 3], [4, 5]], "8301820203820405"], + [{ a: 1, b: [2, 3] }, "a26161016162820203"], +]; + +describe("CBOR codec", () => { + test.each(knownVectors)("encodes and decodes RFC 8949 vector %#", (value, wire) => { + const encoded = encodeCbor(value); + expect(toHex(encoded)).toBe(wire); + const decoded = decodeCbor(fromHex(wire)); + if (typeof value === "number" && Object.is(value, -0)) expect(Object.is(decoded, -0)).toBe(true); + else expect(decoded).toEqual(value); + }); + + test("omits undefined object properties without omitting falsey values", () => { + const value = { omitted: undefined, zero: 0, empty: "", no: false, nil: null }; + expect(decodeCbor(encodeCbor(value))).toEqual({ zero: 0, empty: "", no: false, nil: null }); + }); + + test("preserves a leading Unicode BOM and treats __proto__ as data", () => { + expect(decodeCbor(fromHex("63efbbbf"))).toBe("\ufeff"); + const value: Record = {}; + Object.defineProperty(value, "__proto__", { enumerable: true, value: "safe" }); + const decoded = decodeCbor(encodeCbor(value)); + expect(decoded).toEqual(value); + expect(Object.getPrototypeOf(decoded)).toBe(Object.prototype); + expect(Object.hasOwn(decoded as object, "__proto__")).toBe(true); + }); + + test.each([ + ["top-level undefined", undefined], + ["undefined array element", [undefined]], + ["array hole", new Array(1)], + ["NaN", Number.NaN], + ["positive infinity", Number.POSITIVE_INFINITY], + ["negative infinity", Number.NEGATIVE_INFINITY], + ["unsafe positive integer", Number.MAX_SAFE_INTEGER + 1], + ["unsafe negative integer", Number.MIN_SAFE_INTEGER - 1], + ["bigint", 1n], + ["symbol", Symbol("unsupported")], + ["function", () => undefined], + ["Date", new Date(0)], + ["Map", new Map()], + ] as const)("rejects unsupported encoder value: %s", (_label, value) => { + expect(() => encodeCbor(value)).toThrow(CborError); + }); + + test("rejects maps with enumerable symbol keys", () => { + const value = { valid: true }; + Object.defineProperty(value, Symbol("key"), { enumerable: true, value: false }); + expect(() => encodeCbor(value)).toThrow(CborError); + }); + + test("rejects lossy strings, cycles, and excessive encoder depth", () => { + expect(() => encodeCbor("\ud800")).toThrow(/Unicode/i); + + const cyclic: unknown[] = []; + cyclic.push(cyclic); + expect(() => encodeCbor(cyclic)).toThrow(/cycles/i); + + let tooDeep: unknown = null; + for (let depth = 0; depth <= DEFAULT_MAX_CBOR_DEPTH; depth++) tooDeep = [tooDeep]; + expect(() => encodeCbor(tooDeep)).toThrow(/depth/i); + }); + + test.each([ + ["empty input", ""], + ["truncated integer", "18"], + ["reserved additional information", "1c"], + ["indefinite byte string", "5f"], + ["indefinite text string", "7f"], + ["indefinite array", "9f"], + ["indefinite map", "bf"], + ["tag", "c000"], + ["undefined", "f7"], + ["unsupported simple value", "e0"], + ["break outside an indefinite item", "ff"], + ["float16", "f93c00"], + ["float32", "fa3f800000"], + ["positive infinity", "fb7ff0000000000000"], + ["NaN", "fb7ff8000000000000"], + ["truncated float64", "fb3ff00000"], + ["truncated byte string", "44010203"], + ["truncated text string", "636162"], + ["truncated array", "8201"], + ["truncated map", "a16161"], + ["trailing data", "0000"], + ["non-string map key", "a10102"], + ["duplicate map key", "a2616101616102"], + ["invalid UTF-8 byte", "61ff"], + ["overlong UTF-8", "62c080"], + ["UTF-8 surrogate", "63eda080"], + ["unsafe positive integer", "1b0020000000000000"], + ["unsafe negative integer", "3b001fffffffffffff"], + ["unsafe integer encoded as float64", "fb4340000000000000"], + ] as const)("rejects invalid decoder input: %s", (_label, wire) => { + expect(() => decodeCbor(fromHex(wire))).toThrow(CborError); + }); + + test("enforces depth and declared length limits before traversing values", () => { + const tooDeep = new Uint8Array(DEFAULT_MAX_CBOR_DEPTH + 2); + tooDeep.fill(0x81, 0, -1); + tooDeep[tooDeep.length - 1] = 0xf6; + expect(() => decodeCbor(tooDeep)).toThrow(/depth/i); + + const oversizedBytes = fromHex(`5a${(DEFAULT_MAX_CBOR_BYTE_LENGTH + 1).toString(16).padStart(8, "0")}`); + const oversizedText = fromHex(`7a${(DEFAULT_MAX_CBOR_BYTE_LENGTH + 1).toString(16).padStart(8, "0")}`); + const oversizedArray = fromHex(`9a${(DEFAULT_MAX_CBOR_CONTAINER_LENGTH + 1).toString(16).padStart(8, "0")}`); + const oversizedMap = fromHex(`ba${(DEFAULT_MAX_CBOR_CONTAINER_LENGTH + 1).toString(16).padStart(8, "0")}`); + for (const wire of [oversizedBytes, oversizedText, oversizedArray, oversizedMap]) { + expect(() => decodeCbor(wire)).toThrow(/limit/i); + } + }); + + test("supports stricter caller-provided limits", () => { + expect(() => decodeCbor(fromHex("83010203"), { maxContainerLength: 2 })).toThrow(/limit/i); + expect(() => decodeCbor(fromHex("626162"), { maxByteLength: 2 })).toThrow(/limit/i); + expect(() => encodeCbor([1, 2, 3], { maxContainerLength: 2 })).toThrow(/limit/i); + expect(() => encodeCbor("ab", { maxByteLength: 2 })).toThrow(/limit/i); + }); +}); diff --git a/packages/protocol/test/framing.test.ts b/packages/protocol/test/framing.test.ts new file mode 100644 index 000000000..1cb893658 --- /dev/null +++ b/packages/protocol/test/framing.test.ts @@ -0,0 +1,117 @@ +import { describe, expect, test } from "vitest"; +import { assertCompleteFrame, DEFAULT_MAX_FRAME_LENGTH, encodeFrame, FrameDecoder, FrameError } from "../src/index.ts"; + +function concatenate(...chunks: Uint8Array[]): Uint8Array { + const result = new Uint8Array(chunks.reduce((length, chunk) => length + chunk.byteLength, 0)); + let offset = 0; + for (const chunk of chunks) { + result.set(chunk, offset); + offset += chunk.byteLength; + } + return result; +} + +describe("binary framing", () => { + test("prefixes payloads with a four-byte big-endian length", () => { + expect(encodeFrame(new Uint8Array([0xaa, 0xbb, 0xcc]))).toEqual( + new Uint8Array([0x00, 0x00, 0x00, 0x03, 0xaa, 0xbb, 0xcc]), + ); + expect(encodeFrame(new Uint8Array())).toEqual(new Uint8Array([0, 0, 0, 0])); + }); + + test("validates one complete bounded frame without accepting trailing or partial bytes", () => { + expect(() => assertCompleteFrame(new Uint8Array([0, 0, 0, 2, 1, 2]), { maxFrameLength: 2 })).not.toThrow(); + expect(() => assertCompleteFrame(new Uint8Array([0, 0, 0, 2, 1]))).toThrow(/complete/i); + expect(() => assertCompleteFrame(new Uint8Array([0, 0, 0, 1, 1, 2]))).toThrow(/exactly/i); + expect(() => assertCompleteFrame(new Uint8Array([0, 0, 0, 3, 1, 2, 3]), { maxFrameLength: 2 })).toThrow(/limit/i); + }); + + test("decodes fragmented, coalesced, and empty frames in order", () => { + const wire = concatenate( + encodeFrame(new Uint8Array([1, 2, 3])), + encodeFrame(new Uint8Array()), + encodeFrame(new Uint8Array([4])), + ); + const decoder = new FrameDecoder(); + const frames: Uint8Array[] = []; + for (const byte of wire) frames.push(...decoder.push(new Uint8Array([byte]))); + decoder.end(); + expect(frames).toEqual([new Uint8Array([1, 2, 3]), new Uint8Array(), new Uint8Array([4])]); + + const coalesced = new FrameDecoder(); + expect(coalesced.push(wire)).toEqual(frames); + coalesced.end(); + }); + + test("assembles payloads spanning multiple internal blocks", () => { + const payload = Uint8Array.from({ length: 70_000 }, (_, index) => index % 251); + const wire = encodeFrame(payload); + const decoder = new FrameDecoder(); + const frames = [ + ...decoder.push(wire.subarray(0, 101)), + ...decoder.push(wire.subarray(101, 65_541)), + ...decoder.push(wire.subarray(65_541)), + ]; + decoder.end(); + expect(frames).toEqual([payload]); + }); + + test("handles every split point across a frame", () => { + const wire = encodeFrame(new Uint8Array([10, 20, 30, 40])); + for (let split = 0; split <= wire.byteLength; split++) { + const decoder = new FrameDecoder(); + const frames = [...decoder.push(wire.subarray(0, split)), ...decoder.push(wire.subarray(split))]; + decoder.end(); + expect(frames).toEqual([new Uint8Array([10, 20, 30, 40])]); + } + }); + + test("copies payload bytes instead of retaining or aliasing input chunks", () => { + const chunk = encodeFrame(new Uint8Array([1, 2, 3])); + const decoder = new FrameDecoder(); + const frames = decoder.push(chunk); + chunk.fill(9); + expect(frames).toEqual([new Uint8Array([1, 2, 3])]); + }); + + test("accepts empty chunks and a clean empty stream", () => { + const decoder = new FrameDecoder(); + expect(decoder.push(new Uint8Array())).toEqual([]); + expect(() => decoder.end()).not.toThrow(); + }); + + test.each([ + ["partial header", new Uint8Array([0, 0, 0])], + ["partial payload", new Uint8Array([0, 0, 0, 2, 1])], + ] as const)("rejects a truncated stream at end: %s", (_label, wire) => { + const decoder = new FrameDecoder(); + expect(decoder.push(wire)).toEqual([]); + expect(() => decoder.end()).toThrow(FrameError); + }); + + test("rejects an oversized declared length as soon as its header is complete", () => { + const decoder = new FrameDecoder({ maxFrameLength: 3 }); + expect(() => decoder.push(new Uint8Array([0, 0, 0, 4]))).toThrow(/limit/i); + expect(() => decoder.push(new Uint8Array([1]))).toThrow(/failed/i); + }); + + test("accepts a frame exactly at the configured maximum", () => { + const decoder = new FrameDecoder({ maxFrameLength: 3 }); + expect(decoder.push(encodeFrame(new Uint8Array([1, 2, 3])))).toEqual([new Uint8Array([1, 2, 3])]); + decoder.end(); + }); + + test("cannot be pushed after end", () => { + const decoder = new FrameDecoder(); + decoder.end(); + expect(() => decoder.push(new Uint8Array())).toThrow(/ended/i); + expect(() => decoder.end()).toThrow(/ended/i); + }); + + test.each([-1, 1.5, Number.NaN, DEFAULT_MAX_FRAME_LENGTH * 1_000])( + "rejects invalid maximum frame length: %s", + (maxFrameLength) => { + expect(() => new FrameDecoder({ maxFrameLength })).toThrow(RangeError); + }, + ); +}); diff --git a/packages/protocol/test/protocol.test.ts b/packages/protocol/test/protocol.test.ts new file mode 100644 index 000000000..23621ab47 --- /dev/null +++ b/packages/protocol/test/protocol.test.ts @@ -0,0 +1,248 @@ +import { describe, expect, test } from "vitest"; +import { + type ClientHello, + type ClientMessage, + ClientMessageDecoder, + decodeCbor, + encodeCbor, + encodeClientMessage, + encodeFrame, + encodeServerMessage, + FrameDecoder, + isSupportedProtocolVersion, + PROTOCOL_VERSION, + ProtocolValidationError, + parseClientMessage, + parseServerMessage, + type ServerHello, + type ServerMessage, + ServerMessageDecoder, + type ServerSnapshot, +} from "../src/index.ts"; + +const emptyServerSnapshot: ServerSnapshot = { + serverId: "server-1", + protocolVersion: PROTOCOL_VERSION, + revision: 0, + sessions: [], + models: [], +}; + +const clientHello: ClientHello = { + type: "hello", + version: PROTOCOL_VERSION, + token: "secret", +}; + +const serverHello: ServerHello = { + type: "hello", + version: PROTOCOL_VERSION, + connectionId: "connection-1", + snapshot: emptyServerSnapshot, +}; + +describe("protocol validation", () => { + test("uses protocol version 2", () => { + expect(PROTOCOL_VERSION).toBe(2); + expect(isSupportedProtocolVersion(2)).toBe(true); + expect(isSupportedProtocolVersion(1)).toBe(false); + expect(isSupportedProtocolVersion(2.5)).toBe(false); + }); + + test.each([0, 1, PROTOCOL_VERSION])("accepts integer client hello version %s for negotiation", (version) => { + const message = { ...clientHello, version }; + expect(parseClientMessage(message)).toEqual(message); + }); + + test.each([ + { type: "hello", version: "2", token: "secret" }, + { type: "hello", version: 2.5, token: "secret" }, + { type: "hello", version: 2, token: "" }, + { type: "hello", version: 2, token: "secret", extra: true }, + ])("rejects an invalid strict handshake: $version", (message) => { + expect(() => parseClientMessage(message)).toThrow(ProtocolValidationError); + }); + + test("does not parse JSON strings as wire messages", () => { + expect(() => parseClientMessage(JSON.stringify(clientHello))).toThrow(ProtocolValidationError); + expect(() => parseServerMessage(JSON.stringify(serverHello))).toThrow(ProtocolValidationError); + }); + + test("rejects image input while the MVP remains text-only", () => { + expect(() => + parseClientMessage({ + type: "request", + id: "request-1", + request: { + command: "prompt", + sessionId: "session-1", + text: "inspect", + images: [{ type: "image", data: "abc", mimeType: "image/png" }], + }, + }), + ).toThrow(ProtocolValidationError); + }); + + test("parses a server handshake snapshot", () => { + expect(parseServerMessage(serverHello)).toEqual(serverHello); + }); + + test.each([ + { type: "hello", version: 1, connectionId: "connection-1", snapshot: emptyServerSnapshot }, + { type: "response", id: "request-1", ok: true, result: { command: "unknown" } }, + { type: "event", event: { type: "session_removed", sessionId: 42 } }, + ])("rejects invalid server messages", (wire) => { + expect(() => parseServerMessage(wire)).toThrow(ProtocolValidationError); + }); + + test("validates nested JSON tool details", () => { + const message = { + type: "event", + event: { + type: "session_progress", + sessionId: "session-1", + progress: { + type: "item_finished", + item: { + id: "tool-1", + role: "tool", + toolCallId: "call-1", + toolName: "read", + input: { path: "/tmp/file" }, + content: [{ type: "text", text: "done" }], + details: { lines: [1, 2, 3], cached: false }, + status: "complete", + isError: false, + timestamp: 1, + }, + }, + }, + }; + expect(parseServerMessage(message)).toEqual(message); + }); + + test("rejects cyclic protocol values with a protocol validation error", () => { + const details: Record = {}; + details.self = details; + const message = { + type: "response", + id: "request-1", + ok: false, + error: { code: "invalid_request", message: "invalid", details }, + }; + + expect(() => parseServerMessage(message)).toThrow(ProtocolValidationError); + }); + + test("validation errors do not retain rejected payloads", () => { + let thrown: unknown; + try { + parseClientMessage({ type: "hello", version: "2", token: "x".repeat(2_000_000) }); + } catch (error) { + thrown = error; + } + expect(thrown).toBeInstanceOf(ProtocolValidationError); + expect(Object.hasOwn(thrown as object, "value")).toBe(false); + expect((thrown as Error).message.length).toBeLessThan(1_000); + }); +}); + +describe("validated framed protocol APIs", () => { + test("encodes complete client and server frames", () => { + const clientFrames = new FrameDecoder().push(encodeClientMessage(clientHello)); + expect(clientFrames).toHaveLength(1); + expect(parseClientMessage(decodeCbor(clientFrames[0]!))).toEqual(clientHello); + + const serverFrames = new FrameDecoder().push(encodeServerMessage(serverHello)); + expect(serverFrames).toHaveLength(1); + expect(parseServerMessage(decodeCbor(serverFrames[0]!))).toEqual(serverHello); + }); + + test("enforces an outbound frame limit before returning encoded bytes", () => { + expect(() => encodeClientMessage(clientHello, { maxFrameLength: 8 })).toThrow(ProtocolValidationError); + expect(() => encodeServerMessage(serverHello, { maxFrameLength: 8 })).toThrow(ProtocolValidationError); + }); + + test("validates messages before encoding", () => { + expect(() => encodeClientMessage({ type: "hello", version: 2, token: "" })).toThrow(ProtocolValidationError); + }); + + test("omits explicit undefined optional properties on the wire", () => { + const message: ClientMessage = { + type: "request", + id: "request-1", + request: { command: "create", cwd: undefined, name: undefined }, + }; + const [payload] = new FrameDecoder().push(encodeClientMessage(message)); + expect(decodeCbor(payload!)).toEqual({ + type: "request", + id: "request-1", + request: { command: "create" }, + }); + }); + + test("incrementally decodes fragmented and coalesced client messages", () => { + const request: ClientMessage = { + type: "request", + id: "request-1", + request: { command: "list" }, + }; + const first = encodeClientMessage(clientHello); + const second = encodeClientMessage(request); + const wire = new Uint8Array(first.byteLength + second.byteLength); + wire.set(first); + wire.set(second, first.byteLength); + + for (let split = 0; split <= wire.byteLength; split++) { + const decoder = new ClientMessageDecoder(); + const messages = [...decoder.push(wire.subarray(0, split)), ...decoder.push(wire.subarray(split))]; + decoder.end(); + expect(messages).toEqual([clientHello, request]); + } + }); + + test("incrementally decodes server messages", () => { + const errorMessage: ServerMessage = { + type: "hello_error", + error: { code: "auth", message: "Invalid token" }, + }; + const decoder = new ServerMessageDecoder(); + expect(decoder.push(encodeServerMessage(errorMessage))).toEqual([errorMessage]); + decoder.end(); + }); + + test.each([ + ["empty CBOR payload", encodeFrame(new Uint8Array())], + ["malformed CBOR", encodeFrame(new Uint8Array([0xff]))], + ["schema-invalid CBOR", encodeFrame(encodeCbor({ type: "hello", version: 2, token: "", extra: true }))], + ] as const)("rejects invalid framed client input: %s", (_label, wire) => { + const decoder = new ClientMessageDecoder(); + expect(() => decoder.push(wire)).toThrow(ProtocolValidationError); + expect(() => decoder.push(encodeClientMessage(clientHello))).toThrow(/failed/i); + }); + + test("rejects CBOR byte strings nested in JSON-valued fields", () => { + const wire = encodeFrame( + encodeCbor({ + type: "response", + id: "request-1", + ok: false, + error: { + code: "invalid_request", + message: "invalid", + details: { nested: new Uint8Array([1, 2, 3]) }, + }, + }), + ); + expect(() => new ServerMessageDecoder().push(wire)).toThrow(ProtocolValidationError); + }); + + test("rejects truncated and oversized framing through the validated decoder", () => { + const truncated = new ServerMessageDecoder(); + expect(truncated.push(new Uint8Array([0, 0, 0, 2, 1]))).toEqual([]); + expect(() => truncated.end()).toThrow(ProtocolValidationError); + + const oversized = new ClientMessageDecoder({ maxFrameLength: 3 }); + expect(() => oversized.push(new Uint8Array([0, 0, 0, 4]))).toThrow(ProtocolValidationError); + }); +}); diff --git a/packages/protocol/tsconfig.build.json b/packages/protocol/tsconfig.build.json new file mode 100644 index 000000000..695dd9ad2 --- /dev/null +++ b/packages/protocol/tsconfig.build.json @@ -0,0 +1,9 @@ +{ + "extends": "../../tsconfig.base.json", + "compilerOptions": { + "outDir": "./dist", + "rootDir": "./src" + }, + "include": ["src/**/*.ts"], + "exclude": ["node_modules", "dist", "**/*.d.ts", "src/**/*.d.ts"] +} diff --git a/packages/protocol/tsconfig.test.json b/packages/protocol/tsconfig.test.json new file mode 100644 index 000000000..e1276ec0b --- /dev/null +++ b/packages/protocol/tsconfig.test.json @@ -0,0 +1,10 @@ +{ + "extends": "../../tsconfig.base.json", + "compilerOptions": { + "noEmit": true, + "module": "NodeNext", + "moduleResolution": "NodeNext", + "types": ["node", "vitest"] + }, + "include": ["src/**/*.ts", "test/**/*.ts"] +} diff --git a/packages/protocol/vitest.config.ts b/packages/protocol/vitest.config.ts new file mode 100644 index 000000000..157f467a0 --- /dev/null +++ b/packages/protocol/vitest.config.ts @@ -0,0 +1,9 @@ +import { defineConfig } from "vitest/config"; + +export default defineConfig({ + test: { + globals: true, + environment: "node", + reporters: process.env.GITHUB_ACTIONS ? ["dot", "github-actions"] : ["dot"], + }, +}); diff --git a/scripts/browser-smoke-entry.ts b/scripts/browser-smoke-entry.ts index 31dada999..8b201ed36 100644 --- a/scripts/browser-smoke-entry.ts +++ b/scripts/browser-smoke-entry.ts @@ -17,6 +17,7 @@ import { toError, truncateHead, } from "@earendil-works/pi-agent-core"; +import { decodeCbor, encodeCbor, PROTOCOL_VERSION } from "@earendil-works/pi-protocol"; // Keep this entry browser-safe. It is bundled by scripts/check-browser-smoke.mjs // to catch accidental Node-only runtime imports in browser-facing package exports. @@ -58,4 +59,6 @@ console.log( new FileError("not_found", "missing").code, toError("boom").message, typeof streamProxy, + PROTOCOL_VERSION, + decodeCbor(encodeCbor({ browser: true })), ); diff --git a/scripts/local-release.mjs b/scripts/local-release.mjs index 16ec5ab9e..3b275fc88 100644 --- a/scripts/local-release.mjs +++ b/scripts/local-release.mjs @@ -9,6 +9,7 @@ const packages = [ { directory: "packages/ai", name: "@earendil-works/pi-ai" }, { directory: "packages/tui", name: "@earendil-works/pi-tui" }, { directory: "packages/agent", name: "@earendil-works/pi-agent-core" }, + { directory: "packages/protocol", name: "@earendil-works/pi-protocol" }, { directory: "packages/storage/sqlite-node", name: "@earendil-works/pi-storage-sqlite-node" }, { directory: "packages/coding-agent", name: "@earendil-works/pi-coding-agent" }, ]; diff --git a/scripts/publish.mjs b/scripts/publish.mjs index e0b1410cf..967f61373 100644 --- a/scripts/publish.mjs +++ b/scripts/publish.mjs @@ -7,6 +7,7 @@ import { join } from "node:path"; const packages = [ { directory: "packages/ai", name: "@earendil-works/pi-ai" }, { directory: "packages/agent", name: "@earendil-works/pi-agent-core" }, + { directory: "packages/protocol", name: "@earendil-works/pi-protocol" }, { directory: "packages/storage/sqlite-node", name: "@earendil-works/pi-storage-sqlite-node" }, { directory: "packages/tui", name: "@earendil-works/pi-tui" }, { directory: "packages/coding-agent", name: "@earendil-works/pi-coding-agent" }, diff --git a/tsconfig.json b/tsconfig.json index fac44acf6..4d36db2f7 100644 --- a/tsconfig.json +++ b/tsconfig.json @@ -16,6 +16,8 @@ "@earendil-works/pi-coding-agent": ["./packages/coding-agent/src/index.ts"], "@earendil-works/pi-coding-agent/hooks": ["./packages/coding-agent/src/core/hooks/index.ts"], "@earendil-works/pi-coding-agent/*": ["./packages/coding-agent/src/*"], + "@earendil-works/pi-protocol": ["./packages/protocol/src/index.ts"], + "@earendil-works/pi-protocol/*": ["./packages/protocol/src/*"], "@earendil-works/pi-server": ["./packages/server/src/index.ts"], "@earendil-works/pi-server/*": ["./packages/server/src/*"], "typebox": ["./node_modules/typebox"],