fix(shell): the job store could crash a supervisor on Windows with WinError 5 (2.118.1)

CHANGELOG.md

@@ -4,6 +4,38 @@ All notable changes to Code Context Control (C3) are documented here.
 The format is loosely based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [2.118.1] - 2026-09-04
+
+### Fixed — the job store could crash a supervisor on Windows with WinError 5
+
+`services/shell_output._write_json_atomic` is the publish step for every
+`c3_shell_job` status update. It used a single shared temp name and attempted
+`os.replace` exactly once, so a supervisor could die before running anything —
+reporting `failed before running` for a command that was fine. It surfaced as
+an intermittent CI failure on windows-latest (`test_shell_jobs_s3.py`,
+`PermissionError: [WinError 5] Access is denied`), which is the visible half;
+the invisible half is that the same race can lose a job on a real Windows box.
+
+Three defects, all now covered by tests that fail against the old code:
+
+- **A shared temp name.** `<name>.tmp` is identical for every writer, so the
+  parent creating the job record and the supervisor immediately saving status
+  raced on the temp itself — one mid-write while the other tried to publish
+  it. The suffix is now a pid **and** a random token: a pid alone is not
+  enough, because this store is written from a threaded server and threads
+  share a pid. (The concurrency test caught that second-order case.)
+- **No retry.** Another process holding a read handle on the target — AV
+  scanner, Search indexer, a concurrent reader — makes `os.replace` raise
+  transiently. One attempt turned that into a lost write. Now four attempts
+  with exponential backoff, and a genuinely persistent error still raises
+  rather than being swallowed.
+- **Orphaned temp files.** A failed replace abandoned its temp in the store
+  forever. Cleaned up on every failure path.
+
+Same construction as `cli/_hook_utils._atomic_write_json`, which already had
+it. The duplication is deliberate — a PreToolUse hook subprocess imports
+nothing from `services` — and is now noted in both.
+
 ## [2.118.0] - 2026-09-04
 
 ### Added — one approval can cover a whole rule for the session, not one file

cli/c3.py

@@ -92,7 +92,7 @@
 # Config
 CONFIG_DIR = ".c3"
 CONFIG_FILE = ".c3/config.json"
-__version__ = "2.118.0"
+__version__ = "2.118.1"
 
 
 def _compress_file_cli(compressor, path, mode="smart", **kw):

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "code-context-control"
-version = "2.118.0"
+version = "2.118.1"
 description = "Local MCP code-intelligence for AI coding tools: surgical search/read/edit, agent-config version history, path-level access + masking guards, and a multi-project hub."
 readme = "README.md"
 requires-python = ">=3.10"

services/shell_output.py

@@ -960,10 +960,64 @@ def _load_meta(path: Path) -> OutputMeta | None:
         return None
 
 
+#: os.replace retries for when Windows briefly holds a handle on the target.
+#: Same construction as ``cli/_hook_utils._atomic_write_json``; the hook layer
+#: keeps its own copy on purpose (a PreToolUse subprocess imports nothing from
+#: services), so the two are duplicated deliberately rather than by accident.
+_REPLACE_ATTEMPTS = 4
+_REPLACE_BACKOFF_S = 0.02
+
+
 def _write_json_atomic(path: Path, data: dict) -> None:
-    tmp = path.with_name(path.name + ".tmp")
-    tmp.write_text(json.dumps(data, ensure_ascii=False, indent=1), encoding="utf-8")
-    os.replace(tmp, path)
+    """Write JSON durably: unique temp file in the same directory + os.replace.
+
+    Two bugs this fixes, both observed on Windows CI (2026-09-04, job store):
+
+    1. **A shared temp name.** ``<name>.tmp`` is the same string for every
+       writer, so two processes writing the SAME job file — the parent that
+       creates the record and the supervisor that immediately saves status —
+       raced on the temp itself. One would be mid-write while the other tried
+       to publish it, and ``os.replace`` raised
+       ``PermissionError: [WinError 5] Access is denied``. The suffix is a pid
+       AND a random token, not a pid alone: this store is written from a
+       threaded server, and threads share a pid, so ``.tmp<pid>`` collides
+       exactly as the shared name did. With a per-WRITE temp the race is only
+       on the target, where ``os.replace`` is genuinely atomic.
+    2. **No retry.** Even with unique temps, another process holding a read
+       handle on the target (an AV scanner, the Search indexer, a concurrent
+       reader) makes ``os.replace`` raise transiently. A single attempt turned
+       that into a crashed supervisor — the job never ran, and the caller got
+       "failed before running" for a command that was fine.
+
+    A partial write can still be published without fsync, but this store is
+    ephemeral spill state rather than durable config, so the cost of an fsync
+    per status update is not worth paying here.
+    """
+    path.parent.mkdir(parents=True, exist_ok=True)
+    tmp = path.with_name(
+        f"{path.name}.tmp{os.getpid()}-{secrets.token_hex(4)}")
+    try:
+        tmp.write_text(json.dumps(data, ensure_ascii=False, indent=1),
+                       encoding="utf-8")
+        last_exc: OSError | None = None
+        for attempt in range(_REPLACE_ATTEMPTS):
+            try:
+                os.replace(tmp, path)
+                return
+            except PermissionError as exc:
+                last_exc = exc
+                if attempt < _REPLACE_ATTEMPTS - 1:
+                    time.sleep(_REPLACE_BACKOFF_S * (2 ** attempt))
+        if last_exc is not None:
+            raise last_exc
+    finally:
+        # A successful replace already consumed tmp, so this only fires on the
+        # failure paths — otherwise abandoned temps accumulate in the store.
+        try:
+            if tmp.exists():
+                tmp.unlink()
+        except OSError:
+            pass
 
 
 def _move(src: Path, dst: Path) -> None:

tests/test_shell_output.py

@@ -723,5 +723,103 @@ def test_end_to_end_with_spaces_and_unicode(self):
         self.assertEqual(self.store.list(project_path=project, session_id=session), [])
 
 
+class TestAtomicWrite(unittest.TestCase):
+    """`_write_json_atomic` — the job store's publish step.
+
+    Both cases below fail against the pre-2.118.1 version, which used a single
+    shared `<name>.tmp` and attempted `os.replace` exactly once. That is what
+    crashed a supervisor on Windows CI with WinError 5 and reported "failed
+    before running" for a command that was fine.
+    """
+
+    def setUp(self):
+        self._tmp = tempfile.TemporaryDirectory()
+        self.path = Path(self._tmp.name) / "job.json"
+
+    def tearDown(self):
+        self._tmp.cleanup()
+
+    def test_temp_name_is_per_process(self):
+        """Two writers must not collide on the temp file itself."""
+        seen = []
+        real = os.replace
+
+        def spy(src, dst):
+            seen.append(Path(src).name)
+            return real(src, dst)
+
+        with patch.object(so.os, "replace", spy):
+            so._write_json_atomic(self.path, {"a": 1})
+        self.assertEqual(len(seen), 1)
+        self.assertNotEqual(seen[0], "job.json.tmp",
+                            "shared temp name — concurrent writers will race")
+        self.assertTrue(seen[0].startswith("job.json.tmp"))
+        self.assertIn(str(os.getpid()), seen[0])
+        # …and unique per WRITE, not just per process: this store is written
+        # from a threaded server, and threads share a pid.
+        seen.clear()
+        with patch.object(so.os, "replace", spy):
+            so._write_json_atomic(self.path, {"a": 2})
+            so._write_json_atomic(self.path, {"a": 3})
+        self.assertEqual(len(set(seen)), 2, f"temp name reused: {seen}")
+
+    def test_transient_permission_error_is_retried(self):
+        calls = {"n": 0}
+        real = os.replace
+
+        def flaky(src, dst):
+            calls["n"] += 1
+            if calls["n"] < 3:          # two transient failures, then success
+                raise PermissionError(5, "Access is denied")
+            return real(src, dst)
+
+        with patch.object(so.os, "replace", flaky):
+            so._write_json_atomic(self.path, {"ok": True})
+        self.assertEqual(calls["n"], 3)
+        self.assertEqual(json.loads(self.path.read_text(encoding="utf-8")),
+                         {"ok": True})
+
+    def test_a_persistent_permission_error_still_raises(self):
+        """Retrying must not swallow a real, non-transient failure."""
+        def always(src, dst):
+            raise PermissionError(5, "Access is denied")
+
+        with patch.object(so.os, "replace", always):
+            with self.assertRaises(PermissionError):
+                so._write_json_atomic(self.path, {"a": 1})
+
+    def test_no_temp_file_is_left_behind_when_replace_fails(self):
+        def always(src, dst):
+            raise PermissionError(5, "Access is denied")
+
+        with patch.object(so.os, "replace", always):
+            with self.assertRaises(PermissionError):
+                so._write_json_atomic(self.path, {"a": 1})
+        leftovers = list(self.path.parent.glob("job.json.tmp*"))
+        self.assertEqual(leftovers, [], f"orphaned temp files: {leftovers}")
+
+    def test_concurrent_writers_all_publish_valid_json(self):
+        """The end-to-end property: N threads hammering one path never leave
+        the target truncated or half-written."""
+        import threading
+        errors = []
+
+        def writer(i):
+            try:
+                for _ in range(20):
+                    so._write_json_atomic(self.path, {"writer": i})
+            except Exception as exc:      # noqa: BLE001 — surfaced below
+                errors.append(exc)
+
+        threads = [threading.Thread(target=writer, args=(i,)) for i in range(6)]
+        for t in threads:
+            t.start()
+        for t in threads:
+            t.join()
+        self.assertEqual(errors, [])
+        self.assertIn("writer", json.loads(self.path.read_text(encoding="utf-8")))
+        self.assertEqual(list(self.path.parent.glob("job.json.tmp*")), [])
+
+
 if __name__ == "__main__":
     unittest.main()