source.py

"""Miner2 v4: risk-gated, deadline-bounded differential verification for SN99."""

import hashlib
import json
import os
import re
import resource
import signal
import subprocess
import sys
import tempfile
import time


_FORMAT = "miner2-risk-gated-stress-v4"
_MODEL = "openai/gpt-5.6-luna"
_EFFORTS = ("low", "medium", "high")
_OUTPUT_LIMIT = 8 * 1024 * 1024
_CASE_TIMEOUT_S = 8.0
_GEN_TIMEOUT_S = 4.0
_CHILD_CPU_S = 9
_CHILD_AS_BYTES = 1 << 30
_CHILD_NPROC = 16
_CHILD_NOFILE = 32
_STRESS_ROUNDS = 24
_MIN_VALID_STRESS = 8
_SIZES = (2, 3, 5, 8, 12, 20)

_SAMPLE = re.compile(r"^Sample (Input|Output)\s*(\d+)\s*$", re.MULTILINE)
_FENCE = re.compile(r"```(?:python|py)?\s*\n(.*?)```", re.DOTALL | re.IGNORECASE)
_NAMED = re.compile(r"```(reference|generator)\s*\n(.*?)```", re.DOTALL | re.IGNORECASE)
_UNSAFE = re.compile(
    r"\b(?:subprocess|multiprocessing|socket)\b|"
    r"\bos\s*\.\s*(?:fork|forkpty|posix_spawn|system|popen)\b|"
    r"\bpty\s*\.\s*spawn\b"
)
_CODE_REQUEST = (
    "Use the full constraints to choose a provably correct algorithm and suitable buffered I/O. "
    "Check every explicit condition and published example. Return only complete raw Python 3 "
    "source, without Markdown fences or prose."
)
_TOOLS_REQUEST = (
    "Independently test the proposed solution. Return exactly two fenced blocks: `reference` must "
    "be a simple correctness-first program for small legal inputs and must not reuse the optimized "
    "algorithm; `generator` must accept seed and size arguments, seed Python random, and print one "
    "legal varied input. Output no text outside those blocks."
)
_REPAIR_REQUEST = (
    "A generated legal case disagrees with an independently derived small-case reference. Solve "
    "the original problem again from first principles; do not patch or special-case this case.\n"
    "Input:\n%s\nPrevious output:\n%s\nReference output:\n%s\nReturn only complete raw Python 3 source."
)
_INCONCLUSIVE = object()


def _is_code_prompt(text):
    value = str(text)
    return (
        "Write a complete Python 3 program" in value
        and "standard input" in value
        and "standard output" in value
    )


def _is_choice_prompt(text):
    body = "\n" + str(text)
    return all("\n" + letter + ")" in body for letter in "ABCD")


def _samples(prompt, maximum):
    text = str(prompt)
    marks = [(m.start(), m.end(), m.group(1), m.group(2)) for m in _SAMPLE.finditer(text)]
    blocks = {}
    for index, (_start, end, kind, number) in enumerate(marks):
        stop = marks[index + 1][0] if index + 1 < len(marks) else len(text)
        body = text[end:stop].replace("\r\n", "\n").replace("\r", "\n").strip("\n")
        if kind == "Output":
            body = body.split("\n\n", 1)[0]
        blocks.setdefault(number, {})[kind] = body.strip("\n")
    pairs = []
    for number in sorted(blocks, key=lambda item: int(item) if item.isdigit() else item):
        row = blocks[number]
        if row.get("Input", "").strip() and "Output" in row:
            pairs.append((row["Input"] + "\n", row["Output"]))
    return pairs[:maximum]


def _program(response):
    value = str(response or "")
    match = _FENCE.search(value)
    return (match.group(1) if match else value).strip()


def _named_blocks(response):
    return {name.lower(): code.strip() for name, code in _NAMED.findall(str(response or ""))}


def _limits():  # pragma: no cover - child only
    resource.setrlimit(resource.RLIMIT_CPU, (_CHILD_CPU_S, _CHILD_CPU_S))
    resource.setrlimit(resource.RLIMIT_AS, (_CHILD_AS_BYTES, _CHILD_AS_BYTES))
    resource.setrlimit(resource.RLIMIT_NPROC, (_CHILD_NPROC, _CHILD_NPROC))
    resource.setrlimit(resource.RLIMIT_NOFILE, (_CHILD_NOFILE, _CHILD_NOFILE))
    resource.setrlimit(resource.RLIMIT_FSIZE, (_OUTPUT_LIMIT, _OUTPUT_LIMIT))
    resource.setrlimit(resource.RLIMIT_CORE, (0, 0))
    os.setsid()


def _kill_group(process):
    try:
        os.killpg(os.getpgid(process.pid), signal.SIGKILL)
    except Exception:  # noqa: BLE001
        try:
            process.kill()
        except Exception:  # noqa: BLE001
            pass


def _execute(code, stdin_text, timeout, argv=()):
    if not str(code).strip():
        return "empty", ""
    if _UNSAFE.search(str(code)):
        return "rejected", ""
    path = None
    output = None
    process = None
    try:
        descriptor, path = tempfile.mkstemp(suffix=".py")
        with os.fdopen(descriptor, "w") as handle:
            handle.write(str(code))
        output = tempfile.TemporaryFile()
        process = subprocess.Popen(
            [sys.executable, path, *[str(item) for item in argv]],
            stdin=subprocess.PIPE,
            stdout=output,
            stderr=subprocess.DEVNULL,
            preexec_fn=_limits,
            close_fds=True,
            cwd=tempfile.gettempdir(),
            env={"PATH": "/usr/bin:/bin", "PYTHONIOENCODING": "utf-8"},
        )
        try:
            process.communicate(str(stdin_text).encode("utf-8"), timeout=timeout)
        except subprocess.TimeoutExpired:
            _kill_group(process)
            process.communicate(timeout=2)
            return "timeout", ""
        output.seek(0)
        raw = output.read(_OUTPUT_LIMIT + 1)
        if len(raw) > _OUTPUT_LIMIT:
            return "output_limit", ""
        if process.returncode != 0:
            return "crash", raw.decode("utf-8", "replace")
        return "ok", raw.decode("utf-8", "replace")
    except Exception:  # noqa: BLE001
        return "harness", ""
    finally:
        if process is not None and process.poll() is None:
            _kill_group(process)
        if output is not None:
            output.close()
        if path:
            try:
                os.unlink(path)
            except OSError:
                pass


def _first_failure(answer, cases):
    if not cases:
        return None
    code = _program(answer)
    if not code:
        return cases[0][0], "<empty>", cases[0][1]
    if _UNSAFE.search(code):
        return _INCONCLUSIVE
    for stdin_text, expected in cases:
        status, observed = _execute(code, stdin_text, _CASE_TIMEOUT_S)
        if status in ("rejected", "timeout", "output_limit", "harness"):
            return _INCONCLUSIVE
        if status != "ok":
            return stdin_text, "<%s>" % status, expected
        if observed.split() != expected.split():
            return stdin_text, observed.strip(), expected.strip()
    return None


def _same_values(left, right, rel=1e-7):
    a, b = left.split(), right.split()
    if len(a) != len(b):
        return False
    for x, y in zip(a, b):
        if x == y:
            continue
        try:
            fx, fy = float(x), float(y)
        except ValueError:
            return False
        if abs(fx - fy) > rel * max(1.0, abs(fx), abs(fy)):
            return False
    return True


def _stress(solution, reference, generator):
    valid = 0
    first_mismatch = None
    for index in range(_STRESS_ROUNDS):
        size = _SIZES[index % len(_SIZES)]
        status, case = _execute(generator, "", _GEN_TIMEOUT_S, argv=(7919 + index, size))
        if status != "ok" or not case.strip():
            continue
        ref_status, wanted = _execute(reference, case, _CASE_TIMEOUT_S)
        if ref_status != "ok":
            continue
        valid += 1
        if first_mismatch is not None:
            continue
        got_status, got = _execute(solution, case, _CASE_TIMEOUT_S)
        if got_status == "ok" and (got.split() == wanted.split() or _same_values(got, wanted)):
            continue
        observed = got.strip() if got_status == "ok" else "<%s>" % got_status
        first_mismatch = (case, observed, wanted.strip())
    return first_mismatch, valid


def _needs_verification(prompt, policy):
    text = str(prompt)
    lower = text.lower()
    return (
        len(text) >= policy["verify_min_chars"]
        and (
            lower.count("operation") >= policy["verify_operation_mentions"]
            or lower.count("swap") >= policy["verify_swap_mentions"]
        )
    )


def _load_policy(weights):
    try:
        policy = json.loads(bytes(weights).decode("utf-8"))
    except Exception as exc:
        raise ValueError("miner2-v4 weights are not valid JSON") from exc
    allowed = {
        "format", "model", "draft_effort", "floor_effort", "repair_effort",
        "tools_effort", "max_tokens", "max_examples", "prompt_revision",
        "verify_min_chars", "verify_operation_mentions", "verify_swap_mentions",
        "verify_start_deadline_s", "repair_start_deadline_s",
    }
    if not isinstance(policy, dict) or set(policy) != allowed:
        raise ValueError("miner2-v4 policy schema is malformed")
    if policy.get("format") != _FORMAT or policy.get("model") != _MODEL:
        raise ValueError("miner2-v4 policy identity is malformed")
    if any(policy.get(key) not in _EFFORTS for key in
           ("draft_effort", "floor_effort", "repair_effort", "tools_effort")):
        raise ValueError("miner2-v4 effort policy is malformed")
    if type(policy.get("max_tokens")) is not int or policy["max_tokens"] != 8192:
        raise ValueError("miner2-v4 token limit is malformed")
    if type(policy.get("max_examples")) is not int or not 1 <= policy["max_examples"] <= 8:
        raise ValueError("miner2-v4 example limit is malformed")
    if policy.get("prompt_revision") != 4:
        raise ValueError("miner2-v4 prompt revision is malformed")
    if (policy.get("verify_min_chars") != 1400
            or policy.get("verify_operation_mentions") != 4
            or policy.get("verify_swap_mentions") != 2):
        raise ValueError("miner2-v4 risk policy is malformed")
    if (policy.get("verify_start_deadline_s") != 28
            or policy.get("repair_start_deadline_s") != 30):
        raise ValueError("miner2-v4 deadline policy is malformed")
    return policy


def build_agent(weights):
    policy = _load_policy(weights)
    batch_started = [None]

    def params(effort):
        return {"max_tokens": policy["max_tokens"], "reasoning": {"effort": effort}}

    def ask(call_model, content, effort):
        return call_model(_MODEL, [{"role": "user", "content": content}], params(effort))

    def agent(prompt, call_model):
        original = str(prompt)
        if batch_started[0] is None:
            batch_started[0] = time.monotonic()
        if _is_choice_prompt(original):
            return ask(call_model, original, policy["floor_effort"])
        if not _is_code_prompt(original):
            marker = int.from_bytes(hashlib.sha256(original.encode("utf-8")).digest()[:8], "big")
            request = (
                original + "\n\nSolve in the requested units and put the final numeric answer "
                "alone on the last line. Ignore audit marker %d; it is answer-independent "
                "metadata and must not appear in the answer." % marker
            )
            return ask(call_model, request, policy["floor_effort"])

        draft = ask(call_model, original + "\n\n" + _CODE_REQUEST, policy["draft_effort"])
        cases = _samples(original, policy["max_examples"])
        sample_failure = _first_failure(draft, cases)
        if sample_failure is not None and sample_failure is not _INCONCLUSIVE:
            repaired = ask(
                call_model, original + "\n\n" + (_REPAIR_REQUEST % sample_failure),
                policy["repair_effort"],
            )
            return repaired if _first_failure(repaired, cases) is None else draft
        if sample_failure is _INCONCLUSIVE or not cases:
            return draft

        if not _needs_verification(original, policy):
            return draft
        if time.monotonic() - batch_started[0] > policy["verify_start_deadline_s"]:
            return draft

        tools = ask(call_model, original + "\n\n" + _TOOLS_REQUEST, policy["tools_effort"])
        blocks = _named_blocks(tools)
        reference, generator = blocks.get("reference"), blocks.get("generator")
        if not reference or not generator or _first_failure(reference, cases) is not None:
            return draft
        mismatch, valid = _stress(_program(draft), reference, generator)
        if mismatch is None or valid < _MIN_VALID_STRESS:
            return draft
        if time.monotonic() - batch_started[0] > policy["repair_start_deadline_s"]:
            return draft
        repaired = ask(
            call_model, original + "\n\n" + (_REPAIR_REQUEST % mismatch),
            policy["repair_effort"],
        )
        if _first_failure(repaired, cases) is not None:
            return draft
        disagreement, checked = _stress(_program(repaired), reference, generator)
        return repaired if disagreement is None and checked >= _MIN_VALID_STRESS else draft

    return agent