fix(stream): Wave C.2 — durable retry/ack for terminal host lifecycle actions

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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CZwZPDBgxy4mCWWg6zQSjq

packages/api/src/stream/ApprovalLifecycle.ts

@@ -391,6 +391,7 @@ export class ApprovalLifecycle {
     streamId: string,
     expectedActionId?: string,
     expectedCreatedAt?: number,
+    options?: { markHostActionPending?: boolean },
   ): Promise<SerializableJobData | null> {
     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,
     });

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<void> {
     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;

packages/api/src/stream/__tests__/RedisJobStore.spec.ts

@@ -665,6 +665,7 @@ describe('RedisJobStore', () => {
     const evalResult = createDeferred<number>();
     const runningMembership = createDeferred<number>();
     const requiresActionRemoval = createDeferred<number>();
+    const terminalHostActionRemoval = createDeferred<number>();
     const userMembership = createDeferred<number>();
     const userExpiry = createDeferred<number>();
     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',

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<void> }).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<Promise<void>, [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();
   });

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<SerializableJobData[]> {
+    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<void> {
+    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<number> {
     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

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<SerializableJobData[]> {
+    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<void> {
+    // 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<number> {
     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

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<SerializableJobData[]>;
+  /** 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<SerializableJobData[]>;
+  /** 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<void>;
   cleanup(): Promise<number>;
   recordActivity?(streamId: string, expectedCreatedAt?: number): void;
   getJobCount(): Promise<number>;