diff --git a/packages/agent/CHANGELOG.md b/packages/agent/CHANGELOG.md index dd235f964b..fbbeed90e5 100644 --- a/packages/agent/CHANGELOG.md +++ b/packages/agent/CHANGELOG.md @@ -1,10 +1,18 @@ # Changelog ## [Unreleased] +- Managed fallback now transfers safety-stop authority only to the adjudicated final assistant shell; intermediate partial snapshots and hostile accessor-backed final messages cannot retain or bypass the provenance boundary (#4777 review). + +- Managed assistant reconstruction now copies provider metadata through guarded property reads instead of an unguarded spread, so accessor-trapped metadata degrades without aborting the attempt or creating managed retry authority (#4777 review). +- Hostile Proxy-wrapped final messages no longer reintroduce a forged `provider_safety_stop` label through the sanitizer fallback shell; discarded failure outcomes are now label-free before session policy can suppress provider fallback (#4777 review). + ### Fixed - Staged-payload sizing no longer materializes what it is bounding (#4602 fix-forward of the exact-head 078e22c0 review). All staging measurements now walk the JSON surface directly: exact byte counts come from a code-point walk (quotes, escapes, separators, delimiters, nulls, array holes, and keys all charged) instead of building the full `JSON.stringify` string plus its UTF-8 encoding, and lone surrogates are charged as the six-byte `\udXXX` escape JSON emits rather than their three-byte UTF-8 form, closing a ~2x undercount on surrogate-heavy strings. `structuredClone` is additionally preflighted by a clone-surface walk that never dispatches `toJSON`, accessors, or proxy traps: a live payload class whose compact `toJSON` hides an oversized own payload is rejected as the typed `local_buffer_overflow` at `overflow.preMeasure` — before the duplicate is allocated — instead of being cloned first and rejected at `overflow.staged`. Accessors are no longer invoked at all while sizing (a staged witness getter is read zero times), `undefined`-valued record properties are skipped exactly as `JSON.stringify` omits them, an unmeasurable assistant pair now fails closed like its `#stage` twin instead of being retained with a zero-byte charge, the `overflow.preMeasure` diagnostic reports the incoming event's real bounded size instead of a constant fabricated after `discard()`, and above-ceiling clamp warnings are logged once per distinct knob value with a bounded digest. | - Provider safety-stop messages now retain their explicitly allowlisted `errorKind: "provider_safety_stop"` through managed assistant snapshots and remain terminal even when transport facts are present on a multi-model fallback chain, while provider payloads still cannot forge the runtime-owned local diagnostic kinds (#4777). +- Terminal safety-stop authority is now provenance-bound instead of data-bound: a provider or custom stream payload that self-labels `errorKind: "provider_safety_stop"` without the adapter-minted mark is stripped at the stream exit before any retry/discard gate or the managed snapshot shell reads it, so a compromised provider can no longer force refusal by naming the field (#4777 review). Authenticated first-party envelopes (structured refusal signals parsed by the anthropic, openai-completions, and google adapters) keep terminal treatment, and only the agent loop's module-private rebuild set carries that authority onto its own destination — clones, JSON/persistence round-trips, and re-emitted payloads are all unauthenticated. +- Safety-stop minting is now limited to the package-private adapter capability, and public AI consumers cannot transfer authority from a genuine marked source to an arbitrary destination. The trailing stream-completion path also sanitizes provenance before rebuilding managed assistant messages, covering streams that end without a `done` or `error` event and keeping forged labels fallback-eligible (#4777 review). +- Safety-stop authority now expires at every stream dispatch entry: committed assistant messages (including a previously adjudicated stop) are handed to the next — possibly custom — stream through `convertToLlm`, and the dispatch-entry expiry guarantees no live authority mark is ever exposed to a stream, so re-use of a committed stop object cannot forge a terminal failure (#4777 review). The provenance strip is also stopReason-independent and rebuilds frozen or Proxy-trapped final messages as plain mutable copies, so a forged label on a nominally successful response cannot survive into the committed message (where it could skip session compaction checks) and cannot abort the run through a rejection trap. - A foreign error that self-declares a local failure kind no longer gets one either (#4618). `errorKind` and the structured `bufferOverflow` shape now come from a single identity-checked extractor (`managedLocalErrorDiagnostic`) used by both terminal-message producers — `managedFailureMessage` and the `Agent` run catch. Previously the shape was identity-gated but the label was not, so a provider or custom-stream failure carrying `errorKind: "local_buffer_overflow"` reached the parent receipt preview as `Local staging-buffer overflow; structured diagnostic unavailable.` and pointed whoever read it at the wrong subsystem. - Local diagnostic authority fields are no longer foreign-settable through the managed snapshot shell (#4618). `managedAssistantShell` spreads the provider/stream message snapshot into the rebuilt assistant message; a payload that smuggled a local `errorKind` or `bufferOverflow` through that spread could masquerade as the runtime's own identity-checked diagnostic at the parent boundary. Local kinds and `bufferOverflow` remain stripped from the snapshot spread, while the provider-owned safety-stop kind is copied only through its explicit closed-literal guard. diff --git a/packages/agent/src/agent-loop.ts b/packages/agent/src/agent-loop.ts index 4daddc8736..f75f948383 100644 --- a/packages/agent/src/agent-loop.ts +++ b/packages/agent/src/agent-loop.ts @@ -12,6 +12,7 @@ import { classifyFallbackTrigger, EMPTY_RESPONSE_PROVIDER_CODE, EventStream, + isProviderSafetyStopAuthenticated, isZodSchema, streamSimple, type ToolChoice, @@ -34,6 +35,7 @@ import { } from "@gajae-code/ai/utils"; import { isCursorExecResolved } from "@gajae-code/ai/utils/block-symbols"; import { $credentialEnv, logger, sanitizeText } from "@gajae-code/utils"; +import { revokeProviderSafetyStop } from "../../ai/src/adapter-internals/provider-safety-stop"; import type { AttemptScope } from "./attempt-scope"; import { createHarmonyAuditEvent, @@ -491,23 +493,53 @@ function managedContextOverflow(message: AssistantMessage, config: AgentLoopConf } /** Managed fallback owns retry policy; only attached typed transport facts may discard an attempt. */ -function managedProperty(value: unknown, key: string): unknown { - if (!value || typeof value !== "object") return undefined; +function managedPropertyRead(value: unknown, key: string): { ok: boolean; value: unknown } { + if (!value || typeof value !== "object") return { ok: true, value: undefined }; try { - return Reflect.get(value, key); + return { ok: true, value: Reflect.get(value, key) }; } catch { - return undefined; + return { ok: false, value: undefined }; } } +function managedProperty(value: unknown, key: string): unknown { + return managedPropertyRead(value, key).value; +} + function managedTransportFailure(failure: unknown) { const facts = managedProperty(failure, "transportFailure"); return facts && typeof facts === "object" ? transportFailureFacts(facts) : undefined; } +// AI owns provider-originated authority. The agent loop owns authority for +// the rebuilt message objects it creates; this second WeakSet is deliberately +// module-private so a public AI consumer cannot transfer authority to an +// arbitrary destination. A destination is marked only while this managed +// runtime is rebuilding a source that AI authenticated. +const managedProviderSafetyStops = new WeakSet(); + +function isManagedProviderSafetyStopAuthenticated(value: unknown): boolean { + return ( + isProviderSafetyStopAuthenticated(value) || + (typeof value === "object" && value !== null && managedProviderSafetyStops.has(value)) + ); +} + function managedRetryableFailure(failure: unknown): boolean { const facts = managedTransportFailure(failure); if (!facts) return false; + // A typed provider safety stop is terminal evidence ahead of any transport + // class, but only with adapter-minted provenance: unauthenticated labels + // are stripped at the stream exit (`sanitizeProviderSafetyStopProvenance`) + // and must fall through to ordinary transport classification so the chain + // can still advance (#4777). + if ( + managedProperty(failure, "stopReason") === "error" && + managedProperty(failure, "errorKind") === "provider_safety_stop" && + isManagedProviderSafetyStopAuthenticated(failure) + ) { + return false; + } const trigger = classifyFallbackTrigger(facts); // A plain `forbidden` is terminal: retrying it just re-sends a request the // caller is not authorized to make, and the credential-mutating consumers @@ -532,6 +564,62 @@ function promoteTypedEmptyResponseStop(message: AssistantMessage): void { message.stopReason = "error"; message.errorMessage = "Provider returned an empty response with zero token usage"; } +/** + * Terminal safety-stop authority is provenance-bound, not data-bound: a + * provider or custom stream payload that self-labels + * `errorKind: "provider_safety_stop"` without the adapter-minted mark must not + * terminalize the failure, because terminal treatment suppresses the user's + * configured fallback chain (#4777 review follow-up). Strip the unauthenticated + * field from the live final message at the single stream-exit point, before + * the managed snapshot shell clones it and before any retry/discard policy + * reads it, so a forged label degrades to an ordinary (fallback-eligible) + * error everywhere downstream — loop gates, session policy, and persistence. + * + * The label is stripped regardless of stopReason: the field is reserved for + * adapter-minted terminal stops, and downstream consumers (session compaction + * checks among them) read it without re-checking the error state, so a forged + * label on a nominally successful response must not survive either. A frozen + * or Proxy-trapped final message is rebuilt as a plain mutable copy instead of + * letting the strip abort the run. + */ + +function sanitizeProviderSafetyStopProvenance( + message: AssistantMessage, + model: AgentLoopConfig["model"], +): AssistantMessage { + const errorKindRead = managedPropertyRead(message, "errorKind"); + if ( + errorKindRead.ok && + (errorKindRead.value !== "provider_safety_stop" || isManagedProviderSafetyStopAuthenticated(message)) + ) { + return message; + } + const detached = managedAttemptSnapshotDetailed(message).snapshot; + if (isManagedPlainRecord(detached)) { + const rebuilt = { ...detached } as AssistantMessage; + delete rebuilt.errorKind; + return rebuilt; + } + const rebuilt = managedAssistantShell(message, model); + delete rebuilt.errorKind; + return rebuilt; +} + +/** + * Expire residual terminal safety-stop authority before a dispatch exposes + * committed history to a stream. Once a stop has been adjudicated, its + * committed assistant message may be handed unchanged to a later — possibly + * custom — stream through `convertToLlm`; a live mark would let that stream + * re-use the authenticated object (or a mutation of it) to forge a terminal + * failure and suppress the fallback chain (#4777 review follow-up). + */ +function expireProviderSafetyStopAuthority(messages: AgentMessage[]): void { + for (const message of messages) { + if (message.role !== "assistant") continue; + revokeProviderSafetyStop(message); + managedProviderSafetyStops.delete(message); + } +} /** * Neutralize leaked reserved control tokens in-place across the outgoing @@ -1693,6 +1781,7 @@ function managedAssistantShell( value: unknown, model: AgentLoopConfig["model"], degradedFieldDiagnostics: Set = new Set(), + transferSafetyStopAuthority = false, ): AssistantMessage { const detailed = managedAttemptSnapshotDetailed(value); const snapshotRecord = isManagedPlainRecord(detailed.snapshot) ? detailed.snapshot : undefined; @@ -1758,9 +1847,13 @@ function managedAssistantShell( stopReason === "error" && managedProperty(source, "errorKind") === "provider_safety_stop" ? ("provider_safety_stop" as const) : undefined; - const safeMetadata: Record = isManagedPlainRecord(detailed.snapshot) - ? { ...detailed.snapshot } - : {}; + const safeMetadata: Record = {}; + if (isManagedPlainRecord(detailed.snapshot)) { + for (const key of Object.keys(detailed.snapshot)) { + const metadata = managedProperty(detailed.snapshot, key); + if (metadata !== undefined) safeMetadata[key] = metadata; + } + } delete safeMetadata.errorMessage; delete safeMetadata.errorStatus; delete safeMetadata.transportFailure; @@ -1770,7 +1863,7 @@ function managedAssistantShell( // runtime failure in the executor's parent-facing summary (#4618). delete safeMetadata.errorKind; delete safeMetadata.bufferOverflow; - return { + const rebuilt: AssistantMessage = { ...safeMetadata, role: "assistant", content, @@ -1785,6 +1878,16 @@ function managedAssistantShell( ...(errorKind ? { errorKind } : {}), ...(typeof errorStatus === "number" && Number.isFinite(errorStatus) ? { errorStatus } : {}), }; + // The closed-literal copy above is fed by the stream-exit provenance + // sanitize, so an unauthenticated label never reaches here. Mark the + // runtime-owned destination only when this source is already authenticated; + // no public AI API can perform this transfer (#4777 review). + if (transferSafetyStopAuthority && errorKind && isManagedProviderSafetyStopAuthenticated(value)) { + managedProviderSafetyStops.add(rebuilt); + revokeProviderSafetyStop(value); + if (typeof value === "object" && value !== null) managedProviderSafetyStops.delete(value); + } + return rebuilt; } function managedContentBlock(block: unknown): AssistantMessage["content"] { @@ -3479,7 +3582,7 @@ async function runLoopBody( return; } if (attemptTransaction) { - message = managedAssistantShell(message, config.model); + message = managedAssistantShell(message, config.model, new Set(), true); const index = currentContext.messages.length - 1; if (index >= 0 && currentContext.messages[index]?.role === "assistant") { currentContext.messages[index] = message; @@ -3785,10 +3888,22 @@ async function streamAssistantResponse( const managedDegradedFieldDiagnostics = new Set(); // Apply context transform if configured (AgentMessage[] → AgentMessage[]) let messages = context.messages; + // Revoke before invoking any caller-controlled transform so it cannot retain + // a live authenticated object and restore its role for a later custom stream. + expireProviderSafetyStopAuthority(messages); + if (messages !== context.messages) expireProviderSafetyStopAuthority(context.messages); if (config.transformContext) { messages = await config.transformContext(messages, signal, scope); } + // Expire residual terminal safety-stop authority again after the transform: + // committed history (including a previously adjudicated stop) is handed + // to the stream through convertToLlm below, and a live mark would let a + // custom stream re-use the authenticated object to forge a terminal + // failure (#4777 review follow-up). + expireProviderSafetyStopAuthority(messages); + if (messages !== context.messages) expireProviderSafetyStopAuthority(context.messages); + // Convert to LLM-compatible messages (AgentMessage[] → Message[]) and normalize at the LLM boundary. // Cache hits are keyed by provider-visible content hashes, never message object identity. const normalizedMessages = await convertAndNormalizeMessages(messages, context, config); @@ -4176,9 +4291,10 @@ async function streamAssistantResponse( case "done": case "error": { + const finished = sanitizeProviderSafetyStopProvenance(await finishResponse(), config.model); const finalMessage = config.fallbackManaged - ? managedAssistantShell(await finishResponse(), config.model, managedDegradedFieldDiagnostics) - : await finishResponse(); + ? managedAssistantShell(finished, config.model, managedDegradedFieldDiagnostics, true) + : finished; promoteTypedEmptyResponseStop(finalMessage); if (addedPartial) { context.messages[context.messages.length - 1] = finalMessage; @@ -4199,9 +4315,10 @@ async function streamAssistantResponse( closeIterator(); } + const finished = sanitizeProviderSafetyStopProvenance(await finishResponse(), config.model); const trailing = config.fallbackManaged - ? managedAssistantShell(await finishResponse(), config.model, managedDegradedFieldDiagnostics) - : await finishResponse(); + ? managedAssistantShell(finished, config.model, managedDegradedFieldDiagnostics, true) + : finished; await finishChat(trailing); return trailing; }); diff --git a/packages/agent/test/managed-attempt-transaction.test.ts b/packages/agent/test/managed-attempt-transaction.test.ts index 353eb5da62..29698e7b45 100644 --- a/packages/agent/test/managed-attempt-transaction.test.ts +++ b/packages/agent/test/managed-attempt-transaction.test.ts @@ -12,10 +12,14 @@ import { } from "@gajae-code/agent-core/agent-loop"; import type { AgentContext, AgentEvent, AgentLoopConfig } from "@gajae-code/agent-core/types"; import type { AssistantMessage, AssistantMessageEvent, Message } from "@gajae-code/ai"; - import { createMockModel } from "@gajae-code/ai/providers/mock"; import { AssistantMessageEventStream } from "@gajae-code/ai/utils/event-stream"; import { logger } from "@gajae-code/utils"; +import { + mintProviderSafetyStop, + PROVIDER_SAFETY_STOP_ADAPTER_CAPABILITY, + PROVIDER_SAFETY_STOP_ADAPTER_INVOCATION, +} from "../../ai/src/adapter-internals/provider-safety-stop"; /** * Capture the bounded local-failure diagnostics emitted for one run. Returns @@ -133,18 +137,33 @@ describe("managed attempt transaction", () => { // These values are deliberately supplied by the provider envelope. The // runtime-authored local kinds tested below and in the local snapshot/ // overflow cases must not be confused with this untrusted input surface. + // The provider-owned safety-stop kind survives only when the envelope + // carries adapter-minted provenance; a wire-assignable field alone is + // stripped (#4777 review follow-up). const cases = [ { errorKind: "provider_safety_stop" as const, stopReason: "error" as const, + authenticated: true, expected: "provider_safety_stop" as const, }, - { errorKind: "provider_safety_stop" as const, stopReason: "stop" as const, expected: undefined }, + { + errorKind: "provider_safety_stop" as const, + stopReason: "error" as const, + authenticated: false, + expected: undefined, + }, + { + errorKind: "provider_safety_stop" as const, + stopReason: "stop" as const, + authenticated: true, + expected: undefined, + }, { errorKind: "local_buffer_overflow" as const, stopReason: "error" as const, expected: undefined }, { errorKind: undefined, stopReason: "error" as const, expected: undefined }, ]; - for (const { errorKind, stopReason, expected } of cases) { + for (const { errorKind, stopReason, authenticated, expected } of cases) { const mock = createMockModel(); const streamFn = () => { const stream = new AssistantMessageEventStream(); @@ -154,6 +173,15 @@ describe("managed attempt transaction", () => { errorMessage: "provider response", ...(errorKind ? { errorKind } : {}), }; + if (authenticated) { + mintProviderSafetyStop( + message, + "refusal", + PROVIDER_SAFETY_STOP_ADAPTER_CAPABILITY, + undefined, + PROVIDER_SAFETY_STOP_ADAPTER_INVOCATION, + ); + } queueMicrotask(() => { stream.push({ type: "start", partial: message }); stream.push({ type: "done", reason: "stop", message }); @@ -173,6 +201,352 @@ describe("managed attempt transaction", () => { } }); + it("keeps an authenticated transport-fact-carrying provider safety stop terminal instead of discarding it", async () => { + // A first-party adapter that reports its safety stop on an HTTP envelope + // (OpenAI content_filter with status + transportFailure) keeps terminal + // authority even when the facts classify as retryable (5xx): the + // attempt is committed for the parent, not discarded into a retryable + // managed failure outcome that would advance the chain (#4777). + const mock = createMockModel(); + let dispatches = 0; + const streamFn = () => { + dispatches += 1; + const stream = new AssistantMessageEventStream(); + const message: AssistantMessage = { + ...assistantMessage(mock.model), + stopReason: "error", + errorMessage: "The response was filtered by the content management policy", + errorStatus: 500, + transportFailure: { kind: "transport", status: 500 }, + }; + mintProviderSafetyStop( + message, + "content_filter", + PROVIDER_SAFETY_STOP_ADAPTER_CAPABILITY, + undefined, + PROVIDER_SAFETY_STOP_ADAPTER_INVOCATION, + ); + queueMicrotask(() => { + stream.push({ type: "start", partial: message }); + stream.push({ type: "error", reason: "error", error: message }); + }); + return stream; + }; + const agent = new Agent({ + initialState: { model: mock.model, systemPrompt: ["test"], tools: [], messages: [] }, + streamFn, + }); + + await agent.prompt("run", { fallbackManaged: true }); + + expect(dispatches).toBe(1); + const terminal = agent.state.messages.at(-1); + expect(terminal?.role).toBe("assistant"); + expect(terminal).toMatchObject({ + stopReason: "error", + errorKind: "provider_safety_stop", + }); + }); + + it("discards an unauthenticated transport-fact-carrying safety-stop label for retry", async () => { + // The forged counterpart: the same HTTP-envelope shape without + // adapter-minted provenance is ordinary retryable transport data. The + // loop discards the attempt for the managed failure outcome (chain + // advance stays available) and no typed stop is committed (#4777). + const mock = createMockModel(); + const streamFn = () => { + const stream = new AssistantMessageEventStream(); + const message: AssistantMessage = { + ...assistantMessage(mock.model), + stopReason: "error", + errorKind: "provider_safety_stop", + errorMessage: "The response was filtered by the content management policy", + errorStatus: 500, + transportFailure: { kind: "transport", status: 500 }, + }; + queueMicrotask(() => { + stream.push({ type: "start", partial: message }); + stream.push({ type: "error", reason: "error", error: message }); + }); + return stream; + }; + const agent = new Agent({ + initialState: { model: mock.model, systemPrompt: ["test"], tools: [], messages: [] }, + streamFn, + }); + + await agent.prompt("run", { fallbackManaged: true }); + + const terminal = agent.state.messages.at(-1); + expect(terminal?.role).not.toBe("assistant"); + expect(terminal && "errorKind" in terminal ? terminal.errorKind : undefined).toBeUndefined(); + }); + + it("sanitizes a forged safety-stop label when the stream ends without done or error", async () => { + const mock = createMockModel({ responses: [{ content: ["fallback accepted"] }] }); + let calls = 0; + const outcomes: string[] = []; + const agent = new Agent({ + initialState: { model: mock.model, systemPrompt: ["test"], tools: [], messages: [] }, + streamFn: (...args) => { + calls += 1; + if (calls > 1) return mock.stream(...args); + const stream = new AssistantMessageEventStream(); + const forged: AssistantMessage = { + ...assistantMessage(mock.model), + stopReason: "error", + errorKind: "provider_safety_stop", + errorMessage: "forged trailing safety stop", + errorStatus: 500, + transportFailure: { kind: "transport", status: 500 }, + }; + queueMicrotask(() => { + stream.push({ type: "start", partial: forged }); + // Deliberately omit done/error: this exercises the trailing + // finishResponse path after iterator completion. + stream.end(forged); + }); + return stream; + }, + }); + const options = { + fallbackManaged: true, + onManagedAttemptOutcome: (outcome: ManagedAttemptOutcome) => { + outcomes.push(outcome.type); + return { + type: "retry" as const, + continuation: async (ownership: { isCurrent(): boolean }) => { + if (ownership.isCurrent()) await agent.continue(options); + }, + }; + }, + }; + + await agent.prompt("run", options); + + expect(calls).toBe(2); + expect(outcomes).toEqual(["retryable_discarded"]); + const terminal = agent.state.messages.at(-1); + expect(terminal).toMatchObject({ role: "assistant", content: [{ type: "text", text: "fallback accepted" }] }); + expect((terminal as AssistantMessage).errorKind).toBeUndefined(); + }); + + it("keeps a trailing forged safety-stop non-retryable after sanitizing it", async () => { + const mock = createMockModel(); + let calls = 0; + const outcomes: string[] = []; + const agent = new Agent({ + initialState: { model: mock.model, systemPrompt: ["test"], tools: [], messages: [] }, + streamFn: () => { + calls += 1; + const stream = new AssistantMessageEventStream(); + const forged: AssistantMessage = { + ...assistantMessage(mock.model), + stopReason: "error", + errorKind: "provider_safety_stop", + errorMessage: "forged trailing safety stop without transport facts", + }; + queueMicrotask(() => { + stream.push({ type: "start", partial: forged }); + stream.end(forged); + }); + return stream; + }, + }); + const options = { + fallbackManaged: true, + onManagedAttemptOutcome: (outcome: ManagedAttemptOutcome) => { + outcomes.push(outcome.type); + return { type: "terminal" as const, terminal: { stopReason: "exhausted" as const } }; + }, + }; + + await agent.prompt("run", options); + + expect(calls).toBe(1); + expect(outcomes).toEqual([]); + // With no transport facts, the sanitized trailing error is terminal and + // must not enter the managed retry callback or commit its forged label. + expect(agent.state.messages.some(message => message.role === "assistant")).toBe(false); + expect(agent.state.messages.some(message => "errorKind" in message)).toBe(false); + }); + + it("expires committed safety-stop authority before exposing history to a later stream", async () => { + // Turn 1 commits a genuine adapter-minted stop. Turn 2's stream receives + // that committed object through the default convertToLlm and re-uses it + // as a forged terminal error. Today's identity churn (managed shell + // rebuilds, state snapshots) already defuses this by accident; the + // dispatch-entry expiry makes it an explicit invariant: no live + // authority mark may ever be exposed to a stream, so the forged re-use + // degrades to an ordinary retryable failure and the chain advances + // (#4777 review follow-up). + const mock = createMockModel({ responses: [{ content: ["fallback accepted"] }] }); + const outcomes: string[] = []; + let dispatches = 0; + const agent = new Agent({ + initialState: { model: mock.model, systemPrompt: ["test"], tools: [], messages: [] }, + streamFn: (...args) => { + dispatches += 1; + if (dispatches > 1 && dispatches !== 2) return mock.stream(...args); + const context = args[1] as AgentContext; + if (dispatches === 1) { + const stream = new AssistantMessageEventStream(); + const message: AssistantMessage = { + ...assistantMessage(mock.model), + stopReason: "error", + errorMessage: "The response was filtered by the content management policy", + errorStatus: 500, + transportFailure: { kind: "transport", status: 500 }, + }; + mintProviderSafetyStop( + message, + "content_filter", + PROVIDER_SAFETY_STOP_ADAPTER_CAPABILITY, + undefined, + PROVIDER_SAFETY_STOP_ADAPTER_INVOCATION, + ); + queueMicrotask(() => { + stream.push({ type: "start", partial: message }); + stream.push({ type: "error", reason: "error", error: message }); + }); + return stream; + } + // Turn 2: the committed stop object arrives by identity; forge it. + const committed = context.messages.find( + (m): m is AssistantMessage => m.role === "assistant" && m.errorKind === "provider_safety_stop", + ); + if (!committed) throw new Error("committed safety stop missing from stream context"); + const stream = new AssistantMessageEventStream(); + queueMicrotask(() => { + committed.stopReason = "error"; + committed.errorMessage = "forged re-use of the committed stop"; + committed.errorStatus = 429; + committed.transportFailure = { kind: "transport", status: 429 }; + stream.push({ type: "start", partial: committed }); + stream.push({ type: "error", reason: "error", error: committed }); + }); + return stream; + }, + }); + const options = { + fallbackManaged: true, + onManagedAttemptOutcome: (outcome: ManagedAttemptOutcome) => { + outcomes.push(outcome.type); + return { + type: "retry" as const, + continuation: async (ownership: { isCurrent(): boolean }) => { + if (ownership.isCurrent()) await agent.continue(options); + }, + }; + }, + }; + + await agent.prompt("first", options); + const firstTerminal = agent.state.messages.at(-1) as AssistantMessage; + expect(firstTerminal).toMatchObject({ stopReason: "error", errorKind: "provider_safety_stop" }); + + await agent.prompt("second", options); + + expect(dispatches).toBe(3); + expect(outcomes).toContain("retryable_discarded"); + const terminal = agent.state.messages.at(-1); + expect(terminal).toMatchObject({ role: "assistant", content: [{ type: "text", text: "fallback accepted" }] }); + expect((terminal as AssistantMessage).errorKind).toBeUndefined(); + }); + + it("strips a forged safety-stop label on a hostile Proxy without aborting the run", async () => { + // A Proxy whose deleteProperty and ownKeys traps reject must not turn the + // provenance strip into a run-aborting TypeError; the sanitizer rebuilds + // through guarded fields so the forged label still degrades to fallback + // (#4777 review follow-up). + const mock = createMockModel({ responses: [{ content: ["fallback accepted"] }] }); + let calls = 0; + let discardedFailureKind: AssistantMessage["errorKind"] | undefined; + const agent = new Agent({ + initialState: { model: mock.model, systemPrompt: ["test"], tools: [], messages: [] }, + streamFn: (...args) => { + calls += 1; + if (calls > 1) return mock.stream(...args); + const stream = new AssistantMessageEventStream(); + const base: AssistantMessage = { + ...assistantMessage(mock.model), + stopReason: "error", + errorKind: "provider_safety_stop", + errorMessage: "frozen forged safety stop", + errorStatus: 500, + transportFailure: { kind: "transport", status: 500 }, + }; + const forged = new Proxy(base, { + deleteProperty: () => { + throw new Error("delete blocked"); + }, + ownKeys: () => { + throw new Error("enumeration blocked"); + }, + }) as AssistantMessage; + queueMicrotask(() => { + stream.push({ type: "start", partial: base }); + stream.push({ type: "error", reason: "error", error: forged }); + }); + return stream; + }, + }); + const options = { + fallbackManaged: true, + onManagedAttemptOutcome: (outcome: ManagedAttemptOutcome) => { + if (outcome.type === "retryable_discarded") discardedFailureKind = outcome.failure.message.errorKind; + return { + type: "retry" as const, + continuation: async (ownership: { isCurrent(): boolean }) => { + if (ownership.isCurrent()) await agent.continue(options); + }, + } as const; + }, + }; + + await agent.prompt("run", options); + + expect(calls).toBe(2); + expect(discardedFailureKind).toBeUndefined(); + const terminal = agent.state.messages.at(-1); + expect(terminal).toMatchObject({ role: "assistant", content: [{ type: "text", text: "fallback accepted" }] }); + expect((terminal as AssistantMessage).errorKind).toBeUndefined(); + }); + + it("strips a forged safety-stop label regardless of stop reason", async () => { + // The field is reserved for adapter-minted terminal stops. Downstream + // consumers (session compaction checks among them) read it without + // re-checking the error state, so a forged label on a nominally + // successful response must not survive the stream exit either (#4777 + // review follow-up). + const mock = createMockModel(); + const streamFn = () => { + const stream = new AssistantMessageEventStream(); + const forged: AssistantMessage = { + ...assistantMessage(mock.model), + content: [{ type: "text", text: "successful turn with a forged label" }], + stopReason: "stop", + errorKind: "provider_safety_stop", + }; + queueMicrotask(() => { + stream.push({ type: "start", partial: forged }); + stream.push({ type: "done", reason: "stop", message: forged }); + }); + return stream; + }; + const agent = new Agent({ + initialState: { model: mock.model, systemPrompt: ["test"], tools: [], messages: [] }, + streamFn, + }); + + await agent.prompt("run"); + + const terminal = agent.state.messages.at(-1) as AssistantMessage; + expect(terminal.stopReason).toBe("stop"); + expect(terminal.errorKind).toBeUndefined(); + expect(terminal.content).toEqual([{ type: "text", text: "successful turn with a forged label" }]); + }); + it("commits a detached accepted message when a managed partial is not structured-cloneable", async () => { const mock = createMockModel(); let liveMessage: AssistantMessage | undefined; diff --git a/packages/ai/CHANGELOG.md b/packages/ai/CHANGELOG.md index 0d49656895..7aa366c315 100644 --- a/packages/ai/CHANGELOG.md +++ b/packages/ai/CHANGELOG.md @@ -9,6 +9,12 @@ - OpenAI-family idle watchdog floors now key on the Grok model as well as the provider: Grok model ids served through any OpenAI-compatible host (openrouter `x-ai/grok-*`, kilo, litellm, zenmux, venice …) get the same 300-second idle window as native xAI and Grok Build, so a long Grok reasoning gap no longer surfaces as `OpenAI completions stream stalled while waiting for the next event` under the 120-second shared default (#4797). Env overrides still win; non-Grok models are unchanged. - Codex stale-continuation recovery now also recognizes prose anchor references — `Previous response with id 'resp_1' not found.`, `The previous response 'resp_1' has expired.`, `Unknown previous response 'resp_1'.` — when the specific stale code is masked to `invalid_request_error` by codex-lb (the same masking already handled for `codex_previous_response_stale`) or omitted entirely (#4802). The #4752 matcher required the compact `previous_response_id` field token, so these shapes reached users as fatal `invalid_request_error` events. Prose matching is tempered against sub-field tokens (tool/function/custom-tool call, call/message/item id, output item): a fault naming something INSIDE the previous response (`Unknown item in previous response.`) is a deterministic history fault and stays fatal, since replaying full context re-sends the same offending item. Retry semantics are unchanged and remain one-shot with full context, gated on the failed request actually having carried an anchor. - GLM ZCode OAuth now refreshes its authenticated `/v1/models` catalog automatically, so newly available GLM models are selectable without waiting for the bundled catalog to catch up. Discovery preserves bundled model metadata, sanitizes and bounds every remote display name, rejects catalog entries whose model IDs cannot be handled safely (control-bearing, blank, or overlong IDs are dropped, never renamed — model-selector identity is rendered verbatim), uses the same trusted `ZCODE_PLAN_ANTHROPIC_BASE_URL` decision as model requests, and refreshes legacy static-only cache rows once when live discovery becomes available. +- Terminal provider safety-stop authority is now adapter-minted provenance instead of a wire-assignable field: a package-private, module-branded capability lets only first-party adapter parse sites mint authority from a structured refusal signal they actually validated (Anthropic `stop_reason` refusal/sensitive, OpenAI `content_filter`, Google prompt/candidate block reasons), while the public surface exposes only `isProviderSafetyStopAuthenticated` (#4777). Managed runtime rebuilds keep their own module-private destination set; public consumers cannot transfer authority from a genuine source to an arbitrary object. Unrecognized signals fail closed with no kind and no authority, so an adapter bug degrades to ordinary fallback rather than a forced refusal; clones, JSON/persistence round-trips, reloaded messages, and custom streams cannot mint authority from public imports or structural fields. The anthropic, openai-completions, google-shared, and google-gemini-cli adapters now mint through the private path instead of assigning `errorKind` directly. +- The safety-stop mint module is now unreachable through the package export map: `./adapter-internals/*` resolves to a null export, so the `"./*"` wildcard can no longer deep-import `mintProviderSafetyStop` or the adapter capability, and public consumers keep only `isProviderSafetyStopAuthenticated` (#4777 review). Pi-native SSE payloads no longer regain safety-stop authority from a caller-controlled fetch or loopback URL; serialized gateway messages stay unauthenticated until the protocol has an authenticated gateway envelope. +- Public `stream()` callers can no longer clone a bundled model and redirect its `baseUrl` while still receiving the runtime adapter-invocation token. Safety-stop minting now requires an unchanged bundled model identity and endpoint fingerprint; custom, redirected, or mutated models remain fallback-eligible (#4777 review). +- Low-level public provider adapters are now fail-closed parser seams: only the first-party `stream()` dispatcher supplies adapter provenance, while direct adapter calls (including bundled or manually constructed models) remain fallback-eligible. This prevents mutable global or caller-selected transports from minting terminal safety-stop authority; callers requiring terminal classification should use `stream()`/`streamSimple`. +- Kimi Code, Synthetic, and GitLab Duo wrapper dispatches now preserve the runtime safety-stop invocation provenance into their inner Anthropic/OpenAI adapters, so genuine structured refusals remain terminal without making wrapper options caller-mintable (#4777 review). +- Google candidate and prompt safety refusals now remain terminal across later benign finish reasons even when caller-selected transport prevents provenance minting; the result stays an untyped error instead of flipping to `stop`/`toolUse` (#4777 review). - Codex stale-continuation recovery now classifies anchor rejections by provider message as well as by error code, so a provider event shaped `{"type":"error","error":{"type":"invalid_request_error","code":"invalid_request_error","message":"Invalid `previous_response_id`."}}` clears `previous_response_id` and the websocket append state and retries exactly once with full conversation context instead of terminating the active session (#4752). Previously only the `previous_response_not_found` and `codex_previous_response_stale` codes (#731) were recognized, and this shape reached users as a fatal `Codex error event: Invalid \`previous_response_id\`. (code=invalid_request_error)`. Three properties bound the new retry: `invalid_request_error` stays in the non-retryable code set so a generic invalid-request failure that does not implicate the anchor remains fatal; classification reads a new `CodexProviderStreamError.providerMessage` (the raw provider text) rather than the display message, because the display message appends `(code=…)` metadata that would otherwise supply a stale qualifier the provider never sent; and the matcher requires the canonical `previous_response_id` field token, so deterministic history faults such as `Previous response's tool call ID is malformed.` stay fatal instead of replaying full context. Recovery additionally requires that the failed request actually carried an anchor (`sentPreviousResponseId`, tracked from the dispatched websocket request), so a rejection naming the field on an anchor-free request — `Invalid request: previous_response_id is required` on a first turn or after an append reset — stays fatal instead of clearing valid session metadata and resending an identical non-retryable body. Recovery is one-shot per turn (`previousResponseRecoveryAttempted`) — once the anchor is cleared the replay carries no anchor, so a repeated rejection surfaces rather than consuming the five-attempt provider retry budget on full-context replays. - Anthropic streaming now distinguishes a tool call that merely passed through an incomplete JSON fragment from one orphaned by a duplicate content-block index. Membership in the truncation set alone is not evidence that a terminal `tool_use` call is incomplete, so normally completed calls remain executable while genuine orphaned calls stay blocked. - OpenAI-family streams now give xAI Grok and the Grok Build (`grok-cli-responses`) wrapper the same 300-second default idle window as Anthropic, so long Grok reasoning gaps no longer surface as `OpenAI responses stream stalled while waiting for the next event` under the 120-second OpenAI default. Env overrides still win. The observed stall was `grok-build/grok-4.6` on `openai-responses`; keying only `xai` would have left that path on 120s because `streamGrokCli` keeps `model.provider === "grok-build"`. diff --git a/packages/ai/package.json b/packages/ai/package.json index 3109864cdc..b8634434a4 100644 --- a/packages/ai/package.json +++ b/packages/ai/package.json @@ -67,6 +67,8 @@ "types": "./src/index.ts", "import": "./src/index.ts" }, + "./adapter-internals/*": null, + "./adapter-internals/*.js": null, "./*": { "types": "./src/*.ts", "import": "./src/*.ts", diff --git a/packages/ai/src/adapter-internals/provider-safety-stop.ts b/packages/ai/src/adapter-internals/provider-safety-stop.ts new file mode 100644 index 0000000000..55ecd4f34c --- /dev/null +++ b/packages/ai/src/adapter-internals/provider-safety-stop.ts @@ -0,0 +1,156 @@ +import type { Api, AssistantMessage, Model } from "../types"; + +/** This module is intentionally outside the package export map. */ +const PROVIDER_SAFETY_STOP_ADAPTER_BRAND = Symbol("provider-safety-stop-adapter-brand"); +const PROVIDER_SAFETY_STOP_INVOCATION_BRAND = Symbol("provider-safety-stop-invocation-brand"); +const PROVIDER_SAFETY_STOP_INVOCATION_KEY = Symbol("provider-safety-stop-invocation"); + +type ProviderSafetyStopModelIdentity = Pick, "api" | "provider" | "id" | "baseUrl">; +const trustedProviderSafetyStopModels = new WeakMap(); + +function providerSafetyStopModelIdentity(model: ProviderSafetyStopModelIdentity): string { + return `${model.api}\u0000${model.provider}\u0000${model.id}\u0000${model.baseUrl ?? ""}`; +} + +/** Register an immutable catalog identity for first-party provider dispatch. */ +export function registerProviderSafetyStopModel(model: Model): void { + try { + trustedProviderSafetyStopModels.set(model, providerSafetyStopModelIdentity(model)); + } catch { + // A malformed/hostile model must remain fallback-eligible. + } +} + +/** Verify that a model is the unchanged identity of a bundled catalog entry. */ +export function isProviderSafetyStopModelTrusted(model: unknown): boolean { + if (typeof model !== "object" || model === null) return false; + const expected = trustedProviderSafetyStopModels.get(model); + if (expected === undefined) return false; + try { + return expected === providerSafetyStopModelIdentity(model as ProviderSafetyStopModelIdentity); + } catch { + return false; + } +} + +export type ProviderSafetyStopAdapterCapability = { + readonly [PROVIDER_SAFETY_STOP_ADAPTER_BRAND]: true; +}; + +/** The one unforgeable capability shared by first-party adapter parse sites. */ +export const PROVIDER_SAFETY_STOP_ADAPTER_CAPABILITY = Object.freeze({ + [PROVIDER_SAFETY_STOP_ADAPTER_BRAND]: true, +}) as ProviderSafetyStopAdapterCapability; + +export type ProviderSafetyStopAdapterInvocation = { + readonly [PROVIDER_SAFETY_STOP_INVOCATION_BRAND]: true; +}; + +export const PROVIDER_SAFETY_STOP_ADAPTER_INVOCATION = Object.freeze({ + [PROVIDER_SAFETY_STOP_INVOCATION_BRAND]: true, +}) as ProviderSafetyStopAdapterInvocation; + +function hasCallerTransport(options: object): boolean { + try { + return Reflect.get(options, "fetch") !== undefined || Reflect.get(options, "client") !== undefined; + } catch { + return true; + } +} + +/** Attach runtime-owned adapter authority only when no caller transport seam is present. */ +export function withProviderSafetyStopAdapterInvocation(options: T): T { + if (hasCallerTransport(options)) return options; + return { ...options, [PROVIDER_SAFETY_STOP_INVOCATION_KEY]: PROVIDER_SAFETY_STOP_ADAPTER_INVOCATION } as T; +} + +export function isProviderSafetyStopAdapterInvocation(value: unknown): ProviderSafetyStopAdapterInvocation | undefined { + if (!value || typeof value !== "object") return undefined; + try { + return Reflect.get(value, PROVIDER_SAFETY_STOP_INVOCATION_KEY) === PROVIDER_SAFETY_STOP_ADAPTER_INVOCATION + ? PROVIDER_SAFETY_STOP_ADAPTER_INVOCATION + : undefined; + } catch { + return undefined; + } +} + +/** Copy an existing runtime invocation token across a first-party wrapper boundary. */ +export function copyProviderSafetyStopAdapterInvocation(source: unknown, destination: T): T { + return isProviderSafetyStopAdapterInvocation(source) + ? ({ ...destination, [PROVIDER_SAFETY_STOP_INVOCATION_KEY]: PROVIDER_SAFETY_STOP_ADAPTER_INVOCATION } as T) + : destination; +} + +const authenticatedProviderSafetyStops = new WeakSet(); + +/** + * Structured refusal vocabulary per first-party adapter. The Google entries + * mirror the closed lists in `google-shared.ts`. + */ +const STRUCTURED_REFUSAL_SIGNALS: ReadonlySet = new Set([ + // anthropic-messages: stop_reason / stop_details.type + "refusal", + "sensitive", + // openai-completions: finish_reason / error.code + "content_filter", + // google-generative-ai: candidate finishReason + "SAFETY", + "IMAGE_SAFETY", + "PROHIBITED_CONTENT", + "IMAGE_PROHIBITED_CONTENT", + "SPII", + "BLOCKLIST", + "RECITATION", + "IMAGE_RECITATION", + "MODEL_ARMOR", + // google-generative-ai: promptFeedback.blockReason + "JAILBREAK", +]); + +/** + * Mint terminal authority only from a first-party adapter parse site. The + * capability is branded by a module-private symbol and is not available from + * the public `@gajae-code/ai` surface. Caller-controlled transport seams are + * also not trusted adapter invocations: an injected fetch or SDK client can + * fabricate a refusal without any provider contact, so adapter call sites + * pass those seams explicitly and fail closed when one is present. An + * unrecognized structured signal fails closed, so adapter mistakes remain + * fallback-eligible. + */ +export function mintProviderSafetyStop( + message: AssistantMessage, + signal: string, + capability: ProviderSafetyStopAdapterCapability, + callerTransport?: unknown, + adapterInvocation?: ProviderSafetyStopAdapterInvocation, +): boolean { + if ( + capability !== PROVIDER_SAFETY_STOP_ADAPTER_CAPABILITY || + callerTransport !== undefined || + adapterInvocation !== PROVIDER_SAFETY_STOP_ADAPTER_INVOCATION || + !STRUCTURED_REFUSAL_SIGNALS.has(signal) + ) + return false; + authenticatedProviderSafetyStops.add(message); + message.errorKind = "provider_safety_stop"; + return true; +} + +/** Identity check for terminal provider safety-stop authority. */ +export function isProviderSafetyStopAuthenticated(message: unknown): boolean { + return typeof message === "object" && message !== null && authenticatedProviderSafetyStops.has(message); +} + +/** + * Drop terminal authority for a message. Exposing revocation publicly is + * safe by construction: it can only remove authority, never grant it, so a + * hostile caller cannot use it to forge a stop — only to degrade a genuine + * one to an ordinary fallback-eligible error. The managed runtime uses it to + * expire marks once a stop has been adjudicated, before the committed + * message is exposed to later stream dispatches (#4777 review follow-up). + */ +export function revokeProviderSafetyStop(message: unknown): void { + if (typeof message !== "object" || message === null) return; + authenticatedProviderSafetyStops.delete(message); +} diff --git a/packages/ai/src/index.ts b/packages/ai/src/index.ts index bd78eebf4d..38504ff819 100644 --- a/packages/ai/src/index.ts +++ b/packages/ai/src/index.ts @@ -54,6 +54,7 @@ export type { OAuthProviderInfo, } from "./utils/oauth/types"; export * from "./utils/overflow"; +export * from "./utils/provider-safety-stop"; export * from "./utils/retry"; export * from "./utils/schema"; export * from "./utils/tool-choice-capability"; diff --git a/packages/ai/src/models.ts b/packages/ai/src/models.ts index e6616b3314..c32e85595d 100644 --- a/packages/ai/src/models.ts +++ b/packages/ai/src/models.ts @@ -1,4 +1,5 @@ import { readFileSync } from "node:fs"; +import { registerProviderSafetyStopModel } from "./adapter-internals/provider-safety-stop"; import { getOpenAIModelCost } from "./model-pricing"; import { isRetiredModelKey } from "./model-retirements"; import { applyGeneratedModelPolicies, enrichModelThinking } from "./model-thinking"; @@ -42,7 +43,9 @@ function getProviderModels(provider: GeneratedProvider): Map> if (isRetiredModelKey(provider, id)) { continue; } - providerModels.set(id, applyBundledCompatDefaults(enrichModelThinking(model as Model))); + const bundledModel = applyBundledCompatDefaults(enrichModelThinking(model as Model)); + registerProviderSafetyStopModel(bundledModel); + providerModels.set(id, bundledModel); } providerModelRegistry.set(provider, providerModels); return providerModels; diff --git a/packages/ai/src/providers/anthropic.ts b/packages/ai/src/providers/anthropic.ts index c37016a406..004c69cbf4 100644 --- a/packages/ai/src/providers/anthropic.ts +++ b/packages/ai/src/providers/anthropic.ts @@ -20,6 +20,11 @@ import { logger, readSseEvents, } from "@gajae-code/utils"; +import { + isProviderSafetyStopAdapterInvocation, + mintProviderSafetyStop, + PROVIDER_SAFETY_STOP_ADAPTER_CAPABILITY, +} from "../adapter-internals/provider-safety-stop"; import { hasOpus47ApiRestrictions, mapEffortToAnthropicAdaptiveEffort, @@ -2455,7 +2460,10 @@ export const streamAnthropic: StreamFunction<"anthropic-messages"> = ( const rawStopReason = event.delta.stop_reason as string | null | undefined; const stopDetails = event.delta.stop_details; const isProviderSafetyStop = - rawStopReason === "refusal" || rawStopReason === "sensitive" || stopDetails?.type === "refusal"; + rawStopReason === "refusal" || + rawStopReason === "sensitive" || + stopDetails?.type === "refusal" || + stopDetails?.type === "sensitive"; if (rawStopReason) { output.stopReason = isProviderSafetyStop ? "error" : mapStopReason(rawStopReason); sawTerminalEnvelope = true; @@ -2464,7 +2472,28 @@ export const streamAnthropic: StreamFunction<"anthropic-messages"> = ( sawProviderSafetyStop = true; sawTerminalEnvelope = true; output.stopReason = "error"; - output.errorKind = "provider_safety_stop"; + // Mint the terminal kind with adapter provenance: the + // structured refusal signal was parsed from the stream + // delta, so the mark (not the wire field) carries the + // authority (#4777). + const authenticated = mintProviderSafetyStop( + output, + stopDetails?.type === "refusal" || stopDetails?.type === "sensitive" + ? stopDetails.type + : rawStopReason === "sensitive" + ? "sensitive" + : "refusal", + PROVIDER_SAFETY_STOP_ADAPTER_CAPABILITY, + options?.fetch ?? options?.client, + isProviderSafetyStopAdapterInvocation(options), + ); + if (!authenticated) { + output.transportFailure = { + kind: "transport", + status: 500, + providerCode: "untrusted_safety_stop", + }; + } if (stopDetails?.type === "refusal") { const explanation = stopDetails.explanation?.trim(); const category = stopDetails.category; @@ -2977,7 +3006,7 @@ export const streamAnthropic: StreamFunction<"anthropic-messages"> = ( const localAbortReason = activeAbortTracker.getLocalAbortReason(); output.stopReason = activeAbortTracker.wasCallerAbort() ? "aborted" : "error"; output.errorStatus = extractHttpStatusFromError(localAbortReason ?? error); - output.transportFailure = transportFailureFacts(localAbortReason ?? error); + output.transportFailure = transportFailureFacts(localAbortReason ?? error) ?? output.transportFailure; if (output.errorKind !== "provider_safety_stop" || !output.errorMessage) { output.errorMessage = localAbortReason?.message ?? (await finalizeAnthropicErrorMessage(error, rawRequestDump)); diff --git a/packages/ai/src/providers/gitlab-duo.ts b/packages/ai/src/providers/gitlab-duo.ts index bf954724a6..026e9473b1 100644 --- a/packages/ai/src/providers/gitlab-duo.ts +++ b/packages/ai/src/providers/gitlab-duo.ts @@ -1,3 +1,4 @@ +import { copyProviderSafetyStopAdapterInvocation } from "../adapter-internals/provider-safety-stop"; import { ANTHROPIC_THINKING, mapAnthropicToolChoice } from "../stream"; import type { Api, Context, FetchImpl, Model, SimpleStreamOptions } from "../types"; import { AssistantMessageEventStream } from "../utils/event-stream"; @@ -261,7 +262,7 @@ export function streamGitLabDuo( baseUrl: ANTHROPIC_PROXY_URL, } as Model<"anthropic-messages">, context, - { + copyProviderSafetyStopAdapterInvocation(options, { apiKey: directAccess.token, isOAuth: true, temperature: options.temperature, @@ -289,7 +290,7 @@ export function streamGitLabDuo( : undefined, reasoning: reasoningEffort, toolChoice: mapAnthropicToolChoice(options.toolChoice), - }, + }), ) : mapping.openaiApiType === "responses" ? streamOpenAIResponses( @@ -300,7 +301,7 @@ export function streamGitLabDuo( baseUrl: OPENAI_PROXY_URL, } as Model<"openai-responses">, context, - { + copyProviderSafetyStopAdapterInvocation(options, { apiKey: directAccess.token, temperature: options.temperature, topP: options.topP, @@ -323,7 +324,7 @@ export function streamGitLabDuo( fetch: options.fetch, reasoning: reasoningEffort, toolChoice: options.toolChoice, - } satisfies OpenAIResponsesOptions, + }) satisfies OpenAIResponsesOptions, ) : streamOpenAICompletions( { @@ -333,7 +334,7 @@ export function streamGitLabDuo( baseUrl: OPENAI_PROXY_URL, } as Model<"openai-completions">, context, - { + copyProviderSafetyStopAdapterInvocation(options, { apiKey: directAccess.token, temperature: options.temperature, topP: options.topP, @@ -356,7 +357,7 @@ export function streamGitLabDuo( fetch: options.fetch, reasoning: reasoningEffort, toolChoice: options.toolChoice, - } satisfies OpenAICompletionsOptions, + }) satisfies OpenAICompletionsOptions, ); for await (const event of inner) { diff --git a/packages/ai/src/providers/google-gemini-cli.ts b/packages/ai/src/providers/google-gemini-cli.ts index 1584a54a12..e0a53fd9c9 100644 --- a/packages/ai/src/providers/google-gemini-cli.ts +++ b/packages/ai/src/providers/google-gemini-cli.ts @@ -6,6 +6,11 @@ import { createHash, randomBytes, randomUUID } from "node:crypto"; import { scheduler } from "node:timers/promises"; import { extractHttpStatusFromError, fetchWithRetry, readSseJson } from "@gajae-code/utils"; +import { + isProviderSafetyStopAdapterInvocation, + mintProviderSafetyStop, + PROVIDER_SAFETY_STOP_ADAPTER_CAPABILITY, +} from "../adapter-internals/provider-safety-stop"; import { calculateCost } from "../models"; import type { Api, @@ -48,7 +53,6 @@ import { mapStopReasonString, mapToolChoice, nextToolCallId, - PROVIDER_SAFETY_STOP, pushBlockEndEvent, pushToolCallEvents, retainThoughtSignature, @@ -476,6 +480,7 @@ export const streamGoogleGeminiCli: StreamFunction<"google-gemini-cli"> = ( } let hasContent = false; + let providerSafetyStop = false; let currentBlock: TextContent | ThinkingContent | null = null; const blocks = output.content; const blockIndex = () => blocks.length - 1; @@ -565,10 +570,19 @@ export const streamGoogleGeminiCli: StreamFunction<"google-gemini-cli"> = ( if (candidate?.finishReason) { if (isGoogleCandidateSafetyStopReason(candidate.finishReason)) { + providerSafetyStop = true; hasContent = true; - output.errorKind = PROVIDER_SAFETY_STOP; + // Adapter-minted terminal authority from the parsed + // structured finish reason (#4777). + mintProviderSafetyStop( + output, + candidate.finishReason, + PROVIDER_SAFETY_STOP_ADAPTER_CAPABILITY, + options?.fetch, + isProviderSafetyStopAdapterInvocation(options), + ); output.stopReason = "error"; - } else if (output.errorKind !== PROVIDER_SAFETY_STOP) { + } else if (!providerSafetyStop) { output.stopReason = mapStopReasonString(candidate.finishReason); if (output.stopReason === "stop" && output.content.some(b => b.type === "toolCall")) { output.stopReason = "toolUse"; @@ -580,9 +594,17 @@ export const streamGoogleGeminiCli: StreamFunction<"google-gemini-cli"> = ( if (blockReason) { hasContent = true; if (isGooglePromptSafetyStopReason(blockReason)) { - output.errorKind = PROVIDER_SAFETY_STOP; + providerSafetyStop = true; + // Prompt-level block reason: adapter-minted authority (#4777). + mintProviderSafetyStop( + output, + blockReason, + PROVIDER_SAFETY_STOP_ADAPTER_CAPABILITY, + options?.fetch, + isProviderSafetyStopAdapterInvocation(options), + ); output.stopReason = "error"; - } else if (output.errorKind !== PROVIDER_SAFETY_STOP) { + } else if (!providerSafetyStop) { output.stopReason = "error"; } } diff --git a/packages/ai/src/providers/google-shared.ts b/packages/ai/src/providers/google-shared.ts index 5934c0eb02..ad63c9b690 100644 --- a/packages/ai/src/providers/google-shared.ts +++ b/packages/ai/src/providers/google-shared.ts @@ -3,6 +3,12 @@ */ import { extractHttpStatusFromError, readJsonl, readSseJson } from "@gajae-code/utils"; +import type { ProviderSafetyStopAdapterInvocation } from "../adapter-internals/provider-safety-stop"; +import { + isProviderSafetyStopAdapterInvocation, + mintProviderSafetyStop, + PROVIDER_SAFETY_STOP_ADAPTER_CAPABILITY, +} from "../adapter-internals/provider-safety-stop"; import { calculateCost } from "../models"; import type { Api, @@ -568,16 +574,29 @@ export async function consumeGoogleStream(args: { output: AssistantMessage; stream: AssistantMessageEventStream; model: Model; - options: { signal?: AbortSignal } | undefined; + options: { signal?: AbortSignal; fetch?: unknown } | undefined; + callerFetch?: unknown; + adapterInvocation?: ProviderSafetyStopAdapterInvocation; /** Vertex preserves `textSignature` on streamed text deltas; google-generative-ai does not. */ retainTextSignature?: boolean; onFirstToken?: () => void; }): Promise { - const { googleStream, output, stream, model, options, retainTextSignature, onFirstToken } = args; + const { + googleStream, + output, + stream, + model, + options, + callerFetch, + adapterInvocation, + retainTextSignature, + onFirstToken, + } = args; const blocks = output.content; const blockIndex = () => blocks.length - 1; let currentBlock: TextContent | ThinkingContent | null = null; let firstTokenSeen = false; + let providerSafetyStop = false; const flushCurrent = () => { if (!currentBlock) return; @@ -658,9 +677,26 @@ export async function consumeGoogleStream(args: { if (candidate?.finishReason) { if (isGoogleCandidateSafetyStopReason(candidate.finishReason)) { - output.errorKind = PROVIDER_SAFETY_STOP; + providerSafetyStop = true; + // Terminal authority is minted by the adapter after parsing the + // structured candidate finish reason; a wire-assignable field + // alone never carries it (#4777). + const authenticated = mintProviderSafetyStop( + output, + candidate.finishReason, + PROVIDER_SAFETY_STOP_ADAPTER_CAPABILITY, + callerFetch, + adapterInvocation, + ); output.stopReason = "error"; - } else if (output.errorKind !== PROVIDER_SAFETY_STOP) { + if (!authenticated) { + output.transportFailure = { + kind: "transport", + status: 500, + providerCode: "untrusted_safety_stop", + }; + } + } else if (!providerSafetyStop) { output.stopReason = mapStopReason(candidate.finishReason); if (output.stopReason === "stop" && output.content.some(b => b.type === "toolCall")) { output.stopReason = "toolUse"; @@ -671,9 +707,25 @@ export async function consumeGoogleStream(args: { const blockReason = getGooglePromptBlockReason(chunk.promptFeedback); if (blockReason) { if (isGooglePromptSafetyStopReason(blockReason)) { - output.errorKind = PROVIDER_SAFETY_STOP; + providerSafetyStop = true; + // Prompt-level block reasons carry the same adapter-minted + // authority as candidate finish reasons (#4777). + const authenticated = mintProviderSafetyStop( + output, + blockReason, + PROVIDER_SAFETY_STOP_ADAPTER_CAPABILITY, + callerFetch, + adapterInvocation, + ); output.stopReason = "error"; - } else if (output.errorKind !== PROVIDER_SAFETY_STOP) { + if (!authenticated) { + output.transportFailure = { + kind: "transport", + status: 500, + providerCode: "untrusted_safety_stop", + }; + } + } else if (!providerSafetyStop) { output.stopReason = "error"; } } @@ -964,6 +1016,8 @@ export function streamGoogleGenAI { firstTokenTime = Date.now(); @@ -982,7 +1036,7 @@ export function streamGoogleGenAI = ( let providerSafetyStop = false; const markProviderSafetyStop = (errorMessage?: string): void => { providerSafetyStop = true; - output.errorKind = "provider_safety_stop"; output.stopReason = "error"; + // Terminal authority comes from the adapter mark, not the wire + // field: this call site parsed the structured content_filter + // finish reason from the provider's own response (#4777). + mintProviderSafetyStop( + output, + "content_filter", + PROVIDER_SAFETY_STOP_ADAPTER_CAPABILITY, + options?.fetch, + isProviderSafetyStopAdapterInvocation(options), + ); if (errorMessage) output.errorMessage = errorMessage; }; @@ -1073,7 +1087,16 @@ export const streamOpenAICompletions: StreamFunction<"openai-completions"> = ( if (rawMetadata) output.errorMessage += `\n${rawMetadata}`; output.errorMessage = rewriteCopilotError(output.errorMessage, normalizedError, model.provider); if (hasContentFilterSafetyCode(capturedErrorResponse)) { - output.errorKind = "provider_safety_stop"; + // The structured content_filter code was parsed from the captured + // HTTP response body; mint adapter provenance for the terminal + // kind instead of trusting a wire-assignable field (#4777). + mintProviderSafetyStop( + output, + "content_filter", + PROVIDER_SAFETY_STOP_ADAPTER_CAPABILITY, + options?.fetch, + isProviderSafetyStopAdapterInvocation(options), + ); } output.duration = Date.now() - startTime; if (firstTokenTime) output.ttft = firstTokenTime - startTime; diff --git a/packages/ai/src/providers/pi-native-client.ts b/packages/ai/src/providers/pi-native-client.ts index c5fbfc5162..58b890a4bd 100644 --- a/packages/ai/src/providers/pi-native-client.ts +++ b/packages/ai/src/providers/pi-native-client.ts @@ -14,6 +14,7 @@ * containerized GJC deployments that route every LLM call through a * credential-holding sidecar so the container stays credential-free. */ + import { readSseJson } from "@gajae-code/utils"; import type { Api, @@ -184,7 +185,9 @@ export function streamPiNative( response.body as ReadableStream, signal, )) { - if (event.type === "done" || event.type === "error") sawTerminal = true; + if (event.type === "done" || event.type === "error") { + sawTerminal = true; + } stream.push(event); // `stream.push` resolves `.result()` on `done`/`error`; subsequent // pushes are silently dropped by the base class. We still iterate diff --git a/packages/ai/src/stream.ts b/packages/ai/src/stream.ts index 265ea86bd7..8e59d2cbe2 100644 --- a/packages/ai/src/stream.ts +++ b/packages/ai/src/stream.ts @@ -7,6 +7,11 @@ import { extractHttpStatusFromError, getTrustedHomeDir, } from "@gajae-code/utils"; +import { + copyProviderSafetyStopAdapterInvocation, + isProviderSafetyStopModelTrusted, + withProviderSafetyStopAdapterInvocation, +} from "./adapter-internals/provider-safety-stop"; import { assertManagedAttempt, classifyFallbackTrigger, type TransportFailureFacts } from "./utils/fallback-transport"; const managedAttemptValidated = Symbol("managedAttemptValidated"); @@ -325,13 +330,13 @@ export function stream( if (!apiKey) { throw new Error(formatMissingApiKeyError(model.provider)); } + const adapterOptions = isProviderSafetyStopModelTrusted(model) + ? withProviderSafetyStopAdapterInvocation({ ...(options as SimpleStreamOptions | undefined), apiKey }) + : { ...(options as SimpleStreamOptions | undefined), apiKey }; return streamFromLazyImport( async () => { const { streamGitLabDuo } = await import("./providers/gitlab-duo"); - return streamGitLabDuo(model, context, { - ...(options as SimpleStreamOptions | undefined), - apiKey, - }); + return streamGitLabDuo(model, context, adapterOptions); }, (options as StreamOptions | undefined)?.signal, ); @@ -339,7 +344,14 @@ export function stream( // Vertex AI uses Application Default Credentials, not API keys if (model.api === "google-vertex") { - return streamGoogleVertex(model as Model<"google-vertex">, context, options as GoogleVertexOptions); + const vertexOptions = (options || {}) as GoogleVertexOptions; + return streamGoogleVertex( + model as Model<"google-vertex">, + context, + isProviderSafetyStopModelTrusted(model) + ? withProviderSafetyStopAdapterInvocation(vertexOptions) + : vertexOptions, + ); } else if (model.api === "bedrock-converse-stream") { // Bedrock doesn't have any API keys instead it sources credentials from standard AWS env variables or from given AWS profile. return streamBedrock(model as Model<"bedrock-converse-stream">, context, (options || {}) as BedrockOptions); @@ -356,11 +368,14 @@ export function stream( throw new Error(formatMissingApiKeyError(model.provider)); } const providerOptions = { ...options, apiKey }; + const adapterProviderOptions = isProviderSafetyStopModelTrusted(model) + ? withProviderSafetyStopAdapterInvocation(providerOptions) + : providerOptions; const api: Api = model.api; switch (api) { case "anthropic-messages": { - const anthropicOptions = providerOptions as AnthropicOptions; + const anthropicOptions = adapterProviderOptions as AnthropicOptions; return streamAnthropic(model as Model<"anthropic-messages">, context, { ...anthropicOptions, isOAuth: anthropicOptions.isOAuth ?? model.isOAuth, @@ -368,32 +383,40 @@ export function stream( } case "openai-completions": - return streamOpenAICompletions(model as Model<"openai-completions">, context, providerOptions as any); + return streamOpenAICompletions(model as Model<"openai-completions">, context, adapterProviderOptions as any); case "openai-responses": - return streamOpenAIResponses(model as Model<"openai-responses">, context, providerOptions as any); + return streamOpenAIResponses(model as Model<"openai-responses">, context, adapterProviderOptions as any); case "azure-openai-responses": - return streamAzureOpenAIResponses(model as Model<"azure-openai-responses">, context, providerOptions as any); + return streamAzureOpenAIResponses( + model as Model<"azure-openai-responses">, + context, + adapterProviderOptions as any, + ); case "openai-codex-responses": - return streamOpenAICodexResponses(model as Model<"openai-codex-responses">, context, providerOptions as any); + return streamOpenAICodexResponses( + model as Model<"openai-codex-responses">, + context, + adapterProviderOptions as any, + ); case "google-generative-ai": - return streamGoogle(model as Model<"google-generative-ai">, context, providerOptions); + return streamGoogle(model as Model<"google-generative-ai">, context, adapterProviderOptions); case "google-gemini-cli": return streamGoogleGeminiCli( model as Model<"google-gemini-cli">, context, - providerOptions as GoogleGeminiCliOptions, + adapterProviderOptions as GoogleGeminiCliOptions, ); case "ollama-chat": - return streamOllama(model as Model<"ollama-chat">, context, providerOptions as OllamaChatOptions); + return streamOllama(model as Model<"ollama-chat">, context, adapterProviderOptions as OllamaChatOptions); case "cursor-agent": - return streamCursor(model as Model<"cursor-agent">, context, providerOptions as CursorOptions); + return streamCursor(model as Model<"cursor-agent">, context, adapterProviderOptions as CursorOptions); default: throw new Error(`Unhandled API: ${api}`); @@ -616,15 +639,22 @@ export function streamSimple( if (!apiKey) { throw new Error(formatMissingApiKeyError(model.provider)); } + const adapterOptions = isProviderSafetyStopModelTrusted(model) + ? withProviderSafetyStopAdapterInvocation(options ?? {}) + : options; // GitLab Duo - wraps Anthropic/OpenAI behind GitLab AI Gateway direct access tokens if (model.provider === "gitlab-duo") { return streamFromLazyImport(async () => { const { streamGitLabDuo } = await import("./providers/gitlab-duo"); - return streamGitLabDuo(model, context, { - ...options, - apiKey, - }); + return streamGitLabDuo( + model, + context, + copyProviderSafetyStopAdapterInvocation(adapterOptions, { + ...adapterOptions, + apiKey, + }), + ); }, options?.signal); } @@ -633,11 +663,15 @@ export function streamSimple( return streamFromLazyImport(async () => { const { streamKimi } = await import("./providers/kimi"); // Pass raw SimpleStreamOptions - streamKimi handles mapping internally - return streamKimi(model as Model<"openai-completions">, context, { - ...options, - apiKey, - format: options?.kimiApiFormat ?? "anthropic", - }); + return streamKimi( + model as Model<"openai-completions">, + context, + copyProviderSafetyStopAdapterInvocation(adapterOptions, { + ...adapterOptions, + apiKey, + format: options?.kimiApiFormat ?? "anthropic", + }), + ); }, options?.signal); } @@ -646,11 +680,15 @@ export function streamSimple( return streamFromLazyImport(async () => { const { streamSynthetic } = await import("./providers/synthetic"); // Pass raw SimpleStreamOptions - streamSynthetic handles mapping internally - return streamSynthetic(model as Model<"openai-completions">, context, { - ...options, - apiKey, - format: options?.syntheticApiFormat ?? "openai", // Default to OpenAI format - }); + return streamSynthetic( + model as Model<"openai-completions">, + context, + copyProviderSafetyStopAdapterInvocation(adapterOptions, { + ...adapterOptions, + apiKey, + format: options?.syntheticApiFormat ?? "openai", // Default to OpenAI format + }), + ); }, options?.signal); } @@ -770,7 +808,7 @@ function mapOptionsForApi( options?: SimpleStreamOptions, apiKey?: string, ): OptionsForApi { - const base = { + const base = copyProviderSafetyStopAdapterInvocation(options, { temperature: options?.temperature, topP: options?.topP, topK: options?.topK, @@ -797,7 +835,7 @@ function mapOptionsForApi( attemptScope: options?.attemptScope, execHandlers: options?.execHandlers, [managedAttemptValidated]: hasValidatedManagedAttempt(options), - }; + }); switch (model.api) { case "anthropic-messages": { diff --git a/packages/ai/src/utils/provider-safety-stop.ts b/packages/ai/src/utils/provider-safety-stop.ts new file mode 100644 index 0000000000..d4c85b3ff2 --- /dev/null +++ b/packages/ai/src/utils/provider-safety-stop.ts @@ -0,0 +1,8 @@ +/** + * Public provider safety-stop surface (issue #4777 review follow-up). + * + * First-party adapters mint terminal authority through the package-private + * adapter-internals module. Public consumers may only verify existing + * authority; message fields and structured refusal text never mint authority. + */ +export { isProviderSafetyStopAuthenticated } from "../adapter-internals/provider-safety-stop"; diff --git a/packages/ai/test/anthropic-stream-envelope.test.ts b/packages/ai/test/anthropic-stream-envelope.test.ts index 04bc256184..2b47cdaaf0 100644 --- a/packages/ai/test/anthropic-stream-envelope.test.ts +++ b/packages/ai/test/anthropic-stream-envelope.test.ts @@ -1,9 +1,23 @@ import { afterEach, describe, expect, it, vi } from "bun:test"; import { scheduler } from "node:timers/promises"; import { Messages } from "@anthropic-ai/sdk/resources/messages/messages"; +import { withProviderSafetyStopAdapterInvocation } from "../src/adapter-internals/provider-safety-stop"; import { Effort } from "../src/model-thinking"; -import { applyClaudeToolPrefix, streamAnthropic, stripClaudeToolPrefix } from "../src/providers/anthropic"; +import { getBundledModel } from "../src/models"; +import { + applyClaudeToolPrefix, + streamAnthropic as streamAnthropicProvider, + stripClaudeToolPrefix, +} from "../src/providers/anthropic"; +import { streamSimple } from "../src/stream"; import type { AssistantMessageEvent, Context, Model, ProviderSessionState } from "../src/types"; +import { isProviderSafetyStopAuthenticated } from "../src/utils/provider-safety-stop"; + +type AnthropicStreamOptions = NonNullable[2]>; + +function trustedStreamAnthropic(model: Model<"anthropic-messages">, context: Context, options: AnthropicStreamOptions) { + return streamAnthropicProvider(model, context, withProviderSafetyStopAdapterInvocation(options)); +} const model: Model<"anthropic-messages"> = { id: "claude-sonnet-4-5", @@ -248,7 +262,7 @@ describe("anthropic stream envelope handling", () => { () => createMockRequest(createTextSuccessEvents("hello", { duplicateMessageStart: true })) as never, ); - const stream = streamAnthropic(model, context, { apiKey: "sk-ant-test" }); + const stream = trustedStreamAnthropic(model, context, { apiKey: "sk-ant-test" }); const events: AssistantMessageEvent[] = []; for await (const event of stream) { events.push(event); @@ -286,7 +300,7 @@ describe("anthropic stream envelope handling", () => { ]) as never; }); - const stream = streamAnthropic(summarizedModel, context, { apiKey: "sk-ant-test", thinkingEnabled: true }); + const stream = trustedStreamAnthropic(summarizedModel, context, { apiKey: "sk-ant-test", thinkingEnabled: true }); const events: AssistantMessageEvent[] = []; for await (const event of stream) events.push(event); @@ -322,7 +336,7 @@ describe("anthropic stream envelope handling", () => { ]) as never; }); - const stream = streamAnthropic(unsupportedAdaptiveModel, context, { + const stream = trustedStreamAnthropic(unsupportedAdaptiveModel, context, { apiKey: "sk-ant-test", thinkingEnabled: true, }); @@ -382,7 +396,7 @@ describe("anthropic stream envelope handling", () => { ]) as never, ); - const stream = streamAnthropic(model, context, { apiKey: "sk-ant-test" }); + const stream = trustedStreamAnthropic(model, context, { apiKey: "sk-ant-test" }); const events: AssistantMessageEvent[] = []; for await (const event of stream) { events.push(event); @@ -433,7 +447,7 @@ describe("anthropic stream envelope handling", () => { ]) as never, ); - const stream = streamAnthropic(model, context, { apiKey: "sk-ant-test" }); + const stream = trustedStreamAnthropic(model, context, { apiKey: "sk-ant-test" }); const events: AssistantMessageEvent[] = []; for await (const event of stream) { events.push(event); @@ -497,7 +511,7 @@ describe("anthropic stream envelope handling", () => { ]) as never, ); - const stream = streamAnthropic(model, context, { apiKey: "sk-ant-test" }); + const stream = trustedStreamAnthropic(model, context, { apiKey: "sk-ant-test" }); for await (const _ of stream) { // drain stream } @@ -548,7 +562,7 @@ describe("anthropic stream envelope handling", () => { ]) as never, ); - const stream = streamAnthropic(model, context, { apiKey: "sk-ant-test" }); + const stream = trustedStreamAnthropic(model, context, { apiKey: "sk-ant-test" }); for await (const _ of stream) { // drain stream } @@ -601,7 +615,7 @@ describe("anthropic stream envelope handling", () => { ]) as never, ); - const stream = streamAnthropic(model, context, { apiKey: "sk-ant-test" }); + const stream = trustedStreamAnthropic(model, context, { apiKey: "sk-ant-test" }); const events: AssistantMessageEvent[] = []; for await (const event of stream) { events.push(event); @@ -633,7 +647,7 @@ describe("anthropic stream envelope handling", () => { return createMockRequest(createTextSuccessEventsWithPreamble("hello", [{ type: "ping" }])) as never; }); - const stream = streamAnthropic(model, context, { apiKey: "sk-ant-test" }); + const stream = trustedStreamAnthropic(model, context, { apiKey: "sk-ant-test" }); const events: AssistantMessageEvent[] = []; for await (const event of stream) { events.push(event); @@ -660,7 +674,7 @@ describe("anthropic stream envelope handling", () => { ) as never; }); - const stream = streamAnthropic(model, context, { apiKey: "sk-ant-test" }); + const stream = trustedStreamAnthropic(model, context, { apiKey: "sk-ant-test" }); const events: AssistantMessageEvent[] = []; for await (const event of stream) { events.push(event); @@ -688,7 +702,7 @@ describe("anthropic stream envelope handling", () => { }); vi.spyOn(scheduler, "wait").mockResolvedValue(undefined); - const stream = streamAnthropic(model, context, { apiKey: "sk-ant-test" }); + const stream = trustedStreamAnthropic(model, context, { apiKey: "sk-ant-test" }); const events: AssistantMessageEvent[] = []; for await (const event of stream) { events.push(event); @@ -728,7 +742,7 @@ describe("anthropic stream envelope handling", () => { return createMockRequest(createTextSuccessEvents(attempt === 2 ? "recovered" : "later")) as never; }); - const stream = streamAnthropic(model, toolContext, { apiKey: "sk-ant-test", providerSessionState }); + const stream = trustedStreamAnthropic(model, toolContext, { apiKey: "sk-ant-test", providerSessionState }); const events: AssistantMessageEvent[] = []; for await (const event of stream) { events.push(event); @@ -746,7 +760,7 @@ describe("anthropic stream envelope handling", () => { ?.strictToolsDisabled, ).toBe(true); - const nextStream = streamAnthropic(model, toolContext, { apiKey: "sk-ant-test", providerSessionState }); + const nextStream = trustedStreamAnthropic(model, toolContext, { apiKey: "sk-ant-test", providerSessionState }); const nextEvents: AssistantMessageEvent[] = []; for await (const event of nextStream) { nextEvents.push(event); @@ -781,7 +795,7 @@ describe("anthropic stream envelope handling", () => { return createRejectedMockRequest(createOtherInvalidRequestError()) as never; }); - const stream = streamAnthropic(model, toolContext, { apiKey: "sk-ant-test", providerSessionState }); + const stream = trustedStreamAnthropic(model, toolContext, { apiKey: "sk-ant-test", providerSessionState }); const events: AssistantMessageEvent[] = []; for await (const event of stream) { events.push(event); @@ -845,7 +859,7 @@ describe("anthropic stream envelope handling", () => { return createMockRequest(createTextSuccessEvents("recovered")) as never; }); - const stream = streamAnthropic(gatewayModel, toolLoopContext, { + const stream = trustedStreamAnthropic(gatewayModel, toolLoopContext, { apiKey: "sk-ant-test", providerSessionState, }); @@ -870,7 +884,7 @@ describe("anthropic stream envelope handling", () => { // A later turn in the same session starts from the reduced budget instead of // re-triggering the rejection. - const nextStream = streamAnthropic(gatewayModel, toolLoopContext, { + const nextStream = trustedStreamAnthropic(gatewayModel, toolLoopContext, { apiKey: "sk-ant-test", providerSessionState, }); @@ -900,7 +914,7 @@ describe("anthropic stream envelope handling", () => { return createMockRequest(createTextSuccessEvents("recovered")) as never; }); - const stream = streamAnthropic(gatewayModel, context, { apiKey: "sk-ant-test", providerSessionState }); + const stream = trustedStreamAnthropic(gatewayModel, context, { apiKey: "sk-ant-test", providerSessionState }); const events: AssistantMessageEvent[] = []; for await (const event of stream) { events.push(event); @@ -936,7 +950,7 @@ describe("anthropic stream envelope handling", () => { return createMockRequest(createTextSuccessEvents("recovered")) as never; }); - const stream = streamAnthropic(gatewayModel, context, { apiKey: "sk-ant-test" }); + const stream = trustedStreamAnthropic(gatewayModel, context, { apiKey: "sk-ant-test" }); for await (const _ of stream) { // drain stream } @@ -963,7 +977,7 @@ describe("anthropic stream envelope handling", () => { return createRejectedMockRequest(createOtherInvalidRequestError()) as never; }); - const stream = streamAnthropic(gatewayModel, context, { apiKey: "sk-ant-test", providerSessionState }); + const stream = trustedStreamAnthropic(gatewayModel, context, { apiKey: "sk-ant-test", providerSessionState }); const events: AssistantMessageEvent[] = []; for await (const event of stream) { events.push(event); @@ -987,7 +1001,7 @@ describe("anthropic stream envelope handling", () => { return createMockRequest(createMalformedToolUseEvents()) as never; }); - const stream = streamAnthropic(model, context, { apiKey: "sk-ant-test" }); + const stream = trustedStreamAnthropic(model, context, { apiKey: "sk-ant-test" }); const events: AssistantMessageEvent[] = []; for await (const event of stream) { events.push(event); @@ -1020,7 +1034,7 @@ describe("anthropic stream envelope handling", () => { ) as never, ); - const stream = streamAnthropic(model, context, { apiKey: "sk-ant-test" }); + const stream = trustedStreamAnthropic(model, context, { apiKey: "sk-ant-test" }); const events: AssistantMessageEvent[] = []; for await (const event of stream) { events.push(event); @@ -1039,7 +1053,7 @@ describe("anthropic stream envelope handling", () => { ); vi.spyOn(Messages.prototype, "create").mockImplementation(() => createRawSseRequest(incompleteFrames) as never); - const stream = streamAnthropic(model, context, { apiKey: "sk-ant-test" }); + const stream = trustedStreamAnthropic(model, context, { apiKey: "sk-ant-test" }); const events: AssistantMessageEvent[] = []; for await (const event of stream) { events.push(event); @@ -1067,7 +1081,7 @@ describe("anthropic stream envelope handling", () => { ]; vi.spyOn(Messages.prototype, "create").mockImplementation(() => createRawSseRequest(frames) as never); - const stream = streamAnthropic(model, context, { apiKey: "sk-ant-test" }); + const stream = trustedStreamAnthropic(model, context, { apiKey: "sk-ant-test" }); for await (const _ of stream) { // drain stream } @@ -1099,7 +1113,7 @@ describe("anthropic stream envelope handling", () => { ]; vi.spyOn(Messages.prototype, "create").mockImplementation(() => createMockRequest(refusalEvents) as never); - const stream = streamAnthropic(model, context, { apiKey: "sk-ant-test" }); + const stream = trustedStreamAnthropic(model, context, { apiKey: "sk-ant-test" }); const events: AssistantMessageEvent[] = []; for await (const event of stream) { events.push(event); @@ -1144,7 +1158,7 @@ describe("anthropic stream envelope handling", () => { ]; vi.spyOn(Messages.prototype, "create").mockImplementation(() => createMockRequest(refusalEvents) as never); - const stream = streamAnthropic(model, context, { apiKey: "sk-ant-test" }); + const stream = trustedStreamAnthropic(model, context, { apiKey: "sk-ant-test" }); const events: AssistantMessageEvent[] = []; for await (const event of stream) { events.push(event); @@ -1181,7 +1195,7 @@ describe("anthropic stream envelope handling", () => { ]; vi.spyOn(Messages.prototype, "create").mockImplementation(() => createMockRequest(sensitiveEvents) as never); - const stream = streamAnthropic(model, context, { apiKey: "sk-ant-test" }); + const stream = trustedStreamAnthropic(model, context, { apiKey: "sk-ant-test" }); for await (const _ of stream) { // drain stream } @@ -1190,7 +1204,108 @@ describe("anthropic stream envelope handling", () => { expect(result.stopReason).toBe("error"); expect(result.errorMessage).toBe("Content flagged by safety filters"); expect(result.errorKind).toBe("provider_safety_stop"); + expect(isProviderSafetyStopAuthenticated(result)).toBe(true); }); + + it("keeps direct provider calls unauthenticated without dispatcher provenance", async () => { + const refusalEvents: MockAnthropicEvent[] = [ + { + type: "message_start", + message: { + id: "msg_direct_refusal", + usage: { + input_tokens: 5, + output_tokens: 0, + cache_read_input_tokens: 0, + cache_creation_input_tokens: 0, + }, + }, + }, + { + type: "message_delta", + delta: { + stop_reason: "end_turn", + stop_details: { type: "refusal", category: "safety", explanation: "Direct refusal" }, + }, + usage: { input_tokens: 5, output_tokens: 0 }, + }, + { type: "message_stop" }, + ]; + vi.spyOn(Messages.prototype, "create").mockImplementation(() => createMockRequest(refusalEvents) as never); + + const bundled = getBundledModel("anthropic", "claude-sonnet-4-5") as Model<"anthropic-messages"> | undefined; + if (!bundled) throw new Error("Expected bundled Anthropic model"); + const stream = streamAnthropicProvider(bundled, context, { apiKey: "sk-ant-test" }); + for await (const _ of stream) { + // drain stream + } + const result = await stream.result(); + + expect(result.errorKind).toBeUndefined(); + expect(isProviderSafetyStopAuthenticated(result)).toBe(false); + + const cloned = { ...bundled, baseUrl: "https://attacker.example/anthropic" }; + const clonedStream = streamAnthropicProvider(cloned, context, { apiKey: "sk-ant-test" }); + for await (const _ of clonedStream) { + // drain stream + } + const clonedResult = await clonedStream.result(); + expect(clonedResult.errorKind).toBeUndefined(); + expect(isProviderSafetyStopAuthenticated(clonedResult)).toBe(false); + + const callerTransportStream = streamAnthropicProvider(bundled, context, { + client: { messages: { create: () => createMockRequest(refusalEvents) } } as never, + }); + for await (const _ of callerTransportStream) { + // drain stream + } + const callerTransportResult = await callerTransportStream.result(); + expect(callerTransportResult.errorKind).toBeUndefined(); + expect(callerTransportResult.transportFailure).toMatchObject({ + kind: "transport", + status: 500, + providerCode: "untrusted_safety_stop", + }); + }); + + it("preserves adapter provenance through streamSimple option mapping", async () => { + const bundled = getBundledModel("anthropic", "claude-sonnet-4-5") as Model<"anthropic-messages"> | undefined; + if (!bundled) throw new Error("Expected bundled Anthropic model"); + const refusalEvents: MockAnthropicEvent[] = [ + { + type: "message_start", + message: { + id: "msg_simple_refusal", + usage: { + input_tokens: 5, + output_tokens: 0, + cache_read_input_tokens: 0, + cache_creation_input_tokens: 0, + }, + }, + }, + { + type: "message_delta", + delta: { + stop_reason: "end_turn", + stop_details: { type: "refusal", category: "safety", explanation: "Simple refusal" }, + }, + usage: { input_tokens: 5, output_tokens: 0 }, + }, + { type: "message_stop" }, + ]; + vi.spyOn(Messages.prototype, "create").mockImplementation(() => createMockRequest(refusalEvents) as never); + + const stream = streamSimple(bundled, context, { apiKey: "sk-ant-test" }); + for await (const _ of stream) { + // drain stream + } + const result = await stream.result(); + + expect(result.errorKind).toBe("provider_safety_stop"); + expect(isProviderSafetyStopAuthenticated(result)).toBe(true); + }); + it("keeps a safety stop terminal when later stop reasons and tool events arrive", async () => { const eventsAfterSafety: MockAnthropicEvent[] = [ { @@ -1227,7 +1342,7 @@ describe("anthropic stream envelope handling", () => { ]; vi.spyOn(Messages.prototype, "create").mockImplementation(() => createMockRequest(eventsAfterSafety) as never); - const stream = streamAnthropic(model, context, { apiKey: "sk-ant-test" }); + const stream = trustedStreamAnthropic(model, context, { apiKey: "sk-ant-test" }); const observedEvents: AssistantMessageEvent[] = []; for await (const event of stream) { observedEvents.push(event); @@ -1279,7 +1394,7 @@ describe("anthropic stream envelope handling", () => { }); vi.spyOn(scheduler, "wait").mockResolvedValue(undefined); - const stream = streamAnthropic(model, context, { apiKey: "sk-ant-test" }); + const stream = trustedStreamAnthropic(model, context, { apiKey: "sk-ant-test" }); const observedEvents: AssistantMessageEvent[] = []; for await (const event of stream) { observedEvents.push(event); @@ -1310,13 +1425,13 @@ describe("anthropic stream envelope handling", () => { return createMockRequest(createTextSuccessEvents("ok")) as never; }); - const eagerStream = streamAnthropic(model, toolContext, { apiKey: "sk-ant-test" }); + const eagerStream = trustedStreamAnthropic(model, toolContext, { apiKey: "sk-ant-test" }); for await (const _ of eagerStream) { // drain stream } await eagerStream.result(); - const disabledStream = streamAnthropic( + const disabledStream = trustedStreamAnthropic( { ...model, compat: { supportsEagerToolInputStreaming: false } }, toolContext, { apiKey: "sk-ant-test" }, @@ -1350,7 +1465,7 @@ describe("anthropic stream envelope handling", () => { baseUrl: "https://proxy.example.com/anthropic", }, ]) { - const stream = streamAnthropic(testModel, context, { + const stream = trustedStreamAnthropic(testModel, context, { apiKey: "sk-ant-test", cacheRetention: "long", }); @@ -1403,7 +1518,7 @@ describe("anthropic stream envelope handling", () => { }, ]) { // No cacheRetention passed: the provider default should drive the TTL. - const stream = streamAnthropic(testModel, context, { apiKey: "sk-ant-test" }); + const stream = trustedStreamAnthropic(testModel, context, { apiKey: "sk-ant-test" }); for await (const _ of stream) { // drain stream } @@ -1468,7 +1583,7 @@ describe("anthropic stream envelope handling", () => { ]) as never, ); - const stream = streamAnthropic(thinkingModel, context, { apiKey: "sk-ant-test", thinkingEnabled: true }); + const stream = trustedStreamAnthropic(thinkingModel, context, { apiKey: "sk-ant-test", thinkingEnabled: true }); const events: AssistantMessageEvent[] = []; for await (const event of stream) events.push(event); const result = await stream.result(); diff --git a/packages/ai/test/google-gemini-cli-safety-stop.test.ts b/packages/ai/test/google-gemini-cli-safety-stop.test.ts index ea7555f52c..a1529d7717 100644 --- a/packages/ai/test/google-gemini-cli-safety-stop.test.ts +++ b/packages/ai/test/google-gemini-cli-safety-stop.test.ts @@ -1,4 +1,5 @@ import { describe, expect, it } from "bun:test"; +import { withProviderSafetyStopAdapterInvocation } from "../src/adapter-internals/provider-safety-stop"; import { streamGoogleGeminiCli } from "../src/providers/google-gemini-cli"; import type { Context, Model } from "../src/types"; import { collectEvents, createSseResponse } from "./openai-tool-choice-test-helpers"; @@ -35,16 +36,22 @@ function createSseResponseWithUrl(chunks: unknown[]): Response { async function streamResponse(provider: GeminiCliProvider, chunks: unknown[]) { let requestCount = 0; - const stream = streamGoogleGeminiCli(createModel(provider), context, { - apiKey: JSON.stringify({ token: "token", projectId: "project" }), - fetch: async () => { - requestCount += 1; - return createSseResponseWithUrl(chunks); - }, - }); - - const events = await collectEvents(stream); - return { events, requestCount, result: await stream.result() }; + const originalFetch = globalThis.fetch; + globalThis.fetch = (async () => { + requestCount += 1; + return createSseResponseWithUrl(chunks); + }) as unknown as typeof fetch; + try { + const stream = streamGoogleGeminiCli(createModel(provider), context, { + ...withProviderSafetyStopAdapterInvocation({ + apiKey: JSON.stringify({ token: "token", projectId: "project" }), + }), + }); + const events = await collectEvents(stream); + return { events, requestCount, result: await stream.result() }; + } finally { + globalThis.fetch = originalFetch; + } } describe("Google Gemini CLI safety stops", () => { diff --git a/packages/ai/test/google-safety-stop.test.ts b/packages/ai/test/google-safety-stop.test.ts index fbe7c736dd..ae95420360 100644 --- a/packages/ai/test/google-safety-stop.test.ts +++ b/packages/ai/test/google-safety-stop.test.ts @@ -1,4 +1,5 @@ import { describe, expect, it } from "bun:test"; +import { withProviderSafetyStopAdapterInvocation } from "../src/adapter-internals/provider-safety-stop"; import { streamGoogleGenAI } from "../src/providers/google-shared"; import { collectEvents, createBaseModel, createSseResponse } from "./openai-tool-choice-test-helpers"; @@ -78,23 +79,91 @@ const promptBlockReasonFixtures = { async function streamGoogleResponse(response: unknown | unknown[], api: GoogleStreamApi = "google-generative-ai") { const model = createBaseModel(api); - const stream = streamGoogleGenAI({ - model, - api, - options: undefined, - prepare: () => ({ - params: { model: model.id, contents: [] }, - url: "https://google.example.test/stream", - headers: {}, - fetch: async () => createSseResponse(Array.isArray(response) ? response : [response]), - }), - }); + const originalFetch = globalThis.fetch; + globalThis.fetch = (async () => + createSseResponse(Array.isArray(response) ? response : [response])) as unknown as typeof fetch; + try { + const stream = streamGoogleGenAI({ + model, + api, + options: withProviderSafetyStopAdapterInvocation({}), + prepare: () => ({ + params: { model: model.id, contents: [] }, + url: "https://google.example.test/stream", + headers: {}, + }), + }); - await collectEvents(stream); - return stream.result(); + await collectEvents(stream); + return stream.result(); + } finally { + globalThis.fetch = originalFetch; + } } describe("Google safety stops", () => { + it("does not authenticate a safety chunk from caller-selected prepare fetch", async () => { + const model = createBaseModel("google-generative-ai"); + const stream = streamGoogleGenAI({ + model, + api: "google-generative-ai", + options: undefined, + prepare: () => ({ + params: { model: model.id, contents: [] }, + url: "https://google.example.test/stream", + headers: {}, + fetch: async () => createSseResponse([{ candidates: [{ finishReason: "SAFETY" }] }]), + }), + }); + + await collectEvents(stream); + const result = await stream.result(); + expect(result.stopReason).toBe("error"); + expect(result.errorKind).toBeUndefined(); + expect(result.transportFailure).toMatchObject({ + kind: "transport", + status: 500, + providerCode: "untrusted_safety_stop", + }); + }); + + it("keeps an unauthenticated safety refusal terminal across a later benign finish", async () => { + const model = createBaseModel("google-generative-ai"); + const stream = streamGoogleGenAI({ + model, + api: "google-generative-ai", + options: undefined, + prepare: () => ({ + params: { model: model.id, contents: [] }, + url: "https://google.example.test/stream", + headers: {}, + fetch: async () => + createSseResponse([ + { + candidates: [ + { + content: { parts: [{ functionCall: { name: "blocked", args: {} } }] }, + finishReason: "SAFETY", + }, + ], + }, + { candidates: [{ finishReason: "STOP" }] }, + ]), + }), + }); + + await collectEvents(stream); + const result = await stream.result(); + expect(result.stopReason).toBe("error"); + expect(result.errorKind).toBeUndefined(); + expect(result.content.some(block => block.type === "toolCall")).toBe(true); + expect(result.transportFailure).toMatchObject({ + kind: "transport", + status: 500, + providerCode: "untrusted_safety_stop", + }); + }); + it("classifies the exhaustive candidate finish-reason partition", async () => { for (const finishReason of candidateFinishReasonFixtures.commonSafety) { const result = await streamGoogleResponse({ candidates: [{ finishReason }] }); diff --git a/packages/ai/test/openai-completions-safety-stop.test.ts b/packages/ai/test/openai-completions-safety-stop.test.ts index 259d480355..6a59c677b5 100644 --- a/packages/ai/test/openai-completions-safety-stop.test.ts +++ b/packages/ai/test/openai-completions-safety-stop.test.ts @@ -1,6 +1,13 @@ import { afterEach, describe, expect, it } from "bun:test"; import { streamOpenAICompletions } from "@gajae-code/ai/providers/openai-completions"; import type { AssistantMessageEvent, Context, Model } from "@gajae-code/ai/types"; +import { withProviderSafetyStopAdapterInvocation } from "../src/adapter-internals/provider-safety-stop"; +import { getBundledModel } from "../src/models"; +import { isProviderSafetyStopAuthenticated } from "../src/utils/provider-safety-stop"; + +function trustedOptions(): { apiKey: string } { + return withProviderSafetyStopAdapterInvocation({ apiKey: "test" }); +} const originalFetch = global.fetch; afterEach(() => { @@ -74,6 +81,16 @@ function context(): Context { } describe("chat-completions: provider safety stops", () => { + it("keeps direct calls unauthenticated without dispatcher provenance", async () => { + const bundled = getBundledModel("openai", "gpt-4o-mini") as Model<"openai-completions"> | undefined; + if (!bundled) throw new Error("Expected bundled OpenAI model"); + global.fetch = mockFetch([chunk({}, "content_filter"), "[DONE]"]); + + const result = await streamOpenAICompletions(bundled, context(), { apiKey: "test" }).result(); + expect(result.errorKind).toBeUndefined(); + expect(isProviderSafetyStopAuthenticated(result)).toBe(false); + }); + it("keeps a content-filter safety stop when a later tool block finishes", async () => { global.fetch = mockFetch([ chunk({}, "content_filter"), @@ -93,7 +110,7 @@ describe("chat-completions: provider safety stops", () => { "[DONE]", ]); - const result = await streamOpenAICompletions(model(), context(), { apiKey: "test" }).result(); + const result = await streamOpenAICompletions(model(), context(), trustedOptions()).result(); expect(result.errorKind).toBe("provider_safety_stop"); expect(result.stopReason).toBe("error"); expect(result.errorMessage).toBe("Provider finish_reason: content_filter"); @@ -102,7 +119,7 @@ describe("chat-completions: provider safety stops", () => { it("classifies a streamed refusal as a safety stop despite an ordinary finish reason", async () => { global.fetch = mockFetch([chunk({ refusal: "I cannot help with that." }, "stop"), "[DONE]"]); - const result = await streamOpenAICompletions(model(), context(), { apiKey: "test" }).result(); + const result = await streamOpenAICompletions(model(), context(), trustedOptions()).result(); expect(result.errorKind).toBe("provider_safety_stop"); expect(result.stopReason).toBe("error"); expect(result.content).toEqual([{ type: "text", text: "I cannot help with that." }]); @@ -127,7 +144,7 @@ describe("chat-completions: provider safety stops", () => { "[DONE]", ]); - const result = await streamOpenAICompletions(model(), context(), { apiKey: "test" }).result(); + const result = await streamOpenAICompletions(model(), context(), trustedOptions()).result(); expect(result.content.some(block => block.type === "toolCall")).toBe(true); expect(result.errorKind).toBe("provider_safety_stop"); expect(result.stopReason).toBe("error"); @@ -139,7 +156,7 @@ describe("chat-completions: provider safety stops", () => { "[DONE]", ]); - const result = await streamOpenAICompletions(model(), context(), { apiKey: "test" }).result(); + const result = await streamOpenAICompletions(model(), context(), trustedOptions()).result(); expect(result.errorKind).toBe("provider_safety_stop"); expect(result.stopReason).toBe("error"); expect(result.content).toEqual([{ type: "text", text: "I cannot help with that. Here is ordinary content." }]); @@ -148,7 +165,7 @@ describe("chat-completions: provider safety stops", () => { it("keeps the content-filter error after an earlier refusal", async () => { global.fetch = mockFetch([chunk({ refusal: "I cannot help with that." }), chunk({}, "content_filter"), "[DONE]"]); - const result = await streamOpenAICompletions(model(), context(), { apiKey: "test" }).result(); + const result = await streamOpenAICompletions(model(), context(), trustedOptions()).result(); expect(result.errorKind).toBe("provider_safety_stop"); expect(result.stopReason).toBe("error"); expect(result.errorMessage).toBe("Provider finish_reason: content_filter"); @@ -159,7 +176,7 @@ describe("chat-completions: provider safety stops", () => { error: { code: "content_filter", message: "Prompt rejected by policy" }, }); - const stream = streamOpenAICompletions(model(), context(), { apiKey: "test" }); + const stream = streamOpenAICompletions(model(), context(), trustedOptions()); const events: AssistantMessageEvent[] = []; for await (const event of stream) events.push(event); const result = await stream.result(); @@ -180,7 +197,7 @@ describe("chat-completions: provider safety stops", () => { }, }); - const stream = streamOpenAICompletions(model(), context(), { apiKey: "test" }); + const stream = streamOpenAICompletions(model(), context(), trustedOptions()); const events: AssistantMessageEvent[] = []; for await (const event of stream) events.push(event); const result = await stream.result(); diff --git a/packages/ai/test/pi-native-client.test.ts b/packages/ai/test/pi-native-client.test.ts index 615f723192..2435c5bec0 100644 --- a/packages/ai/test/pi-native-client.test.ts +++ b/packages/ai/test/pi-native-client.test.ts @@ -1,6 +1,8 @@ import { afterEach, describe, expect, it, mock, spyOn } from "bun:test"; import { streamPiNative } from "../src/providers/pi-native-client"; +import { streamSimple } from "../src/stream"; import type { AssistantMessage, AssistantMessageEvent, Context, FetchImpl, Model } from "../src/types"; +import { isProviderSafetyStopAuthenticated } from "../src/utils/provider-safety-stop"; function sseBytes(events: AssistantMessageEvent[]): Uint8Array { const encoder = new TextEncoder(); @@ -298,3 +300,59 @@ describe("streamPiNative event flow", () => { expect(captured.signal).toBe(controller.signal); }); }); + +describe("streamPiNative provider safety-stop provenance", () => { + const typedStop = (): AssistantMessage => + baseAssistant({ + stopReason: "error", + errorKind: "provider_safety_stop", + errorMessage: "Refusal (safety): policy violation", + }); + + it("does not mint authority from a caller-supplied fetch and loopback URL", async () => { + const fetchImpl: FetchImpl = (async () => + fakeResponse([{ type: "error", reason: "error", error: typedStop() }])) as FetchImpl; + const model = fakeModel({ baseUrl: "http://127.0.0.1:4000" }); + + const result = await streamPiNative(model, baseContext, { apiKey: "k", fetch: fetchImpl }).result(); + expect(result.errorKind).toBe("provider_safety_stop"); + expect(isProviderSafetyStopAuthenticated(result)).toBe(false); + }); + + it("keeps the public streamSimple pi-native path fail-closed", async () => { + const fetchImpl: FetchImpl = (async () => + fakeResponse([{ type: "done", reason: "stop", message: typedStop() }])) as FetchImpl; + + const result = await streamSimple(fakeModel({ baseUrl: "http://127.0.0.1:4000" }), baseContext, { + apiKey: "k", + fetch: fetchImpl, + }).result(); + + expect(result.errorKind).toBe("provider_safety_stop"); + expect(isProviderSafetyStopAuthenticated(result)).toBe(false); + }); + + it("does not authenticate a typed stop from any serialized gateway endpoint", async () => { + const fetchImpl: FetchImpl = (async () => + fakeResponse([{ type: "error", reason: "error", error: typedStop() }])) as FetchImpl; + + const result = await streamPiNative(fakeModel(), baseContext, { apiKey: "k", fetch: fetchImpl }).result(); + expect(result.errorKind).toBe("provider_safety_stop"); + expect(isProviderSafetyStopAuthenticated(result)).toBe(false); + }); + + it("never authenticates a done-carried typed stop after SSE serialization", async () => { + const fetchImpl: FetchImpl = (async () => + fakeResponse([{ type: "done", reason: "stop", message: typedStop() }])) as FetchImpl; + const model = fakeModel({ baseUrl: "http://localhost:4000" }); + + const result = await streamPiNative(model, baseContext, { apiKey: "k", fetch: fetchImpl }).result(); + expect(isProviderSafetyStopAuthenticated(result)).toBe(false); + + const plainFetch: FetchImpl = (async () => + fakeResponse([{ type: "done", reason: "stop", message: baseAssistant() }])) as FetchImpl; + const plain = await streamPiNative(model, baseContext, { apiKey: "k", fetch: plainFetch }).result(); + expect(plain.errorKind).toBeUndefined(); + expect(isProviderSafetyStopAuthenticated(plain)).toBe(false); + }); +}); diff --git a/packages/ai/test/provider-safety-stop.test.ts b/packages/ai/test/provider-safety-stop.test.ts new file mode 100644 index 0000000000..31281f3791 --- /dev/null +++ b/packages/ai/test/provider-safety-stop.test.ts @@ -0,0 +1,255 @@ +import { describe, expect, test, vi } from "bun:test"; +import { + mintProviderSafetyStop, + PROVIDER_SAFETY_STOP_ADAPTER_CAPABILITY, + PROVIDER_SAFETY_STOP_ADAPTER_INVOCATION, +} from "../src/adapter-internals/provider-safety-stop"; +import * as publicAi from "../src/index"; +import { getBundledModel } from "../src/models"; +import { streamOpenAICompletions } from "../src/providers/openai-completions"; +import { stream, streamSimple } from "../src/stream"; +import type { AssistantMessage, Context, FetchImpl, Model } from "../src/types"; +import { isProviderSafetyStopAuthenticated } from "../src/utils/provider-safety-stop"; + +function message(): AssistantMessage { + return { + role: "assistant", + content: [], + api: "anthropic-messages", + provider: "anthropic", + model: "claude-sonnet-4-5", + usage: { + input: 0, + output: 0, + cacheRead: 0, + cacheWrite: 0, + totalTokens: 0, + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, + }, + stopReason: "error", + errorMessage: "Refusal (safety): Policy violation", + timestamp: 1, + }; +} + +describe("provider safety-stop provenance authority", () => { + test("does not mint from a public OpenAI adapter with caller-supplied fetch", async () => { + const model = getBundledModel("openai", "gpt-4o-mini") as Model<"openai-completions">; + const context: Context = { + messages: [{ role: "user", content: "hello", timestamp: 0 }], + }; + const fetchImpl: FetchImpl = (async () => + new Response( + JSON.stringify({ + error: { message: "filtered", type: "invalid_request_error", code: "content_filter" }, + }), + { status: 429, headers: { "Content-Type": "application/json" } }, + )) as FetchImpl; + + const result = await streamOpenAICompletions(model, context, { + apiKey: "caller-key", + fetch: fetchImpl, + requestMaxRetries: 0, + streamMaxRetries: 0, + }).result(); + + expect(result.stopReason).toBe("error"); + expect(result.errorKind).toBeUndefined(); + expect(isProviderSafetyStopAuthenticated(result)).toBe(false); + }); + + test("does not mint from a public stream when a caller redirects a cloned model", async () => { + const bundled = getBundledModel("openai", "gpt-4o-mini") as Model<"openai-completions">; + const model = { ...bundled, baseUrl: "https://attacker.example/v1" }; + const fetchSpy = vi.spyOn(globalThis, "fetch").mockResolvedValue( + new Response( + JSON.stringify({ + error: { message: "filtered", type: "invalid_request_error", code: "content_filter" }, + }), + { status: 429, headers: { "Content-Type": "application/json" } }, + ), + ); + try { + const result = await stream( + model, + { messages: [{ role: "user", content: "hello", timestamp: 0 }] }, + { apiKey: "caller-key", requestMaxRetries: 0, streamMaxRetries: 0 }, + ).result(); + + expect(fetchSpy).toHaveBeenCalled(); + expect(result.stopReason).toBe("error"); + expect(result.errorKind).toBeUndefined(); + expect(isProviderSafetyStopAuthenticated(result)).toBe(false); + } finally { + fetchSpy.mockRestore(); + } + }); + + test("preserves authenticated safety stops through the Synthetic wrapper", async () => { + const model = getBundledModel("synthetic", "hf:deepseek-ai/DeepSeek-R1-0528"); + if (!model) throw new Error("Expected bundled Synthetic model"); + const fetchSpy = vi.spyOn(globalThis, "fetch").mockResolvedValue( + new Response( + JSON.stringify({ + error: { message: "filtered", type: "invalid_request_error", code: "content_filter" }, + }), + { status: 429, headers: { "Content-Type": "application/json" } }, + ), + ); + try { + const result = await streamSimple( + model, + { messages: [{ role: "user", content: "hello", timestamp: 0 }] }, + { apiKey: "synthetic-key", requestMaxRetries: 0, streamMaxRetries: 0 }, + ).result(); + + expect(fetchSpy).toHaveBeenCalled(); + expect(result.stopReason).toBe("error"); + expect(result.errorKind).toBe("provider_safety_stop"); + expect(isProviderSafetyStopAuthenticated(result)).toBe(true); + } finally { + fetchSpy.mockRestore(); + } + }); + + test("public AI exports expose verification only, never the minting operation", () => { + const publicSurface = publicAi as unknown as Record; + expect(publicSurface.applyProviderSafetyStop).toBeUndefined(); + expect(typeof publicSurface.isProviderSafetyStopAuthenticated).toBe("function"); + expect(publicSurface.revokeProviderSafetyStop).toBeUndefined(); + expect(publicSurface.transferProviderSafetyStop).toBeUndefined(); + }); + + test("mints the typed kind only for structured first-party refusal signals", () => { + for (const signal of ["refusal", "sensitive", "content_filter", "SAFETY", "JAILBREAK", "RECITATION"]) { + const marked = message(); + expect( + mintProviderSafetyStop( + marked, + signal, + PROVIDER_SAFETY_STOP_ADAPTER_CAPABILITY, + undefined, + PROVIDER_SAFETY_STOP_ADAPTER_INVOCATION, + ), + ).toBe(true); + expect(marked.errorKind).toBe("provider_safety_stop"); + expect(isProviderSafetyStopAuthenticated(marked)).toBe(true); + } + }); + + test("fails closed on an unrecognized signal: no kind, no authority", () => { + const unmarked = message(); + unmarked.errorKind = "provider_safety_stop"; + expect( + mintProviderSafetyStop( + unmarked, + "totally-not-a-refusal", + PROVIDER_SAFETY_STOP_ADAPTER_CAPABILITY, + undefined, + PROVIDER_SAFETY_STOP_ADAPTER_INVOCATION, + ), + ).toBe(false); + // The pre-existing wire-assignable field stays exactly as unauthenticated + // as it was; the adapter bug degraded to ordinary fallback, not a mint. + expect(isProviderSafetyStopAuthenticated(unmarked)).toBe(false); + }); + + test("fails closed when a caller controls the adapter transport seam", () => { + for (const callerTransport of [() => undefined, {}]) { + const forged = message(); + expect( + mintProviderSafetyStop( + forged, + "refusal", + PROVIDER_SAFETY_STOP_ADAPTER_CAPABILITY, + callerTransport, + PROVIDER_SAFETY_STOP_ADAPTER_INVOCATION, + ), + ).toBe(false); + expect(forged.errorKind).toBeUndefined(); + expect(isProviderSafetyStopAuthenticated(forged)).toBe(false); + } + }); + + test("requires the runtime-owned adapter invocation token", () => { + const untrusted = message(); + expect(mintProviderSafetyStop(untrusted, "refusal", PROVIDER_SAFETY_STOP_ADAPTER_CAPABILITY)).toBe(false); + expect(untrusted.errorKind).toBeUndefined(); + expect(isProviderSafetyStopAuthenticated(untrusted)).toBe(false); + }); + + test("a structurally forged capability cannot mint authority", () => { + const forged = message(); + const forgedCapability = {} as Parameters[2]; + expect( + mintProviderSafetyStop( + forged, + "refusal", + forgedCapability, + undefined, + PROVIDER_SAFETY_STOP_ADAPTER_INVOCATION, + ), + ).toBe(false); + expect(isProviderSafetyStopAuthenticated(forged)).toBe(false); + expect(forged.errorKind).toBeUndefined(); + }); + + test("a public consumer cannot clone authority from a genuine marked source", () => { + const marked = message(); + expect( + mintProviderSafetyStop( + marked, + "refusal", + PROVIDER_SAFETY_STOP_ADAPTER_CAPABILITY, + undefined, + PROVIDER_SAFETY_STOP_ADAPTER_INVOCATION, + ), + ).toBe(true); + + const forgedDestination = { ...marked }; + expect(isProviderSafetyStopAuthenticated(marked)).toBe(true); + expect(isProviderSafetyStopAuthenticated(forgedDestination)).toBe(false); + expect((publicAi as unknown as Record).transferProviderSafetyStop).toBeUndefined(); + }); + + test("data alone is never authenticated: clones, JSON round-trips, and fresh copies lose authority", () => { + const marked = message(); + expect( + mintProviderSafetyStop( + marked, + "refusal", + PROVIDER_SAFETY_STOP_ADAPTER_CAPABILITY, + undefined, + PROVIDER_SAFETY_STOP_ADAPTER_INVOCATION, + ), + ).toBe(true); + + const cloned = structuredClone(marked); + expect(cloned.errorKind).toBe("provider_safety_stop"); + expect(isProviderSafetyStopAuthenticated(cloned)).toBe(false); + + const persisted = JSON.parse(JSON.stringify(marked)) as AssistantMessage; + expect(persisted.errorKind).toBe("provider_safety_stop"); + expect(isProviderSafetyStopAuthenticated(persisted)).toBe(false); + + const fresh = message(); + fresh.errorKind = "provider_safety_stop"; + expect(isProviderSafetyStopAuthenticated(fresh)).toBe(false); + expect(isProviderSafetyStopAuthenticated(undefined)).toBe(false); + expect(isProviderSafetyStopAuthenticated("provider_safety_stop")).toBe(false); + }); + + test("the mint module is unreachable through the package export map", async () => { + // Deep imports through the public package name resolve through the + // exports map; `./adapter-internals/*` is null there, so neither the + // mint nor the capability is importable outside first-party relative + // imports (#4777 review follow-up). + // Non-literal specifier so typecheck cannot resolve the blocked subpath; + // the point is runtime resolution failing closed. + const deepImport = "@gajae-code/ai/adapter-internals/provider-safety-stop"; + await expect(import(deepImport)).rejects.toThrow(); + const manifest = (await import("../package.json", { with: { type: "json" } })).default; + expect(manifest.exports["./adapter-internals/*"]).toBeNull(); + expect(manifest.exports["./adapter-internals/*.js"]).toBeNull(); + }); +}); diff --git a/packages/coding-agent/CHANGELOG.md b/packages/coding-agent/CHANGELOG.md index cd6352f8cb..ed207ffd70 100644 --- a/packages/coding-agent/CHANGELOG.md +++ b/packages/coding-agent/CHANGELOG.md @@ -37,6 +37,8 @@ - Queued SDK prompts now retain their dispatch-time ownership across selection fences instead of reclassifying from a contradictory later streaming snapshot. Fresh promotion, earlier follow-up ordering, and terminal-abort cancellation are reachable again, while `/btw` test fixtures now model the user-drainable queue count required by the current empty-submit contract. - Coordinator stop and idle-reap now initialize canonical namespace state before reading durable deletion recovery. Fresh or upgraded projection-only sessions were misreported as `state_corrupt` before the broker close was attempted, and completed deletion receipts were excluded from the idempotent missing-session lookup; both paths now preserve strict malformed-state rejection while allowing safe cleanup and replay. - Fixed provider safety-stop classification being lost before session persistence (#4777). The managed provider-envelope boundary now preserves only the allowlisted `provider_safety_stop` kind, typed safety stops remain terminal even with transport facts on a multi-model fallback chain, and the regression e2e test is selected both by the focused affected-path route and exactly one normal coding-agent shard. +- A forged `errorKind: "provider_safety_stop"` label can no longer suppress model fallback (#4777 review). Terminal safety-stop authority is now adapter-minted (module-scoped provenance set by the first-party anthropic/openai/google adapters when they parse a structured refusal) rather than carried on message data: the loop strips unauthenticated labels at the stream exit, the managed discard gate re-verifies identity, and transport facts never upgrade provenance, so a compromised provider that names the typed kind on an error envelope degrades to an ordinary fallback-eligible failure on every status (400 and 429 both keep advancing the chain). Authenticated envelopes remain single-dispatch terminal with the manual-switch hint intact, and persisted/reloaded sessions retain the legacy display classification without retaining live terminal authority, so restart/replay stays fail-closed. +- The provider safety-stop minting path is now package-private and capability-branded; public AI imports and structural custom-stream payloads cannot forge terminal authority. Trailing stream completion now sanitizes the final message before managed-shell rebuilding when no `done` or `error` event was emitted, preserving fallback for forged labels (#4777 review). - Retained-publication acquisition diagnostics now name the stage that actually refused instead of always claiming a failed open, and the native errno vocabulary is closed over the actionable `open(2)` refusal set (#4764). `nativeRetainedObstruction()` rendered every unmatched reason as "could not be opened", so a `broker.json` FIFO — which opens fine `O_RDWR | O_NOFOLLOW | O_NONBLOCK` and then refuses `read_to_end` with `WouldBlock` — was reported as an open failure that had in fact succeeded, `clone`/`metadata` refusals of already-open descriptors were reported as opens, and `unsupported-platform` read as an open error. The structured native reason now drives the prose: `errno-`/`io-` render as open refusals, `read-` as a read failure, `clone-`/`metadata[-kind]` as inspection failures of an already-open object, `unsupported-platform` as unimplemented authority, and any future reason falls back to a stable "withheld publication authority (reason)" sentence instead of misnaming the stage; the structured reason is preferred whenever the native message parses, so the observed-state fallback is never entered for a stage-named refusal. The Rust `errno_name()` vocabulary adds `EAGAIN`, `EMFILE`, `ENAMETOOLONG`, `ENFILE`, `ENOMEM`, `EOVERFLOW`, and `EPERM` alongside the existing six — descriptor exhaustion and LSM/seccomp denial no longer collapse to `UNKNOWN` — while every unlisted errno still renders as `UNKNOWN` and no errno number or OS string ever leaves the native layer. - `move_session` no longer lets a cwd transition commit inside an in-flight tool's execution (#4629). The previous fence re-checked the cwd generation and then yielded, so a move that started inside a tool's first `await` retargeted work already admitted against the launcher root: bash computed `commandCwd` from the live session cwd only after `expandInternalUrls`, running a root-A command in root B. Relative-path tools now hold a shared read lease on the session cwd for their whole execution (`SessionManager.runWithCwdReadLease`), and writers drain outstanding leases before committing, so the cwd observed at admission stays authoritative until the tool finishes. Announced writers also block newly arriving readers, so a stream of tool calls cannot starve a queued move, and async bash jobs stay bound to the cwd they were admitted for. - `move_session` re-roots the state the model is shown at the new cwd (#4629). The volatile per-turn message reported the post-move cwd while attaching the launch-bound workspace tree, and subagents launched at the live cwd inherited startup `contextFiles`, `skills`, and `workspaceTree` — pairing the new cwd with the retired root's files and project instructions, which drives wrong path selection rather than merely stale display. Project context files and skills are now re-discovered after a committed move, the workspace-tree service resolves the live cwd per scan instead of capturing the launch root, the cached tree is retired so the next turn re-scans, and the system prompt is rebuilt from the live cwd. diff --git a/packages/coding-agent/test/agent-session-resilient-retry.test.ts b/packages/coding-agent/test/agent-session-resilient-retry.test.ts index 29d6bf5832..744626e11d 100644 --- a/packages/coding-agent/test/agent-session-resilient-retry.test.ts +++ b/packages/coding-agent/test/agent-session-resilient-retry.test.ts @@ -15,6 +15,11 @@ import { AuthStorage } from "@gajae-code/coding-agent/session/auth-storage"; import { SessionManager } from "@gajae-code/coding-agent/session/session-manager"; import { TempDir } from "@gajae-code/utils"; import * as z from "zod/v4"; +import { + mintProviderSafetyStop, + PROVIDER_SAFETY_STOP_ADAPTER_CAPABILITY, + PROVIDER_SAFETY_STOP_ADAPTER_INVOCATION, +} from "../../ai/src/adapter-internals/provider-safety-stop"; /** * Anthropic's statusless capacity-overload envelope exactly as observed in a @@ -189,6 +194,19 @@ describe("AgentSession resilient retry", () => { : { transportFailure: callFailure?.transportFailure ?? options.transportFailure }), timestamp: Date.now(), }; + // The typed safety stop is adapter-minted: this helper simulates a + // first-party provider envelope, so the structured refusal signal + // carries the terminal authority rather than the wire field + // alone (#4777). + if (options.errorKind === "provider_safety_stop") { + mintProviderSafetyStop( + message, + "refusal", + PROVIDER_SAFETY_STOP_ADAPTER_CAPABILITY, + undefined, + PROVIDER_SAFETY_STOP_ADAPTER_INVOCATION, + ); + } stream.push({ type: "start", partial: message }); stream.push({ type: "error", reason: "error", error: message }); }); diff --git a/packages/coding-agent/test/provider-safety-stop-hint.e2e.test.ts b/packages/coding-agent/test/provider-safety-stop-hint.e2e.test.ts index c0b4522ea4..7fc55712d2 100644 --- a/packages/coding-agent/test/provider-safety-stop-hint.e2e.test.ts +++ b/packages/coding-agent/test/provider-safety-stop-hint.e2e.test.ts @@ -22,6 +22,11 @@ import { resolveProviderSafetyStopHint, } from "@gajae-code/coding-agent/session/provider-safety-stop-hint"; import { TempDir } from "@gajae-code/utils"; +import { + mintProviderSafetyStop, + PROVIDER_SAFETY_STOP_ADAPTER_CAPABILITY, + PROVIDER_SAFETY_STOP_ADAPTER_INVOCATION, +} from "../../ai/src/adapter-internals/provider-safety-stop"; import { AgentSession, type AgentSessionEvent } from "../src/session/agent-session"; import { AuthStorage } from "../src/session/auth-storage"; import { SessionManager } from "../src/session/session-manager"; @@ -32,6 +37,7 @@ function safetyStopStream( model: Model, refusal: string, transportFacts?: { status: number }, + options?: { authenticated?: boolean }, ): AssistantMessageEventStream { const stream = new AssistantMessageEventStream(); queueMicrotask(() => { @@ -52,7 +58,6 @@ function safetyStopStream( cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, }, stopReason: "error", - errorKind: "provider_safety_stop", errorMessage: refusal, ...(transportFacts ? { @@ -62,6 +67,19 @@ function safetyStopStream( : {}), timestamp: Date.now(), }; + // Simulate the first-party adapter envelope: a structured refusal signal + // parsed from the provider's own response mints the terminal authority. + // Omitting the mark simulates a wire/custom-stream payload that only + // carries the forged field. + if (options?.authenticated !== false) { + mintProviderSafetyStop( + message, + "refusal", + PROVIDER_SAFETY_STOP_ADAPTER_CAPABILITY, + undefined, + PROVIDER_SAFETY_STOP_ADAPTER_INVOCATION, + ); + } else message.errorKind = "provider_safety_stop"; stream.push({ type: "start", partial: message }); stream.push({ type: "error", reason: "error", error: message }); }); @@ -215,6 +233,55 @@ describe("provider safety stop hint e2e (#4650)", () => { session = undefined; } }); + it("does not let a forged safety-stop field suppress fallback (#4777 review)", async () => { + // A wire/custom-stream payload that self-labels the typed kind without + // adapter-minted provenance must never terminalize: unauthenticated + // provider metadata stays fallback-eligible so a compromised provider + // cannot force refusal by naming the field. Transport facts do not + // upgrade provenance — retryable and non-retryable statuses both keep + // the chain advancing. + for (const status of [400, 429] as const) { + const primary = getBundledModel("anthropic", "claude-sonnet-4-5"); + const alternate = getBundledModel("openai", "gpt-4o-mini"); + if (!primary || !alternate) throw new Error("Expected bundled test models"); + const calls: string[] = []; + const agent = new Agent({ + getApiKey: provider => `${provider}-test-key`, + initialState: { model: primary, systemPrompt: ["Test"], tools: [], messages: [] }, + streamFn: ((model, _context, _options) => { + calls.push(selector(model)); + return safetyStopStream( + model, + "Upstream returned an unclassified error envelope", + { status }, + { + authenticated: false, + }, + ); + }) satisfies AgentOptions["streamFn"], + }); + const settings = Settings.isolated({ "compaction.enabled": false, "retry.baseDelayMs": 1 }); + settings.set("modelRoles", { default: [selector(primary), selector(alternate)] }); + session = new AgentSession({ + agent, + sessionManager: SessionManager.inMemory(), + settings, + modelRegistry: new ModelRegistry(authStorage), + }); + + await session.prompt("trigger forged safety-stop label"); + await session.waitForIdle(); + + // The forged label never terminalized: the chain advanced past the + // primary to the configured alternate. + expect(calls).toContain(selector(alternate)); + const last = [...session.state.messages].reverse().find(message => message.role === "assistant"); + expect((last as AssistantMessage).errorKind).toBeUndefined(); + expect(resolveProviderSafetyStopHint(last as AssistantMessage, session)).toBeUndefined(); + await session.dispose(); + session = undefined; + } + }); it("falls back to bounded static guidance when no alternate is configured", async () => { const primary = getBundledModel("anthropic", "claude-sonnet-4-5");