From d4e0bb806020268ee322ad31d2ed119ba0dd79fc Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 18 Aug 2026 19:35:52 +0000 Subject: [PATCH] =?UTF-8?q?fix(stream):=20Wave=20C.2=20=E2=80=94=20durable?= =?UTF-8?q?=20retry/ack=20for=20terminal=20host=20lifecycle=20actions?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Thread r3804518375. Approval expiry won the `requires_action → aborted` CAS and then invoked the host hook best-effort: `runApprovalExpiredHandler` swallowed a failure, and because later sweeps enumerate only `requires_action` jobs, the now- aborted job was never offered again. In the clustered entrypoint (no schedule reconciler) the ScheduleRun stayed `requires_action` and its retained job persisted indefinitely. Make the host lifecycle work durable rather than schedule-specific: - Add a generic `terminalHostActionPending` marker, set ATOMICALLY in the same terminal transition (ApprovalLifecycle.expireWithIdentity), only when a host adapter is installed. - Retain and index such jobs: both stores keep them out of terminal reaping and expose getTerminalHostActionJobs(); Redis adds a set + extended (24h-bounded) TTL, in-memory a bounded 24h retention so a permanently-failing hook cannot leak. - The manager clears the marker only after the adapter acknowledges success, fenced by generation identity (clearTerminalHostAction), so a replacement generation can neither clear nor execute its predecessor's action. - cleanup()/expireStaleApprovals() enumerates unacknowledged terminal host actions across restarts and replicas and retries the idempotent hook; the relay only re-invokes while the marker is unacknowledged, so a successful ack prevents duplicate work. Store-won expiry marks it too, so a loser-replica relay still crosses the hook. - Terminal SSE notification continues regardless of host-hook outcome. Covers in-memory behavior (retry after failure, restart/other-replica retry, ack prevents duplicates, identity fence, terminal notification on failure, no marker accumulation for non-scheduled jobs) and updates the Redis cluster-membership contract test for the new index. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01CZwZPDBgxy4mCWWg6zQSjq --- packages/api/src/stream/ApprovalLifecycle.ts | 12 +- .../api/src/stream/GenerationJobManager.ts | 35 ++++- .../stream/__tests__/RedisJobStore.spec.ts | 28 +++- .../stream/__tests__/pendingAction.spec.ts | 143 +++++++++++++++++- .../implementations/InMemoryJobStore.ts | 46 +++++- .../stream/implementations/RedisJobStore.ts | 68 +++++++++ .../api/src/stream/interfaces/IJobStore.ts | 17 +++ 7 files changed, 330 insertions(+), 19 deletions(-) diff --git a/packages/api/src/stream/ApprovalLifecycle.ts b/packages/api/src/stream/ApprovalLifecycle.ts index f8db79b839..ac6961af51 100644 --- a/packages/api/src/stream/ApprovalLifecycle.ts +++ b/packages/api/src/stream/ApprovalLifecycle.ts @@ -391,6 +391,7 @@ export class ApprovalLifecycle { streamId: string, expectedActionId?: string, expectedCreatedAt?: number, + options?: { markHostActionPending?: boolean }, ): Promise { const job = await this.waitForPausePersistence(streamId, expectedCreatedAt); if ( @@ -410,8 +411,15 @@ export class ApprovalLifecycle { to: 'aborted', clear: ['pendingAction', 'pendingActionId'], // completedAt lets the stores' terminal-cleanup reclaim the job; without - // it an expired approval lingers in the in-memory map indefinitely. - patch: { error: 'Approval expired before a decision was made', completedAt: Date.now() }, + // it an expired approval lingers in the in-memory map indefinitely. When a host + // action is owed, `terminalHostActionPending` retains the job (and keeps it + // enumerable) until the adapter acknowledges — set ATOMICALLY here so a crash right + // after this CAS still leaves the durable retry evidence. + patch: { + error: 'Approval expired before a decision was made', + completedAt: Date.now(), + ...(options?.markHostActionPending === true && { terminalHostActionPending: true }), + }, expectActionId: expectedActionId, expectCreatedAt: createdAt, }); diff --git a/packages/api/src/stream/GenerationJobManager.ts b/packages/api/src/stream/GenerationJobManager.ts index eb5ab037b5..efd0046269 100644 --- a/packages/api/src/stream/GenerationJobManager.ts +++ b/packages/api/src/stream/GenerationJobManager.ts @@ -7024,6 +7024,9 @@ class GenerationJobManagerClass { streamId, actionId, expectedCreatedAt ?? observedJob?.createdAt, + // Only retain a durable host-action marker when a host adapter is installed to + // consume it — a store with no handler owes no action and accumulates nothing. + { markHostActionPending: this.approvalExpiredHandler != null }, ); if (expiredJob == null) { return false; @@ -7040,10 +7043,15 @@ class GenerationJobManagerClass { ): Promise { try { await this.approvalExpiredHandler?.(streamId, job); + // Success: the durable host action is settled. Clear its pending marker, fenced to + // this exact generation so a replacement at the same streamId keeps its own state. + // A no-op handler (or none) clears harmlessly, so non-scheduled jobs never linger. + await this.jobStore.clearTerminalHostAction?.(streamId, job.createdAt); } catch (err) { - // Expiry itself already won its exact CAS. Keep terminal notification moving; - // replica relays may deliver this idempotent hook again, and host-owned durable - // evidence remains available to that host's own recovery path. + // Expiry itself already won its exact CAS. Keep terminal notification moving; the + // `terminalHostActionPending` marker stays set so a later cleanup pass (this or + // another replica, across restarts) re-enumerates the job and retries this + // idempotent hook until it acknowledges. logger.error(`[GenerationJobManager] Approval-expiry host hook failed: ${streamId}`, err); } } @@ -7102,6 +7110,19 @@ class GenerationJobManagerClass { logger.error('[GenerationJobManager] Failed to enumerate pending approvals', err); } } + // Also scan terminal jobs that still owe a host lifecycle hook. An expired approval is + // no longer in the requires_action index, so without this a failed host hook (e.g. the + // schedule outcome write) would never be retried on a replica or after a restart that + // has no local runtime — the exact clustered-entrypoint gap, which runs no reconciler. + if (this.jobStore.getTerminalHostActionJobs) { + try { + for (const job of await this.jobStore.getTerminalHostActionJobs()) { + candidates.set(job.streamId, job); + } + } catch (err) { + logger.error('[GenerationJobManager] Failed to enumerate pending host actions', err); + } + } const streamIds = new Set([...this.runtimeState.keys(), ...candidates.keys()]); for (const streamId of streamIds) { let job: SerializableJobData | null; @@ -7123,7 +7144,13 @@ class GenerationJobManagerClass { // The `errorEvent` flag (set by emitError) keeps this idempotent vs the win path. const runtime = this.runtimeState.get(streamId); if (job?.status === 'aborted' && job.error === APPROVAL_EXPIRED_ERROR) { - await this.runApprovalExpiredHandler(streamId, job); + // Retry the durable host hook ONLY while its marker is unacknowledged, so a + // successful ack prevents duplicate work. The terminal SSE relay below is + // separately idempotent (emitError's errorEvent flag) and always runs so a + // loser-replica subscriber still gets a terminal event. + if (job.terminalHostActionPending === true) { + await this.runApprovalExpiredHandler(streamId, job); + } await this.notifyApprovalExpiredRuntime(streamId, job.createdAt, runtime); changed = this.releaseJobOwnership(streamId, job.createdAt) || changed; continue; diff --git a/packages/api/src/stream/__tests__/RedisJobStore.spec.ts b/packages/api/src/stream/__tests__/RedisJobStore.spec.ts index 8713b38992..0b9c4a46ae 100644 --- a/packages/api/src/stream/__tests__/RedisJobStore.spec.ts +++ b/packages/api/src/stream/__tests__/RedisJobStore.spec.ts @@ -665,6 +665,7 @@ describe('RedisJobStore', () => { const evalResult = createDeferred(); const runningMembership = createDeferred(); const requiresActionRemoval = createDeferred(); + const terminalHostActionRemoval = createDeferred(); const userMembership = createDeferred(); const userExpiry = createDeferred(); const started: string[] = []; @@ -688,9 +689,13 @@ describe('RedisJobStore', () => { started.push('user'); return userMembership.promise; }), - srem: jest.fn(() => { - started.push('requires_action'); - return requiresActionRemoval.promise; + srem: jest.fn((key: string) => { + if (key === 'stream:requires_action') { + started.push('requires_action'); + return requiresActionRemoval.promise; + } + started.push('terminal_host_action'); + return terminalHostActionRemoval.promise; }), hgetall: jest.fn(() => jobHashFromCreationCall(evalJobCreation.mock.calls[0])), expire, @@ -705,16 +710,23 @@ describe('RedisJobStore', () => { expect(started).toEqual(['job']); evalResult.resolve(1); - await waitFor(() => started.length === 4); + await waitFor(() => started.length === 5); - expect(started).toEqual(['job', 'running', 'requires_action', 'user']); + expect(started).toEqual(['job', 'running', 'requires_action', 'terminal_host_action', 'user']); expect(settled).toBe(false); expect(expire).not.toHaveBeenCalled(); userMembership.resolve(1); await waitFor(() => expire.mock.calls.length === 1); - expect(started).toEqual(['job', 'running', 'requires_action', 'user', 'user_expiry']); + expect(started).toEqual([ + 'job', + 'running', + 'requires_action', + 'terminal_host_action', + 'user', + 'user_expiry', + ]); expect(expire).toHaveBeenCalledWith('stream:user:{user-1}:jobs', 60); expect(settled).toBe(false); @@ -726,6 +738,10 @@ describe('RedisJobStore', () => { await Promise.resolve(); expect(settled).toBe(false); + terminalHostActionRemoval.resolve(1); + await Promise.resolve(); + expect(settled).toBe(false); + requiresActionRemoval.resolve(1); await expect(creating).resolves.toMatchObject({ streamId: 'stream-1', diff --git a/packages/api/src/stream/__tests__/pendingAction.spec.ts b/packages/api/src/stream/__tests__/pendingAction.spec.ts index 89bb58350e..0a0c034dd6 100644 --- a/packages/api/src/stream/__tests__/pendingAction.spec.ts +++ b/packages/api/src/stream/__tests__/pendingAction.spec.ts @@ -978,6 +978,133 @@ describe('ApprovalLifecycle via GenerationJobManager.approvals (in-memory)', () }); }); + describe('durable approval-expiry host action', () => { + const sweep = (m: GenerationJobManagerClass) => + (m as unknown as { expireStaleApprovals: () => Promise }).expireStaleApprovals(); + + async function pauseScheduled(streamId: string) { + const job = await manager.createJob(streamId, 'user-1'); + await manager.updateMetadata(streamId, { + scheduleId: 's1', + scheduledFor: '2026-08-17T12:00:00.000Z', + }); + await manager.approvals.pause(streamId, buildAction(streamId)); + return job; + } + + test('retains the marker on hook failure and retries it on a later cleanup pass', async () => { + const streamId = 'stream-host-retry'; + await pauseScheduled(streamId); + const handler = jest + .fn, [string, { scheduleId?: string }]>() + .mockRejectedValueOnce(new Error('mongo down')) + .mockResolvedValue(undefined); + manager.setApprovalExpiredHandler(handler); + + // First expiry wins its CAS, but the host hook FAILS: the durable marker is retained. + expect(await manager.expireApproval(streamId)).toBe(true); + expect(handler).toHaveBeenCalledTimes(1); + expect((await jobStore.getJob(streamId))?.terminalHostActionPending).toBe(true); + expect(await jobStore.getTerminalHostActionJobs()).toHaveLength(1); + + // A later cleanup pass re-enumerates the aborted job and retries; it now succeeds + // and clears the marker. + await sweep(manager); + expect(handler).toHaveBeenCalledTimes(2); + expect((await jobStore.getJob(streamId))?.terminalHostActionPending).toBeUndefined(); + expect(await jobStore.getTerminalHostActionJobs()).toHaveLength(0); + }); + + test('a fresh manager (restart / other replica) retries the pending host action', async () => { + const streamId = 'stream-host-restart'; + const job = await pauseScheduled(streamId); + manager.setApprovalExpiredHandler(jest.fn().mockRejectedValue(new Error('down'))); + expect(await manager.expireApproval(streamId)).toBe(true); + expect((await jobStore.getJob(streamId))?.terminalHostActionPending).toBe(true); + + // Another replica / a restarted process: a fresh manager over the SAME durable store, + // with NO local runtime for this stream. + const other = new GenerationJobManagerClass(); + other.configure({ + jobStore, + eventTransport: new InMemoryEventTransport(), + isRedis: false, + cleanupOnComplete: false, + }); + other.initialize(); + const succeeding = jest.fn().mockResolvedValue(undefined); + other.setApprovalExpiredHandler(succeeding); + + await sweep(other); + + expect(succeeding).toHaveBeenCalledWith( + streamId, + expect.objectContaining({ createdAt: job.createdAt, scheduleId: 's1' }), + ); + expect((await jobStore.getJob(streamId))?.terminalHostActionPending).toBeUndefined(); + await other.destroy(); + }); + + test('a successful acknowledgement prevents duplicate work on later sweeps', async () => { + const streamId = 'stream-host-ack'; + await pauseScheduled(streamId); + const handler = jest.fn().mockResolvedValue(undefined); + manager.setApprovalExpiredHandler(handler); + + expect(await manager.expireApproval(streamId)).toBe(true); + expect(handler).toHaveBeenCalledTimes(1); + expect(await jobStore.getTerminalHostActionJobs()).toHaveLength(0); + + await sweep(manager); + expect(handler).toHaveBeenCalledTimes(1); // no duplicate invocation + }); + + test('clearTerminalHostAction is identity-fenced against a replacement generation', async () => { + const streamId = 'stream-host-fence'; + const job = await pauseScheduled(streamId); + manager.setApprovalExpiredHandler(jest.fn().mockRejectedValue(new Error('down'))); + await manager.expireApproval(streamId); + expect((await jobStore.getJob(streamId))?.terminalHostActionPending).toBe(true); + + // A clear fenced to a DIFFERENT generation must not clear this one's action. + await jobStore.clearTerminalHostAction(streamId, job.createdAt + 1); + expect((await jobStore.getJob(streamId))?.terminalHostActionPending).toBe(true); + + // The exact generation clears it. + await jobStore.clearTerminalHostAction(streamId, job.createdAt); + expect((await jobStore.getJob(streamId))?.terminalHostActionPending).toBeUndefined(); + }); + + test('terminalizes and delivers the terminal state even when the first hook fails', async () => { + const streamId = 'stream-host-notify'; + await pauseScheduled(streamId); + manager.setApprovalExpiredHandler(jest.fn().mockRejectedValue(new Error('down'))); + + // The CAS + terminal notification proceed regardless of the host hook outcome. + expect(await manager.expireApproval(streamId)).toBe(true); + expect(await manager.getJobStatus(streamId)).toBe('aborted'); + }); + + test('a non-scheduled job with a no-op host hook accumulates no marker', async () => { + const streamId = 'stream-host-nonsched'; + await manager.createJob(streamId, 'user-1'); // no schedule metadata + await manager.approvals.pause(streamId, buildAction(streamId)); + // The schedule adapter is installed but owns no action for a non-scheduled job. + manager.setApprovalExpiredHandler( + jest.fn(async (_streamId: string, job: { scheduleId?: string }) => { + if (!job.scheduleId) { + return; + } + }), + ); + + expect(await manager.expireApproval(streamId)).toBe(true); + // Marked atomically then cleared on the no-op success — nothing lingers. + expect((await jobStore.getJob(streamId))?.terminalHostActionPending).toBeUndefined(); + expect(await jobStore.getTerminalHostActionJobs()).toHaveLength(0); + }); + }); + describe('facade integration', () => { test('requires_action drops the running count but keeps the user-active set', async () => { const streamId = 'stream-counts'; @@ -1047,9 +1174,9 @@ describe('InMemoryJobStore — approval expiry cleanup', () => { expect((await store.getJob('epoch-guard'))?.status).toBe('error'); }); - test('cleanup() finalizes and reclaims a past-expiry pending-approval job', async () => { + test('cleanup() finalizes a past-expiry approval, retaining it for host-hook retry', async () => { const store = new InMemoryJobStore({ ttlAfterComplete: 0 }); - await store.createJob('s1', 'u1'); + const job = await store.createJob('s1', 'u1'); const action = buildPendingAction( buildToolApprovalPayload([{ name: 'shell', arguments: {}, tool_call_id: 'c1' }]), @@ -1061,7 +1188,17 @@ describe('InMemoryJobStore — approval expiry cleanup', () => { patch: { pendingAction: action, pendingActionId: action.actionId }, }); - // A past-expiry approval must be finalized + reclaimed, not left resident. + // A past-expiry approval is finalized (aborted) but RETAINED with a pending host-action + // marker so the manager relay can still run its lifecycle hook (the store cannot know + // whether one is owed); it is enumerable for that retry. + await store.cleanup(); + const aborted = await store.getJob('s1'); + expect(aborted?.status).toBe('aborted'); + expect(aborted?.terminalHostActionPending).toBe(true); + expect(await store.getTerminalHostActionJobs()).toHaveLength(1); + + // Once the host action acknowledges, the marker clears and the next cleanup reclaims it. + await store.clearTerminalHostAction('s1', job.createdAt); await store.cleanup(); expect(await store.getJob('s1')).toBeNull(); }); diff --git a/packages/api/src/stream/implementations/InMemoryJobStore.ts b/packages/api/src/stream/implementations/InMemoryJobStore.ts index 08b0515e63..227d711cbb 100644 --- a/packages/api/src/stream/implementations/InMemoryJobStore.ts +++ b/packages/api/src/stream/implementations/InMemoryJobStore.ts @@ -46,6 +46,10 @@ const STEER_RECEIPT_MAX_PER_STREAM: number = 100; const CREATE_REPLACEMENT_RECEIPT_MAX = 32; const CLIENT_REQUEST_ID_PATTERN = /^[A-Za-z0-9:_-]{1,128}$/; const TERMINAL_PERSISTENCE_RETENTION_MS = 5 * 60 * 1000; +/** Bounded window a terminal job owing a host lifecycle hook is retained for retry before + * it is reaped even if unacknowledged (a permanently-failing hook cannot leak forever). + * Mirrors the Redis store's 24h host-action TTL. */ +const HOST_ACTION_RETENTION_MS = 24 * 60 * 60 * 1000; function assertCreateIdempotencyArguments( claimKey?: string, @@ -1018,6 +1022,26 @@ export class InMemoryJobStore implements IJobStoreV2 { return paused; } + async getTerminalHostActionJobs(): Promise { + const pending: SerializableJobData[] = []; + for (const job of this.jobs.values()) { + if (job.terminalHostActionPending === true) { + pending.push(job); + } + } + return pending; + } + + async clearTerminalHostAction(streamId: string, expectedCreatedAt?: number): Promise { + const job = this.jobs.get(streamId); + // Identity-fenced: a replacement generation at this streamId (a newer createdAt) must + // not have a predecessor's marker cleared on its behalf. + if (job == null || (expectedCreatedAt != null && job.createdAt !== expectedCreatedAt)) { + return; + } + delete job.terminalHostActionPending; + } + async cleanup(): Promise { const now = Date.now(); const toDelete: Array<{ streamId: string; createdAt: number }> = []; @@ -1081,7 +1105,19 @@ export class InMemoryJobStore implements IJobStoreV2 { } const isFinished = ['complete', 'error', 'aborted'].includes(job.status); - if (isFinished && job.completedAt && job.preserveForScheduleReconcile !== true) { + // A terminal job that still owes a host lifecycle hook is retained (like a + // reconcile-preserved job) so a later cleanup pass can enumerate and retry it — + // but only within a bounded window, so a permanently-failing hook cannot leak. + const hostActionHeld = + job.terminalHostActionPending === true && + job.completedAt != null && + now - job.completedAt <= HOST_ACTION_RETENTION_MS; + if ( + isFinished && + job.completedAt && + job.preserveForScheduleReconcile !== true && + !hostActionHeld + ) { // A pending final event is a persistence barrier, not an ordinary // completed row. Give its owner/recovery path a bounded window even // when normal terminal retention is configured to zero. @@ -1109,11 +1145,13 @@ export class InMemoryJobStore implements IJobStoreV2 { job.status = 'aborted'; job.completedAt = now; job.error = 'Approval expired before a decision was made'; + // Store-won expiry (the manager's own sweep did not win the CAS): mark the host + // action pending so the manager relay still runs its lifecycle hook, and RETAIN the + // job (even under zero terminal retention) so a sweep can enumerate it — cleared and + // reaped once the hook acknowledges. + job.terminalHostActionPending = true; delete job.pendingAction; delete job.pendingActionId; - if (this.ttlAfterComplete === 0) { - toDelete.push({ streamId, createdAt: job.createdAt }); - } } else if (this.staleJobTimeout > 0 && job.status === 'running') { // Failsafe: reap jobs stuck in "running" with no generation activity for // longer than the stale timeout. These are crashed/hung generations that diff --git a/packages/api/src/stream/implementations/RedisJobStore.ts b/packages/api/src/stream/implementations/RedisJobStore.ts index f7a440e21f..38e2c2f6eb 100644 --- a/packages/api/src/stream/implementations/RedisJobStore.ts +++ b/packages/api/src/stream/implementations/RedisJobStore.ts @@ -1535,6 +1535,9 @@ const KEYS = { runningJobs: 'stream:running', /** Jobs paused for human review (global set - single slot) */ requiresActionJobs: 'stream:requires_action', + /** Terminal jobs that still owe a durable host lifecycle hook (global set). Retains + * the aborted approval-expiry job for cross-replica / post-restart hook retry. */ + terminalHostActionJobs: 'stream:terminal_host_action', /** User's active jobs set, tenant-qualified when tenantId is available */ userJobs: (userId: string, tenantId?: string) => tenantId ? `stream:user:{${tenantId}:${userId}}:jobs` : `stream:user:{${userId}}:jobs`, @@ -2236,6 +2239,11 @@ export class RedisJobStore implements IJobStoreV2 { statusKey === KEYS.requiresActionJobs ? this.redis.sadd(KEYS.requiresActionJobs, streamId) : this.redis.srem(KEYS.requiresActionJobs, streamId), + // Terminal host-action membership follows the durable hash field, not status, so + // an aborted approval-expiry job stays enumerable for hook retry until acked. + job?.terminalHostActionPending === true + ? this.redis.sadd(KEYS.terminalHostActionJobs, streamId) + : this.redis.srem(KEYS.terminalHostActionJobs, streamId), ]; for (const userJobsKey of observedUserKeys) { if (userJobsKey !== activeUserKey) { @@ -2267,6 +2275,11 @@ export class RedisJobStore implements IJobStoreV2 { } else { pipeline.srem(KEYS.requiresActionJobs, streamId); } + if (job?.terminalHostActionPending === true) { + pipeline.sadd(KEYS.terminalHostActionJobs, streamId); + } else { + pipeline.srem(KEYS.terminalHostActionJobs, streamId); + } for (const userJobsKey of observedUserKeys) { if (userJobsKey !== activeUserKey) { pipeline.srem(userJobsKey, streamId); @@ -2428,6 +2441,12 @@ export class RedisJobStore implements IJobStoreV2 { if (terminal && patch?.terminalPersistencePending === true) { ttl = Math.max(ttl, TERMINAL_PERSISTENCE_RETENTION_TTL_S); } + if (terminal && patch?.terminalHostActionPending === true) { + // A terminal job owing a host hook must outlive the normal completed TTL so cleanup + // can still enumerate and retry it across restarts; the pause backstop (24h) bounds + // the retry window. Cleared to the completed TTL on acknowledgement. + ttl = Math.max(ttl, this.ttl.requiresAction); + } if (to === 'requires_action') { // A paused job must outlive its approval window, even when that window is // longer than the running TTL — otherwise Redis evicts it before a @@ -2682,6 +2701,49 @@ export class RedisJobStore implements IJobStoreV2 { ); } + async getTerminalHostActionJobs(): Promise { + const streamIds = await this.redis.smembers(KEYS.terminalHostActionJobs); + if (streamIds.length === 0) { + return []; + } + const jobs = await Promise.all(streamIds.map((streamId) => this.getJob(streamId))); + // The durable hash field is the source of truth; a stale set entry (job reaped, or the + // marker already cleared) is filtered out and self-heals via reconcileJobMembership. + const stale: string[] = []; + const pending: SerializableJobData[] = []; + for (let i = 0; i < streamIds.length; i++) { + const job = jobs[i]; + if (job != null && job.terminalHostActionPending === true) { + pending.push(job); + } else { + stale.push(streamIds[i]); + } + } + if (stale.length > 0) { + await this.redis.srem(KEYS.terminalHostActionJobs, ...stale).catch(() => undefined); + } + return pending; + } + + async clearTerminalHostAction(streamId: string, expectedCreatedAt?: number): Promise { + // Identity-fenced: only clear when the hash still holds this exact generation, so a + // replacement at the same streamId is never cleared through its predecessor. The HDEL + // and the completed-TTL reset happen atomically; membership is then reconciled to SREM. + const cleared = (await this.redis.eval( + 'if redis.call("HGET", KEYS[1], "createdAt") ~= ARGV[1] then return 0 end ' + + 'redis.call("HDEL", KEYS[1], "terminalHostActionPending") ' + + 'if tonumber(ARGV[2]) > 0 then redis.call("EXPIRE", KEYS[1], ARGV[2]) end ' + + 'return 1', + 1, + KEYS.job(streamId), + expectedCreatedAt != null ? String(expectedCreatedAt) : '', + String(this.ttl.completed), + )) as number; + if (cleared === 1) { + await this.redis.srem(KEYS.terminalHostActionJobs, streamId).catch(() => undefined); + } + } + async cleanup(): Promise { const now = Date.now(); const streamIds = await this.redis.smembers(KEYS.runningJobs); @@ -2849,6 +2911,10 @@ export class RedisJobStore implements IJobStoreV2 { patch: { error: 'Approval expired before a decision was made', completedAt: Date.now(), + // Store-won expiry: mark the host action pending (and extend retention via + // the TTL rule above) so the manager relay still runs its lifecycle hook on + // a replica that owns the runtime; cleared once that hook acknowledges. + terminalHostActionPending: true, }, // Scope the CAS to the action we observed as stale: if the user resolved it // and the run re-paused on a fresh action between the read and here, the @@ -4496,6 +4562,8 @@ export class RedisJobStore implements IJobStoreV2 { data.preserveForScheduleReconcile != null ? data.preserveForScheduleReconcile === '1' : undefined, + terminalHostActionPending: + data.terminalHostActionPending != null ? data.terminalHostActionPending === '1' : undefined, // Deferred tools discovered before a HITL pause; replayed into createRun on resume. discoveredTools: data.discoveredTools ? JSON.parse(data.discoveredTools) : undefined, activityPhaseSnapshot: data.activityPhaseSnapshot diff --git a/packages/api/src/stream/interfaces/IJobStore.ts b/packages/api/src/stream/interfaces/IJobStore.ts index 95525109c7..99a60a2af6 100644 --- a/packages/api/src/stream/interfaces/IJobStore.ts +++ b/packages/api/src/stream/interfaces/IJobStore.ts @@ -132,6 +132,14 @@ export interface SerializableJobData { scheduleOutcome?: 'success' | 'error' | 'interrupted' | 'skipped_balance'; scheduleOutcomeError?: string; preserveForScheduleReconcile?: boolean; + /** + * A terminal transition (currently approval expiry) still owes a durable host + * lifecycle hook. Set atomically with that transition and cleared only once the host + * adapter acknowledges success, so the job is retained (not reaped) and enumerable by + * cleanup across restarts and replicas until the hook completes. Generic: a host with + * no action clears it immediately on its no-op success, so nothing accumulates. + */ + terminalHostActionPending?: boolean; /** Serialized final event for replay */ finalEvent?: string; @@ -681,6 +689,15 @@ export interface IJobStore { /** Optional durable paused-job enumeration. Built-in stores implement it so * the manager can own approval expiry even after the original runtime died. */ getRequiresActionJobs?(): Promise; + /** Optional durable enumeration of terminal jobs that still owe a host lifecycle + * hook (see `terminalHostActionPending`). Built-in stores implement it so cleanup can + * retry the host adapter after a restart / on another replica, even though the job is + * no longer in the requires_action index. */ + getTerminalHostActionJobs?(): Promise; + /** Clears the pending-host-action marker once the adapter acknowledges success. + * Identity-fenced on `expectedCreatedAt` so a replacement generation at the same + * streamId is never cleared through its predecessor. */ + clearTerminalHostAction?(streamId: string, expectedCreatedAt?: number): Promise; cleanup(): Promise; recordActivity?(streamId: string, expectedCreatedAt?: number): void; getJobCount(): Promise;