#!/usr/bin/env python3
"""bodhi-state — deterministic writer for BodhiKit tracking files.

Skills decide WHAT happened (the pedagogical judgment); this script performs
the file mutations (the mechanical part). It owns: Leitner box math, the
bloomLevel ratchet, the consecutiveCorrectAtL4Plus counter, sessionHistory
type vocabulary, unknown-field preservation, the mastery formula, the
prerequisite gate verdict, and the v2->v3 spaced-review migration.

Canonical shapes: skills/state-schema/SKILL.md. Canonical intervals and
update rules: skills/spaced-repetition/SKILL.md. This script implements
those KBs; if they change, change this script in the same PR (dev/check.sh
pins the constants).

No dependencies beyond the Python 3 standard library.

All subcommands print a JSON result to stdout. Exit 0 = success, 1 = error
(human-readable message on stderr plus JSON error object on stdout when
possible), 2 = bad usage.
"""

import argparse
import datetime
import json
import os
import re
import shutil
import sys
import tempfile

try:
    import fcntl  # POSIX only; locking degrades gracefully elsewhere
except ImportError:  # pragma: no cover
    fcntl = None

# --- Canonical constants (mirrors spaced-repetition KB; lint pins these) ---

BOX_INTERVALS = {1: 1, 2: 3, 3: 7, 4: 14, 5: 30}
MAX_BOX = 5

# Canonical sessionHistory[].type vocabulary (state-ops KB).
SESSION_TYPES = {
    "spaced-review", "quiz", "targeted-reteach", "diagnostic-after-gap",
    "learner-forget", "learner-park", "pair", "practice", "evaluate", "other",
}

CONFIDENCE_VALUES = {"sure", "mostly", "guessing"}

# Evidence kinds (state-schema KB): what shows the learner can use a concept.
# A project or module with no kind recorded is `code`.
EVIDENCE_KINDS = ("code", "problems", "writing", "performed")
# How applied work was seen: `observed` (the tutor read it in the project; the
# absent default, never written), `inspected` (a photo or transcript the
# learner shared), `reported` (the learner's word).
APPLIED_BASES = ("observed", "inspected", "reported")
# The kinds whose applied work the tutor cannot see: the two unseen bases
# count toward the gate and Solid there and nowhere else.
UNSEEN_BASIS_KINDS = ("performed",)

# Learner-facing rendering of a Bloom level (blooms-taxonomy KB, canonical).
# The number is an instructor-facing instrument. A learner reads the OUTCOME
# clause on its own; the LABEL is spoken only at a rung-crossing
# (record-review's crossedLevel) and in /progress's legend. Every emitter that
# returns a bloomLevel also returns bloomLabel/bloomOutcome, so skills render
# and never translate.
BLOOM_LABELS = {0: None, 1: "Remember", 2: "Understand", 3: "Apply",
                4: "Analyze", 5: "Evaluate", 6: "Create"}
BLOOM_OUTCOMES = {
    0: "nothing observed yet",
    1: "you can recall the terms and what they refer to",
    2: "you can explain it in your own words",
    3: "you can apply it to a real case with some guidance",
    4: "you can work out why it works as it does and find where it breaks, on your own",
    5: "you can weigh the alternatives and defend a choice between them",
    6: "you can create something new with it and teach it",
}


def bloom_render(level):
    """{bloomLabel, bloomOutcome} for a level; clamps junk to the 0..6 scale."""
    try:
        lvl = int(level)
    except (TypeError, ValueError):
        lvl = 0
    lvl = max(0, min(6, lvl))
    return {"bloomLabel": BLOOM_LABELS[lvl], "bloomOutcome": BLOOM_OUTCOMES[lvl]}


def bloom_scale():
    """The full ladder, for dashboards that render a legend."""
    return [{"level": n, "label": BLOOM_LABELS[n], "outcome": BLOOM_OUTCOMES[n]}
            for n in range(1, 7)]

# Prerequisite-gate recency window (1.11.0): a bloomLevel >= 3 concept only
# auto-satisfies the gate if its retention evidence is current — box >= 3 OR
# reviewed within this many days. Otherwise it is "stale" and the gate offers
# a quick reconfirm instead of a free pass.
GATE_RECENCY_DAYS = 30

# lastActivity length guidance (state-ops KB: "one short sentence (<=120
# chars)"). One home for the value: touch-state truncates to it, verify warns
# above it, and the message quotes it. Before 1.14.x these were three different
# numbers — truncate at 120, warn above 160, message said 120 — so a 140-char
# lastActivity passed silently and a 165-char one was told the limit was 120.
LAST_ACTIVITY_MAX = 120

PROFILE_COUNTERS = {
    "totalSessions", "totalExercises", "totalConceptsLearned",
    "totalMilestonesReached", "totalProjects", "teachBacksWritten",
    "teachBacksPublished",
}

# Required fields on .bodhi-profile.projects.json list entries (state-schema
# KB). Since 1.16.0 the profile-* subcommands own these mutations and construct
# schema-complete entries; verify additionally backstops the shape so a
# fallback-path hand-edit that drops a field cannot pass silently.
PROFILE_ACTIVE_FIELDS = {
    "name", "topic", "startedAt", "currentPhase", "currentModule",
    "bloomLevel", "pace", "status", "trackPurpose",
}
PROFILE_COMPLETED_FIELDS = {
    "name", "completedAt", "finalBloomLevel", "trackPurpose",
}

# profile-update-patterns threshold (state-schema KB): a sub-topic with this
# many assessment-history entries at Bloom <3 is a persistent challenge; at
# Bloom 4+ a consistent strength. One home for the value — skills cite the
# subcommand, never re-tally assessments in prose.
PATTERNS_MIN_ASSESSMENTS = 3


# --- Helpers ---------------------------------------------------------------

def _today_override():
    """BODHI_TODAY=YYYY-MM-DD pins the script's clock. For the test suite
    (which otherwise races midnight against its own `date.today()`) and for
    date-travel checks of the Leitner schedule. Never set in a real
    session; a malformed value fails loudly rather than silently shifting
    every review date."""
    raw = os.environ.get("BODHI_TODAY")
    if not raw:
        return None
    try:
        return datetime.date.fromisoformat(raw.strip())
    except ValueError:
        die(f"BODHI_TODAY {raw!r} is not an ISO date (YYYY-MM-DD)")


def today():
    return _today_override() or datetime.date.today()


def iso(d):
    return d.isoformat()


def now_iso():
    now = datetime.datetime.now().replace(microsecond=0)
    pinned = _today_override()
    if pinned:
        now = datetime.datetime.combine(pinned, now.time())
    return now.isoformat()


_DATE_RE = re.compile(r"^\d{4}-\d{2}-\d{2}($|T)")


def parse_date(s):
    """YYYY-MM-DD, or an ISO date-time; anything else (ints, prose, junk
    suffixes) is None so callers report it instead of computing on it."""
    if not isinstance(s, str) or not _DATE_RE.match(s):
        return None
    try:
        return datetime.date.fromisoformat(s[:10])
    except ValueError:
        return None


def die(msg, code=1):
    print(json.dumps({"ok": False, "error": msg}))
    print(f"bodhi-state: {msg}", file=sys.stderr)
    sys.exit(code)


_WRITE_NOTES = {}   # merged into the next emit(): e.g. migratedFromVersion


def emit(obj):
    obj.setdefault("ok", True)
    obj.update(_WRITE_NOTES)
    print(json.dumps(obj, indent=2, ensure_ascii=False))


def load_json_raw(path):
    """Raw loader — raises. Used by `verify`, which reports errors itself."""
    with open(path, "r", encoding="utf-8") as f:
        return json.load(f)


def load_json(path, expect=dict):
    """Loader for every other subcommand: clean error instead of a traceback."""
    try:
        data = load_json_raw(path)
    except json.JSONDecodeError as e:
        die(f"{path} is not valid JSON ({e.msg} at line {e.lineno}) — do not "
            f"hand-edit; run `bodhi-state verify` and restore from a backup "
            f"if needed")
    except OSError as e:
        die(f"cannot read {path}: {e}")
    if expect is not None and not isinstance(data, expect):
        die(f"{path} top level is {type(data).__name__}, expected "
            f"{expect.__name__} — the file is structurally broken; run "
            f"`bodhi-state verify`")
    return data


def write_json(path, obj):
    """Atomic write: unique temp file in the same directory, then rename.
    The temp file takes the target's existing mode (or the umask default
    for a new file) — mkstemp creates 0600 and the rename would otherwise
    carry that over, making the learner's tracking files owner-only."""
    fd, tmp = tempfile.mkstemp(dir=os.path.dirname(os.path.abspath(path)),
                               prefix=os.path.basename(path) + ".",
                               suffix=".tmp")
    try:
        try:
            mode = os.stat(path).st_mode & 0o777
        except OSError:
            umask = os.umask(0)
            os.umask(umask)
            mode = 0o666 & ~umask
        try:
            os.chmod(tmp, mode)
        except OSError:
            pass  # a filesystem without modes: keep going, the write matters more
        with os.fdopen(fd, "w", encoding="utf-8") as f:
            json.dump(obj, f, indent=2, ensure_ascii=False)
            f.write("\n")
        os.replace(tmp, path)
    except BaseException:
        try:
            os.unlink(tmp)
        except OSError:
            pass
        raise


_LOCK_HANDLES = []  # held until process exit


def acquire_lock(directory, shared=False):
    """Per-directory lock spanning the whole read-mutate-write run.

    Writers take it exclusive (prevents the two-terminal lost-update race).
    Read-only subcommands take it shared, and only if a writer has already
    created the lock file — a read must never leave a .bodhi-state.lock
    behind that makes a merely-inspected project look touched. POSIX-only;
    on platforms without fcntl the script still works, just without the guard.
    """
    if fcntl is None:
        return
    lock_path = os.path.join(directory, ".bodhi-state.lock")
    try:
        if shared:
            if not os.path.exists(lock_path):
                return
            handle = open(lock_path, "r")
            fcntl.flock(handle, fcntl.LOCK_SH)
        else:
            handle = open(lock_path, "a")
            fcntl.flock(handle, fcntl.LOCK_EX)
        _LOCK_HANDLES.append(handle)
    except OSError:
        pass  # locking is best-effort; never block the actual work on it


def bodhi_dir(project):
    d = os.path.join(project, ".bodhi")
    if not os.path.isdir(d):
        die(f"no .bodhi/ directory under {project!r} — is this a BodhiKit project?")
    return d


def sr_path(project):
    return os.path.join(bodhi_dir(project), "spaced-review.json")


def state_path(project):
    return os.path.join(bodhi_dir(project), "state.json")


_LOADED_VERSION = {}   # spaced-review path -> version found on disk at load
_UPGRADE_FIELDS = {}   # spaced-review path -> per-concept fields the load added

V3_CONCEPT_DEFAULTS = (("bloomLevel", 0), ("feynmanPassed", False),
                       ("consecutiveCorrectAtL4Plus", 0), ("question", ""),
                       ("lastResult", ""), ("reviewHistory", list))


def upgrade_to_v3(data):
    """THE v1/v2 -> v3 upgrade, shared by the load path (in memory, persisted
    by the next write) and `migrate-spaced-review` (explicit). One function
    so both paths produce the same file — before 1.18.x the write path
    added three fields and no marker, migrate added five and a marker, and
    migrate then reported noop on a file the write path had half-upgraded.
    Returns the number of per-concept fields added."""
    added = 0
    for c in data.get("concepts", []):
        if not isinstance(c, dict):
            continue
        for key, default in V3_CONCEPT_DEFAULTS:
            if key not in c:
                c[key] = default() if callable(default) else default
                added += 1
    data.setdefault("sessionHistory", [])
    return added


def is_v3_complete(data):
    return data.get("version") == 3 and all(
        isinstance(c, dict) and all(k in c for k, _ in V3_CONCEPT_DEFAULTS)
        for c in data.get("concepts", []))


def write_migration_marker(bdir, from_version, concepts, fields_added, performed_by):
    """`.bodhi/.migration-1.10.md`, written once by whichever path upgraded
    the file first. Returns the marker path."""
    marker = os.path.join(bdir, ".migration-1.10.md")
    if not os.path.exists(marker):
        with open(marker, "w", encoding="utf-8") as f:
            f.write(f"# Migration to 1.10 — {iso(today())}\n\n"
                    f"Performed by {performed_by}.\n\n"
                    f"- spaced-review.json: v{from_version} -> v3\n"
                    f"- concepts touched: {concepts}\n"
                    f"- fields added: {fields_added}\n"
                    f"- backup: `.bodhi/.pre-1.10-backup/spaced-review.json`\n")
    return marker


def ensure_pre_v3_backup(path):
    """Copy a v1/v2 spaced-review.json to .pre-1.10-backup/ once. Never
    overwrites — the first copy may be the only pre-v3 one."""
    backup_dir = os.path.join(os.path.dirname(path), ".pre-1.10-backup")
    backup = os.path.join(backup_dir, "spaced-review.json")
    if not os.path.exists(backup):
        os.makedirs(backup_dir, exist_ok=True)
        shutil.copyfile(path, backup)
        load_json(backup)  # backup must parse before we mutate the source
    return backup


def write_spaced_review(path, data):
    """Every mutating subcommand's write. A file loaded at v1/v2 is backed up
    before it is stamped v3 — previously any write silently upgraded the file
    and `migrate-spaced-review` then reported noop, so the pre-v3 backup its
    contract promises was never made."""
    loaded = _LOADED_VERSION.get(path, 3)
    if loaded < 3 and os.path.exists(path):
        backup = ensure_pre_v3_backup(path)
        _WRITE_NOTES["migratedFromVersion"] = loaded
        _WRITE_NOTES["backup"] = backup
        _WRITE_NOTES["marker"] = write_migration_marker(
            os.path.dirname(path), loaded, len(data.get("concepts", [])),
            _UPGRADE_FIELDS.get(path, 0), "the first `bodhi-state` write on the file")
    data["version"] = 3
    write_json(path, data)


def load_spaced_review(project, create=False):
    path = sr_path(project)
    if not os.path.exists(path):
        if create:
            return path, {"version": 3, "lastReviewCheck": None,
                          "concepts": [], "sessionHistory": []}
        die(f"{path} does not exist")
    data = load_json(path)
    v = data.get("version", 1)
    _LOADED_VERSION[path] = v if isinstance(v, int) else 1
    # Read-tolerate v1/v2: fill v3 per-concept fields in memory; persisted on
    # the next write (state-migration KB pattern).
    if not isinstance(data.get("concepts", []), list):
        die(f"{path}: concepts is {type(data.get('concepts')).__name__}, "
            f"expected list — run `bodhi-state verify`")
    for i, c in enumerate(data.get("concepts", [])):
        if not isinstance(c, dict):
            die(f"{path}: concepts[{i}] is {type(c).__name__}, expected object "
                f"— run `bodhi-state verify`")
    _UPGRADE_FIELDS[path] = upgrade_to_v3(data)
    validate_spaced_review(data, path)
    return path, data


def _is_int(x):
    return isinstance(x, int) and not isinstance(x, bool)


def _coerce_int(x, lo, hi):
    """The lossless int repair: a numeric string in range. Anything else
    (a bool, prose, a float) is None — normalize never guesses."""
    if isinstance(x, str) and re.fullmatch(r"\s*\d+\s*", x):
        v = int(x)
        if lo <= v <= hi:
            return v
    return None


def _coerce_bool(x):
    """The lossless bool repair: the strings true/false, or the ints 0/1."""
    if isinstance(x, str) and x.strip().lower() in ("true", "false"):
        return x.strip().lower() == "true"
    if _is_int(x) and x in (0, 1):
        return bool(x)
    return None


def concept_shape_errors(c):
    """THE shape table for one spaced-review concept — the single source
    behind load-time validation (dies on the first), `verify` (reports them
    all) and `normalize` (repairs the ones with a lossless fix). Returns
    [(field, actual, expected, repair)] where repair is the coerced value
    or None. Two hand-kept copies of these rules drifted in 1.18.0 (verify
    accepted a bool box that every other subcommand then died on)."""
    errs = []
    name = c.get("name")
    if not isinstance(name, str) or not name.strip():
        errs.append(("name", name, "non-empty string", None))
    b = c.get("box")
    if not (_is_int(b) and 1 <= b <= MAX_BOX):
        errs.append(("box", b, f"int 1-{MAX_BOX}", _coerce_int(b, 1, MAX_BOX)))
    # The v3 fields may be absent on a v1/v2 file that verify inspects
    # without upgrading; the load path fills them first.
    if "bloomLevel" in c:
        bl = c["bloomLevel"]
        if not (_is_int(bl) and 0 <= bl <= 6):
            errs.append(("bloomLevel", bl, "int 0-6", _coerce_int(bl, 0, 6)))
    if "feynmanPassed" in c:
        fp = c["feynmanPassed"]
        if not isinstance(fp, bool):
            errs.append(("feynmanPassed", fp, "true|false", _coerce_bool(fp)))
    if "consecutiveCorrectAtL4Plus" in c:
        cc = c["consecutiveCorrectAtL4Plus"]
        if not (_is_int(cc) and cc >= 0):
            errs.append(("consecutiveCorrectAtL4Plus", cc, "int >= 0",
                         _coerce_int(cc, 0, 10**6)))
    rh = c.get("reviewHistory", [])
    if not isinstance(rh, list):
        errs.append(("reviewHistory", rh, "list", None))
    elif any(not isinstance(h, dict) for h in rh):
        errs.append(("reviewHistory", "…", "list of objects", None))
    return errs


def history_entry_errors(h):
    """Shape table for one reviewHistory entry — the fields readers compute
    on. Review finding 7 (2026-09-07): `verify` passed a history bloomLevel of
    "three" that session-brief then died on with a TypeError, so the Stop
    hook certified a project /teach could not read. Rule: whatever verify
    accepts, every read subcommand must handle. Returns the same
    (field, actual, expected, repair) tuples as concept_shape_errors; dates
    are not here — readers skip an unparseable date, so verify only warns."""
    errs = []
    if h.get("deferred"):
        if "days" in h and not (_is_int(h["days"]) and h["days"] >= 0):
            errs.append(("days", h["days"], "int >= 0", _coerce_int(h["days"], 0, 10**6)))
        return errs
    if "bloomLevel" in h:
        bl = h["bloomLevel"]
        if not (_is_int(bl) and 0 <= bl <= 6):
            errs.append(("bloomLevel", bl, "int 0-6", _coerce_int(bl, 0, 6)))
    if "boxBefore" in h:
        bb = h["boxBefore"]
        if not (_is_int(bb) and 1 <= bb <= MAX_BOX):
            errs.append(("boxBefore", bb, f"int 1-{MAX_BOX}", _coerce_int(bb, 1, MAX_BOX)))
    for flag in ("applied", "retry", "selfReport"):
        if flag in h and not isinstance(h[flag], bool):
            errs.append((flag, h[flag], "true|false", _coerce_bool(h[flag])))
    if "appliedBasis" in h and h["appliedBasis"] not in APPLIED_BASES:
        errs.append(("appliedBasis", h["appliedBasis"], "|".join(APPLIED_BASES),
                     _coerce_choice(h["appliedBasis"], APPLIED_BASES)))
    if "evidence" in h and not isinstance(h["evidence"], str):
        errs.append(("evidence", h["evidence"], "a project-relative path", None))
    return errs


def shape_error_text(name, field, actual, expected, repair):
    """One wording for both paths; names `normalize` only when it can fix it."""
    msg = f"concept {name!r} {field} {actual!r} not {expected}"
    if repair is not None:
        msg += " — run `bodhi-state normalize`"
    return msg


def validate_spaced_review(data, path):
    """Type-check the fields every subcommand computes on, once, at load.
    A string box or a null name used to traceback in some subcommands and
    pass silently through others; now every command fails the same way,
    naming the field and — when there is one — the repair."""
    for i, c in enumerate(data.get("concepts", [])):
        errs = concept_shape_errors(c)
        if errs:
            field, actual, expected, repair = errs[0]
            label = c.get("name") if isinstance(c.get("name"), str) else f"index {i}"
            die(f"{path}: " + shape_error_text(label, field, actual, expected, repair)
                + ("; " if repair is not None else " — fix it by hand; ")
                + "`bodhi-state verify` lists every problem")
        for h in c.get("reviewHistory", []) if isinstance(c.get("reviewHistory"), list) else []:
            if not isinstance(h, dict):
                continue
            herrs = history_entry_errors(h)
            if herrs:
                field, actual, expected, repair = herrs[0]
                label = c.get("name") if isinstance(c.get("name"), str) else f"index {i}"
                die(f"{path}: " + shape_error_text(label, f"reviewHistory {field}",
                                                   actual, expected, repair)
                    + ("; " if repair is not None else " — fix it by hand; ")
                    + "`bodhi-state verify` lists every problem")
    if not isinstance(data.get("sessionHistory"), list):
        die(f"{path}: sessionHistory is not a list — run `bodhi-state verify`")


def note_activity(project, text):
    """Point state.json.lastActivity at what just happened (forget, park).
    No-op when the project has no state.json yet."""
    state = state_if_present(project)
    if state is None:
        return
    sp = state_path(project)
    state["lastActivity"] = text[:LAST_ACTIVITY_MAX]
    write_json(sp, state)


def load_state(project):
    """state.json with the fields the script branches on type-checked; the
    drift patterns `normalize` repairs (dict currentModule, nested session
    bookkeeping) fail here with the repair named instead of a traceback."""
    sp = state_path(project)
    state = load_json(sp)
    for key in ("currentModule", "previousModule", "lastActivity"):
        v = state.get(key)
        if v is not None and not isinstance(v, str):
            die(f"{sp}: {key} is {type(v).__name__}, expected string — "
                f"run `bodhi-state normalize`")
    if not isinstance(state.get("sessionDates", []), list):
        die(f"{sp}: sessionDates is not a list — run `bodhi-state normalize`")
    return sp, state


def require_state(project):
    """state.json for the subcommands that cannot run without it."""
    sp = state_path(project)
    if not os.path.exists(sp):
        die(f"{sp} does not exist")
    return sp, load_state(project)[1]


def state_if_present(project, default=None):
    """state.json for the subcommands that degrade gracefully without it."""
    return load_state(project)[1] if os.path.exists(state_path(project)) else default


def resolve_concepts(data, names, hint=""):
    """The shared preamble of forget/park/defer: no names or an untracked
    name dies; otherwise the concept dicts, in the order given, each
    looked up once."""
    names = [n for n in names if n]
    if not names:
        die("no concepts given (use --concept, repeatable" + hint + ")")
    found = [(n, find_concept(data, n)) for n in names]
    missing = [n for n, c in found if c is None]
    if missing:
        die(f"not tracked: {', '.join(missing)} — resolve names first "
            f"(or add-concept them)")
    return [c for _, c in found]


def find_concept(data, name):
    for c in data.get("concepts", []):
        if c.get("name", "").strip().lower() == name.strip().lower():
            return c
    return None


def new_concept(name, module, question="", bloom=0):
    """bloom: the level a /learn or /assess classification observed (0 = not
    classified). Before this flag every seed entered at 0 even though /learn
    only seeds sub-topics it classified at >= 1, so the assessment's own
    reading was dropped at the door: the gate saw `no-opinion`, /continue saw
    `neverTaught`, and session-brief opened with a pretest on material the
    learner had just been assessed on (review finding 3, 2026-09-07)."""
    t = today()
    return {
        "name": name,
        "module": module,
        "introduced": iso(t),
        "box": 1,
        "nextReview": iso(t + datetime.timedelta(days=BOX_INTERVALS[1])),
        "lastReviewed": None,
        "question": question,
        "lastResult": "",
        "bloomLevel": bloom,
        "feynmanPassed": False,
        "consecutiveCorrectAtL4Plus": 0,
        "reviewHistory": [],
    }


def find_profile(project):
    """Walk up from the project dir looking for .bodhi-profile.json."""
    d = os.path.abspath(project)
    for _ in range(4):
        p = os.path.join(d, ".bodhi-profile.json")
        if os.path.exists(p):
            return p
        parent = os.path.dirname(d)
        if parent == d:
            break
        d = parent
    return None


HISTORY_CAP = 100  # per concept; older entries roll into reviewHistoryArchived


def promotion_hold(concept, t):
    """Why a correct on date `t` earns no box movement, or None when it is a
    spaced recall (spaced-repetition KB). A Leitner interval is earned by
    recalling AFTER the gap, so the box and the L4+ streak move only on a
    DUE review; any other correct is evidence (history, Bloom ratchet,
    applied) but not a spaced success.

    Until 1.23.0 only a second review on the same date was held, so corrects
    on consecutive days climbed Box 1 -> 5 and reached mastery in four days
    with no gap longer than one. Performance right after instruction is a
    poor index of what is retained (Soderstrom & Bjork 2015); the rule makes
    the box measure the delayed recall.

    - "already reviewed today": a same-day miss still demotes (forgetting is
      forgetting); the correct that follows it is the relearning rep and
      holds, exactly as --retry does.
    - "first review": nothing has been spaced yet — a lesson's own check, or
      the first grade on a /learn seed (overdue on paper since the day after
      seeding) is first contact, not a delayed recall.
    - "not yet due": reviewed before nextReview. With no readable
      nextReview (parked, legacy, malformed) the interval is measured from
      lastReviewed; with neither date there is nothing to be early against.
    """
    if any(h.get("date") == iso(t) and not h.get("deferred")
           and h.get("result") in ("correct", "incorrect", "partial")
           for h in concept.get("reviewHistory", [])):
        return "already reviewed today"
    last = parse_date(concept.get("lastReviewed"))
    if last is None and exposure_of(concept) == "seeded":
        return "first review"
    due = parse_date(concept.get("nextReview"))
    if due is None and last is not None:
        due = last + datetime.timedelta(days=BOX_INTERVALS.get(concept.get("box", 1), 1))
    if due is not None and t < due:
        return "not yet due"
    return None


def apply_review(concept, result, tested_bloom, confidence=None, note=None,
                 source=None, retry=False, applied=None, evidence=None):
    """Canonical per-concept update (spaced-repetition + state-schema KBs).

    correct   -> box up one (max 5), nextReview = today + new interval, when
                 the review is due; otherwise evidence only (promotion_hold)
    incorrect -> box 1, nextReview tomorrow, counter reset
    partial   -> box held, nextReview tomorrow (re-test soon), counter reset
                 (1.11.2: not a Leitner demotion, but it breaks the
                 consecutive-correct mastery streak)
    retry     -> successive-relearning rep (1.11.1): history entry only.
                 The Rawson & Dunlosky in-session retry is additional evidence,
                 not a substitute — the original miss's Box-1 demotion and
                 tomorrow's review stand; no box/counter/bloom movement.
    bloomLevel only ever ratchets up (demotion is /forget's job, and even
    /forget demotes the box, not the Bloom classification).
    `applied` marks the entry as demonstrated in work the tutor read (code
    that ran, or a written application on a non-code topic; blooms-taxonomy
    KB *Applied Evidence*) rather than in an explanation or a quiz answer. It is a separate axis from the level: the rubric still sets
    `--tested-bloom`, and the flag is what the gate and the mastery formula
    read for "can the learner actually build with it" (1.20.0). `applied`
    is its basis (APPLIED_BASES); only an unseen one is written, as
    appliedBasis, and `evidence` is the work's project-relative path.
    """
    t = today()
    old_box = concept.get("box", 1)
    old_bloom = concept.get("bloomLevel", 0)
    if result not in ("correct", "incorrect", "partial"):
        die(f"result must be correct|incorrect|partial, got {result!r}")
    box_held = None
    if not retry:
        if result == "correct":
            box_held = promotion_hold(concept, t)
            if box_held is None:
                concept["box"] = min(old_box + 1, MAX_BOX)
                concept["nextReview"] = iso(t + datetime.timedelta(days=BOX_INTERVALS[concept["box"]]))
            elif box_held == "first review":
                # The clock starts at first contact: a /learn seed taught
                # weeks after seeding is overdue on paper, and without this
                # it would stay in `due` for the rest of the day it was taught.
                nr = parse_date(concept.get("nextReview"))
                if nr is None or nr <= t:
                    concept["nextReview"] = iso(t + datetime.timedelta(days=BOX_INTERVALS.get(old_box, 1)))
            if tested_bloom is not None:
                concept["bloomLevel"] = max(concept.get("bloomLevel", 0), tested_bloom)
                if tested_bloom >= 4 and box_held is None:
                    concept["consecutiveCorrectAtL4Plus"] = concept.get("consecutiveCorrectAtL4Plus", 0) + 1
        elif result == "incorrect":
            concept["box"] = 1
            concept["nextReview"] = iso(t + datetime.timedelta(days=1))
            concept["consecutiveCorrectAtL4Plus"] = 0
        elif result == "partial":
            concept["nextReview"] = iso(t + datetime.timedelta(days=1))
            # 1.11.2: a partial retrieval breaks the consecutive-correct
            # mastery streak (state-schema KB) — the box is held, but
            # "3 consecutive correct at L4+" means uninterrupted corrects.
            concept["consecutiveCorrectAtL4Plus"] = 0
    concept["lastReviewed"] = iso(t)
    concept["lastResult"] = note or (f"{result} (relearning retry)" if retry else result)
    # bloomLevel is recorded only when a level was tested: a bare `correct`
    # used to write 0, indistinguishable from "graded at 0".
    entry = {"date": iso(t), "result": result, "boxBefore": old_box}
    if tested_bloom is not None:
        entry["bloomLevel"] = tested_bloom
    if retry:
        entry["retry"] = True
    if confidence:
        entry["confidence"] = confidence
    if source:
        entry["source"] = source
    if applied:
        entry["applied"] = True
    if applied in ("inspected", "reported"):
        entry["appliedBasis"] = applied
    if evidence:
        entry["evidence"] = evidence
    history = concept.setdefault("reviewHistory", [])
    history.append(entry)
    if len(history) > HISTORY_CAP:
        overflow = len(history) - HISTORY_CAP
        archive_history(concept, history[:overflow])
        del history[:overflow]
    new_bloom = concept.get("bloomLevel", 0)
    out = {"concept": concept["name"], "box": f"{old_box} -> {concept['box']}",
            "retry": retry,
            "applied": bool(applied),
            "appliedEvidence": applied_evidence(concept),
            "bloomLevel": new_bloom,
            **bloom_render(new_bloom),
            # 1.14.0: THIS write crossed the Bloom-3 line — the exact condition
            # for /teach's bump-profile totalConceptsLearned duty. Executors
            # previously re-derived it ("if unsure, scan progress.md").
            "crossedBloom3": old_bloom < 3 <= new_bloom,
            "crossedLevel": new_bloom > old_bloom,   # the one moment the label is spoken
            "consecutiveCorrectAtL4Plus": concept.get("consecutiveCorrectAtL4Plus", 0),
            "nextReview": concept["nextReview"]}
    if applied in ("inspected", "reported"):
        out["appliedBasis"] = applied
    if evidence:
        out["evidence"] = evidence
    if box_held:
        out["boxHeld"] = box_held
    return out


# The exercise key the tutor saves before a non-code attempt: worked answers
# for `problems`, the criteria for `writing` (and a performed rep's task).
KEY_FILES = ("answers.md", "criteria.md")


def evidence_path(project, raw, kind, basis):
    """The checked, project-relative path of applied work (state-schema KB
    reviewHistory[].evidence). Observed work off code is the learner's own
    file beside the exercise's key, so neither the key, the brief nor a
    tracking file can stand in for it."""
    root = os.path.realpath(project)
    full = os.path.realpath(raw if os.path.isabs(raw) else os.path.join(root, raw))
    rel = os.path.relpath(full, root)
    if rel == os.pardir or rel.startswith(os.pardir + os.sep):
        die(f"--evidence {raw!r} is outside the project")
    if not os.path.isfile(full):
        die(f"--evidence {raw!r} does not exist in the project")
    parts = rel.split(os.sep)
    if parts[0] == ".bodhi":
        die(f"--evidence {raw!r} is a tracking file, not the learner's work")
    if kind != "code" and basis == "observed":
        name = parts[-1].casefold()
        if name in KEY_FILES:
            die(f"--evidence {raw!r} is the exercise's key, not the learner's work")
        if name == "readme.md":
            die(f"--evidence {raw!r} is the exercise's README (the brief), not the "
                f"learner's work")
        if not any(os.path.isfile(os.path.join(os.path.dirname(full), k)) for k in KEY_FILES):
            die(f"--evidence {raw!r} has no key beside it: save the exercise's "
                f"{' or '.join(KEY_FILES)} in its folder before the attempt")
    return "/".join(parts)


# --- Subcommands ------------------------------------------------------------

def cmd_add_concept(args):
    path, data = load_spaced_review(args.project, create=True)
    if find_concept(data, args.concept):
        emit({"action": "noop", "reason": f"concept {args.concept!r} already tracked"})
        return
    bloom = args.bloom or 0
    if not 0 <= bloom <= 6:
        die(f"--bloom must be 0-6, got {bloom}")
    data.setdefault("concepts", []).append(
        new_concept(args.concept, args.module, args.question or "", bloom))
    write_spaced_review(path, data)
    emit({"action": "added", "concept": args.concept, "module": args.module,
          "box": 1, "nextReview": data["concepts"][-1]["nextReview"],
          "bloomLevel": bloom, **bloom_render(bloom)})


def cmd_record_review(args):
    if args.confidence and args.confidence not in CONFIDENCE_VALUES:
        die(f"confidence must be one of {sorted(CONFIDENCE_VALUES)}")
    path, data = load_spaced_review(args.project, create=True)
    concept = find_concept(data, args.concept)
    created = False
    new_module = False
    if concept is None:
        if not args.module:
            die(f"concept {args.concept!r} not tracked; pass --module to auto-create it")
        known = {str(c.get("module", "")).strip().lower()
                 for c in data.get("concepts", [])}
        st = state_if_present(args.project)
        if st is not None:
            known |= {str(st.get(k) or "").strip().lower()
                      for k in ("currentModule", "previousModule")}
        # A module name nothing else has seen is usually a typo of one that
        # exists; the write still happens (the skill's judgment stands), the
        # flag lets it notice.
        new_module = args.module.strip().lower() not in known
        concept = new_concept(args.concept, args.module, args.question or "")
        data.setdefault("concepts", []).append(concept)
        created = True
    kind = effective_kind(concept)
    if args.evidence is not None and args.applied is None:
        die("--evidence is the path of applied work; pass it with --applied")
    if args.applied is not None and not admits(kind, args.applied):
        die(f"concept {concept['name']!r} is a {kind} concept: --applied "
            f"{args.applied} (the learner's word, or a photo the tutor examined) "
            f"counts only on a performed concept — record it without --applied, "
            f"or as --applied observed with the saved work")
    evidence = None
    if args.evidence is not None:
        evidence = evidence_path(args.project, args.evidence, kind, args.applied)
    elif args.applied == "observed" and kind != "code":
        die(f"--applied observed on a {kind} concept needs --evidence <path>: the "
            f"learner's work saved in the project, beside the exercise's key "
            f"({' or '.join(KEY_FILES)})")
    change = apply_review(concept, args.result, args.tested_bloom,
                          confidence=args.confidence, note=args.note,
                          source=args.source, retry=args.retry,
                          applied=args.applied, evidence=evidence)
    data["lastReviewCheck"] = now_iso()
    write_spaced_review(path, data)
    change["created"] = created
    if created:
        change["newModule"] = new_module
    emit(change)


def cmd_set_feynman(args):
    path, data = load_spaced_review(args.project)
    concept = find_concept(data, args.concept)
    if concept is None:
        die(f"concept {args.concept!r} not tracked")
    concept["feynmanPassed"] = True  # set, never unset
    # 1.23.0: the date of the most recent pass. Mastery counts the
    # explain-back only when it postdates the last miss (feynman_current).
    concept["feynmanPassedAt"] = iso(today())
    write_spaced_review(path, data)
    emit({"concept": concept["name"], "feynmanPassed": True,
          "feynmanPassedAt": concept["feynmanPassedAt"],
          "feynmanCurrent": feynman_current(concept)})


def cmd_record_session(args):
    if args.type not in SESSION_TYPES:
        die(f"type {args.type!r} is not in the canonical vocabulary "
            f"{sorted(SESSION_TYPES)} (state-schema KB). Use 'other' with "
            f"--subtype for genuinely novel sessions.")
    if args.type == "other" and not args.subtype:
        die("type 'other' requires --subtype (state-schema KB)")
    extra = {}
    if args.data:
        try:
            extra = json.loads(args.data)
        except json.JSONDecodeError as e:
            die(f"--data is not valid JSON: {e}")
        if not isinstance(extra, dict):
            die("--data must be a JSON object")
    # Reserved keys come from flags only — --data must not bypass the
    # vocabulary check (1.11.1: a smuggled "type" defeated enforcement and
    # then tripped the Stop hook on the script's own write).
    for reserved in ("type", "subtype", "date"):
        extra.pop(reserved, None)
    path, data = load_spaced_review(args.project, create=True)
    entry = {"date": iso(today()), "type": args.type}
    if args.subtype:
        entry["subtype"] = args.subtype
    entry.update(extra)
    data.setdefault("sessionHistory", []).append(entry)
    write_spaced_review(path, data)
    emit({"action": "session-recorded", "entry": entry})


ASSESSMENT_TRIGGERS = {"learn-phase2", "assess", "evaluate", "plan-regenerate"}


def cmd_record_assessment(args):
    """Append an entry to the append-only assessment-history.json."""
    if args.trigger not in ASSESSMENT_TRIGGERS:
        die(f"trigger must be one of {sorted(ASSESSMENT_TRIGGERS)}")
    try:
        entry = json.loads(args.data)
    except json.JSONDecodeError as e:
        die(f"--data is not valid JSON: {e}")
    if not isinstance(entry, dict):
        die("--data must be a JSON object")
    entry["date"] = entry.get("date") or iso(today())
    entry["trigger"] = args.trigger
    path = os.path.join(bodhi_dir(args.project), "assessment-history.json")
    if os.path.exists(path):
        data = load_json(path)
    else:
        data = {"version": 1, "entries": []}
    data.setdefault("entries", []).append(entry)
    write_json(path, data)
    emit({"action": "assessment-recorded", "trigger": args.trigger,
          "entries": len(data["entries"])})


def cmd_forget(args):
    # --concept (repeatable) is the exact-name path — required for names that
    # themselves contain commas. --concepts remains the comma-list convenience.
    names = list(args.concept or [])
    if args.concepts:
        names.extend(n.strip() for n in args.concepts.split(",") if n.strip())
    path, data = load_spaced_review(args.project)
    concepts = resolve_concepts(data, names, ', or --concepts "a, b"')
    names = [c["name"] for c in concepts]
    t = today()
    box_changes = {}
    for c in concepts:
        box_changes[c["name"]] = f"{c.get('box', 1)} -> 1"
        c["box"] = 1
        c["nextReview"] = iso(t + datetime.timedelta(days=1))
        c["consecutiveCorrectAtL4Plus"] = 0
        # feynmanPassed and bloomLevel preserved by design (state-schema KB).
        c.setdefault("reviewHistory", []).append(
            {"date": iso(t), "result": "incorrect", "selfReport": True,
             "note": "learner-initiated demote"})  # nothing was tested: no bloomLevel
        c["lastReviewed"] = iso(t)
        c["lastResult"] = "learner-initiated demote"
    entry = {"date": iso(t), "type": "learner-forget",
             "conceptsDemoted": names,
             "boxChanges": box_changes}
    if args.note:
        entry["notes"] = args.note
    data.setdefault("sessionHistory", []).append(entry)
    write_spaced_review(path, data)
    note_activity(args.project, args.activity or
                  f"Demoted {len(names)} concept(s): {', '.join(names)}")
    emit({"action": "demoted", "concepts": names, "boxChanges": box_changes})


def cmd_park(args):
    """Take a concept out of review rotation, or return it (--resume).

    /forget --park (1.16.0): a working learner who has
    consciously deprioritized a concept needs "stop scheduling it", not
    "review it tomorrow, harder" — otherwise review rot accumulates on
    concepts they no longer maintain and the due pile stops being trusted.
    Parking is scheduling, never an outcome: box, bloom, counters,
    feynmanPassed, and history all stand. --resume re-enters rotation with a
    review tomorrow and the box preserved.
    """
    path, data = load_spaced_review(args.project)
    concepts = resolve_concepts(data, args.concept or [])
    names = [c["name"] for c in concepts]
    t = today()
    changed = {}
    for c in concepts:
        if args.resume:
            if not c.get("parked"):
                die(f"{c['name']!r} is not parked — nothing to resume")
            c["parked"] = False
            c["nextReview"] = iso(t + datetime.timedelta(days=1))
        else:
            if c.get("parked"):
                die(f"{c['name']!r} is already parked")
            c["parked"] = True
            c["nextReview"] = None
        changed[c["name"]] = c["nextReview"]
    key = "conceptsResumed" if args.resume else "conceptsParked"
    entry = {"date": iso(t), "type": "learner-park",
             key: names}
    if args.note:
        entry["notes"] = args.note
    data.setdefault("sessionHistory", []).append(entry)
    write_spaced_review(path, data)
    verb = "Resumed" if args.resume else "Parked"
    note_activity(args.project, f"{verb} {len(names)} concept(s): {', '.join(names)}")
    emit({"action": "resumed" if args.resume else "parked",
          "concepts": changed})


def cmd_defer(args):
    """Roll a due-but-unreviewed concept forward WITHOUT inventing an outcome.

    1.12.1 — found in the wild: sessions that ran out of time hand-wrote
    result: "skipped" and rolled nextReview by hand. Deferral is scheduling,
    never an outcome: box, bloom, counters, and lastReviewed stay untouched.
    """
    days = args.days if args.days is not None else 1
    if days <= 0:
        die(f"--days must be a positive number of days (got {days})")
    path, data = load_spaced_review(args.project)
    concepts = resolve_concepts(data, args.concept or [])
    t = today()
    rolled = {}
    for c in concepts:
        c["nextReview"] = iso(t + datetime.timedelta(days=days))
        entry = {"date": iso(t), "deferred": True, "days": days}
        if args.note:
            entry["note"] = args.note
        c.setdefault("reviewHistory", []).append(entry)
        rolled[c["name"]] = c["nextReview"]
    write_spaced_review(path, data)
    emit({"action": "deferred", "days": days, "concepts": rolled})


def cmd_touch_state(args):
    sp, state = require_state(args.project)
    t = iso(today())
    dates = state.setdefault("sessionDates", [])
    if not isinstance(dates, list):
        die(f"state.json sessionDates is {type(dates).__name__}, expected a "
            f"list — run `bodhi-state verify`")
    new_session = t not in dates
    profile_bumped = False
    if new_session:
        yesterday = iso(today() - datetime.timedelta(days=1))
        state["currentStreak"] = (state.get("currentStreak", 0) + 1
                                  if yesterday in dates else 1)
        dates.append(t)
        state["totalSessions"] = state.get("totalSessions", 0) + 1
        # 1.11.1: the script owns the cross-project session counter too —
        # the first touch-state of the day bumps it, so it no longer depends
        # on which skill in the chain happens to run touch-state first.
        pp = find_profile(args.project)
        if pp:
            profile = load_json(pp)
            stats = profile.setdefault("cumulativeStats", {})
            stats["totalSessions"] = stats.get("totalSessions", 0) + 1
            profile["lastUpdated"] = now_iso()
            write_json(pp, profile)
            profile_bumped = True
    state["lastSessionAt"] = now_iso()
    if args.activity:
        state["lastActivity"] = args.activity[:LAST_ACTIVITY_MAX]
    if args.module:
        if state.get("currentModule") and state["currentModule"] != args.module:
            state["previousModule"] = state["currentModule"]
        state["currentModule"] = args.module
    if args.module_index is not None:
        state["currentModuleIndex"] = args.module_index
    if args.phase:
        state["currentPhase"] = args.phase
    if args.completion is not None:
        state["overallCompletion"] = max(0, min(100, args.completion))
    state["version"] = 2
    write_json(sp, state)
    emit({"action": "state-updated", "newSession": new_session,
          "currentStreak": state.get("currentStreak"),
          "totalSessions": state.get("totalSessions"),
          "profileSessionsBumped": profile_bumped,
          "lastActivity": state.get("lastActivity")})


def cmd_set_evidence_kind(args):
    """Record the project's evidence kind, or one module's override
    (state-schema KB). Writes state.json only: every review is kept, and a
    kind changes how reviews are read, never what they say."""
    if args.clear and args.kind is not None:
        die("pass --kind to record a kind or --clear to remove one, not both")
    if not args.clear and args.kind is None:
        die(f"--kind is required: one of {'|'.join(EVIDENCE_KINDS)} (or --clear)")
    if args.kind is not None and args.kind not in EVIDENCE_KINDS:
        die(f"kind {args.kind!r} is not one of {'|'.join(EVIDENCE_KINDS)}")
    sp, state = require_state(args.project)
    kinds = state.get("moduleEvidenceKinds", {})
    if not isinstance(kinds, dict):
        die(f"{sp}: moduleEvidenceKinds is {type(kinds).__name__}, expected an "
            f"object — fix it by hand; `bodhi-state verify` lists every problem")
    key = None
    if args.module is not None:
        _, plan = read_plan_modules(args.project)
        by_id = {m["id"]: norm_module_name(m["name"]) for m in plan}
        key = by_id.get(norm_module_name(args.module), norm_module_name(args.module))
        _, data = load_spaced_review(args.project, create=True)
        known = set(by_id.values()) | {norm_module_name(c.get("module"))
                                       for c in data.get("concepts", [])}
        # Clearing must reach an orphan (a renamed module), so only a set
        # is held to the typo guard.
        if not args.clear and key not in known:
            die(f"module {args.module!r} matches no plan module heading and no "
                f"tracked concept's module — `syllabus` lists the plan's names")
    if key is None and args.clear:
        changed = state.pop("evidenceKind", None) is not None
    elif key is None:
        state["evidenceKind"] = args.kind
        changed = True
    else:
        spellings = [k for k in kinds if norm_module_name(k) == key]
        for k in spellings:
            kinds.pop(k)
        if not args.clear:
            kinds[key] = args.kind
        changed = bool(spellings) or not args.clear
        if kinds:
            state["moduleEvidenceKinds"] = kinds
        else:
            state.pop("moduleEvidenceKinds", None)
    if not changed:
        emit({"action": "noop", "reason": "no kind recorded there to clear",
              "module": key})
        return
    write_json(sp, state)
    install_kinds(args.project)
    emit({"action": "evidence-kind-cleared" if args.clear else "evidence-kind-set",
          "module": key, "kind": args.kind,
          "projectKind": _KINDS["project"],
          "projectKindRecorded": "evidenceKind" in state,
          "moduleEvidenceKinds": state.get("moduleEvidenceKinds", {})})


def cmd_bump_profile(args):
    if args.counter not in PROFILE_COUNTERS:
        die(f"counter must be one of {sorted(PROFILE_COUNTERS)}")
    p = find_profile(args.project)
    if not p:
        die("no .bodhi-profile.json found walking up from the project "
            "(it is created by /learn)")
    profile = load_json(p)
    stats = profile.setdefault("cumulativeStats", {})
    stats[args.counter] = stats.get(args.counter, 0) + 1
    profile["lastUpdated"] = now_iso()
    write_json(p, profile)
    emit({"action": "profile-bumped", "counter": args.counter,
          "value": stats[args.counter], "file": p})


# --- Cross-project profile-list ownership (1.16.0) --------------------------
# Until 1.15.x the .bodhi-profile.projects.json list was "the one mutation the
# script does not own" — /learn and /evaluate hand-edited it, guarded only by
# verify's entry-shape backstop. These subcommands close that last
# hand-edit hole: the script constructs schema-complete entries, the skills
# supply only the pedagogical values.

def load_profile_projects(project):
    """Locate .bodhi-profile.projects.json beside .bodhi-profile.json."""
    pp = find_profile(project)
    if not pp:
        die("no .bodhi-profile.json found walking up from the project "
            "(it is scaffolded by /learn) — the projects list lives beside it")
    lp = os.path.join(os.path.dirname(pp), ".bodhi-profile.projects.json")
    if os.path.exists(lp):
        data = load_json(lp)
    else:
        data = {"version": 2, "activeProjects": [], "completedProjects": []}
    for key in ("activeProjects", "completedProjects"):
        if not isinstance(data.get(key), list):
            data[key] = []
        for i, e in enumerate(data[key]):
            if not isinstance(e, dict):
                die(f"{lp}: {key}[{i}] is {type(e).__name__}, expected object "
                    f"— run `bodhi-state verify` and repair the entry")
    return pp, lp, data


def find_project_entry(entries, name):
    want = name.strip().lower()
    return next((e for e in entries if isinstance(e, dict)
                 and str(e.get("name", "")).strip().lower() == want), None)


def cmd_profile_add_project(args):
    pp, lp, data = load_profile_projects(args.project)
    if find_project_entry(data["activeProjects"], args.name):
        die(f"{args.name!r} is already in activeProjects — "
            f"use profile-update-project to refresh it")
    if find_project_entry(data["completedProjects"], args.name):
        die(f"{args.name!r} is already in completedProjects — "
            f"pick a distinct project name")
    entry = {
        "name": args.name,
        "topic": args.topic,
        "startedAt": iso(today()),
        "currentPhase": args.phase or "1",
        "currentModule": args.module or "",
        "bloomLevel": args.bloom if args.bloom is not None else 0,
        "pace": args.pace or "steady",
        "status": args.status or "active",
        "trackPurpose": args.track_purpose or "",
    }
    data["activeProjects"].append(entry)
    data["version"] = 2
    write_json(lp, data)
    emit({"action": "profile-project-added", "entry": entry,
          "activeProjects": len(data["activeProjects"]), "file": lp})


def cmd_profile_update_project(args):
    _, lp, data = load_profile_projects(args.project)
    entry = find_project_entry(data["activeProjects"], args.name)
    if entry is None:
        names = [str(e.get("name")) for e in data["activeProjects"]]
        die(f"{args.name!r} is not in activeProjects {names} — "
            f"add it with profile-add-project first")
    updated = []
    for flag, field in (("topic", "topic"), ("phase", "currentPhase"),
                        ("module", "currentModule"), ("bloom", "bloomLevel"),
                        ("pace", "pace"), ("status", "status"),
                        ("track_purpose", "trackPurpose")):
        value = getattr(args, flag)
        if value is not None:
            entry[field] = value
            updated.append(field)
    if not updated:
        die("nothing to update — pass at least one field flag")
    write_json(lp, data)
    emit({"action": "profile-project-updated", "entry": entry,
          "updatedFields": updated, "file": lp})


def cmd_profile_complete_project(args):
    _, lp, data = load_profile_projects(args.project)
    entry = find_project_entry(data["activeProjects"], args.name)
    if entry is None:
        names = [str(e.get("name")) for e in data["activeProjects"]]
        die(f"{args.name!r} is not in activeProjects {names} — "
            f"nothing to complete")
    data["activeProjects"].remove(entry)
    completed = {
        "name": entry.get("name", args.name),
        "completedAt": iso(today()),
        "finalBloomLevel": (args.final_bloom if args.final_bloom is not None
                            else entry.get("bloomLevel", 0)),
        "trackPurpose": entry.get("trackPurpose", ""),
    }
    if args.status:
        # /learn Phase 1.5(c) replace-archive: "archived: replaced by <new>".
        completed["status"] = args.status
    data["completedProjects"].append(completed)
    write_json(lp, data)
    emit({"action": "profile-project-completed", "entry": completed,
          "activeProjects": len(data["activeProjects"]),
          "completedProjects": len(data["completedProjects"]), "file": lp})


def cmd_profile_update_patterns(args):
    """Append persistent challenges / consistent strengths from assessment
    counts. Pure counting (state-schema KB) — the model never tallies."""
    pp = find_profile(args.project)
    if not pp:
        die("no .bodhi-profile.json found walking up from the project "
            "(it is scaffolded by /learn)")
    hist_path = os.path.join(bodhi_dir(args.project), "assessment-history.json")
    if not os.path.exists(hist_path):
        emit({"action": "profile-patterns-updated", "addedChallenges": [],
              "addedStrengths": [],
              "note": "no assessment-history.json yet — nothing to count"})
        return
    hist = load_json(hist_path)
    low, high = {}, {}
    for e in hist.get("entries", []):
        if not isinstance(e, dict):
            continue
        for st in e.get("subTopics", []):
            if not isinstance(st, dict):
                continue
            name = str(st.get("name", "")).strip()
            bloom = st.get("bloomLevel")
            if not name or not isinstance(bloom, int):
                continue
            if bloom < 3:
                low[name] = low.get(name, 0) + 1
            if bloom >= 4:
                high[name] = high.get(name, 0) + 1
    profile = load_json(pp)
    patterns = profile.setdefault("patterns", {})
    added = {}
    for key, counts in (("persistentChallenges", low),
                        ("consistentStrengths", high)):
        arr = patterns.setdefault(key, [])
        if not isinstance(arr, list):
            arr = patterns[key] = []
        have = {str(x).strip().lower() for x in arr}
        added[key] = sorted(n for n, c in counts.items()
                            if c >= PATTERNS_MIN_ASSESSMENTS
                            and n.strip().lower() not in have)
        arr.extend(added[key])
    if added["persistentChallenges"] or added["consistentStrengths"]:
        profile["lastUpdated"] = now_iso()
        write_json(pp, profile)
    emit({"action": "profile-patterns-updated",
          "addedChallenges": added["persistentChallenges"],
          "addedStrengths": added["consistentStrengths"],
          "counts": {"belowBloom3": low, "bloom4Plus": high},
          "threshold": PATTERNS_MIN_ASSESSMENTS, "file": pp})


def cmd_due(args):
    _, data = load_spaced_review(args.project)
    t = today()
    due, unparseable = [], []
    parked_count = 0
    for c in data.get("concepts", []):
        if c.get("parked") is True:
            # Consciously out of rotation (park) — excluded from the due
            # pile, surfaced as a count so it is never a silent hole.
            parked_count += 1
            continue
        raw = c.get("nextReview")
        nr = parse_date(raw)
        if nr is None:
            # A concept with an unparseable schedule silently leaves rotation
            # forever — surface it, never skip it quietly (1.11.1).
            unparseable.append({"name": c.get("name", "(unnamed)"),
                                "nextReview": raw})
            continue
        if nr <= t:
            # neverTaught / exposure: see exposure_of. A concept /learn seeded
            # from the assessment, or one only ever quizzed below the apply
            # rung, has a schedule but nothing to space — /continue routes
            # these to /teach first. Knowledge shown at the apply rung, or
            # taught, or built, is real review material whatever its source.
            exposure = exposure_of(c)
            # The due list is the surface skills narrate from, so it carries
            # the words a learner may hear (dueSince / overdueDays / the
            # outcome clause) and NOT the box or Bloom number — 1/2 Fable
            # runs read "box: 1" straight into "Query planning — Box 1".
            # Ordering is by box then date; `priority` is the rank.
            due.append({"name": c["name"], "module": c.get("module", ""),
                        "dueSince": raw, "overdueDays": (t - nr).days,
                        "neverTaught": exposure in ("seeded", "quizzed-only"),
                        "exposure": exposure,
                        "bloomOutcome": bloom_render(c.get("bloomLevel", 0))["bloomOutcome"],
                        "question": c.get("question", ""),
                        "evidenceKind": effective_kind(c),
                        "_sort": (c.get("box", 1), raw)})
    due.sort(key=lambda d: d.pop("_sort"))
    for i, d in enumerate(due, 1):
        d["priority"] = i
    out = {"dueToday": len(due),
           "neverTaughtCount": sum(1 for d in due if d["neverTaught"]),
           "concepts": due}
    if parked_count:
        out["parkedCount"] = parked_count
    if args.limit is not None and len(due) > args.limit:
        out["concepts"] = due[:args.limit]
        out["truncated"] = len(due) - args.limit
    if unparseable:
        out["unparseableDates"] = unparseable
    emit(out)


def is_mastered(c):
    """Canonical mastery formula (state-ops KB; field semantics in state-schema).

    Five conjuncts since 1.20.0: the fifth, one correct demonstrated in
    applied work since the last miss, is what keeps "Solid" from being
    earned entirely in conversation. Since 1.23.0 the explain-back must
    also postdate the last miss (feynman_current). Do not redefine inline.
    """
    return (_mastery_core(c)
            and feynman_current(c)
            and applied_evidence(c) >= 1)


def _mastery_core(c):
    """The three retention conjuncts of is_mastered: level, streak, box."""
    return (c.get("bloomLevel", 0) >= 4
            and c.get("consecutiveCorrectAtL4Plus", 0) >= 3
            and c.get("box", 1) >= 4)


def concept_tier(c):
    """Canonical concept tier (blooms-taxonomy KB ladder). Do not re-derive.

    Ordered top-down; first match wins. `mastered` is the predicate name and
    reuses is_mastered so the four-conjunct formula has exactly one home; the
    learner-facing words (Solid/Working/Introduced) are /progress's rendering
    of these keys, not separate logic.

    Implements the KB's `familiar` and `introduced` tiers in code so /progress
    renders a computed tier instead of inferring one from two booleans.
    """
    if c.get("bloomLevel", 0) <= 0:
        return "unclassified"
    if is_mastered(c):
        return "mastered"
    if c.get("bloomLevel", 0) >= 3 and c.get("box", 1) >= 2:
        return "familiar"
    return "introduced"


def due_for_check(c, t):
    """A mastered concept whose last check has aged: overdue by more than
    its box interval, i.e. more than twice the gap the schedule chose.

    It stays mastered (is_mastered does not read the clock): a skipped
    check is unverified, not known forgotten — access fades while storage
    persists and relearning is fast (Bjork & Bjork 1992) — so mastery
    counts, tiers and the anonymized export do not flicker with absence.
    /progress renders it *Solid, due for a check* (1.23.0). Parked and
    unscheduled concepts are out of rotation and never flagged."""
    if c.get("parked") is True or not is_mastered(c):
        return False
    nr = parse_date(c.get("nextReview"))
    return nr is not None and (t - nr).days > BOX_INTERVALS.get(c.get("box", 1), 1)


def new_module_row():
    """Per-module tally shape shared by `mastery` and `snapshot`."""
    return {"concepts": 0, "mastered": 0, "classified": 0, "applied": 0,
            "dueForCheck": 0, "masteredByReport": 0,
            "tiers": {"unclassified": 0, "introduced": 0,
                      "familiar": 0, "mastered": 0}}


def tally_module(m, c, t):
    """Fold one concept into a module row (counts + tier distribution)."""
    m["concepts"] += 1
    if c.get("bloomLevel", 0) > 0:
        m["classified"] += 1
    if is_mastered(c):
        m["mastered"] += 1
        if strongest_basis(c) == "reported":
            m["masteredByReport"] += 1
    if applied_evidence(c) >= 1:
        m["applied"] += 1
    if due_for_check(c, t):
        m["dueForCheck"] += 1
    m["tiers"][concept_tier(c)] += 1


def finalize_module_rows(modules):
    """masteryPct per module; an all-unclassified module is not computable
    (None), never 0% — legacy concepts have no opinion, not a bad one."""
    for m in modules.values():
        m["masteryPct"] = (None if m["classified"] == 0
                           else round(100 * m["mastered"] / m["concepts"]))


def retention_tier(b):
    """spaced-repetition KB 3-tier rollup: Box 4-5 strong, 2-3 building,
    1 needs_review. The one home for these thresholds (lint pins them)."""
    if isinstance(b, int) and b >= 4:
        return "strong"
    if isinstance(b, int) and b >= 2:
        return "building"
    return "needs_review"


def blocked_on_feynman(concepts):
    """is_mastered minus its explain-back conjunct: every other criterion met
    (the applied one included), and /progress names the one step left."""
    return [c["name"] for c in concepts
            if _mastery_core(c)
            and not feynman_current(c)
            and applied_evidence(c) >= 1]


def blocked_on_applied(concepts):
    """is_mastered minus its working-code conjunct: a learner who has
    explained and recalled a concept to every other bar but never built
    with it since the last miss. /progress names the one step left."""
    return [c["name"] for c in concepts
            if _mastery_core(c)
            and feynman_current(c)
            and applied_evidence(c) == 0]


def review_rollup(concepts, t):
    """The per-concept scan shared by `mastery` and `snapshot`: module
    tallies, retention tiers, due windows, box distribution. Parked concepts
    keep their module/mastery standing but leave the retention and due
    surfaces — a parked concept must not nag as "needs review"."""
    modules = {}
    rollup = {"strong": [], "building": [], "needs_review": []}
    due_today, due_week, parked, due_check = [], [], [], []
    boxes = {str(b): 0 for b in range(1, MAX_BOX + 1)}
    overdue10 = unparseable = waiting = unscheduled = 0
    for c in concepts:
        tally_module(modules.setdefault(c.get("module", "(none)"),
                                        new_module_row()), c, t)
        if due_for_check(c, t):
            due_check.append(c["name"])
        if c.get("parked") is True:
            parked.append(c["name"])
            continue
        b = c.get("box", 1)
        boxes[str(b)] += 1
        rollup[retention_tier(b)].append(c["name"])
        nr = parse_date(c.get("nextReview"))
        if nr is None:
            if c.get("nextReview") is not None:
                unparseable += 1
            else:
                unscheduled += 1
            continue
        if nr <= t:
            due_today.append(c["name"])
            if (t - nr).days >= 10:
                overdue10 += 1
        else:
            # Scheduled later than today: nothing to do for it yet. Since
            # 1.23.0 a correct before the due date holds the box, so a
            # learner who only teaches sees no movement; this count is what
            # /progress names so a flat dashboard reads as "waiting", not
            # "stuck".
            waiting += 1
            if nr <= t + datetime.timedelta(days=7):
                due_week.append(c["name"])
    finalize_module_rows(modules)
    for name, row in modules.items():
        row["evidenceKind"] = kind_of_module(name)
    return {"modules": modules, "rollup": rollup, "dueToday": due_today,
            "dueThisWeek": due_week, "parked": parked, "boxes": boxes,
            "dueForCheck": due_check, "waiting": waiting,
            "unscheduled": unscheduled,
            "overdue10": overdue10, "unparseable": unparseable}


def cmd_mastery(args):
    _, data = load_spaced_review(args.project)
    concepts = data.get("concepts", [])
    r = review_rollup(concepts, today())
    out = {"modules": r["modules"],
           "retentionRollup": {k: len(v) for k, v in r["rollup"].items()},
           "retentionConcepts": r["rollup"], "dueToday": r["dueToday"],
           "dueThisWeek": r["dueThisWeek"],
           "blockedOnFeynman": blocked_on_feynman(concepts),
           "blockedOnApplied": blocked_on_applied(concepts),
           "masteredDueForCheck": r["dueForCheck"]}
    out["solidByReport"], out["solidBasis"] = solid_by_report(concepts)
    if r["parked"]:
        out["parked"] = r["parked"]
    emit(out)


def calibration_summary(data):
    """Confidence-vs-outcome calibration from reviewHistory[].confidence.

    Shared by `calibration` and `export-anonymized` — the export strips the
    per-concept event lists (they carry concept names).
    """
    buckets = {c: {"correct": 0, "incorrect": 0, "partial": 0}
               for c in CONFIDENCE_VALUES}
    overconfident_events = []
    underconfident_events = []
    total = 0
    for c in data.get("concepts", []):
        for h in c.get("reviewHistory", []):
            conf = h.get("confidence")
            if conf not in CONFIDENCE_VALUES:
                continue
            res = h.get("result")
            if res not in ("correct", "incorrect", "partial"):
                continue
            buckets[conf][res] += 1
            total += 1
            if conf == "sure" and res == "incorrect":
                overconfident_events.append({"concept": c["name"], "date": h.get("date")})
            if conf == "guessing" and res == "correct":
                underconfident_events.append({"concept": c["name"], "date": h.get("date")})
    def rate(n, d):
        return round(n / d, 2) if d else None
    sure = buckets["sure"]
    guess = buckets["guessing"]
    return {
        "taggedAnswers": total,
        "buckets": buckets,
        "overconfidenceRate": rate(sure["incorrect"], sum(sure.values())),
        "underconfidenceRate": rate(guess["correct"], sum(guess.values())),
        "overconfidentEvents": overconfident_events[-5:],
        "underconfidentEvents": underconfident_events[-5:],
    }


def cmd_calibration(args):
    _, data = load_spaced_review(args.project)
    emit(calibration_summary(data))


# Spacing-gap buckets for retention analysis (name, min days, max days).
GAP_BUCKETS = (
    ("same-day", 0, 0),
    ("1d", 1, 1),
    ("2-3d", 2, 3),
    ("4-7d", 4, 7),
    ("8-14d", 8, 14),
    ("15-30d", 15, 30),
    ("31d+", 31, None),
)


def is_self_report(h):
    """A /forget entry: the learner asked for a reset, nobody tested them.
    Entries written before the `selfReport` field carry the note."""
    return h.get("selfReport") is True or h.get("note") == "learner-initiated demote"


def bucket_for_gap(days):
    for name, lo, hi in GAP_BUCKETS:
        if days >= lo and (hi is None or days <= hi):
            return name
    return None  # negative gap: clock skew or hand-edited dates


def retention_summary(data):
    """Retention-at-review: % correct grouped by actual spacing gap and by
    box-at-review-time (boxBefore, 1.11.3).

    reviewHistory is a longitudinal retention dataset — every due-review
    outcome is a natural experiment on whether the Leitner intervals are
    calibrated. Relearning retries are excluded (same-session reps, not
    spacing evidence); so are self-reports (/forget writes an `incorrect`
    nobody tested — review finding 5: one /forget on a fresh concept read as
    "1 review, 0% recall"). The gap for a concept's first review runs from
    its `introduced` date. `delayedSuccessRate` drops the same-day bucket,
    which is immediate post-instruction performance, not retention.
    `byKind` and `byBasis` split the same counts by the concept's evidence
    kind and by the entry's applied-work basis (`notApplied` without one).
    A practice rep's entry is graded on its debrief, so it counts here.
    """
    def new_bucket():
        return {"correct": 0, "incorrect": 0, "partial": 0}
    by_gap = {name: new_bucket() for name, _, _ in GAP_BUCKETS}
    by_box = {str(b): new_bucket() for b in range(1, MAX_BOX + 1)}
    totals = new_bucket()
    relearning = 0
    skipped = 0
    deferrals = 0
    self_reports = 0
    legacy_no_box = 0
    delayed = new_bucket()
    by_kind, by_basis = {}, {}

    def tally(groups, key, res, is_delayed):
        g = groups.setdefault(key, {"all": new_bucket(), "delayed": new_bucket()})
        g["all"][res] += 1
        if is_delayed:
            g["delayed"][res] += 1

    for c in data.get("concepts", []):
        kind = effective_kind(c)
        # After a truncation the first retained gap runs from the last
        # archived review, not from `introduced` (review finding 6).
        prev = (parse_date(archived_summary(c).get("through"))
                or parse_date(c.get("introduced")))
        for h in c.get("reviewHistory", []):
            if h.get("deferred"):
                deferrals += 1  # scheduling event, not retrieval evidence
                continue
            if is_self_report(h):
                self_reports += 1  # nothing was asked; not retrieval evidence
                continue
            d = parse_date(h.get("date"))
            res = h.get("result")
            if d is None or res not in ("correct", "incorrect", "partial"):
                skipped += 1
                continue
            if h.get("retry"):
                relearning += 1
                continue
            is_delayed = False
            if prev is not None:
                bucket = bucket_for_gap((d - prev).days)
                if bucket:
                    by_gap[bucket][res] += 1
                    if bucket != "same-day":
                        delayed[res] += 1
                        is_delayed = True
            prev = d
            totals[res] += 1
            tally(by_kind, kind, res, is_delayed)
            tally(by_basis, entry_basis(h) or "notApplied", res, is_delayed)
            bb = h.get("boxBefore")
            if isinstance(bb, int) and 1 <= bb <= MAX_BOX:
                by_box[str(bb)][res] += 1
            else:
                legacy_no_box += 1
    def with_rate(bucket):
        n = sum(bucket.values())
        out = dict(bucket)
        out["reviews"] = n
        out["successRate"] = round(bucket["correct"] / n, 2) if n else None
        return out
    def rates(g):
        n, dn = sum(g["all"].values()), sum(g["delayed"].values())
        return {"reviews": n,
                "successRate": round(g["all"]["correct"] / n, 2) if n else None,
                "delayedReviews": dn,
                "delayedSuccessRate": round(g["delayed"]["correct"] / dn, 2) if dn else None}
    total_n = sum(totals.values())
    delayed_n = sum(delayed.values())
    return {
        "reviews": total_n,
        "overallSuccessRate": round(totals["correct"] / total_n, 2) if total_n else None,
        "delayedReviews": delayed_n,
        "delayedSuccessRate": round(delayed["correct"] / delayed_n, 2) if delayed_n else None,
        "byGap": {k: with_rate(v) for k, v in by_gap.items()},
        "byBoxAtReview": {k: with_rate(v) for k, v in by_box.items()},
        "relearningRetriesExcluded": relearning,
        "deferralsExcluded": deferrals,
        "selfReportsExcluded": self_reports,
        "entriesWithoutBoxBefore": legacy_no_box,
        "entriesSkipped": skipped,
        "byKind": {k: rates(by_kind[k]) for k in EVIDENCE_KINDS if k in by_kind},
        "byBasis": {b: rates(by_basis[b]) for b in APPLIED_BASES + ("notApplied",)
                    if b in by_basis},
    }


def cmd_retention(args):
    _, data = load_spaced_review(args.project)
    out = retention_summary(data)
    out["note"] = ("Leitner targets roughly 80-90% success at review time: "
                   "persistently above suggests intervals too conservative; "
                   "persistently below, too aggressive")
    emit(out)


def cmd_export_anonymized(args):
    """Shareable stats block: counts and rates only — no concept names,
    no questions, no notes, no free text of any kind (1.11.3)."""
    _, data = load_spaced_review(args.project)
    boxes = {str(b): 0 for b in range(1, MAX_BOX + 1)}
    blooms = {str(b): 0 for b in range(0, 7)}
    feynman = 0
    mastered = 0
    applied = 0
    # By kind and basis: counts only, keyed by vocabulary, never by a name.
    by_kind, by_basis = {}, {}

    def basis_row(basis):
        return by_basis.setdefault(basis, {"appliedCorrects": 0, "solid": 0})

    for c in data.get("concepts", []):
        row = by_kind.setdefault(effective_kind(c), {"concepts": 0, "applied": 0, "mastered": 0})
        row["concepts"] += 1
        row["applied"] += applied_evidence(c) >= 1
        row["mastered"] += is_mastered(c)
        for h in c.get("reviewHistory", []):
            basis = applied_correct_basis(h) if isinstance(h, dict) else None
            if basis:
                basis_row(basis)["appliedCorrects"] += 1
        if is_mastered(c):
            basis_row(strongest_basis(c))["solid"] += 1
        b = c.get("box", 1)
        if isinstance(b, int) and 1 <= b <= MAX_BOX:
            boxes[str(b)] += 1
        bl = c.get("bloomLevel", 0)
        if isinstance(bl, int) and 0 <= bl <= 6:
            blooms[str(bl)] += 1
        if c.get("feynmanPassed") is True:
            feynman += 1
        if is_mastered(c):
            mastered += 1
        if applied_evidence(c) >= 1:
            applied += 1
    session_types = {}
    for s in data.get("sessionHistory", []):
        t = s.get("type") if isinstance(s.get("type"), str) else "(untyped)"
        session_types[t] = session_types.get(t, 0) + 1
    calibration = calibration_summary(data)
    calibration.pop("overconfidentEvents", None)   # carry concept names
    calibration.pop("underconfidentEvents", None)
    out = {
        "exportVersion": 1,
        "generated": iso(today()),
        "concepts": len(data.get("concepts", [])),
        "boxDistribution": boxes,
        "bloomDistribution": blooms,
        "feynmanPassed": feynman,
        "applied": applied,
        "mastered": mastered,
        "sessionTypeCounts": session_types,
        "retention": retention_summary(data),
        "calibration": calibration,
        "byKind": {k: by_kind[k] for k in EVIDENCE_KINDS if k in by_kind},
        "byBasis": {b: by_basis[b] for b in APPLIED_BASES if b in by_basis},
    }
    state = state_if_present(args.project)
    if state is not None:
        created = parse_date(state.get("createdAt"))
        out["project"] = {
            "totalSessions": state.get("totalSessions", 0),
            "currentStreak": state.get("currentStreak", 0),
            "overallCompletion": state.get("overallCompletion", 0),
            "daysSinceStart": (today() - created).days if created else None,
        }
    emit(out)


def cmd_session_brief(args):
    """Mechanical branch detection for /teach (1.14.0).

    Pretest-vs-retrieval-open and the targeted-reteach duty are pure state
    predicates; executors were re-deriving them from prose — the same
    judgment-tree residue class 1.11.0 closed for writes. Read-only.
    """
    state = state_if_present(args.project, {})
    _, data = load_spaced_review(args.project, create=True)
    t = today()
    c = find_concept(data, args.concept)
    if c is None:
        emit({"concept": args.concept, "tracked": False,
              "firstExposure": True, "pretestApplies": True,
              "isReteach": False, "dueForReview": False, "reviews": 0,
              "currentModule": state.get("currentModule"),
              "evidenceKind": kind_of_module(state.get("currentModule"))})
        return
    history = [h for h in c.get("reviewHistory", []) if not h.get("deferred")]
    # First exposure: never classified by a v3 writer AND no real review has
    # ever happened (deferrals are scheduling, not exposure evidence).
    first_exposure = c.get("bloomLevel", 0) == 0 and not history
    last = parse_date(c.get("lastReviewed"))
    last_result = history[-1].get("result") if history else None
    nr = parse_date(c.get("nextReview"))
    emit({
        "concept": c["name"],
        "tracked": True,
        "firstExposure": first_exposure,
        "pretestApplies": first_exposure,  # pretesting research covers untaught material only
        # Targeted re-teach: the concept has real history and sits demoted
        # (Box 1) or its latest real outcome was a demonstrated forgetting.
        "isReteach": bool(history) and (c.get("box", 1) == 1
                                        or last_result == "incorrect"),
        "box": c.get("box", 1),
        "bloomLevel": c.get("bloomLevel", 0),
        **bloom_render(c.get("bloomLevel", 0)),
        "evidenceAt3Plus": evidence_at_3_plus(c),
        "appliedEvidence": applied_evidence(c),
        "feynmanPassed": c.get("feynmanPassed", False),
        "feynmanCurrent": feynman_current(c),
        "lastReviewed": c.get("lastReviewed"),
        "daysSinceLastReview": (t - last).days if last else None,
        "lastResult": last_result,
        "dueForReview": nr is not None and nr <= t,
        "reviews": len(history) + (c.get("reviewHistoryArchived", 0) or 0),
        "currentModule": state.get("currentModule"),
        "evidenceKind": effective_kind(c),
    })


def slugify(text, limit=40):
    """kebab-case for a file name: 'B-tree indexes' -> 'b-tree-indexes'."""
    out = re.sub(r"[^a-z0-9]+", "-", str(text).lower()).strip("-")
    return out[:limit].rstrip("-") or "session"


# --- syllabus (1.24.0) ------------------------------------------------------
# The plan says the order; the review file says the evidence; state.json says
# the position. No skill could join the three reliably in prose: /plan view
# read every phase file and showed order with no performance, /progress
# showed per-module tiers with no order (its modules come from the review
# file), and "where am I and how did I do" needs both. One read-only call.

PHASE_FILE_RE = re.compile(r"^phase-(\d+)\.md$")
# `### Module 1.2 — name` is canonical; `### Module 7 — name` (one number,
# counted across the course) is what the OpenAI build's /learn wrote in a
# real session, and a plan the reader cannot see empties /plan, /progress and
# /continue's position line. The id is the number(s) as written.
MODULE_HEADING_RE = re.compile(r"^###\s+Module\s+(\d+)(?:\.(\d+))?\s*(?:[—–:-]+\s*)?(.*?)\s*$")
PLAN_UNREADABLE = ("plan/ has phase files but no `### Module N.M — name` (or "
                   "`### Module N — name`) headings; /plan, /progress and the "
                   "position line read an empty plan")
MODULE_LINE_RE = re.compile(
    r"^\s*(?:[-*]\s+)?\*\*(Status|Target Bloom(?:'s)?(?: level)?|Concepts|Key terms?)\s*:?\*\*\s*:?\s*(.+?)\s*$",
    re.IGNORECASE)
MODULE_STATUS_WORDS = {"skipped": "skipped", "completed": "done", "done": "done",
                       "complete": "done"}
MODULE_STATUS_RE = re.compile(r"\b(skipped|completed|complete|done)\b", re.IGNORECASE)


def norm_module_name(s):
    """Join key for a module name: a `Module 1.2 — ` or `Module 2 — ` prefix
    dropped, whitespace collapsed, case folded. Concept rows carry whatever
    `--module` was passed, so the join tolerates the executor's spelling —
    including a plan renumbered from `Module 2` to `Module 0.2` after rows
    were written."""
    if not isinstance(s, str):
        return ""
    stripped = re.sub(r"^\s*module\s+\d+(?:\.\d+)?\s*[—–:-]*\s*", "", s, flags=re.IGNORECASE)
    if stripped.strip():
        s = stripped
    return re.sub(r"\s+", " ", s).strip().casefold()


# --- Evidence-kind resolver -------------------------------------------------
# The kind reaches the applied-work counters through this per-process table,
# installed once by main() (one process serves one --project), not through a
# parameter on every reader. Uninstalled, everything reads as code: the
# strictest reading, since code admits only observed work.
_KINDS = {"project": "code", "modules": {}}


def read_kinds(project):
    """(project kind, {normalized module name: kind}) from state.json. Never
    fails a reader: a missing or unreadable file, or a kind outside
    EVIDENCE_KINDS, reads as code (verify reports the bad value)."""
    try:
        with open(state_path(project), encoding="utf-8") as f:
            state = json.load(f)
    except (OSError, ValueError):
        return "code", {}
    if not isinstance(state, dict):
        return "code", {}
    kind = state.get("evidenceKind")
    modules = state.get("moduleEvidenceKinds")
    modules = modules if isinstance(modules, dict) else {}
    return (kind if kind in EVIDENCE_KINDS else "code",
            {norm_module_name(name): k if k in EVIDENCE_KINDS else "code"
             for name, k in modules.items()})


def install_kinds(project):
    _KINDS["project"], _KINDS["modules"] = read_kinds(project)


def kind_of_module(module):
    return _KINDS["modules"].get(norm_module_name(module), _KINDS["project"])


def effective_kind(c):
    """A concept's evidence kind: its module's override, else the project's."""
    return kind_of_module(c.get("module"))


def applied_basis(h):
    """How an applied entry's work was seen. Absent is `observed`; a value
    outside APPLIED_BASES reads as `reported`, the strictest."""
    b = h.get("appliedBasis")
    if b is None:
        return "observed"
    return b if b in APPLIED_BASES else "reported"


def admits(kind, basis):
    """Whether applied work on `basis` counts toward the gate and Solid for a
    concept of `kind`: observed work everywhere, unseen work only where the
    tutor cannot see it."""
    return basis == "observed" or kind in UNSEEN_BASIS_KINDS


def _coerce_choice(x, choices):
    """The lossless vocabulary repair: case and surrounding spaces."""
    if isinstance(x, str) and x not in choices and x.strip().casefold() in choices:
        return x.strip().casefold()
    return None


def parse_bloom_target(val):
    """`3`, `3 (Apply)`, `Apply` → 3; anything else None."""
    m = re.search(r"\b([1-6])\b", val)
    if m:
        return int(m.group(1))
    low = val.casefold()
    for lvl, label in BLOOM_LABELS.items():
        if label and label.casefold() in low:
            return lvl
    return None


def read_plan_modules(project):
    """The plan's module order from plan/phase-*.md (state-schema KB *plan/*):
    `### Module <N>.<M> — <name>` headings in phase-number order, with the
    optional `**Status:**`, `**Target Bloom's:**`, `**Concepts:**` and
    `**Key terms:**` lines of each section. Returns (phases, modules) or (None, []) with no plan/."""
    pdir = os.path.join(project, ".bodhi", "plan")
    if not os.path.isdir(pdir):
        return None, []
    files = []
    try:
        names = os.listdir(pdir)
    except OSError:
        names = []
    for f in names:
        m = PHASE_FILE_RE.match(f)
        if m:
            files.append((int(m.group(1)), f))
    phases, modules = [], []
    for n, f in sorted(files):
        try:
            # A plan is prose the learner pastes into; one Latin-1 byte must
            # not take `snapshot` (and with it /continue) down. Replace, skip
            # nothing readable.
            with open(os.path.join(pdir, f), encoding="utf-8", errors="replace") as fh:
                text = fh.read()
        except OSError:
            continue
        title = next((ln[2:].strip() for ln in text.splitlines()
                      if ln.startswith("# ")), f"Phase {n}")
        cur, count, fenced = None, 0, False
        for line in text.splitlines():
            if line.lstrip().startswith("```"):
                fenced = not fenced
                continue
            if fenced:
                continue
            m = MODULE_HEADING_RE.match(line)
            if m:
                mid = f"{m.group(1)}.{m.group(2)}" if m.group(2) else m.group(1)
                cur = {"phase": n, "id": mid, "name": m.group(3) or f"Module {mid}",
                       "file": f, "declaredStatus": None, "targetBloom": None,
                       "plannedConcepts": [], "keyTerms": []}
                modules.append(cur)
                count += 1
                continue
            if line.startswith("## ") or line.startswith("# "):
                cur = None
            if cur is None:
                continue
            ml = MODULE_LINE_RE.match(line)
            if not ml:
                continue
            key, val = ml.group(1).casefold(), ml.group(2)
            if key == "status":
                sw = MODULE_STATUS_RE.search(val)
                cur["declaredStatus"] = MODULE_STATUS_WORDS.get(sw.group(1).casefold()) if sw else None
            elif key.startswith("target bloom"):
                cur["targetBloom"] = parse_bloom_target(val)
            elif key == "concepts":
                cur["plannedConcepts"] = [x.strip() for x in val.split(",") if x.strip()]
            elif key.startswith("key term"):
                cur["keyTerms"] = [x.strip(" `*") for x in val.split(",") if x.strip(" `*")]
        phases.append({"phase": n, "title": title, "file": f, "modules": count})
    return phases, modules


def plan_position(state, plan):
    """Index of the current module in the plan's order: state.json
    currentModule by name or id, else currentModuleIndex, else None. The
    second and third cases carry a warning. Shared by `syllabus` and
    `snapshot` (whose `project.position` is what /continue's check-in
    prints), so "Module 3 of 8" has one home."""
    warnings = []
    cur_name = norm_module_name(state.get("currentModule"))
    cur_id = re.sub(r"(?i)^\s*module\s+", "", str(state.get("currentModule") or "")).strip()
    cur_phase = str(state.get("currentPhase") or "").strip()
    matches = [i for i, m in enumerate(plan)
               if cur_name and (norm_module_name(m["name"]) == cur_name or m["id"] == cur_id)]
    if matches:
        # A spiral revisit may declare the same name in a later phase; the
        # phase the state names breaks the tie, else the first occurrence.
        in_phase = [i for i in matches if str(plan[i]["phase"]) == cur_phase]
        return (in_phase or matches)[0], warnings
    if plan:
        idx = state.get("currentModuleIndex")
        if isinstance(idx, str) and idx.strip().isdigit():
            idx = int(idx)
        if isinstance(idx, int) and not isinstance(idx, bool) and 0 <= idx < len(plan):
            if cur_name:
                warnings.append(f"state.json currentModule {state.get('currentModule')!r} "
                                f"is not a plan module; position taken from "
                                f"currentModuleIndex {idx}")
            return idx, warnings
        warnings.append("state.json names no plan module as current; every "
                        "undeclared module reads as upcoming")
    return None, warnings


def position_row(plan, idx):
    """The learner-facing position: 1-based index over all plan modules."""
    if idx is None:
        return None
    return {"index": idx + 1, "of": len(plan),
            "module": plan[idx]["name"], "phase": plan[idx]["phase"]}


def summary_tier(tiers):
    """The /progress rendering rule in code: a mixed module reads at its
    lowest occupied tier above `unclassified`; all-unclassified is None."""
    for key in ("introduced", "familiar", "mastered"):
        if tiers.get(key, 0) > 0:
            return key
    return None


def cmd_syllabus(args):
    """Plan order × review evidence × state position, one read-only call.
    Every module the plan declares, in order, with its status (declared
    `skipped`/`done`, else `done`/`current`/`upcoming` from the position the
    plugin already tracks), the review file's module row, the earliest next
    review and today's due count. Modules on either side that the other side
    does not know are reported, never dropped: `untracked` names plan modules
    no concept was tracked under (not reached yet, or a spelling the review
    file does not use); `unplanned` carries the rows of tracked modules the
    plan does not declare, so no evidence disappears from a dashboard."""
    state = state_if_present(args.project, {})
    _, data = load_spaced_review(args.project, create=True)
    t = today()
    concepts = data.get("concepts", [])
    warnings = []

    phases, plan = read_plan_modules(args.project)
    if phases is None:
        warnings.append("no .bodhi/plan/ directory — a pre-1.7.0 monolithic "
                        "plan.md is not read; /housekeep migrate converts it")
        phases = []
    elif not plan:
        warnings.append(PLAN_UNREADABLE)

    # Evidence per tracked module, by join key — the same tally `mastery`
    # and `snapshot` use, re-keyed so "schema design" and "Schema Design"
    # are one module here (the rollup keys by the raw spelling).
    tracked = {}
    for c in concepts:
        raw = c.get("module", "(none)")
        entry = tracked.setdefault(norm_module_name(raw),
                                   {"module": raw, "aliases": [], "row": new_module_row(),
                                    "nextDue": None, "dueToday": 0, "parked": 0})
        if raw != entry["module"] and raw not in entry["aliases"]:
            entry["aliases"].append(raw)
        tally_module(entry["row"], c, t)
        if c.get("parked") is True:
            entry["parked"] += 1
            continue
        nr = parse_date(c.get("nextReview"))
        if nr is None:
            continue
        if nr <= t:
            entry["dueToday"] += 1
        if entry["nextDue"] is None or nr < entry["nextDue"]:
            entry["nextDue"] = nr
    finalize_module_rows({k: e["row"] for k, e in tracked.items()})

    def evidence(entry):
        if entry is None:
            return {"tracked": False, "concepts": 0, "classified": 0, "applied": 0,
                    "mastered": 0, "masteredByReport": 0, "dueForCheck": 0, "masteryPct": None,
                    "tiers": new_module_row()["tiers"], "summaryTier": None,
                    "nextDue": None, "dueToday": 0, "parked": 0}
        row = entry["row"]
        out = {"tracked": True, "trackedAs": entry["module"]}
        if entry["aliases"]:
            out["trackedAlso"] = entry["aliases"]
        out.update({k: row[k] for k in ("concepts", "classified", "applied", "mastered",
                                        "masteredByReport", "dueForCheck", "masteryPct",
                                        "tiers")})
        out["summaryTier"] = summary_tier(row["tiers"])
        out["nextDue"] = iso(entry["nextDue"]) if entry["nextDue"] else None
        out["dueToday"] = entry["dueToday"]
        out["parked"] = entry["parked"]
        return out

    cur_idx, pos_warnings = plan_position(state, plan)
    warnings.extend(pos_warnings)

    modules, matched_keys = [], set()
    for i, m in enumerate(plan):
        key = norm_module_name(m["name"])
        entry = tracked.get(key) or tracked.get(m["id"])
        if entry is not None:
            matched_keys.add(norm_module_name(entry["module"]))
        if m["declaredStatus"]:
            status, source = m["declaredStatus"], "declared"
        elif cur_idx is None:
            status, source = "upcoming", "unknown-position"
        elif i < cur_idx:
            status, source = "done", "plan-order"
        elif i == cur_idx:
            status, source = "current", "plan-order"
        else:
            status, source = "upcoming", "plan-order"
        row = {"index": i + 1, "phase": m["phase"], "id": m["id"], "name": m["name"],
               "status": status, "statusSource": source,
               "targetBloom": m["targetBloom"], "plannedConcepts": m["plannedConcepts"],
               "keyTerms": m["keyTerms"], "file": m["file"]}
        if m["targetBloom"] is not None:
            tb = bloom_render(m["targetBloom"])
            row["targetBloomLabel"] = tb["bloomLabel"]
            row["targetBloomOutcome"] = tb["bloomOutcome"]
        row["evidence"] = evidence(entry)
        row["evidenceKind"] = kind_of_module(m["name"])
        modules.append(row)

    untracked = [m["name"] for m in modules if not m["evidence"]["tracked"]]
    unplanned = [{"module": e["module"], **{k: v for k, v in evidence(e).items()
                                            if k not in ("tracked", "trackedAs")},
                  "evidenceKind": kind_of_module(e["module"])}
                 for k, e in tracked.items() if k not in matched_keys]

    emit({
        "project": state.get("projectName"),
        "evidenceKind": _KINDS["project"],
        "plan": phases is not None and bool(plan),
        "phases": phases,
        "position": position_row(plan, cur_idx),
        "currentPhase": state.get("currentPhase"),
        "overallCompletion": state.get("overallCompletion", 0),
        "counts": {"modules": len(plan),
                   "done": sum(1 for m in modules if m["status"] == "done"),
                   "skipped": sum(1 for m in modules if m["status"] == "skipped"),
                   "upcoming": sum(1 for m in modules if m["status"] == "upcoming")},
        "modules": modules,
        "untracked": untracked,
        "unplanned": unplanned,
        "warnings": warnings,
    })


def cmd_revision_brief(args):
    """What today's revision sheet is about (1.18.0). Read-only.

    The sheet is the learner's take-home; the model writes the prose, this
    reports the facts: every concept with a non-deferred review dated today
    (what was actually studied — seeding via /learn or an /assess writes no
    review, so those days need no sheet), the file name to write, and any
    sheet already written today (append, never duplicate). The Stop hook
    reads `sessionToday` and `existing` to refuse a stop without a sheet."""
    t = iso(today())
    state = state_if_present(args.project, {})
    _, data = load_spaced_review(args.project, create=True)
    studied = []
    for c in data.get("concepts", []):
        todays = [h for h in c.get("reviewHistory", [])
                  if not h.get("deferred") and h.get("date") == t]
        if not todays and c.get("introduced") != t:
            continue
        studied.append({
            "name": c["name"], "module": c.get("module", ""),
            "resultsToday": [h.get("result") for h in todays],
            "sourcesToday": sorted({h.get("source") for h in todays if h.get("source")}),
            "box": c.get("box", 1), "nextReview": c.get("nextReview"),
            "bloomLevel": c.get("bloomLevel", 0),
            **bloom_render(c.get("bloomLevel", 0)),
            "question": c.get("question", ""),
            "feynmanPassed": c.get("feynmanPassed", False),
            "evidenceKind": effective_kind(c),
        })
    # the sheet is named after the taught concept when there is one
    studied.sort(key=lambda d: (0 if "teach" in d["sourcesToday"] else 1,
                                -len(d["resultsToday"]), d["name"].lower()))
    rev_dir = os.path.join(args.project, "revision")
    existing = sorted(f for f in (os.listdir(rev_dir) if os.path.isdir(rev_dir) else [])
                      if f.startswith(t) and f.endswith(".md"))
    if studied:
        slug = slugify(studied[0]["name"] if len(studied) == 1 or "teach" in studied[0]["sourcesToday"]
                       else (studied[0]["module"] or studied[0]["name"]))
    else:
        slug = "session"
    emit({"date": t, "project": os.path.basename(os.path.abspath(args.project)),
          "module": state.get("currentModule"),
          "session": state.get("totalSessions"),
          "sessionToday": bool(studied),
          "concepts": studied,
          "suggestedFile": os.path.join("revision", f"{t}-{slug}.md"),
          "existing": [os.path.join("revision", f) for f in existing],
          "resourcesFile": ".bodhi/resources.md"
                           if os.path.exists(os.path.join(args.project, ".bodhi", "resources.md")) else None})


def cmd_snapshot(args):
    """Single-call dashboard rollup for /progress (1.14.0). Read-only.

    Merges the state.json position, session cadence, due/box/mastery counts,
    and the calibration rates into one JSON so /progress reads this plus the
    live progress.md — its context cost stays O(1) in session count instead
    of growing with the tracking files.
    """
    sp, state = require_state(args.project)
    _, data = load_spaced_review(args.project, create=True)
    t = today()

    dates = sorted({d for d in state.get("sessionDates", [])
                    if isinstance(d, str) and parse_date(d)})
    def sessions_since(days):
        cutoff = t - datetime.timedelta(days=days)
        return sum(1 for d in dates if parse_date(d) >= cutoff)
    last_date = parse_date(dates[-1]) if dates else None

    concepts = data.get("concepts", [])
    r = review_rollup(concepts, t)
    mastered = sum(1 for c in concepts if is_mastered(c))
    feynman = sum(1 for c in concepts if c.get("feynmanPassed") is True)
    classified = sum(1 for c in concepts if c.get("bloomLevel", 0) > 0)
    applied = sum(1 for c in concepts if applied_evidence(c) >= 1)
    by_report, solid_basis = solid_by_report(concepts)

    calibration = calibration_summary(data)
    calibration.pop("overconfidentEvents", None)
    calibration.pop("underconfidentEvents", None)
    _, plan = read_plan_modules(args.project)
    plan_idx, _ = plan_position(state, plan)

    emit({
        "project": {
            "name": state.get("projectName"),
            "topic": state.get("topic"),
            "createdAt": state.get("createdAt"),
            "currentPhase": state.get("currentPhase"),
            "currentModule": state.get("currentModule"),
            "currentModuleIndex": state.get("currentModuleIndex"),
            "position": position_row(plan, plan_idx),
            "lastActivity": state.get("lastActivity"),
            "overallCompletion": state.get("overallCompletion", 0),
            "initialBloomLevel": state.get("initialBloomLevel", {}),
            "currentBloomLevel": state.get("currentBloomLevel", {}),
            "evidenceKind": _KINDS["project"],
            # /continue proposes a kind once when none was ever recorded.
            "evidenceKindRecorded": state.get("evidenceKind") in EVIDENCE_KINDS,
        },
        "cadence": {
            "totalSessions": state.get("totalSessions", 0),
            "currentStreak": state.get("currentStreak", 0),
            "lastSessionAt": state.get("lastSessionAt"),
            "daysSinceLastSession": (t - last_date).days if last_date else None,
            "sessionsLast7d": sessions_since(7),
            "sessionsLast30d": sessions_since(30),
        },
        "review": {
            "concepts": len(concepts),
            "dueToday": len(r["dueToday"]), "dueTodayConcepts": r["dueToday"],
            "dueThisWeek": len(r["dueThisWeek"]), "dueThisWeekConcepts": r["dueThisWeek"],
            "overdue10dPlus": r["overdue10"],
            "waitingForReviewDay": r["waiting"],
            "unscheduled": r["unscheduled"],
            "boxDistribution": r["boxes"],
            "retentionRollup": {k: len(v) for k, v in r["rollup"].items()},
            "retentionConcepts": r["rollup"],
            "unparseableDates": r["unparseable"],
            "parked": len(r["parked"]),
        },
        "mastery": {
            "mastered": mastered,
            "feynmanPassed": feynman,
            "applied": applied,
            "classified": classified,
            "modules": r["modules"],
            "blockedOnFeynman": blocked_on_feynman(concepts),
            "blockedOnApplied": blocked_on_applied(concepts),
            "masteredDueForCheck": r["dueForCheck"],
            "solidByReport": by_report,
            "solidBasis": solid_basis,
        },
        "calibration": calibration,
        "bloomScale": bloom_scale(),
    })


def evidence_at_3_plus(c):
    """Independent observations of the apply rung SINCE THE LAST MISS: correct
    results graded at Bloom >= 3, counted from the most recent `incorrect`
    (a learner-initiated /forget writes one, so it resets the count too).
    Evidence that predates a demonstrated forgetting event is not current
    evidence; without the reset, two old corrects kept a concept `satisfied`
    through any number of later misses. Deferred entries are scheduling
    notes, not observations, and are skipped; archived entries are carried
    by archivedSummary."""
    return _evidence_counts(c.get("reviewHistory", []), archived_summary(c))["at3plus"]


EXPOSURE_RANK = {"seeded": 0, "quizzed-only": 1, "demonstrated": 2,
                 "taught": 3, "built": 4}


def archived_summary(c):
    s = c.get("archivedSummary")
    return s if isinstance(s, dict) else {}


def built_by_of(summary):
    """The unseen bases archived work was built on (archivedSummary.builtBy)."""
    b = summary.get("builtBy")
    return [x for x in b if x in ("inspected", "reported")] if isinstance(b, list) else []


# archivedSummary key that seeds each count (state-schema KB).
ARCHIVE_SEEDS = {"observed": "appliedEvidence", "inspected": "appliedInspected",
                 "reported": "appliedReported", "at3plus": "evidenceAt3Plus"}


def entry_basis(h):
    """The basis of an applied entry, else None: the one place the `applied`
    flag is read for counting (t_single_applied_reader), so no reader can
    count unseen work as observed."""
    return applied_basis(h) if h.get("applied") is True else None


def applied_correct_basis(h):
    """The basis of an applied correct, else None."""
    return entry_basis(h) if h.get("result") == "correct" else None


def _evidence_counts(entries, seed):
    """Applied corrects per basis and level-3+ corrects, since the most
    recent miss over `entries`, continuing from an archivedSummary `seed` —
    the one loop behind applied_evidence, evidence_at_3_plus and
    archive_history. Raw counts: which bases a concept admits is the
    reader's question, asked through admits()."""
    counts = {k: seed.get(key, 0) or 0 for k, key in ARCHIVE_SEEDS.items()}
    for h in entries:
        if not isinstance(h, dict) or h.get("deferred"):
            continue
        result = h.get("result")
        if result == "incorrect":
            counts = dict.fromkeys(counts, 0)
        elif result == "correct":
            basis = applied_correct_basis(h)
            if basis:
                counts[basis] += 1
            if (h.get("bloomLevel") or 0) >= 3:
                counts["at3plus"] += 1
    return counts


def archive_history(c, archived):
    """Fold the entries about to leave reviewHistory into archivedSummary so
    no reader's answer changes at the cut (review finding 6, 2026-09-07:
    truncation used to turn a taught, built concept into neverTaught with
    zero applied evidence, and to measure the first retained gap from
    `introduced`). Accumulates across repeated truncations."""
    prev = archived_summary(c)
    counts = _evidence_counts(archived, prev)
    summary = {"appliedEvidence": counts["observed"], "evidenceAt3Plus": counts["at3plus"]}
    dates = [h.get("date") for h in archived
             if isinstance(h, dict) and parse_date(h.get("date"))]
    through = dates[-1] if dates else prev.get("through")
    if through:
        summary["through"] = through
    misses = [d for d in (parse_date(h.get("date")) for h in archived
                          if isinstance(h, dict) and h.get("result") == "incorrect")
              if d] + [d for d in [parse_date(prev.get("lastMiss"))] if d]
    if misses:
        summary["lastMiss"] = iso(max(misses))
    # Instruction facts never reset; the evidence-derived states are
    # recomputed from the seeds, so only these two are worth carrying.
    graded = [h for h in archived if isinstance(h, dict) and not h.get("deferred")
              and h.get("result") in ("correct", "incorrect", "partial")]
    # `built` is the floor observed work sets; work on an unseen basis is
    # kept as builtBy, so a later kind change can still admit or refuse it.
    exposure = prev.get("exposure")
    bases = {applied_correct_basis(h) for h in graded}
    if "observed" in bases:
        exposure = "built"
    elif exposure != "built" and any(h.get("source") in INSTRUCTIONAL_SOURCES for h in graded):
        exposure = "taught"
    if exposure in ("taught", "built"):
        summary["exposure"] = exposure
    for basis in ("inspected", "reported"):
        if counts[basis]:
            summary[ARCHIVE_SEEDS[basis]] = counts[basis]
    built_by = [b for b in ("inspected", "reported") if b in bases or b in built_by_of(prev)]
    if built_by:
        summary["builtBy"] = built_by
    c["reviewHistoryArchived"] = (c.get("reviewHistoryArchived", 0) or 0) + len(archived)
    c["archivedSummary"] = summary


def applied_evidence(c):
    """Correct results demonstrated in applied work (`applied: true`: code
    that ran, or a written application on a non-code topic) SINCE THE LAST
    MISS, the same reset as evidence_at_3_plus. The level says how well the
    learner can *talk about* a concept; this says whether they have *used* it. Every recorded level before 1.20.0 came from an
    explanation or a quiz answer, so the gate and the mastery formula could
    be satisfied without the learner ever writing a line: the learner
    finding that motivated the flag. Deferred entries are skipped; archived
    entries are carried by archivedSummary."""
    counts = _evidence_counts(c.get("reviewHistory", []), archived_summary(c))
    kind = effective_kind(c)
    return sum(counts[b] for b in APPLIED_BASES if admits(kind, b))


def strongest_basis(c):
    """The best-seen basis among the applied corrects this concept's kind
    admits since the last miss — observed, then inspected, then reported —
    or None. A Solid resting on `reported` is Solid by the learner's
    account, and every surface that shows it says so."""
    counts = _evidence_counts(c.get("reviewHistory", []), archived_summary(c))
    kind = effective_kind(c)
    return next((b for b in APPLIED_BASES if counts[b] and admits(kind, b)), None)


def solid_by_report(concepts):
    """(names Solid on the learner's word alone, {Solid name: strongest basis})."""
    basis = {c["name"]: strongest_basis(c) for c in concepts if is_mastered(c)}
    return [n for n, b in basis.items() if b == "reported"], basis


def last_miss(c):
    """Date of the most recent `incorrect` on record — live history, else
    archivedSummary.lastMiss. The anchor for "since the last miss" when a
    reader needs a date rather than a count (feynman_current)."""
    dates = [parse_date(h.get("date")) for h in c.get("reviewHistory", [])
             if isinstance(h, dict) and not h.get("deferred")
             and h.get("result") == "incorrect"]
    dates = [d for d in dates if d]
    archived = parse_date(archived_summary(c).get("lastMiss"))
    if archived:
        dates.append(archived)
    return max(dates) if dates else None


def feynman_current(c):
    """The explain-back conjunct of is_mastered (1.23.0): passed, and dated
    AFTER the most recent miss. An explanation given before the concept was
    forgotten says nothing about the understanding that came back — a real
    learner's delayed quiz (2026-09-25) missed a concept whose explain-back
    had passed 17 days earlier, and the flag still counted. A pass on the day of
    the miss does not count (dates carry no order within a day, and an
    explanation right after relearning is the fluency peak); an undated
    pre-1.23.0 flag counts only when no miss is on record."""
    if c.get("feynmanPassed") is not True:
        return False
    miss = last_miss(c)
    if miss is None:
        return True
    passed = parse_date(c.get("feynmanPassedAt"))
    return passed is not None and passed > miss


# Review sources that are instruction: the tutor taught, or watched the
# learner build. A quiz answer or a /reflect retrieval is evidence of
# knowledge, not of teaching. debug-together writes no reviews today; listed
# so a future write from it counts as instruction without a script change.
INSTRUCTIONAL_SOURCES = {"teach", "practice", "pair", "debug-together"}


def exposure_of(c):
    """How the learner has met this concept, from the evidence on record:
      seeded        — tracked (by /learn's assessment or a skill), never graded
      quizzed-only  — graded, but only by quiz/reflect retrievals and none of
                      them reached the apply rung since the last miss
      demonstrated  — no instruction on record, but the learner has answered
                      at the apply rung or above (a seed assessed high and
                      quizzed correctly): knowledge acquired elsewhere
      taught        — at least one review from an instructional source
      built         — at least one `applied` correct on record
    Review finding 3 (2026-09-07): `neverTaught` used to mean only "no
    source=teach entry", so a concept mastered through /practice with working
    code was routed to first teaching, and knowledge shown in quizzes did not
    count. It now means exposure in {seeded, quizzed-only} — the two states
    where quizzing it would test nothing the learner has been given."""
    graded = [h for h in c.get("reviewHistory", []) if isinstance(h, dict)
              and not h.get("deferred")
              and h.get("result") in ("correct", "incorrect", "partial")]
    summary = archived_summary(c)
    kind = effective_kind(c)
    bases = [applied_correct_basis(h) for h in graded] + built_by_of(summary)
    if summary.get("exposure") == "built" or any(b and admits(kind, b) for b in bases):
        return "built"
    if summary.get("exposure") == "taught" or any(h.get("source") in INSTRUCTIONAL_SOURCES
                                                  for h in graded):
        return "taught"
    if not graded and not c.get("reviewHistoryArchived"):
        return "seeded"
    if evidence_at_3_plus(c) >= 1:
        return "demonstrated"
    return "quizzed-only"


def never_taught(c):
    return exposure_of(c) in ("seeded", "quizzed-only")


def gate_verdict(c, t):
    """(status, reason) for one prerequisite (canonical).

    The Bloom ratchet is one-way and a single LLM grade is noisy, so a level-3
    reached by exactly one review is not treated as settled: it earns one
    reconfirm question (the stale-reconfirm path), not a free pass. Two
    level-3+ corrects since the last miss, or Box >= 3 (which already implies
    two spaced successes), clear the gate. `single-evidence` means "fewer than
    two such corrects on record since the last miss" — after a /forget that
    count is zero and the gate asks again, which is the point of forgetting.

    1.20.0: recall alone never satisfies the gate. A concept whose box or
    explain-back evidence would pass, but with no `applied` correct since the
    last miss, earns one reconfirm that is a small piece of code, not a
    question (`stale-reconfirm`, reason `no-applied-evidence`). That covers
    every concept tracked before the flag existed exactly once.
    """
    bloom = c.get("bloomLevel", 0)
    box = c.get("box", 1)
    last = parse_date(c.get("lastReviewed"))
    recent = last is not None and (t - last).days <= GATE_RECENCY_DAYS
    built = applied_evidence(c) >= 1
    if bloom == 0:
        return "no-opinion", "unclassified"   # never classified by a v3 writer; allow
    if bloom >= 3:
        if box >= 3:
            return ("satisfied", "box") if built else ("stale-reconfirm", "no-applied-evidence")
        if recent and evidence_at_3_plus(c) >= 2:
            return ("satisfied", "evidence") if built else ("stale-reconfirm", "no-applied-evidence")
        if recent:
            return "stale-reconfirm", "single-evidence"
        return "stale-reconfirm", "stale"    # Bloom ratchet alone is not current evidence
    # 1 <= bloom < 3: strong v2 retention evidence fallthrough (1.10.10)
    hist = [h.get("result") for h in c.get("reviewHistory", [])
            if not h.get("deferred")]
    if box >= 3 and len(hist) >= 2 and hist[-1] == hist[-2] == "correct":
        # gate-time read only; bloomLevel untouched. Reliable recall of a
        # concept never built with is still a code reconfirm, not a pass.
        return ("apply-equivalent", "box") if built else ("stale-reconfirm", "no-applied-evidence")
    return "gap", "below-apply"


def gate_status(c, t):
    return gate_verdict(c, t)[0]


def cmd_gate_check(args):
    sp, state = require_state(args.project)
    current = args.module or state.get("currentModule", "")
    if not str(current).strip():
        emit({"fires": False,
              "reason": "no currentModule set in state.json — set it via "
                        "touch-state --module before gating",
              "currentModule": current})
        return
    _, data = load_spaced_review(args.project, create=True)
    concepts = data.get("concepts", [])
    t = today()

    # Module entry is detected by graded ACTIVITY in the module, not by
    # tracked membership: /learn seeds every assessed sub-topic into its
    # module with add-concept on day 1, so "any concept tracked here" made a
    # seeded module a continuation session before it was ever taught and
    # the gate never fired on the journey it exists for (review finding 2b).
    in_module = [c for c in concepts
                 if norm_module_name(c.get("module", "")) == norm_module_name(current)]
    started = [c for c in in_module if any(
        not h.get("deferred") and h.get("result") in ("correct", "incorrect", "partial")
        for h in c.get("reviewHistory", []) if isinstance(h, dict))]
    if not concepts:
        emit({"fires": False, "reason": "first-ever session — nothing to gate against",
              "currentModule": current})
        return
    if started:
        emit({"fires": False,
              "reason": f"continuation session — {len(started)} concept(s) graded "
                        f"in module {current!r}",
              "currentModule": current})
        return
    seeded_only = len(in_module)

    # First session on a new module: identify prerequisites. A plan may
    # declare `none` — a module that builds on nothing tracked. Read as a
    # concept name it matched nothing and reported a vacuous "clear".
    if args.prereqs and re.match(r"\s*none\b", args.prereqs, re.IGNORECASE):
        emit({"fires": False,
              "reason": "the plan declares no prerequisites for this module",
              "currentModule": current, "prerequisiteSource": "declared"})
        return
    if args.prereqs:
        names = [n.strip().lower() for n in args.prereqs.split(",") if n.strip()]
        prereqs = [c for c in concepts
                   if c.get("name", "").strip().lower() in names]
        source = "declared"
    else:
        prior = args.prior_module or state.get("previousModule")
        if not prior:
            # 1.15.x: the old fallback here INFERRED a prior module by
            # string-sorting `introduced` dates — a weak heuristic that was
            # honestly flagged as inferred but still did real gating work in
            # real sessions. A noisy gate is worse than no gate (false
            # reconfirm questions erode trust in the real ones), so with
            # nothing declared and no tracked previousModule the gate now
            # declines to fire instead of guessing.
            emit({"fires": False,
                  "reason": "no prerequisites declared and no previousModule "
                            "tracked — pass --prereqs (the plan's "
                            "'Prerequisites for next module' line) or "
                            "--prior-module to gate this module",
                  "currentModule": current})
            return
        prereqs = [c for c in concepts
                   if norm_module_name(c.get("module", "")) == norm_module_name(str(prior))]
        source = "prior-module"

    report = []
    for c in prereqs:
        status, reason = gate_verdict(c, t)
        report.append({"name": c["name"], "bloomLevel": c.get("bloomLevel", 0),
                       **bloom_render(c.get("bloomLevel", 0)),
                       "box": c.get("box", 1), "lastReviewed": c.get("lastReviewed"),
                       "evidenceAt3Plus": evidence_at_3_plus(c),
                       "appliedEvidence": applied_evidence(c),
                       "evidenceKind": effective_kind(c),
                       "status": status, "reason": reason})
    gaps = [r for r in report if r["status"] == "gap"]
    stale = [r for r in report if r["status"] == "stale-reconfirm"]
    emit({"fires": True, "currentModule": current,
          "seededOnly": seeded_only,   # tracked in the module, none graded yet
          "prerequisiteSource": source,
          "prerequisites": report,
          "gaps": [r["name"] for r in gaps],
          "staleReconfirm": [r["name"] for r in stale],
          "verdict": ("offer" if gaps or stale else "clear")})


def cmd_migrate(args):
    """spaced-review.json v1/v2 -> v3. Idempotent, backed up, marker-writing.

    Replaces the prose 5f-bis procedure in /housekeep migrate. Preserves every
    non-canonical field by mutating the parsed JSON in place.
    """
    bdir = bodhi_dir(args.project)
    path = os.path.join(bdir, "spaced-review.json")
    if not os.path.exists(path):
        die(f"{path} does not exist")
    data = load_json(path)
    version = data.get("version", 1)
    concepts = data.get("concepts", [])
    if is_v3_complete(data):
        emit({"action": "noop", "reason": "already at v3 with all per-concept fields"})
        return

    # Snapshot THIS RUN's input before any mutation — post-write verification
    # compares against this, never against an on-disk backup that may predate
    # this run (1.11.1: a stale backup produced a false mismatch whose error
    # message advised a data-destroying restore).
    snapshot = json.loads(json.dumps(data))

    # Backup (never overwrite an existing backup — it may be the only pre-v3 copy).
    backup = ensure_pre_v3_backup(path)

    fields_added = upgrade_to_v3(data)
    data["version"] = 3
    write_json(path, data)

    # Post-write verification against this run's input snapshot.
    after = load_json(path)
    if len(after.get("concepts", [])) != len(snapshot.get("concepts", [])):
        die("verification failed: concept count changed during this run — "
            "the live file may be partially written; inspect it before "
            f"considering the backup at {backup}")
    for b, a in zip(snapshot.get("concepts", []), after.get("concepts", [])):
        lost = set(b.keys()) - set(a.keys())
        if lost:
            die(f"verification failed: concept {b.get('name')!r} lost fields "
                f"{sorted(lost)} during this run — inspect the live file; "
                f"backup at {backup}")

    marker = write_migration_marker(bdir, version, len(concepts), fields_added,
                                    "`bodhi-state migrate-spaced-review`")
    emit({"action": "migrated", "fromVersion": version, "toVersion": 3,
          "concepts": len(concepts), "fieldsAdded": fields_added,
          "backup": backup, "marker": marker})


def cmd_normalize(args):
    """One-shot repair of pre-1.11.0 executor drift (1.12.1). Idempotent.

    Patterns come from real learning projects: nested session bookkeeping,
    dict lastActivity/previousModule, plural *BloomLevels, duplicate
    sessionDates, invented reviewHistory results, invented sessionHistory
    types. Canonical fields are repaired; everything the drift invented is
    preserved (moved under *Legacy or kept in place) — learner data is sacred.
    """
    bdir = bodhi_dir(args.project)
    backup_dir = os.path.join(bdir, ".pre-normalize-backup")
    changes = []

    def backup(path):
        os.makedirs(backup_dir, exist_ok=True)
        dest = os.path.join(backup_dir, os.path.basename(path))
        if not os.path.exists(dest):  # first normalize wins; never overwrite
            shutil.copyfile(path, dest)

    sp = state_path(args.project)
    state = load_json(sp) if os.path.exists(sp) else None
    state_changes = 0
    if state is not None:
        for nested_key in ("session", "sessions"):
            nested = state.get(nested_key)
            if isinstance(nested, dict) and ("sessionDates" in nested
                                             or "totalSessions" in nested):
                for k in ("totalSessions", "currentStreak"):
                    if k in nested and k not in state:
                        state[k] = nested.pop(k)
                if "sessionDates" in nested:
                    lifted = nested.pop("sessionDates")
                    dates = list(state.get("sessionDates", []))
                    if isinstance(lifted, list):
                        dates += [d for d in lifted if isinstance(d, str)]
                    state["sessionDates"] = dates
                if not nested:
                    state.pop(nested_key)
                changes.append(f"lifted {nested_key!r} bookkeeping to top level")
                state_changes += 1
        dates = state.get("sessionDates")
        if isinstance(dates, list) and (len(dates) != len(set(dates))
                                        or dates != sorted(dates)):
            state["sessionDates"] = sorted(set(dates))
            changes.append("deduped and sorted sessionDates")
            state_changes += 1
        for key in ("lastActivity", "previousModule"):
            v = state.get(key)
            if v is not None and not isinstance(v, str):
                legacy_key = key + "Legacy"
                if legacy_key not in state:
                    state[legacy_key] = v
                if isinstance(v, dict):
                    s = v.get("name") or v.get("id") or v.get("result") or json.dumps(v)
                else:
                    s = str(v)
                state[key] = str(s)[:LAST_ACTIVITY_MAX]
                changes.append(f"stringified {key} (original kept in {legacy_key})")
                state_changes += 1
        # Scalar-but-wrong-type currentModule (int module numbers in the wild):
        # str() is lossless, no legacy copy needed.
        cm = state.get("currentModule")
        if cm is not None and not isinstance(cm, str):
            if isinstance(cm, (int, float)):
                state["currentModule"] = str(cm)
            else:
                if "currentModuleLegacy" not in state:
                    state["currentModuleLegacy"] = cm
                state["currentModule"] = (cm.get("name") or cm.get("id") or
                                          json.dumps(cm))[:LAST_ACTIVITY_MAX] if isinstance(cm, dict) else str(cm)[:LAST_ACTIVITY_MAX]
            changes.append("stringified currentModule")
            state_changes += 1
        for plural, singular in (("initialBloomLevels", "initialBloomLevel"),
                                 ("currentBloomLevels", "currentBloomLevel")):
            if plural in state and singular not in state:
                state[singular] = state.pop(plural)
                changes.append(f"renamed {plural} -> {singular}")
                state_changes += 1
        kinds = state.get("moduleEvidenceKinds")
        for holder, key in [(state, "evidenceKind")] + [
                (kinds, name) for name in (kinds if isinstance(kinds, dict) else {})]:
            fixed = _coerce_choice(holder.get(key), EVIDENCE_KINDS)
            if fixed is not None:
                changes.append(f"evidence kind {holder[key]!r} -> {fixed!r}")
                holder[key] = fixed
                state_changes += 1

    srp = sr_path(args.project)
    sr = load_json(srp) if os.path.exists(srp) else None
    sr_changes = 0
    if sr is not None:
        for c in sr.get("concepts", []):
            if not isinstance(c, dict):
                continue
            # Typed-field repairs from the shape table — only the lossless
            # ones (numeric strings, string booleans); anything else stays
            # for verify to report and a human to fix.
            for field, actual, expected, repair in concept_shape_errors(c):
                if repair is not None:
                    c[field] = repair
                    changes.append(f"{field} {actual!r} -> {repair!r} ({c.get('name')})")
                    sr_changes += 1
            if not isinstance(c.get("reviewHistory"), list):
                continue
            for h in c.get("reviewHistory", []):
                if not isinstance(h, dict):
                    continue
                for field, actual, expected, repair in history_entry_errors(h):
                    if repair is not None:
                        changes.append(f"reviewHistory {field} {actual!r} -> "
                                       f"{repair!r} ({c.get('name')})")
                        h[field] = repair
                        sr_changes += 1
                if h.get("deferred"):
                    continue
                res = h.get("result")
                if res is not None and res not in ("correct", "incorrect", "partial"):
                    h.pop("result")
                    h["deferred"] = True
                    h.setdefault("note", f"normalized from invented result {res!r}")
                    changes.append(f"reviewHistory result {res!r} -> deferral "
                                   f"({c.get('name')})")
                    sr_changes += 1
        for s in sr.get("sessionHistory", []):
            st = s.get("type")
            if isinstance(st, str) and st not in SESSION_TYPES:
                s["type"] = "other"
                s["subtype"] = st
                changes.append(f"sessionHistory type {st!r} -> other+subtype")
                sr_changes += 1

    if not changes:
        emit({"action": "noop", "reason": "nothing to normalize"})
        return
    if state_changes and state is not None:
        backup(sp)
        write_json(sp, state)
    if sr_changes and sr is not None:
        backup(srp)
        write_json(srp, sr)
    emit({"action": "normalized", "changes": changes, "backup": backup_dir})


def cmd_verify(args):
    """Schema sanity check. Used by dev/check.sh and the Stop hook."""
    errors, warnings = [], []
    module_kinds, tracked_modules = {}, set()
    project = args.project
    bdir = os.path.join(project, ".bodhi")
    if not os.path.isdir(bdir):
        die(f"no .bodhi/ under {project!r}")

    sp = os.path.join(bdir, "state.json")
    if os.path.exists(sp):
        try:
            state = load_json_raw(sp)
            if not isinstance(state, dict):
                raise TypeError(f"top level is {type(state).__name__}, expected object")
            if state.get("version") != 2:
                warnings.append(f"state.json version is {state.get('version')!r}, expected 2")
            for legacy in ("lastSessionSummary", "bloomResetNote"):
                if legacy in state:
                    errors.append(f"state.json carries v1 narrative field {legacy!r}")
            if not isinstance(state.get("sessionDates", []), list):
                errors.append("state.json sessionDates is not a list")
            la = state.get("lastActivity", "")
            if isinstance(la, str) and len(la) > LAST_ACTIVITY_MAX:
                warnings.append(
                    f"state.json lastActivity exceeds {LAST_ACTIVITY_MAX}-char "
                    f"guidance (state-ops KB)")
            # 1.12.1 drift checks — patterns found in real pre-1.11.0 data,
            # where executors invented a parallel schema that shadows the
            # canonical fields. All repairable by `bodhi-state normalize`.
            for nested in ("session", "sessions"):
                v = state.get(nested)
                if isinstance(v, dict) and ("sessionDates" in v or "totalSessions" in v):
                    errors.append(f"state.json nests session bookkeeping under "
                                  f"{nested!r} — canonical fields are top-level; "
                                  f"run `bodhi-state normalize`")
            for key in ("lastActivity", "previousModule", "currentModule"):
                v = state.get(key)
                if v is not None and not isinstance(v, str):
                    errors.append(f"state.json {key} is {type(v).__name__}, "
                                  f"expected string — run `bodhi-state normalize`")
            for plural in ("initialBloomLevels", "currentBloomLevels"):
                if plural in state:
                    warnings.append(f"state.json {plural!r} should be the singular "
                                    f"map — run `bodhi-state normalize`")
            dates = state.get("sessionDates")
            if isinstance(dates, list) and len(dates) != len(set(dates)):
                warnings.append("state.json sessionDates has duplicates — "
                                "run `bodhi-state normalize`")
            # Evidence kinds: readers treat a bad value as code, so it is an
            # error only here, where it can be seen.
            kinds = state.get("moduleEvidenceKinds")
            if kinds is not None and not isinstance(kinds, dict):
                errors.append(f"state.json moduleEvidenceKinds is "
                              f"{type(kinds).__name__}, expected an object of "
                              f"module name -> kind")
                kinds = None
            module_kinds = kinds or {}
            for label, k in [("evidenceKind", state.get("evidenceKind"))] + [
                    (f"moduleEvidenceKinds {name!r}", k) for name, k in module_kinds.items()]:
                if k is not None and k not in EVIDENCE_KINDS:
                    fix = _coerce_choice(k, EVIDENCE_KINDS)
                    errors.append(f"state.json {label} {k!r} not "
                                  f"{'|'.join(EVIDENCE_KINDS)} — readers treat it as code"
                                  + (" — run `bodhi-state normalize`" if fix else ""))
        except (json.JSONDecodeError, OSError) as e:
            errors.append(f"state.json unreadable: {e}")
        except Exception as e:  # drift the checks above did not anticipate
            errors.append(f"state.json structurally broken ({type(e).__name__}: {e}) "
                          f"— run `bodhi-state normalize`")
    else:
        errors.append("state.json missing")

    srp = os.path.join(bdir, "spaced-review.json")
    if os.path.exists(srp):
        try:
            sr = load_json_raw(srp)
            if not isinstance(sr, dict):
                raise TypeError(f"top level is {type(sr).__name__}, expected object")
            v = sr.get("version")
            if v not in (1, 2, 3):
                errors.append(f"spaced-review.json version is {v!r}")
            elif v != 3:
                warnings.append(f"spaced-review.json at v{v} — run "
                                "`bodhi-state migrate-spaced-review`")
            seen_names = {}
            concepts = sr.get("concepts", [])
            if not isinstance(concepts, list):
                errors.append(f"spaced-review.json concepts is "
                              f"{type(concepts).__name__}, expected list")
                concepts = []
            for i, c in enumerate(concepts):
                if not isinstance(c, dict):
                    errors.append(f"spaced-review concepts[{i}] is "
                                  f"{type(c).__name__}, expected object")
                    continue
                name = c.get("name")
                if not isinstance(name, str) or not name.strip():
                    errors.append(f"spaced-review concepts[{i}] name is "
                                  f"{name!r}, expected non-empty string")
                    name = f"index {i}"
                key = name.strip().lower()
                if key in seen_names:
                    errors.append(f"duplicate concept names (case-insensitive): "
                                  f"{seen_names[key]!r} and {name!r} — writers "
                                  f"match first-found; merge them")
                seen_names[key] = name
                for field, actual, expected, repair in concept_shape_errors(c):
                    if field == "name":
                        continue  # reported above with the index fallback
                    errors.append(shape_error_text(name, field, actual, expected, repair))
                if c.get("nextReview") is not None and parse_date(c.get("nextReview")) is None:
                    errors.append(f"concept {name!r} nextReview "
                                  f"{c.get('nextReview')!r} is not an ISO date — "
                                  f"the concept has silently left review rotation")
                if c.get("lastReviewed") is not None and parse_date(c.get("lastReviewed")) is None:
                    warnings.append(f"concept {name!r} lastReviewed "
                                    f"{c.get('lastReviewed')!r} is not an ISO date")
                if c.get("feynmanPassedAt") is not None and parse_date(c.get("feynmanPassedAt")) is None:
                    warnings.append(f"concept {name!r} feynmanPassedAt "
                                    f"{c.get('feynmanPassedAt')!r} is not an ISO date — "
                                    f"read as an undated explain-back")
                if v == 3:
                    for k in ("bloomLevel", "feynmanPassed", "consecutiveCorrectAtL4Plus"):
                        if k not in c:
                            errors.append(f"concept {name!r} missing v3 field {k!r}")
                if isinstance(c.get("module"), str):
                    tracked_modules.add(c["module"])
                rh = c.get("reviewHistory", [])
                if not isinstance(rh, list):
                    rh = []
                for h in rh:
                    if not isinstance(h, dict):
                        continue  # shape table reported it
                    for field, actual, expected, repair in history_entry_errors(h):
                        errors.append(shape_error_text(name, f"reviewHistory {field}",
                                                       actual, expected, repair)
                                      + ("" if repair is not None else " — fix it by hand"))
                    if "appliedBasis" in h and h.get("applied") is not True:
                        warnings.append(f"concept {name!r} reviewHistory "
                                        f"{h.get('date')!r} has appliedBasis "
                                        f"{h['appliedBasis']!r} without applied: "
                                        f"true — the basis is ignored")
                    if h.get("date") is not None and parse_date(h.get("date")) is None:
                        warnings.append(f"concept {name!r} reviewHistory date "
                                        f"{h.get('date')!r} is not an ISO date — "
                                        f"readers skip this entry")
                    if h.get("deferred"):
                        if "result" in h:
                            errors.append(f"concept {name!r} deferral entry carries "
                                          f"a result — deferral is scheduling, not "
                                          f"an outcome")
                        continue
                    hres = h.get("result")
                    if hres not in ("correct", "incorrect", "partial"):
                        errors.append(f"concept {name!r} reviewHistory result "
                                      f"{hres!r} not in canonical vocabulary "
                                      f"(correct|incorrect|partial) — run "
                                      f"`bodhi-state normalize`")
            sh = sr.get("sessionHistory", [])
            if not isinstance(sh, list):
                errors.append("spaced-review.json sessionHistory is not a list")
                sh = []
            for s in sh:
                if not isinstance(s, dict):
                    errors.append(f"sessionHistory entry {s!r} is not an object")
                    continue
                st = s.get("type")
                if st not in SESSION_TYPES:
                    errors.append(f"sessionHistory type {st!r} not in canonical vocabulary")
                elif st == "other" and not s.get("subtype"):
                    errors.append("sessionHistory 'other' entry missing subtype")
        except (json.JSONDecodeError, OSError) as e:
            errors.append(f"spaced-review.json unreadable: {e}")
        except Exception as e:
            errors.append(f"spaced-review.json structurally broken "
                          f"({type(e).__name__}: {e}) — run `bodhi-state normalize`")
    elif os.path.exists(os.path.join(bdir, "state.json")):
        warnings.append("spaced-review.json missing while state.json exists — "
                        "the review schedule may have been deleted")

    # Phase files nothing can parse: syllabus would read an empty plan.
    phases, plan = read_plan_modules(project)
    if phases and not plan:
        warnings.append(PLAN_UNREADABLE)

    # A module override joins on the normalized module name. Neither case
    # below can be repaired by a subcommand, so neither blocks the Stop hook.
    overridden = {norm_module_name(name) for name in module_kinds}
    known = {norm_module_name(m["name"]) for m in plan} | {
        norm_module_name(raw) for raw in tracked_modules}
    for name in module_kinds:
        if norm_module_name(name) not in known:
            warnings.append(f"state.json moduleEvidenceKinds {name!r} matches no plan "
                            f"module and no tracked concept (a renamed module?) — its "
                            f"kind reaches nothing; record it under the current name")
    by_id = {m["id"]: m for m in plan}
    for raw in sorted(tracked_modules):
        m = by_id.get(norm_module_name(raw))
        if m and norm_module_name(raw) not in overridden \
                and norm_module_name(m["name"]) in overridden:
            warnings.append(f"concepts filed under plan id {raw!r} do not get module "
                            f"{m['name']!r}'s evidence kind — the kind joins on the "
                            f"module name")

    pp = find_profile(project)
    if pp:
        try:
            profile = load_json_raw(pp)
            if profile.get("version") != 2:
                warnings.append(f"profile version is {profile.get('version')!r}, expected 2")
            for inline in ("activeProjects", "completedProjects"):
                if inline in profile:
                    errors.append(f"profile carries {inline!r} inline — belongs in "
                                  ".bodhi-profile.projects.json (v2 split)")
        except (json.JSONDecodeError, OSError) as e:
            errors.append(f"profile unreadable: {e}")
        except Exception as e:
            errors.append(f"profile structurally broken ({type(e).__name__}: {e})")

        # Project-list entry shape. Hand-edited by /learn and /evaluate; no
        # write path enforces it, so a dropped field would otherwise survive
        # silently until a cross-project skill read the entry and found a hole.
        lp = os.path.join(os.path.dirname(pp), ".bodhi-profile.projects.json")
        if os.path.exists(lp):
            try:
                plists = load_json_raw(lp)
                if plists.get("version") != 2:
                    warnings.append(f"profile projects list version is "
                                    f"{plists.get('version')!r}, expected 2")
                for key, required in (("activeProjects", PROFILE_ACTIVE_FIELDS),
                                      ("completedProjects", PROFILE_COMPLETED_FIELDS)):
                    entries = plists.get(key)
                    if entries is None:
                        continue
                    if not isinstance(entries, list):
                        errors.append(f"profile projects list {key!r} is "
                                      f"{type(entries).__name__}, expected list")
                        continue
                    for i, entry in enumerate(entries):
                        if not isinstance(entry, dict):
                            errors.append(f"{key}[{i}] is {type(entry).__name__}, "
                                          f"expected object")
                            continue
                        label = entry.get("name") or f"index {i}"
                        missing = sorted(required - set(entry))
                        if missing:
                            errors.append(
                                f"{key} entry {label!r} missing required "
                                f"field(s): {', '.join(missing)} — the project "
                                f"list is hand-edited by /learn and /evaluate; "
                                f"restore the field(s) per the state-schema KB")
            except (json.JSONDecodeError, OSError) as e:
                errors.append(f"profile projects list unreadable: {e}")
            except Exception as e:
                errors.append(f"profile projects list structurally broken "
                              f"({type(e).__name__}: {e})")

    result = {"ok": not errors, "project": project,
              "errors": errors, "warnings": warnings}
    print(json.dumps(result, indent=2))
    sys.exit(0 if not errors else 1)


# --- CLI --------------------------------------------------------------------

READ_ONLY_COMMANDS = {cmd_due, cmd_mastery, cmd_calibration, cmd_retention,
                      cmd_export_anonymized, cmd_session_brief, cmd_snapshot,
                      cmd_gate_check, cmd_verify, cmd_revision_brief,
                      cmd_syllabus}

def main():
    p = argparse.ArgumentParser(prog="bodhi-state", description=__doc__,
                                formatter_class=argparse.RawDescriptionHelpFormatter)
    p.add_argument("--project", default=".",
                   help="path to the learning project (the dir containing .bodhi/)")
    sub = p.add_subparsers(dest="cmd", required=True)

    s = sub.add_parser("add-concept", help="track a new concept (Box 1, review tomorrow)")
    s.add_argument("--concept", required=True)
    s.add_argument("--module", required=True)
    s.add_argument("--question", default="")
    s.add_argument("--bloom", type=int, default=0,
                   help="level an assessment classified it at (0-6; 0 = unclassified)")
    s.set_defaults(fn=cmd_add_concept)

    s = sub.add_parser("record-review",
                       help="record a retrieval outcome (Leitner + Bloom ratchet + history)")
    s.add_argument("--concept", required=True)
    s.add_argument("--result", required=True, choices=["correct", "incorrect", "partial"])
    s.add_argument("--tested-bloom", type=int, choices=range(0, 7), default=None,
                   help="Bloom level the question actually tested at")
    s.add_argument("--confidence", choices=sorted(CONFIDENCE_VALUES), default=None,
                   help="learner's pre-reveal confidence tag")
    s.add_argument("--module", default=None, help="auto-create the concept under this module if untracked")
    s.add_argument("--question", default="")
    s.add_argument("--note", default=None, help="short lastResult note")
    s.add_argument("--source", default=None, help="which skill produced this review")
    s.add_argument("--applied", nargs="?", const="observed", default=None,
                   choices=APPLIED_BASES,
                   help="the outcome was demonstrated in applied work, not an "
                        "explanation or quiz answer. Bare = observed: work the "
                        "tutor read (code that ran; off code, a saved file, with "
                        "--evidence). inspected (a photo the tutor examined) and "
                        "reported (the learner's word) only on a performed "
                        "concept. Feeds the gate and the mastery formula")
    s.add_argument("--evidence", default=None,
                   help="project-relative path of the applied work; required for "
                        "observed work off code, beside the exercise's key")
    s.add_argument("--retry", action="store_true",
                   help="successive-relearning retry: history entry only, no "
                        "box/counter/bloom movement (the original demotion stands)")
    s.set_defaults(fn=cmd_record_review)

    s = sub.add_parser("set-feynman", help="mark a concept's Feynman explain-back gate passed")
    s.add_argument("--concept", required=True)
    s.set_defaults(fn=cmd_set_feynman)

    s = sub.add_parser("record-session", help="append a sessionHistory entry (vocabulary-checked)")
    s.add_argument("--type", required=True)
    s.add_argument("--subtype", default=None)
    s.add_argument("--data", default=None, help="JSON object of optional fields")
    s.set_defaults(fn=cmd_record_session)

    s = sub.add_parser("record-assessment",
                       help="append an entry to assessment-history.json (append-only)")
    s.add_argument("--trigger", required=True)
    s.add_argument("--data", required=True, help="JSON object: topic, subTopics[], overallNote, predictionDelta")
    s.set_defaults(fn=cmd_record_assessment)

    s = sub.add_parser("forget", help="learner-initiated demote (box 1, counter reset, history)")
    s.add_argument("--concept", action="append", default=None,
                   help="exact concept name (repeatable; use for names containing commas)")
    s.add_argument("--concepts", default=None, help="comma-separated concept names")
    s.add_argument("--note", default=None)
    s.add_argument("--activity", default=None, help="override the state.json lastActivity line")
    s.set_defaults(fn=cmd_forget)

    s = sub.add_parser("park",
                       help="take a concept out of review rotation "
                            "(learner-deprioritized; --resume returns it)")
    s.add_argument("--concept", action="append", default=None,
                   help="exact concept name (repeatable)")
    s.add_argument("--resume", action="store_true",
                   help="return a parked concept to rotation (review tomorrow, box preserved)")
    s.add_argument("--note", default=None)
    s.set_defaults(fn=cmd_park)

    s = sub.add_parser("defer",
                       help="roll a due-but-unreviewed concept forward (no outcome invented)")
    s.add_argument("--concept", action="append", default=None,
                   help="exact concept name (repeatable)")
    s.add_argument("--days", type=int, default=1,
                   help="days to roll nextReview forward (default 1)")
    s.add_argument("--note", default=None)
    s.set_defaults(fn=cmd_defer)

    s = sub.add_parser("normalize",
                       help="repair pre-1.11.0 executor drift (backed up, idempotent)")
    s.set_defaults(fn=cmd_normalize)

    s = sub.add_parser("touch-state", help="update state.json session bookkeeping")
    s.add_argument("--activity", default=None, help="lastActivity one-liner (<=120 chars)")
    s.add_argument("--module", default=None, help="advance currentModule (records previousModule)")
    s.add_argument("--module-index", type=int, default=None)
    s.add_argument("--phase", default=None)
    s.add_argument("--completion", type=int, default=None)
    s.set_defaults(fn=cmd_touch_state)

    s = sub.add_parser("set-evidence-kind",
                       help="record the project's evidence kind, or one module's "
                            "(state.json only; every review is kept)")
    s.add_argument("--kind", default=None, help="|".join(EVIDENCE_KINDS))
    s.add_argument("--module", default=None,
                   help="a plan module's name or id, or a tracked module; omit for the project")
    s.add_argument("--clear", action="store_true",
                   help="remove the kind: the project reads as code, a module "
                        "takes the project's")
    s.set_defaults(fn=cmd_set_evidence_kind)

    s = sub.add_parser("profile-add-project",
                       help="append a schema-complete activeProjects entry "
                            "(creates .bodhi-profile.projects.json if missing)")
    s.add_argument("--name", required=True)
    s.add_argument("--topic", required=True)
    s.add_argument("--phase", default=None, help="currentPhase (default 1)")
    s.add_argument("--module", default=None, help="currentModule (default empty)")
    s.add_argument("--bloom", type=int, choices=range(0, 7), default=None,
                   help="starting bloomLevel (default 0)")
    s.add_argument("--pace", default=None, help="default steady")
    s.add_argument("--status", default=None, help="default active")
    s.add_argument("--track-purpose", default=None)
    s.set_defaults(fn=cmd_profile_add_project)

    s = sub.add_parser("profile-update-project",
                       help="refresh fields on an activeProjects entry in place")
    s.add_argument("--name", required=True)
    s.add_argument("--topic", default=None)
    s.add_argument("--phase", default=None, help="currentPhase")
    s.add_argument("--module", default=None, help="currentModule")
    s.add_argument("--bloom", type=int, choices=range(0, 7), default=None)
    s.add_argument("--pace", default=None)
    s.add_argument("--status", default=None)
    s.add_argument("--track-purpose", default=None)
    s.set_defaults(fn=cmd_profile_update_project)

    s = sub.add_parser("profile-complete-project",
                       help="move an entry activeProjects -> completedProjects "
                            "(learner-confirmed completion, or /learn replace-archive)")
    s.add_argument("--name", required=True)
    s.add_argument("--final-bloom", type=int, choices=range(0, 7), default=None,
                   help="finalBloomLevel (defaults to the entry's bloomLevel)")
    s.add_argument("--status", default=None,
                   help="optional note, e.g. 'archived: replaced by <new> on <date>'")
    s.set_defaults(fn=cmd_profile_complete_project)

    s = sub.add_parser("profile-update-patterns",
                       help="append persistent challenges / consistent strengths "
                            "from assessment-history counts (append-only, deduped)")
    s.set_defaults(fn=cmd_profile_update_patterns)

    s = sub.add_parser("bump-profile", help="increment a cumulativeStats counter")
    s.add_argument("--counter", required=True)
    s.set_defaults(fn=cmd_bump_profile)

    s = sub.add_parser("due", help="list concepts due for review")
    s.add_argument("--limit", type=int, default=None,
                   help="cap the listed concepts (full count still reported)")
    s.set_defaults(fn=cmd_due)

    s = sub.add_parser("mastery", help="per-module mastery + retention rollup (canonical formula)")
    s.set_defaults(fn=cmd_mastery)

    s = sub.add_parser("calibration", help="confidence-vs-outcome calibration summary")
    s.set_defaults(fn=cmd_calibration)

    s = sub.add_parser("retention",
                       help="retention-at-review rates by spacing gap and box (outcome data)")
    s.set_defaults(fn=cmd_retention)

    s = sub.add_parser("export-anonymized",
                       help="shareable anonymized stats: counts and rates only, no concept names or free text")
    s.set_defaults(fn=cmd_export_anonymized)

    s = sub.add_parser("session-brief",
                       help="mechanical branch detection for /teach: firstExposure, pretestApplies, isReteach (read-only)")
    s.add_argument("--concept", required=True)
    s.set_defaults(fn=cmd_session_brief)

    s = sub.add_parser("snapshot",
                       help="single-call dashboard rollup for /progress: position, cadence, due, mastery, calibration (read-only)")
    s.set_defaults(fn=cmd_snapshot)

    s = sub.add_parser("syllabus",
                       help="plan order × review evidence × position: every plan module in order with status and module evidence (read-only)")
    s.set_defaults(fn=cmd_syllabus)

    s = sub.add_parser("revision-brief",
                       help="today's studied concepts + the revision sheet file to write (read-only)")
    s.set_defaults(fn=cmd_revision_brief)

    s = sub.add_parser("gate-check", help="prerequisite Bloom gate verdict for /teach Phase 1")
    s.add_argument("--module", default=None, help="module to gate (default: state.json currentModule)")
    s.add_argument("--prior-module", default=None)
    s.add_argument("--prereqs", default=None,
                   help="comma-separated declared prerequisite concepts (from the plan file)")
    s.set_defaults(fn=cmd_gate_check)

    s = sub.add_parser("migrate-spaced-review",
                       help="one-shot spaced-review.json v1/v2 -> v3 (backed up, idempotent)")
    s.set_defaults(fn=cmd_migrate)

    s = sub.add_parser("verify", help="schema sanity check (lint + Stop hook)")
    s.set_defaults(fn=cmd_verify)

    args = p.parse_args()
    # One exclusive lock per project for the whole read-mutate-write run —
    # closes the two-terminal lost-update race (1.11.1). Profile writes lock
    # the profile's directory too.
    shared = args.fn in READ_ONLY_COMMANDS
    bdir = os.path.join(args.project, ".bodhi")
    if os.path.isdir(bdir):
        acquire_lock(bdir, shared=shared)
    profile = find_profile(args.project)
    if profile:
        acquire_lock(os.path.dirname(profile), shared=shared)
    install_kinds(args.project)
    args.fn(args)


if __name__ == "__main__":
    main()
