"""Persists the Agent's RTC write audit records (append-only), the candidate preview and the rollback bookkeeping; mirrors a summary event for the panel.""" from __future__ import annotations from sqlalchemy import select, update from .events import record_event from .models import RtcCandidate, RtcWriteAudit MAX_AUDITS_PER_JOB = 2000 MAX_CANDIDATES = 25000 WRITE_MODES = ("DISABLED", "DRY_RUN", "APPROVAL_REQUIRED", "ENABLED") MAX_APPROVED_PER_JOB = 500 MAX_ROLLBACKS_PER_JOB = 500 def _clip(value, size): return None if value is None else str(value)[:size] def job_authorizations(session, installation) -> dict: """What the authenticated job carries to the Agent. Decided centrally per installation; the Agent has no local switch.""" mode = installation.rtc_write_mode if installation.rtc_write_mode in WRITE_MODES else "DISABLED" out = {"rtc_write_mode": mode, "rtc_approved_codes": [], "rtc_rollbacks": [], "features": {"corrector_enabled": bool(installation.corrector_enabled), "enrichment_mode": installation.enrichment_mode, "features_configured": bool(installation.features_configured)}} if mode == "APPROVAL_REQUIRED": out["rtc_approved_codes"] = list(session.scalars(select(RtcCandidate.product_code).where(RtcCandidate.installation_id == installation.id, RtcCandidate.status == "APPROVED").limit(MAX_APPROVED_PER_JOB))) for audit in session.scalars(select(RtcWriteAudit).where(RtcWriteAudit.installation_id == installation.id, RtcWriteAudit.action == "APPLY", RtcWriteAudit.outcome == "APPLIED", RtcWriteAudit.rollback_state == "REQUESTED").limit(MAX_ROLLBACKS_PER_JOB)): out["rtc_rollbacks"].append({"audit_id": audit.id, "product_code": audit.product_code, "written_value": audit.new_value, "rule_id": audit.rule_id}) return out def _store_preview(session, tenant_id, installation, summary, evaluated_all): preview = summary.get("preview") if not isinstance(preview, list) or installation.rtc_write_mode == "DISABLED": return existing = {c.product_code: c for c in session.scalars(select(RtcCandidate).where(RtcCandidate.installation_id == installation.id))} seen = set() for item in preview[:MAX_CANDIDATES]: code = _clip(item.get("product_code"), 64) if not code or not item.get("proposed"): continue seen.add(code) row = existing.get(code) if row is None: session.add(RtcCandidate(tenant_id=tenant_id, installation_id=installation.id, product_code=code, description=_clip(item.get("description"), 255), ncm=_clip(item.get("ncm"), 16), current_value=_clip(item.get("current") or None, 16), proposed=_clip(item["proposed"], 16), cst_ibscbs=_clip(item.get("cst"), 8), rule_id=_clip(item.get("rule_id"), 32), basis=_clip(item.get("basis"), 255), status="PENDING")) else: if row.status in ("APPLIED", "STALE"): row.status = "PENDING" # empty again (e.g. after a rollback) or newly eligible if row.proposed != item["proposed"]: row.proposed = _clip(item["proposed"], 16) if row.status == "APPROVED": row.status = "PENDING" # the proposal changed: the earlier approval does not carry over row.description, row.ncm, row.cst_ibscbs = _clip(item.get("description"), 255), _clip(item.get("ncm"), 16), _clip(item.get("cst"), 8) row.rule_id, row.basis = _clip(item.get("rule_id"), 32), _clip(item.get("basis"), 255) if evaluated_all and not summary.get("preview_truncated"): for code, row in existing.items(): if code not in seen and row.status in ("PENDING", "APPROVED"): row.status = "STALE" def record_rtc_write_audits(session, *, tenant_id: int, installation, agent, job_id: int | None, rtc_auto_apply: dict | None, evaluated_all: bool = False) -> int: """Stores every applied/aborted/rolled-back write of the job. Never trusts the Agent for identity: tenant/installation/agent come from the authenticated context.""" if not isinstance(rtc_auto_apply, dict): return 0 _store_preview(session, tenant_id, installation, rtc_auto_apply, evaluated_all) records = list(rtc_auto_apply.get("applied") or []) + list(rtc_auto_apply.get("aborted") or []) + list(rtc_auto_apply.get("rollbacks") or []) stored = applied = aborted = rolled = 0 for item in records[:MAX_AUDITS_PER_JOB]: if not isinstance(item, dict) or item.get("field") != "CCLASSTRIB": continue action = "ROLLBACK" if item.get("action") == "ROLLBACK" else "APPLY" outcome = str(item.get("outcome", ""))[:16] rollback_of = item.get("rollback_of") if action == "ROLLBACK" else None code = str(item.get("product_code", ""))[:64] session.add(RtcWriteAudit( tenant_id=tenant_id, installation_id=installation.id, agent_id=agent.id, job_id=job_id, client=getattr(installation, "display_name", None), product_code=code, field="CCLASSTRIB", outcome=outcome, reason=(item.get("reason") or None) and str(item["reason"])[:96], old_value=_clip(item.get("old_value"), 64), new_value=_clip(item.get("new_value"), 64), rule_id=(item.get("rule_id") or None) and str(item["rule_id"])[:32], evidence=item.get("evidence") or {}, official_source_version=item.get("official_source_version") or {}, local_catalog_hash=item.get("local_catalog_hash"), written_at=item.get("timestamp"), rule_version=_clip(item.get("rule_version"), 32), company_crt=_clip(item.get("company_crt"), 8), authorization_origin=_clip(item.get("authorization_origin"), 48), description_snapshot=_clip(item.get("description_snapshot"), 255), installation_uuid=_clip(getattr(installation, "installation_id", None), 64), rtc_status=_clip(item.get("rtc_status"), 24), local_catalog_state=_clip(item.get("local_catalog_state"), 32), action=action, rollback_of=rollback_of if isinstance(rollback_of, int) else None)) stored += 1 candidate_of = (RtcCandidate.installation_id == installation.id) & (RtcCandidate.product_code == code) if action == "ROLLBACK": if isinstance(rollback_of, int): session.execute(update(RtcWriteAudit).where(RtcWriteAudit.id == rollback_of, RtcWriteAudit.installation_id == installation.id).values(rollback_state="DONE" if outcome == "ROLLED_BACK" else "BLOCKED")) if outcome == "ROLLED_BACK": rolled += 1 session.execute(update(RtcCandidate).where(candidate_of).values(status="PENDING")) elif outcome == "APPLIED": applied += 1 session.execute(update(RtcCandidate).where(candidate_of).values(status="APPLIED")) else: aborted += 1 if stored: record_event(session, tenant_id=tenant_id, installation_id=installation.id, agent_id=agent.id, severity="WARNING" if (applied or rolled) else "INFO", event_type="RTC_AUTO_APPLY", job_id=job_id, message=f"cClassTrib: {applied} preenchido(s), {rolled} revertido(s), {aborted} abortado(s)", context={"applied": applied, "rolled_back": rolled, "aborted": aborted, "reasons": rtc_auto_apply.get("aborted_reasons") or {}, "mode": rtc_auto_apply.get("mode")}) session.commit() return stored