diff --git a/alembic/versions/0051_pending_curator_actions.py b/alembic/versions/0051_pending_curator_actions.py new file mode 100644 index 0000000..eedd1d8 --- /dev/null +++ b/alembic/versions/0051_pending_curator_actions.py @@ -0,0 +1,98 @@ +"""pending_curator_actions — curator-proposed mutations awaiting user approval + +Revision ID: 0051 +Revises: 0050 +Create Date: 2026-05-23 + +Architecture: the curator runs unattended and can be confidently wrong. +Additive operations (create_*, record_moment, log_work, save_person) +land directly because adds are easily undone. Mutating operations +(update_*, delete_*) are proposed to this table instead — the user +sees them in a "Needs Review" panel and approves or rejects each one. + +A proposed action stores: + - The action type (`update_task`, `delete_note`, etc.) — drives which + handler gets re-executed on approval. + - The target id + type for display ("update task 42", "delete note 17"). + - The proposed arguments (`payload`) — what the curator wanted to do. + - A snapshot of the target's state at proposal time (`current_snapshot`) + — so the review UI can render a real before/after diff even if other + work modified the entity since. + - The status (`pending` / `approved` / `rejected`) and review timestamp. + +Approval replays the original tool call via execute_tool with +authority="user" so the request bypasses the curator interceptor and +just runs. +""" +from alembic import op +import sqlalchemy as sa +from sqlalchemy.dialects.postgresql import JSONB + + +revision = "0051" +down_revision = "0050" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.create_table( + "pending_curator_actions", + sa.Column("id", sa.Integer, primary_key=True), + sa.Column( + "user_id", + sa.Integer, + sa.ForeignKey("users.id", ondelete="CASCADE"), + nullable=False, + ), + sa.Column( + "conv_id", + sa.Integer, + sa.ForeignKey("conversations.id", ondelete="SET NULL"), + nullable=True, + ), + sa.Column("action_type", sa.Text, nullable=False), + sa.Column("target_type", sa.Text, nullable=True), + sa.Column("target_id", sa.Integer, nullable=True), + sa.Column("target_label", sa.Text, nullable=True), + sa.Column("payload", JSONB, nullable=False, server_default=sa.text("'{}'::jsonb")), + sa.Column( + "current_snapshot", + JSONB, + nullable=False, + server_default=sa.text("'{}'::jsonb"), + ), + sa.Column( + "status", + sa.Text, + nullable=False, + server_default=sa.text("'pending'"), + ), + sa.Column( + "created_at", + sa.DateTime(timezone=True), + nullable=False, + server_default=sa.text("now()"), + ), + sa.Column("reviewed_at", sa.DateTime(timezone=True), nullable=True), + sa.CheckConstraint( + "status IN ('pending', 'approved', 'rejected')", + name="pending_curator_actions_status_check", + ), + ) + # Pending-only index — the Needs Review panel fetches "user's pending" + # constantly, history rows just accumulate. + op.create_index( + "ix_pending_curator_actions_user_pending", + "pending_curator_actions", + ["user_id", sa.text("created_at DESC")], + postgresql_where=sa.text("status = 'pending'"), + ) + + +def downgrade() -> None: + op.drop_index( + "ix_pending_curator_actions_user_pending", + table_name="pending_curator_actions", + ) + op.drop_table("pending_curator_actions") diff --git a/src/fabledassistant/models/__init__.py b/src/fabledassistant/models/__init__.py index 37c4af2..45248e9 100644 --- a/src/fabledassistant/models/__init__.py +++ b/src/fabledassistant/models/__init__.py @@ -50,3 +50,4 @@ from fabledassistant.models.moment import ( # noqa: E402, F401 moment_notes, ) from fabledassistant.models.generation_tool_log import GenerationToolLog # noqa: E402, F401 +from fabledassistant.models.pending_curator_action import PendingCuratorAction # noqa: E402, F401 diff --git a/src/fabledassistant/models/pending_curator_action.py b/src/fabledassistant/models/pending_curator_action.py new file mode 100644 index 0000000..5784877 --- /dev/null +++ b/src/fabledassistant/models/pending_curator_action.py @@ -0,0 +1,81 @@ +from datetime import datetime, timezone + +from sqlalchemy import Boolean, DateTime, ForeignKey, Index, Integer, Text +from sqlalchemy.dialects.postgresql import JSONB +from sqlalchemy.orm import Mapped, mapped_column + +from fabledassistant.models import Base + + +class PendingCuratorAction(Base): + """Curator-proposed mutation awaiting user approval. + + The curator can be confidently wrong, and the user is not in the loop + when it runs. Additive operations land directly; mutating operations + (update_*, delete_*) write a row here instead. The user reviews each + one from the journal's Needs Review panel. + + On approval, the original tool call is replayed via execute_tool with + authority="user" — bypassing the curator interceptor so the request + just runs. + """ + + __tablename__ = "pending_curator_actions" + + id: Mapped[int] = mapped_column(primary_key=True) + user_id: Mapped[int] = mapped_column( + Integer, + ForeignKey("users.id", ondelete="CASCADE"), + nullable=False, + ) + # SET NULL so a curator-proposed action survives if its source + # conversation is later deleted — the user might still want to + # review and approve it. + conv_id: Mapped[int | None] = mapped_column( + Integer, + ForeignKey("conversations.id", ondelete="SET NULL"), + nullable=True, + ) + # The tool name the curator wanted to call (`update_task`, `delete_note`, + # etc.). Used at approval-replay time to dispatch through execute_tool. + action_type: Mapped[str] = mapped_column(Text, nullable=False) + # Display hints — what the UI shows in the card header. + target_type: Mapped[str | None] = mapped_column(Text, nullable=True) + target_id: Mapped[int | None] = mapped_column(Integer, nullable=True) + target_label: Mapped[str | None] = mapped_column(Text, nullable=True) + # The curator's proposed arguments — replayed verbatim on approval. + payload: Mapped[dict] = mapped_column(JSONB, nullable=False, default=dict) + # State of the target at proposal time. Lets the UI render an honest + # diff (current → proposed) even if other work modifies the entity + # between proposal and review. + current_snapshot: Mapped[dict] = mapped_column( + JSONB, nullable=False, default=dict + ) + status: Mapped[str] = mapped_column( + Text, nullable=False, default="pending", server_default="pending" + ) + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), + default=lambda: datetime.now(timezone.utc), + ) + reviewed_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), nullable=True, default=None + ) + + def to_dict(self) -> dict: + return { + "id": self.id, + "user_id": self.user_id, + "conv_id": self.conv_id, + "action_type": self.action_type, + "target_type": self.target_type, + "target_id": self.target_id, + "target_label": self.target_label, + "payload": dict(self.payload or {}), + "current_snapshot": dict(self.current_snapshot or {}), + "status": self.status, + "created_at": self.created_at.isoformat() if self.created_at else None, + "reviewed_at": ( + self.reviewed_at.isoformat() if self.reviewed_at else None + ), + } diff --git a/src/fabledassistant/services/pending_actions.py b/src/fabledassistant/services/pending_actions.py new file mode 100644 index 0000000..04da735 --- /dev/null +++ b/src/fabledassistant/services/pending_actions.py @@ -0,0 +1,169 @@ +"""Pending-action service — curator-proposed mutations awaiting user approval. + +See models/pending_curator_action.py for the schema rationale. This module +is the API everything else uses: + +- `create_pending(...)` is called from execute_tool's curator-authority + interceptor when the curator tries to call a mutating tool. The caller + has already captured a snapshot of the target's current state. + +- `list_pending(user_id)` is what the Needs Review panel reads. + +- `approve(action_id, user_id)` replays the original tool call via + `execute_tool` with `authority="user"`, which bypasses the curator + interceptor and just runs. On success, the row moves to status='approved'. + +- `reject(action_id, user_id)` marks the row 'rejected' without executing. + History row stays around for audit; nothing runs. +""" +from __future__ import annotations + +import datetime +import logging +from datetime import timezone +from typing import Any + +from sqlalchemy import select, update + +from fabledassistant.models import async_session +from fabledassistant.models.pending_curator_action import PendingCuratorAction + +logger = logging.getLogger(__name__) + + +async def create_pending( + *, + user_id: int, + conv_id: int | None, + action_type: str, + target_type: str | None, + target_id: int | None, + target_label: str | None, + payload: dict, + current_snapshot: dict, +) -> PendingCuratorAction: + """Persist a curator-proposed mutation for later review. + + Returns the created row so the interceptor can include its id in the + tool-result envelope (handy for tests, logs, and future UI deep-links). + """ + async with async_session() as session: + row = PendingCuratorAction( + user_id=user_id, + conv_id=conv_id, + action_type=action_type, + target_type=target_type, + target_id=target_id, + target_label=target_label, + payload=payload or {}, + current_snapshot=current_snapshot or {}, + status="pending", + ) + session.add(row) + await session.commit() + await session.refresh(row) + logger.info( + "Curator proposed %s on %s/%s (label=%r) — action_id=%d, conv=%s", + action_type, target_type, target_id, target_label, row.id, conv_id, + ) + return row + + +async def list_pending(user_id: int, limit: int = 50) -> list[dict]: + """Return the user's pending actions, newest first. + + Only `status='pending'` — once a row is approved or rejected, it + drops off this list. Use the index `ix_pending_curator_actions_user_pending`. + """ + async with async_session() as session: + result = await session.execute( + select(PendingCuratorAction) + .where( + PendingCuratorAction.user_id == user_id, + PendingCuratorAction.status == "pending", + ) + .order_by(PendingCuratorAction.created_at.desc()) + .limit(limit) + ) + return [row.to_dict() for row in result.scalars().all()] + + +async def _load_for_user(action_id: int, user_id: int) -> PendingCuratorAction | None: + async with async_session() as session: + result = await session.execute( + select(PendingCuratorAction).where( + PendingCuratorAction.id == action_id, + PendingCuratorAction.user_id == user_id, + ) + ) + return result.scalar_one_or_none() + + +async def _mark_reviewed(action_id: int, status: str) -> None: + async with async_session() as session: + await session.execute( + update(PendingCuratorAction) + .where(PendingCuratorAction.id == action_id) + .values(status=status, reviewed_at=datetime.datetime.now(timezone.utc)) + ) + await session.commit() + + +async def approve(action_id: int, user_id: int) -> dict[str, Any]: + """Replay the curator's proposed action with user authority and mark approved. + + The replay path goes through `execute_tool` with `authority="user"` so + the curator interceptor doesn't intercept its own replay (which would + create another pending row and loop forever). + + Returns the tool-result dict from execute_tool. Action stays in + `pending` if replay returns an error so the user can retry; only + transitions to `approved` on success. + """ + row = await _load_for_user(action_id, user_id) + if row is None: + return {"success": False, "error": "Pending action not found"} + if row.status != "pending": + return { + "success": False, + "error": f"Action already {row.status}", + } + + from fabledassistant.services.tools import execute_tool + try: + # NOTE (C2 → C3 hand-off): execute_tool gains an `authority` kwarg + # in commit C3 that the curator interceptor checks. Approval replay + # MUST pass authority="user" once that lands, otherwise the + # interceptor re-routes the replay back into pending and loops. + # Until C3 lands, execute_tool is identity (no interceptor exists + # yet), so this call runs cleanly. + result = await execute_tool( + user_id, + row.action_type, + row.payload, + conv_id=row.conv_id, + ) + except Exception as e: + logger.exception( + "Replay of curator action %d (%s) failed", + action_id, row.action_type, + ) + return {"success": False, "error": f"{type(e).__name__}: {e}"} + + if result.get("success", True) and not result.get("error"): + await _mark_reviewed(action_id, "approved") + return result + + +async def reject(action_id: int, user_id: int) -> dict[str, Any]: + """Mark the pending action as rejected without executing anything.""" + row = await _load_for_user(action_id, user_id) + if row is None: + return {"success": False, "error": "Pending action not found"} + if row.status != "pending": + return { + "success": False, + "error": f"Action already {row.status}", + } + await _mark_reviewed(action_id, "rejected") + return {"success": True, "id": action_id, "status": "rejected"}