From 031f7d671edd05cad54b6aa9ab54a4e8103be163 Mon Sep 17 00:00:00 2001 From: Serhan Date: Sat, 11 Jul 2026 14:42:32 -0700 Subject: [PATCH] Fix nested checkup drift input resolution --- .pdd/meta/core_cli_python.json | 8 +- pdd/commands/drift.py | 46 +++- pdd/core/cli.py | 2 +- pdd/drift_main.py | 369 ++++++++++++++++++++++++--- pdd/prompts/core/cli_python.prompt | 2 +- tests/commands/test_drift_cli.py | 393 ++++++++++++++++++++++++++++- 6 files changed, 772 insertions(+), 48 deletions(-) diff --git a/.pdd/meta/core_cli_python.json b/.pdd/meta/core_cli_python.json index 0fc153705d..986c1fc353 100644 --- a/.pdd/meta/core_cli_python.json +++ b/.pdd/meta/core_cli_python.json @@ -1,9 +1,9 @@ { - "pdd_version": "0.0.301.dev0", - "timestamp": "2026-07-11T20:13:58.087488+00:00", + "pdd_version": "0.0.302.dev0", + "timestamp": "2026-07-11T21:34:57.998075+00:00", "command": "fix", - "prompt_hash": "38d36818bb40cab97145d9d92d484e1066c90d194e2b0311e8b1947338de38db", - "code_hash": "75eebbcab0f52fdeee132dbfe2760d1161738ea8509e0017c8f58d192bc78024", + "prompt_hash": "041ab8f4feba77b02ce03ce3e6063becb71d17ef1b32d4906eeca29e555aefc8", + "code_hash": "bd12418a3af2519ad7a365ae4f27588489d78e56af22845df697fbd758319160", "example_hash": null, "test_hash": null, "test_files": {}, diff --git a/pdd/commands/drift.py b/pdd/commands/drift.py index 202b6e7f39..582aa50eaa 100644 --- a/pdd/commands/drift.py +++ b/pdd/commands/drift.py @@ -1,4 +1,5 @@ """``pdd checkup drift`` subcommand implementation.""" + from __future__ import annotations import json @@ -7,12 +8,18 @@ import click -from ..drift_main import DEFAULT_MAX_COST_USD, run_drift +from ..drift_main import DEFAULT_MAX_COST_USD, DriftInputError, run_drift @click.command("drift") @click.argument("devunit") -@click.option("--runs", default=1, show_default=True, type=int, help="Number of regeneration runs.") +@click.option( + "--runs", + default=1, + show_default=True, + type=int, + help="Number of regeneration runs.", +) @click.option("--model", default=None, help="Model override for regeneration runs.") @click.option( "--from-evidence", @@ -32,7 +39,9 @@ default=False, help="Do not regenerate; compare current artifact stability only.", ) -@click.option("--json", "as_json", is_flag=True, default=False, help="Emit machine-readable JSON.") +@click.option( + "--json", "as_json", is_flag=True, default=False, help="Emit machine-readable JSON." +) @click.option( "--max-cost", "max_cost_usd", @@ -79,7 +88,34 @@ def drift_cmd( dry_run=dry_run, max_cost_usd=effective_max_cost, ) + except DriftInputError as exc: + if as_json: + click.echo( + json.dumps( + { + "status": "error", + "error": {"code": exc.code, "message": str(exc)}, + }, + indent=2, + ) + ) + raise click.exceptions.Exit(1) from exc + raise click.ClickException(str(exc)) from exc except (FileNotFoundError, RuntimeError) as exc: + if as_json: + click.echo( + json.dumps( + { + "status": "error", + "error": { + "code": "drift_runtime_error", + "message": str(exc), + }, + }, + indent=2, + ) + ) + raise click.exceptions.Exit(1) from exc raise click.ClickException(str(exc)) from exc if as_json: @@ -93,7 +129,9 @@ def drift_cmd( click.echo(f"Verify: {payload['verify']}") click.echo(f"Policy: {payload['policy']}") if payload.get("policy_check_unavailable"): - click.echo("Policy note: gate command unavailable; policy check failed closed.") + click.echo( + "Policy note: gate command unavailable; policy check failed closed." + ) if payload.get("max_cost_usd") is not None: click.echo(f"Max cost: ${payload['max_cost_usd']:.4f}") click.echo(f"Total cost: ${payload['total_cost_usd']:.4f}") diff --git a/pdd/core/cli.py b/pdd/core/cli.py index 25a6cefd8e..5d8d8dfc09 100644 --- a/pdd/core/cli.py +++ b/pdd/core/cli.py @@ -880,7 +880,7 @@ def _restore_estimate_env(_snapshot=_estimate_env_snapshot): pass # Warn users who have not completed interactive setup unless they are running it now - if not estimate_mode and _should_show_onboarding_reminder(ctx): + if not estimate_mode and not json_mode and _should_show_onboarding_reminder(ctx): console.print( "[warning]Complete onboarding with `pdd setup` to install tab completion and configure API keys.[/warning]" ) diff --git a/pdd/drift_main.py b/pdd/drift_main.py index af8dc2d094..14b5ea64e3 100644 --- a/pdd/drift_main.py +++ b/pdd/drift_main.py @@ -1,4 +1,5 @@ """Regeneration stability checks for PDD dev units (``pdd checkup drift``).""" + from __future__ import annotations import ast @@ -13,7 +14,7 @@ from pathlib import Path from typing import Any, Optional -from .evidence_store import ManifestView, resolve_prompt_path +from .evidence_store import ManifestView try: from .evidence_manifest import resolve_test_output_paths @@ -24,7 +25,10 @@ from .gate_main import run_gate_policy as _run_gate_policy_impl _GATE_POLICY_AVAILABLE = True -except (ModuleNotFoundError, ImportError): # pragma: no cover - gate optional on this branch +except ( + ModuleNotFoundError, + ImportError, +): # pragma: no cover - gate optional on this branch _GATE_POLICY_AVAILABLE = False def _run_gate_policy_impl(*_args, **_kwargs): @@ -33,6 +37,7 @@ class _Result: return _Result() + DEFAULT_MAX_COST_USD = 20.0 _COST_RE = re.compile(r"(?:Total\s+)?Cost:\s*\$([0-9]+(?:\.[0-9]+)?)", re.IGNORECASE) _POLICY_VALIDATION_KEYS = ("policy", "gate", "checkup_gate", "policy_gate") @@ -42,6 +47,14 @@ class _Result: ) +class DriftInputError(FileNotFoundError): + """A stable, user-actionable failure while resolving drift inputs.""" + + def __init__(self, code: str, message: str): + self.code = code + super().__init__(message) + + @dataclass class RunSnapshot: run_index: int @@ -141,8 +154,29 @@ def _public_api(path: Path) -> list[str]: return sorted(names) -def _resolve_code_path(prompt_path: Path, project_root: Path) -> Path: - stem = prompt_path.stem.replace("_python", "").replace("_typescript", "") +def _prompt_identity(prompt_path: Path) -> tuple[str, str]: + """Return ``(basename, language)`` from a PDD prompt filename.""" + if prompt_path.suffix.lower() != ".prompt": + raise DriftInputError( + "drift_input_invalid", + f"Prompt path must end in .prompt: {prompt_path}", + ) + basename, separator, language = prompt_path.stem.rpartition("_") + if not separator or not basename or not language: + raise DriftInputError( + "drift_input_invalid", + f"Prompt filename must use _.prompt: {prompt_path.name}", + ) + return basename, language.lower() + + +def _resolve_code_path( + prompt_path: Path, + project_root: Path, + *, + basename: Optional[str] = None, +) -> Path: + stem = basename or _prompt_identity(prompt_path)[0] candidates = [ project_root / "pdd" / f"{stem}.py", project_root / "src" / f"{stem}.py", @@ -156,6 +190,242 @@ def _resolve_code_path(prompt_path: Path, project_root: Path) -> Path: ) +def _resolve_project_file( + raw_path: str | Path, + project_root: Path, + *, + label: str, +) -> Path: + """Resolve an existing file without allowing symlink or ``..`` escape.""" + root = project_root.resolve() + path = Path(raw_path) + candidate = path if path.is_absolute() else root / path + try: + resolved = candidate.resolve() + resolved.relative_to(root) + except (OSError, RuntimeError, ValueError) as exc: + raise DriftInputError( + "drift_input_outside_project", + f"{label} must resolve inside project root {root}: {raw_path}", + ) from exc + if not resolved.is_file(): + raise DriftInputError( + "drift_input_not_found", + f"{label} not found: {raw_path}", + ) + return resolved + + +def _active_prompt_roots(project_root: Path) -> list[Path]: + """Return contained prompt roots declared by the active project config.""" + from .construct_paths import _find_pddrc_file, _load_pddrc_config + + root = project_root.resolve() + candidates = [root / "prompts"] + pddrc_path = _find_pddrc_file(root) + if pddrc_path: + try: + config = _load_pddrc_config(pddrc_path) + except ValueError as exc: + raise DriftInputError( + "drift_input_invalid", + f"Could not load active .pddrc: {exc}", + ) from exc + contexts = config.get("contexts", {}) + if isinstance(contexts, dict): + for context in contexts.values(): + if not isinstance(context, dict): + continue + defaults = context.get("defaults", {}) + if not isinstance(defaults, dict): + continue + configured = defaults.get("prompts_dir") + if isinstance(configured, str) and configured.strip(): + candidates.append(pddrc_path.parent / configured) + + contained: list[tuple[Path, Path]] = [] + for candidate in candidates: + try: + resolved = candidate.resolve() + resolved.relative_to(root) + except (OSError, RuntimeError, ValueError): + continue + if not resolved.is_dir(): + continue + contained.append((candidate, resolved)) + + # Scan each physical tree once. A canonical ``prompts/`` root commonly + # contains several context-specific roots such as ``prompts/services``. + roots: list[Path] = [] + resolved_roots: list[Path] = [] + for candidate, resolved in sorted(contained, key=lambda item: len(item[1].parts)): + if any(resolved.is_relative_to(parent) for parent in resolved_roots): + continue + resolved_roots.append(resolved) + roots.append(candidate) + return roots + + +def _path_shaped_devunit(devunit: str) -> bool: + return ( + Path(devunit).is_absolute() + or "/" in devunit + or "\\" in devunit + or devunit.lower().endswith(".prompt") + ) + + +def _normalized_devunit_parts(devunit: str) -> tuple[str, ...]: + if devunit != devunit.strip() or "\\" in devunit: + raise DriftInputError( + "drift_input_invalid", + f"Invalid dev unit name: {devunit!r}", + ) + parts = tuple(devunit.split("/")) + if not parts or any(part in {"", ".", ".."} for part in parts): + raise DriftInputError( + "drift_input_invalid", + f"Invalid dev unit name: {devunit!r}", + ) + return parts + + +def _candidate_matches_devunit( + candidate: Path, + prompt_root: Path, + devunit_parts: tuple[str, ...], +) -> bool: + try: + relative = candidate.relative_to(prompt_root) + except ValueError: + return False + basename, _language = _prompt_identity(candidate) + identity = (*relative.parent.parts, basename) + if len(devunit_parts) == 1: + return basename.casefold() == devunit_parts[0].casefold() + if len(identity) < len(devunit_parts): + return False + return tuple(part.casefold() for part in identity[-len(devunit_parts) :]) == tuple( + part.casefold() for part in devunit_parts + ) + + +def _owning_prompt_root(prompt_path: Path, roots: list[Path]) -> Path: + owners: list[Path] = [] + for root in roots: + try: + prompt_path.relative_to(root.resolve()) + except ValueError: + continue + owners.append(root) + if not owners: + return prompt_path.parent + return min(owners, key=lambda path: len(path.resolve().parts)) + + +def _resolve_prompt_input( + devunit: str, + project_root: Path, +) -> tuple[Path, Path]: + """Resolve explicit prompt paths or unambiguous basenames in active roots.""" + roots = _active_prompt_roots(project_root) + if _path_shaped_devunit(devunit) and devunit.lower().endswith(".prompt"): + prompt = _resolve_project_file(devunit, project_root, label="Prompt file") + _prompt_identity(prompt) + return prompt, _owning_prompt_root(prompt, roots) + if _path_shaped_devunit(devunit) and Path(devunit).is_absolute(): + _resolve_project_file(devunit, project_root, label="Prompt file") + raise DriftInputError( + "drift_input_invalid", + f"Prompt path must end in .prompt: {devunit}", + ) + + devunit_parts = _normalized_devunit_parts(devunit) + matches: dict[Path, tuple[Path, Path]] = {} + project_root_resolved = project_root.resolve() + for prompt_root in roots: + for candidate in prompt_root.rglob("*.prompt"): + if not candidate.is_file(): + continue + try: + resolved = candidate.resolve() + resolved.relative_to(project_root_resolved) + if not _candidate_matches_devunit( + candidate, prompt_root, devunit_parts + ): + continue + except (DriftInputError, OSError, RuntimeError, ValueError): + continue + matches.setdefault(resolved, (candidate, prompt_root)) + + if not matches: + raise DriftInputError( + "drift_input_not_found", + f"Could not resolve prompt for dev unit {devunit!r}", + ) + if len(matches) > 1: + choices = sorted( + candidate.relative_to(project_root).as_posix() + for candidate, _prompt_root in matches.values() + ) + rendered = ", ".join(choices) + raise DriftInputError( + "drift_input_ambiguous", + f"Ambiguous prompt for dev unit {devunit!r}; use an explicit path: {rendered}", + ) + resolved, (_candidate, prompt_root) = next(iter(matches.items())) + return resolved, prompt_root + + +def _derive_code_from_prompt( + prompt_path: Path, + prompt_root: Path, + project_root: Path, +) -> Path: + """Use the shared PDD path resolver, then retain the flat legacy fallback.""" + basename, language = _prompt_identity(prompt_path) + try: + relative_prompt = prompt_path.relative_to(prompt_root.resolve()) + except ValueError as exc: + raise DriftInputError( + "drift_input_outside_project", + f"Prompt file is not contained by its active prompt root: {prompt_path}", + ) from exc + qualified_basename = (relative_prompt.parent / basename).as_posix() + try: + from .sync_determine_operation import AmbiguousModuleError, get_pdd_file_paths + + paths = get_pdd_file_paths( + qualified_basename, + language, + prompts_dir=str(prompt_root), + ) + resolved_prompt = paths.get("prompt") + if ( + resolved_prompt is not None + and resolved_prompt.resolve() != prompt_path.resolve() + ): + raise DriftInputError( + "drift_input_ambiguous", + "Resolved code belongs to a different prompt; pass --code-file: " + f"{resolved_prompt}", + ) + resolved = paths.get("code") + if resolved is not None and resolved.is_file(): + try: + return _resolve_project_file(resolved, project_root, label="Code file") + except DriftInputError: + pass + except AmbiguousModuleError as exc: + raise DriftInputError("drift_input_ambiguous", str(exc)) from exc + if qualified_basename != basename: + raise DriftInputError( + "drift_input_not_found", + f"Could not locate generated code for {prompt_path.name}; pass --code-file", + ) + return _resolve_code_path(prompt_path, project_root, basename=basename) + + def _path_score(path: Path, *, expected_stem: str, devunit: str) -> int: score = 0 if path.is_file(): @@ -196,36 +466,64 @@ def _select_manifest_output( return None -def _load_manifest_paths( +def _resolve_drift_inputs( devunit: str, project_root: Path, from_evidence: Optional[Path], + code_file: Optional[Path], ) -> tuple[Path, Path, Optional[ManifestView]]: - if from_evidence is None: - latest = project_root / ".pdd" / "evidence" / "devunits" / f"{devunit}.latest.json" + """Resolve the authoritative prompt/code pair for one drift invocation.""" + root = project_root.resolve() + if from_evidence is None and not _path_shaped_devunit(devunit): + latest = ( + project_root / ".pdd" / "evidence" / "devunits" / f"{devunit}.latest.json" + ) if latest.is_file(): from_evidence = latest - if from_evidence is None or not from_evidence.is_file(): - prompt = resolve_prompt_path(project_root, devunit) - if prompt is None: - raise FileNotFoundError(f"Could not resolve prompt for dev unit {devunit!r}") - return prompt, _resolve_code_path(prompt, project_root), None - - manifest = ManifestView.from_file(from_evidence.resolve(), project_root) - prompt = manifest.prompt_path or resolve_prompt_path(project_root, devunit, manifest.raw) - if prompt is None: - raise FileNotFoundError(f"Evidence manifest missing prompt path: {from_evidence}") - - expected_stem = prompt.stem.replace("_python", "").replace("_typescript", "") - picked = _select_manifest_output( - manifest, - project_root=project_root, - expected_stem=expected_stem, - devunit=devunit, - ) - if picked is not None: - return prompt, picked, manifest - return prompt, _resolve_code_path(prompt, project_root), manifest + manifest = None + if from_evidence is not None: + if not from_evidence.is_file(): + raise DriftInputError( + "drift_input_not_found", + f"Evidence manifest not found: {from_evidence}", + ) + try: + manifest = ManifestView.from_file(from_evidence.resolve(), root) + except (OSError, TypeError, ValueError) as exc: + raise DriftInputError( + "drift_input_invalid", + f"Could not load evidence manifest {from_evidence}: {exc}", + ) from exc + + roots = _active_prompt_roots(root) + if manifest is not None and manifest.prompt_path is not None: + prompt = _resolve_project_file( + manifest.prompt_path, + root, + label="Evidence prompt file", + ) + prompt_root = _owning_prompt_root(prompt, roots) + else: + prompt, prompt_root = _resolve_prompt_input(devunit, root) + + # An explicit code path is authoritative. Do not derive or validate a + # manifest/default output first: that ordering was the core of issue #2001. + if code_file is not None: + code = _resolve_project_file(code_file, root, label="Code file") + return prompt, code, manifest + + if manifest is not None: + basename, _language = _prompt_identity(prompt) + picked = _select_manifest_output( + manifest, + project_root=root, + expected_stem=basename, + devunit=devunit, + ) + if picked is not None: + code = _resolve_project_file(picked, root, label="Evidence code file") + return prompt, code, manifest + return prompt, _derive_code_from_prompt(prompt, prompt_root, root), manifest def _parse_cost_usd(stdout: str, stderr: str) -> float: @@ -458,7 +756,9 @@ def _run_pytest_for_candidate( return completed.returncode == 0 -def _validation_key_configured(manifest: Optional[ManifestView], keys: tuple[str, ...]) -> bool: +def _validation_key_configured( + manifest: Optional[ManifestView], keys: tuple[str, ...] +) -> bool: if manifest is None: return False for key in keys: @@ -641,12 +941,12 @@ def run_drift( dry_run: bool = False, max_cost_usd: Optional[float] = None, ) -> DriftReport: - prompt_path, resolved_code, manifest = _load_manifest_paths( + prompt_path, code_path, manifest = _resolve_drift_inputs( devunit, project_root, from_evidence, + code_file, ) - code_path = code_file.resolve() if code_file else resolved_code if not code_path.is_file(): raise FileNotFoundError(f"Code file not found: {code_path}") @@ -729,9 +1029,10 @@ def run_drift( public_api_unchanged = bool(candidate_apis) and all( api == baseline_api for api in candidate_apis ) - implementation_changed = any(h != baseline_hash for h in candidate_hashes) or len( - set(candidate_hashes) - ) > 1 + implementation_changed = ( + any(h != baseline_hash for h in candidate_hashes) + or len(set(candidate_hashes)) > 1 + ) behavior_unchanged = all( snap.tests_passed and snap.stories_passed diff --git a/pdd/prompts/core/cli_python.prompt b/pdd/prompts/core/cli_python.prompt index fae6cb423c..e26efae222 100644 --- a/pdd/prompts/core/cli_python.prompt +++ b/pdd/prompts/core/cli_python.prompt @@ -56,7 +56,7 @@ - When `--review-examples` is supplied, set `ctx.obj["review_examples"] = True` AND initialize `ctx.obj["grounding_review_decisions"] = []` so downstream grounding consumers can append per-module review decisions (`{"module": str, "decision": "accept"|"reject", "reason": Optional[str]}`) for the evidence-manifest writer to record under `generation.grounding.reviewed`. When `--review-examples` is not supplied, do NOT pre-create the key — the manifest writer treats its absence as "no review observed" and records `reviewed: false`. - Only set `strength`/`temperature` in `ctx.obj` if explicitly provided. - Early Actions: Handle `--list-contexts` and `--context` validation (against `list_available_contexts()`). Run `auto_update()` unless `PDD_AUTO_UPDATE=false`. - - Show onboarding reminder via `_should_show_onboarding_reminder`. + - Show onboarding reminder via `_should_show_onboarding_reminder` for human-output invocations only; suppress it for machine JSON mode so stdout remains one parseable JSON document. - Output Capture: If `--core-dump` enabled, wrap `sys.stdout/stderr` with `OutputCapture` and store in `ctx.obj`. 4. **Result Callback (`process_commands`)**: diff --git a/tests/commands/test_drift_cli.py b/tests/commands/test_drift_cli.py index 429ee4e671..01326613e6 100644 --- a/tests/commands/test_drift_cli.py +++ b/tests/commands/test_drift_cli.py @@ -1,4 +1,5 @@ """CLI smoke tests for ``pdd checkup drift`` (PR #1261 manual test plan).""" + from __future__ import annotations import hashlib @@ -49,6 +50,35 @@ def _write_smoke_project(project: Path) -> None: ) +def _write_nested_project( + project: Path, + *, + prompt_root: str = "prompts/src", + prompt_name: str = "widget_Python.prompt", + code_path: str = "generated/widget_impl.py", +) -> tuple[Path, Path]: + prompt = project / prompt_root / prompt_name + prompt.parent.mkdir(parents=True, exist_ok=True) + prompt.write_text("\nNested widget module.\n\n", encoding="utf-8") + code = project / code_path + code.parent.mkdir(parents=True, exist_ok=True) + code.write_text("def widget() -> str:\n return 'ok'\n", encoding="utf-8") + (project / ".pddrc").write_text( + "version: '1.0'\n" + "contexts:\n" + " nested:\n" + " paths:\n" + f" - '{prompt_root}/**'\n" + f" - '{Path(code_path).parent.as_posix()}/**'\n" + " defaults:\n" + f" prompts_dir: '{prompt_root}'\n" + f" generate_output_path: '{Path(code_path).parent.as_posix()}'\n" + " default_language: python\n", + encoding="utf-8", + ) + return prompt, code + + @pytest.fixture def runner() -> CliRunner: return CliRunner() @@ -62,7 +92,9 @@ def test_checkup_drift_help(runner: CliRunner) -> None: assert "--json" in result.output -def test_checkup_drift_dry_run_json(runner: CliRunner, tmp_path: Path, monkeypatch) -> None: +def test_checkup_drift_dry_run_json( + runner: CliRunner, tmp_path: Path, monkeypatch +) -> None: _write_smoke_project(tmp_path) monkeypatch.chdir(tmp_path) result = runner.invoke( @@ -76,10 +108,14 @@ def test_checkup_drift_dry_run_json(runner: CliRunner, tmp_path: Path, monkeypat assert payload["dry_run"] is True -def test_checkup_drift_from_evidence_json(runner: CliRunner, tmp_path: Path, monkeypatch) -> None: +def test_checkup_drift_from_evidence_json( + runner: CliRunner, tmp_path: Path, monkeypatch +) -> None: _write_smoke_project(tmp_path) monkeypatch.chdir(tmp_path) - manifest = tmp_path / ".pdd" / "evidence" / "devunits" / "refund_payment.latest.json" + manifest = ( + tmp_path / ".pdd" / "evidence" / "devunits" / "refund_payment.latest.json" + ) result = runner.invoke( checkup, [ @@ -153,7 +189,7 @@ def _fake_drift_main(args, **kwargs): return 0 with patch("pdd.commands.checkup.drift_cmd.main", side_effect=_fake_drift_main): - result = runner.invoke( + runner.invoke( checkup, ["drift", "refund_payment", "--preview"], catch_exceptions=False, @@ -179,3 +215,352 @@ def test_explicit_dry_run_still_forwarded_to_drift( assert result.exit_code == 0, result.output payload = json.loads(result.output) assert payload["dry_run"] is True + + +def test_checkup_drift_accepts_explicit_nested_prompt_and_authoritative_code_file( + runner: CliRunner, tmp_path: Path, monkeypatch +) -> None: + prompt, code = _write_nested_project(tmp_path) + monkeypatch.chdir(tmp_path) + + result = runner.invoke( + checkup, + [ + "drift", + prompt.relative_to(tmp_path).as_posix(), + "--code-file", + code.relative_to(tmp_path).as_posix(), + "--dry-run", + "--json", + ], + catch_exceptions=False, + ) + + assert result.exit_code == 0, result.output + payload = json.loads(result.output) + assert Path(payload["prompt_path"]) == prompt.resolve() + assert Path(payload["code_path"]) == code.resolve() + assert payload["status"] == "stable" + + +def test_checkup_drift_resolves_nested_basename_from_active_pddrc( + runner: CliRunner, tmp_path: Path, monkeypatch +) -> None: + prompt, code = _write_nested_project( + tmp_path, + prompt_root="specs/backend", + prompt_name="widget_Python.prompt", + ) + monkeypatch.chdir(tmp_path) + + result = runner.invoke( + checkup, + [ + "drift", + "widget", + "--code-file", + code.relative_to(tmp_path).as_posix(), + "--dry-run", + "--json", + ], + catch_exceptions=False, + ) + + assert result.exit_code == 0, result.output + payload = json.loads(result.output) + assert Path(payload["prompt_path"]) == prompt.resolve() + assert Path(payload["code_path"]) == code.resolve() + + +def test_checkup_drift_explicit_nested_prompt_derives_own_sibling_context_code( + runner: CliRunner, tmp_path: Path, monkeypatch +) -> None: + """A prompt path must retain its context when ``--code-file`` is omitted.""" + for context in ("backend", "frontend"): + prompt = tmp_path / "prompts" / context / "widget_Python.prompt" + prompt.parent.mkdir(parents=True) + prompt.write_text(f"{context} widget\n", encoding="utf-8") + code = tmp_path / "src" / context / "widget.py" + code.parent.mkdir(parents=True) + code.write_text( + f"def widget() -> str:\n return '{context}'\n", + encoding="utf-8", + ) + (tmp_path / ".pddrc").write_text( + "version: '1.0'\n" + "contexts:\n" + " backend:\n" + " paths:\n" + " - 'backend/**'\n" + " - 'src/backend/**'\n" + " - 'prompts/backend/**'\n" + " defaults:\n" + " prompts_dir: prompts/backend\n" + " generate_output_path: src/backend\n" + " default_language: python\n" + " frontend:\n" + " paths:\n" + " - 'frontend/**'\n" + " - 'src/frontend/**'\n" + " - 'prompts/frontend/**'\n" + " defaults:\n" + " prompts_dir: prompts/frontend\n" + " generate_output_path: src/frontend\n" + " default_language: python\n", + encoding="utf-8", + ) + monkeypatch.chdir(tmp_path) + + result = runner.invoke( + checkup, + [ + "drift", + "prompts/frontend/widget_Python.prompt", + "--dry-run", + "--json", + ], + catch_exceptions=False, + ) + + assert result.exit_code == 0, result.output + payload = json.loads(result.output) + assert ( + Path(payload["prompt_path"]) + == (tmp_path / "prompts/frontend/widget_Python.prompt").resolve() + ) + assert Path(payload["code_path"]) == (tmp_path / "src/frontend/widget.py").resolve() + + +def test_checkup_drift_nested_prompt_does_not_fall_back_to_unrelated_flat_code( + runner: CliRunner, tmp_path: Path, monkeypatch +) -> None: + prompt = tmp_path / "prompts/frontend/widget_Python.prompt" + prompt.parent.mkdir(parents=True) + prompt.write_text("frontend widget\n", encoding="utf-8") + unrelated = tmp_path / "src/widget.py" + unrelated.parent.mkdir() + unrelated.write_text("def widget() -> str:\n return 'root'\n", encoding="utf-8") + (tmp_path / ".pddrc").write_text( + "version: '1.0'\n" + "contexts:\n" + " frontend:\n" + " paths:\n" + " - 'frontend/**'\n" + " - 'prompts/frontend/**'\n" + " defaults:\n" + " prompts_dir: prompts/frontend\n" + " generate_output_path: src/frontend\n" + " default_language: python\n", + encoding="utf-8", + ) + monkeypatch.chdir(tmp_path) + + result = runner.invoke( + checkup, + [ + "drift", + "prompts/frontend/widget_Python.prompt", + "--dry-run", + "--json", + ], + catch_exceptions=False, + ) + + assert result.exit_code == 1 + payload = json.loads(result.output) + assert payload["error"]["code"] == "drift_input_not_found" + assert "pass --code-file" in payload["error"]["message"] + + +def test_checkup_drift_matches_language_suffix_case_insensitively( + runner: CliRunner, tmp_path: Path, monkeypatch +) -> None: + prompt, code = _write_nested_project( + tmp_path, + prompt_name="widget_PyThOn.prompt", + ) + monkeypatch.chdir(tmp_path) + + result = runner.invoke( + checkup, + [ + "drift", + "widget", + "--code-file", + code.relative_to(tmp_path).as_posix(), + "--dry-run", + "--json", + ], + catch_exceptions=False, + ) + + assert result.exit_code == 0, result.output + payload = json.loads(result.output) + assert Path(payload["prompt_path"]) == prompt.resolve() + + +def test_checkup_drift_json_resolution_failure_is_structured( + runner: CliRunner, tmp_path: Path, monkeypatch +) -> None: + monkeypatch.chdir(tmp_path) + + result = runner.invoke( + checkup, + ["drift", "missing_widget", "--dry-run", "--json"], + catch_exceptions=False, + ) + + assert result.exit_code == 1 + payload = json.loads(result.output) + assert payload == { + "status": "error", + "error": { + "code": "drift_input_not_found", + "message": "Could not resolve prompt for dev unit 'missing_widget'", + }, + } + + +def test_checkup_drift_rejects_explicit_paths_outside_project_root( + runner: CliRunner, tmp_path: Path, monkeypatch +) -> None: + project = tmp_path / "project" + project.mkdir() + outside_prompt = tmp_path / "outside_Python.prompt" + outside_prompt.write_text("outside\n", encoding="utf-8") + code = project / "src" / "widget.py" + code.parent.mkdir() + code.write_text("def widget() -> None:\n pass\n", encoding="utf-8") + monkeypatch.chdir(project) + + result = runner.invoke( + checkup, + [ + "drift", + str(outside_prompt), + "--code-file", + str(code), + "--dry-run", + "--json", + ], + catch_exceptions=False, + ) + + assert result.exit_code == 1 + payload = json.loads(result.output) + assert payload["status"] == "error" + assert payload["error"]["code"] == "drift_input_outside_project" + + +def test_checkup_drift_rejects_explicit_code_outside_project_root( + runner: CliRunner, tmp_path: Path, monkeypatch +) -> None: + project = tmp_path / "project" + prompt = project / "prompts" / "widget_python.prompt" + prompt.parent.mkdir(parents=True) + prompt.write_text("widget\n", encoding="utf-8") + outside_code = tmp_path / "widget.py" + outside_code.write_text("def widget() -> None:\n pass\n", encoding="utf-8") + monkeypatch.chdir(project) + + result = runner.invoke( + checkup, + [ + "drift", + "prompts/widget_python.prompt", + "--code-file", + str(outside_code), + "--dry-run", + "--json", + ], + catch_exceptions=False, + ) + + assert result.exit_code == 1 + payload = json.loads(result.output) + assert payload["error"]["code"] == "drift_input_outside_project" + + +def test_checkup_drift_rejects_ambiguous_nested_basename( + runner: CliRunner, tmp_path: Path, monkeypatch +) -> None: + code = tmp_path / "src" / "widget.py" + code.parent.mkdir(parents=True) + code.write_text("def widget() -> None:\n pass\n", encoding="utf-8") + for prompt_dir in ("specs/backend", "specs/frontend"): + prompt = tmp_path / prompt_dir / "widget_python.prompt" + prompt.parent.mkdir(parents=True) + prompt.write_text("widget\n", encoding="utf-8") + (tmp_path / ".pddrc").write_text( + "version: '1.0'\n" + "contexts:\n" + " backend:\n" + " defaults:\n" + " prompts_dir: specs/backend\n" + " default_language: python\n" + " frontend:\n" + " defaults:\n" + " prompts_dir: specs/frontend\n" + " default_language: python\n", + encoding="utf-8", + ) + monkeypatch.chdir(tmp_path) + + result = runner.invoke( + checkup, + [ + "drift", + "widget", + "--code-file", + "src/widget.py", + "--dry-run", + "--json", + ], + catch_exceptions=False, + ) + + assert result.exit_code == 1 + payload = json.loads(result.output) + assert payload["error"]["code"] == "drift_input_ambiguous" + assert "specs/backend/widget_python.prompt" in payload["error"]["message"] + assert "specs/frontend/widget_python.prompt" in payload["error"]["message"] + + +def test_checkup_drift_flat_prompt_control_without_manifest( + runner: CliRunner, tmp_path: Path, monkeypatch +) -> None: + prompt = tmp_path / "prompts" / "widget_python.prompt" + prompt.parent.mkdir() + prompt.write_text("widget\n", encoding="utf-8") + code = tmp_path / "src" / "widget.py" + code.parent.mkdir() + code.write_text("def widget() -> str:\n return 'ok'\n", encoding="utf-8") + monkeypatch.chdir(tmp_path) + + result = runner.invoke( + checkup, + ["drift", "widget", "--code-file", "src/widget.py", "--dry-run", "--json"], + catch_exceptions=False, + ) + + assert result.exit_code == 0, result.output + payload = json.loads(result.output) + assert Path(payload["prompt_path"]) == prompt.resolve() + assert Path(payload["code_path"]) == code.resolve() + + +def test_root_cli_preserves_structured_drift_json_failure( + runner: CliRunner, tmp_path: Path, monkeypatch +) -> None: + monkeypatch.chdir(tmp_path) + + result = runner.invoke( + cli, + ["checkup", "drift", "missing_widget", "--dry-run", "--json"], + catch_exceptions=False, + ) + + assert result.exit_code == 1 + payload = json.loads(result.output) + assert payload["status"] == "error" + assert payload["error"]["code"] == "drift_input_not_found"