#!/usr/bin/env python3
"""Closed JSON-to-argv bridge for Ahel-owned OpenClaw cron tasks."""

from __future__ import annotations

import json
import re
import subprocess
import sys
import threading
import time
from typing import Any, BinaryIO, Callable


OPENCLAW = "/usr/bin/openclaw"
MAX_BYTES = 64 * 1024
MAX_PROMPT_SCALARS = 12_000
MAX_CANCEL_TARGETS = 32
CLI_TIMEOUT_SECONDS = 35.0
TERMINATE_GRACE_SECONDS = 2.0
ENV = {"HOME": "/home/admin", "PATH": "/usr/bin:/bin"}
WEBHOOK = "https://192.168.8.168:8443/v1/openclaw/callbacks/run"
TASKS_LIST_ARGV = [
    OPENCLAW,
    "tasks",
    "list",
    "--json",
    "--runtime",
    "cron",
    "--status",
    "running",
]

SAFE_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$")
TASK_ID = re.compile(r"^task-[A-Za-z0-9._:-]{1,123}$")
ROUTES = {
    "ahel-local": "ahel-pool/qwen3.6-35b-hauhau-aggressive-mtp-q4",
    "ahel-cloud": "ahel-claude/claude-code-sonnet",
    "ahel-deep": "ahel-claude/claude-code-opus",
}
# Routines are recurring declarations. They keep every restriction `create` has
# except two: the schedule is a real (closed-grammar) expression instead of the
# one-shot "+1s", and the model comes from a prefix allowlist instead of ROUTES,
# so a routine can be pinned to a specific local, OpenRouter or subscription model.
ROUTINE_ID = re.compile(r"^routine-[A-Za-z0-9._:-]{1,120}$")
MODEL_PREFIXES = ("ahel-pool/", "openrouter/", "claude-code/")
SCHEDULE_PATTERNS = (
    re.compile(r"^every day (?:[01]\d|2[0-3]):[0-5]\d$"),
    re.compile(
        r"^every (?:monday|tuesday|wednesday|thursday|friday|saturday|sunday) "
        r"(?:[01]\d|2[0-3]):[0-5]\d$"
    ),
    re.compile(r"^every (?:[1-9]|1\d|2[0-4]) hours$"),
    re.compile(r"^every (?:1[5-9]|[2-5]\d) minutes$"),
)
ALLOWED_KEYS = {
    "status": frozenset(("version", "operation")),
    "list": frozenset(("version", "operation")),
    "get": frozenset(("version", "operation", "jobId")),
    "runResult": frozenset(("version", "operation", "jobId")),
    "cancel": frozenset(("version", "operation", "jobId")),
    "cancelRoutine": frozenset(("version", "operation", "jobId")),
    "cancelDeclaration": frozenset(("version", "operation", "taskId")),
    "declareRoutine": frozenset(
        (
            "version",
            "operation",
            "routineId",
            "name",
            "at",
            "prompt",
            "agent",
            "model",
            "fallbacks",
            "webhook",
            "timeoutSeconds",
        )
    ),
    "create": frozenset(
        (
            "version",
            "operation",
            "taskId",
            "name",
            "at",
            "prompt",
            "agent",
            "model",
            "fallbacks",
            "webhook",
            "timeoutSeconds",
        )
    ),
}


class WrapperError(Exception):
    """A content-free failure safe to return across the adapter boundary."""

    def __init__(self, code: str):
        self.code = code
        super().__init__(code)


def _is_int(value: Any) -> bool:
    return isinstance(value, int) and not isinstance(value, bool)


def _safe_id(value: Any, *, task: bool = False) -> str:
    pattern = TASK_ID if task else SAFE_ID
    if not isinstance(value, str) or pattern.fullmatch(value) is None:
        raise ValueError("invalid identifier")
    return value


def _validated_operation(payload: Any) -> str:
    if not isinstance(payload, dict):
        raise ValueError("request must be an object")
    operation = payload.get("operation")
    if not isinstance(operation, str) or operation not in ALLOWED_KEYS:
        raise ValueError("invalid operation")
    if set(payload) != ALLOWED_KEYS[operation]:
        raise ValueError("invalid request keys")
    if not _is_int(payload["version"]) or payload["version"] != 1:
        raise ValueError("invalid version")
    return operation


WEEKDAYS = {
    "sunday": 0, "monday": 1, "tuesday": 2, "wednesday": 3,
    "thursday": 4, "friday": 5, "saturday": 6,
}
SCHEDULE_TZ = "Europe/Tallinn"


def _validated_schedule(value: Any) -> str:
    """Translate the closed human grammar into a 5-field cron expression.

    OpenClaw's --at is the ONE-SHOT flag ("run once at time"); recurrence is
    --cron or --every. The coordinator keeps speaking the readable grammar and
    the wrapper — which owns CLI specifics — does the translation.
    """
    if not isinstance(value, str):
        raise ValueError("invalid schedule")
    match = re.fullmatch(r"every day ([01]\d|2[0-3]):([0-5]\d)", value)
    if match:
        return f"{int(match.group(2))} {int(match.group(1))} * * *"
    match = re.fullmatch(
        r"every (" + "|".join(WEEKDAYS) + r") ([01]\d|2[0-3]):([0-5]\d)", value)
    if match:
        return (f"{int(match.group(3))} {int(match.group(2))} * * "
                f"{WEEKDAYS[match.group(1)]}")
    match = re.fullmatch(r"every ([1-9]|1\d|2[0-4]) hours", value)
    if match:
        return f"0 */{int(match.group(1))} * * *"
    match = re.fullmatch(r"every (1[5-9]|[2-5]\d) minutes", value)
    if match:
        return f"*/{int(match.group(1))} * * * *"
    raise ValueError("invalid schedule")


def _validated_model(value: Any) -> str:
    if not isinstance(value, str) or not value.startswith(MODEL_PREFIXES):
        raise ValueError("invalid model")
    if len(value) > 128 or ".." in value or any(c in value for c in " \t\n\r"):
        raise ValueError("invalid model")
    if not any(len(value) > len(prefix) for prefix in MODEL_PREFIXES if value.startswith(prefix)):
        raise ValueError("invalid model")
    return value


def _validate_prompt(value: Any) -> str:
    if not isinstance(value, str) or not value or len(value) > MAX_PROMPT_SCALARS:
        raise ValueError("invalid prompt")
    if any(0xD800 <= ord(char) <= 0xDFFF for char in value):
        raise ValueError("invalid prompt")
    return value


def build_argv(payload: dict[str, Any]) -> list[str]:
    operation = _validated_operation(payload)
    if operation == "status":
        return [OPENCLAW, "cron", "status", "--json"]
    if operation == "list":
        return [OPENCLAW, "cron", "list", "--all", "--json"]
    if operation == "get":
        return [OPENCLAW, "cron", "get", _safe_id(payload["jobId"])]
    if operation == "runResult":
        _safe_id(payload["jobId"])
        return [OPENCLAW, "tasks", "list", "--json", "--runtime", "cron"]
    if operation == "cancel" or operation == "cancelRoutine":
        return [OPENCLAW, "cron", "rm", _safe_id(payload["jobId"]), "--json"]
    if operation == "cancelDeclaration":
        _safe_id(payload["taskId"], task=True)
        return [OPENCLAW, "cron", "list", "--all", "--json"]

    if operation == "declareRoutine":
        routine_id = payload["routineId"]
        if not isinstance(routine_id, str) or ROUTINE_ID.fullmatch(routine_id) is None:
            raise ValueError("invalid identifier")
        if payload["name"] != f"Ahel routine {routine_id}":
            raise ValueError("invalid routine declaration")
        cron_expression = _validated_schedule(payload["at"])
        routine_prompt = _validate_prompt(payload["prompt"])
        routine_model = _validated_model(payload["model"])
        if payload["agent"] not in ROUTES:
            raise ValueError("invalid route")
        if not isinstance(payload["fallbacks"], list) or payload["fallbacks"]:
            raise ValueError("fallbacks must be empty")
        if payload["webhook"] != WEBHOOK:
            raise ValueError("invalid webhook")
        if not _is_int(payload["timeoutSeconds"]) or payload["timeoutSeconds"] != 1800:
            raise ValueError("invalid timeout")
        return [
            OPENCLAW,
            "cron",
            "add",
            "--name",
            payload["name"],
            "--cron",
            cron_expression,
            "--tz",
            SCHEDULE_TZ,
            "--message",
            routine_prompt,
            "--agent",
            payload["agent"],
            "--model",
            routine_model,
            "--session",
            "isolated",
            "--declaration-key",
            routine_id,
            "--webhook",
            WEBHOOK,
            "--timeout-seconds",
            "1800",
            "--json",
        ]

    task_id = _safe_id(payload["taskId"], task=True)
    if payload["name"] != f"Ahel task {task_id}" or payload["at"] != "+1s":
        raise ValueError("invalid one-shot declaration")
    prompt = _validate_prompt(payload["prompt"])
    agent = payload["agent"]
    model = payload["model"]
    if not isinstance(agent, str) or ROUTES.get(agent) != model:
        raise ValueError("invalid route")
    if not isinstance(payload["fallbacks"], list) or payload["fallbacks"]:
        raise ValueError("fallbacks must be empty")
    if payload["webhook"] != WEBHOOK:
        raise ValueError("invalid webhook")
    if not _is_int(payload["timeoutSeconds"]) or payload["timeoutSeconds"] != 1800:
        raise ValueError("invalid timeout")
    return [
        OPENCLAW,
        "cron",
        "add",
        "--name",
        payload["name"],
        "--at",
        "+1s",
        "--message",
        prompt,
        "--agent",
        agent,
        "--model",
        model,
        "--session",
        "isolated",
        "--declaration-key",
        task_id,
        "--delete-after-run",
        "--webhook",
        WEBHOOK,
        "--timeout-seconds",
        "1800",
        "--json",
    ]


def _read_stream(
    stream: BinaryIO,
    *,
    retain: bool,
    state: dict[str, Any],
    lock: threading.Lock,
) -> None:
    try:
        while True:
            if state["overflow_event"].is_set():
                return
            chunk = stream.read(8192)
            if not chunk:
                return
            if not isinstance(chunk, bytes):
                chunk = bytes(chunk)
            with lock:
                previous_total = state["total"]
                next_total = previous_total + len(chunk)
                retained_budget = max(0, MAX_BYTES - previous_total)
                state["total"] = min(MAX_BYTES + 1, next_total)
                if retain and retained_budget:
                    state["stdout"].extend(chunk[:retained_budget])
                if next_total > MAX_BYTES:
                    state["overflow"] = True
                    state["overflow_event"].set()
                    return
    except (OSError, ValueError):
        with lock:
            state["read_error"] = True


def _stop_process(process: Any) -> None:
    try:
        process.terminate()
    except OSError:
        pass
    try:
        process.wait(timeout=TERMINATE_GRACE_SECONDS)
        return
    except subprocess.TimeoutExpired:
        pass
    try:
        process.kill()
    except OSError:
        pass
    try:
        process.wait(timeout=TERMINATE_GRACE_SECONDS)
    except subprocess.TimeoutExpired:
        pass


def _bounded_process(
    argv: list[str],
    *,
    popen_factory: Callable[..., Any],
    timeout: float,
    retain_stdout: bool,
) -> bytes:
    try:
        process = popen_factory(
            argv,
            stdin=subprocess.DEVNULL,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=False,
            shell=False,
            env=ENV,
        )
    except (OSError, ValueError):
        raise WrapperError("cli_failed") from None

    state: dict[str, Any] = {
        "total": 0,
        "overflow": False,
        "overflow_event": threading.Event(),
        "read_error": False,
        "stdout": bytearray(),
    }
    lock = threading.Lock()
    readers = [
        threading.Thread(
            target=_read_stream,
            kwargs={
                "stream": process.stdout,
                "retain": retain_stdout,
                "state": state,
                "lock": lock,
            },
            daemon=True,
        ),
        threading.Thread(
            target=_read_stream,
            kwargs={
                "stream": process.stderr,
                "retain": False,
                "state": state,
                "lock": lock,
            },
            daemon=True,
        ),
    ]
    for reader in readers:
        reader.start()

    stop_reason: str | None = None
    deadline = time.monotonic() + timeout
    try:
        while True:
            if state["overflow_event"].is_set():
                stop_reason = "cli_output_too_large"
                _stop_process(process)
                break
            remaining = deadline - time.monotonic()
            if remaining <= 0:
                stop_reason = "cli_timeout"
                _stop_process(process)
                break
            try:
                process.wait(timeout=min(0.05, remaining))
                break
            except subprocess.TimeoutExpired:
                continue
    finally:
        for reader in readers:
            reader.join(timeout=TERMINATE_GRACE_SECONDS)
        for stream in (process.stdout, process.stderr):
            try:
                stream.close()
            except (AttributeError, OSError):
                pass
        for reader in readers:
            reader.join(timeout=TERMINATE_GRACE_SECONDS)

    if stop_reason is not None:
        raise WrapperError(stop_reason)
    if any(reader.is_alive() for reader in readers) or state["read_error"]:
        raise WrapperError("cli_failed")
    if state["overflow"]:
        raise WrapperError("cli_output_too_large")
    if process.returncode != 0:
        raise WrapperError("cli_failed")
    return bytes(state["stdout"])


def _reject_duplicate_keys(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
    result: dict[str, Any] = {}
    for key, value in pairs:
        if key in result:
            raise ValueError("duplicate key")
        result[key] = value
    return result


def _parse_json(raw: bytes) -> Any:
    try:
        return json.loads(raw.decode("utf-8"), object_pairs_hook=_reject_duplicate_keys)
    except (UnicodeDecodeError, json.JSONDecodeError, ValueError):
        raise WrapperError("invalid_cli_response") from None


def _run_json(
    argv: list[str], popen_factory: Callable[..., Any], timeout: float
) -> Any:
    return _parse_json(
        _bounded_process(
            argv, popen_factory=popen_factory, timeout=timeout, retain_stdout=True
        )
    )


def _run_discard(
    argv: list[str], popen_factory: Callable[..., Any], timeout: float
) -> None:
    _bounded_process(
        argv, popen_factory=popen_factory, timeout=timeout, retain_stdout=False
    )


def _parse_create(value: Any) -> str:
    if not isinstance(value, dict) or set(value) not in (
        {"created", "job"},
        {"created", "updated", "job"},
    ):
        raise WrapperError("invalid_cli_response")
    if not isinstance(value["created"], bool):
        raise WrapperError("invalid_cli_response")
    if "updated" in value and not isinstance(value["updated"], bool):
        raise WrapperError("invalid_cli_response")
    if not value["created"] and not value.get("updated", False):
        raise WrapperError("invalid_cli_response")
    job = value["job"]
    try:
        return _safe_id(job["id"] if isinstance(job, dict) else None)
    except (KeyError, ValueError):
        raise WrapperError("invalid_cli_response") from None


def _parse_declare(value: Any) -> str:
    """Accept an already-declared routine as success.

    `openclaw cron add` with a declaration key returns created=false,
    updated=false when the job already exists unchanged. For a one-shot that is
    suspicious, so _parse_create rejects it — but for a RECURRING routine it is
    the normal idempotent outcome, and rejecting it makes approve unretryable.
    """
    if not isinstance(value, dict) or set(value) not in (
        {"created", "job"},
        {"created", "updated", "job"},
    ):
        raise WrapperError("invalid_cli_response")
    if not isinstance(value["created"], bool):
        raise WrapperError("invalid_cli_response")
    if "updated" in value and not isinstance(value["updated"], bool):
        raise WrapperError("invalid_cli_response")
    job = value["job"]
    try:
        return _safe_id(job["id"] if isinstance(job, dict) else None)
    except (KeyError, ValueError):
        raise WrapperError("invalid_cli_response") from None


def _parse_status(value: Any) -> None:
    expected = {"enabled", "storePath", "storage", "sqlitePath", "jobs", "nextWakeAtMs"}
    if not isinstance(value, dict) or set(value) != expected:
        raise WrapperError("invalid_cli_response")
    next_wake = value["nextWakeAtMs"]
    if (
        value["enabled"] is not True
        or value["storage"] != "sqlite"
        or not isinstance(value["storePath"], str)
        or not isinstance(value["sqlitePath"], str)
        or not _is_int(value["jobs"])
        or value["jobs"] < 0
        or (next_wake is not None and (not _is_int(next_wake) or next_wake < 0))
    ):
        raise WrapperError("invalid_cli_response")


def _parse_job(value: Any, expected_id: str | None = None) -> dict[str, str]:
    if not isinstance(value, dict):
        raise WrapperError("invalid_cli_response")
    try:
        job_id = _safe_id(value.get("id"))
        if expected_id is not None and job_id != expected_id:
            raise ValueError("unexpected id")
        normalized = {"jobId": job_id}
        if "declarationKey" in value:
            normalized["taskId"] = _safe_id(value["declarationKey"], task=True)
        return normalized
    except ValueError:
        raise WrapperError("invalid_cli_response") from None


def _parse_cron_list(value: Any) -> list[dict[str, str]]:
    required = {"jobs", "total", "offset", "limit", "hasMore", "nextOffset"}
    if not isinstance(value, dict) or set(value) not in (
        required,
        required | {"deliveryPreviews"},
    ):
        raise WrapperError("invalid_cli_response")
    jobs = value["jobs"]
    integers = (value["total"], value["offset"], value["limit"])
    if (
        not isinstance(jobs, list)
        or len(jobs) > 200
        or any(not _is_int(item) or item < 0 for item in integers)
        or value["offset"] != 0
        or value["total"] != len(jobs)
        or not isinstance(value["hasMore"], bool)
        or value["hasMore"]
        or value["nextOffset"] is not None
    ):
        raise WrapperError("invalid_cli_response")
    return [_parse_job(job) for job in jobs]


_RUN_STATUSES = frozenset(
    ("queued", "running", "succeeded", "failed", "cancelled", "error", "timeout")
)
_MAX_RESULT_SCALARS = 32_768


def _parse_run_result(value: Any, job_id: str) -> dict[str, Any]:
    """Newest run record for one cron job, normalized for the coordinator.

    The delivery leg of the installed OpenClaw never fires its cron webhook
    (2026.7.1-2), so the coordinator reconciles by polling this projection.
    Returns {"found": False} when no run for the job exists yet.
    """
    if not isinstance(value, dict) or not isinstance(value.get("tasks"), list):
        raise WrapperError("cli_failed")
    newest: dict[str, Any] | None = None
    for entry in value["tasks"]:
        if not isinstance(entry, dict) or entry.get("sourceId") != job_id:
            continue
        created = entry.get("createdAt")
        if not _is_int(created):
            continue
        if newest is None or created > newest.get("createdAt", 0):
            newest = entry
    if newest is None:
        return {"run": {"found": False}}
    status = newest.get("status")
    if not isinstance(status, str) or status not in _RUN_STATUSES:
        raise WrapperError("cli_failed")
    run_id = newest.get("runId")
    if not isinstance(run_id, str) or SAFE_ID.fullmatch(run_id) is None:
        run_id = "reconcile"
    result = newest.get("terminalSummary")
    if not isinstance(result, str):
        result = ""
    return {
        "run": {
            "found": True,
            "status": status,
            "runId": run_id,
            "result": result[:_MAX_RESULT_SCALARS],
        }
    }


def _job_present(job_id: str, popen_factory: Callable[..., Any], timeout: float) -> bool:
    """Is this cron job still declared? Any listing failure answers 'yes', so an
    unverifiable cancellation is reported as a failure rather than assumed good."""
    try:
        jobs = _parse_cron_list(
            _run_json([OPENCLAW, "cron", "list", "--all", "--json"], popen_factory, timeout),
        )
    except WrapperError:
        return True
    return any(job.get("jobId") == job_id for job in jobs)


def _parse_remove(value: Any) -> None:
    if value != {"ok": True, "removed": True}:
        raise WrapperError("invalid_cli_response")


def _parse_running_tasks(value: Any, source_ids: set[str]) -> list[str]:
    if not isinstance(value, dict) or set(value) != {
        "count",
        "runtime",
        "status",
        "tasks",
    }:
        raise WrapperError("invalid_cli_response")
    tasks = value["tasks"]
    if (
        not _is_int(value["count"])
        or not isinstance(tasks, list)
        or value["count"] != len(tasks)
        or value["runtime"] != "cron"
        or value["status"] != "running"
    ):
        raise WrapperError("invalid_cli_response")
    targets: list[str] = []
    for record in tasks:
        if not isinstance(record, dict):
            raise WrapperError("invalid_cli_response")
        if record.get("sourceId") not in source_ids:
            continue
        if record.get("runtime") != "cron" or record.get("status") != "running":
            raise WrapperError("invalid_cli_response")
        target = None
        for key in ("runId", "taskId"):
            try:
                target = _safe_id(record.get(key))
                break
            except ValueError:
                pass
        if target is None:
            raise WrapperError("invalid_cli_response")
        if target not in targets:
            targets.append(target)
        if len(targets) > MAX_CANCEL_TARGETS:
            raise WrapperError("invalid_cli_response")
    return targets


def _running_targets(
    source_ids: set[str], popen_factory: Callable[..., Any], timeout: float
) -> list[str]:
    return _parse_running_tasks(
        _run_json(TASKS_LIST_ARGV, popen_factory, timeout), source_ids
    )


def _cancel_active(
    source_ids: set[str], popen_factory: Callable[..., Any], timeout: float
) -> bool:
    succeeded = True
    for _ in range(2):
        try:
            targets = _running_targets(source_ids, popen_factory, timeout)
        except WrapperError:
            succeeded = False
            continue
        for target in targets:
            try:
                _run_discard(
                    [OPENCLAW, "tasks", "cancel", target], popen_factory, timeout
                )
            except WrapperError:
                succeeded = False
    return succeeded


def execute(
    payload: dict[str, Any],
    *,
    popen_factory: Callable[..., Any] = subprocess.Popen,
    timeout: float = CLI_TIMEOUT_SECONDS,
) -> dict[str, Any]:
    operation = _validated_operation(payload)
    argv = build_argv(payload)
    if operation == "declareRoutine":
        # execute() is a SECOND dispatch alongside build_argv — an operation
        # must be handled in both or it falls through to the cancelDeclaration
        # tail and dies on payload["taskId"].
        return {"jobId": _parse_declare(_run_json(argv, popen_factory, timeout))}
    if operation == "create":
        return {"jobId": _parse_create(_run_json(argv, popen_factory, timeout))}
    if operation == "status":
        _parse_status(_run_json(argv, popen_factory, timeout))
        return {"ok": True}
    if operation == "list":
        return {"jobs": _parse_cron_list(_run_json(argv, popen_factory, timeout))}
    if operation == "get":
        return _parse_job(_run_json(argv, popen_factory, timeout), payload["jobId"])
    if operation == "runResult":
        return _parse_run_result(
            _run_json(argv, popen_factory, timeout), payload["jobId"]
        )
    if operation == "cancelRoutine":
        # Pausing a RECURRING routine means unscheduling it. It deliberately
        # does NOT kill an in-flight run: that costs a running-task listing plus
        # a recheck (~40s on the Pi), blowing the adapter ceiling, and a run
        # already underway may as well finish.
        job_id = payload["jobId"]
        try:
            _parse_remove(_run_json(argv, popen_factory, timeout))
        except WrapperError:
            if _job_present(job_id, popen_factory, timeout):
                raise WrapperError("cli_failed") from None
        return {"cancelled": True}

    if operation == "cancel":
        job_id = payload["jobId"]
        succeeded = True
        try:
            _parse_remove(_run_json(argv, popen_factory, timeout))
        except WrapperError:
            # Cancelling is idempotent: the CLI errors with "id not found" when
            # the job is already gone, and a caller retrying after a timeout
            # must not be told the cancellation failed. Verify by listing —
            # absent means cancelled, whatever `rm` said.
            succeeded = not _job_present(job_id, popen_factory, timeout)
        if not _cancel_active({job_id}, popen_factory, timeout):
            succeeded = False
        if not succeeded:
            raise WrapperError("cli_failed")
        return {"cancelled": True}

    task_id = payload["taskId"]
    jobs = _parse_cron_list(_run_json(argv, popen_factory, timeout))
    matched = {job["jobId"] for job in jobs if job.get("taskId") == task_id}
    if not matched:
        return {"cancelled": True}
    succeeded = True
    for job_id in sorted(matched):
        try:
            _parse_remove(
                _run_json(
                    [OPENCLAW, "cron", "rm", job_id, "--json"],
                    popen_factory,
                    timeout,
                )
            )
        except WrapperError:
            succeeded = False
    if not _cancel_active(matched, popen_factory, timeout):
        succeeded = False
    if not succeeded:
        raise WrapperError("cli_failed")
    return {"cancelled": True}


def read_request(stream: BinaryIO) -> dict[str, Any]:
    raw = stream.read(MAX_BYTES + 1)
    if len(raw) > MAX_BYTES:
        raise WrapperError("invalid_request")
    try:
        value = json.loads(
            raw.decode("utf-8"), object_pairs_hook=_reject_duplicate_keys
        )
    except (UnicodeDecodeError, json.JSONDecodeError, ValueError):
        raise WrapperError("invalid_request") from None
    if not isinstance(value, dict):
        raise WrapperError("invalid_request")
    return value


def main() -> int:
    try:
        request = read_request(sys.stdin.buffer)
        result = execute(request)
    except (WrapperError, ValueError) as error:
        code = error.code if isinstance(error, WrapperError) else "invalid_request"
        sys.stdout.write(json.dumps({"error": code}, separators=(",", ":")) + "\n")
        return 1
    sys.stdout.write(json.dumps(result, separators=(",", ":")) + "\n")
    return 0


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