90 lines
3.3 KiB
Python
90 lines
3.3 KiB
Python
#!/usr/bin/env python3
|
|
"""Validate the small task pointer loaded at the start of a work session."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import re
|
|
from pathlib import Path
|
|
|
|
|
|
MAX_POINTER_BYTES = 8192
|
|
MAX_POINTER_LINES = 180
|
|
REQUIRED_HEADINGS = ("## Current", "## Read", "## Task", "## Next", "## Verify", "## Stop")
|
|
FORBIDDEN_HEADINGS = (
|
|
"## 6. 已验证证据",
|
|
"## Evidence",
|
|
"## History",
|
|
"## 历史",
|
|
"## Chat",
|
|
)
|
|
|
|
|
|
class PointerError(ValueError):
|
|
"""Raised when the bootstrap pointer would expand the model context."""
|
|
|
|
|
|
def validate_pointer(
|
|
path: Path,
|
|
*,
|
|
max_bytes: int = MAX_POINTER_BYTES,
|
|
max_lines: int = MAX_POINTER_LINES,
|
|
lock_path: Path | None = None,
|
|
) -> dict[str, int]:
|
|
text = path.read_text(encoding="utf-8")
|
|
size = len(text.encode("utf-8"))
|
|
lines = len(text.splitlines())
|
|
if size > max_bytes:
|
|
raise PointerError(f"task pointer exceeds byte budget ({size}>{max_bytes})")
|
|
if lines > max_lines:
|
|
raise PointerError(f"task pointer exceeds line budget ({lines}>{max_lines})")
|
|
missing = [heading for heading in REQUIRED_HEADINGS if heading not in text]
|
|
if missing:
|
|
raise PointerError(f"task pointer missing sections: {', '.join(missing)}")
|
|
found = [heading for heading in FORBIDDEN_HEADINGS if re.search(rf"^\s*{re.escape(heading)}\s*$", text, re.MULTILINE)]
|
|
if found:
|
|
raise PointerError(f"task pointer contains history section: {found[0]}")
|
|
if "coverage/" not in text:
|
|
raise PointerError("task pointer must direct large evidence to paths")
|
|
pointer_match = re.search(r"^context_id:\s*([^\s]+)\s*$", text, re.MULTILINE)
|
|
if pointer_match is None:
|
|
raise PointerError("task pointer is missing context_id")
|
|
if lock_path is not None and lock_path.exists():
|
|
try:
|
|
lock = json.loads(lock_path.read_text(encoding="utf-8"))
|
|
except (OSError, UnicodeError, json.JSONDecodeError) as exc:
|
|
raise PointerError(f"cannot read context lock: {exc}") from exc
|
|
lock_id = lock.get("context_id") if isinstance(lock, dict) else None
|
|
if not lock_id:
|
|
raise PointerError("context lock is missing context_id")
|
|
if pointer_match.group(1) != str(lock_id):
|
|
raise PointerError(
|
|
f"task pointer context_id does not match lock ({pointer_match.group(1)}!={lock_id})"
|
|
)
|
|
return {"bytes": size, "lines": lines, "max_bytes": max_bytes, "max_lines": max_lines}
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("path", nargs="?", type=Path, default=Path("nextTask.md"))
|
|
parser.add_argument("--max-bytes", type=int, default=MAX_POINTER_BYTES)
|
|
parser.add_argument("--max-lines", type=int, default=MAX_POINTER_LINES)
|
|
parser.add_argument("--lock", type=Path, default=Path("context-lock.json"))
|
|
args = parser.parse_args()
|
|
try:
|
|
result = validate_pointer(
|
|
args.path,
|
|
max_bytes=args.max_bytes,
|
|
max_lines=args.max_lines,
|
|
lock_path=args.lock,
|
|
)
|
|
except (OSError, UnicodeError, PointerError) as exc:
|
|
parser.error(str(exc))
|
|
print(f"task_pointer=valid bytes={result['bytes']} lines={result['lines']}")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|