{"version":3,"file":"chain-append.d.ts","sourceRoot":"","sources":["../../../../src/runs/background/chain-append.ts"],"names":[],"mappings":"AAKA,OAAO,KAAK,EAAE,WAAW,EAA4C,MAAM,uBAAuB,CAAC;AAEnG,OAAO,KAAK,EAIX,UAAU,EAEV,MAAM,6BAA6B,CAAC;AAKrC,MAAM,WAAW,kBAAkB;IAClC,EAAE,EAAE,MAAM,CAAC;IACX,SAAS,EAAE,MAAM,CAAC;IAClB,KAAK,EAAE,UAAU,EAAE,CAAC;CACpB;AAED,MAAM,WAAW,iBAAiB;IACjC,OAAO,EAAE,kBAAkB,CAAC;IAC5B,YAAY,EAAE,MAAM,CAAC;CACrB;AA0BD,wBAAgB,+BAA+B,CAAC,QAAQ,EAAE,MAAM,GAAG,MAAM,CAExE;AAED,wBAAgB,qBAAqB,CAAC,KAAK,EAAE,UAAU,EAAE,GAAG,MAAM,EAAE,CAanE;AAED,wBAAgB,yBAAyB,CAAC,KAAK,EAAE;IAChD,QAAQ,EAAE,MAAM,CAAC;IACjB,KAAK,EAAE,MAAM,CAAC;IACd,KAAK,EAAE,UAAU,EAAE,CAAC;IACpB,GAAG,CAAC,EAAE,MAAM,CAAC;CACb,GAAG,iBAAiB,CAyCpB;AAWD,wBAAgB,8BAA8B,CAAC,QAAQ,EAAE,MAAM,GAAG,kBAAkB,EAAE,CAKrF;AAED,wBAAgB,0BAA0B,CAAC,QAAQ,EAAE,MAAM,GAAG,kBAAkB,EAAE,CAYjF;AAID,+FAA+F;AAC/F,wBAAgB,qBAAqB,CAAC,IAAI,EAAE,MAAM,GAAG,SAAS,GAAG,MAAM,GAAG,SAAS,CAMlF;AAuLD,wBAAgB,yBAAyB,CAAC,KAAK,EAAE;IAChD,MAAM,EAAE,WAAW,CAAC;IACpB,KAAK,EAAE,UAAU,EAAE,CAAC;IACpB,GAAG,CAAC,EAAE,MAAM,CAAC;IACb,cAAc,CAAC,EAAE,MAAM,CAAC;CACxB,GAAG;IAAE,eAAe,EAAE,MAAM,CAAC;IAAC,cAAc,EAAE,MAAM,CAAA;CAAE,CAwBtD","sourcesContent":["import { randomUUID } from \"node:crypto\";\nimport * as fs from \"node:fs\";\nimport * as path from \"node:path\";\nimport { appendJsonl } from \"../../shared/artifacts.ts\";\nimport { writeAtomicJson } from \"../../shared/atomic-json.ts\";\nimport type { AsyncStatus, WorkflowGraphNode, WorkflowGraphSnapshot } from \"../../shared/types.ts\";\nimport { readStatus } from \"../../shared/utils.ts\";\nimport type {\n\tDynamicRunnerGroup,\n\tParallelStepGroup,\n\tRunnerCheckpointStep,\n\tRunnerStep,\n\tRunnerSubagentStep,\n} from \"../shared/parallel-utils.ts\";\nimport { isCheckpointRunnerStep, isDynamicRunnerGroup, isParallelGroup } from \"../shared/parallel-utils.ts\";\n\nconst APPEND_REQUESTS_DIR = \"append-requests\";\n\nexport interface ChainAppendRequest {\n\tid: string;\n\tcreatedAt: number;\n\tsteps: RunnerStep[];\n}\n\nexport interface ChainAppendResult {\n\trequest: ChainAppendRequest;\n\tpendingCount: number;\n}\n\ntype StatusStep = NonNullable<AsyncStatus[\"steps\"]>[number];\n\nfunction appendDir(asyncDir: string): string {\n\treturn path.join(asyncDir, APPEND_REQUESTS_DIR);\n}\n\nfunction appendRequestPath(asyncDir: string, request: ChainAppendRequest): string {\n\treturn path.join(appendDir(asyncDir), `${request.createdAt}-${request.id}.json`);\n}\n\nfunction listAppendRequestFiles(asyncDir: string): string[] {\n\tconst dir = appendDir(asyncDir);\n\ttry {\n\t\treturn fs\n\t\t\t.readdirSync(dir)\n\t\t\t.filter((entry) => entry.endsWith(\".json\"))\n\t\t\t.map((entry) => path.join(dir, entry))\n\t\t\t.sort();\n\t} catch (error) {\n\t\tif ((error as NodeJS.ErrnoException).code === \"ENOENT\") return [];\n\t\tthrow error;\n\t}\n}\n\nexport function countPendingChainAppendRequests(asyncDir: string): number {\n\treturn listAppendRequestFiles(asyncDir).length;\n}\n\nexport function runnerStepOutputNames(steps: RunnerStep[]): string[] {\n\tconst names: string[] = [];\n\tfor (const step of steps) {\n\t\tif (isCheckpointRunnerStep(step)) {\n\t\t} else if (isParallelGroup(step)) {\n\t\t\tnames.push(...step.parallel.map((task) => task.outputName).filter((name): name is string => Boolean(name)));\n\t\t} else if (isDynamicRunnerGroup(step)) {\n\t\t\tif (step.collect.as) names.push(step.collect.as);\n\t\t} else if (step.outputName) {\n\t\t\tnames.push(step.outputName);\n\t\t}\n\t}\n\treturn names;\n}\n\nexport function enqueueChainAppendRequest(input: {\n\tasyncDir: string;\n\trunId: string;\n\tsteps: RunnerStep[];\n\tnow?: number;\n}): ChainAppendResult {\n\tconst status = readStatus(input.asyncDir);\n\tif (!status) throw new Error(`No async run status found for '${input.runId}'.`);\n\tif (status.runId !== input.runId)\n\t\tthrow new Error(`Async run id mismatch: expected '${input.runId}', found '${status.runId}'.`);\n\tif (status.mode !== \"chain\")\n\t\tthrow new Error(`Run '${input.runId}' is ${status.mode}; only active chain runs accept appended steps.`);\n\tif (status.state !== \"running\")\n\t\tthrow new Error(`Run '${input.runId}' is ${status.state}; only running chain runs accept appended steps.`);\n\tconst stillInProgress =\n\t\t(status.steps ?? []).some((step) => step.status === \"running\" || step.status === \"pending\") ||\n\t\t(status.pendingAppends ?? 0) > 0;\n\tif (!stillInProgress)\n\t\tthrow new Error(\n\t\t\t`Run '${input.runId}' has no running or pending chain steps left; append-step must target an in-progress chain.`,\n\t\t);\n\tif (input.steps.length === 0) throw new Error(\"append-step requires one chain step.\");\n\n\tconst request: ChainAppendRequest = {\n\t\tid: randomUUID(),\n\t\tcreatedAt: input.now ?? Date.now(),\n\t\tsteps: input.steps,\n\t};\n\tfs.mkdirSync(appendDir(input.asyncDir), { recursive: true });\n\twriteAtomicJson(appendRequestPath(input.asyncDir, request), request);\n\tconst pendingCount = countPendingChainAppendRequests(input.asyncDir);\n\tconst statusPath = path.join(input.asyncDir, \"status.json\");\n\tconst updatedStatus = { ...status, pendingAppends: pendingCount, lastUpdate: request.createdAt };\n\twriteAtomicJson(statusPath, updatedStatus);\n\tappendJsonl(\n\t\tpath.join(input.asyncDir, \"events.jsonl\"),\n\t\tJSON.stringify({\n\t\t\ttype: \"subagent.chain.append.requested\",\n\t\t\tts: request.createdAt,\n\t\t\trunId: input.runId,\n\t\t\trequestId: request.id,\n\t\t\tstepCount: input.steps.length,\n\t\t\tpendingAppends: pendingCount,\n\t\t}),\n\t);\n\treturn { request, pendingCount };\n}\n\nfunction readAppendRequest(filePath: string): ChainAppendRequest | undefined {\n\tconst raw = JSON.parse(fs.readFileSync(filePath, \"utf-8\")) as Partial<ChainAppendRequest>;\n\tif (!raw.id || typeof raw.id !== \"string\") return undefined;\n\tconst createdAt = raw.createdAt;\n\tif (typeof createdAt !== \"number\" || !Number.isFinite(createdAt)) return undefined;\n\tif (!Array.isArray(raw.steps) || raw.steps.length === 0) return undefined;\n\treturn { id: raw.id, createdAt, steps: raw.steps as RunnerStep[] };\n}\n\nexport function readPendingChainAppendRequests(asyncDir: string): ChainAppendRequest[] {\n\treturn listAppendRequestFiles(asyncDir)\n\t\t.map((filePath) => readAppendRequest(filePath))\n\t\t.filter((request): request is ChainAppendRequest => Boolean(request))\n\t\t.sort((left, right) => left.createdAt - right.createdAt || left.id.localeCompare(right.id));\n}\n\nexport function consumeChainAppendRequests(asyncDir: string): ChainAppendRequest[] {\n\tconst requests: ChainAppendRequest[] = [];\n\tfor (const filePath of listAppendRequestFiles(asyncDir)) {\n\t\tconst request = readAppendRequest(filePath);\n\t\ttry {\n\t\t\tfs.unlinkSync(filePath);\n\t\t} catch {\n\t\t\t// The runner should not execute a consumed request twice.\n\t\t}\n\t\tif (request) requests.push(request);\n\t}\n\treturn requests.sort((left, right) => left.createdAt - right.createdAt || left.id.localeCompare(right.id));\n}\n\nconst MAX_STATUS_STEP_DESCRIPTION_CHARS = 160;\n\n/** Bounded one-line per-step task description persisted into status.json for fleet display. */\nexport function statusStepDescription(task: string | undefined): string | undefined {\n\tconst description = task?.replace(/\\s+/g, \" \").trim();\n\tif (!description) return undefined;\n\treturn description.length > MAX_STATUS_STEP_DESCRIPTION_CHARS\n\t\t? `${description.slice(0, MAX_STATUS_STEP_DESCRIPTION_CHARS - 1)}…`\n\t\t: description;\n}\n\nfunction statusStepForTask(task: RunnerSubagentStep): StatusStep {\n\tconst description = statusStepDescription(task.task);\n\treturn {\n\t\tagent: task.agent,\n\t\t...(description ? { description } : {}),\n\t\t...(task.context ? { context: task.context } : {}),\n\t\tphase: task.phase,\n\t\tlabel: task.label,\n\t\toutputName: task.outputName,\n\t\tstructured: task.structured,\n\t\tstatus: \"pending\",\n\t\t...(task.sessionFile ? { sessionFile: task.sessionFile } : {}),\n\t\tskills: task.skills,\n\t\tmodel: task.model,\n\t\tthinking: task.thinking,\n\t\tattemptedModels:\n\t\t\ttask.modelCandidates && task.modelCandidates.length > 0\n\t\t\t\t? task.modelCandidates\n\t\t\t\t: task.model\n\t\t\t\t\t? [task.model]\n\t\t\t\t\t: undefined,\n\t\trecentTools: [],\n\t\trecentOutput: [],\n\t};\n}\n\nfunction statusStepForCheckpoint(step: RunnerCheckpointStep, stepIndex: number): StatusStep {\n\treturn {\n\t\tagent: `checkpoint:${step.checkpoint}`,\n\t\tphase: step.phase,\n\t\tlabel: step.label ?? step.checkpoint,\n\t\tstatus: \"pending\",\n\t\tcheckpoint: {\n\t\t\tname: step.checkpoint,\n\t\t\t...(step.message ? { message: step.message } : {}),\n\t\t\tstatus: \"pending\",\n\t\t\tstepIndex,\n\t\t},\n\t\trecentTools: [],\n\t\trecentOutput: [],\n\t};\n}\n\nfunction statusStepsForRunnerStep(step: RunnerStep, stepIndex: number): StatusStep[] {\n\tif (isCheckpointRunnerStep(step)) return [statusStepForCheckpoint(step, stepIndex)];\n\tif (isParallelGroup(step)) return step.parallel.map(statusStepForTask);\n\tif (isDynamicRunnerGroup(step)) {\n\t\treturn [\n\t\t\t{\n\t\t\t\tagent: `expand:${step.parallel.agent}`,\n\t\t\t\t...(step.parallel.context ? { context: step.parallel.context } : {}),\n\t\t\t\tphase: step.phase ?? step.parallel.phase,\n\t\t\t\tlabel: step.label ?? step.parallel.label ?? `Dynamic fanout (${step.collect.as})`,\n\t\t\t\toutputName: step.collect.as,\n\t\t\t\tstructured: Boolean(step.collect.outputSchema),\n\t\t\t\tstatus: \"pending\",\n\t\t\t\trecentTools: [],\n\t\t\t\trecentOutput: [],\n\t\t\t},\n\t\t];\n\t}\n\treturn [statusStepForTask(step)];\n}\n\nfunction pushPhase(graph: WorkflowGraphSnapshot, phase: string | undefined, nodeId: string): void {\n\tif (!phase) return;\n\tlet group = graph.phases.find((candidate) => candidate.title === phase);\n\tif (!group) {\n\t\tgroup = { title: phase, nodeIds: [] };\n\t\tgraph.phases.push(group);\n\t}\n\tgroup.nodeIds.push(nodeId);\n}\n\nfunction graphNodeForSequential(step: RunnerSubagentStep, stepIndex: number, flatIndex: number): WorkflowGraphNode {\n\treturn {\n\t\tid: `step-${stepIndex}`,\n\t\tkind: \"step\",\n\t\tagent: step.agent,\n\t\tphase: step.phase,\n\t\tlabel: step.label?.trim() || step.agent || `Step ${stepIndex + 1}`,\n\t\tstatus: \"pending\",\n\t\tflatIndex,\n\t\tstepIndex,\n\t\toutputName: step.outputName,\n\t\tstructured: step.structured,\n\t};\n}\n\nfunction graphNodeForParallel(\n\tstep: ParallelStepGroup,\n\tstepIndex: number,\n\tflatIndex: number,\n\tgraph: WorkflowGraphSnapshot,\n): WorkflowGraphNode {\n\tconst children = step.parallel.map((task, taskIndex) => {\n\t\tconst childId = `step-${stepIndex}-agent-${taskIndex}`;\n\t\tpushPhase(graph, task.phase, childId);\n\t\treturn {\n\t\t\tid: childId,\n\t\t\tkind: \"agent\" as const,\n\t\t\tagent: task.agent,\n\t\t\tphase: task.phase,\n\t\t\tlabel: task.label?.trim() || task.agent || `Agent ${taskIndex + 1}`,\n\t\t\tstatus: \"pending\" as const,\n\t\t\tflatIndex: flatIndex + taskIndex,\n\t\t\tstepIndex,\n\t\t\toutputName: task.outputName,\n\t\t\tstructured: task.structured,\n\t\t};\n\t});\n\treturn {\n\t\tid: `step-${stepIndex}`,\n\t\tkind: \"parallel-group\",\n\t\tlabel: step.parallel.length === 1 ? \"Parallel task\" : `Parallel group (${step.parallel.length})`,\n\t\tstatus: \"pending\",\n\t\tstepIndex,\n\t\tchildren,\n\t};\n}\n\nfunction graphNodeForDynamic(step: DynamicRunnerGroup, stepIndex: number): WorkflowGraphNode {\n\treturn {\n\t\tid: `step-${stepIndex}`,\n\t\tkind: \"dynamic-parallel-group\",\n\t\tlabel: step.label?.trim() || step.parallel.label?.trim() || `Dynamic fanout (${step.collect.as})`,\n\t\tstatus: \"pending\",\n\t\tstepIndex,\n\t\toutputName: step.collect.as,\n\t\tstructured: Boolean(step.collect.outputSchema),\n\t\tdynamic: {\n\t\t\tsourceOutput: step.expand.from.output,\n\t\t\tsourcePath: step.expand.from.path,\n\t\t\titemName: step.expand.item ?? \"item\",\n\t\t\tmaxItems: step.expand.maxItems,\n\t\t\tcollectAs: step.collect.as,\n\t\t},\n\t\tchildren: [],\n\t};\n}\n\nfunction appendWorkflowNode(\n\tgraph: WorkflowGraphSnapshot | undefined,\n\tstep: RunnerStep,\n\tstepIndex: number,\n\tflatIndex: number,\n): void {\n\tif (!graph) return;\n\tif (isCheckpointRunnerStep(step)) {\n\t\tconst node: WorkflowGraphNode = {\n\t\t\tid: `step-${stepIndex}`,\n\t\t\tkind: \"checkpoint\",\n\t\t\tphase: step.phase,\n\t\t\tlabel: step.label?.trim() || step.checkpoint,\n\t\t\tstatus: \"pending\",\n\t\t\tflatIndex,\n\t\t\tstepIndex,\n\t\t\tcheckpoint: {\n\t\t\t\tname: step.checkpoint,\n\t\t\t\t...(step.message ? { message: step.message } : {}),\n\t\t\t\tstatus: \"pending\",\n\t\t\t\tstepIndex,\n\t\t\t},\n\t\t};\n\t\tgraph.nodes.push(node);\n\t\tpushPhase(graph, step.phase, node.id);\n\t\treturn;\n\t}\n\tif (isParallelGroup(step)) {\n\t\tgraph.nodes.push(graphNodeForParallel(step, stepIndex, flatIndex, graph));\n\t\treturn;\n\t}\n\tif (isDynamicRunnerGroup(step)) {\n\t\tgraph.nodes.push(graphNodeForDynamic(step, stepIndex));\n\t\treturn;\n\t}\n\tconst node = graphNodeForSequential(step, stepIndex, flatIndex);\n\tgraph.nodes.push(node);\n\tpushPhase(graph, step.phase, node.id);\n}\n\nexport function appendRunnerStepsToStatus(input: {\n\tstatus: AsyncStatus;\n\tsteps: RunnerStep[];\n\tnow?: number;\n\tpendingAppends?: number;\n}): { addedChainSteps: number; addedFlatSteps: number } {\n\tlet addedChainSteps = 0;\n\tlet addedFlatSteps = 0;\n\tfor (const step of input.steps) {\n\t\tconst stepIndex = input.status.chainStepCount ?? input.status.steps?.length ?? 0;\n\t\tconst flatIndex = input.status.steps?.length ?? 0;\n\t\tconst statusSteps = statusStepsForRunnerStep(step, stepIndex);\n\t\tinput.status.steps ??= [];\n\t\tinput.status.steps.push(...statusSteps);\n\t\tif (isParallelGroup(step)) {\n\t\t\tinput.status.parallelGroups ??= [];\n\t\t\tinput.status.parallelGroups.push({ start: flatIndex, count: step.parallel.length, stepIndex });\n\t\t} else if (isDynamicRunnerGroup(step)) {\n\t\t\tinput.status.parallelGroups ??= [];\n\t\t\tinput.status.parallelGroups.push({ start: flatIndex, count: 1, stepIndex });\n\t\t}\n\t\tappendWorkflowNode(input.status.workflowGraph, step, stepIndex, flatIndex);\n\t\tinput.status.chainStepCount = stepIndex + 1;\n\t\taddedChainSteps++;\n\t\taddedFlatSteps += statusSteps.length;\n\t}\n\tinput.status.pendingAppends = input.pendingAppends ?? 0;\n\tinput.status.lastUpdate = input.now ?? Date.now();\n\treturn { addedChainSteps, addedFlatSteps };\n}\n"]}