Coverage for little_loops / learning_tests / gate.py: 27%

1"""Shared gate utilities for learning-test staleness checks (ENH-2208). 

2 

3Exposes ``is_record_stale()`` and ``format_nudge_message()`` as standalone 

4importable helpers consumed by the discoverability gate hook, the install 

5learning gate hook (ENH-2212), and downstream sprint/release gates 

6(ENH-2209, ENH-2210, ENH-2214, ENH-2217). 

7 

8Also exposes ``run_learning_gate_for_issue()`` — the shared subprocess wrapper 

9for the ``proof-first-task`` loop used by ll-auto (ENH-2319). 

10""" 

11 

12from __future__ import annotations 

13 

14import datetime 

15import json 

16import subprocess 

17from pathlib import Path 

18from typing import TYPE_CHECKING, Literal 

19 

20if TYPE_CHECKING: 

21 from little_loops.learning_tests import LearnTestRecord 

22 

23 

24def format_nudge_message(pkg: str, stale: bool = False) -> str: 

25 """Return a nudge message for a package with no proven or stale learning test. 

26 

27 Args: 

28 pkg: The package name (already normalized). 

29 stale: If True, the record exists but is stale; otherwise it is absent. 

30 """ 

31 if stale: 

32 body = f'Learning test for "{pkg}" is stale.' 

33 else: 

34 body = f'No learning test for "{pkg}".' 

35 return f'[ll: new dependency] {body} Consider: /ll:explore-api "{pkg}"' 

36 

37 

38def is_record_stale(record: LearnTestRecord, stale_after_days: int) -> bool: 

39 """Return True if the record's proof date exceeds the staleness threshold. 

40 

41 Args: 

42 record: A LearnTestRecord with a ``date`` field (ISO 8601: YYYY-MM-DD). 

43 stale_after_days: Age threshold in days. Clamped to minimum 1 to 

44 prevent ``stale_after_days=0`` from being a footgun that passes 

45 all records whose date is exactly today. 

46 

47 Returns: 

48 True if age in days exceeds the threshold; False if fresh or unparseable. 

49 """ 

50 threshold = max(1, stale_after_days) 

51 try: 

52 record_date = datetime.date.fromisoformat(record.date) 

53 except (ValueError, TypeError, AttributeError): 

54 return False 

55 age_days = (datetime.date.today() - record_date).days 

56 return age_days > threshold 

57 

58 

59def _read_loop_final_state(cwd: Path, loop_name: str) -> str | None: 

60 """Read the terminal state from a completed loop's running state file. 

61 

62 After ``ll-loop run`` exits, the state file remains at 

63 ``<cwd>/.loops/.running/<loop_name>.state.json``. The ``current_state`` 

64 field holds the terminal state name (e.g. ``"done"``, ``"blocked"``). 

65 """ 

66 state_file = cwd / ".loops" / ".running" / f"{loop_name}.state.json" 

67 if not state_file.exists(): 

68 return None 

69 try: 

70 data = json.loads(state_file.read_text()) 

71 return data.get("current_state") 

72 except (json.JSONDecodeError, OSError): 

73 return None 

74 

75 

76def run_learning_gate_for_issue( 

77 issue_path: Path, 

78 *, 

79 skip: bool = False, 

80 cwd: Path | None = None, 

81) -> Literal["passed", "blocked", "skipped"]: 

82 """Invoke proof-first-task loop for an issue and return the gate verdict. 

83 

84 ``skip=True`` short-circuits to "skipped" (honours --skip-learning-gate). 

85 All terminal states (done, blocked, impl_failed) exit 0 — blocked is 

86 distinguished from done by reading the loop state file left after execution. 

87 

88 Args: 

89 issue_path: Absolute path to the issue file. 

90 skip: If True, return "skipped" immediately without running the loop. 

91 cwd: Working directory for the subprocess (and state-file lookup). 

92 Defaults to ``Path.cwd()`` when None. 

93 """ 

94 if skip: 

95 return "skipped" 

96 

97 working_dir = cwd or Path.cwd() 

98 subprocess.run( 

99 [ 

100 "ll-loop", 

101 "run", 

102 "proof-first-task", 

103 "--context", 

104 f"issue_file={issue_path}", 

105 ], 

106 capture_output=True, 

107 text=True, 

108 cwd=working_dir, 

109 ) 

110 

111 final_state = _read_loop_final_state(working_dir, "proof-first-task") 

112 if final_state == "blocked": 

113 return "blocked" 

114 return "passed"