stream: labeled_trailer_counts option (SPEC 8.4.1)

Opt-in labeled trailer counts for the graph streaming encoder: labeled_trailer_counts=True
emits the '##! summary' counts field as label:count (counts=targets:1,related:1,edges:1)
instead of the positional default. Conformance runner reads options.labeledTrailerCounts
and the skip-guard now treats labeledTrailerCounts as supported, so fixture 005 runs and
passes (default output unchanged; 004 positional still passes).

Author: blackwell-systems

src/gcf/stream.py

@@ -38,8 +38,10 @@ def __init__(
         tokens_used: int = 0,
         pack_root: str = "",
         session: bool = False,
+        labeled_trailer_counts: bool = False,
     ) -> None:
         self._w = writer
+        self._labeled_trailer_counts = labeled_trailer_counts
         self._lock = threading.Lock()
         self._sym_index: dict[str, int] = {}
         self._next_id = 0
@@ -122,22 +124,29 @@ def write_bare_ref(self, qname: str, distance: int) -> None:
     def close(self) -> None:
         """Emit ##! summary trailer with final counts."""
         with self._lock:
-            counts: list[str] = []
+            # Build label:count sections, then either emit as-is (labeled form,
+            # SPEC 8.4.1) or strip to values (default positional form).
+            sections: list[str] = []
             group_order = ["targets", "related", "extended"]
 
             for g in group_order:
                 c = self._group_counts.get(g, 0)
                 if c > 0:
-                    counts.append(str(c))
+                    sections.append(f"{g}:{c}")
             for g, c in self._group_counts.items():
                 if g not in group_order and c > 0:
-                    counts.append(str(c))
+                    sections.append(f"{g}:{c}")
             if self._edge_count > 0:
-                counts.append(str(self._edge_count))
+                sections.append(f"edges:{self._edge_count}")
+
+            if self._labeled_trailer_counts:
+                counts_str = ",".join(sections)
+            else:
+                counts_str = ",".join(s.split(":", 1)[1] for s in sections)
 
             self._w.write(
                 f"##! summary symbols={self._next_id} edges={self._edge_count}"
-                f" counts={','.join(counts)}\n"
+                f" counts={counts_str}\n"
             )
 
     @property

tests/test_conformance_v2.py

@@ -199,9 +199,10 @@ def _set(s):
             )
 
     elif op == "graph-stream-encode":
-        # Skip a fixture requesting stream options this runner does not support
-        # (e.g. labeledTrailerCounts, SPEC 8.4.1). This runner supports none.
-        if data.get("options"):
+        # labeledTrailerCounts (SPEC 8.4.1) is supported; skip a fixture only if
+        # it requests some OTHER stream option this runner does not support.
+        options = data.get("options", {})
+        if any(k != "labeledTrailerCounts" for k in options):
             pytest.skip("unsupported stream options")
         import io
 
@@ -216,6 +217,7 @@ def _set(s):
             token_budget=inp.get("tokenBudget", 0),
             tokens_used=inp.get("tokensUsed", 0),
             pack_root=inp.get("packRoot", ""),
+            labeled_trailer_counts=options.get("labeledTrailerCounts", False),
         )
         for s in inp.get("symbols", []):
             enc.write_symbol(