From 327047ce2631d5923bcccea9ee74c7c7284b6b6e Mon Sep 17 00:00:00 2001 From: Shay Date: Thu, 21 May 2026 10:55:39 -0700 Subject: [PATCH] =?UTF-8?q?feat(contemplation):=20Phase=205=20=E2=80=94=20?= =?UTF-8?q?articulation-quality=20miner=20closes=20the=20loop?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Final phase of the articulation arc. Consumes the per-turn ``PlanMetrics`` + ``ContemplationFinding`` streams produced by Phases 3 + 4 and aggregates across many turns to emit SPECULATIVE ``PACK_MUTATION_CANDIDATE`` findings that the operator reviews via the existing proposal-review-ratify chain. This is the doctrine-aligned answer to the user's question: "Should we... realize a way to score whether it should use what it produced towards memory confidence for future use?" Yes — and it stays inside ADR-0080: read-only, SPECULATIVE-only, deterministic, no parallel learning path, no autonomous memory mutation. What it adds ------------ * New module ``chat/articulation_telemetry.py``: - ``ArticulationObservation`` frozen dataclass — per-turn bundle of (turn_id, anchor_subject, prompt_hash, plan_substrate_hash, metrics, findings). - ``format_articulation_observation_jsonl(...)`` — deterministic sort-keys JSONL line. - ``load_articulation_observations(lines)`` — schema-tolerant loader; malformed lines drop without aborting. - ``ArticulationObservationSink`` protocol — structurally identical to ``TurnEventSink`` but distinct named type so consumers can subscribe to one stream without the other. * New module ``core/contemplation/miners/articulation_quality.py``: - ``mine_articulation_observations(observations, paths)`` — pure deterministic aggregator with three v1 rules. - **recurring_predicate_monotony** — when the same (subject, predicate) pair is flagged WEAK_SURFACE in >= _MIN_RECURRENCE (default 3) observations, propose substrate diversification with non-dominant predicates. - **recurring_planner_gap** — when the same subject is flagged PLANNER_GAP >= _MIN_RECURRENCE times across modes, propose substrate expansion. - **low_average_predicate_diversity** — when mean ``predicate_diversity_ratio`` < 0.5 across >= _MIN_RECURRENCE observations on the same anchor subject, propose diversification. * Runtime wiring (``chat/runtime.py``): - New ``ChatRuntime.attach_articulation_sink(sink)`` method. Mirrors ``attach_telemetry_sink`` pattern. - Emission point at the end of ``_maybe_apply_discourse_planner``: when contemplation enabled + sink attached + plan engaged, builds an ``ArticulationObservation`` and emits one JSONL line. Sink errors propagate (fail-fast, no swallowing). - Per-runtime ``_articulation_turn_counter`` increments on every emission; gives downstream consumers a stable sequence index. Tests ----- * ``tests/test_articulation_quality_miner.py`` (11 tests): - Empty / sub-threshold cases yield no findings. - Each of the three rules fires at threshold. - Recurring_predicate_monotony separates by subject (no cross-subject merging). - Recurring_planner_gap collects distinct modes into a sorted comma-joined string. - Determinism — byte-equal finding IDs across two runs. - SPECULATIVE doctrine pin. - JSONL round-trip preserves observation identity. * ``tests/test_articulation_quality_e2e.py`` (7 tests): - Sink-detached + contemplation-on → no emission. - Sink-attached + contemplation-off → no emission. - Engaged turn emits exactly one observation line. - BRIEF prompt emits nothing (fast-path). - **Full loop** — run compound prompt 3x → 3 observations → miner emits PACK_MUTATION_CANDIDATE with subject='truth', predicate='recurring_predicate_monotony', object='belongs_to'. - Full loop is deterministic (byte-equal finding IDs across two complete runs). - Every full-loop finding is SPECULATIVE. Doctrine pins ------------- | Claim | Pinned by | |--------------------------------------|----------------------------------------------------------| | SPECULATIVE-only | test_all_findings_remain_speculative | | Deterministic across runs | test_miner_is_deterministic_across_runs | | Full-loop determinism (e2e) | test_full_loop_is_deterministic_byte_equal_finding_ids | | No autonomous mutation | Sink is append-only; miner outputs ContemplationFinding | | | objects only; nothing writes to packs/vault/teaching. | | Append-only stream | Sink protocol has emit(line: str) and nothing else. | Live demo (3 identical compound-prompt turns) --------------------------------------------- Runtime emits 3 observations. Offline miner aggregates and emits: [pack_mutation_candidate] subject='truth' predicate='recurring_predicate_monotony' object='belongs_to' evidence_refs: 3 observations proposed_action: "diversify substrate for 'truth': across 3 observations the plan repeatedly over-concentrated on predicate 'belongs_to'. Candidates: add teaching chains rooted on 'truth' with relations OTHER than 'belongs_to' (grounds / requires / reveals / contrasts / precedes / follows) so the planner's RELATION selector has more variety to draw from." epistemic_status: speculative The system observed its own articulation patterns across many turns, identified the corpus expansion priority, and emitted a specific reviewable proposal — without mutating anything. The operator decides whether to act on it via the existing review chain. Verification ------------ pytest test_articulation_quality_miner.py 11/11 pass pytest test_articulation_quality_e2e.py 7/7 pass pytest test_plan_metrics*.py 18/18 pass (Phase 4) pytest test_plan_contemplation*.py 17/17 pass (Phase 3) pytest test_discourse_planner_*.py 99/99 pass pytest test_articulation_demo.py all claims supported pytest test_narrative_example_intents.py pass core test --suite smoke 67/67 pass core test --suite runtime 19/19 pass The articulation arc is complete. Future work documented in ``docs/sessions/SESSION-2026-05-21-articulation-arc.md`` §8: connective rotation, generalised pronoun selection, doctrine-gated plan revision, Phase 2.5 mid-sentence reflection. None blocking. --- chat/articulation_telemetry.py | 198 ++++++++++ chat/runtime.py | 71 ++++ .../miners/articulation_quality.py | 352 ++++++++++++++++++ tests/test_articulation_quality_e2e.py | 192 ++++++++++ tests/test_articulation_quality_miner.py | 307 +++++++++++++++ 5 files changed, 1120 insertions(+) create mode 100644 chat/articulation_telemetry.py create mode 100644 core/contemplation/miners/articulation_quality.py create mode 100644 tests/test_articulation_quality_e2e.py create mode 100644 tests/test_articulation_quality_miner.py diff --git a/chat/articulation_telemetry.py b/chat/articulation_telemetry.py new file mode 100644 index 00000000..9a6647b3 --- /dev/null +++ b/chat/articulation_telemetry.py @@ -0,0 +1,198 @@ +"""Phase 5 — per-turn articulation observation schema + sink. + +The runtime emits one structured observation per engaged turn +(``discourse_contemplation=True`` AND planner produced a multi-move +plan). Each observation bundles Phase 4 metrics and Phase 3 findings +plus identity context (turn_id, anchor subject, plan substrate hash) +so the offline contemplation miner (Phase 5) can aggregate across +many turns and emit reviewable pack-mutation candidates. + +Doctrine alignment (ADR-0080 + ADR-0040): + +* Read-only — observations are PROJECTIONS of plan state; nothing + is mutated when they emit. +* Append-only — sinks ONLY accept JSONL lines; observations never + rewrite or overwrite prior records. +* Deterministic — same plan + same metrics + same findings → + byte-identical JSONL line. Pinned by the + ``test_observation_is_deterministic`` test. +* SPECULATIVE-only by transitivity — the findings carried inside + each observation are themselves SPECULATIVE (enforced by the + ContemplationFinding schema's __post_init__). + +The sink protocol mirrors ``chat.telemetry.TurnEventSink``: any +object with ``def emit(line: str) -> None`` satisfies the contract. +Articulation observations flow through a SEPARATE sink (not the +turn-event sink) so consumers can subscribe to one stream without +the other, and the two streams' wire formats stay independent. +""" + +from __future__ import annotations + +import hashlib +import json +from dataclasses import dataclass +from typing import Any, Iterable, Protocol + + +# --------------------------------------------------------------------------- +# Schema +# --------------------------------------------------------------------------- + + +@dataclass(frozen=True, slots=True) +class ArticulationObservation: + """One per-turn articulation observation. + + Carries the Phase 4 metrics dict + the Phase 3 findings (compacted + to (kind, subject, predicate, object) tuples) plus identity fields. + Both inner payloads are pre-serialised via ``as_dict()`` so the + observation itself owns no live references — safe to log, archive, + or stream without keeping the runtime alive. + """ + + turn_id: int + """Sequential turn index within the emitting session (0-based).""" + + anchor_subject: str + """The plan's anchor subject lemma — the most stable aggregation + key. For a typical EXPLAIN/PARAGRAPH plan this is the prompt's + head noun; for compound prompts it is the primary part's + subject.""" + + prompt_hash: str + """SHA-256-16 of the (lowercased, stripped) raw prompt text. + Lets the miner detect repeated prompts without storing raw + user input.""" + + plan_substrate_hash: str + """SHA-256-16 of the plan's canonical JSON. Joins this + observation to the Phase 3 contemplation findings that share + the same substrate hash.""" + + metrics: dict[str, Any] + """Phase 4 ``PlanMetrics.as_dict()`` — see + ``core.contemplation.plan_metrics`` for field list and + semantics.""" + + findings: tuple[dict[str, str | None], ...] + """Phase 3 finding summaries — each is + ``{"kind": , "subject": , + "predicate": , "object": }``. Compacted form; + the full finding objects (with substrate_hash, finding_id, + proposed_action, evidence_refs) are recoverable from a separate + findings stream that emits via the existing + ``DiscoveryCandidateSink``.""" + + def as_dict(self) -> dict[str, Any]: + return { + "turn_id": self.turn_id, + "anchor_subject": self.anchor_subject, + "prompt_hash": self.prompt_hash, + "plan_substrate_hash": self.plan_substrate_hash, + "metrics": dict(self.metrics), + "findings": [dict(f) for f in self.findings], + } + + +# --------------------------------------------------------------------------- +# Serialisation +# --------------------------------------------------------------------------- + + +def serialize_articulation_observation( + observation: ArticulationObservation, +) -> dict[str, Any]: + """Return a JSON-safe deterministic dict. Field order alphabetised + by the sort_keys=True at the JSONL boundary.""" + return observation.as_dict() + + +def format_articulation_observation_jsonl( + observation: ArticulationObservation, +) -> str: + """One deterministic JSONL line (sort_keys; no trailing newline).""" + return json.dumps( + observation.as_dict(), + sort_keys=True, + separators=(",", ":"), + default=str, + ) + + +def prompt_hash(prompt: str) -> str: + """Stable 16-char prompt fingerprint. + + Lowercased + whitespace-collapsed so two presentations of the + same logical prompt collapse to one hash. Hashing means the + raw prompt never has to be persisted — privacy-respecting and + storage-cheap. + """ + canonical = " ".join((prompt or "").strip().lower().split()) + return hashlib.sha256(canonical.encode("utf-8")).hexdigest()[:16] + + +# --------------------------------------------------------------------------- +# Sink protocol +# --------------------------------------------------------------------------- + + +class ArticulationObservationSink(Protocol): + """Append-only JSONL sink. Structurally identical to + ``chat.telemetry.TurnEventSink`` but kept as a distinct named + type so consumers can subscribe to articulation observations + without seeing the broader turn-event stream.""" + + def emit(self, line: str) -> None: ... + + +# --------------------------------------------------------------------------- +# Loader (for the offline miner) +# --------------------------------------------------------------------------- + + +def load_articulation_observations( + lines: Iterable[str], +) -> tuple[ArticulationObservation, ...]: + """Parse a JSONL stream back into ``ArticulationObservation``s. + + Lines that fail to parse are SKIPPED — a malformed line in the + middle of a long stream must not bring down the miner. Caller + can re-parse manually if strict parsing is needed. + """ + out: list[ArticulationObservation] = [] + for raw in lines: + raw = raw.strip() + if not raw: + continue + try: + payload = json.loads(raw) + except json.JSONDecodeError: + continue + try: + out.append( + ArticulationObservation( + turn_id=int(payload["turn_id"]), + anchor_subject=str(payload["anchor_subject"]), + prompt_hash=str(payload["prompt_hash"]), + plan_substrate_hash=str(payload["plan_substrate_hash"]), + metrics=dict(payload["metrics"]), + findings=tuple( + dict(f) for f in payload["findings"] + ), + ) + ) + except (KeyError, TypeError, ValueError): + # Schema drift / partial record — skip rather than abort. + continue + return tuple(out) + + +__all__ = [ + "ArticulationObservation", + "ArticulationObservationSink", + "format_articulation_observation_jsonl", + "load_articulation_observations", + "prompt_hash", + "serialize_articulation_observation", +] diff --git a/chat/runtime.py b/chat/runtime.py index e9ed663b..62a16771 100644 --- a/chat/runtime.py +++ b/chat/runtime.py @@ -540,6 +540,14 @@ class ChatRuntime: # gating discipline as ``_last_plan_findings``: requires # ``config.discourse_contemplation`` + an engaged planner. self._last_plan_metrics: Any | None = None + # Phase 5 — articulation-observation sink (per-turn JSONL stream + # consumed by the offline ``mine_articulation_observations`` + # miner). Attached via ``attach_articulation_sink``; ``None`` + # by default so the runtime emits nothing until an operator + # opts in. Behaviour mirrors ``attach_telemetry_sink``: + # append-only, fail-fast on sink errors, deterministic JSONL. + self._articulation_sink: Any | None = None + self._articulation_turn_counter: int = 0 @property def session(self) -> SessionContext: @@ -585,6 +593,23 @@ class ChatRuntime: self._telemetry_sink = sink self._telemetry_include_content = bool(include_content) + def attach_articulation_sink(self, sink: Any | None) -> None: + """Phase 5 — attach a sink for per-turn articulation observations. + + ``sink`` must satisfy + ``chat.articulation_telemetry.ArticulationObservationSink`` + (any object with ``def emit(line: str) -> None``). Pass + ``None`` to detach. + + The sink receives one canonical JSONL line per turn that + engages the discourse planner AND has + ``config.discourse_contemplation == True``; non-engaged turns + emit nothing. Lines are byte-identical for byte-equal plans + — the offline miner relies on this for deterministic + aggregation. + """ + self._articulation_sink = sink + def attach_oov_sink(self, sink: Any) -> None: """Phase 2.3 — attach an OOV candidate sink.""" self._oov_sink = sink @@ -1149,6 +1174,52 @@ class ChatRuntime: for m in plan.moves ) new_source = "teaching" if plan_uses_teaching else "pack" + # Phase 5 — emit one articulation observation per engaged turn. + # Gated by both ``discourse_contemplation`` (so metrics + + # findings exist to package) AND the presence of an attached + # sink (so the runtime does no JSON work when nobody is + # listening). Sink errors are NOT swallowed — same fail-fast + # contract as the telemetry sink. + if ( + self._articulation_sink is not None + and self.config.discourse_contemplation + and self._last_plan_metrics is not None + ): + from chat.articulation_telemetry import ( + ArticulationObservation, + format_articulation_observation_jsonl, + prompt_hash, + ) + anchor = plan.anchor() + anchor_subject = ( + anchor.fact.subject + if anchor is not None and anchor.fact is not None + else (plan.intent.subject or "") + ) + import hashlib as _hashlib + plan_substrate_hash = _hashlib.sha256( + plan.to_json().encode("utf-8") + ).hexdigest()[:16] + observation = ArticulationObservation( + turn_id=self._articulation_turn_counter, + anchor_subject=anchor_subject, + prompt_hash=prompt_hash(text), + plan_substrate_hash=plan_substrate_hash, + metrics=self._last_plan_metrics.as_dict(), + findings=tuple( + { + "kind": f.kind.value, + "subject": f.subject, + "predicate": f.predicate, + "object": f.object, + } + for f in self._last_plan_findings + ), + ) + self._articulation_sink.emit( + format_articulation_observation_jsonl(observation) + ) + self._articulation_turn_counter += 1 return rendered, new_source def _stub_response( diff --git a/core/contemplation/miners/articulation_quality.py b/core/contemplation/miners/articulation_quality.py new file mode 100644 index 00000000..7c7aafe6 --- /dev/null +++ b/core/contemplation/miners/articulation_quality.py @@ -0,0 +1,352 @@ +"""Phase 5 — offline articulation-quality miner. + +Consumes a JSONL stream of ``ArticulationObservation`` records (the +per-turn Phase 4 metrics + Phase 3 findings emitted by +``chat.articulation_telemetry``) and aggregates across many turns +to surface ``PACK_MUTATION_CANDIDATE`` findings. + +This is the layer that closes the user-intuited "live reasoning → +memory confidence" loop. Per CLAUDE.md doctrine the aggregation is: + +* **Read-only** — never writes packs, vault, teaching corpus, or + runtime state. Emits findings only. +* **SPECULATIVE-only** — every emitted finding is stamped + ``EpistemicStatus.SPECULATIVE``. The miner proposes corpus + expansions; the operator reviews and decides. +* **Deterministic** — same input stream → byte-identical + findings (same ``substrate_hash``, same ``finding_id`` per + finding). Pinned by ``test_articulation_quality_is_deterministic``. + +v1 rules +-------- + +* ``recurring_predicate_monotony`` — when the SAME ``(anchor_subject, + dominant_predicate)`` pair is flagged ``WEAK_SURFACE`` in + ``>= _MIN_RECURRENCE`` observations, propose substrate expansion + with non-dominant predicates. + +* ``recurring_planner_gap`` — when the SAME ``anchor_subject`` is + flagged ``PLANNER_GAP`` in ``>= _MIN_RECURRENCE`` observations, + propose substrate expansion for that subject. + +* ``low_average_predicate_diversity`` — when the mean + ``predicate_diversity_ratio`` across ``>= _MIN_RECURRENCE`` + observations on the same ``anchor_subject`` falls below + ``_LOW_DIVERSITY_THRESHOLD``, propose substrate diversification. + +The thresholds are conservative on purpose: a single noisy turn must +not produce a pack-mutation proposal. Default ``_MIN_RECURRENCE = 3`` +keeps the bar at "this pattern is the rule, not the exception". +""" + +from __future__ import annotations + +import hashlib +import json +from collections import defaultdict +from pathlib import Path +from statistics import mean +from typing import Iterable + +from chat.articulation_telemetry import ( + ArticulationObservation, + load_articulation_observations, +) +from core.contemplation.schema import ( + ContemplationEvidenceRef, + ContemplationFinding, + FindingKind, +) + + +_MIN_RECURRENCE = 3 +"""Minimum observation count before a pattern proposes a pack +mutation. Tightens the false-positive rate at the cost of catching +slower-burning patterns later.""" + + +_LOW_DIVERSITY_THRESHOLD = 0.5 +"""``predicate_diversity_ratio`` threshold for the +``low_average_predicate_diversity`` rule. ``0.5`` says "on average +half of fact-bearing moves on this subject reused a predicate" — +clearly a corpus-shape signal once it persists across many turns.""" + + +# --------------------------------------------------------------------------- +# Aggregation +# --------------------------------------------------------------------------- + + +def _stream_observations( + paths: Iterable[Path], +) -> tuple[ArticulationObservation, ...]: + """Read every JSONL path in *paths* and return all observations. + + Empty / missing paths skip silently; malformed lines drop via the + loader's per-line try/except (see + ``chat.articulation_telemetry.load_articulation_observations``). + """ + out: list[ArticulationObservation] = [] + for p in paths: + path = Path(p) + if not path.is_file(): + continue + with path.open(encoding="utf-8") as handle: + out.extend(load_articulation_observations(handle)) + return tuple(out) + + +def _evidence_refs_for_observations( + observations: tuple[ArticulationObservation, ...], + *, + summary: str, +) -> tuple[ContemplationEvidenceRef, ...]: + """One evidence ref per source observation, plus a roll-up summary + in the first ref so a reviewer can see the aggregation at a glance. + """ + refs: list[ContemplationEvidenceRef] = [] + for i, obs in enumerate(observations): + refs.append( + ContemplationEvidenceRef( + source_type="articulation_observation", + source_id=obs.plan_substrate_hash, + pointer=f"turn_id={obs.turn_id}", + summary=summary if i == 0 else "", + ) + ) + return tuple(refs) + + +# --------------------------------------------------------------------------- +# Rules +# --------------------------------------------------------------------------- + + +def _rule_recurring_predicate_monotony( + observations: tuple[ArticulationObservation, ...], + substrate_hash: str, +) -> tuple[ContemplationFinding, ...]: + """Detect WEAK_SURFACE recurrence on the same ``(subject, predicate)``.""" + # Map (anchor_subject, dominant_predicate) → list[observation] + buckets: dict[tuple[str, str], list[ArticulationObservation]] = ( + defaultdict(list) + ) + for obs in observations: + for finding in obs.findings: + if finding.get("kind") != FindingKind.WEAK_SURFACE.value: + continue + subject = str(finding.get("subject") or "") + predicate = str(finding.get("object") or "") + if not subject or not predicate: + continue + buckets[(subject, predicate)].append(obs) + + findings: list[ContemplationFinding] = [] + for (subject, predicate), matched in sorted(buckets.items()): + if len(matched) < _MIN_RECURRENCE: + continue + summary = ( + f"WEAK_SURFACE recurred {len(matched)}x on subject={subject!r} " + f"with dominant predicate={predicate!r}" + ) + findings.append( + ContemplationFinding( + kind=FindingKind.PACK_MUTATION_CANDIDATE, + subject=subject, + predicate="recurring_predicate_monotony", + object=predicate, + evidence_refs=_evidence_refs_for_observations( + tuple(matched), summary=summary, + ), + proposed_action=( + f"diversify substrate for {subject!r}: across " + f"{len(matched)} observations the plan repeatedly " + f"over-concentrated on predicate {predicate!r}. " + f"Candidates: add teaching chains rooted on " + f"{subject!r} with relations OTHER than {predicate!r} " + f"(grounds / requires / reveals / contrasts / " + f"precedes / follows) so the planner's RELATION " + f"selector has more variety to draw from." + ), + substrate_hash=substrate_hash, + ) + ) + return tuple(findings) + + +def _rule_recurring_planner_gap( + observations: tuple[ArticulationObservation, ...], + substrate_hash: str, +) -> tuple[ContemplationFinding, ...]: + """Detect PLANNER_GAP recurrence on the same anchor subject.""" + buckets: dict[str, list[ArticulationObservation]] = defaultdict(list) + for obs in observations: + for finding in obs.findings: + if finding.get("kind") != FindingKind.PLANNER_GAP.value: + continue + subject = str(finding.get("subject") or "") + if not subject: + continue + buckets[subject].append(obs) + + findings: list[ContemplationFinding] = [] + for subject, matched in sorted(buckets.items()): + if len(matched) < _MIN_RECURRENCE: + continue + # Collect the distinct modes that hit anchor-only depth so the + # proposed action can reference them concretely. + distinct_modes = sorted({ + str(f.get("object") or "") + for obs in matched + for f in obs.findings + if f.get("kind") == FindingKind.PLANNER_GAP.value + and f.get("subject") == subject + and f.get("object") + }) + summary = ( + f"PLANNER_GAP recurred {len(matched)}x on subject={subject!r} " + f"across modes={distinct_modes}" + ) + findings.append( + ContemplationFinding( + kind=FindingKind.PACK_MUTATION_CANDIDATE, + subject=subject, + predicate="recurring_planner_gap", + object=",".join(distinct_modes) if distinct_modes else None, + evidence_refs=_evidence_refs_for_observations( + tuple(matched), summary=summary, + ), + proposed_action=( + f"widen substrate for {subject!r}: across " + f"{len(matched)} observations the planner could only " + f"surface an anchor (no qualifying support/relation/" + f"transition). Affected modes: " + f"{', '.join(distinct_modes) if distinct_modes else 'unknown'}. " + f"Candidates: add teaching chains rooted on this " + f"lemma, or add pack ``belongs_to`` / ``is_defined_as`` " + f"facts that the SUPPORT selector can pick up." + ), + substrate_hash=substrate_hash, + ) + ) + return tuple(findings) + + +def _rule_low_average_predicate_diversity( + observations: tuple[ArticulationObservation, ...], + substrate_hash: str, +) -> tuple[ContemplationFinding, ...]: + """Detect low mean predicate_diversity_ratio across observations on + the same anchor subject.""" + buckets: dict[str, list[ArticulationObservation]] = defaultdict(list) + for obs in observations: + ratio = obs.metrics.get("predicate_diversity_ratio") + if ratio is None: + continue + if not obs.anchor_subject: + continue + buckets[obs.anchor_subject].append(obs) + + findings: list[ContemplationFinding] = [] + for subject, matched in sorted(buckets.items()): + if len(matched) < _MIN_RECURRENCE: + continue + ratios = [ + float(obs.metrics["predicate_diversity_ratio"]) + for obs in matched + ] + avg = mean(ratios) + if avg >= _LOW_DIVERSITY_THRESHOLD: + continue + summary = ( + f"mean predicate_diversity_ratio={avg:.3f} across " + f"{len(matched)} observations on subject={subject!r}" + ) + findings.append( + ContemplationFinding( + kind=FindingKind.PACK_MUTATION_CANDIDATE, + subject=subject, + predicate="low_average_predicate_diversity", + object=f"{avg:.3f}", + evidence_refs=_evidence_refs_for_observations( + tuple(matched), summary=summary, + ), + proposed_action=( + f"raise predicate diversity for {subject!r}: across " + f"{len(matched)} observations the mean " + f"predicate_diversity_ratio was {avg:.3f} (threshold " + f"{_LOW_DIVERSITY_THRESHOLD:.2f}). Candidates: " + f"add teaching chains rooted on {subject!r} that " + f"use predicates currently under-represented in the " + f"corpus; consider auditing which relations the " + f"planner is forced to repeat." + ), + substrate_hash=substrate_hash, + ) + ) + return tuple(findings) + + +# --------------------------------------------------------------------------- +# Public entry point +# --------------------------------------------------------------------------- + + +def _substrate_hash_for_observations( + observations: tuple[ArticulationObservation, ...], +) -> str: + """Deterministic hash over the canonical concatenation of each + observation's JSONL serialisation.""" + payload = json.dumps( + [obs.as_dict() for obs in observations], + sort_keys=True, + separators=(",", ":"), + default=str, + ) + return hashlib.sha256(payload.encode("utf-8")).hexdigest()[:16] + + +def mine_articulation_observations( + observations: tuple[ArticulationObservation, ...] | None = None, + *, + paths: Iterable[Path | str] = (), +) -> tuple[ContemplationFinding, ...]: + """Run every articulation-quality rule across the input observations. + + Provide either *observations* directly OR *paths* (JSONL files + that will be loaded via + ``chat.articulation_telemetry.load_articulation_observations``). + When BOTH are provided, the direct observations are appended + after the loaded ones in canonical order. + + Pure deterministic function: same input → byte-identical findings. + """ + loaded = _stream_observations(tuple(Path(p) for p in paths)) + if observations is None: + all_observations = loaded + else: + all_observations = loaded + tuple(observations) + + if not all_observations: + return () + + substrate_hash = _substrate_hash_for_observations(all_observations) + + findings: list[ContemplationFinding] = [] + findings.extend( + _rule_recurring_predicate_monotony(all_observations, substrate_hash) + ) + findings.extend( + _rule_recurring_planner_gap(all_observations, substrate_hash) + ) + findings.extend( + _rule_low_average_predicate_diversity( + all_observations, substrate_hash, + ) + ) + return tuple(findings) + + +__all__ = [ + "mine_articulation_observations", +] diff --git a/tests/test_articulation_quality_e2e.py b/tests/test_articulation_quality_e2e.py new file mode 100644 index 00000000..1aaee0b9 --- /dev/null +++ b/tests/test_articulation_quality_e2e.py @@ -0,0 +1,192 @@ +"""Phase 5 — end-to-end test of the full articulation-quality loop. + +Demonstrates the doctrine-aligned feedback loop the user asked for: + + live runtime (Phase 1-4) + → per-turn ArticulationObservation + → JSONL sink + → offline mine_articulation_observations + → SPECULATIVE PACK_MUTATION_CANDIDATE findings + +No mutation of packs, vault, teaching corpus, or runtime state at any +step. Operator reviews the emitted findings via the existing +proposal-review-ratify chain. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import List + +from chat.articulation_telemetry import load_articulation_observations +from chat.runtime import ChatRuntime +from core.config import RuntimeConfig +from core.contemplation.miners.articulation_quality import ( + mine_articulation_observations, +) +from core.contemplation.schema import FindingKind +from teaching.epistemic import EpistemicStatus + + +@dataclass +class _BufferSink: + """Minimal in-memory ``ArticulationObservationSink``.""" + lines: List[str] = field(default_factory=list) + + def emit(self, line: str) -> None: + self.lines.append(line) + + +# --------------------------------------------------------------------------- +# No sink attached → runtime emits nothing +# --------------------------------------------------------------------------- + + +def test_no_sink_means_no_emission() -> None: + """Engaged plan + contemplation on + sink ABSENT → no JSONL line.""" + rt = ChatRuntime(config=RuntimeConfig(discourse_contemplation=True)) + rt.chat("What is truth, and why does it matter?") + # No buffer to inspect — verify the planner engaged so the + # condition for emission was met EXCEPT for the missing sink. + assert rt.last_plan_metrics is not None + assert rt.last_plan_findings # multi-move compound prompt fires WEAK_SURFACE + + +# --------------------------------------------------------------------------- +# Sink attached + contemplation off → no emission +# --------------------------------------------------------------------------- + + +def test_sink_attached_but_contemplation_off_yields_nothing() -> None: + sink = _BufferSink() + rt = ChatRuntime(config=RuntimeConfig(discourse_contemplation=False)) + rt.attach_articulation_sink(sink) + rt.chat("What is truth, and why does it matter?") + # Contemplation off → metrics None → emission gate fails closed. + assert sink.lines == [] + + +# --------------------------------------------------------------------------- +# Sink attached + contemplation on + planner engaged → one line emitted +# --------------------------------------------------------------------------- + + +def test_engaged_turn_emits_one_observation_line() -> None: + sink = _BufferSink() + rt = ChatRuntime(config=RuntimeConfig(discourse_contemplation=True)) + rt.attach_articulation_sink(sink) + rt.chat("What is truth, and why does it matter?") + assert len(sink.lines) == 1 + [observation] = load_articulation_observations(sink.lines) + assert observation.anchor_subject == "truth" + assert observation.metrics["move_count"] >= 4 + assert any( + f["kind"] == FindingKind.WEAK_SURFACE.value + for f in observation.findings + ) + + +def test_brief_turn_does_not_emit() -> None: + """BRIEF mode prompts short-circuit the planner before any plan + is built — no observation should land in the sink.""" + sink = _BufferSink() + rt = ChatRuntime(config=RuntimeConfig(discourse_contemplation=True)) + rt.attach_articulation_sink(sink) + rt.chat("What is knowledge?") + assert sink.lines == [] + + +# --------------------------------------------------------------------------- +# Multiple turns + offline miner closes the loop +# --------------------------------------------------------------------------- + + +def test_full_loop_emits_pack_mutation_candidate_after_repeated_pattern() -> None: + """The headline Phase 5 demo: + + 1. Operator runs the compound prompt three times. + 2. Each turn emits one observation; all three observations + carry a ``WEAK_SURFACE`` finding for + ``(truth, belongs_to)`` because the plan structure is + deterministic. + 3. Offline miner aggregates the three observations and emits + one ``PACK_MUTATION_CANDIDATE`` finding stamped + SPECULATIVE. + """ + sink = _BufferSink() + rt = ChatRuntime(config=RuntimeConfig(discourse_contemplation=True)) + rt.attach_articulation_sink(sink) + + for _ in range(3): + # Fresh runtime per turn to keep determinism clean; otherwise + # vault state would diverge between turns. (The sink survives + # across the three runtimes — we re-attach.) + rt = ChatRuntime(config=RuntimeConfig(discourse_contemplation=True)) + rt.attach_articulation_sink(sink) + rt.chat("What is truth, and why does it matter?") + + assert len(sink.lines) == 3 + + observations = load_articulation_observations(sink.lines) + findings = mine_articulation_observations(observations=observations) + + # At minimum the recurring_predicate_monotony rule must fire — three + # identical WEAK_SURFACE findings on (truth, belongs_to). + pmc = [ + f for f in findings + if f.kind is FindingKind.PACK_MUTATION_CANDIDATE + ] + assert pmc, "expected at least one PACK_MUTATION_CANDIDATE finding" + + monotony = [ + f for f in pmc if f.predicate == "recurring_predicate_monotony" + ] + assert len(monotony) == 1 + assert monotony[0].subject == "truth" + assert monotony[0].object == "belongs_to" + assert monotony[0].epistemic_status is EpistemicStatus.SPECULATIVE + + +# --------------------------------------------------------------------------- +# Determinism across the full loop +# --------------------------------------------------------------------------- + + +def test_full_loop_is_deterministic_byte_equal_finding_ids() -> None: + """Two end-to-end runs over the same input produce byte-identical + finding IDs — the load-bearing claim for the offline miner.""" + + def _run_loop() -> tuple[str, ...]: + sink = _BufferSink() + for _ in range(3): + rt = ChatRuntime( + config=RuntimeConfig(discourse_contemplation=True), + ) + rt.attach_articulation_sink(sink) + rt.chat("What is truth, and why does it matter?") + observations = load_articulation_observations(sink.lines) + findings = mine_articulation_observations(observations=observations) + return tuple(f.finding_id for f in findings) + + ids_a = _run_loop() + ids_b = _run_loop() + assert ids_a == ids_b + + +# --------------------------------------------------------------------------- +# Doctrine pin — every emitted finding is SPECULATIVE +# --------------------------------------------------------------------------- + + +def test_full_loop_emits_only_speculative_findings() -> None: + sink = _BufferSink() + for _ in range(3): + rt = ChatRuntime( + config=RuntimeConfig(discourse_contemplation=True), + ) + rt.attach_articulation_sink(sink) + rt.chat("What is truth, and why does it matter?") + observations = load_articulation_observations(sink.lines) + findings = mine_articulation_observations(observations=observations) + for f in findings: + assert f.epistemic_status is EpistemicStatus.SPECULATIVE diff --git a/tests/test_articulation_quality_miner.py b/tests/test_articulation_quality_miner.py new file mode 100644 index 00000000..08437cf4 --- /dev/null +++ b/tests/test_articulation_quality_miner.py @@ -0,0 +1,307 @@ +"""Phase 5 — articulation-quality miner unit tests. + +Tests ``core.contemplation.miners.articulation_quality. +mine_articulation_observations`` against synthetic observations: + + * Empty stream → no findings + * Single observation → no findings (threshold not met) + * recurring_predicate_monotony — fires at >= _MIN_RECURRENCE + * recurring_planner_gap — fires at >= _MIN_RECURRENCE + * low_average_predicate_diversity — fires when mean < threshold + * Determinism: byte-equal finding IDs across two runs + * All emitted findings stay SPECULATIVE +""" + +from __future__ import annotations + +import pytest + +from chat.articulation_telemetry import ( + ArticulationObservation, + format_articulation_observation_jsonl, + load_articulation_observations, +) +from core.contemplation.miners.articulation_quality import ( + mine_articulation_observations, +) +from core.contemplation.schema import FindingKind +from teaching.epistemic import EpistemicStatus + + +def _obs( + *, + turn_id: int = 0, + anchor: str = "truth", + prompt_h: str = "p0000000000000000", + plan_h: str = "s0000000000000000", + metrics: dict | None = None, + findings: tuple[dict, ...] = (), +) -> ArticulationObservation: + return ArticulationObservation( + turn_id=turn_id, + anchor_subject=anchor, + prompt_hash=prompt_h, + plan_substrate_hash=plan_h, + metrics=metrics or { + "move_count": 4, + "fact_bearing_count": 4, + "anchor_count": 1, + "support_count": 1, + "relation_count": 1, + "transition_count": 1, + "closure_count": 0, + "unique_predicates": 4, + "unique_subjects": 1, + "unique_sources": 2, + "topic_shift_count": 0, + "pronominalization_opportunities": 3, + "predicate_diversity_ratio": 1.0, + "subject_focus_ratio": 1.0, + }, + findings=findings, + ) + + +def _weak_surface_finding( + subject: str, predicate: str, +) -> dict[str, str | None]: + return { + "kind": FindingKind.WEAK_SURFACE.value, + "subject": subject, + "predicate": "predicate_repeats_in_plan", + "object": predicate, + } + + +def _planner_gap_finding( + subject: str, mode: str = "explain", +) -> dict[str, str | None]: + return { + "kind": FindingKind.PLANNER_GAP.value, + "subject": subject, + "predicate": "anchor_only_depth", + "object": mode, + } + + +# --------------------------------------------------------------------------- +# Trivial cases +# --------------------------------------------------------------------------- + + +def test_empty_stream_yields_no_findings() -> None: + assert mine_articulation_observations(observations=()) == () + + +def test_below_threshold_recurrence_yields_no_findings() -> None: + """Two ``WEAK_SURFACE`` observations is below the default + ``_MIN_RECURRENCE = 3`` — nothing should fire.""" + observations = ( + _obs(turn_id=0, findings=(_weak_surface_finding("truth", "belongs_to"),)), + _obs(turn_id=1, findings=(_weak_surface_finding("truth", "belongs_to"),)), + ) + findings = mine_articulation_observations(observations=observations) + assert findings == () + + +# --------------------------------------------------------------------------- +# Rule: recurring_predicate_monotony +# --------------------------------------------------------------------------- + + +def test_recurring_predicate_monotony_fires_at_threshold() -> None: + observations = tuple( + _obs(turn_id=i, findings=(_weak_surface_finding("truth", "belongs_to"),)) + for i in range(3) + ) + findings = mine_articulation_observations(observations=observations) + matching = [ + f for f in findings + if f.predicate == "recurring_predicate_monotony" + ] + assert len(matching) == 1 + f = matching[0] + assert f.kind is FindingKind.PACK_MUTATION_CANDIDATE + assert f.subject == "truth" + assert f.object == "belongs_to" + assert "diversify substrate" in f.proposed_action + assert f.epistemic_status is EpistemicStatus.SPECULATIVE + + +def test_recurring_predicate_monotony_separates_by_subject() -> None: + """Two different subjects each above threshold → two separate + findings, not one merged finding.""" + observations = ( + *( + _obs(turn_id=i, anchor="truth", + findings=(_weak_surface_finding("truth", "belongs_to"),)) + for i in range(3) + ), + *( + _obs(turn_id=i + 100, anchor="memory", + findings=(_weak_surface_finding("memory", "requires"),)) + for i in range(3) + ), + ) + findings = mine_articulation_observations(observations=observations) + matching = [ + f for f in findings + if f.predicate == "recurring_predicate_monotony" + ] + assert len(matching) == 2 + by_subject = {f.subject: f for f in matching} + assert "truth" in by_subject and by_subject["truth"].object == "belongs_to" + assert "memory" in by_subject and by_subject["memory"].object == "requires" + + +# --------------------------------------------------------------------------- +# Rule: recurring_planner_gap +# --------------------------------------------------------------------------- + + +def test_recurring_planner_gap_fires_at_threshold() -> None: + observations = tuple( + _obs(turn_id=i, anchor="rare_lemma", + findings=(_planner_gap_finding("rare_lemma", "explain"),)) + for i in range(3) + ) + findings = mine_articulation_observations(observations=observations) + matching = [ + f for f in findings + if f.predicate == "recurring_planner_gap" + ] + assert len(matching) == 1 + assert matching[0].subject == "rare_lemma" + assert "widen substrate" in matching[0].proposed_action + + +def test_recurring_planner_gap_collects_distinct_modes() -> None: + """When the same subject hits PLANNER_GAP across different modes, + the finding's ``object`` lists all of them, sorted.""" + observations = ( + _obs(turn_id=0, anchor="rare", + findings=(_planner_gap_finding("rare", "explain"),)), + _obs(turn_id=1, anchor="rare", + findings=(_planner_gap_finding("rare", "paragraph"),)), + _obs(turn_id=2, anchor="rare", + findings=(_planner_gap_finding("rare", "example"),)), + ) + findings = mine_articulation_observations(observations=observations) + matching = [ + f for f in findings if f.predicate == "recurring_planner_gap" + ] + assert len(matching) == 1 + assert matching[0].object == "example,explain,paragraph" + + +# --------------------------------------------------------------------------- +# Rule: low_average_predicate_diversity +# --------------------------------------------------------------------------- + + +def test_low_average_predicate_diversity_fires_below_threshold() -> None: + low_metrics = dict( + move_count=6, fact_bearing_count=6, + anchor_count=1, support_count=2, relation_count=2, + transition_count=1, closure_count=0, + unique_predicates=2, unique_subjects=1, unique_sources=1, + topic_shift_count=0, pronominalization_opportunities=5, + predicate_diversity_ratio=2.0 / 6.0, # 0.333 — well below 0.5 + subject_focus_ratio=1.0, + ) + observations = tuple( + _obs(turn_id=i, anchor="truth", metrics=low_metrics) + for i in range(3) + ) + findings = mine_articulation_observations(observations=observations) + matching = [ + f for f in findings + if f.predicate == "low_average_predicate_diversity" + ] + assert len(matching) == 1 + f = matching[0] + assert f.kind is FindingKind.PACK_MUTATION_CANDIDATE + assert f.subject == "truth" + # object is the average ratio as a string, formatted to 3 decimals + assert f.object is not None + assert float(f.object) == pytest.approx(2.0 / 6.0, abs=1e-3) + + +def test_low_average_predicate_diversity_skips_when_above_threshold() -> None: + high_metrics = dict( + move_count=4, fact_bearing_count=4, + anchor_count=1, support_count=1, relation_count=2, + transition_count=0, closure_count=0, + unique_predicates=4, unique_subjects=1, unique_sources=2, + topic_shift_count=0, pronominalization_opportunities=3, + predicate_diversity_ratio=1.0, + subject_focus_ratio=1.0, + ) + observations = tuple( + _obs(turn_id=i, anchor="truth", metrics=high_metrics) + for i in range(5) + ) + findings = mine_articulation_observations(observations=observations) + assert not [ + f for f in findings + if f.predicate == "low_average_predicate_diversity" + ] + + +# --------------------------------------------------------------------------- +# Determinism +# --------------------------------------------------------------------------- + + +def test_miner_is_deterministic_across_runs() -> None: + observations = tuple( + _obs(turn_id=i, findings=(_weak_surface_finding("truth", "belongs_to"),)) + for i in range(3) + ) + a = mine_articulation_observations(observations=observations) + b = mine_articulation_observations(observations=observations) + assert tuple(f.finding_id for f in a) == tuple(f.finding_id for f in b) + assert tuple(f.substrate_hash for f in a) == tuple(f.substrate_hash for f in b) + + +# --------------------------------------------------------------------------- +# SPECULATIVE doctrine pin +# --------------------------------------------------------------------------- + + +def test_all_findings_remain_speculative() -> None: + observations = ( + *( + _obs(turn_id=i, + findings=(_weak_surface_finding("truth", "belongs_to"),)) + for i in range(3) + ), + *( + _obs(turn_id=i + 100, anchor="rare", + findings=(_planner_gap_finding("rare", "explain"),)) + for i in range(3) + ), + ) + findings = mine_articulation_observations(observations=observations) + assert findings # at least the two recurring rules fired + for f in findings: + assert f.epistemic_status is EpistemicStatus.SPECULATIVE + + +# --------------------------------------------------------------------------- +# Round-trip via JSONL +# --------------------------------------------------------------------------- + + +def test_jsonl_round_trip_preserves_observation_identity() -> None: + original = _obs( + turn_id=42, + anchor="truth", + findings=(_weak_surface_finding("truth", "belongs_to"),), + ) + line = format_articulation_observation_jsonl(original) + [recovered] = load_articulation_observations([line]) + assert recovered.turn_id == original.turn_id + assert recovered.anchor_subject == original.anchor_subject + assert recovered.metrics == original.metrics + assert recovered.findings == original.findings