diff --git a/api/package.json b/api/package.json index 8759bb02e3..c8a2b20c7e 100644 --- a/api/package.json +++ b/api/package.json @@ -46,7 +46,7 @@ "@azure/storage-blob": "^12.30.0", "@google/genai": "^2.8.0", "@keyv/redis": "5.1.6", - "@librechat/agents": "^3.7.13", + "@librechat/agents": "^3.7.14", "@librechat/api": "*", "@librechat/data-schemas": "*", "@microsoft/microsoft-graph-client": "^3.0.7", diff --git a/api/server/controllers/agents/client.js b/api/server/controllers/agents/client.js index bfe23974fb..34f3e254af 100644 --- a/api/server/controllers/agents/client.js +++ b/api/server/controllers/agents/client.js @@ -640,7 +640,7 @@ class AgentClient extends BaseClient { hook: createSteerDrainHook(drainOptions), ...(isSteerPreemptSupported() && { preemptHook: createSteerPreemptBoundaryHook(drainOptions), - preemption: createSteerPreemptPoll(streamId), + preemption: createSteerPreemptPoll(streamId, this.jobCreatedAt), }), ...(isSteerTerminalContinuationSupported() && { terminalHook: createSteerTerminalContinuationHook(drainOptions), diff --git a/package-lock.json b/package-lock.json index 13af8c4a47..2496056d3d 100644 --- a/package-lock.json +++ b/package-lock.json @@ -63,7 +63,7 @@ "@azure/storage-blob": "^12.30.0", "@google/genai": "^2.8.0", "@keyv/redis": "5.1.6", - "@librechat/agents": "^3.7.13", + "@librechat/agents": "^3.7.14", "@librechat/api": "*", "@librechat/data-schemas": "*", "@microsoft/microsoft-graph-client": "^3.0.7", @@ -8286,41 +8286,41 @@ } }, "node_modules/@humanfs/core": { - "version": "0.19.1", - "resolved": "https://registry.npmjs.org/@humanfs/core/-/core-0.19.1.tgz", - "integrity": "sha512-5DyQ4+1JEUzejeK1JGICcideyfUbGixgS9jNgex5nqkW+cY7WZhxBigmieN5Qnw9ZosSNVC9KQKyb+GUaGyKUA==", + "version": "0.19.2", + "resolved": "https://registry.npmjs.org/@humanfs/core/-/core-0.19.2.tgz", + "integrity": "sha512-UhXNm+CFMWcbChXywFwkmhqjs3PRCmcSa/hfBgLIb7oQ5HNb1wS0icWsGtSAUNgefHeI+eBrA8I1fxmbHsGdvA==", "dev": true, "license": "Apache-2.0", + "dependencies": { + "@humanfs/types": "^0.15.0" + }, "engines": { "node": ">=18.18.0" } }, "node_modules/@humanfs/node": { - "version": "0.16.6", - "resolved": "https://registry.npmjs.org/@humanfs/node/-/node-0.16.6.tgz", - "integrity": "sha512-YuI2ZHQL78Q5HbhDiBA1X4LmYdXCKCMQIfw0pw7piHJwyREFebJUvrQN4cMssyES6x+vfUbx1CIpaQUKYdQZOw==", + "version": "0.16.8", + "resolved": "https://registry.npmjs.org/@humanfs/node/-/node-0.16.8.tgz", + "integrity": "sha512-gE1eQNZ3R++kTzFUpdGlpmy8kDZD/MLyHqDwqjkVQI0JMdI1D51sy1H958PNXYkM2rAac7e5/CnIKZrHtPh3BQ==", "dev": true, "license": "Apache-2.0", "dependencies": { - "@humanfs/core": "^0.19.1", - "@humanwhocodes/retry": "^0.3.0" + "@humanfs/core": "^0.19.2", + "@humanfs/types": "^0.15.0", + "@humanwhocodes/retry": "^0.4.0" }, "engines": { "node": ">=18.18.0" } }, - "node_modules/@humanfs/node/node_modules/@humanwhocodes/retry": { - "version": "0.3.1", - "resolved": "https://registry.npmjs.org/@humanwhocodes/retry/-/retry-0.3.1.tgz", - "integrity": "sha512-JBxkERygn7Bv/GbN5Rv8Ul6LVknS+5Bp6RgDC/O8gEBU/yeH5Ui5C/OlWrTb6qct7LjjfT6Re2NxB0ln0yYybA==", + "node_modules/@humanfs/types": { + "version": "0.15.0", + "resolved": "https://registry.npmjs.org/@humanfs/types/-/types-0.15.0.tgz", + "integrity": "sha512-ZZ1w0aoQkwuUuC7Yf+7sdeaNfqQiiLcSRbfI08oAxqLtpXQr9AIVX7Ay7HLDuiLYAaFPu8oBYNq/QIi9URHJ3Q==", "dev": true, "license": "Apache-2.0", "engines": { - "node": ">=18.18" - }, - "funding": { - "type": "github", - "url": "https://github.com/sponsors/nzakas" + "node": ">=18.18.0" } }, "node_modules/@humanwhocodes/module-importer": { @@ -10633,9 +10633,9 @@ } }, "node_modules/@librechat/agents": { - "version": "3.7.13", - "resolved": "https://registry.npmjs.org/@librechat/agents/-/agents-3.7.13.tgz", - "integrity": "sha512-V9KGPap07qgOgAzO5yZUAhav+KxzxfETj/gXDPmDHWw7IVyv4nILBDZzJ/Bu7TnvlzmleikKgnJDa6r9TNTMkQ==", + "version": "3.7.14", + "resolved": "https://registry.npmjs.org/@librechat/agents/-/agents-3.7.14.tgz", + "integrity": "sha512-aKWy0sJgR84eRUOujZAwL15nEfclDZ48ULlSGlazw5ozN/Ixiqr+bc+eY/cvDEBWPXpvoZKb/G+XjQJVUP3xVg==", "license": "MIT", "dependencies": { "@anthropic-ai/sdk": "^0.115.0", @@ -20581,9 +20581,9 @@ } }, "node_modules/@xmldom/xmldom": { - "version": "0.8.13", - "resolved": "https://registry.npmjs.org/@xmldom/xmldom/-/xmldom-0.8.13.tgz", - "integrity": "sha512-KRYzxepc14G/CEpEGc3Yn+JKaAeT63smlDr+vjB8jRfgTBBI9wRj/nkQEO+ucV8p8I9bfKLWp37uHgFrbntPvw==", + "version": "0.8.15", + "resolved": "https://registry.npmjs.org/@xmldom/xmldom/-/xmldom-0.8.15.tgz", + "integrity": "sha512-/5NV/vDALVFDXgLmfsy9TRCBlKwO2LNBFzpzvb9iIj+jR+eSc6DLYYvVOdivT/jm7MtU6TebYuRmzEOI7w40UA==", "license": "MIT", "engines": { "node": ">=10.0.0" @@ -36196,12 +36196,13 @@ } }, "node_modules/qs": { - "version": "6.15.2", - "resolved": "https://registry.npmjs.org/qs/-/qs-6.15.2.tgz", - "integrity": "sha512-Rzq0KEyX/w/tEybncDgdkZrJgVUsUMk3xjh3t5bv3S1HTAtg+uOYt72+ZfwiQwKdysThkTBdL/rTi6HDmX9Ddw==", + "version": "6.16.0", + "resolved": "https://registry.npmjs.org/qs/-/qs-6.16.0.tgz", + "integrity": "sha512-h6fhOIaRrID2CbEY2fqs+7t+UXZo+MLAnU5gRIq85uFtdiUPCdsApMlHhXogKVM4HM2DVbIjGNTTYH2OcmP1vA==", "license": "BSD-3-Clause", "dependencies": { - "side-channel": "^1.1.0" + "es-define-property": "^1.0.1", + "side-channel": "^1.1.1" }, "engines": { "node": ">=0.6" @@ -38687,14 +38688,14 @@ "license": "BSD-2-Clause" }, "node_modules/side-channel": { - "version": "1.1.0", - "resolved": "https://registry.npmjs.org/side-channel/-/side-channel-1.1.0.tgz", - "integrity": "sha512-ZX99e6tRweoUXqR+VBrslhda51Nh5MTQwou5tnUDgbtyM0dBgmhEDtWGP/xbKn6hqfPRHujUNwz5fy/wbbhnpw==", + "version": "1.1.1", + "resolved": "https://registry.npmjs.org/side-channel/-/side-channel-1.1.1.tgz", + "integrity": "sha512-6x6dK6zJdpTzF4sQeNYxwtvBzf6Eg4GtlesS94HOvTudUeyK2WXAaIfmDgsyslYrRBeFIlsi54AYsFGUuhmvrQ==", "license": "MIT", "dependencies": { "es-errors": "^1.3.0", - "object-inspect": "^1.13.3", - "side-channel-list": "^1.0.0", + "object-inspect": "^1.13.4", + "side-channel-list": "^1.0.1", "side-channel-map": "^1.0.1", "side-channel-weakmap": "^1.0.2" }, @@ -38706,13 +38707,13 @@ } }, "node_modules/side-channel-list": { - "version": "1.0.0", - "resolved": "https://registry.npmjs.org/side-channel-list/-/side-channel-list-1.0.0.tgz", - "integrity": "sha512-FCLHtRD/gnpCiCHEiJLOwdmFP+wzCmDEkc9y7NsYxeF4u7Btsn1ZuwgwJGxImImHicJArLP4R0yX4c2KCrMrTA==", + "version": "1.0.1", + "resolved": "https://registry.npmjs.org/side-channel-list/-/side-channel-list-1.0.1.tgz", + "integrity": "sha512-mjn/0bi/oUURjc5Xl7IaWi/OJJJumuoJFQJfDDyO46+hBWsfaVM65TBHq2eoZBhzl9EchxOijpkbRC8SVBQU0w==", "license": "MIT", "dependencies": { "es-errors": "^1.3.0", - "object-inspect": "^1.13.3" + "object-inspect": "^1.13.4" }, "engines": { "node": ">= 0.4" @@ -42638,7 +42639,7 @@ "@azure/storage-blob": "^12.30.0", "@google/genai": "^2.8.0", "@keyv/redis": "5.1.6", - "@librechat/agents": "^3.7.13", + "@librechat/agents": "^3.7.14", "@librechat/data-schemas": "*", "@modelcontextprotocol/sdk": "^1.30.0", "@opentelemetry/api": "^1.9.0", diff --git a/package.json b/package.json index a60c89de17..9e7659573d 100644 --- a/package.json +++ b/package.json @@ -197,7 +197,9 @@ "minimatch@10": { "brace-expansion": "^5.0.8" }, - "@xmldom/xmldom": "^0.8.13", + "@xmldom/xmldom": "^0.8.15", + "@humanfs/node": "^0.16.8", + "qs": "^6.16.0", "baseline-browser-mapping": "^2.11.0", "browserslist": "^4.28.7", "elliptic": "^6.6.1", diff --git a/packages/api/package.json b/packages/api/package.json index 2ead02b4fb..11721140b1 100644 --- a/packages/api/package.json +++ b/packages/api/package.json @@ -113,7 +113,7 @@ "@azure/storage-blob": "^12.30.0", "@google/genai": "^2.8.0", "@keyv/redis": "5.1.6", - "@librechat/agents": "^3.7.13", + "@librechat/agents": "^3.7.14", "@librechat/data-schemas": "*", "@modelcontextprotocol/sdk": "^1.30.0", "@opentelemetry/api": "^1.9.0", diff --git a/packages/api/src/agents/steering/__tests__/runtime.spec.ts b/packages/api/src/agents/steering/__tests__/runtime.spec.ts index ff37d594d4..ddc4d04355 100644 --- a/packages/api/src/agents/steering/__tests__/runtime.spec.ts +++ b/packages/api/src/agents/steering/__tests__/runtime.spec.ts @@ -20,6 +20,7 @@ import { createSteerPreemptPoll, isSteeringSupported, isSteerPreemptSupported, + isSteerPreemptRestartSupported, isSteerTerminalContinuationSupported, } from '../runtime'; import type { TerminalSteerHookInput } from '../runtime'; @@ -484,6 +485,20 @@ describe('isSteerPreemptSupported', () => { }); }); +describe('isSteerPreemptRestartSupported', () => { + /** + * A THIRD probe on top of the preempt one. An SDK that can seal still cannot + * act on an interrupt armed while the model is silent or merely thinking — + * exactly the window users reach for it most — so a host that conflated the + * two would hand out a wake channel nothing reads. + */ + it('requires the restart capability ON TOP of preempt support', () => { + const sdk = agentsSdk as { HOOK_PREEMPT_RESTART_CAPABLE?: boolean }; + const expected = isSteerPreemptSupported() && sdk.HOOK_PREEMPT_RESTART_CAPABLE === true; + expect(isSteerPreemptRestartSupported()).toBe(expected); + }); +}); + describe('createSteerPreemptBoundaryHook', () => { beforeEach(() => { GenerationJobManager.configure({ @@ -747,3 +762,170 @@ describe('createSteerPreemptPoll', () => { expect(createSteerPreemptPoll('no-such-stream').shouldPreempt()).toBe(false); }); }); + +/** + * The wake channel exists because the poll above is only read per streamed + * chunk. A steer armed while the provider is silent, or while it streams + * reasoning that will never become sealable, cannot reach the run any other + * way — it would wait out the whole turn and land as a terminal continuation. + */ +describe('preempt wake channel', () => { + beforeEach(() => { + GenerationJobManager.configure({ + jobStore: new InMemoryJobStore({ ttlAfterComplete: 60000 }), + eventTransport: new InMemoryEventTransport(), + isRedis: false, + cleanupOnComplete: false, + }); + GenerationJobManager.initialize(); + }); + + afterEach(async () => { + await GenerationJobManager.destroy(); + }); + + it('wakes a subscribed run when a steer arms', async () => { + const streamId = `preempt-wake-${Date.now()}`; + const job = await GenerationJobManager.createJob(streamId, 'user-1'); + const wake = jest.fn(); + GenerationJobManager.subscribePreempt(streamId, wake, job.createdAt); + + expect(wake).not.toHaveBeenCalled(); + await GenerationJobManager.requestPreempt(streamId, 's1', job.createdAt); + expect(wake).toHaveBeenCalledTimes(1); + expect(GenerationJobManager.isPreemptRequested(streamId)).toBe(true); + }); + + /** + * The wake is a hint, not the request. An arm the store REFUSES — a steer + * already drained at an ordinary boundary, so its id is tombstoned — leaves + * nothing for the run to act on, and waking would spend a look for nothing. + */ + it('stays quiet when the arm is refused', async () => { + const streamId = `preempt-wake-refused-${Date.now()}`; + const job = await GenerationJobManager.createJob(streamId, 'user-1'); + GenerationJobManager.noteSteersRemoved(streamId, ['s1'], job.createdAt); + const wake = jest.fn(); + GenerationJobManager.subscribePreempt(streamId, wake, job.createdAt); + + await GenerationJobManager.requestPreempt(streamId, 's1', job.createdAt); + + expect(wake).not.toHaveBeenCalled(); + expect(GenerationJobManager.isPreemptRequested(streamId)).toBe(false); + }); + + /** + * Requests are level-triggered, so an arm that landed before the run + * installed its listener has no callback to notify — and on a silent or + * reasoning-only turn no later chunk poll may ever run, which is the exact + * stall this channel removes. + */ + it('replays an arm that landed before the run subscribed', async () => { + const streamId = `preempt-wake-replay-${Date.now()}`; + const job = await GenerationJobManager.createJob(streamId, 'user-1'); + await GenerationJobManager.requestPreempt(streamId, 's1', job.createdAt); + + const wake = jest.fn(); + GenerationJobManager.subscribePreempt(streamId, wake, job.createdAt); + + expect(wake).toHaveBeenCalledTimes(1); + }); + + it('does not replay to a run that already looked', async () => { + const streamId = `preempt-wake-replay-once-${Date.now()}`; + const job = await GenerationJobManager.createJob(streamId, 'user-1'); + const first = jest.fn(); + GenerationJobManager.subscribePreempt(streamId, first, job.createdAt); + await GenerationJobManager.requestPreempt(streamId, 's1', job.createdAt); + expect(first).toHaveBeenCalledTimes(1); + + const second = jest.fn(); + GenerationJobManager.subscribePreempt(streamId, second, job.createdAt); + + expect(second).toHaveBeenCalledTimes(1); + expect(first).toHaveBeenCalledTimes(1); + }); + + it('does not replay when nothing is armed', async () => { + const streamId = `preempt-wake-no-replay-${Date.now()}`; + const job = await GenerationJobManager.createJob(streamId, 'user-1'); + const wake = jest.fn(); + + GenerationJobManager.subscribePreempt(streamId, wake, job.createdAt); + + expect(wake).not.toHaveBeenCalled(); + }); + + it('stops waking once the run unsubscribes', async () => { + const streamId = `preempt-wake-unsub-${Date.now()}`; + const job = await GenerationJobManager.createJob(streamId, 'user-1'); + const wake = jest.fn(); + const unsubscribe = GenerationJobManager.subscribePreempt(streamId, wake, job.createdAt); + + unsubscribe(); + await GenerationJobManager.requestPreempt(streamId, 's1', job.createdAt); + + expect(wake).not.toHaveBeenCalled(); + }); + + /** + * The same `createdAt` fence every other preempt entry point carries: a run + * wired to a generation that has since been replaced must not be woken by + * the replacement's arms. + */ + it('refuses to subscribe against a replaced generation', async () => { + const streamId = `preempt-wake-fence-${Date.now()}`; + const job = await GenerationJobManager.createJob(streamId, 'user-1'); + const wake = jest.fn(); + const unsubscribe = GenerationJobManager.subscribePreempt(streamId, wake, job.createdAt - 1); + + await GenerationJobManager.requestPreempt(streamId, 's1', job.createdAt); + + expect(wake).not.toHaveBeenCalled(); + expect(() => unsubscribe()).not.toThrow(); + }); + + /** + * A wake is an optimization over the per-chunk poll. A listener that throws + * must not fail the arm a durably queued steer already depends on. + */ + it('survives a throwing listener without losing the arm', async () => { + const streamId = `preempt-wake-throws-${Date.now()}`; + const job = await GenerationJobManager.createJob(streamId, 'user-1'); + const healthy = jest.fn(); + GenerationJobManager.subscribePreempt( + streamId, + () => { + throw new Error('listener exploded'); + }, + job.createdAt, + ); + GenerationJobManager.subscribePreempt(streamId, healthy, job.createdAt); + + await expect(GenerationJobManager.requestPreempt(streamId, 's1', job.createdAt)).resolves.toBe( + true, + ); + expect(healthy).toHaveBeenCalledTimes(1); + expect(GenerationJobManager.isPreemptRequested(streamId)).toBe(true); + }); + + /** + * The capability probe is what keeps the promise honest: an SDK that cannot + * discard an unstarted turn is handed no wake channel at all, so the run + * falls back to the per-chunk poll rather than subscribing to a signal + * nothing reads. + */ + it('supplies the channel only when the SDK can act on it', async () => { + const streamId = `preempt-wake-probe-${Date.now()}`; + await GenerationJobManager.createJob(streamId, 'user-1'); + /** Widened locally, the same way this file reads the SDK's capability + * constants: an SDK predating the restart contract types + * `StreamPreemption` without `subscribe`, and the probe below is exactly + * what decides whether the field is there to read. */ + const poll = createSteerPreemptPoll(streamId) as { + subscribe?: (wake: () => void) => () => void; + }; + + expect(typeof poll.subscribe === 'function').toBe(isSteerPreemptRestartSupported()); + }); +}); diff --git a/packages/api/src/agents/steering/index.ts b/packages/api/src/agents/steering/index.ts index 22af745578..11df15426a 100644 --- a/packages/api/src/agents/steering/index.ts +++ b/packages/api/src/agents/steering/index.ts @@ -5,6 +5,7 @@ export { createSteerPreemptPoll, isSteeringSupported, isSteerPreemptSupported, + isSteerPreemptRestartSupported, isSteerTerminalContinuationSupported, } from './runtime'; export type { diff --git a/packages/api/src/agents/steering/runtime.ts b/packages/api/src/agents/steering/runtime.ts index 44a0a06b46..aea73d4a57 100644 --- a/packages/api/src/agents/steering/runtime.ts +++ b/packages/api/src/agents/steering/runtime.ts @@ -315,10 +315,26 @@ export function createSteerTerminalContinuationHook( * job's armed preempt requests, exactly as the SDK contract requires: it * keeps returning true until the boundary drain (either boundary) clears the * request via `noteSteersRemoved`. Never consumes on read. + * + * `subscribe` adds the wake channel the poll alone cannot cover. The SDK only + * reads `shouldPreempt` per streamed chunk, so an interrupt armed while the + * model is silent — or while it streams reasoning that will never become + * sealable — would otherwise wait for the entire turn and land as a terminal + * continuation. Woken, the SDK discards the unstarted turn and re-issues it + * with the steer appended. The callback is a hint only; the poll above stays + * the authority, so the level-triggered contract is unchanged. + * + * Fenced on `jobCreatedAt` for the same reason every other preempt entry point + * is: a run wired to a generation that has since been replaced must not be + * woken by the replacement's arms. */ -export function createSteerPreemptPoll(streamId: string): StreamPreemption { +export function createSteerPreemptPoll(streamId: string, jobCreatedAt?: number): StreamPreemption { return { shouldPreempt: () => GenerationJobManager.isPreemptRequested(streamId), + ...(isSteerPreemptRestartSupported() && { + subscribe: (wake: () => void) => + GenerationJobManager.subscribePreempt(streamId, wake, jobCreatedAt), + }), }; } @@ -333,6 +349,22 @@ export function isSteerPreemptSupported(): boolean { return isSteeringSupported() && sdk.HOOK_PREEMPT_BOUNDARY_CAPABLE === true; } +/** + * Whether the installed SDK can honor a preempt on a turn that has produced + * nothing to keep — discarding the in-flight model call and re-issuing it + * rather than waiting for a sealable chunk that a silent or thinking turn + * never reaches. + * + * A THIRD probe, separate from `isSteerPreemptSupported()`, because the two + * differ in exactly the window users reach for an interrupt most. An SDK with + * only the seal path still accepts the request and still labels the chip + * "interrupting" — it simply cannot act until the model starts writing. + */ +export function isSteerPreemptRestartSupported(): boolean { + const sdk = agentsSdk as { HOOK_PREEMPT_RESTART_CAPABLE?: boolean }; + return isSteerPreemptSupported() && sdk.HOOK_PREEMPT_RESTART_CAPABLE === true; +} + export function isSteerTerminalContinuationSupported(): boolean { const sdk = agentsSdk as { HOOK_STOP_CONTINUATION_CAPABLE?: boolean }; return isSteeringSupported() && sdk.HOOK_STOP_CONTINUATION_CAPABLE === true; diff --git a/packages/api/src/stream/GenerationJobManager.ts b/packages/api/src/stream/GenerationJobManager.ts index df3d1663bb..aec4f26a9d 100644 --- a/packages/api/src/stream/GenerationJobManager.ts +++ b/packages/api/src/stream/GenerationJobManager.ts @@ -604,6 +604,16 @@ interface RuntimeJobState { createdAt: number; /** Server-minted steerIds requested but not yet drained/cancelled. */ ids: Set; + /** + * Run-scoped wake callbacks handed to the SDK's `StreamPreemption`. + * Notified after an arm is ACCEPTED so the model attempt can look at a + * request the per-chunk poll cannot reach — a provider that has gone + * silent, or one streaming reasoning that will never become sealable. + * + * Hints only. `ids` above stays the sole authority, so a spurious or + * duplicate wake costs one O(1) re-read and nothing else. + */ + wakes?: Set<() => void>; /** * steerIds whose removal was observed BEFORE their arm. Arm and clear are * published by different replicas over different connections, so a steer @@ -1301,9 +1311,70 @@ class GenerationJobManagerClass { state.ids.add(id); accepted += 1; } + if (accepted > 0) { + this.notePreemptArmed(state); + } return accepted; } + /** + * Wakes the live model attempt after an accepted arm. Called for local and + * cross-replica arms alike, since both land here through `armPreemptIds`. + * + * Errors are swallowed per listener: a wake is an optimization over the + * per-chunk poll, and a throwing listener must not fail the arm that a + * queued steer already depends on. + */ + private notePreemptArmed(state: NonNullable): void { + if (state.wakes == null) { + return; + } + for (const wake of state.wakes) { + try { + wake(); + } catch (error) { + logger.error('[GenerationJobManager] Preempt wake listener failed:', error); + } + } + } + + /** + * Registers a wake listener for the SDK's `StreamPreemption.subscribe`. + * Returns the unsubscribe the SDK calls when the model attempt ends. + * + * Fenced on `createdAt` like every other preempt entry point: a run wiring + * itself to a replaced generation must not be woken by the replacement's + * arms. A mismatch yields an inert unsubscribe rather than throwing, so the + * run still starts — it simply falls back to the per-chunk poll. + * + * An arm that ALREADY happened is replayed on registration. Requests are + * level-triggered, and the window between a job becoming steerable and the + * SDK installing its listener is real: an interrupt landing there would be + * recorded with no callbacks to notify, and on a silent or reasoning-only + * turn no later chunk poll may ever run — the exact stall this channel + * exists to remove. The replay makes the host correct on its own terms + * rather than resting on when the SDK first reads the flag. + */ + subscribePreempt(streamId: string, wake: () => void, jobCreatedAt?: number): () => void { + const runtime = this.runtimeState.get(streamId); + if (runtime == null || (jobCreatedAt != null && runtime.createdAt !== jobCreatedAt)) { + return () => {}; + } + const state = this.ensurePreemptState(runtime, jobCreatedAt ?? runtime.createdAt); + state.wakes ??= new Set(); + const wakes = state.wakes; + wakes.add(wake); + if (state.ids.size > 0) { + /** Only the new listener, and only after it is registered: waking the + * whole set would re-notify runs that already looked, and waking before + * registration would leave a concurrent arm with nowhere to land. */ + this.notePreemptArmed({ ...state, wakes: new Set([wake]) }); + } + return () => { + wakes.delete(wake); + }; + } + /** Non-terminal disarm used by a capable→incapable owner handover. The * queue item remains valid, so this records a revision floor instead of the * permanent removal tombstone used by `clearPreemptIds`. */