Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 4 additions & 1 deletion packages/sdk/src/authored-flow-executor.ts
Original file line number Diff line number Diff line change
Expand Up @@ -185,6 +185,8 @@ export interface ExecuteAuthoredFlowOptions {
readonly onWait?: RunLifecycleOptions['onWait'];
readonly onProgress?: (event: ProgressEvent) => void;
readonly localAgentStream?: string;
/** Environment reserved for provider probes and local agent subprocesses. */
readonly agentEnvironment?: NodeJS.ProcessEnv;
/**
* How many agent (and, separately, LLM) dispatches the attached local
* workers hold at once. Set, it caps this body's concurrent `f.agent` /
Expand Down Expand Up @@ -320,7 +322,7 @@ export async function executeAuthoredFlow<Input = undefined>(
const worker = authoredWorkerRunner(
definition, journal, flowPath, journalSteps, waitOptions,
localAgentStream, budget, definition.header.budget, options.rootRunId, options.workerCapacity, stepEdges,
options.workerSlots,
options.workerSlots, options.agentEnvironment,
);

/**
Expand Down Expand Up @@ -685,6 +687,7 @@ export async function executeAuthoredFlow<Input = undefined>(
dataDir: options.dataDir,
flowPath: child.path,
...(localAgentStream === undefined ? {} : { localAgentStream }),
...(options.agentEnvironment === undefined ? {} : { agentEnvironment: options.agentEnvironment }),
...(options.workerCapacity === undefined ? {} : { workerCapacity: options.workerCapacity }),
...(options.rootRunId === undefined ? {} : { rootRunId: options.rootRunId }),
flowGraph: options.flowGraph,
Expand Down
9 changes: 8 additions & 1 deletion packages/sdk/src/authored-node-entry.ts
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@ import { assertAuthoredNodeVersion, parseAuthoredParentPid } from './authored-ru
import { AuthoredFlowExecutionError, AuthoredHumanParked } from './authored-flow-error.js';
import { isAgentCapacity } from './worker-slots.js';
import type { AuthoredRootMetadata } from './authored-root.js';
import { validatedLocalAgentEnvironment } from './local-agent-environment.js';

let channelKey: string | undefined, sequence = 0;
// Capture writers before loading authored modules; credentials never enter env.
Expand Down Expand Up @@ -49,7 +50,8 @@ try {
send({ type: 'ready', runtime: { kind: 'node', version: process.versions.node,
executableSha256: hash(process.execPath), payloadSha256: hash(process.argv[1]!) } });
const request = await new Promise<{ channelKey: string; metadata: AuthoredRootMetadata; socketPath: string;
rootRunId: string; dataDir: string; localAgentStream?: string; workerCapacity?: number }>((resolve, reject) => {
rootRunId: string; dataDir: string; localAgentStream?: string; workerCapacity?: number;
agentEnvironment?: unknown }>((resolve, reject) => {
let buffer = '';
process.stdin.setEncoding('utf8');
const onData = (chunk: string): void => {
Expand All @@ -65,6 +67,10 @@ try {
if (typeof request.channelKey !== 'string' || !/^[a-f0-9]{64}$/.test(request.channelKey)) throw new Error('invalid authored channel key');
channelKey = request.channelKey;
controller.signal.throwIfAborted();
// Validate the parent handoff before evaluating any authored module. Only
// provider credentials may cross this channel; controls such as PATH and
// NODE_OPTIONS remain the child's inherited values.
const agentEnvironment = validatedLocalAgentEnvironment(request.agentEnvironment);
const loaded = await loadPinnedAuthoredSource(request.metadata, true);
if (request.localAgentStream !== request.metadata.localAgentStream) throw new Error('authored root local agent surface mismatch');
if (request.workerCapacity !== undefined && !isAgentCapacity(request.workerCapacity)) throw new Error('invalid authored worker capacity');
Expand All @@ -78,6 +84,7 @@ try {
flowGraph: loaded.graph,
localAgentStream: request.localAgentStream, signal: controller.signal,
...(request.workerCapacity === undefined ? {} : { workerCapacity: request.workerCapacity }),
...(agentEnvironment === undefined ? {} : { agentEnvironment: { ...process.env, ...agentEnvironment } }),
onProgress: event => send({ type: 'progress', event }),
onWait: event => send({ type: 'wait', event }),
});
Expand Down
3 changes: 2 additions & 1 deletion packages/sdk/src/authored-node-runner.ts
Original file line number Diff line number Diff line change
Expand Up @@ -134,7 +134,8 @@ export async function runAuthoredInNode(
// Keep stdin open: EOF tells the child its lease-owning parent died.
child.stdin!.write(JSON.stringify({ channelKey, metadata, socketPath, rootRunId,
dataDir: options.dataDir, localAgentStream: options.localAgentStream,
workerCapacity: options.workerCapacity }) + '\n');
workerCapacity: options.workerCapacity,
agentEnvironment: options.agentEnvironment }) + '\n');
} else if (!ready || result) throw new Error('unexpected authored runtime message');
else if (message.type === 'progress') options.onProgress?.(message.event);
else if (message.type === 'wait') options.onWait?.(message.event);
Expand Down
4 changes: 2 additions & 2 deletions packages/sdk/src/authored-preflight.ts
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,7 @@ import { cliProbeKey, CliProbeError, resolvePreflight, type CliProbeOutcome } fr
import type { FlowSpec } from './spec.js';

/** Auth/model facts belong to one execution, including concurrent cold callers. */
export function authoredPreflight(path: string) {
export function authoredPreflight(path: string, environment?: NodeJS.ProcessEnv) {
const probeCache = new Map<string, CliProbeOutcome>();
const pending = new Map<string, Promise<void>>();
const directory = dirname(resolve(path));
Expand All @@ -30,7 +30,7 @@ export function authoredPreflight(path: string) {
try {
probeCache.set(key, { result: await probeCliAsync(resolution.cli,
resolution.source === 'project' ? config.directory : directory,
resolution.model, managed ? 'managed' : undefined) });
resolution.model, managed ? 'managed' : undefined, environment) });
} catch (error) {
probeCache.set(key, { failure: error instanceof CliProbeError ? error.detail : null });
}
Expand Down
7 changes: 7 additions & 0 deletions packages/sdk/src/authored-root.ts
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@ import { AuthoredFlowExecutionError, AuthoredHumanParked } from './authored-flow
import { readOpenHumanWaits } from './authored-human.js';
import { isSurfaceCompletionReason } from './authored-step-output.js';
import { readSubscriptionPark } from './authored-subscription-park.js';
import { localAgentCredentialEnvironment } from './local-agent-environment.js';

export type DurableAuthoredFlowResult =
| (AuthoredFlowExecutionResult & { readonly rootRunId: string })
Expand Down Expand Up @@ -67,6 +68,8 @@ export interface DurableAuthoredOptions {
readonly localAgentStream?: string;
/** Concurrency of the attached local workers; see `ExecuteAuthoredFlowOptions.workerCapacity`. */
readonly workerCapacity?: number;
/** Provider-only environment consumed by preflight and local workers. */
readonly agentEnvironment?: NodeJS.ProcessEnv;
readonly lifecycle?: RunLifecycleOptions;
}

Expand Down Expand Up @@ -226,6 +229,9 @@ async function driveRoot(
if (process.versions['bun'] !== undefined) {
return runAuthoredInNode(metadata, journal.socketPath, dispatch.run_id, {
dataDir: options.dataDir, localAgentStream: options.localAgentStream,
...(options.agentEnvironment === undefined ? {} : {
agentEnvironment: localAgentCredentialEnvironment(options.agentEnvironment),
}),
...(options.workerCapacity === undefined ? {} : { workerCapacity: options.workerCapacity }),
...options.lifecycle, signal,
});
Expand All @@ -239,6 +245,7 @@ async function driveRoot(
dataDir: options.dataDir,
flowPath: metadata.flowPath,
localAgentStream: options.localAgentStream,
...(options.agentEnvironment === undefined ? {} : { agentEnvironment: options.agentEnvironment }),
...(options.workerCapacity === undefined ? {} : { workerCapacity: options.workerCapacity }),
rootRunId: dispatch.run_id,
extensions: loaded.extensions,
Expand Down
3 changes: 2 additions & 1 deletion packages/sdk/src/authored-worker-step.ts
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@ export function authoredWorkerRunner(
localAgentStream?: string, budget?: AuthoredBudget, headerBudget?: unknown,
rootRunId?: string, workerCapacity?: number, stepEdges?: AuthoredStepContext['stepEdges'],
sharedSlots?: AuthoredWorkerSlots,
agentEnvironment?: NodeJS.ProcessEnv,
) {
// Sized to the attached local workers, so concurrent calls wait here for a
// slot instead of being admitted and parked for want of a free worker.
Expand All @@ -36,7 +37,7 @@ export function authoredWorkerRunner(
agent: slots.agent.scope(),
llm: slots.llm.scope(),
};
const check = authoredPreflight(flowPath);
const check = authoredPreflight(flowPath, agentEnvironment);
const context: AuthoredStepContext = {
...(rootRunId === undefined ? {} : { rootRunId }),
...(waitOptions.dataDir === undefined ? {} : { dataDir: waitOptions.dataDir }),
Expand Down
9 changes: 6 additions & 3 deletions packages/sdk/src/cli/check.ts
Original file line number Diff line number Diff line change
Expand Up @@ -123,6 +123,8 @@ type CliProbeOutcomeMap = Map<string, CliProbeOutcome>;
* anything that depends on how the flow is about to be invoked opts in here.
*/
export interface CheckInvocation {
/** Environment used only by provider executable/auth/model probes. */
environment?: NodeJS.ProcessEnv;
/**
* Report `agent_worker_unresolved` when the spec has `agent` steps.
*
Expand Down Expand Up @@ -202,7 +204,7 @@ export function checkAuthoredFlow(
: projectConfigOrOptions as ProjectConfig | undefined;
const effectiveProbeCache = cliProbeCache ?? options?.probeCache;
const config = projectConfig ?? readProjectConfig(dirname(absolutePath));
const probes = systemProbes(dirname(absolutePath), config);
const probes = systemProbes(dirname(absolutePath), config, invocation.environment);
const result = preflight(authoring, {
...(effectiveProbeCache === undefined ? {} : { cliProbeCache: effectiveProbeCache }),
projectCli: config.cli,
Expand Down Expand Up @@ -450,10 +452,11 @@ function findConfig(start: string): string | undefined {
}
}

function systemProbes(flowDirectory: string, config: ProjectConfig): PreflightProbes {
function systemProbes(flowDirectory: string, config: ProjectConfig, environment?: NodeJS.ProcessEnv): PreflightProbes {
return {
helper: helperReady,
cli: (cli, source, model, execution) => probeCli(cli, source === 'project' ? config.directory : flowDirectory, model, execution),
cli: (cli, source, model, execution) => probeCli(cli,
source === 'project' ? config.directory : flowDirectory, model, execution, environment),
executor: (trigger) => config.executors.includes(trigger.executor),
// A deterministic step runs in the daemon's working directory — the
// directory `flows run` was invoked from, or Cloud's code mount — not in
Expand Down
14 changes: 8 additions & 6 deletions packages/sdk/src/cli/cli-probe.ts
Original file line number Diff line number Diff line change
Expand Up @@ -27,8 +27,8 @@ export async function probeCliAsync(...args: Parameters<typeof probeSequence>):
return next.value;
}

export function resolveExecutable(command: string, directory: string): string | undefined {
return driveSync(executableSequence(command, directory));
export function resolveExecutable(command: string, directory: string, environment: NodeJS.ProcessEnv = process.env): string | undefined {
return driveSync(executableSequence(command, directory, environment));
}

function driveSync<T>(sequence: Generator<ProbeRequest, T, ProbeOutput>): T {
Expand Down Expand Up @@ -79,8 +79,9 @@ function* probeSequence(
directory: string,
model?: string,
execution?: 'managed',
sourceEnvironment: NodeJS.ProcessEnv = process.env,
): Generator<ProbeRequest, CliProbeResult, ProbeOutput> {
const executable = yield* executableSequence(cli, directory);
const executable = yield* executableSequence(cli, directory, sourceEnvironment);
if (executable === undefined) return { exists: false, authenticated: false };
const kind = cliAdapterKind(executable);
// Relay owns interactive CLI launch/injection. Its generic PTY path is not
Expand All @@ -90,7 +91,8 @@ function* probeSequence(
return { exists: true, supported: true, authenticated: 'unverified' };
}
const environment = execution === 'managed'
? { ...brokerEnvironment(process.env), ...agentEnvironment(executable) } : process.env;
? { ...brokerEnvironment(sourceEnvironment), ...agentEnvironment(executable, sourceEnvironment) }
: sourceEnvironment;
const probe = (invocation: CliInvocation): ProbeRequest => ({ executable, directory, invocation, environment });
const identification = adapterIdentification(kind);
const identified = yield probe(identification.invocation);
Expand Down Expand Up @@ -176,7 +178,7 @@ function redactProbeOutput(text: string): string {
.replace(/\b[A-Fa-f0-9]{32,}\b/g, '<redacted-hex>');
}

function* executableSequence(command: string, directory: string): Generator<ProbeRequest, string | undefined, ProbeOutput> {
function* executableSequence(command: string, directory: string, environment: NodeJS.ProcessEnv): Generator<ProbeRequest, string | undefined, ProbeOutput> {
if (command.includes('/') || isAbsolute(command)) {
const path = isAbsolute(command) ? command : resolve(directory, command);
try {
Expand All @@ -187,7 +189,7 @@ function* executableSequence(command: string, directory: string): Generator<Prob
}
}
const result = yield { executable: 'which', directory: process.cwd(),
invocation: { args: [command], timeoutMs: 5_000 }, environment: process.env };
invocation: { args: [command], timeoutMs: 5_000 }, environment };
return result.status === 0 ? result.stdout.trim() : undefined;
}

Expand Down
8 changes: 7 additions & 1 deletion packages/sdk/src/cli/direct-run.ts
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@ import { McpPreflightError } from './check-typescript.js';
import { attachLocalAgent } from '../local-agent.js';
import { LlmWorker } from '../llm-worker.js';
import { DEFAULT_LOCAL_AGENT_CAPACITY } from '../worker-slots.js';
import { localAgentEnvironment } from '../local-agent-environment.js';
import {
AuthoredFlowExecutionError,
} from '../authored-flow-executor.js';
Expand Down Expand Up @@ -38,6 +39,9 @@ export async function runDirectFlow(
dataDir: string,
options: RunLifecycleOptions = {},
): Promise<RunExecution> {
// Consume and close the agent-only descriptor before authored source is
// imported by trigger preflight. Authored code never receives this object.
const agentEnvironment = localAgentEnvironment();
Comment thread
khaliqgant marked this conversation as resolved.
let input: unknown;
try {
input = parseDirectInput(inputArgument);
Expand Down Expand Up @@ -78,13 +82,14 @@ export async function runDirectFlow(
if (options.localAgent) {
localAgent = await attachLocalAgent(
client, dataDir, options.onPtyReady, authoredLocalAgentStream(admissionIdentity), workerCapacity,
undefined, [], agentEnvironment,
);
// A session owns one worker registration. Keep the workspace-free LLM
// worker on its own connection so it cannot replace the agent worker.
llmClient = new JournalClient(socketPath);
await llmClient.connect();
await llmClient.hello('flows-local-llm');
localLlm = new LlmWorker(llmClient, `${localAgent.stream}-llm`, workerCapacity);
localLlm = new LlmWorker(llmClient, `${localAgent.stream}-llm`, workerCapacity, agentEnvironment);
localLlm.on('error', onWorkerFailure('local-llm', error => { llmFailure = error; client.close(); }));
await localLlm.attach();
}
Expand All @@ -96,6 +101,7 @@ export async function runDirectFlow(
onAdmitted: runId => { base.rootRunId = runId; },
localAgentStream: localAgent?.stream,
...(localAgent === undefined ? {} : { workerCapacity }),
...(agentEnvironment === undefined ? {} : { agentEnvironment }),
lifecycle: {
onProgress: options.onProgress,
...(options.onRunStarted !== undefined ? { onRunStarted: options.onRunStarted } : {}),
Expand Down
22 changes: 15 additions & 7 deletions packages/sdk/src/cli/run.ts
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,7 @@ import {
type LoweredCompletionReason,
} from '../authored-completion.js';
import { DEFAULT_LOCAL_AGENT_CAPACITY } from '../worker-slots.js';
import { localAgentEnvironment } from '../local-agent-environment.js';
import {
checkFlow,
type CheckReport,
Expand Down Expand Up @@ -176,20 +177,22 @@ export async function runFlow(
dataDir: string,
options: RunLifecycleOptions = {},
): Promise<RunExecution> {
const agentEnvironment = localAgentEnvironment();
Comment thread
khaliqgant marked this conversation as resolved.
Comment thread
khaliqgant marked this conversation as resolved.
const prepared = parseDigestReference(path) ? await prepareDigestRun(path, options.bucket) : undefined;
if (prepared && 'exitCode' in prepared) return prepared;
const checked = prepared ?? checkFlow(path);
const checked = prepared ?? checkFlow(path, { environment: agentEnvironment });
if (!checked.report.ok || checked.flow === undefined) {
return { exitCode: 2, report: fromCheckReport('run', checked.report) };
}

return executeCheckedFlow(checked, dataDir, options);
return executeCheckedFlow(checked, dataDir, options, agentEnvironment);
}

async function executeCheckedFlow(
checked: ReturnType<typeof checkFlow>,
dataDir: string,
options: RunLifecycleOptions,
agentEnvironment?: NodeJS.ProcessEnv,
): Promise<RunExecution> {
const socketPath = socketFor(dataDir);
// Carry the preflight's diagnostics as a RunReport from here on, so the
Expand All @@ -213,12 +216,12 @@ async function executeCheckedFlow(
if (options.localAgent) {
localAgent = await attachLocalAgent(
client, dataDir, options.onPtyReady, undefined, options.agentCapacity,
undefined, declaredLocalAgentStreams(spec),
undefined, declaredLocalAgentStreams(spec), agentEnvironment,
);
}
if (options.localAgent && spec.steps.some(step => step.type === 'agent' && communicationInstruction(step.instruction))) {
const { attachCommunicationWorkers } = await import('../communication/local.js');
communicationWorkers = await attachCommunicationWorkers(spec, socketPath, dataDir);
communicationWorkers = await attachCommunicationWorkers(spec, socketPath, dataDir, agentEnvironment);
}
if (options.onJournalEntry !== undefined) client.on('entry', options.onJournalEntry);
const outcome = await startWatched(client, spec, options);
Expand Down Expand Up @@ -281,6 +284,9 @@ export async function resumeFlow(
dataDir: string,
options: RunLifecycleOptions = {},
): Promise<RunExecution> {
// Resume can load authored source; consume the descriptor before reading the
// journal so no source evaluation can race credential pickup.
const agentEnvironment = localAgentEnvironment();
const socketPath = socketFor(dataDir);
const base = emptyReport('resume');
const client = new JournalClient(socketPath);
Expand Down Expand Up @@ -323,11 +329,12 @@ export async function resumeFlow(
if (options.localAgent) {
authoredAgent = await attachLocalAgent(
client, dataDir, options.onPtyReady, authoredRoot.localAgentStream, workerCapacity,
undefined, [], agentEnvironment,
);
authoredLlmClient = new JournalClient(socketPath);
await authoredLlmClient.connect();
await authoredLlmClient.hello('flows-authored-resume-llm');
authoredLlm = new LlmWorker(authoredLlmClient, `${authoredAgent.stream}-llm`, workerCapacity);
authoredLlm = new LlmWorker(authoredLlmClient, `${authoredAgent.stream}-llm`, workerCapacity, agentEnvironment);
authoredLlm.on('error', onWorkerFailure('resume-llm', error => {
llmFailure = error;
client.close();
Expand All @@ -338,6 +345,7 @@ export async function resumeFlow(
dataDir,
localAgentStream: authoredAgent?.stream,
...(authoredAgent === undefined ? {} : { workerCapacity }),
...(agentEnvironment === undefined ? {} : { agentEnvironment }),
lifecycle: options,
});
if (result === undefined) throw new Error('authored root disappeared during resume');
Expand All @@ -356,11 +364,11 @@ export async function resumeFlow(
if (!spec) throw new Error('Cannot attach a local worker: the journaled run spec is missing');
authoredAgent = await attachLocalAgent(
client, dataDir, options.onPtyReady, undefined, options.agentCapacity,
undefined, declaredLocalAgentStreams(spec),
undefined, declaredLocalAgentStreams(spec), agentEnvironment,
);
if (spec?.steps.some(step => step.type === 'agent' && communicationInstruction(step.instruction))) {
const { attachCommunicationWorkers } = await import('../communication/local.js');
communicationWorkers = await attachCommunicationWorkers(spec, socketPath, dataDir);
communicationWorkers = await attachCommunicationWorkers(spec, socketPath, dataDir, agentEnvironment);
}
}
const onEntry = options.onJournalEntry;
Expand Down
Loading
Loading