diff --git a/devlog/_plan/260907_axis1_bugfixes/021_source_review.md b/devlog/_plan/260907_axis1_bugfixes/021_source_review.md new file mode 100644 index 0000000000..516a760813 --- /dev/null +++ b/devlog/_plan/260907_axis1_bugfixes/021_source_review.md @@ -0,0 +1,7 @@ +# wp1 source review + +Three bounded patches implemented with regression coverage. Hooke independently passed the physical-response quota observer wiring; Tesla independently passed quota/recovery security and source review with zero blockers. Version comparator and status/doctor projections inspected by main. All source workers report no local suite/typecheck/build execution. + +Quota source: #3809, Éverton Toffanetto; Co-authored-by included in f215f79b4. Version report: garysassano; Reported-by included in f91e3953a. Recovery report: Hu9956; Reported-by included in recovery commit. + +Source-only checks: git diff --check and documentation fence/whitespace inspection. These do not prove runtime correctness. wp2 final cumulative hosted CI is still mandatory. Final CI dispatch includes Windows because ordinary PR workflow omits it. No release/deploy workflow will be dispatched. diff --git a/docs-site/src/content/docs/reference/architecture.md b/docs-site/src/content/docs/reference/architecture.md index 8e1e361a81..e0fbc8bcba 100644 --- a/docs-site/src/content/docs/reference/architecture.md +++ b/docs-site/src/content/docs/reference/architecture.md @@ -233,7 +233,17 @@ response is not cacheable. Post-commit and 5xx errors keep the no-resend path. When encrypted agent-task recovery refuses a routed task, its existing 400 error can include a bounded `recovery_reason`: `unsupported_envelope`, -`admission_denied`, `recovery_unavailable`, `caller_cancelled`, or `input_changed`. -The field is omitted when no classified recovery result exists. +`admission_denied`, `recovery_unavailable`, `caller_cancelled`, `input_changed`, +`recovery_http_rejected`, `recovery_timeout`, `recovery_aborted`, +`recovery_transport_error`, or `recovery_invalid_output`. +HTTP rejection requires an observed non-success response. Invalid output includes +invalid UTF-8, oversized bodies, malformed or incomplete recovery streams, and +invalid or conflicting assignments. A caller's cancellation takes precedence over +an owned deadline, which takes precedence over decode/transport failures. +`recovery_aborted` describes a shared recovery cancelled independently of that caller. +Shared-flight waiters receive the same underlying failure unless individually cancelled; +only successful plaintext is cached. Diagnostics contain no upstream error or payload text. +The field is omitted when no classified recovery result exists, and existing combo +branches that return the original target failure keep that response. `recovery_unavailable` includes cache/singleflight capacity and does not prove an upstream request was attempted. No retry or broader envelope acceptance is enabled. diff --git a/scripts/test-layout/layout.json b/scripts/test-layout/layout.json index f560dffeb5..7547d467c7 100644 --- a/scripts/test-layout/layout.json +++ b/scripts/test-layout/layout.json @@ -239,6 +239,7 @@ "aside-profiles-routes.test.ts": "server", "aside-profiles.test.ts": "clients", "aside-profile-paths.test.ts": "clients", + "aside-profile-identity.test.ts": "clients", "aside-profile-sync-owner.test.ts": "clients", "assert-mergeable-review.test.ts": "ci-workflows", "auto-compact-budget.test.ts": "providers", diff --git a/src/clients/aside-profiles.ts b/src/clients/aside-profiles.ts index 31f13d9b76..770857611a 100644 --- a/src/clients/aside-profiles.ts +++ b/src/clients/aside-profiles.ts @@ -1,4 +1,4 @@ -import { lstatSync, readFileSync, readlinkSync, realpathSync, statSync, type Stats } from "node:fs"; +import { lstatSync, readFileSync, readlinkSync, realpathSync, statSync, type BigIntStats } from "node:fs"; import { homedir } from "node:os"; import { basename, dirname, isAbsolute, join, resolve } from "node:path"; import type { IntegrationIO } from "../integrations/config-io"; @@ -14,7 +14,7 @@ export interface AsideProfile { } const MAX_PROFILES = 128; -const MAX_MANIFEST_BYTES = 4 * 1024 * 1024; +const MAX_MANIFEST_BYTES = 4n * 1024n * 1024n; const MAX_LEAF_LINKS = 40; function refuse(message: string): never { @@ -30,9 +30,10 @@ function object(value: unknown): value is Record { return value !== null && typeof value === "object" && !Array.isArray(value); } -function inspect(path: string, follow = false): Stats | null { +function inspect(path: string, follow = false): BigIntStats | null { try { - return follow ? statSync(path) : lstatSync(path); + // File IDs can exceed Number's exact integer range; never round identities. + return follow ? statSync(path, { bigint: true }) : lstatSync(path, { bigint: true }); } catch (error) { if ((error as NodeJS.ErrnoException).code === "ENOENT") return null; return refuse("a filesystem boundary could not be inspected."); @@ -110,10 +111,10 @@ export function listAsideProfiles(env: NodeJS.ProcessEnv = process.env, home: st return readProfiles(root); } -type DirectoryIdentity = { path: string; dev: number; ino: number }; +type DirectoryIdentity = { path: string; dev: bigint; ino: bigint }; type Boundary = Array; -function sameIdentity(a: Pick, b: Pick): boolean { +function sameIdentity(a: Pick, b: Pick): boolean { return a.dev === b.dev && a.ino === b.ino; } @@ -162,7 +163,7 @@ function boundary(profile: AsideProfile, profiles: AsideProfile[], mutation: boo } if (absent) return identities; const leaf = inspect(profile.configPath); - if (leaf && (leaf.isSymbolicLink() || !leaf.isFile() || leaf.nlink > 1)) { + if (leaf && (leaf.isSymbolicLink() || !leaf.isFile() || leaf.nlink > 1n)) { refuse("the model catalog is a link, shared file or non-regular file."); } if (leaf && canonical(profile.configPath) !== join(parent!, "models.json")) { diff --git a/src/lib/bounded-body.ts b/src/lib/bounded-body.ts index 4016a0a753..0975268560 100644 --- a/src/lib/bounded-body.ts +++ b/src/lib/bounded-body.ts @@ -212,13 +212,28 @@ export async function readBoundedResponseBytes( } } -function decodeUtf8(chunks: readonly Uint8Array[], fatal: boolean): string { +// Mark only exceptions thrown by our decoder, preserving their identity and TypeError contract. +// Timeout-path flushing may fail too; retain that origin so callers do not lose the deadline. +const decodeFailures = new WeakMap(); + +export function boundedBodyDecodeFailure(error: unknown): "invalid_utf8" | "timeout" | undefined { + return error !== null && typeof error === "object" ? decodeFailures.get(error) : undefined; +} + +function decodeUtf8(chunks: readonly Uint8Array[], fatal: boolean, timedOut = false): string { const decoder = new TextDecoder("utf-8", { fatal }); - let text = ""; - for (const chunk of chunks) text += decoder.decode(chunk, { stream: true }); - // Flush an incomplete trailing UTF-8 sequence deterministically. - text += decoder.decode(); - return text; + try { + let text = ""; + for (const chunk of chunks) text += decoder.decode(chunk, { stream: true }); + // Flush an incomplete trailing UTF-8 sequence deterministically. + text += decoder.decode(); + return text; + } catch (error) { + if (error !== null && typeof error === "object") { + decodeFailures.set(error, timedOut ? "timeout" : "invalid_utf8"); + } + throw error; + } } /** @@ -297,7 +312,7 @@ export async function readBoundedResponseBody( "TimeoutError", ); return { - text: decodeUtf8([retained.subarray(0, retainedBytes)], options.fatalUtf8 === true), + text: decodeUtf8([retained.subarray(0, retainedBytes)], options.fatalUtf8 === true, true), truncated: true, timedOut: true, totalTimedOut: outcome === TOTAL_TIMEOUT, diff --git a/src/server/responses/agent-task-recovery-cache.ts b/src/server/responses/agent-task-recovery-cache.ts index 93d0c1778b..398a0feba4 100644 --- a/src/server/responses/agent-task-recovery-cache.ts +++ b/src/server/responses/agent-task-recovery-cache.ts @@ -2,6 +2,20 @@ const MAX_CACHE_BYTES = 8 * 1024 * 1024; const MAX_CONCURRENT_RECOVERIES = 32; const CACHE_TTL_MS = 15 * 60 * 1000; +export type AgentTaskRecoveryResolutionFailureReason = + | "recovery_unavailable" + | "caller_cancelled" + | "recovery_http_rejected" + | "recovery_timeout" + | "recovery_aborted" + | "recovery_transport_error" + | "recovery_invalid_output"; + +/** Shared flights carry bounded failures; only successful plaintext enters the cache. */ +export type AgentTaskRecoveryResolution = + | { readonly recovered: true; readonly assignment: string } + | { readonly recovered: false; readonly reason: AgentTaskRecoveryResolutionFailureReason }; + interface RecoveryCacheEntry { assignment: string; bytes: number; @@ -11,7 +25,7 @@ interface RecoveryCacheEntry { interface RecoveryFlight { controller: AbortController; - promise: Promise; + promise: Promise; waiters: number; settled: boolean; } @@ -63,7 +77,7 @@ function insertRecoveryCacheEntry(key: string, assignment: string, maxEntries: n function startRecoveryFlight( key: string, maxEntries: number, - request: (signal: AbortSignal) => Promise, + request: (signal: AbortSignal) => Promise, ): RecoveryFlight | null { const active = RECOVERY_FLIGHTS.get(key); if (active) return active; @@ -72,15 +86,15 @@ function startRecoveryFlight( const controller = new AbortController(); const flight: RecoveryFlight = { controller, - promise: Promise.resolve(null), + promise: Promise.resolve({ recovered: false, reason: "recovery_unavailable" }), waiters: 0, settled: false, }; flight.promise = request(controller.signal) - .then((assignment) => { - if (!assignment || controller.signal.aborted) return null; - insertRecoveryCacheEntry(key, assignment, maxEntries); - return assignment; + .then((result): AgentTaskRecoveryResolution => { + if (controller.signal.aborted) return { recovered: false, reason: "recovery_aborted" }; + if (result.recovered) insertRecoveryCacheEntry(key, result.assignment, maxEntries); + return result; }) .finally(() => { flight.settled = true; @@ -93,14 +107,14 @@ function startRecoveryFlight( async function waitForRecoveryFlight( flight: RecoveryFlight, abortSignal?: AbortSignal, -): Promise { - if (abortSignal?.aborted) return null; +): Promise { + if (abortSignal?.aborted) return { recovered: false, reason: "caller_cancelled" }; flight.waiters += 1; let onAbort: (() => void) | undefined; try { if (!abortSignal) return await flight.promise; - const cancelled = new Promise((resolve) => { - onAbort = () => resolve(null); + const cancelled = new Promise((resolve) => { + onAbort = () => resolve({ recovered: false, reason: "caller_cancelled" }); abortSignal.addEventListener("abort", onAbort, { once: true }); if (abortSignal.aborted) onAbort(); }); @@ -120,12 +134,27 @@ export async function resolveCachedAgentTaskRecovery( request: (signal: AbortSignal) => Promise, abortSignal?: AbortSignal, ): Promise { - if (abortSignal?.aborted) return null; + const result = await resolveCachedAgentTaskRecoveryWithResult(key, maxEntries, async signal => { + const assignment = await request(signal); + return assignment + ? { recovered: true, assignment } + : { recovered: false, reason: "recovery_unavailable" }; + }, abortSignal); + return result.recovered ? result.assignment : null; +} + +export async function resolveCachedAgentTaskRecoveryWithResult( + key: string, + maxEntries: number, + request: (signal: AbortSignal) => Promise, + abortSignal?: AbortSignal, +): Promise { + if (abortSignal?.aborted) return { recovered: false, reason: "caller_cancelled" }; sweepRecoveryCache(Date.now(), maxEntries); const cached = RECOVERY_CACHE.get(key)?.assignment; - if (cached) return cached; + if (cached) return { recovered: true, assignment: cached }; const flight = startRecoveryFlight(key, maxEntries, request); - return flight ? waitForRecoveryFlight(flight, abortSignal) : null; + return flight ? waitForRecoveryFlight(flight, abortSignal) : { recovered: false, reason: "recovery_unavailable" }; } export function discardCachedAgentTaskRecovery(key: string): void { diff --git a/src/server/responses/agent-task-recovery.ts b/src/server/responses/agent-task-recovery.ts index 22b7a4e66b..a15a2563ca 100644 --- a/src/server/responses/agent-task-recovery.ts +++ b/src/server/responses/agent-task-recovery.ts @@ -1,14 +1,16 @@ import { createHash, createHmac, randomBytes } from "node:crypto"; import { decodeJwtPayload, extractAccountId } from "../../oauth/chatgpt"; import type { OcxConfig } from "../../types"; -import { readBoundedResponseBody } from "../../lib/bounded-body"; +import { boundedBodyDecodeFailure, readBoundedResponseBody } from "../../lib/bounded-body"; import { isApiAuthRequired, isProxyAdmissionSecret } from "../auth-cors"; import { structurallyValidFernetTokens } from "./encrypted-payload"; import { cachedAgentTaskRecovery, discardCachedAgentTaskRecovery, resetAgentTaskRecoveryCache, - resolveCachedAgentTaskRecovery, + resolveCachedAgentTaskRecoveryWithResult, + type AgentTaskRecoveryResolution, + type AgentTaskRecoveryResolutionFailureReason, } from "./agent-task-recovery-cache"; /** Experimental opt-in normalization through ChatGPT's fixed Codex endpoint. */ @@ -44,9 +46,8 @@ export interface AgentTaskRecoveryOptions { export type AgentTaskRecoveryFailureReason = | "unsupported_envelope" | "admission_denied" - // Includes cache capacity rejection; does not imply an upstream request was attempted. - | "recovery_unavailable" - | "caller_cancelled" + // recovery_unavailable includes capacity rejection, which does not imply an upstream attempt. + | AgentTaskRecoveryResolutionFailureReason | "input_changed"; export type AgentTaskRecoveryResult = @@ -436,7 +437,7 @@ async function requestRecovery( envelope: AgentEnvelope, options: AgentTaskRecoveryOptions, abortSignal?: AbortSignal, -): Promise { +): Promise { const controller = new AbortController(); const timeout = setTimeout( () => controller.abort(new DOMException("Agent task recovery timed out", "TimeoutError")), @@ -454,8 +455,11 @@ async function requestRecovery( redirect: "error", }); if (!response.ok) { - try { await response.body?.cancel(); } catch { /* already closed */ } - return null; + // A rejected or never-settling cancellation must not extend the recovery deadline. + try { void response.body?.cancel().catch(() => undefined); } catch { /* already closed */ } + if (abortSignal?.aborted) return { recovered: false, reason: "recovery_aborted" }; + if (controller.signal.aborted) return { recovered: false, reason: "recovery_timeout" }; + return { recovered: false, reason: "recovery_http_rejected" }; } const body = await readBoundedResponseBody(response, { signal, @@ -465,10 +469,18 @@ async function requestRecovery( inactivityTimeoutMs: options.timeoutMs ?? 45_000, firstByteTimeoutMs: options.timeoutMs ?? 45_000, }); - if (body.truncated || body.oversized || body.timedOut || !body.displaySafe) return null; - return assignmentFromRecoverySse(body.text, envelope); - } catch { - return null; + if (abortSignal?.aborted) return { recovered: false, reason: "recovery_aborted" }; + if (controller.signal.aborted || body.timedOut) return { recovered: false, reason: "recovery_timeout" }; + if (body.truncated || body.oversized || !body.displaySafe) return { recovered: false, reason: "recovery_invalid_output" }; + const assignment = assignmentFromRecoverySse(body.text, envelope); + return assignment === null + ? { recovered: false, reason: "recovery_invalid_output" } + : { recovered: true, assignment }; + } catch (error) { + if (abortSignal?.aborted) return { recovered: false, reason: "recovery_aborted" }; + const decodeFailure = boundedBodyDecodeFailure(error); + if (controller.signal.aborted || decodeFailure === "timeout") return { recovered: false, reason: "recovery_timeout" }; + return { recovered: false, reason: decodeFailure === "invalid_utf8" ? "recovery_invalid_output" : "recovery_transport_error" }; } finally { clearTimeout(timeout); } @@ -497,23 +509,23 @@ export async function recoverEncryptedAgentTaskWithResult( const admitted = admittedRecovery(req, input, config, context.parentThreadId); if (!admitted.admitted) return { recovered: false, reason: admitted.reason }; const { admission, cacheKey, envelope } = admitted.recovery; - const assignment = await resolveCachedAgentTaskRecovery( + const result = await resolveCachedAgentTaskRecoveryWithResult( cacheKey, options.cacheEntries ?? 200, signal => requestRecovery(admission, envelope, options, signal), context.abortSignal, ); - if (!assignment) { + if (!result.recovered) { return { recovered: false, - reason: context.abortSignal?.aborted ? "caller_cancelled" : "recovery_unavailable", + reason: context.abortSignal?.aborted ? "caller_cancelled" : result.reason, }; } if (context.abortSignal?.aborted) { discardCachedAgentTaskRecovery(cacheKey); return { recovered: false, reason: "caller_cancelled" }; } - if (!injectAssignment(input, envelope, assignment)) { + if (!injectAssignment(input, envelope, result.assignment)) { discardCachedAgentTaskRecovery(cacheKey); return { recovered: false, reason: "input_changed" }; } diff --git a/structure/04_transports-and-sidecars.md b/structure/04_transports-and-sidecars.md index 7e41dc2222..d27022c2ad 100644 --- a/structure/04_transports-and-sidecars.md +++ b/structure/04_transports-and-sidecars.md @@ -1719,7 +1719,17 @@ response is not cacheable. Post-commit and 5xx errors keep the no-resend path. When encrypted agent-task recovery refuses a routed task, its existing 400 error can include a bounded `recovery_reason`: `unsupported_envelope`, -`admission_denied`, `recovery_unavailable`, `caller_cancelled`, or `input_changed`. -The field is omitted when no classified recovery result exists. +`admission_denied`, `recovery_unavailable`, `caller_cancelled`, `input_changed`, +`recovery_http_rejected`, `recovery_timeout`, `recovery_aborted`, +`recovery_transport_error`, or `recovery_invalid_output`. +HTTP rejection requires an observed non-success response. Invalid output includes +invalid UTF-8, oversized bodies, malformed or incomplete recovery streams, and +invalid or conflicting assignments. A caller's cancellation takes precedence over +an owned deadline, which takes precedence over decode/transport failures. +`recovery_aborted` describes a shared recovery cancelled independently of that caller. +Shared-flight waiters receive the same underlying failure unless individually cancelled; +only successful plaintext is cached. Diagnostics contain no upstream error or payload text. +The field is omitted when no classified recovery result exists, and existing combo +branches that return the original target failure keep that response. `recovery_unavailable` includes cache/singleflight capacity and does not prove an upstream request was attempted. No retry or broader envelope acceptance is enabled. diff --git a/tests/clients/aside-profile-identity.test.ts b/tests/clients/aside-profile-identity.test.ts new file mode 100644 index 0000000000..771966a1c4 --- /dev/null +++ b/tests/clients/aside-profile-identity.test.ts @@ -0,0 +1,132 @@ +import { expect, spyOn, test } from "bun:test"; +import * as fs from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import type { IntegrationIO } from "../../src/integrations/config-io"; +import { removeTreeWithRetry } from "../helpers/remove-tree"; + +// Capture real delegates before spying. Only fixture inode values are controlled; +// existence, file type, link count, realpath and link resolution remain native. +const nativeLstat = fs.lstatSync; +const nativeStat = fs.statSync; +const FIRST_INODE = 2n ** 53n; +const SECOND_INODE = FIRST_INODE + 1n; + +test("Aside preserves high file identities without admitting shared targets or directory replacement", async () => { + const home = fs.mkdtempSync(join(tmpdir(), "ocx-aside-identity-")); + const root = join(home, ".aside"); + const paths = [0, 1].map(id => join(root, "u", String(id), "models.json")); + const identities = new Map(); + const reads = new Set(); + let observingBoundary = false; + const restoreSpies: Array<() => void> = []; + + function controlledStat(delegate: typeof fs.statSync, kind: "stat" | "lstat"): typeof fs.statSync { + // Preserve fs's overload contract: the native delegate determines the result + // type, including undefined for throwIfNoEntry:false and number vs bigint. + return ((path: fs.PathLike, options?: fs.StatOptions) => { + const stats = delegate(path, options); + const inode = typeof path === "string" ? identities.get(path) : undefined; + if (stats && inode !== undefined) { + if (observingBoundary) reads.add(`${kind}:${path}`); + // Mutate this fresh native result, retaining its prototype and method + // receiver. Spreading Stats would lose native isFile/isDirectory methods. + stats.ino = options?.bigint ? inode : Number(inode); + } + return stats; + }) as typeof fs.statSync; + } + + function observe(run: () => T): T { + reads.clear(); + observingBoundary = true; + try { return run(); } finally { observingBoundary = false; } + } + + try { + for (const id of [0, 1]) fs.mkdirSync(join(root, "u", String(id)), { recursive: true }); + fs.writeFileSync(join(root, "accounts.json"), JSON.stringify({ + currentAccountId: 0, accounts: [{ id: 0 }, { id: 1 }], + })); + for (const path of paths) fs.writeFileSync(path, "{}"); + // Controlled IDs must not hide a runtime lacking native BigInt stat support. + expect(typeof nativeStat(paths[0]!, { bigint: true }).ino).toBe("bigint"); + expect(typeof nativeLstat(paths[0]!, { bigint: true }).ino).toBe("bigint"); + const lstatSpy = spyOn(fs, "lstatSync"); + restoreSpies.push(() => lstatSpy.mockRestore()); + lstatSpy.mockImplementation(controlledStat(nativeLstat, "lstat")); + const statSpy = spyOn(fs, "statSync"); + restoreSpies.push(() => statSpy.mockRestore()); + statSpy.mockImplementation(controlledStat(nativeStat, "stat")); + + // Load after spies so the regression also covers the native named-import seam. + const { assertAsideProfileBoundary, guardAsideProfileIO, listAsideProfiles } = + await import("../../src/clients/aside-profiles"); + const [selected, peer] = listAsideProfiles({}, home); + if (!selected || !peer) throw new Error("fixture requires two profiles"); + const profiles = [selected, peer]; + expect(Number(FIRST_INODE)).toBe(Number(SECOND_INODE)); + expect(FIRST_INODE).not.toBe(SECOND_INODE); + expect(nativeStat(selected.configPath, { bigint: true }).dev) + .toBe(nativeStat(peer.configPath, { bigint: true }).dev); + // Distinct catalogs and directories are allowed even though their Number + // representations collide. + // Reads are recorded only DURING boundary calls, so a missed spy binding + // cannot silently turn this into a passing ordinary-filesystem test. + for (const target of ["configPath", "detectDir"] as const) { + identities.clear(); + identities.set(selected[target], FIRST_INODE); + identities.set(peer[target], SECOND_INODE); + for (const profile of profiles) { + const sibling = profile === selected ? peer : selected; + observe(() => expect(() => assertAsideProfileBoundary(profile, profiles, true)).not.toThrow()); + expect(reads.has(`lstat:${profile[target]}`)).toBe(true); + expect(reads.has(`stat:${sibling[target]}`)).toBe(true); + } + } + + identities.clear(); + identities.set(selected.detectDir, FIRST_INODE); + let delegatedReads = 0; + const io: IntegrationIO = { + readText: () => { delegatedReads++; return { kind: "text", text: "{}" }; }, + statKind: () => "file", + writeText: () => {}, removeFile: () => {}, mkdirp: () => {}, + now: () => 0, appendJournal: () => {}, putRecord: () => {}, dropRecord: () => {}, + }; + const guarded = observe(() => guardAsideProfileIO(selected, io, profiles)); + expect(reads.has(`lstat:${selected.detectDir}`)).toBe(true); + observe(() => expect(guarded.readText(selected.configPath)).toEqual({ kind: "text", text: "{}" })); + expect(reads.has(`lstat:${selected.detectDir}`)).toBe(true); + expect(delegatedReads).toBe(1); + identities.set(selected.detectDir, SECOND_INODE); + observe(() => expect(() => guarded.readText(selected.configPath)) + .toThrow("the account directory changed after the operation began.")); + expect(reads.has(`lstat:${selected.detectDir}`)).toBe(true); + expect(delegatedReads).toBe(1); + + // No synthetic IDs for these controls: real hardlinks and symlinks must + // continue to be refused by the same boundary, with native stat delegates. + identities.clear(); + fs.unlinkSync(peer.configPath); + fs.linkSync(selected.configPath, peer.configPath); + expect(nativeLstat(selected.configPath, { bigint: true }).nlink).toBe(2n); + for (const profile of profiles) { + expect(() => assertAsideProfileBoundary(profile, profiles, true)) + .toThrow("the model catalog is a link, shared file or non-regular file."); + } + fs.unlinkSync(peer.configPath); + fs.symlinkSync(selected.configPath, peer.configPath, "file"); + expect(nativeLstat(peer.configPath, { bigint: true }).isSymbolicLink()).toBe(true); + expect(() => assertAsideProfileBoundary(selected, profiles, true)) + .toThrow("account catalogs share a target."); + expect(() => assertAsideProfileBoundary(peer, profiles, true)) + .toThrow("the model catalog is a link, shared file or non-regular file."); + } finally { + observingBoundary = false; + identities.clear(); + reads.clear(); + for (const restore of restoreSpies.reverse()) restore(); + removeTreeWithRetry(home); + } +}); diff --git a/tests/fixtures/test-layout-expected.json b/tests/fixtures/test-layout-expected.json index 2a1cac7ec3..cac40976e1 100644 --- a/tests/fixtures/test-layout-expected.json +++ b/tests/fixtures/test-layout-expected.json @@ -74,6 +74,7 @@ "aside-profiles-routes.test.ts": "server", "aside-profiles.test.ts": "clients", "aside-profile-paths.test.ts": "clients", + "aside-profile-identity.test.ts": "clients", "aside-profile-sync-owner.test.ts": "clients", "assert-mergeable-review.test.ts": "ci-workflows", "auto-compact-budget.test.ts": "providers", diff --git a/tests/server/agent-task-recovery-cache.test.ts b/tests/server/agent-task-recovery-cache.test.ts index 2ee994f8ce..35b5ba7050 100644 --- a/tests/server/agent-task-recovery-cache.test.ts +++ b/tests/server/agent-task-recovery-cache.test.ts @@ -29,16 +29,23 @@ describe("agent task recovery cache", () => { resetAgentTaskRecoveryCache(); }); - test("shared failure gives each waiter its own result without contaminating another key", async () => { + test.each([ + { kind: "http", reason: "recovery_http_rejected" }, + { kind: "reader", reason: "recovery_transport_error" }, + { kind: "decode", reason: "recovery_invalid_output" }, + ] as const)("shared $kind failure gives each waiter its own result without contaminating another key", async ({ kind, reason }) => { let release: (() => void) | undefined; const gate = new Promise(resolve => { release = resolve; }); let fetches = 0; globalThis.fetch = (async () => { const requestNumber = ++fetches; await gate; - return requestNumber === 1 - ? new Response("raw-failure-sentinel", { status: 503 }) - : new Response(recoverySse("Independent assignment.")); + if (requestNumber !== 1) return new Response(recoverySse("Independent assignment.")); + if (kind === "decode") return new Response(new Uint8Array([0xff])); + if (kind === "reader") return new Response(new ReadableStream({ + pull(controller) { controller.error(new TypeError("private-reader-failure")); }, + })); + return new Response("raw-failure-sentinel", { status: 503 }); }) as typeof fetch; const req = new Request("http://localhost/v1/responses", { headers: codexHeaders() }); const config = routedConfig(); @@ -53,8 +60,8 @@ describe("agent task recovery cache", () => { expect(fetches).toBe(2); release?.(); const [firstResult, secondResult, otherResult] = await Promise.all([first, second, other]); - expect(firstResult).toEqual({ recovered: false, reason: "recovery_unavailable" }); - expect(secondResult).toEqual({ recovered: false, reason: "recovery_unavailable" }); + expect(firstResult).toEqual({ recovered: false, reason }); + expect(secondResult).toEqual({ recovered: false, reason }); expect(firstResult).not.toBe(secondResult); expect(otherResult).toEqual({ recovered: true }); expect(firstInput).toEqual(encryptedInput()); @@ -68,6 +75,39 @@ describe("agent task recovery cache", () => { } }); + test("shared flight reset reports abort to surviving callers and never caches late plaintext", async () => { + let release!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + let fetches = 0; + globalThis.fetch = (async () => { + fetches++; + await gate; + return new Response(recoverySse("private-late-assignment")); + }) as typeof fetch; + const req = new Request("http://localhost/v1/responses", { headers: codexHeaders() }); + const firstInput = encryptedInput(); + const secondInput = encryptedInput(); + const first = recoverEncryptedAgentTaskWithResult(req, firstInput, {}, routedConfig()); + const second = recoverEncryptedAgentTaskWithResult(req, secondInput, {}, routedConfig()); + try { + expect(fetches).toBe(1); + resetAgentTaskRecoveryCache(); + release(); + const results = await Promise.all([first, second]); + expect(results).toEqual([ + { recovered: false, reason: "recovery_aborted" }, + { recovered: false, reason: "recovery_aborted" }, + ]); + expect(results[0]).not.toBe(results[1]); + expect(firstInput).toEqual(encryptedInput()); + expect(secondInput).toEqual(encryptedInput()); + expect(agentTaskRecoveryCacheSnapshotForTests()).toEqual({ entries: 0, bytes: 0 }); + } finally { + release(); + await Promise.all([first, second]); + } + }); + for (const succeeds of [true, false]) { test(`caller cancellation stays local when the remaining waiter ${succeeds ? "succeeds" : "fails"}`, async () => { let release: (() => void) | undefined; @@ -95,7 +135,7 @@ describe("agent task recovery cache", () => { release?.(); expect(await second).toEqual(succeeds ? { recovered: true } - : { recovered: false, reason: "recovery_unavailable" }); + : { recovered: false, reason: "recovery_http_rejected" }); expect(fetches).toBe(1); expect(restoreCachedEncryptedAgentTasks(req, encryptedInput(), config)).toBe(succeeds ? 1 : 0); } finally { diff --git a/tests/server/agent-task-recovery.test.ts b/tests/server/agent-task-recovery.test.ts index ceb1c5b6b5..a168f2c364 100644 --- a/tests/server/agent-task-recovery.test.ts +++ b/tests/server/agent-task-recovery.test.ts @@ -1,4 +1,4 @@ -import { afterEach, beforeEach, describe, expect, test } from "bun:test"; +import { afterEach, beforeEach, describe, expect, spyOn, test } from "bun:test"; import { createTranslatorBudget } from "../../src/lib/translator-budget"; import { warnAgentTaskRecoveryStartup } from "../../src/server"; import { @@ -7,6 +7,7 @@ import { recoverEncryptedAgentTaskWithResult, resetAgentTaskRecoveryState, restoreCachedEncryptedAgentTasks, + type AgentTaskRecoveryFailureReason, } from "../../src/server/responses/agent-task-recovery"; import { agentTaskRecoveryWaiterCountForTests } from "../../src/server/responses/agent-task-recovery-cache"; import { @@ -78,24 +79,36 @@ describe("agent task recovery (opt-in, default off)", () => { }); } - const failedRecoveries: Array<[string, () => Response]> = [ - ["HTTP 503", () => new Response("raw-error-sentinel", { status: 503 })], - ["network exception", () => { throw new Error("raw-error-sentinel"); }], - ["malformed SSE", () => new Response("data: {not-json}\n\n")], - ["missing completion", () => new Response(recoverySse("payload-sentinel").split("data: {\"type\":\"response.completed\"")[0])], - ["conflicting assignment", () => new Response(recoverySse("payload-sentinel") + recoveryCompletedSse("other-payload-sentinel"))], - ["failed terminal", () => new Response(recoverySse("payload-sentinel") + 'data: {"type":"response.failed","response":{"error":{"message":"raw-error-sentinel"}}}\n\n')], - ["incomplete terminal", () => new Response(recoverySse("payload-sentinel") + 'data: {"type":"response.incomplete"}\n\n')], - ["bare error", () => new Response(recoverySse("payload-sentinel") + 'data: {"type":"error","error":{"message":"raw-error-sentinel"}}\n\n')], + const failedRecoveries: Array<[string, () => Response, AgentTaskRecoveryFailureReason]> = [ + ["HTTP 401", () => new Response("private-error", { status: 401 }), "recovery_http_rejected"], + ["HTTP 403", () => new Response("private-error", { status: 403 }), "recovery_http_rejected"], + ["HTTP 429", () => new Response("private-error", { status: 429 }), "recovery_http_rejected"], + ["fetch TypeError", () => { throw new TypeError("private-error"); }, "recovery_transport_error"], + ["unowned TimeoutError", () => { throw new DOMException("private-error", "TimeoutError"); }, "recovery_transport_error"], + ["reader TypeError", () => new Response(new ReadableStream({ + pull(controller) { controller.error(new TypeError("private-reader-error")); }, + })), "recovery_transport_error"], + ["invalid UTF-8", () => new Response(new Uint8Array([0xff])), "recovery_invalid_output"], + ["trailing UTF-8", () => new Response(new Uint8Array([0xe2, 0x82])), "recovery_invalid_output"], + ["oversized body", () => new Response(new Uint8Array(4 * 1024 * 1024 + 1)), "recovery_invalid_output"], + ["invalid arguments", () => new Response(recoverySse("task").replace('{\\"assignment\\":\\"task\\"}', '{broken')), "recovery_invalid_output"], + ["HTTP 503", () => new Response("raw-error-sentinel", { status: 503 }), "recovery_http_rejected"], + ["network exception", () => { throw new Error("raw-error-sentinel"); }, "recovery_transport_error"], + ["malformed SSE", () => new Response("data: {not-json}\n\n"), "recovery_invalid_output"], + ["missing completion", () => new Response(recoverySse("payload-sentinel").split("data: {\"type\":\"response.completed\"")[0]), "recovery_invalid_output"], + ["conflicting assignment", () => new Response(recoverySse("payload-sentinel") + recoveryCompletedSse("other-payload-sentinel")), "recovery_invalid_output"], + ["failed terminal", () => new Response(recoverySse("payload-sentinel") + 'data: {"type":"response.failed","response":{"error":{"message":"raw-error-sentinel"}}}\n\n'), "recovery_invalid_output"], + ["incomplete terminal", () => new Response(recoverySse("payload-sentinel") + 'data: {"type":"response.incomplete"}\n\n'), "recovery_invalid_output"], + ["bare error", () => new Response(recoverySse("payload-sentinel") + 'data: {"type":"error","error":{"message":"raw-error-sentinel"}}\n\n'), "recovery_invalid_output"], // Exact-case events are also used by the pinned official Codex source. Recovery's // additional completed-status requirement remains deliberately stricter. - ["mixed-case completion", () => new Response(recoverySse("payload-sentinel").replace("response.completed", "Response.Completed"))], - ["mixed-case status", () => new Response(recoverySse("payload-sentinel").replace('"status":"completed"', '"status":"Completed"'))], - ["missing status", () => new Response(recoverySse("payload-sentinel").replace('"status":"completed",', ""))], - ["ciphertext assignment", () => new Response(recoverySse(FERNET_TASK))], + ["mixed-case completion", () => new Response(recoverySse("payload-sentinel").replace("response.completed", "Response.Completed")), "recovery_invalid_output"], + ["mixed-case status", () => new Response(recoverySse("payload-sentinel").replace('"status":"completed"', '"status":"Completed"')), "recovery_invalid_output"], + ["missing status", () => new Response(recoverySse("payload-sentinel").replace('"status":"completed",', "")), "recovery_invalid_output"], + ["ciphertext assignment", () => new Response(recoverySse(FERNET_TASK)), "recovery_invalid_output"], ]; - for (const [name, response] of failedRecoveries) { - test(`typed recovery keeps ${name} coarse and preserves false without retrying`, async () => { + for (const [name, response, reason] of failedRecoveries) { + test(`typed recovery classifies ${name} and preserves false without retrying`, async () => { const req = new Request("http://localhost/v1/responses", { headers: codexHeaders() }); const config = routedConfig(); let fetches = 0; @@ -103,7 +116,7 @@ describe("agent task recovery (opt-in, default off)", () => { const input = encryptedInput(); const original = structuredClone(input); expect(await recoverEncryptedAgentTaskWithResult(req, input, {}, config)) - .toEqual({ recovered: false, reason: "recovery_unavailable" }); + .toEqual({ recovered: false, reason }); expect(input).toEqual(original); expect(fetches).toBe(1); expect(restoreCachedEncryptedAgentTasks(req, encryptedInput(), config)).toBe(0); @@ -113,6 +126,69 @@ describe("agent task recovery (opt-in, default off)", () => { }); } + test.each(["pending", "rejecting"] as const)("HTTP refusal does not await %s body cancellation", async mode => { + let cancels = 0; + let reads = 0; + let releaseCancel: (() => void) | undefined; + const cancellation = new Promise(resolve => { releaseCancel = resolve; }); + globalThis.fetch = (async () => new Response(new ReadableStream({ + pull() { reads++; }, + cancel() { + cancels++; + return mode === "pending" ? cancellation : Promise.reject(new Error("private-cancel-error")); + }, + }, { highWaterMark: 0 }), { status: 503 })) as typeof fetch; + try { + const result = await recoverEncryptedAgentTaskWithResult( + new Request("http://localhost/v1/responses", { headers: codexHeaders() }), encryptedInput(), {}, routedConfig(), + ); + expect(result).toEqual({ recovered: false, reason: "recovery_http_rejected" }); + expect(cancels).toBe(1); + expect(reads).toBe(0); + } finally { + releaseCancel?.(); + } + }); + + test.each(["headers", "body", "caller"] as const)("owned deadline classification at %s preserves cancellation precedence", async site => { + const callbacks: Array<() => void> = []; + const timers = spyOn(globalThis, "setTimeout").mockImplementation(((callback: () => void) => { + callbacks.push(callback); + return 0 as unknown as ReturnType; + }) as typeof setTimeout); + const caller = new AbortController(); + let started!: () => void; + const ready = new Promise(resolve => { started = resolve; }); + let fetches = 0; + globalThis.fetch = ((_, init) => { + fetches++; + if (site === "body") return Promise.resolve(new Response(new ReadableStream({ + pull(controller) { + controller.enqueue(new Uint8Array([0xe2, 0x82])); + started(); + return new Promise(() => {}); + }, + }, { highWaterMark: 0 }))); + return new Promise((_resolve, reject) => { + init?.signal?.addEventListener("abort", () => reject(init.signal?.reason), { once: true }); + started(); + }); + }) as typeof fetch; + try { + const pending = recoverEncryptedAgentTaskWithResult( + new Request("http://localhost/v1/responses", { headers: codexHeaders() }), encryptedInput(), {}, routedConfig(), + { abortSignal: caller.signal }, + ); + await ready; + callbacks[0]!(); // Fire the owned deadline without wall-clock sleeps. + if (site === "caller") caller.abort(new TypeError("private-caller-error")); + expect(await pending).toEqual({ recovered: false, reason: site === "caller" ? "caller_cancelled" : "recovery_timeout" }); + expect(fetches).toBe(1); + } finally { + timers.mockRestore(); + } + }); + test("keeps the disabled fail-fast response byte-identical to the absent feature", async () => { const snapshot = async (config: ReturnType) => { let fetchCalls = 0; @@ -226,7 +302,7 @@ describe("agent task recovery (opt-in, default off)", () => { expect(response.status).toBe(400); expect(json.error?.code).toBe("unreadable_encrypted_agent_task"); - expect(json.error?.recovery_reason).toBe("recovery_unavailable"); + expect(json.error?.recovery_reason).toBe("recovery_invalid_output"); expect(fetchedUrls.length).toBeGreaterThan(0); expect(fetchedUrls[0]).toContain("chatgpt.com/backend-api/codex"); }); @@ -778,7 +854,7 @@ describe("agent task recovery (opt-in, default off)", () => { expect(fetchedUrls).toHaveLength(1); expect(fetchedUrls[0]).toContain("chatgpt.com/backend-api/codex/responses"); expect(await response.json()).toMatchObject({ - error: { code: "unreadable_encrypted_agent_task", recovery_reason: "recovery_unavailable" }, + error: { code: "unreadable_encrypted_agent_task", recovery_reason: "recovery_transport_error" }, }); }); }); diff --git a/tests/server/bounded-body.test.ts b/tests/server/bounded-body.test.ts index f5223d34a4..0bf5e0ae1b 100644 --- a/tests/server/bounded-body.test.ts +++ b/tests/server/bounded-body.test.ts @@ -1,7 +1,8 @@ -import { describe, expect, test } from "bun:test"; +import { describe, expect, spyOn, test } from "bun:test"; import { BOUNDED_BODY_MAX_BYTES, boundedBodyBufferGrowthsForTests, + boundedBodyDecodeFailure, readBoundedResponseBytes, readBoundedResponseBody, } from "../../src/lib/bounded-body"; @@ -21,6 +22,65 @@ function responseFromChunks(...chunks: Uint8Array[]): Response { } describe("readBoundedResponseBody", () => { + test("only actual decoder exceptions carry the decode discriminator", async () => { + for (const bytes of [new Uint8Array([0xff]), new Uint8Array([0xe2, 0x82])]) { + let caught: unknown; + try { await readBoundedResponseBody(responseFromChunks(bytes), { fatalUtf8: true }); } + catch (error) { caught = error; } + expect(caught).toBeInstanceOf(TypeError); + expect(boundedBodyDecodeFailure(caught)).toBe("invalid_utf8"); + } + const readerError = new TypeError("private-reader-error"); + const response = new Response(new ReadableStream({ pull(controller) { controller.error(readerError); } })); + let caught: unknown; + try { await readBoundedResponseBody(response, { fatalUtf8: true }); } + catch (error) { caught = error; } + expect(caught).toBe(readerError); + expect(boundedBodyDecodeFailure(caught)).toBeUndefined(); + }); + + test("fatal UTF-8 abort retains the exact caller reason without a decode mark", async () => { + const caller = new AbortController(); + const reason = new TypeError("private-caller-error"); + const pending = readBoundedResponseBody(new Response(new ReadableStream({})), { signal: caller.signal, fatalUtf8: true }); + caller.abort(reason); + let caught: unknown; + try { await pending; } catch (error) { caught = error; } + expect(caught).toBe(reason); + expect(boundedBodyDecodeFailure(caught)).toBeUndefined(); + }); + + test.each([0, 1])("fatal timeout flush retains deadline origin %s and cancels without waiting", async deadline => { + const callbacks: Array<() => void> = []; + const timers = spyOn(globalThis, "setTimeout").mockImplementation(((callback: () => void) => { + callbacks.push(callback); + return 0 as unknown as ReturnType; + }) as typeof setTimeout); + let stalled!: () => void; + const ready = new Promise(resolve => { stalled = resolve; }); + let pulls = 0; + let cancelled = false; + const response = new Response(new ReadableStream({ + pull(controller) { + if (pulls++ === 0) controller.enqueue(new Uint8Array([0xe2, 0x82])); + else { stalled(); return new Promise(() => {}); } + }, + cancel() { cancelled = true; return new Promise(() => {}); }, + }, { highWaterMark: 0 })); + try { + const pending = readBoundedResponseBody(response, { fatalUtf8: true }); + await ready; + callbacks[deadline === 0 ? 0 : callbacks.length - 1]!(); + let caught: unknown; + try { await pending; } catch (error) { caught = error; } + expect(caught).toBeInstanceOf(TypeError); + expect(boundedBodyDecodeFailure(caught)).toBe("timeout"); + expect(cancelled).toBe(true); + } finally { + timers.mockRestore(); + } + }); + test("the bounded JSON caller allows a full total deadline for its first byte", () => { expect(UPSTREAM_JSON_BODY_READ_OPTIONS.firstByteTimeoutMs) .toBe(UPSTREAM_JSON_BODY_READ_OPTIONS.totalTimeoutMs);