#!/usr/bin/env python3
"""Verify FlintTrade against the exact Kotak Neo v3 main and release commits.

The gate builds one disposable environment per immutable upstream revision. It
installs only locked dependencies before the selected Git SDK, installs the
four FlintTrade packages needed by the adapter without dependency resolution,
then executes an offline import/API probe. No broker credentials are read and
no broker endpoint is contacted.
"""

from __future__ import annotations

import argparse
import json
import os
import re
import shutil
import subprocess
import sys
import tempfile
import tomllib
from collections.abc import Callable, Mapping, Sequence
from dataclasses import dataclass
from pathlib import Path
from typing import Any
from urllib.parse import urlsplit

REPO = Path(__file__).resolve().parents[1]
IS_WINDOWS = os.name == "nt"
_KOTAK_NAME = "kotakneoapi"
_KOTAK_HOMEPAGE = "https://github.com/Kotak-Neo/kotak-neo-python"
_FULL_GIT_HASH = re.compile(r"[0-9a-f]{40}")


class ContractError(RuntimeError):
    """Raised when provenance, compatibility, or migration evidence is incomplete."""


@dataclass(frozen=True)
class ContractConfig:
    """Authoritative dual-track pins loaded from ``brokers.lock``."""

    repo_url: str
    version: str
    release_version: str
    runtime_main_commit: str
    release_tag: str
    release_commit: str


def load_contract_config(repo: Path = REPO) -> ContractConfig:
    """Load and validate the single Kotak Neo attestation record."""

    lock_path = repo / "brokers.lock"
    try:
        data = tomllib.loads(lock_path.read_text(encoding="utf-8"))
    except (OSError, tomllib.TOMLDecodeError) as exc:
        raise ContractError(f"cannot read authoritative broker lock: {lock_path}") from exc
    broker_rows = data.get("broker")
    if not isinstance(broker_rows, list) or not all(isinstance(entry, dict) for entry in broker_rows):
        raise ContractError("brokers.lock broker inventory is malformed")
    entries = [entry for entry in broker_rows if entry.get("name") == _KOTAK_NAME]
    if len(entries) != 1:
        raise ContractError("brokers.lock must contain exactly one kotakneoapi record")
    entry = entries[0]
    homepage = entry.get("homepage")
    version = entry.get("version")
    release_version = entry.get("release_version")
    runtime_main_commit = entry.get("source_commit")
    release_tag = entry.get("release_tag")
    release_commit = entry.get("release_commit")
    if homepage != _KOTAK_HOMEPAGE:
        raise ContractError("brokers.lock Kotak Neo homepage is not the official upstream")
    if not isinstance(version, str) or not re.fullmatch(r"\d+\.\d+\.\d+", version):
        raise ContractError("brokers.lock Kotak Neo version is malformed")
    if not isinstance(release_version, str) or not re.fullmatch(r"\d+\.\d+\.\d+", release_version):
        raise ContractError("brokers.lock Kotak Neo release version is malformed or missing")
    if release_tag != f"v{release_version}":
        raise ContractError("brokers.lock Kotak Neo release tag does not match its release version")
    for label, value in (
        ("runtime main commit", runtime_main_commit),
        ("release commit", release_commit),
    ):
        if not isinstance(value, str) or _FULL_GIT_HASH.fullmatch(value) is None:
            raise ContractError(f"brokers.lock Kotak Neo {label} is not a full Git hash")
    return ContractConfig(
        repo_url=f"{homepage}.git",
        version=version,
        release_version=release_version,
        runtime_main_commit=runtime_main_commit,
        release_tag=release_tag,
        release_commit=release_commit,
    )


@dataclass(frozen=True)
class SdkTrack:
    """One immutable upstream contract target."""

    name: str
    revision: str
    version: str
    repo_url: str
    release_tag: str | None = None


def build_tracks(config: ContractConfig) -> tuple[SdkTrack, SdkTrack]:
    """Build the runtime-main and stable-release tracks from one lock record."""

    return (
        SdkTrack("runtime-main", config.runtime_main_commit, config.version, config.repo_url),
        SdkTrack(
            f"release-{config.release_tag}",
            config.release_commit,
            config.release_version,
            config.repo_url,
            config.release_tag,
        ),
    )


# Compatibility aliases for the unit-test surface. These values are derived
# from the authoritative lock rather than maintained as a second pin set.
_DEFAULT_CONFIG = load_contract_config()
KOTAK_REPO = _DEFAULT_CONFIG.repo_url
KOTAK_VERSION = _DEFAULT_CONFIG.version
RUNTIME_MAIN_COMMIT = _DEFAULT_CONFIG.runtime_main_commit
RELEASE_TAG = _DEFAULT_CONFIG.release_tag
RELEASE_COMMIT = _DEFAULT_CONFIG.release_commit
TRACKS = build_tracks(_DEFAULT_CONFIG)

SCANNER_TARGETS = (
    "packages/integrations/gateway/src/flinttrade_gateway/brokers/kotakneo.py",
    "packages/integrations/gateway/src/flinttrade_gateway/brokers/kotakneo_sdk.py",
    "packages/integrations/gateway/src/flinttrade_gateway/brokers/kotakneo_streaming.py",
    "packages/integrations/gateway/src/flinttrade_gateway/native_login.py",
    "packages/integrations/gateway/src/flinttrade_gateway/connected_read_smoke.py",
    "scripts/probe_native_broker_live.py",
    "scripts/probe_kotakneo_live.py",
)

Run = Callable[..., subprocess.CompletedProcess[str]]


def host_python(repo: Path = REPO) -> Path:
    """Return the repository-managed interpreter required by the local gate."""

    candidate = repo / ".venv" / ("Scripts/python.exe" if IS_WINDOWS else "bin/python")
    if not candidate.is_file():
        raise ContractError(f"repository-managed Python is missing: {candidate}")
    return candidate


def environment_python(environment: Path) -> Path:
    """Return the interpreter path inside a disposable uv environment."""

    return environment / ("Scripts/python.exe" if IS_WINDOWS else "bin/python")


_PASSTHROUGH_ENVIRONMENT = frozenset(
    {
        "ALL_PROXY",
        "COMSPEC",
        "CURL_CA_BUNDLE",
        "HTTPS_PROXY",
        "HTTP_PROXY",
        "LANG",
        "LC_ALL",
        "NO_PROXY",
        "PATH",
        "PATHEXT",
        "REQUESTS_CA_BUNDLE",
        "SSL_CERT_DIR",
        "SSL_CERT_FILE",
        "SYSTEMROOT",
        "WINDIR",
        "all_proxy",
        "https_proxy",
        "http_proxy",
        "no_proxy",
    }
)


def build_subprocess_environment(
    workspace: Path,
    *,
    source: Mapping[str, str] | None = None,
) -> dict[str, str]:
    """Return a minimal child environment with isolated home/cache paths.

    Installation needs the host PATH, platform runtime variables, certificate
    paths, and (where configured) a network proxy. Broker, cloud, API, Python,
    Git, and package-index credentials are deliberately not inherited.
    """

    parent = os.environ if source is None else source
    environment = {
        key: value for key, value in parent.items() if key in _PASSTHROUGH_ENVIRONMENT and isinstance(value, str)
    }
    environment.setdefault("PATH", os.defpath)
    home = workspace / "home"
    cache = workspace / "cache"
    temporary = workspace / "tmp"
    for path in (home, cache, temporary):
        path.mkdir(parents=True, exist_ok=True)
    environment.update(
        {
            "GIT_CONFIG_GLOBAL": os.devnull,
            "GIT_CONFIG_NOSYSTEM": "1",
            "GIT_TERMINAL_PROMPT": "0",
            "HOME": str(home),
            "PIP_CONFIG_FILE": os.devnull,
            "PYTHONNOUSERSITE": "1",
            "PYTHONUTF8": "1",
            "TEMP": str(temporary),
            "TMP": str(temporary),
            "TMPDIR": str(temporary),
            "USERPROFILE": str(home),
            "UV_CACHE_DIR": str(cache / "uv"),
            "XDG_CACHE_HOME": str(cache / "xdg"),
        }
    )
    return environment


def build_export_command(requirements: Path, *, repo: Path = REPO) -> list[str]:
    """Build the frozen export command for the gateway's non-workspace base."""

    del repo  # The caller supplies the repository as the command cwd.
    return [
        "uv",
        "export",
        "--frozen",
        "--package",
        "flinttrade-gateway",
        "--no-dev",
        "--no-emit-workspace",
        "--no-emit-package",
        "kotakneoapi",
        "--output-file",
        str(requirements),
    ]


def build_track_commands(
    track: SdkTrack,
    environment: Path,
    requirements: Path,
    *,
    repo: Path = REPO,
    gateway_project: Path | None = None,
) -> list[list[str]]:
    """Build the ordered installation/check commands for one SDK track."""

    python = environment_python(environment)
    editable_paths = (
        repo / "packages/core/core",
        repo / "packages/core/data",
        repo / "packages/services/engine",
        gateway_project or repo / "packages/integrations/gateway",
    )
    editable_args: list[str] = []
    for path in editable_paths:
        editable_args.extend(("--editable", str(path)))
    return [
        ["uv", "venv", str(environment), "--python", str(host_python(repo))],
        ["uv", "pip", "install", "--python", str(python), "--require-hashes", "-r", str(requirements)],
        [
            "uv",
            "pip",
            "install",
            "--python",
            str(python),
            "--require-hashes",
            "--only-binary=:all:",
            "--no-deps",
            "-r",
            str(repo / "broker-sdk-build.lock"),
        ],
        [
            "uv",
            "pip",
            "install",
            "--python",
            str(python),
            "--no-build-isolation",
            "--no-deps",
            f"git+{track.repo_url}@{track.revision}",
        ],
        [
            "uv",
            "pip",
            "install",
            "--python",
            str(python),
            "--no-deps",
            *editable_args,
        ],
        ["uv", "pip", "check", "--python", str(python)],
    ]


def _normalise_distribution(name: object) -> str:
    return re.sub(r"[-_.]+", "-", str(name).strip().lower())


def _require_mapping(value: object, label: str) -> Mapping[str, Any]:
    if not isinstance(value, Mapping):
        raise ContractError(f"Kotak Neo {label} is missing or malformed")
    return value


def validate_probe_result(result: Mapping[str, Any], track: SdkTrack) -> None:
    """Validate exact distribution, namespace, import, and offline probe evidence."""

    distributions = _require_mapping(result.get("distributions"), "distribution inventory")
    if "neo-api-client" in distributions:
        raise ContractError("forbidden neo-api-client v2 distribution is installed")
    installed = _require_mapping(distributions.get("kotakneoapi"), "kotakneoapi distribution")
    if installed.get("version") != track.version:
        raise ContractError(f"Kotak Neo version must be exactly {track.version}")

    counts = result.get("distribution_counts")
    if not isinstance(counts, Mapping) or counts.get("kotakneoapi") != 1:
        raise ContractError("kotakneoapi distribution inventory is ambiguous")

    owners = result.get("namespace_owners")
    if owners != ["kotakneoapi"]:
        raise ContractError("neo_api_client namespace must have exactly one kotakneoapi owner")

    direct = _require_mapping(installed.get("direct_url"), "Git provenance")
    raw_url = direct.get("url")
    try:
        parsed = urlsplit(raw_url if isinstance(raw_url, str) else "")
        exact_origin = (
            parsed.scheme == "https"
            and parsed.hostname == "github.com"
            and f"{parsed.scheme}://{parsed.hostname}{parsed.path}" == track.repo_url
            and parsed.username is None
            and parsed.password is None
            and parsed.port is None
            and not parsed.query
            and not parsed.fragment
        )
    except ValueError as exc:
        raise ContractError("Kotak Neo Git origin is malformed") from exc
    if not exact_origin:
        raise ContractError("Kotak Neo Git origin must be the exact HTTPS upstream")

    vcs = _require_mapping(direct.get("vcs_info"), "VCS provenance")
    commit = vcs.get("commit_id")
    if vcs.get("vcs") != "git" or not isinstance(commit, str) or not re.fullmatch(r"[0-9a-f]{40}", commit):
        raise ContractError("Kotak Neo provenance must contain a full Git commit")
    if commit != track.revision:
        raise ContractError(f"Kotak Neo commit does not match {track.name}")
    if vcs.get("requested_revision") != track.revision:
        raise ContractError(f"Kotak Neo requested revision does not match {track.name}")

    module_path = result.get("module_path")
    environment_root = result.get("environment_root")
    if not isinstance(module_path, str) or not isinstance(environment_root, str):
        raise ContractError("target-environment import evidence is missing")
    try:
        imported_from_environment = Path(module_path).resolve().is_relative_to(Path(environment_root).resolve())
    except (OSError, RuntimeError, ValueError):
        imported_from_environment = False
    if not imported_from_environment:
        raise ContractError("neo_api_client was not imported from the target environment")

    cwd_files = result.get("cwd_files")
    if cwd_files != []:
        raise ContractError("Kotak Neo SDK created a log or another file in the probe working directory")
    if result.get("contract_ok") is not True:
        raise ContractError("Kotak Neo offline API contract probe did not complete")
    if result.get("read_contract_ok") is not True:
        raise ContractError("Kotak Neo offline read contract probe did not complete")
    if result.get("read_failure_contract_ok") is not True:
        raise ContractError("Kotak Neo offline read failure contract probe did not complete")
    if result.get("feed_cancellation_contract_ok") is not True:
        raise ContractError("Kotak Neo offline feed cancellation contract probe did not complete")


def parse_probe_output(stdout: str) -> dict[str, Any]:
    """Parse a probe's single JSON object, rejecting banners or partial output."""

    try:
        value = json.loads(stdout)
    except (TypeError, json.JSONDecodeError) as exc:
        raise ContractError("Kotak Neo probe did not return one JSON object") from exc
    if not isinstance(value, dict):
        raise ContractError("Kotak Neo probe JSON must be an object")
    return value


_PROBE_SCRIPT = r"""
from __future__ import annotations

import asyncio
import importlib.metadata as md
import inspect
import json
import os
import re
import selectors
import socket
import sys
from datetime import date
from pathlib import Path


def denied(*_args, **_kwargs):
    raise AssertionError("network access is forbidden during the Kotak Neo contract probe")


def deny_network_and_process(event, _args):
    if event.startswith("socket.") or event in {
        "subprocess.Popen",
        "os.system",
        "os.posix_spawn",
        "os.spawn",
        "os.exec",
        "os.fork",
        "os.forkpty",
        "pty.spawn",
    }:
        denied()


sys.addaudithook(deny_network_and_process)
socket.socket.connect = denied
socket.socket.connect_ex = denied
socket.create_connection = denied

# Fail closed if this interpreter does not emit the socket-construction audit
# event. This also proves the guard before any third-party module is imported.
try:
    socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
except AssertionError:
    pass
else:
    raise AssertionError("network audit guard is not active")

import httpx
import neo_api_client
import neo_api_client.neo_api as neo_module
import neo_api_client.websocket.feed.client as feed_module
from neo_api_client import NeoAPI
from neo_api_client.websocket.feed import (
    SFeedIndex,
    SFeedScrip,
    SFeedScripLite,
    SFeedWebSocket,
    WsToken,
)
from neo_api_client.websocket.orderfeed import OrderFeedWebSocket, OrderUpdate, PositionUpdate

from flinttrade_gateway.brokers import kotakneo_streaming
from flinttrade_gateway.brokers.kotakneo_sdk import KotakNeoSdkSession, validate_read_envelope
from flinttrade_core.exceptions import BrokerInternal


def canonical(name):
    return re.sub(r"[-_.]+", "-", str(name).strip().lower())


distributions = {}
counts = {}
owners = []
for distribution in md.distributions():
    name = canonical(distribution.metadata.get("Name", ""))
    if name in {"kotakneoapi", "neo-api-client"}:
        direct_text = distribution.read_text("direct_url.json")
        try:
            direct = json.loads(direct_text) if direct_text else None
        except (TypeError, ValueError):
            direct = None
        distributions[name] = {"version": distribution.version, "direct_url": direct}
        counts[name] = counts.get(name, 0) + 1
    top_level = (distribution.read_text("top_level.txt") or "").splitlines()
    record_owns = any(
        str(file).replace("\\", "/").split("/")[0] == "neo_api_client"
        for file in (distribution.files or ())
    )
    if "neo_api_client" in top_level or record_owns:
        owners.append(name)

expected = {
    "totp_login": ("mobile_number", "ucc", "totp"),
    "totp_validate": ("mpin",),
    "place_order": (
        "exchange_segment", "product", "price", "order_type", "quantity", "validity",
        "trading_symbol", "transaction_type", "amo", "disclosed_quantity", "trigger_price", "tag",
    ),
    "modify_order": (
        "order_id", "price", "order_type", "quantity", "validity", "trigger_price",
        "disclosed_quantity", "amo",
    ),
    "cancel_order": ("order_id", "amo", "isVerify"),
    "margin_required": (
        "exchange_segment", "price", "order_type", "product", "quantity", "instrument_token",
        "transaction_type", "trigger_price", "broker_name", "branch_id", "stop_loss_type",
        "stop_loss_value", "square_off_type", "square_off_value", "trailing_stop_loss",
        "trailing_sl_value",
    ),
    "create_websocket": ("url", "kwargs"),
    "create_order_feed": ("kwargs",),
    "whatsmyip": (),
    "order_report": ("order_id",),
    "order_history": ("order_id",),
    "trade_report": (),
    "positions": (),
    "holdings": (),
    "limits": (),
    "quotes": ("instrument_tokens", "quote_type"),
    "scrip_master": ("exchange_segment",),
    "search_scrip": (
        "exchange_segment", "symbol", "expiry", "option_type", "strike_price", "ignore_50multiple",
    ),
    "expiries": ("exchange", "underlying", "instrument_type"),
    "option_chain": ("exchange", "underlying", "expiry", "instrument_type", "count"),
    "historical_data": ("neosymbol", "interval", "from_date", "to_date"),
    "logout": (),
}
for name, parameters in expected.items():
    assert tuple(inspect.signature(getattr(NeoAPI, name)).parameters)[1:] == parameters, name
assert tuple(inspect.signature(NeoAPI).parameters) == (
    "consumer_key", "environment", "access_token", "neo_fin_key", "transport", "limits", "http2", "timeout",
)
assert tuple(inspect.signature(WsToken).parameters) == ("exchange_segment", "instrument_token")

feed_signatures = {
    SFeedWebSocket: {
        "connect": ("self",),
        "close": ("self",),
        "subscribe_scrips": ("self", "tokens"),
        "unsubscribe_scrips": ("self", "tokens"),
        "subscribe_depth": ("self", "tokens"),
        "unsubscribe_depth": ("self", "tokens"),
        "subscribe_index": ("self", "tokens"),
        "unsubscribe_index": ("self", "tokens"),
        "__aiter__": ("self",),
    },
    OrderFeedWebSocket: {
        "connect": ("self",),
        "close": ("self",),
        "__aiter__": ("self",),
    },
}
for feed_class, methods in feed_signatures.items():
    for name, parameters in methods.items():
        assert tuple(inspect.signature(getattr(feed_class, name)).parameters) == parameters, (
            feed_class.__name__, name
        )

for removed in ("NeoWebSocket", "HSWebSocket", "HSIWebSocket"):
    assert not hasattr(neo_api_client, removed), removed
for method, args in (
    ("subscribe", ([{"instrument_token": "11536", "exchange_segment": "nse_cm"}],)),
    ("un_subscribe", ([{"instrument_token": "11536", "exchange_segment": "nse_cm"}],)),
    ("subscribe_to_orderfeed", ()),
):
    instance = NeoAPI.__new__(NeoAPI)
    try:
        getattr(instance, method)(*args)
    except NotImplementedError:
        pass
    else:
        raise AssertionError(f"legacy {method} unexpectedly remains active")


class FakeTotp:
    def __init__(self, _api_client):
        pass

    def totp_login(self, **_kwargs):
        return {"data": {"status": "success", "token": "view", "sid": "view-sid", "ucc": "SYNTHETIC", "kType": "View"}}

    def totp_validate(self, **_kwargs):
        return {"data": {"status": "success", "token": "trade", "sid": "trade-sid", "baseUrl": "https://example.invalid", "kType": "Trade"}}


class FakeOrder:
    def __init__(self, _api_client):
        pass

    def order_placing(self, **_kwargs):
        return {"stat": "Ok", "nOrdNo": "SYNTHETIC"}

    def order_cancelling(self, **_kwargs):
        return {"stat": "Ok", "nOrdNo": "SYNTHETIC"}


class FakeOrders:
    def __init__(self, _api_client):
        pass

    def ordered_books(self):
        return {"stat": "Ok", "stCode": 200, "data": []}


neo_module.TotpAPI = FakeTotp
neo_module.OrderAPI = FakeOrder
neo_module.OrderReportAPI = FakeOrders
neo = NeoAPI(
    consumer_key="synthetic",
    environment="prod",
    access_token=None,
    transport=httpx.MockTransport(lambda _request: denied()),
)
view = neo.totp_login(mobile_number="+910000000000", ucc="SYNTHETIC", totp="000000")
trade = neo.totp_validate(mpin="000000")
assert view["data"]["kType"] == "View"
assert trade["data"]["kType"] == "Trade"
neo.configuration.edit_token = "synthetic-trade-token"
neo.configuration.edit_sid = "synthetic-trade-sid"
neo.configuration.ucc = "SYNTHETIC"
neo.configuration.base_url = "https://example.invalid"
assert neo.order_report()["data"] == []
assert neo.place_order(
    exchange_segment="nse_cm",
    product="MIS",
    price="0",
    order_type="MKT",
    quantity="1",
    validity="DAY",
    trading_symbol="SYNTHETIC-EQ",
    transaction_type="B",
)["nOrdNo"] == "SYNTHETIC"

# Exercise the changed read orchestration against synthetic services. The
# process-wide network guard is already active and HOME belongs to the
# disposable contract workspace, including 3.0.8's on-disk holdings cache.
read_calls = {"holdings": 0, "positions": 0, "quotes": 0}
position_has_ltp = True
position_ltp = "120.00"


class FakePortfolio:
    failure = None

    def __init__(self, _api_client):
        pass

    def portfolio_holdings(self):
        if self.failure is not None:
            raise self.failure
        read_calls["holdings"] += 1
        return {"data": [{"exchangeIdentifier": "SYNTHETIC-TOKEN", "averagePrice": 100.0}]}


class FakePositions:
    failure = None

    def __init__(self, _api_client):
        pass

    def position_init(self):
        if self.failure is not None:
            raise self.failure
        read_calls["positions"] += 1
        position = {"exSeg": "nse_cm", "tok": "SYNTHETIC-TOKEN", "cfBuyQty": "2"}
        if position_has_ltp:
            position["ltp"] = position_ltp
        return {"data": [position]}


def synthetic_quotes(*, instrument_tokens, quote_type):
    assert instrument_tokens == [{"exchange_segment": "nse_cm", "instrument_token": "SYNTHETIC-TOKEN"}]
    assert quote_type == "ltp"
    read_calls["quotes"] += 1
    return [{"exchange": "nse_cm", "exchange_token": "SYNTHETIC-TOKEN", "ltp": "120.00"}]


neo_module.PortfolioAPI = FakePortfolio
neo_module.PositionsAPI = FakePositions
neo.quotes = synthetic_quotes
assert neo.holdings()["data"][0]["averagePrice"] == 100.0
position = neo.positions()["data"][0]
if md.version("kotakneoapi") == "3.0.9":
    # positions() returns the broker row. Enrichment (netQty/positionPnl) stays
    # unwired because a missing or zero LTP used to be treated as a price.
    raw = {"exSeg": "nse_cm", "tok": "SYNTHETIC-TOKEN", "cfBuyQty": "2", "ltp": "120.00"}
    assert position == raw
    assert "netQty" not in position
    assert "positionPnl" not in position
    assert read_calls == {"holdings": 1, "positions": 1, "quotes": 0}
    position_has_ltp = False
    missing_ltp = neo.positions()["data"][0]
    assert missing_ltp == {"exSeg": "nse_cm", "tok": "SYNTHETIC-TOKEN", "cfBuyQty": "2"}
    assert "netQty" not in missing_ltp
    assert "positionPnl" not in missing_ltp
    position_has_ltp = True
    position_ltp = 0
    zero_ltp = neo.positions()["data"][0]
    assert zero_ltp == {"exSeg": "nse_cm", "tok": "SYNTHETIC-TOKEN", "cfBuyQty": "2", "ltp": 0}
    assert "netQty" not in zero_ltp
    assert "positionPnl" not in zero_ltp
    assert read_calls == {"holdings": 1, "positions": 3, "quotes": 0}
elif md.version("kotakneoapi") == "3.0.8":
    assert position["netQty"] == 2.0
    assert position["averagePrice"] == 100.0
    assert position["positionPnl"] == 40.0
    assert position["pnlCalculationError"] is None
    assert read_calls == {"holdings": 1, "positions": 1, "quotes": 0}
    position_has_ltp = False
    assert neo.positions()["data"][0]["positionPnl"] == 40.0
    assert read_calls == {"holdings": 1, "positions": 2, "quotes": 1}
elif md.version("kotakneoapi") == "3.0.7":
    assert position == {"exSeg": "nse_cm", "tok": "SYNTHETIC-TOKEN", "cfBuyQty": "2", "ltp": "120.00"}
    assert read_calls == {"holdings": 1, "positions": 1, "quotes": 0}
else:
    raise AssertionError("unreviewed Kotak Neo read contract version")

token = WsToken("nse_cm", "11536")
assert hash(token)
assert all(inspect.isclass(model) for model in (SFeedScrip, SFeedScripLite, SFeedIndex, OrderUpdate, PositionUpdate))

facade = KotakNeoSdkSession.__new__(KotakNeoSdkSession)
facade._neo = neo
facade._closed = False


def require_canonical_read_failure(method):
    try:
        getattr(facade, method)()
    except BrokerInternal as error:
        assert type(error) is BrokerInternal
        assert "cache-private-detail" not in str(error)
        assert "AttributeError" not in str(error)
        assert "PermissionError" not in str(error)
        assert "has no attribute" not in str(error)
    else:
        raise AssertionError(f"failed {method} became a successful read")


# Both SDK tracks must keep errors inside the redacted gateway taxonomy.
for service, method, failure in (
    (FakePositions, "positions", AttributeError("cache-private-detail")),
    (FakePortfolio, "holdings", PermissionError("cache-private-detail")),
):
    service.failure = failure
    try:
        require_canonical_read_failure(method)
    finally:
        service.failure = None

if md.version("kotakneoapi") in {"3.0.8", "3.0.9"}:
    # 3.0.8 positions() reads this on-disk holdings cache. 3.0.9 positions()
    # does not; holdings() still writes it, so a failed write must stay a
    # redacted read failure rather than an empty holdings result.
    from neo_api_client.utils import holdings_cache

    cache_path = holdings_cache._cache_path(neo.configuration.ucc, date.today())
    assert cache_path.resolve().is_relative_to(Path.home().resolve())
    cache_path.write_text("[]", encoding="utf-8")
    neo._holdings_cache = None
    neo._holdings_cache_date = None
    if md.version("kotakneoapi") == "3.0.8":
        require_canonical_read_failure("positions")
    else:
        ignored_cache = neo.positions()["data"][0]
        assert "netQty" not in ignored_cache
        assert "positionPnl" not in ignored_cache

    # A successful synthetic portfolio response must not become empty/zero
    # holdings when the real SDK's cache write fails. Preserve every other
    # Path.write_text call and restore the method even if an assertion fails.
    original_write = Path.write_text

    def denied_cache_write(path, *args, **kwargs):
        if path == cache_path:
            raise PermissionError("cache-private-detail")
        return original_write(path, *args, **kwargs)

    Path.write_text = denied_cache_write
    try:
        require_canonical_read_failure("holdings")
    finally:
        Path.write_text = original_write

market_feed = facade.create_websocket(
    max_reconnect_attempts=0,
    max_connect_retries=0,
    max_subscriptions=3000,
)
order_feed = facade.create_order_feed(
    max_reconnect_attempts=0,
    max_connect_retries=0,
)
assert isinstance(market_feed, SFeedWebSocket)
assert market_feed.max_reconnect_attempts == 0
assert market_feed.max_connect_retries == 0
assert market_feed.max_subscriptions == 3000
assert isinstance(order_feed, OrderFeedWebSocket)
assert order_feed.max_reconnect_attempts == 0
assert order_feed.max_connect_retries == 0
for method in (
    "connect", "close", "subscribe_scrips", "unsubscribe_scrips", "subscribe_depth",
    "unsubscribe_depth", "subscribe_index", "unsubscribe_index",
):
    assert callable(getattr(market_feed, method))
assert callable(facade.logout_sdk)
assert callable(facade.close_rest)
for legacy in ("subscribe", "un_subscribe", "subscribe_to_orderfeed"):
    assert not hasattr(KotakNeoSdkSession, legacy), legacy


async def require_callback_cancellation_and_cleanup():
    original_decode = feed_module.decode_packet

    def failed_decode(*_args):
        raise ValueError("synthetic malformed frame")

    def failed_callback(*_args):
        raise RuntimeError("synthetic callback failure")

    def cancelled_callback(*_args):
        raise asyncio.CancelledError

    class SyntheticSocket:
        def __init__(self):
            self.close_calls = 0

        async def close(self):
            self.close_calls += 1

    async def blocked_receiver(started):
        started.set()
        await asyncio.Event().wait()

    feed_module.decode_packet = failed_decode
    try:
        market_feed.on_error = failed_callback
        if md.version("kotakneoapi") in {"3.0.8", "3.0.9"}:
            assert market_feed._decode_packet(b"synthetic") is None
        else:
            try:
                market_feed._decode_packet(b"synthetic")
            except RuntimeError:
                pass
            else:
                raise AssertionError("release callback behaviour changed")

        market_feed.on_error = cancelled_callback
        try:
            market_feed._decode_packet(b"synthetic")
        except asyncio.CancelledError:
            pass
        else:
            raise AssertionError("market callback swallowed cancellation")

        order_feed.on_disconnect = cancelled_callback
        try:
            await order_feed._handle_disconnect()
        except asyncio.CancelledError:
            pass
        else:
            raise AssertionError("order callback swallowed cancellation")
    finally:
        feed_module.decode_packet = original_decode
        market_feed.on_error = None
        order_feed.on_disconnect = None
        for feed in (market_feed, order_feed):
            socket = SyntheticSocket()
            started = asyncio.Event()
            receiver = asyncio.create_task(blocked_receiver(started))
            await started.wait()
            feed._ws = socket
            feed._receive_task = receiver
            await feed.close()
            assert receiver.cancelled()
            assert feed._receive_task is None
            assert feed._ws is None
            assert feed.is_connected is False
            assert socket.close_calls == 1


class NoIOSelector(selectors.BaseSelector):
    "Schedule ready synthetic tasks without sockets or real I/O."

    def __init__(self):
        self.turns = 0

    def register(self, *_args):
        raise AssertionError("real I/O is forbidden in the callback probe")

    def unregister(self, *_args):
        raise AssertionError("real I/O is forbidden in the callback probe")

    def get_map(self):
        return {}

    def select(self, timeout=None):
        self.turns += 1
        assert self.turns <= 256, "callback probe exceeded its scheduling bound"
        assert timeout == 0, "callback probe stalled without a ready task"
        return []


class NoSocketEventLoop(asyncio.SelectorEventLoop):
    def __init__(self):
        super().__init__(selector=NoIOSelector())

    def _make_self_pipe(self):
        # No threads, OS events or real sockets participate in this probe.
        # Keep the all-socket process audit guard unchanged, including asyncio's
        # usual wake-up socketpair. Base _write_to_self safely sees None.
        self._ssock = self._csock = None

    def _close_self_pipe(self):
        pass

    def _deny_io(self, *_args, **_kwargs):
        raise AssertionError("real I/O is forbidden in the callback probe")

    add_reader = add_writer = _deny_io
    sock_recv = sock_recv_into = sock_recvfrom = sock_recvfrom_into = _deny_io
    sock_sendall = sock_sendto = sock_connect = sock_accept = sock_sendfile = _deny_io


with asyncio.Runner(loop_factory=NoSocketEventLoop) as runner:
    runner.run(require_callback_cancellation_and_cleanup())

main_envelope = {"stat": "Ok", "stCode": 200, "data": [], "rateLimit": {"remaining": 2}}
release_envelope = {"data": {
    "common_data": {
        "mktLot": "65", "multiplier": "1", "unlSymbol": "NIFTY",
        "exSeg": "nse_fo", "expiryDt": "2026-09-24",
    },
    "call": [{"instrument": {"neoSymbol": "nse_fo|71472"}}],
    "put": [],
}}
assert validate_read_envelope(main_envelope, operation="order_report") is main_envelope
assert validate_read_envelope(release_envelope, operation="option_chain") is release_envelope

# Importing the runtime module above is part of the contract. Keep a concrete
# reference so optimisers and static rewrites cannot erase it as unused.
assert kotakneo_streaming.__name__.endswith("kotakneo_streaming")
neo.api_client.rest_client.close()

cwd = Path.cwd()
cwd_files = sorted(str(path.relative_to(cwd)) for path in cwd.rglob("*") if path.is_file())
print(json.dumps({
    "distributions": distributions,
    "distribution_counts": counts,
    "namespace_owners": sorted(owners),
    "module_path": str(Path(neo_api_client.__file__).resolve()),
    "environment_root": str(Path(os.environ["KOTAK_CONTRACT_ENV"]).resolve()),
    "cwd_files": cwd_files,
    "contract_ok": True,
    "read_contract_ok": True,
    "read_failure_contract_ok": True,
    "feed_cancellation_contract_ok": True,
}, sort_keys=True))
"""


def _run_checked(
    args: Sequence[str | Path],
    *,
    run: Run,
    cwd: Path,
    env: Mapping[str, str] | None = None,
) -> subprocess.CompletedProcess[str]:
    command = list(map(str, args))
    result = run(command, cwd=cwd, env=env, capture_output=True, text=True, check=False)
    if result.returncode != 0:
        detail = (result.stderr or result.stdout or "command failed without output").strip()
        raise ContractError(f"command failed ({' '.join(command[:4])}): {detail}")
    return result


def resolve_scanner_targets(repo: Path = REPO) -> list[Path]:
    """Resolve exactly the seven migration-scan inputs and require each file."""

    if len(SCANNER_TARGETS) != 7 or len(set(SCANNER_TARGETS)) != 7:
        raise ContractError("Kotak Neo migration scanner target inventory must contain exactly seven unique paths")
    targets = [repo / relative for relative in SCANNER_TARGETS]
    missing = [str(path) for path in targets if not path.is_file()]
    if missing:
        raise ContractError(f"Kotak Neo migration scanner target is missing: {', '.join(missing)}")
    return targets


def materialise_scanner(
    workspace: Path,
    *,
    config: ContractConfig = _DEFAULT_CONFIG,
    run: Run = subprocess.run,
    env: Mapping[str, str] | None = None,
) -> Path:
    """Read the official migration scanner from the exact runtime commit."""

    bare = workspace / "kotak-upstream.git"
    _run_checked(["git", "init", "--bare", bare], run=run, cwd=workspace, env=env)
    _run_checked(
        [
            "git",
            f"--git-dir={bare}",
            "fetch",
            "--depth=1",
            config.repo_url,
            config.runtime_main_commit,
        ],
        run=run,
        cwd=workspace,
        env=env,
    )
    shown = _run_checked(
        [
            "git",
            f"--git-dir={bare}",
            "show",
            f"{config.runtime_main_commit}:docs/scripts/migrate_from_v2.py",
        ],
        run=run,
        cwd=workspace,
        env=env,
    )
    scanner = workspace / "migrate_from_v2.py"
    scanner.write_text(shown.stdout, encoding="utf-8")
    return scanner


_SCANNER_SUMMARY = re.compile(
    r"(?m)^(?P<errors>\d+) error\(s\), (?P<warnings>\d+) warning\(s\) across (?P<files>\d+) file\(s\)\.$"
)


def validate_scanner_result(result: subprocess.CompletedProcess[str], *, expected_files: int) -> None:
    """Require a parsed zero-error/zero-warning summary for every target."""

    matches = list(_SCANNER_SUMMARY.finditer(result.stdout or ""))
    if len(matches) != 1:
        raise ContractError("official Kotak Neo migration scanner must emit exactly one auditable summary")
    match = matches[0]
    errors = int(match.group("errors"))
    warnings = int(match.group("warnings"))
    files = int(match.group("files"))
    if files != expected_files:
        raise ContractError(f"official Kotak Neo migration scanner covered {files}, expected {expected_files}")
    if errors or warnings:
        raise ContractError(f"official Kotak Neo migration scanner found {errors} error(s) and {warnings} warning(s)")
    if result.returncode != 0:
        raise ContractError(f"official Kotak Neo migration scanner exited {result.returncode}")


def _release_gateway_project(
    track: SdkTrack,
    workspace: Path,
    *,
    repo: Path,
    run: Run,
    env: Mapping[str, str],
) -> Path:
    """Copy tracked gateway source for release-baseline compatibility only.

    The production runtime dependency stays exact. This disposable manifest
    substitutes only that dependency with the exact historical release SDK;
    every source byte and other dependency remains the production checkout's.
    It is never installed into the repository environment or used for runtime
    attestation, and the surrounding temporary workspace owns its cleanup.
    """
    source = repo / "packages/integrations/gateway"
    text = (source / "pyproject.toml").read_text(encoding="utf-8")
    runtime_dependency = f'"kotakneoapi=={load_contract_config(repo).version}"'
    if text.count(runtime_dependency) != 1:
        raise ContractError("gateway must declare exactly one exact runtime SDK dependency")
    listed = _run_checked(
        [
            "git",
            "ls-files",
            "-z",
            "--",
            "packages/integrations/gateway/pyproject.toml",
            "packages/integrations/gateway/src/",
        ],
        run=run,
        cwd=repo,
        env=env,
    ).stdout
    files = [Path(path) for path in listed.split("\0") if path and (repo / path).is_file()]
    manifest = Path("packages/integrations/gateway/pyproject.toml")
    if manifest not in files or not any(path.is_relative_to("packages/integrations/gateway/src") for path in files):
        raise ContractError("tracked gateway source and manifest are required for release compatibility")
    target = workspace / "release-gateway"
    for relative in files:
        original = repo / relative
        if original.is_symlink() or not original.resolve().is_relative_to(source.resolve()):
            raise ContractError("release compatibility source must be a tracked gateway file")
        destination = target / relative.relative_to("packages/integrations/gateway")
        destination.parent.mkdir(parents=True, exist_ok=True)
        shutil.copyfile(original, destination)
    (target / "pyproject.toml").write_text(
        text.replace(runtime_dependency, f'"kotakneoapi=={track.version}"'), encoding="utf-8"
    )
    return target


def _probe_track(
    track: SdkTrack,
    environment: Path,
    workspace: Path,
    *,
    run: Run,
    repo: Path,
    base_env: Mapping[str, str],
) -> None:
    requirements = workspace / "base-requirements.txt"
    gateway_project = (
        _release_gateway_project(track, workspace, repo=repo, run=run, env=base_env) if track.release_tag else None
    )
    for command in build_track_commands(track, environment, requirements, repo=repo, gateway_project=gateway_project):
        _run_checked(command, run=run, cwd=workspace, env=base_env)

    probe_cwd = workspace / f"probe-{track.name}"
    probe_cwd.mkdir()
    probe_env = build_subprocess_environment(workspace / f"process-{track.name}", source=base_env)
    probe_env.update(
        {
            "KOTAK_CONTRACT_ENV": str(environment),
            "NEO_LOG_FILE_ENABLED": "false",
            "NEO_LOG_LEVEL": "CRITICAL",
            "NO_PROXY": "*",
            "no_proxy": "*",
        }
    )
    result = _run_checked(
        [environment_python(environment), "-I", "-c", _PROBE_SCRIPT],
        run=run,
        cwd=probe_cwd,
        env=probe_env,
    )
    validate_probe_result(parse_probe_output(result.stdout), track)


def run_contract(
    *,
    repo: Path = REPO,
    run: Run = subprocess.run,
    temporary_parent: Path | None = None,
) -> ContractConfig:
    """Run both disposable SDK tracks and the exact official migration scan."""

    config = load_contract_config(repo)
    tracks = build_tracks(config)
    parent = str(temporary_parent) if temporary_parent is not None else None
    with tempfile.TemporaryDirectory(prefix="kotakneo-contract-", dir=parent) as temporary:
        workspace = Path(temporary)
        base_env = build_subprocess_environment(workspace / "process")
        requirements = workspace / "base-requirements.txt"
        _run_checked(
            build_export_command(requirements, repo=repo),
            run=run,
            cwd=repo,
            env=base_env,
        )

        for track in tracks:
            _probe_track(
                track,
                workspace / track.name,
                workspace,
                run=run,
                repo=repo,
                base_env=base_env,
            )

        scanner = materialise_scanner(workspace, config=config, run=run, env=base_env)
        targets = resolve_scanner_targets(repo)
        result = run(
            [str(host_python(repo)), str(scanner), *map(str, targets)],
            cwd=repo,
            env=base_env,
            capture_output=True,
            text=True,
            check=False,
        )
        validate_scanner_result(result, expected_files=len(targets))
    return config


def main(argv: list[str] | None = None) -> int:
    """CLI entry point."""

    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--repo", type=Path, default=REPO, help="FlintTrade checkout to verify")
    args = parser.parse_args(argv)
    try:
        repo = args.repo.resolve()
        config = run_contract(repo=repo)
    except ContractError as exc:
        print(f"Kotak Neo SDK contract: FAIL: {exc}", file=sys.stderr)
        return 1
    print(
        "Kotak Neo SDK contract: PASS "
        f"(runtime main {config.runtime_main_commit}; "
        f"{config.release_tag} {config.release_commit}; migration scan clean)"
    )
    return 0


if __name__ == "__main__":
    raise SystemExit(main())