"""The judgement pass. Reads what the rail found and thinks about it. THE ONE RULE: this can only ADD. It opens findings and the register read-only and writes exactly one file, `data/judgement/.json`. There is no code path here that edits a finding, removes a register row, or changes a bucket. The brief renders what this produces underneath what the rules found, never instead of it. That is deliberate and it is the whole reason the rail has no model calls in it. A component that can score a message can also silently drop one, and the value of the register is that a date captured in January still surfaces in March whether or not anything was feeling clever that day. So judgement lives out here, where the worst it can do is be wrong in an extra paragraph. If it does not run, or returns nothing, or returns garbage, the brief is unchanged. Failure of this file is invisible to the thing that matters. """ from __future__ import annotations import json from datetime import date, datetime, timedelta, timezone from pathlib import Path from typing import Any from notify import ROOT, env_path JUDGEMENT_DIR = ROOT / "data" / "judgement" FINDINGS_DIR = ROOT / "data" / "findings" REGISTER_PATH = ROOT / "data" / "deadlines.jsonl" CONTEXT_PATH = ROOT / "CONTEXT.md" PERSONA_PATH = ROOT / "COO.md" # How stale a judgement file may be before the brief stops showing it. A thought from # last week rendered as if it were today's is worse than no thought at all. MAX_AGE_HOURS = 36 def judgement_for(day: date | None = None) -> dict[str, Any] | None: """Today's judgement, or None. Never raises — the brief must not care.""" try: day = day or datetime.now().astimezone().date() path = JUDGEMENT_DIR / f"{day.isoformat()}.json" if not path.exists(): return None value = json.loads(path.read_text(encoding="utf-8")) written = datetime.fromisoformat(value["written_at_utc"]) if datetime.now(timezone.utc) - written > timedelta(hours=MAX_AGE_HOURS): return None lines = [str(line).strip() for line in value.get("lines", []) if str(line).strip()] return {"lines": lines[:3]} if lines else None except Exception: # A malformed judgement file must never take the brief down with it. return None def _recent_findings(days: int = 1) -> list[dict[str, Any]]: cutoff = (datetime.now().astimezone().date() - timedelta(days=days)).isoformat() rows: list[dict[str, Any]] = [] if not FINDINGS_DIR.exists(): return rows for path in sorted(FINDINGS_DIR.glob("*.jsonl")): if path.stem < cutoff: continue for line in path.read_text(encoding="utf-8").splitlines(): if line.strip(): try: rows.append(json.loads(line)) except ValueError: continue return rows def _horizon(days: int = 90) -> list[dict[str, Any]]: if not REGISTER_PATH.exists(): return [] today = datetime.now().astimezone().date() seen: dict[tuple[str, str], dict[str, Any]] = {} for line in REGISTER_PATH.read_text(encoding="utf-8").splitlines(): if not line.strip(): continue record = json.loads(line) try: due = date.fromisoformat(record["date"]) except (KeyError, ValueError): continue if 0 <= (due - today).days <= days: seen[(record["date"], record["source_subject"])] = record return sorted(seen.values(), key=lambda item: item["date"]) def build_prompt() -> str: """Everything the judgement pass gets to see. Read-only, assembled here.""" findings = _recent_findings() act = [f for f in findings if f.get("bucket") == "ACT"] horizon = _horizon() persona = PERSONA_PATH.read_text(encoding="utf-8") if PERSONA_PATH.exists() else "" context = CONTEXT_PATH.read_text(encoding="utf-8") if CONTEXT_PATH.exists() else "" act_text = "\n".join( f"- [{f.get('account')}] {f.get('subject', '')[:90]}" f" ({', '.join(f.get('reasons', [])[:2])})" for f in act[:20] ) or "- nothing" horizon_text = "\n".join( f"- {h['date']} {h['source_subject'][:80]}" for h in horizon ) or "- nothing dated in the next 90 days" return ( f"{persona}\n\n" "---\n\n" f"{context}\n\n" "---\n\n" "## What the rules found today\n\n" f"{act_text}\n\n" "## Dated obligations, next 90 days\n\n" f"{horizon_text}\n\n" "---\n\n" "Give at most three lines. Each must tell him something he does not already\n" "know from the list above — a connection, a consequence, or something in\n" "CONTEXT.md that is going quiet and has no email to remind him. If the honest\n" "answer is that there is nothing worth saying, return an empty list. A quiet\n" "day reported quietly is a correct answer and padding is not.\n\n" 'Reply as JSON only: {"lines": ["...", "..."]}\n' ) def _write(lines: list[str], day: date | None = None) -> Path: day = day or datetime.now().astimezone().date() JUDGEMENT_DIR.mkdir(parents=True, exist_ok=True) path = JUDGEMENT_DIR / f"{day.isoformat()}.json" payload = { "written_at_utc": datetime.now(timezone.utc).isoformat(), "lines": lines[:3], } temporary = path.with_suffix(".json.tmp") temporary.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8") temporary.replace(path) return path def _configured_api_key() -> str | None: """The key from the CONFIGURED secrets file only. Never the ambient environment. read_env() merges os.environ, which is right for the rest of the rail. It is wrong here: on 2026-07-29 this picked up the ANTHROPIC_API_KEY belonging to the Claude Code session that was building it, and tried to bill a model against a credential Rocky had never provided. It 401'd and the failure was harmless, but the next one might have worked, which is worse. A credential he has not put in the secrets file is a credential this does not have. """ path = env_path() if not path.exists(): return None for raw_line in path.read_text(encoding="utf-8").splitlines(): line = raw_line.strip() if not line or line.startswith("#") or "=" not in line: continue key, value = line.split("=", 1) if key.strip() == "ANTHROPIC_API_KEY" and value.strip(): return value.strip() return None def main() -> int: api_key = _configured_api_key() if not api_key: # Not an error. The rail is complete without this; judgement is an extra. print("no ANTHROPIC_API_KEY configured — judgement pass is dark, rail unaffected") return 0 import requests prompt = build_prompt() response = requests.post( "https://api.anthropic.com/v1/messages", headers={ "x-api-key": api_key, "anthropic-version": "2023-06-01", "content-type": "application/json", }, json={ "model": "claude-sonnet-5", "max_tokens": 400, "messages": [{"role": "user", "content": prompt}], }, timeout=60, ) response.raise_for_status() text = "".join( block.get("text", "") for block in response.json().get("content", []) ).strip() if text.startswith("```"): text = text.split("\n", 1)[1].rsplit("```", 1)[0] lines = json.loads(text).get("lines", []) path = _write([str(line) for line in lines]) print(f"judgement written: {path.name} ({len(lines)} lines)") return 0 if __name__ == "__main__": from error_reporting import record_failure try: raise SystemExit(main()) except SystemExit: raise except Exception as error: # Judgement failing must never be able to stop a brief. Log it, say so, exit 0. summary = record_failure("Judgement pass failed", error) print(f"Judgement pass failed (rail unaffected): {summary}") raise SystemExit(0)