diff --git a/pdd/continuous_sync.py b/pdd/continuous_sync.py index 82e7c95584..0991012bee 100644 --- a/pdd/continuous_sync.py +++ b/pdd/continuous_sync.py @@ -1,31 +1,52 @@ """Deterministic continuous-sync classification and reconciliation reports.""" +# pylint: disable=too-many-lines from __future__ import annotations import json import os +import stat import subprocess from dataclasses import dataclass from datetime import datetime, timezone from functools import lru_cache -from itertools import combinations from pathlib import Path, PurePosixPath from typing import Any, Dict, Iterable, List, Optional from .operation_log import ( _safe_basename, - get_fingerprint_path, - infer_module_identity, ) from .sync_determine_operation import ( Fingerprint, + _architecture_artifact_paths, + _expand_output_templates, + _module_filepath_matches_basename, + _relative_basename_for_context, + _resolve_context_name_for_basename, calculate_current_hashes, calculate_sha256, - get_pdd_file_paths, - read_fingerprint, + get_extension, ) +from .architecture_registry import extract_modules from .sync_core import CanonicalReportOptions, build_canonical_report - - +from .construct_paths import _load_pddrc_config + + +MAX_PROMPT_DISCOVERY_FILES = 10000 +MAX_PROMPT_DISCOVERY_ENTRIES = 50000 +MAX_PROMPT_DISCOVERY_ROOTS = 100 +MAX_CONFIG_CONTEXTS = 1000 +MAX_METADATA_INFERENCE_CANDIDATES = 32 +MAX_REPAIR_DISCOVERY_ENTRIES = 50000 +SKIPPED_DISCOVERY_DIRS = { + ".git", + ".hg", + ".svn", + ".pdd", + "__pycache__", + "node_modules", + ".venv", + "venv", +} DRIFT_CLASSIFICATIONS = { "DOC_CHANGED", "PROMPT_CHANGED", @@ -47,6 +68,18 @@ class SyncUnit: language: str prompt_path: Path prompts_dir: Path + context_name: str | None = None + pddrc_path: Path | None = None + config: Dict[str, Any] | None = None + + +@dataclass(frozen=True) +class DiscoveryFailure: + """A prompt discovery failure that must be visible in JSON output.""" + + prompt_root: Path + reason: str + failure: str def project_root(start: Optional[Path] = None) -> Path: @@ -238,21 +271,290 @@ def _prompts_dir_for(prompt_path: Path) -> Path: return prompt_path.parent -def _prompt_units(prompt_root: Path) -> List[SyncUnit]: +def _is_hidden_or_system_dir(path: Path) -> bool: + return path.name.startswith(".") or path.name in SKIPPED_DISCOVERY_DIRS + + +def _bounded_prompt_paths( + prompt_root: Path, budget: Dict[str, int] +) -> tuple[List[Path], Optional[DiscoveryFailure]]: + prompt_paths: List[Path] = [] + pending = [prompt_root] + while pending: + current = pending.pop() + try: + entries = os.scandir(current) + with entries: + for entry in entries: + budget["entries"] += 1 + if budget["entries"] > MAX_PROMPT_DISCOVERY_ENTRIES: + return prompt_paths, DiscoveryFailure( + prompt_root=prompt_root, + reason=( + "configured prompt root exceeded traversal budget " + f"of {MAX_PROMPT_DISCOVERY_ENTRIES} directory entries" + ), + failure="prompt_traversal_budget", + ) + entry_path = current / entry.name + try: + is_directory = entry.is_dir(follow_symlinks=False) + except OSError as exc: + raise OSError(f"cannot inspect {entry_path}: {exc}") from exc + if is_directory: + if not _is_hidden_or_system_dir(entry_path): + pending.append(entry_path) + continue + if not entry.name.endswith(".prompt") or "_" not in entry.name: + continue + prompt_paths.append(entry_path) + budget["files"] += 1 + if budget["files"] > MAX_PROMPT_DISCOVERY_FILES: + return prompt_paths, DiscoveryFailure( + prompt_root=prompt_root, + reason=( + "configured prompt root exceeded traversal budget " + f"of {MAX_PROMPT_DISCOVERY_FILES} prompt files" + ), + failure="prompt_traversal_budget", + ) + except OSError as exc: + return prompt_paths, DiscoveryFailure( + prompt_root=prompt_root, + reason=f"configured prompt root traversal failed: {exc}", + failure="prompt_traversal_error", + ) + return sorted(prompt_paths), None + + +def _prompt_units( + prompt_root: Path, budget: Dict[str, int], identity_roots: List[Path], base: Path, + config_cache: Dict[Path, Dict[str, Any]], +) -> tuple[List[SyncUnit], List[DiscoveryFailure]]: units: List[SyncUnit] = [] - for prompt_path in sorted(prompt_root.rglob("*_*.prompt")): - basename, language = infer_module_identity(prompt_path) - if basename is None or language is None: + failures: List[DiscoveryFailure] = [] + prompt_paths, failure = _bounded_prompt_paths(prompt_root, budget) + if failure is not None: + failures.append(failure) + for prompt_path in prompt_paths[:MAX_PROMPT_DISCOVERY_FILES]: + nested_configs = [ + parent / ".pddrc" + for parent in (prompt_path.parent, *prompt_path.parents) + if _is_within_root(parent, base) and parent != base + ] + unsafe = next( + ( + path + for path in nested_configs + if (path.is_symlink() or path.exists()) + and not _safe_control_file(path, base) + ), + None, + ) + if unsafe is not None: + failures.append(DiscoveryFailure(unsafe, f"unsafe nested config: {unsafe}", "unsafe_config")) + continue + owner = max( + (root for root in identity_roots if _is_within_root(prompt_path, root)), + key=lambda root: len(root.parts), + default=prompt_root, + ) + stem, separator, language = prompt_path.stem.rpartition("_") + if not separator or not stem or not language: + continue + try: + basename, context_name, pddrc_path, owner = _prompt_ownership( + prompt_path, stem, owner, base, config_cache + ) + except (OSError, RuntimeError, ValueError) as exc: + failures.append( + DiscoveryFailure( + prompt_path, + f"invalid owning config: {exc}", + "invalid_pddrc", + ) + ) continue + language = language.lower() units.append( SyncUnit( basename=basename, language=language, prompt_path=prompt_path, - prompts_dir=_prompts_dir_for(prompt_path), + prompts_dir=owner, + context_name=context_name, + pddrc_path=pddrc_path, + config=config_cache.get(pddrc_path) if pddrc_path else None, ) ) - return units + return units, failures + + +def _is_safe_prompt_root(base: Path, prompt_root: Path) -> bool: + """Return whether a configured prompt root is within the project boundary.""" + try: + prompt_root.resolve(strict=False).relative_to(base.resolve(strict=False)) + except ValueError: + return False + return True + + +def _safe_control_file(path: Path, base: Path) -> bool: + """Validate a candidate-controlled file path without following links.""" + return _safe_architecture_candidate(path, base) + + +def _project_config(base: Path) -> tuple[Path | None, Dict[str, Any]]: + """Load the project config only after validating its logical path.""" + path = base / ".pddrc" + if not _safe_control_file(path, base): + raise UnsafeConfigError(f"unsafe project config: {path}") + try: + mode = path.lstat().st_mode + except FileNotFoundError: + return None, {} + if not stat.S_ISREG(mode): + raise UnsafeConfigError(f"unsafe project config: {path}") + return path, _load_pddrc_config(path) + + +def _prompt_ownership( + prompt_path: Path, + stem: str, + fallback_root: Path, + base: Path, + config_cache: Dict[Path, Dict[str, Any]] | None = None, +) -> tuple[str, str | None, Path | None, Path]: + """Return basename and nearest validated config/context ownership.""" + config_candidates = [ + parent / ".pddrc" + for parent in (prompt_path.parent, *prompt_path.parents) + if _is_within_root(parent, base) + ] + for pddrc_path in config_candidates: + if not pddrc_path.exists(): + continue + if not _safe_control_file(pddrc_path, base): + raise UnsafeConfigError(f"unsafe nested config: {pddrc_path}") + if config_cache is not None and pddrc_path in config_cache: + config = config_cache[pddrc_path] + else: + config = _load_pddrc_config(pddrc_path) + if config_cache is not None: + config_cache[pddrc_path] = config + contexts = _validate_pddrc_structure(config) + owned: list[tuple[int, str, Path]] = [] + if isinstance(contexts, dict): + for context_name, context in contexts.items(): + defaults = context.get("defaults", {}) + raw_root = defaults.get("prompts_dir", "prompts") + prompt_root = Path(raw_root).expanduser() + if not prompt_root.is_absolute(): + prompt_root = pddrc_path.parent / prompt_root + prompt_root = prompt_root.resolve(strict=False) + if _is_safe_prompt_root(base, prompt_root) and _is_within_root( + prompt_path, prompt_root + ): + owned.append((len(prompt_root.parts), context_name, prompt_root)) + if owned: + _depth, context_name, prompt_root = max(owned) + parent = prompt_path.parent.relative_to(prompt_root) + basename = (parent / stem).as_posix() if parent.parts else stem + return basename, context_name, pddrc_path, prompt_root + parent = prompt_path.parent.relative_to(fallback_root) + basename = (parent / stem).as_posix() if parent.parts else stem + return basename, None, None, fallback_root + + +def _configured_prompt_roots( + base: Path, pddrc_path: Path | None, config: Dict[str, Any] +) -> List[Path]: + # pylint: disable=too-many-branches + """Return normalized prompt roots configured for the legacy report. + + The report owns discovery, so configured roots must be resolved before it + passes a fixed relative pattern to ``Path.rglob``. In particular, an + absolute ``prompts_dir`` is a root, never a glob pattern to append to one. + """ + configured: List[Path] = [] + if pddrc_path is not None: + contexts = _validate_pddrc_structure(config) + for context in contexts.values(): + defaults = context.get("defaults", {}) + raw_root = defaults.get("prompts_dir", "prompts") + root = Path(raw_root).expanduser() + if not root.is_absolute(): + root = pddrc_path.parent / root + configured.append(root.resolve(strict=False)) + + if not configured: + configured.append((base / "prompts").resolve(strict=False)) + unique = sorted(dict.fromkeys(configured), key=lambda item: len(item.parts)) + collapsed: List[Path] = [] + for candidate in unique: + collapsed.append(candidate) + if len(collapsed) > MAX_PROMPT_DISCOVERY_ROOTS: + raise ValueError( + f".pddrc exceeds configured prompt root limit of {MAX_PROMPT_DISCOVERY_ROOTS}" + ) + return collapsed + + +def _validate_pddrc_structure(config: Any) -> Dict[str, Any]: + """Return strictly validated contexts shared by root and nested configs.""" + if not isinstance(config, dict): + raise ValueError(".pddrc must contain a mapping") + contexts = config.get("contexts", {}) + if not isinstance(contexts, dict): + raise ValueError(".pddrc contexts must contain a mapping") + if len(contexts) > MAX_CONFIG_CONTEXTS: + raise ValueError( + f".pddrc exceeds context limit of {MAX_CONFIG_CONTEXTS}" + ) + for context_name, context in contexts.items(): + if not isinstance(context_name, str) or not isinstance(context, dict): + raise ValueError(".pddrc context must contain a mapping") + defaults = context.get("defaults", {}) + if not isinstance(defaults, dict): + raise ValueError(".pddrc defaults must contain a mapping") + raw_root = defaults.get("prompts_dir", "prompts") + if not isinstance(raw_root, str) or not raw_root.strip(): + raise ValueError(".pddrc prompts_dir must be a non-empty string") + return contexts + + +def _validated_prompt_roots( + base: Path, config_cache: Dict[Path, Dict[str, Any]] | None = None +) -> tuple[List[Path], List[DiscoveryFailure]]: + roots: List[Path] = [] + failures: List[DiscoveryFailure] = [] + try: + pddrc_path, config = _project_config(base) + if pddrc_path is not None and config_cache is not None: + config_cache[pddrc_path] = config + configured_roots = _configured_prompt_roots(base, pddrc_path, config) + except UnsafeConfigError as exc: + return [], [DiscoveryFailure(base / ".pddrc", str(exc), "unsafe_config")] + except (OSError, RuntimeError, ValueError) as exc: + return [], [ + DiscoveryFailure( + prompt_root=base / ".pddrc", + reason=f"invalid .pddrc: {exc}", + failure="invalid_pddrc", + ) + ] + for prompt_root in configured_roots: + if _is_safe_prompt_root(base, prompt_root): + roots.append(prompt_root) + continue + failures.append( + DiscoveryFailure( + prompt_root=prompt_root, + reason="configured prompt root is outside project", + failure="unsafe_prompt_root", + ) + ) + return roots, failures def _matches_module(unit: SyncUnit, wanted: set[str]) -> bool: @@ -315,34 +617,46 @@ def _slash_candidates(safe_basename: str) -> List[str]: if len(parts) <= 1: return [] - candidates: List[str] = [] - separators = range(1, len(parts)) - for count in range(1, len(parts)): - for slash_positions in combinations(separators, count): - chunks: List[str] = [parts[0]] - for index, part in enumerate(parts[1:], start=1): - if index in slash_positions: - chunks.append("/") - else: - chunks.append("_") - chunks.append(part) - candidates.append("".join(chunks)) - return candidates + candidates = [safe_basename.replace("_", "/")] + for index in range(1, len(parts)): + candidates.append("_".join(parts[:index]) + "/" + "_".join(parts[index:])) + return list(dict.fromkeys(candidates))[:MAX_METADATA_INFERENCE_CANDIDATES] + + +def _validated_artifact_exists(path: Path, base: Path) -> bool: + """Stat an inferred artifact only after logical containment/link validation.""" + return _safe_architecture_candidate(path, base) and path.exists() def _existing_artifact_score( basename: str, language: str, prompt_root: Path, + pddrc_path: Path | None = None, + config: Dict[str, Any] | None = None, + budget: Optional[Dict[str, int]] = None, ) -> int: try: - paths = get_pdd_file_paths(basename, language, prompts_dir=str(prompt_root)) - except Exception: + prompt_path = prompt_root / f"{Path(basename).name}_{language}.prompt" + paths = _resolve_report_paths( + SyncUnit( + basename, + language, + prompt_path, + prompt_root, + pddrc_path=pddrc_path, + config=config, + ), + project_root(prompt_root), + budget, + ) + except ValueError: return 0 + base = project_root(prompt_root) score = 0 for artifact in ("code", "example", "test"): path = paths.get(artifact) - if path is not None and path.exists(): + if path is not None and _validated_artifact_exists(path, base): score += 1 return score @@ -351,11 +665,21 @@ def _infer_basename_from_artifacts( safe_basename: str, language: str, prompt_root: Path, + budget: Dict[str, int], + pddrc_path: Path | None = None, + config: Dict[str, Any] | None = None, ) -> str: best = safe_basename - best_score = _existing_artifact_score(best, language, prompt_root) + best_score = _existing_artifact_score( + best, language, prompt_root, pddrc_path, config, budget + ) for candidate in _slash_candidates(safe_basename): - score = _existing_artifact_score(candidate, language, prompt_root) + if budget["entries"] >= MAX_PROMPT_DISCOVERY_ENTRIES: + break + budget["entries"] += 1 + score = _existing_artifact_score( + candidate, language, prompt_root, pddrc_path, config, budget + ) if score > best_score: best = candidate best_score = score @@ -367,6 +691,9 @@ def _unit_from_metadata_identity( prompt_root: Path, prompt_index: Dict[tuple[str, str], SyncUnit], requested_basename: Optional[str] = None, + budget: Optional[Dict[str, int]] = None, + pddrc_path: Path | None = None, + config: Dict[str, Any] | None = None, ) -> SyncUnit: safe_basename, language = identity prompt_unit = prompt_index.get((safe_basename, language)) @@ -377,6 +704,9 @@ def _unit_from_metadata_identity( safe_basename, language, prompt_root, + budget or {"entries": 0, "files": 0}, + pddrc_path, + config, ) prompt_path = _prompt_path_for_basename(prompt_root, basename, language) return SyncUnit( @@ -384,21 +714,93 @@ def _unit_from_metadata_identity( language=language, prompt_path=prompt_path, prompts_dir=prompt_root, + pddrc_path=pddrc_path, + config=config, ) -def _metadata_identities(meta_root: Path) -> List[tuple[str, str]]: +def _path_component_violation(path: Path, root: Path, leaf_directory: bool) -> Optional[str]: + # pylint: disable=too-many-return-statements + """Return why a project-contained path is unsafe without following symlinks.""" + root_path = root.resolve() + try: + parts = path.relative_to(root_path).parts + except ValueError: + return "outside_project" + cursor = root_path + for index, part in enumerate(parts): + cursor /= part + try: + mode = cursor.lstat().st_mode + except FileNotFoundError: + return "missing" + except OSError: + return "invalid" + if stat.S_ISLNK(mode): + return "symlink" + is_leaf = index == len(parts) - 1 + if is_leaf: + expected = stat.S_ISDIR(mode) if leaf_directory else stat.S_ISREG(mode) + else: + expected = stat.S_ISDIR(mode) + if not expected: + return "nonregular" + try: + path.resolve(strict=True).relative_to(root_path) + except (OSError, ValueError): + return "outside_project" + return None + + +def _metadata_identities( + meta_root: Path, + base: Path, + budget: Dict[str, int], +) -> tuple[List[tuple[str, str]], Optional[DiscoveryFailure]]: identities: List[tuple[str, str]] = [] seen: set[tuple[str, str]] = set() - if not meta_root.exists(): - return identities - for path in sorted(meta_root.glob("*_*.json")): + violation = _path_component_violation(meta_root, base, leaf_directory=True) + if violation == "missing": + return identities, None + if violation is not None: + return [], DiscoveryFailure( + prompt_root=meta_root, + reason=f"unsafe metadata directory: {violation}", + failure="unsafe_metadata", + ) + remaining = max(0, MAX_PROMPT_DISCOVERY_ENTRIES - budget["entries"]) + try: + paths = [] + with os.scandir(meta_root) as entries: + for entry in entries: + paths.append(meta_root / entry.name) + if len(paths) > remaining: + break + except OSError as exc: + return [], DiscoveryFailure(meta_root, f"metadata traversal failed: {exc}", "metadata_traversal_error") + exhausted = len(paths) > remaining + paths = sorted(paths[:remaining]) + for path in paths: + budget["entries"] += 1 + if budget["entries"] > MAX_PROMPT_DISCOVERY_ENTRIES: + return identities, DiscoveryFailure(meta_root, "command discovery entry budget exhausted", "discovery_budget") + if not path.name.endswith(".json") or "_" not in path.stem: + continue identity = _metadata_identity(path) if identity is None or identity in seen: continue seen.add(identity) identities.append(identity) - return identities + budget["files"] += 1 + if budget["files"] > MAX_PROMPT_DISCOVERY_FILES: + return identities[:-1], DiscoveryFailure(meta_root, "command discovery unit budget exhausted", "discovery_budget") + if exhausted: + return identities, DiscoveryFailure( + meta_root, + "command discovery entry budget exhausted", + "discovery_budget", + ) + return identities, None def discover_units( @@ -406,65 +808,147 @@ def discover_units( modules: Optional[Iterable[str]] = None, ) -> List[SyncUnit]: """Discover prompt-backed units under ``root``.""" - base = project_root(root) - wanted = set(modules or []) - prompt_root = base / "prompts" - prompt_units = _prompt_units(prompt_root) if prompt_root.exists() else [] - meta_root = base / ".pdd" / "meta" - metadata_identities = _metadata_identities(meta_root) + units, _failures = _discover_units_and_failures(root, modules) + return units - prompt_index: Dict[tuple[str, str], SyncUnit] = {} - for unit in prompt_units: - prompt_index.setdefault((_safe_basename(unit.basename), unit.language), unit) - if wanted: - units: List[SyncUnit] = [] - seen: set[tuple[str, str, Path]] = set() - for identity in metadata_identities: - if not _identity_matches_wanted(identity, wanted): - continue - unit = _unit_from_metadata_identity( - identity, - prompt_root, - prompt_index, - requested_basename=_requested_basename_for_identity(identity, wanted), - ) - dedupe_key = (unit.basename, unit.language, unit.prompt_path) - if dedupe_key in seen: - continue - seen.add(dedupe_key) - units.append(unit) - for unit in prompt_units: - if not _matches_module(unit, wanted): - continue - dedupe_key = (unit.basename, unit.language, unit.prompt_path) - if dedupe_key in seen: - continue - seen.add(dedupe_key) - units.append(unit) - return units +def _append_unique_unit( + units: List[SyncUnit], + seen: set[tuple[str, str, Path]], + unit: SyncUnit, +) -> None: + dedupe_key = (unit.basename, unit.language, unit.prompt_path) + if dedupe_key in seen: + return + seen.add(dedupe_key) + units.append(unit) + +def _metadata_units( + metadata_identities: List[tuple[str, str]], + prompt_root: Path, + prompt_index: Dict[tuple[str, str], SyncUnit], + wanted: set[str], + budget: Dict[str, int], + pddrc_path: Path | None, + config: Dict[str, Any] | None, +) -> List[SyncUnit]: units: List[SyncUnit] = [] seen: set[tuple[str, str, Path]] = set() for identity in metadata_identities: - unit = _unit_from_metadata_identity(identity, prompt_root, prompt_index) - dedupe_key = (unit.basename, unit.language, unit.prompt_path) - if dedupe_key in seen: - continue - seen.add(dedupe_key) - units.append(unit) - for unit in prompt_units: - dedupe_key = (unit.basename, unit.language, unit.prompt_path) - if dedupe_key in seen: + if wanted and not _identity_matches_wanted(identity, wanted): continue - seen.add(dedupe_key) - units.append(unit) + unit = _unit_from_metadata_identity( + identity, + prompt_root, + prompt_index, + requested_basename=_requested_basename_for_identity(identity, wanted), + budget=budget, + pddrc_path=pddrc_path, + config=config, + ) + _append_unique_unit(units, seen, unit) return units -def _load_fingerprint_json(path: Path) -> tuple[Optional[Dict[str, Any]], Optional[str]]: - if not path.exists(): +def _discover_units_and_failures( + root: Optional[Path] = None, + modules: Optional[Iterable[str]] = None, +) -> tuple[List[SyncUnit], List[DiscoveryFailure]]: + """Discover prompt-backed units and discovery failures under ``root``.""" + base = project_root(root) + wanted = set(modules or []) + config_cache: Dict[Path, Dict[str, Any]] = {} + prompt_roots, failures = _validated_prompt_roots(base, config_cache) + if any(failure.failure == "invalid_pddrc" for failure in failures): + return [], failures + prompt_units: List[SyncUnit] = [] + budget = {"entries": 0, "files": 0} + traversal_roots = [ + root for root in prompt_roots + if not any(root != other and _is_within_root(root, other) for other in prompt_roots) + ] + for prompt_root in traversal_roots: + if not prompt_root.is_dir(): + continue + units, unit_failures = _prompt_units( + prompt_root, budget, prompt_roots, base, config_cache + ) + prompt_units.extend(units) + failures.extend(unit_failures) + metadata_identities, metadata_failure = _metadata_identities( + base / ".pdd" / "meta", base, budget + ) + if metadata_failure is not None: + failures.append(metadata_failure) + if metadata_failure.failure != "discovery_budget": + return [], failures + + units = _metadata_units( + metadata_identities, + prompt_roots[0] if prompt_roots else base / "prompts", + _prompt_index(prompt_units), + wanted, + budget, + base / ".pddrc" if base / ".pddrc" in config_cache else None, + config_cache.get(base / ".pddrc"), + ) + seen = {(unit.basename, unit.language, unit.prompt_path) for unit in units} + if wanted: + for unit in prompt_units: + if not _matches_module(unit, wanted): + continue + _append_unique_unit(units, seen, unit) + return units, failures + + for unit in prompt_units: + _append_unique_unit(units, seen, unit) + return units, failures + + +def _prompt_index(prompt_units: List[SyncUnit]) -> Dict[tuple[str, str], SyncUnit]: + prompt_index: Dict[tuple[str, str], SyncUnit] = {} + for unit in prompt_units: + prompt_index.setdefault((_safe_basename(unit.basename), unit.language), unit) + return prompt_index + + +def _discovery_failure_payload(failure: DiscoveryFailure, root: Path) -> Dict[str, Any]: + try: + prompt_root = failure.prompt_root.resolve(strict=False).relative_to( + root.resolve() + ).as_posix() + except ValueError: + prompt_root = str(failure.prompt_root) + return { + "basename": "", + "language": "", + "classification": FAILURE_CLASSIFICATION, + "operation": "none", + "reason": failure.reason, + "changed_files": [], + "metadata_valid": False, + "paths": {"prompt_root": prompt_root}, + "failure": failure.failure, + } + + +def _load_fingerprint_json( + path: Path, + root: Path, +) -> tuple[Optional[Dict[str, Any]], Optional[str]]: + # pylint: disable=too-many-return-statements + violation = _path_component_violation(path, root, leaf_directory=False) + if violation == "missing": + return None, "missing" + if violation is not None: + return None, "unsafe_metadata" + try: + mode = path.lstat().st_mode + except FileNotFoundError: return None, "missing" + if stat.S_ISLNK(mode) or not stat.S_ISREG(mode): + return None, "invalid" try: data = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): @@ -474,21 +958,500 @@ def _load_fingerprint_json(path: Path) -> tuple[Optional[Dict[str, Any]], Option return data, None +def _fingerprint_from_payload(payload: Dict[str, Any]) -> Optional[Fingerprint]: + """Decode the legacy fingerprint without invoking its directory-creating reader.""" + try: + return Fingerprint( + pdd_version=payload["pdd_version"], + timestamp=payload["timestamp"], + command=payload["command"], + prompt_hash=payload.get("prompt_hash"), + code_hash=payload.get("code_hash"), + example_hash=payload.get("example_hash"), + test_hash=payload.get("test_hash"), + test_files=payload.get("test_files"), + include_deps=payload.get("include_deps"), + ) + except (KeyError, TypeError): + return None + + +def _relative_or_raw(path: Path, root: Path) -> str: + try: + return path.relative_to(root).as_posix() + except ValueError: + return str(path) + + +def _include_dep_violation(dep_path_value: Any, root: Path) -> Optional[Dict[str, str]]: + # pylint: disable=too-many-return-statements + if not isinstance(dep_path_value, str) or not dep_path_value.strip(): + return {"path": str(dep_path_value), "reason": "invalid_path"} + + raw_path = Path(dep_path_value) + dep_path = raw_path if raw_path.is_absolute() else root / raw_path + dep_path = Path(os.path.abspath(os.path.normpath(os.fspath(dep_path)))) + root_path = root.resolve() + try: + common_path = os.path.commonpath([str(root_path), str(dep_path)]) + except ValueError: + return {"path": dep_path_value, "reason": "outside_project"} + if common_path != str(root_path): + return {"path": dep_path_value, "reason": "outside_project"} + + try: + relative_parts = Path(os.path.relpath(dep_path, root_path)).parts + except ValueError: + return {"path": dep_path_value, "reason": "outside_project"} + + cursor = root_path + for index, part in enumerate(relative_parts): + cursor = cursor / part + is_leaf = index == len(relative_parts) - 1 + try: + dep_stat = cursor.lstat() + except ValueError: + return {"path": dep_path_value, "reason": "invalid_path"} + except OSError: + return {"path": dep_path_value, "reason": "missing"} + if stat.S_ISLNK(dep_stat.st_mode): + return {"path": dep_path_value, "reason": "symlink"} + if is_leaf: + if not stat.S_ISREG(dep_stat.st_mode): + return {"path": dep_path_value, "reason": "nonregular"} + elif not stat.S_ISDIR(dep_stat.st_mode): + return {"path": dep_path_value, "reason": "nonregular"} + + return None + + +def _unsafe_include_deps( + include_deps: Any, + root: Path, +) -> List[Dict[str, str]]: + if include_deps is None: + return [] + if not isinstance(include_deps, dict): + return [{"path": str(include_deps), "reason": "invalid_shape"}] + if not include_deps: + return [] + violations = [] + for dep_path_value, digest in sorted( + include_deps.items(), key=lambda item: str(item[0]) + ): + violation = _include_dep_violation(dep_path_value, root) + if violation is None and not isinstance(digest, str): + violation = {"path": str(dep_path_value), "reason": "invalid_digest"} + if violation is not None: + violations.append(violation) + return violations + + def _paths_as_json(paths: Dict[str, Any], root: Path) -> Dict[str, Any]: payload: Dict[str, Any] = {} + resolved_root = root.resolve() for key, value in paths.items(): if isinstance(value, Path): try: - payload[key] = value.resolve().relative_to(root.resolve()).as_posix() - except (OSError, ValueError): + payload[key] = value.resolve().relative_to(resolved_root).as_posix() + except (OSError, RuntimeError, ValueError): payload[key] = str(value) elif isinstance(value, list): - payload[key] = [str(item) for item in value] + payload[key] = [ + _relative_or_raw(item, resolved_root) + if isinstance(item, Path) else str(item) + for item in value + ] else: payload[key] = str(value) return payload +class UnsafeArchitectureError(ValueError): + """Raised when architecture metadata is unsafe to inspect.""" + + +class UnsafeConfigError(ValueError): + """Raised when project configuration is unsafe to inspect.""" + + +def _safe_architecture_candidate(path: Path, base: Path) -> bool: + """Validate a logical architecture path without following symlinks.""" + base = base.resolve() + candidate = path if path.is_absolute() else base / path + candidate = Path(os.path.abspath(os.path.normpath(os.fspath(candidate)))) + try: + parts = candidate.relative_to(base).parts + except ValueError: + return False + cursor = base + for part in parts: + cursor /= part + try: + mode = cursor.lstat().st_mode + except FileNotFoundError: + return True + except (OSError, ValueError): + return False + if stat.S_ISLNK(mode): + return False + return True + + +def _architecture_filepath( + unit: SyncUnit, + base: Path, + context_name: str | None = None, +) -> tuple[Path | None, str | None, Path | None]: + # pylint: disable=too-many-branches + """Return an architecture.json filepath match without mutating project state.""" + candidates: list[Path] = [] + for parent in (unit.prompt_path.parent, *unit.prompt_path.parents): + if not _is_within_root(parent, base): + break + candidates.append(parent / "architecture.json") + if parent == base: + break + for architecture_path in dict.fromkeys(candidates): + if not _safe_architecture_candidate(architecture_path, base): + raise UnsafeArchitectureError(f"unsafe architecture config: {architecture_path}") + if not architecture_path.is_file(): + continue + try: + payload = json.loads(architecture_path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as exc: + raise ValueError(f"architecture config is invalid: {architecture_path}") from exc + modules = extract_modules(payload) + exact_matches: list[Path] = [] + leaf_matches: list[Path] = [] + for item in modules: + if not isinstance(item, dict): + continue + filename = item.get("filename") + filepath = item.get("filepath") + if not isinstance(filename, str) or not isinstance(filepath, str): + continue + filename_path = Path(filename) + try: + prompt_relative = unit.prompt_path.relative_to(architecture_path.parent) + except ValueError: + prompt_relative = unit.prompt_path + configured_name = f"{unit.basename}_{unit.language}.prompt".lower() + is_exact = filename_path.as_posix().lower() in { + prompt_relative.as_posix().lower(), + configured_name, + } + path_aligned = _module_filepath_matches_basename( + _module_token_basename(filename, unit.language), + unit.basename, + context_name=context_name or unit.context_name, + ) + flat_basename = len(Path(unit.basename).parts) == 1 + leaf_match = ( + flat_basename + and filename_path.name.lower() == unit.prompt_path.name.lower() + ) + if (is_exact and (flat_basename or path_aligned)) or leaf_match: + output = Path(filepath) + if output.is_absolute() or ".." in output.parts: + raise ValueError(f"architecture output is invalid: {filepath}") + (exact_matches if is_exact else leaf_matches).append(output) + matches = exact_matches or leaf_matches + if len(matches) > 1: + raise ValueError("ambiguous module configuration") + if matches: + matched = matches[0] + duplicate_leaf = sum( + 1 + for module in modules + if isinstance(module, dict) + and isinstance(module.get("filepath"), str) + and Path(module["filepath"]).stem == matched.stem + ) > 1 + derived_stem = ( + _safe_basename(matched.with_suffix("").as_posix()) + if duplicate_leaf + else matched.stem + ) + return matched, derived_stem, architecture_path.parent + return None, None, None + + +def _configured_output_defaults( + unit: SyncUnit, base: Path +) -> tuple[Dict[str, Any], str | None, Path | None, Dict[str, Any]]: + """Return complete output defaults for the context owning ``unit``.""" + if unit.pddrc_path is not None: + if not _safe_control_file(unit.pddrc_path, base): + raise UnsafeConfigError(f"unsafe unit config: {unit.pddrc_path}") + pddrc_path = unit.pddrc_path + config = unit.config if unit.config is not None else _load_pddrc_config(pddrc_path) + else: + pddrc_path, config = _project_config(base) + if pddrc_path is None: + return {}, None, None, config + contexts = _validate_pddrc_structure(config) + context_name = _resolve_context_name_for_basename( + unit.basename, + context_override=unit.context_name, + pddrc_path=pddrc_path, + config=config, + ) + if context_name is None: + stem, separator, language = unit.prompt_path.stem.rpartition("_") + if separator and stem and language: + _basename, owned_context, owned_config, _root = _prompt_ownership( + unit.prompt_path, stem, unit.prompts_dir, base, + {pddrc_path: config}, + ) + if owned_config == pddrc_path: + context_name = owned_context + context = contexts.get(context_name, {}) if context_name else {} + defaults = context.get("defaults", {}) if isinstance(context, dict) else {} + if not isinstance(defaults, dict): + raise ValueError(".pddrc defaults must contain a mapping") + return defaults, context_name, pddrc_path, config + + +def _bounded_report_test_files( + test_path: Path, + base: Path, + extension: str, + budget: Optional[Dict[str, int]], +) -> List[Path]: + """Discover contained matching tests after validating every path component.""" + if not _safe_architecture_candidate(test_path, base): + raise ValueError("configured test path is outside project or symlinked") + parent = test_path.parent + if not _safe_architecture_candidate(parent, base): + raise ValueError("configured test directory is outside project or symlinked") + shared_budget = budget if budget is not None else {} + matches: List[Path] = [] + try: + entries = os.scandir(parent) + with entries: + for entry in entries: + shared_budget["report_entries"] = ( + shared_budget.get("report_entries", 0) + 1 + ) + if shared_budget["report_entries"] > MAX_REPAIR_DISCOVERY_ENTRIES: + raise ValueError("report test discovery budget exhausted") + if not entry.is_file(follow_symlinks=False): + continue + if not entry.name.startswith(test_path.stem): + continue + if Path(entry.name).suffix != f".{extension}": + continue + matches.append(parent / entry.name) + except FileNotFoundError: + return [test_path] + except OSError as exc: + raise ValueError(f"report test discovery failed: {exc}") from exc + return sorted(matches) or [test_path] + + +def _resolve_report_paths( + unit: SyncUnit, + base: Path, + command_budget: Optional[Dict[str, int]] = None, +) -> Dict[str, Any]: + """Resolve report paths without creating directories or files.""" + extension = get_extension(unit.language) + suffix = f".{extension}" if extension else "" + defaults, context_name, pddrc_path, config = _configured_output_defaults(unit, base) + architecture_filepath, architecture_stem, architecture_root = _architecture_filepath( + unit, base, context_name + ) + code_path = architecture_filepath + architecture_path = code_path + relative_basename = _relative_basename_for_context( + unit.basename, context_name, pddrc_path=pddrc_path, config=config + ) + code_stem = architecture_stem or relative_basename + code_dir = defaults.get("generate_output_path", "") + example_dir = defaults.get("example_output_path", "examples") + test_dir = defaults.get("test_output_path", "tests") + for value in (code_dir, example_dir, test_dir): + if not isinstance(value, str): + raise ValueError("configured output directory must be a string") + architecture_paths = None + if code_path is not None: + architecture_paths = _architecture_artifact_paths( + architecture_root or base, + code_path, + code_stem, + extension, + code_dir, + example_dir, + test_dir, + ) + code_path = architecture_paths["code"] + else: + code_path = base / code_dir / f"{code_stem}{suffix}" + artifact_root = architecture_root or base + example_path = ( + architecture_paths["example"] if architecture_paths is not None + else artifact_root / example_dir / f"{code_stem}_example{suffix}" + ) + test_path = ( + architecture_paths["test"] if architecture_paths is not None + else artifact_root / test_dir / f"test_{code_stem}{suffix}" + ) + outputs = defaults.get("outputs") + if isinstance(outputs, dict) and outputs: + generated = _expand_output_templates( + relative_basename, + unit.language, + extension, + outputs, + str(unit.prompt_path), + ) + for output_name, rendered in generated.items(): + if output_name not in {"prompt", "code", "example", "test"}: + continue + raw_config = outputs.get(output_name, {}) + raw_template = ( + raw_config.get("path", "") if isinstance(raw_config, dict) else "" + ) + if ".." in Path(raw_template).parts: + raise ValueError(f"configured {output_name} path is outside project") + candidate = rendered if rendered.is_absolute() else base / rendered + if not _safe_architecture_candidate(candidate, base): + raise ValueError(f"configured {output_name} path is outside project") + if architecture_path is None and "code" in outputs: + code_path = ( + generated["code"] + if generated["code"].is_absolute() + else base / generated["code"] + ) + if "example" in outputs: + example_path = ( + generated["example"] + if generated["example"].is_absolute() + else base / generated["example"] + ) + if "test" in outputs: + test_path = ( + generated["test"] + if generated["test"].is_absolute() + else base / generated["test"] + ) + test_files = [test_path] + if architecture_paths is not None: + paths_are_safe = all( + _safe_architecture_candidate(artifact_path, base) + for artifact_path in (code_path, example_path, test_path) + ) + if paths_are_safe: + test_files = _bounded_report_test_files( + test_path, base, extension, command_budget + ) + return { + "prompt": unit.prompt_path, + "code": code_path, + "example": example_path, + "test": test_path, + "test_files": test_files, + } + + +def _is_within_root(path: Path, root: Path) -> bool: + try: + path.relative_to(root) + return True + except ValueError: + return False + + +def _artifact_path_violation(path: Path, root: Path) -> Optional[str]: + # pylint: disable=too-many-return-statements + root = root.resolve() + candidate = path if path.is_absolute() else root / path + normalized = Path(os.path.abspath(os.path.normpath(os.fspath(candidate)))) + try: + relative_parts = normalized.relative_to(root).parts + except ValueError: + return "resolves outside project" + cursor = root + for part in relative_parts: + cursor /= part + try: + component_mode = cursor.lstat().st_mode + except FileNotFoundError: + break + except (OSError, RuntimeError, ValueError): + return "is an invalid path" + if stat.S_ISLNK(component_mode): + return "is a symlink" if cursor == normalized else "contains a symlink" + try: + mode = candidate.lstat().st_mode + except FileNotFoundError: + mode = None + except (OSError, RuntimeError, ValueError): + return "is an invalid path" + if mode is not None and stat.S_ISLNK(mode): + return "is a symlink" + try: + resolved = candidate.resolve(strict=False) + except (OSError, RuntimeError, ValueError): + return "is an invalid path" + if not _is_within_root(resolved, root): + return "resolves outside project" + if mode is None: + return None + if not stat.S_ISREG(mode): + return "is not a regular file" + return None + + +def _unsafe_fingerprinted_artifacts( + paths: Dict[str, Any], + root: Path, +) -> Dict[str, str]: + unsafe: Dict[str, str] = {} + for artifact in ("prompt", "code", "example", "test"): + path = paths.get(artifact) + if isinstance(path, Path): + violation = _artifact_path_violation(path, root) + if violation: + unsafe[artifact] = violation + for path in paths.get("test_files", []): + if isinstance(path, Path): + violation = _artifact_path_violation(path, root) + if violation: + unsafe[f"test_files:{_relative_or_raw(path, root.resolve())}"] = violation + return unsafe + + +def _unsafe_artifact_result( + unit: SyncUnit, + paths: Dict[str, Any], + root: Path, + fingerprint_path: Path, +) -> Optional[Dict[str, Any]]: + """Return a unit failure when any resolved artifact violates project safety.""" + unsafe_artifacts = _unsafe_fingerprinted_artifacts(paths, root) + if not unsafe_artifacts: + return None + changed_files = sorted(key.split(":", 1)[0] for key in unsafe_artifacts) + return { + "basename": unit.basename, + "language": unit.language, + "classification": FAILURE_CLASSIFICATION, + "operation": "none", + "reason": "unsafe fingerprinted artifacts: " + + ", ".join( + f"{artifact} {reason}" + for artifact, reason in sorted(unsafe_artifacts.items()) + ), + "changed_files": changed_files, + "metadata_valid": False, + "fingerprint_path": str(fingerprint_path), + "paths": _paths_as_json(paths, root), + "failure": "unsafe_artifacts", + } + + def _changed_artifacts( fingerprint: Fingerprint, paths: Dict[str, Path], @@ -561,16 +1524,56 @@ def _find_matching_artifact( root: Path, filename: str, expected_hash: str, -) -> Optional[Path]: + budget: Optional[Dict[str, int]] = None, +) -> tuple[Optional[Path], Optional[DiscoveryFailure]]: matches: List[Path] = [] - for candidate in root.rglob(filename): - if any(part in {".git", ".pdd", "__pycache__"} for part in candidate.parts): - continue - if candidate.is_file() and calculate_sha256(candidate) == expected_hash: - matches.append(candidate) + resolved_root = root.resolve() + if budget is None: + budget = {"repair_entries": 0} + pending = [root] + while pending: + current = pending.pop() + try: + entries = os.scandir(current) + with entries: + for entry in entries: + budget["repair_entries"] = budget.get("repair_entries", 0) + 1 + if budget["repair_entries"] > MAX_REPAIR_DISCOVERY_ENTRIES: + return None, DiscoveryFailure( + prompt_root=root, + reason=( + "repair search exceeded traversal budget " + f"of {MAX_REPAIR_DISCOVERY_ENTRIES} directory entries" + ), + failure="repair_traversal_budget", + ) + candidate = current / entry.name + try: + is_directory = entry.is_dir(follow_symlinks=False) + except OSError as exc: + raise OSError(f"cannot inspect {candidate}: {exc}") from exc + if is_directory: + if not _is_hidden_or_system_dir(candidate): + pending.append(candidate) + continue + if entry.name != filename: + continue + if _artifact_path_violation(candidate, resolved_root): + continue + if ( + entry.is_file(follow_symlinks=False) + and calculate_sha256(candidate) == expected_hash + ): + matches.append(candidate) + except OSError as exc: + return None, DiscoveryFailure( + prompt_root=root, + reason=f"repair search traversal failed: {exc}", + failure="repair_traversal_error", + ) if len(matches) == 1: - return matches[0] - return None + return matches[0], None + return None, None def _repair_missing_fingerprinted_paths( @@ -578,7 +1581,8 @@ def _repair_missing_fingerprinted_paths( fingerprint: Fingerprint, basename: str, root: Path, -) -> Dict[str, Path]: + budget: Optional[Dict[str, int]] = None, +) -> tuple[Dict[str, Path], Optional[DiscoveryFailure]]: repaired = dict(paths) field_by_artifact = { "code": "code_hash", @@ -593,13 +1597,15 @@ def _repair_missing_fingerprinted_paths( filename = _artifact_search_name(artifact, basename, path) if not filename: continue - match = _find_matching_artifact(root, filename, expected_hash) + match, failure = _find_matching_artifact(root, filename, expected_hash, budget) + if failure is not None: + return repaired, failure if match is None: continue repaired[artifact] = match if artifact == "test": repaired["test_files"] = [match] - return repaired + return repaired, None def _missing_required_hashes(fingerprint: Fingerprint, paths: Dict[str, Path]) -> List[str]: @@ -662,17 +1668,26 @@ def _classification_for_changes(changes: List[str]) -> str: return "IN_SYNC" -def classify_unit(unit: SyncUnit, root: Optional[Path] = None) -> Dict[str, Any]: - # pylint: disable=broad-exception-caught +def classify_unit( + unit: SyncUnit, + root: Optional[Path] = None, + command_budget: Optional[Dict[str, int]] = None, +) -> Dict[str, Any]: + # pylint: disable=broad-exception-caught,too-many-locals,too-many-return-statements """Classify one sync unit without mutating files.""" base = project_root(root) + # A report must not create `.pdd/meta` just to discover that no fingerprint + # exists. The legacy helper creates its parent directory as a write-side + # convenience, so derive the read-only project-relative location here. + fp_path = base / ".pdd" / "meta" / f"{_safe_basename(unit.basename)}_{unit.language}.json" try: - paths = get_pdd_file_paths( - unit.basename, - unit.language, - prompts_dir=str(unit.prompts_dir), + paths = _resolve_report_paths(unit, base, command_budget) + except Exception as exc: # surfaced in JSON report + failure = ( + "unsafe_architecture" + if isinstance(exc, UnsafeArchitectureError) + else "path_resolution" ) - except Exception as exc: # pragma: no cover - surfaced in JSON report return { "basename": unit.basename, "language": unit.language, @@ -681,32 +1696,43 @@ def classify_unit(unit: SyncUnit, root: Optional[Path] = None) -> Dict[str, Any] "reason": f"path resolution failed: {exc}", "changed_files": [], "metadata_valid": False, + "fingerprint_path": str(fp_path), "paths": {"prompt": str(unit.prompt_path)}, + "failure": failure, } - fp_path = get_fingerprint_path(unit.basename, unit.language, paths=paths) - _raw_fp, raw_error = _load_fingerprint_json(fp_path) - if raw_error is not None: + unsafe_result = _unsafe_artifact_result(unit, paths, base, fp_path) + if unsafe_result is not None: + return unsafe_result + + _raw_fp, raw_error = _load_fingerprint_json(fp_path, base) + if raw_error == "unsafe_metadata": return { "basename": unit.basename, "language": unit.language, - "classification": UNBASELINED_CLASSIFICATION, + "classification": FAILURE_CLASSIFICATION, "operation": "none", - "reason": f"fingerprint {raw_error}", + "reason": "unsafe fingerprint metadata path", "changed_files": [], "metadata_valid": False, "fingerprint_path": str(fp_path), "paths": _paths_as_json(paths, base), + "failure": "unsafe_metadata", } - - fingerprint = read_fingerprint(unit.basename, unit.language, paths=paths) - if fingerprint is None: + fingerprint = ( + None if raw_error is not None else _fingerprint_from_payload(_raw_fp) + ) + if raw_error is not None or fingerprint is None: return { "basename": unit.basename, "language": unit.language, "classification": UNBASELINED_CLASSIFICATION, "operation": "none", - "reason": "fingerprint invalid", + "reason": ( + f"fingerprint {raw_error}" + if raw_error is not None + else "fingerprint invalid" + ), "changed_files": [], "metadata_valid": False, "fingerprint_path": str(fp_path), @@ -714,12 +1740,30 @@ def classify_unit(unit: SyncUnit, root: Optional[Path] = None) -> Dict[str, Any] } if not unit.prompt_path.exists(): - paths = _repair_missing_fingerprinted_paths( + paths, repair_failure = _repair_missing_fingerprinted_paths( paths, fingerprint, unit.basename, base, + command_budget, ) + if repair_failure is not None: + return { + "basename": unit.basename, + "language": unit.language, + "classification": FAILURE_CLASSIFICATION, + "operation": "none", + "reason": repair_failure.reason, + "changed_files": [], + "metadata_valid": False, + "fingerprint_path": str(fp_path), + "paths": _paths_as_json(paths, base), + "failure": repair_failure.failure, + } + + unsafe_result = _unsafe_artifact_result(unit, paths, base, fp_path) + if unsafe_result is not None: + return unsafe_result missing_hashes = _missing_required_hashes(fingerprint, paths) if missing_hashes: @@ -752,10 +1796,41 @@ def classify_unit(unit: SyncUnit, root: Optional[Path] = None) -> Dict[str, Any] "failure": "missing_artifacts", } - current_hashes = calculate_current_hashes( - paths, - stored_include_deps=fingerprint.include_deps, - ) + unsafe_include_deps = _unsafe_include_deps(fingerprint.include_deps, base) + if unsafe_include_deps: + return { + "basename": unit.basename, + "language": unit.language, + "classification": FAILURE_CLASSIFICATION, + "operation": "none", + "reason": "unsafe legacy include dependency metadata", + "changed_files": [], + "metadata_valid": False, + "fingerprint_path": str(fp_path), + "paths": _paths_as_json(paths, base), + "failure": "unsafe_include_deps", + "unsafe_include_deps": unsafe_include_deps, + } + + try: + current_hashes = calculate_current_hashes( + paths, + stored_include_deps=fingerprint.include_deps, + dependency_root=base, + ) + except Exception as exc: # surfaced as a unit-scoped machine-readable failure + return { + "basename": unit.basename, + "language": unit.language, + "classification": FAILURE_CLASSIFICATION, + "operation": "none", + "reason": f"fingerprint hash calculation failed: {exc}", + "changed_files": [], + "metadata_valid": False, + "fingerprint_path": str(fp_path), + "paths": _paths_as_json(paths, base), + "failure": "hash_calculation", + } changes = _changed_artifacts(fingerprint, paths, current_hashes) classification = _classification_for_changes(changes) return { @@ -850,8 +1925,12 @@ def build_report( "canonical read-only reporting cannot append a repository ledger" ) return _canonical_compatibility_report(base, consumer, modules) - units = discover_units(base, modules=modules) - classified = [classify_unit(unit, base) for unit in units] + units, discovery_failures = _discover_units_and_failures(base, modules=modules) + command_budget = {"repair_entries": 0} + classified = [classify_unit(unit, base, command_budget) for unit in units] + classified.extend( + _discovery_failure_payload(failure, base) for failure in discovery_failures + ) summary = _build_summary(classified) ledger_path = _append_ledger(base, consumer, classified) if ledger else None ok = ( diff --git a/pdd/core/cli.py b/pdd/core/cli.py index b387b684fc..38245628d5 100644 --- a/pdd/core/cli.py +++ b/pdd/core/cli.py @@ -870,7 +870,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/sync_core/fingerprint_store.py b/pdd/sync_core/fingerprint_store.py index 4765f3a634..b4f8acec09 100644 --- a/pdd/sync_core/fingerprint_store.py +++ b/pdd/sync_core/fingerprint_store.py @@ -275,12 +275,18 @@ def read_legacy(self, path: Path) -> LegacyFingerprintRecord: candidate = Path(path) if not candidate.is_absolute(): candidate = self.checkout_root / candidate + try: + candidate_mode = candidate.lstat().st_mode + except OSError as exc: + raise FingerprintStoreError("legacy fingerprint is not a regular file") from exc + if stat.S_ISLNK(candidate_mode) or not stat.S_ISREG(candidate_mode): + raise FingerprintStoreError("legacy fingerprint is not a regular file") resolved = candidate.resolve(strict=True) try: resolved.relative_to(self.checkout_root) except ValueError as exc: raise FingerprintStoreError("legacy fingerprint escapes checkout") from exc - if candidate.is_symlink() or not resolved.is_file(): + if not resolved.is_file(): raise FingerprintStoreError("legacy fingerprint is not a regular file") try: payload = json.loads(resolved.read_text(encoding="utf-8")) diff --git a/pdd/sync_core/supervisor.py b/pdd/sync_core/supervisor.py index 3fa340e92a..d690b1b4ec 100644 --- a/pdd/sync_core/supervisor.py +++ b/pdd/sync_core/supervisor.py @@ -19,6 +19,12 @@ import sysconfig +# Capture the executable that loaded this trusted module. Tests and callers may +# replace ``sys.executable`` to model argv-prefix portability; that synthetic +# spelling must never become a measured file or sandbox mount source. +_SUPERVISOR_EXECUTABLE = Path(sys.executable) + + @dataclass(frozen=True) class SupervisorLimits: """Hard limits applied to every untrusted validator process tree.""" @@ -86,7 +92,7 @@ def _runtime_directories() -> tuple[tuple[str, Path], ...]: def released_runtime_closure_paths() -> tuple[tuple[str, Path], ...]: """Return every regular file exposed by the sandbox with logical names.""" entries: dict[str, Path] = {} - native: set[Path] = {Path(sys.executable).resolve()} + native: set[Path] = {_SUPERVISOR_EXECUTABLE.resolve()} for label, directory in _runtime_directories(): for path in sorted(directory.rglob("*")): if path.is_file() and not path.is_symlink(): @@ -105,7 +111,7 @@ def released_runtime_closure_paths() -> tuple[tuple[str, Path], ...]: path = Path(value).resolve() entries[f"sandbox/{name}"] = path native.add(path) - entries["interpreter/python"] = Path(sys.executable).resolve() + entries["interpreter/python"] = _SUPERVISOR_EXECUTABLE.resolve() for path in sorted(native): for library in _linked_libraries(path): entries.setdefault( @@ -120,7 +126,7 @@ def _runtime_roots(command: list[str], cwd: Path) -> tuple[Path, ...]: directories = tuple(directory for _label, directory in _runtime_directories()) roots.update(directories) executables = ( - Path(sys.executable), Path(shutil.which(command[0]) or command[0]), + _SUPERVISOR_EXECUTABLE, Path(shutil.which(command[0]) or command[0]), ) for executable in executables: resolved_executable = executable.resolve() @@ -163,7 +169,7 @@ def _limited_command(command: list[str], limits: SupervisorLimits) -> list[str]: "resource.setrlimit(resource.RLIMIT_NOFILE,(v[4],v[4]));" "os.execvpe(sys.argv[6],sys.argv[6:],os.environ)" ) - return [sys.executable, "-c", script, str(limits.max_memory_bytes), + return [str(_SUPERVISOR_EXECUTABLE), "-c", script, str(limits.max_memory_bytes), str(limits.max_cpu_seconds), str(limits.max_processes), str(limits.max_output_bytes), "256", *command] @@ -189,7 +195,7 @@ def _staged_bwrap(argv: list[str], sources: list[Path]) -> list[str]: " shutil.rmtree(base,ignore_errors=True)", "raise SystemExit(result.returncode)", )) - return ["sudo", "-n", "-E", sys.executable, "-c", helper, + return ["sudo", "-n", "-E", str(_SUPERVISOR_EXECUTABLE), "-c", helper, json.dumps(argv), json.dumps([str(path) for path in sources])] def _supervised_descendants(token: str) -> set[int]: diff --git a/pdd/sync_determine_operation.py b/pdd/sync_determine_operation.py index 122e6c953c..e3c8286d7c 100644 --- a/pdd/sync_determine_operation.py +++ b/pdd/sync_determine_operation.py @@ -12,6 +12,7 @@ import glob import json import hashlib +import stat import subprocess import fnmatch from pathlib import Path, PurePosixPath @@ -330,19 +331,23 @@ def _case_insensitive_path_lookup(candidate: Path) -> Optional[Path]: def _resolve_context_name_for_basename( basename: str, context_override: Optional[str] = None, + *, + pddrc_path: Optional[Path] = None, + config: Optional[Dict[str, Any]] = None, ) -> Optional[str]: """Resolve the context for a basename when no explicit override is provided.""" if context_override: return context_override - pddrc_path = _find_pddrc_file() + pddrc_path = pddrc_path or _find_pddrc_file() if not pddrc_path: return None - try: - config = _load_pddrc_config(pddrc_path) - except ValueError: - return None + if config is None: + try: + config = _load_pddrc_config(pddrc_path) + except ValueError: + return None return _detect_context_from_basename(basename, config, pddrc_path=pddrc_path) @@ -951,19 +956,26 @@ def _resolve_prompts_root(prompts_dir: str) -> Path: return prompts_root -def _relative_basename_for_context(basename: str, context_name: Optional[str]) -> str: +def _relative_basename_for_context( + basename: str, + context_name: Optional[str], + *, + pddrc_path: Optional[Path] = None, + config: Optional[Dict[str, Any]] = None, +) -> str: """Strip context-specific prefixes from basename when possible.""" if not context_name: return basename - pddrc_path = _find_pddrc_file() + pddrc_path = pddrc_path or _find_pddrc_file() if not pddrc_path: return basename - try: - config = _load_pddrc_config(pddrc_path) - except ValueError: - return basename + if config is None: + try: + config = _load_pddrc_config(pddrc_path) + except ValueError: + return basename contexts = config.get("contexts", {}) context_config = contexts.get(context_name, {}) @@ -994,29 +1006,14 @@ def _relative_basename_for_context(basename: str, context_name: Optional[str]) - return matches[0][1] -def _generate_paths_from_templates( +def _expand_output_templates( basename: str, language: str, extension: str, outputs_config: Dict[str, Any], - prompt_path: str + prompt_path: str, ) -> Dict[str, Path]: - """ - Generate file paths from template configuration. - - This function is used by Issue #237 to support extensible output path patterns - for different project layouts (Next.js, Vue, Python backend, etc.). - - Args: - basename: The relative basename (e.g., 'marketplace/AssetCard' or 'credit_helpers') - language: The full language name (e.g., 'python', 'typescript') - extension: The file extension (e.g., 'py', 'tsx') - outputs_config: The 'outputs' section from .pddrc context config - prompt_path: The prompt file path to use as fallback - - Returns: - Dictionary mapping file types ('prompt', 'code', 'test', etc.) to Path objects - """ + """Purely expand configured output templates without filesystem access.""" import logging logger = logging.getLogger(__name__) @@ -1025,35 +1022,6 @@ def _generate_paths_from_templates( name = parts[-1] if parts else basename category = '/'.join(parts[:-1]) if len(parts) > 1 else '' - # Issue #237 fix: If category is empty but we have an actual prompt_path, - # try to derive the category from the prompt path by comparing with template - if not category and prompt_path and Path(prompt_path).exists(): - prompt_template = outputs_config.get('prompt', {}).get('path', '') - if prompt_template and '{category}' in prompt_template: - # Extract category from actual prompt path - # Template: prompts/frontend/{category}/{name}_{language}.prompt - # Actual: prompts/frontend/app/page_TypescriptReact.prompt - # Category: app - prompt_path_obj = Path(prompt_path) - prompt_parts = prompt_path_obj.parts - - # Find where the template's fixed prefix ends - # E.g., "prompts/frontend/" -> look for index after "frontend" - template_prefix = prompt_template.split('{category}')[0].rstrip('/') - template_prefix_parts = Path(template_prefix).parts if template_prefix else () - - # Find the matching index in the actual path - if template_prefix_parts: - for i, part in enumerate(prompt_parts): - if prompt_parts[i:i+len(template_prefix_parts)] == template_prefix_parts: - # Category starts after the prefix, ends before the filename - category_start = i + len(template_prefix_parts) - category_end = len(prompt_parts) - 1 # Exclude filename - if category_start < category_end: - category = '/'.join(prompt_parts[category_start:category_end]) - logger.info(f"Derived category '{category}' from prompt path: {prompt_path}") - break - # Build dir_prefix (for legacy template compatibility) dir_prefix = '/'.join(parts[:-1]) + '/' if len(parts) > 1 else '' if category and not dir_prefix: @@ -1077,10 +1045,9 @@ def _generate_paths_from_templates( if isinstance(config, dict) and 'path' in config: template = config['path'] expanded = expand_template(template, template_context) + if Path(template).is_absolute() and not Path(expanded).is_absolute(): + expanded = str(Path(Path(template).anchor) / expanded) result[output_type] = Path(expanded) - if output_type == 'prompt': - from pdd.sync_main import _case_insensitive_prompt_lookup - result[output_type] = _case_insensitive_prompt_lookup(result[output_type]) logger.debug(f"Template {output_type}: {template} -> {expanded}") # Ensure prompt is always present (fallback to provided prompt_path) @@ -1097,6 +1064,26 @@ def _generate_paths_from_templates( if 'test' not in result: result['test'] = Path(f"tests/test_{name}{_dot(extension)}") + result['test_files'] = [result['test']] + return result + + +def _generate_paths_from_templates( + basename: str, + language: str, + extension: str, + outputs_config: Dict[str, Any], + prompt_path: str +) -> Dict[str, Path]: + """Expand output templates and perform legacy live-path discovery.""" + result = _expand_output_templates( + basename, language, extension, outputs_config, prompt_path + ) + name = basename.split('/')[-1] if basename else basename + if 'prompt' in outputs_config: + from pdd.sync_main import _case_insensitive_prompt_lookup + result['prompt'] = _case_insensitive_prompt_lookup(result['prompt']) + # Handle test_files for Bug #156 compatibility if 'test' in result: test_path = result['test'] @@ -1111,6 +1098,29 @@ def _generate_paths_from_templates( return result +def _architecture_artifact_paths( + project_root: Path, + architecture_filepath: Path, + artifact_stem: str, + extension: str, + generate_dir: str = "", + example_dir: str = "examples/", + test_dir: str = "tests/", +) -> Dict[str, Any]: + """Construct architecture artifact paths without inspecting the filesystem.""" + code_path = project_root / architecture_filepath + if generate_dir and architecture_filepath.parent == Path("."): + code_path = project_root / generate_dir / architecture_filepath.name + example_path = project_root / example_dir / f"{artifact_stem}_example{_dot(extension)}" + test_path = project_root / test_dir / f"test_{artifact_stem}{_dot(extension)}" + return { + "code": code_path, + "example": example_path, + "test": test_path, + "test_files": [test_path], + } + + def get_pdd_file_paths(basename: str, language: str, prompts_dir: str = "prompts", context_override: Optional[str] = None) -> Dict[str, Path]: """Returns a dictionary mapping file types to their expected Path objects. @@ -1134,6 +1144,7 @@ def get_pdd_file_paths(basename: str, language: str, prompts_dir: str = "prompts name = basename.split('/')[-1] if '/' in basename else basename resolved_context_name = _resolve_context_name_for_basename(basename, context_override) construct_paths_basename = _relative_basename_for_context(basename, resolved_context_name) + template_basename = construct_paths_basename # Anchor configuration lookups (architecture.json, .pddrc) at the resolved # prompts root so nested subprojects (e.g. extensions//prompts/) find @@ -1160,6 +1171,20 @@ def get_pdd_file_paths(basename: str, language: str, prompts_dir: str = "prompts resolved_prompt = _find_prompt_file(basename, language, prompts_root, arch_path, context_override=context_override) if resolved_prompt: prompt_path = str(resolved_prompt) + try: + relative_prompt = resolved_prompt.resolve().relative_to(prompts_root.resolve()) + prompt_stem = relative_prompt.stem + suffix = f"_{language}" + if prompt_stem.lower().endswith(suffix.lower()): + prompt_stem = prompt_stem[:-len(suffix)] + discovered_basename = ( + relative_prompt.parent / prompt_stem + ).as_posix() + template_basename = _relative_basename_for_context( + discovered_basename, resolved_context_name + ) + except ValueError: + pass else: # File doesn't exist yet (new module being created) — construct expected path # Respect .pddrc context's prompts_dir prefix so new modules land in the @@ -1260,17 +1285,7 @@ def get_pdd_file_paths(basename: str, language: str, prompts_dir: str = "prompts except ValueError: pass - # Apply generate_output_path only when arch_filepath is a bare filename - # at the project root (no directory component). When arch_filepath already - # contains a subdirectory structure, that structure takes precedence. - # Preserve the explicit filename (including extension) from architecture.json; - # only the parent directory is overridden by .pddrc generate_output_path. arch_filepath_path = Path(arch_filepath) - if generate_dir and str(arch_filepath_path.parent) in (".", ""): - code_path = project_root / f"{generate_dir}{arch_filepath_path.name}" - logger.debug(f"Path source: generate={code_path} (from pddrc generate_output_path)") - else: - logger.debug(f"Path source: generate={code_path} (from architecture.json)") # Issue #1677: when the leaf basename is ambiguous (several architecture # modules share it, e.g. Next.js `page`), two path-qualified modules @@ -1282,8 +1297,10 @@ def get_pdd_file_paths(basename: str, language: str, prompts_dir: str = "prompts if arch_path and len(_architecture_module_choices(arch_path, name, language)) > 1: example_stem = _safe_basename(Path(arch_filepath).with_suffix("").as_posix()) - example_path = project_root / f"{example_dir}{example_stem}_example{_dot(extension)}" - test_path = project_root / f"{test_dir}test_{example_stem}{_dot(extension)}" + artifacts = _architecture_artifact_paths( + project_root, arch_filepath_path, example_stem, extension, + generate_dir, example_dir, test_dir, + ) # If the flattened prompt basename already has corresponding example/test # artifacts, prefer those over the architecture filepath stem. This keeps @@ -1295,10 +1312,11 @@ def get_pdd_file_paths(basename: str, language: str, prompts_dir: str = "prompts preferred_example = False preferred_test = False if basename_example_path.exists(): - example_path = basename_example_path + artifacts["example"] = basename_example_path preferred_example = True if basename_test_path.exists(): - test_path = basename_test_path + artifacts["test"] = basename_test_path + artifacts["test_files"] = [basename_test_path] preferred_test = True if preferred_example or preferred_test: logger.info( @@ -1310,20 +1328,7 @@ def get_pdd_file_paths(basename: str, language: str, prompts_dir: str = "prompts preferred_test, ) - test_dir_path = test_path.parent - test_stem = glob.escape(test_path.stem) - if test_dir_path.exists(): - matching_test_files = sorted(test_dir_path.glob(f"{test_stem}*.{extension}")) - else: - matching_test_files = [test_path] if test_path.exists() else [] - - result = { - 'prompt': Path(prompt_path), - 'code': code_path, - 'example': example_path, - 'test': test_path, - 'test_files': matching_test_files or [test_path] - } + result = {"prompt": Path(prompt_path), **artifacts} logger.info(f"get_pdd_file_paths returning (from architecture.json): {result}") return result @@ -1510,7 +1515,7 @@ def get_pdd_file_paths(basename: str, language: str, prompts_dir: str = "prompts extension = get_extension(language) logger.info(f"Using template-based paths from outputs config (prompt exists)") context_name = context_override or resolved_config.get('_matched_context') - basename_for_templates = _relative_basename_for_context(basename, context_name) + basename_for_templates = template_basename result = _generate_paths_from_templates( basename=basename_for_templates, language=language, @@ -1727,7 +1732,71 @@ def calculate_sha256(file_path: Path) -> Optional[str]: return None -def extract_include_deps(prompt_path: Path, *, version: int = 2) -> Dict[str, str]: +def _safe_report_include(reference: str, prompt_path: Path, root: Path) -> Optional[Path]: + """Resolve one legacy report include without CWD or symlink traversal.""" + declared = Path(reference) + if declared.is_absolute(): + raise ValueError(f"absolute include path is unsafe: {reference}") + root = root.resolve() + candidates = (prompt_path.parent / declared, root / declared) + for candidate in dict.fromkeys(candidates): + normalized = Path(os.path.abspath(os.path.normpath(os.fspath(candidate)))) + try: + parts = normalized.relative_to(root).parts + except ValueError as exc: + raise ValueError(f"include path escapes project: {reference}") from exc + cursor = root + missing = False + for index, part in enumerate(parts): + cursor /= part + try: + mode = cursor.lstat().st_mode + except FileNotFoundError: + missing = True + break + except (OSError, ValueError) as exc: + raise ValueError(f"invalid include path: {reference}") from exc + if stat.S_ISLNK(mode): + raise ValueError(f"symlink include path is unsafe: {reference}") + if index < len(parts) - 1 and not stat.S_ISDIR(mode): + raise ValueError(f"non-directory include ancestor: {reference}") + if index == len(parts) - 1 and not stat.S_ISREG(mode): + raise ValueError(f"non-regular include path: {reference}") + if not missing: + return normalized + return None + + +def _validated_report_live_includes( + prompt_path: Path, root: Path +) -> tuple[bool, Optional[List[tuple[str, Path]]]]: + """Resolve all live legacy includes once, before any dependency is read.""" + from pdd.continuous_sync import canonical_sync_enabled + if canonical_sync_enabled(prompt_path) or not prompt_path.exists(): + return False, None + try: + content = prompt_path.read_text(encoding="utf-8", errors="ignore") + except OSError: + return False, None + references = _legacy_include_references(content) + if not references: + return False, None + resolved: List[tuple[str, Path]] = [] + for reference in references: + dependency = _safe_report_include(reference.strip(), prompt_path, root) + if dependency is None: + continue + resolved.append((reference.strip(), dependency)) + return True, resolved + + +def extract_include_deps( + prompt_path: Path, + dependency_root: Optional[Path] = None, + *, + version: int = 2, + resolved_live_dependencies: Optional[List[tuple[str, Path]]] = None, +) -> Dict[str, str]: """Extract include dependency paths and their hashes from a prompt file. Returns a dict mapping resolved dependency paths to their SHA256 hashes. @@ -1742,6 +1811,18 @@ def extract_include_deps(prompt_path: Path, *, version: int = 2) -> Dict[str, st from pdd.sync_core.path_policy import PathPolicy, PathPolicyError if not canonical_sync_enabled(prompt_path): + if resolved_live_dependencies is not None: + dependencies: Dict[str, str] = {} + for _declared, dependency in resolved_live_dependencies: + digest = calculate_sha256(dependency) + if digest: + key_root = dependency_root or Path.cwd() + try: + key = str(dependency.relative_to(key_root)) + except ValueError: + key = str(dependency) + dependencies[key] = digest + return dependencies if not prompt_path.exists(): return {} try: @@ -1755,19 +1836,25 @@ def extract_include_deps(prompt_path: Path, *, version: int = 2) -> Dict[str, st else [reference.path for reference in parse_include_references(content)] ) for declared_text in declared_paths: - declared = Path(declared_text.strip()) - candidates = ( - (declared,) - if declared.is_absolute() - else (prompt_path.parent / declared, Path.cwd() / declared) - ) - dependency = next((item for item in candidates if item.is_file()), None) + if dependency_root is not None: + dependency = _safe_report_include( + declared_text.strip(), prompt_path, dependency_root + ) + else: + declared = Path(declared_text.strip()) + candidates = ( + (declared,) + if declared.is_absolute() + else (prompt_path.parent / declared, Path.cwd() / declared) + ) + dependency = next((item for item in candidates if item.is_file()), None) if dependency is None: continue digest = calculate_sha256(dependency) if digest: try: - key = str(dependency.relative_to(Path.cwd())) + key_root = dependency_root or Path.cwd() + key = str(dependency.relative_to(key_root)) except ValueError: key = str(dependency) dependencies[key] = digest @@ -1806,8 +1893,10 @@ def _legacy_include_references(content: str) -> list[str]: def calculate_prompt_hash( prompt_path: Path, stored_deps: Optional[Dict[str, str]] = None, + dependency_root: Optional[Path] = None, *, hash_version: int = 1, + resolved_live_dependencies: Optional[List[tuple[str, Path]]] = None, ) -> Optional[str]: """Hash a prompt file including the content of all its dependencies. @@ -1819,6 +1908,7 @@ def calculate_prompt_hash( Args: prompt_path: Path to the prompt file. stored_deps: Previously stored dependency paths from fingerprint (issue #522). + dependency_root: Explicit base for stored relative dependency paths. Returns: SHA256 hex digest of the prompt + dependency contents, or None. @@ -1837,24 +1927,39 @@ def calculate_prompt_hash( if hash_version == 1 else [reference.path for reference in parse_include_references(prompt_content)] ) - declared_dependencies = ( - references - if references else list((stored_deps or {}).keys()) - ) - if hash_version == 1: - declared_dependencies = sorted(set(item.strip() for item in declared_dependencies)) - resolved_dependencies = [] - for declared in declared_dependencies: - if hash_version == 1 and not references: - candidate = Path(declared) - candidate = candidate if candidate.exists() else None + if references and resolved_live_dependencies is not None: + validated_dependencies = list(resolved_live_dependencies) + if hash_version == 1: + by_declaration = dict(validated_dependencies) + resolved_dependencies = [ + by_declaration[declared] + for declared in sorted(set(by_declaration)) + ] else: - candidate = _legacy_dependency_path(prompt_path, declared) - if candidate is None: - if hash_version == 1: - continue - return None - resolved_dependencies.append(candidate.resolve()) + resolved_dependencies = [path for _declared, path in validated_dependencies] + else: + declared_dependencies = ( + references + if references else list((stored_deps or {}).keys()) + ) + if hash_version == 1: + declared_dependencies = sorted( + set(item.strip() for item in declared_dependencies) + ) + resolved_dependencies = [] + for declared in declared_dependencies: + if hash_version == 1 and not references: + candidate = Path(declared) + if dependency_root is not None and not candidate.is_absolute(): + candidate = dependency_root / candidate + candidate = candidate if candidate.exists() else None + else: + candidate = _legacy_dependency_path(prompt_path, declared) + if candidate is None: + if hash_version == 1: + continue + return None + resolved_dependencies.append(candidate.resolve()) hasher = hashlib.sha256() hasher.update(prompt_path.read_bytes()) @@ -1948,13 +2053,18 @@ def read_run_report( return None -def calculate_current_hashes(paths: Dict[str, Any], stored_include_deps: Optional[Dict[str, str]] = None) -> Dict[str, Any]: +def calculate_current_hashes( + paths: Dict[str, Any], + stored_include_deps: Optional[Dict[str, str]] = None, + dependency_root: Optional[Path] = None, +) -> Dict[str, Any]: """Computes the hashes for all current files on disk. Args: paths: Dictionary of PDD file paths. stored_include_deps: Previously stored include dependency paths from fingerprint. Used when the prompt no longer has tags (issue #522). + dependency_root: Explicit base for stored relative dependency paths. """ hashes = {} for file_type, file_path in paths.items(): @@ -1967,15 +2077,37 @@ def calculate_current_hashes(paths: Dict[str, Any], stored_include_deps: Optiona } elif file_type == 'prompt' and isinstance(file_path, Path): # Issue #522: Hash prompt with dependencies - hashes['prompt_hash'] = calculate_prompt_hash(file_path, stored_deps=stored_include_deps) + has_live_includes = False + resolved_live_dependencies = None + if dependency_root is not None: + has_live_includes, resolved_live_dependencies = ( + _validated_report_live_includes(file_path, dependency_root) + ) + hashes['prompt_hash'] = calculate_prompt_hash( + file_path, + stored_deps=stored_include_deps, + dependency_root=dependency_root, + resolved_live_dependencies=resolved_live_dependencies, + ) # Also extract current include deps for persistence - hashes['include_deps'] = extract_include_deps(file_path, version=1) + hashes['include_deps'] = ( + extract_include_deps( + file_path, + dependency_root, + version=1, + resolved_live_dependencies=resolved_live_dependencies, + ) + if not has_live_includes or resolved_live_dependencies is not None + else {} + ) # If no deps found in prompt but we have stored deps, preserve them if not hashes['include_deps'] and stored_include_deps: # Re-hash stored deps to check for changes updated_deps = {} for dep_path_str, old_hash in stored_include_deps.items(): dep_path = Path(dep_path_str) + if not dep_path.is_absolute(): + dep_path = (dependency_root or Path.cwd()) / dep_path if dep_path.exists(): new_hash = calculate_sha256(dep_path) if new_hash: diff --git a/tests/e2e/test_issue_1932_continuous_sync_guarantee.py b/tests/e2e/test_issue_1932_continuous_sync_guarantee.py index be77a352fe..863f3701be 100644 --- a/tests/e2e/test_issue_1932_continuous_sync_guarantee.py +++ b/tests/e2e/test_issue_1932_continuous_sync_guarantee.py @@ -7,9 +7,11 @@ import sys from pathlib import Path from typing import Callable, Dict, List +from unittest.mock import patch import pytest +from pdd import continuous_sync from pdd.operation_log import save_fingerprint from pdd.sync_determine_operation import get_pdd_file_paths @@ -182,6 +184,13 @@ def _fingerprint(repo: Path) -> dict: return json.loads((repo / ".pdd/meta/widget_python.json").read_text(encoding="utf-8")) +def _write_fingerprint(repo: Path, fingerprint: dict) -> None: + (repo / ".pdd/meta/widget_python.json").write_text( + json.dumps(fingerprint, indent=2), + encoding="utf-8", + ) + + def test_issue_1932_complete_inventory_blocks_unbaselined_units(pdd_project: Path) -> None: reconcile = _pdd_json( pdd_project, "reconcile", "--json", "--strict", check=False @@ -202,6 +211,320 @@ def test_issue_1932_complete_inventory_blocks_unbaselined_units(pdd_project: Pat } +def test_issue_1996_mixed_contexts_keep_implicit_default_prompt_inventory( + pdd_project: Path, +) -> None: + (pdd_project / ".pddrc").write_text( + "contexts:\n" + " default:\n" + " paths: ['**']\n" + " defaults: {}\n" + " frontend:\n" + " paths: ['frontend/**']\n" + " defaults:\n" + " prompts_dir: prompts/frontend\n", + encoding="utf-8", + ) + frontend = pdd_project / "prompts/frontend/card_typescript.prompt" + frontend.parent.mkdir(parents=True, exist_ok=True) + frontend.write_text("Build a card.\n", encoding="utf-8") + + report = _pdd_json(pdd_project, "sync", "--dry-run", "--json", check=False) + + assert report["ok"] is False + assert {unit["basename"] for unit in report["units"]} >= { + "widget", + "unbaselined_helper", + "card", + } + + +def test_issue_1996_discovered_context_owns_configured_report_paths( + pdd_project: Path, +) -> None: + """Discovery must retain the context identity used for output resolution.""" + (pdd_project / ".pddrc").write_text( + "contexts:\n" + " default:\n" + " paths: ['**']\n" + " defaults: {}\n" + " frontend:\n" + " paths: ['frontend/**']\n" + " defaults:\n" + " prompts_dir: prompts/frontend\n" + " outputs:\n" + " code: {path: 'web/{category}/{name}.{ext}'}\n" + " example: {path: 'demo/{category}/{name}_example.{ext}'}\n" + " test: {path: 'checks/{category}/test_{name}.{ext}'}\n", + encoding="utf-8", + ) + prompt = pdd_project / "prompts/frontend/components/AssetCard_typescriptreact.prompt" + prompt.parent.mkdir(parents=True, exist_ok=True) + prompt.write_text("Build an asset card.\n", encoding="utf-8") + + report = continuous_sync.build_report(consumer="sync", root=pdd_project) + unit = next( + item for item in report["units"] + if item["paths"]["prompt"].endswith(prompt.name) + ) + + assert unit["paths"]["code"] == "web/components/AssetCard.tsx" + assert unit["paths"]["example"] == "demo/components/AssetCard_example.tsx" + assert unit["paths"]["test"] == "checks/components/test_AssetCard.tsx" + + +def test_issue_1996_read_only_resolver_uses_full_context_template_contract( + pdd_project: Path, +) -> None: + (pdd_project / ".pddrc").write_text( + "contexts:\n" + " default:\n" + " paths: ['**']\n" + " defaults: {}\n" + " frontend:\n" + " paths: ['frontend/**']\n" + " defaults:\n" + " prompts_dir: prompts/frontend\n" + " outputs:\n" + " code:\n" + " path: 'web/{category}/{name}/{name_snake}-{name_pascal}-{name_kebab}.{ext}'\n" + " example:\n" + " path: 'demo/{dir_prefix}{name}.{ext}'\n", + encoding="utf-8", + ) + prompt = pdd_project / "prompts/frontend/components/AssetCard_typescriptreact.prompt" + prompt.parent.mkdir(parents=True, exist_ok=True) + prompt.write_text("Build an asset card.\n", encoding="utf-8") + unit = continuous_sync.SyncUnit( + basename="frontend/components/AssetCard", + language="typescriptreact", + prompt_path=prompt, + prompts_dir=pdd_project / "prompts/frontend", + ) + + paths = continuous_sync._resolve_report_paths(unit, pdd_project) + + assert paths["code"] == pdd_project / "web/components/AssetCard/asset_card-Assetcard-asset-card.tsx" + assert paths["example"] == pdd_project / "demo/components/AssetCard.tsx" + + +def test_issue_1996_object_architecture_uses_context_derived_artifact_paths( + pdd_project: Path, +) -> None: + (pdd_project / ".pddrc").write_text( + "contexts:\n" + " frontend:\n" + " paths: ['frontend/**']\n" + " defaults:\n" + " prompts_dir: prompts/frontend\n" + " generate_output_path: web/src\n" + " example_output_path: web/examples\n" + " test_output_path: web/tests\n", + encoding="utf-8", + ) + nested = pdd_project / "prompts/frontend/admin" + nested.mkdir(parents=True, exist_ok=True) + prompt = nested / "page_typescriptreact.prompt" + prompt.write_text("Build admin page.\n", encoding="utf-8") + (pdd_project / "prompts/frontend/architecture.json").write_text( + json.dumps({"modules": [{"filename": "admin/page_typescriptreact.prompt", "filepath": "page.tsx"}]}), + encoding="utf-8", + ) + unit = continuous_sync.SyncUnit( + basename="frontend/admin/page", + language="typescriptreact", + prompt_path=prompt, + prompts_dir=pdd_project / "prompts/frontend", + ) + + paths = continuous_sync._resolve_report_paths(unit, pdd_project) + + assert paths["code"] == pdd_project / "prompts/frontend/web/src/page.tsx" + assert paths["example"] == pdd_project / "prompts/frontend/web/examples/page_example.tsx" + assert paths["test"] == pdd_project / "prompts/frontend/web/tests/test_page.tsx" + + +def test_issue_1996_nested_architecture_anchors_qualified_filepath_at_owner( + pdd_project: Path, +) -> None: + nested = pdd_project / "packages/store" + prompt = nested / "prompts/pages/cart_typescriptreact.prompt" + prompt.parent.mkdir(parents=True) + prompt.write_text("Build cart.\n", encoding="utf-8") + (nested / "architecture.json").write_text( + json.dumps([ + { + "filename": "pages/cart_typescriptreact.prompt", + "filepath": "src/pages/cart.tsx", + } + ]), + encoding="utf-8", + ) + unit = continuous_sync.SyncUnit( + "pages/cart", "typescriptreact", prompt, nested / "prompts" + ) + + paths = continuous_sync._resolve_report_paths(unit, pdd_project) + + assert paths["code"] == nested / "src/pages/cart.tsx" + + +def test_issue_1996_qualified_prompt_rejects_wrong_directory_architecture_leaf( + pdd_project: Path, +) -> None: + prompt = pdd_project / "prompts/bar/page_typescriptreact.prompt" + prompt.parent.mkdir(parents=True) + prompt.write_text("Build bar page.\n", encoding="utf-8") + (pdd_project / "architecture.json").write_text( + json.dumps([ + { + "filename": "foo/page_typescriptreact.prompt", + "filepath": "src/foo/page.tsx", + } + ]), + encoding="utf-8", + ) + unit = continuous_sync.SyncUnit( + "bar/page", "typescriptreact", prompt, pdd_project / "prompts" + ) + + paths = continuous_sync._resolve_report_paths(unit, pdd_project) + + assert paths["code"] == pdd_project / "bar/page.tsx" + + +def test_issue_1996_report_templates_do_not_inspect_unvalidated_external_paths( + pdd_project: Path, +) -> None: + outside = pdd_project.parent / "outside" + outside.mkdir() + (pdd_project / ".pddrc").write_text( + "contexts:\n app:\n paths: ['app/**']\n defaults:\n" + " prompts_dir: prompts/app\n outputs:\n" + f" prompt: {{path: '{outside}/{{name}}_{{language}}.prompt'}}\n" + f" test: {{path: '{outside}/test_{{name}}.{{ext}}'}}\n", + encoding="utf-8", + ) + prompt = pdd_project / "prompts/app/widget_python.prompt" + prompt.parent.mkdir(parents=True) + prompt.write_text("Build widget.\n", encoding="utf-8") + unit = continuous_sync.SyncUnit("app/widget", "python", prompt, prompt.parent) + + with patch( + "pdd.sync_main._case_insensitive_prompt_lookup", + side_effect=AssertionError("external prompt lookup occurred"), + ) as lookup: + with pytest.raises(ValueError, match="outside project"): + continuous_sync._resolve_report_paths(unit, pdd_project) + + lookup.assert_not_called() + + +def test_issue_1996_duplicate_architecture_leaves_get_distinct_derived_stems( + pdd_project: Path, +) -> None: + architecture = { + "modules": [ + {"filename": "admin/page_typescriptreact.prompt", "filepath": "app/admin/page.tsx"}, + {"filename": "settings/page_typescriptreact.prompt", "filepath": "app/settings/page.tsx"}, + ] + } + (pdd_project / "architecture.json").write_text(json.dumps(architecture), encoding="utf-8") + prompt = pdd_project / "prompts/admin/page_typescriptreact.prompt" + prompt.parent.mkdir(parents=True, exist_ok=True) + prompt.write_text("Build admin page.\n", encoding="utf-8") + unit = continuous_sync.SyncUnit( + basename="admin/page", + language="typescriptreact", + prompt_path=prompt, + prompts_dir=pdd_project / "prompts", + ) + + paths = continuous_sync._resolve_report_paths(unit, pdd_project) + + assert paths["example"] == pdd_project / "examples/app_admin_page_example.tsx" + assert paths["test"] == pdd_project / "tests/test_app_admin_page.tsx" + + +@pytest.mark.skipif(not hasattr(os, "symlink"), reason="symlink unavailable") +def test_issue_1996_nested_pddrc_symlink_fails_before_identity_read( + pdd_project: Path, +) -> None: + nested = pdd_project / "prompts/sub" + nested.mkdir() + (nested / "thing_python.prompt").write_text("Build thing.\n", encoding="utf-8") + outside = pdd_project.parent / "outside-pddrc" + outside.write_text("contexts: {}\n", encoding="utf-8") + os.symlink(outside, nested / ".pddrc") + + report = _pdd_json(pdd_project, "sync", "--dry-run", "--json", check=False) + + assert report["ok"] is False + assert any(item["failure"] == "unsafe_config" for item in report["failures"]) + assert all(item["basename"] != "sub/thing" for item in report["units"]) + + +def test_issue_1996_metadata_units_share_command_discovery_budget( + pdd_project: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr(continuous_sync, "MAX_PROMPT_DISCOVERY_FILES", 4) + meta = pdd_project / ".pdd/meta" + for index in range(8): + (meta / f"extra_{index}_python.json").write_text("{}", encoding="utf-8") + + report = continuous_sync.build_report(consumer="sync", root=pdd_project) + + assert report["ok"] is False + assert report["summary"]["total"] <= 5 + assert any(item["failure"] == "discovery_budget" for item in report["failures"]) + + +def test_issue_1996_metadata_enumeration_stops_at_remaining_budget( + pdd_project: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + meta = pdd_project / ".pdd/meta" + enumerated = 0 + real_scandir = os.scandir + + def observing_scandir(path: Path): + nonlocal enumerated + if path != meta: + return real_scandir(path) + iterator = real_scandir(path) + + class Observed: + def __enter__(self): + return self + + def __exit__(self, *_args): + iterator.close() + + def __iter__(self): + return self + + def __next__(self): + nonlocal enumerated + entry = next(iterator) + enumerated += 1 + return entry + + return Observed() + + for index in range(20): + (meta / f"candidate_{index}_python.json").write_text("{}", encoding="utf-8") + monkeypatch.setattr(continuous_sync.os, "scandir", observing_scandir) + monkeypatch.setattr(continuous_sync, "MAX_PROMPT_DISCOVERY_ENTRIES", 2) + + _identities, failure = continuous_sync._metadata_identities( + meta, pdd_project, {"entries": 0, "files": 0} + ) + + assert failure is not None and failure.failure == "discovery_budget" + assert enumerated == 3 + + @pytest.mark.parametrize( ("name", "path", "edit", "classification"), [ @@ -327,7 +650,7 @@ def test_issue_1932_incomplete_metadata_reports_failure_not_success(pdd_project: fingerprint_path = pdd_project / ".pdd/meta/widget_python.json" fingerprint = _fingerprint(pdd_project) fingerprint["code_hash"] = None - fingerprint_path.write_text(json.dumps(fingerprint, indent=2), encoding="utf-8") + _write_fingerprint(pdd_project, fingerprint) report = _pdd_json(pdd_project, "reconcile", "--json", "--strict", check=False) assert report["ok"] is False @@ -335,6 +658,198 @@ def test_issue_1932_incomplete_metadata_reports_failure_not_success(pdd_project: assert report["failures"][0]["failure"] == "incomplete_metadata" +@pytest.mark.parametrize( + ("dep_factory", "expected_reason"), + [ + ( + lambda repo: repo.parent / "outside.md", + "outside_project", + ), + ( + lambda _repo: "../outside.md", + "outside_project", + ), + ( + lambda repo: repo / "docs" / "missing.md", + "missing", + ), + ], +) +def test_issue_1996_legacy_include_deps_reject_unsafe_paths_before_hashing( + pdd_project: Path, + monkeypatch: pytest.MonkeyPatch, + dep_factory: Callable[[Path], object], + expected_reason: str, +) -> None: + dep_path = dep_factory(pdd_project) + if isinstance(dep_path, Path) and dep_path.name == "outside.md": + dep_path.write_text("outside bytes must not be read\n", encoding="utf-8") + if dep_path == "../outside.md": + (pdd_project.parent / "outside.md").write_text( + "outside bytes must not be read\n", + encoding="utf-8", + ) + + fingerprint = _fingerprint(pdd_project) + fingerprint["include_deps"] = {str(dep_path): "stored-hash"} + _write_fingerprint(pdd_project, fingerprint) + + def fail_if_hashing_reached(*_args: object, **_kwargs: object) -> dict: + raise AssertionError("unsafe include deps reached current hash calculation") + + monkeypatch.setattr( + continuous_sync, + "calculate_current_hashes", + fail_if_hashing_reached, + ) + + result = continuous_sync.classify_unit( + continuous_sync.SyncUnit( + basename="widget", + language="python", + prompt_path=pdd_project / "prompts/widget_python.prompt", + prompts_dir=pdd_project / "prompts", + ), + pdd_project, + ) + + assert result["classification"] == "FAILURE" + assert result["failure"] == "unsafe_include_deps" + assert result["unsafe_include_deps"] == [ + {"path": str(dep_path), "reason": expected_reason} + ] + + +@pytest.mark.skipif(not hasattr(os, "symlink"), reason="symlink unavailable") +def test_issue_1996_legacy_include_deps_reject_symlink_before_hashing( + pdd_project: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + target = pdd_project / "docs" / "target.md" + target.write_text("target bytes must not be read through symlink\n", encoding="utf-8") + link = pdd_project / "docs" / "link.md" + os.symlink(target, link) + + fingerprint = _fingerprint(pdd_project) + fingerprint["include_deps"] = {str(link): "stored-hash"} + _write_fingerprint(pdd_project, fingerprint) + + monkeypatch.setattr( + continuous_sync, + "calculate_current_hashes", + lambda *_args, **_kwargs: (_ for _ in ()).throw( + AssertionError("symlink include dep reached current hash calculation") + ), + ) + + result = continuous_sync.classify_unit( + continuous_sync.SyncUnit( + basename="widget", + language="python", + prompt_path=pdd_project / "prompts/widget_python.prompt", + prompts_dir=pdd_project / "prompts", + ), + pdd_project, + ) + + assert result["classification"] == "FAILURE" + assert result["failure"] == "unsafe_include_deps" + assert result["unsafe_include_deps"] == [ + {"path": str(link), "reason": "symlink"} + ] + + +@pytest.mark.skipif(not hasattr(os, "mkfifo"), reason="FIFO unavailable") +def test_issue_1996_legacy_include_deps_reject_fifo_before_hashing( + pdd_project: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + fifo = pdd_project / "docs" / "dep.fifo" + os.mkfifo(fifo) + + fingerprint = _fingerprint(pdd_project) + fingerprint["include_deps"] = {str(fifo): "stored-hash"} + _write_fingerprint(pdd_project, fingerprint) + + monkeypatch.setattr( + continuous_sync, + "calculate_current_hashes", + lambda *_args, **_kwargs: (_ for _ in ()).throw( + AssertionError("FIFO include dep reached current hash calculation") + ), + ) + + result = continuous_sync.classify_unit( + continuous_sync.SyncUnit( + basename="widget", + language="python", + prompt_path=pdd_project / "prompts/widget_python.prompt", + prompts_dir=pdd_project / "prompts", + ), + pdd_project, + ) + + assert result["classification"] == "FAILURE" + assert result["failure"] == "unsafe_include_deps" + assert result["unsafe_include_deps"] == [ + {"path": str(fifo), "reason": "nonregular"} + ] + + +@pytest.mark.parametrize("include_deps", [["docs/widget.md"], "docs/widget.md", {"docs/widget.md": 7}]) +def test_issue_1996_malformed_include_deps_report_failure_json( + pdd_project: Path, + include_deps: object, +) -> None: + fingerprint = _fingerprint(pdd_project) + fingerprint["include_deps"] = include_deps + _write_fingerprint(pdd_project, fingerprint) + + report = _pdd_json(pdd_project, "sync", "--dry-run", "--json", check=False) + + widget = next(unit for unit in report["units"] if unit["basename"] == "widget") + assert widget["classification"] == "FAILURE" + assert widget["failure"] == "unsafe_include_deps" + assert widget["metadata_valid"] is False + + +@pytest.mark.skipif(not hasattr(os, "symlink"), reason="symlink unavailable") +def test_issue_1996_symlinked_metadata_ancestor_reports_failure_without_reading( + pdd_project: Path, +) -> None: + meta = pdd_project / ".pdd" / "meta" + outside_meta = pdd_project.parent / "outside-meta" + meta.rename(outside_meta) + os.symlink(outside_meta, meta) + + report = _pdd_json(pdd_project, "sync", "--dry-run", "--json", check=False) + + assert report["ok"] is False + assert any(item["failure"] == "unsafe_metadata" for item in report["failures"]) + assert all(item["basename"] != "widget" for item in report["units"]) + + +@pytest.mark.parametrize( + "pddrc", + [ + "contexts: []\n", + "contexts:\n default:\n defaults: []\n", + "contexts: [\n", + ], +) +def test_issue_1996_invalid_pddrc_reports_discovery_failure_json( + pdd_project: Path, + pddrc: str, +) -> None: + (pdd_project / ".pddrc").write_text(pddrc, encoding="utf-8") + + report = _pdd_json(pdd_project, "sync", "--dry-run", "--json", check=False) + + assert report["ok"] is False + assert any(item["failure"] == "invalid_pddrc" for item in report["failures"]) + assert all(item["basename"] != "widget" for item in report["units"]) + + def test_issue_1932_deleted_generated_artifact_is_failure_not_in_sync( pdd_project: Path, ) -> None: @@ -357,6 +872,39 @@ def test_issue_1932_deleted_generated_artifact_is_failure_not_in_sync( assert (pdd_project / ".pdd/meta/widget_python.json").read_text(encoding="utf-8") == fingerprint_before +def test_issue_1996_symlinked_generated_artifact_is_failure_not_in_sync( + pdd_project: Path, + tmp_path: Path, +) -> None: + outside = tmp_path / "outside-widget.py" + outside.write_text( + (pdd_project / "src/widget.py").read_text(encoding="utf-8"), + encoding="utf-8", + ) + artifact = pdd_project / "src/widget.py" + artifact.unlink() + try: + artifact.symlink_to(outside) + except OSError as exc: # pragma: no cover - platform permission guard + pytest.skip(f"symlink creation unavailable: {exc}") + + report = _pdd_json( + pdd_project, + "reconcile", + "--json", + "--strict", + "--module", + "widget", + check=False, + ) + assert report["ok"] is False + assert report["summary"]["synced"] == 0 + assert report["failures"][0]["classification"] == "FAILURE" + assert report["failures"][0]["failure"] == "unsafe_artifacts" + assert report["failures"][0]["changed_files"] == ["code"] + assert "code is a symlink" in report["failures"][0]["reason"] + + def test_issue_1932_deleted_canonical_artifact_is_not_masked_by_duplicate( pdd_project: Path, ) -> None: diff --git a/tests/test_cloud_global_dry_run.py b/tests/test_cloud_global_dry_run.py new file mode 100644 index 0000000000..b0c2802783 --- /dev/null +++ b/tests/test_cloud_global_dry_run.py @@ -0,0 +1,1262 @@ +"""Regression coverage for the bounded global sync dry-run report.""" + +from __future__ import annotations + +import json +import hashlib +from pathlib import Path +from unittest.mock import patch + +from click.testing import CliRunner +import pytest + +from pdd import cli +from pdd.continuous_sync import SyncUnit, _find_matching_artifact, classify_unit +from pdd.sync_determine_operation import ( + _architecture_artifact_paths, + calculate_current_hashes, + calculate_prompt_hash, + get_pdd_file_paths, +) + + +def _write_fingerprint(project: Path, basename: str, hashes: dict) -> None: + meta = project / ".pdd" / "meta" + meta.mkdir(parents=True, exist_ok=True) + (meta / f"{basename}_python.json").write_text( + json.dumps( + { + "pdd_version": "0.0-test", + "timestamp": "2026-07-11T00:00:00+00:00", + "command": f"pdd sync {basename}", + **hashes, + } + ), + encoding="utf-8", + ) + + +def test_global_dry_run_json_discovers_absolute_configured_prompt_root_once( + tmp_path: Path, + monkeypatch, +) -> None: + """Configured absolute prompt roots are report roots, not glob patterns.""" + project = tmp_path / "project" + prompts = project / "shared-prompts" + project.mkdir() + prompts.mkdir() + (prompts / "alpha_python.prompt").write_text("alpha\n", encoding="utf-8") + (prompts / "beta_python.prompt").write_text("beta\n", encoding="utf-8") + (project / ".pddrc").write_text( + "contexts:\n" + " default:\n" + " paths: [\"**\"]\n" + " defaults:\n" + f" prompts_dir: {prompts}\n", + encoding="utf-8", + ) + monkeypatch.chdir(project) + before = sorted(path.relative_to(project) for path in project.rglob("*")) + + result = CliRunner().invoke( + cli.cli, + ["--no-core-dump", "sync", "--dry-run", "--json"], + catch_exceptions=False, + ) + + assert result.exit_code == 0, result.output + report = json.loads(result.output) + assert report["summary"]["total"] == 2 + assert {unit["basename"] for unit in report["units"]} == {"alpha", "beta"} + assert sorted(path.relative_to(project) for path in project.rglob("*")) == before + + +def test_global_dry_run_json_rejects_parent_relative_prompt_root( + tmp_path: Path, + monkeypatch, +) -> None: + """Candidate configs cannot escape to a parent workspace with relative paths.""" + workspace = tmp_path / "workspace" + project = workspace / "project" + sibling = workspace / "sibling-secret" + project.mkdir(parents=True) + sibling.mkdir() + (sibling / "secret_python.prompt").write_text("secret\n", encoding="utf-8") + (project / ".pddrc").write_text( + "contexts:\n" + " default:\n" + " paths: [\"**\"]\n" + " defaults:\n" + " prompts_dir: ..\n", + encoding="utf-8", + ) + monkeypatch.chdir(project) + + result = CliRunner().invoke( + cli.cli, + ["--no-core-dump", "sync", "--dry-run", "--json"], + catch_exceptions=False, + ) + + assert result.exit_code == 0, result.output + report = json.loads(result.output) + assert report["summary"]["failures"] == 1 + assert report["failures"][0]["failure"] == "unsafe_prompt_root" + assert all(unit.get("basename") != "secret" for unit in report["units"]) + + +def test_global_dry_run_json_rejects_sibling_relative_prompt_root( + tmp_path: Path, + monkeypatch, +) -> None: + """Candidate configs cannot point discovery at a sibling checkout.""" + workspace = tmp_path / "workspace" + project = workspace / "project" + sibling = workspace / "sibling" + project.mkdir(parents=True) + sibling.mkdir() + (sibling / "secret_python.prompt").write_text("secret\n", encoding="utf-8") + (project / ".pddrc").write_text( + "contexts:\n" + " default:\n" + " paths: [\"**\"]\n" + " defaults:\n" + " prompts_dir: ../sibling\n", + encoding="utf-8", + ) + monkeypatch.chdir(project) + + result = CliRunner().invoke( + cli.cli, + ["--no-core-dump", "sync", "--dry-run", "--json"], + catch_exceptions=False, + ) + + assert result.exit_code == 0, result.output + report = json.loads(result.output) + assert report["summary"]["failures"] == 1 + assert report["failures"][0]["failure"] == "unsafe_prompt_root" + assert all(unit.get("basename") != "secret" for unit in report["units"]) + + +def test_global_dry_run_json_rejects_unsafe_absolute_prompt_root( + tmp_path: Path, + monkeypatch, +) -> None: + """Candidate configs cannot make dry-run traverse arbitrary absolute trees.""" + project = tmp_path / "workspace" / "project" + outside = tmp_path / "outside" + project.mkdir(parents=True) + outside.mkdir() + (outside / "secret_python.prompt").write_text("secret\n", encoding="utf-8") + (project / ".pddrc").write_text( + "contexts:\n" + " default:\n" + " paths: [\"**\"]\n" + " defaults:\n" + f" prompts_dir: {outside}\n", + encoding="utf-8", + ) + monkeypatch.chdir(project) + + result = CliRunner().invoke( + cli.cli, + ["--no-core-dump", "sync", "--dry-run", "--json"], + catch_exceptions=False, + ) + + assert result.exit_code == 0, result.output + report = json.loads(result.output) + assert report["summary"]["failures"] == 1 + assert report["summary"]["total"] == 1 + assert report["failures"][0]["failure"] == "unsafe_prompt_root" + assert "outside project" in report["failures"][0]["reason"] + + +def test_global_dry_run_json_rejects_absolute_workspace_parent_prompt_root( + tmp_path: Path, + monkeypatch, +) -> None: + """An absolute parent workspace is not a trusted prompt root.""" + workspace = tmp_path / "workspace" + project = workspace / "project" + sibling = workspace / "sibling" + project.mkdir(parents=True) + sibling.mkdir() + (sibling / "secret_python.prompt").write_text("secret\n", encoding="utf-8") + (project / ".pddrc").write_text( + "contexts:\n" + " default:\n" + " paths: [\"**\"]\n" + " defaults:\n" + f" prompts_dir: {workspace}\n", + encoding="utf-8", + ) + monkeypatch.chdir(project) + + result = CliRunner().invoke( + cli.cli, + ["--no-core-dump", "sync", "--dry-run", "--json"], + catch_exceptions=False, + ) + + assert result.exit_code == 0, result.output + report = json.loads(result.output) + assert report["summary"]["failures"] == 1 + assert report["failures"][0]["failure"] == "unsafe_prompt_root" + assert all(unit.get("basename") != "secret" for unit in report["units"]) + + +def test_global_dry_run_json_reports_prompt_traversal_budget( + tmp_path: Path, + monkeypatch, +) -> None: + """Large configured prompt trees fail closed instead of hanging discovery.""" + project = tmp_path / "project" + prompts = project / "prompts" + prompts.mkdir(parents=True) + for index in range(3): + (prompts / f"unit{index}_python.prompt").write_text( + f"unit {index}\n", + encoding="utf-8", + ) + (project / ".pddrc").write_text( + "contexts:\n" + " default:\n" + " paths: [\"**\"]\n" + " defaults:\n" + " prompts_dir: prompts\n", + encoding="utf-8", + ) + monkeypatch.chdir(project) + monkeypatch.setattr("pdd.continuous_sync.MAX_PROMPT_DISCOVERY_FILES", 2) + + result = CliRunner().invoke( + cli.cli, + ["--no-core-dump", "sync", "--dry-run", "--json"], + catch_exceptions=False, + ) + + assert result.exit_code == 0, result.output + report = json.loads(result.output) + assert report["summary"]["failures"] == 1 + assert report["failures"][0]["failure"] == "prompt_traversal_budget" + + +def test_global_dry_run_json_reports_directory_entry_traversal_budget( + tmp_path: Path, + monkeypatch, +) -> None: + """Traversal budget counts directory entries, not only matching prompt files.""" + project = tmp_path / "project" + prompts = project / "prompts" + prompts.mkdir(parents=True) + for index in range(4): + (prompts / f"file{index}.txt").write_text("not a prompt\n", encoding="utf-8") + monkeypatch.chdir(project) + monkeypatch.setattr("pdd.continuous_sync.MAX_PROMPT_DISCOVERY_ENTRIES", 3) + + result = CliRunner().invoke( + cli.cli, + ["--no-core-dump", "sync", "--dry-run", "--json"], + catch_exceptions=False, + ) + + assert result.exit_code == 0, result.output + report = json.loads(result.output) + assert report["summary"]["failures"] == 1 + assert report["failures"][0]["failure"] == "prompt_traversal_budget" + + +def test_prompt_traversal_stops_scandir_at_entry_budget( + tmp_path: Path, + monkeypatch, +) -> None: + """A single wide directory is consumed only through allowance plus one.""" + project = tmp_path / "project" + prompts = project / "prompts" + prompts.mkdir(parents=True) + for index in range(8): + (prompts / f"file{index}.txt").write_text("noise\n", encoding="utf-8") + real_scandir = __import__("os").scandir + yielded = 0 + + def observing_scandir(path): + iterator = real_scandir(path) + + class Observed: + def __enter__(self): + return self + + def __exit__(self, *_args): + iterator.close() + + def __iter__(self): + return self + + def __next__(self): + nonlocal yielded + entry = next(iterator) + yielded += 1 + return entry + + return Observed() + + monkeypatch.chdir(project) + monkeypatch.setattr("pdd.continuous_sync.os.scandir", observing_scandir) + monkeypatch.setattr("pdd.continuous_sync.MAX_PROMPT_DISCOVERY_ENTRIES", 3) + + result = CliRunner().invoke( + cli.cli, + ["--no-core-dump", "sync", "--dry-run", "--json"], + catch_exceptions=False, + ) + + assert json.loads(result.output)["failures"][0]["failure"] == "prompt_traversal_budget" + assert yielded == 4 + + +def test_metadata_traversal_stops_scandir_at_entry_budget( + tmp_path: Path, monkeypatch, +) -> None: + """Metadata enumeration consumes only the remaining allowance plus one.""" + project = tmp_path / "project" + meta = project / ".pdd" / "meta" + meta.mkdir(parents=True) + for index in range(8): + (meta / f"unit{index}_python.json").write_text("{}", encoding="utf-8") + real_scandir = __import__("os").scandir + yielded = 0 + + def observing_scandir(path): + nonlocal yielded + iterator = real_scandir(path) + + class Observed: + def __enter__(self): + return self + + def __exit__(self, *_args): + iterator.close() + + def __iter__(self): + return self + + def __next__(self): + nonlocal yielded + entry = next(iterator) + if Path(path) == meta: + yielded += 1 + return entry + + return Observed() + + monkeypatch.chdir(project) + monkeypatch.setattr("pdd.continuous_sync.os.scandir", observing_scandir) + monkeypatch.setattr("pdd.continuous_sync.MAX_PROMPT_DISCOVERY_ENTRIES", 3) + + result = CliRunner().invoke( + cli.cli, ["--no-core-dump", "sync", "--dry-run", "--json"], + catch_exceptions=False, + ) + + assert json.loads(result.output)["failures"][0]["failure"] == "discovery_budget" + assert yielded == 4 + + +def test_missing_fingerprint_does_not_mask_path_resolution_failure( + tmp_path: Path, +) -> None: + """Path failures remain distinct even when no legacy fingerprint exists.""" + project = tmp_path / "project" + prompts = project / "prompts" + prompts.mkdir(parents=True) + prompt = prompts / "broken_python.prompt" + prompt.write_text("broken\n", encoding="utf-8") + unit = SyncUnit("broken", "python", prompt, prompts) + + with patch( + "pdd.continuous_sync._resolve_report_paths", + side_effect=ValueError("ambiguous module configuration"), + ): + report = classify_unit(unit, project) + + assert report["classification"] == "FAILURE" + assert report["failure"] == "path_resolution" + assert "ambiguous module configuration" in report["reason"] + + +@pytest.mark.parametrize( + "output_key", + ["generate_output_path", "example_output_path", "test_output_path"], +) +def test_missing_fingerprint_does_not_mask_unsafe_legacy_artifact( + tmp_path: Path, + output_key: str, +) -> None: + """Every resolved legacy artifact is validated before fingerprint branching.""" + project = tmp_path / "project" + prompts = project / "prompts" + prompts.mkdir(parents=True) + prompt = prompts / "widget_python.prompt" + prompt.write_text("widget\n", encoding="utf-8") + (project / ".pddrc").write_text( + "contexts:\n default:\n paths: ['**']\n defaults:\n" + " prompts_dir: prompts\n" + f" {output_key}: ../outside\n", + encoding="utf-8", + ) + + report = classify_unit(SyncUnit("widget", "python", prompt, prompts), project) + + assert report["classification"] == "FAILURE" + assert report["failure"] == "unsafe_artifacts" + assert "resolves outside project" in report["reason"] + + +def test_classify_unit_does_not_remove_concurrent_empty_directories( + tmp_path: Path, +) -> None: + """Read-only classification must not clean up directories it did not create.""" + project = tmp_path / "project" + prompts = project / "prompts" + prompts.mkdir(parents=True) + prompt = prompts / "widget_python.prompt" + prompt.write_text("widget\n", encoding="utf-8") + unit = SyncUnit("widget", "python", prompt, prompts) + + def create_concurrent_dir(*_args, **_kwargs): + (project / "concurrent-empty").mkdir() + raise ValueError("ambiguous module configuration") + + with patch( + "pdd.continuous_sync._resolve_report_paths", + side_effect=create_concurrent_dir, + ): + report = classify_unit(unit, project) + + assert report["failure"] == "path_resolution" + assert (project / "concurrent-empty").is_dir() + + +def test_missing_prompt_repair_uses_bounded_pruned_traversal( + tmp_path: Path, + monkeypatch, +) -> None: + """Stale fingerprints cannot trigger unbounded whole-project artifact search.""" + project = tmp_path / "project" + prompts = project / "prompts" + meta = project / ".pdd" / "meta" + prompts.mkdir(parents=True) + meta.mkdir(parents=True) + code = project / "node_modules" / "deep" / "widget.py" + code.parent.mkdir(parents=True) + code.write_text("value = 1\n", encoding="utf-8") + for index in range(8): + (project / f"entry-{index}.txt").write_text("noise\n", encoding="utf-8") + (meta / "widget_python.json").write_text( + json.dumps( + { + "pdd_version": "0.0-test", + "timestamp": "2026-07-11T00:00:00+00:00", + "command": "pdd sync widget", + "prompt_hash": None, + "code_hash": hashlib.sha256(code.read_bytes()).hexdigest(), + "example_hash": None, + "test_hash": None, + "test_files": None, + "include_deps": {}, + } + ), + encoding="utf-8", + ) + unit = SyncUnit("widget", "python", prompts / "widget_python.prompt", prompts) + monkeypatch.setattr("pdd.continuous_sync.MAX_REPAIR_DISCOVERY_ENTRIES", 3) + + report = classify_unit(unit, project) + + assert report["classification"] == "FAILURE" + assert report["failure"] == "repair_traversal_budget" + assert "repair search exceeded traversal budget" in report["reason"] + assert "node_modules" not in json.dumps(report["paths"]) + + +def test_repair_traversal_stops_scandir_at_entry_budget( + tmp_path: Path, + monkeypatch, +) -> None: + """Repair does not materialize a wide directory before enforcing its cap.""" + project = tmp_path / "project" + project.mkdir() + for index in range(8): + (project / f"entry-{index}.txt").write_text("noise\n", encoding="utf-8") + real_scandir = __import__("os").scandir + yielded = 0 + + def observing_scandir(path): + iterator = real_scandir(path) + + class Observed: + def __enter__(self): + return self + + def __exit__(self, *_args): + iterator.close() + + def __iter__(self): + return self + + def __next__(self): + nonlocal yielded + entry = next(iterator) + yielded += 1 + return entry + + return Observed() + + monkeypatch.setattr("pdd.continuous_sync.os.scandir", observing_scandir) + monkeypatch.setattr("pdd.continuous_sync.MAX_REPAIR_DISCOVERY_ENTRIES", 3) + + _match, failure = _find_matching_artifact(project, "widget.py", "stored") + + assert failure is not None + assert failure.failure == "repair_traversal_budget" + assert yielded == 4 + + +def test_configured_outputs_without_architecture_remain_in_sync( + tmp_path: Path, +) -> None: + project = tmp_path / "project" + prompts = project / "prompts" + prompts.mkdir(parents=True) + prompt = prompts / "widget_python.prompt" + code = project / "src" / "widget.py" + example = project / "samples" / "widget_example.py" + test_file = project / "checks" / "test_widget.py" + for path, content in ( + (prompt, "widget\n"), + (code, "value = 1\n"), + (example, "print('widget')\n"), + (test_file, "def test_widget(): pass\n"), + ): + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(content, encoding="utf-8") + (project / ".pddrc").write_text( + "contexts:\n default:\n paths: ['**']\n defaults:\n" + " prompts_dir: prompts\n generate_output_path: src/\n" + " example_output_path: samples/\n test_output_path: checks/\n", + encoding="utf-8", + ) + paths = { + "prompt": prompt, + "code": code, + "example": example, + "test": test_file, + "test_files": [test_file], + } + _write_fingerprint(project, "widget", calculate_current_hashes(paths)) + + report = classify_unit(SyncUnit("widget", "python", prompt, prompts), project) + + assert report["classification"] == "IN_SYNC" + assert report["paths"]["code"] == "src/widget.py" + assert report["paths"]["example"] == "samples/widget_example.py" + assert report["paths"]["test"] == "checks/test_widget.py" + + +def test_default_context_templates_preserve_nested_basename(tmp_path: Path) -> None: + """Read-only reports use the same default/template semantics as sync.""" + project = tmp_path / "project" + prompts = project / "prompts" / "core" + prompts.mkdir(parents=True) + prompt = prompts / "cloud_python.prompt" + code = project / "src" / "cloud.py" + example = project / "usage" / "cloud_demo.py" + test_file = project / "spec" / "cloud_spec.py" + for path, content in ((prompt, "cloud\n"), (code, "VALUE = 1\n"), + (example, "print('cloud')\n"), + (test_file, "def test_cloud(): pass\n")): + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(content, encoding="utf-8") + (project / ".pddrc").write_text( + "contexts:\n default:\n paths: ['**']\n defaults:\n" + " outputs:\n" + " code: {path: 'src/{name}.py'}\n" + " example: {path: 'usage/{name}_demo.py'}\n" + " test: {path: 'spec/{name}_spec.py'}\n", + encoding="utf-8", + ) + paths = {"prompt": prompt, "code": code, "example": example, + "test": test_file, "test_files": [test_file]} + _write_fingerprint(project, "core_cloud", calculate_current_hashes(paths)) + + report = classify_unit(SyncUnit("core/cloud", "python", prompt, prompts), project) + + assert report["classification"] == "IN_SYNC" + assert report["paths"]["code"] == "src/cloud.py" + assert report["paths"]["example"] == "usage/cloud_demo.py" + assert report["paths"]["test"] == "spec/cloud_spec.py" + + +def test_report_and_live_resolver_share_leaf_name_semantics(tmp_path: Path, monkeypatch) -> None: + project = tmp_path / "project" + prompt = project / "prompts" / "core" / "cloud_python.prompt" + prompt.parent.mkdir(parents=True) + prompt.write_text("cloud\n", encoding="utf-8") + (project / ".pddrc").write_text( + "contexts:\n default:\n paths: ['**']\n defaults:\n" + " outputs:\n code: {path: 'src/{name}.py'}\n", + encoding="utf-8", + ) + monkeypatch.chdir(project) + + report = classify_unit(SyncUnit("core/cloud", "python", prompt, project / "prompts"), project) + live = get_pdd_file_paths("core/cloud", "python", "prompts") + + assert report["paths"]["code"] == "src/cloud.py" + assert live["code"] == Path("src/cloud.py") + + +def test_live_flat_basename_uses_discovered_nested_prompt_context(tmp_path: Path, monkeypatch) -> None: + project = tmp_path / "project" + prompt = project / "prompts" / "frontend" / "app" / "page_python.prompt" + prompt.parent.mkdir(parents=True) + prompt.write_text("page\n", encoding="utf-8") + (project / ".pddrc").write_text( + "contexts:\n default:\n paths: ['**']\n defaults:\n" + " outputs:\n code: {path: 'src/{category}/{name}.py'}\n", + encoding="utf-8", + ) + monkeypatch.chdir(project) + + live = get_pdd_file_paths("page", "python", "prompts") + + assert live["code"] == Path("src/frontend/app/page.py") + + +@pytest.mark.parametrize("malformed", ["contexts: []\n", "contexts:\n local:\n defaults: []\n", "contexts:\n local:\n defaults:\n prompts_dir: []\n"]) +def test_nested_malformed_config_fails_closed(tmp_path: Path, monkeypatch, malformed: str) -> None: + project = tmp_path / "project" + prompt = project / "prompts" / "nested" / "widget_python.prompt" + prompt.parent.mkdir(parents=True) + prompt.write_text("widget\n", encoding="utf-8") + (prompt.parent / ".pddrc").write_text(malformed, encoding="utf-8") + monkeypatch.chdir(project) + + result = CliRunner().invoke(cli.cli, ["--no-core-dump", "sync", "--dry-run", "--json"], catch_exceptions=False) + report = json.loads(result.output) + + assert report["failures"][0]["failure"] == "invalid_pddrc" + + +def test_metadata_inference_is_bounded_and_validates_before_stat(tmp_path: Path, monkeypatch) -> None: + project = tmp_path / "project" + meta = project / ".pdd" / "meta" + meta.mkdir(parents=True) + safe_name = "_".join("x" for _ in range(40)) + (meta / f"{safe_name}_python.json").write_text("{}", encoding="utf-8") + monkeypatch.chdir(project) + monkeypatch.setattr("pdd.continuous_sync.MAX_METADATA_INFERENCE_CANDIDATES", 3, raising=False) + calls = [] + + def guarded_exists(path: Path, _base: Path) -> bool: + calls.append(path) + assert path.resolve(strict=False).is_relative_to(project.resolve()) + return False + + with patch("pdd.continuous_sync._validated_artifact_exists", side_effect=guarded_exists, create=True): + result = CliRunner().invoke(cli.cli, ["--no-core-dump", "sync", "--dry-run", "--json"], catch_exceptions=False) + + assert result.exit_code == 0 + assert len(calls) <= 12 + + +@pytest.mark.parametrize("target_exists", [True, False]) +def test_metadata_symlink_is_rejected_without_target_stat(tmp_path: Path, monkeypatch, target_exists: bool) -> None: + project = tmp_path / "project" + (project / ".pdd").mkdir(parents=True) + target = tmp_path / "outside-meta" + if target_exists: + target.mkdir() + try: + (project / ".pdd" / "meta").symlink_to(target, target_is_directory=True) + except OSError as exc: + pytest.skip(f"symlink creation unavailable: {exc}") + monkeypatch.chdir(project) + meta_root = project / ".pdd" / "meta" + real_exists = Path.exists + + def guarded_exists(path: Path) -> bool: + assert path != meta_root, "metadata exists() followed the symlink before lstat validation" + return real_exists(path) + + with patch.object(Path, "exists", guarded_exists): + result = CliRunner().invoke(cli.cli, ["--no-core-dump", "sync", "--dry-run", "--json"], catch_exceptions=False) + report = json.loads(result.output) + + assert report["failures"][0]["failure"] == "unsafe_metadata" + + +def test_architecture_duplicate_leaves_match_relative_prompt_path(tmp_path: Path) -> None: + """Architecture selection distinguishes path-qualified duplicate leaves.""" + project = tmp_path / "project" + prompts = project / "prompts" / "app" / "settings" + prompts.mkdir(parents=True) + prompt = prompts / "page_typescriptreact.prompt" + prompt.write_text("page\n", encoding="utf-8") + code = project / "src" / "app" / "settings" / "page.tsx" + code.parent.mkdir(parents=True) + code.write_text("export default 1\n", encoding="utf-8") + (project / "architecture.json").write_text(json.dumps([ + {"filename": "app/login/page_typescriptreact.prompt", "filepath": "src/app/login/page.tsx"}, + {"filename": "app/settings/page_typescriptreact.prompt", "filepath": "src/app/settings/page.tsx"}, + ]), encoding="utf-8") + _write_fingerprint(project, "app_settings_page", { + "prompt_hash": hashlib.sha256(prompt.read_bytes()).hexdigest(), + "code_hash": hashlib.sha256(code.read_bytes()).hexdigest(), + "example_hash": None, "test_hash": None, "test_files": None, + "include_deps": {}, + }) + (project / ".pdd" / "meta" / "app_settings_page_python.json").rename( + project / ".pdd" / "meta" / "app_settings_page_typescriptreact.json" + ) + + report = classify_unit( + SyncUnit("app/settings/page", "typescriptreact", prompt, project / "prompts"), + project, + ) + + assert report["classification"] == "IN_SYNC" + assert report["paths"]["code"] == "src/app/settings/page.tsx" + + +def test_nested_architecture_report_matches_all_live_artifact_paths( + tmp_path: Path, monkeypatch, +) -> None: + """Nested architecture ownership applies to every derived artifact.""" + project = tmp_path / "project" + owner = project / "packages" / "store" + prompts = owner / "prompts" + prompts.mkdir(parents=True) + prompt = prompts / "cart_typescriptreact.prompt" + prompt.write_text("cart\n", encoding="utf-8") + (owner / "architecture.json").write_text(json.dumps([ + {"filename": "cart_typescriptreact.prompt", "filepath": "src/pages/cart.tsx"}, + ]), encoding="utf-8") + monkeypatch.chdir(project) + + live = get_pdd_file_paths("cart", "typescriptreact", str(prompts)) + report = classify_unit( + SyncUnit("cart", "typescriptreact", prompt, prompts), project + )["paths"] + + for key in ("code", "example", "test"): + assert report[key] == live[key].relative_to(project).as_posix() + assert report["test_files"] == [ + path.relative_to(project).as_posix() for path in live["test_files"] + ] + + +def test_global_dry_run_json_rejects_symlinked_pddrc_before_read( + tmp_path: Path, monkeypatch, +) -> None: + """Candidate config links are rejected before their targets are opened.""" + project = tmp_path / "project" + project.mkdir() + outside = tmp_path / "outside.yml" + outside.write_text("contexts: {}\n", encoding="utf-8") + try: + (project / ".pddrc").symlink_to(outside) + except OSError as exc: + pytest.skip(f"symlink creation unavailable: {exc}") + monkeypatch.chdir(project) + + result = CliRunner().invoke( + cli.cli, ["--no-core-dump", "sync", "--dry-run", "--json"], + catch_exceptions=False, + ) + + report = json.loads(result.output) + assert report["failures"][0]["failure"] == "unsafe_config" + + +def test_global_discovery_budget_is_shared_across_nested_roots( + tmp_path: Path, monkeypatch, +) -> None: + """Nested configured roots cannot multiply the traversal allowance.""" + project = tmp_path / "project" + nested = project / "prompts" / "nested" + nested.mkdir(parents=True) + (nested / "unit_python.prompt").write_text("unit\n", encoding="utf-8") + (project / ".pddrc").write_text( + "contexts:\n" + " outer: {defaults: {prompts_dir: prompts}}\n" + " inner: {defaults: {prompts_dir: prompts/nested}}\n", + encoding="utf-8", + ) + monkeypatch.chdir(project) + monkeypatch.setattr("pdd.continuous_sync.MAX_PROMPT_DISCOVERY_ENTRIES", 4) + + result = CliRunner().invoke( + cli.cli, ["--no-core-dump", "sync", "--dry-run", "--json"], + catch_exceptions=False, + ) + + report = json.loads(result.output) + assert report["summary"]["total"] == 1 + assert report["failures"] == [] + + +def test_malformed_architecture_output_is_unit_scoped_json_failure( + tmp_path: Path, monkeypatch, +) -> None: + """Malformed artifact paths cannot abort the machine-readable report.""" + project = tmp_path / "project" + prompts = project / "prompts" + prompts.mkdir(parents=True) + prompt = prompts / "widget_python.prompt" + prompt.write_text("widget\n", encoding="utf-8") + (project / "architecture.json").write_text(json.dumps([ + {"filename": prompt.name, "filepath": "bad\u0000path.py"}, + ]), encoding="utf-8") + _write_fingerprint(project, "widget", { + "prompt_hash": "stored", "code_hash": "stored", "example_hash": None, + "test_hash": None, "test_files": None, "include_deps": {}, + }) + monkeypatch.chdir(project) + + result = CliRunner().invoke( + cli.cli, ["--no-core-dump", "sync", "--dry-run", "--json"], + catch_exceptions=False, + ) + + report = json.loads(result.output) + assert report["units"][0]["classification"] == "FAILURE" + assert report["units"][0]["failure"] in {"path_resolution", "unsafe_artifacts"} + + +def test_live_include_hash_is_independent_of_nested_cwd( + tmp_path: Path, + monkeypatch, +) -> None: + project = tmp_path / "project" + prompts = project / "prompts" + nested = project / "nested" + prompts.mkdir(parents=True) + nested.mkdir() + prompt = prompts / "widget_python.prompt" + dependency = prompts / "shared.txt" + prompt.write_text("shared.txt\nwidget\n", encoding="utf-8") + dependency.write_text("trusted\n", encoding="utf-8") + (nested / "shared.txt").write_text("cwd-dependent\n", encoding="utf-8") + paths = {"prompt": prompt, "code": project / "widget.py", "example": project / "examples/widget_example.py", "test": project / "tests/test_widget.py", "test_files": [project / "tests/test_widget.py"]} + for key in ("code", "example", "test"): + paths[key].parent.mkdir(parents=True, exist_ok=True) + paths[key].write_text(f"{key}\n", encoding="utf-8") + monkeypatch.chdir(project) + hashes = calculate_current_hashes(paths, dependency_root=project) + _write_fingerprint(project, "widget", hashes) + monkeypatch.chdir(nested) + + report = classify_unit(SyncUnit("widget", "python", prompt, prompts), project) + + assert report["classification"] == "IN_SYNC" + + +def test_live_include_is_validated_once_before_dependency_read( + tmp_path: Path, +) -> None: + """Hash and dependency-map generation share one validated include resolution.""" + project = tmp_path / "project" + prompts = project / "prompts" + prompts.mkdir(parents=True) + outside = tmp_path / "secret.txt" + outside.write_text("secret\n", encoding="utf-8") + prompt = prompts / "widget_python.prompt" + prompt.write_text(f"{outside}\nwidget\n", encoding="utf-8") + _write_fingerprint(project, "widget", {"prompt_hash": "stored", "code_hash": None, "example_hash": None, "test_hash": None, "test_files": None, "include_deps": {}}) + original_read_bytes = Path.read_bytes + outside_reads = 0 + + def observing_read_bytes(path: Path) -> bytes: + nonlocal outside_reads + if path == outside: + outside_reads += 1 + return original_read_bytes(path) + + with patch.object(Path, "read_bytes", observing_read_bytes): + report = classify_unit(SyncUnit("widget", "python", prompt, prompts), project) + + assert report["failure"] == "hash_calculation" + assert outside_reads == 0 + + +def test_unresolved_live_include_preserves_v1_skip_without_ancestor_read( + tmp_path: Path, +) -> None: + """Safe report resolution preserves v1 missing-dependency semantics.""" + project = tmp_path / "workspace" / "project" + prompts = project / "prompts" + prompts.mkdir(parents=True) + ancestor = project.parent / "ancestor-only.txt" + ancestor.write_text("outside\n", encoding="utf-8") + prompt = prompts / "widget_python.prompt" + prompt.write_text("ancestor-only.txt\n", encoding="utf-8") + reads = 0 + original = Path.read_bytes + + def observing(path: Path) -> bytes: + nonlocal reads + if path == ancestor: + reads += 1 + return original(path) + + with patch.object(Path, "read_bytes", observing): + hashes = calculate_current_hashes({"prompt": prompt}, dependency_root=project) + + assert hashes["prompt_hash"] == hashlib.sha256(prompt.read_bytes()).hexdigest() + assert reads == 0 + + +def test_validated_v1_hash_preserves_declared_alias_multiplicity(tmp_path: Path) -> None: + """v1 deduplicates declared strings, not their resolved path aliases.""" + project = tmp_path / "project" + project.mkdir() + dependency = project / "dep.txt" + dependency.write_text("dependency\n", encoding="utf-8") + prompt = project / "widget_python.prompt" + prompt.write_text( + "dep.txt\n./dep.txt\n", + encoding="utf-8", + ) + + legacy = calculate_prompt_hash(prompt, dependency_root=project, hash_version=1) + validated = calculate_current_hashes( + {"prompt": prompt}, dependency_root=project + )["prompt_hash"] + + assert validated == legacy + + +def test_report_caches_config_ownership_and_caps_contexts( + tmp_path: Path, monkeypatch, +) -> None: + """One config is parsed once and context ownership has a hard command cap.""" + project = tmp_path / "project" + prompts = project / "prompts" + prompts.mkdir(parents=True) + for name in ("alpha", "beta"): + (prompts / f"{name}_python.prompt").write_text(name, encoding="utf-8") + (project / ".pddrc").write_text( + "contexts:\n default:\n defaults:\n prompts_dir: prompts\n", + encoding="utf-8", + ) + monkeypatch.chdir(project) + from pdd import continuous_sync + original = continuous_sync._load_pddrc_config + loads = 0 + + def observing(path): + nonlocal loads + loads += 1 + return original(path) + + monkeypatch.setattr(continuous_sync, "_load_pddrc_config", observing) + first = CliRunner().invoke( + cli.cli, ["--no-core-dump", "sync", "--dry-run", "--json"], + catch_exceptions=False, + ) + assert json.loads(first.output)["summary"]["total"] == 2 + assert loads == 1 + + monkeypatch.setattr(continuous_sync, "MAX_CONFIG_CONTEXTS", 1, raising=False) + (project / ".pddrc").write_text( + "contexts:\n one: {defaults: {prompts_dir: prompts}}\n" + " two: {defaults: {prompts_dir: prompts}}\n", + encoding="utf-8", + ) + second = CliRunner().invoke( + cli.cli, ["--no-core-dump", "sync", "--dry-run", "--json"], + catch_exceptions=False, + ) + assert json.loads(second.output)["failures"][0]["failure"] == "invalid_pddrc" + + +def test_metadata_only_report_reuses_parsed_config( + tmp_path: Path, monkeypatch, +) -> None: + """Metadata inference and classification share the report config cache.""" + project = tmp_path / "project" + meta = project / ".pdd" / "meta" + meta.mkdir(parents=True) + for name in ("alpha", "beta"): + (meta / f"{name}_python.json").write_text("{}", encoding="utf-8") + (project / ".pddrc").write_text( + "contexts:\n default:\n defaults:\n prompts_dir: prompts\n", + encoding="utf-8", + ) + monkeypatch.chdir(project) + from pdd import continuous_sync + original = continuous_sync._load_pddrc_config + loads = 0 + + def observing(path): + nonlocal loads + loads += 1 + return original(path) + + monkeypatch.setattr(continuous_sync, "_load_pddrc_config", observing) + + result = CliRunner().invoke( + cli.cli, + ["--no-core-dump", "sync", "--dry-run", "--json"], + catch_exceptions=False, + ) + + assert json.loads(result.output)["summary"]["total"] == 2 + assert loads == 1 + + +def test_duplicate_prompt_roots_cannot_bypass_context_cap_without_prompts( + tmp_path: Path, monkeypatch, +) -> None: + """Context validation runs before duplicate prompt roots are collapsed.""" + project = tmp_path / "project" + project.mkdir() + (project / ".pddrc").write_text( + "contexts:\n" + " one: {defaults: {prompts_dir: prompts}}\n" + " two: {defaults: {prompts_dir: prompts}}\n", + encoding="utf-8", + ) + monkeypatch.chdir(project) + monkeypatch.setattr("pdd.continuous_sync.MAX_CONFIG_CONTEXTS", 1) + + result = CliRunner().invoke( + cli.cli, + ["--no-core-dump", "sync", "--dry-run", "--json"], + catch_exceptions=False, + ) + + assert json.loads(result.output)["failures"][0]["failure"] == "invalid_pddrc" + + +def test_architecture_artifact_construction_performs_no_discovery( + tmp_path: Path, +) -> None: + """Architecture path construction is pure until report validation.""" + project = tmp_path / "project" + + with patch.object(Path, "exists", side_effect=AssertionError("exists called")), \ + patch.object(Path, "glob", side_effect=AssertionError("glob called")): + paths = _architecture_artifact_paths( + project, Path("src/widget.py"), "widget", "py" + ) + + assert paths["test_files"] == [project / "tests" / "test_widget.py"] + + +def test_architecture_test_output_is_validated_before_discovery( + tmp_path: Path, monkeypatch, +) -> None: + """Outside architecture test directories are rejected before scandir.""" + project = tmp_path / "project" + prompts = project / "prompts" + prompts.mkdir(parents=True) + prompt = prompts / "widget_python.prompt" + prompt.write_text("widget\n", encoding="utf-8") + (project / "architecture.json").write_text( + json.dumps([{"filename": prompt.name, "filepath": "src/widget.py"}]), + encoding="utf-8", + ) + outside = tmp_path / "outside-tests" + (project / ".pddrc").write_text( + "contexts:\n default:\n defaults:\n" + " prompts_dir: prompts\n" + f" test_output_path: {outside}\n", + encoding="utf-8", + ) + original_scandir = __import__("os").scandir + + def guarded_scandir(path): + assert Path(path) != outside + return original_scandir(path) + + monkeypatch.setattr("pdd.continuous_sync.os.scandir", guarded_scandir) + + report = classify_unit(SyncUnit("widget", "python", prompt, prompts), project) + + assert report["classification"] == "FAILURE" + assert report["failure"] == "unsafe_artifacts" + + +def test_symlinked_architecture_test_output_is_validated_before_discovery( + tmp_path: Path, monkeypatch, +) -> None: + """Symlink-directed architecture test directories are never enumerated.""" + project = tmp_path / "project" + prompts = project / "prompts" + prompts.mkdir(parents=True) + prompt = prompts / "widget_python.prompt" + prompt.write_text("widget\n", encoding="utf-8") + (project / "architecture.json").write_text( + json.dumps([{"filename": prompt.name, "filepath": "src/widget.py"}]), + encoding="utf-8", + ) + outside = tmp_path / "outside-tests" + outside.mkdir() + linked = project / "linked-tests" + try: + linked.symlink_to(outside, target_is_directory=True) + except OSError as exc: + pytest.skip(f"symlink creation unavailable: {exc}") + (project / ".pddrc").write_text( + "contexts:\n default:\n defaults:\n" + " prompts_dir: prompts\n" + " test_output_path: linked-tests\n", + encoding="utf-8", + ) + original_scandir = __import__("os").scandir + + def guarded_scandir(path): + assert Path(path) != linked + return original_scandir(path) + + monkeypatch.setattr("pdd.continuous_sync.os.scandir", guarded_scandir) + + report = classify_unit(SyncUnit("widget", "python", prompt, prompts), project) + + assert report["classification"] == "FAILURE" + assert report["failure"] == "unsafe_artifacts" + + +def test_architecture_test_discovery_uses_shared_budget( + tmp_path: Path, monkeypatch, +) -> None: + """Wide architecture test directories stop at the command budget.""" + project = tmp_path / "project" + prompts = project / "prompts" + tests = project / "tests" + prompts.mkdir(parents=True) + tests.mkdir() + prompt = prompts / "widget_python.prompt" + prompt.write_text("widget\n", encoding="utf-8") + (project / "architecture.json").write_text( + json.dumps([{"filename": prompt.name, "filepath": "src/widget.py"}]), + encoding="utf-8", + ) + for index in range(4): + (tests / f"test_widget_{index}.py").write_text("pass\n", encoding="utf-8") + monkeypatch.setattr("pdd.continuous_sync.MAX_REPAIR_DISCOVERY_ENTRIES", 2) + budget = {"repair_entries": 0} + + report = classify_unit( + SyncUnit("widget", "python", prompt, prompts), project, budget + ) + + assert report["classification"] == "FAILURE" + assert report["failure"] == "path_resolution" + assert budget["report_entries"] == 3 + + +@pytest.mark.parametrize("include_kind", ["absolute", "symlink"]) +def test_live_include_rejects_unsafe_target( + tmp_path: Path, + include_kind: str, +) -> None: + project = tmp_path / "project" + prompts = project / "prompts" + prompts.mkdir(parents=True) + outside = tmp_path / "secret.txt" + outside.write_text("secret\n", encoding="utf-8") + if include_kind == "absolute": + reference = str(outside) + else: + link = project / "linked.txt" + try: + link.symlink_to(outside) + except OSError as exc: + pytest.skip(f"symlink creation unavailable: {exc}") + reference = "linked.txt" + prompt = prompts / "widget_python.prompt" + prompt.write_text(f"{reference}\nwidget\n", encoding="utf-8") + _write_fingerprint(project, "widget", {"prompt_hash": "stored", "code_hash": None, "example_hash": None, "test_hash": None, "test_files": None, "include_deps": {}}) + + report = classify_unit(SyncUnit("widget", "python", prompt, prompts), project) + + assert report["classification"] == "FAILURE" + assert report["failure"] == "hash_calculation" + + +def test_symlinked_architecture_is_rejected_before_read( + tmp_path: Path, +) -> None: + project = tmp_path / "project" + prompts = project / "prompts" + prompts.mkdir(parents=True) + prompt = prompts / "widget_python.prompt" + prompt.write_text("widget\n", encoding="utf-8") + outside = tmp_path / "architecture.json" + outside.write_text("[]", encoding="utf-8") + try: + (project / "architecture.json").symlink_to(outside) + except OSError as exc: + pytest.skip(f"symlink creation unavailable: {exc}") + + report = classify_unit(SyncUnit("widget", "python", prompt, prompts), project) + + assert report["classification"] == "FAILURE" + assert report["failure"] == "unsafe_architecture" + + +@pytest.mark.parametrize("value", ["[]", "7", "true", "' '"]) +def test_invalid_prompts_dir_value_emits_invalid_pddrc_json( + tmp_path: Path, + monkeypatch, + value: str, +) -> None: + project = tmp_path / "project" + project.mkdir() + (project / ".pddrc").write_text( + "contexts:\n default:\n defaults:\n prompts_dir: " + value + "\n", + encoding="utf-8", + ) + monkeypatch.chdir(project) + + result = CliRunner().invoke(cli.cli, ["--no-core-dump", "sync", "--dry-run", "--json"], catch_exceptions=False) + + report = json.loads(result.output) + assert report["failures"][0]["failure"] == "invalid_pddrc" + + +def test_unexpandable_prompts_dir_emits_invalid_pddrc_json( + tmp_path: Path, + monkeypatch, +) -> None: + project = tmp_path / "project" + project.mkdir() + (project / ".pddrc").write_text( + "contexts:\n default:\n defaults:\n prompts_dir: ~pdd_user_that_does_not_exist/prompts\n", + encoding="utf-8", + ) + monkeypatch.chdir(project) + + result = CliRunner().invoke(cli.cli, ["--no-core-dump", "sync", "--dry-run", "--json"], catch_exceptions=False) + + report = json.loads(result.output) + assert report["failures"][0]["failure"] == "invalid_pddrc" + + +def test_malformed_string_include_path_is_unit_scoped_failure(tmp_path: Path) -> None: + project = tmp_path / "project" + prompts = project / "prompts" + prompts.mkdir(parents=True) + prompt = prompts / "widget_python.prompt" + prompt.write_text("widget\n", encoding="utf-8") + _write_fingerprint(project, "widget", {"prompt_hash": "stored", "code_hash": None, "example_hash": None, "test_hash": None, "test_files": None, "include_deps": {"bad\u0000path": "digest"}}) + + report = classify_unit(SyncUnit("widget", "python", prompt, prompts), project) + + assert report["failure"] == "unsafe_include_deps" + assert report["unsafe_include_deps"] == [{"path": "bad\u0000path", "reason": "invalid_path"}] diff --git a/tests/test_continuous_sync_path_policy.py b/tests/test_continuous_sync_path_policy.py new file mode 100644 index 0000000000..dc1efe0a95 --- /dev/null +++ b/tests/test_continuous_sync_path_policy.py @@ -0,0 +1,44 @@ +"""Path-policy regressions for legacy continuous-sync metadata repair.""" + +from pathlib import Path + +import pytest + +from pdd import continuous_sync + + +def test_artifact_path_violation_rejects_resolved_project_escape(tmp_path: Path) -> None: + root = tmp_path / "repo" + root.mkdir() + outside = tmp_path / "outside.py" + outside.write_text("def value():\n return 1\n", encoding="utf-8") + + assert ( + continuous_sync._artifact_path_violation(root / "../outside.py", root) + == "resolves outside project" + ) + + +def test_repair_search_skips_symlink_candidate_without_hashing( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + root = tmp_path / "repo" + root.mkdir() + archive = root / "archive" + archive.mkdir() + outside = tmp_path / "widget.py" + outside.write_text("def value():\n return 1\n", encoding="utf-8") + try: + (archive / "widget.py").symlink_to(outside) + except OSError as exc: # pragma: no cover - platform permission guard + pytest.skip(f"symlink creation unavailable: {exc}") + + def fail_if_hashed(path: Path) -> str: + raise AssertionError(f"unexpected hash read: {path}") + + monkeypatch.setattr(continuous_sync, "calculate_sha256", fail_if_hashed) + + match, failure = continuous_sync._find_matching_artifact(root, "widget.py", "unused") + assert match is None + assert failure is None diff --git a/tests/test_sync_core_fingerprint_store.py b/tests/test_sync_core_fingerprint_store.py index 9c5458957b..3fb43583be 100644 --- a/tests/test_sync_core_fingerprint_store.py +++ b/tests/test_sync_core_fingerprint_store.py @@ -158,6 +158,21 @@ def test_legacy_record_is_readable_but_not_promoted(tmp_path) -> None: assert store.load(UNIT_ID) is None +def test_legacy_record_symlink_is_rejected_before_target_read(tmp_path) -> None: + store = _store(tmp_path) + outside = tmp_path.parent / "outside-fingerprint.json" + outside.write_text("{not-json", encoding="utf-8") + legacy_path = tmp_path / ".pdd/meta/widget_python.json" + legacy_path.parent.mkdir(parents=True, exist_ok=True) + try: + legacy_path.symlink_to(outside) + except OSError as exc: # pragma: no cover - platform permission guard + pytest.skip(f"symlink creation unavailable: {exc}") + + with pytest.raises(FingerprintStoreError, match="not a regular file"): + store.read_legacy(legacy_path) + + def test_embedded_identity_mismatch_is_corrupt(tmp_path) -> None: store = _store(tmp_path) path = store.write(_record()) diff --git a/tests/test_sync_core_supervisor.py b/tests/test_sync_core_supervisor.py index 3d76375d47..641bfa1014 100644 --- a/tests/test_sync_core_supervisor.py +++ b/tests/test_sync_core_supervisor.py @@ -11,6 +11,7 @@ import pytest +from pdd.sync_core import supervisor from pdd.sync_core.supervisor import ( SupervisorLimits, _linked_libraries, @@ -23,6 +24,23 @@ ) +def test_runtime_closure_ignores_synthetic_argv_interpreter_prefix( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Identity-only argv prefixes must never become measured or mounted paths.""" + actual_executable = Path(sys.executable).resolve() + supervisor.released_runtime_closure_paths.cache_clear() + try: + monkeypatch.setattr( + "pdd.sync_core.runner.sys.executable", "/venv-a/bin/python" + ) + closure = dict(supervisor.released_runtime_closure_paths()) + assert closure["interpreter/python"] == actual_executable + assert closure["interpreter/python"].is_file() + finally: + supervisor.released_runtime_closure_paths.cache_clear() + + def test_runtime_directories_collapse_nested_but_keep_disjoint_roots( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: @@ -121,7 +139,10 @@ def test_sandbox_binds_resolved_runtime_sources_at_original_destinations( workdir.mkdir() monkeypatch.setattr(sys, "platform", "linux") - monkeypatch.setattr(sys, "executable", str(executable_destination)) + monkeypatch.setattr( + "pdd.sync_core.supervisor._SUPERVISOR_EXECUTABLE", + executable_destination, + ) sandbox_tools = { "bwrap": "/usr/bin/bwrap", "sudo": "/usr/bin/sudo", diff --git a/tests/test_sync_determine_operation.py b/tests/test_sync_determine_operation.py index 96c0c60330..250b5168cf 100644 --- a/tests/test_sync_determine_operation.py +++ b/tests/test_sync_determine_operation.py @@ -4272,6 +4272,36 @@ def test_calculate_prompt_hash_detects_dep_change_via_stored_deps(self, pdd_test "even when the prompt itself has no tags" ) + def test_calculate_prompt_hash_anchors_relative_stored_deps_to_explicit_root( + self, pdd_test_environment, monkeypatch + ): + """Stored relative dependency keys must not be interpreted from process CWD.""" + prompts_dir = pdd_test_environment / "prompts" + prompt_path = prompts_dir / f"{BASENAME}_{LANGUAGE}.prompt" + create_file(prompt_path, "Create a helper using project docs.\n") + project_dep = pdd_test_environment / "docs" / "contract.md" + create_file(project_dep, "trusted project dependency\n") + + nested = pdd_test_environment / "nested" + alternate_dep = nested / "docs" / "contract.md" + create_file(alternate_dep, "wrong nested dependency\n") + monkeypatch.chdir(nested) + + stored_deps = {"docs/contract.md": calculate_sha256(project_dep)} + anchored_hash = calculate_prompt_hash( + prompt_path, + stored_deps=stored_deps, + dependency_root=pdd_test_environment, + ) + create_file(alternate_dep, "changed wrong nested dependency\n") + anchored_hash_after_alternate_change = calculate_prompt_hash( + prompt_path, + stored_deps=stored_deps, + dependency_root=pdd_test_environment, + ) + + assert anchored_hash == anchored_hash_after_alternate_change + def test_fingerprint_stores_include_deps(self, pdd_test_environment): """Fingerprint dataclass should correctly store and serialize include_deps.""" fp = Fingerprint(