mirror of
https://github.com/razzant/ouroboros.git
synced 2026-10-03 04:07:04 +00:00
1500 lines
94 KiB
Python
1500 lines
94 KiB
Python
"""Pins of the live E2E stand runner (``devtools/e2e_live``). Default lane: no server, no sockets, no network (the
|
|
provider probes are monkeypatched and the lane pool is replaced by a fake). The one real-server test at the end is
|
|
the keyless ``--stub`` rehearsal of SM1 and carries the same three gates as the system_e2e lane."""
|
|
from __future__ import annotations
|
|
|
|
import collections
|
|
import dataclasses
|
|
import json
|
|
import os
|
|
import pathlib
|
|
import subprocess
|
|
import sys
|
|
import threading
|
|
import time
|
|
import types
|
|
import urllib.error
|
|
import urllib.request
|
|
|
|
import pytest
|
|
|
|
REPO_ROOT = pathlib.Path(__file__).resolve().parents[1]
|
|
if str(REPO_ROOT) not in sys.path:
|
|
sys.path.insert(0, str(REPO_ROOT))
|
|
|
|
from devtools.benchmarks.common import launcher_audit # noqa: E402
|
|
from devtools.benchmarks.common.manifests import repo_provenance # noqa: E402
|
|
from devtools.e2e_live import run_live_lanes, scenarios, stub_lane, ui_probe # noqa: E402
|
|
|
|
FAKE_KEY = "sk-or-v1-e2e-live-test-key-value-never-printed-0123456789"
|
|
|
|
|
|
def _commit(repo: pathlib.Path, message: str) -> str:
|
|
subprocess.run(["git", "add", "-A"], cwd=str(repo), check=True)
|
|
subprocess.run(["git", "-c", "user.name=t", "-c", "user.email=t@e.invalid", "commit", "-q", "-m", message],
|
|
cwd=str(repo), check=True)
|
|
return subprocess.run(["git", "rev-parse", "HEAD"], cwd=str(repo), check=True,
|
|
capture_output=True, text=True).stdout.strip()
|
|
|
|
|
|
def _git_seed(root: pathlib.Path, *, dirty: bool = False) -> pathlib.Path:
|
|
"""A tiny SOURCE checkout: one committed VERSION, optionally a TRACKED uncommitted edit."""
|
|
seed = root / "source"
|
|
seed.mkdir()
|
|
(seed / "VERSION").write_text("7.0.0-test\n", encoding="utf-8")
|
|
subprocess.run(["git", "init", "-q"], cwd=str(seed), check=True)
|
|
_commit(seed, "seed")
|
|
if dirty:
|
|
(seed / "VERSION").write_text("7.0.0-dirty\n", encoding="utf-8") # describe says -dirty
|
|
return seed
|
|
|
|
|
|
def _fake_lane(job, args, out, template, stagger, states, seed, budget=None, *, key="", seed_sha=""):
|
|
sid, attempt = job
|
|
lane = out / "lanes" / f"{sid}_a{attempt}"
|
|
ceiling = budget.ceiling(job) if budget is not None else None # the real lane reads it before spending
|
|
lane.mkdir(parents=True)
|
|
row = {"scenario": sid, "attempt": attempt, "status": "pass", "checks": {"fake": True}, "error": "",
|
|
"duration_sec": 0.1, "model_slots": {"OUROBOROS_MODEL": template.get("OUROBOROS_MODEL")},
|
|
"lane_total_budget_usd": ceiling,
|
|
"template_has_key": "OPENROUTER_API_KEY" in template, "key_handed": bool(key), "seed_sha": seed_sha}
|
|
(lane / "result.json").write_text(json.dumps(row), encoding="utf-8")
|
|
return row
|
|
|
|
|
|
class _Response:
|
|
def __init__(self, body: bytes) -> None:
|
|
self._body = body
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *_args):
|
|
return False
|
|
|
|
def read(self) -> bytes:
|
|
return self._body
|
|
|
|
|
|
def _fake_urlopen(key_body: bytes, credits_body: bytes, calls: list):
|
|
def fake(req, timeout=0):
|
|
assert req.headers["Authorization"] == f"Bearer {FAKE_KEY}"
|
|
calls.append(req.full_url)
|
|
return _Response(key_body if req.full_url.endswith("/key") else credits_body)
|
|
return fake
|
|
|
|
|
|
def _short_tmp(monkeypatch) -> None:
|
|
monkeypatch.setattr(run_live_lanes.tempfile, "gettempdir", lambda: "/tmp/short")
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Structure: the shared launcher gate, the table, the argv bounds
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
def test_launcher_passes_the_shared_structural_gate():
|
|
"""Admission is the outer boundary, confinement follows the handed source, only the seam
|
|
publishes the manifest — the SAME gate the benchmark family is held to, by source."""
|
|
source = (REPO_ROOT / "devtools" / "e2e_live" / "run_live_lanes.py").read_text(encoding="utf-8")
|
|
assert launcher_audit.audit_source(source, name="run_live_lanes.py") == []
|
|
|
|
|
|
def test_scenario_table_shape():
|
|
assert set(scenarios.SCENARIOS) == {"SM1", "SW1", "SK1"}
|
|
for sid, row in scenarios.SCENARIOS.items():
|
|
assert row.id == sid and row.prompt.strip() and row.title.strip()
|
|
assert isinstance(row.settings_overrides, dict)
|
|
assert callable(row.acceptance) and callable(row.stub_script)
|
|
sm1 = scenarios.SCENARIOS["SM1"].settings_overrides
|
|
assert sm1 == {"OUROBOROS_RUNTIME_MODE": "advanced", "OUROBOROS_REVIEW_ENFORCEMENT": "blocking"}
|
|
roster = json.loads(scenarios.SCENARIOS["SW1"].overrides("openai-compatible::mock-child")["OUROBOROS_SUBAGENTS"])
|
|
assert roster["items"][0]["route"]["target_id"] == "openai-compatible::mock-child"
|
|
assert scenarios.SCENARIOS["SW1"].overrides("x")["OUROBOROS_MAX_SUBAGENT_DEPTH"] == 1
|
|
# The budget reservation unit: SK1 mints two root tasks (author + dispatch), the others one
|
|
# (SW1's scouts spend under their single root's ceiling).
|
|
assert {sid: row.root_tasks for sid, row in scenarios.SCENARIOS.items()} == {"SM1": 1, "SW1": 1, "SK1": 2}
|
|
# Only SM1 lands a commit the post-task evolution absorbs: the --self-mod wait and check follow this flag.
|
|
assert {sid: row.expects_absorb for sid, row in scenarios.SCENARIOS.items()} == {"SM1": True, "SW1": False, "SK1": False}
|
|
# Stub scripts are role-keyed queues; SW1 needs every role the swarm wire interleaves.
|
|
sw1 = scenarios.SCENARIOS["SW1"].stub_script(REPO_ROOT)
|
|
assert set(sw1) == {"agent", "child", "probe"}
|
|
assert [s["tool"] for s in sw1["agent"] if isinstance(s, dict) and "tool" in s] == [
|
|
"plan_task", "schedule_subagent", "schedule_subagent"]
|
|
assert sum(1 for s in sw1["agent"] if callable(s)) == 3 # wait_tasks + two child dispositions
|
|
|
|
|
|
def test_css_accent_helpers_only_touch_the_root_token():
|
|
css = ":root {\n --accent: #c93545;\n --accent-light: #f07a86;\n}\n.x { --accent: red; }\n"
|
|
assert scenarios.accent_value(css) == "#c93545"
|
|
changed = scenarios.css_with_accent(css, "#2f7de1")
|
|
assert scenarios.accent_value(changed) == "#2f7de1"
|
|
assert changed.count("#2f7de1") == 1 and "--accent-light: #f07a86" in changed and "--accent: red" in changed
|
|
|
|
|
|
def test_lane_count_and_stagger_bounds(monkeypatch):
|
|
_short_tmp(monkeypatch)
|
|
assert run_live_lanes.parse_args(["--stub"]).lanes == 4
|
|
assert run_live_lanes.parse_args(["--stub", "--lanes", "6"]).lanes == 6
|
|
assert run_live_lanes.parse_args(["--stub", "--stagger", "10"]).stagger == 3.0
|
|
assert run_live_lanes.parse_args(["--stub", "--stagger", "0.1"]).stagger == 2.0
|
|
assert run_live_lanes.parse_args(["--stub", "--stagger", "2.4"]).stagger == 2.4
|
|
for argv in (["--lanes", "7"], ["--lanes", "0"], ["--model", "x/y"], # --model: the stub IS the model
|
|
["--scenarios", "SM1,NOPE"], ["--attempts", "2", "--pass-of", "3"]):
|
|
with pytest.raises(SystemExit):
|
|
run_live_lanes.parse_args(["--stub", *argv])
|
|
args = run_live_lanes.parse_args(["--total-budget", "30"])
|
|
assert args.min_credit_usd == 30.0 and args.key_env == run_live_lanes.DEFAULT_KEY_ENV
|
|
assert args.seed == "HEAD" and args.source_repo == ""
|
|
|
|
|
|
def test_money_and_interval_arguments_must_be_finite_and_positive(monkeypatch):
|
|
"""A non-positive TOTAL_BUDGET means NO cap to the runtime and a non-positive tick is a hot
|
|
loop: both are argument-shaped refusals, before anything touches the world."""
|
|
_short_tmp(monkeypatch)
|
|
for argv in (["--total-budget", "0"], ["--total-budget", "-5"], ["--total-budget", "inf"],
|
|
["--total-budget", "nan"], ["--per-task-usd", "0"], ["--min-credit-usd", "0"],
|
|
["--min-credit-usd", "inf"], ["--watch-interval", "0"], ["--watch-interval", "-1"],
|
|
["--watch-interval", "nan"], ["--watch-interval", "1"], ["--seed", " "]):
|
|
with pytest.raises(SystemExit):
|
|
run_live_lanes.parse_args(["--stub", *argv])
|
|
ok = run_live_lanes.parse_args(["--stub", "--watch-interval", str(run_live_lanes.WATCH_INTERVAL_MIN_SEC)])
|
|
assert ok.watch_interval == run_live_lanes.WATCH_INTERVAL_MIN_SEC
|
|
|
|
|
|
def test_tmpdir_length_guard_refuses_fail_closed(monkeypatch):
|
|
monkeypatch.setattr(run_live_lanes.tempfile, "gettempdir", lambda: "/tmp/" + "x" * 80)
|
|
with pytest.raises(SystemExit):
|
|
run_live_lanes.parse_args(["--stub"])
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Effective settings: the tree's defaults, the budget knobs as settings keys, no env guesses
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
def test_budget_and_per_task_caps_are_written_into_the_applied_settings(monkeypatch):
|
|
_short_tmp(monkeypatch)
|
|
args = run_live_lanes.parse_args(["--total-budget", "30", "--per-task-usd", "8"])
|
|
cfg = run_live_lanes.effective_settings(args, FAKE_KEY)
|
|
assert cfg["TOTAL_BUDGET"] == 30.0 and cfg["OUROBOROS_PER_TASK_COST_USD"] == 8.0
|
|
assert cfg["OPENROUTER_API_KEY"] == FAKE_KEY and cfg["OUROBOROS_RUNTIME_MODE"] == "advanced"
|
|
from ouroboros.provider_models import declared_model_settings
|
|
for key, value in declared_model_settings({}).items():
|
|
assert cfg[key] == value # the defaults of the tree under test, written explicitly
|
|
pinned = run_live_lanes.effective_settings(run_live_lanes.parse_args(["--model", "argv/model-y"]), FAKE_KEY)
|
|
assert pinned["OUROBOROS_MODEL"] == "argv/model-y"
|
|
# The run-root copy is redacted: the key survives only in memory and in the lane files.
|
|
redacted = run_live_lanes.redacted_template(cfg)
|
|
assert "OPENROUTER_API_KEY" not in redacted and redacted["OUROBOROS_MODEL"] == cfg["OUROBOROS_MODEL"]
|
|
assert run_live_lanes.template_credentials(cfg) == {"OPENROUTER_API_KEY": FAKE_KEY}
|
|
|
|
|
|
def test_self_mod_is_off_by_default(monkeypatch):
|
|
_short_tmp(monkeypatch)
|
|
off = run_live_lanes.effective_settings(run_live_lanes.parse_args(["--stub"]), "")
|
|
assert off["OUROBOROS_POST_TASK_EVOLUTION"] == "false" and "OUROBOROS_POST_TASK_EVOLUTION_CADENCE" not in off
|
|
on = run_live_lanes.effective_settings(run_live_lanes.parse_args(["--stub", "--self-mod"]), "")
|
|
assert on["OUROBOROS_POST_TASK_EVOLUTION"] == "true" and on["OUROBOROS_POST_TASK_EVOLUTION_CADENCE"] == "every_n:1"
|
|
|
|
|
|
def test_preflight_worker_cap_reaches_every_lane_and_is_recorded(tmp_path, monkeypatch):
|
|
"""The commit gate's hermetic pytest pass runs INSIDE the lane server and resolves ``-n auto``
|
|
to the host CPU count (the 2026-09-04 paid run fanned out to >= 104 xdist workers per lane): the
|
|
stand must set the runtime's own lever to ``max(2, 16 // lanes)`` in the process every lane
|
|
server inherits, override an ambient value, and record the applied number in the manifest
|
|
and in each lane row. The runtime reads exactly that key (pinned here, not modified)."""
|
|
from ouroboros import preflight_runner
|
|
_short_tmp(monkeypatch)
|
|
assert run_live_lanes.PREFLIGHT_WORKERS_ENV == preflight_runner._PREFLIGHT_WORKERS_ENV
|
|
assert run_live_lanes.PREFLIGHT_WORKERS_FLOOR == preflight_runner._MIN_PREFLIGHT_WORKERS
|
|
assert run_live_lanes.parse_args(["--stub", "--lanes", "1"]).preflight_test_workers == 16
|
|
assert run_live_lanes.parse_args(["--stub", "--lanes", "4"]).preflight_test_workers == 4
|
|
assert run_live_lanes.parse_args(["--stub", "--lanes", "6"]).preflight_test_workers == 2 # floor at MAX_LANES
|
|
monkeypatch.setenv(run_live_lanes.PREFLIGHT_WORKERS_ENV, "128") # the operator shell must lose
|
|
seen: dict = {}
|
|
|
|
def lane(job, args, out, template, stagger, states, seed, budget=None, *, key="", seed_sha=""):
|
|
seen[job] = (os.environ.get(run_live_lanes.PREFLIGHT_WORKERS_ENV),
|
|
run_live_lanes._lane_row(job, args)["preflight_test_workers"],
|
|
preflight_runner._preflight_worker_count())
|
|
return _fake_lane(job, args, out, template, stagger, states, seed, budget, key=key, seed_sha=seed_sha)
|
|
|
|
monkeypatch.setattr(run_live_lanes, "run_lane", lane)
|
|
out = tmp_path / "out"
|
|
rc = run_live_lanes.main(["--stub", "--source-repo", str(_git_seed(tmp_path)), "--out", str(out),
|
|
"--scenarios", "SM1,SW1", "--lanes", "3", "--watch-interval", "600"])
|
|
assert rc == 0
|
|
assert seen == {("SM1", 1): ("5", 5, 5), ("SW1", 1): ("5", 5, 5)}
|
|
manifest = json.loads((out / "run_manifest.json").read_text(encoding="utf-8"))
|
|
assert manifest["extra"]["lanes"] == 3 and manifest["extra"]["preflight_test_workers"] == 5
|
|
|
|
|
|
def test_isolated_server_forwards_the_preflight_worker_cap_through_the_authoritative_sweep(tmp_path, monkeypatch):
|
|
"""The lane servers start in settings-authoritative mode, which strips the whole OUROBOROS_
|
|
namespace; the worker cap is the one operational lever that must survive, while an ambient
|
|
model slot still does not."""
|
|
from devtools.benchmarks.common.server_runner import _AUTHORITATIVE_ENV_KEEP, IsolatedServer
|
|
assert run_live_lanes.PREFLIGHT_WORKERS_ENV in _AUTHORITATIVE_ENV_KEEP
|
|
settings = tmp_path / "settings.json"
|
|
settings.write_text("{}", encoding="utf-8")
|
|
monkeypatch.setenv(run_live_lanes.PREFLIGHT_WORKERS_ENV, "4")
|
|
monkeypatch.setenv("OUROBOROS_MODEL", "ambient/model")
|
|
env = IsolatedServer(tmp_path / "clone", tmp_path / "data", settings, settings_authoritative_env=True)._env()
|
|
assert env[run_live_lanes.PREFLIGHT_WORKERS_ENV] == "4" and "OUROBOROS_MODEL" not in env
|
|
|
|
|
|
def test_stub_template_carries_only_the_loopback_slots(monkeypatch):
|
|
_short_tmp(monkeypatch)
|
|
cfg = run_live_lanes.effective_settings(run_live_lanes.parse_args(["--stub"]), "")
|
|
assert cfg["OUROBOROS_MODEL"] == stub_lane.STUB_MODEL_SLUG == cfg["OUROBOROS_MODEL_LIGHT"]
|
|
assert not any(k.startswith("OUROBOROS_MODEL") and v and v != stub_lane.STUB_MODEL_SLUG for k, v in cfg.items())
|
|
assert "OPENROUTER_API_KEY" not in cfg
|
|
|
|
|
|
def test_config_sha256_is_secret_free_and_key_independent():
|
|
base = {"OUROBOROS_MODEL": "m", "TOTAL_BUDGET": 1.0}
|
|
a = run_live_lanes.config_sha256({**base, "OPENROUTER_API_KEY": "key-one"})
|
|
b = run_live_lanes.config_sha256({**base, "OPENROUTER_API_KEY": "key-two"})
|
|
assert a != b # a different key is a different (fingerprinted) config...
|
|
assert a == run_live_lanes.config_sha256({**base, "OPENROUTER_API_KEY": "key-one"})
|
|
assert a != run_live_lanes.config_sha256({**base, "OPENROUTER_API_KEY": "key-one", "TOTAL_BUDGET": 2.0})
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# The run-wide budget ledger
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
def test_lane_spend_sums_the_settled_product_ledger_and_counts_unknown_costs(tmp_path):
|
|
"""rc.15 run3: telemetry summed 114.81, the product ledger 141.63 (skill review, advisory, synthesis write no row)."""
|
|
(state := tmp_path / "data" / "state").mkdir(parents=True)
|
|
rows = [{"state": "settled", "cost_final": True, "cost_usd": 1.5}, {"state": "settled", "cost_final": True, "cost_usd": 0.25},
|
|
{"state": "settled", "cost_final": True, "cost_usd": None}, {"state": "settled", "cost_final": False, "cost_usd": 99.0},
|
|
{"state": "pending", "cost_usd": 99.0}, {"state": "settled", "cost_final": True, "cost_usd": True}, "not json"]
|
|
(state / "usage_attempts.jsonl").write_text("\n".join(r if isinstance(r, str) else json.dumps(r) for r in rows) + "\n", encoding="utf-8")
|
|
assert run_live_lanes.lane_spend(tmp_path / "data") == (1.75, 2)
|
|
assert run_live_lanes.lane_spend(tmp_path / "absent") == (0.0, 0)
|
|
|
|
|
|
def _ask(budget, job, root_tasks, root, waits: list | None = None, *, index: int = 0):
|
|
"""``admit`` on its own thread (it may block): ``(thread, box)``; ``box["r"]`` is the answer."""
|
|
box: dict = {}
|
|
thread = threading.Thread(target=lambda: box.__setitem__("r", budget.admit(
|
|
job, root_tasks, root, dispatch_index=index, on_wait=waits.append if waits is not None else None)), daemon=True)
|
|
thread.start()
|
|
thread.join(0.3)
|
|
return thread, box
|
|
|
|
|
|
def test_run_budget_waits_on_in_flight_reservations_and_refuses_only_what_can_never_fit(tmp_path):
|
|
"""Per attempt: spent (durable, re-read) + reservation > cap -> refused, no run-wide halt; fits the cap but not
|
|
the reservations in flight -> waits and re-asks after EVERY settle (the first paid run wrote SW1/SK1 off at t=+21
|
|
min behind two SM1 reservations still in flight); spent only grows, so a waiter can end refused with its wait
|
|
recorded. A lane's TOTAL_BUDGET is its OWN reservation: the ceilings in flight are disjoint and settled spend +
|
|
in-flight ceilings never exceeds the cap (the first draft handed each lane cap - others' reservations). The
|
|
reservation unit is per-task x root tasks: $8 per task reserves $8 per root, $16 for SK1's two."""
|
|
spend = {}
|
|
budget = run_live_lanes.RunBudget(20.0, 8.0, reader=lambda root: (spend.get(root.name, 0.0), 0))
|
|
assert budget.reservation(1) == 8.0 and budget.reservation(2) == 16.0 and budget.reservation(0) == 8.0
|
|
ok, facts = budget.admit(("SM1", 1), 1, tmp_path / "a", dispatch_index=0)
|
|
assert ok and facts == {"cap_usd": 20.0, "spent_usd": 0.0, "reserved_usd": 0.0, "reservation_usd": 8.0,
|
|
"unknown_cost_rows": 0, "waited_sec": 0.0}
|
|
assert budget.ceiling(("SM1", 1)) == 8.0 # its own reservation, never the whole cap
|
|
ok, facts = budget.admit(("SW1", 1), 1, tmp_path / "b", dispatch_index=1)
|
|
assert ok and facts["reserved_usd"] == 8.0
|
|
assert budget.ceiling(("SW1", 1)) == 8.0 # disjoint from lane a: 8 + 8 + spent 0 <= cap 20
|
|
assert budget.ceiling(("SM1", 1)) + budget.ceiling(("SW1", 1)) <= 20.0
|
|
spend["a"] = 5.0 # lane a spends while in flight: visible now
|
|
ok, facts = budget.admit(("SK1", 1), 2, tmp_path / "c", dispatch_index=2) # 5 + 16 > 20: can NEVER fit -> refused at once
|
|
assert not ok and facts["spent_usd"] == 5.0 and facts["waited_sec"] == 0.0 and budget.not_run == ["SK1_a1"]
|
|
waits: list = []
|
|
thread, box = _ask(budget, ("SM1", 2), 1, tmp_path / "d", waits, index=3) # 5 + 8 <= 20 but 5 + 16 + 8 > 20: waits
|
|
assert thread.is_alive() and budget.not_run == ["SK1_a1"] # not refused: the blocker is in flight
|
|
assert waits == ["waiting — in flight reserved $16.00, needs $8.00, spent $5.00, cap $20.00"]
|
|
budget.settle(("SM1", 1)) # 5 + 8 + 8 > 20: re-asked, still waiting
|
|
thread.join(0.3)
|
|
assert thread.is_alive() and waits[1:] == [] # told once per wait, not per wake-up
|
|
budget.settle(("SW1", 1)) # 5 + 0 + 8: admitted after the wait
|
|
thread.join(5.0)
|
|
assert not thread.is_alive() and box["r"][0] and box["r"][1]["reserved_usd"] == 0.0 and box["r"][1]["waited_sec"] > 0
|
|
thread, box = _ask(budget, ("SM1", 3), 1, tmp_path / "e", index=4) # 5 + 8 <= 20 but 5 + 8 + 8 > 20: waits
|
|
assert thread.is_alive()
|
|
spend["d"] = 10.0 # the lane in flight overruns: spent 15 on the next question
|
|
budget.settle(("SM1", 2)) # 15 + 8 > 20: refused AFTER the wait
|
|
thread.join(5.0)
|
|
assert not thread.is_alive() and not box["r"][0] and box["r"][1]["spent_usd"] == 15.0 and box["r"][1]["waited_sec"] > 0
|
|
snap = budget.snapshot()
|
|
assert snap["spent_usd"] == 15.0 and snap["reserved_usd"] == 0.0 and snap["lanes_settled"] == 3
|
|
assert snap["attempts_not_run"] == ["SK1_a1", "SM1_a3"] and snap["first_refused"] == "SK1_a1" and "halted" not in snap
|
|
assert [(r["attempt"], r["reason"], r["reservation_usd"]) for r in snap["refusals"]] == [
|
|
("SK1_a1", "budget_cap", 16.0), ("SM1_a3", "budget_cap", 8.0)]
|
|
assert snap["refusals"][0]["waited_sec"] == 0.0 and snap["refusals"][1]["waited_sec"] > 0
|
|
assert snap["reservation_rule"] == run_live_lanes.RESERVATION_RULE
|
|
# The ceiling ignores what OTHER lanes spend (it is this lane's reservation), and the floor
|
|
# keeps it positive (the runtime reads a non-positive TOTAL_BUDGET as NO cap).
|
|
tiny = run_live_lanes.RunBudget(10.0, 8.0, reader=lambda root: (20.0, 0))
|
|
assert tiny.admit(("SM1", 1), 1, tmp_path / "x", dispatch_index=0)[0]
|
|
assert tiny.ceiling(("SM1", 1)) == 8.0
|
|
assert tiny.ceiling(("never", 9)) == run_live_lanes.LANE_BUDGET_FLOOR_USD # not admitted: the floor, not the cap
|
|
# The floor is part of the ONE effective ceiling: admission reserves it, the lane receives it,
|
|
# so micro reservations cannot sum past the cap (5 x 0.01 fit a 0.05 cap, the 6th waits on them).
|
|
micro = run_live_lanes.RunBudget(0.05, 0.001, reader=lambda root: (0.0, 0))
|
|
assert micro.reservation(1) == run_live_lanes.LANE_BUDGET_FLOOR_USD
|
|
for n in range(5):
|
|
ok, facts = micro.admit(("SM1", n), 1, tmp_path / f"m{n}", dispatch_index=n)
|
|
assert ok and facts["reservation_usd"] == 0.01 and micro.ceiling(("SM1", n)) == 0.01
|
|
sixth, box = _ask(micro, ("SM1", 5), 1, tmp_path / "m5", index=5)
|
|
assert sixth.is_alive() and micro.not_run == []
|
|
assert sum(micro.ceiling(("SM1", n)) for n in range(5)) <= 0.05
|
|
for n in range(5):
|
|
micro.settle(("SM1", n))
|
|
sixth.join(5.0)
|
|
assert not sixth.is_alive() and box["r"][0] # admitted once the five settled at $0
|
|
below = run_live_lanes.RunBudget(0.005, 0.001, reader=lambda root: (0.0, 0))
|
|
assert not below.admit(("SM1", 1), 1, tmp_path / "z", dispatch_index=0)[0] # the floored reservation exceeds the cap
|
|
# Fractional reservations are never rounded upward (round(0.01006, 4) would hand out 0.0101):
|
|
# two exact 0.01006 reservations fill a 0.02012 cap and each lane receives exactly 0.01006.
|
|
frac = run_live_lanes.RunBudget(0.02012, 0.01006, reader=lambda root: (0.0, 0))
|
|
assert frac.admit(("SM1", 1), 1, tmp_path / "f1", dispatch_index=0)[0] and frac.admit(("SM1", 2), 1, tmp_path / "f2", dispatch_index=1)[0]
|
|
assert frac.ceiling(("SM1", 1)) == 0.01006 and frac.ceiling(("SM1", 2)) == 0.01006
|
|
assert frac.ceiling(("SM1", 1)) + frac.ceiling(("SM1", 2)) <= 0.02012
|
|
third, box = _ask(frac, ("SM1", 3), 1, tmp_path / "f3", index=2)
|
|
assert third.is_alive() # full: waits, not refused
|
|
frac.settle(("SM1", 1))
|
|
frac.settle(("SM1", 2))
|
|
third.join(5.0)
|
|
assert not third.is_alive() and box["r"][0]
|
|
|
|
|
|
def test_admission_is_fifo_by_dispatch_index_and_a_refused_head_frees_the_line(tmp_path):
|
|
"""A later-dispatched attempt that WOULD fit waits while an earlier one is still asking (the freed lane's next
|
|
job can no longer leapfrog the woken waiter); a head that can never fit is refused and leaves the line. cap 30 /
|
|
per-task 8: SK1 #0 (16) in flight; SK1 #1 (16) waits on it (32 > 30); SM1 #2 (8) would fit (24 <= 30) but waits
|
|
behind #1, its wait naming that; lane #0 spends 15 and settles: #1 refused (31 > 30), #2 admitted at reserved $0."""
|
|
spend: dict = {}
|
|
budget = run_live_lanes.RunBudget(30.0, 8.0, reader=lambda root: (spend.get(root.name, 0.0), 0))
|
|
assert budget.admit(("SK1", 1), 2, tmp_path / "a", dispatch_index=0)[0]
|
|
head_waits, later_waits = [], []
|
|
head, head_box = _ask(budget, ("SK1", 2), 2, tmp_path / "b", head_waits, index=1)
|
|
later, later_box = _ask(budget, ("SM1", 1), 1, tmp_path / "c", later_waits, index=2)
|
|
assert head.is_alive() and later.is_alive() and budget.not_run == []
|
|
assert head_waits == ["waiting — in flight reserved $16.00, needs $16.00, spent $0.00, cap $30.00"]
|
|
assert later_waits == ["waiting — behind SK1_a2 in dispatch order, needs $8.00, spent $0.00, cap $30.00"]
|
|
spend["a"] = 15.0
|
|
budget.settle(("SK1", 1))
|
|
head.join(5.0)
|
|
later.join(5.0)
|
|
assert not head.is_alive() and not head_box["r"][0] and head_box["r"][1]["spent_usd"] == 15.0
|
|
assert not later.is_alive() and later_box["r"][0] and later_box["r"][1]["reserved_usd"] == 0.0
|
|
assert budget.not_run == ["SK1_a2"] and budget.ceiling(("SM1", 1)) == 8.0 and later_box["r"][1]["waited_sec"] > 0
|
|
|
|
|
|
def test_reservation_counts_roots_plus_the_evolution_root_and_is_the_lane_total_budget(tmp_path, monkeypatch):
|
|
"""EQUALITY pins of the rc.14/rc.15 finding: the reservation is per-task x root tasks, +1 with --self-mod (the one
|
|
post-task cycle; rc.14: SM1_a1 task $3.84 + cycles $12.40 + $2.84 of $20 — the lane's TOTAL_BUDGET is the fence); the 2x
|
|
factor and its product import are gone and no bench budget profile is projected. Per-task $20 and one root reserve
|
|
$20 ($40 for the absorbing SM1 root with --self-mod; SK1's two roots stay $40, it does not promote) and that exact number reaches the lane's settings file as
|
|
TOTAL_BUDGET through ``run_lane`` (never the run-wide cap)."""
|
|
_short_tmp(monkeypatch)
|
|
rule = run_live_lanes.RESERVATION_RULE
|
|
assert not hasattr(run_live_lanes, "HARD_STOP_INVERSE") and rule == run_live_lanes.RunBudget(1, 1).snapshot()["reservation_rule"]
|
|
assert rule.startswith("max(0.01, per_task_usd x (root_tasks + 1 if --self-mod and the scenario absorbs else root_tasks))")
|
|
assert "post-task cycle of a lane that promotes" in rule and "the true fence" in rule and "cost_hard_stop" not in rule
|
|
budget = run_live_lanes.RunBudget(100.0, 20.0, reader=lambda root: (0.0, 0))
|
|
assert budget.reservation(1) == 20.0 and budget.reservation(2) == 40.0 and not budget.self_mod
|
|
evolving = run_live_lanes.RunBudget(100.0, 20.0, reader=lambda root: (0.0, 0), self_mod=True)
|
|
assert evolving.reservation(1, absorbs=True) == 40.0 and evolving.reservation(2) == 40.0 and evolving.reservation(1) == 20.0
|
|
seed = _git_seed(tmp_path)
|
|
out, job = tmp_path / "out", ("SM1", 1)
|
|
ok, facts = budget.admit(job, 1, out / "lanes" / "SM1_a1" / "data", dispatch_index=0)
|
|
assert ok and facts["reservation_usd"] == 20.0 and budget.ceiling(job) == 20.0
|
|
|
|
class _NoServer(_NoopServer): # the real path up to the written settings, then stop
|
|
def start(self, **_k) -> None:
|
|
raise RuntimeError("no server in this pin: the settings file on disk is the evidence")
|
|
|
|
monkeypatch.setattr(run_live_lanes, "IsolatedServer", _NoServer)
|
|
args = run_live_lanes.parse_args(["--per-task-usd", "20", "--total-budget", "100", "--scenarios", "SM1",
|
|
"--out", str(out), "--watch-interval", "600"])
|
|
template = run_live_lanes.effective_settings(args, FAKE_KEY)
|
|
assert template["TOTAL_BUDGET"] == 100.0 # the run cap; every lane rewrites it with its ceiling
|
|
row = run_live_lanes.run_lane(job, args, out, template, run_live_lanes.Stagger(2.0), {}, seed, budget,
|
|
key=FAKE_KEY, seed_sha=run_live_lanes.head_sha(seed))
|
|
applied = json.loads((out / "lanes" / "SM1_a1" / "data" / "settings.json").read_text(encoding="utf-8"))
|
|
assert applied["TOTAL_BUDGET"] == 20.0 == budget.ceiling(job) == budget.reservation(1)
|
|
assert applied["OUROBOROS_PER_TASK_COST_USD"] == 20.0 and applied["OPENROUTER_API_KEY"] == FAKE_KEY
|
|
assert row["budget"] == {"reservation_usd": 20.0, "lane_total_budget_usd": 20.0, "per_task_usd": 20.0,
|
|
"spent_usd": 0.0, "unknown_cost_rows": 0}
|
|
assert row["status"] == "infra_error" and row["refusal"]["type"] == "RuntimeError"
|
|
|
|
|
|
def test_submit_injects_no_budget_profile_into_the_stand_roots(monkeypatch):
|
|
"""Every stand root runs under the PRODUCT'S default in-task ceiling: no ``metadata.budget_profile``
|
|
(a bench profile would change the pacing path under test; with a reservation >= 2 x per-task the
|
|
per-task axis binds first). Only the stand's identity and the scenario's own metadata are sent."""
|
|
bodies: list = []
|
|
monkeypatch.setattr(scenarios, "_api", lambda base, method, path, payload=None, timeout=0:
|
|
bodies.append((method, path, payload)) or {"task_id": "t-1"})
|
|
assert _ctx().submit("do it", metadata={"force_plan": True, "force_plan_source": "swarm"}) == "t-1"
|
|
(method, path, body), = bodies
|
|
assert (method, path) == ("POST", "/api/tasks") and body["timeout_sec"] == 1
|
|
assert body["metadata"] == {"source": "e2e_live", "delegation_role": "root", "force_plan": True, "force_plan_source": "swarm"}
|
|
assert not hasattr(scenarios, "STAND_BUDGET_PROFILE") and "budget_profile" not in json.dumps(body)
|
|
|
|
|
|
def test_budget_preflight_refuses_reservations_that_can_never_all_be_admitted(tmp_path, monkeypatch):
|
|
"""The rc.15 plan under the 2x rule (cap 200, SK1 reserving the whole cap, attempts 3) would have burned SM1/SW1 and
|
|
refused every SK1 attempt by construction. The preflight refuses BEFORE any spend a reservation above the cap, or
|
|
equal to it with attempts >= 2 (the second can never be admitted after any spend), in the credit preflight's typed
|
|
shape, before the key, the seed or a lane; no override. The per-ROUND worst case is the --lanes largest reservations,
|
|
ONE attempt per scenario: the owner's cap 300 / per-task 50 / --self-mod / 3 lanes = $250 (SM1 100, SK1 100, SW1 50)."""
|
|
def rows(budget, attempts, ids=("SM1", "SW1", "SK1"), lanes=3):
|
|
pre = run_live_lanes.budget_preflight(budget, list(ids), attempts, lanes)
|
|
return ({r["scenario"]: r["reservation_usd"] for r in pre["scenarios"]}, pre["worst_case_usd"], pre["unreachable"],
|
|
pre["round_worst_case_usd"])
|
|
|
|
reader = lambda root: (0.0, 0) # noqa: E731 - a stub reader
|
|
assert rows(run_live_lanes.RunBudget(300.0, 50.0, reader, self_mod=True), 3) == ({"SM1": 100.0, "SW1": 50.0, "SK1": 100.0}, 750.0, [], 250.0)
|
|
assert rows(run_live_lanes.RunBudget(300.0, 50.0, reader, self_mod=True), 3, lanes=2)[3] == 200.0
|
|
assert rows(run_live_lanes.RunBudget(90.0, 50.0, reader, self_mod=True), 1) == ({"SM1": 100.0, "SW1": 50.0, "SK1": 100.0}, 250.0, ["SM1", "SK1"], 250.0)
|
|
at_cap = run_live_lanes.RunBudget(100.0, 50.0, reader, self_mod=True) # SM1/SK1 == cap: one attempt fits at $0, never a second
|
|
assert rows(at_cap, 1)[2] == [] and rows(at_cap, 2)[2] == ["SM1", "SK1"]
|
|
assert rows(run_live_lanes.RunBudget(200.0, 100.0, reader), 3)[2] == ["SK1"] # the shipped 2x rule's SK1 = 200 of 200
|
|
pre = run_live_lanes.budget_preflight(run_live_lanes.RunBudget(200.0, 50.0, reader), ["SK1"], 2, 4)
|
|
assert pre == {"cap_usd": 200.0, "per_task_usd": 50.0, "self_mod": False, "reservation_rule": run_live_lanes.RESERVATION_RULE,
|
|
"scenarios": [{"scenario": "SK1", "root_tasks": 2, "reservation_usd": 100.0, "attempts": 2,
|
|
"worst_case_usd": 200.0, "unreachable": False}], "worst_case_usd": 200.0, "lanes": 4,
|
|
"round_worst_case_usd": 100.0, "unreachable": []}
|
|
out, manifest = _fake_run(tmp_path, monkeypatch, ["--total-budget", "90", "--per-task-usd", "50", "--self-mod",
|
|
"--scenarios", "SM1,SW1,SK1"],
|
|
lane=lambda *a, **k: pytest.fail("a lane started after a budget refusal"), expect_rc=3)
|
|
refusal = manifest["extra"]["refusal"]
|
|
assert refusal["stage"] == "budget_preflight" and refusal["reason"] == "reservation_unreachable"
|
|
assert refusal["unreachable"] == ["SM1", "SK1"] and refusal["cap_usd"] == 90.0 and refusal["self_mod"] is True
|
|
assert manifest["extra"]["budget_preflight"]["unreachable"] == ["SM1", "SK1"] and manifest["extra"]["exit_code"] == 3
|
|
assert "credential_fingerprint" not in manifest["extra"] and not (out / "seed").exists() and not (out / "lanes").exists()
|
|
assert manifest["requested_task_ids"] == ["SM1_a1", "SW1_a1", "SK1_a1"] # SM1/SW1 = $100 = cap: one attempt fits
|
|
|
|
|
|
def test_jobs_are_dispatched_round_robin_by_attempt_largest_reservation_first_within_a_round(tmp_path, monkeypatch):
|
|
"""``dispatch_order``: a1 of every scenario, then a2 (the verdict is pass-of PER scenario: the order protects the
|
|
MINIMUM admitted per scenario); within a round SK1 (two roots) asks before SM1 and SW1 (stable among equals), and
|
|
admission keeps that order (FIFO by index). Requested ids keep the argument order; the per-round worst case is recorded."""
|
|
order: list = []
|
|
|
|
def lane(job, *a, **k):
|
|
order.append(f"{job[0]}_a{job[1]}")
|
|
return _fake_lane(job, *a, **k)
|
|
|
|
_out, manifest = _fake_run(tmp_path, monkeypatch, ["--total-budget", "200", "--per-task-usd", "50", "--lanes", "1",
|
|
"--scenarios", "SM1,SW1,SK1", "--attempts", "2"], lane=lane)
|
|
assert order == ["SK1_a1", "SM1_a1", "SW1_a1", "SK1_a2", "SM1_a2", "SW1_a2"]
|
|
assert manifest["requested_task_ids"] == ["SM1_a1", "SM1_a2", "SW1_a1", "SW1_a2", "SK1_a1", "SK1_a2"]
|
|
assert manifest["extra"]["budget_preflight"]["unreachable"] == [] and manifest["extra"]["outcome"] == "completed"
|
|
assert manifest["extra"]["budget_preflight"]["worst_case_usd"] == 400.0
|
|
assert manifest["extra"]["budget_preflight"]["lanes"] == 1 and manifest["extra"]["budget_preflight"]["round_worst_case_usd"] == 100.0
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Feasibility pins with POSITIVE spends: the audit's driver over the REAL ledger
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
class _Driver:
|
|
"""``main()``'s pool replaced by a virtual clock over the REAL ``RunBudget``: ``admit``/``settle`` as ``run_attempt``
|
|
makes them (indices from ``dispatch_order``), spend visible at settle, lane durations in virtual minutes (rc.14 SM1
|
|
22-54, rc.11 SW1 ~10, SK1 ~7). The ledger's ``wait`` is a park the driver releases one thread at a time, so the
|
|
schedule is the runner's OBSERVED one, never the OS's: after a settle the freed lane's next job asks FIRST (it wins
|
|
the lock on CPython, 300/300), then the parked attempts re-ask in dispatch order; a refusal frees its lane at once."""
|
|
DURATION = {"SM1": 50, "SW1": 10, "SK1": 7}
|
|
|
|
def __init__(self, cap, per_task, scenario_ids, attempts, spends, *, lanes=3, self_mod=False) -> None:
|
|
self.spend: dict = {}
|
|
self.budget = run_live_lanes.RunBudget(cap, per_task, reader=lambda root: (self.spend.get(root.name, 0.0), 0),
|
|
self_mod=self_mod)
|
|
requested = [(sid, n) for sid in scenario_ids for n in range(1, attempts + 1)]
|
|
self.pending = collections.deque(enumerate(run_live_lanes.dispatch_order(self.budget, requested)))
|
|
self.lanes, self.spends, self.now = lanes, spends, 0.0
|
|
self.in_flight, self.parked, self.threads, self.admitted, self.refused = {}, {}, {}, [], []
|
|
lock = self.budget._lock
|
|
|
|
def park(*_a, **_k) -> None: # the ledger's wait: release the lock, hold until the driver wakes this thread
|
|
gate = self.parked[threading.current_thread().name] = threading.Event()
|
|
lock.release()
|
|
gate.wait()
|
|
lock.acquire()
|
|
|
|
lock.wait = park
|
|
|
|
def _ask(self, index: int, job) -> None:
|
|
name, box, row = f"{job[0]}_a{job[1]}", {}, scenarios.SCENARIOS[job[0]]
|
|
thread = threading.Thread(name=name, daemon=True, target=lambda: box.__setitem__("r", self.budget.admit(
|
|
job, row.root_tasks, pathlib.Path("/x") / name, dispatch_index=index, absorbs=row.expects_absorb)))
|
|
self.threads[name] = (index, job, thread, box)
|
|
thread.start()
|
|
self._settle_thread(name)
|
|
|
|
def _settle_thread(self, name: str) -> None:
|
|
"""Spin until the thread has answered or parked; the deadline is a hang guard, never a timing assumption."""
|
|
_index, job, thread, box = self.threads[name]
|
|
deadline = time.monotonic() + 10.0
|
|
while thread.is_alive() and name not in self.parked:
|
|
assert time.monotonic() < deadline, f"{name} neither answered nor parked"
|
|
time.sleep(0.0005)
|
|
if not thread.is_alive(): # answered: admitted (in flight now) or refused
|
|
del self.threads[name]
|
|
(self.admitted if box["r"][0] else self.refused).append(name)
|
|
if box["r"][0]:
|
|
self.in_flight[job] = self.now + self.DURATION[job[0]]
|
|
|
|
def run(self) -> tuple[list, list, float]: # (admitted in admission order, refused in refusal order, spend)
|
|
while self.pending or self.in_flight:
|
|
while self.pending and len(self.in_flight) + len(self.parked) < self.lanes:
|
|
self._ask(*self.pending.popleft())
|
|
assert self.in_flight, "parked attempts with nothing in flight (the ledger contract forbids it)"
|
|
job = min(self.in_flight, key=lambda j: (self.in_flight[j], j))
|
|
self.now = self.in_flight.pop(job)
|
|
self.spend[f"{job[0]}_a{job[1]}"] = self.spends[job[0]]
|
|
self.budget.settle(job)
|
|
if self.pending: # the freed lane's next job asks before the line re-asks
|
|
self._ask(*self.pending.popleft())
|
|
for name in sorted(self.parked, key=lambda n: self.threads[n][0]): # then the line, earliest first
|
|
self.parked.pop(name).set()
|
|
self._settle_thread(name)
|
|
return self.admitted, self.refused, round(self.budget.snapshot()["spent_usd"], 2)
|
|
|
|
|
|
REALISTIC_SPEND = {"SM1": 30.0, "SW1": 8.0, "SK1": 15.0} # assumed per-attempt spends: rc.14 SM1 lanes, rc.11 SW1/SK1
|
|
PESSIMISTIC_SPEND = {"SM1": 45.0, "SW1": 8.0, "SK1": 30.0}
|
|
OWNER_CONFIGURATION = dict(cap=300.0, per_task=50.0, scenario_ids=["SM1", "SW1", "SK1"], attempts=3, lanes=3, self_mod=True)
|
|
DISPATCH_ORDER = ["SM1_a1", "SK1_a1", "SW1_a1", "SM1_a2", "SK1_a2", "SW1_a2", "SM1_a3", "SK1_a3", "SW1_a3"]
|
|
|
|
|
|
def test_owner_configuration_cap_300_per_task_50_three_attempts_self_mod_is_exact_under_fifo_admission():
|
|
"""The live configuration (cap 300, per-task 50, attempts 3, pass-of 2, 3 lanes, --self-mod: SM1 reserves 100 — its
|
|
root plus the post-task cycle only it promotes — SK1 100 for two roots, SW1 50; round 1 = 250 fits) — EXACT
|
|
sequences, no wake-order range. Realistic spends: all nine admitted in dispatch order, $159. Pessimistic: SK1_a3
|
|
refused ($219 + 100 > 300): 8/9 at $219, every scenario keeping two = pass-of. Largest-first dispatch under the
|
|
earlier +1-for-every-lane rule refused all of SW1 ($225, 0/3): the handbook's traced reason, prose, not a pin."""
|
|
assert _Driver(spends=REALISTIC_SPEND, **OWNER_CONFIGURATION).run() == (DISPATCH_ORDER, [], 159.0)
|
|
admitted, refused, spent = _Driver(spends=PESSIMISTIC_SPEND, **OWNER_CONFIGURATION).run()
|
|
assert (admitted, refused, spent) == (DISPATCH_ORDER[:7] + ["SW1_a3"], ["SK1_a3"], 219.0)
|
|
assert {s: sum(n.startswith(s) for n in admitted) for s in ("SM1", "SW1", "SK1")} == {"SM1": 3, "SW1": 3, "SK1": 2}
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# The watcher's key probe: informational, bounded, backing off, never on the tick's path
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
def test_key_probe_failures_are_informational_and_back_off():
|
|
stop = threading.Event()
|
|
calls = {"n": 0}
|
|
|
|
def flaky() -> float | None:
|
|
calls["n"] += 1
|
|
if calls["n"] <= 2:
|
|
raise TimeoutError("timed out")
|
|
return 3.0
|
|
|
|
probe = run_live_lanes.KeyProbe(flaky, floor=5.0, interval=30.0, stop=stop)
|
|
assert probe.interval == run_live_lanes.PROBE_MIN_INTERVAL_SEC # never more often than the floor
|
|
assert probe.fragment() == "key probe pending"
|
|
probe.poll_once()
|
|
assert probe.failures == 1 and "ALERT" not in probe.fragment()
|
|
assert probe.fragment().startswith("key probe failed: TimeoutError") and "informational" in probe.fragment()
|
|
assert probe.next_wait() == 2 * run_live_lanes.PROBE_MIN_INTERVAL_SEC
|
|
probe.poll_once()
|
|
assert probe.failures == 2 and probe.next_wait() == 4 * run_live_lanes.PROBE_MIN_INTERVAL_SEC
|
|
probe.failures = 10
|
|
assert probe.next_wait() == run_live_lanes.PROBE_BACKOFF_MAX_SEC
|
|
probe.poll_once() # a good reading resets the back-off
|
|
assert probe.failures == 0 and probe.next_wait() == run_live_lanes.PROBE_MIN_INTERVAL_SEC
|
|
assert probe.fragment() == "key remaining $3.00 ALERT" # ALERT only on a GOOD reading under the floor
|
|
probe.seed(None)
|
|
assert probe.fragment() == "key uncapped"
|
|
|
|
|
|
def test_watcher_tick_never_waits_on_the_key_probe(capsys, monkeypatch):
|
|
"""A probe stuck in a provider call must not delay the tick: the watcher reads the probe's
|
|
last fragment and prints the ledger's spend regardless."""
|
|
monkeypatch.setattr(run_live_lanes.shutil, "disk_usage", lambda _mount: types.SimpleNamespace(free=100 * 2**30))
|
|
stop = threading.Event()
|
|
|
|
def stuck() -> float | None:
|
|
stop.wait(10)
|
|
return None
|
|
|
|
probe = run_live_lanes.KeyProbe(stuck, floor=1.0, interval=30.0, stop=stop)
|
|
probe.interval = 0.01
|
|
probe.start()
|
|
budget = run_live_lanes.RunBudget(50.0, 16.0, reader=lambda root: (2.5, 0))
|
|
budget.admit(("SM1", 1), 1, pathlib.Path("/nonexistent/lane/data"), dispatch_index=0)
|
|
states = {("SM1", 1): ("running scenario", time.time())}
|
|
thread = threading.Thread(target=run_live_lanes.watcher, args=(stop, states, 0.05, budget, probe), daemon=True)
|
|
thread.start()
|
|
seen = ""
|
|
deadline = time.time() + 5
|
|
while "[watch]" not in seen and time.time() < deadline:
|
|
time.sleep(0.05)
|
|
seen += capsys.readouterr().out
|
|
stop.set()
|
|
thread.join(timeout=5)
|
|
line = next(ln for ln in seen.splitlines() if "[watch]" in ln)
|
|
assert "spent $2.50/$50.00 reserved $16.00" in line and "SM1_a1=running scenario" in line # $16 per task, one root
|
|
# Disk alerts are independent of the key probe's pending state.
|
|
assert line.endswith(" | key probe pending")
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Self-modification: a confirmed absorb, never an assumed one
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
def _campaign(data_root: pathlib.Path, cycles: int, tx: dict | None = None) -> None:
|
|
(data_root / "state").mkdir(parents=True, exist_ok=True)
|
|
(data_root / "state" / "evolution_campaign.json").write_text(json.dumps(
|
|
{"absorbed_cycles_done": cycles, "transaction_history": [tx] if tx else []}), encoding="utf-8")
|
|
|
|
|
|
def test_confirm_absorb_requires_positive_evidence(tmp_path, monkeypatch):
|
|
clone = _git_seed(tmp_path)
|
|
(clone / "f").write_text("1\n", encoding="utf-8")
|
|
first = _commit(clone, "one")
|
|
data_root = tmp_path / "data"
|
|
_campaign(data_root, 0)
|
|
state = {"sha": first[:8], "uptime": 100}
|
|
monkeypatch.setattr(run_live_lanes, "_api", lambda base, method, path, payload=None, timeout=0: dict(state))
|
|
pre = run_live_lanes.self_mod_snapshot(_FakeServer(), clone, data_root)
|
|
assert pre["head"] == first and pre["sha"] == first[:8] and pre["cycles"] == 0 and pre["state_read"]
|
|
# No promotion: the runtime declined, and a liveness check alone would have said PASS.
|
|
out = run_live_lanes.confirm_absorb(_FakeServer(absorb={"absorbed": False, "reason": "no_promotion"}), clone,
|
|
data_root, pre, timeout=1, ready_timeout=1)
|
|
assert out["confirmed"] is False and out["reason"] == "no_promotion" and out["head_moved"] is False
|
|
# The wait said absorbed and the counter advanced, but the served uptime never reset: not restarted.
|
|
(clone / "f").write_text("2\n", encoding="utf-8")
|
|
second = _commit(clone, "two")
|
|
_campaign(data_root, 1, {"commit_sha": second, "cycle_outcome": "absorbed", "restart_verified": True,
|
|
"verified_by": "boot_reconciliation"})
|
|
state.update({"sha": second[:8], "uptime": 100})
|
|
out = run_live_lanes.confirm_absorb(_FakeServer(absorb={"absorbed": True, "reason": "absorbed"}), clone,
|
|
data_root, pre, timeout=1, ready_timeout=1)
|
|
assert out["confirmed"] is False and out["reason"] == "not_restarted" and out["head_moved"] is True
|
|
assert out["transaction"] == {"commit_sha": second, "cycle_outcome": "absorbed", "restart_verified": True,
|
|
"verified_by": "boot_reconciliation"}
|
|
# Every fact present: counter advanced, sha moved, uptime reset, healthy, serving the clone HEAD.
|
|
state["uptime"] = 0
|
|
out = run_live_lanes.confirm_absorb(_FakeServer(absorb={"absorbed": True, "reason": "absorbed"}), clone,
|
|
data_root, pre, timeout=1, ready_timeout=1)
|
|
assert out["confirmed"] is True and out["reason"] == "absorbed" and out["serving_head"] is True
|
|
assert out["post"]["cycles"] == 1 and out["post"]["head"] == second
|
|
unhealthy = run_live_lanes.confirm_absorb(_FakeServer(absorb={"absorbed": True, "reason": "absorbed"}, healthy=False),
|
|
clone, data_root, pre, timeout=1, ready_timeout=1)
|
|
assert unhealthy["confirmed"] is False and unhealthy["reason"] == "unhealthy"
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Scenario contracts: per-task check keys, the dispatch verdict, typed refusal facts, SM1 parity
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
class _FakeServer: # the scenario-facing surface (wait_task/cancel_task) and the one confirm_absorb reads
|
|
base_url = "http://127.0.0.1:1"
|
|
|
|
def __init__(self, status: str = "completed", *, absorb: dict | None = None, healthy: bool = True) -> None:
|
|
self.status, self.absorb, self.healthy = status, dict(absorb or {}), healthy
|
|
|
|
def wait_task(self, task_id, timeout=0):
|
|
return {"status": self.status, "reason_code": "final_message" if self.status == "completed" else "deadline_local"}
|
|
|
|
def cancel_task(self, task_id):
|
|
return {}
|
|
|
|
def wait_for_absorb(self, prev_sha, prev_absorbed, timeout=0):
|
|
return dict(self.absorb)
|
|
|
|
def wait_for_health(self, timeout=0):
|
|
return self.healthy
|
|
|
|
|
|
class _FakeHarness:
|
|
@staticmethod
|
|
def wait_durable_result(oracle, task_id, timeout=0):
|
|
return {"status": "completed", "reason_code": "final_message", "task_id": task_id}
|
|
|
|
|
|
def _ctx(server=None, *, ui_resolver=None, restart=lambda: None, shots=pathlib.Path("/s")) -> scenarios.LaneContext:
|
|
return scenarios.LaneContext(server=server or _FakeServer(), clone=pathlib.Path("/x"), data_root=pathlib.Path("/y"),
|
|
oracle=None, harness=_FakeHarness(), ui_resolver=ui_resolver, ui_reason="", shots=shots,
|
|
log=lambda m: None, task_timeout=1, restart=restart)
|
|
|
|
|
|
try:
|
|
from playwright.sync_api import TargetClosedError # newer Playwright re-exports it
|
|
except ImportError: # pragma: no cover - depends on the installed Playwright
|
|
try:
|
|
from playwright._impl._errors import TargetClosedError
|
|
except ImportError:
|
|
class TargetClosedError(Exception): # type: ignore[no-redef]
|
|
"""Stand-in with Playwright's class name when Playwright is not installed."""
|
|
|
|
|
|
class _FakeUI:
|
|
"""A UI client recording its lifecycle; ``goto`` raises ``fail_goto`` when given (the dead
|
|
target of the rc.14 incident: the chrome died during the absorb wait, the driver lived)."""
|
|
|
|
def __init__(self, base_url: str, calls: list, fail_goto: Exception | None = None) -> None:
|
|
self.base_url, self.calls, self.fail_goto = base_url, calls, fail_goto
|
|
|
|
def open(self):
|
|
self.calls.append(("open", self.base_url))
|
|
return self
|
|
|
|
def goto(self, path="/", *, ready_selector="#chat-input"):
|
|
self.calls.append(("goto", path))
|
|
if self.fail_goto is not None:
|
|
raise self.fail_goto
|
|
|
|
def computed_property(self, selector, prop):
|
|
self.calls.append(("computed_property", selector, prop))
|
|
return scenarios.SM1_NEW_ACCENT
|
|
|
|
send_chat = rebind = lambda self, *a, **k: None
|
|
|
|
def screenshot(self, path):
|
|
self.calls.append(("screenshot", str(path)))
|
|
|
|
def close(self):
|
|
self.calls.append(("close", self.base_url))
|
|
|
|
|
|
def _ui_resolver(calls: list, fail_goto: Exception | None = None):
|
|
def resolve(base_url: str):
|
|
return _FakeUI(base_url, calls, fail_goto).open(), ""
|
|
return resolve
|
|
|
|
|
|
def _sm1_ui_tail(ctx: scenarios.LaneContext) -> None:
|
|
"""Exercise the production oracle, including both documents, without a paid task."""
|
|
if "commit_landed" not in ctx.checks:
|
|
ctx.check("commit_landed", True)
|
|
scenarios.check_sm1_rendered_palette(ctx, scenarios.SM1_REQUIRED_PALETTE)
|
|
|
|
|
|
def test_ui_client_opens_on_first_use_and_restart_reopens_against_the_new_server(tmp_path):
|
|
calls: list = []
|
|
servers = [_FakeServer(), _FakeServer()]
|
|
servers[1].base_url = "http://127.0.0.1:2"
|
|
ctx = _ctx(servers[0], ui_resolver=_ui_resolver(calls), restart=lambda: servers[1], shots=tmp_path)
|
|
assert calls == [] # nothing opened at construction
|
|
assert ctx.ui is not None and calls == [("open", "http://127.0.0.1:1")]
|
|
assert ctx.ui is not None and len(calls) == 1 # one client per open, not one per access
|
|
_sm1_ui_tail(ctx)
|
|
assert calls[1] == ("close", "http://127.0.0.1:1") and calls[2] == ("open", "http://127.0.0.1:2")
|
|
assert ctx.checks == {"commit_landed": True, "ui_app_palette": True, "ui_onboarding_palette": True,
|
|
"ui_computed_style": True} and ctx.ui_reason == "" and "ui_reason" not in ctx.facts
|
|
assert ctx.screenshots == [str(tmp_path / "sm1_app_after_restart.png"),
|
|
str(tmp_path / "sm1_onboarding_after_restart.png")]
|
|
assert [c for c in calls if c[0] == "goto"] == [("goto", "/"), ("goto", "/onboarding")]
|
|
ctx.close_ui()
|
|
ctx.close_ui() # idempotent
|
|
assert [c for c in calls if c[0] == "close"] == [("close", "http://127.0.0.1:1"), ("close", "http://127.0.0.1:2")]
|
|
|
|
|
|
def test_ui_open_failure_is_a_typed_reason_and_never_retried_before_restart():
|
|
attempts: list = []
|
|
|
|
def refuse(base_url):
|
|
attempts.append(base_url)
|
|
return None, "ui_unavailable:browser_missing"
|
|
|
|
ctx = _ctx(ui_resolver=refuse, restart=_FakeServer)
|
|
assert ctx.ui is None and ctx.ui is None and attempts == ["http://127.0.0.1:1"]
|
|
assert ctx.ui_reason == "ui_unavailable:browser_missing"
|
|
ctx.restart() # a restart is the one re-resolve point
|
|
assert ctx.ui is None and len(attempts) == 2
|
|
|
|
|
|
def test_closed_target_degrades_the_ui_checks_typed_and_keeps_every_other_check(tmp_path):
|
|
calls: list = []
|
|
ctx = _ctx(ui_resolver=_ui_resolver(calls, TargetClosedError("Target page, context or browser has been closed")),
|
|
restart=_FakeServer, shots=tmp_path)
|
|
ctx.check("commit_landed", True)
|
|
_sm1_ui_tail(ctx) # no exception escapes
|
|
assert ctx.checks == {"commit_landed": True, "ui_app_palette": False, "ui_onboarding_palette": False,
|
|
"ui_computed_style": False}
|
|
assert ctx.ui_reason == ctx.facts["ui_reason"] == "ui_unavailable:TargetClosedError"
|
|
assert ctx.facts["ui_errors"] == ["TargetClosedError: Target page, context or browser has been closed"]
|
|
assert not any(ctx.facts["palette_computed"]["app"].values()) and ctx.screenshots == []
|
|
assert [c[0] for c in calls] == ["open", "goto", "close"] # closed on the failure, later calls no-ops
|
|
|
|
|
|
class _NoopServer:
|
|
base_url, attestation = "http://127.0.0.1:1", {}
|
|
__init__ = start = stop = lambda self, *a, **k: None
|
|
|
|
|
|
def _attempt_row(tmp_path, monkeypatch, sid: str, acceptance=lambda ctx: ctx.check("scenario_ok", True), *,
|
|
flags: str = "") -> dict:
|
|
"""One ``run_attempt`` of ``sid`` under ``<tmp_path>/<sid>``: a fresh git seed, the real template, the no-op
|
|
server, ``acceptance`` in the scenario's place (default: one passing check); ``<sid>/out`` keeps the artifacts."""
|
|
_short_tmp(monkeypatch)
|
|
monkeypatch.setattr(run_live_lanes, "IsolatedServer", _NoopServer)
|
|
monkeypatch.setitem(run_live_lanes.SCENARIOS, sid, dataclasses.replace(scenarios.SCENARIOS[sid], acceptance=acceptance))
|
|
root = tmp_path / sid
|
|
root.mkdir()
|
|
seed = _git_seed(root)
|
|
args = run_live_lanes.parse_args(["--out", str(root / "out"), "--watch-interval", "600", *flags.split()])
|
|
return run_live_lanes.run_attempt((sid, 1), args, root / "out", run_live_lanes.effective_settings(args, ""),
|
|
run_live_lanes.Stagger(0.0), {}, seed,
|
|
run_live_lanes.RunBudget(100.0, 8.0, reader=lambda root: (0.0, 0)),
|
|
dispatch_index=0, key="", seed_sha=repo_provenance(seed)["head"])
|
|
|
|
|
|
def test_lane_with_a_dead_browser_target_is_checks_failed_not_infra_error(tmp_path, monkeypatch):
|
|
"""The rc.14 incident at lane level: the probe at lane start opens and closes, the client the
|
|
scenario uses opens after the restart, its ``goto`` meets a closed target — the lane row is
|
|
``fail/checks_failed`` with the UI check typed, the task-side checks kept, no ``refusal``."""
|
|
calls: list = []
|
|
|
|
def acceptance(ctx):
|
|
ctx.check("commit_landed", True)
|
|
_sm1_ui_tail(ctx)
|
|
|
|
monkeypatch.setattr(run_live_lanes, "resolve_ui_client",
|
|
_ui_resolver(calls, TargetClosedError("Target page, context or browser has been closed")))
|
|
row = _attempt_row(tmp_path, monkeypatch, "SM1", acceptance)
|
|
assert row["status"] == "fail" and row["reason_code"] == "checks_failed" and row["error"] == ""
|
|
assert "refusal" not in row
|
|
assert row["checks"]["commit_landed"] is True and row["checks"]["ui_computed_style"] is False
|
|
assert row["facts"]["ui_reason"] == "ui_unavailable:TargetClosedError"
|
|
assert row["ui"] == {"available": False, "reason": "ui_unavailable:TargetClosedError"}
|
|
# lane start: availability probe opened and closed; use: opened after the restart, dead, closed
|
|
assert [c[0] for c in calls] == ["open", "close", "open", "goto", "close"]
|
|
stored = json.loads((tmp_path / "SM1" / "out" / "lanes" / "SM1_a1" / "result.json").read_text(encoding="utf-8"))
|
|
assert stored["status"] == "fail" and stored["facts"]["ui_reason"] == "ui_unavailable:TargetClosedError"
|
|
|
|
|
|
def test_absorb_wait_and_check_follow_the_scenarios_expects_absorb(tmp_path, monkeypatch):
|
|
"""The rc.15 paid stand (2026-09-05, SK1_a1): every ``--self-mod`` lane waited ``--task-timeout`` for an absorb
|
|
only SM1's commit could trigger, then failed ``self_mod_absorb_confirmed`` by construction. Now SM1 waits and
|
|
carries the check; SW1/SK1 stop right after the scenario with ``{"expected": False}``, no check, post-task
|
|
evolution OFF in their settings; every lane seeds ``owner_chat_id`` ONLY, never a campaign (run2's t=0 cycles)."""
|
|
waits: list = []
|
|
monkeypatch.setattr(run_live_lanes, "resolve_ui_client", lambda base_url: (None, "ui_unavailable:test"))
|
|
monkeypatch.setattr(run_live_lanes, "self_mod_snapshot", lambda server, clone, data_root: {"pre": True})
|
|
monkeypatch.setattr(run_live_lanes, "confirm_absorb", lambda server, clone, data_root, pre, **kw: (
|
|
waits.append(pre) or {"confirmed": False, "reason": "no_promotion", "healthy": True}))
|
|
sm1 = _attempt_row(tmp_path, monkeypatch, "SM1", flags="--self-mod")
|
|
assert waits == [{"pre": True}] and sm1["status"] == "fail" and sm1["checks"]["self_mod_absorb_confirmed"] is False
|
|
assert sm1["self_mod_absorb"] == {"expected": True, "confirmed": False, "reason": "no_promotion", "healthy": True}
|
|
for sid in ("SM1", "SW1", "SK1"):
|
|
if sid != "SM1":
|
|
row = _attempt_row(tmp_path, monkeypatch, sid, flags="--self-mod")
|
|
assert row["status"] == "pass" and "self_mod_absorb_confirmed" not in row["checks"], row["checks"]
|
|
assert row["self_mod_absorb"] == {"expected": False} and row["self_mod"] is True and waits == [{"pre": True}]
|
|
lane = tmp_path / sid / "out" / "lanes" / f"{sid}_a1" / "data"
|
|
state = json.loads((lane / "state" / "state.json").read_text(encoding="utf-8"))
|
|
assert json.loads((lane / "settings.json").read_text())["OUROBOROS_POST_TASK_EVOLUTION"] == ("true" if sid == "SM1" else "false")
|
|
assert state["owner_chat_id"] == 1 and "evolution_mode_enabled" not in state, state
|
|
assert not (lane / "state" / "evolution_campaign.json").exists(), sid
|
|
|
|
|
|
def test_wait_task_namespaces_checks_per_task_and_check_refuses_overwrites():
|
|
ctx = _ctx()
|
|
ctx.wait_task("t1", label="author")
|
|
ctx.wait_task("t2", label="dispatch")
|
|
assert set(ctx.checks) == {"author_http_terminal_completed", "author_durable_terminal_completed",
|
|
"dispatch_http_terminal_completed", "dispatch_durable_terminal_completed"}
|
|
assert all(ctx.checks.values())
|
|
assert ctx.facts["author_terminal"]["task_id"] == "t1" and ctx.facts["dispatch_terminal"]["task_id"] == "t2"
|
|
assert ctx.facts["author_http_status"] == "completed" and ctx.facts["runtime_result"]["task_id"] == "t2"
|
|
with pytest.raises(scenarios.DuplicateCheckKey):
|
|
ctx.wait_task("t3", label="author")
|
|
with pytest.raises(scenarios.DuplicateCheckKey):
|
|
ctx.check("author_http_terminal_completed", True)
|
|
# An unlabeled await keeps the plain keys for single-task scenarios.
|
|
plain = _ctx()
|
|
plain.wait_task("t9")
|
|
assert set(plain.checks) == {"http_terminal_completed", "durable_terminal_completed"}
|
|
|
|
|
|
def test_dispatch_verdict_requires_ok_status_and_the_exact_echo():
|
|
gen = "f773dad013e846c793dccd7938188b46"
|
|
failed = [{"tool": "ext_x", "status": "error", "result_preview": "boom",
|
|
"tool_result_meta": {"extension_generation": gen, "physical_dispatch": True}}]
|
|
verdict = scenarios.dispatch_verdict(failed, scenarios.SK1_ECHO_EXPECTED)
|
|
assert verdict["generation_ok"] and verdict["status"] == "error" and not verdict["echo_ok"]
|
|
good = [{"tool": "ext_x", "status": "ok", "result_preview": "echo: ping-e2e-live\n",
|
|
"tool_result_meta": {"extension_generation": gen, "physical_dispatch": True}}]
|
|
verdict = scenarios.dispatch_verdict(good, scenarios.SK1_ECHO_EXPECTED)
|
|
assert verdict == {"row_present": True, "status": "ok", "generation": gen, "generation_ok": True,
|
|
"physical_dispatch": True, "echo_ok": True}
|
|
assert scenarios.dispatch_verdict([], scenarios.SK1_ECHO_EXPECTED)["row_present"] is False
|
|
assert scenarios.SK1_ECHO_EXPECTED == f"echo: {scenarios.SK1_ECHO_MESSAGE}"
|
|
assert scenarios.SCENARIOS["SK1"].stub_script(REPO_ROOT)["agent"][4]["arguments"]["message"] == scenarios.SK1_ECHO_MESSAGE
|
|
# The relayed line opens one owner-chat turn on the stub wire: a second closing final absorbs it.
|
|
assert [list(s)[0] for s in scenarios.SCENARIOS["SK1"].stub_script(REPO_ROOT)["agent"]] == [
|
|
"tool", "tool", "tool", "final", "tool", "final", "final"]
|
|
|
|
|
|
def test_sk1_fixture_declares_exactly_the_permissions_its_plugin_exercises():
|
|
"""The SK1 manifest is honest by construction: every declared permission maps to source the
|
|
plugin actually runs, the ONLY owner-granted one (``inject_chat``) is what the stand grants,
|
|
and the prose states that narrow purpose. The first paid run declared ``inject_chat`` over an
|
|
echo-only plugin and the skill review refused it 3/3 on ``permissions_honesty`` +
|
|
``inject_chat_minimization`` — a fixture defect, so this pins the fixture, not the reviewer."""
|
|
import ast
|
|
|
|
from ouroboros.contracts.skill_manifest import parse_skill_manifest_text
|
|
from ouroboros.skill_loader import requested_skill_permissions
|
|
|
|
manifest = parse_skill_manifest_text(scenarios.SK1_SKILL_MD)
|
|
assert manifest.name == scenarios.SK1_SKILL and manifest.type == "extension" and manifest.entry == "plugin.py"
|
|
exercised_by = { # permission -> the source that performs it
|
|
"tool": "api.register_tool(",
|
|
"inject_chat": "/chat/inject",
|
|
"net": "urllib.request",
|
|
}
|
|
assert set(manifest.permissions) == set(exercised_by)
|
|
for permission, marker in exercised_by.items():
|
|
assert marker in scenarios.SK1_PLUGIN, (permission, marker)
|
|
assert requested_skill_permissions(list(manifest.permissions)) == scenarios.SK1_GRANTS == ["inject_chat"]
|
|
# Host-token discipline (checklist item 12): the token is revealed at the request site only.
|
|
assert scenarios.SK1_PLUGIN.count("get_skill_token().use_in_request()") == 1
|
|
assert "print(" not in scenarios.SK1_PLUGIN and "log(" not in scenarios.SK1_PLUGIN
|
|
# Owner binding: the destination is a module constant, never a tool argument.
|
|
assert f"OWNER_CHAT_ID = {scenarios.SK1_OWNER_CHAT_ID}" in scenarios.SK1_PLUGIN and scenarios.SK1_OWNER_CHAT_ID == 1
|
|
assert "'chat_id': OWNER_CHAT_ID" in scenarios.SK1_PLUGIN
|
|
tree = ast.parse(scenarios.SK1_PLUGIN)
|
|
register_call = next(n for n in ast.walk(tree) if isinstance(n, ast.Call)
|
|
and isinstance(n.func, ast.Attribute) and n.func.attr == "register_tool")
|
|
schema = ast.literal_eval(next(k.value for k in register_call.keywords if k.arg == "schema"))
|
|
assert set(schema["properties"]) == {"message"}
|
|
# The prose names the purpose and no longer denies what the code does.
|
|
body = scenarios.SK1_SKILL_MD.split("---", 2)[2]
|
|
assert "/chat/inject" in body and f"chat_id {scenarios.SK1_OWNER_CHAT_ID}" in body and "127.0.0.1" in body
|
|
assert "no host or network access" not in body
|
|
|
|
|
|
class _FakeExtensionApi:
|
|
"""Only the two PluginAPI members the probe plugin touches (the token object: ``use_in_request`` alone)."""
|
|
|
|
def __init__(self, token: str) -> None:
|
|
self.token, self.tools = token, {}
|
|
|
|
def register_tool(self, name, handler, *, description, schema, timeout_sec=60):
|
|
self.tools[name] = handler
|
|
|
|
def get_skill_token(self):
|
|
return types.SimpleNamespace(use_in_request=lambda: self.token)
|
|
|
|
|
|
def _inject_sink(status: int, hits: list):
|
|
"""A loopback HTTP server standing in for the Host Service ``/chat/inject`` route."""
|
|
import http.server
|
|
|
|
class Handler(http.server.BaseHTTPRequestHandler):
|
|
def do_POST(self):
|
|
body = self.rfile.read(int(self.headers.get("Content-Length") or 0))
|
|
hits.append({"path": self.path, "token": self.headers.get("X-Skill-Token"),
|
|
"body": json.loads(body.decode("utf-8"))})
|
|
self.send_response(status)
|
|
self.send_header("Content-Type", "application/json")
|
|
self.end_headers()
|
|
self.wfile.write(b'{"ok": true}')
|
|
|
|
def log_message(self, *_args): # keep pytest output clean
|
|
return
|
|
|
|
server = http.server.HTTPServer(("127.0.0.1", 0), Handler)
|
|
threading.Thread(target=server.serve_forever, daemon=True).start()
|
|
return server
|
|
|
|
|
|
def test_sk1_plugin_relays_one_bounded_line_into_the_owner_chat(monkeypatch):
|
|
"""The plugin text the model is told to write, executed: one POST to the loopback
|
|
``/chat/inject`` per call, owner chat pinned, the skill token only in the header, the text
|
|
bounded, the same text returned — and a Host Service refusal surfaces as a tool error."""
|
|
hits: list = []
|
|
sink = _inject_sink(202, hits)
|
|
try:
|
|
monkeypatch.setenv("HOST_SERVICE_URL", f"http://127.0.0.1:{sink.server_port}")
|
|
namespace: dict = {}
|
|
exec(compile(scenarios.SK1_PLUGIN, "plugin.py", "exec"), namespace) # noqa: S102 - the fixture under test
|
|
api = _FakeExtensionApi("tok-e2e")
|
|
namespace["register"](api)
|
|
echo = api.tools["echo"]
|
|
assert echo(None, message=scenarios.SK1_ECHO_MESSAGE) == scenarios.SK1_ECHO_EXPECTED
|
|
assert hits == [{"path": "/chat/inject", "token": "tok-e2e", "body": {
|
|
"text": scenarios.SK1_ECHO_EXPECTED, "chat_id": scenarios.SK1_OWNER_CHAT_ID,
|
|
"sender_label": scenarios.SK1_SKILL}}]
|
|
long = echo(None, message="x" * (scenarios.SK1_ECHO_MAX_CHARS + 50))
|
|
assert long == hits[-1]["body"]["text"] == ("echo: " + "x" * scenarios.SK1_ECHO_MAX_CHARS)[:scenarios.SK1_ECHO_MAX_CHARS]
|
|
assert len(long) == scenarios.SK1_ECHO_MAX_CHARS # the cap bounds the FINAL text, prefix included
|
|
multi = echo(None, message="first\r\nsecond\nthird")
|
|
assert multi == hits[-1]["body"]["text"] == "echo: first second third" # ONE line: breaks collapse
|
|
assert len(hits) == 3 # exactly one line per call, no retry
|
|
finally:
|
|
sink.shutdown()
|
|
refusing = _inject_sink(403, [])
|
|
try:
|
|
monkeypatch.setenv("HOST_SERVICE_URL", f"http://127.0.0.1:{refusing.server_port}")
|
|
with pytest.raises(urllib.error.HTTPError):
|
|
echo(None, message="denied")
|
|
finally:
|
|
refusing.shutdown()
|
|
# The acceptance reads the HOST's attribution of that line, never the plugin's claim.
|
|
rows = [{"direction": "in", "chat_id": 1, "source": f"skill:{scenarios.SK1_SKILL}", "text": scenarios.SK1_ECHO_EXPECTED},
|
|
{"direction": "out", "chat_id": 1, "source": f"skill:{scenarios.SK1_SKILL}", "text": scenarios.SK1_ECHO_EXPECTED},
|
|
{"direction": "in", "chat_id": 1, "source": "web", "text": scenarios.SK1_ECHO_EXPECTED},
|
|
{"direction": "in", "chat_id": 2, "source": f"skill:{scenarios.SK1_SKILL}", "text": scenarios.SK1_ECHO_EXPECTED},
|
|
{"direction": "in", "chat_id": 1, "source": f"skill:{scenarios.SK1_SKILL}", "text": "echo: other"}]
|
|
assert scenarios.owner_chat_relay_rows(rows, scenarios.SK1_SKILL, scenarios.SK1_ECHO_EXPECTED) == rows[:1]
|
|
|
|
|
|
def test_commit_refusal_facts_name_every_typed_refusal():
|
|
ledger = {"attempts": [
|
|
{"attempt": 1, "phase": "preflight", "status": "blocked", "block_reason": "tests_preflight_blocked"},
|
|
{"attempt": 2, "phase": "blocking_review", "status": "blocked", "block_reason": "scope_blocked"},
|
|
{"attempt": 3, "phase": "late_wait", "status": "reviewing", "block_reason": "review_late_result_pending"}],
|
|
"advisory_runs": [{"status": "stale"}, {"status": "bypassed"}]}
|
|
tools = [
|
|
{"tool": "preflight_review", "status": "ok",
|
|
"result_preview": '{\n "status": "preflight_blocked",\n "error": "⚠️ PREFLIGHT_BLOCKED: VERSION is not in scope'},
|
|
{"tool": "commit_reviewed", "status": "blocked", "result_preview": "⚠️ TESTS_PREFLIGHT_BLOCKED: Tests must pass"},
|
|
{"tool": "commit_reviewed", "status": "blocked", "result_preview": "⚠️ SCOPE_REVIEW_BLOCKED: the review pack"},
|
|
{"tool": "write_file", "status": "ok", "result_preview": "⚠️ NOT_A_REVIEW_TOOL: ignored"},
|
|
{"tool": "commit_reviewed", "status": "ok", "result_preview": "⚠️ REVIEW_PENDING: physical reviewer work"}]
|
|
facts = scenarios.commit_refusal_facts(ledger, tools, {"status": "failed", "reason_code": "budget_exhausted"})
|
|
assert facts["refusal_codes"] == ["PREFLIGHT_BLOCKED", "REVIEW_PENDING", "SCOPE_REVIEW_BLOCKED", "TESTS_PREFLIGHT_BLOCKED"]
|
|
assert [a["block_reason"] for a in facts["commit_attempts"]] == [
|
|
"tests_preflight_blocked", "scope_blocked", "review_late_result_pending"]
|
|
assert facts["advisory_run_statuses"] == ["stale", "bypassed"]
|
|
assert facts["review_tool_calls"][1] == {"tool": "commit_reviewed", "status": "blocked", "code": "TESTS_PREFLIGHT_BLOCKED"}
|
|
assert facts["terminal_status"] == "failed" and facts["terminal_reason_code"] == "budget_exhausted"
|
|
|
|
|
|
def test_sm1_changes_the_shared_palette_for_both_documents():
|
|
"""One palette edit, same release/review path. Source ownership is not browser evidence:
|
|
separate real-document tests reject missing CSS and page-local palette overrides."""
|
|
assert scenarios.SM1_CSS_PATHS == ("web/ui.css",)
|
|
prompt = scenarios.sm1_prompt()
|
|
assert "web/ui.css" in prompt and "/onboarding" in prompt and "docs/DESIGN.md" in prompt
|
|
assert "both browser documents" in prompt
|
|
assert "skip_" not in prompt.lower() and "do not bump" not in prompt.lower() and "bumped in the same diff" in prompt
|
|
script = scenarios.sm1_stub_script(REPO_ROOT)["agent"]
|
|
writes = [s for s in script if s.get("tool") == "write_file"]
|
|
written = [w["arguments"]["path"] for w in writes]
|
|
assert written[0] == scenarios.SM1_CSS_PATH
|
|
assert "web/style.css" not in written and "web/onboarding.css" not in written
|
|
commit = next(s for s in script if s.get("tool") == "commit_reviewed")["arguments"]
|
|
assert commit["paths"] == written and "commit_message" in commit
|
|
assert not any(key.startswith("skip_") for key in commit), "the stub rehearsal takes the full user path like the paid prompt"
|
|
text = writes[0]["arguments"]["content"]
|
|
original = (REPO_ROOT / scenarios.SM1_CSS_PATH).read_text(encoding="utf-8")
|
|
assert scenarios.accent_value(text) == scenarios.SM1_NEW_ACCENT
|
|
# Hermetic preflight can already be running on the target-accent candidate.
|
|
if scenarios.accent_value(original) != scenarios.SM1_NEW_ACCENT:
|
|
assert text != original
|
|
assert len(text.splitlines()) == len(original.splitlines())
|
|
palette = scenarios.sm1_palette_tokens(text)
|
|
assert scenarios.SM1_REQUIRED_PALETTE <= palette.keys() and all(palette.values())
|
|
assert all(f"--accent-{alpha}" in palette for alpha in ("04", "05", "08", "10", "12", "18", "22", "25", "35", "45", "55", "65"))
|
|
for document in ("web/index.html", "web/onboarding_template.html"):
|
|
assert 'href="/static/ui.css"' in (REPO_ROOT / document).read_text(encoding="utf-8")
|
|
|
|
|
|
def test_sm1_stub_bumps_the_release_carriers_through_the_sync_ssot(tmp_path):
|
|
"""The stub's bump is a strictly-greater release version whose carriers come from
|
|
``release_sync`` (no hand list) and pass the product's own release admission gate; the
|
|
acceptance's advisory-row and vision-evidence readers tell the real rows from the audited ones."""
|
|
from ouroboros.commit_admission import release_metadata_preflight
|
|
from ouroboros.tools.release_sync import CARRIER_SPAN_PATHS
|
|
|
|
writes = [s for s in scenarios.sm1_stub_script(REPO_ROOT)["agent"] if s.get("tool") == "write_file"]
|
|
carriers = {w["arguments"]["path"]: w["arguments"]["content"] for w in writes
|
|
if w["arguments"]["path"] not in scenarios.SM1_CSS_PATHS}
|
|
assert {"VERSION", "README.md"} <= set(carriers) <= CARRIER_SPAN_PATHS
|
|
seed = (REPO_ROOT / "VERSION").read_text(encoding="utf-8").strip()
|
|
bumped = carriers["VERSION"].strip()
|
|
assert scenarios.version_is_bumped(seed, bumped) and f"| {bumped} |" in carriers["README.md"]
|
|
root = tmp_path / "carriers"
|
|
root.mkdir()
|
|
# Release admission reads Git scope, so materialize carriers in a disposable
|
|
# repository rather than a bare directory.
|
|
subprocess.run(["git", "init", "-q"], cwd=str(root), check=True)
|
|
for rel in sorted(CARRIER_SPAN_PATHS):
|
|
if (REPO_ROOT / rel).is_file():
|
|
(root / rel).parent.mkdir(parents=True, exist_ok=True)
|
|
(root / rel).write_text(carriers.get(rel) or (REPO_ROOT / rel).read_text(encoding="utf-8"), encoding="utf-8")
|
|
assert release_metadata_preflight(root, scenarios.SM1_COMMIT_MESSAGE, ["VERSION"]) is None
|
|
assert scenarios.release_carriers_desync_at(root, _commit(root, "coherent release")) == ""
|
|
(root / "pyproject.toml").write_text('[project]\nversion = "0.0.0"\n', encoding="utf-8")
|
|
assert "pyproject.toml" in scenarios.release_carriers_desync_at(root, _commit(root, "broken carrier"))
|
|
assert scenarios.sm1_next_version("7.0.0-rc.14") == "7.0.0-rc.15" and scenarios.sm1_next_version("7.0.0") == "7.0.1"
|
|
# A seed cloned from an older ref carries the newer tags: the stub skips taken versions.
|
|
assert scenarios.sm1_next_version("7.0.0-rc.14", {"v7.0.0-rc.15", "v7.0.0-rc.16"}) == "7.0.0-rc.17"
|
|
assert scenarios.sm1_next_version("7.0.0", {"v7.0.1"}) == "7.0.2"
|
|
assert not scenarios.version_is_bumped("7.0.0-rc.14", "7.0.0-rc.14") and not scenarios.version_is_bumped("7.0.0-rc.14", "7.0.0-rc.13")
|
|
assert scenarios.advisory_run_is_real({"status": "fresh"}) and scenarios.advisory_run_is_real({"status": "stale", "raw_result": "[]"})
|
|
assert not scenarios.advisory_run_is_real({"status": "bypassed", "bypass_reason": "skip_advisory_review"})
|
|
assert not scenarios.advisory_run_is_real({"status": "stale", "raw_result": "⚠️ ADVISORY_SKIPPED: prompt too large"})
|
|
rows = [{"tool": "vlm_query"}, {"tool": "browser_action", "args": {"action": "click"}}, {"tool": "read_file"}]
|
|
assert [r["tool"] for r in scenarios.vision_evidence_rows(rows)] == ["vlm_query"]
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Seed: a clean detached clone of the requested ref, never the operator's live worktree
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
def test_materialize_seed_is_a_clean_detached_clone_of_the_ref(tmp_path):
|
|
source = _git_seed(tmp_path)
|
|
first = subprocess.run(["git", "rev-parse", "HEAD"], cwd=str(source), check=True, capture_output=True, text=True).stdout.strip()
|
|
(source / "VERSION").write_text("7.0.1-test\n", encoding="utf-8")
|
|
second = _commit(source, "bump")
|
|
(source / "VERSION").write_text("7.0.2-wip\n", encoding="utf-8") # dirty source, never under test
|
|
seed = tmp_path / "seed"
|
|
record = run_live_lanes.materialize_seed(source, "HEAD~1", seed)
|
|
assert record["resolved_sha"] == first and record["policy"] == run_live_lanes.SEED_POLICY
|
|
assert (seed / "VERSION").read_text(encoding="utf-8") == "7.0.0-test\n"
|
|
detached = subprocess.run(["git", "symbolic-ref", "-q", "HEAD"], cwd=str(seed), check=False, capture_output=True)
|
|
assert detached.returncode != 0 # no branch checked out
|
|
provenance = repo_provenance(seed)
|
|
assert run_live_lanes.seed_is_clean(provenance, first) and not run_live_lanes.seed_is_clean(provenance, second)
|
|
assert not provenance["describe"].endswith("-dirty")
|
|
with pytest.raises(run_live_lanes.SeedMaterializeRefused) as exists:
|
|
run_live_lanes.materialize_seed(source, "HEAD", seed)
|
|
assert exists.value.reason == "seed_dir_exists"
|
|
with pytest.raises(run_live_lanes.SeedMaterializeRefused) as bogus:
|
|
run_live_lanes.materialize_seed(source, "no-such-ref", tmp_path / "seed2")
|
|
assert bogus.value.reason == "ref_unresolved"
|
|
|
|
|
|
def test_dirty_source_runs_the_committed_ref_from_a_clean_detached_seed(tmp_path, monkeypatch):
|
|
_short_tmp(monkeypatch)
|
|
monkeypatch.setattr(run_live_lanes, "run_lane", _fake_lane)
|
|
source = _git_seed(tmp_path, dirty=True)
|
|
head = subprocess.run(["git", "rev-parse", "HEAD"], cwd=str(source), check=True, capture_output=True, text=True).stdout.strip()
|
|
out = tmp_path / "out"
|
|
rc = run_live_lanes.main(["--stub", "--source-repo", str(source), "--out", str(out), "--scenarios", "SM1",
|
|
"--watch-interval", "600"])
|
|
assert rc == 0
|
|
manifest = json.loads((out / "run_manifest.json").read_text(encoding="utf-8"))
|
|
assert manifest["source"]["dirty"] is True and manifest["seed_gate"]["allow_dirty_seed"] is True
|
|
assert manifest["seed"]["resolved_sha"] == head and manifest["seed"]["clean"] is True
|
|
assert manifest["seed"]["requested_ref"] == "HEAD" and manifest["extra"]["seed_policy"] == run_live_lanes.SEED_POLICY
|
|
assert manifest["extra"]["seed_head"] == head and not manifest["extra"]["seed_describe"].endswith("-dirty")
|
|
assert (out / "seed" / "VERSION").read_text(encoding="utf-8") == "7.0.0-test\n" # committed, not the edit
|
|
row = json.loads((out / "lanes" / "SM1_a1" / "result.json").read_text(encoding="utf-8"))
|
|
assert row["seed_sha"] == head
|
|
|
|
|
|
def test_unresolvable_seed_ref_is_a_typed_refusal_before_any_lane(tmp_path, monkeypatch):
|
|
_short_tmp(monkeypatch)
|
|
monkeypatch.setattr(run_live_lanes, "run_lane", lambda *a, **k: pytest.fail("a lane started without a seed"))
|
|
source = _git_seed(tmp_path)
|
|
out = tmp_path / "out"
|
|
rc = run_live_lanes.main(["--stub", "--source-repo", str(source), "--seed", "no-such-ref", "--out", str(out),
|
|
"--watch-interval", "600"])
|
|
assert rc == 3
|
|
manifest = json.loads((out / "run_manifest.json").read_text(encoding="utf-8"))
|
|
assert manifest["extra"]["outcome"] == "refused" and manifest["extra"]["exit_code"] == 3
|
|
assert manifest["extra"]["refusal"]["stage"] == "seed_materialize"
|
|
assert manifest["extra"]["refusal"]["reason"] == "ref_unresolved"
|
|
assert not (out / "lanes").exists() and not (out / "effective_settings.json").exists()
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Admission and the typed refusals (persisted manifest, no footprint)
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
def test_run_root_confinement_refuses_before_anything_is_created(tmp_path, monkeypatch):
|
|
_short_tmp(monkeypatch)
|
|
source = _git_seed(tmp_path)
|
|
with pytest.raises(ValueError, match="must not be under repo/"):
|
|
run_live_lanes.main(["--stub", "--source-repo", str(source), "--out", str(source / "inside")])
|
|
assert not (source / "inside").exists()
|
|
|
|
|
|
def test_missing_key_env_is_a_typed_refusal_before_any_lane_starts(tmp_path, monkeypatch):
|
|
_short_tmp(monkeypatch)
|
|
monkeypatch.delenv("E2E_TEST_KEY_ENV", raising=False)
|
|
monkeypatch.setattr(run_live_lanes, "run_lane", lambda *a, **k: pytest.fail("a lane started without a key"))
|
|
source = _git_seed(tmp_path)
|
|
out = tmp_path / "out"
|
|
rc = run_live_lanes.main(["--source-repo", str(source), "--out", str(out), "--key-env", "E2E_TEST_KEY_ENV"])
|
|
assert rc == 3
|
|
manifest = json.loads((out / "run_manifest.json").read_text(encoding="utf-8"))
|
|
assert manifest["extra"]["outcome"] == "refused" and manifest["extra"]["exit_code"] == 3
|
|
assert manifest["extra"]["refusal"] == {"stage": "credential", "reason": "key_env_absent", "env": "E2E_TEST_KEY_ENV"}
|
|
assert not (out / "lanes").exists() and not (out / "effective_settings.json").exists() and not (out / "seed").exists()
|
|
|
|
|
|
def test_credit_preflight_takes_the_min_of_both_planes(tmp_path, monkeypatch):
|
|
"""Key limit says $50, the account behind it holds $1: the run is bounded by $1 and refused
|
|
below a $5 floor. Both numbers are recorded; the key value never is."""
|
|
_short_tmp(monkeypatch)
|
|
monkeypatch.setenv("E2E_TEST_KEY_ENV", FAKE_KEY)
|
|
monkeypatch.setattr(run_live_lanes, "run_lane", lambda *a, **k: pytest.fail("a lane started under the floor"))
|
|
calls: list = []
|
|
monkeypatch.setattr(urllib.request, "urlopen", _fake_urlopen(
|
|
b'{"data":{"limit_remaining":50.0}}', b'{"data":{"total_credits":10.0,"total_usage":9.0}}', calls))
|
|
source = _git_seed(tmp_path)
|
|
out = tmp_path / "out"
|
|
rc = run_live_lanes.main(["--source-repo", str(source), "--out", str(out), "--key-env", "E2E_TEST_KEY_ENV",
|
|
"--min-credit-usd", "5"])
|
|
assert rc == 3
|
|
assert calls == ["https://openrouter.ai/api/v1/key", "https://openrouter.ai/api/v1/credits"]
|
|
raw = (out / "run_manifest.json").read_bytes()
|
|
assert FAKE_KEY.encode() not in raw
|
|
manifest = json.loads(raw)
|
|
refusal = manifest["extra"]["refusal"]
|
|
assert refusal["stage"] == "credit_preflight" and refusal["reason"] == "insufficient_remaining"
|
|
assert refusal["remaining_usd"] == 1.0 and refusal["key_limit_remaining_usd"] == 50.0
|
|
assert refusal["account_credits_usd"] == 1.0 and refusal["floor_usd"] == 5.0
|
|
assert manifest["extra"]["credential_fingerprint"].startswith("sha256:")
|
|
|
|
|
|
def test_openrouter_account_credits_is_the_second_bound_only(monkeypatch):
|
|
from devtools.benchmarks.common.manifests import openrouter_account_credits, openrouter_key_remaining
|
|
|
|
calls: list = []
|
|
monkeypatch.setattr(urllib.request, "urlopen", _fake_urlopen(
|
|
b'{"data":{"limit":null}}', b'{"data":{"total_credits":12.5,"total_usage":2.5}}', calls))
|
|
assert openrouter_key_remaining(FAKE_KEY) is None # uncapped key: not "$0", not "plenty"
|
|
assert openrouter_account_credits(FAKE_KEY) == 10.0
|
|
assert run_live_lanes.credit_preflight(FAKE_KEY)["remaining_usd"] == 10.0
|
|
monkeypatch.setattr(urllib.request, "urlopen", _fake_urlopen(b'{"data":{"limit":null}}', b'{"data":{}}', []))
|
|
assert run_live_lanes.credit_preflight(FAKE_KEY, timeout=3) == {
|
|
"key_limit_remaining_usd": None, "account_credits_usd": None, "remaining_usd": None}
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# The manifest names the APPLIED model; secrets stay out of every run-level artifact
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
def _fake_run(tmp_path, monkeypatch, argv: list[str], *, lane=_fake_lane, expect_rc: int = 0) -> tuple[pathlib.Path, dict]:
|
|
_short_tmp(monkeypatch)
|
|
monkeypatch.setenv("E2E_TEST_KEY_ENV", FAKE_KEY)
|
|
monkeypatch.setattr(run_live_lanes, "run_lane", lane)
|
|
monkeypatch.setattr(run_live_lanes, "credit_preflight", lambda key, **_kw: {
|
|
"key_limit_remaining_usd": None, "account_credits_usd": None, "remaining_usd": None})
|
|
source = _git_seed(tmp_path)
|
|
out = tmp_path / "out"
|
|
rc = run_live_lanes.main(["--source-repo", str(source), "--out", str(out), "--key-env", "E2E_TEST_KEY_ENV",
|
|
"--watch-interval", "600", *argv])
|
|
assert rc == expect_rc
|
|
return out, json.loads((out / "run_manifest.json").read_text(encoding="utf-8"))
|
|
|
|
|
|
def test_manifest_names_the_effective_model_not_argv(tmp_path, monkeypatch):
|
|
"""EQUALITY pin: the manifest's model is the one in the APPLIED settings file. argv pins
|
|
model Y, the applied file carries X -> the manifest says X."""
|
|
real = run_live_lanes.effective_settings
|
|
|
|
def applied_differs(args, key):
|
|
return {**real(args, key), "OUROBOROS_MODEL": "applied/model-x"}
|
|
|
|
monkeypatch.setattr(run_live_lanes, "effective_settings", applied_differs)
|
|
out, manifest = _fake_run(tmp_path, monkeypatch, ["--model", "argv/model-y", "--scenarios", "SM1"])
|
|
applied = json.loads((out / "effective_settings.json").read_text(encoding="utf-8"))
|
|
assert manifest["model_slots"]["OUROBOROS_MODEL"] == applied["OUROBOROS_MODEL"] == "applied/model-x"
|
|
assert manifest["extra"]["effective_model"] == "applied/model-x"
|
|
assert "argv/model-y" not in json.dumps(manifest["model_slots"])
|
|
|
|
|
|
def test_run_root_template_is_redacted_and_the_key_reaches_only_the_lanes(tmp_path, monkeypatch):
|
|
out, manifest = _fake_run(tmp_path, monkeypatch, ["--model", "argv/model-y", "--scenarios", "SM1,SK1",
|
|
"--attempts", "2", "--pass-of", "2", "--lanes", "2"])
|
|
assert manifest["model_slots"]["OUROBOROS_MODEL"] == "argv/model-y"
|
|
template_path = out / "effective_settings.json"
|
|
if os.name == "posix":
|
|
assert (template_path.stat().st_mode & 0o777) == 0o600
|
|
else: # Windows: chmod only toggles read-only; the mode reads 0o666 — the redaction is the guarantee there
|
|
assert template_path.is_file()
|
|
template = json.loads(template_path.read_text(encoding="utf-8"))
|
|
assert "OPENROUTER_API_KEY" not in template and template["OUROBOROS_MODEL"] == "argv/model-y"
|
|
for artifact in (out / "run_manifest.json", template_path, *out.glob("lanes/*/result.json")):
|
|
assert FAKE_KEY.encode() not in artifact.read_bytes(), artifact
|
|
creds = manifest["provider_credentials"]
|
|
assert creds["granted"] == {} # the file grant: nothing
|
|
assert creds["runtime_granted"]["OPENROUTER_API_KEY"]["present"] is True
|
|
assert creds["runtime_granted"]["OPENROUTER_API_KEY"]["fingerprint"].startswith("sha256:")
|
|
for row_path in out.glob("lanes/*/result.json"):
|
|
row = json.loads(row_path.read_text(encoding="utf-8"))
|
|
assert row["template_has_key"] is False and row["key_handed"] is True # injected per lane, in memory
|
|
assert manifest["requested_task_ids"] == ["SM1_a1", "SM1_a2", "SK1_a1", "SK1_a2"]
|
|
assert manifest["extra"]["scenarios"] == {
|
|
"SM1": {"attempts": 2, "passed": 2, "infra_errors": 0, "not_run": 0, "verdict": "pass"},
|
|
"SK1": {"attempts": 2, "passed": 2, "infra_errors": 0, "not_run": 0, "verdict": "pass"}}
|
|
assert manifest["extra"]["outcome"] == "completed" and manifest["extra"]["exit_code"] == 0
|
|
assert manifest["extra"]["total_budget_usd"] == 100.0 and manifest["extra"]["per_task_usd"] == 8.0
|
|
budget = manifest["extra"]["budget"]
|
|
assert budget["cap_usd"] == 100.0 and budget["refusals"] == [] and budget["attempts_not_run"] == []
|
|
assert budget["first_refused"] is None and "halted" not in budget
|
|
assert budget["reservation_rule"] == run_live_lanes.RESERVATION_RULE and "stop_reason" not in manifest["extra"]
|
|
|
|
|
|
def test_run_wide_cap_refuses_per_attempt_and_records_not_run_rows(tmp_path, monkeypatch):
|
|
"""cap $16, per-task $4 (SK1 reserves $8, SM1/SW1 $4), one lane, round-robin dispatch; a settled SK1 lane
|
|
reads back $7, every other $2: round 1 runs SK1_a1 (0+8), SM1_a1 (7+4), SW1_a1 (9+4); SK1_a2 (11+8 > 16) is
|
|
refused at once; SM1_a2 (11+4) still RUNS after that refusal — a refusal is per attempt, not a halt; SW1_a2
|
|
(13+4 > 16) is refused. Every refusal is a recorded row and the stop_reason; pass-of 2 fails SW1/SK1, not SM1."""
|
|
monkeypatch.setattr(run_live_lanes, "lane_spend", lambda root: (
|
|
(7.0 if pathlib.Path(root).parent.name.startswith("SK1") else 2.0, 0)
|
|
if pathlib.Path(root).parent.exists() else (0.0, 0)))
|
|
out, manifest = _fake_run(tmp_path, monkeypatch, ["--scenarios", "SM1,SK1,SW1", "--attempts", "2", "--pass-of", "2",
|
|
"--lanes", "1", "--total-budget", "16", "--per-task-usd", "4"], expect_rc=1)
|
|
budget = manifest["extra"]["budget"]
|
|
assert budget["first_refused"] == "SK1_a2" and "halted" not in budget
|
|
assert [(r["attempt"], r["spent_usd"], r["reservation_usd"], r["waited_sec"]) for r in budget["refusals"]] == [
|
|
("SK1_a2", 11.0, 8.0, 0.0), ("SW1_a2", 13.0, 4.0, 0.0)]
|
|
assert budget["attempts_not_run"] == ["SK1_a2", "SW1_a2"] and budget["spent_usd"] == 13.0
|
|
assert manifest["extra"]["stop_reason"] == "budget_cap" and manifest["extra"]["lanes_run"] == 4
|
|
assert manifest["extra"]["scenarios"]["SW1"] == {"attempts": 2, "passed": 1, "infra_errors": 0, "not_run": 1,
|
|
"verdict": "fail"} and manifest["extra"]["scenarios"]["SM1"]["verdict"] == "pass"
|
|
assert manifest["extra"]["scenarios"]["SK1"]["passed"] == 1 and manifest["extra"]["scenarios"]["SK1"]["not_run"] == 1
|
|
rows = {json.loads(p.read_text(encoding="utf-8"))["attempt"]: json.loads(p.read_text(encoding="utf-8"))
|
|
for p in out.glob("lanes/SM1_*/result.json")}
|
|
assert rows[1]["lane_total_budget_usd"] == 4.0 and rows[2]["lane_total_budget_usd"] == 4.0 # each: its reservation
|
|
refused = json.loads((out / "lanes" / "SK1_a2" / "result.json").read_text(encoding="utf-8"))
|
|
assert refused["status"] == "not_run" and refused["reason_code"] == "budget_cap"
|
|
assert refused["refusal"]["code"] == "budget_cap" and refused["budget"]["waited_sec"] == 0.0
|
|
assert refused["budget"]["spent_usd"] == 11.0 and refused["budget"]["reservation_usd"] == 8.0
|
|
index = [json.loads(ln) for ln in (out / "result_index.jsonl").read_text(encoding="utf-8").splitlines() if ln.strip()]
|
|
assert [(r["instance_id"], r["status"], r["reason_code"]) for r in index if r["status"] == "not_run"] == [
|
|
("SK1_a2", "not_run", "budget_cap"), ("SW1_a2", "not_run", "budget_cap")]
|
|
|
|
|
|
def test_self_mod_run_level_gate_fails_every_unconfirmed_absorbing_lane(tmp_path, monkeypatch):
|
|
"""The gate follows ``expects_absorb``: an unconfirmed SM1 fails the run; SW1 (no absorb to confirm) is never listed."""
|
|
def lane(job, *a, **k):
|
|
row = _fake_lane(job, *a, **k)
|
|
row["self_mod_absorb"] = {"expected": True, "confirmed": False} if job[0] == "SM1" else {"expected": False}
|
|
return row
|
|
|
|
_out, manifest = _fake_run(tmp_path, monkeypatch, ["--self-mod", "--scenarios", "SM1,SW1", "--lanes", "1"],
|
|
lane=lane, expect_rc=1)
|
|
assert manifest["extra"]["self_mod"] == {"lanes": 2, "absorb_expected": 1, "absorb_unconfirmed": ["SM1_a1"]}
|
|
assert manifest["extra"]["outcome"] == "failed" and manifest["extra"]["exit_code"] == 1
|
|
assert manifest["extra"]["scenarios"]["SM1"]["verdict"] == "pass" # the lane verdict alone would have passed
|
|
|
|
|
|
def test_lane_infra_failure_is_a_typed_refusal_in_both_artifacts(tmp_path, monkeypatch):
|
|
_short_tmp(monkeypatch)
|
|
|
|
def refuse(seed, clone):
|
|
raise run_live_lanes.SeedMaterializeRefused("clone_failed", "git clone exploded")
|
|
|
|
monkeypatch.setattr(run_live_lanes, "clone_seed", refuse)
|
|
args = run_live_lanes.parse_args(["--stub", "--out", str(tmp_path / "out"), "--watch-interval", "600"])
|
|
out = tmp_path / "out"
|
|
states: dict = {}
|
|
row = run_live_lanes.run_attempt(("SM1", 1), args, out, {}, run_live_lanes.Stagger(2.0), states, tmp_path / "seed",
|
|
run_live_lanes.RunBudget(100.0, 8.0, reader=lambda root: (0.0, 0)),
|
|
dispatch_index=0, key="", seed_sha="abc")
|
|
assert row["status"] == "infra_error" and row["reason_code"] == "infra_error:clone_failed"
|
|
assert row["refusal"] == {"type": "SeedMaterializeRefused", "code": "clone_failed", "message": "git clone exploded"}
|
|
stored = json.loads((out / "lanes" / "SM1_a1" / "result.json").read_text(encoding="utf-8"))
|
|
assert stored["refusal"]["code"] == "clone_failed" and stored["error"].startswith("SeedMaterializeRefused:")
|
|
index = json.loads((out / "result_index.jsonl").read_text(encoding="utf-8").strip())
|
|
assert index["status"] == "infra_error" and index["reason_code"] == "infra_error:clone_failed"
|
|
assert index["details"]["refusal"]["code"] == "clone_failed"
|
|
|
|
|
|
def test_stagger_gate_spaces_lane_starts(monkeypatch):
|
|
clock = {"t": 100.0}
|
|
slept: list = []
|
|
monkeypatch.setattr(run_live_lanes.time, "monotonic", lambda: clock["t"])
|
|
monkeypatch.setattr(run_live_lanes.time, "sleep", lambda s: slept.append(s))
|
|
gate = run_live_lanes.Stagger(2.5)
|
|
gate.wait_turn()
|
|
gate.wait_turn()
|
|
clock["t"] += 1.0
|
|
gate.wait_turn()
|
|
assert slept == [0.0, 2.5, 1.5]
|
|
|
|
|
|
def test_ui_client_degrades_typed_without_playwright(monkeypatch):
|
|
monkeypatch.setattr(ui_probe, "_suite_client", lambda base_url: None)
|
|
monkeypatch.setitem(sys.modules, "playwright", None)
|
|
monkeypatch.setitem(sys.modules, "playwright.sync_api", None)
|
|
assert ui_probe.resolve_ui_client("http://127.0.0.1:1") == (None, "ui_unavailable:playwright_not_installed")
|
|
|
|
|
|
def test_ui_client_prefers_the_suite_interface_when_it_has_this_surface(monkeypatch):
|
|
class Landed:
|
|
def __init__(self, base_url):
|
|
self.base_url = base_url
|
|
self.opened = False
|
|
|
|
def open(self):
|
|
self.opened = True
|
|
return self
|
|
|
|
goto = computed_property = send_chat = screenshot = rebind = close = lambda self, *a, **k: None
|
|
|
|
fake = type(sys)("tests.system_e2e.interfaces")
|
|
fake.PlaywrightUIClient = Landed
|
|
monkeypatch.setitem(sys.modules, "tests.system_e2e.interfaces", fake)
|
|
client, reason = ui_probe.resolve_ui_client("http://127.0.0.1:1")
|
|
assert isinstance(client, Landed) and client.opened and reason == ""
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# The keyless rehearsal: SM1 end-to-end on a real isolated server (--stub)
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
@pytest.mark.integration
|
|
@pytest.mark.serial
|
|
def test_stub_sm1_end_to_end_on_a_real_isolated_server(tmp_path):
|
|
"""Real server, loopback stub model, no key: the commit lands through the review organ
|
|
(the shared palette plus the release-carrier bump, no skip flags, through the same hermetic
|
|
tests preflight as the paid prompt), the durable rows and receipts exist, the
|
|
seed is a clean detached clone of this tree's HEAD and the manifest names the stub as
|
|
the model."""
|
|
if str(os.environ.get("OUROBOROS_E2E_DEEP") or "").strip().lower() != "mock":
|
|
pytest.skip("set OUROBOROS_E2E_DEEP=mock to run the stub rehearsal (spawns a real isolated server)")
|
|
out = tmp_path / "out"
|
|
rc = run_live_lanes.main(["--stub", "--lanes", "1", "--scenarios", "SM1", "--source-repo", str(REPO_ROOT),
|
|
"--seed", "HEAD", "--out", str(out), "--watch-interval", "600"])
|
|
row = json.loads((out / "lanes" / "SM1_a1" / "result.json").read_text(encoding="utf-8"))
|
|
failed = sorted(k for k, v in row["checks"].items() if not v and not k.startswith("ui_"))
|
|
assert rc == 0 and failed == [], (row["status"], failed, row["error"])
|
|
manifest = json.loads((out / "run_manifest.json").read_text(encoding="utf-8"))
|
|
assert manifest["model_slots"]["OUROBOROS_MODEL"] == stub_lane.STUB_MODEL_SLUG == row["model_slots"]["OUROBOROS_MODEL"]
|
|
assert row["digests"]["pre_head"] == manifest["seed"]["resolved_sha"] != row["digests"]["post_head"]
|
|
assert len(row["digests"]["diff_sha256"]) == 64 and not row["digests"]["seed_describe"].endswith("-dirty")
|
|
assert (out / "result_index.jsonl").read_text(encoding="utf-8").count("\n") == 1
|
|
|
|
|
|
def test_orphan_after_stop_fails_a_passing_lane_with_a_typed_reason(tmp_path):
|
|
"""A process still carrying the lane's data root after stop flips a passing lane to fail with
|
|
reason_code=checks_failed, in result.json AND in result_index.jsonl — never an empty reason."""
|
|
def row():
|
|
return {"scenario": "SM1", "attempt": 1, "status": "pass", "reason_code": "", "checks": {"fake": True},
|
|
"error": "", "duration_sec": 1.0, "budget": {}, "refusal": None, "runtime_outcome": "completed"}
|
|
clean = row()
|
|
run_live_lanes._apply_orphan_scan(clean, [])
|
|
assert clean["status"] == "pass" and clean["reason_code"] == "" and clean["checks"]["no_orphans_after_stop"] is True
|
|
assert "orphans" not in clean
|
|
absent = row()
|
|
run_live_lanes._apply_orphan_scan(absent, None) # no procfs (macOS, Windows): a typed fact, no check
|
|
assert absent["status"] == "pass" and absent["orphan_scan"] == "unavailable:no_procfs"
|
|
assert absent["no_orphans_after_stop"] is None and "no_orphans_after_stop" not in absent["checks"]
|
|
dirty = row()
|
|
run_live_lanes._apply_orphan_scan(dirty, [os.getpid()])
|
|
assert dirty["status"] == "fail" and dirty["reason_code"] == "checks_failed"
|
|
assert dirty["checks"]["no_orphans_after_stop"] is False and dirty["no_orphans_after_stop"] is False
|
|
# The survivors are NAMED: pid + the head of its cmdline (this very interpreter here).
|
|
assert [o["pid"] for o in dirty["orphans"]] == [os.getpid()] and "orphans_omitted" not in dirty
|
|
assert len(dirty["orphans"][0]["cmdline"]) <= 120
|
|
if run_live_lanes.PROCFS_AVAILABLE: # the cmdline text is read from /proc: Linux only; elsewhere it is the typed ""
|
|
assert "python" in dirty["orphans"][0]["cmdline"]
|
|
crowded = row()
|
|
run_live_lanes._apply_orphan_scan(crowded, [os.getpid()] + [2 ** 22 + n for n in range(24)])
|
|
assert len(crowded["orphans"]) == 20 and crowded["orphans_omitted"] == 5
|
|
assert crowded["orphans"][1] == {"pid": 2 ** 22, "cmdline": ""} # a pid gone by read time: typed empty
|
|
out = tmp_path / "run"
|
|
lane = out / "lanes" / "SM1_a1"
|
|
lane.mkdir(parents=True)
|
|
run_live_lanes._record_row(out, lane, dirty)
|
|
recorded = json.loads((lane / "result.json").read_text(encoding="utf-8"))
|
|
assert recorded["status"] == "fail" and recorded["reason_code"] == "checks_failed"
|
|
index = [json.loads(ln) for ln in (out / "result_index.jsonl").read_text(encoding="utf-8").splitlines() if ln.strip()]
|
|
assert [(r["instance_id"], r["status"], r["reason_code"]) for r in index] == [("SM1_a1", "fail", "checks_failed")]
|