feat: run durable MCP processing jobs in the existing app

src/app/api/mcp/jobs/tick/route.test.ts

@@ -0,0 +1,17 @@
+import {beforeEach,afterEach,expect,it,vi} from 'vitest';
+import {hash} from 'node:crypto';
+const context=vi.hoisted(()=>vi.fn());
+vi.mock('@/lib/purchasing/client',()=>({createPurchasingContext:context}));
+vi.mock('@/lib/purchasing/jobs',()=>({tickMcpJobs:vi.fn(async()=>({claimed:0,completed:0,failed:0}))}));
+import {POST} from './route';
+const token='a-dedicated-scheduler-test-token';
+beforeEach(()=>{context.mockReset();context.mockResolvedValue({});vi.stubEnv('MCP_PUBLIC_URL','https://mega.test/api/mcp');});afterEach(()=>vi.unstubAllEnvs());
+it('denies browser write credentials before constructing the database client',async()=>{
+ vi.stubEnv('MCP_CLIENTS_JSON',JSON.stringify([{id:'writer',tokenHash:hash('sha256',token),permissions:['read','write']}]));
+ expect((await POST(new Request('https://mega.test/api/mcp/jobs/tick',{method:'POST',headers:{Authorization:`Bearer ${token}`}}))).status).toBe(403);
+ expect(context).not.toHaveBeenCalled();
+});
+it('accepts a dedicated processing credential',async()=>{
+ vi.stubEnv('MCP_CLIENTS_JSON',JSON.stringify([{id:'scheduler',tokenHash:hash('sha256',token),permissions:['process']}]));
+ expect((await POST(new Request('https://mega.test/api/mcp/jobs/tick',{method:'POST',headers:{Authorization:`Bearer ${token}`}}))).status).toBe(200);
+});

src/app/api/mcp/jobs/tick/route.ts

@@ -0,0 +1,12 @@
+import {authenticateMcp,requirePermission,McpHttpError} from '@/mcp/auth';
+import {createPurchasingContext} from '@/lib/purchasing/client';
+import {tickMcpJobs} from '@/lib/purchasing/jobs';
+import {PurchasingError} from '@/lib/purchasing/errors';
+export const runtime='nodejs';export const dynamic='force-dynamic';export const maxDuration=600;
+export async function POST(request:Request):Promise<Response>{
+ try{
+  const principal=authenticateMcp(request);requirePermission(principal,'process');
+  const result=await tickMcpJobs(await createPurchasingContext(principal),{limit:1,deadlineMs:240000});
+  return Response.json(result,{headers:{'Cache-Control':'no-store'}});
+ }catch(error){return Response.json({error:error instanceof PurchasingError||error instanceof McpHttpError?error.message:'job_tick_failed'},{status:error instanceof PurchasingError||error instanceof McpHttpError?error.status:503,headers:{'Cache-Control':'no-store'}});}
+}

src/lib/dn-pipeline/run-staged.ts

@@ -56,14 +56,14 @@ type ExistingCandidate = {
  * Run the non-financial stages for one batch. A successful run only produces
  * candidates; Task 8 is the sole owner of delivery-note/ledger promotion.
  */
-export async function runStagedBatch(pb: PocketBase, batchId: string, options: { freshPages?: readonly number[] } = {}): Promise<StagedRunResult> {
+export async function runStagedBatch(pb: PocketBase, batchId: string, options: { freshPages?: readonly number[]; deadlineAt?:number } = {}): Promise<StagedRunResult> {
   const lease = await withBatchLease(pb, batchId, "staged-runner", async (ownership) => runClaimedStagedBatch(pb, batchId, ownership, options));
   return lease.acquired
     ? lease.value
     : { status: "needs_attention", notesDetected: 0, notesPromoted: 0, pagesRead: 0, pagesFailed: 0, checks: [], notClaimed: true };
 }
 
-async function runClaimedStagedBatch(pb: PocketBase, batchId: string, ownership: BatchLeaseContext, options: { freshPages?: readonly number[] }): Promise<StagedRunResult> {
+async function runClaimedStagedBatch(pb: PocketBase, batchId: string, ownership: BatchLeaseContext, options: { freshPages?: readonly number[]; deadlineAt?:number }): Promise<StagedRunResult> {
   const batch = (await pb.collection("dn_batches").getOne(batchId)) as unknown as Batch;
   await ownership.update("dn_batches", batchId, {
     status: "analysing", stage: "preparing", claimed_at: new Date().toISOString(), last_error: "",
@@ -95,6 +95,7 @@ async function runClaimedStagedBatch(pb: PocketBase, batchId: string, ownership:
   let pageOffset = 0;
 
   for (const source of sources) {
+    assertStagedDeadline(options.deadlineAt);
     let normalised: NormalisedPage[];
     try {
       const original = await fetchSourceBytes(pb, source);
@@ -109,6 +110,7 @@ async function runClaimedStagedBatch(pb: PocketBase, batchId: string, ownership:
     }
 
     for (const image of normalised) {
+      assertStagedDeadline(options.deadlineAt);
       const page = pageOffset + image.sourcePage;
       pageCount += 1;
       if (source.delivery_session_id) pageDeliverySessionIds[page] = source.delivery_session_id;
@@ -180,6 +182,7 @@ async function runClaimedStagedBatch(pb: PocketBase, batchId: string, ownership:
   let review = 0;
 
   for (const [index, group] of assembly.groups.entries()) {
+    assertStagedDeadline(options.deadlineAt);
     const groupReads = group.pages.map((page) => reads.find((read) => read.page === page)).filter((read): read is PageRead => Boolean(read));
     const first = groupReads[0];
     if (!first) continue;
@@ -911,3 +914,7 @@ function filter(pb: PocketBase, raw: string, params: Record<string, unknown>): s
   if (typeof client.filter === "function") return client.filter(raw, params);
   return raw.replace(/\{:(\w+)\}/g, (_match, key: string) => JSON.stringify(params[key]));
 }
+
+export function assertStagedDeadline(deadlineAt?:number):void {
+ if(deadlineAt!==undefined&&Date.now()>=deadlineAt)throw new Error("staged_processing_deadline");
+}

src/lib/purchasing/client.ts

@@ -25,6 +25,7 @@ export function assertAllowedMutation(collection:string,method:string,data?:unkn
 }
 export function guardPurchasingClient(pb:PocketBase,principal:Principal):void {
  const cleanup=new Set<string>();
+ const submittedLocks=new Set<string>();
  const recordPath=(url:string)=>/^\/api\/collections\/([^/]+)\/records(?:\/([^/]+))?$/.exec(decodeURIComponent(new URL(url,pb.baseURL).pathname));
  pb.beforeSend=(url,options)=>{
   const method=(options.method||'GET').toUpperCase();
@@ -35,6 +36,7 @@ export function guardPurchasingClient(pb:PocketBase,principal:Principal):void {
   const validate=(suburl:string,verb:string,body?:unknown)=>{
    const record=recordPath(suburl);if(!record)throw new PurchasingError('collection_write_forbidden',403);
    assertAllowedMutation(record[1],verb,body,cleanup.has(`${record[1]}:${record[2]}`));
+   if(LOCKS.has(record[1])&&verb.toUpperCase()==='POST'){const payload=objectBody(body);if(typeof payload.id==='string')submittedLocks.add(`${record[1]}:${payload.id}`);}
   };
   if(path==='/api/batch'){
    const body=objectBody(options.body);const requests=body.requests;
@@ -53,7 +55,7 @@ export function guardPurchasingClient(pb:PocketBase,principal:Principal):void {
    const rows=Array.isArray(data?.items)?data.items:[data];
    for(const row of rows){
     if(!row||typeof row.id!=='string')continue;
-    if(row.worker===principal.id||String(row.worker??'').startsWith(principal.id+':')||row.owner===principal.id||String(row.owner??'').startsWith(principal.id+':')||Date.parse(row.expires_at)<=Date.now())cleanup.add(`${record[1]}:${row.id}`);
+    if(submittedLocks.has(`${record[1]}:${row.id}`)||row.worker===principal.id||String(row.worker??'').startsWith(principal.id+':')||row.owner===principal.id||String(row.owner??'').startsWith(principal.id+':')||Date.parse(row.expires_at)<=Date.now())cleanup.add(`${record[1]}:${row.id}`);
    }
   }
   return data;

src/lib/purchasing/jobs.test.ts

@@ -0,0 +1,31 @@
+import {beforeEach,afterEach,expect,it,vi} from 'vitest';
+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());
+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);
+ expect(canReclaimJob('invalid',Date.now())).toBe(false);
+});
+it('enqueues once and allows only a dedicated scheduler to process it',async()=>{
+ const {pb,store}=memoryDatabase();await pb.collection('dn_batches').create({id:'abcdefghijklmn1'});
+ const writer={pb,principal:{id:'writer',permissions:new Set<'read'|'write'>(['read','write'])}};
+ const input={operationId:'83f33948-63c7-4a02-a45a-9ab18d3bdc02',batchId:'abcdefghijklmn1'};
+ expect(await enqueueDeliveryJob(writer,input)).toEqual(await enqueueDeliveryJob(writer,input));expect(store('mcp_jobs').size).toBe(1);
+ await expect(tickMcpJobs(writer,{limit:1,deadlineMs:1000})).rejects.toThrow('permission_denied');
+ const scheduler={pb,principal:{id:'scheduler',permissions:new Set<'process'>(['process'])}};
+ expect(await tickMcpJobs(scheduler,{limit:1,deadlineMs:1000})).toEqual({claimed:1,completed:1,failed:0});
+ expect(staged).toHaveBeenCalledTimes(1);expect([...store('mcp_jobs').values()][0].status).toBe('completed');
+});
+it('retries failed processing without creating another job or receipt',async()=>{
+ const {pb,store}=memoryDatabase();await pb.collection('dn_batches').create({id:'abcdefghijklmn1'});
+ const writer={pb,principal:{id:'writer',permissions:new Set<'read'|'write'>(['read','write'])}};
+ await enqueueDeliveryJob(writer,{operationId:'83f33948-63c7-4a02-a45a-9ab18d3bdc02',batchId:'abcdefghijklmn1'});
+ staged.mockRejectedValueOnce(Error('provider unavailable'));
+ const scheduler={pb,principal:{id:'scheduler',permissions:new Set<'process'>(['process'])}};
+ expect((await tickMcpJobs(scheduler,{limit:1,deadlineMs:1000})).failed).toBe(1);
+ expect([...store('mcp_jobs').values()][0].status).toBe('queued');
+ expect(store('mcp_jobs').size).toBe(1);expect(store('delivery_notes').size).toBe(0);
+});

src/lib/purchasing/jobs.ts

@@ -0,0 +1,53 @@
+import {z} from 'zod';
+import {hash} from 'node:crypto';
+import type {PurchasingContext,OperationResult} from './types';
+import {MutationSchema,runOperation,operationRecordId} from './operations';
+import {RecordId} from './documents';
+import {withResourceLease} from './resource-lease';
+import {runStagedBatch} from '@/lib/dn-pipeline/run-staged';
+import {PurchasingError} from './errors';
+export const ProcessBatchSchema=MutationSchema.extend({batchId:RecordId}).strict();
+export function canReclaimJob(expiresAt:string,now:number):boolean{const time=Date.parse(expiresAt);return Number.isFinite(time)&&time<=now;}
+export async function enqueueDeliveryJob(ctx:PurchasingContext,input:Record<string,unknown>):Promise<OperationResult>{
+ const args=ProcessBatchSchema.parse(input);await ctx.pb.collection('dn_batches').getOne(args.batchId,{fields:'id'});
+ 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'};
+ });
+}
+export async function tickMcpJobs(ctx:PurchasingContext,options:{limit:number;deadlineMs:number}):Promise<{claimed:number;completed:number;failed:number}>{
+ if(!ctx.principal.permissions.has('process'))throw new PurchasingError('permission_denied',403);
+ if(!process.env.MINIMAX_API_KEY)throw new PurchasingError('processing_setup_required',503);
+ const limit=z.number().int().min(1).max(3).parse(options.limit),deadlineMs=z.number().int().min(1).max(240000).parse(options.deadlineMs);
+ const deadlineAt=Date.now()+deadlineMs;
+ const jobs=await ctx.pb.collection('mcp_jobs').getList(1,limit,{filter:ctx.pb.filter('(status="queued" || status="running") && attempts<3 && (next_attempt_at="" || next_attempt_at<={:now})',{now:new Date().toISOString()}),sort:'next_attempt_at,id'});
+ const counts={claimed:0,completed:0,failed:0};
+ for(const job of jobs.items){
+  if(Date.now()>=deadlineAt)break;
+  const result=await withResourceLease(ctx.pb,`job:${job.id}`,ctx.principal.id,async lease=>{
+   const current=await ctx.pb.collection('mcp_jobs').getOne(job.id);
+   if(!['queued','running'].includes(current.status)||Number(current.attempts)>=3||current.next_attempt_at&&Date.parse(current.next_attempt_at)>Date.now())return;
+   counts.claimed++;const attempts=Number(current.attempts??0)+1;
+   await lease.update('mcp_jobs',job.id,{status:'running',attempts});
+   try{
+    const summary=await runStagedBatch(ctx.pb,current.batch,{deadlineAt});
+    if(summary.notClaimed)throw new PurchasingError('batch_locked',409);
+    if(summary.pagesFailed>0||summary.checks?.some(check=>check.id==='sources_present'&&check.passed===false))throw new PurchasingError('batch_processing_incomplete',422);
+    await lease.batch([
+     {collection:'mcp_jobs',method:'update',id:job.id,data:{status:'completed',error_code:''}},
+     {collection:'mcp_operations',method:'update',id:current.operation,data:{result:{jobId:job.id,status:'completed',summary}}},
+    ]);counts.completed++;
+   }catch(error){
+    const status=attempts>=3?'failed':'queued';
+    const errorCode=error instanceof PurchasingError?error.code:'batch_processing_failed';
+    await lease.batch([
+     {collection:'mcp_jobs',method:'update',id:job.id,data:{status,next_attempt_at:new Date(Date.now()+Math.min(300000,30000*2**(attempts-1))).toISOString(),error_code:errorCode}},
+     {collection:'mcp_operations',method:'update',id:current.operation,data:{result:{jobId:job.id,status,attempts,error:errorCode}}},
+    ]);counts.failed++;
+   }
+  });
+  if(!result.acquired)continue;
+ }
+ return counts;
+}

src/lib/purchasing/resource-lease.test.ts

@@ -14,3 +14,9 @@ it('reports a live competing lease without running mutations',async()=>{
  const pb={collection:()=>({create:vi.fn(async()=>{throw Error('unique');}),getOne:vi.fn(),getFullList:vi.fn(async()=>[{id:'other',owner:'other',expires_at:new Date(Date.now()+60000).toISOString()}]),update:vi.fn(),delete:vi.fn()}),createBatch:vi.fn()} as unknown as PocketBase;
  expect(await withResourceLease(pb,'po:one','agent',work)).toEqual({acquired:false});expect(work).not.toHaveBeenCalled();
 });
+import {memoryDatabase} from './test-support';
+it('reclaims an expired acquisition after restart and releases only the new owner',async()=>{
+ const {pb,store}=memoryDatabase();await pb.collection('mcp_resource_locks').create({id:'stale',resource_key:'job:one',owner:'old',expires_at:new Date(Date.now()-1000).toISOString()});
+ expect(await withResourceLease(pb,'job:one','scheduler',async lease=>{await lease.assertOwned();return 'resumed';})).toEqual({acquired:true,value:'resumed'});
+ expect(store('mcp_resource_locks').size).toBe(0);
+});

src/lib/purchasing/test-support.ts

@@ -11,6 +11,7 @@ export function memoryDatabase(){
    const record={...input,id};store(collection).set(id,record);return record;
   },
   getOne:async(id:string)=>{const row=store(collection).get(id);if(!row)throw Object.assign(Error('not found'),{status:404});return {...row};},
+  getList:async(page:number,perPage:number)=>({page,perPage,totalItems:store(collection).size,totalPages:Math.ceil(store(collection).size/perPage),items:[...store(collection).values()].slice((page-1)*perPage,page*perPage)}),
   getFullList:async()=>[...store(collection).values()].map(r=>({...r})),
   update:async(id:string,input:Record<string,unknown>)=>{const row=await api(collection).getOne(id);const next={...row,...input};store(collection).set(id,next);return next;},
   delete:async(id:string)=>{store(collection).delete(id);},

src/mcp/purchasing-write-tools.ts

@@ -5,6 +5,7 @@ import {CreatePurchaseOrderSchema,UpdatePurchaseOrderSchema,AddPurchaseOrderItem
 import {SupplierGuidanceSchema,mutateSupplierGuidance} from '@/lib/purchasing/supplier-guidance';
 import {CandidateOperationSchema,EntityResolutionSchema,ItemResolutionSchema,resolveDelivery,processDeliveryCandidate} from '@/lib/purchasing/delivery-matching';
 import {McpPostDeliveryLineSchema,ReceiptCorrectionSchema,postDeliveryLine,correctDeliveryReceipt} from '@/lib/purchasing/delivery-receipts';
+import {ProcessBatchSchema,enqueueDeliveryJob} from '@/lib/purchasing/jobs';
 import {setDeliveryVerification} from '@/lib/purchasing/verification';
 import {UploadPurchasingSchema,uploadPurchasingDocument} from '@/lib/purchasing/uploads';
 import {CreateDeliveryCandidateSchema,UpdateDeliveryCandidateSchema,DeliveryGroupingSchema,mutateDeliveryCandidate} from '@/lib/purchasing/delivery-candidates';
@@ -13,6 +14,7 @@ export function registerPurchasingWrites(server:McpServer,principal:Principal):v
  for(const [name,schema] of [['create_delivery_candidate',CreateDeliveryCandidateSchema],['update_delivery_candidate',UpdateDeliveryCandidateSchema],['set_delivery_grouping',DeliveryGroupingSchema]] as const)registerPurchasingTool(server,principal,{name,description:'Edit receiving candidates or grouping under the batch lease, preserving posted evidence.',schema,permission:'write',handler:(ctx,input)=>mutateDeliveryCandidate(ctx,name,input)});
  for(const [name,schema] of [['resolve_delivery_entities',EntityResolutionSchema],['resolve_delivery_item',ItemResolutionSchema]] as const)registerPurchasingTool(server,principal,{name,description:'Resolve receiving entities or PO lines with ownership checks; does not book a receipt.',schema,permission:'write',handler:(ctx,input)=>resolveDelivery(ctx,name,input)});
  for(const [name,description,schema,handler] of [['process_delivery_candidate','Generate a fresh matching preview without posting.',CandidateOperationSchema,processDeliveryCandidate],['post_delivery_line','Post one allocation with explicit confirmation and a current preview hash.',McpPostDeliveryLineSchema,postDeliveryLine],['correct_delivery_receipt','Correct an eligible legacy receipt with expected quantities and audit history.',ReceiptCorrectionSchema,correctDeliveryReceipt]] as const)registerPurchasingTool(server,principal,{name,description,schema,permission:'write',handler});
+ registerPurchasingTool(server,principal,{name:'process_delivery_batch',description:'Queue durable source reading and matching; no automatic receipt posting. Poll get_mcp_operation.',schema:ProcessBatchSchema,permission:'write',handler:enqueueDeliveryJob});
  const orderTools=[
   ['create_purchase_order','Create a manual order and validated lines atomically.',CreatePurchaseOrderSchema],
   ['update_purchase_order','Update permitted header fields of an unverified order.',UpdatePurchaseOrderSchema],