Coverage for little_loops / decisions.py: 0%
198 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-06-16 13:08 -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@dataclass
196class CouplingEntry:
197 """A coupling rule linking changed files to required audit targets in wire-issue."""
199 id: str
200 type: str = "coupling"
201 timestamp: str = ""
202 category: str = ""
203 labels: list[str] = field(default_factory=list)
204 rationale: str = ""
205 if_changed: str = ""
206 then_check: list[str] = field(default_factory=list)
207 tier: str = "soft" # hard | soft | fyi
208 archetype: str | None = None
209 enforcement: str = "advisory"
210 supersedes: str | None = None
211 issue: str | None = None
213 @classmethod
214 def from_dict(cls, data: dict[str, Any]) -> CouplingEntry:
215 return cls(
216 id=data["id"],
217 type=data.get("type", "coupling"),
218 timestamp=data.get("timestamp", ""),
219 category=data.get("category", ""),
220 labels=data.get("labels", []),
221 rationale=data.get("rationale", ""),
222 if_changed=data.get("if_changed", ""),
223 then_check=data.get("then_check", []),
224 tier=data.get("tier", "soft"),
225 archetype=data.get("archetype"),
226 enforcement=data.get("enforcement", "advisory"),
227 supersedes=data.get("supersedes"),
228 issue=data.get("issue"),
229 )
231 def to_dict(self) -> dict[str, Any]:
232 d: dict[str, Any] = {
233 "id": self.id,
234 "type": self.type,
235 "timestamp": self.timestamp,
236 "category": self.category,
237 "labels": self.labels,
238 "rationale": self.rationale,
239 "if_changed": self.if_changed,
240 "then_check": self.then_check,
241 "tier": self.tier,
242 "enforcement": self.enforcement,
243 }
244 if self.archetype is not None:
245 d["archetype"] = self.archetype
246 if self.supersedes is not None:
247 d["supersedes"] = self.supersedes
248 if self.issue is not None:
249 d["issue"] = self.issue
250 return d
253# Open dispatch registry — add new entry types here without modifying CRUD functions
254AnyEntry = RuleEntry | DecisionEntry | ExceptionEntry | CouplingEntry
256_ENTRY_REGISTRY: dict[str, Any] = {
257 "rule": RuleEntry,
258 "decision": DecisionEntry,
259 "exception": ExceptionEntry,
260 "coupling": CouplingEntry,
261}
264def _entry_from_dict(data: dict[str, Any]) -> AnyEntry:
265 entry_type = data.get("type", "rule")
266 cls = _ENTRY_REGISTRY.get(entry_type)
267 if cls is None:
268 raise ValueError(f"Unknown entry type: {entry_type!r}")
269 return cls.from_dict(data)
272def load_decisions(path: Path | None = None) -> list[AnyEntry]:
273 """Load all decision log entries from YAML; returns empty list if file absent."""
274 resolved = _resolve_path(path)
275 if not resolved.exists():
276 return []
277 raw = resolved.read_text(encoding="utf-8")
278 data = yaml.safe_load(raw)
279 if not data:
280 return []
281 entries = data if isinstance(data, list) else data.get("entries", [])
282 return [_entry_from_dict(e) for e in entries]
285def save_decisions(entries: list[AnyEntry], path: Path | None = None) -> None:
286 """Atomically persist decision log entries to YAML."""
287 resolved = _resolve_path(path)
288 resolved.parent.mkdir(parents=True, exist_ok=True)
289 content = yaml.dump(
290 [e.to_dict() for e in entries],
291 default_flow_style=False,
292 sort_keys=False,
293 allow_unicode=True,
294 )
295 atomic_write(resolved, content)
298def add_entry(entry: AnyEntry, path: Path | None = None) -> None:
299 """Append a new entry to the decisions log."""
300 entries = load_decisions(path)
301 entries.append(entry)
302 save_decisions(entries, path)
305def list_entries(
306 path: Path | None = None,
307 *,
308 type: str | None = None,
309 category: str | None = None,
310 label: str | None = None,
311) -> list[AnyEntry]:
312 """Return entries, optionally filtered by type, category, or label."""
313 entries = load_decisions(path)
314 if type is not None:
315 entries = [e for e in entries if e.type == type]
316 if category is not None:
317 entries = [e for e in entries if e.category == category]
318 if label is not None:
319 entries = [e for e in entries if label in e.labels]
320 return entries
323def resolve_active(entries: list[AnyEntry]) -> list[AnyEntry]:
324 """Return entries excluding those superseded by a newer entry.
326 An entry is inactive if another entry's `supersedes` field references its ID.
327 """
328 superseded_ids = {
329 getattr(e, "supersedes", None) for e in entries if getattr(e, "supersedes", None)
330 }
331 return [e for e in entries if e.id not in superseded_ids]
334def set_outcome(
335 entry_id: str,
336 result: str,
337 measured_at: str,
338 notes: str | None = None,
339 path: Path | None = None,
340 *,
341 force: bool = False,
342) -> None:
343 """Set the outcome on a decision entry; refuses to overwrite without force=True."""
344 entries = load_decisions(path)
345 for entry in entries:
346 if entry.id == entry_id:
347 if not isinstance(entry, DecisionEntry):
348 raise TypeError(f"Entry {entry_id!r} is not a DecisionEntry (got {entry.type!r})")
349 if entry.outcome is not None and not force:
350 raise ValueError(
351 f"Entry {entry_id!r} already has an outcome. Use force=True to overwrite."
352 )
353 entry.outcome = DecisionOutcome(result=result, measured_at=measured_at, notes=notes)
354 save_decisions(entries, path)
355 return
356 raise KeyError(f"No entry with id {entry_id!r}")
359def load_coupling_entries(
360 path: Path | None = None,
361 *,
362 changed_globs: list[str] | None = None,
363 archetype: str | None = None,
364) -> list[CouplingEntry]:
365 """Return coupling entries, optionally filtered by glob match against changed files and archetype.
367 Returns an empty list when decisions.yaml is absent (graceful degradation).
368 """
369 from fnmatch import fnmatch
371 entries = [e for e in load_decisions(path) if isinstance(e, CouplingEntry)]
373 if archetype is not None:
374 entries = [e for e in entries if e.archetype == archetype]
376 if changed_globs is not None:
377 entries = [e for e in entries if any(fnmatch(f, e.if_changed) for f in changed_globs)]
379 return entries
382def generate_from_completed(config: BRConfig) -> int:
383 """Generate DecisionEntry records from completed issues and persist to the log.
385 Prefers the SQLite history DB when present; falls back to filesystem scanning.
386 Skips issues that already have an entry in the log. Returns the count added.
387 """
388 from little_loops.issue_history.parsing import (
389 scan_completed_issues,
390 scan_completed_issues_from_db,
391 )
393 project_root = Path(config.project_root)
394 log_path = project_root / config.decisions.log_path
395 db_path = project_root / ".ll" / "history.db"
397 if db_path.exists():
398 completed = scan_completed_issues_from_db(db_path)
399 else:
400 completed = scan_completed_issues(project_root / config.issues.base_dir)
402 existing = load_decisions(log_path)
403 existing_issue_ids = {e.issue for e in existing if isinstance(e, DecisionEntry) and e.issue}
405 count = 0
406 for issue in completed:
407 if issue.issue_id in existing_issue_ids:
408 continue
409 ts = ""
410 if issue.completed_at:
411 ts = issue.completed_at.strftime("%Y-%m-%dT%H:%M:%SZ")
412 elif issue.captured_at:
413 ts = issue.captured_at.strftime("%Y-%m-%dT%H:%M:%SZ")
414 entry = DecisionEntry(
415 id=f"DEC-{issue.issue_id}",
416 timestamp=ts,
417 category=issue.issue_type.lower(),
418 labels=[issue.priority, issue.issue_type.lower()],
419 rationale=f"Auto-generated from completed issue {issue.issue_id}",
420 issue=issue.issue_id,
421 scope="issue",
422 )
423 add_entry(entry, log_path)
424 count += 1
426 return count