_test_filter.py

"""Standalone test-path filtering logic. No pytest dependencies."""

from __future__ import annotations

import ast
import datetime
import enum
import fnmatch
import json
import logging
import re
import subprocess
import warnings
from pathlib import Path
from typing import TYPE_CHECKING

import pathspec

if TYPE_CHECKING:
    from typing import Any


class FilterMode(enum.StrEnum):
    NONE = "none"
    CONSERVATIVE = "conservative"
    AGGRESSIVE = "aggressive"


class FullRunReason(enum.StrEnum):
    DISABLED = "disabled"
    GIT_UNAVAILABLE = "git_unavailable"
    LARGE_CHANGESET = "large_changeset"
    BUCKET_A = "bucket_a"
    UNMAPPED_FILE = "unmapped_file"


class ImportContext(enum.StrEnum):
    TOP_LEVEL = "top_level"
    CONDITIONAL = "conditional"
    TYPE_CHECKING = "type_checking"
    DEFERRED = "deferred"
    IMPORTLIB = "importlib"


BUCKET_A_PATTERNS: frozenset[str] = frozenset(
    {
        "tests/conftest.py",
        "tests/_helpers.py",
        "tests/_arch_constraint_discovery.py",
        "tests/arch/_helpers.py",
        "tests/arch/_rules.py",
        "pyproject.toml",
        "uv.lock",
        ".pre-commit-config.yaml",
        "src/autoskillit/server/_factory.py",
    }
)

BUCKET_A_GLOBS: tuple[str, ...] = ("tests/*/conftest.py",)

# Matches lines that only change a version string: -version = "0.9.x" / +version = "0.9.y"
_VERSION_LINE_RE: re.Pattern[str] = re.compile(r'^[+-]version\s*=\s*"[^"]*"', re.IGNORECASE)

# Matches removed lines that are class/function definitions or UPPER_CASE constant assignments.
# Used by _is_additive_only to detect breaking changes.
_BREAKING_DEF_RE: re.Pattern[str] = re.compile(
    r"^-\s*(class |def |[A-Z_][A-Z0-9_]*\s*[=:])", re.MULTILINE
)

ALWAYS_RUN_CONSERVATIVE: frozenset[str] = frozenset(
    {
        "arch",
        "contracts",
        "infra",
        "docs",
    }
)

ALWAYS_RUN_AGGRESSIVE: frozenset[str] = frozenset(
    {
        "arch",
        "contracts",
    }
)

# ---------------------------------------------------------------------------
# Tiered always-run configuration (conservative mode)
# ---------------------------------------------------------------------------

# Structural infra files that run unconditionally regardless of trigger conditions.
# These enforce cross-cutting registration, executability, and schema contracts.
_INFRA_UNCONDITIONAL_FILES: frozenset[str] = frozenset(
    {
        "test_manifest_completeness.py",
        "test_manifest_directory_completeness.py",
        "test_guard_coverage.py",
        "test_session_scope_enforcement.py",
        "test_filter_activation.py",
        "test_schema_version_convention.py",
        "test_release_sanity.py",
        "test_background_exec_guard.py",
        "test_adr_runtime_guard_coverage.py",
    }
)

# Structural hook files that run unconditionally regardless of trigger conditions.
# These enforce cross-cutting hook registration and executability contracts.
_HOOKS_UNCONDITIONAL_FILES: frozenset[str] = frozenset(
    {
        "test_hook_executability.py",
        "test_hook_registration_coverage.py",
        "test_hook_registry.py",
    }
)

# Conditions that trigger inclusion of the full infra/ directory.
_INFRA_HOOK_TRIGGER_PREFIX: str = "src/autoskillit/hooks/"
_INFRA_CI_TRIGGER_PREFIX: str = ".github/"
_INFRA_CI_TRIGGER_FILES: frozenset[str] = frozenset(
    {
        "pyproject.toml",
        ".pre-commit-config.yaml",
        "Taskfile.yml",
    }
)

# Conditions that trigger inclusion of the docs/ directory.
_DOCS_TRIGGER_PREFIX: str = "docs/"
_DOCS_TRIGGER_FILES: frozenset[str] = frozenset({"README.md", "CLAUDE.md", "AGENTS.md"})

# Tier-1 unconditional always-run for conservative mode: arch+contracts only.
# Decoupled from ALWAYS_RUN_AGGRESSIVE so future additions to that constant
# cannot silently alter conservative behavior.
_ALWAYS_RUN_CONSERVATIVE_UNCONDITIONAL: frozenset[str] = frozenset({"arch", "contracts"})

_LARGE_CHANGESET_THRESHOLD_CONSERVATIVE: int = 30

# ---------------------------------------------------------------------------
# core/ module-level cascade classification
# ---------------------------------------------------------------------------

# Modules in this set trigger a full 17-directory cascade. A core/ module belongs
# here if:
# (a) it is imported at the top level of tests/conftest.py, OR
# (b) its types have concrete implementations in tests/fakes.py (protocol satisfaction)
#     AND fakes.py is imported at module level by test files across most directories.
# If neither condition holds, use MODULE_CASCADE_CORE with an explicit narrower cascade.
_CORE_UNIVERSAL_MODULES: frozenset[str] = frozenset(
    {
        "io",
        "logging",
        "types",
        "_type_constants",
        "_type_protocols_logging",
        "_type_protocols_execution",
        "_type_protocols_github",
        "_type_protocols_recipe",
        "_type_protocols_infra",
        "_type_enums",
        "_type_subprocess",
    }
)

# Maps universal module stems to test dirs safe to exclude when changes are additive-only.
# Only consulted when stem is in _CORE_UNIVERSAL_MODULES and diff passes _is_additive_only.
_CORE_UNIVERSAL_EXCLUSIONS: dict[str, frozenset[str]] = {
    "_type_enums": frozenset({"hooks", "skills"}),
}

MODULE_CASCADE_CORE: dict[str, frozenset[str]] = {
    "_cmd_runner": frozenset({"cli", "core", "recipe", "_probe_canary"}),
    "_json": frozenset({"core", "execution", "pipeline", "recipe", "server"}),
    "feature_flags": frozenset(
        {"core", "cli", "config", "execution", "recipe", "server", "workspace"}
    ),
    "branch_guard": frozenset({"core", "pipeline", "server", "workspace"}),
    "_plugin_ids": frozenset({"core", "cli", "hook_registry", "server"}),
    "_terminal_table": frozenset({"core", "cli", "pipeline", "recipe"}),
    "_plugin_cache": frozenset({"core", "cli", "server"}),
    "git_remote": frozenset({"core", "execution"}),
    "github_url": frozenset({"core", "cli", "execution", "fleet", "server"}),
    "paths": frozenset(
        {
            "core",
            "cli",
            "config",
            "execution",
            "fleet",
            "hook_registry",
            "migration",
            "recipe",
            "report",
            "server",
            "workspace",
            "infra/test_generated_file_write_guard.py",
            "infra/test_generated_files.py",
        }
    ),
    "_claude_env": frozenset({"core", "execution", "_llm_triage", "cli"}),
    "_version_snapshot": frozenset({"core", "execution"}),
    "claude_conventions": frozenset({"core", "execution", "server", "workspace"}),
    "_type_resume": frozenset({"core", "cli", "execution", "fleet"}),
    "_type_helpers": frozenset({"core", "execution", "fleet", "pipeline", "recipe", "server"}),
    "_type_protocols_workspace": frozenset({"core", "pipeline", "recipe", "workspace"}),
    "_type_protocols_backend": frozenset(
        {"_llm_triage", "cli", "core", "execution", "pipeline", "server", "workspace"}
    ),
    "_type_inspector": frozenset({"core", "execution"}),
    "_type_invariant_registry": frozenset({"core"}),
    "_install_detect": frozenset({"core", "cli", "config"}),
    "_linux_proc": frozenset({"core", "execution", "fleet", "cli"}),
    "_type_plugin_source": frozenset({"core", "execution", "pipeline", "server", "cli"}),
    "kitchen_state": frozenset({"core", "server"}),
    "session_provenance": frozenset({"core", "execution"}),
    "readiness": frozenset({"core", "server"}),
    "session_registry": frozenset({"core"}),
    "tool_sequence_analysis": frozenset({"core", "execution", "server", "cli"}),
    "_type_checkpoint": frozenset({"core", "execution", "fleet", "server"}),
    "_type_results": frozenset(
        {
            "core",
            "execution",
            "pipeline",
            "workspace",
            "recipe",
            "migration",
            "fleet",
            "server",
            "cli",
            "_llm_triage",
            "_test_filter",
            "hook_registry",
            "smoke_utils",
        }
    ),
    "_type_results_execution": frozenset({"core", "execution", "server", "pipeline"}),
    "_type_backend": frozenset({"core", "execution", "cli", "recipe", "workspace"}),
    "_type_dispatch_identity": frozenset({"core", "fleet", "execution"}),
    "_type_figure_spec": frozenset({"core", "report"}),
    "_type_session_env": frozenset({"core", "cli"}),
    "_type_capture": frozenset({"core", "fleet", "recipe", "cli"}),
    "_type_token": frozenset({"core", "execution"}),
    "_type_constants_env": frozenset(
        {"cli", "config", "core", "execution", "recipe", "server", "smoke_utils", "workspace"}
    ),
    "_type_constants_features": frozenset(
        {"cli", "config", "core", "fleet", "recipe", "server", "workspace"}
    ),
    "_type_constants_registries": frozenset(
        {"cli", "config", "core", "pipeline", "recipe", "server", "workspace"}
    ),
    "_type_exceptions": frozenset({"core", "execution", "fleet", "recipe", "server"}),
    "_type_phoropter": frozenset({"core"}),
    "_type_tradition_manifest": frozenset({"core"}),
    "_step_context": frozenset({"core", "execution", "pipeline", "server"}),
    "_execution_marker": frozenset({"core", "execution", "fleet", "server"}),
    "bash_write_targets": frozenset({"core", "execution", "server"}),
}

# Narrow per-module cascade for execution/. Modules not listed here fall through
# to LAYER_CASCADE_CONSERVATIVE["execution"] (fail-open).
MODULE_CASCADE_EXECUTION: dict[str, frozenset[str]] = {
    # --- Narrowest: imports only within tests/execution/ ---
    "anomaly_detection": frozenset({"execution"}),
    "clone_guard": frozenset({"execution"}),
    # --- Medium: execution + specific file-level consumers ---
    "ci": frozenset({"execution"}),
    "diff_annotator": frozenset({"execution", "test_smoke_utils.py"}),
    "pr_analysis": frozenset({"execution", "test_smoke_utils.py"}),
    "testing": frozenset(
        {
            "execution",
            "server/test_factory.py",
            "server/test_server_tool_registration.py",
            "contracts/test_protocol_satisfaction.py",
        }
    ),
    "db": frozenset(
        {
            "execution",
            "server/test_factory.py",
            "contracts/test_l1_packages.py",
            "contracts/test_protocol_satisfaction.py",
            "contracts/test_package_gateways.py",
        }
    ),
    "recording": frozenset(
        {
            "execution",
            "server/test_factory_recording.py",
            "server/test_state.py",
            "server/test_lifespan.py",
            "server/test_tools_run_cmd_unit.py",
        }
    ),
    "github": frozenset(
        {
            "execution",
            "server/test_factory.py",
            "server/test_tools_status_kitchen.py",
        }
    ),
    "remote_resolver": frozenset(
        {
            "execution",
            "server/test_tools_ci.py",
            "workspace/test_clone_ci_contract.py",
        }
    ),
    "quota": frozenset({"execution", "server", "cli/test_doctor.py"}),
    "session_log": frozenset(
        {
            "execution",
            "server",
            "cli",
            "hooks/test_quota_check.py",
        }
    ),
    "linux_tracing": frozenset(
        {
            "execution",
            "cli",
            "fleet/test_fleet_e2e.py",
        }
    ),
    "commands": frozenset(
        {
            "execution",
            "server/test_tools_execution_command.py",
            "cli/test_cook_order_command.py",
            "cli/test_cook_interactive.py",
            "cli/test_cook_env_scrub.py",
            "cli/test_reload_loop.py",
        }
    ),
}

SUBPKG_CASCADE_EXECUTION: dict[str, frozenset[str]] = {
    "merge_queue": frozenset({"execution"}),
    "headless": frozenset(
        {
            "execution",
            "core",
            "workspace",
            "migration",
            "server",
            "cli",
            "infra/test_pretty_output_hook_infra.py",
            "skills/test_skill_output_compliance.py",
            "_llm_triage",
            "smoke_utils",
        }
    ),
    "process": frozenset(
        {
            "execution",
            "core",
            "workspace",
            "migration",
            "server",
            "cli",
            "infra/test_pretty_output_hook_infra.py",
            "skills/test_skill_output_compliance.py",
            "_llm_triage",
            "smoke_utils",
        }
    ),
    "backends": frozenset({"execution", "integration"}),
    "session": frozenset(
        {
            "execution",
            "server",
            "migration/test_engine.py",
            "test_llm_triage.py",
            "cli/test_session_launch.py",
            "cli/test_order_resume.py",
            "cli/test_cook_ide_isolation.py",
            "cli/test_cook_env_scrub.py",
            "cli/test_terminal.py",
            "cli/test_fleet_session.py",
            "cli/test_reload_loop.py",
            "fleet/test_result_parser.py",
            "fleet/test_dispatch_outcome_classifier.py",
            "fleet/test_dispatch_outcome_classifier_timeout.py",
            "fleet/test_dispatch_lifespan.py",
            "fleet/test_dispatch_state_handle.py",
            "fleet/test_research_campaign_dispatch.py",
            "fleet/test_campaign_capture.py",
            "fleet/test_dispatch_failure_semantics.py",
            "fleet/test_dispatch_stderr_forwarding.py",
            "fleet/test_dispatch_envelope_fields.py",
            "fleet/test_dispatch_labels_cleaned.py",
            "fleet/test_resume_preflight.py",
            "fleet/test_helpers_exports.py",
        }
    ),
}

MODULE_CASCADE_CONFIG: dict[str, frozenset[str]] = {
    "_config_loader": frozenset({"config", "cli"}),
    "ingredient_defaults": frozenset({"config", "recipe"}),
}

MODULE_CASCADE_RECIPE: dict[str, frozenset[str]] = {
    # --- Narrowest: recipe/ only (no out-of-recipe importers) ---
    "rules_actions": frozenset({"recipe"}),
    "rules_blocks": frozenset({"recipe"}),
    "rules_bypass": frozenset({"recipe"}),
    "rules_campaign_capture": frozenset({"recipe"}),
    "rules_campaign_deps": frozenset({"recipe"}),
    "rules_campaign_dispatch": frozenset({"recipe"}),
    "rules_campaign_flow": frozenset({"recipe"}),
    "rules_campaign_ingredients": frozenset({"recipe"}),
    "rules_ci": frozenset({"recipe"}),
    "rules_ci_conflict": frozenset({"recipe"}),
    "rules_ci_guards": frozenset({"recipe"}),
    "rules_ci_merge_queue": frozenset({"recipe"}),
    "rules_clone": frozenset({"recipe"}),
    "rules_cmd": frozenset({"recipe"}),
    "rules_contracts": frozenset({"recipe"}),
    "rules_dataflow": frozenset({"recipe"}),
    "rules_dataflow_callable": frozenset({"recipe"}),
    "rules_dataflow_handoff": frozenset({"recipe"}),
    "rules_dataflow_multipart": frozenset({"recipe"}),
    "rules_features": frozenset({"recipe"}),
    "rules_fixing": frozenset({"recipe"}),
    "rules_graph": frozenset({"recipe"}),
    "rules_graph_output": frozenset({"recipe"}),
    "rules_graph_review": frozenset({"recipe"}),
    "rules_graph_routes": frozenset({"recipe"}),
    "rules_inline_script": frozenset({"recipe"}),
    "rules_inputs": frozenset({"recipe"}),
    "rules_isolation": frozenset({"recipe"}),
    "rules_loop_counter": frozenset({"recipe"}),
    "rules_merge": frozenset({"recipe"}),
    "rules_packs": frozenset({"recipe"}),
    "rules_reachability": frozenset({"recipe"}),
    "rules_recipe": frozenset({"recipe"}),
    "rules_skills": frozenset({"recipe"}),
    "rules_temp_path": frozenset({"recipe"}),
    "rules_terminal_convergence": frozenset({"recipe"}),
    "rules_tools": frozenset({"recipe"}),
    "rules_verdict": frozenset({"recipe"}),
    "rules_worktree": frozenset({"recipe"}),
    "rules_skill_content": frozenset(
        {
            "recipe",
            "recipe/test_rules_skill_content.py",
            "skills/test_graphql_invocation_completeness.py",
            "skills/test_skill_placeholder_contracts.py",
            "skills/test_skill_tool_syntax_contracts.py",
            "contracts/test_no_interpreter_writes_in_skills.py",
        }
    ),
    "rules_skill_write_path_alignment": frozenset({"recipe"}),
    # --- Internal utility modules (no external src importers) ---
    "_analysis": frozenset({"recipe"}),
    "_analysis_bfs": frozenset({"recipe"}),
    "_analysis_blocks": frozenset({"recipe"}),
    "_analysis_detectors": frozenset({"recipe"}),
    "_analysis_graph": frozenset({"recipe"}),
    "_git_helpers": frozenset({"recipe"}),
    "_skill_helpers": frozenset({"recipe"}),
    "_skill_placeholder_parser": frozenset(
        {
            "recipe",
            "arch/test_write_restriction_coverage.py",
            "skills/test_graphql_invocation_completeness.py",
            "skills/test_make_campaign_compliance.py",
            "skills/test_review_pr_prior_thread_awareness.py",
            "skills/test_skill_placeholder_contracts.py",
            "skills/test_skill_tool_syntax_contracts.py",
            "skills/test_audit_impl_diff_discipline.py",
            "skills/test_skill_variable_threading.py",
            "contracts/test_no_interpreter_writes_in_skills.py",
        }
    ),
    "_recipe_composition": frozenset({"recipe"}),
    "_recipe_ingredients": frozenset({"recipe"}),
    "diagrams": frozenset({"recipe"}),
    "identity": frozenset({"recipe"}),
    "order": frozenset({"recipe"}),
    "loader": frozenset(
        {
            "recipe",
            "cli/test_cli_prompts.py",
            "cli/test_preview.py",
        }
    ),
    # --- Medium: recipe + specific file-level consumers ---
    "staleness_cache": frozenset(
        {
            "recipe",
            "server/test_tools_load_recipe.py",
        }
    ),
    "repository": frozenset(
        {
            "recipe",
            "server/test_factory.py",
            "server/test_tools_kitchen_envelope.py",
            "server/test_service_wrappers.py",
            "contracts/test_protocol_satisfaction.py",
        }
    ),
    "validator": frozenset(
        {
            "recipe",
            "server/test_smoke_pipeline.py",
            "contracts/test_skill_yaml_validation.py",
        }
    ),
    "experiment_type_registry": frozenset(
        {
            "recipe",
            "skills/test_review_design_guards.py",
            "contracts/test_review_design_contracts.py",
            "integration",
            "cli",
            "smoke_utils",
        }
    ),
    "registry": frozenset(
        {
            "recipe",
            "server/test_smoke_pipeline.py",
            "contracts/test_skill_yaml_validation.py",
            "skills/test_skill_placeholder_contracts.py",
            "skills/test_skill_tool_syntax_contracts.py",
        }
    ),
    # --- Broader: recipe + multiple directory consumers ---
    "contracts": frozenset(
        {
            "recipe",
            "_llm_triage",
            "server/test_factory.py",
            "server/test_tools_load_recipe.py",
            "server/test_service_wrappers.py",
            "execution/test_headless_path_validation.py",
            "execution/test_zero_write_detection.py",
            "skills/test_planner_skill_contracts.py",
            "contracts/test_api_surface_alignment.py",
            "contracts/test_review_pr_diff_annotation.py",
            "contracts/test_instruction_surface.py",
        }
    ),
    "_api": frozenset(
        {
            "recipe",
            "server/test_tools_list_recipes.py",
            "server/test_tools_kitchen_envelope.py",
            "server/test_tools_load_recipe.py",
            "server/test_mcp_overrides.py",
            "server/test_service_wrappers.py",
            "cli/test_cli_prompts.py",
            "infra/test_pretty_output_recipe.py",
            "contracts/test_tools_recipe_contracts.py",
            "contracts/test_package_gateways.py",
        }
    ),
    "io": frozenset(
        {
            "recipe",
            "server/test_server_tool_registration.py",
            "server/test_smoke_pipeline.py",
            "server/test_service_wrappers.py",
            "cli/test_cli_prompts.py",
            "cli/test_fleet_list.py",
            "cli/test_preview.py",
            "fleet/test_pack_enforcement.py",
            "fleet/test_pack_enforcement_e2e.py",
            "skills/test_planner_skill_contracts.py",
            "contracts/test_skill_yaml_validation.py",
            "contracts/test_target_skill_invocability.py",
            "contracts/test_package_gateways.py",
            "contracts/test_instruction_surface.py",
        }
    ),
}

# ---------------------------------------------------------------------------
# Layer cascade maps
# ---------------------------------------------------------------------------

LAYER_CASCADE_CONSERVATIVE: dict[str, frozenset[str]] = {
    # L0 — imported by everything
    "core": frozenset(
        {
            "core",
            "config",
            "execution",
            "pipeline",
            "report",
            "workspace",
            "recipe",
            "migration",
            "fleet",
            "server",
            "cli",
            "hooks",
            "skills",
            "_llm_triage",
            "_test_filter",
            "hook_registry",
            "planner",
            "smoke_utils",
            "_probe_canary",
        }
    ),
    # L1
    "config": frozenset(
        {
            "config",
            "execution",
            "fleet",
            "pipeline",
            "recipe",
            "workspace",
            "server",
            "cli",
        }
    ),
    # skills/test_skill_output_compliance.py is intentionally absent: headless and process
    # subpackages carry it via SUBPKG_CASCADE_EXECUTION. The fallthrough path covers
    # execution modules that are not headless/process, so skills compliance testing is
    # out of scope for generic execution changes.
    "execution": frozenset(
        {
            "execution",
            "core",
            "workspace",
            "migration",
            "server",
            "cli",
            "fleet",
            "infra/test_pretty_output_hook_infra.py",
            "_llm_triage",
            "smoke_utils",
        }
    ),
    "pipeline": frozenset(
        {
            "pipeline",
            # execution/ — direct pipeline importers (8 files):
            "execution/test_clone_guard.py",
            "execution/test_headless_core.py",
            "execution/test_headless_path_validation.py",
            "execution/test_headless_provider_forwarding.py",
            "execution/test_headless_result.py",
            "execution/test_headless_result_write_reconciliation.py",
            "execution/test_planner_write_isolation.py",
            "execution/test_session_log_flush.py",
            # execution/ — fixture-mediated pipeline dependents (16 files):
            # These use minimal_ctx or tool_ctx fixtures which import
            # autoskillit.pipeline at call time. Validated by REQ-GUARD-007.
            "execution/test_backend_dispatch.py",
            "execution/test_boundary_pty_dispatch.py",
            "execution/test_flush_provider_integration.py",
            "execution/test_headless_add_dirs.py",
            "execution/test_headless_backend_mixing.py",
            "execution/test_headless_backend_override.py",
            "execution/test_headless_backend_resolution.py",
            "execution/test_headless_dispatch.py",
            "execution/test_headless_env_injection.py",
            "execution/test_headless_env_scrub.py",
            "execution/test_headless_execute.py",
            "execution/test_headless_provider_fallback.py",
            "execution/test_headless_synthesis.py",
            "execution/test_idle_output_env.py",
            "execution/test_write_evidence.py",
            "execution/test_zero_write_detection.py",
            "execution/test_session_log_fields.py",
            "server",
            "infra",
            "cli",
            "fleet",
            "smoke_utils",
        }
    ),
    "workspace": frozenset(
        {
            "workspace",
            # recipe direct-import entries (import autoskillit.workspace at AST level):
            "recipe/test_contracts.py",
            "recipe/test_backend_reachability.py",
            "recipe/test_recipe_backend_composition_matrix.py",
            "recipe/test_rules_backend_compat.py",
            "recipe/test_rules_skill_content.py",
            "recipe/test_rules_stamp_ownership.py",
            # recipe transitive entries (exercise workspace via deferred imports in
            # recipe/_api.py, _api_listing.py, _skill_helpers.py, _contracts_staleness.py,
            # rules/rules_skills.py, rules/rules_skill_content.py):
            "recipe/test_api.py",
            "recipe/test_api_cache_isolation.py",
            "recipe/test_bem_wrapper_structure.py",
            "recipe/test_bundled_recipes_dispatch_ready.py",
            "recipe/test_bundled_recipes_general.py",
            "recipe/test_callable_contracts.py",
            "recipe/test_contracts_block_fingerprint.py",
            "recipe/test_contract_verdict_output_required.py",
            "recipe/test_deep_staleness.py",
            "recipe/test_diagnose_ci_subtype_output.py",
            "recipe/test_hidden_ingredients.py",
            "recipe/test_io_discovery.py",
            "recipe/test_issue_url_pipeline.py",
            "recipe/test_planner_contracts.py",
            "recipe/test_recipe_temp_substitution.py",
            "recipe/test_repository.py",
            "recipe/test_research_campaign.py",
            "recipe/test_rules_contracts.py",
            "recipe/test_rules_dataflow_handoff.py",
            "recipe/test_rules_skill_routing.py",
            "recipe/test_rules_skills.py",
            "recipe/test_rules_tools.py",
            "recipe/test_skill_contract_completeness.py",
            "recipe/test_skill_emit_consistency.py",
            "recipe/test_skip_guard_deferral.py",
            "recipe/test_staleness_cache.py",
            "recipe/test_sub_recipe_loading.py",
            "recipe/test_sub_recipe_validation.py",
            # Server file-level entries:
            "server/test_backend_ingredient_injection.py",
            "server/test_factory.py",
            "server/test_tools_clone.py",
            "server/test_tools_execution_routing.py",
            "server/test_run_skill_add_dirs.py",
            "server/test_run_skill_backend_compat.py",
            "server/test_tools_workspace.py",
            "cli",
            "fleet",
            "skills",
            "_llm_triage",
        }
    ),
    # L2
    "recipe": frozenset(
        {
            "recipe",
            "_llm_triage",
            "core",
            "hooks/test_recipe_contract_freshness.py",
            "migration",
            # Server file-level entries (13 of 52 import autoskillit.recipe):
            "server/test_backend_ingredient_injection.py",
            "server/test_factory.py",
            "server/test_tools_dispatch_validation.py",
            "server/test_tools_kitchen_gate_features.py",
            "server/test_tools_load_recipe.py",
            "server/test_server_tool_registration.py",
            "server/test_mcp_overrides.py",
            "server/test_smoke_pipeline.py",
            "server/test_tools_kitchen_envelope.py",
            "server/test_tools_kitchen_cache_poison.py",
            "server/test_service_wrappers.py",
            "server/test_tools_list_recipes.py",
            "server/test_tools_bootstrap.py",
            "server/test_tools_kitchen_visibility.py",
            "server/test_tools_execution_step_resolution.py",
            "server/test_tools_execution_input_gates.py",
            "server/test_capability_admission_e2e.py",
            # CLI file-level entries (6 of 38 import autoskillit.recipe):
            "cli/test_cli_prompts.py",
            "cli/test_l3_orchestrator_prompt.py",
            "cli/test_cook_order_picker.py",
            "cli/test_fleet_list.py",
            "cli/test_preview.py",
            "cli/test_validate_registries.py",
            # Execution file-level entries:
            "execution/test_headless_path_validation.py",
            "execution/test_zero_write_detection.py",
            # Fleet file-level entries (8 of N import autoskillit.recipe):
            "fleet/test_fleet_e2e.py",
            "fleet/test_fleet_e2e_codex.py",
            "fleet/test_campaign_capture.py",
            "fleet/test_pack_enforcement.py",
            "fleet/test_pack_enforcement_e2e.py",
            "fleet/test_dispatch_ingredient_validation.py",
            "fleet/test_dispatch_recipe_kind_gate.py",
            "fleet/test_dispatch_state_handle.py",
            "fleet/test_research_campaign_dispatch.py",
            "fleet/test_dispatch_failure_semantics.py",
            "fleet/test_gate_state_persistence.py",
            # Other file-level entries:
            "infra/test_pretty_output_recipe.py",
            "infra/test_generated_files.py",
            "skills/test_graphql_invocation_completeness.py",
            "skills/test_planner_skill_contracts.py",
            "skills/test_review_pr_prior_thread_awareness.py",
            "skills/test_skill_placeholder_contracts.py",
            "skills/test_make_campaign_compliance.py",
            "skills/test_review_design_guards.py",
            "skills/test_skill_compliance.py",
            "skills/test_skill_tool_syntax_contracts.py",
            "skills/test_vis_lens_methodology_norms.py",
            "skills/test_audit_impl_diff_discipline.py",
            "skills/test_skill_variable_threading.py",
            "skills/test_phoropter_structural.py",
            "smoke_utils",
        }
    ),
    "migration": frozenset(
        {
            "migration",
            "server",
            "cli",
        }
    ),
    "fleet": frozenset(
        {
            "fleet",
            "server",
            "cli",
            # file-level: cross-layer test that imports fleet prompt builder
            "execution/backends/test_failure_predicate_spanning.py",
            # file-level: execution test that imports fleet.build_protected_campaign_ids
            "execution/test_session_log_retention.py",
        }
    ),
    # L3
    "server": frozenset(
        {
            "server",
            "cli",
            # file-level: fleet tests that import server tool handlers directly
            "fleet/test_api.py",
            "fleet/test_dispatch_crash_diagnostics.py",
            "fleet/test_fleet_e2e.py",
            "fleet/test_pack_enforcement.py",
            "fleet/test_gate_state_persistence.py",
            "fleet/test_dispatch_identity_continuity.py",
            "fleet/test_dispatch_envelope_fields.py",
            # file-level: cross-layer test that imports server TypedDict schemas
            "execution/backends/test_failure_predicate_spanning.py",
        }
    ),
    "cli": frozenset(
        {
            "cli",
            "__main__",
            # file-level: cross-layer test that imports cli prompt builder
            "execution/backends/test_failure_predicate_spanning.py",
        }
    ),
    # Infra (non-layered)
    "hooks": frozenset(
        {
            "hooks",
            "cli/test_cli_hooks.py",
            "cli/test_install.py",
            "recipe",
            "server/test_kitchen_lifecycle.py",
            "server/test_tools_kitchen_gate.py",
            "server/test_tools_kitchen_gate_hook_config.py",
            "execution/test_quota_sleep.py",
            "execution/test_session_log_fields.py",
        }
    ),
    "hook_registry": frozenset(
        {
            "hooks",
            "cli",
            "execution",
            # server/ narrowed to 4 files
            "server/test_tools_kitchen_envelope.py",
            "server/test_tools_kitchen_preflight.py",
            "server/test_tools_fleet_dispatch_preflight.py",
            "server/test_lifespan.py",
            "server/test_run_skill_backend_compat.py",
            # infra/ narrowed to 7 files
            "infra/test_adr_runtime_guard_coverage.py",
            "infra/test_command_guard_completeness.py",
            "infra/test_guard_coverage.py",
            "infra/test_risky_gh_subcommand_coverage.py",
            "infra/test_risky_git_ops_coverage.py",
            "infra/test_session_scope_enforcement.py",
            "infra/test_session_type_exemption_enforcement.py",
            "infra/test_skill_exemption_enforcement.py",
            # docs/ narrowed to 1 file
            "docs/test_doc_counts.py",
        }
    ),
    # Standalone modules (not subpackage directories)
    # L1
    "report": frozenset({"report", "skills_extended"}),
    "planner": frozenset(
        {
            "planner",
            # recipe file-level entries — exercise planner via _check_result_field_drift
            # in rules/rules_contracts.py (deferred import of PHASE_REQUIRED_KEYS):
            "recipe/test_rules_contracts.py",
            "recipe/test_contracts.py",
            "recipe/test_planner_recipe.py",
        }
    ),
    "_llm_triage": frozenset({"test_llm_triage.py", "server"}),
    "_test_filter": frozenset({"arch", "infra", "contracts"}),
    "smoke_utils": frozenset({"test_smoke_utils.py", "recipe", "smoke_utils"}),
    "version": frozenset({"test_version.py", "server", "cli"}),
    "_probe_canary": frozenset({"core", "execution/backends/test_probe_canary.py"}),
}

# ---------------------------------------------------------------------------
# Transitive override registry for REQ-GUARD-006
# ---------------------------------------------------------------------------
# Entries in LAYER_CASCADE_CONSERVATIVE whose test files do NOT import the
# cascade package directly at AST level — they are transitively reachable via
# deferred imports in recipe source modules (e.g. recipe/_api.py ->
# autoskillit.workspace.DefaultSkillResolver).  These are exempt from the
# direct-import check in test_file_level_entries_import_their_cascade_package.
#
# COUPLING: each set here is a strict subset of the corresponding
# LAYER_CASCADE_CONSERVATIVE entry — only the recipe/* transitive entries.
# Excluded from "workspace": the two direct-import entries
# ("recipe/test_contracts.py", "recipe/test_rules_skill_content.py") and all
# non-recipe entries ("workspace", "fleet", "server", "cli", "skills",
# "_llm_triage").  When adding a new transitive recipe/* entry to
# LAYER_CASCADE_CONSERVATIVE["workspace"] or ["planner"], add it here too —
# otherwise test_file_level_entries_import_their_cascade_package will
# incorrectly require a direct AST import from those test files.
_IMPORT_GUARD_TRANSITIVE_OVERRIDES: dict[str, frozenset[str]] = {
    "workspace": frozenset(
        {
            "recipe/test_api.py",
            "recipe/test_api_cache_isolation.py",
            "recipe/test_bem_wrapper_structure.py",
            "recipe/test_bundled_recipes_dispatch_ready.py",
            "recipe/test_bundled_recipes_general.py",
            "recipe/test_callable_contracts.py",
            "recipe/test_contracts_block_fingerprint.py",
            "recipe/test_contract_verdict_output_required.py",
            "recipe/test_deep_staleness.py",
            "recipe/test_diagnose_ci_subtype_output.py",
            "recipe/test_hidden_ingredients.py",
            "recipe/test_io_discovery.py",
            "recipe/test_issue_url_pipeline.py",
            "recipe/test_planner_contracts.py",
            "recipe/test_recipe_temp_substitution.py",
            "recipe/test_repository.py",
            "recipe/test_research_campaign.py",
            "recipe/test_rules_backend_compat.py",
            "recipe/test_rules_contracts.py",
            "recipe/test_rules_dataflow_handoff.py",
            "recipe/test_rules_skill_routing.py",
            "recipe/test_rules_skills.py",
            "recipe/test_rules_tools.py",
            "recipe/test_skill_contract_completeness.py",
            "recipe/test_skill_emit_consistency.py",
            "recipe/test_skip_guard_deferral.py",
            "recipe/test_staleness_cache.py",
            "recipe/test_sub_recipe_loading.py",
            "recipe/test_sub_recipe_validation.py",
            "server/test_backend_ingredient_injection.py",
        }
    ),
    "planner": frozenset(
        {
            "recipe/test_rules_contracts.py",
            "recipe/test_contracts.py",
            "recipe/test_planner_recipe.py",
        }
    ),
}

LAYER_CASCADE_AGGRESSIVE: dict[str, frozenset[str]] = {
    "core": frozenset({"core"}),
    "config": frozenset({"config"}),
    "execution": frozenset({"execution"}),
    "pipeline": frozenset({"pipeline"}),
    "workspace": frozenset({"workspace"}),
    "recipe": frozenset({"recipe"}),
    "migration": frozenset({"migration"}),
    "fleet": frozenset({"fleet"}),
    "server": frozenset({"server"}),
    "cli": frozenset({"cli"}),
    "hooks": frozenset({"hooks"}),
    "hook_registry": frozenset({"hooks"}),
    "planner": frozenset({"planner"}),
    "_llm_triage": frozenset({"test_llm_triage.py"}),
    "smoke_utils": frozenset({"test_smoke_utils.py"}),
    "version": frozenset({"test_version.py"}),
    "_test_filter": frozenset({"arch", "contracts"}),
    "report": frozenset({"report"}),
    "_probe_canary": frozenset({"core", "execution/backends/test_probe_canary.py"}),
}

# ---------------------------------------------------------------------------
# ASTImportWalker
# ---------------------------------------------------------------------------


class ASTImportWalker(ast.NodeVisitor):
    """Extract all import statements from a Python source file with context tracking."""

    def __init__(self) -> None:
        self.imports: list[tuple[str, ImportContext]] = []
        self._context_stack: list[ImportContext] = []

    @property
    def _current_context(self) -> ImportContext:
        return self._context_stack[-1] if self._context_stack else ImportContext.TOP_LEVEL

    def visit_Import(self, node: ast.Import) -> None:
        for alias in node.names:
            self.imports.append((alias.name, self._current_context))
        self.generic_visit(node)

    def visit_ImportFrom(self, node: ast.ImportFrom) -> None:
        if node.module:
            module = ("." * (node.level or 0)) + node.module
        else:
            module = "." * (node.level or 0)
        self.imports.append((module, self._current_context))
        self.generic_visit(node)

    def visit_FunctionDef(self, node: ast.FunctionDef) -> None:
        self._context_stack.append(ImportContext.DEFERRED)
        self.generic_visit(node)
        self._context_stack.pop()

    def visit_AsyncFunctionDef(self, node: ast.AsyncFunctionDef) -> None:
        self._context_stack.append(ImportContext.DEFERRED)
        self.generic_visit(node)
        self._context_stack.pop()

    def visit_If(self, node: ast.If) -> None:
        if self._is_type_checking_guard(node.test):
            self._context_stack.append(ImportContext.TYPE_CHECKING)
            for child in node.body:
                self.visit(child)
            self._context_stack.pop()
            self._context_stack.append(ImportContext.CONDITIONAL)
            for child in node.orelse:
                self.visit(child)
            self._context_stack.pop()
        else:
            self._context_stack.append(ImportContext.CONDITIONAL)
            self.generic_visit(node)
            self._context_stack.pop()

    def visit_Try(self, node: ast.Try) -> None:
        self._context_stack.append(ImportContext.CONDITIONAL)
        self.generic_visit(node)
        self._context_stack.pop()

    def visit_Call(self, node: ast.Call) -> None:
        if self._is_importlib_call(node) and node.args:
            arg = node.args[0]
            if isinstance(arg, ast.Constant) and isinstance(arg.value, str):
                self.imports.append((arg.value, ImportContext.IMPORTLIB))
        self.generic_visit(node)

    @staticmethod
    def _is_type_checking_guard(test: ast.expr) -> bool:
        if isinstance(test, ast.Name) and test.id == "TYPE_CHECKING":
            return True
        if isinstance(test, ast.Attribute) and test.attr == "TYPE_CHECKING":
            return True
        return False

    @staticmethod
    def _is_importlib_call(node: ast.Call) -> bool:
        func = node.func
        if isinstance(func, ast.Attribute) and func.attr == "import_module":
            return True
        return False


# ---------------------------------------------------------------------------
# Utility functions
# ---------------------------------------------------------------------------


def git_changed_files(
    cwd: str | Path,
    base_ref: str | None = None,
) -> set[str] | None:
    """Return set of changed files relative to base_ref, or None on failure."""
    import os

    if base_ref is None:
        base_ref = os.environ.get(
            "AUTOSKILLIT_TEST_BASE_REF",
            os.environ.get("GITHUB_BASE_REF"),
        )
    if base_ref is None:
        return None

    try:
        merge_base_result = subprocess.run(
            ["git", "merge-base", "HEAD", base_ref],
            cwd=str(cwd),
            capture_output=True,
            text=True,
            timeout=10,
            check=True,
        )
        merge_base_sha = merge_base_result.stdout.strip()
        if not merge_base_sha:
            warnings.warn("git merge-base returned empty output", stacklevel=2)
            return None

        diff_result = subprocess.run(
            ["git", "diff", "--name-only", merge_base_sha],
            cwd=str(cwd),
            capture_output=True,
            text=True,
            timeout=10,
            check=True,
        )
    except subprocess.CalledProcessError as exc:
        warnings.warn(f"git diff failed (exit {exc.returncode}): {exc.stderr or ''}", stacklevel=2)
        return None
    except subprocess.TimeoutExpired:
        warnings.warn("git diff timed out after 10s", stacklevel=2)
        return None
    except FileNotFoundError:
        warnings.warn("git binary not found on PATH", stacklevel=2)
        return None

    files: set[str] = set()
    for line in diff_result.stdout.strip().splitlines():
        if line.strip():
            files.add(line.strip())

    untracked = subprocess.run(
        ["git", "ls-files", "--others", "--exclude-standard"],
        cwd=str(cwd),
        capture_output=True,
        text=True,
        timeout=10,
        check=False,
    )
    if untracked.returncode == 0:
        for line in untracked.stdout.strip().splitlines():
            if line.strip():
                files.add(line.strip())

    return files


def git_changed_files_local(
    cwd: str | Path,
) -> set[str] | None:
    """Return files changed in working tree vs HEAD, or None on failure.

    Uses ``git diff HEAD --name-only`` (staged + unstaged vs last commit)
    plus ``git ls-files --others --exclude-standard`` (untracked).
    """
    try:
        diff_result = subprocess.run(
            ["git", "diff", "HEAD", "--name-only"],
            cwd=str(cwd),
            capture_output=True,
            text=True,
            timeout=10,
            check=True,
        )
    except subprocess.CalledProcessError as exc:
        warnings.warn(
            f"git diff HEAD failed (exit {exc.returncode}): {exc.stderr or ''}", stacklevel=2
        )
        return None
    except subprocess.TimeoutExpired:
        warnings.warn("git diff HEAD timed out after 10s", stacklevel=2)
        return None
    except FileNotFoundError:
        warnings.warn("git binary not found on PATH", stacklevel=2)
        return None

    files: set[str] = set()
    for line in diff_result.stdout.strip().splitlines():
        if line.strip():
            files.add(line.strip())

    try:
        untracked = subprocess.run(
            ["git", "ls-files", "--others", "--exclude-standard"],
            cwd=str(cwd),
            capture_output=True,
            text=True,
            timeout=10,
            check=False,
        )
    except subprocess.TimeoutExpired:
        warnings.warn("git ls-files timed out after 10s", stacklevel=2)
        return files
    except FileNotFoundError:
        warnings.warn("git binary not found on PATH", stacklevel=2)
        return files
    if untracked.returncode == 0:
        for line in untracked.stdout.strip().splitlines():
            if line.strip():
                files.add(line.strip())

    return files


def check_bucket_a(changed_files: set[str]) -> bool:
    """Return True if any changed file triggers a full test run."""
    for f in changed_files:
        if f in BUCKET_A_PATTERNS:
            return True
        for glob_pat in BUCKET_A_GLOBS:
            if fnmatch.fnmatch(f, glob_pat):
                return True
    return False


def _is_only_version_changes_in_diff(
    cwd: str | Path,
    base_ref: str,
    *paths: str,
) -> bool:
    """Return True if every added/removed line in the diff of *paths* is a version string.

    Runs ``git merge-base HEAD base_ref`` then ``git diff --unified=0 <sha> -- *paths``.
    Returns False on any git error (fail-open: caller treats files as Bucket A triggers).
    Returns True if the diff is empty (no changes at all — version bump already absorbed).
    """
    try:
        merge_base_result = subprocess.run(
            ["git", "merge-base", "HEAD", base_ref],
            cwd=str(cwd),
            capture_output=True,
            text=True,
            timeout=10,
            check=True,
        )
        merge_base_sha = merge_base_result.stdout.strip()
        if not re.fullmatch(r"[0-9a-f]{40}", merge_base_sha):
            return False

        diff_result = subprocess.run(
            ["git", "diff", "--unified=0", merge_base_sha, "--", *paths],
            cwd=str(cwd),
            capture_output=True,
            text=True,
            timeout=10,
            check=True,
        )
    except (subprocess.CalledProcessError, subprocess.TimeoutExpired, FileNotFoundError, OSError):
        return False

    for line in diff_result.stdout.splitlines():
        if not (line.startswith("+") or line.startswith("-")):
            continue
        if line.startswith("+++") or line.startswith("---"):
            continue
        if not _VERSION_LINE_RE.match(line):
            return False  # non-version change found — not a pure version bump

    return True  # all diff lines are version strings (or diff is empty)


def _is_additive_only(
    cwd: str | Path,
    base_ref: str,
    filepath: str,
) -> bool:
    """Return True if the diff for *filepath* contains no removed definitions.

    Runs ``git merge-base HEAD base_ref`` then ``git diff --unified=0 <sha> -- filepath``.
    Returns False on any git error (fail-open: caller uses the full cascade).

    Checks for removed lines matching class definitions, function definitions,
    or UPPER_CASE constant assignments. Lowercase field additions/modifications
    are treated as additive — acceptable for the narrowing use case since the
    excluded directories do not import those types.
    """
    if Path(filepath).is_absolute():
        return False
    try:
        merge_base_result = subprocess.run(
            ["git", "merge-base", "HEAD", base_ref],
            cwd=str(cwd),
            capture_output=True,
            text=True,
            timeout=10,
            check=True,
        )
        merge_base_sha = merge_base_result.stdout.strip()
        if not re.fullmatch(r"[0-9a-f]{40}", merge_base_sha):
            return False

        diff_result = subprocess.run(
            ["git", "diff", "--unified=0", merge_base_sha, "--", filepath],
            cwd=str(cwd),
            capture_output=True,
            text=True,
            timeout=10,
            check=True,
        )
    except (subprocess.CalledProcessError, subprocess.TimeoutExpired, FileNotFoundError, OSError):
        return False

    return not _BREAKING_DEF_RE.search(diff_result.stdout)


# Files in BUCKET_A_PATTERNS that may produce false positives from CI version bumps.
# These are given a content check before triggering a full run.
_VERSION_BUMP_FILES: frozenset[str] = frozenset({"pyproject.toml", "uv.lock"})


def check_bucket_a_content_aware(
    changed_files: set[str],
    cwd: str | Path,
    base_ref: str,
) -> bool:
    """Content-aware Bucket A check that skips version-bump-only changes.

    Identical to ``check_bucket_a`` except for ``pyproject.toml`` and ``uv.lock``:
    if those files are present in *changed_files* but their entire diff consists only
    of ``version = "..."`` line changes, they are NOT treated as Bucket A triggers.

    Falls back to treating them as Bucket A triggers on any git failure (fail-open).

    Other Bucket A patterns (conftest.py, _rules.py, etc.) are unaffected — they
    still trigger a full run immediately without any git diff inspection.
    """
    # Fast path: check all non-version-bump patterns first (no git I/O)
    non_version_files = changed_files - _VERSION_BUMP_FILES
    if check_bucket_a(non_version_files):
        return True

    # Check version-bump-candidate files that are also in BUCKET_A_PATTERNS
    version_hits = changed_files & _VERSION_BUMP_FILES & BUCKET_A_PATTERNS
    if not version_hits:
        return False  # no version-bump candidates hit Bucket A

    # Content check: if every diff line is a version string, skip Bucket A
    if _is_only_version_changes_in_diff(cwd, base_ref, *version_hits):
        return False  # version-only bump — not a structural change

    return True  # diff contains non-version changes → full run


def load_manifest(path: str | Path) -> dict[str, Any] | None:
    """Load .autoskillit/test-filter-manifest.yaml, or None if absent."""
    import yaml

    manifest_path = Path(path) / ".autoskillit" / "test-filter-manifest.yaml"
    if not manifest_path.exists():
        return None
    try:
        _Loader = getattr(yaml, "CSafeLoader", yaml.SafeLoader)
        with manifest_path.open() as f:
            return yaml.load(f, Loader=_Loader)
    except OSError as exc:
        warnings.warn(f"Cannot read manifest {manifest_path}: {exc}", stacklevel=2)
        return None
    except yaml.YAMLError as exc:
        warnings.warn(f"Malformed YAML in {manifest_path}: {exc}", stacklevel=2)
        return None


def load_coverage_map(
    map_path: str | Path,
    max_age_days: int = 30,
) -> dict[str, set[str]] | None:
    """Load .autoskillit/test-source-map.json with staleness guard.

    Args:
        map_path: Path to the test-source-map.json file.
        max_age_days: Maximum age in days before the map is considered stale.
                      Defaults to 30 days.

    Returns:
        dict mapping source file paths to sets of test file paths, or None when:
        - The file does not exist
        - The file is older than max_age_days
        - The file cannot be read or parsed
    """
    map_path = Path(map_path)
    try:
        stat = map_path.stat()
    except OSError:
        return None

    age = datetime.datetime.now().timestamp() - stat.st_mtime
    if age > max_age_days * 24 * 3600:
        return None

    try:
        raw = json.loads(map_path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as exc:
        warnings.warn(f"Cannot read coverage map {map_path}: {exc}", stacklevel=2)
        return None

    if not isinstance(raw, dict):
        warnings.warn(f"Coverage map {map_path} is not a JSON object", stacklevel=2)
        return None

    result: dict[str, set[str]] = {}
    for src, tests in raw.items():
        if not isinstance(tests, list):
            warnings.warn(
                f"Coverage map {map_path}: value for {src!r} is not a list",
                stacklevel=2,
            )
            return None
        result[src] = set(tests)
    return result


def apply_manifest(
    changed_files: set[str],
    manifest: dict[str, Any] | None,
) -> set[str] | None:
    """Return test directories matched by manifest patterns for the changed files.

    Returns None when manifest is None (fail-open) or when any changed file matches
    no manifest pattern (fail-open: caller should run the full suite).
    """
    if manifest is None:
        return None
    compiled = {pat: pathspec.PathSpec.from_lines("gitwildmatch", [pat]) for pat in manifest}
    matched_dirs: set[str] = set()
    for f in changed_files:
        file_matched = False
        for pattern, spec in compiled.items():
            if spec.match_file(f):
                test_dirs = manifest[pattern]
                if isinstance(test_dirs, list):
                    matched_dirs.update(test_dirs)
                elif isinstance(test_dirs, str):
                    matched_dirs.add(test_dirs)
                file_matched = True
        if not file_matched:
            return None
    return matched_dirs


# ---------------------------------------------------------------------------
# Re-export closure
# ---------------------------------------------------------------------------


def _expand_reexport_closure(
    changed_src_files: set[str],
    src_root: str | Path,
) -> set[str]:
    """Expand changed files to include __init__.py files that directly re-export from them.

    Walks up parent directories checking each __init__.py via AST parsing.
    Only handles direct re-exports (from .module import X); transitive
    chains through intermediate hubs are covered by cascade maps.
    """
    src_root = Path(src_root)
    expanded = set(changed_src_files)

    for changed in list(changed_src_files):
        changed_path = src_root / changed
        if not changed_path.exists():
            continue
        module_name = changed_path.stem
        parent = changed_path.parent
        while parent >= src_root:
            init_path = parent / "__init__.py"
            if init_path.exists():
                try:
                    source = init_path.read_text(errors="replace")
                    tree = ast.parse(source, filename=str(init_path))
                    walker = ASTImportWalker()
                    walker.visit(tree)
                    for mod, _ctx in walker.imports:
                        bare = mod.split(".")[-1] if "." in mod else mod.lstrip(".")
                        if bare == module_name:
                            rel_init = str(init_path.relative_to(src_root))
                            expanded.add(rel_init)
                            break
                except SyntaxError as exc:
                    warnings.warn(
                        f"Failed to parse {init_path}: {exc}",
                        stacklevel=2,
                    )
            parent = parent.parent

    return expanded


# ---------------------------------------------------------------------------
# build_test_scope
# ---------------------------------------------------------------------------


def _file_to_package(filepath: str) -> str | None:
    """Extract the autoskillit subpackage name from a source file path.

    e.g. 'src/autoskillit/core/io.py' -> 'core'
         'src/autoskillit/execution/headless.py' -> 'execution'
         'src/autoskillit/server/_factory.py' -> 'server'
         'src/autoskillit/hook_registry.py' -> 'hook_registry'
    """
    parts = Path(filepath).parts
    try:
        idx = parts.index("autoskillit")
        if idx + 1 < len(parts):
            candidate = parts[idx + 1]
            if candidate.endswith(".py"):
                return candidate.removesuffix(".py")
            return candidate
    except ValueError:
        pass
    return None


def _file_to_execution_subpkg(filepath: str) -> str | None:
    """Return the execution subpackage name if *filepath* is inside one, else None.

    e.g. 'src/autoskillit/execution/merge_queue/_classifier.py' -> 'merge_queue'
         'src/autoskillit/execution/headless.py' -> None
         'src/autoskillit/execution/__init__.py' -> None
    """
    parts = Path(filepath).parts
    try:
        idx = parts.index("autoskillit")
    except ValueError:
        return None
    # idx + 3 < len(parts) ensures parts[idx + 2] is a subpackage dir (not a bare module)
    # and guards parts[idx + 1].
    if idx + 3 < len(parts) and parts[idx + 1] == "execution":
        return parts[idx + 2]
    return None


def build_test_scope(
    changed_files: set[str] | None,
    mode: FilterMode,
    manifest: dict[str, Any] | None = None,
    tests_root: str | Path = "tests",
    coverage_map_path: str | Path | None = None,
    cwd: str | Path | None = None,
    base_ref: str | None = None,
) -> set[Path] | FullRunReason:
    """Compute the set of test paths to run, or a FullRunReason for a full run.

    Algorithm:
    1. None changed_files -> FullRunReason.GIT_UNAVAILABLE (fail-open)
    2. Conservative + >30 files -> FullRunReason.LARGE_CHANGESET (aggressive mode: no threshold)
    3. Bucket A triggered -> FullRunReason.BUCKET_A (full run)
    4. Classify: src Python -> cascade, test Python -> direct, non-Python -> manifest
    5. Compute always-run set for mode (includes arch/contracts for both modes)
    6. Union all sets
    7. Coverage oracle file-level refinement (same-package narrowing, both modes)
    8. Resolve to concrete paths
    """
    if mode == FilterMode.NONE:
        return FullRunReason.DISABLED

    if changed_files is None:
        return FullRunReason.GIT_UNAVAILABLE

    if (
        mode == FilterMode.CONSERVATIVE
        and len(changed_files) > _LARGE_CHANGESET_THRESHOLD_CONSERVATIVE
    ):
        return FullRunReason.LARGE_CHANGESET

    if cwd is not None and base_ref is not None:
        if check_bucket_a_content_aware(changed_files, cwd, base_ref):
            return FullRunReason.BUCKET_A
        # Exclude version-bump files that passed the content-aware check from classification.
        # Recomputes the same set as `version_hits` inside check_bucket_a_content_aware because
        # that function returns bool; extracting the set here avoids changing its signature.
        version_bump_in_bucket_a = changed_files & _VERSION_BUMP_FILES & BUCKET_A_PATTERNS
        if version_bump_in_bucket_a:
            changed_files = changed_files - version_bump_in_bucket_a
    else:
        if check_bucket_a(changed_files):
            return FullRunReason.BUCKET_A

    tests_root = Path(tests_root)

    cascade_map = (
        LAYER_CASCADE_CONSERVATIVE if mode == FilterMode.CONSERVATIVE else LAYER_CASCADE_AGGRESSIVE
    )
    always_run = (
        ALWAYS_RUN_CONSERVATIVE if mode == FilterMode.CONSERVATIVE else ALWAYS_RUN_AGGRESSIVE
    )

    test_dirs: set[str] = set()
    direct_test_files: set[str] = set()
    for f in changed_files:
        if f.startswith("tests/") and f.endswith(".py"):
            direct_test_files.add(f)
        elif f.startswith("src/") and f.endswith(".py"):
            pkg = _file_to_package(f)
            if pkg == "core" and mode == FilterMode.CONSERVATIVE:
                stem = Path(f).stem
                if stem in _CORE_UNIVERSAL_MODULES:
                    if (
                        stem in _CORE_UNIVERSAL_EXCLUSIONS
                        and cwd is not None
                        and base_ref is not None
                        and _is_additive_only(cwd, base_ref, f)
                    ):
                        test_dirs.update(cascade_map["core"] - _CORE_UNIVERSAL_EXCLUSIONS[stem])
                    else:
                        test_dirs.update(cascade_map["core"])
                elif stem == "__init__":
                    test_dirs.update(cascade_map["core"])
                elif stem in MODULE_CASCADE_CORE:
                    test_dirs.update(MODULE_CASCADE_CORE[stem])
                else:
                    test_dirs.update(cascade_map["core"])  # fail-open: future unclassified stems
            elif pkg == "execution" and mode == FilterMode.CONSERVATIVE:
                subpkg = _file_to_execution_subpkg(f)
                if subpkg and subpkg in SUBPKG_CASCADE_EXECUTION:
                    test_dirs.update(SUBPKG_CASCADE_EXECUTION[subpkg])
                else:
                    stem = Path(f).stem
                    if stem in MODULE_CASCADE_EXECUTION:
                        test_dirs.update(MODULE_CASCADE_EXECUTION[stem])
                    else:
                        test_dirs.update(
                            cascade_map["execution"]
                        )  # fail-open: unknown execution modules
            elif pkg == "recipe" and mode == FilterMode.CONSERVATIVE:
                stem = Path(f).stem
                if stem in MODULE_CASCADE_RECIPE:
                    test_dirs.update(MODULE_CASCADE_RECIPE[stem])
                else:
                    test_dirs.update(cascade_map["recipe"])  # fail-open
            elif pkg == "config" and mode == FilterMode.CONSERVATIVE:
                stem = Path(f).stem
                if stem in MODULE_CASCADE_CONFIG:
                    test_dirs.update(MODULE_CASCADE_CONFIG[stem])
                else:
                    test_dirs.update(
                        cascade_map["config"]
                    )  # fail-open: __init__, settings, _config_dataclasses, etc.
            elif pkg and pkg in cascade_map:
                test_dirs.update(cascade_map[pkg])
            else:
                return FullRunReason.UNMAPPED_FILE
        elif f.endswith(".py"):
            manifest_dirs = apply_manifest({f}, manifest)
            if manifest_dirs is None:
                return FullRunReason.UNMAPPED_FILE
            test_dirs.update(manifest_dirs)
        else:
            manifest_dirs = apply_manifest({f}, manifest)
            if manifest_dirs is None:
                return FullRunReason.UNMAPPED_FILE
            test_dirs.update(manifest_dirs)

    # Expand src Python files via re-export closure: add __init__.py files that
    # directly re-export any of the changed modules, then cascade-classify them.
    changed_src_py = {f for f in changed_files if f.startswith("src/") and f.endswith(".py")}
    if changed_src_py:
        try:
            expanded = _expand_reexport_closure(changed_src_py, tests_root.parent)
            for f in expanded - changed_src_py:
                if f.startswith("src/") and f.endswith(".py"):
                    pkg = _file_to_package(f)
                    if pkg == "core" and mode == FilterMode.CONSERVATIVE:
                        stem = Path(f).stem
                        if stem in _CORE_UNIVERSAL_MODULES:
                            if (
                                stem in _CORE_UNIVERSAL_EXCLUSIONS
                                and cwd is not None
                                and base_ref is not None
                                and _is_additive_only(cwd, base_ref, f)
                            ):
                                test_dirs.update(
                                    cascade_map["core"] - _CORE_UNIVERSAL_EXCLUSIONS[stem]
                                )
                            else:
                                test_dirs.update(cascade_map["core"])
                        elif stem == "__init__":
                            core_cause_stems = {
                                Path(c).stem
                                for c in changed_src_py
                                if _file_to_package(c) == "core" and Path(c).stem != "__init__"
                            }
                            if core_cause_stems and all(
                                s in MODULE_CASCADE_CORE and s not in _CORE_UNIVERSAL_MODULES
                                for s in core_cause_stems
                            ):
                                for s in core_cause_stems:
                                    test_dirs.update(MODULE_CASCADE_CORE[s])
                            else:
                                test_dirs.update(cascade_map["core"])
                        elif stem in MODULE_CASCADE_CORE:
                            test_dirs.update(MODULE_CASCADE_CORE[stem])
                        else:
                            test_dirs.update(
                                cascade_map["core"]
                            )  # fail-open: future unclassified stems
                    elif pkg == "execution" and mode == FilterMode.CONSERVATIVE:
                        subpkg = _file_to_execution_subpkg(f)
                        if subpkg and subpkg in SUBPKG_CASCADE_EXECUTION:
                            test_dirs.update(SUBPKG_CASCADE_EXECUTION[subpkg])
                        else:
                            stem = Path(f).stem
                            if stem == "__init__":
                                exec_cause_files = [
                                    c
                                    for c in changed_src_py
                                    if _file_to_package(c) == "execution"
                                    and Path(c).stem != "__init__"
                                ]
                                # empty list → fail-open: only execution/__init__ changed,
                                # no cause to narrow on
                                all_narrow = bool(exec_cause_files)
                                narrow_dirs: set[str] = set()
                                for c in exec_cause_files:
                                    c_subpkg = _file_to_execution_subpkg(c)
                                    if c_subpkg and c_subpkg in SUBPKG_CASCADE_EXECUTION:
                                        narrow_dirs.update(SUBPKG_CASCADE_EXECUTION[c_subpkg])
                                    elif Path(c).stem in MODULE_CASCADE_EXECUTION:
                                        narrow_dirs.update(MODULE_CASCADE_EXECUTION[Path(c).stem])
                                    else:
                                        all_narrow = False
                                        break
                                if all_narrow:
                                    test_dirs.update(narrow_dirs)
                                else:
                                    test_dirs.update(cascade_map["execution"])  # fail-open
                            elif stem in MODULE_CASCADE_EXECUTION:
                                test_dirs.update(MODULE_CASCADE_EXECUTION[stem])
                            else:
                                test_dirs.update(cascade_map["execution"])  # fail-open
                    elif pkg == "recipe" and mode == FilterMode.CONSERVATIVE:
                        stem = Path(f).stem
                        if stem == "__init__":
                            recipe_cause_stems = {
                                Path(c).stem
                                for c in changed_src_py
                                if _file_to_package(c) == "recipe" and Path(c).stem != "__init__"
                            }
                            if recipe_cause_stems and all(
                                s in MODULE_CASCADE_RECIPE for s in recipe_cause_stems
                            ):
                                for s in recipe_cause_stems:
                                    test_dirs.update(MODULE_CASCADE_RECIPE[s])
                            else:
                                if not recipe_cause_stems:
                                    logging.getLogger(__name__).debug(  # noqa: TID251
                                        "recipe/__init__ backtrace: no non-init recipe cause stems"
                                        " — failing open to full recipe cascade (expected behavior"
                                        " when only recipe/__init__.py changed)"
                                    )
                                test_dirs.update(cascade_map["recipe"])  # fail-open
                        elif stem in MODULE_CASCADE_RECIPE:
                            test_dirs.update(MODULE_CASCADE_RECIPE[stem])
                        else:
                            test_dirs.update(cascade_map["recipe"])  # fail-open
                    elif pkg == "config" and mode == FilterMode.CONSERVATIVE:
                        stem = Path(f).stem
                        if stem == "__init__":
                            config_cause_stems = {
                                Path(c).stem
                                for c in changed_src_py
                                if _file_to_package(c) == "config" and Path(c).stem != "__init__"
                            }
                            if config_cause_stems and all(
                                s in MODULE_CASCADE_CONFIG for s in config_cause_stems
                            ):
                                for s in config_cause_stems:
                                    test_dirs.update(MODULE_CASCADE_CONFIG[s])
                            else:
                                if not config_cause_stems:
                                    logging.getLogger(__name__).debug(  # noqa: TID251
                                        "config/__init__ backtrace: no non-init config cause stems"
                                        " — failing open to full config cascade (expected behavior"
                                        " when only config/__init__.py changed)"
                                    )
                                test_dirs.update(cascade_map["config"])  # fail-open
                        elif stem in MODULE_CASCADE_CONFIG:
                            test_dirs.update(MODULE_CASCADE_CONFIG[stem])
                        else:
                            test_dirs.update(
                                cascade_map["config"]
                            )  # fail-open: non-__init__ unmapped config stems
                    elif pkg and pkg in cascade_map:
                        test_dirs.update(cascade_map[pkg])
        except Exception:
            logging.getLogger(__name__).debug(  # noqa: TID251
                "_expand_reexport_closure suppressed", exc_info=True
            )  # fail-open: expansion errors do not affect the computed scope

    if mode == FilterMode.CONSERVATIVE and changed_files:
        # REQ-TIER-001: arch and contracts always unconditional
        test_dirs.update(_ALWAYS_RUN_CONSERVATIVE_UNCONDITIONAL)

        # REQ-TIER-002: docs gated on documentation file changes
        if any(
            f.startswith(_DOCS_TRIGGER_PREFIX) or f in _DOCS_TRIGGER_FILES for f in changed_files
        ):
            test_dirs.add("docs")
        else:
            direct_test_files.add(str(tests_root / "docs" / "test_doc_counts.py"))

        # REQ-TIER-003: 9 infra + 3 hooks structural files always; full infra dir only on trigger
        for fname in _INFRA_UNCONDITIONAL_FILES:
            direct_test_files.add(str(tests_root / "infra" / fname))
        for fname in _HOOKS_UNCONDITIONAL_FILES:
            direct_test_files.add(str(tests_root / "hooks" / fname))
        if any(
            f.startswith(_INFRA_HOOK_TRIGGER_PREFIX)
            or f.startswith(_INFRA_CI_TRIGGER_PREFIX)
            or f in _INFRA_CI_TRIGGER_FILES
            for f in changed_files
        ):
            test_dirs.add("infra")
    else:
        # REQ-TIER-004: fail-open for empty changeset; aggressive mode uses its own set
        test_dirs.update(always_run)

    # Step 7: File-level refinement via coverage oracle.
    # Uses LAYER_CASCADE_AGGRESSIVE for grouping regardless of mode, ensuring
    # only same-package test directories are narrowed — cross-package cascade
    # directories from conservative mode remain untouched.
    if coverage_map_path is not None:
        cov_map = load_coverage_map(coverage_map_path)
        if cov_map is not None:
            oracle_cascade = LAYER_CASCADE_AGGRESSIVE
            dir_to_src_files: dict[str, set[str]] = {}
            for f in changed_src_py:
                pkg = _file_to_package(f)
                if pkg and pkg in oracle_cascade:
                    for d in oracle_cascade[pkg]:
                        dir_to_src_files.setdefault(d, set()).add(f)

            for d, src_files in dir_to_src_files.items():
                if d in test_dirs and all(f in cov_map and cov_map[f] for f in src_files):
                    test_dirs.discard(d)
                    for f in src_files:
                        direct_test_files.update(str(tests_root.parent / fp) for fp in cov_map[f])

    result: set[Path] = set()
    for d in test_dirs:
        dir_path = tests_root / d
        if dir_path.is_dir() or dir_path.is_file():
            result.add(dir_path)

    for f in direct_test_files:
        result.add(Path(f))

    return result