fix: address overseer review feedback

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>

Author: Copilot

Server/src/overseer/api.py

@@ -23,6 +23,9 @@ def create_overseer_app(
     if not api_key:
         raise ValueError("api_key is required")
     app = FastAPI(title="Overseer Control Plane", version="1.0")
+    scoped_team_ids = (
+        frozenset(authorized_team_ids) if authorized_team_ids is not None else None
+    )
 
     def serialize(value: object) -> object:
         if isinstance(value, Decimal):
@@ -36,12 +39,9 @@ def serialize(value: object) -> object:
     def authorize(requested_team_id: str | None, presented_key: str | None) -> None:
         if presented_key is None or not compare_digest(presented_key, api_key):
             raise HTTPException(status_code=401, detail="authentication required")
-        if (
-            requested_team_id is not None
-            and authorized_team_ids is not None
-            and requested_team_id not in authorized_team_ids
-        ):
-            raise HTTPException(status_code=403, detail="team access denied")
+        if scoped_team_ids is not None:
+            if requested_team_id is None or requested_team_id not in scoped_team_ids:
+                raise HTTPException(status_code=403, detail="team access denied")
 
     @app.get("/events")
     def events(

Server/src/overseer/ledger.py

@@ -11,6 +11,7 @@
 from decimal import Decimal, InvalidOperation
 import json
 import sqlite3
+from threading import RLock
 from typing import Any
 from uuid import uuid4
 
@@ -46,6 +47,7 @@ class EventLedger:
     """SQLite-backed ledger suitable for simulation and a later API adapter."""
 
     def __init__(self, connection: sqlite3.Connection):
+        self._lock = RLock()
         database_path = connection.execute("PRAGMA database_list").fetchone()[2]
         if database_path:
             self._connection = sqlite3.connect(database_path, check_same_thread=False)
@@ -128,58 +130,62 @@ def record(
             metadata=metadata or {},
             created_at=normalized_created_at.isoformat(),
         )
-        self._connection.execute(
-            """
-            INSERT INTO ledger_events
-            (id, category, event_type, team_id, agent_id, task_id, amount, currency,
-             requires_approval, approved_by, metadata_json, created_at)
-            VALUES (:id, :category, :event_type, :team_id, :agent_id, :task_id, :amount,
-                    :currency, :requires_approval, :approved_by, :metadata_json, :created_at)
-            """,
-            {**asdict(event), "requires_approval": int(event.requires_approval),
-             "amount": str(event.amount),
-             "metadata_json": json.dumps(event.metadata, sort_keys=True)},
-        )
-        self._connection.commit()
+        with self._lock:
+            self._connection.execute(
+                """
+                INSERT INTO ledger_events
+                (id, category, event_type, team_id, agent_id, task_id, amount, currency,
+                 requires_approval, approved_by, metadata_json, created_at)
+                VALUES (:id, :category, :event_type, :team_id, :agent_id, :task_id, :amount,
+                        :currency, :requires_approval, :approved_by, :metadata_json, :created_at)
+                """,
+                {**asdict(event), "requires_approval": int(event.requires_approval),
+                 "amount": str(event.amount),
+                 "metadata_json": json.dumps(event.metadata, sort_keys=True)},
+            )
+            self._connection.commit()
         return event
 
     def approve(self, event_id: str, approver_id: str) -> LedgerEvent:
         if not approver_id:
             raise ValueError("approver_id is required")
-        cursor = self._connection.execute(
-            """
-            UPDATE ledger_events
-            SET requires_approval = 0, approved_by = ?
-            WHERE id = ? AND requires_approval = 1 AND approved_by IS NULL
-            """,
-            (approver_id, event_id),
-        )
-        if cursor.rowcount != 1:
-            raise LookupError("pending approval not found")
-        self._connection.commit()
-        return self.get(event_id)
+        with self._lock:
+            cursor = self._connection.execute(
+                """
+                UPDATE ledger_events
+                SET requires_approval = 0, approved_by = ?
+                WHERE id = ? AND requires_approval = 1 AND approved_by IS NULL
+                """,
+                (approver_id, event_id),
+            )
+            if cursor.rowcount != 1:
+                raise LookupError("pending approval not found")
+            self._connection.commit()
+            return self.get(event_id)
 
     def get(self, event_id: str) -> LedgerEvent:
-        row = self._connection.execute(
-            "SELECT * FROM ledger_events WHERE id = ?", (event_id,)
-        ).fetchone()
+        with self._lock:
+            row = self._connection.execute(
+                "SELECT * FROM ledger_events WHERE id = ?", (event_id,)
+            ).fetchone()
         if row is None:
             raise LookupError("ledger event not found")
         return self._row_to_event(row)
 
     def list_events(self, team_id: str | None = None, limit: int | None = None) -> list[LedgerEvent]:
         limit_sql = "" if limit is None else " LIMIT ?"
         limit_params: tuple[Any, ...] = () if limit is None else (limit,)
-        if team_id is None:
-            rows = self._connection.execute(
-                "SELECT * FROM ledger_events ORDER BY created_at DESC" + limit_sql,
-                limit_params,
-            ).fetchall()
-        else:
-            rows = self._connection.execute(
-                "SELECT * FROM ledger_events WHERE team_id = ? ORDER BY created_at DESC" + limit_sql,
-                (team_id, *limit_params),
-            ).fetchall()
+        with self._lock:
+            if team_id is None:
+                rows = self._connection.execute(
+                    "SELECT * FROM ledger_events ORDER BY created_at DESC" + limit_sql,
+                    limit_params,
+                ).fetchall()
+            else:
+                rows = self._connection.execute(
+                    "SELECT * FROM ledger_events WHERE team_id = ? ORDER BY created_at DESC" + limit_sql,
+                    (team_id, *limit_params),
+                ).fetchall()
         return [self._row_to_event(row) for row in rows]
 
     def list_pending_approvals(
@@ -194,21 +200,23 @@ def list_pending_approvals(
         if limit is not None:
             query += " LIMIT ?"
             params += (limit,)
-        rows = self._connection.execute(query, params).fetchall()
+        with self._lock:
+            rows = self._connection.execute(query, params).fetchall()
         return [self._row_to_event(row) for row in rows]
 
     def summarize(self, team_id: str | None = None) -> list[TeamSummary]:
         where = "" if team_id is None else "WHERE team_id = ?"
         params: tuple[Any, ...] = () if team_id is None else (team_id,)
-        rows = self._connection.execute(
-            """
-            SELECT team_id, currency, category, requires_approval, amount
-            FROM ledger_events
-            """ + where + """
-            ORDER BY team_id, currency
-            """,
-            params,
-        ).fetchall()
+        with self._lock:
+            rows = self._connection.execute(
+                """
+                SELECT team_id, currency, category, requires_approval, amount
+                FROM ledger_events
+                """ + where + """
+                ORDER BY team_id, currency
+                """,
+                params,
+            ).fetchall()
         totals: dict[tuple[str, str], dict[str, Any]] = {}
         for row in rows:
             key = (row["team_id"], row["currency"])

Server/src/overseer/simulation.py

@@ -2,7 +2,7 @@
 
 from __future__ import annotations
 
-import math
+from decimal import Decimal, InvalidOperation
 
 from .ledger import EventLedger, LedgerEvent
 
@@ -15,9 +15,16 @@ def run_support_ticket_simulation(
     model_cost: float = 31.25,
 ) -> list[LedgerEvent]:
     """Record one complete support-ticket lifecycle without external side effects."""
-    if not all(
-        math.isfinite(float(amount)) and amount >= 0
-        for amount in (subscription_amount, model_cost)
+    try:
+        exact_subscription_amount = Decimal(str(subscription_amount))
+        exact_model_cost = Decimal(str(model_cost))
+    except (InvalidOperation, TypeError, ValueError):
+        raise ValueError(
+            "subscription_amount and model_cost must be finite and non-negative"
+        ) from None
+    if any(
+        not amount.is_finite() or amount < 0
+        for amount in (exact_subscription_amount, exact_model_cost)
     ):
         raise ValueError("subscription_amount and model_cost must be finite and non-negative")
     task_id = f"ticket:{ticket_id}"
@@ -44,7 +51,7 @@ def run_support_ticket_simulation(
             team_id="support",
             agent_id="support-agent",
             task_id=task_id,
-            amount=model_cost,
+            amount=exact_model_cost,
             currency="USD",
         ),
         ledger.record(
@@ -53,7 +60,7 @@ def run_support_ticket_simulation(
             team_id="support",
             agent_id="billing-agent",
             task_id=task_id,
-            amount=subscription_amount,
+            amount=exact_subscription_amount,
             currency="USD",
         ),
     ]

Server/tests/test_overseer_api.py

@@ -1,10 +1,30 @@
 import sqlite3
 
+import pytest
 from fastapi.testclient import TestClient
 
 from overseer import EventLedger, create_overseer_app, run_support_ticket_simulation
 
 
+@pytest.mark.parametrize(
+    ("subscription_amount", "model_cost"),
+    [(float("nan"), 31.25), (499, float("inf")), (-1, 31.25), (499, -1)],
+)
+def test_simulation_validates_amounts_before_recording_events(
+    subscription_amount, model_cost
+):
+    ledger = EventLedger(sqlite3.connect(":memory:"))
+
+    with pytest.raises(ValueError, match="finite and non-negative"):
+        run_support_ticket_simulation(
+            ledger,
+            subscription_amount=subscription_amount,
+            model_cost=model_cost,
+        )
+
+    assert ledger.list_events() == []
+
+
 def test_api_exposes_simulation_events_and_financial_summary():
     ledger = EventLedger(sqlite3.connect(":memory:"))
     run_support_ticket_simulation(ledger, ticket_id="T-42")
@@ -63,10 +83,23 @@ def test_api_requires_authentication_and_enforces_team_scope():
     )
 
     assert client.get("/events").status_code == 401
+    assert (
+        client.get(
+            "/events",
+            headers={"X-Overseer-Api-Key": "test-key"},
+        ).status_code
+        == 403
+    )
     assert (
         client.get(
             "/events?team_id=lead-generation",
             headers={"X-Overseer-Api-Key": "test-key"},
         ).status_code
         == 403
     )
+
+    for path in ("/summaries", "/approvals"):
+        assert (
+            client.get(path, headers={"X-Overseer-Api-Key": "test-key"}).status_code
+            == 403
+        )

Server/tests/test_overseer_ledger.py

@@ -1,4 +1,5 @@
 import sqlite3
+from concurrent.futures import ThreadPoolExecutor
 from datetime import datetime, timezone
 from decimal import Decimal
 
@@ -78,10 +79,19 @@ def test_exact_money_is_currency_aware_and_pending_amounts_are_excluded(ledger):
     assert by_currency["EUR"].revenue == Decimal("0")
     assert by_currency["EUR"].pending_approvals == 1
 
+    ledger.approve(
+        ledger.list_pending_approvals()[0].id,
+        "operator-1",
+    )
+    by_currency = {summary.currency: summary for summary in ledger.summarize()}
+    assert by_currency["EUR"].revenue == Decimal("100")
+
 
 def test_rejects_non_finite_or_naive_timestamps(ledger):
     with pytest.raises(ValueError, match="finite"):
         ledger.record(category="cost", event_type="x", team_id="t", agent_id="a", amount=float("nan"))
+    with pytest.raises(ValueError, match="finite"):
+        ledger.record(category="cost", event_type="x", team_id="t", agent_id="a", amount=float("inf"))
     with pytest.raises(ValueError, match="timezone-aware"):
         ledger.record(
             category="activity",
@@ -100,6 +110,15 @@ def test_rejects_non_finite_or_naive_timestamps(ledger):
     )
     assert event.created_at.endswith("+00:00")
 
+    offset_event = ledger.record(
+        category="activity",
+        event_type="x",
+        team_id="t",
+        agent_id="a",
+        created_at=datetime.fromisoformat("2025-01-01T01:00:00+01:00"),
+    )
+    assert offset_event.created_at == "2025-01-01T00:00:00+00:00"
+
 
 def test_requires_approval_is_visible_and_can_be_approved(ledger):
     event = ledger.record(
@@ -143,3 +162,18 @@ def test_rejects_unsafe_or_ambiguous_events(ledger):
             requires_approval=True,
             approved_by="operator",
         )
+
+
+def test_serializes_concurrent_worker_thread_access(ledger):
+    def record_event(index):
+        ledger.record(
+            category="activity",
+            event_type="worker_event",
+            team_id="support",
+            agent_id=f"agent-{index}",
+        )
+
+    with ThreadPoolExecutor(max_workers=8) as executor:
+        list(executor.map(record_event, range(40)))
+
+    assert len(ledger.list_events(team_id="support")) == 40

docs/overseer-control-plane.md

@@ -56,13 +56,15 @@ app = create_overseer_app(ledger, api_key="set-this-from-secret-storage")
 
 It provides authenticated `GET /events`, `GET /summaries`, and `GET /approvals`
 using the `X-Overseer-Api-Key` header. Pass `authorized_team_ids` when the
-operator should be restricted to a subset of teams. The
+operator should be restricted to a subset of teams; scoped requests must
+include a team ID from that set. The
 `run_support_ticket_simulation()` helper can populate a local demo ledger so
 the dashboard can be built and reviewed before any provider credentials exist.
-When the API is used, create the SQLite connection with
-`EventLedger` will copy it to a thread-safe connection for API handlers. File-
+When the API is used, `EventLedger` copies the SQLite connection to one that is
+safe for API handlers. File-
 backed databases remain persistent, while in-memory databases are copied into
-an isolated thread-safe database owned by the ledger.
+an isolated database owned by the ledger, with access serialized for FastAPI
+worker threads.
 
 Amounts are stored as exact decimals and summaries are separated by currency.
 Pending approval events remain visible in the activity and approval feeds but