Coverage for little_loops / cli / issues / decisions.py: 0%
180 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-06-08 15:34 -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", "coupling"],
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 "--archetype",
40 default=None,
41 help="Filter coupling entries by archetype label (e.g. add-cli-command)",
42 )
43 list_p.add_argument(
44 "--no-outcome",
45 action="store_true",
46 help="Show only DecisionEntry records with no recorded outcome",
47 )
48 list_p.add_argument(
49 "--before",
50 default=None,
51 metavar="ISO-8601",
52 help="Show entries with timestamp before this date",
53 )
54 list_p.add_argument(
55 "--scope",
56 default=None,
57 help="Filter DecisionEntry records by scope (e.g. 'issue', 'project')",
58 )
59 list_p.add_argument(
60 "--format",
61 "-f",
62 choices=["text", "json"],
63 default="text",
64 help="Output format (default: text)",
65 )
66 list_p.add_argument(
67 "--active-only",
68 action="store_true",
69 help="Exclude entries superseded by a newer entry",
70 )
71 add_config_arg(list_p)
73 # -- add --
74 add_p = subsubs.add_parser("add", help="Add a new entry to the decisions log")
75 add_p.add_argument(
76 "--type",
77 required=True,
78 choices=["rule", "decision", "exception", "coupling"],
79 help="Entry type",
80 )
81 add_p.add_argument("--category", required=True, help="Category (e.g. 'architecture')")
82 add_p.add_argument(
83 "--rule",
84 default=None,
85 help="Rule or decision text (required for type=rule or type=decision)",
86 )
87 add_p.add_argument(
88 "--rationale",
89 required=True,
90 help="Why this rule/decision/exception applies",
91 )
92 add_p.add_argument("--issue", default=None, help="Related issue ID (e.g. FEAT-1894)")
93 add_p.add_argument("--label", default=None, help="Comma-separated labels")
94 add_p.add_argument(
95 "--enforcement",
96 choices=["required", "advisory"],
97 default="advisory",
98 help="Enforcement level for type=rule (default: advisory)",
99 )
100 add_p.add_argument(
101 "--rule-ref",
102 default=None,
103 dest="rule_ref",
104 metavar="RULE-ID",
105 help="Reference to the rule being excepted (required for type=exception)",
106 )
107 add_p.add_argument(
108 "--alternatives-rejected",
109 default=None,
110 dest="alternatives_rejected",
111 help="Alternatives considered and rejected (for type=decision or type=exception)",
112 )
113 add_p.add_argument(
114 "--supersedes",
115 default=None,
116 metavar="ENTRY-ID",
117 help="ID of the entry this one supersedes (for type=rule)",
118 )
119 add_p.add_argument(
120 "--scope",
121 default="issue",
122 help="Decision scope: 'issue' (default) or 'project' (for type=decision)",
123 )
124 add_p.add_argument(
125 "--id",
126 default=None,
127 dest="entry_id",
128 metavar="ENTRY-ID",
129 help="Explicit entry ID (auto-generated if omitted)",
130 )
131 # Coupling-specific flags (only relevant when --type=coupling)
132 add_p.add_argument(
133 "--if-changed",
134 default=None,
135 dest="if_changed",
136 metavar="GLOB",
137 help="Glob pattern matching files being modified (required for type=coupling)",
138 )
139 add_p.add_argument(
140 "--then-check",
141 default=None,
142 dest="then_check",
143 metavar="PATHS",
144 help="Comma-separated file/pattern list to audit for wiring gaps (required for type=coupling)",
145 )
146 add_p.add_argument(
147 "--tier",
148 choices=["hard", "soft", "fyi"],
149 default="soft",
150 help="Coupling tier: hard (must change together), soft (should update), fyi (informational)",
151 )
152 add_p.add_argument(
153 "--archetype",
154 default=None,
155 dest="archetype",
156 help="Named bundle label grouping related coupling rules (e.g. add-cli-command)",
157 )
158 add_config_arg(add_p)
160 # -- outcome --
161 outcome_p = subsubs.add_parser("outcome", help="Record the outcome of a decision entry")
162 outcome_p.add_argument("id", help="Entry ID to record outcome for")
163 outcome_p.add_argument(
164 "--result",
165 required=True,
166 choices=["worked", "did_not_work", "mixed", "reversed"],
167 help="Outcome result",
168 )
169 outcome_p.add_argument("--notes", default=None, help="Free-text notes about the outcome")
170 outcome_p.add_argument(
171 "--measured-at",
172 default=None,
173 dest="measured_at",
174 metavar="ISO-8601",
175 help="When the outcome was measured (default: now)",
176 )
177 outcome_p.add_argument(
178 "--force",
179 action="store_true",
180 help="Overwrite an existing outcome",
181 )
182 add_config_arg(outcome_p)
184 # -- generate (stub) --
185 gen_p = subsubs.add_parser(
186 "generate",
187 help="Generate decisions log entries from completed issues",
188 )
189 gen_p.add_argument(
190 "--from",
191 dest="source",
192 default="completed",
193 choices=["completed"],
194 help="Source to generate from (default: completed)",
195 )
196 add_config_arg(gen_p)
198 # -- sync (stub) --
199 sync_p = subsubs.add_parser("sync", help="Sync active rules to .ll/ll.local.md")
200 add_config_arg(sync_p)
202 return p
205def cmd_decisions(config: BRConfig, args: argparse.Namespace) -> int:
206 """Dispatch decisions sub-sub-commands; returns 0 on success, 1 on error."""
207 from pathlib import Path
209 from little_loops.decisions import (
210 CouplingEntry,
211 DecisionEntry,
212 ExceptionEntry,
213 RuleEntry,
214 add_entry,
215 list_entries,
216 resolve_active,
217 set_outcome,
218 )
220 if not getattr(args, "subcommand", None):
221 args._decisions_parser.print_help()
222 return 1
224 # Resolve the decisions log path from config
225 path = Path(config.project_root) / config.decisions.log_path
227 sub = args.subcommand
229 if sub == "list":
230 return _cmd_list(args, path, list_entries, resolve_active)
232 if sub == "add":
233 return _cmd_add(
234 args,
235 path,
236 RuleEntry,
237 DecisionEntry,
238 ExceptionEntry,
239 CouplingEntry,
240 add_entry,
241 list_entries,
242 )
244 if sub == "outcome":
245 return _cmd_outcome(args, path, set_outcome)
247 if sub == "generate":
248 from little_loops.decisions import generate_from_completed
250 count = generate_from_completed(config)
251 print(
252 f"Generated {count} decision entr{'y' if count == 1 else 'ies'} from completed issues."
253 )
254 return 0
256 if sub == "sync":
257 return _cmd_sync(path)
259 print(f"Unknown subcommand: {sub!r}", file=sys.stderr)
260 return 1
263def _cmd_list(args, path, list_entries, resolve_active) -> int:
264 import json
266 entries = list_entries(
267 path=path,
268 type=getattr(args, "type", None),
269 category=getattr(args, "category", None),
270 label=getattr(args, "label", None),
271 )
273 # Post-filters
274 if getattr(args, "no_outcome", False):
275 from little_loops.decisions import DecisionEntry
277 entries = [e for e in entries if isinstance(e, DecisionEntry) and e.outcome is None]
279 if getattr(args, "before", None):
280 entries = [e for e in entries if e.timestamp < args.before]
282 if getattr(args, "scope", None):
283 from little_loops.decisions import DecisionEntry
285 entries = [e for e in entries if isinstance(e, DecisionEntry) and e.scope == args.scope]
287 if getattr(args, "active_only", False):
288 entries = resolve_active(entries)
290 if getattr(args, "archetype", None):
291 from little_loops.decisions import CouplingEntry as _CE
293 entries = [e for e in entries if isinstance(e, _CE) and e.archetype == args.archetype]
295 fmt = getattr(args, "format", "text") or "text"
296 if fmt == "json":
297 print(json.dumps([e.to_dict() for e in entries], indent=2))
298 return 0
300 if not entries:
301 print("(no entries)")
302 return 0
304 for entry in entries:
305 _print_entry(entry)
306 return 0
309def _print_entry(entry) -> None:
310 from little_loops.decisions import CouplingEntry, DecisionEntry, ExceptionEntry, RuleEntry
312 label_str = ", ".join(entry.labels) if entry.labels else ""
313 print(
314 f"{entry.id} [{entry.type}] {entry.category}" + (f" ({label_str})" if label_str else "")
315 )
316 if isinstance(entry, (RuleEntry, DecisionEntry)):
317 print(f" rule: {entry.rule}")
318 elif isinstance(entry, ExceptionEntry):
319 print(f" rule_ref: {entry.rule_ref}")
320 elif isinstance(entry, CouplingEntry):
321 print(f" if_changed: {entry.if_changed}")
322 print(f" then_check: {', '.join(entry.then_check)}")
323 print(f" tier: {entry.tier}")
324 if entry.archetype:
325 print(f" archetype: {entry.archetype}")
326 if entry.rationale:
327 print(f" rationale: {entry.rationale}")
328 if isinstance(entry, DecisionEntry) and entry.outcome:
329 print(f" outcome: {entry.outcome.result} @ {entry.outcome.measured_at}")
332def _cmd_add(
333 args, path, RuleEntry, DecisionEntry, ExceptionEntry, CouplingEntry, add_entry, list_entries
334) -> int:
335 entry_type = args.type
337 # Validate type-specific required fields
338 if entry_type == "rule" and not getattr(args, "rule", None):
339 print("Error: --rule is required for type 'rule'", file=sys.stderr)
340 return 1
341 if entry_type == "decision" and not getattr(args, "rule", None):
342 print("Error: --rule is required for type 'decision' (the decision text)", file=sys.stderr)
343 return 1
344 if entry_type == "exception" and not getattr(args, "rule_ref", None):
345 print("Error: --rule-ref is required for type 'exception'", file=sys.stderr)
346 return 1
347 if entry_type == "coupling":
348 if not getattr(args, "if_changed", None):
349 print("Error: --if-changed is required for type 'coupling'", file=sys.stderr)
350 return 1
351 if not getattr(args, "then_check", None):
352 print("Error: --then-check is required for type 'coupling'", file=sys.stderr)
353 return 1
355 # Generate an ID if not provided
356 entry_id = getattr(args, "entry_id", None)
357 if not entry_id:
358 existing = list_entries(path=path, type=entry_type)
359 prefix = {
360 "rule": args.category.upper() if args.category else "RULE",
361 "decision": args.category.upper() if args.category else "DEC",
362 "exception": args.category.upper() + "-EX" if args.category else "EX",
363 "coupling": "COUPLING",
364 }[entry_type]
365 entry_id = f"{prefix}-{len(existing) + 1:03d}"
367 timestamp = datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%SZ")
368 labels = [lbl.strip() for lbl in args.label.split(",")] if getattr(args, "label", None) else []
370 if entry_type == "rule":
371 entry = RuleEntry(
372 id=entry_id,
373 timestamp=timestamp,
374 category=args.category,
375 labels=labels,
376 rationale=args.rationale,
377 rule=args.rule,
378 enforcement=getattr(args, "enforcement", "advisory"),
379 supersedes=getattr(args, "supersedes", None),
380 issue=getattr(args, "issue", None),
381 )
382 elif entry_type == "decision":
383 entry = DecisionEntry(
384 id=entry_id,
385 timestamp=timestamp,
386 category=args.category,
387 labels=labels,
388 rationale=args.rationale,
389 rule=args.rule,
390 alternatives_rejected=getattr(args, "alternatives_rejected", None),
391 issue=getattr(args, "issue", None),
392 scope=getattr(args, "scope", "issue"),
393 )
394 elif entry_type == "coupling":
395 then_check = [t.strip() for t in args.then_check.split(",")]
396 entry = CouplingEntry(
397 id=entry_id,
398 timestamp=timestamp,
399 category=args.category,
400 labels=labels,
401 rationale=args.rationale,
402 if_changed=args.if_changed,
403 then_check=then_check,
404 tier=getattr(args, "tier", "soft"),
405 archetype=getattr(args, "archetype", None),
406 enforcement=getattr(args, "enforcement", "advisory"),
407 issue=getattr(args, "issue", None),
408 )
409 else: # exception
410 entry = ExceptionEntry(
411 id=entry_id,
412 timestamp=timestamp,
413 category=args.category,
414 labels=labels,
415 rationale=args.rationale,
416 rule_ref=args.rule_ref,
417 issue=getattr(args, "issue", "") or "",
418 alternatives_rejected=getattr(args, "alternatives_rejected", None),
419 )
421 add_entry(entry, path=path)
422 print(f"Added {entry_type} entry: {entry_id}")
423 return 0
426def _cmd_outcome(args, path, set_outcome) -> int:
427 measured_at = getattr(args, "measured_at", None) or datetime.now(UTC).strftime(
428 "%Y-%m-%dT%H:%M:%SZ"
429 )
430 try:
431 set_outcome(
432 args.id,
433 args.result,
434 measured_at,
435 notes=getattr(args, "notes", None),
436 path=path,
437 force=getattr(args, "force", False),
438 )
439 except KeyError:
440 print(f"Error: entry {args.id!r} not found", file=sys.stderr)
441 return 1
442 except TypeError as exc:
443 print(f"Error: {exc}", file=sys.stderr)
444 return 1
445 except ValueError as exc:
446 print(f"Error: {exc} (use --force to overwrite)", file=sys.stderr)
447 return 1
448 print(f"Recorded outcome for {args.id}: {args.result}")
449 return 0
452def _cmd_sync(path) -> int:
453 try:
454 from little_loops.decisions_sync import sync_to_local_md
456 sync_to_local_md(path=path)
457 except ImportError:
458 print("sync not yet available (requires FEAT-1895)", file=sys.stderr)
459 return 1
460 return 0