legal-ai/mcp-server/src/legal_mcp/case_status_model.py

"""Single source of truth for the case-status lifecycle (the *status model*).

Every consumer derives its order / labels / phase from this one registry:
  • ``CaseStatus`` enum (models.py) — the type-level key set (kept consistent
    by the assertion at the bottom of this module);
  • the forward-only ``STATUS_ORDER`` guard in ``tools/cases.py``;
  • the ``GET /api/status-model`` endpoint (web/app.py) that the frontend
    mirror (``web-ui/src/lib/api/case-status.ts``) is generated against.

To make a status *do something* on entry (notify, kick a job, …), set its
``on_enter`` to an action key — the single place where a status's behaviour is
declared. ``tools/cases.py`` dispatches it when a case transitions into that
status (forward-only), so the behaviour lives next to the definition.

The five **phases** are the coarse pipeline the 12 statuses collapse into for the
header stepper. Intermediate analyst/research states (``analyst_verified``,
``research_complete``) are first-class canonical statuses here — the agents set
them, so they must be in the model rather than fall between the canonical set
and the legacy bucket (the bug that made the chip, stepper and manual-changer
disagree).
"""

from __future__ import annotations

from dataclasses import dataclass

# Ordered 5-phase pipeline (key → Hebrew label) — coarse view of the lifecycle.
PHASES: list[tuple[str, str]] = [
    ("intake", "קליטה ועיבוד"),
    ("prep", "הכנת תיק"),
    ("thinking", "ניתוח וכיוון"),
    ("writing", "כתיבת טיוטה"),
    ("done", "סגירה"),
]
PHASE_LABELS: dict[str, str] = dict(PHASES)


@dataclass(frozen=True)
class StatusDef:
    key: str
    label: str            # Hebrew label (chip / dropdown)
    description: str       # Hebrew one-liner (status guide)
    phase: str            # one of PHASES keys
    selectable: bool = True   # offered in the manual status-changer dropdown
    terminal: bool = False
    on_enter: str | None = None  # future: action key dispatched on entry


# THE lifecycle — ordered. STATUS_ORDER, the enum and the frontend all derive
# from this list. Insert intermediates in workflow order.
STATUS_DEFS: list[StatusDef] = [
    StatusDef("new", "חדש", "התיק נוצר וממתין להעלאת מסמכים", "intake"),
    StatusDef("processing", "בעיבוד", "המערכת מעבדת ומנתחת את המסמכים", "intake"),
    StatusDef("documents_ready", "מסמכים מוכנים", "כל המסמכים עובדו ומוכנים לעבודה", "prep"),
    StatusDef("analyst_verified", "ניתוח אומת", "המנתח סיים ואימת את הניתוח — ממתין להכרעת תוצאה של היו״ר", "thinking"),
    StatusDef("research_complete", "מחקר הושלם", "חקר התקדימים הושלם (מסלול נפרד מהמנתח)", "thinking"),
    StatusDef("outcome_set", "תוצאה נקבעה", "נקבעה תוצאה צפויה לערר", "thinking"),
    StatusDef("direction_approved", "כיוון אושר", "כיוון ההחלטה אושר — בהעמקת ניתוח וכתיבה", "thinking"),
    StatusDef("qa_review", "בדיקת איכות", "הטיוטה בבדיקת איכות אוטומטית", "writing"),
    StatusDef("drafted", "טיוטה", "טיוטה מוכנה לעיון", "writing"),
    StatusDef("exported", "יוצא", "ההחלטה יוצאה לקובץ DOCX", "done"),
    StatusDef("reviewed", "נבדק", 'ההחלטה נבדקה ע"י היו"ר', "done"),
    StatusDef("final", "סופי", "החלטה סופית — מוכנה להגשה", "done", terminal=True),
]

BY_KEY: dict[str, StatusDef] = {d.key: d for d in STATUS_DEFS}
STATUS_ORDER: list[str] = [d.key for d in STATUS_DEFS]


def phase_of(status: str | None) -> str | None:
    """The pipeline phase a status belongs to (None for unknown values)."""
    d = BY_KEY.get(status or "")
    return d.phase if d else None


def label_of(status: str | None) -> str:
    d = BY_KEY.get(status or "")
    return d.label if d else (status or "")


def to_dict() -> dict:
    """Serialisable status model for ``GET /api/status-model`` (frontend SSoT)."""
    return {
        "statuses": [
            {
                "key": d.key,
                "label": d.label,
                "description": d.description,
                "phase": d.phase,
                "selectable": d.selectable,
                "terminal": d.terminal,
                "on_enter": d.on_enter,
            }
            for d in STATUS_DEFS
        ],
        "phases": [{"key": k, "label": v} for k, v in PHASES],
    }


# Drift guard — the registry IS the canonical key set; the CaseStatus enum must
# match it exactly. Importing here (models.py never imports this module) is safe.
def _assert_consistent() -> None:
    from legal_mcp.models import CaseStatus

    registry = {d.key for d in STATUS_DEFS}
    enum_keys = {s.value for s in CaseStatus}
    if registry != enum_keys:
        raise RuntimeError(
            "case_status_model drift: registry vs CaseStatus enum differ — "
            f"only-in-registry={registry - enum_keys}, only-in-enum={enum_keys - registry}"
        )
    phase_keys = {k for k, _ in PHASES}
    bad = {d.key: d.phase for d in STATUS_DEFS if d.phase not in phase_keys}
    if bad:
        raise RuntimeError(f"case_status_model: statuses with unknown phase: {bad}")


_assert_consistent()