Coverage for src/keel/activity.py: 100%
112 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-18 12:05 +0000
1"""Lightweight, additive **command-activity** records for live observability.
3The resumable ``checkpoint`` is the ship backbone's own artifact (s0–s12) and is
4deliberately ship-shaped. Most other keel commands (``triage``, ``morning``,
5``pr-loop`` …) run in the main checkout, never write a checkpoint, and so are
6invisible to ``keel-visual``'s live board.
8This module adds a *separate, additive* channel: a per-run JSON record under
9``.keel/activity/`` that any command's adapter can stamp as it moves through its
10own flow phases (from :mod:`keel.flows`). It never touches the checkpoint
11contract. Records are keyed by ``run_id`` (one file each), so two commands in the
12same repo never clobber one another.
14Pure-core + thin I/O, mirroring :mod:`keel.checkpoint`: the builders/validators
15are deterministic (stable ordering, no wall-clock, no randomness); only
16read/write/remove touch the filesystem.
17"""
19from __future__ import annotations
21import json
22import re
23from pathlib import Path
24from typing import Any
26from . import config as cfg
27from . import flows, workspace
29ACTIVITY_SCHEMA_VERSION = "keel.activity.v1"
30RECORD_TYPE_ACTIVITY = "command_activity"
31DEFAULT_ACTIVITY_DIR = ".keel/activity"
32STATUSES = ("running", "done", "merged")
34#: Whether the phase the run reached was actually *passed*. Separate from
35#: :data:`STATUSES` because "advanced to s8" and "cleared s8" are different facts and
36#: were previously written identically (#636) — a run whose gates came back red
37#: recorded as ``phase: s8, status: running``, carrying no failure signal at all, and
38#: the board painted it as in-progress. ``None`` means the step reached this stamp
39#: without a verdict to report (planning, a phase with nothing to pass), which is not
40#: the same as passing.
41VERDICTS = ("pass", "blocked")
43# A run_id reduces to this slug for its filename; anything else is rejected so a
44# crafted run_id can never escape the activity directory.
45_RUN_ID_SLUG = re.compile(r"[^a-z0-9._-]+")
48class ActivityError(ValueError):
49 """Raised when an activity record or path is malformed."""
52def activity_contract_as_dict() -> dict[str, Any]:
53 """Return the stable activity-record contract consumed by adapters."""
54 return {
55 "schema_version": ACTIVITY_SCHEMA_VERSION,
56 "record_type": RECORD_TYPE_ACTIVITY,
57 "dir": DEFAULT_ACTIVITY_DIR,
58 "keyed_by": "run_id",
59 "statuses": list(STATUSES),
60 "additive": True,
61 "touches_checkpoint": False,
62 "phase_source": "keel.flows.flow_for(command)",
63 }
66def configured_activity_dir(config: cfg.ProjectConfig) -> tuple[str, str]:
67 """Return the configured activity directory and its source."""
68 pack = config.policy_pack or {}
69 reports = pack.get("reports") if isinstance(pack.get("reports"), dict) else {}
70 value = reports.get("activity")
71 if isinstance(value, str) and value.strip():
72 return value, "policy_pack.reports.activity"
73 return DEFAULT_ACTIVITY_DIR, "default"
76def resolve_dir(root: str | Path, config: cfg.ProjectConfig) -> Path:
77 """Resolve the activity directory under ``root`` and reject escapes."""
78 raw, _ = configured_activity_dir(config)
79 path = Path(raw)
80 if path.is_absolute():
81 raise ActivityError("activity dir must be relative to the project root")
82 root_path = Path(root).resolve()
83 resolved = (root_path / path).resolve()
84 try:
85 resolved.relative_to(root_path)
86 except ValueError as exc:
87 raise ActivityError("activity dir escapes the project root") from exc
88 return resolved
91def run_id_slug(run_id: str) -> str:
92 """Reduce a run_id to a safe filename stem (lowercase ``[a-z0-9._-]``)."""
93 if not isinstance(run_id, str) or not run_id.strip():
94 raise ActivityError("run_id must be a non-empty string")
95 slug = _RUN_ID_SLUG.sub("-", run_id.strip().lower()).strip("-.")
96 if not slug:
97 raise ActivityError("run_id has no usable characters")
98 return slug
101def record_path(root: str | Path, config: cfg.ProjectConfig, run_id: str) -> Path:
102 """Path of the activity record for ``run_id`` under ``root``."""
103 return resolve_dir(root, config) / f"{run_id_slug(run_id)}.json"
106def _phase_ids(command: str) -> tuple[str, ...]:
107 return tuple(phase.id for phase in flows.flow_for(command))
110def build_activity_record(
111 *,
112 command: str,
113 run_id: str,
114 phase: str,
115 status: str = "running",
116 verdict: str | None = None,
117 issue: int | None = None,
118 pr: int | None = None,
119 note: str | None = None,
120) -> dict[str, Any]:
121 """Build one deterministic activity record, validating command + phase.
123 ``command`` must be a known :mod:`keel.flows` command and ``phase`` one of
124 that command's flow phase ids. ``status`` is ``running``, ``done`` or
125 ``merged`` (a real merge landed, distinct from a soft ``done``).
127 ``verdict`` (:data:`VERDICTS`) says whether the phase was **passed**, which
128 ``status`` deliberately does not: a blocked gate run is still ``running`` in the
129 board's sense — it advanced, it did not finish — and recording only that made a
130 red gate indistinguishable from an in-progress one (#636). ``None`` = no verdict
131 to report, which must not read as a pass.
132 """
133 if not flows.is_known(command):
134 raise ActivityError(f"unknown command: {command!r}")
135 if phase not in _phase_ids(command):
136 raise ActivityError(f"phase {phase!r} is not a {command} flow phase")
137 if status not in STATUSES:
138 raise ActivityError(f"unsupported status: {status!r}")
139 if verdict is not None and verdict not in VERDICTS:
140 raise ActivityError(f"unsupported verdict: {verdict!r}")
141 return {
142 "schema_version": ACTIVITY_SCHEMA_VERSION,
143 "record_type": RECORD_TYPE_ACTIVITY,
144 "command": command,
145 "run_id": run_id,
146 "phase": phase,
147 "status": status,
148 "verdict": verdict,
149 "issue": issue,
150 "pr": pr,
151 "note": note,
152 }
155def validate_activity(record: Any) -> None:
156 """Validate the stable activity-record shape."""
157 if not isinstance(record, dict):
158 raise ActivityError("activity must be an object")
159 if record.get("schema_version") != ACTIVITY_SCHEMA_VERSION:
160 raise ActivityError("unsupported schema_version")
161 if record.get("record_type") != RECORD_TYPE_ACTIVITY:
162 raise ActivityError("unsupported record_type")
163 command = record.get("command")
164 if not isinstance(command, str) or not flows.is_known(command):
165 raise ActivityError("unsupported command")
166 if not isinstance(record.get("run_id"), str) or not record["run_id"].strip():
167 raise ActivityError("run_id must be a non-empty string")
168 if record.get("phase") not in _phase_ids(command):
169 raise ActivityError("unsupported phase")
170 if record.get("status") not in STATUSES:
171 raise ActivityError("unsupported status")
172 # Absent is fine (older records, phases with nothing to pass); a *wrong* value is
173 # not — a board that trusts this field must never read a typo as a pass.
174 if record.get("verdict") is not None and record.get("verdict") not in VERDICTS:
175 raise ActivityError("unsupported verdict")
178def encode_activity(record: dict[str, Any]) -> str:
179 """Encode one activity record as stable JSON."""
180 validate_activity(record)
181 return json.dumps(record, indent=2, sort_keys=True) + "\n"
184def parse_activity(text: str) -> dict[str, Any]:
185 """Parse and validate one activity record."""
186 try:
187 record = json.loads(text)
188 except json.JSONDecodeError as exc:
189 raise ActivityError("invalid JSON") from exc
190 validate_activity(record)
191 return record
194def read_activity(path: str | Path) -> dict[str, Any] | None:
195 """Read one activity record; a missing file means no such run."""
196 activity_path = Path(path)
197 if not activity_path.exists():
198 return None
199 return parse_activity(activity_path.read_text(encoding="utf-8"))
202def write_activity(path: str | Path, record: dict[str, Any]) -> None:
203 """Write one validated activity record."""
204 activity_path = Path(path)
205 activity_path.parent.mkdir(parents=True, exist_ok=True)
206 workspace.ensure_runtime_gitignore_for(activity_path)
207 activity_path.write_text(encode_activity(record), encoding="utf-8")
210def remove_activity(path: str | Path) -> bool:
211 """Delete an activity record. Returns ``True`` if a file was removed."""
212 activity_path = Path(path)
213 if not activity_path.exists():
214 return False
215 activity_path.unlink()
216 return True
219def read_all_activity(dir_path: str | Path) -> list[dict[str, Any]]:
220 """Every readable activity record in ``dir_path``, sorted by run_id.
222 Fail-soft: an unreadable or malformed file is skipped, never raised — one bad
223 record must not blank the board. The directory missing yields ``[]``.
224 """
225 directory = Path(dir_path)
226 if not directory.is_dir():
227 return []
228 records: list[dict[str, Any]] = []
229 for entry in sorted(directory.glob("*.json")):
230 try:
231 record = parse_activity(entry.read_text(encoding="utf-8"))
232 except (ActivityError, OSError):
233 continue
234 records.append(record)
235 return records