Coverage for little_loops / decisions.py: 0%
163 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-06-04 12:21 -0500
1"""Decisions and rules log data layer.
3Provides typed dataclasses and CRUD operations for managing architectural
4decisions, team-enforced rules, and exceptions stored in `.ll/decisions.yaml`.
5"""
7from __future__ import annotations
9from dataclasses import dataclass, field
10from pathlib import Path
11from typing import TYPE_CHECKING, Any
13if TYPE_CHECKING:
14 from little_loops.config.core import BRConfig
16import yaml
18from little_loops.file_utils import atomic_write
20_DEFAULT_LOG_PATH = Path(".ll") / "decisions.yaml"
23def _resolve_path(path: Path | None) -> Path:
24 return path if path is not None else Path.cwd() / _DEFAULT_LOG_PATH
27@dataclass
28class DecisionOutcome:
29 """Recorded outcome for a decision entry."""
31 result: str
32 measured_at: str
33 notes: str | None = None
35 @classmethod
36 def from_dict(cls, data: dict[str, Any]) -> DecisionOutcome:
37 return cls(
38 result=data["result"],
39 measured_at=data["measured_at"],
40 notes=data.get("notes"),
41 )
43 def to_dict(self) -> dict[str, Any]:
44 d: dict[str, Any] = {"result": self.result, "measured_at": self.measured_at}
45 if self.notes is not None:
46 d["notes"] = self.notes
47 return d
50@dataclass
51class RuleEntry:
52 """An enforced rule in the decisions log."""
54 id: str
55 type: str = "rule"
56 timestamp: str = ""
57 category: str = ""
58 labels: list[str] = field(default_factory=list)
59 rationale: str = ""
60 rule: str = ""
61 enforcement: str = "advisory"
62 supersedes: str | None = None
63 issue: str | None = None
65 @classmethod
66 def from_dict(cls, data: dict[str, Any]) -> RuleEntry:
67 return cls(
68 id=data["id"],
69 type=data.get("type", "rule"),
70 timestamp=data.get("timestamp", ""),
71 category=data.get("category", ""),
72 labels=data.get("labels", []),
73 rationale=data.get("rationale", ""),
74 rule=data.get("rule", ""),
75 enforcement=data.get("enforcement", "advisory"),
76 supersedes=data.get("supersedes"),
77 issue=data.get("issue"),
78 )
80 def to_dict(self) -> dict[str, Any]:
81 d: dict[str, Any] = {
82 "id": self.id,
83 "type": self.type,
84 "timestamp": self.timestamp,
85 "category": self.category,
86 "labels": self.labels,
87 "rationale": self.rationale,
88 "rule": self.rule,
89 "enforcement": self.enforcement,
90 }
91 if self.supersedes is not None:
92 d["supersedes"] = self.supersedes
93 if self.issue is not None:
94 d["issue"] = self.issue
95 return d
98@dataclass
99class DecisionEntry:
100 """A recorded architectural or process decision."""
102 id: str
103 type: str = "decision"
104 timestamp: str = ""
105 category: str = ""
106 labels: list[str] = field(default_factory=list)
107 rationale: str = ""
108 rule: str = ""
109 alternatives_rejected: str | None = None
110 issue: str | None = None
111 scope: str = "issue"
112 outcome: DecisionOutcome | None = None
114 @classmethod
115 def from_dict(cls, data: dict[str, Any]) -> DecisionEntry:
116 outcome_data = data.get("outcome")
117 return cls(
118 id=data["id"],
119 type=data.get("type", "decision"),
120 timestamp=data.get("timestamp", ""),
121 category=data.get("category", ""),
122 labels=data.get("labels", []),
123 rationale=data.get("rationale", ""),
124 rule=data.get("rule", ""),
125 alternatives_rejected=data.get("alternatives_rejected"),
126 issue=data.get("issue"),
127 scope=data.get("scope", "issue"),
128 outcome=DecisionOutcome.from_dict(outcome_data) if outcome_data else None,
129 )
131 def to_dict(self) -> dict[str, Any]:
132 d: dict[str, Any] = {
133 "id": self.id,
134 "type": self.type,
135 "timestamp": self.timestamp,
136 "category": self.category,
137 "labels": self.labels,
138 "rationale": self.rationale,
139 "rule": self.rule,
140 "scope": self.scope,
141 }
142 if self.alternatives_rejected is not None:
143 d["alternatives_rejected"] = self.alternatives_rejected
144 if self.issue is not None:
145 d["issue"] = self.issue
146 if self.outcome is not None:
147 d["outcome"] = self.outcome.to_dict()
148 return d
151@dataclass
152class ExceptionEntry:
153 """A one-time exception to an existing rule."""
155 id: str
156 type: str = "exception"
157 timestamp: str = ""
158 category: str = ""
159 labels: list[str] = field(default_factory=list)
160 rationale: str = ""
161 rule_ref: str = ""
162 issue: str = ""
163 alternatives_rejected: str | None = None
165 @classmethod
166 def from_dict(cls, data: dict[str, Any]) -> ExceptionEntry:
167 return cls(
168 id=data["id"],
169 type=data.get("type", "exception"),
170 timestamp=data.get("timestamp", ""),
171 category=data.get("category", ""),
172 labels=data.get("labels", []),
173 rationale=data.get("rationale", ""),
174 rule_ref=data.get("rule_ref", ""),
175 issue=data.get("issue", ""),
176 alternatives_rejected=data.get("alternatives_rejected"),
177 )
179 def to_dict(self) -> dict[str, Any]:
180 d: dict[str, Any] = {
181 "id": self.id,
182 "type": self.type,
183 "timestamp": self.timestamp,
184 "category": self.category,
185 "labels": self.labels,
186 "rationale": self.rationale,
187 "rule_ref": self.rule_ref,
188 "issue": self.issue,
189 }
190 if self.alternatives_rejected is not None:
191 d["alternatives_rejected"] = self.alternatives_rejected
192 return d
195# Open dispatch registry — add new entry types here without modifying CRUD functions
196AnyEntry = RuleEntry | DecisionEntry | ExceptionEntry
198_ENTRY_REGISTRY: dict[str, Any] = {
199 "rule": RuleEntry,
200 "decision": DecisionEntry,
201 "exception": ExceptionEntry,
202}
205def _entry_from_dict(data: dict[str, Any]) -> AnyEntry:
206 entry_type = data.get("type", "rule")
207 cls = _ENTRY_REGISTRY.get(entry_type)
208 if cls is None:
209 raise ValueError(f"Unknown entry type: {entry_type!r}")
210 return cls.from_dict(data)
213def load_decisions(path: Path | None = None) -> list[AnyEntry]:
214 """Load all decision log entries from YAML; returns empty list if file absent."""
215 resolved = _resolve_path(path)
216 if not resolved.exists():
217 return []
218 raw = resolved.read_text(encoding="utf-8")
219 data = yaml.safe_load(raw)
220 if not data:
221 return []
222 entries = data if isinstance(data, list) else data.get("entries", [])
223 return [_entry_from_dict(e) for e in entries]
226def save_decisions(entries: list[AnyEntry], path: Path | None = None) -> None:
227 """Atomically persist decision log entries to YAML."""
228 resolved = _resolve_path(path)
229 resolved.parent.mkdir(parents=True, exist_ok=True)
230 content = yaml.dump(
231 [e.to_dict() for e in entries],
232 default_flow_style=False,
233 sort_keys=False,
234 allow_unicode=True,
235 )
236 atomic_write(resolved, content)
239def add_entry(entry: AnyEntry, path: Path | None = None) -> None:
240 """Append a new entry to the decisions log."""
241 entries = load_decisions(path)
242 entries.append(entry)
243 save_decisions(entries, path)
246def list_entries(
247 path: Path | None = None,
248 *,
249 type: str | None = None,
250 category: str | None = None,
251 label: str | None = None,
252) -> list[AnyEntry]:
253 """Return entries, optionally filtered by type, category, or label."""
254 entries = load_decisions(path)
255 if type is not None:
256 entries = [e for e in entries if e.type == type]
257 if category is not None:
258 entries = [e for e in entries if e.category == category]
259 if label is not None:
260 entries = [e for e in entries if label in e.labels]
261 return entries
264def resolve_active(entries: list[AnyEntry]) -> list[AnyEntry]:
265 """Return entries excluding those superseded by a newer entry.
267 An entry is inactive if another entry's `supersedes` field references its ID.
268 """
269 superseded_ids = {
270 getattr(e, "supersedes", None) for e in entries if getattr(e, "supersedes", None)
271 }
272 return [e for e in entries if e.id not in superseded_ids]
275def set_outcome(
276 entry_id: str,
277 result: str,
278 measured_at: str,
279 notes: str | None = None,
280 path: Path | None = None,
281 *,
282 force: bool = False,
283) -> None:
284 """Set the outcome on a decision entry; refuses to overwrite without force=True."""
285 entries = load_decisions(path)
286 for entry in entries:
287 if entry.id == entry_id:
288 if not isinstance(entry, DecisionEntry):
289 raise TypeError(f"Entry {entry_id!r} is not a DecisionEntry (got {entry.type!r})")
290 if entry.outcome is not None and not force:
291 raise ValueError(
292 f"Entry {entry_id!r} already has an outcome. Use force=True to overwrite."
293 )
294 entry.outcome = DecisionOutcome(result=result, measured_at=measured_at, notes=notes)
295 save_decisions(entries, path)
296 return
297 raise KeyError(f"No entry with id {entry_id!r}")
300def generate_from_completed(config: BRConfig) -> int:
301 """Generate DecisionEntry records from completed issues and persist to the log.
303 Prefers the SQLite history DB when present; falls back to filesystem scanning.
304 Skips issues that already have an entry in the log. Returns the count added.
305 """
306 from little_loops.issue_history.parsing import (
307 scan_completed_issues,
308 scan_completed_issues_from_db,
309 )
311 project_root = Path(config.project_root)
312 log_path = project_root / config.decisions.log_path
313 db_path = project_root / ".ll" / "history.db"
315 if db_path.exists():
316 completed = scan_completed_issues_from_db(db_path)
317 else:
318 completed = scan_completed_issues(project_root / config.issues.base_dir)
320 existing = load_decisions(log_path)
321 existing_issue_ids = {e.issue for e in existing if isinstance(e, DecisionEntry) and e.issue}
323 count = 0
324 for issue in completed:
325 if issue.issue_id in existing_issue_ids:
326 continue
327 ts = ""
328 if issue.completed_at:
329 ts = issue.completed_at.strftime("%Y-%m-%dT%H:%M:%SZ")
330 elif issue.captured_at:
331 ts = issue.captured_at.strftime("%Y-%m-%dT%H:%M:%SZ")
332 entry = DecisionEntry(
333 id=f"DEC-{issue.issue_id}",
334 timestamp=ts,
335 category=issue.issue_type.lower(),
336 labels=[issue.priority, issue.issue_type.lower()],
337 rationale=f"Auto-generated from completed issue {issue.issue_id}",
338 issue=issue.issue_id,
339 scope="issue",
340 )
341 add_entry(entry, log_path)
342 count += 1
344 return count