fix: case registry locking, watchdog override verification, export paths

courtroom/portal/export_transcript.py

@@ -206,6 +206,21 @@ def apply_personality_styling(html: str) -> str:
     return html
 
 
+def resolve_transcript_path(raw: str) -> Path:
+    """Resolve transcript path relative to courtroom/ (BASE_DIR).
+
+    Accepts paths relative to repo root (courtroom/transcripts/foo.md),
+    courtroom/ (transcripts/foo.md), or absolute paths.
+    """
+    src = Path(raw)
+    if src.is_absolute():
+        return src
+    rel = raw.lstrip("/")
+    if rel.startswith("courtroom/"):
+        rel = rel[len("courtroom/") :]
+    return BASE_DIR / rel
+
+
 def extract_title(md_path: Path, content: str) -> str:
     """Derive a human-readable title from path or first H1."""
     m = re.search(r"^#\s+(.+)$", content, re.MULTILINE)
@@ -229,9 +244,7 @@ def main() -> int:
     parser.add_argument("-o", "--output", help="Output HTML path (default: portal/exports/<basename>.html)")
     args = parser.parse_args()
 
-    src = Path(args.transcript)
-    if not src.is_absolute():
-        src = BASE_DIR / args.transcript.lstrip("/")
+    src = resolve_transcript_path(args.transcript)
     if not src.exists():
         print(f"Error: File not found: {src}", file=sys.stderr)
         return 1

executive/watchdog.py

@@ -119,13 +119,27 @@ def run_checks() -> tuple[bool, list[str]]:
 
     for a in actions:
         rid = a.get("ruling_id", "")
-        if rid and rid not in approved and "override" not in a.get("action_type", "").lower():
-            # Action references ruling not in judicial log
-            proof = a.get("override_proof")
+        action_type = a.get("action_type", "").lower()
+        proof = a.get("override_proof")
+
+        if "override" in action_type:
             if not proof:
+                alerts.append(f"Override action without proof: ruling_id={rid}")
+                continue
+            try:
+                from executive.proof import OverrideProof
+
+                reason = a.get("description", "")
+                if OverrideProof.verify(proof, rid, reason) is None:
+                    alerts.append(f"Invalid override proof: ruling_id={rid}")
+            except FileNotFoundError:
                 alerts.append(
-                    f"Action without judicial approval and no override proof: ruling_id={rid}"
+                    f"Cannot verify override proof (secret missing): ruling_id={rid}"
                 )
+        elif rid and rid not in approved:
+            alerts.append(
+                f"Action without judicial approval and no override proof: ruling_id={rid}"
+            )
 
     # (c) Override frequency
     overrides = [a for a in actions if a.get("override_proof")]

litigation/run.py

@@ -12,6 +12,11 @@
 from datetime import datetime
 from pathlib import Path
 
+try:
+    import fcntl
+except ImportError:  # Windows — no cross-process locking
+    fcntl = None  # type: ignore[assignment,misc]
+
 # Add project root for imports
 REPO_ROOT = Path(__file__).resolve().parent.parent
 sys.path.insert(0, str(REPO_ROOT))
@@ -62,30 +67,71 @@ def slugify(text: str) -> str:
 REGISTRY_PATH = REPO_ROOT / "courtroom" / "case-registry.yaml"
 
 
+def _lock_registry_file(file_obj) -> None:
+    if fcntl is not None:
+        fcntl.flock(file_obj.fileno(), fcntl.LOCK_EX)
+
+
+def _unlock_registry_file(file_obj) -> None:
+    if fcntl is not None:
+        fcntl.flock(file_obj.fileno(), fcntl.LOCK_UN)
+
+
 def allocate_case_no(category: str = "DEL", *, deliberation: int = 1) -> str:
     """
     Allocate the next Case No. from courtroom/case-registry.yaml.
 
     Per core/case-format.md: registry holds next available NNN per category.
+    Uses an exclusive file lock so concurrent saves cannot corrupt the registry
+    or assign duplicate Case Nos. Creates the registry when missing and resets
+    per-category counters when the calendar year advances past the registry year.
     """
     import yaml
 
-    year = datetime.now().strftime("%Y")
+    calendar_year = datetime.now().strftime("%Y")
+    REGISTRY_PATH.parent.mkdir(parents=True, exist_ok=True)
     if not REGISTRY_PATH.exists():
-        return f"{year}-{category}-001-{deliberation:03d}"
-
-    data = yaml.safe_load(REGISTRY_PATH.read_text(encoding="utf-8")) or {}
-    year = str(data.get("year", year))
-    categories = data.setdefault("categories", {})
-    nnn = int(categories.get(category, 1))
-    case_no = f"{year}-{category}-{nnn:03d}-{deliberation:03d}"
-    categories[category] = nnn + 1
-    data["categories"] = categories
-    data["last_updated"] = datetime.now().strftime("%Y-%m-%d")
-    REGISTRY_PATH.write_text(
-        yaml.dump(data, default_flow_style=False, sort_keys=False),
-        encoding="utf-8",
-    )
+        REGISTRY_PATH.write_text(
+            yaml.dump(
+                {
+                    "year": int(calendar_year),
+                    "categories": {},
+                    "last_updated": datetime.now().strftime("%Y-%m-%d"),
+                },
+                default_flow_style=False,
+                sort_keys=False,
+            ),
+            encoding="utf-8",
+        )
+
+    with open(REGISTRY_PATH, "r+", encoding="utf-8") as registry_file:
+        _lock_registry_file(registry_file)
+        try:
+            registry_file.seek(0)
+            raw = registry_file.read()
+            data = yaml.safe_load(raw) or {}
+            year = str(data.get("year", calendar_year))
+            if year != calendar_year:
+                data["year"] = int(calendar_year)
+                data["categories"] = {cat: 1 for cat in (data.get("categories") or {})}
+                year = calendar_year
+            categories = data.setdefault("categories", {})
+            # Registry may seed unused categories at 0; NNN must be 001-999 per case-format.md
+            nnn = max(int(categories.get(category, 1)), 1)
+            case_no = f"{year}-{category}-{nnn:03d}-{deliberation:03d}"
+            categories[category] = nnn + 1
+            data["categories"] = categories
+            data["year"] = int(data.get("year", calendar_year))
+            data["last_updated"] = datetime.now().strftime("%Y-%m-%d")
+            serialized = yaml.dump(data, default_flow_style=False, sort_keys=False)
+            registry_file.seek(0)
+            registry_file.write(serialized)
+            registry_file.truncate()
+            registry_file.flush()
+            os.fsync(registry_file.fileno())
+        finally:
+            _unlock_registry_file(registry_file)
+
     return case_no
 
 

litigation/viewer.py

@@ -85,6 +85,29 @@ def _collect_transcripts() -> List[Tuple[Path, str, str]]:
     return out
 
 
+def _matches_transcript_query(name: str, path: Path) -> bool:
+    """Match transcript by exact name, prefix, or topic suffix (not arbitrary substring)."""
+    stem = path.stem
+    if stem == name or path.name == name:
+        return True
+    if stem.startswith(name):
+        return True
+    return stem.endswith(f"-{name}") or stem.endswith(f"_{name}")
+
+
+def _resolve_transcript(name: str, rows: List[Tuple[Path, str, str]]) -> Optional[Path]:
+    """Pick the best transcript path for a query name."""
+    matches = [r for r in rows if _matches_transcript_query(name, r[0])]
+    if not matches:
+        return None
+    exact = [m for m in matches if m[0].stem == name or m[0].name == name]
+    if exact:
+        return exact[0][0]
+    if len(matches) > 1:
+        matches.sort(key=lambda r: len(r[0].stem))
+    return matches[0][0]
+
+
 def cmd_list(plain: bool) -> None:
     """List transcripts in a table or plain lines."""
     rows = _collect_transcripts()
@@ -116,17 +139,11 @@ def cmd_show(name: Optional[str], pager: bool) -> None:
         path = rows[0][0]
     else:
         name = name.strip()
-        matches = [r for r in rows if r[0].stem == name or r[0].name == name or r[0].stem.startswith(name) or name in r[0].stem]
-        if not matches:
+        path = _resolve_transcript(name, rows)
+        if path is None:
             print(f"No transcript matching '{name}'", file=sys.stderr)
             print("Use 'viewer.py list' to see available transcripts.", file=sys.stderr)
             sys.exit(1)
-        if len(matches) > 1:
-            # Prefer exact stem match
-            exact = [m for m in matches if m[0].stem == name]
-            path = exact[0][0] if exact else matches[0][0]
-        else:
-            path = matches[0][0]
     text = path.read_text(encoding="utf-8")
     if pager:
         import subprocess
@@ -223,15 +240,12 @@ def do_GET(self) -> None:
                 if not name:
                     self.send_error(404)
                     return
-                found = None
-                for p, date_str, topic in rows:
-                    if p.stem == name or name in p.stem:
-                        found = (p, topic)
-                        break
-                if not found:
+                resolved = _resolve_transcript(name, rows)
+                if not resolved:
                     self.send_error(404)
                     return
-                p, topic = found
+                p = resolved
+                topic = next(t for path, _, t in rows if path == p)
                 md = p.read_text(encoding="utf-8")
                 body_html = _markdown_to_html(md)
                 title = f"In Re: {topic}"

tests/test_executive_watchdog.py

@@ -66,6 +66,60 @@ def test_run_checks_alerts_on_unapproved_action_without_proof(
     assert any("without judicial approval" in a for a in alerts)
 
 
+def test_run_checks_alerts_on_override_without_proof(
+    tmp_path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    log_path = tmp_path / "judicial_decisions.log"
+    actions_path = tmp_path / "executive_actions.log"
+
+    JudicialLog(log_path=log_path)  # empty approvals
+
+    _write_action(
+        actions_path,
+        action_type="override",
+        ruling_id="2026-UNKNOWN-999",
+        description="rogue override",
+    )
+
+    monkeypatch.setattr(watchdog, "JudicialLog", lambda: JudicialLog(log_path=log_path))
+    monkeypatch.setattr(watchdog, "_actions_log_path", lambda: actions_path)
+
+    ok, alerts = watchdog.run_checks()
+    assert ok is False
+    assert any("Override action without proof" in a for a in alerts)
+
+
+def test_run_checks_alerts_on_invalid_override_proof(
+    tmp_path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    log_path = tmp_path / "judicial_decisions.log"
+    actions_path = tmp_path / "executive_actions.log"
+    secret_path = tmp_path / ".executive_secret"
+    secret_path.write_bytes(bytes.fromhex("99" * 32))
+
+    proof = OverrideProof.generate(
+        "2026-DEL-001",
+        "emergency",
+        timestamp="2026-05-25T12:00:00+00:00",
+        secret_path=secret_path,
+    )
+    _write_action(
+        actions_path,
+        action_type="override",
+        ruling_id="2026-DEL-002",
+        description="emergency",
+        override_proof=proof,
+    )
+
+    monkeypatch.setattr(watchdog, "JudicialLog", lambda: JudicialLog(log_path=log_path))
+    monkeypatch.setattr(watchdog, "_actions_log_path", lambda: actions_path)
+    monkeypatch.setenv("EXECUTIVE_PROOF_SECRET", "99" * 64)
+
+    ok, alerts = watchdog.run_checks()
+    assert ok is False
+    assert any("Invalid override proof" in a for a in alerts)
+
+
 def test_run_checks_alerts_on_tampered_judicial_log(
     tmp_path, monkeypatch: pytest.MonkeyPatch
 ) -> None:

tests/test_export_transcript.py

@@ -36,3 +36,22 @@ def test_extract_title_from_dated_filename():
     mod = _load_export_module()
     path = Path("2026-02-15-framework-enhancement-analysis.md")
     assert mod.extract_title(path, "") == "Framework Enhancement Analysis"
+
+
+def test_resolve_transcript_path_accepts_repo_root_prefix():
+    """README documents courtroom/transcripts/foo.md from repo root."""
+    mod = _load_export_module()
+    sample = "2026-02-15-framework-enhancement-analysis.md"
+    resolved = mod.resolve_transcript_path(f"courtroom/transcripts/{sample}")
+    expected = REPO_ROOT / "courtroom" / "transcripts" / sample
+    assert resolved == expected
+    assert resolved.exists()
+
+
+def test_resolve_transcript_path_accepts_courtroom_relative_prefix():
+    mod = _load_export_module()
+    sample = "2026-02-15-framework-enhancement-analysis.md"
+    resolved = mod.resolve_transcript_path(f"transcripts/{sample}")
+    expected = REPO_ROOT / "courtroom" / "transcripts" / sample
+    assert resolved == expected
+    assert resolved.exists()