Coverage for little_loops / cli / issues / decisions.py: 0%
156 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-06-04 12:21 -0500
1"""ll-issues decisions: Manage rules, decisions, and exceptions log."""
3from __future__ import annotations
5import argparse
6import sys
7from datetime import UTC, datetime
8from typing import TYPE_CHECKING
10if TYPE_CHECKING:
11 from little_loops.config import BRConfig
14def add_decisions_parser(subs: argparse._SubParsersAction) -> argparse.ArgumentParser:
15 """Register the decisions subparser with all sub-sub-commands on *subs*."""
16 from little_loops.cli_args import add_config_arg
18 p = subs.add_parser(
19 "decisions",
20 help="Manage rules, decisions, and exceptions log",
21 )
22 p.set_defaults(command="decisions")
23 p.set_defaults(_decisions_parser=p)
25 add_config_arg(p)
26 subsubs = p.add_subparsers(dest="subcommand")
28 # -- list --
29 list_p = subsubs.add_parser("list", help="List decisions log entries")
30 list_p.add_argument(
31 "--type",
32 choices=["rule", "decision", "exception"],
33 default=None,
34 help="Filter by entry type",
35 )
36 list_p.add_argument("--category", default=None, help="Filter by category")
37 list_p.add_argument("--label", default=None, help="Filter by label")
38 list_p.add_argument(
39 "--no-outcome",
40 action="store_true",
41 help="Show only DecisionEntry records with no recorded outcome",
42 )
43 list_p.add_argument(
44 "--before",
45 default=None,
46 metavar="ISO-8601",
47 help="Show entries with timestamp before this date",
48 )
49 list_p.add_argument(
50 "--scope",
51 default=None,
52 help="Filter DecisionEntry records by scope (e.g. 'issue', 'project')",
53 )
54 list_p.add_argument(
55 "--format",
56 "-f",
57 choices=["text", "json"],
58 default="text",
59 help="Output format (default: text)",
60 )
61 list_p.add_argument(
62 "--active-only",
63 action="store_true",
64 help="Exclude entries superseded by a newer entry",
65 )
66 add_config_arg(list_p)
68 # -- add --
69 add_p = subsubs.add_parser("add", help="Add a new entry to the decisions log")
70 add_p.add_argument(
71 "--type",
72 required=True,
73 choices=["rule", "decision", "exception"],
74 help="Entry type",
75 )
76 add_p.add_argument("--category", required=True, help="Category (e.g. 'architecture')")
77 add_p.add_argument(
78 "--rule",
79 default=None,
80 help="Rule or decision text (required for type=rule or type=decision)",
81 )
82 add_p.add_argument(
83 "--rationale",
84 required=True,
85 help="Why this rule/decision/exception applies",
86 )
87 add_p.add_argument("--issue", default=None, help="Related issue ID (e.g. FEAT-1894)")
88 add_p.add_argument("--label", default=None, help="Comma-separated labels")
89 add_p.add_argument(
90 "--enforcement",
91 choices=["required", "advisory"],
92 default="advisory",
93 help="Enforcement level for type=rule (default: advisory)",
94 )
95 add_p.add_argument(
96 "--rule-ref",
97 default=None,
98 dest="rule_ref",
99 metavar="RULE-ID",
100 help="Reference to the rule being excepted (required for type=exception)",
101 )
102 add_p.add_argument(
103 "--alternatives-rejected",
104 default=None,
105 dest="alternatives_rejected",
106 help="Alternatives considered and rejected (for type=decision or type=exception)",
107 )
108 add_p.add_argument(
109 "--supersedes",
110 default=None,
111 metavar="ENTRY-ID",
112 help="ID of the entry this one supersedes (for type=rule)",
113 )
114 add_p.add_argument(
115 "--scope",
116 default="issue",
117 help="Decision scope: 'issue' (default) or 'project' (for type=decision)",
118 )
119 add_p.add_argument(
120 "--id",
121 default=None,
122 dest="entry_id",
123 metavar="ENTRY-ID",
124 help="Explicit entry ID (auto-generated if omitted)",
125 )
126 add_config_arg(add_p)
128 # -- outcome --
129 outcome_p = subsubs.add_parser("outcome", help="Record the outcome of a decision entry")
130 outcome_p.add_argument("id", help="Entry ID to record outcome for")
131 outcome_p.add_argument(
132 "--result",
133 required=True,
134 choices=["worked", "did_not_work", "mixed", "reversed"],
135 help="Outcome result",
136 )
137 outcome_p.add_argument("--notes", default=None, help="Free-text notes about the outcome")
138 outcome_p.add_argument(
139 "--measured-at",
140 default=None,
141 dest="measured_at",
142 metavar="ISO-8601",
143 help="When the outcome was measured (default: now)",
144 )
145 outcome_p.add_argument(
146 "--force",
147 action="store_true",
148 help="Overwrite an existing outcome",
149 )
150 add_config_arg(outcome_p)
152 # -- generate (stub) --
153 gen_p = subsubs.add_parser(
154 "generate",
155 help="Generate decisions log entries from completed issues",
156 )
157 gen_p.add_argument(
158 "--from",
159 dest="source",
160 default="completed",
161 choices=["completed"],
162 help="Source to generate from (default: completed)",
163 )
164 add_config_arg(gen_p)
166 # -- sync (stub) --
167 sync_p = subsubs.add_parser("sync", help="Sync active rules to .ll/ll.local.md")
168 add_config_arg(sync_p)
170 return p
173def cmd_decisions(config: BRConfig, args: argparse.Namespace) -> int:
174 """Dispatch decisions sub-sub-commands; returns 0 on success, 1 on error."""
175 from pathlib import Path
177 from little_loops.decisions import (
178 DecisionEntry,
179 ExceptionEntry,
180 RuleEntry,
181 add_entry,
182 list_entries,
183 resolve_active,
184 set_outcome,
185 )
187 if not getattr(args, "subcommand", None):
188 args._decisions_parser.print_help()
189 return 1
191 # Resolve the decisions log path from config
192 path = Path(config.project_root) / config.decisions.log_path
194 sub = args.subcommand
196 if sub == "list":
197 return _cmd_list(args, path, list_entries, resolve_active)
199 if sub == "add":
200 return _cmd_add(
201 args, path, RuleEntry, DecisionEntry, ExceptionEntry, add_entry, list_entries
202 )
204 if sub == "outcome":
205 return _cmd_outcome(args, path, set_outcome)
207 if sub == "generate":
208 from little_loops.decisions import generate_from_completed
210 count = generate_from_completed(config)
211 print(
212 f"Generated {count} decision entr{'y' if count == 1 else 'ies'} from completed issues."
213 )
214 return 0
216 if sub == "sync":
217 return _cmd_sync(path)
219 print(f"Unknown subcommand: {sub!r}", file=sys.stderr)
220 return 1
223def _cmd_list(args, path, list_entries, resolve_active) -> int:
224 import json
226 entries = list_entries(
227 path=path,
228 type=getattr(args, "type", None),
229 category=getattr(args, "category", None),
230 label=getattr(args, "label", None),
231 )
233 # Post-filters
234 if getattr(args, "no_outcome", False):
235 from little_loops.decisions import DecisionEntry
237 entries = [e for e in entries if isinstance(e, DecisionEntry) and e.outcome is None]
239 if getattr(args, "before", None):
240 entries = [e for e in entries if e.timestamp < args.before]
242 if getattr(args, "scope", None):
243 from little_loops.decisions import DecisionEntry
245 entries = [e for e in entries if isinstance(e, DecisionEntry) and e.scope == args.scope]
247 if getattr(args, "active_only", False):
248 entries = resolve_active(entries)
250 fmt = getattr(args, "format", "text") or "text"
251 if fmt == "json":
252 print(json.dumps([e.to_dict() for e in entries], indent=2))
253 return 0
255 if not entries:
256 print("(no entries)")
257 return 0
259 for entry in entries:
260 _print_entry(entry)
261 return 0
264def _print_entry(entry) -> None:
265 from little_loops.decisions import DecisionEntry, ExceptionEntry, RuleEntry
267 label_str = ", ".join(entry.labels) if entry.labels else ""
268 print(
269 f"{entry.id} [{entry.type}] {entry.category}" + (f" ({label_str})" if label_str else "")
270 )
271 if isinstance(entry, (RuleEntry, DecisionEntry)):
272 print(f" rule: {entry.rule}")
273 elif isinstance(entry, ExceptionEntry):
274 print(f" rule_ref: {entry.rule_ref}")
275 if entry.rationale:
276 print(f" rationale: {entry.rationale}")
277 if isinstance(entry, DecisionEntry) and entry.outcome:
278 print(f" outcome: {entry.outcome.result} @ {entry.outcome.measured_at}")
281def _cmd_add(args, path, RuleEntry, DecisionEntry, ExceptionEntry, add_entry, list_entries) -> int:
282 entry_type = args.type
284 # Validate type-specific required fields
285 if entry_type == "rule" and not getattr(args, "rule", None):
286 print("Error: --rule is required for type 'rule'", file=sys.stderr)
287 return 1
288 if entry_type == "decision" and not getattr(args, "rule", None):
289 print("Error: --rule is required for type 'decision' (the decision text)", file=sys.stderr)
290 return 1
291 if entry_type == "exception" and not getattr(args, "rule_ref", None):
292 print("Error: --rule-ref is required for type 'exception'", file=sys.stderr)
293 return 1
295 # Generate an ID if not provided
296 entry_id = getattr(args, "entry_id", None)
297 if not entry_id:
298 existing = list_entries(path=path, type=entry_type)
299 prefix = {
300 "rule": args.category.upper() if args.category else "RULE",
301 "decision": args.category.upper() if args.category else "DEC",
302 "exception": args.category.upper() + "-EX" if args.category else "EX",
303 }[entry_type]
304 entry_id = f"{prefix}-{len(existing) + 1:03d}"
306 timestamp = datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%SZ")
307 labels = [lbl.strip() for lbl in args.label.split(",")] if getattr(args, "label", None) else []
309 if entry_type == "rule":
310 entry = RuleEntry(
311 id=entry_id,
312 timestamp=timestamp,
313 category=args.category,
314 labels=labels,
315 rationale=args.rationale,
316 rule=args.rule,
317 enforcement=getattr(args, "enforcement", "advisory"),
318 supersedes=getattr(args, "supersedes", None),
319 issue=getattr(args, "issue", None),
320 )
321 elif entry_type == "decision":
322 entry = DecisionEntry(
323 id=entry_id,
324 timestamp=timestamp,
325 category=args.category,
326 labels=labels,
327 rationale=args.rationale,
328 rule=args.rule,
329 alternatives_rejected=getattr(args, "alternatives_rejected", None),
330 issue=getattr(args, "issue", None),
331 scope=getattr(args, "scope", "issue"),
332 )
333 else: # exception
334 entry = ExceptionEntry(
335 id=entry_id,
336 timestamp=timestamp,
337 category=args.category,
338 labels=labels,
339 rationale=args.rationale,
340 rule_ref=args.rule_ref,
341 issue=getattr(args, "issue", "") or "",
342 alternatives_rejected=getattr(args, "alternatives_rejected", None),
343 )
345 add_entry(entry, path=path)
346 print(f"Added {entry_type} entry: {entry_id}")
347 return 0
350def _cmd_outcome(args, path, set_outcome) -> int:
351 measured_at = getattr(args, "measured_at", None) or datetime.now(UTC).strftime(
352 "%Y-%m-%dT%H:%M:%SZ"
353 )
354 try:
355 set_outcome(
356 args.id,
357 args.result,
358 measured_at,
359 notes=getattr(args, "notes", None),
360 path=path,
361 force=getattr(args, "force", False),
362 )
363 except KeyError:
364 print(f"Error: entry {args.id!r} not found", file=sys.stderr)
365 return 1
366 except TypeError as exc:
367 print(f"Error: {exc}", file=sys.stderr)
368 return 1
369 except ValueError as exc:
370 print(f"Error: {exc} (use --force to overwrite)", file=sys.stderr)
371 return 1
372 print(f"Recorded outcome for {args.id}: {args.result}")
373 return 0
376def _cmd_sync(path) -> int:
377 try:
378 from little_loops.decisions_sync import sync_to_local_md
380 sync_to_local_md(path=path)
381 except ImportError:
382 print("sync not yet available (requires FEAT-1895)", file=sys.stderr)
383 return 1
384 return 0