state-hub/scripts/quality_assessment.py

#!/usr/bin/env python3
"""DoX quality assessment helpers (STATE-WP-0077).

Canonical recording fields and parsing for Definition assessments.
See docs/work-record-quality-gates.md.
"""
from __future__ import annotations

import re
from dataclasses import dataclass
from typing import Any

POLICIES = ("DoC", "DoR", "DoD")

FIELD_FOR_POLICY = {
    "DoC": "quality_doc",
    "DoR": "quality_dor",
    "DoD": "quality_dod",
}


def _norm_policy(policy: str) -> str:
    p = policy.strip()
    upper = p.upper().replace(" ", "").replace("_", "")
    if upper in ("DOC", "COMPREHENSION", "DEFINITIONOFCOMPREHENSION"):
        return "DoC"
    if upper in ("DOR", "READY", "DEFINITIONOFREADY"):
        return "DoR"
    if upper in ("DOD", "DONE", "DEFINITIONOFDONE"):
        return "DoD"
    if p in POLICIES:
        return p
    raise ValueError(f"unknown policy {policy!r}; use DoC, DoR, or DoD")


def badge(policy: str, outcome: str) -> str:
    """Map policy + Ok|Failed (or full badge) to DoX-Ok / DoX-Failed."""
    pol = _norm_policy(policy)
    out = outcome.strip().strip('"')
    if out in (f"{pol}-Ok", f"{pol}-Failed"):
        return out
    if out.lower() in ("ok", "pass", "passed"):
        return f"{pol}-Ok"
    if out.lower() in ("failed", "fail", "no"):
        return f"{pol}-Failed"
    raise ValueError(f"unknown outcome {outcome!r} for {pol}")


def assessment_from_mapping(meta: dict, policy: str) -> str | None:
    """Read assessment badge for policy from frontmatter / YAML / task fields."""
    pol = _norm_policy(policy)
    key = FIELD_FOR_POLICY[pol]
    raw = meta.get(key)
    if raw is None:
        return None
    s = str(raw).strip().strip('"').strip("'")
    if not s or s.lower() in ("unassessed", "none", "null", "~", "-"):
        return None
    try:
        return badge(pol, s)
    except ValueError:
        return None


def is_ok(badge_value: str | None) -> bool:
    return bool(badge_value) and str(badge_value).endswith("-Ok")


@dataclass(frozen=True)
class QualityDebtItem:
    kind: str
    record_id: str
    lifecycle_status: str
    missing: str
    path: str
    detail: str = ""


def debt_for_workplan_meta(meta: dict, *, path: str) -> list[QualityDebtItem]:
    """Quality debt for a workplan frontmatter dict."""
    items: list[QualityDebtItem] = []
    wp_id = str(meta.get("id", "")).strip() or path
    status = str(meta.get("status", "")).strip().lower()
    dor = assessment_from_mapping(meta, "DoR")
    dod = assessment_from_mapping(meta, "DoD")
    if status == "ready" and not is_ok(dor):
        items.append(
            QualityDebtItem(
                kind="workplan",
                record_id=wp_id,
                lifecycle_status=status,
                missing="DoR-Ok",
                path=path,
                detail=f"quality_dor={dor or 'unassessed'}",
            )
        )
    if status in ("finished", "completed") and not is_ok(dod):
        items.append(
            QualityDebtItem(
                kind="workplan",
                record_id=wp_id,
                lifecycle_status=status,
                missing="DoD-Ok",
                path=path,
                detail=f"quality_dod={dod or 'unassessed'}",
            )
        )
    return items


def debt_for_intake_meta(meta: dict, *, path: str) -> list[QualityDebtItem]:
    """Quality debt for intake blocks in vetted/routed without DoC-Ok."""
    items: list[QualityDebtItem] = []
    iid = str(meta.get("id", "")).strip() or path
    status = str(meta.get("status", "")).strip().lower()
    doc = assessment_from_mapping(meta, "DoC")
    if status in ("vetted", "routed") and not is_ok(doc):
        items.append(
            QualityDebtItem(
                kind="intake",
                record_id=iid,
                lifecycle_status=status,
                missing="DoC-Ok",
                path=path,
                detail=f"quality_doc={doc or 'unassessed'}",
            )
        )
    return items


def intake_has_doc_ok(intake: dict) -> bool:
    """Soft check: hub intake notes/description record DoC-Ok."""
    chunks: list[str] = []
    for key in ("description", "routed_note", "title"):
        v = intake.get(key)
        if v:
            chunks.append(str(v))
    for note in intake.get("notes") or []:
        if isinstance(note, dict):
            chunks.append(str(note.get("content") or ""))
        else:
            chunks.append(str(note))
    # structured future field
    if assessment_from_mapping(intake, "DoC") and is_ok(assessment_from_mapping(intake, "DoC")):
        return True
    text = "\n".join(chunks)
    if re.search(r"\bDoC-Ok\b", text):
        return True
    if re.search(r"quality_doc\s*[:=]\s*[\"']?(DoC-Ok|Ok)\b", text, re.I):
        return True
    return False


def progress_event_body(
    *,
    policy: str,
    outcome: str,
    record_kind: str,
    record_id: str,
    assessed_by: str | None = None,
    note: str | None = None,
    topic_id: str | None = None,
    workplan_id: str | None = None,
    author: str | None = None,
) -> dict[str, Any]:
    """JSON body for POST /progress/ recording a quality assessment."""
    b = badge(policy, outcome)
    pol = b.rsplit("-", 1)[0]
    body: dict[str, Any] = {
        "event_type": "quality_assessment",
        "summary": f"{b} for {record_kind} {record_id}",
        "author": author or assessed_by or "agent",
        "detail": {
            "policy": pol,
            "outcome": "Ok" if b.endswith("-Ok") else "Failed",
            "badge": b,
            "record_kind": record_kind,
            "record_id": record_id,
            "assessed_by": assessed_by,
            "note": note,
        },
    }
    if topic_id:
        body["topic_id"] = topic_id
    if workplan_id:
        body["workplan_id"] = workplan_id
    return body