"""Outcome recording with idempotency, no dummy fallbacks, and timezone-aware timestamps. Also updates per‑skill reliability models when skill provenance is present (v4.3.1). """ import datetime import logging from typing import Optional, Dict, Any from sqlalchemy.orm import Session from sqlalchemy.exc import IntegrityError from agentic_reliability_framework.core.governance.risk_engine import RiskEngine from agentic_reliability_framework.core.governance.intents import ( InfrastructureIntent, ProvisionResourceIntent, GrantAccessIntent, DeployConfigurationIntent, ) from app.database.models_intents import IntentDB, OutcomeDB, BetaStateDB logger = logging.getLogger(__name__) # ── v4.3.1: optional skill registry integration ────────────── try: from agentic_reliability_framework.core.governance.skill_registry import SkillRegistry SKILL_REGISTRY_AVAILABLE = True except ImportError: SkillRegistry = None SKILL_REGISTRY_AVAILABLE = False # --------------------------------------------------------------------------- # Helper: persist the conjugate posterior state # --------------------------------------------------------------------------- def _persist_beta_state(db: Session, tenant_id: str, risk_engine: RiskEngine) -> None: """ Write the current Beta posterior parameters to the beta_state table. This is called after every outcome update so that online learning survives restarts. """ try: state = risk_engine.beta_store.get_state() for cat, (alpha, beta) in state.items(): # Upsert on (tenant_id, category): merge() matches on primary key # only, and these rows are always constructed without an `id`, # so merge() would always attempt an INSERT and collide with the # unique constraint on the second write for the same pair. row = db.query(BetaStateDB).filter( BetaStateDB.tenant_id == tenant_id, BetaStateDB.category == cat.value, ).first() if row is not None: row.alpha = alpha row.beta = beta else: db.add(BetaStateDB(tenant_id=tenant_id, category=cat.value, alpha=alpha, beta=beta)) db.commit() logger.debug("Persisted Beta posterior parameters to database.") except Exception as e: db.rollback() logger.error("Failed to persist beta state: %s", e) class OutcomeConflictError(Exception): """Raised when an outcome already exists for the same intent with a different result.""" pass def reconstruct_oss_intent_from_json( oss_json: Dict[str, Any]) -> InfrastructureIntent: """Reconstruct OSS intent from stored JSON. Raises ValueError on failure.""" intent_type = oss_json.get("intent_type") if intent_type == "provision_resource": return ProvisionResourceIntent(**oss_json) elif intent_type == "grant_access": return GrantAccessIntent(**oss_json) elif intent_type == "deploy_config": return DeployConfigurationIntent(**oss_json) else: raise ValueError( f"Cannot reconstruct intent from JSON: missing or unknown intent_type {intent_type}") def record_outcome( db: Session, tenant_id: str, deterministic_id: str, success: bool, recorded_by: Optional[str], notes: Optional[str], risk_engine: RiskEngine, idempotency_key: Optional[str] = None, skill_id: Optional[str] = None, # v4.3.1 skill_version: Optional[int] = None, # v4.3.1 skill_registry: Optional["SkillRegistry"] = None, # v4.3.1 ) -> OutcomeDB: """ Record an outcome for a previously evaluated intent. Idempotent: calling twice with the same (deterministic_id, success) returns the same record. If the outcome already exists with a different success value, raises OutcomeConflictError. No dummy intents are created. If the OSS intent cannot be reconstructed, the risk engine is NOT updated – we log an error and still record the outcome. The intent lookup is scoped to `tenant_id` so a caller can only record outcomes for intents owned by their own tenant, even if they know or guess another tenant's deterministic_id. Parameters ---------- db : Session SQLAlchemy session. tenant_id : str Tenant of the authenticated caller. Must match the intent's owning tenant. deterministic_id : str Unique identifier of the original intent. success : bool Whether the action succeeded (True) or failed (False). recorded_by : str or None Optional user or system identifier. notes : str or None Optional human-readable notes. risk_engine : RiskEngine ARF risk engine instance (may be updated). idempotency_key : str or None Optional caller-provided idempotency token. skill_id : str or None (v4.3.1) Identifier of the procedural skill that guided the action. skill_version : int or None (v4.3.1) Version number of that skill. skill_registry : SkillRegistry or None (v4.3.1) Optional skill registry instance to update per‑skill reliability. Returns ------- OutcomeDB The recorded outcome object. Raises ------ ValueError If intent not found or reconstruction fails fatally. OutcomeConflictError If a conflicting outcome already exists. """ # 1. Fetch the original intent record, scoped to the caller's tenant intent = db.query(IntentDB).filter( IntentDB.deterministic_id == deterministic_id, IntentDB.tenant_id == tenant_id, ).one_or_none() if not intent: raise ValueError(f"Intent not found: {deterministic_id}") # 2. Idempotency / conflict check with database-level uniqueness existing_outcome = db.query(OutcomeDB).filter( OutcomeDB.intent_id == intent.id).one_or_none() if existing_outcome: if existing_outcome.success == success: return existing_outcome db.rollback() raise OutcomeConflictError( f"Outcome already recorded for intent {deterministic_id} with different result " f"(existing={existing_outcome.success}, new={success})" ) # 3. Create outcome record outcome = OutcomeDB( intent_id=intent.id, success=bool(success), recorded_by=recorded_by, notes=notes, recorded_at=datetime.datetime.now(datetime.timezone.utc), idempotency_key=idempotency_key, ) db.add(outcome) # 4. Attempt to commit; handle duplicate key errors for idempotency try: db.commit() db.refresh(outcome) except IntegrityError as e: db.rollback() if "idempotency_key" in str(e) and idempotency_key: existing = db.query(OutcomeDB).filter( OutcomeDB.idempotency_key == idempotency_key).first() if existing: logger.info( "Idempotent request for key %s, returning existing outcome", idempotency_key) return existing raise # 5. Update RiskEngine ONLY if we can reconstruct a valid OSS intent oss_intent = None if intent.oss_payload: try: oss_intent = reconstruct_oss_intent_from_json(intent.oss_payload) except Exception as e: logger.error( "Failed to reconstruct OSS intent for %s: %s. RiskEngine will NOT be updated.", deterministic_id, e, exc_info=True) else: logger.warning( "No oss_payload stored for intent %s – cannot update RiskEngine.", deterministic_id ) if oss_intent is not None: try: risk_engine.update_outcome(oss_intent, success) # ---------------------------------------------------------------- # PERSISTENCE: after updating the conjugate posterior, write it # ---------------------------------------------------------------- _persist_beta_state(db, tenant_id, risk_engine) except Exception as e: logger.exception( "Failed to update RiskEngine after recording outcome for intent %s: %s", deterministic_id, e) else: logger.info( "Skipped RiskEngine update for intent %s (no valid OSS intent)", deterministic_id ) # 6. v4.3.1: Update per‑skill reliability model if provenance is provided if SKILL_REGISTRY_AVAILABLE and skill_registry is not None and skill_id is not None and skill_version is not None: try: skill_registry.observe_outcome(skill_id, skill_version, success) logger.debug( "Skill reliability updated for '%s' v%d (success=%s)", skill_id, skill_version, success, ) except Exception as e: logger.warning( "Failed to update skill reliability for '%s' v%d: %s", skill_id, skill_version, e, exc_info=True, ) return outcome