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 <noreply@anthropic.com>
This commit is contained in:
gti
Claude Sonnet 5
@@ -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.<field> 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")}
|
||||
@@ -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 <field> 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
|
||||
@@ -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?
|
||||
|
||||
|
||||
@@ -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).
|
||||
+14
-8
@@ -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?
|
||||
|
||||
|
||||
Reference in New Issue
Block a user