fix: harden runtime concurrency and production guardrails

.github/workflows/security.yml

@@ -80,12 +80,13 @@ jobs:
         with:
           python-version: "3.11"
           cache: pip
-      - run: pip install pip-audit
-      - name: Generate audit report
+      - name: Generate dependency-audit report
         run: |
+          python -m pip install --upgrade pip pip-audit
           set +e
-          pip-audit -r requirements.txt --format json --output pip-audit.json
-          echo "Dependency findings retained for triage."
+          pip-audit --requirement requirements.txt --format json --output pip-audit.json
+          status=$?
+          echo "Dependency audit completed with status $status; findings are retained for remediation."
           exit 0
       - uses: actions/upload-artifact@v7
         with:

agent/autonomy/runtime.py

@@ -2,6 +2,8 @@
 
 from __future__ import annotations
 
+import threading
+
 from agent.autonomy.models import (
     AgentRun,
     ApprovalRequest,
@@ -32,6 +34,8 @@ def __init__(
         self.registry = registry or self._default_registry()
         self.store = store or SQLiteRunStore()
         self.event_sink = event_sink or build_event_sink_from_env()
+        self._run_locks: dict[str, threading.Lock] = {}
+        self._run_locks_guard = threading.Lock()
 
     @staticmethod
     def _default_registry() -> ToolRegistry:
@@ -110,7 +114,26 @@ def submit(
         self._emit(EventType.PLAN_CREATED, run, plan_size=len(run.plan))
         return run
 
+    def _lock_for_run(self, run_id: str) -> threading.Lock:
+        with self._run_locks_guard:
+            return self._run_locks.setdefault(run_id, threading.Lock())
+
+    def _discard_terminal_lock(self, run: AgentRun) -> None:
+        if run.status not in (RunStatus.COMPLETED, RunStatus.FAILED):
+            return
+        with self._run_locks_guard:
+            self._run_locks.pop(run.id, None)
+
     def run(self, run_id: str) -> AgentRun:
+        """Execute one durable run at most once per runtime process."""
+
+        lock = self._lock_for_run(run_id)
+        with lock:
+            run = self._run_locked(run_id)
+            self._discard_terminal_lock(run)
+            return run
+
+    def _run_locked(self, run_id: str) -> AgentRun:
         run = self.store.get(run_id)
         if run.status in (RunStatus.COMPLETED, RunStatus.FAILED):
             return run
@@ -213,7 +236,18 @@ def run(self, run_id: str) -> AgentRun:
         return run
 
     def approve(self, run_id: str, task_id: str, approved: bool) -> AgentRun:
+        """Persist a decision without racing execution of the same run."""
+
+        lock = self._lock_for_run(run_id)
+        with lock:
+            run = self._approve_locked(run_id, task_id, approved)
+            self._discard_terminal_lock(run)
+            return run
+
+    def _approve_locked(self, run_id: str, task_id: str, approved: bool) -> AgentRun:
         run = self.store.get(run_id)
+        if run.status != RunStatus.WAITING_APPROVAL:
+            raise ValueError(f"Run {run_id} is not waiting for approval")
         request = next((item for item in run.approvals if item.task_id == task_id), None)
         if request is None:
             raise KeyError(f"No approval request for task {task_id}")

agent/autonomy/store.py

@@ -4,6 +4,7 @@
 
 import os
 import sqlite3
+import threading
 from pathlib import Path
 
 from agent.autonomy.models import AgentRun
@@ -18,7 +19,12 @@ def __init__(self, path: str | Path | None = None) -> None:
         self.path = str(path or os.getenv("HELIXAGENT_RUN_DB", "data/helixagent_runs.db"))
         if self.path != ":memory:":
             Path(self.path).parent.mkdir(parents=True, exist_ok=True)
+        self._lock = threading.RLock()
         self._connection = sqlite3.connect(self.path, check_same_thread=False)
+        self._connection.execute("PRAGMA busy_timeout = 5000")
+        if self.path != ":memory:":
+            self._connection.execute("PRAGMA journal_mode = WAL")
+            self._connection.execute("PRAGMA synchronous = FULL")
         self._connection.execute(
             "CREATE TABLE IF NOT EXISTS agent_runs "
             "(id TEXT PRIMARY KEY, state_json TEXT NOT NULL, updated_at TEXT NOT NULL)"
@@ -27,7 +33,8 @@ def __init__(self, path: str | Path | None = None) -> None:
 
     def close(self) -> None:
         """Release the database handle deterministically."""
-        self._connection.close()
+        with self._lock:
+            self._connection.close()
 
     def __enter__(self) -> SQLiteRunStore:  # noqa: PYI034 - Python 3.10 compatibility
         return self
@@ -37,18 +44,20 @@ def __exit__(self, *_exc: object) -> None:
 
     def save(self, run: AgentRun) -> None:
         run.touch()
-        self._connection.execute(
-            "INSERT INTO agent_runs(id, state_json, updated_at) VALUES (?, ?, ?) "
-            "ON CONFLICT(id) DO UPDATE SET state_json=excluded.state_json, "
-            "updated_at=excluded.updated_at",
-            (run.id, run.model_dump_json(), run.updated_at.isoformat()),
-        )
-        self._connection.commit()
+        with self._lock:
+            self._connection.execute(
+                "INSERT INTO agent_runs(id, state_json, updated_at) VALUES (?, ?, ?) "
+                "ON CONFLICT(id) DO UPDATE SET state_json=excluded.state_json, "
+                "updated_at=excluded.updated_at",
+                (run.id, run.model_dump_json(), run.updated_at.isoformat()),
+            )
+            self._connection.commit()
 
     def get(self, run_id: str) -> AgentRun:
-        row = self._connection.execute(
-            "SELECT state_json FROM agent_runs WHERE id = ?", (run_id,)
-        ).fetchone()
+        with self._lock:
+            row = self._connection.execute(
+                "SELECT state_json FROM agent_runs WHERE id = ?", (run_id,)
+            ).fetchone()
         if row is None:
             raise RunNotFoundError(run_id)
         return AgentRun.model_validate_json(row[0])

api/main.py

@@ -101,6 +101,8 @@ async def decide_approval(
         run = runtime.approve(run_id, task_id, payload.approved)
     except (RunNotFoundError, KeyError) as exc:
         raise HTTPException(status_code=404, detail=str(exc)) from exc
+    except ValueError as exc:
+        raise HTTPException(status_code=409, detail=str(exc)) from exc
     if payload.approved:
         background_tasks.add_task(runtime.run, run.id)
     return run

api/monitoring.py

@@ -33,6 +33,13 @@
 )
 
 
+def _endpoint_label(request: Request) -> str:
+    """Use the matched route template to avoid unbounded metric cardinality."""
+
+    route = request.scope.get("route")
+    return getattr(route, "path", "unmatched")
+
+
 def setup_monitoring(app: FastAPI):
     """Attach monitoring endpoints and tracing to FastAPI"""
 
@@ -44,16 +51,22 @@ async def metrics():
     # Middleware for metrics
     @app.middleware("http")
     async def prometheus_middleware(request: Request, call_next):
-        start = time.time()
-
-        response = await call_next(request)
-        latency = time.time() - start
+        start = time.perf_counter()
+        try:
+            response = await call_next(request)
+        except Exception:
+            endpoint = _endpoint_label(request)
+            REQUEST_COUNT.labels(
+                method=request.method, endpoint=endpoint, status_code=500
+            ).inc()
+            REQUEST_LATENCY.labels(endpoint=endpoint).observe(time.perf_counter() - start)
+            raise
 
+        endpoint = _endpoint_label(request)
         REQUEST_COUNT.labels(
-            method=request.method, endpoint=request.url.path, status_code=response.status_code
+            method=request.method, endpoint=endpoint, status_code=response.status_code
         ).inc()
-        REQUEST_LATENCY.labels(endpoint=request.url.path).observe(latency)
-
+        REQUEST_LATENCY.labels(endpoint=endpoint).observe(time.perf_counter() - start)
         return response
 
     # -----------------------------

tests/test_autonomous_runtime.py

@@ -1,8 +1,11 @@
 """Behavioral tests for the durable, governed autonomous control loop."""
 
 import time
+from concurrent.futures import ThreadPoolExecutor
 from pathlib import Path
 
+import pytest
+
 from agent.autonomy.models import GoalSpec, RiskLevel, RunStatus, Task
 from agent.autonomy.runtime import AutonomousRuntime
 from agent.autonomy.store import SQLiteRunStore
@@ -46,6 +49,34 @@ def test_run_completes_and_persists_observation(tmp_path: Path) -> None:
     assert restored.observations[0].output == "evidence"
 
 
+def test_concurrent_run_requests_execute_a_run_once(tmp_path: Path) -> None:
+    calls = 0
+
+    def slow_echo(_args):
+        nonlocal calls
+        calls += 1
+        time.sleep(0.05)
+        return "evidence"
+
+    registry = ToolRegistry()
+    registry.register(ToolSpec("slow_echo", slow_echo, "Produce evidence"))
+    runtime = runtime_for(
+        tmp_path,
+        [Task(objective="Produce evidence", tool="slow_echo")],
+        registry,
+    )
+    submitted = runtime.submit("Produce evidence")
+
+    with ThreadPoolExecutor(max_workers=2) as executor:
+        results = list(executor.map(runtime.run, [submitted.id, submitted.id]))
+
+    restored = runtime.store.get(submitted.id)
+    assert calls == 1
+    assert all(result.status == RunStatus.COMPLETED for result in results)
+    assert restored.tool_calls == 1
+    assert len(restored.observations) == 1
+
+
 def test_write_tool_pauses_until_explicit_approval(tmp_path: Path) -> None:
     registry = ToolRegistry()
     registry.register(
@@ -63,6 +94,19 @@ def test_write_tool_pauses_until_explicit_approval(tmp_path: Path) -> None:
     assert completed.status == RunStatus.COMPLETED
 
 
+def test_stale_approval_decision_is_rejected(tmp_path: Path) -> None:
+    registry = ToolRegistry()
+    registry.register(
+        ToolSpec("write", lambda args: "written", "Write data", risk=RiskLevel.WRITE)
+    )
+    runtime = runtime_for(tmp_path, [Task(objective="Write", tool="write")], registry)
+    paused = runtime.run(runtime.submit("Write an artifact").id)
+
+    runtime.approve(paused.id, paused.plan[0].id, True)
+    with pytest.raises(ValueError, match="not waiting for approval"):
+        runtime.approve(paused.id, paused.plan[0].id, True)
+
+
 def test_denied_approval_fails_without_tool_execution(tmp_path: Path) -> None:
     calls: list[str] = []
     registry = ToolRegistry()