diff --git a/.changeset/ordinary-server-egress.md b/.changeset/ordinary-server-egress.md new file mode 100644 index 000000000..8f7441a4e --- /dev/null +++ b/.changeset/ordinary-server-egress.md @@ -0,0 +1,8 @@ +--- +"@schlessera/brain-ui-server": minor +"@schlessera/brain-ui-sdk": patch +--- + +Add a typed per-turn egress connection event and an internal proxy recorder with durable connection/refusal facts and cancellation when recording fails. Ordinary backend activation remains unchanged. + +Honor valid zero-byte egress budgets by refusing before credential release or upstream contact. diff --git a/docs/http-api-coverage.md b/docs/http-api-coverage.md index 366622146..26b1b09dc 100644 --- a/docs/http-api-coverage.md +++ b/docs/http-api-coverage.md @@ -369,8 +369,8 @@ the additive `query_activity` scope carries the supported agent contract. - `governance events filter by session, turn and kind, and page in id order` ([source](../packages/ui-server/tests/governance-events-http.test.ts)), (`governance events filter by session, turn and kind, and page in id order`, `packages/ui-server/tests/governance-events-http.test.ts:48-71`) - `governance events refuse malformed parameters` ([source](../packages/ui-server/tests/governance-events-http.test.ts)), (`governance events refuse malformed parameters`, `packages/ui-server/tests/governance-events-http.test.ts:73-82`) - `a governance event storage failure is a 500` ([source](../packages/ui-server/tests/governance-events-http.test.ts)), (`a governance event storage failure is a 500`, `packages/ui-server/tests/governance-events-http.test.ts:84-91`) -- `events survive the Activity span-detail prune` ([source](../packages/ui-server/tests/governance-events.test.ts)), (`events survive the Activity span-detail prune`, `packages/ui-server/tests/governance-events.test.ts:177-191`) -- `deleting a session deletes its events and only its events` ([source](../packages/ui-server/tests/governance-events.test.ts)), (`deleting a session deletes its events and only its events`, `packages/ui-server/tests/governance-events.test.ts:193-202`) +- `events survive the Activity span-detail prune` ([source](../packages/ui-server/tests/governance-events.test.ts)), (`events survive the Activity span-detail prune`, `packages/ui-server/tests/governance-events.test.ts:178-192`) +- `deleting a session deletes its events and only its events` ([source](../packages/ui-server/tests/governance-events.test.ts)), (`deleting a session deletes its events and only its events`, `packages/ui-server/tests/governance-events.test.ts:194-203`) **N — Speech grants.** diff --git a/docs/http-api.md b/docs/http-api.md index 34f763061..6677180ed 100644 --- a/docs/http-api.md +++ b/docs/http-api.md @@ -1161,6 +1161,7 @@ read by any authenticated principal. | Kind | Version | Fields | Meaning | | --- | --- | --- | --- | +| `egress_connection` | 1 | `host`, `port`, `method`, `path` (optional), `bytes`, `accounting`, `durationMs`, `outcome`, `inference` (optional), `reason` (optional) | A completed, denied, failed or cancelled proxy connection. HTTP has a path and conservative wire accounting; opaque CONNECT has no path or plaintext claim. Only the concrete inference relay sets `inference: true`. Headers, query strings and content are absent. | | `steering_change` | 1 | `path`, `sequence`, `beforeHash`, `afterHash`, `taintSourceEventId`, `overlappingTurnIds` | One observed steering change, attributed to its turn. Hashes identify retained regular-file versions or are `null`; the source is an existing prior taint event or `null`. Shared windows remain shared, not exclusive writer claims. | | `policy_change` | 1 | `requestId`, `baseHash`, `proposedHash`, `sections`, `outcome` | Explicit user-approved policy publication. Actor is `user`; outcome is `updated`; hashes identify the exact reviewed and written bytes. | | `scratch_overage` | 1 | `bytes`, `pinnedFiles`, `pinnedBytes`, `overageBytes` | Scratch remains above its cap after a prune that preserved retained links. Counts are remaining bytes and pinned files/bytes; `overageBytes` is the excess. A background pass has actor `server` and no session or turn; unbound retention applies. | diff --git a/docs/integration-contract/http.md b/docs/integration-contract/http.md index cde3d658c..e49932e96 100644 --- a/docs/integration-contract/http.md +++ b/docs/integration-contract/http.md @@ -183,6 +183,12 @@ model, profile, payload }`. The [HTTP specification](../http-api.md#governance-e holds the parameters, the errors, the attribution values and every kind's payload fields. +The additive `egress_connection` kind records the concrete turn proxy's host, +port, visible HTTP method/path, bytes, accounting scope, duration and outcome. +CONNECT records opaque bytes without a path or content claim; the inference +tag comes only from the actual relay. Existing `egress_denial` remains named +and versioned as before. Header values, query strings and bodies are excluded. + The kinds and each kind's payload are versioned separately. A payload change that is not purely additive bumps that kind's `schemaVersion`, and stored events keep the version they were written with, so a new version or a new diff --git a/packages/ui-sdk/src/server/egress-proxy.ts b/packages/ui-sdk/src/server/egress-proxy.ts index ed1fda084..973df2ac8 100644 --- a/packages/ui-sdk/src/server/egress-proxy.ts +++ b/packages/ui-sdk/src/server/egress-proxy.ts @@ -50,7 +50,7 @@ function headers(message: IncomingMessage): Record { } export async function startEgressProxy(options: EgressProxyOptions) { - if (!Number.isSafeInteger(options.maxBytes) || options.maxBytes <= 0) throw new Error("Egress maxBytes must be a finite positive integer"); + if (!Number.isSafeInteger(options.maxBytes) || options.maxBytes < 0) throw new Error("Egress maxBytes must be a finite nonnegative integer"); for (const pattern of [...options.allow, ...options.deny, ...options.outwardHosts]) { if (!/^(?:\*\.)?[a-zA-Z0-9.-]+$/.test(pattern) || pattern.includes("..")) throw new Error("Invalid egress host pattern"); } diff --git a/packages/ui-sdk/tests/egress-proxy.test.ts b/packages/ui-sdk/tests/egress-proxy.test.ts index fd0038227..a1fd464d5 100644 --- a/packages/ui-sdk/tests/egress-proxy.test.ts +++ b/packages/ui-sdk/tests/egress-proxy.test.ts @@ -25,6 +25,13 @@ function tunnel(socketPath: string, target: string, body: string): Promise { + for (const maxBytes of [-1, 0.5, NaN, Infinity, Number.MAX_SAFE_INTEGER + 1]) { + await expect(startEgressProxy({ ...defaults([]), maxBytes }), `INVALID_EGRESS_BYTE_CEILING_${maxBytes}`) + .rejects.toThrow("Egress maxBytes must be a finite nonnegative integer"); + } +}); + test("real Unix HTTP proxy forwards cleartext and records only path and payload accounting", async () => { const records: EgressRecord[] = []; const origin = Bun.serve({ hostname: "127.0.0.1", port: 0, fetch: () => new Response("Ithaca") }); diff --git a/packages/ui-sdk/tests/inference-egress.test.ts b/packages/ui-sdk/tests/inference-egress.test.ts index 3ce79ae8f..86dd7060e 100644 --- a/packages/ui-sdk/tests/inference-egress.test.ts +++ b/packages/ui-sdk/tests/inference-egress.test.ts @@ -40,6 +40,20 @@ test("shared inference destination admission refuses before credentials and upst } finally { f.stop(); } }); +test("a zero-byte budget refuses inference before credential release and upstream contact", async () => { + let credentials = 0, contacts = 0; + const f = await setup(0, () => { contacts++; return new Response("Ithaca"); }, { + credentials: async () => { credentials++; return { authorization: "odysseus-host-held" }; }, + }); + try { + expect((await send(f.relay.socketPath, "/v1/messages")).status).toBe(403); + expect(credentials, "ZERO_CAP_PRECEDES_CREDENTIAL_RELEASE").toBe(0); + expect(contacts, "ZERO_CAP_PRECEDES_PROVIDER_CONTACT").toBe(0); + expect(f.proxy.usedBytes()).toBe(0); + expect(f.records).toMatchObject([{ inference: true, outcome: "denied", reason: "max_bytes_per_turn", bytes: 0 }]); + } finally { f.stop(); } +}); + test("inference and generic HTTP consume one turn counter and preserve redacted destination records", async () => { const f = await setup(5000, request => { if (new URL(request.url).pathname === "/v1/messages") expect(request.headers.get("authorization")).toBe("Bearer odysseus-host-held"); diff --git a/packages/ui-server/src/brain/turn-egress.ts b/packages/ui-server/src/brain/turn-egress.ts new file mode 100644 index 000000000..c6083e552 --- /dev/null +++ b/packages/ui-server/src/brain/turn-egress.ts @@ -0,0 +1,46 @@ +/** One admitted turn's concrete proxy, approval surface and durable records. */ +import type { EffectiveGovernancePolicy } from "@schlessera/brain-common/internal/governance-policy"; +import { startEgressProxy, type EgressRecord } from "@schlessera/brain-ui-sdk/internal"; +import { requestToolPermission, type BackendBridge } from "@schlessera/brain-ui-sdk/server"; +import type { GovernanceEventInput } from "../governance/events.js"; + +export async function prepareTurnEgress(options: { + policy: EffectiveGovernancePolicy; bridge: BackendBridge; controller: AbortController; + turnId: string; sessionId: () => string | null; model: () => string | null; + profile: "governed" | "strict"; origin: "interactive" | "voice"; + record(event: GovernanceEventInput): void; +}) { + let recordingFailure: Error | undefined; + const report = (record: EgressRecord): void => { + try { + const base = { turnId: options.turnId, sessionId: options.sessionId(), model: options.model(), + actor: options.origin, profile: options.profile }; + options.record({ ...base, kind: "egress_connection", payload: record }); + const denied = options.policy.egress.deny.some(pattern => { + const normalized = pattern.toLowerCase().replace(/\.$/, ""); + return normalized.startsWith("*.") ? record.host.endsWith(normalized.slice(1)) && record.host !== normalized.slice(2) : record.host === normalized; + }); + const reason = record.reason === "destination_not_allowed" ? denied ? "denied_host" : "not_allowlisted" + : record.reason === "max_bytes_per_turn" ? "byte_cap" : undefined; + if (reason) options.record({ ...base, kind: "egress_denial", payload: { + host: record.host, port: record.port, method: record.method, reason, bytes: record.bytes, + } }); + } catch { + recordingFailure ??= new Error("Turn egress recording is unavailable"); + options.controller.abort(recordingFailure); + } + }; + const proxy = await startEgressProxy({ mode: options.policy.egress.mode, allow: options.policy.egress.allow, + deny: options.policy.egress.deny, maxBytes: options.policy.egress.max_bytes_per_turn ?? Number.MAX_SAFE_INTEGER, + outwardHosts: [...new Set(["api.github.com", "uploads.github.com", ...options.policy.outward.hosts])], + signal: options.controller.signal, observe: report, + approveOutward: async (destination, signal) => { + const decision = await requestToolPermission(options.bridge, { toolUseId: crypto.randomUUID(), toolName: "Network", + kind: "command", input: destination, + description: `${destination.method} to ${destination.host}:${destination.port}`, + }, { noGrantSurface: options.origin === "voice" }); + return !signal.aborted && decision.behavior === "allow"; + }, + }); + return { ...proxy, recordingFailure: () => recordingFailure }; +} diff --git a/packages/ui-server/src/governance/events.ts b/packages/ui-server/src/governance/events.ts index a2005fe63..595e588c6 100644 --- a/packages/ui-server/src/governance/events.ts +++ b/packages/ui-server/src/governance/events.ts @@ -60,6 +60,14 @@ function kind(version: number, payload: T) { * docs/http-api.md ("Governance events"); keep the two in step. */ export const GOVERNANCE_EVENT_SCHEMAS = { + /** Completed or refused connection, using the proxy's actual visibility. */ + egress_connection: kind(1, z.strictObject({ + host: z.string().min(1).max(255), port: z.number().int().min(1).max(65535), + method: z.string().min(1).max(20), path: z.string().max(16384).optional(), bytes: count, + accounting: z.enum(["http_wire_bound", "opaque_bytes"]), durationMs: amount, + outcome: z.enum(["completed", "denied", "failed", "cancelled"]), + inference: z.literal(true).optional(), reason: z.string().max(200).optional(), + })), steering_change: kind(1, z.strictObject({ path, sequence: count, beforeHash: policyHash, afterHash: policyHash, taintSourceEventId: z.number().int().positive().nullable(), overlappingTurnIds: z.array(z.string().min(1).max(200)).max(1000) })), diff --git a/packages/ui-server/tests/governance-events.test.ts b/packages/ui-server/tests/governance-events.test.ts index 36612dfc9..3dce42fcb 100644 --- a/packages/ui-server/tests/governance-events.test.ts +++ b/packages/ui-server/tests/governance-events.test.ts @@ -27,6 +27,7 @@ const DAY = 24 * 60 * 60 * 1000; /** One valid v1 payload per kind, in the Odysseus world. */ const SAMPLE_PAYLOADS: { [K in GovernanceEventKind]: GovernanceEventPayload } = { + egress_connection: { host: "ithaca.invalid", port: 443, method: "CONNECT", bytes: 100, accounting: "opaque_bytes", durationMs: 5, outcome: "completed" }, policy_change: { requestId: "policy-ithaca", baseHash: null, proposedHash: "a".repeat(64), sections: ["profiles.interactive"], outcome: "updated" }, state_room_warning: { usedBytes: 51 * 1024 * 1024, allowanceBytes: 64 * 1024 * 1024, observedAt: Date.UTC(2026, 6, 12) }, scratch_overage: { bytes: 1_200_000_000, pinnedFiles: 3, pinnedBytes: 1_200_000_000, overageBytes: 126_258_176 }, diff --git a/packages/ui-server/tests/turn-egress.test.ts b/packages/ui-server/tests/turn-egress.test.ts new file mode 100644 index 000000000..a269687cd --- /dev/null +++ b/packages/ui-server/tests/turn-egress.test.ts @@ -0,0 +1,121 @@ +import { expect, test } from "bun:test"; +import { request } from "node:http"; +import { resolveGovernancePolicy, type GovernancePolicyInput } from "@schlessera/brain-common/internal/governance-policy"; +import type { BackendBridge, PermissionDecision } from "@schlessera/brain-ui-sdk/server"; +import { createUiDb } from "../src/db/client.js"; +import { queryGovernanceEvents, recordGovernanceEvent } from "../src/governance/events.js"; +import { prepareTurnEgress } from "../src/brain/turn-egress.js"; + +function get(socketPath: string, target: string): Promise { + return new Promise((resolve, reject) => { + const child = request({ socketPath, path: target }, response => { + response.resume(); response.once("end", () => resolve(response.statusCode!)); + response.once("error", reject); + }); + child.once("error", reject); child.end(); + }); +} +const bridge = (requestPermission: BackendBridge["requestPermission"]): BackendBridge => ({ requestPermission }) as BackendBridge; +const inputs = (controller: AbortController, policy: GovernancePolicyInput = { schema_version: 1 }) => ({ + controller, policy: resolveGovernancePolicy(policy), turnId: "turn-ithaca", sessionId: () => null, + model: () => "claude-haiku-4-5", profile: "governed" as const, origin: "interactive" as const, + bridge: bridge(async () => ({ behavior: "deny", message: "Odysseus declines" })), +}); + +test("a real turn proxy durably records bounded connection facts without the query", async () => { + const db = createUiDb(":memory:"); + const origin = Bun.serve({ hostname: "127.0.0.1", port: 0, fetch: () => new Response("Ithaca") }); + const proxy = await prepareTurnEgress({ ...inputs(new AbortController()), record: event => recordGovernanceEvent(db, event) }); + try { + expect(await get(proxy.socketPath, `${origin.url}voyage?private=odysseus-query`)).toBe(200); + proxy.stop(); + const rows = queryGovernanceEvents(db, { kinds: ["egress_connection"], limit: 10 }); + expect(rows, "TURN_EGRESS_DURABLE_CONNECTION").toHaveLength(1); + expect(rows[0]).toMatchObject({ turnId: "turn-ithaca", actor: "interactive", profile: "governed", + payload: { host: "127.0.0.1", port: origin.port, path: "/voyage", method: "GET", outcome: "completed" } }); + expect(JSON.stringify(rows)).not.toContain("odysseus-query"); + expect(proxy.recordingFailure()).toBeUndefined(); + } finally { proxy.stop(); origin.stop(true); db.close(); } +}); + +test("a zero-byte policy records refusal without contacting the origin", async () => { + const db = createUiDb(":memory:"); let calls = 0; + const origin = Bun.serve({ hostname: "127.0.0.1", port: 0, fetch() { calls++; return new Response("Ithaca"); } }); + let proxy: Awaited> | undefined; + try { + let admissionError: unknown; + try { proxy = await prepareTurnEgress({ ...inputs(new AbortController(), { schema_version: 1, egress: { max_bytes_per_turn: 0 } }), + record: event => recordGovernanceEvent(db, event) }); } catch (error) { admissionError = error; } + expect(admissionError, "TURN_EGRESS_ZERO_CAP_ADMISSION_VALID").toBeUndefined(); + expect(await get(proxy!.socketPath, origin.url.href)).toBe(403); + expect(calls, "TURN_EGRESS_ZERO_CAP_PREVENTS_CONTACT").toBe(0); + expect(queryGovernanceEvents(db, { kinds: ["egress_connection"], limit: 10 })).toMatchObject([ + { payload: { outcome: "denied", reason: "max_bytes_per_turn", bytes: 0 } }, + ]); + expect(queryGovernanceEvents(db, { kinds: ["egress_denial"], limit: 10 })).toMatchObject([ + { payload: { reason: "byte_cap", bytes: 0 } }, + ]); + expect(proxy!.usedBytes()).toBe(0); + } finally { proxy?.stop(); origin.stop(true); db.close(); } +}); + +const refusalCases: Array<[string, NonNullable]> = [ + ["denied_host", { mode: "allowlist", allow: ["127.0.0.1"], deny: ["127.0.0.1"] }], + ["not_allowlisted", { mode: "allowlist", allow: ["ithaca.invalid"] }], + ["byte_cap", { max_bytes_per_turn: 1 }], +]; +for (const [reason, egress] of refusalCases) test(`a real turn refusal durably retains ${reason} before contacting the origin`, async () => { + const db = createUiDb(":memory:"); let calls = 0; + const origin = Bun.serve({ hostname: "127.0.0.1", port: 0, fetch() { calls++; return new Response("Ithaca"); } }); + const proxy = await prepareTurnEgress({ ...inputs(new AbortController(), { schema_version: 1, egress }), + record: event => recordGovernanceEvent(db, event) }); + try { + expect(await get(proxy.socketPath, origin.url.href)).toBe(403); + expect(calls, "TURN_EGRESS_REFUSAL_BEFORE_CONTACT").toBe(0); + expect(queryGovernanceEvents(db, { kinds: ["egress_denial"], limit: 10 }), "TURN_EGRESS_DURABLE_DENIAL").toMatchObject([ + { payload: { host: "127.0.0.1", port: origin.port, method: "GET", reason, bytes: 0 } }, + ]); + expect(proxy.recordingFailure()).toBeUndefined(); + } finally { proxy.stop(); origin.stop(true); db.close(); } +}); + +for (const failingKind of ["egress_connection", "egress_denial"] as const) test(`a failed ${failingKind} recorder aborts the owning turn and retains its failure`, async () => { + const controller = new AbortController(); let attempts = 0; + const proxy = await prepareTurnEgress({ ...inputs(controller, { schema_version: 1, egress: { mode: "allowlist", allow: [] } }), + record(event) { attempts++; if (event.kind === failingKind) throw new Error("Fictional unavailable ledger"); } }); + try { + await get(proxy.socketPath, "http://127.0.0.1:1/voyage").catch(() => undefined); + expect(controller.signal.aborted, "TURN_EGRESS_FAILED_RECORD_ABORTS").toBe(true); + expect(proxy.recordingFailure()?.message).toBe("Turn egress recording is unavailable"); + expect(controller.signal.reason).toBe(proxy.recordingFailure()); + expect(attempts).toBe(failingKind === "egress_connection" ? 1 : 2); + } finally { proxy.stop(); } +}); + +test("a turn's outward connection waits for its bridge decision before contact", async () => { + let calls = 0, approve!: (value: PermissionDecision) => void, observed!: () => void; + let refuseMissingCard!: (error: Error) => void; + const permission = new Promise((resolve, reject) => { observed = resolve; refuseMissingCard = reject; }); + const origin = Bun.serve({ hostname: "127.0.0.1", port: 0, fetch() { calls++; return new Response("Ithaca"); } }); + const proxy = await prepareTurnEgress({ ...inputs(new AbortController(), { schema_version: 1, outward: { hosts: ["127.0.0.1"] } }), + bridge: bridge(() => { observed(); return new Promise(resolve => { approve = resolve; }); }), record() {} }); + const response = get(proxy.socketPath, origin.url.href); + const deadline = setTimeout(() => refuseMissingCard(new Error("TURN_EGRESS_OUTWARD_CARD_REQUIRED")), 5_000); + try { + await permission; + expect(calls, "TURN_EGRESS_OUTWARD_WAITS_FOR_CARD").toBe(0); + approve({ behavior: "allow" }); + expect(await response).toBe(200); expect(calls).toBe(1); + } finally { clearTimeout(deadline); approve?.({ behavior: "deny", message: "Fixture cleanup" }); proxy.stop(); await response.catch(() => undefined); origin.stop(true); } +}); + +test("voice without a grant surface refuses the outward connection", async () => { + let calls = 0, asks = 0; + const origin = Bun.serve({ hostname: "127.0.0.1", port: 0, fetch() { calls++; return new Response("Ithaca"); } }); + const proxy = await prepareTurnEgress({ ...inputs(new AbortController(), { schema_version: 1, outward: { hosts: ["127.0.0.1"] } }), + origin: "voice", bridge: bridge(async () => { asks++; return { behavior: "allow" }; }), record() {} }); + try { + expect(await get(proxy.socketPath, origin.url.href)).toBe(403); + expect(calls).toBe(0); expect(asks, "TURN_EGRESS_VOICE_NO_APPROVAL_SURFACE").toBe(0); + } finally { proxy.stop(); origin.stop(true); } +});