diff --git a/agent.py b/agent.py new file mode 100644 index 0000000..b219b81 --- /dev/null +++ b/agent.py @@ -0,0 +1,1074 @@ +"""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| 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(default="tenant-smoke", 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.1" + + 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_json(ctx, run_id, "leakage.json", analysis.model_dump(mode="json")), + await _persist_json(ctx, run_id, "prioritized-accounts.json", prioritized.model_dump(mode="json")), + await _persist_json(ctx, run_id, "recovery-plans.json", proposed.model_dump(mode="json")), + await _persist_json(ctx, run_id, "compliance-report.json", validation.model_dump(mode="json")), + 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)}") + nonnegative_findings: list[LeakageFinding] = [] + for finding in findings: + if finding.recoverable_amount >= 0: + nonnegative_findings.append(finding) + nonnegative_findings.sort(key=_finding_sort_key) + return nonnegative_findings, sorted(set(unknowns)) + + +def _finding_sort_key(finding: LeakageFinding) -> tuple[float, str, str]: + return (-finding.recoverable_amount, finding.customer_token, finding.leakage_type.value) + + +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)) + leakage_types = sorted({finding.leakage_type for finding in findings}, key=_leakage_type_sort_key) + 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=leakage_types, likelihood=likelihood, customer_risk=risk, policy_blockers=blockers)) + accounts.sort(key=_prioritized_account_sort_key) + ranked_accounts: list[PrioritizedAccount] = [] + for index, account in enumerate(accounts[:max_accounts]): + ranked_accounts.append(account.model_copy(update={"rank": index + 1})) + return ranked_accounts + + +def _leakage_type_sort_key(leakage_type: LeakageType) -> str: + return leakage_type.value + + +def _prioritized_account_sort_key(account: PrioritizedAccount) -> tuple[float, float, str]: + return (-account.priority_score, -account.recoverable_amount, account.customer_token) + + +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=_recoverable_amount_key, 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 _recoverable_amount_key(finding: LeakageFinding) -> float: + return finding.recoverable_amount + + +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: + writer = getattr(ctx.workspace, "write_bytes", None) + if writer is None: + raise RuntimeError("workspace client does not expose direct output writes") + writer(rel_path, data) + 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))