GTI_GERENCIADOR_FISCAL/fiscal/docs/superpowers/plans/2026-09-17-gti-fiscal-hub-mission-02.md

GTI Fiscal Hub V1 Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: Build and exercise a Docker-hosted FastAPI fiscal hub, official local NCM registry, and read-only Windows-ready Firebird agent for 20 real products.

Architecture: A modular FastAPI monolith owns versioned datasets, classification, determination placeholders, cache, and audit in PostgreSQL. A separate agent uses the existing Firebird read-only boundary plus SQLite outbox state to submit product batches and create comparison artifacts, without any ERP writes.

Tech Stack: Python 3.12, FastAPI, Pydantic, SQLAlchemy, Alembic, psycopg, PostgreSQL 16, httpx, Docker Compose, SQLite, pytest, fdb==2.0.4.

Global Constraints

• The only canonical NCM source is https://portalunico.siscomex.gov.br/classif/api/publico/nomenclatura/download/json.
• Verify the downloaded JSON root has Data_Ultima_Atualizacao_NCM, Ato, and Nomenclaturas; each nomenclature must contain Codigo, Descricao, Data_Inicio, and Data_Fim.
• Store raw-payload SHA-256, provenance, record count, and version metadata; preserve prior dataset versions.
• Promote a dataset atomically from READY to ACTIVE while changing the former active version to SUPERSEDED in one transaction.
• No paid APIs, LLMs, photos, BLOB reads, Firebird DML/DDL, ERP schema changes, or Firebird writes.
• Central PostgreSQL is Hub-only; the agent communicates only through authenticated HTTP.
• Every response contains engine_version, dataset_versions, and decision_version.
• NCM_VALIDATED does not authorize fiscal-field writes; can_auto_apply is always false in Mission 02.
• Use the reference fingerprint 1198f89897532b643439f06b729e93b189018f56e7d478e95095f8f56977005b as REFERENCE_SCHEMA_V1.
• Preserve existing worktree changes. Do not configure Git identity or create a commit unless an identity already exists.

Planned File Structure

• central/app/config.py: environment-backed Hub settings.
• central/app/db.py: SQLAlchemy engine/session lifecycle.
• central/app/models.py: PostgreSQL ORM records.
• central/app/schemas.py: HTTP request and response contracts.
• central/app/datasets/ncm.py: verified source parsing, import validation, and normalization.
• central/app/datasets/service.py: dataset lifecycle and atomic promotion.
• central/app/classification.py: exact NCM validation and deterministic candidate ranking.
• central/app/providers.py: fiscal integration statuses and RTC provider interface.
• central/app/auth.py: bearer token verification and tenant/agent binding.
• central/app/api.py: versioned routes and dependencies.
• central/app/main.py: application construction and health checks.
• central/alembic/: migration environment and initial schema migration.
• central/scripts/import_ncm.py: Hub NCM update command.
• agent/src/gti_fiscal_agent/: Firebird reader, local state/outbox, client, and runner.
• agent/scripts/run_agent.py: CLI for the 20-product run.
• central/tests/ and agent/tests/: focused unit/integration tests.
• docker-compose.yml, central/Dockerfile, .env.example: development operation.
• docs/MISSION_02_PRODUCT_RESULTS.json, docs/MISSION_02_REPORT.md, and docs/FISCAL_DATASETS.md: proof and source documentation.

---

Task 1: Create Hub, Agent, and Docker Foundations

Files:

• Create: central/pyproject.toml
• Create: central/__init__.py
• Create: central/app/__init__.py
• Create: central/app/config.py
• Create: central/app/main.py
• Create: central/Dockerfile
• Create: docker-compose.yml
• Modify: .env.example
• Test: central/tests/test_health.py

Interfaces:

• Produces Settings, create_app() -> FastAPI, and a Compose service named fiscal-hub.

• Consumes no new internal components.

• Step 1: Write the failing health test

from fastapi.testclient import TestClient
from central.app.main import create_app

def test_health_reports_database_dependency():
    client = TestClient(create_app(database_probe=lambda: True))
    assert client.get("/health").json() == {"status": "ok"}

• Step 2: Run the focused test to verify it fails

Run: python -m pytest central/tests/test_health.py -q

Expected: import failure because central.app.main does not exist.

• Step 3: Implement the minimal configuration and health application

def create_app(database_probe: Callable[[], bool] = database_is_available) -> FastAPI:
    app = FastAPI(title="GTI Fiscal Hub", version="0.1.0")
    @app.get("/health")
    def health():
        if not database_probe():
            raise HTTPException(status_code=503, detail="postgres unavailable")
        return {"status": "ok"}
    return app

Create Compose services for PostgreSQL 16 and fiscal-hub, each with a healthcheck. Configure DATABASE_URL, HUB_PORT, AGENT_DEV_TOKEN, NCM_DATASET_URL, DATASET_STORAGE_PATH, LOG_LEVEL, and HTTP_TIMEOUT_SECONDS in .env.example without values that are real secrets.

• Step 4: Run focused tests and Compose validation

Run: python -m pytest central/tests/test_health.py -q

Expected: PASS.

Run: docker compose config

Expected: rendered postgres and fiscal-hub services with healthchecks.

• Step 5: Commit if Git identity exists

Run: git config user.name

If it exits nonzero, record GIT_IDENTITY_NOT_CONFIGURED and do not commit. Otherwise: git add central docker-compose.yml .env.example && git commit -m "feat: scaffold fiscal hub services".

Task 2: Add PostgreSQL Schema and Dataset Lifecycle

Files:

• Create: central/app/db.py
• Create: central/app/models.py
• Create: central/alembic.ini
• Create: central/alembic/env.py
• Create: central/alembic/versions/0001_initial.py
• Create: central/app/datasets/service.py
• Test: central/tests/test_dataset_lifecycle.py

Interfaces:

• Produces DatasetService.create_pending(), mark_ready(), and promote(dataset_id).

• Produces ORM tables datasets, ncm_entries, tenants, agents, classification_requests, classification_results, and classification_cache.

• promote() is consumed by the importer in Task 3.

• Step 1: Write lifecycle tests

def test_promotion_supersedes_previous_active_in_one_commit(session):
    old = datasets.create_ready(session, version="old", sha256="a" * 64)
    datasets.promote(session, old.id)
    new = datasets.create_ready(session, version="new", sha256="b" * 64)
    datasets.promote(session, new.id)
    assert datasets.status(session, old.id) == "SUPERSEDED"
    assert datasets.status(session, new.id) == "ACTIVE"

• Step 2: Run lifecycle test to verify failure

Run: python -m pytest central/tests/test_dataset_lifecycle.py -q

Expected: import failure for DatasetService.

• Step 3: Implement models, migration, and transaction-scoped promotion

Define a unique dataset identity on (provider, sha256). Scope ncm_entries to dataset_id with an index on (dataset_id, code). Store only token_hash in agents. In promote, lock current active NCM dataset rows with SELECT ... FOR UPDATE, verify target is READY, update old active rows to SUPERSEDED, update target to ACTIVE, and commit once. Roll back on every exception.

• Step 4: Run migration and tests against PostgreSQL

Run: docker compose up -d postgres

Run: docker compose exec fiscal-hub alembic upgrade head

Run: python -m pytest central/tests/test_dataset_lifecycle.py -q

Expected: migration succeeds; lifecycle test passes.

• Step 5: Commit if allowed

Run the Task 1 identity check. If configured: git add central && git commit -m "feat: add fiscal hub persistence".

Task 3: Implement Official NCM Download, Validation, and Import

Files:

• Create: central/app/datasets/ncm.py
• Create: central/scripts/import_ncm.py
• Create: central/tests/fixtures/ncm_official_sample.json
• Test: central/tests/test_ncm_import.py

Interfaces:

• Produces parse_ncm_payload(payload: bytes) -> NcmPayload, import_ncm(session, payload, source_url, downloaded_at) -> Dataset.

• NcmPayload contains version_label, act, and entries: list[NcmEntryInput].

• Consumed by the import CLI and API dataset status route.

• Step 1: Write parser and idempotency tests using the verified source shape

def test_parser_accepts_verified_siscomex_shape():
    parsed = parse_ncm_payload(b'{"Data_Ultima_Atualizacao_NCM":"Vigente em 17/09/2026","Ato":"Resolução Gecex nº 926/2026","Nomenclaturas":[{"Codigo":"08043000","Descricao":"Abacaxis.","Data_Inicio":"01/04/2022","Data_Fim":"31/12/9999"}]}')
    assert parsed.entries[0].code == "08043000"

def test_import_same_sha_returns_existing_dataset(session):
    first = import_ncm(session, payload, SOURCE_URL, now)
    second = import_ncm(session, payload, SOURCE_URL, now)
    assert first.id == second.id

• Step 2: Run tests to verify failure

Run: python -m pytest central/tests/test_ncm_import.py -q

Expected: import failure for parse_ncm_payload.

• Step 3: Implement strict parser and staged import

Calculate hashlib.sha256(payload).hexdigest() before parsing. Require an object root with the known Data_Ultima_Atualizacao_NCM, Ato, and nonempty Nomenclaturas members; tolerate additional metadata but reject missing or incorrectly typed known members. Require string Codigo, Descricao, Data_Inicio, and Data_Fim per entry. Normalize code by removing non-digits; retain all hierarchy rows but validate product NCMs only against eight-digit entries. Reject duplicate normalized (code, valid_from, valid_to) values and implausible totals below 1,000 entries. Insert the dataset as PENDING, bulk-insert all entries, validate inserted count equals parsed count, mark READY, then call Task 2 promotion. Never delete rows from a prior dataset.

The CLI fetches with httpx, requires a successful response, writes the raw payload under the configured dataset storage path using its SHA-256 filename, then invokes the transactional importer.

• Step 4: Run importer tests and actual source import

Run: python -m pytest central/tests/test_ncm_import.py -q

Expected: PASS.

Run: docker compose exec fiscal-hub python -m central.scripts.import_ncm

Expected: one ACTIVE NCM dataset with recorded source URL, SHA-256, update label, act, downloaded timestamp, and count.

• Step 5: Commit if allowed

Run the Task 1 identity check. If configured: git add central && git commit -m "feat: import versioned official NCM data".

Task 4: Add Deterministic Product Classification and Provider Boundaries

Files:

• Create: central/app/classification.py
• Create: central/app/providers.py
• Test: central/tests/test_classification.py
• Test: central/tests/test_providers.py

Interfaces:

• Produces normalize_code(value) -> str, normalize_description(value) -> str, and Classifier.classify(request) -> ClassificationDecision.

• Produces RtcCalculatorProvider.determine(context) -> ProviderResult and NotConfiguredRtcCalculator.

• Consumed by API routes in Task 5.

• Step 1: Write exact-validation and candidate-ranking tests

def test_valid_existing_ncm_is_validated_without_reclassification(repository):
    decision = Classifier(repository).classify(product(current_ncm="0804.30.00", description="anything"))
    assert decision.status == "NCM_VALIDATED"
    assert decision.classification.ncm == "08043000"
    assert decision.can_auto_apply is False

def test_unmatched_description_is_not_invented(repository):
    decision = Classifier(repository).classify(product(current_ncm=None, description="RISOLE DE CAMARAO"))
    assert decision.status in {"HIGH_CONFIDENCE_CANDIDATE", "AMBIGUOUS", "NOT_FOUND"}
    assert decision.can_auto_apply is False

• Step 2: Run tests to verify failure

Run: python -m pytest central/tests/test_classification.py central/tests/test_providers.py -q

Expected: import failure for Classifier.

• Step 3: Implement normalizers, ranking, and NOT_CONFIGURED RTC adapter

Normalize descriptions with Unicode decomposition, accent removal, uppercase conversion, punctuation-to-space conversion, and collapsed whitespace. Rank candidates with token overlap between normalized description tokens and active NCM descriptions; only return a candidate when its score is at least 0.90 and exceeds the second candidate by 0.15. Otherwise return AMBIGUOUS or NOT_FOUND. Leave CEST and fiscal determination fields null with reason codes such as FISCAL_DETERMINATION_NOT_CONFIGURED.

Implement the RTC protocol to return {status: "NOT_CONFIGURED", values: {}}; do not call an external service.

• Step 4: Run focused tests

Run: python -m pytest central/tests/test_classification.py central/tests/test_providers.py -q

Expected: PASS.

• Step 5: Commit if allowed

Run the Task 1 identity check. If configured: git add central && git commit -m "feat: add deterministic fiscal classification".

Task 5: Implement Authenticated, Cached /v1 API

Files:

• Create: central/app/schemas.py
• Create: central/app/auth.py
• Create: central/app/api.py
• Modify: central/app/main.py
• Test: central/tests/test_api.py

Interfaces:

• Produces POST /v1/validate/ncm, POST /v1/classify, POST /v1/classify/batch, and GET /v1/datasets/status.

• Consumes Classifier, DatasetService, and a bearer-token AgentPrincipal.

• Returns ClassificationResponse containing provenance versions and can_auto_apply=False.

• Step 1: Write API auth, isolation, cache, and idempotency tests

def test_classify_requires_agent_token(client):
    assert client.post("/v1/classify", json=payload()).status_code == 401

def test_same_authenticated_request_uses_cache(client, token):
    first = client.post("/v1/classify", headers={"Authorization": f"Bearer {token}"}, json=payload())
    second = client.post("/v1/classify", headers={"Authorization": f"Bearer {token}"}, json=payload())
    assert first.json()["request_fingerprint"] == second.json()["request_fingerprint"]
    assert second.json()["provenance"]["cache_hit"] is True

• Step 2: Run API tests to verify failure

Run: python -m pytest central/tests/test_api.py -q

Expected: import failure for API routes.

• Step 3: Implement models and route behavior

Hash configured bootstrap tokens with SHA-256 at provisioning and compare bearer tokens with hmac.compare_digest. Resolve tenant_id and agent_id from the token; reject a payload whose supplied IDs differ. Build cache and request fingerprints from resolved tenant, normalized product/context, ENGINE_VERSION, DECISION_VERSION, and active dataset versions. Persist one request/result audit pair per new fingerprint and return cache metadata for repeats. Batch must preserve input order and return one decision per product. Health returns 503 if the database probe fails.

• Step 4: Run API tests and smoke test

Run: python -m pytest central/tests/test_api.py -q

Expected: PASS.

Run: curl http://localhost:8000/health

Expected: {"status":"ok"} after Compose is up.

• Step 5: Commit if allowed

Run the Task 1 identity check. If configured: git add central && git commit -m "feat: expose authenticated fiscal API".

Task 6: Build Agent State, Outbox, and HTTP Client

Files:

• Create: agent/pyproject.toml
• Create: agent/src/gti_fiscal_agent/__init__.py
• Create: agent/src/gti_fiscal_agent/state.py
• Create: agent/src/gti_fiscal_agent/client.py
• Create: agent/src/gti_fiscal_agent/fingerprints.py
• Test: agent/tests/test_state.py
• Test: agent/tests/test_client.py

Interfaces:

• Produces AgentState.enqueue(product), pending_batch(), mark_sent(), mark_result_received(), and product_fingerprint(product) -> str.

• Produces HubClient.classify_batch(products) -> list[dict].

• Consumed by the Firebird runner in Task 7.

• Step 1: Write outbox and offline-client tests

def test_unchanged_product_with_current_versions_is_not_enqueued(state):
    state.mark_result_received("000276", source_fingerprint="a", dataset_version="v1", engine_version="0.1.0")
    assert state.enqueue("000276", "a", "v1", "0.1.0") is False

def test_offline_client_leaves_row_pending(state, offline_client):
    state.enqueue("000276", "a", "v1", "0.1.0")
    send_pending(state, offline_client)
    assert state.status("000276") == "PENDING_API"

• Step 2: Run tests to verify failure

Run: python -m pytest agent/tests/test_state.py agent/tests/test_client.py -q

Expected: import failure for AgentState.

• Step 3: Implement SQLite state machine and client

Create a SQLite product_state table keyed by product ID and an outbox table keyed by product ID plus request fingerprint. Enforce transitions DISCOVERED -> PENDING_API -> SENT -> RESULT_RECEIVED; define but do not use APPLIED. Persist retry count and next-attempt timestamp. Build product fingerprints from description, GTIN, NCM, CLASSIFICACAO_FISCAL, CFOP, CST, aliquota, CEST, and CSOSN. On connection error or non-success HTTP status, increment retry metadata and restore PENDING_API; never raise a Firebird write path.

• Step 4: Run focused tests

Run: python -m pytest agent/tests/test_state.py agent/tests/test_client.py -q

Expected: PASS.

• Step 5: Commit if allowed

Run the Task 1 identity check. If configured: git add agent && git commit -m "feat: add fiscal agent outbox".

Task 7: Integrate Read-Only Firebird Agent and Generate the 20-Product Proof

Files:

• Create: agent/src/gti_fiscal_agent/firebird_reader.py
• Create: agent/src/gti_fiscal_agent/runner.py
• Create: agent/scripts/run_agent.py
• Test: agent/tests/test_firebird_reader.py
• Test: agent/tests/test_runner.py
• Create: docs/MISSION_02_PRODUCT_RESULTS.json
• Create: docs/MISSION_02_REPORT.md

Interfaces:

• Produces read_products(catalog, codes) -> list[AgentProduct] and run_once() -> RunSummary.

• Consumes gti_discovery closed read-only catalog APIs, AgentState, and HubClient.

• Step 1: Write fingerprint gate and payload-mapping tests

def test_unknown_schema_does_not_read_or_send(reader, client):
    result = run_once(reader.with_fingerprint("other"), client)
    assert result.status == "SCHEMA_UNKNOWN"
    assert client.calls == []

def test_payload_uses_classificacao_fiscal_when_ncm_is_empty(reader):
    product = reader.product("000276")
    assert to_hub_product(product).current_ncm == "08043000"

• Step 2: Run tests to verify failure

Run: python -m pytest agent/tests/test_firebird_reader.py agent/tests/test_runner.py -q

Expected: import failure for run_once.

• Step 3: Implement SELECT-only reader and runner

Reuse connect_authorized_read_only, catalog data methods, and existing fingerprint functions; do not create a raw SQL escape hatch. Verify the computed schema fingerprint equals the reference constant before selecting the 20 codes from SAMPLE_PRODUCTS_V1.json. Select only mapped fiscal scalar fields, never IMG_PRODUTO or FOTO. Map current_ncm from NCM when nonempty, otherwise CLASSIFICACAO_FISCAL. Send batch requests with the configured tenant, agent, and company context. Save exactly 20 comparison entries containing current fields, returned classification, proposed fields, provenance, reason codes, and divergences. Mark results received only after a valid response.

• Step 4: Run focused tests and the real end-to-end CLI

Run: python -m pytest agent/tests/test_firebird_reader.py agent/tests/test_runner.py -q

Expected: PASS.

Run: python agent/scripts/run_agent.py --sample docs/SAMPLE_PRODUCTS_V1.json --output docs/MISSION_02_PRODUCT_RESULTS.json

Expected: 20 results, zero Firebird writes, and a persisted SQLite outbox state.

• Step 5: Commit if allowed

Run the Task 1 identity check. If configured: git add agent docs/MISSION_02_PRODUCT_RESULTS.json && git commit -m "feat: process products through fiscal hub".

Task 8: Document Dataset and Provider Investigation Results

Files:

• Create: docs/FISCAL_DATASETS.md
• Modify: docs/MISSION_02_REPORT.md
• Test: central/tests/test_dataset_status_documentation.py

Interfaces:

• Produces a factual provider status registry consumed only by the Mission 02 report.

• Step 1: Write a report consistency test

def test_report_contains_required_cost_and_safety_counters():
    report = Path("docs/MISSION_02_REPORT.md").read_text(encoding="utf-8")
    assert "PAID_API_CALLS=0" in report
    assert "PAID_API_COST=R$0.00" in report
    assert "APPLICATION_DDL_DML_WRITES=0" in report

• Step 2: Run test to verify failure

Run: python -m pytest central/tests/test_dataset_status_documentation.py -q

Expected: FAIL because the Mission 02 report does not exist.

• Step 3: Document exact source and integrations without invented claims

Record NCM as IMPLEMENTED with actual imported source URL, download timestamp, SHA-256, update label, act, and row count. Research primary/free sources for CEST, CFOP, CST/CSOSN, cClassTrib, IBS/CBS, and the official RTC calculator; assign only IMPLEMENTED, PARTIAL, DISCOVERED, or NOT_AVAILABLE supported by evidence. Include RTC installation/API/license/hosting/input/output evidence, or record the concrete absence/blocker and NOT_CONFIGURED adapter. Report the real 20-product metrics, timing, cache hits, and every required safety and cost counter.

• Step 4: Run documentation test

Run: python -m pytest central/tests/test_dataset_status_documentation.py -q

Expected: PASS.

• Step 5: Commit if allowed

Run the Task 1 identity check. If configured: git add docs central/tests/test_dataset_status_documentation.py && git commit -m "docs: report fiscal hub POC coverage".

Task 9: Verify Complete POC and Operational Resilience

Files:

• Modify: docs/MISSION_02_REPORT.md
• Test: central/tests/test_health.py
• Test: central/tests/test_api.py
• Test: agent/tests/test_runner.py

Interfaces:

• Consumes the complete Hub, importer, agent, report, and Docker Compose configuration.

• Step 1: Add failure-mode regression tests

def test_health_is_503_when_postgres_is_down():
    client = TestClient(create_app(database_probe=lambda: False))
    assert client.get("/health").status_code == 503

def test_second_agent_run_skips_unchanged_products_and_counts_cache(agent):
    first = agent.run_once()
    second = agent.run_once()
    assert first.total == 20
    assert second.api_requests == 0

• Step 2: Run these tests to verify their expected initial behavior

Run: python -m pytest central/tests/test_health.py agent/tests/test_runner.py -q

Expected: PASS after Tasks 1-7; otherwise fix the implementation before continuing.

• Step 3: Execute Compose and live proof

Run: docker compose up -d --build

Run: docker compose ps

Expected: postgres and fiscal-hub report healthy.

Run: docker compose exec fiscal-hub python -m central.scripts.import_ncm

Run: python agent/scripts/run_agent.py --sample docs/SAMPLE_PRODUCTS_V1.json --output docs/MISSION_02_PRODUCT_RESULTS.json

Run the same agent command again and record API call count, cache behavior, total timing, per-product timing, and cache-hit timing in the report.

• Step 4: Run full verification

Run: python -m pytest tests central/tests agent/tests -q

Expected: all tests pass.

Run: python -m json.tool docs/MISSION_02_PRODUCT_RESULTS.json > $null

Expected: valid JSON with 20 entries.

Run: git diff --check

Expected: no whitespace errors; line-ending warnings may be reported without diff failures.

• Step 5: Record final state without forcing Git identity

Run: git status --short

Run: git log --oneline -10

Add branch, status, commit state if any, health evidence, test output, performance, and real gaps to docs/MISSION_02_REPORT.md. If git config user.name is unavailable, report GIT_IDENTITY_NOT_CONFIGURED and do not configure it.