"""Production-safe deterministic customer integration engineering agent.

This agent guides an integration from requirements through generated workspace
artifacts and sandbox-style acceptance evidence. It deliberately does not call
an LLM, read provider credentials, execute arbitrary shell, deploy, merge, or
mutate customer systems. All customer credentials stay behind caller-managed
secret references.
"""
from __future__ import annotations

import asyncio
import hashlib
import ipaddress
import json
import re
import socket
import time
from datetime import UTC, datetime
from typing import Any, Literal
from urllib.parse import urlparse

import a2a_pack as a2a
from a2a_pack import (
    A2AAgent,
    ConsumerSetup,
    ConsumerSetupField,
    EgressPolicy,
    NoAuth,
    Pricing,
    Resources,
    RunContext,
    WorkspaceAccess,
    WorkspaceMode,
)
from a2a_pack.context import AgentEvent
from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator


MAX_JSON_BYTES = 256_000
MAX_TEXT_BYTES = 128_000
APPROVAL_PHRASE = "I_APPROVE_THE_LISTED_STEPS"
OUTPUT_ROOT = "outputs/integrations"
SECRET_WORDS = ("secret", "token", "password", "passwd", "apikey", "api_key", "authorization", "bearer")


class StrictModel(BaseModel):
    model_config = ConfigDict(extra="forbid", str_strip_whitespace=True)


class AuditEntry(StrictModel):
    ts: str
    action: str
    integration_id: str
    details: dict[str, Any] = Field(default_factory=dict)


class ArtifactRecord(StrictModel):
    path: str
    sha256: str
    size_bytes: int = Field(ge=0)
    mime_type: str


class SkillStatus(StrictModel):
    status: Literal["ok", "needs_setup", "blocked", "failed"]
    integration_id: str
    dry_run: bool
    audit: list[AuditEntry] = Field(default_factory=list)
    warnings: list[str] = Field(default_factory=list)
    residual_risks: list[str] = Field(default_factory=list)


class SystemSpec(StrictModel):
    name: str = Field(min_length=1, max_length=80)
    kind: Literal[
        "rest_api",
        "webhook",
        "database",
        "object_storage",
        "observability",
        "gitea",
        "kubernetes",
        "argo_cd",
        "queue",
        "other",
    ]
    base_url: str | None = Field(default=None, max_length=2048)
    environment: str = Field(default="sandbox", max_length=64)
    owner: str | None = Field(default=None, max_length=120)
    notes: str | None = Field(default=None, max_length=2000)

    @field_validator("base_url")
    @classmethod
    def _safe_url_text(cls, value: str | None) -> str | None:
        if value is None or not value:
            return None
        parsed = urlparse(value)
        if parsed.scheme not in {"https", "http"} or not parsed.netloc:
            raise ValueError("base_url must be an absolute http(s) URL")
        return value


class AuthMethod(StrictModel):
    system_name: str = Field(min_length=1, max_length=80)
    auth_type: Literal["none", "api_key", "bearer_token", "oauth2_client_credentials", "hmac_signature", "mtls", "basic", "oidc"]
    secret_ref: str | None = Field(
        default=None,
        pattern=r"^[A-Z][A-Z0-9_]{2,127}$",
        description="Caller-managed secret reference name only; never a raw secret value.",
    )
    header_name: str | None = Field(default=None, max_length=80)
    token_url: str | None = Field(default=None, max_length=2048)
    scopes: list[str] = Field(default_factory=list, max_length=20)

    @model_validator(mode="after")
    def _require_secret_ref_for_secret_auth(self) -> "AuthMethod":
        if self.auth_type != "none" and not self.secret_ref:
            raise ValueError("secret_ref is required for non-none auth methods")
        _reject_raw_secret(self.secret_ref or "")
        return self


class FieldMapping(StrictModel):
    source: str = Field(min_length=1, max_length=160)
    target: str = Field(min_length=1, max_length=160)
    transform: str = Field(default="copy", max_length=500)
    required: bool = True
    pii: bool = False


class DataContract(StrictModel):
    name: str = Field(min_length=1, max_length=120)
    direction: Literal["inbound", "outbound", "bidirectional"]
    content_type: str = Field(default="application/json", max_length=120)
    schema_ref: str | None = Field(default=None, max_length=500)
    fields: list[FieldMapping] = Field(default_factory=list, max_length=200)
    example_payload: dict[str, Any] | None = None

    @field_validator("example_payload")
    @classmethod
    def _bounded_example(cls, value: dict[str, Any] | None) -> dict[str, Any] | None:
        if value is not None:
            _ensure_bounded_json(value, MAX_JSON_BYTES)
        return value


class RateLimitSpec(StrictModel):
    system_name: str = Field(min_length=1, max_length=80)
    max_requests: int = Field(ge=1, le=1_000_000)
    per_seconds: int = Field(ge=1, le=86_400)
    burst: int | None = Field(default=None, ge=1, le=1_000_000)
    retry_after_header: str | None = Field(default="Retry-After", max_length=80)


class EnvironmentSpec(StrictModel):
    name: str = Field(min_length=1, max_length=64)
    purpose: Literal["dev", "test", "sandbox", "staging", "prod"]
    endpoint_refs: list[str] = Field(default_factory=list, max_length=20)
    object_storage_ref: str | None = Field(default=None, max_length=128)
    observability_ref: str | None = Field(default=None, max_length=128)
    mutable: bool = False


class ComplianceConstraint(StrictModel):
    name: str = Field(min_length=1, max_length=120)
    category: Literal["pii", "phi", "pci", "soc2", "gdpr", "hipaa", "data_residency", "retention", "audit", "other"]
    requirement: str = Field(min_length=1, max_length=2000)


class SuccessCriterion(StrictModel):
    name: str = Field(min_length=1, max_length=120)
    measurement: str = Field(min_length=1, max_length=500)
    target: str = Field(min_length=1, max_length=200)


class IntegrationRequirements(StrictModel):
    integration_id: str = Field(pattern=r"^[a-z][a-z0-9-]{2,63}$")
    title: str = Field(min_length=1, max_length=160)
    systems: list[SystemSpec] = Field(min_length=1, max_length=12)
    auth_methods: list[AuthMethod] = Field(default_factory=list, max_length=12)
    data_contracts: list[DataContract] = Field(default_factory=list, max_length=20)
    rate_limits: list[RateLimitSpec] = Field(default_factory=list, max_length=20)
    environments: list[EnvironmentSpec] = Field(default_factory=list, max_length=12)
    compliance_constraints: list[ComplianceConstraint] = Field(default_factory=list, max_length=30)
    success_criteria: list[SuccessCriterion] = Field(min_length=1, max_length=20)
    external_host_allowlist: list[str] = Field(default_factory=list, max_length=50)
    repository_write_allowlist: list[str] = Field(default_factory=list, max_length=20)
    notes: str | None = Field(default=None, max_length=4000)
    dry_run: bool = True

    @field_validator("external_host_allowlist")
    @classmethod
    def _clean_hosts(cls, value: list[str]) -> list[str]:
        return [_validate_hostname(host) for host in value]

    @field_validator("repository_write_allowlist")
    @classmethod
    def _clean_repos(cls, value: list[str]) -> list[str]:
        out: list[str] = []
        for repo in value:
            clean = repo.strip()
            if not re.fullmatch(r"[A-Za-z0-9_.-]{1,80}/[A-Za-z0-9_.-]{1,120}", clean):
                raise ValueError("repository allowlist entries must look like owner/repo")
            out.append(clean)
        return out


class DesignIntegrationRequest(StrictModel):
    requirements: IntegrationRequirements
    create_artifacts: bool = True
    dry_run: bool = True


class DesignIntegrationResult(SkillStatus):
    architecture_path: str | None = None
    mapping_path: str | None = None
    plan_digest: str
    artifacts: list[ArtifactRecord] = Field(default_factory=list)
    architecture_summary: str
    field_mapping_count: int = Field(ge=0)
    required_setup: list[str] = Field(default_factory=list)


class GenerateConnectorRequest(StrictModel):
    requirements: IntegrationRequirements
    connector_language: Literal["python", "typescript"] = "python"
    include_tests: bool = True
    dry_run: bool = True
    external_host_allowlist: list[str] = Field(default_factory=list, max_length=50)
    repository_write_allowlist: list[str] = Field(default_factory=list, max_length=20)

    @field_validator("external_host_allowlist")
    @classmethod
    def _clean_hosts(cls, value: list[str]) -> list[str]:
        return [_validate_hostname(host) for host in value]


class GenerateConnectorResult(SkillStatus):
    connector_path: str | None = None
    test_path: str | None = None
    plan_digest: str
    artifacts: list[ArtifactRecord] = Field(default_factory=list)
    generated_files: list[str] = Field(default_factory=list)
    blocked_actions: list[str] = Field(default_factory=list)


class ContractFixture(StrictModel):
    name: str = Field(min_length=1, max_length=120)
    payload: dict[str, Any]
    should_pass: bool = True

    @field_validator("payload")
    @classmethod
    def _bounded_payload(cls, value: dict[str, Any]) -> dict[str, Any]:
        _ensure_bounded_json(value, MAX_JSON_BYTES)
        return value


class ContractArtifact(StrictModel):
    name: str = Field(min_length=1, max_length=120)
    kind: Literal["json_schema", "openapi_3", "webhook"]
    schema_doc: dict[str, Any]
    fixtures: list[ContractFixture] = Field(default_factory=list, max_length=50)
    signature_header: str | None = Field(default=None, max_length=120)
    secret_ref: str | None = Field(default=None, pattern=r"^[A-Z][A-Z0-9_]{2,127}$")

    @field_validator("schema_doc")
    @classmethod
    def _bounded_schema(cls, value: dict[str, Any]) -> dict[str, Any]:
        _ensure_bounded_json(value, MAX_JSON_BYTES)
        return value


class ValidateContractRequest(StrictModel):
    integration_id: str = Field(pattern=r"^[a-z][a-z0-9-]{2,63}$")
    contracts: list[ContractArtifact] = Field(min_length=1, max_length=20)
    fail_on_drift: bool = True
    dry_run: bool = True


class ContractCheck(StrictModel):
    name: str
    kind: str
    status: Literal["pass", "fail", "warn"]
    findings: list[str] = Field(default_factory=list)
    fixture_results: list[dict[str, Any]] = Field(default_factory=list)


class ValidateContractResult(SkillStatus):
    contract_report_path: str | None = None
    artifacts: list[ArtifactRecord] = Field(default_factory=list)
    checks: list[ContractCheck] = Field(default_factory=list)
    drift_detected: bool = False


class EndpointTest(StrictModel):
    name: str = Field(min_length=1, max_length=120)
    method: Literal["GET", "POST", "PUT", "PATCH", "DELETE"] = "POST"
    url: str = Field(max_length=2048)
    expected_status: int = Field(default=200, ge=100, le=599)
    request_json: dict[str, Any] | None = None
    required_secret_refs: list[str] = Field(default_factory=list, max_length=10)
    idempotency_key: str | None = Field(default=None, max_length=160)

    @field_validator("request_json")
    @classmethod
    def _bounded_request(cls, value: dict[str, Any] | None) -> dict[str, Any] | None:
        if value is not None:
            _ensure_bounded_json(value, MAX_JSON_BYTES)
        return value

    @field_validator("required_secret_refs")
    @classmethod
    def _secret_refs_only(cls, value: list[str]) -> list[str]:
        for ref in value:
            if not re.fullmatch(r"[A-Z][A-Z0-9_]{2,127}", ref):
                raise ValueError("secret refs must be uppercase reference names")
            _reject_raw_secret(ref)
        return value


class RunAcceptanceRequest(StrictModel):
    integration_id: str = Field(pattern=r"^[a-z][a-z0-9-]{2,63}$")
    endpoint_tests: list[EndpointTest] = Field(default_factory=list, max_length=20)
    contract_fixtures: list[ContractArtifact] = Field(default_factory=list, max_length=20)
    external_host_allowlist: list[str] = Field(default_factory=list, max_length=50)
    allow_network: bool = False
    max_retries: int = Field(default=1, ge=0, le=3)
    timeout_seconds: float = Field(default=8.0, ge=1.0, le=30.0)
    dry_run: bool = True

    @field_validator("external_host_allowlist")
    @classmethod
    def _clean_hosts(cls, value: list[str]) -> list[str]:
        return [_validate_hostname(host) for host in value]


class AcceptanceStepResult(StrictModel):
    name: str
    status: Literal["pass", "fail", "skipped", "blocked"]
    attempts: int = Field(ge=0)
    detail: str
    idempotency_key: str | None = None


class RunAcceptanceResult(SkillStatus):
    acceptance_report_path: str | None = None
    artifacts: list[ArtifactRecord] = Field(default_factory=list)
    step_results: list[AcceptanceStepResult] = Field(default_factory=list)
    contract_results: list[ContractCheck] = Field(default_factory=list)
    network_calls_made: int = Field(ge=0)


class HandoffSection(StrictModel):
    heading: str = Field(min_length=1, max_length=120)
    body: str = Field(min_length=1, max_length=8000)

    @field_validator("body")
    @classmethod
    def _redact_body(cls, value: str) -> str:
        return redact_secrets(value)


class PrepareHandoffRequest(StrictModel):
    integration_id: str = Field(pattern=r"^[a-z][a-z0-9-]{2,63}$")
    title: str = Field(min_length=1, max_length=160)
    rollout_steps: list[str] = Field(min_length=1, max_length=30)
    rollback_steps: list[str] = Field(min_length=1, max_length=30)
    monitoring_checks: list[str] = Field(min_length=1, max_length=30)
    operator_notes: list[HandoffSection] = Field(default_factory=list, max_length=20)
    delivery_actions: list[str] = Field(default_factory=list, max_length=20)
    plan_digest: str | None = Field(default=None, max_length=128)
    approval_phrase: str | None = Field(default=None, max_length=64)
    approve_delivery: bool = False
    dry_run: bool = True


class PrepareHandoffResult(SkillStatus):
    handoff_path: str | None = None
    plan_digest: str
    artifacts: list[ArtifactRecord] = Field(default_factory=list)
    approval_required: bool
    approval_valid: bool
    delivery_executed: bool = False


class CustomerIntegrationEngineerConfig(StrictModel):
    default_dry_run: bool = True


class CustomerIntegrationEngineer(A2AAgent[CustomerIntegrationEngineerConfig, NoAuth]):
    name = "customer-integration-engineer"
    description = (
        "Guides and validates customer integrations from requirements through a tested handoff while "
        "keeping credentials and production changes under customer control."
    )
    version = "0.1.1"

    config_model = CustomerIntegrationEngineerConfig
    auth_model = NoAuth
    pricing = Pricing(
        price_per_call_usd=0.0,
        caller_pays_llm=False,
        notes="Deterministic no-LLM integration engineering; callers keep credentials and production changes under their control.",
    )
    resources = Resources(cpu="500m", memory="512Mi", max_runtime_seconds=600)
    egress = EgressPolicy(deny_internet_by_default=True)
    workspace_access = WorkspaceAccess.dynamic(
        max_files=256,
        allowed_modes=(WorkspaceMode.READ_ONLY, WorkspaceMode.READ_WRITE_OVERLAY),
        require_reason=True,
        deny_patterns=("**/.env", "**/*secret*", "**/*token*", "**/id_rsa", "**/.git/**"),
        max_total_size_bytes=20 * 1024 * 1024,
    )
    tools_used = ("pydantic", "jsonschema", "httpx")
    consumer_setup = ConsumerSetup.from_fields(
        ConsumerSetupField.config("GITEA_REPOSITORY", label="Gitea repository", description="Optional owner/repo target controlled by the caller.", required=False),
        ConsumerSetupField.secret("GITEA_TOKEN", label="Gitea token", description="Optional caller-managed token reference; never returned or stored.", required=False),
        ConsumerSetupField.config("API_BASE_URL", label="API base URL", description="Optional endpoint used only when allowlisted and dry_run=false.", required=False, input_type="url"),
        ConsumerSetupField.secret("API_AUTH_TOKEN", label="API auth token", description="Optional caller-managed API credential reference.", required=False),
        ConsumerSetupField.config("OBJECT_STORAGE_REF", label="Object storage reference", description="Caller-owned object storage config reference.", required=False),
        ConsumerSetupField.config("OBSERVABILITY_REF", label="Observability reference", description="Caller-owned logs/metrics/traces reference.", required=False),
        ConsumerSetupField.config("KUBERNETES_CONTEXT_REF", label="Kubernetes context reference", description="Optional caller-owned Kubernetes context reference.", required=False),
        ConsumerSetupField.config("ARGOCD_APP_REF", label="Argo CD app reference", description="Optional caller-owned Argo CD app reference.", required=False),
    )

    @a2a.tool(description="Capture requirements, produce architecture and field mapping artifacts", timeout_seconds=120, idempotent=True, max_retries=0, cost_class="deterministic", grant_outputs_prefix="outputs", grant_write_prefixes=("outputs/integrations",))
    async def design_integration(self, ctx: RunContext[NoAuth], request: DesignIntegrationRequest) -> DesignIntegrationResult:
        req = request.requirements.model_copy(
            update={"dry_run": _effective_dry_run(self, request.dry_run, request.requirements.dry_run)}
        )
        audit = [_audit("design_started", req.integration_id, {"dry_run": req.dry_run})]
        await _emit(ctx, audit[-1])
        warnings = _requirements_warnings(req)
        required_setup = _required_setup(req)
        digest = _plan_digest({"phase": "design", "requirements": req.model_dump(mode="json")})
        architecture = _render_architecture(req, digest, warnings, required_setup)
        mapping = _build_mapping(req)
        artifacts: list[ArtifactRecord] = []
        arch_path = mapping_path = None
        if request.create_artifacts:
            base = _integration_base(req.integration_id)
            arch_path = f"{base}/architecture.md"
            mapping_path = f"{base}/mapping.json"
            artifacts.append(await _write_workspace_text(ctx, arch_path, architecture, "text/markdown"))
            artifacts.append(await _write_workspace_json(ctx, mapping_path, mapping))
        audit.append(_audit("design_completed", req.integration_id, {"artifacts": [a.path for a in artifacts]}))
        await _emit(ctx, audit[-1])
        return DesignIntegrationResult(
            status="ok",
            integration_id=req.integration_id,
            dry_run=req.dry_run,
            audit=audit,
            warnings=warnings,
            residual_risks=_residual_risks(req),
            architecture_path=arch_path,
            mapping_path=mapping_path,
            plan_digest=digest,
            artifacts=artifacts,
            architecture_summary=f"{req.title}: {len(req.systems)} systems, {len(req.data_contracts)} contracts, {len(req.success_criteria)} success criteria.",
            field_mapping_count=sum(len(c.fields) for c in req.data_contracts),
            required_setup=required_setup,
        )

    @a2a.tool(description="Generate a minimal workspace-contained connector scaffold with no deployment or repository mutation", timeout_seconds=120, idempotent=True, max_retries=0, cost_class="deterministic", grant_outputs_prefix="outputs", grant_write_prefixes=("outputs/integrations",))
    async def generate_connector(self, ctx: RunContext[NoAuth], request: GenerateConnectorRequest) -> GenerateConnectorResult:
        req = request.requirements.model_copy(
            update={"dry_run": _effective_dry_run(self, request.dry_run, request.requirements.dry_run)}
        )
        audit = [_audit("connector_generation_started", req.integration_id, {"language": request.connector_language, "dry_run": req.dry_run})]
        await _emit(ctx, audit[-1])
        warnings = _requirements_warnings(req)
        blocked_actions: list[str] = []
        if request.repository_write_allowlist or req.repository_write_allowlist:
            blocked_actions.append("repository writes are never performed; generated source is written only under outputs/integrations/{integration_id}/")
        allowlist = sorted(set(req.external_host_allowlist + request.external_host_allowlist))
        digest = _plan_digest({"phase": "generate_connector", "requirements": req.model_dump(mode="json"), "language": request.connector_language})
        base = _integration_base(req.integration_id)
        artifacts: list[ArtifactRecord] = []
        generated_files: list[str] = []
        if request.connector_language == "python":
            connector_path = f"{base}/connector/python/connector.py"
            test_path = f"{base}/connector/python/test_connector.py" if request.include_tests else None
            source = _python_connector(req, allowlist, digest)
            artifacts.append(await _write_workspace_text(ctx, connector_path, source, "text/x-python"))
            generated_files.append(connector_path)
            if test_path:
                test_source = _python_connector_test(req)
                artifacts.append(await _write_workspace_text(ctx, test_path, test_source, "text/x-python"))
                generated_files.append(test_path)
        else:
            connector_path = f"{base}/connector/typescript/connector.ts"
            test_path = f"{base}/connector/typescript/connector.test.ts" if request.include_tests else None
            source = _typescript_connector(req, allowlist, digest)
            artifacts.append(await _write_workspace_text(ctx, connector_path, source, "text/typescript"))
            generated_files.append(connector_path)
            if test_path:
                test_source = _typescript_connector_test(req)
                artifacts.append(await _write_workspace_text(ctx, test_path, test_source, "text/typescript"))
                generated_files.append(test_path)
        audit.append(_audit("connector_generation_completed", req.integration_id, {"files": generated_files, "blocked_actions": blocked_actions}))
        await _emit(ctx, audit[-1])
        status: Literal["ok", "needs_setup", "blocked", "failed"] = "blocked" if blocked_actions and not req.dry_run else "ok"
        return GenerateConnectorResult(
            status=status,
            integration_id=req.integration_id,
            dry_run=req.dry_run,
            audit=audit,
            warnings=warnings,
            residual_risks=_residual_risks(req),
            connector_path=connector_path,
            test_path=test_path,
            plan_digest=digest,
            artifacts=artifacts,
            generated_files=generated_files,
            blocked_actions=blocked_actions,
        )

    @a2a.tool(description="Validate OpenAPI, JSON Schema, and webhook contracts with deterministic fixtures", timeout_seconds=120, idempotent=True, max_retries=0, cost_class="deterministic", grant_outputs_prefix="outputs", grant_write_prefixes=("outputs/integrations",))
    async def validate_contract(self, ctx: RunContext[NoAuth], request: ValidateContractRequest) -> ValidateContractResult:
        dry_run = _effective_dry_run(self, request.dry_run)
        audit = [_audit("contract_validation_started", request.integration_id, {"contracts": len(request.contracts), "dry_run": dry_run})]
        await _emit(ctx, audit[-1])
        checks = [_validate_one_contract(contract) for contract in request.contracts]
        drift = any(check.status == "fail" for check in checks)
        report = {
            "integration_id": request.integration_id,
            "status": "fail" if drift and request.fail_on_drift else "pass",
            "dry_run": dry_run,
            "generated_at": _now(),
            "checks": [c.model_dump(mode="json") for c in checks],
            "secret_handling": "secret_ref values only; raw credentials are rejected and redacted",
        }
        path = f"{_integration_base(request.integration_id)}/contract-report.json"
        artifact = await _write_workspace_json(ctx, path, report)
        audit.append(_audit("contract_validation_completed", request.integration_id, {"drift_detected": drift, "report": path}))
        await _emit(ctx, audit[-1])
        return ValidateContractResult(
            status="failed" if drift and request.fail_on_drift else "ok",
            integration_id=request.integration_id,
            dry_run=dry_run,
            audit=audit,
            warnings=[] if not drift else ["contract drift detected; inspect contract-report.json before handoff"],
            residual_risks=[] if not drift else ["fixtures are deterministic and bounded; live provider behavior still requires caller-controlled staging verification"],
            contract_report_path=path,
            artifacts=[artifact],
            checks=checks,
            drift_detected=drift,
        )

    @a2a.tool(description="Run dry-run or allowlisted acceptance checks using endpoint metadata and secret references", timeout_seconds=180, idempotent=True, max_retries=0, cost_class="deterministic", grant_outputs_prefix="outputs", grant_write_prefixes=("outputs/integrations",))
    async def run_acceptance(self, ctx: RunContext[NoAuth], request: RunAcceptanceRequest) -> RunAcceptanceResult:
        dry_run = _effective_dry_run(self, request.dry_run)
        live_network_block = _live_network_unavailable_reason(self)
        audit = [_audit("acceptance_started", request.integration_id, {"endpoint_tests": len(request.endpoint_tests), "dry_run": dry_run, "allow_network": request.allow_network})]
        await _emit(ctx, audit[-1])
        contract_results = [_validate_one_contract(contract) for contract in request.contract_fixtures]
        step_results: list[AcceptanceStepResult] = []
        network_calls = 0
        warnings: list[str] = []
        for test in request.endpoint_tests:
            validation = await _validate_endpoint_target(test.url, request.external_host_allowlist)
            idem = test.idempotency_key or _plan_digest({"integration_id": request.integration_id, "test": test.model_dump(mode="json")})[:32]
            if validation:
                step_results.append(AcceptanceStepResult(name=test.name, status="blocked", attempts=0, detail=validation, idempotency_key=idem))
                warnings.append(validation)
                continue
            if dry_run or not request.allow_network:
                step_results.append(AcceptanceStepResult(name=test.name, status="skipped", attempts=0, detail="dry_run/default safe mode: external request was validated but not sent", idempotency_key=idem))
                continue
            if live_network_block:
                step_results.append(AcceptanceStepResult(name=test.name, status="blocked", attempts=0, detail=live_network_block, idempotency_key=idem))
                warnings.append(live_network_block)
                continue
            result = await _call_endpoint_with_retries(test, request.max_retries, request.timeout_seconds, idem)
            network_calls += result.attempts
            step_results.append(result)
        report = {
            "integration_id": request.integration_id,
            "dry_run": dry_run,
            "allow_network": request.allow_network,
            "network_calls_made": network_calls,
            "generated_at": _now(),
            "endpoint_results": [s.model_dump(mode="json") for s in step_results],
            "contract_results": [c.model_dump(mode="json") for c in contract_results],
            "safety": {
                "ssrf_protection": "scheme, host allowlist, DNS/IP public-routability, and redirect revalidation enforced",
                "secrets": "only secret references accepted; no raw values stored",
                "payload_limits_bytes": MAX_JSON_BYTES,
                "timeouts_seconds": request.timeout_seconds,
            },
        }
        path = f"{_integration_base(request.integration_id)}/acceptance-report.json"
        artifact = await _write_workspace_json(ctx, path, report)
        failed = any(s.status in {"fail", "blocked"} for s in step_results) or any(c.status == "fail" for c in contract_results)
        audit.append(_audit("acceptance_completed", request.integration_id, {"report": path, "network_calls_made": network_calls, "failed_or_blocked": failed}))
        await _emit(ctx, audit[-1])
        return RunAcceptanceResult(
            status="blocked" if any(s.status == "blocked" for s in step_results) else "failed" if failed else "ok",
            integration_id=request.integration_id,
            dry_run=dry_run,
            audit=audit,
            warnings=sorted(set(warnings)),
            residual_risks=["No production deployment, merge, or environment mutation is performed by this agent."],
            acceptance_report_path=path,
            artifacts=[artifact],
            step_results=step_results,
            contract_results=contract_results,
            network_calls_made=network_calls,
        )

    @a2a.tool(description="Prepare rollout, rollback, monitoring, and operator handoff documentation with approval-gated delivery planning", timeout_seconds=120, idempotent=True, max_retries=0, cost_class="deterministic", grant_outputs_prefix="outputs", grant_write_prefixes=("outputs/integrations",))
    async def prepare_handoff(self, ctx: RunContext[NoAuth], request: PrepareHandoffRequest) -> PrepareHandoffResult:
        dry_run = _effective_dry_run(self, request.dry_run)
        audit = [_audit("handoff_started", request.integration_id, {"dry_run": dry_run, "delivery_actions": len(request.delivery_actions)})]
        await _emit(ctx, audit[-1])
        plan_material = {
            "integration_id": request.integration_id,
            "rollout_steps": request.rollout_steps,
            "rollback_steps": request.rollback_steps,
            "monitoring_checks": request.monitoring_checks,
            "delivery_actions": request.delivery_actions,
        }
        digest = _plan_digest(plan_material)
        approval_required = bool(request.delivery_actions)
        approval_valid = (not approval_required) or (
            request.approve_delivery
            and request.approval_phrase == APPROVAL_PHRASE
            and request.plan_digest == digest
        )
        warnings: list[str] = []
        if approval_required and not approval_valid:
            warnings.append(f"delivery actions require approval_phrase={APPROVAL_PHRASE!r} and plan_digest={digest}")
        if approval_required and approval_valid:
            warnings.append("approval validated for the listed plan, but this agent still does not deploy, merge, or modify customer systems")
        handoff_request = request.model_copy(update={"dry_run": dry_run})
        handoff = _render_handoff(handoff_request, digest, approval_required, approval_valid)
        path = f"{_integration_base(request.integration_id)}/handoff.md"
        artifact = await _write_workspace_text(ctx, path, handoff, "text/markdown")
        audit.append(_audit("handoff_completed", request.integration_id, {"handoff": path, "approval_required": approval_required, "approval_valid": approval_valid}))
        await _emit(ctx, audit[-1])
        return PrepareHandoffResult(
            status="blocked" if approval_required and not approval_valid else "ok",
            integration_id=request.integration_id,
            dry_run=dry_run,
            audit=audit,
            warnings=warnings,
            residual_risks=["Operators must execute rollout or rollback in caller-controlled systems outside this agent."],
            handoff_path=path,
            plan_digest=digest,
            artifacts=[artifact],
            approval_required=approval_required,
            approval_valid=approval_valid,
            delivery_executed=False,
        )


def _now() -> str:
    return datetime.now(UTC).isoformat()


def _effective_dry_run(agent: CustomerIntegrationEngineer, *values: bool) -> bool:
    return bool(agent.config.default_dry_run or any(values))


def _live_network_unavailable_reason(agent: CustomerIntegrationEngineer) -> str | None:
    if getattr(agent.egress, "deny_internet_by_default", False):
        return (
            "live network checks are disabled by this runtime because egress denies internet by default; "
            "acceptance can validate metadata in dry_run mode but cannot contact external endpoints"
        )
    return None


def _audit(action: str, integration_id: str, details: dict[str, Any] | None = None) -> AuditEntry:
    return AuditEntry(ts=_now(), action=action, integration_id=integration_id, details=redact_obj(details or {}))


async def _emit(ctx: RunContext[NoAuth], entry: AuditEntry) -> None:
    await ctx.emit_event(AgentEvent(kind="audit", payload=entry.model_dump(mode="json")))
    await ctx.emit_progress(f"{entry.action}: {entry.integration_id}")


def _integration_base(integration_id: str) -> str:
    if not re.fullmatch(r"[a-z][a-z0-9-]{2,63}", integration_id):
        raise ValueError("invalid integration_id")
    return f"{OUTPUT_ROOT}/{integration_id}"


def _safe_output_path(path: str) -> str:
    clean = path.replace("\\", "/").strip("/")
    if not clean.startswith(f"{OUTPUT_ROOT}/"):
        raise ValueError("writes must stay under outputs/integrations/{integration_id}/")
    if ".." in clean.split("/") or clean.startswith("/"):
        raise ValueError("unsafe output path")
    return clean


async def _write_workspace_text(ctx: RunContext[NoAuth], path: str, text: str, mime_type: str) -> ArtifactRecord:
    clean = _safe_output_path(path)
    data = redact_secrets(text).encode("utf-8")
    if len(data) > MAX_TEXT_BYTES * 4:
        raise ValueError("artifact exceeds bounded text size")
    ws = ctx.workspace
    if hasattr(ws, "is_writable_output") and ws.is_writable_output(clean):  # type: ignore[attr-defined]
        if hasattr(ws, "write_bytes"):
            ws.write_bytes(clean, data)  # type: ignore[attr-defined]
        else:
            view = await ws.open_view(purpose=f"write {clean}", hints=[clean], max_files=1, mode=WorkspaceMode.READ_WRITE_OVERLAY, reason="Persist integration engineering output under outputs/integrations")
            await view.write(clean, data)
    else:
        raise PermissionError("workspace grant does not allow writes under outputs/integrations")
    artifact_name = clean.replace("/", "__")
    ref = await ctx.write_artifact(artifact_name, data, mime_type)
    await ctx.emit_artifact(ref)
    return ArtifactRecord(path=clean, sha256=hashlib.sha256(data).hexdigest(), size_bytes=len(data), mime_type=mime_type)


async def _write_workspace_json(ctx: RunContext[NoAuth], path: str, payload: dict[str, Any]) -> ArtifactRecord:
    safe_payload = redact_obj(payload)
    _ensure_bounded_json(safe_payload, MAX_JSON_BYTES)
    return await _write_workspace_text(ctx, path, json.dumps(safe_payload, indent=2, sort_keys=True) + "\n", "application/json")


def _ensure_bounded_json(value: Any, max_bytes: int) -> None:
    data = json.dumps(value, default=str, sort_keys=True).encode("utf-8")
    if len(data) > max_bytes:
        raise ValueError(f"JSON payload exceeds {max_bytes} bytes")


def _reject_raw_secret(value: str) -> None:
    if not value:
        return
    if any(marker in value.lower() for marker in ("sk-", "ghp_", "xoxb-", "-----begin", "bearer ")):
        raise ValueError("raw credential-looking values are not allowed; use a secret_ref")


def redact_secrets(text: str) -> str:
    redacted = text
    patterns = [
        r"(?i)(authorization\s*[:=]\s*bearer\s+)[A-Za-z0-9._~+/=-]+",
        r"(?i)(api[_-]?key\s*[:=]\s*)[A-Za-z0-9._~+/=-]{8,}",
        r"(?i)(token\s*[:=]\s*)[A-Za-z0-9._~+/=-]{8,}",
        r"(?i)(password\s*[:=]\s*)[^\s,;]+",
        r"sk-[A-Za-z0-9]{12,}",
        r"ghp_[A-Za-z0-9]{12,}",
    ]
    for pattern in patterns:
        redacted = re.sub(pattern, lambda m: (m.group(1) if m.lastindex else "") + "[REDACTED]", redacted)
    return redacted


def redact_obj(value: Any) -> Any:
    if isinstance(value, dict):
        out: dict[str, Any] = {}
        for key, item in value.items():
            if any(word in str(key).lower() for word in SECRET_WORDS):
                out[key] = "[REDACTED_REF]" if isinstance(item, str) else "[REDACTED]"
            else:
                out[key] = redact_obj(item)
        return out
    if isinstance(value, list):
        return [redact_obj(item) for item in value]
    if isinstance(value, str):
        return redact_secrets(value)
    return value


def _plan_digest(payload: dict[str, Any]) -> str:
    safe = redact_obj(payload)
    encoded = json.dumps(safe, sort_keys=True, separators=(",", ":"), default=str).encode("utf-8")
    return hashlib.sha256(encoded).hexdigest()


def _requirements_warnings(req: IntegrationRequirements) -> list[str]:
    warnings: list[str] = []
    if not req.auth_methods:
        warnings.append("no auth methods supplied; confirm this is intentional")
    if not req.data_contracts:
        warnings.append("no data contracts supplied; mapping and validation will be limited")
    if not req.rate_limits:
        warnings.append("no rate limits supplied; connector will use conservative defaults")
    if any(env.purpose == "prod" and env.mutable for env in req.environments):
        warnings.append("production environment marked mutable; this agent will still not mutate it")
    return warnings


def _residual_risks(req: IntegrationRequirements) -> list[str]:
    risks = [
        "Generated connector is minimal and must be reviewed by customer operators before production use.",
        "Provider-specific authentication and observability wiring stay under caller-controlled setup references.",
    ]
    if req.dry_run:
        risks.append("dry_run=true: acceptance evidence is deterministic and non-mutating, not proof of live provider behavior.")
    return risks


def _required_setup(req: IntegrationRequirements) -> list[str]:
    setup = {auth.secret_ref for auth in req.auth_methods if auth.secret_ref}
    for env in req.environments:
        for ref in (env.object_storage_ref, env.observability_ref):
            if ref:
                setup.add(ref)
    return sorted(setup)


def _build_mapping(req: IntegrationRequirements) -> dict[str, Any]:
    return {
        "integration_id": req.integration_id,
        "title": req.title,
        "contracts": [
            {
                "name": c.name,
                "direction": c.direction,
                "content_type": c.content_type,
                "schema_ref": c.schema_ref,
                "fields": [f.model_dump(mode="json") for f in c.fields],
            }
            for c in req.data_contracts
        ],
        "pii_fields": [f"{c.name}.{field.target}" for c in req.data_contracts for field in c.fields if field.pii],
    }


def _render_architecture(req: IntegrationRequirements, digest: str, warnings: list[str], setup: list[str]) -> str:
    systems = "\n".join(f"- **{s.name}** ({s.kind}, {s.environment}) {s.base_url or '(endpoint configured by caller)'}" for s in req.systems)
    auth = "\n".join(f"- {a.system_name}: {a.auth_type}, secret_ref={a.secret_ref or 'none'}" for a in req.auth_methods) or "- No auth methods supplied."
    contracts = "\n".join(f"- {c.name}: {c.direction}, {len(c.fields)} mapped fields, schema_ref={c.schema_ref or 'inline/fixture'}" for c in req.data_contracts) or "- No contracts supplied."
    rates = "\n".join(f"- {r.system_name}: {r.max_requests}/{r.per_seconds}s burst={r.burst or r.max_requests}" for r in req.rate_limits) or "- Conservative default: bounded retries with idempotency keys."
    compliance = "\n".join(f"- {c.category}: {c.name} — {c.requirement}" for c in req.compliance_constraints) or "- No explicit compliance constraints supplied."
    success = "\n".join(f"- {s.name}: {s.measurement} target={s.target}" for s in req.success_criteria)
    warning_text = "\n".join(f"- {w}" for w in warnings) or "- None"
    setup_text = "\n".join(f"- {s}" for s in setup) or "- None"
    return f"""# Integration Architecture: {req.title}

Integration ID: `{req.integration_id}`  
Plan digest: `{digest}`  
Dry run: `{req.dry_run}`

## Systems
{systems}

## Authentication and Secret References
{auth}

Raw credentials are neither accepted nor stored. Operators configure referenced secrets outside this agent.

## Data Contracts and Field Mapping
{contracts}

See `mapping.json` for exact field-level mapping.

## Rate Limits and Retries
{rates}

All mutating calls must carry idempotency keys. Retries are bounded and honor provider retry metadata.

## Environments
{chr(10).join(f"- {e.name}: {e.purpose}, mutable={e.mutable}" for e in req.environments) or '- No environments supplied.'}

## Compliance Constraints
{compliance}

## Success Criteria
{success}

## Required Caller Setup
{setup_text}

## Safety Controls
- Default dry_run=true.
- External requests require explicit host allowlist, DNS/IP validation, redirect revalidation, payload limits, and timeouts.
- Repository writes beyond outputs, deployments, merges, and environment mutations are not performed.
- Delivery actions require `{APPROVAL_PHRASE}` bound to a plan digest and still remain operator-executed outside this agent.

## Warnings
{warning_text}
"""


def _python_connector(req: IntegrationRequirements, allowlist: list[str], digest: str) -> str:
    return f'''"""Minimal connector scaffold for {req.integration_id}.
Generated by customer-integration-engineer. Review before use.
Plan digest: {digest}
"""
from __future__ import annotations

import hashlib
import ipaddress
import json
import socket
from urllib.parse import urlparse

import httpx

ALLOWED_HOSTS = {allowlist!r}
MAX_BYTES = {MAX_JSON_BYTES}
TIMEOUT_SECONDS = 8.0


def validate_url(url: str) -> str:
    parsed = urlparse(url)
    if parsed.scheme != "https" or not parsed.hostname:
        raise ValueError("connector only permits absolute https URLs")
    host = parsed.hostname.lower()
    if host not in ALLOWED_HOSTS:
        raise ValueError(f"host {{host!r}} is not allowlisted")
    for info in socket.getaddrinfo(host, parsed.port or 443, type=socket.SOCK_STREAM):
        ip = ipaddress.ip_address(info[4][0])
        if ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_multicast or ip.is_reserved or ip.is_unspecified:
            raise ValueError(f"host {{host!r}} resolved to unsafe address")
    return url


def idempotency_key(operation: str, payload: dict) -> str:
    material = json.dumps({{"operation": operation, "payload": payload}}, sort_keys=True).encode("utf-8")
    return hashlib.sha256(material).hexdigest()


async def send_json(url: str, payload: dict, *, secret_ref: str | None = None, dry_run: bool = True) -> dict:
    body = json.dumps(payload).encode("utf-8")
    if len(body) > MAX_BYTES:
        raise ValueError("payload too large")
    safe_url = validate_url(url)
    key = idempotency_key("send_json", payload)
    if dry_run:
        return {{"status": "skipped", "reason": "dry_run", "idempotency_key": key, "url": safe_url}}
    # Resolve secret_ref outside this scaffold using your approved secret manager.
    headers = {{"Idempotency-Key": key, "Content-Type": "application/json"}}
    async with httpx.AsyncClient(timeout=TIMEOUT_SECONDS, follow_redirects=False) as client:
        response = await client.post(safe_url, content=body, headers=headers)
    return {{"status_code": response.status_code, "idempotency_key": key}}
'''


def _python_connector_test(req: IntegrationRequirements) -> str:
    return f'''from connector import idempotency_key


def test_idempotency_key_stable():
    payload = {{"integration_id": "{req.integration_id}", "event": "synthetic"}}
    assert idempotency_key("send_json", payload) == idempotency_key("send_json", payload)
'''


def _typescript_connector(req: IntegrationRequirements, allowlist: list[str], digest: str) -> str:
    hosts = json.dumps(allowlist)
    return f'''// Minimal connector scaffold for {req.integration_id}. Plan digest: {digest}
const ALLOWED_HOSTS = new Set<string>({hosts});
const MAX_BYTES = {MAX_JSON_BYTES};

export function validateUrl(raw: string): URL {{
  const url = new URL(raw);
  if (url.protocol !== "https:") throw new Error("connector only permits https URLs");
  if (!ALLOWED_HOSTS.has(url.hostname.toLowerCase())) throw new Error("host is not allowlisted");
  return url;
}}

export async function sendJson(rawUrl: string, payload: unknown, options: {{ dryRun?: boolean, secretRef?: string }} = {{}}) {{
  const url = validateUrl(rawUrl);
  const body = JSON.stringify(payload);
  if (new TextEncoder().encode(body).length > MAX_BYTES) throw new Error("payload too large");
  const idem = await crypto.subtle.digest("SHA-256", new TextEncoder().encode(body));
  const idempotencyKey = Array.from(new Uint8Array(idem)).map(b => b.toString(16).padStart(2, "0")).join("");
  if (options.dryRun !== false) return {{ status: "skipped", reason: "dry_run", idempotencyKey, url: url.toString() }};
  const response = await fetch(url, {{ method: "POST", body, headers: {{ "Content-Type": "application/json", "Idempotency-Key": idempotencyKey }} }});
  return {{ statusCode: response.status, idempotencyKey }};
}}
'''


def _typescript_connector_test(req: IntegrationRequirements) -> str:
    return f'''import {{ validateUrl }} from "./connector";

test("allowlist rejects unknown host for {req.integration_id}", () => {{
  expect(() => validateUrl("https://example.invalid/webhook")).toThrow();
}});
'''


def _validate_one_contract(contract: ContractArtifact) -> ContractCheck:
    findings: list[str] = []
    fixture_results: list[dict[str, Any]] = []
    try:
        if contract.kind == "json_schema":
            from jsonschema import Draft202012Validator
            Draft202012Validator.check_schema(contract.schema_doc)
            validator = Draft202012Validator(contract.schema_doc)
            for fixture in contract.fixtures:
                errors = sorted(validator.iter_errors(fixture.payload), key=lambda e: e.path)
                passed = not errors
                expected = fixture.should_pass
                fixture_results.append({"fixture": fixture.name, "passed": passed, "expected": expected, "errors": [e.message for e in errors[:5]]})
                if passed != expected:
                    findings.append(f"fixture {fixture.name!r} expected should_pass={expected} but got {passed}")
        elif contract.kind == "openapi_3":
            version = str(contract.schema_doc.get("openapi", ""))
            if not version.startswith("3."):
                findings.append("OpenAPI document must declare openapi: 3.x")
            if not isinstance(contract.schema_doc.get("paths"), dict) or not contract.schema_doc.get("paths"):
                findings.append("OpenAPI document must include non-empty paths")
            if "servers" in contract.schema_doc and not isinstance(contract.schema_doc["servers"], list):
                findings.append("OpenAPI servers must be a list")
        elif contract.kind == "webhook":
            if not contract.signature_header:
                findings.append("webhook contract requires signature_header")
            if not contract.secret_ref:
                findings.append("webhook contract requires secret_ref, not a raw secret")
            event_type = contract.schema_doc.get("event_type") or contract.schema_doc.get("type")
            if not event_type:
                findings.append("webhook schema_doc should include event_type or type")
            for fixture in contract.fixtures:
                has_event = "event" in fixture.payload or "type" in fixture.payload
                passed = bool(has_event)
                fixture_results.append({"fixture": fixture.name, "passed": passed, "expected": fixture.should_pass, "errors": [] if passed else ["missing event/type"]})
                if passed != fixture.should_pass:
                    findings.append(f"fixture {fixture.name!r} webhook expectation mismatch")
    except Exception as exc:  # noqa: BLE001
        findings.append(f"contract validation error: {type(exc).__name__}: {exc}")
    status: Literal["pass", "fail", "warn"] = "fail" if findings else "pass"
    return ContractCheck(name=contract.name, kind=contract.kind, status=status, findings=findings, fixture_results=fixture_results)


def _validate_hostname(host: str) -> str:
    clean = host.strip().lower().rstrip(".")
    if not clean or len(clean) > 253:
        raise ValueError("invalid hostname")
    if clean in {"localhost", "metadata.google.internal"} or clean.endswith(".local") or clean.endswith(".internal"):
        raise ValueError("internal hostnames are not allowed")
    try:
        ip = ipaddress.ip_address(clean)
    except ValueError:
        if not re.fullmatch(r"[a-z0-9.-]+", clean) or ".." in clean:
            raise ValueError("hostname contains invalid characters")
        return clean
    if _unsafe_ip(ip):
        raise ValueError("private, loopback, link-local, multicast, reserved, or unspecified IPs are not allowed")
    return clean


def _unsafe_ip(ip: ipaddress._BaseAddress) -> bool:
    return bool(ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_multicast or ip.is_reserved or ip.is_unspecified)


async def _validate_endpoint_target(url: str, allowlist: list[str]) -> str | None:
    parsed = urlparse(url)
    if parsed.scheme != "https" or not parsed.hostname:
        return "endpoint URL must be absolute https with a hostname"
    host = parsed.hostname.lower().rstrip(".")
    if host not in set(allowlist):
        return f"host {host!r} is not in external_host_allowlist"
    try:
        _validate_hostname(host)
    except ValueError as exc:
        return str(exc)
    try:
        infos = await asyncio.to_thread(socket.getaddrinfo, host, parsed.port or 443, type=socket.SOCK_STREAM)
    except socket.gaierror as exc:
        return f"DNS resolution failed for {host!r}: {exc}"
    for info in infos:
        ip = ipaddress.ip_address(info[4][0])
        if _unsafe_ip(ip):
            return f"host {host!r} resolves to unsafe IP {ip}"
    return None


async def _call_endpoint_with_retries(test: EndpointTest, max_retries: int, timeout_seconds: float, idem: str) -> AcceptanceStepResult:
    import httpx

    attempts = 0
    last_detail = "not attempted"
    headers = {"Idempotency-Key": idem, "Content-Type": "application/json"}
    for attempt in range(max_retries + 1):
        attempts = attempt + 1
        try:
            async with httpx.AsyncClient(timeout=timeout_seconds, follow_redirects=False) as client:
                response = await client.request(test.method, test.url, json=test.request_json, headers=headers)
            if 300 <= response.status_code < 400:
                location = response.headers.get("location", "")
                return AcceptanceStepResult(name=test.name, status="blocked", attempts=attempts, detail=f"redirect blocked pending revalidation: {location[:200]}", idempotency_key=idem)
            if response.status_code == test.expected_status:
                return AcceptanceStepResult(name=test.name, status="pass", attempts=attempts, detail=f"received expected status {response.status_code}", idempotency_key=idem)
            last_detail = f"expected {test.expected_status}, got {response.status_code}"
            if response.status_code not in {408, 409, 425, 429, 500, 502, 503, 504}:
                break
            await asyncio.sleep(min(2.0, 0.25 * (attempt + 1)))
        except Exception as exc:  # noqa: BLE001
            last_detail = f"{type(exc).__name__}: {exc}"
            await asyncio.sleep(min(2.0, 0.25 * (attempt + 1)))
    return AcceptanceStepResult(name=test.name, status="fail", attempts=attempts, detail=redact_secrets(last_detail), idempotency_key=idem)


def _render_handoff(request: PrepareHandoffRequest, digest: str, approval_required: bool, approval_valid: bool) -> str:
    operator_sections = "\n\n".join(f"## {s.heading}\n{s.body}" for s in request.operator_notes) or "## Operator Notes\nNo additional operator notes supplied."
    delivery = "\n".join(f"- {item}" for item in request.delivery_actions) or "- No delivery actions requested."
    return f"""# Integration Handoff: {request.title}

Integration ID: `{request.integration_id}`  
Plan digest: `{digest}`  
Dry run: `{request.dry_run}`

## Rollout Plan
{chr(10).join(f"{i+1}. {redact_secrets(step)}" for i, step in enumerate(request.rollout_steps))}

## Rollback Plan
{chr(10).join(f"{i+1}. {redact_secrets(step)}" for i, step in enumerate(request.rollback_steps))}

## Monitoring and Alerting Checks
{chr(10).join(f"- {redact_secrets(check)}" for check in request.monitoring_checks)}

{operator_sections}

## Delivery Actions Requiring Operator Control
{delivery}

Approval required: `{approval_required}`  
Approval valid for this exact digest: `{approval_valid}`

This agent does not deploy, merge, modify customer systems, or execute delivery actions. Operators must perform approved delivery steps in their own controlled systems.

## Evidence Checklist
- Review `architecture.md` and `mapping.json`.
- Review connector source under `connector/`.
- Review `contract-report.json`.
- Review `acceptance-report.json`.
- Confirm all secret references are configured in caller-owned systems.
"""