"""Deterministic revenue recovery preparation agent.

This agent is deliberately read-only and dry-run only. It analyzes bounded,
authorized subscription snapshots, prepares approval-ready recovery plans, and
persists redacted artifacts under outputs/revenue-recovery/{run_id}/.
It never charges, refunds, changes subscriptions, sends messages, updates CRM,
or calls external providers.
"""
from __future__ import annotations

import hashlib
import json
import re
from datetime import UTC, datetime
from enum import Enum
from typing import Any, Literal

from pydantic import BaseModel, ConfigDict, Field, HttpUrl, field_validator, model_validator

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


Money = float
MAX_RECORDS_PER_SNAPSHOT = 500
SUPPORTED_CURRENCIES = {"USD", "EUR", "GBP", "CAD", "AUD"}
PROTECTED_TRAIT_KEYS = {
    "age",
    "birthdate",
    "birthday",
    "disability",
    "ethnicity",
    "gender",
    "health",
    "marital_status",
    "nationality",
    "pregnancy",
    "race",
    "religion",
    "sex",
    "sexual_orientation",
    "veteran_status",
}
PII_KEYS = {"email", "phone", "name", "address", "ip", "card", "bank", "ssn", "tax_id"}
DANGEROUS_TEXT = re.compile(
    r"(?i)(ignore\s+previous|system\s+prompt|developer\s+message|<script|javascript:|curl\s+|wget\s+|http://169\.254|file://|\.\./)"
)
SAFE_ID = re.compile(r"^[A-Za-z0-9_.:-]{1,128}$")


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


class LeakageType(str, Enum):
    payment_failure = "payment_failure"
    entitlement_mismatch = "entitlement_mismatch"
    involuntary_churn = "involuntary_churn"
    plan_leakage = "plan_leakage"
    data_quality = "data_quality"


class RiskLevel(str, Enum):
    low = "low"
    medium = "medium"
    high = "high"


class RecoveryAction(str, Enum):
    collect_payment_update = "collect_payment_update"
    reconcile_entitlement = "reconcile_entitlement"
    save_offer_review = "save_offer_review"
    plan_true_up_review = "plan_true_up_review"
    data_quality_review = "data_quality_review"
    suppress = "suppress"


class OutputFile(StrictModel):
    path: str
    artifact_uri: str | None = None
    sha256: str
    bytes: int = Field(ge=0)


class AuditEvent(StrictModel):
    timestamp: str
    event_type: str
    run_id: str
    detail: dict[str, Any] = Field(default_factory=dict)


class SafetyFinding(StrictModel):
    code: str
    severity: Literal["info", "warning", "critical"]
    message: str
    affected_tokens: list[str] = Field(default_factory=list, max_length=100)


class BillingRecord(StrictModel):
    invoice_id: str = Field(min_length=1, max_length=128)
    customer_id: str = Field(min_length=1, max_length=128)
    amount_due: Money = Field(ge=0, le=1_000_000)
    currency: str = Field(min_length=3, max_length=3)
    status: Literal["paid", "failed", "open", "void", "refunded", "disputed"]
    failure_code: str | None = Field(default=None, max_length=80)
    due_days_ago: int = Field(default=0, ge=0, le=3650)

    @field_validator("currency")
    @classmethod
    def _currency(cls, value: str) -> str:
        clean = value.upper()
        if clean not in SUPPORTED_CURRENCIES:
            raise ValueError(f"unsupported currency: {clean}")
        return clean


class SubscriptionRecord(StrictModel):
    customer_id: str = Field(min_length=1, max_length=128)
    subscription_id: str = Field(min_length=1, max_length=128)
    plan_id: str = Field(min_length=1, max_length=128)
    status: Literal["active", "past_due", "cancelled", "trialing"]
    monthly_recurring_revenue: Money = Field(ge=0, le=1_000_000)
    paid_seats: int = Field(default=0, ge=0, le=100_000)
    entitled_seats: int = Field(default=0, ge=0, le=100_000)
    active_seats: int = Field(default=0, ge=0, le=100_000)
    included_units: int = Field(default=0, ge=0, le=100_000_000)
    unit_price: Money = Field(default=0, ge=0, le=100_000)
    currency: str = Field(default="USD", min_length=3, max_length=3)

    @field_validator("currency")
    @classmethod
    def _currency(cls, value: str) -> str:
        clean = value.upper()
        if clean not in SUPPORTED_CURRENCIES:
            raise ValueError(f"unsupported currency: {clean}")
        return clean


class UsageRecord(StrictModel):
    customer_id: str = Field(min_length=1, max_length=128)
    active_seats_observed: int = Field(default=0, ge=0, le=100_000)
    units_used: int = Field(default=0, ge=0, le=100_000_000)
    last_active_days_ago: int = Field(default=0, ge=0, le=3650)


class CRMRecord(StrictModel):
    customer_id: str = Field(min_length=1, max_length=128)
    account_tier: Literal["self_serve", "growth", "enterprise"] = "self_serve"
    consent_to_contact: bool = True
    suppressed: bool = False
    jurisdiction: str = Field(default="US", min_length=2, max_length=32)
    lifecycle_stage: Literal["customer", "prospect", "former_customer"] = "customer"
    customer_risk: RiskLevel = RiskLevel.low
    raw_notes: str | None = Field(default=None, max_length=1000)

    @field_validator("raw_notes")
    @classmethod
    def _reject_prompt_injection(cls, value: str | None) -> str | None:
        if value and DANGEROUS_TEXT.search(value):
            raise ValueError("raw_notes contained unsafe instruction-like text")
        return value


class SupportRecord(StrictModel):
    customer_id: str = Field(min_length=1, max_length=128)
    open_dispute: bool = False
    open_support_ticket: bool = False
    sentiment: RiskLevel = RiskLevel.low
    tags: list[str] = Field(default_factory=list, max_length=25)

    @field_validator("tags")
    @classmethod
    def _safe_tags(cls, values: list[str]) -> list[str]:
        clean: list[str] = []
        for value in values:
            tag = str(value).strip().lower()[:64]
            if DANGEROUS_TEXT.search(tag):
                raise ValueError("support tag contained unsafe instruction-like text")
            if tag:
                clean.append(tag)
        return clean


class SnapshotBundle(StrictModel):
    billing: list[BillingRecord] = Field(default_factory=list, max_length=MAX_RECORDS_PER_SNAPSHOT)
    subscriptions: list[SubscriptionRecord] = Field(default_factory=list, max_length=MAX_RECORDS_PER_SNAPSHOT)
    usage: list[UsageRecord] = Field(default_factory=list, max_length=MAX_RECORDS_PER_SNAPSHOT)
    crm: list[CRMRecord] = Field(default_factory=list, max_length=MAX_RECORDS_PER_SNAPSHOT)
    support: list[SupportRecord] = Field(default_factory=list, max_length=MAX_RECORDS_PER_SNAPSHOT)
    metadata: dict[str, str] = Field(default_factory=dict, max_length=50)

    @model_validator(mode="after")
    def _minimum_data(self) -> "SnapshotBundle":
        total = sum(len(getattr(self, name)) for name in ("billing", "subscriptions", "usage", "crm", "support"))
        if total > MAX_RECORDS_PER_SNAPSHOT * 5:
            raise ValueError("snapshot is too large")
        for key in self.metadata:
            lowered = key.strip().lower()
            if lowered in PROTECTED_TRAIT_KEYS or any(pii in lowered for pii in PII_KEYS):
                raise ValueError(f"metadata key is not allowed: {key}")
        return self


class IntegrationConfig(StrictModel):
    mode: Literal["synthetic", "consumer_configured"] = "synthetic"
    billing_endpoint: HttpUrl | None = None
    crm_endpoint: HttpUrl | None = None
    warehouse_endpoint: HttpUrl | None = None
    object_storage_endpoint: HttpUrl | None = None


class PolicyConfig(StrictModel):
    dry_run: bool = True
    require_consent: bool = True
    suppressed_customer_ids: list[str] = Field(default_factory=list, max_length=500)
    suppressed_jurisdictions: list[str] = Field(default_factory=lambda: ["UNKNOWN"], max_length=100)
    brand_rules: list[str] = Field(default_factory=list, max_length=50)
    maximum_account_amount: Money = Field(default=250_000, gt=0, le=1_000_000)

    @model_validator(mode="after")
    def _must_be_dry_run(self) -> "PolicyConfig":
        if self.dry_run is not True:
            raise ValueError("this agent only supports dry_run=true")
        return self


class AnalyzeRevenueLeakageInput(StrictModel):
    tenant_id: str = Field(min_length=3, max_length=128)
    run_id: str | None = Field(default=None, min_length=8, max_length=80)
    use_synthetic_data: bool = True
    snapshots: SnapshotBundle | None = None
    integration: IntegrationConfig = Field(default_factory=IntegrationConfig)
    policy: PolicyConfig = Field(default_factory=PolicyConfig)
    confidence_level: float = Field(default=0.90, ge=0.5, le=0.99)
    max_accounts: int = Field(default=100, ge=1, le=500)


class LeakageFinding(StrictModel):
    finding_id: str
    customer_token: str
    leakage_type: LeakageType
    recoverable_amount: Money = Field(ge=0)
    currency: str
    confidence: float = Field(ge=0, le=1)
    confidence_interval_low: Money = Field(ge=0)
    confidence_interval_high: Money = Field(ge=0)
    evidence_codes: list[str] = Field(default_factory=list, max_length=20)
    blockers: list[str] = Field(default_factory=list, max_length=20)
    recommended_action: RecoveryAction


class AnalyzeRevenueLeakageOutput(StrictModel):
    run_id: str
    status: Literal["completed", "rejected"]
    dry_run: bool
    external_actions_performed: Literal[False]
    tenant_token: str
    findings: list[LeakageFinding] = Field(default_factory=list, max_length=500)
    totals_by_currency: dict[str, Money] = Field(default_factory=dict)
    unknowns: list[str] = Field(default_factory=list, max_length=100)
    safety_findings: list[SafetyFinding] = Field(default_factory=list, max_length=100)
    audit_events: list[AuditEvent] = Field(default_factory=list, max_length=200)
    output_files: list[OutputFile] = Field(default_factory=list, max_length=10)


class PrioritizeAccountsInput(StrictModel):
    tenant_id: str = Field(min_length=3, max_length=128)
    run_id: str | None = Field(default=None, min_length=8, max_length=80)
    analysis: AnalyzeRevenueLeakageOutput | None = None
    use_synthetic_data: bool = True
    policy: PolicyConfig = Field(default_factory=PolicyConfig)
    max_accounts: int = Field(default=50, ge=1, le=250)


class PrioritizedAccount(StrictModel):
    rank: int = Field(ge=1)
    customer_token: str
    priority_score: float = Field(ge=0, le=100)
    recoverable_amount: Money = Field(ge=0)
    currency: str
    leakage_types: list[LeakageType] = Field(default_factory=list, max_length=10)
    likelihood: float = Field(ge=0, le=1)
    customer_risk: RiskLevel
    policy_blockers: list[str] = Field(default_factory=list, max_length=20)


class PrioritizeAccountsOutput(StrictModel):
    run_id: str
    status: Literal["completed", "rejected"]
    dry_run: bool
    external_actions_performed: Literal[False]
    prioritized_accounts: list[PrioritizedAccount] = Field(default_factory=list, max_length=250)
    scoring_formula: str
    unknowns: list[str] = Field(default_factory=list, max_length=100)
    safety_findings: list[SafetyFinding] = Field(default_factory=list, max_length=100)
    audit_events: list[AuditEvent] = Field(default_factory=list, max_length=200)
    output_files: list[OutputFile] = Field(default_factory=list, max_length=10)


class ProposeRecoveryInput(StrictModel):
    tenant_id: str = Field(min_length=3, max_length=128)
    run_id: str | None = Field(default=None, min_length=8, max_length=80)
    prioritized: PrioritizeAccountsOutput | None = None
    analysis: AnalyzeRevenueLeakageOutput | None = None
    use_synthetic_data: bool = True
    policy: PolicyConfig = Field(default_factory=PolicyConfig)
    max_plans: int = Field(default=25, ge=1, le=100)


class RecoveryPlan(StrictModel):
    plan_id: str
    customer_token: str
    proposed_action: RecoveryAction
    amount: Money = Field(ge=0)
    currency: str
    rationale: str
    respectful_message_draft: str
    approval_required: Literal[True]
    action_digest: str
    prohibited_actions: list[str] = Field(default_factory=list, max_length=20)


class ProposeRecoveryOutput(StrictModel):
    run_id: str
    status: Literal["completed", "rejected"]
    dry_run: bool
    external_actions_performed: Literal[False]
    recovery_plans: list[RecoveryPlan] = Field(default_factory=list, max_length=100)
    unknowns: list[str] = Field(default_factory=list, max_length=100)
    safety_findings: list[SafetyFinding] = Field(default_factory=list, max_length=100)
    audit_events: list[AuditEvent] = Field(default_factory=list, max_length=200)
    output_files: list[OutputFile] = Field(default_factory=list, max_length=10)


class ValidateRecoveryInput(StrictModel):
    tenant_id: str = Field(min_length=3, max_length=128)
    run_id: str | None = Field(default=None, min_length=8, max_length=80)
    proposed: ProposeRecoveryOutput | None = None
    use_synthetic_data: bool = True
    policy: PolicyConfig = Field(default_factory=PolicyConfig)


class PlanValidation(StrictModel):
    plan_id: str
    customer_token: str
    approved_for_preparation: bool
    blocked: bool
    blockers: list[str] = Field(default_factory=list, max_length=20)
    required_future_approval_phrase: str
    immutable_plan_digest: str


class ValidateRecoveryOutput(StrictModel):
    run_id: str
    status: Literal["completed", "rejected"]
    dry_run: bool
    external_actions_performed: Literal[False]
    validations: list[PlanValidation] = Field(default_factory=list, max_length=100)
    compliance_summary: dict[str, int] = Field(default_factory=dict)
    unknowns: list[str] = Field(default_factory=list, max_length=100)
    safety_findings: list[SafetyFinding] = Field(default_factory=list, max_length=100)
    audit_events: list[AuditEvent] = Field(default_factory=list, max_length=200)
    output_files: list[OutputFile] = Field(default_factory=list, max_length=10)


class GenerateCampaignPackInput(StrictModel):
    tenant_id: str = Field(min_length=3, max_length=128)
    run_id: str | None = Field(default=None, min_length=8, max_length=80)
    analysis: AnalyzeRevenueLeakageOutput | None = None
    prioritized: PrioritizeAccountsOutput | None = None
    proposed: ProposeRecoveryOutput | None = None
    validation: ValidateRecoveryOutput | None = None
    use_synthetic_data: bool = True
    policy: PolicyConfig = Field(default_factory=PolicyConfig)


class GenerateCampaignPackOutput(StrictModel):
    run_id: str
    status: Literal["completed", "rejected"]
    dry_run: bool
    external_actions_performed: Literal[False]
    campaign_pack_markdown: str
    measurement_plan: dict[str, Any]
    output_files: list[OutputFile] = Field(default_factory=list, max_length=20)
    unknowns: list[str] = Field(default_factory=list, max_length=100)
    safety_findings: list[SafetyFinding] = Field(default_factory=list, max_length=100)
    audit_events: list[AuditEvent] = Field(default_factory=list, max_length=200)


class RevenueRecoveryAgentConfig(StrictModel):
    default_currency: str = "USD"


class RevenueRecoveryAgent(A2AAgent[RevenueRecoveryAgentConfig, NoAuth]):
    name = "revenue-recovery-agent"
    description = (
        "Identifies recoverable subscription revenue and prepares respectful, "
        "policy-compliant recovery actions without charging, contacting customers, "
        "or mutating external systems. Synthetic adapters are used by default."
    )
    version = "0.1.0"

    config_model = RevenueRecoveryAgentConfig
    auth_model = NoAuth

    pricing = Pricing(
        price_per_call_usd=0.0,
        caller_pays_llm=False,
        notes="Deterministic dry-run analysis. No LLM calls and no external actions.",
    )
    resources = Resources(cpu="500m", memory="512Mi", max_runtime_seconds=600)
    egress = EgressPolicy(allow_hosts=(), deny_internet_by_default=True)
    tools_used = ("deterministic-reconciliation", "synthetic-adapters")
    consumer_setup = ConsumerSetup.from_fields(
        ConsumerSetupField.config(
            "BILLING_ENDPOINT",
            label="Billing API endpoint",
            description="Optional HTTPS endpoint metadata for future caller-configured read-only adapters. Not called by this agent.",
            required=False,
            input_type="url",
        ),
        ConsumerSetupField.secret(
            "BILLING_TOKEN",
            label="Billing API token",
            description="Optional caller-owned token metadata for future read-only adapters. Not read unless consumer_configured mode is implemented.",
            required=False,
        ),
        ConsumerSetupField.config("CRM_ENDPOINT", required=False, input_type="url"),
        ConsumerSetupField.config("WAREHOUSE_ENDPOINT", required=False, input_type="url"),
        ConsumerSetupField.config("OBJECT_STORAGE_ENDPOINT", required=False, input_type="url"),
    )
    workspace_access = WorkspaceAccess.dynamic(
        max_files=64,
        allowed_modes=(WorkspaceMode.READ_ONLY, WorkspaceMode.READ_WRITE_OVERLAY),
        require_reason=False,
        deny_patterns=("**/.env", "**/*secret*", "**/*credential*", "**/*token*"),
        max_total_size_bytes=10 * 1024 * 1024,
    )

    @a2a.tool(
        description="Analyze bounded authorized snapshots for recoverable subscription revenue leakage and write leakage.json.",
        timeout_seconds=600,
        idempotent=True,
        cost_class="deterministic",
    )
    async def analyze_revenue_leakage(
        self,
        ctx: RunContext[NoAuth],
        request: AnalyzeRevenueLeakageInput,
    ) -> AnalyzeRevenueLeakageOutput:
        run_id = _run_id(request.run_id, request.tenant_id, "analyze", request.model_dump(mode="json"))
        audit = [_audit(run_id, "analysis_started", {"dry_run": True})]
        await _emit(ctx, audit[-1])
        safety = _preflight(request.tenant_id, request.integration, request.policy)
        if _has_critical(safety):
            return AnalyzeRevenueLeakageOutput(
                run_id=run_id,
                status="rejected",
                dry_run=True,
                external_actions_performed=False,
                tenant_token=_token(request.tenant_id, request.tenant_id),
                safety_findings=safety,
                audit_events=audit,
            )
        snapshots = _snapshots(request.use_synthetic_data, request.snapshots)
        findings, unknowns = _analyze_snapshots(request.tenant_id, snapshots, request.policy, request.confidence_level)
        findings = findings[: request.max_accounts]
        totals = _totals(findings)
        output = AnalyzeRevenueLeakageOutput(
            run_id=run_id,
            status="completed",
            dry_run=True,
            external_actions_performed=False,
            tenant_token=_token(request.tenant_id, request.tenant_id),
            findings=findings,
            totals_by_currency=totals,
            unknowns=unknowns,
            safety_findings=safety,
            audit_events=audit + [_audit(run_id, "analysis_completed", {"findings": len(findings)})],
        )
        output.output_files = [await _persist_json(ctx, run_id, "leakage.json", output.model_dump(mode="json"))]
        await _emit(ctx, output.audit_events[-1])
        return output

    @a2a.tool(
        description="Prioritize recoverable accounts by value, likelihood, customer risk, and policy constraints; write prioritized-accounts.json.",
        timeout_seconds=600,
        idempotent=True,
        cost_class="deterministic",
    )
    async def prioritize_accounts(
        self,
        ctx: RunContext[NoAuth],
        request: PrioritizeAccountsInput,
    ) -> PrioritizeAccountsOutput:
        analysis = request.analysis or _offline_analysis(request.tenant_id, request.run_id, request.use_synthetic_data, request.policy)
        run_id = _run_id(request.run_id or analysis.run_id, request.tenant_id, "prioritize", analysis.model_dump(mode="json"))
        audit = [_audit(run_id, "prioritization_started", {})]
        await _emit(ctx, audit[-1])
        accounts = _prioritize(analysis, request.policy, request.max_accounts)
        output = PrioritizeAccountsOutput(
            run_id=run_id,
            status="completed" if analysis.status == "completed" else "rejected",
            dry_run=True,
            external_actions_performed=False,
            prioritized_accounts=accounts,
            scoring_formula="min(100, value_score*0.45 + likelihood*35 + low_risk_bonus - blocker_penalty)",
            unknowns=analysis.unknowns,
            safety_findings=analysis.safety_findings,
            audit_events=audit + [_audit(run_id, "prioritization_completed", {"accounts": len(accounts)})],
        )
        output.output_files = [await _persist_json(ctx, run_id, "prioritized-accounts.json", output.model_dump(mode="json"))]
        await _emit(ctx, output.audit_events[-1])
        return output

    @a2a.tool(
        description="Prepare account-specific remediation recommendations and respectful message drafts; write recovery-plans.json.",
        timeout_seconds=600,
        idempotent=True,
        cost_class="deterministic",
    )
    async def propose_recovery(
        self,
        ctx: RunContext[NoAuth],
        request: ProposeRecoveryInput,
    ) -> ProposeRecoveryOutput:
        analysis = request.analysis or _offline_analysis(request.tenant_id, request.run_id, request.use_synthetic_data, request.policy)
        prioritized = request.prioritized or _offline_prioritized(request.tenant_id, request.run_id, analysis, request.policy, request.max_plans)
        run_id = _run_id(request.run_id or prioritized.run_id, request.tenant_id, "propose", prioritized.model_dump(mode="json"))
        audit = [_audit(run_id, "proposal_started", {})]
        await _emit(ctx, audit[-1])
        plans = _propose(analysis, prioritized, request.max_plans)
        output = ProposeRecoveryOutput(
            run_id=run_id,
            status="completed" if prioritized.status == "completed" else "rejected",
            dry_run=True,
            external_actions_performed=False,
            recovery_plans=plans,
            unknowns=sorted(set(analysis.unknowns + ["No external messages are sent; drafts require separate human approval."])),
            safety_findings=analysis.safety_findings,
            audit_events=audit + [_audit(run_id, "proposal_completed", {"plans": len(plans)})],
        )
        output.output_files = [await _persist_json(ctx, run_id, "recovery-plans.json", output.model_dump(mode="json"))]
        await _emit(ctx, output.audit_events[-1])
        return output

    @a2a.tool(
        description="Validate proposed recovery plans against consent, suppression, jurisdiction, dispute, and brand rules; write compliance-report.json.",
        timeout_seconds=600,
        idempotent=True,
        cost_class="deterministic",
    )
    async def validate_recovery(
        self,
        ctx: RunContext[NoAuth],
        request: ValidateRecoveryInput,
    ) -> ValidateRecoveryOutput:
        proposed = request.proposed or _offline_proposed(request.tenant_id, request.run_id, request.use_synthetic_data, request.policy)
        run_id = _run_id(request.run_id or proposed.run_id, request.tenant_id, "validate", proposed.model_dump(mode="json"))
        audit = [_audit(run_id, "validation_started", {})]
        await _emit(ctx, audit[-1])
        validations = _validate_plans(proposed, request.policy)
        output = ValidateRecoveryOutput(
            run_id=run_id,
            status="completed" if proposed.status == "completed" else "rejected",
            dry_run=True,
            external_actions_performed=False,
            validations=validations,
            compliance_summary={
                "plans_reviewed": len(validations),
                "blocked": sum(1 for item in validations if item.blocked),
                "approval_ready": sum(1 for item in validations if item.approved_for_preparation),
            },
            unknowns=proposed.unknowns,
            safety_findings=proposed.safety_findings,
            audit_events=audit + [_audit(run_id, "validation_completed", {"validations": len(validations)})],
        )
        output.output_files = [await _persist_json(ctx, run_id, "compliance-report.json", output.model_dump(mode="json"))]
        await _emit(ctx, output.audit_events[-1])
        return output

    @a2a.tool(
        description="Generate an approval-ready campaign pack and measurement plan; write campaign-pack.md and measurement-plan.json.",
        timeout_seconds=600,
        idempotent=True,
        cost_class="deterministic",
    )
    async def generate_campaign_pack(
        self,
        ctx: RunContext[NoAuth],
        request: GenerateCampaignPackInput,
    ) -> GenerateCampaignPackOutput:
        analysis = request.analysis or _offline_analysis(request.tenant_id, request.run_id, request.use_synthetic_data, request.policy)
        prioritized = request.prioritized or _offline_prioritized(request.tenant_id, request.run_id, analysis, request.policy, 50)
        proposed = request.proposed or _offline_proposed_from(request.tenant_id, request.run_id, analysis, prioritized, request.policy)
        validation = request.validation or _offline_validation(request.tenant_id, request.run_id, proposed, request.policy)
        run_id = _run_id(request.run_id or validation.run_id, request.tenant_id, "campaign", validation.model_dump(mode="json"))
        audit = [_audit(run_id, "campaign_pack_started", {})]
        await _emit(ctx, audit[-1])
        markdown = _campaign_markdown(run_id, analysis, prioritized, proposed, validation)
        measurement = _measurement_plan(run_id, analysis, prioritized, validation)
        output = GenerateCampaignPackOutput(
            run_id=run_id,
            status="completed",
            dry_run=True,
            external_actions_performed=False,
            campaign_pack_markdown=markdown,
            measurement_plan=measurement,
            unknowns=sorted(set(analysis.unknowns + proposed.unknowns + validation.unknowns)),
            safety_findings=analysis.safety_findings + proposed.safety_findings + validation.safety_findings,
            audit_events=audit + [_audit(run_id, "campaign_pack_completed", {"files": 2})],
        )
        output.output_files = [
            await _persist_text(ctx, run_id, "campaign-pack.md", markdown, "text/markdown"),
            await _persist_json(ctx, run_id, "measurement-plan.json", measurement),
        ]
        await _emit(ctx, output.audit_events[-1])
        return output


def _snapshots(use_synthetic: bool, provided: SnapshotBundle | None) -> SnapshotBundle:
    if provided is not None:
        return provided
    if not use_synthetic:
        raise ValueError("snapshots are required when use_synthetic_data=false")
    return _synthetic_snapshots()


def _synthetic_snapshots() -> SnapshotBundle:
    return SnapshotBundle(
        billing=[
            BillingRecord(invoice_id="inv_1001", customer_id="cust_alpha", amount_due=240.0, currency="USD", status="failed", failure_code="card_declined", due_days_ago=12),
            BillingRecord(invoice_id="inv_1002", customer_id="cust_beta", amount_due=480.0, currency="USD", status="paid", due_days_ago=0),
            BillingRecord(invoice_id="inv_1003", customer_id="cust_gamma", amount_due=1200.0, currency="USD", status="open", due_days_ago=45),
            BillingRecord(invoice_id="inv_1003", customer_id="cust_gamma", amount_due=1200.0, currency="USD", status="open", due_days_ago=45),
            BillingRecord(invoice_id="inv_1004", customer_id="cust_delta", amount_due=300.0, currency="USD", status="paid", due_days_ago=0),
            BillingRecord(invoice_id="inv_1005", customer_id="cust_echo", amount_due=90.0, currency="USD", status="failed", failure_code="insufficient_funds", due_days_ago=5),
        ],
        subscriptions=[
            SubscriptionRecord(customer_id="cust_alpha", subscription_id="sub_a", plan_id="growth", status="past_due", monthly_recurring_revenue=240, paid_seats=5, entitled_seats=5, active_seats=5, included_units=1000, unit_price=0.2),
            SubscriptionRecord(customer_id="cust_beta", subscription_id="sub_b", plan_id="growth", status="active", monthly_recurring_revenue=480, paid_seats=10, entitled_seats=10, active_seats=14, included_units=2000, unit_price=0.25),
            SubscriptionRecord(customer_id="cust_gamma", subscription_id="sub_c", plan_id="enterprise", status="cancelled", monthly_recurring_revenue=1200, paid_seats=25, entitled_seats=25, active_seats=21, included_units=5000, unit_price=0.3),
            SubscriptionRecord(customer_id="cust_delta", subscription_id="sub_d", plan_id="starter", status="active", monthly_recurring_revenue=300, paid_seats=3, entitled_seats=3, active_seats=3, included_units=500, unit_price=0.5),
            SubscriptionRecord(customer_id="cust_echo", subscription_id="sub_e", plan_id="starter", status="past_due", monthly_recurring_revenue=90, paid_seats=1, entitled_seats=1, active_seats=1, included_units=250, unit_price=0.4),
        ],
        usage=[
            UsageRecord(customer_id="cust_alpha", active_seats_observed=5, units_used=800, last_active_days_ago=1),
            UsageRecord(customer_id="cust_beta", active_seats_observed=14, units_used=1900, last_active_days_ago=2),
            UsageRecord(customer_id="cust_gamma", active_seats_observed=20, units_used=3000, last_active_days_ago=3),
            UsageRecord(customer_id="cust_delta", active_seats_observed=3, units_used=1100, last_active_days_ago=0),
            UsageRecord(customer_id="cust_echo", active_seats_observed=1, units_used=150, last_active_days_ago=6),
        ],
        crm=[
            CRMRecord(customer_id="cust_alpha", account_tier="growth", consent_to_contact=True, suppressed=False, jurisdiction="US", customer_risk=RiskLevel.low),
            CRMRecord(customer_id="cust_beta", account_tier="growth", consent_to_contact=True, suppressed=False, jurisdiction="US", customer_risk=RiskLevel.low),
            CRMRecord(customer_id="cust_gamma", account_tier="enterprise", consent_to_contact=True, suppressed=False, jurisdiction="US", customer_risk=RiskLevel.high),
            CRMRecord(customer_id="cust_delta", account_tier="self_serve", consent_to_contact=True, suppressed=False, jurisdiction="US", customer_risk=RiskLevel.medium),
            CRMRecord(customer_id="cust_echo", account_tier="self_serve", consent_to_contact=False, suppressed=True, jurisdiction="US", customer_risk=RiskLevel.medium),
        ],
        support=[
            SupportRecord(customer_id="cust_alpha", open_dispute=False, sentiment=RiskLevel.low),
            SupportRecord(customer_id="cust_beta", open_dispute=False, sentiment=RiskLevel.low),
            SupportRecord(customer_id="cust_gamma", open_dispute=True, open_support_ticket=True, sentiment=RiskLevel.high, tags=["billing_dispute"]),
            SupportRecord(customer_id="cust_delta", open_dispute=False, sentiment=RiskLevel.medium),
            SupportRecord(customer_id="cust_echo", open_dispute=False, sentiment=RiskLevel.low),
        ],
        metadata={"source": "deterministic_synthetic_fixture", "schema_version": "2026-07-13"},
    )


def _preflight(tenant_id: str, integration: IntegrationConfig | None, policy: PolicyConfig) -> list[SafetyFinding]:
    findings: list[SafetyFinding] = []
    if not SAFE_ID.fullmatch(tenant_id):
        findings.append(SafetyFinding(code="unsafe_tenant_id", severity="critical", message="tenant_id contains unsupported characters"))
    if not policy.dry_run:
        findings.append(SafetyFinding(code="dry_run_required", severity="critical", message="dry_run must be true"))
    if integration:
        for name in ("billing_endpoint", "crm_endpoint", "warehouse_endpoint", "object_storage_endpoint"):
            value = getattr(integration, name)
            if value is None:
                continue
            url = str(value)
            if not url.startswith("https://"):
                findings.append(SafetyFinding(code="endpoint_not_https", severity="critical", message=f"{name} must use HTTPS"))
            if _looks_internal_url(url):
                findings.append(SafetyFinding(code="ssrf_blocked", severity="critical", message=f"{name} points to a blocked internal address"))
        if integration.mode == "consumer_configured":
            findings.append(SafetyFinding(code="synthetic_adapter_only", severity="warning", message="Consumer-configured integrations are declared but not called; synthetic/provided snapshots are used."))
    return findings


def _looks_internal_url(url: str) -> bool:
    lowered = url.lower()
    blocked = ("localhost", "127.0.0.1", "0.0.0.0", "169.254.", "::1", ".local", "metadata.google", "metadata.aws")
    return any(item in lowered for item in blocked)


def _has_critical(findings: list[SafetyFinding]) -> bool:
    return any(item.severity == "critical" for item in findings)


def _analyze_snapshots(tenant_id: str, snapshots: SnapshotBundle, policy: PolicyConfig, confidence_level: float) -> tuple[list[LeakageFinding], list[str]]:
    unknowns: list[str] = []
    by_customer_sub = {s.customer_id: s for s in snapshots.subscriptions}
    by_customer_usage = {u.customer_id: u for u in snapshots.usage}
    by_customer_crm = {c.customer_id: c for c in snapshots.crm}
    by_customer_support = {s.customer_id: s for s in snapshots.support}
    findings: list[LeakageFinding] = []
    seen_invoices: set[str] = set()
    duplicate_invoices: set[str] = set()

    for invoice in snapshots.billing:
        if invoice.invoice_id in seen_invoices:
            duplicate_invoices.add(invoice.invoice_id)
            findings.append(_finding(tenant_id, invoice.customer_id, LeakageType.data_quality, 0, invoice.currency, 0.99, confidence_level, ["duplicate_invoice"], ["duplicate_prevented"], RecoveryAction.data_quality_review))
            continue
        seen_invoices.add(invoice.invoice_id)
        crm = by_customer_crm.get(invoice.customer_id)
        support = by_customer_support.get(invoice.customer_id)
        blockers = _policy_blockers(invoice.customer_id, crm, support, policy)
        if invoice.status == "failed" and invoice.amount_due > 0:
            findings.append(_finding(tenant_id, invoice.customer_id, LeakageType.payment_failure, min(invoice.amount_due, policy.maximum_account_amount), invoice.currency, _confidence([invoice, by_customer_sub.get(invoice.customer_id), crm, support]), confidence_level, ["failed_invoice", f"due_{invoice.due_days_ago}_days"], blockers, RecoveryAction.collect_payment_update))
        elif invoice.status == "open" and invoice.due_days_ago >= 30 and invoice.amount_due > 0:
            findings.append(_finding(tenant_id, invoice.customer_id, LeakageType.involuntary_churn, min(invoice.amount_due, policy.maximum_account_amount), invoice.currency, _confidence([invoice, by_customer_sub.get(invoice.customer_id), crm, support]) - 0.08, confidence_level, ["open_invoice_over_30_days"], blockers, RecoveryAction.save_offer_review))

    for customer_id, sub in by_customer_sub.items():
        usage = by_customer_usage.get(customer_id)
        crm = by_customer_crm.get(customer_id)
        support = by_customer_support.get(customer_id)
        blockers = _policy_blockers(customer_id, crm, support, policy)
        active = max(sub.active_seats, usage.active_seats_observed if usage else 0)
        if sub.status == "active" and sub.paid_seats and active > sub.paid_seats:
            seat_delta = active - sub.paid_seats
            amount = min(seat_delta * max(sub.monthly_recurring_revenue / max(sub.paid_seats, 1), 0), policy.maximum_account_amount)
            findings.append(_finding(tenant_id, customer_id, LeakageType.entitlement_mismatch, amount, sub.currency, _confidence([sub, usage, crm, support]), confidence_level, ["active_seats_exceed_paid_seats", f"seat_delta_{seat_delta}"], blockers, RecoveryAction.reconcile_entitlement))
        if sub.status == "active" and usage and sub.included_units and usage.units_used > sub.included_units:
            overage = usage.units_used - sub.included_units
            amount = min(overage * sub.unit_price, policy.maximum_account_amount)
            findings.append(_finding(tenant_id, customer_id, LeakageType.plan_leakage, amount, sub.currency, _confidence([sub, usage, crm, support]) - 0.05, confidence_level, ["usage_above_included_units", f"overage_units_{overage}"], blockers, RecoveryAction.plan_true_up_review))
        if customer_id not in by_customer_crm:
            unknowns.append(f"CRM record missing for token {_token(tenant_id, customer_id)}")
        if customer_id not in by_customer_usage:
            unknowns.append(f"Usage record missing for token {_token(tenant_id, customer_id)}")

    if duplicate_invoices:
        unknowns.append(f"Duplicate invoices suppressed from amount totals: {len(duplicate_invoices)}")
    findings = [item for item in findings if item.recoverable_amount >= 0]
    findings.sort(key=lambda item: (-item.recoverable_amount, item.customer_token, item.leakage_type.value))
    return findings, sorted(set(unknowns))


def _finding(tenant_id: str, customer_id: str, leakage_type: LeakageType, amount: float, currency: str, confidence: float, confidence_level: float, evidence: list[str], blockers: list[str], action: RecoveryAction) -> LeakageFinding:
    confidence = max(0.05, min(0.99, confidence))
    margin = (1 - confidence) * (1.1 if confidence_level >= 0.9 else 0.9)
    low = max(0.0, amount * (1 - margin))
    high = amount * (1 + margin)
    token = _token(tenant_id, customer_id)
    payload = {"tenant": tenant_id, "customer": token, "type": leakage_type.value, "amount": round(amount, 2), "evidence": evidence}
    finding_id = "lf_" + _digest(payload)[:16]
    return LeakageFinding(
        finding_id=finding_id,
        customer_token=token,
        leakage_type=leakage_type,
        recoverable_amount=round(amount, 2),
        currency=currency,
        confidence=round(confidence, 3),
        confidence_interval_low=round(low, 2),
        confidence_interval_high=round(high, 2),
        evidence_codes=evidence,
        blockers=blockers,
        recommended_action=RecoveryAction.suppress if blockers else action,
    )


def _confidence(parts: list[Any]) -> float:
    present = sum(1 for item in parts if item is not None)
    base = 0.58 + present * 0.09
    return min(0.95, base)


def _policy_blockers(customer_id: str, crm: CRMRecord | None, support: SupportRecord | None, policy: PolicyConfig) -> list[str]:
    blockers: list[str] = []
    if customer_id in policy.suppressed_customer_ids:
        blockers.append("caller_suppression")
    if crm is None:
        blockers.append("missing_crm_policy_context")
    else:
        if policy.require_consent and not crm.consent_to_contact:
            blockers.append("no_contact_consent")
        if crm.suppressed:
            blockers.append("crm_suppression")
        if crm.jurisdiction.upper() in {j.upper() for j in policy.suppressed_jurisdictions}:
            blockers.append("jurisdiction_suppression")
        if crm.customer_risk == RiskLevel.high:
            blockers.append("high_customer_risk")
    if support and support.open_dispute:
        blockers.append("open_dispute")
    if support and support.sentiment == RiskLevel.high:
        blockers.append("high_support_risk")
    return sorted(set(blockers))


def _totals(findings: list[LeakageFinding]) -> dict[str, float]:
    totals: dict[str, float] = {}
    counted: set[str] = set()
    for item in findings:
        if item.leakage_type == LeakageType.data_quality or item.recoverable_amount <= 0:
            continue
        key = f"{item.customer_token}:{item.finding_id}"
        if key in counted:
            continue
        counted.add(key)
        totals[item.currency] = round(totals.get(item.currency, 0.0) + item.recoverable_amount, 2)
    return dict(sorted(totals.items()))


def _prioritize(analysis: AnalyzeRevenueLeakageOutput, policy: PolicyConfig, max_accounts: int) -> list[PrioritizedAccount]:
    grouped: dict[str, list[LeakageFinding]] = {}
    for finding in analysis.findings:
        if finding.leakage_type == LeakageType.data_quality:
            continue
        grouped.setdefault(finding.customer_token, []).append(finding)
    max_amount = max((sum(f.recoverable_amount for f in fs) for fs in grouped.values()), default=1.0)
    accounts: list[PrioritizedAccount] = []
    for token, findings in grouped.items():
        amount = round(sum(f.recoverable_amount for f in findings), 2)
        blockers = sorted(set(b for f in findings for b in f.blockers))
        likelihood = round(sum(f.confidence for f in findings) / len(findings), 3)
        risk = RiskLevel.high if any(b in blockers for b in ("open_dispute", "high_customer_risk", "high_support_risk")) else RiskLevel.medium if blockers else RiskLevel.low
        value_score = min(100.0, (amount / max_amount) * 100.0)
        low_risk_bonus = 15 if risk == RiskLevel.low else 5 if risk == RiskLevel.medium else 0
        blocker_penalty = min(60, len(blockers) * 18)
        score = max(0.0, min(100.0, value_score * 0.45 + likelihood * 35 + low_risk_bonus - blocker_penalty))
        accounts.append(PrioritizedAccount(rank=1, customer_token=token, priority_score=round(score, 2), recoverable_amount=min(amount, policy.maximum_account_amount), currency=findings[0].currency, leakage_types=sorted({f.leakage_type for f in findings}, key=lambda x: x.value), likelihood=likelihood, customer_risk=risk, policy_blockers=blockers))
    accounts.sort(key=lambda item: (-item.priority_score, -item.recoverable_amount, item.customer_token))
    return [item.model_copy(update={"rank": idx + 1}) for idx, item in enumerate(accounts[:max_accounts])]


def _propose(analysis: AnalyzeRevenueLeakageOutput, prioritized: PrioritizeAccountsOutput, max_plans: int) -> list[RecoveryPlan]:
    by_token: dict[str, list[LeakageFinding]] = {}
    for finding in analysis.findings:
        by_token.setdefault(finding.customer_token, []).append(finding)
    plans: list[RecoveryPlan] = []
    for account in prioritized.prioritized_accounts[:max_plans]:
        findings = by_token.get(account.customer_token, [])
        primary = max(findings, key=lambda f: f.recoverable_amount, default=None)
        action = RecoveryAction.suppress if account.policy_blockers else (primary.recommended_action if primary else RecoveryAction.data_quality_review)
        amount = account.recoverable_amount
        rationale = _rationale(action, account, primary)
        draft = _message_draft(action, amount, account.currency)
        plan_base = {"token": account.customer_token, "action": action.value, "amount": amount, "currency": account.currency, "rationale": rationale}
        digest = _digest(plan_base)
        plans.append(RecoveryPlan(plan_id="plan_" + digest[:16], customer_token=account.customer_token, proposed_action=action, amount=amount, currency=account.currency, rationale=rationale, respectful_message_draft=draft, approval_required=True, action_digest=digest, prohibited_actions=["charge_customer", "refund", "change_subscription", "send_message", "update_crm", "make_legal_claim", "infer_protected_traits"]))
    return plans


def _rationale(action: RecoveryAction, account: PrioritizedAccount, primary: LeakageFinding | None) -> str:
    if action == RecoveryAction.suppress:
        return "Preparation only: policy blockers require suppression or specialist review before any customer-facing action."
    if action == RecoveryAction.collect_payment_update:
        return "Failed payment signal with recent usage/subscription evidence suggests respectful payment-method update outreach may recover revenue."
    if action == RecoveryAction.reconcile_entitlement:
        return "Observed active seats exceed paid seats; recommend internal entitlement reconciliation before any customer communication."
    if action == RecoveryAction.plan_true_up_review:
        return "Usage exceeds included plan units; recommend human review of contract terms before true-up messaging."
    if action == RecoveryAction.save_offer_review:
        return "Past-due/open invoice with active usage suggests involuntary churn risk; prepare retention review, not autonomous billing."
    return f"Data-quality review required for {primary.finding_id if primary else 'unknown finding'}."


def _message_draft(action: RecoveryAction, amount: float, currency: str) -> str:
    if action == RecoveryAction.suppress:
        return "No customer message should be sent. This account is blocked pending policy or support review."
    if action == RecoveryAction.collect_payment_update:
        return f"We noticed a recent billing issue on your subscription. If you would like to continue uninterrupted access, please review your payment details. Estimated impacted balance: {currency} {amount:.2f}. If this looks wrong, reply and we will review."
    if action == RecoveryAction.reconcile_entitlement:
        return "Before contacting the customer, verify entitlement and seat counts internally. If confirmed, send a neutral note offering to align licenses with current usage."
    if action == RecoveryAction.plan_true_up_review:
        return "Before contacting the customer, verify plan terms and usage. If confirmed, send a neutral note offering options to align the plan with observed usage."
    if action == RecoveryAction.save_offer_review:
        return "Prepare a helpful retention review. Do not imply fault; offer assistance resolving account access or billing questions."
    return "Internal data-quality review only. Do not contact the customer."


def _validate_plans(proposed: ProposeRecoveryOutput, policy: PolicyConfig) -> list[PlanValidation]:
    validations: list[PlanValidation] = []
    for plan in proposed.recovery_plans:
        blockers: list[str] = []
        if plan.proposed_action == RecoveryAction.suppress:
            blockers.append("suppressed_plan")
        if plan.amount > policy.maximum_account_amount:
            blockers.append("amount_exceeds_policy")
        if "charge_customer" not in plan.prohibited_actions or "send_message" not in plan.prohibited_actions:
            blockers.append("missing_prohibited_action_boundary")
        if DANGEROUS_TEXT.search(plan.respectful_message_draft):
            blockers.append("unsafe_message_text")
        digest = _digest(plan.model_dump(mode="json"))
        validations.append(PlanValidation(plan_id=plan.plan_id, customer_token=plan.customer_token, approved_for_preparation=not blockers, blocked=bool(blockers), blockers=sorted(set(blockers)), required_future_approval_phrase=f"I_APPROVE_THE_LISTED_STEPS:{digest}", immutable_plan_digest=digest))
    return validations


def _campaign_markdown(run_id: str, analysis: AnalyzeRevenueLeakageOutput, prioritized: PrioritizeAccountsOutput, proposed: ProposeRecoveryOutput, validation: ValidateRecoveryOutput) -> str:
    total = ", ".join(f"{cur} {amount:.2f}" for cur, amount in analysis.totals_by_currency.items()) or "none"
    lines = [
        f"# Revenue Recovery Campaign Pack ({run_id})",
        "",
        "## Operating boundary",
        "This pack is approval-ready preparation only. No charges, refunds, subscription changes, CRM updates, or customer messages were performed.",
        "",
        "## Summary",
        f"- Findings: {len(analysis.findings)}",
        f"- Prioritized accounts: {len(prioritized.prioritized_accounts)}",
        f"- Proposed plans: {len(proposed.recovery_plans)}",
        f"- Approval-ready after validation: {validation.compliance_summary.get('approval_ready', 0)}",
        f"- Estimated recoverable amount: {total}",
        "",
        "## Recommended next steps",
        "1. Human owner reviews blocked plans and unknowns.",
        "2. Human owner verifies source-system records and contract terms.",
        "3. If future external action tooling is added, require the exact I_APPROVE_THE_LISTED_STEPS token tied to the immutable plan digest.",
        "",
        "## Top accounts (tokenized)",
    ]
    for account in prioritized.prioritized_accounts[:10]:
        lines.append(f"- Rank {account.rank}: {account.customer_token}, score {account.priority_score}, amount {account.currency} {account.recoverable_amount:.2f}, blockers: {', '.join(account.policy_blockers) or 'none'}")
    lines.extend(["", "## Unknowns", *[f"- {u}" for u in sorted(set(analysis.unknowns + proposed.unknowns + validation.unknowns))]])
    return "\n".join(lines) + "\n"


def _measurement_plan(run_id: str, analysis: AnalyzeRevenueLeakageOutput, prioritized: PrioritizeAccountsOutput, validation: ValidateRecoveryOutput) -> dict[str, Any]:
    return {
        "run_id": run_id,
        "dry_run": True,
        "external_actions_performed": False,
        "primary_metrics": ["verified_recoverable_amount", "human_approved_plan_rate", "resolution_rate", "customer_complaint_rate"],
        "guardrail_metrics": ["suppression_violation_count", "dispute_contact_count", "unsubscribe_or_complaint_rate", "false_positive_rate"],
        "cohorts": ["payment_failure", "entitlement_mismatch", "involuntary_churn", "plan_leakage", "data_quality"],
        "baseline": {"findings": len(analysis.findings), "prioritized_accounts": len(prioritized.prioritized_accounts), "approval_ready": validation.compliance_summary.get("approval_ready", 0)},
        "holdout_recommendation": "Use a human-approved holdout before any future customer-facing campaign; this agent does not send messages.",
    }


def _offline_analysis(tenant_id: str, run_id: str | None, use_synthetic_data: bool, policy: PolicyConfig) -> AnalyzeRevenueLeakageOutput:
    snapshots = _snapshots(use_synthetic_data, None)
    rid = _run_id(run_id, tenant_id, "offline_analysis", snapshots.model_dump(mode="json"))
    findings, unknowns = _analyze_snapshots(tenant_id, snapshots, policy, 0.90)
    return AnalyzeRevenueLeakageOutput(run_id=rid, status="completed", dry_run=True, external_actions_performed=False, tenant_token=_token(tenant_id, tenant_id), findings=findings, totals_by_currency=_totals(findings), unknowns=unknowns, audit_events=[_audit(rid, "offline_analysis", {})])


def _offline_prioritized(tenant_id: str, run_id: str | None, analysis: AnalyzeRevenueLeakageOutput, policy: PolicyConfig, max_accounts: int) -> PrioritizeAccountsOutput:
    rid = _run_id(run_id or analysis.run_id, tenant_id, "offline_prioritized", analysis.model_dump(mode="json"))
    return PrioritizeAccountsOutput(run_id=rid, status="completed", dry_run=True, external_actions_performed=False, prioritized_accounts=_prioritize(analysis, policy, max_accounts), scoring_formula="min(100, value_score*0.45 + likelihood*35 + low_risk_bonus - blocker_penalty)", unknowns=analysis.unknowns, safety_findings=analysis.safety_findings, audit_events=[_audit(rid, "offline_prioritized", {})])


def _offline_proposed(tenant_id: str, run_id: str | None, use_synthetic_data: bool, policy: PolicyConfig) -> ProposeRecoveryOutput:
    analysis = _offline_analysis(tenant_id, run_id, use_synthetic_data, policy)
    prioritized = _offline_prioritized(tenant_id, run_id, analysis, policy, 25)
    return _offline_proposed_from(tenant_id, run_id, analysis, prioritized, policy)


def _offline_proposed_from(tenant_id: str, run_id: str | None, analysis: AnalyzeRevenueLeakageOutput, prioritized: PrioritizeAccountsOutput, policy: PolicyConfig) -> ProposeRecoveryOutput:
    rid = _run_id(run_id or prioritized.run_id, tenant_id, "offline_proposed", prioritized.model_dump(mode="json"))
    return ProposeRecoveryOutput(run_id=rid, status="completed", dry_run=True, external_actions_performed=False, recovery_plans=_propose(analysis, prioritized, 25), unknowns=analysis.unknowns, safety_findings=analysis.safety_findings, audit_events=[_audit(rid, "offline_proposed", {})])


def _offline_validation(tenant_id: str, run_id: str | None, proposed: ProposeRecoveryOutput, policy: PolicyConfig) -> ValidateRecoveryOutput:
    rid = _run_id(run_id or proposed.run_id, tenant_id, "offline_validation", proposed.model_dump(mode="json"))
    validations = _validate_plans(proposed, policy)
    return ValidateRecoveryOutput(run_id=rid, status="completed", dry_run=True, external_actions_performed=False, validations=validations, compliance_summary={"plans_reviewed": len(validations), "blocked": sum(1 for item in validations if item.blocked), "approval_ready": sum(1 for item in validations if item.approved_for_preparation)}, unknowns=proposed.unknowns, safety_findings=proposed.safety_findings, audit_events=[_audit(rid, "offline_validation", {})])


def _token(tenant_id: str, value: str) -> str:
    return "tok_" + hashlib.sha256(f"{tenant_id}:{value}".encode("utf-8")).hexdigest()[:20]


def _digest(value: Any) -> str:
    return hashlib.sha256(json.dumps(value, sort_keys=True, separators=(",", ":"), default=str).encode("utf-8")).hexdigest()


def _run_id(requested: str | None, tenant_id: str, step: str, payload: Any) -> str:
    if requested:
        if not SAFE_ID.fullmatch(requested):
            raise ValueError("run_id contains unsupported characters")
        return requested
    return f"rr_{_digest({'tenant': tenant_id, 'step': step, 'payload': payload})[:18]}"


def _audit(run_id: str, event_type: str, detail: dict[str, Any]) -> AuditEvent:
    return AuditEvent(timestamp=datetime.now(UTC).replace(microsecond=0).isoformat(), event_type=event_type, run_id=run_id, detail=_redact(detail))


def _redact(value: Any) -> Any:
    if isinstance(value, dict):
        out: dict[str, Any] = {}
        for key, item in value.items():
            lowered = str(key).lower()
            if any(marker in lowered for marker in PII_KEYS) or "secret" in lowered or "token" in lowered:
                out[str(key)] = "[REDACTED]"
            else:
                out[str(key)] = _redact(item)
        return out
    if isinstance(value, list):
        return [_redact(item) for item in value]
    return value


async def _emit(ctx: RunContext[NoAuth], event: AuditEvent) -> None:
    await ctx.emit_event(AgentEvent(kind="audit", payload=event.model_dump(mode="json")))
    await ctx.emit_progress(f"{event.event_type}: {event.run_id}")


async def _persist_json(ctx: RunContext[NoAuth], run_id: str, filename: str, data: dict[str, Any]) -> OutputFile:
    return await _persist_text(ctx, run_id, filename, json.dumps(data, indent=2, sort_keys=True) + "\n", "application/json")


async def _persist_text(ctx: RunContext[NoAuth], run_id: str, filename: str, text: str, mime_type: str) -> OutputFile:
    safe_name = filename.replace("/", "_").replace("\\", "_")
    rel_path = f"outputs/revenue-recovery/{run_id}/{safe_name}"
    data = text.encode("utf-8")
    sha = hashlib.sha256(data).hexdigest()
    artifact_uri: str | None = None
    try:
        grant = await ctx.workspace.request_access(files=[rel_path], mode=WorkspaceMode.READ_WRITE_OVERLAY, reason="persist redacted revenue recovery output", purpose="write revenue recovery artifact")
        view = await ctx.workspace.request_access(files=[rel_path], mode=WorkspaceMode.READ_WRITE_OVERLAY, reason="persist redacted revenue recovery output", purpose="write revenue recovery artifact")
        # The workspace API returns grants; open a view for concrete writes.
        opened = await ctx.workspace.open_view(purpose="persist revenue recovery output", hints=[rel_path], max_files=1, mode=WorkspaceMode.READ_WRITE_OVERLAY, reason="write generated output under outputs/revenue-recovery")
        await opened.write(rel_path, data)
        _ = grant, view
    except Exception as exc:  # workspace may be unavailable in local smoke calls; artifact still persists.
        await ctx.emit_event(AgentEvent(kind="workspace_write_warning", payload={"path": rel_path, "message": str(exc)[:300]}))
    try:
        ref = await ctx.write_artifact(safe_name, data, mime_type)
        await ctx.emit_artifact(ref)
        artifact_uri = ref.uri
    except Exception as exc:
        await ctx.emit_event(AgentEvent(kind="artifact_write_warning", payload={"name": safe_name, "message": str(exc)[:300]}))
    return OutputFile(path=rel_path, artifact_uri=artifact_uri, sha256=sha, bytes=len(data))