core-labs/core
core-labs/core
1
0
Fork 0
Code Issues 11 Pull requests 2 Projects Releases Packages Wiki Activity Actions

feat(contemplation): Phase 5 — articulation-quality miner closes the loop

Browse source 
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.
This commit is contained in:
Shay 2026-05-21 10:55:39 -07:00
parent 1740b7d518
commit 327047ce26
5 changed files with 1120 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

198
chat/articulation_telemetry.py Normal file
View file

@ -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": <FindingKind.value>, "subject": <str>,
"predicate": <str>, "object": <str|None>}``. 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",
]

71
chat/runtime.py
View file

@ -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(

352
core/contemplation/miners/articulation_quality.py Normal file
View file

@ -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",
]

192
tests/test_articulation_quality_e2e.py Normal file
View file

@ -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

307
tests/test_articulation_quality_miner.py Normal file
View file

@ -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