Fail closed on invalid JSON/YAML/TOML writes instead of writing then reporting

write_file() previously called _atomic_write() first and only ran the
JSON/YAML/TOML/Python syntax check afterward as an informational lint
delta -- a parse failure never set the top-level `error` key, so a
corrupt structured-data write still landed on disk (and file_tools.py's
files_modified gating, which keys off `error`, silently reported it as
a successful modification).

Move the in-process syntax check for JSON/YAML/TOML ahead of
_atomic_write() and refuse the write outright on a parse failure: no
temp file, no rename, nothing touches disk, and the result carries a
top-level `error` so callers correctly see it as unmodified.

Deliberately scoped to _FAIL_CLOSED_INPROC_EXTS (JSON/YAML/TOML), not
all of LINTERS_INPROC -- .py is excluded because this codebase's own
test fixtures (TestPatchReplacePostWriteVerification et al.) write
arbitrary non-Python text through *.py paths purely to exercise
write-mechanics; a hard block there broke 3 previously-passing tests
during development. Python keeps its pre-existing non-blocking
lint-delta report.

Adds tests/tools/test_write_file_syntax_gate.py: invalid JSON/YAML/YML/
TOML refused with nothing written (new file) and nothing modified
(existing file); valid JSON/YAML still written byte-for-byte; a
non-linted extension with garbage content is unaffected; invalid Python
is confirmed NOT hard-refused (still just reported).
This commit is contained in:
Neo Guyver 2026-07-07 13:35:15 -07:00 committed by Teknium
parent 4d7f8ade3e
commit 2e1982f83d
2 changed files with 169 additions and 7 deletions

View file

@ -0,0 +1,101 @@
"""Tests for the fail-closed pre-write syntax gate on write_file.
Structured formats with an in-process linter (JSON/YAML/TOML) are validated
BEFORE any bytes touch disk: a candidate write that doesn't parse is refused
outright -- nothing lands on disk -- instead of being written and merely
reported afterward via the post-write lint delta.
These run against a REAL LocalEnvironment (actual shell commands / actual
files under tmp_path), matching the existing pattern in
tests/tools/test_file_write_safety.py::TestAtomicWrite.
"""
import json
from pathlib import Path
import pytest
from tools.environments.local import LocalEnvironment
from tools.file_operations import ShellFileOperations
@pytest.fixture
def ops(tmp_path: Path):
env = LocalEnvironment(cwd=str(tmp_path))
return ShellFileOperations(env, cwd=str(tmp_path))
class TestFailClosedSyntaxGate:
def test_invalid_json_refused_file_not_created(self, ops, tmp_path: Path):
target = tmp_path / "config.json"
res = ops.write_file(str(target), '{"a": 1,') # truncated / invalid
assert res.error is not None
assert "json" in res.error.lower()
assert not target.exists(), "invalid JSON must NOT be written to disk"
def test_invalid_json_refused_existing_file_not_modified(self, ops, tmp_path: Path):
target = tmp_path / "config.json"
target.write_text('{"a": 1}')
res = ops.write_file(str(target), '{"a": 1,')
assert res.error is not None
assert target.read_text() == '{"a": 1}', (
"existing valid file must be left untouched by a refused write"
)
def test_invalid_yaml_refused_file_not_created(self, ops, tmp_path: Path):
target = tmp_path / "config.yaml"
res = ops.write_file(str(target), 'key: "unclosed\n')
assert res.error is not None
assert "yaml" in res.error.lower()
assert not target.exists(), "invalid YAML must NOT be written to disk"
def test_invalid_yml_extension_also_refused(self, ops, tmp_path: Path):
target = tmp_path / "config.yml"
res = ops.write_file(str(target), 'key: "unclosed\n')
assert res.error is not None
assert not target.exists()
def test_valid_json_written_exactly(self, ops, tmp_path: Path):
target = tmp_path / "config.json"
content = json.dumps({"a": 1, "b": [1, 2, 3]})
res = ops.write_file(str(target), content)
assert res.error is None, res.error
assert target.read_text() == content
def test_valid_yaml_written_exactly(self, ops, tmp_path: Path):
target = tmp_path / "config.yaml"
content = "a: 1\nb:\n - 1\n - 2\n"
res = ops.write_file(str(target), content)
assert res.error is None, res.error
assert target.read_text() == content
def test_non_linted_extension_with_garbage_still_written(self, ops, tmp_path: Path):
"""Behavior for extensions with NO in-process linter is unchanged --
garbage content is written as-is, no refusal."""
target = tmp_path / "notes.txt"
garbage = "{{{ not json, not yaml, not anything ]]] <<<"
res = ops.write_file(str(target), garbage)
assert res.error is None, res.error
assert target.read_text() == garbage
def test_invalid_python_is_NOT_hard_refused(self, ops, tmp_path: Path):
"""Deliberate scope decision: .py keeps the pre-existing NON-BLOCKING
lint-delta report rather than a hard refusal (see
``_FAIL_CLOSED_INPROC_EXTS`` in tools/file_operations.py for why --
this codebase's own test suite writes arbitrary non-Python content
through *.py paths as generic write-mechanics fixtures)."""
target = tmp_path / "broken.py"
bad_python = "def foo(:\n pass\n"
res = ops.write_file(str(target), bad_python)
assert res.error is None, res.error
assert target.read_text() == bad_python
# Still surfaced via the (non-blocking) lint report:
assert res.lint is not None
assert res.lint.get("status") == "error"
assert "SyntaxError" in res.lint.get("output", "")
def test_invalid_toml_refused_file_not_created(self, ops, tmp_path: Path):
target = tmp_path / "config.toml"
res = ops.write_file(str(target), "[section\nk = 'v'")
assert res.error is not None
assert not target.exists()

View file

@ -677,6 +677,21 @@ LINTERS_INPROC = {
'.toml': _lint_toml_inproc,
}
# Subset of LINTERS_INPROC that the pre-write fail-closed gate in
# ``write_file`` (see below) refuses on, rather than merely reporting.
# Deliberately excludes ``.py``: unlike JSON/YAML/TOML (atomic structured
# data blobs where "doesn't parse" always means "corrupt"), ``.py`` is
# used throughout this codebase's own test fixtures as a generic
# stand-in extension for arbitrary non-Python text content (e.g.
# ``tests/tools/test_file_operations.py``'s
# ``TestPatchReplacePostWriteVerification`` writes "hello world" /
# "hi world" through a ``*.py`` path purely to exercise write-mechanics,
# not Python validity). Hard-refusing on invalid Python would treat that
# established, exercised pattern as an error and break it. Python source
# keeps the existing (unchanged) post-write lint-delta *report* — still
# visible to the caller, just not a write-blocking refusal.
_FAIL_CLOSED_INPROC_EXTS = frozenset({'.json', '.yaml', '.yml', '.toml'})
# Max limits for read operations
MAX_LINES = 2000
MAX_LINE_LENGTH = 2000
@ -1316,12 +1331,21 @@ class ShellFileOperations(FileOperations):
files. The content never appears in the shell command string
only the file path does.
After the write, runs a post-first / pre-lazy lint check via
``_check_lint_delta()``. If the new content is clean, the lint
call is O(one parse). If the new content has errors, the pre-write
content is linted too and only errors newly introduced by this
write are surfaced pre-existing problems are filtered out so
the agent isn't distracted chasing them.
Before anything touches disk, a fail-closed syntax gate runs
against the CANDIDATE content: if ``path``'s extension is in
``_FAIL_CLOSED_INPROC_EXTS`` (JSON/YAML/TOML structured data
formats where a parse failure always means corruption) and the
candidate content doesn't parse, the write is refused outright.
No temp file, no rename, nothing on disk changes.
After a write that clears the gate, runs a post-first / pre-lazy
lint check via ``_check_lint_delta()``. If the new content is
clean, the lint call is O(one parse). If the new content has
errors the gate didn't already catch (i.e. errors from a linter
outside ``_FAIL_CLOSED_INPROC_EXTS``, such as Python), the
pre-write content is linted too and only errors newly introduced
by this write are surfaced pre-existing problems are filtered
out so the agent isn't distracted chasing them.
Args:
path: File path to write
@ -1337,6 +1361,44 @@ class ShellFileOperations(FileOperations):
if _is_write_denied(path):
return WriteResult(error=f"Write denied: '{path}' is a protected system/credential file.")
# ── Fail-closed pre-write syntax gate ───────────────────────────
# Validate the CANDIDATE content BEFORE any bytes touch disk —
# previously this only ran as a post-write lint *report* that the
# caller could ignore (or that ``files_modified`` gating wouldn't
# catch, since a lint failure never set the top-level ``error``
# key). A structured-format write that doesn't even parse (mashed
# quotes, truncated generation, wrong indentation dialect) is a
# corrupt write, not a style nit — refuse it outright instead of
# writing first and reporting the damage afterward.
#
# Scope: only extensions in ``_FAIL_CLOSED_INPROC_EXTS`` (JSON/
# YAML/TOML). ``.py`` deliberately keeps its pre-existing,
# non-blocking lint-delta *report* instead of a hard refusal — see
# ``_FAIL_CLOSED_INPROC_EXTS``'s docstring above for why. Extensions
# with no in-process linter at all (including ones only covered by
# a shell linter) are completely unaffected — this gate never runs
# for them, so behavior there is unchanged.
#
# Checked against the raw ``content`` argument, before the
# BOM/CRLF preservation shims below run. Those shims exist purely
# to match the on-disk file's existing conventions; linting
# post-shim would false-positive a JSONDecodeError on a
# legitimately BOM-marked JSON file purely because this method
# re-adds the marker the read layer strips — see
# ``_file_has_bom``/``_UTF8_BOM`` below.
ext = os.path.splitext(path)[1].lower()
inproc_linter = LINTERS_INPROC.get(ext) if ext in _FAIL_CLOSED_INPROC_EXTS else None
if inproc_linter is not None:
_ok, _lint_err = inproc_linter(content)
if not _ok and _lint_err != "__SKIP__":
return WriteResult(
error=(
f"Refusing to write '{path}': candidate content fails "
f"{ext} syntax validation ({_lint_err}). The file was "
"NOT created or modified. Fix the content and retry."
)
)
# Capture pre-write content. Two consumers want it:
#
# 1. The lint-delta layer (for in-process linters like ast.parse
@ -1352,7 +1414,6 @@ class ShellFileOperations(FileOperations):
# the UNION of in-process lint coverage and LSP coverage. For
# extensions outside both sets (binaries, opaque formats),
# skipping the read keeps the hot path fast.
ext = os.path.splitext(path)[1].lower()
pre_content: Optional[str] = None
want_pre = ext in LINTERS_INPROC or self._lsp_handles_extension(ext)
if want_pre: