feat: TECHNICAL_PLACEHOLDER (NCM 00000000) is replaceable by the product enrichment; real values preserved; Agent 1.3.4

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

central/app/enrichment/api.py

@@ -10,6 +10,7 @@ from sqlalchemy import select
 from ..blocked_products import upsert_blocker
 from ..config import settings
 from ..models import Installation, ProductEnrichment, ProductFieldAudit
+from .placeholders import is_placeholder
 from .providers import build_providers
 from .quota import ProviderGateway
 from .resolver import FIELDS, resolve
@@ -74,6 +75,9 @@ def router(principal, providers_factory=None):
                 if field in result["fields"]:
                     result["fields"][field] = {"status": "HUMAN_OVERRIDDEN", "value": None, "rule": None, "kb_status": None, "evidence": [], "reason": "valor alterado por uma pessoa; nunca mais preenchido automaticamente"}
             pending = result["state"] in PENDING_STATES
+            for name, decision in result["fields"].items():                    # explain in the audit trail that a technical placeholder was treated as "not classified yet"
+                if is_placeholder(name, current.get(name)):
+                    decision["technical_placeholder"] = str(current.get(name))
             row.gtin, row.description, row.latency_ms = result.get("gtin"), description, latency
             row.attempts = (row.attempts or 0) + 1
             row.status = "PENDING" if pending else "RESOLVED"

central/app/enrichment/placeholders.py

@@ -0,0 +1,27 @@
+"""TECHNICAL_PLACEHOLDER semantics for the product enrichment.
+
+The SGV refuses to save a product without an NCM, so operators type a technical filler to get past the form. That filler is NOT a human fiscal decision: it means "not classified yet".
+Explicit allowlist PER FIELD (never generalised: "0", "99999999", "SEM NCM" ... are NOT placeholders until the real SGV behaviour is confirmed). The Agent carries the same table
+(agent/src/gti_fiscal_agent/enrichment.py PLACEHOLDERS); a test asserts both are identical.
+"""
+import re
+
+PRODUCT_ENRICHMENT_PLACEHOLDERS = {"CLASSIFICACAO_FISCAL": frozenset({"00000000"})}
+
+
+def _norm(field: str, value) -> str:
+    text = "" if value is None else str(value).strip()
+    return re.sub(r"\D", "", text) if field == "CLASSIFICACAO_FISCAL" else text     # 0000.00.00 and 00000000 are the same NCM filler
+
+
+def is_placeholder(field: str, value) -> bool:
+    return _norm(field, value) in PRODUCT_ENRICHMENT_PLACEHOLDERS.get(field, frozenset())
+
+
+def is_blank(value) -> bool:
+    return not ("" if value is None else str(value)).strip()
+
+
+def is_writable_empty(field: str, value) -> bool:
+    """True when the field is empty OR holds the approved technical placeholder. A real value (even a wrong one) is never writable here."""
+    return is_blank(value) or is_placeholder(field, value)

central/app/enrichment/resolver.py

@@ -17,6 +17,7 @@ from ..datasets.service import DatasetService
 from ..models import Dataset, NcmEntry
 from . import kb, textnorm
 from .gtin import normalize_gtin
+from .placeholders import is_placeholder, is_writable_empty
 from .quota import ProviderGateway
 
 AUTO, SUGGEST, CONFLICT, INSUFFICIENT, ALREADY, NA = "AUTO_FILL_SAFE", "SUGGESTION_ONLY", "CONFLICT", "INSUFFICIENT_EVIDENCE", "ALREADY_SET", "NOT_APPLICABLE"
@@ -75,7 +76,7 @@ def _field(status, value=None, rule=None, kb_status=None, evidence=None, reason=
 
 
 def _empty(current: dict, name: str) -> bool:
-    return not str(current.get(name) or "").strip()
+    return is_writable_empty(name, current.get(name))
 
 
 def _fiscal_decisions(session, ncm_eval: dict, gtin_kb_verified: bool, current: dict, description: str, *, name_verified=False) -> dict:
@@ -92,7 +93,8 @@ def _fiscal_decisions(session, ncm_eval: dict, gtin_kb_verified: bool, current:
     else:
         out["CLASSIFICACAO_FISCAL"] = _field(INSUFFICIENT, reason="nenhuma fonte com NCM válido no dataset oficial")
     ncm = out["CLASSIFICACAO_FISCAL"]
-    ncm_for_engine = ncm["value"] if ncm["status"] == AUTO else (normalize_code(current.get("CLASSIFICACAO_FISCAL")) or None)
+    own = current.get("CLASSIFICACAO_FISCAL")
+    ncm_for_engine = ncm["value"] if ncm["status"] == AUTO else (None if is_placeholder("CLASSIFICACAO_FISCAL", own) else (normalize_code(own) or None))
     if not ncm_for_engine or (ncm["status"] == ALREADY and not ncm_is_official(session, ncm_for_engine)):
         for name in ("CEST", "IPI"):
             out[name] = _field(ALREADY if not _empty(current, name) else INSUFFICIENT, reason="depende de um NCM oficial válido")

central/app/version.py

@@ -4,5 +4,5 @@ kept identical to this) and what an AgentRelease row's `version` column holds
 once published — never a second, independent version number. Fiscal semantics
 are versioned separately (ENGINE_VERSION/DECISION_VERSION in classification.py,
 DECISION_ENGINE_VERSION etc.) and are NOT changed by a release bump."""
-GTI_FISCAL_VERSION = "1.3.3"
+GTI_FISCAL_VERSION = "1.3.4"
 API_COMPAT_MAJOR_MINOR = "1.0"  # an Agent 1.0.x/1.1.x talks to any Hub 1.0.x

central/pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "gti-fiscal-hub"
-version = "1.3.3"
+version = "1.3.4"
 requires-python = ">=3.10"
 dependencies = [
   "fastapi>=0.115,<1",

central/tests/test_health.py

@@ -17,7 +17,7 @@ def test_health_is_unavailable_when_database_is_down():
 
 def test_version_endpoint_reports_release_and_fiscal_versions():
     body = TestClient(create_app(database_probe=lambda: True)).get("/version").json()
-    assert body["version"] == "1.3.3" and body["api_compat"] == "1.0" and body["engine_version"] and body["decision_version"]
+    assert body["version"] == "1.3.4" and body["api_compat"] == "1.0" and body["engine_version"] and body["decision_version"]
 
 
 def test_production_refuses_the_dev_token(monkeypatch):

central/tests/test_product_enrichment_resolver.py

@@ -313,3 +313,15 @@ def test_backfill_respects_operational_reserve(env):
     outs = [gw.lookup(fake, gtin(f"78970000{i:04d}"), "BACKFILL", 20).state for i in range(10)]
     assert outs.count("SKIPPED") == 2 and fake.calls == 8                                    # 20% of 10 kept for new products
     assert gw.lookup(fake, gtin("789700009999"), "OPERATIONAL").state in ("FOUND", "NOT_FOUND")
+
+
+def test_technical_placeholder_ncm_is_treated_as_not_classified_but_real_values_are_not(env):
+    p = kb.upsert_product(env, "0" + G1, "COCA-COLA ORIGINAL LATA 350 ML")
+    for fam in ("PROVIDER:a", "PROVIDER:b"):
+        kb.add_evidence(env, p, "ncm", "22021000", "x", family=fam)
+    env.commit()
+    for value, expected in (("00000000", "AUTO_FILL_SAFE"), ("0000.00.00", "AUTO_FILL_SAFE"), ("22021000", "ALREADY_SET"), ("99999999", "ALREADY_SET"), ("0", "ALREADY_SET"), ("", "AUTO_FILL_SAFE")):
+        r = run(env, [], G1, "COCA-COLA ORIGINAL LATA 350 ML", current={"CLASSIFICACAO_FISCAL": value})
+        assert r["fields"]["CLASSIFICACAO_FISCAL"]["status"] == expected, value
+    r = run(env, [], G1, "COCA-COLA ORIGINAL LATA 350 ML", current={"CLASSIFICACAO_FISCAL": "00000000"})
+    assert r["fields"]["IPI"]["value"] == "0.0"                    # CEST/IPI follow the NCM that is about to replace the filler