fix(file-tools): native writes create the temp file at its final mode before any byte lands

The in-process atomic_write created its temp 0o666 (umask-filtered, 0644 under 022), wrote the new
contents, and only then chmod-ed it to the target's mode, so a rewritten 0600 file's new contents sat
world-readable in the target's directory until the chmod. The shell path it replaces used mktemp's
0600 temp, which never exposed them.

An existing target's temp is now created 0o600 and fchmod-ed to the target's mode before the write
(best effort, as before: a failed fchmod leaves it 0600, never wider). A new target keeps the umask
default (O_CREAT 0o666), the same mode the shell path's chmod "=rw" gave it.

Probe (mode of the temp at the first write, umask 022, existing 0600 target): 0o644 on f980eca8e4,
0o600 now; final modes unchanged (0600 existing, 0644 new).

tests/agent/lsp/_mock_lsp_server.py

@@ -34,8 +34,9 @@ Behaviour (all behaviours selectable via env var ``MOCK_LSP_SCRIPT``):
   no ``version`` field in any publishDiagnostics (the client credits
   each push with its current document version at receipt).  Push-only:
   the pull endpoint rejects.
-- ``"by_content"`` — push-only; after ``MOCK_LSP_PUSH_DELAY`` (default 0.4s) pushes one error iff the
-  synced text contains ``bad``, else empty diagnostics.
+- ``"by_content"`` — push-only; pushes one error iff the synced text contains ``bad``, else empty
+  diagnostics.  Holds every push until the file named by ``MOCK_LSP_GATE`` exists (a server slower
+  than the write, released by the test), else after ``MOCK_LSP_PUSH_DELAY`` (default 0.4s).
 - ``"incremental"`` — apply ranged edits as UTF-16 and expose the server's
   document mirror through hover, for synchronization contract tests.
 - ``"clean_eof"`` — closes stdout after ``didOpen`` but keeps the
@@ -58,12 +59,18 @@ import time
 
 
 def _push_by_content(params, is_change, error_diag):
-    """``by_content``: after MOCK_LSP_PUSH_DELAY, one error iff the synced text contains "bad" (so a
-    baseline of the pre-edit text differs from the edit's verdict)."""
+    """``by_content``: once MOCK_LSP_GATE exists (else after MOCK_LSP_PUSH_DELAY), one error iff the
+    synced text contains "bad" (so a baseline of the pre-edit text differs from the edit's verdict)."""
     td = params.get("textDocument") or {}
     text = (params.get("contentChanges") or [{}])[-1].get("text", "") if is_change else td.get("text")
-    # health: allow HX013 -- the delay IS the behaviour under test (a server slower than the write)
-    time.sleep(float(os.environ.get("MOCK_LSP_PUSH_DELAY", "0.4")))
+    gate = os.environ.get("MOCK_LSP_GATE")
+    deadline = time.monotonic() + 60
+    while gate and not os.path.exists(gate) and time.monotonic() < deadline:
+        # health: allow HX013 -- a separate process can only poll the test's gate file
+        time.sleep(0.01)
+    if not gate:
+        # health: allow HX013 -- the delay IS the behaviour under test (a server slower than the write)
+        time.sleep(float(os.environ.get("MOCK_LSP_PUSH_DELAY", "0.4")))
     write_message({"jsonrpc": "2.0", "method": "textDocument/publishDiagnostics",
                    "params": {"uri": td.get("uri", ""), "version": td.get("version", 0),
                               "diagnostics": error_diag if "bad" in (text or "") else []}})

tests/agent/lsp/test_backend_gate.py

@@ -62,7 +62,7 @@ def test_edit_check_started_for_local_env(tmp_path, monkeypatch):
         def enabled_for(self, path):
             return True
 
-        def start_edit_check(self, path, *, pre_text, inherited):
+        def start_edit_check(self, path, *, pre_text, owner=None, carry=()):
             started.append((path, pre_text))
 
     monkeypatch.setattr("agent.lsp.get_service", lambda: FakeService())

tests/agent/lsp/test_write_never_waits_on_lsp.py

@@ -1,8 +1,10 @@
-"""Invariant: a write never waits on the language server, and its diagnostics still reach the model.
+"""Invariant: a write never waits on the language server, and its diagnostics still reach the model
+(the session that made the edit, and only that session).
 
 The write path used to block on a pre-write baseline wait plus a post-write wait, each up to
 ``lsp.wait_timeout``; a server that stays silent (typescript-language-server on a clean edit) ran
-both to the deadline on every edit.
+both to the deadline on every edit.  The servers here are held until the test releases them, so the
+assertions are about order (the tool result comes back before any verdict exists), not speed.
 """
 from __future__ import annotations
 
@@ -11,34 +13,39 @@ import time
 
 import pytest
 
+from agent.lsp import deferred
 from agent.lsp.manager import LSPService
 from model_tools import handle_function_call
 from tests.agent.lsp.test_service import _install_mock_server
+from tests.waiting import wait_until
+
+_WAIT = 20.0  # lsp.wait_timeout: a write that waited on the server would take 2x this per edit
 
 
 @pytest.fixture
 def mock_workspace(monkeypatch, tmp_path):
+    created = []
+
     def install(script, **env):
         repo = tmp_path / "repo"
         repo.mkdir()
         (repo / ".git").mkdir()
         (repo / "pyproject.toml").write_text("", encoding="utf-8")
         monkeypatch.chdir(str(repo))
+        gate = tmp_path / "release-diagnostics"
+        monkeypatch.setenv("MOCK_LSP_GATE", str(gate))
         for key, value in env.items():
             monkeypatch.setenv(key, value)
         gen = _install_mock_server(monkeypatch, script, "pyright")
         next(gen)
-        svc = LSPService(enabled=True, wait_mode="document", wait_timeout=3.0, install_strategy="manual",
+        svc = LSPService(enabled=True, wait_mode="document", wait_timeout=_WAIT, install_strategy="manual",
                          trusted_workspaces=[str(repo)])
         monkeypatch.setattr("agent.lsp.get_service", lambda: svc)
-        return repo, svc, gen
-    created = []
-
-    def _install(script, **env):
-        created.append(install(script, **env))
-        return created[-1][:2]
-    yield _install
-    for _repo, svc, gen in created:
+        created.append((svc, gen, gate))
+        return repo, gate
+    yield install
+    for svc, gen, gate in created:
+        gate.touch()  # release any held push so shutdown is not stuck behind it
         svc.shutdown()
         try:
             next(gen)
@@ -47,42 +54,43 @@ def mock_workspace(monkeypatch, tmp_path):
 
 
 def _patch(path, old, new, session="S-lsp"):
-    handle_function_call("read_file", {"path": str(path)}, task_id="t-lsp", session_id=session)
+    handle_function_call("read_file", {"path": str(path)}, task_id="t-" + session, session_id=session)
     started = time.monotonic()
     out = handle_function_call("patch", {"path": str(path), "old_string": old, "new_string": new},
-                               task_id="t-lsp", session_id=session)
+                               task_id="t-" + session, session_id=session)
     return json.loads(out), time.monotonic() - started
 
 
+def _settled(*sessions):
+    with deferred._LOCK:
+        return not any(deferred._PENDING.get(s) for s in sessions)
+
+
 def test_silent_server_does_not_hold_the_write(mock_workspace):
-    """A server that never answers costs the write nothing: two edits stay far below one wait budget."""
-    repo, _svc = mock_workspace("silent")
+    """A server that never answers costs the write nothing: each edit returns long before one wait
+    budget (the old path waited the budget twice per edit)."""
+    repo, _gate = mock_workspace("silent")
     target = repo / "a.py"
     target.write_text("x = 1\n", encoding="utf-8")
     for old, new in (("x = 1", "x = 2"), ("x = 2", "x = 3")):
         result, elapsed = _patch(target, old, new)
         assert result["success"] is True, result
-        assert elapsed < 1.0, f"write blocked {elapsed:.2f}s on a silent language server"
+        assert elapsed < _WAIT / 2, f"write blocked {elapsed:.2f}s on a silent language server"
     assert target.read_text(encoding="utf-8-sig") == "x = 3\n"
 
 
 def test_late_diagnostics_reach_the_session(mock_workspace):
-    """A verdict slower than the write is delivered to the editing session afterwards (attached to
-    its next tool result), and only the error the edit introduced is reported."""
-    repo, _svc = mock_workspace("by_content", MOCK_LSP_PUSH_DELAY="0.4")
+    """A verdict that arrives after the write returned is delivered to the editing session afterwards
+    (attached to its next tool result), and only the error the edit introduced is reported."""
+    repo, gate = mock_workspace("by_content")
     target = repo / "b.py"
     target.write_text("x = 1\n", encoding="utf-8")
     result, elapsed = _patch(target, "x = 1", "x = bad")
-    assert result["success"] is True and elapsed < 0.4, (result, elapsed)
-    assert "lsp_diagnostics" not in result
-    from agent.lsp import deferred
-    notes = []
-    deadline = time.monotonic() + 10
-    while not notes and time.monotonic() < deadline:
-        notes = deferred.pop_notes("S-lsp")
-        time.sleep(0.05)
-    assert notes, "the late verdict never reached the session"
-    (path, text), = notes
+    assert result["success"] is True and "lsp_diagnostics" not in result, result
+    assert elapsed < _WAIT / 2, f"write blocked {elapsed:.2f}s on a held language server"
+    gate.touch()
+    (path, text), = wait_until(lambda: deferred.pop_notes("S-lsp"), timeout=_WAIT,
+                               what="the late verdict reaching the session")
     assert path == str(target) and "MOCK001" in text and "earlier edit" in text
 
 
@@ -95,12 +103,13 @@ def test_late_verdict_reaches_the_model_before_the_turn_ends(mock_workspace, mon
     from agent.turn_stop_gates import apply_stop_gates
 
     monkeypatch.setenv("HERMES_VERIFY_ON_STOP", "0")
-    repo, _svc = mock_workspace("by_content", MOCK_LSP_PUSH_DELAY="0.4")
+    repo, gate = mock_workspace("by_content")
     target = repo / "c.py"
     target.write_text("x = 1\n", encoding="utf-8")
     result, _elapsed = _patch(target, "x = 1", "x = bad", session="S-end")
     assert result["success"] is True and "lsp_diagnostics" not in result, result
-    agent = SimpleNamespace(session_id="S-end", _current_task_id="t-lsp",
+    gate.touch()
+    agent = SimpleNamespace(session_id="S-end", _current_task_id="t-S-end",
                             _interim_content_was_streamed=lambda _text: False,
                             _emit_interim_assistant_message=lambda _msg: None,
                             _flush_messages_to_session_db=lambda *_a: None)
@@ -110,3 +119,20 @@ def test_late_verdict_reaches_the_model_before_the_turn_ends(mock_workspace, mon
                                pending_verification_response=None, pending_verification_response_previewed=False)
     assert verdict.continue_turn, "the turn ended with the edit's verdict still undelivered"
     assert messages[-1]["role"] == "user" and "MOCK001" in messages[-1]["content"], messages[-1]
+
+
+def test_diagnostics_stay_with_the_session_that_made_the_edit(mock_workspace):
+    """Two sessions edit one file while the first check is still running: the error session A
+    introduced is reported to A, and B (whose edit kept it, not introduced it) hears nothing."""
+    repo, gate = mock_workspace("by_content")
+    target = repo / "d.py"
+    target.write_text("x = 1\n", encoding="utf-8")
+    result_a, _ = _patch(target, "x = 1", "x = bad", session="S-A")
+    result_b, _ = _patch(target, "x = bad", "x = bad\ny = 2", session="S-B")
+    assert result_a["success"] is True and result_b["success"] is True, (result_a, result_b)
+    gate.touch()
+    wait_until(lambda: _settled("S-A", "S-B"), timeout=_WAIT, what="both edit checks settling")
+    notes_b = deferred.pop_notes("S-B")
+    notes_a = deferred.pop_notes("S-A")
+    assert not notes_b, f"session B was told about session A's error: {notes_b}"
+    assert notes_a and "MOCK001" in notes_a[0][1], f"session A never heard about its own error: {notes_a}"

tests/tools/test_write_verification.py

@@ -56,6 +56,7 @@ class TestWriteVerification:
             r = json.loads(write_file_tool(str(f), "actual content\n", task_id="t-wv"))
         assert "error" in r
 
+    @pytest.mark.platforms("posix")  # the native write path is POSIX-only (_native_read_enabled)
     def test_native_hash_mismatch_is_hard_error(self, workdir):
         f = workdir / "bad_native.txt"
         with mock_patch("tools.file_operations_native.sha256_of", return_value="0" * 64):

tools/file_operations_native.py

@@ -56,7 +56,11 @@ def atomic_write(path: str, data: bytes) -> Optional[str]:
     Same contract as the shell script: a symlink target is written through (the link survives),
     parent directories are created, the temp file lives in the target's directory so the rename is
     atomic, an existing target's mode is kept and a new one gets the umask default (``O_CREAT`` with
-    0o666 applies the process umask itself), and the temp is removed on every failure."""
+    0o666 applies the process umask itself), and the temp is removed on every failure.
+
+    The temp holds its final mode before the first byte lands: an existing target's temp is created
+    0o600 and ``fchmod``-ed to the target's mode, so a private file's new contents are never readable
+    through a wider temp (the shell script's ``mktemp`` 0600 temp gave the same guarantee)."""
     target = os.path.realpath(path) if os.path.islink(path) else path
     parent = os.path.dirname(target) or "."
     try:
@@ -66,17 +70,18 @@ def atomic_write(path: str, data: bytes) -> Optional[str]:
         except FileNotFoundError:
             mode = None
         tmp = os.path.join(parent, f".hermes-tmp.{secrets.token_hex(6)}")
-        fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_CLOEXEC", 0), 0o666)
+        fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_CLOEXEC", 0),
+                     0o666 if mode is None else 0o600)
     except OSError as e:
         return f"{e.strerror or e}: {e.filename or target}"
     try:
-        with open(fd, "wb") as fh:
-            fh.write(data)
         if mode is not None:
             try:
-                os.chmod(tmp, mode)
+                os.fchmod(fd, mode)  # before any content: the bytes never sit in a wider file
             except OSError:
-                pass  # best effort, as in the shell script
+                pass  # best effort, as in the shell script (the temp then stays 0o600)
+        with open(fd, "wb") as fh:
+            fh.write(data)
         os.replace(tmp, target)
     except OSError as e:
         with contextlib.suppress(OSError):