fix: checkpoint handled inbox events before turn completion (TASK-481)

backend/__tests__/unit/routes/agentsRuntime.dm-events.test.js

@@ -11,6 +11,7 @@ jest.mock('jsonwebtoken', () => ({ sign: jest.fn(), verify: jest.fn(), decode: j
 jest.mock('../../../services/agentEventService', () => ({
   list: jest.fn(),
   listInboxPage: jest.fn(),
+  renewDelivery: jest.fn(),
 }));
 jest.mock('../../../services/agentIdentityService', () => ({
   buildAgentUsername: jest.fn((agentName, instanceId = 'default') => (
@@ -138,3 +139,27 @@ describe('agentsRuntime DM and event pod coverage', () => {
     expect(res.json).toHaveBeenCalledWith({ dmPod: { _id: 'dm-pod-1', type: 'agent-admin' } });
   });
 });
+
+
+describe('inbox delivery renewal route', () => {
+  const eventId = '0123456789abcdef01234567';
+  const response = () => ({ status: jest.fn().mockReturnThis(), json: jest.fn() });
+  test('uses authenticated seat/instance, ignoring identity supplied in the body', async () => {
+    AgentEventService.renewDelivery.mockResolvedValue(true);
+    const res = response();
+    await getRouteHandler('/events/renew', 'post')({
+      agentInstallation: { agentName: 'wren', instanceId: 'writer-instance' },
+      body: { agentName: 'victim', instanceId: 'victim-instance', events: [{ eventId, deliveryId: 'nonce' }] },
+    }, res);
+    expect(AgentEventService.renewDelivery).toHaveBeenLastCalledWith(eventId, 'wren', 'writer-instance', 'nonce');
+    expect(res.json).toHaveBeenCalledWith({ renewed: [eventId] });
+  });
+  test.each([undefined, [], Array(51).fill({ eventId, deliveryId: 'nonce' }),
+    [{ eventId, deliveryId: '' }], [{ eventId: {}, deliveryId: 'nonce' }]])('rejects malformed or unbounded input %j', async (events) => {
+    AgentEventService.renewDelivery.mockClear();
+    const res = response();
+    await getRouteHandler('/events/renew', 'post')({ body: { events }, agentInstallation: { agentName: 'wren' } }, res);
+    expect(res.status).toHaveBeenCalledWith(400);
+    expect(AgentEventService.renewDelivery).not.toHaveBeenCalled();
+  });
+});

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

@@ -0,0 +1,63 @@
+const mongoose = require('mongoose');
+const { setupMongoDb, closeMongoDb, clearMongoDb } = require('../../utils/testUtils');
+const AgentEvent = require('../../../models/AgentEvent');
+const AgentEventService = require('../../../services/agentEventService');
+
+describe('live inbox delivery renewal', () => {
+  beforeAll(setupMongoDb);
+  afterAll(closeMongoDb);
+  afterEach(clearMongoDb);
+
+  const delivery = async (fields = {}) => {
+    const row = { _id: new mongoose.Types.ObjectId(), agentName: 'wren', instanceId: 'default',
+      podId: new mongoose.Types.ObjectId(), type: 'chat.mention', status: 'delivered', attempts: 1,
+      deliveryNonce: 'current-nonce', deliveredAt: new Date('2026-01-01'), createdAt: new Date(), ...fields };
+    await AgentEvent.collection.insertOne(row);
+    return row;
+  };
+
+  test('extends only current delivery time, without acking or spending another attempt', async () => {
+    const row = await delivery();
+    expect(await AgentEventService.renewDelivery(String(row._id), 'Wren', 'default', 'current-nonce')).toBe(true);
+    const after = await AgentEvent.findById(row._id).lean();
+    expect(after.status).toBe('delivered');
+    expect(after.deliveryNonce).toBe('current-nonce');
+    expect(after.attempts).toBe(1);
+    expect(after.deliveredAt.getTime()).toBeGreaterThan(row.deliveredAt.getTime());
+  });
+
+  test('GC requeues only two unfinished events after three durable checkpoints', async () => {
+    const rows = [];
+    for (let i = 0; i < 5; i += 1) rows.push(await delivery());
+    for (const row of rows.slice(0, 3)) {
+      expect(await AgentEventService.acknowledge(String(row._id), 'wren', 'default',
+        { outcome: 'acknowledged', reason: 'handled-checkpoint' }, 'current-nonce', 'mcp')).not.toBeNull();
+    }
+    await AgentEventService.garbageCollect();
+    const pending = await AgentEvent.find({ status: 'pending' }).lean();
+    expect(pending.map((row) => String(row._id)).sort()).toEqual(rows.slice(3).map((row) => String(row._id)).sort());
+    expect(await AgentEvent.countDocuments({ status: 'acked' })).toBe(3);
+  });
+
+  test('a renewed live delivery survives the actual stale-delivery GC', async () => {
+    const row = await delivery();
+    await AgentEventService.renewDelivery(String(row._id), 'wren', 'default', 'current-nonce');
+    await AgentEventService.garbageCollect();
+    expect((await AgentEvent.findById(row._id).lean()).status).toBe('delivered');
+  });
+
+  test.each([
+    ['other-seat', 'default', 'current-nonce'], ['wren', 'another-instance', 'current-nonce'],
+    ['wren', 'default', 'old-nonce'], ['wren', 'default', ''],
+  ])('refuses wrong identity or nonce (%s/%s/%s)', async (agent, instance, nonce) => {
+    const row = await delivery();
+    expect(await AgentEventService.renewDelivery(String(row._id), agent, instance, nonce)).toBe(false);
+    expect((await AgentEvent.findById(row._id).lean()).deliveredAt).toEqual(row.deliveredAt);
+  });
+
+  test.each(['pending', 'acked', 'failed'])('does not revive a %s event', async (status) => {
+    const row = await delivery({ status });
+    expect(await AgentEventService.renewDelivery(String(row._id), 'wren', 'default', 'current-nonce')).toBe(false);
+    expect((await AgentEvent.findById(row._id).lean()).status).toBe(status);
+  });
+});

backend/routes/agentsRuntime.ts

@@ -1313,6 +1313,32 @@ router.post('/bot/events/:id/ack', auth, requireApiTokenScopes(['agent:events:ac
   }
 });
 
+// A live inbox turn may outlast the stale-delivery GC interval. Renew only
+// the token owner's current nonce; this never marks work handled.
+router.post('/events/renew', phase4RateLimit, agentRuntimeAuth, async (req: any, res: any) => {
+  const events = req.body?.events;
+  if (!Array.isArray(events) || events.length < 1 || events.length > 50
+    || events.some((entry: any) => !entry || typeof entry.eventId !== 'string' || !/^[a-fA-F0-9]{24}$/.test(entry.eventId)
+      || typeof entry.deliveryId !== 'string' || !entry.deliveryId)) {
+    return res.status(400).json({ message: 'events requires 1–50 eventId/deliveryId pairs' });
+  }
+  const agentName = req.agentInstallation?.agentName
+    || req.agentUser?.botMetadata?.agentName
+    || req.agentUser?.botMetadata?.agentType || req.agentUser?.username;
+  const instanceId = req.agentInstallation?.instanceId || req.agentUser?.botMetadata?.instanceId || 'default';
+  if (!agentName) return res.status(403).json({ message: 'Agent token not authorized for events' });
+  try {
+    const renewed = await Promise.all(events.map(async (entry: { eventId: string; deliveryId: string }) => (
+      await AgentEventService.renewDelivery(entry.eventId, agentName, instanceId, entry.deliveryId)
+        ? entry.eventId : null
+    )));
+    return res.json({ renewed: renewed.filter(Boolean) });
+  } catch (error: any) {
+    console.error('Error renewing inbox deliveries:', error.message);
+    return res.status(500).json({ message: 'Could not renew inbox deliveries' });
+  }
+});
+
 router.post('/events/:id/ack', phase4RateLimit, agentRuntimeAuth, async (req: any, res: any) => {
   try {
     const installation = req.agentInstallation;

backend/services/agentEventService.ts

@@ -1961,6 +1961,23 @@ class AgentEventService {
     return { events, inboxCount: includeInboxCount ? inboxCount : events.length };
   }
 
+  // Keep a live wrapper's delivery from aging into the GC requeue while its
+  // child is handling a batch. A nonce is mandatory; never renew a replacement
+  // runner's delivery, a pending row or a terminal checkpoint.
+  static async renewDelivery(
+    eventId: string, agentName: string, instanceId: string, deliveryId: string,
+  ): Promise<boolean> {
+    if (!eventId || !deliveryId) return false;
+    const result = await AgentEvent.updateOne({
+      _id: eventId,
+      agentName: String(agentName).toLowerCase(),
+      instanceId: String(instanceId || 'default'),
+      status: 'delivered',
+      deliveryNonce: deliveryId,
+    }, { $set: { deliveredAt: new Date() } });
+    return result.modifiedCount === 1;
+  }
+
   // ADR-012 §3: acknowledge is status-gated. Lifecycle:
   //   pending → delivered (on `list()` claim)
   //   delivered → acked (here)

cli/__tests__/run-loop.test.mjs

@@ -1546,6 +1546,162 @@ describe('performRun', () => {
     expect(errors[0]).toMatchObject({ failureClass: 'quota', circuitOpen: true });
   });
 
+  test('a never-resolving child cannot renew inbox deliveries beyond the turn deadline', async () => {
+    const post = jest.fn().mockResolvedValue({});
+    createClient.mockReturnValue({ get: jest.fn(async (route) => {
+      if (route.endsWith('/events')) return { events: [makeEvent({ payload: { content: 'hang', deliveryId: 'nonce' } })] };
+      if (route.endsWith('/messages')) return { messages: [] };
+      return {};
+    }), post });
+    const intervals = [];
+    const deadlines = [];
+    const clearIntervalImpl = jest.fn();
+    const spawn = jest.fn(() => new Promise(() => {}));
+    const handle = performRun({ instanceUrl: 'https://example.test', token: 'cm_agent_test',
+      adapter: { name: 'stub', detect: stubAdapter.detect, spawn }, agentName: 'my-stub',
+      turnTimeoutMs: 1000, setTimeoutImpl: noopTimeout,
+      setIntervalImpl: (callback, delay) => { const timer = { callback, delay }; intervals.push(timer); return timer; },
+      clearIntervalImpl,
+      setDeadlineTimeoutImpl: (callback, delay) => { const timer = { callback, delay }; deadlines.push(timer); return timer; },
+      clearDeadlineTimeoutImpl: jest.fn() });
+    await drainMicrotasks();
+    try {
+      expect(spawn).toHaveBeenCalledTimes(1);
+      expect(spawn.mock.calls[0][1].timeoutMs).toBe(1000);
+      const renewal = intervals.find((timer) => timer.delay === 60_000);
+      await renewal.callback();
+      expect(callsTo(post, '/events/renew')).toHaveLength(1);
+      expect(deadlines).toHaveLength(1);
+      expect(deadlines[0].delay).toBe(1000);
+      deadlines[0].callback(); // child deliberately remains unresolved
+      expect(clearIntervalImpl).toHaveBeenCalledWith(renewal);
+      await renewal.callback(); // even an already queued interval cannot renew
+      expect(callsTo(post, '/events/renew')).toHaveLength(1);
+      expect(ackPosts(post)).toHaveLength(0);
+    } finally { handle.stop(); }
+  });
+
+  test.each([1, 51])('renews %i nonce-bound deliveries in bounded requests and clears the timer afterwards', async (count) => {
+    const events = Array.from({ length: count }, (_, i) => makeEvent({
+      _id: `evt-${i + 1}`, payload: { content: 'long work', deliveryId: `nonce-${i + 1}` },
+    }));
+    const post = jest.fn().mockResolvedValue({});
+    createClient.mockReturnValue({ get: jest.fn(async (route) => {
+      if (route.endsWith('/events')) return { events };
+      if (route.endsWith('/messages')) return { messages: [] };
+      return {};
+    }), post });
+    const timers = [];
+    const clearIntervalImpl = jest.fn();
+    let finish;
+    const spawn = jest.fn(() => new Promise((resolve) => { finish = resolve; }));
+    const handle = performRun({ instanceUrl: 'https://example.test', token: 'cm_agent_test',
+      adapter: { name: 'stub', detect: stubAdapter.detect, spawn }, agentName: 'my-stub',
+      setTimeoutImpl: noopTimeout,
+      setIntervalImpl: (callback, delay) => { const timer = { callback, delay }; timers.push(timer); return timer; },
+      clearIntervalImpl });
+    await drainMicrotasks();
+    const renewal = timers.find((timer) => timer.delay === 60_000);
+    expect(renewal).toBeDefined();
+    await renewal.callback();
+    const renewals = callsTo(post, '/events/renew');
+    expect(renewals).toHaveLength(Math.ceil(count / 50));
+    expect(renewals.every(([, body]) => body.events.length <= 50)).toBe(true);
+    expect(renewals.flatMap(([, body]) => body.events)).toEqual(events.map((event) => ({
+      eventId: event._id, deliveryId: event.payload.deliveryId,
+    })));
+    expect(ackPosts(post)).toHaveLength(0);
+    finish({ text: 'NO_REPLY' });
+    await drainMicrotasks();
+    handle.stop();
+    expect(clearIntervalImpl).toHaveBeenCalledWith(renewal);
+    expect(ackPosts(post)).toHaveLength(count);
+  });
+
+  test('an item without a delivery nonce gets no per-item checkpoint instruction', async () => {
+    createClient.mockReturnValue({ get: jest.fn(async (route) => {
+      if (route.endsWith('/events')) return { events: [makeEvent()] };
+      if (route.endsWith('/messages')) return { messages: [] };
+      return {};
+    }), post: jest.fn().mockResolvedValue({}) });
+    const spawn = jest.fn(async () => ({ text: 'NO_REPLY' }));
+    const handle = performRun({ instanceUrl: 'https://example.test', token: 'cm_agent_test',
+      adapter: { name: 'stub', detect: stubAdapter.detect, spawn }, agentName: 'my-stub', setTimeoutImpl: noopTimeout });
+    await drainMicrotasks();
+    handle.stop();
+    expect(spawn).toHaveBeenCalledTimes(1);
+    expect(spawn.mock.calls[0][0]).not.toContain('call commonly_checkpoint_event(');
+  });
+
+  test('three explicit event checkpoints survive a five-event turn timeout', async () => {
+    const { buildTools } = await import('../../commonly-mcp/src/tools.js');
+    const originalFetch = global.fetch;
+    const events = Array.from({ length: 5 }, (_, index) => makeEvent({
+      _id: `evt-check-${index + 1}`,
+      payload: { content: `handle item ${index + 1}`, deliveryId: `nonce-${index + 1}` },
+    }));
+    const acked = new Set();
+    const pages = [];
+    const post = jest.fn(async (route) => {
+      const match = route.match(/events\/(evt-check-\d)\/ack$/);
+      if (match) acked.add(match[1]);
+      return {};
+    });
+    const get = jest.fn(async (route) => {
+      if (route === '/api/agents/runtime/events') {
+        const remaining = events.filter((event) => !acked.has(event._id));
+        pages.push(remaining.map((event) => event._id));
+        return { events: remaining };
+      }
+      if (route.endsWith('/messages')) return { messages: [] };
+      if (route.endsWith('/memory')) return { sections: {} };
+      return {};
+    });
+    createClient.mockReturnValue({ get, post });
+    global.fetch = jest.fn(async (url, init) => {
+      const id = String(url).match(/events\/(evt-check-\d)\/ack$/)?.[1];
+      expect(id).toBeDefined();
+      const event = events.find((entry) => entry._id === id);
+      expect(JSON.parse(init.body)).toEqual({ deliveryId: event.payload.deliveryId,
+        result: { outcome: 'acknowledged', reason: 'handled-checkpoint' } });
+      acked.add(id);
+      return { ok: true, status: 200, text: async () => '{"success":true}' };
+    });
+    const scheduled = [];
+    let attempt = 0;
+    const spawn = jest.fn(async (prompt, ctx) => {
+      attempt += 1;
+      if (attempt === 1) {
+        const checkpoint = buildTools({ baseUrl: ctx.instanceUrl, token: ctx.runtimeToken })
+          .find((tool) => tool.name === 'commonly_checkpoint_event');
+        expect(checkpoint).toBeDefined();
+        for (const event of events.slice(0, 3)) {
+          expect(prompt).toContain(`After completed tool-side work or intentional NO_REPLY for only this item, call commonly_checkpoint_event(${JSON.stringify({ eventId: event._id, deliveryId: event.payload.deliveryId })})`);
+          expect(prompt).toContain('Do not checkpoint a reply you are holding for final CLI output');
+          const result = await checkpoint.call({ eventId: event._id, deliveryId: event.payload.deliveryId });
+          expect(result.isError).not.toBe(true);
+        }
+        throw Object.assign(new Error('turn timed out'), { code: 'ETIMEDOUT' });
+      }
+      expect(prompt).not.toContain('evt-check-1');
+      expect(prompt).toContain('evt-check-4');
+      expect(prompt).toContain('evt-check-5');
+      return { text: 'NO_REPLY' };
+    });
+    const handle = performRun({ instanceUrl: 'https://example.test', token: 'cm_agent_test',
+      adapter: { name: 'stub', detect: stubAdapter.detect, spawn }, agentName: 'my-stub',
+      retryJitterRatio: 0, setTimeoutImpl: (callback, delayMs) => { scheduled.push({ callback, delayMs }); return 0; } });
+    try {
+      await drainMicrotasks();
+      expect(acked).toEqual(new Set(events.slice(0, 3).map((event) => event._id)));
+      scheduled.shift().callback();
+      await drainMicrotasks();
+      expect(pages.slice(0, 2)).toEqual([events.map((event) => event._id), ['evt-check-4', 'evt-check-5']]);
+      expect(spawn).toHaveBeenCalledTimes(2);
+      expect(acked.size).toBe(5);
+    } finally { handle.stop(); global.fetch = originalFetch; }
+  });
+
   test('a batch spawn failure leaves its no-op sibling unacked for coherent retry', async () => {
     let poll = 0;
     const mockGet = jest.fn(async (route) => {