Skip to content

Commit 020d73f

Browse files
Relayflow Leadclaude
andcommitted
feat(sdk): record a named step DAG for authored flows
Each authored step records a human label (f.agent / f.hook name) and its direct causal predecessors (after), derived from the runtime promise graph, on its authored-steps records and the root completion's journalSteps. Step ids, admission keys and resume identity are unchanged. Consumed by the Cloud run graph (AgentWorkforce/cloud#3945). Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
1 parent 83addb0 commit 020d73f

12 files changed

Lines changed: 919 additions & 13 deletions

‎docs/SURFACE.md‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -858,6 +858,15 @@ child. Helper-provider, MCP and plugin-effect children are not yet included
858858
in this index. Once appended, the index survives process exit and is readable from
859859
the journal on disk, including after a cooperative nonzero exit.
860860

861+
Each record, and the matching root `journalSteps` entry, may also carry the
862+
step's place in the run's DAG. `after` lists the step ids it causally waited
863+
for, transitively reduced and capped at 32. `afterTruncated: true` means that
864+
list is incomplete. `label` is the author-chosen `f.agent` or `f.hook` name,
865+
kept whole or omitted when it is over 256 characters, because a cut label
866+
could end partway through a secret. `f.run` has no label, because a command
867+
can carry literal tokens and URLs. The fields are optional; a step with none
868+
writes the original record shape.
869+
861870
An authored step-failure JSON report keeps the child in `runId` and adds
862871
`rootRunId` for the durable authored root. Consumers must use `rootRunId` for
863872
the resume pointer and root index, and `runId` for the failing child's evidence.

‎packages/sdk/src/authored-flow-executor.ts‎

Lines changed: 14 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -43,6 +43,7 @@ import {
4343
verifyAuthoredOperations,
4444
} from './authored-flow-operation.js';
4545
import { AuthoredFlowLifecycle } from './authored-flow-lifecycle.js';
46+
import { displayLabel, type AuthoredStepEdges } from './authored-step-index.js';
4647
import { JournalClient } from './journal-client.js';
4748
import { PluginError } from './plugin-manifest.js';
4849
import { createHookEvaluator } from './authored-hooks.js';
@@ -77,7 +78,7 @@ type EveryRunCompletionReasonIsAcceptedByDone = Assert<
7778

7879
export { AuthoredFlowExecutionError, type AuthoredFlowExecutionErrorCode };
7980

80-
export interface AuthoredFlowJournalStep {
81+
export interface AuthoredFlowJournalStep extends AuthoredStepEdges {
8182
readonly id: string;
8283
readonly runId: string;
8384
readonly completionReason: ProtocolCompletionReason;
@@ -239,19 +240,21 @@ export async function executeAuthoredFlow<Input = undefined>(
239240
const journalSteps: AuthoredFlowJournalStep[] = [];
240241
const authoredSteps: AuthoredFlowOperation<unknown>[] = [];
241242
const lifecycle = new AuthoredFlowLifecycle();
243+
const stepEdges = (step: string): AuthoredStepEdges | undefined => lifecycle.stepEdges(step);
242244
let nextStep = 1;
243245
let requestedCompletion: LoweredCompletionReason | undefined;
244246

245247
const lowerDeterministic = authoredDeterministicRunner(
246248
definition.name, journal, journalSteps, budget, {
247249
...(options.rootRunId === undefined ? {} : { rootRunId: options.rootRunId }),
248250
...(options.dataDir === undefined ? {} : { dataDir: options.dataDir }),
251+
stepEdges,
249252
},
250253
);
251254

252255
const worker = authoredWorkerRunner(
253256
definition, journal, flowPath, journalSteps, waitOptions,
254-
localAgentStream, budget, definition.header.budget, options.rootRunId, options.workerCapacity,
257+
localAgentStream, budget, definition.header.budget, options.rootRunId, options.workerCapacity, stepEdges,
255258
);
256259

257260
/**
@@ -365,6 +368,7 @@ export async function executeAuthoredFlow<Input = undefined>(
365368
return worker.llm(id, text, undefined, llmOp.namedGate);
366369
}, onProgress),
367370
lifecycle,
371+
{},
368372
);
369373
return trackStep(authoredSteps, llmOp);
370374
}
@@ -442,13 +446,17 @@ export async function executeAuthoredFlow<Input = undefined>(
442446
() => assertOperationAllowed('run', definition.name, requestedCompletion),
443447
() => observeStep(id, 'deterministic', () => lowerDeterministic(id, command, false, leaseMs, runOp.namedGate), options.onProgress),
444448
lifecycle,
449+
// No label: a command is not a display name. It carries literal
450+
// tokens and URLs, and no prefix of it is safe to show (displayLabel).
451+
{},
445452
);
446453
return trackStep(authoredSteps, runOp);
447454
},
448455
llm: llmOperation,
449456
agent(name, options) {
450457
assertOperationAllowed('agent', definition.name, requestedCompletion);
451-
void name; // Authored headers do not yet declare reusable named agents.
458+
// Authored headers do not yet declare reusable named agents, so the name
459+
// identifies nothing to the kernel; it is the step's label in the DAG.
452460
const id = `agent-${nextStep++}`;
453461
let agentOp!: AuthoredFlowOperation<AgentResult>;
454462
agentOp = new AuthoredFlowOperation<AgentResult>(
@@ -457,6 +465,7 @@ export async function executeAuthoredFlow<Input = undefined>(
457465
() => assertOperationAllowed('agent', definition.name, requestedCompletion),
458466
() => observeStep(id, 'agent', () => worker.agent(id, options, agentOp.namedGate), onProgress),
459467
lifecycle,
468+
displayLabel(typeof name === 'string' ? name : undefined),
460469
);
461470
return trackStep(authoredSteps, agentOp);
462471
},
@@ -518,6 +527,7 @@ export async function executeAuthoredFlow<Input = undefined>(
518527
return recorded.answer;
519528
}, options.onProgress),
520529
lifecycle,
530+
{},
521531
);
522532
return trackStep(authoredSteps, humanOp);
523533
},
@@ -540,6 +550,7 @@ export async function executeAuthoredFlow<Input = undefined>(
540550
return verdict;
541551
},
542552
lifecycle,
553+
displayLabel(typeof name === 'string' ? name : undefined),
543554
));
544555
},
545556
done(reason) {

‎packages/sdk/src/authored-flow-lifecycle.ts‎

Lines changed: 45 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,8 @@
11
import { AsyncLocalStorage, executionAsyncId } from 'node:async_hooks';
22
import { AuthoredFlowExecutionError } from './authored-flow-error.js';
33
import { AuthoredPromiseGraph } from './authored-promise-graph.js';
4+
import { AuthoredStepGraph } from './authored-step-graph.js';
5+
import type { AuthoredStepEdges } from './authored-step-index.js';
46

57
type OperationToken = object;
68
type ResolverProbe = Set<number>;
@@ -66,7 +68,7 @@ const observedCombinators: Record<CombinatorName, Combinator> = Object.fromEntri
6668
const members = collectMembers(values);
6769
if (members === undefined) return native.call(this, values);
6870
const aggregate = native.call(this, members);
69-
activeLifecycle.getStore()?.registerCombinator(members, aggregate);
71+
activeLifecycle.getStore()?.registerCombinator(members, aggregate, name);
7072
return aggregate;
7173
};
7274
Object.defineProperty(observed, 'name', { value: name, configurable: true });
@@ -117,6 +119,8 @@ export class AuthoredFlowLifecycle {
117119
*/
118120
applyPredicateGate: (<T>(operation: { id: string; predicateGate: unknown }, value: T) => Promise<T>) | undefined = undefined;
119121
private readonly graph: AuthoredPromiseGraph;
122+
private readonly stepGraph: AuthoredStepGraph;
123+
private readonly operationPromises = new Map<OperationToken, number>();
120124
private readonly activeResolverProbes: ResolverProbe[] = [];
121125
private readonly invocations = new Map<OperationToken, AuthoredOperationInvocation[]>();
122126
private readonly promiseAllAggregates = new Map<OperationToken, Set<number>>();
@@ -135,6 +139,7 @@ export class AuthoredFlowLifecycle {
135139
for (const probe of this.activeResolverProbes) probe.add(asyncId);
136140
},
137141
);
142+
this.stepGraph = new AuthoredStepGraph((asyncId) => this.graph.causesOf(asyncId));
138143
installPromiseAllObserver();
139144
try {
140145
this.graph.enable();
@@ -152,16 +157,25 @@ export class AuthoredFlowLifecycle {
152157
stepOwners.set(step, { lifecycle: this, operation });
153158
}
154159

155-
registerCombinator(values: readonly unknown[], aggregate: Promise<unknown>): void {
160+
registerCombinator(
161+
values: readonly unknown[],
162+
aggregate: Promise<unknown>,
163+
combinator: CombinatorName,
164+
): void {
156165
const aggregateId = this.graph.idOf(aggregate);
157166
if (aggregateId === undefined) return;
158167
this.graph.registerRoot(aggregateId);
159168
const memberIds = new Set<number>();
169+
const awaitedMembers: number[] = [];
160170
for (const value of values) {
161171
if ((typeof value !== 'object' && typeof value !== 'function') || value === null) continue;
162172
const memberId = this.graph.idOf(value);
163173
if (memberId !== undefined) memberIds.add(memberId);
164174
const owner = stepOwners.get(value);
175+
// A Step is a thenable, not a promise: its member identity for the step
176+
// graph is the operation's own promise.
177+
const awaited = owner?.lifecycle === this ? this.operationPromises.get(owner.operation) : memberId;
178+
if (awaited !== undefined) awaitedMembers.push(awaited);
165179
if (owner?.lifecycle !== this) continue;
166180
let operationAggregates = this.promiseAllAggregates.get(owner.operation);
167181
if (operationAggregates === undefined) {
@@ -172,6 +186,33 @@ export class AuthoredFlowLifecycle {
172186
}
173187
this.promiseAllGroups.push({ aggregate: aggregateId, members: memberIds });
174188
this.registrations++;
189+
this.stepGraph.registerAggregate(aggregateId, aggregate, combinator, awaitedMembers);
190+
}
191+
192+
/**
193+
* Record an operation's own promise and, for a step the run's DAG draws, its
194+
* place in that DAG. Called synchronously from the operation's constructor:
195+
* the causal predecessors are read from the context that invoked
196+
* `f.agent`/`f.run`/`f.llm`, which is observable only at that moment.
197+
*
198+
* Every operation's promise is recorded so `Promise.all([step, …])` can
199+
* expand to it; only a `node` becomes a frontier the walk stops at. An
200+
* operation that is not drawn (a helper effect) is walked through instead,
201+
* so an edge never names a step the index has no record for.
202+
*/
203+
registerOperation(
204+
operation: OperationToken,
205+
promise: Promise<unknown>,
206+
node?: { readonly step: string; readonly label?: string },
207+
): AuthoredStepEdges | undefined {
208+
const promiseId = this.graph.idOf(promise);
209+
if (promiseId !== undefined) this.operationPromises.set(operation, promiseId);
210+
if (node === undefined) return undefined;
211+
return this.stepGraph.registerStep(node.step, promiseId, executionAsyncId(), node.label);
212+
}
213+
214+
stepEdges(step: string): AuthoredStepEdges | undefined {
215+
return this.stepGraph.edgesOf(step);
175216
}
176217

177218
registerInvocation(
@@ -304,6 +345,8 @@ export class AuthoredFlowLifecycle {
304345
this.closed = true;
305346
this.graph.disable();
306347
this.graph.clear();
348+
this.stepGraph.release();
349+
this.operationPromises.clear();
307350
this.invocations.clear();
308351
this.promiseAllAggregates.clear();
309352
this.promiseAllGroups.length = 0;

‎packages/sdk/src/authored-flow-operation.ts‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@ import {
55
AuthoredFlowLifecycle,
66
type AuthoredOperationInvocation,
77
} from './authored-flow-lifecycle.js';
8+
import type { AuthoredStepEdges } from './authored-step-index.js';
89

910
/** Slice-P kinds the surface `.gate(config)` accepts and the SDK lowers. */
1011
const NAMED_GATE_KINDS = new Set([
@@ -39,6 +40,11 @@ export class AuthoredFlowOperation<T> {
3940
* function itself is never serialized. `flows check` cannot prove it.
4041
*/
4142
predicateGate: { predicate: (value: T) => boolean; because?: string } | undefined = undefined;
43+
/**
44+
* This step's label and causal predecessors in the run's DAG, captured when
45+
* it was invoked. Undefined for an operation the step index does not draw.
46+
*/
47+
readonly edges: AuthoredStepEdges | undefined;
4248
private state: OperationState = 'created';
4349
private thenInvoked = false;
4450
private rootFailureRecorded = false;
@@ -55,6 +61,7 @@ export class AuthoredFlowOperation<T> {
5561
private readonly assertCanStart: () => void,
5662
private readonly start: () => Promise<T>,
5763
private readonly scope: AuthoredFlowLifecycle,
64+
node?: { readonly label?: string },
5865
) {
5966
let resolve!: (value: T | PromiseLike<T>) => void;
6067
let reject!: (reason?: unknown) => void;
@@ -64,6 +71,9 @@ export class AuthoredFlowOperation<T> {
6471
});
6572
this.resolve = resolve;
6673
this.reject = reject;
74+
this.edges = scope.registerOperation(
75+
this, this.promise, node === undefined ? undefined : { step: id, ...node },
76+
);
6777

6878
observeRejection(this.promise, (error) => this.recordRootFailure(error));
6979
const operation = this;

0 commit comments

Comments
 (0)