"""GTI Fiscal LOCAL COMPLETION V1 runner: snapshot -> dry-run -> guarded apply -> verify -> second pass -> reports/questions. python scripts/local_completion.py [--db NAME ...] stages: snapshot | dryrun | apply | verify | second | report | questions | all Databases are ONLY those in docs/LOCAL_WRITE_ALLOWLIST_V1.json (exact canonical paths). Writes go through scripts/firebird_write_executor.execute_guarded_product_update (path allowlist, schema/type/length guard, compare-before-write, one UPDATE per product, rowcount, independent re-read). Nothing is written by any stage except `apply` (and `second`, which must write 0). State lives in docs/runs/local_completion/ (git-ignored bulk files). """ import argparse import gzip import hashlib import importlib.util import json import os import pickle import subprocess import sys import tempfile from collections import Counter, defaultdict from datetime import datetime, timezone from pathlib import Path 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")) import fdb # noqa: E402 import run_fiscal_decision_v1 as base # noqa: E402 from central.app.field_contract import ALREADY, CHANGE, CONTRACT, CONTRACT_VERSION, CORRECTED, MANUAL, NA, WRITE_FIELDS, Context, resolve_product # noqa: E402 from gti_fiscal_agent import company_context as cc # noqa: E402 from gti_fiscal_agent.coverage import build_refs # noqa: E402 from gti_fiscal_agent.firebird_reader import read_evidence # noqa: E402 STATE = ROOT / "docs" / "runs" / "local_completion" HUB = "http://localhost:8000" # Per-database company configuration (operator decisions recorded in docs/AGENT_COMPANY_CONTEXT_RULES_V1.md): the three databases with a CRT conflict use the # emission authority EMISSOR; the company is always the C000004 row 000001 (the one whose CNPJ equals the EMISSOR CNPJ). COMPANY_CFG = {"CONSTRUCAO": {"company_code": "000001"}, "MMMODAS": {"company_code": "000001", "crt_source": "EMISSOR"}, "RINCAO": {"company_code": "000001", "crt_source": "EMISSOR"}, "UNIAO": {"company_code": "000001", "crt_source": "EMISSOR"}, "TESTESOFICIAIS": {}} BASE_COLS = ["CODIGO", "PRODUTO", "CODGRUPO", "TIPO", "PROD_COZINHA", "PRODUTO_SERVICO"] s_ = lambda v: None if v is None else str(v).strip() # noqa: E731 def load_executor(): spec = importlib.util.spec_from_file_location("firebird_write_executor", ROOT / "scripts" / "firebird_write_executor.py") mod = importlib.util.module_from_spec(spec); sys.modules[spec.name] = mod; spec.loader.exec_module(mod) return mod def allowlist(): doc = json.loads((ROOT / "docs" / "LOCAL_WRITE_ALLOWLIST_V1.json").read_text(encoding="utf-8")) return {d["name"]: d["path"] for d in doc["databases"]} def connect(path, rw=False): kw = dict(database=path, 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 env_for(path): return {"FIREBIRD_DATABASE_PATH": path, "FB_USER": os.environ["FB_USER"], "FB_PASSWORD": os.environ["FB_PASSWORD"], "FIREBIRD_CLIENT_LIBRARY": os.environ.get("FIREBIRD_CLIENT_LIBRARY")} # ------------------------------------------------------------------------------------------------------------------ snapshot def snapshot(name, path): conn = connect(path); cur = conn.cursor() cur.execute("SELECT TRIM(rf.RDB$FIELD_NAME), f.RDB$FIELD_TYPE FROM RDB$RELATION_FIELDS rf JOIN RDB$FIELDS f ON f.RDB$FIELD_NAME = rf.RDB$FIELD_SOURCE WHERE rf.RDB$RELATION_NAME = 'C000025' ORDER BY rf.RDB$FIELD_POSITION") cols = [c for c, t in cur.fetchall() if t != 261] # blobs (images) excluded cur.execute(f"SELECT {', '.join(cols)} FROM C000025 ORDER BY CODIGO") rows = {s_(r[cols.index("CODIGO")]): tuple(r) for r in cur.fetchall()} side = {} for t in ("C000004", "EMISSOR"): try: cur.execute(f"SELECT * FROM {t}") names = [d[0] for d in cur.description] data = [tuple(v.read() if hasattr(v, "read") else v for v in r) for r in cur.fetchall()] side[t] = {"columns": names, "rows": data, "sha256": hashlib.sha256(repr(data).encode("utf-8", "replace")).hexdigest()} except Exception as e: # noqa: BLE001 side[t] = {"error": type(e).__name__} company = cc.resolve(cc.collect_evidence(cur), cc.config_from_env({k.upper(): v for k, v in _cfg_env(name).items()})) conn.close() snap = {"name": name, "path": path, "taken": datetime.now(timezone.utc).isoformat(timespec="seconds"), "columns": cols, "rows": rows, "side_tables": side, "product_count": len(rows), "products_logical_sha256": hashlib.sha256(repr(sorted(rows.items())).encode("utf-8", "replace")).hexdigest(), "company": {k: company.get(k) for k in ("status", "classification", "company_code", "company_cnpj", "company_uf", "company_crt", "source_of_truth", "confidence", "warnings", "reasons")}} STATE.mkdir(parents=True, exist_ok=True) with gzip.open(STATE / f"{name}_before.pkl.gz", "wb") as f: pickle.dump(snap, f) return snap def _cfg_env(name): cfg = COMPANY_CFG.get(name, {}) out = {} if cfg.get("company_code"): out["COMPANY_CODE"] = cfg["company_code"] if cfg.get("crt_source"): out["COMPANY_CRT_SOURCE"] = cfg["crt_source"] return out def load_snapshot(name): with gzip.open(STATE / f"{name}_before.pkl.gz", "rb") as f: return pickle.load(f) # ------------------------------------------------------------------------------------------------------------------ analysis def classify_pairs(pairs): pairs = sorted(set(pairs)) with tempfile.TemporaryDirectory() as tmp: fin, fout = Path(tmp) / "in.json", Path(tmp) / "out.json" fin.write_text(json.dumps(pairs, ensure_ascii=False), encoding="utf-8") subprocess.run(["docker", "cp", str(ROOT / "scripts" / "bulk_classify_in_container.py"), "projeto-fiscal-hub-1:/tmp/bulk.py"], check=True) subprocess.run(["docker", "cp", str(fin), "projeto-fiscal-hub-1:/tmp/in.json"], check=True) subprocess.run(["docker", "exec", "projeto-fiscal-hub-1", "sh", "-c", "cd /app && python /tmp/bulk.py /tmp/in.json /tmp/out.json"], check=True, capture_output=True) subprocess.run(["docker", "cp", "projeto-fiscal-hub-1:/tmp/out.json", str(fout)], check=True) return dict(zip(pairs, json.loads(fout.read_text(encoding="utf-8")))) def hub_get(path, token): return base.http(HUB + path, token) def analyze(name, path, token, snap=None, extra_context=None): """Read the database (read-only) and resolve every product/field. Returns the full analysis.""" conn = connect(path); cur = conn.cursor() cur.execute("SELECT TRIM(RDB$FIELD_NAME) FROM RDB$RELATION_FIELDS WHERE RDB$RELATION_NAME = 'C000025'") have = {r[0] for r in cur.fetchall()} cols = [c for c in BASE_COLS + WRITE_FIELDS if c in have] cur.execute(f"SELECT {', '.join(cols)} FROM C000025 ORDER BY CODIGO") products = [] for r in cur.fetchall(): p = {c: None for c in BASE_COLS + WRITE_FIELDS} p.update(zip(cols, r)) for k in p: if isinstance(p[k], str): p[k] = p[k].strip() products.append(p) company = cc.resolve(cc.collect_evidence(cur), cc.config_from_env(_cfg_env(name))) ev_rows = cc.collect_evidence(cur) sel = next((c for c in ev_rows["c000004"] if c["code"] == company["company_code"]), {}) groups = {} if "C000017" in {r[0] for r in _tables(cur)}: cur.execute("SELECT TRIM(CODIGO), TRIM(GRUPO) FROM C000017") groups = dict(cur.fetchall()) contrib = None if "CONTRIBUINTE_IPI" in _cols(cur, "C000004"): cur.execute("SELECT CODIGO, CONTRIBUINTE_IPI FROM C000004") contrib = {s_(a): s_(b) for a, b in cur.fetchall()}.get(company["company_code"]) conn.close() if company["status"] != "PASS": raise RuntimeError(f"{name}: company context {company['classification']}: {company['reasons']}") comp = {"uf": company["company_uf"], "crt": company["company_crt"], "contribuinte_ipi": contrib, "company_code": company["company_code"], "company_key": company["company_cnpj_fingerprint"]} # ERP-emission evidence: the SGV NFC-e history shows whether the PIS/COFINS cadastral fields are consumed at emission (issued CST_PIS_COFINS non-blank share) try: _c = connect(path); _cur = _c.cursor() _cur.execute("SELECT COUNT(*), COUNT(NULLIF(TRIM(CST_PIS_COFINS), '')) FROM NFCE_ITENS") tot, nonblank = _cur.fetchone(); _c.close() comp["pis_cofins_emission_items"] = tot; comp["pis_cofins_emission_nonblank"] = nonblank comp["pis_cofins_unused_by_emission"] = bool(tot >= 1000 and nonblank / tot < 0.001) except Exception as exc: # noqa: BLE001 comp["pis_cofins_unused_by_emission"] = False; comp["pis_cofins_evidence_error"] = str(exc)[:120] evidence = read_evidence(env_for(path), company["company_code"]) refs = build_refs(lambda p: hub_get(p, token), products, comp["uf"]) entries = list(extra_context or []) try: entries += hub_get(f"/v1/context?company_key={comp['company_key']}", token)["entries"] except Exception: # noqa: BLE001 pass ctx = Context(entries) manual = {} for p in products: v, _ = ctx.get("ncm", s_(p["CODIGO"])) if v: manual[s_(p["CODIGO"])] = v pairs = [(s_(p["CLASSIFICACAO_FISCAL"]) or "", s_(p["PRODUTO"]) or "") for p in products] + [(v, s_(p["PRODUTO"]) or "") for p in products for v in [manual.get(s_(p["CODIGO"]))] if v] dec = classify_pairs(pairs) results = {} dup_codes = {c: n for c, n in Counter(s_(p["CODIGO"]) for p in products).items() if n > 1} seen = Counter() for p in products: code = s_(p["CODIGO"]); desc = s_(p["PRODUTO"]) or "" seen[code] += 1 rk = code if seen[code] == 1 else f"{code}#{seen[code]}" p["_rk"] = rk m = manual.get(code) md = dec[(m, desc)] if m else None valid_manual = bool(md and md.get("status") == "EXACT_NCM") d = md if valid_manual else dec[(s_(p["CLASSIFICACAO_FISCAL"]) or "", desc)] results[rk] = resolve_product(p, d, comp, refs, evidence, ctx, manual_ncm=(m, valid_manual) if m else None) if dup_codes.get(code, 1) > 1: # data-integrity defect (PK violation in the copy): every physical row that shares the code is blocked, all 23 cells stay accounted for f in WRITE_FIELDS: r = results[rk][f] r.update(state=MANUAL, reason="DUPLICATE_KEY_BLOCKED", input_key="duplicate_code", input_scope="PRODUCT", write_eligible=False, proposed=None) results[rk]["_ds"] = {k: (v or "")[:8] for k, v in (d.get("dataset_versions") or {}).items() if k in ("ncm", "tipi_ipi", "cest", "ba_st_mva", "cfop", "nfe_moc_tables")} return {"name": name, "path": path, "products": products, "company": comp, "company_result": company, "results": results, "groups": groups, "context_entries": len(entries), "engine": {"contract": CONTRACT_VERSION}} def _tables(cur): cur.execute("SELECT TRIM(RDB$RELATION_NAME) FROM RDB$RELATIONS WHERE COALESCE(RDB$SYSTEM_FLAG,0)=0") return cur.fetchall() def _cols(cur, table): cur.execute("SELECT TRIM(RDB$FIELD_NAME) FROM RDB$RELATION_FIELDS WHERE RDB$RELATION_NAME = ?", (table,)) return {r[0] for r in cur.fetchall()} def summarize(an): per = {f: Counter() for f in WRITE_FIELDS} reasons = {f: Counter() for f in WRITE_FIELDS} manual_keys = Counter() for code, res in an["results"].items(): for f in WRITE_FIELDS: r = res[f] per[f][r["state"]] += 1 reasons[f][(r["state"], r["reason"])] += 1 if r["state"] == MANUAL: manual_keys[(f, r["input_key"])] += 1 roles = Counter(res["_facts"]["role"] for res in an["results"].values()) return {"name": an["name"], "total_products": len(an["products"]), "company": an["company"], "context_entries": an["context_entries"], "fields": {f: dict(per[f]) for f in WRITE_FIELDS}, "reasons": {f: {f"{s}|{r}": n for (s, r), n in reasons[f].most_common(12)} for f in WRITE_FIELDS}, "manual_inputs": {f"{f}|{k}": n for (f, k), n in manual_keys.most_common()}, "roles": dict(roles), "change_required_cells": sum(per[f][CHANGE] for f in WRITE_FIELDS), "write_eligible_cells": sum(1 for res in an["results"].values() for f in WRITE_FIELDS if res[f]["write_eligible"])} def write_ndjson(name, an, suffix): with gzip.open(STATE / f"{name}_{suffix}.ndjson.gz", "wt", encoding="utf-8") as f: for p in an["products"]: code = s_(p["CODIGO"]); res = an["results"][p["_rk"]] for fld in WRITE_FIELDS: r = res[fld] f.write(json.dumps({"db": an["name"], "codigo": code, "row_key": p["_rk"], "field": fld, "before": r["current"], "proposed": r["proposed"], "status": r["state"], "reason": r["reason"], "input_key": r["input_key"], "provenance": f"{CONTRACT_VERSION}|" + ",".join(f"{k}:{v}" for k, v in res["_ds"].items()) + f"|role={res['_facts']['role']}", "write_eligible": r["write_eligible"]}, ensure_ascii=False, default=str) + "\n") # ------------------------------------------------------------------------------------------------------------------ apply / verify def apply(name, path, an, snap): ex = load_executor() ex.assert_local_completion_target(path) run_id = hashlib.sha256(f"{name}|{snap['taken']}|{snap['products_logical_sha256']}".encode()).hexdigest()[:16] now_company = cc.resolve(cc.collect_evidence(_ro_cursor(path)), cc.config_from_env(_cfg_env(name))) if now_company["status"] != "PASS" or (now_company["company_uf"], now_company["company_crt"]) != (an["company"]["uf"], an["company"]["crt"]): raise RuntimeError("company-context guard failed before writing") cols = snap["columns"] conn = connect(path, rw=True) audit, failed, attempted, succeeded = [], [], 0, 0 try: for p in an["products"]: code = s_(p["CODIGO"]); res = an["results"][code] changes = {} for f in WRITE_FIELDS: r = res[f] if r["write_eligible"]: changes[f] = (snap["rows"][code][cols.index(f)], r["proposed"]) if not changes: continue attempted += len(changes) try: rec = ex.execute_guarded_product_update(conn, database_path=path, product_code=code, changes=changes, now=datetime.now(timezone.utc).isoformat(timespec="seconds")) audit.append({"db": name, "run_id": run_id, **rec, "outcome": "SUCCEEDED"}) succeeded += len(changes) except ex.WriteRefused as e: failed.append({"db": name, "run_id": run_id, "product_code": code, "fields": list(changes), "error": str(e)[:200]}) finally: conn.close() with gzip.open(STATE / f"{name}_write_audit.ndjson.gz", "wt", encoding="utf-8") as f: for a in audit + [{**x, "outcome": "FAILED"} for x in failed]: f.write(json.dumps(a, ensure_ascii=False, default=str) + "\n") return {"name": name, "cells_attempted": attempted, "cells_succeeded": succeeded, "cells_failed": sum(len(x["fields"]) for x in failed), "products_updated": len(audit), "failures": failed[:20]} def _ro_cursor(path): return connect(path).cursor() def verify(name, path, an, snap): conn = connect(path); cur = conn.cursor() cols = snap["columns"] cur.execute(f"SELECT {', '.join(cols)} FROM C000025 ORDER BY CODIGO") after = {s_(r[cols.index("CODIGO")]): tuple(r) for r in cur.fetchall()} conn.close() norm = lambda v: None if v is None else (round(float(v), 6) if isinstance(v, (int, float)) or v.__class__.__name__ == "Decimal" else str(v).strip()) # noqa: E731 planned = verified = bad_planned = unplanned_changes = 0 problems = [] for p in an["products"]: code = s_(p["CODIGO"]); before = snap["rows"][code]; aft = after.get(code) if aft is None: problems.append((code, "MISSING_AFTER")); continue exp = {f: an["results"][code][f]["proposed"] for f in WRITE_FIELDS if an["results"][code][f]["write_eligible"]} for i, c in enumerate(cols): if c in exp: planned += 1 if norm(aft[i]) == norm(exp[c]): verified += 1 else: bad_planned += 1; problems.append((code, c, "AFTER!=PROPOSED", aft[i], exp[c])) elif norm(aft[i]) != norm(before[i]): unplanned_changes += 1; problems.append((code, c, "UNPLANNED_CHANGE", before[i], aft[i])) return {"name": name, "products_after": len(after), "products_before": len(snap["rows"]), "planned_cells": planned, "post_write_verified": verified, "planned_mismatch": bad_planned, "unplanned_changes": unplanned_changes, "problems": problems[:20]} # ------------------------------------------------------------------------------------------------------------------ questions def questions(analyses): groups = defaultdict(lambda: {"products": [], "fields": Counter(), "choices": None, "examples": []}) for an in analyses: by_code = {s_(p["CODIGO"]): p for p in an["products"]} for code, res in an["results"].items(): p = by_code[code] for f in WRITE_FIELDS: r = res[f] if r["state"] != MANUAL: continue key, scope = r["input_key"], r["input_scope"] if key in ("ncm",): gk = (an["name"], key, "PRODUCT", code, "") elif key in ("cest", "ex_tipi"): ncm = "".join(c for c in (p["CLASSIFICACAO_FISCAL"] or "") if c.isdigit()) gk = (an["name"], key, "NCM", ncm, "") elif scope == "COMPANY" or key in ("entry_scope", "rounding", "icms_rate", "credit_rate", "pis_cofins_profile", "crt"): gk = (an["name"], key, "COMPANY", "", "") else: grp = s_(p["CODGRUPO"]) or "(sem grupo)" gk = (an["name"], key, "GROUP", grp, an["groups"].get(grp, "")) g = groups[gk] g["products"].append(code); g["fields"][f] += 1 if r.get("choices") and g["choices"] is None: g["choices"] = r["choices"] if len(g["examples"]) < 3: g["examples"].append(f"{code} {s_(p['PRODUTO'])[:40]}") from central.app.context_keys import ENUMS out = [] for (db, key, scope, skey, gname), g in sorted(groups.items()): allowed = sorted(ENUMS[key]) if key in ENUMS else ("JSON {pis_in_cst,pis_out_cst,pis_rate,cofins_in_cst,cofins_out_cst,cofins_rate}" if key == "pis_cofins_profile" else "free text (see key)") out.append({"database": db, "company_key": next(a["company"]["company_key"] for a in analyses if a["name"] == db), "key": key, "scope": scope, "scope_key": skey, "group_name": gname, "products_affected": len(set(g["products"])), "fields_affected": dict(g["fields"]), "allowed_values": allowed, "candidate_choices": g["choices"], "examples": g["examples"], "answer": ""}) return out def main(): ap = argparse.ArgumentParser() ap.add_argument("stage"); ap.add_argument("--db", nargs="*") a = ap.parse_args() base.load_dotenv(ROOT / ".env") token = os.environ["AGENT_DEV_TOKEN"] allow = allowlist() names = a.db or list(allow) for n in names: if n not in allow: raise SystemExit(f"{n} is not in the exact write allowlist") STATE.mkdir(parents=True, exist_ok=True) analyses = [] for name in names: path = allow[name] print(f"== {name} :: {a.stage}", flush=True) if a.stage in ("snapshot", "all"): snap = snapshot(name, path) print(f" snapshot: {snap['product_count']} products, company {snap['company'].get('classification')}", flush=True) if a.stage in ("dryrun", "apply", "verify", "second", "report", "questions", "all"): snap = load_snapshot(name) if a.stage in ("dryrun", "all", "apply"): an = analyze(name, path, token, snap) (STATE / f"{name}_dryrun_summary.json").write_text(json.dumps(summarize(an), ensure_ascii=False, indent=1, default=str), encoding="utf-8") write_ndjson(name, an, "dryrun") print(" dry-run:", json.dumps({k: summarize(an)[k] for k in ("total_products", "change_required_cells", "write_eligible_cells")}), flush=True) if a.stage in ("apply", "all"): res = apply(name, path, an, snap) (STATE / f"{name}_apply_summary.json").write_text(json.dumps(res, ensure_ascii=False, indent=1, default=str), encoding="utf-8") print(" apply:", json.dumps({k: res[k] for k in ("cells_attempted", "cells_succeeded", "cells_failed", "products_updated")}), flush=True) v = verify(name, path, an, snap) (STATE / f"{name}_verify_summary.json").write_text(json.dumps(v, ensure_ascii=False, indent=1, default=str), encoding="utf-8") print(" verify:", json.dumps({k: v[k] for k in ("planned_cells", "post_write_verified", "planned_mismatch", "unplanned_changes")}), flush=True) if a.stage in ("second", "all"): an2 = analyze(name, path, token, snap) s2 = summarize(an2) second = {"name": name, "SECOND_PASS_CHANGE_REQUIRED": s2["change_required_cells"], "SECOND_PASS_WRITES": 0, "fields": s2["fields"]} if s2["write_eligible_cells"]: second["SECOND_PASS_WRITES"] = apply(name, path, an2, snap)["cells_succeeded"] (STATE / f"{name}_second_pass.json").write_text(json.dumps(second, ensure_ascii=False, indent=1, default=str), encoding="utf-8") write_ndjson(name, an2, "second") print(" second pass:", second["SECOND_PASS_CHANGE_REQUIRED"], second["SECOND_PASS_WRITES"], flush=True) if a.stage in ("questions",): analyses.append(analyze(name, path, token, snap)) if a.stage == "questions": q = questions(analyses) (ROOT / "docs" / "LOCAL_FISCAL_CONTEXT_QUESTIONS_V1.json").write_text(json.dumps({"artifact": "LOCAL_FISCAL_CONTEXT_QUESTIONS_V1", "total_questions": len(q), "questions": q}, ensure_ascii=False, indent=1, default=str), encoding="utf-8") import csv with open(ROOT / "docs" / "LOCAL_FISCAL_CONTEXT_QUESTIONS_V1.csv", "w", newline="", encoding="utf-8-sig") as f: w = csv.writer(f) w.writerow(["database", "company_key", "scope", "scope_key", "key", "group_name", "products_affected", "fields_affected", "allowed_values", "examples", "value", "reason"]) for x in q: w.writerow([x["database"], x["company_key"], x["scope"], x["scope_key"], x["key"], x["group_name"], x["products_affected"], json.dumps(x["fields_affected"], ensure_ascii=False), x["allowed_values"] if isinstance(x["allowed_values"], str) else "|".join(x["allowed_values"]), " ; ".join(x["examples"]), "", ""]) print("questions:", len(q)) if __name__ == "__main__": main()