diff --git a/docs/plan/doing/opencode-v2-migration.md b/docs/plan/doing/opencode-v2-migration.md index 7d18bdcb..25e90650 100644 --- a/docs/plan/doing/opencode-v2-migration.md +++ b/docs/plan/doing/opencode-v2-migration.md @@ -73,6 +73,24 @@ Experimental, unconfigured adapters cover V2 model-context and native transport The permission smoke first captures a genuine decline from its own controlled `ctx.permission.assert` call, then registers the adapter with that independently obtained sample. On both tested versions, a forged matching prototype exposed a synthetic canary through custom inspection without the adapter; with the adapter, inspection was clean and the rejection still stopped the step. This fixture is not a production source of trusted types. The 2.2.6 public plugin API does not expose the host's decline constructor, and a factory that derives its prototype from an arbitrary tool error would reintroduce the vulnerability. A trusted production capability is still required; this adapter does not protect the trusted prototype against same-privilege mutation by other plugins. Other AFTER-hook failures can reach the model as tool errors rather than stop the step, so registration alone does not establish an independent fail-closed boundary. +An alternative, also unconfigured, `registerV2InterruptingToolOutputRedaction` +does not copy a decline prototype. It replaces the error with a generic host +tool error, interrupts the affected session, and blocks its model-request hooks +until a successful interruption settles. Failed, refused, timed-out or +unverifiable settlement leaves that session quarantined for the plugin's +lifetime; unknown identity or exhausted bounds closes the guard globally. +Interruption is session-wide and reports an interrupted outcome. On 2.2.5 and +2.2.6, isolated controls observed a stopped plain rejection, redacted feedback, +forged-cause containment, blocked retry, generate and compaction requests after +a simulated interruption failure, and an unaffected second session. A new +user prompt succeeded after a successful interruption settled. These tests +cover synthetic native HTTP requests, not every provider, title request, +concurrent in-flight operation or native question dismissal. Quarantine is in +memory and is lost on unload or restart; an unresolved host interrupt may also +outlive the plugin. The adapter +does not establish an independent enforcement boundary or authorize V2 +activation. + These are **not** complete gateway protection or parity. Transport-hook ordering and non-native providers, WebSocket behavior, images and other encodings, large/resumed histories, all four request kinds live, other error paths, command and event lifecycles, and the remaining guards also require validation. The adapters are intentionally not registered in the user's configuration. The 2.2.5 and 2.2.6 permission smokes also measure synthetic controls. A rejection caught directly from `ctx.permission.assert` in a tool owned by the fixture is the same cause later exposed to its after-hook; a fresh object with that observed prototype still stops the step. Dropping the cause or throwing a generic error from the after-hook instead allows a tool result to reach the model. An independently constructed Effect tagged-error class with the same name also fails the host's decline check on both versions. These observations neither authenticate an arbitrary tool's cause nor authorize the adapter. CI runs the permission, tool-error and shell smokes on each pinned version; the smoke fails if a supported host does not reproduce the controls. diff --git a/plugin/gateway-core/dist/v2/tool-output-redaction.d.ts b/plugin/gateway-core/dist/v2/tool-output-redaction.d.ts index 014e90a1..677923d4 100644 --- a/plugin/gateway-core/dist/v2/tool-output-redaction.d.ts +++ b/plugin/gateway-core/dist/v2/tool-output-redaction.d.ts @@ -1,5 +1,5 @@ import type { SecretRedactor } from "../hooks/shared/secret-redaction.js"; -import type { ToolHooks } from "./types.js"; +import type { Registration, ToolHooks } from "./types.js"; type ToolErrorFactory = (message: string, cause?: unknown) => unknown; type TrustedPermissionDeclineFactory = () => unknown; /** @@ -8,4 +8,34 @@ type TrustedPermissionDeclineFactory = () => unknown; * result from a tool error. A matching tag cannot authenticate a host type. */ export declare function registerV2ToolOutputRedaction(tool: ToolHooks, redactor: Pick, createToolError: ToolErrorFactory, runtimeVersion: string, createTrustedDecline: TrustedPermissionDeclineFactory): Promise<() => Promise>; +type InterruptEvent = { + sessionID: unknown; + status: unknown; + error?: unknown; + result?: unknown; +}; +type GuardEvent = { + sessionID: unknown; +}; +type GuardKind = "context" | "compaction" | "generate" | "title" | "http.request" | "experimental.ws.send"; +interface InterruptToolHooks { + hook(name: "execute.after", callback: (event: InterruptEvent) => Promise): Promise; +} +interface InterruptSessionHooks { + hook(name: GuardKind, callback: (event: GuardEvent) => Promise): Promise; + interrupt(input: { + sessionID: string; + resume: false; + }): Promise<{ + interrupted: boolean; + }>; + wait(input: { + sessionID: string; + }): Promise; +} +/** + * Experimental alternative to copying a host decline prototype. It is not a + * complete provider boundary and must not be configured as a standalone guard. + */ +export declare function registerV2InterruptingToolOutputRedaction(session: InterruptSessionHooks, tool: InterruptToolHooks, redactor: Pick, createToolError: ToolErrorFactory, runtimeVersion: string): Promise<() => Promise>; export {}; diff --git a/plugin/gateway-core/dist/v2/tool-output-redaction.js b/plugin/gateway-core/dist/v2/tool-output-redaction.js index 09ba6960..62540353 100644 --- a/plugin/gateway-core/dist/v2/tool-output-redaction.js +++ b/plugin/gateway-core/dist/v2/tool-output-redaction.js @@ -119,3 +119,188 @@ export async function registerV2ToolOutputRedaction(tool, redactor, createToolEr }); return () => registration.dispose(); } +const INTERRUPT_FAILURE = "Tool step withheld because session interruption is unresolved"; +const INTERRUPT_TIMEOUT_MS = 2000; +const MAX_INTERRUPTS = 8; +const MAX_QUARANTINED = 64; +const BAD_SESSION_ID = /[\u0000-\u001f\u007f]/; +function validSessionID(id) { + return typeof id === "string" && id.length > 0 && id.length <= 160 && id === id.trim() && !BAD_SESSION_ID.test(id); +} +// Only inspect own data properties. An opaque cause cannot be classified safely. +function cancellationKind(error) { + try { + if (!error || typeof error !== "object") + return "unknown"; + const descriptor = Object.getOwnPropertyDescriptor(error, "error"); + if (!descriptor) + return "other"; + if (!("value" in descriptor)) + return "unknown"; + const cause = descriptor.value; + if (cause === undefined || cause === null) + return "other"; + if (typeof cause !== "object") + return "unknown"; + const tag = Object.getOwnPropertyDescriptor(cause, "_tag"); + if (!tag) + return "other"; + if (!("value" in tag) || typeof tag.value !== "string") + return "unknown"; + return tag.value === PERMISSION_DECLINED || tag.value === "QuestionTool.CancelledError" ? "stop" : "other"; + } + catch { + return "unknown"; + } +} +/** + * Experimental alternative to copying a host decline prototype. It is not a + * complete provider boundary and must not be configured as a standalone guard. + */ +export async function registerV2InterruptingToolOutputRedaction(session, tool, redactor, createToolError, runtimeVersion) { + if (!VERIFIED_OPENCODE_VERSIONS.includes(runtimeVersion)) + throw new Error("V2 tool-output adapter requires verified OpenCode 2.2.5 or 2.2.6"); + if (typeof createToolError !== "function") + throw new Error("V2 Tool.Error factory required"); + const registrations = []; + const quarantined = new Map(); + let globallyClosed = false; + let activeInterrupts = 0; + const maybeRelease = (id, entry) => { + if (!entry.failed && entry.idle && !entry.waiter && entry.inflight === 0 && quarantined.get(id) === entry) { + quarantined.delete(id); + } + }; + const observeSettlement = (id, entry) => { + entry.idle = false; + if (entry.waiter || entry.failed) + return; + entry.waiter = true; + void Promise.resolve().then(() => session.wait({ sessionID: id })).then(() => { + entry.idle = true; + }, () => { + entry.failed = true; + }).finally(() => { + entry.waiter = false; + maybeRelease(id, entry); + }); + }; + const interrupt = async (event) => { + let id; + try { + id = event.sessionID; + } + catch { + globallyClosed = true; + throw new Error(INTERRUPT_FAILURE); + } + if (!validSessionID(id)) { + globallyClosed = true; + throw new Error(INTERRUPT_FAILURE); + } + let entry = quarantined.get(id); + if (!entry) { + if (quarantined.size >= MAX_QUARANTINED) { + globallyClosed = true; + throw new Error(INTERRUPT_FAILURE); + } + entry = { inflight: 0, idle: false, waiter: false, failed: false }; + quarantined.set(id, entry); + } + entry.inflight += 1; + let assignmentFailed = false; + try { + try { + event.error = createToolError("Tool step stopped"); + } + catch { + assignmentFailed = true; + entry.failed = true; + } + if (activeInterrupts >= MAX_INTERRUPTS) { + globallyClosed = true; + throw new Error(INTERRUPT_FAILURE); + } + activeInterrupts += 1; + const pending = Promise.resolve().then(() => session.interrupt({ sessionID: id, resume: false })); + void pending.finally(() => { activeInterrupts -= 1; }).catch(() => { }); + let timer; + try { + const timeout = new Promise((_, reject) => { + timer = setTimeout(() => reject(new Error(INTERRUPT_FAILURE)), INTERRUPT_TIMEOUT_MS); + }); + const outcome = await Promise.race([pending, timeout]); + if (!outcome || outcome.interrupted !== true || assignmentFailed) + throw new Error(INTERRUPT_FAILURE); + observeSettlement(id, entry); + } + finally { + if (timer) + clearTimeout(timer); + } + } + catch { + entry.failed = true; + throw new Error(INTERRUPT_FAILURE); + } + finally { + entry.inflight -= 1; + maybeRelease(id, entry); + } + }; + try { + for (const kind of ["context", "compaction", "generate", "title", "http.request", "experimental.ws.send"]) { + registrations.push(await session.hook(kind, async (event) => { + try { + const id = event.sessionID; + if (globallyClosed || !validSessionID(id) || quarantined.has(id)) + throw new Error(); + } + catch { + throw new Error(INTERRUPT_FAILURE); + } + })); + } + registrations.push(await tool.hook("execute.after", async (event) => { + let status; + let original; + try { + status = event.status; + if (status === "error") + original = event.error; + } + catch { + globallyClosed = true; + throw new Error(REDACTION_FAILURE); + } + if (status === "error") { + if (cancellationKind(original) !== "other") + return interrupt(event); + try { + event.error = createToolError(await safeErrorMessage(original, redactor)); + } + catch { + throw new Error(REDACTION_FAILURE); + } + return; + } + if (status !== "completed") + return interrupt(event); + try { + const wrapper = { result: event.result }; + await redactor.redactMutableValueAsync(wrapper); + event.result = wrapper.result; + } + catch { + throw new Error(REDACTION_FAILURE); + } + })); + } + catch { + await Promise.allSettled(registrations.map(async (registration) => registration.dispose())); + throw new Error("Unable to register V2 tool-output redaction"); + } + return async () => { + await Promise.allSettled(registrations.map(async (registration) => registration.dispose())); + }; +} diff --git a/plugin/gateway-core/src/v2/tool-output-redaction.ts b/plugin/gateway-core/src/v2/tool-output-redaction.ts index 5381aa69..b6db4f11 100644 --- a/plugin/gateway-core/src/v2/tool-output-redaction.ts +++ b/plugin/gateway-core/src/v2/tool-output-redaction.ts @@ -1,5 +1,5 @@ import type { SecretRedactor } from "../hooks/shared/secret-redaction.js" -import type { ToolHooks } from "./types.js" +import type { Registration, ToolHooks } from "./types.js" const REDACTION_FAILURE = "Tool output withheld because redaction failed" const ERROR_WITHHELD = "Tool execution failed; details withheld by gateway" @@ -120,3 +120,186 @@ export async function registerV2ToolOutputRedaction( }) return () => registration.dispose() } + +type InterruptEvent = { sessionID: unknown; status: unknown; error?: unknown; result?: unknown } +type GuardEvent = { sessionID: unknown } +type GuardKind = "context" | "compaction" | "generate" | "title" | "http.request" | "experimental.ws.send" +interface InterruptToolHooks { + hook(name: "execute.after", callback: (event: InterruptEvent) => Promise): Promise +} +interface InterruptSessionHooks { + hook(name: GuardKind, callback: (event: GuardEvent) => Promise): Promise + interrupt(input: { sessionID: string; resume: false }): Promise<{ interrupted: boolean }> + wait(input: { sessionID: string }): Promise +} +type Quarantine = { inflight: number; idle: boolean; waiter: boolean; failed: boolean } + +const INTERRUPT_FAILURE = "Tool step withheld because session interruption is unresolved" +const INTERRUPT_TIMEOUT_MS = 2000 +const MAX_INTERRUPTS = 8 +const MAX_QUARANTINED = 64 +const BAD_SESSION_ID = /[\u0000-\u001f\u007f]/ + +function validSessionID(id: unknown): id is string { + return typeof id === "string" && id.length > 0 && id.length <= 160 && id === id.trim() && !BAD_SESSION_ID.test(id) +} + +// Only inspect own data properties. An opaque cause cannot be classified safely. +function cancellationKind(error: unknown): "stop" | "other" | "unknown" { + try { + if (!error || typeof error !== "object") return "unknown" + const descriptor = Object.getOwnPropertyDescriptor(error, "error") + if (!descriptor) return "other" + if (!("value" in descriptor)) return "unknown" + const cause = descriptor.value + if (cause === undefined || cause === null) return "other" + if (typeof cause !== "object") return "unknown" + const tag = Object.getOwnPropertyDescriptor(cause, "_tag") + if (!tag) return "other" + if (!("value" in tag) || typeof tag.value !== "string") return "unknown" + return tag.value === PERMISSION_DECLINED || tag.value === "QuestionTool.CancelledError" ? "stop" : "other" + } catch { + return "unknown" + } +} + +/** + * Experimental alternative to copying a host decline prototype. It is not a + * complete provider boundary and must not be configured as a standalone guard. + */ +export async function registerV2InterruptingToolOutputRedaction( + session: InterruptSessionHooks, + tool: InterruptToolHooks, + redactor: Pick, + createToolError: ToolErrorFactory, + runtimeVersion: string, +): Promise<() => Promise> { + if (!VERIFIED_OPENCODE_VERSIONS.includes(runtimeVersion)) throw new Error("V2 tool-output adapter requires verified OpenCode 2.2.5 or 2.2.6") + if (typeof createToolError !== "function") throw new Error("V2 Tool.Error factory required") + const registrations: Registration[] = [] + const quarantined = new Map() + let globallyClosed = false + let activeInterrupts = 0 + + const maybeRelease = (id: string, entry: Quarantine): void => { + if (!entry.failed && entry.idle && !entry.waiter && entry.inflight === 0 && quarantined.get(id) === entry) { + quarantined.delete(id) + } + } + const observeSettlement = (id: string, entry: Quarantine): void => { + entry.idle = false + if (entry.waiter || entry.failed) return + entry.waiter = true + void Promise.resolve().then(() => session.wait({ sessionID: id })).then(() => { + entry.idle = true + }, () => { + entry.failed = true + }).finally(() => { + entry.waiter = false + maybeRelease(id, entry) + }) + } + const interrupt = async (event: InterruptEvent): Promise => { + let id: unknown + try { + id = event.sessionID + } catch { + globallyClosed = true + throw new Error(INTERRUPT_FAILURE) + } + if (!validSessionID(id)) { + globallyClosed = true + throw new Error(INTERRUPT_FAILURE) + } + let entry = quarantined.get(id) + if (!entry) { + if (quarantined.size >= MAX_QUARANTINED) { + globallyClosed = true + throw new Error(INTERRUPT_FAILURE) + } + entry = { inflight: 0, idle: false, waiter: false, failed: false } + quarantined.set(id, entry) + } + entry.inflight += 1 + let assignmentFailed = false + try { + try { + event.error = createToolError("Tool step stopped") + } catch { + assignmentFailed = true + entry.failed = true + } + if (activeInterrupts >= MAX_INTERRUPTS) { + globallyClosed = true + throw new Error(INTERRUPT_FAILURE) + } + activeInterrupts += 1 + const pending = Promise.resolve().then(() => session.interrupt({ sessionID: id, resume: false })) + void pending.finally(() => { activeInterrupts -= 1 }).catch(() => {}) + let timer: ReturnType | undefined + try { + const timeout = new Promise((_, reject) => { + timer = setTimeout(() => reject(new Error(INTERRUPT_FAILURE)), INTERRUPT_TIMEOUT_MS) + }) + const outcome = await Promise.race([pending, timeout]) + if (!outcome || outcome.interrupted !== true || assignmentFailed) throw new Error(INTERRUPT_FAILURE) + observeSettlement(id, entry) + } finally { + if (timer) clearTimeout(timer) + } + } catch { + entry.failed = true + throw new Error(INTERRUPT_FAILURE) + } finally { + entry.inflight -= 1 + maybeRelease(id, entry) + } + } + + try { + for (const kind of ["context", "compaction", "generate", "title", "http.request", "experimental.ws.send"] as const) { + registrations.push(await session.hook(kind, async (event) => { + try { + const id = event.sessionID + if (globallyClosed || !validSessionID(id) || quarantined.has(id)) throw new Error() + } catch { + throw new Error(INTERRUPT_FAILURE) + } + })) + } + registrations.push(await tool.hook("execute.after", async (event) => { + let status: unknown + let original: unknown + try { + status = event.status + if (status === "error") original = event.error + } catch { + globallyClosed = true + throw new Error(REDACTION_FAILURE) + } + if (status === "error") { + if (cancellationKind(original) !== "other") return interrupt(event) + try { + event.error = createToolError(await safeErrorMessage(original, redactor)) + } catch { + throw new Error(REDACTION_FAILURE) + } + return + } + if (status !== "completed") return interrupt(event) + try { + const wrapper = { result: event.result } + await redactor.redactMutableValueAsync(wrapper) + event.result = wrapper.result + } catch { + throw new Error(REDACTION_FAILURE) + } + })) + } catch { + await Promise.allSettled(registrations.map(async (registration) => registration.dispose())) + throw new Error("Unable to register V2 tool-output redaction") + } + return async () => { + await Promise.allSettled(registrations.map(async (registration) => registration.dispose())) + } +} diff --git a/plugin/gateway-core/test/v2/tool-output-interruption.test.mjs b/plugin/gateway-core/test/v2/tool-output-interruption.test.mjs new file mode 100644 index 00000000..60cd73c1 --- /dev/null +++ b/plugin/gateway-core/test/v2/tool-output-interruption.test.mjs @@ -0,0 +1,197 @@ +import assert from "node:assert/strict" +import test from "node:test" + +import { registerV2InterruptingToolOutputRedaction } from "../../dist/v2/tool-output-redaction.js" + +function deferred() { + let resolve + let reject + const promise = new Promise((yes, no) => { resolve = yes; reject = no }) + return { promise, resolve, reject } +} + +function fixture() { + const hooks = new Map() + const disposed = [] + const waits = [] + const interrupts = [] + const session = { + async hook(name, callback) { + hooks.set(name, callback) + return { async dispose() { disposed.push(name) } } + }, + async interrupt(input) { interrupts.push(input); return { interrupted: true } }, + wait(input) { const pending = deferred(); waits.push({ ...pending, input }); return pending.promise }, + } + const tool = { async hook(name, callback) { + hooks.set(name, callback) + return { async dispose() { disposed.push(name) } } + } } + const redactor = { async redactMutableValueAsync(value) { + if (typeof value.message === "string") value.message = value.message.replaceAll("secret-sentinel", "[REDACTED]") + if (typeof value.result?.content === "string") value.result.content = value.result.content.replaceAll("secret-sentinel", "[REDACTED]") + } } + const createError = (message) => new Error(message) + return { hooks, disposed, waits, interrupts, session, tool, redactor, createError, + register: () => registerV2InterruptingToolOutputRedaction(session, tool, redactor, createError, "2.2.6"), + } +} + +function rejected(sessionID, tag = "Permission.DeclinedError") { + const untrusted = Object.create({ toString() { return "secret-sentinel" } }) + Object.defineProperty(untrusted, "_tag", { value: tag, enumerable: true }) + Object.defineProperty(untrusted, "private", { value: "secret-sentinel" }) + return { sessionID, status: "error", error: Object.assign(new Error("secret-sentinel"), { error: untrusted }) } +} + +async function tick() { await new Promise((resolve) => setImmediate(resolve)) } + +test("plain V2 rejection is sanitized and quarantined until the interrupted session settles", async () => { + const f = fixture() + const dispose = await f.register() + const event = rejected("ses-one") + await f.hooks.get("execute.after")(event) + assert.equal(event.error.message, "Tool step stopped") + assert.equal(event.error.error, undefined) + assert.deepEqual(f.interrupts, [{ sessionID: "ses-one", resume: false }]) + await tick() + for (const name of ["context", "compaction", "generate", "title", "http.request", "experimental.ws.send"]) { + await assert.rejects(f.hooks.get(name)({ sessionID: "ses-one" }), /interruption is unresolved/) + await f.hooks.get(name)({ sessionID: "ses-other" }) + } + let reads = 0 + const changingIdentity = { get sessionID() { reads += 1; return reads === 1 ? "ses-one" : "ses-other" } } + await assert.rejects(f.hooks.get("context")(changingIdentity), /interruption is unresolved/) + assert.equal(reads, 1) + assert.equal(f.waits.length, 1) + f.waits[0].resolve() + await tick() + await f.hooks.get("context")({ sessionID: "ses-one" }) + await dispose() + assert.equal(f.disposed.length, 7) +}) + +test("ordinary tool output and feedback are redacted without interrupting", async () => { + const f = fixture() + await f.register() + const completed = { sessionID: "ses-one", status: "completed", result: { content: "secret-sentinel" } } + await f.hooks.get("execute.after")(completed) + assert.equal(completed.result.content, "[REDACTED]") + const feedback = rejected("ses-one", "Permission.CorrectedError") + await f.hooks.get("execute.after")(feedback) + assert.equal(feedback.error.message, "[REDACTED]") + assert.equal(feedback.error.error, undefined) + assert.equal(f.interrupts.length, 0) +}) + +test("failed interruption keeps only that session quarantined and does not expose the failure", async () => { + const f = fixture() + f.session.interrupt = async () => { throw new Error("secret-sentinel") } + await f.register() + const event = rejected("ses-one") + await assert.rejects(f.hooks.get("execute.after")(event), /^Error: Tool step withheld because session interruption is unresolved$/) + assert.equal(event.error.message, "Tool step stopped") + await assert.rejects(f.hooks.get("context")({ sessionID: "ses-one" }), /interruption is unresolved/) + await f.hooks.get("context")({ sessionID: "ses-other" }) + assert.equal(f.waits.length, 0) +}) + +test("a refused interruption or failed settlement leaves the affected session blocked", async () => { + const f = fixture() + f.session.interrupt = async () => ({ interrupted: false }) + await f.register() + await assert.rejects(f.hooks.get("execute.after")(rejected("ses-refused")), /interruption is unresolved/) + await assert.rejects(f.hooks.get("context")({ sessionID: "ses-refused" }), /interruption is unresolved/) + await f.hooks.get("context")({ sessionID: "ses-other" }) + + const g = fixture() + g.session.wait = async () => { throw new Error("secret-sentinel") } + await g.register() + await g.hooks.get("execute.after")(rejected("ses-wait-failed")) + await tick() + await assert.rejects(g.hooks.get("context")({ sessionID: "ses-wait-failed" }), /interruption is unresolved/) + await g.hooks.get("context")({ sessionID: "ses-other" }) +}) + +test("an unresponsive interruption times out without releasing the quarantine when it later returns", { timeout: 15000 }, async () => { + const f = fixture() + const pending = deferred() + f.session.interrupt = () => pending.promise + await f.register() + const started = Date.now() + await assert.rejects(f.hooks.get("execute.after")(rejected("ses-hung")), /interruption is unresolved/) + assert.ok(Date.now() - started < 10000) + pending.resolve({ interrupted: true }) + await tick() + await assert.rejects(f.hooks.get("context")({ sessionID: "ses-hung" }), /interruption is unresolved/) + await f.hooks.get("context")({ sessionID: "ses-other" }) +}) + +test("a second interruption cannot release an earlier session quarantine", async () => { + const f = fixture() + const first = deferred() + const second = deferred() + let calls = 0 + f.session.interrupt = () => (++calls === 1 ? first.promise : second.promise) + await f.register() + const a = f.hooks.get("execute.after")(rejected("ses-one")) + const b = f.hooks.get("execute.after")(rejected("ses-one")) + first.resolve({ interrupted: true }) + await a + await tick() + f.waits[0].resolve() + await tick() + await assert.rejects(f.hooks.get("context")({ sessionID: "ses-one" }), /interruption is unresolved/) + second.resolve({ interrupted: true }) + await b + await tick() + assert.equal(f.waits.length, 2) + f.waits[1].resolve() + await tick() + await f.hooks.get("context")({ sessionID: "ses-one" }) +}) + +test("opaque cancellation, missing identity, and partial registration fail safely", async () => { + const f = fixture() + await f.register() + const opaque = { sessionID: "ses-one", status: "error", error: new Proxy({}, { getOwnPropertyDescriptor() { throw new Error("secret-sentinel") } }) } + await f.hooks.get("execute.after")(opaque) + assert.deepEqual(f.interrupts, [{ sessionID: "ses-one", resume: false }]) + await assert.rejects(f.hooks.get("execute.after")(rejected("")), /interruption is unresolved/) + await assert.rejects(f.hooks.get("context")({ sessionID: "ses-other" }), /interruption is unresolved/) + + const g = fixture() + g.session.hook = async (name, callback) => { + if (name === "generate") throw new Error("secret-sentinel") + g.hooks.set(name, callback) + return { async dispose() { g.disposed.push(name) } } + } + await assert.rejects(g.register(), /^Error: Unable to register V2 tool-output redaction$/) + assert.deepEqual(g.disposed.sort(), ["compaction", "context"]) +}) + +test("a failing error factory still requests interruption and keeps the session blocked", async () => { + const f = fixture() + await registerV2InterruptingToolOutputRedaction(f.session, f.tool, f.redactor, () => { throw new Error("secret-sentinel") }, "2.2.6") + await assert.rejects(f.hooks.get("execute.after")(rejected("ses-one")), /^Error: Tool step withheld because session interruption is unresolved$/) + assert.deepEqual(f.interrupts, [{ sessionID: "ses-one", resume: false }]) + await assert.rejects(f.hooks.get("context")({ sessionID: "ses-one" }), /interruption is unresolved/) + await f.hooks.get("context")({ sessionID: "ses-other" }) +}) + +test("exhausting the quarantine limit closes the remaining sessions", async () => { + const f = fixture() + f.session.interrupt = async () => ({ interrupted: false }) + await f.register() + for (let index = 0; index < 65; index += 1) { + await assert.rejects(f.hooks.get("execute.after")(rejected(`ses-${index}`)), /interruption is unresolved/) + } + await assert.rejects(f.hooks.get("context")({ sessionID: "ses-other" }), /interruption is unresolved/) +}) + +test("a dismissed V2 question also stops the affected session", async () => { + const f = fixture() + await f.register() + await f.hooks.get("execute.after")(rejected("ses-question", "QuestionTool.CancelledError")) + assert.deepEqual(f.interrupts, [{ sessionID: "ses-question", resume: false }]) +}) diff --git a/scripts/gateway_v2_dependency_profile_probe.py b/scripts/gateway_v2_dependency_profile_probe.py index 4503c97c..2d7329cf 100644 --- a/scripts/gateway_v2_dependency_profile_probe.py +++ b/scripts/gateway_v2_dependency_profile_probe.py @@ -300,6 +300,8 @@ def start(self) -> None: # thread fails to start. Never spawn inside a fallible constructor. env = child_env(self.root) env["OPENCODE_PASSWORD"] = self.password + if self.cancelled.is_set() or time.monotonic() >= self.deadline: + raise ProbeFailure("case_timeout") self.process = subprocess.Popen( [self.binary, "serve", "--hostname", "127.0.0.1", "--port", str(self.port)], cwd=self.root / "project", env=env, stdin=subprocess.DEVNULL, @@ -1016,7 +1018,11 @@ def expire() -> None: provider_thread.start() provider_started = True config = prepare_fixture(root, state, provider.server_port) + if state.stopped.is_set() or time.monotonic() >= state.deadline: + raise ProbeFailure("case_timeout") runtime = PrivateRuntime(binary, root, state.deadline) + if state.stopped.is_set() or time.monotonic() >= state.deadline: + raise ProbeFailure("case_timeout") runtime.start() control = control_case(root, state, runtime, config) control["ownership"] = {"expected_pid": runtime.process.pid, "observed_pid": runtime.observed_pid} diff --git a/scripts/gateway_v2_permission_rejection_live_smoke.py b/scripts/gateway_v2_permission_rejection_live_smoke.py index b2bd8693..4d514734 100644 --- a/scripts/gateway_v2_permission_rejection_live_smoke.py +++ b/scripts/gateway_v2_permission_rejection_live_smoke.py @@ -87,6 +87,8 @@ def run_case( plugin.mkdir(parents=True) marker = root / "redactor-called" identity_marker = root / "host-cause.json" + quarantine_marker = root / "quarantine-enforced" + settled_marker = root / "interrupt-settled" registered_marker = root / "guard-registered" started, aborted, after = root / "started", root / "aborted", root / "after" dist = ROOT / "plugin/gateway-core/dist" @@ -96,16 +98,21 @@ def run_case( 'import { Error as ToolError } from "@opencode/plugin/promise/tool"\n' 'import { writeFileSync } from "node:fs"\n' 'import { inspect } from "node:util"\n' - f"import {{ registerV2ToolOutputRedaction }} from {json.dumps((dist / 'v2/tool-output-redaction.js').as_uri())}\n" + f"import {{ registerV2ToolOutputRedaction, registerV2InterruptingToolOutputRedaction }} from {json.dumps((dist / 'v2/tool-output-redaction.js').as_uri())}\n" f"import {{ createSecretRedactor }} from {json.dumps((dist / 'hooks/shared/secret-redaction.js').as_uri())}\n" 'export default Plugin.define({ id: "v2-permission-rejection-probe", async setup(ctx) {\n' ' if (!["2.2.5", "2.2.6"].includes(ctx.app.version)) throw new Error("fixture requires OpenCode 2.2.5 or 2.2.6")\n' ' let capturedDecline, registered = false, activeCase = false\n' - ' await ctx.tool.hook("execute.after", event => {\n' + ' await ctx.tool.hook("execute.after", async event => {\n' ' if (event.input?.kind === "bootstrap") return\n' f' writeFileSync({json.dumps(str(after))}, "after")\n' ' if (event.tool !== "synthetic_probe" || event.status !== "error") return\n' ' const mode = process.env.V2_SMOKE_CONTROL\n' + ' if (mode === "interrupt_fail_unprotected") {\n' + ' event.error = new ToolError({ message: "synthetic rejection" })\n' + f' writeFileSync({json.dumps(str(identity_marker))}, JSON.stringify({{ mode, applied: true }}))\n' + f' try {{ throw new Error({json.dumps(REDACTOR_CANARY)}) }} catch {{ throw new Error("Permission interruption failed") }}\n' + ' }\n' ' if (mode === "drop_cause") {\n' ' event.error = new ToolError({ message: "synthetic rejection" })\n' f' writeFileSync({json.dumps(str(identity_marker))}, JSON.stringify({{ mode, applied: true }}))\n' @@ -131,7 +138,7 @@ def run_case( ' event.error = new ToolError({ message: "synthetic rejection", error: clone })\n' f' writeFileSync({json.dumps(str(identity_marker))}, JSON.stringify({{ mode, applied: true, same }}))\n' ' }\n' - ' if (mode === "forged" || mode === "forged_unprotected") {\n' + ' if (mode === "forged" || mode === "forged_unprotected" || mode === "interrupt_forged") {\n' ' const base = Object.create(Error.prototype)\n' ' for (const key of ["constructor", "~effect/Effect", "pipe", "toJSON", "~effect/Effect/identifier", "~effect/Effect/evaluate", "~effect/Effect/successCont", "~effect/Effect/failureCont", "~effect/Effect/ensureCont", Symbol.for("nodejs.util.inspect.custom"), Symbol.iterator])\n' f' Object.defineProperty(base, key, {{ value() {{ return {json.dumps(FORGED_CANARY)} }} }})\n' @@ -143,6 +150,24 @@ def run_case( ' }\n' ' })\n' f' const redactor = process.env.V2_SMOKE_REDACTOR_FAIL === "1" ? {{ async redactMutableValueAsync() {{ if (activeCase) writeFileSync({json.dumps(str(marker))}, "called"); throw new Error({json.dumps(REDACTOR_CANARY)}) }} }} : createSecretRedactor({{ patterns: {json.dumps([FEEDBACK_CANARY, ABORT_CANARY])}, redactionToken: "[REDACTED]", limits: {{ maxDepth: 32, maxNodes: 10000, maxChars: 1000000 }} }})\n' + ' const mode = process.env.V2_SMOKE_CONTROL\n' + ' if (["interrupt_plain", "interrupt_fail", "interrupt_forged", "interrupt_feedback"].includes(mode)) {\n' + ' const session = {\n' + ' hook: (name, callback) => ctx.session.hook(name, async event => {\n' + f' try {{ await callback(event) }} catch (error) {{ writeFileSync({json.dumps(str(quarantine_marker))} + "-" + name, "enforced"); if (name === "context") writeFileSync({json.dumps(str(quarantine_marker))}, "enforced"); throw error }}\n' + ' }),\n' + ' interrupt: async input => {\n' + f' writeFileSync({json.dumps(str(identity_marker))}, JSON.stringify({{ mode, applied: true, phase: "before" }}))\n' + f' if (mode === "interrupt_fail") throw new Error({json.dumps(REDACTOR_CANARY)})\n' + ' const result = await ctx.session.interrupt(input)\n' + f' writeFileSync({json.dumps(str(identity_marker))}, JSON.stringify({{ mode, applied: true, phase: "after", result }}))\n' + ' return result\n' + ' },\n' + f' wait: async input => {{ await ctx.session.wait(input); writeFileSync({json.dumps(str(settled_marker))}, "settled") }},\n' + ' }\n' + ' await registerV2InterruptingToolOutputRedaction(session, ctx.tool, redactor, message => new ToolError({ message }), ctx.app.version)\n' + f' writeFileSync({json.dumps(str(registered_marker))}, "registered")\n' + ' }\n' ' const inspectForged = event => {\n' ' const mode = process.env.V2_SMOKE_CONTROL\n' ' if ((mode !== "forged" && mode !== "forged_unprotected") || event.tool !== "synthetic_probe" || event.status !== "error" || event.input?.kind === "bootstrap") return\n' @@ -194,9 +219,9 @@ def run_case( env.pop("OPENCODE_SESSION_ID", None) password = secrets.token_urlsafe(24) env["OPENCODE_PASSWORD"] = password - if mode in {"baseline", "host_clone", "drop_cause", "throw_after", "forged_unprotected", "schema_clone"}: + if mode in {"baseline", "host_clone", "drop_cause", "throw_after", "forged_unprotected", "schema_clone", "interrupt_plain", "interrupt_fail", "interrupt_fail_unprotected", "interrupt_forged", "interrupt_feedback"}: env["V2_SMOKE_GUARD"] = "off" - if mode in {"host_clone", "drop_cause", "throw_after", "forged", "forged_unprotected", "schema_clone"}: + if mode in {"host_clone", "drop_cause", "throw_after", "forged", "forged_unprotected", "schema_clone", "interrupt_plain", "interrupt_fail", "interrupt_fail_unprotected", "interrupt_forged", "interrupt_feedback"}: env["V2_SMOKE_CONTROL"] = mode if mode == "redactor_failure": env["V2_SMOKE_REDACTOR_FAIL"] = "1" @@ -232,6 +257,16 @@ def run_case( request(base, password, "POST", f"/api/session/{bootstrap}/permission/{pending[0]['id']}/reply", {"decision": "reject"}) request(base, password, "POST", f"/api/experimental/session/{bootstrap}/wait", timeout=25) wait_for(lambda: registered_marker.is_file(), process) + peer = peer_pending_id = None + if mode in {"interrupt_plain", "interrupt_fail", "interrupt_forged"}: + peer = request(base, password, "POST", "/api/session", session_body)["data"]["id"] + request(base, password, "POST", f"/api/session/{peer}/prompt", {"text": "PEER_PERMISSION_PROBE"}) + peer_pending = wait_for( + lambda: request(base, password, "GET", f"/api/session/{peer}/permission").get("data"), process + ) + if len(peer_pending) != 1: + raise RuntimeError("unexpected peer permissions") + peer_pending_id = peer_pending[0]["id"] with state.lock: start = len(state.requests) session = request( @@ -295,6 +330,54 @@ def run_case( if identity_marker.is_file() else {} ) + peer_pending_preserved = peer_completed = None + if peer is not None and peer_pending_id is not None: + pending_peer = request(base, password, "GET", f"/api/session/{peer}/permission").get("data", []) + peer_pending_preserved = len(pending_peer) == 1 and pending_peer[0].get("id") == peer_pending_id + if not peer_pending_preserved: + raise RuntimeError("other session's permission was affected") + with state.lock: + peer_start = len(state.requests) + request(base, password, "POST", f"/api/session/{peer}/permission/{peer_pending_id}/reply", {"decision": "once"}) + request(base, password, "POST", f"/api/experimental/session/{peer}/wait", timeout=25) + with state.lock: + peer_requests = list(state.requests[peer_start:]) + peer_completed = any(item.get("role") == "tool" for req in peer_requests for item in req.get("messages", [])) + retry_blocked = retry_completed = None + auxiliary_blocked = None + if mode == "interrupt_fail": + with state.lock: + retry_start = len(state.requests) + request(base, password, "POST", f"/api/session/{session}/prompt", {"text": "RETRY_PERMISSION_PROBE"}) + request(base, password, "POST", f"/api/experimental/session/{session}/wait", timeout=25) + with state.lock: + retry_blocked = len(state.requests) == retry_start + auxiliary_blocked = {} + for kind, endpoint, body in ( + ("generate", "generate", {"prompt": "AUXILIARY_QUARANTINE_PROBE"}), + ("compaction", "compact", {}), + ): + with state.lock: + auxiliary_start = len(state.requests) + try: + request(base, password, "POST", f"/api/session/{session}/{endpoint}", body, timeout=25) + if kind == "compaction": + request(base, password, "POST", f"/api/experimental/session/{session}/wait", timeout=25) + except urllib.error.HTTPError: + pass + with state.lock: + no_request = len(state.requests) == auxiliary_start + auxiliary_blocked[kind] = no_request and Path(str(quarantine_marker) + "-" + kind).is_file() + if mode == "interrupt_plain": + wait_for(lambda: settled_marker.is_file(), process) + with state.lock: + retry_start = len(state.requests) + request(base, password, "POST", f"/api/session/{session}/prompt", {"text": "RETRY_PERMISSION_PROBE"}) + request(base, password, "POST", f"/api/experimental/session/{session}/wait", timeout=25) + with state.lock: + retry_requests = list(state.requests[retry_start:]) + retry_info = request(base, password, "GET", f"/api/session/{session}")["data"] + retry_completed = bool(retry_requests) and retry_info.get("outcome") == "succeeded" return { "permission_requested": not cancel, "tool_returned_to_model": any( @@ -319,6 +402,15 @@ def run_case( "after_hook_seen": after.is_file(), "control_applied": control_observation.get("mode") == mode and control_observation.get("applied") is True, + "interrupt_phase": control_observation.get("phase"), + "interrupt_result": control_observation.get("result"), + "quarantine_enforced": quarantine_marker.is_file(), + "peer_pending_preserved": peer_pending_preserved, + "peer_completed": peer_completed, + "retry_blocked": retry_blocked, + "auxiliary_blocked": auxiliary_blocked, + "retry_completed": retry_completed, + "interrupt_guard_registered": registered_marker.is_file(), "host_cause_is_captured": control_observation.get("same"), "forged_exposed_to_inspect": control_observation.get("exposed"), "forged_canary_on_wire": FORGED_CANARY in wire, @@ -354,6 +446,62 @@ def control_evidence_passed(controls: dict[str, dict[str, Any]]) -> bool: return controls["host_clone"].get("host_cause_is_captured") is True +def interrupt_evidence_passed(rows: dict[str, dict[str, Any]]) -> bool: + expected = {"plain", "failure", "unprotected", "forged", "feedback"} + if set(rows) != expected or not all(isinstance(row, dict) for row in rows.values()): + return False + for row in rows.values(): + if any(row.get(field) is not True for field in ("permission_requested", "after_hook_seen")): + return False + if any(row.get(field) is not False for field in ( + "permission_bypassed", "redactor_exception_on_wire", "redactor_exception_in_database", "redactor_exception_in_server_log", + "forged_canary_on_wire", "forged_canary_in_database", "forged_canary_in_server_log", + )): + return False + for mode in ("plain", "forged"): + row = rows[mode] + result = row.get("interrupt_result") + if ( + row.get("interrupt_guard_registered") is not True + or row.get("control_applied") is not True + or row.get("interrupt_phase") != "after" + or not isinstance(result, dict) + or result.get("interrupted") is not True + or row.get("session_outcome") != "interrupted" + or row.get("tool_returned_to_model") is not False + or row.get("peer_pending_preserved") is not True + or row.get("peer_completed") is not True + ): + return False + failure = rows["failure"] + auxiliary = failure.get("auxiliary_blocked") + feedback = rows["feedback"] + unprotected = rows["unprotected"] + return ( + rows["plain"].get("retry_completed") is True + and failure.get("interrupt_guard_registered") is True + and failure.get("control_applied") is True + and failure.get("session_outcome") == "failed" + and failure.get("tool_returned_to_model") is False + and failure.get("quarantine_enforced") is True + and failure.get("retry_blocked") is True + and isinstance(auxiliary, dict) + and set(auxiliary) == {"generate", "compaction"} + and all(auxiliary.get(kind) is True for kind in ("generate", "compaction")) + and failure.get("peer_pending_preserved") is True + and failure.get("peer_completed") is True + and unprotected.get("control_applied") is True + and unprotected.get("tool_returned_to_model") is True + and feedback.get("interrupt_guard_registered") is True + and feedback.get("control_applied") is False + and feedback.get("tool_returned_to_model") is True + and feedback.get("redacted_feedback_on_wire") is True + and feedback.get("feedback_on_wire") is False + and feedback.get("feedback_in_database") is False + and feedback.get("feedback_in_server_log") is False + ) + + def main() -> int: capture = ThreadingHTTPServer(("127.0.0.1", 0), ErrorHandler) state = CaptureState() @@ -381,6 +529,16 @@ def main() -> int: for mode in ("forged", "forged_unprotected") } independent_schema = run_case(capture, state, "schema_clone", False) + interrupt_cases = { + name: run_case(capture, state, mode, feedback) + for name, mode, feedback in ( + ("plain", "interrupt_plain", False), + ("failure", "interrupt_fail", False), + ("unprotected", "interrupt_fail_unprotected", False), + ("forged", "interrupt_forged", False), + ("feedback", "interrupt_feedback", True), + ) + } passed = ( all( row["permission_requested"] @@ -421,6 +579,7 @@ def main() -> int: for row in cancellation.values() ) and control_evidence_passed(controls) + and interrupt_evidence_passed(interrupt_cases) and forged["forged"]["control_applied"] and forged["forged"]["host_cause_is_captured"] is True and forged["forged"]["forged_exposed_to_inspect"] is False @@ -444,6 +603,7 @@ def main() -> int: "controls": controls, "forged": forged, "independent_schema": independent_schema, + "interrupt_cases": interrupt_cases, } ) ) diff --git a/tests/test_gateway_v2_dependency_profile_probe.py b/tests/test_gateway_v2_dependency_profile_probe.py index dc744d2e..d2c82f16 100644 --- a/tests/test_gateway_v2_dependency_profile_probe.py +++ b/tests/test_gateway_v2_dependency_profile_probe.py @@ -8,6 +8,7 @@ import threading import time import unittest +from collections.abc import Callable from pathlib import Path from unittest.mock import MagicMock, patch @@ -351,23 +352,66 @@ def test_unstarted_provider_is_never_shutdown_or_joined(self) -> None: self.assertEqual(result["measurement"], "FAIL") self.assertTrue(result["cleanup_complete"]) - def test_absolute_case_deadline_cancels_controller_connections(self) -> None: + def test_watchdog_cancels_active_controller_connections(self) -> None: with tempfile.TemporaryDirectory() as temporary: cancelled = threading.Event() + controller_entered = threading.Event() runtime = MagicMock() runtime.abort_connections.side_effect = cancelled.set + real_timer = threading.Timer def blocked_controller(*args: object) -> None: - self.assertTrue(cancelled.wait(1)) + controller_entered.set() + self.assertTrue(cancelled.wait(3)) raise profile.ProbeFailure("cancelled_controller") - with patch.object(profile, "PRIMARY_SECONDS", 0.02), patch.object(profile, "PrivateRuntime", return_value=runtime), patch.object(profile, "control_case", side_effect=blocked_controller): + def coordinated_timer(_delay: float, expire: Callable[[], None]) -> threading.Timer: + def fire() -> None: + if controller_entered.wait(3): + expire() + return real_timer(0, fire) + + with patch.object(profile, "PRIMARY_SECONDS", 10), patch.object(profile.threading, "Timer", side_effect=coordinated_timer), patch.object(profile, "PrivateRuntime", return_value=runtime), patch.object(profile, "control_case", side_effect=blocked_controller): result = profile.run_case("/not-executed", Path(temporary), "baseline") self.assertEqual(result["reason"], "case_timeout") runtime.abort_connections.assert_called_once() runtime.close.assert_called_once() self.assertTrue(result["cleanup_complete"]) + def test_expiry_during_fixture_setup_does_not_start_a_runtime(self) -> None: + with tempfile.TemporaryDirectory() as temporary: + def slow_fixture(_root: Path, current: profile.ProfileState, _port: int) -> dict: + self.assertTrue(current.stopped.wait(3)) + return {} + + with patch.object(profile, "PRIMARY_SECONDS", 0.02), patch.object(profile, "prepare_fixture", side_effect=slow_fixture), patch.object(profile, "PrivateRuntime") as runtime: + result = profile.run_case("/not-executed", Path(temporary), "baseline") + self.assertEqual(result["reason"], "case_timeout") + self.assertTrue(result["cleanup_complete"]) + runtime.assert_not_called() + + def test_expired_or_cancelled_runtime_never_spawns(self) -> None: + with tempfile.TemporaryDirectory() as temporary: + for expired in (True, False): + with self.subTest(expired=expired): + deadline = time.monotonic() + (-1 if expired else 10) + runtime = profile.PrivateRuntime("/not-executed", Path(temporary), deadline) + if not expired: + runtime.cancelled.set() + with patch.object(profile.subprocess, "Popen") as spawn: + with self.assertRaisesRegex(profile.ProbeFailure, "case_timeout"): + runtime.start() + spawn.assert_not_called() + + runtime = profile.PrivateRuntime("/not-executed", Path(temporary), time.monotonic() + 10) + def cancelled_during_environment(_root: Path) -> dict: + runtime.cancelled.set() + return {} + with patch.object(profile, "child_env", side_effect=cancelled_during_environment), patch.object(profile.subprocess, "Popen") as spawn: + with self.assertRaisesRegex(profile.ProbeFailure, "case_timeout"): + runtime.start() + spawn.assert_not_called() + def test_abort_interrupts_partial_native_response_headers(self) -> None: with socket.socket() as server, tempfile.TemporaryDirectory() as temporary: server.bind(("127.0.0.1", 0));server.listen(1);server.settimeout(1) diff --git a/tests/test_gateway_v2_permission_rejection_controls.py b/tests/test_gateway_v2_permission_rejection_controls.py index cd93608f..75e46f68 100644 --- a/tests/test_gateway_v2_permission_rejection_controls.py +++ b/tests/test_gateway_v2_permission_rejection_controls.py @@ -8,7 +8,7 @@ ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT / "scripts")) -from gateway_v2_permission_rejection_live_smoke import control_evidence_passed +from gateway_v2_permission_rejection_live_smoke import control_evidence_passed, interrupt_evidence_passed class PermissionControlEvidenceTest(unittest.TestCase): @@ -60,5 +60,66 @@ def test_incomplete_or_non_boolean_evidence_is_not_accepted(self) -> None: self.assertFalse(control_evidence_passed(changed)) +class InterruptControlEvidenceTest(unittest.TestCase): + @staticmethod + def valid_controls() -> dict: + common = { + "permission_requested": True, + "after_hook_seen": True, + "permission_bypassed": False, + "redactor_exception_on_wire": False, + "redactor_exception_in_database": False, + "redactor_exception_in_server_log": False, + "forged_canary_on_wire": False, + "forged_canary_in_database": False, + "forged_canary_in_server_log": False, + } + stopped = { + **common, "interrupt_guard_registered": True, "control_applied": True, + "interrupt_phase": "after", "interrupt_result": {"interrupted": True}, + "session_outcome": "interrupted", "tool_returned_to_model": False, + "peer_pending_preserved": True, "peer_completed": True, + } + return { + "plain": {**stopped, "retry_completed": True}, + "forged": dict(stopped), + "failure": { + **common, "interrupt_guard_registered": True, "control_applied": True, + "session_outcome": "failed", "tool_returned_to_model": False, + "quarantine_enforced": True, "retry_blocked": True, + "auxiliary_blocked": {"generate": True, "compaction": True}, + "peer_pending_preserved": True, "peer_completed": True, + }, + "unprotected": {**common, "control_applied": True, "tool_returned_to_model": True}, + "feedback": { + **common, "interrupt_guard_registered": True, "control_applied": False, + "tool_returned_to_model": True, "redacted_feedback_on_wire": True, + "feedback_on_wire": False, "feedback_in_database": False, "feedback_in_server_log": False, + }, + } + + def test_requires_success_failure_peer_and_privacy_controls(self) -> None: + controls = self.valid_controls() + self.assertTrue(interrupt_evidence_passed(controls)) + for mode, field, value in ( + ("plain", "retry_completed", False), + ("plain", "peer_pending_preserved", False), + ("failure", "quarantine_enforced", False), + ("failure", "retry_blocked", False), + ("failure", "auxiliary_blocked", {"generate": True, "compaction": False}), + ("unprotected", "tool_returned_to_model", False), + ("forged", "forged_canary_in_database", True), + ("feedback", "feedback_on_wire", True), + ("feedback", "redacted_feedback_on_wire", False), + ): + with self.subTest(mode=mode, field=field): + changed = copy.deepcopy(controls) + changed[mode][field] = value + self.assertFalse(interrupt_evidence_passed(changed)) + missing = copy.deepcopy(controls) + del missing["failure"] + self.assertFalse(interrupt_evidence_passed(missing)) + + if __name__ == "__main__": unittest.main()