diff --git a/src/episteme/_profile_history.py b/src/episteme/_profile_history.py new file mode 100644 index 0000000..208bae7 --- /dev/null +++ b/src/episteme/_profile_history.py @@ -0,0 +1,326 @@ +"""Operator-profile axis-change history — Cognitive Arm A · Item 1 +(CP-TEMPORAL-INTEGRITY-EXPANSION-01 first slice; Event 82). + +Append-only hash-chained record of operator-profile axis changes. Lives +at ``~/.episteme/memory/reflective/profile_history.jsonl`` and uses the +existing CP7 ``cp7-chained-v1`` envelope schema (see +``core/hooks/_chain.py``). + +## Why this exists + +The kernel's operator profile (`core/memory/global/operator_profile.md`) +encodes 16 cognitive-style axes as YAML in-place. When an axis is +re-elicited, inferred-to-elicited, or value-shifted, the OLD claim is +overwritten — the *trajectory* is lost. The Phase 12 audit detects +drift (axis-claim diverges from observed behavior); the operator +re-elicits or revises; but the journey from old-claim to new-claim has +historically been preserved only in the axis's `note` field as prose, +which doesn't compose into auditable trajectory data. + +This module fixes the gap. Every meaningful axis change can be recorded +as a chain entry with old_value → new_value, the reason for the change, +and optional evidence_refs (e.g., Event numbers that supported the +re-elicitation). Future audits can walk the trajectory of any axis +across its lifetime. + +## Schema + +Single payload type: + +``` +{"type": "profile_axis_change", + "axis_name": "", + "old_value": "", + "new_value": "", + "reason": "<≥15 chars, no lazy tokens>", + "recorded_at": "", + "recorder": "", + "evidence_refs": ["Event 65", ...]} +``` + +Old / new values are FREE-FORM strings — operators may record +"inferred:loss-averse@2026-04-13" or "20% stop-condition rate" or +whatever shape captures the trajectory honestly. The history is +documentary, not enum-strict. + +## Validation discipline + +- `axis_name` must be one of the 16 declared operator-profile axes. +- `reason` must be ≥ 15 chars + must NOT match the lazy-token list + (mirrors `_profile_audit_ack.py` discipline). +- `old_value` and `new_value` must be strings. + +## Auto-instrumentation status + +This module ships the API + CLI for **manual** trajectory recording. +**Auto-instrumentation** of profile-write paths (so every CLI-driven +profile change emits a history entry without operator action) is +deferred to a follow-up Event. Operators use the `episteme history axis + --record` CLI to backfill or record manually until then. + +Spec: ``~/episteme-private/docs/cp-v1.1-architectural.md`` +§ CP-TEMPORAL-INTEGRITY-EXPANSION-01 Item 1. +""" +from __future__ import annotations + +import os +import subprocess +import sys +from datetime import datetime, timezone +from pathlib import Path +from typing import Iterable + +# Locate core/hooks/_chain.py — same lazy-import pattern other src/episteme/ +# library modules use for hook-tier modules. +_REPO_ROOT = Path(__file__).resolve().parent.parent.parent +_CORE_HOOKS_DIR = _REPO_ROOT / "core" / "hooks" +if str(_CORE_HOOKS_DIR) not in sys.path: + sys.path.insert(0, str(_CORE_HOOKS_DIR)) + +import _chain # type: ignore # pyright: ignore[reportMissingImports] + + +# --------------------------------------------------------------------------- +# Constants +# --------------------------------------------------------------------------- + + +DEFAULT_REFLECTIVE_DIR = Path.home() / ".episteme" / "memory" / "reflective" +HISTORY_FILENAME = "profile_history.jsonl" + + +# Valid axis names from `kernel/OPERATOR_PROFILE_SCHEMA.md` v2 schema. +# Must match the YAML keys in `core/memory/global/operator_profile.md`. +VALID_AXIS_NAMES: frozenset[str] = frozenset({ + # Process axes (§ 4a) + "planning_strictness", + "risk_tolerance", + "testing_rigor", + "parallelism_preference", + "documentation_rigor", + "automation_level", + # Cognitive-style axes (§ 4b) + "dominant_lens", + "noise_signature", + "abstraction_entry", + "decision_cadence", + "explanation_depth", + "feedback_mode", + "uncertainty_tolerance", + "asymmetry_posture", + "fence_discipline", + # Expertise map (§ 4c) + "expertise_map", +}) + + +# Mirrors `_profile_audit_ack.py:LAZY_RATIONALE_TOKENS`. Reasoning Surface +# validator's lazy-token discipline applied to the reason field of a +# profile-axis-change record. +LAZY_REASON_TOKENS: frozenset[str] = frozenset({ + # English shortforms + "n/a", "na", "tbd", "todo", + "none", "nothing", "nil", "null", + "ack", "acked", "acknowledged", + "ok", "okay", "fine", + "later", "fix later", "do later", "address later", + "wip", "in progress", + # Korean equivalents + "해당 없음", "없음", "없다", "추후", "나중에", +}) + +MIN_REASON_CHARS = 15 + + +# --------------------------------------------------------------------------- +# Validation +# --------------------------------------------------------------------------- + + +def validate_axis_name(axis_name) -> None: + """Reject empty / non-string / non-schema-axis axis_name. Strict + against the v2 schema's 16-axis enumeration.""" + if not isinstance(axis_name, str): + raise ValueError("axis_name must be a string") + stripped = axis_name.strip() + if not stripped: + raise ValueError("axis_name must be a non-empty string") + if stripped not in VALID_AXIS_NAMES: + raise ValueError( + f"unknown axis_name {axis_name!r}. Must be one of the 16 " + f"declared axes in kernel/OPERATOR_PROFILE_SCHEMA.md. " + f"Use `episteme history axis --list` to see valid axes." + ) + + +def validate_reason(text) -> None: + """Lazy-token + min-char rejection. Mirrors `_profile_audit_ack.py: + validate_rationale` discipline — a reason without substance defeats + the purpose of the trajectory record.""" + if not isinstance(text, str): + raise ValueError("reason must be a string") + stripped = text.strip() + lowered = stripped.lower() + # Lazy-token check first: a lazy token of any length should report + # as lazy, not as too-short. + for token in LAZY_REASON_TOKENS: + if lowered == token.lower(): + raise ValueError( + f"reason matches lazy-token {token!r}. " + f"Provide a substantive reason — what triggered the change?" + ) + if len(stripped) < MIN_REASON_CHARS: + raise ValueError( + f"reason must be at least {MIN_REASON_CHARS} characters; " + f"got {len(stripped)}. Provide a substantive reason — what " + f"triggered the change? (Empty / placeholder reasons rejected.)" + ) + + +def _validate_value(value, field_name: str) -> None: + if not isinstance(value, str): + raise ValueError(f"{field_name} must be a string") + if not value.strip(): + raise ValueError(f"{field_name} must be a non-empty string") + + +# --------------------------------------------------------------------------- +# Recorder identity resolution (same pattern as _profile_audit_ack.py) +# --------------------------------------------------------------------------- + + +def _resolve_recorder() -> str: + """Default recorder identity. Resolution order: + EPISTEME_RECORDER env var → USER env var → git config user.name → 'unknown'.""" + explicit = os.environ.get("EPISTEME_RECORDER", "").strip() + if explicit: + return explicit + user = os.environ.get("USER", "").strip() + if user: + return user + try: + result = subprocess.run( + ["git", "config", "--get", "user.name"], + capture_output=True, text=True, timeout=2, + ) + if result.returncode == 0 and result.stdout.strip(): + return result.stdout.strip() + except (OSError, subprocess.SubprocessError): + pass + return "unknown" + + +# --------------------------------------------------------------------------- +# Path resolution +# --------------------------------------------------------------------------- + + +def _resolve_path(reflective_dir: Path | None = None) -> Path: + base = reflective_dir or DEFAULT_REFLECTIVE_DIR + return base / HISTORY_FILENAME + + +# --------------------------------------------------------------------------- +# Write path +# --------------------------------------------------------------------------- + + +def record_change( + axis_name: str, + old_value: str, + new_value: str, + reason: str, + *, + evidence_refs: Iterable[str] | None = None, + recorder: str | None = None, + reflective_dir: Path | None = None, + _now: datetime | None = None, # test seam +) -> dict: + """Append a ``profile_axis_change`` envelope to the history stream + and return the full chain envelope. + + Raises ValueError on invalid axis_name (must be one of the 16 + declared schema axes), invalid reason (lazy-token / too-short), or + non-string old_value / new_value. + """ + validate_axis_name(axis_name) + _validate_value(old_value, "old_value") + _validate_value(new_value, "new_value") + validate_reason(reason) + now = _now or datetime.now(timezone.utc) + + payload = { + "type": "profile_axis_change", + "axis_name": axis_name, + "old_value": old_value.strip(), + "new_value": new_value.strip(), + "reason": reason.strip(), + "recorded_at": now.isoformat(), + "recorder": recorder or _resolve_recorder(), + "evidence_refs": list(evidence_refs) if evidence_refs else [], + } + return _chain.append(_resolve_path(reflective_dir), payload) + + +# --------------------------------------------------------------------------- +# Read paths +# --------------------------------------------------------------------------- + + +def walk_axis_history( + axis_name: str, + *, + reflective_dir: Path | None = None, +) -> list[dict]: + """Return all envelopes for ``axis_name``, in chronological (chain) + order. Returns empty list if no history file or no entries for the axis. + + Filters out non-`profile_axis_change` payloads (defensive — the + stream is single-payload-type by design but the filter ensures + forward-compat with future payload types).""" + validate_axis_name(axis_name) + path = _resolve_path(reflective_dir) + if not path.exists(): + return [] + + entries: list[dict] = [] + for envelope in _chain.iter_records(path, verify=True): + payload = envelope.get("payload", {}) + if not isinstance(payload, dict): + continue + if payload.get("type") != "profile_axis_change": + continue + if payload.get("axis_name") != axis_name: + continue + entries.append(envelope) + return entries + + +def list_axes_with_history(*, reflective_dir: Path | None = None) -> set[str]: + """Return set of axis_names that have at least one recorded change.""" + path = _resolve_path(reflective_dir) + if not path.exists(): + return set() + axes: set[str] = set() + for envelope in _chain.iter_records(path, verify=True): + payload = envelope.get("payload", {}) + if not isinstance(payload, dict): + continue + if payload.get("type") != "profile_axis_change": + continue + axis = payload.get("axis_name") + if isinstance(axis, str): + axes.add(axis) + return axes + + +# --------------------------------------------------------------------------- +# Chain verification (delegates to _chain) +# --------------------------------------------------------------------------- + + +def verify_chain(reflective_dir: Path | None = None): + """Return ``_chain.ChainVerdict`` for the profile_history stream. + Used by ``episteme chain verify`` to integrate the history stream + into the Pillar 2 verification surface.""" + return _chain.verify_chain(_resolve_path(reflective_dir)) diff --git a/src/episteme/cli.py b/src/episteme/cli.py index 9c78a7a..7237365 100644 --- a/src/episteme/cli.py +++ b/src/episteme/cli.py @@ -3034,6 +3034,104 @@ def _profile_audit_cli(*, since: str, write: bool, as_json: bool) -> int: return 0 +def _profile_history_cli(args) -> int: + """CLI entry for `episteme history axis` (Cognitive Arm A · Item 1 / Event 82). + + Three modes dispatched by args: + - `axis --list`: enumerate axes that have at least one history entry. + - `axis --record --from "..." --to "..." --reason "..."`: record a change. + - `axis `: walk + render the chronological trajectory for the axis. + """ + from episteme import _profile_history as ph_mod + + history_action = getattr(args, "history_action", None) + if history_action != "axis": + print( + f"unknown history action: {history_action!r} " + "(expected: axis)", + file=sys.stderr, + ) + return 2 + + if getattr(args, "list_axes", False): + axes = ph_mod.list_axes_with_history() + if not axes: + print("No profile axes have recorded history yet.") + return 0 + print(f"Profile axes with recorded history ({len(axes)}):") + for axis in sorted(axes): + history = ph_mod.walk_axis_history(axis) + print(f" {axis:30s} {len(history)} entr{'y' if len(history) == 1 else 'ies'}") + return 0 + + axis_name = getattr(args, "axis_name", None) + if not axis_name: + print( + "axis_name is required (or pass --list to enumerate axes with history).", + file=sys.stderr, + ) + return 2 + + record = getattr(args, "record", False) + if record: + old_value = getattr(args, "from_value", None) + new_value = getattr(args, "to_value", None) + reason = getattr(args, "reason", None) + evidence_refs = getattr(args, "evidence_refs", None) or [] + + if not old_value or not new_value or not reason: + print( + "--record requires --from, --to, and --reason " + "(min 15 chars; lazy tokens like 'n/a' / 'tbd' rejected).", + file=sys.stderr, + ) + return 2 + try: + envelope = ph_mod.record_change( + axis_name, + old_value=old_value, + new_value=new_value, + reason=reason, + evidence_refs=evidence_refs, + ) + except ValueError as exc: + print(f"error: {exc}", file=sys.stderr) + return 1 + print(f"Recorded axis change for {axis_name}.") + print(f" entry_hash: {envelope['entry_hash']}") + print(f" recorder: {envelope['payload'].get('recorder', 'unknown')}") + if evidence_refs: + print(f" evidence: {', '.join(evidence_refs)}") + return 0 + + # Default: walk + render + try: + history = ph_mod.walk_axis_history(axis_name) + except ValueError as exc: + print(f"error: {exc}", file=sys.stderr) + return 1 + if not history: + print(f"No recorded history for axis {axis_name!r}.") + print(" Record a change with: " + f"`episteme history axis {axis_name} --record --from \"...\" --to \"...\" --reason \"...\"`") + return 0 + print(f"Trajectory for {axis_name} ({len(history)} entr{'y' if len(history) == 1 else 'ies'}):") + print() + for envelope in history: + payload = envelope.get("payload", {}) + print(f" recorded_at: {payload.get('recorded_at', '?')}") + print(f" recorder: {payload.get('recorder', '?')}") + print(f" old_value: {payload.get('old_value', '?')}") + print(f" new_value: {payload.get('new_value', '?')}") + print(f" reason: {payload.get('reason', '?')}") + evidence = payload.get("evidence_refs") or [] + if evidence: + print(f" evidence: {', '.join(evidence)}") + print(f" entry_hash: {envelope.get('entry_hash', '?')}") + print() + return 0 + + def _profile_audit_ack_cli(args) -> int: """CLI entry for `episteme profile audit ack` (CP-AUDIT-ACK-01 / Event 78). @@ -3971,6 +4069,13 @@ def _chain_dispatch(args) -> int: ack_verdict = _ack_mod.verify_chain() except Exception: # noqa: BLE001 — degrade gracefully ack_verdict = None + # CP-TEMPORAL-INTEGRITY-EXPANSION-01 Item 1 / Event 82 — include + # the profile-history stream in the chain-verify enumeration. + try: + from episteme import _profile_history as _ph_mod + history_verdict = _ph_mod.verify_chain() + except Exception: # noqa: BLE001 — degrade gracefully + history_verdict = None all_intact = True for stream_name, verdict in ( ("protocols", fw.get("protocols")), @@ -3978,6 +4083,7 @@ def _chain_dispatch(args) -> int: ("pending_contracts", pc), ("pending_contracts_archive", pc_arch), ("profile_audit_acks", ack_verdict), + ("profile_history", history_verdict), ): if verdict is None: continue @@ -4829,6 +4935,57 @@ def build_parser() -> argparse.ArgumentParser: help="Stream to upgrade (only `protocols` has legacy records to upgrade at CP7)", ) + # CP-TEMPORAL-INTEGRITY-EXPANSION-01 Item 1 + Item 5 / Event 82 — unified + # `episteme history` CLI; Cognitive Arm A first slice covers `axis` + # subcommand. Future Events add `protocol` and `surface` subcommands. + history_cmd = sub.add_parser( + "history", + help="Walk the supersede-with-history record streams (Cognitive Arm A)", + ) + history_sub = history_cmd.add_subparsers(dest="history_action", required=True) + p_h_axis = history_sub.add_parser( + "axis", + help="Walk profile axis change history (or --record / --list)", + ) + p_h_axis.add_argument( + "axis_name", + nargs="?", + help="One of the 16 declared operator-profile axes (omit when using --list)", + ) + p_h_axis.add_argument( + "--list", + dest="list_axes", + action="store_true", + help="List all axes that have recorded history", + ) + p_h_axis.add_argument( + "--record", + action="store_true", + help="Record a new axis change entry (requires --from / --to / --reason)", + ) + p_h_axis.add_argument( + "--from", + dest="from_value", + help="Prior value (free-form string; e.g., 'inferred:loss-averse@2026-04-13')", + ) + p_h_axis.add_argument( + "--to", + dest="to_value", + help="New value (free-form string; e.g., 'elicited:loss-averse@2026-04-27 with lived-behavior')", + ) + p_h_axis.add_argument( + "--reason", + help="Substantive reason for the change (min 15 chars; lazy tokens 'n/a' / 'tbd' / etc. rejected)", + ) + p_h_axis.add_argument( + "--evidence-refs", + dest="evidence_refs", + nargs="*", + default=[], + metavar="REF", + help="Optional event/episode references (e.g. 'Event 65' 'Event 66')", + ) + # CP-CHAIN-RECOVERY-PROTOCOL-01 / Event 80 — unified recovery surface # covering reset (functional), selective (stub), migrate (stub). c_recover = chain_sub.add_parser( @@ -5144,6 +5301,8 @@ def main(argv: Iterable[str] | None = None) -> int: force=args.force, ) return 0 + if args.command == "history": + return _profile_history_cli(args) if args.command == "profile": if args.profile_action == "show": return _profile_show() diff --git a/tests/test_profile_history.py b/tests/test_profile_history.py new file mode 100644 index 0000000..5c9b4d7 --- /dev/null +++ b/tests/test_profile_history.py @@ -0,0 +1,211 @@ +"""Tests for CP-TEMPORAL-INTEGRITY-EXPANSION-01 Item 1 (Event 82) — +profile axis history hash-chained stream at +~/.episteme/memory/reflective/profile_history.jsonl. + +Coverage: +- axis_name validation (must be one of 16 declared schema axes) +- reason validation (lazy-token + min-char rejection) +- old_value / new_value validation (must be non-empty strings) +- record_change writes valid cp7-chained-v1 envelope +- walk_axis_history returns chronological trajectory for axis +- list_axes_with_history returns set of axes with at least one entry +- chain integrity across multiple writes +""" +from __future__ import annotations + +import tempfile +import unittest +from pathlib import Path + +from episteme import _profile_history as ph + + +class ValidateAxisNameTests(unittest.TestCase): + def test_valid_axis_accepted(self): + ph.validate_axis_name("asymmetry_posture") + ph.validate_axis_name("planning_strictness") + ph.validate_axis_name("expertise_map") + # Should NOT raise + + def test_unknown_axis_rejected(self): + with self.assertRaises(ValueError) as ctx: + ph.validate_axis_name("not_a_real_axis") + self.assertIn("unknown axis_name", str(ctx.exception)) + + def test_empty_axis_rejected(self): + with self.assertRaises(ValueError): + ph.validate_axis_name("") + with self.assertRaises(ValueError): + ph.validate_axis_name(" ") + + def test_non_string_axis_rejected(self): + with self.assertRaises(ValueError): + ph.validate_axis_name(None) # type: ignore[arg-type] + with self.assertRaises(ValueError): + ph.validate_axis_name(123) # type: ignore[arg-type] + + +class ValidateReasonTests(unittest.TestCase): + def test_lazy_token_n_a_rejected(self): + with self.assertRaises(ValueError) as ctx: + ph.validate_reason("n/a") + self.assertIn("lazy-token", str(ctx.exception)) + + def test_lazy_token_korean_rejected(self): + with self.assertRaises(ValueError): + ph.validate_reason("해당 없음") + + def test_short_reason_rejected(self): + with self.assertRaises(ValueError) as ctx: + ph.validate_reason("too short") + self.assertIn("at least", str(ctx.exception)) + + def test_substantive_reason_accepted(self): + ph.validate_reason("Re-elicited after lived-behavior closure across Events 65-67.") + # Should NOT raise + + def test_non_string_reason_rejected(self): + with self.assertRaises(ValueError): + ph.validate_reason(None) # type: ignore[arg-type] + + +class RecordChangeTests(unittest.TestCase): + def test_record_change_writes_valid_envelope(self): + with tempfile.TemporaryDirectory() as td: + envelope = ph.record_change( + "asymmetry_posture", + old_value="inferred:loss-averse@2026-04-13", + new_value="elicited:loss-averse@2026-04-27 with lived-behavior", + reason="Re-elicit; lived-behavior closure across Events 65-67.", + evidence_refs=["Event 65", "Event 66", "Event 67"], + recorder="testuser", + reflective_dir=Path(td), + ) + self.assertEqual(envelope["schema_version"], "cp7-chained-v1") + payload = envelope["payload"] + self.assertEqual(payload["type"], "profile_axis_change") + self.assertEqual(payload["axis_name"], "asymmetry_posture") + self.assertEqual(payload["old_value"], "inferred:loss-averse@2026-04-13") + self.assertEqual(payload["new_value"], "elicited:loss-averse@2026-04-27 with lived-behavior") + self.assertEqual(payload["recorder"], "testuser") + self.assertEqual(payload["evidence_refs"], ["Event 65", "Event 66", "Event 67"]) + self.assertIn("recorded_at", payload) + self.assertTrue(envelope["entry_hash"].startswith("sha256:")) + + def test_record_change_invalid_axis_rejected(self): + with tempfile.TemporaryDirectory() as td: + with self.assertRaises(ValueError): + ph.record_change( + "fake_axis", + "old", "new", + "Substantive reason text here.", + reflective_dir=Path(td), + ) + + def test_record_change_invalid_reason_rejected(self): + with tempfile.TemporaryDirectory() as td: + with self.assertRaises(ValueError): + ph.record_change( + "asymmetry_posture", + "old", "new", + "tbd", # lazy + reflective_dir=Path(td), + ) + + def test_record_change_empty_value_rejected(self): + with tempfile.TemporaryDirectory() as td: + with self.assertRaises(ValueError): + ph.record_change( + "asymmetry_posture", + "", "new", # empty old + "Substantive reason text here.", + reflective_dir=Path(td), + ) + with self.assertRaises(ValueError): + ph.record_change( + "asymmetry_posture", + "old", "", # empty new + "Substantive reason text here.", + reflective_dir=Path(td), + ) + + +class WalkAxisHistoryTests(unittest.TestCase): + def test_walk_returns_empty_when_no_file(self): + with tempfile.TemporaryDirectory() as td: + self.assertEqual( + ph.walk_axis_history("asymmetry_posture", reflective_dir=Path(td)), + [], + ) + + def test_walk_returns_chronological_trajectory(self): + with tempfile.TemporaryDirectory() as td: + d = Path(td) + ph.record_change( + "asymmetry_posture", + "inferred", "elicited:loss-averse@2026-04-13", + "Initial elicitation from cognitive_profile evidence.", + reflective_dir=d, + ) + ph.record_change( + "asymmetry_posture", + "elicited:loss-averse@2026-04-13", + "elicited:loss-averse@2026-04-27 with lived-behavior", + "Re-elicit after Events 65-67 closed the audit drift.", + evidence_refs=["Event 65", "Event 66", "Event 67"], + reflective_dir=d, + ) + history = ph.walk_axis_history("asymmetry_posture", reflective_dir=d) + self.assertEqual(len(history), 2) + self.assertEqual(history[0]["payload"]["old_value"], "inferred") + self.assertEqual(history[1]["payload"]["new_value"], "elicited:loss-averse@2026-04-27 with lived-behavior") + + def test_walk_filters_other_axes(self): + with tempfile.TemporaryDirectory() as td: + d = Path(td) + ph.record_change( + "asymmetry_posture", + "old1", "new1", + "Substantive reason text here.", + reflective_dir=d, + ) + ph.record_change( + "fence_discipline", + "old2", "new2", + "Different axis change reason here.", + reflective_dir=d, + ) + asymmetry_history = ph.walk_axis_history("asymmetry_posture", reflective_dir=d) + self.assertEqual(len(asymmetry_history), 1) + self.assertEqual(asymmetry_history[0]["payload"]["axis_name"], "asymmetry_posture") + + +class ListAxesWithHistoryTests(unittest.TestCase): + def test_list_returns_distinct_axes(self): + with tempfile.TemporaryDirectory() as td: + d = Path(td) + ph.record_change("asymmetry_posture", "a", "b", "Substantive reason text.", reflective_dir=d) + ph.record_change("fence_discipline", "c", "d", "Substantive reason text.", reflective_dir=d) + ph.record_change("asymmetry_posture", "b", "e", "Substantive reason text.", reflective_dir=d) + axes = ph.list_axes_with_history(reflective_dir=d) + self.assertEqual(axes, {"asymmetry_posture", "fence_discipline"}) + + def test_list_empty_when_no_file(self): + with tempfile.TemporaryDirectory() as td: + self.assertEqual(ph.list_axes_with_history(reflective_dir=Path(td)), set()) + + +class ChainIntegrityTests(unittest.TestCase): + def test_chain_intact_after_multiple_writes(self): + with tempfile.TemporaryDirectory() as td: + d = Path(td) + ph.record_change("planning_strictness", "v1", "v2", "Substantive reason.", reflective_dir=d) + ph.record_change("risk_tolerance", "v1", "v2", "Substantive reason.", reflective_dir=d) + ph.record_change("asymmetry_posture", "v1", "v2", "Substantive reason.", reflective_dir=d) + verdict = ph.verify_chain(reflective_dir=d) + self.assertTrue(verdict.intact) + self.assertEqual(verdict.total_entries, 3) + + +if __name__ == "__main__": + unittest.main()