tools/f0_inventory.py

1#!/usr/bin/env python3

2"""Read-only Folio estate census and F0 gate.

4The source estate is never opened for writing. All mutation/no-op probes operate on

5an explicitly supplied disposable copy. This is a baseline tool, not a replacement

6for Folio's parser: its deliberately independent raw-ID pass is the second net used by

7the F0 denominator gate.

8"""

10from __future__ import annotations

12import argparse

13import hashlib

14import json

15import os

16import re

17import shutil

18import statistics

19import sys

20import tempfile

21import time

22from collections import Counter, defaultdict

23from pathlib import Path

25TASK_RE = re.compile(r"^(?P<indent>[ \t]*)- \[(?P<done>[ xX])\] (?P<body>.*)$")

26ID_RE = re.compile(r"(?<![A-Za-z0-9_-])id\s*:\s*(t-[^\s,}]+)")

27LINK_RE = re.compile(r"\[\[([^\]|]+)(?:\|[^\]]*)?\]\]")

28META_RE = re.compile(r"\{([^{}]*)\}\s*$")

29KEY_RE = re.compile(r"(?:^|,)\s*([A-Za-z][A-Za-z0-9_-]*)\s*:")

30CANONICAL_DIRS = ("notes", "projects", "archive/notes", "archive/projects")

31GENERATED_NAMES = {".folio-index.db", ".folio-index.db-shm", ".folio-index.db-wal"}

34def sha256(data: bytes) -> str:

35 return hashlib.sha256(data).hexdigest()

38def indent_width(text: str) -> int:

39 width = 0

40 for ch in text:

41 if ch == " ":

42 width += 1

43 elif ch == "\t":

44 width += 4

45 else:

46 break

47 return width

50def comment_mask(lines: list[str]) -> list[str]:

51 joined = "\n".join(lines)

52 chars = list(joined)

53 offsets = []

54 offset = 0

55 for line in lines:

56 offsets.append(offset)

57 offset += len(line) + 1

58 line_index = 0

59 while line_index < len(lines):

60 if not lines[line_index].startswith("<!--"):

61 line_index += 1

62 continue

63 start = offsets[line_index]

64 end = joined.find("-->", start + 4)

65 if end < 0:

66 break

67 for i in range(start, end + 3):

68 if chars[i] != "\n":

69 chars[i] = " "

70 while line_index < len(offsets) and offsets[line_index] <= end:

71 line_index += 1

72 return "".join(chars).split("\n")

75def task_blocks(lines: list[str]) -> list[dict]:

76 blocks: list[dict] = []

77 i = 0

78 while i < len(lines):

79 match = TASK_RE.match(lines[i])

80 if not match:

81 i += 1

82 continue

83 base = indent_width(match.group("indent"))

84 end = i

85 j = i + 1

86 while j < len(lines):

87 if lines[j].strip():

88 if indent_width(lines[j]) < base + 2:

89 break

90 end = j

91 j += 1

92 continue

93 k = j + 1

94 while k < len(lines) and not lines[k].strip():

95 k += 1

96 if k < len(lines) and indent_width(lines[k]) >= base + 2:

97 end = k

98 j = k + 1

99 continue

100 break

101 meta_match = META_RE.search(match.group("body"))

102 metadata = meta_match.group(1) if meta_match else ""

103 id_match = ID_RE.search(metadata)

104 blocks.append({

105 "start": i + 1,

106 "end": end + 1,

107 "done": match.group("done").lower() == "x",

108 "id": id_match.group(1) if id_match else None,

109 "metadata": metadata,

110 "metadata_keys": KEY_RE.findall(metadata),

111 })

112 i = end + 1

113 return blocks

114

115

116def classify(rel: str) -> tuple[str, str]:

117 if rel == "inbox.md":

118 return "inbox", "inbox parser; task mutation owner"

119 if rel.startswith("projects/") and rel.endswith(".md") and rel.count("/") == 1:

120 return "project", "project parser; project/task mutation owner"

121 if rel.startswith("notes/") and rel.endswith(".md"):

122 return "note", "note parser; note mutation owner"

123 if rel.startswith("archive/projects/") and rel.endswith(".md"):

124 return "archived-project", "not discovered by current project parser; manual/Git owner"

125 if rel.startswith("archive/notes/") and rel.endswith(".md"):

126 return "archived-note", "not discovered by current note parser; manual/Git owner"

127 name = Path(rel).name

128 if name in GENERATED_NAMES or name.startswith(".folio-index.db-"):

129 return "generated-index", "SQLite runtime; disposable"

130 if rel == "README.md":

131 return "auxiliary", "documentation; manual/Git owner"

132 if any(part.startswith(".") for part in Path(rel).parts):

133 return "ignored-hidden", "not discovered; external/manual owner"

134 return "unclassified", "UNKNOWN"

135

136

137def yaml_fields(lines: list[str]) -> tuple[list[str], bool]:

138 if not lines or lines[0] != "---":

139 return [], False

140 try:

141 end = lines[1:].index("---") + 1

142 except ValueError:

143 return [], True

144 fields = []

145 malformed = False

146 for line in lines[1:end]:

147 if not line.strip() or line.lstrip().startswith("-") or line.startswith((" ", "\t", "#")):

148 continue

149 if ":" not in line:

150 malformed = True

151 else:

152 fields.append(line.split(":", 1)[0].strip())

153 return fields, malformed

154

155

156def scan(root: Path, omit_prefix: str | None = None) -> dict:

157 if not root.is_dir():

158 raise ValueError(f"root is not a directory: {root}")

159 all_files = sorted(p for p in root.rglob("*") if p.is_file())

160 selected = []

161 omitted = []

162 for path in all_files:

163 rel = path.relative_to(root).as_posix()

164 if omit_prefix and (rel == omit_prefix or rel.startswith(omit_prefix.rstrip("/") + "/")):

165 omitted.append(rel)

166 else:

167 selected.append(path)

168

169 manifest = []

170 counts = Counter()

171 status_counts = Counter()

172 metadata_keys = Counter()

173 yaml_keys = Counter()

174 tag_counts = Counter()

175 id_locations: dict[str, list[str]] = defaultdict(list)

176 raw_locations: dict[str, list[str]] = defaultdict(list)

177 raw_product_ids = set()

178 unreachable = []

179 accounted_raw_ids = set()

180 links = []

181 parse_failures = []

182 visible_total = 0

183 raw_total = 0

184

185 for path in selected:

186 rel = path.relative_to(root).as_posix()

187 kind, owner = classify(rel)

188 data = path.read_bytes()

189 manifest.append({"path": rel, "kind": kind, "bytes": len(data), "sha256": sha256(data), "owner": owner})

190 counts[f"files.{kind}"] += 1

191 counts[f"bytes.{kind}"] += len(data)

192 if path.suffix != ".md":

193 continue

194 try:

195 text = data.decode("utf-8")

196 except UnicodeDecodeError as exc:

197 parse_failures.append({"path": rel, "kind": "invalid-utf8", "detail": str(exc)})

198 continue

199 lines = text.splitlines()

200 fields, malformed_yaml = yaml_fields(lines)

201 yaml_keys.update(fields)

202 if malformed_yaml:

203 parse_failures.append({"path": rel, "kind": "malformed-frontmatter"})

204 blocks = task_blocks(lines)

205 local_visible_ids = {block["id"] for block in blocks if block["id"]}

206 product_task_file = kind in {"inbox", "project"}

207 if product_task_file:

208 visible_total += len(blocks)

209 counts["tasks.visible"] += len(blocks)

210 else:

211 counts["tasks.outside-current-query"] += len(blocks)

212 for block in blocks:

213 if product_task_file:

214 status_counts["done" if block["done"] else "open"] += 1

215 metadata_keys.update(block["metadata_keys"])

216 for tm in re.finditer(r"(?:^|,)\s*tags\s*:\s*([^,}]*)", block["metadata"]):

217 tag_counts.update(tm.group(1).split())

218 if block["id"]:

219 id_locations[block["id"]].append(f"{rel}:{block['start']}")

220 else:

221 counts["tasks.without-id"] += 1

222 masked = comment_mask(lines)

223 for line_no, line in enumerate(masked, 1):

224 for match in ID_RE.finditer(line):

225 raw_total += 1

226 task_id = match.group(1)

227 raw_locations[task_id].append(f"{rel}:{line_no}")

228 if product_task_file:

229 raw_product_ids.add(task_id)

230 containing = next((b for b in blocks if b["start"] <= line_no <= b["end"]), None)

231 if containing:

232 accounted_raw_ids.add(task_id)

233 if not containing and task_id not in local_visible_ids:

234 unreachable.append({"id": task_id, "location": f"{rel}:{line_no}"})

235 for match in LINK_RE.finditer(text):

236 links.append({"source": rel, "target": match.group(1)})

237

238 selected_rel = {p.relative_to(root).as_posix() for p in selected}

239 full_rel = {p.relative_to(root).as_posix() for p in all_files}

240 unaccounted = sorted(full_rel - selected_rel)

241 unclassified = [m["path"] for m in manifest if m["kind"] == "unclassified"]

242 duplicate_ids = {key: value for key, value in sorted(id_locations.items()) if len(value) > 1}

243 raw_unique = set(raw_locations)

244 visible_unique = set(id_locations)

245 integrity_agrees = raw_product_ids == visible_unique | accounted_raw_ids | {item["id"] for item in unreachable}

246 backlinks = Counter(item["target"] for item in links)

247 return {

248 "schema": "folio-f0-estate-v1",

249 "root": str(root.resolve()),

250 "manifest": manifest,

251 "counts": {

252 "files_total": len(all_files),

253 "files_scanned": len(selected),

254 "markdown_total": sum(1 for m in manifest if m["path"].endswith(".md")),

255 "visible_tasks": visible_total,

256 "visible_task_ids": sum(len(v) for v in id_locations.values()),

257 "raw_id_occurrences": raw_total,

258 "unique_visible_ids": len(visible_unique),

259 "unique_raw_ids": len(raw_unique),

260 "unique_raw_ids_in_task_sources": len(raw_product_ids),

261 "unreachable_id_occurrences": len(unreachable),

262 "links": len(links),

263 "unique_link_targets": len(backlinks),

264 **dict(sorted(counts.items())),

265 },

266 "task_statuses": dict(sorted(status_counts.items())),

267 "task_metadata_keys": dict(sorted(metadata_keys.items())),

268 "frontmatter_keys": dict(sorted(yaml_keys.items())),

269 "task_tags": dict(sorted(tag_counts.items())),

270 "duplicate_visible_ids": duplicate_ids,

271 "unreachable_ids": unreachable,

272 "links": links,

273 "backlink_counts": dict(sorted(backlinks.items())),

274 "parse_failures": parse_failures,

275 "omitted_files": omitted,

276 "unaccounted_files": unaccounted,

277 "unclassified_files": unclassified,

278 "integrity_agrees": integrity_agrees,

279 }

280

281

282def validate(report: dict) -> list[str]:

283 failures = []

284 if report["unaccounted_files"]:

285 failures.append("unaccounted files: " + ", ".join(report["unaccounted_files"][:12]))

286 if report["unclassified_files"]:

287 failures.append("unclassified files: " + ", ".join(report["unclassified_files"][:12]))

288 if not report["integrity_agrees"]:

289 failures.append("visible/unreachable identity population does not account for raw scan")

290 if report["counts"]["files_scanned"] != report["counts"]["files_total"]:

291 failures.append(

292 f"classification denominator mismatch: scanned {report['counts']['files_scanned']} "

293 f"of {report['counts']['files_total']} files"

294 )

295 return failures

296

297

298def tree_hashes(root: Path) -> dict[str, str]:

299 return {p.relative_to(root).as_posix(): sha256(p.read_bytes()) for p in sorted(root.rglob("*")) if p.is_file()}

300

301

302def no_op_roundtrip(copy_root: Path) -> dict:

303 before = tree_hashes(copy_root)

304 for rel, digest in before.items():

305 path = copy_root / rel

306 data = path.read_bytes()

307 path.write_bytes(data)

308 if sha256(path.read_bytes()) != digest:

309 raise RuntimeError(f"no-op changed bytes: {rel}")

310 after = tree_hashes(copy_root)

311 return {"files": len(before), "byte_exact": before == after}

312

313

314def benchmark(root: Path, rounds: int) -> dict:

315 samples = []

316 reports = []

317 for _ in range(rounds):

318 start = time.perf_counter_ns()

319 report = scan(root)

320 samples.append((time.perf_counter_ns() - start) / 1_000_000)

321 reports.append(report)

322 return {

323 "operation": "cold/full filesystem census + task/link/integrity scan",

324 "rounds": rounds,

325 "milliseconds": samples,

326 "median_ms": statistics.median(samples),

327 "min_ms": min(samples),

328 "max_ms": max(samples),

329 "stable_counts": all(r["counts"] == reports[0]["counts"] for r in reports),

330 }

331

332

333def positive_control(root: Path) -> dict:

334 with tempfile.TemporaryDirectory(prefix="folio-f0-positive-") as tmp:

335 copy = Path(tmp) / "folio"

336 shutil.copytree(root, copy, ignore=shutil.ignore_patterns(".folio-index.db*"))

337 marker = copy / "projects" / "f0-positive-control.md"

338 marker.write_text(

339 "---\nstatus: active\ntags:\n - f0-positive\n---\n\n# F0 positive control\n\n## Tasks\n\n"

340 "- [ ] unique pressure task [[f0-positive-target]] "

341 "{id: t-f0positive, tags: f0-positive}\n",

342 encoding="utf-8",

343 )

344 report = scan(copy)

345 manifest = any(m["path"] == "projects/f0-positive-control.md" for m in report["manifest"])

346 # The task count and raw population are independent paths; locate the

347 # marker directly in manifest-derived data for the parser assertion.

348 parser = report["counts"]["visible_tasks"] > scan(root)["counts"]["visible_tasks"]

349 raw = report["counts"]["raw_id_occurrences"] > scan(root)["counts"]["raw_id_occurrences"]

350 link = report["backlink_counts"].get("f0-positive-target", 0) == 1

351 return {"manifest": manifest, "task_parser": parser, "raw_id_scanner": raw, "link_scanner": link,

352 "passed": manifest and parser and raw and link}

353

354

355def main() -> int:

356 parser = argparse.ArgumentParser()

357 parser.add_argument("root", type=Path)

358 parser.add_argument("--output", type=Path)

359 parser.add_argument("--omit-prefix")

360 parser.add_argument("--rounds", type=int, default=3)

361 parser.add_argument("--positive-control", action="store_true")

362 parser.add_argument("--no-op-copy", type=Path)

363 args = parser.parse_args()

364 started = time.time()

365 report = scan(args.root, args.omit_prefix)

366 report["validation_failures"] = validate(report)

367 report["benchmark"] = benchmark(args.root, args.rounds)

368 if args.positive_control:

369 report["positive_control"] = positive_control(args.root)

370 if not report["positive_control"]["passed"]:

371 report["validation_failures"].append("positive control was not found by every inventory path")

372 if args.no_op_copy:

373 report["no_op_roundtrip"] = no_op_roundtrip(args.no_op_copy)

374 if not report["no_op_roundtrip"]["byte_exact"]:

375 report["validation_failures"].append("no-op round trip changed bytes")

376 report["elapsed_seconds"] = time.time() - started

377 rendered = json.dumps(report, indent=2, sort_keys=True) + "\n"

378 if args.output:

379 args.output.parent.mkdir(parents=True, exist_ok=True)

380 args.output.write_text(rendered, encoding="utf-8")

381 else:

382 sys.stdout.write(rendered)

383 if report["validation_failures"]:

384 for failure in report["validation_failures"]:

385 print(f"FAIL: {failure}", file=sys.stderr)

386 return 1

387 print("PASS: complete classification, independent identity accounting, and controls", file=sys.stderr)

388 return 0

389

390

391if __name__ == "__main__":

392 raise SystemExit(main())