diff --git a/agent/src/gti_fiscal_agent/cfop_ippt_auto_apply.py b/agent/src/gti_fiscal_agent/cfop_ippt_auto_apply.py deleted file mode 100644 index 4daa792..0000000 --- a/agent/src/gti_fiscal_agent/cfop_ippt_auto_apply.py +++ /dev/null @@ -1,185 +0,0 @@ -"""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 deleted file mode 100644 index 58b0875..0000000 --- a/agent/tests/test_cfop_ippt_auto_apply.py +++ /dev/null @@ -1,213 +0,0 @@ -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/docs/CFOP_IPPT_WRITE_CAPABILITY_V1.md b/docs/CFOP_IPPT_WRITE_CAPABILITY_V1.md deleted file mode 100644 index fed09e0..0000000 --- a/docs/CFOP_IPPT_WRITE_CAPABILITY_V1.md +++ /dev/null @@ -1,67 +0,0 @@ -# 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).