diff --git a/backend/__tests__/service/assignedTaskWakeService.skipLog.test.js b/backend/__tests__/service/assignedTaskWakeService.skipLog.test.js new file mode 100644 index 0000000000..e9e085685f --- /dev/null +++ b/backend/__tests__/service/assignedTaskWakeService.skipLog.test.js @@ -0,0 +1,116 @@ +const mongoose = require('mongoose'); +const { setupMongoDb, closeMongoDb, clearMongoDb } = require('../utils/testUtils'); + +jest.mock('jsonwebtoken', () => ({})); +jest.mock('../../services/agentEventService', () => ({ enqueue: jest.fn().mockResolvedValue({}) })); +const Task = require('../../models/Task'); +const User = require('../../models/User'); +const AgentEvent = require('../../models/AgentEvent'); +const AgentRun = require('../../models/AgentRun'); +const { AgentInstallation } = require('../../models/AgentRegistry'); +const { dispatchAssignedTaskWakes } = require('../../services/assignedTaskWakeService'); + +// The skip log is rate-limited in memory for the life of the module, so each +// test runs at its own hour and cannot be silenced by an earlier one. +const START = new Date('2026-10-24T11:00:00Z').getTime(); +let hour = 0; +let now; +const minutesFrom = (base, minutes) => new Date(base.getTime() + minutes * 60000); +let podId; +let debug; + +async function completed(minutesAgo) { + const at = minutesFrom(now, -minutesAgo); + return AgentEvent.collection.insertOne({ + agentName: 'codex', + instanceId: 'forge', + podId, + type: 'chat.mention', + status: 'acked', + deliveryMode: 'wake', + delivery: { outcome: 'no_action', reasonRecorded: true, updatedAt: at }, + createdAt: at, + updatedAt: at, + }); +} + +async function fixture({ principal = true } = {}) { + podId = new mongoose.Types.ObjectId(); + const installer = await User.create({ + username: 'forge-skip-installer', email: 'forge-skip-installer@example.test', password: 'test-password', + }); + if (principal) { + await User.create({ + username: 'forge-skip-test', + email: 'forge-skip@example.test', + password: 'test-password', + isBot: true, + botType: 'agent', + botMetadata: { agentName: 'codex', instanceId: 'forge' }, + }); + } + await AgentInstallation.create({ + agentName: 'codex', instanceId: 'forge', podId, version: '1', installedBy: installer._id, status: 'active', + }); + await Task.create({ + podId, taskNum: 1, taskId: 'TASK-1', title: 'Do the assigned work', assignee: 'forge', + }); +} + +const skips = () => debug.mock.calls.map(([line]) => line) + .filter((line) => String(line).startsWith('[assigned-task-wake] skip')); + +beforeAll(setupMongoDb); +afterAll(closeMongoDb); +beforeEach(async () => { + hour += 1; + now = new Date(START + hour * 3600000); + await clearMongoDb(); + debug = jest.spyOn(console, 'debug').mockImplementation(() => {}); +}); +afterEach(() => debug.mockRestore()); + +it('names the seat, the rows and the gate, once per thirty minutes', async () => { + await fixture(); + await completed(2); + await dispatchAssignedTaskWakes(now); + await dispatchAssignedTaskWakes(minutesFrom(now, 1)); + expect(skips()).toEqual(['[assigned-task-wake] skip seat=codex/forge rows=TASK-1 gate=delay']); + await AgentRun.create({ + agentName: 'codex', instanceId: 'forge', podId, trigger: 'mention', status: 'running', + }); + await dispatchAssignedTaskWakes(minutesFrom(now, 10)); + await dispatchAssignedTaskWakes(minutesFrom(now, 20)); + expect(skips()).toEqual([ + '[assigned-task-wake] skip seat=codex/forge rows=TASK-1 gate=delay', + '[assigned-task-wake] skip seat=codex/forge rows=TASK-1 gate=other-work', + ]); + await dispatchAssignedTaskWakes(minutesFrom(now, 40)); + expect(skips()).toHaveLength(3); + expect(skips()[2]).toBe('[assigned-task-wake] skip seat=codex/forge rows=TASK-1 gate=other-work'); +}); + +it('logs nothing when the seat is woken, then names the seat interval', async () => { + await fixture(); + await completed(6); + expect((await dispatchAssignedTaskWakes(now)).woken).toBe(1); + expect(skips()).toEqual([]); + expect((await dispatchAssignedTaskWakes(minutesFrom(now, 10))).woken).toBe(0); + expect(skips()).toEqual(['[assigned-task-wake] skip seat=codex/forge rows=TASK-1 gate=seat-interval']); +}); + +it('names a seat with assigned rows but no principal', async () => { + await fixture({ principal: false }); + await completed(6); + await dispatchAssignedTaskWakes(now); + expect(skips()).toEqual(['[assigned-task-wake] skip seat=codex/forge rows=TASK-1 gate=no-principal']); +}); + +it('names the row a self-created wake is holding for a person', async () => { + await fixture(); + await Task.updateOne({ podId, taskId: 'TASK-1' }, { $set: { source: 'agent', taskId: 'N-1-codex-abc' } }); + await completed(6); + expect((await dispatchAssignedTaskWakes(now)).woken).toBe(1); + expect((await dispatchAssignedTaskWakes(minutesFrom(now, 45))).woken).toBe(0); + expect(skips()).toEqual(['[assigned-task-wake] skip seat=codex/forge rows=N-1-codex-abc gate=awaiting-human']); +}); diff --git a/backend/services/assignedTaskWakeService.ts b/backend/services/assignedTaskWakeService.ts index e9826ba9b8..85d6bab74e 100644 --- a/backend/services/assignedTaskWakeService.ts +++ b/backend/services/assignedTaskWakeService.ts @@ -35,6 +35,25 @@ type RowState = { type WakeEvent = { _id: unknown; status?: string; delivery?: { outcome?: string; reason?: string | null } }; const rowKey = (seat: string, row: Row): string => `${seat}|${String(row._id)}`; + +// The dispatcher runs every minute, so a gate that keeps holding a seat would +// repeat its line each run. Each (seat, rows, gate) logs once per interval. +const skipLoggedAt = new Map(); +function logSkip(install: Installation, rows: Row[], gate: string, now: Date): void { + const seat = `${install.agentName}/${install.instanceId || 'default'}`; + const taskIds = rows.map((row) => row.taskId).join(',') || '-'; + const key = `${seat}|${taskIds}|${gate}`; + const last = skipLoggedAt.get(key); + if (last !== undefined && now.getTime() - last < INTERVAL_MS) return; + if (skipLoggedAt.size > 5000) { + skipLoggedAt.forEach((at, logged) => { + if (now.getTime() - at >= INTERVAL_MS) skipLoggedAt.delete(logged); + }); + } + skipLoggedAt.set(key, now.getTime()); + console.debug(`[assigned-task-wake] skip seat=${seat} rows=${taskIds} gate=${gate}`); +} + // Native rows record no author: their id is minted `N---`. // Rows created through the API open with a "Created by" update carrying the author's id. const NATIVE_TASK_ID = /^N-\d+-(.+)-[0-9a-z]+$/; @@ -166,6 +185,7 @@ async function wakeableRows( if (selfCreated(row, install, principalId)) { // eslint-disable-next-line no-await-in-loop if (!state?.lastWakeAt || await humanActedSince(new Date(state.lastWakeAt))) wakeable.push(row); + else logSkip(install, [row], 'awaiting-human', now); continue; } if (!state) { @@ -180,10 +200,14 @@ async function wakeableRows( wakeable.push(row); continue; } - if (state.stoppedAt) continue; + if (state.stoppedAt) { + logSkip(install, [row], 'stopped', now); + continue; + } if ((state.failures || 0) >= STOP_AFTER_FAILURES) { // eslint-disable-next-line no-await-in-loop await stopRow(row, state, install, principalId, now); + logSkip(install, [row], 'stopped', now); continue; } wakeable.push(row); @@ -199,14 +223,14 @@ async function holdsFloor(install: Installation, principalId: string): Promise String(row.podId) === String(install.podId) && owns(row, install, null)); + if (named.length) logSkip(install, named, 'no-principal', now); + continue; + } const assigned = rows.filter((row) => String(row.podId) === String(install.podId) && owns(row, install, principalId)); if (!assigned.length) continue; @@ -288,13 +316,25 @@ export async function dispatchAssignedTaskWakes(now = new Date()): Promise<{ eli }; // eslint-disable-next-line no-await-in-loop const last = await AgentEvent.findOne(completionFilter).sort({ 'delivery.updatedAt': -1 }).lean(); - if (!last?.delivery?.updatedAt) continue; + if (!last?.delivery?.updatedAt) { + logSkip(install, owned, 'no-completed-turn', now); + continue; + } const dueAt = new Date(new Date(last.delivery.updatedAt).getTime() + DELAY_MS); - if (dueAt > now) continue; + if (dueAt > now) { + logSkip(install, owned, 'delay', now); + continue; + } // eslint-disable-next-line no-await-in-loop - if (await hasOtherWork(install.agentName, install.instanceId)) continue; + if (await hasOtherWork(install.agentName, install.instanceId)) { + logSkip(install, owned, 'other-work', now); + continue; + } // eslint-disable-next-line no-await-in-loop - if (await holdsFloor(install, principalId)) continue; + if (await holdsFloor(install, principalId)) { + logSkip(install, owned, 'floor', now); + continue; + } eligible += 1; let reserved = false; let enqueueStarted = false; @@ -321,6 +361,7 @@ export async function dispatchAssignedTaskWakes(now = new Date()): Promise<{ eli if (!active || busy || !current.length || String(newest?._id) !== String(last._id)) { // eslint-disable-next-line no-await-in-loop await AgentTaskWakeState.updateOne({ _id: key, lastWakeAt: now }, { $unset: { lastWakeAt: 1 } }); + logSkip(install, owned, 'recheck', now); continue; } const shown = current.slice(0, 5); @@ -369,7 +410,10 @@ export async function dispatchAssignedTaskWakes(now = new Date()): Promise<{ eli console.error('[assigned-task-wake] wake not linked to rows:', key, (stateError as Error).message); } } catch (error) { - if ((error as { code?: number }).code === 11000 && !reserved) continue; + if ((error as { code?: number }).code === 11000 && !reserved) { + logSkip(install, owned, 'seat-interval', now); + continue; + } if (reserved && !enqueueStarted) { // No queue write was attempted, so returning this slot is safe. // eslint-disable-next-line no-await-in-loop