"""
bridge_client.py — SINGLE-FILE Cowork-side client for the cowork-to-code bridge.

This is a self-contained, zero-dependency (stdlib-only) copy of the bridge
client, meant to be dropped into a Cowork sandbox with ONE file fetch — no
`pip install`, no package, no outbound network beyond fetching this one file.
(The Cowork sandbox blocks outbound egress and prompts per fetch, so one file =
one prompt.)

It is kept in sync with `cowork_to_code_bridge/client.py`. If you have the full
package installed, prefer `from cowork_to_code_bridge import call_remote`.

Usage in a Cowork session:

    import os
    os.environ["BRIDGE_ROOT"] = "/Users/you/.cowork-to-code-bridge"  # from your Mac's .env
    from bridge_client import call_remote, daemon_alive
    print(daemon_alive())
    r = call_remote("scripts/run_claude.sh",
                    args=["Run the tests and fix failures", "/path/to/repo"],
                    timeout=600, idempotency_key="task-1")
    print(r["exit_code"], r["stdout"])

Or run it directly as a probe:

    BRIDGE_ROOT=/Users/you/.cowork-to-code-bridge python bridge_client.py
    # prints "BRIDGE LIVE" or "DAEMON NOT REACHABLE"
"""
from __future__ import annotations

import json
import os
import time
import uuid
from pathlib import Path
from typing import Any

__version__ = "0.5.1"


def _resolve_bridge_root() -> Path:
    """Find the bridge directory. Order: $BRIDGE_ROOT, $PWD/bridge, ./bridge."""
    env = os.environ.get("BRIDGE_ROOT")
    if env:
        return Path(env)
    cwd_bridge = Path.cwd() / "bridge"
    if cwd_bridge.exists():
        return cwd_bridge
    return Path.cwd() / "bridge"


def _load_token(bridge_root: Path) -> str | None:
    """Load BRIDGE_TOKEN: env var wins, else .env in bridge_root, else None."""
    env_tok = os.environ.get("BRIDGE_TOKEN")
    if env_tok:
        return env_tok
    env_file = bridge_root / ".env"
    if not env_file.exists():
        return None
    for line in env_file.read_text().splitlines():
        if line.strip().startswith("BRIDGE_TOKEN"):
            _, _, v = line.partition("=")
            return v.strip().strip('"').strip("'") or None
    return None


def queue_task(
    script: str,
    args: list[str | int | float] | None = None,
    timeout: int = 60,
    cwd: str | None = None,
    env: dict[str, str] | None = None,
    bridge_root: Path | str | None = None,
    idempotency_key: str | None = None,
    plan: str | None = None,
    max_budget_usd: float | None = None,
    permission_scope: str | None = None,
    model_tier: str | None = None,
    effort: str | None = None,
) -> dict[str, Any]:
    """Queue a task WITHOUT waiting for result (async, non-blocking).

    Returns immediately with task_id. Use poll_task_result(task_id) to check
    for completion later. Useful in time-limited environments (e.g. 45s bash).
    """
    root = Path(bridge_root) if bridge_root else _resolve_bridge_root()
    queue = root / "queue"
    queue.mkdir(parents=True, exist_ok=True)

    task_id = f"{int(time.time())}_{uuid.uuid4().hex[:8]}"
    payload: dict[str, Any] = {
        "id": task_id,
        "script": script,
        "args": args or [],
        "timeout": timeout,
        "ts_submitted": time.time(),
    }
    if cwd:
        payload["cwd"] = cwd
    if env:
        payload["env"] = env
    if idempotency_key:
        payload["idempotency_key"] = idempotency_key
    if plan is not None:
        payload["plan"] = plan
    if max_budget_usd is not None:
        payload["max_budget_usd"] = float(max_budget_usd)
    if permission_scope is not None:
        payload["permission_scope"] = str(permission_scope)
    if model_tier is not None:
        payload["model_tier"] = str(model_tier).strip().lower()
    if effort is not None:
        payload["effort"] = str(effort).strip().lower()

    token = _load_token(root)
    if token:
        payload["token"] = token

    task_file = queue / f"{task_id}.json"
    tmp = task_file.with_suffix(".json.tmp")
    tmp.write_text(json.dumps(payload))
    tmp.rename(task_file)

    return {
        "task_id": task_id,
        "status": "queued",
        "timestamp": payload["ts_submitted"],
    }


# Braille spinner frames — same set Claude Code uses for its own ticker.
_SPINNER_FRAMES = "⣾⣽⣻⢿⡿⣟⣯⣷"


def format_status_line(
    status: dict[str, Any],
    *,
    verb: str = "Working",
    show_last_line: bool = False,
) -> str:
    """Render a daemon status dict into a human-facing ticker line.

    Produces strings like ``⣾ Working… 42s elapsed`` from the
    ``progress/<id>.status.json`` payload the daemon writes (see
    ``poll_task_result`` / ``call_remote(on_status=...)``).

    Args:
        status: a status dict with at least ``elapsed_s`` (int seconds). May
            also carry ``last_line`` (most recent script output) and ``state``.
        verb: the present-participle-ish label shown before the ellipsis
            (e.g. "Building", "Running tests"). Ignored once the task reports
            a terminal ``state`` of "done"/"failed".
        show_last_line: when True, append the script's most recent output line.

    The spinner frame is chosen from ``elapsed_s`` so successive polls advance
    it without the client tracking any frame counter.
    """
    elapsed = int(status.get("elapsed_s", 0) or 0)
    state = status.get("state", "running")

    if state == "done":
        head = f"✓ Done in {elapsed}s"
    elif state == "failed":
        head = f"✗ Failed after {elapsed}s"
    else:
        frame = _SPINNER_FRAMES[elapsed % len(_SPINNER_FRAMES)]
        head = f"{frame} {verb}… {elapsed}s elapsed"

    if show_last_line:
        last = str(status.get("last_line", "")).strip()
        if last:
            head = f"{head}  ·  {last}"
    return head


def poll_task_result(
    task_id: str,
    bridge_root: Path | str | None = None,
) -> dict[str, Any]:
    """Check if a queued task has completed (fully idempotent polling).

    Returns:
      status='queued' - task not yet picked up
      status='running' - daemon is executing
      status='completed' - task finished; also has exit_code, stdout, stderr
      status='unknown' - task not found
    """
    root = Path(bridge_root) if bridge_root else _resolve_bridge_root()
    queue = root / "queue"
    results = root / "results"
    progress = root / "progress"

    result_file = results / f"{task_id}.json"
    if result_file.exists():
        try:
            result = json.loads(result_file.read_text())
            return {"status": "completed", **result}
        except json.JSONDecodeError:
            pass

    progress_file = progress / f"{task_id}.log"
    if progress_file.exists():
        running: dict[str, Any] = {
            "status": "running",
            "task_id": task_id,
            "progress_available": True,
        }
        # Surface the daemon's live status ticker (elapsed_s / last_line /
        # state), written atomically to progress/<id>.status.json every ~2s.
        # This lets the non-blocking queue_task + poll_task_result flow show a
        # spinner without tailing the raw log. Best-effort: a missing or
        # half-written file just means no ticker this poll.
        status_file = progress / f"{task_id}.status.json"
        try:
            status = json.loads(status_file.read_text())
            if isinstance(status, dict):
                for key in ("elapsed_s", "last_line", "state"):
                    if key in status:
                        running[key] = status[key]
                running["status_line"] = format_status_line(status)
        except (OSError, json.JSONDecodeError):
            pass
        return running

    task_file = queue / f"{task_id}.json"
    if task_file.exists():
        return {"status": "queued", "task_id": task_id}

    return {
        "status": "unknown",
        "task_id": task_id,
        "message": "Task not found in queue or results",
    }


def call_remote(
    script: str,
    args: list[str | int | float] | None = None,
    timeout: int = 60,
    poll_interval: float = 1.0,
    cwd: str | None = None,
    env: dict[str, str] | None = None,
    bridge_root: Path | str | None = None,
    idempotency_key: str | None = None,
    plan: str | None = None,
    max_budget_usd: float | None = None,
    permission_scope: str | None = None,
    model_tier: str | None = None,
    effort: str | None = None,
) -> dict[str, Any]:
    """Submit a script invocation to the Mac daemon and wait for its result.

    See the full package docs for details. Key points:
      - `script` must be whitelisted on the Mac (e.g. "scripts/run_claude.sh").
      - `idempotency_key` makes retries safe: same key => the daemon runs the
        script once and returns the cached result (annotated idempotent_replay).
      - exit_code -4 = daemon crashed mid-run; treat as indeterminate.
      - `plan` is an optional plain-English description of what the task will do.
        If approve_plan.sh exists on the machine the daemon runs it first.
    Raises TimeoutError if the daemon doesn't respond within timeout + 5s.
    """
    root = Path(bridge_root) if bridge_root else _resolve_bridge_root()
    queue = root / "queue"
    results = root / "results"
    queue.mkdir(parents=True, exist_ok=True)
    results.mkdir(parents=True, exist_ok=True)

    cmd_id = f"{int(time.time())}_{uuid.uuid4().hex[:8]}"
    payload: dict[str, Any] = {
        "id": cmd_id,
        "script": script,
        "args": args or [],
        "timeout": timeout,
        "ts_submitted": time.time(),
    }
    if cwd:
        payload["cwd"] = cwd
    if env:
        payload["env"] = env
    if idempotency_key:
        payload["idempotency_key"] = idempotency_key
    if plan is not None:
        payload["plan"] = plan
    if max_budget_usd is not None:
        payload["max_budget_usd"] = float(max_budget_usd)
    if permission_scope is not None:
        payload["permission_scope"] = str(permission_scope)
    if model_tier is not None:
        payload["model_tier"] = str(model_tier).strip().lower()
    if effort is not None:
        payload["effort"] = str(effort).strip().lower()

    token = _load_token(root)
    if token:
        payload["token"] = token

    cmd_file = queue / f"{cmd_id}.json"
    tmp = cmd_file.with_suffix(".json.tmp")
    tmp.write_text(json.dumps(payload))
    tmp.rename(cmd_file)

    result_file = results / f"{cmd_id}.json"
    deadline = time.time() + timeout + 5
    while time.time() < deadline:
        if result_file.exists():
            try:
                return json.loads(result_file.read_text())
            except json.JSONDecodeError:
                time.sleep(poll_interval)
                continue
        time.sleep(poll_interval)

    raise TimeoutError(
        f"bridge: no result for {cmd_id} within {timeout + 5}s. "
        f"Is the daemon running on the Mac? Check "
        f"`launchctl list | grep cowork-to-code-bridge` on the Mac, and confirm "
        f"BRIDGE_ROOT matches the path in your Mac's ~/.cowork-to-code-bridge/.env."
    )


def call_remote_streaming(script, args=None, timeout=600, poll_interval=1.0,
                          cwd=None, env=None, bridge_root=None,
                          idempotency_key=None, on_progress=None, on_status=None,
                          plan=None, max_budget_usd=None,
                          interactive=False, permission_scope=None,
                          model_tier=None, effort=None) -> dict[str, Any]:
    """Like call_remote, but streams live output while the task runs.

    The daemon tees the script's output to progress/<id>.log; this polls it and
    calls on_progress(new_text) for each new chunk (or prints it if on_progress
    is None). Use for long tasks (builds, test runs) so they're not blind.
    Returns the same final result dict as call_remote.

    on_status: optional callable receiving status dicts written by the daemon
    every ~2 s to progress/<id>.status.json.  Each dict has:
        elapsed_s  (int)  seconds since the script started
        last_line  (str)  most recent non-empty output line
        state      (str)  "running" | "done" | "error"
        exit_code  (int)  present only when state != "running"
    Called only when the file changes (mtime-gated).

    interactive: when True, each poll iteration also scans to_cowork/*.json for
    a request whose "parent" field matches this cmd_id. If found, returns early
    with shape:
        {"state": "awaiting_reply", "cmd_id": ..., "request_id": ...,
         "question": ..., "from": "claude-code"}
    The caller should show the question, call reply_to_machine(request_id, answer),
    then call resume_remote(cmd_id, remaining_deadline) to re-enter the wait loop.
    """
    root = Path(bridge_root) if bridge_root else _resolve_bridge_root()
    queue = root / "queue"; results = root / "results"; progress = root / "progress"
    to_cowork = root / "to_cowork"
    queue.mkdir(parents=True, exist_ok=True); results.mkdir(parents=True, exist_ok=True)
    cmd_id = f"{int(time.time())}_{uuid.uuid4().hex[:8]}"
    payload: dict[str, Any] = {"id": cmd_id, "script": script, "args": args or [],
                               "timeout": timeout, "ts_submitted": time.time()}
    if cwd: payload["cwd"] = cwd
    if env: payload["env"] = env
    if idempotency_key: payload["idempotency_key"] = idempotency_key
    if plan is not None: payload["plan"] = plan
    if max_budget_usd is not None: payload["max_budget_usd"] = float(max_budget_usd)
    if permission_scope is not None: payload["permission_scope"] = str(permission_scope)
    if model_tier is not None: payload["model_tier"] = str(model_tier).strip().lower()
    if effort is not None: payload["effort"] = str(effort).strip().lower()
    token = _load_token(root)
    if token: payload["token"] = token
    cmd_file = queue / f"{cmd_id}.json"
    tmp = cmd_file.with_suffix(".json.tmp")
    tmp.write_text(json.dumps(payload)); tmp.rename(cmd_file)
    result_file = results / f"{cmd_id}.json"
    progress_file = progress / f"{cmd_id}.log"
    status_file = progress / f"{cmd_id}.status.json"
    emit = on_progress or (lambda chunk: print(chunk, end="", flush=True))
    seen = 0
    last_status_mtime: float = 0.0
    deadline = time.time() + timeout + 5
    while time.time() < deadline:
        try:
            if progress_file.exists():
                data = progress_file.read_text()
                if len(data) > seen:
                    emit(data[seen:]); seen = len(data)
        except OSError:
            pass
        if on_status is not None:
            try:
                mtime = status_file.stat().st_mtime
                if mtime > last_status_mtime:
                    last_status_mtime = mtime
                    on_status(json.loads(status_file.read_text()))
            except (OSError, json.JSONDecodeError):
                pass
        if interactive and to_cowork.exists():
            for req_file in sorted(to_cowork.glob("*.json")):
                try:
                    req = json.loads(req_file.read_text())
                except (OSError, json.JSONDecodeError):
                    continue
                if req.get("parent") == cmd_id:
                    req_file.rename(req_file.with_suffix(".json.answered"))
                    return {
                        "state": "awaiting_reply",
                        "cmd_id": cmd_id,
                        "request_id": req.get("id", req_file.stem),
                        "question": req.get("request", ""),
                        "from": req.get("from", "claude-code"),
                        "_deadline": deadline,
                        "_bridge_root": str(root),
                    }
        if result_file.exists():
            try:
                return json.loads(result_file.read_text())
            except json.JSONDecodeError:
                time.sleep(poll_interval); continue
        time.sleep(poll_interval)
    raise TimeoutError(f"bridge: no result for {cmd_id} within {timeout + 5}s.")


def reply_to_machine(
    request_id: str,
    text: str,
    bridge_root: Path | str | None = None,
) -> None:
    """Write a reply to a mid-task question raised by a running script.

    Call this after call_remote_streaming returns {"state": "awaiting_reply"}.
    Writes the reply JSON to cowork_results/<request_id>.json so request_cowork.sh
    --wait polling on the Mac side picks it up and unblocks the script.

    Args:
        request_id: the "request_id" field from the awaiting_reply dict.
        text:       the answer to send back (plain string).
    """
    root = Path(bridge_root) if bridge_root else _resolve_bridge_root()
    cowork_results = root / "cowork_results"
    cowork_results.mkdir(parents=True, exist_ok=True)
    reply = {"id": request_id, "reply": text, "ts": time.time(), "from": "cowork"}
    out = cowork_results / f"{request_id}.json"
    tmp = out.with_suffix(".json.tmp")
    tmp.write_text(json.dumps(reply))
    tmp.rename(out)


def resume_remote(
    cmd_id: str,
    deadline: float,
    poll_interval: float = 1.0,
    on_progress=None,
    on_status=None,
    interactive: bool = True,
    bridge_root: Path | str | None = None,
) -> dict[str, Any]:
    """Re-enter the wait loop for a task after answering its question.

    Call this after reply_to_machine() to continue waiting for the task's
    final result (or its next question, if interactive=True).

    Args:
        cmd_id:    the "cmd_id" from the awaiting_reply dict.
        deadline:  the "_deadline" from the awaiting_reply dict (absolute epoch time).
                   Preserves the original timeout budget — the clock never resets.
    Returns:
        Same dict shapes as call_remote_streaming: final result or another
        awaiting_reply if the task asks another question.
    """
    root = Path(bridge_root) if bridge_root else _resolve_bridge_root()
    results = root / "results"
    progress = root / "progress"
    to_cowork = root / "to_cowork"
    result_file = results / f"{cmd_id}.json"
    progress_file = progress / f"{cmd_id}.log"
    status_file = progress / f"{cmd_id}.status.json"
    emit = on_progress or (lambda chunk: print(chunk, end="", flush=True))
    seen = 0
    last_status_mtime: float = 0.0
    while time.time() < deadline:
        try:
            if progress_file.exists():
                data = progress_file.read_text()
                if len(data) > seen:
                    emit(data[seen:]); seen = len(data)
        except OSError:
            pass
        if on_status is not None:
            try:
                mtime = status_file.stat().st_mtime
                if mtime > last_status_mtime:
                    last_status_mtime = mtime
                    on_status(json.loads(status_file.read_text()))
            except (OSError, json.JSONDecodeError):
                pass
        if interactive and to_cowork.exists():
            for req_file in sorted(to_cowork.glob("*.json")):
                try:
                    req = json.loads(req_file.read_text())
                except (OSError, json.JSONDecodeError):
                    continue
                if req.get("parent") == cmd_id:
                    req_file.rename(req_file.with_suffix(".json.answered"))
                    return {
                        "state": "awaiting_reply",
                        "cmd_id": cmd_id,
                        "request_id": req.get("id", req_file.stem),
                        "question": req.get("request", ""),
                        "from": req.get("from", "claude-code"),
                        "_deadline": deadline,
                        "_bridge_root": str(root),
                    }
        if result_file.exists():
            try:
                return json.loads(result_file.read_text())
            except json.JSONDecodeError:
                time.sleep(poll_interval); continue
        time.sleep(poll_interval)
    raise TimeoutError(
        f"bridge: no result for {cmd_id} — deadline expired while waiting for reply."
    )


def daemon_alive(bridge_root: Path | str | None = None, ping_timeout: int = 10) -> bool:
    """Quick health check — submits the ping script and waits for exit_code==0."""
    try:
        r = call_remote("scripts/ping.sh", args=[], timeout=ping_timeout,
                        bridge_root=bridge_root)
        return r.get("exit_code") == 0
    except TimeoutError:
        return False


def post_message_to_cowork(
    message_type: str,
    content: str,
    parent_task_id: str | None = None,
    bridge_root: Path | str | None = None,
) -> str:
    """Post a message from Claude Code back to Cowork (bidirectional communication)."""
    root = Path(bridge_root) if bridge_root else _resolve_bridge_root()
    to_cowork = root / "to_cowork"
    to_cowork.mkdir(parents=True, exist_ok=True)

    request_id = f"msg_{int(time.time())}_{uuid.uuid4().hex[:8]}"
    message = {
        "id": request_id,
        "type": message_type,
        "content": content,
        "ts": time.time(),
        "from": "claude-code",
    }
    if parent_task_id:
        message["parent"] = parent_task_id

    msg_file = to_cowork / f"{request_id}.json"
    tmp = msg_file.with_suffix(".json.tmp")
    tmp.write_text(json.dumps(message))
    tmp.rename(msg_file)

    return request_id


def detect_messages_from_claude_code(
    parent_task_id: str | None = None,
    bridge_root: Path | str | None = None,
) -> list[dict[str, Any]]:
    """Detect and retrieve messages posted by Claude Code (bidirectional communication)."""
    root = Path(bridge_root) if bridge_root else _resolve_bridge_root()
    to_cowork = root / "to_cowork"

    if not to_cowork.exists():
        return []

    messages = []
    for msg_file in sorted(to_cowork.glob("*.json")):
        if msg_file.suffix == ".answered":
            continue
        try:
            msg = json.loads(msg_file.read_text())
        except (OSError, json.JSONDecodeError):
            continue

        if parent_task_id and msg.get("parent") != parent_task_id:
            continue

        messages.append(msg)

    return messages


if __name__ == "__main__":
    # Run directly as a probe: prints LIVE / NOT REACHABLE.
    alive = daemon_alive(ping_timeout=10)
    print("BRIDGE LIVE" if alive else "DAEMON NOT REACHABLE")
    raise SystemExit(0 if alive else 1)
