"""Talking to an engine: uploading declared flows, and running them.

Two halves of one job. :func:`sync` puts what the decorators declared into the
flow store; :class:`Client` submits runs and reads back what they produced, so
an experiment is started and inspected from the same script that defines it.
"""

from __future__ import annotations

import json
import os
import subprocess
import time
import uuid
from collections.abc import Iterable
from pathlib import Path
from typing import Any

from fluksio.sdk import MARKER, Flow, SyncError

__all__ = [
    "Client",
    "RunHandle",
    "SyncReport",
    "config_path",
    "data_dir",
    "find_data_dir",
    "ignore_self",
    "login",
    "sync",
    "write_config",
]

API = "/api/v1"
#: A run is over when it reaches one of these.
DONE = frozenset({"ok", "error", "cancelled", "abandoned"})

#: How long to wait for an answer. Split, because the two halves mean
#: different things: an engine that is not there refuses the connection at
#: once, so waiting 30 s for it only ever delays a typo in the URL, while an
#: engine that *is* there can be busy — a rebuild waits 15 s before it answers
#: 503, and compiling a node is given 60 s — so the read has to outlast the
#: slowest thing the engine does on purpose.
CONNECT_TIMEOUT = 5.0
READ_TIMEOUT = 120.0

#: How many times an idempotent call is tried again before it gives up, and
#: what a busy engine answers with. A read timeout used to kill a driver
#: script outright, which is a poor trade for a request that costs nothing to
#: repeat. 503 is the engine's own "ask again": it is what a rebuild answers.
RETRIES = 3
RETRY_STATUS = frozenset({502, 503, 504})

#: How many refreshes in a row may fail before waiting gives up. Each one is
#: already several attempts, so this is minutes of a missing engine, not a
#: blip.
WAIT_TOLERANCE = 5


#: What a project-local installation is called, beside `.venv` and `.git`.
DATA_DIR_NAME = ".fluksio"

#: The shared one, for a machine that wants a single engine rather than one
#: per repository. `fluksio serve --global` is how you ask for it.
GLOBAL_DATA_DIR = Path("~/.fluksio")


def find_data_dir(start: Path | None = None) -> Path | None:
    """The nearest project-local installation, walking up from ``start``.

    The same search `.git` and `.venv` get, and for the same reason: which
    installation you mean is a fact about where you are standing, not about
    which machine you are on. Several repositories on one device each keep
    their own flows, runs and token this way rather than sharing one.
    """
    directory = (start or Path.cwd()).resolve()
    for candidate in (directory, *directory.parents):
        local = candidate / DATA_DIR_NAME
        if local.is_dir():
            return local
    return None


def data_dir(start: Path | None = None) -> Path:
    """The installation this working directory belongs to."""
    found = find_data_dir(start)
    return found if found is not None else GLOBAL_DATA_DIR.expanduser()


def config_path(directory: Path | None = None) -> Path:
    """Where the token for an installation lives — beside the data it opens.

    Not a single file per machine: a token is for one engine, and with an
    installation per repository there is more than one. Keeping it inside the
    data directory means the client finds the credential for the engine whose
    directory it is standing in, without either having to be told.
    """
    return (directory or data_dir()) / "client.json"


def _legacy_config_path() -> Path:
    """Where `fluksio login` used to write, before installations were local."""
    base = os.environ.get("XDG_CONFIG_HOME") or str(Path.home() / ".config")
    return Path(base) / "fluksio" / "client.json"


def _read(path: Path) -> dict[str, str]:
    if not path.exists():
        return {}
    try:
        data = json.loads(path.read_text())
    except ValueError:
        return {}
    return data if isinstance(data, dict) else {}


def _stored() -> dict[str, str]:
    """The nearest credential: this project's, then the machine's."""
    for path in (config_path(), _legacy_config_path()):
        found = _read(path)
        if found:
            return found
    return {}


class ApiError(Exception):
    """The engine refused, with what it said."""

    def __init__(self, status: int, detail: Any) -> None:
        super().__init__(f"{status}: {detail}")
        self.status = status
        self.detail = detail


class Conflict(ApiError):
    """Someone else saved in between — the 409 the optimistic lock answers."""

    def __init__(self, detail: Any) -> None:
        super().__init__(409, detail)
        self.current_version = 0
        if isinstance(detail, dict):
            self.current_version = int(detail.get("current_version") or 0)


class Client:
    """An authenticated engine, addressed over its HTTP API."""

    def __init__(
        self,
        url: str = "",
        token: str = "",
        http: Any = None,
        timeout: Any = None,
        retries: int = RETRIES,
    ) -> None:
        stored = _stored()
        self.url = url or os.environ.get("FLUKSIO_URL") or stored.get("url") or ""
        self.token = (
            token or os.environ.get("FLUKSIO_TOKEN") or stored.get("token") or ""
        )
        self.retries = retries
        if http is None:
            if not self.url:
                raise SyncError(
                    "No engine to talk to. Run `fluksio login --url http://…`, or "
                    "set FLUKSIO_URL and FLUKSIO_TOKEN."
                )
            import httpx

            if timeout is None:
                timeout = httpx.Timeout(
                    30.0, connect=CONNECT_TIMEOUT, read=READ_TIMEOUT
                )
            http = httpx.Client(base_url=self.url, timeout=timeout)
        self.http = http
        if self.token:
            self.http.headers["Authorization"] = f"Bearer {self.token}"

    # -- plumbing ----------------------------------------------------------

    def _request(
        self, method: str, path: str, idempotent: bool = False, **kwargs: Any
    ) -> Any:
        """One request, tried again while it is safe and worth it.

        Only a call that may be repeated without meaning something different
        is retried — every GET, and the writes that carry a key or are already
        a no-op the second time. The last attempt is made outside the loop, so
        whatever it raises is what the caller sees.
        """
        import httpx

        attempts = self.retries + 1 if idempotent else 1
        for attempt in range(attempts - 1):
            try:
                response = self.http.request(method, f"{API}{path}", **kwargs)
                if response.status_code not in RETRY_STATUS:
                    return response
            except httpx.TransportError:
                pass
            time.sleep(2**attempt)
        return self.http.request(method, f"{API}{path}", **kwargs)

    def _call(
        self, method: str, path: str, idempotent: bool | None = None, **kwargs: Any
    ) -> Any:
        response = self._request(
            method,
            path,
            idempotent=(method == "GET") if idempotent is None else idempotent,
            **kwargs,
        )
        if response.status_code == 409:
            raise Conflict(_detail(response))
        if response.status_code >= 400:
            raise ApiError(response.status_code, _detail(response))
        if response.status_code == 204 or not response.content:
            return None
        return response.json()

    # -- flows -------------------------------------------------------------

    def get_flow(self, name: str) -> dict[str, Any] | None:
        """The stored flow, draft included, or ``None`` if there is none."""
        try:
            result: dict[str, Any] = self._call("GET", f"/flows/{name}")
        except ApiError as exc:
            if exc.status == 404:
                return None
            raise
        return result

    def put_flow(self, document: dict[str, Any]) -> dict[str, Any]:
        result: dict[str, Any] = self._call(
            "PUT", f"/flows/{document['name']}", json=document
        )
        return result

    def get_source(self, flow: str, node: str) -> str:
        return str(self.get_source_entry(flow, node).get("code", ""))

    def get_source_entry(self, flow: str, node: str) -> dict[str, Any]:
        """A node's stored body, and whether there is one at all.

        `missing` is what separates a node nobody has written from a node
        somebody edited: the engine answers a new-node template for both, and
        only one of them is work worth refusing to overwrite.
        """
        result: dict[str, Any] = self._call("GET", f"/flows/{flow}/nodes/{node}/source")
        return result or {}

    def put_source(self, flow: str, node: str, code: str) -> dict[str, Any]:
        result: dict[str, Any] = self._call(
            "PUT", f"/flows/{flow}/nodes/{node}/source", json={"code": code}
        )
        return result

    def publish(self, name: str, version: int) -> dict[str, Any]:
        result: dict[str, Any] = self._call(
            "POST", f"/flows/{name}/publish", json={"version": version}
        )
        return result

    def refresh_modules(self) -> None:
        """Retire the engine's workers, so the next run imports the code as it is."""
        # Retiring workers twice is retiring workers, and this is the call most
        # likely to meet a rebuild's 503.
        self._call("POST", "/modules/refresh", idempotent=True)

    def flows(self) -> list[dict[str, Any]]:
        """Every flow with its node and error counts, as the home screen lists."""
        result = self._call("GET", "/flows/")
        return list(result.get("data") or [])

    def flavors(self) -> list[dict[str, Any]]:
        """The named sizes a node can ask for."""
        result = self._call("GET", "/flavors/")
        return list(result.get("data") or [])

    def resources(self) -> dict[str, Any]:
        """Every machine, what is free of it, and which nodes are queued."""
        result: dict[str, Any] = self._call("GET", "/workers/resources")
        return result

    # -- how the engine is doing -------------------------------------------

    def summary(self) -> dict[str, Any]:
        """Health as the dashboard reads it: flows, nodes, queue, loop lag."""
        result: dict[str, Any] = self._call("GET", "/observability/summary")
        return result

    def cloud_status(self) -> dict[str, Any]:
        """Whether this installation is enrolled with a portal, and linked."""
        result: dict[str, Any] = self._call("GET", "/cloud/status")
        return result

    def events(
        self,
        kind: str = "failure",
        limit: int = 10,
        flow: str = "",
        run: str = "",
        since: Any = None,
        until: Any = None,
    ) -> list[dict[str, Any]]:
        """What went wrong, or who changed what. Newest first.

        Engine-wide, and narrowed by flow, by run or by time. ``run`` takes a
        run id and answers what the engine recorded during it. For the failure
        that ended a run, :attr:`RunHandle.failures` is still the shorter road:
        it reads the run's own node rows, which carry the traceback too.
        """
        query: dict[str, Any] = {"kind": kind, "limit": limit}
        if flow:
            query["flow"] = flow
        if run:
            query["run"] = run
        for name, value in (("since", since), ("until", until)):
            if value is not None:
                query[name] = (
                    value.isoformat() if hasattr(value, "isoformat") else value
                )
        result = self._call("GET", "/observability/events", params=query)
        return list(result or [])

    # -- runs --------------------------------------------------------------

    def submit(
        self,
        flow: str,
        params: dict[str, Any] | None = None,
        seed: int | None = None,
        no_cache: bool = False,
        cause: str = "sdk",
    ) -> RunHandle:
        """Queue a run. ``cause`` is what the history records it as coming
        from — a script is the default, `fluksio run` says so itself.

        Safe to retry: the key is minted here, once per call, so an attempt
        that timed out on the way back is answered with the run it made rather
        than starting a second one.
        """
        row = self._call(
            "POST",
            f"/runs/flows/{flow}",
            idempotent=True,
            json={
                "params": params or {},
                "seed": seed,
                "no_cache": no_cache,
                "cause": cause,
                "idempotency_key": uuid.uuid4().hex,
            },
        )
        return RunHandle(self, row["id"], row)

    def sweep(
        self,
        flow: str,
        entries: list[dict[str, Any]],
        no_cache: bool = False,
    ) -> list[RunHandle]:
        """Many runs of one flow at once. The caller decides what varies.

        A key per entry rather than one for the sweep: a retry then recreates
        only the runs whose rows never landed.
        """
        keyed = [{**entry, "idempotency_key": uuid.uuid4().hex} for entry in entries]
        rows = self._call(
            "POST",
            f"/runs/flows/{flow}/sweep",
            idempotent=True,
            json={"runs": keyed, "no_cache": no_cache},
        )
        return [RunHandle(self, row["id"], row) for row in rows]

    def run(self, run_id: str) -> dict[str, Any]:
        result: dict[str, Any] = self._call("GET", f"/runs/{run_id}")
        return result

    def runs(self, flow: str = "", limit: int = 20, **filters: Any) -> Any:
        query = {"limit": limit, **filters}
        if flow:
            query["flow"] = flow
        return self._call("GET", "/runs", params=query)

    def metrics(self, run_id: str, name: str = "", stride: int = 1) -> Any:
        query: dict[str, Any] = {"stride": stride}
        if name:
            query["name"] = name
        return self._call("GET", f"/runs/{run_id}/metrics", params=query)

    def compare(self, ids: Iterable[str], metric: str) -> Any:
        return self._call(
            "GET",
            "/runs/series/compare",
            params={"ids": ",".join(ids), "metric": metric},
        )

    def export_metrics(
        self,
        flow: str = "",
        ids: Iterable[str] = (),
        name: str = "",
        stride: int = 1,
        **filters: Any,
    ) -> list[dict[str, Any]]:
        """Runs' series as long rows: run, name, step, ts, value.

        ``pd.DataFrame(client.export_metrics(flow="train"))`` is every loss
        curve of that flow with the run id on each row. ``name`` keeps the
        metrics it lists, ``stride`` thins each curve, and the selection is
        narrowed the way :meth:`runs` is — ``status=``, ``group=``,
        ``since=``, ``until=``.
        """
        query: dict[str, Any] = {"stride": stride}
        if name:
            query["name"] = name
        return self._export("/runs/export/metrics", flow, ids, query, filters)

    def export_runs(
        self,
        flow: str = "",
        ids: Iterable[str] = (),
        params: str = "",
        metrics: str = "",
        **filters: Any,
    ) -> list[dict[str, Any]]:
        """One row per run: its inputs as columns, its final numbers, its code.

        The arm-comparison table. The inputs kept are the ones that vary
        across the selection unless ``params`` names them, which is the axis a
        sweep is read along.
        """
        query: dict[str, Any] = {}
        if params:
            query["params"] = params
        if metrics:
            query["metrics"] = metrics
        return self._export("/runs/export/runs", flow, ids, query, filters)

    def _export(
        self,
        path: str,
        flow: str,
        ids: Iterable[str],
        query: dict[str, Any],
        filters: dict[str, Any],
    ) -> list[dict[str, Any]]:
        """An export as parsed rows.

        jsonl on the wire rather than csv: it keeps the types the engine has,
        and it is the format the CLI converts to parquet from. Read whole —
        the streaming is the engine's side, and a notebook wants a list.
        """
        if flow:
            query["flow"] = flow
        named = ",".join(ids)
        if named:
            query["ids"] = named
        for key, value in filters.items():
            query[key] = value.isoformat() if hasattr(value, "isoformat") else value
        query["format"] = "jsonl"
        response = self._request("GET", path, idempotent=True, params=query)
        if response.status_code >= 400:
            raise ApiError(response.status_code, _detail(response))
        return [json.loads(line) for line in response.text.splitlines() if line]

    def cancel(self, run_id: str) -> Any:
        # Cancelling a cancelled run is cancelled.
        return self._call("POST", f"/runs/{run_id}/cancel", idempotent=True)

    def download(self, digest: str) -> bytes:
        response = self._request("GET", f"/artifacts/{digest}", idempotent=True)
        if response.status_code >= 400:
            raise ApiError(response.status_code, _detail(response))
        return bytes(response.content)


def _detail(response: Any) -> Any:
    try:
        body = response.json()
    except ValueError:
        return response.text
    return body.get("detail", body) if isinstance(body, dict) else body


class RunHandle:
    """One run, and the answers it accumulates."""

    def __init__(self, client: Client, run_id: str, row: dict[str, Any]) -> None:
        self.client = client
        self.id = run_id
        self._row = row

    def refresh(self) -> RunHandle:
        self._row = self.client.run(self.id)
        return self

    @property
    def status(self) -> str:
        return str(self._row.get("status", ""))

    @property
    def done(self) -> bool:
        return self.status in DONE

    @property
    def result(self) -> dict[str, Any]:
        """What the flow's outputs held when it finished."""
        result = self._row.get("result")
        return result if isinstance(result, dict) else {}

    @property
    def artifacts(self) -> list[dict[str, Any]]:
        rows = self._row.get("artifacts")
        return rows if isinstance(rows, list) else []

    @property
    def failures(self) -> list[dict[str, Any]]:
        """The nodes that failed, each with its error and its logs."""
        if "nodes" not in self._row:
            self.refresh()
        rows = self._row.get("nodes")
        rows = rows if isinstance(rows, list) else []
        return [row for row in rows if row.get("status") == "error"]

    def wait(self, timeout: float = 0.0, poll: float = 1.0) -> RunHandle:
        """Block until the run is over, or ``timeout`` seconds have passed.

        A run outlives the engine being briefly unreachable, so waiting for one
        does too: a few failed refreshes in a row are a busy engine, not a
        finished run. What is not tolerated is the engine answering — a 404
        means the run is gone, and repeating the question will not bring it
        back.
        """
        import httpx

        deadline = time.monotonic() + timeout if timeout else 0.0
        failures = 0
        while True:
            try:
                self.refresh()
            except ApiError as exc:
                if exc.status < 500:
                    raise
                failures += 1
                if failures >= WAIT_TOLERANCE:
                    raise
            except httpx.HTTPError:
                failures += 1
                if failures >= WAIT_TOLERANCE:
                    raise
            else:
                failures = 0
                if self.done:
                    return self
            if deadline and time.monotonic() > deadline:
                raise TimeoutError(f"run {self.id} is still {self.status}")
            time.sleep(poll)

    def metrics(self, name: str = "", stride: int = 1) -> list[dict[str, Any]]:
        """A streamed port's whole series — a run's metrics are its outputs."""
        points: list[dict[str, Any]] = self.client.metrics(self.id, name, stride)
        return points

    def download(self, name: str) -> bytes:
        """The bytes of an artifact this run produced."""
        for row in self.artifacts:
            if row.get("name") == name:
                return self.client.download(str(row["digest"]))
        have = ", ".join(str(row.get("name")) for row in self.artifacts) or "none"
        raise KeyError(f"run {self.id} has no artifact '{name}' (it has {have})")

    def __getitem__(self, key: str) -> Any:
        return self._row[key]

    def __repr__(self) -> str:
        return f"RunHandle({self.id!r}, status={self.status!r})"


def ignore_self(directory: Path) -> None:
    """Keep an installation out of the repository it sits in.

    It holds a token and a database, neither of which belongs in anybody's
    history. A `.gitignore` of `*` inside the directory ignores it from within,
    so nothing has to be added to the project's own — the same thing `uv` does
    for the venv it builds.
    """
    marker = directory / ".gitignore"
    if not marker.exists():
        directory.mkdir(parents=True, exist_ok=True)
        marker.write_text("# A Fluksio installation: a database, and a token.\n*\n")


def write_config(url: str, token: str, directory: Path | None = None) -> Path:
    """Store the credential for an engine, readable only by its owner."""
    path = config_path(directory)
    path.parent.mkdir(parents=True, exist_ok=True)
    ignore_self(path.parent)
    path.write_text(json.dumps({"url": url.rstrip("/"), "token": token}, indent=2))
    path.chmod(0o600)
    return path


def login(
    url: str,
    email: str,
    password: str,
    timeout: float = 30.0,
    directory: Path | None = None,
) -> Path:
    """Exchange credentials for a token and remember the engine."""
    import httpx

    response = httpx.post(
        f"{url.rstrip('/')}{API}/login/access-token",
        data={"username": email, "password": password},
        timeout=timeout,
    )
    if response.status_code >= 400:
        raise ApiError(response.status_code, _detail(response))
    return write_config(url, str(response.json()["access_token"]), directory)


# ---------------------------------------------------------------------------
# sync
# ---------------------------------------------------------------------------


class SyncReport:
    """What one flow's sync did, for the CLI to print."""

    def __init__(self, flow: str) -> None:
        self.flow = flow
        self.created = False
        self.changed: list[str] = []
        self.published = False
        #: A draft exists and was left alone, because publishing was not asked
        #: for. False when there was simply nothing to publish, which is not
        #: the same thing and used to read as if it were.
        self.drafted = False
        #: The engine stored the node document without what each node's code
        #: reaches, which means it is older than this client.
        self.forgot_code = False

    @property
    def unchanged(self) -> bool:
        return not self.created and not self.changed

    def __repr__(self) -> str:
        return f"SyncReport({self.flow!r}, changed={self.changed})"


def origin_of(repo: str) -> dict[str, Any]:
    """Where this flow came from, so a run can name the code that produced it."""
    commit, dirty = "", False
    if repo:
        head = _git(repo, "rev-parse", "HEAD")
        if head is not None:
            commit = head
            dirty = bool(_git(repo, "status", "--porcelain"))
    return {"kind": "python", "repo": repo, "commit": commit, "dirty": dirty}


def _git(repo: str, *args: str) -> str | None:
    try:
        result = subprocess.run(
            ["git", "-C", repo, *args], capture_output=True, text=True, timeout=10
        )
    except (OSError, subprocess.SubprocessError):
        return None
    return result.stdout.strip() if result.returncode == 0 else None


def repo_root(start: str | Path) -> str:
    """The git repository a path sits in, or its directory if there is none."""
    directory = Path(start).resolve()
    if directory.is_file():
        directory = directory.parent
    found = _git(str(directory), "rev-parse", "--show-toplevel")
    return found or str(directory)


def sync(
    flows: Iterable[Flow],
    client: Client,
    *,
    repo: str = "",
    origin: dict[str, Any] | None = None,
    publish: bool = True,
    force: bool = False,
) -> list[SyncReport]:
    """Put declared flows into the store, refusing to overwrite canvas work.

    The order matters: the document first, so the nodes it names exist, then
    each node's generated body, then one publish. Both writes no-op on
    identical content, so a second sync with nothing changed commits nothing.
    """
    stamp = origin if origin is not None else origin_of(repo)
    reports = []
    for target in flows:
        reports.append(_sync_one(target, client, stamp, publish=publish, force=force))
    # Always, even when nothing here changed. A worker holds the imported
    # package in `sys.modules` for as long as it lives, so an edit to the
    # caller's own code is invisible until the process is retired — and that
    # edit is invisible to this function too, since it changes no shim and no
    # document. Refusing to refresh on a "no-op" sync would skip exactly the
    # case the refresh exists for.
    client.refresh_modules()
    return reports


def _sync_one(
    target: Flow,
    client: Client,
    origin: dict[str, Any],
    *,
    publish: bool,
    force: bool,
) -> SyncReport:
    report = SyncReport(target.name)
    stored = client.get_flow(target.name)
    version = 1
    # What the drift check already read, so the write loop below does not ask
    # the engine for every node's body a second time.
    known: dict[str, str] = {}
    if stored is None:
        report.created = True
    else:
        definition = stored.get("definition") or {}
        version = int(definition.get("version") or 1)
        if not force:
            known = _refuse_on_drift(client, target, definition)

    document = target.document(origin)
    moved = _moved_code(stored, document)
    saved = client.put_flow(document | {"version": version})
    # An engine older than the per-node digest parses the document without
    # those fields and stores it without them, silently: no error, no 404, and
    # a cache still keyed on the whole repository.
    kept = _digests(saved.get("definition"))
    report.forgot_code = any(
        digest and not kept.get(node_id)
        for node_id, digest in _digests(document).items()
    )
    stored_version = int((saved.get("definition") or {}).get("version") or version)
    if report.created or stored_version != version:
        # The store bumps the version only when the content actually changed,
        # so this is its own no-op detection rather than a second guess at it.
        report.changed.append("flow")
    version = stored_version
    # Named before the shims are compared, because this is the case a shim
    # comparison cannot see: the function is unchanged and something it calls
    # into is not.
    report.changed.extend(moved)

    for node_id, code in target.shims().items():
        if not report.created:
            stored_code = known.get(node_id)
            if stored_code is None:
                stored_code = client.get_source(target.name, node_id)
            if stored_code == code:
                continue
        client.put_source(target.name, node_id, code)
        if node_id not in report.changed:
            report.changed.append(node_id)

    has_draft = bool((client.get_flow(target.name) or {}).get("has_draft"))
    if publish and has_draft:
        client.publish(target.name, version)
        report.published = True
    else:
        # Only when one is actually there. "Nothing needed publishing" and
        # "left as a draft" are different answers to the same question.
        report.drafted = has_draft
    return report


def _digests(document: dict[str, Any] | None) -> dict[str, str]:
    """Each node's recorded code digest, by node id."""
    nodes = (document or {}).get("nodes") or []
    return {str(node.get("id")): str(node.get("code_digest") or "") for node in nodes}


def _moved_code(stored: dict[str, Any] | None, document: dict[str, Any]) -> list[str]:
    """The nodes whose reachable code changed since the last sync.

    A node is otherwise named only when its generated shim text changes, and
    the shim reflects the declared ports rather than what the function calls
    into — so editing a helper printed as nothing at all, which reads as
    "there was nothing to do".

    A node is named only when *both* sides carry a digest. An engine that does
    not record one drops the field on the way in, so comparing against nothing
    would report every node as changed on every sync, for ever — which is a
    worse answer than the silence this replaced.
    """
    if stored is None:
        return []
    before = _digests(stored.get("definition"))
    return [
        node_id
        for node_id, digest in _digests(document).items()
        if digest and before.get(node_id) and before[node_id] != digest
    ]


def _refuse_on_drift(
    client: Client, target: Flow, definition: dict[str, Any]
) -> dict[str, str]:
    """Stop before overwriting work that was done somewhere else.

    Two ways a stored flow is not ours to replace: it was drawn on the canvas
    and has no origin at all, or one of its node bodies no longer carries the
    line saying it was generated — which means somebody edited the code there.

    A node with *no* stored body is neither. That is what a sync interrupted
    part of the way through leaves, and it used to read as a canvas edit,
    because the engine answers the new-node template for it and the template
    carries no marker either. Re-running the sync is the fix for that state,
    so refusing to is precisely wrong.

    Returns the bodies it read, so the caller can compare against them without
    asking again.
    """
    if not definition.get("origin"):
        raise SyncError(
            f"flow '{target.name}' was not created by sync, so replacing it would "
            "discard whoever drew it. Rename yours, or pass --force."
        )
    seen: dict[str, str] = {}
    for stored_node in definition.get("nodes") or []:
        node_id = str(stored_node.get("id"))
        try:
            entry = client.get_source_entry(target.name, node_id)
        except ApiError:
            continue
        if entry.get("missing"):
            continue
        code = str(entry.get("code", ""))
        seen[node_id] = code
        if code and not code.startswith(MARKER):
            raise SyncError(
                f"node '{target.name}.{node_id}' was edited on the canvas, and "
                "syncing would throw that edit away. Copy it out, or pass --force."
            )
    return seen