feat(workflow): add safe page clone and workflow lab

agent/api/posts.py

@@ -123,6 +123,47 @@ async def list_scheduled_posts():
     return await crud.list_scheduled_posts()
 
 
+@router.post("/{post_id}/queue")
+async def queue_post(post_id: str):
+    """Queue one reviewed draft through the standard dry-run/approval gate."""
+    post = await crud.get_post(post_id)
+    if not post:
+        raise HTTPException(404, "Post not found")
+    if post.get("status") != "DRAFT":
+        raise HTTPException(409, "Only DRAFT posts can be queued")
+
+    task_type = f"POST_{post.get('post_type', 'TEXT')}"
+    payload = {
+        "content": post.get("content", ""),
+        "targetType": post.get("target_type", "TIMELINE"),
+        "targetId": post.get("target_id"),
+    }
+    if post.get("media_paths"):
+        try:
+            payload["mediaPaths"] = json.loads(post["media_paths"])
+        except (json.JSONDecodeError, TypeError):
+            raise HTTPException(409, "Post media paths are unreadable")
+    try:
+        payload = enforce_payload(task_type, payload)
+    except ValueError as exc:
+        raise HTTPException(422, str(exc)) from exc
+
+    queued_post = await crud.claim_draft_post_for_queue(post_id)
+    if not queued_post:
+        raise HTTPException(409, "Post was already queued or changed")
+    try:
+        task = await crud.create_task(
+            account_id=queued_post["account_id"],
+            task_type=task_type,
+            payload=json.dumps(payload),
+            ref_id=queued_post["id"],
+        )
+    except Exception:
+        await crud.update_post(post_id, status="DRAFT")
+        raise
+    return {"post": queued_post, "task": task}
+
+
 @router.get("/{post_id}")
 async def get_post(post_id: str):
     post = await crud.get_post(post_id)

agent/api/tasks.py

@@ -1,12 +1,15 @@
 """FBKit — Task API routes."""
 import json
 from json import JSONDecodeError
+from pathlib import Path
 from fastapi import APIRouter, HTTPException
 from pydantic import BaseModel
 from typing import Optional
 
 from agent import config
+from agent.config import MEDIA_DIR
 from agent.db import crud
+from agent.services.page_clone_contract import normalize_page_clone_task_payload
 from agent.services.safety_gate import enforce_payload, strip_server_owned_payload_fields
 
 router = APIRouter(prefix="/tasks", tags=["tasks"])
@@ -16,6 +19,23 @@ def _strip_external_server_fields(payload: dict) -> dict:
     return strip_server_owned_payload_fields(payload)
 
 
+def _safe_page_clone_media_paths(post: dict) -> list[tuple[str, str]]:
+    root = Path(MEDIA_DIR).resolve()
+    paths = []
+    for item in post.get("media", []) if isinstance(post.get("media"), list) else []:
+        if not isinstance(item, dict) or not isinstance(item.get("media_path"), str):
+            continue
+        try:
+            path = Path(item["media_path"]).resolve()
+            path.relative_to(root)
+            if path.is_file():
+                media_type = "video" if item.get("type") == "video" else "image"
+                paths.append((str(path), media_type))
+        except (OSError, ValueError):
+            continue
+    return paths
+
+
 class TaskCreate(BaseModel):
     account_id: str
     task_type: str
@@ -43,6 +63,13 @@ class LiveArmCreate(BaseModel):
     created_by: Optional[str] = None
 
 
+class PageCloneDraftCreate(BaseModel):
+    """Operator-selected Page Clone evidence to save as local drafts."""
+    account_id: str
+    target_id: str
+    selected_post_indexes: list[int]
+
+
 @router.get("")
 async def list_tasks(status: str = None, task_type: str = None, account_id: str = None):
     return await crud.list_tasks(status=status, task_type=task_type, account_id=account_id)
@@ -64,7 +91,13 @@ async def pending_count():
 async def create_task(body: TaskCreate):
     kwargs = {}
     payload = _strip_external_server_fields(dict(body.payload or {}))
-    payload = enforce_payload(body.task_type, payload)
+    try:
+        if body.task_type == "SCRAPE_PAGE_CLONE":
+            payload = normalize_page_clone_task_payload(payload)
+        else:
+            payload = enforce_payload(body.task_type, payload)
+    except ValueError as exc:
+        raise HTTPException(422, str(exc)) from exc
     if payload:
         kwargs["payload"] = json.dumps(payload)
     if body.ref_id:
@@ -82,6 +115,64 @@ async def create_task(body: TaskCreate):
     )
 
 
+@router.post("/{task_id}/page-clone-drafts")
+async def create_page_clone_drafts(task_id: str, body: PageCloneDraftCreate):
+    """Create reviewed local drafts only; publishing remains a separate guarded task."""
+    source_task = await crud.get_task(task_id)
+    if not source_task:
+        raise HTTPException(404, "Page Clone task not found")
+    if source_task.get("task_type") != "SCRAPE_PAGE_CLONE" or source_task.get("status") != "COMPLETED":
+        raise HTTPException(409, "Page Clone evidence must be completed before creating drafts")
+    if source_task.get("account_id") != body.account_id:
+        raise HTTPException(409, "Draft account must match the Page Clone source account")
+    target_id = body.target_id.strip()
+    if not target_id:
+        raise HTTPException(422, "target_id is required")
+    try:
+        target_id = enforce_payload(
+            "POST_TEXT", {"targetType": "PAGE", "targetId": target_id}
+        )["targetId"]
+    except ValueError as exc:
+        raise HTTPException(422, str(exc)) from exc
+    indexes = list(dict.fromkeys(body.selected_post_indexes))
+    if not indexes or len(indexes) > 8 or any(index < 0 for index in indexes):
+        raise HTTPException(422, "Select between 1 and 8 valid post indexes")
+
+    try:
+        result = json.loads(source_task.get("result") or "{}")
+        posts = result.get("data", {}).get("posts", [])
+    except (JSONDecodeError, AttributeError):
+        raise HTTPException(409, "Page Clone evidence is unreadable")
+    if not isinstance(posts, list):
+        raise HTTPException(409, "Page Clone evidence has no post list")
+
+    draft_contents = []
+    for index in indexes:
+        if index >= len(posts) or not isinstance(posts[index], dict):
+            raise HTTPException(422, f"Invalid post index: {index}")
+        content = str(posts[index].get("message") or "").strip()[:500]
+        cached_media = _safe_page_clone_media_paths(posts[index])
+        media_paths = [path for path, _ in cached_media]
+        if not content and not media_paths:
+            raise HTTPException(422, f"Selected post {index} has no text or cached media to draft")
+        post_type = "VIDEO" if any(media_type == "video" for _, media_type in cached_media) else "IMAGE"
+        draft_contents.append((content, media_paths[:10], post_type))
+
+    drafts = [
+        await crud.create_post(
+            account_id=body.account_id,
+            post_type=post_type if media_paths else "TEXT",
+            content=content,
+            media_paths=json.dumps(media_paths) if media_paths else None,
+            target_type="PAGE",
+            target_id=target_id,
+            status="DRAFT",
+        )
+        for content, media_paths, post_type in draft_contents
+    ]
+    return {"source_task_id": task_id, "drafts": drafts}
+
+
 @router.post("/{task_id}/approve")
 async def approve_task(task_id: str):
     task = await crud.get_task(task_id)

agent/api/workflows.py

@@ -0,0 +1,80 @@
+"""Inspect-only local Workflow Lab API; capture start and event ingest are absent."""
+
+from __future__ import annotations
+
+import os
+import hmac
+from fastapi import APIRouter, Header, HTTPException
+from pydantic import BaseModel, Field
+
+from agent.services.workflow_store import WorkflowStore
+from agent.services.workflow_analyzer import analyze_replayability
+
+router = APIRouter(prefix="/workflow-lab", tags=["workflow-lab"])
+_store = WorkflowStore()
+
+
+class WorkflowReview(BaseModel):
+    decision: str = Field(pattern="^(accept|reject)$")
+    note: str = Field(default="", max_length=500)
+
+
+def _authorize(key: str | None) -> None:
+    expected = os.environ.get("WORKFLOW_LAB_API_KEY", "").strip()
+    if not expected or not key or not hmac.compare_digest(key, expected):
+        raise HTTPException(401, "Workflow Lab key required")
+
+
+@router.get("")
+async def list_workflow_captures(profile_id: str | None = None, x_workflow_profile: str | None = Header(default=None), x_workflow_lab_key: str | None = Header(default=None)):
+    _authorize(x_workflow_lab_key)
+    if not profile_id or profile_id != x_workflow_profile:
+        raise HTTPException(403, "Workflow profile scope required")
+    return {"captures": _store.list_captures(profile_id)}
+
+
+@router.get("/{capture_id}")
+async def inspect_workflow_capture(capture_id: str, x_workflow_profile: str | None = Header(default=None), x_workflow_lab_key: str | None = Header(default=None)):
+    _authorize(x_workflow_lab_key)
+    capture = _store.inspect_capture(capture_id)
+    if not capture or capture["profileId"] != x_workflow_profile:
+        raise HTTPException(404, "Capture not found")
+    return capture
+
+
+@router.post("/{capture_id}/stop")
+async def stop_workflow_capture(capture_id: str, x_workflow_profile: str | None = Header(default=None), x_workflow_lab_key: str | None = Header(default=None)):
+    _authorize(x_workflow_lab_key)
+    current = _store.inspect_capture(capture_id)
+    if not current or current["profileId"] != x_workflow_profile:
+        raise HTTPException(404, "Capture not found")
+    return _store.stop_capture(capture_id)
+
+
+@router.post("/{capture_id}/review")
+async def review_workflow_capture(capture_id: str, body: WorkflowReview, x_workflow_profile: str | None = Header(default=None), x_workflow_lab_key: str | None = Header(default=None)):
+    _authorize(x_workflow_lab_key)
+    capture = _store.inspect_capture(capture_id)
+    if not capture or capture["profileId"] != x_workflow_profile:
+        raise HTTPException(404, "Capture not found")
+    if body.decision == "accept":
+        raise HTTPException(409, "Review cannot promote or execute a workflow in V1")
+    return {"captureId": capture_id, "decision": "reject", "note": body.note[:500], "readOnly": True}
+
+
+@router.get("/{capture_id}/analysis")
+async def analyze_workflow_capture(capture_id: str, x_workflow_profile: str | None = Header(default=None), x_workflow_lab_key: str | None = Header(default=None)):
+    _authorize(x_workflow_lab_key)
+    capture = _store.inspect_capture(capture_id)
+    if not capture or capture["profileId"] != x_workflow_profile:
+        raise HTTPException(404, "Capture not found")
+    return {"captureId": capture_id, **analyze_replayability(capture["events"])}
+
+
+@router.delete("/{capture_id}")
+async def delete_workflow_capture(capture_id: str, x_workflow_profile: str | None = Header(default=None), x_workflow_lab_key: str | None = Header(default=None)):
+    _authorize(x_workflow_lab_key)
+    capture = _store.inspect_capture(capture_id)
+    if not capture or capture["profileId"] != x_workflow_profile or not _store.delete_capture(capture_id):
+        raise HTTPException(404, "Capture not found")
+    return {"ok": True, "logicalRetention": True}

agent/db/crud.py

@@ -15,6 +15,7 @@
 from agent import config
 from agent.config import DATA_ENCRYPTION_KEY
 from agent.services.safety_gate import MUTATING_TASK_TYPES, dry_run_from_payload, enforce_payload, is_mutating_task, strip_server_owned_payload_fields
+from agent.services.page_clone_contract import redact_page_clone_task_payload
 from agent.utils.time import utc_now, utc_now_iso
 
 logger = logging.getLogger(__name__)
@@ -273,6 +274,19 @@ async def claim_scheduled_post(post_id: str, before: str) -> dict | None:
     return await get_post(post_id)
 
 
+async def claim_draft_post_for_queue(post_id: str) -> dict | None:
+    """Atomically transition one reviewed draft to queued status."""
+    db = await get_db()
+    cur = await db.execute(
+        "UPDATE post SET status = 'SCHEDULED', updated_at = ? WHERE id = ? AND status = 'DRAFT'",
+        (utc_now_iso(), post_id),
+    )
+    await db.commit()
+    if cur.rowcount != 1:
+        return None
+    return await get_post(post_id)
+
+
 async def delete_post(post_id: str) -> bool:
     db = await get_db()
     cur = await db.execute("DELETE FROM post WHERE id = ?", (post_id,))
@@ -596,7 +610,17 @@ def _task_requires_live_account_lease(row) -> bool:
 
 
 async def cancel_task(task_id: str) -> dict | None:
-    return await update_task(task_id, status="CANCELLED")
+    task = await get_task(task_id)
+    if not task:
+        return None
+    updates = {"status": "CANCELLED"}
+    if task.get("task_type") == "SCRAPE_PAGE_CLONE":
+        try:
+            payload = json.loads(task.get("payload") or "{}")
+        except JSONDecodeError:
+            payload = {}
+        updates["payload"] = json.dumps(redact_page_clone_task_payload(payload))
+    return await update_task(task_id, **updates)
 
 
 # ─── FB Group ───────────────────────────────────────────────