feat(curator): pending_curator_actions schema + service (C2/5)

The backend foundation for curator-proposed mutations awaiting user
approval. No tools route to this yet — that's C3's job. This commit
just lands the schema and the service API everything else will use.

Migration 0051 — new table:
- id, user_id (CASCADE), conv_id (SET NULL — survives conv deletion).
- action_type (the tool name to replay), target_type/target_id/
  target_label (display hints).
- payload (jsonb — the curator's proposed args, replayed verbatim
  on approval).
- current_snapshot (jsonb — the target's state at proposal time, so
  the review UI can render an honest diff even if other work modified
  the entity between proposal and review).
- status ('pending' / 'approved' / 'rejected') + CHECK constraint.
- created_at / reviewed_at.
- Partial index ix_pending_curator_actions_user_pending narrowed to
  status='pending' — the Needs Review panel hits this constantly,
  history rows just accumulate.

Model: PendingCuratorAction with to_dict() for API serialization.

Service services/pending_actions.py:
- create_pending(...) — called from the curator interceptor (C3).
  Accepts an already-fetched current_snapshot so each mutating tool
  can capture target state in its own way (notes vs milestones vs
  profile have different shapes).
- list_pending(user_id, limit=50) — what the Needs Review panel reads.
- approve(action_id, user_id) — replays via execute_tool and marks
  approved on success. Stays pending on replay error so the user
  can retry. NOTE: approve passes the request through execute_tool
  unchanged for now; C3 will add authority='user' so the upcoming
  curator interceptor doesn't re-intercept the replay and loop.
- reject(action_id, user_id) — marks rejected with no execution.

C3 next: wires the curator interceptor (authority='curator' on
execute_tool routes mutating tools to create_pending instead of
running them), adds the mutating tools back to the curator's
allowlist, and updates approve() to pass authority='user'.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

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")

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

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
+            ),
+        }

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"}