From f79c2aba7e159f297a5ed9f8d921cacbf1433c8b Mon Sep 17 00:00:00 2001 From: Alexandre Teixeira <111787685+alteixeira20@users.noreply.github.com> Date: Sat, 3 Oct 2026 00:58:11 +0100 Subject: [PATCH] fix(effects): make postconditions prove intended mutations edit_file and apply_patch update claims asserted only existence (or, in the uncommitted corrective attempt, any content change), so an unrelated write could verify them. Each filesystem postcondition is now the exact content the producer's own transformation writes from the identity-checked pre-state: edit_file through the extracted pure _edit_file_text (no newline translation), apply_patch updates through _apply_patch_hunks on the universal-newline pre-state. An oversized, replaced or undecodable pre-state, a non-matching hunk, or an underivable write_file body leaves the whole claim without postconditions (UNVERIFIED) instead of letting derivable targets verify the operation or falling back to existence. --- src/agent_runtime/effect_adapters.py | 85 +++++++++++++++++++++++++--- src/agent_tools/filesystem_tools.py | 23 ++++++-- 2 files changed, 94 insertions(+), 14 deletions(-) diff --git a/src/agent_runtime/effect_adapters.py b/src/agent_runtime/effect_adapters.py index 16b1593e6..98b062962 100644 --- a/src/agent_runtime/effect_adapters.py +++ b/src/agent_runtime/effect_adapters.py @@ -12,6 +12,7 @@ from __future__ import annotations import asyncio from dataclasses import dataclass, field import hashlib +import io import json import logging import os @@ -28,6 +29,8 @@ _FILESYSTEM_READS = frozenset({"read_file", "ls", "glob", "grep"}) _JOB_READS = frozenset({"list", "ls", "jobs", "output", "get", "read", "tail", "status", "show"}) _OWNED_READS = frozenset({"vault_get", "vault_search", "list_sessions", "search_chats"}) _JOB_SETTLED = {"done", "failed"} +# Largest pre-state an edit/patch postcondition is derived from. +_PRE_STATE_LIMIT = 10 * 1024 * 1024 logger = logging.getLogger(__name__) @@ -88,28 +91,94 @@ def _write_file_digest(execution_input: str, path: str) -> str: return hashlib.sha256(_unwrap_fenced_source_body(body, path).encode("utf-8")).hexdigest() +def _pre_state_text(resource: Any, *, newline: str | None) -> str | None: + """The exact bound file decoded as its producer decodes it, or None. + + Reads only the admitted target binding (identity-checked). An oversized, + replaced or undecodable file yields None: a truncated read must never + stand in for the whole pre-state. + """ + data = _read_whole(resource, _PRE_STATE_LIMIT) + if data is None or len(data) > _PRE_STATE_LIMIT: + return None + try: + return io.TextIOWrapper(io.BytesIO(data), encoding="utf-8", newline=newline).read() + except (UnicodeDecodeError, ValueError): + return None + + +def _edit_file_digest(execution_input: str, resource: Any) -> str: + """SHA-256 of the exact bytes edit_file writes for this admitted input, or ''.""" + from src.agent_tools.filesystem_tools import _edit_file_text + try: + args = json.loads(execution_input) + except (TypeError, ValueError): + return "" + if not isinstance(args, dict): + return "" + old, new, replace_all = args.get("old_string"), args.get("new_string"), args.get("replace_all", False) + if not isinstance(old, str) or not old or not isinstance(new, str) or type(replace_all) is not bool or old == new: + return "" + # edit_file reads with newline="" and writes with newline="": no translation. + original = _pre_state_text(resource, newline="") + if original is None: + return "" + updated, _ = _edit_file_text(original, old, new, replace_all) + return "" if updated is None else hashlib.sha256(updated.encode("utf-8")).hexdigest() + + +def _patch_update_digest(op: dict, resource: Any) -> str: + """SHA-256 of the exact bytes apply_patch writes for one update, or ''.""" + from src.agent_tools.filesystem_tools import _apply_patch_hunks + # apply_patch reads updates with universal newlines and writes newline="". + original = _pre_state_text(resource, newline=None) + if original is None: + return "" + try: + updated = _apply_patch_hunks(original, op["hunks"], op["path"]) + except ValueError: + return "" + return hashlib.sha256(updated.encode("utf-8")).hexdigest() + + def _filesystem_scope(bound: Any) -> tuple[tuple[ResourceRef, ...], tuple[Postcondition, ...]]: + """Exact bindings and the requested post-state of each mutation target. + + Each postcondition is the exact content (or absence) the producer's own + transformation yields from the admitted pre-state, so an unrelated change + can never satisfy it. When any target's requested state cannot be derived + the claim carries no postcondition at all and stays UNVERIFIED: a partial + set would let the derivable targets verify the whole operation. + """ from src.agent_tools.filesystem_tools import _parse_agent_patch tool = bound.operation.tool refs = tuple(resource_ref(b.resource, b.role) for b in bound.bindings) obligations: list[Postcondition] = [] if tool == "write_file": - target = refs[0] expected = _write_file_digest(bound.execution_input, bound.bindings[0].resource.path) - obligations.append(Postcondition(target, Predicate.CONTENT_SHA256, expected) if expected - else Postcondition(target, Predicate.EXISTS)) + if not expected: + return refs, () + obligations.append(Postcondition(refs[0], Predicate.CONTENT_SHA256, expected)) elif tool == "edit_file": - obligations.append(Postcondition(refs[0], Predicate.EXISTS)) + expected = _edit_file_digest(bound.execution_input, bound.bindings[0].resource) + if not expected: + return refs, () + obligations.append(Postcondition(refs[0], Predicate.CONTENT_SHA256, expected)) elif tool == "apply_patch": ops = _parse_agent_patch(json.loads(bound.execution_input)["patch_text"]) - for op, ref in zip(ops, refs): + if len(ops) != len(bound.bindings): + return refs, () + for op, binding, ref in zip(ops, bound.bindings, refs): if op["kind"] == "add": - digest = hashlib.sha256(op["content"].encode("utf-8")).hexdigest() - obligations.append(Postcondition(ref, Predicate.CONTENT_SHA256, digest)) + obligations.append(Postcondition(ref, Predicate.CONTENT_SHA256, + hashlib.sha256(op["content"].encode("utf-8")).hexdigest())) elif op["kind"] == "delete": obligations.append(Postcondition(ref, Predicate.ABSENT)) else: - obligations.append(Postcondition(ref, Predicate.EXISTS)) + expected = _patch_update_digest(op, binding.resource) + if not expected: + return refs, () + obligations.append(Postcondition(ref, Predicate.CONTENT_SHA256, expected)) return refs, tuple(obligations) diff --git a/src/agent_tools/filesystem_tools.py b/src/agent_tools/filesystem_tools.py index 53854da61..0cdd5282d 100644 --- a/src/agent_tools/filesystem_tools.py +++ b/src/agent_tools/filesystem_tools.py @@ -116,6 +116,20 @@ def _unified_diff(old: str, new: str, path: str) -> Optional[Dict[str, Any]]: "file": os.path.basename(path) or (path or "file"), } +def _edit_file_text(original: str, old: str, new: str, replace_all: bool) -> tuple[str | None, str]: + """The exact text edit_file writes for ``original``, or None and why not. + + Pure: the effect adapter derives the requested post-state from this same + function, so the postcondition is the producer's own transformation. + """ + count = original.count(old) + if count == 0: + return None, "not_found" + if count > 1 and not replace_all: + return None, f"not_unique:{count}" + return (original.replace(old, new) if replace_all else original.replace(old, new, 1)), "ok" + + class EditFileTool: async def execute(self, content: str, ctx: dict) -> dict: from src.tool_execution import _resolve_tool_path, _resolve_search_root, _truncate @@ -150,12 +164,9 @@ class EditFileTool: # Exact replacement must not normalize unrelated CRLF/CR newlines. with open(path, "r", encoding="utf-8", newline="") as f: original = f.read() - count = original.count(old) - if count == 0: - return original, None, "not_found" - if count > 1 and not replace_all: - return original, None, f"not_unique:{count}" - updated = original.replace(old, new) if replace_all else original.replace(old, new, 1) + updated, status = _edit_file_text(original, old, new, replace_all) + if updated is None: + return original, None, status attempted.append(True) with open(path, "w", encoding="utf-8", newline="") as f: f.write(updated)