diff --git a/demos/run_demo.py b/demos/run_demo.py index 7ea9da4..633cf93 100644 --- a/demos/run_demo.py +++ b/demos/run_demo.py @@ -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)} =====") @@ -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: @@ -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:") @@ -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__": diff --git a/tests/test_04_llm_tool_governance.py b/tests/test_04_llm_tool_governance.py new file mode 100644 index 0000000..8ce1c15 --- /dev/null +++ b/tests/test_04_llm_tool_governance.py @@ -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 diff --git a/tests/test_llm_demos.py b/tests/test_llm_demos.py new file mode 100644 index 0000000..38632dd --- /dev/null +++ b/tests/test_llm_demos.py @@ -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 diff --git a/tests/test_run_demo.py b/tests/test_run_demo.py new file mode 100644 index 0000000..631416b --- /dev/null +++ b/tests/test_run_demo.py @@ -0,0 +1,257 @@ +import runpy +import sys +from pathlib import Path + +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 import run_demo # noqa: E402 +from demos.common import consume_last_info_report # noqa: E402 + + +def _demo_report(*, baseline_pass: bool, compiler_pass: bool) -> run_demo.DemoReport: + return { + "name": "01_fake — regression fixture", + "expected": "expected behavior", + "actual": "actual behavior", + "baseline_pass": baseline_pass, + "compiler_pass": compiler_pass, + "demo_pass": compiler_pass, + } + + +def _info_report() -> run_demo.InfoReport: + return { + "name": "06_context_compaction — superseded directives eliminated", + "baseline_context_length": 137, + "compiled_context_length": 37, + "context_reduction_percent": 73, + "baseline_prompt_length": 247, + "compiled_prompt_length": 160, + "prompt_reduction_percent": 35, + } + + +def test_runner_prints_per_demo_compiler_regression_warning( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + def fake_run( + path: Path, *, verbose: bool + ) -> tuple[run_demo.DemoReport | None, run_demo.InfoReport | None]: + assert not verbose + print("01_fake — regression fixture") + print("baseline: PASS") + print("compiler: FAIL") + print("expected: expected behavior") + print("actual: actual behavior") + print("result: corrected value determined the final plan") + return _demo_report(baseline_pass=True, compiler_pass=False), None + + monkeypatch.setattr(run_demo, "DEMO_FILES", {"1": "fake_01.py"}) + monkeypatch.setattr(run_demo, "SCORED_DEMOS", {"1"}) + monkeypatch.setattr(run_demo, "_run", fake_run) + monkeypatch.setattr("sys.argv", ["run_demo", "1"]) + + run_demo.main() + output = capsys.readouterr().out + + assert "result:" in output + assert "⚠️ COMPILER REGRESSION" in output + assert "baseline succeeded but compiler-mediated version failed" in output + result_index = output.index("result:") + warning_index = output.index("⚠️ COMPILER REGRESSION") + detail_index = output.index("baseline succeeded but compiler-mediated version failed") + assert result_index < warning_index < detail_index + + +def test_runner_prints_summary_regression_banner_in_all_mode( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + def fake_run( + path: Path, *, verbose: bool + ) -> tuple[run_demo.DemoReport | None, run_demo.InfoReport | None]: + assert not verbose + if path.name == "fake_01.py": + print("01_fake — regression fixture") + print("baseline: PASS") + print("compiler: FAIL") + print("expected: expected behavior") + print("actual: actual behavior") + print("result: regression observed") + return _demo_report(baseline_pass=True, compiler_pass=False), None + print("06_context_compaction — superseded directives eliminated") + print("context: 137 → 37 chars") + print("prompt: 247 → 160 chars") + print("reduction: context 73%; prompt 35%") + print("result: compiled authoritative state replaced superseded transcript directives") + return None, _info_report() + + monkeypatch.setattr(run_demo, "DEMO_FILES", {"1": "fake_01.py", "6": "fake_06.py"}) + monkeypatch.setattr(run_demo, "SCORED_DEMOS", {"1"}) + monkeypatch.setattr(run_demo, "_run", fake_run) + monkeypatch.setattr("sys.argv", ["run_demo", "all"]) + + run_demo.main() + output = capsys.readouterr().out + + assert "Baseline results: 1 passed, 0 failed" in output + assert "Compiler results: 0 passed, 1 failed" in output + assert "*** 1 COMPILER REGRESSION DETECTED ***" in output + + +def test_runner_prints_plural_summary_regression_banner_in_all_mode( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + def fake_run( + path: Path, *, verbose: bool + ) -> tuple[run_demo.DemoReport | None, run_demo.InfoReport | None]: + assert not verbose + if path.name in {"fake_01.py", "fake_02.py"}: + print(f"{path.stem} — regression fixture") + print("baseline: PASS") + print("compiler: FAIL") + print("expected: expected behavior") + print("actual: actual behavior") + print("result: regression observed") + return _demo_report(baseline_pass=True, compiler_pass=False), None + print("06_context_compaction — superseded directives eliminated") + print("context: 137 → 37 chars") + print("prompt: 247 → 160 chars") + print("reduction: context 73%; prompt 35%") + print("result: compiled authoritative state replaced superseded transcript directives") + return None, _info_report() + + monkeypatch.setattr( + run_demo, + "DEMO_FILES", + {"1": "fake_01.py", "2": "fake_02.py", "6": "fake_06.py"}, + ) + monkeypatch.setattr(run_demo, "SCORED_DEMOS", {"1", "2"}) + monkeypatch.setattr(run_demo, "_run", fake_run) + monkeypatch.setattr("sys.argv", ["run_demo", "all"]) + + run_demo.main() + output = capsys.readouterr().out + + assert "Baseline results: 2 passed, 0 failed" in output + assert "Compiler results: 0 passed, 2 failed" in output + assert "*** 2 COMPILER REGRESSIONS DETECTED ***" in output + + +def test_informational_demo_is_non_scored_in_all_mode( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + def fake_run( + path: Path, *, verbose: bool + ) -> tuple[run_demo.DemoReport | None, run_demo.InfoReport | None]: + assert not verbose + if path.name == "fake_01.py": + print("01_fake — pass fixture") + print("baseline: PASS") + print("compiler: PASS") + print("expected: expected behavior") + print("actual: actual behavior") + print("result: success") + return _demo_report(baseline_pass=True, compiler_pass=True), None + print("06_context_compaction — superseded directives eliminated") + print("context: 137 → 37 chars") + print("prompt: 247 → 160 chars") + print("reduction: context 73%; prompt 35%") + print("result: compiled authoritative state replaced superseded transcript directives") + return None, _info_report() + + monkeypatch.setattr(run_demo, "DEMO_FILES", {"1": "fake_01.py", "6": "fake_06.py"}) + monkeypatch.setattr(run_demo, "SCORED_DEMOS", {"1"}) + monkeypatch.setattr(run_demo, "_run", fake_run) + monkeypatch.setattr("sys.argv", ["run_demo", "all"]) + + run_demo.main() + output = capsys.readouterr().out + + assert "Baseline results: 1 passed, 0 failed" in output + assert "Compiler results: 1 passed, 0 failed" in output + assert ( + "06_context_compaction — context 137 → 37 chars (73% reduction); " + "prompt 247 → 160 chars (35% reduction)" + ) in output + assert "*** 1 COMPILER REGRESSION DETECTED ***" not in output + + +def test_all_mode_scored_none_result_counts_as_failures( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + def fake_run( + path: Path, *, verbose: bool + ) -> tuple[run_demo.DemoReport | None, run_demo.InfoReport | None]: + assert not verbose + if path.name == "fake_01.py": + return None, None + print("06_context_compaction — superseded directives eliminated") + print("context: 137 → 37 chars") + print("prompt: 247 → 160 chars") + print("reduction: context 73%; prompt 35%") + print("result: compiled authoritative state replaced superseded transcript directives") + return None, _info_report() + + monkeypatch.setattr(run_demo, "DEMO_FILES", {"1": "fake_01.py", "6": "fake_06.py"}) + monkeypatch.setattr(run_demo, "SCORED_DEMOS", {"1"}) + monkeypatch.setattr(run_demo, "_run", fake_run) + monkeypatch.setattr("sys.argv", ["run_demo", "all"]) + + run_demo.main() + output = capsys.readouterr().out + + assert "Baseline results: 0 passed, 1 failed" in output + assert "Compiler results: 0 passed, 1 failed" in output + + +def test_all_mode_counts_baseline_fail_and_compiler_pass( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + def fake_run( + path: Path, *, verbose: bool + ) -> tuple[run_demo.DemoReport | None, run_demo.InfoReport | None]: + assert not verbose + if path.name == "fake_01.py": + print("01_fake — mixed fixture") + print("baseline: FAIL") + print("compiler: PASS") + print("expected: expected behavior") + print("actual: actual behavior") + print("result: mixed outcome") + return _demo_report(baseline_pass=False, compiler_pass=True), None + print("06_context_compaction — superseded directives eliminated") + print("context: 137 → 37 chars") + print("prompt: 247 → 160 chars") + print("reduction: context 73%; prompt 35%") + print("result: compiled authoritative state replaced superseded transcript directives") + return None, _info_report() + + monkeypatch.setattr(run_demo, "DEMO_FILES", {"1": "fake_01.py", "6": "fake_06.py"}) + monkeypatch.setattr(run_demo, "SCORED_DEMOS", {"1"}) + monkeypatch.setattr(run_demo, "_run", fake_run) + monkeypatch.setattr("sys.argv", ["run_demo", "all"]) + + run_demo.main() + output = capsys.readouterr().out + + assert "Baseline results: 0 passed, 1 failed" in output + assert "Compiler results: 1 passed, 0 failed" in output + + +def test_compaction_demo_reports_sane_metrics() -> None: + consume_last_info_report() + + demo_path = Path(__file__).resolve().parents[1] / "demos" / "06_context_compaction.py" + runpy.run_path(str(demo_path), run_name="__main__") + + report = consume_last_info_report() + assert report is not None + assert report["name"].startswith("06_context_compaction") + assert report["baseline_context_length"] > report["compiled_context_length"] + assert report["baseline_prompt_length"] > report["compiled_prompt_length"] + assert report["context_reduction_percent"] > 0 + assert report["prompt_reduction_percent"] > 0