fix(server): keep initAgent on registration authorization

server/src/agent_control_server/endpoints/agents.py

@@ -174,23 +174,6 @@ def _ensure_target_principal_matches_namespace(
     )
 
 
-async def _authorize_existing_agent_overwrite(
-    request: Request,
-    principal: Principal,
-) -> None:
-    update_principal = await get_authorizer(Operation.AGENTS_UPDATE).authorize(
-        request,
-        Operation.AGENTS_UPDATE,
-    )
-    if update_principal.namespace_key == principal.namespace_key:
-        return
-    raise ForbiddenError(
-        error_code=ErrorCode.AUTH_INSUFFICIENT_PRIVILEGES,
-        detail="Update authorization resolved to a different namespace.",
-        hint="Ensure the credential is scoped to the requested agent namespace.",
-    )
-
-
 # =============================================================================
 # List Agents Models
 # =============================================================================
@@ -549,7 +532,6 @@ async def list_agents(
 )
 async def init_agent(
     request: InitAgentRequest,
-    http_request: Request,
     db: AsyncSession = Depends(get_async_db),
     principal: Principal = Depends(require_operation(Operation.AGENTS_CREATE)),
     target_principal: Principal | None = Depends(_init_agent_target_principal),
@@ -682,9 +664,6 @@ async def init_agent(
         )
         return InitAgentResponse(created=created, controls=controls)
 
-    if request.force_replace or request.conflict_mode == ConflictMode.OVERWRITE:
-        await _authorize_existing_agent_overwrite(http_request, principal)
-
     # Parse existing data via AgentData Pydantic model
     try:
         data_model = AgentData.model_validate(existing.data)
@@ -912,13 +891,6 @@ 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/tests/test_auth.py

@@ -230,7 +230,7 @@ def test_non_admin_key_denied_on_admin_only_mutations(
         body = response.json()
         assert body["error_code"] == "AUTH_INSUFFICIENT_PRIVILEGES"
 
-    def test_non_admin_key_can_init_agent_and_fetch_controls(
+    def test_non_admin_key_can_register_refresh_agent_and_fetch_controls(
         self, non_admin_client: TestClient
     ) -> None:
         agent_name = f"runtime-agent-{uuid.uuid4().hex[:8]}"
@@ -253,6 +253,32 @@ def test_non_admin_key_can_init_agent_and_fetch_controls(
 
         init_response = non_admin_client.post("/api/v1/agents/initAgent", json=init_payload)
         assert init_response.status_code == 200
+        assert init_response.json()["created"] is True
+
+        updated_payload = {
+            "agent": {
+                "agent_name": agent_name,
+                "agent_description": "Updated runtime agent",
+                "agent_version": "2.0",
+            },
+            "steps": [
+                {
+                    "type": "tool",
+                    "name": "tool_b",
+                    "input_schema": {"type": "object"},
+                    "output_schema": {"type": "object"},
+                }
+            ],
+            "evaluators": [],
+            "conflict_mode": "overwrite",
+        }
+        refresh_response = non_admin_client.post(
+            "/api/v1/agents/initAgent",
+            json=updated_payload,
+        )
+        assert refresh_response.status_code == 200
+        assert refresh_response.json()["created"] is False
+        assert refresh_response.json()["overwrite_applied"] is True
 
         controls_response = non_admin_client.get(f"/api/v1/agents/{agent_name}/controls")
         assert controls_response.status_code == 200

server/tests/test_init_agent_conflict_mode.py

@@ -17,13 +17,17 @@
 
 
 class CreateOnlyAuthorizer:
+    def __init__(self) -> None:
+        self.operations: list[Operation] = []
+
     async def authorize(
         self,
         request: Request,
         operation: Operation,
         context: dict[str, Any] | None = None,
     ) -> Principal:
         del request, context
+        self.operations.append(operation)
         if operation is Operation.AGENTS_UPDATE:
             raise ForbiddenError(
                 error_code=ErrorCode.AUTH_INSUFFICIENT_PRIVILEGES,
@@ -174,55 +178,94 @@ def test_init_agent_overwrite_replaces_steps_and_evaluators(client: TestClient)
     assert {evaluator["name"] for evaluator in get_data["evaluators"]} == {"eval-a", "eval-c"}
 
 
-def test_init_agent_overwrite_existing_agent_requires_update_auth(
+def test_init_agent_overwrite_existing_agent_uses_create_auth(
     client: TestClient,
 ) -> None:
+    # Given: an existing agent and a principal that may register but not update agents.
     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())
+    authorizer = CreateOnlyAuthorizer()
+    set_authorizer(authorizer)
+
+    # When: the registration is refreshed using the SDK's default overwrite mode.
     overwrite_resp = client.post(
         "/api/v1/agents/initAgent",
-        json=_init_payload(agent_name=agent_name, conflict_mode="overwrite"),
+        json=_init_payload(
+            agent_name=agent_name,
+            agent_description="updated",
+            agent_version="2.0",
+            steps=[
+                {
+                    "type": "tool",
+                    "name": "new-tool",
+                    "input_schema": {"type": "object"},
+                    "output_schema": {"type": "object"},
+                }
+            ],
+            conflict_mode="overwrite",
+        ),
     )
 
-    assert overwrite_resp.status_code == 403
+    # Then: registration succeeds without requesting the agent management operation.
+    assert overwrite_resp.status_code == 200
+    assert overwrite_resp.json()["created"] is False
+    assert overwrite_resp.json()["overwrite_applied"] is True
+    assert authorizer.operations == [Operation.AGENTS_CREATE]
 
 
-def test_init_agent_force_replace_existing_agent_requires_update_auth(
+def test_init_agent_force_replace_existing_agent_uses_create_auth(
     client: TestClient,
 ) -> None:
+    # Given: an existing agent and a principal that may register but not update agents.
     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())
+    authorizer = CreateOnlyAuthorizer()
+    set_authorizer(authorizer)
+
+    # When: the existing registration is force-replaced.
     force_resp = client.post(
         "/api/v1/agents/initAgent",
-        json={**_init_payload(agent_name=agent_name), "force_replace": True},
+        json={
+            **_init_payload(
+                agent_name=agent_name,
+                agent_description="force-replaced",
+                agent_version="2.0",
+            ),
+            "force_replace": True,
+        },
     )
 
-    assert force_resp.status_code == 403
+    # Then: registration succeeds without requesting the agent management operation.
+    assert force_resp.status_code == 200
+    assert force_resp.json()["created"] is False
+    assert authorizer.operations == [Operation.AGENTS_CREATE]
 
 
-def test_init_agent_strict_existing_agent_mutation_requires_update_auth(
+def test_init_agent_strict_existing_agent_mutation_uses_create_auth(
     client: TestClient,
 ) -> None:
+    # Given: an existing agent and a principal that may register but not update agents.
     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())
+    authorizer = CreateOnlyAuthorizer()
+    set_authorizer(authorizer)
+
+    # When: strict registration adds a compatible step.
     strict_resp = client.post(
         "/api/v1/agents/initAgent",
         json=_init_payload(
@@ -238,7 +281,10 @@ def test_init_agent_strict_existing_agent_mutation_requires_update_auth(
         ),
     )
 
-    assert strict_resp.status_code == 403
+    # Then: registration succeeds without requesting the agent management operation.
+    assert strict_resp.status_code == 200
+    assert strict_resp.json()["created"] is False
+    assert authorizer.operations == [Operation.AGENTS_CREATE]
 
 
 def test_init_agent_overwrite_warns_on_removed_referenced_evaluator(client: TestClient) -> None:
@@ -369,6 +415,8 @@ def test_init_agent_overwrite_noop_reports_not_applied(client: TestClient) -> No
     assert first_resp.status_code == 200
 
     # When: initAgent is called in overwrite mode with no effective registration changes.
+    authorizer = CreateOnlyAuthorizer()
+    set_authorizer(authorizer)
     second_payload = dict(payload)
     second_payload["conflict_mode"] = "overwrite"
     second_resp = client.post("/api/v1/agents/initAgent", json=second_payload)
@@ -387,3 +435,4 @@ def test_init_agent_overwrite_noop_reports_not_applied(client: TestClient) -> No
         "evaluators_removed": [],
         "evaluator_removals": [],
     }
+    assert authorizer.operations == [Operation.AGENTS_CREATE]