"""Tree-wide budget, cancellation, lineage, and live-steering contracts.""" from __future__ import annotations import json import threading import time import pytest import openai4s.agent.delegation as deleg_mod import openai4s.agent.loop as loop_mod from openai4s.agent.delegation import ( FANOUT_CAP, MAX_DEPTH, SESSION_CAP, DelegationBudget, DelegationError, DelegationRunner, ) from openai4s.agent.models import RunState from openai4s.config import get_config from openai4s.store import get_store #: How long to wait for a worker thread to reach its rendezvous. #: #: These are handshakes, not latency assertions — the claims around them #: are about which child stops, which model a child compacts against, and #: whether a grandchild is reachable. It was 2 seconds, which is plenty on #: an idle laptop and not plenty on a loaded CI runner with the #: frozen-shape recorder installed: that is where it failed, having passed #: locally three times in a row and in the CI run before it. A longer wait #: weakens nothing, because an event that never fires still fails the #: test — it just stops failing when the machine is busy. _RENDEZVOUS_TIMEOUT = 30 def _submitted(output=None): return { "stop_reason": "submitted", "submitted_output": { "output": output or {"ok": True}, "completion_bullets": ["Completed child work"], }, "final_message": None, } def _wait_for(predicate, timeout: float = 2.0) -> None: deadline = time.monotonic() + timeout while time.monotonic() < deadline: if predicate(): return time.sleep(0.001) raise AssertionError("condition not reached before timeout") def test_delegated_agent_writes_its_own_canonical_ledger(monkeypatch): arguments = { "summary": "The delegated task is complete.", "completion_bullets": ["Completed delegated work"], } def finalize_chat(messages, cfg, **kwargs): del messages, cfg, kwargs call = { "id": "delegate-finalize", "wire_id": "delegate-finalize", "name": "finalize_response", "ordinal": 0, "raw_arguments": json.dumps(arguments), "arguments": arguments, "parse_error": None, "provider_meta": {"provider": "test"}, } return { "content": "", "tool_calls": [call], "assistant_message": { "role": "assistant", "content": "", "tool_calls": [call], }, } monkeypatch.setattr(loop_mod, "chat", finalize_chat) cfg = get_config() store = get_store(cfg.db_path) parent = store.new_frame(kind="turn", project_id="default") runner = DelegationRunner(cfg, parent_frame_id=parent, store=store) try: result = runner({"request": "Finish this delegated task"}) finally: runner.close() assert result["stop_reason"] == "submitted" child_frame_id = result["frame_id"] assert child_frame_id assert [group["kind"] for group in store.list_action_groups(child_frame_id)] == [ "user", "finalize", "terminal", ] def test_nested_runners_share_one_tree_budget_and_stats(monkeypatch): monkeypatch.setattr(deleg_mod, "SESSION_CAP", 3) def fake_run(self, task): if self.delegate_depth == 1: nested = self.dispatcher._delegate_fn( {"request": ["grandchild-a", "grandchild-b"]} ) assert len(nested) == 2 return _submitted({"task": task, "depth": self.delegate_depth}) monkeypatch.setattr(loop_mod.Agent, "run", fake_run) runner = DelegationRunner(get_config()) result = runner({"request": "root child"}) assert result["output"]["depth"] == 1 assert runner.delegation_stats() == { "total": 3, "direct_total": 1, "running": 0, "done": 3, "failed": 0, "stopped": 0, "pending": 0, "spawned_session": 3, "active_session": 0, "remaining_session_budget": 0, "budget_root_frame_id": None, "depth": 0, } with pytest.raises(DelegationError, match="already spawned 3"): runner({"request": "one child too many"}) @pytest.mark.parametrize( "delegation_spec", [ {"request": "asynchronous child", "wait": False}, {"request": ["fanout-a", "fanout-b"], "wait": True}, ], ) def test_trusted_capture_rejects_parallel_delegation_before_reservation( delegation_spec, ): """Shared-workspace snapshots cannot prove concurrent child authorship.""" runner = DelegationRunner( get_config(), cell_hooks_factory=lambda _frame_id: object(), ) try: with pytest.raises(DelegationError, match="trusted Artifact capture"): runner(delegation_spec) assert runner.children() == [] assert runner.delegation_stats()["spawned_session"] == 0 finally: runner.close() def test_trusted_capture_keeps_single_synchronous_delegation(monkeypatch): monkeypatch.setattr(loop_mod.Agent, "run", lambda self, task: _submitted(task)) runner = DelegationRunner( get_config(), cell_hooks_factory=lambda _frame_id: object(), ) try: result = runner({"request": "serial child", "wait": True}) finally: runner.close() assert result["stop_reason"] == "submitted" assert result["output"] == "serial child" def test_trusted_capture_rejects_a_second_thread_before_reservation(monkeypatch): entered = threading.Event() release = threading.Event() def held_run(self, task): entered.set() assert release.wait(_RENDEZVOUS_TIMEOUT) return _submitted(task) monkeypatch.setattr(loop_mod.Agent, "run", held_run) runner = DelegationRunner( get_config(), cell_hooks_factory=lambda _frame_id: object(), ) result: list[dict] = [] first = threading.Thread( target=lambda: result.append( runner({"request": "first serial child", "wait": True}) ) ) first.start() assert entered.wait(_RENDEZVOUS_TIMEOUT) try: with pytest.raises(DelegationError, match="owns trusted Artifact capture"): runner({"request": "concurrent serial child", "wait": True}) assert runner.delegation_stats()["spawned_session"] == 1 finally: release.set() first.join(_RENDEZVOUS_TIMEOUT) runner.close() assert not first.is_alive() assert result[0]["output"] == "first serial child" def test_trusted_capture_gate_is_reentrant_for_synchronous_descendants(monkeypatch): observed_depths = [] def nested_run(self, task): observed_depths.append(self.delegate_depth) if self.delegate_depth == 1: grandchild = self.dispatcher._delegate_fn( {"request": "grandchild", "wait": True} ) assert grandchild["output"] == "grandchild" return _submitted(task) monkeypatch.setattr(loop_mod.Agent, "run", nested_run) runner = DelegationRunner( get_config(), cell_hooks_factory=lambda _frame_id: object(), ) try: result = runner({"request": "parent", "wait": True}) finally: runner.close() assert result["output"] == "parent" assert observed_depths == [1, 2] assert runner.delegation_stats()["spawned_session"] == 2 def test_trusted_capture_child_cannot_start_a_background_kernel(monkeypatch): observed = {} def inspect_policy(self, task): del task observed["catalog_tool"] = self.dispatcher.tool_catalog().get("exec_background") observed["result"] = self.dispatcher( "exec_background", [{"code": "open('late.txt', 'w').write('late')"}], ) observed["executor"] = self.dispatcher._bg_executor return _submitted() monkeypatch.setattr(loop_mod.Agent, "run", inspect_policy) runner = DelegationRunner( get_config(), cell_hooks_factory=lambda _frame_id: object(), ) try: runner( { "request": "try background work", "permissions": {"background": "allow"}, } ) child = runner.children()[0] finally: runner.close() assert observed["catalog_tool"] is None assert observed["result"] == { "error": "Permission denied by delegated child policy: exec_background" } assert observed["executor"] is None assert child["overrides"]["permissions"]["background"] == "deny" def test_shared_budget_reservation_is_atomic_across_concurrent_runners(monkeypatch): monkeypatch.setattr(loop_mod.Agent, "run", lambda self, task: _submitted()) budget = DelegationBudget("root-session", limit=8) root = DelegationRunner(get_config(), budget=budget) sibling = DelegationRunner(get_config(), budget=budget) barrier = threading.Barrier(17) successes: list[str] = [] failures: list[str] = [] lock = threading.Lock() def spawn(runner, index): barrier.wait() try: result = runner({"request": f"child-{index}"}) except DelegationError as error: with lock: failures.append(str(error)) else: with lock: successes.append(result["child_id"]) threads = [ threading.Thread( target=spawn, args=(root if index % 2 else sibling, index), name=f"spawn-{index}", ) for index in range(16) ] for thread in threads: thread.start() barrier.wait() # The last wait in this file that never got `_RENDEZVOUS_TIMEOUT`, and it # failed the same way the comment up there describes: three seconds each is # plenty on an idle laptop and not plenty on a CI runner sharing itself with # eight other branches, where this suite took 20 minutes against 13 locally. # Reproduced deterministically by making the workers slower without changing # a thing about what they do -- so the assertion was measuring latency while # claiming to measure atomicity. # # One deadline for the whole set rather than per thread: they are released # together by the barrier and run concurrently, so the meaningful budget is # wall-clock for all of them, and a genuine deadlock still fails inside a # bounded time. Nothing below is weakened -- the counts, the identity of the # winners and the budget ledger are what this test is actually about. deadline = time.monotonic() + _RENDEZVOUS_TIMEOUT for thread in threads: thread.join(max(0.0, deadline - time.monotonic())) stuck = [thread.name for thread in threads if thread.is_alive()] assert not stuck, f"threads never returned within {_RENDEZVOUS_TIMEOUT}s: {stuck}" assert len(successes) == 8 assert len(set(successes)) == 8 assert len(failures) == 8 assert root._spawned == sibling._spawned == 8 assert budget.usage() == { "root_frame_id": "root-session", "limit": 8, "spawned": 8, "active": 0, "remaining": 0, } def test_depth_four_is_an_unconditional_leaf(monkeypatch): observed = [] def fake_run(self, task): observed.append( { "depth": self.delegate_depth, "allow_delegate": self.allow_delegate, "delegate_fn": self.dispatcher._delegate_fn, } ) return _submitted() monkeypatch.setattr(loop_mod.Agent, "run", fake_run) parent = DelegationRunner(get_config(), depth=3) parent({"request": "make a leaf"}) assert observed == [{"depth": 4, "allow_delegate": False, "delegate_fn": None}] leaf = DelegationRunner(get_config(), depth=4) with pytest.raises(DelegationError, match="leaves and cannot delegate"): leaf({"request": "must not run"}) class _FakeKernel: instances = [] action_started = threading.Event() release_action = threading.Event() block_actions = False def __init__(self, *args, **kwargs): del args, kwargs self.interrupt_calls = 0 self.action_codes = [] type(self).instances.append(self) def __enter__(self): return self def __exit__(self, *args): del args def execute(self, code, **kwargs): del kwargs if "_sd =" in code: return {"stdout": "", "stderr": "", "error": None} self.action_codes.append(code) type(self).action_started.set() if type(self).block_actions: assert type(self).release_action.wait(_RENDEZVOUS_TIMEOUT) return { "stdout": "", "stderr": "", "error": "Interrupted" if self.interrupt_calls else None, "interrupted": bool(self.interrupt_calls), } def interrupt(self): self.interrupt_calls += 1 type(self).release_action.set() def _reset_fake_kernel(*, block_actions: bool) -> None: _FakeKernel.instances = [] _FakeKernel.action_started = threading.Event() _FakeKernel.release_action = threading.Event() _FakeKernel.block_actions = block_actions def test_stop_child_interrupts_exact_foreground_kernel_and_engine_cancels(monkeypatch): _reset_fake_kernel(block_actions=True) engine_results = [] original_run = loop_mod.Agent.run def record_run(self, task): result = original_run(self, task) engine_results.append(result) return result def fake_chat(messages, cfg, **kwargs): del messages, cfg, kwargs return { "content": "```python\nprint('long scientific cell')\n```", "tool_calls": [], } monkeypatch.setattr(loop_mod, "Kernel", _FakeKernel) monkeypatch.setattr(loop_mod, "chat", fake_chat) monkeypatch.setattr(loop_mod.Agent, "run", record_run) runner = DelegationRunner(get_config(), child_max_turns=2) handle = runner({"request": "run a long cell", "wait": False}) assert _FakeKernel.action_started.wait(_RENDEZVOUS_TIMEOUT) stopped = runner.stop_child(handle["child_id"]) result = runner.collect({"child_ids": [handle["child_id"]]})[0] assert stopped["status"] == "stopped" assert result["stop_reason"] == "stopped" assert result["output"] is None assert engine_results[0]["stop_reason"] == "cancelled" assert len(_FakeKernel.instances) == 1 assert _FakeKernel.instances[0].interrupt_calls == 1 model_cells = [ code for code in _FakeKernel.instances[0].action_codes if "long scientific cell" in code ] assert model_cells == ["print('long scientific cell')\n"] def test_late_model_reply_after_stop_cannot_execute_or_submit(monkeypatch): _reset_fake_kernel(block_actions=False) model_started = threading.Event() release_model = threading.Event() engine_results = [] model_calls = [] original_run = loop_mod.Agent.run def record_run(self, task): result = original_run(self, task) engine_results.append(result) return result def late_chat(messages, cfg, **kwargs): del messages, cfg, kwargs model_calls.append("started") model_started.set() assert release_model.wait(_RENDEZVOUS_TIMEOUT) return { "content": ( "```python\n" "host.submit_output({'summary':'late'}, ['Submitted late'])\n" "```" ), "tool_calls": [], } monkeypatch.setattr(loop_mod, "Kernel", _FakeKernel) monkeypatch.setattr(loop_mod, "chat", late_chat) monkeypatch.setattr(loop_mod.Agent, "run", record_run) runner = DelegationRunner(get_config(), child_max_turns=2) handle = runner({"request": "wait for the model", "wait": False}) assert model_started.wait(_RENDEZVOUS_TIMEOUT) runner.stop_child(handle["child_id"]) release_model.set() result = runner.collect({"child_ids": [handle["child_id"]]})[0] assert engine_results[0]["stop_reason"] == "cancelled" assert result["stop_reason"] == "stopped" assert result["output"] is None assert model_calls == ["started"] # The CLI/delegation Python worker is lazy: cancellation while the model is # still in flight must not even create a worker, much less run the late # model-authored submit cell. assert _FakeKernel.instances == [] def test_parent_stop_propagates_to_running_descendants(monkeypatch): child_ready = threading.Event() grandchild_ready = threading.Event() sibling_ready = threading.Event() release_sibling = threading.Event() nested_handle = {} def cancellable_run(self, task): if self.delegate_depth == 1 and task == "parent child": nested_handle.update( self.dispatcher._delegate_fn({"request": "grandchild", "wait": False}) ) child_ready.set() assert grandchild_ready.wait(_RENDEZVOUS_TIMEOUT) elif self.delegate_depth == 1: sibling_ready.set() assert release_sibling.wait(_RENDEZVOUS_TIMEOUT) assert not self.cancellation.cancelled() return _submitted({"sibling": "unharmed"}) else: grandchild_ready.set() _wait_for(lambda: self.cancellation.cancelled()) return { "stop_reason": "cancelled", "submitted_output": None, "final_message": None, } monkeypatch.setattr(loop_mod.Agent, "run", cancellable_run) runner = DelegationRunner(get_config()) parent = runner({"request": "parent child", "wait": False}) sibling = runner({"request": "sibling child", "wait": False}) assert child_ready.wait(_RENDEZVOUS_TIMEOUT) assert grandchild_ready.wait(_RENDEZVOUS_TIMEOUT) assert sibling_ready.wait(_RENDEZVOUS_TIMEOUT) runner.stop_child(parent["child_id"]) runner.collect({"child_ids": [parent["child_id"]]}) _wait_for(lambda: runner.delegation_stats()["stopped"] == 2) stats = runner.delegation_stats() assert stats["total"] == 3 assert stats["stopped"] == 2 assert stats["running"] == 1 assert runner._children[sibling["child_id"]].stop_event.is_set() is False descendants = runner._tree.descendants(parent["child_id"], include_self=False) assert [child.child_id for child in descendants] == [nested_handle["child_id"]] assert descendants[0].stop_event.is_set() release_sibling.set() sibling_result = runner.collect({"child_ids": [sibling["child_id"]]})[0] assert sibling_result["output"] == {"sibling": "unharmed"} def test_live_steering_is_delivered_at_next_turn_boundary(monkeypatch): events = [] first_boundary = threading.Event() continue_turn = threading.Event() observed_messages = [] def boundary_run(self, task): state = RunState( [ {"role": "system", "content": "system"}, {"role": "user", "content": task}, ], max_turns=self.max_turns, ) self.context_policy.prepare(state) first_boundary.set() assert continue_turn.wait(_RENDEZVOUS_TIMEOUT) state.turn = 1 self.context_policy.prepare(state) observed_messages.extend(state.messages) return _submitted() monkeypatch.setattr(loop_mod.Agent, "run", boundary_run) runner = DelegationRunner(get_config(), event_sink=events.append) handle = runner({"request": "initial task", "wait": False}) assert first_boundary.wait(_RENDEZVOUS_TIMEOUT) queued = runner.send_message( {"child_id": handle["child_id"], "message": "Use the newer dataset"} ) assert queued["status"] == "queued" snapshot = runner.children()[0] assert snapshot["steering"]["queued"] == 1 assert snapshot["steering"]["delivered"] == 0 continue_turn.set() runner.collect({"child_ids": [handle["child_id"]]}) snapshot = runner.children()[0] assert snapshot["steering"]["queued"] == 0 assert snapshot["steering"]["delivered"] == 1 assert snapshot["steering"]["messages"][0]["boundary"] == 2 assert any( message["role"] == "user" and "Use the newer dataset" in message["content"] for message in observed_messages ) assert "steering_queued" in {event["event"] for event in events} assert "steering_delivered" in {event["event"] for event in events} assert ( runner.send_message({"child_id": handle["child_id"], "message": "too late"})[ "status" ] == "rejected" ) def test_child_model_steps_and_policy_overrides_remain_visible(monkeypatch): observed = [] def fake_run(self, task): observed.append( { "task": task, "provider": self.cfg.llm.provider, "model": self.cfg.llm.model, "max_turns": self.max_turns, } ) return _submitted() monkeypatch.setattr(loop_mod.Agent, "run", fake_run) runner = DelegationRunner(get_config()) runner( { "request": { "request": "special work", "model": {"provider": "chatgpt", "model": "special-model"}, "steps": 3, "permissions": {"bash": "deny"}, "capabilities": ["web", "read_file"], } } ) assert observed == [ { "task": "special work", "provider": "chatgpt", "model": "special-model", "max_turns": 3, } ] child = runner.children()[0] assert child["overrides"] == { "model": {"provider": "chatgpt", "model": "special-model"}, "steps": 3, "permissions": {"bash": "deny"}, "capabilities": ["web", "read_file"], } assert child["depth"] == 1 assert child["parent_child_id"] is None assert child["progress"]["max_turns"] == 3 def test_stopping_one_child_leaves_its_siblings_running(monkeypatch): """P1-B's exit criterion, stated directly: cancelling one child or queued item must not affect its siblings. `_stop_subtree` walks `descendants`, which follows `parent_child_id` links, so siblings are structurally outside the walk. That is the right design and it is exactly the kind of thing that survives a refactor into "stop everything under the parent" without anybody noticing, because the common case — one child — behaves identically either way. """ _reset_fake_kernel(block_actions=True) def fake_chat(messages, cfg, **kwargs): del messages, cfg, kwargs return {"content": "```python\nprint('long cell')\n```", "tool_calls": []} monkeypatch.setattr(loop_mod, "Kernel", _FakeKernel) monkeypatch.setattr(loop_mod, "chat", fake_chat) runner = DelegationRunner(get_config(), child_max_turns=2) doomed = runner({"request": "child A", "wait": False}) spared = runner({"request": "child B", "wait": False}) assert _FakeKernel.action_started.wait(_RENDEZVOUS_TIMEOUT) stopped = runner.stop_child(doomed["child_id"]) assert stopped["status"] == "stopped" # The sibling is untouched: not stopped, and still collectable on its own # terms rather than reporting someone else's cancellation. states = {item["child_id"]: item for item in runner.children()} assert states[spared["child_id"]]["status"] != "stopped" runner.stop_child(spared["child_id"]) # clean up the second kernel def test_stopping_a_parent_stops_its_descendants_but_not_a_cousin(monkeypatch): """The other half. A subtree stop must reach grandchildren — otherwise a stopped branch keeps burning budget underneath — while still not touching a branch that merely shares a root. """ from types import SimpleNamespace from openai4s.agent.delegation import _DelegationTree # `descendants` reads exactly two attributes — `child_id` and # `parent_child_id` — so a minimal stand-in is faithful *for this # function*, and building six real `_Child` objects (each needing a store, # a budget and a clock) would test the constructor rather than the walk. tree = _DelegationTree(clock=lambda: 0.0) for child_id, parent in ( ("a", None), ("a1", "a"), ("a2", "a"), ("a1x", "a1"), ("b", None), ("b1", "b"), ): tree.children[child_id] = SimpleNamespace( child_id=child_id, parent_child_id=parent ) reached = {child.child_id for child in tree.descendants("a")} assert reached == {"a", "a1", "a2", "a1x"}, "the subtree walk is wrong" assert "b" not in reached and "b1" not in reached # And a leaf stop reaches only itself. assert {child.child_id for child in tree.descendants("a1x")} == {"a1x"} # -------------------------------------------------------------------------- # a child compacts against its own model's window # -------------------------------------------------------------------------- def test_a_child_compacts_against_its_own_models_window(): """The defect. `_SteeringContextPolicy` built a bare `CompactionPolicy`, which falls back to `cfg.context_window_tokens` — the daemon default of 262,144 — while the Web session path has always derived the budget from the model's declared capability. A child may run a different model than its parent, which is exactly when the two numbers diverge: a model whose usable window is 136,000 tokens would compact against 262,144 and sail past its real limit, learning about it as a provider rejection rather than as a compaction. Asserted against the real capability rather than a literal, since the number belongs to the model and not to this test — but the *inequality* with the daemon default is the point and is checked explicitly. """ import dataclasses from openai4s.agent.delegation import _child_context_budget from openai4s.config import get_config from openai4s.llm import get_model_capabilities base = get_config() cfg = dataclasses.replace( base, llm=dataclasses.replace( base.llm, provider="claude", model="claude-opus-4-20250514" ), ) expected = get_model_capabilities("claude", "claude-opus-4-20250514") budget = _child_context_budget(cfg)(None) assert budget == expected.usable_context_tokens assert ( budget != base.context_window_tokens ), "this model's window matches the daemon default, so the test proves nothing" def test_an_unknown_model_falls_back_rather_than_guessing(): """Returning None restores the previous behaviour on purpose: a model nobody has capabilities for uses the configured default, and a capability lookup that raises must not take the child down with it.""" import dataclasses from openai4s.agent.delegation import _child_context_budget from openai4s.config import get_config base = get_config() cfg = dataclasses.replace( base, llm=dataclasses.replace( base.llm, provider="not-a-provider", model="not-a-model" ), ) assert _child_context_budget(cfg)(None) is None def test_the_policy_actually_installs_the_provider(): """A budget function nothing calls is the shape of the bug it replaces.""" import inspect from openai4s.agent import delegation source = inspect.getsource(delegation._SteeringContextPolicy.__init__) assert "context_budget_provider=" in source assert "_child_context_budget" in source # -------------------------------------------------------------------------- # child step forwarding into the parent session's step sink (D8) # -------------------------------------------------------------------------- def _step(step_id, kind, phase, **extra): if phase == "begin": base = { "phase": "begin", "step_id": step_id, "kind": kind, "title": f"{kind} title", "input": {"query": "x"}, } else: base = { "phase": "end", "step_id": step_id, "status": "done", "output": {}, "summary": "ok", } base.update(extra) return base def test_child_steps_are_forwarded_bounded_and_decorated(monkeypatch): """Meaningful child steps (skills / env / artifacts / delegate / errors) reach the parent session's step sink decorated with the child identity; per-chunk noise kinds are dropped unless they end in an error.""" forwarded = [] def fake_run(self, task): on_step = self.dispatcher.on_step assert on_step is not None, "the forwarder was not installed" on_step(_step("s-skill", "skill", "begin")) on_step(_step("s-skill", "skill", "end", summary="loaded")) # not meaningful and successful: dropped entirely on_step(_step("s-search", "search", "begin")) on_step(_step("s-search", "search", "end")) # not meaningful but ends in an error: both phases relayed on_step(_step("s-fetch", "fetch", "begin")) on_step(_step("s-fetch", "fetch", "end", status="error", summary="failed")) return _submitted() monkeypatch.setattr(loop_mod.Agent, "run", fake_run) runner = DelegationRunner(get_config(), child_step_sink=forwarded.append) try: result = runner({"request": "child task", "name": "step-scout"}) finally: runner.close() assert result["stop_reason"] == "submitted" ids = [(ev["step_id"], ev["phase"]) for ev in forwarded] assert ("s-skill", "begin") in ids and ("s-skill", "end") in ids assert not any(step_id == "s-search" for step_id, _phase in ids) assert ("s-fetch", "begin") in ids and ("s-fetch", "end") in ids begin = next( ev for ev in forwarded if ev["step_id"] == "s-skill" and ev["phase"] == "begin" ) decoration = begin["input"]["delegation"] assert decoration["delegation_child_id"] == result["child_id"] assert decoration["child_name"] == "step-scout" assert decoration["depth"] == 1 assert "child_frame_id" in decoration # the child's own payload is untouched assert begin["input"]["query"] == "x" def test_child_step_flood_is_capped_with_one_elision_marker(monkeypatch): forwarded = [] def fake_run(self, task): on_step = self.dispatcher.on_step for index in range(205): on_step(_step(f"s-{index}", "skill", "begin")) on_step(_step(f"s-{index}", "skill", "end")) return _submitted() monkeypatch.setattr(loop_mod.Agent, "run", fake_run) runner = DelegationRunner(get_config(), child_step_sink=forwarded.append) try: runner({"request": "flood", "name": "flooder"}) finally: runner.close() plain = [ ev for ev in forwarded if not str(ev.get("step_id", "")).startswith("s-elide") ] markers = [ ev for ev in forwarded if str(ev.get("step_id", "")).startswith("s-elide") ] assert len(plain) == 400 # 200 steps x begin+end # exactly one marker step (begin+end) names the elided count assert len(markers) == 2 marker_end = next(ev for ev in markers if ev["phase"] == "end") assert "5 more" in marker_end["summary"] def test_cli_runner_without_step_sink_leaves_child_dispatcher_unwired(monkeypatch): observed = {} def fake_run(self, task): observed["on_step"] = self.dispatcher.on_step return _submitted() monkeypatch.setattr(loop_mod.Agent, "run", fake_run) runner = DelegationRunner(get_config()) try: runner({"request": "cli child"}) finally: runner.close() assert observed["on_step"] is None def test_collect_is_additive_with_request_attempt_and_artifact_refs(monkeypatch): monkeypatch.setattr(loop_mod.Agent, "run", lambda self, task: _submitted(task)) runner = DelegationRunner(get_config()) try: handle = runner({"request": "collect me", "wait": False}) collected = runner.collect({"child_ids": [handle["child_id"]]}) finally: runner.close() assert set(collected[0]) >= {"request_id", "attempt_id", "artifact_refs"} assert collected[0]["artifact_refs"] == [] def test_private_scratch_does_not_lift_trusted_capture_parallel_limit(): runner = DelegationRunner( get_config(), cell_hooks_factory=lambda _frame_id: object(), private_scratch=True, ) try: with pytest.raises(DelegationError, match="trusted Artifact capture"): runner({"request": ["a", "b"]}) assert runner.delegation_stats()["spawned_session"] == 0 assert (FANOUT_CAP, SESSION_CAP, MAX_DEPTH) == (48, 1000, 4) finally: runner.close() # -------------------------------------------------------------------------- # a child policy receives the providers only the run can build # -------------------------------------------------------------------------- def test_steering_policy_binds_only_the_providers_it_lacks(): """`_SteeringContextPolicy` is built before the child's run exists, so it cannot carry the tool catalogue or the kernel cwd. The loop hands them in afterwards; the child's own context budget (its model's window) must not be overwritten by the parent's default in the process.""" from openai4s.agent.delegation import _SteeringContextPolicy from openai4s.agent.runtime import CompactionPolicy policy = object.__new__(_SteeringContextPolicy) own_budget = lambda _state: 136_000 # noqa: E731 - identity is the point policy._base = CompactionPolicy(get_config(), context_budget_provider=own_budget) policy.bind_runtime( context_budget_provider=lambda _state: 262_144, tool_schema_provider=lambda _state: ({"name": "finalize_response"},), workspace_provider=lambda _state: "/child/cwd", ) assert policy._base.context_budget_provider is own_budget assert policy._base.workspace_provider(None) == "/child/cwd" assert policy._base.tool_schema_provider(None)[0]["name"] == "finalize_response" def test_the_loop_hands_a_wrapped_policy_the_run_providers(monkeypatch): """The wiring half. A context policy supplied from outside `Agent.run` used to short-circuit the `or` that built the CLI policy with its providers, so a delegated child priced its trigger without tool schemas and never got the kernel-readable copy of an oversized output.""" import os from openai4s.agent import Agent from openai4s.agent.runtime import CompactionPolicy arguments = { "summary": "Done.", "completion_bullets": ["Answered the question"], } def finalize_chat(messages, cfg, **kwargs): del messages, cfg, kwargs call = { "id": "wrapped-finalize", "wire_id": "wrapped-finalize", "name": "finalize_response", "ordinal": 0, "raw_arguments": json.dumps(arguments), "arguments": arguments, "parse_error": None, "provider_meta": {"provider": "test"}, } return { "content": "", "tool_calls": [call], "assistant_message": { "role": "assistant", "content": "", "tool_calls": [call], }, } class Wrapped: def __init__(self, cfg): self._base = CompactionPolicy(cfg) self.bound: dict = {} def bind_runtime(self, **providers): self.bound = dict(providers) for name, value in providers.items(): if getattr(self._base, name, None) is None: setattr(self._base, name, value) def prepare(self, state): return self._base.prepare(state) monkeypatch.setattr(loop_mod, "chat", finalize_chat) cfg = get_config() wrapped = Wrapped(cfg) result = Agent( cfg=cfg, use_skills=False, allow_delegate=False, max_turns=1, context_policy=wrapped, ).run("submit a short answer") assert result["stop_reason"] == "submitted" assert wrapped._base.workspace_provider(None) == os.getcwd() schemas = wrapped._base.tool_schema_provider(RunState([])) names = {getattr(s, "name", None) or s.get("name") for s in schemas} assert "finalize_response" in names def _controlled_collect_children(runner, count, future_factory): """Real tree identities/Futures, without starting model or kernel work.""" children = [] for index, child_id in enumerate(runner._reserve(count)): child, _ = runner._child_from_reservation( { "child_ids": [child_id], "request_id": f"request-{index}", "attempt_id": f"attempt-{index}", }, {"request": f"controlled {index}"}, ) assert child.begin(1) child.artifact_refs = [ {"artifact_id": f"artifact-{index}", "version_id": f"version-{index}"} ] future = future_factory() assert future.set_running_or_notify_cancel() child.set_future(future) children.append(child) return children @pytest.mark.parametrize("count", [3, 10]) def test_collect_uses_one_deadline_for_all_unfinished_futures(monkeypatch, count): from concurrent.futures import Future from types import SimpleNamespace clock = [0.0] waits = [] class ClockedFuture(Future): def result(self, timeout=None): assert timeout is not None waits.append(timeout) clock[0] += timeout return super().result(timeout=0) runner = DelegationRunner(get_config()) children = _controlled_collect_children(runner, count, ClockedFuture) monkeypatch.setattr(deleg_mod, "time", SimpleNamespace(monotonic=lambda: clock[0])) try: results = runner.collect({"timeout": 1}) assert clock[0] <= 1.000001 assert all(wait <= 0.25 for wait in waits) assert [item["status"] for item in results] == ["running"] * count assert all(not child.future.done() for child in children) finally: for child in children: child.future.set_result(None) runner.close() def test_collect_none_observes_parent_cancel_without_waiting_for_futures(): from concurrent.futures import Future entered = threading.Event() cancelled = threading.Event() returned = threading.Event() result = [] class ObservedFuture(Future): def result(self, timeout=None): entered.set() return super().result(timeout=timeout) runner = DelegationRunner(get_config()) runner.cancelled = cancelled.is_set children = _controlled_collect_children(runner, 3, ObservedFuture) def collect(): try: result.extend(runner.collect({"timeout": None})) finally: returned.set() collector = threading.Thread(target=collect, daemon=True) collector.start() try: assert entered.wait(_RENDEZVOUS_TIMEOUT) cancelled.set() assert returned.wait(1), "parent cancellation did not end collect(None)" assert len(result) == 3 assert all(child.stop_event.is_set() for child in children) # Cancellation is a request, not proof that a running Future exited. assert all(not child.future.done() for child in children) finally: for child in children: if not child.future.done(): child.future.set_result(None) collector.join(_RENDEZVOUS_TIMEOUT) runner.close() @pytest.mark.parametrize( "value", [True, False, -0.01, 3601, float("inf"), float("-inf"), float("nan"), "1", {}, []], ) def test_collect_rejects_invalid_timeout_before_waiting(value): runner = DelegationRunner(get_config()) try: with pytest.raises(ValueError, match="timeout"): runner.collect({"timeout": value}) finally: runner.close() def test_collect_zero_is_nonblocking_snapshot_with_all_result_identities(): from concurrent.futures import Future class NoWaitFuture(Future): def result(self, timeout=None): raise AssertionError("zero timeout waited on an unfinished Future") runner = DelegationRunner(get_config()) children = _controlled_collect_children(runner, 2, NoWaitFuture) try: results = runner.collect({"timeout": 0}) for index, item in enumerate(results): assert item["status"] == "running" assert item["child_id"] == children[index].child_id assert item["request_id"] == f"request-{index}" assert item["attempt_id"] == f"attempt-{index}" assert item["artifact_refs"] == children[index].artifact_refs assert not children[index].stop_event.is_set() finally: for child in children: child.future.set_result(None) runner.close() def test_collect_fast_result_is_not_lost_while_later_child_uses_remaining_budget( monkeypatch, ): from concurrent.futures import Future from types import SimpleNamespace clock = [0.0] children = [] class CompletingFuture(Future): def result(self, timeout=None): if self.done(): return super().result(timeout=0) clock[0] += timeout if clock[0] >= 0.75 and not children[0].future.done(): children[0].finish_done( {"task_status": "completed", "output": "fast result"} ) children[0].future.set_result(None) return super().result(timeout=0) runner = DelegationRunner(get_config()) children.extend(_controlled_collect_children(runner, 2, CompletingFuture)) monkeypatch.setattr(deleg_mod, "time", SimpleNamespace(monotonic=lambda: clock[0])) try: result = runner.collect({"timeout": 1}) assert clock[0] == 1 assert result[0]["output"] == "fast result" assert result[0]["request_id"] == "request-0" assert result[0]["artifact_refs"] == children[0].artifact_refs assert result[1]["status"] == "running" finally: for child in children: if not child.future.done(): child.future.set_result(None) runner.close() def test_collect_does_not_hold_tree_lock_while_future_waits(): from concurrent.futures import Future entered = threading.Event() errors = [] class ObservedFuture(Future): def result(self, timeout=None): entered.set() return super().result(timeout=timeout) runner = DelegationRunner(get_config()) children = _controlled_collect_children(runner, 1, ObservedFuture) def collect(): try: runner.collect({}) except BaseException as error: errors.append(error) collector = threading.Thread(target=collect, daemon=True) collector.start() try: assert entered.wait(_RENDEZVOUS_TIMEOUT) assert runner._tree.lock.acquire(timeout=1), "tree lock blocked by Future wait" runner._tree.lock.release() finally: children[0].future.set_result(None) collector.join(_RENDEZVOUS_TIMEOUT) runner.close() assert not collector.is_alive() assert not errors def test_collect_one_second_budget_wall_clock_smoke(): from concurrent.futures import Future runner = DelegationRunner(get_config()) children = _controlled_collect_children(runner, 3, Future) try: started = time.monotonic() result = runner.collect({"timeout": 1}) elapsed = time.monotonic() - started assert elapsed < 1.5 assert len(result) == 3 assert all(item["status"] == "running" for item in result) finally: for child in children: child.future.set_result(None) runner.close() @pytest.mark.parametrize("entry", ["native", "python"]) @pytest.mark.stubbed_backend def test_agent_cancel_probe_reaches_collect_through_native_and_python(entry): from concurrent.futures import Future from types import SimpleNamespace from openai4s.host_dispatch import HostDispatcher from openai4s.kernel import Kernel from openai4s.tools import execute_tool_call entered = threading.Event() cancelled = threading.Event() returned = threading.Event() outputs = [] errors = [] class ObservedFuture(Future): def result(self, timeout=None): entered.set() return super().result(timeout=timeout) cfg = get_config() agent = loop_mod.Agent( cfg, dispatcher=HostDispatcher(cfg), cancellation=SimpleNamespace(cancelled=cancelled.is_set), ) runner = agent._delegation_runner children = _controlled_collect_children(runner, 2, ObservedFuture) kernel = Kernel(dispatcher=agent.dispatcher) if entry == "python" else None def collect(): try: if kernel is None: outputs.append( execute_tool_call( agent.dispatcher, {"name": "collect_children", "arguments": {}} ) ) else: outputs.append( kernel.execute( "rows = host.collect()\nprint(len(rows), all(row['stop_reason'] for row in rows))" ) ) except BaseException as error: errors.append(error) finally: returned.set() collector = threading.Thread(target=collect, daemon=True) collector.start() try: assert entered.wait(_RENDEZVOUS_TIMEOUT) cancelled.set() assert returned.wait(1) assert not errors assert all(child.stop_event.is_set() for child in children) assert all(not child.future.done() for child in children) assert agent.dispatcher.last_output is None if kernel is None: text, ok = outputs[0] assert ok and "stopped" in text else: assert outputs[0]["error"] is None assert outputs[0]["stdout"].strip() == "2 True" assert kernel.is_alive() finally: for child in children: if not child.future.done(): child.future.set_result(None) collector.join(_RENDEZVOUS_TIMEOUT) if kernel is not None: kernel.shutdown() runner.close() @pytest.mark.stubbed_backend def test_collect_invalid_timeout_is_rejected_at_native_and_python_boundaries(): from openai4s.host_dispatch import HostDispatcher from openai4s.kernel import Kernel from openai4s.tools import execute_tool_call dispatcher = HostDispatcher(get_config()) invoked = [] dispatcher.steer_fns = {"collect": lambda spec: invoked.append(spec) or []} invalid = [True, False, -1, 3601, float("inf"), float("nan"), "1"] for value in invalid: observation, ok = execute_tool_call( dispatcher, {"name": "collect_children", "arguments": {"timeout": value}} ) assert not ok and "timeout" in observation with Kernel(dispatcher=dispatcher) as kernel: result = kernel.execute( "for timeout in (True, False, -1, 3601, float('inf'), float('nan'), '1'):\n" " try:\n" " host.collect(timeout=timeout)\n" " except RuntimeError:\n" " print('rejected')\n" ) assert result["error"] is None assert result["stdout"].strip().splitlines() == ["rejected"] * len(invalid) assert invoked == [] def test_parent_cancel_targets_owned_subtree_even_when_collect_selects_one(): from concurrent.futures import Future cancelled = threading.Event() runner = DelegationRunner(get_config(), cancelled=cancelled.is_set) children = _controlled_collect_children(runner, 2, Future) nested = DelegationRunner( get_config(), depth=1, delegation_tree=runner._tree, parent_child_id=children[0].child_id, ) grandchildren = _controlled_collect_children(nested, 1, Future) try: cancelled.set() result = runner.collect({"child_ids": [children[0].child_id], "timeout": 0}) assert len(result) == 1 assert all(child.stop_event.is_set() for child in children + grandchildren) assert all(not child.future.done() for child in children + grandchildren) finally: for child in children + grandchildren: child.future.set_result(None) nested.close() runner.close() def test_single_child_stop_and_collect_leave_sibling_running(): from concurrent.futures import Future runner = DelegationRunner(get_config()) children = _controlled_collect_children(runner, 2, Future) try: runner.stop_child(children[0].child_id) result = runner.collect({"timeout": 0}) assert children[0].stop_event.is_set() assert not children[1].stop_event.is_set() assert result[1]["status"] == "running" assert all(not child.future.done() for child in children) finally: for child in children: child.future.set_result(None) runner.close() @pytest.mark.parametrize( "failure", [ValueError("child failed"), TimeoutError("child operation timed out")] ) def test_collect_completed_future_failure_keeps_real_failed_state(failure): from concurrent.futures import Future runner = DelegationRunner(get_config()) child = _controlled_collect_children(runner, 1, Future)[0] child.future.set_exception(failure) try: result = runner.collect({"timeout": 0})[0] assert result["task_status"] == "failed" assert result["error"] == str(failure) assert result["request_id"] == child.request_id assert child.snapshot()["status"] == "failed" finally: runner.close() def test_collect_cancelled_pending_future_reports_stopped_without_affecting_sibling(): from concurrent.futures import Future runner = DelegationRunner(get_config()) children = _controlled_collect_children(runner, 2, Future) children[0].future.set_result(None) children[0].set_future(Future()) assert children[0].future.cancel() try: result = runner.collect({"timeout": 0}) assert children[0].snapshot()["status"] == "stopped" assert result[1]["status"] == "running" assert not children[1].stop_event.is_set() finally: children[1].future.set_result(None) runner.close() @pytest.mark.stubbed_backend def test_web_session_runner_binds_its_own_cancel_event_to_collect(): from concurrent.futures import Future from types import SimpleNamespace from openai4s.host_dispatch import HostDispatcher from openai4s.server.gateway import SessionRunner, SessionState cfg = get_config() session_runner = SessionRunner( cfg, SimpleNamespace(emitter=lambda _: lambda event: None), start_idle_sweeper=False, ) root = session_runner.store.new_frame( kind="turn", project_id="default", status="ready" ) state = SessionState(root, "default", session_runner.workspace_for(root)) dispatcher = HostDispatcher(cfg, frame_id=root) children = [] try: session_runner._wire_delegation(state, dispatcher) runner = state.delegation_runner assert runner is not None assert runner.cancelled() is False children = _controlled_collect_children(runner, 1, Future) state.cancel.set() assert runner.cancelled() is True result = dispatcher("collect", [{"timeout": 0}]) assert result[0]["child_id"] == children[0].child_id assert children[0].stop_event.is_set() assert not children[0].future.done() finally: for child in children: if not child.future.done(): child.future.set_result(None) if state.delegation_runner is not None: state.delegation_runner.close() session_runner.close()