mirror of
https://github.com/danny-avila/LibreChat.git
synced 2026-10-21 17:44:11 +00:00
⏭️ feat: Honor an Interrupt Before the Model Has Answered (#15491)
* ⏭️ feat: Honor an Interrupt Before the Model Has Answered An interrupt armed while the model is still thinking sat as "Interrupting" until the entire turn finished, then landed as a terminal continuation. The cause is upstream: the SDK reads `shouldPreempt` once per streamed chunk, and its seal gate requires non-empty text — so a silent provider is never polled at all, and a reasoning-only turn is never sealable. `@librechat/agents` gains a discard-and-restart path for exactly that window (`HOOK_PREEMPT_RESTART_CAPABLE`). This wires the host half: - `GenerationJobManager.subscribePreempt` registers run-scoped wake listeners, notified after an ACCEPTED arm — local or cross-replica, since both land in `armPreemptIds`. Fenced on `createdAt` like every other preempt entry point, and a throwing listener never fails the arm a queued steer depends on. - `createSteerPreemptPoll` hands that channel to the SDK as `StreamPreemption.subscribe`, gated on a new `isSteerPreemptRestartSupported` probe so an SDK that cannot act on a wake is never handed one. The wake is a hint only: `isPreemptRequested` stays the single authority, so the level-triggered contract the seal path depends on is unchanged. Inert until `@librechat/agents` publishes the restart contract — the probe returns false against the pinned 3.7.11 and the run keeps today's behavior. * 🩹 fix: Replay an Already-Armed Preempt on Subscribe Preempt requests are level-triggered, and the window between a job becoming steerable and the SDK installing its model-attempt listener is real. An interrupt landing there was recorded with no callbacks to notify, and on a silent or reasoning-only turn no later chunk poll may ever run — leaving the request to wait out the whole turn, the exact stall this channel removes. `subscribePreempt` now replays an existing arm to the newly registered listener only, after registration: waking the whole set would re-notify runs that already looked, and waking before registration would leave a concurrent arm with nowhere to land. The `createdAt` fence is unchanged and still gates the replay. The SDK also reads the level flag once at each attempt's start, so this is belt-and-braces rather than the sole cover — but it makes the host correct on its own terms instead of resting on when the SDK happens to look. * 🔧 chore: Update @librechat/agents dependency to version 3.7.14 in package-lock.json and related package.json files * 🔧 chore: Update dependencies in package-lock.json and package.json - Bump versions for @humanfs/core (0.19.1 to 0.19.2), @humanfs/node (0.16.6 to 0.16.8), @xmldom/xmldom (0.8.13 to 0.8.15), and qs (6.15.2 to 6.16.0). - Add @humanfs/types (0.15.0) as a new dependency for @humanfs/node. - Update dependencies for side-channel and side-channel-list to their latest versions.
This commit is contained in:
Danny Avila
GitHub
parent
0219513115
commit
29f738dea8
9 changed files with 332 additions and 43 deletions
|
|
@ -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",
|
||||
|
|
|
|||
|
|
@ -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),
|
||||
|
|
|
|||
77
package-lock.json
generated
77
package-lock.json
generated
|
|
@ -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",
|
||||
|
|
|
|||
|
|
@ -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",
|
||||
|
|
|
|||
|
|
@ -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",
|
||||
|
|
|
|||
|
|
@ -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());
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -5,6 +5,7 @@ export {
|
|||
createSteerPreemptPoll,
|
||||
isSteeringSupported,
|
||||
isSteerPreemptSupported,
|
||||
isSteerPreemptRestartSupported,
|
||||
isSteerTerminalContinuationSupported,
|
||||
} from './runtime';
|
||||
export type {
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
|
|
|
|||
|
|
@ -604,6 +604,16 @@ interface RuntimeJobState {
|
|||
createdAt: number;
|
||||
/** Server-minted steerIds requested but not yet drained/cancelled. */
|
||||
ids: Set<string>;
|
||||
/**
|
||||
* 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<RuntimeJobState['preempt']>): 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`. */
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue