Coverage for little_loops/cli/verify

1"""ll-verify-triggers: Validate skill description trigger accuracy.

3Empirically validates whether each skill's ``description`` field fires correctly:

4it should trigger on a set of realistic should-fire phrasings and stay silent on

5a set of near-miss should-NOT-fire phrasings. Reports per-skill precision/recall

6and a cross-skill collision matrix. Exits non-zero when any skill falls below

7threshold or collides with another skill.

8"""

10from __future__ import annotations

12import argparse

13import json

14import re

15import sys

16from collections import defaultdict

17from dataclasses import dataclass, field

18from pathlib import Path

20from little_loops.session_store import DEFAULT_DB_PATH, cli_event_context

22# ---------------------------------------------------------------------------

23# Data types

24# ---------------------------------------------------------------------------

26STOPWORDS: frozenset[str] = frozenset(

27 {

28 "the",

29 "a",

30 "an",

31 "is",

32 "are",

33 "was",

34 "were",

35 "be",

36 "been",

37 "being",

38 "have",

39 "has",

40 "had",

41 "do",

42 "does",

43 "did",

44 "will",

45 "would",

46 "could",

47 "should",

48 "may",

49 "might",

50 "can",

51 "shall",

52 "to",

53 "of",

54 "in",

55 "for",

56 "on",

57 "with",

58 "at",

59 "by",

60 "from",

61 "as",

62 "or",

63 "and",

64 "not",

65 "no",

66 "but",

67 "if",

68 "then",

69 "else",

70 "when",

71 "where",

72 "how",

73 "all",

74 "each",

75 "every",

76 "both",

77 "few",

78 "more",

79 "most",

80 "other",

81 "some",

82 "such",

83 "only",

84 "own",

85 "same",

86 "so",

87 "than",

88 "too",

89 "very",

90 "just",

91 "about",

92 "above",

93 "after",

94 "again",

95 "against",

96 "any",

97 "because",

98 "before",

99 "between",

100 "into",

101 "through",

102 "during",

103 "under",

104 "over",

105 "its",

106 "it",

107 "this",

108 "that",

109 "these",

110 "those",

111 "use",

112 "used",

113 "using",

114 "also",

115 "get",

116 "got",

117 "make",

118 "made",

119 "see",

120 }

121)

122

123_MIN_WORD_LEN = 3

124

125

126@dataclass

127class TriggerFixtures:

128 """Parsed trigger fixture data from a skill's frontmatter."""

129

130 should_fire: list[str]

131 should_not_fire: list[str]

132

133

134@dataclass

135class PrecisionRecall:

136 """Precision and recall metrics for a single skill."""

137

138 precision: float

139 recall: float

140

141

142@dataclass

143class SkillTriggerResult:

144 """Per-skill validation result."""

145

146 skill_name: str

147 description: str

148 tp: int = 0

149 fp: int = 0

150 fn: int = 0

151 precision: float = 0.0

152 recall: float = 0.0

153 false_positive_phrasings: list[str] = field(default_factory=list)

154 false_negative_phrasings: list[str] = field(default_factory=list)

155

156

157# Naming aliases for test compatibility

158PhrasingFixture = TriggerFixtures

159

160

161# ---------------------------------------------------------------------------

162# Keyword extraction

163# ---------------------------------------------------------------------------

164

165

166def _extract_keywords(description: str) -> set[str]:

167 """Extract matchable keywords from a skill description.

168

169 Looks for an explicit ``Trigger keywords:`` line first; falls back to

170 extracting all substantive words from the description body, filtering

171 stopwords and short tokens.

172 """

173 for line in description.splitlines():

174 stripped = line.strip()

175 if stripped.lower().startswith("trigger keywords"):

176 # Everything after the colon

177 if ":" in stripped:

178 trigger_text = stripped.split(":", 1)[1].strip()

179 else:

180 trigger_text = ""

181 return _tokenize(trigger_text)

182

183 # Fallback: tokenize the full description

184 return _tokenize(description)

185

186

187def _tokenize(text: str) -> set[str]:

188 """Tokenize text into a set of normalized keywords."""

189 tokens: set[str] = set()

190 for word in re.findall(r"[a-zA-Z][a-zA-Z0-9_-]*", text.lower()):

191 if len(word) >= _MIN_WORD_LEN and word not in STOPWORDS:

192 tokens.add(word)

193 return tokens

194

195

196# ---------------------------------------------------------------------------

197# Matching

198# ---------------------------------------------------------------------------

199

200

201def _match_phrasing(phrasing: str, keywords: set[str]) -> bool:

202 """Return True if *phrasing* contains at least one keyword token."""

203 phrasing_tokens = _tokenize(phrasing)

204 return bool(phrasing_tokens & keywords)

205

206

207def _match_score(phrasing: str, keywords: set[str]) -> int:

208 """Number of keyword tokens found in the phrasing."""

209 phrasing_tokens = _tokenize(phrasing)

210 return len(phrasing_tokens & keywords)

211

212

213# ---------------------------------------------------------------------------

214# Fixture loading

215# ---------------------------------------------------------------------------

216

217

218def _load_trigger_fixtures(skill_md_path: Path) -> TriggerFixtures | None:

219 """Parse trigger_fixtures from a SKILL.md frontmatter block.

220

221 Returns None if the file has no trigger_fixtures or empty lists.

222 """

223 try:

224 text = skill_md_path.read_text()

225 except OSError:

226 return None

227

228 if not text.startswith("---"):

229 return None

230

231 end = text.find("---", 3)

232 if end == -1:

233 return None

234

235 # Quick substring check before attempting YAML parse

236 fm_text = text[3:end]

237 if "trigger_fixtures" not in fm_text:

238 return None

239

240 import yaml

241

242 try:

243 fm = yaml.safe_load(fm_text)

244 except yaml.YAMLError:

245 return None

246

247 if not isinstance(fm, dict):

248 return None

249

250 fixtures = fm.get("trigger_fixtures")

251 if not isinstance(fixtures, dict):

252 return None

253

254 should_fire = fixtures.get("should_fire", [])

255 should_not_fire = fixtures.get("should_not_fire", [])

256

257 if not isinstance(should_fire, list) or not isinstance(should_not_fire, list):

258 return None

259

260 if not should_fire and not should_not_fire:

261 return None

262

263 return TriggerFixtures(

264 should_fire=[str(s) for s in should_fire],

265 should_not_fire=[str(s) for s in should_not_fire],

266 )

267

268

269# ---------------------------------------------------------------------------

270# Skill description loading

271# ---------------------------------------------------------------------------

272

273

274def _load_skill_descriptions(skills_dir: Path) -> dict[str, tuple[str, Path]]:

275 """Load (description, path) for all SKILL.md files under *skills_dir*.

276

277 Returns a dict mapping skill name → (description, path).

278 Skills without frontmatter or a description are skipped.

279 """

280 skills: dict[str, tuple[str, Path]] = {}

281

282 if not skills_dir.is_dir():

283 return skills

284

285 for skill_md in sorted(skills_dir.glob("*/SKILL.md")):

286 try:

287 text = skill_md.read_text()

288 except OSError:

289 continue

290

291 if not text.startswith("---"):

292 continue

293

294 end = text.find("---", 3)

295 if end == -1:

296 continue

297

298 fm_text = text[3:end]

299

300 import yaml

301

302 try:

303 fm = yaml.safe_load(fm_text)

304 except yaml.YAMLError:

305 continue

306

307 if not isinstance(fm, dict):

308 continue

309

310 name = fm.get("name") or skill_md.parent.name

311 desc = fm.get("description", "")

312 if desc:

313 skills[name] = (str(desc), skill_md)

314

315 return skills

316

317

318# ---------------------------------------------------------------------------

319# Metrics computation

320# ---------------------------------------------------------------------------

321

322

323def _compute_precision_recall(tp: int, fp: int, fn: int) -> PrecisionRecall:

324 """Return (precision, recall) from confusion-matrix counts."""

325 precision = tp / (tp + fp) if (tp + fp) > 0 else 0.0

326 recall = tp / (tp + fn) if (tp + fn) > 0 else 0.0

327 return PrecisionRecall(precision=precision, recall=recall)

328

329

330def _detect_collisions(

331 results: dict[str, SkillTriggerResult],

332 phrasing_matches: dict[str, set[str]],

333) -> list[dict]:

334 """Detect cross-skill collisions from phrase-level match data.

335

336 A collision occurs when a single phrasing triggered more than one skill.

337 """

338 collisions: list[dict] = []

339 for phrasing, matched_skills in sorted(phrasing_matches.items()):

340 if len(matched_skills) > 1:

341 collisions.append(

342 {

343 "phrasing": phrasing,

344 "skills": sorted(matched_skills),

345 }

346 )

347 return collisions

348

349

350# ---------------------------------------------------------------------------

351# Core validation

352# ---------------------------------------------------------------------------

353

354

355def _run_validation(

356 skills_dir: Path,

357 precision_threshold: float = 0.5,

358 recall_threshold: float = 0.5,

359) -> tuple[dict[str, SkillTriggerResult], list[dict], dict]:

360 """Run full trigger validation across all skills.

361

362 Returns (results_by_skill, collisions, thresholds_dict).

363 """

364 # 1. Load skill descriptions and extract keywords

365 skill_descs = _load_skill_descriptions(skills_dir)

366 skill_keywords: dict[str, set[str]] = {

367 name: _extract_keywords(desc) for name, (desc, _) in skill_descs.items()

368 }

369

370 # 2. Load trigger fixtures per skill

371 skill_fixtures: dict[str, TriggerFixtures] = {}

372 for name, (_, path) in skill_descs.items():

373 fixtures = _load_trigger_fixtures(path)

374 if fixtures is not None:

375 skill_fixtures[name] = fixtures

376

377 # 3. Evaluate each skill with fixtures

378 results: dict[str, SkillTriggerResult] = {}

379 phrasing_matches: dict[str, set[str]] = defaultdict(set)

380

381 for skill_name, (desc, _) in skill_descs.items():

382 fixtures = skill_fixtures.get(skill_name)

383 if fixtures is None:

384 results[skill_name] = SkillTriggerResult(

385 skill_name=skill_name,

386 description=desc,

387 )

388 continue

389

390 tp = 0

391 fp = 0

392 fn = 0

393 fp_phrasings: list[str] = []

394 fn_phrasings: list[str] = []

395

396 # Evaluate should_fire phrasings

397 for phrasing in fixtures.should_fire:

398 # Check which skills this phrasing matches

399 matched: list[str] = []

400 for other_name, other_kw in skill_keywords.items():

401 if _match_phrasing(phrasing, other_kw):

402 matched.append(other_name)

403

404 if matched:

405 phrasing_matches[phrasing] |= set(matched)

406

407 if skill_name in matched:

408 tp += 1

409 else:

410 fn += 1

411 fn_phrasings.append(phrasing)

412

413 # Evaluate should_not_fire phrasings

414 for phrasing in fixtures.should_not_fire:

415 matched_not: list[str] = []

416 for other_name, other_kw in skill_keywords.items():

417 if _match_phrasing(phrasing, other_kw):

418 matched_not.append(other_name)

419

420 if matched_not:

421 phrasing_matches[phrasing] |= set(matched_not)

422

423 if skill_name in matched_not:

424 fp += 1

425 fp_phrasings.append(phrasing)

426

427 pr = _compute_precision_recall(tp, fp, fn)

428 results[skill_name] = SkillTriggerResult(

429 skill_name=skill_name,

430 description=desc,

431 tp=tp,

432 fp=fp,

433 fn=fn,

434 precision=pr.precision,

435 recall=pr.recall,

436 false_positive_phrasings=fp_phrasings,

437 false_negative_phrasings=fn_phrasings,

438 )

439

440 # 4. Detect collisions

441 collisions = _detect_collisions(results, dict(phrasing_matches))

442

443 thresholds = {

444 "precision_threshold": precision_threshold,

445 "recall_threshold": recall_threshold,

446 }

447

448 return results, collisions, thresholds

449

450

451# ---------------------------------------------------------------------------

452# Formatting

453# ---------------------------------------------------------------------------

454

455

456def _format_text_report(

457 results: dict[str, SkillTriggerResult],

458 collisions: list[dict],

459 thresholds: dict,

460) -> str:

461 """Format a human-readable text report."""

462 lines: list[str] = []

463 lines.append("Skill Trigger Validation Report")

464 lines.append("=" * 50)

465 lines.append(

466 f"Thresholds: precision ≥ {thresholds['precision_threshold']:.0%}, "

467 f"recall ≥ {thresholds['recall_threshold']:.0%}"

468 )

469 lines.append("")

470

471 # Per-skill table

472 if results:

473 lines.append(f"{'Skill':<40} {'Precision':>10} {'Recall':>10}")

474 lines.append(f"{'':-<40} {'':->10} {'':->10}")

475 for name in sorted(results):

476 r = results[name]

477 lines.append(f"{name:<40} {r.precision:>10.0%} {r.recall:>10.0%}")

478 lines.append("")

479

480 # Collisions

481 if collisions:

482 lines.append("Cross-Skill Collisions")

483 lines.append("=" * 50)

484 for c in collisions:

485 skills = ", ".join(c["skills"])

486 lines.append(f' Phrasing: "{c["phrasing"]}"')

487 lines.append(f" Colliding skills: {skills}")

488 lines.append("")

489 else:

490 lines.append("No cross-skill collisions detected.")

491 lines.append("")

492

493 # Failures

494 failures: list[str] = []

495 for name in sorted(results):

496 r = results[name]

497 if r.precision < thresholds["precision_threshold"]:

498 failures.append(

499 f" {name}: precision {r.precision:.0%} < {thresholds['precision_threshold']:.0%}"

500 )

501 if r.recall < thresholds["recall_threshold"]:

502 failures.append(

503 f" {name}: recall {r.recall:.0%} < {thresholds['recall_threshold']:.0%}"

504 )

505

506 if collisions:

507 failures.append(" Collisions detected")

508 if failures:

509 lines.append("FAILURES")

510 lines.append("=" * 50)

511 for f in failures:

512 lines.append(f)

513

514 return "\n".join(lines)

515

516

517def _format_json_report(

518 results: dict[str, SkillTriggerResult],

519 collisions: list[dict],

520 thresholds: dict,

521) -> str:

522 """Format results as JSON."""

523 skills_list = []

524 for name in sorted(results):

525 r = results[name]

526 skills_list.append(

527 {

528 "name": r.skill_name,

529 "description": r.description,

530 "precision": r.precision,

531 "recall": r.recall,

532 "tp": r.tp,

533 "fp": r.fp,

534 "fn": r.fn,

535 "false_positive_phrasings": r.false_positive_phrasings,

536 "false_negative_phrasings": r.false_negative_phrasings,

537 }

538 )

539

540 return json.dumps(

541 {

542 "thresholds": thresholds,

543 "skills": skills_list,

544 "collisions": collisions,

545 },

546 indent=2,

547 )

548

549

550# ---------------------------------------------------------------------------

551# Orchestration

552# ---------------------------------------------------------------------------

553

554

555def _any_failures(

556 results: dict[str, SkillTriggerResult],

557 collisions: list[dict],

558 precision_threshold: float,

559 recall_threshold: float,

560) -> bool:

561 """True when any skill is below threshold or collisions exist."""

562 if collisions:

563 return True

564 for r in results.values():

565 if r.precision < precision_threshold:

566 return True

567 if r.recall < recall_threshold:

568 return True

569 return False

570

571

572# ---------------------------------------------------------------------------

573# Entry point

574# ---------------------------------------------------------------------------

575

576

577def main_verify_triggers() -> int:

578 """Entry point for ll-verify-triggers CLI.

579

580 Returns 0 when all skills meet threshold and there are no collisions;

581 returns 1 otherwise.

582 """

583 with cli_event_context(DEFAULT_DB_PATH, "ll-verify-triggers", sys.argv[1:]):

584 parser = argparse.ArgumentParser(

585 prog="ll-verify-triggers",

586 description=(

587 "Validate skill description trigger accuracy against should-fire "

588 "and should-NOT-fire phrasings. Reports per-skill precision/recall "

589 "and cross-skill collision matrix."

590 ),

591 formatter_class=argparse.RawDescriptionHelpFormatter,

592 epilog="""\

593Examples:

594 %(prog)s # Validate all skills against default thresholds

595 %(prog)s --json # Machine-readable JSON output

596 %(prog)s --precision-threshold 0.8 --recall-threshold 0.6

597

598Exit codes:

599 0 - All skills meet thresholds, no collisions

600 1 - One or more skills below threshold, or collisions detected

601""",

602 )

603 parser.add_argument(

604 "-C",

605 "--directory",

606 type=Path,

607 default=None,

608 help="Base directory (default: current directory)",

609 )

610 parser.add_argument(

611 "--json",

612 action="store_true",

613 default=False,

614 help="Output results as JSON",

615 )

616 parser.add_argument(

617 "--precision-threshold",

618 type=float,

619 default=0.5,

620 metavar="F",

621 help="Minimum precision required (default: 0.5)",

622 )

623 parser.add_argument(

624 "--recall-threshold",

625 type=float,

626 default=0.5,

627 metavar="F",

628 help="Minimum recall required (default: 0.5)",

629 )

630

631 args = parser.parse_args()

632

633 base_dir = args.directory or Path.cwd()

634 skills_dir = base_dir / "skills"

635

636 if not skills_dir.exists():

637 print(

638 f"ERROR: skills directory not found: {skills_dir}",

639 file=sys.stderr,

640 )

641 return 1

642

643 results, collisions, thresholds = _run_validation(

644 skills_dir,

645 precision_threshold=args.precision_threshold,

646 recall_threshold=args.recall_threshold,

647 )

648

649 if args.json:

650 print(_format_json_report(results, collisions, thresholds))

651 else:

652 print(_format_text_report(results, collisions, thresholds))

653

654 if _any_failures(

655 results,

656 collisions,

657 args.precision_threshold,

658 args.recall_threshold,

659 ):

660 return 1

661 return 0

Coverage for little_loops / cli / verify_triggers.py: 18%

238 statements