diff --git a/tests/tools/test_file_state_registry.py b/tests/tools/test_file_state_registry.py index f465e6f4e8..bf46723eb0 100644 --- a/tests/tools/test_file_state_registry.py +++ b/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() diff --git a/tools/file_state.py b/tools/file_state.py index 54518cb99c..f4576e82cb 100644 --- a/tools/file_state.py +++ b/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"]