story: DGR-036 Prove dense fixture and real-model range parity

tests/test_native_shard_worker.py

@@ -316,6 +316,72 @@ def test_decode_step_is_served(worker):
     assert echoed.chunk.bundle.tensors[0].fragments[0].payload == payload
 
 
+def test_two_disjoint_fake_worker_processes_preserve_prefill_and_decode_seam():
+    """DGR-036 fixture proof across two actual fake-worker processes.
+
+    The DGR-033 worker deliberately has no model graph: its bounded forward
+    validates and echoes the activation bytes.  That makes it suitable for a
+    deterministic protocol proof only.  Keep the two ranges disjoint at the
+    ``SessionOpen`` boundary, pass each stage-one output through stage two,
+    and exercise a prefill plus 32 sequential decode positions.  Numerical
+    dense-GGUF parity remains an opt-in DGR-036 real-model lane, not a claim
+    made by this fixture.
+    """
+    head = _Worker()
+    tail = _Worker()
+    try:
+        head_open = _open(
+            route_session_id="dgr036-head",
+            shard_range=pb.ShardRange(start_layer=0, end_layer=16, effective_start_layer=0),
+        )
+        tail_open = _open(
+            route_session_id="dgr036-tail",
+            shard_range=pb.ShardRange(start_layer=16, end_layer=32, effective_start_layer=16),
+        )
+        payload = b"dgr036 deterministic dense prefill residual"
+
+        def decode_requests(*, stage: str, payloads: list[bytes]) -> list[pb.SessionRequest]:
+            requests: list[pb.SessionRequest] = []
+            for position, stage_payload in enumerate(payloads, start=1):
+                # The fixture worker's negotiated default window is 16.  Top
+                # it up before decode 16 and 32 so this exercises all 32
+                # sequential positions rather than silently testing only one
+                # credit window.
+                if position in {16, 32}:
+                    requests.append(pb.SessionRequest(flow_control=pb.FlowControl(credits_granted=16)))
+                requests.append(_decode(f"decode-{stage}-{position}", stage_payload, position + 1, position))
+            return requests
+
+        head_responses = head.session(
+            [head_open, _chunk("prefill-head", payload, step=1)]
+            + decode_requests(stage="head", payloads=[payload] * 32)
+        )
+        assert head_responses[0].WhichOneof("kind") == "accepted"
+        assert head_responses[1].WhichOneof("kind") == "chunk"
+        seam_payloads = [
+            response.chunk.bundle.tensors[0].fragments[0].payload
+            for response in head_responses
+            if response.WhichOneof("kind") == "chunk"
+        ]
+        assert len(seam_payloads) == 33
+
+        tail_responses = tail.session(
+            [tail_open, _chunk("prefill-tail", seam_payloads[0], step=1, route_session_id="dgr036-tail")]
+            + decode_requests(stage="tail", payloads=seam_payloads[1:])
+        )
+        assert tail_responses[0].WhichOneof("kind") == "accepted"
+        assert tail_responses[1].WhichOneof("kind") == "chunk"
+        assert tail_responses[1].chunk.bundle.tensors[0].fragments[0].payload == payload
+
+        for tail_response in (response for response in tail_responses if response.WhichOneof("kind") == "chunk"):
+            assert tail_response.chunk.bundle.tensors[0].fragments[0].payload == payload
+    finally:
+        if head.proc.poll() is None:
+            head.close()
+        if tail.proc.poll() is None:
+            tail.close()
+
+
 def test_release_is_terminal(worker):
     responses = worker.session([_open(), _release()])
     assert responses[0].WhichOneof("kind") == "accepted"