"""Bounded engineering witness for the report; no learning, search, or inference.

Only two operators and one local replication pattern are supported. This is not
a general-purpose runtime. Run with ordinary Python 3; results are written beside
this file. The hard wall/CPU limits are 30 seconds, with at most eight cells.
"""
from __future__ import annotations

from dataclasses import dataclass, replace
from pathlib import Path
import hashlib
import json
import math
import platform
import resource
import signal
import sys
import time


class ContractError(ValueError):
    pass


class BudgetError(ValueError):
    pass


@dataclass(frozen=True)
class Port:
    direction: str
    width: int
    semantic: str
    dtype: str = "float64"
    aggregation: str = "single"
    delay: int = 0
    clock: str = "external-step"


@dataclass(frozen=True)
class Rule:
    cue: str
    operator: str
    parameter: float
    replicate_offset: int | None = None


GENOME = (Rule("source", "scale", 2.0, 1),
          Rule("memory", "leaky_accumulator", 0.5))


@dataclass(frozen=True)
class Budget:
    nodes: int = 8
    edges: int = 16  # sum of developmental and execution edges
    ticks: int = 8
    width: int = 8


@dataclass(frozen=True)
class Cell:
    lineage: tuple[int, ...]
    position: int
    width: int
    operator: str | None = None
    parameter: float | None = None


@dataclass(frozen=True)
class Observation:
    """No graph, task labels, global node count, or desired outputs."""
    own: Cell
    cue: str
    neighboring_positions: tuple[int, ...]


@dataclass(frozen=True)
class Proposal:
    operator: str
    parameter: float
    replicate_offset: int | None


@dataclass(frozen=True)
class Model:
    cells: tuple[Cell, ...]
    developmental_edges: tuple[tuple[tuple[int, ...], tuple[int, ...]], ...]
    execution_edges: tuple[tuple[tuple[int, ...], tuple[int, ...]], ...]
    schedule: tuple[tuple[int, ...], ...]
    input_port: Port
    output_port: Port
    initial_state: tuple[float, ...]
    development_ticks: int
    learning_rules: tuple = ()


def local_rule(genome, observation):
    if observation.own.operator is not None:
        return None
    # First matching rule wins; rule order is part of g.
    for rule in genome:
        if rule.cue == observation.cue:
            return Proposal(rule.operator, rule.parameter, rule.replicate_offset)
    return None


def compatible(source, destination):
    if source.direction != "out" or destination.direction != "in":
        raise ContractError("edge directions")
    if replace(source, direction="in") != destination:
        raise ContractError("shape/type/meaning/aggregation/delay/clock mismatch")


def develop_R(genome, seed, environment, budget):
    """Frozen witness R-v1: synchronous snapshots, atomic bounded commits.

    environment maps public positions to cues. A rule can inspect only its cue
    and occupied adjacent positions; it may create only an adjacent child.
    Arbitrary connections, pruning, feedback, randomness, and learning are
    intentionally unsupported here (their proposed semantics are in research.md).
    """
    if not 1 <= seed.width <= budget.width:
        raise BudgetError("width cap")
    if seed.operator is not None or seed.lineage != (0,):
        raise ContractError("seed must be the single undecided root")
    cells, signaling, execution = (seed,), (), ()
    for tick in range(budget.ticks):
        snapshot = cells
        proposals = []
        for cell in snapshot:
            adjacent = tuple(c.position for c in snapshot
                             if abs(c.position - cell.position) == 1)
            observed = Observation(cell, environment.get(cell.position, ""), adjacent)
            proposal = local_rule(genome, observed)
            if proposal is not None:
                proposals.append((cell, proposal))
        if not proposals:
            break
        next_cells = {c.lineage: c for c in snapshot}
        occupied = {c.position for c in snapshot}
        next_signal, next_execution = list(signaling), list(execution)
        # Lineage ordering is solely an explicit commit/conflict convention.
        for cell, proposal in sorted(proposals, key=lambda item: item[0].lineage):
            if proposal.operator not in {"scale", "leaky_accumulator"}:
                raise ContractError("unknown operator")
            if not math.isfinite(proposal.parameter):
                raise ContractError("nonfinite parameter")
            if proposal.operator == "leaky_accumulator" and not 0 <= proposal.parameter < 1:
                raise ContractError("accumulator must be contractive")
            next_cells[cell.lineage] = replace(cell, operator=proposal.operator,
                                                parameter=proposal.parameter)
            if proposal.replicate_offset is not None:
                if proposal.operator != "scale" or abs(proposal.replicate_offset) != 1:
                    raise ContractError("unsupported replication")
                position = cell.position + proposal.replicate_offset
                if position in occupied or position not in environment:
                    raise ContractError("occupied or undeclared child site")
                occupied.add(position)
                child = Cell(cell.lineage + (0,), position, cell.width)
                next_cells[child.lineage] = child
                next_signal.append((cell.lineage, child.lineage))
                next_execution.append((cell.lineage, child.lineage))
        if len(next_cells) > budget.nodes:
            raise BudgetError("node cap")
        if len(next_signal) + len(next_execution) > budget.edges:
            raise BudgetError("combined edge cap")
        cells = tuple(next_cells[key] for key in sorted(next_cells))
        signaling, execution = tuple(next_signal), tuple(next_execution)
    else:
        raise BudgetError("development did not quiesce within tick cap")
    if len(cells) != 2 or any(c.operator is None for c in cells):
        raise ContractError("witness requires two differentiated cells")
    source, sink = cells
    if (source.operator, sink.operator) != ("scale", "leaky_accumulator"):
        raise ContractError("unsupported operator schedule")
    edge = ((source.lineage, sink.lineage),)
    if execution != edge:
        raise ContractError("unsupported execution topology")
    source_out = Port("out", source.width, "scaled-feature")
    sink_in = Port("in", sink.width, "scaled-feature")
    compatible(source_out, sink_in)
    return Model(cells, signaling, execution, (source.lineage, sink.lineage),
                 Port("in", source.width, "raw-feature"),
                 Port("out", sink.width, "low-pass-feature"),
                 (0.0,) * sink.width, tick + 1)


def vector(values, width):
    if len(values) != width:
        raise ContractError("runtime vector width")
    if any(isinstance(x, bool) or not isinstance(x, (int, float))
           or not math.isfinite(x) for x in values):
        raise ContractError("runtime finite numeric values")
    return tuple(float(x) for x in values)


def execute(model, inputs, state=None):
    if len(inputs) > 128:
        raise BudgetError("sequence cap")
    h = model.initial_state if state is None else vector(state, model.output_port.width)
    gain, leak = (c.parameter for c in model.cells)
    outputs = []
    for raw in inputs:
        x = vector(raw, model.input_port.width)
        u = tuple(gain * value for value in x)
        h = tuple(leak * old + (1.0 - leak) * value for old, value in zip(h, u))
        h = vector(h, model.output_port.width)
        outputs.append(h)
    return tuple(outputs), h


def require(condition, label):
    if not condition:
        raise AssertionError(label)


def rejected(fn, exception):
    try:
        fn()
    except exception:
        return True
    raise AssertionError("expected rejection")


def main():
    signal.signal(signal.SIGALRM, lambda *_: (_ for _ in ()).throw(TimeoutError("wall cap")))
    signal.setitimer(signal.ITIMER_REAL, 30)
    resource.setrlimit(resource.RLIMIT_CPU, (30, 30))
    memory_cap = 512 * 1024 * 1024
    memory_enforcement = "RLIMIT_AS plus bounded structures and measured RSS"
    try:
        resource.setrlimit(resource.RLIMIT_AS, (memory_cap, memory_cap))
    except ValueError:
        # Darwin rejects this AS limit. No large allocation is attempted: all
        # structures and loops are capped, and measured RSS must fit the ceiling.
        memory_enforcement = "bounded structures plus measured RSS; RLIMIT_AS unavailable on this host"
    started, cpu_started = time.perf_counter(), time.process_time()
    passed = []

    def check(label, condition):
        require(condition, label)
        passed.append(label)

    seed, environment = Cell((0,), 0, 2), {0: "source", 1: "memory"}
    model = develop_R(GENOME, seed, environment, Budget())
    inputs = ((2.0, -2.0), (0.0, 4.0))
    outputs, state = execute(model, inputs)
    check("heterogeneous composition and numeric recurrence", outputs == ((2.0, -2.0), (1.0, 3.0)))
    check("deterministic development", model == develop_R(GENOME, seed, environment, Budget()))
    check("development/execution edges accounted separately", len(model.developmental_edges) == len(model.execution_edges) == 1)
    continued, _ = execute(model, ((0.0, 0.0),), state)
    reset, _ = execute(model, ((0.0, 0.0),))
    check("explicit carry versus reset", continued == ((0.5, 1.5),) and reset == ((0.0, 0.0),))
    check("node cap rejection", rejected(lambda: develop_R(GENOME, seed, environment, Budget(nodes=1)), BudgetError))
    check("edge cap rejection", rejected(lambda: develop_R(GENOME, seed, environment, Budget(edges=1)), BudgetError))
    check("tick cap rejection", rejected(lambda: develop_R(GENOME, seed, environment, Budget(ticks=1)), BudgetError))
    check("missing cue rejection", rejected(lambda: develop_R(GENOME, seed, {0: "source", 1: ""}, Budget()), ContractError))
    check("shape rejection", rejected(lambda: compatible(Port("out", 2, "feature"), Port("in", 3, "feature")), ContractError))
    check("semantic rejection despite equal shapes", rejected(lambda: compatible(Port("out", 2, "logits"), Port("in", 2, "probabilities")), ContractError))
    check("delay rejection", rejected(lambda: compatible(Port("out", 2, "feature"), Port("in", 2, "feature", delay=1)), ContractError))
    check("invalid runtime input rejection", rejected(lambda: execute(model, ((float("nan"), 0.0),)), ContractError))
    symmetric_a = Observation(Cell((0,), 0, 2), "source", ())
    symmetric_b = Observation(Cell((9,), 0, 2), "source", ())
    check("shared deterministic rule cannot break identical observed roles", local_rule(GENOME, symmetric_a) == local_rule(GENOME, symmetric_b))
    twin_a, twin_b = state, state
    growth_error = 0.0
    for raw in ((1.0, 2.0), (-1.0, 0.0), (0.0, 0.0)):
        _, state = execute(model, (raw,), state)
        _, twin_a = execute(model, (raw,), twin_a)
        _, twin_b = execute(model, (raw,), twin_b)
        mean = tuple((a + b) / 2.0 for a, b in zip(twin_a, twin_b))
        growth_error = max(growth_error, max(abs(a - b) for a, b in zip(state, mean)))
    check("explicit state duplication plus mean readout retains trajectory", growth_error == 0.0)
    check("naive sum duplication changes behavior", tuple(a + b for a, b in zip(twin_a, twin_b)) != state)
    _, remembered = execute(model, inputs)
    sibling = develop_R(GENOME, seed, environment, Budget())
    check("topology regrowth does not recover erased state", sibling.initial_state != remembered)
    check("independent widths accepted by witness", all(develop_R(GENOME, replace(seed, width=w), environment, Budget()).input_port.width == w for w in (1, 3, 8)))
    usage = resource.getrusage(resource.RUSAGE_SELF)
    peak_bytes = usage.ru_maxrss if sys.platform == "darwin" else usage.ru_maxrss * 1024
    require(peak_bytes <= memory_cap, "RSS ceiling")
    result = {
        "status": "passed", "scope": "engineering witness only; no trained models or experiments",
        "runtime_version": "R-v1", "executed_at_utc": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
        "python": platform.python_version(), "source_sha256": hashlib.sha256(Path(__file__).read_bytes()).hexdigest(),
        "checks_passed": len(passed), "checks": passed,
        "development_ticks": model.development_ticks, "nodes": len(model.cells),
        "developmental_edges": len(model.developmental_edges), "execution_edges": len(model.execution_edges),
        "example_inputs": inputs, "example_outputs": outputs,
        "duplicated_state_mean_readout_max_error": growth_error,
        "wall_seconds": time.perf_counter() - started, "check_cpu_seconds": time.process_time() - cpu_started,
        "process_cpu_seconds": usage.ru_utime + usage.ru_stime, "process_peak_rss_bytes": peak_bytes,
        "memory_enforcement": memory_enforcement,
        "earlier_attempts": ["Python 3.9 postponed-annotation compatibility corrected before checks ran",
                             "Darwin rejected RLIMIT_AS before checks ran; bounded structures and RSS validation used"],
        "limits": {"workers": 1, "wall_seconds": 30, "cpu_seconds": 30, "address_space_bytes": 512 * 1024 * 1024,
                   "nodes": 8, "combined_edges": 16, "development_ticks": 8, "vector_width": 8, "sequence_length": 128},
        "limitations": ["hand-authored gain and memory primitive", "no adaptation or capacity advantage tested",
                        "only the two-cell schedule implemented", "no pruning, randomness, or feedback implementation",
                        "numeric equality here is not a theorem for arbitrary operators"]
    }
    Path(__file__).with_name("check_results.json").write_text(json.dumps(result, indent=2) + "\n")
    print(json.dumps({key: result[key] for key in ("status", "checks_passed", "wall_seconds", "process_peak_rss_bytes")}))


if __name__ == "__main__":
    main()