"""AD-733: Camera frame ingestion endpoint.

POST /api/perception/camera/frame accepts a multipart JPEG, validates +
stores it via the shared :func:`_validate_and_store_attachment` chain (AD-731:
SHA-256 keyed bytes in :class:`AttachmentStore`), and broadcasts a
``vision_observation`` :class:`IntentMessage` whose params carry ONLY the
SHA ref — never the bytes.

v1 has no LLM consumer for the intent (AD-733a forward marker). Per the
dynamic intent discovery design, an unconsumed intent is broadcast and
silently dropped; the audit trail still lands in the journal.
"""
from __future__ import annotations

import asyncio
import logging
import time
from typing import Any

from fastapi import APIRouter, Depends, File, Form, HTTPException, UploadFile
from fastapi.responses import JSONResponse
from pydantic import BaseModel

from probos.routers.auth import require_crew_scope
from probos.routers.chat import _validate_and_store_attachment
from probos.routers.deps import get_runtime
from probos.types import AnchorFrame, Episode, IntentMessage
from probos.perception import VISION_OBSERVATION_DESCRIPTOR  # noqa: F401  registers descriptor

router = APIRouter(prefix="/api/perception", tags=["perception"])
logger = logging.getLogger(__name__)

# Per-session token bucket. Module-scoped: one runtime per process.
_buckets: dict[str, tuple[float, float]] = {}

# One anchor episode per session per runtime boot. Cleared on process exit.
_ANCHOR_WRITTEN: set[str] = set()


def _reset_state_for_tests() -> None:
    """Test-only helper — clears module state so each test runs in isolation."""
    _buckets.clear()
    _ANCHOR_WRITTEN.clear()


def _check_rate(session_id: str, max_fps: int, source: str = "camera") -> bool:
    """Token-bucket admission: True when a frame slot is available.

    AD-733-2: the bucket key is ``(session_id, source)``. Camera and screen
    streams have independent budgets — saturating one source does not
    starve the other. Legacy callers pass ``source="camera"`` (the default)
    so existing camera flows are byte-compatible.
    """
    key = f"{session_id}::{source}"
    now = time.monotonic()
    last, tokens = _buckets.get(key, (now, float(max_fps)))
    elapsed = now - last
    tokens = min(float(max_fps), tokens + elapsed * max_fps)
    if tokens < 1.0:
        _buckets[key] = (now, tokens)
        return False
    _buckets[key] = (now, tokens - 1.0)
    return True


async def _write_anchor_episode(
    runtime: Any,
    session_id: str,
    sha: str,
    captured_at: float,
    *,
    source: str = "camera",
    trigger_type: str | None = None,
) -> None:
    """AD-541b: anchored episode marking <source>-stream-began.

    AD-733-2: ``source`` distinguishes camera from screen anchors; the
    anchor-key includes source so each (session, source) pair writes its
    own opening episode. Default ``trigger_type`` derived from source.

    AD-744: explicit Captain shares pass ``trigger_type="captain_explicit_share"``
    so they don't collide with the ambient stream-began anchor.

    Tier-2 honest-degrade: logs WARNING on any failure, never raises.
    """
    if trigger_type is None:
        trigger_type = f"{source}_stream_began"
    anchor_key = f"{session_id}::{trigger_type}"
    if anchor_key in _ANCHOR_WRITTEN:
        return
    _ANCHOR_WRITTEN.add(anchor_key)
    episodic = getattr(runtime, "episodic_memory", None)
    if episodic is None:
        return
    try:
        episode = Episode(
            timestamp=captured_at,
            user_input="",
            outcomes=[{
                "intent": "vision_observation",
                "success": True,
                "session_id": session_id,
                "attachment_ref": sha,
                "source": source,
            }],
            reflection=(
                f"{source.capitalize()} stream began "
                f"(session={session_id[:8]}, sha={sha[:8]}). "
                "AD-733 v1 wire shape proven; consumer is the AD-733a VisionConsumer."
            ),
            source="direct",
            importance=8,
            anchors=AnchorFrame(
                channel="perception",
                trigger_type=trigger_type,
                trigger_agent="captain",
            ),
        )
        await episodic.store(episode)
    except Exception as ex:
        logger.warning(
            "AD-733 anchor episode store failed (session=%s, source=%s, sha=%s): %s; "
            "frame is still stored, observation pipeline unaffected.",
            session_id, source, sha[:8], ex,
        )


_VALID_SOURCES: frozenset[str] = frozenset({"camera", "screen"})


@router.post("/camera/frame", dependencies=[Depends(require_crew_scope)])
async def upload_camera_frame(
    file: UploadFile = File(...),
    session_id: str = Form(...),
    force: str = Form(""),
    agent_ids: str = Form(""),
    # AD-733-2: optional source discriminator. Defaults to "camera" so all
    # existing AD-733 callers are byte-compatible. ``screen`` activates the
    # AD-733-2 passive screen-share flow; values outside the allow-list
    # return HTTP 400.
    source: str = Form("camera"),
    runtime: Any = Depends(get_runtime),
) -> Any:
    cfg = getattr(runtime.config, "perception", None)
    if cfg is None or not cfg.enabled:
        return JSONResponse(status_code=503, content={"error": "perception_disabled"})

    # AD-733-2: source validation BEFORE any downstream branching so that
    # invalid values cannot poison the rate-bucket key or the descriptor
    # params. Allow-list only — defense in depth.
    if source not in _VALID_SOURCES:
        return JSONResponse(status_code=400, content={"error": "invalid_source"})

    # AD-733-2: per-source enable gate. Camera and screen are independent
    # subsystems — disabling one MUST NOT block the other.
    if source == "camera":
        if not cfg.camera.enabled:
            return JSONResponse(status_code=503, content={"error": "camera_disabled"})
        rate_cap = cfg.camera_max_fps_server
    else:  # source == "screen"
        screen_cfg = getattr(cfg, "screen", None)
        if screen_cfg is None or not screen_cfg.enabled:
            return JSONResponse(status_code=503, content={"error": "screen_disabled"})
        rate_cap = cfg.screen_max_fps_server

    # AD-744: explicit-share gate. When the master switch is False, any
    # POST that combines ``force=true`` AND non-empty ``agent_ids`` is
    # treated as an explicit-share invocation and rejected. Ambient
    # streams (no force/no agent_ids) and operator-preview force shots
    # (force only, no agent_ids) remain unaffected.
    is_forced_preflight = force.lower() in {"1", "true", "yes"}
    has_agent_ids_preflight = bool(agent_ids.strip())
    explicit_share_attempt = is_forced_preflight and has_agent_ids_preflight
    if explicit_share_attempt and not getattr(cfg, "explicit_share_enabled", True):
        return JSONResponse(
            status_code=503,
            content={"error": "explicit_share_disabled"},
        )

    if not _check_rate(session_id, rate_cap, source=source):
        return JSONResponse(
            status_code=429,
            content={"error": "rate_limited"},
            headers={"Retry-After": "1"},
        )

    blob = await file.read()
    if len(blob) > cfg.frame_max_size_bytes:
        return JSONResponse(status_code=413, content={"error": "frame_too_large"})
    if len(blob) < 12:
        return JSONResponse(status_code=400, content={"error": "frame_too_small"})

    ok, result = await _validate_and_store_attachment(
        runtime, blob, "image/jpeg",
        declared_filename=None, declared_hash_or_None=None,
        origin="perception_frame",
    )
    if not ok:
        return JSONResponse(
            status_code=result["status_code"],
            content=result["body"],
            headers=result.get("headers"),
        )

    sha = result["attachment_id"]
    captured_at = time.time()
    is_forced = force.lower() in {"1", "true", "yes"}

    # AD-742c: comma-separated agent_ids form field. When provided, the
    # frame is bound to those specific agents and the consumer restricts
    # fan-out. AD-731 invariant preserved: agent_ids is a STRING list,
    # NOT image bytes.
    bound_agent_ids: list[str] = []
    if agent_ids.strip():
        bound_agent_ids = [
            aid.strip() for aid in agent_ids.split(",") if aid.strip()
        ]

    # AD-731 invariant: refs only — NEVER inline bytes in IntentMessage.params.
    # BF-302: ``force`` carries Captain's explicit "describe this frame even
    # if the supervisor would normally drop it" intent. Used by the operator
    # preview panel for testing the pipeline without waiting on novelty.
    # AD-733-2: ``source`` discriminates camera vs screen so downstream
    # consumers can filter (forward marker AD-733-2-1 covers per-source
    # novelty thresholds; v1 fan-out is source-agnostic).
    _params: dict[str, Any] = {
        "attachment_ref": sha,
        "mime": "image/jpeg",
        "captured_at": captured_at,
        "source": source,
        "session_id": session_id,
        "force": is_forced,
    }
    if bound_agent_ids:
        _params["bound_agent_ids"] = bound_agent_ids
    msg = IntentMessage(
        intent="vision_observation",
        params=_params,
    )
    try:
        await runtime.intent_bus.broadcast(msg)
    except Exception as ex:
        logger.warning(
            "AD-733 intent_bus.broadcast failed (session=%s, source=%s, sha=%s): %s; "
            "frame is still stored, observation pipeline unaffected.",
            session_id, source, sha[:8], ex,
        )

    # AD-744: Captain-initiated explicit shares emit a distinct anchor
    # trigger_type so they don't collide with the ambient stream-began
    # anchor (which fires once per session). Detection: force=true +
    # non-empty agent_ids — works for camera AND screen sources.
    anchor_trigger: str | None = None
    if explicit_share_attempt:
        anchor_trigger = "captain_explicit_share"
    await _write_anchor_episode(
        runtime, session_id, sha, captured_at,
        source=source, trigger_type=anchor_trigger,
    )

    return {"ok": True, "attachment_ref": sha, "captured_at": captured_at}


@router.get("/recent", dependencies=[Depends(require_crew_scope)])
async def get_recent_observations(limit: int = 8, runtime: Any = Depends(get_runtime)) -> Any:
    """BF-303 / BF-306: return the most recent vision observations across all
    per-agent working memories, newest first, AND the most recent supervisor
    decisions (kept frames + dropped frames with reason + novelty score).
    Operator-facing debug surface for the preview panel.
    """
    from probos.perception.consumer import _WORKING_MEMORIES

    items: list[dict[str, Any]] = []
    for agent_id, wm in _WORKING_MEMORIES.items():
        for obs in wm.entries():
            items.append(
                {
                    "agent_id": agent_id,
                    "timestamp": obs.timestamp,
                    "attachment_ref": obs.attachment_ref,
                    "description": obs.description,
                    "novelty_score": obs.novelty_score,
                    "subject_identity": obs.subject_identity,
                    "session_id": obs.session_id,
                }
            )
    items.sort(key=lambda it: it["timestamp"], reverse=True)

    # BF-306: surface supervisor decisions so the operator can SEE why
    # frames are or aren't being described. Pulled from the runtime's
    # consumer instance — honest-degrade to empty list when absent.
    decisions: list[dict[str, Any]] = []
    consumer = getattr(runtime, "vision_consumer", None)
    if consumer is not None and hasattr(consumer, "recent_decisions"):
        try:
            decisions = consumer.recent_decisions(limit=max(1, min(limit, 32)))
        except Exception:
            decisions = []

    return {
        "observations": items[: max(1, min(limit, 32))],
        "recent_decisions": decisions,
    }


# AD-733c-2 (Wave 172) - Mode status + manual override.

@router.get("/mode", dependencies=[Depends(require_crew_scope)])
async def get_perception_mode(runtime: Any = Depends(get_runtime)) -> Any:
    """Return the current PerceptionMode, when it transitioned, the last DM
    activity, the three preset bundles, and the most recent transitions.
    """
    controller = getattr(runtime, "perception_mode_controller", None)
    if controller is None:
        return JSONResponse(
            status_code=503,
            content={"error": "perception_mode_controller_unavailable"},
        )
    from probos.perception.mode_controller import PRESETS

    presets = {
        m.value: {
            "min_interval_seconds": p.min_interval_seconds,
            "novelty_threshold": p.novelty_threshold,
            "baseline_max_age_seconds": p.baseline_max_age_seconds,
        }
        for m, p in PRESETS.items()
    }
    transitions = [
        {
            "at": t.at,
            "from_mode": t.from_mode.value,
            "to_mode": t.to_mode.value,
            "trigger": t.trigger,
        }
        for t in controller.recent_transitions(limit=3)
    ]
    # AD-733c-5: per-agent modes for HXI rendering. Defaults to an empty
    # dict when the registry is unwired (back-compat for legacy single-
    # controller deployments).
    per_agent: dict[str, str] = {}
    _engagement = getattr(runtime, "perception_engagement_registry", None)
    if _engagement is not None:
        try:
            per_agent = _engagement.current_modes()
        except Exception:
            logger.debug(
                "AD-733c-5: engagement registry current_modes raised",
                exc_info=True,
            )
    return {
        "mode": controller.current_mode.value,
        "since": controller.mode_since,
        "last_dm_activity": controller.last_dm_activity_at,
        "presets": presets,
        "transitions": transitions,
        "per_agent": per_agent,
    }


class _PerceptionModeRequest(BaseModel):
    mode: str


@router.post("/mode", dependencies=[Depends(require_crew_scope)])
async def post_perception_mode(
    body: _PerceptionModeRequest,
    runtime: Any = Depends(get_runtime),
) -> Any:
    """Manual operator override. Trigger='manual' bypasses the programmatic cooldown."""
    controller = getattr(runtime, "perception_mode_controller", None)
    if controller is None:
        return JSONResponse(
            status_code=503,
            content={"error": "perception_mode_controller_unavailable"},
        )
    from probos.perception.mode_controller import Mode
    try:
        target = Mode(body.mode.strip().lower())
    except ValueError:
        return JSONResponse(
            status_code=400, content={"error": "invalid_mode", "value": body.mode},
        )
    changed = controller.transition_to(target, trigger="manual")
    return {"ok": True, "mode": controller.current_mode.value, "changed": changed}


# AD-733c-3 (Wave 172) - Wake-word engage.

class _PerceptionEngageRequest(BaseModel):
    agent: str | None = None        # callsign of the targeted agent (informational)
    phrase: str | None = None       # the matched phrase (informational, for logs)
    source: str = "wake_word"       # "wake_word" | "manual"


@router.post("/engage", dependencies=[Depends(require_crew_scope)])
async def post_perception_engage(
    body: _PerceptionEngageRequest,
    runtime: Any = Depends(get_runtime),
) -> Any:
    """AD-733c-3: flip the mode controller to ENGAGED on a wake-word event.

    Body fields are informational (logged); the controller only needs the
    side effect. 5s cooldown enforced at the controller level.

    AD-733c-5: when ``body.agent`` resolves to a registered per-agent
    controller (either by agent_id or by callsign), the engagement is
    scoped to that agent only. Otherwise falls back to the legacy
    singleton (runtime-wide engagement).
    """
    controller = getattr(runtime, "perception_mode_controller", None)
    if controller is None:
        return JSONResponse(
            status_code=503,
            content={"error": "perception_mode_controller_unavailable"},
        )
    if body.source not in ("wake_word", "manual"):
        return JSONResponse(
            status_code=400, content={"error": "invalid_source", "value": body.source},
        )

    # AD-733c-5: prefer per-agent routing.
    routed_agent: str | None = None
    _engagement = getattr(runtime, "perception_engagement_registry", None)
    if _engagement is not None and body.agent:
        # Try agent_id first.
        _per = _engagement.get(body.agent)
        if _per is None:
            # Try callsign → agent_id resolution.
            cs_reg = getattr(runtime, "callsign_registry", None)
            if cs_reg is not None:
                try:
                    resolved = cs_reg.resolve(body.agent)
                except Exception:
                    resolved = None
                if resolved is not None:
                    _aid = getattr(resolved, "id", None) or getattr(
                        resolved, "agent_id", None
                    )
                    if _aid is not None:
                        _per = _engagement.get(_aid)
                        if _per is not None:
                            routed_agent = _aid
        else:
            routed_agent = body.agent
        # When agent was specified but no per-agent controller resolved,
        # return 404 honest-degrade per AD-733c-5 contract.
        if _per is None:
            return JSONResponse(
                status_code=404,
                content={"error": "unknown_agent", "value": body.agent},
            )
        controller = _per

    transitioned, reason = controller.note_wake_word()
    logger.info(
        "AD-733c-3: engage agent=%s phrase=%s source=%s transitioned=%s reason=%s "
        "routed_agent=%s",
        (body.agent or "*")[:32], (body.phrase or "*")[:64], body.source,
        transitioned, reason, routed_agent or "*",
    )
    return {
        "ok": True,
        "mode": controller.current_mode.value,
        "transitioned": transitioned,
        "reason": reason,
        "agent_id": routed_agent,
    }


# AD-733c-7 (Wave 176) — Silero VAD secondary engagement trigger.

class _PerceptionVoiceActivityRequest(BaseModel):
    agent: str | None = None        # agent_id or callsign of the targeted agent
    source: str = "vad"             # "vad" | "manual"


@router.post(
    "/voice-activity", dependencies=[Depends(require_crew_scope)]
)
async def post_perception_voice_activity(
    body: _PerceptionVoiceActivityRequest,
    runtime: Any = Depends(get_runtime),
) -> Any:
    """AD-733c-7: register a Silero-VAD speech-detected event.

    Body carries only metadata — audio bytes never leave the browser.
    Honest-degrades to 503 when ``vad_engagement_enabled=False``;
    routes through the AD-733c-5 per-agent registry when ``agent`` is
    specified (mirrors AD-733c-3 /engage shape).
    """
    cfg = getattr(runtime, "config", None)
    pcfg = getattr(cfg, "perception", None) if cfg is not None else None
    if pcfg is None or not getattr(pcfg, "vad_engagement_enabled", False):
        return JSONResponse(
            status_code=503,
            content={"error": "vad_engagement_disabled"},
        )

    controller = getattr(runtime, "perception_mode_controller", None)
    if controller is None:
        return JSONResponse(
            status_code=503,
            content={"error": "perception_mode_controller_unavailable"},
        )
    if body.source not in ("vad", "manual"):
        return JSONResponse(
            status_code=400,
            content={"error": "invalid_source", "value": body.source},
        )

    # AD-733c-5: prefer per-agent routing.
    routed_agent: str | None = None
    _engagement = getattr(runtime, "perception_engagement_registry", None)
    if _engagement is not None and body.agent:
        _per = _engagement.get(body.agent)
        if _per is None:
            cs_reg = getattr(runtime, "callsign_registry", None)
            if cs_reg is not None:
                try:
                    resolved = cs_reg.resolve(body.agent)
                except Exception:
                    resolved = None
                if resolved is not None:
                    _aid = getattr(resolved, "id", None) or getattr(
                        resolved, "agent_id", None
                    )
                    if _aid is not None:
                        _per = _engagement.get(_aid)
                        if _per is not None:
                            routed_agent = _aid
        else:
            routed_agent = body.agent
        if _per is None:
            return JSONResponse(
                status_code=404,
                content={"error": "unknown_agent", "value": body.agent},
            )
        controller = _per

    transitioned, reason = controller.note_voice_activity()
    logger.info(
        "AD-733c-7: voice-activity agent=%s source=%s transitioned=%s reason=%s "
        "routed_agent=%s",
        (body.agent or "*")[:32], body.source,
        transitioned, reason, routed_agent or "*",
    )
    return {
        "ok": True,
        "mode": controller.current_mode.value,
        "transitioned": transitioned,
        "reason": reason,
        "agent_id": routed_agent,
    }


@router.post("/identity/enroll", dependencies=[Depends(require_crew_scope)])
async def enroll_identity(
    file: UploadFile = File(...),
    runtime: Any = Depends(get_runtime),
) -> dict[str, Any]:
    """AD-742b: enroll the Captain's reference face.

    Accepts a multipart image upload (JPEG/PNG). Computes the 512-d
    embedding via facenet-pytorch, persists to ``data/captain_identity.json``,
    and discards the image bytes. The reference photo is NOT stored.
    """
    resolver = getattr(runtime, "identity_resolver", None)
    if resolver is None:
        raise HTTPException(
            status_code=503,
            detail="AD-742b: IdentityResolver not wired. Check that perception.identity_resolver_enabled is True and facenet-pytorch is installed.",
        )
    content = await file.read()
    if not content or len(content) > 10 * 1024 * 1024:  # 10 MB cap
        raise HTTPException(status_code=400, detail="empty or oversized image")
    try:
        # Offload sync inference from the event loop.
        loop = asyncio.get_running_loop()
        await loop.run_in_executor(None, resolver.enroll, content)
    except ValueError as exc:
        raise HTTPException(status_code=422, detail=str(exc))
    return {"enrolled": True, "model_id": "facenet-pytorch-vggface2-1.0"}


@router.delete("/identity", dependencies=[Depends(require_crew_scope)])
async def revoke_identity(
    runtime: Any = Depends(get_runtime),
) -> dict[str, Any]:
    """AD-742b: delete the enrolled face embedding."""
    resolver = getattr(runtime, "identity_resolver", None)
    if resolver is None:
        return {"removed": False, "reason": "resolver not wired"}
    removed = resolver.revoke()
    return {"removed": removed}


@router.get("/identity", dependencies=[Depends(require_crew_scope)])
async def get_identity_status(
    runtime: Any = Depends(get_runtime),
) -> dict[str, Any]:
    """AD-742b: report enrollment status (no embedding returned)."""
    resolver = getattr(runtime, "identity_resolver", None)
    if resolver is None:
        return {"enrolled": False, "resolver_wired": False}
    return {
        "enrolled": resolver.is_enrolled(),
        "resolver_wired": True,
        "model_id": "facenet-pytorch-vggface2-1.0",
    }


@router.get("/budget", dependencies=[Depends(require_crew_scope)])
async def get_vision_budget(
    runtime: Any = Depends(get_runtime),
) -> dict[str, Any]:
    """AD-742e (Wave 174): vision LLM call budget telemetry.

    Returns per-tier (vision / vision_fast) call counts for the current
    session AND for today (UTC). Plus a next-allowed-in estimate based
    on the supervisor's min-interval floor.
    """
    consumer = getattr(runtime, "vision_consumer", None)
    if consumer is None:
        return {
            "session_id": "",
            "calls_this_session": {"vision": 0, "vision_fast": 0},
            "calls_today": {"vision": 0, "vision_fast": 0},
            "total_session": 0,
            "total_today": 0,
            "session_ceiling_estimate": 0,
            "next_allowed_in_seconds": 0.0,
            "consumer_wired": False,
        }
    snapshot = consumer.get_budget_snapshot()
    snapshot["consumer_wired"] = True
    return snapshot


# AD-742c (Wave 176) — Per-agent camera binding.

class _CameraBindingRequest(BaseModel):
    agent_id: str
    device_id: str


@router.get('/cameras', dependencies=[Depends(require_crew_scope)])
async def get_cameras(runtime: Any = Depends(get_runtime)) -> Any:
    """AD-742c: list current camera bindings per crew agent.

    The actual device enumeration happens browser-side (the runtime has
    no view of the operator's hardware). The response exposes the
    persisted bindings so the HXI can render the binding UI.
    """
    bindings: dict[str, str] = {}
    profile_store = getattr(runtime, 'profile_store', None)
    registry = getattr(runtime, 'registry', None)
    ontology = getattr(runtime, 'ontology', None)
    if profile_store is not None and registry is not None:
        from probos.crew_utils import is_crew_agent
        try:
            for agent in registry.all():
                if not is_crew_agent(agent, ontology):
                    continue
                profile = profile_store.get(agent.id)
                if profile is None or profile.perception is None:
                    continue
                bindings[agent.id] = profile.perception.camera_device_id or ''
        except Exception:
            logger.warning(
                'AD-742c: enumerating crew bindings failed', exc_info=True,
            )
    return {'bindings': bindings}


@router.post('/cameras/binding', dependencies=[Depends(require_crew_scope)])
async def post_camera_binding(
    body: _CameraBindingRequest,
    runtime: Any = Depends(get_runtime),
) -> Any:
    """AD-742c: bind `agent_id` to the browser-side `device_id`.

    Empty `device_id` clears the binding (agent falls back to shared
    default camera). 404 honest-degrade when `agent_id` has no profile.
    """
    profile_store = getattr(runtime, 'profile_store', None)
    if profile_store is None:
        return JSONResponse(
            status_code=503, content={'error': 'profile_store_unavailable'},
        )
    profile = profile_store.get(body.agent_id)
    if profile is None:
        return JSONResponse(
            status_code=404,
            content={'error': 'unknown_agent', 'value': body.agent_id},
        )
    profile.perception.camera_device_id = body.device_id.strip()
    try:
        profile_store.update(profile)
    except Exception:
        logger.warning(
            'AD-742c: profile_store update failed agent=%s', body.agent_id,
            exc_info=True,
        )
        return JSONResponse(
            status_code=500, content={'error': 'persist_failed'},
        )
    logger.info(
        'AD-742c: bound agent=%s camera_device_id=%s',
        body.agent_id, body.device_id or '(cleared)',
    )
    return {
        'ok': True,
        'agent_id': body.agent_id,
        'device_id': profile.perception.camera_device_id,
    }


# AD-746 Layer 2 — Per-agent source-binding endpoints.

class _SourceBindingRequest(BaseModel):
    agent_id: str
    sources: list[str]


@router.get('/sources', dependencies=[Depends(require_crew_scope)])
async def get_source_bindings(runtime: Any = Depends(get_runtime)) -> Any:
    """AD-746 Layer 2: list current per-agent source bindings.

    Returns ``{bindings: {agent_id: list[str]}}`` where each value is
    the agent's ``bound_sources`` profile field. Agents with no
    explicit binding return the default ``["camera", "screen"]``.
    """
    bindings: dict[str, list[str]] = {}
    profile_store = getattr(runtime, 'profile_store', None)
    registry = getattr(runtime, 'registry', None)
    ontology = getattr(runtime, 'ontology', None)
    if profile_store is not None and registry is not None:
        from probos.crew_utils import is_crew_agent
        try:
            for agent in registry.all():
                if not is_crew_agent(agent, ontology):
                    continue
                profile = profile_store.get(agent.id)
                if profile is None or profile.perception is None:
                    continue
                bindings[agent.id] = list(profile.perception.bound_sources)
        except Exception:
            logger.warning(
                'AD-746: enumerating source bindings failed', exc_info=True,
            )
    return {'bindings': bindings}


@router.post('/sources/binding', dependencies=[Depends(require_crew_scope)])
async def post_source_binding(
    body: _SourceBindingRequest,
    runtime: Any = Depends(get_runtime),
) -> Any:
    """AD-746 Layer 2: bind ``agent_id`` to a list of visual sources.

    Each source must be in ``_VALID_SOURCES`` (camera/screen).
    Unknown sources rejected with HTTP 400. 404 on unknown agent.
    """
    # Validate against the same frozenset that gates frame upload.
    cleaned: list[str] = []
    for s in body.sources:
        s = (s or '').strip()
        if not s:
            continue
        if s not in _VALID_SOURCES:
            return JSONResponse(
                status_code=400,
                content={'error': 'invalid_source', 'value': s},
            )
        if s not in cleaned:
            cleaned.append(s)
    profile_store = getattr(runtime, 'profile_store', None)
    if profile_store is None:
        return JSONResponse(
            status_code=503, content={'error': 'profile_store_unavailable'},
        )
    profile = profile_store.get(body.agent_id)
    if profile is None:
        return JSONResponse(
            status_code=404,
            content={'error': 'unknown_agent', 'value': body.agent_id},
        )
    profile.perception.bound_sources = cleaned
    try:
        profile_store.update(profile)
    except Exception:
        logger.warning(
            'AD-746: profile_store update failed agent=%s', body.agent_id,
            exc_info=True,
        )
        return JSONResponse(
            status_code=500, content={'error': 'persist_failed'},
        )
    logger.info(
        'AD-746 Layer 2: bound agent=%s sources=%s',
        body.agent_id, cleaned,
    )
    return {
        'ok': True,
        'agent_id': body.agent_id,
        'sources': cleaned,
    }
