source.py

"""Miner2 v8: risk-gated diverse execution consensus under one shared clock."""

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


_FORMAT = "miner2-risk-gated-consensus-v8"
_PRIMARY_MODEL = "openai/gpt-5.6-luna"
_SECONDARY_MODEL = "google/gemini-3.6-flash"
_EFFORTS = ("low", "medium", "high")
_OUTPUT_LIMIT = 4 * 1024 * 1024
_CASE_TIMEOUT_S = 8.0
_GENERATOR_TIMEOUT_S = 4.0
_CHILD_CPU_S = 9
_CHILD_AS_BYTES = 1 << 30
_CHILD_NPROC = 16
_CHILD_NOFILE = 32
_EXPECTED_TASKS = 6
_SIZES = (2, 3, 5, 8, 13, 21)

_SAMPLE = re.compile(r"^Sample (Input|Output)\s*(\d+)\s*$", re.MULTILINE)
_FENCE = re.compile(r"```(?:python|py)?\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"
)

_PRIMARY_INSTRUCTION = " ".join((
    "Derive a complete solution from the full statement and constraints.",
    "Privately create a small-state specification and use it to challenge your algorithm.",
    "Check boundaries, ordering, multiplicity, repeated values, overflow, and complexity.",
    "Trace every published example and return only complete raw Python 3 source with no Markdown or prose.",
))
_REVIEW_INSTRUCTION = " ".join((
    "Work independently without seeing another solution.",
    "Return one raw executable Python program derived by a structurally different route.",
    "Oracle must be a simple correctness-first program for small legal inputs and must not reuse the candidate algorithm.",
    "Generator must accept seed and size arguments, seed Python random, and print one varied legal input.",
    "Append oracle lines between comments `# ORACLE-BEGIN` and `# ORACLE-END`, prefixing every line with `#|`.",
    "Append generator lines the same way between `# GENERATOR-BEGIN` and `# GENERATOR-END`.",
    "Those comment sections must be inert when the complete response is executed as the candidate.",
))
_REPAIR_INSTRUCTION = " ".join((
    "A concrete executed legal case disproved both available approaches.",
    "Derive a fresh general solution from the full statement; do not patch, fingerprint, or special-case the case.",
    "Return only complete raw Python 3 source with no Markdown or prose.",
))
_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 _needs_review(text, policy):
    value = str(text)
    lower = value.lower()
    stateful = any(word in lower for word in ("operation", "replace", "update", "query"))
    ordered = any(phrase in lower for phrase in ("in order", "one by one", "sequential", "after each"))
    repeated = lower.count("operation") >= 3 or lower.count("query") >= 3
    return len(value) >= policy["review_min_chars"] or (stateful and ordered) or repeated


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 value: int(value) if value.isdigit() else value):
        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 _review_parts(response):
    value = str(response or "").strip()
    found = {"oracle": [], "generator": []}
    mode = None
    for line in value.splitlines():
        tag = line.strip()
        if tag == "# ORACLE-BEGIN":
            mode = "oracle"
            continue
        if tag == "# ORACLE-END":
            mode = None
            continue
        if tag == "# GENERATOR-BEGIN":
            mode = "generator"
            continue
        if tag == "# GENERATOR-END":
            mode = None
            continue
        if mode is not None:
            if not line.startswith("#|"):
                return value, "", ""
            found[mode].append(line[2:])
    return value, "\n".join(found["oracle"]).strip(), "\n".join(found["generator"]).strip()


def _limits():  # pragma: no cover - subprocess 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(value) for value 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", ""
        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 _same_output(left, right, rel=1e-7):
    a, b = str(left).split(), str(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 _first_failure(answer, cases):
    if not cases:
        return _INCONCLUSIVE
    code = _program(answer)
    if not code or _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" or not _same_output(observed, expected):
            shown = observed.strip() if status == "ok" else "<%s>" % status
            return stdin_text, shown or "<empty>", expected.strip()
    return None


def _case_bank(oracle, generator, rounds):
    rows = []
    seen = set()
    for index in range(rounds):
        size = _SIZES[index % len(_SIZES)]
        status, case = _execute(
            generator, "", _GENERATOR_TIMEOUT_S, argv=(161803 + index, size)
        )
        if status != "ok" or not case.strip() or case in seen:
            continue
        oracle_status, wanted = _execute(oracle, case, _CASE_TIMEOUT_S)
        if oracle_status != "ok":
            continue
        seen.add(case)
        rows.append((case, wanted))
    return rows


def _counterexample(answer, bank):
    code = _program(answer)
    if not code or _UNSAFE.search(code):
        return _INCONCLUSIVE
    for case, wanted in bank:
        status, observed = _execute(code, case, _CASE_TIMEOUT_S)
        if status in ("rejected", "timeout", "output_limit", "harness"):
            return _INCONCLUSIVE
        if status != "ok" or not _same_output(observed, wanted):
            shown = observed.strip() if status == "ok" else "<%s>" % status
            return case, shown or "<empty>", wanted.strip()
    return None


def _load_policy(weights):
    try:
        policy = json.loads(bytes(weights).decode("utf-8"))
    except Exception as exc:
        raise ValueError("miner2-v8 weights are not valid JSON") from exc
    fields = {
        "code_call_cap", "floor_call_cap", "floor_effort", "format",
        "future_task_reserve_s", "max_examples", "min_valid_stress",
        "primary_effort", "primary_max_tokens", "primary_model",
        "repair_effort", "review_min_chars", "run_deadline_s",
        "secondary_effort", "secondary_max_tokens", "secondary_model",
        "strategy_revision", "stress_rounds",
    }
    if not isinstance(policy, dict) or set(policy) != fields:
        raise ValueError("miner2-v8 policy schema is malformed")
    identities = {
        "format": _FORMAT,
        "primary_model": _PRIMARY_MODEL,
        "secondary_model": _SECONDARY_MODEL,
    }
    if any(policy.get(name) != value for name, value in identities.items()):
        raise ValueError("miner2-v8 policy identity is malformed")
    for name in ("floor_effort", "primary_effort", "secondary_effort", "repair_effort"):
        if policy.get(name) not in _EFFORTS:
            raise ValueError("miner2-v8 effort policy is malformed")
    expected = {
        "code_call_cap": 3, "floor_call_cap": 1, "future_task_reserve_s": 35,
        "max_examples": 6, "min_valid_stress": 8, "primary_max_tokens": 32768,
        "review_min_chars": 1600, "run_deadline_s": 600,
        "secondary_max_tokens": 16384, "strategy_revision": 8, "stress_rounds": 20,
    }
    if any(type(policy.get(name)) is not int or policy[name] != value for name, value in expected.items()):
        raise ValueError("miner2-v8 bounded policy is malformed")
    return policy


def build_agent(weights):
    policy = _load_policy(weights)
    run_started = [None]
    served = [0]

    def agent(prompt, call_model):
        original = str(prompt)
        if run_started[0] is None:
            run_started[0] = time.monotonic()
        task_index = served[0]
        served[0] += 1
        calls = [0]
        is_code = _is_code_prompt(original)
        limit = policy["code_call_cap"] if is_code else policy["floor_call_cap"]

        def can_start(minimum):
            elapsed = time.monotonic() - run_started[0]
            future = max(0, _EXPECTED_TASKS - task_index - 1)
            reserved = future * policy["future_task_reserve_s"]
            return policy["run_deadline_s"] - elapsed >= minimum + reserved

        def request(model, content, effort, max_tokens, minimum):
            if calls[0] >= limit:
                raise RuntimeError("miner2-v8 per-task call limit exceeded")
            if not can_start(minimum):
                raise TimeoutError("miner2-v8 shared deadline reserve reached")
            calls[0] += 1
            params = {"max_tokens": max_tokens, "reasoning": {"effort": effort}}
            return call_model(model, [{"role": "user", "content": content}], params)

        if _is_choice_prompt(original):
            try:
                return request(
                    _PRIMARY_MODEL, original, policy["floor_effort"],
                    policy["secondary_max_tokens"], 20,
                )
            except Exception:
                return ""
        if not is_code:
            marker = int.from_bytes(hashlib.sha256(original.encode("utf-8")).digest()[:8], "big")
            numeric = original + "\n\n" + (
                "Solve in the requested units and put only the final numeric result on the last line. "
                "Audit marker %d is answer-independent metadata; do not reproduce it." % marker
            )
            try:
                return request(
                    _PRIMARY_MODEL, numeric, policy["floor_effort"],
                    policy["secondary_max_tokens"], 20,
                )
            except Exception:
                return ""

        try:
            primary = request(
                _PRIMARY_MODEL, original + "\n\n" + _PRIMARY_INSTRUCTION,
                policy["primary_effort"], policy["primary_max_tokens"], 45,
            )
        except Exception:
            return ""
        cases = _samples(original, policy["max_examples"])
        primary_sample = _first_failure(primary, cases)
        if primary_sample is None and not _needs_review(original, policy):
            return primary
        if primary_sample is _INCONCLUSIVE and not _needs_review(original, policy):
            return primary

        try:
            review = request(
                _SECONDARY_MODEL, original + "\n\n" + _REVIEW_INSTRUCTION,
                policy["secondary_effort"], policy["secondary_max_tokens"], 90,
            )
        except Exception:
            return primary
        secondary, oracle, generator = _review_parts(review)
        secondary_sample = _first_failure(secondary, cases) if secondary else _INCONCLUSIVE

        if primary_sample is None:
            fallback = primary
        elif secondary_sample is None:
            fallback = secondary
        else:
            fallback = primary

        if not oracle or not generator or _first_failure(oracle, cases) is not None:
            if primary_sample not in (None, _INCONCLUSIVE) and secondary_sample is not None:
                return _repair(original, primary_sample, fallback, cases, (), request, policy)
            return fallback
        bank = _case_bank(oracle, generator, policy["stress_rounds"])
        if len(bank) < policy["min_valid_stress"]:
            if primary_sample not in (None, _INCONCLUSIVE) and secondary_sample is not None:
                return _repair(original, primary_sample, fallback, cases, (), request, policy)
            return fallback

        primary_failure = _counterexample(primary, bank)
        secondary_failure = _counterexample(secondary, bank) if secondary else _INCONCLUSIVE
        if primary_sample is None and primary_failure is None:
            return primary
        if secondary_sample is None and secondary_failure is None:
            return secondary

        disproved = primary_failure if primary_failure not in (None, _INCONCLUSIVE) else primary_sample
        if disproved in (None, _INCONCLUSIVE):
            return fallback
        return _repair(original, disproved, fallback, cases, bank, request, policy)

    return agent


def _repair(original, counterexample, fallback, samples, bank, request, policy):
    case, observed, wanted = counterexample
    prompt = original + "\n\n" + _REPAIR_INSTRUCTION + (
        "\nExecuted input:\n%s\nCandidate output:\n%s\nOracle output:\n%s"
        % (case, observed, wanted)
    )
    try:
        repaired = request(
            _PRIMARY_MODEL, prompt, policy["repair_effort"],
            policy["primary_max_tokens"], 70,
        )
    except Exception:
        return fallback
    if _first_failure(repaired, samples) is not None:
        return fallback
    if bank and _counterexample(repaired, bank) is not None:
        return fallback
    return repaired