mirror of
https://github.com/razzant/ouroboros.git
synced 2026-10-03 04:07:04 +00:00
42e04e2e76
Merge user content only when the existing observed-send record proves the tail absent. Keep recorded messages intact for slot-less producers and verify repair, multipart and Local/GigaChat request shapes. Local Phase B checkpoint; complete phase review and publication are pending.
640 lines
34 KiB
Python
640 lines
34 KiB
Python
"""One frozen acceptance operation survives owner input and free collection."""
|
|
|
|
import dataclasses
|
|
import json
|
|
import threading
|
|
import time
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from ouroboros.loop_acceptance_review import acceptance_run_pending
|
|
from ouroboros.review_dispatch import collect_task_acceptance_run
|
|
from ouroboros.review_substrate import ReviewRequest, ReviewSlot, run_review_request
|
|
|
|
|
|
@pytest.mark.parametrize("cold", [False, True])
|
|
def test_released_acceptance_collects_exact_producer_without_another_send(tmp_path, monkeypatch, cold):
|
|
from ouroboros.review_custody import _ACTIVE, _ACTIVE_LOCK, _attempt_key
|
|
from ouroboros.owner_mailbox import OwnerMailboxPeek
|
|
|
|
entered, release, settled = threading.Event(), threading.Event(), threading.Event()
|
|
calls = []
|
|
original_settle = __import__("ouroboros.review_custody", fromlist=["_settle_review_attempt"])._settle_review_attempt
|
|
|
|
def settle(*args, **kwargs):
|
|
try:
|
|
return original_settle(*args, **kwargs)
|
|
finally:
|
|
settled.set()
|
|
|
|
monkeypatch.setattr("ouroboros.review_custody._settle_review_attempt", settle)
|
|
|
|
class HeldModel:
|
|
def chat(self, **kwargs):
|
|
calls.append(kwargs)
|
|
entered.set()
|
|
assert release.wait(10), "fixture did not release its model"
|
|
return {"content": json.dumps({"verdict": "FAIL", "summary": "Keep this actual criticism",
|
|
"findings": []})}, {"prompt_tokens": 5, "completion_tokens": 2}
|
|
|
|
ctx = SimpleNamespace(task_id="acceptance-root", task_attempt=1, drive_root=tmp_path,
|
|
budget_drive_root=tmp_path, task_metadata={}, pending_events=[], event_queue=None)
|
|
request = ReviewRequest(surface="task_acceptance", task_id=ctx.task_id, goal="original goal",
|
|
subject="complete original result", evidence={"requirement": "exact original"},
|
|
retry_key="acceptance-subject-one", drain_deadline=time.monotonic())
|
|
slot = ReviewSlot(slot_id="one", model="model/original", effort="high", timeout_sec=20)
|
|
try:
|
|
first = run_review_request(request, slots=[slot], drive_root=tmp_path, usage_ctx=ctx, llm=HeldModel())
|
|
assert entered.wait(5)
|
|
assert acceptance_run_pending(first)
|
|
frozen = json.loads(json.dumps(dataclasses.asdict(first)))
|
|
assert frozen["slot_roster"][0]["model"] == "model/original"
|
|
assert frozen["request"]["subject"] == "complete original result"
|
|
# Main remains independent of the frozen worker input.
|
|
ctx.messages = [{"role": "user", "content": "How is it going?"}]
|
|
ctx._owner_directives = [{"content": "How is it going?"}]
|
|
still_running = collect_task_acceptance_run(frozen, drive_root=tmp_path, usage_ctx=ctx)
|
|
assert acceptance_run_pending(still_running)
|
|
assert len(calls) == 1
|
|
release.set()
|
|
assert settled.wait(5)
|
|
assert OwnerMailboxPeek().pending(tmp_path, ctx.task_id, set(), 1)
|
|
with _ACTIVE_LOCK:
|
|
assert _attempt_key(request, slot) not in _ACTIVE
|
|
if cold:
|
|
ctx = SimpleNamespace(task_id=ctx.task_id, task_attempt=1, drive_root=tmp_path,
|
|
budget_drive_root=tmp_path, task_metadata={}, pending_events=[], event_queue=None)
|
|
result = collect_task_acceptance_run(frozen, drive_root=tmp_path, usage_ctx=ctx)
|
|
assert not acceptance_run_pending(result)
|
|
assert result.actors[0]["parsed"]["verdict"] == "FAIL"
|
|
assert result.actors[0]["operation_id"] == first.actors[0]["operation_id"]
|
|
assert result.request["subject"] == "complete original result"
|
|
assert result.request["evidence"] == {"requirement": "exact original"}
|
|
assert len(calls) == 1
|
|
finally:
|
|
release.set()
|
|
assert settled.wait(5)
|
|
|
|
|
|
def test_missing_recorded_roster_does_not_dispatch(tmp_path, monkeypatch):
|
|
monkeypatch.setattr("ouroboros.review_substrate.run_review_request",
|
|
lambda *a, **k: pytest.fail("missing source bought another review"))
|
|
request = dataclasses.asdict(ReviewRequest(surface="task_acceptance", goal="g", retry_key="subject"))
|
|
with pytest.raises(ValueError, match="roster is unavailable"):
|
|
collect_task_acceptance_run({"request": request}, drive_root=tmp_path, usage_ctx=SimpleNamespace())
|
|
|
|
|
|
def test_review_park_is_not_a_question_and_preserves_operation(tmp_path):
|
|
from ouroboros.artifacts import read_actor_source_bytes
|
|
from ouroboros.owner_wait import wait_after_tools
|
|
from tests.test_owner_wait import context
|
|
|
|
ctx, captured = context(tmp_path), []
|
|
ctx._owner_wait_requested = ""
|
|
ctx._task_acceptance_pending = "binding-original"
|
|
ctx.owner_wait_callback = lambda owner, checkpoint: captured.append(checkpoint)
|
|
trace = {"review_runs": [{"binding_hash": "binding-original", "request": {"subject": "full result"}}]}
|
|
wait_after_tools(ctx, [], trace, {}, 3, [], set(), review_binding="binding-original")
|
|
assert len(captured) == 1
|
|
assert captured[0]["quiz_id"] == ""
|
|
assert captured[0]["reason"] == "review"
|
|
source = json.loads(read_actor_source_bytes(tmp_path, ctx.task_id, captured[0]["source_ref"]))
|
|
assert source["acceptance"]["_task_acceptance_pending"] == "binding-original"
|
|
assert source["trace"] == trace
|
|
|
|
|
|
def test_unknown_custody_is_not_an_active_wait():
|
|
assert not acceptance_run_pending({"actors": [{"operation_state": "custody_lost", "late_result_pending": True}]})
|
|
assert acceptance_run_pending({"actors": [{"operation_state": "pending_dispatch"}]})
|
|
|
|
|
|
def _host_run(**overrides):
|
|
run = {"authority": "host_root", "request": {"surface": "task_acceptance", "retry_key": "subject"},
|
|
"slot_roster": [{"slot_id": "one", "model": "m", "route": "api_chat"}],
|
|
"actors": [{"operation_state": "pending_dispatch"}]}
|
|
run.update(overrides)
|
|
return run
|
|
|
|
|
|
def test_a_settled_acceptance_run_is_never_collected_twice(tmp_path, monkeypatch):
|
|
"""``acceptance_run_pending`` is the whole idempotency guard: a settled,
|
|
custody-lost or agent-tool run is never re-read, so no marker field exists."""
|
|
from ouroboros import review_dispatch
|
|
|
|
monkeypatch.setattr(review_dispatch, "collect_task_acceptance_run",
|
|
lambda *a, **k: pytest.fail("a run that is not pending was collected again"))
|
|
trace = {"review_runs": [
|
|
_host_run(actors=[{"operation_state": "settled", "parsed": {"verdict": "PASS"}}]),
|
|
_host_run(actors=[{"operation_state": "custody_lost", "late_result_pending": True}]),
|
|
_host_run(authority="agent_tool"),
|
|
_host_run(slot_roster=[]),
|
|
_host_run(request=None),
|
|
]}
|
|
advanced = review_dispatch.reconcile_pending_acceptance_runs(
|
|
trace, drive_root=tmp_path, usage_ctx=SimpleNamespace(),
|
|
)
|
|
assert advanced == 0
|
|
|
|
|
|
def test_an_uncollectable_pending_run_is_left_alone_and_never_raises(tmp_path, monkeypatch):
|
|
"""Fail-soft like ``plan_review_collect.collect_before_gate``: the pass
|
|
continues, the run stays pending, and nothing dispatches."""
|
|
from ouroboros import review_dispatch
|
|
|
|
monkeypatch.setattr("ouroboros.review_substrate.run_review_request",
|
|
lambda *a, **k: pytest.fail("a stranded run bought another review"))
|
|
for error in (ValueError("recorded acceptance roster is unavailable"),
|
|
KeyError("request"), OSError("custody store unavailable"), TimeoutError("slow")):
|
|
def raising(*_a, _error=error, **_k):
|
|
raise _error
|
|
|
|
monkeypatch.setattr(review_dispatch, "collect_task_acceptance_run", raising)
|
|
pending = _host_run()
|
|
advanced = review_dispatch.reconcile_pending_acceptance_runs(
|
|
{"review_runs": [pending]}, drive_root=tmp_path, usage_ctx=SimpleNamespace(),
|
|
)
|
|
assert advanced == 0 and acceptance_run_pending(pending)
|
|
|
|
|
|
def _mailbox_rows(root, task_id):
|
|
from ouroboros.owner_mailbox import _mailbox_path
|
|
|
|
path = _mailbox_path(root, task_id)
|
|
if not path.exists():
|
|
return []
|
|
return [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines() if line.strip()]
|
|
|
|
|
|
def test_the_settlement_wake_carries_the_reviewers_own_verdicts(tmp_path):
|
|
"""A review is advice for its author (owner D4=A): the wake IS the advice,
|
|
not a pointer to a collect verb the model reaches only by not moving on."""
|
|
from ouroboros.acceptance_settlement import announce_acceptance_settlement
|
|
|
|
announce_acceptance_settlement(
|
|
SimpleNamespace(drive_root=tmp_path),
|
|
SimpleNamespace(retry_key="acceptance-subject-one", task_id="root"),
|
|
{"slots": {"one": "ok", "two": ""}, "total": 2,
|
|
"verdicts": {"one": {"verdict": "PASS", "note": "Budget section is complete."}}},
|
|
)
|
|
rows = _mailbox_rows(tmp_path, "root")
|
|
assert len(rows) == 1 and rows[0]["provenance"] == "system"
|
|
text = rows[0]["text"]
|
|
assert "acceptance-subject-one" in text and "1 of 2 reviewer slot(s)" in text
|
|
assert "advice for you, not a signature" in text
|
|
assert "- one: PASS — Budget section is complete." in text and "- two: pending" in text
|
|
assert "keep control" not in text
|
|
|
|
|
|
class _SlotModel:
|
|
"""One chat per roster slot, each held behind its own event."""
|
|
|
|
def __init__(self, gates, verdicts):
|
|
self.gates, self.verdicts, self.calls = gates, verdicts, []
|
|
|
|
def chat(self, **kwargs):
|
|
self.calls.append(kwargs)
|
|
model = str(kwargs.get("model") or "")
|
|
assert self.gates[model].wait(10), f"fixture did not release {model}"
|
|
return ({"content": json.dumps({"verdict": self.verdicts[model], "findings": [],
|
|
"summary": f"{model} says {self.verdicts[model]}"})},
|
|
{"prompt_tokens": 5, "completion_tokens": 2})
|
|
|
|
|
|
def _released_wave(tmp_path, ctx, *, slots, model, min_successful_slots=1, task_id="root",
|
|
retry_key="acceptance-subject-one"):
|
|
request = ReviewRequest(surface="task_acceptance", task_id=task_id, goal="goal",
|
|
subject="complete result", evidence={"requirement": "exact"},
|
|
policy={"min_successful_slots": min_successful_slots},
|
|
retry_key=retry_key, drain_deadline=time.monotonic())
|
|
return run_review_request(request, slots=slots, drive_root=tmp_path, usage_ctx=ctx, llm=model)
|
|
|
|
|
|
def test_the_quorum_wake_carries_the_reviewers_verdicts_before_the_last_slot_settles(tmp_path, monkeypatch):
|
|
"""Fable roast round 1 / owner D4=A: a two-slot wave with quorum 1 wakes Main
|
|
when the first reviewer answers, then again when the straggler settles; the
|
|
straggler's wake lists every slot."""
|
|
settled = threading.Condition()
|
|
count = {"n": 0}
|
|
original_settle = __import__("ouroboros.review_custody", fromlist=["_settle_review_attempt"])._settle_review_attempt
|
|
|
|
def settle(*args, **kwargs):
|
|
try:
|
|
return original_settle(*args, **kwargs)
|
|
finally:
|
|
with settled:
|
|
count["n"] += 1
|
|
settled.notify_all()
|
|
|
|
monkeypatch.setattr("ouroboros.review_custody._settle_review_attempt", settle)
|
|
gates = {"model/a": threading.Event(), "model/b": threading.Event()}
|
|
model = _SlotModel(gates, {"model/a": "PASS", "model/b": "FAIL"})
|
|
ctx = SimpleNamespace(task_id="root", task_attempt=1, drive_root=tmp_path, budget_drive_root=tmp_path,
|
|
task_metadata={}, pending_events=[], event_queue=None)
|
|
slots = [ReviewSlot(slot_id="a", model="model/a", effort="high", timeout_sec=20),
|
|
ReviewSlot(slot_id="b", model="model/b", effort="high", timeout_sec=20)]
|
|
try:
|
|
first = _released_wave(tmp_path, ctx, slots=slots, model=model)
|
|
assert acceptance_run_pending(first) and _mailbox_rows(tmp_path, "root") == []
|
|
gates["model/a"].set()
|
|
with settled:
|
|
assert settled.wait_for(lambda: count["n"] >= 1, timeout=10)
|
|
rows = _mailbox_rows(tmp_path, "root")
|
|
assert len(rows) == 1, rows
|
|
assert "1 of 2 reviewer slot(s)" in rows[0]["text"]
|
|
assert "- a: PASS — model/a says PASS" in rows[0]["text"] and "- b: pending" in rows[0]["text"]
|
|
gates["model/b"].set()
|
|
with settled:
|
|
assert settled.wait_for(lambda: count["n"] >= 2, timeout=10)
|
|
rows = _mailbox_rows(tmp_path, "root")
|
|
assert len(rows) == 2, rows
|
|
assert "2 of 2 reviewer slot(s)" in rows[1]["text"]
|
|
assert "- b: FAIL — model/b says FAIL" in rows[1]["text"]
|
|
assert len(model.calls) == 2, "the wakes bought nothing"
|
|
finally:
|
|
for gate in gates.values():
|
|
gate.set()
|
|
with settled:
|
|
settled.wait_for(lambda: count["n"] >= 2, timeout=10)
|
|
|
|
|
|
def _terminal_ctx(tmp_path, *, task_id, retry_key="acceptance-subject-one"):
|
|
"""A worker context whose task already ended, with the wave's run in its trace."""
|
|
import queue
|
|
|
|
from ouroboros.task_results import write_task_result
|
|
|
|
write_task_result(tmp_path, task_id, "completed", chat_id=1, result="The delivered answer.")
|
|
return SimpleNamespace(task_id=task_id, task_attempt=1, drive_root=tmp_path, budget_drive_root=tmp_path,
|
|
task_metadata={}, pending_events=[], event_queue=queue.Queue(),
|
|
_execution_trace={"review_runs": []}, _retry_key=retry_key)
|
|
|
|
|
|
def test_a_panel_that_settles_after_the_task_ended_is_attached_and_announced_once(tmp_path, monkeypatch):
|
|
"""Owner fork 2=A + 3=A: one System row in the task's room, whatever the
|
|
verdict, placed inside the card's Reviews group; the projection reads the
|
|
collected verdicts and carries the same sentence the row does; the panel
|
|
that settled before the terminal carries none; nothing wakes a model."""
|
|
from ouroboros.task_results import load_task_result
|
|
|
|
settled = threading.Event()
|
|
original_settle = __import__("ouroboros.review_custody", fromlist=["_settle_review_attempt"])._settle_review_attempt
|
|
|
|
def settle(*args, **kwargs):
|
|
try:
|
|
return original_settle(*args, **kwargs)
|
|
finally:
|
|
settled.set()
|
|
|
|
monkeypatch.setattr("ouroboros.review_custody._settle_review_attempt", settle)
|
|
gates = {"model/a": threading.Event()}
|
|
model = _SlotModel(gates, {"model/a": "PASS"})
|
|
ctx = _terminal_ctx(tmp_path, task_id="late-root")
|
|
slot = ReviewSlot(slot_id="a", model="model/a", effort="high", timeout_sec=20)
|
|
try:
|
|
first = _released_wave(tmp_path, ctx, slots=[slot], model=model, task_id="late-root")
|
|
assert acceptance_run_pending(first)
|
|
run = {**json.loads(json.dumps(dataclasses.asdict(first))), "authority": "host_root",
|
|
"panel_id": "panel_1", "binding_hash": "binding-one", "candidate_hash": "c1",
|
|
"superseded_by_revision": True}
|
|
ctx._execution_trace["review_runs"].append(run)
|
|
# A sibling panel of the same task that settled while the task was still
|
|
# alive: it is republished beside the late one and must stay unstamped.
|
|
ctx._execution_trace["review_runs"].append(_host_run(
|
|
request={"surface": "task_acceptance", "retry_key": "settled-in-time"},
|
|
panel_id="panel_in_time", aggregate_signal="PASS",
|
|
actors=[{"operation_state": "settled", "parsed": {"verdict": "PASS"}}]))
|
|
gates["model/a"].set()
|
|
assert settled.wait(10)
|
|
events = []
|
|
while not any(e.get("system_type") == "acceptance_late_settlement" for e in events):
|
|
events.append(ctx.event_queue.get(timeout=10))
|
|
finally:
|
|
gates["model/a"].set()
|
|
assert settled.wait(10)
|
|
assert _mailbox_rows(tmp_path, "late-root") == [], "a terminal task has nobody to wake"
|
|
rows = [e for e in events if e.get("system_type") == "acceptance_late_settlement"]
|
|
# Exactly one chat row; the other queue entries are the typed operation facts
|
|
# every settlement emits, never a second message and never a model turn.
|
|
assert len(rows) == 1 and [e for e in events if e.get("type") == "send_message"] == rows, events
|
|
event = rows[0]
|
|
assert event["type"] == "send_message" and event["role"] == "system"
|
|
assert event["chat_id"] == 1 and event["task_id"] == "late-root"
|
|
assert event["text"].startswith("Reviewers later passed this answer. They reviewed the earlier version")
|
|
assert "- a: PASS — model/a says PASS" in event["text"]
|
|
assert event["delivery_id"] == "acceptance-late:acceptance-subject-one"
|
|
assert event["progress_meta"] == {"card_row": "reviews",
|
|
"card_row_id": "acceptance-late:acceptance-subject-one"}
|
|
assert not acceptance_run_pending(run) and run["actors"][0]["parsed"]["verdict"] == "PASS"
|
|
stored = load_task_result(tmp_path, "late-root")
|
|
assert stored["status"] == "completed", "the supplement never moves a terminal status"
|
|
panels = stored["review_projection"]["panels"]
|
|
actor = panels[0]["actors"][0]
|
|
assert actor["transport_status"] == "success" and actor["parse_status"] == "valid"
|
|
# The panel of THIS wave carries the host's sentence verbatim; the sibling
|
|
# that settled in time carries no settlement at all.
|
|
assert len(panels) == 2 and panels[1]["panel_id"] == "panel_in_time"
|
|
assert panels[0]["late_settlement"] == {"note": event["text"], "reviewed_revision": "earlier",
|
|
"settled_after_terminal": True}
|
|
assert "late_settlement" not in panels[1]
|
|
assert len(model.calls) == 1, "collection is free"
|
|
# A second settlement of the same wave finds nothing to reconcile, announces
|
|
# nothing and stamps nothing new.
|
|
from ouroboros.acceptance_settlement import attach_late_acceptance_settlement
|
|
|
|
before = json.dumps(stored["review_projection"], sort_keys=True)
|
|
assert attach_late_acceptance_settlement(
|
|
ctx, SimpleNamespace(retry_key="acceptance-subject-one", task_id="late-root"),
|
|
{"slots": {"a": "ok"}, "total": 1}, result=stored) is False
|
|
assert ctx.event_queue.empty()
|
|
assert json.dumps(load_task_result(tmp_path, "late-root")["review_projection"], sort_keys=True) == before
|
|
|
|
|
|
def test_the_late_row_never_reports_a_reviewer_whose_outcome_is_unknown_as_answered():
|
|
"""Only PASS and FAIL are verdicts. Anything else is a panel that reached no
|
|
quorum, and a slot whose physical outcome the host does not know is named as
|
|
unknown rather than read as silence."""
|
|
from ouroboros.acceptance_settlement import _late_settlement_text
|
|
|
|
wave = {"slots": {"a": "ok", "b": "ok", "c": ""},
|
|
"verdicts": {"a": {"verdict": "PASS"}, "b": {"verdict": "DEGRADED"}}}
|
|
incident = {"aggregate_signal": "DEGRADED", "actors": [
|
|
{"operation_state": "settled", "parsed": {"verdict": "PASS"}},
|
|
{"operation_state": "settled", "parsed": {"verdict": "DEGRADED"}},
|
|
{"operation_state": "custody_lost", "late_result_pending": True}]}
|
|
text = _late_settlement_text(incident, wave)
|
|
assert text.startswith("Reviewers later returned no settled verdict on this answer — 1 reviewer's outcome "
|
|
"is still unknown. They reviewed the answer that was delivered.")
|
|
assert "no quorum" not in text
|
|
assert "- a: PASS" in text and "- c: pending" in text
|
|
assert "— 2 reviewers' outcomes are still unknown." in _late_settlement_text(
|
|
{**incident, "actors": [incident["actors"][2], {"operation_state": "pending_dispatch"}]}, wave)
|
|
answered = {**incident, "actors": incident["actors"][:2]}
|
|
assert _late_settlement_text(answered, wave).startswith(
|
|
"Reviewers later returned no settled verdict on this answer. They reviewed")
|
|
assert _late_settlement_text({**answered, "aggregate_signal": "PASS"}, wave).startswith(
|
|
"Reviewers later passed this answer. They reviewed the answer that was delivered.")
|
|
assert _late_settlement_text(
|
|
{**answered, "aggregate_signal": "FAIL", "superseded_by_revision": True}, wave).startswith(
|
|
"Reviewers later rejected this answer. They reviewed the earlier version,")
|
|
|
|
|
|
def test_a_terminal_task_gets_one_row_at_completion_not_at_quorum(tmp_path, monkeypatch):
|
|
"""Fable review round 2: a quorum settlement on an already-terminal task must
|
|
not announce a half-settled wave (the straggler's verdict would never reach
|
|
the chat, deduped behind the same delivery id). Nothing is reconciled until
|
|
every slot answered, so the one row carries every reviewer."""
|
|
settled = threading.Condition()
|
|
count = {"n": 0}
|
|
original_settle = __import__("ouroboros.review_custody", fromlist=["_settle_review_attempt"])._settle_review_attempt
|
|
|
|
def settle(*args, **kwargs):
|
|
try:
|
|
return original_settle(*args, **kwargs)
|
|
finally:
|
|
with settled:
|
|
count["n"] += 1
|
|
settled.notify_all()
|
|
|
|
monkeypatch.setattr("ouroboros.review_custody._settle_review_attempt", settle)
|
|
gates = {"model/a": threading.Event(), "model/b": threading.Event()}
|
|
model = _SlotModel(gates, {"model/a": "PASS", "model/b": "PASS"})
|
|
ctx = _terminal_ctx(tmp_path, task_id="late-two")
|
|
slots = [ReviewSlot(slot_id="a", model="model/a", effort="high", timeout_sec=20),
|
|
ReviewSlot(slot_id="b", model="model/b", effort="high", timeout_sec=20)]
|
|
try:
|
|
first = _released_wave(tmp_path, ctx, slots=slots, model=model, task_id="late-two")
|
|
run = {**json.loads(json.dumps(dataclasses.asdict(first))), "authority": "host_root",
|
|
"panel_id": "panel_1", "binding_hash": "binding-one", "candidate_hash": "c1"}
|
|
ctx._execution_trace["review_runs"].append(run)
|
|
gates["model/a"].set()
|
|
with settled:
|
|
assert settled.wait_for(lambda: count["n"] >= 1, timeout=10)
|
|
# The late row is enqueued inside the settle (before the wrapper counts), so this is deterministic.
|
|
assert not [e for e in list(ctx.event_queue.queue) if e.get("system_type") == "acceptance_late_settlement"]
|
|
gates["model/b"].set()
|
|
with settled:
|
|
assert settled.wait_for(lambda: count["n"] >= 2, timeout=10)
|
|
events = []
|
|
while not any(e.get("system_type") == "acceptance_late_settlement" for e in events):
|
|
events.append(ctx.event_queue.get(timeout=10))
|
|
finally:
|
|
for gate in gates.values():
|
|
gate.set()
|
|
with settled:
|
|
settled.wait_for(lambda: count["n"] >= 2, timeout=10)
|
|
rows = [e for e in events if e.get("system_type") == "acceptance_late_settlement"]
|
|
assert len(rows) == 1 and "- a: PASS" in rows[0]["text"] and "- b: PASS" in rows[0]["text"]
|
|
assert "pending" not in rows[0]["text"] and _mailbox_rows(tmp_path, "late-two") == []
|
|
|
|
|
|
def test_two_late_panels_of_one_task_each_announce_their_own_row(tmp_path, monkeypatch):
|
|
"""Astra review round 4: a settlement reconciles only its own wave. Collecting
|
|
every pending panel would let the first callback swallow the sibling's verdicts
|
|
and leave that panel's own callback with nothing to announce."""
|
|
settled = threading.Condition()
|
|
count = {"n": 0}
|
|
original_settle = __import__("ouroboros.review_custody", fromlist=["_settle_review_attempt"])._settle_review_attempt
|
|
|
|
def settle(*args, **kwargs):
|
|
try:
|
|
return original_settle(*args, **kwargs)
|
|
finally:
|
|
with settled:
|
|
count["n"] += 1
|
|
settled.notify_all()
|
|
|
|
monkeypatch.setattr("ouroboros.review_custody._settle_review_attempt", settle)
|
|
# Both settlements are PUBLISHED (custody's lock block ran, the settled attempt
|
|
# is collectible) before EITHER announcement runs, so the two announcements
|
|
# race exactly as two panels settling together would; a broken barrier fails
|
|
# the test instead of degrading it to a serial schedule.
|
|
barrier = threading.Barrier(2, timeout=10)
|
|
broken = []
|
|
from ouroboros import acceptance_settlement as leaf
|
|
original_announce = leaf.announce_acceptance_settlement
|
|
|
|
def announce(*args, **kwargs):
|
|
try:
|
|
barrier.wait()
|
|
except threading.BrokenBarrierError:
|
|
broken.append(True)
|
|
return original_announce(*args, **kwargs)
|
|
|
|
monkeypatch.setattr(leaf, "announce_acceptance_settlement", announce)
|
|
gates = {"model/a": threading.Event(), "model/b": threading.Event()}
|
|
model = _SlotModel(gates, {"model/a": "PASS", "model/b": "FAIL"})
|
|
ctx = _terminal_ctx(tmp_path, task_id="late-pair")
|
|
try:
|
|
for key, slot_model in (("wave-one", "model/a"), ("wave-two", "model/b")):
|
|
slot = ReviewSlot(slot_id=key, model=slot_model, effort="high", timeout_sec=20)
|
|
first = _released_wave(tmp_path, ctx, slots=[slot], model=model, task_id="late-pair", retry_key=key)
|
|
ctx._execution_trace["review_runs"].append(
|
|
{**json.loads(json.dumps(dataclasses.asdict(first))), "authority": "host_root",
|
|
"panel_id": f"panel_{key}", "binding_hash": f"binding-{key}", "candidate_hash": f"c-{key}"})
|
|
gates["model/a"].set()
|
|
gates["model/b"].set()
|
|
with settled:
|
|
assert settled.wait_for(lambda: count["n"] >= 2, timeout=15)
|
|
events = []
|
|
deadline = time.monotonic() + 10
|
|
while time.monotonic() < deadline and sum(1 for e in events if e.get("system_type") == "acceptance_late_settlement") < 2:
|
|
try:
|
|
events.append(ctx.event_queue.get(timeout=2))
|
|
except Exception:
|
|
break
|
|
finally:
|
|
for gate in gates.values():
|
|
gate.set()
|
|
with settled:
|
|
settled.wait_for(lambda: count["n"] >= 2, timeout=10)
|
|
rows = [e for e in events if e.get("system_type") == "acceptance_late_settlement"]
|
|
assert not broken, "both announcements must have raced through the barrier"
|
|
assert sorted(r["delivery_id"] for r in rows) == ["acceptance-late:wave-one", "acceptance-late:wave-two"], events
|
|
assert any("passed" in r["text"] for r in rows) and any("rejected" in r["text"] for r in rows)
|
|
|
|
|
|
def test_an_owner_followup_sets_the_running_panel_aside(tmp_path):
|
|
"""Fable review round 4: after the owner changes the requirements, the answer
|
|
Main writes for them is not a delivery under the panel that reviewed the old
|
|
ones — the latch clears, so the ordinary path decides while the old panel's
|
|
verdicts still arrive as advice."""
|
|
from ouroboros import loop
|
|
from tests.test_delivery_forced_finalization import _forced_test_context
|
|
|
|
_loop, registry, _ctx, trace = _forced_test_context(tmp_path)
|
|
registry._ctx._task_acceptance_pending = "binding-one"
|
|
trace["review_decision"] = {}
|
|
loop._supersede_task_acceptance_for_owner_followup(registry._ctx, trace)
|
|
assert registry._ctx._task_acceptance_pending == ""
|
|
assert trace["acceptance_decision"]["reason"] == "owner_followup"
|
|
|
|
|
|
def test_a_late_settlement_for_another_task_is_never_published(tmp_path):
|
|
"""The worker may already be rebound: a wave whose retry key is not in the
|
|
live trace returns False and writes nothing anywhere."""
|
|
from ouroboros.acceptance_settlement import attach_late_acceptance_settlement
|
|
from ouroboros.task_results import load_task_result
|
|
|
|
ctx = _terminal_ctx(tmp_path, task_id="other-root")
|
|
ctx._execution_trace["review_runs"].append(_host_run(request={"surface": "task_acceptance", "retry_key": "another"}))
|
|
before = json.dumps(load_task_result(tmp_path, "other-root"), sort_keys=True)
|
|
assert attach_late_acceptance_settlement(
|
|
ctx, SimpleNamespace(retry_key="acceptance-subject-one", task_id="other-root"),
|
|
{"slots": {"a": "ok"}, "total": 1}, result={"chat_id": 1}) is False
|
|
assert ctx.event_queue.empty() and _mailbox_rows(tmp_path, "other-root") == []
|
|
assert json.dumps(load_task_result(tmp_path, "other-root"), sort_keys=True) == before
|
|
|
|
|
|
def test_every_acceptance_wake_reoffers_a_changed_keep_contract(tmp_path, monkeypatch):
|
|
"""A replacement candidate inherits ``control_episode_seen``; the contract for
|
|
the NEW candidate must still be shown, while identical bytes are not repeated."""
|
|
from ouroboros.loop_acceptance_review import wait_for_acceptance_feedback
|
|
from tests.test_delivery_forced_finalization import _forced_test_context
|
|
|
|
loop, registry, ctx, trace = _forced_test_context(tmp_path)
|
|
monkeypatch.setattr("ouroboros.owner_wait.wait_after_tools", lambda *_a, **_k: None)
|
|
registry._ctx._task_acceptance_pending = "binding-one"
|
|
blocks = lambda: str(ctx.messages).count("[DELIVERY_FINALIZATION_CONTROL]") # noqa: E731
|
|
|
|
first = loop._replace_delivery_candidate(registry, ctx, trace, "First complete answer.", control="candidate")
|
|
assert first.control_episode_seen is False
|
|
wait_for_acceptance_feedback(registry, ctx, trace, [], set())
|
|
assert first.control_episode_seen is True and blocks() == 1
|
|
# The same candidate renders identical bytes: a second wake adds no noise.
|
|
wait_for_acceptance_feedback(registry, ctx, trace, [], set())
|
|
assert blocks() == 1
|
|
second = loop._replace_delivery_candidate(registry, ctx, trace, "Second complete answer.", control="candidate")
|
|
assert second.control_episode_seen is True, "the inherited flag is what hid the contract"
|
|
wait_for_acceptance_feedback(registry, ctx, trace, [], set())
|
|
assert blocks() == 2 and second.content_sha256[:12] in str(ctx.messages)
|
|
|
|
|
|
def test_the_acceptance_wake_keeps_the_one_repair_already_spent(tmp_path, monkeypatch):
|
|
"""Scope review round 1: re-arming on every wake reset ``repair_attempted``,
|
|
so a candidate could burn one malformed-control repair per wake instead of
|
|
one per episode. The wake's re-offer preserves the spent repair; an ordinary
|
|
arm (something changed) still opens a fresh episode."""
|
|
from ouroboros.loop_acceptance_review import wait_for_acceptance_feedback
|
|
from tests.test_delivery_forced_finalization import _forced_test_context
|
|
|
|
loop, registry, ctx, trace = _forced_test_context(tmp_path)
|
|
monkeypatch.setattr("ouroboros.owner_wait.wait_after_tools", lambda *_a, **_k: None)
|
|
registry._ctx._task_acceptance_pending = "binding-one"
|
|
candidate = loop._replace_delivery_candidate(registry, ctx, trace, "Complete answer.", control="candidate")
|
|
wait_for_acceptance_feedback(registry, ctx, trace, [], set())
|
|
candidate.repair_attempted = True # the one repair was spent on a malformed control
|
|
wait_for_acceptance_feedback(registry, ctx, trace, [], set())
|
|
assert candidate.repair_attempted is True, "the wake re-offer must not refund the repair"
|
|
loop._arm_delivery_control(registry, ctx, trace)
|
|
assert candidate.repair_attempted is False, "an ordinary arm opens a new episode"
|
|
|
|
|
|
def test_the_rearmed_contract_never_rewrites_an_already_sent_row(tmp_path):
|
|
"""Issue #906: merging into a sent row discards the conversation cache. Every
|
|
wake re-arms, so the control block must take the execution slot and append."""
|
|
import copy as _copy
|
|
|
|
from ouroboros.transcript_prefix import observe_send
|
|
from tests.test_delivery_forced_finalization import _forced_test_context
|
|
|
|
loop, registry, ctx, trace = _forced_test_context(tmp_path)
|
|
loop._replace_delivery_candidate(registry, ctx, trace, "Complete answer.", control="candidate")
|
|
ctx.messages.append({"role": "user", "content": "An owner follow-up that already went out."})
|
|
observe_send(registry._ctx, ctx.messages, round_idx=1)
|
|
sent = _copy.deepcopy(ctx.messages[-1])
|
|
|
|
loop._arm_delivery_control(registry, ctx, trace)
|
|
|
|
assert sent in ctx.messages, "an already-sent message was rewritten"
|
|
control = [row for row in ctx.messages
|
|
if "[DELIVERY_FINALIZATION_CONTROL]" in str(row.get("content") or "")]
|
|
assert len(control) == 1 and control[0] is not ctx.messages[ctx.messages.index(sent)]
|
|
# Only the acceptance wake's repeated re-offer is deduplicated. Every other
|
|
# caller arms because something changed, so it always appends.
|
|
loop._arm_delivery_control(registry, ctx, trace)
|
|
assert str(ctx.messages).count("[DELIVERY_FINALIZATION_CONTROL]") == 2
|
|
|
|
|
|
@pytest.mark.parametrize("choice", ["wait", "finish"])
|
|
def test_pending_review_rides_beside_the_verb_and_is_recorded_on_every_answer(tmp_path, choice):
|
|
"""WP-7: the optional wait/finish choice is a sibling of ``acceptance_subject``,
|
|
never an extra key that invalidates the body, and every control answer records
|
|
it (an answer without the key means wait)."""
|
|
from tests.test_delivery_control_lineage import _start_control_episode
|
|
|
|
loop, registry, ctx, trace, candidate = _start_control_episode(tmp_path)
|
|
loop._arm_delivery_control(registry, ctx, trace)
|
|
status, text = loop._resolve_delivery_control(
|
|
json.dumps({"delivery_control": "keep", "pending_review": choice}), registry, ctx, trace,
|
|
)
|
|
assert (status, text) == ("resolved", candidate.full_text)
|
|
assert registry._ctx._acceptance_pending_review_choice == choice
|
|
loop._arm_delivery_control(registry, ctx, trace)
|
|
status, _text = loop._resolve_delivery_control(
|
|
json.dumps({"delivery_control": "keep"}), registry, ctx, trace,
|
|
)
|
|
assert status == "resolved" and registry._ctx._acceptance_pending_review_choice == "wait"
|
|
|
|
|
|
@pytest.mark.parametrize("content", ["", [{"type": "thinking", "thinking": "reasoning only"}], "invalid control"])
|
|
def test_delivery_repair_keeps_the_sent_control_prefix(tmp_path, content):
|
|
import copy
|
|
from ouroboros.transcript_prefix import observe_send
|
|
from tests.test_delivery_forced_finalization import _forced_test_context
|
|
|
|
loop, registry, ctx, trace = _forced_test_context(tmp_path)
|
|
ctx.messages.insert(0, {"role": "system", "content": "Complete the task."})
|
|
candidate = loop._replace_delivery_candidate(registry, ctx, trace, "Complete retained answer.", control="candidate")
|
|
loop._arm_delivery_control(registry, ctx, trace)
|
|
observe_send(registry._ctx, ctx.messages, round_idx=1)
|
|
sent = copy.deepcopy(ctx.messages)
|
|
|
|
status, text = loop._resolve_delivery_control(content, registry, ctx, trace)
|
|
|
|
assert (status, text) == ("retry", "")
|
|
assert ctx.messages[:len(sent)] == sent
|
|
assert "[DELIVERY_CONTROL_REPAIR]" in ctx.messages[-1]["content"]
|
|
assert ctx.messages[-1]["role"] == "user"
|
|
assert observe_send(registry._ctx, ctx.messages, round_idx=2) is None
|
|
assert candidate.full_text == "Complete retained answer."
|