Add GenericDeltaSession re-anchor helper (SPEC §10a.8)

Port the producer-side re-anchor session helper from gcf-go: byte-for-byte
identical cadence, passing the shared generic-delta-session conformance
fixtures (fixed-N, size-guard, schema-change).

- GenericDeltaSession with current_full() and next() -> (wire, is_full)
- ReanchorPolicy / fixed_n(n) / size_guard() / DEFAULT_REANCHOR_N = 15
- Size guard compares UTF-8 byte lengths to match Go's len(string)
- Schema change (or fresh key) forces a full re-anchor (10a.7)
- Thin sugar over the primitives; no new wire syntax

Exports the new public symbols from gcf.__init__. Adds a
generic-delta-session conformance branch and a unit suite mirroring the
Go tests (cadence patterns + consumer-stays-in-sync under both policies).

Author: blackwell-systems

CHANGELOG.md

@@ -10,12 +10,14 @@
   - `diff_generic_sets` (the blessed producer path; centralizes the keyed-diff invariants), `encode_generic_full`, `encode_generic_delta`
   - `decode_generic_full`, `decode_generic_delta` (consumer wire parsing)
   - `verify_generic_delta` (atomic apply + `new_root` verification)
+  - Re-anchor session helper (SPEC §10a.8): `GenericDeltaSession` (`current_full`, `next`) with `ReanchorPolicy` / `fixed_n(n)` / `size_guard()` cadence policies and `DEFAULT_REANCHOR_N = 15`. Producer-side sugar over the primitives; introduces no new wire syntax (every emission is exactly `encode_generic_full` / `encode_generic_delta` output). Re-anchor cadence is byte-for-byte identical to `gcf-go` (size guard uses UTF-8 byte length to match Go's `len(string)`), verified by the shared `generic-delta-session` conformance fixtures.
 - Delta is opt-in and bilateral; the existing `encode_generic` path is unchanged (backward compatible).
 
 ### Tests
 
 - Unit suite mirroring `gcf-go`: self-proving round-trip (diff -> encode -> apply -> recomputed root), determinism / row-order invariance, no-type-collision canonicalization, every invariant/error path, full-payload wire round-trip, the complete server -> wire -> consumer end-to-end loop, and malformed-wire-fails-closed.
 - Conformance runner support for `generic-pack-root`, `generic-delta`, `generic-delta-verify`, `generic-delta-decode` (12 shared fixtures); verified to produce identical pack roots and delta wire to `gcf-go`.
+- Session helper suite (`test_generic_delta_session.py`) mirroring `gcf-go`: FixedN cadence pattern, size-guard triggering, schema-change forced full, FixedN(15)-over-30-turns count, and the load-bearing consumer-stays-in-sync check under both policies. Conformance runner support for `generic-delta-session` (3 shared fixtures: fixed-N, size-guard, schema-change).
 
 ## v2.2.2 (2026-07-10)
 

src/gcf/__init__.py

@@ -49,6 +49,12 @@
     decode_generic_full,
     decode_generic_delta,
     verify_generic_delta,
+    GenericDeltaSession,
+    ReanchorPolicy,
+    ReanchorMode,
+    fixed_n,
+    size_guard,
+    DEFAULT_REANCHOR_N,
 )
 from .session import Session, encode_with_session
 from .decode_generic import decode_generic
@@ -84,6 +90,12 @@
     "decode_generic_full",
     "decode_generic_delta",
     "verify_generic_delta",
+    "GenericDeltaSession",
+    "ReanchorPolicy",
+    "ReanchorMode",
+    "fixed_n",
+    "size_guard",
+    "DEFAULT_REANCHOR_N",
 ]
 
 __version__ = "2.3.0"

src/gcf/generic_delta.py

@@ -9,6 +9,7 @@
 
 import hashlib
 from dataclasses import dataclass, field
+from enum import Enum
 from typing import Any
 
 from .scalar import (
@@ -362,3 +363,125 @@ def decode_generic_delta(text: str) -> GenericDeltaPayload:
         else:
             raise ValueError(f"delta_invalid: unknown delta section {name!r}")
     return d
+
+
+# --- producer-side re-anchor session helper (SPEC Section 10a.8) ---
+#
+# GenericDeltaSession manages the re-anchor cadence for a stream of
+# generic-profile updates. It is thin sugar over the primitives: each ``next``
+# emits either a compact delta or, on its chosen cadence, a full re-anchor,
+# updating its held base. It introduces NO new wire syntax: every payload it
+# emits is exactly what ``encode_generic_full`` or ``encode_generic_delta``
+# produce, and the decoder accepts them cadence-agnostically. N and the size
+# guard are the helper's knobs; they are never wire fields. (Section 10a.8 is
+# non-normative producer policy.)
+
+
+# The working default cadence for FixedN (SPEC Section 10a.8).
+DEFAULT_REANCHOR_N = 15
+
+
+class ReanchorMode(Enum):
+    """Selects the session's cadence policy."""
+
+    FIXED_N = 0  # re-anchor every N turns
+    SIZE_GUARD = 1  # re-anchor once cumulative delta reaches the full payload's size
+
+
+@dataclass
+class ReanchorPolicy:
+    """Selects when a :class:`GenericDeltaSession` re-anchors.
+
+    Construct it with :func:`fixed_n` or :func:`size_guard`.
+    """
+
+    mode: ReanchorMode = ReanchorMode.FIXED_N
+    n: int = 0  # turns between anchors; FixedN only
+
+
+def fixed_n(n: int) -> ReanchorPolicy:
+    """Re-anchor every ``n`` turns. ``n <= 0`` falls back to ``DEFAULT_REANCHOR_N``."""
+    if n <= 0:
+        n = DEFAULT_REANCHOR_N
+    return ReanchorPolicy(mode=ReanchorMode.FIXED_N, n=n)
+
+
+def size_guard() -> ReanchorPolicy:
+    """Re-anchor once the cumulative delta bytes since the last anchor reach the
+    current full payload's byte size: it re-anchors more under heavy churn, rarely
+    under light churn, and bounds the delta spent between anchors to about one full
+    payload. Production-recommended.
+    """
+    return ReanchorPolicy(mode=ReanchorMode.SIZE_GUARD)
+
+
+def _byte_len(s: str) -> int:
+    """UTF-8 byte length, matching Go's ``len(string)`` (bytes, not code points)."""
+    return len(s.encode("utf-8"))
+
+
+class GenericDeltaSession:
+    """Holds the current base and re-anchor state for a producer loop.
+
+    Not safe for concurrent use. Call :meth:`current_full` to get the initial
+    full payload to transmit, then :meth:`next` for each subsequent state.
+    """
+
+    def __init__(self, base: GenericSet, tool: str, policy: ReanchorPolicy) -> None:
+        if policy.mode == ReanchorMode.FIXED_N and policy.n <= 0:
+            policy = ReanchorPolicy(mode=policy.mode, n=DEFAULT_REANCHOR_N)
+        self._base = base
+        self._tool = tool
+        self._policy = policy
+        self._turn = 0
+        self._cum = 0  # cumulative delta bytes since the last anchor
+
+    @property
+    def turn(self) -> int:
+        """The number of :meth:`next` calls so far (the initial full is turn 0)."""
+        return self._turn
+
+    def current_full(self) -> str:
+        """Return the full payload for the current base (:func:`encode_generic_full`).
+
+        Send this first to establish the base; it is also a valid manual re-anchor.
+        """
+        return encode_generic_full(self._base, self._tool)
+
+    def _reanchor(self, nxt: GenericSet) -> str:
+        wire = encode_generic_full(nxt, self._tool)
+        self._base = nxt
+        self._cum = 0
+        return wire
+
+    def next(self, nxt: GenericSet) -> tuple[str, bool]:
+        """Advance the session by one turn to ``nxt``.
+
+        Returns ``(wire, is_full)``: the wire to transmit and whether it is a full
+        re-anchor (``True``) or a delta (``False``). A schema change forces a full
+        (Section 10a.7). The held base becomes ``nxt`` either way. The wire is
+        byte-identical to calling ``encode_generic_full`` / ``encode_generic_delta``
+        directly. May raise on duplicate identity / no key (propagated from
+        :func:`diff_generic_sets`).
+        """
+        self._turn += 1
+
+        # Schema change (or a fresh key) cannot be expressed as a delta -> full.
+        if nxt.key != self._base.key or list(self._base.fields) != list(nxt.fields):
+            return self._reanchor(nxt), True
+
+        d = diff_generic_sets(self._base, nxt)
+        delta_wire = encode_generic_delta(d)
+
+        if self._policy.mode == ReanchorMode.SIZE_GUARD:
+            reanchor = self._cum + _byte_len(delta_wire) >= _byte_len(
+                encode_generic_full(nxt, self._tool)
+            )
+        else:  # FIXED_N
+            reanchor = self._turn % self._policy.n == 0
+
+        if reanchor:
+            return self._reanchor(nxt), True
+        self._base = nxt
+        self._cum += _byte_len(delta_wire)
+        return delta_wire, False

tests/test_conformance_v2.py

@@ -164,5 +164,39 @@ def test_conformance(rel_path, data):
             res = verify_generic_delta(base, decode_generic_delta(inp["wire"]), inp["expectedNewRoot"])
             assert generic_pack_root(res) == data["expected"]
 
+    elif op == "generic-delta-session":
+        from gcf.generic_delta import (
+            GenericDeltaSession,
+            GenericSet,
+            fixed_n,
+            size_guard,
+        )
+
+        inp = data["input"]
+
+        def _set(s):
+            return GenericSet(
+                key=s["key"], fields=s["fields"], rows=s["rows"], name=s.get("name", "rows")
+            )
+
+        pol = inp["policy"]
+        if pol.get("mode") == "sizeGuard":
+            policy = size_guard()
+        else:
+            policy = fixed_n(pol.get("n", 0))
+
+        sess = GenericDeltaSession(_set(inp["base"]), inp.get("tool", ""), policy)
+        exp = data["expected"]
+        assert sess.current_full() == exp["initialFull"], (
+            f"initial full mismatch:\n  got: {sess.current_full()!r}\n  exp: {exp['initialFull']!r}"
+        )
+        for i, up in enumerate(inp["updates"]):
+            wire, is_full = sess.next(_set(up))
+            e = exp["emissions"][i]
+            assert is_full == e["isFull"], f"turn {i + 1}: isFull={is_full}, want {e['isFull']}"
+            assert wire == e["wire"], (
+                f"turn {i + 1} wire mismatch:\n  got: {wire!r}\n  exp: {e['wire']!r}"
+            )
+
     else:
         pytest.skip(f"unknown operation: {op}")