董宏凯andCursor
e690475e6e
Ship an independent, feature-flag-gated MAF roundtable skeleton with six isolated tables, atomic single-transaction run launch/cancel, owner-scoped read/write split, shared config enums, and HTTP/concurrency regression tests. Co-authored-by: Cursor <cursoragent@cursor.com>
202 lines
8.7 KiB
Python
202 lines
8.7 KiB
Python
"""Session repository for the MAF roundtable experiment (owner-scoped CRUD).
|
|
|
|
Owner-scoped by ``user_id``: every read / update / delete checks the session's
|
|
owner against the caller. Admins bypass the owner filter when
|
|
``admin_diagnostic_access`` is true (handled in the service / router layer, not
|
|
here — the repo exposes both scoped and unscoped lookups).
|
|
|
|
Capability snapshot: ``capability_snapshot_json`` is frozen at creation time
|
|
(``create_session``) and never mutated by later updates — it is the audit /
|
|
reproducibility record of which agents / tools / skills were configured when
|
|
the session started. ``fact_ledger_json`` / ``final_result_json`` are mutable
|
|
and updated as the run progresses.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import UTC, datetime
|
|
from typing import Any
|
|
from uuid import uuid4
|
|
|
|
from sqlalchemy import delete, func, select, update
|
|
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
|
|
|
|
from deerflow.persistence.maf_roundtable.model import MafRoundtableSessionRow
|
|
|
|
|
|
def _now() -> datetime:
|
|
return datetime.now(UTC)
|
|
|
|
|
|
def _to_dict(row: MafRoundtableSessionRow) -> dict[str, Any]:
|
|
return {
|
|
"id": row.id,
|
|
"user_id": row.user_id,
|
|
"title": row.title,
|
|
"task_text": row.task_text,
|
|
"task_payload": row.task_payload_json,
|
|
"mode": row.mode,
|
|
"status": row.status,
|
|
"model_name": row.model_name,
|
|
"reasoning_effort": row.reasoning_effort,
|
|
"manager_agent_id": row.manager_agent_id,
|
|
"max_rounds": row.max_rounds,
|
|
"write_policy": row.write_policy,
|
|
"capability_snapshot": row.capability_snapshot_json,
|
|
"fact_ledger": row.fact_ledger_json,
|
|
"final_result": row.final_result_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 MafSessionRepository:
|
|
"""Owner-scoped CRUD for MAF roundtable sessions."""
|
|
|
|
def __init__(self, session_factory: async_sessionmaker[AsyncSession]) -> None:
|
|
self._sf = session_factory
|
|
|
|
async def create_session(self, data: dict[str, Any]) -> dict[str, Any]:
|
|
"""Create a new session. ``data`` must include ``user_id`` and ``task_text``."""
|
|
now = _now()
|
|
row = MafRoundtableSessionRow(
|
|
id=data.get("id") or uuid4().hex,
|
|
user_id=str(data["user_id"]),
|
|
title=str(data.get("title") or ""),
|
|
task_text=str(data["task_text"]),
|
|
task_payload_json=data.get("task_payload"),
|
|
mode=str(data.get("mode") or "structured"),
|
|
status=str(data.get("status") or "draft"),
|
|
model_name=data.get("model_name"),
|
|
reasoning_effort=data.get("reasoning_effort"),
|
|
manager_agent_id=data.get("manager_agent_id"),
|
|
max_rounds=int(data.get("max_rounds", 8)),
|
|
write_policy=str(data.get("write_policy") or "confirm_high_risk"),
|
|
capability_snapshot_json=data.get("capability_snapshot"),
|
|
fact_ledger_json=data.get("fact_ledger"),
|
|
final_result_json=data.get("final_result"),
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
async with self._sf() as session:
|
|
session.add(row)
|
|
await session.commit()
|
|
return await self.get_session(row.id, user_id=None) # type: ignore[arg-type]
|
|
|
|
async def get_session(
|
|
self, session_id: str, *, user_id: str | None = "REQUIRED"
|
|
) -> dict[str, Any] | None:
|
|
"""Read one session.
|
|
|
|
``user_id`` scoping:
|
|
- ``"REQUIRED"`` (default): enforce owner scope — returns ``None`` if the
|
|
session is missing **or** owned by another user.
|
|
- ``None``: bypass owner scope (admin / internal path) — returns the
|
|
row regardless of owner.
|
|
"""
|
|
async with self._sf() as session:
|
|
stmt = select(MafRoundtableSessionRow).where(MafRoundtableSessionRow.id == session_id).limit(1)
|
|
row = (await session.execute(stmt)).scalars().first()
|
|
if row is None:
|
|
return None
|
|
if user_id is not None and row.user_id != user_id:
|
|
return None
|
|
return _to_dict(row)
|
|
|
|
async def list_sessions(
|
|
self,
|
|
*,
|
|
user_id: str | None = None,
|
|
status: str | None = None,
|
|
mode: str | None = None,
|
|
limit: int = 100,
|
|
offset: int = 0,
|
|
) -> list[dict[str, Any]]:
|
|
"""List sessions, newest-first.
|
|
|
|
``user_id=None`` lists across all users (admin / diagnostic). Otherwise
|
|
scoped to the caller. ``status`` / ``mode`` are optional filters.
|
|
"""
|
|
stmt = select(MafRoundtableSessionRow)
|
|
if user_id is not None:
|
|
stmt = stmt.where(MafRoundtableSessionRow.user_id == user_id)
|
|
if status is not None:
|
|
stmt = stmt.where(MafRoundtableSessionRow.status == status)
|
|
if mode is not None:
|
|
stmt = stmt.where(MafRoundtableSessionRow.mode == mode)
|
|
stmt = stmt.order_by(MafRoundtableSessionRow.created_at.desc()).limit(limit).offset(offset)
|
|
async with self._sf() as session:
|
|
rows = (await session.execute(stmt)).scalars().all()
|
|
return [_to_dict(r) for r in rows]
|
|
|
|
async def count_sessions(
|
|
self, *, user_id: str | None = None, status: str | None = None, mode: str | None = None
|
|
) -> int:
|
|
stmt = select(func.count()).select_from(MafRoundtableSessionRow)
|
|
if user_id is not None:
|
|
stmt = stmt.where(MafRoundtableSessionRow.user_id == user_id)
|
|
if status is not None:
|
|
stmt = stmt.where(MafRoundtableSessionRow.status == status)
|
|
if mode is not None:
|
|
stmt = stmt.where(MafRoundtableSessionRow.mode == mode)
|
|
async with self._sf() as session:
|
|
return int((await session.execute(stmt)).scalar_one())
|
|
|
|
async def update_session(
|
|
self, session_id: str, *, user_id: str | None, fields: dict[str, Any]
|
|
) -> dict[str, Any] | None:
|
|
"""Partial update. Only allowed when the session is not yet ``running``
|
|
(enforced by the service layer). ``capability_snapshot`` is **never**
|
|
mutated here — it is frozen at creation. Returns ``None`` if the session
|
|
is missing or owned by another user.
|
|
"""
|
|
async with self._sf() as session:
|
|
stmt = select(MafRoundtableSessionRow).where(MafRoundtableSessionRow.id == session_id).limit(1)
|
|
row = (await session.execute(stmt)).scalars().first()
|
|
if row is None:
|
|
return None
|
|
if user_id is not None and row.user_id != user_id:
|
|
return None
|
|
for key in ("title", "task_text", "task_payload", "mode", "status", "model_name",
|
|
"reasoning_effort", "manager_agent_id", "max_rounds", "write_policy",
|
|
"fact_ledger", "final_result"):
|
|
if key in fields:
|
|
value = fields[key]
|
|
col = {
|
|
"task_payload": "task_payload_json",
|
|
"capability_snapshot": "capability_snapshot_json",
|
|
"fact_ledger": "fact_ledger_json",
|
|
"final_result": "final_result_json",
|
|
}.get(key, key)
|
|
setattr(row, col, value)
|
|
row.updated_at = _now()
|
|
await session.commit()
|
|
return _to_dict(row)
|
|
|
|
async def delete_session(self, session_id: str, *, user_id: str | None) -> bool:
|
|
"""Delete a session (and its children via FK CASCADE). Returns ``True``
|
|
if a row was deleted, ``False`` if missing or owned by another user.
|
|
"""
|
|
async with self._sf() as session:
|
|
stmt = select(MafRoundtableSessionRow).where(MafRoundtableSessionRow.id == session_id).limit(1)
|
|
row = (await session.execute(stmt)).scalars().first()
|
|
if row is None:
|
|
return False
|
|
if user_id is not None and row.user_id != user_id:
|
|
return False
|
|
await session.execute(
|
|
delete(MafRoundtableSessionRow).where(MafRoundtableSessionRow.id == session_id)
|
|
)
|
|
await session.commit()
|
|
return True
|
|
|
|
async def set_status(self, session_id: str, status: str) -> None:
|
|
"""Internal: update only the status column (used by the run lifecycle)."""
|
|
async with self._sf() as session:
|
|
await session.execute(
|
|
update(MafRoundtableSessionRow)
|
|
.where(MafRoundtableSessionRow.id == session_id)
|
|
.values(status=status, updated_at=_now())
|
|
)
|
|
await session.commit()
|