#!/usr/bin/env python3
# SPDX-License-Identifier: GPL-3.0-or-later
"""Offline-only passive LF-batch contract for synthetic shsrv inputs.

The module has no transport, CLI, address, clock, file output or prompt
detector.  It formats one bounded batch from an already validated Phase-1.0W
plan and seals supplied bytes only when the caller supplies a synthetic hard
deadline event.
"""

from __future__ import annotations

from dataclasses import dataclass
from typing import Any

from phase10v_shsrv_collector_model import CollectorError, OfflineCollector
from phase10w_shsrv_client_policy import SessionPlan, WINDOW_COMMANDS


IAC = 0xFF
MAX_LITERAL_PATH_BYTES = 512
MAX_BATCH_BYTES = 1035
KNOWN_HELP_FAMILIES = {
    "OFFICIAL_V07_SOURCE_FAMILY_CANDIDATE",
    "OFFICIAL_V019_SOURCE_FAMILY_CANDIDATE",
}


class PassiveContractError(RuntimeError):
    """Fail-closed contract error that never embeds supplied bytes or paths."""


@dataclass(frozen=True)
class PassiveBatch:
    """Target-free immutable outbound bytes for one synthetic session."""

    window: str
    payload: bytes
    expected_paths: tuple[str, ...]
    command_count: int
    deadline_seconds: int
    outbound_batches: int = 1
    lf_only: bool = True
    telnet_negotiation: bool = False
    retry_allowed: bool = False
    reconnect_allowed: bool = False
    resume_allowed: bool = False


def _validate_payload(payload: bytes) -> None:
    if not payload or len(payload) > MAX_BATCH_BYTES:
        raise PassiveContractError("batch length is invalid")
    if any(value in payload for value in (0x00, 0x0D, IAC)):
        raise PassiveContractError("batch contains a forbidden byte")
    if any(value > 0x7F for value in payload):
        raise PassiveContractError("batch is not ASCII")
    if not payload.endswith(b"\n"):
        raise PassiveContractError("batch is not LF terminated")


def _validate_batch(batch: PassiveBatch) -> None:
    _validate_payload(batch.payload)
    if batch.outbound_batches != 1 or batch.lf_only is not True or \
            batch.telnet_negotiation is not False or \
            batch.retry_allowed is not False or \
            batch.reconnect_allowed is not False or batch.resume_allowed is not False:
        raise PassiveContractError("batch safety fields are invalid")
    if not isinstance(batch.deadline_seconds, int) or isinstance(
            batch.deadline_seconds, bool) or not 1 <= batch.deadline_seconds <= 10:
        raise PassiveContractError("batch deadline is invalid")
    if batch.window == "T2_GREETING_AND_HELP":
        if batch.payload != b"help\n" or batch.expected_paths != () or \
                batch.command_count != 1:
            raise PassiveContractError("help batch contract is invalid")
        return
    if batch.window == "T3_ONE_EXACT_PATH" and \
            len(batch.expected_paths) == 1 and batch.command_count == 2:
        path = batch.expected_paths[0]
        try:
            OfflineCollector._validate_expected_paths({path})
            encoded = path.encode("ascii", errors="strict")
        except (CollectorError, UnicodeEncodeError) as error:
            raise PassiveContractError("path batch contract is invalid") from error
        expected = b"stat " + encoded + b"\nsum " + encoded + b"\n"
        if len(encoded) <= MAX_LITERAL_PATH_BYTES and batch.payload == expected:
            return
    raise PassiveContractError("path batch contract is invalid")


def build_passive_batch(plan: SessionPlan) -> PassiveBatch:
    """Build one target-free LF batch from an exact Phase-1.0W plan."""
    if not isinstance(plan, SessionPlan):
        raise PassiveContractError("session plan type is invalid")
    expected_commands = WINDOW_COMMANDS.get(plan.window)
    if expected_commands is None or plan.commands != expected_commands:
        raise PassiveContractError("window command sequence is invalid")
    if not isinstance(plan.deadline_seconds, int) or isinstance(
            plan.deadline_seconds, bool) or not 1 <= plan.deadline_seconds <= 10:
        raise PassiveContractError("deadline is invalid")

    expected_paths: tuple[str, ...]
    if plan.window == "T2_GREETING_AND_HELP":
        if plan.exact_literal_path is not None:
            raise PassiveContractError("help window contains a path")
        payload = b"help\n"
        expected_paths = ()
    elif plan.window == "T3_ONE_EXACT_PATH":
        path = plan.exact_literal_path
        if not isinstance(path, str):
            raise PassiveContractError("exact path is missing")
        try:
            OfflineCollector._validate_expected_paths({path})
            encoded = path.encode("ascii", errors="strict")
        except (CollectorError, UnicodeEncodeError) as error:
            raise PassiveContractError("exact path is invalid") from error
        if len(encoded) > MAX_LITERAL_PATH_BYTES:
            raise PassiveContractError("exact path is too long")
        payload = b"stat " + encoded + b"\nsum " + encoded + b"\n"
        expected_paths = (path,)
    else:
        raise PassiveContractError("window is not allowlisted")

    batch = PassiveBatch(
        window=plan.window,
        payload=payload,
        expected_paths=expected_paths,
        command_count=len(plan.commands),
        deadline_seconds=plan.deadline_seconds,
    )
    _validate_batch(batch)
    return batch


class PassiveResultAccumulator:
    """Consume supplied chunks once; never observes prompts or remote EOF."""

    def __init__(self, batch: PassiveBatch) -> None:
        if not isinstance(batch, PassiveBatch):
            raise PassiveContractError("batch type is invalid")
        _validate_batch(batch)
        self.batch = batch
        self._collector = OfflineCollector()
        self.state = "READY"

    def _invalidate(self, message: str) -> None:
        if self._collector.state not in {"SEALED", "INVALID", "ABORTED"}:
            try:
                self._collector.abort()
            except CollectorError:
                pass
        self.state = "INVALID"
        raise PassiveContractError(message)

    def feed_supplied_chunk(self, chunk: bytes) -> None:
        if self.state not in {"READY", "RECEIVING"}:
            raise PassiveContractError("accumulator is not accepting input")
        if not isinstance(chunk, bytes):
            self._invalidate("input must be bytes")
        if IAC in chunk:
            self._invalidate("unexpected Telnet control byte")
        try:
            self._collector.feed(chunk)
        except CollectorError as error:
            self._invalidate("collector rejected supplied input")
            raise AssertionError("unreachable") from error
        if chunk:
            self.state = "RECEIVING"

    def abort(self) -> None:
        if self.state not in {"READY", "RECEIVING"}:
            raise PassiveContractError("accumulator can no longer abort")
        try:
            self._collector.abort()
        except CollectorError as error:
            raise PassiveContractError("collector abort failed") from error
        self.state = "ABORTED"

    def _validate_complete_result(self, result: dict[str, Any]) -> None:
        if self.batch.window == "T2_GREETING_AND_HELP":
            fingerprint = result.get("command_fingerprint", {})
            if fingerprint.get("source_family_match") not in KNOWN_HELP_FAMILIES:
                raise PassiveContractError("help response is incomplete or unknown")
            if result.get("classification") != "SOURCE_FAMILY_FINGERPRINT_ONLY":
                raise PassiveContractError("help response classification is invalid")
        elif self.batch.window == "T3_ONE_EXACT_PATH":
            observations = result.get("file_observations")
            if not isinstance(observations, list) or len(observations) != 1:
                raise PassiveContractError("file response is incomplete")
            observation = observations[0]
            required = {
                "path", "metadata_seen", "size", "weak_checksum",
                "weak_checksum_algorithm", "cryptographic_checksum",
                "proves_exact_binary",
            }
            if not required.issubset(observation) or observation.get("path") != \
                    self.batch.expected_paths[0]:
                raise PassiveContractError("file response fields are incomplete")
            if observation.get("metadata_seen") is not True or not isinstance(
                    observation.get("size"), int) or isinstance(
                    observation.get("size"), bool) or observation["size"] < 0:
                raise PassiveContractError("stat response is incomplete")
            if observation.get("weak_checksum_algorithm") != "BSD_ROTATE_16" or \
                    observation.get("cryptographic_checksum") is not False:
                raise PassiveContractError("sum response is incomplete")
            if result.get("classification") != "WEAK_FILE_CORRELATION_ONLY":
                raise PassiveContractError("file response classification is invalid")
        else:
            raise PassiveContractError("batch window is invalid")

    def seal_at_hard_deadline(
        self, hard_deadline_reached: bool,
    ) -> dict[str, Any]:
        """Seal once only after an externally supplied synthetic deadline."""
        if hard_deadline_reached is not True:
            self._invalidate("hard deadline event is absent")
        if self.state not in {"READY", "RECEIVING"}:
            raise PassiveContractError("accumulator cannot be sealed")
        try:
            result = self._collector.finalize(set(self.batch.expected_paths))
            self._validate_complete_result(result)
        except (CollectorError, PassiveContractError) as error:
            self.state = "INVALID"
            raise PassiveContractError("deadline result is invalid") from error
        self.state = "SEALED"
        result["passive_batch_contract"] = {
            "offline_only": True,
            "one_outbound_batch": True,
            "plain_lf_only": True,
            "telnet_negotiation_emitted": False,
            "telnet_control_received": False,
            "prompt_completion_used": False,
            "remote_eof_completion_used": False,
            "sealed_by_synthetic_hard_deadline": True,
            "source_family_selected": False,
            "retry_allowed": False,
            "reconnect_allowed": False,
            "device_behavior_proven": False,
            "exact_identity_proven": False,
        }
        return result