deerflow-code/offline-backend-20260512/backend/packages/harness/deerflow/persistence/maf_roundtable/model.py

"""ORM models for the MAF (Microsoft Agent Framework) roundtable experiment.

Nine independent tables — **no foreign keys to the legacy ``roundtable_*``
tables** (``roundtable_jobs`` / ``roundtable_drafts`` /
``roundtable_task_drafts``). The MAF experiment is fully isolated: it has its
own session / participant / run / event / action / lease / approval tables, keyed by
its own surrogate ids, so the legacy multi-agent roundtable keeps its current
behavior, schema, and recovery semantics untouched.

Tables (see the implementation plan §8):

- ``maf_roundtable_sessions``     — one MAF roundtable session (task + mode + capability snapshot)
- ``maf_roundtable_participants`` — manager + seat agents bound to a session
- ``maf_roundtable_runs``         — one execution run of a session
- ``maf_roundtable_events``       — monotonic per-run event stream (``(run_id, seq)`` unique)
- ``maf_roundtable_actions``      — external write / tool actions (idempotency-keyed)
- ``maf_roundtable_action_leases`` — active target-write leases (one per target key)
- ``maf_roundtable_approvals``    — pending human-approval requests

User isolation: ``maf_roundtable_sessions.user_id`` is the owner; all child
rows are filtered through the owning session. ``user_id`` on child tables
(``actions``) is audit-only. Admins bypass the owner filter when
``maf_roundtable.admin_diagnostic_access`` is true.

All JSON / large-text columns use the portable types so the schema is identical
across sqlite / postgres / mysql (see ``deerflow.persistence.types``).
"""

from __future__ import annotations

from datetime import UTC, datetime

from sqlalchemy import (
    CheckConstraint,
    ForeignKey,
    Integer,
    String,
    UniqueConstraint,
)
from sqlalchemy.orm import Mapped, mapped_column

from deerflow.persistence.base import Base
from deerflow.persistence.types import BeijingDateTime, PortableJSON, PortableLongText


def _now() -> datetime:
    return datetime.now(UTC)


# ── 1. sessions ───────────────────────────────────────────────────────────


class MafRoundtableSessionRow(Base):
    """One MAF roundtable session (the user's task + chosen mode + frozen capability snapshot).

    Owner-scoped by ``user_id``. ``capability_snapshot_json`` freezes the
    agents / tools / skills configuration at creation time so a run can be
    reproduced even if the agent config changes later. ``fact_ledger_json``
    holds the shared fact ledger (confirmed facts / tool results / pending /
    failed actions) updated as the run progresses.
    """

    __tablename__ = "maf_roundtable_sessions"

    id: Mapped[str] = mapped_column(String(64), primary_key=True)
    user_id: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
    title: Mapped[str] = mapped_column(String(255), default="")
    task_text: Mapped[str] = mapped_column(PortableLongText, nullable=False)
    task_payload_json: Mapped[dict | None] = mapped_column(PortableJSON(), nullable=True)
    # structured | group_chat
    mode: Mapped[str] = mapped_column(String(16), default="structured", index=True)
    # draft | running | waiting_approval | waiting_input | completed | failed | cancelled
    status: Mapped[str] = mapped_column(String(32), default="draft", index=True)
    model_name: Mapped[str | None] = mapped_column(String(128), nullable=True)
    reasoning_effort: Mapped[str | None] = mapped_column(String(32), nullable=True)
    manager_agent_id: Mapped[str | None] = mapped_column(String(64), nullable=True)
    max_rounds: Mapped[int] = mapped_column(Integer, default=8)
    # auto | confirm_high_risk | confirm_all_writes
    write_policy: Mapped[str] = mapped_column(String(32), default="confirm_high_risk")
    capability_snapshot_json: Mapped[dict | None] = mapped_column(PortableJSON(), nullable=True)
    fact_ledger_json: Mapped[dict | None] = mapped_column(PortableJSON(), nullable=True)
    final_result_json: Mapped[dict | None] = mapped_column(PortableJSON(), nullable=True)
    created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), nullable=False, default=_now)
    updated_at: Mapped[datetime] = mapped_column(
        BeijingDateTime(), nullable=False, default=_now, onupdate=_now
    )


# ── 2. participants ──────────────────────────────────────────────────────


class MafRoundtableParticipantRow(Base):
    """Manager + seat agents bound to a session.

    One row per participant agent. ``role`` is ``manager`` / ``seat``.
    ``thread_id`` / ``sandbox_id`` bind the participant to its own DeerFlow
    thread / sandbox (one seat = one thread = one sandbox). ``capability_snapshot_json``
    freezes the agent's tools / skills at session creation time.
    """

    __tablename__ = "maf_roundtable_participants"
    __table_args__ = (
        UniqueConstraint("session_id", "agent_id", name="uq_maf_participants_session_agent"),
    )

    id: Mapped[str] = mapped_column(String(64), primary_key=True)
    session_id: Mapped[str] = mapped_column(
        String(64), ForeignKey("maf_roundtable_sessions.id", ondelete="CASCADE"), nullable=False, index=True
    )
    agent_id: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
    display_name: Mapped[str] = mapped_column(String(255), default="")
    # manager | seat
    role: Mapped[str] = mapped_column(String(16), default="seat")
    sort_order: Mapped[int] = mapped_column(Integer, default=0)
    thread_id: Mapped[str | None] = mapped_column(String(64), nullable=True)
    sandbox_id: Mapped[str | None] = mapped_column(String(64), nullable=True)
    # idle | running | done | failed | cancelled
    status: Mapped[str] = mapped_column(String(16), default="idle")
    last_turn_index: Mapped[int | None] = mapped_column(Integer, nullable=True)
    capability_snapshot_json: Mapped[dict | None] = mapped_column(PortableJSON(), nullable=True)
    created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), nullable=False, default=_now)
    updated_at: Mapped[datetime] = mapped_column(
        BeijingDateTime(), nullable=False, default=_now, onupdate=_now
    )


# ── 3. runs ───────────────────────────────────────────────────────────────


class MafRoundtableRunRow(Base):
    """One execution run of a session.

    ``maf_checkpoint_json`` stores the MAF workflow checkpoint (or a reference)
    for resume-after-restart. ``last_event_seq`` is a denormalized cache of the
    latest event seq for fast polling (the authoritative seq lives in
    ``maf_roundtable_events``). ``token_usage_json`` accumulates per-run token
    usage. ``latency_ms`` is the wall-clock duration.
    """

    __tablename__ = "maf_roundtable_runs"

    id: Mapped[str] = mapped_column(String(64), primary_key=True)
    session_id: Mapped[str] = mapped_column(
        String(64), ForeignKey("maf_roundtable_sessions.id", ondelete="CASCADE"), nullable=False, index=True
    )
    # pending | running | waiting_approval | waiting_input | completed | failed | cancelled
    status: Mapped[str] = mapped_column(String(32), default="pending", index=True)
    workflow_version: Mapped[str | None] = mapped_column(String(64), nullable=True)
    maf_checkpoint_json: Mapped[dict | None] = mapped_column(PortableJSON(), nullable=True)
    last_event_seq: Mapped[int] = mapped_column(Integer, default=0)
    started_at: Mapped[datetime | None] = mapped_column(BeijingDateTime(), nullable=True)
    finished_at: Mapped[datetime | None] = mapped_column(BeijingDateTime(), nullable=True)
    error_code: Mapped[str | None] = mapped_column(String(64), nullable=True)
    error_message_safe: Mapped[str | None] = mapped_column(PortableLongText, nullable=True)
    token_usage_json: Mapped[dict | None] = mapped_column(PortableJSON(), nullable=True)
    latency_ms: Mapped[int | None] = mapped_column(Integer, nullable=True)
    created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), nullable=False, default=_now)
    updated_at: Mapped[datetime] = mapped_column(
        BeijingDateTime(), nullable=False, default=_now, onupdate=_now
    )


# ── 4. events ─────────────────────────────────────────────────────────────


class MafRoundtableEventRow(Base):
    """Monotonic per-run event stream.

    ``(run_id, seq)`` is unique — seq is monotonically increasing per run. The
    store allocates seq by ``SELECT MAX(seq) + 1`` and retries on
    ``IntegrityError`` so concurrent appends from parallel seats cannot collide
    or skip seq values. ``payload_json`` carries the event-specific body (tool
    result, agent message delta, fact-recorded, etc.); sensitive fields must
    already be redacted by the writer.
    """

    __tablename__ = "maf_roundtable_events"
    __table_args__ = (
        UniqueConstraint("run_id", "seq", name="uq_maf_events_run_seq"),
        # Dedup key for at-most-once event persistence. The authoritative
        # dedup is this DB constraint (survives process restarts) — not the
        # in-process ``_seen_source_ids`` set, which is only a fast-path cache.
        # ``source_event_id`` is nullable; rows with NULL are not deduped
        # (SQL treats NULLs as distinct in a UNIQUE constraint on sqlite /
        # postgres / mysql), so events without a source id stay append-only.
        UniqueConstraint("run_id", "source_event_id", name="uq_maf_events_run_source"),
    )

    id: Mapped[str] = mapped_column(String(64), primary_key=True)
    session_id: Mapped[str] = mapped_column(
        String(64), ForeignKey("maf_roundtable_sessions.id", ondelete="CASCADE"), nullable=False, index=True
    )
    run_id: Mapped[str] = mapped_column(
        String(64), ForeignKey("maf_roundtable_runs.id", ondelete="CASCADE"), nullable=False, index=True
    )
    seq: Mapped[int] = mapped_column(Integer, nullable=False)
    # run.started | run.status_changed | workflow.checkpointed | manager.decision |
    # agent.turn_started | agent.message.delta | agent.message.completed |
    # skill.selected | skill.started | skill.completed | tool.requested | tool.started |
    # tool.progress | tool.completed | tool.failed | action.waiting_approval |
    # action.approved | action.rejected | artifact.created | fact.recorded |
    # run.completed | run.failed
    event_type: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
    agent_id: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
    turn_index: Mapped[int | None] = mapped_column(Integer, nullable=True)
    # Upstream identity of the source runtime event (e.g. LangGraph message id
    # / tool_call id). Used by the ``(run_id, source_event_id)`` unique
    # constraint for durable at-most-once dedup across restarts.
    source_event_id: Mapped[str | None] = mapped_column(String(191), nullable=True, index=True)
    payload_json: Mapped[dict | None] = mapped_column(PortableJSON(), nullable=True)
    created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), nullable=False, default=_now, index=True)


# ── 5. actions ─────────────────────────────────────────────────────────────


class MafRoundtableActionRow(Base):
    """External write / tool action record (idempotency-keyed).

    The Action Coordinator (Phase 5) uses ``idempotency_key`` to guarantee
    at-most-once execution across retries / restarts. ``target_key`` carries a
    per-business-object lease so parallel seats writing the same object are
    serialized. ``risk_level`` is R0–R4. ``status`` is the action state machine:
    pending | running | succeeded | failed | unknown | waiting_approval |
    approved | rejected | compensated. ``arguments_redacted_json`` /
    ``result_redacted_json`` must already have credentials / tokens / cookies
    stripped by the writer.
    """

    __tablename__ = "maf_roundtable_actions"
    __table_args__ = (
        UniqueConstraint("idempotency_key", name="uq_maf_actions_idempotency"),
    )

    id: Mapped[str] = mapped_column(String(64), primary_key=True)
    session_id: Mapped[str] = mapped_column(
        String(64), ForeignKey("maf_roundtable_sessions.id", ondelete="CASCADE"), nullable=False, index=True
    )
    run_id: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
    agent_id: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
    tool_name: Mapped[str] = mapped_column(String(128), nullable=False, index=True)
    logical_action_id: Mapped[str | None] = mapped_column(String(128), nullable=True)
    # The seat turn that produced this action. Persisted (not only embedded in
    # the idempotency_key) so pending-approval matching can key on
    # run_id + agent_id + turn_index + logical_action_id precisely — never
    # mixing a seat's earlier-turn waiting actions into the current approval.
    turn_index: Mapped[int | None] = mapped_column(Integer, nullable=True)
    idempotency_key: Mapped[str] = mapped_column(String(191), nullable=False, unique=True)
    target_key: Mapped[str | None] = mapped_column(String(255), nullable=True, index=True)
    # R0 | R1 | R2 | R3 | R4
    risk_level: Mapped[str] = mapped_column(String(8), default="R0")
    # pending | running | succeeded | failed | unknown | waiting_approval | approved | rejected | compensated
    status: Mapped[str] = mapped_column(String(32), default="pending", index=True)
    arguments_redacted_json: Mapped[dict | None] = mapped_column(PortableJSON(), nullable=True)
    result_redacted_json: Mapped[dict | None] = mapped_column(PortableJSON(), nullable=True)
    external_object_id: Mapped[str | None] = mapped_column(String(255), nullable=True)
    approval_id: Mapped[str | None] = mapped_column(String(64), nullable=True)
    attempt_count: Mapped[int] = mapped_column(Integer, default=0)
    started_at: Mapped[datetime | None] = mapped_column(BeijingDateTime(), nullable=True)
    finished_at: Mapped[datetime | None] = mapped_column(BeijingDateTime(), nullable=True)
    created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), nullable=False, default=_now)
    updated_at: Mapped[datetime] = mapped_column(
        BeijingDateTime(), nullable=False, default=_now, onupdate=_now
    )


# ── 6. target leases ──────────────────────────────────────────────────────


class MafRoundtableActionLeaseRow(Base):
    """One active, short-lived lease for a server-derived external target key.

    The key is opaque (a stable SHA-256 digest) and is never sourced from a
    browser/model field.  ``target_key`` is the primary key, which makes the
    database — not an in-process mutex — arbitrate concurrent writes across
    workers and restarts.  A lease belongs to exactly one action and is removed
    with a known terminal result; an ``unknown`` result deliberately retains a
    manual-reconciliation barrier.
    """

    __tablename__ = "maf_roundtable_action_leases"
    __table_args__ = (
        UniqueConstraint("action_id", name="uq_maf_action_leases_action"),
    )

    target_key: Mapped[str] = mapped_column(String(255), primary_key=True)
    action_id: Mapped[str] = mapped_column(
        String(64),
        ForeignKey("maf_roundtable_actions.id", ondelete="CASCADE"),
        nullable=False,
        index=True,
    )
    lease_expires_at: Mapped[datetime] = mapped_column(
        BeijingDateTime(), nullable=False, index=True
    )
    created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), nullable=False, default=_now)
    updated_at: Mapped[datetime] = mapped_column(
        BeijingDateTime(), nullable=False, default=_now, onupdate=_now
    )


# ── 7. approvals ──────────────────────────────────────────────────────────


class MafRoundtableApprovalRow(Base):
    """Pending human-approval request for a high-risk external write.

    ``status`` is ``pending | approved | rejected | expired``. ``payload_redacted_json``
    carries the action summary the user sees (redacted). ``expires_at`` is the
    deadline after which the approval auto-expires and the action is marked
    ``rejected`` (or ``unknown`` if it may already have been sent).
    """

    __tablename__ = "maf_roundtable_approvals"

    id: Mapped[str] = mapped_column(String(64), primary_key=True)
    session_id: Mapped[str] = mapped_column(
        String(64), ForeignKey("maf_roundtable_sessions.id", ondelete="CASCADE"), nullable=False, index=True
    )
    run_id: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
    action_id: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
    requested_by_agent_id: Mapped[str | None] = mapped_column(String(64), nullable=True)
    # pending | approved | rejected | expired
    status: Mapped[str] = mapped_column(String(16), default="pending", index=True)
    summary: Mapped[str] = mapped_column(PortableLongText, default="")
    payload_redacted_json: Mapped[dict | None] = mapped_column(PortableJSON(), nullable=True)
    decided_by: Mapped[str | None] = mapped_column(String(64), nullable=True)
    decided_at: Mapped[datetime | None] = mapped_column(BeijingDateTime(), nullable=True)
    expires_at: Mapped[datetime | None] = mapped_column(BeijingDateTime(), nullable=True, index=True)
    created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), nullable=False, default=_now)
    updated_at: Mapped[datetime] = mapped_column(
        BeijingDateTime(), nullable=False, default=_now, onupdate=_now
    )


# ── 8. seat runs (Phase 2: MAF↔DeerFlow run mapping) ─────────────────────


class MafRoundtableSeatRunRow(Base):
    """Persistent mapping between a MAF workflow seat turn and its DeerFlow run.

    Replaces the in-memory ``_active_runs`` dict so that after a service restart
    the system can still locate and cancel outstanding DeerFlow runs, and the
    MAF checkpoint can reference which seat turns completed.

    Phase-3 unique constraint: ``(maf_run_id, agent_id, turn_index,
    execution_stage)``. ``turn_index`` is the business round (kept stable for
    frontend display / approval matching / event tracing); ``execution_stage``
    disambiguates the distinct DeerFlow executions that happen within one
    business round (planner / expert / format_repair / cross_review /
    finalizer / group_chat). The business turn_index is NEVER mutated to dodge
    a collision.
    """

    __tablename__ = "maf_roundtable_seat_runs"
    __table_args__ = (
        UniqueConstraint(
            "maf_run_id",
            "agent_id",
            "turn_index",
            "execution_stage",
            name="uq_maf_seat_runs_run_agent_turn_stage",
        ),
        # Phase-3 P1-2: enforce the closed execution_stage enum at the DB layer.
        # Mirrors seat_runs.VALID_EXECUTION_STAGES — a typo (e.g. "exprt") would
        # otherwise silently bypass the four-column idempotency key by creating
        # a brand-new stage. This CHECK keeps both fresh creates and any raw
        # migration / manual SQL honest. Keep in sync with VALID_EXECUTION_STAGES.
        CheckConstraint(
            "execution_stage IN ('legacy','planner','expert','format_repair',"
            "'cross_review','cross_review_repair','finalizer','group_chat')",
            name="ck_maf_seat_runs_execution_stage",
        ),
    )

    id: Mapped[str] = mapped_column(String(64), primary_key=True)
    maf_run_id: Mapped[str] = mapped_column(
        String(64), ForeignKey("maf_roundtable_runs.id", ondelete="CASCADE"), nullable=False, index=True
    )
    agent_id: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
    turn_index: Mapped[int] = mapped_column(Integer, nullable=False)
    # Phase-3: which kind of DeerFlow execution this row represents. Pre-Phase-3
    # rows are migrated to ``legacy``. Closed enum (see seat_runs.VALID_EXECUTION_STAGES):
    # planner | expert | format_repair | cross_review | cross_review_repair |
    # finalizer | group_chat | legacy.
    execution_stage: Mapped[str] = mapped_column(
        String(32), nullable=False, default="legacy", server_default="legacy"
    )
    deerflow_run_id: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
    deerflow_thread_id: Mapped[str | None] = mapped_column(String(64), nullable=True)
    sandbox_id: Mapped[str | None] = mapped_column(String(64), nullable=True)
    workflow_checkpoint_id: Mapped[str | None] = mapped_column(String(128), nullable=True)
    # pending | running | completed | failed | cancelled
    status: Mapped[str] = mapped_column(String(32), default="pending", index=True)
    # Phase-3 P0-1: a sanitized, replayable snapshot of the SeatResponse written
    # atomically with the completed transition. Survives a crash that happens
    # after the seat run is marked completed but before the workflow checkpoint
    # is persisted, so recovery never needs to re-call DeerFlow.
    result_snapshot: Mapped[dict | None] = mapped_column(PortableJSON, nullable=True)
    started_at: Mapped[datetime | None] = mapped_column(BeijingDateTime(), nullable=True)
    finished_at: Mapped[datetime | None] = mapped_column(BeijingDateTime(), nullable=True)
    error: Mapped[str | None] = mapped_column(PortableLongText, nullable=True)
    created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), nullable=False, default=_now)
    updated_at: Mapped[datetime] = mapped_column(
        BeijingDateTime(), nullable=False, default=_now, onupdate=_now
    )


# ── 9. action secrets (encrypted actual execution params, server-only) ────


class MafRoundtableActionSecretRow(Base):
    """Encrypted actual execution parameters for a protected action.

    Stores the **real** (un-redacted) tool-call arguments so that after
    approval or a service restart the system can recover and execute the
    approved action with the original parameters. This table is **never**
    joined in any read query exposed to the API, events, logs, or frontend —
    it is server-only. The ActionCoordinator encrypts the payload before
    writing and decrypts only at execution time.

    One row per action (unique ``action_id``). The encryption key is derived
    from the deployment's secret key; the ciphertext is opaque to the store.
    """

    __tablename__ = "maf_roundtable_action_secrets"
    __table_args__ = (
        UniqueConstraint("action_id", name="uq_maf_action_secrets_action"),
    )

    id: Mapped[str] = mapped_column(String(64), primary_key=True)
    action_id: Mapped[str] = mapped_column(
        String(64), ForeignKey("maf_roundtable_actions.id", ondelete="CASCADE"),
        nullable=False, index=True,
    )
    arguments_encrypted: Mapped[str] = mapped_column(PortableLongText, nullable=False)
    created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), nullable=False, default=_now)


# ── 10. conversations (V2: user-visible conversation aggregation) ─────────


class MafRoundtableConversationRow(Base):
    """One user-visible MAF roundtable conversation.

    This is the **only** entity the frontend knows about. A conversation can
    span multiple user messages, each of which may trigger an internal
    ``maf_roundtable_runs`` execution. The user never sees session/run/thread
    ids — only ``conversation_id``.

    ``active_run_id`` points to the currently executing internal run (nullable
    when idle). ``safe_status`` is a coarse user-facing status derived from
    the active run's state (idle / running / waiting_input / waiting_approval
    / completed / failed / cancelled).
    """

    __tablename__ = "maf_roundtable_conversations"

    id: Mapped[str] = mapped_column(String(64), primary_key=True)
    owner_user_id: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
    title: Mapped[str] = mapped_column(String(255), default="")
    task_context_json: Mapped[dict | None] = mapped_column(PortableJSON(), nullable=True)
    # idle | running | waiting_input | waiting_approval | completed | failed | cancelled
    safe_status: Mapped[str] = mapped_column(String(32), default="idle", index=True)
    active_run_id: Mapped[str | None] = mapped_column(
        String(64),
        ForeignKey("maf_roundtable_runs.id", ondelete="SET NULL"),
        nullable=True,
    )
    # Internal session this conversation is linked to (lazy-created on first message).
    internal_session_id: Mapped[str | None] = mapped_column(
        String(64),
        ForeignKey("maf_roundtable_sessions.id", ondelete="SET NULL"),
        nullable=True,
    )
    last_stream_seq: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
    created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), nullable=False, default=_now)
    updated_at: Mapped[datetime] = mapped_column(
        BeijingDateTime(), nullable=False, default=_now, onupdate=_now
    )


# ── 11. conversation messages ─────────────────────────────────────────────


class MafRoundtableMessageRow(Base):
    """Stable message history for a conversation.

    Every user-visible bubble (user message, manager message, seat message,
    clarification card, approval card, result card) has exactly one row here.
    ``seq`` is a conversation-global monotonic counter (unique per conversation)
    used for SSE cursor and ordering. ``message_kind`` classifies the bubble
    type for the frontend reducer.

    ``run_id`` links the message to the internal run that produced it (nullable
    for user messages that haven't started a run yet). ``agent_id`` identifies
    the speaking agent (null for user messages).
    """

    __tablename__ = "maf_roundtable_messages"
    __table_args__ = (
        UniqueConstraint("conversation_id", "seq", name="uq_maf_messages_conv_seq"),
    )

    id: Mapped[str] = mapped_column(String(64), primary_key=True)
    conversation_id: Mapped[str] = mapped_column(
        String(64), ForeignKey("maf_roundtable_conversations.id", ondelete="CASCADE"),
        nullable=False, index=True,
    )
    seq: Mapped[int] = mapped_column(Integer, nullable=False)
    # user | manager | seat | system | clarification | approval | result | tool_progress
    message_kind: Mapped[str] = mapped_column(String(32), nullable=False, default="user")
    role: Mapped[str] = mapped_column(String(16), nullable=False, default="user")
    agent_id: Mapped[str | None] = mapped_column(String(64), nullable=True)
    run_id: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
    # Stable message identity for SSE delta merging (e.g. LangGraph message id).
    message_ref: Mapped[str | None] = mapped_column(String(191), nullable=True, index=True)
    content: Mapped[str] = mapped_column(PortableLongText, default="")
    metadata_json: Mapped[dict | None] = mapped_column(PortableJSON(), nullable=True)
    created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), nullable=False, default=_now, index=True)


# ── 12. conversation events (SSE stream, conversation-global seq) ─────────


class MafRoundtableConversationEventRow(Base):
    """Conversation-dimension event stream for SSE.

    Unlike ``maf_roundtable_events`` (per-run seq), this table uses a
    **conversation-global** monotonic ``seq`` so the frontend only needs one
    cursor (``after_seq``) regardless of how many internal runs happen.
    The SSE hub replays from this table on connect/reconnect, then subscribes
    to real-time notifications.

    ``event_type`` follows the §4.3 contract (conversation.snapshot,
    user.message.created, manager.intent, clarification.required, etc.).
    ``data_json`` is the already-sanitized event payload safe for the browser.
    """

    __tablename__ = "maf_roundtable_conversation_events"
    __table_args__ = (
        UniqueConstraint("conversation_id", "seq", name="uq_maf_conv_events_conv_seq"),
        # P0-4: dedup key for projection idempotency. The source audit event's
        # stable id (e.g. LangGraph event id / tool_call id) prevents duplicate
        # conversation events when the bridge retries or the adapter re-projects
        # after a restart. NULL source_event_ids are not deduped (SQL treats
        # NULLs as distinct in UNIQUE constraints).
        UniqueConstraint("conversation_id", "source_event_id", name="uq_maf_conv_events_conv_source"),
    )

    id: Mapped[str] = mapped_column(String(64), primary_key=True)
    conversation_id: Mapped[str] = mapped_column(
        String(64), ForeignKey("maf_roundtable_conversations.id", ondelete="CASCADE"),
        nullable=False, index=True,
    )
    seq: Mapped[int] = mapped_column(Integer, nullable=False)
    run_id: Mapped[str | None] = mapped_column(String(64), nullable=True)
    # Stable upstream id from the source audit event (maf_roundtable_events.id
    # or the LangGraph source_event_id). Used for projection dedup via the
    # (conversation_id, source_event_id) unique constraint.
    source_event_id: Mapped[str | None] = mapped_column(String(191), nullable=True, index=True)
    event_type: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
    data_json: Mapped[dict | None] = mapped_column(PortableJSON(), nullable=True)
    created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), nullable=False, default=_now, index=True)


__all__ = [
    "MafRoundtableActionRow",
    "MafRoundtableActionLeaseRow",
    "MafRoundtableActionSecretRow",
    "MafRoundtableApprovalRow",
    "MafRoundtableConversationEventRow",
    "MafRoundtableConversationRow",
    "MafRoundtableEventRow",
    "MafRoundtableMessageRow",
    "MafRoundtableParticipantRow",
    "MafRoundtableRunRow",
    "MafRoundtableSeatRunRow",
    "MafRoundtableSessionRow",
]