feat(status): single source of truth for the case-status model + canonicalize intermediates
All checks were successful
G12 Leak-Guard / leak-guard (pull_request) Successful in 4s
Lint — undefined names / undefined-names (pull_request) Successful in 11s

Fixes the case where the H1 chip, the pipeline stepper and the manual-changer
dropdown disagreed (e.g. 1043-02-26 on analyst_verified): the status sat between
the canonical 10 and a legacy bucket, so each surface read a different list.

Root fix — ONE authoritative status model + canonicalize the real intermediate
states (chair-approved):

- New mcp-server/.../case_status_model.py — the single registry: ordered
  StatusDef list (key/label/description/phase/selectable/terminal/on_enter),
  the 5 phases, STATUS_ORDER, phase_of/label_of, and a drift assertion that the
  CaseStatus enum matches it. `on_enter` is the seam for a status to *do*
  something on entry, declared next to its definition (no dispatcher wired yet).
- models.CaseStatus enum: analyst_verified + research_complete promoted to
  first-class canonical statuses (the agents set them — they belong in the
  model, not a legacy fallback).
- tools/cases.py: the forward-only STATUS_ORDER guard now derives from the
  registry instead of an inline list.
- GET /api/status-model exposes the model so the frontend mirror can be
  generated against it.
- web-ui case-status.ts mirrors it: the two intermediates moved from the
  legacy map into CASE_STATUSES / PHASES(thinking) / STATUS_LABELS /
  STATUS_DESCRIPTIONS; status-badge icons+tones extended. So the chip (status),
  stepper (its phase) and dropdown (now includes it) all agree.

Invariants: G2 — collapses the scattered status definitions (enum, inline
STATUS_ORDER, two frontend label maps) onto one backend authority + a documented
frontend mirror; no parallel status list remains. Agent-prompt alignment +
backfill follow in a separate change. py_compile + tsc + eslint clean.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-06-30 19:31:21 +00:00
parent 0bf311041b
commit 44f6915f0c
6 changed files with 170 additions and 34 deletions

View File

@@ -0,0 +1,118 @@
"""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()

View File

@@ -9,13 +9,18 @@ from uuid import UUID
from pydantic import BaseModel, Field
# Core case lifecycle — kept in sync with STATUS_ORDER in tools/cases.py and the
# frontend SSoT web-ui/src/lib/api/case-status.ts. Trimmed from 17 → 10 (the
# decorative mid-stage markers that no pipeline code ever set were removed).
# Core case lifecycle. The canonical key set, order, labels, phases and (future)
# per-status actions all live in ONE place — legal_mcp/case_status_model.py —
# which asserts this enum matches it. The frontend mirror is
# web-ui/src/lib/api/case-status.ts (generated against GET /api/status-model).
# The analyst/research intermediate states are first-class canonical statuses
# (the agents set them) — not legacy.
class CaseStatus(str, enum.Enum):
NEW = "new"
PROCESSING = "processing"
DOCUMENTS_READY = "documents_ready"
ANALYST_VERIFIED = "analyst_verified"
RESEARCH_COMPLETE = "research_complete"
OUTCOME_SET = "outcome_set"
DIRECTION_APPROVED = "direction_approved"
QA_REVIEW = "qa_review"

View File

@@ -13,6 +13,7 @@ from uuid import UUID
import httpx
from legal_mcp import config
from legal_mcp.case_status_model import STATUS_ORDER # status SSoT
from legal_mcp.services import audit, db, extractor, git_sync, practice_area as pa
from legal_mcp.tools.envelope import empty, err, ok # GAP-48: SSoT envelope
@@ -341,16 +342,8 @@ async def case_update(
"""
from datetime import date as date_type
# Ordered core lifecycle — regression protection (forward-only).
# Single source of truth, mirrored by web-ui/src/lib/api/case-status.ts and
# models.CaseStatus. Trimmed from 17 → 10 (decorative statuses removed).
STATUS_ORDER = [
"new", "processing", "documents_ready",
"outcome_set", "direction_approved",
"qa_review", "drafted",
"exported", "reviewed", "final",
]
# Ordered core lifecycle (forward-only regression guard). Single source of
# truth: legal_mcp/case_status_model.py — STATUS_ORDER/BY_KEY derive from it.
case = await db.get_case_by_number(case_number)
if not case:
return err(f"תיק {case_number} לא נמצא.")
@@ -363,6 +356,9 @@ async def case_update(
# Only update if advancing or status is unknown to the order
if new_idx >= cur_idx or new_idx == -1:
fields["status"] = status
# per-status on_enter hook (case_status_model) — the single place a
# status's behaviour is declared. No dispatcher wired yet; when the
# first action is added, dispatch BY_KEY[status].on_enter here.
if title:
fields["title"] = title
if subject: