From 9f0698f1d15fcb372154c2c7325b65fee165605e Mon Sep 17 00:00:00 2001 From: DEls Date: Mon, 5 Oct 2026 02:04:15 +0200 Subject: [PATCH] feat: run durable MCP processing jobs in the existing app --- src/app/api/mcp/jobs/tick/route.test.ts | 17 ++++++++ src/app/api/mcp/jobs/tick/route.ts | 12 +++++ src/lib/dn-pipeline/run-staged.ts | 11 ++++- src/lib/purchasing/client.ts | 4 +- src/lib/purchasing/jobs.test.ts | 31 +++++++++++++ src/lib/purchasing/jobs.ts | 53 +++++++++++++++++++++++ src/lib/purchasing/resource-lease.test.ts | 6 +++ src/lib/purchasing/test-support.ts | 1 + src/mcp/purchasing-write-tools.ts | 2 + 9 files changed, 134 insertions(+), 3 deletions(-) create mode 100644 src/app/api/mcp/jobs/tick/route.test.ts create mode 100644 src/app/api/mcp/jobs/tick/route.ts create mode 100644 src/lib/purchasing/jobs.test.ts create mode 100644 src/lib/purchasing/jobs.ts diff --git a/src/app/api/mcp/jobs/tick/route.test.ts b/src/app/api/mcp/jobs/tick/route.test.ts new file mode 100644 index 0000000..a819926 --- /dev/null +++ b/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); +}); diff --git a/src/app/api/mcp/jobs/tick/route.ts b/src/app/api/mcp/jobs/tick/route.ts new file mode 100644 index 0000000..580de45 --- /dev/null +++ b/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{ + 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'}});} +} diff --git a/src/lib/dn-pipeline/run-staged.ts b/src/lib/dn-pipeline/run-staged.ts index eb4591e..c6f786b 100644 --- a/src/lib/dn-pipeline/run-staged.ts +++ b/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 { +export async function runStagedBatch(pb: PocketBase, batchId: string, options: { freshPages?: readonly number[]; deadlineAt?:number } = {}): Promise { 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 { +async function runClaimedStagedBatch(pb: PocketBase, batchId: string, ownership: BatchLeaseContext, options: { freshPages?: readonly number[]; deadlineAt?:number }): Promise { 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): 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"); +} diff --git a/src/lib/purchasing/client.ts b/src/lib/purchasing/client.ts index bc3afdc..5e1ea27 100644 --- a/src/lib/purchasing/client.ts +++ b/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(); + const submittedLocks=new Set(); 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; diff --git a/src/lib/purchasing/jobs.test.ts b/src/lib/purchasing/jobs.test.ts new file mode 100644 index 0000000..8b7bd81 --- /dev/null +++ b/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); +}); diff --git a/src/lib/purchasing/jobs.ts b/src/lib/purchasing/jobs.ts new file mode 100644 index 0000000..4533778 --- /dev/null +++ b/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):Promise{ + 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; +} diff --git a/src/lib/purchasing/resource-lease.test.ts b/src/lib/purchasing/resource-lease.test.ts index a9fe75b..21099b7 100644 --- a/src/lib/purchasing/resource-lease.test.ts +++ b/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); +}); diff --git a/src/lib/purchasing/test-support.ts b/src/lib/purchasing/test-support.ts index e2e2bc1..2b970d7 100644 --- a/src/lib/purchasing/test-support.ts +++ b/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)=>{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);}, diff --git a/src/mcp/purchasing-write-tools.ts b/src/mcp/purchasing-write-tools.ts index 8f95d29..d5fad66 100644 --- a/src/mcp/purchasing-write-tools.ts +++ b/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],