feat(telemetry): ADR-0078 Phase 1 — composer/graph atom equivalence (observational)

Wires observational telemetry on the composer-vs-graph atom-set
relationship.  Phase 1 is strictly observational: no enforcement,
no surface mutation, no grounding-source change, no trace-hash impact.

New telemetry fields on TurnEvent + ChatResponse:
  composer_graph_atom_status         ∈ {equivalent, divergent,
                                         graph_unconstrained,
                                         composer_no_atoms,
                                         not_applicable, ""}
  composer_atom_set_hash             SHA-256 over sorted unique atoms
  graph_atom_set_hash                SHA-256 over sorted unique atoms
  composer_graph_atom_overlap_count  int

Composer atoms come from existing pack candidate metadata
(pack_semantic_domains channel through _maybe_pack_grounded_surface).
Graph atoms come from build_graph_from_input + resolve_lemma on
node.subject/predicate/obj — no prose parsing.  When a grounded
composer path lacks explicit atom provenance, status is
'composer_no_atoms'.

New pure helper:
  chat/atom_equivalence.py — normalize_atoms, hash_atoms,
  atoms_for_graph_nodes, compare_atom_sets

Tests (tests/test_composer_graph_atom_equivalence.py):
  - Pack DEFINITION path produces observable equivalence
  - Divergent atom sets produce distinct hashes
  - Register invariance: atom hashes + status identical across
    {neutral, terse, convivial}; trace_hash also constant (R5 axis)
  - Anchor lens engaged case still ASCII-only on surface
  - No prose-parsing helper symbols introduced in runtime.py
    (extract_candidate_surface_lemmas, surface_lemma,
    parse_surface_atoms) — enforces Phase 1 boundary

Performance note: build_graph_from_input now runs on every warm
English turn (previously only when forward_graph_constraint=True).
Phase 1 accepts this cost to make the telemetry universally
available; Phase 2+ can introduce a feature flag if needed.

Validation:
  - Cognition eval byte-identical: 100/100/91.7/100
  - Full lane: 2864 passed, 3 skipped, 0 failed (+5 over baseline)
  - Targeted lane: 72 passed in tests/test_{graph_constraint,
    pack_grounding,register_tour_demo,anchor_lens_tour_demo,
    orthogonality_tour_demo,realizer_guard_holdout,
    composer_graph_atom_equivalence}.py

chat/atom_equivalence.py

@@ -0,0 +1,80 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from hashlib import sha256
+
+from chat.pack_resolver import resolve_lemma
+
+
+@dataclass(frozen=True, slots=True)
+class AtomEquivalence:
+    status: str
+    composer_atom_set_hash: str
+    graph_atom_set_hash: str
+    overlap_count: int
+
+
+def normalize_atoms(atoms: tuple[str, ...]) -> tuple[str, ...]:
+    cleaned = sorted(
+        {
+            atom.strip()
+            for atom in atoms
+            if atom and atom.strip() and atom.strip() != "<pending>"
+        }
+    )
+    return tuple(cleaned)
+
+
+def hash_atoms(atoms: tuple[str, ...]) -> str:
+    cleaned = normalize_atoms(atoms)
+    if not cleaned:
+        return ""
+    blob = "\n".join(cleaned).encode("utf-8")
+    return sha256(blob).hexdigest()
+
+
+def atoms_for_graph_nodes(graph) -> tuple[str, ...]:
+    atoms: list[str] = []
+    for node in getattr(graph, "nodes", ()) or ():
+        for surface in (
+            getattr(node, "subject", ""),
+            getattr(node, "predicate", ""),
+            getattr(node, "obj", ""),
+        ):
+            resolved = resolve_lemma(str(surface))
+            if resolved is None:
+                continue
+            _, domains = resolved
+            atoms.extend(domains)
+    return normalize_atoms(tuple(atoms))
+
+
+def compare_atom_sets(
+    *,
+    composer_atoms: tuple[str, ...],
+    graph_atoms: tuple[str, ...],
+    graph_unconstrained: bool,
+    applicable: bool,
+) -> AtomEquivalence:
+    composer_norm = normalize_atoms(composer_atoms)
+    graph_norm = normalize_atoms(graph_atoms)
+    composer_hash = hash_atoms(composer_norm)
+    graph_hash = hash_atoms(graph_norm)
+    overlap = len(set(composer_norm).intersection(graph_norm))
+
+    if not applicable:
+        status = "not_applicable"
+    elif not composer_norm:
+        status = "composer_no_atoms"
+    elif graph_unconstrained or not graph_norm:
+        status = "graph_unconstrained"
+    elif overlap > 0:
+        status = "equivalent"
+    else:
+        status = "divergent"
+    return AtomEquivalence(
+        status=status,
+        composer_atom_set_hash=composer_hash,
+        graph_atom_set_hash=graph_hash,
+        overlap_count=overlap,
+    )

chat/runtime.py

@@ -59,6 +59,7 @@ from packs.ethics.loader import (
 from packs.identity.loader import load_identity_manifold
 from chat.register_substantive import apply_substantive_register
 from chat.register_variation import decorate_surface
+from chat.atom_equivalence import atoms_for_graph_nodes, compare_atom_sets
 from generate.realizer_guard import (
     DISCLOSURE_SURFACE as _GUARD_DISCLOSURE_SURFACE,
     check_surface as _check_realizer_surface,
@@ -363,6 +364,12 @@ class ChatResponse:
     # ``trace_hash``.  Empty string ⇒ pre-R6 caller; pipeline falls
     # back to ``pre_decoration_surface`` (byte-identity preserved).
     register_canonical_surface: str = ""
+    # ADR-0078 (Phase 1) — observational composer/graph atom
+    # equivalence telemetry mirrored from TurnEvent.
+    composer_graph_atom_status: str = ""
+    composer_atom_set_hash: str = ""
+    graph_atom_set_hash: str = ""
+    composer_graph_atom_overlap_count: int = 0
 
 
 class ChatRuntime:
@@ -841,6 +848,39 @@ class ChatRuntime:
                 return (oov_surface, "oov", ())
         return None
 
+    def _graph_atom_context(
+        self,
+        text: str,
+        articulation: ArticulationPlan,
+        *,
+        region=None,
+    ) -> tuple[tuple[str, ...], bool]:
+        """Return ``(graph_atoms, graph_unconstrained)`` for observational telemetry."""
+        if self.config.output_language != "en":
+            return ((), True)
+        graph = build_graph_from_input(text, articulation)
+        graph_atoms = atoms_for_graph_nodes(graph)
+        unconstrained = len(graph_atoms) == 0
+        if region is not None:
+            unconstrained = unconstrained or getattr(region, "allowed_indices", None) is None
+        return (graph_atoms, unconstrained)
+
+    def _composer_graph_atom_equivalence(
+        self,
+        *,
+        grounding_source: str,
+        composer_atoms: tuple[str, ...],
+        graph_atoms: tuple[str, ...],
+        graph_unconstrained: bool,
+    ):
+        applicable = grounding_source in {"pack", "teaching"}
+        return compare_atom_sets(
+            composer_atoms=composer_atoms,
+            graph_atoms=graph_atoms,
+            graph_unconstrained=graph_unconstrained,
+            applicable=applicable,
+        )
+
     def _maybe_apply_discourse_planner(
         self, text: str, source_tag: str
     ) -> tuple[str, str] | None:
@@ -954,6 +994,8 @@ class ChatRuntime:
         pack_grounded_surface: str | None = None,
         grounded_source_tag: str = "pack",
         pack_semantic_domains: tuple[str, ...] = (),
+        graph_atoms: tuple[str, ...] = (),
+        graph_unconstrained: bool = True,
         discovery_intent_tag: Any = None,
         discovery_intent_subject: str | None = None,
     ) -> ChatResponse:
@@ -1094,6 +1136,12 @@ class ChatRuntime:
         anchor_lens_mode_label_stub = _extract_anchor_lens_mode_label(
             pre_decoration_surface_stub, anchor_lens_id_stub,
         )
+        atom_equivalence_stub = self._composer_graph_atom_equivalence(
+            grounding_source=grounding_source,
+            composer_atoms=pack_semantic_domains,
+            graph_atoms=graph_atoms,
+            graph_unconstrained=graph_unconstrained,
+        )
         verdicts_bundle = TurnVerdicts(
             identity_score=None,
             safety_verdict=safety_verdict,
@@ -1126,6 +1174,10 @@ class ChatRuntime:
                 realizer_guard_status=realizer_guard_status_stub,
                 realizer_guard_rule=realizer_guard_rule_stub,
                 register_canonical_surface=register_canonical_surface_stub,
+                composer_graph_atom_status=atom_equivalence_stub.status,
+                composer_atom_set_hash=atom_equivalence_stub.composer_atom_set_hash,
+                graph_atom_set_hash=atom_equivalence_stub.graph_atom_set_hash,
+                composer_graph_atom_overlap_count=atom_equivalence_stub.overlap_count,
             )
             self.turn_log.append(stub_event)
             self._emit_turn_event(stub_event)
@@ -1177,6 +1229,10 @@ class ChatRuntime:
             realizer_guard_status=realizer_guard_status_stub,
             realizer_guard_rule=realizer_guard_rule_stub,
             register_canonical_surface=register_canonical_surface_stub,
+            composer_graph_atom_status=atom_equivalence_stub.status,
+            composer_atom_set_hash=atom_equivalence_stub.composer_atom_set_hash,
+            graph_atom_set_hash=atom_equivalence_stub.graph_atom_set_hash,
+            composer_graph_atom_overlap_count=atom_equivalence_stub.overlap_count,
         )
 
     def chat(self, text: str, max_tokens: int | None = None) -> ChatResponse:
@@ -1230,6 +1286,8 @@ class ChatRuntime:
             )
             discovery_intent_tag = None
             discovery_intent_subject: str | None = None
+            stub_graph_atoms: tuple[str, ...] = ()
+            stub_graph_unconstrained = True
             if (
                 gate_decision.source == "empty_vault"
                 and self.config.output_language == "en"
@@ -1238,12 +1296,26 @@ class ChatRuntime:
                 _intent = classify_intent_from_input(text)
                 discovery_intent_tag = _intent.tag
                 discovery_intent_subject = _intent.subject
+                stub_articulation = ArticulationPlan(
+                    subject=_intent.subject or "",
+                    predicate="",
+                    object=None,
+                    surface="",
+                    output_language=self.config.output_language,
+                    frame_id="unknown_domain",
+                )
+                stub_graph_atoms, stub_graph_unconstrained = self._graph_atom_context(
+                    text,
+                    stub_articulation,
+                )
             return self._stub_response(
                 committed,
                 tokens=tuple(filtered),
                 pack_grounded_surface=pack_surface,
                 grounded_source_tag=pack_source_tag,
                 pack_semantic_domains=pack_semantic_domains,
+                graph_atoms=stub_graph_atoms,
+                graph_unconstrained=stub_graph_unconstrained,
                 discovery_intent_tag=discovery_intent_tag,
                 discovery_intent_subject=discovery_intent_subject,
             )
@@ -1276,9 +1348,20 @@ class ChatRuntime:
         self._context.record_dialogue(proposition)
 
         forward_region = None
-        if self.config.forward_graph_constraint and self.config.output_language == "en":
+        graph_atoms_main: tuple[str, ...] = ()
+        graph_unconstrained_main = True
+        if self.config.output_language == "en":
             pre_gen_graph = build_graph_from_input(text, articulation)
-            forward_region = build_graph_constraint(pre_gen_graph, self._context.vocab)
+            graph_atoms_main = atoms_for_graph_nodes(pre_gen_graph)
+            if self.config.forward_graph_constraint:
+                forward_region = build_graph_constraint(pre_gen_graph, self._context.vocab)
+            graph_unconstrained_main = (
+                len(graph_atoms_main) == 0
+                or (
+                    forward_region is not None
+                    and getattr(forward_region, "allowed_indices", None) is None
+                )
+            )
 
         result = generate(
             field_state,
@@ -1518,6 +1601,12 @@ class ChatRuntime:
         anchor_lens_mode_label_main = _extract_anchor_lens_mode_label(
             pre_decoration_surface_main, anchor_lens_id_main,
         )
+        atom_equivalence_main = self._composer_graph_atom_equivalence(
+            grounding_source=warm_grounding_source or "vault",
+            composer_atoms=warm_pack_semantic_domains,
+            graph_atoms=graph_atoms_main,
+            graph_unconstrained=graph_unconstrained_main,
+        )
         verdicts_bundle = TurnVerdicts(
             identity_score=identity_score,
             safety_verdict=safety_verdict,
@@ -1549,6 +1638,10 @@ class ChatRuntime:
             realizer_guard_status=realizer_guard_status_main,
             realizer_guard_rule=realizer_guard_rule_main,
             register_canonical_surface=register_canonical_surface_main,
+            composer_graph_atom_status=atom_equivalence_main.status,
+            composer_atom_set_hash=atom_equivalence_main.composer_atom_set_hash,
+            graph_atom_set_hash=atom_equivalence_main.graph_atom_set_hash,
+            composer_graph_atom_overlap_count=atom_equivalence_main.overlap_count,
         )
         self.turn_log.append(turn_event)
         self._emit_turn_event(turn_event)
@@ -1589,6 +1682,10 @@ class ChatRuntime:
             realizer_guard_status=realizer_guard_status_main,
             realizer_guard_rule=realizer_guard_rule_main,
             register_canonical_surface=register_canonical_surface_main,
+            composer_graph_atom_status=atom_equivalence_main.status,
+            composer_atom_set_hash=atom_equivalence_main.composer_atom_set_hash,
+            graph_atom_set_hash=atom_equivalence_main.graph_atom_set_hash,
+            composer_graph_atom_overlap_count=atom_equivalence_main.overlap_count,
         )
 
     def _unknown_domain_response(self, field_state: FieldState, filtered: list[str]) -> ChatResponse:

chat/telemetry.py

@@ -83,6 +83,18 @@ def serialize_turn_event(
         # Empty strings on pre-C1 events; closed enums otherwise.
         "realizer_guard_status": str(getattr(event, "realizer_guard_status", "") or ""),
         "realizer_guard_rule": str(getattr(event, "realizer_guard_rule", "") or ""),
+        "composer_graph_atom_status": str(
+            getattr(event, "composer_graph_atom_status", "") or ""
+        ),
+        "composer_atom_set_hash": str(
+            getattr(event, "composer_atom_set_hash", "") or ""
+        ),
+        "graph_atom_set_hash": str(
+            getattr(event, "graph_atom_set_hash", "") or ""
+        ),
+        "composer_graph_atom_overlap_count": int(
+            getattr(event, "composer_graph_atom_overlap_count", 0) or 0
+        ),
     }
     safety = getattr(event, "safety_verdict", None)
     if safety is not None:

core/physics/identity.py

@@ -333,3 +333,10 @@ class TurnEvent:
     # pipeline falls back to the existing ``pre_decoration_surface``
     # source in that case (byte-identity preserved).
     register_canonical_surface: str = ""
+    # ADR-0078 (Phase 1) — observational composer/graph atom
+    # equivalence telemetry.  Empty defaults preserve back-compat for
+    # pre-ADR-0078 callers and non-applicable turns.
+    composer_graph_atom_status: str = ""
+    composer_atom_set_hash: str = ""
+    graph_atom_set_hash: str = ""
+    composer_graph_atom_overlap_count: int = 0

docs/decisions/ADR-0078-phase1-implementation-note.md

@@ -0,0 +1,27 @@
+# ADR-0078 Phase 1 — Pre-Implementation Planning Note
+
+1. Where composer atoms come from.
+- On DEFINITION/RECALL pack-grounded paths, composer provenance is available from existing pack candidate metadata (`build_pack_surface_candidate(...).semantic_domains`) and from the existing `_maybe_pack_grounded_surface(...)->pack_semantic_domains` return channel.
+- Other composer paths do not always expose explicit atom provenance today; those will report `composer_no_atoms` when telemetry is applicable and grounded but atom provenance is absent.
+
+2. Where graph-side atoms/indices are derived.
+- Graph topology comes from `build_graph_from_input(text, articulation)` and forward constraints from `build_graph_constraint(graph, vocab)`.
+- Phase 1 graph atom telemetry will be observational and derived by resolving graph node surfaces (`subject/predicate/obj`) through `chat.pack_resolver.resolve_lemma`, unioning resolved semantic domains.
+- If no graph nodes resolve to atoms (or graph constraint is unconstrained), graph hash remains empty and status can become `graph_unconstrained`.
+
+3. Exact telemetry hook location.
+- Hook in `chat/runtime.py` after final composer surface / grounding source are known (cold/stub and warm paths) and after pre-generation graph context is known, but before `TurnEvent` and `ChatResponse` are finalized.
+- Keep this observational only: compute status/hash/overlap and attach fields; do not alter surface selection, guard outcomes, grounding source, or trace behavior.
+
+4. Why register variation does not affect atom hashes.
+- Register decoration/substantive transforms operate on rendered surface layers (`register_canonical_surface` -> pre-decoration -> decorated surface).
+- Composer atom provenance and graph atoms come from pack/graph structures, not register text transforms; therefore register changes should not perturb atom-set hashes/status for same prompt/lens/runtime state.
+
+5. Why anchor-lens engagement may affect substantive/proposition telemetry.
+- Anchor-lens changes composer proposition selection by engaging substrate-aligned semantic preferences (ADR-0073c/0073d), so resulting semantic domains and graph realization can legitimately differ when lens changes.
+- This is substantive-axis movement, not register-style presentation variation; telemetry must allow divergence/equivalence outcomes without forcing false invariants.
+
+6. Confirmation that no final surface prose parsing will be added.
+- No parsing of `ChatResponse.surface`/final prose will be introduced for atom inference.
+- No helpers like `extract_candidate_surface_lemmas`, `surface_lemma`, or `parse_surface_atoms` will be added.
+- If a grounded composer path lacks explicit atom provenance, status will be `composer_no_atoms`.

tests/test_composer_graph_atom_equivalence.py

@@ -0,0 +1,79 @@
+from __future__ import annotations
+
+from chat.atom_equivalence import compare_atom_sets
+from chat.runtime import ChatRuntime
+from core.cognition.pipeline import CognitiveTurnPipeline
+from core.config import RuntimeConfig
+
+
+def _run(prompt: str, *, register: str = "default_neutral_v1", lens: str | None = None):
+    rt = ChatRuntime(
+        config=RuntimeConfig(
+            register_pack_id=register,
+            anchor_lens_id=lens,
+        )
+    )
+    pipeline = CognitiveTurnPipeline(runtime=rt)
+    result = pipeline.run(prompt)
+    response = rt.turn_log[-1]
+    return result, response
+
+
+def test_pack_definition_equivalence_observable():
+    _, event = _run("What is truth?")
+    assert event.grounding_source == "pack"
+    assert event.composer_graph_atom_status in {"equivalent", "graph_unconstrained"}
+    if event.composer_graph_atom_status == "equivalent":
+        assert event.composer_atom_set_hash != ""
+        assert event.graph_atom_set_hash != ""
+        assert event.composer_graph_atom_overlap_count > 0
+
+
+def test_corrupt_graph_divergence_observable_without_surface_behavior():
+    eq = compare_atom_sets(
+        composer_atoms=("logos.aletheia.verity",),
+        graph_atoms=("logos.unrelated.test_atom",),
+        graph_unconstrained=False,
+        applicable=True,
+    )
+    assert eq.status == "divergent"
+    assert eq.overlap_count == 0
+    assert eq.composer_atom_set_hash != ""
+    assert eq.graph_atom_set_hash != ""
+    assert eq.composer_atom_set_hash != eq.graph_atom_set_hash
+
+
+def test_register_invariance_of_atom_equivalence():
+    prompt = "What is truth?"
+    neutral_result, neutral = _run(prompt, register="default_neutral_v1")
+    terse_result, terse = _run(prompt, register="terse_v1")
+    convivial_result, convivial = _run(prompt, register="convivial_v1")
+
+    assert neutral.composer_atom_set_hash == terse.composer_atom_set_hash == convivial.composer_atom_set_hash
+    assert neutral.graph_atom_set_hash == terse.graph_atom_set_hash == convivial.graph_atom_set_hash
+    assert neutral.composer_graph_atom_status == terse.composer_graph_atom_status == convivial.composer_graph_atom_status
+    assert neutral_result.trace_hash == terse_result.trace_hash == convivial_result.trace_hash
+
+
+def test_anchor_lens_engaged_case_telemetry_computes_without_glyph_leak():
+    _, engaged = _run("What is knowledge?", lens="grc_logos_v1")
+    _, unanchored = _run("What is knowledge?")
+
+    assert engaged.composer_graph_atom_status != ""
+    assert unanchored.composer_graph_atom_status != ""
+    assert all(ord(ch) < 128 for ch in engaged.surface)
+    assert all(ord(ch) < 128 for ch in unanchored.surface)
+
+
+def test_no_final_surface_lemma_parser_symbols():
+    import pathlib
+
+    forbidden = (
+        "extract_candidate_surface_lemmas",
+        "surface_lemma",
+        "parse_surface_atoms",
+    )
+    root = pathlib.Path(__file__).resolve().parent.parent
+    runtime_src = (root / "chat" / "runtime.py").read_text(encoding="utf-8")
+    for token in forbidden:
+        assert token not in runtime_src