feat(W-020b): DerivedRecognizer producer wiring (ADR-0154) (#278)

W-007/ADR-0149 wired the consumer side of the recognizer registry
(first_admitted_recognizer → graph derivation, opt-in via
recognition_grounded_graph). The producer side — capturing
(tokens, bundle) from admitted turns so derive_recognizer at
checkpoint can anti-unify them — had no production caller.
record_recognition_example existed but was only invoked by tests,
so _pending_recognizer_examples stayed empty in live sessions and
the registry could never grow from traffic.

Observed: 103-turn session wrote recognizers.jsonl empty even with
recognition running.

- CognitiveTurnPipeline.run calls runtime.record_recognition_example
  at the admitted-recognition boundary
- Producer fires unconditionally; consumer (derive_recognizer at
  checkpoint) stays opt-in behind the same flag — flipping it later
  is no longer a cold start
- hasattr guard keeps the pipeline tolerant of non-ChatRuntime
  runtimes

Validated: tests/test_adr_0154_recognizer_producer_wiring.py (5
tests covering admit/refuse, flag-off producer, end-to-end loop,
accumulation); core test --suite cognition/smoke + recognition
phase 1/2/refusal-propagation all green.

Out of scope: bootstrap of the first recognizer from operator
review (substrate-liveness audit scope); bounded growth of the
producer queue when consumer flag stays off (future LRU cap).

core/cognition/pipeline.py

@@ -175,6 +175,25 @@ class CognitiveTurnPipeline:
                     nodes=(_ep_node,),
                     recognizer_id=self._recognizer.teaching_set_id,
                 )
+                # ADR-0154 (W-020b) — producer-side wiring for the
+                # DerivedRecognizer registry.  When a recognizer admits a
+                # turn, capture (tokens, bundle) so the registry can
+                # derive tighter recognizers via anti-unification at the
+                # next checkpoint.  Pre-ADR-0154 the producer hook had
+                # no production caller (only tests invoked
+                # ``record_recognition_example``), so the registry
+                # could never grow from live traffic regardless of
+                # whether ``recognition_grounded_graph`` was enabled.
+                # The producer fires unconditionally; the consumer
+                # (``checkpoint_engine_state``'s derive_recognizer
+                # call) stays opt-in behind the same flag.
+                if (
+                    _rec_outcome.proposition is not None
+                    and hasattr(self.runtime, "record_recognition_example")
+                ):
+                    self.runtime.record_recognition_example(
+                        raw_tokens, _rec_outcome.proposition
+                    )
             elif _rec_outcome.refusal_reason is not None:
                 from generate.exhaustion import RefusalReason as _ExhaustionRefusalReason
                 _recognition_refusal_reason = _ExhaustionRefusalReason.RECOGNITION_REFUSED.value

docs/decisions/ADR-0154-recognizer-producer-wiring.md

@@ -0,0 +1,101 @@
+# ADR-0154 — DerivedRecognizer producer wiring (W-020b)
+
+Status: accepted
+Date: 2026-05-25
+
+## Context
+
+ADR-0149 (W-007) wired the `DerivedRecognizer` registry's **consumer**
+side: `runtime.first_admitted_recognizer()` is read by
+`CognitiveTurnPipeline.__init__` and feeds the optional
+recognition-grounded graph at `pipeline.py` ~line 217 (gated by
+`recognition_grounded_graph`, default off).
+
+The **producer** side — capturing `(tokens, bundle)` from admitted
+turns so `derive_recognizer` at checkpoint can anti-unify them into
+tighter recognizers — was never connected in production code.
+`runtime.record_recognition_example` had zero non-test callers:
+
+```bash
+$ grep -rn record_recognition_example --include="*.py" | grep -v test
+chat/runtime.py:703:    def record_recognition_example(
+```
+
+Consequence: `_pending_recognizer_examples` stayed permanently empty,
+so the conditional at `chat/runtime.py:684-691` —
+
+```python
+if (
+    self.config.recognition_grounded_graph
+    and self._pending_recognizer_examples
+):
+    recognizer = derive_recognizer(...)
+    ...
+```
+
+— never fired, even with the flag enabled. The registry could only
+grow via tests calling `record_recognition_example` directly.
+Observed symptom: a 103-turn session wrote `recognizers.jsonl` as
+empty even though recognition was running.
+
+## Decision
+
+In `CognitiveTurnPipeline.run`, at the admitted-recognition boundary
+(directly after `EpistemicGraph` construction), call
+`runtime.record_recognition_example(raw_tokens, _rec_outcome.proposition)`.
+
+- **Producer fires unconditionally** when a turn is admitted — the
+  bucket is filled regardless of `recognition_grounded_graph`. This
+  means flipping the consumer flag later is not a cold start.
+- **Consumer stays opt-in** behind the same flag — no change to
+  `checkpoint_engine_state`'s `derive_recognizer` gate.
+- `hasattr` guard on `runtime.record_recognition_example` keeps the
+  pipeline tolerant of non-`ChatRuntime` runtimes (test doubles,
+  alternative shells).
+
+## Invariants
+
+- Refused recognition: no producer call (gated inside
+  `if _rec_outcome.admitted:`).
+- No attached recognizer: no recognition runs at all, no producer
+  call.
+- Per-turn FeatureBundle is the validated proposition emitted by
+  `recognize` — no shape massaging in the pipeline.
+- `recognize` is unchanged; `derive_recognizer` is unchanged; trace
+  hash bytes are unchanged for any given turn.
+
+## Out of scope
+
+- **Bootstrap of the very first recognizer.** This ADR closes the
+  loop *given* a recognizer is attached. No path in production code
+  seeds the first recognizer from operator review or reviewed
+  teaching examples; that is a substrate-liveness concern tracked
+  separately under the ADR-0143 / substrate-liveness audit family.
+- **Unbounded growth of `_pending_recognizer_examples` when the
+  consumer flag stays off.** With flag=False, the producer
+  accumulates forever. Acceptable for short sessions; a future
+  bound (LRU or cap) should ship before long-running operators
+  enable the producer with the consumer off.
+
+## Validation
+
+`tests/test_adr_0154_recognizer_producer_wiring.py`:
+- admitted turn appends `(tokens, bundle)` to the producer queue
+  (flag=False so the queue is not drained at checkpoint)
+- producer fires when consumer flag is off
+- refused turn does not populate the queue
+- end-to-end loop: with flag=True, an admitted turn feeds the
+  producer queue, then `checkpoint_engine_state` drains it via
+  `derive_recognizer` and registers the result
+- multiple admitted turns accumulate in order
+
+CLI lanes: `core test --suite cognition` (120 + 1 skipped),
+`core test --suite smoke` (67), recognition phase 1/2 + refusal
+propagation (25) all green.
+
+## Closure
+
+After this ADR, the DerivedRecognizer registry can grow from live
+traffic. The remaining gap is bootstrap — getting the first
+recognizer into the registry without test-only injection. That is a
+substrate-liveness scope concern, not W-020b.

tests/test_adr_0154_recognizer_producer_wiring.py

@@ -0,0 +1,222 @@
+"""ADR-0154 (W-020b) — producer-side wiring for DerivedRecognizer registry.
+
+Pre-ADR-0154, ``ChatRuntime.record_recognition_example`` had no
+production caller — only tests invoked it.  Result: the
+``_pending_recognizer_examples`` bucket stayed empty regardless of
+how many turns were admitted by an attached recognizer, so
+``derive_recognizer`` at the next checkpoint had nothing to
+anti-unify.  The registry could never grow from live traffic, even
+when ``recognition_grounded_graph`` was enabled.
+
+Fix: in ``CognitiveTurnPipeline.run`` at the admitted-recognition
+boundary, capture ``(raw_tokens, _rec_outcome.proposition)`` via
+``runtime.record_recognition_example``.  Producer fires
+unconditionally; consumer (``derive_recognizer`` in
+``checkpoint_engine_state``) stays opt-in behind the same flag.
+"""
+from __future__ import annotations
+
+from dataclasses import replace
+from pathlib import Path
+
+from chat.runtime import ChatRuntime
+from core.cognition import CognitiveTurnPipeline
+from core.config import DEFAULT_CONFIG
+from recognition.anti_unifier import derive_recognizer
+from recognition.outcome import EvidenceSpan, FeatureBundle, NegativeEvidence
+
+
+def _config(*, recognition_grounded_graph: bool):
+    return replace(
+        DEFAULT_CONFIG,
+        recognition_grounded_graph=recognition_grounded_graph,
+    )
+
+
+def _span(tokens: tuple[str, ...], start: int, end: int) -> EvidenceSpan:
+    return EvidenceSpan(start=start, end=end, text=" ".join(tokens[start:end]))
+
+
+def _bundle(
+    tokens: tuple[str, ...],
+    agent_span: tuple[int, int],
+    count_span: tuple[int, int],
+    unit_span: tuple[int, int],
+    agent: str,
+    count: int,
+    unit: str,
+) -> FeatureBundle:
+    return FeatureBundle.from_mapping(
+        {
+            "agent": (agent, _span(tokens, *agent_span)),
+            "count": (count, _span(tokens, *count_span)),
+            "intentionality": (
+                "possession",
+                _span(
+                    tokens,
+                    1 if tokens[0] in {"A", "The"} else 0,
+                    3 if tokens[0] in {"A", "The"} else 2,
+                ),
+            ),
+            "modality": (
+                "actual",
+                NegativeEvidence(0, len(tokens), "no modal counter-marker present"),
+            ),
+            "polarity": ("+", NegativeEvidence(0, len(tokens), "no negator present")),
+            "relation": ("has", _span(tokens, count_span[0] - 1, count_span[0])),
+            "tense": ("present", _span(tokens, count_span[0] - 1, count_span[0])),
+            "unit": (unit, _span(tokens, *unit_span)),
+        }
+    )
+
+
+def _examples() -> list[tuple[tuple[str, ...], FeatureBundle]]:
+    john = ("John", "has", "5", "apples")
+    mary = ("Mary", "has", "3", "books")
+    school = ("A", "school", "has", "100", "students")
+    library = ("The", "library", "has", "12", "chairs")
+    return [
+        (john, _bundle(john, (0, 1), (2, 3), (3, 4), "John", 5, "apple")),
+        (mary, _bundle(mary, (0, 1), (2, 3), (3, 4), "Mary", 3, "book")),
+        (
+            school,
+            _bundle(school, (1, 2), (3, 4), (4, 5), "school", 100, "student"),
+        ),
+        (
+            library,
+            _bundle(library, (1, 2), (3, 4), (4, 5), "library", 12, "chair"),
+        ),
+    ]
+
+
+def _recognizer():
+    return derive_recognizer(_examples())
+
+
+def test_admitted_turn_records_recognition_example(tmp_path: Path) -> None:
+    """Admitted recognition appends (tokens, bundle) to the producer queue.
+
+    Uses flag=False so the consumer (``checkpoint_engine_state``'s
+    ``derive_recognizer``) does not drain the queue at end-of-turn;
+    that lets us assert the producer's output directly.
+    Recognizer attached via pipeline constructor because
+    ``runtime.first_admitted_recognizer`` is gated on the flag.
+    """
+    runtime = ChatRuntime(
+        config=_config(recognition_grounded_graph=False),
+        engine_state_path=tmp_path,
+    )
+    pipe = CognitiveTurnPipeline(runtime, recognizer=_recognizer())
+    assert runtime._pending_recognizer_examples == []
+
+    result = pipe.run("A baker has 24 loaves", max_tokens=4)
+
+    assert result.epistemic_graph is not None, (
+        "fixture must admit; otherwise the producer hook is not exercised"
+    )
+    assert len(runtime._pending_recognizer_examples) == 1
+    tokens, bundle = runtime._pending_recognizer_examples[0]
+    assert tokens == ("A", "baker", "has", "24", "loaves")
+    assert isinstance(bundle, FeatureBundle)
+    # Bundle must be complete (anti-unifier invariant).
+    assert {f.name for f in bundle.features} >= {
+        "agent",
+        "count",
+        "unit",
+    }
+
+
+def test_producer_fires_when_consumer_flag_off(tmp_path: Path) -> None:
+    """Producer must NOT be gated on ``recognition_grounded_graph``.
+
+    The consumer (derive_recognizer at checkpoint) is opt-in; the
+    producer is unconditional so flipping the flag later is not a
+    cold start.  Without an attached recognizer (registry empty +
+    flag off), no recognition runs at all, so we attach one
+    directly to the pipeline.
+    """
+    runtime = ChatRuntime(
+        config=_config(recognition_grounded_graph=False),
+        engine_state_path=tmp_path,
+    )
+    pipe = CognitiveTurnPipeline(runtime, recognizer=_recognizer())
+
+    result = pipe.run("A baker has 24 loaves", max_tokens=4)
+
+    # Flag is off → graph derivation skipped, but producer must still
+    # have captured the admitted example.
+    assert result.epistemic_graph is not None  # pipeline-level admit
+    assert len(runtime._pending_recognizer_examples) == 1
+
+
+def test_refused_turn_does_not_record_example(tmp_path: Path) -> None:
+    """Refused recognition must not populate the producer queue."""
+    runtime = ChatRuntime(
+        config=_config(recognition_grounded_graph=False),
+        engine_state_path=tmp_path,
+    )
+    pipe = CognitiveTurnPipeline(runtime, recognizer=_recognizer())
+
+    # Input that does not match the (agent, has, count, unit) pattern.
+    pipe.run("Hello world", max_tokens=4)
+
+    assert runtime._pending_recognizer_examples == []
+
+
+def test_full_loop_admit_then_derive_registers_new_recognizer(
+    tmp_path: Path,
+) -> None:
+    """End-to-end producer→consumer: with flag=True, an admitted turn
+    feeds the producer queue, then ``checkpoint_engine_state`` drains
+    the queue via ``derive_recognizer`` and registers the result.
+    Pre-ADR-0154 this loop could not close from live traffic because
+    the producer was never wired.
+    """
+    runtime = ChatRuntime(
+        config=_config(recognition_grounded_graph=True),
+        engine_state_path=tmp_path,
+    )
+    seed = _recognizer()
+    runtime._recognizer_registry.register(seed)
+    registry_size_before = len(runtime._recognizer_registry)
+
+    CognitiveTurnPipeline(runtime).run(
+        "A baker has 24 loaves", max_tokens=4
+    )
+
+    # The consumer drained the queue at checkpoint and registered the
+    # newly-derived recognizer (overwriting the seed under the same
+    # teaching_set_id, since derive_recognizer is byte-deterministic).
+    assert runtime._pending_recognizer_examples == []
+    assert len(runtime._recognizer_registry) >= registry_size_before
+
+
+def test_examples_accumulate_across_admitted_turns(tmp_path: Path) -> None:
+    """Multiple admitted turns append in order.
+
+    Flag=False so the consumer does not drain between turns;
+    that lets us assert producer accumulation directly.
+    """
+    runtime = ChatRuntime(
+        config=_config(recognition_grounded_graph=False),
+        engine_state_path=tmp_path,
+    )
+    pipe = CognitiveTurnPipeline(runtime, recognizer=_recognizer())
+    pipe.run("A baker has 24 loaves", max_tokens=4)
+    pipe.run("The farmer has 7 sheep", max_tokens=4)
+
+    assert len(runtime._pending_recognizer_examples) == 2
+    assert runtime._pending_recognizer_examples[0][0] == (
+        "A",
+        "baker",
+        "has",
+        "24",
+        "loaves",
+    )
+    assert runtime._pending_recognizer_examples[1][0] == (
+        "The",
+        "farmer",
+        "has",
+        "7",
+        "sheep",
+    )