diff --git a/backend/__tests__/service/taskEventService.nudgeThrottle.test.js b/backend/__tests__/service/taskEventService.nudgeThrottle.test.js new file mode 100644 index 0000000000..b8c5a08049 --- /dev/null +++ b/backend/__tests__/service/taskEventService.nudgeThrottle.test.js @@ -0,0 +1,114 @@ +/* eslint-disable import/no-unresolved, import/extensions */ +const mongoose = require('mongoose'); +const { setupMongoDb, closeMongoDb, clearMongoDb } = require('../utils/testUtils'); + +jest.mock('../../services/agentEventService', () => ({ enqueue: jest.fn().mockResolvedValue({}) })); +jest.mock('../../models/AgentEvent', () => ({ findOneAndUpdate: jest.fn().mockResolvedValue(null) })); + +const { AgentInstallation } = require('../../models/AgentRegistry'); +const Task = require('../../models/Task'); +const User = require('../../models/User'); +const AgentEventService = require('../../services/agentEventService'); +const { notifyFoundWork } = require('../../services/taskEventService'); + +const podId = new mongoose.Types.ObjectId(); +const now = new Date('2026-08-20T20:00:00.000Z'); +const work = [{ + _id: new mongoose.Types.ObjectId(), taskId: 'TASK-UNCLAIMED', title: 'Unclaimed work', updatedAt: now, +}]; + +beforeAll(setupMongoDb); +beforeEach(async () => { + await clearMongoDb(); + jest.clearAllMocks(); + await AgentInstallation.create({ + agentName: 'scout', + instanceId: 'default', + podId, + status: 'active', + version: '1.0.0', + installedBy: new mongoose.Types.ObjectId(), + config: { boardWake: { enabled: true }, workDiscovery: { enabled: true, role: 'implementer' } }, + }); +}); +afterAll(async () => { + await clearMongoDb(); + await closeMongoDb(); +}); + +it('atomically permits one nudge per seat and pod in a rolling one-hour window', async () => { + const [a, b] = await Promise.all([ + notifyFoundWork(podId, work, 1, now), + notifyFoundWork(podId, work, 1, now), + ]); + expect(a.woken + b.woken).toBe(1); + expect(AgentEventService.enqueue).toHaveBeenCalledTimes(1); + + const withinWindow = await notifyFoundWork(podId, work, 1, new Date(now.getTime() + 59 * 60 * 1000)); + expect(withinWindow.woken).toBe(0); + expect(AgentEventService.enqueue).toHaveBeenCalledTimes(1); + + const nextWindow = await notifyFoundWork(podId, work, 1, new Date(now.getTime() + 60 * 60 * 1000)); + expect(nextWindow.woken).toBe(1); + expect(AgentEventService.enqueue).toHaveBeenCalledTimes(2); + const install = await AgentInstallation.findOne({ podId, agentName: 'scout' }).lean(); + expect(install.lastFoundWorkAt).toEqual(new Date(now.getTime() + 60 * 60 * 1000)); +}); + +it('still nudges an idle uppercase DEFAULT installation using its stored identity', async () => { + await AgentInstallation.updateOne({ podId, agentName: 'scout' }, { $set: { instanceId: 'DEFAULT' } }); + + expect((await notifyFoundWork(podId, work, 1, now)).woken).toBe(1); + + expect(AgentEventService.enqueue).toHaveBeenCalledWith(expect.objectContaining({ + agentName: 'scout', instanceId: 'DEFAULT', + })); + const install = await AgentInstallation.findOne({ podId, agentName: 'scout' }).lean(); + expect(install.lastFoundWorkAt).toEqual(now); +}); + +it.each([ + ['assigned', { assignee: 'scout', status: 'pending' }, 'default'], + ['mixed-case assignment', { assignee: 'Scout', status: 'pending' }, 'default'], + ['claimed', { claimedBy: 'scout', status: 'claimed' }, 'default'], + ['mixed-case claim', { claimedBy: 'Scout', status: 'claimed' }, 'default'], + ['assignment with uppercase DEFAULT', { assignee: 'scout', status: 'pending' }, 'DEFAULT'], + ['claim with uppercase DEFAULT', { claimedBy: 'Scout', status: 'claimed' }, 'DEFAULT'], +])('does not reserve a slot for a reviewer or an implementer with %s', async (_kind, ownership, instanceId) => { + if (instanceId) await AgentInstallation.updateOne({ podId, agentName: 'scout' }, { $set: { instanceId } }); + await AgentInstallation.create({ + agentName: 'sprint-review', instanceId: 'default', podId, status: 'active', + version: '1.0.0', installedBy: new mongoose.Types.ObjectId(), + config: { boardWake: { enabled: true }, workDiscovery: { enabled: true, role: 'reviewer' } }, + }); + await Task.create({ + podId, taskNum: 1, taskId: 'TASK-BUSY', title: 'Already owned', + ...ownership, + }); + + const result = await notifyFoundWork(podId, work, 1, now); + + expect(result.woken).toBe(0); + expect(AgentEventService.enqueue).not.toHaveBeenCalled(); + const installs = await AgentInstallation.find({ podId }).lean(); + expect(installs.every((install) => install.lastFoundWorkAt == null)).toBe(true); +}); + +it.each(['default', 'DEFAULT'])('treats a bot-principal ObjectId claim as occupied capacity for %s', async (instanceId) => { + await AgentInstallation.updateOne({ podId, agentName: 'scout' }, { $set: { instanceId } }); + const principalId = new mongoose.Types.ObjectId(); + await User.collection.insertOne({ + _id: principalId, isBot: true, botMetadata: { agentName: 'Scout', instanceId: 'DeFaUlT' }, + }); + await Task.create({ + podId, taskNum: 2, taskId: 'TASK-PRINCIPAL', title: 'Claimed by bot principal', + status: 'claimed', claimedBy: String(principalId), + }); + + const result = await notifyFoundWork(podId, work, 1, now); + + expect(result.woken).toBe(0); + expect(AgentEventService.enqueue).not.toHaveBeenCalled(); + const install = await AgentInstallation.findOne({ podId, agentName: 'scout' }).lean(); + expect(install.lastFoundWorkAt).toBeNull(); +}); diff --git a/backend/__tests__/service/taskEventService.offeredAt.test.js b/backend/__tests__/service/taskEventService.offeredAt.test.js index f4f56a582c..a996a8f234 100644 --- a/backend/__tests__/service/taskEventService.offeredAt.test.js +++ b/backend/__tests__/service/taskEventService.offeredAt.test.js @@ -15,6 +15,8 @@ jest.mock('../../models/AgentRegistry', () => ({ AgentInstallation: { find: jest.fn(), findOne: jest.fn(), + findOneAndUpdate: jest.fn(), + updateOne: jest.fn(), }, })); @@ -41,14 +43,16 @@ const KernelWorkSweepService = require('../../services/kernelWorkSweepService'); const POD_ID = new mongoose.Types.ObjectId(); const NOW = new Date('2026-08-20T20:00:00.000Z'); -const SINCE = new Date(NOW.getTime() - 10 * 60 * 1000); const install = (agentName = 'scout') => ({ agentName, instanceId: 'default', podId: POD_ID.toString(), status: 'active', - config: { runtime: { runtimeType: 'wrapper' }, boardWake: { enabled: true } }, + config: { + runtime: { runtimeType: 'wrapper' }, boardWake: { enabled: true }, + workDiscovery: { enabled: true, role: 'implementer' }, + }, }); const setInstalls = (installs) => { @@ -97,6 +101,8 @@ beforeEach(async () => { taskNum = 0; setInstalls([install()]); AgentEvent.findOneAndUpdate.mockResolvedValue(null); + AgentInstallation.findOneAndUpdate.mockResolvedValue({}); + AgentInstallation.updateOne.mockResolvedValue({ matchedCount: 1 }); AgentEvent.updateOne.mockResolvedValue({}); AgentEventService.enqueue.mockResolvedValue({}); mockBoardWakeEnabled.mockReturnValue(true); @@ -113,6 +119,19 @@ afterAll(async () => { }); describe('kernel sweep offeredAt filter', () => { + it('still offers a revision deferred past the ten-minute scan window by the hourly cap', async () => { + const old = await createTask({ + taskId: 'TASK-DEFERRED', updatedAt: new Date(NOW.getTime() - 45 * 60 * 1000), + }); + + const sweep = await KernelWorkSweepService.wakeForFoundWork(NOW); + + expect(sweep.woken).toBe(1); + expect(AgentEventService.enqueue.mock.calls[0][0].payload.content).toContain('TASK-DEFERRED'); + const after = await Task.findById(old._id).lean(); + expect(after.offeredAt).toEqual(after.updatedAt); + }); + it('keeps starter rows due until a Guide installation can receive them', async () => { const starter = await createTask({ taskId: 'TASK-STARTER', @@ -135,7 +154,7 @@ describe('kernel sweep offeredAt filter', () => { expect(afterOffer.offeredAt).toEqual(afterOffer.updatedAt); }); - it('retries a swallowed enqueue while the unchanged revision remains in the scan window', async () => { + it('retries a swallowed enqueue while the unchanged revision remains due', async () => { const task = await createTask({ taskId: 'TASK-RETRY', updatedAt: new Date(NOW.getTime() - 30000), @@ -158,7 +177,7 @@ describe('kernel sweep offeredAt filter', () => { expect(afterRetry.offeredAt).toEqual(afterRetry.updatedAt); }); - it('offers missing/rewritten revisions, skips offered/out-of-window rows, and offers each once', async () => { + it('offers missing, rewritten and old due revisions, but skips offered and assigned rows', async () => { const missing = await createTask({ taskId: 'TASK-MISSING', updatedAt: new Date(NOW.getTime() - 30000), @@ -173,9 +192,9 @@ describe('kernel sweep offeredAt filter', () => { updatedAt: new Date(NOW.getTime() - 20000), offeredAt: new Date(NOW.getTime() - 40000), }); - const outsideWindow = await createTask({ + const olderDue = await createTask({ taskId: 'TASK-OLD', - updatedAt: new Date(SINCE.getTime() - 1), + updatedAt: new Date(NOW.getTime() - 11 * 60 * 1000), }); const assigned = await createTask({ taskId: 'TASK-ASSIGNED', @@ -191,7 +210,7 @@ describe('kernel sweep offeredAt filter', () => { expect(content).toContain('TASK-MISSING'); expect(content).toContain('TASK-REWRITTEN'); expect(content).not.toContain('TASK-OFFERED'); - expect(content).not.toContain('TASK-OLD'); + expect(content).toContain('TASK-OLD'); expect(content).not.toContain('TASK-ASSIGNED'); const offeredMissing = await Task.findById(missing._id).lean(); @@ -200,7 +219,7 @@ describe('kernel sweep offeredAt filter', () => { expect(offeredRewritten.offeredAt).toEqual(offeredRewritten.updatedAt); expect((await Task.findById(alreadyOffered._id).lean()).offeredAt) .toEqual(alreadyOffered.offeredAt); - expect((await Task.findById(outsideWindow._id).lean()).offeredAt).toBeUndefined(); + expect((await Task.findById(olderDue._id).lean()).offeredAt).toEqual(olderDue.updatedAt); expect((await Task.findById(assigned._id).lean()).offeredAt).toBeUndefined(); const second = await KernelWorkSweepService.wakeForFoundWork(NOW); @@ -481,7 +500,6 @@ describe('offeredAt race probe', () => { { $match: { _id: { $in: [casTask._id, nowTask._id] }, - updatedAt: { $gte: SINCE }, $expr: { $lt: ['$offeredAt', '$updatedAt'] }, }, }, diff --git a/backend/__tests__/service/tasks.claim-lease.test.js b/backend/__tests__/service/tasks.claim-lease.test.js index 66c82362f6..2332ad214d 100644 --- a/backend/__tests__/service/tasks.claim-lease.test.js +++ b/backend/__tests__/service/tasks.claim-lease.test.js @@ -478,7 +478,12 @@ describe('POST /api/v1/tasks/:podId/:taskId/claim — lease semantics (ADR-018 D } }); - test('PATCH unassignment offers the exact stored revision before the work sweep', async () => { + test('PATCH unassignment leaves the stored revision due for the work sweep', async () => { + await AgentInstallation.updateOne( + { agentName: 'claim-audit-agent', podId: pod._id }, + { $set: { 'config.boardWake.enabled': true, + 'config.workDiscovery': { enabled: true, role: 'implementer' } } }, + ); 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. @@ -501,10 +506,16 @@ describe('POST /api/v1/tasks/:podId/:taskId/claim — lease semantics (ADR-018 D 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()); + expect(offered.offeredAt).toBeUndefined(); - const sweep = await require('../../services/kernelWorkSweepService').wakeForFoundWork(new Date()); - expect(sweep.scannedPods).toBe(0); + // The prior-owner notice does not offer work to idle implementers. + // Only the opted-in discovery sweep offers the unassigned row. Use a clock past the forced monotonic revision above. + const sweep = await require('../../services/kernelWorkSweepService') + .wakeForFoundWork(new Date(stored.updatedAt.getTime() + 1000)); + expect(sweep.scannedPods).toBe(1); + expect(sweep.woken).toBe(1); + const swept = await Task.findById(stored._id); + expect(swept.offeredAt?.getTime()).toBe(swept.updatedAt.getTime()); const note = await request(app) .post(`/api/v1/tasks/${pod._id}/T-1/updates`) @@ -512,8 +523,9 @@ describe('POST /api/v1/tasks/:podId/:taskId/claim — lease semantics (ADR-018 D .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(afterNote.updatedAt.getTime()).toBeGreaterThan(swept.offeredAt.getTime()); + const dueSweep = await require('../../services/kernelWorkSweepService') + .wakeForFoundWork(new Date(afterNote.updatedAt.getTime() + 1000)); expect(dueSweep.scannedPods).toBe(1); const title = await request(app) diff --git a/backend/__tests__/unit/services/kernelWorkSweepService.test.js b/backend/__tests__/unit/services/kernelWorkSweepService.test.js index 5c623fb37d..02fa315462 100644 --- a/backend/__tests__/unit/services/kernelWorkSweepService.test.js +++ b/backend/__tests__/unit/services/kernelWorkSweepService.test.js @@ -338,14 +338,14 @@ describe('sweep', () => { expect(result.woken).toBe(1); }); - it('matches due revisions within the 10-minute scan bound', async () => { + it('matches unoffered revisions regardless of age so the hourly batch can retry', async () => { Task.aggregate.mockResolvedValue([]); await KernelWorkSweepService.sweep(NOW); const [pipeline] = Task.aggregate.mock.calls[0]; const match = pipeline[0].$match; - expect(match.updatedAt.$gte).toEqual(new Date(NOW.getTime() - 10 * 60 * 1000)); + expect(match).not.toHaveProperty('updatedAt'); expect(match.status).toBe('pending'); expect(match.$expr).toEqual({ $lt: ['$offeredAt', '$updatedAt'] }); expect(match.$or).toEqual([ diff --git a/backend/__tests__/unit/services/taskEventService.notifyPodAgents.test.js b/backend/__tests__/unit/services/taskEventService.notifyPodAgents.test.js index 48e7c104eb..062df0f29e 100644 --- a/backend/__tests__/unit/services/taskEventService.notifyPodAgents.test.js +++ b/backend/__tests__/unit/services/taskEventService.notifyPodAgents.test.js @@ -11,7 +11,7 @@ const mongoose = require('mongoose'); jest.mock('../../../models/AgentRegistry', () => ({ - AgentInstallation: { find: jest.fn(), findOne: jest.fn() }, + AgentInstallation: { find: jest.fn(), findOne: jest.fn(), findOneAndUpdate: jest.fn(), updateOne: jest.fn() }, })); jest.mock('../../../services/agentEventService', () => ({ @@ -25,12 +25,14 @@ jest.mock('../../../models/AgentEvent', () => ({ jest.mock('../../../models/User', () => ({ findById: jest.fn(), + find: jest.fn(), })); const mockTaskUpdateOne = jest.fn(); jest.mock('../../../models/Task', () => ({ updateOne: (...args) => mockTaskUpdateOne(...args), find: jest.fn(), + exists: jest.fn(), })); const { AgentInstallation } = require('../../../models/AgentRegistry'); @@ -54,7 +56,10 @@ const install = (agentName, installedBy) => ({ podId: POD_ID, status: 'active', installedBy, - config: { runtime: { runtimeType: 'wrapper' }, wakeOnMessage: { enabled: true } }, + config: { + runtime: { runtimeType: 'wrapper' }, wakeOnMessage: { enabled: true }, + workDiscovery: { enabled: true, role: 'implementer' }, + }, }); const task = (overrides = {}) => ({ @@ -77,12 +82,16 @@ beforeEach(() => { // unless it explicitly seeds a pending wake. AgentEvent.findOneAndUpdate.mockResolvedValue(null); AgentInstallation.findOne.mockReturnValue({ sort: () => ({ lean: async () => null }) }); + AgentInstallation.findOneAndUpdate.mockResolvedValue({}); + AgentInstallation.updateOne.mockResolvedValue({ matchedCount: 1 }); User.findById.mockReturnValue({ select: () => ({ lean: async () => null }) }); + User.find.mockReturnValue({ select: () => ({ lean: async () => [] }) }); AgentEvent.updateOne.mockResolvedValue({}); mockTaskUpdateOne.mockResolvedValue({ matchedCount: 1 }); Task.find.mockReturnValue({ sort: () => ({ limit: () => ({ lean: async () => [] }) }), }); + Task.exists.mockResolvedValue(null); }); describe('notifyPodAgents', () => { @@ -150,6 +159,18 @@ describe('notifyPodAgents', () => { expect(mockTaskUpdateOne).not.toHaveBeenCalled(); }); + it('keeps a pending unassigned revision due even with a stale claim id', async () => { + mockInstalls([install('scout', HUMAN_ID)]); + + await notifyPodAgents(POD_ID, task({ + _id: 'task-1', status: 'pending', assignee: null, claimedBy: 'default', + updatedAt: new Date('2026-08-20T20:00:00.000Z'), + }), 'updated', { userId: HUMAN_ID, isAgent: false }); + + expect(AgentEventService.enqueue).not.toHaveBeenCalled(); + expect(mockTaskUpdateOne).not.toHaveBeenCalled(); + }); + it('rides message.posted, because a bespoke task.* type reaches neither runtime', async () => { mockInstalls([install('scout', HUMAN_ID)]); @@ -443,6 +464,98 @@ describe('notifyFoundWork', () => { updatedAt: new Date(`2026-08-20T20:00:${String(index).padStart(2, '0')}.000Z`), })); + it('caps unclaimed-work nudges per seat and pod for one hour without losing due work', async () => { + mockInstalls([install('scout', HUMAN_ID)]); + const now = new Date('2026-08-20T20:00:00.000Z'); + AgentInstallation.findOneAndUpdate.mockResolvedValue(null); + + await notifyFoundWork(POD_ID, offerRows(1), 1, now); + + expect(AgentInstallation.findOneAndUpdate).toHaveBeenCalledWith( + { + agentName: 'scout', instanceId: 'default', podId: POD_ID, status: 'active', + $or: [ + { lastFoundWorkAt: { $lte: new Date(now.getTime() - 60 * 60 * 1000) } }, + { lastFoundWorkAt: null }, + ], + }, + { $set: { lastFoundWorkAt: now } }, + { new: true, timestamps: false }, + ); + expect(AgentEventService.enqueue).not.toHaveBeenCalled(); + expect(mockTaskUpdateOne).not.toHaveBeenCalled(); + }); + + it('never nudges a reviewer or a seat without explicit implementer opt-in', async () => { + mockInstalls([ + { ...install('sprint-review', HUMAN_ID), config: { + wakeOnMessage: { enabled: true }, workDiscovery: { enabled: true, role: 'reviewer' }, + } }, + { ...install('legacy', HUMAN_ID), config: { wakeOnMessage: { enabled: true } } }, + install('sprint-impl', HUMAN_ID), + ]); + + await notifyFoundWork(POD_ID, offerRows(1), 1); + + expect(AgentEventService.enqueue.mock.calls.map(([arg]) => arg.agentName)).toEqual(['sprint-impl']); + expect(Task.exists).toHaveBeenCalledTimes(1); + }); + + it('skips an implementer with assigned or claimed open work', async () => { + mockInstalls([install('sprint-impl', HUMAN_ID)]); + Task.exists.mockResolvedValue({ _id: 'busy-task' }); + + await notifyFoundWork(POD_ID, offerRows(1), 1); + + expect(Task.exists).toHaveBeenCalledWith({ + podId: POD_ID, + status: { $in: ['pending', 'claimed'] }, + $or: [ + { assignee: /^sprint-impl$/i }, + { status: 'claimed', claimedBy: /^sprint-impl$/i }, + ], + }); + expect(AgentInstallation.findOneAndUpdate).not.toHaveBeenCalled(); + expect(AgentEventService.enqueue).not.toHaveBeenCalled(); + expect(mockTaskUpdateOne).not.toHaveBeenCalled(); + }); + + it('skips a seat whose claim is stored as its bot User id', async () => { + mockInstalls([install('sprint-impl', HUMAN_ID)]); + const principalId = new mongoose.Types.ObjectId(); + User.find.mockReturnValue({ select: () => ({ lean: async () => [{ _id: principalId }] }) }); + Task.exists.mockResolvedValueOnce(null).mockResolvedValueOnce({ _id: 'busy-task' }); + + await notifyFoundWork(POD_ID, offerRows(1), 1); + + expect(User.find).toHaveBeenCalledWith({ + isBot: true, + 'botMetadata.agentName': /^sprint-impl$/i, + 'botMetadata.instanceId': { $in: [/^default$/i, /^sprint-impl$/i, null] }, + }); + expect(Task.exists).toHaveBeenLastCalledWith({ + podId: POD_ID, + status: 'claimed', + claimedBy: { $in: [String(principalId)] }, + }); + expect(AgentEventService.enqueue).not.toHaveBeenCalled(); + }); + + it('releases a reserved slot when the nudge enqueue fails', async () => { + mockInstalls([install('scout', HUMAN_ID)]); + const now = new Date('2026-08-20T20:00:00.000Z'); + AgentEventService.enqueue.mockRejectedValueOnce(new Error('store down')); + + await notifyFoundWork(POD_ID, offerRows(1), 1, now); + + expect(AgentInstallation.updateOne).toHaveBeenCalledWith( + { agentName: 'scout', instanceId: 'default', podId: POD_ID, lastFoundWorkAt: now }, + { $unset: { lastFoundWorkAt: 1 } }, + { timestamps: false }, + ); + expect(mockTaskUpdateOne).not.toHaveBeenCalled(); + }); + it('names five rows but stamps every aggregate row, including rows 6+', async () => { mockInstalls([install('scout', HUMAN_ID)]); const rows = offerRows(9); diff --git a/backend/models/AgentRegistry.ts b/backend/models/AgentRegistry.ts index d0de075984..38e980e859 100644 --- a/backend/models/AgentRegistry.ts +++ b/backend/models/AgentRegistry.ts @@ -210,6 +210,7 @@ export interface IAgentInstallationRegistry extends Document { status: AgentInstallationStatus; errorMessage?: string; staleSince?: Date; + lastFoundWorkAt?: Date | null; installedBy: Types.ObjectId; skillLoad?: { at: Date; mounted: string[]; conflicted: string[]; status?: 'available' | 'unavailable' }; usage: { @@ -248,6 +249,8 @@ const AgentInstallationSchema = new Schema( status: { type: String, enum: ['active', 'paused', 'uninstalled', 'error', 'stale'], default: 'active' }, errorMessage: String, staleSince: { type: Date }, + // CAS throttle for the kernel's unclaimed-work nudge, per seat and pod. + lastFoundWorkAt: { type: Date, default: null }, installedBy: { type: Schema.Types.ObjectId, ref: 'User', required: true }, skillLoad: { type: new Schema({ at: Date, mounted: [String], conflicted: [String], status: { type: String, enum: ['available', 'unavailable'] } }, { _id: false }), diff --git a/backend/services/kernelWorkSweepService.ts b/backend/services/kernelWorkSweepService.ts index 4a25caf10f..c3588ec069 100644 --- a/backend/services/kernelWorkSweepService.ts +++ b/backend/services/kernelWorkSweepService.ts @@ -69,15 +69,9 @@ const LIVE_STATES = new Set(['listening']); // "at most three deferrals" — verbatim. Per LEASE, not per task: the claim // route resets the counter, so this bounds one holder's silence at -// 3 x SWEEP_INTERVAL_MS (~30 min) past a lapsed lease, never a task's lifetime. +// Three ten-minute sweeps (~30 min) past a lapsed lease, never a task's lifetime. const MAX_RESCUE_DEFERRALS = 3; -// MUST match the cron cadence in schedulerService ('4,14,24,... * * * *' = -// every 10 minutes). This is only a scan bound now; offeredAt vs updatedAt -// records whether a revision was delivered. One cadence window catches rows -// changed between passes without scanning older standing stock. -const SWEEP_INTERVAL_MS = 10 * 60 * 1000; - interface SweepResult { scannedPods: number; rescued: number; @@ -300,15 +294,14 @@ class KernelWorkSweepService { // eslint-disable-next-line @typescript-eslint/no-require-imports, global-require const { notifyFoundWork } = require('./taskEventService'); - // updatedAt bounds the scan to the current sweep window; offeredAt records - // whether this exact revision was sent. A missing offeredAt sorts below a - // Date and is due. Rescue writes advance updatedAt as well. - const since = new Date(now.getTime() - SWEEP_INTERVAL_MS); + // offeredAt records whether this exact revision was sent. Do not bound + // updatedAt to this sweep window: a seat's hourly nudge cap can defer an + // unassigned revision beyond ten minutes. It must remain due for the next + // eligible hourly batch. A missing offeredAt sorts below a Date and is due. const actionable = await Task.aggregate([ { $match: { status: 'pending', - updatedAt: { $gte: since }, $expr: { $lt: ['$offeredAt', '$updatedAt'] }, $or: [{ assignee: null }, { assignee: '' }, { assignee: { $exists: false } }], }, diff --git a/backend/services/taskEventService.ts b/backend/services/taskEventService.ts index 6bfb5479a8..2fa5f411d1 100644 --- a/backend/services/taskEventService.ts +++ b/backend/services/taskEventService.ts @@ -19,6 +19,7 @@ let ioRef: SocketIOLike | null = null; export type TaskEventKind = 'created' | 'updated' | 'deleted'; const MAX_NAMED_ITEMS = 5; const MAX_ASSIGNED_ROWS = 5; +const FOUND_WORK_BATCH_INTERVAL_MS = 60 * 60 * 1000; type BoardTask = Record; function seatMatchesAssignee(install: Record, assignee: unknown): boolean { @@ -29,6 +30,11 @@ function seatMatchesAssignee(install: Record, assignee: unknown return instance === 'default' ? assigned === name : assigned === instance; } +function exactSeatKeyIgnoringCase(key: unknown): RegExp { + const escaped = String(key).replace(/[.*+?^${}()|[\]\\]/g, '\\$&'); + return new RegExp(`^${escaped}$`, 'i'); +} + async function claimHolderKey(claimedBy: unknown): Promise { if (!claimedBy) return null; const claim = String(claimedBy); @@ -412,7 +418,12 @@ export async function notifyPodAgents( })); const failed = outcomes.some((outcome) => outcome === undefined); const offered = outcomes.some((outcome) => outcome === true); - if (offered && !failed) { + // A wake to the prior owner after unassignment cannot retire a pending + // unassigned row from the kernel's found-work scan. + // Match wakeForFoundWork's status/assignee predicate exactly; legacy rows + // can retain a stale claimedBy even while pending and unassigned. + const unclaimedWork = String(task.status) === 'pending' && !task.assignee; + if (offered && !failed && !unclaimedWork) { await stampOfferedRevisions([task]); } } catch (err) { @@ -458,11 +469,21 @@ export async function notifyFoundWork( const AgentEventService = require('./agentEventService'); const { boardWakeEnabled } = require('./agentMentionService'); const AgentEvent = require('../models/AgentEvent'); + const Task = require('../models/Task'); + const User = require('../models/User'); /* eslint-enable global-require, @typescript-eslint/no-require-imports */ const normalizedPodId = String(podId); const active = await AgentInstallation.find({ podId: normalizedPodId, status: 'active' }).lean(); - const installs = (active || []).filter(boardWakeEnabled); + // Board subscriptions include reviewers. Discovery is an implementer-only + // opt-in, independent of whether a seat receives addressed row changes. + const installs = (active || []).filter((install: Record) => { + if (!boardWakeEnabled(install)) return false; + const config = install.config as Map | Record | undefined; + const discovery = config instanceof Map ? config.get('workDiscovery') : config?.workDiscovery; + return (discovery as { enabled?: unknown; role?: unknown } | undefined)?.enabled === true + && (discovery as { role?: unknown })?.role === 'implementer'; + }); if (!installs.length) return { woken: 0 }; // #1080 part 3: a rescued row names who it lapsed from. Without it the wake @@ -495,7 +516,59 @@ export async function notifyFoundWork( const outcomes = await Promise.all(installs.map(async (install: Record) => { const agentName = install.agentName; const instanceId = install.instanceId || 'default'; + const isDefaultInstance = String(instanceId).toLowerCase() === 'default'; + let reservedSlot = false; try { + // Assigned pending work and active claims already occupy this seat in + // this pod. The unclaimed nudge is only for spare capacity; do not + // spend a turn offering more work to a seat with an open row. + const seatKey = exactSeatKeyIgnoringCase(isDefaultInstance ? agentName : instanceId); + let hasOpenWork = await Task.exists({ + podId: normalizedPodId, + status: { $in: ['pending', 'claimed'] }, + $or: [ + { assignee: seatKey }, + { status: 'claimed', claimedBy: seatKey }, + ], + }); + if (!hasOpenWork) { + // Wrapper/MCP claim writers often store the bot User ObjectId, + // whereas board assignments store the seat key. Resolve those + // principals before deciding that an unassigned-looking seat is idle. + const principalInstances = isDefaultInstance + ? [exactSeatKeyIgnoringCase('default'), exactSeatKeyIgnoringCase(agentName), null] + : [exactSeatKeyIgnoringCase(instanceId)]; + const principals = await User.find({ + isBot: true, + 'botMetadata.agentName': exactSeatKeyIgnoringCase(agentName), + 'botMetadata.instanceId': { $in: principalInstances }, + }).select('_id').lean(); + if (principals.length) { + hasOpenWork = await Task.exists({ + podId: normalizedPodId, + status: 'claimed', + claimedBy: { $in: principals.map((principal: { _id: unknown }) => String(principal._id)) }, + }); + } + } + if (hasOpenWork) return false; + + // Reserve the one-hour slot on the installation itself. A read of + // prior events followed by enqueue races another sweep during rollout; + // this CAS makes at most one worker win for a seat and pod. + const reserved = await AgentInstallation.findOneAndUpdate( + { + agentName, instanceId, podId: normalizedPodId, status: 'active', + $or: [ + { lastFoundWorkAt: { $lte: new Date(now.getTime() - FOUND_WORK_BATCH_INTERVAL_MS) } }, + { lastFoundWorkAt: null }, + ], + }, + { $set: { lastFoundWorkAt: now } }, + { new: true, timestamps: false }, + ); + if (!reserved) return false; + reservedSlot = true; const folded = await AgentEvent.findOneAndUpdate( { agentName, instanceId, podId: normalizedPodId, status: 'pending', 'payload.boardWake': true, @@ -523,6 +596,17 @@ export async function notifyFoundWork( return true; } catch (err) { console.warn(`[task-event] found-work enqueue failed for ${agentName}:`, (err as Error).message); + // A failed enqueue did not send a nudge. Restore eligibility only if + // this reservation is still ours; work remains due for the next sweep. + if (reservedSlot) { + await AgentInstallation.updateOne( + { agentName, instanceId, podId: normalizedPodId, lastFoundWorkAt: now }, + { $unset: { lastFoundWorkAt: 1 } }, + { timestamps: false }, + ).catch((rollbackErr: Error) => { + console.warn(`[task-event] found-work reservation rollback failed for ${agentName}:`, rollbackErr.message); + }); + } } })); if (woken > 0 && outcomes.every((outcome) => outcome === true)) { diff --git a/docs/adr/ADR-018-agent-attention-claims.md b/docs/adr/ADR-018-agent-attention-claims.md index d7f4e19fa0..c9973a89a6 100644 --- a/docs/adr/ADR-018-agent-attention-claims.md +++ b/docs/adr/ADR-018-agent-attention-claims.md @@ -426,6 +426,12 @@ Implementation: `backend/services/wakeDefaultService.ts` (the rule), the registr Prior ownership must be the pre-image of the successful board write. A separate read before a concurrent write can name a stale assignee and miss the seat just unassigned. The wake's `updatedAt` must equal the stored revision and advance under concurrent writes, so `offeredAt` can mark that exact revision once. Every later task write must advance the same revision, including claims, completions, comments, non-ownership edits, and source-ref reopens; a wall-clock timestamp can otherwise move it behind `offeredAt` and hide an unassigned pending row from the sweep. +### D8.3 — AMENDMENT (2026-10-24): discovery wakes require an idle implementer + +The D8.2 discovery exception woke reviewers who had no implementation capacity: one review seat received 351 unclaimed-work wakes in about 11.5 hours and answered every one with `NO_REPLY` (TASK-464). This was the kernel finding work correctly and choosing the wrong audience. + +Discovery now requires an explicit `config.workDiscovery: { enabled: true, role: 'implementer' }` on the installation. It skips a seat with assigned pending work or an active claim in that pod, whether the claim stores a seat key or its bot User id, and caps delivery at one nudge per seat and pod per hour. Missing role configuration fails closed. An unoffered revision remains eligible beyond the sweep's ten-minute tick so the next hourly batch can carry it. A capped nudge does not mark an unassigned row as offered. Browser board updates and addressed owner wakes are unaffected. + ### D9 — Heartbeat shrinks; it does not die Wake-on-message replaces heartbeat's *"did anything happen?"* role in pods