diff --git a/code_puppy_core_plugins/read_before_write/LICENSE.deepseek b/code_puppy_core_plugins/read_before_write/LICENSE.deepseek new file mode 100644 index 0000000..0b105e4 --- /dev/null +++ b/code_puppy_core_plugins/read_before_write/LICENSE.deepseek @@ -0,0 +1,28 @@ +DeepSeek Harness fs-observation-policy attribution + +The read-before-write functionality in this package is based on the +fs-observation-policy package from DeepSeek Harness: + +https://github.com/deepseek-ai/DeepSeek-Harness + +MIT License + +Copyright (c) 2026 DeepSeek + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/code_puppy_core_plugins/read_before_write/__init__.py b/code_puppy_core_plugins/read_before_write/__init__.py new file mode 100644 index 0000000..f03b330 --- /dev/null +++ b/code_puppy_core_plugins/read_before_write/__init__.py @@ -0,0 +1 @@ +"""Version-guarded read-before-write protection for Code Puppy file tools.""" diff --git a/code_puppy_core_plugins/read_before_write/policy.py b/code_puppy_core_plugins/read_before_write/policy.py new file mode 100644 index 0000000..5395586 --- /dev/null +++ b/code_puppy_core_plugins/read_before_write/policy.py @@ -0,0 +1,374 @@ +"""Observation state and decisions for the read-before-write plugin. + +The design ports DeepSeek Harness's MIT-licensed ``fs-observation-policy``. +See ``LICENSE.deepseek`` for attribution. Files are versioned pragmatically by +``(st_mtime_ns, st_size)``; this catches stale context but leaves a tiny stat -> +mutation race because Code Puppy's file tools do not expose an atomic CAS API. +""" + +from __future__ import annotations + +import ast +import logging +import os +from dataclasses import dataclass +from typing import Any, Literal + +from pydantic import TypeAdapter, ValidationError + +from code_puppy.tools.common import resolve_path + +Version = tuple[int, int] +ScopeKey = tuple[str, tuple[str, ...]] + +EDIT_TOOLS = frozenset({"delete_snippet", "replace_in_file"}) +GUARDED_TOOLS = EDIT_TOOLS | {"create_file"} +MUTATION_TOOLS = frozenset({"create_file", "delete_snippet", "replace_in_file"}) +OBSERVATION_TOOLS = MUTATION_TOOLS | {"delete_file", "read_file"} + +logger = logging.getLogger(__name__) + +_PATH_ADAPTER = TypeAdapter(str) +_OVERWRITE_ADAPTER = TypeAdapter(bool) +_HOOK_CONTEXT_PREFIX = "[hook context]\n" + + +@dataclass(frozen=True, slots=True) +class Observation: + """An authoritative observation that a canonical path exists or is absent.""" + + kind: Literal["present", "absent"] + version: Version | None = None + + +@dataclass(frozen=True, slots=True) +class ReadSnapshot: + """The tool-effective identity/version immediately before a read call.""" + + normalized_path: str + version: Version | None + + +@dataclass(frozen=True, slots=True) +class MutationSnapshot: + """The canonical target identity immediately before a mutation call.""" + + normalized_path: str + + +_observations: dict[ScopeKey, dict[str, Observation]] = {} + + +def _reset_state() -> None: + """Drop every conversation's observations (tests and defensive re-init).""" + _observations.clear() + + +def _normalize_path(file_path: Any) -> str | None: + """Resolve like the tools, then return one canonical local state key.""" + try: + raw_path = os.fspath(file_path) + except TypeError: + return None + if not isinstance(raw_path, str) or not raw_path or "\x00" in raw_path: + return None + effective_path = resolve_path(raw_path) + return os.path.realpath(os.path.abspath(effective_path)) + + +def _path_details(tool_args: Any) -> tuple[str, str] | None: + """Coerce a raw tool path exactly as downstream Pydantic validation does.""" + if not isinstance(tool_args, dict): + return None + try: + file_path = _PATH_ADAPTER.validate_python(tool_args.get("file_path")) + except ValidationError: + return None + normalized = _normalize_path(file_path) + if normalized is None: + return None + return file_path, normalized + + +def _stat_version(path: str) -> Version: + stat_result = os.stat(path) + return stat_result.st_mtime_ns, stat_result.st_size + + +def _scope_observations(scope: ScopeKey) -> dict[str, Observation]: + return _observations.setdefault(scope, {}) + + +def _set_observation( + scope: ScopeKey, + normalized_path: str, + observation: Observation, +) -> None: + _scope_observations(scope)[normalized_path] = observation + + +def _get_observation( + scope: ScopeKey, + normalized_path: str, +) -> Observation | None: + return _observations.get(scope, {}).get(normalized_path) + + +def _block(reason: str) -> dict[str, bool | str]: + return {"blocked": True, "reason": reason} + + +def _stale_read(display_path: str) -> dict[str, bool | str]: + return _block( + f"STALE READ: '{display_path}' changed on disk since you last read it " + "(external edit?). Call read_file again before editing." + ) + + +def _overwrite_requested(value: Any) -> bool: + """Coerce raw overwrite values exactly like the downstream bool field.""" + try: + return _OVERWRITE_ADAPTER.validate_python(value) + except ValidationError: + # Let the actual tool validator explain invalid values to the model. + return False + + +def enforce( + tool_name: str, + tool_args: dict[str, Any], + scope: ScopeKey, +) -> dict[str, bool | str] | None: + """Return a model-actionable block decision, or ``None`` to allow.""" + if tool_name not in GUARDED_TOOLS: + return None + + details = _path_details(tool_args) + if details is None: + return None + display_path, normalized_path = details + observation = _get_observation(scope, normalized_path) + + if tool_name in EDIT_TOOLS: + if observation is None: + return _block( + f"READ-BEFORE-WRITE: '{display_path}' has not been read this " + "session. Call read_file on it first, then retry your edit." + ) + if observation.kind == "absent": + return _block( + f"'{display_path}' does not exist (you observed it missing " + "earlier). Check the path with list_files or grep." + ) + try: + current_version = _stat_version(normalized_path) + except FileNotFoundError: + return _stale_read(display_path) + if current_version != observation.version: + return _stale_read(display_path) + return None + + if not _overwrite_requested(tool_args.get("overwrite", False)): + return None + + try: + current_version = _stat_version(normalized_path) + except FileNotFoundError: + # No target exists to clobber, regardless of an older observation. + return None + + if observation is None or observation.kind != "present": + return _block( + f"'{display_path}' already exists but hasn't been read this session. " + "Call read_file first (or use replace_in_file for a targeted edit)." + ) + if current_version != observation.version: + return _stale_read(display_path) + return None + + +def capture_mutation_snapshot( + tool_args: dict[str, Any], +) -> MutationSnapshot | None: + """Freeze a mutation target so post-call symlink drift cannot bless another.""" + details = _path_details(tool_args) + if details is None: + return None + _, normalized_path = details + return MutationSnapshot(normalized_path) + + +def capture_read_snapshot(tool_args: dict[str, Any]) -> ReadSnapshot | None: + """Capture the tool-effective identity/version immediately before a read.""" + details = _path_details(tool_args) + if details is None: + return None + _, normalized_path = details + try: + version = _stat_version(normalized_path) + except FileNotFoundError: + version = None + return ReadSnapshot(normalized_path, version) + + +def _context_wrapped_result(result: str) -> dict[Any, Any] | None: + """Recover core's structured result after hook-context string decoration.""" + if not result.startswith(_HOOK_CONTEXT_PREFIX): + return None + _, separator, payload = result.rpartition("\n\n") + if not separator: + return None + + try: + parsed = ast.literal_eval(payload) + except (SyntaxError, ValueError): + # ReadFileOutput's string form ends in its final ``error=...`` field. + if not payload.startswith("content=") or " num_tokens=" not in payload: + return None + _, error_separator, error_literal = payload.rpartition(" error=") + if not error_separator: + return None + try: + error = ast.literal_eval(error_literal) + except (SyntaxError, ValueError): + return None + return {"error": error} + return parsed if isinstance(parsed, dict) else None + + +def _result_dict(result: Any) -> dict[Any, Any] | None: + """Return dict/Pydantic results, including core context-wrapped variants.""" + if isinstance(result, dict): + return result + if isinstance(result, str): + return _context_wrapped_result(result) + + model_dump = getattr(result, "model_dump", None) + if not callable(model_dump): + return None + try: + dumped = model_dump(exclude_none=True) + except TypeError: + dumped = model_dump() + return dumped if isinstance(dumped, dict) else None + + +def _result_indicates_not_found(result: dict[Any, Any], display_path: str) -> bool: + """Match only the target-specific not-found forms emitted by read_file.""" + error = result.get("error") + if not isinstance(error, str): + return False + stripped = error.strip() + if stripped == "FILE NOT FOUND": + return True + return stripped == f"File {resolve_path(display_path)} does not exist" + + +def _record_present( + scope: ScopeKey, + normalized_path: str, + read_snapshot: ReadSnapshot | None = None, +) -> None: + """Best-effort stat and record a version the model could have observed.""" + try: + version = _stat_version(normalized_path) + except OSError: + logger.warning( + "Could not stat %s while recording a file observation", + normalized_path, + exc_info=True, + ) + return + + if read_snapshot is not None and ( + read_snapshot.normalized_path != normalized_path + or (read_snapshot.version is not None and read_snapshot.version != version) + ): + logger.warning( + "File identity/version changed while read_file was running; " + "observation skipped for %s", + normalized_path, + ) + return + _set_observation(scope, normalized_path, Observation("present", version)) + + +def record( + tool_name: str, + tool_args: dict[str, Any], + result: Any, + scope: ScopeKey, + read_snapshot: ReadSnapshot | None = None, + mutation_snapshot: MutationSnapshot | None = None, +) -> None: + """Record authoritative reads and successful mutations for one scope.""" + if tool_name not in OBSERVATION_TOOLS: + return + + details = _path_details(tool_args) + if details is None: + return + display_path, normalized_path = details + result_dict = _result_dict(result) + if result_dict is None: + return + + if tool_name == "read_file": + if _result_indicates_not_found(result_dict, display_path): + if ( + read_snapshot is not None + and read_snapshot.normalized_path != normalized_path + ): + logger.warning( + "File identity changed while missing read_file was running; " + "absent observation skipped for %s", + normalized_path, + ) + return + _set_observation(scope, normalized_path, Observation("absent")) + return + if "error" not in result_dict or result_dict.get("error") is None: + # Ranged reads count: even partial content authoritatively observed + # this path and therefore supplies a fresh stat version. + _record_present(scope, normalized_path, read_snapshot) + return + + if not result_dict.get("success"): + return + if tool_name == "delete_file": + _set_observation(scope, normalized_path, Observation("absent")) + return + + mutation_path = ( + mutation_snapshot.normalized_path + if mutation_snapshot is not None + else normalized_path + ) + if mutation_path != normalized_path: + logger.warning( + "Mutation target identity changed before observation; recording the " + "pre-call target %s instead of %s", + mutation_path, + normalized_path, + ) + _record_present(scope, mutation_path) + + +__all__ = [ + "EDIT_TOOLS", + "GUARDED_TOOLS", + "MUTATION_TOOLS", + "OBSERVATION_TOOLS", + "MutationSnapshot", + "Observation", + "ReadSnapshot", + "ScopeKey", + "Version", + "_normalize_path", + "_observations", + "_reset_state", + "capture_mutation_snapshot", + "capture_read_snapshot", + "enforce", + "record", +] diff --git a/code_puppy_core_plugins/read_before_write/register_callbacks.py b/code_puppy_core_plugins/read_before_write/register_callbacks.py new file mode 100644 index 0000000..948118e --- /dev/null +++ b/code_puppy_core_plugins/read_before_write/register_callbacks.py @@ -0,0 +1,240 @@ +"""Plugin: require fresh reads before Code Puppy file-tool edits. + +This ports DeepSeek Harness's MIT-licensed ``fs-observation-policy`` into Code +Puppy's callback architecture. A successful read (including a ranged read) or +mutation records the canonical path's ``(st_mtime_ns, st_size)`` version for the +active conversation/subagent scope. Targeted edits require that observation and +must still match it; full-file overwrites may not blindly clobber an unread file. + +This is a correctness guard, not a permission prompt, so YOLO mode does not +bypass it. Set ``read_before_write_enabled = 0`` (or ``false``) in ``puppy.cfg`` +to disable enforcement; observations continue to be recorded while disabled. +``delete_file`` is deliberately unguarded in v1 and retains its normal +interactive permission flow, matching the source policy's treatment of deletes. +Shell redirection and browser/MCP file tools are out of scope: only Code Puppy's +named file tools pass these hooks. Raw paths are Pydantic-coerced and resolved +through the same session working-directory helper as those tools before +``realpath(abspath(...))`` canonicalization. + +Versions use local metadata rather than content hashes. A pre-read snapshot +prevents a changed path/version from being blessed by the post hook, but tiny +read-syscall-to-stat and pre-stat-to-mutation races remain because the tools do +not expose atomic revision/CAS operations. Likewise, the filesystem-backend +protocol exposes no content revision, so host-only unsaved-buffer and virtual +filesystem changes cannot be versioned until core grows that API. +""" + +from __future__ import annotations + +import logging +from contextvars import ContextVar +from dataclasses import dataclass +from typing import Any + +from code_puppy.callbacks import register_callback +from code_puppy.tools.subagent_context import ( + get_conversation_root_id, + get_subagent_chain, +) + +from . import policy + +logger = logging.getLogger(__name__) + +CONFIG_KEY = "read_before_write_enabled" +ENABLED_CONFIG_KEY = CONFIG_KEY +DEFAULT_ENABLED = True + +# Re-export the state primitives from the logic module for focused tests and +# debugging without making callback registration itself chunky. +MutationSnapshot = policy.MutationSnapshot +Observation = policy.Observation +ReadSnapshot = policy.ReadSnapshot +_observations = policy._observations + + +@dataclass(frozen=True, slots=True) +class _ReadAttempt: + tool_args: dict + snapshot: policy.ReadSnapshot | None + + +@dataclass(frozen=True, slots=True) +class _MutationAttempt: + tool_name: str + tool_args: dict + snapshot: policy.MutationSnapshot | None + + +_read_attempt: ContextVar[_ReadAttempt | None] = ContextVar( + "read_before_write_read_attempt", default=None +) +_mutation_attempt: ContextVar[_MutationAttempt | None] = ContextVar( + "read_before_write_mutation_attempt", default=None +) + + +def _scope_key() -> policy.ScopeKey: + """Identify the active conversation and exact subagent ancestry.""" + return (get_conversation_root_id() or "global", get_subagent_chain()) + + +def _is_enabled() -> bool: + """Read defensive boolean config; config failures disable enforcement.""" + try: + from code_puppy.config import get_value + + raw = get_value(CONFIG_KEY) + except Exception: + logger.warning( + "Could not read %s; read-before-write enforcement is fail-open", + CONFIG_KEY, + exc_info=True, + ) + return False + + if raw is None: + return DEFAULT_ENABLED + try: + text = str(raw).strip().lower() + except Exception: + logger.warning( + "Invalid %s value; read-before-write enforcement is fail-open", + CONFIG_KEY, + exc_info=True, + ) + return False + if not text: + return DEFAULT_ENABLED + if text in {"0", "false"}: + return False + if text in {"1", "true"}: + return True + + logger.warning( + "Invalid %s value %r; falling back to enabled", + CONFIG_KEY, + raw, + ) + return DEFAULT_ENABLED + + +def _on_pre_tool_call( + tool_name: str, + tool_args: dict, + context: Any = None, +) -> dict[str, bool | str] | None: + """Enforce observation/version rules and otherwise allow the tool call.""" + _ = context + _read_attempt.set(None) + _mutation_attempt.set(None) + if tool_name in policy.MUTATION_TOOLS: + try: + snapshot = policy.capture_mutation_snapshot(tool_args) + _mutation_attempt.set(_MutationAttempt(tool_name, tool_args, snapshot)) + except Exception: + logger.warning( + "read-before-write mutation snapshot failed open", + exc_info=True, + ) + if tool_name == "read_file": + try: + snapshot = policy.capture_read_snapshot(tool_args) + _read_attempt.set(_ReadAttempt(tool_args, snapshot)) + except Exception: + logger.warning( + "read-before-write pre-read snapshot failed open", + exc_info=True, + ) + return None + if tool_name not in policy.GUARDED_TOOLS: + return None + try: + if not _is_enabled(): + return None + decision = policy.enforce(tool_name, tool_args, _scope_key()) + if isinstance(decision, dict) and decision.get("blocked"): + _mutation_attempt.set(None) + return decision + except Exception: + logger.warning( + "read-before-write pre-tool guard failed open for %s", + tool_name, + exc_info=True, + ) + return None + + +def _on_post_tool_call( + tool_name: str, + tool_args: dict, + result: Any, + duration_ms: float, + context: Any = None, +) -> None: + """Best-effort record reads and successful file mutations.""" + _ = duration_ms, context + read_attempt = _read_attempt.get() + mutation_attempt = _mutation_attempt.get() + _read_attempt.set(None) + _mutation_attempt.set(None) + if tool_name not in policy.OBSERVATION_TOOLS: + return None + read_snapshot = ( + read_attempt.snapshot + if tool_name == "read_file" + and read_attempt is not None + and read_attempt.tool_args is tool_args + else None + ) + mutation_snapshot = ( + mutation_attempt.snapshot + if tool_name in policy.MUTATION_TOOLS + and mutation_attempt is not None + and mutation_attempt.tool_name == tool_name + and mutation_attempt.tool_args is tool_args + else None + ) + try: + policy.record( + tool_name, + tool_args, + result, + _scope_key(), + read_snapshot=read_snapshot, + mutation_snapshot=mutation_snapshot, + ) + except Exception: + logger.warning( + "read-before-write observation failed for %s", + tool_name, + exc_info=True, + ) + return None + + +def _reset_state() -> None: + """Clear every recorded scope (used by tests and defensive re-init).""" + policy._reset_state() + _read_attempt.set(None) + _mutation_attempt.set(None) + + +register_callback("pre_tool_call", _on_pre_tool_call) +register_callback("post_tool_call", _on_post_tool_call) + + +__all__ = [ + "CONFIG_KEY", + "DEFAULT_ENABLED", + "ENABLED_CONFIG_KEY", + "MutationSnapshot", + "Observation", + "ReadSnapshot", + "_is_enabled", + "_observations", + "_on_post_tool_call", + "_on_pre_tool_call", + "_reset_state", + "_scope_key", +] diff --git a/plugin-names.txt b/plugin-names.txt index 9effc6f..673d154 100644 --- a/plugin-names.txt +++ b/plugin-names.txt @@ -38,6 +38,7 @@ prune puppy_kennel puppy_spinner quick_resume +read_before_write review_pr shell_safety spill diff --git a/pyproject.toml b/pyproject.toml index 08702ff..b2efbef 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -67,6 +67,7 @@ prune = "code_puppy_core_plugins.prune.register_callbacks" puppy_kennel = "code_puppy_core_plugins.puppy_kennel.register_callbacks" puppy_spinner = "code_puppy_core_plugins.puppy_spinner.register_callbacks" quick_resume = "code_puppy_core_plugins.quick_resume.register_callbacks" +read_before_write = "code_puppy_core_plugins.read_before_write.register_callbacks" review_pr = "code_puppy_core_plugins.review_pr.register_callbacks" shell_safety = "code_puppy_core_plugins.shell_safety.register_callbacks" spill = "code_puppy_core_plugins.spill.register_callbacks" diff --git a/tests/test_read_before_write.py b/tests/test_read_before_write.py new file mode 100644 index 0000000..ca218ec --- /dev/null +++ b/tests/test_read_before_write.py @@ -0,0 +1,568 @@ +"""Tests for version-guarded read-before-write file-tool enforcement.""" + +from __future__ import annotations + +import logging +import os +import time +from decimal import Decimal + +import pytest + +from code_puppy import callbacks, config +from code_puppy_core_plugins.read_before_write import policy +from code_puppy_core_plugins.read_before_write import register_callbacks as rbw + + +@pytest.fixture(autouse=True) +def _isolated_observations(): + rbw._reset_state() + yield + rbw._reset_state() + + +def _pre_raw(tool_name: str, path, **tool_args): + return rbw._on_pre_tool_call( + tool_name, + {"file_path": path, **tool_args}, + ) + + +def _pre(tool_name: str, path, **tool_args): + return _pre_raw(tool_name, str(path), **tool_args) + + +def _record_read(path, *, start_line: int | None = None): + args = {"file_path": str(path)} + if start_line is not None: + args.update(start_line=start_line, num_lines=1) + rbw._on_post_tool_call( + "read_file", + args, + {"content": "observed", "num_tokens": 2}, + 1.0, + ) + + +def _record_success(tool_name: str, path): + rbw._on_post_tool_call( + tool_name, + {"file_path": str(path)}, + {"success": True}, + 1.0, + ) + + +@pytest.mark.parametrize("tool_name", ["replace_in_file", "delete_snippet"]) +def test_edit_without_observation_is_blocked(tool_name, tmp_path): + path = tmp_path / "unread.txt" + path.write_text("hello", encoding="utf-8") + + decision = _pre(tool_name, path) + + assert decision == { + "blocked": True, + "reason": ( + f"READ-BEFORE-WRITE: '{path}' has not been read this session. " + "Call read_file on it first, then retry your edit." + ), + } + + +def test_ranged_read_then_edit_is_allowed(tmp_path): + path = tmp_path / "observed.txt" + path.write_text("one\ntwo\n", encoding="utf-8") + + _record_read(path, start_line=2) + + assert _pre("replace_in_file", path) is None + + +def test_core_read_result_model_records_observation(tmp_path): + from code_puppy.tools.file_operations import ReadFileOutput + + path = tmp_path / "pydantic-result.txt" + path.write_text("observed", encoding="utf-8") + result = ReadFileOutput(content="observed", num_tokens=2) + + rbw._on_post_tool_call("read_file", {"file_path": str(path)}, result, 1.0) + + assert _pre("replace_in_file", path) is None + + +def test_edit_after_successful_own_write_is_allowed(tmp_path): + path = tmp_path / "owned.txt" + path.write_text("first", encoding="utf-8") + _record_read(path) + assert _pre("replace_in_file", path) is None + + path.write_text("second version", encoding="utf-8") + _record_success("replace_in_file", path) + + assert _pre("delete_snippet", path) is None + + +def test_observed_absent_then_edit_has_not_found_guidance(tmp_path): + path = tmp_path / "missing.txt" + rbw._on_post_tool_call( + "read_file", + {"file_path": str(path)}, + {"error": f"File {path} does not exist"}, + 1.0, + ) + + decision = _pre("replace_in_file", path) + + assert decision == { + "blocked": True, + "reason": ( + f"'{path}' does not exist (you observed it missing earlier). " + "Check the path with list_files or grep." + ), + } + + +def test_external_change_after_read_is_stale(tmp_path): + path = tmp_path / "stale.txt" + path.write_text("before", encoding="utf-8") + _record_read(path) + observed_stat = path.stat() + + path.write_text("after, with a different size", encoding="utf-8") + bumped_mtime = max(time.time_ns(), observed_stat.st_mtime_ns + 1_000_000) + os.utime(path, ns=(observed_stat.st_atime_ns, bumped_mtime)) + + decision = _pre("replace_in_file", path) + + assert decision == { + "blocked": True, + "reason": ( + f"STALE READ: '{path}' changed on disk since you last read it " + "(external edit?). Call read_file again before editing." + ), + } + + +def test_create_overwrite_guards_only_clobbers(tmp_path): + existing = tmp_path / "existing.txt" + existing.write_text("keep", encoding="utf-8") + missing = tmp_path / "new.txt" + + existing_decision = _pre("create_file", existing, overwrite=True) + + assert existing_decision == { + "blocked": True, + "reason": ( + f"'{existing}' already exists but hasn't been read this session. " + "Call read_file first (or use replace_in_file for a targeted edit)." + ), + } + assert _pre("create_file", missing, overwrite=True) is None + assert _pre("create_file", existing, overwrite=False) is None + assert _pre("create_file", existing, overwrite="false") is None + assert _pre("create_file", existing, overwrite="yes")["blocked"] is True + + +def test_create_overwrite_uses_observed_version(tmp_path): + path = tmp_path / "overwrite.txt" + path.write_text("original", encoding="utf-8") + _record_read(path) + assert _pre("create_file", path, overwrite=True) is None + + path.write_text("external change", encoding="utf-8") + + decision = _pre("create_file", path, overwrite=True) + assert decision and "STALE READ" in decision["reason"] + # A no-clobber create remains delegated to the tool even with stale state. + assert _pre("create_file", path, overwrite=False) is None + + +def test_create_overwrite_allows_target_deleted_since_read(tmp_path): + path = tmp_path / "recreate.txt" + path.write_text("observed", encoding="utf-8") + _record_read(path) + path.unlink() + + assert _pre("create_file", path, overwrite=True) is None + + +def test_absent_read_then_successful_create_allows_edit(tmp_path): + path = tmp_path / "created-after-read.txt" + rbw._on_post_tool_call( + "read_file", + {"file_path": str(path)}, + {"error": f"File {path} does not exist"}, + 1.0, + ) + assert _pre("create_file", path, overwrite=False) is None + + path.write_text("new content", encoding="utf-8") + _record_success("create_file", path) + + assert _pre("replace_in_file", path) is None + + +def test_successful_delete_snippet_refreshes_version(tmp_path): + path = tmp_path / "snippet.txt" + path.write_text("keep remove", encoding="utf-8") + _record_read(path) + assert _pre("delete_snippet", path) is None + + path.write_text("keep ", encoding="utf-8") + _record_success("delete_snippet", path) + + assert _pre("replace_in_file", path) is None + + +def test_successful_guarded_write_updates_recorded_version(tmp_path): + path = tmp_path / "twice.txt" + path.write_text("version one", encoding="utf-8") + _record_read(path) + assert _pre("replace_in_file", path) is None + + path.write_text("version two is longer", encoding="utf-8") + _record_success("replace_in_file", path) + + normalized = policy._normalize_path(path) + observation = rbw._observations[rbw._scope_key()][normalized] + current = path.stat() + assert observation.version == (current.st_mtime_ns, current.st_size) + assert _pre("replace_in_file", path) is None + + +@pytest.mark.parametrize("change", ["mtime", "size"]) +def test_each_version_component_independently_detects_stale_read(tmp_path, change): + path = tmp_path / f"stale-{change}.txt" + path.write_text("same-size", encoding="utf-8") + _record_read(path) + observed = path.stat() + + if change == "mtime": + path.write_text("new-value", encoding="utf-8") + os.utime( + path, + ns=(observed.st_atime_ns, observed.st_mtime_ns + 1_000_000), + ) + else: + path.write_text("different-size", encoding="utf-8") + os.utime(path, ns=(observed.st_atime_ns, observed.st_mtime_ns)) + + decision = _pre("replace_in_file", path) + assert decision and "STALE READ" in decision["reason"] + + +def test_delete_success_records_absent_without_guarding_delete(tmp_path): + path = tmp_path / "delete-me.txt" + path.write_text("bye", encoding="utf-8") + + assert _pre("delete_file", path) is None + path.unlink() + _record_success("delete_file", path) + + decision = _pre("delete_snippet", path) + assert decision and "observed it missing" in decision["reason"] + + +def test_non_file_tools_and_reads_pass_through(tmp_path): + path = tmp_path / "anything.txt" + + assert _pre("read_file", path) is None + assert ( + rbw._on_pre_tool_call("agent_run_shell_command", {"command": "echo hi"}) is None + ) + + +def test_observations_are_isolated_by_conversation_and_subagent(tmp_path, monkeypatch): + path = tmp_path / "scoped.txt" + path.write_text("scope", encoding="utf-8") + active = {"root": "conversation-a", "chain": ()} + monkeypatch.setattr(rbw, "get_conversation_root_id", lambda: active["root"]) + monkeypatch.setattr(rbw, "get_subagent_chain", lambda: active["chain"]) + + _record_read(path) + assert _pre("replace_in_file", path) is None + + active["root"] = "conversation-b" + assert _pre("replace_in_file", path)["blocked"] is True + + active["root"] = "conversation-a" + active["chain"] = ("reviewer",) + assert _pre("replace_in_file", path)["blocked"] is True + + active["chain"] = () + assert _pre("replace_in_file", path) is None + + +def test_config_disabled_allows_guarded_operations(tmp_path): + path = tmp_path / "disabled.txt" + path.write_text("unread", encoding="utf-8") + config.set_value(rbw.ENABLED_CONFIG_KEY, "false") + + assert _pre("replace_in_file", path) is None + assert _pre("delete_snippet", path) is None + assert _pre("create_file", path, overwrite=True) is None + + +def test_config_disabled_still_records_observations(tmp_path): + path = tmp_path / "recorded-while-disabled.txt" + path.write_text("content", encoding="utf-8") + config.set_value(rbw.ENABLED_CONFIG_KEY, "0") + + _record_read(path) + config.set_value(rbw.ENABLED_CONFIG_KEY, "1") + + assert _pre("replace_in_file", path) is None + + +def test_unexpected_stat_error_fails_open_and_warns(tmp_path, monkeypatch, caplog): + path = tmp_path / "stat-error.txt" + path.write_text("content", encoding="utf-8") + _record_read(path) + + def broken_stat(*args, **kwargs): + raise OSError("surprise stat failure") + + # Keep the config read from consuming the mocked stat failure first; this + # test targets the version check itself. + monkeypatch.setattr(rbw, "_is_enabled", lambda: True) + monkeypatch.setattr(policy.os, "stat", broken_stat) + with caplog.at_level(logging.WARNING): + decision = _pre("replace_in_file", path) + + assert decision is None + assert "failed open" in caplog.text + + +def test_recording_stat_error_is_best_effort_and_warns(tmp_path, monkeypatch, caplog): + path = tmp_path / "post-stat-error.txt" + path.write_text("content", encoding="utf-8") + + def broken_stat(*args, **kwargs): + raise OSError("post-hook stat failure") + + monkeypatch.setattr(policy.os, "stat", broken_stat) + with caplog.at_level(logging.WARNING): + _record_read(path) + + assert "while recording a file observation" in caplog.text + assert rbw._observations.get(rbw._scope_key(), {}) == {} + + +def test_callbacks_are_registered(): + assert rbw._on_pre_tool_call in callbacks.get_callbacks( + "pre_tool_call", include_disabled=True + ) + assert rbw._on_post_tool_call in callbacks.get_callbacks( + "post_tool_call", include_disabled=True + ) + + +def test_yolo_mode_does_not_bypass_guard(tmp_path): + path = tmp_path / "yolo-unread.txt" + path.write_text("content", encoding="utf-8") + previous = config.get_cli_yolo_override() + config.set_cli_yolo_override(True) + try: + decision = _pre("replace_in_file", path) + finally: + config.set_cli_yolo_override(previous) + + assert decision and decision["blocked"] is True + + +@pytest.mark.parametrize( + "raw_path_factory", + [os.fsencode, lambda path: bytearray(os.fsencode(path))], +) +def test_raw_paths_use_downstream_pydantic_coercion(tmp_path, raw_path_factory): + path = tmp_path / "coerced-path.txt" + path.write_text("content", encoding="utf-8") + + decision = _pre_raw("replace_in_file", raw_path_factory(path)) + + assert decision and decision["blocked"] is True + assert str(path) in decision["reason"] + + +def test_raw_overwrite_uses_downstream_pydantic_coercion(tmp_path): + path = tmp_path / "decimal-overwrite.txt" + path.write_text("content", encoding="utf-8") + + assert _pre_raw("create_file", str(path), overwrite=Decimal(1))["blocked"] + # Path objects fail the downstream ``str`` validator, so policy lets the + # normal validation error through instead of inventing a policy denial. + assert _pre_raw("replace_in_file", path) is None + + +def test_session_working_directory_is_used_instead_of_process_cwd( + tmp_path, monkeypatch +): + from code_puppy.tools.common import reset_working_directory, set_working_directory + + process_cwd = tmp_path / "process-cwd" + workspace = tmp_path / "workspace" + process_cwd.mkdir() + workspace.mkdir() + shadow = process_cwd / "relative.txt" + target = workspace / "relative.txt" + shadow.write_text("shadow", encoding="utf-8") + target.write_text("workspace", encoding="utf-8") + monkeypatch.chdir(process_cwd) + token = set_working_directory(str(workspace)) + try: + _record_read("relative.txt") + target.write_text("workspace changed", encoding="utf-8") + decision = _pre("replace_in_file", "relative.txt") + finally: + reset_working_directory(token) + + assert decision and "STALE READ" in decision["reason"] + + +def test_tilde_path_resolves_like_file_tools(tmp_path, monkeypatch): + home = tmp_path / "home" + home.mkdir() + target = home / "tilde.txt" + target.write_text("content", encoding="utf-8") + monkeypatch.setenv("HOME", str(home)) + + _record_read("~/tilde.txt") + + assert _pre("replace_in_file", "~/tilde.txt") is None + assert set(rbw._observations[rbw._scope_key()]) == {os.path.realpath(target)} + + +def test_hook_context_wrapped_mutation_result_refreshes_version(tmp_path): + path = tmp_path / "context-mutation.txt" + path.write_text("first", encoding="utf-8") + _record_read(path) + path.write_text("second version", encoding="utf-8") + wrapped = "[hook context]\nemoji_filter changed args\n\n{'success': True}" + + rbw._on_post_tool_call( + "replace_in_file", + {"file_path": str(path)}, + wrapped, + 1.0, + ) + + assert _pre("replace_in_file", path) is None + + +def test_hook_context_wrapped_read_model_records_present_and_absent(tmp_path): + from code_puppy.tools.file_operations import ReadFileOutput + + present = tmp_path / "wrapped-present.txt" + present.write_text("content", encoding="utf-8") + present_args = {"file_path": str(present)} + rbw._on_pre_tool_call("read_file", present_args) + present_result = ReadFileOutput(content="content", num_tokens=2) + rbw._on_post_tool_call( + "read_file", + present_args, + f"[hook context]\nhook note\n\n{present_result}", + 1.0, + ) + assert _pre("replace_in_file", present) is None + + missing = tmp_path / "wrapped-missing.txt" + missing_args = {"file_path": str(missing)} + error = f"File {missing} does not exist" + rbw._on_pre_tool_call("read_file", missing_args) + missing_result = ReadFileOutput(content=error, num_tokens=0, error=error) + rbw._on_post_tool_call( + "read_file", + missing_args, + f"[hook context]\nhook note\n\n{missing_result}", + 1.0, + ) + assert "observed it missing" in _pre("replace_in_file", missing)["reason"] + + +def test_changed_file_during_read_is_not_blessed(tmp_path, caplog): + path = tmp_path / "read-race.txt" + path.write_text("version one", encoding="utf-8") + args = {"file_path": str(path)} + assert rbw._on_pre_tool_call("read_file", args) is None + path.write_text("version two is different", encoding="utf-8") + + with caplog.at_level(logging.WARNING): + rbw._on_post_tool_call( + "read_file", + args, + {"content": "version one", "num_tokens": 3}, + 1.0, + ) + + decision = _pre("replace_in_file", path) + assert decision and "READ-BEFORE-WRITE" in decision["reason"] + assert "changed while read_file was running" in caplog.text + + +def test_unrelated_not_found_text_does_not_record_absence(tmp_path): + path = tmp_path / "backend-error.txt" + path.write_text("content", encoding="utf-8") + + rbw._on_post_tool_call( + "read_file", + {"file_path": str(path)}, + {"error": "Backend dependency file not found"}, + 1.0, + ) + + decision = _pre("replace_in_file", path) + assert decision and "has not been read" in decision["reason"] + + +def test_missing_or_invalid_path_is_not_guarded(): + assert rbw._on_pre_tool_call("replace_in_file", {}) is None + assert rbw._on_pre_tool_call("replace_in_file", {"file_path": None}) is None + assert ( + rbw._on_pre_tool_call("replace_in_file", {"file_path": "bad\x00path"}) is None + ) + + +def test_symlink_retargeted_during_read_is_not_blessed(tmp_path): + first = tmp_path / "first.txt" + second = tmp_path / "second.txt" + alias = tmp_path / "alias.txt" + first.write_text("first", encoding="utf-8") + second.write_text("second", encoding="utf-8") + try: + alias.symlink_to(first) + except OSError as exc: # pragma: no cover - platform privilege fallback + pytest.skip(f"symlinks unavailable: {exc}") + + args = {"file_path": str(alias)} + rbw._on_pre_tool_call("read_file", args) + alias.unlink() + alias.symlink_to(second) + rbw._on_post_tool_call( + "read_file", + args, + {"content": "first", "num_tokens": 1}, + 1.0, + ) + + decision = _pre("replace_in_file", alias) + assert decision and "READ-BEFORE-WRITE" in decision["reason"] + + +def test_dotdot_and_symlink_paths_share_one_canonical_observation(tmp_path): + directory = tmp_path / "real" + nested = directory / "nested" + nested.mkdir(parents=True) + target = directory / "target.txt" + target.write_text("canonical", encoding="utf-8") + hostile_path = nested / ".." / "target.txt" + symlink_path = tmp_path / "alias.txt" + try: + symlink_path.symlink_to(target) + except OSError as exc: # pragma: no cover - platform privilege fallback + pytest.skip(f"symlinks unavailable: {exc}") + + _record_read(hostile_path) + + assert _pre("replace_in_file", symlink_path) is None + scope_state = rbw._observations[rbw._scope_key()] + assert set(scope_state) == {os.path.realpath(os.path.abspath(target))} diff --git a/tests/test_read_before_write_identity_races.py b/tests/test_read_before_write_identity_races.py new file mode 100644 index 0000000..6b6ec37 --- /dev/null +++ b/tests/test_read_before_write_identity_races.py @@ -0,0 +1,126 @@ +"""Adversarial identity-race tests for read-before-write observations.""" + +from __future__ import annotations + +import logging + +import pytest + +from code_puppy_core_plugins.read_before_write import register_callbacks as rbw + + +@pytest.fixture(autouse=True) +def _isolated_observations(): + rbw._reset_state() + yield + rbw._reset_state() + + +def _symlink(link, target): + try: + link.symlink_to(target) + except OSError as exc: # pragma: no cover - platform privilege fallback + pytest.skip(f"symlinks unavailable: {exc}") + + +def test_retarget_after_successful_mutation_does_not_bless_new_target(tmp_path, caplog): + first = tmp_path / "first.txt" + second = tmp_path / "second.txt" + alias = tmp_path / "alias.txt" + first.write_text("first", encoding="utf-8") + second.write_text("second", encoding="utf-8") + _symlink(alias, first) + + rbw._on_post_tool_call( + "read_file", + {"file_path": str(alias)}, + {"content": "first", "num_tokens": 1}, + 1.0, + ) + args = {"file_path": str(alias), "content": "written", "overwrite": True} + assert rbw._on_pre_tool_call("create_file", args) is None + + first.write_text("written", encoding="utf-8") + alias.unlink() + alias.symlink_to(second) + with caplog.at_level(logging.WARNING): + rbw._on_post_tool_call("create_file", args, {"success": True}, 1.0) + + decision = rbw._on_pre_tool_call("create_file", args) + assert decision and "hasn't been read" in decision["reason"] + assert "recording the pre-call target" in caplog.text + + +def test_blocked_snapshot_cannot_poison_later_direct_post(tmp_path): + unread = tmp_path / "unread.txt" + direct = tmp_path / "direct.txt" + unread.write_text("unread", encoding="utf-8") + direct.write_text("created directly", encoding="utf-8") + unread_args = { + "file_path": str(unread), + "content": "clobber", + "overwrite": True, + } + + decision = rbw._on_pre_tool_call("create_file", unread_args) + assert decision and decision["blocked"] is True + assert rbw._mutation_attempt.get() is None + + direct_args = {"file_path": str(direct)} + rbw._on_post_tool_call("create_file", direct_args, {"success": True}, 1.0) + + assert rbw._on_pre_tool_call("replace_in_file", direct_args) is None + assert rbw._on_pre_tool_call("create_file", unread_args)["blocked"] is True + + +def test_paired_posts_clear_attempt_state_even_on_failure_or_exception( + tmp_path, monkeypatch +): + path = tmp_path / "paired.txt" + path.write_text("content", encoding="utf-8") + read_args = {"file_path": str(path)} + rbw._on_pre_tool_call("read_file", read_args) + rbw._on_post_tool_call( + "read_file", + read_args, + {"content": "content", "num_tokens": 2}, + 1.0, + ) + assert rbw._read_attempt.get() is None + assert rbw._mutation_attempt.get() is None + + mutation_args = {"file_path": str(path), "replacements": []} + assert rbw._on_pre_tool_call("replace_in_file", mutation_args) is None + rbw._on_post_tool_call("replace_in_file", mutation_args, {"success": False}, 1.0) + assert rbw._read_attempt.get() is None + assert rbw._mutation_attempt.get() is None + + assert rbw._on_pre_tool_call("replace_in_file", mutation_args) is None + + def broken_record(*args, **kwargs): + raise RuntimeError("post observer exploded") + + monkeypatch.setattr(rbw.policy, "record", broken_record) + rbw._on_post_tool_call("replace_in_file", mutation_args, {"success": True}, 1.0) + assert rbw._read_attempt.get() is None + assert rbw._mutation_attempt.get() is None + + +def test_retarget_after_missing_read_does_not_mark_new_target_absent(tmp_path, caplog): + missing = tmp_path / "missing.txt" + existing = tmp_path / "existing.txt" + alias = tmp_path / "alias.txt" + existing.write_text("existing", encoding="utf-8") + _symlink(alias, missing) + + args = {"file_path": str(alias)} + assert rbw._on_pre_tool_call("read_file", args) is None + alias.unlink() + alias.symlink_to(existing) + result = {"error": f"File {alias} does not exist"} + with caplog.at_level(logging.WARNING): + rbw._on_post_tool_call("read_file", args, result, 1.0) + + decision = rbw._on_pre_tool_call("replace_in_file", args) + assert decision and "has not been read" in decision["reason"] + assert "absent observation skipped" in caplog.text