#!/usr/bin/env python3
"""Production full-generation peer conversation over a transport directory.

The directory is the object-level seam implemented by the MTP adapter in a
deployed bundle and directly supplied by isolated tests. Business documents are
always produced and consumed by the existing generation services.
"""

from __future__ import annotations

import argparse
import hashlib
import json
import os
import shutil
import subprocess
import sys
import tempfile
import time
import uuid
from contextlib import closing, redirect_stdout
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent))
import sync_generation_exchange as generation

CAPS = {
    "generation-manifest-v1",
    "generation-ack-v1",
    "causal-delete-v1",
    "mobile-history-v4",
    "execution-draft-v1",
    "generation-archive-v1",
}


def emit(run_id: str, phase: str, **values: object) -> None:
    value = {
        "format": "trainlog-sync-progress",
        "version": 1,
        "run_id": run_id,
        "phase": phase,
    } | values
    print(json.dumps(value, sort_keys=True, separators=(",", ":")), flush=True)


def bounded(path: Path, limit: int = 65536) -> bytes:
    with path.open("rb") as stream:
        raw = stream.read(limit + 1)
    if len(raw) > limit:
        raise RuntimeError(f"{path.name} exceeds bound")
    return raw


def publish_json(path: Path, value: dict) -> None:
    """Publish a coordination object only after all bytes are durable."""
    raw = generation.canonical(value)
    if len(raw) > 65536:
        raise RuntimeError(f"{path.name} exceeds bound")
    temporary = path.with_name("." + path.name + ".tmp-" + str(os.getpid()))
    with temporary.open("wb") as stream:
        stream.write(raw)
        stream.flush()
        os.fsync(stream.fileno())
    os.chmod(temporary, 0o600)
    os.replace(temporary, path)


def run_adapter(
    adapter: Path,
    operation: str,
    peer: str,
    root: Path,
    deadline: float,
    allow_missing_peer: bool = False,
) -> bool:
    remaining = max(1, int(deadline - time.monotonic()))
    timeout = min(remaining, 30)
    try:
        result = subprocess.run(
            [str(adapter), operation, peer, str(root)],
            stdin=subprocess.DEVNULL,
            capture_output=True,
            timeout=timeout,
            check=False,
        )
    except subprocess.TimeoutExpired as error:
        # CONTRACT: an adapter timeout is an ambiguous transport failure. It
        # never implies rollback, replay, or permission to retry a mutation.
        raise RuntimeError(
            f"transport_timeout: MTP {operation} did not finish within {timeout} seconds"
        ) from error
    if result.returncode != 0:
        diagnostic = (
            result.stderr.decode("utf-8", "replace")[:1024] or "MTP adapter failed"
        ).strip()
        missing_peer = diagnostic in {
            "double failed 6 expected Android peer not found",
            "MTP adapter failed status=6 diagnostic=expected Android peer not found",
        }
        if allow_missing_peer and missing_peer:
            return False
        raise RuntimeError(diagnostic)
    return True


def push_adapter(
    adapter: Path,
    peer: str,
    root: Path,
    relative_paths: list[str],
    deadline: float,
) -> None:
    """Publish only the phase-owned bounded MTP objects.

    WHY: the durable transport root also retains inbound objects, legacy files,
    and staging evidence; recursively uploading that root makes each exchange
    grow with history and can time out before Android sees the request.
    CONTRACT: callers name fixed relative files/directories already durably
    published below ``root``. The adapter receives a private disposable view.
    INVARIANT: no retained generation, ACK, tombstone, or source object is
    deleted or modified while constructing the view.
    """
    with tempfile.TemporaryDirectory(prefix="trainlog-mtp-outbox-", dir=root.parent) as directory:
        outbox = Path(directory)
        os.chmod(outbox, 0o700)
        for relative in relative_paths:
            source = root / relative
            destination = outbox / relative
            if (
                source.is_symlink()
                or not source.exists()
                or not source.resolve().is_relative_to(root.resolve())
            ):
                raise RuntimeError(f"invalid MTP outbox source: {relative}")
            if source.is_dir() and any(path.is_symlink() for path in source.rglob("*")):
                raise RuntimeError(f"MTP outbox source contains a symbolic link: {relative}")
            destination.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
            if source.is_dir():
                shutil.copytree(source, destination, copy_function=shutil.copy2)
            elif source.is_file():
                shutil.copy2(source, destination)
            else:
                raise RuntimeError(f"invalid MTP outbox object: {relative}")
        run_adapter(adapter, "push", peer, outbox, deadline)


def wait_file(path: Path, deadline: float, pump=None) -> None:
    while not path.is_file():
        if time.monotonic() >= deadline:
            raise RuntimeError(f"timeout waiting for {path.name}")
        if pump is not None:
            pump()
        time.sleep(0.05)


def wait_json(path: Path, deadline: float, predicate, pump=None) -> dict:
    while True:
        wait_file(path, deadline, pump)
        try:
            value = json.loads(bounded(path))
        except (OSError, json.JSONDecodeError):
            value = None
        if isinstance(value, dict) and predicate(value):
            return value
        if time.monotonic() >= deadline:
            raise RuntimeError(f"timeout waiting for correlated {path.name}")
        if pump is not None:
            pump()
        time.sleep(0.05)


def correlated_peer_error(root: Path, run_id: str) -> None:
    path = root / "android-generation-error-v1.json"
    if not path.is_file():
        return
    try:
        value = json.loads(bounded(path))
    except (OSError, json.JSONDecodeError):
        return
    if (
        isinstance(value, dict)
        and value.get("format") == "trainlog-sync-generation-error"
        and value.get("version") == 1
        and value.get("run_id") == run_id
        and value.get("code") == "peer_capacity_exhausted"
    ):
        raise RuntimeError(
            "peer_capacity_exhausted: Android active generation capacity is exhausted; "
            "archive acknowledged generations, then retry explicitly"
        )


def load_peer(root: Path, expected: str) -> dict:
    value = json.loads(bounded(root / "android-peer-v1.json"))
    if (
        set(value) != {"format", "version", "peer_id", "capabilities"}
        or value["format"] != "trainlog-sync-peer"
        or type(value["version"]) is not int
        or value["version"] != 1
        or value["peer_id"] != expected
        or not isinstance(value["capabilities"], list)
        or not CAPS.issubset(set(value["capabilities"]))
    ):
        raise RuntimeError("incompatible or unpaired Android peer advertisement")
    return value


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--database", required=True, type=Path)
    parser.add_argument("--transport-root", required=True, type=Path)
    parser.add_argument("--owned-root", required=True, type=Path)
    parser.add_argument("--run-id", required=True)
    parser.add_argument("--expected-peer", required=True)
    parser.add_argument("--timeout", required=True, type=int)
    parser.add_argument("--mode", choices=("directory", "mtp"), default="directory")
    parser.add_argument("--mtp-adapter", type=Path)
    args = parser.parse_args()
    deadline = time.monotonic() + args.timeout
    pump = None
    if args.mode == "mtp":
        if args.mtp_adapter is None:
            raise RuntimeError("MTP adapter is required")
        pump = lambda: run_adapter(
            args.mtp_adapter,
            "pull",
            args.expected_peer,
            args.transport_root,
            deadline,
            True,
        )
        # Android publishes its durable identity only while participating.  A
        # missing expected advertisement is therefore a bounded wait state;
        # device, transport, ambiguity and malformed-object failures remain
        # immediate and distinct adapter errors.
        wait_file(args.transport_root / "android-peer-v1.json", deadline, pump)
    peer = load_peer(args.transport_root, args.expected_peer)
    with closing(generation.connect_database(args.database)) as db:
        generation.require_schema(db)
        desktop_peer = generation.peer_identity(db, "desktop")
        db.commit()
    request = {
        "format": "trainlog-sync-generation-request",
        "version": 1,
        "run_id": args.run_id,
        "desktop_peer_id": desktop_peer,
        "android_peer_id": peer["peer_id"],
        "capabilities": sorted(CAPS),
    }
    request_path = args.transport_root / "request-v1.json"
    publish_json(request_path, request)
    with closing(generation.connect_database(args.database)) as db:
        acknowledgement_rows = db.execute(
            "SELECT ack_json FROM sync_consumed_generations WHERE producer_peer_id=? "
            "AND consumer_peer_id=? AND result='consumed' ORDER BY consumed_at,generation_id LIMIT 32",
            (peer["peer_id"], desktop_peer),
        ).fetchall()
    archive_acknowledgements = {
        "format": "trainlog-sync-archive-acknowledgements",
        "version": 1,
        "run_id": args.run_id,
        "android_peer_id": peer["peer_id"],
        "desktop_peer_id": desktop_peer,
        "acknowledgements": [json.loads(row[0]) for row in acknowledgement_rows],
    }
    archive_ack_path = args.transport_root / "desktop-archive-acknowledgements-v1.json"
    publish_json(archive_ack_path, archive_acknowledgements)
    if args.mode == "mtp":
        push_adapter(
            args.mtp_adapter,
            args.expected_peer,
            args.transport_root,
            ["request-v1.json", "desktop-archive-acknowledgements-v1.json"],
            deadline,
        )
    emit(
        args.run_id,
        "waiting_android_publication",
        producer_peer_id=peer["peer_id"],
        consumer_peer_id=desktop_peer,
    )
    inbound_ref = args.transport_root / "android-generation-v1.json"
    def pump_inbound():
        if pump is not None:
            pump()
        correlated_peer_error(args.transport_root, args.run_id)

    inbound = wait_json(
        inbound_ref, deadline, lambda value: value.get("run_id") == args.run_id, pump_inbound
    )
    inbound_dir = args.transport_root / inbound["relative_path"]
    ack = generation.consume_desktop(args.database, inbound_dir)
    ack_path = args.transport_root / "desktop-consumption-ack-v1.json"
    publish_json(ack_path, ack)
    if args.mode == "mtp":
        push_adapter(
            args.mtp_adapter,
            args.expected_peer,
            args.transport_root,
            ["request-v1.json", "desktop-archive-acknowledgements-v1.json", "desktop-consumption-ack-v1.json"],
            deadline,
        )
    if ack["result"] != "consumed":
        raise RuntimeError("desktop rejected Android generation")
    emit(
        args.run_id,
        "local_import_committed",
        inbound_generation_id=ack["generation_id"],
        manifest_sha256=ack["manifest_sha256"],
    )
    outgoing_id = "gen_" + str(uuid.uuid4())
    # Export codecs retain human diagnostics on stdout when invoked as CLIs.
    # The worker stdout contract is typed NDJSON only, so route those bounded
    # diagnostics to the separately bounded stderr channel.
    with redirect_stdout(sys.stderr):
        stage, manifest, digest = generation.capture_desktop(
            args.database, args.owned_root, peer["peer_id"], args.run_id, outgoing_id
        )
    del stage
    published = generation.publish(
        args.database, outgoing_id, args.transport_root / "desktop-objects"
    )
    ref = {
        "format": "trainlog-sync-generation-reference",
        "version": 1,
        "run_id": args.run_id,
        "generation_id": outgoing_id,
        "manifest_sha256": digest,
        "relative_path": str(published.relative_to(args.transport_root)),
    }
    publish_json(args.transport_root / "desktop-generation-v1.json", ref)
    if args.mode == "mtp":
        push_adapter(
            args.mtp_adapter,
            args.expected_peer,
            args.transport_root,
            [
                "request-v1.json",
                "desktop-archive-acknowledgements-v1.json",
                "desktop-consumption-ack-v1.json",
                "desktop-generation-v1.json",
                ref["relative_path"],
            ],
            deadline,
        )
    emit(
        args.run_id,
        "published",
        outbound_generation_id=outgoing_id,
        manifest_sha256=digest,
    )
    emit(args.run_id, "waiting_acknowledgement", outbound_generation_id=outgoing_id)
    android_ack = args.transport_root / "android-consumption-ack-v1.json"
    android_ack_value = wait_json(
        android_ack,
        deadline,
        lambda value: value.get("run_id") == args.run_id
        and value.get("generation_id") == outgoing_id,
        pump,
    )
    result = generation.accept_ack(args.database, android_ack)
    if result not in ("acknowledged", "unchanged"):
        raise RuntimeError(
            f"Android did not durably consume desktop generation: {result}: {android_ack_value.get('diagnostic','')}"
        )
    with closing(generation.connect_database(args.database)) as db:
        inbound_row = db.execute(
            "SELECT result FROM sync_consumed_generations WHERE generation_id=?",
            (ack["generation_id"],),
        ).fetchone()
        outbound_row = db.execute(
            "SELECT status FROM sync_generations WHERE generation_id=?", (outgoing_id,)
        ).fetchone()
        drafts = [
            {
                "draft_id": row[0],
                "state": row[1],
                "session_type": json.loads(row[2]).get("session_type", "training"),
                "occurrence_count": len(json.loads(row[2]).get("exercises", [])),
            }
            for row in db.execute(
                "SELECT session_id,state,payload_json FROM execution_drafts ORDER BY session_id LIMIT 32"
            )
        ]
    if inbound_row != ("consumed",) or outbound_row != ("acknowledged",):
        raise RuntimeError("generation evidence did not corroborate completion")
    emit(args.run_id, "peer_consumed", outbound_generation_id=outgoing_id)
    report = {
        "format": "trainlog-sync-worker-report",
        "version": 1,
        "run_id": args.run_id,
        "producer_peer_id": peer["peer_id"],
        "consumer_peer_id": desktop_peer,
        "inbound_generation_id": ack["generation_id"],
        "outbound_generation_id": outgoing_id,
        "manifest_sha256": digest,
        "result": "completed",
        "sessions_reconciled": 0,
        "domains": {
            name: True
            for name in (
                "history-v4",
                "execution-draft-v1",
                "causal-delete-v1",
                "equipment",
                "aliases",
                "profiles",
                "body-zones",
                "feedback",
            )
        },
        "drafts": drafts,
        "ai_midpoint": {"result": "not_configured"},
        "ai_post_sync": {"result": "not_configured"},
    }
    print(json.dumps(report, sort_keys=True, separators=(",", ":")), flush=True)
    return 0


if __name__ == "__main__":
    try:
        raise SystemExit(main())
    except RuntimeError as error:
        print(f"sync peer failed: {error}", file=sys.stderr)
        raise SystemExit(1) from None