diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 34c2de5..1d6046d 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -69,7 +69,7 @@ holding --pricing allowed--> committed (dispatch wins; cancelHold reports "com holding --cancelHold-------> refused (cancellation wins; the request is refused at the boundary) ``` -A `refused` hold returns from the loop and arms a one-shot refusal keyed by session, provider, model, and kind. The transport-boundary hooks registered before the pricing guards consume it and replace the request with a credential-free placeholder aimed at a closed loopback endpoint, so OpenCode sees a retryable transport failure and its own retry policy decides what happens next. A placeholder is never re-evaluated by the guards. An armed refusal expires after 60 seconds and is dropped when a fresh hold starts for the same session, so it cannot refuse an unrelated request. `consumeRefusal` also suppresses this plugin's own retry recovery for that one failure, because a refusal is not a provider failure and the caller owns the follow-up. +A session can hold several requests at once (the agent loop and an auxiliary title request are prepared concurrently), so holds are tracked per request and `cancelHold` cancels every waiting request of that session. A `refused` hold returns from the loop and arms a one-shot refusal keyed by session, provider, model, and kind. The transport-boundary hooks registered before the pricing guards consume it and replace the request with a credential-free placeholder aimed at a closed loopback endpoint, so OpenCode sees a retryable transport failure and its own retry policy decides what happens next. A placeholder is never re-evaluated by the guards. An armed refusal expires after 60 seconds and is dropped when a fresh hold starts for the same session, so it cannot refuse an unrelated request. `consumeRefusal` also suppresses this plugin's own retry recovery for that one failure, because a refusal is not a provider failure and the caller owns the follow-up. The mechanism is deliberately generic: this plugin exports a capability, not a policy. It knows nothing about who calls it or why, a queue-owned hold is never cancellable this way, and an unused capability changes nothing. diff --git a/CHANGELOG.md b/CHANGELOG.md index 5db8416..1242359 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,11 @@ # Changelog +## 0.1.8 - 2026-10-16 + +- **A session can hold several requests at once, and cancelling cancels all of them.** OpenCode prepares auxiliary requests (the automatic title, generate, compaction) concurrently with the agent loop, so one session can have more than one request parked. The held-request bookkeeping is now per request rather than per session: `cancelHold` cancels every request of that session that is still waiting, each gets its own one-shot refusal at the transport boundary, and none can be left behind to dispatch at the next allowed window. A refusal is matched by provider, model, and request kind, and only a fresh request *of the same kind* supersedes an un-consumed refusal. +- Evidence: 107 tests pass, including a new case with a held primary request and a held title request cancelled together, each refusal consumed independently. +- Found while verifying the consuming plugin end to end: the title request and the prompt were both parked, only the newest hold was cancellable, and the title's hold could still have dispatched paid work at the window. + ## 0.1.7 - 2026-10-16 - **Generic hold cancellation (`cancelHold`).** A held request can now be cancelled by whoever owns the routing decision that made it obsolete, through one new RPC method on the existing `opencode-offpeak` channel: input `{ sessionID, reason? }`, output `{ cancelled, state, model? }`. It is generic infrastructure — the plugin learns nothing about the caller — and it removes the need for a session interruption: no `session.interrupt`, no synthetic continuation prompt, no lost turn. `cancelled: true` means the hold lost the race and the request will be refused; `state: "committed"` means the dispatch decision had already been taken and the caller must wait for the next safe boundary; `state: "none"` means nothing was parked. diff --git a/package.json b/package.json index 654eb71..c42558a 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "opencode-offpeak", - "version": "0.1.7", + "version": "0.1.8", "description": "Queue OpenCode agent tasks and guard model requests against expensive pricing windows.", "author": "Lutz Kind", "license": "MIT", diff --git a/src/engine.ts b/src/engine.ts index 332d12c..5e91e58 100644 --- a/src/engine.ts +++ b/src/engine.ts @@ -270,15 +270,19 @@ export class OffpeakEngine { */ readonly #holdAborts = new Map(); /** - * Requests parked by an interactive hold, keyed by session. At most one hold - * exists per session, so the entry is the session's authoritative request - * state while it waits. + * Requests parked by an interactive hold, keyed by session. OpenCode prepares + * auxiliary requests (for example the automatic title) concurrently with the + * agent loop, so a session can hold several requests at once and each of them + * is cancellable. */ - readonly #holds = new Map(); - /** Armed refusals, keyed by session, consumed once at the transport boundary. */ - readonly #refusals = new Map(); + readonly #holds = new Map>(); + /** Armed refusals, keyed by session, consumed once each at the boundary. */ + readonly #refusals = new Map(); /** Consumed refusals, used to keep this plugin's own retry recovery out of the way. */ - readonly #consumedRefusals = new Map(); + readonly #consumedRefusals = new Map< + string, + { at: number; providerID: string; modelID: string } + >(); /** Last settled hold per session, so a late `cancelHold` can report the truth. */ readonly #holdOutcomes = new Map(); #nextHoldID = 0; @@ -536,8 +540,8 @@ export class OffpeakEngine { : {}), waiting: tasks.filter((task) => task.state === "WAITING_OFFPEAK").length, interactiveWaiting: this.#interactiveWaiters.size, - interactiveHolds: this.#holds.size, - armedRefusals: this.#refusals.size, + interactiveHolds: this.holdDiagnostics().interactiveHolds, + armedRefusals: this.holdDiagnostics().armedRefusals, cancelledHolds: this.#cancelledHolds, refusedRequests: this.#refusedRequests, recoveryWaiting: this.#recoveries.size, @@ -749,38 +753,49 @@ export class OffpeakEngine { * discoverable and cancellable before it is dispatched. */ #beginHold(input: GuardInput): HeldRequest { - const existing = this.#holds.get(input.sessionID); - if (existing !== undefined) this.#endHold(existing); - // A fresh request is a fresh decision: a refusal armed for an earlier - // request of the same session must never leak onto this one. - this.#refusals.delete(input.sessionID); + const kind = input.kind ?? "primary"; + const session = this.#holds.get(input.sessionID) ?? new Map(); + // A fresh request of the same kind is a fresh decision: a refusal armed for + // an earlier request of that kind must never leak onto this one. Other + // kinds (a held title beside a held prompt) are untouched. + const remaining = (this.#refusals.get(input.sessionID) ?? []).filter( + (refusal) => refusal.kind !== kind, + ); + if (remaining.length === 0) this.#refusals.delete(input.sessionID); + else this.#refusals.set(input.sessionID, remaining); const hold: HeldRequest = { id: ++this.#nextHoldID, sessionID: input.sessionID, providerID: input.providerID, modelID: input.modelID, - kind: input.kind ?? "primary", + kind, state: "holding", at: this.#clock.now(), }; - this.#holds.set(input.sessionID, hold); + session.set(hold.id, hold); + this.#holds.set(input.sessionID, session); return hold; } #endHold(hold: HeldRequest): void { - const current = this.#holds.get(hold.sessionID); - if (current !== undefined && current.id === hold.id) this.#holds.delete(hold.sessionID); + const session = this.#holds.get(hold.sessionID); + if (session !== undefined) { + session.delete(hold.id); + if (session.size === 0) this.#holds.delete(hold.sessionID); + } this.#holdOutcomes.set(hold.sessionID, { state: hold.state, at: this.#clock.now() }); } /** Arm the one-shot transport refusal for a cancelled request. */ #armRefusal(hold: HeldRequest): void { - this.#refusals.set(hold.sessionID, { + const refusals = this.#refusals.get(hold.sessionID) ?? []; + refusals.push({ providerID: hold.providerID, modelID: hold.modelID, kind: hold.kind, expiresAt: this.#clock.now() + REFUSAL_TTL_MS, }); + this.#refusals.set(hold.sessionID, refusals); this.#refusedRequests += 1; this.#announceResume(hold.sessionID); } @@ -799,31 +814,37 @@ export class OffpeakEngine { * which only its owner may cancel. */ async cancelHold(input: { sessionID: string; reason?: string }): Promise { - const hold = this.#holds.get(input.sessionID); - if (hold === undefined) { + const session = this.#holds.get(input.sessionID); + const holds = [...(session?.values() ?? [])]; + const cancellable = holds.filter((hold) => hold.state === "holding"); + const first = cancellable[0] ?? holds[0]; + if (cancellable.length === 0 || first === undefined) { const last = this.#holdOutcomes.get(input.sessionID); const state = - last !== undefined && this.#clock.now() - last.at < HOLD_OUTCOME_TTL_MS ? last.state : "none"; - return { cancelled: false, state }; - } - if (hold.state !== "holding") { + first !== undefined + ? first.state + : last !== undefined && this.#clock.now() - last.at < HOLD_OUTCOME_TTL_MS + ? last.state + : "none"; return { cancelled: false, - state: hold.state, - model: `${hold.providerID}/${hold.modelID}`, + state, + ...(first === undefined ? {} : { model: `${first.providerID}/${first.modelID}` }), }; } - hold.state = "refused"; + for (const hold of cancellable) { + hold.state = "refused"; + this.#cancelledHolds += 1; + } this.#holdOutcomes.set(input.sessionID, { state: "refused", at: this.#clock.now() }); - this.#cancelledHolds += 1; - // The request is obsolete: nothing may continue it later. + // The requests are obsolete: nothing may continue them later. await this.#removeRecovery(input.sessionID, PRIMARY_RECOVERY_ORIGINS); this.#signal.notify(); this.#requestPump(); return { cancelled: true, state: "refused", - model: `${hold.providerID}/${hold.modelID}`, + model: `${first.providerID}/${first.modelID}`, }; } @@ -838,16 +859,19 @@ export class OffpeakEngine { modelID: string; kind: string; }): boolean { - const refusal = this.#refusals.get(input.sessionID); + const refusals = this.#refusals.get(input.sessionID); + const index = + refusals?.findIndex( + (refusal) => + refusal.providerID === input.providerID && + refusal.modelID === input.modelID && + refusal.kind === input.kind, + ) ?? -1; + if (refusals === undefined || index < 0) return false; + const [refusal] = refusals.splice(index, 1); if (refusal === undefined) return false; - if ( - refusal.providerID !== input.providerID || - refusal.modelID !== input.modelID || - refusal.kind !== input.kind - ) { - return false; - } - this.#refusals.delete(input.sessionID); + + if (refusals.length === 0) this.#refusals.delete(input.sessionID); if (this.#clock.now() > refusal.expiresAt) return false; this.#consumedRefusals.set(input.sessionID, { at: this.#clock.now(), @@ -865,8 +889,14 @@ export class OffpeakEngine { refusedRequests: number; } { return { - interactiveHolds: this.#holds.size, - armedRefusals: this.#refusals.size, + interactiveHolds: [...this.#holds.values()].reduce( + (total, session) => total + session.size, + 0, + ), + armedRefusals: [...this.#refusals.values()].reduce( + (total, refusals) => total + refusals.length, + 0, + ), cancelledHolds: this.#cancelledHolds, refusedRequests: this.#refusedRequests, }; diff --git a/test/hold-cancel.test.mjs b/test/hold-cancel.test.mjs index 3740305..346a8e3 100644 --- a/test/hold-cancel.test.mjs +++ b/test/hold-cancel.test.mjs @@ -186,6 +186,38 @@ test("cancelling a parked request refuses it at the transport boundary instead o } }); +test("cancelling a session cancels every parked request it holds", async () => { + const clock = new ManualClock("2026-10-12T02:00:00Z"); + const h = await harness("2026-10-12T02:00:00Z", { clock }); + try { + // OpenCode prepares the automatic title concurrently with the agent loop. + const primary = park(h.engine, "cancel-concurrent", "primary"); + const title = park(h.engine, "cancel-concurrent", "title"); + await until(async () => (await h.engine.status(selectedModel)).interactiveHolds === 2); + + const result = await h.engine.cancelHold({ sessionID: "cancel-concurrent" }); + assert.equal(result.cancelled, true); + await primary.settled; + await title.settled; + + const status = await h.engine.status(selectedModel); + assert.equal(status.interactiveHolds, 0); + assert.equal(status.armedRefusals, 2, "both requests are refused, not just the newest"); + const consume = (kind) => + h.engine.consumeRefusal({ + sessionID: "cancel-concurrent", + providerID: "opencode-go", + modelID: "deepseek-v4.1-flash", + kind, + }); + assert.equal(consume("primary"), true); + assert.equal(consume("title"), true); + assert.equal(consume("primary"), false); + } finally { + await h.close(); + } +}); + test("a refusal never leaks onto a different request", async () => { const clock = new ManualClock("2026-10-12T02:00:00Z"); const h = await harness("2026-10-12T02:00:00Z", { clock });