feat(moderation): add batch review system with mobile-friendly UI (#672)

* feat(moderation): add batch review system with mobile-friendly UI

- Add review batch tables (review_batches, batch_flags) with migrations
- Add /admin/batches POST endpoint to create review batches
- Add /review/:id endpoints for auth-protected review UI
- Review page renders server-side HTML with embedded JS
- Same auth middleware as admin endpoints (X-Moderation-Key header)
- Update moderation_loop.py to create batches and send DM links
- Simplify loop: DM is now just a notification channel, not for parsing

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <[e-mail]>

* feat(ci): add workflow dispatch for moderation loop

Allows triggering the moderation loop from GitHub Actions UI with:
- dry_run toggle (default: true for safety)
- limit input for testing with subset of flags
- env selector (prod/staging/dev)

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <[e-mail]>

---------

Co-authored-by: Claude Opus 4.5 <[e-mail]>

.github/workflows/run-moderation-loop.yml

···
+# run moderation loop via workflow dispatch
+#
+# analyzes pending copyright flags, auto-resolves false positives,
+# creates review batches for human review, sends DM notification
+#
+# required secrets:
+#   MODERATION_SERVICE_URL - moderation service base URL
+#   MODERATION_AUTH_TOKEN - X-Moderation-Key header value
+#   ANTHROPIC_API_KEY - for flag analysis
+#   NOTIFY_BOT_HANDLE - bluesky bot handle for DMs
+#   NOTIFY_BOT_PASSWORD - bluesky bot app password
+#   NOTIFY_RECIPIENT_HANDLE - who receives DM notifications
+
+name: run moderation loop
+
+on:
+  workflow_dispatch:
+    inputs:
+      dry_run:
+        description: "dry run (analyze only, don't resolve or send DMs)"
+        type: boolean
+        default: true
+      limit:
+        description: "max flags to process (leave empty for all)"
+        type: string
+        default: ""
+      env:
+        description: "environment (for DM header)"
+        type: choice
+        options:
+          - prod
+          - staging
+          - dev
+        default: prod
+
+jobs:
+  run:
+    runs-on: ubuntu-latest
+
+    steps:
+      - uses: actions/checkout@v4
+
+      - uses: astral-sh/setup-uv@v4
+
+      - name: Run moderation loop
+        run: |
+          ARGS=""
+          if [ "${{ inputs.dry_run }}" = "true" ]; then
+            ARGS="$ARGS --dry-run"
+          fi
+          if [ -n "${{ inputs.limit }}" ]; then
+            ARGS="$ARGS --limit ${{ inputs.limit }}"
+          fi
+          ARGS="$ARGS --env ${{ inputs.env }}"
+
+          echo "Running: uv run scripts/moderation_loop.py $ARGS"
+          uv run scripts/moderation_loop.py $ARGS
+        env:
+          MODERATION_SERVICE_URL: ${{ secrets.MODERATION_SERVICE_URL }}
+          MODERATION_AUTH_TOKEN: ${{ secrets.MODERATION_AUTH_TOKEN }}
+          ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }}
+          NOTIFY_BOT_HANDLE: ${{ secrets.NOTIFY_BOT_HANDLE }}
+          NOTIFY_BOT_PASSWORD: ${{ secrets.NOTIFY_BOT_PASSWORD }}
+          NOTIFY_RECIPIENT_HANDLE: ${{ secrets.NOTIFY_RECIPIENT_HANDLE }}

moderation/Cargo.toml

···
 [dependencies]
 anyhow = "1.0"
 axum = { version = "0.7", features = ["macros", "json", "ws"] }
+rand = "0.8"
 bytes = "1.0"
 chrono = { version = "0.4", features = ["serde"] }
 futures = "0.3"
···
 tracing-subscriber = { version = "0.3", features = ["env-filter", "fmt"] }
 
 [dev-dependencies]
-rand = "0.8"

moderation/src/admin.rs

···
     pub message: String,
 }
 
+/// Request to create a review batch.
+#[derive(Debug, Deserialize)]
+pub struct CreateBatchRequest {
+    /// URIs to include. If empty, uses all pending flags.
+    #[serde(default)]
+    pub uris: Vec<String>,
+    /// Who created this batch.
+    pub created_by: Option<String>,
+}
+
+/// Response after creating a review batch.
+#[derive(Debug, Serialize)]
+pub struct CreateBatchResponse {
+    pub id: String,
+    pub url: String,
+    pub flag_count: usize,
+}
+
 /// List all flagged tracks - returns JSON for API, HTML for htmx.
 pub async fn list_flagged(
     State(state): State<AppState>,
···
     Ok(Json(StoreContextResponse {
         message: format!("context stored for {}", request.uri),
     }))
+}
+
+/// Create a review batch from pending flags.
+pub async fn create_batch(
+    State(state): State<AppState>,
+    Json(request): Json<CreateBatchRequest>,
+) -> Result<Json<CreateBatchResponse>, AppError> {
+    let db = state.db.as_ref().ok_or(AppError::LabelerNotConfigured)?;
+
+    // Get URIs to include
+    let uris = if request.uris.is_empty() {
+        let pending = db.get_pending_flags().await?;
+        pending
+            .into_iter()
+            .filter(|t| !t.resolved)
+            .map(|t| t.uri)
+            .collect()
+    } else {
+        request.uris
+    };
+
+    if uris.is_empty() {
+        return Err(AppError::BadRequest("no flags to review".to_string()));
+    }
+
+    let id = generate_batch_id();
+    let flag_count = uris.len();
+
+    tracing::info!(
+        batch_id = %id,
+        flag_count = flag_count,
+        "creating review batch"
+    );
+
+    db.create_batch(&id, &uris, request.created_by.as_deref())
+        .await?;
+
+    let url = format!("/review/{}", id);
+
+    Ok(Json(CreateBatchResponse { id, url, flag_count }))
+}
+
+/// Generate a short, URL-safe batch ID.
+fn generate_batch_id() -> String {
+    use std::time::{SystemTime, UNIX_EPOCH};
+    let now = SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .unwrap()
+        .as_millis();
+    let rand_part: u32 = rand::random();
+    format!("{:x}{:x}", (now as u64) & 0xFFFFFFFF, rand_part & 0xFFFF)
 }
 
 /// Add a sensitive image entry.

moderation/src/db.rs

···
     pub flagged_by: Option<String>,
 }
 
+/// Review batch for mobile-friendly flag review.
+#[derive(Debug, Clone, FromRow, Serialize, Deserialize)]
+pub struct ReviewBatch {
+    pub id: String,
+    pub created_at: DateTime<Utc>,
+    pub expires_at: Option<DateTime<Utc>>,
+    /// Status: pending, completed.
+    pub status: String,
+    /// Who created this batch.
+    pub created_by: Option<String>,
+}
+
+/// A flag within a review batch.
+#[derive(Debug, Clone, FromRow, Serialize, Deserialize)]
+pub struct BatchFlag {
+    pub id: i64,
+    pub batch_id: String,
+    pub uri: String,
+    pub reviewed: bool,
+    pub reviewed_at: Option<DateTime<Utc>>,
+    /// Decision: approved, rejected, or null.
+    pub decision: Option<String>,
+}
+
 /// Type alias for context row from database query.
 type ContextRow = (
     Option<i64>,    // track_id
···
             .execute(&self.pool)
             .await?;
         sqlx::query("CREATE INDEX IF NOT EXISTS idx_sensitive_images_url ON sensitive_images(url)")
+            .execute(&self.pool)
+            .await?;
+
+        // Review batches for mobile-friendly flag review
+        sqlx::query(
+            r#"
+            CREATE TABLE IF NOT EXISTS review_batches (
+                id TEXT PRIMARY KEY,
+                created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+                expires_at TIMESTAMPTZ,
+                status TEXT NOT NULL DEFAULT 'pending',
+                created_by TEXT
+            )
+            "#,
+        )
+        .execute(&self.pool)
+        .await?;
+
+        // Flags within review batches
+        sqlx::query(
+            r#"
+            CREATE TABLE IF NOT EXISTS batch_flags (
+                id BIGSERIAL PRIMARY KEY,
+                batch_id TEXT NOT NULL REFERENCES review_batches(id) ON DELETE CASCADE,
+                uri TEXT NOT NULL,
+                reviewed BOOLEAN NOT NULL DEFAULT FALSE,
+                reviewed_at TIMESTAMPTZ,
+                decision TEXT,
+                UNIQUE(batch_id, uri)
+            )
+            "#,
+        )
+        .execute(&self.pool)
+        .await?;
+
+        sqlx::query("CREATE INDEX IF NOT EXISTS idx_batch_flags_batch_id ON batch_flags(batch_id)")
             .execute(&self.pool)
             .await?;
 
···
             .collect();
 
         Ok(tracks)
+    }
+
+    // -------------------------------------------------------------------------
+    // Review batches
+    // -------------------------------------------------------------------------
+
+    /// Create a review batch with the given flags.
+    pub async fn create_batch(
+        &self,
+        id: &str,
+        uris: &[String],
+        created_by: Option<&str>,
+    ) -> Result<ReviewBatch, sqlx::Error> {
+        let batch = sqlx::query_as::<_, ReviewBatch>(
+            r#"
+            INSERT INTO review_batches (id, created_by)
+            VALUES ($1, $2)
+            RETURNING id, created_at, expires_at, status, created_by
+            "#,
+        )
+        .bind(id)
+        .bind(created_by)
+        .fetch_one(&self.pool)
+        .await?;
+
+        for uri in uris {
+            sqlx::query(
+                r#"
+                INSERT INTO batch_flags (batch_id, uri)
+                VALUES ($1, $2)
+                ON CONFLICT (batch_id, uri) DO NOTHING
+                "#,
+            )
+            .bind(id)
+            .bind(uri)
+            .execute(&self.pool)
+            .await?;
+        }
+
+        Ok(batch)
+    }
+
+    /// Get a batch by ID.
+    pub async fn get_batch(&self, id: &str) -> Result<Option<ReviewBatch>, sqlx::Error> {
+        sqlx::query_as::<_, ReviewBatch>(
+            r#"
+            SELECT id, created_at, expires_at, status, created_by
+            FROM review_batches
+            WHERE id = $1
+            "#,
+        )
+        .bind(id)
+        .fetch_optional(&self.pool)
+        .await
+    }
+
+    /// Get all flags in a batch with their context.
+    pub async fn get_batch_flags(&self, batch_id: &str) -> Result<Vec<FlaggedTrack>, sqlx::Error> {
+        let rows: Vec<FlaggedRow> = sqlx::query_as(
+            r#"
+            SELECT l.seq, l.uri, l.val, l.cts,
+                   c.track_id, c.track_title, c.artist_handle, c.artist_did, c.highest_score, c.matches,
+                   c.resolution_reason, c.resolution_notes
+            FROM batch_flags bf
+            JOIN labels l ON l.uri = bf.uri AND l.val = 'copyright-violation' AND l.neg = false
+            LEFT JOIN label_context c ON l.uri = c.uri
+            WHERE bf.batch_id = $1
+            ORDER BY l.seq DESC
+            "#,
+        )
+        .bind(batch_id)
+        .fetch_all(&self.pool)
+        .await?;
+
+        let batch_uris: Vec<String> = rows.iter().map(|r| r.1.clone()).collect();
+        let negated_uris: std::collections::HashSet<String> = if !batch_uris.is_empty() {
+            sqlx::query_scalar::<_, String>(
+                r#"
+                SELECT DISTINCT uri
+                FROM labels
+                WHERE val = 'copyright-violation' AND neg = true AND uri = ANY($1)
+                "#,
+            )
+            .bind(&batch_uris)
+            .fetch_all(&self.pool)
+            .await?
+            .into_iter()
+            .collect()
+        } else {
+            std::collections::HashSet::new()
+        };
+
+        let tracks = rows
+            .into_iter()
+            .map(
+                |(
+                    seq,
+                    uri,
+                    val,
+                    cts,
+                    track_id,
+                    track_title,
+                    artist_handle,
+                    artist_did,
+                    highest_score,
+                    matches,
+                    resolution_reason,
+                    resolution_notes,
+                )| {
+                    let context = if track_id.is_some()
+                        || track_title.is_some()
+                        || artist_handle.is_some()
+                        || resolution_reason.is_some()
+                    {
+                        Some(LabelContext {
+                            track_id,
+                            track_title,
+                            artist_handle,
+                            artist_did,
+                            highest_score,
+                            matches: matches.and_then(|v| serde_json::from_value(v).ok()),
+                            resolution_reason: resolution_reason
+                                .and_then(|s| ResolutionReason::from_str(&s)),
+                            resolution_notes,
+                        })
+                    } else {
+                        None
+                    };
+
+                    FlaggedTrack {
+                        seq,
+                        uri: uri.clone(),
+                        val,
+                        created_at: cts.format("%Y-%m-%d %H:%M:%S").to_string(),
+                        resolved: negated_uris.contains(&uri),
+                        context,
+                    }
+                },
+            )
+            .collect();
+
+        Ok(tracks)
+    }
+
+    /// Update batch status.
+    pub async fn update_batch_status(&self, id: &str, status: &str) -> Result<bool, sqlx::Error> {
+        let result = sqlx::query("UPDATE review_batches SET status = $1 WHERE id = $2")
+            .bind(status)
+            .bind(id)
+            .execute(&self.pool)
+            .await?;
+        Ok(result.rows_affected() > 0)
+    }
+
+    /// Mark a flag in a batch as reviewed.
+    pub async fn mark_flag_reviewed(
+        &self,
+        batch_id: &str,
+        uri: &str,
+        decision: &str,
+    ) -> Result<bool, sqlx::Error> {
+        let result = sqlx::query(
+            r#"
+            UPDATE batch_flags
+            SET reviewed = true, reviewed_at = NOW(), decision = $1
+            WHERE batch_id = $2 AND uri = $3
+            "#,
+        )
+        .bind(decision)
+        .bind(batch_id)
+        .bind(uri)
+        .execute(&self.pool)
+        .await?;
+        Ok(result.rows_affected() > 0)
+    }
+
+    /// Get pending (non-reviewed) flags from a batch.
+    pub async fn get_batch_pending_uris(&self, batch_id: &str) -> Result<Vec<String>, sqlx::Error> {
+        sqlx::query_scalar::<_, String>(
+            r#"
+            SELECT uri FROM batch_flags
+            WHERE batch_id = $1 AND reviewed = false
+            "#,
+        )
+        .bind(batch_id)
+        .fetch_all(&self.pool)
+        .await
     }
 
     // -------------------------------------------------------------------------

moderation/src/main.rs

···
 mod db;
 mod handlers;
 mod labels;
+mod review;
 mod state;
 mod xrpc;
 
···
             "/admin/sensitive-images/remove",
             post(admin::remove_sensitive_image),
         )
+        .route("/admin/batches", post(admin::create_batch))
+        // Review endpoints (auth protected)
+        .route("/review/:id", get(review::review_page))
+        .route("/review/:id/data", get(review::review_data))
+        .route("/review/:id/submit", post(review::submit_review))
         // Static files (CSS, JS for admin UI)
         .nest_service("/static", ServeDir::new("static"))
         // ATProto XRPC endpoints (public)

moderation/src/review.rs

···
+//! Review endpoints for batch flag review.
+//!
+//! These endpoints are behind the same auth as admin endpoints.
+
+use axum::{
+    extract::{Path, State},
+    http::header::CONTENT_TYPE,
+    response::{IntoResponse, Response},
+    Json,
+};
+use serde::{Deserialize, Serialize};
+
+use crate::admin::FlaggedTrack;
+use crate::state::{AppError, AppState};
+
+/// Response for review page data.
+#[derive(Debug, Serialize)]
+pub struct ReviewPageData {
+    pub batch_id: String,
+    pub flags: Vec<FlaggedTrack>,
+    pub status: String,
+}
+
+/// Request to submit review decisions.
+#[derive(Debug, Deserialize)]
+pub struct SubmitReviewRequest {
+    pub decisions: Vec<ReviewDecision>,
+}
+
+/// A single review decision.
+#[derive(Debug, Deserialize)]
+pub struct ReviewDecision {
+    pub uri: String,
+    pub decision: String, // "approved" or "rejected"
+}
+
+/// Response after submitting review.
+#[derive(Debug, Serialize)]
+pub struct SubmitReviewResponse {
+    pub resolved_count: usize,
+    pub message: String,
+}
+
+/// Get review page HTML.
+pub async fn review_page(
+    State(state): State<AppState>,
+    Path(batch_id): Path<String>,
+) -> Result<Response, AppError> {
+    let db = state.db.as_ref().ok_or(AppError::LabelerNotConfigured)?;
+
+    let batch = db
+        .get_batch(&batch_id)
+        .await?
+        .ok_or(AppError::NotFound("batch not found".to_string()))?;
+
+    let flags = db.get_batch_flags(&batch_id).await?;
+    let html = render_review_page(&batch_id, &flags, &batch.status);
+
+    Ok(([(CONTENT_TYPE, "text/html; charset=utf-8")], html).into_response())
+}
+
+/// Get review data as JSON.
+pub async fn review_data(
+    State(state): State<AppState>,
+    Path(batch_id): Path<String>,
+) -> Result<Json<ReviewPageData>, AppError> {
+    let db = state.db.as_ref().ok_or(AppError::LabelerNotConfigured)?;
+
+    let batch = db
+        .get_batch(&batch_id)
+        .await?
+        .ok_or(AppError::NotFound("batch not found".to_string()))?;
+
+    let flags = db.get_batch_flags(&batch_id).await?;
+
+    Ok(Json(ReviewPageData {
+        batch_id,
+        flags,
+        status: batch.status,
+    }))
+}
+
+/// Submit review decisions.
+pub async fn submit_review(
+    State(state): State<AppState>,
+    Path(batch_id): Path<String>,
+    Json(request): Json<SubmitReviewRequest>,
+) -> Result<Json<SubmitReviewResponse>, AppError> {
+    let db = state.db.as_ref().ok_or(AppError::LabelerNotConfigured)?;
+    let signer = state
+        .signer
+        .as_ref()
+        .ok_or(AppError::LabelerNotConfigured)?;
+
+    let _batch = db
+        .get_batch(&batch_id)
+        .await?
+        .ok_or(AppError::NotFound("batch not found".to_string()))?;
+
+    let mut resolved_count = 0;
+
+    for decision in &request.decisions {
+        tracing::info!(
+            batch_id = %batch_id,
+            uri = %decision.uri,
+            decision = %decision.decision,
+            "processing review decision"
+        );
+
+        db.mark_flag_reviewed(&batch_id, &decision.uri, &decision.decision)
+            .await?;
+
+        if decision.decision == "approved" {
+            let label =
+                crate::labels::Label::new(signer.did(), &decision.uri, "copyright-violation")
+                    .negated();
+            let label = signer.sign_label(label)?;
+            let seq = db.store_label(&label).await?;
+
+            db.store_resolution(
+                &decision.uri,
+                crate::db::ResolutionReason::FingerprintNoise,
+                Some("batch review"),
+            )
+            .await?;
+
+            if let Some(tx) = &state.label_tx {
+                let _ = tx.send((seq, label));
+            }
+
+            resolved_count += 1;
+        }
+    }
+
+    let pending = db.get_batch_pending_uris(&batch_id).await?;
+    if pending.is_empty() {
+        db.update_batch_status(&batch_id, "completed").await?;
+    }
+
+    Ok(Json(SubmitReviewResponse {
+        resolved_count,
+        message: format!(
+            "processed {} decisions, resolved {} flags",
+            request.decisions.len(),
+            resolved_count
+        ),
+    }))
+}
+
+/// Render the review page.
+fn render_review_page(batch_id: &str, flags: &[FlaggedTrack], status: &str) -> String {
+    let pending: Vec<_> = flags.iter().filter(|f| !f.resolved).collect();
+    let resolved: Vec<_> = flags.iter().filter(|f| f.resolved).collect();
+
+    let pending_cards: Vec<String> = pending.iter().map(|f| render_review_card(f)).collect();
+    let resolved_cards: Vec<String> = resolved.iter().map(|f| render_review_card(f)).collect();
+
+    let pending_html = if pending_cards.is_empty() {
+        "<div class=\"empty\">all flags reviewed!</div>".to_string()
+    } else {
+        pending_cards.join("\n")
+    };
+
+    let resolved_html = if resolved_cards.is_empty() {
+        String::new()
+    } else {
+        format!(
+            r#"<details class="resolved-section">
+                <summary>{} resolved</summary>
+                {}
+            </details>"#,
+            resolved_cards.len(),
+            resolved_cards.join("\n")
+        )
+    };
+
+    let status_badge = if status == "completed" {
+        r#"<span class="status-badge completed">completed</span>"#
+    } else {
+        ""
+    };
+
+    format!(
+        r#"<!DOCTYPE html>
+<html lang="en">
+<head>
+    <meta charset="utf-8">
+    <meta name="viewport" content="width=device-width, initial-scale=1">
+    <title>review batch - plyr.fm</title>
+    <style>{}</style>
+</head>
+<body>
+    <div class="container">
+        <header>
+            <h1>plyr.fm moderation</h1>
+            <div class="batch-info">{} pending {}</div>
+        </header>
+
+        <form id="review-form" class="review-form">
+            <div class="flags-list">
+                {}
+            </div>
+
+            {}
+
+            <div class="submit-bar">
+                <button type="submit" class="btn-submit" id="submit-btn" disabled>
+                    submit decisions
+                </button>
+            </div>
+        </form>
+    </div>
+
+    <script>
+        const form = document.getElementById('review-form');
+        const submitBtn = document.getElementById('submit-btn');
+        const batchId = '{}';
+
+        const decisions = {{}};
+
+        function updateSubmitBtn() {{
+            const count = Object.keys(decisions).length;
+            submitBtn.disabled = count === 0;
+            submitBtn.textContent = count > 0 ? `submit ${{count}} decision${{count > 1 ? 's' : ''}}` : 'submit decisions';
+        }}
+
+        function setDecision(uri, decision) {{
+            decisions[uri] = decision;
+            const card = document.querySelector(`[data-uri="${{CSS.escape(uri)}}"]`);
+            if (card) {{
+                card.classList.remove('approved', 'rejected');
+                card.classList.add(decision);
+            }}
+            updateSubmitBtn();
+        }}
+
+        form.addEventListener('submit', async (e) => {{
+            e.preventDefault();
+            submitBtn.disabled = true;
+            submitBtn.textContent = 'submitting...';
+
+            try {{
+                const response = await fetch(`/review/${{batchId}}/submit`, {{
+                    method: 'POST',
+                    headers: {{ 'Content-Type': 'application/json' }},
+                    body: JSON.stringify({{
+                        decisions: Object.entries(decisions).map(([uri, decision]) => ({{ uri, decision }}))
+                    }})
+                }});
+
+                if (response.ok) {{
+                    const result = await response.json();
+                    alert(result.message);
+                    location.reload();
+                }} else {{
+                    const err = await response.json();
+                    alert('error: ' + (err.message || 'unknown error'));
+                    submitBtn.disabled = false;
+                    updateSubmitBtn();
+                }}
+            }} catch (err) {{
+                alert('network error: ' + err.message);
+                submitBtn.disabled = false;
+                updateSubmitBtn();
+            }}
+        }});
+    </script>
+</body>
+</html>"#,
+        REVIEW_CSS,
+        pending.len(),
+        status_badge,
+        pending_html,
+        resolved_html,
+        html_escape(batch_id)
+    )
+}
+
+/// Render a single review card.
+fn render_review_card(track: &FlaggedTrack) -> String {
+    let ctx = track.context.as_ref();
+
+    let title = ctx
+        .and_then(|c| c.track_title.as_deref())
+        .unwrap_or("unknown track");
+    let artist = ctx
+        .and_then(|c| c.artist_handle.as_deref())
+        .unwrap_or("unknown");
+    let track_id = ctx.and_then(|c| c.track_id);
+
+    let title_html = if let Some(id) = track_id {
+        format!(
+            r#"<a href="https://plyr.fm/track/{}" target="_blank">{}</a>"#,
+            id,
+            html_escape(title)
+        )
+    } else {
+        html_escape(title)
+    };
+
+    let matches_html = ctx
+        .and_then(|c| c.matches.as_ref())
+        .filter(|m| !m.is_empty())
+        .map(|matches| {
+            let items: Vec<String> = matches
+                .iter()
+                .take(2)
+                .map(|m| {
+                    format!(
+                        r#"<span class="match">{} - {}</span>"#,
+                        html_escape(&m.title),
+                        html_escape(&m.artist)
+                    )
+                })
+                .collect();
+            format!(r#"<div class="matches">{}</div>"#, items.join(""))
+        })
+        .unwrap_or_default();
+
+    let resolved_badge = if track.resolved {
+        r#"<span class="badge resolved">resolved</span>"#
+    } else {
+        ""
+    };
+
+    let action_buttons = if !track.resolved {
+        format!(
+            r#"<div class="actions">
+                <button type="button" class="btn-approve" onclick="setDecision('{}', 'approved')">approve</button>
+                <button type="button" class="btn-reject" onclick="setDecision('{}', 'rejected')">reject</button>
+            </div>"#,
+            html_escape(&track.uri),
+            html_escape(&track.uri)
+        )
+    } else {
+        String::new()
+    };
+
+    format!(
+        r#"<div class="review-card{}" data-uri="{}">
+            <div class="track-info">
+                <div class="title">{}</div>
+                <div class="artist">@{}</div>
+                {}
+            </div>
+            {}
+            {}
+        </div>"#,
+        if track.resolved { " resolved" } else { "" },
+        html_escape(&track.uri),
+        title_html,
+        html_escape(artist),
+        matches_html,
+        resolved_badge,
+        action_buttons
+    )
+}
+
+fn html_escape(s: &str) -> String {
+    s.replace('&', "&amp;")
+        .replace('<', "&lt;")
+        .replace('>', "&gt;")
+        .replace('"', "&quot;")
+        .replace('\'', "&#039;")
+}
+
+const REVIEW_CSS: &str = r#"
+* { box-sizing: border-box; margin: 0; padding: 0; }
+
+body {
+    font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, sans-serif;
+    background: #0a0a0a;
+    color: #e0e0e0;
+    min-height: 100vh;
+}
+
+.container {
+    max-width: 600px;
+    margin: 0 auto;
+    padding: 16px;
+    padding-bottom: 80px;
+}
+
+header {
+    display: flex;
+    justify-content: space-between;
+    align-items: center;
+    margin-bottom: 20px;
+    padding-bottom: 12px;
+    border-bottom: 1px solid #333;
+}
+
+h1 { font-size: 1.25rem; font-weight: 600; color: #fff; }
+.batch-info { font-size: 0.875rem; color: #888; }
+.status-badge { font-size: 0.7rem; background: #1a3a1a; color: #6d9; padding: 2px 6px; border-radius: 4px; margin-left: 8px; }
+.flags-list { display: flex; flex-direction: column; gap: 12px; }
+
+.review-card {
+    background: #1a1a1a;
+    border: 1px solid #333;
+    border-radius: 8px;
+    padding: 12px;
+    transition: border-color 0.2s, background 0.2s;
+}
+
+.review-card.approved { border-color: #2d5a27; background: #1a2a18; }
+.review-card.rejected { border-color: #5a2727; background: #2a1818; }
+.review-card.resolved { opacity: 0.6; }
+.track-info { margin-bottom: 8px; }
+.title { font-weight: 600; font-size: 1rem; margin-bottom: 2px; }
+.title a { color: inherit; text-decoration: none; }
+.title a:hover { text-decoration: underline; }
+.artist { font-size: 0.875rem; color: #888; }
+.matches { margin-top: 6px; display: flex; flex-wrap: wrap; gap: 4px; }
+.match { font-size: 0.75rem; background: #2a2a2a; padding: 2px 6px; border-radius: 4px; color: #aaa; }
+.badge { display: inline-block; font-size: 0.7rem; padding: 2px 6px; border-radius: 4px; text-transform: uppercase; font-weight: 500; }
+.badge.resolved { background: #1a3a1a; color: #6d9; }
+.actions { display: flex; gap: 8px; margin-top: 10px; }
+.actions button { flex: 1; padding: 10px; border: none; border-radius: 6px; font-size: 0.875rem; font-weight: 500; cursor: pointer; transition: opacity 0.2s; }
+.btn-approve { background: #2d5a27; color: #fff; }
+.btn-reject { background: #5a2727; color: #fff; }
+.actions button:active { opacity: 0.8; }
+
+.submit-bar {
+    position: fixed;
+    bottom: 0;
+    left: 0;
+    right: 0;
+    padding: 12px 16px;
+    background: #111;
+    border-top: 1px solid #333;
+}
+
+.btn-submit {
+    width: 100%;
+    max-width: 600px;
+    margin: 0 auto;
+    display: block;
+    padding: 14px;
+    background: #4a9eff;
+    color: #fff;
+    border: none;
+    border-radius: 8px;
+    font-size: 1rem;
+    font-weight: 600;
+    cursor: pointer;
+}
+
+.btn-submit:disabled { background: #333; color: #666; cursor: not-allowed; }
+.empty { text-align: center; padding: 40px 20px; color: #666; }
+.resolved-section { margin-top: 20px; border-top: 1px solid #333; padding-top: 16px; }
+.resolved-section summary { cursor: pointer; color: #888; font-size: 0.875rem; margin-bottom: 12px; }
+"#;

moderation/src/state.rs

···
     #[error("bad request: {0}")]
     BadRequest(String),
 
+    #[error("not found: {0}")]
+    NotFound(String),
+
     #[error("label error: {0}")]
     Label(#[from] LabelError),
 
···
                 (StatusCode::SERVICE_UNAVAILABLE, "LabelerNotConfigured")
             }
             AppError::BadRequest(_) => (StatusCode::BAD_REQUEST, "BadRequest"),
+            AppError::NotFound(_) => (StatusCode::NOT_FOUND, "NotFound"),
             AppError::Label(_) => (StatusCode::INTERNAL_SERVER_ERROR, "LabelError"),
             AppError::Database(_) => (StatusCode::INTERNAL_SERVER_ERROR, "DatabaseError"),
             AppError::Io(_) => (StatusCode::INTERNAL_SERVER_ERROR, "IoError"),

scripts/moderation_loop.py

···
+#!/usr/bin/env -S uv run --script --quiet
+# /// script
+# requires-python = ">=3.12"
+# dependencies = [
+#     "pydantic-ai>=0.1.0",
+#     "anthropic",
+#     "httpx",
+#     "pydantic>=2.0",
+#     "pydantic-settings",
+#     "atproto>=0.0.55",
+#     "rich",
+# ]
+# ///
+"""autonomous moderation loop for plyr.fm.
+
+workflow:
+1. fetch pending flags from moderation service
+2. analyze each flag with LLM (FALSE_POSITIVE, VIOLATION, NEEDS_HUMAN)
+3. auto-resolve false positives
+4. create review batch for needs_human flags
+5. send DM with link to review UI
+
+the review UI handles human decisions - DM is just a notification channel.
+"""
+
+import argparse
+import asyncio
+from dataclasses import dataclass, field
+from pathlib import Path
+
+import httpx
+from atproto import AsyncClient, models
+from pydantic import BaseModel, Field
+from pydantic_ai import Agent
+from pydantic_ai.models.anthropic import AnthropicModel
+from pydantic_settings import BaseSettings, SettingsConfigDict
+from rich.console import Console
+
+console = Console()
+
+
+class LoopSettings(BaseSettings):
+    model_config = SettingsConfigDict(
+        env_file=Path(__file__).parent.parent / ".env",
+        case_sensitive=False,
+        extra="ignore",
+    )
+    moderation_service_url: str = Field(
+        default="https://moderation.plyr.fm", validation_alias="MODERATION_SERVICE_URL"
+    )
+    moderation_auth_token: str = Field(
+        default="", validation_alias="MODERATION_AUTH_TOKEN"
+    )
+    anthropic_api_key: str = Field(default="", validation_alias="ANTHROPIC_API_KEY")
+    anthropic_model: str = Field(
+        default="claude-sonnet-4-20250514", validation_alias="ANTHROPIC_MODEL"
+    )
+    bot_handle: str = Field(default="", validation_alias="NOTIFY_BOT_HANDLE")
+    bot_password: str = Field(default="", validation_alias="NOTIFY_BOT_PASSWORD")
+    recipient_handle: str = Field(
+        default="", validation_alias="NOTIFY_RECIPIENT_HANDLE"
+    )
+
+
+class FlagAnalysis(BaseModel):
+    """result of analyzing a single flag."""
+
+    uri: str
+    category: str = Field(description="FALSE_POSITIVE, VIOLATION, or NEEDS_HUMAN")
+    reason: str
+
+
+@dataclass
+class DMClient:
+    handle: str
+    password: str
+    recipient_handle: str
+    _client: AsyncClient = field(init=False, repr=False)
+    _dm_client: AsyncClient = field(init=False, repr=False)
+    _recipient_did: str = field(init=False, repr=False)
+    _convo_id: str = field(init=False, repr=False)
+
+    async def setup(self) -> None:
+        self._client = AsyncClient()
+        await self._client.login(self.handle, self.password)
+        self._dm_client = self._client.with_bsky_chat_proxy()
+        profile = await self._client.app.bsky.actor.get_profile(
+            {"actor": self.recipient_handle}
+        )
+        self._recipient_did = profile.did
+        convo = await self._dm_client.chat.bsky.convo.get_convo_for_members(
+            models.ChatBskyConvoGetConvoForMembers.Params(members=[self._recipient_did])
+        )
+        self._convo_id = convo.convo.id
+
+    async def get_messages(self, limit: int = 30) -> list[dict]:
+        response = await self._dm_client.chat.bsky.convo.get_messages(
+            models.ChatBskyConvoGetMessages.Params(convo_id=self._convo_id, limit=limit)
+        )
+        return [
+            {
+                "text": m.text,
+                "is_bot": m.sender.did != self._recipient_did,
+                "sent_at": getattr(m, "sent_at", None),
+            }
+            for m in response.messages
+            if hasattr(m, "text") and hasattr(m, "sender")
+        ]
+
+    async def send(self, text: str) -> None:
+        await self._dm_client.chat.bsky.convo.send_message(
+            models.ChatBskyConvoSendMessage.Data(
+                convo_id=self._convo_id,
+                message=models.ChatBskyConvoDefs.MessageInput(text=text),
+            )
+        )
+
+
+@dataclass
+class ModClient:
+    base_url: str
+    auth_token: str
+    _client: httpx.AsyncClient = field(init=False, repr=False)
+
+    def __post_init__(self) -> None:
+        self._client = httpx.AsyncClient(
+            base_url=self.base_url,
+            headers={"X-Moderation-Key": self.auth_token},
+            timeout=30.0,
+        )
+
+    async def close(self) -> None:
+        await self._client.aclose()
+
+    async def list_pending(self) -> list[dict]:
+        r = await self._client.get("/admin/flags", params={"filter": "pending"})
+        r.raise_for_status()
+        return r.json().get("tracks", [])
+
+    async def resolve(self, uri: str, reason: str, notes: str = "") -> None:
+        r = await self._client.post(
+            "/admin/resolve",
+            json={
+                "uri": uri,
+                "val": "copyright-violation",
+                "reason": reason,
+                "notes": notes,
+            },
+        )
+        r.raise_for_status()
+
+    async def create_batch(
+        self, uris: list[str], created_by: str | None = None
+    ) -> dict:
+        """create a review batch and return {id, url, flag_count}."""
+        r = await self._client.post(
+            "/admin/batches",
+            json={"uris": uris, "created_by": created_by},
+        )
+        r.raise_for_status()
+        return r.json()
+
+
+def get_header(env: str) -> str:
+    return f"[PLYR-MOD:{env.upper()}]"
+
+
+def create_flag_analyzer(api_key: str, model: str) -> Agent[None, list[FlagAnalysis]]:
+    from pydantic_ai.providers.anthropic import AnthropicProvider
+
+    return Agent(
+        model=AnthropicModel(model, provider=AnthropicProvider(api_key=api_key)),
+        output_type=list[FlagAnalysis],
+        system_prompt="""\
+analyze each copyright flag. categorize as:
+- FALSE_POSITIVE: fingerprint noise, uploader is the artist, unrelated matches
+- VIOLATION: clearly copyrighted commercial content
+- NEEDS_HUMAN: ambiguous, need human review
+
+return a FlagAnalysis for each flag with uri, category, and brief reason.
+""",
+    )
+
+
+async def run_loop(
+    dry_run: bool = False, limit: int | None = None, env: str = "prod"
+) -> None:
+    settings = LoopSettings()
+    for attr in [
+        "moderation_auth_token",
+        "anthropic_api_key",
+        "bot_handle",
+        "bot_password",
+        "recipient_handle",
+    ]:
+        if not getattr(settings, attr):
+            console.print(f"[red]missing {attr}[/red]")
+            return
+
+    console.print(f"[bold]moderation loop[/bold] ({settings.anthropic_model})")
+    if dry_run:
+        console.print("[yellow]DRY RUN[/yellow]")
+
+    dm = DMClient(settings.bot_handle, settings.bot_password, settings.recipient_handle)
+    mod = ModClient(settings.moderation_service_url, settings.moderation_auth_token)
+
+    try:
+        await dm.setup()
+
+        # get pending flags
+        pending = await mod.list_pending()
+        if not pending:
+            console.print("[green]no pending flags[/green]")
+            return
+
+        console.print(f"[bold]{len(pending)} pending flags[/bold]")
+
+        # analyze flags
+        if limit:
+            pending = pending[:limit]
+
+        analyzer = create_flag_analyzer(
+            settings.anthropic_api_key, settings.anthropic_model
+        )
+        desc = "\n---\n".join(
+            f"URI: {f['uri']}\nTrack: {f.get('context', {}).get('track_title', '?')}\n"
+            f"Uploader: @{f.get('context', {}).get('artist_handle', '?')}\n"
+            f"Matches: {', '.join(m['artist'] for m in f.get('context', {}).get('matches', [])[:3])}"
+            for f in pending
+        )
+        result = await analyzer.run(f"analyze {len(pending)} flags:\n\n{desc}")
+        analyses = result.output
+
+        # auto-resolve false positives
+        auto = [a for a in analyses if a.category == "FALSE_POSITIVE"]
+        human = [a for a in analyses if a.category == "NEEDS_HUMAN"]
+        console.print(f"analysis: {len(auto)} auto-resolve, {len(human)} need human")
+
+        for a in auto:
+            if not dry_run:
+                try:
+                    await mod.resolve(
+                        a.uri, "fingerprint_noise", f"auto: {a.reason[:50]}"
+                    )
+                    console.print(f"  [green]✓[/green] {a.uri[-40:]}")
+                except Exception as e:
+                    console.print(f"  [red]✗[/red] {e}")
+
+        # create batch and send link for needs_human (if any)
+        if human:
+            human_uris = [h.uri for h in human]
+            console.print(f"[dim]creating batch for {len(human_uris)} flags...[/dim]")
+
+            if not dry_run:
+                batch = await mod.create_batch(human_uris, created_by="moderation_loop")
+                msg = f"{get_header(env)} {batch['flag_count']} need review:\n{batch['url']}"
+                await dm.send(msg)
+                console.print(f"[green]sent batch {batch['id']}[/green]")
+            else:
+                console.print(
+                    f"[yellow]would create batch with {len(human_uris)} flags[/yellow]"
+                )
+
+        console.print("[bold]done[/bold]")
+
+    finally:
+        await mod.close()
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--dry-run", action="store_true")
+    parser.add_argument("--limit", type=int, default=None)
+    parser.add_argument("--env", default="prod", choices=["dev", "staging", "prod"])
+    args = parser.parse_args()
+    asyncio.run(run_loop(dry_run=args.dry_run, limit=args.limit, env=args.env))
+
+
+if __name__ == "__main__":
+    main()