fix(file_state): purge stale writer claims in forget_task and add writer TTL (#114446)

(cherry picked from commit b5c1ea32151e56d197e371dd7a5d682218903e8b)

tests/tools/test_file_state_registry.py

@@ -172,6 +172,45 @@ class FileStateRegistryUnitTests(unittest.TestCase):
         self.assertNotIn(task_id, rt._read_tracker)
         self.assertNotIn(task_id, rt._patch_failure_tracker)
 
+    def test_forget_task_clears_last_writer_claims(self):
+        """Regression test for issue #114446: forget_task must prune _last_writer
+        entries owned by the ended task so sequential runs don't false-positive as
+        concurrent sibling conflicts."""
+        p = self._mk()
+        file_state.note_write("cron:JOB:run1", p)
+        registry = file_state.get_registry()
+        self.assertIn(p, registry._last_writer)
+        self.assertEqual(registry._last_writer[p][0], "cron:JOB:run1")
+
+        # Now task lifecycle ends
+        file_state.forget_task("cron:JOB:run1")
+
+        # The writer claim must be gone
+        self.assertNotIn(p, registry._last_writer)
+
+        # Next sequential run touching the file should not trigger a sibling warning
+        warn = file_state.check_stale("cron:JOB:run2", p)
+        self.assertIsNone(warn)
+
+    def test_clear_file_ops_cache_clears_last_writer_claims(self):
+        """Ensure file_tools.clear_file_ops_cache propagates forget_task to _last_writer."""
+        p = self._mk()
+        file_state.note_write("worker-1", p)
+        clear_file_ops_cache("worker-1")
+        warn = file_state.check_stale("worker-2", p)
+        self.assertIsNone(warn)
+
+    def test_last_writer_ttl_expiration(self):
+        """Entries older than TTL must not report stale conflicts for long-lived processes."""
+        p = self._mk()
+        # Simulate a write from 2 hours ago
+        old_time = time.time() - 7200
+        with file_state.get_registry()._state_lock:
+            file_state.get_registry()._last_writer[p] = ("old-task", old_time)
+
+        warn = file_state.check_stale("new-task", p)
+        self.assertIsNone(warn)
+        self.assertNotIn(p, file_state.get_registry()._last_writer)
 
     def test_kill_switch_env_var(self):
         p = self._mk()

tools/file_state.py

@@ -38,6 +38,17 @@ def guard_disabled() -> bool:
     return _disabled()
 
 
+def _writer_ttl_seconds() -> float:
+    # TTL for _last_writer entries to bound concurrent conflict detection window
+    raw = os.environ.get("HERMES_FILE_STATE_WRITER_TTL")
+    if raw:
+        try:
+            return float(raw)
+        except ValueError:
+            pass
+    return 3600.0  # default: 1 hour
+
+
 def _mtime_or_none(resolved: str) -> Optional[float]:
     try:
         return os.path.getmtime(resolved)
@@ -132,6 +143,11 @@ class FileStateRegistry:
         with self._state_lock:
             stamp = self._reads.get(task_id, {}).get(resolved)
             last_writer = self._last_writer.get(resolved)
+            if last_writer is not None:
+                ttl = _writer_ttl_seconds()
+                if ttl > 0 and (time.time() - last_writer[1]) > ttl:
+                    self._last_writer.pop(resolved, None)
+                    last_writer = None
 
         if stamp is None and last_writer is None:  # net-new file / first touch
             return None
@@ -197,9 +213,15 @@ class FileStateRegistry:
             return list(self._reads.get(task_id, {}).keys())
 
     def forget_task(self, task_id: str) -> None:
-        """Release read stamps owned by a task after its lifecycle ends."""
+        """Release read stamps and writer claims owned by a task after its lifecycle ends."""
         with self._state_lock:
             self._reads.pop(task_id, None)
+            stale_paths = [
+                p for p, (writer_tid, _) in self._last_writer.items()
+                if writer_tid == task_id
+            ]
+            for p in stale_paths:
+                self._last_writer.pop(p, None)
 
     def clear(self) -> None:
         """Reset all state. Intended for tests only."""
@@ -243,6 +265,10 @@ def known_reads(task_id: str) -> List[str]:
     return _registry.known_reads(task_id)
 
 
+def forget_task(task_id: str) -> None:
+    _registry.forget_task(task_id)
+
+
 __all__ = [
     "FileStateRegistry",
     "get_registry",
@@ -251,4 +277,5 @@ __all__ = [
     "check_stale",
     "lock_path",
     "writes_since",
-    "known_reads"]
+    "known_reads",
+    "forget_task"]