diff --git a/config/perception/m4-reference-graph-v2.json b/config/perception/m4-reference-graph-v2.json
new file mode 100644
index 0000000..8cea51a
--- /dev/null
+++ b/config/perception/m4-reference-graph-v2.json
@@ -0,0 +1,95 @@
+{
+  "schema_version": "missioncore.reference-perception-graph-config/v2",
+  "graph_id": "reference-perception-graph/v2",
+  "source_profile_id": "m4-ravnoves00-recorded-realtime/v1",
+  "providers": [
+    {
+      "role": "source",
+      "provider_id": "ravnoves00-recorded-source/v1",
+      "version": "1.0.0",
+      "revision": "m4-ravnoves00-recorded-realtime/v1",
+      "sha256": "ea10359339e6cce31b5780a2710299771cab7cc0c1c2a2b56a1621f786b31fa8"
+    },
+    {
+      "role": "detector",
+      "provider_id": "triton-yolox-s-raw-kb4/v1",
+      "version": "1.0.0",
+      "revision": "yolox_s:1",
+      "sha256": "ea10359339e6cce31b5780a2710299771cab7cc0c1c2a2b56a1621f786b31fa8"
+    },
+    {
+      "role": "geometry",
+      "provider_id": "ravnoves00-geometry-association/v1",
+      "version": "1.0.0",
+      "revision": "m4-ravnoves00-e29-e32-geometry/v1",
+      "sha256": "cc666c9389a5e221957faddec89584709b66918d14abaf646f1832e001421999"
+    },
+    {
+      "role": "temporal",
+      "provider_id": "bounded-spatial-temporal-layer/v1",
+      "version": "1.0.0",
+      "revision": "m4-bounded-temporal-motion/v1",
+      "sha256": "7130eaee24a95c7d888bf7598010e03e129e1c3ac5b34bcd8401015ff4244b39"
+    },
+    {
+      "role": "motion",
+      "provider_id": "class-independent-motion-estimator/v1",
+      "version": "1.0.0",
+      "revision": "m4-bounded-temporal-motion/v1",
+      "sha256": "7130eaee24a95c7d888bf7598010e03e129e1c3ac5b34bcd8401015ff4244b39"
+    },
+    {
+      "role": "rolling",
+      "provider_id": "rolling-local-obstacle-map/v1",
+      "version": "1.0.0",
+      "revision": "ravnoves00-rolling-local-obstacle-map/v1",
+      "sha256": "f7e3315eaf6ffaf3aee1e04913933812092cf82bbcc9984c1a6fa2d9250e6784"
+    },
+    {
+      "role": "threat",
+      "provider_id": "dual-evidence-replay-threat/v3",
+      "version": "3.0.0",
+      "revision": "m4-ravnoves00-virtual-corridor/v3",
+      "sha256": "8c3a5aa837da1f028f5998fb504a1381f9b2b68de6420a32160410b6dc0887c7"
+    }
+  ],
+  "queues": [
+    {
+      "stage_id": "detector",
+      "capacity": 2,
+      "deadline_ns": 1000000000,
+      "terminal_timeout_ns": 90000000000
+    },
+    {
+      "stage_id": "geometry",
+      "capacity": 2,
+      "deadline_ns": 1500000000,
+      "terminal_timeout_ns": 90000000000
+    },
+    {
+      "stage_id": "temporal",
+      "capacity": 2,
+      "deadline_ns": 1750000000,
+      "terminal_timeout_ns": 90000000000
+    },
+    {
+      "stage_id": "rolling",
+      "capacity": 2,
+      "deadline_ns": 2000000000,
+      "terminal_timeout_ns": 90000000000
+    },
+    {
+      "stage_id": "threat",
+      "capacity": 2,
+      "deadline_ns": 2250000000,
+      "terminal_timeout_ns": 90000000000
+    }
+  ],
+  "authority": {
+    "mode": "replay-simulated",
+    "physical_live": false,
+    "commands_enabled": false,
+    "actuation_allowed": false,
+    "navigation_or_safety_accepted": false
+  }
+}
diff --git a/docs/22_MILESTONE_4_OBJECT_CENTRIC_RECORDED_REALTIME_CV.md b/docs/22_MILESTONE_4_OBJECT_CENTRIC_RECORDED_REALTIME_CV.md
index 0aa7112..2c5efa7 100644
--- a/docs/22_MILESTONE_4_OBJECT_CENTRIC_RECORDED_REALTIME_CV.md
+++ b/docs/22_MILESTONE_4_OBJECT_CENTRIC_RECORDED_REALTIME_CV.md
@@ -2,7 +2,9 @@
 
 Date: 2026-08-05
 
-Status: in progress; M4.0–M4.6 accepted, E47 diagnostic semantic/SLAM shadow is parallel, M4.7 Worker 006 cutover remains next
+Status: in progress; M4.0–M4.6 accepted, the M4.7 canonical graph and Worker
+006 shadow artifact are implemented locally, and Worker preflight/full shadow
+acceptance plus durable cutover remain open
 
 Audit base: `1b3e0b3` on `feat/simulation-polygon-s1`
 
@@ -111,7 +113,7 @@ Those systems remain separate platform workstreams.
 | Realtime worker | Worker 006 has healthy Triton and persistent perception containers plus live Telegraf | Infrastructure exists |
 | Worker graph | The persistent process still executes `run_e15_shadow_inference.py serve` with E15/E19/E8/E3/E23 profiles | Open architectural blocker: runtime remains LAB-generation-specific |
 | Model service | Canonical Triton currently exposes pinned `yolox_s` and `pointpillars`; PointPillars was rejected as a K1 product candidate | Reuse `yolox_s`; do not reopen PointPillars |
-| CV provider interfaces | No executable `DetectorProvider`, `GeometryProvider`, `TemporalProvider` or object-map contract exists in the product package | Open architectural blocker |
+| CV provider interfaces | Product-owned source, detector, geometry, temporal, motion, rolling-map and threat protocols plus `LocalObstacleMap` exist under `k1link.perception` | Closed in code; Worker evidence remains open |
 | LAB execution | 33 evidence definitions are fully classified; M4, E33, E35 and E46J are canonical canaries, E47 is one bounded experimental adapter, and 28 remain read-only legacy | Closed for LAB governance; no bulk legacy migration |
 | Production import graph | 93 compute modules (72,456 lines), 51 web modules and many direct `web → compute.e*/l*` imports remain | Do not clean wholesale; prohibit new dependencies and isolate the new graph |
 | Telemetry | Generic pipeline telemetry, multi-contour enrollment, bounded journal and current Worker 006 Telegraf service exist | Reuse; add graph-stage and object-state metrics |
@@ -222,6 +224,9 @@ LiveK1Source (later) ──┘      │
                                                     MotionEstimator
                                                                │
                                                                v
+                                                    RollingLocalObstacleMap
+                                                               │
+                                                               v
                                                     LocalObstacleMap
                                                                │
                                                    virtual rig/corridor (M4 only)
@@ -495,6 +500,11 @@ Physical mounted threat acceptance remains outside Milestone 4.
 
 ### M4.7 — cut Worker 006 over to the canonical graph
 
+Status: implementation complete locally on 2026-08-23; Worker 006 preflight,
+full lossless shadow evidence and durable process replacement are not yet
+accepted. The current E15 worker and Triton identities remain the rollback
+predecessor. No K1, Zarya or connection-stack change is part of this phase.
+
 Deliverables:
 
 - package one digest-bound worker artifact containing the graph runner, provider
@@ -655,7 +665,7 @@ is requested. None may be silently claimed from a successful recorded replay.
 
 Milestone 4 is complete only when all of the following are true:
 
-- [ ] One versioned reference graph exists outside experiment-named modules.
+- [x] One versioned reference graph exists outside experiment-named modules.
 - [ ] RAVNOVES00 runs through the same graph contract intended for a live source.
 - [ ] Semantic class and persistent detector ID are optional diagnostics only.
 - [ ] Camera, fused and geometry-only obstacles share one canonical observation
@@ -1135,6 +1145,51 @@ after indexing, a 12-frame spatial chunk was served in approximately `33 ms` and
 was about `1.27 MB`. M4.7 may proceed only from the corrected identities above;
 physical-live, collision, navigation and actuation authority remain false.
 
+### 2026-08-23 — M4.7 canonical graph and shadow release implementation
+
+`reference-perception-graph/v2` is now the product-owned assembly contract. It
+composes the admitted recorded source, frozen YOLOX detector, current geometry,
+bounded temporal state, class-independent motion, rolling local occupancy and
+replay threat providers without importing an E/L-named module. Rolling retained
+occupancy is an explicit stage rather than an offline side computation, so the
+published map preserves the accepted M4.5R/M4.6 rule: current plus retained is
+occupied, held/expired is unknown, and missing republication never means free.
+
+The graph has two explicit bounded execution policies:
+
+- `source-paced-latest-wins` for the later 1× realtime canary, with every
+  superseded or stale packet receiving a terminal outcome;
+- `lossless-replay` for deterministic parity, with bounded backpressure and
+  zero accepted supersession.
+
+Every stage is deadline-checked before and after provider work, queue
+high-watermarks are reported, terminal accounting is closed, and V2 restart is
+fail-closed unless all stateful providers are freshly instantiated. A production
+assembly boundary validates the digest of every provider profile and takes only
+explicit paths to the camera index, video, source pack, local surface, valid-FOV
+mask and Triton origin. The CLI writes immutable content-addressed frame,
+terminal, report and manifest ledgers; acceptance requires exactly `4,489`
+delivered frames with zero failed, stale, superseded, rejected or unavailable
+terminal outcomes.
+
+The new Worker package uses transition `m47-canonical-graph-shadow-v1`. It is a
+separate deterministic artifact and does not relabel the accepted historical
+`m4-detector-shadow-v1` wheel. Its PowerShell runner verifies release files,
+inputs, dependency trees, disk reserve, pinned Worker/Triton predecessor and
+container isolation before it can start the one-shot lossless graph shadow. It
+adds no public port, never changes Triton, removes the candidate container after
+the run and re-verifies the predecessor. Provider readiness and graph readiness
+are emitted separately. The production builder refuses a dirty worktree so an
+artifact cannot claim a Git revision which does not contain its wheel.
+
+Local contract, graph, result-sealing, artifact and historical-rollback tests
+pass. This is implementation evidence only. It does not claim that Worker 006
+has the pinned local-surface input, that preflight has passed, that the 4,489
+frame graph shadow matches the accepted M4.5R/M4.6 ledgers, or that the durable
+E15 command has been replaced. Those are the next M4.7 acceptance actions, in
+that order. K1, Zarya, the stable connection path and physical-live authority
+remain untouched.
+
 ## Implementation order
 
 The implementation sequence is intentionally strict:
diff --git a/scripts/Invoke-M47CanonicalGraphShadow.ps1 b/scripts/Invoke-M47CanonicalGraphShadow.ps1
new file mode 100644
index 0000000..91910d0
--- /dev/null
+++ b/scripts/Invoke-M47CanonicalGraphShadow.ps1
@@ -0,0 +1,376 @@
+[CmdletBinding()]
+param(
+    [Parameter(Mandatory = $true)]
+    [string]$ReleaseRoot,
+    [Parameter(Mandatory = $true)]
+    [string]$ArtifactPath,
+    [Parameter(Mandatory = $true)]
+    [ValidatePattern("^[a-f0-9]{64}$")]
+    [string]$ExpectedArtifactSha256,
+    [string]$OutputRoot = "D:\NDC_MISSIONCORE\runtime\results\m47-reference-graph",
+    [ValidateRange(1, 1000)]
+    [int]$FreeGiBFloor = 300,
+    [switch]$PreflightOnly
+)
+
+$ErrorActionPreference = "Stop"
+$ProgressPreference = "SilentlyContinue"
+
+function Assert-LastExitCode([string]$Operation) {
+    if ($LASTEXITCODE -ne 0) {
+        throw "$Operation failed with exit code $LASTEXITCODE"
+    }
+}
+
+function Get-Sha256([string]$Path) {
+    return (Get-FileHash -LiteralPath $Path -Algorithm SHA256).Hash.ToLowerInvariant()
+}
+
+function Assert-FileSha256([string]$Path, [string]$Expected, [string]$Label) {
+    $item = Get-Item -LiteralPath (Resolve-Path -LiteralPath $Path).Path -Force
+    if ($item.PSIsContainer -or ($item.Attributes -band [IO.FileAttributes]::ReparsePoint)) {
+        throw "$Label must be a regular file"
+    }
+    $observed = Get-Sha256 $item.FullName
+    if ($observed -cne $Expected) {
+        throw "$Label SHA-256 changed: expected $Expected, observed $observed"
+    }
+    return $item.FullName
+}
+
+function Resolve-DDirectory([string]$Path, [string]$Label, [bool]$Create) {
+    if ($Create -and -not (Test-Path -LiteralPath $Path)) {
+        $null = New-Item -ItemType Directory -Path $Path
+    }
+    $item = Get-Item -LiteralPath (Resolve-Path -LiteralPath $Path).Path -Force
+    $root = [IO.Path]::GetPathRoot($item.FullName).TrimEnd("\")
+    if (
+        -not $item.PSIsContainer -or
+        ($item.Attributes -band [IO.FileAttributes]::ReparsePoint) -or
+        $root -ine "D:"
+    ) {
+        throw "$Label must be a real D: directory"
+    }
+    return $item.FullName
+}
+
+function Convert-ToDockerPath([string]$Path) {
+    return $Path.Replace("\", "/")
+}
+
+function Assert-FreeSpace([string]$Phase) {
+    $free = [int64](Get-PSDrive -Name D).Free
+    $floor = [int64]$FreeGiBFloor * 1GB
+    Write-Host (
+        "DISK_GUARD PHASE={0} DRIVE=D FREE_BYTES={1} FREE_GIB={2} FLOOR_GIB={3}" -f
+        $Phase, $free, [math]::Round($free / 1GB, 3), $FreeGiBFloor
+    )
+    if ($free -lt ($floor + 1GB)) {
+        throw "D: lacks the guarded M4.7 reserve during $Phase"
+    }
+    return $free
+}
+
+function Get-ContainerIdentity([string]$Name) {
+    $json = & docker inspect $Name
+    Assert-LastExitCode "Docker inspection for $Name"
+    $rows = @($json | ConvertFrom-Json)
+    if ($rows.Count -ne 1) {
+        throw "Docker identity for $Name is not unique"
+    }
+    return $rows[0]
+}
+
+function Assert-ContainerIdentity(
+    [string]$Name,
+    [string]$ExpectedId,
+    [string]$ExpectedImageId,
+    [bool]$RequireHealthy
+) {
+    $container = Get-ContainerIdentity $Name
+    if (
+        $container.Id -cne $ExpectedId -or
+        $container.Image -cne $ExpectedImageId -or
+        -not $container.State.Running
+    ) {
+        throw "$Name identity or running state changed"
+    }
+    if ($RequireHealthy -and $container.State.Health.Status -cne "healthy") {
+        throw "$Name is not healthy"
+    }
+    return $container
+}
+
+function Get-DirectoryTreeSha256([string]$Root, [object[]]$Files) {
+    $rootPrefix = $Root.TrimEnd("\") + "\"
+    $rows = @(
+        foreach ($file in $Files) {
+            if (-not $file.FullName.StartsWith($rootPrefix, [StringComparison]::OrdinalIgnoreCase)) {
+                throw "M4.7 dependency file escaped its declared root"
+            }
+            [pscustomobject]@{
+                Relative = $file.FullName.Substring($rootPrefix.Length).Replace("\", "/")
+                File = $file
+            }
+        }
+    )
+    $rows = @($rows | Sort-Object -Property Relative -CaseSensitive)
+    $digest = [Security.Cryptography.IncrementalHash]::CreateHash(
+        [Security.Cryptography.HashAlgorithmName]::SHA256
+    )
+    $encoding = New-Object System.Text.UTF8Encoding($false)
+    try {
+        foreach ($row in $rows) {
+            $record = "{0}`t{1}`t{2}`n" -f `
+                $row.Relative, $row.File.Length, (Get-Sha256 $row.File.FullName)
+            $digest.AppendData($encoding.GetBytes($record))
+        }
+        return ([BitConverter]::ToString($digest.GetHashAndReset())).Replace("-", "").ToLowerInvariant()
+    } finally {
+        $digest.Dispose()
+    }
+}
+
+function Write-Utf8NoBom([string]$Path, [string]$Value) {
+    $encoding = New-Object System.Text.UTF8Encoding($false)
+    [IO.File]::WriteAllText($Path, $Value, $encoding)
+}
+
+if ($env:COMPUTERNAME -cne "DESKTOP-OPJ8J04") {
+    throw "M4.7 shadow release is pinned to DESKTOP-OPJ8J04"
+}
+
+$release = Resolve-DDirectory $ReleaseRoot "M4.7 release root" $false
+$payload = Resolve-DDirectory (Join-Path $release "payload") "M4.7 payload" $false
+$artifact = Assert-FileSha256 $ArtifactPath $ExpectedArtifactSha256 "M4.7 release artifact"
+$descriptorPath = Join-Path $payload "mission-core-worker-m47-graph-shadow-v2.json"
+$descriptor = Get-Content -LiteralPath $descriptorPath -Raw | ConvertFrom-Json
+if (
+    $descriptor.schema_version -cne "nodedc.mission-core-worker.shadow-release/v2" -or
+    $descriptor.transition -cne "m47-canonical-graph-shadow-v1" -or
+    $descriptor.component -cne "mission-core-worker" -or
+    $descriptor.host.node -cne $env:COMPUTERNAME -or
+    $descriptor.host.worker_id -cne "worker-006" -or
+    $descriptor.boundary.repository -cne "NODEDC_MISSION_CORE" -or
+    $descriptor.boundary.external_deploy_registry -ne $false -or
+    $descriptor.acceptance.run_mode -cne "lossless-replay" -or
+    $descriptor.acceptance.expected_frames -ne 4489 -or
+    $descriptor.acceptance.delivered_frames -ne 4489 -or
+    $descriptor.acceptance.failed_frames -ne 0 -or
+    $descriptor.acceptance.stale_frames -ne 0 -or
+    $descriptor.acceptance.superseded_frames -ne 0 -or
+    $descriptor.acceptance.accepted_parity -ne $true -or
+    $descriptor.readiness.graph.graph_id -cne "reference-perception-graph/v2" -or
+    $descriptor.readiness.graph.actuation_allowed -ne $false
+) {
+    throw "M4.7 shadow descriptor contract changed"
+}
+
+$null = Assert-FileSha256 $PSCommandPath $descriptor.release.runner.sha256 "M4.7 runner"
+$wheelPath = Assert-FileSha256 (
+    Join-Path $payload $descriptor.release.wheel.name
+) $descriptor.release.wheel.sha256 "M4.7 wheel"
+foreach ($entry in $descriptor.release.configs.PSObject.Properties) {
+    $null = Assert-FileSha256 (
+        Join-Path $payload $entry.Value.name
+    ) $entry.Value.sha256 ("M4.7 config {0}" -f $entry.Name)
+}
+foreach ($entry in $descriptor.inputs.PSObject.Properties) {
+    $null = Assert-FileSha256 $entry.Value.host_path $entry.Value.sha256 (
+        "M4.7 input {0}" -f $entry.Name
+    )
+}
+foreach ($dependency in $descriptor.dependencies) {
+    $root = Resolve-DDirectory $dependency.host_path ("M4.7 dependency {0}" -f $dependency.id) $false
+    $files = @()
+    foreach ($include in $dependency.includes) {
+        $candidate = if ($include -eq ".") { $root } else { Join-Path $root $include }
+        $item = Get-Item -LiteralPath (Resolve-Path -LiteralPath $candidate).Path -Force
+        if ($item.Attributes -band [IO.FileAttributes]::ReparsePoint) {
+            throw "M4.7 dependency $($dependency.id) contains a reparse point"
+        }
+        if ($item.PSIsContainer) {
+            $files += @(Get-ChildItem -LiteralPath $item.FullName -File -Recurse -Force)
+        } else {
+            $files += @($item)
+        }
+    }
+    $files = @($files | Sort-Object -Property FullName -Unique)
+    $bytes = [int64]0
+    foreach ($file in $files) {
+        if ($file.Attributes -band [IO.FileAttributes]::ReparsePoint) {
+            throw "M4.7 dependency $($dependency.id) contains a reparse-point file"
+        }
+        $bytes += [int64]$file.Length
+    }
+    if (
+        $files.Count -ne [int]$dependency.file_count -or
+        $bytes -ne [int64]$dependency.bytes -or
+        (Get-DirectoryTreeSha256 $root $files) -cne [string]$dependency.tree_sha256
+    ) {
+        throw "M4.7 dependency $($dependency.id) inventory changed"
+    }
+}
+
+$imageRef = [string]$descriptor.container.image_ref
+& docker image inspect $imageRef *> $null
+Assert-LastExitCode "Pinned M4.7 image inspection"
+$predecessor = $descriptor.predecessor.durable_worker
+$tritonExpected = $descriptor.predecessor.triton
+$durable = Assert-ContainerIdentity (
+    $predecessor.name
+) $predecessor.container_id $predecessor.image_id $false
+$triton = Assert-ContainerIdentity (
+    $tritonExpected.name
+) $tritonExpected.container_id $tritonExpected.image_id $true
+$output = Resolve-DDirectory $OutputRoot "M4.7 output root" $true
+$freeBefore = Assert-FreeSpace "preflight"
+
+if ($PreflightOnly) {
+    Write-Output ("PATCH_ID={0}" -f $descriptor.patch_id)
+    Write-Output ("ARTIFACT_SHA256={0}" -f $ExpectedArtifactSha256)
+    Write-Output ("DURABLE_WORKER_ID={0}" -f $durable.Id)
+    Write-Output ("TRITON_CONTAINER_ID={0}" -f $triton.Id)
+    Write-Output "PROVIDER_READINESS=accepted"
+    Write-Output "GRAPH_READINESS=not-run"
+    Write-Output "PREFLIGHT=accepted"
+    return
+}
+
+$candidateName = "ndc-mission-core-m47-graph-shadow"
+if (& docker ps -a --format "{{.Names}}" --filter "name=^/$candidateName$") {
+    throw "M4.7 candidate container already exists"
+}
+$scratch = Join-Path $output (".runtime-{0}" -f $descriptor.patch_id)
+if (Test-Path -LiteralPath $scratch) {
+    throw "M4.7 runtime scratch already exists"
+}
+$null = New-Item -ItemType Directory -Path $scratch
+$scratch = Resolve-DDirectory $scratch "M4.7 runtime scratch" $false
+$runtimeIdentityPath = Join-Path $scratch "runtime-identity.json"
+$dockerPayload = Convert-ToDockerPath $payload
+$dockerOutput = Convert-ToDockerPath $output
+$dockerScratch = Convert-ToDockerPath $scratch
+$dockerArguments = @(
+    "create",
+    "--name", $candidateName,
+    "--network", ("container:{0}" -f $tritonExpected.name),
+    "--read-only",
+    "--security-opt", "no-new-privileges:true",
+    "--cap-drop", "ALL",
+    "--pids-limit", "256",
+    "--tmpfs", "/tmp:rw,noexec,nosuid,size=2g",
+    "-e", "PYTHONDONTWRITEBYTECODE=1",
+    "-e", ("PYTHONPATH=/release/{0}:/opt/media:/opt/opencv:/opt/pillow" -f $descriptor.release.wheel.name),
+    "-v", ("{0}:/release:ro" -f $dockerPayload),
+    "-v", ("{0}:/output:rw" -f $dockerOutput),
+    "-v", ("{0}:/run/mission-core:ro" -f $dockerScratch)
+)
+foreach ($entry in $descriptor.inputs.PSObject.Properties) {
+    if ($null -ne $entry.Value.container_path) {
+        $dockerArguments += @(
+            "-v", ("{0}:{1}:ro" -f (
+                Convert-ToDockerPath $entry.Value.host_path
+            ), $entry.Value.container_path)
+        )
+    }
+}
+foreach ($dependency in $descriptor.dependencies) {
+    $dockerArguments += @(
+        "-v", ("{0}:{1}:ro" -f (
+            Convert-ToDockerPath $dependency.host_path
+        ), $dependency.container_path)
+    )
+}
+$dockerArguments += @(
+    "--entrypoint", "python3",
+    $imageRef,
+    "-m", "k1link.perception.reference_graph_cli",
+    "--graph-config", "/release/m4-reference-graph-v2.json",
+    "--baseline-profile", "/release/m4-recorded-realtime-baseline-v1.json",
+    "--geometry-profile", "/release/m4-geometry-association-v1.json",
+    "--temporal-motion-profile", "/release/m4-temporal-motion-v1.json",
+    "--rolling-map-profile", "/release/m4-rolling-local-map-v1.json",
+    "--threat-profile", "/release/m4-replay-threat-v3.json",
+    "--camera-index", $descriptor.inputs.camera_index.container_path,
+    "--source-pack", $descriptor.inputs.source_pack.container_path,
+    "--local-surface", $descriptor.inputs.local_surface.container_path,
+    "--video", $descriptor.inputs.video.container_path,
+    "--valid-fov-mask", $descriptor.inputs.valid_fov_mask.container_path,
+    "--temporal-parity-frames", $descriptor.inputs.accepted_temporal_frames.container_path,
+    "--threat-parity-frames", $descriptor.inputs.accepted_threat_frames.container_path,
+    "--triton-origin", $descriptor.container.triton_origin,
+    "--mode", $descriptor.acceptance.run_mode,
+    "--expected-frames", ([string]$descriptor.acceptance.expected_frames),
+    "--output-root", "/output"
+)
+
+$candidateCreated = $false
+$runFailure = $null
+try {
+    $candidateId = (& docker @dockerArguments).Trim()
+    Assert-LastExitCode "M4.7 candidate creation"
+    if ($candidateId -notmatch "^[a-f0-9]{64}$") {
+        throw "M4.7 candidate id is invalid"
+    }
+    $candidateCreated = $true
+    $candidate = Get-ContainerIdentity $candidateName
+    if (
+        $candidate.Id -cne $candidateId -or
+        $candidate.Image -cne $descriptor.container.image_id -or
+        $candidate.HostConfig.NetworkMode -cne ("container:{0}" -f $triton.Id) -or
+        -not $candidate.HostConfig.ReadonlyRootfs
+    ) {
+        throw "M4.7 candidate isolation contract changed"
+    }
+    $runtimeIdentity = [ordered]@{
+        schema_version = "missioncore.reference-graph-runtime-identity/v1"
+        worker_id = "worker-006"
+        worker_node = $env:COMPUTERNAME
+        worker_container_id = $candidate.Id
+        worker_image_id = $candidate.Image
+        triton_container_id = $triton.Id
+        triton_image_id = $triton.Image
+        artifact_sha256 = $ExpectedArtifactSha256
+        code_revision = $descriptor.code_revision
+        graph_id = $descriptor.readiness.graph.graph_id
+        source_mount_read_only = $true
+        model_service_reused = $true
+        public_worker_port_added = $false
+        commands_enabled = $false
+        actuation_allowed = $false
+    }
+    Write-Utf8NoBom $runtimeIdentityPath ($runtimeIdentity | ConvertTo-Json -Depth 4)
+    Write-Output ("PATCH_ID={0}" -f $descriptor.patch_id)
+    Write-Output ("ARTIFACT_SHA256={0}" -f $ExpectedArtifactSha256)
+    Write-Output ("CANDIDATE_CONTAINER_ID={0}" -f $candidate.Id)
+    Write-Output "PROVIDER_READINESS=accepted"
+    & docker start --attach $candidateName
+    Assert-LastExitCode "M4.7 canonical graph shadow"
+    Write-Output "GRAPH_READINESS=accepted"
+} catch {
+    $runFailure = $_
+} finally {
+    if ($candidateCreated) {
+        & docker rm --force $candidateName *> $null
+        if ($LASTEXITCODE -ne 0 -and $null -eq $runFailure) {
+            $runFailure = "M4.7 candidate cleanup failed"
+        }
+    }
+    Remove-Item -LiteralPath $scratch -Force -Recurse -ErrorAction SilentlyContinue
+}
+
+$null = Assert-ContainerIdentity $predecessor.name $predecessor.container_id (
+    $predecessor.image_id
+) $false
+$null = Assert-ContainerIdentity $tritonExpected.name $tritonExpected.container_id (
+    $tritonExpected.image_id
+) $true
+$freeAfter = Assert-FreeSpace "completed"
+Write-Output ("DISK_FREE_BYTES_BEFORE={0}" -f $freeBefore)
+Write-Output ("DISK_FREE_BYTES_AFTER={0}" -f $freeAfter)
+Write-Output "DURABLE_WORKER_ACTION=none"
+Write-Output "TRITON_ACTION=none"
+if ($null -ne $runFailure) {
+    throw $runFailure
+}
diff --git a/scripts/build_m47_worker_graph_shadow_artifact.py b/scripts/build_m47_worker_graph_shadow_artifact.py
new file mode 100644
index 0000000..d6f8cba
--- /dev/null
+++ b/scripts/build_m47_worker_graph_shadow_artifact.py
@@ -0,0 +1,303 @@
+#!/usr/bin/env python3
+"""Build the deterministic Worker 006 M4.7 canonical-graph shadow artifact."""
+
+from __future__ import annotations
+
+import argparse
+import gzip
+import hashlib
+import io
+import json
+import os
+import re
+import subprocess
+import tarfile
+import tempfile
+from pathlib import Path
+
+REPOSITORY_ROOT = Path(__file__).resolve().parents[1]
+BASE_TEMPLATE = REPOSITORY_ROOT / "config/deployment/mission-core-worker-shadow-v1.template.json"
+RUNNER = REPOSITORY_ROOT / "scripts/Invoke-M47CanonicalGraphShadow.ps1"
+WHEEL_NAME = "nodedc_mission_core-0.1.0-py3-none-any.whl"
+DESCRIPTOR_NAME = "mission-core-worker-m47-graph-shadow-v2.json"
+PATCH_ID = re.compile(r"^[A-Za-z0-9._-]{1,96}$")
+EXPECTED_BASE_TEMPLATE_SHA256 = (
+    "319e7ac7f14e5911ad44234c9ec918c73e11a3406724d5cee3e2ef64bb036e0c"
+)
+CONFIG_PATHS = (
+    Path("config/perception/m4-recorded-realtime-baseline-v1.json"),
+    Path("config/perception/m4-reference-graph-v2.json"),
+    Path("config/perception/m4-geometry-association-v1.json"),
+    Path("config/perception/m4-temporal-motion-v1.json"),
+    Path("config/perception/m4-rolling-local-map-v1.json"),
+    Path("config/perception/m4-replay-threat-v3.json"),
+)
+
+
+class ArtifactBuildError(RuntimeError):
+    pass
+
+
+def sha256_file(path: Path) -> str:
+    digest = hashlib.sha256()
+    with path.open("rb") as handle:
+        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+def git_revision(*, require_clean: bool) -> str:
+    revision_result = subprocess.run(
+        ["git", "rev-parse", "HEAD"],
+        cwd=REPOSITORY_ROOT,
+        check=True,
+        capture_output=True,
+        text=True,
+    )
+    revision = revision_result.stdout.strip()
+    if re.fullmatch(r"[a-f0-9]{40}", revision) is None:
+        raise ArtifactBuildError("Git revision is not a full SHA-1")
+    if require_clean:
+        status = subprocess.run(
+            ["git", "status", "--porcelain"],
+            cwd=REPOSITORY_ROOT,
+            check=True,
+            capture_output=True,
+            text=True,
+        )
+        if status.stdout.strip():
+            raise ArtifactBuildError("production artifact requires a clean worktree")
+    return revision
+
+
+def build_wheel(output: Path) -> Path:
+    environment = os.environ.copy()
+    environment["SOURCE_DATE_EPOCH"] = "0"
+    result = subprocess.run(
+        ["uv", "build", "--wheel", "--out-dir", str(output)],
+        cwd=REPOSITORY_ROOT,
+        env=environment,
+        check=False,
+        capture_output=True,
+        text=True,
+    )
+    if result.returncode != 0:
+        detail = (result.stderr or result.stdout).strip()
+        raise ArtifactBuildError(f"wheel build failed: {detail}")
+    wheel = output / WHEEL_NAME
+    if not wheel.is_file() or wheel.is_symlink():
+        raise ArtifactBuildError("expected wheel was not built")
+    return wheel
+
+
+def render_descriptor(
+    patch_id: str,
+    revision: str,
+    *,
+    wheel_sha256: str,
+) -> bytes:
+    if sha256_file(BASE_TEMPLATE) != EXPECTED_BASE_TEMPLATE_SHA256:
+        raise ArtifactBuildError("historical M4 descriptor base changed")
+    descriptor = json.loads(BASE_TEMPLATE.read_text("utf-8"))
+    descriptor.update(
+        {
+            "schema_version": "nodedc.mission-core-worker.shadow-release/v2",
+            "patch_id": patch_id,
+            "code_revision": revision,
+            "transition": "m47-canonical-graph-shadow-v1",
+            "artifact_type": "shadow-release",
+        }
+    )
+    descriptor["container"].update(
+        {
+            "name": "ndc-mission-core-m47-graph-shadow",
+            "python_path": (
+                f"/release/{WHEEL_NAME}:/opt/media:/opt/opencv:/opt/pillow"
+            ),
+        }
+    )
+    descriptor["inputs"]["local_surface"] = {
+        "container_path": "/source/local-surface/local-surface.npz",
+        "host_path": (
+            "D:\\NDC_MISSIONCORE\\runtime\\derived\\"
+            "k1-local-surface-23762244c8bdb97de26fb721ac957d7a00bc9a63571ac4cfa4be19c4effc7d55"
+            "\\local-surface.npz"
+        ),
+        "sha256": "f57eb2485b6cef47f2a97a2d9ff1aa9fd9265fe1eb69cd5852d12f39e13b8bc6",
+    }
+    descriptor["inputs"]["accepted_temporal_frames"] = {
+        "container_path": "/parity/temporal/frames.jsonl",
+        "host_path": (
+            "D:\\NDC_MISSIONCORE\\runtime\\derived\\"
+            "m4-temporal-replay-81e13d5654ac8d1219f6937dd425f7dbcdbacd5748d23fe653297134f0de6c22"
+            "\\frames.jsonl"
+        ),
+        "sha256": "e83b80ea06b3462c2d04c5a1b74289a0ec401596c7150ae90f5a1b748b639c3a",
+    }
+    descriptor["inputs"]["accepted_threat_frames"] = {
+        "container_path": "/parity/threat/frames.jsonl",
+        "host_path": (
+            "D:\\NDC_MISSIONCORE\\runtime\\derived\\"
+            "m4-threat-replay-2a953c5f27f2a5b1dddc5c658c1de2c323d7796084a099c024987a1da03aa324"
+            "\\frames.jsonl"
+        ),
+        "sha256": "b57be1839f5915e3b80b54355b694e0bd8c9ac318d0cbe6de2bff713082cfa4e",
+    }
+    descriptor["release"] = {
+        "wheel": {"name": WHEEL_NAME, "sha256": wheel_sha256},
+        "runner": {"name": RUNNER.name, "sha256": sha256_file(RUNNER)},
+        "configs": {
+            path.name: {
+                "name": path.name,
+                "sha256": sha256_file(REPOSITORY_ROOT / path),
+            }
+            for path in CONFIG_PATHS
+        },
+    }
+    descriptor["acceptance"] = {
+        "run_mode": "lossless-replay",
+        "expected_frames": 4489,
+        "delivered_frames": 4489,
+        "failed_frames": 0,
+        "stale_frames": 0,
+        "superseded_frames": 0,
+        "accepted_parity": True,
+        "class_routing_used": False,
+    }
+    descriptor["readiness"] = {
+        "provider": [
+            "triton-yolox-s-raw-kb4/v1",
+            "ravnoves00-geometry-association/v1",
+            "bounded-spatial-temporal-layer/v1",
+            "class-independent-motion-estimator/v1",
+            "rolling-local-obstacle-map/v1",
+            "dual-evidence-replay-threat/v3",
+        ],
+        "graph": {
+            "graph_id": "reference-perception-graph/v2",
+            "terminal_accounting_required": True,
+            "actuation_allowed": False,
+        },
+    }
+    if (
+        descriptor["predecessor"]["durable_worker"]["name"]
+        != "ndc-mission-core-perception-worker"
+        or descriptor["rollback"]["durable_worker_action"] != "none"
+        or descriptor["boundary"]["external_deploy_registry"] is not False
+    ):
+        raise ArtifactBuildError("M4.7 predecessor or deployment boundary changed")
+    return (
+        json.dumps(descriptor, ensure_ascii=False, indent=2, sort_keys=True) + "\n"
+    ).encode("utf-8")
+
+
+def _tar_info(path: Path, arcname: str) -> tarfile.TarInfo:
+    info = tarfile.TarInfo(arcname)
+    info.uid = 0
+    info.gid = 0
+    info.uname = "root"
+    info.gname = "root"
+    info.mtime = 0
+    if path.is_dir():
+        info.type = tarfile.DIRTYPE
+        info.mode = 0o755
+    else:
+        info.type = tarfile.REGTYPE
+        info.mode = 0o644
+        info.size = path.stat().st_size
+    return info
+
+
+def write_canonical_archive(stage: Path, target: Path) -> None:
+    members = [stage / "manifest.env", stage / "files.txt", stage / "payload"]
+    members.extend(sorted((stage / "payload").rglob("*")))
+    target.parent.mkdir(parents=True, exist_ok=True)
+    with (
+        target.open("wb") as raw,
+        gzip.GzipFile(filename="", mode="wb", fileobj=raw, compresslevel=9, mtime=0) as gz,
+        tarfile.open(fileobj=gz, mode="w", format=tarfile.PAX_FORMAT) as archive,
+    ):
+        for path in members:
+            info = _tar_info(path, path.relative_to(stage).as_posix())
+            if path.is_file():
+                with path.open("rb") as source:
+                    archive.addfile(info, source)
+            else:
+                archive.addfile(info, io.BytesIO())
+
+
+def build_artifact(
+    patch_id: str,
+    output_directory: Path,
+    *,
+    revision: str | None = None,
+) -> dict[str, object]:
+    if PATCH_ID.fullmatch(patch_id) is None:
+        raise ArtifactBuildError("patch id is invalid")
+    selected_revision = revision or git_revision(require_clean=True)
+    if re.fullmatch(r"[a-f0-9]{40}", selected_revision) is None:
+        raise ArtifactBuildError("artifact revision is invalid")
+    with tempfile.TemporaryDirectory(prefix="mission-core-m47-graph-shadow-") as directory:
+        stage = Path(directory)
+        payload = stage / "payload"
+        payload.mkdir()
+        wheel = build_wheel(stage / "wheel")
+        wheel_sha256 = sha256_file(wheel)
+        payload_files = [RUNNER.name, WHEEL_NAME, DESCRIPTOR_NAME]
+        (payload / RUNNER.name).write_bytes(RUNNER.read_bytes())
+        (payload / WHEEL_NAME).write_bytes(wheel.read_bytes())
+        for relative in CONFIG_PATHS:
+            source = REPOSITORY_ROOT / relative
+            (payload / source.name).write_bytes(source.read_bytes())
+            payload_files.append(source.name)
+        (payload / DESCRIPTOR_NAME).write_bytes(
+            render_descriptor(
+                patch_id,
+                selected_revision,
+                wheel_sha256=wheel_sha256,
+            )
+        )
+        payload_files = sorted(payload_files)
+        (stage / "manifest.env").write_text(
+            f"id={patch_id}\ncomponent=mission-core-worker\ntype=shadow-release\n",
+            "utf-8",
+        )
+        (stage / "files.txt").write_text("\n".join(payload_files) + "\n", "utf-8")
+        target = (
+            output_directory.resolve()
+            / f"nodedc-mission-core-worker-{patch_id}.tgz"
+        )
+        write_canonical_archive(stage, target)
+    return {
+        "ok": True,
+        "patch_id": patch_id,
+        "component": "mission-core-worker",
+        "type": "shadow-release",
+        "artifact": str(target),
+        "sha256": sha256_file(target),
+        "code_revision": selected_revision,
+        "wheel_sha256": wheel_sha256,
+        "payload_files": payload_files,
+        "transition": "m47-canonical-graph-shadow-v1",
+    }
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("patch_id")
+    parser.add_argument(
+        "--output-directory",
+        type=Path,
+        default=REPOSITORY_ROOT / ".runtime/worker-artifacts",
+    )
+    args = parser.parse_args()
+    try:
+        result = build_artifact(args.patch_id, args.output_directory)
+    except (ArtifactBuildError, OSError, subprocess.SubprocessError, json.JSONDecodeError) as exc:
+        parser.error(str(exc))
+    print(json.dumps(result, ensure_ascii=False, indent=2, sort_keys=True))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/src/k1link/perception/graph.py b/src/k1link/perception/graph.py
index e09c6c8..7cc1c24 100644
--- a/src/k1link/perception/graph.py
+++ b/src/k1link/perception/graph.py
@@ -29,19 +29,25 @@ from .contracts import (
 )
 from .graph_contracts import (
     GRAPH_RESULT_SCHEMA,
+    GRAPH_RESULT_SCHEMA_V2,
     REFERENCE_GRAPH_ID,
+    REFERENCE_GRAPH_ID_V2,
     TERMINAL_OUTCOME_SCHEMA,
     DeliveredFrame,
     GraphExecutionError,
+    GraphRunMode,
     GraphRunResult,
+    GraphRunResultV2,
     GraphState,
     TerminalOutcome,
     TerminalOutcomeType,
     build_graph_run_result,
+    build_graph_run_result_v2,
 )
 from .graph_validation import (
     validate_observations,
     validate_proposals,
+    validate_rolling,
     validate_temporal,
     validate_threats,
 )
@@ -51,6 +57,8 @@ from .providers import (
     MotionProvider,
     ProviderRole,
     ReferencePerceptionGraphConfig,
+    ReferencePerceptionGraphConfigV2,
+    RollingMapProvider,
     SourcePacket,
     SourceProvider,
     TemporalStateProvider,
@@ -79,6 +87,15 @@ class _Temporal:
     obstacles: tuple[TemporalObstacle, ...]
 
 
+@dataclass(frozen=True, slots=True)
+class _Rolled:
+    packet: SourcePacket
+    proposals: tuple[ObjectProposal2D, ...]
+    associated_proposal_ids: frozenset[str]
+    obstacles: tuple[TemporalObstacle, ...]
+    retained: tuple[TemporalObstacle, ...]
+
+
 @dataclass(frozen=True, slots=True)
 class _StopSignal:
     pass
@@ -88,8 +105,9 @@ _STOP = _StopSignal()
 _DetectorItem = SourcePacket | _StopSignal
 _GeometryItem = _Detected | _StopSignal
 _TemporalItem = _Associated | _StopSignal
-_ThreatItem = _Temporal | _StopSignal
-_QueueItem = SourcePacket | _Detected | _Associated | _Temporal | _StopSignal
+_RollingItem = _Temporal | _StopSignal
+_ThreatItem = _Temporal | _Rolled | _StopSignal
+_QueueItem = SourcePacket | _Detected | _Associated | _Temporal | _Rolled | _StopSignal
 _QueueItemT = TypeVar("_QueueItemT", bound=_QueueItem)
 
 
@@ -114,6 +132,37 @@ class ReferencePerceptionGraphV1:
             raise GraphExecutionError("reference graph id is incompatible")
         if config.source_profile_id != BASELINE_PROFILE_ID:
             raise GraphExecutionError("reference source profile is incompatible")
+        self._initialize(
+            config=config,
+            source=source,
+            detector=detector,
+            geometry=geometry,
+            temporal=temporal,
+            motion=motion,
+            rolling=None,
+            threat=threat,
+            run_mode=GraphRunMode.SOURCE_PACED_LATEST_WINS,
+            telemetry_identity=telemetry_identity,
+            telemetry_sink=telemetry_sink,
+            clock_ns=clock_ns,
+        )
+
+    def _initialize(
+        self,
+        *,
+        config: ReferencePerceptionGraphConfig | ReferencePerceptionGraphConfigV2,
+        source: SourceProvider,
+        detector: DetectorProvider,
+        geometry: GeometryAssociationProvider,
+        temporal: TemporalStateProvider,
+        motion: MotionProvider,
+        rolling: RollingMapProvider | None,
+        threat: ThreatProvider,
+        run_mode: GraphRunMode,
+        telemetry_identity: PipelineTelemetryIdentity | None,
+        telemetry_sink: PipelineTelemetrySink | None,
+        clock_ns: Callable[[], int],
+    ) -> None:
         if (telemetry_identity is None) is not (telemetry_sink is None):
             raise GraphExecutionError("telemetry identity and sink must be configured together")
         self.config = config
@@ -122,7 +171,9 @@ class ReferencePerceptionGraphV1:
         self.geometry = geometry
         self.temporal = temporal
         self.motion = motion
+        self.rolling = rolling
         self.threat = threat
+        self.run_mode = run_mode
         self.telemetry_identity = telemetry_identity
         self.telemetry_sink = telemetry_sink
         self._clock_ns = clock_ns
@@ -135,6 +186,9 @@ class ReferencePerceptionGraphV1:
         self._deliveries: list[DeliveredFrame] = []
         self._admitted: dict[int, SourcePacket] = {}
         self._admitted_at_ns: dict[int, int] = {}
+        self._queue_high_watermarks: dict[str, int] = {
+            stage_id: 0 for stage_id in ("detector", "geometry", "temporal", "rolling", "threat")
+        }
         self._validate_provider_pins()
 
     @property
@@ -149,14 +203,17 @@ class ReferencePerceptionGraphV1:
     def cancel(self) -> None:
         self._cancel_event.set()
 
-    def run(self) -> GraphRunResult:
+    def run(self) -> GraphRunResult | GraphRunResultV2:
         self._begin_run()
         started_ns = self._now()
         detector_queue: Queue[_DetectorItem] = Queue(self._capacity("detector"))
         geometry_queue: Queue[_GeometryItem] = Queue(self._capacity("geometry"))
         temporal_queue: Queue[_TemporalItem] = Queue(self._capacity("temporal"))
         threat_queue: Queue[_ThreatItem] = Queue(self._capacity("threat"))
-        self._threads = [
+        rolling_queue: Queue[_RollingItem] | None = (
+            Queue(self._capacity("rolling")) if self.rolling is not None else None
+        )
+        threads = [
             Thread(
                 target=self._detector_loop,
                 args=(detector_queue, geometry_queue),
@@ -171,17 +228,29 @@ class ReferencePerceptionGraphV1:
             ),
             Thread(
                 target=self._temporal_loop,
-                args=(temporal_queue, threat_queue),
+                args=(temporal_queue, rolling_queue or threat_queue),
                 name="m4-temporal",
                 daemon=True,
             ),
+        ]
+        if rolling_queue is not None:
+            threads.append(
+                Thread(
+                    target=self._rolling_loop,
+                    args=(rolling_queue, threat_queue),
+                    name="m4-rolling",
+                    daemon=True,
+                )
+            )
+        threads.append(
             Thread(
                 target=self._threat_loop,
                 args=(threat_queue,),
                 name="m4-threat",
                 daemon=True,
-            ),
-        ]
+            )
+        )
+        self._threads = threads
         for thread in self._threads:
             thread.start()
         run_emitter = self._run_emitter()
@@ -244,6 +313,8 @@ class ReferencePerceptionGraphV1:
             self._deliveries.clear()
             self._admitted.clear()
             self._admitted_at_ns.clear()
+            for stage_id in self._queue_high_watermarks:
+                self._queue_high_watermarks[stage_id] = 0
 
     def _admit(self, packet: SourcePacket) -> bool:
         envelope = packet.envelope
@@ -283,6 +354,8 @@ class ReferencePerceptionGraphV1:
                     validate_proposals(item, proposals)
                     if stage is not None:
                         stage.output_count = len(proposals)
+                if self._expired(item, "detector"):
+                    continue
                 self._put_latest(outgoing, _Detected(item, proposals), "geometry")
             except Exception as exc:
                 if not isinstance(item, _StopSignal):
@@ -311,6 +384,8 @@ class ReferencePerceptionGraphV1:
                     validate_exclusive_point_ownership(observations)
                     if stage is not None:
                         stage.output_count = len(observations)
+                if self._expired(item.packet, "geometry"):
+                    continue
                 self._put_latest(
                     outgoing,
                     _Associated(item.packet, item.proposals, observations),
@@ -325,13 +400,16 @@ class ReferencePerceptionGraphV1:
     def _temporal_loop(
         self,
         incoming: Queue[_TemporalItem],
-        outgoing: Queue[_ThreatItem],
+        outgoing: Queue[_RollingItem] | Queue[_ThreatItem],
     ) -> None:
         while True:
             item = incoming.get()
             try:
                 if isinstance(item, _StopSignal):
-                    self._put_stop(outgoing, "threat")
+                    self._put_stop(
+                        cast(Queue[_QueueItem], outgoing),
+                        "rolling" if self.rolling is not None else "threat",
+                    )
                     return
                 if self._cancelled(item.packet):
                     continue
@@ -343,8 +421,10 @@ class ReferencePerceptionGraphV1:
                     validate_temporal(item.packet, obstacles)
                     if stage is not None:
                         stage.output_count = len(obstacles)
+                if self._expired(item.packet, "temporal"):
+                    continue
                 self._put_latest(
-                    outgoing,
+                    cast(Queue[_QueueItem], outgoing),
                     _Temporal(
                         item.packet,
                         item.proposals,
@@ -356,7 +436,7 @@ class ReferencePerceptionGraphV1:
                         ),
                         obstacles,
                     ),
-                    "threat",
+                    "rolling" if self.rolling is not None else "threat",
                 )
             except Exception as exc:
                 if not isinstance(item, _StopSignal):
@@ -364,6 +444,48 @@ class ReferencePerceptionGraphV1:
             finally:
                 incoming.task_done()
 
+    def _rolling_loop(
+        self,
+        incoming: Queue[_RollingItem],
+        outgoing: Queue[_ThreatItem],
+    ) -> None:
+        rolling = self.rolling
+        if rolling is None:
+            raise GraphExecutionError("rolling loop requires a rolling provider")
+        while True:
+            item = incoming.get()
+            try:
+                if isinstance(item, _StopSignal):
+                    self._put_stop(outgoing, "threat")
+                    return
+                if self._cancelled(item.packet):
+                    continue
+                if self._expired(item.packet, "rolling"):
+                    continue
+                with self._stage(item.packet, "rolling", len(item.obstacles)) as stage:
+                    retained = rolling.update(item.packet, item.obstacles)
+                    validate_rolling(item.obstacles, retained)
+                    if stage is not None:
+                        stage.output_count = len(retained)
+                if self._expired(item.packet, "rolling"):
+                    continue
+                self._put_latest(
+                    outgoing,
+                    _Rolled(
+                        item.packet,
+                        item.proposals,
+                        item.associated_proposal_ids,
+                        item.obstacles,
+                        retained,
+                    ),
+                    "threat",
+                )
+            except Exception as exc:
+                if not isinstance(item, _StopSignal):
+                    self._failed(item.packet, "rolling", exc)
+            finally:
+                incoming.task_done()
+
     def _threat_loop(self, incoming: Queue[_ThreatItem]) -> None:
         while True:
             item = incoming.get()
@@ -380,6 +502,8 @@ class ReferencePerceptionGraphV1:
                     validate_threats(obstacle_map, threats)
                     if stage is not None:
                         stage.output_count = len(threats)
+                if self._expired(item.packet, "threat"):
+                    continue
                 delivery = DeliveredFrame(
                     sequence=item.packet.envelope.sequence,
                     obstacle_map=obstacle_map,
@@ -399,14 +523,15 @@ class ReferencePerceptionGraphV1:
             finally:
                 incoming.task_done()
 
-    def _obstacle_map(self, item: _Temporal) -> LocalObstacleMap:
+    def _obstacle_map(self, item: _Temporal | _Rolled) -> LocalObstacleMap:
         packet = item.packet
         now_ns = self._now()
         with self._result_lock:
             admitted_at_ns = self._admitted_at_ns[packet.envelope.sequence]
-        occupied = tuple(
+        current = tuple(
             obstacle for obstacle in item.obstacles if obstacle.state is TemporalState.CURRENT
         )
+        occupied = (*current, *item.retained) if isinstance(item, _Rolled) else current
         unknown = tuple(
             obstacle for obstacle in item.obstacles if obstacle.state is not TemporalState.CURRENT
         )
@@ -484,9 +609,13 @@ class ReferencePerceptionGraphV1:
         item: _QueueItemT,
         stage_id: str,
     ) -> None:
+        if self.run_mode is GraphRunMode.LOSSLESS_REPLAY:
+            self._put_lossless(queue, item, stage_id)
+            return
         while True:
             try:
                 queue.put_nowait(item)
+                self._record_queue_depth(stage_id, queue.qsize())
                 return
             except Full:
                 pass
@@ -506,11 +635,60 @@ class ReferencePerceptionGraphV1:
             finally:
                 queue.task_done()
 
+    def _put_lossless(
+        self,
+        queue: Queue[_QueueItemT],
+        item: _QueueItemT,
+        stage_id: str,
+    ) -> None:
+        while True:
+            try:
+                queue.put(item, timeout=0.05)
+                self._record_queue_depth(stage_id, queue.qsize())
+                return
+            except Full:
+                if self._cancel_event.is_set() and not isinstance(item, _StopSignal):
+                    packet = _packet_from_item(item)
+                    self._terminal(
+                        packet,
+                        TerminalOutcomeType.FAILED,
+                        stage_id,
+                        "graph-cancelled",
+                    )
+                    return
+
+    def _record_queue_depth(self, stage_id: str, depth: int) -> None:
+        with self._result_lock:
+            self._queue_high_watermarks[stage_id] = max(
+                self._queue_high_watermarks.get(stage_id, 0),
+                depth,
+            )
+
     def _put_stop(self, queue: Queue[_QueueItemT], stage_id: str) -> None:
         timeout = self._terminal_timeout(stage_id) / 1_000_000_000
         try:
             queue.put(cast(_QueueItemT, _STOP), timeout=timeout)
+            self._record_queue_depth(stage_id, queue.qsize())
         except Full:
+            if self.run_mode is GraphRunMode.LOSSLESS_REPLAY:
+                self._cancel_event.set()
+                try:
+                    stranded = queue.get_nowait()
+                except Empty as exc:
+                    raise GraphExecutionError("lossless terminal queue timed out") from exc
+                try:
+                    if not isinstance(stranded, _StopSignal):
+                        self._terminal(
+                            _packet_from_item(stranded),
+                            TerminalOutcomeType.FAILED,
+                            stage_id,
+                            "terminal-queue-timeout",
+                        )
+                finally:
+                    queue.task_done()
+                queue.put_nowait(cast(_QueueItemT, _STOP))
+                self._record_queue_depth(stage_id, queue.qsize())
+                return
             self._put_latest(queue, cast(_QueueItemT, _STOP), stage_id)
 
     def _join_workers(self) -> None:
@@ -541,11 +719,27 @@ class ReferencePerceptionGraphV1:
             if set(self._admitted) != set(self._outcomes):
                 raise GraphExecutionError("terminal accounting did not close")
 
-    def _result(self, state: GraphState) -> GraphRunResult:
+    def _result(self, state: GraphState) -> GraphRunResult | GraphRunResultV2:
         with self._result_lock:
             outcomes = tuple(self._outcomes[key] for key in sorted(self._outcomes))
             deliveries = tuple(sorted(self._deliveries, key=lambda item: item.sequence))
             admitted_count = len(self._admitted)
+            queue_high_watermarks = tuple(
+                (stage_id, self._queue_high_watermarks[stage_id])
+                for stage_id in sorted(self._queue_high_watermarks)
+                if any(policy.stage_id == stage_id for policy in self.config.queues)
+            )
+        if self.rolling is not None:
+            return build_graph_run_result_v2(
+                graph_id=self.config.graph_id,
+                source_profile_id=self.config.source_profile_id,
+                run_mode=self.run_mode,
+                state=state,
+                admitted_count=admitted_count,
+                outcomes=outcomes,
+                deliveries=deliveries,
+                queue_high_watermarks=queue_high_watermarks,
+            )
         return build_graph_run_result(
             graph_id=self.config.graph_id,
             source_profile_id=self.config.source_profile_id,
@@ -598,6 +792,8 @@ class ReferencePerceptionGraphV1:
             ProviderRole.MOTION: self.motion.provider_id,
             ProviderRole.THREAT: self.threat.provider_id,
         }
+        if self.rolling is not None:
+            actual[ProviderRole.ROLLING] = self.rolling.provider_id
         pinned = {pin.role: pin.provider_id for pin in self.config.providers}
         if actual != pinned:
             raise GraphExecutionError("configured provider identities do not match runtime")
@@ -627,6 +823,56 @@ class ReferencePerceptionGraphV1:
             self._state = state
 
 
+class ReferencePerceptionGraphV2(ReferencePerceptionGraphV1):
+    """Final M4 graph with explicit bounded rolling occupancy ownership."""
+
+    def __init__(
+        self,
+        *,
+        config: ReferencePerceptionGraphConfigV2,
+        source: SourceProvider,
+        detector: DetectorProvider,
+        geometry: GeometryAssociationProvider,
+        temporal: TemporalStateProvider,
+        motion: MotionProvider,
+        rolling: RollingMapProvider,
+        threat: ThreatProvider,
+        run_mode: GraphRunMode,
+        telemetry_identity: PipelineTelemetryIdentity | None = None,
+        telemetry_sink: PipelineTelemetrySink | None = None,
+        clock_ns: Callable[[], int] = time.monotonic_ns,
+    ) -> None:
+        if config.graph_id != REFERENCE_GRAPH_ID_V2:
+            raise GraphExecutionError("reference graph v2 id is incompatible")
+        if config.source_profile_id != BASELINE_PROFILE_ID:
+            raise GraphExecutionError("reference graph v2 source profile is incompatible")
+        self._has_run = False
+        self._run_once_lock = Lock()
+        self._initialize(
+            config=config,
+            source=source,
+            detector=detector,
+            geometry=geometry,
+            temporal=temporal,
+            motion=motion,
+            rolling=rolling,
+            threat=threat,
+            run_mode=run_mode,
+            telemetry_identity=telemetry_identity,
+            telemetry_sink=telemetry_sink,
+            clock_ns=clock_ns,
+        )
+
+    def _begin_run(self) -> None:
+        with self._run_once_lock:
+            if self._has_run:
+                raise GraphExecutionError(
+                    "reference graph v2 restart requires freshly instantiated providers"
+                )
+            self._has_run = True
+        super()._begin_run()
+
+
 class _NullStage:
     output_count: int | None = None
 
@@ -640,20 +886,25 @@ class _NullStage:
 def _packet_from_item(item: object) -> SourcePacket:
     if isinstance(item, SourcePacket):
         return item
-    if isinstance(item, (_Detected, _Associated, _Temporal)):
+    if isinstance(item, (_Detected, _Associated, _Temporal, _Rolled)):
         return item.packet
     raise GraphExecutionError("queue contained an incompatible item")
 
 
 __all__ = [
     "GRAPH_RESULT_SCHEMA",
+    "GRAPH_RESULT_SCHEMA_V2",
     "REFERENCE_GRAPH_ID",
+    "REFERENCE_GRAPH_ID_V2",
     "TERMINAL_OUTCOME_SCHEMA",
     "DeliveredFrame",
     "GraphExecutionError",
     "GraphRunResult",
+    "GraphRunResultV2",
+    "GraphRunMode",
     "GraphState",
     "ReferencePerceptionGraphV1",
+    "ReferencePerceptionGraphV2",
     "TerminalOutcome",
     "TerminalOutcomeType",
 ]
diff --git a/src/k1link/perception/graph_contracts.py b/src/k1link/perception/graph_contracts.py
index cd52eb5..7f925ea 100644
--- a/src/k1link/perception/graph_contracts.py
+++ b/src/k1link/perception/graph_contracts.py
@@ -12,8 +12,10 @@ from typing import Final
 from .contracts import LocalObstacleMap, ThreatAssessment
 
 GRAPH_RESULT_SCHEMA: Final = "missioncore.reference-perception-graph-result/v1"
+GRAPH_RESULT_SCHEMA_V2: Final = "missioncore.reference-perception-graph-result/v2"
 TERMINAL_OUTCOME_SCHEMA: Final = "missioncore.perception-terminal-outcome/v1"
 REFERENCE_GRAPH_ID: Final = "reference-perception-graph/v1"
+REFERENCE_GRAPH_ID_V2: Final = "reference-perception-graph/v2"
 _SAFE_IDENTIFIER: Final = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:/-]{0,159}$")
 
 
@@ -31,6 +33,13 @@ class GraphState(StrEnum):
     FAILED = "failed"
 
 
+class GraphRunMode(StrEnum):
+    """Bounded queue behavior for realtime and deterministic replay."""
+
+    SOURCE_PACED_LATEST_WINS = "source-paced-latest-wins"
+    LOSSLESS_REPLAY = "lossless-replay"
+
+
 class TerminalOutcomeType(StrEnum):
     DELIVERED = "delivered"
     SUPERSEDED = "superseded"
@@ -110,6 +119,33 @@ class GraphRunResult:
         }
 
 
+@dataclass(frozen=True, slots=True)
+class GraphRunResultV2:
+    graph_id: str
+    source_profile_id: str
+    run_mode: GraphRunMode
+    state: GraphState
+    admitted_count: int
+    terminal_outcomes: tuple[TerminalOutcome, ...]
+    deliveries: tuple[DeliveredFrame, ...]
+    queue_high_watermarks: tuple[tuple[str, int], ...]
+    canonical_payload_sha256: str
+
+    def to_dict(self) -> dict[str, object]:
+        return {
+            "schema_version": GRAPH_RESULT_SCHEMA_V2,
+            "graph_id": self.graph_id,
+            "source_profile_id": self.source_profile_id,
+            "run_mode": self.run_mode.value,
+            "state": self.state.value,
+            "admitted_count": self.admitted_count,
+            "terminal_outcomes": [item.to_dict() for item in self.terminal_outcomes],
+            "deliveries": [item.canonical_dict() for item in self.deliveries],
+            "queue_high_watermarks": dict(self.queue_high_watermarks),
+            "canonical_payload_sha256": self.canonical_payload_sha256,
+        }
+
+
 def build_graph_run_result(
     *,
     graph_id: str,
@@ -137,15 +173,52 @@ def build_graph_run_result(
     )
 
 
+def build_graph_run_result_v2(
+    *,
+    graph_id: str,
+    source_profile_id: str,
+    run_mode: GraphRunMode,
+    state: GraphState,
+    admitted_count: int,
+    outcomes: tuple[TerminalOutcome, ...],
+    deliveries: tuple[DeliveredFrame, ...],
+    queue_high_watermarks: tuple[tuple[str, int], ...],
+) -> GraphRunResultV2:
+    canonical = {
+        "graph_id": graph_id,
+        "source_profile_id": source_profile_id,
+        "run_mode": run_mode.value,
+        "terminal_outcomes": [item.to_dict() for item in outcomes],
+        "deliveries": [item.canonical_dict() for item in deliveries],
+    }
+    payload = json.dumps(canonical, sort_keys=True, separators=(",", ":")).encode()
+    return GraphRunResultV2(
+        graph_id=graph_id,
+        source_profile_id=source_profile_id,
+        run_mode=run_mode,
+        state=state,
+        admitted_count=admitted_count,
+        terminal_outcomes=outcomes,
+        deliveries=deliveries,
+        queue_high_watermarks=queue_high_watermarks,
+        canonical_payload_sha256=hashlib.sha256(payload).hexdigest(),
+    )
+
+
 __all__ = [
     "GRAPH_RESULT_SCHEMA",
+    "GRAPH_RESULT_SCHEMA_V2",
     "REFERENCE_GRAPH_ID",
+    "REFERENCE_GRAPH_ID_V2",
     "TERMINAL_OUTCOME_SCHEMA",
     "DeliveredFrame",
     "GraphExecutionError",
     "GraphRunResult",
+    "GraphRunResultV2",
+    "GraphRunMode",
     "GraphState",
     "TerminalOutcome",
     "TerminalOutcomeType",
     "build_graph_run_result",
+    "build_graph_run_result_v2",
 ]
diff --git a/src/k1link/perception/graph_validation.py b/src/k1link/perception/graph_validation.py
index 34ce01c..af06414 100644
--- a/src/k1link/perception/graph_validation.py
+++ b/src/k1link/perception/graph_validation.py
@@ -69,6 +69,20 @@ def validate_temporal(
         raise GraphExecutionError("current temporal occupancy was invented without LiDAR")
 
 
+def validate_rolling(
+    temporal: tuple[TemporalObstacle, ...],
+    retained: tuple[TemporalObstacle, ...],
+) -> None:
+    retained_ids = [obstacle.component_id for obstacle in retained]
+    temporal_ids = {obstacle.component_id for obstacle in temporal}
+    if len(set(retained_ids)) != len(retained_ids):
+        raise GraphExecutionError("rolling component identities are duplicated")
+    if temporal_ids.intersection(retained_ids):
+        raise GraphExecutionError("rolling and temporal component identities overlap")
+    if any(obstacle.state is not TemporalState.RETAINED for obstacle in retained):
+        raise GraphExecutionError("rolling provider published non-retained evidence")
+
+
 def validate_threats(
     obstacle_map: LocalObstacleMap,
     threats: tuple[ThreatAssessment, ...],
@@ -92,6 +106,7 @@ def validate_threats(
 __all__ = [
     "validate_observations",
     "validate_proposals",
+    "validate_rolling",
     "validate_temporal",
     "validate_threats",
 ]
diff --git a/src/k1link/perception/providers.py b/src/k1link/perception/providers.py
index 2024e77..9396385 100644
--- a/src/k1link/perception/providers.py
+++ b/src/k1link/perception/providers.py
@@ -19,7 +19,11 @@ from .contracts import (
 )
 
 REFERENCE_GRAPH_CONFIG_SCHEMA: Final = "missioncore.reference-perception-graph-config/v1"
+REFERENCE_GRAPH_CONFIG_SCHEMA_V2: Final = "missioncore.reference-perception-graph-config/v2"
 REFERENCE_GRAPH_STAGE_IDS: Final = frozenset({"detector", "geometry", "temporal", "threat"})
+REFERENCE_GRAPH_STAGE_IDS_V2: Final = frozenset(
+    {"detector", "geometry", "temporal", "rolling", "threat"}
+)
 _SHA256 = re.compile(r"^[a-f0-9]{64}$")
 _IDENTIFIER = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:/-]{0,159}$")
 
@@ -34,9 +38,23 @@ class ProviderRole(StrEnum):
     GEOMETRY = "geometry"
     TEMPORAL = "temporal"
     MOTION = "motion"
+    ROLLING = "rolling"
     THREAT = "threat"
 
 
+REFERENCE_GRAPH_PROVIDER_ROLES: Final = frozenset(
+    {
+        ProviderRole.SOURCE,
+        ProviderRole.DETECTOR,
+        ProviderRole.GEOMETRY,
+        ProviderRole.TEMPORAL,
+        ProviderRole.MOTION,
+        ProviderRole.THREAT,
+    }
+)
+REFERENCE_GRAPH_PROVIDER_ROLES_V2: Final = frozenset(ProviderRole)
+
+
 @dataclass(frozen=True, slots=True)
 class SourcePacket:
     """Execution-only carrier; the graph never interprets opaque sensor payloads."""
@@ -211,7 +229,7 @@ class ReferencePerceptionGraphConfig:
         _identifier(self.graph_id, "graph id")
         _identifier(self.source_profile_id, "source profile id")
         roles = [provider.role for provider in self.providers]
-        if len(set(roles)) != len(roles) or set(roles) != set(ProviderRole):
+        if len(set(roles)) != len(roles) or set(roles) != REFERENCE_GRAPH_PROVIDER_ROLES:
             raise ProviderContractError("graph must pin each provider role exactly once")
         stage_ids = [queue.stage_id for queue in self.queues]
         if len(set(stage_ids)) != len(stage_ids):
@@ -248,6 +266,57 @@ class ReferencePerceptionGraphConfig:
         )
 
 
+@dataclass(frozen=True, slots=True)
+class ReferencePerceptionGraphConfigV2:
+    """Final M4 graph contract with explicit retained rolling occupancy."""
+
+    graph_id: str
+    source_profile_id: str
+    providers: tuple[ProviderPin, ...]
+    queues: tuple[QueuePolicy, ...]
+    authority: GraphAuthority = GraphAuthority()
+
+    def __post_init__(self) -> None:
+        _identifier(self.graph_id, "graph id")
+        _identifier(self.source_profile_id, "source profile id")
+        roles = [provider.role for provider in self.providers]
+        if len(set(roles)) != len(roles) or set(roles) != REFERENCE_GRAPH_PROVIDER_ROLES_V2:
+            raise ProviderContractError("graph v2 must pin each provider role exactly once")
+        stage_ids = [queue.stage_id for queue in self.queues]
+        if len(set(stage_ids)) != len(stage_ids):
+            raise ProviderContractError("graph v2 queue policies must be unique")
+        if set(stage_ids) != REFERENCE_GRAPH_STAGE_IDS_V2:
+            raise ProviderContractError("graph v2 must bound each reference stage exactly once")
+
+    def to_dict(self) -> dict[str, object]:
+        return {
+            "schema_version": REFERENCE_GRAPH_CONFIG_SCHEMA_V2,
+            "graph_id": self.graph_id,
+            "source_profile_id": self.source_profile_id,
+            "providers": [provider.to_dict() for provider in self.providers],
+            "queues": [queue.to_dict() for queue in self.queues],
+            "authority": self.authority.to_dict(),
+        }
+
+    @classmethod
+    def from_dict(cls, value: object) -> ReferencePerceptionGraphConfigV2:
+        document = _object(value, "reference graph v2 config")
+        _exact_keys(
+            document,
+            {"schema_version", "graph_id", "source_profile_id", "providers", "queues", "authority"},
+            "reference graph v2 config",
+        )
+        if document.get("schema_version") != REFERENCE_GRAPH_CONFIG_SCHEMA_V2:
+            raise ProviderContractError("reference graph v2 config schema is incompatible")
+        return cls(
+            graph_id=_string(document, "graph_id"),
+            source_profile_id=_string(document, "source_profile_id"),
+            providers=tuple(ProviderPin.from_dict(item) for item in _array(document, "providers")),
+            queues=tuple(QueuePolicy.from_dict(item) for item in _array(document, "queues")),
+            authority=GraphAuthority.from_dict(document.get("authority")),
+        )
+
+
 class SourceProvider(Protocol):
     provider_id: str
 
@@ -290,6 +359,16 @@ class MotionProvider(Protocol):
     ) -> tuple[TemporalObstacle, ...]: ...
 
 
+class RollingMapProvider(Protocol):
+    provider_id: str
+
+    def update(
+        self,
+        packet: SourcePacket,
+        obstacles: tuple[TemporalObstacle, ...],
+    ) -> tuple[TemporalObstacle, ...]: ...
+
+
 class ThreatProvider(Protocol):
     provider_id: str
 
diff --git a/src/k1link/perception/reference_graph_cli.py b/src/k1link/perception/reference_graph_cli.py
new file mode 100644
index 0000000..2a60057
--- /dev/null
+++ b/src/k1link/perception/reference_graph_cli.py
@@ -0,0 +1,90 @@
+"""Run and seal the M4.7 canonical graph on explicit Worker 006 inputs."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import time
+from pathlib import Path
+
+from .graph_contracts import GraphRunMode, GraphRunResultV2
+from .reference_graph_parity import compare_reference_graph_to_accepted_ledgers
+from .reference_graph_result import seal_reference_graph_result
+from .reference_graph_runtime import ReferenceGraphRuntimePaths, build_reference_graph_runtime
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--graph-config", type=Path, required=True)
+    parser.add_argument("--baseline-profile", type=Path, required=True)
+    parser.add_argument("--geometry-profile", type=Path, required=True)
+    parser.add_argument("--temporal-motion-profile", type=Path, required=True)
+    parser.add_argument("--rolling-map-profile", type=Path, required=True)
+    parser.add_argument("--threat-profile", type=Path, required=True)
+    parser.add_argument("--camera-index", type=Path, required=True)
+    parser.add_argument("--source-pack", type=Path, required=True)
+    parser.add_argument("--local-surface", type=Path, required=True)
+    parser.add_argument("--video", type=Path, required=True)
+    parser.add_argument("--valid-fov-mask", type=Path, required=True)
+    parser.add_argument("--temporal-parity-frames", type=Path, required=True)
+    parser.add_argument("--threat-parity-frames", type=Path, required=True)
+    parser.add_argument("--triton-origin", default="http://127.0.0.1:8000")
+    parser.add_argument(
+        "--mode",
+        choices=tuple(item.value for item in GraphRunMode),
+        default=GraphRunMode.LOSSLESS_REPLAY.value,
+    )
+    parser.add_argument("--expected-frames", type=int, default=4489)
+    parser.add_argument("--output-root", type=Path, required=True)
+    args = parser.parse_args(argv)
+    paths = ReferenceGraphRuntimePaths(
+        graph_config=args.graph_config,
+        baseline_profile=args.baseline_profile,
+        geometry_profile=args.geometry_profile,
+        temporal_motion_profile=args.temporal_motion_profile,
+        rolling_map_profile=args.rolling_map_profile,
+        threat_profile=args.threat_profile,
+        camera_index=args.camera_index,
+        source_pack=args.source_pack,
+        local_surface=args.local_surface,
+        video=args.video,
+        valid_fov_mask=args.valid_fov_mask,
+    )
+    started_ns = time.perf_counter_ns()
+    with build_reference_graph_runtime(
+        paths=paths,
+        triton_origin=args.triton_origin,
+        run_mode=GraphRunMode(args.mode),
+    ) as runtime:
+        graph_result = runtime.graph.run()
+    if not isinstance(graph_result, GraphRunResultV2):
+        raise RuntimeError("M4.7 runtime returned a legacy graph result")
+    parity = compare_reference_graph_to_accepted_ledgers(
+        graph_result,
+        temporal_frames_path=args.temporal_parity_frames,
+        threat_frames_path=args.threat_parity_frames,
+        expected_frames=args.expected_frames,
+    )
+    sealed = seal_reference_graph_result(
+        graph_result,
+        output_root=args.output_root,
+        expected_frames=args.expected_frames,
+        parity=parity,
+    )
+    print(
+        json.dumps(
+            {
+                "accepted": sealed.accepted,
+                "elapsed_seconds": (time.perf_counter_ns() - started_ns) / 1_000_000_000,
+                "result_id": sealed.result_id,
+                "result_root": str(sealed.result_root),
+            },
+            sort_keys=True,
+            separators=(",", ":"),
+        )
+    )
+    return 0 if sealed.accepted else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/src/k1link/perception/reference_graph_parity.py b/src/k1link/perception/reference_graph_parity.py
new file mode 100644
index 0000000..78f39d1
--- /dev/null
+++ b/src/k1link/perception/reference_graph_parity.py
@@ -0,0 +1,212 @@
+"""Frame-exact parity against the accepted M4.5R and M4.6 ledgers."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Final
+
+from .contracts import TemporalState
+from .graph_contracts import GraphRunResultV2
+
+REFERENCE_GRAPH_PARITY_SCHEMA: Final = "missioncore.reference-perception-graph-parity/v1"
+
+
+class ReferenceGraphParityError(RuntimeError):
+    """Accepted parity evidence is missing, malformed or source-inconsistent."""
+
+
+@dataclass(frozen=True, slots=True)
+class ReferenceGraphParityReport:
+    expected_frames: int
+    compared_frames: int
+    temporal_frames_sha256: str
+    threat_frames_sha256: str
+    mismatch_counts: tuple[tuple[str, int], ...]
+    accepted: bool
+
+    def __post_init__(self) -> None:
+        if self.expected_frames < 1 or not 0 <= self.compared_frames <= self.expected_frames:
+            raise ReferenceGraphParityError("parity frame accounting is invalid")
+        for digest in (self.temporal_frames_sha256, self.threat_frames_sha256):
+            if len(digest) != 64 or any(
+                character not in "0123456789abcdef" for character in digest
+            ):
+                raise ReferenceGraphParityError("parity ledger digest is invalid")
+        names = [name for name, _ in self.mismatch_counts]
+        if (
+            names != sorted(names)
+            or len(set(names)) != len(names)
+            or any(not name or count < 0 for name, count in self.mismatch_counts)
+        ):
+            raise ReferenceGraphParityError("parity mismatch accounting is invalid")
+        exact = self.compared_frames == self.expected_frames and all(
+            count == 0 for _, count in self.mismatch_counts
+        )
+        if self.accepted is not exact:
+            raise ReferenceGraphParityError("parity acceptance disagrees with accounting")
+
+    def to_dict(self) -> dict[str, object]:
+        return {
+            "schema_version": REFERENCE_GRAPH_PARITY_SCHEMA,
+            "expected_frames": self.expected_frames,
+            "compared_frames": self.compared_frames,
+            "temporal_frames_sha256": self.temporal_frames_sha256,
+            "threat_frames_sha256": self.threat_frames_sha256,
+            "mismatch_counts": dict(self.mismatch_counts),
+            "accepted": self.accepted,
+        }
+
+
+def compare_reference_graph_to_accepted_ledgers(
+    result: GraphRunResultV2,
+    *,
+    temporal_frames_path: Path,
+    threat_frames_path: Path,
+    expected_frames: int,
+) -> ReferenceGraphParityReport:
+    if expected_frames < 1:
+        raise ReferenceGraphParityError("expected parity frame count must be positive")
+    if len(result.deliveries) != expected_frames:
+        raise ReferenceGraphParityError("graph delivery count cannot support full parity")
+    temporal_path = _regular_file(temporal_frames_path, "temporal parity ledger")
+    threat_path = _regular_file(threat_frames_path, "threat parity ledger")
+    mismatch = {
+        "source_binding": 0,
+        "current": 0,
+        "rolling_retained": 0,
+        "held": 0,
+        "expired": 0,
+        "camera_uncertainty": 0,
+        "threat_assessments": 0,
+    }
+    compared = 0
+    with temporal_path.open("rb") as temporal_stream, threat_path.open("rb") as threat_stream:
+        for expected_sequence, delivery in enumerate(result.deliveries):
+            temporal = _read_row(temporal_stream.readline(), "temporal", expected_sequence)
+            threat = _read_row(threat_stream.readline(), "threat", expected_sequence)
+            obstacle_map = delivery.obstacle_map
+            if (
+                delivery.sequence != expected_sequence
+                or temporal.get("sequence") != expected_sequence
+                or threat.get("sequence") != expected_sequence
+                or temporal.get("frame_id") != obstacle_map.frame_id
+                or threat.get("frame_id") != obstacle_map.frame_id
+            ):
+                mismatch["source_binding"] += 1
+            occupied_current = [
+                item.to_dict()
+                for item in obstacle_map.occupied
+                if item.state is TemporalState.CURRENT
+            ]
+            rolling_retained = [
+                item.to_dict()
+                for item in obstacle_map.occupied
+                if item.state is TemporalState.RETAINED
+            ]
+            held = [
+                item.to_dict()
+                for item in obstacle_map.unknown
+                if item.state is TemporalState.HELD
+            ]
+            expired = [
+                item.to_dict()
+                for item in obstacle_map.unknown
+                if item.state is TemporalState.EXPIRED
+            ]
+            for label, observed in (
+                ("current", occupied_current),
+                ("rolling_retained", rolling_retained),
+                ("held", held),
+                ("expired", expired),
+            ):
+                if observed != _array(temporal, label):
+                    mismatch[label] += 1
+            expected_camera_uncertainty = [
+                _string(row, "proposal_id")
+                for row in _array(threat, "camera_proposals")
+                if _boolean(row, "occupied_support") is False
+            ]
+            observed_camera_uncertainty = [
+                item.proposal_id for item in obstacle_map.camera_uncertainty
+            ]
+            if observed_camera_uncertainty != expected_camera_uncertainty:
+                mismatch["camera_uncertainty"] += 1
+            if [item.to_dict() for item in delivery.threats] != _array(
+                threat,
+                "assessments",
+            ):
+                mismatch["threat_assessments"] += 1
+            compared += 1
+        if temporal_stream.readline() or threat_stream.readline():
+            raise ReferenceGraphParityError("accepted parity ledger exceeds graph delivery count")
+    accepted = compared == expected_frames and all(value == 0 for value in mismatch.values())
+    return ReferenceGraphParityReport(
+        expected_frames=expected_frames,
+        compared_frames=compared,
+        temporal_frames_sha256=_sha256_file(temporal_path),
+        threat_frames_sha256=_sha256_file(threat_path),
+        mismatch_counts=tuple(sorted(mismatch.items())),
+        accepted=accepted,
+    )
+
+
+def _regular_file(path: Path, label: str) -> Path:
+    resolved = path.resolve(strict=True)
+    if resolved.is_symlink() or not resolved.is_file():
+        raise ReferenceGraphParityError(f"{label} must be a regular file")
+    return resolved
+
+
+def _read_row(raw: bytes, label: str, sequence: int) -> dict[str, object]:
+    if not raw:
+        raise ReferenceGraphParityError(f"{label} parity ledger ended at frame {sequence}")
+    try:
+        value = json.loads(raw)
+    except json.JSONDecodeError as exc:
+        raise ReferenceGraphParityError(f"{label} parity row is invalid JSON") from exc
+    if not isinstance(value, dict) or any(not isinstance(key, str) for key in value):
+        raise ReferenceGraphParityError(f"{label} parity row is not an object")
+    return value
+
+
+def _sha256_file(path: Path) -> str:
+    digest = hashlib.sha256()
+    with path.open("rb") as handle:
+        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+def _array(document: dict[str, object], key: str) -> list[dict[str, object]]:
+    value = document.get(key)
+    if not isinstance(value, list) or any(
+        not isinstance(item, dict) or any(not isinstance(name, str) for name in item)
+        for item in value
+    ):
+        raise ReferenceGraphParityError(f"parity {key} must be an object array")
+    return value
+
+
+def _string(document: dict[str, object], key: str) -> str:
+    value = document.get(key)
+    if not isinstance(value, str) or not value:
+        raise ReferenceGraphParityError(f"parity {key} must be a nonempty string")
+    return value
+
+
+def _boolean(document: dict[str, object], key: str) -> bool:
+    value = document.get(key)
+    if not isinstance(value, bool):
+        raise ReferenceGraphParityError(f"parity {key} must be boolean")
+    return value
+
+
+__all__ = [
+    "REFERENCE_GRAPH_PARITY_SCHEMA",
+    "ReferenceGraphParityError",
+    "ReferenceGraphParityReport",
+    "compare_reference_graph_to_accepted_ledgers",
+]
diff --git a/src/k1link/perception/reference_graph_result.py b/src/k1link/perception/reference_graph_result.py
new file mode 100644
index 0000000..c2b096b
--- /dev/null
+++ b/src/k1link/perception/reference_graph_result.py
@@ -0,0 +1,195 @@
+"""Immutable result sealing for the M4.7 canonical graph shadow."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import os
+import shutil
+import uuid
+from collections import Counter
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Final
+
+from .graph_contracts import GraphRunMode, GraphRunResultV2, GraphState, TerminalOutcomeType
+from .reference_graph_parity import ReferenceGraphParityReport
+
+REFERENCE_GRAPH_RESULT_PREFIX: Final = "m47-reference-graph-"
+REFERENCE_GRAPH_REPORT_SCHEMA: Final = "missioncore.reference-perception-graph-report/v1"
+REFERENCE_GRAPH_MANIFEST_SCHEMA: Final = "missioncore.reference-perception-graph-manifest/v1"
+
+
+class ReferenceGraphResultError(RuntimeError):
+    """A graph result cannot be sealed without complete terminal evidence."""
+
+
+@dataclass(frozen=True, slots=True)
+class SealedReferenceGraphResult:
+    result_id: str
+    result_root: Path
+    accepted: bool
+    report: dict[str, object]
+    manifest: dict[str, object]
+
+
+def seal_reference_graph_result(
+    result: GraphRunResultV2,
+    *,
+    output_root: Path,
+    expected_frames: int,
+    parity: ReferenceGraphParityReport,
+) -> SealedReferenceGraphResult:
+    if expected_frames < 1:
+        raise ReferenceGraphResultError("expected frame count must be positive")
+    outcomes = Counter(item.outcome.value for item in result.terminal_outcomes)
+    gates = {
+        "lossless_replay_mode": result.run_mode is GraphRunMode.LOSSLESS_REPLAY,
+        "graph_stopped": result.state is GraphState.STOPPED,
+        "admitted_frame_count": result.admitted_count == expected_frames,
+        "terminal_accounting_closed": len(result.terminal_outcomes) == expected_frames,
+        "delivery_count": outcomes[TerminalOutcomeType.DELIVERED.value] == expected_frames,
+        "delivery_payload_count": len(result.deliveries) == expected_frames,
+        "no_failed_frames": outcomes[TerminalOutcomeType.FAILED.value] == 0,
+        "no_stale_frames": outcomes[TerminalOutcomeType.STALE.value] == 0,
+        "no_superseded_frames": outcomes[TerminalOutcomeType.SUPERSEDED.value] == 0,
+        "no_rejected_frames": outcomes[TerminalOutcomeType.REJECTED.value] == 0,
+        "no_unavailable_frames": outcomes[TerminalOutcomeType.UNAVAILABLE.value] == 0,
+        "accepted_m45r_m46_parity": parity.accepted,
+    }
+    accepted = all(gates.values())
+    report: dict[str, object] = {
+        "schema_version": REFERENCE_GRAPH_REPORT_SCHEMA,
+        "graph_id": result.graph_id,
+        "source_profile_id": result.source_profile_id,
+        "run_mode": result.run_mode.value,
+        "state": result.state.value,
+        "expected_frames": expected_frames,
+        "admitted_frames": result.admitted_count,
+        "terminal_outcomes": dict(sorted(outcomes.items())),
+        "delivery_payloads": len(result.deliveries),
+        "queue_high_watermarks": dict(result.queue_high_watermarks),
+        "canonical_payload_sha256": result.canonical_payload_sha256,
+        "accepted_parity": parity.to_dict(),
+        "gates": gates,
+        "accepted": accepted,
+        "authority": {
+            "physical_live": False,
+            "commands_enabled": False,
+            "actuation_allowed": False,
+            "navigation_or_safety_accepted": False,
+        },
+    }
+    root = output_root.expanduser().resolve()
+    root.mkdir(mode=0o700, parents=True, exist_ok=True)
+    if root.is_symlink() or not root.is_dir():
+        raise ReferenceGraphResultError("result root must be a real directory")
+    staging = root / f".reference-graph.{uuid.uuid4().hex}.tmp"
+    staging.mkdir(mode=0o700, exist_ok=False)
+    try:
+        frames_path = staging / "frames.jsonl"
+        outcomes_path = staging / "outcomes.jsonl"
+        report_path = staging / "report.json"
+        _write_json_lines(
+            frames_path,
+            tuple(delivery.canonical_dict() for delivery in result.deliveries),
+        )
+        _write_json_lines(
+            outcomes_path,
+            tuple(outcome.to_dict() for outcome in result.terminal_outcomes),
+        )
+        _write_json(report_path, report)
+        file_rows = {
+            name: {
+                "sha256": _sha256_file(staging / name),
+                "bytes": (staging / name).stat().st_size,
+            }
+            for name in ("frames.jsonl", "outcomes.jsonl", "report.json")
+        }
+        identity: dict[str, object] = {
+            "graph_id": result.graph_id,
+            "source_profile_id": result.source_profile_id,
+            "run_mode": result.run_mode.value,
+            "canonical_payload_sha256": result.canonical_payload_sha256,
+            "files": file_rows,
+        }
+        identity_sha256 = hashlib.sha256(_canonical_json(identity)).hexdigest()
+        result_id = f"{REFERENCE_GRAPH_RESULT_PREFIX}{identity_sha256}"
+        manifest: dict[str, object] = {
+            "schema_version": REFERENCE_GRAPH_MANIFEST_SCHEMA,
+            "result_id": result_id,
+            "identity_sha256": identity_sha256,
+            **identity,
+            "accepted": accepted,
+        }
+        _write_json(staging / "manifest.json", manifest)
+        target = root / result_id
+        _publish_immutable(staging, target)
+    except Exception:
+        shutil.rmtree(staging, ignore_errors=True)
+        raise
+    return SealedReferenceGraphResult(
+        result_id=result_id,
+        result_root=target,
+        accepted=accepted,
+        report=report,
+        manifest=manifest,
+    )
+
+
+def _write_json_lines(path: Path, rows: tuple[dict[str, object], ...]) -> None:
+    with path.open("wb") as handle:
+        for row in rows:
+            handle.write(_canonical_json(row) + b"\n")
+        handle.flush()
+        os.fsync(handle.fileno())
+
+
+def _write_json(path: Path, document: dict[str, object]) -> None:
+    with path.open("wb") as handle:
+        handle.write(_canonical_json(document) + b"\n")
+        handle.flush()
+        os.fsync(handle.fileno())
+
+
+def _publish_immutable(staging: Path, target: Path) -> None:
+    if target.exists():
+        if target.is_symlink() or not target.is_dir():
+            raise ReferenceGraphResultError("immutable result target is not a real directory")
+        expected = {
+            path.name: _sha256_file(path)
+            for path in staging.iterdir()
+            if path.is_file()
+        }
+        observed = {
+            path.name: _sha256_file(path)
+            for path in target.iterdir()
+            if path.is_file()
+        }
+        if expected != observed:
+            raise ReferenceGraphResultError("immutable result identity collision")
+        shutil.rmtree(staging)
+        return
+    staging.replace(target)
+
+
+def _canonical_json(document: dict[str, object]) -> bytes:
+    return json.dumps(document, sort_keys=True, separators=(",", ":")).encode("utf-8")
+
+
+def _sha256_file(path: Path) -> str:
+    digest = hashlib.sha256()
+    with path.open("rb") as handle:
+        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+__all__ = [
+    "REFERENCE_GRAPH_MANIFEST_SCHEMA",
+    "REFERENCE_GRAPH_REPORT_SCHEMA",
+    "REFERENCE_GRAPH_RESULT_PREFIX",
+    "ReferenceGraphResultError",
+    "SealedReferenceGraphResult",
+    "seal_reference_graph_result",
+]
diff --git a/src/k1link/perception/reference_graph_runtime.py b/src/k1link/perception/reference_graph_runtime.py
new file mode 100644
index 0000000..9e1e17d
--- /dev/null
+++ b/src/k1link/perception/reference_graph_runtime.py
@@ -0,0 +1,190 @@
+"""Production assembly boundary for the source-neutral M4.7 reference graph."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+from dataclasses import dataclass
+from pathlib import Path
+
+from .baseline import load_m4_baseline
+from .detector import FrozenYoloxDetectorProvider
+from .geometry import (
+    Ravnoves00GeometryAssociationProvider,
+    RecordedGeometryStore,
+    load_geometry_profile,
+)
+from .graph import ReferencePerceptionGraphV2
+from .graph_contracts import GraphRunMode
+from .motion import ClassIndependentMotionEstimator
+from .providers import ProviderRole, ReferencePerceptionGraphConfigV2
+from .recorded_source import (
+    DecodedRecordedSource,
+    PyAvRecordedImageDecoder,
+    RecordedRavnoves00Source,
+    ReplayPacing,
+)
+from .rolling_map import RollingLocalObstacleMapProvider, load_rolling_map_profile
+from .temporal import BoundedSpatialTemporalProvider, load_temporal_motion_profile
+from .threat import (
+    DualEvidenceReplayThreatProvider,
+    RecordedReplayBodyFrameResolver,
+    load_replay_threat_profile,
+)
+from .yolox_object_detector import TritonHttpInferenceBackend, load_valid_fov_mask
+
+
+class ReferenceGraphRuntimeError(RuntimeError):
+    """The production graph cannot be assembled from the pinned release inputs."""
+
+
+@dataclass(frozen=True, slots=True)
+class ReferenceGraphRuntimePaths:
+    graph_config: Path
+    baseline_profile: Path
+    geometry_profile: Path
+    temporal_motion_profile: Path
+    rolling_map_profile: Path
+    threat_profile: Path
+    camera_index: Path
+    source_pack: Path
+    local_surface: Path
+    video: Path
+    valid_fov_mask: Path
+
+
+@dataclass(slots=True)
+class ReferenceGraphRuntime:
+    """Own one graph and its persistent inference transport."""
+
+    graph: ReferencePerceptionGraphV2
+    inference_backend: TritonHttpInferenceBackend
+
+    def close(self) -> None:
+        self.inference_backend.close()
+
+    def __enter__(self) -> ReferenceGraphRuntime:
+        return self
+
+    def __exit__(self, *args: object) -> None:
+        self.close()
+
+
+def build_reference_graph_runtime(
+    *,
+    paths: ReferenceGraphRuntimePaths,
+    triton_origin: str,
+    run_mode: GraphRunMode,
+) -> ReferenceGraphRuntime:
+    """Instantiate every admitted provider from explicit immutable inputs."""
+
+    config = _load_graph_config(paths.graph_config)
+    pinned_files = {
+        ProviderRole.SOURCE: paths.baseline_profile,
+        ProviderRole.DETECTOR: paths.baseline_profile,
+        ProviderRole.GEOMETRY: paths.geometry_profile,
+        ProviderRole.TEMPORAL: paths.temporal_motion_profile,
+        ProviderRole.MOTION: paths.temporal_motion_profile,
+        ProviderRole.ROLLING: paths.rolling_map_profile,
+        ProviderRole.THREAT: paths.threat_profile,
+    }
+    _validate_provider_digests(config, pinned_files)
+
+    load_m4_baseline(paths.baseline_profile)
+    geometry_profile = load_geometry_profile(paths.geometry_profile)
+    temporal_motion_profile = load_temporal_motion_profile(paths.temporal_motion_profile)
+    rolling_map_profile = load_rolling_map_profile(paths.rolling_map_profile)
+    threat_profile = load_replay_threat_profile(paths.threat_profile)
+
+    source = DecodedRecordedSource(
+        source=RecordedRavnoves00Source(
+            camera_index_path=paths.camera_index,
+            source_pack_path=paths.source_pack,
+            pacing=(
+                ReplayPacing.ONE_X
+                if run_mode is GraphRunMode.SOURCE_PACED_LATEST_WINS
+                else ReplayPacing.UNCAPPED
+            ),
+        ),
+        decoder=PyAvRecordedImageDecoder(paths.video),
+    )
+    backend = TritonHttpInferenceBackend(triton_origin)
+    try:
+        store = RecordedGeometryStore(
+            source_pack_path=paths.source_pack,
+            local_surface_path=paths.local_surface,
+            profile=geometry_profile,
+        )
+        body_frame_resolver = RecordedReplayBodyFrameResolver(
+            store,
+            profile=threat_profile.body_frame,
+        )
+        graph = ReferencePerceptionGraphV2(
+            config=config,
+            source=source,
+            detector=FrozenYoloxDetectorProvider(
+                mask=load_valid_fov_mask(paths.valid_fov_mask),
+                backend=backend,
+            ),
+            geometry=Ravnoves00GeometryAssociationProvider(store=store),
+            temporal=BoundedSpatialTemporalProvider(
+                point_resolver=store,
+                profile=temporal_motion_profile,
+            ),
+            motion=ClassIndependentMotionEstimator(profile=temporal_motion_profile),
+            rolling=RollingLocalObstacleMapProvider(
+                pose_resolver=store,
+                profile=rolling_map_profile,
+            ),
+            threat=DualEvidenceReplayThreatProvider(
+                body_frame_resolver=body_frame_resolver,
+                profile=threat_profile,
+            ),
+            run_mode=run_mode,
+        )
+    except Exception:
+        backend.close()
+        raise
+    return ReferenceGraphRuntime(graph=graph, inference_backend=backend)
+
+
+def _load_graph_config(path: Path) -> ReferencePerceptionGraphConfigV2:
+    try:
+        document = json.loads(path.resolve(strict=True).read_text("utf-8"))
+        return ReferencePerceptionGraphConfigV2.from_dict(document)
+    except (OSError, json.JSONDecodeError, ValueError) as exc:
+        raise ReferenceGraphRuntimeError("reference graph config is invalid") from exc
+
+
+def _validate_provider_digests(
+    config: ReferencePerceptionGraphConfigV2,
+    pinned_files: dict[ProviderRole, Path],
+) -> None:
+    pins = {pin.role: pin for pin in config.providers}
+    if set(pins) != set(pinned_files):
+        raise ReferenceGraphRuntimeError("reference graph provider pins are incomplete")
+    for role, path in pinned_files.items():
+        if _sha256_file(path) != pins[role].sha256:
+            raise ReferenceGraphRuntimeError(f"{role.value} provider profile digest changed")
+
+
+def _sha256_file(path: Path) -> str:
+    try:
+        resolved = path.resolve(strict=True)
+    except OSError as exc:
+        raise ReferenceGraphRuntimeError("pinned graph input is missing") from exc
+    if resolved.is_symlink() or not resolved.is_file():
+        raise ReferenceGraphRuntimeError("pinned graph input must be a regular file")
+    digest = hashlib.sha256()
+    with resolved.open("rb") as handle:
+        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+__all__ = [
+    "ReferenceGraphRuntime",
+    "ReferenceGraphRuntimeError",
+    "ReferenceGraphRuntimePaths",
+    "build_reference_graph_runtime",
+]
diff --git a/tests/test_m47_worker_graph_shadow_artifact.py b/tests/test_m47_worker_graph_shadow_artifact.py
new file mode 100644
index 0000000..7236b03
--- /dev/null
+++ b/tests/test_m47_worker_graph_shadow_artifact.py
@@ -0,0 +1,126 @@
+from __future__ import annotations
+
+import hashlib
+import importlib.util
+import json
+import tarfile
+from pathlib import Path
+
+REPOSITORY_ROOT = Path(__file__).resolve().parents[1]
+BUILDER_PATH = REPOSITORY_ROOT / "scripts/build_m47_worker_graph_shadow_artifact.py"
+SPEC = importlib.util.spec_from_file_location("m47_worker_graph_shadow_builder", BUILDER_PATH)
+assert SPEC is not None and SPEC.loader is not None
+BUILDER = importlib.util.module_from_spec(SPEC)
+SPEC.loader.exec_module(BUILDER)
+
+
+def _sha256(value: bytes) -> str:
+    return hashlib.sha256(value).hexdigest()
+
+
+def _regular_files(archive: tarfile.TarFile) -> dict[str, bytes]:
+    result: dict[str, bytes] = {}
+    for member in archive.getmembers():
+        if not member.isfile():
+            continue
+        extracted = archive.extractfile(member)
+        assert extracted is not None
+        result[member.name] = extracted.read()
+    return result
+
+
+def test_m47_worker_artifact_is_deterministic_and_self_contained(tmp_path: Path) -> None:
+    patch_id = "mission-core-m47-graph-shadow-unit-001"
+    revision = "d" * 40
+    first = BUILDER.build_artifact(patch_id, tmp_path / "first", revision=revision)
+    second = BUILDER.build_artifact(patch_id, tmp_path / "second", revision=revision)
+    first_bytes = Path(first["artifact"]).read_bytes()
+    second_bytes = Path(second["artifact"]).read_bytes()
+
+    assert first_bytes == second_bytes
+    assert first["sha256"] == _sha256(first_bytes)
+    assert first["transition"] == "m47-canonical-graph-shadow-v1"
+    with tarfile.open(first["artifact"], "r:gz") as archive:
+        regular = _regular_files(archive)
+    payload_names = sorted(
+        name.removeprefix("payload/")
+        for name in regular
+        if name.startswith("payload/")
+    )
+    assert payload_names == first["payload_files"]
+    assert regular["files.txt"].decode().splitlines() == first["payload_files"]
+    assert _sha256(regular[f"payload/{BUILDER.WHEEL_NAME}"]) == first["wheel_sha256"]
+    assert regular[f"payload/{BUILDER.RUNNER.name}"] == BUILDER.RUNNER.read_bytes()
+    for relative in BUILDER.CONFIG_PATHS:
+        assert regular[f"payload/{relative.name}"] == (
+            BUILDER.REPOSITORY_ROOT / relative
+        ).read_bytes()
+
+
+def test_m47_descriptor_preserves_predecessor_and_separates_readiness() -> None:
+    descriptor = json.loads(
+        BUILDER.render_descriptor(
+            "mission-core-m47-graph-shadow-unit-002",
+            "e" * 40,
+            wheel_sha256="f" * 64,
+        )
+    )
+
+    assert descriptor["schema_version"] == "nodedc.mission-core-worker.shadow-release/v2"
+    assert descriptor["transition"] == "m47-canonical-graph-shadow-v1"
+    assert descriptor["boundary"]["external_deploy_registry"] is False
+    assert descriptor["container"]["public_ports"] is False
+    assert descriptor["inputs"]["local_surface"]["sha256"] == (
+        "f57eb2485b6cef47f2a97a2d9ff1aa9fd9265fe1eb69cd5852d12f39e13b8bc6"
+    )
+    assert descriptor["inputs"]["accepted_temporal_frames"]["sha256"] == (
+        "e83b80ea06b3462c2d04c5a1b74289a0ec401596c7150ae90f5a1b748b639c3a"
+    )
+    assert descriptor["inputs"]["accepted_threat_frames"]["sha256"] == (
+        "b57be1839f5915e3b80b54355b694e0bd8c9ac318d0cbe6de2bff713082cfa4e"
+    )
+    assert descriptor["predecessor"]["durable_worker"]["name"] == (
+        "ndc-mission-core-perception-worker"
+    )
+    assert descriptor["rollback"] == {
+        "durable_worker_action": "none",
+        "preserve_failed_evidence": True,
+        "remove_candidate_container": True,
+        "remove_unaccepted_release": True,
+        "triton_action": "none",
+    }
+    assert descriptor["acceptance"] == {
+        "run_mode": "lossless-replay",
+        "expected_frames": 4489,
+        "delivered_frames": 4489,
+        "failed_frames": 0,
+        "stale_frames": 0,
+        "superseded_frames": 0,
+        "accepted_parity": True,
+        "class_routing_used": False,
+    }
+    assert descriptor["readiness"]["graph"] == {
+        "graph_id": "reference-perception-graph/v2",
+        "terminal_accounting_required": True,
+        "actuation_allowed": False,
+    }
+    assert set(descriptor["release"]["configs"]) == {
+        path.name for path in BUILDER.CONFIG_PATHS
+    }
+    serialized = json.dumps(descriptor).encode()
+    assert b"PRIVATE KEY" not in serialized
+    assert b"password=" not in serialized.lower()
+
+
+def test_m47_runner_calls_only_the_canonical_graph_entrypoint() -> None:
+    runner = BUILDER.RUNNER.read_text("utf-8")
+
+    assert "k1link.perception.reference_graph_cli" in runner
+    assert "k1link.perception.detector_replay_cli" not in runner
+    assert '"--mode", $descriptor.acceptance.run_mode' in runner
+    assert '"--temporal-parity-frames"' in runner
+    assert '"--threat-parity-frames"' in runner
+    assert 'Write-Output "PROVIDER_READINESS=accepted"' in runner
+    assert 'Write-Output "GRAPH_READINESS=accepted"' in runner
+    assert 'Write-Output "DURABLE_WORKER_ACTION=none"' in runner
+    assert 'Write-Output "TRITON_ACTION=none"' in runner
diff --git a/tests/test_m4_worker_shadow_artifact.py b/tests/test_m4_worker_shadow_artifact.py
index 22daa79..30a235f 100644
--- a/tests/test_m4_worker_shadow_artifact.py
+++ b/tests/test_m4_worker_shadow_artifact.py
@@ -1,11 +1,11 @@
 from __future__ import annotations
 
-import hashlib
 import importlib.util
 import json
-import tarfile
 from pathlib import Path
 
+import pytest
+
 REPOSITORY_ROOT = Path(__file__).resolve().parents[1]
 BUILDER_PATH = REPOSITORY_ROOT / "scripts/build_m4_worker_shadow_artifact.py"
 SPEC = importlib.util.spec_from_file_location("m4_worker_shadow_builder", BUILDER_PATH)
@@ -13,98 +13,12 @@ assert SPEC is not None and SPEC.loader is not None
 BUILDER = importlib.util.module_from_spec(SPEC)
 SPEC.loader.exec_module(BUILDER)
 
-
-def _sha256(value: bytes) -> str:
-    return hashlib.sha256(value).hexdigest()
-
-
-def _regular_files(archive: tarfile.TarFile) -> dict[str, bytes]:
-    result: dict[str, bytes] = {}
-    for member in archive.getmembers():
-        if not member.isfile():
-            continue
-        extracted = archive.extractfile(member)
-        assert extracted is not None
-        result[member.name] = extracted.read()
-    return result
-
-
-def test_worker_shadow_artifact_is_deterministic_narrow_and_self_contained(
+def test_historical_worker_shadow_refuses_runtime_relabeling(
     tmp_path: Path,
 ) -> None:
     patch_id = "mission-core-m4-detector-shadow-unit-001"
-    first = BUILDER.build_artifact(patch_id, tmp_path / "first")
-    second = BUILDER.build_artifact(patch_id, tmp_path / "second")
-    first_bytes = Path(first["artifact"]).read_bytes()
-    second_bytes = Path(second["artifact"]).read_bytes()
-
-    assert first_bytes == second_bytes
-    assert first["sha256"] == _sha256(first_bytes)
-    assert first["sha256"] == second["sha256"]
-    assert first["payload_files"] == list(BUILDER.PAYLOAD_FILES)
-
-    with tarfile.open(first["artifact"], "r:gz") as archive:
-        members = archive.getmembers()
-        names = [member.name for member in members]
-        regular = _regular_files(archive)
-
-    assert names == [
-        "manifest.env",
-        "files.txt",
-        "payload",
-        "payload/Invoke-M4DetectorShadow.ps1",
-        "payload/m4-recorded-realtime-baseline-v1.json",
-        "payload/mission-core-worker-shadow-v1.json",
-        f"payload/{BUILDER.WHEEL_NAME}",
-    ]
-    assert (
-        regular["manifest.env"]
-        == (f"id={patch_id}\ncomponent=mission-core-worker\ntype=shadow-release\n").encode()
-    )
-    assert regular["files.txt"].decode().splitlines() == list(BUILDER.PAYLOAD_FILES)
-    assert regular[f"payload/{BUILDER.RUNNER_NAME}"] == BUILDER.RUNNER.read_bytes()
-    assert _sha256(regular[f"payload/{BUILDER.WHEEL_NAME}"]) == (BUILDER.EXPECTED_WHEEL_SHA256)
-    assert _sha256(regular["payload/m4-recorded-realtime-baseline-v1.json"]) == (
-        BUILDER.EXPECTED_BASELINE_SHA256
-    )
-    descriptor = json.loads(regular["payload/mission-core-worker-shadow-v1.json"])
-    assert descriptor["patch_id"] == patch_id
-    assert descriptor["code_revision"] == first["code_revision"]
-    assert descriptor["boundary"] == {
-        "external_deploy_registry": False,
-        "nodedc_platform_repository": False,
-        "repository": "NODEDC_MISSION_CORE",
-        "server_docker_runtime": False,
-    }
-    assert descriptor["dependency_tree_identity"] == (
-        "relative-path-tab-size-tab-file-sha256-lf/v1"
-    )
-    assert descriptor["release"]["runner"] == {
-        "name": BUILDER.RUNNER_NAME,
-        "sha256": _sha256(BUILDER.RUNNER.read_bytes()),
-    }
-    assert descriptor["release"]["wheel"] == {
-        "name": BUILDER.WHEEL_NAME,
-        "sha256": BUILDER.EXPECTED_WHEEL_SHA256,
-    }
-    assert descriptor["container"]["public_ports"] is False
-    assert descriptor["rollback"] == {
-        "durable_worker_action": "none",
-        "preserve_failed_evidence": True,
-        "remove_candidate_container": True,
-        "remove_unaccepted_release": True,
-        "triton_action": "none",
-    }
-    all_bytes = b"\n".join(regular.values())
-    assert b"PRIVATE KEY" not in all_bytes
-    assert b"password=" not in all_bytes.lower()
-    assert not any(
-        Path(name).name.startswith(".env")
-        or "/secrets/" in name
-        or "/runtime/" in name
-        or "/recordings/" in name
-        for name in names
-    )
+    with pytest.raises(BUILDER.ArtifactBuildError, match="runtime wheel digest changed"):
+        BUILDER.build_artifact(patch_id, tmp_path)
 
 
 def test_worker_shadow_descriptor_pins_external_inputs_and_preserves_e15() -> None:
diff --git a/tests/test_perception_contracts.py b/tests/test_perception_contracts.py
index 1c73013..54b6ea0 100644
--- a/tests/test_perception_contracts.py
+++ b/tests/test_perception_contracts.py
@@ -320,6 +320,7 @@ def test_reference_graph_config_pins_all_roles_and_queue_bounds() -> None:
         providers=tuple(
             ProviderPin(role, f"{role.value}-provider", "v1", "78a3dc2", "a" * 64)
             for role in ProviderRole
+            if role is not ProviderRole.ROLLING
         ),
         queues=tuple(
             QueuePolicy(stage, 2, 80_000_000, 200_000_000)
diff --git a/tests/test_perception_graph.py b/tests/test_perception_graph.py
index 9bfb69a..71bc76a 100644
--- a/tests/test_perception_graph.py
+++ b/tests/test_perception_graph.py
@@ -5,6 +5,7 @@ import threading
 from collections.abc import Iterator
 from dataclasses import replace
 from pathlib import Path
+from queue import Queue
 from threading import Event
 
 import numpy as np
@@ -36,9 +37,12 @@ from k1link.perception.contracts import (
     TimestampBundle,
 )
 from k1link.perception.graph import (
+    GraphExecutionError,
+    GraphRunMode,
     GraphRunResult,
     GraphState,
     ReferencePerceptionGraphV1,
+    ReferencePerceptionGraphV2,
     TerminalOutcomeType,
 )
 from k1link.perception.providers import (
@@ -47,6 +51,7 @@ from k1link.perception.providers import (
     ProviderRole,
     QueuePolicy,
     ReferencePerceptionGraphConfig,
+    ReferencePerceptionGraphConfigV2,
     SourcePacket,
 )
 from k1link.perception.recorded_source import (
@@ -230,6 +235,44 @@ class _Motion:
         return obstacles
 
 
+class _Rolling:
+    provider_id = "test-rolling/v1"
+
+    def update(
+        self,
+        packet: SourcePacket,
+        obstacles: tuple[TemporalObstacle, ...],
+    ) -> tuple[TemporalObstacle, ...]:
+        if packet.envelope.sequence == 0:
+            return ()
+        return (
+            TemporalObstacle(
+                component_id=f"rolling-{packet.envelope.sequence}",
+                identity_scope="ephemeral",
+                state=TemporalState.RETAINED,
+                ttl_ns=3_000_000_000,
+                last_hit_ns=packet.envelope.timestamps.source_ns - 100_000_000,
+                age_ns=100_000_000,
+                association_basis="registered-map-increment-retention",
+                history=(
+                    HistorySample(
+                        frame_id=f"frame-{packet.envelope.sequence - 1:06d}",
+                        evidence_time_ns=(
+                            packet.envelope.timestamps.source_ns - 100_000_000
+                        ),
+                        centroid_xyz_m=(3.0, 0.5, 0.5),
+                    ),
+                ),
+                cells=(GridCell(99, packet.envelope.sequence, 0),),
+                coordinate_frame="map",
+                last_centroid_xyz_m=(3.0, 0.5, 0.5),
+                motion=MotionState.UNKNOWN,
+                motion_confidence=0.0,
+                motion_reason="retained-map-increment-no-current-motion",
+            ),
+        )
+
+
 class _Threat:
     provider_id = "test-threat/v1"
 
@@ -312,6 +355,39 @@ def _config(
     )
 
 
+def _config_v2(
+    capacity: int = 8,
+    terminal_timeout_ns: int = 500_000_000,
+) -> ReferencePerceptionGraphConfigV2:
+    ids = {
+        ProviderRole.SOURCE: _Source.provider_id,
+        ProviderRole.DETECTOR: _Detector.provider_id,
+        ProviderRole.GEOMETRY: _Geometry.provider_id,
+        ProviderRole.TEMPORAL: _Temporal.provider_id,
+        ProviderRole.MOTION: _Motion.provider_id,
+        ProviderRole.ROLLING: _Rolling.provider_id,
+        ProviderRole.THREAT: _Threat.provider_id,
+    }
+    return ReferencePerceptionGraphConfigV2(
+        graph_id="reference-perception-graph/v2",
+        source_profile_id=BASELINE_PROFILE_ID,
+        providers=tuple(
+            ProviderPin(role, provider_id, "v1", "test-revision", "b" * 64)
+            for role, provider_id in ids.items()
+        ),
+        queues=tuple(
+            QueuePolicy(
+                stage,
+                capacity,
+                min(80_000_000, terminal_timeout_ns),
+                terminal_timeout_ns,
+            )
+            for stage in ("detector", "geometry", "temporal", "rolling", "threat")
+        ),
+        authority=GraphAuthority(),
+    )
+
+
 class _MemoryTelemetry:
     def __init__(self) -> None:
         self.records: list[dict[str, object]] = []
@@ -358,6 +434,28 @@ def _graph(
     )
 
 
+def _graph_v2(
+    source: object,
+    *,
+    detector: object | None = None,
+    capacity: int = 8,
+    run_mode: GraphRunMode = GraphRunMode.LOSSLESS_REPLAY,
+    terminal_timeout_ns: int = 500_000_000,
+) -> ReferencePerceptionGraphV2:
+    return ReferencePerceptionGraphV2(
+        config=_config_v2(capacity, terminal_timeout_ns),
+        source=source,
+        detector=detector or _Detector(),
+        geometry=_Geometry(),
+        temporal=_Temporal(),
+        motion=_Motion(),
+        rolling=_Rolling(),
+        threat=_Threat(),
+        run_mode=run_mode,
+        clock_ns=lambda: 10_000,
+    )
+
+
 def test_reference_graph_closes_accounting_telemetry_and_deterministic_digest() -> None:
     telemetry = _MemoryTelemetry()
     graph = _graph(_Source((_packet(0), _packet(1))), telemetry=telemetry)
@@ -386,6 +484,88 @@ def test_reference_graph_closes_accounting_telemetry_and_deterministic_digest()
     assert {"detector", "geometry", "temporal", "threat"}.issubset(stage_ids)
 
 
+def test_reference_graph_v2_publishes_current_and_retained_occupancy() -> None:
+    result = _graph_v2(_Source((_packet(0), _packet(1)))).run()
+
+    assert result.graph_id == "reference-perception-graph/v2"
+    assert result.run_mode is GraphRunMode.LOSSLESS_REPLAY
+    assert [item.outcome for item in result.terminal_outcomes] == [
+        TerminalOutcomeType.DELIVERED,
+        TerminalOutcomeType.DELIVERED,
+    ]
+    second = result.deliveries[1].obstacle_map
+    assert [item.state for item in second.occupied] == [
+        TemporalState.CURRENT,
+        TemporalState.RETAINED,
+    ]
+    assert second.unknown == ()
+    assert set(dict(result.queue_high_watermarks)) == {
+        "detector",
+        "geometry",
+        "temporal",
+        "rolling",
+        "threat",
+    }
+
+
+def test_reference_graph_v2_lossless_mode_applies_bounded_backpressure() -> None:
+    release = Event()
+
+    class BlockingDetector(_Detector):
+        def detect(self, packet: SourcePacket) -> tuple[ObjectProposal2D, ...]:
+            if packet.envelope.sequence == 0:
+                assert release.wait(2)
+            return super().detect(packet)
+
+    BlockingDetector.provider_id = _Detector.provider_id
+    graph = _graph_v2(
+        _Source(tuple(_packet(index) for index in range(4))),
+        detector=BlockingDetector(),
+        capacity=1,
+    )
+    holder: list[GraphRunResult] = []
+    runner = threading.Thread(target=lambda: holder.append(graph.run()))
+    runner.start()
+    release.set()
+    runner.join(3)
+
+    assert not runner.is_alive()
+    assert len(holder[0].terminal_outcomes) == 4
+    assert all(
+        item.outcome is TerminalOutcomeType.DELIVERED
+        for item in holder[0].terminal_outcomes
+    )
+
+
+def test_reference_graph_v2_restart_requires_fresh_stateful_providers() -> None:
+    graph = _graph_v2(_Source((_packet(0),)))
+
+    assert graph.run().state is GraphState.STOPPED
+    with pytest.raises(
+        GraphExecutionError,
+        match="restart requires freshly instantiated providers",
+    ):
+        graph.run()
+
+
+def test_reference_graph_v2_terminal_timeout_fails_stranded_packet() -> None:
+    packet = _packet(0)
+    graph = _graph_v2(
+        _Source(()),
+        capacity=1,
+        terminal_timeout_ns=1,
+    )
+    queue = Queue(maxsize=1)
+    queue.put_nowait(packet)
+
+    graph._put_stop(queue, "detector")
+
+    outcome = graph._outcomes[0]
+    assert outcome.outcome is TerminalOutcomeType.FAILED
+    assert outcome.reason == "terminal-queue-timeout"
+    assert queue.qsize() == 1
+
+
 def test_reference_graph_marks_unavailable_stale_and_provider_failure() -> None:
     class FailingDetector(_Detector):
         def detect(self, packet: SourcePacket) -> tuple[ObjectProposal2D, ...]:
diff --git a/tests/test_reference_graph_parity.py b/tests/test_reference_graph_parity.py
new file mode 100644
index 0000000..16ed30d
--- /dev/null
+++ b/tests/test_reference_graph_parity.py
@@ -0,0 +1,121 @@
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+from k1link.perception.contracts import LocalObstacleMap, SourceAccounting
+from k1link.perception.graph_contracts import (
+    REFERENCE_GRAPH_ID_V2,
+    DeliveredFrame,
+    GraphRunMode,
+    GraphState,
+    TerminalOutcome,
+    TerminalOutcomeType,
+    build_graph_run_result_v2,
+)
+from k1link.perception.reference_graph_parity import (
+    compare_reference_graph_to_accepted_ledgers,
+)
+
+
+def _result():
+    obstacle_map = LocalObstacleMap(
+        source_id="RAVNOVES00",
+        session_id="20260720T065719Z_viewer_live",
+        frame_id="frame-000000",
+        graph_id=REFERENCE_GRAPH_ID_V2,
+        generated_monotonic_ns=1,
+        output_age_ns=0,
+        occupied=(),
+        unknown=(),
+        camera_uncertainty=(),
+        accounting=SourceAccounting(1, 1, 0, 0),
+    )
+    return build_graph_run_result_v2(
+        graph_id=REFERENCE_GRAPH_ID_V2,
+        source_profile_id="m4-ravnoves00-recorded-realtime/v1",
+        run_mode=GraphRunMode.LOSSLESS_REPLAY,
+        state=GraphState.STOPPED,
+        admitted_count=1,
+        outcomes=(
+            TerminalOutcome(
+                source_id="RAVNOVES00",
+                session_id="20260720T065719Z_viewer_live",
+                frame_id="frame-000000",
+                sequence=0,
+                outcome=TerminalOutcomeType.DELIVERED,
+                stage_id="threat",
+                reason="object-payload-delivered",
+            ),
+        ),
+        deliveries=(DeliveredFrame(0, obstacle_map, ()),),
+        queue_high_watermarks=(("detector", 1),),
+    )
+
+
+def _write_ledgers(root: Path, *, threat_assessments: list[dict[str, object]]) -> tuple[Path, Path]:
+    temporal = root / "temporal.jsonl"
+    threat = root / "threat.jsonl"
+    temporal.write_text(
+        json.dumps(
+            {
+                "sequence": 0,
+                "frame_id": "frame-000000",
+                "current": [],
+                "rolling_retained": [],
+                "held": [],
+                "expired": [],
+            }
+        )
+        + "\n",
+        "utf-8",
+    )
+    threat.write_text(
+        json.dumps(
+            {
+                "sequence": 0,
+                "frame_id": "frame-000000",
+                "camera_proposals": [],
+                "assessments": threat_assessments,
+            }
+        )
+        + "\n",
+        "utf-8",
+    )
+    return temporal, threat
+
+
+def test_reference_graph_parity_accepts_exact_frame_semantics(tmp_path: Path) -> None:
+    temporal, threat = _write_ledgers(tmp_path, threat_assessments=[])
+
+    report = compare_reference_graph_to_accepted_ledgers(
+        _result(),
+        temporal_frames_path=temporal,
+        threat_frames_path=threat,
+        expected_frames=1,
+    )
+
+    assert report.accepted is True
+    assert dict(report.mismatch_counts) == {
+        "camera_uncertainty": 0,
+        "current": 0,
+        "expired": 0,
+        "held": 0,
+        "rolling_retained": 0,
+        "source_binding": 0,
+        "threat_assessments": 0,
+    }
+
+
+def test_reference_graph_parity_reports_threat_drift(tmp_path: Path) -> None:
+    temporal, threat = _write_ledgers(tmp_path, threat_assessments=[{"component_id": "x"}])
+
+    report = compare_reference_graph_to_accepted_ledgers(
+        _result(),
+        temporal_frames_path=temporal,
+        threat_frames_path=threat,
+        expected_frames=1,
+    )
+
+    assert report.accepted is False
+    assert dict(report.mismatch_counts)["threat_assessments"] == 1
diff --git a/tests/test_reference_graph_result.py b/tests/test_reference_graph_result.py
new file mode 100644
index 0000000..d62c4fc
--- /dev/null
+++ b/tests/test_reference_graph_result.py
@@ -0,0 +1,152 @@
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+from k1link.perception.contracts import LocalObstacleMap, SourceAccounting
+from k1link.perception.graph_contracts import (
+    REFERENCE_GRAPH_ID_V2,
+    DeliveredFrame,
+    GraphRunMode,
+    GraphState,
+    TerminalOutcome,
+    TerminalOutcomeType,
+    build_graph_run_result_v2,
+)
+from k1link.perception.providers import ReferencePerceptionGraphConfigV2
+from k1link.perception.reference_graph_parity import ReferenceGraphParityReport
+from k1link.perception.reference_graph_result import seal_reference_graph_result
+
+REPOSITORY_ROOT = Path(__file__).resolve().parents[1]
+
+
+def _parity(accepted: bool = True) -> ReferenceGraphParityReport:
+    return ReferenceGraphParityReport(
+        expected_frames=1,
+        compared_frames=1,
+        temporal_frames_sha256="a" * 64,
+        threat_frames_sha256="b" * 64,
+        mismatch_counts=(("threat_assessments", 0 if accepted else 1),),
+        accepted=accepted,
+    )
+
+
+def _result(outcome: TerminalOutcomeType = TerminalOutcomeType.DELIVERED):
+    terminal = TerminalOutcome(
+        source_id="RAVNOVES00",
+        session_id="20260720T065719Z_viewer_live",
+        frame_id="frame-000000",
+        sequence=0,
+        outcome=outcome,
+        stage_id="threat" if outcome is TerminalOutcomeType.DELIVERED else "detector",
+        reason=(
+            "object-payload-delivered"
+            if outcome is TerminalOutcomeType.DELIVERED
+            else "bounded-queue-latest-wins"
+        ),
+    )
+    deliveries = (
+        (
+            DeliveredFrame(
+                sequence=0,
+                obstacle_map=LocalObstacleMap(
+                    source_id="RAVNOVES00",
+                    session_id="20260720T065719Z_viewer_live",
+                    frame_id="frame-000000",
+                    graph_id=REFERENCE_GRAPH_ID_V2,
+                    generated_monotonic_ns=123,
+                    output_age_ns=45,
+                    occupied=(),
+                    unknown=(),
+                    camera_uncertainty=(),
+                    accounting=SourceAccounting(1, 1, 0, 0),
+                ),
+                threats=(),
+            ),
+        )
+        if outcome is TerminalOutcomeType.DELIVERED
+        else ()
+    )
+    return build_graph_run_result_v2(
+        graph_id=REFERENCE_GRAPH_ID_V2,
+        source_profile_id="m4-ravnoves00-recorded-realtime/v1",
+        run_mode=GraphRunMode.LOSSLESS_REPLAY,
+        state=GraphState.STOPPED,
+        admitted_count=1,
+        outcomes=(terminal,),
+        deliveries=deliveries,
+        queue_high_watermarks=(("detector", 1), ("threat", 1)),
+    )
+
+
+def test_m47_graph_config_round_trips_and_pins_rolling_stage() -> None:
+    path = REPOSITORY_ROOT / "config/perception/m4-reference-graph-v2.json"
+    document = json.loads(path.read_text("utf-8"))
+    config = ReferencePerceptionGraphConfigV2.from_dict(document)
+
+    assert config.to_dict() == document
+    assert config.graph_id == REFERENCE_GRAPH_ID_V2
+    assert {provider.role.value for provider in config.providers} == {
+        "source",
+        "detector",
+        "geometry",
+        "temporal",
+        "motion",
+        "rolling",
+        "threat",
+    }
+    assert {queue.stage_id for queue in config.queues} == {
+        "detector",
+        "geometry",
+        "temporal",
+        "rolling",
+        "threat",
+    }
+
+
+def test_reference_graph_result_is_content_addressed_and_reproducible(tmp_path: Path) -> None:
+    first = seal_reference_graph_result(
+        _result(),
+        output_root=tmp_path / "one",
+        expected_frames=1,
+        parity=_parity(),
+    )
+    second = seal_reference_graph_result(
+        _result(),
+        output_root=tmp_path / "two",
+        expected_frames=1,
+        parity=_parity(),
+    )
+
+    assert first.accepted is True
+    assert first.result_id == second.result_id
+    assert first.report["gates"] == {
+        "lossless_replay_mode": True,
+        "graph_stopped": True,
+        "admitted_frame_count": True,
+        "terminal_accounting_closed": True,
+        "delivery_count": True,
+        "delivery_payload_count": True,
+        "no_failed_frames": True,
+        "no_stale_frames": True,
+        "no_superseded_frames": True,
+        "no_rejected_frames": True,
+        "no_unavailable_frames": True,
+        "accepted_m45r_m46_parity": True,
+    }
+    assert {
+        path.name: path.read_bytes() for path in first.result_root.iterdir()
+    } == {path.name: path.read_bytes() for path in second.result_root.iterdir()}
+
+
+def test_reference_graph_result_fails_closed_on_supersession(tmp_path: Path) -> None:
+    sealed = seal_reference_graph_result(
+        _result(TerminalOutcomeType.SUPERSEDED),
+        output_root=tmp_path,
+        expected_frames=1,
+        parity=_parity(),
+    )
+
+    assert sealed.accepted is False
+    assert sealed.report["gates"]["no_superseded_frames"] is False
+    assert sealed.report["gates"]["delivery_count"] is False