"""Synthetic harness contracts and bounded Docker workload ownership experiment.

Python 3.10+, standard library only. --contracts-only needs no Docker.
No model, real provider, credentials, host mounts, or published network ports.
This is a worked experiment, not a production supervisor or gateway.
"""

import argparse
import json
import platform
import subprocess
import threading
import time
import uuid
from concurrent.futures import ThreadPoolExecutor


def check(condition, message):
    if not condition:
        raise AssertionError(message)


def row(case_id, title, evidence, **observed):
    return {"id": case_id, "title": title, "evidence": evidence, "result": "PASS", **observed}


def valid_lease(lease, now, workload):
    if not isinstance(lease, dict) or not all(
        type(lease.get(key)) is int for key in ("approved_at", "prior_expiry", "expires_at")
    ):
        return False
    if not isinstance(workload, dict) or not all(
        lease.get(key) == workload.get(key) and workload.get(key) is not None
        for key in ("owner", "workload_id", "generation")
    ):
        return False
    return (
        lease.get("authorized") is True
        and lease["approved_at"] <= now
        and lease["approved_at"] < lease["prior_expiry"]
        and now < lease["expires_at"]
        and lease["expires_at"] > lease["prior_expiry"]
    )


def normalize(adapter, payload):
    """Two invented fixture shapes; neither represents a provider API."""
    if type(payload) is not dict:
        raise ValueError("Unsupported adapter payload")
    if adapter == "fixture-a" and set(payload) == {"status", "receipt"}:
        status, receipt = payload["status"], payload["receipt"]
    elif adapter == "fixture-b" and set(payload) == {"result"}:
        value = payload["result"]
        if not isinstance(value, dict) or set(value) != {"outcome", "evidence"}:
            raise ValueError("Unsupported adapter schema")
        status, receipt = value["outcome"], value["evidence"]
    else:
        raise ValueError("Unsupported adapter schema")
    if type(status) is not str or status not in {
        "completed",
        "no_op",
        "blocked",
        "failed",
        "cancelled",
        "unknown",
    }:
        raise ValueError("Unsupported outcome")
    if type(receipt) is not dict:
        raise ValueError("Missing receipt")
    return status, receipt


def decide(outcome, receipt, now=20):
    """Evidence is supplied by this trusted fixture, never inferred from stdout."""
    if not isinstance(receipt, dict) or receipt.get("capture_complete") is not True:
        return "reconcile"
    effect = receipt.get("effect")
    if effect not in {"none", "confirmed", "partial"}:
        return "reconcile"
    if not isinstance(receipt.get("workloads"), list):
        return "reconcile"
    if outcome in {"blocked", "no_op"} and effect != "none":
        return "reconcile"
    if outcome == "completed" and effect == "partial":
        return "reconcile"
    if outcome == "completed" and receipt.get("task_verified") is not True:
        return "reconcile"
    if outcome == "no_op" and (
        receipt.get("task_verified") is not True or receipt.get("no_op_permitted") is not True
    ):
        return "reconcile"
    workloads = receipt["workloads"]
    if workloads:
        if outcome == "completed" and all(
            isinstance(item, dict) and valid_lease(item.get("lease"), now, item.get("identity"))
            for item in workloads
        ):
            return "handoff"
        return "reconcile"
    if outcome in {"completed", "no_op"}:
        return "advance"
    if outcome == "blocked":
        return "stop_policy"
    if outcome == "cancelled" and effect == "none":
        return "stop_cancelled"
    return "reconcile"


def retry_fixture(outcomes, max_attempts=3):
    """Retry permission requires explicit transient classification and no effect."""
    for count, (kind, effect) in enumerate(outcomes[:max_attempts], 1):
        if effect != "none" or kind == "unknown":
            return count, "reconcile"
        if kind == "terminal":
            return count, "stop_policy"
        if kind != "transient":
            return count, "stop_unknown_classification"
    return min(len(outcomes), max_attempts), "exhausted"


class Budget:
    """Local mock ledger. Integer units are artificial, not money or tokens."""

    def __init__(self, limit):
        self.limit = limit
        self.spent = 0
        self.pending = {}
        self.settled = set()
        self.available = True
        self.lock = threading.Lock()

    def reserve(self, attempt, bound):
        with self.lock:
            if not self.available or type(bound) is not int or bound <= 0:
                return False
            if attempt in self.pending or attempt in self.settled:
                return False
            if self.spent + sum(self.pending.values()) + bound > self.limit:
                return False
            self.pending[attempt] = bound
            return True

    def settle(self, attempt, charge):
        with self.lock:
            if attempt not in self.pending or type(charge) is not int:
                raise ValueError("Unknown reservation or charge")
            if not 0 <= charge <= self.pending[attempt]:
                raise ValueError("Charge exceeds declared bound")
            self.spent += charge
            del self.pending[attempt]
            self.settled.add(attempt)


def contracts():
    rows = []
    base = {"capture_complete": True, "effect": "confirmed", "task_verified": True, "workloads": []}
    identity = {"owner": "session-a", "workload_id": "fixture-workload-a", "generation": 1}
    lease = {**identity, "authorized": True, "approved_at": 9, "prior_expiry": 10, "expires_at": 30}

    def retained(value):
        return {**base, "workloads": [{"identity": identity, "lease": value}]}

    fixtures = [
        ("completed task", "completed", base, "advance"),
        (
            "verified read-only no-op",
            "no_op",
            {**base, "effect": "none", "no_op_permitted": True},
            "advance",
        ),
        ("failure after effect", "failed", {**base, "effect": "partial"}, "reconcile"),
        ("false success claim", "completed", {**base, "task_verified": False}, "reconcile"),
        ("missing receipt", "completed", None, "reconcile"),
        ("blocked without effect", "blocked", {**base, "effect": "none"}, "stop_policy"),
        ("cancelled without effect", "cancelled", {**base, "effect": "none"}, "stop_cancelled"),
        ("authorized retained service", "completed", retained(lease), "handoff"),
        ("blocked claim with effect", "blocked", base, "reconcile"),
        ("capture gap", "completed", {**base, "capture_complete": False}, "reconcile"),
        ("unknown cancellation effect", "cancelled", {**base, "effect": "unknown"}, "reconcile"),
        (
            "late retention approval",
            "completed",
            retained({**lease, "approved_at": 11}),
            "reconcile",
        ),
        (
            "completed claim with partial effect",
            "completed",
            {**base, "effect": "partial"},
            "reconcile",
        ),
        ("malformed retention receipt", "completed", {**base, "workloads": [{}]}, "reconcile"),
        ("task does not permit no-op", "no_op", {**base, "effect": "none"}, "reconcile"),
        ("stale lease generation", "completed", retained({**lease, "generation": 0}), "reconcile"),
        ("wrong lease owner", "completed", retained({**lease, "owner": "session-b"}), "reconcile"),
        (
            "wrong lease workload",
            "completed",
            retained({**lease, "workload_id": "fixture-workload-b"}),
            "reconcile",
        ),
        (
            "future lease approval",
            "completed",
            retained({**lease, "approved_at": 21, "prior_expiry": 25}),
            "reconcile",
        ),
        ("expired lease", "completed", retained({**lease, "expires_at": 19}), "reconcile"),
    ]
    for index, (title, outcome, receipt, expected) in enumerate(fixtures, 1):
        actual = decide(outcome, receipt)
        check(actual == expected, title)
        rows.append(row(f"RC-{index:02}", title, "synthetic decision fixture", decision=actual))
    a = normalize("fixture-a", {"status": "completed", "receipt": base})
    b = normalize("fixture-b", {"result": {"outcome": "completed", "evidence": base}})
    check(a == b and decide(*a) == "advance", "Adapter contract mismatch")
    rows.append(
        row("AD-01", "two adapter shapes", "synthetic adapter fixtures", decision="advance")
    )
    rejected = 0
    for adapter, payload in [
        ("unknown", {}),
        ("fixture-b", {"result": []}),
        ("fixture-a", {"status": "completed", "receipt": None}),
        ("fixture-a", None),
        ("fixture-a", []),
        ("fixture-a", {"status": [], "receipt": base}),
    ]:
        try:
            normalize(adapter, payload)
        except ValueError:
            rejected += 1
    check(rejected == 6, "Malformed schemas accepted")
    rows.append(
        row(
            "AD-02",
            "unsupported and malformed schemas",
            "synthetic adapter fixtures",
            rejected=rejected,
        )
    )
    for case_id, title, outcomes, expected in [
        ("RT-01", "terminal refusal", [("terminal", "none")] * 5, (1, "stop_policy")),
        ("RT-02", "bounded transient retries", [("transient", "none")] * 5, (3, "exhausted")),
        ("RT-03", "unknown outcome", [("unknown", "unknown")] * 5, (1, "reconcile")),
        (
            "RT-04",
            "transient label after effect",
            [("transient", "confirmed")] * 5,
            (1, "reconcile"),
        ),
    ]:
        actual = retry_fixture(outcomes)
        check(actual == expected, title)
        rows.append(
            row(case_id, title, "synthetic retry fixtures", attempts=actual[0], decision=actual[1])
        )

    ledger = Budget(10)
    with ThreadPoolExecutor(max_workers=8) as pool:
        granted = list(pool.map(lambda i: ledger.reserve(f"concurrent-{i}", 3), range(8)))
    check(sum(granted) == 3 and sum(ledger.pending.values()) == 9, "Atomic cap violated")
    rows.append(
        row(
            "BU-01",
            "concurrent reservation",
            "local mock ledger",
            admitted=3,
            denied=5,
            reserved=9,
            limit=10,
        )
    )
    ledger = Budget(10)
    check(ledger.reserve("original", 6), "Initial reserve denied")
    check(
        not ledger.reserve("retry", 6) and not ledger.reserve("fallback", 5),
        "Retry bypassed reservation",
    )
    rows.append(
        row("BU-02", "retry and fallback share cap", "local mock ledger", reserved=6, denied=2)
    )
    check(not ledger.reserve("original", 6), "Duplicate attempt reserved")
    check(ledger.pending == {"original": 6}, "Uncertain charge released")
    rows.append(
        row(
            "BU-03",
            "lost acknowledgement and cancellation",
            "simulated uncertainty; local mock ledger",
            reserved=6,
            charge="UNKNOWN",
            cancellation="UNKNOWN",
        )
    )
    ledger.settle("original", 4)
    check(
        ledger.spent == 4 and not ledger.pending and ledger.reserve("later", 6), "Settlement failed"
    )
    try:
        ledger.settle("original", 4)
    except ValueError:
        pass
    else:
        raise AssertionError("Duplicate settlement accepted")
    rows.append(row("BU-04", "reconcile exactly once", "local mock ledger", spent=4, reserved=6))
    admission = Budget(10)
    admission.available = False
    dispatches = int(admission.reserve("outage", 1))
    admission.available = True
    dispatches += int(admission.reserve("unpriced", None))
    check(dispatches == 0 and not admission.pending, "Unavailable or unpriced route admitted")
    check(admission.reserve("priced-control", 1), "Spare-capacity control failed")
    rows.append(
        row(
            "BU-05",
            "accounting outage and unpriced route",
            "local mock ledger",
            invalid_dispatches=dispatches,
            priced_control_admitted=True,
        )
    )
    try:
        ledger.settle("later", 7)
    except ValueError:
        pass
    else:
        raise AssertionError("Broken price bound accepted")
    check(ledger.pending["later"] == 6, "Over-bound evidence discarded reservation")
    rows.append(
        row(
            "BU-06",
            "charge exceeds declared bound",
            "local mock ledger",
            decision="contract_violation",
            reservation_retained=6,
        )
    )
    return rows


SERVER = r"""
import http.server, json, signal, subprocess, sys
from pathlib import Path
signal.signal(signal.SIGALRM, lambda *_: sys.exit(124))
signal.alarm(90)
marker = json.loads(sys.argv[1])
child = subprocess.Popen([sys.executable, '-c', 'import time; time.sleep(120)', 'owned-child'], start_new_session=True)
Path('/tmp/child.pid').write_text(str(child.pid))
class Handler(http.server.BaseHTTPRequestHandler):
    def do_GET(self):
        self.send_response(200)
        self.end_headers()
        self.wfile.write(json.dumps(marker).encode())
    def log_message(self, *args): pass
http.server.HTTPServer(('127.0.0.1', 3187), Handler).serve_forever()
"""


def docker(*args, timeout=20):
    return subprocess.run(
        ["docker", *args], capture_output=True, text=True, timeout=timeout, check=True
    ).stdout.strip()


def lifecycle(image):
    image_info = json.loads(docker("image", "inspect", image))[0]
    image_id = image_info["Id"]
    run_id = uuid.uuid4().hex
    owned = []
    rows = []

    def inventory():
        return set(docker("ps", "-a", "--no-trunc", "--format", "{{.ID}}").splitlines())

    def marker(container):
        return json.loads(
            docker(
                "exec",
                container,
                "python",
                "-c",
                "import urllib.request; print(urllib.request.urlopen('http://127.0.0.1:3187',timeout=2).read().decode())",
            )
        )

    def remove(container):
        info = json.loads(docker("inspect", container))[0]
        check(
            info["Id"] == container and info["Config"]["Labels"].get("ra.operations.run") == run_id,
            "Ownership mismatch",
        )
        docker("rm", "--force", container)
        check(container not in inventory(), "Cleanup unverified")

    try:
        expected = []
        for session in ("session-a", "session-b"):
            identity = {"run": run_id, "session": session, "generation": 1, "build": "synthetic-v1"}
            expected.append(identity)
            container = docker(
                "run",
                "--detach",
                "--rm",
                "--pull=never",
                "--network=none",
                "--read-only",
                "--cap-drop=ALL",
                "--security-opt=no-new-privileges",
                "--user=65534:65534",
                "--pids-limit=32",
                "--memory=128m",
                "--cpus=0.5",
                "--tmpfs",
                "/tmp:rw,noexec,nosuid,size=1m,mode=1777",
                "--label",
                f"ra.operations.run={run_id}",
                image_id,
                "python",
                "-c",
                SERVER,
                json.dumps(identity),
            )
            check(len(container) == 64, "Expected immutable container ID")
            owned.append(container)
            deadline = time.monotonic() + 10
            while True:
                try:
                    actual = marker(container)
                    break
                except subprocess.CalledProcessError:
                    if time.monotonic() >= deadline:
                        raise
                    time.sleep(0.1)
            check(actual == identity, "Wrong service marker")
        a, b = owned
        rows.append(
            row(
                "LC-01",
                "two session endpoints",
                "real Docker / internal HTTP",
                distinct_markers=True,
                host_ports=0,
            )
        )
        busy = docker(
            "exec",
            a,
            "python",
            "-c",
            "import errno,socket; s=socket.socket();\ntry: s.bind(('127.0.0.1',3187))\nexcept OSError as e: assert e.errno==errno.EADDRINUSE; print('busy')\nelse: raise AssertionError('port not busy')",
        )
        check(
            busy == "busy" and marker(a) == expected[0] and marker(b) == expected[1],
            "Busy-port handling changed owner",
        )
        rows.append(
            row(
                "LC-02",
                "occupied port preserves listener",
                "real Docker / socket bind",
                conflict="EADDRINUSE",
                listeners_preserved=2,
            )
        )
        detached = docker(
            "exec",
            a,
            "python",
            "-c",
            "import os; from pathlib import Path; p=int(Path('/tmp/child.pid').read_text()); assert b'owned-child' in Path(f'/proc/{p}/cmdline').read_bytes(); assert os.getsid(p)==p; print('detached child observed')",
        )
        check(detached == "detached child observed", "Detached child not observed")
        rows.append(
            row(
                "LC-03",
                "detached descendant in owning container",
                "real Docker / proc observation",
                detached=True,
            )
        )
        stale = {**expected[0], "generation": 0}
        allowed = stale == marker(a)
        check(not allowed and a in inventory(), "Stale generation authorized cleanup")
        rows.append(
            row(
                "LC-04",
                "stale generation denied",
                "controller equality check over real endpoint",
                cleanup_authorized=False,
            )
        )
        workload = {"owner": "session-a", "workload_id": a, "generation": expected[0]["generation"]}
        lease = {
            **workload,
            "authorized": True,
            "approved_at": 9,
            "prior_expiry": 10,
            "expires_at": 30,
        }
        check(
            valid_lease(lease, 20, workload)
            and not valid_lease({**lease, "approved_at": 11}, 20, workload),
            "Lease admission failed",
        )
        check(marker(a) == expected[0], "Retained workload lost")
        rows.append(
            row(
                "LC-05",
                "prior authorization for retention",
                "simulated clock/lease; real retained container",
                current_lease_valid=True,
                late_approval_denied=True,
            )
        )
        # Simulate an unavailable cleanup acknowledgement. Do not pretend a failed
        # observation established absence; actual teardown still happens below.
        cleanup_ack = None
        cleanup_state = "UNKNOWN" if cleanup_ack is None else "STOPPED"
        check(cleanup_state == "UNKNOWN" and a in inventory(), "Cleanup uncertainty lost")
        rows.append(
            row(
                "LC-06",
                "missing cleanup acknowledgement",
                "simulated acknowledgement loss; real live container",
                cleanup="UNKNOWN",
            )
        )
        remove(a)
        check(marker(b) == expected[1], "Cross-session cleanup")
        rows.append(
            row(
                "LC-07",
                "cleanup preserves other session",
                "real Docker removal and HTTP receipt",
                first_absent=True,
                second_alive=True,
            )
        )
        remove(b)
        check(not (set(owned) & inventory()), "Owned workload remains")
        rows.append(
            row("LC-08", "all owned workloads removed", "real Docker inventory", remaining=0)
        )
    finally:
        # Recover a lost create acknowledgement only through this random run label.
        # Every recovered immutable ID is label-checked again before removal.
        errors = []
        try:
            remaining = set(
                docker(
                    "ps",
                    "-a",
                    "--no-trunc",
                    "--filter",
                    f"label=ra.operations.run={run_id}",
                    "--format",
                    "{{.ID}}",
                ).splitlines()
            )
        except (subprocess.SubprocessError, OSError) as error:
            errors.append(str(error))
            remaining = set(owned)
        for container in remaining:
            try:
                remove(container)
            except (subprocess.SubprocessError, OSError, AssertionError) as error:
                errors.append(str(error))
        if errors:
            raise RuntimeError(f"Cleanup UNKNOWN for run {run_id}; known IDs {owned}: {errors}")
    return rows, {
        "image_id": image_id,
        "image_digests": image_info.get("RepoDigests", []),
        "run_id": run_id,
    }


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--contracts-only", action="store_true")
    parser.add_argument(
        "--image",
        default="python:3.12-slim",
        help="Already-present local Docker image; never pulled",
    )
    args = parser.parse_args()
    rows = contracts()
    environment = {"python": platform.python_version(), "kernel": platform.release()}
    if not args.contracts_only:
        observed, docker_environment = lifecycle(args.image)
        rows.extend(observed)
        environment.update(docker_environment)
    print(
        json.dumps(
            {
                "kind": "synthetic harness contracts; optional actual Docker lifecycle",
                "mode": "contracts-only" if args.contracts_only else "full",
                "environment": environment,
                "cases": rows,
                "passed": len(rows),
            },
            indent=2,
        )
    )


if __name__ == "__main__":
    main()