feat: graph delta wire decode + verify (decode_delta)

Add decode_delta to parse a `GCF profile=graph delta=true` wire back
into removed/added symbols and edge changes, and wire it to
verify_delta for atomic apply against a base snapshot plus pack_root
recomputation, rejecting a wrong new_root with root_mismatch
(SPEC 10.4). encode_delta's ## added line now carries the trailing
distance field (SPEC 3.4.1, Section 10.1). The conformance runner
exercises graph-delta 001/002/003 end to end.

Also refresh the hand-written unit tests to the current encoder
behavior: the delta ## added lines carry distance; the graph header
omits zero-valued budget/tokens/edges; edges decode in canonical
(source, target, type) order rather than input order (compared as a
set); and new session symbols take the next stable session-global ID.

Author: blackwell-systems

CHANGELOG.md

@@ -4,7 +4,8 @@
 
 ### Fixes
 
-- The conformance runner now hard-fails on any unhandled operation (instead of silently skipping it) and exercises session, delta-encode, roundtrip, and pack-root fixtures; only `delta-verify` is explicitly allow-listed, pending a graph delta wire decoder.
+- The conformance runner now hard-fails on any unhandled operation (instead of silently skipping it) and exercises session, delta, roundtrip, and pack-root fixtures end to end; the graph delta wire decode and verify path is now covered, so no operations remain allow-listed.
+- Implemented the graph delta wire decoder and verifier (`decode_delta` / `verify_delta`): parse a `GCF profile=graph delta=true` wire back into removed/added symbols and edge changes, apply them atomically to a base snapshot, recompute `pack_root`, and reject a wrong `new_root` with `root_mismatch` (SPEC 10.4). The `## added` encoder now emits the trailing `distance` field (SPEC 3.4.1, Section 10.1). The shared `graph-delta` fixtures now run end to end: 001 (encode, gains the trailing distance), 002 (verified apply), 003 (`root_mismatch` rejection).
 - **Session encoding correctness fix.** `encode_with_session` assigned per-response local IDs instead of stable session-global IDs, so the cross-call dedup references (`@N  # previously transmitted`) pointed at the wrong symbols, and the header emitted zero-valued `budget`/`tokens`/`edges`. Both are fixed to match the reference; graph session output is now byte-identical across all six SDKs. This had gone undetected because the conformance runner skipped the shared graph-session fixtures (now wired).
 - Added the graph-profile PackRoot (`pack_root(symbols, edges)`, gcf-pack-root-v1, SPEC 10.2): the content-addressed sha256 over canonical, independently-sorted symbol/edge records, byte-identical to gcf-go/rust/typescript/swift/kotlin. The conformance runner now exercises the shared `graph-pack-root` fixtures, which it had been skipping (so this primitive was previously unimplemented and untested).
 - Buffered graph encoder now matches the reference byte-for-byte: symbols are ordered by distance then descending score with local IDs assigned in output order, and the header omits `budget`/`tokens`/`edges` when zero (previously symbols kept input order and zero-valued header fields were always emitted). The conformance runner now exercises the shared `graph-encode` fixtures (001-003), which it had been skipping - which is how this divergence went uncaught.

src/gcf/__init__.py

@@ -36,7 +36,7 @@
 
 from .constants import KIND_ABBREV, KIND_EXPAND
 from .decode import DecodeError, decode
-from .delta import encode_delta
+from .delta import decode_delta, encode_delta, verify_delta
 from .encode import encode
 from .generic import encode_generic, GenericOptions
 from .generic_delta import (
@@ -76,9 +76,11 @@
     "StreamEncoder",
     "Symbol",
     "decode",
+    "decode_delta",
     "decode_generic",
     "encode",
     "encode_delta",
+    "verify_delta",
     "encode_generic",
     "GenericOptions",
     "encode_with_session",

src/gcf/delta.py

@@ -1,7 +1,8 @@
 """GCF delta encoding: only added/removed symbols for incremental delivery."""
 
-from .constants import KIND_ABBREV
-from .types import DeltaPayload
+from .constants import KIND_ABBREV, KIND_EXPAND
+from .packroot import pack_root
+from .types import DeltaPayload, Edge, Symbol
 
 
 def encode_delta(d: DeltaPayload) -> str:
@@ -37,7 +38,9 @@ def encode_delta(d: DeltaPayload) -> str:
         parts.append("## added")
         for i, s in enumerate(d.added):
             kind = KIND_ABBREV.get(s.kind, s.kind)
-            parts.append(f"@{i} {kind} {s.qualified_name} {s.score:.2f} {s.provenance}")
+            parts.append(
+                f"@{i} {kind} {s.qualified_name} {s.score:.2f} {s.provenance} {s.distance}"
+            )
 
     # Removed edges.
     if d.removed_edges:
@@ -52,3 +55,183 @@ def encode_delta(d: DeltaPayload) -> str:
             parts.append(f"{e.source} -> {e.target} {e.edge_type}")
 
     return "\n".join(parts) + "\n"
+
+
+def _expand_kind(k: str) -> str:
+    """Reverse a kind abbreviation to its full form (identity if unknown)."""
+    return KIND_EXPAND.get(k, k)
+
+
+def _parse_delta_edge(line: str) -> Edge:
+    """Parse a `source -> target type` delta edge line."""
+    idx = line.find(" -> ")
+    if idx < 0:
+        raise ValueError(f"malformed_delta: edge line missing ' -> ': {line!r}")
+    source = line[:idx]
+    rest = line[idx + 4 :].split()
+    if len(rest) != 2:
+        raise ValueError(
+            f"malformed_delta: edge line {line!r} must be 'source -> target type'"
+        )
+    return Edge(source=source, target=rest[0], edge_type=rest[1])
+
+
+def decode_delta(wire: str) -> DeltaPayload:
+    """Parse a GCF graph delta wire payload back into a DeltaPayload.
+
+    Kind abbreviations on removed/added lines are expanded to their full form so the
+    result matches a base snapshot's symbol identities. Raises ValueError containing
+    ``malformed_delta`` on bad lines or unknown sections.
+    """
+    lines = wire.rstrip("\n").split("\n")
+    if not lines or lines[0] == "":
+        raise ValueError("missing_header: empty delta payload")
+    header = lines[0].rstrip("\r")
+    if not header.startswith("GCF profile=graph"):
+        raise ValueError(
+            "missing_profile: delta header must begin with 'GCF profile=graph'"
+        )
+
+    d = DeltaPayload()
+    for field in header.split():
+        kv = field.split("=", 1)
+        if len(kv) != 2:
+            continue
+        key, value = kv
+        if key == "tool":
+            d.tool = value
+        elif key == "base_root":
+            d.base_root = value
+        elif key == "new_root":
+            d.new_root = value
+
+    section = ""
+    for raw in lines[1:]:
+        line = raw.rstrip("\r")
+        if line == "":
+            continue
+        if line.startswith("## "):
+            section = line[3:].strip()
+            if section not in ("removed", "added", "edges_removed", "edges_added"):
+                raise ValueError(f"malformed_delta: unknown section {section!r}")
+            continue
+        if section == "removed":
+            parts = line.split()
+            if len(parts) != 2:
+                raise ValueError(
+                    f"malformed_delta: removed line {line!r} must be 'kind qname'"
+                )
+            d.removed.append(
+                Symbol(kind=_expand_kind(parts[0]), qualified_name=parts[1])
+            )
+        elif section == "added":
+            parts = line.split()
+            if len(parts) != 6:
+                raise ValueError(
+                    f"malformed_delta: added line {line!r} must be "
+                    "'@id kind qname score provenance distance'"
+                )
+            try:
+                score = float(parts[3])
+            except ValueError:
+                raise ValueError(f"malformed_delta: invalid added score {parts[3]!r}")
+            try:
+                dist = int(parts[5])
+            except ValueError:
+                raise ValueError(
+                    f"malformed_delta: invalid added distance {parts[5]!r}"
+                )
+            d.added.append(
+                Symbol(
+                    kind=_expand_kind(parts[1]),
+                    qualified_name=parts[2],
+                    score=score,
+                    provenance=parts[4],
+                    distance=dist,
+                )
+            )
+        elif section in ("edges_removed", "edges_added"):
+            e = _parse_delta_edge(line)
+            if section == "edges_removed":
+                d.removed_edges.append(e)
+            else:
+                d.added_edges.append(e)
+        else:
+            raise ValueError(
+                f"malformed_delta: data line {line!r} before any section header"
+            )
+    return d
+
+
+def verify_delta(
+    base_symbols: list[Symbol],
+    base_edges: list[Edge],
+    removed: list[Symbol],
+    added: list[Symbol],
+    removed_edges: list[Edge],
+    added_edges: list[Edge],
+    expected_new_root: str,
+) -> tuple[list[Symbol], list[Edge]]:
+    """Apply a delta to a base snapshot and verify the resulting pack root.
+
+    Symbols are matched by identity ``(kind, qualified_name)``; edges by
+    ``(source, target, edge_type)``. Raises ValueError containing ``delta_invalid``
+    when removing a symbol/edge that does not exist or adding one that already exists,
+    and ``root_mismatch`` when the recomputed pack root differs from
+    ``expected_new_root``. On success returns the applied ``(symbols, edges)``.
+    """
+    sym_map: dict[tuple[str, str], Symbol] = {}
+    for s in base_symbols:
+        sym_map[(s.kind, s.qualified_name)] = s
+
+    for s in removed:
+        key = (s.kind, s.qualified_name)
+        if key not in sym_map:
+            raise ValueError(
+                f"delta_invalid: removing symbol {s.kind} {s.qualified_name} "
+                "that does not exist in base"
+            )
+        del sym_map[key]
+
+    for s in added:
+        key = (s.kind, s.qualified_name)
+        if key in sym_map:
+            raise ValueError(
+                f"delta_invalid: adding symbol {s.kind} {s.qualified_name} "
+                "that already exists"
+            )
+        sym_map[key] = s
+
+    result_symbols = list(sym_map.values())
+
+    edge_map: dict[tuple[str, str, str], Edge] = {}
+    for e in base_edges:
+        edge_map[(e.source, e.target, e.edge_type)] = e
+
+    for e in removed_edges:
+        key = (e.source, e.target, e.edge_type)
+        if key not in edge_map:
+            raise ValueError(
+                f"delta_invalid: removing edge {e.source} -> {e.target} "
+                f"{e.edge_type} that does not exist"
+            )
+        del edge_map[key]
+
+    for e in added_edges:
+        key = (e.source, e.target, e.edge_type)
+        if key in edge_map:
+            raise ValueError(
+                f"delta_invalid: adding edge {e.source} -> {e.target} "
+                f"{e.edge_type} that already exists"
+            )
+        edge_map[key] = e
+
+    result_edges = list(edge_map.values())
+
+    computed_root = pack_root(result_symbols, result_edges)
+    if computed_root != expected_new_root:
+        raise ValueError(
+            f"root_mismatch: computed {computed_root}, expected {expected_new_root}"
+        )
+
+    return result_symbols, result_edges

tests/test_conformance_v2.py

@@ -343,16 +343,13 @@ def _set(s):
                 f"  got: {got!r}\n  exp: {call['expected']!r}"
             )
 
-    elif op == "delta":
-        # graph-delta fixtures share the `delta` operation but come in two shapes:
-        #   - encode: `input` is a structured object -> byte-compare encode_delta.
-        #   - verify: `input` is a WIRE STRING (with a base_snapshot) -> the wire
-        #     decoder + verifier is not yet implemented, so skip like delta-verify.
-        inp = data["input"]
-        if isinstance(inp, str) or "base_snapshot" in data:
-            pytest.skip("graph delta wire decoder not yet implemented")
-
-        from gcf.delta import encode_delta
+    elif op in ("delta", "delta-verify"):
+        # graph-delta fixtures come in two shapes:
+        #   - encode (`delta`, object input): byte-compare encode_delta.
+        #   - verify (`delta` with string input + base_snapshot, or `delta-verify`):
+        #     decode the wire, apply + verify against the base_snapshot pack_root.
+        from gcf.delta import decode_delta, encode_delta, verify_delta
+        from gcf.packroot import pack_root
         from gcf.types import DeltaPayload, Edge, Symbol
 
         def _sym(s):
@@ -367,24 +364,57 @@ def _sym(s):
         def _edge(e):
             return Edge(source=e["source"], target=e["target"], edge_type=e["edgeType"])
 
-        d = DeltaPayload(
-            tool=inp.get("tool", ""),
-            base_root=inp["baseRoot"],
-            new_root=inp["newRoot"],
-            removed=[_sym(s) for s in inp.get("removed", [])],
-            added=[_sym(s) for s in inp.get("added", [])],
-            removed_edges=[_edge(e) for e in inp.get("removedEdges", [])],
-            added_edges=[_edge(e) for e in inp.get("addedEdges", [])],
-            delta_tokens=inp.get("deltaTokens", 0),
-            full_tokens=inp.get("fullTokens", 0),
-        )
-        got = encode_delta(d)
-        assert got == data["expected"], (
-            f"delta encode mismatch:\n  got: {got!r}\n  exp: {data['expected']!r}"
-        )
+        inp = data["input"]
+
+        if op == "delta" and not isinstance(inp, str) and "base_snapshot" not in data:
+            # Structured-object encode.
+            d = DeltaPayload(
+                tool=inp.get("tool", ""),
+                base_root=inp["baseRoot"],
+                new_root=inp["newRoot"],
+                removed=[_sym(s) for s in inp.get("removed", [])],
+                added=[_sym(s) for s in inp.get("added", [])],
+                removed_edges=[_edge(e) for e in inp.get("removedEdges", [])],
+                added_edges=[_edge(e) for e in inp.get("addedEdges", [])],
+                delta_tokens=inp.get("deltaTokens", 0),
+                full_tokens=inp.get("fullTokens", 0),
+            )
+            got = encode_delta(d)
+            assert got == data["expected"], (
+                f"delta encode mismatch:\n  got: {got!r}\n  exp: {data['expected']!r}"
+            )
+            return
+
+        # Wire-string decode + verify.
+        snap = data["base_snapshot"]
+        base_symbols = [_sym(s) for s in snap.get("symbols", [])]
+        base_edges = [_edge(e) for e in snap.get("edges", [])]
+        d = decode_delta(inp)
 
-    elif op == "delta-verify":
-        pytest.skip("graph delta wire decoder not yet implemented")
+        if data.get("expectedError"):
+            with pytest.raises(Exception) as ei:
+                verify_delta(
+                    base_symbols, base_edges,
+                    d.removed, d.added, d.removed_edges, d.added_edges,
+                    d.new_root,
+                )
+            assert data["expectedError"] in str(ei.value), (
+                f"expected error containing {data['expectedError']!r}, got {ei.value!r}"
+            )
+        else:
+            got_symbols, got_edges = verify_delta(
+                base_symbols, base_edges,
+                d.removed, d.added, d.removed_edges, d.added_edges,
+                d.new_root,
+            )
+            exp = data["expected_snapshot"]
+            exp_symbols = [_sym(s) for s in exp.get("symbols", [])]
+            exp_edges = [_edge(e) for e in exp.get("edges", [])]
+            assert pack_root(got_symbols, got_edges) == pack_root(exp_symbols, exp_edges), (
+                "delta verify pack_root mismatch:\n"
+                f"  got: {pack_root(got_symbols, got_edges)}\n"
+                f"  exp: {pack_root(exp_symbols, exp_edges)}"
+            )
 
     else:
         pytest.fail(

tests/test_delta.py

@@ -35,7 +35,7 @@ def test_encode_delta_basic():
     assert "## removed\n" in output
     assert "fn pkg.OldHandler\n" in output
     assert "## added\n" in output
-    assert "@0 fn pkg.NewHandler 0.85 rwr\n" in output
+    assert "@0 fn pkg.NewHandler 0.85 rwr 0\n" in output
     assert "## edges_removed\n" in output
     assert "pkg.Router -> pkg.OldHandler calls\n" in output
     assert "## edges_added\n" in output
@@ -105,7 +105,7 @@ def test_encode_delta_only_added():
     )
     output = encode_delta(d)
     assert "## added\n" in output
-    assert "@0 class pkg.New 0.75 lsp_resolved\n" in output
+    assert "@0 class pkg.New 0.75 lsp_resolved 0\n" in output
     assert "## removed" not in output
 
 
@@ -124,9 +124,9 @@ def test_encode_delta_multiple_added_sequential_ids():
         full_tokens=200,
     )
     output = encode_delta(d)
-    assert "@0 fn pkg.A 0.90 x\n" in output
-    assert "@1 fn pkg.B 0.80 x\n" in output
-    assert "@2 fn pkg.C 0.70 x\n" in output
+    assert "@0 fn pkg.A 0.90 x 0\n" in output
+    assert "@1 fn pkg.B 0.80 x 0\n" in output
+    assert "@2 fn pkg.C 0.70 x 0\n" in output
 
 
 def test_encode_delta_kind_abbreviation():

tests/test_encode.py

@@ -176,7 +176,8 @@ def test_encode_skips_edges_with_missing_symbols():
     output = encode(p)
     # Section header is emitted (matches Go), but no edge lines beneath it
     assert "## edges [0]" in output
-    assert "edges=0" in output
+    # The zero edge count is omitted from the header (matches Go)
+    assert "edges=0" not in output
     lines_after_edges = output.split("## edges [0]\n")[1]
     assert lines_after_edges.strip() == ""
 
@@ -185,4 +186,4 @@ def test_encode_empty_payload():
     """Empty payload produces only header."""
     p = Payload(tool="test", token_budget=100, tokens_used=0)
     output = encode(p)
-    assert output == "GCF profile=graph tool=test budget=100 tokens=0 symbols=0 edges=0\n"
+    assert output == "GCF profile=graph tool=test budget=100 symbols=0\n"

tests/test_roundtrip.py

@@ -119,11 +119,11 @@ def test_roundtrip_multiple_edges():
 
     decoded = decode(encode(original))
     assert len(decoded.edges) == 3
-    # Verify edges exist (order preserved).
-    for orig, dec in zip(original.edges, decoded.edges):
-        assert dec.source == orig.source
-        assert dec.target == orig.target
-        assert dec.edge_type == orig.edge_type
+    # The encoder orders edges by source ID, then target ID, then edge type
+    # (SPEC 16.1), so compare as a set rather than by input order.
+    orig_set = {(e.source, e.target, e.edge_type) for e in original.edges}
+    dec_set = {(e.source, e.target, e.edge_type) for e in decoded.edges}
+    assert dec_set == orig_set
 
 
 def test_roundtrip_empty_payload():