mirror of
https://github.com/razzant/ouroboros.git
synced 2026-10-02 19:58:46 +00:00
79b9b4b11c
The steer-level pin modelled only a headless pooled root (no stamp, no client id). The common geometry is a root promoted from an owner message: it inherits the door's stamp and client id by value but is not a direct turn, and its steer must still travel as the task's own words. Parametrized over both shapes (found by the Fable verification workflow and the astra delta review).
829 lines
42 KiB
Python
829 lines
42 KiB
Python
"""Task-authored messages never travel as owner text (14.09 incident, wave 2).
|
|
|
|
``_handle_steer_task`` used to decide WHO was speaking five times from proxies:
|
|
a routing-contract lane a Swarm root never has, a room veto keyed on the chat, an
|
|
empty client id read as "agent-issued", and every steer written as owner text —
|
|
so a Project root's words reached the Main root as ``[Message from my human]``,
|
|
entered the owner-directive corpus and superseded a paid acceptance panel. The
|
|
host now mints ONE issuer fact by value (``control_routing._routing_issuer``): an
|
|
owner turn keeps today's path; a task's own words are written as a task-message
|
|
row with ``independent_task`` provenance, render under their own prefix, enter no
|
|
owner corpus, bump no owner generation, and are confirmed (or
|
|
refused) to the issuer as WRITTEN with the host's reason. Fixture geometry
|
|
matters: the issuing roots below are POOLED/Swarm roots with no chat ingress id.
|
|
A drained owner message keys the visible receipt without changing authorship.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import queue
|
|
import types
|
|
|
|
import pytest
|
|
|
|
from ouroboros.utils import append_jsonl
|
|
|
|
_OWNER_ORIGIN = "the owner's twenty-minute-old Swarm message"
|
|
|
|
|
|
def _pooled_root_ctx(tmp_path, *, task_id="swarm-root", chat_id=1, metadata=None):
|
|
"""A pooled/Swarm root: force_plan metadata, an origin the agent copied by value,
|
|
NO client_message_id, NO is_direct_chat, nothing drained yet."""
|
|
md = {
|
|
"force_plan": True, "force_plan_source": "swarm", "root_task_id": task_id,
|
|
"origin_message_text": _OWNER_ORIGIN,
|
|
}
|
|
md.update(metadata or {})
|
|
return types.SimpleNamespace(
|
|
pending_events=[], event_queue=None, current_chat_id=chat_id, drive_root=tmp_path,
|
|
task_id=task_id, task_metadata=md, last_owner_delivery=None, project_id="",
|
|
)
|
|
|
|
|
|
def _owner_turn_ctx(tmp_path, *, client_message_id="cm-1"):
|
|
"""A direct turn the owner door stamped (``origin_message_ref``): the one shape
|
|
that speaks as an owner turn; a client id alone never does."""
|
|
return types.SimpleNamespace(
|
|
pending_events=[], event_queue=None, current_chat_id=1, drive_root=tmp_path,
|
|
task_id="turn-1", is_direct_chat=True, last_owner_delivery=None,
|
|
task_metadata={"client_message_id": client_message_id, "origin_message_text": _OWNER_ORIGIN,
|
|
"origin_message_ref": {"chat_id": 1, "client_message_id": client_message_id}},
|
|
)
|
|
|
|
|
|
def _owner_started_receiver():
|
|
"""The receiving root was started by the owner: its first text keeps the owner label."""
|
|
return types.SimpleNamespace(
|
|
task_attempt=1, task_metadata={"origin_message_ref": {"chat_id": 42, "client_message_id": "t-target-origin"}},
|
|
)
|
|
|
|
|
|
def _drain_owner_followup(tmp_path, ctx, *, client_message_id="owner-followup"):
|
|
from ouroboros.loop_round_limits import _drain_incoming_messages
|
|
from ouroboros.owner_mailbox import write_owner_message
|
|
|
|
write_owner_message(tmp_path, "Please coordinate the review", task_id=ctx.task_id,
|
|
msg_id="followup-mailbox", client_message_id=client_message_id)
|
|
_drain_incoming_messages([], queue.Queue(), tmp_path, ctx.task_id, None, set(), owner_ctx=ctx)
|
|
assert ctx.last_owner_delivery["client_message_id"] == client_message_id
|
|
|
|
|
|
@pytest.fixture(params=["pooled", "direct"])
|
|
def target_lane(request):
|
|
import threading
|
|
from supervisor.active_activity import get_direct_activity_registry
|
|
|
|
actor = types.SimpleNamespace(
|
|
_owner_message_admission_lock=threading.RLock(), _busy=True,
|
|
_accepting_owner_messages=True, _current_task_id="t-target", _current_chat_id=1,
|
|
_current_task_metadata={}, _current_task_text="Review", _owner_message_generation=3,
|
|
)
|
|
registry = get_direct_activity_registry()
|
|
if request.param == "direct":
|
|
registry.register("t-target", 1, actor=actor)
|
|
try:
|
|
yield actor
|
|
finally:
|
|
if request.param == "direct":
|
|
registry.unregister("t-target")
|
|
|
|
|
|
def _supervisor(tmp_path, *, running=None, acks=None, notices=None):
|
|
acks = [] if acks is None else acks
|
|
notices = [] if notices is None else notices
|
|
return types.SimpleNamespace(
|
|
DRIVE_ROOT=tmp_path, RUNNING=dict(running or {}), PENDING=[],
|
|
send_with_budget=lambda _chat_id, text, *a, **k: notices.append(text),
|
|
bridge=types.SimpleNamespace(send_routing_ack=lambda _chat_id, **payload: acks.append(payload)),
|
|
append_jsonl=append_jsonl,
|
|
persist_queue_snapshot=lambda **_k: None,
|
|
)
|
|
|
|
|
|
def _wire(ctx, supervisor, emitted):
|
|
from supervisor.events import _handle_steer_task
|
|
|
|
ctx.event_queue = types.SimpleNamespace(
|
|
put_nowait=lambda event: (emitted.append(event), _handle_steer_task(event, supervisor))[0],
|
|
)
|
|
return ctx
|
|
|
|
|
|
def _events(tmp_path, kind):
|
|
path = tmp_path / "logs" / "events.jsonl"
|
|
if not path.exists():
|
|
return []
|
|
rows = [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines() if line.strip()]
|
|
return [row for row in rows if row.get("type") == kind]
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _queue_root(tmp_path, monkeypatch):
|
|
import supervisor.queue as queue_mod
|
|
|
|
monkeypatch.setattr(queue_mod, "DRIVE_ROOT", str(tmp_path))
|
|
monkeypatch.setattr(queue_mod, "ACCEPTANCE_FENCES", {})
|
|
return tmp_path
|
|
|
|
|
|
# --- (a) a pooled root's words are a task message, not the owner's -----------
|
|
|
|
@pytest.mark.parametrize("drained_owner", [None, "owner-followup", ""], ids=["standalone", "owner-id", "legacy-no-id"])
|
|
@pytest.mark.parametrize("project_sender", [False, True])
|
|
@pytest.mark.parametrize("root_shape", ["headless", "promoted"])
|
|
def test_a_pooled_root_steer_is_written_as_its_own_words_and_supersedes_nothing(
|
|
tmp_path, target_lane, drained_owner, project_sender, root_shape,
|
|
):
|
|
"""``promoted`` is the common geometry: the root inherited the owner door's stamp
|
|
and client id from the message that promoted it (ancestry, by value) but is not
|
|
a direct turn, so it still speaks as a task; ``headless`` carries no stamp at all."""
|
|
import supervisor.queue as queue_mod
|
|
from ouroboros.loop_messages import _initialize_owner_directives, owner_source_sha256
|
|
from ouroboros.loop_round_limits import _drain_incoming_messages
|
|
from ouroboros.owner_mailbox import KIND_TASK_MESSAGE, acknowledged_task_message_ids, drain_owner_entries
|
|
from ouroboros.project_dialogue import latest_chat_annotations
|
|
from ouroboros.projects_registry import create_project
|
|
from ouroboros.tools.control import _steer_task
|
|
|
|
fence = {"status": "active", "owner_message_generation": 3}
|
|
queue_mod.ACCEPTANCE_FENCES["t-target"] = fence
|
|
acks, notices, emitted = [], [], []
|
|
supervisor = _supervisor(
|
|
tmp_path, running={"t-target": {"task": {"id": "t-target", "chat_id": 1, "root_task_id": "t-target"}}},
|
|
acks=acks, notices=notices,
|
|
)
|
|
chat_id = create_project(tmp_path, "source", name="Source")["chat_id"] if project_sender else 1
|
|
inherited = {
|
|
"client_message_id": "cm-origin",
|
|
"origin_message_ref": {"chat_id": 1, "client_message_id": "cm-origin", "ts": "t", "text_sha256": "x" * 64},
|
|
} if root_shape == "promoted" else {}
|
|
ctx = _wire(_pooled_root_ctx(tmp_path, chat_id=chat_id, metadata=inherited), supervisor, emitted)
|
|
if drained_owner is not None:
|
|
_drain_owner_followup(tmp_path, ctx, client_message_id=drained_owner)
|
|
|
|
out = _steer_task(ctx, "t-target", "the PR is ready; please review it")
|
|
|
|
# The issuer fact is the host's, by value; the message is the task's OWN
|
|
# words -- never the owner origin the root carries in its metadata.
|
|
assert emitted[0]["issuer"] == {"kind": "task", "task_id": "swarm-root", "root_task_id": "swarm-root"}
|
|
assert emitted[0]["message"] == "the PR is ready; please review it"
|
|
assert "attachment_uploads" not in emitted[0]
|
|
assert out.startswith("✉️ Message to task t-target written to its mailbox (durably confirmed")
|
|
assert "not as owner text" in out and "cannot be attached" in out
|
|
[row] = drain_owner_entries(tmp_path, "t-target")
|
|
assert (row["kind"], row["provenance"], row["source_task_id"]) == (
|
|
KIND_TASK_MESSAGE, "independent_task", "swarm-root",
|
|
)
|
|
assert "client_message_id" not in row
|
|
# A drained owner message gets its receipt, never authorship of the relay.
|
|
assert fence["owner_message_generation"] == 3
|
|
assert target_lane._owner_message_generation == 3
|
|
assert notices == []
|
|
if drained_owner:
|
|
assert emitted[0]["client_message_id"] == "owner-followup"
|
|
assert len(acks) == 1 and acks[0]["client_message_id"] == "owner-followup"
|
|
assert acks[0]["status"] == "delivered"
|
|
else:
|
|
assert emitted[0]["client_message_id"].startswith("agent-steer:")
|
|
assert acks == []
|
|
receipt = latest_chat_annotations(tmp_path)[emitted[0]["client_message_id"]]
|
|
assert (receipt["action"], receipt["status"], receipt["target"]) == ("steer_task", "delivered", "t-target")
|
|
assert "options" not in receipt
|
|
[logged] = _events(tmp_path, "task_message_routed")
|
|
assert (logged["task_id"], logged["target_task_id"], logged["status"]) == ("swarm-root", "t-target", "written")
|
|
|
|
# The RECEIVER drains it as context: rendered under its own prefix, the owner
|
|
# corpus untouched (so owner_source_sha256 cannot supersede a reviewed answer),
|
|
# no owner delivery stamped, the row acknowledged, the injected event typed.
|
|
receiver = _owner_started_receiver()
|
|
messages = [{"role": "user", "content": "Initial requirement verbatim"}]
|
|
_initialize_owner_directives(receiver, messages)
|
|
corpus_before = owner_source_sha256(receiver)
|
|
events: queue.Queue = queue.Queue()
|
|
_drain_incoming_messages(messages, queue.Queue(), tmp_path, "t-target", events, set(), owner_ctx=receiver)
|
|
delivered = str(messages[-1]["content"])
|
|
assert "[Message from independent task swarm-root]\nthe PR is ready; please review it" in delivered
|
|
assert "[Message from my human]" not in delivered and "ancestor" not in delivered
|
|
assert owner_source_sha256(receiver) == corpus_before
|
|
assert [d["source"] for d in receiver._owner_directives] == ["initial_user"]
|
|
assert getattr(receiver, "last_owner_delivery", None) is None
|
|
assert row["msg_id"] in acknowledged_task_message_ids(tmp_path, "t-target", attempt_key=1)
|
|
injected = events.get_nowait()
|
|
assert (injected["type"], injected["source_task_id"], injected["provenance"]) == (
|
|
"task_message_injected", "swarm-root", "independent_task",
|
|
)
|
|
|
|
|
|
# --- (b) an owner turn keeps today's exact path -------------------------------
|
|
|
|
@pytest.mark.parametrize("stamp", ["logged-ref", "suppressed-log"])
|
|
def test_an_owner_turn_from_main_still_steers_with_owner_text_and_bumps_the_generation(tmp_path, stamp):
|
|
import supervisor.queue as queue_mod
|
|
from ouroboros.loop_messages import _initialize_owner_directives, owner_source_sha256
|
|
from ouroboros.loop_round_limits import _drain_incoming_messages
|
|
from ouroboros.owner_mailbox import KIND_OWNER_TEXT, drain_owner_entries
|
|
from ouroboros.tools.control import _steer_task
|
|
|
|
fence = {"status": "active", "owner_message_generation": 3}
|
|
queue_mod.ACCEPTANCE_FENCES["t-target"] = fence
|
|
acks, notices, emitted = [], [], []
|
|
supervisor = _supervisor(
|
|
tmp_path, running={"t-target": {"task": {"id": "t-target", "chat_id": 42, "root_task_id": "t-target"}}},
|
|
acks=acks, notices=notices,
|
|
)
|
|
ctx = _wire(_owner_turn_ctx(tmp_path), supervisor, emitted)
|
|
if stamp == "suppressed-log": # a never-logged owner message: the door's designed absence of a ref
|
|
del ctx.task_metadata["origin_message_ref"]
|
|
ctx.task_metadata["origin_suppressed"] = True
|
|
|
|
out = _steer_task(ctx, "t-target", "model paraphrase")
|
|
|
|
assert out.startswith("✉️ Steering task t-target: mailbox delivery is durably confirmed")
|
|
assert emitted[0]["issuer"] == {"kind": "owner_turn"}
|
|
[row] = drain_owner_entries(tmp_path, "t-target")
|
|
# The first act relays the owner's exact bytes, as owner text, under the owner id.
|
|
assert (row["kind"], row["text"], row["client_message_id"]) == (KIND_OWNER_TEXT, _OWNER_ORIGIN, "cm-1")
|
|
assert fence["owner_message_generation"] == 4
|
|
assert acks[-1]["status"] == "delivered" and acks[-1]["client_message_id"] == "cm-1"
|
|
# Main addresses a root in another chat: the lane is the registry's answer.
|
|
assert notices == []
|
|
assert _events(tmp_path, "task_message_routed") == []
|
|
|
|
receiver = _owner_started_receiver()
|
|
messages = [{"role": "user", "content": "Initial requirement verbatim"}]
|
|
_initialize_owner_directives(receiver, messages)
|
|
corpus_before = owner_source_sha256(receiver)
|
|
_drain_incoming_messages(messages, queue.Queue(), tmp_path, "t-target", None, set(), owner_ctx=receiver)
|
|
assert "[Message from my human]" in messages[-1]["content"]
|
|
assert owner_source_sha256(receiver) != corpus_before
|
|
assert [directive["source"] for directive in receiver._owner_directives] == ["initial_user", "owner_mailbox"]
|
|
assert receiver.last_owner_delivery["client_message_id"] == "cm-1"
|
|
|
|
|
|
def test_a_relay_retry_deduplicates_without_losing_the_owner_receipt(tmp_path):
|
|
from ouroboros.owner_mailbox import KIND_TASK_MESSAGE, drain_owner_entries
|
|
from ouroboros.tools.control import _steer_task
|
|
from supervisor.steering import _handle_steer_task
|
|
|
|
emitted, acks = [], []
|
|
supervisor = _supervisor(tmp_path, acks=acks,
|
|
running={"t-target": {"task": {"id": "t-target", "chat_id": 1}}})
|
|
ctx = _pooled_root_ctx(tmp_path)
|
|
_drain_owner_followup(tmp_path, ctx)
|
|
_wire(ctx, supervisor, emitted)
|
|
|
|
_steer_task(ctx, "t-target", "My review is complete")
|
|
_handle_steer_task(emitted[0], supervisor)
|
|
|
|
[row] = drain_owner_entries(tmp_path, "t-target")
|
|
assert row["kind"] == KIND_TASK_MESSAGE and "client_message_id" not in row
|
|
assert row["text"] == "My review is complete"
|
|
assert acks and all(ack["client_message_id"] == "owner-followup" for ack in acks)
|
|
assert all(ack["status"] == "delivered" for ack in acks)
|
|
|
|
|
|
# --- (c) refusals to a task issuer: typed, silent in chat, one Logs row -------
|
|
|
|
@pytest.mark.parametrize("drained_owner", [False, True])
|
|
def test_a_task_steer_of_a_cancelling_target_is_refused_typed_with_no_chat_and_no_picker(
|
|
tmp_path, monkeypatch, drained_owner,
|
|
):
|
|
import ouroboros.cancel_intents as cancel_intents
|
|
from ouroboros.project_dialogue import latest_chat_annotations
|
|
from ouroboros.tools.control import _steer_task
|
|
|
|
monkeypatch.setattr(cancel_intents, "cancel_pending", lambda _root, task_id, **_k: task_id == "t-target")
|
|
acks, notices, emitted = [], [], []
|
|
supervisor = _supervisor(
|
|
tmp_path, running={"t-target": {"task": {"id": "t-target", "chat_id": 1}}}, acks=acks, notices=notices,
|
|
)
|
|
ctx = _wire(_pooled_root_ctx(tmp_path), supervisor, emitted)
|
|
if drained_owner:
|
|
_drain_owner_followup(tmp_path, ctx)
|
|
|
|
out = _steer_task(ctx, "t-target", "stop after this file")
|
|
|
|
assert out.startswith("⚠️ STEER_REJECTED: task t-target was not steered (cancel_pending)")
|
|
assert notices == []
|
|
assert [ack["client_message_id"] for ack in acks] == (["owner-followup"] if drained_owner else [])
|
|
receipt = latest_chat_annotations(tmp_path)[emitted[0]["client_message_id"]]
|
|
assert (receipt["status"], receipt["reason"]) == ("rejected", "cancel_pending")
|
|
assert "options" not in receipt
|
|
[logged] = _events(tmp_path, "task_message_routed")
|
|
assert (logged["task_id"], logged["target_task_id"], logged["status"], logged["reason"]) == (
|
|
"swarm-root", "t-target", "refused", "cancel_pending",
|
|
)
|
|
|
|
|
|
def test_a_task_relay_to_a_closing_direct_turn_keeps_the_owner_receipt(tmp_path, monkeypatch):
|
|
import threading
|
|
from ouroboros.owner_mailbox import drain_owner_entries
|
|
from ouroboros.project_dialogue import latest_chat_annotations, routing_refusal_cause
|
|
from ouroboros.tools.control import _steer_task
|
|
from supervisor import workers
|
|
|
|
actor = types.SimpleNamespace(_owner_message_admission_lock=threading.RLock(),
|
|
_busy=True, _accepting_owner_messages=False,
|
|
_current_task_id="t-target", _owner_message_generation=3)
|
|
# Discovery succeeded, but admission closed before the transaction acquired its lock.
|
|
monkeypatch.setattr(workers, "get_direct_chat_agent", lambda _target: actor)
|
|
monkeypatch.setattr(workers, "direct_chat_turn", lambda _target: {
|
|
"id": "t-target", "chat_id": 1, "_is_direct_chat": True,
|
|
})
|
|
acks, notices, emitted = [], [], []
|
|
supervisor = _supervisor(tmp_path, acks=acks, notices=notices)
|
|
ctx = _wire(_pooled_root_ctx(tmp_path), supervisor, emitted)
|
|
_drain_owner_followup(tmp_path, ctx)
|
|
|
|
out = _steer_task(ctx, "t-target", "My review is complete")
|
|
|
|
assert "STEER_REJECTED" in out and "target_closed" in out
|
|
assert drain_owner_entries(tmp_path, "t-target") == [] and notices == []
|
|
assert actor._owner_message_generation == 3
|
|
assert len(acks) == 1 and acks[0]["client_message_id"] == "owner-followup"
|
|
assert latest_chat_annotations(tmp_path)["owner-followup"]["reason"] == "target_closed"
|
|
assert acks[0]["cause"] == routing_refusal_cause("steer_task", "needs_manual_target", "target_closed")
|
|
|
|
|
|
def test_a_headless_root_steering_a_finished_target_is_refused_without_any_chat_row(tmp_path):
|
|
"""The issuer lives in the hidden partition (chat_id 0): its refusal is its
|
|
tool result and its Logs row, and nothing is addressed to any chat."""
|
|
from ouroboros.tools.control import _steer_task
|
|
|
|
acks, notices, emitted = [], [], []
|
|
supervisor = _supervisor(tmp_path, acks=acks, notices=notices)
|
|
ctx = _wire(_pooled_root_ctx(tmp_path, task_id="headless-root", chat_id=0), supervisor, emitted)
|
|
|
|
out = _steer_task(ctx, "t-gone", "are you there")
|
|
|
|
assert out.startswith("⚠️ STEER_REJECTED: task t-gone was not steered (target_unknown)")
|
|
assert notices == [] and acks == []
|
|
[logged] = _events(tmp_path, "task_message_routed")
|
|
assert (logged["task_id"], logged["target_task_id"], logged["reason"]) == ("headless-root", "t-gone", "target_unknown")
|
|
|
|
|
|
def test_a_task_may_message_a_hidden_partition_root(tmp_path):
|
|
"""Owner 6=A: a headless (chat_id 0) root is a host-listed root like any other."""
|
|
from ouroboros.owner_mailbox import drain_owner_entries
|
|
from ouroboros.tools.control import _steer_task
|
|
|
|
emitted = []
|
|
supervisor = _supervisor(tmp_path, running={"t-hidden": {"task": {"id": "t-hidden", "chat_id": 0}}})
|
|
ctx = _wire(_pooled_root_ctx(tmp_path), supervisor, emitted)
|
|
|
|
out = _steer_task(ctx, "t-hidden", "hello there")
|
|
|
|
assert out.startswith("✉️ Message to task t-hidden written")
|
|
assert [row["text"] for row in drain_owner_entries(tmp_path, "t-hidden")] == ["hello there"]
|
|
|
|
|
|
# --- (d) ProjectA root -> ProjectB root, two texts in order -------------------
|
|
|
|
def test_a_project_root_messages_another_projects_root_twice_in_order(tmp_path):
|
|
from ouroboros.owner_mailbox import drain_owner_entries
|
|
from ouroboros.projects_registry import create_project
|
|
from ouroboros.tools.control import _steer_task
|
|
|
|
room_a = create_project(tmp_path, "proj-a", name="Project A")
|
|
room_b = create_project(tmp_path, "proj-b", name="Project B")
|
|
emitted, notices = [], []
|
|
supervisor = _supervisor(
|
|
tmp_path, notices=notices,
|
|
running={"t-b": {"task": {"id": "t-b", "chat_id": room_b["chat_id"], "project_id": "proj-b"}}},
|
|
)
|
|
ctx = _wire(_pooled_root_ctx(
|
|
tmp_path, task_id="t-a", chat_id=room_a["chat_id"], metadata={"project_id": "proj-a"},
|
|
), supervisor, emitted)
|
|
|
|
first = _steer_task(ctx, "t-b", "first: the schema is frozen")
|
|
second = _steer_task(ctx, "t-b", "second: migrations may start")
|
|
|
|
assert first.startswith("✉️ Message to task t-b written") and second.startswith("✉️ Message to task t-b written")
|
|
assert [row["text"] for row in drain_owner_entries(tmp_path, "t-b")] == [
|
|
"first: the schema is frozen", "second: migrations may start",
|
|
]
|
|
assert notices == []
|
|
|
|
|
|
@pytest.mark.parametrize("stamp", ["logged-ref", "suppressed-log"])
|
|
def test_an_owner_turn_in_a_project_room_still_gets_the_room_veto(tmp_path, stamp):
|
|
"""Today's veto for owner turns, computed from the registry lane: a Project
|
|
room turn cannot steer another room's root; Main can (it sees the manifest)."""
|
|
from ouroboros.owner_mailbox import drain_owner_entries
|
|
from ouroboros.projects_registry import create_project
|
|
from ouroboros.tools.control import _steer_task
|
|
|
|
room_a = create_project(tmp_path, "proj-a", name="Project A")
|
|
room_b = create_project(tmp_path, "proj-b", name="Project B")
|
|
emitted, notices = [], []
|
|
supervisor = _supervisor(
|
|
tmp_path, notices=notices,
|
|
running={"t-b": {"task": {"id": "t-b", "chat_id": room_b["chat_id"], "project_id": "proj-b"}}},
|
|
)
|
|
ctx = _wire(_owner_turn_ctx(tmp_path), supervisor, emitted)
|
|
if stamp == "suppressed-log":
|
|
del ctx.task_metadata["origin_message_ref"]
|
|
ctx.task_metadata["origin_suppressed"] = True
|
|
ctx.current_chat_id = room_a["chat_id"]
|
|
|
|
out = _steer_task(ctx, "t-b", "cross-room owner words")
|
|
|
|
assert out.startswith("⚠️ STEER_REJECTED: task t-b was not steered (chat_mismatch)")
|
|
assert drain_owner_entries(tmp_path, "t-b") == []
|
|
|
|
|
|
# --- forward_to_worker: one writer, two verb names ----------------------------
|
|
|
|
def _snapshot(tmp_path, rows):
|
|
from ouroboros.utils import atomic_write_json, utc_now_iso
|
|
|
|
(tmp_path / "state").mkdir(parents=True, exist_ok=True)
|
|
atomic_write_json(tmp_path / "state" / "queue_snapshot.json", {
|
|
"ts": utc_now_iso(), "running": rows, "pending": [], "worker_total": 1,
|
|
})
|
|
|
|
|
|
def test_forward_to_worker_reaches_a_host_listed_root_on_its_own_drive(tmp_path):
|
|
from ouroboros.owner_mailbox import drain_owner_entries
|
|
from ouroboros.task_results import STATUS_RUNNING, write_task_result
|
|
from ouroboros.tools.core import _forward_to_worker
|
|
|
|
root_drive = tmp_path / "root-drive"
|
|
root_drive.mkdir()
|
|
write_task_result(tmp_path, "root-x", STATUS_RUNNING, root_task_id="root-x", result="running")
|
|
write_task_result(tmp_path, "stranger", STATUS_RUNNING, root_task_id="stranger", result="running")
|
|
_snapshot(tmp_path, [{"id": "root-x", "task": {"id": "root-x", "chat_id": 0, "drive_root": str(root_drive)}}])
|
|
ctx = types.SimpleNamespace(drive_root=tmp_path, task_id="sender")
|
|
|
|
out = _forward_to_worker(ctx, "root-x", "the shared schema changed")
|
|
forbidden = _forward_to_worker(ctx, "stranger", "not listed")
|
|
relayed = _forward_to_worker(ctx, "root-x", "relay", relayed_from_task_id="sibling")
|
|
|
|
assert out.startswith("Message forwarded to task root-x: written to its mailbox as a message from this task")
|
|
[row] = drain_owner_entries(root_drive, "root-x")
|
|
assert (row["provenance"], row["source_task_id"], row["text"]) == ("independent_task", "sender", "the shared schema changed")
|
|
assert "TASK_FORBIDDEN" in forbidden and "nor an active independent root" in forbidden
|
|
assert "TASK_FORBIDDEN" in relayed and "independent root" in relayed
|
|
assert drain_owner_entries(tmp_path, "stranger") == []
|
|
|
|
|
|
# --- (e) the roster note: only on change, a fresh row after a send -----------
|
|
|
|
def test_the_roster_note_is_appended_on_change_and_never_rewrites_a_sent_row(tmp_path):
|
|
from ouroboros.peer_roster import maybe_append_roster_note
|
|
from ouroboros.transcript_prefix import observe_send
|
|
|
|
_snapshot(tmp_path, [{"id": "r-1", "task": {"id": "r-1", "title": "Deploy docs", "chat_id": 0, "project_id": "docs"}}])
|
|
ctx = types.SimpleNamespace(task_id="me", task_metadata={"budget_drive_root": str(tmp_path)})
|
|
messages = [{"role": "system", "content": "s"}, {"role": "user", "content": "task"}]
|
|
|
|
assert maybe_append_roster_note(ctx, messages, tmp_path) is True
|
|
assert maybe_append_roster_note(ctx, messages, tmp_path) is False, "unchanged roster: no note"
|
|
# No send witness proves the task tail is unsent, so append the note.
|
|
assert len(messages) == 3
|
|
assert messages[1] == {"role": "user", "content": "task"}
|
|
note = str(messages[-1]["content"])
|
|
assert note.startswith("[System task message]\n[INDEPENDENT_ROOTS]")
|
|
assert "- r-1 · Deploy docs · project=docs · running" in note
|
|
assert "objective" not in note.lower()
|
|
# A root that is the reader itself and a subagent row never appear.
|
|
observe_send(ctx, messages, round_idx=1)
|
|
sent_bytes = json.dumps(messages, ensure_ascii=False).encode("utf-8")
|
|
_snapshot(tmp_path, [
|
|
{"id": "r-1", "task": {"id": "r-1", "title": "Deploy docs", "chat_id": 0, "project_id": "docs"}},
|
|
{"id": "me", "task": {"id": "me", "title": "Myself", "chat_id": 1}},
|
|
{"id": "kid", "task": {"id": "kid", "parent_task_id": "r-1", "delegation_role": "subagent"}},
|
|
{"id": "r-2", "task": {"id": "r-2", "title": "Audit", "chat_id": 7}},
|
|
])
|
|
assert maybe_append_roster_note(ctx, messages, tmp_path) is True
|
|
# The sent row is byte-frozen: the changed roster is a NEW tail row.
|
|
assert len(messages) == 4
|
|
second = str(messages[-1]["content"])
|
|
assert second.startswith("[System task message]\n[INDEPENDENT_ROOTS]")
|
|
assert "- r-2 · Audit · chat=7 · running" in second
|
|
assert "- me ·" not in second and "kid" not in second
|
|
assert note == str(messages[2]["content"])
|
|
assert json.dumps(messages[:-1], ensure_ascii=False).encode("utf-8") == sent_bytes
|
|
|
|
|
|
def test_the_roster_note_includes_direct_roots_but_skips_subagents_and_discloses_gaps(tmp_path):
|
|
from ouroboros.peer_roster import maybe_append_roster_note, render_roster_note
|
|
|
|
_snapshot(tmp_path, [{"id": "r-1", "task": {"id": "r-1", "title": "Deploy docs", "chat_id": 0}}])
|
|
direct = types.SimpleNamespace(task_id="turn", is_direct_chat=True, task_metadata={})
|
|
child = types.SimpleNamespace(task_id="kid", task_metadata={"delegation_role": "subagent"})
|
|
# Main's routing manifest does not carry authored focus. Its first roster
|
|
# view is required, just like any root's, and remains restorable.
|
|
assert maybe_append_roster_note(direct, [], tmp_path) is True
|
|
_snapshot(tmp_path, [{"id": "r-1", "task": {"id": "r-1", "title": "Deploy docs", "chat_id": 0}},
|
|
{"id": "r-2", "task": {"id": "r-2", "title": "New work", "chat_id": 0}}])
|
|
assert maybe_append_roster_note(direct, [], tmp_path) is True
|
|
assert maybe_append_roster_note(child, [], tmp_path) is False
|
|
rendered = render_roster_note({
|
|
"roots": [{"task_id": f"r-{i}", "title": "", "chat_id": 1, "project_id": "", "status": "pending"} for i in range(45)],
|
|
"incomplete": True,
|
|
})
|
|
assert "…and 5 more not shown." in rendered and "roster incomplete" in rendered
|
|
|
|
|
|
@pytest.mark.parametrize("origin", [{}, {"initiator": "consciousness"}], ids=["owner", "consciousness"])
|
|
def test_the_direct_roots_fragment_never_blocks_on_a_held_actor_lock(tmp_path, monkeypatch, origin):
|
|
import threading
|
|
|
|
from supervisor import direct_roots
|
|
from supervisor.active_activity import get_direct_activity_registry
|
|
from ouroboros.peer_roster import independent_roots
|
|
|
|
registry = get_direct_activity_registry()
|
|
lock_free = threading.RLock()
|
|
free_actor = types.SimpleNamespace(
|
|
_owner_message_admission_lock=lock_free, _busy=True, _accepting_owner_messages=True,
|
|
_current_task_id="turn-free", _current_chat_id=1, _current_task_metadata={"title": "Chat", **origin},
|
|
_current_task_text="hello",
|
|
)
|
|
lock_held = threading.Lock()
|
|
lock_held.acquire()
|
|
held_actor = types.SimpleNamespace(
|
|
_owner_message_admission_lock=lock_held, _busy=True, _accepting_owner_messages=True,
|
|
_current_task_id="turn-held", _current_chat_id=1, _current_task_metadata={},
|
|
_current_task_text="busy",
|
|
)
|
|
registry.register("turn-free", 1, actor=free_actor)
|
|
registry.register("turn-held", 1, actor=held_actor)
|
|
try:
|
|
payload = direct_roots.publish_direct_roots(tmp_path)
|
|
finally:
|
|
registry.unregister("turn-free")
|
|
registry.unregister("turn-held")
|
|
lock_held.release()
|
|
|
|
assert [row["task_id"] for row in payload["roots"]] == ["turn-free"]
|
|
assert payload["incomplete"] is True
|
|
_snapshot(tmp_path, [])
|
|
roster = independent_roots(tmp_path)
|
|
assert [row["task_id"] for row in roster["roots"]] == ["turn-free"]
|
|
assert roster["roots"][0]["direct_chat"] is True and roster["incomplete"] is True
|
|
from ouroboros.peer_roster import render_roster_note
|
|
|
|
note = render_roster_note(roster)
|
|
assert "live direct conversation" in note
|
|
assert "live owner conversation" not in note and "person is having" not in note
|
|
assert "turn-free" in note, "the direct lane remains addressable"
|
|
direct_roots.clear_direct_roots(tmp_path)
|
|
assert independent_roots(tmp_path)["roots"] == []
|
|
|
|
|
|
# --- wave 3: ensure on the rail, obligation follows the work ------------------
|
|
|
|
def _ensure_live(tmp_path, supervisor, **ctx_kw):
|
|
from supervisor.events_project_routing import _handle_ensure_project_scope
|
|
|
|
ctx = types.SimpleNamespace(
|
|
project_id="", task_metadata={}, task_contract={}, task_id="t-root", pending_events=[],
|
|
drive_root=tmp_path, event_queue=None,
|
|
)
|
|
for key, value in ctx_kw.items():
|
|
setattr(ctx, key, value)
|
|
ctx.event_queue = types.SimpleNamespace(
|
|
put_nowait=lambda event: _handle_ensure_project_scope(event, supervisor),
|
|
)
|
|
return ctx
|
|
|
|
|
|
@pytest.fixture
|
|
def _projects_root(tmp_path, monkeypatch):
|
|
import ouroboros.config as cfg
|
|
import supervisor.message_bus as mb
|
|
from supervisor import workers
|
|
|
|
monkeypatch.setattr(cfg, "DATA_DIR", tmp_path)
|
|
monkeypatch.setattr(workers, "DRIVE_ROOT", tmp_path)
|
|
monkeypatch.setattr(mb, "get_bridge", lambda: types.SimpleNamespace(broadcast=lambda payload: None))
|
|
monkeypatch.setattr(workers, "_announce_created_project", lambda *a, **kw: None)
|
|
(tmp_path / "logs").mkdir(parents=True, exist_ok=True)
|
|
return tmp_path
|
|
|
|
|
|
def test_a_refused_bind_reaches_the_ensure_caller_as_a_typed_refusal(_projects_root, monkeypatch):
|
|
"""(f) The bind is forced to fail: the tool says so, restores its scope, and
|
|
never says OK; the receipt is readable under the act's own id."""
|
|
import ouroboros.projects_registry as reg
|
|
from ouroboros.project_dialogue import latest_chat_annotations
|
|
from ouroboros.tools.control import _ensure_project_scope
|
|
|
|
tmp_path = _projects_root
|
|
|
|
def _refuse(*_a, **_k):
|
|
raise ValueError("project 'cyber-racing' is deleting; it cannot accept bindings")
|
|
|
|
monkeypatch.setattr(reg, "bind_task_to_project", _refuse)
|
|
supervisor = _supervisor(tmp_path, running={"t-root": {"task": {"id": "t-root", "project_id": ""}}})
|
|
ctx = _ensure_live(tmp_path, supervisor, project_id="")
|
|
|
|
out = _ensure_project_scope(ctx, project_name="Cyber Racing")
|
|
|
|
assert out.startswith("⚠️ SCOPE_REJECTED (project_binding_failed)")
|
|
assert "OK" not in out and "durably bound to no project" in out
|
|
assert ctx.project_id == "" # scope unchanged
|
|
assert supervisor.RUNNING["t-root"]["task"]["project_id"] == ""
|
|
[receipt] = [row for row in latest_chat_annotations(tmp_path).values() if row["action"] == "ensure_project_scope"]
|
|
assert (receipt["status"], receipt["reason"]) == ("rejected", "project_binding_failed")
|
|
assert receipt["client_message_id"].startswith("agent-steer:")
|
|
|
|
|
|
def test_a_bind_that_lands_after_a_lost_ack_is_discoverable_from_the_durable_binding(
|
|
_projects_root, monkeypatch,
|
|
):
|
|
"""(f) The wait times out while the handler still binds: the tool reports
|
|
unconfirmed (never scoped), and the next call reads the durable truth."""
|
|
import ouroboros.tools.control_events as control_events
|
|
from ouroboros.project_facts import project_id_from_display_name
|
|
from ouroboros.projects_registry import project_id_for_task
|
|
from ouroboros.tools.control import _ensure_project_scope
|
|
|
|
tmp_path = _projects_root
|
|
monkeypatch.setattr(
|
|
control_events, "_wait_for_routing_annotation",
|
|
lambda *_a, **_k: {"status": "unconfirmed", "reason": "confirmation_timeout"},
|
|
)
|
|
supervisor = _supervisor(tmp_path, running={"t-root": {"task": {"id": "t-root", "project_id": ""}}})
|
|
ctx = _ensure_live(tmp_path, supervisor)
|
|
pid = project_id_from_display_name("Cyber Racing")
|
|
|
|
out = _ensure_project_scope(ctx, project_name="Cyber Racing")
|
|
|
|
assert out.startswith("⚠️ SCOPE_UNCONFIRMED") and "do not report it as scoped" in out
|
|
assert ctx.project_id == pid # journal writes target it meanwhile
|
|
assert project_id_for_task(tmp_path, "t-root") == pid # the bind did land
|
|
again = _ensure_project_scope(_ensure_live(tmp_path, supervisor), project_name="Cyber Racing")
|
|
assert "already scoped" in again
|
|
|
|
|
|
def _promote_supervisor(tmp_path, running, enqueued):
|
|
from ouroboros.utils import append_jsonl
|
|
|
|
return types.SimpleNamespace(
|
|
DRIVE_ROOT=tmp_path, RUNNING=running, PENDING=[],
|
|
WORKERS={0: types.SimpleNamespace()},
|
|
bridge=types.SimpleNamespace(send_routing_ack=lambda *a, **k: None, broadcast=lambda *a, **k: None),
|
|
enqueue_task=lambda task: enqueued.append(task),
|
|
persist_queue_snapshot=lambda **_k: True,
|
|
load_state=lambda: {"owner_chat_id": 1},
|
|
append_jsonl=append_jsonl,
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize("drained_owner", [False, True])
|
|
def test_an_unmet_swarm_obligation_moves_to_the_promoted_root(_projects_root, monkeypatch, drained_owner):
|
|
"""(h) Through the real admission handler: the new root carries force_plan,
|
|
the promoter is released with a transferred receipt on its live row and its
|
|
task details, the tool result names the move, and the worker's own copy of
|
|
the flag is released so force_plan_decision stops requiring a plan."""
|
|
import supervisor.queue as queue_mod
|
|
from ouroboros.owner_hurry import force_plan_decision
|
|
from ouroboros.task_results import load_task_result
|
|
from ouroboros.tools.control import _promote_chat_to_task
|
|
from supervisor.events_project_routing import _handle_promote_chat_to_task
|
|
|
|
tmp_path = _projects_root
|
|
monkeypatch.setattr(queue_mod, "DRIVE_ROOT", str(tmp_path))
|
|
enqueued: list = []
|
|
running = {"swarm-root": {"task": {"id": "swarm-root", "chat_id": 1,
|
|
"metadata": {"force_plan": True, "force_plan_source": "swarm"}}}}
|
|
supervisor = _promote_supervisor(tmp_path, running, enqueued)
|
|
ctx = _pooled_root_ctx(tmp_path)
|
|
if drained_owner:
|
|
_drain_owner_followup(tmp_path, ctx)
|
|
ctx.event_queue = types.SimpleNamespace(put_nowait=lambda event: _handle_promote_chat_to_task(event, supervisor))
|
|
|
|
out = _promote_chat_to_task(ctx, "Implement the plan in a new root", predecessor_task_id="")
|
|
|
|
assert out.startswith("OK: task")
|
|
[new_root] = enqueued
|
|
assert new_root["metadata"]["force_plan"] is True and new_root["metadata"]["force_plan_source"] == "swarm"
|
|
assert f"Your planning obligation (force_plan) moved to task {new_root['id']}" in out
|
|
assert "use ensure_project_scope" in out and "unplanned" in out
|
|
promoter_row = running["swarm-root"]["task"]["metadata"]
|
|
assert promoter_row["force_plan"] is False and promoter_row["force_plan_transferred_to"] == new_root["id"]
|
|
admission = load_task_result(tmp_path, new_root["id"])["promotion_admission"]
|
|
assert admission["force_plan_transfer"]["from"] == "swarm-root"
|
|
assert admission["force_plan_transfer"]["released"] is True
|
|
assert load_task_result(tmp_path, "swarm-root")["force_plan_transfer"]["to"] == new_root["id"]
|
|
# The worker's copy is released too: no plan is required of the promoter now.
|
|
assert ctx.task_metadata["force_plan"] is False
|
|
assert ctx.task_metadata["force_plan_transferred_to"] == new_root["id"]
|
|
assert force_plan_decision(ctx, {})["status"] == "not_required"
|
|
|
|
|
|
def test_ensure_keeps_the_obligation_and_a_met_one_transfers_nothing(_projects_root, monkeypatch):
|
|
import ouroboros.task_results as task_results
|
|
from ouroboros.owner_hurry import unmet_force_plan_obligation
|
|
from ouroboros.tools.control import _ensure_project_scope
|
|
|
|
tmp_path = _projects_root
|
|
supervisor = _supervisor(tmp_path, running={"swarm-root": {"task": {"id": "swarm-root", "project_id": ""}}})
|
|
ctx = _ensure_live(tmp_path, supervisor, task_id="swarm-root",
|
|
task_metadata={"force_plan": True, "force_plan_source": "swarm"})
|
|
|
|
out = _ensure_project_scope(ctx, project_name="Cyber Racing")
|
|
|
|
assert out.startswith("OK: this task is now durably bound")
|
|
assert ctx.task_metadata["force_plan"] is True # same task: the obligation stays
|
|
assert unmet_force_plan_obligation(ctx) == {"unmet": True, "source": "swarm"}
|
|
monkeypatch.setattr(task_results, "load_plan_review_state",
|
|
lambda _root, _tid: {"schema_version": 2, "waves": [{"request_fingerprint": "f1"}]})
|
|
assert unmet_force_plan_obligation(ctx) == {"unmet": False, "reason": "plan_review_engaged"}
|
|
assert unmet_force_plan_obligation(_owner_turn_ctx(tmp_path)) == {"unmet": False, "reason": "not_required"}
|
|
|
|
|
|
def test_a_promote_into_another_project_discloses_the_second_project(_projects_root, monkeypatch):
|
|
"""(13) Owner 5A: promote stays free, so the result says a second Project
|
|
now holds the work when the request already had one."""
|
|
import ouroboros.tools.control_events as control_events
|
|
from ouroboros.projects_registry import bind_task_to_project, create_project
|
|
from ouroboros.tools.control import _promote_chat_to_task
|
|
|
|
tmp_path = _projects_root
|
|
create_project(tmp_path, "first-room", name="First Room")
|
|
create_project(tmp_path, "second-room", name="Second Room")
|
|
bind_task_to_project(tmp_path, "swarm-root", "first-room", origin={"absent": "system"})
|
|
monkeypatch.setattr(
|
|
control_events, "_wait_for_promotion_admission",
|
|
lambda *_a, **_k: {"status": "scheduled", "effective_project_id": "second-room"},
|
|
)
|
|
ctx = _pooled_root_ctx(tmp_path)
|
|
|
|
out = _promote_chat_to_task(ctx, "Do it elsewhere", project_id="second-room", predecessor_task_id="")
|
|
|
|
assert "already has project 'first-room'" in out and "a second project 'second-room' now holds this promote" in out
|
|
|
|
|
|
def test_a_transfer_admitted_after_the_wait_returned_still_releases_the_worker(tmp_path):
|
|
"""Late admission: the supervisor recorded the transfer on the promoter's task
|
|
result while the tool had already returned unconfirmed; the worker's own
|
|
obligation readers reconcile from that durable record instead of holding
|
|
finalization for a plan the new root owes."""
|
|
from ouroboros.owner_hurry import force_plan_decision, unmet_force_plan_obligation
|
|
from ouroboros.task_results import STATUS_RUNNING, write_task_result
|
|
|
|
write_task_result(tmp_path, "swarm-root", STATUS_RUNNING, result="running",
|
|
force_plan_transfer={"from": "swarm-root", "to": "new-root", "released": True})
|
|
ctx = _pooled_root_ctx(tmp_path)
|
|
assert unmet_force_plan_obligation(ctx) == {"unmet": False, "reason": "transferred"}
|
|
assert ctx.task_metadata["force_plan"] is False
|
|
assert ctx.task_metadata["force_plan_transferred_to"] == "new-root"
|
|
assert force_plan_decision(ctx, {}, enforcement="blocking")["status"] == "not_required"
|
|
|
|
|
|
def test_a_returning_roster_is_re_announced_after_an_intervening_change(tmp_path):
|
|
"""A → B → A: the OLD A row must not suppress the fresh A tail (the model
|
|
would otherwise keep reading B). Only the latest representation counts,
|
|
whether it stands alone or was merged into an unsent owner row."""
|
|
from ouroboros.peer_roster import maybe_append_roster_note
|
|
|
|
ctx = types.SimpleNamespace(task_id="me", task_metadata={"budget_drive_root": str(tmp_path)})
|
|
roster_a = [{"id": "r-1", "task": {"id": "r-1", "title": "Deploy docs", "chat_id": 0, "project_id": "docs"}}]
|
|
roster_b = roster_a + [{"id": "r-2", "task": {"id": "r-2", "title": "Audit", "chat_id": 7}}]
|
|
messages = [{"role": "system", "content": "s"}, {"role": "user", "content": "task"}]
|
|
|
|
_snapshot(tmp_path, roster_a)
|
|
assert maybe_append_roster_note(ctx, messages, tmp_path) is True
|
|
note_a = str(messages[-1]["content"])
|
|
_snapshot(tmp_path, roster_b)
|
|
assert maybe_append_roster_note(ctx, messages, tmp_path) is True
|
|
assert "- r-2 · Audit" in str(messages[-1]["content"])
|
|
_snapshot(tmp_path, roster_a)
|
|
assert maybe_append_roster_note(ctx, messages, tmp_path) is True, "the roster returned to A: announce it again"
|
|
assert str(messages[-1]["content"]) == note_a
|
|
assert maybe_append_roster_note(ctx, messages, tmp_path) is False, "unchanged since the latest note"
|
|
|
|
# The latest representation may be a note merged into an unsent owner row
|
|
# (string or text blocks); it deduplicates exactly like a standalone row.
|
|
merged = [{"role": "user", "content": "owner text\n\n" + note_a}]
|
|
assert maybe_append_roster_note(ctx, merged, tmp_path) is False
|
|
blocks = [{"role": "user", "content": [{"type": "text", "text": "owner text"}, {"type": "text", "text": note_a}]}]
|
|
assert maybe_append_roster_note(ctx, blocks, tmp_path) is False
|
|
|
|
|
|
def test_live_roots_refusals_are_typed_failures_at_the_result_boundary(tmp_path):
|
|
"""A refused or stale-snapshot catalogue read is recorded as a FAILED call,
|
|
not as a successful one carrying an ``error`` key."""
|
|
from ouroboros.tools.recent_tasks import _handle_live_roots
|
|
from ouroboros.tools.tool_result import _structured_failure
|
|
from ouroboros.tool_capabilities import tool_result_limit
|
|
|
|
child = types.SimpleNamespace(drive_root=tmp_path, task_id="kid",
|
|
task_metadata={"parent_task_id": "root", "delegation_role": "subagent"})
|
|
refused = _handle_live_roots(child)
|
|
assert _structured_failure(refused) and json.loads(refused)["host_code"] == "TOOL_FORBIDDEN"
|
|
|
|
_snapshot(tmp_path, [{"id": f"r-{i}", "task": {"id": f"r-{i}", "title": "T" * 80, "chat_id": i, "project_id": f"proj-{i}"}}
|
|
for i in range(100)])
|
|
root = types.SimpleNamespace(drive_root=tmp_path, task_id="me", task_metadata={"budget_drive_root": str(tmp_path)})
|
|
page = _handle_live_roots(root, limit=100)
|
|
assert not _structured_failure(page) and json.loads(page)["returned"] == 100
|
|
# A maximum page is structured JSON: it must fit the result bound the truncator applies.
|
|
assert len(page) < tool_result_limit("live_roots")
|
|
stale = _handle_live_roots(root, limit=100, snapshot="not-the-current-token")
|
|
assert _structured_failure(stale) and json.loads(stale)["host_code"] == "LIVE_ROOTS_SNAPSHOT_CHANGED"
|