"""Run one tick, observe everything, deliver nothing.

    python3 -m scaffold.preflight <launch-config.json>

Runs the author's `scan()` exactly once against live data, through the same loop production uses,
and reports what happened as one JSON document, written to the `result_path` the caller names.

Signals are collected and returned rather than delivered — the runtime's own delivery funnel is
closed for the duration and raises if used. That covers the sanctioned path, which is the one
authored code is supposed to take; a scanner reaching for a socket directly is caught statically at
the import depth, not stopped here. So: nothing leaves by the route the scaffold owns.

## What this adds beyond running the loop

**Reads are counted.** A tick that returns nothing having read nothing is indistinguishable, from
the outside, from a healthy scanner with nothing to trade. Counting successful reads is what tells
those apart — and it is the difference between reporting a strategy as fine and reporting that
nothing about it was established.

**Suppressed exceptions are recorded.** The author contract tells scanners to return `[]` on any
failure rather than crash, which is right for production and blinding for validation: a scanner
whose every read fails looks exactly like one with no setups. Tracing catches exceptions as they
are raised, before the author's own handler swallows them.

**The early exit is located.** When a tick returns without reading, the line it returned from turns
"something stopped early" into an edit.

**Author output cannot corrupt the result.** The tick shares this process's stdout, so the result
does not travel on it: the caller names a `result_path` and the document is written there, atomically.
Author stdout is captured as evidence and never parsed. Redirecting `sys.stdout` alone would only
cover `print()` — `os.write(1, ...)`, a subprocess, or a C extension write to the same descriptor.
"""

import io
import json
import math
import os
import sys
import time
import traceback
from typing import Any, Dict, List, Optional

from . import http_post
from .signal_envelope import candidate_to_signal

# Evidence is meant to be complete, but a scanner that prints in a loop will produce megabytes of
# it. Bounded with the ends kept and the middle marked, so the output stays honest about what was
# dropped rather than trailing off.
_STREAM_LIMIT = 32_768


def _bounded(text: str, limit: int = _STREAM_LIMIT) -> str:
    if len(text) <= limit:
        return text
    half = limit // 2
    dropped = len(text) - limit
    return "{}\n… truncated {} characters …\n{}".format(text[:half], dropped, text[-half:])


class _Watched(dict):
    """A response that remembers whether the scanner looked inside it.

    Fetching a payload and reading nothing out of it usually means the shape is not what the code
    expects — a renamed key, a nesting level missed. That failure is silent: the read succeeded, the
    scanner carried on, and the score was computed from a default. Recording access is what turns it
    into something reportable.
    """

    def __init__(self, source: dict) -> None:
        super().__init__(source)
        self.touched = False

    def _touch(self) -> None:
        self.touched = True

    def __getitem__(self, key: Any) -> Any:
        self._touch()
        return super().__getitem__(key)

    def get(self, key: Any, default: Any = None) -> Any:
        self._touch()
        return super().get(key, default)

    def __contains__(self, key: Any) -> bool:
        self._touch()
        return super().__contains__(key)

    def __iter__(self) -> Any:
        self._touch()
        return super().__iter__()

    def keys(self) -> Any:
        self._touch()
        return super().keys()

    def values(self) -> Any:
        self._touch()
        return super().values()

    def items(self) -> Any:
        self._touch()
        return super().items()


class CountingMcp:
    """Records every call and forwards it unchanged.

    A pass-through, not a stand-in: the real client is called, over the real network, and the real
    payload is returned. Only the record is new. A substituted payload could not surface the type
    and shape problems that make up most of what a live tick is for.

    Nothing here decides what a scanner may call. The client refuses money- and state-changing
    tools itself, before any transport is opened, and validation inherits that — which is the right
    place for it, since the same boundary protects production.
    """

    def __init__(self, inner: Any) -> None:
        self._inner = inner
        self.calls: List[Dict[str, Any]] = []
        self._watched: List[Dict[str, Any]] = []

    def call_tool(self, name: str, args: Optional[dict] = None) -> Any:
        started = time.monotonic()
        subject = _subject(args)
        try:
            result = self._inner.call_tool(name, args)
        except Exception as exc:  # noqa: BLE001
            # Deliberately not BaseException: a tick timeout unwinds through here and must not be
            # recorded as a failed read or, worse, absorbed.
            record = {
                "tool": name,
                "ms": round((time.monotonic() - started) * 1000),
                "ok": False,
                "error_type": type(exc).__name__,
                "error": str(exc)[:500],
            }
            if subject is not None:
                record["subject"] = subject
            # The client refuses money- and state-changing tools before it opens a transport, and
            # owns the list of which those are. Marked rather than re-derived here: a second
            # opinion about what counts as a write would be a worse one, and would drift.
            if isinstance(exc, PermissionError):
                record["blocked"] = True
            self.calls.append(record)
            raise

        record: Dict[str, Any] = {
            "tool": name,
            "ms": round((time.monotonic() - started) * 1000),
            "ok": _payload_ok(result),
        }
        if subject is not None:
            record["subject"] = subject
        if isinstance(result, dict):
            # Inspected before wrapping — reading keys here would otherwise count as the scanner
            # having looked. Only recorded when it might explain something: a tick that read
            # successfully and emitted nothing is the case where the shape is worth seeing.
            record["response_keys"] = sorted(result.keys())[:20]
            if "_cast_dropped" in result:
                record["cast_dropped"] = True
        if not record["ok"]:
            record["error"] = _envelope_error(result)
        self.calls.append(record)

        if isinstance(result, dict) and record["ok"]:
            watched = _Watched(result)
            self._watched.append({"record": record, "response": watched})
            return watched
        return result

    def settle_unread(self) -> None:
        """Mark the responses nothing ever looked inside. Called once the tick is over."""
        for entry in self._watched:
            if not entry["response"].touched:
                entry["record"]["unread"] = True

    def __getattr__(self, name: str) -> Any:
        # Keeps the wrapper transparent for anything beyond the documented surface.
        return getattr(self._inner, name)

    @property
    def reads_ok(self) -> int:
        return sum(1 for c in self.calls if c["ok"])

    @property
    def reads_failed(self) -> int:
        return len(self.calls) - self.reads_ok


#: Argument names that carry the instrument a call is about, checked in this order. `asset` and
#: `coin` are what the senpi tools actually take; the rest are here so a tool this file has never
#: met still gets its subject named rather than reported as "some instrument".
_SUBJECT_KEYS = ("asset", "coin", "symbol", "ticker", "instrument", "assets", "coins", "symbols")

#: A subject is a label, not a payload. Long enough for a dex-prefixed name or a short basket.
_SUBJECT_LIMIT = 120


def _subject(args: Optional[dict]) -> Optional[str]:
    """The instrument a call is about, taken from the arguments as they were passed.

    Recorded here, at the call, and never recovered from the error message afterwards. A finding
    that names an instrument has to name the one that was actually sent: the error is the server's
    prose, free to reword, to quote a normalised form, or to name nothing at all — and a diagnosis
    is worth less than nothing if the name in it is reconstructed.

    Absent rather than guessed when no argument carries a name. A call with no subject is an
    ordinary thing (`market_list_instruments` takes none), and the reader is better served by a
    finding that says it saw none than by one that invents one.
    """
    if not isinstance(args, dict):
        return None
    for key in _SUBJECT_KEYS:
        value = args.get(key)
        if isinstance(value, str) and value.strip():
            return value[:_SUBJECT_LIMIT]
        if isinstance(value, (list, tuple)):
            names = [v for v in value if isinstance(v, str) and v.strip()]
            if names:
                return ", ".join(names[:10])[:_SUBJECT_LIMIT]
    return None


def _payload_ok(payload: Any) -> bool:
    """Whether a returned payload represents a successful read.

    Not simply "did not raise". The transport raises on a protocol-level error, but an
    application-level failure comes back as an ordinary response carrying a false success flag —
    counting that as a read would reproduce, one layer down, exactly the confusion this is here to
    remove. Shape-tolerant, because not every response carries the flag.
    """
    return not (isinstance(payload, dict) and "success" in payload and not payload["success"])


def _envelope_error(payload: Any) -> str:
    if isinstance(payload, dict):
        err = payload.get("error")
        if isinstance(err, dict):
            return str(err.get("message") or err)
        if err is not None:
            return str(err)
    return "the response reported failure"


class _Tracer:
    """Records exceptions raised inside the scanner's own files, and where it returned from.

    A trace function sees an exception at the moment it is raised, not when it propagates — so it
    sees the ones the author's handler is about to swallow, which is the entire point.

    Three things this deliberately does NOT do:

    **Line tracing.** Returning a handler from `'call'` arms per-line tracing for that frame, which
    costs ~12x on compute-heavy code — enough to make a scanner blow a budget here that it meets in
    production, which would be this command lying about production. `'exception'` and `'return'`
    still fire with lines off.

    **Module frames.** The tick runs after the entrypoint is imported, but the import itself is
    traced, and `try: import optional_dep / except ImportError` is a completely ordinary module-level
    idiom. Reporting it as a suppressed exception would flag a clean scanner.

    **Unbounded growth.** A `try/except` inside a large loop raises once per row; the report caps at
    50, but building 100k dicts to throw away is the sort of cost that only shows up on the scanner
    least able to afford it.
    """

    #: Kept during the tick, not just at report time — see the class docstring.
    LIMIT = 50

    def __init__(self, root: str) -> None:
        # The separator matters: a bare prefix test makes root `/x/scanner` claim
        # `/x/scanner-utils/evil.py`, which is plausible with sibling per-instance scanner dirs.
        self.root = os.path.abspath(root)
        self._prefix = self.root.rstrip(os.sep) + os.sep
        self.suppressed: List[Dict[str, Any]] = []
        self.dropped = 0
        self.last_return: Optional[Dict[str, Any]] = None
        # Grouping is by object identity, held as a live reference — `id()` alone is unsound here,
        # because CPython reuses an address as soon as the previous exception is collected, so two
        # unrelated failures of the same class routinely share one. Exactly one reference is kept
        # at a time (replaced on the next raise, dropped at the end of the tick), so this cannot
        # pin a chain of tracebacks.
        self._live_exc: Any = None
        self._group = 0

    def _own(self, frame: Any) -> bool:
        return os.path.abspath(frame.f_code.co_filename).startswith(self._prefix)

    def __call__(self, frame: Any, event: str, arg: Any) -> Any:
        if event == "call":
            if not self._own(frame) or frame.f_code.co_name == "<module>":
                return None
            # Exceptions and returns still fire; per-line events do not.
            frame.f_trace_lines = False
            return self
        if event == "exception":
            exc_type, exc_value, _tb = arg
            # One exception propagating outward fires once per author frame. Consecutive events for
            # the same object are one failure; a different object starts a new group.
            if exc_value is not self._live_exc:
                self._live_exc = exc_value
                self._group += 1
            if len(self.suppressed) >= self.LIMIT:
                self.dropped += 1
                return self
            self.suppressed.append(
                {
                    "file": os.path.relpath(frame.f_code.co_filename, self.root),
                    "line": frame.f_lineno,
                    "type": exc_type.__name__,
                    "message": str(exc_value)[:300],
                    "_group": self._group,
                }
            )
        elif event == "return":
            self.last_return = {
                "file": os.path.relpath(frame.f_code.co_filename, self.root),
                "line": frame.f_lineno,
                "function": frame.f_code.co_name,
            }
        return self

    def claim_ending(self, error_type: str) -> Optional[Dict[str, Any]]:
        """Take every entry belonging to the exception that ended the tick; return its raise site.

        Two things this gets right that matching one entry by type does not.

        The ending exception is not something the scanner "carried on from" — leaving *any* of its
        entries in `suppressed` says the opposite of what happened. It fires once per author frame
        it propagates through, so a helper raising into `scan()` leaves two, and removing one leaves
        the other reading as a caught failure.

        And of those entries, the FIRST is the raise site; the later ones are call sites further out.
        Reporting a call site as the location points an edit a line above the defect. The tick ends
        at the last exception raised, so the last entry of that type identifies it — and identity,
        not type, gathers the rest of its frames.
        """
        for entry in reversed(self.suppressed):
            if entry["type"] != error_type:
                continue
            group = entry.get("_group")
            mine = [e for e in self.suppressed if e.get("_group") == group]
            self.suppressed = [e for e in self.suppressed if e.get("_group") != group]
            return mine[0]
        return None

    def release(self) -> None:
        """Drop the reference held for grouping, so no traceback outlives the tick."""
        self._live_exc = None

    def report(self) -> List[Dict[str, Any]]:
        """The suppressed list as it goes on the wire — identity stripped, it is bookkeeping."""
        return [{k: v for k, v in e.items() if k != "_group"} for e in self.suppressed]


class _CollectingSink:
    """Takes what the tick produced, and what went wrong, without sending either anywhere.

    Both halves matter. The loop does not raise a scanner's own failure — it records the tick as
    failed and reports it through the sink's error channel, so a preflight that only watched for an
    exception would call a tick that raised "ok". Having the attribute is what opts into that
    channel, which is why this is an object rather than a function.
    """

    def __init__(self) -> None:
        self.signals: List[Any] = []
        self.errors: List[Dict[str, Any]] = []

    def __call__(self, candidates: Any) -> None:
        self.signals.extend(candidates)

    def post_error(self, error: Dict[str, Any]) -> None:
        self.errors.append(error)


def run_preflight(config: Dict[str, Any]) -> Dict[str, Any]:
    """Execute one tick and describe it. Never raises for a scanner's own failure."""
    from . import scaffold as scaffold_mod

    root = config["path"]
    sink = _CollectingSink()
    tracer = _Tracer(root)

    inner = scaffold_mod._build_senpi_mcp()  # noqa: SLF001 — the module's own construction seam
    if inner is None:
        return {
            "ok": False,
            "reason": "no_credentials",
            "detail": "no data-client credentials are configured in this environment",
        }
    counter = CountingMcp(inner)

    captured_out = io.StringIO()
    captured_err = io.StringIO()
    real_stdout, real_stderr = sys.stdout, sys.stderr
    status = "ok"
    failure: Optional[Dict[str, Any]] = None

    http_post.PREFLIGHT = True
    sys.stdout = captured_out
    sys.stderr = captured_err
    sys.settrace(tracer)
    try:
        scaffold_mod.run_scaffold(
            root,
            config["entrypoint"],
            interval=float(config["interval_seconds"]),
            timeout_seconds=config.get("timeout_seconds"),
            inputs=config.get("inputs") or {},
            state_path=config.get("state_path"),
            scanner_name=config["scanner_name"],
            wallet=config["wallet"],
            default_signal_validity_seconds=int(config["default_signal_validity_seconds"]),
            state_history_max_count=config.get("state_history_max_count") or 0,
            sink=sink,
            max_ticks=1,
            install_signal_handlers=False,
            senpi_mcp=counter,
            dry_run=True,
        )
    except BaseException as exc:  # noqa: BLE001 — a scanner may raise anything at all
        # Only launch failures reach here: the loop handles a failing tick itself.
        status = "error"
        failure = {
            "type": type(exc).__name__,
            "message": str(exc)[:1000],
            "traceback": _bounded("".join(traceback.format_exception(type(exc), exc, exc.__traceback__))),
        }
    finally:
        sys.settrace(None)
        sys.stdout, sys.stderr = real_stdout, real_stderr
        http_post.PREFLIGHT = False
        counter.settle_unread()
        tracer.release()

    # A tick that failed was reported through the sink rather than raised. The loop's error record
    # carries the type and message but not the location — so it is composed with what the tracer
    # saw. That gives a file and line, which is what an edit needs and a stack string is not.
    if status == "ok" and sink.errors:
        first = sink.errors[0]
        status = "timeout" if first.get("type") == "timeout" else "error"
        error_type = first.get("error_type") or first.get("type") or "tick_error"

        # The loop reports a failed tick through the sink rather than raising it, and its record
        # carries the type and message but not the location — so it is composed with what the
        # tracer saw. A file and line is what an edit needs; a stack string is not.
        where = tracer.claim_ending(error_type)

        message = str(first.get("message", ""))[:1000]
        if not message:
            # A malformed return arrives with an empty message and its reason only in stderr, which
            # is the one place a finding cannot quote. Recover what the loop knows.
            message = str(first.get("error") or first.get("reason") or error_type)[:1000]

        failure = {
            "type": error_type,
            "message": message,
            "where": {"file": where["file"], "line": where["line"]} if where else None,
            "reported_by": "scaffold",
        }

    return {
        "ok": True,
        "status": status,
        "failure": failure,
        "reads": counter.calls,
        "reads_ok": counter.reads_ok,
        "reads_failed": counter.reads_failed,
        "signals": _to_wire(sink.signals, config),
        "suppressed": tracer.report(),
        "suppressed_dropped": tracer.dropped,
        "last_return": tracer.last_return,
        "stdout": _bounded(captured_out.getvalue()),
        "stderr": _bounded(captured_err.getvalue()),
    }


def _to_wire(candidates: List[Any], config: Dict[str, Any]) -> List[Dict[str, Any]]:
    """Build each candidate into the shape intake would actually receive.

    Uses the runtime's own envelope builder rather than describing what it would do. Judging a
    candidate on its pre-envelope shape would answer a different question from "would this be
    accepted", which is the only question worth asking here.
    """
    produced_at = int(time.time() * 1000)
    out: List[Dict[str, Any]] = []
    for candidate in candidates:
        rejections: List[Dict[str, Any]] = []
        signal = candidate_to_signal(
            candidate,
            produced_at,
            signal_data_schema=config.get("signal_data_schema"),
            default_signal_validity_seconds=int(config["default_signal_validity_seconds"]),
            on_reject=rejections.append,
        )
        out.append(
            {
                "emitted": candidate if isinstance(candidate, dict) else repr(candidate),
                "wire": signal,
                "rejected_by_scaffold": rejections,
            }
        )
    return out


def _finite(value: Any) -> Any:
    """Replace non-finite floats with their text, recursively.

    A 0/0 in scoring produces NaN, and `json.dumps` writes it as a bare `NaN` token that no strict
    parser accepts — so the tick that surfaced the bug would be reported as the harness breaking.
    Keeping the value as text says what happened instead of losing the document over it.
    """
    if isinstance(value, float):
        return value if math.isfinite(value) else repr(value)
    if isinstance(value, dict):
        return {k: _finite(v) for k, v in value.items()}
    if isinstance(value, (list, tuple)):
        return [_finite(v) for v in value]
    return value


def _serialize(result: Dict[str, Any]) -> str:
    """One document, always parseable.

    `allow_nan=False` turns a non-finite float into a raised error rather than a token that only
    Python reads back. Anything still unserialisable after sanitising — a circular candidate, an
    exotic object — must not cost the whole report, so the signals are dropped to text and the rest
    survives.
    """
    try:
        return json.dumps(_finite(result), allow_nan=False, default=str)
    except (ValueError, TypeError, RecursionError) as exc:
        salvaged = dict(result)
        salvaged["signals"] = [{"emitted": "<unserialisable>", "wire": None, "rejected_by_scaffold": []}]
        salvaged["serialization_error"] = str(exc)[:300]
        try:
            return json.dumps(_finite(salvaged), allow_nan=False, default=str)
        except Exception:  # noqa: BLE001 — a report is worth more than the detail in it
            return json.dumps({"ok": True, "status": result.get("status", "error"),
                               "serialization_error": str(exc)[:300]})


def _emit(result: Dict[str, Any], config: Dict[str, Any]) -> None:
    """Deliver the result where author code cannot reach it.

    The tick runs in this process and shares its stdout. Swapping `sys.stdout` catches `print()`,
    which is what authors do — but `os.write(1, ...)`, a subprocess, a C extension, or a thread
    printing after the restore all write to the same descriptor, and the reader parses the whole
    buffer. That fails closed (never a false pass) but reports a scanner's debug line as a broken
    environment, which points at the wrong person entirely.

    The caller names a file it owns; writing there makes the guarantee structural rather than
    conventional. Stdout stays available as evidence, and is never parsed.
    """
    document = _serialize(result)
    path = config.get("result_path")
    if path:
        tmp = "{}.tmp".format(path)
        with open(tmp, "w", encoding="utf-8") as handle:
            handle.write(document)
        os.replace(tmp, path)   # the reader never sees a half-written document
        return
    # No path given: the caller is reading stdout (older harness, or a direct invocation).
    sys.stdout.write(document)


def main(argv: Optional[List[str]] = None) -> int:
    argv = list(sys.argv if argv is None else argv)
    if len(argv) < 2:
        sys.stderr.write("usage: python3 -m scaffold.preflight <launch-config.json>\n")
        return 2
    try:
        with open(argv[1], "r", encoding="utf-8") as handle:
            config = json.load(handle)
    except (OSError, json.JSONDecodeError, UnicodeDecodeError) as exc:
        sys.stdout.write(json.dumps({"ok": False, "reason": "bad_config", "detail": str(exc)}))
        return 0

    _emit(run_preflight(config), config)
    return 0


if __name__ == "__main__":
    sys.exit(main())