1071 lines
56 KiB
Python
1071 lines
56 KiB
Python
"""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_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=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:
|
|
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))
|