fix(hosts): tolerate Windows ctime skew in the rollout same-read fence

loopx/codex_app_thread_activity.py

@@ -246,7 +246,13 @@ def _rollout_activity(
                 if Path(os.path.realpath(path)) != path or (end.st_dev, end.st_ino) != (named.st_dev, named.st_ino) or end.st_size < stat.st_size:
                     _rollout_cache.pop(binding, None)
                     return HostThreadActivity.unknown(HostThreadUnknownReason.STORE_UNAVAILABLE)
-                if end.st_size == stat.st_size and (end.st_mtime_ns != stat.st_mtime_ns or end.st_ctime_ns != stat.st_ctime_ns):
+                # Windows ctime-only skew between two stats of one open
+                # descriptor is metadata noise, not a same-size rewrite; drop
+                # ctime from this one-read fence only when both observations
+                # carry a usable volume/file index.
+                fence_ctime = os.name != "nt" or not (stat.st_dev and stat.st_ino and end.st_dev and end.st_ino)
+                if end.st_size == stat.st_size and (end.st_mtime_ns != stat.st_mtime_ns
+                        or (fence_ctime and end.st_ctime_ns != stat.st_ctime_ns)):
                     _rollout_cache.pop(binding, None)
                     return HostThreadActivity.unknown(HostThreadUnknownReason.RECORD_UNRECOGNIZED)
                 cursor.save(cache_root)

tests/test_host_thread_activity.py

@@ -1153,3 +1153,53 @@ def observe(thread_ids: Any) -> dict[str, HostThreadActivity]:
         expectations=lambda scopes: pytest.fail("no observed lane may ask for a cadence"),
     )
     assert unattached == expected
+
+
+@pytest.mark.parametrize("unavailable_identity", [None, "st_dev", "st_ino"])
+@pytest.mark.parametrize("platform", ["nt", "posix"])
+@pytest.mark.parametrize("field", ["st_mtime_ns", "st_ctime_ns"])
+def test_same_size_read_fence_metadata_policy(home, monkeypatch, platform, field, unavailable_identity):
+    rollout = home.thread("t", [_event(T2, "task_complete")])
+    rollout_stat = rollout.stat()
+    assert rollout_stat.st_dev and rollout_stat.st_ino
+    original_fstat = os.fstat
+    original_stat = Path.stat
+    descriptor_observations = 0
+
+    def partial_identity(observed):
+        if unavailable_identity is None or (observed.st_dev, observed.st_ino) != (
+            rollout_stat.st_dev, rollout_stat.st_ino
+        ):
+            return observed
+        # Unavailable volume/file indexes are injected at every rollout stat.
+        fields = {name: getattr(observed, name) for name in dir(observed) if name.startswith("st_")}
+        fields[unavailable_identity] = 0
+        return SimpleNamespace(**fields)
+
+    def path_stat(self, *args, **kwargs):
+        return partial_identity(original_stat(self, *args, **kwargs))
+
+    def descriptor_stat(fd):
+        nonlocal descriptor_observations
+        observed = original_fstat(fd)
+        if (observed.st_dev, observed.st_ino) != (rollout_stat.st_dev, rollout_stat.st_ino):
+            return observed
+        descriptor_observations += 1
+        observed = partial_identity(observed)
+        if descriptor_observations != 2:
+            return observed
+        fields = {name: getattr(observed, name) for name in dir(observed) if name.startswith("st_")}
+        fields[field] += 1
+        return SimpleNamespace(**fields)
+
+    monkeypatch.setattr(Path, "stat", path_stat)
+    monkeypatch.setattr(codex_activity, "os", SimpleNamespace(**{
+        **vars(os), "name": platform, "fstat": descriptor_stat,
+    }))
+    result = observe_codex_threads(["t"], homes=[home.root])["t"]
+    if platform == "nt" and field == "st_ctime_ns" and unavailable_identity is None:
+        assert result.state is HostThreadState.IDLE
+        assert result.last_turn_ended_at == T2
+    else:
+        assert result == HostThreadActivity.unknown(HostThreadUnknownReason.RECORD_UNRECOGNIZED)
+    assert descriptor_observations == 2