Play Policy Insights

scripts/orchestrator.py

#!/usr/bin/env python3

# Copyright 2026 The Android Open Source Project
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""Coordinates the end-to-end Play Policy audit flow.

This manages everything from triage to analysis aggregation.
"""

import argparse
import concurrent.futures
import glob
import json
import os
import re
import sys
import uuid
import xml.etree.ElementTree as ET

import scanner
from template_engine import render_template

# Constants for Gradle parsing
GRADLE_APP_ID_PATTERN = (
    r'applicationId\s*(?:[:=]|\.set\(?)\s*[\'"]?([a-zA-Z0-9._]+)[\'"]?\)?'
)
GRADLE_APP_ID_PROP_PATTERN = (
    r'applicationId\s*[:=]?\s*project\.property\(\s*[\'"]([^\'"]+)[\'"]\s*\)'
)
GRADLE_NAMESPACE_PATTERN = (
    r'namespace\s*(?:[:=]|\.set\(?)\s*[\'"]?([a-zA-Z0-9._]+)[\'"]?\)?'
)
GRADLE_NAMESPACE_PROP_PATTERN = (
    r'namespace\s*[:=]?\s*project\.property\(\s*[\'"]([^\'"]+)[\'"]\s*\)'
)
GRADLE_GROUP_PATTERN = (
    r'group(?:Id)?\s*(?:[:=]|\.set\(?)\s*[\'"]?([a-zA-Z0-9._]+)[\'"]?\)?'
)
GRADLE_GROUP_PROP_PATTERN = (
    r'group(?:Id)?\s*[:=]?\s*project\.property\(\s*[\'"]([^\'"]+)[\'"]\s*\)'
)
GRADLE_TARGET_SDK_PATTERN = (
    r"targetSdk(?:Version)?(?:\.set\(?)?\s*[:="
    r" (]*\s*([\'\"]?[\w\d_\.]+[\'\"]?)\)?"
)

# Well-known framework defaults for unresolved dynamic variables (e.g. Flutter)
FRAMEWORK_DEFAULTS = {
    "flutter.targetSdkVersion": 35,
    "flutter.compileSdkVersion": 35,
    "flutter.minSdkVersion": 21,
    "flutter.ndkVersion": "25.1.8937393",
}

# .NET MAUI / .NET for Android express the target SDK through the target framework
# moniker rather than a manifest or Gradle file. An explicit platform level wins
# (net9.0-android35.0 -> 35); an unqualified moniker (net9.0-android) resolves to
# the highest API the installed workload supports, which is fixed per .NET release.
DOTNET_ANDROID_TARGET_SDK = {
    "8.0": 34,
    "9.0": 35,
    "10.0": 36,
}

# net10.0-android, net10.0-android36.0, net8.0-android34.0 ...
DOTNET_TFM_PATTERN = re.compile(
    r"net(\d+\.\d+)-android(\d+)?(?:\.\d+)?", re.IGNORECASE
)

# Patterns for diverse assignment styles (Groovy & Kotlin DSL)
# 1. Standard: targetSdkVersion = 35, ext.targetSdkVersion = 35, etc.
# 2. Delegated: val targetSdkVersion by extra(35)
# 3. Functional: set("targetSdkVersion", 35) or
#    extra.set("targetSdkVersion", 35)
# 4. Indexer: extra["targetSdkVersion"] = 35
GRADLE_ASSIGNMENT_PATTERNS = [
    re.compile(
        r"(?:\b(?:val|var|def|const|static|internal|private|"
        r"public|protected|ext|final)\s+)*"
        r"([\w\d_\.]+)(?:\s*:\s*[\w\d_\.]+)?\s*=\s*['\"]?"
        r"([^\x27\"\s\(\)<>\{\}]+)['\"]?"
    ),
    re.compile(
        r"val\s+([\w\d_\.]+)(?:\s*:\s*[\w\d_\.]+)?\s+by\s+"
        r"extra\s*\(\s*['\"]?([^\x27\"\s\(\)]+)['\"]?\s*\)"
    ),
    re.compile(
        r"(?:extra\.)?set\s*\(\s*['\"]([\w\d_\.]+)['\"]\s*,\s*"
        r"['\"]?([^\x27\"\s\(\)]+)['\"]?\s*\)"
    ),
    re.compile(
        r"extra\s*\[\s*['\"]([\w\d_\.]+)['\"]\s*\]\s*=\s*"
        r"['\"]?([^\x27\"\s]+)['\"]?"
    ),
]


def _strip_inline_comment(line, support_quotes=True):
  """Strips inline comments starting with '#' while preserving quotes."""
  in_double = False
  in_single = False
  for i, char in enumerate(line):
    if support_quotes:
      if char == '"' and not in_single:
        in_double = not in_double
      elif char == "'" and not in_double:
        in_single = not in_single

    if char == "#" and not in_double and not in_single:
      return line[:i]
  return line


def load_gradle_properties(target_dir):
  """Loads properties from gradle.properties and local.properties."""
  properties = {}
  for filename in ["gradle.properties", "local.properties"]:
    path = os.path.join(target_dir, filename)
    if os.path.exists(path):
      try:
        with open(path, "r", encoding="utf-8", errors="ignore") as f:
          for line in f:
            line = _strip_inline_comment(line, support_quotes=False).strip()
            if line and not line.startswith("#") and "=" in line:
              key, val = line.split("=", 1)
              properties[key.strip()] = val.strip()
      except Exception as e:  # pylint: disable=broad-exception-caught
        print(
            f"Warning: Failed to load properties from {path}: {e}",
            file=sys.stderr,
        )
  return properties


def load_version_catalog(target_dir):
  """Loads versions from gradle/libs.versions.toml if it exists."""
  catalog = {}
  toml_path = os.path.join(target_dir, "gradle", "libs.versions.toml")
  if os.path.exists(toml_path):
    try:
      with open(toml_path, "r", encoding="utf-8", errors="ignore") as f:
        current_section = None
        for line in f:
          line = _strip_inline_comment(line).strip()
          if not line or line.startswith("#"):
            continue
          if line.startswith("[") and line.endswith("]"):
            current_section = line[1:-1].strip()
            continue
          if current_section == "versions" and "=" in line:
            key, val = line.split("=", 1)
            val = val.strip().strip('"').strip("'")
            catalog[key.strip()] = val.strip()
    except Exception as e:  # pylint: disable=broad-exception-caught
      print(
          f"Warning: Failed to load version catalog from {toml_path}: {e}",
          file=sys.stderr,
      )
  return catalog


def load_custom_gradle_versions(target_dir, file_inventory=None):
  """Loads versions from shared Gradle files and project configuration."""
  versions = {}
  if not target_dir or not os.path.isdir(target_dir):
    return versions

  # 1. Identify Candidate Files (Root files are the primary targets)
  candidate_files = []

  # Leaking Strategy: If target_dir is a sub-module, leak up to find the root
  # e.g. /path/to/project/app -> check /path/to/project for root build files
  search_roots = [target_dir]
  parent = os.path.dirname(os.path.normpath(target_dir))
  if parent and parent != target_dir:
    search_roots.append(parent)

  for s_root in search_roots:
    for f in ["build.gradle", "build.gradle.kts"]:
      root_file = os.path.normpath(os.path.join(s_root, f))
      if os.path.exists(root_file) and root_file not in candidate_files:
        candidate_files.append(root_file)

  # Use inventory if provided to avoid redundant file walk
  if file_inventory:
    search_lists = [
        file_inventory.get("gradles", []),
        file_inventory.get("source_code", []),
    ]
    for file_list in search_lists:
      for path in file_list:
        norm_path = os.path.normpath(path)
        if norm_path not in candidate_files:
          file_lower = os.path.basename(norm_path).lower()
          if any(
              name in file_lower
              for name in [
                  "versions.gradle",
                  "dependencies.gradle",
                  "variables.gradle",
              ]
          ):
            candidate_files.append(norm_path)
          elif any(
              d in norm_path.lower() for d in ["build-logic", "buildsrc"]
          ) and file_lower.endswith((".gradle", ".gradle.kts", ".kt")):
            candidate_files.append(norm_path)
  else:
    # Fallback to walk if no inventory provided
    config = scanner.get_scanner_config()
    ignored_dirs = set(config.get("ignored_directories", []))
    for root, dirs, files in os.walk(target_dir):
      dirs[:] = [d for d in dirs if d not in ignored_dirs]
      for f in files:
        if f.lower().endswith((".gradle", ".gradle.kts", ".kt")):
          candidate_files.append(os.path.join(root, f))

  # 2. Parse candidate files for assignments
  for f_path in candidate_files:
    try:
      with open(f_path, "r", encoding="utf-8", errors="ignore") as f:
        content = f.read()
        for pattern in GRADLE_ASSIGNMENT_PATTERNS:
          for key, val in pattern.findall(content):
            # Clean up the key (remove 'ext.' or 'project.' prefixes)
            clean_key = key.split(".")[-1]
            if clean_key not in versions:
              versions[clean_key] = val.strip().strip('"').strip("'")
    except Exception as e:  # pylint: disable=broad-exception-caught
      print(
          f"Warning: Failed to parse Gradle file {f_path}: {e}", file=sys.stderr
      )

  return versions


def _load_json_file(path):
  """Safely loads a JSON file."""
  if os.path.exists(path):
    try:
      with open(path, "r", encoding="utf-8") as f:
        return json.load(f)
    except Exception as e:  # pylint: disable=broad-exception-caught
      print(f"Warning: Failed to load JSON from {path}: {e}", file=sys.stderr)
  return {}


def _read_text_file(path):
  """Safely reads a text file."""
  if os.path.exists(path):
    try:
      with open(path, "r", encoding="utf-8") as f:
        return f.read()
    except Exception as e:  # pylint: disable=broad-exception-caught
      print(f"Warning: Failed to read file {path}: {e}", file=sys.stderr)
  return ""


def _write_json_file(path, data):
  """Safely writes a JSON file."""
  try:
    with open(path, "w", encoding="utf-8") as f:
      json.dump(data, f, indent=4, sort_keys=True)
  except Exception as e:  # pylint: disable=broad-exception-caught
    print(f"Warning: Failed to write JSON to {path}: {e}", file=sys.stderr)


def _write_text_file(path, content):
  """Safely writes a text file."""
  try:
    with open(path, "w", encoding="utf-8") as f:
      f.write(content)
  except Exception as e:  # pylint: disable=broad-exception-caught
    print(f"Warning: Failed to write file {path}: {e}", file=sys.stderr)


def _filter_by_flavor(data, detected_flavors, prioritized_flavors):
  """Recursively filters lists and dicts based on flavor path segments."""
  if isinstance(data, dict):
    filtered_dict = {}
    for k, v in data.items():
      filtered_val = _filter_by_flavor(v, detected_flavors, prioritized_flavors)
      if isinstance(filtered_val, (dict, list)) and not filtered_val:
        continue
      filtered_dict[k] = filtered_val
    return filtered_dict

  elif isinstance(data, list):
    filtered_list = []
    for entry in data:
      # Handle tuple entries (like in codebase_map.all_files)
      entry_str = (
          entry[1]
          if isinstance(entry, tuple) and len(entry) > 1
          else str(entry)
      )

      is_excluded = any(
          f not in prioritized_flavors and f"src/{f}/" in entry_str
          for f in detected_flavors
      )
      if not is_excluded:
        filtered_list.append(entry)
    return filtered_list

  return data  # Fallback for primitive types


def parse_application_modules(
    gradle_files,
    manifest_files,
    gradle_properties,
    version_catalog,
    custom_gradle_versions=None,
):
  """Identifies distinct application modules in the project."""
  modules = []
  lookups = {
      "properties": gradle_properties,
      "catalog": version_catalog,
      "custom": custom_gradle_versions or {},
  }

  # First, check Gradle files for the application plugin
  for g_file in gradle_files:
    try:
      with open(g_file, "r", encoding="utf-8", errors="ignore") as f:
        content = f.read()
        if re.search(GRADLE_APP_PLUGIN_PATTERN, content):
          # Found an app module!
          mod_dir = os.path.dirname(g_file)
          mod_name = os.path.basename(mod_dir)

          # Try to extract details from this Gradle file
          app_id = None
          target_sdk = None
          app_label = None

          id_match = re.search(GRADLE_APP_ID_PATTERN, content)
          if id_match:
            app_id = id_match.group(1)
          else:
            prop_match = re.search(GRADLE_APP_ID_PROP_PATTERN, content)
            if prop_match:
              app_id = lookups["properties"].get(prop_match.group(1))

          sdk_match = re.search(GRADLE_TARGET_SDK_PATTERN, content)
          if sdk_match:
            target_sdk = _resolve_to_int(sdk_match.group(1), lookups)

          # Fallback to namespace if applicationId is missing
          if not app_id:
            ns_match = re.search(GRADLE_NAMESPACE_PATTERN, content)
            if ns_match:
              app_id = ns_match.group(1)
            else:
              ns_prop_match = re.search(GRADLE_NAMESPACE_PROP_PATTERN, content)
              if ns_prop_match:
                app_id = lookups["properties"].get(ns_prop_match.group(1))

          # Fallback to group / groupId if applicationId and namespace are missing
          if not app_id:
            group_match = re.search(GRADLE_GROUP_PATTERN, content)
            if group_match:
              app_id = group_match.group(1)
            else:
              group_prop_match = re.search(GRADLE_GROUP_PROP_PATTERN, content)
              if group_prop_match:
                app_id = lookups["properties"].get(group_prop_match.group(1))

          modules.append({
              "name": mod_name,
              "path": mod_dir,
              "application_id": app_id,
              "target_sdk": target_sdk,
              "app_label": app_label,
          })
    except Exception as e:  # pylint: disable=broad-exception-caught
      print(
          f"Warning: Failed to parse Gradle file {g_file}: {e}", file=sys.stderr
      )

  return modules


def determine_primary_identity(
    app_modules,
    gradle_files,
    manifest_files,
    gradle_properties,
    version_catalog,
    custom_gradle_versions,
    app_dir,
):
  """Heuristically determines the primary app ID and target SDK."""
  primary_id = None
  primary_sdk = None
  primary_label = None

  # 1. Prefer explicitly defined app modules
  for mod in app_modules:
    if mod.get("application_id"):
      primary_id = mod["application_id"]
      primary_sdk = mod["target_sdk"]
      primary_label = mod["app_label"]
      break

  # 2. Fallback to global Gradle properties
  if not primary_id:
    primary_id = (
        gradle_properties.get("applicationId")
        or gradle_properties.get("namespace")
        or gradle_properties.get("group")
        or gradle_properties.get("groupId")
        or gradle_properties.get("GROUP")
    )
  if not primary_sdk:
    primary_sdk = _resolve_to_int(
        gradle_properties.get("targetSdkVersion")
        or gradle_properties.get("targetSdk"),
        {"properties": gradle_properties, "catalog": version_catalog},
    )

  # 3. Fallback to scanning manifests if still missing
  if not primary_id or not primary_sdk:
    for m_file in manifest_files:
      try:
        with open(m_file, "r", encoding="utf-8-sig") as f:
          content = f.read()
          details = extract_manifest_details(content)
          if not primary_id:
            primary_id = details.get("package_name")
          if not primary_sdk:
            primary_sdk = details.get("target_sdk")
          if not primary_label:
            primary_label = details.get("app_label")
      except Exception:  # pylint: disable=broad-exception-caught
        pass

  # 4. Cross-Platform Framework Detection
  if not primary_id or not primary_sdk:
    # Check for Flutter (pubspec.yaml)
    pubspec_path = os.path.join(app_dir, "pubspec.yaml")
    if os.path.exists(pubspec_path):
      try:
        with open(pubspec_path, "r", encoding="utf-8") as f:
          for line in f:
            if line.startswith("name:"):
              primary_id = line.split(":")[1].strip()
              break
        primary_sdk = FRAMEWORK_DEFAULTS.get("flutter.targetSdkVersion")
      except Exception:  # pylint: disable=broad-exception-caught
        pass

    # Check for React Native (package.json)
    pkg_json_path = os.path.join(app_dir, "package.json")
    if os.path.exists(pkg_json_path):
      try:
        with open(pkg_json_path, "r", encoding="utf-8") as f:
          data = json.load(f)
          primary_id = data.get("name")
      except Exception:  # pylint: disable=broad-exception-caught
        pass

    # Check for .NET MAUI / .NET for Android (*.csproj)
    if not primary_id or not primary_sdk:
      dotnet_id, dotnet_sdk, dotnet_label = extract_dotnet_android_details(
          app_dir
      )
      primary_id = primary_id or dotnet_id
      primary_sdk = primary_sdk or dotnet_sdk
      primary_label = primary_label or dotnet_label

  # 5. Last-resort directory fallback if primary_id is still missing
  if not primary_id:
    dir_name = os.path.basename(os.path.normpath(app_dir))
    if dir_name and dir_name not in (".", "/", "\\"):
      primary_id = dir_name

  return primary_id, primary_sdk, primary_label


def _msbuild_property(content, tag):
  """Reads a single MSBuild property value out of a project file."""
  match = re.search(
      rf"<{tag}[^>]*>([^<]+)</{tag}>", content, re.IGNORECASE
  )
  return match.group(1).strip() if match else None


def extract_dotnet_android_details(app_dir):
  """Finds the .NET MAUI / .NET for Android project and reads its identity.

  A .NET for Android app has no Gradle files, and its AndroidManifest.xml carries
  neither the package name nor uses-sdk -- both are MSBuild properties merged into
  the manifest at build time. Without this the audit sees an app with no id and no
  target SDK and reports nothing, which reads as a clean bill of health.

  Args:
    app_dir: Absolute path to the directory containing the app's code.

  Returns:
    A tuple of (application_id, target_sdk, application_title). Any element is
    None when it could not be determined.
  """
  app_id = None
  target_sdk = None
  label = None

  config = scanner.get_scanner_config()
  ignored_dirs = set(config.get("ignored_directories", []))

  candidates = []
  for root, dirs, files in os.walk(app_dir):
    dirs[:] = [d for d in dirs if d not in ignored_dirs]
    for f in files:
      if f.lower().endswith(".csproj"):
        candidates.append(os.path.join(root, f))

  for csproj in sorted(candidates):
    try:
      with open(csproj, "r", encoding="utf-8", errors="ignore") as f:
        content = f.read()
    except Exception:  # pylint: disable=broad-exception-caught
      continue

    content = re.sub(r"<!--.*?-->", "", content, flags=re.DOTALL)

    tfm_match = DOTNET_TFM_PATTERN.search(content)
    if not tfm_match:
      # Not an Android head project (class libraries, test projects).
      continue

    app_id = app_id or _msbuild_property(content, "ApplicationId")
    label = label or _msbuild_property(content, "ApplicationTitle")

    if target_sdk is None:
      # Explicit override wins, then the moniker's platform level, then the
      # workload default for that .NET release.
      explicit = _msbuild_property(content, "AndroidTargetSdkVersion")
      if explicit and explicit.isdigit():
        target_sdk = int(explicit)
      elif tfm_match.group(2):
        target_sdk = int(tfm_match.group(2))
      else:
        target_sdk = DOTNET_ANDROID_TARGET_SDK.get(tfm_match.group(1))

    if app_id and target_sdk and label:
      break

  return app_id, target_sdk, label


def extract_manifest_details(xml_content):
  """Parses AndroidManifest.xml for key policy signals."""
  details = {
      "package_name": None,
      "target_sdk": None,
      "app_label": None,
      "permissions": [],
      "foreground_services": [],
  }

  try:
    # Strip a UTF-8 BOM before parsing. Visual Studio writes one into
    # AndroidManifest.xml by default, and ElementTree rejects it as an invalid
    # token at line 1. The permission scan below is regex-based and survives, so
    # the failure is silent: the manifest still yields permissions while quietly
    # losing the package name, target SDK and app label.
    root = ET.fromstring(xml_content.lstrip("\ufeff"))
    ns = {"android": "http://schemas.android.com/apk/res/android"}

    details["package_name"] = root.get("package")

    # Target SDK
    uses_sdk = root.find("uses-sdk", ns)
    if uses_sdk is not None:
      details["target_sdk"] = uses_sdk.get(
          "{http://schemas.android.com/apk/res/android}targetSdkVersion"
      )

    # App Label
    application = root.find("application", ns)
    if application is not None:
      label = application.get(
          "{http://schemas.android.com/apk/res/android}label"
      )
      if label and label.startswith("@string/"):
        details["app_label"] = label[8:]
      else:
        details["app_label"] = label

    # Permissions
    for tag in ["uses-permission", "uses-permission-sdk-23"]:
      for perm in root.findall(tag, ns):
        name = perm.get("{http://schemas.android.com/apk/res/android}name")
        if name:
          details["permissions"].append(name)

    # Permissions from component declarations
    for tag in [
        "service",
        "receiver",
        "activity",
        "provider",
        "activity-alias",
    ]:
      for component in root.findall(f".//{tag}", ns):
        perm = component.get(
            "{http://schemas.android.com/apk/res/android}permission"
        )
        if perm:
          details["permissions"].append(perm)

        if tag == "provider":
          read_perm = component.get(
              "{http://schemas.android.com/apk/res/android}readPermission"
          )
          if read_perm:
            details["permissions"].append(read_perm)
          write_perm = component.get(
              "{http://schemas.android.com/apk/res/android}writePermission"
          )
          if write_perm:
            details["permissions"].append(write_perm)

    # Foreground Services
    for service in root.findall(".//service", ns):
      # Check for foregroundServiceType
      fgs_type = service.get(
          "{http://schemas.android.com/apk/res/android}foregroundServiceType"
      )
      if fgs_type:
        details["foreground_services"].append({
            "name": service.get(
                "{http://schemas.android.com/apk/res/android}name"
            ),
            "type": fgs_type,
        })
  except Exception as e:  # pylint: disable=broad-exception-caught
    print(f"Warning: Failed to parse XML: {e}", file=sys.stderr)

  return details


def generate_triage_summary(manifest_details, data_safety_scan):
  """Generates a high-level summary of the audit landscape."""
  summary = []
  
  package = manifest_details.get("package_name") or "Unknown"
  summary.append(f"Audit Target: {package}")
  
  sdk = manifest_details.get("target_sdk") or "Unknown"
  summary.append(f"Target SDK: {sdk}")

  perms_count = len(manifest_details.get("permissions", []))
  summary.append(f"Permissions Requested: {perms_count}")

  signals_count = 0
  if isinstance(data_safety_scan, dict):
    inner_scan = data_safety_scan.get("data_safety_scan", {})
    if "data_sources" in inner_scan:
      signals_count = len(inner_scan["data_sources"])
  summary.append(f"Sensitive Data Signals Found: {signals_count}")

  return "\n".join(summary)


def run_scanner_direct(target_dir, file_inventory, output_file):
  """Invokes the scanner directly using the provided inventory."""
  try:
    # This calls the internal scan logic from scanner.py
    results = scanner.perform_scan(target_dir, file_inventory)
    _write_json_file(output_file, results)
    return results
  except Exception as e:  # pylint: disable=broad-exception-caught
    print(f"Error during scan: {e}", file=sys.stderr)
    return {}


def write_agent_prompts(env_data, goal_map, output_data):
  """Generates surgical prompt files for worker agents and infers activation."""
  repo_root = env_data.get("repo_root")
  temp_dir = env_data.get("temp_dir")
  app_dir = env_data.get("app_dir")
  activated_goals = []

  # 1. Load Shared Resources
  policies_path = os.path.join(repo_root, "resources", "policies.json")
  policies_data = _load_json_file(policies_path)
  taxonomy = policies_data.get("data_safety_section", {}).get("taxonomy", {})
  taxonomy_keys = set(taxonomy.keys())

  common_mandates_path = os.path.join(
      repo_root, "resources", "common_mandates.md"
  )
  common_mandates = _read_text_file(common_mandates_path)

  config = scanner.get_scanner_config()
  permission_groups = config.get("permission_groups", {})

  # 2. Write Slices
  if output_data:
    for slice_name in ["manifest_details", "codebase_map"]:
      slice_path = os.path.join(temp_dir, f"{slice_name}.json")
      _write_json_file(slice_path, output_data.get(slice_name, {}))

  # 3. Build Base Context
  scan_results = output_data.get("data_safety_scan", {})
  base_context = scan_results.get("data_safety_scan", {}).copy()
  
  # Also ensure other metadata is included
  base_context["found_urls"] = scan_results.get("found_urls", [])
  base_context["semantic_files"] = scan_results.get("semantic_files", {})

  # Flatten anonymous categories (those with only a "-" key)
  for cat in [
      "exact_alarm",
      "accessibility",
      "foreground_service",
      "disclosure",
  ]:
    if (
        cat in base_context
        and isinstance(base_context[cat], dict)
        and "-" in base_context[cat]
    ):
      base_context[cat] = base_context[cat]["-"]

  manifest_details = output_data.get("manifest_details", {})
  base_context["TARGET_SDK"] = manifest_details.get("target_sdk")
  base_context["APP_NAME"] = output_data.get("app_name")
  base_context["PACKAGE_NAME"] = output_data.get("package_name")
  manifest_perms = manifest_details.get("permissions", [])

  # Populate requested_permissions dictionary based on groups
  requested_perms_dict = {}
  for group_name, perms_to_check in permission_groups.items():
    matched = [p for p in perms_to_check if p in manifest_perms]
    if matched:
      requested_perms_dict[group_name] = matched

  base_context["requested_permissions"] = requested_perms_dict
  base_context["TEMP_DIR"] = temp_dir
  base_context["APP_DIR"] = app_dir
  base_context["REPO_ROOT"] = repo_root

  # 4. Smart Filtering for data_sources (Noise Reduction)
  if "data_sources" in base_context:
    filtered_sources = {}
    for data_type, findings in base_context["data_sources"].items():
      # Deduplicate by File (max 2 per unique file)
      file_counts = {}
      diverse_findings = []
      for f in findings:
        file_path = f.split(":L")[0] if ":L" in f else f
        if file_counts.get(file_path, 0) < 2:
          diverse_findings.append(f)
          file_counts[file_path] = file_counts.get(file_path, 0) + 1

      # Global Cap (max 3 per data type)
      filtered_sources[data_type] = diverse_findings[:3]
    base_context["data_sources"] = filtered_sources

  max_evidence_per_worker = 6

  # Helper to evaluate and save a goal instance
  def _save_goal_instance(instance_name, context, template_content):
    full_template = template_content + "\n\n" + common_mandates
    final_prompt = render_template(full_template, context)

    if "{{ACTIVATE_GOAL}}" in final_prompt:
      final_prompt = final_prompt.replace("{{ACTIVATE_GOAL}}", "")

      goal_input_file = os.path.join(
          temp_dir, f"input_worker_{instance_name}.json"
      )
      _write_json_file(goal_input_file, context)

      output_path = os.path.join(temp_dir, f"prompt_worker_{instance_name}.md")
      _write_text_file(output_path, final_prompt)
      return True
    return False

  # 5. Process Goals Sequentially (Removes Concurrency GIL Overhead & State
  # Copies)
  for goal_name, goal_conf in goal_map.items():
    template_path = os.path.join(
        repo_root, "resources", goal_conf["prompt_file"]
    )
    template_content = _read_text_file(template_path)
    if not template_content:
      continue

    # Identify chunks for this goal
    chunks = []

    if goal_name == "data_safety" and "data_sources" in base_context:
      all_evidence = []
      for data_type, findings in base_context["data_sources"].items():
        if not taxonomy_keys or data_type in taxonomy_keys:
          for finding in findings:
            all_evidence.append((data_type, finding))

      for i in range(0, len(all_evidence), max_evidence_per_worker):
        chunk = all_evidence[i : i + max_evidence_per_worker]
        chunk_dict = {}
        for dt, finding in chunk:
          if dt not in chunk_dict:
            chunk_dict[dt] = {
                "description": taxonomy.get(dt, {}).get(
                    "description", "User data type."
                ),
                "findings": [],
            }
          chunk_dict[dt]["findings"].append(finding)
        chunks.append(chunk_dict)

    if not chunks:
      # Single pass for small goals
      if _save_goal_instance(goal_name, base_context, template_content):
        activated_goals.append(goal_name)
    else:
      # Multipass for large data safety sets
      for idx, chunk_data in enumerate(chunks, 1):
        instance_name = f"{goal_name}_part_{idx}"
        chunk_context = base_context.copy()
        chunk_context["data_sources"] = chunk_data
        chunk_context["GOAL_NAME"] = instance_name
        if _save_goal_instance(instance_name, chunk_context, template_content):
          activated_goals.append(instance_name)

  return activated_goals


def run_aggregation(temp_dir, repo_root):
  """Aggregates and chunks findings for the Critic."""
  worker_pattern = os.path.join(temp_dir, "worker_*.json")
  worker_files = sorted(glob.glob(worker_pattern))

  all_findings = []
  finding_id_counter = 1

  for w_file in worker_files:
    basename = os.path.basename(w_file)
    try:
      with open(w_file, "r", encoding="utf-8") as f:
        w_data = json.load(f)
      findings_list = w_data.get("findings", [])
      if isinstance(findings_list, list):
        for finding in findings_list:
          # Store the source filename so generate_report can associate it
          # correctly
          finding["worker_file"] = basename
          # Assign sequential ID
          finding["finding_id"] = str(finding_id_counter)
          finding_id_counter += 1
          all_findings.append(finding)
    except Exception as e:  # pylint: disable=broad-exception-caught
      print(
          f"Warning: Failed to load worker file {w_file}: {e}", file=sys.stderr
      )

  # Save the master list of all findings
  aggregated_path = os.path.join(temp_dir, "aggregated_findings.json")
  with open(aggregated_path, "w", encoding="utf-8") as f:
    json.dump({"findings": all_findings}, f, indent=4)

  # Filter findings that need Critic evaluation: skip only SUGGESTION findings
  critic_findings = []
  for f in all_findings:
    severity = str(f.get("severity", "SUGGESTION")).upper().strip()
    if severity == "SUGGESTION":
      continue
    critic_findings.append(f)

  # Chunk the findings for the Critic
  chunk_size = 3
  chunks = [
      critic_findings[i : i + chunk_size]
      for i in range(0, len(critic_findings), chunk_size)
  ]

  # Render Critic template for each chunk
  critic_template_path = os.path.join(repo_root, "resources", "critic.md")
  critic_template = ""
  if os.path.exists(critic_template_path):
    with open(critic_template_path, "r", encoding="utf-8") as f:
      critic_template = f.read()

  common_mandates_path = os.path.join(
      repo_root, "resources", "common_mandates.md"
  )
  common_mandates = ""
  if os.path.exists(common_mandates_path):
    with open(common_mandates_path, "r", encoding="utf-8") as f:
      common_mandates = f.read()

  full_template = critic_template + "\n\n" + common_mandates

  # Load contextual values from manifest_details.json
  manifest_path = os.path.join(temp_dir, "manifest_details.json")
  base_context = {}
  if os.path.exists(manifest_path):
    try:
      with open(manifest_path, "r", encoding="utf-8", errors="ignore") as f:
        m_details = json.load(f)
        base_context["APP_DIR"] = m_details.get("app_dir")
    except Exception:  # pylint: disable=broad-exception-caught
      pass

  if not repo_root:
    repo_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))

  base_context["TEMP_DIR"] = temp_dir
  base_context["REPO_ROOT"] = repo_root

  for idx, chunk in enumerate(chunks, 1):
    # Create input_critic_<idx>.json
    chunk_dict = {f["finding_id"]: f for f in chunk}
    chunk_input_path = os.path.join(temp_dir, f"input_critic_{idx}.json")
    with open(chunk_input_path, "w", encoding="utf-8") as f:
      json.dump(chunk_dict, f, indent=4)

    # Render prompt_critic_<idx>.md
    chunk_context = base_context.copy()
    chunk_context["CHUNK_INDEX"] = str(idx)

    final_prompt = render_template(full_template, chunk_context)
    chunk_prompt_path = os.path.join(temp_dir, f"prompt_critic_{idx}.md")
    with open(chunk_prompt_path, "w", encoding="utf-8") as f:
      f.write(final_prompt)

  return len(chunks)


GRADLE_APP_PLUGIN_PATTERN = (
    r"(?:id\s*\(?\s*[\x27\x22]com\.android\.application[\x27\x22]\s*\)?|"
    r"alias\s*\([^)]*android\.application[^)]*\)|"
    r"alias\s*\([^)]*android\.app[^)]*\)|"
    r"apply\s*\(?\s*(?:plugin:\s*)?[\x27\x22]com\.android\.application"
    r"[\x27\x22]\s*\)?)(?!\s*(?:\.?apply\s*\(?\s*false\s*\)?|apply\s*=\s*false))"
)


def _resolve_to_int(val, lookups, visited=None):
  """Recursively resolves a variable name to an integer value."""
  if visited is None:
    visited = set()

  if not val:
    return None

  # Clean and normalize the value
  val_str = str(val).strip().strip('"').strip("'")

  # Support project.property("key") syntax
  prop_match = re.search(
      r'project\.property\s*\(\s*[\'"](.*?)[\'"]\s*\)', val_str
  )
  if prop_match:
    val_str = prop_match.group(1).strip()

  # Robust resolution for Gradle Version Catalogs and dynamic getters
  # Strip common accessors and indirection prefixes
  val_str = re.sub(
      r"\.(?:get|toInt|getOrElse|provider)(?:\s*\([^)]*\))?", "", val_str
  )
  if val_str.startswith("libs.versions."):
    val_str = val_str[14:]
  elif val_str.startswith("libs."):
    val_str = val_str[5:]
  elif val_str.startswith("versions."):
    val_str = val_str[9:]

  if val_str.isdigit():
    return int(val_str)

  if val_str in visited:
    return None  # Circular reference protection
  visited.add(val_str)

  # 1. Direct and Namespaced Lookups
  search_keys = [val_str]
  if "." in val_str:
    # If A.B.C fails, try B.C, then C
    parts = val_str.split(".")
    for i in range(1, len(parts)):
      search_keys.append(".".join(parts[i:]))

  for key in search_keys:
    # Try custom versions first (usually has highest signal)
    if key in lookups.get("custom", {}):
      res = _resolve_to_int(lookups["custom"][key], lookups, visited)
      if res is not None:
        return res

    # Try gradle properties
    if key in lookups.get("properties", {}):
      res = _resolve_to_int(lookups["properties"][key], lookups, visited)
      if res is not None:
        return res

    # Try version catalog
    if key in lookups.get("catalog", {}):
      res = _resolve_to_int(lookups["catalog"][key], lookups, visited)
      if res is not None:
        return res

  return None


def main():
  """Deterministic context-gathering tool for the Reviewer skill."""
  parser = argparse.ArgumentParser(
      description="Deterministic context-gathering and state-query tool."
  )
  subparsers = parser.add_subparsers(dest="command", required=True)

  # Init subcommand
  init_parser = subparsers.add_parser(
      "init", help="Initialize scratch environment."
  )
  init_parser.add_argument("app_dir", help="Path to the Android application.")

  # Aggregate subcommand
  agg_parser = subparsers.add_parser("aggregate", help="Aggregate findings.")
  agg_parser.add_argument(
      "temp_dir", help="Path to the temporary scratch directory."
  )

  args = parser.parse_args()

  repo_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))

  if args.command == "aggregate":
    temp_dir = os.path.abspath(args.temp_dir)
    if not os.path.isdir(temp_dir):
      print(
          f"Error: Specified temporary directory does not exist: {temp_dir}",
          file=sys.stderr,
      )
      sys.exit(1)

    chunk_count = run_aggregation(temp_dir, repo_root)
    print(json.dumps({"temp_dir": temp_dir, "critic_chunks": chunk_count}))
    sys.exit(0)

  # Initialization Mode
  app_dir = os.path.abspath(args.app_dir)
  if not os.path.isdir(app_dir):
    print(f"Error: {app_dir} is not a valid directory.", file=sys.stderr)
    sys.exit(1)

  scratch_id = str(uuid.uuid4())
  workspace_root = os.getcwd()
  temp_dir = os.path.join(
      workspace_root, ".scratch", f"play_policy_insights_{scratch_id}"
  )
  os.makedirs(temp_dir, exist_ok=True)

  # Perform a single master filesystem walk using the scanner
  file_inventory = scanner.collect_target_files(app_dir)

  gradle_properties = load_gradle_properties(app_dir)
  version_catalog = load_version_catalog(app_dir)
  # Pass inventory to avoid redundant walk
  custom_gradle_versions = load_custom_gradle_versions(app_dir, file_inventory)

  codebase_map = {
      "key_files": {
          "manifests": file_inventory["manifests"],
          "gradles": [
              g
              for g in file_inventory["gradles"]
              if g.endswith((".gradle", ".gradle.kts"))
          ],
      }
  }

  # 1. Detect and Filter Flavors (Play Store Prioritization)
  detected_flavors = set()
  src_dirs = []
  
  config = scanner.get_scanner_config()
  ignored_dirs = set(config.get("ignored_directories", []))
  
  # Find all 'src' directories
  for root, dirs, _ in os.walk(app_dir):
    dirs[:] = [d for d in dirs if d not in ignored_dirs]
    if "src" in dirs:
      src_dirs.append(os.path.join(root, "src"))
      dirs.remove("src")  # Don't recurse deeper for 'src' search

  for s_dir in src_dirs:
    for flavor in os.listdir(s_dir):
      if (
          os.path.isdir(os.path.join(s_dir, flavor))
          and flavor not in ignored_dirs
      ):
        detected_flavors.add(flavor)

  prioritized_flavors = []
  if "play" in detected_flavors:
    prioritized_flavors = ["main", "play"]

  if prioritized_flavors:
    # Filter codebase_map to only include prioritized flavors
    print(
        f"Prioritizing flavors: {', '.join(prioritized_flavors)}",
        file=sys.stderr,
    )

    filtered_files = {}
    for key, paths in codebase_map.get("key_files", {}).items():
      filtered_paths = []
      for p in paths:
        # Check if path contains any non-prioritized flavor
        is_excluded = False
        for f in detected_flavors:
          if f not in prioritized_flavors and f"src/{f}/" in p:
            is_excluded = True
            break
        if not is_excluded:
          filtered_paths.append(p)
      filtered_files[key] = filtered_paths
    codebase_map["key_files"] = filtered_files

  with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor:
    # 1. Start the Data Safety Scanner immediately (independent task)
    data_safety_scan_file = os.path.join(temp_dir, "data_safety_scan.json")
    scanner_future = executor.submit(
        run_scanner_direct, app_dir, file_inventory, data_safety_scan_file
    )

    # 2. Extract Application Modules and Primary Identity
    app_modules = parse_application_modules(
        codebase_map.get("key_files", {}).get("gradles", []),
        codebase_map.get("key_files", {}).get("manifests", []),
        gradle_properties,
        version_catalog,
        custom_gradle_versions=custom_gradle_versions,
    )
    primary_app_id, primary_target_sdk, primary_app_label = (
        determine_primary_identity(
            app_modules,
            codebase_map.get("key_files", {}).get("gradles", []),
            codebase_map.get("key_files", {}).get("manifests", []),
            gradle_properties,
            version_catalog,
            custom_gradle_versions,
            app_dir,
        )
    )

    manifest_content = ""
    manifest_details = {
        "package_name": primary_app_id,
        "app_label": primary_app_label,
        "target_sdk": primary_target_sdk,
        "app_dir": app_dir,
        "permissions": [],
        "foreground_services": [],
    }

    if codebase_map.get("key_files", {}).get("manifests"):
      for _, manifest_path in enumerate(codebase_map["key_files"]["manifests"]):
        # Filter Test-Bleed based on relative path
        rel_path = os.path.relpath(manifest_path, app_dir)
        normalized_rel_path = "/" + rel_path.replace("\\", "/")
        if any(
            x in normalized_rel_path
            for x in [
                "/debug/",
                "/test/",
                "/androidTest/",
                "/testFixtures/",
            ]
        ):
          continue

        if os.path.exists(manifest_path):
          try:
            # Manifests are UTF-8. Without an explicit encoding this falls back
            # to the platform default, which on Windows is the locale codepage:
            # the file arrives mojibake and every downstream parse fails.
            with open(manifest_path, "r", encoding="utf-8-sig") as f:
              content = f.read()
              if not manifest_content:
                manifest_content = content

              details = extract_manifest_details(content)

              # Identity Fallbacks
              if details.get("package_name") and not manifest_details.get(
                  "package_name"
              ):
                manifest_details["package_name"] = details["package_name"]

              if details.get("target_sdk") and not manifest_details.get(
                  "target_sdk"
              ):
                manifest_details["target_sdk"] = details["target_sdk"]

              if details.get("app_label") and not manifest_details.get(
                  "app_label"
              ):
                manifest_details["app_label"] = details["app_label"]

              manifest_details["permissions"].extend(
                  details.get("permissions", [])
              )
              manifest_details["foreground_services"].extend(
                  details.get("foreground_services", [])
              )
          except Exception as e:  # pylint: disable=broad-exception-caught
            print(f"Warning: Failed to read manifest: {e}", file=sys.stderr)

      manifest_details["permissions"] = sorted(
          list(set(manifest_details["permissions"]))
      )

      # Deduplicate and sort foreground services
      unique_services = []
      seen_services = set()
      for svc in manifest_details["foreground_services"]:
        svc_key = (svc.get("name"), svc.get("type"))
        if svc_key not in seen_services:
          seen_services.add(svc_key)
          unique_services.append(svc)
      
      # Final stable sort
      manifest_details["foreground_services"] = sorted(
          unique_services,
          key=lambda x: (x.get("name") or "", x.get("type") or ""),
      )

    # 3. Gather results
    data_safety_scan = scanner_future.result()

  # 5. Post-Scan Flavor Filtering
  if prioritized_flavors:
    data_safety_scan = _filter_by_flavor(
        data_safety_scan, detected_flavors, prioritized_flavors
    )

  # 6. Ingest local Play Store info
  local_info_path = os.path.join(
      app_dir, "play_store_assets", "play_store_info.json"
  )
  if os.path.exists(local_info_path):
    try:
      with open(local_info_path, "r", encoding="utf-8") as f:
        play_store_info = json.load(f)
        if "is_published" not in play_store_info:
          play_store_info["is_published"] = True
        if "store_url" not in play_store_info:
          play_store_info["store_url"] = f"file://{local_info_path}"

        with open(
            os.path.join(temp_dir, "play_store_info.json"),
            "w",
            encoding="utf-8",
        ) as out_f:
          json.dump(play_store_info, out_f, indent=4)
      print(
          f"Ingested local Play Store info from {local_info_path}",
          file=sys.stderr,
      )
    except Exception as e:  # pylint: disable=broad-exception-caught
      print(
          f"Warning: Failed to ingest local Play Store info: {e}",
          file=sys.stderr,
      )

  reasoning = []
  goal_map = {}
  resources_dir = os.path.join(repo_root, "resources")
  if os.path.exists(resources_dir):
    for f in os.listdir(resources_dir):
      if f.startswith("goal_") and f.endswith(".md"):
        goal_name = f[5:-3]
        goal_map[goal_name] = {"prompt_file": f}

  if prioritized_flavors:
    reasoning.append(
        f"Prioritizing Play Store flavors: {', '.join(prioritized_flavors)}"
    )
  reasoning.append("Activated via template evaluation")

  output_data = {
      "app_name": (
          manifest_details.get("app_label")
          or os.path.basename(os.path.normpath(app_dir))
      ),
      "package_name": manifest_details.get("package_name"),
      "target_dir": app_dir,
      "codebase_map": codebase_map,
      "manifest_content": manifest_content,
      "manifest_details": manifest_details,
      "application_modules": app_modules,
      "data_safety_scan": data_safety_scan,
      "triage_summary": generate_triage_summary(
          manifest_details, data_safety_scan
      ),
      "triage_reasoning": "\n".join(reasoning),
      "detected_flavors": sorted(list(detected_flavors)),
      "prioritized_flavors": prioritized_flavors,
  }

  try:
    activated_goals = write_agent_prompts(
        {"repo_root": repo_root, "temp_dir": temp_dir, "app_dir": app_dir},
        goal_map,
        output_data,
    )
    print(f"Prompts generated in {temp_dir}", file=sys.stderr)
  except Exception as e:  # pylint: disable=broad-exception-caught
    print(f"Warning: Failed to generate prompts: {e}", file=sys.stderr)

  print(
      json.dumps(
          {"temp_dir": temp_dir, "activated_goals": activated_goals}, indent=4
      )
  )


if __name__ == "__main__":
  main()