🧳 fix: Port Subagent Control Receipt Writes to DocumentDB-Safe Operators (#15171)

* fix: harden subagent control receipt persistence

* fix: harden durable subagent control replay

* fix: await terminal control receipts on shutdown

* fix: close subagent control replay races

* test: type stale-owner transport fixture

* fix: quiesce durable subagent controls

* test: await subagent shutdown durability boundary

* fix: serialize durable subagent controls

* fix: fail shutdown on cleanup errors

* fix: report cancellable result availability accurately

* fix: fence subagent control receipt ownership

* fix: close distributed control receipt races

* test: type control receipt race fixture

* chore: require authoritative control receipts

* fix: close subagent control lifecycle races

* style: separate control reservation member

* test: harden subagent settlement wait

* fix: preserve authoritative control replay state

api/server/services/Endpoints/agents/subagentThreadStore.js

@@ -62,6 +62,7 @@ const subagentThreadTaskStore = createSubagentThreadTaskStore(
     deleteConvos: db.deleteConvos,
     deleteMessages: db.deleteMessages,
     getConvo: db.getConvo,
+    getSubagentTaskControlReplay: db.getSubagentTaskControlReplay,
     getMessages: db.getMessages,
     listActiveSubagentThreadLeases: db.listActiveSubagentThreadLeases,
     recordSubagentTaskControlReceipt: db.recordSubagentTaskControlReceipt,
@@ -93,6 +94,20 @@ registerShutdownTask(
 );
 
 let taskRoutingConfigured = false;
+let disconnectTaskRouting = () => {};
+
+/** Store quiescence is required even without Redis. Optional transport cleanup
+ * is attached after configuration, but local child cancellation and the final
+ * durable receipt flush always participate in graceful shutdown. */
+registerShutdownTask(
+  'subagent task store',
+  async () => {
+    await subagentThreadTaskStore.destroyTaskControlTransport();
+    subagentThreadTaskStore.destroyActivityStream();
+    disconnectTaskRouting();
+  },
+  { priority: 90 },
+);
 
 /** Starts the optional Redis owner directory before HTTP admission opens. */
 async function configureSubagentTaskRouting() {
@@ -126,17 +141,11 @@ async function configureSubagentTaskRouting() {
     throw error;
   }
   taskRoutingConfigured = true;
-  registerShutdownTask(
-    'subagent task control transport',
-    async () => {
-      await subagentThreadTaskStore.destroyTaskControlTransport();
-      subagentThreadTaskStore.destroyActivityStream();
-      publisher.disconnect();
-      activitySubscriber.disconnect();
-      activityPublisher.disconnect();
-    },
-    { priority: 90 },
-  );
+  disconnectTaskRouting = () => {
+    publisher.disconnect();
+    activitySubscriber.disconnect();
+    activityPublisher.disconnect();
+  };
 }
 
 module.exports = subagentThreadTaskStore;

api/server/services/Endpoints/agents/subagentThreadStore.spec.js

@@ -28,6 +28,7 @@ jest.mock('~/models', () => ({
   deleteConvos: jest.fn(),
   deleteMessages: jest.fn(),
   getConvo: jest.fn(),
+  getSubagentTaskControlReplay: jest.fn(),
   getMessages: jest.fn(),
   listActiveSubagentThreadLeases: jest.fn(),
   recordSubagentTaskControlReceipt: jest.fn(),
@@ -61,12 +62,16 @@ const db = require('~/models');
 const activityPrepareRegistration = registerShutdownTask.mock.calls.find(
   ([name]) => name === 'subagent activity streams prepare',
 );
+const taskStoreShutdownRegistration = registerShutdownTask.mock.calls.find(
+  ([name]) => name === 'subagent task store',
+);
 
 describe('subagent thread Redis lifecycle', () => {
   it('wires durable control receipt persistence into the host store', () => {
     expect(taskStoreMethods.recordSubagentTaskControlReceipt).toBe(
       db.recordSubagentTaskControlReceipt,
     );
+    expect(taskStoreMethods.getSubagentTaskControlReplay).toBe(db.getSubagentTaskControlReplay);
   });
 
   it('reads completion wakeup rollout state at task preparation time', async () => {
@@ -83,6 +88,14 @@ describe('subagent thread Redis lifecycle', () => {
     expect(subagentThreadTaskStore.completionWakeupsEnabled).toBe(false);
   });
 
+  it('registers local task-store quiescence independently of optional Redis setup', () => {
+    expect(taskStoreShutdownRegistration).toEqual([
+      'subagent task store',
+      expect.any(Function),
+      { priority: 90 },
+    ]);
+  });
+
   it('closes activity SSE before drain and disconnects its subscriber after drain', async () => {
     const taskSubscriber = { disconnect: jest.fn() };
     const activitySubscriber = { disconnect: jest.fn() };
@@ -102,18 +115,16 @@ describe('subagent thread Redis lifecycle', () => {
       expect.any(Function),
       { phase: 'pre-drain', priority: 100 },
     ]);
-    expect(registerShutdownTask).toHaveBeenCalledWith(
-      'subagent task control transport',
+    expect(taskStoreShutdownRegistration).toEqual([
+      'subagent task store',
       expect.any(Function),
       { priority: 90 },
-    );
+    ]);
     const prepare = activityPrepareRegistration[1];
     prepare();
     expect(mockTaskStore.prepareActivityForShutdown).toHaveBeenCalledTimes(1);
 
-    const shutdown = registerShutdownTask.mock.calls.find(
-      ([name]) => name === 'subagent task control transport',
-    )[1];
+    const shutdown = taskStoreShutdownRegistration[1];
     await shutdown();
 
     expect(mockTaskStore.destroyTaskControlTransport).toHaveBeenCalledTimes(1);

packages/api/src/agents/guard.spec.ts

@@ -37,6 +37,7 @@ function makeStore(): SubagentThreadTaskStore {
     deleteConvos: unused as AllMethods['deleteConvos'],
     deleteMessages: unused as AllMethods['deleteMessages'],
     getConvo: unused as AllMethods['getConvo'],
+    getSubagentTaskControlReplay: unused as AllMethods['getSubagentTaskControlReplay'],
     getMessages: unused as AllMethods['getMessages'],
     listActiveSubagentThreadLeases: unused as AllMethods['listActiveSubagentThreadLeases'],
     recordSubagentTaskControlReceipt: unused as AllMethods['recordSubagentTaskControlReceipt'],

packages/api/src/agents/subagentTaskRouting.spec.ts

@@ -177,6 +177,7 @@ function taskHandler(
     claim: () => ({ status: 'not_found' }),
     control: () => ({ status: 'not_found' }),
     list: () => [],
+    retainsTaskOwnership: () => false,
     cancelScope: () => 0,
     ...overrides,
   };
@@ -475,6 +476,49 @@ describe('RedisSubagentTaskControlTransport', () => {
     await Promise.all([owner.destroy(), requester.destroy()]);
   });
 
+  it('keeps a receipt-only owner registered until deletion cleanup can reach it', async () => {
+    const bus = new FakeRedisBus();
+    const owner = new RedisSubagentTaskControlTransport(
+      asRedis(bus.createClient()),
+      asRedis(bus.createClient()),
+      { namespace: 'test', instanceId: 'owner', registrationHeartbeatMs: 5 },
+    );
+    const requester = new RedisSubagentTaskControlTransport(
+      asRedis(bus.createClient()),
+      asRedis(bus.createClient()),
+      { namespace: 'test', instanceId: 'requester', requestTimeoutMs: 100, retryDelayMs: 5 },
+    );
+    let receiptPending = true;
+    const cancelScope = jest.fn(() => 0);
+    await owner.bind(
+      taskHandler({
+        retainsTaskOwnership: (_scopeId, taskId) => receiptPending && taskId === 'task-1',
+        cancelScope,
+      }),
+    );
+    await requester.bind(taskHandler());
+    await owner.registerTask('scope-1', 'task-1', 60_000);
+
+    /** Model the SDK task/result buckets dropping the task and Redis losing the
+     * directory entry before the next owner heartbeat. Pending receipt work is
+     * the only remaining reason this process can still handle deletion cleanup. */
+    bus.hashes.clear();
+    for (let attempt = 0; attempt < 100 && !(await requester.hasTasks('scope-1')); attempt += 1) {
+      await new Promise<void>((resolve) => setTimeout(resolve, 5));
+    }
+    await expect(requester.hasTasks('scope-1')).resolves.toBe(true);
+    await expect(requester.cancelScope('scope-1', null, ['deleted-child-thread'])).resolves.toBe(0);
+    expect(cancelScope).toHaveBeenCalledWith('scope-1', null, ['deleted-child-thread']);
+
+    receiptPending = false;
+    bus.hashes.clear();
+    for (let attempt = 0; attempt < 100 && (await requester.hasTasks('scope-1')); attempt += 1) {
+      await new Promise<void>((resolve) => setTimeout(resolve, 5));
+    }
+    await expect(requester.hasTasks('scope-1')).resolves.toBe(false);
+    await Promise.all([owner.destroy(), requester.destroy()]);
+  });
+
   it('expires a dead owner independently while another owner keeps the scope active', async () => {
     const bus = new FakeRedisBus();
     const deadOwner = new RedisSubagentTaskControlTransport(

packages/api/src/agents/subagentTaskRouting.ts

@@ -36,6 +36,7 @@ const MAX_PROGRESS_LABEL_CHARS = 1_024;
 /** Bounds the model-facing task list, per owner reply and across the merged result. */
 export const MAX_TASK_SNAPSHOTS = 200;
 const MAX_CANCEL_THREAD_IDS = 200;
+const MAX_REMOVED_CONVERSATION_IDS = MAX_CANCEL_THREAD_IDS + 1;
 /** Matches the deletion drain so bounded fan-out stays well inside the lease TTL. */
 const ROUTING_FANOUT_CONCURRENCY = 32;
 /** Contains every bounded response even when JSON escapes each retained character. */
@@ -97,7 +98,13 @@ type RoutedRequest = RoutedRequestBase &
         invocationId: string;
       }
     | { operation: 'list' }
-    | { operation: 'cancel'; threadIds: string[] | null }
+    | {
+        operation: 'cancel';
+        threadIds: string[] | null;
+        /** Rows already committed as deleted by the requester. Owners must drop
+         * receipt retry work for these exact conversations after cancellation. */
+        removedConversationIds?: string[];
+      }
   );
 
 type RoutedRequestPayload =
@@ -110,7 +117,12 @@ type RoutedRequestPayload =
       invocationId: string;
     }
   | { operation: 'list'; scopeId: string }
-  | { operation: 'cancel'; scopeId: string; threadIds: string[] | null };
+  | {
+      operation: 'cancel';
+      scopeId: string;
+      threadIds: string[] | null;
+      removedConversationIds?: string[];
+    };
 
 interface RoutedResponse {
   version: typeof PROTOCOL_VERSION;
@@ -174,7 +186,14 @@ export interface SubagentTaskControlHandler {
     invocationId: string,
   ): Promise<SubagentTaskControlResult> | SubagentTaskControlResult;
   list(scopeId: string): SubagentTaskSnapshot[];
-  cancelScope(scopeId: string, threadIds: string[] | null): number;
+  /** Receipt retry work can outlive the SDK task/result buckets. Keep its owner
+   * addressable so deletion can revoke work whose durable target was removed. */
+  retainsTaskOwnership(scopeId: string, taskId: string): boolean;
+  cancelScope(
+    scopeId: string,
+    threadIds: string[] | null,
+    removedConversationIds?: string[],
+  ): number;
 }
 
 /** Optional host transport for reaching the process that owns a live child task. */
@@ -190,7 +209,11 @@ export interface SubagentTaskControlTransport {
     invocationId: string,
   ): Promise<SubagentTaskControlResult | undefined>;
   list(scopeId: string): Promise<SubagentTaskSnapshot[]>;
-  cancelScope(scopeId: string, threadIds: string[] | null): Promise<number>;
+  cancelScope(
+    scopeId: string,
+    threadIds: string[] | null,
+    removedConversationIds?: string[],
+  ): Promise<number>;
   destroy(): Promise<void>;
 }
 
@@ -547,6 +570,7 @@ function parseRequest(value: unknown): RoutedRequest | undefined {
     taskId?: unknown;
     command?: unknown;
     threadIds?: unknown;
+    removedConversationIds?: unknown;
     invocationId?: unknown;
     expiresAt?: unknown;
   };
@@ -579,6 +603,14 @@ function parseRequest(value: unknown): RoutedRequest | undefined {
     if (candidate.threadIds !== null && !isCancelThreadIds(candidate.threadIds)) {
       return undefined;
     }
+    if (
+      candidate.removedConversationIds !== undefined &&
+      (!Array.isArray(candidate.removedConversationIds) ||
+        candidate.removedConversationIds.length > MAX_REMOVED_CONVERSATION_IDS ||
+        !candidate.removedConversationIds.every((id) => isBoundedString(id, MAX_THREAD_ID_CHARS)))
+    ) {
+      return undefined;
+    }
     return {
       version: PROTOCOL_VERSION,
       kind: 'request',
@@ -588,6 +620,9 @@ function parseRequest(value: unknown): RoutedRequest | undefined {
       operation: 'cancel',
       scopeId: candidate.scopeId,
       threadIds: candidate.threadIds,
+      ...(candidate.removedConversationIds === undefined
+        ? {}
+        : { removedConversationIds: candidate.removedConversationIds }),
     };
   }
   if (!isBoundedString(candidate.taskId, MAX_TASK_ID_CHARS)) {
@@ -852,7 +887,11 @@ export class RedisSubagentTaskControlTransport implements SubagentTaskControlTra
    * predicate to its complete local task set, so deletion never depends on the
    * bounded model-facing list and cannot miss a task beyond that cap.
    */
-  async cancelScope(scopeId: string, threadIds: string[] | null): Promise<number> {
+  async cancelScope(
+    scopeId: string,
+    threadIds: string[] | null,
+    removedConversationIds: string[] = [],
+  ): Promise<number> {
     this.assertScope(scopeId);
     if (threadIds != null && threadIds.length === 0) {
       return 0;
@@ -880,11 +919,24 @@ export class RedisSubagentTaskControlTransport implements SubagentTaskControlTra
     }
     const cancelSlot = createConcurrencyLimiter(ROUTING_FANOUT_CONCURRENCY);
     const requests: Array<Promise<unknown>> = [];
+    const allTargetThreadIds = threadIds == null ? null : new Set(threadIds);
     for (const ownerId of owners) {
       for (const batch of batches) {
+        const batchThreadIds = batch == null ? null : new Set(batch);
+        const removedForBatch = removedConversationIds.filter(
+          (conversationId) =>
+            allTargetThreadIds == null ||
+            !allTargetThreadIds.has(conversationId) ||
+            batchThreadIds?.has(conversationId) === true,
+        );
         requests.push(
           cancelSlot(() =>
-            this.sendRequest(ownerId, { operation: 'cancel', scopeId, threadIds: batch }),
+            this.sendRequest(ownerId, {
+              operation: 'cancel',
+              scopeId,
+              threadIds: batch,
+              ...(removedForBatch.length === 0 ? {} : { removedConversationIds: removedForBatch }),
+            }),
           ),
         );
       }
@@ -1014,7 +1066,13 @@ export class RedisSubagentTaskControlTransport implements SubagentTaskControlTra
           truncated: tasks.length > bounded.length,
         };
       } else if (request.operation === 'cancel') {
-        result = { cancelled: handler.cancelScope(request.scopeId, request.threadIds) };
+        result = {
+          cancelled: handler.cancelScope(
+            request.scopeId,
+            request.threadIds,
+            request.removedConversationIds,
+          ),
+        };
       } else if (request.operation === 'claim') {
         const claim = boundedClaim(handler.claim(request.scopeId, request.taskId));
         replayable = consumesResult(claim);
@@ -1286,7 +1344,8 @@ export class RedisSubagentTaskControlTransport implements SubagentTaskControlTra
        * address outlives the task itself until the result is acknowledged. */
       if (
         localTaskIds.has(taskId) ||
-        this.claimReplays.entries.has(this.claimReplayKey(scopeId, taskId))
+        this.claimReplays.entries.has(this.claimReplayKey(scopeId, taskId)) ||
+        handler.retainsTaskOwnership(scopeId, taskId)
       ) {
         retained.push(registration);
         continue;

packages/api/src/agents/view.spec.ts

@@ -207,6 +207,14 @@ describe('subagent thread parent-scoped view', () => {
   it('returns bounded authoritative control receipts without private fingerprints', async () => {
     const input = message('task-1:user', 'running', true);
     input.subagentTask!.controlReceipts = [
+      {
+        invocationId: 'private-reservation',
+        fingerprint: 'private-reservation-fingerprint',
+        action: 'queue' as const,
+        status: 'reserved' as const,
+        createdAt: new Date('2026-08-21T09:59:59.000Z'),
+        updatedAt: new Date('2026-08-21T09:59:59.000Z'),
+      },
       ...Array.from({ length: 32 }, (_, index) => ({
         invocationId: `earlier-${index}`,
         fingerprint: `private-${index}`,
@@ -258,6 +266,7 @@ describe('subagent thread parent-scoped view', () => {
     expect(Buffer.byteLength(projected?.message ?? '', 'utf8')).toBeLessThanOrEqual(512);
     expect(view.controlReceipts).toHaveLength(32);
     expect(view.controlReceiptsTruncated).toBe(true);
+    expect(JSON.stringify(view)).not.toContain('private-reservation');
     expect(JSON.stringify(view)).not.toContain('private-fingerprint');
   });
 

packages/api/src/agents/view.ts

@@ -125,8 +125,17 @@ const publicControlReceipts = (
 ): { receipts: SubagentControlReceipt[]; truncated: boolean } => {
   const input = messages.find((message) => message.messageId === `${taskId}:user`);
   const stored = input?.subagentTask?.controlReceipts ?? [];
-  const accepted = stored.filter((receipt) => receipt.status === 'accepted');
-  const terminal = stored.filter((receipt) => receipt.status !== 'accepted');
+  /** A reservation only fences at-most-once application; it does not claim that
+   * guidance was accepted and must never appear in the public activity view. */
+  const visible = stored.filter(
+    (
+      receipt,
+    ): receipt is typeof receipt & {
+      status: 'accepted' | 'applied' | 'rejected' | 'failed';
+    } => receipt.status !== 'reserved',
+  );
+  const accepted = visible.filter((receipt) => receipt.status === 'accepted');
+  const terminal = visible.filter((receipt) => receipt.status !== 'accepted');
   const terminalLimit = Math.max(0, MAX_PUBLIC_CONTROL_RECEIPTS - accepted.length);
   const retained = [...accepted, ...(terminalLimit === 0 ? [] : terminal.slice(-terminalLimit))]
     .slice(0, MAX_PUBLIC_CONTROL_RECEIPTS)
@@ -154,7 +163,7 @@ const publicControlReceipts = (
           : {}),
       };
     });
-  return { receipts: retained, truncated: retained.length < stored.length };
+  return { receipts: retained, truncated: retained.length < visible.length };
 };
 
 const publicStatus = (

packages/api/src/app/shutdown.spec.ts

@@ -301,7 +301,7 @@ describe('setupGracefulShutdown', () => {
     expect(order).toEqual(['generation streams', 'default-first', 'default-second', 'telemetry']);
   });
 
-  it('continues subsequent tasks and still exits if one task throws', async () => {
+  it('continues subsequent tasks and exits nonzero if one task throws', async () => {
     const calls: string[] = [];
     jest.spyOn(server, 'close').mockImplementation((cb?: (err?: Error) => void) => {
       if (cb) {
@@ -324,7 +324,7 @@ describe('setupGracefulShutdown', () => {
     await flush();
     await flush();
     expect(calls).toEqual(['ok-before', 'throws', 'ok-after']);
-    expect(exitSpy).toHaveBeenCalledWith(0);
+    expect(exitSpy).toHaveBeenCalledWith(1);
   });
 
   it('awaits async tasks before exiting', async () => {

packages/api/src/app/shutdown.ts

@@ -81,7 +81,7 @@ export function __resetShutdownStateForTests(): void {
   clearForceExitTimer();
 }
 
-async function runShutdownTasks(phase: ShutdownPhase): Promise<void> {
+async function runShutdownTasks(phase: ShutdownPhase): Promise<boolean> {
   const orderedTasks = tasks
     .filter((task) => task.phase === phase)
     .sort(
@@ -89,14 +89,17 @@ async function runShutdownTasks(phase: ShutdownPhase): Promise<void> {
         right.priority - left.priority || left.registrationOrder - right.registrationOrder,
     );
 
+  let failed = false;
   for (const task of orderedTasks) {
     try {
       logger.info(`Running ${phase} shutdown task: ${task.name}`);
       await task.fn();
     } catch (err) {
+      failed = true;
       logger.error(`Shutdown task "${task.name}" failed:`, err);
     }
   }
+  return failed;
 }
 
 function clearForceExitTimer(): void {
@@ -130,9 +133,9 @@ async function shutdown(signal: NodeJS.Signals): Promise<void> {
       exitCode = 1;
     });
 
-    await runShutdownTasks('pre-drain');
+    if (await runShutdownTasks('pre-drain')) exitCode = 1;
     await serverClosePromise;
-    await runShutdownTasks('post-drain');
+    if (await runShutdownTasks('post-drain')) exitCode = 1;
   } finally {
     clearTimeout(forceExit);
     if (forceExitTimer === forceExit) {

packages/data-schemas/src/methods/message.spec.ts

@@ -42,6 +42,12 @@ let claimSubagentTaskResult: ReturnType<typeof createMessageMethods>['claimSubag
 let recordSubagentTaskControlReceipt: ReturnType<
   typeof createMessageMethods
 >['recordSubagentTaskControlReceipt'];
+let getSubagentTaskControlReceipt: ReturnType<
+  typeof createMessageMethods
+>['getSubagentTaskControlReceipt'];
+let getSubagentTaskControlReplay: ReturnType<
+  typeof createMessageMethods
+>['getSubagentTaskControlReplay'];
 let releaseSubagentTaskResultClaim: ReturnType<
   typeof createMessageMethods
 >['releaseSubagentTaskResultClaim'];
@@ -68,6 +74,8 @@ beforeAll(async () => {
   recordMessage = methods.recordMessage;
   claimSubagentTaskResult = methods.claimSubagentTaskResult;
   recordSubagentTaskControlReceipt = methods.recordSubagentTaskControlReceipt;
+  getSubagentTaskControlReceipt = methods.getSubagentTaskControlReceipt;
+  getSubagentTaskControlReplay = methods.getSubagentTaskControlReplay;
   releaseSubagentTaskResultClaim = methods.releaseSubagentTaskResultClaim;
 
   await mongoose.connect(mongoUri);
@@ -97,6 +105,7 @@ describe('Message Operations', () => {
 
     // Clear database
     await Message.deleteMany({});
+    await mongoose.models.Conversation.deleteMany({});
 
     mockCtx = {
       userId: 'user123',
@@ -2334,6 +2343,25 @@ describe('Message Operations', () => {
         message: 'Use the primary source.',
       };
 
+      await expect(
+        recordSubagentTaskControlReceipt({
+          userId: 'user123',
+          conversationId,
+          taskId: 'task-1',
+          receipt: { ...accepted, controlId: undefined, status: 'reserved' },
+        }),
+      ).resolves.toBe(true);
+      /** A reservation fences competing owners but is not public evidence that
+       * the command was accepted or applied. */
+      await expect(
+        getSubagentTaskControlReceipt({
+          userId: 'user123',
+          conversationId,
+          taskId: 'task-1',
+          invocationId: 'invocation-1',
+        }),
+      ).resolves.toBeNull();
+
       await expect(
         recordSubagentTaskControlReceipt({
           userId: 'user123',
@@ -2356,12 +2384,14 @@ describe('Message Operations', () => {
         }),
       ).resolves.toBe(true);
       /** A delayed accepted replay cannot downgrade the durable terminal receipt. */
-      await recordSubagentTaskControlReceipt({
-        userId: 'user123',
-        conversationId,
-        taskId: 'task-1',
-        receipt: accepted,
-      });
+      await expect(
+        recordSubagentTaskControlReceipt({
+          userId: 'user123',
+          conversationId,
+          taskId: 'task-1',
+          receipt: accepted,
+        }),
+      ).resolves.toBe('unchanged');
       await expect(
         recordSubagentTaskControlReceipt({
           userId: 'another-user',
@@ -2448,6 +2478,317 @@ describe('Message Operations', () => {
       ]);
     });
 
+    it('reads an exact authorized receipt and persists a new terminal rejection', async () => {
+      const conversationId = uuidv4();
+      await createTaskInput(conversationId);
+      await mongoose.models.Conversation.create({
+        user: 'user123',
+        conversationId,
+        title: 'Child thread',
+        endpoint: 'agents',
+        subagentThread: {
+          rootConversationId: 'parent-conversation',
+          parentConversationId: 'parent-conversation',
+          parentMessageId: 'parent-message',
+          parentToolCallId: 'parent-tool',
+          subagentType: 'researcher',
+          subagentKind: 'agent',
+          depth: 1,
+        },
+      });
+      await Message.create({
+        user: 'user123',
+        conversationId,
+        messageId: 'task-1:assistant',
+        parentMessageId: 'task-1:user',
+        sender: 'researcher',
+        text: 'Done.',
+        endpoint: 'agents',
+        isCreatedByUser: false,
+        subagentTask: {
+          attemptKey: 'task-1-attempt',
+          parentRunId: 'parent-message',
+          status: 'completed',
+          resultClaim: {
+            kind: 'manual',
+            claimId: 'poll-1',
+            claimedAt: new Date('2026-08-24T12:00:02.000Z'),
+          },
+        },
+      });
+      const now = new Date('2026-08-24T12:00:00.000Z');
+      await expect(
+        recordSubagentTaskControlReceipt({
+          userId: 'user123',
+          conversationId,
+          taskId: 'task-1',
+          receipt: {
+            invocationId: 'terminal-invocation',
+            fingerprint: 'terminal-fingerprint',
+            action: 'cancel',
+            status: 'rejected',
+            reason: 'task_not_running',
+            createdAt: now,
+            updatedAt: now,
+          },
+        }),
+      ).resolves.toBe(true);
+
+      await expect(
+        getSubagentTaskControlReceipt({
+          userId: 'user123',
+          conversationId,
+          taskId: 'task-1',
+          invocationId: 'terminal-invocation',
+        }),
+      ).resolves.toEqual(
+        expect.objectContaining({
+          invocationId: 'terminal-invocation',
+          fingerprint: 'terminal-fingerprint',
+          status: 'rejected',
+          reason: 'task_not_running',
+        }),
+      );
+      await expect(
+        getSubagentTaskControlReceipt({
+          userId: 'another-user',
+          conversationId,
+          taskId: 'task-1',
+          invocationId: 'terminal-invocation',
+        }),
+      ).resolves.toBeNull();
+      await expect(
+        getSubagentTaskControlReplay({
+          userId: 'user123',
+          parentConversationId: 'parent-conversation',
+          taskId: 'task-1',
+          invocationId: 'terminal-invocation',
+        }),
+      ).resolves.toEqual({
+        receipt: expect.objectContaining({ invocationId: 'terminal-invocation' }),
+        task: expect.objectContaining({
+          taskId: 'task-1',
+          threadId: conversationId,
+          subagentType: 'researcher',
+          status: 'completed',
+          resultAvailable: true,
+          resultClaimed: true,
+        }),
+      });
+      await expect(
+        getSubagentTaskControlReplay({
+          userId: 'user123',
+          parentConversationId: 'different-parent',
+          taskId: 'task-1',
+          invocationId: 'terminal-invocation',
+        }),
+      ).resolves.toBeNull();
+
+      const ordinaryConversationId = uuidv4();
+      await Message.create({
+        user: 'user123',
+        conversationId: ordinaryConversationId,
+        messageId: 'ordinary-task:user',
+        parentMessageId: Constants.NO_PARENT,
+        sender: 'User',
+        text: 'An ordinary message with a colliding id.',
+        endpoint: 'agents',
+        isCreatedByUser: true,
+      });
+      await expect(
+        recordSubagentTaskControlReceipt({
+          userId: 'user123',
+          conversationId: ordinaryConversationId,
+          taskId: 'ordinary-task',
+          receipt: {
+            invocationId: 'terminal-invocation',
+            fingerprint: 'terminal-fingerprint',
+            action: 'cancel',
+            status: 'rejected',
+            reason: 'task_not_running',
+            createdAt: now,
+            updatedAt: now,
+          },
+        }),
+      ).resolves.toBe(false);
+    });
+
+    it('replays an applied cancellation as cancelled before its terminal row exists', async () => {
+      const conversationId = uuidv4();
+      await createTaskInput(conversationId);
+      await mongoose.models.Conversation.create({
+        user: 'user123',
+        conversationId,
+        title: 'Cancelling child thread',
+        endpoint: 'agents',
+        subagentThread: {
+          rootConversationId: 'parent-conversation',
+          parentConversationId: 'parent-conversation',
+          parentMessageId: 'parent-message',
+          parentToolCallId: 'parent-tool',
+          subagentType: 'researcher',
+          subagentKind: 'agent',
+          depth: 1,
+        },
+      });
+      const now = new Date('2026-08-24T12:00:00.000Z');
+      await expect(
+        recordSubagentTaskControlReceipt({
+          userId: 'user123',
+          conversationId,
+          taskId: 'task-1',
+          receipt: {
+            invocationId: 'cancel-invocation',
+            fingerprint: 'cancel-fingerprint',
+            action: 'cancel',
+            status: 'applied',
+            createdAt: now,
+            updatedAt: now,
+          },
+        }),
+      ).resolves.toBe(true);
+
+      await expect(
+        getSubagentTaskControlReplay({
+          userId: 'user123',
+          parentConversationId: 'parent-conversation',
+          taskId: 'task-1',
+          invocationId: 'cancel-invocation',
+        }),
+      ).resolves.toEqual({
+        receipt: expect.objectContaining({ invocationId: 'cancel-invocation' }),
+        task: expect.objectContaining({
+          taskId: 'task-1',
+          threadId: conversationId,
+          status: 'cancelled',
+          resultAvailable: false,
+          resultClaimed: false,
+          updatedAt: now,
+        }),
+      });
+    });
+
+    it('reports every accepted control when replaying one durable invocation', async () => {
+      const conversationId = uuidv4();
+      await createTaskInput(conversationId);
+      await mongoose.models.Conversation.create({
+        user: 'user123',
+        conversationId,
+        title: 'Controlled child thread',
+        endpoint: 'agents',
+        subagentThread: {
+          rootConversationId: 'parent-conversation',
+          parentConversationId: 'parent-conversation',
+          parentMessageId: 'parent-message',
+          parentToolCallId: 'parent-tool',
+          subagentType: 'researcher',
+          subagentKind: 'agent',
+          depth: 1,
+        },
+      });
+      const now = new Date('2026-08-24T12:00:00.000Z');
+      for (const index of [1, 2]) {
+        await expect(
+          recordSubagentTaskControlReceipt({
+            userId: 'user123',
+            conversationId,
+            taskId: 'task-1',
+            receipt: {
+              invocationId: `pending-invocation-${index}`,
+              fingerprint: `pending-fingerprint-${index}`,
+              controlId: `pending-control-${index}`,
+              action: 'queue',
+              status: 'accepted',
+              createdAt: now,
+              updatedAt: now,
+            },
+          }),
+        ).resolves.toBe(true);
+      }
+
+      await expect(
+        getSubagentTaskControlReplay({
+          userId: 'user123',
+          parentConversationId: 'parent-conversation',
+          taskId: 'task-1',
+          invocationId: 'pending-invocation-1',
+        }),
+      ).resolves.toEqual({
+        receipt: expect.objectContaining({ invocationId: 'pending-invocation-1' }),
+        task: expect.objectContaining({ pendingControls: 2 }),
+      });
+    });
+
+    it('rejects an invocation fingerprint conflict without reporting persistence', async () => {
+      const conversationId = uuidv4();
+      await createTaskInput(conversationId);
+      const now = new Date('2026-08-24T12:00:00.000Z');
+      const receipt = {
+        invocationId: 'conflicting-invocation',
+        fingerprint: 'first-fingerprint',
+        controlId: 'first-control',
+        action: 'queue' as const,
+        status: 'accepted' as const,
+        createdAt: now,
+        updatedAt: now,
+      };
+      await expect(
+        recordSubagentTaskControlReceipt({
+          userId: 'user123',
+          conversationId,
+          taskId: 'task-1',
+          receipt,
+        }),
+      ).resolves.toBe(true);
+      await expect(
+        recordSubagentTaskControlReceipt({
+          userId: 'user123',
+          conversationId,
+          taskId: 'task-1',
+          receipt: { ...receipt, fingerprint: 'different-fingerprint' },
+        }),
+      ).resolves.toBe('conflict');
+    });
+
+    it('retains concurrent receipts without requiring an aggregation-pipeline update', async () => {
+      const conversationId = uuidv4();
+      await createTaskInput(conversationId);
+      const createdAt = new Date('2026-08-24T12:00:00.000Z');
+
+      const invocationIds = Array.from({ length: 32 }, (_, index) => `concurrent-${index}`);
+      await Promise.all(
+        invocationIds.map((invocationId, index) =>
+          recordSubagentTaskControlReceipt({
+            userId: 'user123',
+            conversationId,
+            taskId: 'task-1',
+            receipt: {
+              invocationId,
+              fingerprint: `${invocationId}-fingerprint`,
+              controlId: `${invocationId}-control`,
+              action: 'queue',
+              status: 'accepted',
+              createdAt: new Date(createdAt.getTime() + index),
+              updatedAt: new Date(createdAt.getTime() + index),
+            },
+          }),
+        ),
+      );
+
+      const stored = await Message.findOne({
+        user: 'user123',
+        conversationId,
+        messageId: 'task-1:user',
+      })
+        .select('+subagentTask')
+        .lean<IMessage>();
+      expect(stored?.subagentTask?.controlReceipts).toEqual(
+        expect.arrayContaining(
+          invocationIds.map((invocationId) => expect.objectContaining({ invocationId })),
+        ),
+      );
+    });
+
     it('retains accepted commands while bounding terminal receipt history', async () => {
       const conversationId = uuidv4();
       await createTaskInput(conversationId);
@@ -2537,26 +2878,31 @@ describe('Message Operations', () => {
       );
     });
 
-    it('defensively caps accepted receipts outside the supported task-store path', async () => {
+    it('refuses to evict active receipt fences at durable capacity', async () => {
       const conversationId = uuidv4();
       await createTaskInput(conversationId);
       const createdAt = new Date('2026-08-24T12:00:00.000Z');
+      const results: Array<boolean | 'unchanged' | 'conflict'> = [];
       for (let index = 0; index < 70; index += 1) {
-        await recordSubagentTaskControlReceipt({
-          userId: 'user123',
-          conversationId,
-          taskId: 'task-1',
-          receipt: {
-            invocationId: `accepted-${index}`,
-            fingerprint: `fingerprint-${index}`,
-            controlId: `control-${index}`,
-            action: 'queue',
-            status: 'accepted',
-            createdAt: new Date(createdAt.getTime() + index),
-            updatedAt: new Date(createdAt.getTime() + index),
-          },
-        });
+        results.push(
+          await recordSubagentTaskControlReceipt({
+            userId: 'user123',
+            conversationId,
+            taskId: 'task-1',
+            receipt: {
+              invocationId: `accepted-${index}`,
+              fingerprint: `fingerprint-${index}`,
+              controlId: `control-${index}`,
+              action: 'queue',
+              status: 'accepted',
+              createdAt: new Date(createdAt.getTime() + index),
+              updatedAt: new Date(createdAt.getTime() + index),
+            },
+          }),
+        );
       }
+      expect(results.slice(0, 64)).toEqual(Array.from({ length: 64 }, () => true));
+      expect(results.slice(64)).toEqual(Array.from({ length: 6 }, () => false));
 
       const stored = await Message.findOne({
         user: 'user123',
@@ -2566,7 +2912,73 @@ describe('Message Operations', () => {
         .select('+subagentTask')
         .lean<IMessage>();
       expect(stored?.subagentTask?.controlReceipts).toHaveLength(64);
-      expect(stored?.subagentTask?.controlReceipts?.[0]?.invocationId).toBe('accepted-6');
+      expect(stored?.subagentTask?.controlReceipts?.[0]?.invocationId).toBe('accepted-0');
+      expect(stored?.subagentTask?.controlReceipts?.[63]?.invocationId).toBe('accepted-63');
+
+      await expect(
+        recordSubagentTaskControlReceipt({
+          userId: 'user123',
+          conversationId,
+          taskId: 'task-1',
+          receipt: {
+            invocationId: 'terminal-with-no-allowance',
+            fingerprint: 'terminal-fingerprint',
+            controlId: 'terminal-control',
+            action: 'queue',
+            status: 'applied',
+            createdAt: new Date(createdAt.getTime() + 100),
+            updatedAt: new Date(createdAt.getTime() + 100),
+            boundary: 'turn',
+          },
+        }),
+      ).resolves.toBe(false);
+      const afterTerminal = await Message.findOne({
+        user: 'user123',
+        conversationId,
+        messageId: 'task-1:user',
+      })
+        .select('+subagentTask')
+        .lean<IMessage>();
+      expect(afterTerminal?.subagentTask?.controlReceipts).toHaveLength(64);
+      expect(afterTerminal?.subagentTask?.controlReceipts).not.toEqual(
+        expect.arrayContaining([
+          expect.objectContaining({ invocationId: 'terminal-with-no-allowance' }),
+        ]),
+      );
+
+      /** Completing an existing active fence always frees its own slot and wins
+       * over terminal history, even though its occurrence timestamp is oldest. */
+      await expect(
+        recordSubagentTaskControlReceipt({
+          userId: 'user123',
+          conversationId,
+          taskId: 'task-1',
+          receipt: {
+            invocationId: 'accepted-0',
+            fingerprint: 'fingerprint-0',
+            controlId: 'control-0',
+            action: 'queue',
+            status: 'applied',
+            createdAt,
+            updatedAt: new Date(createdAt.getTime() + 101),
+            boundary: 'turn',
+          },
+        }),
+      ).resolves.toBe(true);
+      const afterTransition = await Message.findOne({
+        user: 'user123',
+        conversationId,
+        messageId: 'task-1:user',
+      })
+        .select('+subagentTask')
+        .lean<IMessage>();
+      expect(afterTransition?.subagentTask?.controlReceipts).toHaveLength(64);
+      expect(afterTransition?.subagentTask?.controlReceipts).toEqual(
+        expect.arrayContaining([
+          expect.objectContaining({ invocationId: 'accepted-0', status: 'applied' }),
+          expect.objectContaining({ invocationId: 'accepted-63', status: 'accepted' }),
+        ]),
+      );
     });
   });
 

packages/data-schemas/src/methods/message.ts

@@ -1,9 +1,9 @@
 import { HITL_MESSAGE_FILTER_FIELDS, RetentionMode } from 'librechat-data-provider';
 import type { UserSubmittedMessageFieldPath } from 'librechat-data-provider';
 import type { DeleteResult, FilterQuery, Model, Types } from 'mongoose';
-import type { AppConfig, IMessage } from '~/types';
+import type { AppConfig, IConversation, IMessage } from '~/types';
+import { activeExpirationFilter, createFallbackRetentionDate } from '~/utils/retention';
 import { createTempChatExpirationDate } from '~/utils/tempChatRetention';
-import { createFallbackRetentionDate } from '~/utils/retention';
 import { tenantSafeBulkWrite } from '~/utils/tenantBulkWrite';
 import logger from '~/config/winston';
 
@@ -15,6 +15,9 @@ const MAX_STORED_USER_SUBMITTED_FIELD_PATHS = MAX_NORMALIZED_USER_SUBMITTED_PATH
 const MAX_USER_SUBMITTED_PATH_LENGTH = 2048;
 const MAX_SUBAGENT_CONTROL_RECEIPTS = 64;
 const MAX_SUBAGENT_CONTROL_MESSAGE_LENGTH = 4 * 1024;
+/** One owner admits at most 64 terminal control invocations. The optimistic
+ * writer therefore has enough rounds for every admitted receipt to converge. */
+const MAX_SUBAGENT_CONTROL_RECEIPT_CAS_ATTEMPTS = 64;
 const PROVENANCE_PATHS_UNION_FIELD = '__lcProvenancePathsUnion';
 const PROVENANCE_FIELD_PATHS_UNION_FIELD = '__lcProvenanceFieldPathsUnion';
 const HITL_MESSAGE_FILTER_FIELD_SET = new Set<string>(HITL_MESSAGE_FILTER_FIELDS);
@@ -119,6 +122,92 @@ function getStrictPipelineUpdate(Message: Model<IMessage>, update: Record<string
   return Message.castObject(candidate) as unknown as Record<string, unknown>;
 }
 
+type StoredSubagentControlReceipt = NonNullable<
+  NonNullable<IMessage['subagentTask']>['controlReceipts']
+>[number];
+
+const terminalControlReceipt = (receipt: StoredSubagentControlReceipt): boolean =>
+  receipt.status === 'applied' || receipt.status === 'rejected' || receipt.status === 'failed';
+
+function retainSubagentControlReceipts(
+  current: StoredSubagentControlReceipt[],
+  receipt: StoredSubagentControlReceipt,
+): {
+  status: 'updated' | 'unchanged' | 'conflict' | 'capacity';
+  receipts: StoredSubagentControlReceipt[];
+} {
+  const existingIndex = current.findIndex(
+    (candidate) => candidate.invocationId === receipt.invocationId,
+  );
+  let merged: StoredSubagentControlReceipt[];
+  if (existingIndex < 0) {
+    merged = [...current, receipt];
+  } else {
+    const existing = current[existingIndex];
+    if (existing.fingerprint !== receipt.fingerprint) {
+      return { status: 'conflict', receipts: current };
+    }
+    if (
+      terminalControlReceipt(existing) ||
+      existing.status === receipt.status ||
+      (existing.status === 'accepted' && receipt.status === 'reserved')
+    ) {
+      return { status: 'unchanged', receipts: current };
+    }
+    merged = current.map((candidate, index) => (index === existingIndex ? receipt : candidate));
+  }
+  const accepted = merged.filter(
+    (candidate) => candidate.status === 'reserved' || candidate.status === 'accepted',
+  );
+  /** Reserved and accepted receipts are idempotency fences for commands that can
+   * still take effect. Never evict one to admit another receipt: report capacity
+   * so the caller refuses the command before mutating the live task. */
+  if (accepted.length > MAX_SUBAGENT_CONTROL_RECEIPTS) {
+    return { status: 'capacity', receipts: current };
+  }
+  const terminalAllowance = Math.max(0, MAX_SUBAGENT_CONTROL_RECEIPTS - accepted.length);
+  let terminal =
+    terminalAllowance === 0
+      ? []
+      : merged
+          .filter((candidate) => candidate.status !== 'reserved' && candidate.status !== 'accepted')
+          .sort(
+            (left, right) =>
+              left.createdAt.getTime() - right.createdAt.getTime() ||
+              left.invocationId.localeCompare(right.invocationId),
+          )
+          .slice(-terminalAllowance);
+  const advancesActiveFence =
+    existingIndex >= 0 &&
+    !terminalControlReceipt(current[existingIndex]) &&
+    terminalControlReceipt(receipt);
+  if (
+    advancesActiveFence &&
+    !terminal.some((candidate) => candidate.invocationId === receipt.invocationId)
+  ) {
+    /** A terminal transition for an active fence must outrank unrelated terminal
+     * history even though it retains the command's older occurrence timestamp. */
+    const otherAllowance = Math.max(0, terminalAllowance - 1);
+    terminal = [
+      ...(otherAllowance === 0
+        ? []
+        : terminal
+            .filter((candidate) => candidate.invocationId !== receipt.invocationId)
+            .slice(-otherAllowance)),
+      receipt,
+    ].sort(
+      (left, right) =>
+        left.createdAt.getTime() - right.createdAt.getTime() ||
+        left.invocationId.localeCompare(right.invocationId),
+    );
+  }
+  const receipts = [...accepted, ...terminal];
+  if (!receipts.some((candidate) => candidate.invocationId === receipt.invocationId)) {
+    return { status: 'capacity', receipts: current };
+  }
+  return { status: 'updated', receipts };
+}
+
 /**
  * Builds one Mongo aggregation update that merges and caps both provenance
  * sets. Generic path overflow promotes the message to whole-message user
@@ -291,7 +380,34 @@ export interface MessageMethods {
     taskId: string;
     tenantId?: string;
     receipt: NonNullable<NonNullable<IMessage['subagentTask']>['controlReceipts']>[number];
-  }): Promise<boolean>;
+  }): Promise<boolean | 'unchanged' | 'conflict'>;
+  getSubagentTaskControlReceipt(input: {
+    userId: string;
+    conversationId: string;
+    taskId: string;
+    invocationId: string;
+    tenantId?: string;
+  }): Promise<NonNullable<NonNullable<IMessage['subagentTask']>['controlReceipts']>[number] | null>;
+  getSubagentTaskControlReplay(input: {
+    userId: string;
+    parentConversationId: string;
+    taskId: string;
+    invocationId: string;
+    tenantId?: string;
+  }): Promise<{
+    receipt: NonNullable<NonNullable<IMessage['subagentTask']>['controlReceipts']>[number];
+    task: {
+      taskId: string;
+      threadId: string;
+      subagentType: string;
+      status: NonNullable<IMessage['subagentTask']>['status'];
+      resultAvailable: boolean;
+      resultClaimed: boolean;
+      pendingControls: number;
+      createdAt: Date;
+      updatedAt: Date;
+    };
+  } | null>;
   bulkSaveMessages(
     messages: Array<Partial<IMessage>>,
     overrideTimestamp?: boolean,
@@ -892,9 +1008,9 @@ export function createMessageMethods(mongoose: typeof import('mongoose')): Messa
     taskId: string;
     tenantId?: string;
     receipt: NonNullable<NonNullable<IMessage['subagentTask']>['controlReceipts']>[number];
-  }): Promise<boolean> {
+  }): Promise<boolean | 'unchanged' | 'conflict'> {
     const validActions = new Set(['steer', 'queue', 'interrupt', 'cancel', 'cancel_message']);
-    const validStatuses = new Set(['accepted', 'applied', 'rejected', 'failed']);
+    const validStatuses = new Set(['reserved', 'accepted', 'applied', 'rejected', 'failed']);
     if (
       userId.length === 0 ||
       conversationId.length === 0 ||
@@ -915,233 +1031,211 @@ export function createMessageMethods(mongoose: typeof import('mongoose')): Messa
       throw new TypeError('Invalid subagent task control receipt');
     }
     const Message = mongoose.models.Message as Model<IMessage>;
-    const terminalStatuses = ['applied', 'rejected', 'failed'];
-    const updated = await Message.findOneAndUpdate(
-      {
-        user: userId,
-        conversationId,
-        ...(tenantId == null ? { tenantId: { $exists: false } } : { tenantId }),
-        messageId: `${taskId}:user`,
-        'subagentTask.status': 'running',
+    const recordsTerminalRejection =
+      receipt.status === 'rejected' && receipt.reason === 'task_not_running';
+    const identity = {
+      user: userId,
+      conversationId,
+      ...(tenantId == null ? { tenantId: { $exists: false } } : { tenantId }),
+      messageId: `${taskId}:user`,
+      /** A genuinely new command can arrive after its task settles or its final
+       * lease expires. Persist that authoritative rejection for retries; every
+       * command that could still be applied remains fenced to a running task. */
+      ...(recordsTerminalRejection
+        ? { 'subagentTask.status': { $in: ['running', 'completed', 'error', 'cancelled'] } }
+        : { 'subagentTask.status': 'running' }),
+    };
+    /** Amazon DocumentDB does not support aggregation-pipeline updates. Use a
+     * bounded optimistic compare-and-swap: the read is small, the write uses
+     * only plain operators, and concurrent writers retry rather than overwrite. */
+    for (let attempt = 0; attempt < MAX_SUBAGENT_CONTROL_RECEIPT_CAS_ATTEMPTS; attempt += 1) {
+      const currentMessage = await Message.findOne(identity)
+        .select({ 'subagentTask.controlReceipts': 1, _id: 0 })
+        .lean<Pick<IMessage, 'subagentTask'> | null>();
+      if (currentMessage == null) return false;
+      const current = currentMessage.subagentTask?.controlReceipts ?? [];
+      const retained = retainSubagentControlReceipts(current, receipt);
+      if (retained.status === 'conflict') return 'conflict';
+      if (retained.status === 'unchanged') return 'unchanged';
+      if (retained.status === 'capacity') return false;
+      const next = retained.receipts;
+      const currentFilter =
+        currentMessage.subagentTask?.controlReceipts == null
+          ? { 'subagentTask.controlReceipts': { $exists: false } }
+          : { 'subagentTask.controlReceipts': current };
+      const updated = await Message.findOneAndUpdate(
+        { ...identity, ...currentFilter },
+        { $set: { 'subagentTask.controlReceipts': next } },
+        { new: false, projection: { messageId: 1 } },
+      ).lean<{ messageId: string } | null>();
+      if (updated != null) return true;
+    }
+    throw new Error('Subagent control receipt write contention exceeded its retry bound.');
+  }
+
+  /** Reads one bounded authoritative receipt by its exact durable task identity.
+   * The stored projection is already capped, and no task/runtime metadata leaves
+   * this method. Authorization remains part of the Mongo identity. */
+  async function getSubagentTaskControlReceipt({
+    userId,
+    conversationId,
+    taskId,
+    invocationId,
+    tenantId,
+  }: {
+    userId: string;
+    conversationId: string;
+    taskId: string;
+    invocationId: string;
+    tenantId?: string;
+  }): Promise<StoredSubagentControlReceipt | null> {
+    if (
+      userId.length === 0 ||
+      conversationId.length === 0 ||
+      conversationId.length > 256 ||
+      taskId.length === 0 ||
+      taskId.length > 256 ||
+      invocationId.length === 0 ||
+      invocationId.length > 128
+    ) {
+      throw new TypeError('Invalid subagent task control receipt identity');
+    }
+    const Message = mongoose.models.Message as Model<IMessage>;
+    const input = await Message.findOne({
+      user: userId,
+      conversationId,
+      ...(tenantId == null ? { tenantId: { $exists: false } } : { tenantId }),
+      messageId: `${taskId}:user`,
+      'subagentTask.controlReceipts.invocationId': invocationId,
+    })
+      .select({ 'subagentTask.controlReceipts': 1, _id: 0 })
+      .lean<Pick<IMessage, 'subagentTask'> | null>();
+    const receipt = input?.subagentTask?.controlReceipts?.find(
+      (candidate) => candidate.invocationId === invocationId,
+    );
+    /** Reservations are a server-private at-most-once fence, not proof that a
+     * control was applied. Public HTTP callers retry through the owning store. */
+    return receipt?.status === 'reserved' ? null : (receipt ?? null);
+  }
+
+  /** Resolves one authoritative receipt after its live owner disappears. The
+   * child conversation must still belong to the caller's parent thread, so a
+   * task id learned in another chat cannot cross orchestration scopes. */
+  async function getSubagentTaskControlReplay({
+    userId,
+    parentConversationId,
+    taskId,
+    invocationId,
+    tenantId,
+  }: {
+    userId: string;
+    parentConversationId: string;
+    taskId: string;
+    invocationId: string;
+    tenantId?: string;
+  }): Promise<{
+    receipt: StoredSubagentControlReceipt;
+    task: {
+      taskId: string;
+      threadId: string;
+      subagentType: string;
+      status: 'running' | 'completed' | 'error' | 'cancelled';
+      resultAvailable: boolean;
+      resultClaimed: boolean;
+      pendingControls: number;
+      createdAt: Date;
+      updatedAt: Date;
+    };
+  } | null> {
+    if (
+      userId.length === 0 ||
+      parentConversationId.length === 0 ||
+      parentConversationId.length > 256 ||
+      taskId.length === 0 ||
+      taskId.length > 256 ||
+      invocationId.length === 0 ||
+      invocationId.length > 128
+    ) {
+      throw new TypeError('Invalid subagent task control replay identity');
+    }
+    const Message = mongoose.models.Message as Model<IMessage>;
+    const input = await Message.findOne({
+      user: userId,
+      ...(tenantId == null ? { tenantId: { $exists: false } } : { tenantId }),
+      messageId: `${taskId}:user`,
+      'subagentTask.controlReceipts.invocationId': invocationId,
+    })
+      .select({
+        conversationId: 1,
+        createdAt: 1,
+        updatedAt: 1,
+        'subagentTask.status': 1,
+        'subagentTask.controlReceipts': 1,
+        _id: 0,
+      })
+      .lean<Pick<IMessage, 'conversationId' | 'createdAt' | 'updatedAt' | 'subagentTask'> | null>();
+    const receipt = input?.subagentTask?.controlReceipts?.find(
+      (candidate) => candidate.invocationId === invocationId,
+    );
+    const status = input?.subagentTask?.status;
+    if (
+      input == null ||
+      receipt == null ||
+      status == null ||
+      input.createdAt == null ||
+      input.updatedAt == null
+    ) {
+      return null;
+    }
+    const Conversation = mongoose.models.Conversation as Model<IConversation>;
+    const conversationQuery = Conversation.findOne({
+      user: userId,
+      conversationId: input.conversationId,
+      ...(tenantId == null ? { tenantId: { $exists: false } } : { tenantId }),
+      'subagentThread.parentConversationId': parentConversationId,
+      ...activeExpirationFilter<IConversation>(),
+    })
+      .select({ 'subagentThread.subagentType': 1, _id: 0 })
+      .lean<Pick<IConversation, 'subagentThread'> | null>();
+    const terminalQuery = Message.findOne({
+      user: userId,
+      conversationId: input.conversationId,
+      ...(tenantId == null ? { tenantId: { $exists: false } } : { tenantId }),
+      messageId: `${taskId}:assistant`,
+      'subagentTask.status': { $in: ['completed', 'error', 'cancelled'] },
+    })
+      .select({ updatedAt: 1, 'subagentTask.status': 1, 'subagentTask.resultClaim': 1, _id: 0 })
+      .lean<Pick<IMessage, 'updatedAt' | 'subagentTask'> | null>();
+    const [conversation, terminal] = await Promise.all([conversationQuery, terminalQuery]);
+    const subagentType = conversation?.subagentThread?.subagentType;
+    if (subagentType == null || subagentType === '') return null;
+    /** A committed cancel receipt is itself the authoritative cancellation
+     * boundary. The terminal row is written asynchronously and may not exist if
+     * the owner exits between those two durable commits. */
+    const replayStatus =
+      terminal?.subagentTask?.status ??
+      (receipt.action === 'cancel' && receipt.status === 'applied' ? 'cancelled' : status);
+    return {
+      receipt,
+      task: {
+        taskId,
+        threadId: input.conversationId,
+        subagentType,
+        status: replayStatus,
+        resultAvailable: terminal != null,
+        resultClaimed: terminal?.subagentTask?.resultClaim != null,
+        pendingControls:
+          input.subagentTask?.controlReceipts?.filter(
+            (candidate) => candidate.status === 'accepted',
+          ).length ?? 0,
+        createdAt: input.createdAt,
+        updatedAt:
+          terminal?.updatedAt ??
+          (receipt.action === 'cancel' && receipt.status === 'applied'
+            ? receipt.updatedAt
+            : input.updatedAt),
       },
-      [
-        {
-          $set: {
-            'subagentTask.controlReceipts': {
-              $let: {
-                vars: {
-                  current: {
-                    $cond: [
-                      { $isArray: '$subagentTask.controlReceipts' },
-                      '$subagentTask.controlReceipts',
-                      [],
-                    ],
-                  },
-                },
-                in: {
-                  $let: {
-                    vars: {
-                      existing: {
-                        $arrayElemAt: [
-                          {
-                            $filter: {
-                              input: '$$current',
-                              as: 'candidate',
-                              cond: { $eq: ['$$candidate.invocationId', receipt.invocationId] },
-                            },
-                          },
-                          0,
-                        ],
-                      },
-                    },
-                    in: {
-                      $let: {
-                        vars: {
-                          next: {
-                            $cond: [
-                              {
-                                $or: [
-                                  {
-                                    $in: [{ $ifNull: ['$$existing.status', ''] }, terminalStatuses],
-                                  },
-                                  {
-                                    $and: [
-                                      { $ne: [{ $ifNull: ['$$existing', null] }, null] },
-                                      { $ne: ['$$existing.fingerprint', receipt.fingerprint] },
-                                    ],
-                                  },
-                                ],
-                              },
-                              '$$existing',
-                              { $literal: receipt },
-                            ],
-                          },
-                        },
-                        in: {
-                          $let: {
-                            vars: {
-                              merged: {
-                                $concatArrays: [
-                                  {
-                                    $filter: {
-                                      input: '$$current',
-                                      as: 'candidate',
-                                      cond: {
-                                        $ne: ['$$candidate.invocationId', receipt.invocationId],
-                                      },
-                                    },
-                                  },
-                                  ['$$next'],
-                                ],
-                              },
-                            },
-                            in: {
-                              $let: {
-                                vars: {
-                                  accepted: {
-                                    /** The supported task store admits at most 32 live
-                                     * controls. Keep a defensive storage bound here so
-                                     * custom callers cannot grow the private projection. */
-                                    $slice: [
-                                      {
-                                        $filter: {
-                                          input: '$$merged',
-                                          as: 'candidate',
-                                          cond: { $eq: ['$$candidate.status', 'accepted'] },
-                                        },
-                                      },
-                                      -MAX_SUBAGENT_CONTROL_RECEIPTS,
-                                    ],
-                                  },
-                                },
-                                in: {
-                                  $concatArrays: [
-                                    '$$accepted',
-                                    {
-                                      $slice: [
-                                        {
-                                          /** DocumentDB 5 does not support $sortArray.
-                                           * Insert each bounded receipt into a stable
-                                           * createdAt/invocationId order using baseline
-                                           * aggregation expressions instead. */
-                                          $reduce: {
-                                            input: {
-                                              $filter: {
-                                                input: '$$merged',
-                                                as: 'candidate',
-                                                cond: {
-                                                  $ne: ['$$candidate.status', 'accepted'],
-                                                },
-                                              },
-                                            },
-                                            initialValue: [],
-                                            in: {
-                                              $concatArrays: [
-                                                {
-                                                  $filter: {
-                                                    input: '$$value',
-                                                    as: 'ordered',
-                                                    cond: {
-                                                      $or: [
-                                                        {
-                                                          $lt: [
-                                                            '$$ordered.createdAt',
-                                                            '$$this.createdAt',
-                                                          ],
-                                                        },
-                                                        {
-                                                          $and: [
-                                                            {
-                                                              $eq: [
-                                                                '$$ordered.createdAt',
-                                                                '$$this.createdAt',
-                                                              ],
-                                                            },
-                                                            {
-                                                              $lte: [
-                                                                '$$ordered.invocationId',
-                                                                '$$this.invocationId',
-                                                              ],
-                                                            },
-                                                          ],
-                                                        },
-                                                      ],
-                                                    },
-                                                  },
-                                                },
-                                                ['$$this'],
-                                                {
-                                                  $filter: {
-                                                    input: '$$value',
-                                                    as: 'ordered',
-                                                    cond: {
-                                                      $or: [
-                                                        {
-                                                          $gt: [
-                                                            '$$ordered.createdAt',
-                                                            '$$this.createdAt',
-                                                          ],
-                                                        },
-                                                        {
-                                                          $and: [
-                                                            {
-                                                              $eq: [
-                                                                '$$ordered.createdAt',
-                                                                '$$this.createdAt',
-                                                              ],
-                                                            },
-                                                            {
-                                                              $gt: [
-                                                                '$$ordered.invocationId',
-                                                                '$$this.invocationId',
-                                                              ],
-                                                            },
-                                                          ],
-                                                        },
-                                                      ],
-                                                    },
-                                                  },
-                                                },
-                                              ],
-                                            },
-                                          },
-                                        },
-                                        {
-                                          $multiply: [
-                                            -1,
-                                            {
-                                              $max: [
-                                                0,
-                                                {
-                                                  $subtract: [
-                                                    MAX_SUBAGENT_CONTROL_RECEIPTS,
-                                                    { $size: '$$accepted' },
-                                                  ],
-                                                },
-                                              ],
-                                            },
-                                          ],
-                                        },
-                                      ],
-                                    },
-                                  ],
-                                },
-                              },
-                            },
-                          },
-                        },
-                      },
-                    },
-                  },
-                },
-              },
-            },
-          },
-        },
-      ],
-      { new: true, projection: { messageId: 1 } },
-    ).lean<{ messageId: string } | null>();
-    return updated != null;
+    };
   }
 
   /** Atomically assigns one durable terminal child result to either its
@@ -1749,6 +1843,8 @@ export function createMessageMethods(mongoose: typeof import('mongoose')): Messa
     updateToolCallResult,
     updateMessage,
     recordSubagentTaskControlReceipt,
+    getSubagentTaskControlReceipt,
+    getSubagentTaskControlReplay,
     claimSubagentTaskResult,
     releaseSubagentTaskResultClaim,
     deleteMessagesSince,

packages/data-schemas/src/schema/message.ts

@@ -188,7 +188,7 @@ const messageSchema: Schema<IMessage> = new Schema(
               },
               status: {
                 type: String,
-                enum: ['accepted', 'applied', 'rejected', 'failed'],
+                enum: ['reserved', 'accepted', 'applied', 'rejected', 'failed'],
                 required: true,
               },
               createdAt: { type: Date, required: true },

packages/data-schemas/src/types/message.ts

@@ -12,7 +12,12 @@ export type SubagentTaskControlAction =
   | 'cancel'
   | 'cancel_message';
 
-export type SubagentTaskControlReceiptStatus = 'accepted' | 'applied' | 'rejected' | 'failed';
+export type SubagentTaskControlReceiptStatus =
+  | 'reserved'
+  | 'accepted'
+  | 'applied'
+  | 'rejected'
+  | 'failed';
 
 /** Server-private durable receipt for one parent-to-child control invocation. */
 export interface ISubagentTaskControlReceipt {