model.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/registry/roadmap/reconcile/model.py
 
# OCT-LINT: disable=tests_exist reason="Covered by oc_chat/tests/registry/roadmap/test_reconcile.py"
"""
Result types for the roadmap reconciliation check (Plan 205).
 
Purpose
-------
Carry the findings, counts and plan-link join rows the reconciliation check
produces, in a shape that is **visibly distinguishable from drift**.
 
Responsibilities
----------------
- Define the finding vocabulary (``FINDING_KINDS``) and the two severities.
- Carry the plan-to-item join rows the check derives from ``advances:``.
- Expose ``ReconcileReport.to_dict()`` — the JSON surface, which names the
  question it answers so a reader can never mistake it for the drift check.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - report construction lifecycle
    L3 - counts and severity summaries
    L4 - per-finding construction
 
Contracts
---------
- Every type here is frozen and side-effect free.
- ``ok`` reflects the *advisory* landing mode: it stays True with violations
  present. ``would_fail`` is the signal a failing mode would act on. Callers
  choosing to fail must read ``would_fail``, never ``ok``.
- ``question`` is never the string ``"drift"``; the two checks answer different
  questions and conflating them is what Plan 205 exists to prevent.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
"""
from __future__ import annotations
 
from dataclasses import dataclass, field
from typing import Literal, Mapping, Sequence
 
from oc_diagnostics import _dbg
 
_dbg("OC_CHAT.MAIN", "module", "loaded", 4)
 
Severity = Literal["violation", "advisory"]
 
#: The question this check answers, stamped into every payload. The drift check
#: answers "do the rendered views match the source records?"; this one answers
#: "do the source records match what actually landed?".
QUESTION = "source-records-vs-what-landed"
 
#: The one-paragraph disclaimer rendered above human output, so a reader never
#: again reads a green drift line as "the roadmap is true".
NOT_DRIFT_BANNER = (
    "roadmap reconciliation — this is NOT the drift check\n"
    "  drift : rendered views vs source records (internal consistency of two\n"
    "          artifacts generated seconds apart; cannot see reality)\n"
    "  this  : source records vs what actually landed (claims vs state)"
)
 
#: Finding kinds, grouped by the plan's three classes.
FINDING_KINDS: tuple[str, ...] = (
    # class 1 — changelog-vs-status divergence
    "changelog_status_divergence",
    # class 2 — advances: conformance
    "advances_missing",
    "advances_not_advanced",
    "advances_unknown_record",
    # class 3 — stale-completion drift (advisory by construction)
    "stale_completion",
    "stale_deliverables",
    # input quality — reported separately so a corpus the check could not read
    # never masquerades as a corpus that read clean
    "plan_frontmatter_unparseable",
)
 
#: Kinds that can never be violations, whatever the evidence looks like.
ADVISORY_ONLY_KINDS: frozenset[str] = frozenset(
    {"stale_completion", "stale_deliverables", "plan_frontmatter_unparseable"}
)
 
 
@dataclass(frozen=True)
class Finding:
    """One reconciliation finding: a claim that the roadmap state does not back."""
 
    kind: str
    severity: Severity
    subject: str
    detail: str
    source_ref: str
    evidence: Mapping[str, object] = field(default_factory=dict)
 
    def to_dict(self) -> dict[str, object]:
        """The JSON projection of this finding."""
 
        return {
            "kind": self.kind,
            "severity": self.severity,
            "subject": self.subject,
            "detail": self.detail,
            "source_ref": self.source_ref,
            "evidence": dict(self.evidence),
        }
 
 
@dataclass(frozen=True)
class PlanLink:
    """A plan-to-roadmap-item join row, derived from a plan's ``advances:`` field.
 
    This is the row Plan 205 wants living in ``roadmap_plan_links`` rather than
    in prose. ``contribution`` follows the schema's CHECK vocabulary: ``full``
    when the advanced record is terminal-done, ``partial`` otherwise.
    """
 
    plan_id: str
    item_id: str
    contribution: Literal["full", "partial", "context_only"]
    plan_path: str
 
    def to_dict(self) -> dict[str, str]:
        """The JSON projection of this join row."""
 
        return {
            "plan_id": self.plan_id,
            "item_id": self.item_id,
            "contribution": self.contribution,
            "plan_path": self.plan_path,
        }
 
 
@dataclass(frozen=True)
class StatusTransition:
    """A status movement observed in git history, keyed by roadmap record id."""
 
    record_id: str
    from_status: str | None
    to_status: str
    sha: str
    date: str
 
    def to_dict(self) -> dict[str, object]:
        """The JSON projection of this transition."""
 
        return {
            "record_id": self.record_id,
            "from_status": self.from_status,
            "to_status": self.to_status,
            "sha": self.sha,
            "date": self.date,
        }
 
 
@dataclass(frozen=True)
class ReconcileReport:
    """The full result of one reconciliation run."""
 
    findings: tuple[Finding, ...]
    links: tuple[PlanLink, ...]
    transitions: tuple[StatusTransition, ...]
    counts: Mapping[str, int]
    advances_none_count: int
    plans_checked: int
    changelog_records_checked: int
    records_checked: int
    unscannable_ids: tuple[str, ...]
    question: str = QUESTION
    #: Set when git history could not be read at all.
    history_error: str | None = None
    #: Set when class 1 examined nothing although there was something to examine
    #: — a window or ``--source`` pathspec that matched no commits. Distinct from
    #: ``history_error``: git worked, it just was not pointed at anything.
    changelog_scope_error: str | None = None
 
    @property
    def changelog_class_evaluated(self) -> bool:
        """False when class 1 examined nothing, for ANY reason.
 
        A zero class-1 count with this False means "not checked", NOT "clean".
        Keeping those two indistinguishable is the failure this plan exists to
        eliminate — it is how a green drift line let the roadmap rot for three
        weeks, and the tool must not reproduce it.
        """
 
        return self.history_error is None and self.changelog_scope_error is None
 
    @property
    def changelog_not_evaluated_reason(self) -> str | None:
        """Why class 1 examined nothing, or None when it really did run."""
 
        return self.history_error or self.changelog_scope_error
 
    @property
    def violation_count(self) -> int:
        """How many findings a failing mode would act on."""
 
        return sum(1 for finding in self.findings if finding.severity == "violation")
 
    @property
    def advisory_count(self) -> int:
        """How many findings are advisory-only."""
 
        return sum(1 for finding in self.findings if finding.severity == "advisory")
 
    @property
    def counts_by_severity(self) -> dict[str, dict[str, int]]:
        """Per-kind ``{violation, advisory}`` split.
 
        The flat per-kind count alone is misleading once a kind can carry both
        severities: ``advances_missing: 300`` reads as 300 violations when they
        may all be grandfathered advisories.
        """
 
        split = {kind: {"violation": 0, "advisory": 0} for kind in FINDING_KINDS}
        for finding in self.findings:
            bucket = split.setdefault(finding.kind, {"violation": 0, "advisory": 0})
            bucket[finding.severity] = bucket.get(finding.severity, 0) + 1
        return split
 
    @property
    def would_fail(self) -> bool:
        """True when a failing mode would exit non-zero on this run."""
 
        return self.violation_count > 0
 
    @property
    def ok(self) -> bool:
        """Always True — the check lands advisory (see the module Contracts)."""
 
        return True
 
    def to_dict(self) -> dict[str, object]:
        """The JSON surface, stamped with the question this check answers."""
 
        return {
            "ok": self.ok,
            "question": self.question,
            "not_drift": True,
            "would_fail": self.would_fail,
            "violations": self.violation_count,
            "advisories": self.advisory_count,
            "counts": dict(self.counts),
            "counts_by_severity": self.counts_by_severity,
            "history_error": self.history_error,
            "changelog_class_evaluated": self.changelog_class_evaluated,
            "changelog_not_evaluated_reason": self.changelog_not_evaluated_reason,
            "advances_none": self.advances_none_count,
            "checked": {
                "plans": self.plans_checked,
                "changelog_records": self.changelog_records_checked,
                "roadmap_records": self.records_checked,
            },
            "unscannable_ids": list(self.unscannable_ids),
            "findings": [finding.to_dict() for finding in self.findings],
            "links": [link.to_dict() for link in self.links],
            "transitions": [transition.to_dict() for transition in self.transitions],
        }
 
 
def count_by_kind(findings: Sequence[Finding]) -> dict[str, int]:
    """Count ``findings`` by kind, with every known kind present (zeros included).
 
    Zeros are emitted deliberately: a kind silently absent from the payload reads
    as "not checked", which is the same confusion Plan 205 exists to remove.
    """
 
    func = "count_by_kind"
    counts = {kind: 0 for kind in FINDING_KINDS}
    for finding in findings:
        counts[finding.kind] = counts.get(finding.kind, 0) + 1
    _dbg("OC_CHAT.MAIN", func, f"counted {len(findings)} findings", 3)
    return counts