mirror of
https://github.com/razzant/ouroboros.git
synced 2026-10-03 04:07:04 +00:00
75d763f5fa
WS1: a burst of schedule_subagent calls in one tool-call round now runs in the existing tool ThreadPool instead of sequentially. tool_capabilities.PARALLEL_SAFE_ENQUEUE_TOOLS marks schedule_subagent as a fire-and-forget enqueue tool; loop_tool_execution. tool_calls_can_run_parallel parallelizes any round whose calls are all read-only-parallel or enqueue-safe. WS4: a process-local _SCHEDULE_EMIT_LOCK in tools/control.py serializes the parent-side _last_scheduled_subagents read-modify-write (_record_scheduled_subagent) and the pending_events fallback, so concurrent emission cannot lose records. The supervisor still drains its event queue serially, so cap/dedup/enqueue stay single-threaded. No tool-schema or gateway-contract change. Tests: tests/test_swarm_parallel_spawn.py (gate semantics + concurrency no-lost-update).
76 lines
2.3 KiB
Python
76 lines
2.3 KiB
Python
"""WS1+WS4: parallel schedule_subagent emission gate + parent-side thread safety."""
|
|
from __future__ import annotations
|
|
|
|
import threading
|
|
import types
|
|
|
|
from ouroboros.loop_tool_execution import tool_calls_can_run_parallel
|
|
from ouroboros.tool_capabilities import (
|
|
PARALLEL_SAFE_ENQUEUE_TOOLS,
|
|
READ_ONLY_PARALLEL_TOOLS,
|
|
)
|
|
from ouroboros.tools.control import _emit_control_event, _record_scheduled_subagent
|
|
|
|
|
|
def _call(name):
|
|
return {"function": {"name": name}}
|
|
|
|
|
|
# --- WS1: parallel gate ---
|
|
def test_schedule_subagent_parallel_safe_not_readonly():
|
|
assert "schedule_subagent" in PARALLEL_SAFE_ENQUEUE_TOOLS
|
|
assert "schedule_subagent" not in READ_ONLY_PARALLEL_TOOLS
|
|
|
|
|
|
def test_burst_of_spawns_runs_parallel():
|
|
assert tool_calls_can_run_parallel([_call("schedule_subagent")] * 4) is True
|
|
|
|
|
|
def test_mixed_reads_and_spawns_runs_parallel():
|
|
assert tool_calls_can_run_parallel(
|
|
[_call("read_file"), _call("schedule_subagent"), _call("search_code")]
|
|
) is True
|
|
|
|
|
|
def test_any_non_parallel_tool_forces_sequential():
|
|
assert tool_calls_can_run_parallel(
|
|
[_call("schedule_subagent"), _call("write_file")]
|
|
) is False
|
|
|
|
|
|
def test_single_call_is_not_parallel():
|
|
assert tool_calls_can_run_parallel([_call("schedule_subagent")]) is False
|
|
|
|
|
|
# --- WS4: parent-side thread safety (no lost updates under concurrency) ---
|
|
def _fake_ctx():
|
|
# event_queue=None forces the pending_events fallback in _emit_control_event.
|
|
return types.SimpleNamespace(event_queue=None, pending_events=[], _last_scheduled_subagents=[])
|
|
|
|
|
|
def _run_concurrently(fn, n=200):
|
|
start = threading.Event()
|
|
|
|
def worker(i):
|
|
start.wait()
|
|
fn(i)
|
|
|
|
threads = [threading.Thread(target=worker, args=(i,)) for i in range(n)]
|
|
for t in threads:
|
|
t.start()
|
|
start.set()
|
|
for t in threads:
|
|
t.join()
|
|
|
|
|
|
def test_concurrent_record_scheduled_subagent_no_lost_updates():
|
|
ctx = _fake_ctx()
|
|
_run_concurrently(lambda i: _record_scheduled_subagent(ctx, {"task_ids": [f"t{i}"]}))
|
|
assert len(ctx._last_scheduled_subagents) == 200
|
|
assert {r["task_ids"][0] for r in ctx._last_scheduled_subagents} == {f"t{i}" for i in range(200)}
|
|
|
|
|
|
def test_concurrent_emit_control_event_fallback_no_lost_updates():
|
|
ctx = _fake_ctx()
|
|
_run_concurrently(lambda i: _emit_control_event(ctx, {"type": "x", "i": i}))
|
|
assert len(ctx.pending_events) == 200
|