feat(generation): grounded scenario generation with deterministic checks and live voice wiring

oss/simulation-acceptance/run_voice_case.py

@@ -2,7 +2,9 @@
 
 import argparse
 import asyncio
+import contextlib
 import json
+import os
 import subprocess
 import sys
 from pathlib import Path
@@ -19,7 +21,32 @@ def main() -> int:
     parser.add_argument("case_id", choices=sorted(CASES))
     parser.add_argument("--output-root", default="artifacts/simulation-acceptance")
     parser.add_argument("--dry-run", action="store_true")
+    # A generated scenario brings its own caller, its own mocked tools and its own checks.
+    # Supplying one turns this into a graded test rather than a transport check.
+    parser.add_argument(
+        "--scenario",
+        default=os.environ.get("ALK_SCENARIO", ""),
+        help="path to a generated scenario; drives the caller, the mocks and the checks",
+    )
+    parser.add_argument(
+        "--agent",
+        default=os.environ.get("ALK_AGENT", ""),
+        help="registered agent whose assistant serves the scenario's tools",
+    )
+    parser.add_argument(
+        "--no-mock-tools",
+        action="store_true",
+        help="do not serve the scenario's tools; the agent's own tools answer instead",
+    )
+    parser.add_argument(
+        "--no-grade", action="store_true", help="skip the scenario's checkpoints"
+    )
+    parser.add_argument(
+        "--no-trace", action="store_true", help="skip writing the run trace"
+    )
     args = parser.parse_args()
+    if args.scenario:
+        os.environ["ALK_SCENARIO"] = args.scenario
 
     case = CASES[args.case_id]
     missing = missing_env(case)
@@ -81,33 +108,86 @@ def main() -> int:
         )
         return 0
 
+    record = None
+    if args.scenario:
+        with open(args.scenario, encoding="utf-8") as fh:
+            record = json.load(fh)
+
     trigger = _start_livekit_outbound_trigger(case.case_id)
+    tools = contextlib.nullcontext(None)
+    if record is not None and not args.no_mock_tools:
+        from fi.alk.generation.live_run import tool_session
+
+        tools = tool_session(record, agent=args.agent)
     try:
-        report = asyncio.run(
-            simulate.run_voice_simulation(
-                agent_definition=inputs.agent_definition,
-                livekit_runtime=inputs.livekit_runtime,
-                scenario=inputs.scenario,
-                simulator=inputs.simulator,
-                simulation_run_id=run_id,
-                record_audio=True,
-                recording_root=output_dir / "recordings",
-                recording_case_directory=output_dir / "recordings",
-                min_turn_messages=6,
-                max_seconds=inputs.max_seconds,
-                connect_timeout=60,
-                readiness_timeout=120,
-                cleanup_timeout=30,
-                conversation_direction=inputs.conversation_direction,
-                agent_first_silence_timeout_seconds=30,
+        with tools as tool_state:
+            report = asyncio.run(
+                simulate.run_voice_simulation(
+                    agent_definition=inputs.agent_definition,
+                    livekit_runtime=inputs.livekit_runtime,
+                    scenario=inputs.scenario,
+                    simulator=inputs.simulator,
+                    simulation_run_id=run_id,
+                    record_audio=True,
+                    recording_root=output_dir / "recordings",
+                    recording_case_directory=output_dir / "recordings",
+                    min_turn_messages=6,
+                    max_seconds=inputs.max_seconds,
+                    connect_timeout=60,
+                    readiness_timeout=120,
+                    cleanup_timeout=30,
+                    conversation_direction=inputs.conversation_direction,
+                    agent_first_silence_timeout_seconds=30,
+                )
             )
-        )
-        evaluation = evaluate_agent_report(report, attach=True)
+            evaluation = evaluate_agent_report(report, attach=True)
+            recorded_calls = tool_state.calls() if tool_state is not None else []
+            final_state = tool_state.final_state if tool_state is not None else {}
     finally:
         _finish_livekit_outbound_trigger(trigger)
     report_path = output_dir / "report.json"
     report_path.write_text(report.model_dump_json(indent=2), encoding="utf-8")
     result = report.results[0]
+
+    grading = None
+    trace_path = None
+    if record is not None:
+        messages = [
+            m if isinstance(m, dict) else m.model_dump()
+            for m in (result.messages or [])
+        ]
+        # Provider evidence keeps only a count, so the mock server's record is the tool truth.
+        calls = recorded_calls or _provider_tool_calls(result)
+        if not args.no_grade:
+            from fi.alk.generation.live_run import grade
+
+            grading = grade(
+                record,
+                messages=messages,
+                tool_calls=calls,
+                final_state=final_state,
+            )
+            (output_dir / "checks.json").write_text(
+                json.dumps(grading, indent=2), encoding="utf-8"
+            )
+        if not args.no_trace:
+            from fi.alk.generation.live_run import write_trace
+
+            trace_path = write_trace(
+                str(output_dir),
+                record=record,
+                messages=messages,
+                tool_calls=calls,
+                final_state=final_state,
+                grading=grading,
+                metadata={
+                    "case_id": case.case_id,
+                    "run_id": run_id,
+                    "stop_reason": result.metadata.get("stop_reason"),
+                    "status": result.metadata.get("status"),
+                    "message_count": len(messages),
+                },
+            )
     status = str(result.metadata.get("status") or "unknown")
     print(
         json.dumps(
@@ -119,18 +199,37 @@ def main() -> int:
                 "failure": result.metadata.get("failure"),
                 "evaluation_passed": evaluation.passed,
                 "evaluation_score": evaluation.score,
+                **(
+                    {
+                        "scenario": grading.get("scenario_id"),
+                        "checks_passed": grading.get("passed"),
+                        "checks_failed": grading.get("failed"),
+                        "checks_skipped": grading.get("skipped"),
+                        "scenario_verdict": grading.get("verdict"),
+                    }
+                    if grading
+                    else {}
+                ),
+                **({"trace": trace_path} if trace_path else {}),
                 "manifest": str(manifest_path),
                 "report": str(report_path),
             },
             indent=2,
         )
     )
+    if grading and grading.get("verdict") == "fail":
+        return 1
     return _result_exit_code(
         status=status,
         evaluation_passed=evaluation.passed,
     )
 
 
+def _provider_tool_calls(result) -> list[dict]:
+    raw = getattr(result, "tool_calls", None) or []
+    return [r if isinstance(r, dict) else r.model_dump() for r in raw]
+
+
 def _result_exit_code(*, status: str, evaluation_passed: bool) -> int:
     return 0 if status == "completed" and evaluation_passed else 1
 

oss/simulation-acceptance/voice_cases.py

@@ -10,11 +10,11 @@
 _GOOGLE_PROVIDERS = {"gemini", "google", "vertex"}
 _MODEL_DEFAULTS = {
     "llm": {
-        "gemini": "gemini-2.5-flash-lite",
-        "google": "gemini-2.5-flash-lite",
+        "gemini": "gemini-2.5-flash",
+        "google": "gemini-2.5-flash",
         "openai": "gpt-4o",
         "openai_compatible": "gpt-4o",
-        "vertex": "gemini-2.5-flash-lite",
+        "vertex": "gemini-2.5-flash",
     },
     "stt": {
         "cartesia": "ink-2",
@@ -206,19 +206,35 @@ def build_inputs(case_id: str, run_id: str) -> VoiceInputs:
         room_mode="managed",
         room_name_verbatim=bool(room_override),
     )
-    scenario = simulate.Scenario(
-        name=f"acceptance-{case_id}",
-        dataset=[
-            simulate.Persona(
-                persona={"name": "Morgan", "role": "customer"},
-                situation=(
-                    "A delivery is late. Ask for its current status, expected arrival, "
-                    "and the next action."
-                ),
-                outcome="Complete a natural multi-turn conversation and close politely.",
-            )
-        ],
-    )
+    # A generated scenario, when one is supplied, replaces the built-in persona. The scenario
+    # carries the caller's situation, objective and disclosure rules, which is what turns a
+    # generic conversation into a specific test.
+    generated_path = os.environ.get("ALK_SCENARIO", "").strip()
+    if generated_path:
+        import json as _json
+
+        from fi.alk.generation.simulate_bridge import persona_from_record
+
+        with open(generated_path, encoding="utf-8") as fh:
+            record = _json.load(fh)
+        scenario = simulate.Scenario(
+            name=str(record.get("id") or f"acceptance-{case_id}"),
+            dataset=[persona_from_record(record)],
+        )
+    else:
+        scenario = simulate.Scenario(
+            name=f"acceptance-{case_id}",
+            dataset=[
+                simulate.Persona(
+                    persona={"name": "Morgan", "role": "customer"},
+                    situation=(
+                        "A delivery is late. Ask for its current status, expected arrival, "
+                        "and the next action."
+                    ),
+                    outcome="Complete a natural multi-turn conversation and close politely.",
+                )
+            ],
+        )
     llm_provider = os.environ.get("SIMULATOR_LLM_PROVIDER", "google")
     stt_provider = os.environ.get("SIMULATOR_STT_PROVIDER", "deepgram")
     tts_provider = os.environ.get("SIMULATOR_TTS_PROVIDER", "deepgram")

scripts/audit_generated_scenarios.py

@@ -0,0 +1,113 @@
+"""Audit a generation run against the agent's real source. Deterministic, no model calls.
+
+Usage: python scripts/audit_generated_scenarios.py <run_dir> <agent_repo>
+
+Checks, per scenario and in aggregate:
+- every tool named in a checkpoint or mock exists in the agent source;
+- every identifier-shaped argument value in a tool_call_args checkpoint appears in the agent source
+  (menu ids, enum values), so no checkpoint asserts an id that does not exist;
+- checkpoint kind mix and deterministic share;
+- sub-goal reuse across scenarios (the roll-up property);
+- input/checkpoint separation smells: the agent_input leaking seeded ids that facts do not cover.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import re
+import sys
+from collections import Counter
+
+
+def _load(path: str) -> dict:
+    with open(path, encoding="utf-8") as fh:
+        return json.load(fh)
+
+
+def _source_blob(agent_repo: str) -> str:
+    parts = []
+    for dirpath, dirnames, filenames in os.walk(agent_repo):
+        dirnames[:] = [d for d in dirnames if d not in {".git", "__pycache__", ".venv"}]
+        for filename in filenames:
+            if filename.endswith((".py", ".md", ".json", ".yaml", ".yml", ".toml")):
+                try:
+                    with open(os.path.join(dirpath, filename), encoding="utf-8", errors="ignore") as fh:
+                        parts.append(fh.read())
+                except OSError:
+                    pass
+    return "\n".join(parts)
+
+
+_IDENTIFIER = re.compile(r"^[a-z][a-z0-9_]{2,}$")
+
+
+def audit(run_dir: str, agent_repo: str) -> int:
+    scenarios_dir = os.path.join(run_dir, "scenarios")
+    files = sorted(os.listdir(scenarios_dir)) if os.path.isdir(scenarios_dir) else []
+    if not files:
+        print("no scenarios found")
+        return 1
+    source = _source_blob(agent_repo)
+
+    kind_mix: Counter = Counter()
+    subgoal_uses: Counter = Counter()
+    deterministic = total = 0
+    failures: list[str] = []
+    run_declared_handles: list[str] = []
+
+    for name in files:
+        record = _load(os.path.join(scenarios_dir, name))
+        slug = record.get("id", name)
+        fact_values = {str(f.get("value", "")).lower() for f in record.get("facts") or []}
+        # A scenario may introduce a handle the agent's source cannot contain (an order
+        # reference) provided its own mock declares that value, which is what creates it during
+        # the run. Those are grounded; anything else invented is not.
+        declared = json.dumps(record.get("environment") or {})
+        pass
+        for sub_goal in record.get("sub_goals") or []:
+            checkpoint = (sub_goal or {}).get("checkpoint") or {}
+            definition = checkpoint.get("definition") or {}
+            kind = checkpoint.get("kind", "?")
+            kind_mix[kind] += 1
+            subgoal_uses[str(sub_goal.get("name"))] += 1
+            total += 1
+            if checkpoint.get("deterministic"):
+                deterministic += 1
+            tool = definition.get("tool") or definition.get("no_tool_call") or (
+                definition.get("no_tool_call_with") or {}
+            ).get("tool")
+            if tool and f"{tool}" not in source:
+                failures.append(f"{slug}: tool `{tool}` not found in agent source")
+            for arg, value in (definition.get("args_equal") or {}).items():
+                text = str(value)
+                if _IDENTIFIER.match(text) and text not in source:
+                    # Match the bare token: a handle may be declared in state_updates or
+                    # inside the mock's response text, where JSON escaping hides the quotes.
+                    if re.search(rf"\b{re.escape(text)}\b", declared):
+                        run_declared_handles.append(f"{slug}: {arg}={text}")
+                        continue
+                    failures.append(f"{slug}: args_equal {arg}={text} not found in agent source")
+        agent_input = str(record.get("agent_input", "")).lower()
+        for token in re.findall(r"[a-z][a-z0-9_]{4,}", agent_input):
+            if "_" in token and token in source and token not in fact_values:
+                failures.append(f"{slug}: agent_input leaks internal identifier `{token}`")
+
+    reused = sum(1 for count in subgoal_uses.values() if count >= 2)
+    print(f"scenarios: {len(files)}")
+    print(f"checkpoints: {total}, deterministic: {deterministic} ({100 * deterministic // max(total, 1)}%)")
+    print(f"kind mix: {dict(kind_mix)}")
+    print(f"sub-goal names reused in >=2 scenarios: {reused} of {len(subgoal_uses)}")
+    if run_declared_handles:
+        print(
+            f"run-created handles declared by the scenario's own mocks: "
+            f"{len(run_declared_handles)} (grounded, not failures)"
+        )
+    print(f"grounding failures: {len(failures)}")
+    for failure in failures[:30]:
+        print(f"  - {failure}")
+    return 0 if not failures else 2
+
+
+if __name__ == "__main__":
+    raise SystemExit(audit(sys.argv[1], sys.argv[2]))