Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 8 additions & 0 deletions .changeset/ordinary-server-egress.md
Original file line number Diff line number Diff line change
@@ -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.
4 changes: 2 additions & 2 deletions docs/http-api-coverage.md
Original file line number Diff line number Diff line change
Expand Up @@ -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.**

Expand Down
1 change: 1 addition & 0 deletions docs/http-api.md
Original file line number Diff line number Diff line change
Expand Up @@ -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. |
Expand Down
6 changes: 6 additions & 0 deletions docs/integration-contract/http.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
2 changes: 1 addition & 1 deletion packages/ui-sdk/src/server/egress-proxy.ts
Original file line number Diff line number Diff line change
Expand Up @@ -50,7 +50,7 @@ function headers(message: IncomingMessage): Record<string, string | string[]> {
}

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");
}
Expand Down
7 changes: 7 additions & 0 deletions packages/ui-sdk/tests/egress-proxy.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,13 @@ function tunnel(socketPath: string, target: string, body: string): Promise<strin
});
}

test("invalid byte ceilings are rejected before creating a proxy", async () => {
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") });
Expand Down
14 changes: 14 additions & 0 deletions packages/ui-sdk/tests/inference-egress.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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");
Expand Down
46 changes: 46 additions & 0 deletions packages/ui-server/src/brain/turn-egress.ts
Original file line number Diff line number Diff line change
@@ -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 };
}
8 changes: 8 additions & 0 deletions packages/ui-server/src/governance/events.ts
Original file line number Diff line number Diff line change
Expand Up @@ -60,6 +60,14 @@ function kind<T extends z.ZodType>(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) })),
Expand Down
1 change: 1 addition & 0 deletions packages/ui-server/tests/governance-events.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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<K> } = {
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 },
Expand Down
121 changes: 121 additions & 0 deletions packages/ui-server/tests/turn-egress.test.ts
Original file line number Diff line number Diff line change
@@ -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<number> {
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<ReturnType<typeof prepareTurnEgress>> | 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<GovernancePolicyInput["egress"]>]> = [
["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<void>((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); }
});
Loading