diff --git a/agent/src/gti_fiscal_agent/unified_fiscal.py b/agent/src/gti_fiscal_agent/unified_fiscal.py
index 6835d37..67df3bb 100644
--- a/agent/src/gti_fiscal_agent/unified_fiscal.py
+++ b/agent/src/gti_fiscal_agent/unified_fiscal.py
@@ -8,9 +8,10 @@ from __future__ import annotations
import hashlib
import json
-from datetime import date
+from collections import Counter
+from datetime import date, datetime, timezone
-from central.app.unified_fiscal_coverage import PIS_COFINS_FIELDS, catalog_integrity, evaluate_product, rtc_enabled
+from central.app.unified_fiscal_coverage import PIS_COFINS_FIELDS, catalog_integrity, evaluate_product, legal_effective_context, rtc_enabled
PRODUCT_TABLE = "C000025"
RTC_CCLASSTRIB_TABLE = "RTC_CCLASSTRIB"
@@ -44,12 +45,17 @@ def load_rtc_context(cursor) -> tuple[dict, dict | None, dict]:
cursor.execute("SELECT TRIM(rf.RDB$RELATION_NAME), TRIM(rf.RDB$FIELD_NAME) FROM RDB$RELATION_FIELDS rf WHERE rf.RDB$RELATION_NAME IN ('C000025','RTC_CCLASSTRIB','RTC_CST','PARAMETROS')")
for table, column in cursor.fetchall():
schema.setdefault(table, set()).add(column)
- extras: dict = {"param_value": None, "physical_fk": False}
+ extras: dict = {"param_value": None, "param_rows": 0, "param_conflict": False, "physical_fk": False}
if "USAR_REFORMA_TRIBUTARIA" in schema.get("PARAMETROS", ()):
try:
- cursor.execute("SELECT FIRST 1 USAR_REFORMA_TRIBUTARIA FROM PARAMETROS")
- row = cursor.fetchone()
- extras["param_value"] = None if row is None or row[0] is None else row[0]
+ # ALL rows are read: the source of truth is what THIS installation's database says NOW (never a cached or Hub-side value).
+ cursor.execute("SELECT FIRST 20 USAR_REFORMA_TRIBUTARIA FROM PARAMETROS")
+ values = [None if r[0] is None else str(r[0]).strip().upper() for r in cursor.fetchall()]
+ extras["param_rows"] = len(values)
+ if len(set(values)) > 1:
+ extras["param_conflict"] = True # rows disagree: never guess "on" (RTC stays out of scope, and the conflict is reported)
+ elif values:
+ extras["param_value"] = values[0]
except Exception: # noqa: BLE001 - unreadable parameter => RTC stays off (not enabled by the ERP)
extras["param_value"] = None
try:
@@ -92,11 +98,26 @@ def assess_from_connection(conn, company: dict | None, *, today: date | None = N
today = today or date.today()
products = read_products(cursor, schema)
findings = {}
+ warn_by_type: Counter = Counter()
+ warn_products: set = set()
+ samples: list = []
+ applicability: Counter = Counter()
for p in products:
found = evaluate_product(p, company, rtc, today)
found["input_fingerprint"] = hashlib.sha256(json.dumps({k: (None if v is None else str(v).strip()) for k, v in p.items()} | {"rtc": enabled}, sort_keys=True).encode()).hexdigest()
- findings[str(p.get("CODIGO") or "")] = found
+ code = str(p.get("CODIGO") or "")
+ findings[code] = found
+ applicability[found["applicability"]["PIS_COFINS"]] += 1
+ for w in found.get("warnings", []):
+ warn_by_type[w["type"]] += 1
+ warn_products.add(code)
+ if len(samples) < 10:
+ samples.append({"product_code": code, **w})
integrity = catalog_integrity(catalog) if catalog else None
- return {"context": {"rtc_enabled": enabled, "rtc_supported": caps["HAS_RTC_RELATIONSHIP"], "rtc_catalog_valid": caps["RTC_CATALOG_VALID"], "rtc_param": extras["param_value"],
- "catalog_integrity": integrity, "crt": (company or {}).get("crt"), "capabilities": caps, "products_evaluated": len(products)},
+ return {"context": {"rtc_enabled": enabled, "erp_rtc_enabled": enabled, "rtc_supported": caps["HAS_RTC_RELATIONSHIP"], "rtc_catalog_valid": caps["RTC_CATALOG_VALID"], "rtc_param": extras["param_value"],
+ "rtc_source": "SIMULATION" if rtc_override is not None else "PARAMETROS.USAR_REFORMA_TRIBUTARIA", "rtc_param_rows": extras["param_rows"], "rtc_param_conflict": extras["param_conflict"],
+ "rtc_observed_at": datetime.now(timezone.utc).isoformat(timespec="seconds"), "legal_effective_context": legal_effective_context((company or {}).get("crt")),
+ "catalog_integrity": integrity, "crt": (company or {}).get("crt"), "capabilities": caps, "products_evaluated": len(products),
+ "pis_cofins_applicability": dict(applicability),
+ "non_canonical_warnings": {"products": len(warn_products), "by_type": dict(sorted(warn_by_type.items())), "samples": samples}},
"findings": findings}
diff --git a/agent/tests/test_unified_fiscal.py b/agent/tests/test_unified_fiscal.py
index d75bb2c..7908ea2 100644
--- a/agent/tests/test_unified_fiscal.py
+++ b/agent/tests/test_unified_fiscal.py
@@ -16,8 +16,8 @@ CLASSES = [("000001", "000", date(2025, 5, 5), None), ("200003", "200", date(202
class FakeCursor:
"""READ ONLY fake: answers only SELECTs and records every statement so tests can prove nothing else ran."""
- def __init__(self, param=0, schema=SCHEMA_ROWS, products=PRODUCTS, classes=CLASSES, cst=(("000",), ("200",))):
- self.param, self.schema, self.products, self.classes, self.cst, self.statements, self._rows = param, schema, products, classes, cst, [], []
+ def __init__(self, param=0, params=None, schema=SCHEMA_ROWS, products=PRODUCTS, classes=CLASSES, cst=(("000",), ("200",))):
+ self.param, self.params, self.schema, self.products, self.classes, self.cst, self.statements, self._rows = param, params, schema, products, classes, cst, [], []
def execute(self, sql, *args):
self.statements.append(sql)
@@ -26,7 +26,7 @@ class FakeCursor:
if "FROM RDB$RELATION_FIELDS" in text:
self._rows = list(self.schema)
elif "FROM PARAMETROS" in text:
- self._rows = [(self.param,)]
+ self._rows = [(v,) for v in (self.params if self.params is not None else [self.param])]
elif "RDB$RELATION_CONSTRAINTS" in text:
self._rows = [(1,)]
elif "FROM RTC_CCLASSTRIB" in text:
@@ -62,7 +62,7 @@ def test_rtc_off_in_the_erp_is_out_of_scope_and_never_blocks():
out = uf.assess_from_connection(conn, {"crt": "1"}, today=date(2026, 9, 24))
assert out["context"]["rtc_enabled"] is False and out["context"]["rtc_supported"] is True and out["context"]["rtc_catalog_valid"] is True
assert not [b for f in out["findings"].values() for b in f["blockers"] if b["category"] == "RTC"]
- assert kinds(out["findings"]["1"]) == [] and kinds(out["findings"]["2"]) == [] and kinds(out["findings"]["3"]) == ["COFINS_CST_INVALID", "PIS_CST_INVALID"]
+ assert kinds(out["findings"]["1"]) == [] and kinds(out["findings"]["2"]) == [] and kinds(out["findings"]["3"]) == [] # Simples + explicit "no special treatment": not applicable, typed "2" is only a warning
assert all(s.strip().upper().startswith("SELECT") for s in conn.cur.statements) # read-only: SELECT statements only
@@ -70,19 +70,19 @@ def test_rtc_on_in_the_erp_brings_cclasstrib_into_the_same_findings():
out = uf.assess_from_connection(FakeConn(param=1), {"crt": "1"}, today=date(2026, 9, 24))
assert out["context"]["rtc_enabled"] is True
assert kinds(out["findings"]["1"]) == ["CCLASSTRIB_MISSING"] and kinds(out["findings"]["2"]) == []
- assert kinds(out["findings"]["3"]) == ["CCLASSTRIB_MISSING", "COFINS_CST_INVALID", "PIS_CST_INVALID"] # one queue, one verdict per product
+ assert kinds(out["findings"]["3"]) == ["CCLASSTRIB_MISSING"] # one queue, one verdict per product
def test_simulation_flag_never_touches_the_erp_parameter():
conn = FakeConn(param=0)
on = uf.assess_from_connection(conn, {"crt": "1"}, rtc_override=True)
assert on["context"]["rtc_enabled"] is True and kinds(on["findings"]["1"]) == ["CCLASSTRIB_MISSING"]
- assert on["context"]["rtc_param"] == 0 and not any("UPDATE" in s.upper() for s in conn.cur.statements)
+ assert on["context"]["rtc_param"] == "0" and on["context"]["rtc_source"] == "SIMULATION" and not any("UPDATE" in s.upper() for s in conn.cur.statements)
def test_old_schema_without_rtc_columns_reports_unsupported_and_still_evaluates_pis_cofins():
old = [("C000025", c) for c in ("CODIGO", "COD_SAIDA_PIS", "PISCOFINS", "INCIDENCIA_PISCOFINS")]
- rows = [("1", None, "N", "NENHUMA"), ("2", "9", "N", "NENHUMA")]
+ rows = [("1", None, "N", "NENHUMA"), ("2", "049", "N", "NENHUMA")]
out = uf.assess_from_connection(FakeConn(schema=old, products=rows, classes=[], cst=()), {"crt": "3"})
assert out["context"]["rtc_supported"] is False and out["context"]["rtc_enabled"] is False
assert kinds(out["findings"]["1"]) == [] and kinds(out["findings"]["2"]) == ["PIS_CST_INVALID"]
@@ -108,3 +108,43 @@ def test_input_fingerprint_changes_with_the_fiscal_values_and_the_rtc_switch():
changed = [("1", None, None, "99", None, None, None, None, "N", "NENHUMA")] + PRODUCTS[1:]
c = uf.assess_from_connection(FakeConn(param=0, products=changed), {"crt": "1"})["findings"]["1"]["input_fingerprint"]
assert len({a, b, c}) == 3 and re.fullmatch(r"[0-9a-f]{64}", a)
+
+
+def test_rtc_state_is_read_from_this_database_now_with_provenance_and_timestamp():
+ out = uf.assess_from_connection(FakeConn(param=1), {"crt": "1"})
+ ctx = out["context"]
+ assert ctx["rtc_enabled"] is True and ctx["erp_rtc_enabled"] is True and ctx["rtc_source"] == "PARAMETROS.USAR_REFORMA_TRIBUTARIA" and ctx["rtc_param"] == "1" and ctx["rtc_param_rows"] == 1
+ assert ctx["rtc_param_conflict"] is False and ctx["rtc_observed_at"].endswith("+00:00")
+ assert ctx["legal_effective_context"]["informational_only"] is True and ctx["legal_effective_context"]["crt"] == "1"
+
+
+def test_toggle_between_syncs_is_followed_in_both_directions_without_any_cache():
+ conn = FakeConn(param=0)
+ first = uf.assess_from_connection(conn, {"crt": "1"})
+ assert first["context"]["rtc_enabled"] is False and not [b for f in first["findings"].values() for b in f["blockers"] if b["category"] == "RTC"]
+ conn.cur.param = 1 # the operator switches the reform on in the ERP between two syncs
+ second = uf.assess_from_connection(conn, {"crt": "1"})
+ assert second["context"]["rtc_enabled"] is True and kinds(second["findings"]["1"]) == ["CCLASSTRIB_MISSING"]
+ assert second["findings"]["1"]["input_fingerprint"] != first["findings"]["1"]["input_fingerprint"] # a new blocker fingerprint: the old (RTC off) state is not reused
+ conn.cur.param = 0 # and off again
+ third = uf.assess_from_connection(conn, {"crt": "1"})
+ assert third["context"]["rtc_enabled"] is False and not [b for f in third["findings"].values() for b in f["blockers"] if b["category"] == "RTC"]
+ assert third["findings"]["1"]["input_fingerprint"] == first["findings"]["1"]["input_fingerprint"]
+
+
+def test_conflicting_or_repeated_parameter_rows_never_guess_on():
+ same = uf.assess_from_connection(FakeConn(params=[1, 1]), {"crt": "1"})["context"]
+ assert same["rtc_enabled"] is True and same["rtc_param_rows"] == 2 and same["rtc_param_conflict"] is False
+ conflict = uf.assess_from_connection(FakeConn(params=[1, 0]), {"crt": "1"})
+ assert conflict["context"]["rtc_enabled"] is False and conflict["context"]["rtc_param_conflict"] is True
+ assert not [b for f in conflict["findings"].values() for b in f["blockers"] if b["category"] == "RTC"]
+ assert uf.assess_from_connection(FakeConn(params=[]), {"crt": "1"})["context"]["rtc_enabled"] is False
+
+
+def test_non_canonical_cst_warnings_are_reported_without_blocking():
+ rows = [("1", None, None, "4", None, None, "4", None, "N", "NENHUMA"), ("2", None, None, None, None, None, None, None, "N", "NENHUMA")]
+ out = uf.assess_from_connection(FakeConn(products=rows), {"crt": "3"})
+ assert kinds(out["findings"]["1"]) == [] and kinds(out["findings"]["2"]) == []
+ w = out["context"]["non_canonical_warnings"]
+ assert w["products"] == 1 and w["by_type"] == {"CST_COFINS_NON_CANONICAL_FORMAT": 1, "CST_PIS_NON_CANONICAL_FORMAT": 1} and w["samples"][0]["product_code"] == "1"
+ assert out["context"]["pis_cofins_applicability"] == {"OPTIONAL": 2}
diff --git a/agent/tests/test_zero_touch_agent.py b/agent/tests/test_zero_touch_agent.py
index 2f513cf..729ad62 100644
--- a/agent/tests/test_zero_touch_agent.py
+++ b/agent/tests/test_zero_touch_agent.py
@@ -185,3 +185,26 @@ def test_partial_reprocess_does_not_claim_a_full_fiscal_evaluation(env_setup, mo
hub, state = env_setup
result, _ = run_job("REPROCESS_PRODUCTS", [product("1")], state, monkeypatch)
assert result["fiscal_context"] is None and result["operational"] is None
+
+
+def test_rtc_toggle_between_two_syncs_is_followed_by_the_next_job(env_setup, monkeypatch):
+ """The Agent asks the database EVERY sync: 0 -> 1 makes the next job report RTC blockers, 1 -> 0 makes the following job report none (Hub then closes the old ones)."""
+ from gti_fiscal_agent import unified_fiscal
+ switch = {"on": False}
+
+ def assess(conn, company, **kw):
+ found = _findings(**{"1": [("CCLASSTRIB_MISSING", "RTC")]}) if switch["on"] else {}
+ return {"context": {"rtc_enabled": switch["on"], "rtc_supported": True, "rtc_catalog_valid": True, "rtc_source": "PARAMETROS.USAR_REFORMA_TRIBUTARIA", "rtc_observed_at": "2026-09-24T20:00:00+00:00"}, "findings": found}
+ monkeypatch.setattr(unified_fiscal, "assess_from_connection", assess)
+ hub, state = env_setup
+ products = [product("1"), product("2")]
+ off1, _ = run_job("INITIAL_SYNC", products, state, monkeypatch)
+ switch["on"] = True
+ on, _ = run_job("SYNC_INCREMENTAL", products, state, monkeypatch) # nothing changed in C000025: the delta cache must not hide the RTC change
+ switch["on"] = False
+ off2, _ = run_job("SYNC_INCREMENTAL", products, state, monkeypatch)
+ assert off1["blocked_products"] == [] and off1["fiscal_context"]["rtc_enabled"] is False
+ assert [b["field"] for b in on["blocked_products"]] == ["RTC"] and on["fiscal_context"]["rtc_enabled"] is True and on["operational"]["BLOCKED_FISCAL_DATA"] == 1
+ assert on["fiscal_context"]["rtc_observed_at"] and on["fiscal_context"]["rtc_source"] == "PARAMETROS.USAR_REFORMA_TRIBUTARIA"
+ assert off2["blocked_products"] == [] and off2["fiscal_context"]["rtc_enabled"] is False and off2["fiscal_context"]["evaluated_all"] is True
+ assert off2["operational"]["COMPLETE_PRODUCTS"] == 2
diff --git a/central/app/panel_presentation.py b/central/app/panel_presentation.py
index d2239ee..f63d3ec 100644
--- a/central/app/panel_presentation.py
+++ b/central/app/panel_presentation.py
@@ -162,9 +162,18 @@ def priority_label(value):
def rtc_status_view(summary):
- """Discreet Reforma Tributária status from the last completed job's `fiscal_context`. None when the Agent did not report it (older Agents)."""
+ """Discreet Reforma Tributaria status from the last completed job's `fiscal_context`, exactly as the Agent OBSERVED it in that installation's own database.
+ None when the Agent did not report it (older Agents)."""
context = (summary or {}).get("fiscal_context") if isinstance(summary, dict) else None
if not isinstance(context, dict) or "rtc_enabled" not in context:
return None
+ observed = None
+ if context.get("rtc_observed_at"):
+ try:
+ observed = format_ui_datetime(datetime.fromisoformat(str(context["rtc_observed_at"]).replace("Z", "+00:00")))
+ except ValueError:
+ observed = None
+ source = context.get("rtc_source") or "PARAMETROS.USAR_REFORMA_TRIBUTARIA"
return {"enabled": bool(context["rtc_enabled"]), "label": "Ativa" if context["rtc_enabled"] else "Desativada", "supported": bool(context.get("rtc_supported")),
- "catalog_valid": bool(context.get("rtc_catalog_valid"))}
+ "catalog_valid": bool(context.get("rtc_catalog_valid")), "observed_at": observed, "conflict": bool(context.get("rtc_param_conflict")),
+ "title": f"Lido no banco do cliente ({source})" + (f" em {observed}" if observed else "") + (" - linhas de PARAMETROS divergentes: RTC tratada como desativada" if context.get("rtc_param_conflict") else "")}
diff --git a/central/app/unified_fiscal_coverage.py b/central/app/unified_fiscal_coverage.py
index fea3c17..f113694 100644
--- a/central/app/unified_fiscal_coverage.py
+++ b/central/app/unified_fiscal_coverage.py
@@ -54,17 +54,25 @@ def incidence_kind(value) -> str:
return "UNKNOWN"
+def canonical_cst(value) -> str | None:
+ """The official 2-digit CST this text stands for, or None. Exactly two official digits, or ONE digit whose zero-padded form is an official code
+ ("4" -> "04"). Nothing else is ever normalised ("049", "000", "2x" stay invalid)."""
+ code = _text(value)
+ if code in CST_DOMAIN:
+ return code
+ if len(code) == 1 and code.isdigit() and code.zfill(2) in CST_DOMAIN:
+ return code.zfill(2)
+ return None
+
+
def validate_cst(value) -> str:
+ """EMPTY | VALID | VALID_AFTER_NORMALIZATION | INVALID."""
code = _text(value)
if not code:
return "EMPTY"
- return "VALID" if code in CST_DOMAIN else "INVALID"
-
-
-def cst_suggestion(value) -> str | None:
- """A short numeric code that becomes an official CST once zero-padded to two digits ('4' -> '04'). Only a hint for the operator; never applied."""
- code = _text(value)
- return code.zfill(2) if code.isdigit() and len(code) == 1 and code.zfill(2) in CST_DOMAIN else None
+ if code in CST_DOMAIN:
+ return "VALID"
+ return "VALID_AFTER_NORMALIZATION" if canonical_cst(code) else "INVALID"
def validate_rate(value) -> str:
@@ -82,56 +90,71 @@ def validate_rate(value) -> str:
def pis_cofins_applicability(product: dict, crt) -> dict:
"""{"state": NOT_APPLICABLE | REQUIRED | OPTIONAL | NEEDS_REVIEW, "basis": ...}.
- NOT_APPLICABLE the product carries no product-level PIS/COFINS treatment at all (flag N/blank, incidence NENHUMA/blank, all six fields blank).
- The tax is then settled elsewhere: inside the DAS (Simples family) or from the company default (regime normal).
- REQUIRED the product itself claims a special treatment (flag S, or incidence MONOFASICA/SUBSTITUICAO): the exit CST (and a rate when the CST needs one) must be valid.
- OPTIONAL no special claim, but some CST/rate was typed: it is not mandatory, yet whatever is there must be valid.
- NEEDS_REVIEW the incidence text is not one the SGV defines (NENHUMA / MONOFASICA / SUBST.TRIBUTARIA): applicability cannot be decided.
+ Decided ONLY from signals that are independent of the six CST/rate fields - an empty field is a data state, never proof of "not applicable":
+ * the company regime (CRT);
+ * the product's own declaration: PISCOFINS flag and INCIDENCIA_PISCOFINS;
+ * official rules (LC 123/2006 art. 13 IV/V and art. 18 par. 4-A I; Leis 9.718/98, 10.637/02, 10.833/03; SPED 4.3.x for the CST domain/semantics).
+
+ REQUIRED the product declares a special treatment (flag S, or incidence MONOFASICA / SUBSTITUICAO): the exit CST must be valid, whatever the fields hold.
+ NOT_APPLICABLE Simples family (CRT 1/2/4) with no special treatment declared: PIS/Cofins are collected inside the DAS (LC 123 art. 13), so the product carries no
+ obligation. The same verdict holds whether the fields are empty or typed.
+ OPTIONAL taxed regime (CRT 3 or other) AND the product explicitly declares "no special treatment" (flag N + incidence NENHUMA): the tax applies but the
+ company default covers it, so the product default is not mandatory; anything typed must still be valid.
+ NEEDS_REVIEW the declaration cannot be read: incidence text outside the SGV vocabulary, or a taxed regime with a blank flag/incidence.
"""
flag = _text(product.get("PISCOFINS")).upper()
- kind = incidence_kind(product.get("INCIDENCIA_PISCOFINS"))
- has_values = any(_text(product.get(c)) for c in PIS_COFINS_FIELDS)
+ raw_incidence = _text(product.get("INCIDENCIA_PISCOFINS"))
+ kind = incidence_kind(raw_incidence)
if kind == "UNKNOWN":
return {"state": "NEEDS_REVIEW", "basis": "INCIDENCE_NOT_DEFINED_BY_SGV"}
if flag == "S" or kind in ("MONOFASICA", "SUBSTITUICAO"):
return {"state": "REQUIRED", "basis": "PRODUCT_DECLARES_SPECIAL_TREATMENT"}
- if has_values:
- return {"state": "OPTIONAL", "basis": "VALUES_TYPED_WITHOUT_SPECIAL_CLAIM"}
- return {"state": "NOT_APPLICABLE", "basis": "SIMPLES_DAS" if _text(crt) in SIMPLES_FAMILY else "COMPANY_DEFAULT"}
+ if _text(crt) in SIMPLES_FAMILY:
+ return {"state": "NOT_APPLICABLE", "basis": "SIMPLES_DAS_NO_SPECIAL_TREATMENT"}
+ if flag == "N" and raw_incidence:
+ return {"state": "OPTIONAL", "basis": "TAXED_REGIME_COMPANY_DEFAULT_COVERS"}
+ return {"state": "NEEDS_REVIEW", "basis": "NO_DECLARATION_IN_TAXED_REGIME"}
def evaluate_pis_cofins(product: dict, crt) -> dict:
- """{"applicability": {...}, "blockers": [(type, reason)]}. Entry and exit are separate; a missing OPTIONAL value never blocks."""
+ """{"applicability": {...}, "blockers": [(type, reason)], "warnings": [(type, reason)]}. Entry and exit are separate; PIS and COFINS independent.
+ Blockers exist only where the field applies (REQUIRED / OPTIONAL). Non-blocking warnings report data-quality facts (non-canonical or ignored values)."""
app = pis_cofins_applicability(product, crt)
blockers: list[tuple[str, str]] = []
- if app["state"] in ("NOT_APPLICABLE", "NEEDS_REVIEW"):
- return {"applicability": app, "blockers": blockers}
+ warnings: list[tuple[str, str]] = []
kind = incidence_kind(product.get("INCIDENCIA_PISCOFINS"))
flag = _text(product.get("PISCOFINS")).upper()
+ applies = app["state"] in ("REQUIRED", "OPTIONAL")
for tax in ("PIS", "COFINS"):
entry, exit_, rate = product.get(f"COD_ENTRADA_{tax}"), product.get(f"COD_SAIDA_{tax}"), product.get(f"ALIQ_{tax}")
entry_state, exit_state, rate_state = validate_cst(entry), validate_cst(exit_), validate_rate(rate)
- if entry_state == "INVALID":
- blockers.append((f"{tax}_CST_INVALID", "ENTRY" + (f":UNPADDED_USE_{cst_suggestion(entry)}" if cst_suggestion(entry) else "")))
- if exit_state == "INVALID":
- blockers.append((f"{tax}_CST_INVALID", "EXIT" + (f":UNPADDED_USE_{cst_suggestion(exit_)}" if cst_suggestion(exit_) else "")))
+ for slot, value, state in (("ENTRY", entry, entry_state), ("EXIT", exit_, exit_state)):
+ if state == "VALID_AFTER_NORMALIZATION":
+ warnings.append((f"CST_{tax}_NON_CANONICAL_FORMAT", f"{slot}:{_text(value)}->{canonical_cst(value)}"))
+ elif state == "INVALID":
+ if applies:
+ blockers.append((f"{tax}_CST_INVALID", slot))
+ else:
+ warnings.append((f"CST_{tax}_INVALID_IGNORED", f"{slot}:{_text(value)}"))
+ if not applies:
+ continue
if app["state"] == "REQUIRED" and exit_state == "EMPTY":
blockers.append((f"{tax}_CST_MISSING", "EXIT"))
if rate_state == "INVALID":
blockers.append((f"{tax}_RATE_INVALID", "RATE"))
- if exit_state == "VALID":
- code = _text(exit_)
+ code = canonical_cst(exit_)
+ if code:
if code in RATE_REQUIRED_CST and rate_state == "EMPTY":
blockers.append((f"{tax}_RATE_MISSING", f"CST_{code}_NEEDS_RATE"))
if kind in INCIDENCE_CST and code not in INCIDENCE_CST[kind]:
blockers.append((f"{tax}_CST_INCIDENCE_MISMATCH", f"{kind}:CST_{code}"))
if kind == "NONE" and app["state"] == "REQUIRED" and code in RATE_NOT_APPLICABLE_CST - {"03"} and flag == "S":
blockers.append((f"{tax}_CST_INCIDENCE_MISMATCH", f"NENHUMA:CST_{code}"))
- if flag == "N" and kind != "NONE":
+ if applies and flag == "N" and kind != "NONE":
blockers.append(("PIS_COFINS_FLAG_INCONSISTENT", "FLAG_N_WITH_SPECIAL_INCIDENCE"))
- if flag == "S" and kind == "NONE" and not any(_text(product.get(c)) for c in PIS_COFINS_FIELDS):
+ if applies and flag == "S" and kind == "NONE" and not any(_text(product.get(c)) for c in PIS_COFINS_FIELDS):
blockers.append(("PIS_COFINS_FLAG_INCONSISTENT", "FLAG_S_WITHOUT_INCIDENCE_OR_CODES"))
- return {"applicability": app, "blockers": _dedupe(blockers)}
+ return {"applicability": app, "blockers": _dedupe(blockers), "warnings": _dedupe(warnings)}
def _dedupe(items):
@@ -201,11 +224,21 @@ def evaluate_product(product: dict, company: dict | None, rtc: dict | None, toda
pis = evaluate_pis_cofins(product, crt)
rtc_result = evaluate_rtc(product, rtc, today)
blockers = [{"type": kind, "category": blocker_category(kind), "reason": reason} for kind, reason in pis["blockers"] + rtc_result["blockers"]]
+ warnings = [{"type": kind, "reason": reason} for kind, reason in pis["warnings"]]
mandatory = []
if pis["applicability"]["state"] == "REQUIRED":
mandatory += ["COD_SAIDA_PIS", "COD_SAIDA_COFINS"]
if rtc_result["applicability"]["state"] == "REQUIRED":
mandatory.append("CCLASSTRIB")
return {"blockers": blockers, "applicability": {"PIS_COFINS": pis["applicability"]["state"], "RTC": rtc_result["applicability"]["state"]},
- "basis": {"PIS_COFINS": pis["applicability"]["basis"], "RTC": rtc_result["applicability"]["basis"]}, "mandatory_fields": mandatory,
+ "basis": {"PIS_COFINS": pis["applicability"]["basis"], "RTC": rtc_result["applicability"]["basis"]}, "mandatory_fields": mandatory, "warnings": warnings,
"needs_review": pis["applicability"]["state"] == "NEEDS_REVIEW"}
+
+
+def legal_effective_context(crt) -> dict:
+ """INFORMATIONAL ONLY (never a blocker, never part of completeness). The operational rule stays ERP_RTC_ENABLED; this records what the legislation
+ says about IBS/CBS in documents for the company regime so a difference in timing can be explained without inventing a rule."""
+ simples = _text(crt) in SIMPLES_FAMILY
+ return {"crt": _text(crt) or None, "informational_only": True,
+ "ibs_cbs_in_documents": "FROM_2027_FOR_SIMPLES_ORIENTATION_PENDING" if simples else ("MANDATORY_FROM_2026-01-01" if _text(crt) == "3" else "UNKNOWN_REGIME"),
+ "sources": ["NT 2025.002-RTC v1.51 (Portal NF-e, 04/08/2026): CRT 1/2/4 orientations in a future NT; IBS/CBS taxation of these taxpayers only from 2027", "LC 214/2025 art. 348"]}
diff --git a/central/templates/agent_detail.html b/central/templates/agent_detail.html
index 85a98c8..2af1a4c 100644
--- a/central/templates/agent_detail.html
+++ b/central/templates/agent_detail.html
@@ -8,7 +8,7 @@
Reprocessar tudo
{% if ops.enabled %}{% else %}{% endif %}
-
Processamento
{{ "Ativo" if ops.enabled else "Pausado" }} · sincronização automática a cada {{ ops.interval_minutes }} min
Última sincronização
{{ ops.last_sync or "ainda não sincronizado" }}{% if ops.last_changed is not none %} · {{ ops.last_changed }} produto(s) alterado(s){% endif %}
Sincronização agendada
{{ ops.next_sync or ("aguardando o primeiro sync" if ops.enabled else "pausado") }}