diff --git a/packages/api/src/agents/background.spec.ts b/packages/api/src/agents/background.spec.ts index 6e6fc52fff..1a0fc9460c 100644 --- a/packages/api/src/agents/background.spec.ts +++ b/packages/api/src/agents/background.spec.ts @@ -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({ diff --git a/packages/api/src/agents/background.ts b/packages/api/src/agents/background.ts index 6b33da66ea..776cc53ad0 100644 --- a/packages/api/src/agents/background.ts +++ b/packages/api/src/agents/background.ts @@ -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; /** toolCallId -> taskId, for dispatch idempotency across graph re-execution. */ byToolCall: Map; @@ -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(); + private readonly retainedUsage = new WeakMap(); 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): 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): void { + const touched = new Set(); + 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(); + + 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(); + + 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, - ): 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, }); } diff --git a/packages/api/src/agents/handlers.ts b/packages/api/src/agents/handlers.ts index 23ddeacd9e..bff8bfe79a 100644 --- a/packages/api/src/agents/handlers.ts +++ b/packages/api/src/agents/handlers.ts @@ -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 => { + 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) {