fix(native): a turn that dies with its process is re-run, not parked where nothing polls (TASK-264)

backend/__tests__/unit/services/agentEventService.deliveryNonce.test.js

@@ -422,7 +422,7 @@ describe('the native path carries its own delivery nonce', () => {
   test('a native run that outlives the requeue does not terminate its replacement', async () => {
     let settle;
     runAgent.mockImplementation(() => new Promise((resolve) => { settle = resolve; }));
-    jest.spyOn(AgentInstallation, 'findOne').mockReturnValue({
+    const findOne = jest.spyOn(AgentInstallation, 'findOne').mockReturnValue({
       lean: () => Promise.resolve(nativeInstall),
     });
 
@@ -439,6 +439,10 @@ describe('the native path carries its own delivery nonce', () => {
     // Minting is not enough — the settle handler has to CARRY it.
     expect(born.deliveryNonce).toEqual(expect.any(String));
 
+    // By the sweep the agent no longer runs in-process (TASK-264 re-runs the
+    // ones that still do; see the next test), so the requeue hands the event
+    // to an external poller.
+    findOne.mockReturnValue({ lean: () => Promise.resolve(null) });
     await requeue(event._id);
     const replacement = await claim();
     expect(replacement.payload.deliveryId).not.toBe(born.deliveryNonce);
@@ -454,6 +458,110 @@ describe('the native path carries its own delivery nonce', () => {
   });
 });
 
+// TASK-264: a native turn that died with its process is re-run in-process by
+// the sweep, under a fresh nonce, instead of being requeued to a queue no
+// native agent polls. The dead (or merely slow) run can still not end it.
+describe('the sweep re-runs a stuck in-process event', () => {
+  const { runAgent } = require('../../../services/nativeRuntimeService');
+  const { AgentInstallation } = require('../../../models/AgentRegistry');
+  const AgentRun = require('../../../models/AgentRun');
+  const nativeInstall = {
+    _id: new mongoose.Types.ObjectId(),
+    agentName: AGENT,
+    instanceId: INSTANCE,
+    podId: POD_ID,
+    status: 'active',
+    config: { runtime: { runtimeType: 'native' } },
+  };
+
+  afterEach(() => { jest.restoreAllMocks(); });
+
+  const enqueueStuck = async () => {
+    const settles = [];
+    runAgent.mockImplementation(() => new Promise((resolve) => { settles.push(resolve); }));
+    jest.spyOn(AgentInstallation, 'findOne').mockReturnValue({ lean: () => Promise.resolve(nativeInstall) });
+    const event = await AgentEventService.enqueue({
+      agentName: AGENT, instanceId: INSTANCE, podId: POD_ID, type: 'chat.mention', payload: { text: 'hi' },
+    });
+    const born = await AgentEvent.findById(event._id).lean();
+    return { event, born, settles };
+  };
+
+  test('re-runs it under a fresh nonce, and the dead run cannot end the new delivery', async () => {
+    const { event, born, settles } = await enqueueStuck();
+    await requeue(event._id);
+
+    const rerun = await AgentEvent.findById(event._id).lean();
+    expect(rerun.status).toBe('delivered');
+    expect(rerun.deliveryNonce).toEqual(expect.any(String));
+    expect(rerun.deliveryNonce).not.toBe(born.deliveryNonce);
+    expect(rerun.attempts).toBe(born.attempts + 1);
+    expect(runAgent).toHaveBeenCalledTimes(2);
+
+    settles[0]('done'); // the dead turn's late settle
+    await new Promise((resolve) => setImmediate(resolve));
+    await new Promise((resolve) => setImmediate(resolve));
+    expect((await AgentEvent.findById(event._id).lean()).status).toBe('delivered');
+
+    settles[1]('done'); // the re-run's settle
+    let status;
+    for (let i = 0; i < 50 && status !== 'acked'; i += 1) {
+      // eslint-disable-next-line no-await-in-loop
+      await new Promise((resolve) => setTimeout(resolve, 10));
+      // eslint-disable-next-line no-await-in-loop
+      ({ status } = await AgentEvent.findById(event._id).lean());
+    }
+    expect(status).toBe('acked');
+  });
+
+  test('two sweeps racing (two replicas) re-run it once', async () => {
+    const { event } = await enqueueStuck();
+    await AgentEvent.updateOne({ _id: event._id }, { $set: { deliveredAt: new Date(Date.now() - 60 * 60 * 1000) } });
+    const threshold = new Date(Date.now() - 10 * 60 * 1000);
+    const results = await Promise.all([
+      AgentEventService.redispatchStuckInProcess(threshold, 3),
+      AgentEventService.redispatchStuckInProcess(threshold, 3),
+    ]);
+
+    expect(results.reduce((sum, r) => sum + r.redispatched, 0)).toBe(1);
+    expect(runAgent).toHaveBeenCalledTimes(2); // the original turn plus one re-run
+  });
+
+  test('acks instead of answering twice when the run had already succeeded', async () => {
+    const { event } = await enqueueStuck();
+    jest.spyOn(AgentRun, 'exists').mockResolvedValue({ _id: 'run-1' });
+    await requeue(event._id);
+
+    expect((await AgentEvent.findById(event._id).lean()).status).toBe('acked');
+    expect(runAgent).toHaveBeenCalledTimes(1);
+  });
+
+  test('marks a run its process abandoned as interrupted, and leaves live and finished runs alone', async () => {
+    const base = { podId: new mongoose.Types.ObjectId(), agentName: AGENT, instanceId: INSTANCE, trigger: 'mention' };
+    const hourAgo = new Date(Date.now() - 60 * 60 * 1000);
+    const [abandoned, live, finished] = await AgentRun.create([
+      { ...base, status: 'running', startedAt: hourAgo },
+      { ...base, status: 'running', startedAt: new Date() },
+      { ...base, status: 'succeeded', startedAt: hourAgo },
+    ]);
+    const result = await AgentEventService.garbageCollect({ requeueDeliveredMinutes: 10 });
+
+    expect(result.interruptedRuns).toBe(1);
+    expect((await AgentRun.findById(abandoned._id).lean()).status).toBe('interrupted');
+    expect((await AgentRun.findById(live._id).lean()).status).toBe('running');
+    expect((await AgentRun.findById(finished._id).lean()).status).toBe('succeeded');
+  });
+
+  test('leaves an event at the attempt cap to the expiry pass', async () => {
+    const { event } = await enqueueStuck();
+    await AgentEvent.updateOne({ _id: event._id }, { $set: { attempts: 3 } });
+    await requeue(event._id);
+
+    expect((await AgentEvent.findById(event._id).lean()).status).toBe('failed');
+    expect(runAgent).toHaveBeenCalledTimes(1);
+  });
+});
+
 describe('recordFailure is as terminal as an ack, and gated the same way', () => {
   test('a stale runner\'s failure does not retire the replacement\'s delivery', async () => {
     const event = await seedEvent();

backend/__tests__/unit/services/agentEventService.lifecycle.test.js

@@ -81,6 +81,9 @@ const { postToPublicAddress } = require('../../../services/outboundUrlGuard');
 // Let fire-and-forget promise chains (webhook delivery, native settle) run.
 const flush = () => new Promise((resolve) => setImmediate(resolve));
 
+// A real-shaped pod id: the in-process routing lookup refuses anything else.
+const POD_P1 = '64b000000000000000000001';
+
 const lastCall = (mockFn) => mockFn.mock.calls[mockFn.mock.calls.length - 1];
 
 describe('attempts is a delivery counter with exactly one writer', () => {
@@ -100,7 +103,7 @@ describe('attempts is a delivery counter with exactly one writer', () => {
     });
     AgentEvent.findOneAndUpdate.mockReturnValue({
       lean: () => Promise.resolve({
-        _id: 'evt-1', agentName: 'pixel', instanceId: 'default', podId: 'p1',
+        _id: 'evt-1', agentName: 'pixel', instanceId: 'default', podId: POD_P1,
         type: 'chat.mention', payload: {}, status: 'delivered', attempts: 1,
       }),
     });
@@ -119,7 +122,7 @@ describe('attempts is a delivery counter with exactly one writer', () => {
 
   test('acknowledge() does NOT increment — an ack is not a delivery', async () => {
     AgentEvent.findOneAndUpdate.mockResolvedValue({
-      _id: 'evt-1', agentName: 'pixel', instanceId: 'default', podId: 'p1',
+      _id: 'evt-1', agentName: 'pixel', instanceId: 'default', podId: POD_P1,
       type: 'chat.mention', payload: {}, status: 'acked', memoryRevisionAtDelivery: 0,
     });
 
@@ -134,7 +137,7 @@ describe('attempts is a delivery counter with exactly one writer', () => {
 
   test('recordFailure() does NOT increment either', async () => {
     AgentEvent.findOneAndUpdate.mockResolvedValue({
-      _id: 'evt-1', agentName: 'pixel', instanceId: 'default', podId: 'p1',
+      _id: 'evt-1', agentName: 'pixel', instanceId: 'default', podId: POD_P1,
       type: 'chat.mention', payload: {}, status: 'failed',
     });
 
@@ -369,7 +372,7 @@ describe('driver paths reach a terminal state', () => {
 
   test('a successful webhook POST terminates as acked, not delivered', async () => {
     AgentEvent.create.mockResolvedValue({
-      _id: 'evt-w', agentName: 'hookbot', instanceId: 'default', podId: 'p1',
+      _id: 'evt-w', agentName: 'hookbot', instanceId: 'default', podId: POD_P1,
       type: 'chat.mention', payload: {}, status: 'pending', attempts: 0,
     });
     AgentInstallation.find.mockReturnValue({
@@ -381,7 +384,7 @@ describe('driver paths reach a terminal state', () => {
     postToPublicAddress.mockResolvedValueOnce({ ok: true, status: 200, text: JSON.stringify({ outcome: 'no_action' }) });
 
     await AgentEventService.enqueue({
-      agentName: 'hookbot', instanceId: 'default', podId: 'p1', type: 'chat.mention', payload: {},
+      agentName: 'hookbot', instanceId: 'default', podId: POD_P1, type: 'chat.mention', payload: {},
     });
     await flush();
 
@@ -402,7 +405,7 @@ describe('driver paths reach a terminal state', () => {
     const httpsSpy = jest.spyOn(https, 'request').mockImplementation(noNetwork);
     const warn = jest.spyOn(console, 'warn').mockImplementation(() => {});
     AgentEvent.create.mockResolvedValue({
-      _id: 'evt-r', agentName: 'hookbot', instanceId: 'default', podId: 'p1',
+      _id: 'evt-r', agentName: 'hookbot', instanceId: 'default', podId: POD_P1,
       type: 'chat.mention', payload: {}, status: 'pending', attempts: 0,
     });
     AgentInstallation.find.mockReturnValue({
@@ -413,7 +416,7 @@ describe('driver paths reach a terminal state', () => {
     });
 
     await AgentEventService.enqueue({
-      agentName: 'hookbot', instanceId: 'default', podId: 'p1', type: 'chat.mention', payload: {},
+      agentName: 'hookbot', instanceId: 'default', podId: POD_P1, type: 'chat.mention', payload: {},
     });
     await flush();
 
@@ -435,16 +438,27 @@ describe('driver paths reach a terminal state', () => {
       AgentInstallation.findOne.mockReturnValue({ lean: () => Promise.resolve(nativeInstall) });
       AgentInstallation.find.mockReturnValue({ lean: () => Promise.resolve([]) });
       AgentEvent.create.mockResolvedValue({
-        _id: 'evt-n', agentName: 'clerk', instanceId: 'default', podId: 'p1',
+        _id: 'evt-n', agentName: 'clerk', instanceId: 'default', podId: POD_P1,
         type: 'task.assigned', payload: {}, status: 'delivered', attempts: 1,
       });
       await AgentEventService.enqueue({
-        agentName: 'clerk', instanceId: 'default', podId: 'p1', type: 'task.assigned', payload: {},
+        agentName: 'clerk', instanceId: 'default', podId: POD_P1, type: 'task.assigned', payload: {},
       });
       await flush();
       await flush();
     };
 
+    test('a podId shaped like a query operator is never looked up or run in-process', async () => {
+      AgentInstallation.findOne.mockReturnValue({ lean: () => Promise.resolve(nativeInstall) });
+      AgentEvent.create.mockResolvedValue({ _id: 'evt-x', agentName: 'clerk', instanceId: 'default', type: 'chat.mention', payload: {} });
+      await AgentEventService.enqueue({
+        agentName: 'clerk', instanceId: 'default', podId: { $ne: null }, type: 'chat.mention', payload: {},
+      });
+      await flush();
+      expect(AgentInstallation.findOne).not.toHaveBeenCalled();
+      expect(runAgent).not.toHaveBeenCalled();
+    });
+
     test('creates the event already counted as one delivery', async () => {
       runAgent.mockResolvedValue({ ok: true });
 
@@ -462,7 +476,7 @@ describe('driver paths reach a terminal state', () => {
     test('a completed run acks — without this the requeue hands it to an external poller', async () => {
       runAgent.mockResolvedValue({ ok: true });
       AgentEvent.findOneAndUpdate.mockResolvedValue({
-        _id: 'evt-n', agentName: 'clerk', instanceId: 'default', podId: 'p1',
+        _id: 'evt-n', agentName: 'clerk', instanceId: 'default', podId: POD_P1,
         type: 'task.assigned', payload: {}, status: 'acked', memoryRevisionAtDelivery: 0,
       });
 
@@ -478,7 +492,7 @@ describe('driver paths reach a terminal state', () => {
       runAgent
         .mockRejectedValue(new Error('model timeout'));
       AgentEvent.findOneAndUpdate.mockResolvedValue({
-        _id: 'evt-n', agentName: 'clerk', instanceId: 'default', podId: 'p1',
+        _id: 'evt-n', agentName: 'clerk', instanceId: 'default', podId: POD_P1,
         type: 'task.assigned', payload: {}, status: 'failed',
       });