fix(trace): make trace records and their sidecars durable (#662)

TraceWriter.write() only flushed, so a host crash or container kill lost
the last records. Each record is now fsynced, and — because a fsynced
record must never point at a blob that is not on disk — sidecar writes go
through temp-write -> fsync -> os.replace -> parent-directory fsync
before the referencing record is written. New trace.jsonl and sidecar
directory entries are also fsynced.

fsync failures (unsupported filesystem) log one warning per writer and
degrade to flush-only rather than breaking the run.

agent/src/agent/trace.py

@@ -1,19 +1,25 @@
 """TraceWriter: crash-safe JSONL trace writer.
 
-One JSON record per line; append + flush keeps the trace useful after crashes.
+One JSON record per line; each write is flushed and fsynced so the record
-Large fields are written to sidecar files so traces can preserve full content
+survives an immediate host crash. Large fields are written to sidecar files
-without turning every CLI/history read into a giant file load.
+(temp-write -> fsync -> ``os.replace`` -> directory fsync) BEFORE the record
+that references them is fsynced, so a durable record never points at a
+missing or truncated sidecar. Sidecars keep traces complete without turning
+every CLI/history read into a giant file load.
 """
 
 from __future__ import annotations
 
 import hashlib
 import json
+import logging
 import os
 import time
 from pathlib import Path
 from typing import Any, Dict, Iterable, List, Optional
 
+logger = logging.getLogger(__name__)
+
 
 def _env_int(name: str, default: int) -> int:
     """Read a positive integer environment variable with a safe fallback.
@@ -70,12 +76,28 @@ class TraceWriter:
             dir_path: Directory where trace files are written.
         """
         self.dir_path = dir_path
-        self.dir_path.mkdir(parents=True, exist_ok=True)
         self.path = self.dir_path / "trace.jsonl"
+        self._fsync_warned = False
+        self.dir_path.mkdir(parents=True, exist_ok=True)
+        created = not self.path.exists()
         self._file = open(self.path, "a", encoding="utf-8")
+        try:
+            if created:
+                # Make the new trace.jsonl directory entry itself durable.
+                self._fsync_dir(self.dir_path)
+        except Exception:
+            self._file.close()
+            raise
 
     def write(self, entry: Dict[str, Any]) -> None:
-        """Write a trace record.
+        """Write a trace record, flushed and fsynced for crash safety.
+
+        ``flush`` alone only moves bytes into the OS page cache; a host
+        crash or container kill would lose the last records. Trace volume
+        is low (one entry per LLM/tool event), so one fsync per record is
+        an acceptable cost for post-mortem visibility. If fsync fails
+        (e.g. unsupported filesystem), a warning is logged once and writes
+        continue flush-only.
 
         Args:
             entry: Trace entry; a ``ts`` field is added automatically.
@@ -84,6 +106,10 @@ class TraceWriter:
             entry["ts"] = time.time()
         self._file.write(json.dumps(entry, ensure_ascii=False) + "\n")
         self._file.flush()
+        try:
+            os.fsync(self._file.fileno())
+        except OSError as exc:
+            self._warn_fsync_failure(exc, self.path)
 
     def write_text_entry(
         self,
@@ -167,20 +193,90 @@ class TraceWriter:
         offload_dir_name: str,
         preview_field: str | None = None,
     ) -> None:
-        """Attach text inline or as a sidecar path."""
+        """Attach text inline or as a sidecar path.
+
+        The sidecar is written atomically and durably here, BEFORE the
+        caller writes (and fsyncs) the record that references it, so a
+        crash never leaves a durable record pointing at a missing or
+        truncated sidecar.
+        """
         if len(value) <= threshold:
             entry[field] = value
             return
 
         offload_dir = self.dir_path / offload_dir_name
-        offload_dir.mkdir(parents=True, exist_ok=True)
+        if not offload_dir.is_dir():
+            offload_dir.mkdir(parents=True, exist_ok=True)
+            # Make the new sidecar directory entry itself durable.
+            self._fsync_dir(self.dir_path)
         digest = hashlib.sha256(f"{offload_kind}\0{value}".encode("utf-8")).hexdigest()
         path = offload_dir / f"{digest[:24]}.txt"
-        path.write_text(value, encoding="utf-8")
+        self._write_sidecar_durable(path, value)
         entry[f"{field}_path"] = f"{offload_dir_name}/{path.name}"
         entry[preview_field or f"{field}_preview"] = value[:OFFLOAD_PREVIEW_CHARS]
         entry[f"{field}_size"] = len(value)
 
+    def _write_sidecar_durable(self, path: Path, value: str) -> None:
+        """Atomically and durably write a sidecar blob.
+
+        Write sequence: temp file in same dir -> fsync -> ``os.replace`` ->
+        parent-directory fsync. A crash at any step leaves either no sidecar
+        or a complete one, never a partial write under the final name.
+
+        Args:
+            path: Final sidecar path.
+            value: Full text to persist.
+
+        Raises:
+            OSError: When the temp file cannot be written or renamed. The
+                caller must not write the referencing record in that case.
+        """
+        tmp = path.with_name(f".{path.name}.{os.getpid()}.tmp")
+        fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
+        try:
+            os.write(fd, value.encode("utf-8"))
+            try:
+                os.fsync(fd)
+            except OSError as exc:
+                self._warn_fsync_failure(exc, tmp)
+        finally:
+            os.close(fd)
+        os.replace(tmp, path)
+        self._fsync_dir(path.parent)
+
+    def _fsync_dir(self, directory: Path) -> None:
+        """Fsync a directory so new/renamed entries survive a crash.
+
+        Args:
+            directory: Directory whose entry table should be flushed.
+        """
+        try:
+            dir_fd = os.open(directory, os.O_RDONLY)
+        except OSError:
+            return
+        try:
+            os.fsync(dir_fd)
+        except OSError as exc:
+            self._warn_fsync_failure(exc, directory)
+        finally:
+            os.close(dir_fd)
+
+    def _warn_fsync_failure(self, exc: OSError, target: Path) -> None:
+        """Log the first fsync failure for this writer, then stay quiet.
+
+        Args:
+            exc: The fsync error.
+            target: File or directory whose fsync failed.
+        """
+        if self._fsync_warned:
+            return
+        self._fsync_warned = True
+        logger.warning(
+            "trace fsync failed on %s (%s); trace durability degraded to flush-only for this writer",
+            target,
+            exc,
+        )
+
     @staticmethod
     def read(
         dir_path: Path,

agent/tests/test_trace_writer.py

@@ -3,6 +3,9 @@
 from __future__ import annotations
 
 import json
+import logging
+import os
+import stat
 from pathlib import Path
 
 import pytest
@@ -127,3 +130,138 @@ def test_find_trace_dir_prefers_sessions_then_runs(tmp_path: Path) -> None:
 
     assert TraceWriter.find_trace_dir("abc", runs_dir=runs, sessions_dir=sessions) == session_dir
     assert TraceWriter.find_trace_dir("missing", runs_dir=runs, sessions_dir=sessions) is None
+
+
+def test_write_calls_fsync_for_crash_safety(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+) -> None:
+    """``write`` flushes AND fsyncs so the last record survives a hard crash.
+
+    Regression for the historical "crash-safe" docstring claim that only
+    flushed. Without ``os.fsync`` the kernel page cache can lose the last
+    record on a host kill / power event.
+    """
+    trace = TraceWriter(tmp_path)
+    fsync_fds: list[int] = []
+    real_fsync = os.fsync
+
+    def _tracking_fsync(fd: int) -> None:
+        fsync_fds.append(fd)
+        real_fsync(fd)
+
+    monkeypatch.setattr(trace_mod.os, "fsync", _tracking_fsync)
+    try:
+        trace.write({"type": "answer", "iter": 1, "content": "ok"})
+        assert fsync_fds == [trace._file.fileno()]
+    finally:
+        trace.close()
+
+
+def test_new_trace_file_fsyncs_parent_directory(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+) -> None:
+    """Creating trace.jsonl durably records the new directory entry."""
+    dir_fsyncs: list[int] = []
+    real_fsync = os.fsync
+
+    def _tracking_fsync(fd: int) -> None:
+        if stat.S_ISDIR(os.fstat(fd).st_mode):
+            dir_fsyncs.append(fd)
+        real_fsync(fd)
+
+    monkeypatch.setattr(trace_mod.os, "fsync", _tracking_fsync)
+    trace = TraceWriter(tmp_path)
+    trace.close()
+    assert len(dir_fsyncs) == 1
+
+    dir_fsyncs.clear()
+    reopened = TraceWriter(tmp_path)
+    reopened.close()
+    assert dir_fsyncs == []
+
+
+def test_fsync_oserror_warns_once_and_keeps_writing(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """fsync failure degrades to flush-only with one warning, never silence."""
+    trace = TraceWriter(tmp_path)
+
+    def _failing_fsync(fd: int) -> None:
+        raise OSError("fsync unsupported")
+
+    monkeypatch.setattr(trace_mod.os, "fsync", _failing_fsync)
+    with caplog.at_level(logging.WARNING, logger="src.agent.trace"):
+        trace.write({"type": "answer", "iter": 1, "content": "a"})
+        trace.write({"type": "answer", "iter": 2, "content": "b"})
+    trace.close()
+
+    warnings = [rec for rec in caplog.records if "fsync" in rec.message]
+    assert len(warnings) == 1
+    assert [entry["iter"] for entry in _raw_entries(tmp_path)] == [1, 2]
+
+
+def test_sidecar_write_is_atomic_and_leaves_no_temp_files(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+) -> None:
+    """Sidecars are written temp-file-then-rename with no leftover litter."""
+    monkeypatch.setattr(trace_mod, "TOOL_RESULT_OFFLOAD_THRESHOLD", 8)
+    replace_targets: list[str] = []
+    real_replace = os.replace
+
+    def _tracking_replace(src: object, dst: object) -> None:
+        replace_targets.append(str(dst))
+        real_replace(src, dst)
+
+    monkeypatch.setattr(trace_mod.os, "replace", _tracking_replace)
+    trace = TraceWriter(tmp_path)
+    trace.write_tool_result(
+        call_id="call-1",
+        result="large result body",
+        tool_name="tool",
+        status="ok",
+        elapsed_ms=1,
+        iteration=1,
+    )
+    trace.close()
+
+    [entry] = _raw_entries(tmp_path)
+    sidecar = tmp_path / entry["result_path"]
+    assert sidecar.read_text(encoding="utf-8") == "large result body"
+    assert replace_targets == [str(sidecar)]
+    assert [p.name for p in sidecar.parent.iterdir()] == [sidecar.name]
+
+
+def test_record_never_references_missing_sidecar_on_crash(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+) -> None:
+    """A failed sidecar rename must abort BEFORE the referencing record is written.
+
+    Otherwise a crash could persist a durable record whose ``result_path``
+    target is missing or truncated.
+    """
+    monkeypatch.setattr(trace_mod, "TOOL_RESULT_OFFLOAD_THRESHOLD", 8)
+    trace = TraceWriter(tmp_path)
+
+    def _failing_replace(src: object, dst: object) -> None:
+        raise OSError("simulated crash before sidecar rename")
+
+    monkeypatch.setattr(trace_mod.os, "replace", _failing_replace)
+    with pytest.raises(OSError, match="simulated crash"):
+        trace.write_tool_result(
+            call_id="call-1",
+            result="large result body",
+            tool_name="tool",
+            status="ok",
+            elapsed_ms=1,
+            iteration=1,
+        )
+    trace.close()
+
+    assert _raw_entries(tmp_path) == []
+    assert list((tmp_path / "tool-results").glob("*.txt")) == []