modify documents
This commit is contained in:
@@ -0,0 +1,109 @@
|
||||
"""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
|
||||
)
|
||||
Reference in New Issue
Block a user