GTI_GERENCIADOR_FISCAL/scripts/rtc_copy_write_e2e.py

"""COPY-ONLY write E2E for the centrally controlled RTC auto-apply (C000025.CCLASSTRIB).

    python scripts/rtc_copy_write_e2e.py

Runs ONLY against scratch_rtc_e2e/TESTESOFICIAIS_RTC_COPY.FDB (restored from its own .fbk at the start, so every run is reproducible). It refuses any other path and every
real database. Everything goes through the Agent's own code path `run_agent_service._rtc_auto_apply`, driven ONLY by the authenticated job `authorizations` (no env switches):
  0 restore + company -> regime normal (documented experiment on the copy);   1 whole-database baseline hashes;
  2 DISABLED (0 writes, no preview) / DRY_RUN (preview, 0 writes) / stale decision (RTC off between decision and write => 0 writes);
  3 APPROVAL_REQUIRED with 5 approved products (only those 5 written) / then ENABLED until nothing is left;
  4 reread, FK + CST integrity, coverage;   5 second run = 0 writes;
  6 rollback of 3 audit records (+1 conflict: value changed by an operator => refused);   7 reread, reapply, idempotency;
  8 invariants: NON_TARGET_COLUMNS_CHANGED, CATALOG_CHANGED, PARAMETERS_CHANGED, DOCUMENT_TABLES_CHANGED, OTHER_TABLES_CHANGED all 0.
Writes go to the copy and nowhere else.
"""
import hashlib
import importlib.util
import json
import os
import subprocess
import sys
from pathlib import Path
from types import SimpleNamespace

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT)); sys.path.insert(0, str(ROOT / "scripts")); sys.path.insert(0, str(ROOT / "agent" / "src"))
from dotenv import load_dotenv  # noqa: E402

import fdb  # noqa: E402
import local_completion as lc  # noqa: E402
from central.app.operational_state import classify_product_state, summarize_operational_states  # noqa: E402
from gti_fiscal_agent import firebird_reader, unified_fiscal  # noqa: E402

COPY = str((ROOT / "scratch_rtc_e2e" / "TESTESOFICIAIS_RTC_COPY.FDB").resolve())
FBK = str((ROOT / "scratch_rtc_e2e" / "TESTESOFICIAIS_RTC_COPY.fbk").resolve())
NCM_OK = set((ROOT / "docs" / "runs" / "ncm_valid_codes.txt").read_text().split())
DOC_TABLE_HINTS = ("NFCE", "NFE", "NF_", "CUPOM", "VENDA", "CTE", "MDFE")


def guard(path):
    canon = os.path.normcase(os.path.abspath(path))
    if canon != os.path.normcase(COPY):
        raise SystemExit(f"REFUSED: {path} is not the disposable copy")
    real = {os.path.normcase(os.path.abspath(p)) for p in lc.allowlist().values()}
    if canon in real or "scratch_rtc_e2e" not in canon:
        raise SystemExit("REFUSED: real database")


def connect(rw):
    kw = dict(database=COPY, user=os.environ["FB_USER"], password=os.environ["FB_PASSWORD"], fb_library_name=os.environ.get("FIREBIRD_CLIENT_LIBRARY"))
    if not rw:
        kw["isolation_level"] = fdb.ISOLATION_LEVEL_READ_COMMITED_RO
    return fdb.connect(**kw)


def load_service():
    spec = importlib.util.spec_from_file_location("run_agent_service", ROOT / "agent" / "scripts" / "run_agent_service.py")
    mod = importlib.util.module_from_spec(spec)
    sys.modules[spec.name] = mod
    spec.loader.exec_module(mod)
    return mod


def digits(v):
    return "".join(ch for ch in str(v or "") if ch.isdigit())


def restore_copy():
    guard(COPY)
    gbak = Path(os.environ["FIREBIRD_CLIENT_LIBRARY"]).parent / "gbak.exe"
    done = subprocess.run([str(gbak), "-c", "-rep", "-n", "-user", os.environ["FB_USER"], "-pas", os.environ["FB_PASSWORD"], FBK, COPY], capture_output=True, text=True, timeout=600)
    if done.returncode != 0:
        raise SystemExit(f"restore failed: {done.stderr[-300:]}")


def tables(cur):
    cur.execute("SELECT TRIM(RDB$RELATION_NAME) FROM RDB$RELATIONS WHERE COALESCE(RDB$SYSTEM_FLAG, 0) = 0 AND RDB$VIEW_BLR IS NULL AND RDB$EXTERNAL_FILE IS NULL ORDER BY 1")
    return [r[0] for r in cur.fetchall()]


def table_hash(cur, table, skip_cols=()):
    cur.execute(f"SELECT * FROM {table}")
    cols = [d[0] for d in cur.description]
    keep = [i for i, c in enumerate(cols) if c not in skip_cols]
    digest = hashlib.sha256()
    rows = 0
    for row in cur.fetchall():
        digest.update(repr([row[i] for i in keep]).encode("utf-8", "replace"))
        rows += 1
    return f"{rows}:{digest.hexdigest()[:20]}"


def baseline():
    conn = connect(False); cur = conn.cursor()
    out = {t: table_hash(cur, t, skip_cols=("CCLASSTRIB",) if t == "C000025" else ()) for t in tables(cur)}
    cur.execute("SELECT CODIGO, CCLASSTRIB FROM C000025")
    values = {str(r[0]).strip(): (r[1] or "").strip() for r in cur.fetchall()}
    conn.close()
    return out, values


def diff_tables(before, after):
    return sorted(t for t in before if before[t] != after.get(t))


def coverage(env, company, ncm_valid):
    conn = firebird_reader._connect(env)
    try:
        out = unified_fiscal.assess_from_connection(conn, company, ncm_valid=ncm_valid)
    finally:
        conn.close()
    products = firebird_reader.read_products(env)
    states = [classify_product_state(p, {"status": "EXACT_NCM"}, fiscal=out["findings"].get(str(p["CODIGO"]))) for p in products]
    s = summarize_operational_states(states)
    return out, products, {"total": s["TOTAL_PRODUCTS"], "complete": s["COMPLETE_PRODUCTS"], "blocked": s["SAFE_BLOCKED"], "unclassified": s["UNCLASSIFIED"], "coverage_percent": round(100 * s["COMPLETE_PRODUCTS"] / s["TOTAL_PRODUCTS"], 2)}


def main():
    load_dotenv(ROOT / ".env")            # credentials only when the script runs, never as an import side effect
    guard(COPY)
    service = load_service()
    restore_copy()
    env = {"FIREBIRD_DATABASE_PATH": COPY, "FB_USER": os.environ["FB_USER"], "FB_PASSWORD": os.environ["FB_PASSWORD"], "FIREBIRD_CLIENT_LIBRARY": os.environ.get("FIREBIRD_CLIENT_LIBRARY")}
    report = {"database": "scratch_rtc_e2e/TESTESOFICIAIS_RTC_COPY.FDB (restored from its .fbk at start)", "real_database_writes": 0, "real_database_write_attempts": 0, "mechanism": "job authorizations only (no env switches)"}
    rw = connect(True); cur = rw.cursor()
    cur.execute("SELECT DISTINCT TRIM(CRT) FROM C000004"); report["copy_original_crt"] = [r[0] for r in cur.fetchall()]
    cur.execute("UPDATE C000004 SET CRT = '3'"); rw.commit()
    cur.execute("SELECT DISTINCT TRIM(CRT) FROM C000004"); company = {"crt": cur.fetchone()[0]}
    cur.execute("SELECT USAR_REFORMA_TRIBUTARIA FROM PARAMETROS"); report["copy_rtc_param"] = cur.fetchone()[0]
    cur.execute("SELECT CODIGO, CLASSIFICACAO_FISCAL FROM C000025"); ncm_valid = {str(c).strip(): digits(n) in NCM_OK for c, n in cur.fetchall()}
    rw.commit(); rw.close()
    report["copy_crt_for_experiment"] = company["crt"]
    base, base_values = baseline()
    report["tables_hashed"] = len(base)
    config = SimpleNamespace(rtc_auto_apply_max_per_job=500, database_path=COPY)          # NOTE: no firebird_writes, no rtc switch: the Agent has none

    def run_job(mode, approved=(), rollbacks=(), job_id=1):
        out, products, _ = coverage(env, company, ncm_valid)
        job = {"job_id": job_id, "authorizations": {"rtc_write_mode": mode, "rtc_approved_codes": list(approved), "rtc_rollbacks": list(rollbacks)}}
        return service._rtc_auto_apply(env, config, job, products, out, company), out

    _, _, cov_before = coverage(env, company, ncm_valid)
    report["coverage_before"] = cov_before
    # ---- 2. DISABLED / DRY_RUN / stale decision: zero writes
    disabled, _ = run_job("DISABLED")
    dry, out0 = run_job("DRY_RUN")
    report["disabled_run"] = {"write_attempts": disabled["write_attempts"], "preview": len(disabled["preview"]), "ready_total": disabled["ready_total"]}
    report["dry_run"] = {"write_attempts": dry["write_attempts"], "preview": len(dry["preview"]), "ready_total": dry["ready_total"], "rules": sorted({c["rule_id"] for c in dry["preview"]}), "sample": dry["preview"][:3]}
    report["resolution_before"] = {k: v for k, v in out0["context"]["rtc_resolution"].items() if k != "by_reason"}
    assert base == baseline()[0], "DISABLED/DRY_RUN wrote something"
    rw = connect(True); c = rw.cursor(); c.execute("UPDATE PARAMETROS SET USAR_REFORMA_TRIBUTARIA = 0"); rw.commit()
    stale, _ = run_job("ENABLED")
    c.execute("UPDATE PARAMETROS SET USAR_REFORMA_TRIBUTARIA = 1"); rw.commit(); rw.close()
    report["stale_decision_run"] = {"write_attempts": stale["write_attempts"], "write_successes": stale["write_successes"]}
    # ---- 3. APPROVAL_REQUIRED (5 approved) then ENABLED
    preview_codes = [c["product_code"] for c in dry["preview"]]
    approved = preview_codes[:5]
    unapproved = run_job("APPROVAL_REQUIRED", approved=[])[0]
    part, _ = run_job("APPROVAL_REQUIRED", approved=approved, job_id=2)
    report["approval_run"] = {"without_approval_attempts": unapproved["write_attempts"], "approved": len(approved), "write_successes": part["write_successes"], "only_approved": sorted(a["product_code"] for a in part["applied"]) == sorted(approved)}
    applied = list(part["applied"])
    jobs = 0
    while True:
        jobs += 1
        step, _ = run_job("ENABLED", job_id=100 + jobs)
        applied += step["applied"]
        report.setdefault("enabled_jobs", []).append({"write_attempts": step["write_attempts"], "write_successes": step["write_successes"], "aborted": step["aborted_reasons"]})
        if not step["write_attempts"] or jobs >= 200:
            break
    report["total_applied"] = len(applied)
    report["audit_sample"] = applied[:2]
    report["audit_fields_complete"] = all(a.get(k) is not None for a in applied for k in ("rule_id", "rule_version", "company_crt", "authorization_origin", "description_snapshot", "timestamp", "job_id", "local_catalog_hash", "official_source_version"))
    # ---- 4. reread + integrity + coverage
    mid, mid_values = baseline()
    report["rows_changed"] = sum(1 for k, v in mid_values.items() if base_values.get(k) != v)
    ro = connect(False); cur = ro.cursor()
    cur.execute("SELECT COUNT(*) FROM C000025 p WHERE p.CCLASSTRIB IS NOT NULL AND NOT EXISTS (SELECT 1 FROM RTC_CCLASSTRIB c WHERE c.CCLASSTRIB = p.CCLASSTRIB)"); report["fk_orphans"] = cur.fetchone()[0]
    cur.execute("SELECT COUNT(*) FROM C000025 p JOIN RTC_CCLASSTRIB c ON c.CCLASSTRIB = p.CCLASSTRIB WHERE NOT EXISTS (SELECT 1 FROM RTC_CST s WHERE s.CSTIBSCBS = c.CSTIBSCBS)"); report["cst_orphans"] = cur.fetchone()[0]
    ro.close()
    _, _, cov_after = coverage(env, company, ncm_valid)
    report["coverage_after"] = cov_after
    # ---- 5. second run
    second, _ = run_job("ENABLED", job_id=300)
    report["second_run"] = {"write_attempts": second["write_attempts"], "write_successes": second["write_successes"], "ready_total": second["ready_total"]}
    # ---- 6. rollback of 3 records + 1 conflict
    print(json.dumps({k: report[k] for k in ("disabled_run","dry_run","stale_decision_run","approval_run","enabled_jobs","coverage_before","resolution_before")}, default=str)[:2500]) if len(applied) < 4 else None
    victims = applied[:4]
    rw = connect(True); c = rw.cursor(); c.execute("UPDATE C000025 SET CCLASSTRIB = '200035' WHERE CODIGO = ?", (victims[3]["product_code"],)); rw.commit(); rw.close()      # an operator changed it afterwards
    requests = [{"audit_id": 900 + i, "product_code": v["product_code"], "written_value": v["new_value"], "rule_id": v["rule_id"]} for i, v in enumerate(victims)]
    roll, _ = run_job("DISABLED", rollbacks=requests, job_id=400)                       # rollback needs the central request, not a write mode
    outcomes = {r["product_code"]: (r["outcome"], r.get("reason")) for r in roll["rollbacks"]}
    report["rollback"] = {"rolled_back": sum(1 for o, _ in outcomes.values() if o == "ROLLED_BACK"), "conflict": outcomes[victims[3]["product_code"]], "records": [outcomes[v["product_code"]][0] for v in victims]}
    ro = connect(False); cur = ro.cursor()
    cur.execute("SELECT CODIGO, CCLASSTRIB FROM C000025 WHERE CODIGO IN (%s)" % ",".join("?" * 4), tuple(v["product_code"] for v in victims))
    after_roll = {str(r[0]).strip(): (r[1] or "").strip() for r in cur.fetchall()}
    ro.close()
    report["rollback"]["reread"] = {k: v for k, v in after_roll.items()}
    # ---- 7. reapply + idempotency
    again, _ = run_job("ENABLED", job_id=500)
    report["reapply"] = {"write_successes": again["write_successes"], "codes": sorted(a["product_code"] for a in again["applied"]), "expected": sorted(v["product_code"] for v in victims[:3])}
    third, _ = run_job("ENABLED", job_id=501)
    report["idempotency_after_reapply"] = {"write_attempts": third["write_attempts"], "write_successes": third["write_successes"]}
    # ---- 8. invariants
    final, final_values = baseline()
    changed_tables = diff_tables(base, final)
    report["NON_TARGET_COLUMNS_CHANGED"] = 1 if "C000025" in changed_tables else 0
    report["CATALOG_CHANGED"] = len([t for t in changed_tables if t.startswith("RTC_")])
    report["PARAMETERS_CHANGED"] = len([t for t in changed_tables if t == "PARAMETROS"])
    report["DOCUMENT_TABLES_CHANGED"] = len([t for t in changed_tables if any(h in t for h in DOC_TABLE_HINTS)])
    report["OTHER_TABLES_CHANGED"] = len(changed_tables)
    report["changed_tables"] = changed_tables
    report["IDEMPOTENT"] = second["write_attempts"] == 0 and third["write_attempts"] == 0
    report["PASS"] = bool(report["IDEMPOTENT"] and report["fk_orphans"] == 0 and report["cst_orphans"] == 0 and not changed_tables and report["total_applied"] > 5
                          and report["disabled_run"]["write_attempts"] == 0 and report["dry_run"]["write_attempts"] == 0 and report["dry_run"]["rules"] == ["RTC-R3"]
                          and report["stale_decision_run"]["write_successes"] == 0 and report["approval_run"]["without_approval_attempts"] == 0 and report["approval_run"]["only_approved"]
                          and report["rows_changed"] == report["total_applied"] and report["rollback"]["rolled_back"] == 3 and report["rollback"]["conflict"][0] == "ABORTED"
                          and report["rollback"]["reread"][victims[3]["product_code"]] == "200035" and all(report["rollback"]["reread"][v["product_code"]] == "" for v in victims[:3])
                          and report["reapply"]["codes"] == report["reapply"]["expected"] and report["audit_fields_complete"])
    (ROOT / "docs" / "runs" / "rtc_copy_write_e2e.json").write_text(json.dumps(report, ensure_ascii=False, indent=1, default=str), encoding="utf-8")
    print(json.dumps({k: report[k] for k in report if k not in ("audit_sample", "coverage_before", "resolution_before")}, ensure_ascii=False, default=str)[:3500])


if __name__ == "__main__":
    main()