From 92da03fcff8e2047cefe05c86d813ad64a54cd30 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Andr=C3=A9=20Padez?= Date: Thu, 12 Mar 2026 07:25:11 +0000 Subject: [PATCH] pipeline executor improvements, proxy refresh, task runner step list Co-Authored-By: Claude Opus 4.6 --- src/servers/api/tasks/pipeline-executor.ts | 115 +++++-------- src/servers/api/tasks/pipeline-job-manager.ts | 4 + src/servers/sidecar/claude/claude-manager.ts | 2 +- src/servers/sidecar/claude/proxy.ts | 161 +++++++++++++++++- .../components/TaskRunnerModal.tsx | 80 ++++++++- .../components/usePipelineRunner.ts | 4 +- 6 files changed, 280 insertions(+), 86 deletions(-) diff --git a/src/servers/api/tasks/pipeline-executor.ts b/src/servers/api/tasks/pipeline-executor.ts index 6ab68286..3f7dde8a 100644 --- a/src/servers/api/tasks/pipeline-executor.ts +++ b/src/servers/api/tasks/pipeline-executor.ts @@ -31,7 +31,7 @@ type PipelineStep = { inputs?: Record; 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 { +async function refreshProxyToken(): Promise { + 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 { 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(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; 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 { +export async function executePipeline({ userId, email, username, role, taskDirName, inputs, cwd, model: modelOverride, startAt, abortSignal, emit }: ExecutePipelineParams): Promise { 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 = {}; - 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; cwd?: string } + | { type: 'run'; taskDirName: string; inputs: Record; 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, raw: string | Buffer) inputs: msg.inputs, cwd: msg.cwd, config: task.config, + model: msg.model, + startAt: msg.startAt, }); send(ws, { type: 'job:created', jobId }); diff --git a/src/servers/api/tasks/pipeline-job-manager.ts b/src/servers/api/tasks/pipeline-job-manager.ts index 696adf0f..8799d1f2 100644 --- a/src/servers/api/tasks/pipeline-job-manager.ts +++ b/src/servers/api/tasks/pipeline-job-manager.ts @@ -67,6 +67,8 @@ type StartJobParams = { inputs: Record; cwd?: string; config: unknown; + model?: string; + startAt?: number; }; export async function startJob(params: StartJobParams): Promise { @@ -146,6 +148,8 @@ export async function startJob(params: StartJobParams): Promise { taskDirName: params.taskDirName, inputs: params.inputs, cwd: params.cwd, + model: params.model, + startAt: params.startAt, abortSignal: job.abortSignal, emit, }).then(async () => { diff --git a/src/servers/sidecar/claude/claude-manager.ts b/src/servers/sidecar/claude/claude-manager.ts index ed4cd39f..d8762de6 100644 --- a/src/servers/sidecar/claude/claude-manager.ts +++ b/src/servers/sidecar/claude/claude-manager.ts @@ -6,7 +6,7 @@ import { buildSandboxPrefix, buildRunuserSuffix, SANDBOX_HOME } from '../sandbox import { setClaudeSession, clearClaudeSession, getClaudeSession } from './state'; import { parseStream } from './stream-parser'; -const SEND_TIMEOUT_MS = 5 * 60 * 1000; +const SEND_TIMEOUT_MS = 30 * 60 * 1000; // Use /usr/local/bin/claude so it's visible inside bwrap sandbox (which ro-binds /usr). // The actual binary lives at ~/.local/bin/claude, symlinked from /usr/local/bin/claude. diff --git a/src/servers/sidecar/claude/proxy.ts b/src/servers/sidecar/claude/proxy.ts index f265490b..1dee1aee 100644 --- a/src/servers/sidecar/claude/proxy.ts +++ b/src/servers/sidecar/claude/proxy.ts @@ -5,24 +5,119 @@ import { getState, updateState } from './state'; const PROXY_PORT = Number(process.env.ANTHROPIC_PROXY_PORT ?? '5051'); const ANTHROPIC_API_BASE = 'https://api.anthropic.com'; const CREDENTIALS_PATH = join(homedir(), '.claude', '.credentials.json'); +const TOKEN_URL = 'https://platform.claude.com/v1/oauth/token'; +const CLIENT_ID = '9d1c250a-e61b-44d9-88ed-5944d1962f5e'; -type CredentialsFile = { - claudeAiOauth?: { - accessToken?: string; - }; +// Buffer: refresh 5 minutes before expiry +const EXPIRY_BUFFER_MS = 5 * 60 * 1000; + +type OAuthCredentials = { + accessToken?: string; + refreshToken?: string; + expiresAt?: number; }; -async function readOAuthToken(): Promise { +type CredentialsFile = { + claudeAiOauth?: OAuthCredentials; +}; + +async function readCredentials(): Promise { try { const file = Bun.file(CREDENTIALS_PATH); if (!(await file.exists())) return null; - const data = (await file.json()) as CredentialsFile; - return data.claudeAiOauth?.accessToken?.trim() || null; + return (await file.json()) as CredentialsFile; } catch { return null; } } +async function writeCredentials(creds: CredentialsFile): Promise { + try { + await Bun.write(CREDENTIALS_PATH, JSON.stringify(creds, null, 2)); + } catch (err) { + console.error('[claude:proxy] failed to write credentials:', err); + } +} + +async function refreshOAuthToken(refreshToken: string): Promise { + try { + console.log('[claude:proxy] refreshing OAuth token...'); + const res = await fetch(TOKEN_URL, { + method: 'POST', + headers: { 'Content-Type': 'application/x-www-form-urlencoded' }, + body: new URLSearchParams({ + grant_type: 'refresh_token', + refresh_token: refreshToken, + client_id: CLIENT_ID, + }), + }); + + if (!res.ok) { + const text = await res.text(); + console.error(`[claude:proxy] token refresh failed (${res.status}): ${text}`); + return null; + } + + const data = (await res.json()) as { access_token?: string; refresh_token?: string; expires_in?: number }; + if (!data.access_token) { + console.error('[claude:proxy] token refresh response missing access_token'); + return null; + } + + const expiresAt = data.expires_in ? Date.now() + data.expires_in * 1000 : undefined; + console.log(`[claude:proxy] token refreshed, expires in ${data.expires_in ?? '?'}s`); + + return { + accessToken: data.access_token, + refreshToken: data.refresh_token ?? refreshToken, + expiresAt, + }; + } catch (err) { + console.error('[claude:proxy] token refresh error:', err); + return null; + } +} + +async function getValidToken(): Promise { + const creds = await readCredentials(); + if (!creds?.claudeAiOauth) return null; + + const oauth = creds.claudeAiOauth; + const now = Date.now(); + + // Check if token is still valid (with buffer) + if (oauth.accessToken && oauth.expiresAt && oauth.expiresAt - now > EXPIRY_BUFFER_MS) { + return oauth.accessToken.trim(); + } + + // Token expired or about to expire — try refresh + if (oauth.refreshToken) { + const refreshed = await refreshOAuthToken(oauth.refreshToken); + if (refreshed?.accessToken) { + creds.claudeAiOauth = { ...oauth, ...refreshed }; + await writeCredentials(creds); + return refreshed.accessToken.trim(); + } + } + + // Fallback: return current token even if potentially expired (let upstream decide) + return oauth.accessToken?.trim() || null; +} + +async function forceRefresh(): Promise { + const creds = await readCredentials(); + const refreshToken = creds?.claudeAiOauth?.refreshToken; + if (!refreshToken) return null; + + const refreshed = await refreshOAuthToken(refreshToken); + if (refreshed?.accessToken && creds?.claudeAiOauth) { + creds.claudeAiOauth = { ...creds.claudeAiOauth, ...refreshed }; + await writeCredentials(creds); + return refreshed.accessToken; + } + return null; +} + export function getProxySecret(): string { return getState().proxySecret; } @@ -48,6 +143,18 @@ export function startAnthropicProxy() { async fetch(req) { const url = new URL(req.url); + // Internal refresh endpoint (no auth needed — localhost only) + if (url.pathname === '/refresh') { + const token = await forceRefresh(); + if (token) { + return new Response(JSON.stringify({ ok: true }), { headers: { 'Content-Type': 'application/json' } }); + } + return new Response(JSON.stringify({ error: 'Refresh failed' }), { + status: 502, + headers: { 'Content-Type': 'application/json' }, + }); + } + // Validate proxy secret const incomingKey = req.headers.get('x-api-key'); if (incomingKey !== secret) { @@ -57,8 +164,8 @@ export function startAnthropicProxy() { }); } - // Read fresh OAuth token - const token = await readOAuthToken(); + // Get valid token (refreshes if needed) + const token = await getValidToken(); if (!token) { return new Response(JSON.stringify({ error: 'No OAuth token available' }), { status: 502, @@ -84,6 +191,42 @@ export function startAnthropicProxy() { body, }); + // If upstream returns 401, try one refresh and retry + if (upstreamRes.status === 401) { + const creds = await readCredentials(); + const refreshToken = creds?.claudeAiOauth?.refreshToken; + if (refreshToken) { + const refreshed = await refreshOAuthToken(refreshToken); + if (refreshed?.accessToken) { + if (creds?.claudeAiOauth) { + creds.claudeAiOauth = { ...creds.claudeAiOauth, ...refreshed }; + await writeCredentials(creds); + } + + headers.set('x-api-key', refreshed.accessToken.trim()); + const retryBody = body ? new Uint8Array(body) : null; + const retryRes = await fetch(upstream, { + method: req.method, + headers, + body: retryBody, + }); + + const retryHeaders = new Headers(); + for (const [key, value] of retryRes.headers) { + const lower = key.toLowerCase(); + if (lower === 'content-encoding' || lower === 'content-length' || lower === 'transfer-encoding') continue; + retryHeaders.set(key, value); + } + + return new Response(retryRes.body, { + status: retryRes.status, + statusText: retryRes.statusText, + headers: retryHeaders, + }); + } + } + } + // Build clean response headers const resHeaders = new Headers(); for (const [key, value] of upstreamRes.headers) { diff --git a/src/workspaces/officerdev/src/apps/FileBrowser/FileBrowserApp/components/TaskRunnerModal.tsx b/src/workspaces/officerdev/src/apps/FileBrowser/FileBrowserApp/components/TaskRunnerModal.tsx index 333b55ff..09f8f3c7 100644 --- a/src/workspaces/officerdev/src/apps/FileBrowser/FileBrowserApp/components/TaskRunnerModal.tsx +++ b/src/workspaces/officerdev/src/apps/FileBrowser/FileBrowserApp/components/TaskRunnerModal.tsx @@ -491,11 +491,18 @@ const PipelineRunner = ({ taskDirName, context, cwd }: PipelineRunnerProps) => { const bottomRef = useRef(null); const [inputDefs, setInputDefs] = useState | null>(null); const [formValues, setFormValues] = useState>({}); + const [hasConcurrentSteps, setHasConcurrentSteps] = useState(false); + const [concurrency, setConcurrency] = useState('1'); + const [pipelineSteps, setPipelineSteps] = useState>([]); + const [startAt, setStartAt] = useState(0); + const { settings } = useSettings(); + const availableModels = useUserVisibleModels(); + const [selectedModel, setSelectedModel] = useState(settings.tasks.defaultModel ?? 'claude-haiku-4-5-20251001'); - // Fetch task detail for inputs + // Fetch task detail for inputs + check for concurrent steps useEffect(() => { - client.get<{ inputs?: Record }>(`/tasks/${taskDirName}`).then((task) => { - const defs = task.inputs ?? {}; + client.get<{ inputs?: Record; config?: { steps?: Array<{ task: string; foreach?: string; concurrency?: string | boolean }> } }>(`/tasks/${taskDirName}`).then((task) => { + const defs: Record = task.inputs ?? {}; setInputDefs(defs); const initial: Record = {}; for (const [key, def] of Object.entries(defs)) { @@ -503,6 +510,10 @@ const PipelineRunner = ({ taskDirName, context, cwd }: PipelineRunnerProps) => { else if (def.default !== undefined) initial[key] = def.default; } setFormValues(initial); + + const steps = task.config?.steps ?? []; + setPipelineSteps(steps.map((s) => ({ task: s.task, foreach: s.foreach }))); + setHasConcurrentSteps(steps.some((s: { concurrency?: string | boolean }) => s.concurrency)); }); }, [taskDirName]); @@ -525,7 +536,11 @@ const PipelineRunner = ({ taskDirName, context, cwd }: PipelineRunnerProps) => { }, [pipeline.phase]); const handleRun = () => { - pipeline.run(taskDirName, formValues, cwd); + const allInputs = { ...formValues }; + if (hasConcurrentSteps && parseInt(concurrency, 10) > 1) { + allInputs._concurrency = concurrency; + } + pipeline.run(taskDirName, allInputs, cwd, selectedModel, startAt > 0 ? startAt : undefined); }; const hasConfigurableInputs = inputDefs && Object.keys(inputDefs).length > 0; @@ -541,6 +556,61 @@ const PipelineRunner = ({ taskDirName, context, cwd }: PipelineRunnerProps) => { autoFilledKeys={new Set()} /> )} + {hasConcurrentSteps && ( +
+
+ Concurrency + parallel runners per step +
+ setConcurrency(e.target.value)} + className="w-16 px-2 py-1 text-sm text-center rounded-lg border border-duck-dark/15 dark:border-foreground/15 bg-background focus:outline-none focus:ring-1 focus:ring-duck-teal" + /> +
+ )} +
+ Model + +
+ {pipelineSteps.length > 0 && ( +
+ Steps +
+ {pipelineSteps.map((step, i) => ( + + ))} +
+
+ )}
diff --git a/src/workspaces/officerdev/src/apps/FileBrowser/FileBrowserApp/components/usePipelineRunner.ts b/src/workspaces/officerdev/src/apps/FileBrowser/FileBrowserApp/components/usePipelineRunner.ts index b228de35..0d0873f7 100644 --- a/src/workspaces/officerdev/src/apps/FileBrowser/FileBrowserApp/components/usePipelineRunner.ts +++ b/src/workspaces/officerdev/src/apps/FileBrowser/FileBrowserApp/components/usePipelineRunner.ts @@ -290,7 +290,7 @@ export function usePipelineRunner() { }; }, []); - const run = useCallback((taskDirName: string, inputs: Record, cwd?: string) => { + const run = useCallback((taskDirName: string, inputs: Record, cwd?: string, model?: string, startAt?: number) => { if (!wsRef.current || wsRef.current.readyState !== WebSocket.OPEN) return; setPhase('running'); @@ -313,7 +313,7 @@ export function usePipelineRunner() { setElapsed(Math.floor((Date.now() - startTimeRef.current) / 1000)); }, 1000); - wsRef.current.send(JSON.stringify({ type: 'run', taskDirName, inputs, cwd })); + wsRef.current.send(JSON.stringify({ type: 'run', taskDirName, inputs, cwd, model, startAt })); }, []); const stop = useCallback(() => {