alkdev/flowgraph
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'feat/graph-construction-call'

Browse Source 
# Conflicts:
#	src/graph/construction.ts
#	test/graph/construction.test.ts
This commit is contained in:
glm-5.1
2026-05-21 22:06:51 +00:00
parent d7fea47214 2e0350e87c
commit c57a8558c7
6 changed files with 784 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files

209
src/graph/construction.ts
View File

@@ -8,6 +8,7 @@ import {
NodeNotFoundError,
CycleError,
InvalidInputError,
InvalidTransitionError,
} from "../error/index.js";
import type { CallStatus, AnyValidationError, ValidationError } from "../error/index.js";
import {
@@ -20,8 +21,10 @@ import {
OperationEdgeAttrs as OperationEdgeAttrsSchema,
OperationGraphSerialized,
CallGraphSerialized,
CallNodeAttrs as CallNodeAttrsSchema,
CallEdgeAttrs as CallEdgeAttrsSchema,
} from "../schema/index.js";
import type { OperationNodeAttrs, FlowGraphSerialized } from "../schema/index.js";
import type { OperationNodeAttrs, FlowGraphSerialized, CallNodeAttrs } from "../schema/index.js";
import { typeCompat, type TypeCompatResult } from "../analysis/type-compat.js";
export interface FlowGraphOptions {
@@ -41,8 +44,55 @@ export interface OperationSpec {
tags?: string[];
}
export interface CallRequestedEvent {
type: "call.requested";
requestId: string;
operationId: string;
input: unknown;
timestamp: string;
parentRequestId?: string;
identity?: { id: string; scopes: string[]; resources?: Record<string, string[]> };
startedAt?: string;
}
export interface CallRespondedEvent {
type: "call.responded";
requestId: string;
output: unknown;
timestamp: string;
}
export interface CallErrorEvent {
type: "call.error";
requestId: string;
error: { code: string; message: string; details?: unknown };
timestamp: string;
}
export interface CallAbortedEvent {
type: "call.aborted";
requestId: string;
timestamp: string;
}
export interface CallCompletedEvent {
type: "call.completed";
requestId: string;
output?: unknown;
timestamp: string;
}
export type CallEventMapValue =
| CallRequestedEvent
| CallRespondedEvent
| CallErrorEvent
| CallAbortedEvent
| CallCompletedEvent;
type OperationGraph = FlowGraph<typeof OperationNodeAttrsSchema, typeof OperationEdgeAttrsSchema>;
type CallGraph = FlowGraph<typeof CallNodeAttrsSchema, typeof CallEdgeAttrsSchema>;
type TypedEdgeAttrs = {
edgeType: "typed";
compatible: boolean;
@@ -52,6 +102,14 @@ type TypedEdgeAttrs = {
type Attrs = Record<string, unknown>;
const VALID_TRANSITIONS: Record<CallStatus, CallStatus[]> = {
pending: ["running", "aborted"],
running: ["completed", "failed", "aborted"],
completed: [],
failed: [],
aborted: [],
};
export class FlowGraph<
NodeAttrs extends TSchema = TSchema,
EdgeAttrs extends TSchema = TSchema,
@@ -318,6 +376,145 @@ export class FlowGraph<
return chain;
}
updateFromEvent(event: CallEventMapValue): void {
switch (event.type) {
case "call.requested": {
const attrs: CallNodeAttrs = {
requestId: event.requestId,
operationId: event.operationId,
status: "pending",
input: event.input,
...(event.parentRequestId !== undefined ? { parentRequestId: event.parentRequestId } : {}),
...(event.identity !== undefined ? { identity: event.identity } : {}),
...(event.startedAt !== undefined ? { startedAt: event.startedAt } : {}),
};
this.addCall(attrs);
break;
}
case "call.responded": {
if (!this._graph.hasNode(event.requestId)) return;
const current = this._graph.getNodeAttributes(event.requestId) as Record<string, unknown>;
const currentStatus = current.status as CallStatus;
if (currentStatus === "completed" || currentStatus === "failed" || currentStatus === "aborted") return;
this._graph.mergeNodeAttributes(event.requestId, {
status: "completed",
output: event.output,
completedAt: event.timestamp,
} as Attrs);
break;
}
case "call.error": {
if (!this._graph.hasNode(event.requestId)) return;
const current = this._graph.getNodeAttributes(event.requestId) as Record<string, unknown>;
const currentStatus = current.status as CallStatus;
if (currentStatus === "completed" || currentStatus === "failed" || currentStatus === "aborted") return;
this._graph.mergeNodeAttributes(event.requestId, {
status: "failed",
error: event.error,
completedAt: event.timestamp,
} as Attrs);
break;
}
case "call.aborted": {
if (!this._graph.hasNode(event.requestId)) return;
const current = this._graph.getNodeAttributes(event.requestId) as Record<string, unknown>;
const currentStatus = current.status as CallStatus;
if (currentStatus === "completed" || currentStatus === "failed" || currentStatus === "aborted") return;
this._graph.mergeNodeAttributes(event.requestId, {
status: "aborted",
completedAt: event.timestamp,
} as Attrs);
break;
}
case "call.completed": {
if (!this._graph.hasNode(event.requestId)) return;
const current = this._graph.getNodeAttributes(event.requestId) as Record<string, unknown>;
const currentStatus = current.status as CallStatus;
if (currentStatus === "completed") {
if (!current.completedAt) {
this._graph.mergeNodeAttributes(event.requestId, { completedAt: event.timestamp } as Attrs);
}
return;
}
if (currentStatus === "failed" || currentStatus === "aborted") return;
this._graph.mergeNodeAttributes(event.requestId, {
status: "completed",
...(event.output !== undefined ? { output: event.output } : {}),
completedAt: event.timestamp,
} as Attrs);
break;
}
}
}
addCall(attrs: CallNodeAttrs): void {
if (this._graph.hasNode(attrs.requestId)) return;
this._graph.addNode(attrs.requestId, attrs as Attrs);
if (attrs.parentRequestId !== undefined) {
if (this._graph.hasNode(attrs.parentRequestId)) {
if (willCreateCycle(this._graph, attrs.parentRequestId, attrs.requestId)) {
this._graph.dropNode(attrs.requestId);
const path = this._findPath(attrs.requestId, attrs.parentRequestId);
const cycle = [attrs.parentRequestId, ...path, attrs.parentRequestId];
throw new CycleError([cycle]);
}
const edgeKey = this._edgeKey(attrs.parentRequestId, attrs.requestId);
this._graph.addEdgeWithKey(edgeKey, attrs.parentRequestId, attrs.requestId, { edgeType: "triggered" } as Attrs);
}
}
}
addDependency(source: string, target: string): void {
if (!this._graph.hasNode(source)) {
throw new NodeNotFoundError(source);
}
if (!this._graph.hasNode(target)) {
throw new NodeNotFoundError(target);
}
const edgeKey = `${source}->${target}:depends_on`;
if (this._graph.hasEdge(edgeKey)) return;
if (willCreateCycle(this._graph, source, target)) {
const path = this._findPath(target, source);
const cycle = [source, ...path, source];
throw new CycleError([cycle]);
}
this._graph.addEdgeWithKey(edgeKey, source, target, { edgeType: "depends_on" } as Attrs);
}
updateStatus(requestId: string, status: CallStatus, extra?: Partial<CallNodeAttrs>): void {
if (!this._graph.hasNode(requestId)) {
throw new NodeNotFoundError(requestId);
}
const current = this._graph.getNodeAttributes(requestId) as Record<string, unknown>;
const currentStatus = current.status as CallStatus;
if (currentStatus === status) return;
const allowed = VALID_TRANSITIONS[currentStatus];
if (!allowed || !allowed.includes(status)) {
throw new InvalidTransitionError(requestId, currentStatus, status);
}
const update: Record<string, unknown> = { status };
if (extra) {
for (const [key, value] of Object.entries(extra)) {
if (value !== undefined) {
update[key] = value;
}
}
}
this._graph.mergeNodeAttributes(requestId, update as Attrs);
}
updateCall(requestId: string, attrs: Partial<CallNodeAttrs>): void {
if (!this._graph.hasNode(requestId)) {
throw new NodeNotFoundError(requestId);
}
this._graph.mergeNodeAttributes(requestId, attrs as Attrs);
}
removeCall(requestId: string): void {
if (!this._graph.hasNode(requestId)) return;
this._graph.dropNode(requestId);
}
validate(schema: TSchema): AnyValidationError[] {
return _validate(this, schema as NodeAttrs);
}
@@ -365,10 +562,12 @@ export class FlowGraph<
return graph;
}
static fromCallEvents(
_events: unknown[],
): FlowGraph<TSchema, TSchema> {
throw new Error("not implemented");
static fromCallEvents(events: CallEventMapValue[]): CallGraph {
const graph = new FlowGraph<typeof CallNodeAttrsSchema, typeof CallEdgeAttrsSchema>();
for (const event of events) {
graph.updateFromEvent(event);
}
return graph;
}
export(): FlowGraphSerialized {

2
src/graph/index.ts
View File

@@ -1,4 +1,4 @@
export { FlowGraph, buildTypeEdges, type FlowGraphOptions, type OperationSpec } from "./construction.js";
export { FlowGraph, buildTypeEdges, type FlowGraphOptions, type OperationSpec, type CallEventMapValue, type CallRequestedEvent, type CallRespondedEvent, type CallErrorEvent, type CallAbortedEvent, type CallCompletedEvent } from "./construction.js";
export {
topologicalOrder,
hasCycles,

2
src/index.ts
View File

@@ -1,6 +1,6 @@
export * from "./error/index.js";
export { FlowGraph, buildTypeEdges, type FlowGraphOptions, type OperationSpec } from "./graph/index.js";
export { FlowGraph, buildTypeEdges, type FlowGraphOptions, type OperationSpec, type CallEventMapValue, type CallRequestedEvent, type CallRespondedEvent, type CallErrorEvent, type CallAbortedEvent, type CallCompletedEvent } from "./graph/index.js";
export {
validateSchema,
validateGraph,

52
src/reactive/workflow.ts
View File

@@ -88,6 +88,7 @@ export class WorkflowReactiveRoot implements EventLogProjection {
blockedByFailure: Map<string, ReadonlySignal<boolean>>;
resultMap: Map<string, ReadonlySignal<CallResult | undefined>>;
nodeKeyToRequestId: Map<string, string>;
requestIdToNodeKey: Map<string, string>;
private graph: DirectedGraph;
private effectDisposers: (() => void)[];
@@ -106,10 +107,16 @@ export class WorkflowReactiveRoot implements EventLogProjection {
this.effectDisposers = [];
this.eventLog = [];
this.nodeKeyToRequestId = new Map();
this.requestIdToNodeKey = new Map();
this._failurePolicy = options?.failurePolicy ?? "continue-running";
this.initializeSignals();
}
setRequestId(nodeKey: string, requestId: string): void {
this.nodeKeyToRequestId.set(nodeKey, requestId);
this.requestIdToNodeKey.set(requestId, nodeKey);
}
private initializeSignals(): void {
for (const node of this.graph.nodes()) {
const predecessors: string[] = this.graph.inNeighbors(node) ?? [];
@@ -213,15 +220,29 @@ export class WorkflowReactiveRoot implements EventLogProjection {
if (!("requestId" in event)) return;
const nodeId = this.findNodeByRequestId(event.requestId);
let nodeId = this.requestIdToNodeKey.get(event.requestId);
if (nodeId === undefined) {
for (const [nId, rid] of this.nodeKeyToRequestId) {
if (rid === event.requestId) {
nodeId = nId;
this.requestIdToNodeKey.set(event.requestId, nId);
break;
}
}
}
if (nodeId === undefined) return;
const statusSignal = this.statusMap.get(nodeId);
if (!statusSignal) return;
const currentRequestId = this.nodeKeyToRequestId.get(nodeId);
if (currentRequestId === event.requestId) {
const statusSignal = this.statusMap.get(nodeId);
if (!statusSignal) return;
const derived = EVENT_TO_STATUS[event.type];
if (derived !== undefined) {
statusSignal.value = derived;
const derived = EVENT_TO_STATUS[event.type];
if (derived !== undefined) {
statusSignal.value = derived;
}
}
}
@@ -254,12 +275,17 @@ export class WorkflowReactiveRoot implements EventLogProjection {
}
getEvents(nodeId: string): CallEventMapValue[] {
const requestId = this.nodeKeyToRequestId.get(nodeId);
if (!requestId) return [];
const requestIds = new Set<string>();
for (const [rid, nId] of this.requestIdToNodeKey) {
if (nId === nodeId) {
requestIds.add(rid);
}
}
if (requestIds.size === 0) return [];
const events: CallEventMapValue[] = [];
for (const e of this.eventLog) {
if ("requestId" in e && e.requestId === requestId) {
if ("requestId" in e && requestIds.has(e.requestId)) {
events.push(e);
}
}
@@ -325,13 +351,7 @@ export class WorkflowReactiveRoot implements EventLogProjection {
this.blockedByFailure.clear();
this.resultMap.clear();
this.nodeKeyToRequestId.clear();
this.requestIdToNodeKey.clear();
this.eventLog = [];
}
private findNodeByRequestId(requestId: string): string | undefined {
for (const [nodeId, rid] of this.nodeKeyToRequestId) {
if (rid === requestId) return nodeId;
}
return undefined;
}
}