fix(dn): the n8n tick also runs queued agent batch jobs

Co-Authored-By: MiniMax Code <noreply@minimax.io>

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());

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<Permission>(["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<ReturnType<typeof pbAdmin>>, id: string): Promise<boolean> {

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});
+});

src/lib/purchasing/jobs.ts

@@ -13,7 +13,10 @@ export async function enqueueDeliveryJob(ctx:PurchasingContext,input:Record<stri
  const operation=operationRecordId(ctx.principal.id,'process_delivery_batch',args.operationId),jobId=hash('sha256',operation+':job').slice(0,15);
  return runOperation(ctx,'process_delivery_batch',args,async lease=>{
   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}>{