KOKO\Mimi
110 lines
3.3 KiB
Python
110 lines
3.3 KiB
Python
"""Track whether a Codex turn edited Markdown files only."""
|
|
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Any, Iterable
|
|
|
|
|
|
EDIT_TOOL_NAMES = frozenset({"apply_patch", "Edit", "MultiEdit", "Write"})
|
|
STATE_VERSION = 1
|
|
|
|
|
|
def _identifier(payload: dict[str, Any], name: str) -> str | None:
|
|
value = payload.get(name)
|
|
if not isinstance(value, str) or not value.strip():
|
|
return None
|
|
return value
|
|
|
|
|
|
def _digest(value: str) -> str:
|
|
return hashlib.sha256(value.encode("utf-8")).hexdigest()
|
|
|
|
|
|
def _state_path(root: Path, session_id: str) -> Path:
|
|
return root / ".harness" / "hook-state" / f"{_digest(session_id)}.json"
|
|
|
|
|
|
def _is_markdown_in_root(path: Path, root: Path) -> bool:
|
|
resolved_root = root.resolve()
|
|
resolved_path = path.resolve()
|
|
try:
|
|
resolved_path.relative_to(resolved_root)
|
|
except ValueError:
|
|
return False
|
|
return resolved_path.suffix.lower() == ".md"
|
|
|
|
|
|
def record_edit_scope(
|
|
payload: dict[str, Any],
|
|
root: Path,
|
|
paths: Iterable[Path],
|
|
) -> bool:
|
|
"""Record an allowed edit; return False when turn identity is unavailable."""
|
|
session_id = _identifier(payload, "session_id")
|
|
turn_id = _identifier(payload, "turn_id")
|
|
if session_id is None or turn_id is None:
|
|
return False
|
|
|
|
edit_paths = list(paths)
|
|
if not edit_paths:
|
|
raise ValueError("edit scope requires at least one path")
|
|
|
|
turn_digest = _digest(turn_id)
|
|
state_path = _state_path(root, session_id)
|
|
document_only = all(
|
|
_is_markdown_in_root(path, root) for path in edit_paths
|
|
)
|
|
|
|
if state_path.is_file():
|
|
try:
|
|
previous = json.loads(state_path.read_text(encoding="utf-8"))
|
|
valid_previous = (
|
|
isinstance(previous, dict)
|
|
and previous.get("version") == STATE_VERSION
|
|
and isinstance(previous.get("turn"), str)
|
|
and isinstance(previous.get("document_only"), bool)
|
|
)
|
|
if not valid_previous:
|
|
document_only = False
|
|
elif previous["turn"] == turn_digest:
|
|
document_only = previous["document_only"] and document_only
|
|
except (json.JSONDecodeError, OSError):
|
|
document_only = False
|
|
|
|
state_path.parent.mkdir(parents=True, exist_ok=True)
|
|
temporary_path = state_path.with_suffix(".tmp")
|
|
temporary_path.write_text(
|
|
json.dumps(
|
|
{
|
|
"version": STATE_VERSION,
|
|
"turn": turn_digest,
|
|
"document_only": document_only,
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
temporary_path.replace(state_path)
|
|
return True
|
|
|
|
|
|
def is_document_only_turn(payload: dict[str, Any], root: Path) -> bool:
|
|
"""Return True only for a valid state proving Markdown-only edits."""
|
|
session_id = _identifier(payload, "session_id")
|
|
turn_id = _identifier(payload, "turn_id")
|
|
if session_id is None or turn_id is None:
|
|
return False
|
|
|
|
state_path = _state_path(root, session_id)
|
|
try:
|
|
state = json.loads(state_path.read_text(encoding="utf-8"))
|
|
except (FileNotFoundError, json.JSONDecodeError, OSError):
|
|
return False
|
|
|
|
return (
|
|
isinstance(state, dict)
|
|
and state.get("version") == STATE_VERSION
|
|
and state.get("turn") == _digest(turn_id)
|
|
and state.get("document_only") is True
|
|
)
|