wehub-resource-sync
256 lines
8.3 KiB
Python
256 lines
8.3 KiB
Python
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from opensquilla.engine.types import AgentConfig, DoneEvent
|
|
from opensquilla.skills.creator.runtime_e2e import (
|
|
make_runtime_e2e_context,
|
|
run_runtime_e2e_gate,
|
|
)
|
|
from opensquilla.tool_boundary import ToolCall
|
|
|
|
SKILL_MD = """---
|
|
name: synth-test-pipeline
|
|
description: "Sample synthetic pipeline for runtime E2E tests"
|
|
kind: meta
|
|
meta_priority: 50
|
|
triggers:
|
|
- "synth test trigger"
|
|
provenance:
|
|
origin: opensquilla-user
|
|
composition:
|
|
steps:
|
|
- id: a
|
|
skill: summarize
|
|
with:
|
|
task: "{{ inputs.user_message | xml_escape | truncate(512) }}"
|
|
---
|
|
"""
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_runtime_e2e_gate_runs_meta_and_no_meta_baseline() -> None:
|
|
calls: list[tuple[str, str, str]] = []
|
|
|
|
async def runner(*, route: str, prompt: str, skill_md: str, baseline_model: str) -> dict:
|
|
calls.append((route, prompt, baseline_model))
|
|
return {
|
|
"text": (
|
|
"meta answer with concrete summary"
|
|
if route == "meta"
|
|
else "baseline generic answer"
|
|
),
|
|
"model": baseline_model if route == "baseline" else "meta-route",
|
|
}
|
|
|
|
async def judge(*, prompt: str, meta: dict, baseline: dict) -> dict:
|
|
assert "synth test trigger" in prompt
|
|
assert meta["text"].startswith("meta answer")
|
|
assert baseline["text"].startswith("baseline")
|
|
return {"winner": "meta", "regression": "", "reason": "meta follows the trigger"}
|
|
|
|
result = await run_runtime_e2e_gate(
|
|
skill_md=SKILL_MD,
|
|
eval_prompts=["please use synth test trigger"],
|
|
baseline_model="frontier/highest",
|
|
runner=runner,
|
|
judge=judge,
|
|
)
|
|
|
|
assert result["status"] == "ok"
|
|
assert result["passed"] is True
|
|
assert result["winner"] == "meta"
|
|
assert calls == [
|
|
("meta", "please use synth test trigger", "frontier/highest"),
|
|
("baseline", "please use synth test trigger", "frontier/highest"),
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_runtime_e2e_gate_blocks_baseline_winner() -> None:
|
|
async def runner(*, route: str, prompt: str, skill_md: str, baseline_model: str) -> dict:
|
|
return {"text": f"{route} output", "model": baseline_model}
|
|
|
|
async def judge(*, prompt: str, meta: dict, baseline: dict) -> dict:
|
|
return {
|
|
"winner": "baseline",
|
|
"regression": "meta omits the requested evidence",
|
|
"reason": "baseline is more complete",
|
|
}
|
|
|
|
result = await run_runtime_e2e_gate(
|
|
skill_md=SKILL_MD,
|
|
eval_prompts=["please use synth test trigger"],
|
|
baseline_model="frontier/highest",
|
|
runner=runner,
|
|
judge=judge,
|
|
)
|
|
|
|
assert result["passed"] is False
|
|
assert result["winner"] == "baseline"
|
|
assert result["cases"][0]["regression"] == "meta omits the requested evidence"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_runtime_e2e_gate_blocks_invalid_baseline_refusal() -> None:
|
|
async def runner(*, route: str, prompt: str, skill_md: str, baseline_model: str) -> dict:
|
|
if route == "baseline":
|
|
return {
|
|
"text": (
|
|
"Runtime E2E baseline mode: meta-skill creator tools are "
|
|
"disabled, so I cannot complete this request."
|
|
),
|
|
"model": baseline_model,
|
|
}
|
|
return {"text": "meta output", "model": "meta"}
|
|
|
|
async def judge(*, prompt: str, meta: dict, baseline: dict) -> dict:
|
|
raise AssertionError("blocked/refusal baseline should not be sent to judge")
|
|
|
|
result = await run_runtime_e2e_gate(
|
|
skill_md=SKILL_MD,
|
|
eval_prompts=["create a useful meta-skill from this workflow"],
|
|
baseline_model="frontier/highest",
|
|
runner=runner,
|
|
judge=judge,
|
|
)
|
|
|
|
assert result["passed"] is False
|
|
assert result["winner"] == "invalid"
|
|
assert result["cases"][0]["regression"] == "baseline_invalid_or_blocked"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_runtime_e2e_context_baseline_runs_without_meta_loader() -> None:
|
|
seen_configs: list[AgentConfig] = []
|
|
|
|
class FakeAgent:
|
|
def __init__(self, **kwargs) -> None:
|
|
seen_configs.append(kwargs["config"])
|
|
|
|
async def run_turn(self, prompt: str):
|
|
yield DoneEvent(text=f"baseline handled {prompt}")
|
|
|
|
ctx = make_runtime_e2e_context(
|
|
provider=object(),
|
|
base_config=AgentConfig(
|
|
model_id="frontier/highest",
|
|
metadata={"skill_loader": object(), "meta_match": object(), "keep": "yes"},
|
|
),
|
|
skill_loader=object(),
|
|
tool_definitions=[],
|
|
tool_handler=None,
|
|
agent_factory=FakeAgent,
|
|
llm_chat=None,
|
|
tool_invoker=None,
|
|
session_key="test",
|
|
baseline_model="frontier/highest",
|
|
)
|
|
|
|
result = await ctx["runner"](
|
|
route="baseline",
|
|
prompt="compare this",
|
|
skill_md=SKILL_MD,
|
|
baseline_model="frontier/highest",
|
|
)
|
|
|
|
assert result["text"] == "baseline handled compare this"
|
|
assert seen_configs[0].metadata == {"keep": "yes"}
|
|
assert seen_configs[0].model_id == "frontier/highest"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_runtime_e2e_context_baseline_blocks_creator_side_effect_tools() -> None:
|
|
observed: list[tuple[str, bool, str]] = []
|
|
|
|
async def unsafe_tool_handler(tc: ToolCall):
|
|
raise AssertionError(f"baseline leaked creator tool call: {tc.tool_name}")
|
|
|
|
class FakeAgent:
|
|
def __init__(self, **kwargs) -> None:
|
|
self.tool_handler = kwargs["tool_handler"]
|
|
|
|
async def run_turn(self, prompt: str):
|
|
result = await self.tool_handler(ToolCall(
|
|
tool_use_id="tool-1",
|
|
tool_name="meta_skill_persist_proposal",
|
|
arguments={},
|
|
))
|
|
observed.append((result.tool_name, result.is_error, result.content))
|
|
yield DoneEvent(text="baseline done")
|
|
|
|
ctx = make_runtime_e2e_context(
|
|
provider=object(),
|
|
base_config=AgentConfig(model_id="frontier/highest"),
|
|
skill_loader=object(),
|
|
tool_definitions=[],
|
|
tool_handler=unsafe_tool_handler,
|
|
agent_factory=FakeAgent,
|
|
llm_chat=None,
|
|
tool_invoker=None,
|
|
session_key="test",
|
|
baseline_model="frontier/highest",
|
|
)
|
|
|
|
result = await ctx["runner"](
|
|
route="baseline",
|
|
prompt="compare this",
|
|
skill_md=SKILL_MD,
|
|
baseline_model="frontier/highest",
|
|
)
|
|
|
|
assert result["text"] == "baseline done"
|
|
assert observed == [(
|
|
"meta_skill_persist_proposal",
|
|
False,
|
|
"Continue without this tool and write the strongest standalone answer "
|
|
"directly in the final response.",
|
|
)]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_runtime_e2e_context_baseline_hides_meta_tools_and_instructs_direct_answer() -> None:
|
|
captured: dict[str, object] = {}
|
|
|
|
class FakeAgent:
|
|
def __init__(self, **kwargs) -> None:
|
|
captured["config"] = kwargs["config"]
|
|
captured["tool_definitions"] = kwargs["tool_definitions"]
|
|
|
|
async def run_turn(self, prompt: str):
|
|
yield DoneEvent(text="baseline direct answer")
|
|
|
|
ctx = make_runtime_e2e_context(
|
|
provider=object(),
|
|
base_config=AgentConfig(model_id="frontier/highest"),
|
|
skill_loader=object(),
|
|
tool_definitions=[
|
|
{"type": "function", "function": {"name": "meta_invoke"}},
|
|
{"type": "function", "function": {"name": "meta_skill_persist_proposal"}},
|
|
{"type": "function", "function": {"name": "memory_search"}},
|
|
],
|
|
tool_handler=None,
|
|
agent_factory=FakeAgent,
|
|
llm_chat=None,
|
|
tool_invoker=None,
|
|
session_key="test",
|
|
baseline_model="frontier/highest",
|
|
)
|
|
|
|
result = await ctx["runner"](
|
|
route="baseline",
|
|
prompt="create a meta-skill from my history",
|
|
skill_md=SKILL_MD,
|
|
baseline_model="frontier/highest",
|
|
)
|
|
|
|
assert result["text"] == "baseline direct answer"
|
|
assert captured["tool_definitions"] == [
|
|
{"type": "function", "function": {"name": "memory_search"}},
|
|
]
|
|
config = captured["config"]
|
|
assert isinstance(config, AgentConfig)
|
|
assert "highest-tier single model" in (config.request_context_prompt or "")
|
|
assert "standalone proposal" in (config.request_context_prompt or "")
|
|
assert "disabled" not in (config.request_context_prompt or "").lower()
|