Function buildAgentGraphTraceSnapshot

packages/runtime/src/stream-graph-trace.ts:101–218

(
  input: BuildAgentGraphTraceSnapshotInput,
)

}

export function buildAgentGraphTraceSnapshot(
  input: BuildAgentGraphTraceSnapshotInput,
): AgentGraphTraceSnapshot {
  const validated = validateTopology(input.topology);
  const topologyFingerprint = fingerprintTopology(input.topology.graphId, validated);
  const replay =
    input.records.length > 0
      ? replayAgentGraphRecords(input.records)
      : {
          graphId: input.topology.graphId,
          appliedRecordIds: [],
          operators: {},
        };

  if (replay.graphId !== input.topology.graphId) {
    throw new Error(
      `Trace topology ${input.topology.graphId} cannot observe records from graph ${replay.graphId}`,
    );
  }

  const recordsById = new Map(input.records.map((record) => [record.recordId, record]));
  const orderedRecords = replay.appliedRecordIds.map((recordId) => recordsById.get(recordId)!);
  for (const record of orderedRecords) {
    const binding = validated.operatorsById.get(record.operatorId);
    if (!binding) {
      throw new Error(
        `Graph record ${record.recordId} references unknown topology operator ${record.operatorId}`,
      );
    }
    if (binding.sessionId !== record.sessionId) {
      throw new Error(
        `Topology operator ${record.operatorId} is bound to ${binding.sessionId}, record uses ${record.sessionId}`,
      );
    }
  }

  const topologicalIndex = new Map(
    validated.topologicalOrder.map((operatorId, index) => [operatorId, index]),
  );
  const compareOperators = (a: string, b: string): number =>
    topologicalIndex.get(a)! - topologicalIndex.get(b)! || compareAgentGraphIdentity(a, b);

  const replayOperators = new Map(Object.entries(replay.operators));
  const operators = new Map<string, AgentGraphTraceOperatorState>();
  for (const operatorId of validated.topologicalOrder) {
    const binding = validated.operatorsById.get(operatorId)!;
    const runtimeState = replayOperators.get(operatorId);
    operators.set(operatorId, {
      operatorId,
      sessionId: binding.sessionId,
      topologicalIndex: topologicalIndex.get(operatorId)!,
      upstreamOperatorIds: uniqueOperatorIds(
        (validated.incoming.get(operatorId) ?? []).map((edge) => edge.fromOperatorId),
        compareOperators,
      ),
      downstreamOperatorIds: uniqueOperatorIds(
        (validated.outgoing.get(operatorId) ?? []).map((edge) => edge.toOperatorId),
        compareOperators,