Skip to content

Commit 5e3c680

Browse files
danny-avilaclaude
andauthored
🪃 feat: Wake Parent Agents on Child Completion (#14975)
* feat: wake parent agents on child completion * wip: harden child completion wakeup lifecycle * fix: close the completion-wakeup static failures Type the durable-claim store fixture, the continue-envelope test helper, and the terminal message's task metadata so the wakeup suites compile against the shapes they actually exercise. Replace `Array.prototype.at`, which the package target library does not provide. Capture the prepared child thread in a non-optional local before the provider callback closes over it, and narrow the trigger envelope itself on `mode === 'continue'` rather than a separately copied mode, so reading the continue target is sound. Lift the parent-message fallback out of a nested ternary into a named resolver. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01H1cCMDrTWaRNkmtKjpWELZ * test: cover the active-predecessor admission fence The Redis job-creation call gained a thirteenth scalar argument, so the spec helper reconstructed the HSET pairs one slot early and rebuilt an invalid job hash; three creation tests failed on that alone. Give the fence itself direct coverage in both store adapters, which it had none of despite deciding whether an automatic continuation may replace a live parent turn. Each proves a running and a requires_action predecessor are refused with the state a controller needs for a finite 409, that an absent or settled predecessor is admitted, and that an ordinary user turn without the policy still replaces its predecessor. The Redis case also asserts a refused continuation leaves the parent's durable job and chunks untouched. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01H1cCMDrTWaRNkmtKjpWELZ * fix: harden completion wakeup rollout and claims * fix: close completion wakeup race windows * test: keep the child store fixture exact * fix: close final subagent wakeup gaps * fix: preserve ambiguous completion claims * fix: release pre-admission wakeup claims * fix: stabilize subagent completion recovery --------- Co-authored-by: Claude <noreply@anthropic.com>
1 parent d175741 commit 5e3c680

34 files changed

Lines changed: 2526 additions & 126 deletions

‎.env.example‎

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1211,11 +1211,17 @@ OPENWEATHER_API_KEY=
12111211
# Agent Trigger Delivery #
12121212
#===========================#
12131213

1214-
# Base URL used by trusted in-process event producers to dispatch agent fires and steers.
1214+
# Base URL used by trusted in-process event producers to dispatch agent fires, continuations,
1215+
# and steers.
12151216
# Defaults to this process's bound listener. Set only when internal trigger admission must
12161217
# traverse another trusted HTTP origin, such as a TLS front door.
12171218
# AGENT_TRIGGERS_SELF_URL=http://127.0.0.1:3080
12181219

1220+
# Automatically continue a saved parent agent after a detached subagent settles.
1221+
# Rolling-deploy safety: deploy support with this disabled first, wait until every API
1222+
# replica is upgraded, then enable it in a subsequent rollout.
1223+
# ENABLE_SUBAGENT_COMPLETION_WAKEUPS=false
1224+
12191225
# Trusted event adapters enqueue through the shared durable trigger service. Mongo-backed
12201226
# leases make its workers safe across replicas; successful delivery records expire after
12211227
# 90 days, while dead letters remain available for explicit operator requeue.

‎CONTEXT.md‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3,4 +3,5 @@
33
- **Agent run envelope**: the versioned, JSON-safe request contract created after ingress authentication and protocol validation but before agent, provider, tool, or MCP initialization. It carries only the validated protocol payload and the minimum trusted principal identifiers. The execution host rehydrates all runtime state from those identifiers.
44
- **Subagent thread**: a durable, view-only child conversation owned by one parent conversation and subagent identity. A parent agent may continue it by stable `threadId`; each continuation uses a fresh execution lease restored from the canonical child transcript. It is not an ordinary human-writable chat.
55
- **Live subagent task owner**: the one API process holding a detached child execution, its abort controller, and its bounded control queue. Redis may route trusted poll/control envelopes to that owner, but it does not migrate or persist the executor; Mongo persists only the logical child thread and its continuation fence.
6+
- **Subagent completion wakeup**: a durable internal `continue` trigger pre-registered before detached child execution so a process crash cannot lose the wakeup. Delivery defers until the child's terminal transcript is persisted, targets the initiating agent and exact parent response branch, carries task metadata rather than child output, waits for the parent generation to settle, and starts the parent turn that collects the result through the existing task store.
67
- **Theme definition**: a versioned, data-only description of LibreChat semantic colors and shared appearance roles, optionally specialized by light or dark mode. The theme module validates and resolves partial definitions against bundled defaults before adapters apply them. A theme definition does not contain arbitrary CSS, application behavior, or alternate feature layouts.

‎api/server/controllers/agents/__tests__/request.resumeMetadata.spec.js‎

Lines changed: 114 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -519,6 +519,120 @@ describe('ResumableAgentController resume metadata', () => {
519519
);
520520
});
521521

522+
it('defers a trusted trigger resume while its parent generation is still active', async () => {
523+
const conversationId = 'conversation-123';
524+
mockGetMessages.mockResolvedValue([{ _id: 'persisted-parent' }]);
525+
mockGenerationJobManager.getJob.mockResolvedValue({
526+
status: 'running',
527+
metadata: { userId: 'user-123' },
528+
});
529+
const initializeClient = jest.fn();
530+
const req = {
531+
_isAgentTrigger: true,
532+
user: { id: 'user-123' },
533+
body: {
534+
text: 'Collect the completed child.',
535+
messageId: 'wakeup-user-message',
536+
parentMessageId: 'persisted-response_',
537+
conversationId,
538+
clientRequestId: 'trigger_resume_1',
539+
endpointOption: { endpoint: 'agents', modelOptions: { model: 'gpt-4.1' } },
540+
},
541+
config: {},
542+
};
543+
const res = createResumableResponse();
544+
545+
await AgentController(req, res, jest.fn(), initializeClient, null);
546+
547+
expect(res.status).toHaveBeenCalledWith(409);
548+
expect(res.json).toHaveBeenCalledWith(expect.objectContaining({ code: 'PARENT_NOT_READY' }));
549+
expect(mockGenerationJobManager.claimGeneration).not.toHaveBeenCalled();
550+
expect(mockCheckAndIncrementPendingRequest).not.toHaveBeenCalled();
551+
expect(initializeClient).not.toHaveBeenCalled();
552+
});
553+
554+
it('labels a trigger parent-state lookup failure as provably pre-admission', async () => {
555+
const conversationId = 'conversation-123';
556+
mockGetMessages.mockResolvedValue([{ _id: 'persisted-parent' }]);
557+
mockGenerationJobManager.getJob.mockRejectedValue(new Error('redis unavailable'));
558+
const initializeClient = jest.fn();
559+
const req = {
560+
_isAgentTrigger: true,
561+
user: { id: 'user-123' },
562+
body: {
563+
text: 'Collect the completed child.',
564+
messageId: 'wakeup-user-message',
565+
parentMessageId: 'persisted-response_',
566+
conversationId,
567+
clientRequestId: 'trigger_resume_1',
568+
endpointOption: { endpoint: 'agents', modelOptions: { model: 'gpt-4.1' } },
569+
},
570+
config: {},
571+
};
572+
const res = createResumableResponse();
573+
574+
await AgentController(req, res, jest.fn(), initializeClient, null);
575+
576+
expect(res.set).toHaveBeenCalledWith('Retry-After', '1');
577+
expect(res.status).toHaveBeenCalledWith(503);
578+
expect(res.json).toHaveBeenCalledWith(
579+
expect.objectContaining({ code: 'PARENT_STATE_UNAVAILABLE' }),
580+
);
581+
expect(mockGenerationJobManager.claimGeneration).not.toHaveBeenCalled();
582+
expect(mockCheckAndIncrementPendingRequest).not.toHaveBeenCalled();
583+
expect(initializeClient).not.toHaveBeenCalled();
584+
});
585+
586+
it('deduplicates the active continuation whose admission response was lost', async () => {
587+
const conversationId = 'conversation-123';
588+
mockGetMessages.mockResolvedValue([{ _id: 'persisted-parent' }]);
589+
mockGenerationJobManager.getJob.mockResolvedValue({
590+
createdAt: 1000,
591+
status: 'requires_action',
592+
metadata: {
593+
userId: 'user-123',
594+
idempotencyClientRequestId: 'trigger_resume_1',
595+
},
596+
});
597+
mockGenerationJobManager.claimGeneration.mockResolvedValue({
598+
claimed: false,
599+
existing: {
600+
streamId: conversationId,
601+
conversationId,
602+
claimedAt: 100,
603+
claimToken: 'existing-token',
604+
startedAt: 1000,
605+
},
606+
});
607+
const req = {
608+
_isAgentTrigger: true,
609+
user: { id: 'user-123' },
610+
body: {
611+
text: 'Collect the completed child.',
612+
messageId: 'wakeup-user-message',
613+
parentMessageId: 'persisted-response_',
614+
conversationId,
615+
clientRequestId: 'trigger_resume_1',
616+
endpointOption: { endpoint: 'agents', modelOptions: { model: 'gpt-4.1' } },
617+
},
618+
config: {},
619+
};
620+
const res = createResumableResponse();
621+
622+
await AgentController(req, res, jest.fn(), jest.fn(), null);
623+
624+
expect(res.status).not.toHaveBeenCalledWith(409);
625+
expect(res.json).toHaveBeenCalledWith({
626+
streamId: conversationId,
627+
conversationId,
628+
generationCreatedAt: 1000,
629+
status: 'resumed',
630+
generationProtocolVersion: 1,
631+
});
632+
expect(mockGenerationJobManager.createJob).not.toHaveBeenCalled();
633+
expect(mockCheckAndIncrementPendingRequest).not.toHaveBeenCalled();
634+
});
635+
522636
it('creates the job with the in-flight turn before MCP initialization can emit OAuth', async () => {
523637
const conversationId = 'conversation-123';
524638
const initializeClient = jest.fn().mockRejectedValue(new Error('stop before tool loading'));

‎api/server/controllers/agents/request.js‎

Lines changed: 132 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -301,13 +301,26 @@ function rejectPreliminaryParentMessageId(res, generationProtocolVersion) {
301301
res,
302302
409,
303303
{
304+
code: 'PARENT_NOT_READY',
304305
error:
305306
'Cannot submit a follow-up while the selected parent response is still being saved. Please wait and try again.',
306307
},
307308
generationProtocolVersion,
308309
);
309310
}
310311

312+
function rejectMissingTriggerParentMessageId(res, generationProtocolVersion) {
313+
return sendGenerationJson(
314+
res,
315+
404,
316+
{
317+
code: 'PARENT_NOT_FOUND',
318+
error: 'The selected parent response is no longer available.',
319+
},
320+
generationProtocolVersion,
321+
);
322+
}
323+
311324
/**
312325
* Resumable Agent Controller - Generation runs independently of HTTP connection.
313326
* Returns streamId immediately, client subscribes separately via SSE.
@@ -449,6 +462,9 @@ const ResumableAgentController = async (req, res, next, initializeClient, addTit
449462
: undefined,
450463
});
451464

465+
const isTriggerContinuation =
466+
req._isAgentTrigger === true && !isNewConvo && parentMessageId !== Constants.NO_PARENT;
467+
452468
if (
453469
await isUnpersistedPreliminaryParent({
454470
userId,
@@ -457,6 +473,38 @@ const ResumableAgentController = async (req, res, next, initializeClient, addTit
457473
getMessages,
458474
})
459475
) {
476+
if (isTriggerContinuation) {
477+
let parentJob;
478+
try {
479+
parentJob = await GenerationJobManager.getJob(conversationId);
480+
} catch (error) {
481+
logger.warn('[ResumableAgentController] Trigger parent lookup failed', error);
482+
res.set('Retry-After', '1');
483+
startupTelemetry?.end('rejected');
484+
return sendGenerationJson(
485+
res,
486+
503,
487+
{ code: 'PARENT_STATE_UNAVAILABLE', error: 'Parent generation state is unavailable.' },
488+
generationProtocolVersion,
489+
);
490+
}
491+
if (
492+
parentJob != null &&
493+
liveJobBelongsToRequester(parentJob, req.user) &&
494+
(parentJob.status === 'running' ||
495+
parentJob.status === 'requires_action' ||
496+
parentJob.metadata?.terminalPersistencePending === true) &&
497+
!(
498+
typeof clientRequestId === 'string' &&
499+
parentJob.metadata?.idempotencyClientRequestId === clientRequestId
500+
)
501+
) {
502+
startupTelemetry?.end('rejected');
503+
return rejectPreliminaryParentMessageId(res, generationProtocolVersion);
504+
}
505+
startupTelemetry?.end('rejected');
506+
return rejectMissingTriggerParentMessageId(res, generationProtocolVersion);
507+
}
460508
startupTelemetry?.end('rejected');
461509
return rejectPreliminaryParentMessageId(res, generationProtocolVersion);
462510
}
@@ -472,6 +520,51 @@ const ResumableAgentController = async (req, res, next, initializeClient, addTit
472520
const streamId = conversationId;
473521
req.body.conversationId = conversationId;
474522

523+
/** A durable continuation trigger appends below a completed parent response. If
524+
* that response belongs to a still-running or paused generation, admitting
525+
* another generation on the same conversation stream would replace it.
526+
* Defer without claiming the continuation idempotency key so the delivery engine
527+
* can retry after the parent reaches a terminal state. */
528+
if (isTriggerContinuation) {
529+
let parentJob;
530+
try {
531+
parentJob = await GenerationJobManager.getJob(streamId);
532+
} catch (error) {
533+
logger.warn('[ResumableAgentController] Trigger continuation parent lookup failed', error);
534+
res.set('Retry-After', '1');
535+
startupTelemetry?.end('rejected');
536+
return sendGenerationJson(
537+
res,
538+
503,
539+
{
540+
code: 'PARENT_STATE_UNAVAILABLE',
541+
error: 'Parent generation state is temporarily unavailable.',
542+
},
543+
generationProtocolVersion,
544+
);
545+
}
546+
if (
547+
parentJob != null &&
548+
liveJobBelongsToRequester(parentJob, req.user) &&
549+
(parentJob.status === 'running' ||
550+
parentJob.status === 'requires_action' ||
551+
parentJob.metadata?.terminalPersistencePending === true) &&
552+
!(
553+
typeof clientRequestId === 'string' &&
554+
parentJob.metadata?.idempotencyClientRequestId === clientRequestId
555+
)
556+
) {
557+
res.set('Retry-After', '1');
558+
startupTelemetry?.end('rejected');
559+
return sendGenerationJson(
560+
res,
561+
409,
562+
{ code: 'PARENT_NOT_READY', error: 'The parent generation has not settled yet.' },
563+
generationProtocolVersion,
564+
);
565+
}
566+
}
567+
475568
// Idempotency: a lost/reset start-generation response makes the client re-POST the
476569
// identical payload, which would otherwise start a second fully-billed generation.
477570
// Claim the submission's clientRequestId before creating the job so a retry attaches
@@ -881,6 +974,7 @@ const ResumableAgentController = async (req, res, next, initializeClient, addTit
881974
...(recoveredSteerId && { recoveredSteerId }),
882975
...(recoveredSteerPayload && { recoveredSteerPayload }),
883976
...(expectedPredecessorCreatedAt != null && { expectedPredecessorCreatedAt }),
977+
...(isTriggerContinuation && { rejectActivePredecessor: true }),
884978
...(ownedIdempotencyClaim?.claimToken && {
885979
idempotencyClientRequestId: clientRequestId,
886980
idempotencyClaimToken: ownedIdempotencyClaim.claimToken,
@@ -1908,31 +2002,44 @@ const ResumableAgentController = async (req, res, next, initializeClient, addTit
19082002
if (error?.code === 'GENERATION_PREDECESSOR_MISMATCH') {
19092003
const currentJob = error.currentJob;
19102004
const currentStatus = currentJob?.status;
1911-
const predecessorVerified =
1912-
currentJob != null &&
1913-
Number.isSafeInteger(currentJob.createdAt) &&
1914-
currentJob.createdAt >= 0 &&
1915-
currentJob.verified !== false;
1916-
sendGenerationJson(
1917-
res,
1918-
409,
1919-
{
1920-
status: 'predecessor_mismatch',
1921-
code: 'GENERATION_PREDECESSOR_MISMATCH',
1922-
error: predecessorVerified
1923-
? 'A newer generation became current before this request could start.'
1924-
: 'The prior generation could not be verified. Please retry.',
1925-
streamId,
1926-
conversationId: currentJob?.conversationId ?? conversationId,
1927-
generationCreatedAt: currentJob?.createdAt,
1928-
predecessorVerified,
1929-
active:
1930-
typeof currentJob?.active === 'boolean'
1931-
? currentJob.active
1932-
: currentStatus === 'running' || currentStatus === 'requires_action',
1933-
},
1934-
generationProtocolVersion,
1935-
);
2005+
if (isTriggerContinuation && currentJob?.active === true) {
2006+
res.set('Retry-After', '1');
2007+
sendGenerationJson(
2008+
res,
2009+
409,
2010+
{
2011+
code: 'PARENT_NOT_READY',
2012+
error: 'Another generation became active before the continuation could start.',
2013+
},
2014+
generationProtocolVersion,
2015+
);
2016+
} else {
2017+
const predecessorVerified =
2018+
currentJob != null &&
2019+
Number.isSafeInteger(currentJob.createdAt) &&
2020+
currentJob.createdAt >= 0 &&
2021+
currentJob.verified !== false;
2022+
sendGenerationJson(
2023+
res,
2024+
409,
2025+
{
2026+
status: 'predecessor_mismatch',
2027+
code: 'GENERATION_PREDECESSOR_MISMATCH',
2028+
error: predecessorVerified
2029+
? 'A newer generation became current before this request could start.'
2030+
: 'The prior generation could not be verified. Please retry.',
2031+
streamId,
2032+
conversationId: currentJob?.conversationId ?? conversationId,
2033+
generationCreatedAt: currentJob?.createdAt,
2034+
predecessorVerified,
2035+
active:
2036+
typeof currentJob?.active === 'boolean'
2037+
? currentJob.active
2038+
: currentStatus === 'running' || currentStatus === 'requires_action',
2039+
},
2040+
generationProtocolVersion,
2041+
);
2042+
}
19362043
} else if (error?.code === 'RECOVERY_PAYLOAD_MISMATCH') {
19372044
sendGenerationJson(
19382045
res,

‎api/server/services/Agents/triggers.js‎

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,9 +1,17 @@
1-
const { createAgentTriggerService } = require('@librechat/api');
1+
const {
2+
createAgentTriggerService,
3+
createSubagentCompletionWakeupResolver,
4+
GenerationJobManager,
5+
} = require('@librechat/api');
26
const methods = require('~/models');
37

48
const service = createAgentTriggerService({
59
methods,
610
isPrincipalActive: methods.isAgentTriggerPrincipalActive,
11+
prepareContinue: createSubagentCompletionWakeupResolver({
12+
methods,
13+
getGenerationJob: (conversationId) => GenerationJobManager.getJob(conversationId),
14+
}),
715
});
816

917
module.exports = {

0 commit comments

Comments
 (0)