董宏凯andCursor
1a080311bd
Make ensure_participants concurrent-safe, restore clarification checkpoints without re-planning, and only create threads for validated seats. Co-authored-by: Cursor <cursoragent@cursor.com>
263 lines
10 KiB
Python
263 lines
10 KiB
Python
"""Participant repository for the MAF roundtable experiment.
|
|
|
|
Participants are the manager + seat agents bound to a session. They are
|
|
created together with the session (``replace_participants`` is the atomic
|
|
full-replacement API mirroring ``task_buttons``) and read via the session.
|
|
``capability_snapshot_json`` freezes each agent's tools / skills at session
|
|
creation time.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from datetime import UTC, datetime
|
|
from typing import Any
|
|
from uuid import uuid4
|
|
|
|
from sqlalchemy import delete, select
|
|
from sqlalchemy.exc import IntegrityError
|
|
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
|
|
|
|
from deerflow.persistence.maf_roundtable.model import MafRoundtableParticipantRow
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def _now() -> datetime:
|
|
return datetime.now(UTC)
|
|
|
|
|
|
def _to_dict(row: MafRoundtableParticipantRow) -> dict[str, Any]:
|
|
return {
|
|
"id": row.id,
|
|
"session_id": row.session_id,
|
|
"agent_id": row.agent_id,
|
|
"display_name": row.display_name,
|
|
"role": row.role,
|
|
"sort_order": row.sort_order,
|
|
"thread_id": row.thread_id,
|
|
"sandbox_id": row.sandbox_id,
|
|
"status": row.status,
|
|
"last_turn_index": row.last_turn_index,
|
|
"capability_snapshot": row.capability_snapshot_json,
|
|
"created_at": row.created_at.isoformat() if isinstance(row.created_at, datetime) else row.created_at,
|
|
"updated_at": row.updated_at.isoformat() if isinstance(row.updated_at, datetime) else row.updated_at,
|
|
}
|
|
|
|
|
|
class MafParticipantRepository:
|
|
def __init__(self, session_factory: async_sessionmaker[AsyncSession]) -> None:
|
|
self._sf = session_factory
|
|
|
|
async def list_participants(self, session_id: str) -> list[dict[str, Any]]:
|
|
stmt = (
|
|
select(MafRoundtableParticipantRow)
|
|
.where(MafRoundtableParticipantRow.session_id == session_id)
|
|
.order_by(MafRoundtableParticipantRow.sort_order.asc())
|
|
)
|
|
async with self._sf() as session:
|
|
rows = (await session.execute(stmt)).scalars().all()
|
|
return [_to_dict(r) for r in rows]
|
|
|
|
async def replace_participants(
|
|
self, session_id: str, participants: list[dict[str, Any]]
|
|
) -> list[dict[str, Any]]:
|
|
"""Atomically replace all participants of a session.
|
|
|
|
Each item: ``{agent_id, display_name?, role?, sort_order?, thread_id?,
|
|
sandbox_id?, capability_snapshot?}``. De-dupes by ``agent_id`` (last
|
|
wins) to honor the ``(session_id, agent_id)`` unique constraint.
|
|
"""
|
|
by_agent: dict[str, dict[str, Any]] = {}
|
|
for p in participants:
|
|
agent_id = str(p.get("agent_id") or "")
|
|
if not agent_id:
|
|
continue
|
|
by_agent[agent_id] = p
|
|
now = _now()
|
|
new_rows = [
|
|
MafRoundtableParticipantRow(
|
|
id=uuid4().hex,
|
|
session_id=session_id,
|
|
agent_id=agent_id,
|
|
display_name=str(p.get("display_name") or ""),
|
|
role=str(p.get("role") or "seat"),
|
|
sort_order=int(p.get("sort_order", 0)),
|
|
thread_id=p.get("thread_id"),
|
|
sandbox_id=p.get("sandbox_id"),
|
|
status=str(p.get("status") or "idle"),
|
|
last_turn_index=p.get("last_turn_index"),
|
|
capability_snapshot_json=p.get("capability_snapshot"),
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
for agent_id, p in by_agent.items()
|
|
]
|
|
async with self._sf() as session:
|
|
await session.execute(
|
|
delete(MafRoundtableParticipantRow).where(
|
|
MafRoundtableParticipantRow.session_id == session_id
|
|
)
|
|
)
|
|
session.add_all(new_rows)
|
|
await session.commit()
|
|
return [_to_dict(r) for r in new_rows]
|
|
|
|
async def ensure_participants(
|
|
self, session_id: str, participants: list[dict[str, Any]]
|
|
) -> list[dict[str, Any]]:
|
|
"""Insert missing participants using a database-level idempotent insert.
|
|
|
|
Concurrent-safe: two coroutines calling this simultaneously with the
|
|
same seat set will never raise a unique-constraint error and will
|
|
never produce duplicate rows.
|
|
|
|
- SQLite / PostgreSQL: ``INSERT … ON CONFLICT (session_id, agent_id) DO NOTHING``
|
|
- MySQL: ``INSERT IGNORE …``
|
|
|
|
After the insert attempt the method **re-queries** the database and
|
|
returns the final persisted records (sorted by ``sort_order, agent_id``).
|
|
Existing ``thread_id``, ``sandbox_id``, ``status`` and
|
|
``capability_snapshot_json`` are never overwritten.
|
|
"""
|
|
# Filter: skip empty agent_ids and de-duplicate input.
|
|
candidates: list[dict[str, Any]] = []
|
|
seen: set[str] = set()
|
|
for participant in participants:
|
|
agent_id = str(participant.get("agent_id") or "").strip()
|
|
if not agent_id or agent_id in seen:
|
|
continue
|
|
seen.add(agent_id)
|
|
candidates.append(participant)
|
|
|
|
if candidates:
|
|
async with self._sf() as session:
|
|
await self._idempotent_insert(session, session_id, candidates)
|
|
|
|
# Re-query: always return the authoritative DB state.
|
|
return await self._list_sorted(session_id)
|
|
|
|
async def _idempotent_insert(
|
|
self,
|
|
session: AsyncSession,
|
|
session_id: str,
|
|
candidates: list[dict[str, Any]],
|
|
) -> None:
|
|
"""Execute a conflict-safe INSERT for the given candidate rows."""
|
|
from sqlalchemy import insert as sa_insert
|
|
|
|
now = _now()
|
|
rows = [
|
|
{
|
|
"id": uuid4().hex,
|
|
"session_id": session_id,
|
|
"agent_id": str(p.get("agent_id") or "").strip(),
|
|
"display_name": str(p.get("display_name") or ""),
|
|
"role": str(p.get("role") or "seat"),
|
|
"sort_order": int(p.get("sort_order", 0)),
|
|
"thread_id": None,
|
|
"sandbox_id": None,
|
|
"status": "idle",
|
|
"last_turn_index": None,
|
|
"capability_snapshot_json": p.get("capability_snapshot"),
|
|
"created_at": now,
|
|
"updated_at": now,
|
|
}
|
|
for p in candidates
|
|
]
|
|
|
|
dialect_name = session.bind.dialect.name if session.bind else "sqlite"
|
|
|
|
if dialect_name in ("sqlite", "postgresql"):
|
|
# Use ON CONFLICT DO NOTHING for SQLite and PostgreSQL.
|
|
from sqlalchemy.dialects.postgresql import insert as pg_insert
|
|
from sqlalchemy.dialects.sqlite import insert as sqlite_insert
|
|
|
|
dialect_insert = sqlite_insert if dialect_name == "sqlite" else pg_insert
|
|
stmt = dialect_insert(MafRoundtableParticipantRow).values(rows)
|
|
stmt = stmt.on_conflict_do_nothing(
|
|
index_elements=["session_id", "agent_id"],
|
|
)
|
|
await session.execute(stmt)
|
|
await session.commit()
|
|
elif dialect_name == "mysql":
|
|
# MySQL: INSERT IGNORE skips rows that violate a unique constraint.
|
|
from sqlalchemy.dialects.mysql import insert as mysql_insert
|
|
|
|
stmt = mysql_insert(MafRoundtableParticipantRow).values(rows)
|
|
stmt = stmt.prefix_with("IGNORE")
|
|
await session.execute(stmt)
|
|
await session.commit()
|
|
else:
|
|
# Fallback: try/except IntegrityError per row (rare dialects).
|
|
for row_data in rows:
|
|
try:
|
|
session.add(MafRoundtableParticipantRow(**row_data))
|
|
await session.commit()
|
|
except IntegrityError:
|
|
await session.rollback()
|
|
logger.debug(
|
|
"ensure_participants: skipped duplicate (%s, %s)",
|
|
session_id,
|
|
row_data["agent_id"],
|
|
)
|
|
|
|
async def _list_sorted(self, session_id: str) -> list[dict[str, Any]]:
|
|
"""Return all participants for a session sorted by sort_order, agent_id."""
|
|
async with self._sf() as session:
|
|
rows = (
|
|
await session.execute(
|
|
select(MafRoundtableParticipantRow)
|
|
.where(MafRoundtableParticipantRow.session_id == session_id)
|
|
.order_by(
|
|
MafRoundtableParticipantRow.sort_order.asc(),
|
|
MafRoundtableParticipantRow.agent_id.asc(),
|
|
)
|
|
)
|
|
).scalars().all()
|
|
return [_to_dict(r) for r in rows]
|
|
|
|
async def update_participant_status(
|
|
self, session_id: str, agent_id: str, *, status: str, last_turn_index: int | None = None
|
|
) -> dict[str, Any] | None:
|
|
async with self._sf() as session:
|
|
stmt = (
|
|
select(MafRoundtableParticipantRow)
|
|
.where(
|
|
MafRoundtableParticipantRow.session_id == session_id,
|
|
MafRoundtableParticipantRow.agent_id == agent_id,
|
|
)
|
|
.limit(1)
|
|
)
|
|
row = (await session.execute(stmt)).scalars().first()
|
|
if row is None:
|
|
return None
|
|
row.status = status
|
|
if last_turn_index is not None:
|
|
row.last_turn_index = last_turn_index
|
|
row.updated_at = _now()
|
|
await session.commit()
|
|
return _to_dict(row)
|
|
|
|
async def bind_thread(
|
|
self, session_id: str, agent_id: str, *, thread_id: str, sandbox_id: str | None = None
|
|
) -> dict[str, Any] | None:
|
|
async with self._sf() as session:
|
|
stmt = (
|
|
select(MafRoundtableParticipantRow)
|
|
.where(
|
|
MafRoundtableParticipantRow.session_id == session_id,
|
|
MafRoundtableParticipantRow.agent_id == agent_id,
|
|
)
|
|
.limit(1)
|
|
)
|
|
row = (await session.execute(stmt)).scalars().first()
|
|
if row is None:
|
|
return None
|
|
row.thread_id = thread_id
|
|
if sandbox_id is not None:
|
|
row.sandbox_id = sandbox_id
|
|
row.updated_at = _now()
|
|
await session.commit()
|
|
return _to_dict(row)
|