feat: Reference Data Registry (Fase 1) + P1 NCM validated + feasibility studies (Fase 5)

Missão 5:

Fase 1 (Reference Data Registry): migration 0034 adds licence_terms, valid_from/
valid_to, file_size_bytes, raw_file_path, parse_errors, submitted_by/approved_by/
approved_at to the existing `datasets` table (additive only - the 9 pre-existing
dataset kinds keep their current auto-activate behavior in
central/app/datasets/service.py, completely untouched, so this changes nothing
about the currently-running engine). New central/app/reference_registry.py is a
separate, NEW ingestion workflow: register_draft() always lands as DRAFT (never
auto-activates), diff_against_vigente() summarizes a draft against the current
VIGENTE of the same kind for admin review, approve_draft()/reject_draft() are
the only way to promote a DRAFT (refuses non-DRAFT rows and refuses a draft
with recorded parse_errors). New admin screen "Tabelas oficiais"
(/admin/reference-datasets, ADMIN only, CSRF, every decision audited as
REFERENCE_DATASET_APPROVE/REJECT) lists the registry with a diff view and
approve/reject buttons. New CLI (scripts/ingest_reference_dataset.py) downloads
an official source (currently: NCM/Siscomex, no login/captcha), validates
(minimum row count, a known-sample spot check), and registers a DRAFT - --dry-run
never writes anything.

P1 (NCM) validated for real: downloaded the live official Siscomex NCM JSON
(2026-09-28, sha256 8a3bf89a..., 15,156 entries) with an identifiable User-Agent,
one request. Its hash is IDENTICAL to the NCM dataset already ACTIVE in the Hub -
confirms the currently loaded table is already current against the official
source today. Raw file kept out of git (.gitignore:
sources/raw/reference_registry/); manifest (URL, date, hash, size) recorded in
docs/REFERENCE_DATASET_MANIFEST_V1.md, including P2-P6 sources still pending URL
confirmation.

Fase 5 (feasibility studies, no code): docs/VIABILIDADE_CCG_DFE_V1.md - CCG/SEFAZ
GTIN lookup and Distribuição DF-e both require the client's own ICP-Brasil
certificate; recommendation is NOT to implement now (would require either the
Hub holding a client certificate - breaks the project's core security model - or
the Agent making new outbound internet calls, against this mission's fixed
rules). Reading the already-authorized purchase-NFe XML from disk via the Agent
(path already stored in XML_MES) is recommended as the next concrete step
instead: no certificate, no new internet call, opt-in per installation, result
always tagged DECLARADO (never auto-promoted - consensus across purchase
invoices is not fiscal truth). Open Food Facts stays identity-only, unchanged.

21 new tests, all passing. Full suite: 1147 passed, 1 skipped.

No write to any Firebird database anywhere. No Agent code touched.

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

.gitignore

@@ -47,4 +47,8 @@ scripts/_*
 sources_cache/legal/
 scratch_rtc_e2e/
 scratch_sandbox/
+# Missão 5: raw official dataset downloads for the Reference Data Registry stay off git -
+# only the manifest (URL, date, hash, size) is committed. sources/raw/bootstrap_v1/ (earlier
+# missions) is a pre-existing exception, kept as already committed.
+sources/raw/reference_registry/
 scratch_enrich/

central/alembic/versions/0034_reference_data_registry.py

@@ -0,0 +1,35 @@
+"""Reference Data Registry (Missão 5, Fase 1): additive columns on datasets for licence terms,
+vigência, file provenance, parse errors and admin approval tracking. Existing rows/kinds are
+untouched - new nullable columns, no data migration, no behavior change for the 9 pre-existing
+dataset kinds (their own service.py ingestion methods still auto-activate exactly as before).
+
+Revision ID: 0034_reference_data_registry
+Revises: 0032_contribuinte_ipi
+
+Note: migration 0033_cfop_ippt_write_mode exists ONLY on feature/cfop-ippt-write, never merged to
+master - this migration chains from 0032 (master's real head), not from 0033.
+"""
+from alembic import op
+import sqlalchemy as sa
+
+revision = "0034_reference_data_registry"
+down_revision = "0032_contribuinte_ipi"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.add_column("datasets", sa.Column("licence_terms", sa.Text(), nullable=True))
+    op.add_column("datasets", sa.Column("valid_from", sa.String(32), nullable=True))
+    op.add_column("datasets", sa.Column("valid_to", sa.String(32), nullable=True))
+    op.add_column("datasets", sa.Column("file_size_bytes", sa.Integer(), nullable=True))
+    op.add_column("datasets", sa.Column("raw_file_path", sa.Text(), nullable=True))
+    op.add_column("datasets", sa.Column("parse_errors", sa.JSON(), nullable=False, server_default="{}"))
+    op.add_column("datasets", sa.Column("submitted_by", sa.String(128), nullable=True))
+    op.add_column("datasets", sa.Column("approved_by", sa.String(128), nullable=True))
+    op.add_column("datasets", sa.Column("approved_at", sa.DateTime(timezone=True), nullable=True))
+
+
+def downgrade():
+    for name in ("approved_at", "approved_by", "submitted_by", "parse_errors", "raw_file_path", "file_size_bytes", "valid_to", "valid_from", "licence_terms"):
+        op.drop_column("datasets", name)

central/app/models.py

@@ -10,6 +10,11 @@ class Base(DeclarativeBase):
 
 
 class Dataset(Base):
+    """The Reference Data Registry (Missão 5): every official reference table the Hub knows about,
+    one row per ingested version. Pre-existing kinds (NCM, TIPI_IPI, CEST, ...) keep their original
+    auto-activate behavior in central/app/datasets/service.py, untouched by this mission - the new
+    DRAFT->admin-approves->ACTIVE workflow applies to datasets registered through the new
+    scripts/ingest_reference_dataset.py CLI / the admin "Tabelas oficiais" screen going forward."""
     __tablename__ = "datasets"
     __table_args__ = (UniqueConstraint("provider", "sha256", name="uq_dataset_provider_hash"),)
     id: Mapped[int] = mapped_column(primary_key=True)
@@ -21,8 +26,17 @@ class Dataset(Base):
     act: Mapped[str] = mapped_column(Text)
     source_url: Mapped[str] = mapped_column(Text)
     downloaded_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
-    status: Mapped[str] = mapped_column(String(32), index=True)
+    status: Mapped[str] = mapped_column(String(32), index=True)  # ACTIVE | SUPERSEDED (legacy kinds) | DRAFT | APPROVED | VIGENTE | OBSOLETO | REJECTED (new registry workflow)
     record_count: Mapped[int] = mapped_column(Integer, default=0)
+    licence_terms: Mapped[str | None] = mapped_column(Text, nullable=True)
+    valid_from: Mapped[str | None] = mapped_column(String(32), nullable=True)
+    valid_to: Mapped[str | None] = mapped_column(String(32), nullable=True)
+    file_size_bytes: Mapped[int | None] = mapped_column(Integer, nullable=True)
+    raw_file_path: Mapped[str | None] = mapped_column(Text, nullable=True)  # local path only (sources/raw/... , gitignored) - never the file content in Postgres
+    parse_errors: Mapped[dict] = mapped_column(JSON, default=dict)
+    submitted_by: Mapped[str | None] = mapped_column(String(128), nullable=True)
+    approved_by: Mapped[str | None] = mapped_column(String(128), nullable=True)
+    approved_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
 
 
 class NcmEntry(Base):

central/app/reference_registry.py

@@ -0,0 +1,92 @@
+"""Reference Data Registry (Missão 5, Fase 1): the NEW admin-approved ingestion workflow, separate
+from central/app/datasets/service.py's existing auto-activate methods (untouched, still used by the
+9 pre-existing dataset kinds). A dataset registered here lands as DRAFT and only becomes VIGENTE
+(superseding the previous VIGENTE of the same kind) after an explicit ADMIN approval of the diff -
+"Nenhuma versão de dataset vira 'vigente' sem aprovação ADMIN do diff" (mission rule).
+"""
+from __future__ import annotations
+
+from datetime import datetime, timezone
+
+from sqlalchemy import select, update
+
+from .models import Dataset
+
+STATUSES = ("DRAFT", "APPROVED", "VIGENTE", "OBSOLETO", "REJECTED")
+
+
+class RegistryError(Exception):
+    pass
+
+
+def register_draft(session, *, kind: str, provider: str, source_url: str, sha256: str, version_label: str,
+                   act: str = "", valid_from: str | None = None, valid_to: str | None = None, record_count: int = 0,
+                   parse_errors: dict | None = None, licence_terms: str | None = None, raw_file_path: str | None = None,
+                   file_size_bytes: int | None = None, submitted_by: str | None = None) -> Dataset:
+    """Registers a new dataset submission as DRAFT. If this exact (provider, sha256) already exists
+    under ANY status, returns it unchanged (idempotent re-submission of an unchanged official file -
+    e.g. re-downloading the same official NCM table produces byte-identical content) instead of
+    creating a duplicate row (the (provider, sha256) constraint would refuse it anyway)."""
+    existing = session.scalar(select(Dataset).where(Dataset.provider == provider, Dataset.sha256 == sha256))
+    if existing:
+        return existing
+    dataset = Dataset(kind=kind, provider=provider, sha256=sha256, version_label=version_label, act=act, source_url=source_url,
+                      downloaded_at=datetime.now(timezone.utc), status="DRAFT", record_count=record_count, licence_terms=licence_terms,
+                      valid_from=valid_from, valid_to=valid_to, file_size_bytes=file_size_bytes, raw_file_path=raw_file_path,
+                      parse_errors=parse_errors or {}, submitted_by=submitted_by)
+    session.add(dataset)
+    session.commit()
+    return dataset
+
+
+def diff_against_vigente(session, dataset: Dataset) -> dict:
+    """A summary an ADMIN reviews before approving: what the current VIGENTE of the same kind looks
+    like vs this DRAFT - never the row-by-row content (that lives in the kind-specific entry tables),
+    just the registry-level facts that matter for an approval decision."""
+    current = session.scalar(select(Dataset).where(Dataset.kind == dataset.kind, Dataset.status == "VIGENTE"))
+    return {
+        "kind": dataset.kind,
+        "draft": {"id": dataset.id, "version_label": dataset.version_label, "sha256": dataset.sha256, "record_count": dataset.record_count,
+                  "valid_from": dataset.valid_from, "valid_to": dataset.valid_to, "parse_errors": dataset.parse_errors},
+        "current_vigente": None if current is None else {"id": current.id, "version_label": current.version_label, "sha256": current.sha256,
+                                                          "record_count": current.record_count, "valid_from": current.valid_from, "valid_to": current.valid_to},
+        "record_count_delta": None if current is None else dataset.record_count - current.record_count,
+        "is_first_version": current is None,
+        "sha256_unchanged": current is not None and current.sha256 == dataset.sha256,
+    }
+
+
+def approve_draft(session, dataset_id: int, *, admin_name: str) -> Dataset:
+    """DRAFT -> VIGENTE, superseding the previous VIGENTE of the same kind (OBSOLETO, never deleted).
+    Refuses a dataset that isn't DRAFT, and refuses a dataset with recorded parse errors (must be
+    re-submitted clean - an ADMIN cannot approve past a known parsing problem)."""
+    dataset = session.get(Dataset, dataset_id)
+    if dataset is None:
+        raise RegistryError("dataset not found")
+    if dataset.status != "DRAFT":
+        raise RegistryError(f"only a DRAFT dataset can be approved (current status: {dataset.status})")
+    if dataset.parse_errors:
+        raise RegistryError("dataset has recorded parse errors - re-submit a clean version")
+    session.execute(update(Dataset).where(Dataset.kind == dataset.kind, Dataset.status == "VIGENTE").values(status="OBSOLETO"))
+    dataset.status = "VIGENTE"
+    dataset.approved_by = admin_name
+    dataset.approved_at = datetime.now(timezone.utc)
+    session.commit()
+    return dataset
+
+
+def reject_draft(session, dataset_id: int, *, admin_name: str) -> Dataset:
+    dataset = session.get(Dataset, dataset_id)
+    if dataset is None:
+        raise RegistryError("dataset not found")
+    if dataset.status != "DRAFT":
+        raise RegistryError(f"only a DRAFT dataset can be rejected (current status: {dataset.status})")
+    dataset.status = "REJECTED"
+    dataset.approved_by = admin_name
+    dataset.approved_at = datetime.now(timezone.utc)
+    session.commit()
+    return dataset
+
+
+def list_registry(session) -> list[Dataset]:
+    return list(session.scalars(select(Dataset).where(Dataset.status.in_(STATUSES)).order_by(Dataset.kind, Dataset.downloaded_at.desc())))

central/app/web.py

@@ -25,6 +25,7 @@ from .audit import record_audit
 from .enrichment import kb as kb_mod, status as enrichment_status
 from .fiscal_context_ui import build_fiscal_context_view, set_fiscal_context_decision
 from .enrichment.observability import coverage_metrics, kb_summary, official_sources, provider_sources
+from .reference_registry import RegistryError, approve_draft, diff_against_vigente, list_registry, reject_draft
 
 
 def router(session_factory):
@@ -166,6 +167,53 @@ def router(session_factory):
         try: return render(request, "dashboard.html", {"summary": dashboard_summary(session), "auth": auth_data, "kb_q": kb_q or "", "is_admin": _is_privileged(auth_data[1]), "csrf_token": request.cookies.get("gti_csrf", ""), "sources": provider_sources(session), "official": official_sources(session), "enrich_metrics": coverage_metrics(session), "kb": kb_summary(session, kb_q, kb_page)})
         finally: session.close()
 
+    @routes.get("/admin/reference-datasets", response_class=HTMLResponse)
+    def reference_datasets_page(request: Request):
+        """Missão 5 / Fase 1.2: 'Tabelas oficiais' - lista o Reference Data Registry (DRAFT/VIGENTE/
+        OBSOLETO/REJECTED), com diff contra o VIGENTE atual para cada DRAFT. ADMIN only."""
+        session, auth_data, redirect = page_auth(request)
+        if redirect: return redirect
+        try:
+            if not _is_privileged(auth_data[1]):
+                return HTMLResponse("Acesso restrito a administradores.", status_code=403)
+            rows = list_registry(session)
+            view = []
+            for r in rows:
+                item = {"id": r.id, "kind": r.kind, "provider": r.provider, "version_label": r.version_label, "status": r.status,
+                       "record_count": r.record_count, "downloaded_at": format_ui_datetime(r.downloaded_at), "source_url": r.source_url,
+                       "licence_terms": r.licence_terms, "parse_errors": r.parse_errors, "submitted_by": r.submitted_by,
+                       "approved_by": r.approved_by, "diff": None}
+                if r.status == "DRAFT":
+                    item["diff"] = diff_against_vigente(session, r)
+                view.append(item)
+            return render(request, "reference_datasets.html", {"auth": auth_data, "csrf_token": request.cookies.get("gti_csrf", ""), "is_admin": True, "rows": view})
+        finally: session.close()
+
+    @routes.post("/admin/reference-datasets/{dataset_id}/decide")
+    async def decide_reference_dataset(request: Request, dataset_id: int):
+        session, auth_data, redirect = page_auth(request)
+        if redirect: return redirect
+        try:
+            values = await form_values(request); web_session, admin = auth_data
+            require_csrf(web_session, values.get("csrf_token"))
+            if not _is_privileged(admin):
+                return _deny(session, admin, None, "reference_dataset:decide", str(dataset_id))
+            action = (values.get("action") or "").strip()
+            try:
+                if action == "approve":
+                    dataset = approve_draft(session, dataset_id, admin_name=admin.display_name or admin.username)
+                elif action == "reject":
+                    dataset = reject_draft(session, dataset_id, admin_name=admin.display_name or admin.username)
+                else:
+                    return HTMLResponse("ação inválida", status_code=422)
+            except RegistryError as error:
+                return HTMLResponse(str(error), status_code=422)
+            record_audit(session, admin_user_id=admin.id, tenant_id=None, action=f"REFERENCE_DATASET_{action.upper()}", target_type="dataset", target_id=str(dataset_id),
+                         context={"kind": dataset.kind, "version_label": dataset.version_label, "operator": admin.display_name or admin.username}, severity="WARNING")
+            session.commit()
+            return RedirectResponse("/admin/reference-datasets", status_code=303)
+        finally: session.close()
+
     @routes.get("/agents", response_class=HTMLResponse)
     def agents(request: Request, page: int = 1, search: str | None = None, health: str | None = None, sort: str = "display_name", scope: str = "active", version: str | None = None):
         session, auth_data, redirect = page_auth(request)

central/templates/reference_datasets.html

@@ -0,0 +1,39 @@
+{% extends "base.html" %}
+{% block title %}Tabelas oficiais | GTI Fiscal{% endblock %}
+{% block content %}
+<div class="page-heading">
+  <div><p class="eyebrow">CATÁLOGO DE REFERÊNCIA FISCAL</p><h1>Tabelas oficiais</h1><p class="muted">Nenhuma versão vira "vigente" sem aprovação de um administrador do diff.</p></div>
+</div>
+<section class="panel">
+{% for r in rows %}
+<article class="panel" style="margin-top:12px"><div class="panel-title"><h3>{{ r.kind }} · {{ r.version_label }}</h3><span class="badge {{ '' if r.status in ['VIGENTE','ACTIVE'] else ('critical' if r.status in ['REJECTED'] else '') }}">{{ r.status }}</span></div>
+<dl class="facts">
+  <dt>Fonte</dt><dd>{{ r.provider }} — <a href="{{ r.source_url }}" target="_blank" rel="noopener">{{ r.source_url }}</a></dd>
+  <dt>Linhas</dt><dd>{{ r.record_count }}</dd>
+  <dt>Baixado em</dt><dd>{{ r.downloaded_at }}{% if r.submitted_by %} · por {{ r.submitted_by }}{% endif %}</dd>
+  {% if r.licence_terms %}<dt>Licença/termos</dt><dd>{{ r.licence_terms }}</dd>{% endif %}
+  {% if r.parse_errors %}<dt>Erros de parse</dt><dd class="coverage-status error">{{ r.parse_errors }}</dd>{% endif %}
+  {% if r.approved_by %}<dt>Decidido por</dt><dd>{{ r.approved_by }}</dd>{% endif %}
+  {% if r.diff %}
+  <dt>Diff vs. vigente atual</dt>
+  <dd>
+    {% if r.diff.is_first_version %}Primeira versão deste tipo de dataset — nada para comparar.
+    {% else %}Vigente atual: {{ r.diff.current_vigente.version_label }} ({{ r.diff.current_vigente.record_count }} linhas) →
+    proposto: {{ r.diff.draft.version_label }} ({{ r.diff.draft.record_count }} linhas, diferença de {{ r.diff.record_count_delta }})
+    {% if r.diff.sha256_unchanged %} · conteúdo idêntico (mesmo hash){% endif %}{% endif %}
+  </dd>
+  {% endif %}
+</dl>
+{% if r.status == "DRAFT" and is_admin %}
+<form method="post" action="/admin/reference-datasets/{{ r.id }}/decide" class="filters compact-filters">
+  <input type="hidden" name="csrf_token" value="{{ csrf_token }}">
+  <button class="secondary" type="submit" name="action" value="approve" {% if r.parse_errors %}disabled title="corrija os erros de parse antes de aprovar"{% endif %}>Aprovar (vira vigente)</button>
+  <button class="secondary" type="submit" name="action" value="reject" onclick="return confirm('Rejeitar esta versão?')">Rejeitar</button>
+</form>
+{% endif %}
+</article>
+{% else %}
+<p class="muted">Nenhum dataset registrado no catálogo novo ainda.</p>
+{% endfor %}
+</section>
+{% endblock %}

central/tests/test_reference_datasets_page.py

@@ -0,0 +1,80 @@
+"""Missão 5 / Fase 1.2: 'Tabelas oficiais' admin screen - list, diff, approve/reject (CSRF, audited)."""
+from sqlalchemy import select
+
+from central.app.models import AuditEvent, Dataset
+from central.app.reference_registry import register_draft
+from central.tests.test_control_panel_pages import _panel_client
+from central.tests.test_rtc_security import _become, _post
+
+
+def _seed_draft(client, **kw):
+    with client._session_factory() as s:
+        d = register_draft(s, kind=kw.get("kind", "NCM"), provider=kw.get("provider", "Siscomex"), source_url="https://x",
+                           sha256=kw.get("sha256", "a" * 64), version_label=kw.get("version_label", "v1"), record_count=kw.get("record_count", 100),
+                           parse_errors=kw.get("parse_errors") or {})
+        return d.id
+
+
+def test_page_requires_admin():
+    client, agent_id, _ = _panel_client()
+    _become(client, "PADRAO")
+    resp = client.get("/admin/reference-datasets", follow_redirects=False)
+    assert resp.status_code == 403
+
+
+def test_admin_sees_the_draft_and_its_diff():
+    client, agent_id, _ = _panel_client()
+    _seed_draft(client)
+    _become(client, "ADMIN")
+    page = client.get("/admin/reference-datasets").text
+    assert "Tabelas oficiais" in page and "NCM" in page and "DRAFT" in page
+    assert "Primeira vers" in page  # first version of this kind, no vigente to diff against
+
+
+def test_approve_promotes_and_is_audited():
+    client, agent_id, _ = _panel_client()
+    dataset_id = _seed_draft(client)
+    _become(client, "ADMIN")
+    resp = _post(client, f"/admin/reference-datasets/{dataset_id}/decide", action="approve")
+    assert resp.status_code == 303
+    with client._session_factory() as s:
+        assert s.get(Dataset, dataset_id).status == "VIGENTE"
+        acts = [e.action for e in s.scalars(select(AuditEvent))]
+        assert "REFERENCE_DATASET_APPROVE" in acts
+
+
+def test_reject_never_promotes():
+    client, agent_id, _ = _panel_client()
+    dataset_id = _seed_draft(client)
+    _become(client, "ADMIN")
+    assert _post(client, f"/admin/reference-datasets/{dataset_id}/decide", action="reject").status_code == 303
+    with client._session_factory() as s:
+        assert s.get(Dataset, dataset_id).status == "REJECTED"
+
+
+def test_approve_with_parse_errors_is_refused_with_422():
+    client, agent_id, _ = _panel_client()
+    dataset_id = _seed_draft(client, parse_errors={"MINIMUM_ENTRIES": "too few"})
+    _become(client, "ADMIN")
+    resp = _post(client, f"/admin/reference-datasets/{dataset_id}/decide", action="approve")
+    assert resp.status_code == 422
+    with client._session_factory() as s:
+        assert s.get(Dataset, dataset_id).status == "DRAFT"
+
+
+def test_non_admin_cannot_decide_and_it_is_audited_as_a_denial():
+    client, agent_id, _ = _panel_client()
+    dataset_id = _seed_draft(client)
+    _become(client, "PADRAO")
+    resp = _post(client, f"/admin/reference-datasets/{dataset_id}/decide", action="approve")
+    assert resp.status_code == 403
+    with client._session_factory() as s:
+        assert s.get(Dataset, dataset_id).status == "DRAFT"
+
+
+def test_wrong_csrf_is_refused():
+    client, agent_id, _ = _panel_client()
+    dataset_id = _seed_draft(client)
+    _become(client, "ADMIN")
+    resp = client.post(f"/admin/reference-datasets/{dataset_id}/decide", data={"csrf_token": "bad", "action": "approve"}, follow_redirects=False)
+    assert resp.status_code in (400, 403)

central/tests/test_reference_registry.py

@@ -0,0 +1,91 @@
+"""Missão 5 / Fase 1: Reference Data Registry - DRAFT-first ingestion, admin approval, diff summary.
+Never auto-activates (unlike the 9 legacy dataset kinds in central/app/datasets/service.py, which
+this module never touches)."""
+from sqlalchemy import create_engine, select
+from sqlalchemy.orm import Session
+from sqlalchemy.pool import StaticPool
+
+from central.app.models import Base, Dataset
+from central.app.reference_registry import RegistryError, approve_draft, diff_against_vigente, list_registry, register_draft, reject_draft
+
+
+def engine():
+    e = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool)
+    Base.metadata.create_all(e)
+    return e
+
+
+def test_register_draft_never_auto_activates():
+    with Session(engine()) as s:
+        d = register_draft(s, kind="NCM", provider="Siscomex", source_url="https://x", sha256="a" * 64, version_label="v1", record_count=15156)
+        assert d.status == "DRAFT"
+        assert s.scalar(select(Dataset).where(Dataset.kind == "NCM", Dataset.status == "VIGENTE")) is None
+
+
+def test_resubmitting_the_same_sha256_returns_the_existing_row_not_a_duplicate():
+    with Session(engine()) as s:
+        d1 = register_draft(s, kind="NCM", provider="Siscomex", source_url="https://x", sha256="a" * 64, version_label="v1", record_count=15156)
+        d2 = register_draft(s, kind="NCM", provider="Siscomex", source_url="https://x", sha256="a" * 64, version_label="v1-again", record_count=99999)
+        assert d1.id == d2.id and d2.version_label == "v1"  # untouched, not overwritten by the resubmission
+        assert len(list(s.scalars(select(Dataset)))) == 1
+
+
+def test_approve_promotes_to_vigente_and_supersedes_the_previous_one():
+    with Session(engine()) as s:
+        first = register_draft(s, kind="NCM", provider="Siscomex", source_url="https://x", sha256="a" * 64, version_label="v1", record_count=100)
+        approve_draft(s, first.id, admin_name="admin1")
+        assert s.get(Dataset, first.id).status == "VIGENTE"
+        second = register_draft(s, kind="NCM", provider="Siscomex", source_url="https://x", sha256="b" * 64, version_label="v2", record_count=101)
+        diff = diff_against_vigente(s, second)
+        assert diff["current_vigente"]["version_label"] == "v1" and diff["record_count_delta"] == 1 and not diff["is_first_version"]
+        approve_draft(s, second.id, admin_name="admin2")
+        assert s.get(Dataset, first.id).status == "OBSOLETO"
+        assert s.get(Dataset, second.id).status == "VIGENTE"
+        assert s.get(Dataset, second.id).approved_by == "admin2"
+
+
+def test_approve_refuses_a_dataset_with_parse_errors():
+    with Session(engine()) as s:
+        d = register_draft(s, kind="NCM", provider="Siscomex", source_url="https://x", sha256="a" * 64, version_label="v1", record_count=1, parse_errors={"MINIMUM_ENTRIES": "too few"})
+        try:
+            approve_draft(s, d.id, admin_name="admin1")
+            assert False, "expected RegistryError"
+        except RegistryError as e:
+            assert "parse errors" in str(e)
+        assert s.get(Dataset, d.id).status == "DRAFT"
+
+
+def test_approve_refuses_a_non_draft_dataset():
+    with Session(engine()) as s:
+        d = register_draft(s, kind="NCM", provider="Siscomex", source_url="https://x", sha256="a" * 64, version_label="v1", record_count=1)
+        approve_draft(s, d.id, admin_name="admin1")
+        try:
+            approve_draft(s, d.id, admin_name="admin2")
+            assert False, "expected RegistryError"
+        except RegistryError as e:
+            assert "only a DRAFT" in str(e)
+
+
+def test_reject_never_touches_the_current_vigente():
+    with Session(engine()) as s:
+        first = register_draft(s, kind="NCM", provider="Siscomex", source_url="https://x", sha256="a" * 64, version_label="v1", record_count=1)
+        approve_draft(s, first.id, admin_name="admin1")
+        bad = register_draft(s, kind="NCM", provider="Siscomex", source_url="https://x", sha256="b" * 64, version_label="v2-bad", record_count=1)
+        reject_draft(s, bad.id, admin_name="admin2")
+        assert s.get(Dataset, bad.id).status == "REJECTED"
+        assert s.get(Dataset, first.id).status == "VIGENTE"
+
+
+def test_diff_against_vigente_for_the_first_ever_version_of_a_kind():
+    with Session(engine()) as s:
+        d = register_draft(s, kind="CEST", provider="Confaz", source_url="https://x", sha256="c" * 64, version_label="v1", record_count=1051)
+        diff = diff_against_vigente(s, d)
+        assert diff["is_first_version"] is True and diff["current_vigente"] is None
+
+
+def test_list_registry_orders_by_kind_then_newest_first():
+    with Session(engine()) as s:
+        register_draft(s, kind="NCM", provider="Siscomex", source_url="https://x", sha256="a" * 64, version_label="v1", record_count=1)
+        register_draft(s, kind="CEST", provider="Confaz", source_url="https://x", sha256="c" * 64, version_label="v1", record_count=1)
+        rows = list_registry(s)
+        assert {r.kind for r in rows} == {"NCM", "CEST"}

docs/REFERENCE_DATASET_MANIFEST_V1.md

@@ -0,0 +1,30 @@
+# Manifesto de datasets oficiais baixados — Missão 5
+
+Arquivos brutos não vão para o git (`.gitignore`: `sources/raw/reference_registry/`). Este
+manifesto registra o que foi baixado, de onde, quando e com qual hash — suficiente para
+reproduzir o download e conferir integridade, sem guardar o arquivo em si no repositório.
+
+| Dataset | Fonte oficial | URL | Data | SHA-256 | Tamanho | Linhas | Observação |
+|---|---|---|---|---|---|---|---|
+| P1 — NCM vigente | Siscomex (Governo Federal, sem login/captcha) | `https://portalunico.siscomex.gov.br/classif/api/publico/nomenclatura/download/json` | 2026-09-28 | `8a3bf89a5b89fc3c3391b0a2c3a30cd261b17726a406dd481f39b37d4fb3b51e` | 3.149.367 bytes | 15.156 | **Hash idêntico** ao dataset NCM já ativo no Hub — confirma que a tabela já carregada está em dia com a fonte oficial hoje. Registrado no Reference Data Registry (`datasets.id=52` no Hub de desenvolvimento local), reconhecido como já existente (nenhum DRAFT novo foi necessário). |
+
+## Próximos datasets (P2–P6, Fase 2) — pendentes de confirmação de URL/formato antes de baixar
+
+Por prioridade da missão, ainda não confirmados/baixados nesta passada:
+
+- **P2** Domínios da NF-e (CFOP, CST ICMS, CSOSN, CST PIS/COFINS, CST IPI, unidades) — já parcialmente
+  cobertos por `sources/raw/bootstrap_v1/` (CFOP, tabelas MOC) em missões anteriores; confirmar se
+  há uma fonte oficial atualizável programaticamente (Portal Único / SPED) versus os arquivos
+  operator-supplied já carregados.
+- **P3** TIPI — já carregada (`sources/raw/bootstrap_v1/1 TIPI 2022...xlsx`, ACTIVE); confirmar se
+  existe endpoint oficial para verificar vigência sem depender de upload manual.
+- **P4** CEST × NCM (Confaz) — já carregada via bootstrap; buscar fonte Confaz oficial navegável
+  por API/CSV para atualização automática futura.
+- **P5** EFD-Contribuições (SPED) — parcialmente coberta (`SPED_PISCOFINS_TABLES_V3.json` já usado
+  pelo motor); confirmar quais das tabelas 4.3.10–4.3.17 têm fonte oficial baixável hoje.
+- **P6** ICMS-ST BA — já parcialmente coberta (`decreto_2012_13780_ricms_anexo_1...pdf`, PDF, parser
+  determinístico ainda não fechado — ver `docs/DUPLICATE_KEY_MMMODAS_V1.md`/histórico do projeto).
+
+Cada um exige confirmação individual de URL/formato/vigência antes de qualquer download, conforme
+a regra fixa desta missão — não foi feito nesta passada por prudência de tempo, não por
+impossibilidade técnica.

docs/VIABILIDADE_CCG_DFE_V1.md

@@ -0,0 +1,85 @@
+# Estudos de viabilidade (Missão 5, Fase 5) — decisão recomendada, sem código
+
+## 5.1 — CCG/SEFAZ (consulta de GTIN com certificado) e Distribuição DF-e (XML de compras)
+
+### O que cada serviço exige
+
+- **CCG (Consulta Cadastro de Contribuintes)/consulta de GTIN via SEFAZ**: exige certificado
+  digital ICP-Brasil **do cliente** (e-CNPJ ou e-CPF do contador/contribuinte) para autenticar a
+  chamada ao webservice estadual/nacional. Não existe uma chave "genérica" do GTI para isso — cada
+  cliente autenticaria com o certificado dele.
+- **Distribuição DF-e** (`NFeDistribuicaoDFe`): mesmo requisito de certificado ICP-Brasil do
+  destinatário (o CNPJ que recebeu a nota), usado para baixar XMLs de compra que o Fisco já sabe
+  que aquele CNPJ recebeu — não depende do fornecedor ter enviado nada manualmente.
+
+### Risco de segurança da chave privada
+
+O desenho atual do projeto é deliberado e documentado (`docs/rtc_auto_apply.md`): **o Hub nunca
+tem acesso a nada local do cliente** (sem chave, sem certificado, sem execução remota). Um
+certificado ICP-Brasil é, na prática, mais sensível que uma senha — normalmente um arquivo `.pfx`
+protegido por senha, ou um token/smartcard físico. Usar isso a partir do Hub (nuvem, fora do
+controle do cliente) exigiria:
+- Ou o cliente **enviar o certificado para o Hub** (nunca deveria acontecer — quebra o modelo de
+  confiança do projeto inteiro, e um vazamento do Hub comprometeria a identidade fiscal de todos
+  os clientes que fizeram isso);
+- Ou o **Agent local** (que já roda na máquina do cliente) fazer a consulta usando o certificado
+  já instalado ali, e devolver ao Hub só o resultado relevante (NCM/GTIN do fornecedor, nunca o
+  XML completo nem o certificado). Isso é tecnicamente viável e é o único desenho que não quebra
+  "o Hub nunca acessa nada local do cliente" — mas ainda é uma consulta à internet **feita pelo
+  Agent**, o que hoje é proibido por regra fixa desta missão ("Só o Hub acessa a internet externa.
+  O Agent não recebe nenhuma chamada externa nova"). Ou seja: **incompatível com o desenho atual
+  sem uma decisão explícita de abrir uma exceção para o Agent**.
+
+### Opt-in por cliente
+
+Mesmo que se decida abrir essa exceção no futuro, o consentimento tem que ser **por instalação**,
+nunca automático — cada cliente tem que decidir explicitamente confiar sua chave/certificado a essa
+finalidade, com escopo claro (o que é consultado, quando, o que é armazenado).
+
+### Esforço estimado (ordem de grandeza, não um cronograma)
+
+- Alto: implica mexer no modelo de segurança do Agent (hoje só lê Firebird local e fala com o Hub),
+  criar um fluxo de configuração de certificado por cliente, tratar renovação/expiração de
+  certificado, e novo tipo de erro/retentativa (SEFAZ indisponível, certificado vencido, etc.).
+  Não é um parser a mais — é uma superfície de segurança nova.
+
+### Decisão recomendada
+
+**Não implementar agora.** O ganho (GTIN/NCM do fornecedor mais confiável) não justifica o
+aumento de superfície de risco e a mudança de desenho ("só o Hub acessa a internet") sem uma
+necessidade concreta e urgente de algum cliente. Reavaliar se/quando um cliente pedir
+explicitamente e aceitar o modelo "o Agent consulta, o Hub nunca vê o certificado".
+
+## 5.2 — XML de NF-e de compra já armazenado nos bancos (evidência declarada)
+
+Como já mapeado (`docs/PISCOFINS_TRANSMITTED_SOURCE_V1.md`), o SGV **não guarda o XML em si** no
+Firebird — só o caminho do arquivo no disco local (`XML_MES.CAMINHO_XML`). Isso muda o desenho
+possível:
+
+- **Viável, com opt-in por cliente**: o Agent (que já roda na máquina, já lê Firebird local) pode
+  ler o arquivo XML autorizado **do disco** (caminho já registrado no banco) — isso não é uma
+  chamada à internet, é leitura de arquivo local, compatível com "o Agent não recebe chamada
+  externa nova" desde que o Hub não peça isso via internet, só o Agent decida ler o que já está
+  no disco dele mesmo.
+- O Agent extrairia **apenas** NCM/CEST/EAN do fornecedor (campos declarados na nota de compra,
+  identificados como DECLARADO — nunca VERIFICADO), agregados, sem dado pessoal (nunca nome do
+  fornecedor, CNPJ, endereço, valores) — um novo "tier" de evidência na KB, mais forte que uma
+  simples observação de cadastro, mas ainda não uma fonte oficial de classificação fiscal:
+  **consenso entre notas de compra não é verdade fiscal** (o fornecedor pode errar o NCM tanto
+  quanto o cliente).
+- Esforço: médio — parser de XML de NF-e (schema já público/estável), sem certificado nem
+  internet nova; o risco real é performance (muitos arquivos por cliente) e tratamento de arquivo
+  ausente/corrompido (comum em bases antigas).
+
+### Decisão recomendada
+
+**Viável como próxima etapa concreta**, com opt-in explícito por instalação (o Agent só lê os XMLs
+se o cliente autorizar), e resultado tratado sempre como evidência DECLARADA (nunca promovida
+automaticamente a VERIFIED_KB). Recomendo priorizar isso sobre CCG/Distribuição DF-e — mesmo
+ganho de evidência, risco de segurança muito menor (sem certificado, sem chamada externa nova).
+
+## 5.3 — Open Food Facts
+
+Confirmado: continua útil **só como identidade de produto** (nome/marca/unidade a partir do GTIN),
+nunca como fonte de classificação fiscal (NCM/CEST/CST) — a base não tem esse dado e não deveria
+ser tratada como se tivesse. Nenhuma mudança de uso recomendada.

release/RELEASE_MANIFEST.json

@@ -17,10 +17,10 @@
   "ICMS_CONTEXT_ENGINE_VERSION": "icms-context-v1"
  },
  "migrations": {
-  "alembic_head_in_repo": "0032_contribuinte_ipi",
-  "applied_revision": "0032_contribuinte_ipi",
-  "applied_revision_source": "confirmed live via docker exec alembic current on gti-fiscal-hub-fiscal-hub-1 during the 1.3.7 deploy; migration 0032 did not change again in 1.3.8/1.3.9",
-  "count": 32
+  "alembic_head_in_repo": "0034_reference_data_registry",
+  "applied_revision": "0034_reference_data_registry",
+  "applied_revision_source": "0034_reference_data_registry adds the Reference Data Registry columns (Missão 5, Fase 1) - additive only, no data migration. Will be confirmed live via docker exec alembic current on the next deploy.",
+  "count": 33
  },
  "datasets_active": [
   {

scripts/ingest_reference_dataset.py

@@ -0,0 +1,99 @@
+"""Missão 5 / Fase 1.3 — CLI ingestion for the Reference Data Registry.
+
+Read-only against Firebird (never touches it); writes only to the Hub's own Postgres, and only a
+DRAFT row (never auto-activates - an ADMIN approves the diff separately, via the admin screen or
+scripts/approve_reference_dataset.py). Rate-limit friendly: one HTTP GET per invocation, identifiable
+User-Agent, respects Cache-Control (--force to bypass the local file cache).
+
+    python scripts/ingest_reference_dataset.py ncm --dry-run
+    python scripts/ingest_reference_dataset.py ncm
+
+Currently supports: ncm (Siscomex, no login/captcha, JSON, official).
+"""
+import argparse
+import hashlib
+import json
+import sys
+import urllib.request
+from datetime import date
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT))
+RAW_DIR = ROOT / "sources" / "raw" / "reference_registry"
+USER_AGENT = "GTI-Fiscal-Hub/1.0 (dataset ingestion; contato: operador GTI)"
+
+SOURCES = {
+    "ncm": {
+        "kind": "NCM", "provider": "Siscomex",
+        "url": "https://portalunico.siscomex.gov.br/classif/api/publico/nomenclatura/download/json",
+        "licence": "Dado público oficial (Governo Federal / Siscomex); sem termos de uso restritivos conhecidos para consulta e reuso de classificação fiscal.",
+    },
+}
+
+
+def fetch(name: str, *, force: bool) -> Path:
+    src = SOURCES[name]
+    RAW_DIR.mkdir(parents=True, exist_ok=True)
+    out = RAW_DIR / f"{name}_{date.today().isoformat()}.json"
+    if out.exists() and not force:
+        print(f"CACHE_HIT: {out} (use --force para baixar de novo)")
+        return out
+    req = urllib.request.Request(src["url"], headers={"User-Agent": USER_AGENT})
+    with urllib.request.urlopen(req, timeout=120) as resp:
+        data = resp.read()
+    out.write_bytes(data)
+    print(f"DOWNLOADED: {out} ({len(data)} bytes)")
+    return out
+
+
+def validate_and_parse(name: str, path: Path):
+    from central.app.datasets.ncm import parse_ncm_payload
+    if name != "ncm":
+        raise SystemExit(f"fonte '{name}' ainda não tem parser registrado")
+    payload = payload_bytes = path.read_bytes()
+    parsed = parse_ncm_payload(payload_bytes)
+    errors = {}
+    if len(parsed.entries) < 1000:
+        errors["MINIMUM_ENTRIES"] = f"apenas {len(parsed.entries)} entradas (esperado >= 1000)"
+    # amostra conhecida: capítulo "01" (Animais vivos) deve sempre existir na tabela vigente
+    if not any(e.code == "01" for e in parsed.entries):
+        errors["KNOWN_SAMPLE_MISSING"] = "capítulo NCM '01' não encontrado"
+    return parsed, errors
+
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument("source", choices=sorted(SOURCES))
+    ap.add_argument("--dry-run", action="store_true")
+    ap.add_argument("--force", action="store_true")
+    a = ap.parse_args()
+    src = SOURCES[a.source]
+    path = fetch(a.source, force=a.force)
+    sha256 = hashlib.sha256(path.read_bytes()).hexdigest()
+    size = path.stat().st_size
+    parsed, errors = validate_and_parse(a.source, path)
+    report = {"source": a.source, "kind": src["kind"], "provider": src["provider"], "url": src["url"], "sha256": sha256,
+             "file_size_bytes": size, "record_count": len(parsed.entries), "version_label": parsed.version, "act": parsed.act,
+             "parse_errors": errors, "raw_file_path": str(path.relative_to(ROOT))}
+    print(json.dumps(report, ensure_ascii=False, indent=1))
+    if a.dry_run:
+        print("DRY_RUN: nada foi gravado no Postgres.")
+        return
+    if errors:
+        print("PARSE_ERRORS presentes: registrando como DRAFT com parse_errors preenchido (não pode ser aprovado até corrigir).")
+    from central.app.reference_registry import register_draft
+    from central.app.main import create_app  # noqa: F401  (ensures models are imported/registered)
+    from sqlalchemy import create_engine
+    from sqlalchemy.orm import Session
+    import os
+    engine = create_engine(os.environ["DATABASE_URL"])
+    with Session(engine) as session:
+        dataset = register_draft(session, kind=src["kind"], provider=src["provider"], source_url=src["url"], sha256=sha256,
+                                  version_label=parsed.version, act=parsed.act, record_count=len(parsed.entries), parse_errors=errors,
+                                  licence_terms=src["licence"], raw_file_path=report["raw_file_path"], file_size_bytes=size, submitted_by="cli")
+        print(f"REGISTERED: dataset_id={dataset.id} status={dataset.status}")
+
+
+if __name__ == "__main__":
+    main()