#!/usr/bin/env bash
set -euo pipefail

ROOT="${AGENTTEAM_NO_ISOLATED_EXECUTION_ROOT:-$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd -P)}"

if ! git -C "${ROOT}" rev-parse --is-inside-work-tree >/dev/null 2>&1; then
  echo "retired execution guardrail: root is not a Git worktree: ${ROOT}" >&2
  exit 1
fi

python3 - "${ROOT}" <<'PY'
from __future__ import annotations

import os
import re
import stat
import subprocess
import sys
from dataclasses import dataclass
from pathlib import Path


@dataclass(frozen=True)
class Rule:
    name: str
    pattern: re.Pattern[str]


root = Path(sys.argv[1]).resolve()

# Build retired spellings from fragments so this guardrail does not need a
# blanket content exemption for the vocabulary it protects.
isolated = "isolat" + "(?:ed|ion)"
execution = "execu" + "tion"
worker = "work" + "er"

rules = (
    Rule(
        "retired-phrase",
        re.compile(rf"\b{isolated}[ _-]+{execution}s?\b", re.IGNORECASE),
    ),
    Rule(
        "retired-identifier",
        re.compile(
            rf"\bisolated(?:_|){execution}(?:_?(?:enabled|mode))?\b",
            re.IGNORECASE,
        ),
    ),
    Rule(
        "retired-worker-scope",
        re.compile(
            rf"\b{execution}[ _-]+{worker}\b|\bExecutionWorkerAccessControl\b",
            re.IGNORECASE,
        ),
    ),
    Rule(
        "retired-runtime-capability",
        re.compile(r"\bSupports" + "Isolated" + "Execution" + r"\b"),
    ),
    Rule(
        "retired-lifecycle-tool",
        re.compile(
            r"""["'`]"""
            + execution
            + r"_(?:start|observe|control|close|report|wait|get|resume|"
            r"submit_result|report_progress|ack|cancel|steer|signal)\b"
        ),
    ),
    Rule(
        "retired-lifecycle-identifier",
        re.compile(
            r"\b(?:ExecutionEnvelope|EventExecutionPacket|executionLaunchOwner|"
            r"ExecutionHandler|ExecutionOption|ExecutionService|ExecutionJobs|"
            r"NewExecutionHandler|ExecutionToolDefs|"
            r"WithExecutionService|WithExecutionJobs|"
            r"WithExecutionStartResumeTools|WithExecutionResultTools|"
            r"waiting_on_"
            + execution
            + r"|AgentExecutionPolicySection|ExecutionPolicyDisplay|"
            r"AgentExecutionResponse|AgentExecutionRequest|"
            r"useAgentExecutionSettings|agentIsolatedExecution)\b"
        ),
    ),
    Rule(
        "retired-tool-switch",
        re.compile(
            r"\b(?:ToolAccessSettings|MachineAgentToolAccess|"
            r"AgentToolAccess(?:UpdateRequest|Response|Update)?|ToolAccessStore|"
            r"MCPToolsEnabled|CLIToolsEnabled)\b|"
            r"\b(?:mcp|cli)_tools_enabled\b|"
            r"""["'`]toolAccess["'`]|/tool-access\b"""
        ),
    ),
    Rule(
        "retired-execution-workdir",
        re.compile(
            r"""["'`]executions["'`]|\b(?:executionResidue|executionsRoot|DrainAndJoin)\b"""
        ),
    ),
)

historical_migrations = {
    "internal/tidb/migrations/00175_execution_policy_contract.sql",
    "internal/tidb/migrations/00176_account_execution_policy_user_column.sql",
    "internal/tidb/migrations/00201_user_preferences.sql",
    "internal/tidb/migrations/00204_agent_tool_switches.sql",
    "internal/tidb/migrations/00205_user_preferences_default_isolated_execution_disabled.sql",
    "internal/tidb/migrations/00207_message_pins.sql",
    "internal/tidb/migrations/00210_agent_visible_reply_contract.sql",
    "internal/tidb/migrations/00213_user_preferences_visible_reply_contract.sql",
    "internal/tidb/migrations/00215_agent_isolated_execution_default_disabled.sql",
    "internal/tidb/migrations/00313_drop_isolated_execution_contract.sql",
}

allow: dict[str, frozenset[str]] = {
    **{
        path: frozenset(
            {"retired-phrase", "retired-identifier", "retired-tool-switch"}
        )
        for path in historical_migrations
    },
    # Schema cleanup and migration-history regressions.
    "internal/tidb/isolated_execution_schema_test.go": frozenset(
        {"retired-phrase", "retired-identifier", "retired-tool-switch"}
    ),
    "internal/tidb/migrate_test.go": frozenset(
        {"retired-phrase", "retired-identifier", "retired-tool-switch"}
    ),
    "internal/tidb/admin_test.go": frozenset(
        {"retired-phrase", "retired-identifier"}
    ),
    # The guardrail owner must name its exact allowed paths and public filename.
    "guardrails/desktop/check-no-isolated-execution.sh": frozenset(
        {
            "retired-phrase",
            "retired-identifier",
            "retired-lifecycle-identifier",
            "retired-tool-switch",
            "retired-execution-workdir",
        }
    ),
    "guardrails/desktop/check-no-isolated-execution-test.sh": frozenset(
        {
            "retired-phrase",
            "retired-identifier",
            "retired-tool-switch",
            "retired-execution-workdir",
        }
    ),
    "guardrails/desktop/README.md": frozenset({"retired-phrase"}),
    "scripts/harness-fast.sh": frozenset({"retired-phrase"}),
}


def git_paths(*arguments: str) -> set[str]:
    try:
        result = subprocess.run(
            ["git", "-C", str(root), "ls-files", "-z", *arguments],
            check=True,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
        )
    except (OSError, subprocess.CalledProcessError) as error:
        detail = getattr(error, "stderr", b"").decode("utf-8", errors="replace")
        raise RuntimeError(f"git ls-files {' '.join(arguments)} failed: {detail.strip()}") from error
    return {
        raw.decode("utf-8", errors="surrogateescape")
        for raw in result.stdout.split(b"\0")
        if raw
    }


def tracked_paths() -> list[str]:
    # `git ls-files` includes index entries deleted from the worktree. Exclude
    # those paths so a deletion can remove retired vocabulary without making
    # this guardrail try to read a file that intentionally no longer exists.
    return sorted(git_paths("--cached") - git_paths("--deleted"))


def read_tracked_text(relative: str) -> str | None:
    path = root / relative
    try:
        info = path.lstat()
    except OSError as error:
        raise RuntimeError(f"cannot inspect tracked path {relative}: {error}") from error
    if stat.S_ISLNK(info.st_mode):
        try:
            return os.readlink(path)
        except OSError as error:
            raise RuntimeError(f"cannot read tracked symlink {relative}: {error}") from error
    if not stat.S_ISREG(info.st_mode):
        return None
    try:
        data = path.read_bytes()
    except OSError as error:
        raise RuntimeError(f"cannot read tracked path {relative}: {error}") from error
    if b"\0" in data:
        return None
    return data.decode("utf-8", errors="replace")


def allowed(relative: str, rule: Rule) -> bool:
    return rule.name in allow.get(relative, frozenset())


violations: list[str] = []
try:
    paths = tracked_paths()
    for relative in paths:
        for rule in rules:
            if rule.pattern.search(relative) and not allowed(relative, rule):
                violations.append(f"{rule.name}: {relative}:path")
        text = read_tracked_text(relative)
        if text is None:
            continue
        for line_number, line in enumerate(text.splitlines(), 1):
            for rule in rules:
                if rule.pattern.search(line) and not allowed(relative, rule):
                    violations.append(
                        f"{rule.name}: {relative}:{line_number}"
                    )
except RuntimeError as error:
    print(
        f"retired execution/tool-switch guardrail failed closed: {error}",
        file=sys.stderr,
    )
    raise SystemExit(1)

if violations:
    print("Retired execution/tool-switch surface guardrail failed:", file=sys.stderr)
    for violation in violations:
        print(f"  - {violation}", file=sys.stderr)
    raise SystemExit(1)

print(
    f"Retired execution/tool-switch surface guardrail passed ({len(paths)} tracked paths)."
)
PY