diff --git a/src/core/brain_cloud.ts b/src/core/brain_cloud.ts index 9ae337f..f80ea2f 100644 --- a/src/core/brain_cloud.ts +++ b/src/core/brain_cloud.ts @@ -35,6 +35,7 @@ import { decodeSse, type StreamFrame } from "./stream.js"; import { HttpError, StreamIncompleteError, StreamUnavailableError } from "./errors.js"; import { appendCustody } from "./custody.js"; import { hintFor } from "./error_hints.js"; +import { fallbackCapabilities, type ResolvedCapabilities } from "./capabilities.js"; /** Version of the dev-session wire protocol this client speaks. */ export const DEV_PROTOCOL_VERSION = 1; @@ -50,6 +51,75 @@ interface DevSessionCreated { tools?: string[]; } +/** + * The dev-session protocol versions this build can decode, read from the + * active capability contract (`dev_session_protocol_versions`) and falling back + * to the single version the client declares on the wire. + * + * The contract has always carried this list; nothing read it. Sourcing the + * accepted set from the resolved contract is what makes the handshake real: a + * client shipped with a newer contract accepts the versions that contract + * names, without another release of this file. + */ +export function devProtocolVersions(contract?: Record): number[] { + const raw = (contract ?? fallbackCapabilities().contract)["dev_session_protocol_versions"]; + const versions = Array.isArray(raw) + ? raw + .filter((v): v is number => typeof v === "number" && Number.isFinite(v)) + .map((v) => Math.trunc(v)) + : []; + return versions.length ? [...new Set(versions)] : [DEV_PROTOCOL_VERSION]; +} + +/** + * Validate the create response before a single byte of the stream is read. + * Returns null when the session is usable, or the message to fail the run with. + * + * Two distinct holes, both reachable from a 200: + * + * - No `session_id`. Nothing checked it, so `undefined` flowed straight into + * the path builders and the client issued requests against + * `/agent/dev/sessions/undefined/stream`. + * - A `protocol_version` this build does not speak. The client declares its + * version on create and the server echoes one back; the answer was never + * read. When the server ships a version with any changed frame shape, the + * tolerant `??` / `Number()` mapping in stream.ts turns fields this build no + * longer finds into zeros and empty strings rather than errors — a silently + * mis-decoded session instead of a refused one. + * + * This deliberately does NOT route into the legacy 404/403 downgrade: a + * version mismatch is not "this server has no dev route", and quietly dropping + * to the one-way chat stream would convert a loud, fixable incompatibility into + * an unexplained loss of the local tool round-trip. + */ +export function checkDevSession(created: unknown, speaks: readonly number[]): string | null { + const record = (created ?? {}) as Partial; + const spoken = speaks.length ? speaks : [DEV_PROTOCOL_VERSION]; + const id = record.session_id; + if (typeof id !== "string" || id.trim() === "") { + return "cloud dev session was created without a session_id — refusing to open a stream against an unnamed session"; + } + const version = record.protocol_version; + const list = spoken.map((v) => "v" + String(v)).join(", "); + if (typeof version !== "number" || !Number.isFinite(version)) { + return ( + "cloud dev session did not declare a protocol_version; this build speaks " + + list + + " and will not attach to an unversioned session — upgrade the agent (npm i -g aether-agents@latest)" + ); + } + if (!spoken.includes(Math.trunc(version))) { + return ( + "cloud dev session speaks protocol v" + + String(version) + + " but this build speaks " + + list + + " — upgrade the agent (npm i -g aether-agents@latest)" + ); + } + return null; +} + export class CloudBrain implements Brain { private aborted = false; private net: AbortController | null = null; @@ -57,8 +127,22 @@ export class CloudBrain implements Brain { private lastSeq = 0; /** Serializes upstream result POSTs so they arrive in execution order. */ private upstream: Promise = Promise.resolve(); + /** Dev-session protocol versions this build will attach to. */ + private readonly speaks: number[]; - constructor(private readonly api: ApiClient) {} + /** + * `capabilities` is optional so existing call sites are unchanged; without it + * the packaged contract snapshot supplies the accepted version set. A caller + * that has already resolved the server contract should pass it, so a server + * that legitimately advertises a newer dev protocol is honored rather than + * refused on stale packaged data. + */ + constructor( + private readonly api: ApiClient, + capabilities?: ResolvedCapabilities, + ) { + this.speaks = devProtocolVersions(capabilities?.contract); + } run(task: TaskCommand): AsyncIterable { const queue = new EventQueue(); @@ -87,6 +171,11 @@ export class CloudBrain implements Brain { } throw err; } + // Deliberately outside the catch above: a malformed or version-mismatched + // create is NOT a legacy server, and must not degrade to the one-way chat + // stream. Fail the run loudly instead. + const refusal = checkDevSession(created, this.speaks); + if (refusal) throw new Error(refusal); this.sessionId = created.session_id; queue.push({ type: "stage", name: "execute", face: "⟨◉⟩" }); // uplink face await this.devPump(queue); diff --git a/test/brain_cloud_dev.test.ts b/test/brain_cloud_dev.test.ts index 839cfca..1cf02ba 100644 --- a/test/brain_cloud_dev.test.ts +++ b/test/brain_cloud_dev.test.ts @@ -8,7 +8,7 @@ import { test } from "node:test"; import assert from "node:assert/strict"; -import { CloudBrain } from "../src/core/brain_cloud.js"; +import { CloudBrain, checkDevSession, devProtocolVersions } from "../src/core/brain_cloud.js"; import { ApiClient } from "../src/core/transport.js"; import type { BrainEvent } from "../src/core/brain_protocol.js"; import type { TokenStore } from "../src/core/auth.js"; @@ -23,10 +23,16 @@ interface Call { /** A dev-protocol server fake: create → JSON; stream attempts → scripted SSE * bodies (one per reconnect); tool-results/control/DELETE → recorded JSON. */ -function devServer(streams: string[][], opts?: { failToolResultTimes?: number }) { +function devServer( + streams: string[][], + opts?: { failToolResultTimes?: number; sessionId?: string | null; protocolVersion?: number | null }, +) { const calls: Call[] = []; let attempt = 0; let toolResultFailures = opts?.failToolResultTimes ?? 0; + // null means "omit the field entirely" — a 200 that answers neither question. + const sessionId = opts?.sessionId === undefined ? "devs_abc" : opts.sessionId; + const protocolVersion = opts?.protocolVersion === undefined ? 1 : opts.protocolVersion; const fetchImpl = (async (input: RequestInfo | URL, init?: RequestInit) => { const url = String(input); const method = init?.method ?? "GET"; @@ -44,12 +50,10 @@ function devServer(streams: string[][], opts?: { failToolResultTimes?: number }) }) as unknown as Response; if (url.endsWith("/agent/dev/sessions") && method === "POST") { - return json(200, { - session_id: "devs_abc", - protocol_version: 1, - model: "sonnet", - tools: ["read_file", "write_file"], - }); + const created: Record = { model: "sonnet", tools: ["read_file", "write_file"] }; + if (sessionId !== null) created["session_id"] = sessionId; + if (protocolVersion !== null) created["protocol_version"] = protocolVersion; + return json(200, created); } if (url.includes("/stream") && method === "GET") { const frames = streams[Math.min(attempt, streams.length - 1)] ?? []; @@ -345,3 +349,81 @@ test("a non-404 create failure surfaces as an error, not a silent legacy downgra assert.ok(!out.some((e) => e.type === "monologue")); }); }); + +test("devProtocolVersions reads the accepted set from the capability contract", () => { + // The contract has always carried this list; until now nothing consumed it. + assert.deepEqual(devProtocolVersions({ dev_session_protocol_versions: [1, 2] }), [1, 2]); + // Packaged contract when no resolved contract is supplied. + assert.deepEqual(devProtocolVersions(), [1]); + // A contract with a missing or unusable list still leaves the client with the + // version it actually declares on the wire, never an empty accept-set. + assert.deepEqual(devProtocolVersions({}), [1]); + assert.deepEqual(devProtocolVersions({ dev_session_protocol_versions: ["two"] }), [1]); +}); + +test("checkDevSession names both versions so the message is actionable", () => { + assert.equal(checkDevSession({ session_id: "s", protocol_version: 1 }, [1]), null); + const mismatch = checkDevSession({ session_id: "s", protocol_version: 2 }, [1]); + assert.ok(mismatch); + assert.match(mismatch, /v2/); + assert.match(mismatch, /v1/); + assert.match(mismatch, /upgrade/i); + assert.match(String(checkDevSession({ protocol_version: 1 }, [1])), /session_id/); + assert.match(String(checkDevSession({ session_id: " ", protocol_version: 1 }, [1])), /session_id/); +}); + +test("a create response with no session_id fails the run instead of streaming /undefined/stream", async () => { + const { fetchImpl, calls } = devServer([[]], { sessionId: null }); + await withFetch(fetchImpl, async () => { + const brain = new CloudBrain(new ApiClient("https://stub.test", tokens)); + const out: BrainEvent[] = []; + for await (const ev of brain.run(TASK)) out.push(ev); + assert.ok(out.some((e) => e.type === "error"), "must surface an error"); + assert.ok( + !calls.some((c) => c.url.includes("undefined")), + "must never build a request path out of an absent session id", + ); + }); +}); + +test("an unsupported dev protocol version fails the run and does NOT downgrade to legacy", async () => { + // A version mismatch is not "this server has no dev route". Folding it into + // the 404/403 downgrade would trade a loud, fixable incompatibility for a + // silent loss of the local tool round-trip. + const { fetchImpl, calls } = devServer([[]], { protocolVersion: 2 }); + await withFetch(fetchImpl, async () => { + const brain = new CloudBrain(new ApiClient("https://stub.test", tokens)); + const out: BrainEvent[] = []; + for await (const ev of brain.run(TASK)) out.push(ev); + const err = out.find((e) => e.type === "error"); + assert.ok(err && err.type === "error"); + assert.match(err.msg, /v2/); + assert.match(err.msg, /v1/); + assert.ok(!calls.some((c) => c.url.includes("/agent/chat/stream")), "must not fall back to the legacy stream"); + assert.ok(!calls.some((c) => c.url.includes("/stream") && c.method === "GET"), "must not attach to the dev stream"); + }); +}); + +test("a build whose contract advertises v2 attaches to a v2 session", async () => { + // The accepted set comes from the resolved contract, so a client shipped with + // a newer contract negotiates upward without another edit to brain_cloud.ts. + const { fetchImpl, calls } = devServer( + [[frame({ type: "done", seq: 1, ok: true, uvt: 1, cents: 0 })]], + { protocolVersion: 2 }, + ); + await withFetch(fetchImpl, async () => { + const brain = new CloudBrain(new ApiClient("https://stub.test", tokens), { + contract: { contract_version: 1, dev_session_protocol_versions: [1, 2] }, + digest: "test", + source: "fallback", + overlay: null, + warnings: [], + }); + const out: BrainEvent[] = []; + for await (const ev of brain.run(TASK)) out.push(ev); + assert.ok(!out.some((e) => e.type === "error"), JSON.stringify(out)); + assert.ok(calls.some((c) => c.url.includes("/stream") && c.method === "GET")); + const done = out.find((e) => e.type === "done"); + assert.ok(done && done.type === "done" && done.ok === true); + }); +});