mirror of
https://github.com/danny-avila/LibreChat.git
synced 2026-10-21 17:44:11 +00:00
🗿 fix: Bound Background Task Retention (#15477)
* fix: bound background task registry memory * fix: preserve background registry capacity state * fix: align background payload accounting * fix: harden background capacity admission * fix: honor background payload admission * fix: make background eviction atomic
This commit is contained in:
Danny Avila
GitHub
parent
ce209102b8
commit
a290677d21
3 changed files with 1233 additions and 70 deletions
|
|
@ -17,6 +17,7 @@ import {
|
|||
applyBackgroundToolCalls,
|
||||
synthesizeBackgroundToolOptions,
|
||||
registerBackgroundTaskTool,
|
||||
buildBackgroundCapacityContent,
|
||||
buildBackgroundHandleContent,
|
||||
runCheckBackgroundTask,
|
||||
getBackgroundCodeDelivery,
|
||||
|
|
@ -859,6 +860,60 @@ describe('BackgroundTaskRegistryClass', () => {
|
|||
expect(registry.get('u1', 'c1', created.task.id)?.attachments).toEqual(attachments);
|
||||
});
|
||||
|
||||
it('releases claimed artifact capacity before detached harvest attachments arrive', () => {
|
||||
const registry = new BackgroundTaskRegistryClass();
|
||||
const created = registry.create({
|
||||
userId: 'u-artifact-budget',
|
||||
conversationId: 'c-artifact-budget',
|
||||
toolCallId: 'call_code_budget',
|
||||
toolName: 'execute_code',
|
||||
});
|
||||
if ('atCapacity' in created) {
|
||||
throw new Error('unexpected capacity');
|
||||
}
|
||||
registry.complete('u-artifact-budget', 'c-artifact-budget', created.task.id, {
|
||||
content: 'stdout',
|
||||
artifact: { payload: 'a'.repeat(9_000_000) },
|
||||
});
|
||||
expect(
|
||||
registry.claimArtifact('u-artifact-budget', 'c-artifact-budget', created.task.id),
|
||||
).toBeDefined();
|
||||
|
||||
const attachments = [{ payload: 'b'.repeat(8_000_000) }];
|
||||
registry.attachHarvest('u-artifact-budget', 'c-artifact-budget', created.task.id, attachments);
|
||||
expect(
|
||||
registry.get('u-artifact-budget', 'c-artifact-budget', created.task.id)?.attachments,
|
||||
).toBe(attachments);
|
||||
});
|
||||
|
||||
it('replaces retained attachments without double-counting their payload', () => {
|
||||
const registry = new BackgroundTaskRegistryClass();
|
||||
const created = registry.create({
|
||||
userId: 'u-attachment-replace',
|
||||
conversationId: 'c-attachment-replace',
|
||||
toolCallId: 'call_attachment_replace',
|
||||
toolName: 'execute_code',
|
||||
});
|
||||
if ('atCapacity' in created) {
|
||||
throw new Error('unexpected capacity');
|
||||
}
|
||||
registry.complete('u-attachment-replace', 'c-attachment-replace', created.task.id, {
|
||||
content: 'stdout',
|
||||
});
|
||||
const first = [{ payload: 'a'.repeat(9_000_000) }];
|
||||
const replacement = [{ payload: 'b'.repeat(9_000_000) }];
|
||||
registry.attachHarvest('u-attachment-replace', 'c-attachment-replace', created.task.id, first);
|
||||
registry.attachHarvest(
|
||||
'u-attachment-replace',
|
||||
'c-attachment-replace',
|
||||
created.task.id,
|
||||
replacement,
|
||||
);
|
||||
expect(
|
||||
registry.get('u-attachment-replace', 'c-attachment-replace', created.task.id)?.attachments,
|
||||
).toBe(replacement);
|
||||
});
|
||||
|
||||
it('revokeHarvest hands a pending artifact to the fallback path', () => {
|
||||
const registry = new BackgroundTaskRegistryClass();
|
||||
const created = registry.create({
|
||||
|
|
@ -944,6 +999,41 @@ describe('BackgroundTaskRegistryClass', () => {
|
|||
expect(JSON.stringify(task)).not.toContain('raw failure');
|
||||
});
|
||||
|
||||
it('does not account payloads from rejected late completion updates', () => {
|
||||
const registry = new BackgroundTaskRegistryClass();
|
||||
const blocked = registry.create({
|
||||
userId: 'u-blocked-accounting',
|
||||
conversationId: 'c-blocked',
|
||||
toolCallId: 'call_blocked',
|
||||
toolName: 'execute_code',
|
||||
});
|
||||
if ('atCapacity' in blocked) {
|
||||
throw new Error('unexpected capacity');
|
||||
}
|
||||
registry.blockArtifact('u-blocked-accounting', 'c-blocked', blocked.task.id, 'blocked');
|
||||
registry.complete('u-blocked-accounting', 'c-blocked', blocked.task.id, {
|
||||
content: 'late',
|
||||
artifact: { payload: 'a'.repeat(10_000_000 - 20) },
|
||||
});
|
||||
|
||||
const next = registry.create({
|
||||
userId: 'u-blocked-accounting',
|
||||
conversationId: 'c-next',
|
||||
toolCallId: 'call_next',
|
||||
toolName: 'execute_code',
|
||||
});
|
||||
if ('atCapacity' in next) {
|
||||
throw new Error('unexpected capacity');
|
||||
}
|
||||
registry.complete('u-blocked-accounting', 'c-next', next.task.id, {
|
||||
content: 'next',
|
||||
artifact: { payload: 'b'.repeat(8_000_000) },
|
||||
});
|
||||
expect(registry.get('u-blocked-accounting', 'c-blocked', blocked.task.id)?.error).toBe(
|
||||
'blocked',
|
||||
);
|
||||
});
|
||||
|
||||
it('keeps abort-resistant tasks nonterminal instead of exposing false timeout evidence', () => {
|
||||
jest.useFakeTimers();
|
||||
try {
|
||||
|
|
@ -1016,6 +1106,55 @@ describe('BackgroundTaskRegistryClass', () => {
|
|||
expect(stored).toContain('[truncated: 150000 chars exceeded 100000 limit]');
|
||||
});
|
||||
|
||||
it('drops artifacts whose JSON serialization returns undefined', () => {
|
||||
const registry = new BackgroundTaskRegistryClass();
|
||||
const created = registry.create({
|
||||
userId: 'u-unmeasurable',
|
||||
conversationId: 'c-unmeasurable',
|
||||
toolCallId: 'call_unmeasurable',
|
||||
toolName: 'search_mcp_docs',
|
||||
});
|
||||
if ('atCapacity' in created) {
|
||||
throw new Error('unexpected capacity');
|
||||
}
|
||||
registry.complete('u-unmeasurable', 'c-unmeasurable', created.task.id, {
|
||||
content: 'done',
|
||||
artifact: { payload: 'x'.repeat(1_000_000), toJSON: () => undefined },
|
||||
});
|
||||
expect(registry.get('u-unmeasurable', 'c-unmeasurable', created.task.id)?.artifact).toBe(
|
||||
undefined,
|
||||
);
|
||||
});
|
||||
|
||||
it('retains only an artifact JSON projection without hidden object state', () => {
|
||||
const registry = new BackgroundTaskRegistryClass();
|
||||
const created = registry.create({
|
||||
userId: 'u-hidden-artifact',
|
||||
conversationId: 'c-hidden-artifact',
|
||||
toolCallId: 'call_hidden_artifact',
|
||||
toolName: 'search_mcp_docs',
|
||||
});
|
||||
if ('atCapacity' in created) {
|
||||
throw new Error('unexpected capacity');
|
||||
}
|
||||
const artifact = { visible: 'safe' };
|
||||
Object.defineProperty(artifact, 'hidden', {
|
||||
value: 'x'.repeat(1_000_000),
|
||||
enumerable: false,
|
||||
});
|
||||
registry.complete('u-hidden-artifact', 'c-hidden-artifact', created.task.id, {
|
||||
content: 'done',
|
||||
artifact,
|
||||
});
|
||||
const stored = registry.get(
|
||||
'u-hidden-artifact',
|
||||
'c-hidden-artifact',
|
||||
created.task.id,
|
||||
)?.artifact;
|
||||
expect(stored).toEqual({ visible: 'safe' });
|
||||
expect(stored).not.toBe(artifact);
|
||||
});
|
||||
|
||||
it('restores a claimed artifact after a failed delivery so a later claim retries', () => {
|
||||
const registry = new BackgroundTaskRegistryClass();
|
||||
const created = registry.create({
|
||||
|
|
@ -1098,6 +1237,58 @@ describe('BackgroundTaskRegistryClass', () => {
|
|||
expect(atCapacity).toBe(true);
|
||||
});
|
||||
|
||||
it('caps concurrent running tasks per user across conversations', () => {
|
||||
const registry = new BackgroundTaskRegistryClass();
|
||||
for (let i = 0; i < 40; i++) {
|
||||
const created = registry.create({
|
||||
userId: 'u-cap',
|
||||
conversationId: `c-${i}`,
|
||||
toolCallId: `call_${i}`,
|
||||
toolName: 'search_mcp_docs',
|
||||
});
|
||||
expect('atCapacity' in created).toBe(false);
|
||||
}
|
||||
expect(
|
||||
registry.create({
|
||||
userId: 'u-cap',
|
||||
conversationId: 'c-rejected',
|
||||
toolCallId: 'call_rejected',
|
||||
toolName: 'search_mcp_docs',
|
||||
}),
|
||||
).toEqual({ atCapacity: true, scope: 'user_running' });
|
||||
});
|
||||
|
||||
it('describes aggregate capacity rejections without blaming the conversation', () => {
|
||||
const content = JSON.parse(buildBackgroundCapacityContent('search', 'user_running')) as {
|
||||
scope: string;
|
||||
message: string;
|
||||
};
|
||||
expect(content.scope).toBe('user_running');
|
||||
expect(content.message).toContain('for this user');
|
||||
expect(content.message).not.toContain('in this conversation');
|
||||
});
|
||||
|
||||
it('caps concurrent running tasks process-wide', () => {
|
||||
const registry = new BackgroundTaskRegistryClass();
|
||||
for (let i = 0; i < 200; i++) {
|
||||
const created = registry.create({
|
||||
userId: `u-${i}`,
|
||||
conversationId: `c-${i}`,
|
||||
toolCallId: `call_${i}`,
|
||||
toolName: 'search_mcp_docs',
|
||||
});
|
||||
expect('atCapacity' in created).toBe(false);
|
||||
}
|
||||
expect(
|
||||
registry.create({
|
||||
userId: 'u-rejected',
|
||||
conversationId: 'c-rejected',
|
||||
toolCallId: 'call_rejected',
|
||||
toolName: 'search_mcp_docs',
|
||||
}),
|
||||
).toEqual({ atCapacity: true, scope: 'global_running' });
|
||||
});
|
||||
|
||||
it('holds capacity permits before task registration and releases them explicitly', () => {
|
||||
const registry = new BackgroundTaskRegistryClass();
|
||||
const permits = Array.from({ length: 10 }, (_, index) =>
|
||||
|
|
@ -1116,7 +1307,7 @@ describe('BackgroundTaskRegistryClass', () => {
|
|||
toolCallId: 'call_rejected',
|
||||
runId: 'run-1',
|
||||
}),
|
||||
).toEqual({ atCapacity: true });
|
||||
).toEqual({ atCapacity: true, scope: 'conversation_running' });
|
||||
const first = permits[0];
|
||||
if (!('permit' in first)) {
|
||||
throw new Error('expected capacity permit');
|
||||
|
|
@ -1207,6 +1398,508 @@ describe('BackgroundTaskRegistryClass', () => {
|
|||
expect(registry.list('u1', 'c-full')).toHaveLength(200);
|
||||
});
|
||||
|
||||
it('evicts oldest settled tasks at the per-user cap across conversations', () => {
|
||||
const registry = new BackgroundTaskRegistryClass();
|
||||
let firstTaskId = '';
|
||||
let latestTaskId = '';
|
||||
for (let i = 0; i <= 400; i++) {
|
||||
const created = registry.create({
|
||||
userId: 'u-aggregate',
|
||||
conversationId: `c-${i}`,
|
||||
toolCallId: `call_${i}`,
|
||||
toolName: 't',
|
||||
});
|
||||
if ('atCapacity' in created) {
|
||||
throw new Error(`unexpected capacity at ${i}`);
|
||||
}
|
||||
firstTaskId ||= created.task.id;
|
||||
latestTaskId = created.task.id;
|
||||
registry.complete('u-aggregate', `c-${i}`, created.task.id, { content: 'x' });
|
||||
}
|
||||
expect(registry.get('u-aggregate', 'c-0', firstTaskId)).toBeUndefined();
|
||||
expect(registry.get('u-aggregate', 'c-400', latestTaskId)?.status).toBe('completed');
|
||||
});
|
||||
|
||||
it('recreates a target bucket when aggregate eviction removes it', () => {
|
||||
const registry = new BackgroundTaskRegistryClass();
|
||||
for (let i = 0; i < 400; i++) {
|
||||
const created = registry.create({
|
||||
userId: 'u-reused-bucket',
|
||||
conversationId: `c-${i}`,
|
||||
toolCallId: `call_${i}`,
|
||||
toolName: 't',
|
||||
});
|
||||
if ('atCapacity' in created) {
|
||||
throw new Error(`unexpected capacity at ${i}`);
|
||||
}
|
||||
registry.complete('u-reused-bucket', `c-${i}`, created.task.id, { content: 'x' });
|
||||
}
|
||||
|
||||
const replacement = registry.create({
|
||||
userId: 'u-reused-bucket',
|
||||
conversationId: 'c-0',
|
||||
toolCallId: 'call_replacement',
|
||||
toolName: 't',
|
||||
});
|
||||
if ('atCapacity' in replacement) {
|
||||
throw new Error('unexpected replacement capacity');
|
||||
}
|
||||
registry.complete('u-reused-bucket', 'c-0', replacement.task.id, { content: 'replacement' });
|
||||
expect(registry.get('u-reused-bucket', 'c-0', replacement.task.id)?.result).toBe('replacement');
|
||||
});
|
||||
|
||||
it('uses one local eviction to satisfy bucket and aggregate task caps', () => {
|
||||
const registry = new BackgroundTaskRegistryClass();
|
||||
let otherOldestId = '';
|
||||
let targetOldestId = '';
|
||||
for (let i = 0; i < 200; i++) {
|
||||
const other = registry.create({
|
||||
userId: 'u-local-first',
|
||||
conversationId: 'c-other',
|
||||
toolCallId: `call_other_${i}`,
|
||||
toolName: 't',
|
||||
});
|
||||
if ('atCapacity' in other) {
|
||||
throw new Error(`unexpected other capacity at ${i}`);
|
||||
}
|
||||
otherOldestId ||= other.task.id;
|
||||
registry.complete('u-local-first', 'c-other', other.task.id, { content: 'other' });
|
||||
}
|
||||
for (let i = 0; i < 200; i++) {
|
||||
const target = registry.create({
|
||||
userId: 'u-local-first',
|
||||
conversationId: 'c-target',
|
||||
toolCallId: `call_target_${i}`,
|
||||
toolName: 't',
|
||||
});
|
||||
if ('atCapacity' in target) {
|
||||
throw new Error(`unexpected target capacity at ${i}`);
|
||||
}
|
||||
targetOldestId ||= target.task.id;
|
||||
registry.complete('u-local-first', 'c-target', target.task.id, { content: 'target' });
|
||||
}
|
||||
|
||||
const replacement = registry.create({
|
||||
userId: 'u-local-first',
|
||||
conversationId: 'c-target',
|
||||
toolCallId: 'call_replacement',
|
||||
toolName: 't',
|
||||
});
|
||||
expect('atCapacity' in replacement).toBe(false);
|
||||
expect(registry.get('u-local-first', 'c-other', otherOldestId)).toBeDefined();
|
||||
expect(registry.get('u-local-first', 'c-target', targetOldestId)).toBeUndefined();
|
||||
});
|
||||
|
||||
it('evicts by settlement time instead of dispatch time', () => {
|
||||
const registry = new BackgroundTaskRegistryClass();
|
||||
let now = Date.now();
|
||||
const nowSpy = jest.spyOn(Date, 'now').mockImplementation(() => now);
|
||||
try {
|
||||
const slow = registry.create({
|
||||
userId: 'u-settlement-order',
|
||||
conversationId: 'c-slow',
|
||||
toolCallId: 'call_slow',
|
||||
toolName: 't',
|
||||
});
|
||||
if ('atCapacity' in slow) {
|
||||
throw new Error('unexpected slow-task capacity');
|
||||
}
|
||||
|
||||
let oldestSettledId = '';
|
||||
for (let i = 0; i < 399; i++) {
|
||||
now++;
|
||||
const fast = registry.create({
|
||||
userId: 'u-settlement-order',
|
||||
conversationId: `c-fast-${i}`,
|
||||
toolCallId: `call_fast_${i}`,
|
||||
toolName: 't',
|
||||
});
|
||||
if ('atCapacity' in fast) {
|
||||
throw new Error(`unexpected fast-task capacity at ${i}`);
|
||||
}
|
||||
oldestSettledId ||= fast.task.id;
|
||||
registry.complete('u-settlement-order', `c-fast-${i}`, fast.task.id, {
|
||||
content: 'fast',
|
||||
});
|
||||
}
|
||||
|
||||
now += 1_000;
|
||||
registry.complete('u-settlement-order', 'c-slow', slow.task.id, { content: 'slow' });
|
||||
now++;
|
||||
const replacement = registry.create({
|
||||
userId: 'u-settlement-order',
|
||||
conversationId: 'c-replacement',
|
||||
toolCallId: 'call_replacement',
|
||||
toolName: 't',
|
||||
});
|
||||
expect('atCapacity' in replacement).toBe(false);
|
||||
expect(registry.get('u-settlement-order', 'c-slow', slow.task.id)?.result).toBe('slow');
|
||||
expect(registry.get('u-settlement-order', 'c-fast-0', oldestSettledId)).toBeUndefined();
|
||||
} finally {
|
||||
nowSpy.mockRestore();
|
||||
}
|
||||
});
|
||||
|
||||
it('evicts oldest settled tasks at the process-wide cap', () => {
|
||||
const registry = new BackgroundTaskRegistryClass();
|
||||
let firstTaskId = '';
|
||||
let latestTaskId = '';
|
||||
for (let i = 0; i <= 2_000; i++) {
|
||||
const created = registry.create({
|
||||
userId: `u-${i}`,
|
||||
conversationId: `c-${i}`,
|
||||
toolCallId: `call_${i}`,
|
||||
toolName: 't',
|
||||
});
|
||||
if ('atCapacity' in created) {
|
||||
throw new Error(`unexpected capacity at ${i}`);
|
||||
}
|
||||
firstTaskId ||= created.task.id;
|
||||
latestTaskId = created.task.id;
|
||||
registry.complete(`u-${i}`, `c-${i}`, created.task.id, { content: 'x' });
|
||||
}
|
||||
expect(registry.get('u-0', 'c-0', firstTaskId)).toBeUndefined();
|
||||
expect(registry.get('u-2000', 'c-2000', latestTaskId)?.status).toBe('completed');
|
||||
});
|
||||
|
||||
it('bounds retained payloads per user across conversations', () => {
|
||||
const registry = new BackgroundTaskRegistryClass();
|
||||
let firstTaskId = '';
|
||||
let latestTaskId = '';
|
||||
for (let i = 0; i < 18; i++) {
|
||||
const created = registry.create({
|
||||
userId: 'u-payload',
|
||||
conversationId: `c-${i}`,
|
||||
toolCallId: `call_${i}`,
|
||||
toolName: 't',
|
||||
});
|
||||
if ('atCapacity' in created) {
|
||||
throw new Error(`unexpected capacity at ${i}`);
|
||||
}
|
||||
firstTaskId ||= created.task.id;
|
||||
latestTaskId = created.task.id;
|
||||
registry.complete('u-payload', `c-${i}`, created.task.id, {
|
||||
content: 'x',
|
||||
artifact: { payload: 'x'.repeat(1_000_000) },
|
||||
});
|
||||
}
|
||||
expect(registry.get('u-payload', 'c-0', firstTaskId)).toBeUndefined();
|
||||
expect(registry.get('u-payload', 'c-17', latestTaskId)?.artifact).toBeDefined();
|
||||
});
|
||||
|
||||
it('does not evict a completed task while its harvest is pending', () => {
|
||||
const registry = new BackgroundTaskRegistryClass();
|
||||
const first = registry.create({
|
||||
userId: 'u-pending-harvest',
|
||||
conversationId: 'c-first',
|
||||
toolCallId: 'call_first',
|
||||
toolName: 'execute_code',
|
||||
harvestStarted: true,
|
||||
});
|
||||
const second = registry.create({
|
||||
userId: 'u-pending-harvest',
|
||||
conversationId: 'c-second',
|
||||
toolCallId: 'call_second',
|
||||
toolName: 'execute_code',
|
||||
harvestStarted: true,
|
||||
});
|
||||
if ('atCapacity' in first || 'atCapacity' in second) {
|
||||
throw new Error('unexpected capacity');
|
||||
}
|
||||
registry.complete('u-pending-harvest', 'c-first', first.task.id, {
|
||||
content: 'first',
|
||||
artifact: { payload: 'a'.repeat(9_000_000) },
|
||||
harvestStarted: true,
|
||||
});
|
||||
registry.complete('u-pending-harvest', 'c-second', second.task.id, {
|
||||
content: 'second',
|
||||
artifact: { payload: 'b'.repeat(9_000_000) },
|
||||
harvestStarted: true,
|
||||
});
|
||||
expect(registry.get('u-pending-harvest', 'c-first', first.task.id)?.artifact).toBeDefined();
|
||||
expect(registry.get('u-pending-harvest', 'c-second', second.task.id)?.artifact).toBeUndefined();
|
||||
expect(registry.get('u-pending-harvest', 'c-second', second.task.id)?.result).toBeUndefined();
|
||||
});
|
||||
|
||||
it('does not evict an ordinary task while completion persistence is pending', () => {
|
||||
const registry = new BackgroundTaskRegistryClass();
|
||||
const first = registry.create({
|
||||
userId: 'u-pending-persistence',
|
||||
conversationId: 'c-first',
|
||||
toolCallId: 'call_first',
|
||||
toolName: 'search_mcp_docs',
|
||||
});
|
||||
const second = registry.create({
|
||||
userId: 'u-pending-persistence',
|
||||
conversationId: 'c-second',
|
||||
toolCallId: 'call_second',
|
||||
toolName: 'search_mcp_docs',
|
||||
});
|
||||
if ('atCapacity' in first || 'atCapacity' in second) {
|
||||
throw new Error('unexpected capacity');
|
||||
}
|
||||
registry.complete('u-pending-persistence', 'c-first', first.task.id, {
|
||||
content: 'first',
|
||||
artifact: { payload: 'a'.repeat(9_000_000) },
|
||||
});
|
||||
registry.markCompletionPersistencePending('u-pending-persistence', 'c-first', first.task.id);
|
||||
registry.complete('u-pending-persistence', 'c-second', second.task.id, {
|
||||
content: 'second',
|
||||
artifact: { payload: 'b'.repeat(9_000_000) },
|
||||
});
|
||||
expect(registry.get('u-pending-persistence', 'c-first', first.task.id)).toBeDefined();
|
||||
expect(
|
||||
registry.get('u-pending-persistence', 'c-second', second.task.id)?.result,
|
||||
).toBeUndefined();
|
||||
});
|
||||
|
||||
it('skips zero-byte records when evicting retained payloads', () => {
|
||||
const registry = new BackgroundTaskRegistryClass();
|
||||
const empty = registry.create({
|
||||
userId: 'u-zero-byte',
|
||||
conversationId: 'c-empty',
|
||||
toolCallId: 'call_empty',
|
||||
toolName: 't',
|
||||
});
|
||||
const retained = registry.create({
|
||||
userId: 'u-zero-byte',
|
||||
conversationId: 'c-retained',
|
||||
toolCallId: 'call_retained',
|
||||
toolName: 't',
|
||||
});
|
||||
const incoming = registry.create({
|
||||
userId: 'u-zero-byte',
|
||||
conversationId: 'c-incoming',
|
||||
toolCallId: 'call_incoming',
|
||||
toolName: 't',
|
||||
});
|
||||
if ('atCapacity' in empty || 'atCapacity' in retained || 'atCapacity' in incoming) {
|
||||
throw new Error('unexpected capacity');
|
||||
}
|
||||
registry.complete('u-zero-byte', 'c-empty', empty.task.id, { content: '' });
|
||||
registry.complete('u-zero-byte', 'c-retained', retained.task.id, {
|
||||
content: 'retained',
|
||||
artifact: { payload: 'a'.repeat(9_000_000) },
|
||||
});
|
||||
registry.complete('u-zero-byte', 'c-incoming', incoming.task.id, {
|
||||
content: 'incoming',
|
||||
artifact: { payload: 'b'.repeat(8_000_000) },
|
||||
});
|
||||
expect(registry.get('u-zero-byte', 'c-empty', empty.task.id)).toBeDefined();
|
||||
expect(registry.get('u-zero-byte', 'c-retained', retained.task.id)).toBeUndefined();
|
||||
expect(registry.get('u-zero-byte', 'c-incoming', incoming.task.id)?.artifact).toBeDefined();
|
||||
});
|
||||
|
||||
it('does not partially evict global tasks when user retention cannot be satisfied', () => {
|
||||
const registry = new BackgroundTaskRegistryClass();
|
||||
for (let index = 0; index < 400; index++) {
|
||||
const conversationId = `c-atomic-${Math.floor(index / 200)}`;
|
||||
const created = registry.create({
|
||||
userId: 'u-atomic',
|
||||
conversationId,
|
||||
toolCallId: `call_atomic_${index}`,
|
||||
toolName: 't',
|
||||
harvestStarted: true,
|
||||
});
|
||||
if ('atCapacity' in created) {
|
||||
throw new Error(`unexpected capacity at ${index}: ${created.scope}`);
|
||||
}
|
||||
registry.complete('u-atomic', conversationId, created.task.id, {
|
||||
content: '',
|
||||
harvestStarted: true,
|
||||
});
|
||||
}
|
||||
|
||||
let oldestGlobalTask: { userId: string; conversationId: string; taskId: string } | undefined;
|
||||
for (let userIndex = 0; userIndex < 8; userIndex++) {
|
||||
for (let taskIndex = 0; taskIndex < 200; taskIndex++) {
|
||||
const userId = `u-global-${userIndex}`;
|
||||
const conversationId = `c-global-${userIndex}`;
|
||||
const created = registry.create({
|
||||
userId,
|
||||
conversationId,
|
||||
toolCallId: `call_global_${userIndex}_${taskIndex}`,
|
||||
toolName: 't',
|
||||
});
|
||||
if ('atCapacity' in created) {
|
||||
throw new Error('unexpected capacity');
|
||||
}
|
||||
registry.complete(userId, conversationId, created.task.id, { content: '' });
|
||||
oldestGlobalTask ??= { userId, conversationId, taskId: created.task.id };
|
||||
}
|
||||
}
|
||||
|
||||
const rejected = registry.create({
|
||||
userId: 'u-atomic',
|
||||
conversationId: 'c-atomic-new',
|
||||
toolCallId: 'call_atomic_rejected',
|
||||
toolName: 't',
|
||||
});
|
||||
expect(rejected).toEqual({ atCapacity: true, scope: 'user_retention' });
|
||||
if (oldestGlobalTask == null) {
|
||||
throw new Error('expected a global eviction candidate');
|
||||
}
|
||||
expect(
|
||||
registry.get(
|
||||
oldestGlobalTask.userId,
|
||||
oldestGlobalTask.conversationId,
|
||||
oldestGlobalTask.taskId,
|
||||
),
|
||||
).toBeDefined();
|
||||
});
|
||||
|
||||
it('does not partially evict global payloads when user payload retention cannot be satisfied', () => {
|
||||
const registry = new BackgroundTaskRegistryClass();
|
||||
for (let index = 0; index < 2; index++) {
|
||||
const conversationId = `c-payload-protected-${index}`;
|
||||
const created = registry.create({
|
||||
userId: 'u-payload-atomic',
|
||||
conversationId,
|
||||
toolCallId: `call_payload_protected_${index}`,
|
||||
toolName: 't',
|
||||
harvestStarted: true,
|
||||
});
|
||||
if ('atCapacity' in created) {
|
||||
throw new Error('unexpected capacity');
|
||||
}
|
||||
registry.complete('u-payload-atomic', conversationId, created.task.id, {
|
||||
content: 'protected',
|
||||
artifact: { payload: 'p'.repeat(7_999_000) },
|
||||
harvestStarted: true,
|
||||
});
|
||||
}
|
||||
|
||||
let oldestGlobalTask: { userId: string; conversationId: string; taskId: string } | undefined;
|
||||
for (let index = 0; index < 6; index++) {
|
||||
const userId = `u-payload-global-${index}`;
|
||||
const conversationId = `c-payload-global-${index}`;
|
||||
const created = registry.create({
|
||||
userId,
|
||||
conversationId,
|
||||
toolCallId: `call_payload_global_${index}`,
|
||||
toolName: 't',
|
||||
});
|
||||
if ('atCapacity' in created) {
|
||||
throw new Error('unexpected capacity');
|
||||
}
|
||||
registry.complete(userId, conversationId, created.task.id, {
|
||||
content: 'global',
|
||||
artifact: { payload: 'g'.repeat(7_999_000) },
|
||||
});
|
||||
oldestGlobalTask ??= { userId, conversationId, taskId: created.task.id };
|
||||
}
|
||||
|
||||
const incoming = registry.create({
|
||||
userId: 'u-payload-atomic',
|
||||
conversationId: 'c-payload-incoming',
|
||||
toolCallId: 'call_payload_incoming',
|
||||
toolName: 't',
|
||||
});
|
||||
if ('atCapacity' in incoming) {
|
||||
throw new Error('unexpected capacity');
|
||||
}
|
||||
registry.complete('u-payload-atomic', 'c-payload-incoming', incoming.task.id, {
|
||||
content: 'incoming',
|
||||
artifact: { payload: 'i'.repeat(20_000) },
|
||||
});
|
||||
|
||||
expect(registry.get('u-payload-atomic', 'c-payload-incoming', incoming.task.id)?.result).toBe(
|
||||
undefined,
|
||||
);
|
||||
if (oldestGlobalTask == null) {
|
||||
throw new Error('expected a global payload eviction candidate');
|
||||
}
|
||||
expect(
|
||||
registry.get(
|
||||
oldestGlobalTask.userId,
|
||||
oldestGlobalTask.conversationId,
|
||||
oldestGlobalTask.taskId,
|
||||
),
|
||||
).toBeDefined();
|
||||
});
|
||||
|
||||
it('drops terminal errors when non-evictable payloads exhaust the user budget', () => {
|
||||
const registry = new BackgroundTaskRegistryClass();
|
||||
const first = registry.create({
|
||||
userId: 'u-error-budget',
|
||||
conversationId: 'c-first',
|
||||
toolCallId: 'call_first',
|
||||
toolName: 'execute_code',
|
||||
harvestStarted: true,
|
||||
});
|
||||
const second = registry.create({
|
||||
userId: 'u-error-budget',
|
||||
conversationId: 'c-second',
|
||||
toolCallId: 'call_second',
|
||||
toolName: 'execute_code',
|
||||
harvestStarted: true,
|
||||
});
|
||||
const failing = registry.create({
|
||||
userId: 'u-error-budget',
|
||||
conversationId: 'c-failing',
|
||||
toolCallId: 'call_failing',
|
||||
toolName: 'execute_code',
|
||||
});
|
||||
if ('atCapacity' in first || 'atCapacity' in second || 'atCapacity' in failing) {
|
||||
throw new Error('unexpected capacity');
|
||||
}
|
||||
registry.complete('u-error-budget', 'c-first', first.task.id, {
|
||||
content: 'a',
|
||||
artifact: { payload: 'a'.repeat(9_000_000) },
|
||||
harvestStarted: true,
|
||||
});
|
||||
registry.complete('u-error-budget', 'c-second', second.task.id, {
|
||||
content: 'b',
|
||||
artifact: { payload: 'b'.repeat(6_999_900) },
|
||||
harvestStarted: true,
|
||||
});
|
||||
registry.fail('u-error-budget', 'c-failing', failing.task.id, 'e'.repeat(100_000));
|
||||
const failed = registry.get('u-error-budget', 'c-failing', failing.task.id);
|
||||
expect(failed?.status).toBe('error');
|
||||
expect(failed?.error).toBeUndefined();
|
||||
});
|
||||
|
||||
it('finishes an empty harvest without evicting payload capacity', () => {
|
||||
const registry = new BackgroundTaskRegistryClass();
|
||||
const first = registry.create({
|
||||
userId: 'u-empty-harvest',
|
||||
conversationId: 'c-first',
|
||||
toolCallId: 'call_first',
|
||||
toolName: 'execute_code',
|
||||
});
|
||||
const second = registry.create({
|
||||
userId: 'u-empty-harvest',
|
||||
conversationId: 'c-second',
|
||||
toolCallId: 'call_second',
|
||||
toolName: 'execute_code',
|
||||
});
|
||||
const empty = registry.create({
|
||||
userId: 'u-empty-harvest',
|
||||
conversationId: 'c-empty',
|
||||
toolCallId: 'call_empty',
|
||||
toolName: 'execute_code',
|
||||
harvestStarted: true,
|
||||
});
|
||||
if ('atCapacity' in first || 'atCapacity' in second || 'atCapacity' in empty) {
|
||||
throw new Error('unexpected capacity');
|
||||
}
|
||||
registry.complete('u-empty-harvest', 'c-first', first.task.id, {
|
||||
content: 'a',
|
||||
artifact: { payload: 'a'.repeat(9_000_000) },
|
||||
});
|
||||
registry.complete('u-empty-harvest', 'c-second', second.task.id, {
|
||||
content: 'b',
|
||||
artifact: { payload: 'b'.repeat(6_999_970) },
|
||||
});
|
||||
registry.complete('u-empty-harvest', 'c-empty', empty.task.id, {
|
||||
content: '',
|
||||
harvestStarted: true,
|
||||
});
|
||||
registry.finishHarvest('u-empty-harvest', 'c-empty', empty.task.id);
|
||||
expect(registry.get('u-empty-harvest', 'c-first', first.task.id)).toBeDefined();
|
||||
expect(registry.get('u-empty-harvest', 'c-empty', empty.task.id)?.harvestPending).toBe(false);
|
||||
});
|
||||
|
||||
it('scopes tasks by user and conversation', () => {
|
||||
const registry = new BackgroundTaskRegistryClass();
|
||||
const created = registry.create({
|
||||
|
|
|
|||
|
|
@ -646,6 +646,8 @@ export interface BackgroundTask {
|
|||
* finalize and deadlocking that same generation. */
|
||||
liveArtifactPollRequired?: boolean;
|
||||
completionWakeup?: boolean;
|
||||
/** True while the terminal result is being persisted for automatic delivery. */
|
||||
completionPersistencePending?: boolean;
|
||||
/** Process-local cancellation handle for the preregistered durable delivery.
|
||||
* A same-generation manual claim retires it before exposing the result. */
|
||||
completionWakeupRetire?: BackgroundToolWakeupAdmission['retire'];
|
||||
|
|
@ -655,6 +657,8 @@ export interface BackgroundTask {
|
|||
}
|
||||
|
||||
interface TaskBucket {
|
||||
key: string;
|
||||
userId: string;
|
||||
tasks: Map<string, BackgroundTask>;
|
||||
/** toolCallId -> taskId, for dispatch idempotency across graph re-execution. */
|
||||
byToolCall: Map<string, string>;
|
||||
|
|
@ -665,18 +669,46 @@ interface TaskBucket {
|
|||
lastAccess: number;
|
||||
}
|
||||
|
||||
interface RetainedPayloadUsage {
|
||||
result: number;
|
||||
artifact: number;
|
||||
attachments: number;
|
||||
error: number;
|
||||
}
|
||||
|
||||
export interface BackgroundTaskCapacityPermit {
|
||||
id: string;
|
||||
userId: string;
|
||||
conversationId: string;
|
||||
}
|
||||
|
||||
export type BackgroundTaskCapacityScope =
|
||||
| 'conversation_running'
|
||||
| 'conversation_retention'
|
||||
| 'user_running'
|
||||
| 'user_retention'
|
||||
| 'global_running'
|
||||
| 'global_retention';
|
||||
|
||||
type BackgroundTaskCapacityRejection = {
|
||||
atCapacity: true;
|
||||
scope: BackgroundTaskCapacityScope;
|
||||
};
|
||||
|
||||
const COMPLETED_TASK_TTL_MS = 60 * 60 * 1000;
|
||||
const IDLE_BUCKET_TTL_MS = 6 * 60 * 60 * 1000;
|
||||
const MAX_RUNNING_PER_BUCKET = 10;
|
||||
const MAX_TASKS_PER_BUCKET = 200;
|
||||
/** Cross-conversation limits prevent one principal from multiplying the bucket allowance. */
|
||||
const MAX_RUNNING_PER_USER = 40;
|
||||
const MAX_RUNNING_GLOBAL = 200;
|
||||
const MAX_TASKS_PER_USER = 400;
|
||||
const MAX_TASKS_GLOBAL = 2_000;
|
||||
const MAX_RESULT_CHARS = 100_000;
|
||||
const MAX_ARTIFACT_CHARS = 10_000_000;
|
||||
/** JSON-character budgets bound large settled payloads independently of task metadata. */
|
||||
const MAX_RETAINED_CHARS_PER_USER = 16_000_000;
|
||||
const MAX_RETAINED_CHARS_GLOBAL = 64_000_000;
|
||||
const GLOBAL_SWEEP_INTERVAL_MS = 60 * 1000;
|
||||
|
||||
/**
|
||||
|
|
@ -720,31 +752,43 @@ function nextDispatchStamp(now: number): number {
|
|||
}
|
||||
|
||||
function toStoredContent(content: unknown): string {
|
||||
const asString = typeof content === 'string' ? content : JSON.stringify(content ?? '');
|
||||
const serialized = typeof content === 'string' ? content : JSON.stringify(content ?? '');
|
||||
const asString = serialized ?? String(content ?? '');
|
||||
return truncateMiddle(asString, MAX_RESULT_CHARS);
|
||||
}
|
||||
|
||||
/**
|
||||
* Bounds retained artifact memory: an artifact is held for up to the completed
|
||||
* TTL, so a runaway payload (huge base64 blobs) is dropped rather than pinned.
|
||||
* Measurement failures (circular refs) keep the artifact.
|
||||
* Measurement failures (such as circular references) drop the artifact because
|
||||
* an unmeasurable value cannot safely participate in the aggregate budget.
|
||||
*/
|
||||
function toStoredArtifact(taskId: string, artifact: unknown): unknown {
|
||||
function toStoredArtifact(
|
||||
taskId: string,
|
||||
artifact: unknown,
|
||||
): { artifact?: unknown; chars: number } {
|
||||
if (artifact == null) {
|
||||
return undefined;
|
||||
return { chars: 0 };
|
||||
}
|
||||
try {
|
||||
const size = JSON.stringify(artifact)?.length ?? 0;
|
||||
const serialized = JSON.stringify(artifact);
|
||||
if (serialized == null) {
|
||||
logger.warn(`[background] Dropping unmeasurable artifact for task ${taskId}.`);
|
||||
return { chars: 0 };
|
||||
}
|
||||
const size = serialized.length;
|
||||
if (size > MAX_ARTIFACT_CHARS) {
|
||||
logger.warn(
|
||||
`[background] Dropping oversized artifact for task ${taskId} (${size} chars > ${MAX_ARTIFACT_CHARS}).`,
|
||||
);
|
||||
return undefined;
|
||||
return { chars: 0 };
|
||||
}
|
||||
const storedArtifact: unknown = JSON.parse(serialized);
|
||||
return { artifact: storedArtifact, chars: size };
|
||||
} catch {
|
||||
/* unmeasurable artifact: keep it */
|
||||
logger.warn(`[background] Dropping unmeasurable artifact for task ${taskId}.`);
|
||||
return { chars: 0 };
|
||||
}
|
||||
return artifact;
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -758,6 +802,7 @@ function toStoredArtifact(taskId: string, artifact: unknown): unknown {
|
|||
*/
|
||||
export class BackgroundTaskRegistryClass {
|
||||
private readonly buckets = new Map<string, TaskBucket>();
|
||||
private readonly retainedUsage = new WeakMap<BackgroundTask, RetainedPayloadUsage>();
|
||||
private lastGlobalSweepAt = 0;
|
||||
|
||||
private key(userId: string, conversationId: string): string {
|
||||
|
|
@ -796,6 +841,9 @@ export class BackgroundTaskRegistryClass {
|
|||
continue;
|
||||
}
|
||||
this.sweepBucketTasks(bucket, now);
|
||||
if (bucket.tasks.size === 0 && bucket.capacityPermits.size === 0) {
|
||||
this.buckets.delete(bucketKey);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -804,6 +852,8 @@ export class BackgroundTaskRegistryClass {
|
|||
let bucket = this.buckets.get(bucketKey);
|
||||
if (!bucket) {
|
||||
bucket = {
|
||||
key: bucketKey,
|
||||
userId,
|
||||
tasks: new Map(),
|
||||
byToolCall: new Map(),
|
||||
capacityPermits: new Map(),
|
||||
|
|
@ -829,6 +879,235 @@ export class BackgroundTaskRegistryClass {
|
|||
return running;
|
||||
}
|
||||
|
||||
private payloadUsage(task: BackgroundTask): RetainedPayloadUsage {
|
||||
return (
|
||||
this.retainedUsage.get(task) ?? {
|
||||
result: 0,
|
||||
artifact: 0,
|
||||
attachments: 0,
|
||||
error: 0,
|
||||
}
|
||||
);
|
||||
}
|
||||
|
||||
private payloadChars(task: BackgroundTask): number {
|
||||
const usage = this.payloadUsage(task);
|
||||
return usage.result + usage.artifact + usage.attachments + usage.error;
|
||||
}
|
||||
|
||||
private updatePayloadUsage(task: BackgroundTask, patch: Partial<RetainedPayloadUsage>): void {
|
||||
this.retainedUsage.set(task, { ...this.payloadUsage(task), ...patch });
|
||||
}
|
||||
|
||||
private aggregateUsage(userId: string): {
|
||||
runningForUser: number;
|
||||
runningGlobal: number;
|
||||
tasksForUser: number;
|
||||
tasksGlobal: number;
|
||||
retainedForUser: number;
|
||||
retainedGlobal: number;
|
||||
} {
|
||||
let runningForUser = 0;
|
||||
let runningGlobal = 0;
|
||||
let tasksForUser = 0;
|
||||
let tasksGlobal = 0;
|
||||
let retainedForUser = 0;
|
||||
let retainedGlobal = 0;
|
||||
for (const bucket of this.buckets.values()) {
|
||||
const isUser = bucket.userId === userId;
|
||||
tasksGlobal += bucket.tasks.size + bucket.capacityPermits.size;
|
||||
if (isUser) {
|
||||
tasksForUser += bucket.tasks.size + bucket.capacityPermits.size;
|
||||
}
|
||||
for (const task of bucket.tasks.values()) {
|
||||
if (task.status === 'running') {
|
||||
runningGlobal++;
|
||||
if (isUser) {
|
||||
runningForUser++;
|
||||
}
|
||||
}
|
||||
const retained = this.payloadChars(task);
|
||||
retainedGlobal += retained;
|
||||
if (isUser) {
|
||||
retainedForUser += retained;
|
||||
}
|
||||
}
|
||||
runningGlobal += bucket.capacityPermits.size;
|
||||
if (isUser) {
|
||||
runningForUser += bucket.capacityPermits.size;
|
||||
}
|
||||
}
|
||||
return {
|
||||
runningForUser,
|
||||
runningGlobal,
|
||||
tasksForUser,
|
||||
tasksGlobal,
|
||||
retainedForUser,
|
||||
retainedGlobal,
|
||||
};
|
||||
}
|
||||
|
||||
private runningCapacityScope(userId: string): 'user_running' | 'global_running' | undefined {
|
||||
const usage = this.aggregateUsage(userId);
|
||||
if (usage.runningForUser >= MAX_RUNNING_PER_USER) {
|
||||
return 'user_running';
|
||||
}
|
||||
if (usage.runningGlobal >= MAX_RUNNING_GLOBAL) {
|
||||
return 'global_running';
|
||||
}
|
||||
return undefined;
|
||||
}
|
||||
|
||||
private settledCandidates(params: {
|
||||
userId?: string;
|
||||
bucket?: TaskBucket;
|
||||
excludeTask?: BackgroundTask;
|
||||
requirePayload?: boolean;
|
||||
}): Array<{ bucket: TaskBucket; task: BackgroundTask }> {
|
||||
const candidates: Array<{ bucket: TaskBucket; task: BackgroundTask }> = [];
|
||||
for (const bucket of this.buckets.values()) {
|
||||
if (params.bucket != null && bucket !== params.bucket) {
|
||||
continue;
|
||||
}
|
||||
if (params.userId != null && bucket.userId !== params.userId) {
|
||||
continue;
|
||||
}
|
||||
for (const task of bucket.tasks.values()) {
|
||||
if (
|
||||
task.status !== 'running' &&
|
||||
task.harvestPending !== true &&
|
||||
task.completionPersistencePending !== true &&
|
||||
task !== params.excludeTask &&
|
||||
(params.requirePayload !== true || this.payloadChars(task) > 0)
|
||||
) {
|
||||
candidates.push({ bucket, task });
|
||||
}
|
||||
}
|
||||
}
|
||||
return candidates.sort((a, b) => a.task.updatedAt - b.task.updatedAt);
|
||||
}
|
||||
|
||||
private evictSelected(selected: Map<BackgroundTask, TaskBucket>): void {
|
||||
const touched = new Set<TaskBucket>();
|
||||
for (const [task, bucket] of selected) {
|
||||
bucket.tasks.delete(task.id);
|
||||
touched.add(bucket);
|
||||
}
|
||||
const now = Date.now();
|
||||
for (const bucket of touched) {
|
||||
this.sweepBucketTasks(bucket, now);
|
||||
if (bucket.tasks.size === 0 && bucket.capacityPermits.size === 0) {
|
||||
this.buckets.delete(bucket.key);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private makeTaskRoom(userId: string, bucket?: TaskBucket): boolean {
|
||||
const usage = this.aggregateUsage(userId);
|
||||
const bucketRequired = Math.max(
|
||||
0,
|
||||
(bucket?.tasks.size ?? 0) + (bucket?.capacityPermits.size ?? 0) - MAX_TASKS_PER_BUCKET + 1,
|
||||
);
|
||||
const userRequired = Math.max(0, usage.tasksForUser - MAX_TASKS_PER_USER + 1);
|
||||
const globalRequired = Math.max(0, usage.tasksGlobal - MAX_TASKS_GLOBAL + 1);
|
||||
const selected = new Map<BackgroundTask, TaskBucket>();
|
||||
|
||||
const selectCount = (
|
||||
candidates: Array<{ bucket: TaskBucket; task: BackgroundTask }>,
|
||||
required: number,
|
||||
): boolean => {
|
||||
let remaining = required;
|
||||
for (const candidate of candidates) {
|
||||
if (remaining <= 0) {
|
||||
break;
|
||||
}
|
||||
if (selected.has(candidate.task)) {
|
||||
continue;
|
||||
}
|
||||
selected.set(candidate.task, candidate.bucket);
|
||||
remaining--;
|
||||
}
|
||||
return remaining === 0;
|
||||
};
|
||||
|
||||
if (bucket != null && !selectCount(this.settledCandidates({ bucket }), bucketRequired)) {
|
||||
return false;
|
||||
}
|
||||
const selectedForUser = [...selected.values()].filter(
|
||||
(selectedBucket) => selectedBucket.userId === userId,
|
||||
).length;
|
||||
if (
|
||||
!selectCount(this.settledCandidates({ userId }), Math.max(0, userRequired - selectedForUser))
|
||||
) {
|
||||
return false;
|
||||
}
|
||||
if (!selectCount(this.settledCandidates({}), Math.max(0, globalRequired - selected.size))) {
|
||||
return false;
|
||||
}
|
||||
this.evictSelected(selected);
|
||||
return true;
|
||||
}
|
||||
|
||||
private taskCapacityScope(
|
||||
userId: string,
|
||||
bucket?: TaskBucket,
|
||||
): 'conversation_retention' | 'user_retention' | 'global_retention' {
|
||||
if (bucket != null && bucket.tasks.size + bucket.capacityPermits.size >= MAX_TASKS_PER_BUCKET) {
|
||||
return 'conversation_retention';
|
||||
}
|
||||
return this.aggregateUsage(userId).tasksForUser >= MAX_TASKS_PER_USER
|
||||
? 'user_retention'
|
||||
: 'global_retention';
|
||||
}
|
||||
|
||||
private makeRetainedRoom(userId: string, task: BackgroundTask, chars: number): boolean {
|
||||
const usage = this.aggregateUsage(userId);
|
||||
const userRequired = Math.max(0, usage.retainedForUser + chars - MAX_RETAINED_CHARS_PER_USER);
|
||||
const globalRequired = Math.max(0, usage.retainedGlobal + chars - MAX_RETAINED_CHARS_GLOBAL);
|
||||
const selected = new Map<BackgroundTask, TaskBucket>();
|
||||
|
||||
const selectChars = (
|
||||
candidates: Array<{ bucket: TaskBucket; task: BackgroundTask }>,
|
||||
required: number,
|
||||
): boolean => {
|
||||
let retained = 0;
|
||||
for (const candidate of candidates) {
|
||||
if (retained >= required) {
|
||||
break;
|
||||
}
|
||||
if (selected.has(candidate.task)) {
|
||||
continue;
|
||||
}
|
||||
selected.set(candidate.task, candidate.bucket);
|
||||
retained += this.payloadChars(candidate.task);
|
||||
}
|
||||
return retained >= required;
|
||||
};
|
||||
|
||||
if (
|
||||
!selectChars(
|
||||
this.settledCandidates({ userId, excludeTask: task, requirePayload: true }),
|
||||
userRequired,
|
||||
)
|
||||
) {
|
||||
return false;
|
||||
}
|
||||
let selectedChars = 0;
|
||||
for (const selectedTask of selected.keys()) {
|
||||
selectedChars += this.payloadChars(selectedTask);
|
||||
}
|
||||
if (
|
||||
!selectChars(
|
||||
this.settledCandidates({ excludeTask: task, requirePayload: true }),
|
||||
Math.max(0, globalRequired - selectedChars),
|
||||
)
|
||||
) {
|
||||
return false;
|
||||
}
|
||||
this.evictSelected(selected);
|
||||
return true;
|
||||
}
|
||||
|
||||
/** Acquires process-local capacity before a caller persists launch authority.
|
||||
* The synchronous permit closes the capacity-rejection crash window without
|
||||
* making ordinary background tasks durable. */
|
||||
|
|
@ -841,20 +1120,37 @@ export class BackgroundTaskRegistryClass {
|
|||
}):
|
||||
| { permit: BackgroundTaskCapacityPermit }
|
||||
| { task: BackgroundTask; isNew: false }
|
||||
| { atCapacity: true } {
|
||||
| BackgroundTaskCapacityRejection {
|
||||
const now = Date.now();
|
||||
this.sweep(now);
|
||||
const bucket = this.getBucket(params.userId, params.conversationId, now);
|
||||
this.sweepBucketTasks(bucket, now);
|
||||
const bucketKey = this.key(params.userId, params.conversationId);
|
||||
const existingBucket = this.buckets.get(bucketKey);
|
||||
if (existingBucket != null) {
|
||||
existingBucket.lastAccess = now;
|
||||
this.sweepBucketTasks(existingBucket, now);
|
||||
}
|
||||
const dedupeKey = this.dedupeKey(params);
|
||||
const existingId = bucket.byToolCall.get(dedupeKey);
|
||||
const existing = existingId == null ? undefined : bucket.tasks.get(existingId);
|
||||
const existingId = existingBucket?.byToolCall.get(dedupeKey);
|
||||
const existing = existingId == null ? undefined : existingBucket?.tasks.get(existingId);
|
||||
if (existing != null) {
|
||||
return { task: existing, isNew: false };
|
||||
}
|
||||
if (this.runningCount(bucket) + bucket.capacityPermits.size >= MAX_RUNNING_PER_BUCKET) {
|
||||
return { atCapacity: true };
|
||||
if (
|
||||
existingBucket != null &&
|
||||
this.runningCount(existingBucket) + existingBucket.capacityPermits.size >=
|
||||
MAX_RUNNING_PER_BUCKET
|
||||
) {
|
||||
return { atCapacity: true, scope: 'conversation_running' };
|
||||
}
|
||||
const runningScope = this.runningCapacityScope(params.userId);
|
||||
if (runningScope != null) {
|
||||
return { atCapacity: true, scope: runningScope };
|
||||
}
|
||||
if (!this.makeTaskRoom(params.userId, existingBucket)) {
|
||||
return { atCapacity: true, scope: this.taskCapacityScope(params.userId, existingBucket) };
|
||||
}
|
||||
const bucket =
|
||||
this.buckets.get(bucketKey) ?? this.getBucket(params.userId, params.conversationId, now);
|
||||
const permit: BackgroundTaskCapacityPermit = {
|
||||
id: randomUUID(),
|
||||
userId: params.userId,
|
||||
|
|
@ -869,9 +1165,12 @@ export class BackgroundTaskRegistryClass {
|
|||
}
|
||||
|
||||
releaseCapacity(permit: BackgroundTaskCapacityPermit): void {
|
||||
this.buckets
|
||||
.get(this.key(permit.userId, permit.conversationId))
|
||||
?.capacityPermits.delete(permit.id);
|
||||
const bucketKey = this.key(permit.userId, permit.conversationId);
|
||||
const bucket = this.buckets.get(bucketKey);
|
||||
bucket?.capacityPermits.delete(permit.id);
|
||||
if (bucket?.tasks.size === 0 && bucket.capacityPermits.size === 0) {
|
||||
this.buckets.delete(bucketKey);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -900,16 +1199,20 @@ export class BackgroundTaskRegistryClass {
|
|||
harvestStarted?: boolean;
|
||||
liveArtifactPollRequired?: boolean;
|
||||
capacityPermit?: BackgroundTaskCapacityPermit;
|
||||
}): { task: BackgroundTask; isNew: boolean } | { atCapacity: true } {
|
||||
}): { task: BackgroundTask; isNew: boolean } | BackgroundTaskCapacityRejection {
|
||||
const now = Date.now();
|
||||
this.sweep(now);
|
||||
const bucket = this.getBucket(params.userId, params.conversationId, now);
|
||||
this.sweepBucketTasks(bucket, now);
|
||||
const bucketKey = this.key(params.userId, params.conversationId);
|
||||
const existingBucket = this.buckets.get(bucketKey);
|
||||
if (existingBucket != null) {
|
||||
existingBucket.lastAccess = now;
|
||||
this.sweepBucketTasks(existingBucket, now);
|
||||
}
|
||||
|
||||
const dedupeKey = this.dedupeKey(params);
|
||||
const existingId = bucket.byToolCall.get(dedupeKey);
|
||||
const existingId = existingBucket?.byToolCall.get(dedupeKey);
|
||||
if (existingId) {
|
||||
const existing = bucket.tasks.get(existingId);
|
||||
const existing = existingBucket?.tasks.get(existingId);
|
||||
if (existing) {
|
||||
if (params.capacityPermit != null) {
|
||||
this.releaseCapacity(params.capacityPermit);
|
||||
|
|
@ -919,6 +1222,10 @@ export class BackgroundTaskRegistryClass {
|
|||
}
|
||||
|
||||
if (params.capacityPermit != null) {
|
||||
if (existingBucket == null) {
|
||||
throw new Error('Background task capacity permit is stale');
|
||||
}
|
||||
const bucket = existingBucket;
|
||||
const permit = bucket.capacityPermits.get(params.capacityPermit.id);
|
||||
if (
|
||||
params.capacityPermit.userId !== params.userId ||
|
||||
|
|
@ -930,30 +1237,26 @@ export class BackgroundTaskRegistryClass {
|
|||
bucket.capacityPermits.delete(params.capacityPermit.id);
|
||||
}
|
||||
/** Only *running* tasks gate dispatch. */
|
||||
if (
|
||||
params.capacityPermit == null &&
|
||||
this.runningCount(bucket) + bucket.capacityPermits.size >= MAX_RUNNING_PER_BUCKET
|
||||
) {
|
||||
return { atCapacity: true };
|
||||
}
|
||||
/** The total-tasks cap bounds memory but must NOT block new work: evict the
|
||||
* oldest SETTLED tasks to make room rather than rejecting (settled tasks
|
||||
* aren't removed by polling, so 200 quick calls would otherwise block for
|
||||
* up to the completed-TTL). Running is already capped, so room always frees. */
|
||||
if (bucket.tasks.size >= MAX_TASKS_PER_BUCKET) {
|
||||
const settledOldestFirst = [...bucket.tasks.values()]
|
||||
.filter((t) => t.status !== 'running')
|
||||
.sort((a, b) => a.createdAt - b.createdAt);
|
||||
let toEvict = bucket.tasks.size - MAX_TASKS_PER_BUCKET + 1;
|
||||
for (const stale of settledOldestFirst) {
|
||||
if (toEvict <= 0) {
|
||||
break;
|
||||
}
|
||||
bucket.tasks.delete(stale.id);
|
||||
toEvict--;
|
||||
if (params.capacityPermit == null) {
|
||||
if (
|
||||
existingBucket != null &&
|
||||
this.runningCount(existingBucket) + existingBucket.capacityPermits.size >=
|
||||
MAX_RUNNING_PER_BUCKET
|
||||
) {
|
||||
return { atCapacity: true, scope: 'conversation_running' };
|
||||
}
|
||||
const runningScope = this.runningCapacityScope(params.userId);
|
||||
if (runningScope != null) {
|
||||
return { atCapacity: true, scope: runningScope };
|
||||
}
|
||||
}
|
||||
|
||||
if (!this.makeTaskRoom(params.userId, existingBucket)) {
|
||||
return { atCapacity: true, scope: this.taskCapacityScope(params.userId, existingBucket) };
|
||||
}
|
||||
/** Aggregate eviction may have removed `existingBucket` when its only
|
||||
* settled task was the oldest candidate. Never register into that detached map. */
|
||||
const bucket =
|
||||
this.buckets.get(bucketKey) ?? this.getBucket(params.userId, params.conversationId, now);
|
||||
const task: BackgroundTask = {
|
||||
id: params.taskId ?? randomUUID(),
|
||||
toolName: params.toolName,
|
||||
|
|
@ -977,13 +1280,14 @@ export class BackgroundTaskRegistryClass {
|
|||
conversationId: string,
|
||||
taskId: string,
|
||||
patch: Partial<BackgroundTask>,
|
||||
): void {
|
||||
): boolean {
|
||||
const bucket = this.buckets.get(this.key(userId, conversationId));
|
||||
const task = bucket?.tasks.get(taskId);
|
||||
if (!task || (task.artifactBlocked === true && patch.artifactBlocked !== true)) {
|
||||
return;
|
||||
return false;
|
||||
}
|
||||
Object.assign(task, patch, { updatedAt: Date.now() });
|
||||
return true;
|
||||
}
|
||||
|
||||
complete(
|
||||
|
|
@ -993,15 +1297,39 @@ export class BackgroundTaskRegistryClass {
|
|||
result: { content: unknown; artifact?: unknown; harvestStarted?: boolean },
|
||||
): string {
|
||||
const storedContent = toStoredContent(result.content);
|
||||
this.update(userId, conversationId, taskId, {
|
||||
const task = this.buckets.get(this.key(userId, conversationId))?.tasks.get(taskId);
|
||||
if (task == null || task.status !== 'running' || task.artifactBlocked === true) {
|
||||
return storedContent;
|
||||
}
|
||||
const storedArtifact = toStoredArtifact(taskId, result.artifact);
|
||||
const usage = this.payloadUsage(task);
|
||||
const desiredChars = storedContent.length + storedArtifact.chars;
|
||||
const currentChars = usage.result + usage.artifact;
|
||||
const hasRetainedCapacity = this.makeRetainedRoom(
|
||||
userId,
|
||||
task,
|
||||
Math.max(0, desiredChars - currentChars),
|
||||
);
|
||||
const retainedContent = hasRetainedCapacity ? storedContent : undefined;
|
||||
const artifact = hasRetainedCapacity ? storedArtifact.artifact : undefined;
|
||||
const artifactChars = hasRetainedCapacity ? storedArtifact.chars : 0;
|
||||
const updated = this.update(userId, conversationId, taskId, {
|
||||
status: 'completed',
|
||||
result: storedContent,
|
||||
artifact: toStoredArtifact(taskId, result.artifact),
|
||||
result: retainedContent,
|
||||
artifact,
|
||||
error: undefined,
|
||||
...(result.harvestStarted === true ? { harvestStarted: true, harvestPending: true } : {}),
|
||||
/** Marks that an artifact existed even after `claimArtifact` clears it,
|
||||
* so re-polls keep the "produced an artifact" note. */
|
||||
artifactDelivered: false,
|
||||
});
|
||||
if (updated) {
|
||||
this.updatePayloadUsage(task, {
|
||||
result: retainedContent?.length ?? 0,
|
||||
artifact: artifactChars,
|
||||
error: 0,
|
||||
});
|
||||
}
|
||||
return storedContent;
|
||||
}
|
||||
|
||||
|
|
@ -1020,7 +1348,18 @@ export class BackgroundTaskRegistryClass {
|
|||
if (attachments.length === 0) {
|
||||
return;
|
||||
}
|
||||
this.update(userId, conversationId, taskId, { attachments });
|
||||
const task = this.buckets.get(this.key(userId, conversationId))?.tasks.get(taskId);
|
||||
if (task == null || task.status !== 'completed' || task.artifactBlocked === true) {
|
||||
return;
|
||||
}
|
||||
const measured = toStoredArtifact(taskId, attachments);
|
||||
const additionalChars = Math.max(0, measured.chars - this.payloadUsage(task).attachments);
|
||||
if (measured.artifact == null || !this.makeRetainedRoom(userId, task, additionalChars)) {
|
||||
return;
|
||||
}
|
||||
if (this.update(userId, conversationId, taskId, { attachments })) {
|
||||
this.updatePayloadUsage(task, { attachments: measured.chars });
|
||||
}
|
||||
}
|
||||
|
||||
/** Marks completion-time inspection/persistence successful, unlocking artifact collection. */
|
||||
|
|
@ -1030,10 +1369,25 @@ export class BackgroundTaskRegistryClass {
|
|||
taskId: string,
|
||||
attachments: unknown[] = [],
|
||||
): void {
|
||||
this.update(userId, conversationId, taskId, {
|
||||
const task = this.buckets.get(this.key(userId, conversationId))?.tasks.get(taskId);
|
||||
if (task == null || task.status !== 'completed' || task.artifactBlocked === true) {
|
||||
return;
|
||||
}
|
||||
if (attachments.length === 0) {
|
||||
this.update(userId, conversationId, taskId, { harvestPending: false });
|
||||
return;
|
||||
}
|
||||
const measured = toStoredArtifact(taskId, attachments);
|
||||
const additionalChars = Math.max(0, measured.chars - this.payloadUsage(task).attachments);
|
||||
const canStoreAttachments =
|
||||
measured.artifact != null && this.makeRetainedRoom(userId, task, additionalChars);
|
||||
const updated = this.update(userId, conversationId, taskId, {
|
||||
harvestPending: false,
|
||||
...(attachments.length > 0 ? { attachments } : {}),
|
||||
...(attachments.length > 0 && canStoreAttachments ? { attachments } : {}),
|
||||
});
|
||||
if (updated && attachments.length > 0 && canStoreAttachments) {
|
||||
this.updatePayloadUsage(task, { attachments: measured.chars });
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -1074,6 +1428,7 @@ export class BackgroundTaskRegistryClass {
|
|||
const artifact = task.artifact;
|
||||
task.artifactDelivered = true;
|
||||
task.artifact = undefined;
|
||||
this.updatePayloadUsage(task, { artifact: 0 });
|
||||
return {
|
||||
toolName: task.toolName,
|
||||
toolCallId: task.toolCallId,
|
||||
|
|
@ -1093,13 +1448,27 @@ export class BackgroundTaskRegistryClass {
|
|||
restoreArtifact(userId: string, conversationId: string, taskId: string, artifact: unknown): void {
|
||||
const bucket = this.buckets.get(this.key(userId, conversationId));
|
||||
const task = bucket?.tasks.get(taskId);
|
||||
if (!task || task.artifactBlocked === true || task.artifact != null) {
|
||||
if (
|
||||
!task ||
|
||||
task.status !== 'completed' ||
|
||||
task.artifactBlocked === true ||
|
||||
task.artifact != null
|
||||
) {
|
||||
return;
|
||||
}
|
||||
/** Same size bound as `complete()` — a restore path must not resurrect
|
||||
* an artifact the memory cap already discarded. */
|
||||
task.artifact = toStoredArtifact(taskId, artifact);
|
||||
const storedArtifact = toStoredArtifact(taskId, artifact);
|
||||
if (
|
||||
storedArtifact.artifact == null ||
|
||||
!this.makeRetainedRoom(userId, task, storedArtifact.chars)
|
||||
) {
|
||||
return;
|
||||
}
|
||||
task.artifact = storedArtifact.artifact;
|
||||
task.artifactDelivered = false;
|
||||
this.updatePayloadUsage(task, { artifact: storedArtifact.chars });
|
||||
task.updatedAt = Date.now();
|
||||
}
|
||||
|
||||
fail(
|
||||
|
|
@ -1109,11 +1478,29 @@ export class BackgroundTaskRegistryClass {
|
|||
error: string,
|
||||
options?: { harvestStarted?: boolean },
|
||||
): void {
|
||||
this.update(userId, conversationId, taskId, {
|
||||
const storedError = truncateMiddle(error, MAX_RESULT_CHARS);
|
||||
const task = this.buckets.get(this.key(userId, conversationId))?.tasks.get(taskId);
|
||||
if (task == null || task.status !== 'running' || task.artifactBlocked === true) {
|
||||
return;
|
||||
}
|
||||
const hasRetainedCapacity = this.makeRetainedRoom(userId, task, storedError.length);
|
||||
const retainedError = hasRetainedCapacity ? storedError : undefined;
|
||||
const updated = this.update(userId, conversationId, taskId, {
|
||||
status: 'error',
|
||||
error,
|
||||
error: retainedError,
|
||||
result: undefined,
|
||||
artifact: undefined,
|
||||
attachments: undefined,
|
||||
...(options?.harvestStarted === true ? { harvestStarted: true, harvestPending: true } : {}),
|
||||
});
|
||||
if (updated) {
|
||||
this.retainedUsage.set(task, {
|
||||
result: 0,
|
||||
artifact: 0,
|
||||
attachments: 0,
|
||||
error: retainedError?.length ?? 0,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
markCompletionWakeup(
|
||||
|
|
@ -1128,8 +1515,17 @@ export class BackgroundTaskRegistryClass {
|
|||
});
|
||||
}
|
||||
|
||||
markCompletionPersistencePending(userId: string, conversationId: string, taskId: string): void {
|
||||
this.update(userId, conversationId, taskId, { completionPersistencePending: true });
|
||||
}
|
||||
|
||||
markCompletionPersistenceFinished(userId: string, conversationId: string, taskId: string): void {
|
||||
this.update(userId, conversationId, taskId, { completionPersistencePending: undefined });
|
||||
}
|
||||
|
||||
markCompletionPersistenceFailed(userId: string, conversationId: string, taskId: string): void {
|
||||
this.update(userId, conversationId, taskId, {
|
||||
completionPersistencePending: undefined,
|
||||
completionPersistenceFailed: true,
|
||||
completionWakeupRetire: undefined,
|
||||
});
|
||||
|
|
@ -1189,9 +1585,20 @@ export class BackgroundTaskRegistryClass {
|
|||
|
||||
/** Permanently removes a policy-rejected artifact and exposes only the raw-free policy error. */
|
||||
blockArtifact(userId: string, conversationId: string, taskId: string, error: string): void {
|
||||
this.update(userId, conversationId, taskId, {
|
||||
const task = this.buckets.get(this.key(userId, conversationId))?.tasks.get(taskId);
|
||||
if (task == null) {
|
||||
return;
|
||||
}
|
||||
const storedError = truncateMiddle(error, MAX_RESULT_CHARS);
|
||||
const hasRetainedCapacity = this.makeRetainedRoom(
|
||||
userId,
|
||||
task,
|
||||
Math.max(0, storedError.length - this.payloadChars(task)),
|
||||
);
|
||||
const retainedError = hasRetainedCapacity ? storedError : undefined;
|
||||
const updated = this.update(userId, conversationId, taskId, {
|
||||
status: 'error',
|
||||
error,
|
||||
error: retainedError,
|
||||
result: undefined,
|
||||
artifact: undefined,
|
||||
attachments: undefined,
|
||||
|
|
@ -1200,6 +1607,14 @@ export class BackgroundTaskRegistryClass {
|
|||
artifactDelivered: false,
|
||||
artifactBlocked: true,
|
||||
});
|
||||
if (updated) {
|
||||
this.retainedUsage.set(task, {
|
||||
result: 0,
|
||||
artifact: 0,
|
||||
attachments: 0,
|
||||
error: retainedError?.length ?? 0,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -1217,8 +1632,7 @@ export class BackgroundTaskRegistryClass {
|
|||
task.harvestStarted = undefined;
|
||||
task.harvestPending = undefined;
|
||||
if (task.artifact == null && artifact != null) {
|
||||
task.artifact = artifact;
|
||||
task.artifactDelivered = false;
|
||||
this.restoreArtifact(userId, conversationId, taskId, artifact);
|
||||
}
|
||||
task.updatedAt = Date.now();
|
||||
}
|
||||
|
|
@ -1234,6 +1648,10 @@ export class BackgroundTaskRegistryClass {
|
|||
}
|
||||
bucket.lastAccess = now;
|
||||
this.sweepBucketTasks(bucket, now);
|
||||
if (bucket.tasks.size === 0 && bucket.capacityPermits.size === 0) {
|
||||
this.buckets.delete(bucket.key);
|
||||
return undefined;
|
||||
}
|
||||
return bucket.tasks.get(taskId);
|
||||
}
|
||||
|
||||
|
|
@ -1246,6 +1664,10 @@ export class BackgroundTaskRegistryClass {
|
|||
}
|
||||
bucket.lastAccess = now;
|
||||
this.sweepBucketTasks(bucket, now);
|
||||
if (bucket.tasks.size === 0 && bucket.capacityPermits.size === 0) {
|
||||
this.buckets.delete(bucket.key);
|
||||
return [];
|
||||
}
|
||||
return [...bucket.tasks.values()].sort((a, b) => a.createdAt - b.createdAt);
|
||||
}
|
||||
}
|
||||
|
|
@ -1273,12 +1695,30 @@ export function buildBackgroundHandleContent(
|
|||
});
|
||||
}
|
||||
|
||||
/** Content returned when the per-conversation background running cap is hit. */
|
||||
export function buildBackgroundCapacityContent(toolName: string): string {
|
||||
/** Content returned when a background registry capacity limit is hit. */
|
||||
export function buildBackgroundCapacityContent(
|
||||
toolName: string,
|
||||
scope: BackgroundTaskCapacityScope = 'conversation_running',
|
||||
): string {
|
||||
let message: string;
|
||||
if (scope === 'user_running') {
|
||||
message = `Too many background tasks are already active for this user (limit ${MAX_RUNNING_PER_USER}, including pending launch reservations). Wait for existing background work to settle, or run this call in the foreground.`;
|
||||
} else if (scope === 'user_retention') {
|
||||
message = `This user is retaining the maximum number of background tasks (${MAX_TASKS_PER_USER}), and pending result processing prevents safe eviction. Wait for background result processing to finish, or run this call in the foreground.`;
|
||||
} else if (scope === 'global_running') {
|
||||
message = `The server-wide background task registry is at capacity (running limit ${MAX_RUNNING_GLOBAL}). Retry later, or run this call in the foreground.`;
|
||||
} else if (scope === 'global_retention') {
|
||||
message = `The server-wide background task registry is retaining its maximum number of tasks (${MAX_TASKS_GLOBAL}), and pending result processing prevents safe eviction. Retry later, or run this call in the foreground.`;
|
||||
} else if (scope === 'conversation_retention') {
|
||||
message = `This conversation is retaining the maximum number of background tasks (${MAX_TASKS_PER_BUCKET}), and pending result processing prevents safe eviction. Wait for background result processing to finish, or run this call in the foreground.`;
|
||||
} else {
|
||||
message = `Too many background tasks are already running in this conversation (limit ${MAX_RUNNING_PER_BUCKET}). Poll ${CHECK_BACKGROUND_TASK_NAME} to collect finished results before dispatching more, or run this call in the foreground.`;
|
||||
}
|
||||
return JSON.stringify({
|
||||
status: 'rejected',
|
||||
tool: toolName,
|
||||
message: `Too many background tasks are already running in this conversation (limit ${MAX_RUNNING_PER_BUCKET}). Poll ${CHECK_BACKGROUND_TASK_NAME} to collect finished results before dispatching more, or run this call in the foreground.`,
|
||||
scope,
|
||||
message,
|
||||
});
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -4584,7 +4584,7 @@ export function createToolExecuteHandler(options: ToolExecuteOptions): EventHand
|
|||
return {
|
||||
toolCallId: tc.id,
|
||||
status: 'success' as const,
|
||||
content: buildBackgroundCapacityContent(tc.name),
|
||||
content: buildBackgroundCapacityContent(tc.name, capacityAdmission.scope),
|
||||
};
|
||||
}
|
||||
const capacityPermit =
|
||||
|
|
@ -4665,7 +4665,7 @@ export function createToolExecuteHandler(options: ToolExecuteOptions): EventHand
|
|||
return {
|
||||
toolCallId: tc.id,
|
||||
status: 'success' as const,
|
||||
content: buildBackgroundCapacityContent(tc.name),
|
||||
content: buildBackgroundCapacityContent(tc.name, created.scope),
|
||||
};
|
||||
}
|
||||
const { task, isNew } = created;
|
||||
|
|
@ -4928,6 +4928,30 @@ export function createToolExecuteHandler(options: ToolExecuteOptions): EventHand
|
|||
);
|
||||
}
|
||||
};
|
||||
const persistSettledBackgroundResult = async (params: {
|
||||
output?: string;
|
||||
artifact?: unknown;
|
||||
status: 'completed' | 'error';
|
||||
}): Promise<void> => {
|
||||
if (harvestEnabled) {
|
||||
await persistBackgroundResult(params);
|
||||
return;
|
||||
}
|
||||
backgroundTaskRegistry.markCompletionPersistencePending(
|
||||
backgroundUserId,
|
||||
backgroundConversationId,
|
||||
task.id,
|
||||
);
|
||||
try {
|
||||
await persistBackgroundResult(params);
|
||||
} finally {
|
||||
backgroundTaskRegistry.markCompletionPersistenceFinished(
|
||||
backgroundUserId,
|
||||
backgroundConversationId,
|
||||
task.id,
|
||||
);
|
||||
}
|
||||
};
|
||||
let invokePromise: Promise<{ content?: unknown; artifact?: unknown }>;
|
||||
const backgroundAbortController = new AbortController();
|
||||
try {
|
||||
|
|
@ -5105,7 +5129,10 @@ export function createToolExecuteHandler(options: ToolExecuteOptions): EventHand
|
|||
registryError,
|
||||
{ harvestStarted: harvestEnabled },
|
||||
);
|
||||
await persistBackgroundResult({ output: errorOutput, status: 'error' });
|
||||
await persistSettledBackgroundResult({
|
||||
output: errorOutput,
|
||||
status: 'error',
|
||||
});
|
||||
await wakeDetachedActor();
|
||||
return;
|
||||
}
|
||||
|
|
@ -5162,7 +5189,7 @@ export function createToolExecuteHandler(options: ToolExecuteOptions): EventHand
|
|||
task.id,
|
||||
{ content, artifact: result.artifact, harvestStarted: harvestEnabled },
|
||||
);
|
||||
await persistBackgroundResult({
|
||||
await persistSettledBackgroundResult({
|
||||
/** Use the registry's canonical bounded serialization so
|
||||
* structured content cannot leave the durable card on its
|
||||
* synthetic running handle. */
|
||||
|
|
@ -5211,7 +5238,10 @@ export function createToolExecuteHandler(options: ToolExecuteOptions): EventHand
|
|||
* the dispatch card on the handle JSON forever. */
|
||||
{ harvestStarted: harvestEnabled },
|
||||
);
|
||||
await persistBackgroundResult({ output: deliveredError, status: 'error' });
|
||||
await persistSettledBackgroundResult({
|
||||
output: deliveredError,
|
||||
status: 'error',
|
||||
});
|
||||
await wakeDetachedActor();
|
||||
} finally {
|
||||
if (producerRetirementTimeout != null) {
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue