From 2c4fed314e60fe99af9a6a3af7991c2a5179ca19 Mon Sep 17 00:00:00 2001 From: hthmkt12 Date: Wed, 22 Jul 2026 16:25:25 +0700 Subject: [PATCH 1/3] feat(page-clone): add safe page cloning workflow --- agent/api/posts.py | 41 +++ agent/api/tasks.py | 93 ++++++- agent/db/crud.py | 26 +- agent/db/schema.py | 102 +++++++- agent/services/fb_client.py | 22 ++ agent/services/page_clone_contract.py | 244 ++++++++++++++++++ agent/services/page_clone_media.py | 120 +++++++++ agent/services/safety_gate.py | 7 + agent/worker/processor.py | 91 ++++++- docs/page-clone-usage.md | 50 ++++ extension/content-fb.js | 95 +++++++ tests/unit/test_extension_dry_run.py | 24 ++ tests/unit/test_page_clone_contract.py | 125 +++++++++ tests/unit/test_page_clone_dispatch.py | 164 ++++++++++++ tests/unit/test_page_clone_draft_queue.py | 68 +++++ tests/unit/test_page_clone_drafts.py | 170 ++++++++++++ tests/unit/test_page_clone_media.py | 115 +++++++++ .../unit/test_page_clone_schema_migration.py | 53 ++++ tests/unit/test_page_clone_task_api.py | 85 ++++++ tests/unit/test_safety_gate.py | 13 + 20 files changed, 1696 insertions(+), 12 deletions(-) create mode 100644 agent/services/page_clone_contract.py create mode 100644 agent/services/page_clone_media.py create mode 100644 docs/page-clone-usage.md create mode 100644 tests/unit/test_page_clone_contract.py create mode 100644 tests/unit/test_page_clone_dispatch.py create mode 100644 tests/unit/test_page_clone_draft_queue.py create mode 100644 tests/unit/test_page_clone_drafts.py create mode 100644 tests/unit/test_page_clone_media.py create mode 100644 tests/unit/test_page_clone_schema_migration.py create mode 100644 tests/unit/test_page_clone_task_api.py diff --git a/agent/api/posts.py b/agent/api/posts.py index 7bb62d00..95155afb 100644 --- a/agent/api/posts.py +++ b/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) diff --git a/agent/api/tasks.py b/agent/api/tasks.py index 76e504df..c08eca83 100644 --- a/agent/api/tasks.py +++ b/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) diff --git a/agent/db/crud.py b/agent/db/crud.py index 8c73adda..f407d3bb 100644 --- a/agent/db/crud.py +++ b/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 ─────────────────────────────────────────────── diff --git a/agent/db/schema.py b/agent/db/schema.py index 45b79eb9..f4372340 100644 --- a/agent/db/schema.py +++ b/agent/db/schema.py @@ -90,7 +90,7 @@ 'ADD_FRIEND','ACCEPT_FRIEND', 'JOIN_GROUP','LEAVE_GROUP', 'FOLLOW_PAGE','UNFOLLOW_PAGE', - 'SCRAPE_PROFILE','SCRAPE_GROUP', + 'SCRAPE_PROFILE','SCRAPE_GROUP','SCRAPE_PAGE_CLONE', 'CHECK_LOGIN' )), payload TEXT, -- JSON payload @@ -290,6 +290,105 @@ async def get_db() -> aiosqlite.Connection: ] +_TASK_COLUMNS = ( + "id", "account_id", "task_type", "payload", "ref_id", "status", "priority", + "retry_count", "max_retries", "scheduled_at", "started_at", "completed_at", + "result", "metrics_synced_at", "error_message", "created_at", "updated_at", +) + +_TASK_TABLE_SQL = """ +CREATE TABLE task ( + id TEXT PRIMARY KEY, + account_id TEXT REFERENCES account(id), + task_type TEXT NOT NULL CHECK(task_type IN ( + 'POST_TEXT','POST_IMAGE','POST_VIDEO','POST_LINK', + 'POST_STORY','POST_REEL','REUP_VIDEO', + 'SEND_MESSAGE','SEND_BULK_MESSAGE', + 'LIKE_POST','COMMENT_POST','SHARE_POST', + 'ADD_FRIEND','ACCEPT_FRIEND', + 'JOIN_GROUP','LEAVE_GROUP', + 'FOLLOW_PAGE','UNFOLLOW_PAGE', + 'SCRAPE_PROFILE','SCRAPE_GROUP','SCRAPE_PAGE_CLONE', + 'CHECK_LOGIN' + )), + payload TEXT, + ref_id TEXT, + status TEXT DEFAULT 'PENDING' + CHECK(status IN ('PENDING','PROCESSING','COMPLETED','FAILED','CANCELLED')), + priority INTEGER DEFAULT 0, + retry_count INTEGER DEFAULT 0, + max_retries INTEGER DEFAULT 3, + scheduled_at DATETIME, + started_at DATETIME, + completed_at DATETIME, + result TEXT, + metrics_synced_at DATETIME, + error_message TEXT, + created_at DATETIME DEFAULT (datetime('now')), + updated_at DATETIME DEFAULT (datetime('now')) +) +""" + + +async def _migrate_task_table_for_page_clone(db: aiosqlite.Connection) -> None: + """Rebuild legacy ``task`` tables without losing queued task records. + + SQLite cannot alter a CHECK constraint in place. Keeping foreign-key + references unchanged during the temporary rename avoids rebuilding the + dependent trace/lease tables too. + """ + row = await (await db.execute( + "SELECT sql FROM sqlite_master WHERE type = 'table' AND name = 'task'" + )).fetchone() + if not row or "SCRAPE_PAGE_CLONE" in (row[0] or ""): + return + + legacy_columns = { + column[1] for column in await (await db.execute("PRAGMA table_info(task)")).fetchall() + } + copy_columns = [column for column in _TASK_COLUMNS if column in legacy_columns] + if not {"id", "task_type"}.issubset(copy_columns): + raise RuntimeError("Cannot migrate task table: required columns are missing") + + # PRAGMA foreign_keys must change outside a transaction. legacy_alter_table + # keeps child FK declarations pointing at the replacement table name. + foreign_keys = (await (await db.execute("PRAGMA foreign_keys")).fetchone())[0] + await db.commit() + await db.execute("PRAGMA foreign_keys=OFF") + await db.execute("PRAGMA legacy_alter_table=ON") + try: + await db.execute("BEGIN IMMEDIATE") + await db.execute("ALTER TABLE task RENAME TO task_legacy_page_clone") + await db.execute(_TASK_TABLE_SQL) + names = ", ".join(copy_columns) + await db.execute( + f"INSERT INTO task ({names}) SELECT {names} FROM task_legacy_page_clone" + ) + await db.execute("DROP TABLE task_legacy_page_clone") + await db.executescript(""" + CREATE INDEX IF NOT EXISTS idx_task_status ON task(status); + CREATE INDEX IF NOT EXISTS idx_task_type ON task(task_type); + CREATE INDEX IF NOT EXISTS idx_task_scheduled ON task(scheduled_at); + CREATE INDEX IF NOT EXISTS idx_task_priority ON task(priority DESC); + CREATE INDEX IF NOT EXISTS idx_task_status_scheduled_priority ON task(status, scheduled_at, priority DESC); + CREATE INDEX IF NOT EXISTS idx_task_account_status ON task(account_id, status); + CREATE UNIQUE INDEX IF NOT EXISTS idx_task_zoopost_ref ON task(ref_id) WHERE ref_id LIKE 'zoopost:%'; + CREATE INDEX IF NOT EXISTS idx_task_metrics_due ON task(status, metrics_synced_at, completed_at); + """) + await db.commit() + except Exception: + await db.rollback() + raise + finally: + await db.execute("PRAGMA legacy_alter_table=OFF") + await db.execute(f"PRAGMA foreign_keys={1 if foreign_keys else 0}") + + violations = await (await db.execute("PRAGMA foreign_key_check")).fetchall() + if violations: + raise RuntimeError("Task table migration failed foreign-key integrity check") + logger.info("Migrated task table to support SCRAPE_PAGE_CLONE") + + async def init_db(): db = await get_db() await db.executescript(SCHEMA) @@ -301,6 +400,7 @@ async def init_db(): # Safe for re-run / legacy DBs where column/index already exists. logger.debug("Migration skipped (%s): %s", stmt, exc) + await _migrate_task_table_for_page_clone(db) await db.commit() logger.info("Database initialized: %s", DB_PATH) diff --git a/agent/services/fb_client.py b/agent/services/fb_client.py index a9d2c36b..9a3b9cf0 100644 --- a/agent/services/fb_client.py +++ b/agent/services/fb_client.py @@ -176,6 +176,13 @@ def session_live_guard_enabled(self, fb_uid: Optional[str] = None) -> bool: session = self.get_session_for(fb_uid) return bool(session and session.extension_live_actions_enabled is True) + def page_clone_session_ready(self, fb_uid: str | None) -> bool: + """Require an exact, fresh, authenticated session for Page Clone reads.""" + if not fb_uid: + return False + session = self.get_session_for(fb_uid) + return bool(session and session.logged_in) + @property def connected(self) -> bool: return bool(self._sessions) @@ -455,6 +462,21 @@ async def scrape_group(self, group_url: str, fb_uid: str = None, fb_uid=fb_uid, ) + async def scrape_page_clone(self, source_url: str, max_posts: int = 25, + max_media_per_post: int = 10, + deadline_seconds: int = 30, fb_uid: str = None, + strategy: dict | None = None) -> dict: + return await self._send( + "scrape_page_clone", + self._with_strategy({ + "sourceUrl": source_url, + "maxPosts": max_posts, + "maxMediaPerPost": max_media_per_post, + }, strategy), + fb_uid=fb_uid, + timeout=deadline_seconds, + ) + async def get_page_state(self, fb_uid: str = None) -> dict: return await self._send("get_page_state", {}, fb_uid=fb_uid) diff --git a/agent/services/page_clone_contract.py b/agent/services/page_clone_contract.py new file mode 100644 index 00000000..10ccd827 --- /dev/null +++ b/agent/services/page_clone_contract.py @@ -0,0 +1,244 @@ +"""Contracts and redaction helpers for the read-only page-clone slice.""" + +from __future__ import annotations + +import hashlib +from pathlib import Path +import re +from urllib.parse import parse_qs, urlsplit, urlunsplit + +from agent.config import MEDIA_DIR + + +MAX_POSTS = 25 +MAX_CANDIDATES = 8 +MAX_MEDIA_PER_POST = 10 +MAX_DEADLINE_SECONDS = 30 +MAX_TEXT_CHARS = 500 + +_ALLOWED_KEYS = { + "source_url", + "max_posts", + "candidate_limit", + "max_media_per_post", + "deadline_seconds", + "download_media", +} +_WIRE_KEYS = { + "sourceUrl", + "maxPosts", + "candidateLimit", + "maxMediaPerPost", + "deadlineSeconds", + "downloadMedia", +} +_RESERVED_PAGE_PATHS = { + "groups", + "watch", + "reel", + "reels", + "posts", + "events", + "marketplace", + "login", + "share", + "photo", +} + + +def _hash(value: object) -> str: + return "sha256:" + hashlib.sha256(str(value).encode("utf-8")).hexdigest() + + +def canonicalize_page_url(value: str) -> str: + """Return a canonical HTTPS Facebook page URL or raise ``ValueError``.""" + + if not isinstance(value, str) or not value.strip() or len(value.strip()) > 2048: + raise ValueError("source_url must be a non-empty string") + parsed = urlsplit(value.strip()) + host = (parsed.hostname or "").lower().rstrip(".") + if parsed.scheme.lower() != "https" or not ( + host == "facebook.com" or host.endswith(".facebook.com") + ): + raise ValueError("source_url must use an HTTPS Facebook host") + if parsed.username or parsed.password or parsed.fragment: + raise ValueError("source_url contains unsupported URL components") + + path_parts = [part for part in parsed.path.split("/") if part] + if not path_parts: + raise ValueError("source_url must identify a page") + first = path_parts[0].lower() + query = parse_qs(parsed.query, keep_blank_values=False) + if first == "profile.php": + page_id = query.get("id", [""])[0] + if len(path_parts) != 1 or not page_id.isdigit() or len(page_id) < 5: + raise ValueError("profile.php page URL requires a numeric id") + return "https://www.facebook.com/profile.php?id=" + page_id + if first in _RESERVED_PAGE_PATHS or len(path_parts) > 1 and first != "pages": + raise ValueError("URL does not identify a Facebook page") + if first == "pages": + if len(path_parts) != 3 or not path_parts[2].isdigit() or len(path_parts[2]) < 5: + raise ValueError("/pages URL requires a numeric page id") + path = "/" + "/".join(path_parts) + else: + path = "/" + path_parts[0] + return urlunsplit(("https", "www.facebook.com", path, "", "")) + + +def _bounded_int(payload: dict, key: str, default: int, maximum: int) -> int: + value = payload.get(key, default) + if isinstance(value, bool): + raise ValueError(f"{key} must be an integer") + try: + value = int(value) + except (TypeError, ValueError) as exc: + raise ValueError(f"{key} must be an integer") from exc + if value < 1 or value > maximum: + raise ValueError(f"{key} must be between 1 and {maximum}") + return value + + +def normalize_page_clone_request(payload: dict) -> dict: + if not isinstance(payload, dict): + raise ValueError("page-clone request must be an object") + unknown = set(payload) - _ALLOWED_KEYS + if unknown: + raise ValueError("unsupported page-clone fields: " + ", ".join(sorted(unknown))) + download_media = payload.get("download_media", False) + if not isinstance(download_media, bool): + raise ValueError("download_media must be a boolean") + return { + "source_url": canonicalize_page_url(payload.get("source_url", "")), + "max_posts": _bounded_int(payload, "max_posts", MAX_POSTS, MAX_POSTS), + "candidate_limit": _bounded_int(payload, "candidate_limit", MAX_CANDIDATES, MAX_CANDIDATES), + "max_media_per_post": _bounded_int( + payload, "max_media_per_post", MAX_MEDIA_PER_POST, MAX_MEDIA_PER_POST + ), + "deadline_seconds": _bounded_int( + payload, "deadline_seconds", MAX_DEADLINE_SECONDS, MAX_DEADLINE_SECONDS + ), + "download_media": download_media, + } + + +def normalize_page_clone_task_payload(payload: dict) -> dict: + """Validate the public camelCase task payload and return its canonical form.""" + if not isinstance(payload, dict): + raise ValueError("page-clone task payload must be an object") + unknown = set(payload) - _WIRE_KEYS + if unknown: + raise ValueError("unsupported page-clone fields: " + ", ".join(sorted(unknown))) + request = normalize_page_clone_request({ + "source_url": payload.get("sourceUrl", ""), + "max_posts": payload.get("maxPosts", MAX_POSTS), + "candidate_limit": payload.get("candidateLimit", MAX_CANDIDATES), + "max_media_per_post": payload.get("maxMediaPerPost", MAX_MEDIA_PER_POST), + "deadline_seconds": payload.get("deadlineSeconds", MAX_DEADLINE_SECONDS), + "download_media": payload.get("downloadMedia", False), + }) + return { + "sourceUrl": request["source_url"], + "maxPosts": request["max_posts"], + "candidateLimit": request["candidate_limit"], + "maxMediaPerPost": request["max_media_per_post"], + "deadlineSeconds": request["deadline_seconds"], + "downloadMedia": request["download_media"], + } + + +def redact_page_clone_task_payload(payload: dict) -> dict: + """Replace a terminal task's raw source URL with a durable local reference.""" + raw_payload = payload if isinstance(payload, dict) else {} + try: + request = normalize_page_clone_task_payload(raw_payload) + except ValueError: + return { + "schemaVersion": 1, + "sourceRef": _hash(raw_payload.get("sourceUrl", "")), + } + return { + "schemaVersion": 1, + "sourceRef": _hash(request["sourceUrl"]), + "maxPosts": request["maxPosts"], + "candidateLimit": request["candidateLimit"], + "maxMediaPerPost": request["maxMediaPerPost"], + "deadlineSeconds": request["deadlineSeconds"], + } + + +def _safe_text(value: object) -> str: + return str(value or "")[:MAX_TEXT_CHARS] + + +def _safe_warnings(value: object) -> list[str]: + """Keep bounded diagnostics while removing raw media URLs and query tokens.""" + if not isinstance(value, list): + return [] + warnings = [] + for warning in value: + if not isinstance(warning, str): + continue + warning = re.sub(r"https?://\S+", "[redacted URL]", warning).strip() + if warning: + warnings.append(warning[:MAX_TEXT_CHARS]) + if len(warnings) >= MAX_MEDIA_PER_POST: + break + return warnings + + +def redact_page_clone_result(result: dict) -> dict: + """Create a durable, secret-free evidence representation.""" + + if not isinstance(result, dict): + raise ValueError("page-clone result must be an object") + source_url = canonicalize_page_url(result.get("source_url", "")) + profile = result.get("profile") if isinstance(result.get("profile"), dict) else {} + redacted = { + "schema_version": 1, + "source_ref": _hash(source_url), + "profile": { + "id_hash": _hash(profile.get("id", "")), + "name": _safe_text(profile.get("name")), + "category": _safe_text(profile.get("category")), + }, + "posts": [], + "warnings": _safe_warnings(result.get("warnings")), + } + for post in result.get("posts", []) if isinstance(result.get("posts"), list) else []: + if not isinstance(post, dict): + continue + media = [] + for item in post.get("media", []) if isinstance(post.get("media"), list) else []: + if not isinstance(item, dict): + continue + parsed = urlsplit(str(item.get("url", ""))) + if parsed.scheme != "https" or not parsed.hostname: + continue + redacted_media = { + "host": parsed.hostname.lower(), + "url_hash": _hash(parsed.geturl()), + "type": _safe_text(item.get("type")), + } + local_path = item.get("local_path") + if isinstance(local_path, str) and local_path: + try: + media_root = Path(MEDIA_DIR).resolve() + safe_path = Path(local_path).resolve() + safe_path.relative_to(media_root) + redacted_media["media_path"] = str(safe_path) + except (OSError, ValueError): + pass + media.append(redacted_media) + if len(media) >= MAX_MEDIA_PER_POST: + break + redacted["posts"].append( + { + "id_hash": _hash(post.get("id", "")), + "message": _safe_text(post.get("message")), + "created_time": _safe_text(post.get("created_time")), + "media": media, + } + ) + if len(redacted["posts"]) >= MAX_POSTS: + break + return redacted diff --git a/agent/services/page_clone_media.py b/agent/services/page_clone_media.py new file mode 100644 index 00000000..509c0d20 --- /dev/null +++ b/agent/services/page_clone_media.py @@ -0,0 +1,120 @@ +"""Bounded, allowlisted Page Clone image caching.""" + +from __future__ import annotations + +from pathlib import Path +import re +from urllib.parse import urlsplit + +import httpx + +from agent.config import MEDIA_DIR + +MAX_IMAGE_BYTES = 8 * 1024 * 1024 +MAX_VIDEO_BYTES = 100 * 1024 * 1024 +MAX_TOTAL_BYTES = 256 * 1024 * 1024 +_ALLOWED_SUFFIXES = (".facebook.com", ".fbcdn.net", ".fbsbx.com") + + +def _image_extension(body: bytes) -> str | None: + if body.startswith(b"\xff\xd8\xff"): + return ".jpg" + if body.startswith(b"\x89PNG\r\n\x1a\n"): + return ".png" + if body.startswith((b"GIF87a", b"GIF89a")): + return ".gif" + if body.startswith(b"RIFF") and body[8:12] == b"WEBP": + return ".webp" + return None + + +def _video_extension(body: bytes) -> str | None: + return ".mp4" if len(body) >= 12 and body[4:8] == b"ftyp" else None + + +def _allowed_media_url(value: object) -> str | None: + try: + parsed = urlsplit(str(value)) + except ValueError: + return None + host = (parsed.hostname or "").lower().rstrip(".") + if parsed.scheme != "https" or not host: + return None + if host != "facebook.com" and not host.endswith(_ALLOWED_SUFFIXES): + return None + return parsed.geturl() + + +async def cache_page_clone_media(result: dict, task_id: str) -> dict: + """Download bounded image media without following redirects or arbitrary hosts.""" + data = result.get("data") if isinstance(result.get("data"), dict) else result + posts = data.get("posts") if isinstance(data, dict) else None + if not isinstance(posts, list): + return result + + root = Path(MEDIA_DIR).resolve() + root.mkdir(parents=True, exist_ok=True) + total_bytes = 0 + warnings = data.setdefault("warnings", []) if isinstance(data, dict) else [] + async with httpx.AsyncClient(follow_redirects=False, timeout=10.0) as client: + for post_index, post in enumerate(posts[:25]): + if not isinstance(post, dict) or not isinstance(post.get("media"), list): + continue + for media_index, item in enumerate(post["media"][:10]): + if not isinstance(item, dict): + continue + url = _allowed_media_url(item.get("url")) + if not url: + warnings.append(f"media {post_index}:{media_index} rejected by host policy") + continue + try: + async with client.stream("GET", url) as response: + if response.is_redirect or response.status_code >= 400: + raise ValueError("redirect or HTTP error") + content_type = (response.headers.get("content-type") or "").split(";", 1)[0].lower() + is_video = item.get("type") == "video" + expected_type = "video/" if is_video else "image/" + if not content_type.startswith(expected_type): + raise ValueError(f"media is not a {expected_type[:-1]}") + content_length = int(response.headers.get("content-length") or 0) + max_bytes = MAX_VIDEO_BYTES if is_video else MAX_IMAGE_BYTES + if content_length > max_bytes: + raise ValueError("media exceeds size limit") + chunks = [] + received = 0 + async for chunk in response.aiter_bytes(): + received += len(chunk) + if received > max_bytes or total_bytes + received > MAX_TOTAL_BYTES: + raise ValueError("media cache size limit exceeded") + chunks.append(chunk) + body = b"".join(chunks) + extension = _video_extension(body) if is_video else _image_extension(body) + if not extension: + raise ValueError("media content does not match its declared type") + path = (root / f"page-clone-{task_id[:12]}-{post_index}-{media_index}{extension}").resolve() + path.relative_to(root) + path.write_bytes(body) + total_bytes += len(body) + item["local_path"] = str(path) + except (httpx.HTTPError, OSError, ValueError) as exc: + warnings.append(f"media {post_index}:{media_index} skipped: {exc}") + return result + + +def cleanup_page_clone_media(task_id: str) -> int: + """Remove only this task's bounded cache files after cancellation/failure.""" + safe_task_id = re.sub(r"[^A-Za-z0-9_-]", "", str(task_id))[:12] + if not safe_task_id: + return 0 + root = Path(MEDIA_DIR).resolve() + removed = 0 + for path in root.glob(f"page-clone-{safe_task_id}-*"): + try: + resolved = path.resolve() + resolved.relative_to(root) + if resolved.is_file(): + resolved.unlink() + removed += 1 + except OSError: + continue + return removed diff --git a/agent/services/safety_gate.py b/agent/services/safety_gate.py index 6df4eb16..dd14287e 100644 --- a/agent/services/safety_gate.py +++ b/agent/services/safety_gate.py @@ -7,6 +7,7 @@ from __future__ import annotations from copy import deepcopy +import re from typing import Any from agent import config @@ -86,6 +87,12 @@ def enforce_payload(task_type: str, payload: dict | None) -> dict: safe_payload["groupUrl"] = f"https://facebook.com/groups/{target_id}" else: raise ValueError("group targetType requires a non-empty groupUrl or targetId") + elif safe_payload.get("targetType") == "PAGE": + target_id = safe_payload.get("targetId") + if not isinstance(target_id, str) or not target_id.strip(): + raise ValueError("page targetType requires a non-empty targetId") + if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._-]{0,99}", target_id.strip()): + raise ValueError("page targetType targetId must be a Facebook page id or slug") elif safe_payload.get("targetType") == "POST": post_url = safe_payload.get("postUrl") if not post_url or not isinstance(post_url, str) or not post_url.strip(): diff --git a/agent/worker/processor.py b/agent/worker/processor.py index 4678f48c..b8f7a88a 100644 --- a/agent/worker/processor.py +++ b/agent/worker/processor.py @@ -22,10 +22,32 @@ from agent.services.event_bus import event_bus from agent.services.notifier import get_notifier from agent.services.safety_gate import dry_run_from_payload, enforce_payload, is_mutating_task +from agent.services.page_clone_contract import ( + normalize_page_clone_task_payload, + redact_page_clone_result, + redact_page_clone_task_payload, +) from agent.utils.time import utc_from_timestamp_iso, utc_now_iso, utc_now_ms logger = logging.getLogger(__name__) + +def _persistable_result(task_type: str, result: dict) -> dict: + """Redact page-clone URLs/IDs before any task result is persisted.""" + if task_type != "SCRAPE_PAGE_CLONE" or not isinstance(result, dict): + return result + data = result.get("data") if isinstance(result.get("data"), dict) else result + try: + redacted = redact_page_clone_result(data) + except ValueError as exc: + return {"success": False, "error": str(exc), "code": "INVALID_PAGE_CLONE_RESULT"} + return {"success": bool(result.get("success", True)), "data": redacted} + + +async def _task_is_cancelled(task_id: str) -> bool: + current = await crud.get_task(task_id) + return bool(current and current.get("status") == "CANCELLED") + # Map task_type → daily counter field in account table _COUNTER_MAP = { "POST_TEXT": "daily_posts", @@ -396,6 +418,11 @@ async def _process_task( strategy.get("fail_count", 0), ) + if await _task_is_cancelled(task_id): + logger.info("Task %s was cancelled before dispatch", task_id[:8]) + await event_bus.emit("task_cancelled", {"task_id": task_id, "type": task_type}) + return + # Mark as processing await crud.update_task(task_id, status="PROCESSING", started_at=utc_now_iso()) @@ -413,17 +440,27 @@ async def _process_task( result = await self._dispatch(task_type, payload, task, fb_uid=fb_uid, strategy=strategy) + if await _task_is_cancelled(task_id): + logger.info("Task %s was cancelled while dispatching", task_id[:8]) + if task_type == "SCRAPE_PAGE_CLONE": + from agent.services.page_clone_media import cleanup_page_clone_media + cleanup_page_clone_media(task_id) + await event_bus.emit("task_cancelled", {"task_id": task_id, "type": task_type}) + return + if result.get("error"): raise Exception(result["error"]) # Success duration_ms = utc_now_ms() - started_at_ms - await crud.update_task( - task_id, - status="COMPLETED", - completed_at=utc_now_iso(), - result=json.dumps(result), - ) + completed_update = { + "status": "COMPLETED", + "completed_at": utc_now_iso(), + "result": json.dumps(_persistable_result(task_type, result)), + } + if task_type == "SCRAPE_PAGE_CLONE": + completed_update["payload"] = json.dumps(redact_page_clone_task_payload(payload)) + await crud.update_task(task_id, **completed_update) from agent.services.health_monitor import get_health_monitor if task.get("account_id"): @@ -516,11 +553,16 @@ async def _process_task( max_retries, ) else: + terminal_update = { + "status": "FAILED", + "completed_at": utc_now_iso(), + "error_message": error_message, + } + if task_type == "SCRAPE_PAGE_CLONE": + terminal_update["payload"] = json.dumps(redact_page_clone_task_payload(payload)) await crud.update_task( task_id, - status="FAILED", - completed_at=utc_now_iso(), - error_message=error_message, + **terminal_update, ) await event_bus.emit( "task_failed", @@ -798,6 +840,37 @@ async def _dispatch(self, task_type: str, payload: dict, task: dict, strategy=strategy_hints, ) + elif task_type == "SCRAPE_PAGE_CLONE": + try: + request = normalize_page_clone_task_payload(payload) + except ValueError as exc: + return {"error": str(exc), "code": "INVALID_PAGE_CLONE_REQUEST"} + if not client.page_clone_session_ready(fb_uid): + return { + "error": "Page Clone requires an exact, fresh logged-in Facebook session", + "code": "PAGE_CLONE_SESSION_UNAVAILABLE", + } + result = await client.scrape_page_clone( + source_url=request["sourceUrl"], + max_posts=request["maxPosts"], + max_media_per_post=request["maxMediaPerPost"], + deadline_seconds=request["deadlineSeconds"], + fb_uid=fb_uid, + strategy=strategy_hints, + ) + if request["downloadMedia"] and isinstance(result, dict) and not result.get("error"): + from agent.services.page_clone_media import cache_page_clone_media + try: + result = await asyncio.wait_for( + cache_page_clone_media(result, task.get("id", "page-clone")), + timeout=request["deadlineSeconds"], + ) + except asyncio.TimeoutError: + data = result.get("data") if isinstance(result.get("data"), dict) else result + if isinstance(data, dict): + data.setdefault("warnings", []).append("media download deadline exceeded") + return result + elif task_type == "REUP_VIDEO": from agent.services.downloader import download_video source_url = payload.get("sourceUrl") diff --git a/docs/page-clone-usage.md b/docs/page-clone-usage.md new file mode 100644 index 00000000..aebf2275 --- /dev/null +++ b/docs/page-clone-usage.md @@ -0,0 +1,50 @@ +# Page Clone usage + +This worktree implements a bounded, read-only source-page reader plus an operator-review draft flow. It does not create a Facebook page or publish automatically. + +## 1. Create a source scrape task + +`POST /api/tasks` + +```json +{ + "account_id": "ACCOUNT_ID", + "task_type": "SCRAPE_PAGE_CLONE", + "payload": { + "sourceUrl": "https://www.facebook.com/source-page", + "maxPosts": 25, + "candidateLimit": 8, + "maxMediaPerPost": 10, + "deadlineSeconds": 30, + "downloadMedia": false + } +} +``` + +The source URL must be an HTTPS Facebook page URL. Values above the limits are rejected. Set `downloadMedia: true` only when local image caching is wanted; downloads are restricted to HTTPS Facebook media hosts, no redirects, image content, and bounded sizes. The task requires the account's exact, fresh, logged-in `fb_uid` session. + +## 2. Review evidence + +Read `GET /api/tasks/{task_id}` after the task reaches `COMPLETED`. The durable result is redacted: source URLs, IDs, permalinks, media URLs, and tokens are not stored raw. + +## 3. Create local drafts + +`POST /api/tasks/{task_id}/page-clone-drafts` + +```json +{ + "account_id": "ACCOUNT_ID", + "target_id": "DESTINATION_PAGE_ID_OR_SLUG", + "selected_post_indexes": [0, 2] +} +``` + +This creates local `DRAFT` posts only. It requires a completed Page Clone task, matching account, and 1–8 selected posts. + +## 4. Queue a reviewed draft + +`POST /api/posts/{post_id}/queue` + +The draft is atomically claimed to prevent duplicate queueing. The resulting `POST_TEXT` task passes through the existing safety gate and is dry-run by default. Live publishing requires all existing global auth, live-arm, and approval controls. + +With `downloadMedia: false`, media is metadata-only. With it enabled, allowlisted images and MP4 videos are cached under `MEDIA_DIR`; selected drafts become `IMAGE` or `VIDEO` posts. Facebook upload remains behind the existing dry-run/live approval gates. diff --git a/extension/content-fb.js b/extension/content-fb.js index ee625116..b3cea329 100644 --- a/extension/content-fb.js +++ b/extension/content-fb.js @@ -205,6 +205,9 @@ async function handlePostText(params) { if (targetType === "GROUP" && targetId) { window.location.href = `https://www.facebook.com/groups/${targetId}`; await sleep(3000); + } else if (targetType === "PAGE" && targetId) { + window.location.href = `https://www.facebook.com/${targetId}`; + await sleep(3000); } // Click the composer ("What's on your mind?") @@ -647,6 +650,95 @@ async function handleScrapeProfile(params) { } } +/** + * Read a bounded page snapshot for the Page Clone evidence flow. + * This handler never navigates, uploads, clicks, types, or fetches media URLs. + */ +async function handleScrapePageClone(params = {}) { + const sourceUrl = String(params.sourceUrl || "").trim(); + const maxPosts = Math.min(25, Math.max(1, Number(params.maxPosts) || 25)); + const maxMediaPerPost = Math.min(10, Math.max(1, Number(params.maxMediaPerPost) || 10)); + const isAllowedMediaHost = (hostname) => { + const host = String(hostname || "").toLowerCase(); + return host === "facebook.com" || host.endsWith(".facebook.com") + || host.endsWith(".fbcdn.net") || host.endsWith(".fbsbx.com"); + }; + if (!/^https:\/\/(?:[a-z0-9-]+\.)*facebook\.com\//i.test(sourceUrl)) { + return { error: "sourceUrl must be an HTTPS Facebook page URL" }; + } + if (!/^https:\/\/(?:[a-z0-9-]+\.)*facebook\.com\//i.test(window.location.href)) { + return { error: "Current tab is not a Facebook page" }; + } + const pageKey = (value) => { + const parsed = new URL(value); + const queryId = parsed.searchParams.get("id"); + if (queryId) return `id:${queryId}`; + const parts = parsed.pathname.split("/").filter(Boolean); + return parts[0]?.toLowerCase() === "pages" ? `id:${parts[2] || ""}` : `slug:${(parts[0] || "").toLowerCase()}`; + }; + if (pageKey(sourceUrl) !== pageKey(window.location.href)) { + return { error: "Open the source Facebook page in the active tab before cloning" }; + } + + const profileName = document.querySelector('meta[property="og:title"]')?.content?.trim() + || document.querySelector("h1")?.textContent?.trim() + || document.title.trim(); + const profile = { + id: "", + name: profileName.slice(0, 500), + category: document.querySelector('[data-pagelet="ProfileTilesBio"]')?.textContent?.trim()?.slice(0, 500) || "", + }; + const posts = []; + const seen = new Set(); + for (const article of document.querySelectorAll('[role="article"]')) { + if (posts.length >= maxPosts) break; + const link = [...article.querySelectorAll("a[href]")] + .map(anchor => anchor.href) + .find(href => /facebook\.com\/[^/]+\/posts\/|facebook\.com\/[^/]+\/videos\//i.test(href)); + const id = (link || "").match(/(?:posts|videos)\/(\d+)/i)?.[1] || ""; + const key = id || (link || article.textContent || "").slice(0, 120); + if (seen.has(key)) continue; + seen.add(key); + const imageMedia = [...article.querySelectorAll("img[src]")].map(image => { + const url = String(image.currentSrc || image.src || ""); + try { + const parsed = new URL(url); + if (parsed.protocol !== "https:" || !isAllowedMediaHost(parsed.hostname)) return null; + return { url, type: "image" }; + } catch { + return null; + } + }).filter(Boolean); + const videoMedia = [...article.querySelectorAll("video")].map(video => { + const url = String(video.currentSrc || video.src || ""); + try { + const parsed = new URL(url); + if (parsed.protocol !== "https:" || !isAllowedMediaHost(parsed.hostname)) return null; + return { url, type: "video" }; + } catch { + return null; + } + }).filter(Boolean); + const media = [...imageMedia, ...videoMedia].slice(0, maxMediaPerPost); + posts.push({ + id, + message: String(article.textContent || "").trim().slice(0, 500), + permalink: link || "", + created_time: "", + media, + }); + } + return { + success: true, + data: { + source_url: window.location.href, + profile, + posts, + warnings: posts.length ? [] : ["No visible page posts found"], + }, + }; +} + /** * Get current page state. */ @@ -1570,6 +1662,9 @@ chrome.runtime.onMessage.addListener((message, sender, sendResponse) => { case "get_post_metrics": result = await handleGetPostMetrics(params); break; + case "scrape_page_clone": + result = await handleScrapePageClone(params); + break; default: result = { error: `Unknown method: ${method}` }; } diff --git a/tests/unit/test_extension_dry_run.py b/tests/unit/test_extension_dry_run.py index d231a379..90efa4c2 100644 --- a/tests/unit/test_extension_dry_run.py +++ b/tests/unit/test_extension_dry_run.py @@ -45,6 +45,7 @@ "scrape_live_comments", "get_page_state", "get_post_metrics", + "scrape_page_clone", ] @@ -98,6 +99,20 @@ def test_extension_live_actions_disabled_by_default(): assert "!EXTENSION_LIVE_ACTIONS_ENABLED" in source +def test_text_post_handler_supports_explicit_page_destination(): + source = _source() + handler = _handler_body(source, "handlePostText") + assert 'targetType === "PAGE" && targetId' in handler + assert "https://www.facebook.com/${targetId}" in handler + + +def test_page_clone_media_uses_exact_host_suffix_allowlist(): + source = _source() + assert "const isAllowedMediaHost" in source + assert 'host.endsWith(".fbcdn.net")' in source + assert "!isAllowedMediaHost(parsed.hostname)" in source + + def test_background_reports_extension_live_guard_state(): background = (EXTENSION_SCRIPT.parent / "background.js").read_text(encoding="utf-8") @@ -167,6 +182,15 @@ def test_router_dispatches_read_only_post_metrics_handler(): assert "result = await handleGetPostMetrics(params);" in body +def test_router_dispatches_page_clone_as_read_only(): + source = _source() + body = _router_body(source) + + assert "async function handleScrapePageClone(" in source + assert 'case "scrape_page_clone":' in body + assert "result = await handleScrapePageClone(params);" in body + + def test_mutating_handlers_check_dry_run_before_dangerous_dom_actions(): source = _source() diff --git a/tests/unit/test_page_clone_contract.py b/tests/unit/test_page_clone_contract.py new file mode 100644 index 00000000..52ea08b5 --- /dev/null +++ b/tests/unit/test_page_clone_contract.py @@ -0,0 +1,125 @@ +import pytest + +from agent.services.page_clone_contract import ( + MAX_CANDIDATES, + MAX_MEDIA_PER_POST, + MAX_POSTS, + canonicalize_page_url, + normalize_page_clone_request, + redact_page_clone_result, + redact_page_clone_task_payload, +) + + +def test_canonicalizes_supported_facebook_page_url(): + assert canonicalize_page_url("https://facebook.com/pages/Acme/123456789") == ( + "https://www.facebook.com/pages/Acme/123456789" + ) + + +@pytest.mark.parametrize( + "url", + [ + "https://facebook.com/groups/123", + "https://facebook.com/watch/?v=123", + "http://facebook.com/acme", + "https://evil.example/acme", + "https://facebook.com/profile.php", + ], +) +def test_rejects_non_page_or_unsafe_urls(url): + with pytest.raises(ValueError): + canonicalize_page_url(url) + + +def test_request_applies_safe_bounds_and_rejects_unknown_fields(): + request = normalize_page_clone_request({"source_url": "https://www.facebook.com/acme"}) + assert request["source_url"] == "https://www.facebook.com/acme" + assert request["max_posts"] == MAX_POSTS + assert request["candidate_limit"] == MAX_CANDIDATES + assert request["max_media_per_post"] == MAX_MEDIA_PER_POST + + with pytest.raises(ValueError, match="max_posts must be between"): + normalize_page_clone_request({"source_url": "https://facebook.com/acme", "max_posts": 9999}) + + with pytest.raises(ValueError): + normalize_page_clone_request({"source_url": "https://facebook.com/acme", "token": "secret"}) + + +def test_redacts_ids_urls_and_tokens_from_durable_result(): + result = redact_page_clone_result( + { + "source_url": "https://www.facebook.com/acme", + "profile": {"id": "123", "name": "Acme"}, + "posts": [ + { + "id": "456", + "permalink": "https://www.facebook.com/acme/posts/456", + "message": "hello", + "media": [{"url": "https://scontent.example/x.jpg", "type": "image"}], + } + ], + "access_token": "secret-token", + } + ) + assert result["source_ref"].startswith("sha256:") + assert result["profile"]["id_hash"].startswith("sha256:") + assert result["posts"][0]["id_hash"].startswith("sha256:") + assert "source_url" not in result + assert "permalink" not in result["posts"][0] + assert "access_token" not in str(result) + assert result["posts"][0]["media"][0]["host"] == "scontent.example" + + +def test_redacts_urls_from_persisted_page_clone_warnings(): + result = redact_page_clone_result({ + "source_url": "https://www.facebook.com/acme", + "warnings": ["media skipped: https://scontent.xx.fbcdn.net/photo.jpg?token=secret"], + "posts": [], + }) + + assert result["warnings"] == ["media skipped: [redacted URL]"] + assert "scontent" not in str(result) + assert "token=secret" not in str(result) + + +def test_redaction_uses_configured_media_dir_for_cached_media(tmp_path, monkeypatch): + from agent.services import page_clone_contract + + monkeypatch.setattr(page_clone_contract, "MEDIA_DIR", str(tmp_path)) + cached = tmp_path / "page-clone-task-0-0.jpg" + cached.write_bytes(b"image") + + result = redact_page_clone_result({ + "source_url": "https://www.facebook.com/acme", + "posts": [{"media": [{ + "url": "https://scontent.xx.fbcdn.net/image.jpg", + "local_path": str(cached), + }]}], + }) + + assert result["posts"][0]["media"][0]["media_path"] == str(cached.resolve()) + + +def test_redacts_terminal_task_payload_source_url(): + redacted = redact_page_clone_task_payload({ + "sourceUrl": "https://www.facebook.com/acme", + "maxPosts": 3, + }) + + assert redacted["sourceRef"].startswith("sha256:") + assert "sourceUrl" not in redacted + assert "facebook.com/acme" not in str(redacted) + + +def test_download_media_is_strict_boolean_opt_in(): + request = normalize_page_clone_request({ + "source_url": "https://facebook.com/acme", + "download_media": True, + }) + assert request["download_media"] is True + with pytest.raises(ValueError, match="download_media must be a boolean"): + normalize_page_clone_request({ + "source_url": "https://facebook.com/acme", + "download_media": "true", + }) diff --git a/tests/unit/test_page_clone_dispatch.py b/tests/unit/test_page_clone_dispatch.py new file mode 100644 index 00000000..56cf2999 --- /dev/null +++ b/tests/unit/test_page_clone_dispatch.py @@ -0,0 +1,164 @@ +import pytest + +from agent.worker import processor + + +@pytest.mark.asyncio +async def test_page_clone_dispatches_as_read_only_with_bounded_contract(monkeypatch): + seen = {} + + class FakeClient: + def page_clone_session_ready(self, fb_uid): + return True + + async def scrape_page_clone(self, **kwargs): + seen.update(kwargs) + return {"success": True, "data": {"posts": []}} + + monkeypatch.setattr(processor, "get_fb_client", lambda: FakeClient()) + result = await processor.WorkerController()._dispatch( + "SCRAPE_PAGE_CLONE", + {"sourceUrl": "https://www.facebook.com/acme", "maxPosts": 25}, + {"id": "t1", "task_type": "SCRAPE_PAGE_CLONE", "account_id": "a1"}, + fb_uid="fb-1", + ) + + assert result["success"] is True + assert seen == { + "source_url": "https://www.facebook.com/acme", + "max_posts": 25, + "max_media_per_post": 10, + "deadline_seconds": 30, + "fb_uid": "fb-1", + "strategy": None, + } + + +@pytest.mark.asyncio +async def test_page_clone_rejects_invalid_source_before_extension_dispatch(monkeypatch): + class UnexpectedClient: + async def scrape_page_clone(self, **kwargs): + raise AssertionError("invalid source reached extension") + + monkeypatch.setattr(processor, "get_fb_client", lambda: UnexpectedClient()) + result = await processor.WorkerController()._dispatch( + "SCRAPE_PAGE_CLONE", + {"sourceUrl": "https://example.com/not-facebook"}, + {"id": "t2", "task_type": "SCRAPE_PAGE_CLONE", "account_id": "a1"}, + fb_uid="fb-1", + ) + assert "Facebook host" in result["error"] + + +@pytest.mark.asyncio +async def test_page_clone_requires_an_exact_logged_in_account_session(monkeypatch): + class UnexpectedClient: + def page_clone_session_ready(self, fb_uid): + return False + + async def scrape_page_clone(self, **kwargs): + raise AssertionError("unbound page clone reached extension") + + monkeypatch.setattr(processor, "get_fb_client", lambda: UnexpectedClient()) + result = await processor.WorkerController()._dispatch( + "SCRAPE_PAGE_CLONE", + {"sourceUrl": "https://facebook.com/acme"}, + {"id": "t-session", "task_type": "SCRAPE_PAGE_CLONE", "account_id": "a1"}, + fb_uid=None, + ) + assert result["code"] == "PAGE_CLONE_SESSION_UNAVAILABLE" + + +@pytest.mark.asyncio +async def test_page_clone_propagates_bounded_media_and_deadline(monkeypatch): + seen = {} + + class FakeClient: + def page_clone_session_ready(self, fb_uid): + return True + + async def scrape_page_clone(self, **kwargs): + seen.update(kwargs) + return {"success": True} + + monkeypatch.setattr(processor, "get_fb_client", lambda: FakeClient()) + await processor.WorkerController()._dispatch( + "SCRAPE_PAGE_CLONE", + { + "sourceUrl": "https://facebook.com/acme", + "maxPosts": 2, + "maxMediaPerPost": 3, + "deadlineSeconds": 5, + }, + {"id": "t3", "task_type": "SCRAPE_PAGE_CLONE", "account_id": "a1"}, + fb_uid="fb-1", + ) + + assert seen["max_posts"] == 2 + assert seen["max_media_per_post"] == 3 + assert seen["deadline_seconds"] == 5 + + +@pytest.mark.asyncio +async def test_page_clone_cancellation_check_preserves_cancelled_state(monkeypatch): + async def cancelled_task(task_id): + return {"id": task_id, "status": "CANCELLED"} + + monkeypatch.setattr(processor.crud, "get_task", cancelled_task) + assert await processor._task_is_cancelled("t-cancelled") is True + + +@pytest.mark.asyncio +async def test_page_clone_opt_in_media_cache_runs_before_evidence_redaction(monkeypatch): + class FakeClient: + def page_clone_session_ready(self, fb_uid): + return True + + async def scrape_page_clone(self, **kwargs): + return {"success": True, "data": {"posts": []}} + + cached = {} + + async def fake_cache(result, task_id): + cached["task_id"] = task_id + result["data"]["cached"] = True + return result + + from agent.services import page_clone_media + + monkeypatch.setattr(processor, "get_fb_client", lambda: FakeClient()) + monkeypatch.setattr(page_clone_media, "cache_page_clone_media", fake_cache) + result = await processor.WorkerController()._dispatch( + "SCRAPE_PAGE_CLONE", + {"sourceUrl": "https://facebook.com/acme", "downloadMedia": True}, + {"id": "task-cache", "task_type": "SCRAPE_PAGE_CLONE", "account_id": "a1"}, + fb_uid="fb-1", + ) + + assert cached["task_id"] == "task-cache" + assert result["data"]["cached"] is True + + +def test_task_schema_allows_page_clone_without_mutation_arm(): + from pathlib import Path + + schema = (Path(__file__).parents[2] / "agent" / "db" / "schema.py").read_text() + assert "'SCRAPE_PAGE_CLONE'" in schema + + +def test_page_clone_result_is_redacted_before_persistence(): + persisted = processor._persistable_result( + "SCRAPE_PAGE_CLONE", + { + "success": True, + "data": { + "source_url": "https://www.facebook.com/acme", + "profile": {"id": "123", "name": "Acme"}, + "posts": [], + "access_token": "must-not-persist", + }, + }, + ) + assert "source_url" not in str(persisted) + assert "must-not-persist" not in str(persisted) + assert persisted["data"]["source_ref"].startswith("sha256:") diff --git a/tests/unit/test_page_clone_draft_queue.py b/tests/unit/test_page_clone_draft_queue.py new file mode 100644 index 00000000..f4484a91 --- /dev/null +++ b/tests/unit/test_page_clone_draft_queue.py @@ -0,0 +1,68 @@ +"""Reviewed Page Clone drafts queue through the existing guarded post pipeline.""" +import asyncio +import json + +import pytest + +from agent.api import posts as posts_api +from agent.db import crud + + +@pytest.fixture +async def draft_account(db_ready): + return await crud.create_account("Draft Queue Account") + + +@pytest.mark.asyncio +async def test_queue_page_clone_draft_uses_dry_run_post_task(draft_account, monkeypatch): + monkeypatch.setattr("agent.config.LIVE_ACTIONS_ENABLED", False, raising=False) + draft = await crud.create_post( + draft_account["id"], + post_type="TEXT", + content="Reviewed clone content", + target_type="PAGE", + target_id="destination-page", + status="DRAFT", + ) + + queued = await posts_api.queue_post(draft["id"]) + + assert queued["post"]["status"] == "SCHEDULED" + assert queued["task"]["task_type"] == "POST_TEXT" + assert queued["task"]["ref_id"] == draft["id"] + payload = json.loads(queued["task"]["payload"]) + assert payload["targetType"] == "PAGE" + assert payload["targetId"] == "destination-page" + assert payload["dryRun"] is True + + with pytest.raises(Exception) as exc_info: + await posts_api.queue_post(draft["id"]) + assert getattr(exc_info.value, "status_code", None) == 409 + + +@pytest.mark.asyncio +async def test_queue_rejects_non_draft_post(draft_account): + post = await crud.create_post(draft_account["id"], content="already queued", status="SCHEDULED") + + with pytest.raises(Exception) as exc_info: + await posts_api.queue_post(post["id"]) + + assert getattr(exc_info.value, "status_code", None) == 409 + + +@pytest.mark.asyncio +async def test_concurrent_queue_attempts_create_only_one_task(draft_account): + draft = await crud.create_post( + draft_account["id"], content="one queue only", status="DRAFT" + ) + + results = await asyncio.gather( + posts_api.queue_post(draft["id"]), + posts_api.queue_post(draft["id"]), + return_exceptions=True, + ) + + assert sum(not isinstance(result, Exception) for result in results) == 1 + assert sum(isinstance(result, Exception) for result in results) == 1 + tasks = await crud.list_tasks(account_id=draft_account["id"]) + assert len(tasks) == 1 diff --git a/tests/unit/test_page_clone_drafts.py b/tests/unit/test_page_clone_drafts.py new file mode 100644 index 00000000..069a66a6 --- /dev/null +++ b/tests/unit/test_page_clone_drafts.py @@ -0,0 +1,170 @@ +"""Operator review converts approved Page Clone evidence into local drafts only.""" +import json +from pathlib import Path + +import pytest + +from agent.api import tasks as tasks_api +from agent.db import crud + + +@pytest.fixture +async def page_clone_draft_account(db_ready): + return await crud.create_account("Page Clone Draft Account") + + +@pytest.mark.asyncio +async def test_page_clone_evidence_creates_selected_local_page_drafts(page_clone_draft_account): + source_task = await crud.create_task( + page_clone_draft_account["id"], + "SCRAPE_PAGE_CLONE", + status="COMPLETED", + result=json.dumps({ + "success": True, + "data": { + "posts": [ + {"message": "First cloned draft"}, + {"message": "Second cloned draft"}, + ], + }, + }), + ) + + drafts = await tasks_api.create_page_clone_drafts( + source_task["id"], + tasks_api.PageCloneDraftCreate( + account_id=page_clone_draft_account["id"], + target_id="destination-page", + selected_post_indexes=[1], + ), + ) + + assert drafts["source_task_id"] == source_task["id"] + assert len(drafts["drafts"]) == 1 + draft = drafts["drafts"][0] + assert draft["status"] == "DRAFT" + assert draft["target_type"] == "PAGE" + assert draft["target_id"] == "destination-page" + assert draft["content"] == "Second cloned draft" + + +@pytest.mark.asyncio +async def test_page_clone_drafts_reject_non_completed_or_cross_account_source(page_clone_draft_account): + source_task = await crud.create_task( + page_clone_draft_account["id"], + "SCRAPE_PAGE_CLONE", result=json.dumps({"data": {"posts": []}}), + ) + + with pytest.raises(Exception) as exc_info: + await tasks_api.create_page_clone_drafts( + source_task["id"], + tasks_api.PageCloneDraftCreate( + account_id=page_clone_draft_account["id"], + target_id="destination-page", + selected_post_indexes=[0], + ), + ) + + assert getattr(exc_info.value, "status_code", None) == 409 + + +@pytest.mark.asyncio +async def test_page_clone_drafts_only_attach_cached_media_inside_media_dir( + page_clone_draft_account, tmp_path, monkeypatch +): + from agent.api import tasks as tasks_module + + media_dir = tmp_path / "media" + media_dir.mkdir() + cached_image = media_dir / "cached.jpg" + cached_image.write_bytes(b"image") + monkeypatch.setattr(tasks_module, "MEDIA_DIR", str(media_dir)) + + source_task = await crud.create_task( + page_clone_draft_account["id"], + "SCRAPE_PAGE_CLONE", + status="COMPLETED", + result=json.dumps({"data": {"posts": [{ + "message": "Image draft", + "media": [ + {"media_path": str(cached_image)}, + {"media_path": str(tmp_path / "outside.jpg")}, + ], + }]}}), + ) + + response = await tasks_api.create_page_clone_drafts( + source_task["id"], + tasks_api.PageCloneDraftCreate( + account_id=page_clone_draft_account["id"], + target_id="destination-page", + selected_post_indexes=[0], + ), + ) + + draft = response["drafts"][0] + assert draft["post_type"] == "IMAGE" + assert json.loads(draft["media_paths"]) == [str(cached_image.resolve())] + + +@pytest.mark.asyncio +async def test_page_clone_drafts_allow_image_only_post(page_clone_draft_account, tmp_path, monkeypatch): + from agent.api import tasks as tasks_module + + media_dir = tmp_path / "media" + media_dir.mkdir() + cached_image = media_dir / "image.jpg" + cached_image.write_bytes(b"image") + monkeypatch.setattr(tasks_module, "MEDIA_DIR", str(media_dir)) + source_task = await crud.create_task( + page_clone_draft_account["id"], + "SCRAPE_PAGE_CLONE", + status="COMPLETED", + result=json.dumps({"data": {"posts": [{ + "message": "", + "media": [{"media_path": str(cached_image)}], + }]}}), + ) + + response = await tasks_api.create_page_clone_drafts( + source_task["id"], + tasks_api.PageCloneDraftCreate( + account_id=page_clone_draft_account["id"], + target_id="destination-page", + selected_post_indexes=[0], + ), + ) + + assert response["drafts"][0]["post_type"] == "IMAGE" + assert response["drafts"][0]["content"] == "" + + +@pytest.mark.asyncio +async def test_page_clone_drafts_create_video_post_for_cached_video(page_clone_draft_account, tmp_path, monkeypatch): + from agent.api import tasks as tasks_module + + media_dir = tmp_path / "media" + media_dir.mkdir() + cached_video = media_dir / "video.mp4" + cached_video.write_bytes(b"video") + monkeypatch.setattr(tasks_module, "MEDIA_DIR", str(media_dir)) + source_task = await crud.create_task( + page_clone_draft_account["id"], + "SCRAPE_PAGE_CLONE", + status="COMPLETED", + result=json.dumps({"data": {"posts": [{ + "message": "Video draft", + "media": [{"media_path": str(cached_video), "type": "video"}], + }]}}), + ) + + response = await tasks_api.create_page_clone_drafts( + source_task["id"], + tasks_api.PageCloneDraftCreate( + account_id=page_clone_draft_account["id"], + target_id="destination-page", + selected_post_indexes=[0], + ), + ) + + assert response["drafts"][0]["post_type"] == "VIDEO" diff --git a/tests/unit/test_page_clone_media.py b/tests/unit/test_page_clone_media.py new file mode 100644 index 00000000..428a11af --- /dev/null +++ b/tests/unit/test_page_clone_media.py @@ -0,0 +1,115 @@ +"""Bounded Page Clone media cache tests use an in-process HTTP transport.""" +import httpx +import pytest + +from agent.services import page_clone_media + + +@pytest.mark.asyncio +async def test_media_cache_accepts_allowlisted_image_and_stays_under_media_dir(tmp_path, monkeypatch): + monkeypatch.setattr(page_clone_media, "MEDIA_DIR", str(tmp_path)) + + def handler(request): + return httpx.Response(200, headers={"content-type": "image/jpeg"}, content=b"\xff\xd8\xffjpeg-bytes") + + real_client = httpx.AsyncClient + monkeypatch.setattr( + page_clone_media.httpx, + "AsyncClient", + lambda **kwargs: real_client(transport=httpx.MockTransport(handler), **kwargs), + ) + result = await page_clone_media.cache_page_clone_media( + {"data": {"posts": [{"media": [{"url": "https://scontent.xx.fbcdn.net/image.jpg"}]}]}}, + "task-media", + ) + + path = result["data"]["posts"][0]["media"][0]["local_path"] + assert path.startswith(str(tmp_path)) + assert open(path, "rb").read() == b"\xff\xd8\xffjpeg-bytes" + + +@pytest.mark.asyncio +async def test_media_cache_rejects_redirects_and_untrusted_hosts(tmp_path, monkeypatch): + monkeypatch.setattr(page_clone_media, "MEDIA_DIR", str(tmp_path)) + + def handler(request): + return httpx.Response(302, headers={"location": "https://evil.example/x"}) + + real_client = httpx.AsyncClient + monkeypatch.setattr( + page_clone_media.httpx, + "AsyncClient", + lambda **kwargs: real_client(transport=httpx.MockTransport(handler), **kwargs), + ) + result = await page_clone_media.cache_page_clone_media( + {"data": {"posts": [{"media": [ + {"url": "https://scontent.xx.fbcdn.net/image.jpg"}, + {"url": "https://evil.example/image.jpg"}, + ]}]}}, + "task-media", + ) + + assert "local_path" not in result["data"]["posts"][0]["media"][0] + assert "local_path" not in result["data"]["posts"][0]["media"][1] + assert len(result["data"]["warnings"]) == 2 + + +@pytest.mark.asyncio +async def test_media_cache_rejects_fake_image_content_type(tmp_path, monkeypatch): + monkeypatch.setattr(page_clone_media, "MEDIA_DIR", str(tmp_path)) + + def handler(request): + return httpx.Response(200, headers={"content-type": "image/jpeg"}, content=b"not-an-image") + + real_client = httpx.AsyncClient + monkeypatch.setattr( + page_clone_media.httpx, + "AsyncClient", + lambda **kwargs: real_client(transport=httpx.MockTransport(handler), **kwargs), + ) + result = await page_clone_media.cache_page_clone_media( + {"data": {"posts": [{"media": [{"url": "https://scontent.xx.fbcdn.net/image.jpg"}]}]}}, + "task-media", + ) + + assert "local_path" not in result["data"]["posts"][0]["media"][0] + assert "does not match" in result["data"]["warnings"][0] + + +@pytest.mark.asyncio +async def test_media_cache_accepts_allowlisted_mp4(tmp_path, monkeypatch): + monkeypatch.setattr(page_clone_media, "MEDIA_DIR", str(tmp_path)) + + def handler(request): + return httpx.Response( + 200, + headers={"content-type": "video/mp4"}, + content=b"\x00\x00\x00\x18ftypisomvideo-bytes", + ) + + real_client = httpx.AsyncClient + monkeypatch.setattr( + page_clone_media.httpx, + "AsyncClient", + lambda **kwargs: real_client(transport=httpx.MockTransport(handler), **kwargs), + ) + result = await page_clone_media.cache_page_clone_media( + {"data": {"posts": [{"media": [{ + "url": "https://scontent.xx.fbcdn.net/video.mp4", "type": "video" + }]}]}}, + "task-video", + ) + + assert result["data"]["posts"][0]["media"][0]["local_path"].endswith(".mp4") + + +def test_media_cleanup_removes_only_matching_task_cache(tmp_path, monkeypatch): + monkeypatch.setattr(page_clone_media, "MEDIA_DIR", str(tmp_path)) + matching = tmp_path / "page-clone-task-media-0-0.jpg" + other = tmp_path / "page-clone-other-task-0-0.jpg" + matching.write_bytes(b"x") + other.write_bytes(b"x") + + assert page_clone_media.cleanup_page_clone_media("task-media") == 1 + assert not matching.exists() + assert other.exists() diff --git a/tests/unit/test_page_clone_schema_migration.py b/tests/unit/test_page_clone_schema_migration.py new file mode 100644 index 00000000..7e10bebc --- /dev/null +++ b/tests/unit/test_page_clone_schema_migration.py @@ -0,0 +1,53 @@ +"""Regression coverage for upgrading existing task tables.""" +import sqlite3 + +import pytest + + +@pytest.mark.asyncio +async def test_init_db_upgrades_legacy_task_check_constraint(tmp_path, monkeypatch): + """Existing queue records survive while SCRAPE_PAGE_CLONE becomes valid.""" + db_path = tmp_path / "legacy.db" + monkeypatch.setenv("DB_PATH", str(db_path)) + + import agent.db.schema as schema_mod + + legacy_schema = schema_mod.SCHEMA.replace("'SCRAPE_PAGE_CLONE',", "") + connection = sqlite3.connect(db_path) + connection.executescript(legacy_schema) + connection.execute("INSERT INTO account (id, name) VALUES ('a1', 'Legacy account')") + connection.execute( + "INSERT INTO task (id, account_id, task_type, payload) VALUES (?, ?, ?, ?)", + ("legacy-task", "a1", "SCRAPE_PROFILE", "{}"), + ) + connection.execute( + "INSERT INTO task_trace (task_id, task_type, status) VALUES (?, ?, ?)", + ("legacy-task", "SCRAPE_PROFILE", "SUCCESS"), + ) + connection.commit() + connection.close() + + schema_mod._db = None + schema_mod.DB_PATH = str(db_path) + await schema_mod.init_db() + db = await schema_mod.get_db() + + row = await (await db.execute("SELECT task_type FROM task WHERE id = 'legacy-task'")).fetchone() + assert row["task_type"] == "SCRAPE_PROFILE" + trace = await (await db.execute( + "SELECT task_id FROM task_trace WHERE task_id = 'legacy-task'" + )).fetchone() + assert trace["task_id"] == "legacy-task" + + await db.execute( + "INSERT INTO task (id, account_id, task_type, payload) VALUES (?, ?, ?, ?)", + ("page-clone-task", "a1", "SCRAPE_PAGE_CLONE", "{}"), + ) + await db.commit() + + task_sql = (await (await db.execute( + "SELECT sql FROM sqlite_master WHERE type = 'table' AND name = 'task'" + )).fetchone())["sql"] + assert "SCRAPE_PAGE_CLONE" in task_sql + + await schema_mod.close_db() diff --git a/tests/unit/test_page_clone_task_api.py b/tests/unit/test_page_clone_task_api.py new file mode 100644 index 00000000..84a99c74 --- /dev/null +++ b/tests/unit/test_page_clone_task_api.py @@ -0,0 +1,85 @@ +"""Page Clone API ingress must enforce the same bounded contract as the worker.""" +import json + +import pytest +from fastapi import HTTPException + +from agent.api import tasks as tasks_api +from agent.db import crud + + +@pytest.fixture +async def page_clone_account(db_ready): + return await crud.create_account("Page Clone API Account") + + +@pytest.mark.asyncio +async def test_page_clone_task_api_normalizes_and_bounds_request(page_clone_account): + task = await tasks_api.create_task( + tasks_api.TaskCreate( + account_id=page_clone_account["id"], + task_type="SCRAPE_PAGE_CLONE", + payload={ + "sourceUrl": "https://m.facebook.com/acme/?ref=share", + "maxPosts": 25, + "candidateLimit": 8, + "maxMediaPerPost": 10, + "deadlineSeconds": 30, + }, + ) + ) + + assert task["task_type"] == "SCRAPE_PAGE_CLONE" + assert json.loads(task["payload"]) == { + "sourceUrl": "https://www.facebook.com/acme", + "maxPosts": 25, + "candidateLimit": 8, + "maxMediaPerPost": 10, + "deadlineSeconds": 30, + "downloadMedia": False, + } + + +@pytest.mark.asyncio +async def test_page_clone_task_api_rejects_over_limit_request(page_clone_account): + with pytest.raises(HTTPException) as exc_info: + await tasks_api.create_task( + tasks_api.TaskCreate( + account_id=page_clone_account["id"], + task_type="SCRAPE_PAGE_CLONE", + payload={"sourceUrl": "https://facebook.com/acme", "maxPosts": 26}, + ) + ) + + assert exc_info.value.status_code == 422 + + +@pytest.mark.asyncio +async def test_page_clone_task_api_rejects_unknown_or_unsafe_payload(page_clone_account): + with pytest.raises(HTTPException) as exc_info: + await tasks_api.create_task( + tasks_api.TaskCreate( + account_id=page_clone_account["id"], + task_type="SCRAPE_PAGE_CLONE", + payload={"sourceUrl": "https://example.com/not-facebook", "token": "nope"}, + ) + ) + + assert exc_info.value.status_code == 422 + + +@pytest.mark.asyncio +async def test_cancelling_page_clone_redacts_queued_source_url(page_clone_account): + task = await tasks_api.create_task( + tasks_api.TaskCreate( + account_id=page_clone_account["id"], + task_type="SCRAPE_PAGE_CLONE", + payload={"sourceUrl": "https://facebook.com/acme"}, + ) + ) + + cancelled = await crud.cancel_task(task["id"]) + payload = json.loads(cancelled["payload"]) + assert cancelled["status"] == "CANCELLED" + assert payload["sourceRef"].startswith("sha256:") + assert "sourceUrl" not in payload diff --git a/tests/unit/test_safety_gate.py b/tests/unit/test_safety_gate.py index 02a91268..c53770fd 100644 --- a/tests/unit/test_safety_gate.py +++ b/tests/unit/test_safety_gate.py @@ -1093,6 +1093,19 @@ def test_group_target_type_requires_group_url(): assert payload["groupUrl"] == "https://facebook.com/groups/test" +def test_page_target_type_requires_destination_target_id(): + from agent.services.safety_gate import enforce_payload + + with pytest.raises(ValueError, match="page targetType requires a non-empty targetId"): + enforce_payload("POST_TEXT", {"targetType": "PAGE"}) + + payload = enforce_payload("POST_TEXT", {"targetType": "PAGE", "targetId": "destination-page"}) + assert payload["targetId"] == "destination-page" + + with pytest.raises(ValueError, match="Facebook page id or slug"): + enforce_payload("POST_TEXT", {"targetType": "PAGE", "targetId": "page/?redirect=evil"}) + + def test_post_target_type_requires_post_url(): from agent.services.safety_gate import enforce_payload with pytest.raises(ValueError, match="post targetType requires a non-empty postUrl"): From 833b08f76f75227107e156884c67a36635df614a Mon Sep 17 00:00:00 2001 From: hthmkt12 Date: Thu, 23 Jul 2026 10:59:28 +0700 Subject: [PATCH 2/3] feat(workflow-lab): add read-only inspection tools --- agent/api/workflows.py | 80 +++++++++++ agent/main.py | 2 + agent/services/workflow_adapters.py | 25 ++++ agent/services/workflow_analyzer.py | 26 ++++ agent/services/workflow_capability.py | 14 ++ agent/services/workflow_contract.py | 133 ++++++++++++++++++ agent/services/workflow_redaction.py | 29 ++++ agent/services/workflow_store.py | 95 +++++++++++++ dashboard/src/App.tsx | 2 + dashboard/src/pages/WorkflowLabPage.test.tsx | 18 +++ dashboard/src/pages/WorkflowLabPage.tsx | 28 ++++ dashboard/src/types/workflows.ts | 19 +++ docs/workflow-lab-handoff.md | 20 +++ docs/workflow-lab-threat-model.md | 14 ++ docs/workflow-lab.md | 16 +++ extension/capture.mjs | 66 +++++++++ extension/lease.mjs | 22 +++ extension/tests/capture.test.mjs | 66 +++++++++ integrations/__init__.py | 1 + integrations/workflow_lab_mcp/__init__.py | 1 + integrations/workflow_lab_mcp/server.py | 54 +++++++ .../workflow_lab_mcp/tests/test_server.py | 30 ++++ .../fixtures/workflow_lab/capture-event.json | 7 + tests/unit/test_workflow_adapters.py | 21 +++ tests/unit/test_workflow_analyzer.py | 13 ++ tests/unit/test_workflow_api.py | 42 ++++++ tests/unit/test_workflow_contract.py | 87 ++++++++++++ tests/unit/test_workflow_e2e.py | 26 ++++ tests/unit/test_workflow_store.py | 32 +++++ 29 files changed, 989 insertions(+) create mode 100644 agent/api/workflows.py create mode 100644 agent/services/workflow_adapters.py create mode 100644 agent/services/workflow_analyzer.py create mode 100644 agent/services/workflow_capability.py create mode 100644 agent/services/workflow_contract.py create mode 100644 agent/services/workflow_redaction.py create mode 100644 agent/services/workflow_store.py create mode 100644 dashboard/src/pages/WorkflowLabPage.test.tsx create mode 100644 dashboard/src/pages/WorkflowLabPage.tsx create mode 100644 dashboard/src/types/workflows.ts create mode 100644 docs/workflow-lab-handoff.md create mode 100644 docs/workflow-lab-threat-model.md create mode 100644 docs/workflow-lab.md create mode 100644 extension/capture.mjs create mode 100644 extension/lease.mjs create mode 100644 extension/tests/capture.test.mjs create mode 100644 integrations/__init__.py create mode 100644 integrations/workflow_lab_mcp/__init__.py create mode 100644 integrations/workflow_lab_mcp/server.py create mode 100644 integrations/workflow_lab_mcp/tests/test_server.py create mode 100644 tests/fixtures/workflow_lab/capture-event.json create mode 100644 tests/unit/test_workflow_adapters.py create mode 100644 tests/unit/test_workflow_analyzer.py create mode 100644 tests/unit/test_workflow_api.py create mode 100644 tests/unit/test_workflow_contract.py create mode 100644 tests/unit/test_workflow_e2e.py create mode 100644 tests/unit/test_workflow_store.py diff --git a/agent/api/workflows.py b/agent/api/workflows.py new file mode 100644 index 00000000..49260980 --- /dev/null +++ b/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} diff --git a/agent/main.py b/agent/main.py index 9c52998b..f8c612a9 100644 --- a/agent/main.py +++ b/agent/main.py @@ -210,6 +210,7 @@ def _extension_ws_api_key(request_path: str) -> str | None: from agent.api.seeding import router as seeding_router from agent.api.spy import router as spy_router from agent.api.strategies import router as strategies_router +from agent.api.workflows import router as workflows_router api_dependencies = [Depends(require_api_key)] @@ -221,6 +222,7 @@ def _extension_ws_api_key(request_path: str) -> str | None: app.include_router(seeding_router, prefix="/api", dependencies=api_dependencies) app.include_router(spy_router, prefix="/api", dependencies=api_dependencies) app.include_router(strategies_router, prefix="/api", dependencies=api_dependencies) +app.include_router(workflows_router, prefix="/api", dependencies=api_dependencies) # ─── Root & Status ─────────────────────────────────────────── diff --git a/agent/services/workflow_adapters.py b/agent/services/workflow_adapters.py new file mode 100644 index 00000000..2b006f50 --- /dev/null +++ b/agent/services/workflow_adapters.py @@ -0,0 +1,25 @@ +"""Source-controlled, read-only adapter registry for Workflow Lab.""" + +from __future__ import annotations + +from .workflow_capability import WorkflowReadOnlyCapabilityGate + + +class WorkflowAdapterRegistry: + def __init__(self, gate: WorkflowReadOnlyCapabilityGate | None = None) -> None: + if gate is not None and type(gate) is not WorkflowReadOnlyCapabilityGate: + raise TypeError("Workflow Lab uses the sealed capability gate") + self._gate = gate or WorkflowReadOnlyCapabilityGate() + + def list_adapters(self) -> list[str]: + return ["get_post_metrics", "read_page_clone"] + + def inspect(self, command: str, evidence: dict) -> dict: + self._gate.require(command, "inspect") + if not isinstance(evidence, dict): + raise ValueError("adapter evidence must be an object") + mode = "DOM_FALLBACK" if command == "get_post_metrics" else "SCRAPE_PAGE_CLONE" + return {"schemaVersion": 1, "command": command, "mode": mode, "readOnly": True} + + def execute(self, command: str, payload: dict) -> None: + self._gate.require(command, "execute") diff --git a/agent/services/workflow_analyzer.py b/agent/services/workflow_analyzer.py new file mode 100644 index 00000000..be2a7d9b --- /dev/null +++ b/agent/services/workflow_analyzer.py @@ -0,0 +1,26 @@ +"""Analysis-only replayability classifier; it never constructs or sends requests.""" + +from __future__ import annotations + +from .workflow_contract import Replayability + + +def analyze_replayability(events: list[dict]) -> dict: + if not isinstance(events, list) or len(events) > 1000: + raise ValueError("events must be a bounded list") + if not events: + mode = Replayability.NON_REPLAYABLE + elif any(event.get("method") in {"POST", "PUT", "PATCH", "DELETE"} for event in events if isinstance(event, dict)): + mode = Replayability.BROWSER_SESSION_REQUIRED + elif any(event.get("resourceType") == "Document" for event in events if isinstance(event, dict)): + mode = Replayability.DOM_FALLBACK + else: + mode = Replayability.OBSERVED_REQUEST_CANDIDATE + return { + "schemaVersion": 1, + "replayability": mode.value, + "readOnly": True, + "executeAllowed": False, + "reason": "metadata-only analysis; no response body or executor", + "eventCount": len(events), + } diff --git a/agent/services/workflow_capability.py b/agent/services/workflow_capability.py new file mode 100644 index 00000000..d0a7dbe6 --- /dev/null +++ b/agent/services/workflow_capability.py @@ -0,0 +1,14 @@ +"""Sealed capability boundary for read-only Workflow Lab adapters.""" + +from __future__ import annotations + + +class WorkflowReadOnlyCapabilityGate: + _ALLOW = frozenset({("get_post_metrics", "inspect"), ("read_page_clone", "inspect")}) + + def allows(self, adapter: str, operation: str) -> bool: + return (adapter, operation) in self._ALLOW + + def require(self, adapter: str, operation: str) -> None: + if not self.allows(adapter, operation): + raise PermissionError("Workflow Lab capability is read-only and sealed") diff --git a/agent/services/workflow_contract.py b/agent/services/workflow_contract.py new file mode 100644 index 00000000..811a63a5 --- /dev/null +++ b/agent/services/workflow_contract.py @@ -0,0 +1,133 @@ +"""Positive-schema contracts for the local, read-only Workflow Lab.""" + +from __future__ import annotations + +from dataclasses import dataclass +from enum import StrEnum +from urllib.parse import parse_qsl, urlsplit + + +_EVENT_KEYS = {"method", "url", "status", "resourceType", "timingMs", "responseBody"} +_DRAFT_KEYS = {"name", "adapter", "sourceCaptureId", "steps", "ttlSeconds", "evidenceRefs"} +_ADAPTERS = {"get_post_metrics", "read_page_clone"} + + +class Replayability(StrEnum): + DOM_FALLBACK = "DOM_FALLBACK" + OBSERVED_REQUEST_CANDIDATE = "OBSERVED_REQUEST_CANDIDATE" + TOKEN_REFRESH_REQUIRED = "TOKEN_REFRESH_REQUIRED" + BROWSER_SESSION_REQUIRED = "BROWSER_SESSION_REQUIRED" + NON_REPLAYABLE = "NON_REPLAYABLE" + MUTATION_BLOCKED = "MUTATION_BLOCKED" + + +@dataclass(frozen=True) +class CaptureEvent: + capture_id: str + method: str + host: str + path: str + status: int | None + resource_type: str + timing_ms: float | None + query_shape: list[str] + value_aliases: dict[str, str] + + +@dataclass(frozen=True) +class WorkflowDraft: + name: str + adapter: str + source_capture_id: str + read_only: bool = True + schema_version: int = 1 + steps: tuple[str, ...] = () + ttl_seconds: int = 3600 + evidence_refs: tuple[str, ...] = () + + +@dataclass(frozen=True) +class WorkflowEnvelope: + schema_version: int + risk: str + replayability: Replayability + ttl_seconds: int + evidence_refs: tuple[str, ...] + + +def normalize_capture_event(payload: dict, capture_id: str) -> CaptureEvent: + if not isinstance(payload, dict) or not isinstance(capture_id, str) or not capture_id.strip(): + raise ValueError("capture event and capture_id are required") + unknown = set(payload) - _EVENT_KEYS + if unknown: + raise ValueError("unsupported capture fields: " + ", ".join(sorted(unknown))) + if "responseBody" in payload: + raise ValueError("response body is not accepted") + method = payload.get("method", "GET") + if not isinstance(method, str) or method.upper() not in {"GET", "POST", "PUT", "PATCH", "DELETE", "HEAD", "OPTIONS"}: + raise ValueError("unsupported HTTP method") + raw_url = payload.get("url") + if not isinstance(raw_url, str) or len(raw_url) > 4096: + raise ValueError("url must be a bounded string") + parsed = urlsplit(raw_url) + if parsed.scheme not in {"http", "https"} or not parsed.hostname: + raise ValueError("url must be an absolute HTTP URL") + if parsed.username or parsed.password or parsed.fragment: + raise ValueError("url contains credentials or fragment") + host = parsed.hostname.lower().rstrip(".") + if not (host == "facebook.com" or host.endswith(".facebook.com") or host == "fbcdn.net" or host.endswith(".fbcdn.net")): + raise ValueError("url must use an approved host") + query_shape = tuple(sorted({key[:128] for key, _ in parse_qsl(parsed.query, keep_blank_values=True)})) + aliases = {key: "v1" for key in query_shape} + status = payload.get("status") + if status is not None and (isinstance(status, bool) or not isinstance(status, int) or not 100 <= status <= 599): + raise ValueError("status must be an HTTP status") + timing = payload.get("timingMs") + if timing is not None and (isinstance(timing, bool) or not isinstance(timing, (int, float)) or timing < 0 or timing > 600000): + raise ValueError("timingMs must be bounded") + resource_type = payload.get("resourceType", "Other") + if not isinstance(resource_type, str): + raise ValueError("resourceType must be a string") + return CaptureEvent( + capture_id=capture_id.strip()[:128], method=method.upper(), host=host, + path=(parsed.path or "/")[:2048], status=status, + resource_type=resource_type[:64], + timing_ms=float(timing) if timing is not None else None, + query_shape=list(query_shape), value_aliases=aliases, + ) + + +def normalize_workflow_draft(payload: dict) -> WorkflowDraft: + if not isinstance(payload, dict): + raise ValueError("workflow draft must be an object") + unknown = set(payload) - _DRAFT_KEYS + if unknown: + raise ValueError("unsupported workflow draft fields: " + ", ".join(sorted(unknown))) + name, adapter, capture_id = (payload.get("name"), payload.get("adapter"), payload.get("sourceCaptureId")) + if not all(isinstance(value, str) and value.strip() for value in (name, adapter, capture_id)): + raise ValueError("name, adapter, and sourceCaptureId are required") + if adapter not in _ADAPTERS: + raise ValueError("unsupported workflow adapter") + steps = payload.get("steps", []) + refs = payload.get("evidenceRefs", []) + ttl = payload.get("ttlSeconds", 3600) + if not isinstance(steps, list) or len(steps) > 32 or any(not isinstance(item, str) or not item.strip() for item in steps): + raise ValueError("steps must be a bounded list of strings") + if not isinstance(refs, list) or len(refs) > 64 or any(not isinstance(item, str) or not item.strip() for item in refs): + raise ValueError("evidenceRefs must be a bounded list of strings") + if isinstance(ttl, bool) or not isinstance(ttl, int) or ttl < 60 or ttl > 86400: + raise ValueError("ttlSeconds must be between 60 and 86400") + return WorkflowDraft(name=name.strip()[:128], adapter=adapter, source_capture_id=capture_id.strip()[:128], + steps=tuple(item.strip()[:128] for item in steps), ttl_seconds=ttl, + evidence_refs=tuple(item.strip()[:128] for item in refs)) + + +def make_workflow_envelope(*, replayability: Replayability, ttl_seconds: int = 3600, + evidence_refs: list[str] | tuple[str, ...] = ()) -> WorkflowEnvelope: + if not isinstance(replayability, Replayability): + raise ValueError("replayability must use the Workflow Lab enum") + if isinstance(ttl_seconds, bool) or not 60 <= ttl_seconds <= 86400: + raise ValueError("ttlSeconds must be between 60 and 86400") + if not isinstance(evidence_refs, (list, tuple)) or len(evidence_refs) > 64 or any(not isinstance(ref, str) or not ref.strip() for ref in evidence_refs): + raise ValueError("evidenceRefs must be bounded opaque strings") + return WorkflowEnvelope(1, "READ_ONLY", replayability, ttl_seconds, tuple(ref.strip()[:128] for ref in evidence_refs)) diff --git a/agent/services/workflow_redaction.py b/agent/services/workflow_redaction.py new file mode 100644 index 00000000..e2d41a85 --- /dev/null +++ b/agent/services/workflow_redaction.py @@ -0,0 +1,29 @@ +"""Secret-safe serialization for Workflow Lab evidence.""" + +from __future__ import annotations + +from .workflow_contract import normalize_capture_event +import hashlib + + +def _safe_query_key(key: str) -> str: + lowered = key.lower() + if any(marker in lowered for marker in ("token", "cookie", "auth", "pass", "secret", "session", "jwt", "fb_dtsg", "lsd", "c_user")): + return "key_sha256:" + hashlib.sha256(key.encode("utf-8")).hexdigest()[:16] + return key[:64] + + +def redact_capture_event(payload: dict, capture_id: str) -> dict: + event = normalize_capture_event(payload, capture_id) + return { + "schemaVersion": 1, + "captureId": event.capture_id, + "method": event.method, + "host": event.host, + "path": event.path, + "status": event.status, + "resourceType": event.resource_type, + "timingMs": event.timing_ms, + "queryShape": [_safe_query_key(key) for key in event.query_shape], + "valueAliases": {_safe_query_key(key): value for key, value in event.value_aliases.items()}, + } diff --git a/agent/services/workflow_store.py b/agent/services/workflow_store.py new file mode 100644 index 00000000..0f3f16bf --- /dev/null +++ b/agent/services/workflow_store.py @@ -0,0 +1,95 @@ +"""Dedicated local Workflow Lab SQLite store; never shares task tables.""" + +from __future__ import annotations + +import json +import sqlite3 +import time +from pathlib import Path + + +class WorkflowStore: + def __init__(self, path: str | Path = "runtime/workflow_lab.sqlite3", ttl_seconds: int = 3600) -> None: + self.path = str(path) + self.ttl_seconds = max(60, min(int(ttl_seconds), 86400)) + Path(self.path).parent.mkdir(parents=True, exist_ok=True) + with self._connect() as db: + db.execute("CREATE TABLE IF NOT EXISTS workflow_captures (id TEXT PRIMARY KEY, profile_id TEXT NOT NULL, fb_uid TEXT, tab_id TEXT, status TEXT NOT NULL, created_at REAL NOT NULL, updated_at REAL NOT NULL)") + db.execute("CREATE TABLE IF NOT EXISTS workflow_events (capture_id TEXT NOT NULL, seq INTEGER NOT NULL, payload TEXT NOT NULL, PRIMARY KEY(capture_id, seq))") + + def _connect(self): + db = sqlite3.connect(self.path) + db.row_factory = sqlite3.Row + return db + + def create_capture(self, capture_id: str, profile_id: str, fb_uid: str | None = None, tab_id: str | None = None) -> dict: + now = time.time() + with self._connect() as db: + if not capture_id.strip() or len(capture_id) > 128 or not profile_id.strip() or len(profile_id) > 128: + raise ValueError("capture and profile identifiers must be bounded") + if fb_uid is not None and (not fb_uid.strip() or len(fb_uid) > 128): + raise ValueError("fb_uid must be bounded") + if tab_id is not None and (not tab_id.strip() or len(tab_id) > 128): + raise ValueError("tab_id must be bounded") + db.execute("INSERT INTO workflow_captures VALUES (?, ?, ?, ?, 'running', ?, ?)", (capture_id.strip(), profile_id.strip(), fb_uid, tab_id, now, now)) + return self.inspect_capture(capture_id) + + def append_event(self, capture_id: str, payload: dict, sequence: int | None = None) -> None: + if not isinstance(payload, dict) or len(json.dumps(payload, separators=(",", ":"))) > 16384: + raise ValueError("workflow event must be bounded JSON") + if any(key.lower() in {"body", "responsebody", "headers", "cookie", "authorization", "postdata"} for key in payload): + raise ValueError("workflow event contains forbidden secret-bearing fields") + with self._connect() as db: + row = db.execute("SELECT status FROM workflow_captures WHERE id = ?", (capture_id,)).fetchone() + if not row or row["status"] != "running": + raise ValueError("capture is not running") + count = db.execute("SELECT COUNT(*) FROM workflow_events WHERE capture_id = ?", (capture_id,)).fetchone()[0] + if count >= 1000: + raise ValueError("capture event quota exceeded") + seq = count if sequence is None else sequence + if not isinstance(seq, int) or seq < 0 or seq > 1000000: + raise ValueError("event sequence must be bounded") + existing = db.execute("SELECT payload FROM workflow_events WHERE capture_id = ? AND seq = ?", (capture_id, seq)).fetchone() + if existing: + if existing[0] != json.dumps(payload, separators=(",", ":")): + raise ValueError("duplicate sequence has different payload") + return + db.execute("INSERT INTO workflow_events VALUES (?, ?, ?)", (capture_id, seq, json.dumps(payload, separators=(",", ":")))) + db.execute("UPDATE workflow_captures SET updated_at = ? WHERE id = ?", (time.time(), capture_id)) + + def list_captures(self, profile_id: str | None = None) -> list[dict]: + self.gc() + with self._connect() as db: + if profile_id: + rows = db.execute("SELECT id FROM workflow_captures WHERE profile_id = ? ORDER BY created_at DESC", (profile_id,)) + else: + rows = db.execute("SELECT id FROM workflow_captures ORDER BY created_at DESC") + return [self.inspect_capture(row[0]) for row in rows] + + def gc(self) -> int: + cutoff = time.time() - self.ttl_seconds + with self._connect() as db: + ids = [row[0] for row in db.execute("SELECT id FROM workflow_captures WHERE updated_at < ?", (cutoff,))] + for capture_id in ids: + db.execute("DELETE FROM workflow_events WHERE capture_id = ?", (capture_id,)) + db.execute("DELETE FROM workflow_captures WHERE id = ?", (capture_id,)) + return len(ids) + + def stop_capture(self, capture_id: str) -> dict | None: + with self._connect() as db: + db.execute("UPDATE workflow_captures SET status = 'stopped', updated_at = ? WHERE id = ?", (time.time(), capture_id)) + return self.inspect_capture(capture_id) + + def inspect_capture(self, capture_id: str) -> dict | None: + with self._connect() as db: + row = db.execute("SELECT * FROM workflow_captures WHERE id = ?", (capture_id,)).fetchone() + if not row: + return None + events = [json.loads(item[0]) for item in db.execute("SELECT payload FROM workflow_events WHERE capture_id = ? ORDER BY seq", (capture_id,))] + return {"id": row["id"], "profileId": row["profile_id"], "fbUid": row["fb_uid"], "tabId": row["tab_id"], "status": row["status"], "events": events} + + def delete_capture(self, capture_id: str) -> bool: + with self._connect() as db: + db.execute("DELETE FROM workflow_events WHERE capture_id = ?", (capture_id,)) + changed = db.execute("DELETE FROM workflow_captures WHERE id = ?", (capture_id,)).rowcount + return changed > 0 diff --git a/dashboard/src/App.tsx b/dashboard/src/App.tsx index 1fc188aa..521df417 100644 --- a/dashboard/src/App.tsx +++ b/dashboard/src/App.tsx @@ -23,6 +23,7 @@ import AutoPostFanpagePage from './pages/AutoPostFanpagePage' import AgentOnboardingPage from './pages/AgentOnboardingPage' import ProjectsPage from './pages/ProjectsPage' import CampaignsPage from './pages/CampaignsPage' +import WorkflowLabPage from './pages/WorkflowLabPage' type DemoNavItem = { to: string @@ -42,6 +43,7 @@ const NAV: DemoNavItem[] = [ { to: '/accounts', icon: Users, label: 'Local Profiles', exact: false, element: }, { to: '/tasks', icon: FlaskConical, label: 'Local Tasks', exact: false, element: }, { to: '/reports', icon: BarChart3, label: 'Evidence Log', exact: false, element: }, + { to: '/workflow-lab', icon: FlaskConical, label: 'Workflow Lab', exact: false, element: }, { to: '/seeding', icon: Zap, label: 'Seeding (Coming Soon)', exact: false, element: , future: true }, { to: '/comments', icon: MessageCircle, label: 'Comments (Coming Soon)', exact: false, element: , future: true }, { to: '/inbox', icon: Inbox, label: 'Inbox (Coming Soon)', exact: false, element: , future: true }, diff --git a/dashboard/src/pages/WorkflowLabPage.test.tsx b/dashboard/src/pages/WorkflowLabPage.test.tsx new file mode 100644 index 00000000..018489d0 --- /dev/null +++ b/dashboard/src/pages/WorkflowLabPage.test.tsx @@ -0,0 +1,18 @@ +import { render, screen } from '@testing-library/react' +import { describe, expect, it } from 'vitest' +import WorkflowLabPage from './WorkflowLabPage' + +describe('WorkflowLabPage', () => { + it('shows safe read-only empty state', () => { + render() + expect(screen.getByText('No captures available.')).toBeTruthy() + expect(screen.getByLabelText('read-only status')).toBeTruthy() + expect(screen.queryByRole('button')).toBeNull() + }) + it('renders bounded sanitized metadata without links', () => { + render() + expect(screen.getByText(/DOM_FALLBACK/)).toBeTruthy() + expect(screen.queryAllByRole('link')).toHaveLength(0) + expect(screen.getByText('/api/:segment')).toBeTruthy() + }) +}) diff --git a/dashboard/src/pages/WorkflowLabPage.tsx b/dashboard/src/pages/WorkflowLabPage.tsx new file mode 100644 index 00000000..e711ee71 --- /dev/null +++ b/dashboard/src/pages/WorkflowLabPage.tsx @@ -0,0 +1,28 @@ +import type { WorkflowAnalysis, WorkflowEvent } from '../types/workflows' + +type Props = { events?: WorkflowEvent[]; analysis?: WorkflowAnalysis; loading?: boolean; error?: string } + +export default function WorkflowLabPage({ events = [], analysis, loading = false, error }: Props) { + if (loading) return
Loading Workflow Lab…
+ if (error) return
Unable to load Workflow Lab: {error}
+ return ( +
+
+
+

Workflow Lab

+

Local metadata inspection · read-only · no replay

+
+ READ_ONLY +
+ {analysis &&
+
Replayability: {analysis.replayability}
+
{analysis.eventCount} sanitized events · execution disabled
+
} + {!events.length ?

No captures available.

:
+ + {events.slice(0, 100).map((event, index) => )} +
MethodHostPath shapeStatus
{event.method}{event.host}{event.path}{event.status ?? '—'}
+
} +
+ ) +} diff --git a/dashboard/src/types/workflows.ts b/dashboard/src/types/workflows.ts new file mode 100644 index 00000000..2c6d0852 --- /dev/null +++ b/dashboard/src/types/workflows.ts @@ -0,0 +1,19 @@ +export type WorkflowEvent = { + captureId: string + method: string + host: string + path: string + status: number | null + resourceType: string + timingMs: number | null + queryShape: string[] +} + +export type WorkflowAnalysis = { + captureId: string + schemaVersion: number + replayability: string + readOnly: boolean + executeAllowed: false + eventCount: number +} diff --git a/docs/workflow-lab-handoff.md b/docs/workflow-lab-handoff.md new file mode 100644 index 00000000..cc7668fe --- /dev/null +++ b/docs/workflow-lab-handoff.md @@ -0,0 +1,20 @@ +# Workflow Lab handoff state + +## Verified locally + +- Versioned read-only contracts, redaction, replayability analyzer, adapter registry, isolated SQLite store and profile-scoped inspect API. +- Pure extension capture/controller and exclusive lease primitives with Node tests. +- Read-only dashboard route and disabled-by-default local MCP. +- Synthetic capture-to-MCP proof is secret-free; full Python suite passed with `PYTHONUTF8=1`. + +## Explicitly not claimed + +- No real Facebook HAR is imported or uploaded. +- No response-body capture, arbitrary HTTP replay, mutation, adapter promotion, or Cloud MCP exposure exists. +- `lease.mjs` is not yet wired into the existing `background.js` upload/debugger lifecycle. +- Chrome lifecycle evidence (debugger conflict, tab close, logout/UID drift, worker wake reconciliation) is still manual and requires the Page Clone P4 shared-seam handoff. +- Root ZooPost MCP verifier remains environment-blocked by the optional `mcp` package dependency. + +## Safe rollback + +Disable the capture flag/key, stop active controllers, revoke local capture credentials, release leases, then run Workflow Lab TTL GC. Remove only the isolated Workflow Lab modules and runtime database; preserve unrelated pilot/Page Clone changes. diff --git a/docs/workflow-lab-threat-model.md b/docs/workflow-lab-threat-model.md new file mode 100644 index 00000000..fbb1b4dc --- /dev/null +++ b/docs/workflow-lab-threat-model.md @@ -0,0 +1,14 @@ +# Workflow Lab threat model + +Workflow Lab is local-only and read-only. Its durable evidence uses a positive schema: method, host, path shape, status, resource type, bounded timing, query-key shape, and capture-scoped value aliases. It never accepts or persists response bodies, headers, cookies, bearer tokens, POST bodies, URL fragments, or arbitrary CDP payloads. + +The sealed capability gate permits only `inspect` for the reviewed `post_metrics` and `page_reader` adapters. There is no generic request executor, replay promotion, mutation operation, or user-controlled URL fetch path. Capture authentication is independent from the existing local agent transport and must be bound to the active profile and Facebook identity before a capture is accepted. + +Retention is bounded by TTL, per-capture quotas, and explicit stop/delete controls. Redaction is applied before persistence, dashboard display, MCP exposure, or export. Any uncertainty fails closed and leaves the capture as analysis-only evidence. + +| Boundary | Owner | Mitigation | Detection | Rollback | +| --- | --- | --- | --- | --- | +| Browser capture | extension capture module | debugger lease, dedicated nonce, metadata-only schema | capture audit and quota errors | stop capture, release lease, revoke nonce | +| Local persistence | WorkflowStore | isolated SQLite, TTL/GC, profile binding | startup/periodic GC and integrity checks | delete Workflow Lab store only | +| Adapter execution | sealed capability gate | `get_post_metrics`/`read_page_clone` inspect-only allowlist | denied-operation audit | unregister adapter and clear drafts | +| Export/MCP | local integration | opaque IDs, recursive redaction, off by default | secret-canary tests | disable integration and rotate local key | diff --git a/docs/workflow-lab.md b/docs/workflow-lab.md new file mode 100644 index 00000000..b0ef4ea8 --- /dev/null +++ b/docs/workflow-lab.md @@ -0,0 +1,16 @@ +# FBKit Workflow Lab + +Workflow Lab is a local, read-only inspection surface for attended capture metadata. It stores bounded positive-schema evidence in a dedicated SQLite store, classifies replayability, and exposes only reviewed adapters (`get_post_metrics` and `read_page_clone`) in inspect mode. + +The capture layer never stores response bodies, headers, cookies, post bodies, query values, credentials, or arbitrary URLs. The dashboard and local MCP render opaque IDs and sanitized metadata only. MCP is disabled unless `WORKFLOW_LAB_MCP_ENABLED=1` (or `true`/`yes`) and requires a profile-scoped capture key. + +Operational rollback is fail-closed: stop captures, revoke capture keys/nonces, release debugger leases, then run bounded GC. The debugger lease wiring into the existing service worker remains deferred until the Page Clone shared-seam handoff; no claim of synchronous cleanup is made during worker suspension. + +## Verification + +- `PYTHONUTF8=1 .venv\\Scripts\\python.exe -m pytest -q`: 655 passed. +- `node --test extension\\tests\\capture.test.mjs`: 6 passed. +- Dashboard: 61 tests passed, lint passed, build passed. +- Local Workflow Lab MCP: 2 tests passed. +- Synthetic capture → store → analyzer → adapter proof: 1 test passed with no fixture secret, query value, response body, or credential in serialized evidence. +- Root ZooPost MCP verifier: **16 passed**, import check OK after installing the declared `integrations/zoopost-mcp-readonly/requirements-runtime.txt` dependencies into the nested venv. diff --git a/extension/capture.mjs b/extension/capture.mjs new file mode 100644 index 00000000..6d3f452d --- /dev/null +++ b/extension/capture.mjs @@ -0,0 +1,66 @@ +const APPROVED = (host) => (host === 'facebook.com' || host.endsWith('.facebook.com')) && !host.includes('graph.facebook.com') && !host.endsWith('.fbcdn.net') && host !== 'fbcdn.net'; +const FORBIDDEN_PATHS = new Set(['/login', '/login/identify', '/checkpoint']); +const SENSITIVE = ['token', 'cookie', 'auth', 'pass', 'secret', 'session', 'jwt', 'fb_dtsg', 'lsd', 'c_user']; + +function safeKey(key) { + const lower = key.toLowerCase(); + const digest = [...key].reduce((hash, char) => ((hash ^ char.charCodeAt(0)) * 16777619) >>> 0, 2166136261).toString(16).padStart(8, '0'); + return SENSITIVE.some((part) => lower.includes(part)) + ? `key_hash:${digest}` + : key.slice(0, 64); +} + +export function sanitizeNetworkEvent(input, scope) { + if (!input || typeof input !== 'object' || !scope?.captureId || !scope?.profileId) throw new Error('invalid capture scope'); + if (Object.hasOwn(input, 'responseBody') || Object.hasOwn(input, 'requestHeaders') || Object.hasOwn(input, 'postData')) throw new Error('secret-bearing fields rejected'); + const parsed = new URL(String(input.url || '')); + if (!['http:', 'https:'].includes(parsed.protocol) || parsed.username || parsed.password || parsed.hash) throw new Error('unsafe URL'); + const host = parsed.hostname.toLowerCase().replace(/\.$/, ''); + if (!APPROVED(host) || host.includes('fbcdn')) throw new Error('host not approved'); + const decodedPath = decodeURIComponent(parsed.pathname || '/'); + const path = decodedPath.split('/').map((part) => { + if (!part) return ''; + if (/^\d+$/.test(part) || part.length > 48 || /token|auth|secret|cookie|session|dtsg|lsd/i.test(part)) return ':segment'; + return part; + }).join('/'); + if (FORBIDDEN_PATHS.has(path) || path.startsWith('/login/') || path.startsWith('/checkpoint')) throw new Error('auth path excluded'); + const queryShape = [...new Set([...parsed.searchParams.keys()].sort())].map(safeKey); + const valueAliases = Object.fromEntries(queryShape.map((key) => [key, 'v1'])); + const method = String(input.method || 'GET').toUpperCase(); + if (!['GET', 'POST', 'PUT', 'PATCH', 'DELETE', 'HEAD', 'OPTIONS'].includes(method)) throw new Error('method not allowed'); + const status = input.status == null ? null : Number(input.status); + const timingMs = input.timingMs == null ? null : Number(input.timingMs); + if (status != null && (!Number.isInteger(status) || status < 100 || status > 599)) throw new Error('status invalid'); + if (timingMs != null && (!Number.isFinite(timingMs) || timingMs < 0 || timingMs > 600000)) throw new Error('timing invalid'); + return { schemaVersion: 1, captureId: String(scope.captureId).slice(0, 128), method, host, path: path.slice(0, 2048), status, resourceType: String(input.resourceType || 'Other').slice(0, 64), timingMs, queryShape, valueAliases }; +} + +export function createCaptureController({ captureId, profileId, fbUid = null, maxEvents = 1000, maxBytes = 1024 * 1024, ttlMs = 15 * 60 * 1000, now = () => Date.now() }) { + if (!Number.isInteger(maxEvents) || maxEvents < 1 || !Number.isInteger(maxBytes) || maxBytes < 1) throw new Error('capture quotas must be positive integers'); + let stopped = false; + let reason = null; + const expiresAt = now() + Math.min(Math.max(ttlMs, 1000), 24 * 60 * 60 * 1000); + let events = 0; + let bytes = 0; + return { + push(input) { + if (stopped) throw new Error('capture is stopped'); + if (now() >= expiresAt) { stopped = true; reason = 'ttl'; throw new Error('capture TTL expired'); } + if (events >= maxEvents) throw new Error('event quota exceeded'); + const event = sanitizeNetworkEvent(input, { captureId, profileId }); + const size = JSON.stringify(event).length; + if (bytes + size > maxBytes) throw new Error('byte quota exceeded'); + events += 1; bytes += size; + return event; + }, + stop(value = 'manual') { stopped = true; reason = value; }, + reconcile({ currentFbUid = fbUid, debuggerAttached = true } = {}) { + if (!debuggerAttached) { stopped = true; reason = 'debugger_detached'; } + else if (fbUid !== null && currentFbUid !== fbUid) { stopped = true; reason = 'uid_changed'; } + else if (now() >= expiresAt) { stopped = true; reason = 'ttl'; } + return { status: stopped ? 'stopped' : 'running', reason }; + }, + status() { return stopped ? 'stopped' : 'running'; }, + stopReason() { return reason; }, + }; +} diff --git a/extension/lease.mjs b/extension/lease.mjs new file mode 100644 index 00000000..0eb944b8 --- /dev/null +++ b/extension/lease.mjs @@ -0,0 +1,22 @@ +export function createDebuggerLeaseManager() { + const leases = new Map(); + return { + acquire(tabId, owner) { + if (leases.has(tabId)) throw new Error('debugger lease busy'); + const lease = { tabId, owner, generation: 1, active: true }; + leases.set(tabId, lease); + return { ...lease }; + }, + release(tabId, owner) { + const lease = leases.get(tabId); + if (!lease || lease.owner !== owner) return false; + leases.delete(tabId); + return true; + }, + reconcile(tabId, owner) { + const lease = leases.get(tabId); + if (!lease || lease.owner !== owner) return { active: false, reason: 'stale_or_missing' }; + return { active: true, generation: lease.generation }; + }, + }; +} diff --git a/extension/tests/capture.test.mjs b/extension/tests/capture.test.mjs new file mode 100644 index 00000000..9c32a530 --- /dev/null +++ b/extension/tests/capture.test.mjs @@ -0,0 +1,66 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import { sanitizeNetworkEvent, createCaptureController } from '../capture.mjs'; +import { createDebuggerLeaseManager } from '../lease.mjs'; + +test('sanitizes allowlisted Facebook metadata and drops secrets', () => { + const value = sanitizeNetworkEvent({ + method: 'POST', url: 'https://www.facebook.com/api/post?token=secret', + status: 200, resourceType: 'XHR', timingMs: 12, + }, { captureId: 'cap-1', profileId: 'profile-1' }); + assert.deepEqual(value, { + schemaVersion: 1, captureId: 'cap-1', method: 'POST', host: 'www.facebook.com', + path: '/api/post', status: 200, resourceType: 'XHR', timingMs: 12, + queryShape: ['key_hash:5a88237a'], valueAliases: { 'key_hash:5a88237a': 'v1' } + }); +}); + +test('rejects non-Facebook, login and response-body events', () => { + for (const event of [ + { url: 'https://example.com/a' }, + { url: 'https://graph.facebook.com/v1/12345678901234567890' }, + { url: 'https://fbcdn.facebook.com/image/1' }, + { url: 'https://www.facebook.com/login/identify' }, + { url: 'https://www.facebook.com/a', responseBody: 'secret' } + ]) assert.throws(() => sanitizeNetworkEvent(event, { captureId: 'c', profileId: 'p' })); +}); + +test('normalizes numeric and opaque path segments', () => { + const value = sanitizeNetworkEvent({ url: 'https://www.facebook.com/api/12345678901234567890/very-long-opaque-token-value-that-must-not-persist' }, { captureId: 'c', profileId: 'p' }); + assert.equal(value.path, '/api/:segment/:segment'); +}); + +test('normalizes encoded sensitive path segments', () => { + const value = sanitizeNetworkEvent({ url: 'https://www.facebook.com/api/%74oken-value' }, { captureId: 'c', profileId: 'p' }); + assert.equal(value.path, '/api/:segment'); +}); + +test('controller enforces event and byte caps and stop state', () => { + const controller = createCaptureController({ captureId: 'c', profileId: 'p', maxEvents: 1 }); + controller.push({ method: 'GET', url: 'https://www.facebook.com/a' }); + assert.throws(() => controller.push({ method: 'GET', url: 'https://www.facebook.com/b' }), /event quota/); + controller.stop('ttl'); + assert.equal(controller.status(), 'stopped'); + assert.throws(() => controller.push({ method: 'GET', url: 'https://www.facebook.com/c' }), /stopped/); +}); + +test('controller fails closed on TTL, UID drift, and debugger detach', () => { + let clock = 1000; + const controller = createCaptureController({ captureId: 'c', profileId: 'p', fbUid: 'uid-1', ttlMs: 1000, now: () => clock }); + assert.equal(controller.reconcile({ currentFbUid: 'uid-1', debuggerAttached: true }).status, 'running'); + assert.equal(controller.reconcile({ currentFbUid: 'uid-2' }).reason, 'uid_changed'); + const detached = createCaptureController({ captureId: 'c2', profileId: 'p', ttlMs: 1000, now: () => clock }); + assert.equal(detached.reconcile({ debuggerAttached: false }).reason, 'debugger_detached'); + const expired = createCaptureController({ captureId: 'c3', profileId: 'p', ttlMs: 1000, now: () => clock }); + clock = 2500; + assert.equal(expired.reconcile().reason, 'ttl'); +}); + +test('debugger lease is exclusive and owner-bound', () => { + const leases = createDebuggerLeaseManager(); + leases.acquire(7, 'capture'); + assert.throws(() => leases.acquire(7, 'upload'), /busy/); + assert.deepEqual(leases.reconcile(7, 'upload'), { active: false, reason: 'stale_or_missing' }); + assert.equal(leases.release(7, 'upload'), false); + assert.equal(leases.release(7, 'capture'), true); +}); diff --git a/integrations/__init__.py b/integrations/__init__.py new file mode 100644 index 00000000..a6cfad95 --- /dev/null +++ b/integrations/__init__.py @@ -0,0 +1 @@ +"""Local integrations namespace.""" diff --git a/integrations/workflow_lab_mcp/__init__.py b/integrations/workflow_lab_mcp/__init__.py new file mode 100644 index 00000000..8a4f5c31 --- /dev/null +++ b/integrations/workflow_lab_mcp/__init__.py @@ -0,0 +1 @@ +"""Disabled-by-default local Workflow Lab MCP.""" diff --git a/integrations/workflow_lab_mcp/server.py b/integrations/workflow_lab_mcp/server.py new file mode 100644 index 00000000..a4af7313 --- /dev/null +++ b/integrations/workflow_lab_mcp/server.py @@ -0,0 +1,54 @@ +"""Local-only Workflow Lab MCP facade. Disabled unless WORKFLOW_LAB_MCP_ENABLED=1.""" + +from __future__ import annotations + +import os +from typing import Any + +from agent.services.workflow_analyzer import analyze_replayability +from agent.services.workflow_store import WorkflowStore + + +TOOLS = ("list_captures", "inspect_workflow", "explain_request", "compare_runs", "check_replayability") +_FORBIDDEN = {"cookie", "body", "headers", "authorization", "execute", "promote", "url"} +_ARGUMENTS = {"list_captures": set(), "inspect_workflow": {"captureId"}, "explain_request": {"captureId"}, "check_replayability": {"captureId"}, "compare_runs": {"captureId", "otherCaptureId"}} + + +class WorkflowLabMCP: + def __init__(self, store: WorkflowStore | None = None) -> None: + self.enabled = os.environ.get("WORKFLOW_LAB_MCP_ENABLED", "").lower() in {"1", "true", "yes"} + self.store = store or WorkflowStore() + + def call(self, tool: str, args: dict[str, Any], profile_id: str) -> dict[str, Any]: + if not self.enabled: + raise PermissionError("Workflow Lab MCP is disabled") + if not isinstance(profile_id, str) or not profile_id.strip() or len(profile_id) > 128: + raise ValueError("bounded profile_id is required") + if tool not in TOOLS or not isinstance(args, dict) or set(args) - _ARGUMENTS[tool] or any(key.lower() in _FORBIDDEN for key in args): + raise ValueError("unsupported or unsafe MCP operation") + capture_id = args.get("captureId") + if tool == "list_captures": + return {"captures": self.store.list_captures(profile_id)[:100]} + if not isinstance(capture_id, str) or not capture_id.strip() or len(capture_id) > 128: + raise ValueError("opaque captureId is required") + capture = self.store.inspect_capture(capture_id) + if not capture or capture["profileId"] != profile_id: + raise LookupError("capture not found") + if tool == "check_replayability": + return analyze_replayability(capture["events"]) + if tool == "compare_runs": + return {"observed": True, "captureId": capture_id, "comparison": "unavailable"} + if tool == "explain_request": + return {"observed": True, "captureId": capture_id, "explanation": "metadata-only request shape"} + safe_events = [] + for event in capture["events"][:100]: + if not isinstance(event, dict): + continue + safe_events.append({key: event[key] for key in ("schemaVersion", "captureId", "method", "host", "path", "status", "resourceType", "timingMs", "queryShape", "valueAliases") if key in event}) + return {"id": capture["id"], "status": capture["status"], "events": safe_events, "readOnly": True} + + +def main() -> None: + if os.environ.get("WORKFLOW_LAB_MCP_ENABLED", "").lower() not in {"1", "true", "yes"}: + raise SystemExit("Workflow Lab MCP disabled; set WORKFLOW_LAB_MCP_ENABLED=1 for local stdio use") + raise SystemExit("stdio launcher is intentionally not auto-started by FBKit") diff --git a/integrations/workflow_lab_mcp/tests/test_server.py b/integrations/workflow_lab_mcp/tests/test_server.py new file mode 100644 index 00000000..7dd76ba9 --- /dev/null +++ b/integrations/workflow_lab_mcp/tests/test_server.py @@ -0,0 +1,30 @@ +import pytest + +from agent.services.workflow_store import WorkflowStore +from integrations.workflow_lab_mcp.server import WorkflowLabMCP + + +def test_mcp_is_off_by_default_and_read_only(tmp_path, monkeypatch): + monkeypatch.delenv("WORKFLOW_LAB_MCP_ENABLED", raising=False) + mcp = WorkflowLabMCP(WorkflowStore(tmp_path / "workflow.sqlite3")) + with pytest.raises(PermissionError): + mcp.call("list_captures", {"captureId": "cap"}, "profile") + + +def test_mcp_scopes_and_rejects_unsafe_tools(tmp_path, monkeypatch): + monkeypatch.setenv("WORKFLOW_LAB_MCP_ENABLED", "1") + store = WorkflowStore(tmp_path / "workflow.sqlite3") + store.create_capture("cap", "profile") + mcp = WorkflowLabMCP(store) + assert len(mcp.call("list_captures", {}, "profile")["captures"]) == 1 + assert mcp.call("check_replayability", {"captureId": "cap"}, "profile")["readOnly"] is True + with pytest.raises(LookupError): + mcp.call("inspect_workflow", {"captureId": "cap"}, "other") + with pytest.raises(ValueError): + mcp.call("execute", {"captureId": "cap"}, "profile") + with pytest.raises(ValueError): + mcp.call("inspect_workflow", {"captureId": "cap", "cookie": "secret"}, "profile") + with pytest.raises(ValueError): + mcp.call("inspect_workflow", {"captureId": "cap", "unexpected": True}, "profile") + with pytest.raises(ValueError): + mcp.call("inspect_workflow", {"captureId": "cap"}, "") diff --git a/tests/fixtures/workflow_lab/capture-event.json b/tests/fixtures/workflow_lab/capture-event.json new file mode 100644 index 00000000..d0127db1 --- /dev/null +++ b/tests/fixtures/workflow_lab/capture-event.json @@ -0,0 +1,7 @@ +{ + "method": "GET", + "url": "https://www.facebook.com/api/post?access_token=fixture-secret", + "status": 200, + "resourceType": "XHR", + "timingMs": 18 +} diff --git a/tests/unit/test_workflow_adapters.py b/tests/unit/test_workflow_adapters.py new file mode 100644 index 00000000..26da1ad4 --- /dev/null +++ b/tests/unit/test_workflow_adapters.py @@ -0,0 +1,21 @@ +import pytest + +from agent.services.workflow_adapters import WorkflowAdapterRegistry + + +def test_registry_exposes_only_reviewed_read_only_adapters(): + registry = WorkflowAdapterRegistry() + assert registry.list_adapters() == ["get_post_metrics", "read_page_clone"] + assert registry.inspect("get_post_metrics", {"source": "dom"})["mode"] == "DOM_FALLBACK" + assert registry.inspect("read_page_clone", {"source": "page"})["mode"] == "SCRAPE_PAGE_CLONE" + with pytest.raises(PermissionError): + registry.execute("get_post_metrics", {}) + + +def test_registry_rejects_injected_capability_bypass(): + class AllowEverything: + def require(self, *_args): + return None + + with pytest.raises(TypeError, match="sealed"): + WorkflowAdapterRegistry(AllowEverything()) diff --git a/tests/unit/test_workflow_analyzer.py b/tests/unit/test_workflow_analyzer.py new file mode 100644 index 00000000..863ca9ca --- /dev/null +++ b/tests/unit/test_workflow_analyzer.py @@ -0,0 +1,13 @@ +from agent.services.workflow_analyzer import analyze_replayability + + +def test_analyzer_is_analysis_only_and_blocks_replay(): + result = analyze_replayability([{"method": "POST", "resourceType": "XHR"}]) + assert result["replayability"] == "BROWSER_SESSION_REQUIRED" + assert result["executeAllowed"] is False + assert result["readOnly"] is True + + +def test_analyzer_classifies_dom_and_empty_capture(): + assert analyze_replayability([{"method": "GET", "resourceType": "Document"}])["replayability"] == "DOM_FALLBACK" + assert analyze_replayability([])["replayability"] == "NON_REPLAYABLE" diff --git a/tests/unit/test_workflow_api.py b/tests/unit/test_workflow_api.py new file mode 100644 index 00000000..c9ad5e94 --- /dev/null +++ b/tests/unit/test_workflow_api.py @@ -0,0 +1,42 @@ +import pytest +from fastapi import HTTPException + +from agent.api import workflows + + +@pytest.mark.asyncio +async def test_workflow_api_is_inspect_only(tmp_path, monkeypatch): + monkeypatch.setenv("WORKFLOW_LAB_API_KEY", "test-key") + workflows._store = workflows.WorkflowStore(tmp_path / "workflow_lab.sqlite3") + workflows._store.create_capture("cap-api", "profile-1") + inspected = await workflows.inspect_workflow_capture("cap-api", "profile-1", "test-key") + assert inspected["status"] == "running" + analysis = await workflows.analyze_workflow_capture("cap-api", "profile-1", "test-key") + assert analysis["executeAllowed"] is False + with pytest.raises(HTTPException) as exc: + await workflows.review_workflow_capture("cap-api", workflows.WorkflowReview(decision="accept"), "profile-1", "test-key") + assert exc.value.status_code == 409 + stopped = await workflows.stop_workflow_capture("cap-api", "profile-1", "test-key") + assert stopped["status"] == "stopped" + assert (await workflows.delete_workflow_capture("cap-api", "profile-1", "test-key"))["logicalRetention"] is True + + +@pytest.mark.asyncio +async def test_workflow_api_has_no_start_or_ingest_route(): + paths = {route.path for route in workflows.router.routes} + assert not any(route.endswith("/start") or route.endswith("/events") for route in paths) + + +@pytest.mark.asyncio +async def test_workflow_api_wrong_profile_cannot_stop_or_delete(tmp_path, monkeypatch): + monkeypatch.setenv("WORKFLOW_LAB_API_KEY", "test-key") + workflows._store = workflows.WorkflowStore(tmp_path / "workflow_lab.sqlite3") + workflows._store.create_capture("cap-scope", "owner") + with pytest.raises(HTTPException) as stopped: + await workflows.stop_workflow_capture("cap-scope", "attacker", "test-key") + assert stopped.value.status_code == 404 + assert workflows._store.inspect_capture("cap-scope")["status"] == "running" + with pytest.raises(HTTPException) as deleted: + await workflows.delete_workflow_capture("cap-scope", "attacker", "test-key") + assert deleted.value.status_code == 404 + assert workflows._store.inspect_capture("cap-scope") is not None diff --git a/tests/unit/test_workflow_contract.py b/tests/unit/test_workflow_contract.py new file mode 100644 index 00000000..4a180324 --- /dev/null +++ b/tests/unit/test_workflow_contract.py @@ -0,0 +1,87 @@ +import pytest + +from agent.services.workflow_contract import ( + CaptureEvent, + WorkflowDraft, + Replayability, + make_workflow_envelope, + normalize_capture_event, + normalize_workflow_draft, +) +from agent.services.workflow_redaction import redact_capture_event +from agent.services.workflow_capability import WorkflowReadOnlyCapabilityGate + + +def test_capture_event_keeps_metadata_only_and_aliases_values(): + event = normalize_capture_event( + { + "method": "POST", + "url": "https://www.facebook.com/api?token=secret", + "status": 200, + "resourceType": "XHR", + "timingMs": 12.5, + }, + capture_id="cap-1", + ) + assert isinstance(event, CaptureEvent) + assert event.method == "POST" + assert event.host == "www.facebook.com" + assert event.path == "/api" + assert event.query_shape == ["token"] + assert event.value_aliases == {"token": "v1"} + assert not hasattr(event, "requestHeaders") + + +def test_capture_event_rejects_raw_body_and_unbounded_input(): + with pytest.raises(ValueError, match="response body"): + normalize_capture_event({"method": "GET", "url": "https://example.com", "responseBody": "x"}, "cap") + with pytest.raises(ValueError, match="unsupported capture fields"): + normalize_capture_event({"method": "GET", "url": "https://www.facebook.com", "requestHeaders": {}}, "cap") + with pytest.raises(ValueError, match="approved host"): + normalize_capture_event({"method": "GET", "url": "https://example.com"}, "cap") + with pytest.raises(ValueError, match="resourceType"): + normalize_capture_event({"method": "GET", "url": "https://www.facebook.com", "resourceType": {"secret": "x"}}, "cap") + with pytest.raises(ValueError, match="credentials or fragment"): + normalize_capture_event({"method": "GET", "url": "https://user:pass@www.facebook.com/a#frag"}, "cap") + + +def test_workflow_draft_is_positive_schema_and_read_only(): + draft = normalize_workflow_draft( + {"name": "post metrics", "adapter": "get_post_metrics", "sourceCaptureId": "cap-1"} + ) + assert isinstance(draft, WorkflowDraft) + assert draft.adapter == "get_post_metrics" + assert draft.read_only is True + with pytest.raises(ValueError, match="unsupported"): + normalize_workflow_draft({"name": "x", "adapter": "get_post_metrics", "execute": True}) + + +def test_redaction_is_recursive_and_secret_free(): + value = redact_capture_event( + {"method": "GET", "url": "https://www.facebook.com/a?token=secret", "status": 200}, + capture_id="cap-1", + ) + assert value["captureId"] == "cap-1" + assert value["queryShape"][0].startswith("key_sha256:") + assert "secret" not in repr(value) + assert "url" not in value + + +def test_capability_gate_has_sealed_read_only_allowlist(): + gate = WorkflowReadOnlyCapabilityGate() + assert gate.allows("get_post_metrics", "inspect") + assert gate.allows("read_page_clone", "inspect") + assert not gate.allows("get_post_metrics", "execute") + assert not gate.allows("unknown", "inspect") + with pytest.raises(PermissionError): + gate.require("get_post_metrics", "execute") + + +def test_versioned_read_only_envelope_is_bounded(): + envelope = make_workflow_envelope( + replayability=Replayability.OBSERVED_REQUEST_CANDIDATE, + evidence_refs=["cap-1"], + ) + assert envelope.schema_version == 1 + assert envelope.risk == "READ_ONLY" + assert envelope.evidence_refs == ("cap-1",) diff --git a/tests/unit/test_workflow_e2e.py b/tests/unit/test_workflow_e2e.py new file mode 100644 index 00000000..4cfcdd50 --- /dev/null +++ b/tests/unit/test_workflow_e2e.py @@ -0,0 +1,26 @@ +import json + +from agent.services.workflow_adapters import WorkflowAdapterRegistry +from agent.services.workflow_analyzer import analyze_replayability +from agent.services.workflow_redaction import redact_capture_event +from agent.services.workflow_store import WorkflowStore + + +def test_synthetic_capture_store_analyzer_adapter_chain_is_secret_free(tmp_path, monkeypatch): + store = WorkflowStore(tmp_path / "workflow_lab.sqlite3") + store.create_capture("cap-e2e", "profile-e2e", fb_uid="uid-e2e", tab_id="tab-e2e") + event = redact_capture_event( + {"method": "GET", "url": "https://www.facebook.com/api/post?access_token=fixture-secret", "status": 200, "resourceType": "XHR"}, + "cap-e2e", + ) + store.append_event("cap-e2e", event) + capture = store.inspect_capture("cap-e2e") + analysis = analyze_replayability(capture["events"]) + adapter = WorkflowAdapterRegistry().inspect("get_post_metrics", capture["events"][0]) + assert analysis["executeAllowed"] is False + assert adapter["readOnly"] is True + serialized = json.dumps({"capture": capture, "analysis": analysis, "adapter": adapter}) + assert "fixture-secret" not in serialized + assert "access_token" not in serialized + assert "responseBody" not in serialized + assert "tab-e2e" in serialized diff --git a/tests/unit/test_workflow_store.py b/tests/unit/test_workflow_store.py new file mode 100644 index 00000000..5054289a --- /dev/null +++ b/tests/unit/test_workflow_store.py @@ -0,0 +1,32 @@ +from agent.services.workflow_store import WorkflowStore + + +def test_store_is_local_ttl_bounded_and_reviewable(tmp_path): + store = WorkflowStore(tmp_path / "workflow_lab.sqlite3", ttl_seconds=60) + store.create_capture("cap-1", "profile-1") + store.append_event("cap-1", {"schemaVersion": 1, "method": "GET", "host": "example.com", "path": "/"}) + assert store.inspect_capture("cap-1")["events"][0]["host"] == "example.com" + assert store.stop_capture("cap-1")["status"] == "stopped" + assert store.delete_capture("cap-1") is True + assert store.inspect_capture("cap-1") is None + + +def test_store_sequence_is_idempotent_and_restart_recoverable(tmp_path): + path = tmp_path / "workflow_lab.sqlite3" + store = WorkflowStore(path) + store.create_capture("cap-2", "profile-2") + event = {"schemaVersion": 1, "method": "GET", "host": "www.facebook.com", "path": "/"} + store.append_event("cap-2", event, sequence=4) + store.append_event("cap-2", event, sequence=4) + restarted = WorkflowStore(path) + assert len(restarted.inspect_capture("cap-2")["events"]) == 1 + + +def test_store_gc_removes_expired_capture_only(tmp_path): + path = tmp_path / "workflow_lab.sqlite3" + store = WorkflowStore(path, ttl_seconds=60) + store.create_capture("cap-old", "profile") + with store._connect() as db: + db.execute("UPDATE workflow_captures SET updated_at = 0 WHERE id = 'cap-old'") + assert store.gc() == 1 + assert store.inspect_capture("cap-old") is None From de233a65489280807911e4d0c4482fe3a557d2d0 Mon Sep 17 00:00:00 2001 From: hthmkt12 Date: Sat, 25 Jul 2026 15:32:31 +0700 Subject: [PATCH 3/3] fix(dashboard): add pilot readiness and safety visibility --- .../src/components/SafetyGateStatus.test.tsx | 17 ++ dashboard/src/components/SafetyGateStatus.tsx | 32 ++++ .../components/pilot-readiness-strip.test.tsx | 110 ++++++++++++ .../src/components/pilot-readiness-strip.tsx | 166 ++++++++++++++++++ .../src/components/pilot-readiness.test.ts | 26 +++ dashboard/src/components/pilot-readiness.ts | 20 +++ dashboard/src/pages/DashboardPage.tsx | 46 ++++- dashboard/src/pages/LogsPage.test.tsx | 23 +++ dashboard/src/pages/LogsPage.tsx | 21 ++- .../src/router-security-contract.test.ts | 86 +++++++++ dashboard/src/types/index.ts | 1 + dashboard/tsconfig.app.json | 2 +- 12 files changed, 543 insertions(+), 7 deletions(-) create mode 100644 dashboard/src/components/pilot-readiness-strip.test.tsx create mode 100644 dashboard/src/components/pilot-readiness-strip.tsx create mode 100644 dashboard/src/components/pilot-readiness.test.ts create mode 100644 dashboard/src/components/pilot-readiness.ts create mode 100644 dashboard/src/router-security-contract.test.ts diff --git a/dashboard/src/components/SafetyGateStatus.test.tsx b/dashboard/src/components/SafetyGateStatus.test.tsx index 995ba65d..62464f97 100644 --- a/dashboard/src/components/SafetyGateStatus.test.tsx +++ b/dashboard/src/components/SafetyGateStatus.test.tsx @@ -46,4 +46,21 @@ describe('SafetyGateStatus', () => { expect(screen.getByText(/live auth ready: not ready/i)).toBeTruthy() expect(screen.getByText(/active live arms: none/i)).toBeTruthy() }) + + it('shows live block reasons in protected dry-run mode', () => { + render() + + expect(screen.getByText(/Live is blocked because/i)).toBeTruthy() + expect(screen.getByText(/LIVE_ACTIONS_ENABLED=false/i)).toBeTruthy() + expect(screen.getByText(/DRY_RUN_DEFAULT=true/i)).toBeTruthy() + expect(screen.getByText(/APPROVAL_REQUIRED=true/i)).toBeTruthy() + expect(screen.getByText(/Live auth not ready/i)).toBeTruthy() + expect(screen.getByText(/live-mode-readiness/i)).toBeTruthy() + }) + + it('does not show block reasons when live is enabled (not protected)', () => { + render() + + expect(screen.queryByText(/Live is blocked because/i)).toBeNull() + }) }) diff --git a/dashboard/src/components/SafetyGateStatus.tsx b/dashboard/src/components/SafetyGateStatus.tsx index 031ec00e..a8374675 100644 --- a/dashboard/src/components/SafetyGateStatus.tsx +++ b/dashboard/src/components/SafetyGateStatus.tsx @@ -42,6 +42,17 @@ function InfoPill({ label, value, safe }: { label: string; value: string; safe: ) } +/** Human-readable reasons why live actions are blocked in the current phase. */ +function liveBlockReasons(liveEnabled: boolean, dryRunDefault: boolean, approvalRequired: boolean, liveAuthReady: boolean): string[] { + const reasons: string[] = [] + if (!liveEnabled) reasons.push('LIVE_ACTIONS_ENABLED=false - live mutation disabled by phase policy') + if (dryRunDefault) reasons.push('DRY_RUN_DEFAULT=true - all tasks forced to dry-run') + if (approvalRequired) reasons.push('APPROVAL_REQUIRED=true - manual approval gate is active') + if (!liveAuthReady) reasons.push('Live auth not ready - API/WS auth and live arm prerequisites not met') + reasons.push('Live mode requires docs/live-mode-readiness.md checklist completion + operator approval') + return reasons +} + export default function SafetyGateStatus({ status }: SafetyGateStatusProps) { const safety = status?.safety_gate const safetyKnown = Boolean(safety) @@ -85,6 +96,8 @@ export default function SafetyGateStatus({ status }: SafetyGateStatusProps) { ) } + const blockReasons = liveBlockReasons(liveEnabled, dryRunDefault, approvalRequired, liveAuthReady) + return (
+ + {protectedMode && ( +
+ Live is blocked because: +
    + {blockReasons.map((reason, i) => ( +
  • {reason}
    • + ))} +
+
+ )} ) } diff --git a/dashboard/src/components/pilot-readiness-strip.test.tsx b/dashboard/src/components/pilot-readiness-strip.test.tsx new file mode 100644 index 00000000..6b2b664b --- /dev/null +++ b/dashboard/src/components/pilot-readiness-strip.test.tsx @@ -0,0 +1,110 @@ +import { cleanup, render, screen } from '@testing-library/react' +import { afterEach, describe, expect, it } from 'vitest' +import { PilotReadinessStrip } from './pilot-readiness-strip' + +describe('PilotReadinessStrip', () => { + afterEach(() => { + cleanup() + }) + + it('shows all-pass state when every check passes', () => { + render( + , + ) + + expect(screen.getByText('All checks pass')).toBeTruthy() + expect(screen.getByText('Cloud reachable')).toBeTruthy() + expect(screen.getByText('ZooPost Cloud online')).toBeTruthy() + expect(screen.getByText('FB session logged-in')).toBeTruthy() + expect(screen.getByText('fb_uid detected')).toBeTruthy() + expect(screen.getByText('Agent publish-dry-run')).toBeTruthy() + expect(screen.getByText('capability reported')).toBeTruthy() + expect(screen.getByText('Live disabled')).toBeTruthy() + expect(screen.getByText('dry-run phase enforced')).toBeTruthy() + expect(screen.getByText('2 selectable channel(s)')).toBeTruthy() + }) + + it('shows action required when a critical check fails', () => { + render( + , + ) + + expect(screen.getByText('Action required')).toBeTruthy() + expect(screen.getByText('FBKit offline')).toBeTruthy() + expect(screen.getByText('no logged-in profile')).toBeTruthy() + }) + + it('shows STOP warning when live is not disabled', () => { + render( + , + ) + + expect(screen.getByText('Action required')).toBeTruthy() + expect(screen.getByText('LIVE IS ENABLED - STOP')).toBeTruthy() + }) + + it('shows needs attention when only warn-level checks fail', () => { + render( + , + ) + + expect(screen.getByText('Needs attention')).toBeTruthy() + expect(screen.getByText('evidence stale or missing')).toBeTruthy() + expect(screen.getByText('0 selectable channel(s)')).toBeTruthy() + }) + + it('does not pass when safety status is unavailable', () => { + render( + , + ) + + expect(screen.getByText('Needs attention')).toBeTruthy() + expect(screen.getByText('safety status unavailable - do not proceed')).toBeTruthy() + expect(screen.getByText('evidence freshness unavailable')).toBeTruthy() + }) +}) diff --git a/dashboard/src/components/pilot-readiness-strip.tsx b/dashboard/src/components/pilot-readiness-strip.tsx new file mode 100644 index 00000000..03aad10a --- /dev/null +++ b/dashboard/src/components/pilot-readiness-strip.tsx @@ -0,0 +1,166 @@ +import { CheckCircle2, CircleAlert, Radio } from 'lucide-react' +import type { CSSProperties, ReactNode } from 'react' + +type Props = { + cloudReachable: boolean + fbkitReachable: boolean + dashboardReachable: boolean + fbSessionLoggedIn: boolean + agentHasPublishDryRun: boolean + liveDisabled: boolean | null + evidenceFresh: boolean | null + selectableChannels: number +} + +type CheckStatus = 'pass' | 'fail' | 'warn' | 'skip' + +function checkStatus(ok: boolean | null): CheckStatus { + if (ok === null) return 'skip' + return ok ? 'pass' : 'fail' +} + +export function PilotReadinessStrip(props: Props) { + const checks: Array<{ label: string; status: CheckStatus; detail: string }> = [ + { label: 'Cloud reachable', status: checkStatus(props.cloudReachable), detail: props.cloudReachable ? 'ZooPost Cloud online' : 'Cloud offline' }, + { label: 'FBKit reachable', status: checkStatus(props.fbkitReachable), detail: props.fbkitReachable ? 'FBKit agent online' : 'FBKit offline' }, + { label: 'Dashboard reachable', status: checkStatus(props.dashboardReachable), detail: props.dashboardReachable ? 'Dashboard serving' : 'Dashboard offline' }, + { label: 'FB session logged-in', status: checkStatus(props.fbSessionLoggedIn), detail: props.fbSessionLoggedIn ? 'fb_uid detected' : 'no logged-in profile' }, + { label: 'Agent publish-dry-run', status: checkStatus(props.agentHasPublishDryRun), detail: props.agentHasPublishDryRun ? 'capability reported' : 'capability missing' }, + { label: 'Live disabled', status: props.liveDisabled === null ? 'warn' : props.liveDisabled ? 'pass' : 'fail', detail: props.liveDisabled === null ? 'safety status unavailable - do not proceed' : props.liveDisabled ? 'dry-run phase enforced' : 'LIVE IS ENABLED - STOP' }, + { label: 'Evidence fresh', status: props.evidenceFresh === null ? 'warn' : props.evidenceFresh ? 'pass' : 'warn', detail: props.evidenceFresh === null ? 'evidence freshness unavailable' : props.evidenceFresh ? 'evidence report current' : 'evidence stale or missing' }, + { label: 'Channel readiness', status: props.selectableChannels > 0 ? 'pass' : 'warn', detail: `${props.selectableChannels} selectable channel(s)` }, + ] + + const allPass = checks.every(c => c.status === 'pass') + const hasFail = checks.some(c => c.status === 'fail') + + return ( +
+
+
+
Pilot Readiness
+
Cloud, FBKit, Dashboard, FB session, agent capability, live block, evidence, channels.
+
+ + {allPass ? : } + {allPass ? 'All checks pass' : hasFail ? 'Action required' : 'Needs attention'} + +
+
+ {checks.map(check => ( + + ))} +
+
+ ) +} + +function ReadinessItem({ status, label, detail }: { status: CheckStatus; label: string; detail: string }) { + const color = statusColor(status) + const icon = statusIcon(status) + return ( +
+
{icon}
+
+
{label}
+
{detail}
+
+
+ ) +} + +function statusColor(status: CheckStatus): string { + switch (status) { + case 'pass': return '#16a34a' + case 'fail': return '#dc2626' + case 'warn': return '#d97706' + case 'skip': return 'var(--muted)' + } +} + +function statusIcon(status: CheckStatus): ReactNode { + switch (status) { + case 'pass': return + case 'fail': return + case 'warn': return + case 'skip': return + } +} + +function panelStyle(allPass: boolean, hasFail: boolean): CSSProperties { + const color = allPass ? '#16a34a' : hasFail ? '#dc2626' : '#d97706' + return { + background: `${color}10`, + border: `1px solid ${color}44`, + borderRadius: '14px', + padding: '16px', + } +} + +const headerStyle: CSSProperties = { + display: 'flex', + alignItems: 'center', + justifyContent: 'space-between', + gap: '12px', + flexWrap: 'wrap', + marginBottom: '12px', +} + +const titleStyle: CSSProperties = { + fontSize: '16px', + fontWeight: 850, +} + +const subtitleStyle: CSSProperties = { + color: 'var(--muted)', + fontSize: '12px', + marginTop: '3px', +} + +function statusBadgeStyle(allPass: boolean, hasFail: boolean): CSSProperties { + const color = allPass ? '#16a34a' : hasFail ? '#dc2626' : '#d97706' + return { + display: 'inline-flex', + alignItems: 'center', + gap: '7px', + border: `1px solid ${color}55`, + background: `${color}16`, + color, + borderRadius: '999px', + padding: '7px 10px', + fontSize: '12px', + fontWeight: 850, + whiteSpace: 'nowrap', + } +} + +const gridStyle: CSSProperties = { + display: 'grid', + gridTemplateColumns: 'repeat(auto-fit, minmax(160px, 1fr))', + gap: '10px', +} + +function itemStyle(status: CheckStatus): CSSProperties { + const color = statusColor(status) + return { + display: 'flex', + alignItems: 'center', + gap: '10px', + border: `1px solid ${color}33`, + background: 'var(--card)', + borderRadius: '10px', + padding: '10px 12px', + minHeight: '62px', + } +} + +const itemTitleStyle: CSSProperties = { + fontWeight: 850, + fontSize: '12px', +} + +const detailStyle: CSSProperties = { + color: 'var(--muted)', + fontSize: '11px', + marginTop: '2px', +} diff --git a/dashboard/src/components/pilot-readiness.test.ts b/dashboard/src/components/pilot-readiness.test.ts new file mode 100644 index 00000000..7e0158fb --- /dev/null +++ b/dashboard/src/components/pilot-readiness.test.ts @@ -0,0 +1,26 @@ +import { describe, expect, it } from 'vitest' +import { EVIDENCE_FRESHNESS_MS, isEvidenceFresh } from './pilot-readiness' + +const now = Date.parse('2026-07-25T12:00:00.000Z') + +describe('isEvidenceFresh', () => { + it('accepts activity evidence created within the freshness window', () => { + expect(isEvidenceFresh([ + { id: 'event-1', type: 'job.evidence', severity: 'info', message: 'ready', target_id: null, created_at: '2026-07-25T09:00:01.000Z' }, + ], now)).toBe(true) + }) + + it('rejects stale or timestamp-less activity evidence', () => { + expect(isEvidenceFresh([ + { id: 'event-1', type: 'job.evidence', severity: 'info', message: 'old', target_id: null, created_at: '2026-07-25T07:59:59.000Z' }, + ], now)).toBe(false) + expect(isEvidenceFresh([ + { id: 'event-2', type: 'job.evidence', severity: 'info', message: 'unknown', target_id: null, created_at: null }, + ], now)).toBe(false) + }) + + it('reports unknown when the performance request did not return', () => { + expect(isEvidenceFresh(undefined, now)).toBeNull() + expect(isEvidenceFresh([], now + EVIDENCE_FRESHNESS_MS)).toBe(false) + }) +}) diff --git a/dashboard/src/components/pilot-readiness.ts b/dashboard/src/components/pilot-readiness.ts new file mode 100644 index 00000000..3c6d3246 --- /dev/null +++ b/dashboard/src/components/pilot-readiness.ts @@ -0,0 +1,20 @@ +import type { DashboardPerformance } from '../types' + +export const EVIDENCE_FRESHNESS_MS = 4 * 60 * 60 * 1000 + +export function isEvidenceFresh( + activityLog: DashboardPerformance['activity_log'] | undefined, + now = Date.now(), +): boolean | null { + if (!activityLog) return null + + const newestTimestamp = activityLog.reduce((newest, event) => { + if (!event.created_at) return newest + const parsed = Date.parse(event.created_at) + if (Number.isNaN(parsed)) return newest + return newest === null || parsed > newest ? parsed : newest + }, null) + + if (newestTimestamp === null || newestTimestamp > now) return false + return now - newestTimestamp <= EVIDENCE_FRESHNESS_MS +} diff --git a/dashboard/src/pages/DashboardPage.tsx b/dashboard/src/pages/DashboardPage.tsx index 311d8344..12338930 100644 --- a/dashboard/src/pages/DashboardPage.tsx +++ b/dashboard/src/pages/DashboardPage.tsx @@ -3,7 +3,9 @@ import { Activity, BarChart3, CheckCircle, Clock, Radio, ShieldCheck, Users, Wif import { fetchAPI } from '../api/client' import { useWebSocket } from '../api/useWebSocket' import SafetyGateStatus from '../components/SafetyGateStatus' -import type { AgentStatus, DashboardPerformance, DashboardSummary, WSEvent } from '../types' +import { PilotReadinessStrip } from '../components/pilot-readiness-strip' +import { isEvidenceFresh } from '../components/pilot-readiness' +import type { AgentStatus, AgentInstallation, AgentSessionReadiness, ChannelSelectorResponse, DashboardPerformance, DashboardSummary, WSEvent } from '../types' interface LiveEvent { id: number @@ -57,19 +59,38 @@ export default function DashboardPage() { const [agentStatus, setAgentStatus] = useState(null) const [loadError, setLoadError] = useState(null) const [events, setEvents] = useState([]) + const [channels, setChannels] = useState(null) + const [agentSessions, setAgentSessions] = useState([]) const { isConnected, lastEvent } = useWebSocket() const load = useCallback(async () => { - const [status, dashboardSummary, dashboardPerformance] = await Promise.allSettled([ + const [status, dashboardSummary, dashboardPerformance, channelSelector, installations] = await Promise.allSettled([ fetchAPI('/api/status'), fetchAPI('/api/dashboard/summary'), fetchAPI('/api/dashboard/performance?range=7d&limit=30'), + fetchAPI('/api/channels/selector?platform=facebook&channel_type=fanpage&limit=5'), + fetchAPI('/api/agent-installations'), ]) setAgentStatus(status.status === 'fulfilled' ? status.value : null) setSummary(dashboardSummary.status === 'fulfilled' ? dashboardSummary.value : null) setPerformance(dashboardPerformance.status === 'fulfilled' ? dashboardPerformance.value : null) + setChannels(channelSelector.status === 'fulfilled' ? channelSelector.value : null) setLoadError(dashboardSummary.status === 'rejected' || dashboardPerformance.status === 'rejected' ? 'Không tải được dữ liệu ZooPost cloud.' : null) + + // Fetch agent sessions to check publish-dry-run capability + if (installations.status === 'fulfilled') { + const insts = installations.value + const sessionResults = await Promise.allSettled( + insts.map(inst => fetchAPI(`/api/agent-installations/${inst.id}/sessions`)) + ) + const allSessions = sessionResults + .filter((r): r is PromiseFulfilledResult => r.status === 'fulfilled') + .flatMap(r => r.value) + setAgentSessions(allSessions) + } else { + setAgentSessions([]) + } }, []) useEffect(() => { load() }, [load]) @@ -97,6 +118,16 @@ export default function DashboardPage() { const maxChartValue = Math.max(1, ...(performance?.line_chart ?? []).map(row => Math.max(row.scheduled, row.published, row.failed))) + // Pilot readiness props + const safety = agentStatus?.safety_gate + const liveDisabled = safety ? !safety.live_actions_enabled : null + const fbSessionLoggedIn = (agentStatus?.extension?.sessions ?? []).some(s => s.logged_in && s.fb_uid && !s.stale) + const agentHasPublishDryRun = agentSessions.some(s => s.capability_names?.includes('publish-dry-run')) + const selectableChannelCount = channels?.items.filter(c => c.is_selectable).length ?? 0 + const cloudReachable = summary !== null + const fbkitReachable = agentStatus !== null + const evidenceFresh = isEvidenceFresh(performance?.activity_log) + return (
@@ -134,6 +165,17 @@ export default function DashboardPage() { + +
{cards.map(card => (
diff --git a/dashboard/src/pages/LogsPage.test.tsx b/dashboard/src/pages/LogsPage.test.tsx index de0b004f..9db1bff9 100644 --- a/dashboard/src/pages/LogsPage.test.tsx +++ b/dashboard/src/pages/LogsPage.test.tsx @@ -64,4 +64,27 @@ describe('LogsPage evidence view', () => { }) expect(screen.getByText(/dry_run=true, redacted_fields=6/)).toBeTruthy() }) + + it('shows evidence refresh timestamp and dry-run session online indicator', async () => { + render() + + expect(await screen.findByText('Ready agent sessions')).toBeTruthy() + expect(screen.getByText(/Dry-run session ONLINE/i)).toBeTruthy() + expect(screen.getByText(/Last refreshed:/i)).toBeTruthy() + }) + + it('shows dry-run session offline when no ready sessions', async () => { + vi.stubGlobal('fetch', vi.fn((url: string) => { + if (url === '/api/dashboard/summary') return jsonResponse({ kpis: { scheduled_posts: 0, published_posts: 0, total_channels: 0, total_reach: 0 }, status_bar: { buffer_api: 'not_configured', imgbb_api: 'not_configured', pancake: 'not_synced' }, scheduled_targets: 0, published_targets: 0, failed_targets: 0, total_channels: 0 }) + if (url === '/api/publish-jobs?limit=8') return jsonResponse([]) + if (url === '/api/audit-logs?limit=12') return jsonResponse({ items: [], limit: 12 }) + if (url === '/api/channels/selector?limit=20') return jsonResponse({ items: [], limit: 20 }) + if (url === '/api/agent-installations') return jsonResponse([]) + return Promise.resolve({ ok: false, status: 404, text: () => Promise.resolve('not found') } as Response) + })) + + render() + + expect(await screen.findByText(/Dry-run session OFFLINE/i)).toBeTruthy() + }) }) diff --git a/dashboard/src/pages/LogsPage.tsx b/dashboard/src/pages/LogsPage.tsx index 38cc1e27..73fb4998 100644 --- a/dashboard/src/pages/LogsPage.tsx +++ b/dashboard/src/pages/LogsPage.tsx @@ -1,6 +1,6 @@ import { useCallback, useEffect, useMemo, useState } from 'react' import type { CSSProperties, ReactNode } from 'react' -import { CheckCircle2, ClipboardList, FileText, Radio, RefreshCw, ShieldCheck, Wifi } from 'lucide-react' +import { CheckCircle2, ClipboardList, Clock, FileText, Radio, RefreshCw, ShieldCheck, Wifi, WifiOff } from 'lucide-react' import { fetchAPI } from '../api/client' import { LocalPilotChecklist } from '../components/local-pilot-checklist' import { LocalPilotDemoScript } from '../components/local-pilot-demo-script' @@ -23,6 +23,7 @@ export default function LogsPage() { const [state, setState] = useState(emptyState) const [loading, setLoading] = useState(true) const [message, setMessage] = useState(null) + const [evidenceTimestamp, setEvidenceTimestamp] = useState(null) const load = useCallback(async () => { setLoading(true) @@ -39,6 +40,7 @@ export default function LogsPage() { await fetchAPI(`/api/agent-installations/${item.id}/sessions`), ] as const)) setState({ summary, jobs, audit, channels, installations, sessions: Object.fromEntries(sessionPairs) }) + setEvidenceTimestamp(new Date().toISOString()) setMessage(null) } catch (error) { setMessage(error instanceof Error ? error.message : 'Khong tai duoc evidence log.') @@ -60,10 +62,21 @@ export default function LogsPage() {
Evidence Log
Browser-safe proof for the local dry-run pilot: readiness, jobs, audit trail, and safety boundary.
+ {evidenceTimestamp && ( +
+ Last refreshed: {new Date(evidenceTimestamp).toLocaleString('vi-VN')} +
+ )} +
+
+ 0 ? '#16a34a' : 'var(--muted)' }}> + {readySessions.length > 0 ? : } + {readySessions.length > 0 ? 'Dry-run session ONLINE' : 'Dry-run session OFFLINE'} + +
-
{message &&
{message}
} diff --git a/dashboard/src/router-security-contract.test.ts b/dashboard/src/router-security-contract.test.ts new file mode 100644 index 00000000..55c77d61 --- /dev/null +++ b/dashboard/src/router-security-contract.test.ts @@ -0,0 +1,86 @@ +import { describe, expect, it } from 'vitest' +import { readFileSync } from 'node:fs' +import { dirname, join } from 'node:path' +import { fileURLToPath } from 'node:url' + +const here = dirname(fileURLToPath(import.meta.url)) +const srcRoot = here + +/** Read every TS/TSX file under src/ as a single string for static contract checks. */ +function readSourceTree(): string { + // Scan the same files the build ships. We only assert import patterns, + // so a concatenated read is sufficient and avoids dynamic import edge cases. + const files: string[] = [ + 'App.tsx', + 'main.tsx', + 'api/client.ts', + 'api/useWebSocket.ts', + 'components/SafetyGateStatus.tsx', + 'components/pilot-readiness-strip.tsx', + 'components/local-pilot-checklist.tsx', + 'components/local-pilot-demo-readiness-strip.tsx', + 'components/local-pilot-demo-script.tsx', + 'components/local-pilot-evidence-summary.tsx', + 'components/projects/ApprovalQueueTab.tsx', + 'components/projects/CampaignsRunsTab.tsx', + 'components/projects/ProjectLiveFlagCopy.tsx', + 'pages/AccountsPage.tsx', + 'pages/AutoPostFanpagePage.tsx', + 'pages/DashboardPage.tsx', + 'pages/LogsPage.tsx', + 'pages/ProjectsPage.tsx', + 'pages/WorkflowLabPage.tsx', + 'utils/safe-external-url.ts', + ] + return files + .map((rel) => { + try { + return readFileSync(join(srcRoot, rel), 'utf-8') + } catch { + return '' + } + }) + .join('\n') +} + +describe('React Router RSC non-applicability contract', () => { + // Tracks GHSA-qwww-vcr4-c8h2 (react-router-dom advisory). The dashboard + // retains React Router v7 and relies only on declarative APIs, so the + // unstable RSC/data-strategy surfaces named in the advisory are not used. + // This test fails closed if any unstable RSC surface appears in src/. + it('does not import any unstable RSC or data-strategy surface', () => { + const source = readSourceTree() + const forbidden = [ + 'unstable_', + 'unstableLoader', + 'unstableDataStrategy', + 'dataStrategy', + 'isRouteErrorResponse', + 'renderMatches', + 'RouterProvider', + 'createStaticHandler', + 'createStaticRouter', + ] + const hits = forbidden.filter((token) => source.includes(token)) + expect(hits, `unexpected unstable RSC surfaces: ${hits.join(', ')}`).toEqual([]) + }) + + it('uses only declarative router exports from react-router-dom', () => { + const source = readSourceTree() + // Only these declarative exports are permitted while the RSC exception holds. + const allowed = ['BrowserRouter', 'NavLink', 'Routes', 'Route', 'useLocation', 'useNavigate', 'Link', 'useParams', 'Outlet', 'Navigate'] + const importLines = source.split('\n').filter((line) => line.includes("from 'react-router-dom'") || line.includes('from "react-router-dom"')) + expect(importLines.length).toBeGreaterThan(0) + // Extract imported names from each react-router-dom import line. + const imported = new Set() + for (const line of importLines) { + const match = line.match(/\{([^}]*)\}/) + if (!match) continue + for (const name of match[1].split(',').map((s) => s.trim()).filter(Boolean)) { + imported.add(name) + } + } + const unexpected = [...imported].filter((name) => !allowed.includes(name)) + expect(unexpected, `unexpected router imports: ${unexpected.join(', ')}`).toEqual([]) + }) +}) diff --git a/dashboard/src/types/index.ts b/dashboard/src/types/index.ts index 24549548..ee3e87ce 100644 --- a/dashboard/src/types/index.ts +++ b/dashboard/src/types/index.ts @@ -199,6 +199,7 @@ export interface DashboardPerformance { severity: string message: string target_id: string | null + created_at: string | null }> } diff --git a/dashboard/tsconfig.app.json b/dashboard/tsconfig.app.json index af516fcc..9c4b6a70 100644 --- a/dashboard/tsconfig.app.json +++ b/dashboard/tsconfig.app.json @@ -5,7 +5,7 @@ "useDefineForClassFields": true, "lib": ["ES2023", "DOM", "DOM.Iterable"], "module": "ESNext", - "types": ["vite/client"], + "types": ["vite/client", "node"], "skipLibCheck": true, /* Bundler mode */