feat(dataplane): wrap read_program return in Sourced provenance (LocalValue)

Sourced[T, S] + the six aliases (LocalValue, CachedValue, ReplayedValue,
GossipedValue, ExternalValue, SanitizedValue) were exported with zero
production consumers — the provenance vocabulary was dead code that
the doc §3.4 says should distinguish local-fresh-read vs cached-stale-
read at every call site.

dp.read_program now returns Result[LocalValue[Versioned[ProgramSnapshot]]
| None, DataPlaneError]. The phantom-type wrapper makes the read's
provenance explicit at the type level. Future replay paths would
return ReplayedValue[Versioned[...]]; future cache reads would return
CachedValue[Versioned[...]]; mypy then enforces that a function
demanding LocalValue[Program] cannot accept a CachedValue[Program]
by mistake (bug class #13 closed at the type system level).

Production caller updated: gigaevo/runner/dag_runner.py
_timeout_read_is_fresh unwraps the LocalValue via .value.value to
reach the underlying Versioned.

5 tests updated to unwrap the LocalValue wrapper; 1 new test pins
the phantom-wrapper shape returned by read_program.

This is the foothold: a follow-up provenance audit can
rg "LocalValue\|CachedValue\|ReplayedValue" gigaevo/ and find the
first real consumer; subsequent reads (crdt_read, lwwr_get) can
adopt the same pattern.

Author: GrigoryEvko

gigaevo/dataplane/tests/test_transition_state.py

@@ -337,8 +337,10 @@ async def test_existing_returns_versioned(self, coord: dp.DataPlane) -> None:
         result = await coord.read_program(dp.ProgramId("p-13"))
         assert isinstance(result, dp.Ok)
         assert result.value is not None
-        assert result.value.value["state"] == "RUNNING"
-        assert result.value.epoch >= 1
+        # ``LocalValue`` wraps the freshness-checked ``Versioned``.
+        versioned = result.value.value
+        assert versioned.value["state"] == "RUNNING"
+        assert versioned.epoch >= 1
 
     async def test_stale_floor_raises(self, coord: dp.DataPlane) -> None:
         await _put_program(coord, "p-14", dp.ProgramState.QUEUED)
@@ -351,11 +353,29 @@ async def test_stale_floor_raises(self, coord: dp.DataPlane) -> None:
         current = await coord.read_program(dp.ProgramId("p-14"))
         assert isinstance(current, dp.Ok) and current.value is not None
         future = await coord.read_program(
-            dp.ProgramId("p-14"), min_epoch=current.value.epoch + 10
+            dp.ProgramId("p-14"), min_epoch=current.value.value.epoch + 10
         )
         assert isinstance(future, dp.Err)
         assert isinstance(future.error, dp.StaleReadError)
 
+    async def test_returns_localvalue_phantom_wrapper(
+        self, coord: dp.DataPlane
+    ) -> None:
+        """The successful return is wrapped in :data:`LocalValue` — the
+        :class:`Sourced` phantom-tag alias for a fresh local read. This
+        is the structural discriminator for bug class #13 (stale cache
+        returns as authoritative)."""
+        await _put_program(coord, "p-localvalue", dp.ProgramState.QUEUED)
+        result = await coord.read_program(dp.ProgramId("p-localvalue"))
+        assert isinstance(result, dp.Ok)
+        assert result.value is not None
+        # The outer wrapper is a Sourced instance (the LocalValue alias).
+        assert isinstance(result.value, dp.Sourced)
+        # The inner ``Versioned`` is reachable via ``.value`` and carries
+        # the freshness witness.
+        versioned = result.value.value
+        assert isinstance(versioned, dp.Versioned)
+
 
 # ── token discipline ─────────────────────────────────────────────────
 
@@ -547,4 +567,4 @@ async def test_batch_partial_failure_returns_err_after_partial_commit(
         # The first item's commit survived.
         read = await coord.read_program(dp.ProgramId("b-good"))
         assert isinstance(read, dp.Ok) and read.value is not None
-        assert read.value.value["state"] == "RUNNING"
+        assert read.value.value.value["state"] == "RUNNING"

gigaevo/runner/dag_runner.py

@@ -283,8 +283,13 @@ async def _timeout_read_is_fresh(self, prog: Program) -> bool:
             return False
         # ``Ok(None)`` means the blob is gone (someone deleted it).
         # That is not the freshness contract we wanted to assert; treat
-        # as "not fresh enough to act on" and defer.
-        return result.value is not None
+        # as "not fresh enough to act on" and defer. Otherwise the
+        # coordinator returns a :data:`LocalValue` (the phantom-tag
+        # alias for a fresh local read) wrapping the
+        # :class:`Versioned` admission record; we only need its
+        # presence here, not the inner value.
+        local_value = result.value
+        return local_value is not None
 
     async def _run(self) -> None:
         logger.info("[DagScheduler] start")

tests/database/test_redis_storage_dataplane.py

@@ -652,7 +652,8 @@ async def test_eventual_returns_value_at_any_epoch(
         )
         assert isinstance(result, Ok)
         assert result.value is not None
-        assert result.value.value["state"] == ProgramState.RUNNING.value
+        # ``LocalValue`` wraps the freshness-checked ``Versioned``.
+        assert result.value.value.value["state"] == ProgramState.RUNNING.value
 
     async def test_at_least_below_floor_returns_stale_read_error(
         self,
@@ -684,7 +685,7 @@ async def test_at_least_below_floor_returns_stale_read_error(
         )
         assert isinstance(eventual, Ok)
         assert eventual.value is not None
-        observed_epoch = eventual.value.epoch
+        observed_epoch = eventual.value.value.epoch
 
         # Asking for one beyond the observed epoch fires the stale
         # branch. The same call at FreshnessEventual succeeded above —
@@ -741,7 +742,7 @@ async def test_legacy_min_epoch_kwarg_still_works(
         # First read picks up the current epoch.
         result = await coord.read_program(ProgramId(prog.id))
         assert isinstance(result, Ok)
-        observed_epoch = result.value.epoch  # type: ignore[union-attr]
+        observed_epoch = result.value.value.epoch  # type: ignore[union-attr]
         stale = await coord.read_program(
             ProgramId(prog.id), min_epoch=observed_epoch + 1
         )