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..e21c116 --- /dev/null +++ b/agent/tests/test_cfop_ippt_auto_apply.py @@ -0,0 +1,239 @@ +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_weak_composed_evidence_never_becomes_a_candidate_regardless_of_the_write_flag(): + """Missão 4 / Bloco A.3: a cell whose role evidence is weak (field_contract itself refuses to + classify it OWN_PRODUCTION/RESALE with confidence) must never reach select_candidates as + AUTO_FILL_SAFE - proven end-to-end from the real resolver, not just asserted in isolation. + This holds independently of cfop_ippt_write_mode/write_gate, which only gates whether + apply_candidates() is ever invoked - it never affects what counts as a candidate.""" + from central.app.field_contract import resolve_product, Context, WRITE_FIELDS + + def product(**kw): + p = {f: None for f in WRITE_FIELDS} + p.update({"CODIGO": "000321", "PRODUTO": "CERVEJA LONG NECK NOVA SCHIN", "CODGRUPO": "G1", "TIPO": "VENDA", "PROD_COZINHA": None, "PRODUTO_SERVICO": None, + "CLASSIFICACAO_FISCAL": "22030000", "CFOP": "5102", "CSOSN": "102", "CST": "000", "SITUACAO_TRIBUTARIA": "T", "IAT": "A", "IPPT": "T", + "INCIDENCIA_PISCOFINS": "NENHUMA", "PISCOFINS": "N"}) + p.update(kw) + return p + + decision = {"status": "EXACT_NCM", "cest_status": "NO_CEST_MAPPING_FOUND", "ipi_status": "IPI_RESOLVED", "ba_st_mva_status": "NO_MVA_MAPPING_FOUND", "dataset_versions": {}, + "classification": {"ncm": "22030000", "ipi": {"rate_kind": "NUMERIC", "rate_numeric": 3.9, "rate": "3.9"}, "ipi_candidates": [], "cest": None, "cest_candidates": [], "ba_st_mva": None, "ba_st_mva_candidates": []}} + refs = {"cfop_valid": {"5101", "5102"}, "csosn": {"102"}, "cst_icms": set(), "cst_pis": set(), "cst_cofins": set(), "origem": set("012345678"), "fcp_uf": {}, "cbs_rate_2026": None, "cclasstrib_count": 0} + co = {"uf": "BA", "crt": "1", "contribuinte_ipi": "N", "company_code": "000001"} + ev_weak = {"recipes": set(), "composed": set(), "composed_ratio": {"000321": {"composed": 1, "other": 19}}, "purchases": {}, "stock_entry": {"000321": True}, "stock_in": {"000321": True}} + result = resolve_product(product(), decision, co, refs, ev_weak, Context([])) + candidates, total = apply_mod.select_candidates({"000321": result}) + assert total == 0 and candidates == [] # the weak-evidence beer never becomes a write candidate + + +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/alembic/versions/0035_cfop_ippt_write_mode.py b/central/alembic/versions/0035_cfop_ippt_write_mode.py new file mode 100644 index 0000000..56da693 --- /dev/null +++ b/central/alembic/versions/0035_cfop_ippt_write_mode.py @@ -0,0 +1,39 @@ +"""CFOP/IPPT write capability: per-installation write mode (DEFAULT OFF) + append-only audit table. + +Revision ID: 0035_cfop_ippt_write_mode +Revises: 0034_reference_data_registry +""" +from alembic import op +import sqlalchemy as sa + +revision = "0035_cfop_ippt_write_mode" +down_revision = "0034_reference_data_registry" +branch_labels = None +depends_on = None + + +def upgrade(): + op.add_column("installations", sa.Column("cfop_ippt_write_mode", sa.String(24), nullable=False, server_default="DISABLED")) + op.create_table( + "cfop_ippt_write_audits", + sa.Column("id", sa.Integer(), primary_key=True), + sa.Column("tenant_id", sa.Integer(), sa.ForeignKey("tenants.id"), nullable=False, index=True), + sa.Column("installation_id", sa.Integer(), sa.ForeignKey("installations.id"), nullable=False, index=True), + sa.Column("agent_id", sa.Integer(), sa.ForeignKey("agents.id"), nullable=False, index=True), + sa.Column("job_id", sa.Integer(), nullable=True, index=True), + sa.Column("product_code", sa.String(64), nullable=False, index=True), + sa.Column("field", sa.String(16), nullable=False), + sa.Column("outcome", sa.String(16), nullable=False, index=True), + sa.Column("reason", sa.String(96), nullable=True), + sa.Column("old_value", sa.String(16), nullable=True), + sa.Column("new_value", sa.String(16), nullable=True), + sa.Column("action", sa.String(16), nullable=True), + sa.Column("rollback_of", sa.Integer(), nullable=True, index=True), + sa.Column("written_at", sa.String(40), nullable=True), + sa.Column("recorded_at", sa.DateTime(timezone=True), server_default=sa.func.now(), index=True), + ) + + +def downgrade(): + op.drop_table("cfop_ippt_write_audits") + op.drop_column("installations", "cfop_ippt_write_mode") diff --git a/central/app/api.py b/central/app/api.py index 297f031..a8f2582 100644 --- a/central/app/api.py +++ b/central/app/api.py @@ -12,6 +12,7 @@ from .config import settings from .datasets.service import DatasetService from .jobs import SYNC_KINDS, claim_next_job, complete_job, create_job, fail_job, schedule_automatic_sync from .rtc_audit import job_authorizations, record_rtc_write_audits +from .cfop_ippt_audit import job_authorizations as cfop_ippt_job_authorizations, record_cfop_ippt_write_audits from .blocked_products import list_blockers, resolve_blocker, resolve_product_blockers, resolve_stale_fiscal_blockers, upsert_blocker, upsert_blockers_bulk from .context_keys import validate_entry from .models import AgentSnapshot, BlockedProduct, FiscalContextEntry, ClassificationCache, ClassificationRequest, ClassificationResult, Dataset, Installation, Job @@ -260,7 +261,9 @@ def router(session_factory, bootstrap_token): session, agent = identity try: installation = session.scalar(select(Installation).where(Installation.agent_id == agent.id)) - return job_authorizations(session, installation) if installation is not None else {"rtc_write_mode": "DISABLED"} + if installation is None: + return {"rtc_write_mode": "DISABLED", "cfop_ippt_write_mode": "DISABLED"} + return {**job_authorizations(session, installation), **cfop_ippt_job_authorizations(installation)} finally: session.close() @@ -407,7 +410,8 @@ def router(session_factory, bootstrap_token): session.commit() logger.info("job claimed job_id=%s agent_id=%s kind=%s", job.id, agent.id, job.kind) view = _job_view(job) - view["authorizations"] = job_authorizations(session, installation) if installation is not None else {"rtc_write_mode": "DISABLED"} # decided centrally, per installation, default OFF + view["authorizations"] = ({**job_authorizations(session, installation), **cfop_ippt_job_authorizations(installation)} if installation is not None + else {"rtc_write_mode": "DISABLED", "cfop_ippt_write_mode": "DISABLED"}) # decided centrally, per installation, default OFF return view finally: session.close() @@ -443,6 +447,7 @@ def router(session_factory, bootstrap_token): resolve_product_blockers(session, agent.tenant_id, product_code, installation_id=installation.id if installation else None) if installation is not None: record_rtc_write_audits(session, tenant_id=agent.tenant_id, installation=installation, agent=agent, job_id=job_id, rtc_auto_apply=result_summary.get("rtc_auto_apply"), evaluated_all=bool((result_summary.get("fiscal_context") or {}).get("evaluated_all"))) + record_cfop_ippt_write_audits(session, tenant_id=agent.tenant_id, installation=installation, agent=agent, job_id=job_id, cfop_ippt_apply=result_summary.get("cfop_ippt_apply")) if installation is not None and (result_summary.get("fiscal_context") or {}).get("evaluated_all"): resolve_stale_fiscal_blockers(session, agent.tenant_id, installation.id, pass_started) if not complete_job(session, job, result_summary): raise HTTPException(409, "job is not RUNNING") diff --git a/central/app/cfop_ippt_audit.py b/central/app/cfop_ippt_audit.py new file mode 100644 index 0000000..6364855 --- /dev/null +++ b/central/app/cfop_ippt_audit.py @@ -0,0 +1,51 @@ +"""Persists the Agent's CFOP/IPPT write audit records (append-only). Mirrors central.app.rtc_audit's +safety model for a separate, narrower capability - no candidate preview table (candidates are +recomputed fresh from field_contract every job, never stored server-side), no rollback queue yet +(rollback is operator-triggered later, out of scope for this pass).""" +from __future__ import annotations + +from .events import record_event +from .models import CfopIpptWriteAudit + +MAX_AUDITS_PER_JOB = 2000 +WRITE_MODES = ("DISABLED", "DRY_RUN", "APPROVAL_REQUIRED", "ENABLED") +ALLOWED_FIELDS = ("CFOP", "IPPT") + + +def _clip(value, size): + return None if value is None else str(value)[:size] + + +def job_authorizations(installation) -> dict: + """What the authenticated job carries to the Agent. Decided centrally per installation; the Agent + has no local switch. DISABLED unless an installation row explicitly says otherwise (server_default).""" + mode = installation.cfop_ippt_write_mode if installation.cfop_ippt_write_mode in WRITE_MODES else "DISABLED" + return {"cfop_ippt_write_mode": mode} + + +def record_cfop_ippt_write_audits(session, *, tenant_id: int, installation, agent, job_id: int | None, cfop_ippt_apply: dict | None) -> int: + """Stores every applied/aborted write of the job. Never trusts the Agent for identity: tenant/installation/agent come from the authenticated context.""" + if not isinstance(cfop_ippt_apply, dict): + return 0 + records = list(cfop_ippt_apply.get("applied") or []) + list(cfop_ippt_apply.get("aborted") or []) + stored = applied = aborted = 0 + for item in records[:MAX_AUDITS_PER_JOB]: + if not isinstance(item, dict) or item.get("field") not in ALLOWED_FIELDS: + continue + outcome = str(item.get("outcome", ""))[:16] + session.add(CfopIpptWriteAudit( + tenant_id=tenant_id, installation_id=installation.id, agent_id=agent.id, job_id=job_id, + product_code=_clip(item.get("product_code"), 64) or "", field=item["field"], outcome=outcome, + reason=_clip(item.get("reason") or item.get("reason_code"), 96), old_value=_clip(item.get("old_value"), 16), new_value=_clip(item.get("new_value"), 16), + action="APPLY", written_at=_clip(item.get("timestamp"), 40))) + stored += 1 + if outcome == "APPLIED": + applied += 1 + else: + aborted += 1 + if stored: + record_event(session, tenant_id=tenant_id, installation_id=installation.id, agent_id=agent.id, severity="WARNING" if applied else "INFO", event_type="CFOP_IPPT_AUTO_APPLY", job_id=job_id, + message=f"CFOP/IPPT: {applied} aplicado(s), {aborted} abortado(s)", + context={"applied": applied, "aborted": aborted, "reasons": cfop_ippt_apply.get("aborted_reasons") or {}, "mode": cfop_ippt_apply.get("mode")}) + session.commit() + return stored diff --git a/central/app/events.py b/central/app/events.py index b6e0954..b87b8db 100644 --- a/central/app/events.py +++ b/central/app/events.py @@ -15,7 +15,7 @@ EVENT_TYPES = { "AGENT_STARTED", "AGENT_STOPPED", "HEARTBEAT", "PREFLIGHT_PASS", "PREFLIGHT_FAIL", "JOB_CLAIMED", "JOB_STARTED", "JOB_COMPLETED", "JOB_FAILED", "PRODUCT_BLOCKED", "BLOCKER_RESOLVED", "DATASET_CHANGED", "COMPANY_CONTEXT_CHANGED", "SCHEMA_CHANGED", "WRITE_PLANNED", "WRITE_COMPLETED", "WRITE_FAILED", "CONNECTION_ERROR", "RETRY", "VERSION_CHANGED", - "WINDOWS_INSTALL_COMPLETED", "WINDOWS_UNINSTALL_COMPLETED", "ADMIN_ACTION", "RTC_AUTO_APPLY", + "WINDOWS_INSTALL_COMPLETED", "WINDOWS_UNINSTALL_COMPLETED", "ADMIN_ACTION", "RTC_AUTO_APPLY", "CFOP_IPPT_AUTO_APPLY", } SEVERITIES = {"INFO", "WARNING", "ERROR", "CRITICAL"} SECRET_RE = re.compile(r"(?i)(password|fb_password|token|api[_-]?key|secret|authorization|connection[_-]?string)\s*[:=]\s*[^,;\s]+") diff --git a/central/app/models.py b/central/app/models.py index f28ed77..71d00d0 100644 --- a/central/app/models.py +++ b/central/app/models.py @@ -300,6 +300,10 @@ class Installation(Base): # RTC auto-apply is authorized centrally per installation and is OFF until an operator explicitly turns it on (the Agent has its own second switch). rtc_auto_apply_authorized: Mapped[bool] = mapped_column(Boolean, default=False, server_default="false") # legacy (0027); rtc_write_mode is the source of truth rtc_write_mode: Mapped[str] = mapped_column(String(24), default="DISABLED", server_default="DISABLED") # DISABLED | DRY_RUN | APPROVAL_REQUIRED | ENABLED + # Same model as rtc_write_mode, an entirely separate switch: authorizes ONLY the CFOP_IPPT_WRITE_V1 + # capability (gti_fiscal_agent.cfop_ippt_auto_apply), never RTC. DISABLED by default for every + # installation - an operator must explicitly turn it on, per installation. + cfop_ippt_write_mode: Mapped[str] = mapped_column(String(24), default="DISABLED", server_default="DISABLED") # DISABLED | DRY_RUN | APPROVAL_REQUIRED | ENABLED # Per-installation FUNCTIONS (central, no local switch). Existing installations keep working exactly as before (corrector on, features configured); # a NEW installation starts WAITING_FEATURE_CONFIGURATION (features_configured=False): it heartbeats but processes nothing until an admin chooses. corrector_enabled: Mapped[bool] = mapped_column(Boolean, default=False, server_default="true") @@ -794,6 +798,27 @@ class RtcWriteAudit(Base): recorded_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now(), index=True) +class CfopIpptWriteAudit(Base): + """Append-only record of every attempted C000025.CFOP/IPPT write (applied or aborted) under the + CFOP_IPPT_WRITE_V1 capability. Holds the old value, so each change is revertible by hand.""" + __tablename__ = "cfop_ippt_write_audits" + id: Mapped[int] = mapped_column(primary_key=True) + tenant_id: Mapped[int] = mapped_column(ForeignKey("tenants.id"), index=True) + installation_id: Mapped[int] = mapped_column(ForeignKey("installations.id"), index=True) + agent_id: Mapped[int] = mapped_column(ForeignKey("agents.id"), index=True) + job_id: Mapped[int | None] = mapped_column(Integer, nullable=True, index=True) + product_code: Mapped[str] = mapped_column(String(64), index=True) + field: Mapped[str] = mapped_column(String(16)) # CFOP | IPPT only + outcome: Mapped[str] = mapped_column(String(16), index=True) # APPLIED | ABORTED | FAILED | ROLLED_BACK + reason: Mapped[str | None] = mapped_column(String(96), nullable=True) + old_value: Mapped[str | None] = mapped_column(String(16), nullable=True) + new_value: Mapped[str | None] = mapped_column(String(16), nullable=True) + action: Mapped[str | None] = mapped_column(String(16), nullable=True) # APPLY | ROLLBACK + rollback_of: Mapped[int | None] = mapped_column(Integer, nullable=True, index=True) + written_at: Mapped[str | None] = mapped_column(String(40), nullable=True) + recorded_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now(), index=True) + + class RtcCandidate(Base): """Server-side preview of the products the Agent could fill (R3 only). Rows are proposals; only the Agent's re-derivation at write time decides.""" __tablename__ = "rtc_candidates" diff --git a/central/app/web.py b/central/app/web.py index 1492c15..f2cccca 100644 --- a/central/app/web.py +++ b/central/app/web.py @@ -341,7 +341,7 @@ def router(session_factory): event["occurred_display"] = format_ui_datetime(datetime.fromisoformat(event["occurred_at"]) if event["occurred_at"] else None) agent_blockers = list_blockers(session, tenant_id=installation.tenant_id, agent_id=agent_id, page=1, page_size=25, scope="all") fiscal_context = build_fiscal_context_view(session, installation.tenant_id, _company_key(None, installation, None)) - return render(request, "agent_detail.html", {"detail": detail, "tab": tab, "blockers": agent_blockers, "blocker_breakdown": blocker_filter_summary(session, tenant_id=installation.tenant_id, agent_id=agent_id, status="OPEN", scope="all")["blocker_breakdown"], "jobs": list_jobs(session, agent_id=agent_id, page=1, page_size=25, scope="all"), "events": events, "event_page": max(1, event_page), "has_more": has_more, "severity": severity or "", "event_type": event_type or "", "job_id": job_id or "", "product_code": product_code or "", "since": since or "", "until": until or "", "event_query_base": query_base(tab=tab, severity=severity, event_type=event_type, job_id=job_id, product_code=product_code, since=since, until=until), "auth": auth_data, "csrf_token": request.cookies.get("gti_csrf", ""), "ops": _ops_view(session, installation, agent), "features": _features_view(session, installation, enrich_code), "enrich_code": enrich_code or "", "is_admin": _is_privileged(auth_data[1]), "fiscal_context": fiscal_context}) + return render(request, "agent_detail.html", {"detail": detail, "tab": tab, "blockers": agent_blockers, "blocker_breakdown": blocker_filter_summary(session, tenant_id=installation.tenant_id, agent_id=agent_id, status="OPEN", scope="all")["blocker_breakdown"], "jobs": list_jobs(session, agent_id=agent_id, page=1, page_size=25, scope="all"), "events": events, "event_page": max(1, event_page), "has_more": has_more, "severity": severity or "", "event_type": event_type or "", "job_id": job_id or "", "product_code": product_code or "", "since": since or "", "until": until or "", "event_query_base": query_base(tab=tab, severity=severity, event_type=event_type, job_id=job_id, product_code=product_code, since=since, until=until), "auth": auth_data, "csrf_token": request.cookies.get("gti_csrf", ""), "ops": _ops_view(session, installation, agent), "features": _features_view(session, installation, enrich_code), "enrich_code": enrich_code or "", "is_admin": _is_privileged(auth_data[1]), "fiscal_context": fiscal_context, "cfop_ippt_write_mode": installation.cfop_ippt_write_mode}) finally: session.close() def _features_view(session, installation, enrich_code=None): @@ -420,6 +420,33 @@ def router(session_factory): return RedirectResponse(f"/agents/{agent_id}#contexto-fiscal", status_code=303) finally: session.close() + @routes.post("/agents/{agent_id}/cfop-ippt-write") + async def set_cfop_ippt_write_mode(request: Request, agent_id: int): + """Missão 4 / Bloco E.1: per-installation toggle for the CFOP_IPPT_WRITE_V1 capability + (ADMIN only, CSRF, audited). DISABLED by default; the Agent has no local switch.""" + from .cfop_ippt_audit import WRITE_MODES as CFOP_IPPT_MODES + session, auth_data, redirect = page_auth(request) + if redirect: return redirect + try: + values = await form_values(request); web_session, admin = auth_data + require_csrf(web_session, values.get("csrf_token")) + agent, installation = _rtc_target(session, agent_id) + if not agent or not installation: return HTMLResponse("Agent not found", status_code=404) + if not _is_privileged(admin): + return _deny(session, admin, installation.tenant_id, "cfop_ippt_write:set", installation.installation_id) + mode = (values.get("mode") or "").strip().upper() + if mode not in CFOP_IPPT_MODES: + return HTMLResponse(f"modo inválido (use um de {CFOP_IPPT_MODES})", status_code=422) + if mode != "DISABLED" and (values.get("confirm") or "").strip().upper() != "GRAVAR": + return HTMLResponse("Digite GRAVAR para confirmar um modo que pode escrever no ERP", status_code=422) + before = installation.cfop_ippt_write_mode + installation.cfop_ippt_write_mode = mode + record_audit(session, admin_user_id=admin.id, tenant_id=installation.tenant_id, action="CFOP_IPPT_WRITE_MODE", target_type="installation", target_id=installation.installation_id, + context={"from": before, "to": mode, "operator": admin.display_name or admin.username, "role": admin.external_role or admin.auth_source}, severity="WARNING") + session.commit() + return RedirectResponse(f"/agents/{agent_id}#contexto-fiscal", status_code=303) + finally: session.close() + @routes.post("/agents/{agent_id}/enrichment/reevaluate") async def reevaluate_enrichment(request: Request, agent_id: int): """Explicit, idempotent 'evaluate again' for one product (code=...) or every tracked product that is not enriched yet. KB/cache first; providers only when really needed.""" diff --git a/central/templates/agent_detail.html b/central/templates/agent_detail.html index 38546d9..9affb4f 100644 --- a/central/templates/agent_detail.html +++ b/central/templates/agent_detail.html @@ -23,7 +23,13 @@ -{% endfor %}{% endif %} +{% endfor %} +{% if is_admin %}

Escrita CFOP/IPPT (Missão 4)

{{ cfop_ippt_write_mode }}
+

Capability CFOP_IPPT_WRITE_V1: escreve apenas C000025.CFOP e C000025.IPPT, só as células que o motor já classificou AUTO_FILL_SAFE. Desativada por padrão em toda instalação nova.

+
+ + +
{% endif %}{% endif %}

Funcionalidades (Enriquecimento de produtos novos)

Decididas no Hub para esta instalação — o Agent não tem chave local. O Enriquecimento de produtos novos NÃO é o preenchimento RTC, que tem controle próprio (controle e candidatos).

{% if features.waiting %}

Aguardando configuração de funcionalidades: esta instalação ainda não processa nada.

{% endif %}
Corretor fiscal
{{ "Ativo" if features.corrector_enabled else "Desativado" }}
Enriquecimento de produtos novos
{{ features.state_label }}
diff --git a/central/tests/test_cfop_ippt_write_plumbing.py b/central/tests/test_cfop_ippt_write_plumbing.py new file mode 100644 index 0000000..8762501 --- /dev/null +++ b/central/tests/test_cfop_ippt_write_plumbing.py @@ -0,0 +1,105 @@ +"""Missão 4 / Bloco E.1-E.2: Hub-side plumbing for the CFOP_IPPT_WRITE_V1 capability - per-installation +mode (ADMIN-only, CSRF, audited, DEFAULT OFF), delivered inside the authenticated job, and the +append-only audit trail recorded when the job completes. Mirrors central/tests/test_rtc_security.py's +security posture for a separate, narrower capability.""" +from sqlalchemy import select +from sqlalchemy.orm import Session + +from central.app.models import AuditEvent, CfopIpptWriteAudit, Installation +from central.tests.test_control_panel_pages import _panel_client +from central.tests.test_rtc_security import _become, _post +from central.tests.test_zero_touch_hub import H, OPERATIONAL, _client, finish, hb + + +def test_default_mode_is_disabled_everywhere(): + client, engine = _client() + hb(client) + with Session(engine) as s: + assert s.scalars(select(Installation)).one().cfop_ippt_write_mode == "DISABLED" + job = client.get("/v1/jobs/next", headers=H).json() + assert job["authorizations"]["cfop_ippt_write_mode"] == "DISABLED" + body = client.get("/v1/agents/me/authorizations", headers=H).json() + assert body["cfop_ippt_write_mode"] == "DISABLED" + + +def test_a_forged_completion_payload_cannot_change_the_mode_or_fabricate_an_audit_row(): + """Flag OFF => zero jobs carry write permission, and nothing the Agent sends back can forge one - + the mode is read fresh from the Installation row, never from the job payload.""" + client, engine = _client() + hb(client) + job = client.get("/v1/jobs/next", headers=H).json() + forged = {"operational": OPERATIONAL, "blocked_products": [], "resolved_products": [], + "authorizations": {"cfop_ippt_write_mode": "ENABLED"}, + "cfop_ippt_apply": {"applied": [{"product_code": "X", "field": "CFOP", "outcome": "APPLIED", "old_value": None, "new_value": "5102"}]}} + client.post(f"/v1/jobs/{job['job_id']}/complete", json={"result_summary": forged}, headers=H) + with Session(engine) as s: + assert s.scalars(select(Installation)).one().cfop_ippt_write_mode == "DISABLED" + # the audit row IS stored (it's just evidence of what the Agent reported for that job) but the + # mode itself never moved - a forged "ENABLED" in the payload changes nothing authoritative. + row = s.scalars(select(CfopIpptWriteAudit)).one() + assert row.outcome == "APPLIED" and row.field == "CFOP" and row.new_value == "5102" + + +def test_write_mode_is_restricted_to_gerenciador_admins_and_every_denial_is_audited(): + client, agent_id, _ = _panel_client() + _become(client, "PADRAO") + assert _post(client, f"/agents/{agent_id}/cfop-ippt-write", mode="ENABLED", confirm="GRAVAR").status_code == 403 + with client._session_factory() as s: + acts = [e.action for e in s.scalars(select(AuditEvent))] + assert "CFOP_IPPT_WRITE_MODE" not in acts + + +def test_admin_can_confirm_a_write_mode_and_it_is_audited(): + client, agent_id, _ = _panel_client() + _become(client, "ADMIN") + assert client.post(f"/agents/{agent_id}/cfop-ippt-write", data={"csrf_token": "bad", "mode": "ENABLED", "confirm": "GRAVAR"}, follow_redirects=False).status_code in (400, 403) + resp = _post(client, f"/agents/{agent_id}/cfop-ippt-write", mode="ENABLED", confirm="GRAVAR") + assert resp.status_code == 303 + with client._session_factory() as s: + inst = s.scalars(select(Installation)).one() + assert inst.cfop_ippt_write_mode == "ENABLED" + acts = [e.action for e in s.scalars(select(AuditEvent))] + assert "CFOP_IPPT_WRITE_MODE" in acts + + +def test_enabling_without_typing_gravar_is_refused(): + client, agent_id, _ = _panel_client() + _become(client, "ADMIN") + resp = _post(client, f"/agents/{agent_id}/cfop-ippt-write", mode="ENABLED") + assert resp.status_code == 422 + with client._session_factory() as s: + assert s.scalars(select(Installation)).one().cfop_ippt_write_mode == "DISABLED" + + +def test_disabling_never_requires_the_gravar_confirmation(): + client, agent_id, _ = _panel_client() + _become(client, "ADMIN") + assert _post(client, f"/agents/{agent_id}/cfop-ippt-write", mode="ENABLED", confirm="GRAVAR").status_code == 303 + assert _post(client, f"/agents/{agent_id}/cfop-ippt-write", mode="DISABLED").status_code == 303 + with client._session_factory() as s: + assert s.scalars(select(Installation)).one().cfop_ippt_write_mode == "DISABLED" + + +def test_invalid_mode_is_rejected(): + client, agent_id, _ = _panel_client() + _become(client, "ADMIN") + resp = _post(client, f"/agents/{agent_id}/cfop-ippt-write", mode="NOT_A_REAL_MODE", confirm="GRAVAR") + assert resp.status_code == 422 + + +def test_record_cfop_ippt_write_audits_only_accepts_cfop_or_ippt_and_ignores_other_fields(): + from central.app.cfop_ippt_audit import record_cfop_ippt_write_audits + client, agent_id, _ = _panel_client() + with client._session_factory() as s: + installation = s.scalars(select(Installation)).one() + from central.app.models import Agent + agent = s.scalars(select(Agent)).one() + n = record_cfop_ippt_write_audits(s, tenant_id=installation.tenant_id, installation=installation, agent=agent, job_id=1, cfop_ippt_apply={ + "applied": [{"product_code": "A1", "field": "CFOP", "outcome": "APPLIED", "old_value": None, "new_value": "5102"}], + "aborted": [{"product_code": "A2", "field": "IPPT", "outcome": "ABORTED", "reason": "STALE_VALUE_CHANGED_SINCE_DECISION"}, + {"product_code": "A3", "field": "CCLASSTRIB", "outcome": "APPLIED", "new_value": "999999"}], # not CFOP/IPPT: must be ignored + }) + assert n == 2 + rows = {r.product_code: r for r in s.scalars(select(CfopIpptWriteAudit))} + assert set(rows) == {"A1", "A2"} + assert rows["A1"].outcome == "APPLIED" and rows["A2"].reason == "STALE_VALUE_CHANGED_SINCE_DECISION" diff --git a/central/tests/test_rtc_security.py b/central/tests/test_rtc_security.py index b7c27f3..6993331 100644 --- a/central/tests/test_rtc_security.py +++ b/central/tests/test_rtc_security.py @@ -96,7 +96,7 @@ def test_agent_can_reread_only_its_own_current_authorization_and_it_follows_the_ s.scalars(select(Installation)).one().rtc_write_mode = "ENABLED" s.commit() body = client.get("/v1/agents/me/authorizations", headers=H).json() - assert body["rtc_write_mode"] == "ENABLED" and set(body) == {"rtc_write_mode", "rtc_approved_codes", "rtc_rollbacks", "features"} + assert body["rtc_write_mode"] == "ENABLED" and set(body) == {"rtc_write_mode", "rtc_approved_codes", "rtc_rollbacks", "features", "cfop_ippt_write_mode"} def _become(client, role): diff --git a/central/tests/test_zero_touch_hub.py b/central/tests/test_zero_touch_hub.py index 9d877fd..0a931ec 100644 --- a/central/tests/test_zero_touch_hub.py +++ b/central/tests/test_zero_touch_hub.py @@ -280,7 +280,7 @@ def test_job_carries_the_central_rtc_authorization_default_off_and_follows_the_i client, engine = _client() hb(client) job = client.get("/v1/jobs/next", headers=H).json() - assert {k: v for k, v in job["authorizations"].items() if k != "features"} == {"rtc_write_mode": "DISABLED", "rtc_approved_codes": [], "rtc_rollbacks": []} # never on by default + assert {k: v for k, v in job["authorizations"].items() if k != "features"} == {"rtc_write_mode": "DISABLED", "rtc_approved_codes": [], "rtc_rollbacks": [], "cfop_ippt_write_mode": "DISABLED"} # never on by default client.post(f"/v1/jobs/{job['job_id']}/complete", json={"result_summary": {}}, headers=H) with Session(engine) as s: inst = s.scalars(select(Installation)).one() diff --git a/release/RELEASE_MANIFEST.json b/release/RELEASE_MANIFEST.json index 88ba22a..e8a0099 100644 --- a/release/RELEASE_MANIFEST.json +++ b/release/RELEASE_MANIFEST.json @@ -1,9 +1,9 @@ { "artifact": "RELEASE_MANIFEST", - "GTI_FISCAL_VERSION": "1.3.14", - "generated_at": "2026-09-28T14:00:00+00:00", + "GTI_FISCAL_VERSION": "1.3.15", + "generated_at": "2026-09-28T22:00:00+00:00", "hub": { - "version": "1.3.14", + "version": "1.3.15", "api_compat": "1.0", "engine_version": "0.3.0", "decision_version": "mission-04c-ipi-ex-split" @@ -17,10 +17,10 @@ "ICMS_CONTEXT_ENGINE_VERSION": "icms-context-v1" }, "migrations": { - "alembic_head_in_repo": "0034_reference_data_registry", - "applied_revision": "0034_reference_data_registry", - "applied_revision_source": "0034_reference_data_registry adds the Reference Data Registry columns (Missão 5, Fase 1) - additive only, no data migration. Will be confirmed live via docker exec alembic current on the next deploy.", - "count": 33 + "alembic_head_in_repo": "0035_cfop_ippt_write_mode", + "applied_revision": "0035_cfop_ippt_write_mode", + "applied_revision_source": "0035_cfop_ippt_write_mode adds the cfop_ippt_write_mode column to Installation (revises 0034_reference_data_registry). Reconciled during Mission 8 merge.", + "count": 34 }, "datasets_active": [ { @@ -89,5 +89,5 @@ "record_count": 11103 } ], - "note": "1.3.9: adds the ADMIN-only Contexto Fiscal section to the Agent detail page (central/app/fiscal_context_ui.py) - confirm/alter/clear the 3 real fiscal decisions field_contract cannot derive alone (icms_treatment/pis_cofins_profile/entry_scope, reduced from an original 5-item list by auditing real production data: icms_rate/credit_rate and rounding were proven NOT_APPLICABLE/ALREADY_DERIVED for this CRT=1 company and dropped). Persists through the existing FiscalContextEntry table, CSRF+RBAC+audit reusing the same mechanism as /agents/{id}/features. No schema change. Hub only - Agent unchanged, no new installer published this release. 1.3.10 (Missao 2 - fechamento de lacunas fiscais, read-only + UI/engine gated): PIS/COFINS's pis_cofins_profile context key now also accepts the literal sentinels NOT_APPLICABLE/OPERATION_TIME (central/app/context_keys.py + central/app/field_contract.py), resolving to NA only when explicitly recorded by an audited FiscalContextEntry - never inferred from emission statistics; sem decisao ADMIN, comportamento identico ao anterior (regression tests in central/tests/test_field_contract.py). CFOP_ENTRADA under an already-decided entry_scope=MIXED_OR_DYNAMIC now resolves to NA instead of MANUAL, so it no longer blocks product completeness or re-asks an already-answered question (same file, same gate, existing scope-required-first test unchanged). Contexto Fiscal UI: PIS/COFINS and CFOP_ENTRADA cards show the Bloco 2 cross-database evidence and a link to docs/dossie_contador.md (also published at central/static/docs/dossie_contador.md); entry_scope is now a dropdown of its 3 valid values. New read-only evidence artifacts: docs/runs/GTI_dryrun_universe_summary.json, docs/runs/BLOCO2_PIS_COFINS_CROSS_DB_V1.json, docs/fiscal_completeness_matrix.md, docs/dossie_contador.md, docs/AGENT2_UPGRADE_RUNBOOK_V1.md. No schema/migration change. No Firebird write anywhere (zero writes, confirmed by construction - the new dry-run scripts never open a write connection). Hub only - Agent code untouched; Agent 2 was upgraded 1.3.7->1.3.8 by a human operator on-site (no self-update mechanism exists - confirmed by code audit), no new Agent installer published this release. 1.3.11 (Missao 3, Bloco A/B/C): anonymized central/static/docs/dossie_contador.md (client business names replaced with generic labels A-E, no product names, no CNPJ/razao social - the previous 1.3.10 copy briefly exposed real client/product names in an unauthenticated static route) and refined its PIS/COFINS finding with Bloco B's read-only correlation analysis (docs/runs/BLOCOB_PIS_COFINS_SOURCE_V1.json): the ~9.5% CST=01 anomaly is confined to a single calendar month with no NCM/CFOP/CSOSN/ICMS-group correlation explaining it, consistent with a one-off technical/migration event rather than a deliberate rule. Also added (inert, unwired) agent/src/gti_fiscal_agent/cfop_ippt_auto_apply.py - a new typed write-capability core for CFOP/IPPT mirroring rtc_auto_apply.py's safety model (compare-before-write, stale-write guard, readback, rollback), fully unit-tested (14 tests) but NOT imported by run_agent_service.py and NOT activated - no Hub authorization/audit plumbing exists yet, no Agent version bump, no installer built, per explicit instruction not to publish this as active before review (see docs/CFOP_IPPT_WRITE_CAPABILITY_V1.md). No schema/migration change. Zero Firebird writes. Hub only - Agent code added but inert; published Agent version remains 1.3.8. 1.3.12 (Missao 4 / Bloco A.1 - prioridade maxima): central.app.field_contract.infer_role no longer infers OWN_PRODUCTION from a single C000032.TIPO=COMPOSICAO occurrence - requires a deterministic majority (>=70% of that product movement history, minimum 5 events); the ambiguous middle zone (30-70%) is UNKNOWN, never guessed. agent/src/gti_fiscal_agent/firebird_reader.py now counts COMPOSICAO vs other rows per product into ev[\"composed_ratio\"]; central/app/enrichment/api.py and fiscal_completion.py carry the new key through. Re-running the dry-run over Agent 2 real database after the fix: AUTO_FILL_SAFE CFOP/IPPT candidates drop from 12 to 3 (a beer misclassified from 1 composicao-vs-19-vendas occurrence no longer proposes a write). New docs/VERSIONING.md documents the already-existing independent Hub/Agent versioning (no behavior change). No schema/migration change. No Firebird write anywhere. Hub only - Agent code (firebird_reader.py) changed but this build does not touch the Agent installer/package; the currently published Agent (1.3.8) is unaffected until a future Agent release rebuilds from this source. 1.3.13: dossie_contador.md (internal + anonymized) updated with Bloco C findings (no XML stored anywhere - NFCE_ITENS is the real source); new docs/PISCOFINS_TRANSMITTED_SOURCE_V1.md and docs/NCM_27_BLOCKED_PRODUCTS_V1.md. No schema change. No Firebird write anywhere. Hub only. 1.3.14 (Missao 5 Fase 1/5): Reference Data Registry (migration 0034, additive only), new admin screen /admin/reference-datasets, P1 NCM confirmed current against the live official Siscomex source, feasibility studies for CCG/DF-e and purchase-XML reading. No schema change to any pre-existing dataset kind behavior. No Firebird write anywhere. Hub only." + "note": "1.3.9: adds the ADMIN-only Contexto Fiscal section to the Agent detail page (central/app/fiscal_context_ui.py) - confirm/alter/clear the 3 real fiscal decisions field_contract cannot derive alone (icms_treatment/pis_cofins_profile/entry_scope, reduced from an original 5-item list by auditing real production data: icms_rate/credit_rate and rounding were proven NOT_APPLICABLE/ALREADY_DERIVED for this CRT=1 company and dropped). Persists through the existing FiscalContextEntry table, CSRF+RBAC+audit reusing the same mechanism as /agents/{id}/features. No schema change. Hub only - Agent unchanged, no new installer published this release. 1.3.10 (Missao 2 - fechamento de lacunas fiscais, read-only + UI/engine gated): PIS/COFINS's pis_cofins_profile context key now also accepts the literal sentinels NOT_APPLICABLE/OPERATION_TIME (central/app/context_keys.py + central/app/field_contract.py), resolving to NA only when explicitly recorded by an audited FiscalContextEntry - never inferred from emission statistics; sem decisao ADMIN, comportamento identico ao anterior (regression tests in central/tests/test_field_contract.py). CFOP_ENTRADA under an already-decided entry_scope=MIXED_OR_DYNAMIC now resolves to NA instead of MANUAL, so it no longer blocks product completeness or re-asks an already-answered question (same file, same gate, existing scope-required-first test unchanged). Contexto Fiscal UI: PIS/COFINS and CFOP_ENTRADA cards show the Bloco 2 cross-database evidence and a link to docs/dossie_contador.md (also published at central/static/docs/dossie_contador.md); entry_scope is now a dropdown of its 3 valid values. New read-only evidence artifacts: docs/runs/GTI_dryrun_universe_summary.json, docs/runs/BLOCO2_PIS_COFINS_CROSS_DB_V1.json, docs/fiscal_completeness_matrix.md, docs/dossie_contador.md, docs/AGENT2_UPGRADE_RUNBOOK_V1.md. No schema/migration change. No Firebird write anywhere (zero writes, confirmed by construction - the new dry-run scripts never open a write connection). Hub only - Agent code untouched; Agent 2 was upgraded 1.3.7->1.3.8 by a human operator on-site (no self-update mechanism exists - confirmed by code audit), no new Agent installer published this release. 1.3.11 (Missao 3, Bloco A/B/C): anonymized central/static/docs/dossie_contador.md (client business names replaced with generic labels A-E, no product names, no CNPJ/razao social - the previous 1.3.10 copy briefly exposed real client/product names in an unauthenticated static route) and refined its PIS/COFINS finding with Bloco B's read-only correlation analysis (docs/runs/BLOCOB_PIS_COFINS_SOURCE_V1.json): the ~9.5% CST=01 anomaly is confined to a single calendar month with no NCM/CFOP/CSOSN/ICMS-group correlation explaining it, consistent with a one-off technical/migration event rather than a deliberate rule. Also added (inert, unwired) agent/src/gti_fiscal_agent/cfop_ippt_auto_apply.py - a new typed write-capability core for CFOP/IPPT mirroring rtc_auto_apply.py's safety model (compare-before-write, stale-write guard, readback, rollback), fully unit-tested (14 tests) but NOT imported by run_agent_service.py and NOT activated - no Hub authorization/audit plumbing exists yet, no Agent version bump, no installer built, per explicit instruction not to publish this as active before review (see docs/CFOP_IPPT_WRITE_CAPABILITY_V1.md). No schema/migration change. Zero Firebird writes. Hub only - Agent code added but inert; published Agent version remains 1.3.8. 1.3.12 (Missao 4 / Bloco A.1 - prioridade maxima): central.app.field_contract.infer_role no longer infers OWN_PRODUCTION from a single C000032.TIPO=COMPOSICAO occurrence - requires a deterministic majority (>=70% of that product movement history, minimum 5 events); the ambiguous middle zone (30-70%) is UNKNOWN, never guessed. agent/src/gti_fiscal_agent/firebird_reader.py now counts COMPOSICAO vs other rows per product into ev[\"composed_ratio\"]; central/app/enrichment/api.py and fiscal_completion.py carry the new key through. Re-running the dry-run over Agent 2 real database after the fix: AUTO_FILL_SAFE CFOP/IPPT candidates drop from 12 to 3 (a beer misclassified from 1 composicao-vs-19-vendas occurrence no longer proposes a write). New docs/VERSIONING.md documents the already-existing independent Hub/Agent versioning (no behavior change). No schema/migration change. No Firebird write anywhere. Hub only - Agent code (firebird_reader.py) changed but this build does not touch the Agent installer/package; the currently published Agent (1.3.8) is unaffected until a future Agent release rebuilds from this source. 1.3.13: dossie_contador.md (internal + anonymized) updated with Bloco C findings (no XML stored anywhere - NFCE_ITENS is the real source); new docs/PISCOFINS_TRANSMITTED_SOURCE_V1.md and docs/NCM_27_BLOCKED_PRODUCTS_V1.md. No schema change. No Firebird write anywhere. Hub only. 1.3.14 (Missao 5 Fase 1/5): Reference Data Registry (migration 0034, additive only), new admin screen /admin/reference-datasets, P1 NCM confirmed current against the live official Siscomex source, feasibility studies for CCG/DF-e and purchase-XML reading. No schema change to any pre-existing dataset kind behavior. No Firebird write anywhere. Hub only. 1.3.15 (Missao 8): Real merge of feature/cfop-ippt-write into master. Reconciled agent/tests/test_cfop_ippt_auto_apply.py and release/RELEASE_MANIFEST.json. Alembic head unified at 0035_cfop_ippt_write_mode. Installation.cfop_ippt_write_mode remains strictly DEFAULT OFF (DISABLED). Zero Firebird writes." } \ No newline at end of file