wehub-resource-sync
385 lines
13 KiB
Python
385 lines
13 KiB
Python
"""End-to-end: MetaOrchestrator + MetaRunWriter wiring."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from opensquilla.persistence.meta_run_writer import open_meta_run_writer
|
|
from opensquilla.persistence.migrator import apply_pending
|
|
from opensquilla.skills.meta.types import MetaMatch, MetaPlan, MetaResult, MetaStep
|
|
|
|
MIGRATIONS_DIR = Path(__file__).resolve().parents[1].parent / "migrations"
|
|
|
|
|
|
@pytest.fixture
|
|
def writer_db(tmp_path: Path):
|
|
db = str(tmp_path / "test.db")
|
|
apply_pending(db, MIGRATIONS_DIR)
|
|
w = open_meta_run_writer(db)
|
|
yield w
|
|
w.close()
|
|
|
|
|
|
def _stub_runner_for(outputs: dict[str, str]):
|
|
from opensquilla.skills.meta.events import _StepDone
|
|
|
|
async def _runner(step, effective_skill, inputs, outputs_so_far):
|
|
text = outputs.get(step.id, f"output-of-{step.id}")
|
|
if text == "__FAIL__":
|
|
raise RuntimeError(f"{step.id} exploded")
|
|
yield _StepDone(text=text)
|
|
return _runner
|
|
|
|
|
|
async def _drive_orchestrator(writer, plan: MetaPlan, outputs: dict[str, str]) -> MetaResult:
|
|
from opensquilla.skills.meta.orchestrator import MetaOrchestrator
|
|
|
|
async def stub_skill_loader_dummy():
|
|
return None
|
|
|
|
orch = MetaOrchestrator(
|
|
agent_runner=lambda *a, **kw: None,
|
|
skill_loader=stub_skill_loader_dummy,
|
|
run_writer=writer,
|
|
triggered_by="soft_meta_invoke",
|
|
session_key="sess-test",
|
|
turn_id="turn-test",
|
|
)
|
|
|
|
# Patch dispatch to use stub
|
|
orch._dispatch_step_stream = _stub_runner_for(outputs) # type: ignore[assignment]
|
|
|
|
async def empty_preface(step_id, effective_skill):
|
|
if False:
|
|
yield None
|
|
return
|
|
orch._yield_skill_view_preface = empty_preface # type: ignore[assignment]
|
|
|
|
match = MetaMatch(plan=plan, inputs={"user_message": "hi"})
|
|
final: MetaResult | None = None
|
|
async for item in orch.iter_events(match):
|
|
if isinstance(item, MetaResult):
|
|
final = item
|
|
return final or MetaResult(ok=False, error="no result")
|
|
|
|
|
|
async def _drive_orchestrator_with_usage(writer, plan: MetaPlan) -> MetaResult:
|
|
from opensquilla.engine.usage import UsageTracker
|
|
from opensquilla.skills.meta.events import _StepDone
|
|
from opensquilla.skills.meta.orchestrator import MetaOrchestrator
|
|
|
|
tracker = UsageTracker()
|
|
orch = MetaOrchestrator(
|
|
agent_runner=lambda *a, **kw: None,
|
|
skill_loader=lambda: None,
|
|
run_writer=writer,
|
|
triggered_by="soft_meta_invoke",
|
|
session_key="sess-test",
|
|
turn_id="turn-test",
|
|
usage_tracker=tracker,
|
|
)
|
|
|
|
async def dispatch(step, effective_skill, inputs, outputs_so_far):
|
|
tracker.add(
|
|
"sess-test",
|
|
input_tokens=31,
|
|
output_tokens=7,
|
|
model_id="usage-model",
|
|
billed_cost=0.019,
|
|
)
|
|
yield _StepDone(text="usage-output")
|
|
|
|
async def empty_preface(step_id, effective_skill):
|
|
if False:
|
|
yield None
|
|
return
|
|
|
|
orch._dispatch_step_stream = dispatch # type: ignore[assignment]
|
|
orch._yield_skill_view_preface = empty_preface # type: ignore[assignment]
|
|
final: MetaResult | None = None
|
|
async for item in orch.iter_events(MetaMatch(plan=plan, inputs={"user_message": "hi"})):
|
|
if isinstance(item, MetaResult):
|
|
final = item
|
|
return final or MetaResult(ok=False, error="no result")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_linear_success_writes_run_and_steps(writer_db) -> None:
|
|
plan = MetaPlan(
|
|
name="linear",
|
|
triggers=("t",),
|
|
priority=10,
|
|
steps=(
|
|
MetaStep(id="s1", skill="alpha", kind="agent"),
|
|
MetaStep(id="s2", skill="beta", kind="agent", depends_on=("s1",)),
|
|
),
|
|
)
|
|
result = await _drive_orchestrator(writer_db, plan, outputs={"s1": "A", "s2": "B"})
|
|
assert result.ok
|
|
|
|
runs = writer_db.list_runs(name="linear")
|
|
assert len(runs) == 1
|
|
run = writer_db.get_run(runs[0].run_id)
|
|
assert run is not None
|
|
assert run.status == "ok"
|
|
assert run.triggered_by == "soft_meta_invoke"
|
|
assert run.session_key == "sess-test"
|
|
assert {s.step_id for s in run.steps} == {"s1", "s2"}
|
|
assert all(s.status == "ok" for s in run.steps)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_manual_command_trigger_writes_run(writer_db) -> None:
|
|
from opensquilla.skills.meta.orchestrator import MetaOrchestrator
|
|
|
|
plan = MetaPlan(
|
|
name="manual-linear",
|
|
triggers=("t",),
|
|
priority=10,
|
|
steps=(MetaStep(id="s1", skill="alpha", kind="agent"),),
|
|
)
|
|
orch = MetaOrchestrator(
|
|
agent_runner=lambda *a, **kw: None,
|
|
skill_loader=lambda: None,
|
|
run_writer=writer_db,
|
|
triggered_by="manual_command",
|
|
session_key="sess-test",
|
|
turn_id="turn-test",
|
|
)
|
|
orch._dispatch_step_stream = _stub_runner_for({"s1": "A"}) # type: ignore[assignment]
|
|
|
|
async def empty_preface(step_id, effective_skill):
|
|
if False:
|
|
yield None
|
|
return
|
|
|
|
orch._yield_skill_view_preface = empty_preface # type: ignore[assignment]
|
|
|
|
async for _ in orch.iter_events(MetaMatch(plan=plan, inputs={"user_message": "/meta"})):
|
|
pass
|
|
|
|
[row] = writer_db.list_runs(name="manual-linear")
|
|
run = writer_db.get_run(row.run_id)
|
|
assert run is not None
|
|
assert run.status == "ok"
|
|
assert run.triggered_by == "manual_command"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_live_run_persists_scoped_step_usage(writer_db) -> None:
|
|
plan = MetaPlan(
|
|
name="usage-linear",
|
|
triggers=("t",),
|
|
priority=10,
|
|
steps=(MetaStep(id="s1", skill="alpha", kind="agent"),),
|
|
)
|
|
result = await _drive_orchestrator_with_usage(writer_db, plan)
|
|
assert result.ok
|
|
|
|
[row] = writer_db.list_runs(name="usage-linear")
|
|
run = writer_db.get_run(row.run_id)
|
|
assert run is not None
|
|
summary = writer_db.hydrate_runs([row])[0]
|
|
assert summary.run_id == run.run_id
|
|
|
|
from opensquilla.persistence.meta_run_writer import summarize_run_record
|
|
|
|
run_summary = summarize_run_record(run)
|
|
assert run_summary["usage"]["available"] is True
|
|
assert run_summary["usage"]["input_tokens"] == 31
|
|
assert run_summary["usage"]["output_tokens"] == 7
|
|
assert run_summary["usage"]["cost_usd"] == pytest.approx(0.019)
|
|
assert run_summary["steps"][0]["usage"]["model"] == "usage-model"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_on_failure_records_substituted(writer_db) -> None:
|
|
"""C3: original step gets status='substituted', substitute gets own ok."""
|
|
plan = MetaPlan(
|
|
name="failover",
|
|
triggers=("t",),
|
|
priority=10,
|
|
steps=(
|
|
MetaStep(id="primary", skill="alpha", kind="agent", on_failure="backup"),
|
|
MetaStep(id="backup", skill="beta", kind="agent"),
|
|
),
|
|
)
|
|
await _drive_orchestrator(writer_db, plan, outputs={"primary": "__FAIL__", "backup": "OK"})
|
|
runs = writer_db.list_runs(name="failover")
|
|
run = writer_db.get_run(runs[0].run_id)
|
|
assert run is not None
|
|
by_id = {s.step_id: s for s in run.steps}
|
|
assert by_id["primary"].status == "substituted"
|
|
assert by_id["primary"].substitute_step_id == "backup"
|
|
assert by_id["backup"].status == "ok"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_on_failure_persists_failed_step_usage(writer_db) -> None:
|
|
from opensquilla.engine.usage import UsageTracker
|
|
from opensquilla.skills.meta.events import _StepDone
|
|
from opensquilla.skills.meta.orchestrator import MetaOrchestrator
|
|
|
|
plan = MetaPlan(
|
|
name="failover-usage",
|
|
triggers=("t",),
|
|
priority=10,
|
|
steps=(
|
|
MetaStep(id="primary", skill="alpha", kind="agent", on_failure="backup"),
|
|
MetaStep(id="backup", skill="beta", kind="agent"),
|
|
),
|
|
)
|
|
tracker = UsageTracker()
|
|
orch = MetaOrchestrator(
|
|
agent_runner=lambda *a, **kw: None,
|
|
skill_loader=lambda: None,
|
|
run_writer=writer_db,
|
|
triggered_by="soft_meta_invoke",
|
|
session_key="sess-test",
|
|
turn_id="turn-test",
|
|
usage_tracker=tracker,
|
|
)
|
|
|
|
async def dispatch(step, effective_skill, inputs, outputs_so_far):
|
|
if step.id == "primary":
|
|
tracker.add(
|
|
"sess-test",
|
|
input_tokens=40,
|
|
output_tokens=4,
|
|
model_id="primary-model",
|
|
billed_cost=0.04,
|
|
)
|
|
raise RuntimeError("primary exploded after tokens")
|
|
tracker.add(
|
|
"sess-test",
|
|
input_tokens=5,
|
|
output_tokens=1,
|
|
model_id="backup-model",
|
|
billed_cost=0.005,
|
|
)
|
|
yield _StepDone(text="backup-ok")
|
|
|
|
async def empty_preface(step_id, effective_skill):
|
|
if False:
|
|
yield None
|
|
return
|
|
|
|
orch._dispatch_step_stream = dispatch # type: ignore[assignment]
|
|
orch._yield_skill_view_preface = empty_preface # type: ignore[assignment]
|
|
|
|
async for _ in orch.iter_events(MetaMatch(plan=plan, inputs={"user_message": "hi"})):
|
|
pass
|
|
|
|
[row] = writer_db.list_runs(name="failover-usage")
|
|
run = writer_db.get_run(row.run_id)
|
|
assert run is not None
|
|
from opensquilla.persistence.meta_run_writer import summarize_run_record
|
|
|
|
run_summary = summarize_run_record(run)
|
|
status_by_step = {step.step_id: step.status for step in run.steps}
|
|
by_step = {
|
|
step["step_id"]: step["usage"]
|
|
for step in run_summary["steps"]
|
|
}
|
|
assert status_by_step["primary"] == "substituted"
|
|
assert by_step["primary"]["available"] is True
|
|
assert by_step["primary"]["input_tokens"] == 40
|
|
assert by_step["primary"]["model"] == "primary-model"
|
|
assert run_summary["usage"]["cost_usd"] == pytest.approx(0.045)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_hard_failure_marks_step_failed(writer_db) -> None:
|
|
"""codex-a P2 fix: hard step failure (no on_failure) marks the step row as 'failed'."""
|
|
plan = MetaPlan(
|
|
name="hard-fail", triggers=("t",), priority=10,
|
|
steps=(MetaStep(id="boom", skill="alpha", kind="agent"),),
|
|
)
|
|
await _drive_orchestrator(writer_db, plan, outputs={"boom": "__FAIL__"})
|
|
runs = writer_db.list_runs(name="hard-fail")
|
|
run = writer_db.get_run(runs[0].run_id)
|
|
assert run is not None
|
|
assert run.status == "failed"
|
|
# Critical: step row must be 'failed' not 'running'
|
|
assert run.steps[0].status == "failed"
|
|
assert run.steps[0].error # non-empty
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cancellation_marks_cancelled(writer_db) -> None:
|
|
"""W5: orchestrator cancelled mid-run → status='cancelled'."""
|
|
plan = MetaPlan(
|
|
name="slow", triggers=("t",), priority=10,
|
|
steps=(MetaStep(id="s1", skill="alpha", kind="agent"),),
|
|
)
|
|
from opensquilla.skills.meta.orchestrator import MetaOrchestrator
|
|
orch = MetaOrchestrator(
|
|
agent_runner=lambda *a, **kw: None,
|
|
skill_loader=lambda: None,
|
|
run_writer=writer_db,
|
|
triggered_by="soft_meta_invoke",
|
|
session_key=None, turn_id=None,
|
|
)
|
|
|
|
dispatch_started = asyncio.Event()
|
|
|
|
async def slow_dispatch(*args, **kwargs):
|
|
dispatch_started.set()
|
|
await asyncio.sleep(10)
|
|
if False:
|
|
yield None
|
|
|
|
async def empty_preface(step_id, effective_skill):
|
|
if False:
|
|
yield None
|
|
return
|
|
|
|
orch._dispatch_step_stream = slow_dispatch # type: ignore[assignment]
|
|
orch._yield_skill_view_preface = empty_preface # type: ignore[assignment]
|
|
match = MetaMatch(plan=plan, inputs={})
|
|
|
|
async def consume():
|
|
async for _ in orch.iter_events(match):
|
|
pass
|
|
|
|
task = asyncio.create_task(consume())
|
|
await asyncio.wait_for(dispatch_started.wait(), timeout=2.0)
|
|
task.cancel()
|
|
try:
|
|
await task
|
|
except asyncio.CancelledError:
|
|
pass
|
|
|
|
runs = writer_db.list_runs(name="slow")
|
|
assert len(runs) == 1
|
|
assert runs[0].status == "cancelled"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_writer_none_legacy_path(writer_db) -> None:
|
|
"""Regression: run_writer=None means zero rows written."""
|
|
from opensquilla.skills.meta.orchestrator import MetaOrchestrator
|
|
plan = MetaPlan(
|
|
name="legacy", triggers=("t",), priority=10,
|
|
steps=(MetaStep(id="s1", skill="alpha", kind="agent"),),
|
|
)
|
|
orch = MetaOrchestrator(
|
|
agent_runner=lambda *a, **kw: None, skill_loader=lambda: None,
|
|
)
|
|
orch._dispatch_step_stream = _stub_runner_for({"s1": "X"}) # type: ignore[assignment]
|
|
|
|
async def empty_preface(step_id, effective_skill):
|
|
if False:
|
|
yield None
|
|
return
|
|
orch._yield_skill_view_preface = empty_preface # type: ignore[assignment]
|
|
|
|
match = MetaMatch(plan=plan, inputs={})
|
|
async for _ in orch.iter_events(match):
|
|
pass
|
|
|
|
assert writer_db.list_runs(name="legacy") == []
|