Coverage for src/keel/workspace.py: 100%

1"""keel runtime workspace: the ``.keel`` directory, its gitignore, and scratch. 

2 

3Keel writes runtime artifacts (checkpoints, activity records, the run ledger, 

4merge locks) and its agentic steps stage transient scratch (PR diffs, issue 

5dumps, draft review/closure prose) while driving a workflow. None of this 

6belongs in the consumer's primary checkout: it is keel-owned, disposable 

7runtime state. This module owns the one place those artifacts live - the 

8project's ``.keel`` directory - and keeps them out of ``git status`` by 

9scaffolding a ``.keel/.gitignore`` that ignores the runtime subtrees while 

10leaving committed config (``project.yaml``) and extensions tracked. 

11 

12It also exposes the **scratch directory** (``.keel/scratch``): the sanctioned 

13place for any ad-hoc transient file an agent needs to stage. Routing scratch 

14here - instead of the repo root - is why a consumer no longer sees 

15``plan.json``, ``pr_<n>.diff``, ``issue.md`` and friends accumulate in their 

16checkout. 

17 

18Finally it owns **reclamation** of the disposable runtime trees so they do not 

19grow without bound: :func:`clean_scratch` empties ``.keel/scratch`` and 

20:func:`prune_activity` applies count-based retention to ``.keel/activity``. 

21Reclamation is deliberately scoped to those two trees - it never touches the 

22run ledger, checkpoint, or locks, which have their own bounded lifecycles and 

23(for the ledger) are durable by design. 

24 

25Pure-core + thin I/O, mirroring :mod:`keel.checkpoint` and :mod:`keel.activity`: 

26:func:`runtime_gitignore_body` is a deterministic pure function; only the 

27``ensure_*``, :func:`scratch_dir`, and the reclamation helpers touch the 

28filesystem. 

29""" 

30 

31from __future__ import annotations 

32 

33import shutil 

34from pathlib import Path 

35 

36KEEL_DIRNAME = ".keel" 

37GITIGNORE_NAME = ".gitignore" 

38SCRATCH_DIRNAME = "scratch" 

39 

40# Runtime subtrees of ``.keel`` that are disposable per-run state, expressed as 

41# patterns relative to the ``.keel/.gitignore`` that lists them. ``project.yaml`` 

42# and ``extensions/`` are intentionally absent: they are committed config. 

43RUNTIME_IGNORE_ENTRIES: tuple[str, ...] = ("state/", "activity/", "scratch/", "*.tmp") 

44 

45_GITIGNORE_HEADER = ( 

46 "# keel runtime artifacts - generated by `keel`; commit this file.", 

47 "# These subtrees are disposable per-run state (checkpoints, activity,", 

48 "# locks, agent scratch); keeping them ignored stops keel from polluting", 

49 "# your checkout. Delete an entry only if you deliberately track that path.", 

50) 

51 

52 

53def keel_dir(root: str | Path = ".") -> Path: 

54 """Return the project's ``.keel`` directory under ``root`` (not created).""" 

55 return Path(root) / KEEL_DIRNAME 

56 

57 

58def runtime_gitignore_body() -> str: 

59 """Return the canonical ``.keel/.gitignore`` body (deterministic).""" 

60 lines = [*_GITIGNORE_HEADER, *RUNTIME_IGNORE_ENTRIES] 

61 return "\n".join(lines) + "\n" 

62 

63 

64def ensure_runtime_gitignore(keel_directory: str | Path) -> bool: 

65 """Scaffold or top up ``<keel_directory>/.gitignore``; idempotent. 

66 

67 Creates the gitignore with the canonical body when absent. When it already 

68 exists, appends only the runtime entries that are missing - preserving any 

69 operator additions and never rewriting an already-complete file. Returns 

70 ``True`` when the file was created or changed, ``False`` when it was already 

71 a superset of the runtime entries (or the ``.keel`` directory is absent). 

72 """ 

73 directory = Path(keel_directory) 

74 if not directory.is_dir(): 

75 return False 

76 gitignore = directory / GITIGNORE_NAME 

77 if not gitignore.exists(): 

78 gitignore.write_text(runtime_gitignore_body(), encoding="utf-8") 

79 return True 

80 existing = gitignore.read_text(encoding="utf-8") 

81 present = {line.strip() for line in existing.splitlines()} 

82 missing = [entry for entry in RUNTIME_IGNORE_ENTRIES if entry not in present] 

83 if not missing: 

84 return False 

85 prefix = existing if existing.endswith("\n") or existing == "" else existing + "\n" 

86 gitignore.write_text(prefix + "\n".join(missing) + "\n", encoding="utf-8") 

87 return True 

88 

89 

90def ensure_runtime_gitignore_for(artifact_path: str | Path) -> bool: 

91 """Self-heal the gitignore for a runtime artifact about to be written. 

92 

93 Walks the artifact's ancestors for a ``.keel`` directory and scaffolds its 

94 gitignore. A no-op (returns ``False``) when the artifact lives outside any 

95 ``.keel`` tree - e.g. an operator pointed an output path elsewhere on 

96 purpose, which keel must not silently ignore. 

97 """ 

98 target = Path(artifact_path) 

99 for ancestor in target.parents: 

100 if ancestor.name == KEEL_DIRNAME and ancestor.is_dir(): 

101 return ensure_runtime_gitignore(ancestor) 

102 return False 

103 

104 

105def scratch_dir(root: str | Path = ".", *, create: bool = True) -> Path: 

106 """Return ``.keel/scratch`` - the sanctioned home for transient artifacts. 

107 

108 With ``create`` (the default) the directory and the runtime gitignore are 

109 materialised so callers can write into it immediately and it never surfaces 

110 in ``git status``. 

111 """ 

112 directory = keel_dir(root) 

113 scratch = directory / SCRATCH_DIRNAME 

114 if create: 

115 scratch.mkdir(parents=True, exist_ok=True) 

116 ensure_runtime_gitignore(directory) 

117 return scratch 

118 

119 

120def scratch_entries(root: str | Path = ".") -> list[str]: 

121 """Sorted top-level names currently under ``.keel/scratch`` (``[]`` if none).""" 

122 scratch = keel_dir(root) / SCRATCH_DIRNAME 

123 if not scratch.is_dir(): 

124 return [] 

125 return sorted(p.name for p in scratch.iterdir()) 

126 

127 

128def clean_scratch(root: str | Path = ".") -> list[str]: 

129 """Reclaim ``.keel/scratch`` entirely; return the entries that were removed. 

130 

131 Scratch is transient by definition, so this empties it wholesale. The 

132 directory is recreated on the next :func:`scratch_dir` call. A no-op (``[]``) 

133 when scratch does not exist. 

134 """ 

135 entries = scratch_entries(root) 

136 scratch = keel_dir(root) / SCRATCH_DIRNAME 

137 if scratch.is_dir(): 

138 shutil.rmtree(scratch) 

139 return entries 

140 

141 

142def activity_prune_plan(activity_dir: str | Path, *, keep_last: int) -> list[str]: 

143 """Names of activity records that exceed ``keep_last`` (oldest first removed). 

144 

145 Retention is count-based: the newest ``keep_last`` ``.json`` records (by 

146 mtime, then name for a stable tiebreak) are kept; the rest are returned as 

147 the prune plan. Only ``.json`` files are considered, so a stray file never 

148 counts against - or gets caught by - retention. ``keep_last`` must be 

149 non-negative. A no-op (``[]``) when the directory is absent or within budget. 

150 """ 

151 if keep_last < 0: 

152 raise ValueError("keep_last must be non-negative") 

153 directory = Path(activity_dir) 

154 if not directory.is_dir(): 

155 return [] 

156 records = [p for p in directory.iterdir() if p.is_file() and p.suffix == ".json"] 

157 if len(records) <= keep_last: 

158 return [] 

159 ordered = sorted(records, key=lambda p: (p.stat().st_mtime, p.name), reverse=True) 

160 return sorted(p.name for p in ordered[keep_last:]) 

161 

162 

163def prune_activity(activity_dir: str | Path, *, keep_last: int) -> list[str]: 

164 """Apply count-based retention to ``activity_dir``; return the names removed. 

165 

166 Removes exactly the records named by :func:`activity_prune_plan`. Never 

167 touches non-``.json`` entries, and is only ever pointed at the activity 

168 directory - the run ledger, checkpoint, and locks live elsewhere and are 

169 out of reach by construction. 

170 """ 

171 directory = Path(activity_dir) 

172 doomed = activity_prune_plan(directory, keep_last=keep_last) 

173 for name in doomed: 

174 (directory / name).unlink() 

175 return doomed