diff --git a/Dockerfile b/Dockerfile index 6f4bd108..e8074dec 100644 --- a/Dockerfile +++ b/Dockerfile @@ -19,7 +19,7 @@ RUN apt-get update && apt-get install -y --no-install-recommends \ libglib2.0-0 libgtk-3-0 libpangocairo-1.0-0 libcairo-gobject2 \ libgdk-pixbuf-2.0-0 libxss1 libxtst6 fonts-liberation \ libgl1-mesa-dri libegl-mesa0 \ - procps wget ca-certificates xclip \ + procps wget ca-certificates xclip ffmpeg \ && rm -rf /var/lib/apt/lists/* # Playwright system deps (matches test-infra) diff --git a/backend/control_lease.py b/backend/control_lease.py new file mode 100644 index 00000000..d9bc5fab --- /dev/null +++ b/backend/control_lease.py @@ -0,0 +1,131 @@ +"""Exclusive per-profile control leases for agent and human input.""" + +from __future__ import annotations + +import math +import threading +import time +import uuid +from collections.abc import Callable +from dataclasses import dataclass +from datetime import datetime, timezone +from typing import Literal + +ControlHolder = Literal["agent", "human"] +ControlState = Literal["idle", "agent", "human"] + + +class ControlLeaseError(RuntimeError): + """Base class for control-lease transition failures.""" + + +class ControlLeaseConflictError(ControlLeaseError): + """Raised when an agent tries to take control from a human.""" + + +class ControlLeaseMismatchError(ControlLeaseError): + """Raised when release does not present the active lease identifier.""" + + +@dataclass(frozen=True, slots=True) +class ControlLease: + lease_id: str + state: Literal["agent", "human"] + holder: ControlHolder + expires_at: float + + def to_wire(self) -> dict[str, str]: + return { + "lease_id": self.lease_id, + "state": self.state, + "holder": self.holder, + "expires_at": datetime.fromtimestamp(self.expires_at, tz=timezone.utc) + .isoformat() + .replace("+00:00", "Z"), + } + + +class ControlLeaseManager: + """Atomic in-memory IDLE/AGENT/HUMAN FSM keyed by profile id. + + Human acquisition preempts an agent lease. An agent cannot preempt a human; + the human must release its exact lease id before the agent reacquires. + Expired leases are discarded at every lease boundary. + """ + + def __init__( + self, + *, + clock: Callable[[], float] = time.time, + lease_id_factory: Callable[[], str] | None = None, + ) -> None: + self._clock = clock + self._lease_id_factory = lease_id_factory or (lambda: str(uuid.uuid4())) + self._leases: dict[str, ControlLease] = {} + self._lock = threading.RLock() + + def acquire( + self, + profile_id: str, + holder: ControlHolder, + ttl_s: float, + ) -> ControlLease: + if holder not in ("agent", "human"): + raise ValueError("holder must be 'agent' or 'human'") + if not math.isfinite(ttl_s) or ttl_s <= 0: + raise ValueError("ttl_s must be a positive finite number") + + with self._lock: + now = self._clock() + current = self._active_lease(profile_id, now) + if current and current.holder == "human" and holder == "agent": + raise ControlLeaseConflictError( + f"Profile {profile_id} is controlled by a human" + ) + + lease = ControlLease( + lease_id=self._lease_id_factory(), + state=holder, + holder=holder, + expires_at=now + ttl_s, + ) + self._leases[profile_id] = lease + return lease + + def release(self, profile_id: str, lease_id: str) -> None: + with self._lock: + current = self._active_lease(profile_id, self._clock()) + if current is None or current.lease_id != lease_id: + raise ControlLeaseMismatchError( + f"Lease {lease_id} is not active for profile {profile_id}" + ) + del self._leases[profile_id] + + def status(self, profile_id: str) -> ControlLease | None: + with self._lock: + return self._active_lease(profile_id, self._clock()) + + def state(self, profile_id: str) -> ControlState: + lease = self.status(profile_id) + return lease.state if lease else "idle" + + def agent_can_dispatch(self, profile_id: str, lease_id: str) -> bool: + """Return whether this exact agent lease may dispatch CDP input.""" + lease = self.status(profile_id) + return bool(lease and lease.holder == "agent" and lease.lease_id == lease_id) + + def release_profile(self, profile_id: str) -> None: + """Release any lease because its browser profile stopped.""" + with self._lock: + self._leases.pop(profile_id, None) + + def clear(self) -> None: + with self._lock: + self._leases.clear() + + def _active_lease(self, profile_id: str, now: float) -> ControlLease | None: + lease = self._leases.get(profile_id) + if lease and lease.expires_at <= now: + del self._leases[profile_id] + return None + return lease diff --git a/backend/main.py b/backend/main.py index f4c56bdc..538470c0 100644 --- a/backend/main.py +++ b/backend/main.py @@ -10,22 +10,29 @@ import hmac import logging import os +import shutil import signal import struct -import shutil import time import webbrowser from contextlib import asynccontextmanager -from logging.handlers import RotatingFileHandler from http.cookies import SimpleCookie +from logging.handlers import RotatingFileHandler from pathlib import Path from urllib.parse import urlparse import httpx -from fastapi import FastAPI, HTTPException, Request, Response, WebSocket, WebSocketDisconnect +import starlette.requests +from fastapi import ( + FastAPI, + HTTPException, + Request, + Response, + WebSocket, + WebSocketDisconnect, +) from fastapi.responses import FileResponse, JSONResponse from fastapi.staticfiles import StaticFiles -import starlette.requests from starlette.types import ASGIApp, Receive, Scope, Send from .env_file import load_env_file @@ -34,13 +41,14 @@ # (database.resolve_runtime, AUTH_TOKEN, the license config below). load_env_file() -from . import database as db from cloakbrowser.license import CloakBrowserLicenseError +from . import database as db +from . import recorder as teach_replay_recorder from .browser_manager import ( + SCREENSHOT_FILENAME, BrowserManager, ProfileBusyError, - SCREENSHOT_FILENAME, is_seat_limit_error, license_error_detail, test_proxy, @@ -65,6 +73,8 @@ ) from .runtime import bundle_dir from .settings_store import load_settings, save_settings +from .teach_replay_api import control_leases +from .teach_replay_api import router as teach_replay_router logger = logging.getLogger("cloakbrowser.manager") @@ -442,11 +452,13 @@ def _rewrite_pointer_event(data: bytes, offset: int) -> bytes: return struct.pack(">BHHHhh", 5, mask, x, y, 0, 0) -def _filter_rfb_client_messages(data: bytes) -> bytes: +def _filter_rfb_client_messages(data: bytes, *, input_allowed: bool = True) -> bytes: """Parse concatenated RFB messages, keep only standard types (0-6). Rewrites PointerEvents from 6-byte standard to 11-byte KasmVNC format - and strips unsupported pseudo-encodings from SetEncodings. + and strips unsupported pseudo-encodings from SetEncodings. When input is + blocked, KeyEvent, PointerEvent, and ClientCutText messages are removed + while view-maintenance messages continue to the server. """ _log = logging.getLogger("cloakbrowser.manager") result = bytearray() @@ -467,6 +479,15 @@ def _filter_rfb_client_messages(data: bytes) -> bytes: break msg_idx += 1 if msg_type in _RFB_MSG_SIZE: + if not input_allowed and msg_type in (4, 5, 6): + _log.debug( + "RFB filter: BLOCK input type=%d len=%d at offset=%d", + msg_type, + msg_len, + offset, + ) + offset += msg_len + continue # Standard RFB type — keep (with rewrites for KasmVNC compatibility) _log.debug("RFB filter: KEEP type=%d len=%d at offset=%d (msg #%d in frame)", msg_type, msg_len, offset, msg_idx) if msg_type == 2: # SetEncodings — whitelist safe encodings @@ -501,11 +522,23 @@ async def lifespan(app: FastAPI): if browser_mgr._auto_launch_task and not browser_mgr._auto_launch_task.done(): browser_mgr._auto_launch_task.cancel() await asyncio.gather(browser_mgr._auto_launch_task, return_exceptions=True) - await browser_mgr.cleanup_all() + try: + for profile_id in tuple(browser_mgr.running): + await _stop_running_profile(profile_id) + await browser_mgr.cleanup_all() + finally: + control_leases.clear() app = FastAPI(title="CloakBrowser Manager", lifespan=lifespan) app.add_middleware(AuthMiddleware) +app.include_router(teach_replay_router) + + +async def _stop_running_profile(profile_id: str) -> None: + await teach_replay_recorder.on_profile_stopped(profile_id) + await browser_mgr.stop(profile_id) + control_leases.release_profile(profile_id) # ── Authentication ──────────────────────────────────────────────────────────── @@ -615,7 +648,7 @@ async def update_profile(profile_id: str, req: ProfileUpdate): async def delete_profile(profile_id: str): # Stop browser if running if profile_id in browser_mgr.running: - await browser_mgr.stop(profile_id) + await _stop_running_profile(profile_id) profile = db.get_profile(profile_id) if not profile: @@ -681,7 +714,7 @@ async def reset_profile(profile_id: str): raise HTTPException(status_code=404, detail="Profile not found") if profile_id in browser_mgr.running: - await browser_mgr.stop(profile_id) + await _stop_running_profile(profile_id) user_data_dir = Path(profile["user_data_dir"]) default_dir = user_data_dir / "Default" @@ -855,7 +888,7 @@ async def launch_profile(profile_id: str): async def stop_profile(profile_id: str): if profile_id not in browser_mgr.running: raise HTTPException(status_code=404, detail="Profile is not running") - await browser_mgr.stop(profile_id) + await _stop_running_profile(profile_id) return {"ok": True} @@ -1291,7 +1324,10 @@ async def client_to_vnc(): continue # Parse RFB messages and strip unsupported types - filtered = _filter_rfb_client_messages(data) + filtered = _filter_rfb_client_messages( + data, + input_allowed=control_leases.state(profile_id) == "human", + ) if filtered: # Safety: verify first byte is a valid RFB client type if filtered[0] not in _RFB_MSG_SIZE: @@ -1478,7 +1514,10 @@ async def _proxy_cdp_websocket( ) -> None: """Bidirectional WebSocket proxy between a FastAPI client and a CDP target. - Used by both browser-level and page-level CDP proxy endpoints. + Used by both browser-level and page-level CDP proxy endpoints. CDP input + gating is cooperative: callers must hold the exact AGENT lease and check + ``control_leases.agent_can_dispatch`` at the dispatch boundary. The proxy + remains bidirectional so observation traffic continues during handoff. """ import websockets diff --git a/backend/recorder.py b/backend/recorder.py new file mode 100644 index 00000000..0fa77c36 --- /dev/null +++ b/backend/recorder.py @@ -0,0 +1,526 @@ +"""Per-profile recording of the manager's private X displays.""" + +from __future__ import annotations + +import asyncio +import json +import logging +import math +import re +import signal +import subprocess +import threading +import uuid +from collections.abc import Callable +from dataclasses import dataclass +from datetime import datetime, timezone +from pathlib import Path +from typing import Literal + +from fastapi import HTTPException +from fastapi.responses import FileResponse +from pydantic import BaseModel, ConfigDict + +from . import database as db +from .browser_manager import RunningProfile +from .teach_replay_api import router +from .vnc_manager import VNCManager + +logger = logging.getLogger("cloakbrowser.manager.recorder") + +MAX_RECORDING_SECONDS = 600 +MIN_RECORDING_SECONDS = 0.5 +RECORDING_FPS = 15 + + +class RecordingError(RuntimeError): + """Base class for recording lifecycle failures.""" + + +class RecordingConflictError(RecordingError): + """Raised when a profile already has a recording operation in progress.""" + + +class RecordingNotActiveError(RecordingError): + """Raised when a profile has no active recording to stop.""" + + +class RecordingUnavailableError(RecordingError): + """Raised when a running profile has no recordable X display.""" + + +class RecordingProcessError(RecordingError): + """Raised when ffmpeg cannot be started or cleanly stopped.""" + + +class RecordingValidationError(RecordingError): + """Raised when ffprobe rejects a completed recording.""" + + +class RecordingNotFoundError(RecordingError): + """Raised when a requested completed recording is unavailable.""" + + +@dataclass(slots=True) +class _ActiveRecording: + profile_id: str + recording_id: str + path: Path + process: subprocess.Popen[bytes] + stopping: bool = False + + +@dataclass(frozen=True, slots=True) +class CompletedRecording: + profile_id: str + recording_id: str + path: Path + duration_s: float + + +@dataclass(frozen=True, slots=True) +class RecordingSnapshot: + recording_id: str | None + state: Literal["recording", "stopped"] + + +class RecordingStatusResponse(BaseModel): + model_config = ConfigDict(extra="forbid") + + recording_id: str | None + state: Literal["recording", "stopped"] + + +class RecordingStopResponse(BaseModel): + model_config = ConfigDict(extra="forbid") + + recording_id: str + duration_s: float + path: str + + +class RecordingManager: + """Own ffmpeg processes and validated MP4s for running profiles.""" + + def __init__( + self, + running_profile_getter: Callable[[str], RunningProfile | None], + *, + recordings_dir: Path | None = None, + screen_size_getter: Callable[[str], tuple[int, int] | None] | None = None, + now: Callable[[], datetime] | None = None, + ) -> None: + self._running_profile_getter = running_profile_getter + self._configured_recordings_dir = recordings_dir + self._screen_size_getter = screen_size_getter + self._now = now or (lambda: datetime.now(timezone.utc)) + self._active: dict[str, _ActiveRecording] = {} + self._completed: dict[str, CompletedRecording] = {} + self._last_by_profile: dict[str, str] = {} + self._lock = threading.RLock() + + @property + def recordings_dir(self) -> Path: + return self._configured_recordings_dir or db.DATA_DIR / "recordings" + + def start(self, profile_id: str) -> RecordingSnapshot: + self._reap_finished_before_start(profile_id) + + with self._lock: + if profile_id in self._active: + raise RecordingConflictError( + f"Profile {profile_id} already has an active recording" + ) + + running = self._running_profile_getter(profile_id) + if running is None: + raise RecordingUnavailableError(f"Profile {profile_id} is not running") + display = self._resolve_display(running) + recording_id, output_path = self._new_recording_path(profile_id) + output_path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) + command = self._ffmpeg_command(profile_id, display, output_path) + + try: + process = subprocess.Popen( + command, + stdin=subprocess.PIPE, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + except OSError as exc: + output_path.unlink(missing_ok=True) + raise RecordingProcessError( + "Could not start the screen recorder" + ) from exc + + self._active[profile_id] = _ActiveRecording( + profile_id=profile_id, + recording_id=recording_id, + path=output_path, + process=process, + ) + self._last_by_profile.pop(profile_id, None) + + logger.info("Started recording %s for profile %s", recording_id, profile_id) + return RecordingSnapshot(recording_id=recording_id, state="recording") + + def stop(self, profile_id: str) -> CompletedRecording: + with self._lock: + active = self._active.get(profile_id) + if active is None: + raise RecordingNotActiveError( + f"Profile {profile_id} has no active recording" + ) + if active.stopping: + raise RecordingConflictError( + f"Profile {profile_id} recording is already stopping" + ) + active.stopping = True + + try: + self._stop_process(active.process) + except Exception as exc: + self._discard(active) + if isinstance(exc, RecordingError): + raise + raise RecordingProcessError("Could not stop the screen recorder") from exc + + return self._finalize(active) + + def status(self, profile_id: str) -> RecordingSnapshot: + with self._lock: + active = self._active.get(profile_id) + if active is None: + return RecordingSnapshot( + recording_id=self._last_by_profile.get(profile_id), + state="stopped", + ) + if active.stopping or active.process.poll() is None: + return RecordingSnapshot( + recording_id=active.recording_id, + state="recording", + ) + active.stopping = True + + completed = self._finalize(active) + return RecordingSnapshot( + recording_id=completed.recording_id, + state="stopped", + ) + + def recording_path(self, recording_id: str) -> Path: + with self._lock: + completed = self._completed.get(recording_id) + if completed is None or not completed.path.is_file(): + self._completed.pop(recording_id, None) + raise RecordingNotFoundError(f"Recording {recording_id} was not found") + return completed.path + + def _reap_finished_before_start(self, profile_id: str) -> None: + with self._lock: + active = self._active.get(profile_id) + if active is None: + return + if active.stopping or active.process.poll() is None: + raise RecordingConflictError( + f"Profile {profile_id} already has an active recording" + ) + active.stopping = True + + try: + self._finalize(active) + except RecordingValidationError: + logger.warning( + "Discarded invalid recording %s before starting another", + active.recording_id, + ) + + def _new_recording_path(self, profile_id: str) -> tuple[str, Path]: + profile_key = re.sub(r"[^A-Za-z0-9_-]", "_", profile_id).strip("_") + profile_key = (profile_key or "profile")[:100] + timestamp = self._now().astimezone(timezone.utc).strftime("%Y%m%dT%H%M%S%fZ") + suffix = uuid.uuid4().hex[:8] + recording_id = f"{profile_key}-{timestamp}-{suffix}" + path = self.recordings_dir / profile_key / f"{timestamp}-{suffix}.mp4" + return recording_id, path + + def _ffmpeg_command( + self, + profile_id: str, + display: int, + output_path: Path, + ) -> list[str]: + command = [ + "ffmpeg", + "-y", + "-hide_banner", + "-loglevel", + "error", + "-f", + "x11grab", + "-framerate", + str(RECORDING_FPS), + ] + if self._screen_size_getter is not None: + screen_size = self._screen_size_getter(profile_id) + if screen_size is not None: + width, height = screen_size + command.extend(("-video_size", f"{width}x{height}")) + command.extend( + ( + "-i", + f":{display}.0", + "-t", + str(MAX_RECORDING_SECONDS), + "-an", + "-c:v", + "libx264", + "-pix_fmt", + "yuv420p", + "-movflags", + "+faststart", + str(output_path), + ) + ) + return command + + @staticmethod + def _resolve_display(running: RunningProfile) -> int: + display = running.display + if isinstance(display, int) and not isinstance(display, bool) and display >= 0: + return display + + ws_port = running.ws_port + if isinstance(ws_port, int) and not isinstance(ws_port, bool): + offset = ws_port - VNCManager.BASE_WS_PORT + if offset >= 0: + return VNCManager.BASE_DISPLAY + offset + + raise RecordingUnavailableError( + f"Profile {running.profile_id} has no private X display" + ) + + @staticmethod + def _stop_process(process: subprocess.Popen[bytes]) -> None: + if process.poll() is not None: + process.wait(timeout=0) + return + + wrote_quit = False + if process.stdin is not None: + try: + process.stdin.write(b"q\n") + process.stdin.flush() + wrote_quit = True + except (BrokenPipeError, OSError): + pass + + if not wrote_quit: + process.send_signal(signal.SIGINT) + + try: + process.wait(timeout=10) + except subprocess.TimeoutExpired: + process.send_signal(signal.SIGINT) + try: + process.wait(timeout=5) + except subprocess.TimeoutExpired: + process.kill() + process.wait(timeout=5) + finally: + if process.stdin is not None: + try: + process.stdin.close() + except OSError: + pass + + def _finalize(self, active: _ActiveRecording) -> CompletedRecording: + try: + duration_s = self._probe_duration(active.path) + except Exception: + self._discard(active) + raise + + completed = CompletedRecording( + profile_id=active.profile_id, + recording_id=active.recording_id, + path=active.path, + duration_s=duration_s, + ) + with self._lock: + if self._active.get(active.profile_id) is active: + del self._active[active.profile_id] + self._completed[active.recording_id] = completed + self._last_by_profile[active.profile_id] = active.recording_id + logger.info( + "Finalized recording %s (%.3fs)", + active.recording_id, + duration_s, + ) + return completed + + def _discard(self, active: _ActiveRecording) -> None: + active.path.unlink(missing_ok=True) + with self._lock: + if self._active.get(active.profile_id) is active: + del self._active[active.profile_id] + self._last_by_profile.pop(active.profile_id, None) + + @staticmethod + def _probe_duration(path: Path) -> float: + if not path.is_file() or path.stat().st_size == 0: + raise RecordingValidationError("Recording did not produce an MP4") + + try: + result = subprocess.run( + [ + "ffprobe", + "-v", + "error", + "-show_entries", + "format=duration", + "-of", + "json", + str(path), + ], + capture_output=True, + text=True, + timeout=15, + check=False, + ) + except (OSError, subprocess.TimeoutExpired) as exc: + raise RecordingValidationError( + "Could not validate the completed recording" + ) from exc + + if result.returncode != 0: + raise RecordingValidationError("The completed recording is corrupt") + try: + payload = json.loads(result.stdout) + duration_s = float(payload["format"]["duration"]) + except (KeyError, TypeError, ValueError, json.JSONDecodeError) as exc: + raise RecordingValidationError( + "The completed recording has no valid duration" + ) from exc + if not math.isfinite(duration_s) or duration_s < MIN_RECORDING_SECONDS: + raise RecordingValidationError( + f"Recording must be at least {MIN_RECORDING_SECONDS} seconds" + ) + return duration_s + + +def _running_profile(profile_id: str) -> RunningProfile | None: + from .main import browser_mgr + + return browser_mgr.running.get(profile_id) + + +def _profile_screen_size(profile_id: str) -> tuple[int, int] | None: + profile = db.get_profile(profile_id) + if profile is None: + return None + width = profile.get("screen_width") + height = profile.get("screen_height") + if ( + isinstance(width, int) + and not isinstance(width, bool) + and width > 0 + and isinstance(height, int) + and not isinstance(height, bool) + and height > 0 + ): + return width, height + return None + + +recording_manager = RecordingManager( + _running_profile, + screen_size_getter=_profile_screen_size, +) + + +async def on_profile_stopped(profile_id: str) -> CompletedRecording | None: + """Finalize an active capture before main tears down the profile display.""" + try: + return await asyncio.to_thread(recording_manager.stop, profile_id) + except RecordingNotActiveError: + return None + except RecordingError as exc: + logger.warning( + "Could not finalize recording while profile %s stopped: %s", + profile_id, + exc, + ) + return None + + +def _recording_http_exception(exc: RecordingError) -> HTTPException: + if isinstance(exc, RecordingNotFoundError): + return HTTPException(status_code=404, detail=str(exc)) + if isinstance(exc, RecordingValidationError): + return HTTPException(status_code=422, detail=str(exc)) + if isinstance( + exc, + (RecordingConflictError, RecordingNotActiveError, RecordingUnavailableError), + ): + return HTTPException(status_code=409, detail=str(exc)) + return HTTPException(status_code=500, detail=str(exc)) + + +@router.post( + "/api/profiles/{profile_id}/recording/start", + response_model=RecordingStatusResponse, +) +async def start_recording(profile_id: str) -> RecordingStatusResponse: + try: + snapshot = await asyncio.to_thread(recording_manager.start, profile_id) + except RecordingError as exc: + raise _recording_http_exception(exc) from exc + return RecordingStatusResponse( + recording_id=snapshot.recording_id, + state=snapshot.state, + ) + + +@router.post( + "/api/profiles/{profile_id}/recording/stop", + response_model=RecordingStopResponse, +) +async def stop_recording(profile_id: str) -> RecordingStopResponse: + try: + completed = await asyncio.to_thread(recording_manager.stop, profile_id) + except RecordingError as exc: + raise _recording_http_exception(exc) from exc + return RecordingStopResponse( + recording_id=completed.recording_id, + duration_s=completed.duration_s, + path=f"/api/recording/{completed.recording_id}", + ) + + +@router.get( + "/api/profiles/{profile_id}/recording/status", + response_model=RecordingStatusResponse, +) +async def recording_status(profile_id: str) -> RecordingStatusResponse: + try: + snapshot = await asyncio.to_thread(recording_manager.status, profile_id) + except RecordingError as exc: + raise _recording_http_exception(exc) from exc + return RecordingStatusResponse( + recording_id=snapshot.recording_id, + state=snapshot.state, + ) + + +@router.get("/api/recording/{recording_id}", response_class=FileResponse) +async def download_recording(recording_id: str) -> FileResponse: + try: + path = recording_manager.recording_path(recording_id) + except RecordingError as exc: + raise _recording_http_exception(exc) from exc + return FileResponse( + path, + media_type="video/mp4", + filename=f"{recording_id}.mp4", + ) diff --git a/backend/teach_replay_api.py b/backend/teach_replay_api.py new file mode 100644 index 00000000..169c54d6 --- /dev/null +++ b/backend/teach_replay_api.py @@ -0,0 +1,67 @@ +"""Shared API router for teach/replay manager capabilities.""" + +from __future__ import annotations + +from typing import Literal + +from fastapi import APIRouter, HTTPException, Response +from pydantic import BaseModel, ConfigDict, Field + +from .control_lease import ( + ControlLeaseConflictError, + ControlLeaseManager, + ControlLeaseMismatchError, +) + +router = APIRouter() +control_leases = ControlLeaseManager() + + +class AcquireControlRequest(BaseModel): + model_config = ConfigDict(extra="forbid") + + holder: Literal["agent", "human"] + ttl_s: float = Field(gt=0, allow_inf_nan=False) + + +class ReleaseControlRequest(BaseModel): + model_config = ConfigDict(extra="forbid") + + lease_id: str = Field(min_length=1) + + +def _lease_payload(profile_id: str) -> dict[str, object]: + lease = control_leases.status(profile_id) + if lease: + return {"lease": lease.to_wire()} + return { + "lease": { + "lease_id": None, + "state": "idle", + "holder": None, + "expires_at": None, + } + } + + +@router.post("/api/profiles/{profile_id}/control/acquire") +async def acquire_control(profile_id: str, request: AcquireControlRequest): + try: + lease = control_leases.acquire(profile_id, request.holder, request.ttl_s) + except ControlLeaseConflictError as exc: + raise HTTPException(status_code=409, detail=str(exc)) from exc + return {"lease": lease.to_wire()} + + +@router.post("/api/profiles/{profile_id}/control/release") +async def release_control(profile_id: str, request: ReleaseControlRequest): + try: + control_leases.release(profile_id, request.lease_id) + except ControlLeaseMismatchError as exc: + raise HTTPException(status_code=409, detail=str(exc)) from exc + return Response(status_code=204) + + +@router.get("/api/profiles/{profile_id}/control/status") +async def control_status(profile_id: str): + return _lease_payload(profile_id) diff --git a/backend/tests/test_control_lease.py b/backend/tests/test_control_lease.py new file mode 100644 index 00000000..4c9a4339 --- /dev/null +++ b/backend/tests/test_control_lease.py @@ -0,0 +1,217 @@ +"""Tests for per-profile agent/human control leases.""" + +from __future__ import annotations + +import struct +from unittest.mock import AsyncMock, MagicMock + +import pytest +from starlette.testclient import TestClient + +from backend import main +from backend.control_lease import ( + ControlLeaseConflictError, + ControlLeaseManager, + ControlLeaseMismatchError, +) +from backend.teach_replay_api import control_leases + + +class _Clock: + def __init__(self, now: float = 1_000.0) -> None: + self.now = now + + def __call__(self) -> float: + return self.now + + +@pytest.fixture(autouse=True) +def _clear_shared_leases(): + control_leases.clear() + yield + control_leases.clear() + + +def _manager(clock: _Clock) -> ControlLeaseManager: + ids = iter(("lease-1", "lease-2", "lease-3", "lease-4")) + return ControlLeaseManager(clock=clock, lease_id_factory=lambda: next(ids)) + + +def test_idle_profile_can_be_acquired_by_either_holder() -> None: + clock = _Clock() + leases = _manager(clock) + + agent = leases.acquire("agent-profile", "agent", 30) + human = leases.acquire("human-profile", "human", 45) + + assert agent.to_wire() == { + "lease_id": "lease-1", + "state": "agent", + "holder": "agent", + "expires_at": "1970-01-01T00:17:10Z", + } + assert human.state == "human" + assert leases.state("missing-profile") == "idle" + + +def test_human_takeover_preempts_agent_and_invalidates_agent_lease() -> None: + clock = _Clock() + leases = _manager(clock) + agent = leases.acquire("profile-1", "agent", 60) + + human = leases.acquire("profile-1", "human", 120) + + assert human.lease_id != agent.lease_id + assert leases.status("profile-1") == human + assert not leases.agent_can_dispatch("profile-1", agent.lease_id) + with pytest.raises(ControlLeaseMismatchError): + leases.release("profile-1", agent.lease_id) + assert leases.status("profile-1") == human + + +def test_agent_reacquires_only_after_human_handback() -> None: + clock = _Clock() + leases = _manager(clock) + human = leases.acquire("profile-1", "human", 60) + + with pytest.raises(ControlLeaseConflictError): + leases.acquire("profile-1", "agent", 60) + + leases.release("profile-1", human.lease_id) + agent = leases.acquire("profile-1", "agent", 60) + assert leases.agent_can_dispatch("profile-1", agent.lease_id) + + +def test_lease_ttl_expires_back_to_idle() -> None: + clock = _Clock() + leases = _manager(clock) + lease = leases.acquire("profile-1", "agent", 10) + clock.now = lease.expires_at + + assert leases.status("profile-1") is None + assert leases.state("profile-1") == "idle" + with pytest.raises(ControlLeaseMismatchError): + leases.release("profile-1", lease.lease_id) + + +def test_release_requires_exact_active_lease_id() -> None: + clock = _Clock() + leases = _manager(clock) + lease = leases.acquire("profile-1", "agent", 60) + + with pytest.raises(ControlLeaseMismatchError): + leases.release("profile-1", "wrong-id") + assert leases.status("profile-1") == lease + + leases.release("profile-1", lease.lease_id) + assert leases.state("profile-1") == "idle" + + +def test_release_profile_clears_any_holder() -> None: + clock = _Clock() + leases = _manager(clock) + leases.acquire("profile-1", "human", 60) + + leases.release_profile("profile-1") + + assert leases.state("profile-1") == "idle" + + +def test_control_api_matches_wire_contract_and_reports_conflict( + app_client: TestClient, +) -> None: + idle = app_client.get("/api/profiles/profile-1/control/status") + assert idle.status_code == 200 + assert idle.json() == { + "lease": { + "lease_id": None, + "state": "idle", + "holder": None, + "expires_at": None, + } + } + + acquired = app_client.post( + "/api/profiles/profile-1/control/acquire", + json={"holder": "human", "ttl_s": 60}, + ) + assert acquired.status_code == 200 + lease = acquired.json()["lease"] + assert set(lease) == {"lease_id", "state", "holder", "expires_at"} + assert lease["state"] == "human" + assert lease["holder"] == "human" + + conflict = app_client.post( + "/api/profiles/profile-1/control/acquire", + json={"holder": "agent", "ttl_s": 60}, + ) + assert conflict.status_code == 409 + + wrong_release = app_client.post( + "/api/profiles/profile-1/control/release", + json={"lease_id": "not-the-active-lease"}, + ) + assert wrong_release.status_code == 409 + + released = app_client.post( + "/api/profiles/profile-1/control/release", + json={"lease_id": lease["lease_id"]}, + ) + assert released.status_code == 204 + assert released.content == b"" + assert ( + app_client.get("/api/profiles/profile-1/control/status").json()["lease"][ + "state" + ] + == "idle" + ) + + +def test_profile_stop_releases_control_lease( + app_client: TestClient, + monkeypatch: pytest.MonkeyPatch, +) -> None: + created = app_client.post("/api/profiles", json={"name": "Lease Stop"}) + profile_id = created.json()["id"] + main.browser_mgr.running[profile_id] = MagicMock() + monkeypatch.setattr(main.browser_mgr, "stop", AsyncMock()) + control_leases.acquire(profile_id, "human", 60) + + try: + response = app_client.post(f"/api/profiles/{profile_id}/stop") + assert response.status_code == 200 + assert control_leases.state(profile_id) == "idle" + finally: + main.browser_mgr.running.pop(profile_id, None) + + +def test_rfb_gate_drops_input_while_preserving_view_requests() -> None: + key = struct.pack(">BBxxI", 4, 1, 0x61) + pointer = struct.pack(">BBHH", 5, 1, 100, 200) + clipboard_text = b"hello" + clipboard = struct.pack(">BxxxI", 6, len(clipboard_text)) + clipboard_text + framebuffer_request = struct.pack(">BBHHHH", 3, 1, 0, 0, 1920, 1080) + + profile_id = "profile-1" + idle = main._filter_rfb_client_messages( + key + framebuffer_request + pointer + clipboard, + input_allowed=control_leases.state(profile_id) == "human", + ) + control_leases.acquire(profile_id, "agent", 60) + blocked = main._filter_rfb_client_messages( + key + framebuffer_request + pointer + clipboard, + input_allowed=control_leases.state(profile_id) == "human", + ) + control_leases.release_profile(profile_id) + control_leases.acquire(profile_id, "human", 60) + allowed = main._filter_rfb_client_messages( + key + framebuffer_request + pointer + clipboard, + input_allowed=control_leases.state(profile_id) == "human", + ) + + assert idle == framebuffer_request + assert blocked == framebuffer_request + assert allowed[0] == 4 + assert allowed[8:18] == framebuffer_request + assert allowed[18] == 5 + assert allowed[29] == 6 diff --git a/backend/tests/test_recorder.py b/backend/tests/test_recorder.py new file mode 100644 index 00000000..751f75e2 --- /dev/null +++ b/backend/tests/test_recorder.py @@ -0,0 +1,257 @@ +"""Tests for per-profile X display recording and its HTTP contract.""" + +from __future__ import annotations + +import json +import subprocess +from datetime import datetime, timezone +from pathlib import Path +from unittest.mock import MagicMock + +import pytest +from starlette.testclient import TestClient + +from backend import recorder +from backend.browser_manager import RunningProfile +from backend.recorder import ( + RecordingConflictError, + RecordingManager, + RecordingUnavailableError, + RecordingValidationError, +) + +PROFILE_ID = "profile-1" + + +@pytest.fixture() +def recording_fixture( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +): + process = MagicMock() + process.poll.return_value = None + process.wait.return_value = 0 + process.stdin = MagicMock() + + popen = MagicMock() + + def spawn(command: list[str], **kwargs): + Path(command[-1]).write_bytes(b"mock mp4") + return process + + popen.side_effect = spawn + probe = MagicMock( + return_value=subprocess.CompletedProcess( + args=["ffprobe"], + returncode=0, + stdout=json.dumps({"format": {"duration": "8.25"}}), + stderr="", + ) + ) + monkeypatch.setattr(recorder.subprocess, "Popen", popen) + monkeypatch.setattr(recorder.subprocess, "run", probe) + + running = RunningProfile( + profile_id=PROFILE_ID, + context=object(), + cdp_port=9222, + display=None, + ws_port=6103, + ) + manager = RecordingManager( + lambda profile_id: running if profile_id == PROFILE_ID else None, + recordings_dir=tmp_path / "recordings", + screen_size_getter=lambda _profile_id: (1280, 720), + now=lambda: datetime(2026, 9, 12, 12, 30, tzinfo=timezone.utc), + ) + return manager, popen, probe, process + + +def test_recording_lifecycle_spawns_expected_ffmpeg_and_finalizes( + recording_fixture, +) -> None: + manager, popen, probe, process = recording_fixture + + started = manager.start(PROFILE_ID) + + assert started.state == "recording" + assert started.recording_id is not None + command = popen.call_args.args[0] + assert command[command.index("-f") + 1] == "x11grab" + assert command[command.index("-framerate") + 1] == "15" + assert command[command.index("-video_size") + 1] == "1280x720" + assert command[command.index("-i") + 1] == ":103.0" + assert command[command.index("-t") + 1] == "600" + assert command[command.index("-c:v") + 1] == "libx264" + assert command[command.index("-pix_fmt") + 1] == "yuv420p" + assert "-an" in command + assert popen.call_args.kwargs == { + "stdin": subprocess.PIPE, + "stdout": subprocess.DEVNULL, + "stderr": subprocess.DEVNULL, + } + assert manager.status(PROFILE_ID) == started + + completed = manager.stop(PROFILE_ID) + + process.stdin.write.assert_called_once_with(b"q\n") + process.stdin.flush.assert_called_once_with() + process.wait.assert_called_once_with(timeout=10) + assert completed.recording_id == started.recording_id + assert completed.duration_s == pytest.approx(8.25) + assert completed.path.parent.name == PROFILE_ID + assert manager.recording_path(completed.recording_id) == completed.path + assert manager.status(PROFILE_ID).state == "stopped" + assert probe.call_args.args[0][0] == "ffprobe" + + +def test_only_one_recording_can_be_active_per_profile(recording_fixture) -> None: + manager, popen, _probe, _process = recording_fixture + manager.start(PROFILE_ID) + + with pytest.raises(RecordingConflictError, match="active recording"): + manager.start(PROFILE_ID) + + popen.assert_called_once() + + +def test_recording_requires_a_running_private_display(tmp_path: Path) -> None: + native = RunningProfile( + profile_id=PROFILE_ID, + context=object(), + cdp_port=9222, + display=None, + ws_port=None, + ) + manager = RecordingManager( + lambda _profile_id: native, + recordings_dir=tmp_path / "recordings", + ) + + with pytest.raises(RecordingUnavailableError, match="no private X display"): + manager.start(PROFILE_ID) + + +@pytest.mark.parametrize( + ("probe_result", "message"), + [ + ( + subprocess.CompletedProcess( + args=["ffprobe"], + returncode=1, + stdout="", + stderr="invalid data", + ), + "corrupt", + ), + ( + subprocess.CompletedProcess( + args=["ffprobe"], + returncode=0, + stdout=json.dumps({"format": {"duration": "0.49"}}), + stderr="", + ), + "at least 0.5 seconds", + ), + ], +) +def test_validation_rejects_and_deletes_unusable_recordings( + recording_fixture, + probe_result: subprocess.CompletedProcess[str], + message: str, +) -> None: + manager, _popen, probe, _process = recording_fixture + manager.start(PROFILE_ID) + probe.return_value = probe_result + + with pytest.raises(RecordingValidationError, match=message): + manager.stop(PROFILE_ID) + + assert list(manager.recordings_dir.rglob("*.mp4")) == [] + assert manager.status(PROFILE_ID).recording_id is None + + +async def test_profile_stop_hook_finalizes_an_active_recording( + recording_fixture, + monkeypatch: pytest.MonkeyPatch, +) -> None: + manager, _popen, _probe, process = recording_fixture + monkeypatch.setattr(recorder, "recording_manager", manager) + started = manager.start(PROFILE_ID) + + completed = await recorder.on_profile_stopped(PROFILE_ID) + + assert completed is not None + assert completed.recording_id == started.recording_id + process.stdin.write.assert_called_once_with(b"q\n") + assert manager.status(PROFILE_ID).state == "stopped" + assert await recorder.on_profile_stopped(PROFILE_ID) is None + + +def test_recording_endpoints_match_client_contract( + app_client: TestClient, + recording_fixture, + monkeypatch: pytest.MonkeyPatch, +) -> None: + manager, _popen, _probe, _process = recording_fixture + monkeypatch.setattr(recorder, "recording_manager", manager) + + idle = app_client.get(f"/api/profiles/{PROFILE_ID}/recording/status") + assert idle.status_code == 200 + assert idle.json() == {"recording_id": None, "state": "stopped"} + + started = app_client.post(f"/api/profiles/{PROFILE_ID}/recording/start") + assert started.status_code == 200 + assert started.json()["state"] == "recording" + recording_id = started.json()["recording_id"] + + duplicate = app_client.post(f"/api/profiles/{PROFILE_ID}/recording/start") + assert duplicate.status_code == 409 + assert app_client.get(f"/api/profiles/{PROFILE_ID}/recording/status").json() == { + "recording_id": recording_id, + "state": "recording", + } + + stopped = app_client.post(f"/api/profiles/{PROFILE_ID}/recording/stop") + assert stopped.status_code == 200 + assert stopped.json() == { + "recording_id": recording_id, + "duration_s": 8.25, + "path": f"/api/recording/{recording_id}", + } + + downloaded = app_client.get(f"/api/recording/{recording_id}") + assert downloaded.status_code == 200 + assert downloaded.headers["content-type"] == "video/mp4" + assert downloaded.content == b"mock mp4" + + missing = app_client.get("/api/recording/not-a-recording") + assert missing.status_code == 404 + + +def test_profile_stop_finalizes_recording_before_display_teardown( + app_client: TestClient, + recording_fixture, + monkeypatch: pytest.MonkeyPatch, +) -> None: + from backend import main + + manager, _popen, _probe, process = recording_fixture + monkeypatch.setattr(recorder, "recording_manager", manager) + running = manager._running_profile_getter(PROFILE_ID) + main.browser_mgr.running[PROFILE_ID] = running + browser_stop = MagicMock() + + async def stop_profile(profile_id: str) -> None: + browser_stop(profile_id) + assert process.stdin.write.call_count == 1 + main.browser_mgr.running.pop(profile_id, None) + + monkeypatch.setattr(main.browser_mgr, "stop", stop_profile) + manager.start(PROFILE_ID) + + response = app_client.post(f"/api/profiles/{PROFILE_ID}/stop") + + assert response.status_code == 200 + browser_stop.assert_called_once_with(PROFILE_ID) + assert manager.status(PROFILE_ID).state == "stopped" diff --git a/nodes.json b/nodes.json new file mode 100644 index 00000000..daf589f5 --- /dev/null +++ b/nodes.json @@ -0,0 +1,263 @@ +{ + "nodes": [ + { + "id": "spike-cdp-input-loop", + "task": "POC0 computer-use spike on the Cloak farm (de-risk before building the executor). Standalone script, no product-code changes: read SYNAPSE_WEB_CLOAK_MANAGER_BASE_URL + SYNAPSE_WEB_CLOAK_MANAGER_AUTH_TOKEN, use the existing CloakManagerClient to find-or-create then launch a profile, connect the profile's CDP WebSocket, and run the primitive loop twice: Page.captureScreenshot -> Input.dispatchMouseEvent (move+press+release) on a button + Input.insertText into a text field on a self-contained data-URL test page that flips a DOM flag on click -> second Page.captureScreenshot -> Runtime.evaluate reads the flag. Assert: screenshots differ (hash), the click flag flipped, the typed text landed. Measure and print per-action dispatch latency and the screenshot byte size as WebP (that is the per-turn payload cost the replay loop will pay every model call). Document the env vars in the script header.", + "repo": "LoopKitchen/ai-integration", + "files": [ + "ai_ingestion/web/replay/scripts/spike_cdp_input_loop.py" + ], + "depends_on": [], + "check": "uv run python ai_ingestion/web/replay/scripts/spike_cdp_input_loop.py exits 0 and stdout prints differing before/after screenshot hashes, the flipped click flag, and a per-action latency table", + "human_gate": false, + "status": "done" + }, + { + "id": "skill-schema", + "task": "The learned-skill document contract every other node reads — single-sourced here. ai_ingestion/web/skills/schema.py: SkillDoc Pydantic model — id, name, description, params[] (name, required, default, secret_ref for credential-backed params — NEVER the secret value), steps[] each carrying {action: navigate|click|type|scroll|wait|extract|assert|confirm, ordered target candidates (selector -> accessible-text/role -> coordinates-last-resort), preconditions[], postconditions[], confirm_first, consequential}, provenance {teach_session_id, recording_ref}, version, status draft->approved->deprecated. store.py: versioned on-disk CRUD under the teach data dir (draft write, approve, list, get, version diff); approve() validates that every consequential step is confirm_first and that no field matches a credential pattern (password/otp/token literals). Tests cover schema round-trip, approval rejection of creds and coords-only consequential steps, and version diffing.", + "repo": "LoopKitchen/ai-integration", + "files": [ + "ai_ingestion/web/skills/__init__.py", + "ai_ingestion/web/skills/schema.py", + "ai_ingestion/web/skills/store.py", + "tests/web/skills/test_skill_schema.py", + "tests/web/skills/test_skill_store.py" + ], + "depends_on": [], + "check": "uv run pytest tests/web/skills/test_skill_schema.py tests/web/skills/test_skill_store.py -q green; uv run ruff check ai_ingestion/web/skills", + "human_gate": false, + "status": "done" + }, + { + "id": "cloak-client-ext", + "task": "Extend the existing CloakManagerClient with the control-lease + screen-recording wire surface — this file is the SINGLE SOURCE OF TRUTH for the wire contract; mgr-control-lease and mgr-screen-recorder implement against it verbatim. New methods: acquire_control(profile_id, holder: 'agent'|'human', ttl_s) -> lease dict {lease_id, state, holder, expires_at}, raising CloakLeaseConflictError on HTTP 409 (already held by another holder); release_control(profile_id, lease_id); control_status(profile_id). Recording: start_recording(profile_id); stop_recording(profile_id) -> {recording_id, duration_s, path}; recording_status(profile_id); download_recording(recording_id) -> bytes-or-URL. Typed errors CloakLeaseConflictError + CloakRecordingError; stdlib+httpx only (web/ layer rule). Unit tests fake the transport (httpx MockTransport) covering the 200 path, the 409 path, and status endpoints.", + "repo": "LoopKitchen/ai-integration", + "files": [ + "ai_ingestion/web/browser/cloak/manager_client.py", + "tests/web/test_cloak_control_client.py" + ], + "depends_on": [], + "check": "uv run pytest tests/web/test_cloak_control_client.py -q green; uv run ruff check ai_ingestion/web/browser/cloak/manager_client.py", + "human_gate": false, + "status": "done" + }, + { + "id": "mgr-control-lease", + "task": "Exclusive per-profile control lease on the Cloak manager (AGENT/HUMAN/IDLE) — the 'Take control' moment from the reference UX (Operator/Grok/Devin patterns): the human takes over the remote screen for logins, CAPTCHAs, and irreversible moments instead of pasting credentials, then hands back. backend/control_lease.py: lease FSM — IDLE->AGENT, IDLE->HUMAN, AGENT->HUMAN (human takeover always wins, agent must yield), HUMAN->AGENT (reacquire), TTL with auto-expiry back to IDLE, release-on-profile-stop, single-flight per profile, release requires the lease_id. backend/teach_replay_api.py: a shared APIRouter the teach/replay feature mounts — this node adds POST /api/profiles/{id}/control/acquire {holder, ttl_s} -> {lease:{lease_id, state, holder, expires_at}} returning HTTP 409 when held by the other class, POST /control/release {lease_id}, GET /control/status; wire contract is single-sourced in the ai-integration manager client — mirror it verbatim. backend/main.py: include the router AND make the /api/profiles/{id}/vnc proxy lease-aware — while the holder is AGENT the filtered RFB client stream drops PointerEvent/KeyEvent/ClientCutText entirely (view frames still flow, so the operator watches but cannot touch), while HUMAN they pass; document that CDP input gating stays cooperative (the synapse executor must not dispatch while the lease is not AGENT). backend/tests/test_control_lease.py: FSM transitions incl. takeover, TTL expiry, 409 conflict, and an RFB-gating unit test proving input types are dropped while AGENT-held.", + "repo": "LoopKitchen/CloakBrowser-Manager", + "files": [ + "backend/control_lease.py", + "backend/teach_replay_api.py", + "backend/main.py", + "backend/tests/test_control_lease.py" + ], + "depends_on": [], + "check": "python -m pytest backend/tests/test_control_lease.py -q green", + "human_gate": false, + "status": "done" + }, + { + "id": "mgr-screen-recorder", + "task": "Per-profile screen recording of the private X display (the grok capture path). backend/recorder.py: spawn ffmpeg x11grab on the running profile's display (RunningProfile.display / ws_port display index) at 15 FPS H.264 MP4 into a recordings dir keyed by profile_id+timestamp; hard cap 600s; finalize the MP4 cleanly on stop; auto-stop when the profile stops; one active recording per profile. After stop, ffprobe-validate (reject <0.5s or corrupt) before reporting success. Recording endpoints are registered INSIDE this module by importing the shared router from teach_replay_api (POST /api/profiles/{id}/recording/start, /stop, GET /recording/status, GET /recording/{recording_id} for the MP4 download) — wire contract single-sourced in the ai-integration manager client, mirror verbatim. backend/tests/test_recorder.py: ffmpeg/ffprobe subprocess calls mocked; lifecycle, one-at-a-time enforcement, validation rejection, and endpoint tests via TestClient.", + "repo": "LoopKitchen/CloakBrowser-Manager", + "files": [ + "backend/recorder.py", + "backend/tests/test_recorder.py" + ], + "depends_on": [ + "mgr-control-lease" + ], + "check": "python -m pytest backend/tests/test_recorder.py -q green", + "human_gate": false, + "status": "done" + }, + { + "id": "teach-store", + "task": "Teach-session domain layer. queue.py: durable signed queue — learn entries land as JSON under teach_sessions/queues//pending/.json with an HMAC-SHA256 signature over the canonical JSON (key from env TEACH_QUEUE_HMAC_KEY, documented in module header); a '.prompt-delivered' marker is written after the learner is dispatched; claim uses flock with a 12h lease and lease-free claims older than 24h are treated abandoned. store.py: sqlite records for teach sessions and learning runs (session_id, profile_id/name, operator intent text, recording_ref, status, produced skill_id). service.py: start_teach(profile_name, intent) -> find_profile_by_name on the Cloak farm -> launch -> acquire the control lease holder='human' (the operator demonstrates through noVNC; agent-side input is blocked because the executor refuses to run without holding AGENT) -> start_recording -> session handle; stop_teach -> stop_recording -> release the lease -> write the signed learn entry into pending; status() surfaces recording state and produced skill. Tests fake the manager client and use tmp dirs.", + "repo": "LoopKitchen/ai-integration", + "files": [ + "ai_ingestion/web/teach/__init__.py", + "ai_ingestion/web/teach/queue.py", + "ai_ingestion/web/teach/store.py", + "ai_ingestion/web/teach/service.py", + "tests/web/teach/test_teach_queue.py", + "tests/web/teach/test_teach_service.py" + ], + "depends_on": [ + "skill-schema", + "cloak-client-ext" + ], + "check": "uv run pytest tests/web/teach -q green; uv run ruff check ai_ingestion/web/teach", + "human_gate": false, + "status": "done" + }, + { + "id": "replay-gates", + "task": "Run-admission and approval policy for replay. gates.py: validate_run(skill, params, idempotency_key) — bind params (all required present; secret_ref params resolve to references, never values), persist the idempotency key so a retried submission returns the original run_id instead of re-executing, and enforce declared budget caps (campaign-creating skills must carry a spend-cap param that clamps the run). Per-step gate: steps with confirm_first or consequential=true require a valid approval token before dispatch. approvals.py: signed approval tokens scoped to run_id+step_index, single-use, expiring — plus approve(run, step) / is_approved(run, step). Tests cover the validation matrix, idempotent dedup on retry, spend-cap enforcement, and an unapproved consequential step blocking dispatch.", + "repo": "LoopKitchen/ai-integration", + "files": [ + "ai_ingestion/web/replay/__init__.py", + "ai_ingestion/web/replay/gates.py", + "ai_ingestion/web/replay/approvals.py", + "tests/web/replay/test_replay_gates.py" + ], + "depends_on": [ + "skill-schema" + ], + "check": "uv run pytest tests/web/replay/test_replay_gates.py -q green; uv run ruff check ai_ingestion/web/replay", + "human_gate": false, + "status": "done" + }, + { + "id": "replay-evidence", + "task": "Run evidence bundle — the audit trail the product promise rests on. evidence.py: per-run dir containing actions.jsonl (one line per executed step: timestamp, step action, resolved target, pre/post screenshot refs, gate verdict, expected-vs-actual note), screenshots/, video_ref (manager recording_id started for the run via the extended Cloak client), skill version + params hash, and manifest.json tying it together. expected-vs-actual compares each step's postcondition text against the post-action screenshot through an injectable vision-compare function (default impl goes through the service's existing Anthropic path; tests inject a fake). Also emits the compact run-summary dict the UI renders.", + "repo": "LoopKitchen/ai-integration", + "files": [ + "ai_ingestion/web/replay/evidence.py", + "tests/web/replay/test_evidence.py" + ], + "depends_on": [ + "skill-schema", + "cloak-client-ext" + ], + "check": "uv run pytest tests/web/replay/test_evidence.py -q green; uv run ruff check ai_ingestion/web/replay/evidence.py", + "human_gate": false, + "status": "done" + }, + { + "id": "teach-learner", + "task": "The learn-from-demonstration pipeline (grok flow). validate.py: claim the oldest pending queue entry via flock (12h lease; lease-free claims stale >24h are abandoned and re-queued); ffprobe-validate the MP4 (reject <0.5s or corrupt); extract PNG stills at 20% and 70% as sanity frames; files >15MB split into lossless keyframe-aligned segments of ~12MB. watch.py: dispatch the recording (or its segments) to a vision-analysis call carrying the fixed watch contract verbatim — report starting state, chronological action list, ending state, per-action timing, exact non-secret typed text (credential-looking strings are NEVER transcribed — redacted as [secret]), and URLs/page-titles/UI-controls observed; plus a read-only Chrome-history slice restricted to the recording window via CDP /json/list targets on the profile (history wins for URLs, video wins for in-page actions). learner.py: merge the watch output + history slice into a SkillDoc draft under skill-schema rules — parameterize inputs, prefer stable targets over coordinates, mark consequential steps confirm_first, never embed credentials or harness mechanics (no CDP ports, no profile paths); persist the draft via the skill store linked to the teach session; the claimed video is deleted ONLY after the skill draft is persisted.", + "repo": "LoopKitchen/ai-integration", + "files": [ + "ai_ingestion/web/teach/validate.py", + "ai_ingestion/web/teach/watch.py", + "ai_ingestion/web/teach/learner.py", + "tests/web/teach/test_teach_validate.py", + "tests/web/teach/test_teach_learner.py" + ], + "depends_on": [ + "teach-store", + "skill-schema" + ], + "check": "uv run pytest tests/web/teach -q green (synthetic-fixture MP4 path, or an explicit skip flag when ffmpeg is absent); uv run ruff check ai_ingestion/web/teach; uv run python -m pytest tests/test_import_rules.py -x green", + "human_gate": false, + "status": "todo" + }, + { + "id": "replay-executor", + "task": "The computer-use loop that replays a SkillDoc on a live Cloak profile. cdp_input.py: CDP input driver over the profile's CDP WebSocket — dispatchMouseEvent (move/press/release + wheel scroll), dispatchKeyEvent, insertText, Page.captureScreenshot returning WebP, and a Runtime.evaluate helper. target_resolver.py: resolve a step's ordered target candidates (selector -> accessible text/role via Runtime.evaluate -> coordinates fallback) into viewport coordinates. handoff.py: HITL arbitration — pause the loop, release the AGENT lease via the extended client, emit a handoff event (reason + current screenshot ref), await the HUMAN lease release, reacquire AGENT, then force a FRESH screenshot before the next model turn. executor.py: acquire the AGENT lease -> start the run recording -> for each step: run gates (a consequential step blocks until an approval token exists), resolve the target, screenshot -> model turn (step context + accumulated action log) -> dispatch the returned action batch -> append the post-action WebP screenshot into the next model turn -> write the evidence step -> repeat until done/fail/handoff; on any terminal state release the lease and finalize the evidence bundle (video_ref from the run recording). Never logs secrets or forwards credential values to the model.", + "repo": "LoopKitchen/ai-integration", + "files": [ + "ai_ingestion/web/replay/cdp_input.py", + "ai_ingestion/web/replay/target_resolver.py", + "ai_ingestion/web/replay/handoff.py", + "ai_ingestion/web/replay/executor.py", + "tests/web/replay/test_cdp_input.py", + "tests/web/replay/test_executor.py" + ], + "depends_on": [ + "spike-cdp-input-loop", + "skill-schema", + "cloak-client-ext", + "replay-gates" + ], + "check": "uv run pytest tests/web/replay/test_executor.py tests/web/replay/test_cdp_input.py -q green (fake CDP transport + scripted model); uv run ruff check ai_ingestion/web/replay", + "human_gate": false, + "status": "todo" + }, + { + "id": "agent-skill-tools", + "task": "Agent-facing tool surface over learned skills — this is how 'our agent' invokes a taught task. skill_tools.py follows the claude_agent_sdk @tool pattern and guard posture of browser_tools.py: list_skills(), get_skill(skill_id), run_skill(skill_id, params, idempotency_key) — launches the replay executor against the merchant's Cloak profile and streams progress/evidence refs back, request_takeover(run_id, reason) — flips control to HUMAN via the handoff path. Refuses with actionable text when no browser/cloak surface exists for the session; never accepts or forwards raw credential values — params carry secret_refs only.", + "repo": "LoopKitchen/ai-integration", + "files": [ + "ai_ingestion/web/agent/tools/skill_tools.py", + "tests/web/test_skill_tools.py" + ], + "depends_on": [ + "replay-executor", + "skill-schema" + ], + "check": "uv run pytest tests/web/test_skill_tools.py -q green; uv run ruff check ai_ingestion/web/agent/tools/skill_tools.py; uv run python -m pytest tests/test_import_rules.py -x green", + "human_gate": false, + "status": "todo" + }, + { + "id": "teach-replay-api", + "task": "The HTTP surface — this node owns ALL new route wiring. routes/teach.py: POST /api/teach/sessions {profile_name, intent} -> start_teach; POST /api/teach/sessions/{id}/stop; GET /api/teach/sessions/{id} (recording state + produced skill_id); GET /api/teach/skills (list + detail). routes/replay.py: POST /api/skills/{id}/runs {params, idempotency_key}; GET /api/runs/{id} (status, current step, live evidence refs, pending-approval step); POST /api/runs/{id}/approve {step_index}; POST /api/runs/{id}/takeover; GET /api/runs/{id}/evidence (bundle manifest). web/app.py: import both routers and include_router them — the ONLY file that mounts new routes. Thin routes only: request-model validation then delegate to the domain services; no business logic inline.", + "repo": "LoopKitchen/ai-integration", + "files": [ + "ai_ingestion/web/app.py", + "ai_ingestion/web/routes/teach.py", + "ai_ingestion/web/routes/replay.py", + "tests/web/test_routes_teach.py", + "tests/web/test_routes_replay.py" + ], + "depends_on": [ + "teach-store", + "teach-learner", + "replay-executor", + "replay-gates", + "replay-evidence" + ], + "check": "uv run pytest tests/web/test_routes_teach.py tests/web/test_routes_replay.py -q green; uv run ruff check ai_ingestion/web/routes/teach.py ai_ingestion/web/routes/replay.py ai_ingestion/web/app.py; uv run python -m pytest tests/test_import_rules.py -x green", + "human_gate": false, + "status": "todo" + }, + { + "id": "teach-replay-ui", + "task": "Operator UI in synapse-web — the experience is the Grok Bot agent-screen pattern verified from the reference media: the screen lives as a card in/next to chat, expands to a viewer, 'Teach a task' records a demonstration, a 'Take control' handoff flips control to the human (logins/CAPTCHAs/irreversible moments — they act instead of pasting credentials), and the evidence video plays back at the end. Vendor the beautifului AgentScreen element — the registry URL currently 404s so the pasted component source in the design thread is the source of truth (synapse-web/src/components/agent-screen/AgentScreen.tsx): resting card = framed capture with hover Open; expanded viewer with 'Teach a task' (REC badge + timer, collapse keeps recording running), End, and the cursor overlay. Rebuild the trimmed FauxWindow as the no-stream placeholder in Loop tokens (three traffic-light dots, fake URL bar, skeleton rows — it is the empty state before a stream connects); adapt to Tailwind 4 + React 19 with no shadcn dependency. Map it: streamSrc = the Cloak profile's live view — embed the existing CloakVncView (the /ws/cloak/ proxy) inside the Screen slot, or a recorded MP4 for evidence playback. Teach a task -> POST /api/teach/sessions; REC badge mirrors live recording state; Take control -> POST /api/runs/{id}/takeover flips the lease to HUMAN and back. Grok-pattern surfaces that must exist: a chat-embedded 'Learn from demonstration' pill action on a posted recording; the screen card labeled ''s screen'; a Routines list under the screen showing learned skills with their cadence; a learned-skill card in chat rendering the SkillDoc as numbered steps plus an Inputs line plus assumption text ('I ask before checkout'); a dry-run approval card. ActionTimeline.tsx renders the run's timestamped steps (task-rows style) from GET /api/runs/{id}/evidence. ApprovalCard.tsx is the inline HITL approve gate posting /api/runs/{id}/approve. lib/teachReplayApi.ts is the typed client for the teach + replay routes. TeachReplayPane.tsx wires the pane into the session view next to BrowserPane. Vitest coverage on the API client and the approval flow.", + "repo": "LoopKitchen/ai-integration", + "files": [ + "synapse-web/src/components/agent-screen/AgentScreen.tsx", + "synapse-web/src/components/agent-screen/ActionTimeline.tsx", + "synapse-web/src/components/agent-screen/ApprovalCard.tsx", + "synapse-web/src/components/agent-screen/TeachControls.tsx", + "synapse-web/src/components/agent-screen/AgentScreen.test.tsx", + "synapse-web/src/components/TeachReplayPane.tsx", + "synapse-web/src/lib/teachReplayApi.ts" + ], + "depends_on": [ + "teach-replay-api" + ], + "check": "cd synapse-web && npm run lint && npm run build clean && npm run test -- agent-screen green", + "human_gate": false, + "status": "todo" + }, + { + "id": "e2e-teach-replay", + "task": "End-to-end proof script. Run against a dev Cloak manager (CLOAK_MANAGER_BASE_URL + token envs documented in the script header): create a test profile -> teach capture — either a scripted CDP demonstration driving fixture_portal.html (a self-contained fake portal: login-ish form, a 'create campaign' button, a confirmation banner) or a canned fixture MP4 -> the learner produces a SkillDoc draft -> approve it -> a replay run executes every step through the real executor + CDP -> assert: actions.jsonl complete, before/after screenshots present, video_ref resolvable, and the fixture's confirmation DOM state reached. When no manager is reachable the same path runs against a local headed Chromium over CDP — the script prints which mode ran.", + "repo": "LoopKitchen/ai-integration", + "files": [ + "ai_ingestion/web/replay/scripts/e2e_teach_replay.py", + "ai_ingestion/web/replay/scripts/fixture_portal.html" + ], + "depends_on": [ + "teach-learner", + "replay-executor", + "agent-skill-tools", + "teach-replay-api", + "replay-evidence", + "mgr-screen-recorder", + "mgr-control-lease" + ], + "check": "uv run python ai_ingestion/web/replay/scripts/e2e_teach_replay.py exits 0 and prints PASS for teach -> learn -> replay -> evidence assertions", + "human_gate": false, + "status": "todo" + }, + { + "id": "human-gate-campaign-demo", + "task": "Human gate — run the complete demo human-in-the-loop on Mirus, the approved restaurant reporting SaaS portal. A human obtains the Mirus credentials from the Loop Connect sheet and enters them only after taking exclusive control; credentials never enter code, git, recordings, evidence, logs, or the learned SkillDoc. The doc is the checklist the operator works through: (1) teach a reporting workflow on Mirus through the AgentScreen UI, (2) review the learned SkillDoc — parameterized inputs, stable targets, every irreversible step confirm_first, zero credentials recorded, (3) watch the replay evidence — continuous video, the actions timeline, before/after shots, the approval gate firing before every irreversible action, (4) verify exclusivity — the agent paused while the operator held the lease and a fresh screenshot was taken after hand-back. Sign-off is recorded in the doc.", + "repo": "LoopKitchen/ai-integration", + "files": [ + "docs/teach-replay/demo-checklist.md" + ], + "depends_on": [ + "e2e-teach-replay", + "teach-replay-ui" + ], + "check": "human reviews the evidence bundle against demo-checklist.md and signs off — looks for: complete video, parameterized steps, the approval gate before the irreversible submit, exclusive agent/human control during takeover, and exact non-secret typed text captured", + "human_gate": true, + "status": "todo" + } + ] +}