Files
董宏凯 1aeee86984 fix(maf): harden stage-4 quality gate (P0/P1)
P0-1 migration: detect the same-named stage-3 execution_stage CHECK by sqltext (must contain rework + manager_review), SQLite rebuild / PG-MySQL drop+add, PG boolean default TRUE, validate_schema checks constraint content + seat_tasks status CHECK/unique/FKs; tests start from a real stage-3 schema.

P0-2: a missing/invalid manager review (timeout/exception/empty/invalid JSON/missing per-seat decision) is never an acceptance - system rework under the cap, contract fallback (reassign/controlled_gap/ask_user) afterwards, source=system events.

P0-3: unaccepted deliveries are isolated from the fact ledger, cross-review and final report (deferred merge, delivery_json envelope, accepted_task_results in checkpoint, accepted-only synthesis, no turn_results fallback).

P0-4: seat-task init is atomic get-or-create over the full expected set; recovery fills missing tasks without overwriting frozen contracts; quality gate blocks missing required tasks.

P1: zero config values honoured; duplicate detection persists SHA-256 conclusion digests in history (survives restarts); findings never satisfy evidence, evidence_refs must resolve to trusted ledger refs (unverified_evidence_ref); conversation SSE projection tests added and manager/finalizer messages are no longer mislabelled as seat.delivery.submitted.
2026-08-07 00:04:55 +08:00

642 lines
33 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""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 (
Boolean,
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',"
"'rework','manager_review')",
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
)
# ── 8b. seat tasks (Stage 4 / R3-Q quality state machine) ─────────────────
class MafRoundtableSeatTaskRow(Base):
"""One seat task with its frozen contract and quality lifecycle.
Stage 4 (R3-Q): each manager dispatch assignment becomes one seat task.
The row persists the whole acceptance state machine
(``assigned → delivering → submitted → manager_review →
accepted / rework / reassigned / exhausted / controlled_gap``) together
with the frozen ``contract_json``, the latest ``delivery_json``,
``deterministic_check_json``, ``review_decision_json`` and a per-attempt
``history_json``. When every rework/reassign cap is exhausted the task
ends as ``controlled_gap`` with a ``gap_json`` risk record that the final
report must carry — it is never disguised as an accepted delivery.
``task_key`` is stable per (run, original assignment); reassignment only
changes ``agent_id`` and never creates a new task.
"""
__tablename__ = "maf_roundtable_seat_tasks"
__table_args__ = (
UniqueConstraint("run_id", "task_key", name="uq_maf_seat_tasks_run_task"),
CheckConstraint(
"status IN ('assigned','delivering','submitted','manager_review',"
"'accepted','rework','reassigned','exhausted','controlled_gap')",
name="ck_maf_seat_tasks_status",
),
)
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
)
task_key: Mapped[str] = mapped_column(String(128), nullable=False)
# Current owning seat (changes on reassignment).
agent_id: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
original_agent_id: Mapped[str] = mapped_column(String(64), nullable=False)
status: Mapped[str] = mapped_column(String(32), default="assigned", index=True)
required: Mapped[bool] = mapped_column(Boolean, default=True)
attempt_count: Mapped[int] = mapped_column(Integer, default=0)
rework_count: Mapped[int] = mapped_column(Integer, default=0)
reassign_count: Mapped[int] = mapped_column(Integer, default=0)
contract_json: Mapped[dict | None] = mapped_column(PortableJSON(), nullable=True)
delivery_json: Mapped[dict | None] = mapped_column(PortableJSON(), nullable=True)
deterministic_check_json: Mapped[dict | None] = mapped_column(PortableJSON(), nullable=True)
review_decision_json: Mapped[dict | None] = mapped_column(PortableJSON(), nullable=True)
history_json: Mapped[list | None] = mapped_column(PortableJSON(), nullable=True)
gap_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
)
# ── 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",
"MafRoundtableSeatTaskRow",
"MafRoundtableSessionRow",
]