personal memory agent
Merge pull request #128 from kognova/codex/add-utility-to-validate-crumbs
Add crumb validation utility
authored by
Jer Miller
and committed by
GitHub
c7a807ae
a53be6b3
+67
+19
tests/test_crumbs.py
+19
tests/test_crumbs.py
···
1
1
import importlib
2
+
import os
2
3
from pathlib import Path
3
4
4
5
···
15
16
assert Path(crumb_path).is_file()
16
17
data = Path(crumb_path).read_text()
17
18
assert "generator" in data
19
+
20
+
21
+
def test_validate_crumb(tmp_path):
22
+
mod = importlib.import_module("think.crumbs")
23
+
src = tmp_path / "dep.txt"
24
+
src.write_text("x")
25
+
out = tmp_path / "out.txt"
26
+
out.write_text("o")
27
+
mod.CrumbBuilder(generator="test").add_file(src).commit(str(out))
28
+
29
+
assert mod.validate_crumb(str(out)) is mod.CrumbState.OK
30
+
31
+
src.write_text("changed")
32
+
os.utime(src, (os.path.getmtime(src) + 1, os.path.getmtime(src) + 1))
33
+
assert mod.validate_crumb(str(out)) is mod.CrumbState.STALE
34
+
35
+
out.with_suffix(out.suffix + ".crumb").unlink()
36
+
assert mod.validate_crumb(str(out)) is mod.CrumbState.MISSING
+48
think/crumbs.py
+48
think/crumbs.py
···
7
7
import os
8
8
import sys
9
9
from datetime import datetime, timezone
10
+
from enum import Enum
10
11
from pathlib import Path
11
12
from typing import Any, Dict, Iterable, List
12
13
···
56
57
with open(crumb_path, "w", encoding="utf-8") as f:
57
58
json.dump(crumb, f, indent=2)
58
59
return crumb_path
60
+
61
+
62
+
class CrumbState(str, Enum):
63
+
"""Result of :func:`validate_crumb`."""
64
+
65
+
MISSING = "missing"
66
+
STALE = "stale"
67
+
OK = "ok"
68
+
69
+
70
+
def validate_crumb(output: str | Path) -> CrumbState:
71
+
"""Return status for ``output`` based on its ``.crumb`` file."""
72
+
73
+
output_path = Path(output)
74
+
crumb_path = Path(str(output) + ".crumb")
75
+
76
+
if not output_path.exists() or not crumb_path.exists():
77
+
return CrumbState.MISSING
78
+
79
+
try:
80
+
with open(crumb_path, "r", encoding="utf-8") as f:
81
+
data = json.load(f)
82
+
except Exception:
83
+
return CrumbState.STALE
84
+
85
+
for dep in data.get("dependencies", []):
86
+
dep_type = dep.get("type")
87
+
if dep_type == "file":
88
+
path = dep.get("path")
89
+
mtime = dep.get("mtime")
90
+
if not path or not os.path.exists(path) or int(os.path.getmtime(path)) != mtime:
91
+
return CrumbState.STALE
92
+
elif dep_type == "glob":
93
+
pattern = dep.get("pattern", "")
94
+
recorded = dep.get("files", {})
95
+
matches = glob.glob(pattern)
96
+
if set(matches) != set(recorded):
97
+
return CrumbState.STALE
98
+
for m in matches:
99
+
if int(os.path.getmtime(m)) != recorded[m]:
100
+
return CrumbState.STALE
101
+
elif dep_type == "model":
102
+
continue
103
+
else:
104
+
return CrumbState.STALE
105
+
106
+
return CrumbState.OK