#!/usr/bin/env python3
"""Manifest-backed, idempotent SQL patch replay for offline deployments."""
from __future__ import annotations

import argparse
import hashlib
import json
import os
import re
import shutil
import subprocess
from pathlib import Path
from time import sleep
from typing import Any, Iterable, Mapping

from common import (
    DeployError,
    load_patch_bundle_definition as load_canonical_patch_bundle_definition,
    log,
    mysql_client,
    require_patch_bundle_environment,
    resolve_patch_sql_files,
    split_sql_statements,
    validate_patch_sql_idempotency,
    write_text_atomic,
)
from deployment_runtime_state import (
    append_stage_summary,
    deploy_artifact_dir,
    runtime_state,
)


PATCH_SQL_TRANSIENT_RETRY_DELAY_SECONDS = 240
PATCH_SQL_MAX_ATTEMPTS = 2
PATCH_SQL_TRANSIENT_CONNECTION_MARKERS = (
    "error 2026",
    "error 2013",
    "unexpected eof",
    "lost connection",
)
PATCH_BUNDLE_OPERATION_SCHEMA = "nexuscrm.patch-bundle-operation/v1"
PATCH_OPERATION_ID_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9_.-]{0,127}")
PATCH_SOURCE_COMMIT_RE = re.compile(r"[0-9a-f]{40}")
PATCH_TREE_DIGEST_RE = re.compile(r"sha256:[0-9a-f]{64}")
PATCH_SAFE_PROCESS_ENV_KEYS = ("SSL_CERT_DIR", "SSL_CERT_FILE")


def require_patch_operation_id(value: str) -> str:
    operation_id = str(value or "").strip()
    if not PATCH_OPERATION_ID_RE.fullmatch(operation_id):
        raise DeployError(
            "patch operation id must match [A-Za-z0-9][A-Za-z0-9_.-]{0,127}"
        )
    return operation_id


def patch_bundle_members_digest(members: Iterable[tuple[str, bytes]]) -> str:
    """Bind canonical relative names and bytes independently of their storage."""
    digest = hashlib.sha256()
    count = 0
    for relative_name, payload in sorted(members, key=lambda item: item[0]):
        relative = str(relative_name).encode("utf-8")
        digest.update(relative)
        digest.update(b"\0")
        digest.update(str(len(payload)).encode("ascii"))
        digest.update(b"\0")
        digest.update(payload)
        digest.update(b"\0")
        count += 1
    if count == 0:
        raise DeployError("patch bundle is empty")
    return "sha256:" + digest.hexdigest()


def patch_bundle_tree_digest(root: Path) -> str:
    """Bind every regular file in a Patch Bundle without following symlinks."""
    bundle = root.expanduser().resolve()
    if root.is_symlink() or not bundle.is_dir():
        raise DeployError(f"patch bundle root is missing or unsafe: {root}")
    members: list[tuple[str, bytes]] = []
    for path in bundle.rglob("*"):
        if path.is_symlink():
            raise DeployError(f"patch bundle refuses a symlink: {path}")
        if path.is_file():
            members.append((path.relative_to(bundle).as_posix(), path.read_bytes()))
    return patch_bundle_members_digest(members)


def require_patch_target_env(envv: Mapping[str, str]) -> dict[str, str]:
    required = (
        "ENV_NAME",
        "TIDB_HOST",
        "TIDB_PORT",
        "TIDB_USER",
        "TIDB_DATABASE",
    )
    missing = [name for name in required if not str(envv.get(name, "")).strip()]
    if missing:
        raise DeployError(
            "patch target runtime env is missing: " + ", ".join(missing)
        )
    return {str(key): str(value) for key, value in envv.items()}


def patch_target_identity_sha256(envv: Mapping[str, str]) -> str:
    projection = "\0".join(
        str(envv[name]).strip()
        for name in ("TIDB_HOST", "TIDB_PORT", "TIDB_DATABASE", "TIDB_USER")
    )
    return hashlib.sha256(projection.encode("utf-8")).hexdigest()


def require_patch_evidence_dir(path: Path) -> Path:
    if path.is_symlink():
        raise DeployError(f"patch evidence directory refuses a symlink: {path}")
    path.mkdir(parents=True, mode=0o700, exist_ok=True)
    resolved = path.resolve()
    if not resolved.is_dir():
        raise DeployError(f"patch evidence directory is unavailable: {resolved}")
    os.chmod(resolved, 0o700)
    return resolved


def seal_patch_bundle_for_operation(
    source: Path, evidence_dir: Path
) -> tuple[Path, str]:
    source_digest = patch_bundle_tree_digest(source)
    sealed = evidence_dir / "sealed-bundle"
    if sealed.is_symlink():
        raise DeployError("sealed Patch Bundle refuses a symlink")
    if sealed.exists():
        if not sealed.is_dir():
            raise DeployError("sealed Patch Bundle path is not a directory")
        sealed_digest = patch_bundle_tree_digest(sealed)
        if sealed_digest != source_digest:
            raise DeployError("sealed Patch Bundle differs from the reviewed source")
        return sealed, sealed_digest

    shutil.copytree(source, sealed, symlinks=True)
    for path in sorted(sealed.rglob("*")):
        if path.is_symlink():
            raise DeployError(f"sealed Patch Bundle refuses a symlink: {path}")
        os.chmod(path, 0o700 if path.is_dir() else 0o600)
    sealed_digest = patch_bundle_tree_digest(sealed)
    if sealed_digest != source_digest:
        raise DeployError("Patch Bundle changed while its operation copy was sealed")
    return sealed, sealed_digest


def require_operation_seal(sealed: Path, expected_digest: str) -> None:
    if patch_bundle_tree_digest(sealed) != expected_digest:
        raise DeployError("sealed Patch Bundle changed during the operation")


def require_patch_dir(args: argparse.Namespace) -> Path:
    if not args.patch_dir:
        raise DeployError("--patch-dir is required for stage apply-patch")
    path = Path(args.patch_dir).expanduser()
    if not path.is_absolute():
        path = Path(__file__).resolve().parents[3] / path
    path = path.resolve()
    if not path.is_dir():
        raise DeployError(f"patch dir not found: {path}")
    return path


def load_patch_bundle_definition(patch_dir: Path) -> dict[str, Any]:
    definition = load_canonical_patch_bundle_definition(patch_dir)
    manifest_path = Path(str(definition["manifest_path"]))
    return {
        **definition,
        "raw": json.loads(manifest_path.read_text()),
    }


def require_patch_bundle_has_no_release_refs(definition: Mapping[str, Any]) -> None:
    refs = definition.get("refs")
    if isinstance(refs, Mapping) and any(
        str(value or "").strip() for value in refs.values()
    ):
        raise DeployError(
            "independent Patch Bundle must not carry application release refs"
        )


def patch_artifact_dir(envv: dict[str, str]) -> Path:
    path = deploy_artifact_dir(envv) / "patch-replay"
    path.mkdir(parents=True, exist_ok=True)
    return path


def is_transient_patch_sql_connection_error(output: str) -> bool:
    lowered = output.lower()
    return any(marker in lowered for marker in PATCH_SQL_TRANSIENT_CONNECTION_MARKERS)


def database_error_location(output: str) -> str:
    match = re.search(
        r"\bERROR\s+(\d+)(?:\s+\([^\r\n)]*\))?\s+at\s+line\s+(\d+)\s*:",
        output,
        re.IGNORECASE,
    )
    if match is None:
        return ""
    return f" database error {match.group(1)} at input line {match.group(2)}"


def patch_process_env(envv: Mapping[str, str]) -> dict[str, str]:
    process_env = {
        "PATH": os.environ.get("PATH", "/usr/local/bin:/usr/bin:/bin"),
        "LANG": "C",
        "LC_ALL": "C",
    }
    for key in PATCH_SAFE_PROCESS_ENV_KEYS:
        value = str(envv.get(key, "") or os.environ.get(key, "")).strip()
        if value:
            process_env[key] = value
    if envv.get("TIDB_PASSWORD"):
        process_env["MYSQL_PWD"] = str(envv["TIDB_PASSWORD"])
    return process_env


def execute_sql_file(envv: dict[str, str], sql_file: Path) -> str:
    args = mysql_client(envv) + [
        f"-h{envv['TIDB_HOST']}",
        f"-P{envv['TIDB_PORT']}",
        f"-u{envv['TIDB_USER']}",
    ]
    args.append(envv["TIDB_DATABASE"])
    process_env = patch_process_env(envv)
    sql_text = sql_file.read_text()
    for attempt in range(1, PATCH_SQL_MAX_ATTEMPTS + 1):
        result = subprocess.run(
            args,
            text=True,
            input=sql_text,
            capture_output=True,
            check=False,
            env=process_env,
        )
        output = (result.stdout or "") + (result.stderr or "")
        if result.returncode == 0:
            return output
        if (
            attempt < PATCH_SQL_MAX_ATTEMPTS
            and is_transient_patch_sql_connection_error(output)
        ):
            log(
                "patch sql transient connection loss "
                f"(attempt {attempt}/{PATCH_SQL_MAX_ATTEMPTS}): {sql_file}; "
                f"retrying in {PATCH_SQL_TRANSIENT_RETRY_DELAY_SECONDS}s"
            )
            sleep(PATCH_SQL_TRANSIENT_RETRY_DELAY_SECONDS)
            continue
        raise DeployError(
            f"patch sql failed: {sql_file.name}{database_error_location(output)}"
        )
    raise DeployError(f"sql file failed after retry: {sql_file}")


def resolve_assertion_sql_file(assertion_id: str, patch_dir: Path, sql_name: str) -> Path:
    """Resolve an assertion's sql file, refusing anything outside the bundle or not read-only.

    `sql_files` entries are covered by validate_patch_sql_idempotency; assertions are a second
    execution path into the same database, so they carry their own guard rails:

      * the path must stay inside the bundle directory — an absolute path or a `..` traversal
        would let a manifest run arbitrary sql from elsewhere in the checkout;
      * every statement must be a SELECT — assertions exist to observe state, and a mutation
        here would run before the idempotency validation that only `sql_files` receives.

    Statements are split with the shared comment- and quote-aware splitter, so a `SELECT`
    appearing inside a comment does not count as an executable statement.
    """
    candidate = Path(sql_name)
    resolved = (patch_dir / candidate).resolve()
    try:
        resolved.relative_to(patch_dir.resolve())
    except ValueError:
        raise DeployError(
            f"patch bundle assertion {assertion_id} sql_file must stay inside the bundle: {sql_name}"
        ) from None
    if not resolved.is_file():
        raise DeployError(
            f"patch bundle assertion {assertion_id} references missing sql file: {resolved}"
        )
    raw_sql = resolved.read_text()
    # split_sql_statements strips block comments, but MySQL/MariaDB executes version comments
    # (`/*! ... */`, `/*+ ... */`). The raw file is what reaches the client, so a stripped
    # comment could smuggle a mutation past a statement-kind check.
    for marker in ("/*!", "/*+"):
        if marker in raw_sql:
            raise DeployError(
                f"patch bundle assertion {assertion_id} sql file must not contain executable "
                f"comments ({marker}): {resolved}"
            )
    # The shared splitter does not model MySQL backslash escapes, so `SELECT 'x\\''; UPDATE ...;`
    # parses as one statement while the client executes two. The whole file reaches the client,
    # so the guard cannot rely on the splitter alone for this boundary. Two lexical rules make
    # the weakness unreachable: without a backslash there is no escape ambiguity, and without an
    # interior semicolon there is no statement delimiter to smuggle a second statement behind.
    if "\\" in raw_sql:
        raise DeployError(
            f"patch bundle assertion {assertion_id} sql file must not contain backslashes: {resolved}"
        )
    body = raw_sql.rstrip()
    if body.endswith(";"):
        body = body[:-1]
    if ";" in body:
        raise DeployError(
            f"patch bundle assertion {assertion_id} sql file must contain a single statement "
            f"with at most one trailing semicolon: {resolved}"
        )
    statements = [item.strip() for item in split_sql_statements(raw_sql) if item.strip()]
    if not statements:
        raise DeployError(
            f"patch bundle assertion {assertion_id} sql file has no executable statement: {resolved}"
        )
    if len(statements) != 1:
        # The parser reads the client's output as a single result set. A second statement would
        # append another header line, which would be read as a data row and silently corrupt the
        # row count in either direction.
        raise DeployError(
            f"patch bundle assertion {assertion_id} sql file must contain exactly one statement, "
            f"found {len(statements)}: {resolved}"
        )
    statement = statements[0]
    if not statement.upper().startswith("SELECT"):
        raise DeployError(
            f"patch bundle assertion {assertion_id} sql file must be read-only; "
            f"statement starts with {statement.split()[0]!r}: {resolved}"
        )
    return resolved


def parse_patch_assertions(definition: dict[str, Any], patch_dir: Path) -> list[dict[str, Any]]:
    """Read the optional `assertions` array from the bundle manifest.

    The replay harness reports a sql file as passed purely from the client exit code and
    never looks at the result set, so a `SELECT`-only "verify" file cannot fail a deploy.
    Assertions close that hole: each one runs a query and is enforced against the parsed
    rows, so an unmet precondition raises and the deploy fails closed instead of letting
    reconcile proceed on unrepaired state.
    """
    raw = definition.get("raw")
    if not isinstance(raw, dict) or "assertions" not in raw:
        return []
    # Keyed on presence, not on `get()` returning None. An explicit `assertions: null` is a
    # malformed declaration, and treating it as absent would silently skip evaluation here while
    # the D3 guard — which keys on presence — refuses the same manifest.
    payload = raw["assertions"]
    if not isinstance(payload, list) or not payload:
        raise DeployError(
            f"patch bundle manifest assertions must be a non-empty array: {definition['manifest_path']}"
        )
    assertions: list[dict[str, Any]] = []
    for index, item in enumerate(payload, start=1):
        if not isinstance(item, dict):
            raise DeployError(f"patch bundle assertion {index} must be an object")
        assertion_id = str(item.get("id", "")).strip() or f"assertion-{index}"
        sql_name = str(item.get("sql_file", "")).strip()
        if not sql_name:
            raise DeployError(f"patch bundle assertion {assertion_id} is missing sql_file")
        sql_path = resolve_assertion_sql_file(assertion_id, patch_dir, sql_name)
        expect_rows = item.get("expect_rows")
        # bool is an int subclass in Python, so `expect_rows: true` would otherwise be read as
        # an expected count of 1 instead of being rejected.
        if expect_rows is not None and (
            isinstance(expect_rows, bool) or not isinstance(expect_rows, int) or expect_rows < 0
        ):
            raise DeployError(
                f"patch bundle assertion {assertion_id} expect_rows must be a non-negative integer"
            )
        # `or {}` would turn a falsy invalid value such as `expect_columns: []` into an empty
        # object, silently discarding the column check while expect_rows still passed the
        # assertion. Only an absent key defaults.
        expect_columns_payload = item.get("expect_columns", {})
        if expect_columns_payload is None or not isinstance(expect_columns_payload, dict):
            raise DeployError(
                f"patch bundle assertion {assertion_id} expect_columns must be an object"
            )
        expect_columns = {str(key): str(value) for key, value in expect_columns_payload.items()}
        if expect_rows is None and not expect_columns:
            raise DeployError(
                f"patch bundle assertion {assertion_id} must declare expect_rows or expect_columns"
            )
        assertions.append(
            {
                "id": assertion_id,
                "sql_file": sql_path,
                "expect_rows": expect_rows,
                "expect_columns": expect_columns,
            }
        )
    return assertions


def validate_patch_bundle_assertions(patch_dir: Path) -> list[dict[str, Any]]:
    """Validate a bundle's assertion block without touching the database.

    Called from the prepare stage so a malformed manifest is rejected before the deploy
    performs anything destructive. apply-patch runs after reset-branch, so validating only
    there would let a bad bundle reset the TiDB branch first and fail afterwards.
    """
    definition = load_patch_bundle_definition(patch_dir)
    return parse_patch_assertions(definition, patch_dir)


def parse_tabular_rows(stdout: str) -> list[dict[str, str]]:
    """Parse one tab-separated result set as emitted by the mysql/mariadb batch client."""
    # Only the trailing newline is discarded. A blank line is a real row whose single column is
    # the empty string — with -B -r that is exactly how it is emitted — and dropping it would
    # report one row as zero, letting an `expect_rows: 0` assertion pass on an unmet state.
    lines = stdout.split("\n")
    if lines and lines[-1] == "":
        # Exactly one trailing empty element, produced by the final newline. Popping more would
        # also swallow a genuine last row whose single column is the empty string.
        lines.pop()
    if not lines:
        return []
    header = lines[0].split("\t")
    rows: list[dict[str, str]] = []
    for row_index, line in enumerate(lines[1:], start=1):
        values = line.split("\t")
        if len(values) != len(header):
            raise DeployError(
                "unexpected patch assertion result shape: "
                f"row_index={row_index} expected_columns={len(header)} "
                f"actual_columns={len(values)}"
            )
        rows.append(dict(zip(header, values)))
    return rows


def query_patch_assertion_rows(envv: dict[str, str], sql_file: Path) -> list[dict[str, str]]:
    """Run an assertion query and return its rows.

    Unlike execute_sql_file this keeps stdout separate from stderr, because client warnings
    on stderr would otherwise be parsed as result rows.
    """
    args = mysql_client(envv) + [
        f"-h{envv['TIDB_HOST']}",
        f"-P{envv['TIDB_PORT']}",
        f"-u{envv['TIDB_USER']}",
        # Batch + raw, matching the repo's tabular-query convention in common.py
        # (db_query_scalar). Batch output is tab-separated, which is what
        # parse_tabular_rows expects; raw disables escaping so values round-trip
        # unchanged. `-N` is deliberately NOT used: the header row carries the column
        # names that expect_columns is matched against.
        "-B",
        "-r",
    ]
    args.append(envv["TIDB_DATABASE"])
    process_env = patch_process_env(envv)
    result = subprocess.run(
        args,
        text=True,
        input=sql_file.read_text(),
        capture_output=True,
        check=False,
        env=process_env,
    )
    if result.returncode != 0:
        output = (result.stdout or "") + (result.stderr or "")
        raise DeployError(
            f"patch assertion query failed: {sql_file.name}"
            f"{database_error_location(output)}"
        )
    return parse_tabular_rows(result.stdout or "")


def evaluate_patch_assertion(
    envv: dict[str, str],
    assertion: dict[str, Any],
    artifact_dir: Path,
) -> dict[str, str]:
    sql_file = assertion["sql_file"]
    assert isinstance(sql_file, Path)
    log(f"evaluating patch assertion: {assertion['id']} ({sql_file.name})")
    rows = query_patch_assertion_rows(envv, sql_file)
    out_file = artifact_dir / f"{sql_file.stem}.assertion.json"
    bounded_evidence = {
        "schema": "nexuscrm.patch-bundle-assertion-evidence/v1",
        "id": str(assertion["id"]),
        "row_count": len(rows),
        "rows_sha256": hashlib.sha256(
            json.dumps(
                rows,
                sort_keys=True,
                separators=(",", ":"),
                ensure_ascii=False,
            ).encode("utf-8")
        ).hexdigest(),
    }
    write_text_atomic(
        out_file, json.dumps(bounded_evidence, indent=2, sort_keys=True) + "\n"
    )
    os.chmod(out_file, 0o600)

    failures: list[str] = []
    expect_rows = assertion["expect_rows"]
    if expect_rows is not None and len(rows) != expect_rows:
        failures.append(f"expected {expect_rows} row(s), got {len(rows)}")
    expect_columns = assertion["expect_columns"]
    assert isinstance(expect_columns, dict)
    if expect_columns:
        if not rows:
            failures.append(f"expected columns {sorted(expect_columns)} but the query returned no rows")
        for row_index, row in enumerate(rows):
            for column, expected in expect_columns.items():
                if column not in row:
                    failures.append(f"row {row_index}: column {column} not in result set")
                elif row[column] != expected:
                    failures.append(
                        f"row {row_index}: column {column} did not match its reviewed expectation"
                    )
    if failures:
        raise DeployError(
            f"patch assertion failed: {assertion['id']} ({sql_file.name}): " + "; ".join(failures)
        )
    return {
        "id": str(assertion["id"]),
        "file": sql_file.name,
        "artifact": out_file.name,
        "rows": str(len(rows)),
        "status": "passed",
    }


def execute_patch_bundle(
    envv: Mapping[str, str],
    patch_dir: Path,
    evidence_dir: Path,
    *,
    operation_id: str,
    source_commit: str,
    source_path: str,
    expected_tree_digest: str = "",
) -> dict[str, object]:
    """Validate and execute one exact Patch Bundle through one evidence seam."""
    target_env = require_patch_target_env(envv)
    operation_id = require_patch_operation_id(operation_id)
    source_commit = str(source_commit or "").strip()
    if not PATCH_SOURCE_COMMIT_RE.fullmatch(source_commit):
        raise DeployError("patch operation source commit must be an exact Git commit")
    source_path = str(source_path or "").strip()
    if not source_path:
        raise DeployError("patch operation source path is required")
    expected_tree_digest = str(expected_tree_digest or "").strip()
    if expected_tree_digest and not PATCH_TREE_DIGEST_RE.fullmatch(
        expected_tree_digest
    ):
        raise DeployError("patch operation expected tree digest is malformed")
    source_definition = load_patch_bundle_definition(patch_dir)
    require_patch_bundle_has_no_release_refs(source_definition)
    require_patch_bundle_environment(
        source_definition,
        environment=target_env["ENV_NAME"],
        patch_dir=patch_dir,
    )
    artifact_dir = require_patch_evidence_dir(evidence_dir)
    sealed_dir, tree_digest = seal_patch_bundle_for_operation(
        patch_dir, artifact_dir
    )
    if expected_tree_digest and tree_digest != expected_tree_digest:
        raise DeployError(
            "sealed Patch Bundle does not match the reviewed Git source tree"
        )
    definition = load_patch_bundle_definition(sealed_dir)
    require_patch_bundle_has_no_release_refs(definition)
    require_patch_bundle_environment(
        definition,
        environment=target_env["ENV_NAME"],
        patch_dir=sealed_dir,
    )
    sql_files = resolve_patch_sql_files(sealed_dir, definition)
    assertions = parse_patch_assertions(definition, sealed_dir)

    for sql_file in sql_files:
        validate_patch_sql_idempotency(sql_file)

    results: list[dict[str, str]] = []
    for sql_file in sql_files:
        require_operation_seal(sealed_dir, tree_digest)
        require_patch_bundle_environment(
            load_patch_bundle_definition(sealed_dir),
            environment=target_env["ENV_NAME"],
            patch_dir=sealed_dir,
        )
        require_operation_seal(sealed_dir, tree_digest)
        log(f"applying patch sql: {sql_file}")
        execute_sql_file(target_env, sql_file)
        results.append(
            {
                "file": sql_file.name,
                "status": "passed",
            }
        )

    assertion_results: list[dict[str, str]] = []
    for assertion in assertions:
        require_operation_seal(sealed_dir, tree_digest)
        require_patch_bundle_environment(
            load_patch_bundle_definition(sealed_dir),
            environment=target_env["ENV_NAME"],
            patch_dir=sealed_dir,
        )
        require_operation_seal(sealed_dir, tree_digest)
        assertion_results.append(
            evaluate_patch_assertion(target_env, assertion, artifact_dir)
        )
    require_operation_seal(sealed_dir, tree_digest)

    summary = {
        "schema": PATCH_BUNDLE_OPERATION_SCHEMA,
        "operation_id": operation_id,
        "source_commit": source_commit,
        "source_path": source_path,
        "bundle_tree_digest": tree_digest,
        "target_environment": target_env["ENV_NAME"],
        "target_identity_sha256": patch_target_identity_sha256(target_env),
        "patch_dir": source_path,
        "manifest_path": str(
            Path(source_path) / Path(str(definition["manifest_path"])).name
        ),
        "title": str(definition["title"]),
        "created_at": str(definition["created_at"]),
        "sql_files": [item["file"] for item in results],
        "status": "passed",
        "results": results,
        "assertions": assertion_results,
    }
    summary_path = artifact_dir / "patch-summary.json"
    write_text_atomic(
        summary_path, json.dumps(summary, indent=2, sort_keys=True) + "\n"
    )
    os.chmod(summary_path, 0o600)
    return summary


def apply_patch_bundle(env_file: Path, patch_dir: Path) -> dict[str, object]:
    """Staging convenience adapter that also records deployment stage evidence."""
    state = runtime_state(env_file)
    envv = state["envv"]
    assert isinstance(envv, dict)
    deploy_id = str(envv.get("DEPLOY_ID", "")).strip()
    summary = execute_patch_bundle(
        envv,
        patch_dir,
        patch_artifact_dir(envv),
        operation_id=f"{deploy_id}-patch",
        source_commit=str(envv.get("DOWNSTREAM_REF", "")),
        source_path=str(patch_dir),
    )
    append_stage_summary(envv, "apply-patch", summary)
    return summary