mirror of
https://github.com/razzant/ouroboros.git
synced 2026-10-03 04:07:04 +00:00
7ae0dbe3fd
Owner batch #6 item 3=A ordered two SSOT fixes; only the first (the flat timeout pair, D04) landed. supervisor/workers.py declared its own TOTAL_BUDGET_LIMIT and workers.init accepted a total_budget_limit to write it, and nothing read the result — not the module, not a caller, not a test. An accepted argument that is written to a global nobody reads is worse than a dead constant: it reads as a live tunable to anyone grepping for the budget, and server.py fed it the real TOTAL_BUDGET on every boot. The two copies that ARE read stay untouched: supervisor.state (the authority budget_remaining and every dispatch gate read, hot-reloaded through set_budget_limit) and supervisor.message_bus (the reporting plane). Pin: tests/test_settings_budget_hotreload.py — the module carries no such attribute, init no longer asks for the value, the source no longer names it, and the one authority still feeds budget_remaining. Red before this commit on the attribute assertion.
320 lines
12 KiB
Python
320 lines
12 KiB
Python
"""What the evolution scheduler enqueues, replaces and refuses, and what assignment dispatches.
|
|
|
|
Split out of ``tests/test_evolution_state_integrity_v3.py`` by theme: the bare flag with no
|
|
campaign, the active campaign with no source, the owner resume that repairs a legacy
|
|
source, the transaction attach and its owner-stop recheck, the uncommitted transaction
|
|
replaced only when no worker is reaping, and the exact claim assignment must see.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from types import SimpleNamespace
|
|
|
|
from tests._evolution_state_shared import (
|
|
_CaptureQueue,
|
|
_active_transaction,
|
|
)
|
|
|
|
|
|
def _assignment_case(tmp_path, monkeypatch, task_id="assign-evo"):
|
|
from supervisor import evolution_lifecycle, queue, state, workers
|
|
|
|
state.init(tmp_path)
|
|
monkeypatch.setattr(state, "TOTAL_BUDGET_LIMIT", 0.0)
|
|
pending, running = [], {}
|
|
monkeypatch.setattr(workers, "PENDING", pending)
|
|
monkeypatch.setattr(workers, "RUNNING", running)
|
|
workers.init(tmp_path, tmp_path, 1)
|
|
campaign = evolution_lifecycle.start_evolution_campaign("Improve", source="test")
|
|
state.update_state(lambda live: live.update(
|
|
evolution_mode_enabled=True,
|
|
evolution_owner_stopped=False,
|
|
))
|
|
tx = evolution_lifecycle.begin_evolution_transaction(task_id, cycle=1, campaign=campaign)
|
|
task = {
|
|
"id": task_id,
|
|
"type": "evolution",
|
|
"text": "Improve",
|
|
"metadata": {"evolution_transaction": dict(tx)},
|
|
}
|
|
pending.append(task)
|
|
inbox, events = _CaptureQueue(), _CaptureQueue()
|
|
worker = SimpleNamespace(wid=1, busy_task_id=None, reaping=False, in_q=inbox)
|
|
monkeypatch.setattr(workers, "WORKERS", {1: worker})
|
|
monkeypatch.setattr(workers, "get_event_q", lambda: events)
|
|
monkeypatch.setattr(queue, "persist_queue_snapshot", lambda reason="": None)
|
|
monkeypatch.setattr(evolution_lifecycle, "evolution_block_reason", lambda: "")
|
|
return workers, task, tx, worker, inbox, events
|
|
|
|
|
|
def test_scheduler_disables_a_bare_flag_without_campaign(tmp_path, monkeypatch):
|
|
from supervisor import queue, state
|
|
|
|
state.init(tmp_path)
|
|
queue.init(tmp_path)
|
|
pending = []
|
|
queue.init_queue_refs(pending, {}, {"value": 0})
|
|
live = state.load_state()
|
|
live.update({
|
|
"owner_chat_id": 1,
|
|
"evolution_mode_enabled": True,
|
|
"post_task_autostop": True,
|
|
})
|
|
state.save_state(live)
|
|
sent = []
|
|
monkeypatch.setattr(queue, "send_with_budget", lambda *args, **kwargs: sent.append(args[1]))
|
|
|
|
queue.enqueue_evolution_task_if_needed()
|
|
|
|
assert pending == []
|
|
assert state.load_state()["evolution_mode_enabled"] is False
|
|
assert state.load_state()["post_task_autostop"] is False
|
|
assert "active campaign authority" in sent[0]
|
|
event = json.loads((tmp_path / "logs" / "events.jsonl").read_text().splitlines()[-1])
|
|
assert event["type"] == "evolution_authority_missing"
|
|
|
|
|
|
def test_scheduler_refuses_active_campaign_without_source(tmp_path, monkeypatch):
|
|
from supervisor import evolution_lifecycle, queue, state
|
|
|
|
state.init(tmp_path)
|
|
queue.init(tmp_path)
|
|
queue.init_queue_refs([], {}, {"value": 0})
|
|
campaign = evolution_lifecycle.start_evolution_campaign("Improve", source="test")
|
|
campaign.pop("source")
|
|
assert evolution_lifecycle._write_evolution_campaign(campaign) is True
|
|
live = state.load_state()
|
|
live.update({"owner_chat_id": 1, "evolution_mode_enabled": True})
|
|
state.save_state(live)
|
|
monkeypatch.setattr(queue, "send_with_budget", lambda *a, **k: None)
|
|
|
|
queue.enqueue_evolution_task_if_needed()
|
|
|
|
assert state.load_state()["evolution_mode_enabled"] is False
|
|
|
|
|
|
def test_owner_resume_repairs_missing_legacy_campaign_source(tmp_path):
|
|
from supervisor import evolution_lifecycle, queue, state
|
|
|
|
state.init(tmp_path)
|
|
queue.init(tmp_path)
|
|
campaign = evolution_lifecycle.start_evolution_campaign("Improve", source="test")
|
|
campaign["status"] = "paused"
|
|
campaign.pop("source")
|
|
assert evolution_lifecycle._write_evolution_campaign(campaign) is True
|
|
|
|
resumed = evolution_lifecycle.start_evolution_campaign("", source="owner_chat")
|
|
|
|
assert resumed["status"] == "active"
|
|
assert resumed["source"] == "owner_chat"
|
|
|
|
|
|
def test_scheduler_does_not_enqueue_when_transaction_attach_fails(tmp_path, monkeypatch):
|
|
from supervisor import evolution_lifecycle, queue, state
|
|
|
|
state.init(tmp_path)
|
|
queue.init(tmp_path)
|
|
pending = []
|
|
queue.init_queue_refs(pending, {}, {"value": 0})
|
|
evolution_lifecycle.start_evolution_campaign("Improve", source="test")
|
|
live = state.load_state()
|
|
live.update({"owner_chat_id": 1, "evolution_mode_enabled": True})
|
|
state.save_state(live)
|
|
monkeypatch.setattr(queue, "begin_evolution_transaction", lambda *a, **k: {})
|
|
monkeypatch.setattr(queue, "send_with_budget", lambda *a, **k: None)
|
|
|
|
queue.enqueue_evolution_task_if_needed()
|
|
|
|
assert pending == []
|
|
assert state.load_state()["evolution_mode_enabled"] is False
|
|
|
|
|
|
def test_transaction_attach_rechecks_owner_stop_under_state_lock(tmp_path):
|
|
from supervisor import evolution_lifecycle, queue, state
|
|
|
|
state.init(tmp_path)
|
|
queue.init(tmp_path)
|
|
campaign = evolution_lifecycle.start_evolution_campaign("Improve", source="test")
|
|
live = state.load_state()
|
|
live.update({"evolution_mode_enabled": False, "evolution_owner_stopped": True})
|
|
state.save_state(live)
|
|
|
|
tx = evolution_lifecycle.begin_evolution_transaction(
|
|
"too-late", cycle=1, campaign=campaign,
|
|
)
|
|
|
|
assert tx == {}
|
|
assert "active_transaction" not in evolution_lifecycle._read_evolution_campaign()
|
|
|
|
|
|
def test_scheduler_replaces_uncommitted_transaction_lost_before_enqueue(tmp_path, monkeypatch):
|
|
from supervisor import evolution_lifecycle, queue, state
|
|
|
|
state.init(tmp_path)
|
|
queue.init(tmp_path)
|
|
pending = []
|
|
queue.init_queue_refs(pending, {}, {"value": 0})
|
|
campaign = evolution_lifecycle.start_evolution_campaign("Improve", source="test")
|
|
live = state.load_state()
|
|
live.update({"owner_chat_id": 1, "evolution_mode_enabled": True})
|
|
state.save_state(live)
|
|
lost = evolution_lifecycle.begin_evolution_transaction(
|
|
"lost-before-enqueue", cycle=1, campaign=campaign,
|
|
)
|
|
monkeypatch.setattr(queue, "send_with_budget", lambda *a, **k: None)
|
|
|
|
queue.enqueue_evolution_task_if_needed()
|
|
|
|
assert len(pending) == 1
|
|
replacement = pending[0]["metadata"]["evolution_transaction"]
|
|
assert replacement["transaction_id"] != lost["transaction_id"]
|
|
stored = evolution_lifecycle._read_evolution_campaign()
|
|
assert stored["active_transaction"]["transaction_id"] == replacement["transaction_id"]
|
|
assert stored["transaction_history"][-1]["abandoned_reason"] == "dispatch_not_persisted"
|
|
|
|
|
|
def test_scheduler_does_not_replace_transaction_while_worker_is_reaping(tmp_path, monkeypatch):
|
|
from supervisor import evolution_lifecycle, queue
|
|
|
|
_campaign, tx = _active_transaction(tmp_path, task_id="reaping-evolution")
|
|
pending = []
|
|
queue.init_queue_refs(pending, {}, {"value": 0})
|
|
assert evolution_lifecycle.update_evolution_transaction(
|
|
tx["task_id"], dispatch_status="reaping",
|
|
)
|
|
monkeypatch.setattr(queue, "send_with_budget", lambda *args, **kwargs: None)
|
|
|
|
queue.enqueue_evolution_task_if_needed()
|
|
|
|
assert pending == []
|
|
stored = evolution_lifecycle._read_evolution_campaign()["active_transaction"]
|
|
assert stored["transaction_id"] == tx["transaction_id"]
|
|
assert stored["dispatch_status"] == "reaping"
|
|
|
|
|
|
def test_timeout_marks_evolution_reaping_before_scheduler_can_replace_it(tmp_path, monkeypatch):
|
|
from supervisor import evolution_lifecycle, queue
|
|
|
|
_campaign, tx = _active_transaction(tmp_path, task_id="timeout-evolution")
|
|
pending = []
|
|
running = {
|
|
tx["task_id"]: {
|
|
"task": {
|
|
"id": tx["task_id"],
|
|
"type": "evolution",
|
|
"chat_id": 1,
|
|
"metadata": {"evolution_transaction": dict(tx)},
|
|
},
|
|
"started_at": 1.0,
|
|
"last_heartbeat_at": 1.0,
|
|
"worker_id": 7,
|
|
"attempt": 1,
|
|
}
|
|
}
|
|
queue.init_queue_refs(pending, running, {"value": 0})
|
|
worker = SimpleNamespace(busy_task_id=tx["task_id"], proc=None, reaping=False)
|
|
workers_view = SimpleNamespace(WORKERS={7: worker})
|
|
reaper_jobs = _CaptureQueue()
|
|
monkeypatch.setattr(queue, "FINALIZATION_GRACE_SEC", 0)
|
|
monkeypatch.setattr(queue, "get_task_idle_timeout_sec", lambda: 1)
|
|
monkeypatch.setattr(queue, "get_per_call_timeout_ceiling_sec", lambda: 1)
|
|
monkeypatch.setattr(queue, "get_task_abs_ceiling_sec", lambda: 10)
|
|
monkeypatch.setattr(queue, "_ensure_reaper_started", lambda: None)
|
|
monkeypatch.setattr(queue, "_reap_queue", reaper_jobs)
|
|
monkeypatch.setattr(queue, "persist_queue_snapshot", lambda reason="": True)
|
|
monkeypatch.setattr(queue, "send_with_budget", lambda *args, **kwargs: None)
|
|
|
|
queue._enforce_task_timeouts_locked(
|
|
workers_view, now=1000.0, owner_chat_id=1,
|
|
st={"evolution_mode_enabled": True},
|
|
)
|
|
|
|
assert running == {}
|
|
assert worker.reaping is True
|
|
assert len(reaper_jobs.items) == 1
|
|
stored = evolution_lifecycle._read_evolution_campaign()["active_transaction"]
|
|
assert stored["dispatch_status"] == "reaping"
|
|
|
|
queue.enqueue_evolution_task_if_needed()
|
|
assert pending == []
|
|
assert evolution_lifecycle._read_evolution_campaign()["active_transaction"][
|
|
"transaction_id"
|
|
] == tx["transaction_id"]
|
|
|
|
|
|
def test_assignment_dispatches_exact_uncommitted_evolution_claim(tmp_path, monkeypatch):
|
|
workers, task, _tx, worker, inbox, events = _assignment_case(tmp_path, monkeypatch)
|
|
|
|
workers.assign_tasks()
|
|
|
|
assert inbox.items == [task]
|
|
assert worker.busy_task_id == task["id"]
|
|
assert workers.RUNNING[task["id"]]["task"] == task
|
|
assert events.items == []
|
|
|
|
|
|
def test_assignment_rejects_stale_or_committed_evolution_claim(tmp_path, monkeypatch):
|
|
from ouroboros.task_results import load_task_result
|
|
from supervisor import evolution_lifecycle
|
|
|
|
workers, task, tx, worker, inbox, events = _assignment_case(tmp_path, monkeypatch)
|
|
task["metadata"]["evolution_transaction"]["task_id"] = "other-task"
|
|
|
|
workers.assign_tasks()
|
|
|
|
assert inbox.items == []
|
|
assert workers.RUNNING == {}
|
|
assert worker.busy_task_id is None
|
|
stored = load_task_result(tmp_path, task["id"])
|
|
assert stored["status"] == "cancelled"
|
|
assert stored["reason_code"] == "evolution_authority_missing"
|
|
assert stored["authority_reason"] == "task_mismatch"
|
|
assert events.items[-1]["metadata"]["evolution_transaction"]["task_id"] == "other-task"
|
|
|
|
workers, task, tx, _worker, inbox, _events = _assignment_case(
|
|
tmp_path / "committed", monkeypatch, task_id="committed-evo",
|
|
)
|
|
campaign = evolution_lifecycle._read_evolution_campaign()
|
|
assert evolution_lifecycle.record_evolution_commit(
|
|
campaign["id"], tx["transaction_id"], tx["task_id"], "a" * 40,
|
|
)["ok"] is True
|
|
|
|
workers.assign_tasks()
|
|
|
|
assert inbox.items == []
|
|
assert load_task_result(tmp_path / "committed", task["id"])["authority_reason"] == (
|
|
"transaction_already_committed"
|
|
)
|
|
|
|
|
|
def test_assignment_keeps_invalid_evolution_pending_when_cancel_write_fails(
|
|
tmp_path, monkeypatch,
|
|
):
|
|
from supervisor import workers as workers_module
|
|
|
|
workers, task, _tx, worker, inbox, events = _assignment_case(tmp_path, monkeypatch)
|
|
task["metadata"]["evolution_transaction"]["task_id"] = "other-task"
|
|
monkeypatch.setattr(
|
|
"ouroboros.task_results.write_task_result",
|
|
lambda *_a, **_k: (_ for _ in ()).throw(OSError("disk full")),
|
|
)
|
|
|
|
workers.assign_tasks()
|
|
|
|
assert workers_module.PENDING == [task]
|
|
assert worker.busy_task_id is None
|
|
assert inbox.items == []
|
|
assert events.items == []
|
|
|
|
|
|
def test_benchmark_seed_creates_campaign_before_enabling(tmp_path):
|
|
from devtools.benchmarks.common.server_runner import seed_owner_state
|
|
|
|
seed_owner_state(tmp_path, evolution_enabled=True)
|
|
|
|
state = json.loads((tmp_path / "state" / "state.json").read_text())
|
|
campaign = json.loads((tmp_path / "state" / "evolution_campaign.json").read_text())
|
|
assert campaign["status"] == "active"
|
|
assert campaign["id"]
|
|
assert state["evolution_mode_enabled"] is True
|