pgw#628: residency reporting v2 — content-addressed idempotent observations (0.52.0)

CHANGELOG.md

@@ -1,5 +1,22 @@
 # Changelog
 
+## 0.52.0 (2026-07-24)
+
+- **pgw#628: residency reporting v2 — content-addressed idempotent
+  observations (worker half of th#1070).** Every applied HelloAck opens a
+  republish epoch: the reconcile pass re-announces verified cached
+  identities (ON_DISK/IN_RAM/IN_VRAM with exact ref+digest) once per epoch
+  even when unchanged — a re-received plan (hub redrive, overdue resend,
+  reconnect) is the hub asking for a resync, and success observations are
+  now safe to emit late, twice, or across plan revisions (the v2 hub
+  accepts them by digest, generations survive only for cancel/evict/
+  failure attribution). Job-path cache hits within one epoch stay deduped
+  (no event spam). gw#614's no-cancel-on-same-set reconcile behavior is
+  unchanged — under the v2 hub it is simply correct instead of a trap.
+  Version floor note: endpoint images need no forced rebuild — v2 hubs
+  accept 0.44–0.51 success reports fine; images pick up >=0.52.0 on their
+  next routine rebuild for the lost-observation resync hardening.
+
 ## 0.51.0 (2026-07-24)
 
 - **gw#627: Conv2d additive-branch support in the runtime LoRA overlay.**

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "gen-worker"
-version = "0.51.0"
+version = "0.52.0"
 description = "A library used to build custom functions in Cozy Creator's serverless function platform."
 readme = "README.md"
 license = "MIT"

src/gen_worker/executor.py

@@ -696,6 +696,15 @@ def __init__(
         # pipeline is still in RAM/VRAM. Keep the disk identity separately
         # until record teardown makes it the highest residency tier.
         self._disk_identities: Dict[str, _ResidencyIdentity] = {}
+        # pgw#628 (th#1070 residency v2): every applied HelloAck opens a new
+        # republish epoch. The reconcile pass re-announces verified cached
+        # identities the hub re-asked about even when unchanged — observations
+        # are content-addressed and idempotent hub-side, and a force-resent
+        # plan is exactly the hub saying "tell me again" (redrive/overdue
+        # resends could otherwise never heal a lost success observation).
+        # Job-path ensure_local calls within the same epoch stay deduped.
+        self._residency_republish_epoch = 0
+        self._identity_publish_epochs: Dict[str, int] = {}
         self._identity_lock = threading.RLock()
         # Cold-ref waiters (th#763): ensure_local blocks here until the
         # hub's re-minted DOWNLOAD banks a snapshot for the ref.
@@ -781,6 +790,7 @@ def _on_residency_event(
             with self._identity_lock:
                 self._resident_identities.pop(ref, None)
                 self._disk_identities.pop(ref, None)
+                self._identity_publish_epochs.pop(ref, None)
         try:
             loop = asyncio.get_running_loop()
         except RuntimeError:
@@ -979,6 +989,7 @@ def replace_desired_snapshots(
             for ref, snapshot in stored.items()
         }
         with self._identity_lock:
+            self._residency_republish_epoch += 1
             self._desired_snapshot_identities = desired
             # Generations belong only to the current desired identity. Leave
             # actual resident identity untouched: those bytes may still be in
@@ -1044,8 +1055,15 @@ def activate_disk_identity(self, ref: str) -> _ResidencyIdentity:
     async def _confirm_cached_identity(
         self, ref: str, identity: _ResidencyIdentity,
     ) -> None:
-        """Publish exact identity when verified cached bytes satisfy a newer
-        desired generation without requiring a redundant download."""
+        """Publish exact identity when verified cached bytes satisfy the
+        desired state without requiring a redundant download.
+
+        pgw#628 (th#1070 residency v2): the emission is content-addressed and
+        idempotent hub-side, so it is re-sent once per applied-HelloAck epoch
+        even when the identity is unchanged — a re-received plan (redrive,
+        overdue resend, reconnect) is the hub asking for a resync, and a
+        worker that never re-announces can strand a lost success observation
+        forever. Job-path calls within the same epoch remain deduped."""
         tier = self.residency.tier(ref)
         digest, _ = identity
         if not digest:
@@ -1060,7 +1078,12 @@ async def _confirm_cached_identity(
         # serving the new snapshot.
         if tier in (residency_mod.Tier.RAM, residency_mod.Tier.VRAM) and current[0] != digest:
             return
-        if not self._set_resident_identity(ref, identity):
+        changed = self._set_resident_identity(ref, identity)
+        with self._identity_lock:
+            epoch = self._residency_republish_epoch
+            republish = self._identity_publish_epochs.get(ref) != epoch
+            self._identity_publish_epochs[ref] = epoch
+        if not changed and not republish:
             return
         state = {
             residency_mod.Tier.DISK: pb.MODEL_STATE_ON_DISK,

tests/test_residency_republish_pgw628.py

@@ -0,0 +1,100 @@
+"""pgw#628 (th#1070 residency protocol v2, worker half): success observations
+are (ref, digest, state) — content-addressed, idempotent, and safe to emit
+twice. A re-received desired plan (hub redrive, overdue resend, reconnect) is
+the hub asking for a resync, so the reconcile pass re-announces verified
+cached identities once per applied-HelloAck epoch even when nothing changed.
+Within one epoch the identity dedupe still holds (no event spam). The gw#614
+no-cancel-on-same-set behavior is untouched — under v2 it is simply correct
+instead of a trap.
+"""
+
+from __future__ import annotations
+
+import time
+
+from gen_worker.pb import worker_scheduler_pb2 as pb
+
+from harness.blob_host import BlobHost
+from harness.hub_double import Conn, hub_double, is_model_event, is_ready
+
+_MODEL_REF = "harness/residency-tiny"
+
+
+def _disk_only_ack(snapshot: pb.Snapshot, generation: int) -> pb.HelloAck:
+    return pb.HelloAck(
+        protocol_version=pb.PROTOCOL_VERSION_CURRENT,
+        desired_residency=pb.DesiredResidency(
+            generation=generation,
+            disk_refs=[_MODEL_REF],
+            snapshots={_MODEL_REF: snapshot},
+        ),
+    )
+
+
+def _count_on_disk(conn: Conn) -> int:
+    with conn._recv_cond:
+        return sum(
+            1
+            for m in conn.received
+            if m.WhichOneof("msg") == "model_event"
+            and m.model_event.ref == _MODEL_REF
+            and m.model_event.state == pb.MODEL_STATE_ON_DISK
+        )
+
+
+def _wait_on_disk_count(conn: Conn, want: int, timeout: float = 15.0) -> None:
+    deadline = time.monotonic() + timeout
+    while _count_on_disk(conn) < want:
+        if time.monotonic() > deadline:
+            raise TimeoutError(
+                f"expected {want} ON_DISK re-reports, saw {_count_on_disk(conn)}"
+            )
+        time.sleep(0.05)
+
+
+def test_reissued_plan_republishes_held_identity(tmp_path) -> None:
+    """Each applied plan re-send yields exactly ONE fresh ON_DISK re-report
+    carrying the exact (ref, digest): the idempotent resync a v2 hub can
+    always absorb and that heals a lost success observation. Within an
+    epoch the dedupe holds — re-announce is per applied ack, never spam."""
+    blobs = BlobHost(tmp_path)
+    try:
+        snapshot = blobs.one_file_snapshot("snap-1", "blob", b"tiny-weights")
+        with hub_double() as (scheduler, _harness):
+            conn = scheduler.wait_connection(0)
+            conn.wait_for(is_ready)
+
+            conn.send(hello_ack=_disk_only_ack(snapshot, generation=1))
+            first = conn.wait_for(
+                is_model_event(_MODEL_REF, pb.MODEL_STATE_ON_DISK)
+            ).model_event
+            assert first.snapshot_digest == "snap-1"
+            assert first.residency_generation == 1
+
+            # The hub re-sends the SAME plan (redrive / overdue resend): the
+            # worker must re-announce the held bytes, not stay silent behind
+            # its identity dedupe.
+            baseline = _count_on_disk(conn)
+            conn.send(hello_ack=_disk_only_ack(snapshot, generation=1))
+            _wait_on_disk_count(conn, baseline + 1)
+            events = [
+                m.model_event
+                for m in conn.received
+                if m.WhichOneof("msg") == "model_event"
+                and m.model_event.ref == _MODEL_REF
+                and m.model_event.state == pb.MODEL_STATE_ON_DISK
+            ]
+            assert events[-1].snapshot_digest == "snap-1"
+            assert events[-1].residency_generation == 1
+
+            # And exactly one per applied plan: no runaway re-announce loop.
+            time.sleep(0.5)
+            assert _count_on_disk(conn) == baseline + 1
+
+            # A third re-send opens a third epoch: one more re-report.
+            conn.send(hello_ack=_disk_only_ack(snapshot, generation=1))
+            _wait_on_disk_count(conn, baseline + 2)
+            time.sleep(0.5)
+            assert _count_on_disk(conn) == baseline + 2
+    finally:
+        blobs.shutdown()