Coverage for little_loops / cli / issues / search.py: 0%
267 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-06-29 00:54 -0500
1"""ll-issues search: Search issues with filters and sorting."""
3from __future__ import annotations
5import argparse
6import re
7from collections.abc import Callable
8from datetime import date, datetime
9from typing import TYPE_CHECKING
11from little_loops.cli.output import PRIORITY_COLOR, TYPE_COLOR, colorize, print_json
13if TYPE_CHECKING:
14 from pathlib import Path
16 from little_loops.config import BRConfig
17 from little_loops.config.features import NextIssueConfig
18 from little_loops.issue_parser import IssueInfo
21def _parse_discovered_date(content: str, file_path: Path | None = None) -> datetime | None:
22 """Extract creation datetime from YAML frontmatter.
24 Prefers ``captured_at`` (ISO datetime, sub-day resolution) when present,
25 falling back to ``discovered_date`` (date-granular), then to file mtime
26 when ``file_path`` is provided and both frontmatter fields are absent.
27 Always returns a ``datetime`` so sort comparisons never mix ``date`` and
28 ``datetime``.
29 """
30 from little_loops.frontmatter import parse_frontmatter
32 fm = parse_frontmatter(content)
33 captured = fm.get("captured_at")
34 if isinstance(captured, str) and captured:
35 try:
36 return datetime.fromisoformat(captured.rstrip("Z")).replace(tzinfo=None)
37 except ValueError:
38 pass
40 match = re.match(r"^---\n(.*?)\n---", content, re.DOTALL)
41 if match:
42 date_match = re.search(r"discovered_date:\s*(\S+)", match.group(1))
43 if date_match:
44 date_str = date_match.group(1).strip("\"'")
45 try:
46 d = date.fromisoformat(date_str[:10])
47 return datetime.combine(d, datetime.min.time())
48 except ValueError:
49 pass
51 if file_path is not None:
52 try:
53 return datetime.fromtimestamp(file_path.stat().st_mtime)
54 except OSError:
55 pass
57 return None
60def _parse_updated_date(content: str, file_path: Path) -> date | None:
61 r"""Extract last-activity date from the ## Session Log section, or fall back to file mtime.
63 Scans for the most recent timestamp entry in the Session Log section.
64 Entry format: `- \`/ll:cmd\` - YYYY-MM-DDTHH:MM:SS - \`path\``
65 Falls back to file mtime when no session log timestamps are found.
66 """
67 import re as _re
69 _SESSION_LOG_RE = _re.compile(
70 r"^## Session Log\s*\n+(.*?)(?:\n##|\n---|\Z)", _re.MULTILINE | _re.DOTALL
71 )
72 _TIMESTAMP_RE = _re.compile(r"- `[^`]+` - (\d{4}-\d{2}-\d{2})T\d{2}:\d{2}:\d{2}")
74 match = _SESSION_LOG_RE.search(content)
75 if match:
76 timestamps = _TIMESTAMP_RE.findall(match.group(1))
77 if timestamps:
78 try:
79 return date.fromisoformat(timestamps[-1])
80 except ValueError:
81 pass
83 # Fallback to file mtime
84 try:
85 return date.fromtimestamp(file_path.stat().st_mtime)
86 except OSError:
87 return None
90def _parse_labels_from_content(content: str) -> list[str]:
91 """Extract labels from ## Labels section (backtick-wrapped items)."""
92 match = re.search(r"## Labels\s*\n(.*?)(?:\n##|\Z)", content, re.DOTALL)
93 if not match:
94 return []
95 return [m.lower() for m in re.findall(r"`([^`]+)`", match.group(1))]
98def _parse_summary_from_content(content: str) -> str:
99 """Extract plain text of the ## Summary section body."""
100 match = re.search(r"^## Summary\s*\n+(.*?)(?:\n\n|\n##|\Z)", content, re.MULTILINE | re.DOTALL)
101 if not match:
102 return ""
103 return match.group(1).strip()
106def _parse_priority_filter(priority_values: list[str]) -> set[str]:
107 """Parse priority filter values, supporting ranges like P0-P2."""
108 priorities: set[str] = set()
109 for val in priority_values:
110 range_match = re.match(r"^(P\d)-(P\d)$", val)
111 if range_match:
112 start = int(range_match.group(1)[1:])
113 end = int(range_match.group(2)[1:])
114 for i in range(start, end + 1):
115 priorities.add(f"P{i}")
116 elif re.match(r"^P\d$", val):
117 priorities.add(val)
118 return priorities
121def _load_issues_with_status(
122 config: BRConfig,
123 include_open: bool,
124 include_done: bool,
125 include_deferred: bool,
126) -> list[tuple[IssueInfo, str]]:
127 """Load issues from type directories, tagged with their frontmatter status.
129 Scans only type-scoped directories (bugs/, features/, etc.) and reads
130 ``IssueInfo.status`` from frontmatter instead of inferring status from the
131 directory name.
133 Returns:
134 List of (IssueInfo, status) where status is the frontmatter value
135 (e.g. 'open', 'in_progress', 'blocked', 'done', 'cancelled', 'deferred').
136 """
137 from little_loops.issue_parser import IssueParser
139 parser = IssueParser(config)
140 results: list[tuple[IssueInfo, str]] = []
142 for category in config.issue_categories:
143 issue_dir = config.get_issue_dir(category)
144 if not issue_dir.exists():
145 continue
146 for f in sorted(issue_dir.glob("*.md")):
147 try:
148 issue = parser.parse_file(f)
149 status = issue.status # frontmatter field, default "open"
150 if status in ("open", "in_progress", "blocked"):
151 if include_open:
152 results.append((issue, status))
153 elif status in ("done", "cancelled"):
154 if include_done:
155 results.append((issue, status))
156 elif status == "deferred":
157 if include_deferred:
158 results.append((issue, status))
159 elif include_open:
160 # Unknown status: treat as open
161 results.append((issue, status))
162 except Exception:
163 continue
165 return results
168_STRATEGY_SORT_KEYS: dict[str, tuple[tuple[str, str], ...]] = {
169 "confidence_first": (
170 ("outcome_confidence", "desc"),
171 ("confidence_score", "desc"),
172 ("priority", "asc"),
173 ),
174 "priority_first": (
175 ("priority", "asc"),
176 ("outcome_confidence", "desc"),
177 ("confidence_score", "desc"),
178 ),
179}
182def build_sort_key(config: NextIssueConfig) -> Callable[[IssueInfo], tuple]:
183 """Return a sort-key callable for an IssueInfo list based on NextIssueConfig.
185 Resolution order:
186 - If ``config.sort_keys`` is set, it takes precedence over ``config.strategy``.
187 - Otherwise, the named strategy preset is used.
189 The returned callable produces a multi-field tuple where each component is
190 transformed per-field so that a plain ``sorted(..., reverse=False)`` call
191 yields the desired order. Do NOT pass ``reverse=True`` — mixed directions
192 are baked into each tuple component.
194 None-handling (per-field sentinel):
195 - ``direction="desc"`` (higher = better): component = ``-value`` when set, ``1`` when None.
196 This mirrors ``-(value or -1)`` — values > 0 sort first, None sorts after 0.
197 - ``direction="asc"`` (lower = better): component = ``value`` when set, ``9999`` when None.
199 The schema key ``"priority"`` maps to ``IssueInfo.priority_int``; all other
200 keys are read directly off the dataclass.
202 Raises:
203 ValueError: If ``config.strategy`` is unknown (``from_dict`` normally
204 guards this, but direct instantiation can bypass validation).
205 """
206 if config.sort_keys is not None:
207 entries: tuple[tuple[str, str], ...] = tuple(
208 (sk.key, sk.direction) for sk in config.sort_keys
209 )
210 else:
211 preset = _STRATEGY_SORT_KEYS.get(config.strategy)
212 if preset is None:
213 raise ValueError(f"Unknown strategy: {config.strategy!r}")
214 entries = preset
216 def key(issue: IssueInfo) -> tuple:
217 parts: list[int] = []
218 for field_name, direction in entries:
219 attr = "priority_int" if field_name == "priority" else field_name
220 value = getattr(issue, attr)
221 if direction == "desc":
222 parts.append(-value if value is not None else 1)
223 else:
224 parts.append(value if value is not None else 9999)
225 return tuple(parts)
227 return key
230def _sort_issues(
231 items: list[tuple],
232 sort_field: str,
233 descending: bool,
234) -> list[tuple]:
235 """Sort issues by the requested field."""
237 def key(item: tuple) -> tuple:
238 issue, _status, disc_date, comp_date, *_rest = item
239 if sort_field == "priority":
240 return (issue.priority_int, issue.issue_id)
241 if sort_field == "id":
242 m = re.search(r"-(\d+)$", issue.issue_id)
243 num = int(m.group(1)) if m else 0
244 return (issue.issue_id.split("-", 1)[0], num)
245 if sort_field in ("date", "created"):
246 sentinel = datetime.min if descending else datetime.max
247 return (disc_date if disc_date is not None else sentinel,)
248 if sort_field == "completed":
249 sentinel_d = date.min if descending else date.max
250 return (comp_date if comp_date is not None else sentinel_d,)
251 if sort_field == "type":
252 return (issue.issue_id.split("-", 1)[0], issue.priority_int, issue.issue_id)
253 if sort_field == "title":
254 return (issue.title.lower(),)
255 if sort_field == "confidence":
256 score = (
257 issue.confidence_score
258 if issue.confidence_score is not None
259 else (-1 if descending else 9999)
260 )
261 return (score,)
262 if sort_field == "outcome":
263 score = (
264 issue.outcome_confidence
265 if issue.outcome_confidence is not None
266 else (-1 if descending else 9999)
267 )
268 return (score,)
269 if sort_field == "refinement":
270 refinement_commands = {
271 "/ll:verify-issues",
272 "/ll:refine-issue",
273 "/ll:tradeoff-review-issues",
274 "/ll:map-dependencies",
275 "/ll:ready-issue",
276 }
277 counts: dict[str, int] = getattr(issue, "session_command_counts", {}) or {}
278 total = sum(counts.get(cmd, 0) for cmd in refinement_commands)
279 return (total,)
280 return (issue.priority_int, issue.issue_id)
282 return sorted(items, key=key, reverse=descending)
285def cmd_search(config: BRConfig, args: argparse.Namespace) -> int:
286 """Search issues with optional text query, filters, and sorting.
288 Args:
289 config: Project configuration
290 args: Parsed arguments
292 Returns:
293 Exit code (0 = success)
294 """
295 # Resolve status flags
296 status = getattr(args, "status", "open")
297 if getattr(args, "include_completed", False):
298 status = "all"
300 include_open = status in ("open", "in_progress", "blocked", "all")
301 include_done = status in ("done", "cancelled", "all")
302 include_deferred = status in ("deferred", "all")
304 # Load issues
305 raw = _load_issues_with_status(config, include_open, include_done, include_deferred)
307 # Parse additional metadata (dates, labels) only when needed
308 query: str | None = getattr(args, "query", None)
309 since_date: date | None = None
310 until_date: date | None = None
311 raw_since = getattr(args, "since", None)
312 raw_until = getattr(args, "until", None)
313 if raw_since:
314 try:
315 since_date = date.fromisoformat(raw_since)
316 except ValueError:
317 print(f"Invalid --since date: {raw_since!r}. Use YYYY-MM-DD.")
318 return 1
319 if raw_until:
320 try:
321 until_date = date.fromisoformat(raw_until)
322 except ValueError:
323 print(f"Invalid --until date: {raw_until!r}. Use YYYY-MM-DD.")
324 return 1
326 sort_field = getattr(args, "sort", "priority") or "priority"
327 need_content = bool(
328 query
329 or since_date
330 or until_date
331 or getattr(args, "label", None)
332 or sort_field in {"date", "created", "completed"}
333 or getattr(args, "date_field", "discovered") == "updated"
334 )
336 # Build enriched list: (IssueInfo, status, discovered_date, completed_date, labels)
337 enriched: list[tuple[IssueInfo, str, datetime | None, date | None, list[str]]] = []
338 for issue, stat in raw:
339 if need_content:
340 try:
341 content = issue.path.read_text(encoding="utf-8")
342 except Exception:
343 content = ""
344 disc_date = _parse_discovered_date(content, issue.path)
345 labels = _parse_labels_from_content(content)
346 else:
347 content = ""
348 disc_date = None
349 labels = []
351 # --- Filter: text query ---
352 if query:
353 haystack = (issue.title + "\n" + content).lower()
354 if query.lower() not in haystack:
355 continue
357 # --- Filter: type ---
358 type_filters: list[str] = getattr(args, "type", None) or []
359 if type_filters:
360 issue_type = issue.issue_id.split("-", 1)[0]
361 if issue_type not in type_filters:
362 continue
364 # --- Filter: priority ---
365 priority_filters: list[str] = getattr(args, "priority", None) or []
366 if priority_filters:
367 allowed = _parse_priority_filter(priority_filters)
368 if issue.priority not in allowed:
369 continue
371 # --- Filter: label ---
372 label_filters: list[str] = getattr(args, "label", None) or []
373 if label_filters:
374 if not any(lf.lower() in labels for lf in label_filters):
375 continue
377 # --- Filter: date range ---
378 date_field = getattr(args, "date_field", "discovered")
379 ref_date: date | None
380 if date_field == "discovered":
381 # disc_date is datetime for sort precision; normalize to date for day-granular filters
382 ref_date = disc_date.date() if disc_date is not None else None
383 else:
384 ref_date = _parse_updated_date(content, issue.path)
386 if since_date and (ref_date is None or ref_date < since_date):
387 continue
388 if until_date and (ref_date is None or ref_date > until_date):
389 continue
391 comp_date: date | None = None
392 if sort_field == "completed" and need_content:
393 from little_loops.issue_history.parsing import _parse_completion_date
395 comp_date = _parse_completion_date(content, issue.path)
396 enriched.append((issue, stat, disc_date, comp_date, labels))
398 # --- Sort ---
399 # Default direction: desc for date/created/completed (newest first), asc for everything else
400 if getattr(args, "desc", False):
401 descending = True
402 elif getattr(args, "asc", False):
403 descending = False
404 else:
405 descending = sort_field in {"date", "created", "completed"}
407 enriched = _sort_issues(enriched, sort_field, descending)
409 # --- Limit ---
410 limit = getattr(args, "limit", None)
411 if limit and limit > 0:
412 enriched = enriched[:limit]
414 if not enriched:
415 print("No issues found.")
416 return 0
418 # --- Output ---
419 issues_out = [item[0] for item in enriched]
420 statuses_out = [item[1] for item in enriched]
421 dates_out = [item[2] for item in enriched]
422 labels_out = [item[4] for item in enriched]
424 if getattr(args, "json", False):
425 print_json(
426 [
427 {
428 "id": issue.issue_id,
429 "priority": issue.priority,
430 "type": issue.issue_id.split("-", 1)[0],
431 "title": issue.title,
432 "path": str(issue.path),
433 "status": stat,
434 "discovered_date": d.date().isoformat() if d else None,
435 "parent": issue.parent,
436 "labels": lbls,
437 }
438 for issue, stat, d, lbls in zip(
439 issues_out, statuses_out, dates_out, labels_out, strict=True
440 )
441 ]
442 )
443 return 0
445 fmt = getattr(args, "format", "table") or "table"
447 if fmt == "ids":
448 for issue in issues_out:
449 print(issue.issue_id)
450 return 0
452 if fmt == "list":
453 for issue in issues_out:
454 print(f"{issue.path.name} {issue.title}")
455 return 0
457 # Default: table (grouped by type, similar to ll-issues list)
458 buckets: dict[str, list[tuple[IssueInfo, str]]] = {"BUG": [], "FEAT": [], "ENH": [], "EPIC": []}
459 for issue, stat in zip(issues_out, statuses_out, strict=True):
460 prefix = issue.issue_id.split("-", 1)[0]
461 if prefix in buckets:
462 buckets[prefix].append((issue, stat))
464 type_labels = {"BUG": "Bugs", "FEAT": "Features", "ENH": "Enhancements", "EPIC": "Epics"}
465 lines: list[str] = []
466 for prefix, label in type_labels.items():
467 group = buckets[prefix]
468 if not group:
469 continue
470 header = colorize(f"{label} ({len(group)})", f"{TYPE_COLOR.get(prefix, '0')};1")
471 lines.append(header)
472 for issue, stat in group:
473 issue_type = issue.issue_id.split("-", 1)[0]
474 colored_id = colorize(issue.issue_id, TYPE_COLOR.get(issue_type, "0"))
475 colored_priority = colorize(issue.priority, PRIORITY_COLOR.get(issue.priority, "0"))
476 status_tag = f" [{stat}]" if stat not in ("open", "in_progress") else ""
477 lines.append(f" {colored_priority} {colored_id} {issue.title}{status_tag}")
478 lines.append("")
479 lines.append(f"Total: {len(issues_out)} issue(s) found")
480 print("\n".join(lines))
481 return 0