feat(graph)!: native Dynamo trace replay -- Graph IR agentic workload lane with unified segment store and session routing#1132
Draft
ajcasagrande wants to merge 7 commits into
Draft
feat(graph)!: native Dynamo trace replay -- Graph IR agentic workload lane with unified segment store and session routing#1132ajcasagrande wants to merge 7 commits into
ajcasagrande wants to merge 7 commits into
Conversation
…nt store, graph runtime, replay timing, session routing
Add a full agentic-workload benchmarking lane to AIPerf: LLM workflows
represented as dataflow graphs (LLM nodes wired by static edges, reading
and writing channels, with traces supplying initial state) instead of
flat request lists or linear conversations.
Ingest / IR (aiperf.dataset.graph): ParsedGraph schema, parser,
structural and semantic validators, and adapters lowering weka_trace,
dynamo_trace, native YAML/JSONL, and dag_jsonl workloads onto one IR.
Node prompts lower into a content-addressed unified segment store keyed
by (trace_id, node_ordinal, phase_variant) with a graph_meta sidecar;
graph runs broadcast DatasetMetadata.graph + GraphSegmentClientMetadata
(mandatory sidecar, no stub conversations). Parse dispatch is
registry-driven through one GraphParseContext; GraphStoreBuilder owns
the store build. Dynamo captures build per session-tree (subagent
linkage via parent_trajectory_id) through a fused read+build parallel
path; weka loaders unify on one streaming work-item dispatch and seed
ladder. Trie emission splices content-parent segment chains; hash-int
and segment-id interning plus incremental content spill bound
corpus-scale build memory.
Runtime (aiperf.graph): executor, scheduler, channel store, credit
dispatch adapter, dynamic pools, and worker materialization -- graph
credits materialize payloads on workers from the unified store, filling
dynamic slots from ancestor responses at run time. Identity follows the
legacy contract: data-inherent {scope}:{turn} node ids, per-trajectory
x_correlation_id, instance-keyed sticky sessions with whole-tree
co-placement. Per-call body/header params are Turn-named native node
fields (model, max_tokens, raw_tools, extra_headers, extra_body,
theoretical prefix-cache counts).
Timing: graph_ir_replay strategy replays recorded inter-node pacing
(shared idle-gap warp, both recorded formats), with a scenario-scoped
t* snapshot window, extended-warmup cache-pressure stage with a
profiling handoff, warmup failure aborts, dataset selection
(--num-dataset-entries, --max-context-length, --allow-dataset-wrap,
sampling strategies) behind a fail-loud wrap-guard, and single-pass
semantics for bare graph runs. Recorded output lengths pin generation
caps on both adapters.
Session routing: a session_routing plugin category (--session-routing)
unifies router-affinity signaling (dynamo_headers, dynamo_nvext with
the v1.2 open contract, smg_routing_key, session_id_header) at the
request-serialization chokepoint, with per-request lineage/finality
facts from SessionTreeRegistry, wired into both the linear and graph
planes.
Fidelity gates: dag_jsonl byte-parity vs the legacy plane, weka/dynamo
cross-format parity (one recording, two encodings, one lowering),
golden store digests, and live mock-server E2E runs. Docs: Agentic
Workloads user guide (docs/benchmark-modes/agentic.md) plus graph-*
reference internals.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Signed-off-by: Anthony Casagrande <acasagrande@nvidia.com>
ajcasagrande
requested review from
FrankD412,
debermudez,
ganeshku1,
ilana-n,
lkomali and
matthewkotila
as code owners
Try out this PRQuick install: pip install --upgrade --force-reinstall git+https://github.com/ai-dynamo/aiperf.git@5f5cdf7f7f24f3dcac3ed221a7043d25350fcbb7Recommended with virtual environment (using uv): uv venv --python 3.12 && source .venv/bin/activate
uv pip install --upgrade --force-reinstall git+https://github.com/ai-dynamo/aiperf.git@5f5cdf7f7f24f3dcac3ed221a7043d25350fcbb7Last updated for commit: |
This comment has been minimized.
This comment has been minimized.
ajcasagrande
changed the title
…tale artifacts
Fixes every failing check from the branch's first CI run (the branch was
never pushed before the squash):
- chat CLI (linux+windows): main's new `aiperf chat` builds records
without request.streamed, and this branch gates TTFT/ITL behind the
streamed_request predicate -- every stat rendered "(no tokens
received)". chat hardcodes stream=True on the wire, so build_record
now stamps streamed=True.
- Windows HF repo ids: a weka HF `org/name` id that round-trips through
Path flips to backslashes on Windows, failing the single-slash repo-id
regex -- the id then resolved as a local path (FileNotFoundError in
the DatasetResolver, no adapter match in parse, scenario
require_loader/hf-repo pins never engaging). New _hf_dataset_id_str
normalizes separators (IS_WINDOWS-gated) in _looks_like_hf_dataset_id,
both _load_hf_rows call sites, and pin_weka_hf_repo.
- Windows tests: the forkserver cached-start test skips on Windows (no
forkserver context exists to construct); the corpus-scale memory
module uses pytest.importorskip("resource") (POSIX-only RSS
accounting).
- fern/MDX: rewrap a backtick code span in
graph-ingest-build-pipeline.md whose line break put `<=` at line
start, which the MDX parser reads as a JSX tag opener.
- pre-commit: regenerate the stale aiperf-config.schema.json (missing
agentic_cache_warmup_duration + drifted descriptions); extract
Worker._resolve_graph_session_headers to bring _process_graph_credit
back under the C901 threshold; accept CreditPhaseConfig (31 fields)
into the ergonomics baseline -- the sub-model split ripples across
every phase-config consumer and is deferred rather than rushed into a
CI fix.
Full unit suite: 15447 passed, 95 skipped.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Signed-off-by: Anthony Casagrande <acasagrande@nvidia.com>
…ang-to-failure timeouts Round 2 of first-CI-run fixes: - orchestrator: _execute_sync called _failure_from_subprocess / _build_result_from_metrics with 5 positional args against their 3-positional + keyword-only signatures -- a TypeError on every sweep / multi-run / adaptive execution path (all 25 integration failures). Unit tests never reach this seam; the integration tier does. Call with keyword args; test_multi_run_basic verified green locally. - component tests: the authored-delay lower bound gains a Windows timer allowance (two ~15.6ms ticks; the loop's timer clock can fire early relative to the perf_counter seam timestamps -- observed 237.6ms for an authored 250ms); the duplication-report execute_phase ceilings go 15s -> 60s for loaded Windows runners. - CI: unit + component tiers now run under pytest-timeout (--timeout 600 --timeout-method thread) on POSIX and Windows, so a hung worker fails loudly with a named test instead of silently burning the job budget (linux/macos 3.11 stalled at ~95% for 26 minutes this run); job budget 30 -> 45 minutes for the windows-x64 runners that finish component tests just past the old cap. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Signed-off-by: Anthony Casagrande <acasagrande@nvidia.com>
…nes recycling unbounded Both remaining 3.11 CI failures were one bug: the duration-budget wrappers end their stage by cancelling the lane fan-out task (asyncio.wait_for), but on Python 3.11 a cancel delivered into a TaskGroup whose children complete constantly can be lost mid-abort (rewritten internals make 3.12+ immune). With an instantly-completing issuer the recycle loop then runs unbounded: the winarm duplication test minted 287,685 instances past its 0.1s budget before the outer 60s ceiling fired, and the linux pressure test recycled until the xdist worker died -- the silent ~95% unit-suite stall the earlier pytest-timeout instrumentation converted into a named worker crash. Lanes now ALSO check a loop-clock deadline cooperatively: the two budget wrappers stamp self._duration_deadline before dispatch (cleared in finally so a stage budget never leaks into the next stage), and both recycle loops (_run_lanes lane loop, _run_pressure_lanes lane loop) return once past it. The wait_for cancel stays as the fast path for lanes parked mid-instance on recorded idle delays; the cooperative check makes stage end independent of cancellation delivery. Unit suite 15447 passed; graph component suite 116 passed. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Signed-off-by: Anthony Casagrande <acasagrande@nvidia.com>
…n to pin The fork-minimal byte-parity gate flaked on 3.13 cells with the LEGACY plane sending a fork child bare (no inherited history). Root cause: with two children forking off the parent's terminal turn, an early child can complete its whole conversation (pin -> seed -> run -> release) before the sibling's credit reaches the worker. release_fork_child then sees refcount 0 with pending_fork_eviction set and evicts the parent -- the late sibling hits the "arrived after parent was evicted" branch and dispatches with no seed context. Python 3.13's task scheduling widens the window enough to fire ~1-in-3 suite runs; 3.11/3.12 kept it latent. UserSession now stamps fork_children_expected (count of declared FORK-mode branches, same wire-round-trip rationale as is_fork_parent) and counts fork_children_pinned; release_fork_child collects a pending-eviction parent only once every declared child has pinned, so refcount 0 no longer conflates "all children joined" with "a sibling has not arrived yet". Regression tests pin the late-sibling sequence directly on the session manager (parent must survive the first child's full lifecycle) plus the mid-conversation no-pending case. Parity suite on Python 3.13: 20/20 clean runs (previously ~1-in-3 failed). Full unit suite 15449 passed. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Signed-off-by: Anthony Casagrande <acasagrande@nvidia.com>
…n accounting The previous fix gated pending-eviction collection on every declared FORK child having pinned, but counted ConversationBranchInfo ENTRIES. The dag_jsonl loader packs all forks declared on one turn into a SINGLE entry whose child_conversation_ids lists every child, so a two-child fork read expected=1: the first child's release still evicted the parent before the sibling's credit arrived, reproducing the exact fork-minimal parity failure on the next CI run (windows 3.13). Count children across FORK entries instead. The regression test now parametrizes both shapes -- one-entry-per-child AND the real loader's single-entry-multi-children packing (the variant that would have caught the miscount). Parity suite on Python 3.13: 30/30 clean; full unit suite 15450 passed. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Signed-off-by: Anthony Casagrande <acasagrande@nvidia.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Tip
Q: How does this compare to AIPerf - AgentX?
A: This branch contains more Dynamo specific functaionality than AgentX, but also includes a graph-native alternative implementation of core AgentX concepts, which may be based on this runtime in the future. TBD.
Summary
Native Dynamo trace replay: point AIPerf at a recorded Dynamo capture (
.jsonl/.jsonl.gz, segmentedtrace.NNNNNN.jsonl.gzfiles, or a directory of them) and replay it faithfully against any OpenAI-compatible endpoint — recorded topology, pacing, token lengths, prefix-cache structure, and session identity included. No conversion step:aiperf profile \ --model my-model \ --url http://localhost:8000 \ --endpoint-type chat \ --input-file ./captures/trace.jsonl.gz \ --streaming \ --tokenizer builtin \ --random-seed 1234 \ --num-dataset-entries 50 \ --num-conversations 50 \ --concurrency 8 \ --concurrency-ramp-duration 60 \ --workers-max 8 \ --session-routing dynamo_headers \ --benchmark-duration 600 \ --artifact-dir ./artifacts/dynamo-replay \ --ui simpleThe capture is auto-detected as
dynamo_trace(pass--graph-format dynamo_traceto force it explicitly).--concurrency-ramp-durationperforms a lane-level ramp on the graph replay plane: replay lanes park at phase start and are admitted 1 →--concurrencyover the ramp window, spreading load onto a cold server.Dynamo replay rides a new general agentic-workload lane (Graph IR): LLM workflows represented as dataflow graphs (LLM nodes wired by static edges, reading and writing channels) instead of flat request lists or linear conversations. The same lane also ingests
weka_trace, hand-authored native graph YAML/JSONL, and legacydag_jsonlfiles.Dynamo trace replay
dynamo.request.trace.v1records lower per session-tree (root + descendants linked viaagent_context.parent_trajectory_id), so independent trees never share causality edges; schema-less uploader marker lines are tolerated; mixed multi-file captures group cross-file trees correctly.--synthesis-idle-gap-cap), interval-order causality edges enforce recorded finished-before relations, and recorded output lengths pin per-request generation caps.--session-routing dynamo_headersstampsX-Dynamo-Session-ID/parent headers;--session-routing dynamo_nvextemitsnvext.session_controlbody metadata, withcontract=openmatching the released Dynamo v1.2.x contract (open once, then baresession_id) andcontract=bind(default) matching >= v1.3.0-dev re-bind-per-turn.Build plane at a glance
flowchart LR subgraph sources["Workload sources"] dynamo["dynamo_trace<br/>.jsonl / .jsonl.gz capture"] weka["weka_trace<br/>.json / dir / HF corpus"] native["native<br/>graph YAML / JSONL"] dag["dag_jsonl<br/>legacy DAG files"] end subgraph ingest["Ingest: aiperf.dataset.graph"] ctx["GraphParseContext<br/>run knobs, tri-state idle-gap cap"] adapters["graph_adapter registry<br/>parse(path, ctx)"] ir["ParsedGraph IR<br/>LlmNodes + edges + channels"] end subgraph build["GraphStoreBuilder"] store["unified segment store<br/>content-addressed, mmap"] sidecar["graph_meta sidecar"] end dynamo --> adapters weka --> adapters native --> adapters dag --> adapters ctx --> adapters adapters --> ir ir --> store ir --> sidecar store --> bc["DatasetMetadata.graph +<br/>GraphSegmentClientMetadata<br/>broadcast"] sidecar --> bcSupporting infrastructure (Graph IR lane)
Ingest / IR (
aiperf.dataset.graph)ParsedGraphschema, parser, structural and semantic validators, and adapters loweringdynamo_trace,weka_trace, native YAML/JSONL, anddag_jsonlonto one IR.(trace_id, node_ordinal, phase_variant)with agraph_metasidecar; graph runs broadcastDatasetMetadata.graph+GraphSegmentClientMetadata(mandatory sidecar, no stub conversations).GraphParseContext;GraphStoreBuilderowns the store build; trie emission splices content-parent segment chains for corpus-scale CPU bounds.Runtime (
aiperf.graph){scope}:{turn}node ids, per-trajectoryx_correlation_id, instance-keyed sticky sessions with whole-tree co-placement.model,max_tokens,raw_tools,extra_headers,extra_body, theoretical prefix-cache counts).sequenceDiagram participant TM as TimingManager<br/>graph_ir_replay participant R as StickyCreditRouter participant W as Worker participant SR as session_routing plugin participant S as Inference server TM->>TM: replay recorded pacing<br/>idle-gap warp, t* window TM->>R: credit (trace instance, x_correlation_id) R->>W: route (instance pinned to ONE worker) W->>W: materialize payload from unified store<br/>fill dynamic slots from ancestor responses W->>SR: headers() / transform_body() at serialization SR-->>W: session identity (headers or nvext body) W->>S: HTTP request S-->>W: streamed response W->>W: capture reply into dynamic pool W-->>TM: credit return (unblocks dependent nodes)Timing
graph_ir_replaystrategy with a scenario-scoped t* snapshot window, extended-warmup cache-pressure stage with a profiling handoff, and warmup failure aborts.--num-dataset-entries,--max-context-length,--allow-dataset-wrap, sampling strategies) behind a fail-loud wrap-guard; single-pass semantics for bare graph runs.Session routing
session_routingplugin category (--session-routing) unifying router-affinity signaling (dynamo_headers,dynamo_nvext,smg_routing_key,session_id_header) at the request-serialization chokepoint, with per-request lineage/finality facts fromSessionTreeRegistry, wired into both the linear and graph planes.x-dynamo-session-id; pair runs with--session-routing dynamo_headersto restore stamping (now available for any dataset and endpoint).Fidelity gates
dag_jsonlbyte-parity vs the legacy plane, golden store digests, and live mock-server E2E runs.Documentation
User guides (rendered on this branch):
Reference internals:
🤖 Generated with Claude Code