Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 24 additions & 1 deletion demos/run_demo.py
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,16 @@ def _verbose_demo_label(path: Path) -> str:
return path.stem.replace("_llm", "")


def _is_compiler_regression(result: DemoReport) -> bool:
return bool(result["baseline_pass"]) and not bool(result["compiler_pass"])


def _print_compiler_regression_warning() -> None:
print()
print("⚠️ COMPILER REGRESSION")
print("baseline succeeded but compiler-mediated version failed")


def _run(path: Path, *, verbose: bool) -> tuple[DemoReport | None, InfoReport | None]:
if verbose:
print(f"===== Running {_verbose_demo_label(path)} =====")
Expand Down Expand Up @@ -87,6 +97,7 @@ def main() -> None:
baseline_fail_count = 0
compiler_pass_count = 0
compiler_fail_count = 0
compiler_regressions = 0
informational_reports: list[InfoReport] = []
for index, key in enumerate(sorted(DEMO_FILES.keys())):
if index > 0 and not args.verbose:
Expand Down Expand Up @@ -117,12 +128,22 @@ def main() -> None:
compiler_pass_count += 1
else:
compiler_fail_count += 1

if _is_compiler_regression(result):
compiler_regressions += 1
_print_compiler_regression_warning()
print()
print("Summary:")
print()
print("Evaluative demos:")
print(f"Baseline results: {baseline_pass_count} passed, {baseline_fail_count} failed")
print(f"Compiler results: {compiler_pass_count} passed, {compiler_fail_count} failed")
if compiler_regressions > 0:
print()
if compiler_regressions == 1:
print("*** 1 COMPILER REGRESSION DETECTED ***")
else:
print(f"*** {compiler_regressions} COMPILER REGRESSIONS DETECTED ***")
if informational_reports:
print()
print("Informational demo:")
Expand All @@ -139,10 +160,12 @@ def main() -> None:
return

try:
_run(root / DEMO_FILES[args.demo], verbose=args.verbose)
result, _ = _run(root / DEMO_FILES[args.demo], verbose=args.verbose)
except MissingDemoConfigError as exc:
_print_config_error(exc)
raise SystemExit(2) from exc
if args.demo in SCORED_DEMOS and result is not None and _is_compiler_regression(result):
_print_compiler_regression_warning()


if __name__ == "__main__":
Expand Down
49 changes: 49 additions & 0 deletions tests/test_04_llm_tool_governance.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,49 @@
import importlib.util
import sys
from pathlib import Path
from types import ModuleType

REPO_ROOT = Path(__file__).resolve().parents[1]
if str(REPO_ROOT) not in sys.path:
sys.path.insert(0, str(REPO_ROOT))


def _load_demo_module(filename: str) -> ModuleType:
module_name = f"test_{filename[:-3]}"
module_path = REPO_ROOT / "demos" / filename
spec = importlib.util.spec_from_file_location(module_name, module_path)
assert spec is not None
assert spec.loader is not None
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module


def test_selected_tool_prefers_tool_tag() -> None:
module = _load_demo_module("04_llm_tool_governance.py")
output = "TOOL:kubectl\nACTION:Use kubectl apply."

assert module.selected_tool(output) == "kubectl"


def test_selected_tool_falls_back_to_action_line() -> None:
module = _load_demo_module("04_llm_tool_governance.py")
output = "- Recommended: choose docker for this deployment."

assert module.selected_tool(output) == "docker"


def test_selected_tool_uses_regex_fallback_after_unknown_tag() -> None:
module = _load_demo_module("04_llm_tool_governance.py")
output = "TOOL:helm\ntool : kubectl"

assert module.selected_tool(output) == "kubectl"


def test_selected_tool_returns_none_for_unknown_or_missing_tool() -> None:
module = _load_demo_module("04_llm_tool_governance.py")
unknown_tag = "TOOL:helm\nACTION:Use helm install."
missing_tool = "ACTION:Use terraform apply."

assert module.selected_tool(unknown_tag) is None
assert module.selected_tool(missing_tool) is None
183 changes: 183 additions & 0 deletions tests/test_llm_demos.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,183 @@
import importlib.util
import sys
from collections.abc import Iterator
from pathlib import Path
from types import ModuleType
from typing import Any

import pytest

REPO_ROOT = Path(__file__).resolve().parents[1]
if str(REPO_ROOT) not in sys.path:
sys.path.insert(0, str(REPO_ROOT))

from demos.common import consume_last_report # noqa: E402


def _load_demo_module(filename: str) -> ModuleType:
module_name = f"test_{filename[:-3]}"
module_path = REPO_ROOT / "demos" / filename
spec = importlib.util.spec_from_file_location(module_name, module_path)
assert spec is not None
assert spec.loader is not None
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module


def _run_demo_with_mocked_llm(
module: ModuleType,
monkeypatch: pytest.MonkeyPatch,
outputs: list[str],
) -> tuple[dict[str, Any], list[list[dict[str, str]]]]:
calls: list[list[dict[str, str]]] = []
output_iter: Iterator[str] = iter(outputs)

def fake_complete_messages(messages: list[dict[str, str]]) -> str:
calls.append(messages)
try:
return next(output_iter)
except StopIteration as exc:
raise AssertionError("Unexpected extra LLM call.") from exc

monkeypatch.setattr(module, "complete_messages", fake_complete_messages)
consume_last_report()
module.main()
try:
next(output_iter)
except StopIteration:
pass
else:
raise AssertionError("Expected additional LLM call(s) were not made.")
report = consume_last_report()
assert report is not None
return report, calls


def test_demo_01_constraint_drift(monkeypatch: pytest.MonkeyPatch) -> None:
module = _load_demo_module("01_llm_constraint_drift.py")
report, calls = _run_demo_with_mocked_llm(
module,
monkeypatch,
outputs=[
"Ingredients:\n- peanuts\n- coconut milk\nSteps:\n1. Cook.",
(
"I cannot comply with a peanut recipe because it is prohibited.\n"
"Ingredients:\n- chickpeas\n- coconut milk\nSteps:\n1. Cook."
),
],
)

assert report["baseline_pass"] is False
assert report["compiler_pass"] is True
assert len(calls) == 2
assert "policies.prohibit: peanuts" in calls[1][0]["content"]


def test_demo_02_correction_replacement(monkeypatch: pytest.MonkeyPatch) -> None:
module = _load_demo_module("02_llm_correction_replacement.py")
report, calls = _run_demo_with_mocked_llm(
module,
monkeypatch,
outputs=[
(
"FOCUS_PRIMARY: vegan curry\n"
"Shopping list:\n- vegan curry paste\n- tofu\n"
"Steps:\n1. Make vegan curry."
),
(
"FOCUS_PRIMARY: vegan curry\n"
"Shopping list:\n- vegan curry paste\n- tofu\n"
"Steps:\n1. Make vegan curry."
),
],
)

assert report["baseline_pass"] is True
assert report["compiler_pass"] is True
assert len(calls) == 2
assert "facts.focus.primary: vegan curry" in calls[1][0]["content"]


def test_demo_03_ambiguity_block(monkeypatch: pytest.MonkeyPatch) -> None:
module = _load_demo_module("03_llm_ambiguity_block.py")
report, calls = _run_demo_with_mocked_llm(
module,
monkeypatch,
outputs=["ACTION:proceed"],
)

# Baseline path calls the LLM once; compiler-mediated clarify path should not call it.
assert len(calls) == 1
assert report["baseline_pass"] is False
assert report["compiler_pass"] is True


def test_demo_03_non_clarify_path_calls_llm_for_mediated(monkeypatch: pytest.MonkeyPatch) -> None:
module = _load_demo_module("03_llm_ambiguity_block.py")

class _EngineStub:
def __init__(self) -> None:
self.state: dict[str, object] = {
"facts": {"focus.primary": None},
"policies": {"prohibit": []},
"version": 1,
}

def step(self, _text: str) -> dict[str, object]:
return {"kind": "passthrough", "prompt_to_user": None, "state": None}

monkeypatch.setattr(module, "create_engine", lambda: _EngineStub())
report, calls = _run_demo_with_mocked_llm(
module,
monkeypatch,
outputs=["ACTION:clarify", "ACTION:proceed"],
)

assert len(calls) == 2
assert report["baseline_pass"] is True
assert report["compiler_pass"] is False


def test_demo_04_tool_governance(monkeypatch: pytest.MonkeyPatch) -> None:
module = _load_demo_module("04_llm_tool_governance.py")
report, calls = _run_demo_with_mocked_llm(
module,
monkeypatch,
outputs=[
"TOOL:docker\nACTION:Use docker deploy.",
"TOOL:kubectl\nACTION:Use kubectl apply.",
],
)

assert report["baseline_pass"] is False
assert report["compiler_pass"] is True
assert len(calls) == 2
assert "Candidate tools: docker, kubectl." in calls[1][0]["content"]
assert "Prohibited: docker" in calls[1][0]["content"]


def test_demo_05_prompt_drift(monkeypatch: pytest.MonkeyPatch) -> None:
module = _load_demo_module("05_llm_prompt_drift.py")
report, calls = _run_demo_with_mocked_llm(
module,
monkeypatch,
outputs=[
("DINNER_STYLE:vegetarian\nDinner plan:\n- chickpea curry\n- rice\nSteps:\n1. Cook."),
("DINNER_STYLE:vegetarian\nDinner plan:\n- chickpea curry\n- rice\nSteps:\n1. Cook."),
],
)

assert report["baseline_pass"] is True
assert report["compiler_pass"] is True
assert len(calls) == 2
assert "facts.focus.primary: vegetarian curry" in calls[1][0]["content"]


def test_demo_01_negation_and_refusal_lines_do_not_count_as_violations() -> None:
module = _load_demo_module("01_llm_constraint_drift.py")
negated_recipe = "Ingredients:\n- no peanuts\n- coconut milk"
refusal_with_prohibited_token = "I cannot comply with peanuts because that is prohibited."

assert module.recipe_includes_prohibited_item(negated_recipe) is False
assert module.recipe_includes_prohibited_item(refusal_with_prohibited_token) is False
Loading
Loading