Repository: aden-hive/hive (python)

Issue:
[Bug]:  state.json progress writes are non-atomic and silently ignored on failure"

 ## Describe the Bug

   Hive updates `state.json` during graph execution in `core/framework/graph/executor.py` via `_write_progress()`, but the write path is fragile in two important ways:

   1. It writes `state.json` directly with `write_text()` instead of using an atomic temp-file + rename strategy.
   2. It catches `Exception` and silently ignores all failures.

   This is especially risky because the runtime later relies on the same `state.json` file for resume and shared-memory recovery. A partial write, file lock, permission error, or disk failure can leave session state stale or corrupted
   with no visible signal to the user.

   ## To Reproduce

   Steps to reproduce the behavior:

   1. Start an agent execution that updates progress across multiple node transitions.
   2. Force `_write_progress()` to fail while writing `state.json` (for example via file locking, permission changes, or an injected write error).
   3. Observe that execution continues because the exception is swallowed.
   4. Attempt resume/shared-memory recovery from the affected session.
   5. Observe stale, missing, or corrupted state with little or no actionable error reporting.

   ## Expected Behavior

   `state.json` updates should be crash-safe and observable.

   Expected behavior:
   - Writes should be atomic.
   - Failures should be logged clearly.
   - Resume/shared-memory consumers should not depend on silently corrupted or partially written state.
   - The system should preserve execution reliability even when progress persistence fails.

   ## Screenshots

   If applicable, add screenshots to help explain your problem.

   ## Environment

   - OS: Windows / Linux / macOS
   - Python version: 3.11+
   - Docker version (if applicable): N/A

   ## Configuration

   Relevant parts of your agent configuration or environment setup (remove any sensitive data):

   ```yaml
   Any graph/agent that relies on session resume or shared-memory recovery from state.json

  Logs

  Relevant code paths:

   core/framework/graph/executor.py
   - _write_progress() rewrites state.json directly
   - catches Exception and silently ignores failures

   core/framework/runtime/agent_runtime.py
   - reads state.json for resume/shared-memory recovery

  Additional Context

  Relevant references:

   - core/framework/graph/executor.py:197
   - core/framework/graph/executor.py:242-244
   - core/framework/runtime/agent_runtime.py:1327

  Suggested fix direction:

   - Replace direct writes with the existing atomic write utility / SessionStore-style temp-file swap.
   - Log failures instead of swallowing them silently.
   - Consider whether resume/shared-memory consumers need guardrails when state.json is unreadable or stale.


Patch:
diff --git a/core/framework/graph/executor.py b/core/framework/graph/executor.py
index 04d2a14ad8..9c0f911811 100644
--- a/core/framework/graph/executor.py
+++ b/core/framework/graph/executor.py
@@ -32,6 +32,7 @@
 from framework.runtime.core import Runtime
 from framework.schemas.checkpoint import Checkpoint
 from framework.storage.checkpoint_store import CheckpointStore
+from framework.utils.io import atomic_write
 
 logger = logging.getLogger(__name__)
 
@@ -226,11 +227,11 @@ def _write_progress(
         """
         if not self._storage_path:
             return
+        state_path = self._storage_path / "state.json"
         try:
             import json as _json
             from datetime import datetime
 
-            state_path = self._storage_path / "state.json"
             if state_path.exists():
                 state_data = _json.loads(state_path.read_text(encoding="utf-8"))
             else:
@@ -253,9 +254,14 @@ def _write_progress(
             state_data["memory"] = memory_snapshot
             state_data["memory_keys"] = list(memory_snapshot.keys())
 
-            state_path.write_text(_json.dumps(state_data, indent=2), encoding="utf-8")
+            with atomic_write(state_path, encoding="utf-8") as f:
+                _json.dump(state_data, f, indent=2)
         except Exception:
-            pass  # Best-effort — never block execution
+            logger.warning(
+                "Failed to persist progress state to %s",
+                state_path,
+                exc_info=True,
+            )
 
     def _validate_tools(self, graph: GraphSpec) -> list[str]:
         """


Test patch:
diff --git a/core/tests/test_graph_executor.py b/core/tests/test_graph_executor.py
index 3d30d36ba9..5eb0ad3dd5 100644
--- a/core/tests/test_graph_executor.py
+++ b/core/tests/test_graph_executor.py
@@ -3,12 +3,16 @@
 Focused on minimal success and failure scenarios.
 """
 
+import json
+import logging
+
 import pytest
 
 from framework.graph.edge import GraphSpec
 from framework.graph.executor import GraphExecutor
 from framework.graph.goal import Goal
 from framework.graph.node import NodeResult, NodeSpec
+from framework.utils.io import atomic_write
 
 
 # ---- Dummy runtime (no real logging) ----
@@ -25,6 +29,14 @@ def report_problem(self, **kwargs):
         pass
 
 
+class DummyMemory:
+    def __init__(self, data):
+        self._data = data
+
+    def read_all(self):
+        return self._data
+
+
 # ---- Fake node that always succeeds ----
 class SuccessNode:
     def validate_input(self, ctx):
@@ -245,3 +257,61 @@ async def test_executor_no_events_without_event_bus():
     result = await executor.execute(graph=graph, goal=goal)
 
     assert result.success is True
+
+
+def test_write_progress_uses_atomic_write_and_updates_state(tmp_path, monkeypatch):
+    runtime = DummyRuntime()
+    executor = GraphExecutor(runtime=runtime, storage_path=tmp_path)
+    state_path = tmp_path / "state.json"
+    state_path.write_text(json.dumps({"entry_point": "primary"}), encoding="utf-8")
+    memory = DummyMemory({"foo": "bar"})
+
+    called = {}
+
+    def recording_atomic_write(path, *args, **kwargs):
+        called["path"] = path
+        return atomic_write(path, *args, **kwargs)
+
+    monkeypatch.setattr("framework.graph.executor.atomic_write", recording_atomic_write)
+
+    executor._write_progress(
+        current_node="node-b",
+        path=["node-a", "node-b"],
+        memory=memory,
+        node_visit_counts={"node-a": 1, "node-b": 1},
+    )
+
+    state = json.loads(state_path.read_text(encoding="utf-8"))
+    assert called["path"] == state_path
+    assert state["entry_point"] == "primary"
+    assert state["progress"]["current_node"] == "node-b"
+    assert state["progress"]["path"] == ["node-a", "node-b"]
+    assert state["progress"]["node_visit_counts"] == {"node-a": 1, "node-b": 1}
+    assert state["progress"]["steps_executed"] == 2
+    assert state["memory"] == {"foo": "bar"}
+    assert state["memory_keys"] == ["foo"]
+    assert "updated_at" in state["timestamps"]
+
+
+def test_write_progress_logs_warning_on_atomic_write_failure(tmp_path, monkeypatch, caplog):
+    runtime = DummyRuntime()
+    executor = GraphExecutor(runtime=runtime, storage_path=tmp_path)
+    state_path = tmp_path / "state.json"
+    state_path.write_text(json.dumps({"entry_point": "primary"}), encoding="utf-8")
+    memory = DummyMemory({"foo": "bar"})
+
+    def failing_atomic_write(*args, **kwargs):
+        raise OSError("disk full")
+
+    monkeypatch.setattr("framework.graph.executor.atomic_write", failing_atomic_write)
+
+    with caplog.at_level(logging.WARNING):
+        executor._write_progress(
+            current_node="node-b",
+            path=["node-a", "node-b"],
+            memory=memory,
+            node_visit_counts={"node-a": 1, "node-b": 1},
+        )
+
+    assert "Failed to persist progress state to" in caplog.text
+    assert str(state_path) in caplog.text