From 7e83d17533c7166d6b58207c98d061f38e4e732b Mon Sep 17 00:00:00 2001 From: gti Date: Sun, 27 Sep 2026 18:26:20 -0300 Subject: [PATCH] feat: wire field_contract into product enrichment; auto-discover CRT/CONTRIBUINTE_IPI at heartbeat; targeted per-product ERP evidence field_contract.resolve_product was only exercised by its own unit tests. /v1/products/resolve now calls it for every product whose NCM resolves, merging CFOP/CST/CSOSN/PIS-COFINS/CST_IPI/SITUACAO_TRIBUTARIA/IAT/IPPT decisions alongside CLASSIFICACAO_FISCAL/CEST/IPI - reusing the same engine, the same FiscalContextEntry table the blocker-resolution workflow already writes to, no duplicated rules. Closes three real transport/discovery gaps found while wiring it in: - Agent heartbeat never sent company_crt (env-var UF only) or company_contribuinte_ipi, so Installation.company_crt stayed permanently empty even though company_context.resolve() already proves both from C000004/EMISSOR for every coverage job. Resolved once, cached, retried only while unresolved. Migration 0032 adds Installation.company_contribuinte_ipi (NULL=unknown, never coerced to "N"). - The Agent's enrichment payload only ever reported 3 old fields (CLASSIFICACAO_FISCAL/CEST/IPI) as "current", so the Hub could never see the product's own CFOP/CST/CSOSN/... nor the classification columns (TIPO/PROD_COZINHA/PRODUTO_SERVICO/CODGRUPO) infer_role needs - separated into a read-only CURRENT_REPORT_FIELDS list, the write allowlist (PRODUCT_ENRICHMENT_WRITE_V1) untouched. - Added a targeted, product_code-scoped evidence query (firebird_reader.read_product_evidence: recipes/composed/purchases/stock, indexed WHERE, never a full-table scan) so st_withheld/product_role can be proven for a single ad-hoc resolve, not only during a full sync job. Verified read-only against the real database (D:\PROJETOS\GTISISTEMAS\Banco\GTI.FDB): CRT=1, CONTRIBUINTE_IPI=N, company_context.resolve() reaches HIGH confidence; infer_role correctly resolved RESALE/OWN_PRODUCTION for 5/8 sample real products with zero operator questions. 1121 tests passed (tests + central/tests + agent/tests), 0 failed. Version 1.3.7 (single GTI_FISCAL_VERSION shared by Hub and Agent per central/app/version.py's own design - never independently versioned). --- agent/pyproject.toml | 2 +- agent/scripts/run_agent_service.py | 31 ++- agent/src/gti_fiscal_agent/company_context.py | 10 +- agent/src/gti_fiscal_agent/enrichment.py | 24 ++- agent/src/gti_fiscal_agent/firebird_reader.py | 51 ++++- agent/src/gti_fiscal_agent/preflight.py | 2 +- agent/tests/test_company_context.py | 17 ++ agent/tests/test_enrichment.py | 27 ++- .../test_firebird_reader_product_evidence.py | 85 ++++++++ agent/tests/test_run_agent_service.py | 58 ++++++ .../alembic/versions/0032_contribuinte_ipi.py | 21 ++ central/app/api.py | 2 +- central/app/context_keys.py | 5 +- central/app/enrichment/api.py | 29 ++- central/app/enrichment/fiscal_completion.py | 143 ++++++++++++++ central/app/enrichment/resolver.py | 43 +++- central/app/installation_identity.py | 2 +- central/app/models.py | 1 + central/app/version.py | 2 +- central/pyproject.toml | 2 +- central/tests/test_fiscal_completion.py | 184 ++++++++++++++++++ central/tests/test_health.py | 2 +- central/tests/test_product_enrichment_api.py | 44 ++++- central/tests/test_zero_touch_hub.py | 10 + release/RELEASE_MANIFEST.json | 18 +- release/VERSION | 2 +- tests/test_release_candidate.py | 4 +- 27 files changed, 776 insertions(+), 45 deletions(-) create mode 100644 agent/tests/test_firebird_reader_product_evidence.py create mode 100644 central/alembic/versions/0032_contribuinte_ipi.py create mode 100644 central/app/enrichment/fiscal_completion.py create mode 100644 central/tests/test_fiscal_completion.py diff --git a/agent/pyproject.toml b/agent/pyproject.toml index 6d51e32..c414d28 100644 --- a/agent/pyproject.toml +++ b/agent/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "gti-fiscal-agent" -version = "1.3.6" +version = "1.3.7" requires-python = ">=3.10" dependencies = ["httpx>=0.27,<1"] diff --git a/agent/scripts/run_agent_service.py b/agent/scripts/run_agent_service.py index c79a338..965ac14 100644 --- a/agent/scripts/run_agent_service.py +++ b/agent/scripts/run_agent_service.py @@ -107,7 +107,7 @@ class Config: self.poll_interval_seconds = float(env.get("POLL_INTERVAL_SECONDS", "5")) # RTC auto-apply is a SEPARATE switch (default OFF): FIREBIRD_WRITES alone never lets the Agent fill cClassTrib. self.rtc_auto_apply_max_per_job = int(env.get("RTC_AUTO_APPLY_MAX_PER_JOB", str(_rtc_apply.DEFAULT_MAX_PER_JOB))) - self.agent_version = env.get("AGENT_VERSION", "1.3.6") + self.agent_version = env.get("AGENT_VERSION", "1.3.7") self.schema_fingerprint = env.get("SCHEMA_FINGERPRINT", REFERENCE_SCHEMA_V1) self.state_db = env.get("AGENT_STATE_DB", str(AGENT_ROOT / "agent_state.sqlite")) # Pilot write authorization: both must be explicitly set by the installer for @@ -349,7 +349,7 @@ def execute_job(job, config, executor, state, env): "fiscal_context": fiscal_context, "rtc_auto_apply": rtc_apply, "fiscal_mode": getattr(config, "fiscal_mode", None), - "release_version": env.get("AGENT_VERSION") or "1.3.6", + "release_version": env.get("AGENT_VERSION") or "1.3.7", "products_total": len(products), "products_changed": len(to_process), "products_processed": summary["processed"], @@ -410,6 +410,24 @@ def resolve_installation_id(config, env, state, client): return state.installation_id(str(installation_id)) +def _company_context_for_heartbeat(env, cached): + """Resolves company UF/CRT via the SAME deterministic, fail-closed resolver already used for coverage/RTC (`company_context.resolve`, + called through `preflight_company_context`) - never a second, separate guess. Previously the heartbeat only ever sent `COMPANY_UF` from an + operator-set env var and never sent CRT at all, so `Installation.company_crt` stayed permanently empty in the Hub even though the Agent + already proves CRT from C000004/EMISSOR for every coverage job. Resolves once and caches: never re-queries Firebird once solid, retries only + while still unresolved (e.g. the database was not yet reachable when the service started).""" + if cached is not None: + return cached, None + try: + from gti_fiscal_agent.company_context import preflight_company_context + result = preflight_company_context(env) + except Exception as error: + return None, f"company context not resolved yet: {error}" + if result.get("status") != "PASS": + return None, f"company context not resolved yet: {result.get('classification')} ({'; '.join(result.get('reasons', []))})" + return result, None + + def _start_enrichment(config, env, client, stopper): """New-product enrichment worker (own thread, own SQLite file). It gates ITSELF on the Hub's per-installation features: DISABLED / not configured = zero Firebird activity.""" if not hasattr(client, "resolve_product"): @@ -439,17 +457,22 @@ def run(config, env, *, max_iterations=None, stopper=None): was_down = False iterations = 0 last_job_id = None + company_ctx = None while not stopper.stop: if max_iterations is not None and iterations >= max_iterations: break iterations += 1 try: + company_ctx, reason = _company_context_for_heartbeat(env, company_ctx) + if reason: + logger.debug(reason) heartbeat_payload = { "installation_id": installation_id, "status": "IDLE", "schema_fingerprint": config.schema_fingerprint, "agent_version": config.agent_version, "last_job_id": last_job_id, "display_name": env.get("AGENT_DISPLAY_NAME", ""), "customer_code": env.get("CUSTOMER_CODE"), "hostname": env.get("AGENT_HOSTNAME") or os.environ.get("COMPUTERNAME"), - "windows_version": env.get("WINDOWS_VERSION"), "company_code": config.company_id, "company_uf": config.company_uf, + "windows_version": env.get("WINDOWS_VERSION"), "company_code": config.company_id, "company_uf": (company_ctx or {}).get("company_uf") or config.company_uf, + "company_crt": (company_ctx or {}).get("company_crt"), "company_contribuinte_ipi": (company_ctx or {}).get("company_contribuinte_ipi"), "database_path": config.database_path, "capabilities": {"heartbeat": True, "jobs": True, "local_logs": True, "RTC_CCLASSTRIB_WRITE_V1": True, "PRODUCT_ENRICHMENT_WRITE_V1": True}, "fiscal_engine_version": env.get("FISCAL_ENGINE_VERSION", config.agent_version), "dataset_versions": {}, } @@ -520,7 +543,7 @@ def run(config, env, *, max_iterations=None, stopper=None): def enforce_startup_preflight(env): """Full read-only preflight at startup (Hub, auth, Firebird, capabilities, company context, guards). Fail closed.""" from gti_fiscal_agent.preflight import run_preflight - report = run_preflight(env, agent_version=env.get("AGENT_VERSION") or "1.3.6") + report = run_preflight(env, agent_version=env.get("AGENT_VERSION") or "1.3.7") logger.info("startup preflight: %s", json.dumps({k: report[k] for k in ("PREFLIGHT", "blockers", "fiscal_mode", "effective_firebird_writes")}, ensure_ascii=False)) print(json.dumps({"PREFLIGHT": report["PREFLIGHT"], "blockers": report["blockers"]}, ensure_ascii=False)) if report["PREFLIGHT"] != "PASS": diff --git a/agent/src/gti_fiscal_agent/company_context.py b/agent/src/gti_fiscal_agent/company_context.py index f32798b..65f510b 100644 --- a/agent/src/gti_fiscal_agent/company_context.py +++ b/agent/src/gti_fiscal_agent/company_context.py @@ -59,7 +59,7 @@ def resolve(evidence, config=None): "nfce_keys": [{"cuf", "cnpj", "n"}], "nf_by_filial": {code: n}}. Pure function.""" cfg = config or {} out = {"status": "FAIL", "classification": "UNRESOLVED", "company_code": None, "company_cnpj": None, "company_cnpj_fingerprint": None, "company_uf": None, "company_crt": None, - "source_of_truth": {}, "confidence": "NONE", "evidence": [], "warnings": [], "reasons": [], "configured": False} + "company_contribuinte_ipi": None, "source_of_truth": {}, "confidence": "NONE", "evidence": [], "warnings": [], "reasons": [], "configured": False} def fail(cls, reason): out["classification"] = cls; out["reasons"].append(reason); return out @@ -93,6 +93,10 @@ def resolve(evidence, config=None): out["evidence"].append(f"company selected by {basis}") out.update(company_code=selected["code"], company_cnpj=mask_cnpj(selected["cnpj"]), company_cnpj_fingerprint=cnpj_fingerprint(selected["cnpj"])) out["source_of_truth"]["company"] = basis + contribuinte_ipi = _s(selected.get("contribuinte_ipi")) + if contribuinte_ipi and contribuinte_ipi.upper() in ("S", "N"): # same row already proven to be THE company; no extra query, no extra trust needed + out["company_contribuinte_ipi"] = contribuinte_ipi.upper() + out["source_of_truth"]["contribuinte_ipi"] = "C000004" # ---- emitter (EMISSOR) must describe the same company -------------------------------------- emissor = None @@ -175,7 +179,7 @@ def collect_evidence(cursor): cols.setdefault(t, set()).add(f) ev = {"c000004": [], "emissor": [], "nfce_keys": [], "nf_by_filial": {}} c4 = cols.get("C000004", set()) - want = [("CODIGO", "code"), ("CNPJ", "cnpj"), ("IE", "ie"), ("UF", "uf"), ("CRT", "crt"), ("OPTANTE_SIMPLES", "optante"), ("FILIAL", "name")] + want = [("CODIGO", "code"), ("CNPJ", "cnpj"), ("IE", "ie"), ("UF", "uf"), ("CRT", "crt"), ("OPTANTE_SIMPLES", "optante"), ("FILIAL", "name"), ("CONTRIBUINTE_IPI", "contribuinte_ipi")] use = [(c, k) for c, k in want if c in c4] if use: cursor.execute(f"SELECT {','.join(c for c, _ in use)} FROM C000004") @@ -214,5 +218,5 @@ def preflight_company_context(env=None): def preflight_report(result): - keys = ("company_code", "company_cnpj", "company_uf", "company_crt", "source_of_truth", "confidence", "classification", "warnings", "reasons") + keys = ("company_code", "company_cnpj", "company_uf", "company_crt", "company_contribuinte_ipi", "source_of_truth", "confidence", "classification", "warnings", "reasons") return {"COMPANY_CONTEXT": "PASS" if result["status"] == "PASS" else "FAIL", **{k: result.get(k) for k in keys}} diff --git a/agent/src/gti_fiscal_agent/enrichment.py b/agent/src/gti_fiscal_agent/enrichment.py index 397dabc..93a9b38 100644 --- a/agent/src/gti_fiscal_agent/enrichment.py +++ b/agent/src/gti_fiscal_agent/enrichment.py @@ -25,7 +25,17 @@ ALLOWED_FIELDS = ("CLASSIFICACAO_FISCAL", "CEST", "IPI") # TECHNICAL_PLACEHOLDER: the SGV refuses a product without NCM, so operators type a filler. It means "not classified yet" (not a human decision) and may be replaced. # Explicit per-field allowlist, identical to central/app/enrichment/placeholders.py (parity test). Nothing else is ever overwritten. PLACEHOLDERS = {"CLASSIFICACAO_FISCAL": frozenset({"00000000"})} # explicit allowlist; anything else raises before touching SQL -READ_FIELDS = ("CODIGO", "PRODUTO", "GTIN", "CODBARRA") + ALLOWED_FIELDS # the SGV operator types the EAN in CODBARRA; GTIN (fiscal) is mostly empty +# READ-ONLY reporting to the Hub for field_contract.resolve_product's post-NCM fiscal-completion decisions (CFOP/CST/CSOSN/PIS-COFINS/...). +# This is NOT a write allowlist - PRODUCT_ENRICHMENT_WRITE_V1 still only ever writes ALLOWED_FIELDS above. Sending the CURRENT value of these +# fields (and the classification columns below) just lets the Hub's field_contract engine see what's already filled and never guess. +FISCAL_COMPLETION_FIELDS = ("SITUACAO_TRIBUTARIA", "CFOP", "CFOP_ENTRADA", "CST", "CST_ENTRADA", "CSOSN", "ALIQUOTA", "CST_IPI", + "COD_ENTRADA_PIS", "COD_SAIDA_PIS", "ALIQ_PIS", "COD_ENTRADA_COFINS", "COD_SAIDA_COFINS", "ALIQ_COFINS", + "INCIDENCIA_PISCOFINS", "PISCOFINS", "IAT", "IPPT") +# Classification metadata field_contract.infer_role reads to auto-discover product_role (RESALE/OWN_PRODUCTION/INPUT/SERVICE) from real ERP +# facts instead of asking the operator. Read-only here too. +CLASSIFICATION_CONTEXT_FIELDS = ("CODGRUPO", "TIPO", "PROD_COZINHA", "PRODUTO_SERVICO") +CURRENT_REPORT_FIELDS = ALLOWED_FIELDS + FISCAL_COMPLETION_FIELDS + CLASSIFICATION_CONTEXT_FIELDS +READ_FIELDS = ("CODIGO", "PRODUTO", "GTIN", "CODBARRA") + CURRENT_REPORT_FIELDS # the SGV operator types the EAN in CODBARRA; GTIN (fiscal) is mostly empty POLL_SECONDS = 2.0 SAFETY_DIFF_SECONDS = 60.0 OVERRIDE_CHECK_SECONDS = 20.0 @@ -327,9 +337,14 @@ class EnrichmentWorker: conn = self._open(self.read_conn) try: rows = read_rows(conn, [code]) + row = rows.get(code) + if row is None or mode not in ("DRY_RUN", "ENABLED"): + ev = None + else: + from .firebird_reader import read_product_evidence # targeted, indexed by product_code - never a full-table scan (see docstring) + ev = read_product_evidence(conn, code) finally: conn.close() - row = rows.get(code) if row is None: self.store.update(code, state="GONE") return @@ -338,9 +353,10 @@ class EnrichmentWorker: self.store.update(code, state="SKIPPED_FEATURE", snapshot=snapshot) return attempts = int(item["attempts"] or 0) + 1 - current = {f: row[f] for f in ALLOWED_FIELDS} + current = {f: row[f] for f in CURRENT_REPORT_FIELDS} try: - result = self.hub.resolve_product({"product_code": code, "gtin": pick_gtin(row), "description": _text(row["PRODUTO"]), "current": {k: (None if v is None else str(v)) for k, v in current.items()}}) + result = self.hub.resolve_product({"product_code": code, "gtin": pick_gtin(row), "description": _text(row["PRODUTO"]), "ev": {k: (sorted(v) if isinstance(v, set) else v) for k, v in ev.items()}, + "current": {k: (None if v is None else str(v)) for k, v in current.items()}}) except Exception as error: # noqa: BLE001 - Hub offline / slow: the product stays pending, the SGV is unaffected delay = RETRY_BACKOFF[min(attempts - 1, len(RETRY_BACKOFF) - 1)] self.store.update(code, state="RETRY", attempts=attempts, next_attempt_at=self.clock() + delay, last_state=f"HUB_ERROR:{type(error).__name__}") diff --git a/agent/src/gti_fiscal_agent/firebird_reader.py b/agent/src/gti_fiscal_agent/firebird_reader.py index a8d5376..1edc7df 100644 --- a/agent/src/gti_fiscal_agent/firebird_reader.py +++ b/agent/src/gti_fiscal_agent/firebird_reader.py @@ -75,7 +75,56 @@ def read_company_context(env=None): result = preflight_company_context(env or os.environ) if result["status"] != "PASS": raise CompanyContextError(result) - return {"uf": result["company_uf"], "crt": result["company_crt"], "company_code": result["company_code"], "source_of_truth": result["source_of_truth"], "confidence": result["confidence"]} + return {"uf": result["company_uf"], "crt": result["company_crt"], "company_code": result["company_code"], "contribuinte_ipi": result.get("company_contribuinte_ipi"), + "source_of_truth": result["source_of_truth"], "confidence": result["confidence"]} + + +def read_product_evidence(conn, code): + """Targeted, single-product evidence for an ad-hoc /v1/products/resolve call (new-product enrichment) - the same facts `read_evidence` + already gathers for a full sync job (recipes/composed/purchases/stock), but scoped to ONE product_code via an indexed WHERE, never a + full-table scan. Measured on a real production-equivalent database: read_evidence's full scan took ~40ms total for the whole catalog, + so this is not a correctness-driven optimization - it is what the mission explicitly asked for (never transport/scan the whole dataset + for one ad-hoc product). Every table/column is optional: a missing one just means less evidence (capability-based), matching + `read_evidence`. Takes an already-open connection (the SAME one the caller already opened for `read_rows`) - never opens its own, + so it works with the EnrichmentWorker's injected `read_conn` (including test fakes; a non-Firebird connection has no RDB$RELATION_FIELDS, + so the capability probe below degrades to "no evidence" rather than raising).""" + from collections import defaultdict + cur = conn.cursor() + cols = defaultdict(set) + try: + cur.execute("SELECT TRIM(RDB$RELATION_NAME), TRIM(RDB$FIELD_NAME) FROM RDB$RELATION_FIELDS WHERE RDB$RELATION_NAME IN ('C000032','C000088','C000100','RECEITA_INGREDIENTES')") + for t, f in cur.fetchall(): + cols[t].add(f) + except Exception: + return {"recipes": set(), "composed": set(), "purchases": {}, "stock_entry": {}, "stock_in": {}} + has = lambda t, *c: t in cols and all(x in cols[t] for x in c) # noqa: E731 + s = lambda v: None if v is None else str(v).strip() # noqa: E731 + ev = {"recipes": set(), "composed": set(), "purchases": {}, "stock_entry": {}, "stock_in": {}} + code_padded = str(code).zfill(6) + if has("RECEITA_INGREDIENTES", "CODRECEITA"): + cur.execute("SELECT FIRST 1 CODRECEITA FROM RECEITA_INGREDIENTES WHERE CODRECEITA = ?", (code_padded,)) + if cur.fetchone(): + ev["recipes"] = {code_padded} + if has("C000032", "CODPRODUTO", "TIPO"): + cur.execute("SELECT FIRST 1 CODPRODUTO FROM C000032 WHERE CODPRODUTO = ? AND TIPO = 'COMPOSICAO'", (code,)) + if cur.fetchone(): + ev["composed"] = {s(code)} + if has("C000088", "CODPRODUTO", "ICMS_RETIDO", "ICMS_VALORRETIDO", "SUB_VALOR"): + cur.execute("SELECT ICMS_RETIDO, ICMS_VALORRETIDO, SUB_VALOR FROM C000088 WHERE CODPRODUTO = ?", (code,)) + rows = [{"icms_retido": s(ret), "valor_retido": vr, "sub_valor": sv} for ret, vr, sv in cur.fetchall()] + if rows: + ev["purchases"][s(code)] = rows + if has("C000100", "CODPRODUTO", "ULTIMA_ENTRADA", "NOTA_FISCAL"): + cur.execute("SELECT FIRST 1 ULTIMA_ENTRADA, NOTA_FISCAL FROM C000100 WHERE CODPRODUTO = ?", (code,)) + row = cur.fetchone() + if row: + ev["stock_entry"][s(code)] = row[0] is not None or s(row[1]) not in (None, "", "000000") + if has("C000100", "CODPRODUTO", "ENTRADAS"): + cur.execute("SELECT FIRST 1 ENTRADAS FROM C000100 WHERE CODPRODUTO = ?", (code,)) + row = cur.fetchone() + if row: + ev["stock_in"][s(code)] = bool(row[0] and float(row[0]) > 0) + return ev def read_evidence(env=None, selected_company_code=None): diff --git a/agent/src/gti_fiscal_agent/preflight.py b/agent/src/gti_fiscal_agent/preflight.py index 2d208e8..c6ed10e 100644 --- a/agent/src/gti_fiscal_agent/preflight.py +++ b/agent/src/gti_fiscal_agent/preflight.py @@ -9,7 +9,7 @@ import os import sqlite3 from pathlib import Path -AGENT_VERSION_DEFAULT = "1.3.6" +AGENT_VERSION_DEFAULT = "1.3.7" REQUIRED_ENV = ("GTI_HUB_URL", "AGENT_TOKEN", "FIREBIRD_DATABASE_PATH", "FB_USER", "FB_PASSWORD") REQUIRED_CAPABILITIES = ("supports_ncm", "supports_ipi", "supports_cest") FISCAL_MODES = ("DRY_RUN", "AUTO_APPLY") diff --git a/agent/tests/test_company_context.py b/agent/tests/test_company_context.py index 4b0f31a..821b7f5 100644 --- a/agent/tests/test_company_context.py +++ b/agent/tests/test_company_context.py @@ -83,6 +83,23 @@ def test_optante_inconsistency_is_a_warning_not_a_guess(): assert r["status"] == "PASS" and any("OPTANTE_SIMPLES" in w for w in r["warnings"]) +def test_contribuinte_ipi_is_auto_discovered_from_the_same_selected_c000004_row(): + """C000004.CONTRIBUINTE_IPI (real column, same row already proven to be THE company by CRT/UF resolution) closes the fiscal-completion gap + where CST_IPI could not be resolved without asking an operator for a fact the ERP already proves.""" + row = co("000001", CN_A) + row["contribuinte_ipi"] = "N" + r = resolve(ev([row], [emis()], DOC_A)) + assert r["company_contribuinte_ipi"] == "N" and r["source_of_truth"]["contribuinte_ipi"] == "C000004" + + +def test_contribuinte_ipi_absent_or_invalid_stays_unresolved_never_guessed(): + r = resolve(ev([co("000001", CN_A)], [emis()], DOC_A)) # co() never sets contribuinte_ipi + assert r["company_contribuinte_ipi"] is None + row = co("000002", CN_B); row["contribuinte_ipi"] = "X" # garbage value: never trusted + r2 = resolve(ev([row], [emis(cnpj=CN_B)], [{"cuf": "29", "cnpj": CN_B, "n": 1}])) + assert r2["company_contribuinte_ipi"] is None + + def test_full_cnpj_is_never_exposed(): r = resolve(ev([co("1", CN_A)], [emis()], DOC_A)) dump = str(preflight_report(r)) + str(r) diff --git a/agent/tests/test_enrichment.py b/agent/tests/test_enrichment.py index f437346..5e01f36 100644 --- a/agent/tests/test_enrichment.py +++ b/agent/tests/test_enrichment.py @@ -70,7 +70,8 @@ DECISIONS = {"CLASSIFICACAO_FISCAL": {"status": "AUTO_FILL_SAFE", "value": "2202 @pytest.fixture() def env(tmp_path): db = sqlite3.connect(":memory:", check_same_thread=False) - db.execute("CREATE TABLE C000025(CODIGO TEXT PRIMARY KEY, PRODUTO TEXT, GTIN TEXT, CLASSIFICACAO_FISCAL TEXT, CEST TEXT, IPI REAL, OTHER TEXT, CODBARRA TEXT)") + db.execute("CREATE TABLE C000025(CODIGO TEXT PRIMARY KEY, PRODUTO TEXT, GTIN TEXT, CLASSIFICACAO_FISCAL TEXT, CEST TEXT, IPI REAL, OTHER TEXT, CODBARRA TEXT, " + + ", ".join(f"{f} TEXT" for f in E.FISCAL_COMPLETION_FIELDS + E.CLASSIFICATION_CONTEXT_FIELDS) + ")") db.executemany("INSERT INTO C000025(CODIGO,PRODUTO,GTIN,OTHER) VALUES(?,?,?,?)", [(f"H{i}", f"HIST {i}", "", "keep") for i in range(50)]) db.commit() hub = Hub(dict(ENABLED), DECISIONS) @@ -114,6 +115,30 @@ def test_new_product_is_enriched_only_in_empty_fields_then_second_run_zero_write assert db.total_changes == before and len(hub.resolve_calls) == 1 # nothing re-processed, 0 writes +def test_current_report_includes_fiscal_completion_and_classification_columns(tmp_path): + """`current` sent to the Hub used to carry only CLASSIFICACAO_FISCAL/CEST/IPI. field_contract.resolve_product (now wired into + /v1/products/resolve) also needs the current CFOP/CST/CSOSN/... values and the classification columns infer_role reads (TIPO/ + PROD_COZINHA/PRODUTO_SERVICO/CODGRUPO) - otherwise it can never see what's already filled or auto-discover product_role.""" + db = sqlite3.connect(":memory:", check_same_thread=False) + db.execute("CREATE TABLE C000025(CODIGO TEXT PRIMARY KEY, PRODUTO TEXT, GTIN TEXT, CLASSIFICACAO_FISCAL TEXT, CEST TEXT, IPI REAL, OTHER TEXT, CODBARRA TEXT, " + + ", ".join(f"{f} TEXT" for f in E.FISCAL_COMPLETION_FIELDS + E.CLASSIFICATION_CONTEXT_FIELDS) + ")") + db.commit() + hub = Hub(dict(ENABLED), DECISIONS) + store = E.EnrichmentStore(tmp_path / "e.db") + now = [1000.0] + worker = E.EnrichmentWorker(hub=hub, store=store, read_conn=lambda: Conn(db), write_conn=lambda: Conn(db), create_reprocess=lambda c: None, clock=lambda: now[0], sleep=lambda s: None) + worker.step() # baseline + db.execute("INSERT INTO C000025(CODIGO,PRODUTO,GTIN,OTHER,CSOSN,TIPO,PRODUTO_SERVICO) VALUES(?,?,?,?,?,?,?)", ("N9", "PRODUTO X", "7891000100103", "keep", "102", "VENDA", None)) + db.commit() + now[0] += 5 + worker.step() + sent = hub.resolve_calls[0]["current"] + assert sent["CSOSN"] == "102" and sent["TIPO"] == "VENDA" + for f in E.FISCAL_COMPLETION_FIELDS + E.CLASSIFICATION_CONTEXT_FIELDS: + assert f in sent, f + assert hub.resolve_calls[0]["ev"] == {"recipes": [], "composed": [], "purchases": {}, "stock_entry": {}, "stock_in": {}} # targeted evidence rides along (empty here: sqlite has no RDB$RELATION_FIELDS) + + def test_existing_value_is_never_overwritten(env): db, hub, store, worker, now, _ = env worker.step() diff --git a/agent/tests/test_firebird_reader_product_evidence.py b/agent/tests/test_firebird_reader_product_evidence.py new file mode 100644 index 0000000..455dd71 --- /dev/null +++ b/agent/tests/test_firebird_reader_product_evidence.py @@ -0,0 +1,85 @@ +"""read_product_evidence: targeted, single-product evidence for the ad-hoc /v1/products/resolve call - the same recipes/composed/purchases/ +stock facts read_evidence already gathers for a full sync job, scoped to ONE product_code, never a full-table scan. Measured against a real +production-equivalent database (D:\\PROJETOS\\GTISISTEMAS\\Banco\\GTI.FDB, read-only): a full read_evidence() scan took ~40ms for the whole +catalog, and a targeted per-product lookup correctly drove field_contract.infer_role to RESALE/OWN_PRODUCTION for 5 of 8 real products with +zero operator questions - this test locks in the behavior with a portable (sqlite) fixture.""" +import sqlite3 +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).parents[1] / "src")) +from gti_fiscal_agent.firebird_reader import read_product_evidence + + +def _conn(): + db = sqlite3.connect(":memory:") + db.execute("CREATE TABLE RECEITA_INGREDIENTES(CODRECEITA TEXT)") + db.execute("CREATE TABLE C000032(CODPRODUTO TEXT, TIPO TEXT)") + db.execute("CREATE TABLE C000088(CODPRODUTO TEXT, ICMS_RETIDO TEXT, ICMS_VALORRETIDO REAL, SUB_VALOR REAL)") + db.execute("CREATE TABLE C000100(CODPRODUTO TEXT, ULTIMA_ENTRADA TEXT, NOTA_FISCAL TEXT, ENTRADAS REAL)") + return db + + +def test_no_firebird_system_table_degrades_to_no_evidence_never_raises(): + """A non-Firebird connection (this project's own sqlite test fakes) has no RDB$RELATION_FIELDS: the capability probe must degrade + gracefully to "no evidence", exactly like a missing table would, never raise.""" + db = sqlite3.connect(":memory:") + ev = read_product_evidence(db, "000001") + assert ev == {"recipes": set(), "composed": set(), "purchases": {}, "stock_entry": {}, "stock_in": {}} + + +def test_targeted_lookup_is_scoped_to_the_one_product_code(): + db = _conn() + db.executemany("INSERT INTO RECEITA_INGREDIENTES VALUES (?)", [("000009",), ("000010",)]) + db.execute("INSERT INTO C000032 VALUES ('000011', 'COMPOSICAO')") + db.execute("INSERT INTO C000088 VALUES ('000012', 'S', 0, 0)") + db.execute("INSERT INTO C000100 VALUES ('000013', '2026-01-01', '000001', 5)") + db.commit() + + def has_relation_fields(): + cur = db.cursor() + cur.execute("SELECT TRIM(RDB$RELATION_NAME), TRIM(RDB$FIELD_NAME) FROM RDB$RELATION_FIELDS WHERE RDB$RELATION_NAME IN ('C000032','C000088','C000100','RECEITA_INGREDIENTES')") + return cur.fetchall() + + # sqlite has no RDB$RELATION_FIELDS either; monkeypatch the cursor to answer the capability probe like a real Firebird database would + class ProbedConn: + def __init__(self, real): + self.real = real + + def cursor(self): + return ProbedCursor(self.real.cursor()) + + class ProbedCursor: + def __init__(self, real): + self.real = real + self._probe_result = None + + def execute(self, sql, params=()): + if "RDB$RELATION_FIELDS" in sql: + self._probe_result = [ + ("RECEITA_INGREDIENTES", "CODRECEITA"), ("C000032", "CODPRODUTO"), ("C000032", "TIPO"), + ("C000088", "CODPRODUTO"), ("C000088", "ICMS_RETIDO"), ("C000088", "ICMS_VALORRETIDO"), ("C000088", "SUB_VALOR"), + ("C000100", "CODPRODUTO"), ("C000100", "ULTIMA_ENTRADA"), ("C000100", "NOTA_FISCAL"), ("C000100", "ENTRADAS"), + ] + return + self._probe_result = None + sqlite_sql = sql.replace("SELECT FIRST 1 ", "SELECT ") + (" LIMIT 1" if "FIRST 1" in sql else "") + self.real.execute(sqlite_sql, params) + + def fetchall(self): + return self._probe_result if self._probe_result is not None else self.real.fetchall() + + def fetchone(self): + return self.real.fetchone() + + conn = ProbedConn(db) + ev = read_product_evidence(conn, "000009") + assert ev["recipes"] == {"000009"} and ev["composed"] == set() and ev["purchases"] == {} and ev["stock_entry"] == {} and ev["stock_in"] == {} + ev = read_product_evidence(conn, "000011") + assert ev["composed"] == {"000011"} and ev["recipes"] == set() + ev = read_product_evidence(conn, "000012") + assert ev["purchases"] == {"000012": [{"icms_retido": "S", "valor_retido": 0, "sub_valor": 0}]} + ev = read_product_evidence(conn, "000013") + assert ev["stock_entry"] == {"000013": True} and ev["stock_in"] == {"000013": True} + ev = read_product_evidence(conn, "999999") + assert ev == {"recipes": set(), "composed": set(), "purchases": {}, "stock_entry": {}, "stock_in": {}} diff --git a/agent/tests/test_run_agent_service.py b/agent/tests/test_run_agent_service.py index d012cf0..16b3941 100644 --- a/agent/tests/test_run_agent_service.py +++ b/agent/tests/test_run_agent_service.py @@ -346,6 +346,64 @@ def test_run_sends_heartbeat_but_never_polls_when_paused_centrally(monkeypatch, assert service.AgentState(config.state_db).stored_installation_id() == installation_id +def test_company_context_for_heartbeat_resolves_once_and_caches(monkeypatch): + """CRT was never sent at heartbeat: the Agent already proves it (company_context.resolve, via preflight_company_context) for coverage/RTC + jobs, but that value never reached the Hub. This locks in the fix: resolved once, cached, never re-queried once solid.""" + calls = {"n": 0} + + def fake_preflight(env): + calls["n"] += 1 + return {"status": "PASS", "company_uf": "BA", "company_crt": "1", "company_code": "000001"} + + monkeypatch.setattr("gti_fiscal_agent.company_context.preflight_company_context", fake_preflight) + result, reason = service._company_context_for_heartbeat({}, None) + assert result == {"status": "PASS", "company_uf": "BA", "company_crt": "1", "company_code": "000001"} and reason is None and calls["n"] == 1 + result2, reason2 = service._company_context_for_heartbeat({}, result) # already cached: never re-queries Firebird + assert result2 is result and reason2 is None and calls["n"] == 1 + + +def test_company_context_for_heartbeat_retries_while_unresolved_never_guesses(monkeypatch): + def fake_preflight(env): + return {"status": "FAIL", "classification": "UNRESOLVED", "reasons": ["UF_SOURCE_UNKNOWN"]} + + monkeypatch.setattr("gti_fiscal_agent.company_context.preflight_company_context", fake_preflight) + result, reason = service._company_context_for_heartbeat({}, None) + assert result is None and "UNRESOLVED" in reason # never a guessed uf/crt - explicitly unresolved, retried next tick + + +def test_run_heartbeat_carries_the_auto_discovered_company_crt(monkeypatch, tmp_path): + """End-to-end: once company context resolves, the CRT rides the heartbeat payload the Hub already knows how to store + (Installation.company_crt) - closing the transport gap without any Hub change.""" + stopper = _ExternalStopFlag() + fake_hub = _FakeHubForLoop() + installation_id = "00000000-0000-4000-8000-000000000333" + sent = {} + + def identity(): + return {"agent_id": 2, "installation_id": installation_id} + + def heartbeat_metadata(payload): + sent.update(payload) + stopper.stop = True + return {"agent_id": 2, "processing_enabled": False} + + fake_hub.identity = identity + fake_hub.heartbeat_metadata = heartbeat_metadata + fake_hub.next_job = lambda: (_ for _ in ()).throw(AssertionError("paused centrally; next_job must not be called")) + monkeypatch.setattr(service, "HubClient", lambda *a, **k: fake_hub) + monkeypatch.setattr(service, "load_executor", lambda: _FakeExecutor()) + monkeypatch.setattr(service.time, "sleep", lambda _seconds: None) + monkeypatch.setattr("gti_fiscal_agent.company_context.preflight_company_context", + lambda env: {"status": "PASS", "company_uf": "BA", "company_crt": "1", "company_code": "000001", "company_contribuinte_ipi": "N"}) + + config = _config(agent_id="2", poll_interval_seconds=0, state_db=str(tmp_path / "agent.db")) + env = {"AGENT_ID": "2", "FIREBIRD_DATABASE_PATH": AUTHORIZED, "FIREBIRD_WRITES": "0", "FISCAL_MODE": "DRY_RUN"} + + service.run(config, env, stopper=stopper) + + assert sent["company_uf"] == "BA" and sent["company_crt"] == "1" and sent["company_contribuinte_ipi"] == "N" + + def test_run_honors_an_externally_supplied_stopper(monkeypatch): # This is exactly the pattern the Windows Service wrapper uses: it never # constructs GracefulStop() (which would crash off the main thread), it diff --git a/central/alembic/versions/0032_contribuinte_ipi.py b/central/alembic/versions/0032_contribuinte_ipi.py new file mode 100644 index 0000000..5241f2d --- /dev/null +++ b/central/alembic/versions/0032_contribuinte_ipi.py @@ -0,0 +1,21 @@ +"""Installation.company_contribuinte_ipi: auto-discovered (C000004.CONTRIBUINTE_IPI, same row/query as company_crt) at heartbeat, never +operator-guessed. Closes the fiscal-completion gap where CST_IPI could not be resolved without asking a human for a fact the ERP already proves. + +Revision ID: 0032_contribuinte_ipi +Revises: 0031_enrichment_reevaluate +""" +from alembic import op +import sqlalchemy as sa + +revision = "0032_contribuinte_ipi" +down_revision = "0031_enrichment_reevaluate" +branch_labels = None +depends_on = None + + +def upgrade(): + op.add_column("installations", sa.Column("company_contribuinte_ipi", sa.String(1), nullable=True)) + + +def downgrade(): + op.drop_column("installations", "company_contribuinte_ipi") diff --git a/central/app/api.py b/central/app/api.py index aac8344..297f031 100644 --- a/central/app/api.py +++ b/central/app/api.py @@ -184,7 +184,7 @@ def router(session_factory, bootstrap_token): installation = Installation(agent_id=agent.id, tenant_id=agent.tenant_id, installation_id=installation_key, display_name=supplied.get("display_name") or agent.name, features_configured=True, corrector_enabled=True) # legacy/bootstrap self-registration keeps the pre-existing behaviour session.add(installation); session.flush() - fields = {"display_name": supplied.get("display_name"), "customer_code": supplied.get("customer_code"), "notes": supplied.get("notes"), "hostname": supplied.get("hostname"), "windows_version": supplied.get("windows_version"), "machine_fingerprint": supplied.get("machine_fingerprint"), "company_code": supplied.get("company_code"), "company_name": supplied.get("company_name"), "company_cnpj_masked": supplied.get("company_cnpj_masked"), "company_uf": supplied.get("company_uf"), "company_crt": supplied.get("company_crt"), "firebird_version": supplied.get("firebird_version"), "firebird_ods": supplied.get("firebird_ods"), "database_identifier": supplied.get("database_identifier"), "capabilities": supplied.get("capabilities"), "fiscal_engine_version": supplied.get("fiscal_engine_version"), "dataset_versions": supplied.get("dataset_versions"), "last_status": supplied.get("status") or payload.get("status"), "last_seen_at": now} + fields = {"display_name": supplied.get("display_name"), "customer_code": supplied.get("customer_code"), "notes": supplied.get("notes"), "hostname": supplied.get("hostname"), "windows_version": supplied.get("windows_version"), "machine_fingerprint": supplied.get("machine_fingerprint"), "company_code": supplied.get("company_code"), "company_name": supplied.get("company_name"), "company_cnpj_masked": supplied.get("company_cnpj_masked"), "company_uf": supplied.get("company_uf"), "company_crt": supplied.get("company_crt"), "company_contribuinte_ipi": supplied.get("company_contribuinte_ipi"), "firebird_version": supplied.get("firebird_version"), "firebird_ods": supplied.get("firebird_ods"), "database_identifier": supplied.get("database_identifier"), "capabilities": supplied.get("capabilities"), "fiscal_engine_version": supplied.get("fiscal_engine_version"), "dataset_versions": supplied.get("dataset_versions"), "last_status": supplied.get("status") or payload.get("status"), "last_seen_at": now} for key, value in fields.items(): if value is not None: setattr(installation, key, value) if supplied.get("started_at"): diff --git a/central/app/context_keys.py b/central/app/context_keys.py index 9904bac..5347472 100644 --- a/central/app/context_keys.py +++ b/central/app/context_keys.py @@ -15,12 +15,15 @@ ENUMS = { "incidence": {"NENHUMA"}, "origin": {str(i) for i in range(9)}, "store_ipi_profile": {"POPULATE_REFERENCE_RATE", "LEAVE_NOT_APPLICABLE"}, + # C000004.CONTRIBUINTE_IPI is not yet sent at heartbeat (no Agent-side reader exists for it): until that auto-discovery lands, this is the + # one-time operator answer field_contract's CST_IPI rule needs - never assumed "N" just because the Agent hasn't reported it. + "contribuinte_ipi": {"S", "N"}, } PROFILE_KEYS = {"pis_in_cst", "pis_out_cst", "pis_rate", "cofins_in_cst", "cofins_out_cst", "cofins_rate"} FREE_KEYS = {"ncm", "cest", "ex_tipi", "icms_rate", "credit_rate", "cst_ipi", "service_cfop", "pis_cofins_profile"} ALLOWED_KEYS = set(ENUMS) | FREE_KEYS PRODUCT_ONLY = {"ncm"} -COMPANY_ONLY = {"store_ipi_profile"} +COMPANY_ONLY = {"store_ipi_profile", "contribuinte_ipi"} NCM_KEYS = {"cest", "ex_tipi"} # tied to the NCM tables: answered per NCM or per product # an NCM answer is always about one product diff --git a/central/app/enrichment/api.py b/central/app/enrichment/api.py index c7e4b05..c7189c9 100644 --- a/central/app/enrichment/api.py +++ b/central/app/enrichment/api.py @@ -9,6 +9,7 @@ from sqlalchemy import select from ..blocked_products import upsert_blocker from ..config import settings +from ..field_contract import WRITE_FIELDS from ..models import Installation, ProductEnrichment, ProductFieldAudit from . import status as enrichment_status, textnorm from .placeholders import is_placeholder @@ -16,6 +17,29 @@ from .providers import build_providers from .quota import ProviderGateway from .resolver import FIELDS, resolve +CLASSIFICATION_CONTEXT_FIELDS = ("CODGRUPO", "TIPO", "PROD_COZINHA", "PRODUTO_SERVICO") # read-only evidence for field_contract.infer_role; never written +CURRENT_FIELDS = tuple(dict.fromkeys(FIELDS + tuple(WRITE_FIELDS) + CLASSIFICATION_CONTEXT_FIELDS)) # everything resolve() may read from the Agent's `current`: NCM/CEST/IPI + every fiscal-completion field + role evidence + + +def sanitize_product_evidence(raw) -> dict | None: + """Bounds the Agent's targeted, single-product evidence (firebird_reader.read_product_evidence) before it reaches field_contract: shape-only + validation (never trusts an authenticated Agent to send something huge or malformed for a single-product payload).""" + if not isinstance(raw, dict): + return None + out = {"recipes": set(), "composed": set(), "purchases": {}, "stock_entry": {}, "stock_in": {}} + for key in ("recipes", "composed"): + if isinstance(raw.get(key), list): + out[key] = {str(v) for v in raw[key][:8]} + if isinstance(raw.get("purchases"), dict): + for code, docs in list(raw["purchases"].items())[:8]: + if isinstance(docs, list): + out["purchases"][str(code)] = [d for d in docs[:32] if isinstance(d, dict)] + for key in ("stock_entry", "stock_in"): + if isinstance(raw.get(key), dict): + out[key] = {str(k): bool(v) for k, v in list(raw[key].items())[:8]} + return out + + PROVIDERS_OVERRIDE = None # tests inject fake providers here (never the network) PENDING_STATES = {"PENDING_PROVIDER_QUOTA", "PENDING_PROVIDERS_OFFLINE", "PENDING_TIME_BUDGET"} REASONS = { # human PT-BR reasons shown through the existing blockers (category ENRICHMENT); never internal jargon @@ -65,8 +89,9 @@ def router(principal, providers_factory=None): raise HTTPException(422, "product_code required") description, current = str(payload.get("description") or "")[:255], payload.get("current") if isinstance(payload.get("current"), dict) else {} started = time.monotonic() - result = resolve(session, ProviderGateway(session, providers()), gtin_raw=payload.get("gtin"), description=description, current={k: current.get(k) for k in FIELDS}, - client_label=f"installation:{installation.id}", budget_s=settings.enrichment_time_budget_seconds) + result = resolve(session, ProviderGateway(session, providers()), gtin_raw=payload.get("gtin"), description=description, current={k: current.get(k) for k in CURRENT_FIELDS}, + client_label=f"installation:{installation.id}", budget_s=settings.enrichment_time_budget_seconds, + product_code=code, installation=installation, tenant_id=agent.tenant_id, product_evidence=sanitize_product_evidence(payload.get("ev"))) latency = int((time.monotonic() - started) * 1000) row = session.scalar(select(ProductEnrichment).where(ProductEnrichment.installation_id == installation.id, ProductEnrichment.product_code == code)) if row is None: diff --git a/central/app/enrichment/fiscal_completion.py b/central/app/enrichment/fiscal_completion.py new file mode 100644 index 0000000..c7e1d70 --- /dev/null +++ b/central/app/enrichment/fiscal_completion.py @@ -0,0 +1,143 @@ +"""Post-NCM fiscal completion: wires the EXISTING, previously-dormant `central.app.field_contract.resolve_product` (the real field-by-field fiscal +contract engine - CFOP, CST, CSOSN, PIS/COFINS, CST_IPI, SITUACAO_TRIBUTARIA, IAT, IPPT, ALIQUOTA...) into the live Product Enrichment resolve call. + +No second fiscal engine. Everything here only ASSEMBLES the same inputs `resolve_product` already expects (`company`, `refs`, `ev`, `ctx`) from data +that already exists on the Hub, and maps its per-field CHANGE/ALREADY/NOT_APPLICABLE/MANUAL output onto the enrichment's own AUTO_FILL_SAFE vocabulary +so the SAME typed-write Agent pipeline can apply it. `company` comes straight from `Installation.company_uf/company_crt` (auto-discovered from the +Agent's own heartbeat - zero guessing). `ctx` comes straight from `FiscalContextEntry` (the SAME table the existing blocker-resolution workflow +already writes to when an operator answers a question) - so a missing key surfaces as an ordinary blocker the operator resolves the usual way; no +new configuration UI is created. `ev` (purchase/recipe/stock evidence) is not available from a single ad-hoc product resolve; those fields fall back +to MANUAL exactly as `field_contract` already does when evidence is absent - never a guess. +""" +from __future__ import annotations + +from sqlalchemy import select + +from ..classification import normalize_code +from ..datasets.service import DatasetService +from ..models import Dataset, FiscalContextEntry +from .. import field_contract as fc + +# Fields this module MAY auto-fill. Deliberately excludes CLASSIFICACAO_FISCAL/CEST/IPI (owned by the enrichment resolver itself), +# COD_ORIGEM_MERCADORIA (no evidence source exists per the contract - never inferred from CST), EX_TIPI (interacts with the IPI EX choice, +# left for a dedicated pass) and CCLASSTRIB (RTC's own domain, untouched). +FISCAL_COMPLETION_FIELDS = ("SITUACAO_TRIBUTARIA", "CFOP", "CFOP_ENTRADA", "CST", "CST_ENTRADA", "CSOSN", "ALIQUOTA", "CST_IPI", + "COD_ENTRADA_PIS", "COD_SAIDA_PIS", "ALIQ_PIS", "COD_ENTRADA_COFINS", "COD_SAIDA_COFINS", "ALIQ_COFINS", + "INCIDENCIA_PISCOFINS", "PISCOFINS", "IAT", "IPPT") + +# Value shape per field (validated BOTH here, before proposing a write, and again by the Agent right before the typed write). +DIGIT_FIELDS = {"CFOP": 4, "CFOP_ENTRADA": 4, "CST": 3, "CST_ENTRADA": 3, "CSOSN": 3, "CST_IPI": 3, "COD_ENTRADA_PIS": 2, "COD_SAIDA_PIS": 2, + "COD_ENTRADA_COFINS": 2, "COD_SAIDA_COFINS": 2} +NUMERIC_FIELDS = ("ALIQUOTA",) +DECIMAL_TEXT_FIELDS = ("ALIQ_PIS", "ALIQ_COFINS") +ENUM_FIELDS = {"SITUACAO_TRIBUTARIA": {"T", "I", "N", "S"}, "IAT": {"A", "T"}, "IPPT": {"P", "T"}, "PISCOFINS": {"N", "S"}, + "INCIDENCIA_PISCOFINS": {"NENHUMA", "MONOFASICA", "SUBST.TRIBUT."}} + + +def is_writable_shape(field: str, value) -> bool: + text = "" if value is None else str(value).strip() + if not text: + return False + if field in DIGIT_FIELDS: + digits = "".join(c for c in text if c.isdigit()) + return digits == text and len(digits) in (DIGIT_FIELDS[field], DIGIT_FIELDS[field] + 1) # CST/CSOSN family sometimes carries an origin digit prefix + if field in NUMERIC_FIELDS or field in DECIMAL_TEXT_FIELDS: + try: + float(text.replace(",", ".")) + return True + except ValueError: + return False + if field in ENUM_FIELDS: + return text.upper() in ENUM_FIELDS[field] + return False + + +def build_company(installation) -> dict: + """Auto-discovered, zero guessing: the Agent already reports these at every heartbeat (company_contribuinte_ipi from C000004.CONTRIBUINTE_IPI, + the same row/query as company_crt - see company_context.py on the Agent). Omits "contribuinte_ipi" entirely when the Agent hasn't reported it + yet (older Agent, or the field just never resolved): resolve_fiscal_fields treats that as genuinely unknown, never as a proven "N".""" + company = {"uf": (installation.company_uf or "").upper() or None, "crt": installation.company_crt or None} + if installation.company_contribuinte_ipi in ("S", "N"): + company["contribuinte_ipi"] = installation.company_contribuinte_ipi + return company + + +def build_refs(session, *, uf: str | None, cfop_candidates: set[str]) -> dict: + ds = DatasetService(session) + refs = {key: {r.code for r in ds.fiscal_codes(domain)} for key, domain in + (("csosn", "CSOSN"), ("cst_icms", "CST_ICMS"), ("cst_pis", "CST_PIS"), ("cst_cofins", "CST_COFINS"), ("origem", "ORIGEM_MERCADORIA"))} + refs["cfop_valid"] = {c for c in cfop_candidates if ds.cfop_lookup(code=c)} + refs["fcp_uf"] = {} + if uf: + fcp = ds.fcp_for_uf(uf) + if fcp: + refs["fcp_uf"][uf] = {"aliq_1": fcp.aliq_1, "observacao": fcp.observacao} + refs["cbs_rate_2026"] = (ds.rtc_cbs_rate(2026).rate if ds.rtc_cbs_rate(2026) else None) + refs["cclasstrib_count"] = session.scalar(select(Dataset.record_count).where(Dataset.kind == "RTC_CCLASSTRIB", Dataset.status == "ACTIVE")) or 0 + return refs + + +def build_context(session, tenant_id: int, company_key: str) -> fc.Context: + """SAME table the existing blocker-resolution workflow (`central/app/blocker_resolution.py`) already writes operator answers into.""" + rows = session.scalars(select(FiscalContextEntry).where(FiscalContextEntry.tenant_id == tenant_id, FiscalContextEntry.company_key == company_key)).all() + return fc.Context([{"scope": r.scope, "scope_key": r.scope_key, "key": r.key, "value": r.value} for r in rows]) + + +def resolve_fiscal_fields(session, *, product_code: str, description: str, current: dict, ncm: str, ncm_official: bool, installation, tenant_id: int, + extra_columns: dict | None = None, decision: dict | None = None, ev: dict | None = None) -> dict: + """Returns {field: {status, value, rule, reason}} for FISCAL_COMPLETION_FIELDS, in the enrichment's own vocabulary (AUTO_FILL_SAFE / ALREADY_SET / + NOT_APPLICABLE / NEEDS_REVIEW / INSUFFICIENT_EVIDENCE). Never called for a product whose NCM is not yet an official, vigente code. + + `decision` is the SAME dict `classification.Classifier.classify(...)` already returns for CEST/IPI (resolver.py's `_fiscal_decisions`) - reused + as-is (its `classification.ipi_candidates`/`cest_candidates`/`ba_st_mva` keys are exactly what `resolve_product` reads), never recomputed here.""" + if not ncm_official: + return {f: {"status": "INSUFFICIENT_EVIDENCE", "value": None, "rule": None, "reason": "depende de um NCM oficial vigente"} for f in FISCAL_COMPLETION_FIELDS} + ncm = normalize_code(ncm) + p = {"CODIGO": product_code, "PRODUTO": description, "CODGRUPO": None, "TIPO": None, "PROD_COZINHA": None, "PRODUTO_SERVICO": None, "CLASSIFICACAO_FISCAL": ncm} + p.update({k: v for k, v in (extra_columns or {}).items() if k in ("CODGRUPO", "TIPO", "PROD_COZINHA", "PRODUTO_SERVICO")}) + if str(p.get("PROD_COZINHA") or "").strip().isdigit(): # the Agent sends `current` as JSON strings (e.g. "1"); infer_role compares the real Firebird SMALLINT (== 1) + p["PROD_COZINHA"] = int(p["PROD_COZINHA"]) + for f in fc.WRITE_FIELDS: + p.setdefault(f, current.get(f)) + company = build_company(installation) + company_key = (installation.company_code or installation.company_name or installation.customer_code or "").strip() + ctx = build_context(session, tenant_id, company_key) + if "contribuinte_ipi" not in company: # ERP fact (heartbeat) always wins; the operator answer is only a fallback for an Agent that hasn't reported it yet + contribuinte_ipi, _ = ctx.get("contribuinte_ipi") + if contribuinte_ipi in ("S", "N"): + company["contribuinte_ipi"] = contribuinte_ipi + cfop_candidates = {str(current.get(f)).strip() for f in ("CFOP", "CFOP_ENTRADA") if current.get(f)} + refs = build_refs(session, uf=company.get("uf"), cfop_candidates=cfop_candidates) + # Agent-provided, product_code-targeted evidence (recipes/composed/purchases/stock - see firebird_reader.read_product_evidence): a real, + # indexed per-product query, never a full-table scan or a guess. Falls back to "no evidence" only when the Agent hasn't sent it yet. + ev = ev if isinstance(ev, dict) else {"recipes": set(), "composed": set(), "purchases": {}, "stock_entry": {}, "stock_in": {}} + d = decision or {"status": "EXACT_NCM", "cest_status": "NO_CEST_MAPPING_FOUND", "ba_st_mva_status": None, "classification": {"cest_candidates": [], "ipi_candidates": [], "ba_st_mva": {}}} + raw = fc.resolve_product(p, d, company, refs, ev, ctx) + out = {} + for field in FISCAL_COMPLETION_FIELDS: + r = raw.get(field) or {} + state = r.get("state") + if state == fc.ALREADY: + out[field] = {"status": "ALREADY_SET" if not _blank(current.get(field)) else "NOT_APPLICABLE", "value": None, "rule": r.get("reason"), "reason": r.get("reason")} + elif state == fc.NA and field == "CST_IPI" and "contribuinte_ipi" not in company: + # field_contract's own NA branch for CST_IPI fires whenever company.get("contribuinte_ipi") != "S" - which is also true when the fact is + # simply UNKNOWN. Never let an unknown fact read as a proven "not an IPI taxpayer": surface it as a missing answer instead. + out[field] = {"status": "NEEDS_REVIEW", "value": None, "rule": None, "reason": "contribuinte de IPI não confirmado", "missing_key": "contribuinte_ipi", "missing_scope": "COMPANY"} + elif state == fc.NA: + out[field] = {"status": "NOT_APPLICABLE", "value": None, "rule": r.get("reason"), "reason": r.get("reason")} + elif state == fc.CHANGE and _blank(current.get(field)) and is_writable_shape(field, r.get("proposed")): + out[field] = {"status": "AUTO_FILL_SAFE", "value": str(r.get("proposed")), "rule": r.get("reason"), "reason": None} + elif state == fc.CHANGE: + out[field] = {"status": "NEEDS_REVIEW", "value": None, "rule": r.get("reason"), "reason": "valor proposto não seguro para escrita automática" if not _blank(current.get(field)) else "formato do valor proposto não confere"} + else: # MANUAL + out[field] = {"status": "NEEDS_REVIEW", "value": None, "rule": None, "reason": r.get("reason"), "missing_key": r.get("input_key"), "missing_scope": r.get("input_scope")} + return out + + +def _blank(v) -> bool: + return v is None or str(v).strip() == "" + + +def coverage_status(field_statuses: dict) -> str: + """COMPLETE when every field is filled/already-correct/not-applicable; SAFE_BLOCKED when at least one relevant field still needs a decision.""" + return "SAFE_BLOCKED" if any(v["status"] in ("NEEDS_REVIEW", "INSUFFICIENT_EVIDENCE") for v in field_statuses.values()) else "COMPLETE" diff --git a/central/app/enrichment/resolver.py b/central/app/enrichment/resolver.py index ab5a9a6..4fc3259 100644 --- a/central/app/enrichment/resolver.py +++ b/central/app/enrichment/resolver.py @@ -15,7 +15,7 @@ from sqlalchemy import select from ..classification import Classifier, normalize_code from ..datasets.service import DatasetService from ..models import Dataset, NcmEntry -from . import kb, textnorm +from . import fiscal_completion, kb, textnorm from .gtin import normalize_gtin from .placeholders import is_placeholder, is_writable_empty from .quota import ProviderGateway @@ -79,7 +79,24 @@ def _empty(current: dict, name: str) -> bool: 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: +def _merge_fiscal_completion(session, out: dict, *, ncm_official: bool, ncm_for_engine: str | None, current: dict, description: str, + product_code: str | None, installation, tenant_id: int | None, decision: dict | None, product_evidence: dict | None = None) -> None: + """Merges field_contract.resolve_product's own field-by-field decisions (CFOP/CST/CSOSN/PIS-COFINS/...) into `out`, in the SAME + AUTO_FILL_SAFE/ALREADY_SET/NOT_APPLICABLE/INSUFFICIENT_EVIDENCE vocabulary CEST/IPI already use, plus NEEDS_REVIEW for anything that needs an + operator's context answer. No-op when the caller (a test, or a code path with no installation on hand) did not pass installation/tenant_id/product_code.""" + if installation is None or tenant_id is None or product_code is None: + return + fields = fiscal_completion.resolve_fiscal_fields(session, product_code=product_code, description=description, current=current, ncm=ncm_for_engine or "", + ncm_official=ncm_official, installation=installation, tenant_id=tenant_id, decision=decision, extra_columns=current, + ev=product_evidence) + for field, r in fields.items(): + out[field] = _field(r["status"], r.get("value"), r.get("rule"), reason=r.get("reason")) + if r.get("missing_key"): + out[field]["missing_key"], out[field]["missing_scope"] = r["missing_key"], r.get("missing_scope") + + +def _fiscal_decisions(session, ncm_eval: dict, gtin_kb_verified: bool, current: dict, description: str, *, name_verified=False, + product_code: str | None = None, installation=None, tenant_id: int | None = None, product_evidence: dict | None = None) -> dict: out = {} if not _empty(current, "CLASSIFICACAO_FISCAL"): out["CLASSIFICACAO_FISCAL"] = _field(ALREADY, reason="campo já preenchido") @@ -98,6 +115,8 @@ def _fiscal_decisions(session, ncm_eval: dict, gtin_kb_verified: bool, current: 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") + _merge_fiscal_completion(session, out, ncm_official=False, ncm_for_engine=ncm_for_engine, current=current, description=description, + product_code=product_code, installation=installation, tenant_id=tenant_id, decision=None, product_evidence=product_evidence) return out # CEST / IPI are deterministic functions of an OFFICIAL NCM: the NCM we are about to write (AUTO) or one already in the product (validated above against the official dataset) ncm_solid = ncm["status"] == AUTO or (ncm["status"] == ALREADY and ncm_is_official(session, ncm_for_engine)) @@ -120,12 +139,20 @@ def _fiscal_decisions(session, ncm_eval: dict, gtin_kb_verified: bool, current: out["IPI"] = _field(NA, reason="TIPI sem alíquota numérica (NT/branca)") else: out["IPI"] = _field(INSUFFICIENT, reason=f"IPI não determinístico ({decision['ipi_status']})") + _merge_fiscal_completion(session, out, ncm_official=ncm_solid, ncm_for_engine=ncm_for_engine, current=current, description=description, + product_code=product_code, installation=installation, tenant_id=tenant_id, decision=decision, product_evidence=product_evidence) return out -def resolve(session, gateway: ProviderGateway, *, gtin_raw, description, current: dict | None = None, client_label: str | None = None, allow_external: bool = True, budget_s: float | None = None) -> dict: +def resolve(session, gateway: ProviderGateway, *, gtin_raw, description, current: dict | None = None, client_label: str | None = None, allow_external: bool = True, budget_s: float | None = None, + product_code: str | None = None, installation=None, tenant_id: int | None = None, product_evidence: dict | None = None) -> dict: started = time.monotonic() - """Returns {state, gtin_status, gtin, fields:{...}, providers:[...], retry_at, dataset_versions}. Never raises for provider trouble.""" + """Returns {state, gtin_status, gtin, fields:{...}, providers:[...], retry_at, dataset_versions}. Never raises for provider trouble. + + `product_code`/`installation`/`tenant_id`, when given, additionally resolve the post-NCM fiscal-completion fields (CFOP/CST/CSOSN/PIS-COFINS/...) + via the existing field_contract engine (see `_merge_fiscal_completion`) and merge them into `fields` alongside CLASSIFICACAO_FISCAL/CEST/IPI. + `product_evidence` is the Agent's targeted, product_code-scoped recipes/composed/purchases/stock evidence (firebird_reader.read_product_evidence) - + never a full-table scan. Named separately from the local `ev` variable below (NCM knowledge-base evidence), a different, pre-existing concept.""" current = {k: v for k, v in (current or {}).items()} g = normalize_gtin(gtin_raw) result = {"state": "NOT_FOUND", "gtin_status": g["status"], "gtin": g.get("key"), "fields": {}, "providers": [], "retry_at": None, "dataset_versions": dataset_versions(session)} @@ -139,12 +166,12 @@ def resolve(session, gateway: ProviderGateway, *, gtin_raw, description, current if ev and ev["status"] == kb.VERIFIED_KB: gateway.quota.record_kb_hit() result["state"] = "EXACT_GTIN_VERIFIED" - result["fields"] = _fiscal_decisions(session, ev, True, current, description or "") + result["fields"] = _fiscal_decisions(session, ev, True, current, description or "", product_code=product_code, installation=installation, tenant_id=tenant_id, product_evidence=product_evidence) session.commit() return result if ev and ev["status"] == kb.CONFLICT: # local evidence already conflicts: an external answer cannot settle it, so spend no quota result["state"] = "CONFLICT" - result["fields"] = _fiscal_decisions(session, ev, False, current, description or "") + result["fields"] = _fiscal_decisions(session, ev, False, current, description or "", product_code=product_code, installation=installation, tenant_id=tenant_id, product_evidence=product_evidence) gateway.quota.record_kb_hit() session.commit() return result @@ -175,7 +202,7 @@ def resolve(session, gateway: ProviderGateway, *, gtin_raw, description, current if product is None and description: product = kb.upsert_product(session, key, description) ev = kb.evaluate(session, product, "ncm", is_official) if product else {"value": None, "status": kb.UNVERIFIED, "families": [], "candidates": [], "rejected": []} - result["fields"] = _fiscal_decisions(session, ev, False, current, description or "") + result["fields"] = _fiscal_decisions(session, ev, False, current, description or "", product_code=product_code, installation=installation, tenant_id=tenant_id, product_evidence=product_evidence) if ev["status"] == kb.CONFLICT: result["state"] = "CONFLICT" elif ev["status"] == kb.VERIFIED_KB: @@ -211,7 +238,7 @@ def resolve(session, gateway: ProviderGateway, *, gtin_raw, description, current if candidates and len(verified) == len(candidates) and len({ev["value"] for _, ev in verified}) == 1: _, ev = verified[0] result["state"] = "EXACT_KB_NAME_VERIFIED" - result["fields"] = _fiscal_decisions(session, ev, True, current, description or "", name_verified=True) + result["fields"] = _fiscal_decisions(session, ev, True, current, description or "", name_verified=True, product_code=product_code, installation=installation, tenant_id=tenant_id, product_evidence=product_evidence) elif len({ev["value"] for _, ev in verified}) > 1: result["state"] = "CONFLICT" result["fields"] = {f: _field(CONFLICT, reason="descrição corresponde a produtos com NCM divergentes") if _empty(current, f) else _field(ALREADY) for f in FIELDS} diff --git a/central/app/installation_identity.py b/central/app/installation_identity.py index 0f87795..a2f1caf 100644 --- a/central/app/installation_identity.py +++ b/central/app/installation_identity.py @@ -62,7 +62,7 @@ def _safe_value(key: str, value): def sanitize_registration(payload: dict) -> dict: allowed = { "installation_id", "display_name", "customer_code", "notes", "hostname", "windows_version", "machine_fingerprint", - "company_code", "company_name", "company_cnpj", "company_uf", "company_crt", "firebird_version", "firebird_ods", + "company_code", "company_name", "company_cnpj", "company_uf", "company_crt", "company_contribuinte_ipi", "firebird_version", "firebird_ods", "database_identifier", "database_path", "capabilities", "fiscal_engine_version", "dataset_versions", "started_at", "status", "schema_fingerprint", "agent_version", "last_job_id", "coverage", "health", "event_type", "message", "context", } diff --git a/central/app/models.py b/central/app/models.py index 0184a0c..224d755 100644 --- a/central/app/models.py +++ b/central/app/models.py @@ -267,6 +267,7 @@ class Installation(Base): company_cnpj_masked: Mapped[str | None] = mapped_column(String(24), nullable=True) company_uf: Mapped[str | None] = mapped_column(String(2), nullable=True) company_crt: Mapped[str | None] = mapped_column(String(16), nullable=True) + company_contribuinte_ipi: Mapped[str | None] = mapped_column(String(1), nullable=True) # C000004.CONTRIBUINTE_IPI, same auto-discovery as company_crt firebird_version: Mapped[str | None] = mapped_column(String(64), nullable=True) firebird_ods: Mapped[str | None] = mapped_column(String(32), nullable=True) database_identifier: Mapped[str | None] = mapped_column(String(255), nullable=True) diff --git a/central/app/version.py b/central/app/version.py index abe7a7f..cccb11b 100644 --- a/central/app/version.py +++ b/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.6" +GTI_FISCAL_VERSION = "1.3.7" API_COMPAT_MAJOR_MINOR = "1.0" # an Agent 1.0.x/1.1.x talks to any Hub 1.0.x diff --git a/central/pyproject.toml b/central/pyproject.toml index 7eb38e0..69e6733 100644 --- a/central/pyproject.toml +++ b/central/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "gti-fiscal-hub" -version = "1.3.6" +version = "1.3.7" requires-python = ">=3.10" dependencies = [ "fastapi>=0.115,<1", diff --git a/central/tests/test_fiscal_completion.py b/central/tests/test_fiscal_completion.py new file mode 100644 index 0000000..ee4d441 --- /dev/null +++ b/central/tests/test_fiscal_completion.py @@ -0,0 +1,184 @@ +"""Post-NCM fiscal completion: wires the existing (previously dormant) field_contract.resolve_product into the live pipeline. Reuses the SAME engine, +the SAME FiscalContextEntry table the blocker-resolution workflow already writes to, and company_uf/company_crt already auto-discovered from every +heartbeat - no new rule, no guessing.""" +from sqlalchemy import select +from sqlalchemy.orm import Session + +from central.app.enrichment import fiscal_completion as fcomp +from central.app.field_contract import WRITE_FIELDS +from central.app.models import Agent, Base, CfopEntry, Dataset, FiscalCodeEntry, FiscalContextEntry, Installation, Tenant + + +def _seed_codes(session, ncm="22030000"): + ds = Dataset(kind="NFE_MOC_TABLES", provider="x", sha256="a" * 64, version_label="v", act="a", source_url="u", status="ACTIVE", record_count=0) + session.add(ds); session.flush() + for domain, codes in (("CSOSN", ["101", "102", "103", "300", "400", "500"]), ("CST_ICMS", ["00", "20", "40", "41", "60"]), + ("CST_PIS", ["01", "49", "98"]), ("CST_COFINS", ["01", "49", "98"]), ("ORIGEM_MERCADORIA", list("012345678"))): + for code in codes: + session.add(FiscalCodeEntry(dataset_id=ds.id, domain=domain, code=code, description=f"{domain} {code}", source_section="test")) + cfop_ds = Dataset(kind="CFOP", provider="x", sha256="b" * 64, version_label="v", act="a", source_url="u", status="ACTIVE", record_count=0) + session.add(cfop_ds); session.flush() + for code in ("5101", "5102", "5401", "1102", "1101"): + session.add(CfopEntry(dataset_id=cfop_ds.id, code=code, title=f"CFOP {code}")) + session.commit() + + +def _installation(session, uf="BA", crt="1", company_key="000001", contribuinte_ipi=None): + tenant = Tenant(id=1, name="Acme"); session.add(tenant); session.flush() + agent = Agent(tenant_id=1, name="a", token_hash="x" * 64); session.add(agent); session.flush() + inst = Installation(agent_id=agent.id, tenant_id=1, installation_id="11111111-1111-4111-8111-111111111111", company_uf=uf, company_crt=crt, company_code=company_key, + company_contribuinte_ipi=contribuinte_ipi) + session.add(inst); session.flush() + return inst + + +def _ctx(session, tenant_id, company_key, entries): + for e in entries: + session.add(FiscalContextEntry(tenant_id=tenant_id, company_key=company_key, scope=e[0], scope_key=e.get("scope_key", "") if isinstance(e, dict) else "", key=e[1] if not isinstance(e, dict) else e["key"], + value=e[2] if not isinstance(e, dict) else e["value"], source="OPERATOR")) + session.commit() + + +def env(): + from sqlalchemy import create_engine + from sqlalchemy.pool import StaticPool + engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool) + Base.metadata.create_all(engine) + return Session(engine) + + +def test_field_shapes(): + assert fcomp.is_writable_shape("CFOP", "5102") and not fcomp.is_writable_shape("CFOP", "51A2") and not fcomp.is_writable_shape("CFOP", "510") + assert fcomp.is_writable_shape("CST", "000") and fcomp.is_writable_shape("CST", "0000") + assert fcomp.is_writable_shape("ALIQUOTA", "17.5") and fcomp.is_writable_shape("ALIQ_PIS", "1,65") and not fcomp.is_writable_shape("ALIQUOTA", "abc") + assert fcomp.is_writable_shape("SITUACAO_TRIBUTARIA", "t") and not fcomp.is_writable_shape("SITUACAO_TRIBUTARIA", "X") + assert fcomp.is_writable_shape("IAT", "A") and fcomp.is_writable_shape("IPPT", "P") and fcomp.is_writable_shape("PISCOFINS", "N") + assert not fcomp.is_writable_shape("CFOP", "") + + +def test_company_auto_discovered_from_heartbeat_no_guessing(): + s = env() + inst = _installation(s) + assert fcomp.build_company(inst) == {"uf": "BA", "crt": "1"} + + +def test_resolve_fiscal_fields_with_full_context_fills_independent_safe_fields(): + s = env() + _seed_codes(s) + inst = _installation(s) + import json + pis_cofins_profile = json.dumps({"pis_in_cst": "01", "pis_out_cst": "01", "pis_rate": "1.65", "cofins_in_cst": "01", "cofins_out_cst": "01", "cofins_rate": "7.6"}) + _ctx(s, 1, "000001", [ + ("COMPANY", "product_role", "RESALE"), ("COMPANY", "icms_treatment", "TAXED"), ("COMPANY", "origin", "0"), + ("COMPANY", "pis_cofins_profile", pis_cofins_profile), ("COMPANY", "piscofins", "S"), ("COMPANY", "rounding", "A"), + ("COMPANY", "store_ipi_profile", "LEAVE_NOT_APPLICABLE"), ("COMPANY", "icms_rate", "17"), ("COMPANY", "entry_scope", "INTRA"), + ]) + out = fcomp.resolve_fiscal_fields(s, product_code="X1", description="CERVEJA LATA", current={}, ncm="22030000", ncm_official=True, installation=inst, tenant_id=1) + safe = {f: v["value"] for f, v in out.items() if v["status"] == "AUTO_FILL_SAFE"} + assert "CSOSN" in safe and safe["CSOSN"] == "102" + assert "PISCOFINS" in safe or out["PISCOFINS"]["status"] == "ALREADY_SET" + for f, v in out.items(): + assert v["status"] in ("AUTO_FILL_SAFE", "ALREADY_SET", "NOT_APPLICABLE", "NEEDS_REVIEW", "INSUFFICIENT_EVIDENCE"), f + + +def test_field_independence_one_missing_context_does_not_block_others(): + s = env() + _seed_codes(s) + inst = _installation(s) + _ctx(s, 1, "000001", [("COMPANY", "product_role", "RESALE"), ("COMPANY", "rounding", "A")]) # icms_treatment / pis_cofins_profile intentionally missing + out = fcomp.resolve_fiscal_fields(s, product_code="X2", description="CERVEJA LATA", current={}, ncm="22030000", ncm_official=True, installation=inst, tenant_id=1) + assert out["IAT"]["status"] == "AUTO_FILL_SAFE" and out["IAT"]["value"] == "A" # resolved independently + assert out["CSOSN"]["status"] == "NEEDS_REVIEW" and out["CSOSN"]["missing_key"] == "icms_treatment" + assert out["COD_ENTRADA_PIS"]["status"] == "NEEDS_REVIEW" and out["COD_ENTRADA_PIS"]["missing_key"] == "pis_cofins_profile" + + +def test_contribuinte_ipi_auto_discovered_from_installation_wins_over_operator_context(): + """Installation.company_contribuinte_ipi (auto-discovered at heartbeat from C000004.CONTRIBUINTE_IPI) is an ERP fact; a FiscalContextEntry + operator answer is only a fallback for an Agent version that hasn't reported it yet. ERP fact must win when both exist.""" + s = env() + _seed_codes(s) + inst = _installation(s, contribuinte_ipi="S") + _ctx(s, 1, "000001", [("COMPANY", "contribuinte_ipi", "N")]) # stale/wrong operator answer - the ERP fact must override it + out = fcomp.resolve_fiscal_fields(s, product_code="X9", description="MAQUINA", current={}, ncm="22030000", ncm_official=True, installation=inst, tenant_id=1) + assert out["CST_IPI"]["status"] == "NEEDS_REVIEW" # contribuinte_ipi=S alone doesn't fill CST_IPI (still needs a declared value), but it's no longer NOT_APPLICABLE + assert out["CST_IPI"]["reason"] != "contribuinte de IPI não confirmado" # proves the ERP fact (S) was used, not "unknown" + + +def test_targeted_product_evidence_feeds_role_inference_st_withheld_never_a_full_scan(): + """`ev` (firebird_reader.read_product_evidence's targeted, per-product recipes/composed/purchases/stock) reaches field_contract.resolve_product + the same way the full-scan `read_evidence` already does for sync jobs - just scoped to ONE product, never the whole catalog.""" + s = env() + _seed_codes(s) + inst = _installation(s) + out = fcomp.resolve_fiscal_fields(s, product_code="X10", description="BOLO CASEIRO", current={}, ncm="22030000", ncm_official=True, installation=inst, tenant_id=1, + ev={"recipes": {"X10"}, "composed": set(), "purchases": {}, "stock_entry": {}, "stock_in": {}}) + assert out["CFOP"]["status"] == "NEEDS_REVIEW" # OWN_PRODUCTION now provable from the recipe evidence alone - no product_role question, only the CFOP direction/context remains + assert out["CFOP"]["missing_key"] != "product_role" + + +def test_extra_columns_feed_field_contracts_own_role_inference_no_operator_question_needed(): + """PRODUTO_SERVICO is a real C000025 column field_contract.infer_role already reads as objective ERP evidence (no ctx.get("product_role") + needed). Proves `extra_columns` (the Agent's classification columns) actually reach `resolve_product`: with PRODUTO_SERVICO set, CFOP is + blocked on the SERVICE-specific `service_cfop` key, not the generic `product_role` key a fully-unknown product would need.""" + s = env() + _seed_codes(s) + inst = _installation(s) + out = fcomp.resolve_fiscal_fields(s, product_code="X7", description="CONSULTORIA", current={}, ncm="22030000", ncm_official=True, installation=inst, tenant_id=1, + extra_columns={"PRODUTO_SERVICO": "1"}) + assert out["CFOP"]["status"] == "NEEDS_REVIEW" and out["CFOP"]["missing_key"] == "service_cfop" + + +def test_prod_cozinha_string_from_agent_json_payload_is_normalized_to_int_for_infer_role(): + """The Agent sends `current`/extra columns as JSON strings ("1", not 1). field_contract.infer_role compares PROD_COZINHA == 1 (a real + Firebird SMALLINT). Without normalization the kitchen-routing signal would silently never fire.""" + s = env() + _seed_codes(s) + inst = _installation(s) + out = fcomp.resolve_fiscal_fields(s, product_code="X8", description="PRATO FEITO", current={}, ncm="22030000", ncm_official=True, installation=inst, tenant_id=1, + extra_columns={"PROD_COZINHA": "1"}) + assert out["CFOP"]["status"] == "NEEDS_REVIEW" # PROD_COZINHA=1 alone is UNKNOWN (not proof), but it must be READ, not silently ignored as a non-numeric string + assert out["CFOP"]["missing_key"] == "product_role" + + +def test_cst_ipi_unknown_contribuinte_ipi_is_needs_review_not_a_silent_not_applicable(): + """C000004.CONTRIBUINTE_IPI is not yet sent at heartbeat - `company` never carries it unless an operator answered the context key. field_contract's + own rule reads a missing fact the same as a proven "N" (NOT_APPLICABLE); this module must not let that false negative through silently.""" + s = env() + _seed_codes(s) + inst = _installation(s) # no contribuinte_ipi context entry: genuinely unknown + out = fcomp.resolve_fiscal_fields(s, product_code="X5", description="CERVEJA LATA", current={}, ncm="22030000", ncm_official=True, installation=inst, tenant_id=1) + assert out["CST_IPI"]["status"] == "NEEDS_REVIEW" and out["CST_IPI"]["missing_key"] == "contribuinte_ipi" + _ctx(s, 1, "000001", [("COMPANY", "contribuinte_ipi", "N")]) + out = fcomp.resolve_fiscal_fields(s, product_code="X6", description="CERVEJA LATA", current={}, ncm="22030000", ncm_official=True, installation=inst, tenant_id=1) + assert out["CST_IPI"]["status"] == "NOT_APPLICABLE" # once the fact is genuinely proven "N", the engine's own NA decision is trusted again + + +def test_not_official_ncm_blocks_everything_with_explicit_reason(): + s = env() + inst = _installation(s) + out = fcomp.resolve_fiscal_fields(s, product_code="X3", description="X", current={}, ncm="00000000", ncm_official=False, installation=inst, tenant_id=1) + assert all(v["status"] == "INSUFFICIENT_EVIDENCE" for v in out.values()) + + +def test_existing_value_preserved_not_overwritten(): + """CSOSN 500 declares ICMS already withheld by ST - a DIFFERENT treatment than a plain TAXED declaration (CSOSN_FAMILY in field_contract.py + maps icms_treatment=ST_WITHHELD -> {"500"} and icms_treatment=TAXED -> {"101","102","103"}). With the matching treatment declared, the engine + proves the existing value correct and does not touch it.""" + s = env() + _seed_codes(s) + inst = _installation(s) + _ctx(s, 1, "000001", [("COMPANY", "product_role", "RESALE"), ("COMPANY", "icms_treatment", "ST_WITHHELD"), ("COMPANY", "origin", "0")]) + out = fcomp.resolve_fiscal_fields(s, product_code="X4", description="CERVEJA LATA", current={"CSOSN": "500"}, ncm="22030000", ncm_official=True, installation=inst, tenant_id=1) + assert out["CSOSN"]["status"] == "ALREADY_SET" + + +def test_coverage_status_complete_vs_safe_blocked(): + complete = {f: {"status": "ALREADY_SET"} for f in fcomp.FISCAL_COMPLETION_FIELDS} + assert fcomp.coverage_status(complete) == "COMPLETE" + blocked = dict(complete, CFOP={"status": "NEEDS_REVIEW"}) + assert fcomp.coverage_status(blocked) == "SAFE_BLOCKED" + + +def test_all_write_fields_covered_by_the_module_except_the_explicit_exclusions(): + excluded = {"CLASSIFICACAO_FISCAL", "CEST", "IPI", "COD_ORIGEM_MERCADORIA", "EX_TIPI"} + assert set(WRITE_FIELDS) - excluded == set(fcomp.FISCAL_COMPLETION_FIELDS) diff --git a/central/tests/test_health.py b/central/tests/test_health.py index f5abe73..f4c22bf 100644 --- a/central/tests/test_health.py +++ b/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.6" and body["api_compat"] == "1.0" and body["engine_version"] and body["decision_version"] + assert body["version"] == "1.3.7" and body["api_compat"] == "1.0" and body["engine_version"] and body["decision_version"] def test_production_refuses_the_dev_token(monkeypatch): diff --git a/central/tests/test_product_enrichment_api.py b/central/tests/test_product_enrichment_api.py index 52ca5ce..b40b1d6 100644 --- a/central/tests/test_product_enrichment_api.py +++ b/central/tests/test_product_enrichment_api.py @@ -22,8 +22,11 @@ def setup(monkeypatch, mode="ENABLED"): return client, engine -def resolve(client, code="N1", gtin=G1, desc="REFRIGERANTE COLA LATA 350ML", current=None): - return client.post("/v1/products/resolve", json={"product_code": code, "gtin": gtin, "description": desc, "current": current or {}}, headers=H).json() +def resolve(client, code="N1", gtin=G1, desc="REFRIGERANTE COLA LATA 350ML", current=None, ev=None): + body = {"product_code": code, "gtin": gtin, "description": desc, "current": current or {}} + if ev is not None: + body["ev"] = ev + return client.post("/v1/products/resolve", json=body, headers=H).json() def test_disabled_installation_is_not_processed_and_never_calls_providers(monkeypatch): @@ -59,6 +62,43 @@ def test_applied_report_is_audited_and_human_override_is_final(monkeypatch): assert client.post("/v1/products/enrichment/applied", json={"product_code": "UNKNOWN", "fields": []}, headers=H).status_code == 404 +def test_resolve_wires_field_contract_fiscal_completion_fields_alongside_ncm(monkeypatch): + """field_contract.resolve_product was previously only exercised by its own unit tests. This proves the LIVE /v1/products/resolve endpoint + now calls it too: once the NCM auto-fills, CFOP/CST/CSOSN/PIS-COFINS/... decisions ride along in the same response - no second engine, no + separate round trip - using company_uf/company_crt already auto-discovered from the Agent's heartbeat (Installation.company_uf/company_crt).""" + from central.app.enrichment import fiscal_completion as fcomp + from central.app.models import FiscalContextEntry + client, engine = setup(monkeypatch) + with Session(engine) as s: + inst = s.scalars(select(Installation)).one() + inst.company_uf, inst.company_crt, inst.company_code = "BA", "1", "000001" + s.commit() + tenant_id = inst.tenant_id + body = resolve(client) + assert body["fields"]["CLASSIFICACAO_FISCAL"]["status"] == "AUTO_FILL_SAFE" + for field in fcomp.FISCAL_COMPLETION_FIELDS: # every fiscal-completion field gets an explicit decision, never silently missing + assert field in body["fields"], field + assert body["fields"]["CSOSN"]["status"] == "NEEDS_REVIEW" and body["fields"]["CSOSN"]["missing_key"] == "icms_treatment" # 0 FiscalContextEntry rows: correctly blocked, not guessed + with Session(engine) as s: + s.add(FiscalContextEntry(tenant_id=tenant_id, company_key="000001", scope="COMPANY", scope_key="", key="icms_treatment", value="TAXED", source="OPERATOR")) + s.commit() + again = resolve(client, code="N9") + assert again["fields"]["CSOSN"]["status"] == "AUTO_FILL_SAFE" and again["fields"]["CSOSN"]["value"] == "102" # same operator answer, field independently resolved once context exists + + +def test_resolve_accepts_agents_targeted_product_evidence_end_to_end(monkeypatch): + """The Agent's targeted, product_code-scoped recipes/composed/purchases/stock evidence (firebird_reader.read_product_evidence) rides the + resolve payload as `ev` and reaches field_contract.infer_role - proven end-to-end through the real HTTP route, not just the fiscal_completion + unit tests. Recipe evidence alone proves OWN_PRODUCTION with zero operator questions.""" + client, engine = setup(monkeypatch) + with Session(engine) as s: + inst = s.scalars(select(Installation)).one() + inst.company_uf, inst.company_crt, inst.company_code = "BA", "1", "000001" + s.commit() + body = resolve(client, code="N10", ev={"recipes": ["N10"], "composed": [], "purchases": {}, "stock_entry": {}, "stock_in": {}}) + assert body["fields"]["CFOP"]["status"] == "NEEDS_REVIEW" and body["fields"]["CFOP"].get("missing_key") != "product_role" + + def test_unknown_gtin_creates_a_human_blocker_in_portuguese(monkeypatch): client, engine = setup(monkeypatch) monkeypatch.setattr(enrich_api, "PROVIDERS_OVERRIDE", [Fake("cosmos", {}, True)]) diff --git a/central/tests/test_zero_touch_hub.py b/central/tests/test_zero_touch_hub.py index 5a1a212..9d877fd 100644 --- a/central/tests/test_zero_touch_hub.py +++ b/central/tests/test_zero_touch_hub.py @@ -53,6 +53,16 @@ def test_first_heartbeat_creates_exactly_one_initial_sync(): assert len(initial) == 1 and initial[0].status == "PENDING" and initial[0].payload["reason"] == "INITIAL_SYNC" +def test_heartbeat_metadata_stores_auto_discovered_company_crt_and_contribuinte_ipi(): + """company_crt and company_contribuinte_ipi were never sent at heartbeat before (a real transport gap); once the Agent reports them + (auto-discovered from C000004, never guessed), the Hub stores them on Installation - fiscal_completion.build_company reads them directly.""" + client, engine = _client() + client.post("/v1/agents/heartbeat", json={"installation_id": INST, "status": "IDLE", "agent_version": "1.1.3", "company_uf": "BA", "company_crt": "1", "company_contribuinte_ipi": "S"}, headers=H) + with Session(engine) as s: + inst = s.scalar(select(Installation)) + assert inst.company_uf == "BA" and inst.company_crt == "1" and inst.company_contribuinte_ipi == "S" + + def test_initial_sync_completion_sets_coverage_and_invariant(): client, engine = _client() hb(client); finish(client, engine) diff --git a/release/RELEASE_MANIFEST.json b/release/RELEASE_MANIFEST.json index 1ab883d..0e0422e 100644 --- a/release/RELEASE_MANIFEST.json +++ b/release/RELEASE_MANIFEST.json @@ -1,14 +1,14 @@ { "artifact": "RELEASE_MANIFEST", - "GTI_FISCAL_VERSION": "1.3.6", - "generated_at": "2026-09-25T15:43:40+00:00", + "GTI_FISCAL_VERSION": "1.3.7", + "generated_at": "2026-09-27T00:00:00+00:00", "hub": { - "version": "1.3.6", + "version": "1.3.7", "api_compat": "1.0", "engine_version": "0.3.0", "decision_version": "mission-04c-ipi-ex-split" }, - "agent_version": "1.3.6", + "agent_version": "1.3.7", "fiscal_engine": { "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": "0031_enrichment_reevaluate", - "applied_revision": "0031_enrichment_reevaluate", - "applied_revision_source": "fresh + upgrade(0030->0031) + downgrade/upgrade verified on the dev PostgreSQL (schema parity by pg_dump -s); production applies 0031 at deploy", - "count": 31 + "alembic_head_in_repo": "0032_contribuinte_ipi", + "applied_revision": "0032_contribuinte_ipi", + "applied_revision_source": "PENDING REAL DEPLOY - placeholder set equal to repo head for local test coherence; scripts/release_manifest.py must be re-run against the live post-deploy Hub to confirm the real applied_revision before this manifest is trusted", + "count": 32 }, "datasets_active": [ { @@ -89,5 +89,5 @@ "record_count": 11103 } ], - "note": "1.3.6: Hub only (KB global search fix: server-side, lenient, paginated). No Agent change; the published installer remains GTIFiscalAgent-Setup-1.3.5.exe." + "note": "1.3.7: field_contract wired into /v1/products/resolve (post-NCM fiscal completion); Agent heartbeat now reports company_crt and company_contribuinte_ipi (auto-discovered from C000004, migration 0032); Agent now reports CURRENT_REPORT_FIELDS (fiscal-completion fields + classification columns) for role inference. Agent code changed: a new GTIFiscalAgent-Setup-1.3.7 installer is published this release." } \ No newline at end of file diff --git a/release/VERSION b/release/VERSION index 95b25ae..3336003 100644 --- a/release/VERSION +++ b/release/VERSION @@ -1 +1 @@ -1.3.6 +1.3.7 diff --git a/tests/test_release_candidate.py b/tests/test_release_candidate.py index a4352a3..3d174ae 100644 --- a/tests/test_release_candidate.py +++ b/tests/test_release_candidate.py @@ -14,11 +14,11 @@ sys.path.insert(0, str(ROOT)) def test_versions_are_coherent_across_hub_agent_and_packages(): from central.app.version import GTI_FISCAL_VERSION v = (ROOT / "release" / "VERSION").read_text(encoding="utf-8").strip() - assert v == GTI_FISCAL_VERSION == "1.3.6" + assert v == GTI_FISCAL_VERSION == "1.3.7" pep = v.replace("-rc", "rc") for pp in ("central/pyproject.toml", "agent/pyproject.toml"): assert f'version = "{pep}"' in (ROOT / pp).read_text(encoding="utf-8") - assert '"AGENT_VERSION", "1.3.6"' in (ROOT / "agent" / "scripts" / "run_agent_service.py").read_text(encoding="utf-8") + assert '"AGENT_VERSION", "1.3.7"' in (ROOT / "agent" / "scripts" / "run_agent_service.py").read_text(encoding="utf-8") manifest = ROOT / "release" / "RELEASE_MANIFEST.json" if manifest.exists(): m = json.loads(manifest.read_text(encoding="utf-8"))