Reject concurrent reuse of active handler IDs (#738)

## Summary

- reject HTTP workflow starts that reuse a currently active handler ID
- return `409 Conflict` with a stable, actionable error from both
blocking and non-blocking run endpoints
- keep terminal handler ID reuse intact for existing state-handoff
workflows
- add endpoint regression coverage and a patch changeset for
`llama-agents-server`

## Validation

- `uv run pytest packages/llama-agents-server/tests -n0` (447 passed, 32
deselected)
- `uv run pytest packages/llama-agents-appserver/tests -n0` (107 passed)
- `uv run ruff format --check
packages/llama-agents-server/src/llama_agents/server
packages/llama-agents-server/tests/server`
- `uv run ruff check
packages/llama-agents-server/src/llama_agents/server
packages/llama-agents-server/tests/server`
- `uv run ty check
packages/llama-agents-server/src/llama_agents/server/_service.py
packages/llama-agents-server/src/llama_agents/server/_api.py
packages/llama-agents-server/tests/server/test_server_endpoints.py`
- `git diff --check origin/main...HEAD`

## Known gaps

The store lookup and handler creation are not a cross-process
compare-and-set operation. This change closes the existing HTTP service
path mismatch and does not claim distributed atomicity.

The full-repository pre-commit run could not complete locally because
the workspace environment does not include optional AgentCore, Control
Plane, and llamactl dependencies. The checks scoped to the affected
server and appserver packages pass.

.changeset/reject-active-handler-reuse.md

@@ -0,0 +1,5 @@
+---
+"llama-agents-server": patch
+---
+
+Reject concurrent HTTP runs that reuse an active workflow handler ID

packages/llama-agents-server/src/llama_agents/server/_api.py

@@ -40,6 +40,7 @@ from workflows.utils import _nanoid as nanoid
 
 from ._service import (
     EventSendError,
+    HandlerAlreadyRunningError,
     HandlerCompletedError,
     HandlerNotFoundError,
     _WorkflowService,
@@ -414,6 +415,8 @@ class _WorkflowAPI:
                 context=context,
                 start_event=input_ev,
             )
+        except HandlerAlreadyRunningError as e:
+            raise HTTPException(detail=str(e), status_code=409)
         except Exception as e:
             logger.error(f"Error running workflow: {e}", exc_info=True)
             raise HTTPException(detail=f"Error running workflow: {e}", status_code=500)
@@ -596,6 +599,8 @@ class _WorkflowAPI:
                 context=context,
                 start_event=input_ev,
             )
+        except HandlerAlreadyRunningError as e:
+            raise HTTPException(detail=str(e), status_code=409)
         except Exception as e:
             raise HTTPException(
                 detail=f"Initial persistence failed: {e}", status_code=500

packages/llama-agents-server/src/llama_agents/server/_service.py

@@ -51,6 +51,10 @@ class HandlerCompletedError(Exception):
     pass
 
 
+class HandlerAlreadyRunningError(Exception):
+    pass
+
+
 class EventSendError(Exception):
     pass
 
@@ -213,6 +217,11 @@ class _WorkflowService:
         context: Context | None = None,
     ) -> HandlerData:
         with instrument_tags({"llamaindex.handler_id": handler_id}):
+            existing = await self.load_handler(handler_id)
+            if existing is not None and not is_terminal_status(existing.status):
+                raise HandlerAlreadyRunningError(
+                    f"Handler {handler_id!r} is already running"
+                )
             if context is None:
                 context = await self._context_from_handler_id(workflow, handler_id)
             # Pre-generate run_id and persist the handler record BEFORE starting

packages/llama-agents-server/tests/server/test_server_endpoints.py

@@ -394,6 +394,52 @@ async def test_run_workflow_nowait_success(client: AsyncClient) -> None:
     assert len(data["handler_id"]) == 10  # Default nanoid length
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize("endpoint", ["run", "run-nowait"])
+async def test_run_workflow_rejects_active_handler_id(
+    endpoint: str,
+    client: AsyncClient,
+    server: WorkflowServer,
+) -> None:
+    await server._service.store.update(
+        PersistentHandler(
+            handler_id="shared-handler",
+            workflow_name="test",
+            status="running",
+        )
+    )
+
+    response = await client.post(
+        f"/workflows/test/{endpoint}",
+        json={"handler_id": "shared-handler"},
+    )
+
+    assert response.status_code == 409
+    assert response.json() == {"detail": "Handler 'shared-handler' is already running"}
+
+
+@pytest.mark.asyncio
+async def test_run_workflow_nowait_reuses_terminal_handler_id(
+    client: AsyncClient,
+    server: WorkflowServer,
+) -> None:
+    await server._service.store.update(
+        PersistentHandler(
+            handler_id="completed-handler",
+            workflow_name="test",
+            status="completed",
+        )
+    )
+
+    response = await client.post(
+        "/workflows/test/run-nowait",
+        json={"handler_id": "completed-handler"},
+    )
+
+    assert response.status_code == 200
+    assert response.json()["handler_id"] == "completed-handler"
+
+
 @pytest.mark.asyncio
 async def test_run_workflow_nowait_with_start_event(client: AsyncClient) -> None:
     # Test with start event containing message