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

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions docs/plan/doing/opencode-v2-migration.md
Original file line number Diff line number Diff line change
Expand Up @@ -118,6 +118,10 @@ 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.
An accepted interruption invalidates any earlier settlement wait: if that wait
succeeds, the adapter starts another before releasing quarantine. This local
ordering fence does not prove the host's `session.wait` identifies or settles a
particular interruption.
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
Expand Down
15 changes: 11 additions & 4 deletions plugin/gateway-core/dist/v2/tool-output-redaction.js
Original file line number Diff line number Diff line change
Expand Up @@ -172,17 +172,22 @@ export async function registerV2InterruptingToolOutputRedaction(session, tool, r
}
};
const observeSettlement = (id, entry) => {
entry.idle = false;
if (entry.waiter || entry.failed)
return;
entry.waiter = true;
const generation = entry.generation;
void Promise.resolve().then(() => session.wait({ sessionID: id })).then(() => {
entry.idle = true;
if (entry.generation === generation)
entry.idle = true;
}, () => {
entry.failed = true;
}).finally(() => {
entry.waiter = false;
maybeRelease(id, entry);
// A wait started before a later accepted interruption cannot settle it.
if (!entry.failed && entry.generation !== generation)
observeSettlement(id, entry);
else
maybeRelease(id, entry);
});
};
const interrupt = async (event) => {
Expand All @@ -204,7 +209,7 @@ export async function registerV2InterruptingToolOutputRedaction(session, tool, r
globallyClosed = true;
throw new Error(INTERRUPT_FAILURE);
}
entry = { inflight: 0, idle: false, waiter: false, failed: false };
entry = { inflight: 0, idle: false, waiter: false, failed: false, generation: Symbol() };
quarantined.set(id, entry);
}
entry.inflight += 1;
Expand Down Expand Up @@ -232,6 +237,8 @@ export async function registerV2InterruptingToolOutputRedaction(session, tool, r
const outcome = await Promise.race([pending, timeout]);
if (!outcome || outcome.interrupted !== true || assignmentFailed)
throw new Error(INTERRUPT_FAILURE);
entry.generation = Symbol();
entry.idle = false;
observeSettlement(id, entry);
}
finally {
Expand Down
14 changes: 9 additions & 5 deletions plugin/gateway-core/src/v2/tool-output-redaction.ts
Original file line number Diff line number Diff line change
Expand Up @@ -132,7 +132,7 @@ interface InterruptSessionHooks {
interrupt(input: { sessionID: string; resume: false }): Promise<{ interrupted: boolean }>
wait(input: { sessionID: string }): Promise<unknown>
}
type Quarantine = { inflight: number; idle: boolean; waiter: boolean; failed: boolean }
type Quarantine = { inflight: number; idle: boolean; waiter: boolean; failed: boolean; generation: symbol }

const INTERRUPT_FAILURE = "Tool step withheld because session interruption is unresolved"
const INTERRUPT_TIMEOUT_MS = 2000
Expand Down Expand Up @@ -187,16 +187,18 @@ export async function registerV2InterruptingToolOutputRedaction(
}
}
const observeSettlement = (id: string, entry: Quarantine): void => {
entry.idle = false
if (entry.waiter || entry.failed) return
entry.waiter = true
const generation = entry.generation
void Promise.resolve().then(() => session.wait({ sessionID: id })).then(() => {
entry.idle = true
if (entry.generation === generation) entry.idle = true
}, () => {
entry.failed = true
}).finally(() => {
entry.waiter = false
maybeRelease(id, entry)
// A wait started before a later accepted interruption cannot settle it.
if (!entry.failed && entry.generation !== generation) observeSettlement(id, entry)
else maybeRelease(id, entry)
})
}
const interrupt = async (event: InterruptEvent): Promise<void> => {
Expand All @@ -217,7 +219,7 @@ export async function registerV2InterruptingToolOutputRedaction(
globallyClosed = true
throw new Error(INTERRUPT_FAILURE)
}
entry = { inflight: 0, idle: false, waiter: false, failed: false }
entry = { inflight: 0, idle: false, waiter: false, failed: false, generation: Symbol() }
quarantined.set(id, entry)
}
entry.inflight += 1
Expand All @@ -243,6 +245,8 @@ export async function registerV2InterruptingToolOutputRedaction(
})
const outcome = await Promise.race([pending, timeout])
if (!outcome || outcome.interrupted !== true || assignmentFailed) throw new Error(INTERRUPT_FAILURE)
entry.generation = Symbol()
entry.idle = false
observeSettlement(id, entry)
} finally {
if (timer) clearTimeout(timer)
Expand Down
132 changes: 132 additions & 0 deletions plugin/gateway-core/test/v2/tool-output-interruption.test.mjs
Original file line number Diff line number Diff line change
Expand Up @@ -151,6 +151,138 @@ test("a second interruption cannot release an earlier session quarantine", async
await f.hooks.get("context")({ sessionID: "ses-one" })
})

test("a wait begun before a second accepted interruption cannot release quarantine", async () => {
const f = fixture()
await f.register()
await f.hooks.get("execute.after")(rejected("ses-one"))
await tick()
assert.equal(f.waits.length, 1)
await f.hooks.get("execute.after")(rejected("ses-one"))
await tick()
assert.equal(f.waits.length, 1)
assert.equal(f.interrupts.length, 2)

f.waits[0].resolve()
await tick()
assert.equal(f.waits.length, 2)
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-peer" })
}
f.waits[1].resolve()
await tick()
await f.hooks.get("context")({ sessionID: "ses-one" })
})

test("reordered interruption acknowledgements require a wait after the last acceptance", 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"))
await tick()
second.resolve({ interrupted: true })
await b
await tick()
assert.equal(f.waits.length, 1)
first.resolve({ interrupted: true })
await a
f.waits[0].resolve()
await tick()
assert.equal(f.waits.length, 2)
await assert.rejects(f.hooks.get("context")({ sessionID: "ses-one" }), /interruption is unresolved/)
await f.hooks.get("context")({ sessionID: "ses-peer" })
f.waits[1].resolve()
await tick()
await f.hooks.get("context")({ sessionID: "ses-one" })
})

test("settlement is rechecked for interruptions accepted during successive waits", async () => {
const f = fixture()
await f.register()
for (let index = 0; index < 3; index += 1) {
await f.hooks.get("execute.after")(rejected("ses-one"))
await tick()
}
assert.equal(f.waits.length, 1)
f.waits[0].resolve()
await tick()
assert.equal(f.waits.length, 2)
await f.hooks.get("execute.after")(rejected("ses-one"))
f.waits[1].resolve()
await tick()
assert.equal(f.waits.length, 3)
await assert.rejects(f.hooks.get("context")({ sessionID: "ses-one" }), /interruption is unresolved/)
f.waits[2].resolve()
await tick()
await f.hooks.get("context")({ sessionID: "ses-one" })
})

test("a failed stale or current settlement remains quarantined across later interruptions", async () => {
for (const failStale of [true, false]) {
const f = fixture()
await f.register()
await f.hooks.get("execute.after")(rejected("ses-one"))
await tick()
await f.hooks.get("execute.after")(rejected("ses-one"))
if (failStale) {
f.waits[0].reject(new Error("secret-sentinel"))
} else {
f.waits[0].resolve()
await tick()
assert.equal(f.waits.length, 2)
f.waits[1].reject(new Error("secret-sentinel"))
}
await tick()
const waitCount = f.waits.length
await f.hooks.get("execute.after")(rejected("ses-one"))
await tick()
assert.equal(f.waits.length, waitCount)
await assert.rejects(f.hooks.get("context")({ sessionID: "ses-one" }), /interruption is unresolved/)
await f.hooks.get("context")({ sessionID: "ses-peer" })
}
})

test("a failed later interruption cannot be released by an earlier successful wait", async () => {
const f = fixture()
let calls = 0
f.session.interrupt = async () => ({ interrupted: ++calls === 1 })
await f.register()
await f.hooks.get("execute.after")(rejected("ses-one"))
await tick()
await assert.rejects(f.hooks.get("execute.after")(rejected("ses-one")), /interruption is unresolved/)
f.waits[0].resolve()
await tick()
await assert.rejects(f.hooks.get("context")({ sessionID: "ses-one" }), /interruption is unresolved/)
await f.hooks.get("context")({ sessionID: "ses-peer" })
})

test("settlement generations are independent across quarantined sessions", async () => {
const f = fixture()
await f.register()
await f.hooks.get("execute.after")(rejected("ses-one"))
await tick()
await f.hooks.get("execute.after")(rejected("ses-peer"))
await tick()
assert.deepEqual(f.waits.map((item) => item.input.sessionID), ["ses-one", "ses-peer"])
await f.hooks.get("execute.after")(rejected("ses-one"))
f.waits[1].resolve()
await tick()
await f.hooks.get("context")({ sessionID: "ses-peer" })
await assert.rejects(f.hooks.get("context")({ sessionID: "ses-one" }), /interruption is unresolved/)
f.waits[0].resolve()
await tick()
assert.equal(f.waits[2].input.sessionID, "ses-one")
await f.hooks.get("context")({ sessionID: "ses-peer" })
await assert.rejects(f.hooks.get("context")({ sessionID: "ses-one" }), /interruption is unresolved/)
f.waits[2].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()
Expand Down
Loading