prototype/src/workflow.mjs

import { createStep, createWorkflow } from '@mastra/core/workflows';
import { z } from 'zod';
import { randomUUID } from 'node:crypto';
import { normalizeIssues,suggest,THRESHOLDS,digest } from './core.mjs';
import { embedTexts,issueText,referenceModel,MODEL } from './model.mjs';
const issue=z.object({id:z.number().int().positive(),title:z.string(),body:z.string(),labels:z.array(z.string())});
const validated=z.object({issues:z.array(issue),started:z.number(),sourceHash:z.string()});
const embedded=validated.extend({vectors:z.array(z.array(z.number()))});
const classified=validated.extend({items:z.array(z.any())});
const resultSchema=z.object({runId:z.string(),items:z.array(z.any()),summary:z.object({total:z.number(),ready:z.number(),review:z.number(),duplicateCandidates:z.number()}),sourceHash:z.string(),model:z.any(),thresholds:z.any(),timings:z.object({totalMs:z.number()}),workflowSteps:z.array(z.string())});
export const STEPS=['validate-export','embed-locally','compare-and-abstain','build-review-queue'];
const validate=createStep({id:STEPS[0],inputSchema:z.object({issues:z.array(z.any())}),outputSchema:validated,execute:async({inputData})=>{
  const issues=normalizeIssues(inputData);return {issues,started:performance.now(),sourceHash:digest(issues)};
}});
const embed=createStep({id:STEPS[1],inputSchema:validated,outputSchema:embedded,execute:async({inputData})=>({...inputData,vectors:await embedTexts(inputData.issues.map(issueText))})});
const classify=createStep({id:STEPS[2],inputSchema:embedded,outputSchema:classified,execute:async({inputData})=>{
  const {references,referenceVectors}=await referenceModel();
  const {vectors,...rest}=inputData;return {...rest,items:suggest(inputData.issues,vectors,references,referenceVectors)};
}});
const review=createStep({id:STEPS[3],inputSchema:classified,outputSchema:resultSchema,execute:async({inputData})=>{
  const items=inputData.items,review=items.filter(x=>x.needsReview).length;
  return {runId:randomUUID(),items,summary:{total:items.length,ready:items.length-review,review,duplicateCandidates:items.filter(x=>x.duplicateOf!==null).length},sourceHash:inputData.sourceHash,model:MODEL,thresholds:THRESHOLDS,timings:{totalMs:Math.round(performance.now()-inputData.started)},workflowSteps:STEPS};
}});
export const triageWorkflow=createWorkflow({id:'slow-triage-local-issues',inputSchema:z.object({issues:z.array(z.any())}),outputSchema:resultSchema}).then(validate).then(embed).then(classify).then(review).commit();
export async function runTriage(input){
  const issues=normalizeIssues(input);
  const run=await triageWorkflow.createRun();
  const result=await run.start({inputData:{issues}});
  if(result.status!=='success')throw new Error(result.error?.message || `Workflow did not complete: ${result.status}`);
  return {...result.result,execution:{status:result.status,steps:Object.entries(result.steps??{}).map(([id,value])=>({id,status:value.status}))}};
}