diff --git a/agent.py b/agent.py deleted file mode 100644 index b219b81..0000000 --- a/agent.py +++ /dev/null @@ -1,1074 +0,0 @@ -"""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))