fix(tasks): make named task/create idempotent under concurrent requests

agentex/src/adapters/crud_store/adapter_postgres.py

@@ -212,9 +212,12 @@ async def get(
         id: str | None = None,
         name: str | None = None,
         relationships: list[Relationships] | None = None,
+        use_primary: bool = False,
     ) -> T:
+        # use_primary reads from the read-write engine, for callers that must see
+        # a just-committed write that a lagging read replica may not have yet.
         async with (
-            self.start_async_db_session(allow_writes=False) as session,
+            self.start_async_db_session(allow_writes=use_primary) as session,
             async_sql_exception_handler(),
         ):
             result = await self._get(session, id, name, relationships=relationships)

agentex/src/domain/services/task_service.py

@@ -175,12 +175,15 @@ async def get_task(
         id: str | None = None,
         name: str | None = None,
         relationships: list[TaskRelationships] | None = None,
+        use_primary: bool = False,
     ) -> TaskEntity:
         """
         Get a task from the repository.
+
+        Reads from the read replica unless *use_primary* is set.
         """
         return await self.task_repository.get(
-            id=id, name=name, relationships=relationships
+            id=id, name=name, relationships=relationships, use_primary=use_primary
         )
 
     async def transition_task_status(

agentex/src/domain/use_cases/agents_acp_use_case.py

@@ -9,8 +9,11 @@
 from src.adapters.authentication.exceptions import (
     AuthenticationServiceUnavailableError,
 )
-from src.adapters.crud_store.exceptions import ItemDoesNotExist
-from src.api.schemas.authorization_types import AgentexResource
+from src.adapters.crud_store.exceptions import DuplicateItemError, ItemDoesNotExist
+from src.api.schemas.authorization_types import (
+    AgentexResource,
+    AuthorizedOperationType,
+)
 from src.domain.entities.agents import ACPType, AgentEntity, AgentStatus
 from src.domain.entities.agents_rpc import (
     ACP_TYPE_TO_ALLOWED_RPC_METHODS,
@@ -53,6 +56,7 @@
 from src.domain.services.task_service import DAgentTaskService
 from src.utils.acp_url import resolve_acp_url
 from src.utils.logging import make_logger
+from src.utils.task_authorization import check_task_or_collapse_to_404
 
 logger = make_logger(__name__)
 
@@ -267,6 +271,7 @@ async def _get_or_create_task(
         task_name: str | None = None,
         task_params: dict[str, Any] | None = None,
         task_metadata: dict[str, Any] | None = None,
+        _after_name_conflict: bool = False,
     ) -> TaskEntity:
         """Return the existing task if *task_id* is provided, otherwise create a new one.
 
@@ -278,6 +283,12 @@ async def _get_or_create_task(
         applied at creation time). When neither an id nor a matching name is given, a
         brand-new task is created (name may be None).
 
+        Concurrent calls with the same new *task_name* all resolve to the same task:
+        the lookup and the insert are not atomic, so a caller whose insert loses the
+        race on the unique name constraint re-reads the winner's task from the
+        primary (a read replica may not have it yet) and returns it, provided the
+        caller has update permission on that task.
+
         Note: Only one of task_id or task_name should be provided (enforced by validators).
         """
 
@@ -286,10 +297,23 @@ async def _get_or_create_task(
             task = await self.task_service.get_task(id=task_id)
         elif task_name:
             try:
-                task = await self.task_service.get_task(name=task_name)
+                task = await self.task_service.get_task(
+                    name=task_name, use_primary=_after_name_conflict
+                )
             except ItemDoesNotExist:
                 # Task doesn't exist - will create below
                 task = None
+            else:
+                if _after_name_conflict:
+                    # The request was authorized to create a new task, not to use
+                    # an existing one, and the winner may belong to another
+                    # caller. Require update permission before resuming, changing
+                    # or returning it, as the route does for an existing name.
+                    await check_task_or_collapse_to_404(
+                        self.authorization_service,
+                        task.id,
+                        AuthorizedOperationType.update,
+                    )
         else:
             # No identifier provided - will create below
             task = None
@@ -331,12 +355,36 @@ async def _get_or_create_task(
             return task
 
         # Create a new task if it doesn't exist
-        task = await self.task_service.create_task(
-            agent=agent,
-            task_name=task_name,
-            task_params=task_params,
-            task_metadata=task_metadata,
-        )
+        try:
+            task = await self.task_service.create_task(
+                agent=agent,
+                task_name=task_name,
+                task_params=task_params,
+                task_metadata=task_metadata,
+            )
+        except DuplicateItemError:
+            if not task_name or _after_name_conflict:
+                raise
+            # A concurrent request created a task with this name between our
+            # lookup and insert. Get-or-create semantics mean we return that
+            # task, so re-run once through the existing-task path (which also
+            # handles INTERRUPTED resume and params updates). The re-run reads
+            # the name from the primary: the conflict proves the row is
+            # committed there, but a lagging read replica may still miss it.
+            # It also checks the caller's update permission on that task. The
+            # winner's request owns the access grant, so we don't grant again
+            # here.
+            logger.info(
+                f"[agent_id={agent.id}] Lost create race for task name, "
+                "reusing the existing task"
+            )
+            return await self._get_or_create_task(
+                agent=agent,
+                task_name=task_name,
+                task_params=task_params,
+                task_metadata=task_metadata,
+                _after_name_conflict=True,
+            )
         logger.info(f"[agent_id={agent.id}] Created task {task.id}")
         await self.grant_with_retry(task)
         return task

agentex/tests/unit/use_cases/test_agents_acp_use_case.py

@@ -3,11 +3,16 @@
 from zoneinfo import ZoneInfo
 
 import pytest
-from src.adapters.crud_store.exceptions import ItemDoesNotExist
+from src.adapters.authorization.exceptions import AuthorizationError
+from src.adapters.crud_store.exceptions import DuplicateItemError, ItemDoesNotExist
 from src.api.schemas.agents_rpc import (
     CancelTaskRequest,
     SendEventRequest,
 )
+from src.api.schemas.authorization_types import (
+    AgentexResource,
+    AuthorizedOperationType,
+)
 from src.domain.entities.agent_api_keys import AgentAPIKeyEntity, AgentAPIKeyType
 from src.domain.entities.agents import ACPType, AgentEntity, AgentStatus
 from src.domain.entities.agents_rpc import SendMessageRequestEntity
@@ -2430,3 +2435,190 @@ def create_mock_stream(*args, **kwargs):
         task_after = await task_service.get_task(name="same-params-task")
         # Params should still be the same
         assert task_after.params == task_params
+
+    async def test_get_or_create_task_returns_winner_when_create_loses_name_race(
+        self,
+        agents_acp_use_case,
+        agent_repository,
+        task_service,
+        authorization_service,
+        sample_agent,
+    ):
+        """A concurrent create of the same name returns the existing task, not an error"""
+        # Given - another request has already inserted a task with this name, but
+        # our lookup ran before that insert committed (simulated by making the
+        # first get_task(name=...) miss)
+        await create_or_get_agent(agent_repository, sample_agent)
+        winner = await task_service.create_task(
+            agent=sample_agent, task_name="raced-task"
+        )
+        real_get_task = task_service.get_task
+        missed = False
+
+        async def get_task_missing_once(*args, **kwargs):
+            nonlocal missed
+            if kwargs.get("name") == "raced-task" and not missed:
+                missed = True
+                raise ItemDoesNotExist("not found yet")
+            return await real_get_task(*args, **kwargs)
+
+        task_service.get_task = get_task_missing_once
+        authorization_service.grant.reset_mock()
+
+        # When
+        result = await agents_acp_use_case._get_or_create_task(
+            agent=sample_agent, task_name="raced-task"
+        )
+
+        # Then - the winner's task comes back, and only the winner grants access
+        assert result.id == winner.id
+        authorization_service.grant.assert_not_called()
+
+    async def test_get_or_create_task_name_race_applies_params_to_winner(
+        self,
+        agents_acp_use_case,
+        agent_repository,
+        task_service,
+        sample_agent,
+    ):
+        """The losing request's params are applied to the winner's task, as for any get-or-create hit"""
+        # Given
+        await create_or_get_agent(agent_repository, sample_agent)
+        winner = await task_service.create_task(
+            agent=sample_agent,
+            task_name="raced-params-task",
+            task_params={"version": 1},
+        )
+        real_get_task = task_service.get_task
+        missed = False
+
+        async def get_task_missing_once(*args, **kwargs):
+            nonlocal missed
+            if kwargs.get("name") == "raced-params-task" and not missed:
+                missed = True
+                raise ItemDoesNotExist("not found yet")
+            return await real_get_task(*args, **kwargs)
+
+        task_service.get_task = get_task_missing_once
+
+        # When
+        result = await agents_acp_use_case._get_or_create_task(
+            agent=sample_agent,
+            task_name="raced-params-task",
+            task_params={"version": 2},
+        )
+
+        # Then
+        assert result.id == winner.id
+        assert result.params == {"version": 2}
+
+    async def test_get_or_create_task_name_race_reads_winner_from_primary(
+        self,
+        agents_acp_use_case,
+        agent_repository,
+        task_service,
+        authorization_service,
+        sample_agent,
+    ):
+        """A lagging read replica can't hide the winner's task after a name conflict"""
+        # Given - the winner's insert is committed on the primary, but the read
+        # replica hasn't caught up: every replica lookup by name misses, and only
+        # a read from the primary finds the task
+        await create_or_get_agent(agent_repository, sample_agent)
+        winner = await task_service.create_task(
+            agent=sample_agent, task_name="replica-lag-task"
+        )
+        real_get_task = task_service.get_task
+
+        async def get_task_lagging_replica(*args, **kwargs):
+            if kwargs.get("name") == "replica-lag-task" and not kwargs.get(
+                "use_primary"
+            ):
+                raise ItemDoesNotExist("not on the replica yet")
+            return await real_get_task(*args, **kwargs)
+
+        task_service.get_task = get_task_lagging_replica
+        authorization_service.grant.reset_mock()
+
+        # When
+        result = await agents_acp_use_case._get_or_create_task(
+            agent=sample_agent, task_name="replica-lag-task"
+        )
+
+        # Then
+        assert result.id == winner.id
+        authorization_service.grant.assert_not_called()
+        authorization_service.check.assert_awaited_once_with(
+            resource=AgentexResource.task(winner.id),
+            operation=AuthorizedOperationType.update,
+        )
+
+    @pytest.mark.parametrize(
+        ("can_read", "expected_error"),
+        [(False, ItemDoesNotExist), (True, AuthorizationError)],
+    )
+    async def test_get_or_create_task_name_race_denies_caller_without_update(
+        self,
+        agents_acp_use_case,
+        agent_repository,
+        task_service,
+        authorization_service,
+        sample_agent,
+        can_read,
+        expected_error,
+    ):
+        """Losing the race doesn't let a caller use a task it can't update"""
+        # Given - another caller won the race, and this caller was only
+        # authorized to create a new task
+        await create_or_get_agent(agent_repository, sample_agent)
+        task_name = f"someone-elses-task-{uuid4()}"
+        winner = await task_service.create_task(
+            agent=sample_agent,
+            task_name=task_name,
+            task_params={"owner": "winner"},
+        )
+        real_get_task = task_service.get_task
+        missed = False
+
+        async def get_task_missing_once(*args, **kwargs):
+            nonlocal missed
+            if kwargs.get("name") == task_name and not missed:
+                missed = True
+                raise ItemDoesNotExist("not found yet")
+            return await real_get_task(*args, **kwargs)
+
+        async def check(resource, operation):
+            if operation == AuthorizedOperationType.read and can_read:
+                return
+            raise AuthorizationError("denied")
+
+        task_service.get_task = get_task_missing_once
+        authorization_service.check.side_effect = check
+
+        # When / Then - 404 if the caller can't see the task, 403 if it can
+        with pytest.raises(expected_error):
+            await agents_acp_use_case._get_or_create_task(
+                agent=sample_agent,
+                task_name=task_name,
+                task_params={"owner": "loser"},
+            )
+
+        # The winner's task is left untouched
+        stored = await real_get_task(id=winner.id)
+        assert stored.params == {"owner": "winner"}
+
+    async def test_get_or_create_task_duplicate_without_name_is_raised(
+        self,
+        agents_acp_use_case,
+        sample_agent,
+    ):
+        """Without a name there is no get-or-create key, so a duplicate error propagates"""
+        # Given
+        agents_acp_use_case.task_service = AsyncMock()
+        agents_acp_use_case.task_service.create_task.side_effect = DuplicateItemError(
+            message="Item already exists"
+        )
+
+        # When / Then
+        with pytest.raises(DuplicateItemError):
+            await agents_acp_use_case._get_or_create_task(agent=sample_agent)