diff --git a/docs/index.md b/docs/index.md index 3c83fd02..07d8234f 100644 --- a/docs/index.md +++ b/docs/index.md @@ -32,6 +32,7 @@ The installed CLI entrypoint is documented here as `pare` from `pare/main.py`. ```bash uv run pare scenarios list +uv run pare scenarios split --split full uv run pare benchmark sweep --split full --observe-model gpt-5 --execute-model gpt-5 uv run pare scenarios generate --num-scenarios 1 uv run pare annotation status @@ -46,6 +47,7 @@ Use `pare scenarios list` to see what scenarios are available and filter by app ```bash uv run pare scenarios list --apps StatefulEmailApp +uv run pare scenarios split --split full ``` ### 2. Run a benchmark sweep diff --git a/docs/scenarios/cli_usage.md b/docs/scenarios/cli_usage.md index c6a566e3..7748eaf0 100644 --- a/docs/scenarios/cli_usage.md +++ b/docs/scenarios/cli_usage.md @@ -1,11 +1,12 @@ # `pare scenarios` CLI Usage -This project exposes `pare scenarios` (implemented in `pare/cli/scenarios.py`) for two primary tasks: +This project exposes `pare scenarios` (implemented in `pare/cli/scenarios.py`) to: -- list benchmark scenarios under `pare/scenarios/benchmark/` -- generate new scenarios using the multi-step generator pipeline +- list registered scenarios from `pare/scenarios/` +- inspect benchmark split files and validate scenario ID lists +- generate new scenarios with the multi-step generator pipeline -If `pare` is not available in your shell, run via module entrypoint: +If `pare` is not available in your shell, run via the module entrypoint: ```bash uv run python -m pare.main scenarios --help @@ -13,7 +14,13 @@ uv run python -m pare.main scenarios --help ## List Scenarios -`pare scenarios list` scans benchmark modules for `@register_scenario("...")`. +`pare scenarios list` queries the PARE registry. By default, scenario-loading commands read `PARE_SCENARIOS_DIR`; if it is unset, they fall back to `benchmark`. + +Accepted directory values are relative to `pare/scenarios/`, for example: + +- `benchmark` +- `generator` +- `benchmark,generator` ### Examples @@ -21,6 +28,12 @@ uv run python -m pare.main scenarios --help # List all benchmark scenarios pare scenarios list +# Root shortcut for the list command +pare scenarios --list + +# Explicitly choose scenario directories +pare scenarios list --scenarios-dir benchmark,generator + # Require all listed apps (repeatable flags) pare scenarios list --apps StatefulEmailApp --apps StatefulCalendarApp @@ -39,12 +52,52 @@ pare scenarios list --json ### Flags -- `--benchmark-dir PATH`: override benchmark directory (default `pare/scenarios/benchmark`) -- `--apps, -a TEXT`: required-app filter (repeatable and/or comma-separated) +- `--scenarios-dir, --benchmark-dir TEXT`: override scenario directories to inspect; repeatable and/or comma-separated. Defaults to `PARE_SCENARIOS_DIR` or `benchmark`. +- `--apps, -a TEXT`: required-app filter; repeatable and/or comma-separated. All listed apps must be present in the scenario. - `--id-contains TEXT`: substring match on scenario IDs - `--limit INT`: max number of rows - `--json`: emit JSON output +## Benchmark Split Helpers + +These commands reuse the benchmark-loading helpers in `pare/benchmark/scenario_loader.py`. + +### Examples + +```bash +# Show the active splits directory and available split files +pare scenarios splits + +# List scenarios referenced by the full split +pare scenarios split --split full + +# List scenarios referenced by the ablation split as JSON +pare scenarios split --split ablation --json + +# Validate a scenario-id file against registered scenarios +pare scenarios check-ids-file data/splits/full.txt + +# Validate against other directories under pare/scenarios/ +pare scenarios check-ids-file my_ids.txt --scenarios-dir generator + +# Override the split directory used by the loader +PARE_BENCHMARK_SPLITS_DIR=data/splits pare scenarios split --split full +``` + +### Commands + +- `pare scenarios splits`: show the active benchmark splits directory and available `.txt` split files +- `pare scenarios split --split {full|ablation}`: load scenarios via the benchmark split helpers and print their metadata +- `pare scenarios check-ids-file FILE`: read one scenario ID per line and report which IDs are present or missing in the selected scenario directories + +### Relevant Flags + +- `pare scenarios split --scenarios-dir TEXT`: override scenario directories used during split resolution. Defaults to `PARE_SCENARIOS_DIR` or `benchmark`. +- `pare scenarios split --limit INT`: limit the number of listed scenarios +- `pare scenarios split --json`: output as JSON +- `pare scenarios check-ids-file --scenarios-dir TEXT`: validate IDs against different scenario directories under `pare/scenarios/` +- `pare scenarios check-ids-file --json`: output validation results as JSON and exit nonzero if any IDs are missing + ## Generate Scenarios `pare scenarios generate` runs `ScenarioGeneratingAgentOrchestrator`. @@ -86,7 +139,7 @@ pare scenarios generate --json --full-json ### Flags - `--output-dir PATH`: directory for intermediate step files -- `--trajectory-dir PATH`: trajectory directory (or base dir for multi-run) +- `--trajectory-dir PATH`: trajectory directory or base dir for multi-run generation - `--num-scenarios INT`: number of scenarios to generate - `--max-iterations INT`: retry budget per step - `--resume-from-step TEXT`: `step2`, `step3`, or `step4` diff --git a/docs/scenarios/index.md b/docs/scenarios/index.md index 52fa7fb2..9b13869a 100644 --- a/docs/scenarios/index.md +++ b/docs/scenarios/index.md @@ -13,6 +13,13 @@ If you mainly want to use the benchmark, this is the most important section. The uv run pare scenarios list ``` +### Inspect benchmark split files + +```bash +uv run pare scenarios splits +uv run pare scenarios split --split full +``` + ### Filter scenarios by app usage ```bash diff --git a/pare/cli/scenarios.py b/pare/cli/scenarios.py new file mode 100644 index 00000000..1e264731 --- /dev/null +++ b/pare/cli/scenarios.py @@ -0,0 +1,666 @@ +"""Scenario CLI commands for listing and generating PARE scenarios.""" + +from __future__ import annotations + +import inspect +import json +import logging +import os +from contextlib import contextmanager +from dataclasses import dataclass +from pathlib import Path +from typing import TYPE_CHECKING, Annotated, Any + +import typer +from dotenv import load_dotenv + +from pare.benchmark.scenario_loader import ( + Split, + get_splits_dir, + load_scenario_ids_from_file, + load_scenarios_by_split, + load_scenarios_from_registry, +) + +if TYPE_CHECKING: + from collections.abc import Iterator + + from pare.scenarios import PAREScenario + +logger = logging.getLogger(__name__) + +app = typer.Typer( + name="scenarios", + help="List registered PARE scenarios and run the multi-step scenario generator", + no_args_is_help=True, +) + + +_SCENARIOS_BASE_DIR = Path(__file__).resolve().parents[1] / "scenarios" +_SCENARIO_METADATA_PATH = _SCENARIOS_BASE_DIR / "scenario_metadata.json" + + +def _expand_csv_list(values: list[str] | None) -> list[str]: + if not values: + return [] + expanded: list[str] = [] + for item in values: + expanded.extend(part.strip() for part in item.split(",") if part.strip()) + return expanded + + +def _configured_scenarios_dir_inputs(scenarios_dirs: list[str] | None = None) -> list[str]: + configured = _expand_csv_list(scenarios_dirs) + if not configured: + env_value = os.getenv("PARE_SCENARIOS_DIR", "benchmark") + configured = [part.strip() for part in env_value.split(",") if part.strip()] + return list(dict.fromkeys(configured or ["benchmark"])) + + +def _load_scenario_metadata(path: Path) -> dict[str, dict[str, Any]]: + try: + raw = json.loads(path.read_text(encoding="utf-8")) + except FileNotFoundError: + return {} + except json.JSONDecodeError as exc: + logger.warning("Failed to parse scenario metadata JSON at %s: %s", path, exc) + return {} + if not isinstance(raw, list): + return {} + mapping: dict[str, dict[str, Any]] = {} + for entry in raw: + if not isinstance(entry, dict): + continue + scenario_id = entry.get("scenario_id") + if isinstance(scenario_id, str) and scenario_id: + mapping[scenario_id] = entry + return mapping + + +@dataclass(frozen=True) +class ScenarioListing: + """Lightweight representation of a scenario for CLI output.""" + + scenario_id: str + class_name: str | None + file_path: Path + apps: list[str] + description: str | None + + +def _reset_pare_registry_discovery_state() -> None: + from pare.scenarios import registry + + if hasattr(registry, "_scenarios_discovered"): + registry._scenarios_discovered = False + scenarios = getattr(registry, "_scenarios", None) + if isinstance(scenarios, dict): + scenarios.clear() + + +def _resolve_registry_dir_name(scenarios_dir: str | Path) -> tuple[Path, str]: + raw_dir = Path(scenarios_dir) + candidate_dir = raw_dir if raw_dir.is_absolute() else _SCENARIOS_BASE_DIR / raw_dir + resolved_dir = candidate_dir.resolve() + if not resolved_dir.exists(): + raise typer.BadParameter(f"Scenario directory does not exist: {resolved_dir}") + if not resolved_dir.is_dir(): + raise typer.BadParameter(f"Scenario directory is not a directory: {resolved_dir}") + + try: + relative_dir = resolved_dir.relative_to(_SCENARIOS_BASE_DIR.resolve()) + except ValueError as exc: + raise typer.BadParameter(f"Scenario directory must live under {_SCENARIOS_BASE_DIR}") from exc + + return resolved_dir, relative_dir.as_posix() + + +@contextmanager +def _temporary_scenarios_dirs(scenarios_dirs: list[str] | None = None) -> Iterator[list[Path]]: + resolved_dirs_with_names = [ + _resolve_registry_dir_name(dir_input) for dir_input in _configured_scenarios_dir_inputs(scenarios_dirs) + ] + resolved_dirs = [path for path, _ in resolved_dirs_with_names] + scenarios_dir_names = [dir_name for _, dir_name in resolved_dirs_with_names] + previous_dirs = os.environ.get("PARE_SCENARIOS_DIR") + + try: + os.environ["PARE_SCENARIOS_DIR"] = ",".join(scenarios_dir_names) + _reset_pare_registry_discovery_state() + yield resolved_dirs + finally: + if previous_dirs is None: + os.environ.pop("PARE_SCENARIOS_DIR", None) + else: + os.environ["PARE_SCENARIOS_DIR"] = previous_dirs + _reset_pare_registry_discovery_state() + + +def _scenario_file_belongs_to_dirs(file_path: Path, scenarios_dirs: list[Path]) -> bool: + resolved_file = file_path.resolve() + for scenarios_dir in scenarios_dirs: + try: + resolved_file.relative_to(scenarios_dir.resolve()) + except ValueError: + continue + else: + return True + return False + + +def _load_registered_scenario_ids(scenarios_dirs: list[str] | None = None) -> list[str]: + from pare.scenarios import registry + + with _temporary_scenarios_dirs(scenarios_dirs): + return sorted(registry.get_all_scenarios().keys()) + + +def _load_apps_for_listing(scenario_id: str, scenario: PAREScenario, metadata: dict[str, dict[str, Any]]) -> list[str]: + meta = metadata.get(scenario_id, {}) + apps_from_meta = meta.get("apps") if isinstance(meta, dict) else None + if isinstance(apps_from_meta, list) and all(isinstance(app, str) for app in apps_from_meta): + return list(apps_from_meta) + + try: + scenario.init_and_populate_apps(sandbox_dir=Path("sandbox")) + except TypeError: + try: + scenario.init_and_populate_apps() + except Exception as exc: + logger.warning("Failed to initialize scenario %s for app listing: %s", scenario_id, exc) + return [] + except Exception as exc: + logger.warning("Failed to initialize scenario %s for app listing: %s", scenario_id, exc) + return [] + + apps = getattr(scenario, "apps", None) + if not isinstance(apps, list): + return [] + + app_names = [app.__class__.__name__ for app in apps] + return sorted(dict.fromkeys(app_names)) + + +def _load_registered_scenario_listings(scenarios_dirs: list[str] | None = None) -> list[ScenarioListing]: + from pare.scenarios import registry + + metadata = _load_scenario_metadata(_SCENARIO_METADATA_PATH) + with _temporary_scenarios_dirs(scenarios_dirs) as resolved_dirs: + scenario_ids = sorted(registry.get_all_scenarios().keys()) + scenarios = list(load_scenarios_from_registry(scenario_ids=scenario_ids)) + + listings: list[ScenarioListing] = [] + for scenario in scenarios: + scenario_id = getattr(scenario, "scenario_id", None) + if not isinstance(scenario_id, str) or not scenario_id: + logger.warning("Skipping scenario instance without scenario_id: %r", scenario) + continue + + scenario_class = scenario.__class__ + try: + file_path = Path(inspect.getfile(scenario_class)).resolve() + except TypeError: + logger.warning("Skipping scenario %s because its source file could not be determined", scenario_id) + continue + + if not _scenario_file_belongs_to_dirs(file_path, resolved_dirs): + continue + + meta = metadata.get(scenario_id, {}) + description = None + if isinstance(meta, dict) and isinstance(meta.get("description"), str): + description = meta["description"].strip() or None + else: + class_doc = inspect.getdoc(scenario_class) + if isinstance(class_doc, str): + description = class_doc.strip() or None + + listings.append( + ScenarioListing( + scenario_id=scenario_id, + class_name=getattr(scenario_class, "__name__", None), + file_path=file_path, + apps=_load_apps_for_listing(scenario_id, scenario, metadata), + description=description, + ) + ) + + return listings + + +@app.callback(invoke_without_command=True) +def scenarios_root( + ctx: typer.Context, + list_only: Annotated[ + bool, + typer.Option("--list", help="Alias for `pare scenarios list`"), + ] = False, + scenarios_dirs: Annotated[ + list[str] | None, + typer.Option( + "--scenarios-dir", + "--benchmark-dir", + help=( + "Scenario directories under pare/scenarios/ (repeat or comma-separate). " + "Defaults to PARE_SCENARIOS_DIR or benchmark." + ), + ), + ] = None, + apps: Annotated[ + list[str] | None, + typer.Option("--apps", "-a", help="Filter by required apps (repeat or comma-separate)"), + ] = None, + id_contains: Annotated[ + str | None, + typer.Option("--id-contains", help="Only include scenario_ids containing this substring"), + ] = None, + limit: Annotated[ + int | None, + typer.Option("--limit", help="Optional limit on number of results shown"), + ] = None, + as_json: Annotated[ + bool, + typer.Option("--json", help="Output as JSON"), + ] = False, +) -> None: + """Handle root-level shortcuts such as `pare scenarios --list`.""" + if ctx.invoked_subcommand is not None: + return + if list_only: + ctx.invoke( + list_scenarios, + scenarios_dirs=scenarios_dirs, + apps=apps, + id_contains=id_contains, + limit=limit, + as_json=as_json, + ) + return + typer.echo(ctx.get_help()) + + +def _scenario_to_listing(scenario: PAREScenario, metadata: dict[str, dict[str, Any]]) -> ScenarioListing: + scenario_id = str(getattr(scenario, "scenario_id", "")).strip() + scenario_class = scenario.__class__ + file_path = Path(inspect.getfile(scenario_class)).resolve() + meta = metadata.get(scenario_id, {}) + + description = None + if isinstance(meta, dict) and isinstance(meta.get("description"), str): + description = meta["description"].strip() or None + else: + class_doc = inspect.getdoc(scenario_class) + if isinstance(class_doc, str): + description = class_doc.strip() or None + + return ScenarioListing( + scenario_id=scenario_id, + class_name=getattr(scenario_class, "__name__", None), + file_path=file_path, + apps=_load_apps_for_listing(scenario_id, scenario, metadata), + description=description, + ) + + +def _emit_listings(listings: list[ScenarioListing], *, as_json: bool) -> None: + if as_json: + payload = [ + { + "scenario_id": s.scenario_id, + "class_name": s.class_name, + "file": str(s.file_path), + "apps": s.apps, + "description": s.description, + } + for s in listings + ] + typer.echo(json.dumps(payload, indent=2, ensure_ascii=False)) + return + + if not listings: + typer.echo("No scenarios found.") + return + + for item in listings: + if item.class_name: + header = f"Scenario ID: {item.scenario_id} | Class Name: {item.class_name}" + else: + header = f"Scenario ID: {item.scenario_id}" + typer.echo(header) + typer.echo(f" File Path: {item.file_path}") + typer.echo(f" Apps used: {', '.join(item.apps) if item.apps else '(unknown)'}") + if item.description: + first_line = item.description.splitlines()[0].strip() + if first_line: + typer.echo(f" Description: {first_line}") + typer.echo("") + + +@app.command("list") +def list_scenarios( + scenarios_dirs: Annotated[ + list[str] | None, + typer.Option( + "--scenarios-dir", + "--benchmark-dir", + help=( + "Scenario directories under pare/scenarios/ (repeat or comma-separate). " + "Defaults to PARE_SCENARIOS_DIR or benchmark." + ), + ), + ] = None, + apps: Annotated[ + list[str] | None, + typer.Option("--apps", "-a", help="Filter by required apps (repeat or comma-separate)"), + ] = None, + id_contains: Annotated[ + str | None, + typer.Option("--id-contains", help="Only include scenario_ids containing this substring"), + ] = None, + limit: Annotated[ + int | None, + typer.Option("--limit", help="Optional limit on number of results shown"), + ] = None, + as_json: Annotated[ + bool, + typer.Option("--json", help="Output as JSON"), + ] = False, +) -> None: + """List registered PARE scenarios from configured scenario directories.""" + listings = _load_registered_scenario_listings(scenarios_dirs) + + required_apps = set(_expand_csv_list(apps)) + if id_contains: + needle = id_contains.lower() + listings = [s for s in listings if needle in s.scenario_id.lower()] + + if required_apps: + filtered: list[ScenarioListing] = [] + for item in listings: + item_apps = set(item.apps) + if required_apps.issubset(item_apps): + filtered.append(item) + listings = filtered + + if limit is not None: + listings = listings[: max(0, limit)] + + _emit_listings(listings, as_json=as_json) + + +@app.command("split") +def list_split_scenarios( + split: Annotated[ + Split, + typer.Option("--split", help="Benchmark split to list"), + ] = Split.FULL, + scenarios_dirs: Annotated[ + list[str] | None, + typer.Option( + "--scenarios-dir", + "--benchmark-dir", + help=( + "Scenario directories under pare/scenarios/ (repeat or comma-separate). " + "Defaults to PARE_SCENARIOS_DIR or benchmark." + ), + ), + ] = None, + limit: Annotated[ + int | None, + typer.Option("--limit", help="Optional limit on number of results shown"), + ] = None, + as_json: Annotated[ + bool, + typer.Option("--json", help="Output as JSON"), + ] = False, +) -> None: + """List scenarios referenced by a benchmark split file.""" + metadata = _load_scenario_metadata(_SCENARIO_METADATA_PATH) + with _temporary_scenarios_dirs(scenarios_dirs): + scenarios = list(load_scenarios_by_split(split=split, limit=limit)) + listings = [_scenario_to_listing(scenario, metadata) for scenario in scenarios] + _emit_listings(listings, as_json=as_json) + + +@app.command("check-ids-file") +def check_ids_file( + file_path: Annotated[ + Path, + typer.Argument(help="Path to a file containing one scenario ID per line"), + ], + scenarios_dirs: Annotated[ + list[str] | None, + typer.Option( + "--scenarios-dir", + "--benchmark-dir", + help=( + "Scenario directories under pare/scenarios/ (repeat or comma-separate). " + "Defaults to PARE_SCENARIOS_DIR or benchmark." + ), + ), + ] = None, + as_json: Annotated[ + bool, + typer.Option("--json", help="Output as JSON"), + ] = False, +) -> None: + """Validate that scenario IDs from a file exist in the configured PARE scenario directories.""" + scenario_ids = load_scenario_ids_from_file(file_path) + configured_dirs = _configured_scenarios_dir_inputs(scenarios_dirs) + registered_ids = set(_load_registered_scenario_ids(scenarios_dirs)) + present = [scenario_id for scenario_id in scenario_ids if scenario_id in registered_ids] + missing = [scenario_id for scenario_id in scenario_ids if scenario_id not in registered_ids] + + if as_json: + typer.echo( + json.dumps( + { + "file": str(file_path), + "scenarios_dirs": configured_dirs, + "total_ids": len(scenario_ids), + "present_ids": present, + "missing_ids": missing, + }, + indent=2, + ensure_ascii=False, + ) + ) + if missing: + raise typer.Exit(code=1) + return + + typer.echo(f"Checked {len(scenario_ids)} scenario IDs from {file_path}") + typer.echo(f"Scenario directories: {', '.join(configured_dirs)}") + typer.echo(f"Present: {len(present)}") + typer.echo(f"Missing: {len(missing)}") + if missing: + typer.echo("Missing IDs:") + for scenario_id in missing: + typer.echo(f"- {scenario_id}") + raise typer.Exit(code=1) + + +@app.command("splits") +def show_splits_info( + as_json: Annotated[ + bool, + typer.Option("--json", help="Output as JSON"), + ] = False, +) -> None: + """Show the benchmark splits directory and available split files.""" + splits_dir = get_splits_dir() + available = sorted(path.stem for path in splits_dir.glob("*.txt")) if splits_dir.exists() else [] + + if as_json: + typer.echo( + json.dumps({"splits_dir": str(splits_dir), "available_splits": available}, indent=2, ensure_ascii=False) + ) + return + + typer.echo(f"Splits directory: {splits_dir}") + typer.echo(f"Available splits: {', '.join(available) if available else '(none found)'}") + + +@app.command() +def generate( # noqa: C901 + output_dir: Annotated[ + Path | None, + typer.Option("--output-dir", help="Directory where intermediate step files should be written"), + ] = None, + model: Annotated[ + str, + typer.Option("--model", help="(Currently unused) Kept for parity with the generator entrypoint"), + ] = "gpt-5-mini-2025-08-07", + provider: Annotated[ + str, + typer.Option("--provider", help="(Currently unused) Kept for parity with the generator entrypoint"), + ] = "openai", + endpoint: Annotated[ + str | None, + typer.Option("--endpoint", help="(Currently unused) Optional custom endpoint for the provider"), + ] = None, + max_iterations: Annotated[ + int, + typer.Option("--max-iterations", help="Maximum number of attempts per step"), + ] = 2, + resume_from_step2: Annotated[ + bool, + typer.Option("--resume-from-step2", help="[DEPRECATED] Prefer --resume-from-step step2"), + ] = False, + resume_from_step: Annotated[ + str | None, + typer.Option("--resume-from-step", help="Resume from a specific step (step2|step3|step4)"), + ] = None, + trajectory_dir: Annotated[ + Path | None, + typer.Option( + "--trajectory-dir", help="Optional path to an existing trajectory dir (or base dir for multiple runs)" + ), + ] = None, + num_scenarios: Annotated[ + int, + typer.Option("--num-scenarios", help="Number of distinct scenarios to generate in this invocation"), + ] = 1, + debug_prompts: Annotated[ + bool, + typer.Option("--debug-prompts", help="If set, skip LLM calls and print prompts for all agents instead"), + ] = False, + apps: Annotated[ + list[str] | None, + typer.Option( + "--apps", + "-a", + help=( + "Explicit list of app class names to include (repeat or comma-separate). " + "PAREAgentUserInterface and HomeScreenSystemApp are always available." + ), + ), + ] = None, + as_json: Annotated[ + bool, + typer.Option("--json", help="Output results as JSON (compact by default)"), + ] = False, + full_json: Annotated[ + bool, + typer.Option("--full-json", help="When using --json, include large fields like step conversations"), + ] = False, +) -> None: + """Run the multi-step scenario generator.""" + from pare.scenarios.generator.agent.scenario_generating_agent_orchestrator import ( + ScenarioGeneratingAgentOrchestrator, + ) + from pare.scenarios.generator.scenario_generator import determine_selected_apps, prepare_prompt_context_data + from pare.scenarios.generator.utils.apps_init_instructions import ScenarioWithAllPAREApps + + logging.basicConfig(level=logging.INFO) + + if model or provider or endpoint: + logger.debug( + "Generator CLI params (currently unused): model=%s provider=%s endpoint=%s", model, provider, endpoint + ) + + if resume_from_step is not None and resume_from_step not in {"step2", "step3", "step4"}: + raise typer.BadParameter("--resume-from-step must be one of: step2, step3, step4") + + load_dotenv() + + requested_apps = _expand_csv_list(apps) if apps is not None else None + + app_def_scenario = ScenarioWithAllPAREApps() + app_def_scenario.initialize() + app_instances = { + app_instance.__class__.__name__: app_instance for app_instance in getattr(app_def_scenario, "apps", []) + } + selected_apps = determine_selected_apps(app_instances, requested_apps) + if not selected_apps: + logger.warning("No selectable apps found; continuing with system apps only.") + prompt_context = prepare_prompt_context_data(app_def_scenario, selected_apps) + + if resume_from_step2 and resume_from_step is None: + logger.warning("--resume-from-step2 is deprecated; prefer --resume-from-step step2 instead.") + + total = max(1, num_scenarios) + results: list[dict[str, Any]] = [] + for idx in range(total): + run_trajectory_dir = trajectory_dir + if trajectory_dir is not None and total > 1: + run_trajectory_dir = trajectory_dir / f"run_{idx + 1}" + + agent = ScenarioGeneratingAgentOrchestrator( + output_dir=output_dir, + max_iterations=max_iterations, + trajectory_dir=run_trajectory_dir, + prompt_context=prompt_context, + debug_prompts=debug_prompts, + resume_from_step2=resume_from_step2, + resume_from_step=resume_from_step, + ) + try: + result = agent.run() + except Exception as exc: + logger.exception("Scenario generation failed for run %s/%s.", idx + 1, total) + results.append({ + "run_index": idx + 1, + "status": "failed", + "error": str(exc), + "trajectory_dir": str(run_trajectory_dir) if run_trajectory_dir is not None else None, + }) + continue + + results.append({ + "run_index": idx + 1, + "status": "success", + **result, + }) + + if as_json: + payload: list[dict[str, Any]] = [] + for item in results: + if full_json: + payload.append(item) + continue + compact = dict(item) + step_results = compact.get("step_results") + if isinstance(step_results, list): + compact_steps: list[Any] = [] + for step in step_results: + if isinstance(step, dict): + compact_step = dict(step) + compact_step.pop("conversation", None) + compact_steps.append(compact_step) + else: + compact_steps.append(str(step)) + compact["step_results"] = compact_steps + compact.pop("conversation", None) + payload.append(compact) + typer.echo(json.dumps(payload, indent=2, ensure_ascii=False, default=str)) + else: + failed = [r for r in results if r.get("status") != "success"] + typer.echo(f"Completed {total} scenario runs: {total - len(failed)} success, {len(failed)} failed.") + for item in results: + trajectory = item.get("trajectory_dir") + trajectory_suffix = f" (trajectory_dir={trajectory})" if trajectory else "" + typer.echo(f"- run {item.get('run_index')}: {item.get('status')}{trajectory_suffix}") + + if any(r.get("status") != "success" for r in results): + raise typer.Exit(code=1) diff --git a/pare/main.py b/pare/main.py index dcaa280d..3b973f5d 100644 --- a/pare/main.py +++ b/pare/main.py @@ -4,7 +4,7 @@ import typer -from pare.cli import annotation, benchmark, cache +from pare.cli import annotation, benchmark, cache, scenarios app = typer.Typer( name="pare", @@ -16,6 +16,7 @@ app.add_typer(annotation.app, name="annotation") app.add_typer(benchmark.app, name="benchmark") app.add_typer(cache.app, name="cache") +app.add_typer(scenarios.app, name="scenarios") def main() -> None: diff --git a/pare/scenarios/generator/agent/scenario_generating_agent_orchestrator.py b/pare/scenarios/generator/agent/scenario_generating_agent_orchestrator.py index b5650697..5658b00f 100644 --- a/pare/scenarios/generator/agent/scenario_generating_agent_orchestrator.py +++ b/pare/scenarios/generator/agent/scenario_generating_agent_orchestrator.py @@ -18,7 +18,7 @@ from pare.scenarios.generator.prompt.scenario_generating_agent_prompts import ( configure_dynamic_context, ) -from scripts.run_scenarios import run_scenarios +from pare.scenarios.run_scenarios import run_scenarios from .claude_backend import ClaudeAgentRuntimeConfig, ClaudeFilesystemConfig from .scenario_uniqueness_agent import ScenarioUniquenessCheckAgent diff --git a/pare/scenarios/run_scenarios.py b/pare/scenarios/run_scenarios.py new file mode 100644 index 00000000..b2d1990b --- /dev/null +++ b/pare/scenarios/run_scenarios.py @@ -0,0 +1,368 @@ +"""Helpers and CLI entrypoint for running PARE scenarios.""" + +from __future__ import annotations + +import argparse +import json +import logging +import sys +import time +from dataclasses import asdict, dataclass, field +from datetime import UTC, datetime +from pathlib import Path +from typing import Any + +from dotenv import load_dotenv + +from pare.cli.utils import get_pst_time, run_scenario_by_id, setup_logging +from pare.scenarios.utils.registry import registry + +logger = logging.getLogger(__name__) + + +@dataclass +class ScenarioResult: + """Result of running a single scenario.""" + + scenario_name: str + success: bool + rationale: str | None = None + exception: Exception | None = None + duration_seconds: float = 0.0 + export_path: str | None = None + + +@dataclass +class ResultsSummary: + """Summary of all scenario runs.""" + + timestamp: str + config: dict[str, Any] = field(default_factory=dict) + total_scenarios: int = 0 + passed: int = 0 + failed: int = 0 + results: list[ScenarioResult] = field(default_factory=list) + + +def run_scenarios( + scenario_names: list[str], + user_model: str = "gpt-4o-mini", + user_model_provider: str = "openai", + proactive_model: str = "gpt-4o-mini", + proactive_model_provider: str = "openai", + max_turns: int | None = 10, + user_max_iterations: int = 1, + observe_max_iterations: int = 10, + execute_max_iterations: int = 10, + traces_dir: str = "traces", + logs_dir: str = "logs", + oracle_mode: bool = False, + tool_failure_prob: float = 0.0, + env_events_per_min: float = 0.0, + env_events_seed: int = 42, + stop_on_failure: bool = False, + log_level: str = "WARNING", +) -> ResultsSummary: + """Run specified scenarios and collect results.""" + logger.info(f"Running {len(scenario_names)} scenarios: {scenario_names}") + + summary = ResultsSummary( + timestamp=datetime.now(UTC).isoformat(), + config={ + "user_model": user_model, + "user_model_provider": user_model_provider, + "proactive_model": proactive_model, + "proactive_model_provider": proactive_model_provider, + "max_turns": max_turns, + "user_max_iterations": user_max_iterations, + "observe_max_iterations": observe_max_iterations, + "execute_max_iterations": execute_max_iterations, + "oracle_mode": oracle_mode, + "tool_failure_prob": tool_failure_prob, + "env_events_per_min": env_events_per_min, + "env_events_seed": env_events_seed, + }, + total_scenarios=len(scenario_names), + ) + + green = "\033[92m" + reset = "\033[0m" + + for idx, scenario_name in enumerate(scenario_names, start=1): + print("*" * 80) + print(f"{green}Running Scenario {idx}/{len(scenario_names)}: {scenario_name}{reset}") + print("*" * 80) + + current_logs_dir = Path(logs_dir) / scenario_name + current_logs_dir.mkdir(parents=True, exist_ok=True) + + setup_logging( + level=log_level, + log_dir=current_logs_dir, + use_tqdm=True, + log_to_file=True, + verbose=False, + ) + + start_time = time.time() + result = ScenarioResult(scenario_name=scenario_name, success=False) + + try: + validation_result = run_scenario_by_id( + scenario_name=scenario_name, + user_model=user_model, + user_model_provider=user_model_provider, + proactive_model=proactive_model, + proactive_model_provider=proactive_model_provider, + max_turns=max_turns, + user_max_iterations=user_max_iterations, + observe_max_iterations=observe_max_iterations, + execute_max_iterations=execute_max_iterations, + traces_dir=traces_dir, + oracle_mode=oracle_mode, + tool_failure_prob=tool_failure_prob, + env_events_per_min=env_events_per_min, + env_events_seed=env_events_seed, + ) + + result.success = validation_result.success + result.rationale = validation_result.rationale + result.export_path = validation_result.export_path + + if validation_result.exception: + result.exception = validation_result.exception + + except Exception as exc: + result.success = False + result.exception = exc + logger.exception(f"Exception while running scenario {scenario_name}") + + result.duration_seconds = time.time() - start_time + + summary.results.append(result) + if result.success: + summary.passed += 1 + else: + summary.failed += 1 + + status = "SUCCESS" if result.success else "FAILED" + logger.info(f"Completed {scenario_name}: {status} (took {result.duration_seconds:.2f}s)") + + if stop_on_failure and not result.success: + logger.warning(f"Stopping due to --stop-on-failure flag after {scenario_name}") + break + + return summary + + +def save_results_summary(summary: ResultsSummary, output_path: Path) -> None: + """Save results summary to JSON file.""" + summary_dict = asdict(summary) + + with open(output_path, "w") as file: + json.dump(summary_dict, file, indent=2) + + logger.info(f"Results summary saved to: {output_path}") + + +def print_summary_table(summary: ResultsSummary) -> None: + """Print a summary table to console.""" + print("\n" + "=" * 80) + print("SCENARIO RUN SUMMARY") + print("=" * 80) + print(f"Total: {summary.total_scenarios} | Passed: {summary.passed} | Failed: {summary.failed}") + print("-" * 80) + print(f"{'Scenario':<40} {'Status':<10} {'Duration':<12} {'Rationale'}") + print("-" * 80) + + for result in summary.results: + status = "PARES" if result.success else "FAIL" + duration = f"{result.duration_seconds:.2f}s" + rationale = ( + result.rationale[:30] + "..." + if result.rationale and len(result.rationale) > 30 + else (result.rationale or "") + ) + print(f"{result.scenario_name:<40} {status:<10} {duration:<12} {rationale}") + + print("=" * 80) + + +def main(argv: list[str] | None = None) -> None: + """Parse CLI arguments and run scenarios.""" + parser = argparse.ArgumentParser( + description="Run PARE scenarios", + formatter_class=argparse.ArgumentDefaultsHelpFormatter, + ) + parser.add_argument( + "--scenarios", + nargs="*", + default=None, + help="Scenario IDs to run. If not specified, runs all registered scenarios.", + ) + parser.add_argument( + "--user-model", + default="gpt-4o-mini", + help="LLM model for user agent", + ) + parser.add_argument( + "--user-model-provider", + default="openai", + help="Provider for user model", + ) + parser.add_argument( + "--proactive-model", + default="gpt-4o-mini", + help="LLM model for proactive agents (observe and execute)", + ) + parser.add_argument( + "--proactive-model-provider", + default="openai", + help="Provider for proactive model", + ) + parser.add_argument( + "--max-turns", + type=int, + default=10, + help="Maximum number of agent turns (0 for unlimited)", + ) + parser.add_argument( + "--user-max-iterations", + type=int, + default=1, + help="Maximum number of iterations for the user agent", + ) + parser.add_argument( + "--observe-max-iterations", + type=int, + default=10, + help="Maximum number of iterations for the proactive observe agent", + ) + parser.add_argument( + "--execute-max-iterations", + type=int, + default=10, + help="Maximum number of iterations for the proactive execute agent", + ) + parser.add_argument( + "--traces-dir", + default="traces", + help="Base directory to export traces to", + ) + parser.add_argument( + "--tool-failure-prob", + type=float, + default=0.0, + help="Probability (0.0-1.0) that agent tools fail", + ) + parser.add_argument( + "--env-events-per-min", + type=float, + default=0.0, + help="Average number of environmental noise events per minute", + ) + parser.add_argument( + "--env-events-seed", + type=int, + default=42, + help="Random seed for reproducible noise generation", + ) + parser.add_argument( + "--oracle", + action="store_true", + help="Run in oracle mode (executes predefined oracle events without agents)", + ) + parser.add_argument( + "--experiment-name", + default="experiment", + help="Name of the experiment", + ) + parser.add_argument( + "--stop-on-failure", + action="store_true", + help="Stop execution on first scenario failure", + ) + parser.add_argument( + "--log-level", + default="WARNING", + help="Logging level", + choices=["DEBUG", "INFO", "WARNING", "ERROR"], + ) + + args = parser.parse_args(argv) + + load_dotenv() + + all_scenarios = registry.get_all_scenarios() + + if args.scenarios is None or len(args.scenarios) == 0: + scenario_names = sorted(all_scenarios.keys()) + print(f"Running all {len(scenario_names)} registered scenarios") + else: + if len(args.scenarios) == 1 and Path(args.scenarios[0]).is_file(): + scenarios_file = Path(args.scenarios[0]) + scenario_names = [] + with open(scenarios_file) as file: + for line in file: + line = line.strip() + if not line or line.startswith("#"): + continue + scenario_names.append(line) + print(f"Loaded {len(scenario_names)} scenarios from {scenarios_file}") + else: + scenario_names = args.scenarios + + for name in scenario_names: + if name not in all_scenarios: + print(f"Error: Unknown scenario '{name}'") + print(f"Available scenarios: {sorted(all_scenarios.keys())}") + sys.exit(1) + print(f"Running {len(scenario_names)} scenarios: {scenario_names}") + + config_suffix = ( + f"{args.experiment_name}_user_{args.user_model}" + f"_mt_{args.max_turns}_umi_{args.user_max_iterations}_omi_{args.observe_max_iterations}" + f"_emi_{args.execute_max_iterations}_enmi_{args.env_events_per_min}_es_{args.env_events_seed}" + f"_tfp_{args.tool_failure_prob}" + ) + + run_timestamp = get_pst_time() + + traces_base = Path(args.traces_dir) if Path(args.traces_dir).is_absolute() else (Path.cwd() / args.traces_dir) + traces_dir = traces_base / config_suffix / f"{args.proactive_model}" + traces_dir.mkdir(parents=True, exist_ok=True) + print(f"Traces directory: {traces_dir}") + + logs_dir = Path("logs") / config_suffix / f"{args.proactive_model}_{run_timestamp}" + logs_dir.mkdir(parents=True, exist_ok=True) + print(f"Logs directory: {logs_dir}") + + max_turns = args.max_turns if args.max_turns > 0 else None + + summary = run_scenarios( + scenario_names=scenario_names, + user_model=args.user_model, + user_model_provider=args.user_model_provider, + proactive_model=args.proactive_model, + proactive_model_provider=args.proactive_model_provider, + max_turns=max_turns, + user_max_iterations=args.user_max_iterations, + observe_max_iterations=args.observe_max_iterations, + execute_max_iterations=args.execute_max_iterations, + traces_dir=str(traces_dir), + logs_dir=str(logs_dir), + oracle_mode=args.oracle, + tool_failure_prob=args.tool_failure_prob, + env_events_per_min=args.env_events_per_min, + env_events_seed=args.env_events_seed, + stop_on_failure=args.stop_on_failure, + log_level=args.log_level, + ) + + results_path = traces_dir / "result_summary.json" + save_results_summary(summary, results_path) + print_summary_table(summary) + + +if __name__ == "__main__": # pragma: no cover + main() diff --git a/pyproject.toml b/pyproject.toml index e8bc6d5c..fc108d68 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -23,14 +23,15 @@ dependencies = [ "jinja2>=3.1.6", "numpy>=2.2.6", "pytz>=2025.2", - "boto3>=1.40.70", - "tenacity>=9.1.2", "claude-agent-sdk>=0.1.19", "xxhash>=3.6.0", "typer>=0.20.0", "fastapi>=0.115.0", "uvicorn>=0.34.0", "polars>=1.20.0", + "pydantic>=2.10.6", + "tqdm>=4.67.1", + "docstring-parser>=0.16", ] [project.urls] @@ -214,6 +215,7 @@ source = ["pare"] [tool.deptry] known_first_party = ["pare"] +extend_exclude = ["pare/_archives/.*"] [tool.deptry.package_module_name_map] "meta-agents-research-environments" = "are" diff --git a/tests/cli/test_scenarios_cli.py b/tests/cli/test_scenarios_cli.py new file mode 100644 index 00000000..b418dcad --- /dev/null +++ b/tests/cli/test_scenarios_cli.py @@ -0,0 +1,69 @@ +from __future__ import annotations + +from pathlib import Path + +from typer.testing import CliRunner + +from pare.main import app + +runner = CliRunner() + + +def test_scenarios_help_includes_registered_commands() -> None: + result = runner.invoke(app, ["scenarios", "--help"]) + + assert result.exit_code == 0, result.output + assert "list" in result.output + assert "split" in result.output + assert "splits" in result.output + assert "check-ids-file" in result.output + assert "generate" in result.output + + +def test_scenarios_root_list_shortcut(monkeypatch) -> None: + def fake_list_scenarios( + scenarios_dirs: list[str] | None = None, + apps: list[str] | None = None, + id_contains: str | None = None, + limit: int | None = None, + as_json: bool = False, + ) -> None: + assert scenarios_dirs == ["generator"] + assert apps == ["StatefulEmailApp"] + assert id_contains == "meeting" + assert limit == 1 + assert as_json is True + + monkeypatch.setattr("pare.cli.scenarios.list_scenarios", fake_list_scenarios) + + result = runner.invoke( + app, + [ + "scenarios", + "--list", + "--scenarios-dir", + "generator", + "--apps", + "StatefulEmailApp", + "--id-contains", + "meeting", + "--limit", + "1", + "--json", + ], + ) + + assert result.exit_code == 0, result.output + + +def test_splits_json_reports_available_split_files(tmp_path: Path, monkeypatch) -> None: + (tmp_path / "full.txt").write_text("scenario_a\n", encoding="utf-8") + (tmp_path / "ablation.txt").write_text("scenario_b\n", encoding="utf-8") + monkeypatch.setenv("PARE_BENCHMARK_SPLITS_DIR", str(tmp_path)) + + result = runner.invoke(app, ["scenarios", "splits", "--json"]) + + assert result.exit_code == 0, result.output + assert '"available_splits": [' in result.output + assert '"ablation"' in result.output + assert '"full"' in result.output