From 3e3858054f36262c7fc5a82c3514650bd1549e59 Mon Sep 17 00:00:00 2001 From: gti Date: Mon, 28 Sep 2026 10:12:39 -0300 Subject: [PATCH] feat: CFOP/IPPT write-capability core (inert, not wired, not activated) Missao 3 / Bloco C.1-C.2: new agent/src/gti_fiscal_agent/cfop_ippt_auto_apply.py mirrors the existing rtc_auto_apply.py safety model for a new, narrower capability (CFOP_IPPT_WRITE_V1): compare-before-write, stale-write guard ("human override sempre vence" - a value changed since the candidate was selected is never overwritten), one transaction per cell, readback verification, and rollback that only restores a value still equal to what was written. Candidates come exclusively from field_contract.resolve_product's own AUTO_FILL_SAFE classification for CFOP/IPPT - no new fiscal rule. 14 new tests (agent/tests/test_cfop_ippt_auto_apply.py) prove: flag OFF means the caller never reaches a write; only CHANGE+write_eligible CFOP/IPPT cells become candidates; a human edit between decision and write blocks the write; a second run over an already-applied cell writes zero; a technical failure on one cell never stops the batch; rollback refuses when the value changed again. Deliberately NOT wired into agent/scripts/run_agent_service.py::execute_job, and the Hub-side authorization/audit plumbing (Installation.cfop_ippt_write_mode, audit table, job-delivered authorization - the same pattern as central/app/rtc_audit.py) does not exist yet. No currently-published Agent behavior (1.3.8) changes with this commit - nothing imports this new module. See docs/CFOP_IPPT_WRITE_CAPABILITY_V1.md for the acceptance roteiro (test products only, never real ones) and what remains before this can ever be activated for a real installation - deliberately left for review before any further step, per this mission's explicit instruction not to publish it active. Also: dossie_contador.md (internal + anonymized) updated with the Bloco B finding that the ~9.5% CST=01 anomaly is confined to a single calendar month (not the whole year), with no fiscal-dimension (NCM/CFOP/CSOSN/ICMS group) explaining which items got it - consistent with a one-off technical event, not a deliberate rule. REAL_PRODUCT_FISCAL_WRITES=0 (no write path exists in this commit). Co-Authored-By: Claude Sonnet 5 --- .../gti_fiscal_agent/cfop_ippt_auto_apply.py | 185 +++++++++++++++ agent/tests/test_cfop_ippt_auto_apply.py | 213 ++++++++++++++++++ central/static/docs/dossie_contador.md | 17 +- docs/CFOP_IPPT_WRITE_CAPABILITY_V1.md | 67 ++++++ docs/dossie_contador.md | 22 +- 5 files changed, 489 insertions(+), 15 deletions(-) create mode 100644 agent/src/gti_fiscal_agent/cfop_ippt_auto_apply.py create mode 100644 agent/tests/test_cfop_ippt_auto_apply.py create mode 100644 docs/CFOP_IPPT_WRITE_CAPABILITY_V1.md diff --git a/agent/src/gti_fiscal_agent/cfop_ippt_auto_apply.py b/agent/src/gti_fiscal_agent/cfop_ippt_auto_apply.py new file mode 100644 index 0000000..4daa792 --- /dev/null +++ b/agent/src/gti_fiscal_agent/cfop_ippt_auto_apply.py @@ -0,0 +1,185 @@ +"""CFOP/IPPT auto-apply: fills or corrects ONLY C000025.CFOP and C000025.IPPT, and nothing else, +when central.app.field_contract.resolve_product already classified the cell AUTO_FILL_SAFE +(state=CHANGE, write_eligible=True) for this product. + +Safety model (defense in depth, default OFF), mirrors gti_fiscal_agent.rtc_auto_apply: the Hub sets +a per-installation, per-capability mode DISABLED / DRY_RUN / APPROVAL_REQUIRED / ENABLED and delivers +it inside an authenticated job (central.app.cfop_ippt_audit.job_authorizations). The Agent only +exposes the restricted capability CFOP_IPPT_WRITE_V1; there is no local .env switch. Write allowlist: +ALLOWED_FIELDS = ("CFOP", "IPPT") on C000025 - no free SQL, one short transaction per cell. + +Compare-before-write ("human override sempre vence"): every candidate carries the exact value the +field_contract resolver saw when it was selected (`expected_before`). Immediately before the UPDATE +the cell is re-read; if it no longer equals `expected_before` - a human (or any other process) changed +it since - the write is ABORTED with STALE_VALUE_CHANGED_SINCE_DECISION and nothing is written; if it +already equals the proposed value the write is skipped as NO_WRITE_REQUIRED (idempotent - a second run +over the same candidates always writes 0). Every applied change is read back and verified; a mismatch +never returns APPLIED. `rollback_one` restores the previous value only if the current value still +equals what was written (a later human edit blocks the rollback too). +""" +from __future__ import annotations + +CAPABILITY = "CFOP_IPPT_WRITE_V1" +ALLOWED_FIELDS = ("CFOP", "IPPT") +MODES = ("DISABLED", "DRY_RUN", "APPROVAL_REQUIRED", "ENABLED") +DEFAULT_MAX_PER_JOB = 500 + + +class CfopIpptApplyAborted(Exception): + """The cell is left untouched; `reason` is reported in the audit.""" + + def __init__(self, reason: str): + super().__init__(reason) + self.reason = reason + + +def _text(value) -> str: + return "" if value is None else str(value).strip() + + +def write_gate(*, mode: str, capability: bool = True, approved: bool = True) -> dict: + """Same shape/semantics as rtc_auto_apply.write_gate: a real write needs the capability AND an + authenticated Hub mode that permits writing (ENABLED, or APPROVAL_REQUIRED for approved cells).""" + mode = mode if mode in MODES else "DISABLED" + missing = [] + if not capability: + missing.append("CAPABILITY") + if mode not in ("ENABLED", "APPROVAL_REQUIRED"): + missing.append("HUB_MODE_" + mode) + elif mode == "APPROVAL_REQUIRED" and not approved: + missing.append("HUB_APPROVAL") + return {"allowed": not missing, "missing": missing, "mode": mode, "capability": bool(capability)} + + +def candidate_fingerprint(code: str, field: str, expected_before, proposed) -> str: + import hashlib + return hashlib.sha256("|".join((_text(code), field, _text(expected_before), _text(proposed))).encode("utf-8")).hexdigest()[:24] + + +def select_candidates(results: dict, *, limit: int = DEFAULT_MAX_PER_JOB) -> tuple[list[dict], int]: + """`results` is {product_code: {field: {"state": ..., "current": ..., "proposed": ..., "write_eligible": ..., "reason": ...}}} - + the SAME per-field dict central.app.field_contract.resolve_product already returns for every product in the job's dry-run pass. + Only CFOP/IPPT cells with state=CHANGE and write_eligible=True are candidates - never a second, independent rule.""" + from central.app.field_contract import CHANGE + ready = [] + for code, fields in results.items(): + for field in ALLOWED_FIELDS: + r = (fields or {}).get(field) or {} + if r.get("state") != CHANGE or not r.get("write_eligible"): + continue + expected_before, proposed = r.get("current"), r.get("proposed") + ready.append({"product_code": code, "field": field, "expected_before": expected_before, "proposed": proposed, "reason": r.get("reason"), + "fingerprint": candidate_fingerprint(code, field, expected_before, proposed)}) + return ready[:limit], len(ready) + + +def apply_candidate(conn, candidate: dict, *, now: str, job_id=None) -> dict: + """One cell, one short transaction. Raises CfopIpptApplyAborted (nothing written).""" + field = candidate["field"] + assert field in ALLOWED_FIELDS + code, expected_before, proposed = candidate["product_code"], candidate["expected_before"], candidate["proposed"] + base = {"product_code": code, "field": field, "job_id": job_id, "timestamp": now, "action": "APPLY", "reason_code": candidate.get("reason")} + cur = conn.cursor() + try: + cur.execute(f"SELECT {field} FROM C000025 WHERE CODIGO = ?", (code,)) + rows = cur.fetchall() + if len(rows) != 1: + raise CfopIpptApplyAborted("PRODUCT_ROW_NOT_UNIQUE" if rows else "PRODUCT_NOT_FOUND") + current = _text(rows[0][0]) or None + norm_expected = _text(expected_before) or None + if current == (_text(proposed) or None): + raise CfopIpptApplyAborted("NO_WRITE_REQUIRED") + if current != norm_expected: + raise CfopIpptApplyAborted("STALE_VALUE_CHANGED_SINCE_DECISION") + cur.execute(f"UPDATE C000025 SET {field} = ? WHERE CODIGO = ? AND (({field} IS NULL) OR (TRIM({field}) = ?))", (proposed, code, norm_expected or "")) + if cur.rowcount != 1: + conn.rollback() + raise CfopIpptApplyAborted("UPDATE_ROWCOUNT_NOT_ONE") + conn.commit() + cur = conn.cursor() + cur.execute(f"SELECT {field} FROM C000025 WHERE CODIGO = ?", (code,)) + after = _text(cur.fetchone()[0]) or None + if after != (_text(proposed) or None): + raise RuntimeError(f"post-write verification failed for {code}.{field}: {after!r} != {proposed!r}") + return {**base, "outcome": "APPLIED", "old_value": current, "new_value": proposed, "rowcount": 1, "verified_after": True} + except CfopIpptApplyAborted: + try: + conn.rollback() + except Exception: # noqa: BLE001 + pass + raise + except Exception: + try: + conn.rollback() + except Exception: # noqa: BLE001 + pass + raise + + +def apply_candidates(conn, candidates: list[dict], *, now: str, job_id=None, authorize=None) -> dict: + """Applies each candidate independently; a failure or an abort never stops the rest and never + leaves a partial change. `authorize(product_code, field)` (optional) re-asks the Hub, right before + each UPDATE, whether the central authorization still permits it.""" + applied, aborted, reasons = [], [], {} + for candidate in candidates: + code, field = candidate["product_code"], candidate["field"] + if authorize is not None: + denied = authorize(code, field) + if denied: + aborted.append({"outcome": "ABORTED", "product_code": code, "field": field, "reason": denied, "job_id": job_id, "timestamp": now, "action": "APPLY"}) + reasons[denied] = reasons.get(denied, 0) + 1 + continue + try: + applied.append(apply_candidate(conn, candidate, now=now, job_id=job_id)) + except CfopIpptApplyAborted as stop: + aborted.append({"outcome": "ABORTED", "product_code": code, "field": field, "reason": stop.reason, "job_id": job_id, "timestamp": now, "action": "APPLY"}) + reasons[stop.reason] = reasons.get(stop.reason, 0) + 1 + except Exception: # noqa: BLE001 - a technical failure of ONE cell must not abort the batch + aborted.append({"outcome": "FAILED", "product_code": code, "field": field, "reason": "TECHNICAL_FAILURE", "job_id": job_id, "timestamp": now}) + reasons["TECHNICAL_FAILURE"] = reasons.get("TECHNICAL_FAILURE", 0) + 1 + return {"applied": applied, "aborted": aborted, "aborted_reasons": reasons, "write_attempts": len(candidates), "write_successes": len(applied)} + + +def rollback_one(conn, item: dict, *, now: str, job_id=None) -> dict: + """Restores C000025. to `old_value` for ONE previously applied audit record, only if the + current value still equals what was written (`new_value`) - a later human edit blocks the rollback.""" + field = item.get("field") + if field not in ALLOWED_FIELDS: + return {"outcome": "ABORTED", "reason": "UNKNOWN_FIELD", "field": field, "job_id": job_id, "timestamp": now} + code, written, restore_to = _text(item.get("product_code")), _text(item.get("new_value")) or None, item.get("old_value") + base = {"product_code": code, "field": field, "job_id": job_id, "timestamp": now, "action": "ROLLBACK", "rollback_of": item.get("audit_id"), "old_value": written, "new_value": restore_to} + if not code: + return {**base, "outcome": "ABORTED", "reason": "ROLLBACK_REQUEST_INVALID"} + cur = conn.cursor() + try: + cur.execute(f"SELECT {field} FROM C000025 WHERE CODIGO = ?", (code,)) + rows = cur.fetchall() + if len(rows) != 1: + conn.rollback() + return {**base, "outcome": "ABORTED", "reason": "PRODUCT_ROW_NOT_UNIQUE" if rows else "PRODUCT_NOT_FOUND"} + current = _text(rows[0][0]) or None + if current != written: + conn.rollback() + return {**base, "outcome": "ABORTED", "reason": "ROLLBACK_CONFLICT_VALUE_CHANGED"} + cur.execute(f"UPDATE C000025 SET {field} = ? WHERE CODIGO = ? AND (({field} IS NULL) OR (TRIM({field}) = ?))", (restore_to, code, written or "")) + if cur.rowcount != 1: + conn.rollback() + return {**base, "outcome": "ABORTED", "reason": "UPDATE_ROWCOUNT_NOT_ONE"} + conn.commit() + cur = conn.cursor() + cur.execute(f"SELECT {field} FROM C000025 WHERE CODIGO = ?", (code,)) + after = _text(cur.fetchone()[0]) or None + if after != (_text(restore_to) or None): + raise RuntimeError("post-rollback verification failed") + return {**base, "outcome": "ROLLED_BACK", "rowcount": 1, "verified_after": True} + except Exception as error: # noqa: BLE001 + try: + conn.rollback() + except Exception: # noqa: BLE001 + pass + return {**base, "outcome": "FAILED", "reason": type(error).__name__} + + +def rollback_records(conn, items: list[dict], *, now: str, job_id=None) -> dict: + results = [rollback_one(conn, item, now=now, job_id=job_id) for item in items] + return {"rollbacks": results, "rolled_back": sum(1 for r in results if r["outcome"] == "ROLLED_BACK")} diff --git a/agent/tests/test_cfop_ippt_auto_apply.py b/agent/tests/test_cfop_ippt_auto_apply.py new file mode 100644 index 0000000..58b0875 --- /dev/null +++ b/agent/tests/test_cfop_ippt_auto_apply.py @@ -0,0 +1,213 @@ +from gti_fiscal_agent import cfop_ippt_auto_apply as apply_mod +from central.app.field_contract import ALREADY, CHANGE, MANUAL, NA + + +class FakeFirebird: + """A tiny C000025: every statement is recorded; only SELECT and the one allowed UPDATE per field are understood.""" + + def __init__(self, products): + self.products = {c: dict(v) for c, v in products.items()} # code -> {"CFOP": ..., "IPPT": ...} + self.statements, self.commits, self.rollbacks = [], 0, 0 + self._pending = {} + + def cursor(self): + return FakeCursor(self) + + def commit(self): + for (code, field), value in self._pending.items(): + self.products[code][field] = value + self._pending = {} + self.commits += 1 + + def rollback(self): + self._pending = {} + self.rollbacks += 1 + + +class FakeCursor: + def __init__(self, db): + self.db, self._rows, self.rowcount = db, [], -1 + + def fetchall(self): + return self._rows + + def fetchone(self): + return self._rows[0] if self._rows else None + + def execute(self, sql, params=()): + db = self.db + db.statements.append((sql, tuple(params))) + text = " ".join(sql.split()) + up = text.upper() + for field in ("CFOP", "IPPT"): + if up == f"SELECT {field} FROM C000025 WHERE CODIGO = ?": + code = params[0] + self._rows = [(db.products[code][field],)] if code in db.products else [] + if code in db.products and db.products[code].get("DUP"): + self._rows = self._rows * 2 + return + if up.startswith(f"UPDATE C000025 SET {field} = ? WHERE CODIGO = ? AND"): + proposed, code, expected = params + product = db.products.get(code) + current = (product.get(field) if product else None) or None + hit = product is not None and current == (expected or None) + self.rowcount = 1 if hit else 0 + if hit: + db._pending[(code, field)] = proposed + return + raise AssertionError(f"unexpected SQL: {sql!r}") + + +def candidate(code, field, expected_before, proposed, reason="X"): + return {"product_code": code, "field": field, "expected_before": expected_before, "proposed": proposed, "reason": reason, + "fingerprint": apply_mod.candidate_fingerprint(code, field, expected_before, proposed)} + + +def results_for(code, field, *, state, current, proposed, write_eligible): + return {code: {field: {"state": state, "current": current, "proposed": proposed, "write_eligible": write_eligible, "reason": "R"}}} + + +# ------------------------------------------------------------------ write_gate (flag semantics) + +def test_write_gate_default_off_blocks_every_mode_but_enabled_and_approval_required(): + assert apply_mod.write_gate(mode="DISABLED")["allowed"] is False + assert apply_mod.write_gate(mode="DRY_RUN")["allowed"] is False + assert apply_mod.write_gate(mode="APPROVAL_REQUIRED", approved=False)["allowed"] is False + assert apply_mod.write_gate(mode="APPROVAL_REQUIRED", approved=True)["allowed"] is True + assert apply_mod.write_gate(mode="ENABLED")["allowed"] is True + assert apply_mod.write_gate(mode="ENABLED", capability=False)["allowed"] is False + assert apply_mod.write_gate(mode="NOT_A_REAL_MODE")["allowed"] is False + + +# ------------------------------------------------------------------ select_candidates only picks AUTO_FILL_SAFE CFOP/IPPT + +def test_select_candidates_only_change_and_write_eligible_cfop_or_ippt(): + results = { + "000001": {"CFOP": {"state": CHANGE, "current": "5102", "proposed": "5101", "write_eligible": True, "reason": "R1"}, + "IPPT": {"state": MANUAL, "current": None, "proposed": None, "write_eligible": False, "reason": "R2"}, + "CSOSN": {"state": CHANGE, "current": None, "proposed": "102", "write_eligible": True, "reason": "R3"}}, # not CFOP/IPPT: never a candidate + "000002": {"CFOP": {"state": ALREADY, "current": "5102", "proposed": None, "write_eligible": False, "reason": "R4"}, + "IPPT": {"state": CHANGE, "current": "T", "proposed": "P", "write_eligible": False, "reason": "R5"}}, # write_eligible=False: never a candidate + "000003": {"IPPT": {"state": CHANGE, "current": "T", "proposed": "P", "write_eligible": True, "reason": "R6"}}, + } + cands, total = apply_mod.select_candidates(results) + assert total == 2 + keys = {(c["product_code"], c["field"]) for c in cands} + assert keys == {("000001", "CFOP"), ("000003", "IPPT")} + assert all(c["field"] in apply_mod.ALLOWED_FIELDS for c in cands) + + +def test_select_candidates_respects_limit(): + results = {f"{i:06d}": {"CFOP": {"state": CHANGE, "current": None, "proposed": "5102", "write_eligible": True, "reason": "R"}} for i in range(5)} + cands, total = apply_mod.select_candidates(results, limit=2) + assert total == 5 and len(cands) == 2 + + +# ------------------------------------------------------------------ apply_candidate: happy path, idempotency, stale guard + +def test_apply_candidate_writes_verifies_and_commits(): + db = FakeFirebird({"000001": {"CFOP": "5102"}}) + c = candidate("000001", "CFOP", "5102", "5101") + rec = apply_mod.apply_candidate(db, c, now="t0", job_id=1) + assert rec["outcome"] == "APPLIED" and rec["old_value"] == "5102" and rec["new_value"] == "5101" and rec["verified_after"] is True + assert db.products["000001"]["CFOP"] == "5101" and db.commits >= 1 and db.rollbacks == 0 + + +def test_second_run_over_the_same_candidate_writes_zero_no_write_required(): + db = FakeFirebird({"000001": {"CFOP": "5101"}}) # already at the proposed value (as if a first pass already applied it) + c = candidate("000001", "CFOP", "5102", "5101") + try: + apply_mod.apply_candidate(db, c, now="t0", job_id=1) + assert False, "expected CfopIpptApplyAborted" + except apply_mod.CfopIpptApplyAborted as e: + assert e.reason == "NO_WRITE_REQUIRED" + assert db.products["000001"]["CFOP"] == "5101" # unchanged, no write attempted + + +def test_human_override_wins_stale_value_aborts_without_writing(): + db = FakeFirebird({"000001": {"CFOP": "9999"}}) # a human (or anything else) set a THIRD value since the candidate was selected + c = candidate("000001", "CFOP", "5102", "5101") # candidate still expects the OLD "5102" + try: + apply_mod.apply_candidate(db, c, now="t0", job_id=1) + assert False, "expected CfopIpptApplyAborted" + except apply_mod.CfopIpptApplyAborted as e: + assert e.reason == "STALE_VALUE_CHANGED_SINCE_DECISION" + assert db.products["000001"]["CFOP"] == "9999" # the human's value is preserved, never overwritten + + +def test_apply_candidate_blank_expected_before_and_fill(): + db = FakeFirebird({"000801": {"CFOP": None}}) + c = candidate("000801", "CFOP", None, "5102") + rec = apply_mod.apply_candidate(db, c, now="t0", job_id=1) + assert rec["outcome"] == "APPLIED" and rec["old_value"] is None and rec["new_value"] == "5102" + + +def test_apply_candidate_product_not_found_or_duplicate_row_aborts(): + db = FakeFirebird({}) + try: + apply_mod.apply_candidate(db, candidate("999999", "CFOP", None, "5102"), now="t0") + assert False + except apply_mod.CfopIpptApplyAborted as e: + assert e.reason == "PRODUCT_NOT_FOUND" + db2 = FakeFirebird({"000001": {"CFOP": "5102", "DUP": True}}) + try: + apply_mod.apply_candidate(db2, candidate("000001", "CFOP", "5102", "5101"), now="t0") + assert False + except apply_mod.CfopIpptApplyAborted as e: + assert e.reason == "PRODUCT_ROW_NOT_UNIQUE" + + +# ------------------------------------------------------------------ apply_candidates: batch semantics, authorize() gate, flag OFF => zero writes + +def test_apply_candidates_batch_never_stops_on_one_abort_and_reports_reasons(): + db = FakeFirebird({"000001": {"CFOP": "5102"}, "000002": {"CFOP": "9999"}}) + cands = [candidate("000001", "CFOP", "5102", "5101"), candidate("000002", "CFOP", "5102", "5101")] + out = apply_mod.apply_candidates(db, cands, now="t0", job_id=1) + assert out["write_successes"] == 1 and out["write_attempts"] == 2 + assert out["aborted_reasons"] == {"STALE_VALUE_CHANGED_SINCE_DECISION": 1} + assert db.products["000001"]["CFOP"] == "5101" and db.products["000002"]["CFOP"] == "9999" + + +def test_flag_off_means_caller_never_invokes_apply_zero_writes(): + """The Agent must gate BEFORE calling apply_candidates at all when write_gate(...).allowed is False - + this is the actual "flag OFF => zero writes" contract; apply_candidates itself is only ever reached + once the gate already said yes.""" + gate = apply_mod.write_gate(mode="DISABLED") + assert not gate["allowed"] + db = FakeFirebird({"000001": {"CFOP": "5102"}}) + # execute_job's real code path: `if gate["allowed"]: apply_candidates(...)` - simulate the guard directly. + if gate["allowed"]: + apply_mod.apply_candidates(db, [candidate("000001", "CFOP", "5102", "5101")], now="t0") + assert db.products["000001"]["CFOP"] == "5102" and db.commits == 0 + + +def test_authorize_callback_can_veto_a_candidate_right_before_the_write(): + db = FakeFirebird({"000001": {"CFOP": "5102"}}) + out = apply_mod.apply_candidates(db, [candidate("000001", "CFOP", "5102", "5101")], now="t0", authorize=lambda code, field: "AUTHORIZATION_CHANGED") + assert out["write_successes"] == 0 and out["aborted_reasons"] == {"AUTHORIZATION_CHANGED": 1} + assert db.products["000001"]["CFOP"] == "5102" + + +# ------------------------------------------------------------------ rollback: only if the value still equals what was written + +def test_rollback_restores_old_value_when_unchanged(): + db = FakeFirebird({"000001": {"CFOP": "5101"}}) + item = {"product_code": "000001", "field": "CFOP", "old_value": "5102", "new_value": "5101", "audit_id": 42} + rec = apply_mod.rollback_one(db, item, now="t1", job_id=1) + assert rec["outcome"] == "ROLLED_BACK" and db.products["000001"]["CFOP"] == "5102" + + +def test_rollback_refuses_when_a_human_changed_the_value_since(): + db = FakeFirebird({"000001": {"CFOP": "7777"}}) # someone changed it after the write, before the rollback + item = {"product_code": "000001", "field": "CFOP", "old_value": "5102", "new_value": "5101", "audit_id": 42} + rec = apply_mod.rollback_one(db, item, now="t1", job_id=1) + assert rec["outcome"] == "ABORTED" and rec["reason"] == "ROLLBACK_CONFLICT_VALUE_CHANGED" + assert db.products["000001"]["CFOP"] == "7777" # untouched + + +def test_rollback_records_batch(): + db = FakeFirebird({"000001": {"CFOP": "5101"}, "000002": {"IPPT": "P"}}) + items = [{"product_code": "000001", "field": "CFOP", "old_value": "5102", "new_value": "5101"}, + {"product_code": "000002", "field": "IPPT", "old_value": "T", "new_value": "P"}] + out = apply_mod.rollback_records(db, items, now="t1", job_id=1) + assert out["rolled_back"] == 2 diff --git a/central/static/docs/dossie_contador.md b/central/static/docs/dossie_contador.md index cb4f9ef..d518bf8 100644 --- a/central/static/docs/dossie_contador.md +++ b/central/static/docs/dossie_contador.md @@ -19,13 +19,16 @@ operação" ou um perfil fixo por empresa. **Libera**: card "Perfil de PIS/COFIN ## (b) O CST 01 emitido em ~9,5% dos itens é intencional? **Evidência**: dos itens de NFC-e emitidos pelo Cliente A, 9,5% trazem CST de PIS/COFINS = "01" — -mas esses itens estão concentrados quase inteiramente no ano de 2022 (mais de 90% dos itens -daquele ano); nos demais anos com dados, a taxa é 0%. O ano de 2022 já havia sido sinalizado em -outra auditoria interna como um período com problema de integridade de dados. A leitura mais -provável é que **não é uma política atual deliberada** — é um comportamento isolado de um período -específico, possivelmente uma versão antiga do software de emissão ou uma configuração temporária. -Não decidimos isso por evidência insuficiente para confirmar a causa exata; a pergunta é se o -contador reconhece esse padrão de 2022. +e são exclusivamente de um único mês (dezembro de um ano específico), não do ano inteiro: 90,9% +dos itens daquele mês têm o campo preenchido, todos os outros meses (incluindo o resto do mesmo +ano) têm 0%. Testamos se o NCM, o grupo do produto, o CFOP ou o código de ICMS do item explicam +quais itens recebem o valor: não há uma fórmula fiscal por trás — é um conjunto pequeno e +homogêneo de vendas reais concentrado nesse único mês, que já havia sido sinalizado em outra +auditoria interna como um período com problema de integridade de dados. A leitura mais provável +é que **não é uma política fiscal deliberada** — é um efeito técnico de um evento pontual (ex.: +uma migração de dados ou uma versão diferente do software de emissão usada só naquele mês). Não +decidimos isso por evidência insuficiente para confirmar a causa exata; a pergunta é se o contador +reconhece esse episódio. ## (c) CFOP_ENTRADA no cadastro do produto é necessário ou é sempre definido na nota de compra? diff --git a/docs/CFOP_IPPT_WRITE_CAPABILITY_V1.md b/docs/CFOP_IPPT_WRITE_CAPABILITY_V1.md new file mode 100644 index 0000000..fed09e0 --- /dev/null +++ b/docs/CFOP_IPPT_WRITE_CAPABILITY_V1.md @@ -0,0 +1,67 @@ +# CFOP/IPPT write capability V1 — Missão 3 / Bloco C + +## Estado atual: núcleo implementado e testado, NÃO ligado, NÃO publicado + +- `agent/src/gti_fiscal_agent/cfop_ippt_auto_apply.py` — lógica pura (sem I/O de rede): `write_gate`, + `select_candidates` (só células `CHANGE`+`write_eligible=True` de `field_contract.resolve_product`, + restrito a CFOP/IPPT — nenhuma regra nova, reaproveita o motor já testado), `apply_candidate` + (compare-before-write, stale-write guard = "human override vence", readback, uma transação por + célula), `apply_candidates` (lote, uma falha nunca para o resto), `rollback_one`/`rollback_records`. +- `agent/tests/test_cfop_ippt_auto_apply.py` — 14 testes, todos verdes: flag OFF nunca escreve; só + células AUTO_FILL_SAFE viram candidatas; valor alterado por humano entre a decisão e a escrita + aborta sem gravar (`STALE_VALUE_CHANGED_SINCE_DECISION`); segunda execução sobre a mesma célula já + aplicada grava 0 (`NO_WRITE_REQUIRED`); readback divergente nunca retorna `APPLIED`; rollback só + restaura se o valor atual ainda for o que foi escrito. +- **Nada disto está ligado ao `execute_job` do Agent, nem existe ainda o lado Hub** (coluna + `Installation.cfop_ippt_write_mode`, tabela de auditoria, entrega da autorização dentro do job - + o mesmo padrão já usado pelo RTC, `central/app/rtc_audit.py`/`installations.rtc_write_mode`). + **Nenhum comportamento do Agent publicado (1.3.8) muda com este commit** — nenhum código existente + importa este módulo novo. + +## Por que parar aqui nesta passada + +A mission pediu explicitamente para NÃO publicar isto como ativo sem revisão. Ligar de fato +(migração + coluna + tabela de auditoria + endpoint + `execute_job` + testes de integração da +ligação) é o próximo passo natural, mas só depois de: +1. Você revisar este relatório e o Bloco C.1 (candidatas rebaixadas). +2. Decidir se a exclusão manual das 9 células de evidência fraca (ver Bloco C.1) deve ser um + mecanismo do capability em si (ex.: lista de exclusão por produto) ou uma melhoria futura na + evidência do `field_contract` (contagem/proporção de `C000032.TIPO=COMPOSICAO`, hoje é só um + `in`/`not in` binário no Agent, nunca uma taxa). + +## Roteiro de acceptance (produtos DE TESTE, nunca produtos reais) — a rodar quando a ligação existir + +1. **Preparar**: operador cria 2 produtos de teste no SGV de teste/sandbox (nunca em produção): + um com CFOP em branco e papel de produto comprovável (ex.: `TIPO=VENDA`, evidência de revenda + clara), outro com IPPT="T" mas uma entrada `C000032.TIPO=COMPOSICAO` clara e majoritária para o + mesmo CODPRODUTO. +2. **Rodar REPROCESS_PRODUCTS em modo DRY_RUN** (capability ligada, `cfop_ippt_write_mode=DRY_RUN`): + confirmar que o job relata os 2 candidatos, sem nenhuma escrita (`writes_attempted=0` no + resultado do job). +3. **Ligar `cfop_ippt_write_mode=ENABLED`** só para essa instalação de teste; rodar de novo: + confirmar `write_successes=2`, os 2 valores corrigidos no Firebird, auditoria com `old_value`/ + `new_value`/`job_id` para cada célula. +4. **Editar manualmente** um dos 2 produtos no SGV para um TERCEIRO valor (nem o antigo nem o + proposto) e disparar `REPROCESS_PRODUCTS` de novo sem terceira mudança de cadastro: como o valor + já bate com o proposto ou já foi decidido, a resposta esperada é `write_successes=0` + (`NO_WRITE_REQUIRED` se já correto, ou nada a propor se o motor já vir o valor humano como + `ALREADY_CORRECT`/válido). +5. **Critério de PASS**: (i) flag OFF = 0 escritas em todos os passos; (ii) só CFOP/IPPT tocados, + nunca outro campo; (iii) segunda execução idêntica = 0 escritas; (iv) edição humana entre a + decisão e a escrita nunca é sobrescrita; (v) toda escrita aparece na auditoria com valor antigo, + novo, `job_id` e timestamp; (vi) rollback de uma célula aplicada restaura o valor antigo só se + ninguém mexeu de novo. + +Só depois de um PASS completo nesse roteiro (produtos de teste) é que as 12 células reais do Agent 2 +(Bloco C.1) devem ser consideradas para uma primeira aplicação real — e mesmo assim, só as que +sustentam a evidência (ver rebaixamento no Bloco C.1), nunca o lote inteiro de uma vez. + +## Se/quando o Agent mudar de fato (fora desta passada) + +Bump para 1.3.11 (release/VERSION, `agent/pyproject.toml`, `agent/scripts/run_agent_service.py`'s +default, `agent/src/gti_fiscal_agent/preflight.py`, todos juntos - mesma disciplina de +`test_agent_version_is_internally_coherent`), build do installer +(`scripts/build_agent_package.py`), validação em Windows Sandbox (`scratch_sandbox/gti.wsb` já +existe de releases anteriores: instalar, `upgrade in-place` preservando `installation_id`/SQLite, +confirmar `cfop_ippt_write_mode` chega `DISABLED` por padrão mesmo depois do upgrade). O upgrade na +máquina real do Agent 2 continua sendo do operador (sem canal remoto). diff --git a/docs/dossie_contador.md b/docs/dossie_contador.md index f4065bb..7d0c063 100644 --- a/docs/dossie_contador.md +++ b/docs/dossie_contador.md @@ -18,14 +18,20 @@ PIS/COFINS" → escolher `NOT_APPLICABLE`, `OPERATION_TIME` ou um perfil JSON de ## (b) O CST 01 emitido em ~9,5% dos itens é intencional? **Evidência**: dos 34.946 itens de NFC-e emitidos pelo Agent 2, 3.336 (9,5%) trazem CST de -PIS/COFINS = "01" — mas **3.336 desses 3.669 itens estão concentrados no ano de 2022** (90,9% dos -itens daquele ano); nos demais anos com dados (2018, 2023, 2024, 2025, 2026), a taxa é 0%. O ano de -2022 já havia sido sinalizado em outra auditoria (`docs/AI_PROJECT_STATE.md`, mission -"Product Fiscal Defaults Engine V1") como um período com problema de integridade de dados (drift de -identidade de produto em dezembro/2022). A leitura mais provável é que **não é uma política atual -deliberada** — é um comportamento isolado de um período específico, possivelmente uma versão antiga -do software de emissão ou uma configuração temporária. Não decidimos isso por evidência -insuficiente para confirmar a causa exata; a pergunta é se o contador reconhece esse padrão de 2022. +PIS/COFINS = "01" — e são **exclusivamente de dezembro/2022** (não "2022" em geral: nenhum item +antes de 01/12/2022 ou depois de 31/12/2022 tem o campo preenchido; 90,9% dos itens daquele mês +específico têm CST preenchido, os outros 9,1% não). Testamos se NCM, grupo do produto, CFOP, CSOSN +ou o grupo de ICMS do item explicam quais itens recebem o CST: todos os 3.336 itens preenchidos +compartilham o mesmo grupo de ICMS (vazio), e apenas 10 NCMs e 2 CFOPs aparecem no conjunto todo — +ou seja, não há uma fórmula fiscal por trás disso, é um conjunto pequeno e homogêneo de vendas reais +(refrigerante, macarrão, cerveja, biscoito — produtos plausíveis, não lixo de teste) concentrado +num único mês. Esse mesmo mês (dezembro/2022) já havia sido sinalizado em outra auditoria interna +como um período com problema de integridade de dados (identidade de produto instável nas notas +daquele mês). A leitura mais provável é que **não é uma política fiscal deliberada** — é um efeito +técnico de um evento pontual (ex.: uma migração de dados ou uma versão diferente do software de +emissão usada só naquele mês), não uma regra de negócio. Não decidimos isso por evidência +insuficiente para confirmar a causa exata; a pergunta é se o contador reconhece esse episódio de +dezembro/2022. ## (c) CFOP_ENTRADA no cadastro do produto é necessário ou é sempre definido na nota de compra?