officerdev/platform
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

pipeline executor improvements, proxy refresh, task runner step list

Browse Source 
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
pastilhas
2026-03-12 07:25:11 +00:00
co-authored by Claude Opus 4.6
parent 0f1e1f35f1
commit 92da03fcff
6 changed files with 280 additions and 86 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/servers/api/tasks/pipeline-executor.ts
+46 -69
View File
@@ -31,7 +31,7 @@ type PipelineStep = {
inputs?: Record<string, string>;
foreach?: 'subdirectory';
skip_if?: string;
concurrency?: string;
concurrency?: string | boolean;
};
type PipelineConfig = {
@@ -70,17 +70,29 @@ type RunStepParams = {
taskDirName: string;
prompt: string;
cwd: string;
model: string;
abortSignal: AbortSignal;
emit: EmitEvent;
stepIndex: number;
iterationLabel?: string;
};
async function runAgenticStep({ userId, email, username, role, taskDirName, prompt, cwd, abortSignal, emit, stepIndex, iterationLabel }: RunStepParams): Promise<MessageCost> {
async function refreshProxyToken(): Promise<void> {
const port = process.env.ANTHROPIC_PROXY_PORT ?? '5051';
try {
await fetch(`http://127.0.0.1:${port}/refresh`, { method: 'POST' });
} catch {
// Best effort — proxy may not be running (e.g. using API key directly)
}
}
async function runAgenticStep({ userId, email, username, role, taskDirName, prompt, cwd, model, abortSignal, emit, stepIndex, iterationLabel }: RunStepParams): Promise<MessageCost> {
const sessionId = randomUUID();
const model = await resolveModel(userId);
const isClaudeCode = model.startsWith('claude-code');
// Ensure fresh OAuth token before spawning Claude Code
if (isClaudeCode) await refreshProxyToken();
console.log(`[pipeline] starting step ${stepIndex} for session ${sessionId} (model=${model})${iterationLabel ? ` [${iterationLabel}]` : ''}`);
return new Promise<MessageCost>(async (resolve, reject) => {
@@ -312,7 +324,7 @@ async function runScriptStep({ email, role, task, inputs, cwd, abortSignal, emit
// ── Parallel foreach ──
type ParallelForeachParams = {
type ForeachParams = {
userId: number;
email: string;
username: string;
@@ -327,13 +339,14 @@ type ParallelForeachParams = {
abortSignal: AbortSignal;
totalCost: MessageCost;
emit: EmitEvent;
concurrency: number;
model: string;
};
async function runParallelForeach({
async function runForeach({
userId, email, username, role, stepIdx, step, stepTask, subdirs, baseCwd,
inputs, cwd, abortSignal, totalCost, emit,
}: ParallelForeachParams) {
const concurrency = Math.max(1, parseInt(step.concurrency ?? '1', 10));
inputs, cwd, abortSignal, totalCost, emit, concurrency, model,
}: ForeachParams) {
// Determine skip vs run
const toSkip: string[] = [];
@@ -389,6 +402,7 @@ async function runParallelForeach({
taskDirName: step.task,
prompt,
cwd: resolvedCwd,
model,
abortSignal,
emit,
stepIndex: stepIdx,
@@ -428,11 +442,13 @@ export type ExecutePipelineParams = {
taskDirName: string;
inputs: Record<string, string>;
cwd?: string;
model?: string;
startAt?: number;
abortSignal: AbortSignal;
emit: EmitEvent;
};
export async function executePipeline({ userId, email, username, role, taskDirName, inputs, cwd, abortSignal, emit }: ExecutePipelineParams): Promise<void> {
export async function executePipeline({ userId, email, username, role, taskDirName, inputs, cwd, model: modelOverride, startAt, abortSignal, emit }: ExecutePipelineParams): Promise<void> {
const pipelineTask = await getTaskByDirName(taskDirName, userId);
if (!pipelineTask) {
emit({ type: 'error', message: `Task not found: ${taskDirName}` });
@@ -450,20 +466,30 @@ export async function executePipeline({ userId, email, username, role, taskDirNa
}
const baseCwd = resolveBaseCwd(email, role, cwd);
const model = modelOverride || (await resolveModel(userId));
const totalCost: MessageCost = { inputTokens: 0, outputTokens: 0, totalUSD: 0 };
// Resolve concurrency from user input (default 1)
const runtimeConcurrency = Math.max(1, parseInt(inputs._concurrency ?? '1', 10));
emit({
type: 'pipeline:init',
steps: config.steps.map((s) => ({
task: s.task,
foreach: s.foreach,
concurrency: s.concurrency ? parseInt(s.concurrency, 10) : undefined,
concurrency: s.concurrency ? runtimeConcurrency : undefined,
})),
});
for (let stepIdx = 0; stepIdx < config.steps.length; stepIdx++) {
if (abortSignal.aborted) break;
// Skip steps before startAt
if (startAt && stepIdx < startAt) {
emit({ type: 'step:skip', stepIndex: stepIdx, label: config.steps[stepIdx]!.task, reason: 'skipped (start at)' });
continue;
}
const step = config.steps[stepIdx]!;
const stepTask = await getTaskByDirName(step.task, userId);
@@ -527,65 +553,13 @@ export async function executePipeline({ userId, email, username, role, taskDirNa
return;
}
const concurrency = parseInt(step.concurrency ?? '1', 10);
const concurrency = step.concurrency ? runtimeConcurrency : 1;
if (concurrency > 1) {
await runParallelForeach({
userId, email, username, role,
stepIdx, step, stepTask: { name: stepTask.name, body: stepTask.body! },
subdirs, baseCwd, inputs, cwd, abortSignal, totalCost, emit,
});
} else {
// Sequential
for (let i = 0; i < subdirs.length; i++) {
if (abortSignal.aborted) break;
const subdir = subdirs[i]!;
const subdirPath = join(baseCwd, subdir);
if (step.skip_if && existsSync(join(subdirPath, step.skip_if))) {
emit({ type: 'step:skip', stepIndex: stepIdx, label: subdir, reason: `${step.skip_if} already exists` });
continue;
}
const iterVars = { ...inputs, folder_name: subdir };
const iterInputs: Record<string, string> = {};
if (step.inputs) {
for (const [key, template] of Object.entries(step.inputs)) {
iterInputs[key] = resolveInputTemplate(template, iterVars);
}
}
emit({
type: 'step:start',
stepIndex: stepIdx,
taskName: stepTask.name,
iteration: { current: i + 1, total: subdirs.length, label: subdir },
});
const cwdRelative = cwd ? `${cwd}/${subdir}` : subdir;
const resolvedCwd = resolveBaseCwd(email, role, cwdRelative);
const targetDir = toAgentPath(resolvedCwd, email, role);
const prompt = buildStepPrompt(stepTask.body!,iterInputs, targetDir);
const cost = await runAgenticStep({
userId, email, username, role,
taskDirName: step.task,
prompt,
cwd: resolvedCwd,
abortSignal,
emit,
stepIndex: stepIdx,
iterationLabel: subdir,
});
totalCost.inputTokens += cost.inputTokens;
totalCost.outputTokens += cost.outputTokens;
totalCost.totalUSD += cost.totalUSD;
emit({ type: 'step:complete', stepIndex: stepIdx, cost });
}
}
await runForeach({
userId, email, username, role,
stepIdx, step, stepTask: { name: stepTask.name, body: stepTask.body! },
subdirs, baseCwd, inputs, cwd, abortSignal, totalCost, emit, concurrency, model,
});
} else {
// Single execution step
emit({ type: 'step:start', stepIndex: stepIdx, taskName: stepTask.name });
@@ -598,6 +572,7 @@ export async function executePipeline({ userId, email, username, role, taskDirNa
taskDirName: step.task,
prompt,
cwd: baseCwd,
model,
abortSignal,
emit,
stepIndex: stepIdx,
@@ -629,7 +604,7 @@ type WSData = {
};
type ClientMessage =
| { type: 'run'; taskDirName: string; inputs: Record<string, string>; cwd?: string }
| { type: 'run'; taskDirName: string; inputs: Record<string, string>; cwd?: string; model?: string; startAt?: number }
| { type: 'stop'; jobId: string }
| { type: 'attach'; jobId: string }
| { type: 'list' };
@@ -667,6 +642,8 @@ export async function message(ws: ServerWebSocket<WSData>, raw: string | Buffer)
inputs: msg.inputs,
cwd: msg.cwd,
config: task.config,
model: msg.model,
startAt: msg.startAt,
});
send(ws, { type: 'job:created', jobId });
src/servers/api/tasks/pipeline-job-manager.ts
+4
View File
@@ -67,6 +67,8 @@ type StartJobParams = {
inputs: Record<string, string>;
cwd?: string;
config: unknown;
model?: string;
startAt?: number;
};
export async function startJob(params: StartJobParams): Promise<string> {
@@ -146,6 +148,8 @@ export async function startJob(params: StartJobParams): Promise<string> {
taskDirName: params.taskDirName,
inputs: params.inputs,
cwd: params.cwd,
model: params.model,
startAt: params.startAt,
abortSignal: job.abortSignal,
emit,
}).then(async () => {