diff --git a/src/app/api/dn-batches/tick/route.test.ts b/src/app/api/dn-batches/tick/route.test.ts index e71ed86..bfd2950 100644 --- a/src/app/api/dn-batches/tick/route.test.ts +++ b/src/app/api/dn-batches/tick/route.test.ts @@ -39,13 +39,15 @@ import { BatchLeaseLostError, BatchLockedError, BatchSchemaRequiredError } from /* every constant the factory touches must be hoisted too. */ /* ------------------------------------------------------------------ */ -const { dnBatchesGetFullList, dnBatchesGetOne, dnBatchesUpdate, runBatchMock, runStagedBatchMock, viPbAdmin } = +const { dnBatchesGetFullList, dnBatchesGetOne, dnBatchesUpdate, runBatchMock, runStagedBatchMock, tickMcpJobsMock, createPurchasingContextMock, viPbAdmin } = vi.hoisted(() => ({ dnBatchesGetFullList: vi.fn(), dnBatchesGetOne: vi.fn(), dnBatchesUpdate: vi.fn(), runBatchMock: vi.fn(), runStagedBatchMock: vi.fn(), + tickMcpJobsMock: vi.fn(), + createPurchasingContextMock: vi.fn(), viPbAdmin: vi.fn(), })); @@ -82,6 +84,12 @@ vi.mock("@/lib/dn-pipeline/run", () => ({ vi.mock("@/lib/dn-pipeline/run-staged", () => ({ runStagedBatch: runStagedBatchMock, })); +vi.mock("@/lib/purchasing/jobs", () => ({ + tickMcpJobs: tickMcpJobsMock, +})); +vi.mock("@/lib/purchasing/client", () => ({ + createPurchasingContext: createPurchasingContextMock, +})); import { pbAdmin } from "@/lib/pb"; import { POST } from "./route"; @@ -129,10 +137,16 @@ beforeEach(() => { dnBatchesUpdate.mockReset(); runBatchMock.mockReset(); runStagedBatchMock.mockReset(); + tickMcpJobsMock.mockReset(); + createPurchasingContextMock.mockReset(); viPbAdmin.mockReset(); viPbAdmin.mockResolvedValue(makePb() as never); + // The MCP drain builds its own admin-backed purchasing context. + createPurchasingContextMock.mockResolvedValue({ pb: makePb(), principal: { id: "dn-tick", permissions: new Set(["process"]) } }); + tickMcpJobsMock.mockResolvedValue({ claimed: 0, completed: 0, failed: 0 }); + // Default: no stale claims, one inbox candidate. dnBatchesGetFullList.mockImplementation( async (opts: { filter?: string } = {}) => { @@ -659,6 +673,75 @@ describe("POST /api/dn-batches/tick — admin auth is awaited", () => { }); }); +describe("POST /api/dn-batches/tick — queued MCP jobs", () => { + // process_delivery_batch only writes an mcp_jobs row. Nothing was draining + // it (no MCP_JOB_SCHEDULE_CONFIGURED in production, and /api/mcp/jobs/tick + // had no caller), so agent batch jobs sat in 'queued' forever. + it("drains one queued MCP job even when there are no inbox batches", async () => { + dnBatchesGetFullList.mockImplementation( + async (opts: { filter?: string } = {}) => { + if (opts.filter?.includes("status='inbox'")) return []; + return []; + }, + ); + + const res = await POST(postWithValidAuth()); + + expect(res.status).toBe(200); + expect(tickMcpJobsMock).toHaveBeenCalledTimes(1); + // The drain principal needs 'process' or tickMcpJobs throws permission_denied. + const [contextArg, optionsArg] = tickMcpJobsMock.mock.calls[0]; + expect(contextArg.principal.permissions.has("process")).toBe(true); + expect(optionsArg).toEqual({ limit: 1, deadlineMs: 120000 }); + }); + + it("drains after the inbox batches have run", async () => { + const order: string[] = []; + runBatchMock.mockImplementation(async () => { + order.push("inbox"); + return { status: "processed", notesDetected: 1, notesPromoted: 1, checks: [] }; + }); + tickMcpJobsMock.mockImplementation(async () => { + order.push("mcp"); + return { claimed: 1, completed: 1, failed: 0 }; + }); + + await POST(postWithValidAuth()); + + expect(order).toEqual(["inbox", "mcp"]); + }); + + it("reports the drain alongside the batch results so the n8n log shows it ran", async () => { + tickMcpJobsMock.mockResolvedValue({ claimed: 1, completed: 1, failed: 0 }); + + const body = await (await POST(postWithValidAuth())).json(); + + expect(body.mcpJobs).toEqual({ claimed: 1, completed: 1, failed: 0 }); + // The drain is not a batch: it must not inflate claimed or the results. + expect(body.claimed).toBe(1); + expect(body.results).toHaveLength(1); + }); + + // Without MINIMAX_API_KEY tickMcpJobs throws processing_setup_required, + // and without the n8n secret this route is not the scheduler at all. + it("still returns 200 when the drain throws", async () => { + tickMcpJobsMock.mockRejectedValue(new Error("processing_setup_required")); + + const res = await POST(postWithValidAuth()); + + expect(res.status).toBe(200); + expect(await res.json()).toMatchObject({ ok: true }); + }); + + it("does not drain on an unauthorized request", async () => { + const res = await POST(makeRequest({ headers: { "x-ingest-secret": "wrong" } })); + + expect(res.status).toBe(401); + expect(tickMcpJobsMock).not.toHaveBeenCalled(); + expect(createPurchasingContextMock).not.toHaveBeenCalled(); + }); +}); + describe("POST /api/dn-batches/tick — response shape", () => { it("returns ok=true, claimed and a results array on success", async () => { const res = await POST(postWithValidAuth()); diff --git a/src/app/api/dn-batches/tick/route.ts b/src/app/api/dn-batches/tick/route.ts index 3439c0a..edf80a5 100644 --- a/src/app/api/dn-batches/tick/route.ts +++ b/src/app/api/dn-batches/tick/route.ts @@ -4,6 +4,9 @@ import { pbAdmin } from "@/lib/pb"; import { runBatch } from "@/lib/dn-pipeline/run"; import { runStagedBatch } from "@/lib/dn-pipeline/run-staged"; import { BatchLeaseLostError, BatchLockedError, BatchSchemaRequiredError } from "@/lib/dn-pipeline/lock"; +import { createPurchasingContext } from "@/lib/purchasing/client"; +import { tickMcpJobs } from "@/lib/purchasing/jobs"; +import type { Permission } from "@/lib/purchasing/types"; export const dynamic = "force-dynamic"; export const runtime = "nodejs"; @@ -114,7 +117,32 @@ export async function POST(request: Request) { } } - return NextResponse.json({ ok: true, claimed: results.length, results }); + // Drain one queued agent job. + // + // process_delivery_batch only writes an mcp_jobs row; without this it sat in + // 'queued' forever, because /api/mcp/jobs/tick had no caller and + // MCP_JOB_SCHEDULE_CONFIGURED was never set in production. n8n already + // calls this route, so this is the scheduler that actually exists. + // + // Reported as its own top-level field, not a `results` entry: `results` and + // `claimed` count inbox batches, and an operator reading the n8n execution + // log counts batches, not jobs. + // + // Deferred, never fatal: a missing MINIMAX_API_KEY makes tickMcpJobs throw + // processing_setup_required, and a drain failure must not turn a successful + // inbox tick into an error for the operator. + let mcpJobs: { claimed: number; completed: number; failed: number } | undefined; + try { + const principal = { id: "dn-tick", permissions: new Set(["process"]) }; + mcpJobs = await tickMcpJobs(await createPurchasingContext(principal), { limit: 1, deadlineMs: 120000 }); + } catch (error) { + console.warn("[dn-tick] mcp job drain deferred", { error: String(error) }); + } + + // Only reported when the drain actually did something, so a quiet tick keeps + // the exact response shape it always had. + const drained = mcpJobs && (mcpJobs.claimed > 0 || mcpJobs.completed > 0 || mcpJobs.failed > 0); + return NextResponse.json({ ok: true, claimed: results.length, results, ...(drained ? { mcpJobs } : {}) }); } async function claim(pb: Awaited>, id: string): Promise { diff --git a/src/lib/purchasing/jobs.test.ts b/src/lib/purchasing/jobs.test.ts index 5eb5651..12b87e5 100644 --- a/src/lib/purchasing/jobs.test.ts +++ b/src/lib/purchasing/jobs.test.ts @@ -3,7 +3,7 @@ import {memoryDatabase} from './test-support'; const staged=vi.hoisted(()=>vi.fn(async()=>({status:'needs_attention',notesPromoted:0,notClaimed:false}))); vi.mock('@/lib/dn-pipeline/run-staged',()=>({runStagedBatch:staged})); import {canReclaimJob,enqueueDeliveryJob,tickMcpJobs} from './jobs'; -beforeEach(()=>{vi.stubEnv('MINIMAX_API_KEY','fixture-only');staged.mockClear();});afterEach(()=>vi.unstubAllEnvs()); +beforeEach(()=>{vi.stubEnv('MINIMAX_API_KEY','fixture-only');staged.mockClear();});afterEach(()=>{vi.unstubAllEnvs();}); it('reclaims only expired job acquisitions',()=>{ expect(canReclaimJob('2026-10-05T09:59:59Z',Date.parse('2026-10-05T10:00:00Z'))).toBe(true); expect(canReclaimJob('2026-10-05T10:00:01Z',Date.parse('2026-10-05T10:00:00Z'))).toBe(false); @@ -37,3 +37,16 @@ it('terminates a crashed third attempt after reclaiming its lease without a four expect(store('mcp_jobs').get('abcdefghijklmn2')?.status).toBe('failed');expect(staged).not.toHaveBeenCalled(); expect(store('mcp_operations').get('abcdefghijklmn1')?.result).toMatchObject({status:'failed',error:'final_attempt_interrupted'}); }); + +// The only scheduler that runs is the n8n tick, so its secret being configured +// is the signal that the queue will actually drain. +it('reports the job as scheduled when the n8n tick secret is configured',async()=>{ + const {pb}=memoryDatabase();await pb.collection('dn_batches').create({id:'abcdefghijklmn1'}); + const writer={pb,principal:{id:'writer',permissions:new Set<'read'|'write'>(['read','write'])}}; + vi.stubEnv('N8N_INGEST_SECRET','n8n-secret'); + const enqueued=await enqueueDeliveryJob(writer,{operationId:'7f5a9e21-63c7-4a02-a45a-9ab18d3bdc02',batchId:'abcdefghijklmn1'}); + expect(enqueued.result).toMatchObject({status:'queued',scheduleConfigured:true}); + vi.stubEnv('N8N_INGEST_SECRET',''); + const unscheduled=await enqueueDeliveryJob(writer,{operationId:'1b7c4d92-63c7-4a02-a45a-9ab18d3bdc02',batchId:'abcdefghijklmn1'}); + expect(unscheduled.result).toMatchObject({status:'queued',scheduleConfigured:false}); +}); diff --git a/src/lib/purchasing/jobs.ts b/src/lib/purchasing/jobs.ts index a8f4e1d..4691aab 100644 --- a/src/lib/purchasing/jobs.ts +++ b/src/lib/purchasing/jobs.ts @@ -13,7 +13,10 @@ export async function enqueueDeliveryJob(ctx:PurchasingContext,input:Record{ await lease.create('mcp_jobs',{id:jobId,operation,batch:args.batchId,status:'queued',attempts:0,next_attempt_at:new Date().toISOString()}); - return {jobId,status:'queued',scheduleConfigured:process.env.MCP_JOB_SCHEDULE_CONFIGURED==='true'}; + // The n8n tick drains the queue (see /api/dn-batches/tick), so its secret is + // the real signal that a queued job will ever run. The old flag was never + // set in production, which is why jobs sat in 'queued' forever. + return {jobId,status:'queued',scheduleConfigured:!!process.env.N8N_INGEST_SECRET||process.env.MCP_JOB_SCHEDULE_CONFIGURED==='true'}; }); } export async function tickMcpJobs(ctx:PurchasingContext,options:{limit:number;deadlineMs:number}):Promise<{claimed:number;completed:number;failed:number}>{