TASK-414: wake only owners for task-board changes

backend/__tests__/service/taskEventService.offeredAt.test.js

@@ -48,7 +48,7 @@ const install = (agentName = 'scout') => ({
   instanceId: 'default',
   podId: POD_ID.toString(),
   status: 'active',
-  config: { boardWake: { enabled: true } },
+  config: { runtime: { runtimeType: 'wrapper' }, boardWake: { enabled: true } },
 });
 
 const setInstalls = (installs) => {
@@ -268,7 +268,7 @@ describe('board wake assigned work', () => {
     expect(scout.payload.content).toContain('[Your assigned rows (pending/claimed; newest first):]');
     expect(scout.payload.content).toContain('TASK-ASSIGN — TASK-ASSIGN [pending] — Lily updated: reassigned to scout');
     expect(scout.payload.content).not.toContain('TASK-FOREIGN');
-    expect(peer.payload.content).not.toContain('[Your assigned rows');
+    expect(peer).toBeUndefined();
   });
 
   it('keeps the assigned list to five rows, newest first, after a folded change', async () => {
@@ -319,23 +319,150 @@ describe('board wake assigned work', () => {
     expect(anvil.payload.content).toContain('TASK-INSTANCE — Instance Do it now [pending]');
     expect(anvil.payload.content).not.toContain('Instance\nDo it now');
     expect(anvil.payload.content).not.toContain('TASK-SCOUT');
-    expect(scout.payload.content).toContain('TASK-SCOUT — TASK-SCOUT [pending]');
-    expect(scout.payload.content).not.toContain('TASK-INSTANCE —');
+    expect(scout).toBeUndefined();
+  });
+});
+
+describe('own-row board wake routing', () => {
+  const wakeNames = () => AgentEventService.enqueue.mock.calls.map(([event]) => event.agentName);
+
+  beforeEach(() => setInstalls([install('scout'), install('sprint-impl')]));
+
+  it('does not wake another seat for an unrelated row change', async () => {
+    const changed = await createTask({ taskId: 'TASK-OTHER', updatedAt: NOW });
+
+    await notifyPodAgents(POD_ID, changed, 'updated', { isAgent: false });
+
+    expect(wakeNames()).toEqual([]);
+  });
+
+  it('wakes the new assignee only', async () => {
+    const changed = await createTask({ taskId: 'TASK-ASSIGN-OWNER', updatedAt: NOW, assignee: 'scout' });
+
+    await notifyPodAgents(POD_ID, changed, 'updated', { isAgent: false }, { assignee: null });
+
+    expect(wakeNames()).toEqual(['scout']);
+  });
+
+  it('wakes the previous assignee when a row is unassigned', async () => {
+    const changed = await createTask({ taskId: 'TASK-UNASSIGN-OWNER', updatedAt: NOW });
+
+    await notifyPodAgents(POD_ID, changed, 'updated', { isAgent: false }, { assignee: 'scout' });
+
+    expect(wakeNames()).toEqual(['scout']);
+  });
+
+  it('wakes the row owner for a comment', async () => {
+    const changed = await createTask({ taskId: 'TASK-COMMENT-OWNER', updatedAt: NOW, assignee: 'scout' });
+    await Task.updateOne({ _id: changed._id }, {
+      $push: { updates: { text: 'Peer commented', author: 'Peer', createdAt: NOW } },
+    });
+    const commented = await Task.findById(changed._id).lean();
+
+    await notifyPodAgents(POD_ID, commented, 'updated', { isAgent: false });
+
+    expect(wakeNames()).toEqual(['scout']);
+    expect(AgentEventService.enqueue.mock.calls[0][0].payload.content).toContain('Peer commented');
+  });
+
+  it.each(['native', 'durable-object'])('sends a hosted %s owner a type its engine runs', async (engine) => {
+    setInstalls([
+      { ...install('maya-agent'), config: { runtime: { runtimeType: 'hosted' }, hosted: { engine } } },
+      install('scout'),
+    ]);
+    const changed = await createTask({ taskId: 'TASK-HOSTED', updatedAt: NOW, assignee: 'maya-agent' });
+
+    await notifyPodAgents(POD_ID, changed, 'updated', { isAgent: false });
+
+    expect(wakeNames()).toEqual(['maya-agent']);
+    expect(AgentEventService.enqueue.mock.calls[0][0].type).toBe('chat.mention');
+  });
+
+  it('inherits hosted runtime for a mention-created pod projection', async () => {
+    setInstalls([{ ...install('maya-agent'), config: { heartbeat: { enabled: true }, autoJoinSource: 'mention-resolution' } }]);
+    AgentInstallation.findOne.mockReturnValueOnce({ sort: () => ({ lean: async () => ({
+      ...install('maya-agent'), config: { runtime: { runtimeType: 'hosted' }, hosted: { engine: 'native' } },
+    }) }) });
+    const changed = await createTask({ taskId: 'TASK-HOSTED-PROJECTION', updatedAt: NOW, assignee: 'maya-agent' });
+
+    await notifyPodAgents(POD_ID, changed, 'updated', { isAgent: false });
+
+    expect(AgentInstallation.findOne.mock.calls[0][0]).toMatchObject({
+      agentName: 'maya-agent', instanceId: 'default', status: 'active',
+      'config.runtime.runtimeType': { $exists: true },
+    });
+    expect(AgentEventService.enqueue.mock.calls[0][0].type).toBe('chat.mention');
+  });
+
+  it('keeps an owned wake runnable if a projection runtime lookup fails', async () => {
+    setInstalls([{ ...install('maya-agent'), config: { autoJoinSource: 'mention-resolution' } }]);
+    AgentInstallation.findOne.mockImplementationOnce(() => { throw new Error('runtime lookup unavailable'); });
+    const changed = await createTask({ taskId: 'TASK-LOOKUP-FAIL', updatedAt: NOW, assignee: 'maya-agent' });
+    const warn = jest.spyOn(console, 'warn').mockImplementation(() => {});
+
+    await notifyPodAgents(POD_ID, changed, 'updated', { isAgent: false });
+    warn.mockRestore();
+
+    expect(AgentEventService.enqueue.mock.calls[0][0].type).toBe('chat.mention');
+  });
+
+  it('promotes a pending hosted sweep wake when an own-row change folds into it', async () => {
+    setInstalls([{ ...install('maya-agent'), config: { runtime: { runtimeType: 'hosted' } } }]);
+    const changed = await createTask({ taskId: 'TASK-HOSTED-FOLD', updatedAt: NOW, assignee: 'maya-agent' });
+    AgentEvent.findOneAndUpdate.mockResolvedValueOnce({ _id: 'pending-sweep', payload: { boardChanges: 1 } });
+
+    await notifyPodAgents(POD_ID, changed, 'updated', { isAgent: false });
+
+    expect(AgentEvent.findOneAndUpdate.mock.calls[0][1].$set.type).toBe('chat.mention');
+    expect(AgentEvent.updateOne).toHaveBeenCalledTimes(1);
+  });
+
+  it('wakes the waiting row owner only when its exact dependency clears', async () => {
+    const prerequisite = await createTask({ taskId: 'TASK-PREREQUISITE', updatedAt: NOW, status: 'done' });
+    const waiting = await createTask({ taskId: 'TASK-WAITING', updatedAt: NOW, assignee: 'scout' });
+    await Task.updateOne({ _id: waiting._id }, { $set: { dep: 'TASK-PREREQUISITE' } });
+    const unrelated = await createTask({ taskId: 'TASK-UNRELATED', updatedAt: NOW, assignee: 'sprint-impl' });
+    await Task.updateOne({ _id: unrelated._id }, { $set: { dep: 'TASK-OTHER' } });
+
+    await notifyPodAgents(POD_ID, prerequisite, 'updated', { isAgent: false }, { status: 'claimed' });
+
+    expect(wakeNames()).toEqual(['scout']);
+  });
+
+  it('does not wake an agent for its own write', async () => {
+    const changed = await createTask({ taskId: 'TASK-SELF', updatedAt: NOW, assignee: 'scout' });
+
+    await notifyPodAgents(POD_ID, changed, 'updated', {
+      isAgent: true, agentName: 'scout', instanceId: 'default',
+    });
+
+    expect(wakeNames()).toEqual([]);
+  });
+
+  it('still folds a second own-row change into the pending board wake', async () => {
+    const changed = await createTask({ taskId: 'TASK-FOLD-OWNER', updatedAt: NOW, assignee: 'scout' });
+    AgentEvent.findOneAndUpdate.mockResolvedValueOnce({ _id: 'pending-wake', payload: { boardChanges: 1 } });
+
+    await notifyPodAgents(POD_ID, changed, 'updated', { isAgent: false });
+
+    expect(wakeNames()).toEqual([]);
+    expect(AgentEvent.updateOne).toHaveBeenCalledTimes(1);
+    expect(AgentEvent.updateOne.mock.calls[0][1].$set['payload.content']).toContain('2 changes');
   });
 });
 
 describe('offeredAt race probe', () => {
   it('keeps write two due with a CAS stamp of seen, while a now stamp hides it', async () => {
     const seen = new Date(NOW.getTime() - 5000);
     const writeTwo = new Date(NOW.getTime() - 4000);
-    const casTask = await createTask({ taskId: 'TASK-CAS', updatedAt: seen });
+    const casTask = await createTask({ taskId: 'TASK-CAS', updatedAt: seen, assignee: 'scout' });
     const nowTask = await createTask({ taskId: 'TASK-NOW', updatedAt: seen });
     setInstalls([install()]);
 
     AgentEventService.enqueue.mockImplementationOnce(async () => {
       await Task.updateOne(
         { _id: casTask._id },
-        { $set: { updatedAt: writeTwo } },
+        { $set: { updatedAt: writeTwo, assignee: null } },
         { timestamps: false },
       );
     });

backend/__tests__/service/tasks.claim-lease.test.js

@@ -30,6 +30,7 @@ const Pod = require('../../models/Pod');
 const Task = require('../../models/Task');
 const User = require('../../models/User');
 const { AgentInstallation } = require('../../models/AgentRegistry');
+const KernelWorkSweepService = require('../../services/kernelWorkSweepService').default;
 const tasksApiRoutes = require('../../routes/tasksApi');
 const { hash } = require('../../utils/secret');
 const {
@@ -338,4 +339,201 @@ describe('POST /api/v1/tasks/:podId/:taskId/claim — lease semantics (ADR-018 D
       expect(notifyPodAgents).not.toHaveBeenCalled();
     });
   });
+
+  test('PATCH wakes the owner immediately before its atomic write under interleaving', async () => {
+    await seedTask({ assignee: 'alpha' });
+    notifyPodAgents.mockClear();
+    const originalUpdate = Task.findOneAndUpdate.bind(Task);
+    let releaseGamma;
+    let gammaEntered;
+    const gammaPaused = new Promise((resolve) => { releaseGamma = resolve; });
+    const entered = new Promise((resolve) => { gammaEntered = resolve; });
+    let pausedGamma = false;
+    const updateSpy = jest.spyOn(Task, 'findOneAndUpdate').mockImplementation((filter, update, options) => {
+      if (update?.$set?.assignee === 'gamma' && !pausedGamma) {
+        pausedGamma = true;
+        gammaEntered();
+        return gammaPaused.then(() => originalUpdate(filter, update, options));
+      }
+      return originalUpdate(filter, update, options);
+    });
+    const patch = (assignee) => request(app)
+      .patch(`/api/v1/tasks/${pod._id}/T-1`)
+      .set('Authorization', `Bearer ${ownerToken}`)
+      .send({ assignee });
+
+    try {
+      const gammaRequest = patch('gamma').then((response) => response);
+      await entered;
+      await new Promise((resolve) => setTimeout(resolve, 5));
+      const betaResponse = await patch('beta');
+      expect(betaResponse.status).toBe(200);
+      releaseGamma();
+      const gammaResponse = await gammaRequest;
+      expect(gammaResponse.status).toBe(200);
+
+      const gammaWake = notifyPodAgents.mock.calls.find((call) => call[1].assignee === 'gamma');
+      expect(gammaWake?.[4]).toEqual(expect.objectContaining({ assignee: 'beta' }));
+      expect(gammaWake?.[1]).toEqual(expect.objectContaining({ assignee: 'gamma' }));
+      const betaWake = notifyPodAgents.mock.calls.find((call) => call[1].assignee === 'beta');
+      expect(gammaWake[1].updatedAt.getTime()).toBeGreaterThan(betaWake[1].updatedAt.getTime());
+      expect((await Task.findOne({ podId: pod._id, taskId: 'T-1' })).assignee).toBe('gamma');
+    } finally {
+      releaseGamma();
+      updateSpy.mockRestore();
+    }
+  });
+
+  test('claim wakes the assignee at its successful write after a racing PATCH', async () => {
+    await seedTask({ assignee: 'alpha' });
+    notifyPodAgents.mockClear();
+    const originalUpdate = Task.findOneAndUpdate.bind(Task);
+    let releaseClaim;
+    let claimEntered;
+    const claimPaused = new Promise((resolve) => { releaseClaim = resolve; });
+    const entered = new Promise((resolve) => { claimEntered = resolve; });
+    let pausedClaim = false;
+    const updateSpy = jest.spyOn(Task, 'findOneAndUpdate').mockImplementation((filter, update, options) => {
+      if (update?.$set?.claimedBy === owner._id.toString() && !pausedClaim) {
+        pausedClaim = true;
+        claimEntered();
+        return claimPaused.then(() => originalUpdate(filter, update, options));
+      }
+      return originalUpdate(filter, update, options);
+    });
+
+    try {
+      const claimRequest = claim(ownerToken).then((response) => response);
+      await entered;
+      const patchResponse = await request(app)
+        .patch(`/api/v1/tasks/${pod._id}/T-1`)
+        .set('Authorization', `Bearer ${ownerToken}`)
+        .send({ assignee: 'beta' });
+      expect(patchResponse.status).toBe(200);
+      notifyPodAgents.mockClear();
+      releaseClaim();
+      const claimResponse = await claimRequest;
+      expect(claimResponse.status).toBe(200);
+
+      const claimWake = notifyPodAgents.mock.calls.find((call) => call[1].claimedBy === owner._id.toString());
+      expect(claimWake?.[4]).toEqual(expect.objectContaining({ assignee: 'beta' }));
+      expect(claimWake?.[1]).toEqual(expect.objectContaining({ assignee: 'beta' }));
+    } finally {
+      releaseClaim();
+      updateSpy.mockRestore();
+    }
+  });
+
+  test('claim uses the rescued owner when a sweep writes the same timestamp', async () => {
+    const task = await seedTask({
+      status: 'claimed',
+      assignee: 'alpha',
+      claimedBy: 'alpha',
+      claimedAt: new Date(Date.now() - LEASE_MS - 5000),
+      claimExpiresAt: new Date(Date.now() - 5000),
+    });
+    notifyPodAgents.mockClear();
+    const originalUpdate = Task.findOneAndUpdate.bind(Task);
+    let releaseClaim;
+    let claimEntered;
+    const claimPaused = new Promise((resolve) => { releaseClaim = resolve; });
+    const entered = new Promise((resolve) => { claimEntered = resolve; });
+    let pausedClaim = false;
+    const updateSpy = jest.spyOn(Task, 'findOneAndUpdate').mockImplementation((filter, update, options) => {
+      if (update?.$set?.claimedBy === owner._id.toString() && !pausedClaim) {
+        pausedClaim = true;
+        claimEntered();
+        return claimPaused.then(() => originalUpdate(filter, update, options));
+      }
+      return originalUpdate(filter, update, options);
+    });
+
+    try {
+      const claimRequest = claim(ownerToken).then((response) => response);
+      await entered;
+      // Mongoose normally stamps the sweep with wall-clock time. Hold its
+      // clock at the existing revision to reproduce a timestamp collision.
+      const mongoose = require('mongoose');
+      const clockSpy = jest.spyOn(mongoose, 'now').mockReturnValue(task.updatedAt);
+      try {
+        const { rescued } = await KernelWorkSweepService.rescueLapsed(new Date());
+        expect(rescued).toHaveLength(1);
+      } finally {
+        clockSpy.mockRestore();
+      }
+      const rescuedTask = await Task.findById(task._id);
+      expect(rescuedTask.updatedAt.getTime()).toBe(task.updatedAt.getTime());
+      expect(rescuedTask.assignee).toBeNull();
+      notifyPodAgents.mockClear();
+      releaseClaim();
+      const claimResponse = await claimRequest;
+      expect(claimResponse.status).toBe(200);
+
+      const claimWake = notifyPodAgents.mock.calls.find((call) => call[1].claimedBy === owner._id.toString());
+      expect(claimWake?.[4]).toEqual(expect.objectContaining({ assignee: null, claimedBy: null }));
+      expect(claimWake?.[4].assignee).not.toBe('alpha');
+    } finally {
+      releaseClaim();
+      updateSpy.mockRestore();
+    }
+  });
+
+  test('PATCH unassignment offers the exact stored revision before the work sweep', async () => {
+    await seedTask({ assignee: 'claim-audit-agent' });
+    // Force the same monotonic retry the PATCH takes when a competing writer
+    // has committed after this request chose its timestamp.
+    await Task.updateOne(
+      { podId: pod._id, taskId: 'T-1' },
+      { $set: { updatedAt: new Date(Date.now() + 1000) } },
+      { timestamps: false },
+    );
+    notifyPodAgents.mockClear();
+    const res = await request(app)
+      .patch(`/api/v1/tasks/${pod._id}/T-1`)
+      .set('Authorization', `Bearer ${ownerToken}`)
+      .send({ assignee: null });
+    expect(res.status).toBe(200);
+    const wake = notifyPodAgents.mock.calls.find((call) => call[1].taskId === 'T-1');
+    expect(wake?.[4]).toEqual(expect.objectContaining({ assignee: 'claim-audit-agent' }));
+    const stored = await Task.findOne({ podId: pod._id, taskId: 'T-1' });
+    expect(wake[1].updatedAt.getTime()).toBe(stored.updatedAt.getTime());
+
+    const realNotifyPodAgents = jest.requireActual('../../services/taskEventService').notifyPodAgents;
+    await realNotifyPodAgents(pod._id, wake[1], wake[2], wake[3], wake[4]);
+    const offered = await Task.findById(stored._id);
+    expect(offered.offeredAt?.getTime()).toBe(offered.updatedAt.getTime());
+
+    const sweep = await require('../../services/kernelWorkSweepService').wakeForFoundWork(new Date());
+    expect(sweep.scannedPods).toBe(0);
+
+    const note = await request(app)
+      .post(`/api/v1/tasks/${pod._id}/T-1/updates`)
+      .set('Authorization', `Bearer ${ownerToken}`)
+      .send({ text: 'Still available for pickup' });
+    expect(note.status).toBe(200);
+    const afterNote = await Task.findById(stored._id);
+    expect(afterNote.updatedAt.getTime()).toBeGreaterThan(offered.offeredAt.getTime());
+    const dueSweep = await require('../../services/kernelWorkSweepService').wakeForFoundWork(new Date());
+    expect(dueSweep.scannedPods).toBe(1);
+
+    const title = await request(app)
+      .patch(`/api/v1/tasks/${pod._id}/T-1`)
+      .set('Authorization', `Bearer ${ownerToken}`)
+      .send({ title: 'Still open' });
+    expect(title.status).toBe(200);
+    const afterTitle = await Task.findById(stored._id);
+    expect(afterTitle.updatedAt.getTime()).toBeGreaterThan(afterNote.updatedAt.getTime());
+
+    const claimed = await claim(ownerToken);
+    expect(claimed.status).toBe(200);
+    expect(new Date(claimed.body.task.updatedAt).getTime()).toBeGreaterThan(afterTitle.updatedAt.getTime());
+
+    const completed = await request(app)
+      .post(`/api/v1/tasks/${pod._id}/T-1/complete`)
+      .set('Authorization', `Bearer ${ownerToken}`)
+      .send({});
+    expect(completed.status).toBe(200);
+    expect(new Date(completed.body.task.updatedAt).getTime())
+      .toBeGreaterThan(new Date(claimed.body.task.updatedAt).getTime());
+  });
 });