diff --git a/docs/SURFACE.md b/docs/SURFACE.md
index 19fda3355..af6030066 100644
--- a/docs/SURFACE.md
+++ b/docs/SURFACE.md
@@ -858,6 +858,15 @@ child. Helper-provider, MCP and plugin-effect children are not yet included
in this index. Once appended, the index survives process exit and is readable from
the journal on disk, including after a cooperative nonzero exit.
+Each record, and the matching root `journalSteps` entry, may also carry the
+step's place in the run's DAG. `after` lists the step ids it causally waited
+for, transitively reduced and capped at 32. `afterTruncated: true` means that
+list is incomplete. `label` is the author-chosen `f.agent` or `f.hook` name,
+kept whole or omitted when it is over 256 characters, because a cut label
+could end partway through a secret. `f.run` has no label, because a command
+can carry literal tokens and URLs. The fields are optional; a step with none
+writes the original record shape.
+
An authored step-failure JSON report keeps the child in `runId` and adds
`rootRunId` for the durable authored root. Consumers must use `rootRunId` for
the resume pointer and root index, and `runId` for the failing child's evidence.
diff --git a/packages/sdk/src/authored-flow-executor.ts b/packages/sdk/src/authored-flow-executor.ts
index 7aab04482..ebf9c28a1 100644
--- a/packages/sdk/src/authored-flow-executor.ts
+++ b/packages/sdk/src/authored-flow-executor.ts
@@ -43,6 +43,7 @@ import {
verifyAuthoredOperations,
} from './authored-flow-operation.js';
import { AuthoredFlowLifecycle } from './authored-flow-lifecycle.js';
+import { displayLabel, type AuthoredStepEdges } from './authored-step-index.js';
import { JournalClient } from './journal-client.js';
import { PluginError } from './plugin-manifest.js';
import { createHookEvaluator } from './authored-hooks.js';
@@ -77,7 +78,7 @@ type EveryRunCompletionReasonIsAcceptedByDone = Assert<
export { AuthoredFlowExecutionError, type AuthoredFlowExecutionErrorCode };
-export interface AuthoredFlowJournalStep {
+export interface AuthoredFlowJournalStep extends AuthoredStepEdges {
readonly id: string;
readonly runId: string;
readonly completionReason: ProtocolCompletionReason;
@@ -239,6 +240,7 @@ export async function executeAuthoredFlow(
const journalSteps: AuthoredFlowJournalStep[] = [];
const authoredSteps: AuthoredFlowOperation[] = [];
const lifecycle = new AuthoredFlowLifecycle();
+ const stepEdges = (step: string): AuthoredStepEdges | undefined => lifecycle.stepEdges(step);
let nextStep = 1;
let requestedCompletion: LoweredCompletionReason | undefined;
@@ -246,12 +248,13 @@ export async function executeAuthoredFlow(
definition.name, journal, journalSteps, budget, {
...(options.rootRunId === undefined ? {} : { rootRunId: options.rootRunId }),
...(options.dataDir === undefined ? {} : { dataDir: options.dataDir }),
+ stepEdges,
},
);
const worker = authoredWorkerRunner(
definition, journal, flowPath, journalSteps, waitOptions,
- localAgentStream, budget, definition.header.budget, options.rootRunId, options.workerCapacity,
+ localAgentStream, budget, definition.header.budget, options.rootRunId, options.workerCapacity, stepEdges,
);
/**
@@ -365,6 +368,7 @@ export async function executeAuthoredFlow(
return worker.llm(id, text, undefined, llmOp.namedGate);
}, onProgress),
lifecycle,
+ {},
);
return trackStep(authoredSteps, llmOp);
}
@@ -442,13 +446,17 @@ export async function executeAuthoredFlow(
() => assertOperationAllowed('run', definition.name, requestedCompletion),
() => observeStep(id, 'deterministic', () => lowerDeterministic(id, command, false, leaseMs, runOp.namedGate), options.onProgress),
lifecycle,
+ // No label: a command is not a display name. It carries literal
+ // tokens and URLs, and no prefix of it is safe to show (displayLabel).
+ {},
);
return trackStep(authoredSteps, runOp);
},
llm: llmOperation,
agent(name, options) {
assertOperationAllowed('agent', definition.name, requestedCompletion);
- void name; // Authored headers do not yet declare reusable named agents.
+ // Authored headers do not yet declare reusable named agents, so the name
+ // identifies nothing to the kernel; it is the step's label in the DAG.
const id = `agent-${nextStep++}`;
let agentOp!: AuthoredFlowOperation;
agentOp = new AuthoredFlowOperation(
@@ -457,6 +465,7 @@ export async function executeAuthoredFlow(
() => assertOperationAllowed('agent', definition.name, requestedCompletion),
() => observeStep(id, 'agent', () => worker.agent(id, options, agentOp.namedGate), onProgress),
lifecycle,
+ displayLabel(typeof name === 'string' ? name : undefined),
);
return trackStep(authoredSteps, agentOp);
},
@@ -518,6 +527,7 @@ export async function executeAuthoredFlow(
return recorded.answer;
}, options.onProgress),
lifecycle,
+ {},
);
return trackStep(authoredSteps, humanOp);
},
@@ -540,6 +550,7 @@ export async function executeAuthoredFlow(
return verdict;
},
lifecycle,
+ displayLabel(typeof name === 'string' ? name : undefined),
));
},
done(reason) {
diff --git a/packages/sdk/src/authored-flow-lifecycle.ts b/packages/sdk/src/authored-flow-lifecycle.ts
index 31fe2caea..955f07c72 100644
--- a/packages/sdk/src/authored-flow-lifecycle.ts
+++ b/packages/sdk/src/authored-flow-lifecycle.ts
@@ -1,6 +1,8 @@
import { AsyncLocalStorage, executionAsyncId } from 'node:async_hooks';
import { AuthoredFlowExecutionError } from './authored-flow-error.js';
import { AuthoredPromiseGraph } from './authored-promise-graph.js';
+import { AuthoredStepGraph } from './authored-step-graph.js';
+import type { AuthoredStepEdges } from './authored-step-index.js';
type OperationToken = object;
type ResolverProbe = Set;
@@ -66,7 +68,7 @@ const observedCombinators: Record = Object.fromEntri
const members = collectMembers(values);
if (members === undefined) return native.call(this, values);
const aggregate = native.call(this, members);
- activeLifecycle.getStore()?.registerCombinator(members, aggregate);
+ activeLifecycle.getStore()?.registerCombinator(members, aggregate, name);
return aggregate;
};
Object.defineProperty(observed, 'name', { value: name, configurable: true });
@@ -117,6 +119,8 @@ export class AuthoredFlowLifecycle {
*/
applyPredicateGate: ((operation: { id: string; predicateGate: unknown }, value: T) => Promise) | undefined = undefined;
private readonly graph: AuthoredPromiseGraph;
+ private readonly stepGraph: AuthoredStepGraph;
+ private readonly operationPromises = new Map();
private readonly activeResolverProbes: ResolverProbe[] = [];
private readonly invocations = new Map();
private readonly promiseAllAggregates = new Map>();
@@ -135,6 +139,7 @@ export class AuthoredFlowLifecycle {
for (const probe of this.activeResolverProbes) probe.add(asyncId);
},
);
+ this.stepGraph = new AuthoredStepGraph((asyncId) => this.graph.causesOf(asyncId));
installPromiseAllObserver();
try {
this.graph.enable();
@@ -152,16 +157,25 @@ export class AuthoredFlowLifecycle {
stepOwners.set(step, { lifecycle: this, operation });
}
- registerCombinator(values: readonly unknown[], aggregate: Promise): void {
+ registerCombinator(
+ values: readonly unknown[],
+ aggregate: Promise,
+ combinator: CombinatorName,
+ ): void {
const aggregateId = this.graph.idOf(aggregate);
if (aggregateId === undefined) return;
this.graph.registerRoot(aggregateId);
const memberIds = new Set();
+ const awaitedMembers: number[] = [];
for (const value of values) {
if ((typeof value !== 'object' && typeof value !== 'function') || value === null) continue;
const memberId = this.graph.idOf(value);
if (memberId !== undefined) memberIds.add(memberId);
const owner = stepOwners.get(value);
+ // A Step is a thenable, not a promise: its member identity for the step
+ // graph is the operation's own promise.
+ const awaited = owner?.lifecycle === this ? this.operationPromises.get(owner.operation) : memberId;
+ if (awaited !== undefined) awaitedMembers.push(awaited);
if (owner?.lifecycle !== this) continue;
let operationAggregates = this.promiseAllAggregates.get(owner.operation);
if (operationAggregates === undefined) {
@@ -172,6 +186,33 @@ export class AuthoredFlowLifecycle {
}
this.promiseAllGroups.push({ aggregate: aggregateId, members: memberIds });
this.registrations++;
+ this.stepGraph.registerAggregate(aggregateId, aggregate, combinator, awaitedMembers);
+ }
+
+ /**
+ * Record an operation's own promise and, for a step the run's DAG draws, its
+ * place in that DAG. Called synchronously from the operation's constructor:
+ * the causal predecessors are read from the context that invoked
+ * `f.agent`/`f.run`/`f.llm`, which is observable only at that moment.
+ *
+ * Every operation's promise is recorded so `Promise.all([step, …])` can
+ * expand to it; only a `node` becomes a frontier the walk stops at. An
+ * operation that is not drawn (a helper effect) is walked through instead,
+ * so an edge never names a step the index has no record for.
+ */
+ registerOperation(
+ operation: OperationToken,
+ promise: Promise,
+ node?: { readonly step: string; readonly label?: string },
+ ): AuthoredStepEdges | undefined {
+ const promiseId = this.graph.idOf(promise);
+ if (promiseId !== undefined) this.operationPromises.set(operation, promiseId);
+ if (node === undefined) return undefined;
+ return this.stepGraph.registerStep(node.step, promiseId, executionAsyncId(), node.label);
+ }
+
+ stepEdges(step: string): AuthoredStepEdges | undefined {
+ return this.stepGraph.edgesOf(step);
}
registerInvocation(
@@ -304,6 +345,8 @@ export class AuthoredFlowLifecycle {
this.closed = true;
this.graph.disable();
this.graph.clear();
+ this.stepGraph.release();
+ this.operationPromises.clear();
this.invocations.clear();
this.promiseAllAggregates.clear();
this.promiseAllGroups.length = 0;
diff --git a/packages/sdk/src/authored-flow-operation.ts b/packages/sdk/src/authored-flow-operation.ts
index aea720092..e877c2dda 100644
--- a/packages/sdk/src/authored-flow-operation.ts
+++ b/packages/sdk/src/authored-flow-operation.ts
@@ -5,6 +5,7 @@ import {
AuthoredFlowLifecycle,
type AuthoredOperationInvocation,
} from './authored-flow-lifecycle.js';
+import type { AuthoredStepEdges } from './authored-step-index.js';
/** Slice-P kinds the surface `.gate(config)` accepts and the SDK lowers. */
const NAMED_GATE_KINDS = new Set([
@@ -39,6 +40,11 @@ export class AuthoredFlowOperation {
* function itself is never serialized. `flows check` cannot prove it.
*/
predicateGate: { predicate: (value: T) => boolean; because?: string } | undefined = undefined;
+ /**
+ * This step's label and causal predecessors in the run's DAG, captured when
+ * it was invoked. Undefined for an operation the step index does not draw.
+ */
+ readonly edges: AuthoredStepEdges | undefined;
private state: OperationState = 'created';
private thenInvoked = false;
private rootFailureRecorded = false;
@@ -55,6 +61,7 @@ export class AuthoredFlowOperation {
private readonly assertCanStart: () => void,
private readonly start: () => Promise,
private readonly scope: AuthoredFlowLifecycle,
+ node?: { readonly label?: string },
) {
let resolve!: (value: T | PromiseLike) => void;
let reject!: (reason?: unknown) => void;
@@ -64,6 +71,9 @@ export class AuthoredFlowOperation {
});
this.resolve = resolve;
this.reject = reject;
+ this.edges = scope.registerOperation(
+ this, this.promise, node === undefined ? undefined : { step: id, ...node },
+ );
observeRejection(this.promise, (error) => this.recordRootFailure(error));
const operation = this;
diff --git a/packages/sdk/src/authored-step-graph.ts b/packages/sdk/src/authored-step-graph.ts
new file mode 100644
index 000000000..aaa8e2aac
--- /dev/null
+++ b/packages/sdk/src/authored-step-graph.ts
@@ -0,0 +1,246 @@
+import { inspect } from 'node:util';
+import { MAX_AFTER, displayLabel, type AuthoredStepEdges } from './authored-step-index.js';
+
+/**
+ * The named DAG an authored body draws, recorded as it runs.
+ *
+ * An authored flow is plain TypeScript, so the shape Cloud wants to draw — which
+ * `f.agent` waited for which — is not declared anywhere. It is, however, a fact
+ * the runtime already holds: the promise graph (`authored-promise-graph.ts`)
+ * records, for every promise the body creates, what triggered it, what resolved
+ * it, and which context created it. A step's causal predecessors are the step
+ * promises reachable backward from the context that invoked it.
+ *
+ * Three rules make that walk an answer rather than an approximation:
+ *
+ * 1. **Stop at steps.** A step promise is a frontier; what it waited for is its
+ * own `after`, not ours.
+ * 2. **Expand an aggregate that waited for every member.** The runtime links a
+ * combinator's aggregate only to the member whose settlement settled it.
+ * That member is the whole answer when the aggregate settled on the first
+ * member that could decide it — a rejected `all`, a fulfilled `any`, any
+ * `race` — and a fraction of it when the aggregate had to wait for all of
+ * them: a fulfilled `all`, a rejected `any`, every `allSettled`. Which case
+ * applies is known only once the aggregate settles, so it is decided when a
+ * walk reaches the aggregate (`registerAggregate`, `promiseOutcome`).
+ * 3. **Reduce transitively.** `a → b → c` reaches both `a` and `b` from `c`;
+ * `a` is an ancestor of `b`, so it is dropped.
+ *
+ * Cost is bounded, not quadratic. Each invocation context's frontier is
+ * memoized, and a walk that reaches an already-answered context takes its
+ * answer instead of re-walking the history behind it — so a long sequential
+ * body walks a few nodes per step. A walk that still exceeds `walkLimit` stops
+ * and says so (`afterTruncated`) rather than guessing silently.
+ *
+ * This is observational metadata: it never gates execution, and it changes no
+ * step id, admission key or resume identity.
+ */
+export const MAX_WALK = 10_000;
+
+interface Frontier {
+ readonly steps: readonly string[];
+ readonly truncated: boolean;
+}
+
+interface StepNode {
+ readonly order: number;
+ readonly after: readonly string[];
+ /** `after` may be incomplete, so a reduction that relies on it may keep an ancestor. */
+ readonly truncated: boolean;
+}
+
+type Combinator = 'all' | 'allSettled' | 'any' | 'race';
+type Outcome = 'pending' | 'fulfilled' | 'rejected';
+
+/** The settlement on which a combinator had waited for every member. */
+const EXPANDS_ON: Readonly> = {
+ all: 'fulfilled',
+ allSettled: 'always',
+ any: 'rejected',
+ race: undefined,
+};
+
+interface Aggregate {
+ readonly promise: Promise;
+ /** Cached once settled: a settled promise never changes state. */
+ outcome?: Outcome;
+ readonly expandsOn: Outcome | 'always';
+ readonly members: readonly number[];
+}
+
+export class AuthoredStepGraph {
+ private readonly stepPromises = new Map();
+ private readonly aggregates = new Map();
+ private readonly contexts = new Map();
+ private readonly nodes = new Map();
+ private readonly edges = new Map();
+
+ constructor(
+ private readonly causesOf: (asyncId: number) => readonly number[],
+ private readonly walkLimit = MAX_WALK,
+ ) {}
+
+ /** A combinator's aggregate and its members, for expansion once it settles. */
+ registerAggregate(
+ aggregateId: number,
+ promise: Promise,
+ combinator: Combinator,
+ members: readonly number[],
+ ): void {
+ const expandsOn = EXPANDS_ON[combinator];
+ if (expandsOn === undefined || members.length === 0) return;
+ this.aggregates.set(aggregateId, { promise, expandsOn, members });
+ }
+
+ /**
+ * Record a step invoked from `contextId`. `promiseId` is the step's own
+ * promise, when the promise graph saw it, so later walks stop there.
+ */
+ registerStep(
+ step: string,
+ promiseId: number | undefined,
+ contextId: number,
+ label?: string,
+ ): AuthoredStepEdges {
+ const frontier = this.frontierOf(contextId);
+ this.nodes.set(step, { order: this.nodes.size, after: frontier.steps, truncated: frontier.truncated });
+ if (promiseId !== undefined) this.stepPromises.set(promiseId, step);
+ const truncated = frontier.truncated || frontier.steps.length > MAX_AFTER;
+ const edges: AuthoredStepEdges = Object.freeze({
+ ...displayLabel(label),
+ ...(frontier.steps.length === 0 ? {} : { after: Object.freeze(frontier.steps.slice(0, MAX_AFTER)) }),
+ ...(truncated ? { afterTruncated: true as const } : {}),
+ });
+ this.edges.set(step, edges);
+ return edges;
+ }
+
+ edgesOf(step: string): AuthoredStepEdges | undefined {
+ return this.edges.get(step);
+ }
+
+ /** Drop the walk state; recorded edges stay readable for late record writes. */
+ release(): void {
+ this.stepPromises.clear();
+ this.aggregates.clear();
+ this.contexts.clear();
+ this.nodes.clear();
+ }
+
+ private frontierOf(start: number): Frontier {
+ const memo = this.contexts.get(start);
+ if (memo !== undefined) return memo;
+ const candidates = new Set();
+ let truncated = false;
+ const visited = new Set([start]);
+ const queue = [start];
+ // Breadth-first, so that a walk cut short by the limit has already found
+ // the nearest predecessors and lost only distant history.
+ for (let head = 0; head < queue.length; head++) {
+ const current = queue[head]!;
+ const step = this.stepPromises.get(current);
+ if (step !== undefined) {
+ candidates.add(step);
+ continue;
+ }
+ const answered = current === start ? undefined : this.contexts.get(current);
+ if (answered !== undefined) {
+ for (const found of answered.steps) candidates.add(found);
+ truncated ||= answered.truncated;
+ continue;
+ }
+ for (const id of [...this.waitedFor(current), ...this.causesOf(current)]) {
+ if (visited.has(id)) continue;
+ // At the limit, stop growing the walk but still drain what is queued:
+ // those nodes are already known and cost one lookup each.
+ if (visited.size >= this.walkLimit) {
+ truncated = true;
+ break;
+ }
+ visited.add(id);
+ queue.push(id);
+ }
+ }
+ const reduced = this.reduce(candidates);
+ const frontier = Object.freeze({ steps: reduced.steps, truncated: truncated || reduced.truncated });
+ this.contexts.set(start, frontier);
+ return frontier;
+ }
+
+ /** The members an aggregate waited for beyond its own resolution edge. */
+ private waitedFor(asyncId: number): readonly number[] {
+ const aggregate = this.aggregates.get(asyncId);
+ if (aggregate === undefined) return [];
+ if (aggregate.expandsOn === 'always') return aggregate.members;
+ if (aggregate.outcome === undefined) {
+ const outcome = promiseOutcome(aggregate.promise);
+ if (outcome !== 'pending') aggregate.outcome = outcome;
+ }
+ return aggregate.outcome === aggregate.expandsOn ? aggregate.members : [];
+ }
+
+ /**
+ * Drop every candidate that is an ancestor of another candidate. The
+ * reduction reads recorded `after` lists; when one it reads is incomplete,
+ * a redundant ancestor can survive, and the result says so.
+ */
+ private reduce(candidates: ReadonlySet): Frontier {
+ const ordered = [...candidates].sort((a, b) => this.orderOf(a) - this.orderOf(b));
+ if (ordered.length <= 1) return { steps: ordered, truncated: false };
+ // An ancestor is always registered before its descendant, so nothing
+ // older than the oldest candidate can be one: the walk stops there.
+ const oldest = this.orderOf(ordered[0]!);
+ const reached = new Set();
+ // Predecessor lists are read through cursors, never copied onto the stack, and every entry
+ // inspected counts toward the limit — duplicates included — so a wide fan-in bounds both
+ // allocation and work, not just the number of distinct steps reached.
+ const stack: Array<{ readonly values: readonly string[]; next: number }> = [];
+ const pushList = (values: readonly string[] | undefined): void => {
+ if (values !== undefined && values.length > 0) stack.push({ values, next: 0 });
+ };
+ for (const step of ordered) pushList(this.nodes.get(step)?.after);
+ let inspected = 0;
+ let truncated = ordered.some((step) => this.nodes.get(step)?.truncated === true);
+ while (stack.length > 0) {
+ const frame = stack[stack.length - 1]!;
+ const step = frame.values[frame.next++]!;
+ if (frame.next === frame.values.length) stack.pop();
+ if (++inspected > this.walkLimit) {
+ truncated = true;
+ break;
+ }
+ if (reached.has(step)) continue;
+ reached.add(step);
+ const node = this.nodes.get(step);
+ if (node === undefined || node.order <= oldest) continue;
+ truncated ||= node.truncated;
+ pushList(node.after);
+ }
+ return { steps: ordered.filter((step) => !reached.has(step)), truncated };
+ }
+
+ private orderOf(step: string): number {
+ return this.nodes.get(step)?.order ?? -1;
+ }
+}
+
+/**
+ * Whether a promise has settled, and how, read without attaching a handler.
+ *
+ * Attaching one would mark an author's rejected aggregate as handled and hide
+ * an unhandled rejection the process would otherwise report, so the state is
+ * read the one public way that has no such effect: Node's own inspection.
+ * `customInspect`, `getters` and `showProxy` keep author code and proxy traps
+ * from running; `depth: 0` and the length caps keep the rendered value small.
+ * Every option the match depends on is set here, because `inspect` fills the
+ * rest from `util.inspect.defaultOptions`, which any code in the process can
+ * change — `colors: true` there would wrap `` in ANSI escapes.
+ */
+export function promiseOutcome(promise: Promise): Outcome {
+ const shown = inspect(promise, {
+ depth: 0, customInspect: false, getters: false, showProxy: true, colors: false,
+ compact: 3, sorted: false, maxArrayLength: 0, maxStringLength: 0, breakLength: Infinity,
+ });
+ const state = /^[^{]*\{\s*<(pending|rejected)>/u.exec(shown)?.[1];
+ return state === 'pending' ? 'pending' : state === 'rejected' ? 'rejected' : 'fulfilled';
+}
diff --git a/packages/sdk/src/authored-step-index.ts b/packages/sdk/src/authored-step-index.ts
index e41ac62d8..3ed6de0fc 100644
--- a/packages/sdk/src/authored-step-index.ts
+++ b/packages/sdk/src/authored-step-index.ts
@@ -57,6 +57,40 @@ export interface AuthoredStepRecord {
* `` — saying which is the difference between an index and a guess.
*/
readonly kernelStep?: string;
+ /**
+ * Human name for the step: the author-chosen `f.agent` or `f.hook` name,
+ * whole or not at all (`displayLabel`).
+ */
+ readonly label?: string;
+ /**
+ * The steps this one causally waited for, transitively reduced
+ * (`authored-step-graph.ts`). Absent means none were found, not unknown.
+ */
+ readonly after?: readonly string[];
+ /** Present when `after` is incomplete: the walk hit its limit or the list its cap. */
+ readonly afterTruncated?: true;
+}
+
+/** The graph fields one authored step carries on every record about it. */
+export type AuthoredStepEdges = Pick;
+
+/** Most predecessors one record lists; the rest are marked `afterTruncated`. */
+export const MAX_AFTER = 32;
+
+/** Longest label a record carries. A longer one is omitted, never cut. */
+export const MAX_LABEL = 256;
+
+/**
+ * A label, whole, or nothing.
+ *
+ * A label is displayed and persisted by Cloud after its redactor runs, and the
+ * redactor matches whole secret values: a label cut mid-secret would arrive as
+ * an unredactable prefix. So a label is never truncated — one over the bound
+ * is dropped — and only author-chosen names are labels at all. Commands are
+ * not: they carry literal tokens and URLs, so `f.run` has no label.
+ */
+export function displayLabel(label: string | undefined): { readonly label?: string } {
+ return label === undefined || label.trim() === '' || label.length > MAX_LABEL ? {} : { label };
}
/**
@@ -85,6 +119,10 @@ export async function recordAuthoredChild(
...(record.completionReason === undefined ? {} : { completionReason: bound(record.completionReason) }),
...(record.kernelStep === undefined || record.kernelStep === record.step
? {} : { kernelStep: bound(record.kernelStep) }),
+ ...displayLabel(record.label),
+ ...(record.after === undefined ? {} : { after: record.after.slice(0, MAX_AFTER).map(bound) }),
+ ...(record.afterTruncated === true || (record.after?.length ?? 0) > MAX_AFTER
+ ? { afterTruncated: true } : {}),
});
}
@@ -149,7 +187,12 @@ function isAuthoredStepRecord(value: unknown): value is AuthoredStepRecord {
&& typeof record.runId === 'string' && record.runId.length > 0
&& (record.state === 'admitted' || record.state === 'completed')
&& (record.completionReason === undefined || typeof record.completionReason === 'string')
- && (record.kernelStep === undefined || typeof record.kernelStep === 'string');
+ && (record.kernelStep === undefined || typeof record.kernelStep === 'string')
+ && (record.label === undefined || (typeof record.label === 'string' && record.label.length <= MAX_LABEL))
+ && (record.after === undefined || (Array.isArray(record.after)
+ && record.after.length <= MAX_AFTER
+ && record.after.every((step) => typeof step === 'string' && step.length > 0)))
+ && (record.afterTruncated === undefined || record.afterTruncated === true);
}
function bound(value: string): string {
diff --git a/packages/sdk/src/authored-step-output.ts b/packages/sdk/src/authored-step-output.ts
index c0797527c..308a82a35 100644
--- a/packages/sdk/src/authored-step-output.ts
+++ b/packages/sdk/src/authored-step-output.ts
@@ -5,7 +5,7 @@ import type { CompletionReason as ProtocolCompletionReason, RunCompletionReason
import type { AuthoredFlowJournalStep } from './authored-flow-executor.js';
import type { StepFailedDetails } from './failure-kinds.js';
import { inspectionHint, renderInspection, renderStepEvidence, stepFailureDetails } from './cli/step-failure.js';
-import { alsoRecord, recordAuthoredChild } from './authored-step-index.js';
+import { alsoRecord, recordAuthoredChild, type AuthoredStepEdges } from './authored-step-index.js';
/**
* What an authored operation needs in order to leave readable evidence:
@@ -17,6 +17,8 @@ import { alsoRecord, recordAuthoredChild } from './authored-step-index.js';
export interface AuthoredStepContext {
readonly rootRunId?: string;
readonly dataDir?: string;
+ /** The step's label and causal predecessors, carried on every record about it. */
+ readonly stepEdges?: (step: string) => AuthoredStepEdges | undefined;
}
/**
@@ -58,7 +60,8 @@ export async function readCompletedStepOutput(
}
const reason = completed.payload.completionReason;
- journalSteps.push(Object.freeze({ id: stepId, runId, completionReason: reason }));
+ const edges = context.stepEdges?.(stepId);
+ journalSteps.push(Object.freeze({ id: stepId, runId, completionReason: reason, ...edges }));
if (reason !== 'success') {
let message = `journal step "${stepId}" completed with ${reason}`;
let details: StepFailedDetails | undefined;
@@ -77,11 +80,12 @@ export async function readCompletedStepOutput(
message += await alsoRecord(journal, context.rootRunId, {
step: stepId, runId, state: 'completed', completionReason: reason,
...(details?.stepId === undefined ? {} : { kernelStep: details.stepId }),
+ ...edges,
});
throw new AuthoredFlowExecutionError('step_failed', message, reason, runId, { ...details, ...where });
}
await recordAuthoredChild(journal, context.rootRunId, {
- step: stepId, runId, state: 'completed', completionReason: reason,
+ step: stepId, runId, state: 'completed', completionReason: reason, ...edges,
});
return completed.payload.output;
}
diff --git a/packages/sdk/src/authored-worker-step.ts b/packages/sdk/src/authored-worker-step.ts
index d7793a352..a42244ec6 100644
--- a/packages/sdk/src/authored-worker-step.ts
+++ b/packages/sdk/src/authored-worker-step.ts
@@ -24,7 +24,7 @@ export function authoredWorkerRunner(
definition: { name: string }, journal: JournalClient, flowPath: string,
journalSteps: AuthoredFlowJournalStep[], waitOptions: RunLifecycleOptions,
localAgentStream?: string, budget?: AuthoredBudget, headerBudget?: unknown,
- rootRunId?: string, workerCapacity?: number,
+ rootRunId?: string, workerCapacity?: number, stepEdges?: AuthoredStepContext['stepEdges'],
) {
// 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.
@@ -33,6 +33,7 @@ export function authoredWorkerRunner(
const context: AuthoredStepContext = {
...(rootRunId === undefined ? {} : { rootRunId }),
...(waitOptions.dataDir === undefined ? {} : { dataDir: waitOptions.dataDir }),
+ ...(stepEdges === undefined ? {} : { stepEdges }),
};
async function run(step: StepSpec): Promise {
const id = step.id;
@@ -61,7 +62,9 @@ export function authoredWorkerRunner(
// Written BEFORE the wait, not after it. An agent runs for as long as its
// lease allows; if this process dies mid-step, this record is the only
// thing that still names the child run holding the evidence.
- await recordAuthoredChild(journal, rootRunId, { step: id, runId: outcome.run_id, state: 'admitted' });
+ await recordAuthoredChild(journal, rootRunId, {
+ step: id, runId: outcome.run_id, state: 'admitted', ...stepEdges?.(id),
+ });
// Reuse the declarative CLI's own wait/classification (cli/run.ts) rather
// than a hand-rolled poll: `step.completed` and the run's own terminal
// state are appended as two SEPARATE actions (kernel/relayflowd-core/src/machine.rs
@@ -101,12 +104,13 @@ export function authoredWorkerRunner(
// a manufactured completion.
if (isSurfaceCompletionReason(details?.completionReason)) {
journalSteps.push(Object.freeze({
- id, runId: outcome.run_id, completionReason: details.completionReason,
+ id, runId: outcome.run_id, completionReason: details.completionReason, ...stepEdges?.(id),
}));
message += await alsoRecord(journal, rootRunId, {
step: id, runId: outcome.run_id, state: 'completed',
completionReason: details.completionReason,
...(details.stepId === undefined ? {} : { kernelStep: details.stepId }),
+ ...stepEdges?.(id),
});
}
throw new AuthoredFlowExecutionError(
@@ -242,7 +246,9 @@ export function authoredDeterministicRunner(
}));
const admissionKey = authoredChildAdmissionKey(rootRunId, id);
const consume = async (outcome: import('./protocol.js').RunOutcome): Promise => {
- await recordAuthoredChild(journal, rootRunId, { step: id, runId: outcome.run_id, state: 'admitted' });
+ await recordAuthoredChild(journal, rootRunId, {
+ step: id, runId: outcome.run_id, state: 'admitted', ...context.stepEdges?.(id),
+ });
return readSuccessfulOutput(journal, outcome, id, journalSteps, context);
};
if (terminal) return consume(await journal.runStart(spec, undefined, admissionKey));
diff --git a/packages/sdk/src/journal-client.ts b/packages/sdk/src/journal-client.ts
index 029cf03a3..2b08178b2 100644
--- a/packages/sdk/src/journal-client.ts
+++ b/packages/sdk/src/journal-client.ts
@@ -61,6 +61,8 @@ export class JournalProtocolError extends Error {
export class JournalClient extends EventEmitter {
private socket: Socket | null = null;
+ /** Why the connection ended, so a later "not connected" names its cause rather than hiding it. */
+ private disconnectCause: Error | undefined;
private buffer = '';
private readonly pending = new Map();
private readonly requestTimeoutMs: number;
@@ -103,18 +105,31 @@ export class JournalClient extends EventEmitter {
socket.once('connect', () => {
clearTimeout(timer);
socket.removeListener('error', onError);
- socket.on('error', (err) => this.failAll(err));
+ socket.on('error', (err) => { this.disconnectCause ??= err; this.failAll(err); });
socket.on('data', (chunk) => this.onData(chunk));
- socket.on('close', () => this.failAll(new Error('journal client: connection closed')));
+ socket.on('close', () => {
+ const closed = new Error('journal client: connection closed');
+ this.disconnectCause ??= closed;
+ this.failAll(closed);
+ });
+ this.disconnectCause = undefined;
this.socket = socket;
resolve();
});
});
}
- /** Close the connection and reject any pending requests. */
- close(): void {
- this.failAll(new Error('journal client: closed by caller'));
+ /**
+ * Close the connection and reject any pending requests. A `cause` (for
+ * example the worker failure that forced the close) is carried into every
+ * later "not connected" rejection.
+ */
+ close(cause?: unknown): void {
+ const closed = cause === undefined
+ ? new Error('journal client: closed by caller')
+ : new Error(`journal client: closed after ${cause instanceof Error ? cause.message : String(cause)}`, { cause });
+ this.disconnectCause ??= closed;
+ this.failAll(closed);
this.socket?.destroy();
this.socket = null;
this.buffer = '';
@@ -170,7 +185,11 @@ export class JournalClient extends EventEmitter {
): Promise {
return new Promise((resolve, reject) => {
if (!this.socket || this.socket.destroyed) {
- reject(new Error(`journal client: not connected (${verb})`));
+ const cause = this.disconnectCause;
+ reject(new Error(
+ `journal client: not connected (${verb})${cause === undefined ? '' : `: ${cause.message}`}`,
+ cause === undefined ? undefined : { cause },
+ ));
return;
}
const id = randomUUID();
diff --git a/packages/sdk/src/local-agent.ts b/packages/sdk/src/local-agent.ts
index 0c39c29a9..a836fecb6 100644
--- a/packages/sdk/src/local-agent.ts
+++ b/packages/sdk/src/local-agent.ts
@@ -28,7 +28,8 @@ export async function attachLocalAgent(
pins: { workspace: [], streams: [{ stream, read_offset: 0 }] },
});
let failure: unknown;
- worker.on('error', error => { failure = error; client.close(); });
+ // The cause travels with the close, so the flow's next request names it.
+ worker.on('error', error => { failure = error; client.close(error); });
await worker.attach();
return {
stream,
diff --git a/packages/sdk/tests/agent-transcript-live.test.ts b/packages/sdk/tests/agent-transcript-live.test.ts
index f94f912f1..341fcb0cd 100644
--- a/packages/sdk/tests/agent-transcript-live.test.ts
+++ b/packages/sdk/tests/agent-transcript-live.test.ts
@@ -144,6 +144,8 @@ describe('the transcript digest through the built CLI, a real daemon and the loc
expect(await readAuthoredStepIndex(journal, report.rootRunId)).toEqual([{
index: 'relayflows.authored-step.v1', step: `${kind}-1`, runId: report.runId,
state: 'completed', completionReason: failure!.completionReason,
+ // `f.agent`'s name is the step's label in the run's DAG; `f.llm` has none.
+ ...(kind === 'agent' ? { label: 'prober' } : {}),
}]);
} finally { journal.close(); }
}, 120_000);
diff --git a/packages/sdk/tests/authored-step-graph-live.test.ts b/packages/sdk/tests/authored-step-graph-live.test.ts
new file mode 100644
index 000000000..33fb32212
--- /dev/null
+++ b/packages/sdk/tests/authored-step-graph-live.test.ts
@@ -0,0 +1,94 @@
+import { afterEach, describe, expect, it } from 'vitest';
+import { flow } from '@relayflows/surface';
+import { executeAuthoredFlow } from '../src/authored-flow-executor.js';
+import { attachLocalAgent } from '../src/local-agent.js';
+import { AUTHORED_STEP_STREAM, readAuthoredStepIndex } from '../src/authored-step-index.js';
+import type { JournalClient } from '../src/journal-client.js';
+import { chainFixture } from './flow-chain-fixture.js';
+
+const closes: Array<() => Promise> = [];
+// Every close runs even when an earlier one rejects, so a failed agent close never leaks the daemon.
+afterEach(async () => {
+ const errors: unknown[] = [];
+ for (const close of closes.splice(0).reverse()) {
+ try { await close(); } catch (error) { errors.push(error); }
+ }
+ if (errors.length > 0) throw errors[0];
+});
+
+/** A mutable root the index is appended to, as in authored-run-failure-evidence.test.ts. */
+async function openRoot(journal: JournalClient): Promise {
+ const outcome = await journal.runStart({
+ version: '0.1.0',
+ name: 'graph-root',
+ steps: [{
+ id: 'authored-root', type: 'agent', instruction: '{}',
+ surfaces: { streams: [{ stream: 'graph-root-stream' }] },
+ recovery_mode: 'reset', max_iterations: 8,
+ }],
+ } as never);
+ return outcome.run_id;
+}
+
+describe('the authored step DAG through the live kernel', () => {
+ it('carries labels and predecessors on every index record and journal step, ids unchanged', async () => {
+ const fixture = chainFixture('drafted');
+ closes.push(() => fixture.close());
+ const journal = await fixture.connect();
+ const agent = await attachLocalAgent(journal);
+ closes.push(() => agent.close());
+ const rootRunId = await openRoot(journal);
+
+ const handle = flow('graph-live', async (f) => {
+ await f.run('printf plan');
+ // Fan out on deterministic steps: the local agent worker serves one lease
+ // at a time, so parallel agents would park rather than run.
+ await Promise.all([f.run('printf left'), f.run('printf right')]);
+ await f.agent('writer', { task: 'write it' });
+ await f.run('printf done');
+ f.done('success');
+ });
+ const result = await executeAuthoredFlow(handle, journal, undefined, {
+ rootRunId, flowPath: fixture.flowPath, localAgentStream: agent.stream, dataDir: fixture.data,
+ });
+
+ // Step identity is the kernel's and the admission key's: unchanged.
+ expect(result.journalSteps.map((step) => step.id).sort()).toEqual(
+ ['agent-4', 'complete-6', 'run-1', 'run-2', 'run-3', 'run-5'],
+ );
+ const expected = {
+ // `f.run` has no label: a command is not a safe display name.
+ 'run-1': {},
+ 'run-2': { after: ['run-1'] },
+ 'run-3': { after: ['run-1'] },
+ 'agent-4': { label: 'writer', after: ['run-2', 'run-3'] },
+ 'run-5': { after: ['agent-4'] },
+ };
+ const journaled = Object.fromEntries(result.journalSteps.map((step) => [step.id, step]));
+ for (const [step, edges] of Object.entries(expected)) {
+ expect(journaled[step]).toEqual({ id: step, runId: expect.any(String), completionReason: 'success', ...edges });
+ }
+ // The terminal marker is not an authored step and carries no graph fields.
+ expect(journaled['complete-6']).not.toHaveProperty('label');
+ expect(journaled['complete-6']).not.toHaveProperty('after');
+
+ // Both the `admitted` and the `completed` record carry the fields, so a
+ // reader watching a run in flight can draw the DAG before anything finishes.
+ const raw = (await journal.streamRead(rootRunId, AUTHORED_STEP_STREAM, 0, 1000)).messages
+ .map((message) => (message as { message?: unknown }).message ?? message) as Array>;
+ for (const [step, edges] of Object.entries(expected)) {
+ const records = raw.filter((record) => record['step'] === step);
+ expect(records.map((record) => record['state']).sort()).toEqual(['admitted', 'completed']);
+ for (const record of records) expect(record).toMatchObject(edges);
+ }
+
+ const index = await readAuthoredStepIndex(journal, rootRunId);
+ expect(Object.fromEntries(index.map((record) => [record.step, record]))).toEqual(
+ // The terminal marker is indexed too, with no graph fields of its own.
+ Object.fromEntries(Object.entries({ ...expected, 'complete-6': {} }).map(([step, edges]) => [step, {
+ index: 'relayflows.authored-step.v1', step, runId: journaled[step]!.runId,
+ state: 'completed', completionReason: 'success', ...edges,
+ }])),
+ );
+ }, 60_000);
+});
diff --git a/packages/sdk/tests/authored-step-graph.test.ts b/packages/sdk/tests/authored-step-graph.test.ts
new file mode 100644
index 000000000..362a54cf9
--- /dev/null
+++ b/packages/sdk/tests/authored-step-graph.test.ts
@@ -0,0 +1,409 @@
+import { inspect } from 'node:util';
+import { describe, expect, it } from 'vitest';
+import type { Step } from '@relayflows/surface';
+import {
+ AuthoredFlowOperation,
+ verifyAuthoredOperations,
+} from '../src/authored-flow-operation.js';
+import { AuthoredFlowLifecycle } from '../src/authored-flow-lifecycle.js';
+import { AuthoredStepGraph, promiseOutcome } from '../src/authored-step-graph.js';
+
+type Op = (label: string, delayMs?: number, fails?: boolean, polls?: number) => Step;
+
+const sleep = (ms: number): Promise => new Promise((resolve) => setTimeout(resolve, ms));
+
+/**
+ * Run a body against a real lifecycle, the way the executor does, and return
+ * each step's predecessors by LABEL so the assertions read as the DAG drawn.
+ */
+async function graphOf(body: (op: Op) => Promise, { stepsFail = false } = {}): Promise> {
+ const lifecycle = new AuthoredFlowLifecycle();
+ const operations: AuthoredFlowOperation[] = [];
+ const op: Op = (label, delayMs = 0, fails = false, polls = 0) => {
+ const operation = new AuthoredFlowOperation(
+ `agent-${operations.length + 1}`, 'agent', () => undefined,
+ async () => {
+ // A long-running step's own promise chain, as `classifyOutcome`'s
+ // journal polling builds one: every poll is a fresh await, and every
+ // few a timer, all inside the lifecycle's scope.
+ for (let poll = 0; poll < polls; poll++) {
+ if (poll % 1_000 === 0) await sleep(0);
+ else await Promise.resolve(poll);
+ }
+ if (delayMs > 0) await sleep(delayMs);
+ if (fails) throw new Error(`${label} failed`);
+ return label;
+ },
+ lifecycle,
+ { label },
+ );
+ operations.push(operation as AuthoredFlowOperation);
+ return operation.step;
+ };
+ try {
+ await lifecycle.runBody(async () => {
+ await body(op);
+ lifecycle.markCompletion();
+ });
+ // A failed step fails the flow even when the body caught it; the edges it
+ // drew were recorded at invocation and are what is asserted here.
+ const verified = verifyAuthoredOperations('step-graph', operations, lifecycle);
+ if (stepsFail) await expect(verified).rejects.toThrow(/failed/u);
+ else await verified;
+ } finally {
+ lifecycle.close();
+ }
+ const labels = new Map(operations.map((operation) => [operation.id, operation.edges?.label ?? '?']));
+ return Object.fromEntries(operations.map((operation) => {
+ const edges = lifecycle.stepEdges(operation.id);
+ // The id is the kernel identity and is minted exactly as before.
+ expect(operation.edges).toBe(edges);
+ return [edges?.label ?? operation.id, {
+ // `after` is in step-creation order, which timing decides; compare as a set.
+ after: (edges?.after ?? []).map((step) => labels.get(step) ?? step).sort(),
+ truncated: edges?.afterTruncated === true,
+ }];
+ }));
+}
+
+const after = (...labels: string[]) => ({ after: labels.sort(), truncated: false });
+
+describe('the authored step DAG', () => {
+ it('draws no edge between siblings of a parallel fan-out', async () => {
+ const graph = await graphOf(async (op) => {
+ const a = op('a', 15);
+ const b = op('b', 5);
+ const c = op('c');
+ await Promise.all([a, b, c]);
+ });
+ expect(graph).toEqual({ a: after(), b: after(), c: after() });
+ });
+
+ it('draws a sequential chain as a chain, not as a fan-in from every earlier step', async () => {
+ const graph = await graphOf(async (op) => {
+ await op('a');
+ await op('b');
+ await op('c');
+ });
+ expect(graph).toEqual({ a: after(), b: after('a'), c: after('b') });
+ });
+
+ it('makes a step after `await Promise.all(deps)` depend on every dep, whatever settled last', async () => {
+ const graph = await graphOf(async (op) => {
+ const deps = [op('slow', 20), op('fast'), op('middle', 5)];
+ await Promise.all(deps);
+ await op('join');
+ });
+ expect(graph['join']).toEqual(after('slow', 'fast', 'middle'));
+ });
+
+ it('expands Promise.allSettled the same way', async () => {
+ const graph = await graphOf(async (op) => {
+ await Promise.allSettled([op('x', 10), op('y')]);
+ await op('z');
+ });
+ expect(graph['z']).toEqual(after('x', 'y'));
+ });
+
+ it('follows only the winner of a race', async () => {
+ const graph = await graphOf(async (op) => {
+ const slow = op('slow', 30);
+ await Promise.race([op('fast'), slow]);
+ await op('next');
+ await slow;
+ });
+ expect(graph['next']).toEqual(after('fast'));
+ });
+
+ it('makes a step after a caught Promise.all rejection depend on the rejecting member only', async () => {
+ // `all` rejects on the first rejection; the slow member was still running
+ // when the recovery step started, so it is not a predecessor.
+ const graph = await graphOf(async (op) => {
+ const slow = op('slow', 30);
+ try {
+ await Promise.all([op('boom', 0, true), slow]);
+ } catch { /* the author recovers */ }
+ await op('recover');
+ await slow;
+ }, { stepsFail: true });
+ expect(graph['recover']).toEqual(after('boom'));
+ });
+
+ it('makes a step after a fulfilled Promise.any depend on the winner only', async () => {
+ const graph = await graphOf(async (op) => {
+ const slow = op('slow', 30);
+ await Promise.any([op('fast'), slow]);
+ await op('next');
+ await slow;
+ });
+ expect(graph['next']).toEqual(after('fast'));
+ });
+
+ it('makes a step after a caught Promise.any rejection depend on every member', async () => {
+ // `any` rejects only once every member has rejected, whichever was last.
+ const graph = await graphOf(async (op) => {
+ try {
+ await Promise.any([op('first', 15, true), op('second', 0, true)]);
+ } catch { /* AggregateError: the author recovers */ }
+ await op('recover');
+ }, { stepsFail: true });
+ expect(graph['recover']).toEqual(after('first', 'second'));
+ });
+
+ it('expands Promise.allSettled whatever its members did', async () => {
+ const graph = await graphOf(async (op) => {
+ await Promise.allSettled([op('ok', 10), op('bad', 0, true)]);
+ await op('next');
+ }, { stepsFail: true });
+ expect(graph['next']).toEqual(after('ok', 'bad'));
+ });
+
+ it('reduces a diamond to its direct predecessors', async () => {
+ const graph = await graphOf(async (op) => {
+ await op('root');
+ const left = op('left', 10);
+ const right = op('right');
+ await Promise.all([left, right]);
+ await op('sink');
+ });
+ expect(graph).toEqual({
+ root: after(), left: after('root'), right: after('root'), sink: after('left', 'right'),
+ });
+ });
+
+ it('sees through async wrapper functions', async () => {
+ const graph = await graphOf(async (op) => {
+ const first = (async () => {
+ await op('a1', 10);
+ await op('a2');
+ })();
+ const second = (async () => {
+ await op('b1');
+ })();
+ await Promise.all([first, second]);
+ await op('after-both');
+ });
+ expect(graph).toEqual({
+ a1: after(), a2: after('a1'), b1: after(), 'after-both': after('a2', 'b1'),
+ });
+ });
+
+ it('draws the task-graph pattern: a map of promises, each awaiting its deps', async () => {
+ // The examples/task-graph shape: every subtask is started at once, awaits
+ // the promises of the subtasks it depends on, then runs a chain.
+ const plan = [
+ { id: 'schema', deps: [] as string[], delayMs: 10 },
+ { id: 'api', deps: ['schema'], delayMs: 0 },
+ { id: 'ui', deps: ['schema'], delayMs: 5 },
+ { id: 'docs', deps: [], delayMs: 0 },
+ { id: 'e2e', deps: ['api', 'ui', 'docs'], delayMs: 0 },
+ ];
+ const graph = await graphOf(async (op) => {
+ const done = new Map>();
+ for (const task of plan) {
+ done.set(task.id, (async () => {
+ await Promise.all(task.deps.map((dep) => done.get(dep)));
+ await op(`${task.id}:setup`, task.delayMs);
+ await op(`${task.id}:agent`);
+ await op(`${task.id}:verify`);
+ })());
+ }
+ await Promise.all(done.values());
+ });
+ expect(graph).toEqual({
+ 'schema:setup': after(), 'schema:agent': after('schema:setup'), 'schema:verify': after('schema:agent'),
+ 'api:setup': after('schema:verify'), 'api:agent': after('api:setup'), 'api:verify': after('api:agent'),
+ 'ui:setup': after('schema:verify'), 'ui:agent': after('ui:setup'), 'ui:verify': after('ui:agent'),
+ 'docs:setup': after(), 'docs:agent': after('docs:setup'), 'docs:verify': after('docs:agent'),
+ 'e2e:setup': after('api:verify', 'ui:verify', 'docs:verify'),
+ 'e2e:agent': after('e2e:setup'), 'e2e:verify': after('e2e:agent'),
+ });
+ });
+
+ it('does not walk a long-running step\'s own polling chain to find its dependents\' edges', async () => {
+ // 30 000 polls is three times the walk limit. If a walk entered the
+ // step's internal chain instead of stopping at the step, `next` would
+ // come back truncated or with a wrong predecessor.
+ const graph = await graphOf(async (op) => {
+ const long = op('long', 0, false, 30_000);
+ await op('quick');
+ await op('after-quick');
+ await long;
+ await op('join');
+ await op('next');
+ });
+ expect(graph).toEqual({
+ long: after(), quick: after(), 'after-quick': after('quick'),
+ join: after('after-quick', 'long'), next: after('join'),
+ });
+ }, 30_000);
+
+ it('does not walk a long-running step\'s chain when a wrapper awaits it', async () => {
+ const graph = await graphOf(async (op) => {
+ const branch = (async () => {
+ await op('long', 0, false, 30_000);
+ await op('after-long');
+ })();
+ await op('side');
+ await branch;
+ await op('join');
+ });
+ expect(graph).toEqual({
+ long: after(), 'after-long': after('long'), side: after(), join: after('after-long', 'side'),
+ });
+ }, 30_000);
+
+ it('caps a wide fan-in at 32 predecessors and says it did', async () => {
+ const graph = await graphOf(async (op) => {
+ await Promise.all(Array.from({ length: 40 }, (_, i) => op(`w${i}`)));
+ await op('join');
+ });
+ expect(graph['join']!.after).toHaveLength(32);
+ expect(graph['join']!.truncated).toBe(true);
+ });
+
+ it('keeps a long sequential body linear: 300 steps, each after exactly its predecessor', async () => {
+ const graph = await graphOf(async (op) => {
+ for (let i = 0; i < 300; i++) {
+ await op(`s${i}`);
+ // Ordinary awaits between steps grow the promise graph, not the edges.
+ for (let j = 0; j < 20; j++) await null;
+ }
+ });
+ expect(graph['s0']).toEqual(after());
+ for (let i = 1; i < 300; i++) expect(graph[`s${i}`]).toEqual(after(`s${i - 1}`));
+ });
+});
+
+describe('the walk limit', () => {
+ it('stops at the node cap and marks the result truncated instead of walking on', () => {
+ // A synthetic chain of 50 000 promises between the step and its caller:
+ // with a cap of 1 000 the walk must give up, visit no more than the cap,
+ // and say the answer is incomplete.
+ let visits = 0;
+ const graph = new AuthoredStepGraph((asyncId) => {
+ visits += 1;
+ return asyncId > 1 ? [asyncId - 1] : [];
+ }, 1_000);
+ graph.registerStep('agent-1', 1, 0);
+ visits = 0;
+ const edges = graph.registerStep('agent-2', 50_001, 50_000, 'far');
+ expect(edges).toEqual({ label: 'far', afterTruncated: true });
+ expect(visits).toBeLessThanOrEqual(1_000);
+ });
+
+ it('finds the predecessor when it is within the cap', () => {
+ const graph = new AuthoredStepGraph((asyncId) => (asyncId > 1 ? [asyncId - 1] : []), 1_000);
+ graph.registerStep('agent-1', 1, 0);
+ expect(graph.registerStep('agent-2', 600, 500)).toEqual({ after: ['agent-1'] });
+ });
+});
+
+describe('truncation through the reduction', () => {
+ // A chain of promise ids ending in a step promise: 1 is `s1`'s promise, and
+ // the walk from 2_000 must pass 1_000 ids to reach it.
+ const chain = (asyncId: number): number[] => {
+ if (asyncId === 5_000) return [1, 3_000];
+ return asyncId > 1_000 && asyncId <= 2_000 ? [asyncId - 1] : [];
+ };
+
+ it('marks a join truncated when a predecessor it reduces against has an incomplete list', () => {
+ const graph = new AuthoredStepGraph(chain, 100);
+ graph.registerStep('s1', 1, 0);
+ // `s2` is invoked 1 000 ids after `s1`: its walk gives up and it cannot
+ // say whether `s1` is its ancestor.
+ expect(graph.registerStep('s2', 3_000, 2_000)).toEqual({ afterTruncated: true });
+ // A join of both may therefore keep `s1` when it is really an ancestor of
+ // `s2`, and must say that its list is not known to be reduced.
+ expect(graph.registerStep('s3', 6_000, 5_000)).toEqual({ after: ['s1', 's2'], afterTruncated: true });
+ });
+
+ it('does not mark a join whose predecessors all have complete lists', () => {
+ const graph = new AuthoredStepGraph(chain, 10_000);
+ graph.registerStep('s1', 1, 0);
+ expect(graph.registerStep('s2', 3_000, 2_000)).toEqual({});
+ expect(graph.registerStep('s3', 6_000, 5_000)).toEqual({ after: ['s1', 's2'] });
+ });
+});
+
+describe('a wide fan-in', () => {
+ it('counts every predecessor entry it inspects against the limit, not just distinct steps', () => {
+ // 200 independent steps, 20 siblings that each wait for all of them, then
+ // a join of the siblings. Reducing the join reads 4 000 predecessor
+ // entries but reaches only 200 distinct steps — under a 1 000 limit that
+ // counted distinct steps alone, it would never stop.
+ const bases = Array.from({ length: 200 }, (_, i) => 1 + i);
+ const siblings = Array.from({ length: 20 }, (_, i) => 1_001 + i);
+ const causes = new Map();
+ siblings.forEach((_, i) => causes.set(2_001 + i, bases));
+ causes.set(5_000, siblings);
+ const graph = new AuthoredStepGraph((asyncId) => causes.get(asyncId) ?? [], 1_000);
+ bases.forEach((promise, i) => graph.registerStep(`b${i}`, promise, 0));
+ siblings.forEach((promise, i) => graph.registerStep(`s${i}`, promise, 2_001 + i));
+ expect(graph.registerStep('join', 6_000, 5_000)).toEqual({
+ after: siblings.map((_, i) => `s${i}`),
+ afterTruncated: true,
+ });
+ });
+});
+
+describe('reading a settlement without handling it', () => {
+ it('reports pending, fulfilled and rejected, including for a Promise subclass', async () => {
+ class Sub extends Promise {}
+ const rejected = Promise.reject(new Error('no'));
+ rejected.catch(() => undefined);
+ const subRejected = Sub.reject(1);
+ subRejected.catch(() => undefined);
+ await sleep(0);
+ expect(promiseOutcome(new Promise(() => undefined))).toBe('pending');
+ expect(promiseOutcome(Promise.resolve({ value: 1 }))).toBe('fulfilled');
+ expect(promiseOutcome(Promise.resolve(' in the value'))).toBe('fulfilled');
+ expect(promiseOutcome(rejected)).toBe('rejected');
+ expect(promiseOutcome(subRejected)).toBe('rejected');
+ });
+
+ it('ignores process-wide inspect defaults such as colors', async () => {
+ const rejected = Promise.reject(new Error('no'));
+ rejected.catch(() => undefined);
+ await sleep(0);
+ const defaults = { ...inspect.defaultOptions };
+ Object.assign(inspect.defaultOptions, { colors: true, compact: false, sorted: true });
+ try {
+ expect(promiseOutcome(rejected)).toBe('rejected');
+ expect(promiseOutcome(new Promise(() => undefined))).toBe('pending');
+ } finally {
+ Object.assign(inspect.defaultOptions, defaults);
+ }
+ });
+
+ it('runs no author inspection hook or proxy trap on the settled value', () => {
+ let ran = 0;
+ const value = new Proxy({ [Symbol.for('nodejs.util.inspect.custom')]: () => { ran += 1; return 'x'; } }, {
+ get(target, key) { ran += 1; return Reflect.get(target, key); },
+ ownKeys(target) { ran += 1; return Reflect.ownKeys(target); },
+ });
+ // Resolve with a non-thenable holder: resolving with the proxy itself
+ // would read its `then`, which is the runtime's doing, not the probe's.
+ const settled = Promise.resolve([value]);
+ return settled.then(() => {
+ ran = 0;
+ expect(promiseOutcome(settled)).toBe('fulfilled');
+ expect(ran).toBe(0);
+ });
+ });
+});
+
+describe('step labels', () => {
+ it('keeps a label whole or omits it, never cuts it', () => {
+ // A cut label could end mid-secret, where Cloud's whole-value redactor
+ // cannot match it. Over the bound, there is no label at all.
+ const graph = new AuthoredStepGraph(() => []);
+ expect(graph.registerStep('agent-1', 1, 0, 'writer')).toEqual({ label: 'writer' });
+ expect(graph.registerStep('agent-2', 2, 0, 'w'.repeat(256))).toEqual({ label: 'w'.repeat(256) });
+ expect(graph.registerStep('agent-3', 3, 0, 'w'.repeat(257))).toEqual({});
+ expect(graph.registerStep('agent-4', 4, 0, ' ')).toEqual({});
+ });
+});
diff --git a/packages/sdk/tests/authored-step-index.test.ts b/packages/sdk/tests/authored-step-index.test.ts
index f4042bbef..73eac560a 100644
--- a/packages/sdk/tests/authored-step-index.test.ts
+++ b/packages/sdk/tests/authored-step-index.test.ts
@@ -190,3 +190,69 @@ describe('a failed index append', () => {
})).toBe('');
});
});
+
+describe('the DAG fields on an index record', () => {
+ it('round-trips a label and its predecessors on admitted and completed records', async () => {
+ const { journal, stored, key } = streams();
+ const edges = { label: 'writer', after: ['run-1', 'run-2'] };
+ await recordAuthoredChild(journal, 'root-1', { step: 'agent-3', runId: 'c', state: 'admitted', ...edges });
+ await recordAuthoredChild(journal, 'root-1', {
+ step: 'agent-3', runId: 'c', state: 'completed', completionReason: 'success', ...edges,
+ });
+
+ expect(stored.get(key('root-1', AUTHORED_STEP_STREAM))).toEqual([
+ { index: 'relayflows.authored-step.v1', step: 'agent-3', runId: 'c', state: 'admitted', ...edges },
+ {
+ index: 'relayflows.authored-step.v1', step: 'agent-3', runId: 'c', state: 'completed',
+ completionReason: 'success', ...edges,
+ },
+ ]);
+ expect(await readAuthoredStepIndex(journal, 'root-1')).toEqual([{
+ index: 'relayflows.authored-step.v1', step: 'agent-3', runId: 'c',
+ state: 'completed', completionReason: 'success', ...edges,
+ }]);
+ });
+
+ it('writes no DAG fields for a step that has none, so old readers see the old shape', async () => {
+ const { journal, stored, key } = streams();
+ await recordAuthoredChild(journal, 'root-1', { step: 'run-1', runId: 'a', state: 'admitted' });
+ expect(stored.get(key('root-1', AUTHORED_STEP_STREAM))).toEqual([{
+ index: 'relayflows.authored-step.v1', step: 'run-1', runId: 'a', state: 'admitted',
+ }]);
+ });
+
+ it('omits an oversized label rather than cutting it, and caps the predecessor list', async () => {
+ const { journal, stored, key } = streams();
+ const after = Array.from({ length: 40 }, (_, i) => `run-${i + 1}`);
+ await recordAuthoredChild(journal, 'root-1', {
+ step: 'agent-41', runId: 'x', state: 'admitted', label: 'l'.repeat(4000), after,
+ });
+
+ // A cut label could end mid-secret, where a whole-value redactor cannot
+ // match it; so no prefix of it is ever written.
+ expect(JSON.stringify(stored.get(key('root-1', AUTHORED_STEP_STREAM)))).not.toContain('lll');
+ const [record] = await readAuthoredStepIndex(journal, 'root-1');
+ expect(record).not.toHaveProperty('label');
+ expect(record!.after).toEqual(after.slice(0, 32));
+ expect(record!.afterTruncated).toBe(true);
+ });
+
+ it('skips a record whose DAG fields are malformed rather than trusting them', async () => {
+ const { journal } = streams();
+ const base = { index: 'relayflows.authored-step.v1', runId: 'a', state: 'admitted' };
+ await journal.streamAppend('root-1', AUTHORED_STEP_STREAM, { ...base, step: 'run-1', label: 7 });
+ await journal.streamAppend('root-1', AUTHORED_STEP_STREAM, { ...base, step: 'run-0', label: 'l'.repeat(257) });
+ await journal.streamAppend('root-1', AUTHORED_STEP_STREAM, { ...base, step: 'run-2', after: 'run-1' });
+ await journal.streamAppend('root-1', AUTHORED_STEP_STREAM, { ...base, step: 'run-3', after: [''] });
+ await journal.streamAppend('root-1', AUTHORED_STEP_STREAM, { ...base, step: 'run-4', afterTruncated: 'yes' });
+ // Longer than any writer produces: the list is bounded on read as on write.
+ await journal.streamAppend('root-1', AUTHORED_STEP_STREAM, {
+ ...base, step: 'run-5', after: Array.from({ length: 33 }, (_, i) => `run-${i}`),
+ });
+ await journal.streamAppend('root-1', AUTHORED_STEP_STREAM, {
+ ...base, step: 'run-6', after: Array.from({ length: 32 }, (_, i) => `run-${i}`),
+ });
+
+ expect((await readAuthoredStepIndex(journal, 'root-1')).map((record) => record.step)).toEqual(['run-6']);
+ });
+});
diff --git a/packages/sdk/tests/journal-client.test.ts b/packages/sdk/tests/journal-client.test.ts
index df211408a..d238fda09 100644
--- a/packages/sdk/tests/journal-client.test.ts
+++ b/packages/sdk/tests/journal-client.test.ts
@@ -445,3 +445,19 @@ describe('JournalClient: an effect election is atomic with its provider call', (
}
});
});
+
+describe('a request after the connection ended', () => {
+ it('names the cause the caller closed it with', async () => {
+ const c = new JournalClient('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/nonexistent/relayflowd.sock');
+ const boom = new Error('agent worker: stepComplete rejected');
+ c.close(boom);
+ const error = await c.runGet('run-1').then(() => undefined, (e: unknown) => e as Error);
+ expect(error?.message).toBe('journal client: not connected (run.get): journal client: closed after agent worker: stepComplete rejected');
+ expect((error?.cause as Error | undefined)?.cause).toBe(boom);
+ });
+
+ it('keeps the bare message when nothing ended it', async () => {
+ const c = new JournalClient('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/nonexistent/relayflowd.sock');
+ await expect(c.runGet('run-1')).rejects.toThrow(/^journal client: not connected \(run\.get\)$/);
+ });
+});