diff --git a/.github/workflows/security.yml b/.github/workflows/security.yml index 335e43b..dd1a5fc 100644 --- a/.github/workflows/security.yml +++ b/.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: diff --git a/agent/autonomy/runtime.py b/agent/autonomy/runtime.py index 2d12c04..a38af37 100644 --- a/agent/autonomy/runtime.py +++ b/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}") diff --git a/agent/autonomy/store.py b/agent/autonomy/store.py index cc1d3f4..2fda30a 100644 --- a/agent/autonomy/store.py +++ b/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]) diff --git a/api/main.py b/api/main.py index f2c3920..291d535 100644 --- a/api/main.py +++ b/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 diff --git a/api/monitoring.py b/api/monitoring.py index a02a7cc..995c7d5 100644 --- a/api/monitoring.py +++ b/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 # ----------------------------- diff --git a/tests/test_autonomous_runtime.py b/tests/test_autonomous_runtime.py index a2e5d73..4822118 100644 --- a/tests/test_autonomous_runtime.py +++ b/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()