mirror of
https://github.com/razzant/ouroboros.git
synced 2026-10-03 04:07:04 +00:00
fix(e2e_live): absorb wait proves no cycle is pending; non-absorbing lanes do not promote
Adversarial review of the corrected --self-mod seeding (2026-09-06) traced the post-task path end to end and found that IsolatedServer.wait_for_absorb ended the wait as no_promotion on ONE idle /api/state sample once the grace had passed: no request file, pending_count 0, running_count 0. Those are exactly the readings of the absorb path itself. A cycle that committed keeps its campaign transaction as waiting_for_restart while the supervisor restarts synchronously (RUNNING already popped, the counter unchanged), and the re-exec'd server answers /api/state with zero counts before its supervisor is up; absorbed_cycles_done moves only when the worker boot verifies the restart. The pre-seeded benchmark campaign used to mask this (its t=0 cycle kept running_count above zero and the kept request file blocked the exit); with the owner-id-only seeding every SM1 lane would have failed a healthy absorb as no_promotion. wait_for_absorb now needs PROOF that no cycle is pending, held on six consecutive polls after the grace: the queue idle AND supervisor_ready AND no post_task_evolution_request.json AND no campaign active_transaction. The reason is typed from the durable campaign state (campaign_summary / absorb_idle_reason): no_promotion (no campaign although the post-task decision ran), no_decision, cycle_no_op, cycle_not_absorbed, campaign_<status>, cycle_not_enqueued; the summary travels in the wait dict the stand records. Benchmark callers (evolve_smoke, the CLB adapter) keep their early exit on a genuinely idle campaign. Two more findings from the same review: scenarios that commit nothing (SW1, SK1) now pin OUROBOROS_POST_TASK_EVOLUTION=false in their lane settings — a one-shot cycle promoted from their own roots could commit and re-exec the server in the middle of the lifecycle under test; and the SK1 gate also requires the review call's own executable_review (a pending duplicate job never passes) while _skill_entry reads the listing through the non-raising helper so a failed listing is a failed check with facts, not an infra_error. Tests: a new module pins the wait on a fake clock (waiting_for_restart and a booting server do not end it, one idle sample does not, a pending request blocks it, the absorb confirms, every idle reason), the SK1 module pins the product gate rule and the per-scenario promotion pin, the runner module's settings pin flips for SW1/SK1. Co-authored-by: Ouroboros <311266734+ouroboros-agent@users.noreply.github.com>
This commit is contained in:
Ouroboros
parent
d70f93c3c1
commit
4f3cda85d5
6 changed files with 243 additions and 39 deletions
|
|
@ -310,6 +310,45 @@ def seed_owner_state(data_root: pathlib.Path, *, evolution_enabled: bool = False
|
|||
state_path.write_text(json.dumps(st), encoding="utf-8")
|
||||
|
||||
|
||||
def campaign_summary(data_root: pathlib.Path) -> dict:
|
||||
"""The durable campaign facts an absorb wait reasons about: evolution_campaign.json (presence, status,
|
||||
source, a pending ``active_transaction``, the newest transaction outcome, the absorbed counter) and the
|
||||
post-task promotion counter (``post_task_evolution_counter.json``: the decision ran at least once)."""
|
||||
state_dir = pathlib.Path(data_root) / "state"
|
||||
try:
|
||||
campaign = json.loads((state_dir / "evolution_campaign.json").read_text(encoding="utf-8"))
|
||||
except (OSError, ValueError):
|
||||
campaign = {}
|
||||
campaign = campaign if isinstance(campaign, dict) else {}
|
||||
try:
|
||||
counter = int(json.loads((state_dir / "post_task_evolution_counter.json").read_text(encoding="utf-8"))["n"])
|
||||
except (OSError, ValueError, TypeError, KeyError):
|
||||
counter = 0
|
||||
history = [tx for tx in (campaign.get("transaction_history") or []) if isinstance(tx, dict)]
|
||||
return {"present": bool(campaign), "status": str(campaign.get("status") or ""),
|
||||
"source": str(campaign.get("source") or ""),
|
||||
"active_transaction": isinstance(campaign.get("active_transaction"), dict),
|
||||
"absorbed_cycles_done": int(campaign.get("absorbed_cycles_done") or 0),
|
||||
"newest_outcome": str(history[-1].get("cycle_outcome") or "") if history else "",
|
||||
"post_task_counter": counter}
|
||||
|
||||
|
||||
def absorb_idle_reason(campaign: dict) -> str:
|
||||
"""Typed non-confirmation of an idle lane (``IsolatedServer.wait_for_absorb``): ``no_promotion`` (no
|
||||
campaign although the post-task decision ran), ``no_decision`` (no campaign, no decision), ``cycle_no_op``
|
||||
(the newest cycle committed nothing), ``cycle_not_absorbed`` (a cycle ended otherwise without an absorb),
|
||||
``campaign_<status>`` (paused/stopped/completed) or ``cycle_not_enqueued`` (active campaign, no cycle)."""
|
||||
if not campaign.get("present"):
|
||||
return "no_promotion" if campaign.get("post_task_counter") else "no_decision"
|
||||
if campaign.get("newest_outcome") == "no_op":
|
||||
return "cycle_no_op"
|
||||
if campaign.get("newest_outcome"):
|
||||
return "cycle_not_absorbed"
|
||||
if campaign.get("status") in ("paused", "stopped", "completed"):
|
||||
return f"campaign_{campaign['status']}"
|
||||
return "cycle_not_enqueued"
|
||||
|
||||
|
||||
def absorbed_cycles_done(data_root: pathlib.Path) -> int:
|
||||
"""Read absorbed self-evolution cycle count from evolution_campaign.json."""
|
||||
path = pathlib.Path(data_root) / "state" / "evolution_campaign.json"
|
||||
|
|
@ -647,33 +686,44 @@ class IsolatedServer:
|
|||
return False
|
||||
|
||||
def wait_for_absorb(self, prev_sha: str, prev_absorbed: int, timeout: float = 1800,
|
||||
idle_grace: float = 90) -> dict:
|
||||
"""Between instances, wait for an absorbed self-evolution cycle: the server
|
||||
re-execs onto a new SHA and `absorbed_cycles_done` increments. Returns
|
||||
{absorbed, new_sha, cycles, reason}. When the LLM legitimately declines to
|
||||
promote (the common path), this returns absorbed=False EARLY — once the queue
|
||||
is idle, no post_task_evolution_request.json is pending, and no cycle absorbed
|
||||
within a short grace — instead of stalling the full timeout."""
|
||||
idle_grace: float = 90, idle_polls: int = 6) -> dict:
|
||||
"""Between instances, wait for an absorbed self-evolution cycle: the server re-execs onto a
|
||||
new SHA and ``absorbed_cycles_done`` increments. Returns ``{absorbed, new_sha, cycles, reason,
|
||||
campaign}``. An EARLY ``absorbed=False`` needs PROOF that no cycle is pending, held on
|
||||
``idle_polls`` consecutive polls after ``idle_grace``: the queue idle AND ``supervisor_ready``
|
||||
AND no ``post_task_evolution_request.json`` AND no campaign ``active_transaction``. One idle
|
||||
sample is not proof: a cycle that committed keeps its transaction as ``waiting_for_restart``
|
||||
while the supervisor restarts synchronously (queue empty, counter unchanged), and the re-exec'd
|
||||
server answers ``/api/state`` with zero counts before its supervisor is up — the counter moves
|
||||
only when the worker boot verifies the restart (rc.15 stand, adversarial finding of 2026-09-06).
|
||||
The typed reason is what the durable campaign state proves (``absorb_idle_reason``)."""
|
||||
deadline = time.time() + timeout
|
||||
start = time.time()
|
||||
request_path = self.data_root / "state" / "post_task_evolution_request.json"
|
||||
idle_streak = 0
|
||||
while time.time() < deadline:
|
||||
cycles = absorbed_cycles_done(self.data_root)
|
||||
sha = self.current_sha()
|
||||
if cycles > prev_absorbed and sha and sha != prev_sha:
|
||||
self.wait_for_health(timeout=180)
|
||||
return {"absorbed": True, "new_sha": sha, "cycles": cycles, "reason": "absorbed"}
|
||||
return {"absorbed": True, "new_sha": sha, "cycles": cycles, "reason": "absorbed",
|
||||
"campaign": campaign_summary(self.data_root)}
|
||||
if time.time() - start > idle_grace and cycles == prev_absorbed:
|
||||
campaign = campaign_summary(self.data_root)
|
||||
try:
|
||||
st = self._state(timeout=5)
|
||||
idle = int(st.get("pending_count") or 0) == 0 and int(st.get("running_count") or 0) == 0
|
||||
idle = (int(st.get("pending_count") or 0) == 0 and int(st.get("running_count") or 0) == 0
|
||||
and bool(st.get("supervisor_ready")))
|
||||
except (urllib.error.URLError, OSError, ValueError):
|
||||
idle = False
|
||||
if idle and not request_path.exists():
|
||||
return {"absorbed": False, "new_sha": sha, "cycles": cycles, "reason": "no_promotion"}
|
||||
idle = idle and not request_path.exists() and not campaign["active_transaction"]
|
||||
idle_streak = idle_streak + 1 if idle else 0
|
||||
if idle_streak >= max(1, int(idle_polls)):
|
||||
return {"absorbed": False, "new_sha": sha, "cycles": cycles,
|
||||
"reason": absorb_idle_reason(campaign), "campaign": campaign}
|
||||
time.sleep(5)
|
||||
return {"absorbed": False, "new_sha": self.current_sha(),
|
||||
"cycles": absorbed_cycles_done(self.data_root), "reason": "timeout"}
|
||||
return {"absorbed": False, "new_sha": self.current_sha(), "cycles": absorbed_cycles_done(self.data_root),
|
||||
"reason": "timeout", "campaign": campaign_summary(self.data_root)}
|
||||
|
||||
def stop(self) -> None:
|
||||
if self.proc is not None and self.proc.poll() is None:
|
||||
|
|
|
|||
|
|
@ -765,7 +765,10 @@ def sk1_prompt() -> str:
|
|||
|
||||
|
||||
def _skill_entry(base_url: str, name: str) -> dict:
|
||||
listing = _api(base_url, "GET", "/api/extensions", timeout=30)
|
||||
"""The skill's ``/api/extensions`` row, or ``{}`` — a listing that fails is a failed check with facts
|
||||
(``review_executable``/``live_loaded`` absent), never an infra_error that aborts the lifecycle."""
|
||||
resp = _api_status(base_url, "GET", "/api/extensions", None, timeout=30)
|
||||
listing = resp["body"] if resp["status"] == 200 else {}
|
||||
rows = listing if isinstance(listing, list) else (listing.get("extensions") or listing.get("skills") or [])
|
||||
return next((row for row in rows if isinstance(row, dict) and row.get("name") == name), {})
|
||||
|
||||
|
|
@ -775,11 +778,17 @@ def sk1_review_gate(review: dict, entry: dict, findings: list) -> tuple[bool, di
|
|||
recorded findings) and the ``/api/extensions`` row says ``executable_review`` — clean, warnings, or blockers
|
||||
under advisory enforcement by operator choice (``skill_review_gate``). A clean all-PASS review is a recorded
|
||||
FACT, not the verdict: the rc.15 SK1 rerun on 560f7d71 authored one clean, one warnings and one blockers
|
||||
payload with every other lifecycle check green, so all-PASS measured the author model, not the product."""
|
||||
payload with every other lifecycle check green, so all-PASS measured the author model, not the product.
|
||||
The verdict also needs the review call itself to answer 200 with its own ``executable_review`` (the
|
||||
lifecycle's gate; a ``pending`` duplicate job never passes) and persisted findings (a review really ran).
|
||||
The SK1 lane sets no enforcement, so it runs the tree default (advisory today) under both profiles: the
|
||||
``blockers under blocking enforcement`` branch is the product's rule, not a path the stand exercises."""
|
||||
gate = entry.get("review_gate") if isinstance(entry.get("review_gate"), dict) else {}
|
||||
failed = [f.get("item") for f in findings if str(f.get("verdict") or "") != "PASS"]
|
||||
ok = review["status"] == 200 and entry.get("executable_review") is True and bool(findings)
|
||||
ok = (review["status"] == 200 and review["body"].get("executable_review") is True
|
||||
and entry.get("executable_review") is True and bool(findings))
|
||||
return ok, {"review_status": review["body"].get("status"), "review_executable": entry.get("executable_review"),
|
||||
"review_body_executable": review["body"].get("executable_review"),
|
||||
"review_enforcement": gate.get("review_enforcement"), "review_blocking_reason": gate.get("blocking_reason"),
|
||||
"findings": len(findings), "findings_failed": failed, "review_clean": bool(findings) and not failed}
|
||||
|
||||
|
|
@ -880,6 +889,11 @@ class Scenario:
|
|||
out = dict(self.settings_overrides)
|
||||
if self.id == "SW1":
|
||||
out["OUROBOROS_SUBAGENTS"] = sw1_roster(model)
|
||||
if not self.expects_absorb:
|
||||
# A lane that commits nothing must not promote either: under --self-mod its one-shot cycle could
|
||||
# commit and re-exec the server in the middle of the lifecycle under test (SK1 review/grants/
|
||||
# dispatch), turning an unrelated restart into the lane's verdict. Only SM1 exercises evolution.
|
||||
out["OUROBOROS_POST_TASK_EVOLUTION"] = "false"
|
||||
return out
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -1274,10 +1274,11 @@ computed style read by a browser from the COMMITTED CSS after a restart), SW1
|
|||
(the Swarm button arms `force_plan`; at least two children with causal lineage,
|
||||
the `swarm_fanout` receipt, the with-children cost rollup, no orphan process by
|
||||
the `/proc` environ scan), SK1 (the model authors `SKILL.md`+`plugin.py` and runs
|
||||
`skill_preflight`; the runner reviews — the verdict is the product gate, the
|
||||
`/api/extensions` row's `executable_review` (clean, warnings, or blockers under advisory
|
||||
enforcement), with the review status, non-PASS items and the clean state recorded as
|
||||
facts — grants exactly the manifest's one
|
||||
`skill_preflight`; the runner reviews — the verdict is the product gate: the review call
|
||||
answers 200 with its own `executable_review`, the `/api/extensions` row says
|
||||
`executable_review` (clean, warnings, or blockers under advisory enforcement — the SK1 lane
|
||||
sets none, so it runs the tree default) and findings were persisted; the status, enforcement,
|
||||
blocking reason, non-PASS items and the clean state are recorded facts — grants exactly the manifest's one
|
||||
privileged permission, enables, dispatches, deletes; the author and dispatch
|
||||
tasks keep separate `author_*`/`dispatch_*` terminal checks, and the dispatch
|
||||
counts only on a tools.jsonl row with typed status `ok` and the extension's exact
|
||||
|
|
@ -1378,10 +1379,18 @@ t=0, gets the promotion refused (evolution already enabled) and its kept request
|
|||
`expects_absorb` (SM1, the one that lands a commit): a pre-task snapshot
|
||||
(clone HEAD, served sha, uptime, absorbed-cycle counter) and, afterwards, the
|
||||
counter advanced, the served sha moved, the uptime reset and the server ready;
|
||||
anything less is a typed `self_mod_absorb_confirmed=false` and the run fails.
|
||||
anything less is a typed `self_mod_absorb_confirmed=false` and the run fails. The
|
||||
wait (`IsolatedServer.wait_for_absorb`) ends early only with proof that no cycle is
|
||||
pending — six consecutive polls with the queue idle, `supervisor_ready`, no promotion
|
||||
request and no campaign `active_transaction` (a committed cycle keeps its transaction
|
||||
as `waiting_for_restart` through the synchronous restart and the re-exec boot, when a
|
||||
single idle sample looks exactly like a declined promotion) — and types the reason from
|
||||
the durable campaign state (`no_promotion`, `no_decision`, `cycle_no_op`,
|
||||
`cycle_not_absorbed`, `campaign_<status>`, `cycle_not_enqueued`).
|
||||
SW1/SK1 commit nothing, so under `--self-mod` they stop their server right after
|
||||
the scenario with `self_mod_absorb: {"expected": false}` and no absorb check
|
||||
(evolution stays on in their settings; the stand just does not wait for it) —
|
||||
the scenario with `self_mod_absorb: {"expected": false}` and no absorb check, and their
|
||||
lane settings pin `OUROBOROS_POST_TASK_EVOLUTION=false` (a one-shot cycle promoted from
|
||||
their own roots could re-exec the server inside the lifecycle under test) —
|
||||
the rc.15 paid stand (2026-09-05) had SK1_a1 pass twelve of its thirteen lifecycle
|
||||
checks (the thirteenth was a real reviewer finding on the model-authored plugin),
|
||||
then wait about 27 minutes for a promotion nothing had committed before the
|
||||
|
|
|
|||
|
|
@ -874,7 +874,7 @@ def test_absorb_wait_and_check_follow_the_scenarios_expects_absorb(tmp_path, mon
|
|||
"""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 ON in their settings; every lane seeds ``owner_chat_id`` ONLY, never a campaign (run2's t=0 cycles)."""
|
||||
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})
|
||||
|
|
@ -890,7 +890,7 @@ def test_absorb_wait_and_check_follow_the_scenarios_expects_absorb(tmp_path, mon
|
|||
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"
|
||||
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
|
||||
|
||||
|
|
|
|||
|
|
@ -51,25 +51,48 @@ def test_the_probe_accepts_the_loopback_base_and_only_then_reaches_the_transport
|
|||
|
||||
|
||||
_FINDINGS = [{"item": "manifest_schema", "verdict": "PASS"}, {"item": "bug_hunting", "verdict": "FAIL"}]
|
||||
_CLEAN = [{"item": "a", "verdict": "PASS"}]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("status,http,executable,findings,expected", [
|
||||
("clean", 200, True, [{"item": "a", "verdict": "PASS"}], True),
|
||||
("warnings", 200, True, _FINDINGS, True), # rc.15 SK1_a2: warnings are executable
|
||||
("blockers", 200, True, _FINDINGS, True), # rc.15 SK1_a3: blockers executable under advisory enforcement
|
||||
("blockers", 200, False, _FINDINGS, False), # the same review under blocking enforcement
|
||||
("clean", 500, True, [{"item": "a", "verdict": "PASS"}], False), # the review call itself failed
|
||||
("clean", 200, True, [], False), # no recorded findings: no review actually ran
|
||||
("clean", 200, None, [{"item": "a", "verdict": "PASS"}], False), # the entry carries no gate fact
|
||||
@pytest.mark.parametrize("status,http,body_exec,executable,findings,expected", [
|
||||
("clean", 200, True, True, _CLEAN, True),
|
||||
("warnings", 200, True, True, _FINDINGS, True), # rc.15 SK1_a2: warnings are executable
|
||||
("blockers", 200, True, True, _FINDINGS, True), # rc.15 SK1_a3: blockers executable under advisory
|
||||
("blockers", 200, False, False, _FINDINGS, False), # the same review under blocking enforcement
|
||||
("pending", 200, False, True, _CLEAN, False), # duplicate job: the call's own gate says no
|
||||
("clean", 500, True, True, _CLEAN, False), # the review call itself failed
|
||||
("clean", 200, True, True, [], False), # no recorded findings: no review actually ran
|
||||
("clean", 200, True, None, _CLEAN, False), # the /api/extensions row carries no gate fact
|
||||
])
|
||||
def test_sk1_review_verdict_is_the_product_gate_and_records_the_clean_state_as_a_fact(status, http, executable,
|
||||
findings, expected):
|
||||
review = {"status": http, "body": {"status": status}}
|
||||
def test_sk1_review_verdict_is_the_product_gate_and_records_the_clean_state_as_a_fact(status, http, body_exec,
|
||||
executable, findings, expected):
|
||||
review = {"status": http, "body": {"status": status, "executable_review": body_exec}}
|
||||
entry = {"executable_review": executable,
|
||||
"review_gate": {"review_enforcement": "advisory", "blocking_reason": "x"}}
|
||||
ok, facts = scenarios.sk1_review_gate(review, entry, findings)
|
||||
assert ok is expected
|
||||
failed = [f["item"] for f in findings if f["verdict"] != "PASS"]
|
||||
assert facts == {"review_status": status, "review_executable": executable, "review_enforcement": "advisory",
|
||||
"review_blocking_reason": "x", "findings": len(findings), "findings_failed": failed,
|
||||
"review_clean": bool(findings) and not failed}
|
||||
assert facts == {"review_status": status, "review_executable": executable, "review_body_executable": body_exec,
|
||||
"review_enforcement": "advisory", "review_blocking_reason": "x", "findings": len(findings),
|
||||
"findings_failed": failed, "review_clean": bool(findings) and not failed}
|
||||
|
||||
|
||||
def test_the_product_gate_the_stand_relies_on_executes_warnings_and_advisory_blockers_only():
|
||||
"""The contract behind ``sk1_review_gate``: ``skill_review_gate`` (the same rule the ``/api/extensions``
|
||||
row and the review call project as ``executable_review``)."""
|
||||
from ouroboros.skill_review_status import skill_review_gate
|
||||
|
||||
assert skill_review_gate("clean", enforcement="blocking")["executable_review"] is True
|
||||
assert skill_review_gate("warnings", enforcement="blocking")["executable_review"] is True
|
||||
assert skill_review_gate("blockers", enforcement="advisory")["executable_review"] is True
|
||||
assert skill_review_gate("blockers", enforcement="blocking")["executable_review"] is False
|
||||
assert skill_review_gate("pending", enforcement="advisory")["executable_review"] is False
|
||||
|
||||
|
||||
def test_only_the_absorbing_scenario_promotes_post_task_evolution():
|
||||
"""Under ``--self-mod`` SW1/SK1 would otherwise promote from their own roots and their one-shot cycle could
|
||||
re-exec the server inside the lifecycle under test; the lane overrides pin post-task evolution off for
|
||||
every scenario that commits nothing, and leave SM1 (``expects_absorb``) to the run-level setting."""
|
||||
for sid, row in scenarios.SCENARIOS.items():
|
||||
applied = row.overrides("stub/model").get("OUROBOROS_POST_TASK_EVOLUTION")
|
||||
assert applied == (None if row.expects_absorb else "false"), (sid, applied)
|
||||
|
|
|
|||
108
tests/test_server_runner_absorb_wait.py
Normal file
108
tests/test_server_runner_absorb_wait.py
Normal file
|
|
@ -0,0 +1,108 @@
|
|||
"""``IsolatedServer.wait_for_absorb`` may answer ``absorbed=False`` early only with PROOF that no cycle is
|
||||
pending (rc.15 stand, adversarial finding of 2026-09-06): a cycle that committed keeps its campaign transaction
|
||||
as ``waiting_for_restart`` through the synchronous supervisor restart (queue idle, counter unchanged) and the
|
||||
re-exec'd server answers ``/api/state`` with zero counts before its supervisor is ready; one idle sample used to
|
||||
end the wait as ``no_promotion`` right there. The typed idle reasons come from the durable campaign state."""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import pathlib
|
||||
|
||||
import pytest
|
||||
|
||||
from devtools.benchmarks.common import server_runner
|
||||
|
||||
|
||||
class _Clock:
|
||||
def __init__(self) -> None:
|
||||
self.now = 0.0
|
||||
|
||||
def time(self) -> float:
|
||||
return self.now
|
||||
|
||||
def sleep(self, seconds: float) -> None:
|
||||
self.now += float(seconds)
|
||||
|
||||
|
||||
def _server(tmp_path: pathlib.Path, states: list, monkeypatch, *, sha: str = "aaaaaaaa") -> server_runner.IsolatedServer:
|
||||
"""``states`` feeds the idle poll only (one read per poll after the grace); the served sha is fixed."""
|
||||
clock = _Clock()
|
||||
monkeypatch.setattr(server_runner, "time", clock)
|
||||
srv = object.__new__(server_runner.IsolatedServer)
|
||||
srv.data_root = tmp_path
|
||||
srv.base_url = "http://127.0.0.1:1"
|
||||
calls = {"n": 0}
|
||||
|
||||
def _state(timeout: float = 5) -> dict:
|
||||
st = states[min(calls["n"], len(states) - 1)]
|
||||
calls["n"] += 1
|
||||
if isinstance(st, Exception):
|
||||
raise st
|
||||
return st
|
||||
|
||||
srv._state = _state
|
||||
srv.current_sha = lambda: sha
|
||||
srv.wait_for_health = lambda timeout=180: True
|
||||
return srv
|
||||
|
||||
|
||||
def _campaign(root: pathlib.Path, **fields) -> None:
|
||||
(root / "state").mkdir(parents=True, exist_ok=True)
|
||||
(root / "state" / "evolution_campaign.json").write_text(json.dumps({"schema_version": 1, **fields}), encoding="utf-8")
|
||||
|
||||
|
||||
IDLE = {"sha": "aaaaaaaa", "pending_count": 0, "running_count": 0, "supervisor_ready": True}
|
||||
BOOTING = {"sha": "", "pending_count": 0, "running_count": 0, "supervisor_ready": False}
|
||||
BUSY = {"sha": "aaaaaaaa", "pending_count": 0, "running_count": 1, "supervisor_ready": True}
|
||||
|
||||
|
||||
def test_an_idle_queue_with_a_transaction_waiting_for_restart_is_not_a_declined_promotion(tmp_path, monkeypatch):
|
||||
_campaign(tmp_path, status="active", source="post_task", absorbed_cycles_done=0,
|
||||
active_transaction={"commit_sha": "c" * 40, "cycle_outcome": "waiting_for_restart"})
|
||||
srv = _server(tmp_path, [IDLE], monkeypatch)
|
||||
out = srv.wait_for_absorb("aaaaaaaa", 0, timeout=200, idle_grace=10, idle_polls=2)
|
||||
assert out["reason"] == "timeout" and out["absorbed"] is False and out["campaign"]["active_transaction"] is True
|
||||
|
||||
|
||||
def test_the_booting_reexecd_server_and_a_single_idle_sample_do_not_end_the_wait(tmp_path, monkeypatch):
|
||||
_campaign(tmp_path, status="active", source="post_task", absorbed_cycles_done=0, transaction_history=[])
|
||||
# after the grace: booting, idle, busy, idle, idle -> the streak of two idle polls only forms on the fifth poll
|
||||
srv = _server(tmp_path, [BOOTING, IDLE, BUSY, IDLE, IDLE], monkeypatch)
|
||||
out = srv.wait_for_absorb("aaaaaaaa", 0, timeout=10_000, idle_grace=10, idle_polls=2)
|
||||
assert out["reason"] == "cycle_not_enqueued" and out["absorbed"] is False
|
||||
assert server_runner.time.now == 35 # polls at t=15,20,25,30,35: the streak completes on the fifth
|
||||
|
||||
|
||||
def test_a_pending_request_file_blocks_the_early_exit(tmp_path, monkeypatch):
|
||||
(tmp_path / "state").mkdir()
|
||||
(tmp_path / "state" / "post_task_evolution_request.json").write_text("{}", encoding="utf-8")
|
||||
srv = _server(tmp_path, [IDLE], monkeypatch)
|
||||
assert srv.wait_for_absorb("aaaaaaaa", 0, timeout=100, idle_grace=10, idle_polls=1)["reason"] == "timeout"
|
||||
|
||||
|
||||
def test_the_absorb_is_confirmed_when_the_counter_and_the_served_sha_move(tmp_path, monkeypatch):
|
||||
_campaign(tmp_path, status="active", source="post_task", absorbed_cycles_done=1,
|
||||
transaction_history=[{"cycle_outcome": "absorbed"}])
|
||||
srv = _server(tmp_path, [IDLE], monkeypatch, sha="bbbbbbbb")
|
||||
out = srv.wait_for_absorb("aaaaaaaa", 0, timeout=100, idle_grace=10)
|
||||
assert out["absorbed"] is True and out["new_sha"] == "bbbbbbbb" and out["campaign"]["newest_outcome"] == "absorbed"
|
||||
|
||||
|
||||
@pytest.mark.parametrize("campaign,counter,expected", [
|
||||
(None, 1, "no_promotion"), # decision ran, nothing promoted
|
||||
(None, 0, "no_decision"), # the post-task decision never ran
|
||||
({"status": "active", "transaction_history": [{"cycle_outcome": "no_op"}]}, 1, "cycle_no_op"),
|
||||
({"status": "active", "transaction_history": [{"cycle_outcome": "abandoned"}]}, 1, "cycle_not_absorbed"),
|
||||
({"status": "paused"}, 1, "campaign_paused"),
|
||||
({"status": "active"}, 1, "cycle_not_enqueued"),
|
||||
])
|
||||
def test_idle_reasons_are_typed_from_the_durable_campaign_state(tmp_path, monkeypatch, campaign, counter, expected):
|
||||
if campaign is not None:
|
||||
_campaign(tmp_path, **campaign)
|
||||
(tmp_path / "state").mkdir(exist_ok=True)
|
||||
if counter:
|
||||
(tmp_path / "state" / "post_task_evolution_counter.json").write_text(json.dumps({"n": counter}), encoding="utf-8")
|
||||
srv = _server(tmp_path, [IDLE], monkeypatch)
|
||||
out = srv.wait_for_absorb("aaaaaaaa", 0, timeout=1_000, idle_grace=10, idle_polls=3)
|
||||
assert out["reason"] == expected and out["absorbed"] is False
|
||||
assert out["campaign"]["post_task_counter"] == counter and out["campaign"]["present"] is (campaign is not None)
|
||||
Loading…
Add table
Add a link
Reference in a new issue