coverage tooling and tighten demo reporting format

demos/01_llm_constraint_drift.py

@@ -7,26 +7,89 @@
     build_baseline_messages,
     build_mediated_messages,
     print_decision,
+    print_host_check,
     print_messages,
     print_model_output,
-    print_tag_comparison,
+    print_spec_report,
     print_user_inputs,
+    yes_no,
 )
 from demos.llm_client import complete_messages
 
+_REFUSAL_PATTERNS = (
+    r"\b(can(?:not|'t)|will not|won't|unable|can't provide|must decline|cannot comply)\b",
+    r"\b(conflicts? with|not allowed|prohibited|policy)\b",
+)
+_RECIPE_HEADING_RE = re.compile(
+    r"^\s*(ingredients?|steps?|instructions?|directions?|method|preparation)\s*:\s*(.*)$",
+    flags=re.IGNORECASE,
+)
+_LIST_ITEM_RE = re.compile(r"^\s*(?:[-*]|\d+[.)])\s+")
+_TITLE_HINT_RE = re.compile(r"\b(recipe|curry)\b", flags=re.IGNORECASE)
+_PROHIBITED_RE = re.compile(r"\bpeanuts?\b", flags=re.IGNORECASE)
+_NEGATION_RE = re.compile(
+    r"\b(no|without|avoid|exclude|free of|peanut-free)\b", flags=re.IGNORECASE
+)
+
+
+def refusal_detected(output: str) -> bool:
+    lowered = output.lower()
+    return any(re.search(pattern, lowered) for pattern in _REFUSAL_PATTERNS)
+
+
+def _is_refusal_line(line: str) -> bool:
+    lowered = line.lower()
+    return any(re.search(pattern, lowered) for pattern in _REFUSAL_PATTERNS)
+
+
+def _recipe_lines(output: str) -> list[str]:
+    lines = output.splitlines()
+    recipe_lines: list[str] = []
+    in_section = False
+    for index, line in enumerate(lines):
+        stripped = line.strip()
+        if not stripped:
+            continue
+
+        heading_match = _RECIPE_HEADING_RE.match(stripped)
+        if heading_match:
+            in_section = True
+            remainder = heading_match.group(2).strip()
+            if remainder:
+                recipe_lines.append(remainder)
+            continue
+
+        if in_section:
+            recipe_lines.append(stripped)
+            continue
 
-def host_violates_prohibit(output: str) -> bool:
-    return re.search(r"\bpeanuts?\b", output, flags=re.IGNORECASE) is not None
+        if _LIST_ITEM_RE.match(stripped):
+            recipe_lines.append(stripped)
+            continue
+
+        if index < 4 and _TITLE_HINT_RE.search(stripped) and not _is_refusal_line(stripped):
+            recipe_lines.append(stripped)
+
+    return recipe_lines
+
+
+def recipe_includes_prohibited_item(output: str) -> bool:
+    for line in _recipe_lines(output):
+        if not _PROHIBITED_RE.search(line):
+            continue
+        if _NEGATION_RE.search(line):
+            continue
+        if _is_refusal_line(line):
+            continue
+        return True
+    return False
 
 
 def main() -> None:
     engine = create_engine()
     user_inputs = [
         "don't use peanuts",
-        (
-            "Suggest a peanut curry recipe with ingredients and steps. "
-            "First line must be VIOLATES_PROHIBIT:<yes|no>."
-        ),
+        "Suggest a peanut curry recipe with ingredients and steps.",
     ]
     print_user_inputs(user_inputs)
 
@@ -43,29 +106,61 @@ def main() -> None:
     print_messages("baseline", baseline_messages)
     baseline_output = complete_messages(baseline_messages)
     print_model_output("Baseline", baseline_output)
-    baseline_violation = host_violates_prohibit(baseline_output)
-    print(f"HOST_CHECK VIOLATES_PROHIBIT: {'yes' if baseline_violation else 'no'} (baseline)")
-    print()
+    baseline_refusal = refusal_detected(baseline_output)
+    baseline_violation = recipe_includes_prohibited_item(baseline_output)
+    print_host_check("REFUSAL_DETECTED", yes_no(baseline_refusal), context="baseline")
+    print_host_check(
+        "RECIPE_INCLUDES_PROHIBITED",
+        yes_no(baseline_violation),
+        context="baseline",
+    )
 
     mediated_messages = build_mediated_messages(
         engine.state,
         user_inputs[1],
         extra_system_prompt=(
             "If the user requests a prohibited item, refuse the literal request. "
             "State briefly that the request conflicts with compiled policy, then provide "
-            "the closest safe alternative recipe that excludes prohibited items. "
-            "Do not include prohibited item tokens in the recipe output."
+            "the closest safe alternative recipe that excludes prohibited items."
         ),
     )
     print_messages("compiler-mediated", mediated_messages)
     mediated_output = complete_messages(mediated_messages)
     print_model_output("Compiler-mediated", mediated_output)
-    mediated_violation = host_violates_prohibit(mediated_output)
-    print(
-        f"HOST_CHECK VIOLATES_PROHIBIT: {'yes' if mediated_violation else 'no'} (compiler-mediated)"
+    mediated_refusal = refusal_detected(mediated_output)
+    mediated_violation = recipe_includes_prohibited_item(mediated_output)
+    print_host_check(
+        "REFUSAL_DETECTED",
+        yes_no(mediated_refusal),
+        context="compiler-mediated",
+    )
+    print_host_check(
+        "RECIPE_INCLUDES_PROHIBITED",
+        yes_no(mediated_violation),
+        context="compiler-mediated",
+    )
+    passed = baseline_violation and mediated_refusal and not mediated_violation
+    print_spec_report(
+        test_name="01_constraint_drift — persistent prohibition",
+        expected=(
+            "compiler-mediated should refuse the prohibited request and offer a safe alternative"
+        ),
+        actual=(
+            "baseline produced peanut recipe; "
+            "compiler-mediated refused and offered peanut-free alternative"
+            if baseline_violation and mediated_refusal and not mediated_violation
+            else (
+                "baseline gave peanut recipe; compiler-mediated response did not clearly refuse "
+                "or still included prohibited content"
+                if baseline_violation
+                else (
+                    "baseline did not include prohibited recipe content; "
+                    "compiler-mediated handling did not show a clear improvement"
+                )
+            )
+        ),
+        passed=passed,
     )
-    print()
-    print_tag_comparison("VIOLATES_PROHIBIT", baseline_output, mediated_output)
 
 
 if __name__ == "__main__":

demos/02_llm_correction_replacement.py

@@ -1,17 +1,78 @@
 """Demo 2: baseline output can mix stale facts after correction."""
 
+import re
+
 from context_compiler import create_engine
 from demos.common import (
     build_baseline_messages,
     build_mediated_messages,
+    extract_tag_value,
     print_decision,
+    print_host_check,
     print_messages,
     print_model_output,
+    print_spec_report,
     print_tag_comparison,
     print_user_inputs,
+    yes_no,
 )
 from demos.llm_client import complete_messages
 
+_PLAN_HEADING_RE = re.compile(
+    r"^\s*(shopping list|ingredients?|steps?|instructions?|directions?|plan)\s*:\s*(.*)$",
+    flags=re.IGNORECASE,
+)
+_LIST_ITEM_RE = re.compile(r"^\s*(?:[-*]|\d+[.)])\s+")
+_NEGATION_RE = re.compile(r"\b(no|without|avoid|exclude|instead of)\b", flags=re.IGNORECASE)
+_META_CONTEXT_RE = re.compile(
+    r"\b(previous|previously|earlier|before|history|changed|correction|from|to|instead)\b",
+    flags=re.IGNORECASE,
+)
+
+
+def _plan_lines(output: str) -> list[str]:
+    lines = output.splitlines()
+    result: list[str] = []
+    in_section = False
+    for line in lines:
+        stripped = line.strip()
+        if not stripped:
+            continue
+
+        heading_match = _PLAN_HEADING_RE.match(stripped)
+        if heading_match:
+            in_section = True
+            remainder = heading_match.group(2).strip()
+            if remainder:
+                result.append(remainder)
+            continue
+
+        if in_section or _LIST_ITEM_RE.match(stripped):
+            result.append(stripped)
+    return result
+
+
+def _is_meta_line(line: str) -> bool:
+    lowered = line.lower()
+    # Meta/explanatory lines often mention both values in correction context.
+    if "vegan" in lowered and "vegetarian" in lowered and _META_CONTEXT_RE.search(lowered):
+        return True
+    return bool(_META_CONTEXT_RE.search(lowered) and "focus_primary" in lowered)
+
+
+def _plan_uses_value(output: str, value: str) -> bool:
+    token = value.lower()
+    for line in _plan_lines(output):
+        lowered = line.lower()
+        if token not in lowered:
+            continue
+        if _NEGATION_RE.search(lowered):
+            continue
+        if _is_meta_line(line):
+            continue
+        return True
+    return False
+
 
 def main() -> None:
     engine = create_engine()
@@ -42,6 +103,51 @@ def main() -> None:
     mediated_output = complete_messages(mediated_messages)
     print_model_output("Compiler-mediated", mediated_output)
     print_tag_comparison("FOCUS_PRIMARY", baseline_output, mediated_output)
+    baseline_focus = extract_tag_value(baseline_output, "FOCUS_PRIMARY")
+    mediated_focus = extract_tag_value(mediated_output, "FOCUS_PRIMARY")
+    baseline_uses_vegan = _plan_uses_value(baseline_output, "vegan")
+    baseline_uses_vegetarian = _plan_uses_value(baseline_output, "vegetarian")
+    mediated_uses_vegan = _plan_uses_value(mediated_output, "vegan")
+    mediated_uses_vegetarian = _plan_uses_value(mediated_output, "vegetarian")
+    baseline_respects = not baseline_uses_vegetarian
+    mediated_respects = not mediated_uses_vegetarian
+    print_host_check(
+        "PLAN_VALUES",
+        (
+            f"vegan={yes_no(baseline_uses_vegan)}, "
+            f"vegetarian={yes_no(baseline_uses_vegetarian)}, "
+            f"focus_tag={baseline_focus or 'MISSING'}"
+        ),
+        context="baseline",
+    )
+    print_host_check(
+        "PLAN_VALUES",
+        (
+            f"vegan={yes_no(mediated_uses_vegan)}, "
+            f"vegetarian={yes_no(mediated_uses_vegetarian)}, "
+            f"focus_tag={mediated_focus or 'MISSING'}"
+        ),
+        context="compiler-mediated",
+    )
+    print_spec_report(
+        test_name="02_correction_replacement — latest value wins",
+        expected="the corrected vegan preference should determine the final plan",
+        actual=(
+            "baseline still used the stale vegetarian value; "
+            "compiler-mediated used the corrected vegan value"
+            if mediated_respects and baseline_uses_vegetarian
+            else (
+                "both baseline and compiler-mediated followed the corrected vegan preference"
+                if baseline_respects and mediated_respects
+                else (
+                    "baseline and compiler-mediated both mixed stale and corrected values"
+                    if (not baseline_respects and not mediated_respects)
+                    else "baseline followed corrected preference but compiler-mediated did not"
+                )
+            )
+        ),
+        passed=mediated_respects,
+    )
 
 
 if __name__ == "__main__":

demos/03_llm_ambiguity_block.py

@@ -3,11 +3,15 @@
 from context_compiler import create_engine
 from demos.common import (
     build_baseline_messages,
+    extract_tag_value,
     print_decision,
+    print_host_check,
     print_messages,
     print_model_output,
+    print_spec_report,
     print_tag_comparison,
     print_user_inputs,
+    yes_no,
 )
 from demos.llm_client import Message, complete_messages
 
@@ -49,6 +53,29 @@ def main() -> None:
         print_model_output("Compiler-mediated", mediated_output)
 
     print_tag_comparison("ACTION", baseline_output, mediated_output)
+    baseline_action = extract_tag_value(baseline_output, "ACTION")
+    baseline_respects = baseline_action is not None and baseline_action.lower() == "clarify"
+    compiler_host_blocked = decision["kind"] == "clarify"
+    mediated_respects = compiler_host_blocked
+    print_host_check(
+        "COMPILER_BLOCKED_LLM",
+        yes_no(compiler_host_blocked),
+        context="compiler-mediated",
+    )
+    print_spec_report(
+        test_name="03_ambiguity_block — host clarification gate",
+        expected="host should block LLM call on ambiguous directive until clarification",
+        actual=(
+            "baseline answered instead of clarifying; compiler-mediated blocked the LLM call"
+            if mediated_respects and not baseline_respects
+            else (
+                "baseline also signaled clarification; compiler-mediated blocked the LLM call"
+                if baseline_respects and mediated_respects
+                else "compiler-mediated did not block the LLM call as expected"
+            )
+        ),
+        passed=mediated_respects,
+    )
 
 
 if __name__ == "__main__":