Skip to content

fix: make named task/create idempotent under concurrent requests #478

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Yi-111-a wants to merge 1 commit into
base: main
Choose a base branch
from
Open

fix: make named task/create idempotent under concurrent requests #478

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
32 changes: 26 additions & 6 deletions agentex/src/domain/repositories/task_repository.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@
from typing import Annotated, Literal

from fastapi import Depends
from sqlalchemy import cast, distinct, func, select, update
from sqlalchemy import cast, distinct, exc, func, select, update
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.orm import selectinload
from src.adapters.crud_store.adapter_postgres import (
Expand Down Expand Up @@ -167,15 +167,35 @@ async def list_multi_agent_task_ids(self, *, task_ids: Sequence[str]) -> list[st
async def create(self, agent_id: str, task: TaskEntity) -> TaskEntity:
"""Create task and establish agent relationships"""

async with (
self.start_async_db_session(True) as session,
async_sql_exception_handler(),
):
async with self.start_async_db_session(True) as session:
task_data = task.to_dict()

orm = self.orm(**task_data)
session.add(orm)
await session.flush() # Get the task ID
try:
await session.flush() # Get the task ID
except exc.IntegrityError as e:
if DUPLICATE_KEY_VAL_ERR in str(e) and task.name:
# Another request created the task with the same name concurrently.
# Look up the existing task and return it, applying the losing
# request's params to the winner's task (the winner already
# holds the task, so we merge the losing request's params in).
existing = (
await session.execute(select(TaskORM).where(TaskORM.name == task.name))
).scalar_one()
Comment on lines +183 to +185

@greptile-apps greptile-apps Bot Nov 4, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P1 Concurrent task creation still fails

When two requests insert the same new task name, the losing request's session.flush() raises IntegrityError and leaves the session unusable until rollback. The next session.execute() raises PendingRollbackError, so the caller gets HTTP 500 instead of the winning task. Roll back before the lookup, or isolate the insert in a savepoint.

Prompt To Fix With AI 
This is a comment left during a code review.
Path: agentex/src/domain/repositories/task_repository.py
Line: 183-185

Comment:
**Concurrent task creation still fails**

When two requests insert the same new task name, the losing request's `session.flush()` raises `IntegrityError` and leaves the session unusable until rollback. The next `session.execute()` raises `PendingRollbackError`, so the caller gets HTTP 500 instead of the winning task. Roll back before the lookup, or isolate the insert in a savepoint.

---

For each issue above, determine whether it is valid and should be fixed. If so, fix it directly.

Fix in Cursor Fix in Claude Code Fix in Codex

# Apply the losing request's params to the winner's task
if existing.params != task.params:
existing.params = {
**(existing.params or {}),
**(task.params or {}),
}
await session.merge(existing)
await session.commit()
await session.refresh(existing)
# Do NOT re-add agent relationships — the winner keeps the task.
return TaskEntity.model_validate(existing)
else:
raise

# Add agent relationships
task_agent = TaskAgentORM(task_id=orm.id, agent_id=agent_id)
Expand Down