fix(agent99): bound DB sessions and expire stale outcomes

apps/api/tests/test_agent99_controlled_dispatch_reconciler_job.py

@@ -101,6 +101,7 @@ async def test_reconciler_consumes_authenticated_outcome_and_closes_same_run(
         "status_reconcile_requested": 0,
         "status_reconcile_pending": 0,
         "external_no_write_reconciled": 0,
+        "outcome_timeout_terminalized": 0,
         "verifier_written": 1,
         "closed": 1,
     }
@@ -130,6 +131,7 @@ async def test_reconciler_does_not_finalize_without_outcome(monkeypatch) -> None
         "status_reconcile_requested": 0,
         "status_reconcile_pending": 0,
         "external_no_write_reconciled": 0,
+        "outcome_timeout_terminalized": 0,
         "verifier_written": 0,
         "closed": 0,
     }

apps/api/tests/test_agent99_same_run_reconcile.py

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 import json
 from copy import deepcopy
+from datetime import UTC, datetime, timedelta
 from pathlib import Path
 from types import SimpleNamespace
 from unittest.mock import AsyncMock
@@ -812,6 +813,184 @@ async def test_external_no_write_terminal_rejects_malformed_source_counts(
     assert result["receipt_persisted"] is False
 
 
+def test_dispatch_timeout_elapsed_handles_naive_and_aware_datetimes() -> None:
+    past = datetime.now(UTC) - timedelta(minutes=1)
+    future = datetime.now(UTC) + timedelta(minutes=1)
+
+    assert job._dispatch_timeout_elapsed(past) is True
+    assert job._dispatch_timeout_elapsed(past.replace(tzinfo=None)) is True
+    assert job._dispatch_timeout_elapsed(future) is False
+    assert job._dispatch_timeout_elapsed(None) is False
+
+
+@pytest.mark.asyncio
+async def test_reconciler_terminalizes_expired_run_without_outcome(
+    monkeypatch,
+) -> None:
+    identity = build_agent99_dispatch_identity(
+        project_id="awoooi",
+        incident_id="INC-BRR-EXPIRED",
+        source_fingerprint="backup-restore:" + "b" * 64,
+        route_id="agent99:backup_health:Status",
+        work_item_id="agent99-dispatch:awoooi:INC-BRR-EXPIRED",
+    )
+
+    class TimeoutLedger(_Ledger):
+        def __init__(self) -> None:
+            super().__init__()
+            self.timeout_calls: list[object] = []
+
+        async def list_reconcilable(self, **_kwargs):  # type: ignore[no-untyped-def]
+            return [{
+                "identity": identity,
+                "receipt": self.list_receipt,
+                "timeout_at": datetime.now(UTC) - timedelta(minutes=1),
+            }]
+
+        async def record_outcome_timeout_no_write_terminal(
+            self,
+            **kwargs,
+        ):  # type: ignore[no-untyped-def]
+            self.timeout_calls.append(kwargs["identity"])
+            return {
+                "status": "outcome_timeout_no_write_terminal",
+                "receipt_persisted": True,
+                "runtime_closure_verified": False,
+            }
+
+    ledger = TimeoutLedger()
+    monkeypatch.setattr(job, "get_agent99_dispatch_ledger", lambda: ledger)
+    monkeypatch.setattr(job, "read_agent99_sre_outcome", lambda _run: None)
+
+    result = await job.reconcile_agent99_controlled_dispatches_once()
+
+    assert result["outcome_timeout_terminalized"] == 1
+    assert result["verifier_written"] == 0
+    assert ledger.timeout_calls == [identity]
+
+
+@pytest.mark.asyncio
+async def test_outcome_timeout_terminal_does_not_resolve_incident_or_replay(
+    monkeypatch,
+) -> None:
+    envelope = build_agent99_dispatch_receipt_envelope(
+        identity=IDENTITY,
+        dispatch_receipt={
+            "accepted": True,
+            "inbox_triggered": True,
+            "queue_accepted": True,
+            "dispatch_identity_matched": True,
+            "delivery_certainty": "delivered",
+        },
+        controlled_apply_authorized=True,
+    )
+
+    class TimeoutDB:
+        def __init__(self) -> None:
+            self.call = 0
+            self.statements: list[str] = []
+
+        async def execute(self, statement):  # type: ignore[no-untyped-def]
+            self.call += 1
+            self.statements.append(str(statement))
+            if self.call == 1:
+                return _ScalarResult(
+                    row=SimpleNamespace(
+                        state="waiting_tool",
+                        error_detail=json.dumps(envelope),
+                        timeout_at=(
+                            datetime.now(UTC).replace(tzinfo=None)
+                            - timedelta(minutes=1)
+                        ),
+                    )
+                )
+            if self.call == 2:
+                return _ScalarResult(value=IDENTITY.run_id)
+            return _ScalarResult()
+
+    db = TimeoutDB()
+    monkeypatch.setattr(
+        ledger_module,
+        "get_db_context",
+        lambda _project_id: _Context(db),
+    )
+
+    result = (
+        await PostgresAgent99DispatchLedger()
+        .record_outcome_timeout_no_write_terminal(identity=IDENTITY)
+    )
+
+    assert result["receipt_persisted"] is True
+    assert result["run_terminal_state"] == "failed"
+    assert result["prior_controlled_apply_authorized"] is True
+    assert result["controlled_apply_authorized"] is False
+    assert result["terminalizer_runtime_write_performed"] is False
+    assert result["outcome_runtime_write_status"] == (
+        "unknown_without_authenticated_outcome"
+    )
+    assert result["transport_replayed"] is False
+    assert result["automation_execution_success"] is False
+    assert result["post_verifier_passed"] is False
+    assert result["runtime_closure_verified"] is False
+    assert result["incident_resolution_authorized"] is False
+    assert result["learning_writeback"]["status"] == (
+        "not_applicable_outcome_timeout_no_write"
+    )
+    sql = "\n".join(db.statements).lower()
+    assert "incident_records" not in sql
+    assert "alert_operation_logs" not in sql
+    assert "telegram" not in sql
+
+
+@pytest.mark.asyncio
+async def test_outcome_timeout_terminal_rejects_unexpired_run(
+    monkeypatch,
+) -> None:
+    envelope = build_agent99_dispatch_receipt_envelope(
+        identity=IDENTITY,
+        dispatch_receipt={
+            "accepted": True,
+            "inbox_triggered": True,
+            "queue_accepted": True,
+            "dispatch_identity_matched": True,
+        },
+        controlled_apply_authorized=False,
+    )
+
+    class FutureTimeoutDB:
+        def __init__(self) -> None:
+            self.call = 0
+
+        async def execute(self, _statement):  # type: ignore[no-untyped-def]
+            self.call += 1
+            return _ScalarResult(
+                row=SimpleNamespace(
+                    state="waiting_tool",
+                    error_detail=json.dumps(envelope),
+                    timeout_at=(
+                        datetime.now(UTC).replace(tzinfo=None)
+                        + timedelta(minutes=1)
+                    ),
+                )
+            )
+
+    db = FutureTimeoutDB()
+    monkeypatch.setattr(
+        ledger_module,
+        "get_db_context",
+        lambda _project_id: _Context(db),
+    )
+
+    result = (
+        await PostgresAgent99DispatchLedger()
+        .record_outcome_timeout_no_write_terminal(identity=IDENTITY)
+    )
+
+    assert result["status"] == "outcome_timeout_not_due_no_write"
+    assert result["receipt_persisted"] is False
+    assert db.call == 1
+
+
 def test_windows_relay_and_control_plane_enforce_same_run_no_write_contract() -> None:
     root = Path(__file__).resolve().parents[3]
     relay = (root / "agent99-sre-alert-relay.ps1").read_text(encoding="utf-8")

apps/api/tests/test_config_url_validation.py

@@ -172,3 +172,4 @@ def test_database_pool_budget_defaults_to_production_safe_values():
     assert s.DATABASE_MAX_OVERFLOW == 0
     assert s.DATABASE_POOL_TIMEOUT_SECONDS == 5.0
     assert s.DATABASE_NULL_POOL is False
+    assert s.DATABASE_NULL_POOL_CONCURRENCY_LIMIT == 0

apps/api/tests/test_runtime_bootstrap_guards.py

@@ -8,6 +8,7 @@ Runtime bootstrap guard tests.
 
 from __future__ import annotations
 
+import asyncio
 import inspect
 from collections.abc import Awaitable
 from pathlib import Path
@@ -223,6 +224,52 @@ def test_get_engine_uses_null_pool_for_bounded_background_processes(monkeypatch)
     assert "pool_timeout" not in captured
 
 
+@pytest.mark.asyncio
+async def test_null_pool_session_limiter_serializes_worker_db_sessions(
+    monkeypatch,
+) -> None:
+    from src.db import base as db_base
+
+    assert "async with _database_session_slot()" in inspect.getsource(
+        db_base.get_db
+    )
+    assert "async with _database_session_slot()" in inspect.getsource(
+        db_base.get_db_context
+    )
+    monkeypatch.setattr(db_base.settings, "DATABASE_NULL_POOL", True)
+    monkeypatch.setattr(
+        db_base.settings,
+        "DATABASE_NULL_POOL_CONCURRENCY_LIMIT",
+        1,
+    )
+    monkeypatch.setattr(db_base.settings, "DATABASE_POOL_TIMEOUT_SECONDS", 1.0)
+    monkeypatch.setattr(db_base, "_null_pool_session_limiter", None)
+    monkeypatch.setattr(db_base, "_null_pool_session_limiter_size", 0)
+    first_entered = asyncio.Event()
+    release_first = asyncio.Event()
+    second_entered = asyncio.Event()
+
+    async def first_session() -> None:
+        async with db_base._database_session_slot():
+            first_entered.set()
+            await release_first.wait()
+
+    async def second_session() -> None:
+        async with db_base._database_session_slot():
+            second_entered.set()
+
+    first_task = asyncio.create_task(first_session())
+    await first_entered.wait()
+    second_task = asyncio.create_task(second_session())
+    await asyncio.sleep(0)
+
+    assert second_entered.is_set() is False
+
+    release_first.set()
+    await asyncio.gather(first_task, second_task)
+    assert second_entered.is_set() is True
+
+
 @pytest.mark.asyncio
 async def test_init_db_serializes_bootstrap_ddl_with_advisory_lock(monkeypatch):
     from src.db import base as db_base

apps/api/tests/test_signal_worker_ansible_executor_binding.py

@@ -91,6 +91,8 @@ def test_worker_manifest_has_no_execution_transport_or_write_loop() -> None:
     assert worker_env["MCP_FEDERATION_STARTUP_SLEEP_SECONDS"] == "45"
     assert worker_env["MCP_FEDERATION_SOURCE_TIMEOUT_SECONDS"] == "10"
     assert worker_env["AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_WINDOW_HOURS"] == "168"
+    assert worker_env["DATABASE_NULL_POOL"] == "true"
+    assert worker_env["DATABASE_NULL_POOL_CONCURRENCY_LIMIT"] == "1"
 
 
 def test_execution_broker_is_only_workload_with_ssh_transport() -> None: