fix(server): harden namespace rollout edges

Author: abhinav-galileo

server/alembic/versions/b6f4c2d8e9a1_namespace_observability_events.py

@@ -28,17 +28,43 @@ def upgrade() -> None:
             nullable=False,
         ),
     )
-    op.create_index(
-        "ix_events_namespace_agent_time",
+    op.drop_constraint(
+        "control_execution_events_pkey",
         "control_execution_events",
-        ["namespace_key", "agent_name", sa.literal_column("timestamp DESC")],
-        unique=False,
+        type_="primary",
     )
+    op.create_primary_key(
+        "control_execution_events_pkey",
+        "control_execution_events",
+        ["namespace_key", "control_execution_id"],
+    )
+    with op.get_context().autocommit_block():
+        op.execute("DROP INDEX CONCURRENTLY IF EXISTS ix_events_agent_time")
+        op.execute(
+            """
+            CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_events_namespace_agent_time
+            ON control_execution_events (namespace_key, agent_name, timestamp DESC)
+            """
+        )
 
 
 def downgrade() -> None:
-    op.drop_index(
-        "ix_events_namespace_agent_time",
-        table_name="control_execution_events",
+    with op.get_context().autocommit_block():
+        op.execute("DROP INDEX CONCURRENTLY IF EXISTS ix_events_namespace_agent_time")
+        op.execute(
+            """
+            CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_events_agent_time
+            ON control_execution_events (agent_name, timestamp DESC)
+            """
+        )
+    op.drop_constraint(
+        "control_execution_events_pkey",
+        "control_execution_events",
+        type_="primary",
+    )
+    op.create_primary_key(
+        "control_execution_events_pkey",
+        "control_execution_events",
+        ["control_execution_id"],
     )
     op.drop_column("control_execution_events", "namespace_key")

server/src/agent_control_server/auth_framework/config.py

@@ -232,12 +232,12 @@ def _build_default_provider() -> RequestAuthorizer:
     )
 
 
-def _validate_local_api_key_mode() -> None:
+def _validate_local_api_key_mode(mode_env: str = _MODE_ENV) -> None:
     """Fail startup when local API-key mode has no local key validator."""
     if not auth_settings.api_key_enabled:
         raise RuntimeError(
-            f"{_MODE_ENV}=api_key requires AGENT_CONTROL_API_KEY_ENABLED=true. "
-            f"Use {_MODE_ENV}=none for deployments without credential enforcement."
+            f"{mode_env}=api_key requires AGENT_CONTROL_API_KEY_ENABLED=true. "
+            f"Use {mode_env}=none for deployments without credential enforcement."
         )
     if not auth_settings.get_api_keys() and not auth_settings.get_admin_api_keys():
         raise RuntimeError(
@@ -295,6 +295,7 @@ def _build_runtime_provider(
     if mode == "none":
         return NoAuthProvider()
     if mode == "api_key":
+        _validate_local_api_key_mode(_RUNTIME_MODE_ENV)
         return HeaderAuthProvider()
     if mode == "jwt":
         if config is None:

server/src/agent_control_server/auth_framework/providers/http_upstream.py

@@ -336,12 +336,20 @@ def _ensure_target_context_matches_grant(
     if principal.target_type is None and principal.target_id is None:
         return
     if context is None:
-        return
+        raise ForbiddenError(
+            error_code=ErrorCode.AUTH_INSUFFICIENT_PRIVILEGES,
+            detail="Authorization grant is target-bound but the request target is unavailable.",
+            hint="Use an endpoint that includes target_type and target_id in the authorization context.",
+        )
 
     expected_type = context.get("target_type")
     expected_id = context.get("target_id")
     if not isinstance(expected_type, str) or not isinstance(expected_id, str):
-        return
+        raise ForbiddenError(
+            error_code=ErrorCode.AUTH_INSUFFICIENT_PRIVILEGES,
+            detail="Authorization grant is target-bound but the request target is incomplete.",
+            hint="Provide both target_type and target_id for target-bound credentials.",
+        )
     if principal.target_type == expected_type and principal.target_id == expected_id:
         return
 

server/src/agent_control_server/endpoints/agents.py

@@ -912,6 +912,13 @@ async def init_agent(
 
         data_model.evaluators = new_evaluators
 
+    if (
+        not request.force_replace
+        and request.conflict_mode != ConflictMode.OVERWRITE
+        and (steps_changed or evaluators_changed or metadata_changed)
+    ):
+        await _authorize_existing_agent_overwrite(http_request, principal)
+
     if steps_changed or evaluators_changed or metadata_changed or force_write:
         existing.data = data_model.model_dump(mode="json")
 

server/src/agent_control_server/models.py

@@ -14,6 +14,7 @@
     ForeignKeyConstraint,
     Index,
     Integer,
+    PrimaryKeyConstraint,
     String,
     Table,
     Text,
@@ -354,7 +355,7 @@ class ControlExecutionEventDB(Base):
 
     # Primary key
     control_execution_id: Mapped[str] = mapped_column(
-        String(36), primary_key=True
+        String(36)
     )
 
     # Minimal indexed columns for efficient queries
@@ -377,7 +378,11 @@ class ControlExecutionEventDB(Base):
 
     # Composite index for agent + time queries (primary access pattern)
     __table_args__ = (
+        PrimaryKeyConstraint(
+            "namespace_key",
+            "control_execution_id",
+            name="control_execution_events_pkey",
+        ),
         Index("ix_events_namespace_agent_time", "namespace_key", "agent_name", timestamp.desc()),
-        Index("ix_events_agent_time", "agent_name", timestamp.desc()),
         Index("ix_events_data_control_id", text("(data ->> 'control_id'::text)")),
     )

server/src/agent_control_server/observability/store/postgres.py

@@ -156,7 +156,7 @@ async def store(
                         :namespace_key, :control_execution_id, :timestamp, :agent_name,
                         CAST(:data AS JSONB)
                     )
-                    ON CONFLICT (control_execution_id) DO NOTHING
+                    ON CONFLICT (namespace_key, control_execution_id) DO NOTHING
                 """),
                 values,
             )

server/tests/test_auth_framework.py

@@ -1077,7 +1077,11 @@ async def test_http_upstream_accepts_iso_datetime_and_array_scopes():
             },
         )
     )
-    principal = await provider.authorize(_build_request(), Operation.RUNTIME_TOKEN_EXCHANGE)
+    principal = await provider.authorize(
+        _build_request(),
+        Operation.RUNTIME_TOKEN_EXCHANGE,
+        context={"target_type": "log_stream", "target_id": "ls-1"},
+    )
     assert principal.namespace_key == "org-1"
     assert principal.scopes == ("runtime.use", "runtime.read_only")
     assert principal.target_type == "log_stream"
@@ -1107,6 +1111,44 @@ async def test_http_upstream_rejects_target_grant_mismatch():
         )
 
 
+@pytest.mark.asyncio
+async def test_http_upstream_rejects_target_grant_without_context():
+    provider = _build_upstream(
+        lambda req: httpx.Response(
+            200,
+            json={
+                "namespace_key": "org-1",
+                "target_type": "log_stream",
+                "target_id": "bound",
+            },
+        )
+    )
+
+    with pytest.raises(ForbiddenError, match="request target is unavailable"):
+        await provider.authorize(_build_request(), Operation.CONTROL_BINDINGS_READ)
+
+
+@pytest.mark.asyncio
+async def test_http_upstream_rejects_target_grant_with_incomplete_context():
+    provider = _build_upstream(
+        lambda req: httpx.Response(
+            200,
+            json={
+                "namespace_key": "org-1",
+                "target_type": "log_stream",
+                "target_id": "bound",
+            },
+        )
+    )
+
+    with pytest.raises(ForbiddenError, match="request target is incomplete"):
+        await provider.authorize(
+            _build_request(),
+            Operation.CONTROL_BINDINGS_READ,
+            context={"target_type": "log_stream"},
+        )
+
+
 # ---------------------------------------------------------------------------
 # configure_auth_from_env / teardown_auth lifecycle
 # ---------------------------------------------------------------------------
@@ -1248,6 +1290,18 @@ def test_configure_runtime_api_key_ignores_jwt_secret(monkeypatch):
     assert auth_config.runtime_auth_config() is None
 
 
+def test_configure_runtime_api_key_rejects_without_validator(monkeypatch):
+    from agent_control_server.auth_framework import config as auth_config
+
+    clear_authorizers()
+
+    monkeypatch.setenv("AGENT_CONTROL_RUNTIME_AUTH_MODE", "api_key")
+    monkeypatch.setattr(auth_settings, "api_key_enabled", False)
+
+    with pytest.raises(RuntimeError, match="AGENT_CONTROL_RUNTIME_AUTH_MODE=api_key"):
+        auth_config.configure_auth_from_env()
+
+
 def test_configure_runtime_unset_preserves_no_auth_default(monkeypatch):
     from agent_control_server.auth_framework import config as auth_config
 

server/tests/test_data_model_v1_alembic_migration.py

@@ -16,6 +16,7 @@
 SERVER_DIR = Path(__file__).resolve().parents[1]
 PRE_MIGRATION_REVISION = "c1e9f9c4a1d2"
 MIGRATION_REVISION = "a7f3b1e0d9c5"
+OBSERVABILITY_NAMESPACE_REVISION = "b6f4c2d8e9a1"
 _BASE_DB_URL = make_url(db_config.get_url())
 
 pytestmark = pytest.mark.skipif(
@@ -223,6 +224,21 @@ def test_downgrade_round_trip(alembic_config: Config, temp_engine: Engine) -> No
     assert "control_bindings" in inspect(temp_engine).get_table_names()
 
 
+def test_observability_namespace_migration_scopes_event_primary_key(
+    alembic_config: Config, temp_engine: Engine
+) -> None:
+    command.upgrade(alembic_config, OBSERVABILITY_NAMESPACE_REVISION)
+
+    assert "namespace_key" in _column_names(temp_engine, "control_execution_events")
+    assert _pk_columns(temp_engine, "control_execution_events") == [
+        "namespace_key",
+        "control_execution_id",
+    ]
+    indexes = _index_names(temp_engine, "control_execution_events")
+    assert "ix_events_namespace_agent_time" in indexes
+    assert "ix_events_agent_time" not in indexes
+
+
 def test_downgrade_rejects_cross_namespace_agents_duplicates(
     alembic_config: Config, temp_engine: Engine
 ) -> None:

server/tests/test_init_agent_conflict_mode.py

@@ -212,6 +212,35 @@ def test_init_agent_force_replace_existing_agent_requires_update_auth(
     assert force_resp.status_code == 403
 
 
+def test_init_agent_strict_existing_agent_mutation_requires_update_auth(
+    client: TestClient,
+) -> None:
+    agent_name = f"agent-{uuid.uuid4().hex[:12]}"
+    create_resp = client.post(
+        "/api/v1/agents/initAgent",
+        json=_init_payload(agent_name=agent_name),
+    )
+    assert create_resp.status_code == 200
+
+    set_authorizer(CreateOnlyAuthorizer())
+    strict_resp = client.post(
+        "/api/v1/agents/initAgent",
+        json=_init_payload(
+            agent_name=agent_name,
+            steps=[
+                {
+                    "type": "tool",
+                    "name": "new-tool",
+                    "input_schema": {"type": "object"},
+                    "output_schema": {"type": "object"},
+                }
+            ],
+        ),
+    )
+
+    assert strict_resp.status_code == 403
+
+
 def test_init_agent_overwrite_warns_on_removed_referenced_evaluator(client: TestClient) -> None:
     # Given: an agent whose assigned policy contains a control referencing an agent evaluator.
     agent_name = f"agent-{uuid.uuid4().hex[:12]}"

server/tests/test_observability_store_postgres.py

@@ -175,6 +175,48 @@ async def test_postgres_event_store_scopes_queries_by_namespace() -> None:
     assert stats_a.stats[0].control_id == 1
 
 
+@pytest.mark.asyncio
+async def test_postgres_event_store_idempotency_is_scoped_by_namespace() -> None:
+    session_maker = async_sessionmaker(
+        bind=async_engine,
+        class_=AsyncSession,
+        expire_on_commit=False,
+    )
+    store = PostgresEventStore(session_maker)
+
+    shared_execution_id = str(uuid4())
+    agent_name = f"agent-{uuid4().hex[:12]}"
+    now = datetime.now(UTC)
+    event_a = _event(
+        control_execution_id=shared_execution_id,
+        agent_name=agent_name,
+        control_id=1,
+        action="observe",
+        matched=True,
+        timestamp=now,
+        trace_id="a" * 32,
+    )
+    event_b = _event(
+        control_execution_id=shared_execution_id,
+        agent_name=agent_name,
+        control_id=2,
+        action="deny",
+        matched=True,
+        timestamp=now,
+        trace_id="b" * 32,
+    )
+
+    await store.store([event_a], namespace_key="tenant-a")
+    await store.store([event_b], namespace_key="tenant-b")
+
+    query = EventQueryRequest(agent_name=agent_name, limit=10, offset=0)
+    events_a = await store.query_events(query, namespace_key="tenant-a")
+    events_b = await store.query_events(query, namespace_key="tenant-b")
+
+    assert [event.control_id for event in events_a.events] == [1]
+    assert [event.control_id for event in events_b.events] == [2]
+
+
 @pytest.mark.asyncio
 async def test_postgres_event_store_store_empty_returns_zero() -> None:
     # Given: a Postgres-backed store