packages/runtime/src/__tests__/stream-graph-trace.test.ts - maka - Git at Google

 /*
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
  * regarding copyright ownership.  The ASF licenses this file
  * to you under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance
  * with the License.  You may obtain a copy of the License at
  *
  *     http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing,
  * software distributed under the License is distributed on an
  * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
  * KIND, either express or implied.  See the License for the
  * specific language governing permissions and limitations
  * under the License.
  */

 import assert from 'node:assert/strict';
 import { describe, test } from 'node:test';
 import type { RuntimeInvocationRecord } from '@maka/core/runtime-invocation';
 import type { RuntimeEvent } from '@maka/core/runtime-event';
 import { projectAgentGraphRecords } from '../stream-graph-projection.js';
 import {
   AGENT_GRAPH_TRACE_SCHEMA_VERSION,
   buildAgentGraphTraceSnapshot,
   type AgentGraphTraceTopology,
 } from '../stream-graph-trace.js';
 import { testInvocationRecord } from './invocation-fixture.js';

 const baseTs = 1_800_000_000_000;

 describe('stream graph trace topology', () => {
   test('materializes deterministic direct-edge routes without putting the supervisor in the path', () => {
     const research = runInvocation('research', baseTs);
     const verify = runInvocation('verify', baseTs + 1);
     const synthesize = runInvocation('synthesize', baseTs + 2);
     const audit = runInvocation('audit', baseTs + 3);
     const projection = projectAgentGraphRecords({
       graphId: 'graph-trace',
       streams: [
         stream(research, 'research', [
           runtimeEvent(research, 'research-message', baseTs + 10, 'private-research-payload'),
         ]),
         stream(verify, 'verify', [
           runtimeEvent(verify, 'verify-message', baseTs + 5, 'private-verification-payload'),
         ]),
         stream(synthesize, 'synthesize', [
           runtimeEvent(synthesize, 'synthesis-message', baseTs + 20, 'private-synthesis-payload'),
         ]),
       ],
     });
     const topology: AgentGraphTraceTopology = {
       graphId: 'graph-trace',
       operators: [
         binding(audit, 'audit'),
         binding(synthesize, 'synthesize'),
         binding(verify, 'verify'),
         binding(research, 'research'),
       ],
       edges: [
         {
           edgeId: 'synthesis-to-audit',
           fromOperatorId: 'synthesize',
           toOperatorId: 'audit',
         },
         {
           edgeId: 'verify-to-synthesis',
           fromOperatorId: 'verify',
           toOperatorId: 'synthesize',
         },
         {
           edgeId: 'research-to-synthesis',
           fromOperatorId: 'research',
           toOperatorId: 'synthesize',
         },
       ],
     };

     const snapshot = buildAgentGraphTraceSnapshot({
       topology,
       records: projection.records,
     });

     assert.equal(snapshot.schemaVersion, AGENT_GRAPH_TRACE_SCHEMA_VERSION);
     assert.deepEqual(snapshot.topologicalOrder, ['research', 'verify', 'synthesize', 'audit']);
     assert.deepEqual(snapshot.rootOperatorIds, ['research', 'verify']);
     assert.deepEqual(snapshot.sinkOperatorIds, ['audit']);
     assert.deepEqual(
       snapshot.recordIds,
       projection.records.map((record) => record.recordId),
     );
     assert.deepEqual(
       snapshot.routes.map((route) => [
         route.sourceOperatorId,
         route.targetOperatorId,
         route.sourceRecordId,
       ]),
       [
         ['verify', 'synthesize', projection.records[0]!.recordId],
         ['research', 'synthesize', projection.records[1]!.recordId],
         ['synthesize', 'audit', projection.records[2]!.recordId],
       ],
       'records travel across direct edges only; trace routing does not recursively invent work',
     );
     assert.deepEqual(snapshot.operators.synthesize?.upstreamOperatorIds, ['research', 'verify']);
     assert.equal(snapshot.operators.synthesize?.receivedRouteIds.length, 2);
     assert.equal(snapshot.operators.audit?.receivedRouteIds.length, 1);
     assert.equal(snapshot.operators.audit?.runtimeState, undefined);
     assert.equal(
       snapshot.operators.verify?.runtimeState?.activations[verify.runId]?.recordCount,
       1,
     );
     assert.deepEqual(
       projection.supervisorMetaStream.map((record) => record.recordId),
       projection.records.map((record) => record.recordId),
       'the always-on supervisor continues to observe source records independently',
     );
     assert.doesNotMatch(
       JSON.stringify(snapshot),
       /private-(research|verification|synthesis)-payload/,
     );
   });

   test('is deterministic and idempotent for reordered duplicate observations', () => {
     const source = runInvocation('source', baseTs);
     const target = runInvocation('target', baseTs + 1);
     const projection = projectAgentGraphRecords({
       graphId: 'graph-replay',
       streams: [
         stream(source, 'source', [
           runtimeEvent(source, 'first', baseTs + 1, 'first'),
           runtimeEvent(source, 'second', baseTs + 2, 'second'),
         ]),
       ],
     });
     const topology: AgentGraphTraceTopology = {
       graphId: 'graph-replay',
       operators: [binding(target, 'target'), binding(source, 'source')],
       edges: [
         {
           edgeId: 'source-to-target',
           fromOperatorId: 'source',
           toOperatorId: 'target',
         },
       ],
     };

     const canonical = buildAgentGraphTraceSnapshot({
       topology,
       records: projection.records,
     });
     const replayed = buildAgentGraphTraceSnapshot({
       topology: {
         ...topology,
         operators: [...topology.operators].reverse(),
         edges: [...topology.edges].reverse(),
       },
       records: [
         projection.records[1]!,
         projection.records[0]!,
         projection.records[1]!,
         projection.records[0]!,
       ],
     });

     assert.deepEqual(replayed, canonical);
     assert.equal(new Set(canonical.routes.map((route) => route.routeId)).size, 2);
   });

   test('fingerprints only declared topology fields in raw identity order', () => {
     const precomposed = runInvocation('unicode-precomposed', baseTs);
     const decomposed = runInvocation('unicode-decomposed', baseTs + 1);
     const target = runInvocation('unicode-target', baseTs + 2);
     const precomposedId = '\u00e9';
     const decomposedId = 'e\u0301';
     assert.equal(precomposedId.localeCompare(decomposedId), 0);

     const canonical = buildAgentGraphTraceSnapshot({
       topology: {
         graphId: 'graph-canonical-fingerprint',
         operators: [
           binding(precomposed, precomposedId),
           binding(decomposed, decomposedId),
           binding(target, 'target'),
         ],
         edges: [
           {
             edgeId: 'precomposed-target',
             fromOperatorId: precomposedId,
             toOperatorId: 'target',
           },
           {
             edgeId: 'decomposed-target',
             fromOperatorId: decomposedId,
             toOperatorId: 'target',
           },
         ],
       },
       records: [],
     });
     const noisyOperators = [
       { ...binding(target, 'target'), displayName: 'ignored target' },
       {
         ...binding(decomposed, decomposedId),
         displayName: 'ignored decomposed',
       },
       {
         ...binding(precomposed, precomposedId),
         displayName: 'ignored precomposed',
       },
     ];
     const noisyEdges = [
       {
         edgeId: 'decomposed-target',
         fromOperatorId: decomposedId,
         toOperatorId: 'target',
         debugColor: 'blue',
       },
       {
         edgeId: 'precomposed-target',
         fromOperatorId: precomposedId,
         toOperatorId: 'target',
         debugColor: 'red',
       },
     ];
     const reorderedWithExtras = buildAgentGraphTraceSnapshot({
       topology: {
         graphId: 'graph-canonical-fingerprint',
         operators: noisyOperators,
         edges: noisyEdges,
       },
       records: [],
     });

     assert.deepEqual(reorderedWithExtras, canonical);
     assert.equal(reorderedWithExtras.topologyFingerprint, canonical.topologyFingerprint);
   });

   test('keeps existing route identities stable as later observations arrive', () => {
     const source = runInvocation('source', baseTs);
     const target = runInvocation('target', baseTs + 1);
     const initialProjection = projectAgentGraphRecords({
       graphId: 'graph-incremental',
       streams: [stream(source, 'source', [runtimeEvent(source, 'first', baseTs + 10, 'first')])],
     });
     const expandedProjection = projectAgentGraphRecords({
       graphId: 'graph-incremental',
       streams: [
         stream(source, 'source', [
           runtimeEvent(source, 'first', baseTs + 10, 'first'),
           runtimeEvent(source, 'second', baseTs + 20, 'second'),
         ]),
       ],
     });
     const topology: AgentGraphTraceTopology = {
       graphId: 'graph-incremental',
       operators: [binding(source, 'source'), binding(target, 'target')],
       edges: [
         {
           edgeId: 'source-to-target',
           fromOperatorId: 'source',
           toOperatorId: 'target',
         },
       ],
     };

     const initial = buildAgentGraphTraceSnapshot({
       topology,
       records: initialProjection.records,
     });
     const expanded = buildAgentGraphTraceSnapshot({
       topology,
       records: expandedProjection.records,
     });

     assert.deepEqual(expanded.routes[0], initial.routes[0]);
     assert.equal(expanded.routes.length, 2);
   });

   test('retains an observable topology before any runtime facts arrive', () => {
     const source = runInvocation('source', baseTs);
     const target = runInvocation('target', baseTs + 1);

     const snapshot = buildAgentGraphTraceSnapshot({
       topology: {
         graphId: 'graph-empty-trace',
         operators: [binding(target, 'target'), binding(source, 'source')],
         edges: [
           {
             edgeId: 'source-to-target',
             fromOperatorId: 'source',
             toOperatorId: 'target',
           },
         ],
       },
       records: [],
     });

     assert.deepEqual(snapshot.topologicalOrder, ['source', 'target']);
     assert.deepEqual(snapshot.rootOperatorIds, ['source']);
     assert.deepEqual(snapshot.sinkOperatorIds, ['target']);
     assert.deepEqual(snapshot.recordIds, []);
     assert.deepEqual(snapshot.routes, []);
     assert.equal(snapshot.operators.source?.runtimeState, undefined);
     assert.equal(snapshot.operators.target?.runtimeState, undefined);
   });

   test('materializes reserved JavaScript property names as own snapshot keys', () => {
     const source = {
       ...runInvocation('reserved-source', baseTs),
       runId: 'constructor',
       invocationId: 'reserved-invocation',
     };
     const target = runInvocation('reserved-target', baseTs + 1);
     const projection = projectAgentGraphRecords({
       graphId: 'graph-reserved-keys',
       streams: [
         stream(source, '__proto__', [
           runtimeEvent(source, 'reserved-record', baseTs + 1, 'reserved'),
         ]),
       ],
     });

     const snapshot = buildAgentGraphTraceSnapshot({
       topology: {
         graphId: 'graph-reserved-keys',
         operators: [binding(source, '__proto__'), binding(target, 'toString')],
         edges: [
           {
             edgeId: '__proto__',
             fromOperatorId: '__proto__',
             toOperatorId: 'toString',
           },
         ],
       },
       records: projection.records,
     });

     assert.equal(Object.hasOwn(snapshot.operators, '__proto__'), true);
     assert.equal(Object.hasOwn(snapshot.operators, 'toString'), true);
     assert.equal(Object.hasOwn(snapshot.edges, '__proto__'), true);
     assert.equal(snapshot.edges['__proto__']?.routeIds.length, 1);
     assert.equal(
       Object.hasOwn(
         snapshot.operators['__proto__']?.runtimeState?.activations ?? {},
         'constructor',
       ),
       true,
     );
   });

   test('binds route identity to immutable edge endpoints', () => {
     const source = runInvocation('route-source', baseTs);
     const targetA = runInvocation('route-target-a', baseTs + 1);
     const targetB = runInvocation('route-target-b', baseTs + 2);
     const projection = projectAgentGraphRecords({
       graphId: 'graph-edge-rebinding',
       streams: [
         stream(source, 'source', [runtimeEvent(source, 'source-record', baseTs + 1, 'source')]),
       ],
     });
     const operators = [
       binding(source, 'source'),
       binding(targetA, 'target-a'),
       binding(targetB, 'target-b'),
     ];

     const first = buildAgentGraphTraceSnapshot({
       topology: {
         graphId: 'graph-edge-rebinding',
         operators,
         edges: [
           {
             edgeId: 'edge',
             fromOperatorId: 'source',
             toOperatorId: 'target-a',
           },
         ],
       },
       records: projection.records,
     });
     const rebound = buildAgentGraphTraceSnapshot({
       topology: {
         graphId: 'graph-edge-rebinding',
         operators,
         edges: [
           {
             edgeId: 'edge',
             fromOperatorId: 'source',
             toOperatorId: 'target-b',
           },
         ],
       },
       records: projection.records,
     });

     assert.notEqual(first.routes[0]?.routeId, rebound.routes[0]?.routeId);
     assert.equal(first.routes[0]?.targetOperatorId, 'target-a');
     assert.equal(rebound.routes[0]?.targetOperatorId, 'target-b');
   });

   test('fails closed on invalid topology and record ownership', () => {
     const one = runInvocation('one', baseTs);
     const two = runInvocation('two', baseTs + 1);
     const three = runInvocation('three', baseTs + 2);
     const projection = projectAgentGraphRecords({
       graphId: 'graph-invalid',
       streams: [stream(one, 'one', [runtimeEvent(one, 'one-message', baseTs + 1, 'one')])],
     });

     assert.throws(
       () =>
         buildAgentGraphTraceSnapshot({
           topology: {
             graphId: 'graph-invalid',
             operators: [binding(one, 'one'), binding(two, 'two'), binding(three, 'three')],
             edges: [
               { edgeId: 'one-two', fromOperatorId: 'one', toOperatorId: 'two' },
               {
                 edgeId: 'two-three',
                 fromOperatorId: 'two',
                 toOperatorId: 'three',
               },
               {
                 edgeId: 'three-one',
                 fromOperatorId: 'three',
                 toOperatorId: 'one',
               },
             ],
           },
           records: projection.records,
         }),
       /contains a cycle involving: one, three, two/,
     );
     assert.throws(
       () =>
         buildAgentGraphTraceSnapshot({
           topology: {
             graphId: 'graph-invalid',
             operators: [binding(one, 'one'), binding(two, 'two')],
             edges: [
               {
                 edgeId: 'one-missing',
                 fromOperatorId: 'one',
                 toOperatorId: 'missing',
               },
             ],
           },
           records: projection.records,
         }),
       /unknown target missing/,
     );
     assert.throws(
       () =>
         buildAgentGraphTraceSnapshot({
           topology: {
             graphId: 'graph-invalid',
             operators: [{ operatorId: 'one', sessionId: 'different-session' }, binding(two, 'two')],
             edges: [{ edgeId: 'one-two', fromOperatorId: 'one', toOperatorId: 'two' }],
           },
           records: projection.records,
         }),
       /Topology operator one is bound to different-session/,
     );
     assert.throws(
       () =>
         buildAgentGraphTraceSnapshot({
           topology: {
             graphId: 'different-graph',
             operators: [binding(one, 'one'), binding(two, 'two')],
             edges: [{ edgeId: 'one-two', fromOperatorId: 'one', toOperatorId: 'two' }],
           },
           records: projection.records,
         }),
       /cannot observe records from graph graph-invalid/,
     );
     assert.throws(
       () =>
         buildAgentGraphTraceSnapshot({
           topology: {
             graphId: 'graph-invalid',
             operators: [binding(two, 'two')],
             edges: [],
           },
           records: projection.records,
         }),
       /unknown topology operator one/,
     );
   });
 });

 /** One still-open invocation, as its opening fact describes it. */
 function runInvocation(name: string, openedAt: number): RuntimeInvocationRecord {
   return testInvocationRecord({
     sessionId: `session-${name}`,
     invocationId: `invocation-${name}`,
     runId: `run-${name}`,
     turnId: `turn-${name}`,
     openedAt,
   });
 }

 function binding(run: RuntimeInvocationRecord, operatorId: string) {
   return { operatorId, sessionId: run.sessionId };
 }

 function stream(run: RuntimeInvocationRecord, operatorId: string, events: readonly RuntimeEvent[]) {
   return {
     operator: binding(run, operatorId),
     run,
     events,
   };
 }

 function runtimeEvent(
   run: RuntimeInvocationRecord,
   id: string,
   ts: number,
   text: string,
 ): RuntimeEvent {
   return {
     id,
     invocationId: run.invocationId,
     runId: run.runId,
     sessionId: run.sessionId,
     turnId: run.turnId,
     ts,
     partial: false,
     role: 'model',
     author: 'agent',
     content: { kind: 'text', text },
   };
 }
	/*
	* Licensed to the Apache Software Foundation (ASF) under one
	* or more contributor license agreements. See the NOTICE file
	* distributed with this work for additional information
	* regarding copyright ownership. The ASF licenses this file
	* to you under the Apache License, Version 2.0 (the
	* "License"); you may not use this file except in compliance
	* with the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing,
	* software distributed under the License is distributed on an
	* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
	* KIND, either express or implied. See the License for the
	* specific language governing permissions and limitations
	* under the License.
	*/

	import assert from 'node:assert/strict';
	import { describe, test } from 'node:test';
	import type { RuntimeInvocationRecord } from '@maka/core/runtime-invocation';
	import type { RuntimeEvent } from '@maka/core/runtime-event';
	import { projectAgentGraphRecords } from '../stream-graph-projection.js';
	import {
	AGENT_GRAPH_TRACE_SCHEMA_VERSION,
	buildAgentGraphTraceSnapshot,
	type AgentGraphTraceTopology,
	} from '../stream-graph-trace.js';
	import { testInvocationRecord } from './invocation-fixture.js';

	const baseTs = 1_800_000_000_000;

	describe('stream graph trace topology', () => {
	test('materializes deterministic direct-edge routes without putting the supervisor in the path', () => {
	const research = runInvocation('research', baseTs);
	const verify = runInvocation('verify', baseTs + 1);
	const synthesize = runInvocation('synthesize', baseTs + 2);
	const audit = runInvocation('audit', baseTs + 3);
	const projection = projectAgentGraphRecords({
	graphId: 'graph-trace',
	streams: [
	stream(research, 'research', [
	runtimeEvent(research, 'research-message', baseTs + 10, 'private-research-payload'),
	]),
	stream(verify, 'verify', [
	runtimeEvent(verify, 'verify-message', baseTs + 5, 'private-verification-payload'),
	]),
	stream(synthesize, 'synthesize', [
	runtimeEvent(synthesize, 'synthesis-message', baseTs + 20, 'private-synthesis-payload'),
	]),
	],
	});
	const topology: AgentGraphTraceTopology = {
	graphId: 'graph-trace',
	operators: [
	binding(audit, 'audit'),
	binding(synthesize, 'synthesize'),
	binding(verify, 'verify'),
	binding(research, 'research'),
	],
	edges: [
	{
	edgeId: 'synthesis-to-audit',
	fromOperatorId: 'synthesize',
	toOperatorId: 'audit',
	},
	{
	edgeId: 'verify-to-synthesis',
	fromOperatorId: 'verify',
	toOperatorId: 'synthesize',
	},
	{
	edgeId: 'research-to-synthesis',
	fromOperatorId: 'research',
	toOperatorId: 'synthesize',
	},
	],
	};

	const snapshot = buildAgentGraphTraceSnapshot({
	topology,
	records: projection.records,
	});

	assert.equal(snapshot.schemaVersion, AGENT_GRAPH_TRACE_SCHEMA_VERSION);
	assert.deepEqual(snapshot.topologicalOrder, ['research', 'verify', 'synthesize', 'audit']);
	assert.deepEqual(snapshot.rootOperatorIds, ['research', 'verify']);
	assert.deepEqual(snapshot.sinkOperatorIds, ['audit']);
	assert.deepEqual(
	snapshot.recordIds,
	projection.records.map((record) => record.recordId),
	);
	assert.deepEqual(
	snapshot.routes.map((route) => [
	route.sourceOperatorId,
	route.targetOperatorId,
	route.sourceRecordId,
	]),
	[
	['verify', 'synthesize', projection.records[0]!.recordId],
	['research', 'synthesize', projection.records[1]!.recordId],
	['synthesize', 'audit', projection.records[2]!.recordId],
	],
	'records travel across direct edges only; trace routing does not recursively invent work',
	);
	assert.deepEqual(snapshot.operators.synthesize?.upstreamOperatorIds, ['research', 'verify']);
	assert.equal(snapshot.operators.synthesize?.receivedRouteIds.length, 2);
	assert.equal(snapshot.operators.audit?.receivedRouteIds.length, 1);
	assert.equal(snapshot.operators.audit?.runtimeState, undefined);
	assert.equal(
	snapshot.operators.verify?.runtimeState?.activations[verify.runId]?.recordCount,
	1,
	);
	assert.deepEqual(
	projection.supervisorMetaStream.map((record) => record.recordId),
	projection.records.map((record) => record.recordId),
	'the always-on supervisor continues to observe source records independently',
	);
	assert.doesNotMatch(
	JSON.stringify(snapshot),
	/private-(research\|verification\|synthesis)-payload/,
	);
	});

	test('is deterministic and idempotent for reordered duplicate observations', () => {
	const source = runInvocation('source', baseTs);
	const target = runInvocation('target', baseTs + 1);
	const projection = projectAgentGraphRecords({
	graphId: 'graph-replay',
	streams: [
	stream(source, 'source', [
	runtimeEvent(source, 'first', baseTs + 1, 'first'),
	runtimeEvent(source, 'second', baseTs + 2, 'second'),
	]),
	],
	});
	const topology: AgentGraphTraceTopology = {
	graphId: 'graph-replay',
	operators: [binding(target, 'target'), binding(source, 'source')],
	edges: [
	{
	edgeId: 'source-to-target',
	fromOperatorId: 'source',
	toOperatorId: 'target',
	},
	],
	};

	const canonical = buildAgentGraphTraceSnapshot({
	topology,
	records: projection.records,
	});
	const replayed = buildAgentGraphTraceSnapshot({
	topology: {
	...topology,
	operators: [...topology.operators].reverse(),
	edges: [...topology.edges].reverse(),
	},
	records: [
	projection.records[1]!,
	projection.records[0]!,
	projection.records[1]!,
	projection.records[0]!,
	],
	});

	assert.deepEqual(replayed, canonical);
	assert.equal(new Set(canonical.routes.map((route) => route.routeId)).size, 2);
	});

	test('fingerprints only declared topology fields in raw identity order', () => {
	const precomposed = runInvocation('unicode-precomposed', baseTs);
	const decomposed = runInvocation('unicode-decomposed', baseTs + 1);
	const target = runInvocation('unicode-target', baseTs + 2);
	const precomposedId = '\u00e9';
	const decomposedId = 'e\u0301';
	assert.equal(precomposedId.localeCompare(decomposedId), 0);

	const canonical = buildAgentGraphTraceSnapshot({
	topology: {
	graphId: 'graph-canonical-fingerprint',
	operators: [
	binding(precomposed, precomposedId),
	binding(decomposed, decomposedId),
	binding(target, 'target'),
	],
	edges: [
	{
	edgeId: 'precomposed-target',
	fromOperatorId: precomposedId,
	toOperatorId: 'target',
	},
	{
	edgeId: 'decomposed-target',
	fromOperatorId: decomposedId,
	toOperatorId: 'target',
	},
	],
	},
	records: [],
	});
	const noisyOperators = [
	{ ...binding(target, 'target'), displayName: 'ignored target' },
	{
	...binding(decomposed, decomposedId),
	displayName: 'ignored decomposed',
	},
	{
	...binding(precomposed, precomposedId),
	displayName: 'ignored precomposed',
	},
	];
	const noisyEdges = [
	{
	edgeId: 'decomposed-target',
	fromOperatorId: decomposedId,
	toOperatorId: 'target',
	debugColor: 'blue',
	},
	{
	edgeId: 'precomposed-target',
	fromOperatorId: precomposedId,
	toOperatorId: 'target',
	debugColor: 'red',
	},
	];
	const reorderedWithExtras = buildAgentGraphTraceSnapshot({
	topology: {
	graphId: 'graph-canonical-fingerprint',
	operators: noisyOperators,
	edges: noisyEdges,
	},
	records: [],
	});

	assert.deepEqual(reorderedWithExtras, canonical);
	assert.equal(reorderedWithExtras.topologyFingerprint, canonical.topologyFingerprint);
	});

	test('keeps existing route identities stable as later observations arrive', () => {
	const source = runInvocation('source', baseTs);
	const target = runInvocation('target', baseTs + 1);
	const initialProjection = projectAgentGraphRecords({
	graphId: 'graph-incremental',
	streams: [stream(source, 'source', [runtimeEvent(source, 'first', baseTs + 10, 'first')])],
	});
	const expandedProjection = projectAgentGraphRecords({
	graphId: 'graph-incremental',
	streams: [
	stream(source, 'source', [
	runtimeEvent(source, 'first', baseTs + 10, 'first'),
	runtimeEvent(source, 'second', baseTs + 20, 'second'),
	]),
	],
	});
	const topology: AgentGraphTraceTopology = {
	graphId: 'graph-incremental',
	operators: [binding(source, 'source'), binding(target, 'target')],
	edges: [
	{
	edgeId: 'source-to-target',
	fromOperatorId: 'source',
	toOperatorId: 'target',
	},
	],
	};

	const initial = buildAgentGraphTraceSnapshot({
	topology,
	records: initialProjection.records,
	});
	const expanded = buildAgentGraphTraceSnapshot({
	topology,
	records: expandedProjection.records,
	});

	assert.deepEqual(expanded.routes[0], initial.routes[0]);
	assert.equal(expanded.routes.length, 2);
	});

	test('retains an observable topology before any runtime facts arrive', () => {
	const source = runInvocation('source', baseTs);
	const target = runInvocation('target', baseTs + 1);

	const snapshot = buildAgentGraphTraceSnapshot({
	topology: {
	graphId: 'graph-empty-trace',
	operators: [binding(target, 'target'), binding(source, 'source')],
	edges: [
	{
	edgeId: 'source-to-target',
	fromOperatorId: 'source',
	toOperatorId: 'target',
	},
	],
	},
	records: [],
	});

	assert.deepEqual(snapshot.topologicalOrder, ['source', 'target']);
	assert.deepEqual(snapshot.rootOperatorIds, ['source']);
	assert.deepEqual(snapshot.sinkOperatorIds, ['target']);
	assert.deepEqual(snapshot.recordIds, []);
	assert.deepEqual(snapshot.routes, []);
	assert.equal(snapshot.operators.source?.runtimeState, undefined);
	assert.equal(snapshot.operators.target?.runtimeState, undefined);
	});

	test('materializes reserved JavaScript property names as own snapshot keys', () => {
	const source = {
	...runInvocation('reserved-source', baseTs),
	runId: 'constructor',
	invocationId: 'reserved-invocation',
	};
	const target = runInvocation('reserved-target', baseTs + 1);
	const projection = projectAgentGraphRecords({
	graphId: 'graph-reserved-keys',
	streams: [
	stream(source, '__proto__', [
	runtimeEvent(source, 'reserved-record', baseTs + 1, 'reserved'),
	]),
	],
	});

	const snapshot = buildAgentGraphTraceSnapshot({
	topology: {
	graphId: 'graph-reserved-keys',
	operators: [binding(source, '__proto__'), binding(target, 'toString')],
	edges: [
	{
	edgeId: '__proto__',
	fromOperatorId: '__proto__',
	toOperatorId: 'toString',
	},
	],
	},
	records: projection.records,
	});

	assert.equal(Object.hasOwn(snapshot.operators, '__proto__'), true);
	assert.equal(Object.hasOwn(snapshot.operators, 'toString'), true);
	assert.equal(Object.hasOwn(snapshot.edges, '__proto__'), true);
	assert.equal(snapshot.edges['__proto__']?.routeIds.length, 1);
	assert.equal(
	Object.hasOwn(
	snapshot.operators['__proto__']?.runtimeState?.activations ?? {},
	'constructor',
	),
	true,
	);
	});

	test('binds route identity to immutable edge endpoints', () => {
	const source = runInvocation('route-source', baseTs);
	const targetA = runInvocation('route-target-a', baseTs + 1);
	const targetB = runInvocation('route-target-b', baseTs + 2);
	const projection = projectAgentGraphRecords({
	graphId: 'graph-edge-rebinding',
	streams: [
	stream(source, 'source', [runtimeEvent(source, 'source-record', baseTs + 1, 'source')]),
	],
	});
	const operators = [
	binding(source, 'source'),
	binding(targetA, 'target-a'),
	binding(targetB, 'target-b'),
	];

	const first = buildAgentGraphTraceSnapshot({
	topology: {
	graphId: 'graph-edge-rebinding',
	operators,
	edges: [
	{
	edgeId: 'edge',
	fromOperatorId: 'source',
	toOperatorId: 'target-a',
	},
	],
	},
	records: projection.records,
	});
	const rebound = buildAgentGraphTraceSnapshot({
	topology: {
	graphId: 'graph-edge-rebinding',
	operators,
	edges: [
	{
	edgeId: 'edge',
	fromOperatorId: 'source',
	toOperatorId: 'target-b',
	},
	],
	},
	records: projection.records,
	});

	assert.notEqual(first.routes[0]?.routeId, rebound.routes[0]?.routeId);
	assert.equal(first.routes[0]?.targetOperatorId, 'target-a');
	assert.equal(rebound.routes[0]?.targetOperatorId, 'target-b');
	});

	test('fails closed on invalid topology and record ownership', () => {
	const one = runInvocation('one', baseTs);
	const two = runInvocation('two', baseTs + 1);
	const three = runInvocation('three', baseTs + 2);
	const projection = projectAgentGraphRecords({
	graphId: 'graph-invalid',
	streams: [stream(one, 'one', [runtimeEvent(one, 'one-message', baseTs + 1, 'one')])],
	});

	assert.throws(
	() =>
	buildAgentGraphTraceSnapshot({
	topology: {
	graphId: 'graph-invalid',
	operators: [binding(one, 'one'), binding(two, 'two'), binding(three, 'three')],
	edges: [
	{ edgeId: 'one-two', fromOperatorId: 'one', toOperatorId: 'two' },
	{
	edgeId: 'two-three',
	fromOperatorId: 'two',
	toOperatorId: 'three',
	},
	{
	edgeId: 'three-one',
	fromOperatorId: 'three',
	toOperatorId: 'one',
	},
	],
	},
	records: projection.records,
	}),
	/contains a cycle involving: one, three, two/,
	);
	assert.throws(
	() =>
	buildAgentGraphTraceSnapshot({
	topology: {
	graphId: 'graph-invalid',
	operators: [binding(one, 'one'), binding(two, 'two')],
	edges: [
	{
	edgeId: 'one-missing',
	fromOperatorId: 'one',
	toOperatorId: 'missing',
	},
	],
	},
	records: projection.records,
	}),
	/unknown target missing/,
	);
	assert.throws(
	() =>
	buildAgentGraphTraceSnapshot({
	topology: {
	graphId: 'graph-invalid',
	operators: [{ operatorId: 'one', sessionId: 'different-session' }, binding(two, 'two')],
	edges: [{ edgeId: 'one-two', fromOperatorId: 'one', toOperatorId: 'two' }],
	},
	records: projection.records,
	}),
	/Topology operator one is bound to different-session/,
	);
	assert.throws(
	() =>
	buildAgentGraphTraceSnapshot({
	topology: {
	graphId: 'different-graph',
	operators: [binding(one, 'one'), binding(two, 'two')],
	edges: [{ edgeId: 'one-two', fromOperatorId: 'one', toOperatorId: 'two' }],
	},
	records: projection.records,
	}),
	/cannot observe records from graph graph-invalid/,
	);
	assert.throws(
	() =>
	buildAgentGraphTraceSnapshot({
	topology: {
	graphId: 'graph-invalid',
	operators: [binding(two, 'two')],
	edges: [],
	},
	records: projection.records,
	}),
	/unknown topology operator one/,
	);
	});
	});

	/** One still-open invocation, as its opening fact describes it. */
	function runInvocation(name: string, openedAt: number): RuntimeInvocationRecord {
	return testInvocationRecord({
	sessionId: `session-${name}`,
	invocationId: `invocation-${name}`,
	runId: `run-${name}`,
	turnId: `turn-${name}`,
	openedAt,
	});
	}

	function binding(run: RuntimeInvocationRecord, operatorId: string) {
	return { operatorId, sessionId: run.sessionId };
	}

	function stream(run: RuntimeInvocationRecord, operatorId: string, events: readonly RuntimeEvent[]) {
	return {
	operator: binding(run, operatorId),
	run,
	events,
	};
	}

	function runtimeEvent(
	run: RuntimeInvocationRecord,
	id: string,
	ts: number,
	text: string,
	): RuntimeEvent {
	return {
	id,
	invocationId: run.invocationId,
	runId: run.runId,
	sessionId: run.sessionId,
	turnId: run.turnId,
	ts,
	partial: false,
	role: 'model',
	author: 'agent',
	content: { kind: 'text', text },
	};
	}