Coverage for little_loops / cli / issues / search.py: 0%
262 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-06-15 17:27 -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_priority_filter(priority_values: list[str]) -> set[str]:
99 """Parse priority filter values, supporting ranges like P0-P2."""
100 priorities: set[str] = set()
101 for val in priority_values:
102 range_match = re.match(r"^(P\d)-(P\d)$", val)
103 if range_match:
104 start = int(range_match.group(1)[1:])
105 end = int(range_match.group(2)[1:])
106 for i in range(start, end + 1):
107 priorities.add(f"P{i}")
108 elif re.match(r"^P\d$", val):
109 priorities.add(val)
110 return priorities
113def _load_issues_with_status(
114 config: BRConfig,
115 include_open: bool,
116 include_done: bool,
117 include_deferred: bool,
118) -> list[tuple[IssueInfo, str]]:
119 """Load issues from type directories, tagged with their frontmatter status.
121 Scans only type-scoped directories (bugs/, features/, etc.) and reads
122 ``IssueInfo.status`` from frontmatter instead of inferring status from the
123 directory name.
125 Returns:
126 List of (IssueInfo, status) where status is the frontmatter value
127 (e.g. 'open', 'in_progress', 'blocked', 'done', 'cancelled', 'deferred').
128 """
129 from little_loops.issue_parser import IssueParser
131 parser = IssueParser(config)
132 results: list[tuple[IssueInfo, str]] = []
134 for category in config.issue_categories:
135 issue_dir = config.get_issue_dir(category)
136 if not issue_dir.exists():
137 continue
138 for f in sorted(issue_dir.glob("*.md")):
139 try:
140 issue = parser.parse_file(f)
141 status = issue.status # frontmatter field, default "open"
142 if status in ("open", "in_progress", "blocked"):
143 if include_open:
144 results.append((issue, status))
145 elif status in ("done", "cancelled"):
146 if include_done:
147 results.append((issue, status))
148 elif status == "deferred":
149 if include_deferred:
150 results.append((issue, status))
151 elif include_open:
152 # Unknown status: treat as open
153 results.append((issue, status))
154 except Exception:
155 continue
157 return results
160_STRATEGY_SORT_KEYS: dict[str, tuple[tuple[str, str], ...]] = {
161 "confidence_first": (
162 ("outcome_confidence", "desc"),
163 ("confidence_score", "desc"),
164 ("priority", "asc"),
165 ),
166 "priority_first": (
167 ("priority", "asc"),
168 ("outcome_confidence", "desc"),
169 ("confidence_score", "desc"),
170 ),
171}
174def build_sort_key(config: NextIssueConfig) -> Callable[[IssueInfo], tuple]:
175 """Return a sort-key callable for an IssueInfo list based on NextIssueConfig.
177 Resolution order:
178 - If ``config.sort_keys`` is set, it takes precedence over ``config.strategy``.
179 - Otherwise, the named strategy preset is used.
181 The returned callable produces a multi-field tuple where each component is
182 transformed per-field so that a plain ``sorted(..., reverse=False)`` call
183 yields the desired order. Do NOT pass ``reverse=True`` — mixed directions
184 are baked into each tuple component.
186 None-handling (per-field sentinel):
187 - ``direction="desc"`` (higher = better): component = ``-value`` when set, ``1`` when None.
188 This mirrors ``-(value or -1)`` — values > 0 sort first, None sorts after 0.
189 - ``direction="asc"`` (lower = better): component = ``value`` when set, ``9999`` when None.
191 The schema key ``"priority"`` maps to ``IssueInfo.priority_int``; all other
192 keys are read directly off the dataclass.
194 Raises:
195 ValueError: If ``config.strategy`` is unknown (``from_dict`` normally
196 guards this, but direct instantiation can bypass validation).
197 """
198 if config.sort_keys is not None:
199 entries: tuple[tuple[str, str], ...] = tuple(
200 (sk.key, sk.direction) for sk in config.sort_keys
201 )
202 else:
203 preset = _STRATEGY_SORT_KEYS.get(config.strategy)
204 if preset is None:
205 raise ValueError(f"Unknown strategy: {config.strategy!r}")
206 entries = preset
208 def key(issue: IssueInfo) -> tuple:
209 parts: list[int] = []
210 for field_name, direction in entries:
211 attr = "priority_int" if field_name == "priority" else field_name
212 value = getattr(issue, attr)
213 if direction == "desc":
214 parts.append(-value if value is not None else 1)
215 else:
216 parts.append(value if value is not None else 9999)
217 return tuple(parts)
219 return key
222def _sort_issues(
223 items: list[tuple],
224 sort_field: str,
225 descending: bool,
226) -> list[tuple]:
227 """Sort issues by the requested field."""
229 def key(item: tuple) -> tuple:
230 issue, _status, disc_date, comp_date, *_rest = item
231 if sort_field == "priority":
232 return (issue.priority_int, issue.issue_id)
233 if sort_field == "id":
234 m = re.search(r"-(\d+)$", issue.issue_id)
235 num = int(m.group(1)) if m else 0
236 return (issue.issue_id.split("-", 1)[0], num)
237 if sort_field in ("date", "created"):
238 sentinel = datetime.min if descending else datetime.max
239 return (disc_date if disc_date is not None else sentinel,)
240 if sort_field == "completed":
241 sentinel_d = date.min if descending else date.max
242 return (comp_date if comp_date is not None else sentinel_d,)
243 if sort_field == "type":
244 return (issue.issue_id.split("-", 1)[0], issue.priority_int, issue.issue_id)
245 if sort_field == "title":
246 return (issue.title.lower(),)
247 if sort_field == "confidence":
248 score = (
249 issue.confidence_score
250 if issue.confidence_score is not None
251 else (-1 if descending else 9999)
252 )
253 return (score,)
254 if sort_field == "outcome":
255 score = (
256 issue.outcome_confidence
257 if issue.outcome_confidence is not None
258 else (-1 if descending else 9999)
259 )
260 return (score,)
261 if sort_field == "refinement":
262 refinement_commands = {
263 "/ll:verify-issues",
264 "/ll:refine-issue",
265 "/ll:tradeoff-review-issues",
266 "/ll:map-dependencies",
267 "/ll:ready-issue",
268 }
269 counts: dict[str, int] = getattr(issue, "session_command_counts", {}) or {}
270 total = sum(counts.get(cmd, 0) for cmd in refinement_commands)
271 return (total,)
272 return (issue.priority_int, issue.issue_id)
274 return sorted(items, key=key, reverse=descending)
277def cmd_search(config: BRConfig, args: argparse.Namespace) -> int:
278 """Search issues with optional text query, filters, and sorting.
280 Args:
281 config: Project configuration
282 args: Parsed arguments
284 Returns:
285 Exit code (0 = success)
286 """
287 # Resolve status flags
288 status = getattr(args, "status", "open")
289 if getattr(args, "include_completed", False):
290 status = "all"
292 include_open = status in ("open", "in_progress", "blocked", "all")
293 include_done = status in ("done", "cancelled", "all")
294 include_deferred = status in ("deferred", "all")
296 # Load issues
297 raw = _load_issues_with_status(config, include_open, include_done, include_deferred)
299 # Parse additional metadata (dates, labels) only when needed
300 query: str | None = getattr(args, "query", None)
301 since_date: date | None = None
302 until_date: date | None = None
303 raw_since = getattr(args, "since", None)
304 raw_until = getattr(args, "until", None)
305 if raw_since:
306 try:
307 since_date = date.fromisoformat(raw_since)
308 except ValueError:
309 print(f"Invalid --since date: {raw_since!r}. Use YYYY-MM-DD.")
310 return 1
311 if raw_until:
312 try:
313 until_date = date.fromisoformat(raw_until)
314 except ValueError:
315 print(f"Invalid --until date: {raw_until!r}. Use YYYY-MM-DD.")
316 return 1
318 sort_field = getattr(args, "sort", "priority") or "priority"
319 need_content = bool(
320 query
321 or since_date
322 or until_date
323 or getattr(args, "label", None)
324 or sort_field in {"date", "created", "completed"}
325 or getattr(args, "date_field", "discovered") == "updated"
326 )
328 # Build enriched list: (IssueInfo, status, discovered_date, completed_date, labels)
329 enriched: list[tuple[IssueInfo, str, datetime | None, date | None, list[str]]] = []
330 for issue, stat in raw:
331 if need_content:
332 try:
333 content = issue.path.read_text(encoding="utf-8")
334 except Exception:
335 content = ""
336 disc_date = _parse_discovered_date(content, issue.path)
337 labels = _parse_labels_from_content(content)
338 else:
339 content = ""
340 disc_date = None
341 labels = []
343 # --- Filter: text query ---
344 if query:
345 haystack = (issue.title + "\n" + content).lower()
346 if query.lower() not in haystack:
347 continue
349 # --- Filter: type ---
350 type_filters: list[str] = getattr(args, "type", None) or []
351 if type_filters:
352 issue_type = issue.issue_id.split("-", 1)[0]
353 if issue_type not in type_filters:
354 continue
356 # --- Filter: priority ---
357 priority_filters: list[str] = getattr(args, "priority", None) or []
358 if priority_filters:
359 allowed = _parse_priority_filter(priority_filters)
360 if issue.priority not in allowed:
361 continue
363 # --- Filter: label ---
364 label_filters: list[str] = getattr(args, "label", None) or []
365 if label_filters:
366 if not any(lf.lower() in labels for lf in label_filters):
367 continue
369 # --- Filter: date range ---
370 date_field = getattr(args, "date_field", "discovered")
371 ref_date: date | None
372 if date_field == "discovered":
373 # disc_date is datetime for sort precision; normalize to date for day-granular filters
374 ref_date = disc_date.date() if disc_date is not None else None
375 else:
376 ref_date = _parse_updated_date(content, issue.path)
378 if since_date and (ref_date is None or ref_date < since_date):
379 continue
380 if until_date and (ref_date is None or ref_date > until_date):
381 continue
383 comp_date: date | None = None
384 if sort_field == "completed" and need_content:
385 from little_loops.issue_history.parsing import _parse_completion_date
387 comp_date = _parse_completion_date(content, issue.path)
388 enriched.append((issue, stat, disc_date, comp_date, labels))
390 # --- Sort ---
391 # Default direction: desc for date/created/completed (newest first), asc for everything else
392 if getattr(args, "desc", False):
393 descending = True
394 elif getattr(args, "asc", False):
395 descending = False
396 else:
397 descending = sort_field in {"date", "created", "completed"}
399 enriched = _sort_issues(enriched, sort_field, descending)
401 # --- Limit ---
402 limit = getattr(args, "limit", None)
403 if limit and limit > 0:
404 enriched = enriched[:limit]
406 if not enriched:
407 print("No issues found.")
408 return 0
410 # --- Output ---
411 issues_out = [item[0] for item in enriched]
412 statuses_out = [item[1] for item in enriched]
413 dates_out = [item[2] for item in enriched]
414 labels_out = [item[4] for item in enriched]
416 if getattr(args, "json", False):
417 print_json(
418 [
419 {
420 "id": issue.issue_id,
421 "priority": issue.priority,
422 "type": issue.issue_id.split("-", 1)[0],
423 "title": issue.title,
424 "path": str(issue.path),
425 "status": stat,
426 "discovered_date": d.date().isoformat() if d else None,
427 "parent": issue.parent,
428 "labels": lbls,
429 }
430 for issue, stat, d, lbls in zip(
431 issues_out, statuses_out, dates_out, labels_out, strict=True
432 )
433 ]
434 )
435 return 0
437 fmt = getattr(args, "format", "table") or "table"
439 if fmt == "ids":
440 for issue in issues_out:
441 print(issue.issue_id)
442 return 0
444 if fmt == "list":
445 for issue in issues_out:
446 print(f"{issue.path.name} {issue.title}")
447 return 0
449 # Default: table (grouped by type, similar to ll-issues list)
450 buckets: dict[str, list[tuple[IssueInfo, str]]] = {"BUG": [], "FEAT": [], "ENH": [], "EPIC": []}
451 for issue, stat in zip(issues_out, statuses_out, strict=True):
452 prefix = issue.issue_id.split("-", 1)[0]
453 if prefix in buckets:
454 buckets[prefix].append((issue, stat))
456 type_labels = {"BUG": "Bugs", "FEAT": "Features", "ENH": "Enhancements", "EPIC": "Epics"}
457 lines: list[str] = []
458 for prefix, label in type_labels.items():
459 group = buckets[prefix]
460 if not group:
461 continue
462 header = colorize(f"{label} ({len(group)})", f"{TYPE_COLOR.get(prefix, '0')};1")
463 lines.append(header)
464 for issue, stat in group:
465 issue_type = issue.issue_id.split("-", 1)[0]
466 colored_id = colorize(issue.issue_id, TYPE_COLOR.get(issue_type, "0"))
467 colored_priority = colorize(issue.priority, PRIORITY_COLOR.get(issue.priority, "0"))
468 status_tag = f" [{stat}]" if stat not in ("open", "in_progress") else ""
469 lines.append(f" {colored_priority} {colored_id} {issue.title}{status_tag}")
470 lines.append("")
471 lines.append(f"Total: {len(issues_out)} issue(s) found")
472 print("\n".join(lines))
473 return 0