test: make consolidated workspace self-contained

git-subtree-dir: fiscal
git-subtree-mainline: d95165fc8f
git-subtree-split: 102ba84500

.gitignore

@@ -1,54 +1,26 @@
-*.FDB
-*.FDB-journal
-*.FDB.lck
+# Root-level, conservative .gitignore for the monorepo.
+# Component-specific .gitignore files (gerenciador/.gitignore, fiscal/.gitignore)
+# are preserved as-is and apply within their own subtrees.
+
+# Secrets / environment
 .env
-__pycache__/
-.pytest_cache/
-.superpowers/
-agent/*.sqlite
-agent/logs/
-agent/*.log
-
-dist/
-build/
-
-# --- release closure: secrets, runtime state, local configs, temporary artifacts ---
 .env.*
 !.env.example
-*.env
-!*.env.example
-*.secret
-secrets/
-credentials*
-*.pem
-*.key
-*.pfx
-*.log
-logs/
-*.sqlite
-*.sqlite3
-*.db
-*.dump
-*.zip
-Banco/
-bancos_teste/
-.playwright-mcp/
-.superpowers/
-deploy/hub/backups/
-deploy/hub/.env
-docs/runs/multi_db/_*
-scripts/_*
-.worktrees/
-*.tmp
-*.bak
-*.gz
-*.pkl
-.pytmp/
-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/
+
+# Python
+__pycache__/
+*.pyc
+.venv/
+venv/
+
+# Node
+node_modules/
+
+# OS
+.DS_Store
+Thumbs.db
+
+# Editors
+.vscode/
+.idea/
+*.swp

README.md

@@ -0,0 +1,37 @@
+# GTI Gerenciador + GTI Fiscal
+
+Monorepo corporativo consolidando dois serviços da GTI Sistemas, cada um
+mantendo seu histórico git original:
+
+- `gerenciador/` — o serviço de gestão de clientes e SSO ("GTI Gerenciador"
+  em produção; nome técnico do repositório de origem: `painel-clientes-web`).
+  FastAPI + Jinja2 + MySQL, autoridade de identidade para o login único.
+- `fiscal/` — o Hub (FastAPI + PostgreSQL) e o Agent (leitor Firebird) do
+  sistema fiscal ("GTI Fiscal").
+
+## Os dois continuam sendo serviços independentes em runtime
+
+A consolidação feita aqui é de **código e versionamento**, não de
+infraestrutura. Em produção, `gerenciador/` e `fiscal/` continuam rodando
+como processos separados, com bancos de dados separados (MySQL vs
+PostgreSQL) e deploys independentes.
+
+A integração entre os dois é feita por **HTTPS e configuração** (variáveis
+de ambiente), nunca por caminho de arquivo ou processo compartilhado:
+
+- SSO: o Gerenciador é a autoridade de identidade e emite uma asserção
+  assinada (Ed25519/EdDSA) que o Fiscal valida com sua chave pública.
+- Sincronização de clientes: o Fiscal consome um endpoint interno read-only
+  do Gerenciador (`GET /internal/fiscal/clients`), autenticado por um bearer
+  token dedicado.
+- Smoke de deploy: o deploy do Hub do Fiscal inclui uma checagem read-only
+  contra o Gerenciador em produção.
+
+## Históricos preservados
+
+Os commits de cada componente foram trazidos com autoria e datas originais
+preservadas (sem `--squash`). O histórico do `gerenciador/` foi extraído do
+repositório de origem (`BLOQUEIO SGV`) via `git subtree split` apontando
+só para a pasta `painel-clientes-web/` — outras iniciativas que viviam no
+mesmo repositório de origem (não relacionadas ao Gerenciador) ficaram de
+fora.

docs/LOCAL_WRITE_ALLOWLIST_V1.json

@@ -1,37 +0,0 @@
-{
- "artifact": "LOCAL_WRITE_ALLOWLIST_V1",
- "authorized_by": "operator, GTI FISCAL LOCAL COMPLETION V1 mission message",
- "scope": "disposable local COPIES only; exact canonical paths; never by directory or extension; expires with the mission (no production authorization)",
- "databases": [
-  {
-   "name": "CONSTRUCAO",
-   "path": "D:\\PROJETOS\\GTISISTEMAS\\Projeto\\Banco\\CONSTRUCAO.FDB",
-   "source_artifact": "docs/MULTI_DATABASE_SCHEMA_AUDIT_V1.json",
-   "size_bytes": 88674304
-  },
-  {
-   "name": "MMMODAS",
-   "path": "D:\\PROJETOS\\GTISISTEMAS\\Projeto\\Banco\\MMMODAS.FDB",
-   "source_artifact": "docs/MULTI_DATABASE_SCHEMA_AUDIT_V1.json",
-   "size_bytes": 248487936
-  },
-  {
-   "name": "RINCAO",
-   "path": "D:\\PROJETOS\\GTISISTEMAS\\Projeto\\Banco\\RINCAO.FDB",
-   "source_artifact": "docs/MULTI_DATABASE_SCHEMA_AUDIT_V1.json",
-   "size_bytes": 876544000
-  },
-  {
-   "name": "UNIAO",
-   "path": "D:\\PROJETOS\\GTISISTEMAS\\Projeto\\Banco\\UNIAO.FDB",
-   "source_artifact": "docs/MULTI_DATABASE_SCHEMA_AUDIT_V1.json",
-   "size_bytes": 484052992
-  },
-  {
-   "name": "TESTESOFICIAIS",
-   "path": "D:\\PROJETOS\\GTISISTEMAS\\Projeto\\TESTESOFICIAIS.FDB",
-   "source_artifact": "CLAUDE.md",
-   "size_bytes": 40337408
-  }
- ]
-}

docs/PROJECT_STATE.md

@@ -1,78 +0,0 @@
-# Project State — GTI Fiscal (2026-09-28, fim da Missão 7)
-
-Leitura obrigatória antes de qualquer nova missão fiscal (CLAUDE.md aponta para cá). Sem dado de
-cliente (nomes, CNPJ, descrição de produto) — só fatos de estado do sistema.
-
-> [!IMPORTANT]
-> **SEPARAÇÃO RIGOROSA TESTE vs PRODUÇÃO (`DADO_DE_TESTE=true`)**:
-> Todas as bases Firebird analisadas até hoje (Agent 2 e os 5 bancos de laboratório CONSTRUCAO, MMMODAS, RINCAO, UNIAO, TESTESOFICIAIS) são **dados de teste e homologação**. Nenhum cliente real passou pelo motor ainda. Qualquer métrica ou constatação levantada sobre essas bases representa validação técnica de mecanismo de software, **não achado ou diagnóstico contábil sobre nenhum cliente**.
-
-## Versões
-
-- **Hub (produção)**: 1.3.15 (preparado para deploy), deploy anterior 1.3.14 confirmado (`DEPLOY_OK`, health 200).
-- **Agent (última release Windows PUBLICADA)**: 1.3.8 — nenhum installer novo foi publicado nas
-  Missões 2–7 (o Agent permaneceu 100% inalterado).
-- **Agent 2** (installation `bb50884a-5e78-4f26-8cac-804932dae8de`): **1.3.8, informado pelo
-  operador** — ambiente de teste/homologação.
-- Hub e Agent são versionados **independentemente** — ver `docs/VERSIONING.md`.
-
-## Migration head
-
-- **main (master)**: `0034_reference_data_registry` (33 migrações). Head única linear.
-- **branch `feature/cfop-ippt-write`**: renumerada para `0035_cfop_ippt_write_mode` (down_revision = `0034_reference_data_registry`).
-- **Teste de Merge Real (Missão 7 / A.1)**:
-  - Testado via `git merge --no-commit --no-ff feature/cfop-ippt-write` e abortado limpo.
-  - Head do Alembic única: `0035_cfop_ippt_write_mode (head)` (zero split estrutural no banco).
-  - Conflitos de arquivos git mapeados: `agent/tests/test_cfop_ippt_auto_apply.py` e `release/RELEASE_MANIFEST.json`.
-
-## Reference Data Registry e Datasets P2–P4 (Missão 7)
-
-- **Inventário Oficial dos 9 Datasets Existentes** (`central/app/datasets/inventory.py`):
-  1. NCM (Siscomex)
-  2. TIPI_IPI (Receita Federal)
-  3. CEST (CONFAZ)
-  4. BA_ST_MVA (SEFAZ-BA)
-  5. NFE_MOC_TABLES (Receita Federal / ENCAT)
-  6. CFOP (Portal Nacional NF-e)
-  7. FCP (Portal Nacional NF-e)
-  8. ANP (Portal Nacional NF-e)
-  9. NCM_UTRIB (Portal Nacional NF-e)
-  *(Esclarecimento A.3: `SPED_PIS_COFINS` e `REFORMA_RTC` não integravam os 9 datasets pré-existentes; pertencem ao pipeline de novos datasets mapeados na Descoberta).*
-- **Métricas dos Datasets P2, P3 e P4**:
-  - **CFOP**: 619 códigos (IT 2023.002 v2.10, SHA-256 `c95c...ff6f`), status `DRAFT`.
-  - **NFE_DOMAINS**: 130 códigos (MOC 7.0 Anexo I, SHA-256 `5eb4...55d`), status `DRAFT`.
-  - **TIPI_IPI**: 11.103 NCMs/EX (Decreto nº 11.158/2022 até ADE 001-2026, SHA-256 `d155...2e9a`), status `DRAFT`.
-  - **CEST**: 1.223 CESTs (Convênio ICMS 142/18, SHA-256 `d7fb...54d3`), status `DRAFT`.
-- **Validação de CBS/IBS e GTIN (A.4 - 2026-09-28)**:
-  - CBS 2026: 0,9% (ADCT art. 125, II / LC Reforma Tributária).
-  - IBS 2026: 0,1% (ADCT art. 125, I / LC Reforma Tributária).
-  - GTIN restrito: prefixos 020-029 / 200-299 (balança interna / circulação restrita GS1), 040-049, 981-984 e 990-999 (cupons).
-
-## Separação Teste vs Produção e Artefatos (Missão 7 / Bloco B)
-
-- **Rótulo `DADO_DE_TESTE=true`**: Aplicado no card "Contexto Fiscal" da UI (`agent_detail.html`), no módulo de decisões (`fiscal_context_ui.py`), no dossiê contábil (`dossie_contador.md`) e em todos os relatórios.
-- **Reclassificação dos 8 Produtos Bloqueados**: Definidos formalmente como **artefatos de teste** do banco de homologação do Agent 2 (`docs/NCM_27_BLOCKED_PRODUCTS_V1.md`), nunca achados fiscais sobre clientes reais.
-- **Catálogo de Fontes de Evidência**: Formalizado em `docs/EVIDENCE_SOURCES.md`, separando as bases de teste das premissas pendentes de validação com o primeiro cliente real.
-
-## Prontidão para o Primeiro Cliente Real (Missão 7 / Bloco C)
-
-- **Checklist de Onboarding Seguro**: Detalhado em `docs/ONBOARDING_PRIMEIRO_CLIENTE_REAL.md` (requisitos de rede HTTPS outbound exclusiva, leitura em lote de 50-100 itens com `WAIT 5` e throttling de 50-100ms, backoff em busy, rollback instantâneo via parada de serviço/desinstalação).
-- **Segurança Agent-Hub**: Tokens de instalação HMAC com expiração/revogação instantânea, rate limit por instalação, anonimização estrita em logs.
-- **Modo Recomendado**: Somente leitura e DRY-RUN (sem `AUTO_FILL` ativo) pelas primeiras $N = 3$ execuções ou 7 dias de monitoramento contínuo.
-- **Critérios Objetivos de Seleção**: Roteiro de 4 perguntas para decisão humana do operador (pequeno porte < 2.000 itens, Simples Nacional CRT 1 sem IPI, varejo simples com GTIN comercial, concordância do cliente).
-
-## Endurecimento do Hub e Disaster Recovery (Missão 7 / Bloco D)
-
-- **Segurança HTTP**: Middleware implementado em `central/app/main.py` injetando `X-Content-Type-Options: nosniff`, `X-Frame-Options: SAMEORIGIN`, `X-XSS-Protection` e `Referrer-Policy`.
-- **Cobertura de Testes**: Suíte `central/tests/test_mission_07_hardening.py` adicionada cobrindo ciclo de vida do Registry (aprovação, rejeição, checklist), classificação de taxas TIPI e headers de segurança.
-- **Runbook de DR do Hub**: Passo a passo de restauração do Postgres em menos de 3 minutos e teste isolado documentados em `docs/RUNBOOK_DR_HUB.md`.
-
-## Índice de Documentos-Chave (docs/)
-
-- `EVIDENCE_SOURCES.md` — matriz de fontes de teste vs o que falta validar com cliente real.
-- `ONBOARDING_PRIMEIRO_CLIENTE_REAL.md` — checklist de onboarding, segurança e critério dry-run.
-- `RUNBOOK_DR_HUB.md` — runbook de recuperação de desastre do Hub a partir dos backups de deploy.
-- `REFERENCE_DATASET_MANIFEST_V1.md` — manifesto de datasets oficiais baixados (P1 a P4 com hashes e URLs).
-- `DESCOBERTA_P5_P6_MISSAO_6.md` — levantamento completo de PIS/COFINS SPED e ICMS-ST para o Hub.
-- `dossie_contador.md` — dossiê fiscal com aviso DADO_DE_TESTE=true.
-- `NCM_27_BLOCKED_PRODUCTS_V1.md` — artefatos de teste com bloqueio de NCM.

fiscal/.gitignore

@@ -0,0 +1,59 @@
+*.FDB
+*.FDB-journal
+*.FDB.lck
+.env
+__pycache__/
+.pytest_cache/
+.superpowers/
+agent/*.sqlite
+agent/logs/
+agent/*.log
+
+dist/
+build/
+
+# --- release closure: secrets, runtime state, local configs, temporary artifacts ---
+.env.*
+!.env.example
+*.env
+!*.env.example
+*.secret
+secrets/
+credentials*
+*.pem
+*.key
+*.pfx
+*.log
+logs/
+*.sqlite
+*.sqlite3
+*.db
+*.dump
+*.zip
+Banco/
+bancos_teste/
+.playwright-mcp/
+.superpowers/
+deploy/hub/backups/
+deploy/hub/.env
+docs/runs/multi_db/_*
+scripts/_*
+.worktrees/
+*.tmp
+*.bak
+*.gz
+*.pkl
+.pytmp/
+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/
+# Missão 9: per-installation Agent env files hold real AGENT_TOKEN secrets, plus local SQLite
+# state and logs - never commit any of it. Only the *_original.sha256.txt manifests under
+# work_copies/ (proof the real-client originals were never touched) are tracked.
+work_copies/agent_envs/
+work_copies/*.log
+scratch_enrich/

fiscal/CLAUDE.md

@@ -13,9 +13,14 @@ modify database objects or permissions, or store credentials and other secrets.
 Run the offline tests with:
 
 ```text
+# Suíte completa canônica (1160+ testes: raiz + Hub + Agent):
+python -m pytest tests central/tests agent/tests -q
+
+# Apenas testes unitários offline da raiz:
 python -m pytest tests -q
 ```
 
+
 ## Firebird write policy
 
 Default policy:

fiscal/agent/src/gti_fiscal_agent/cfop_ippt_auto_apply.py

@@ -0,0 +1,185 @@
+"""CFOP/IPPT auto-apply: fills or corrects ONLY C000025.CFOP and C000025.IPPT, and nothing else,
+when central.app.field_contract.resolve_product already classified the cell AUTO_FILL_SAFE
+(state=CHANGE, write_eligible=True) for this product.
+
+Safety model (defense in depth, default OFF), mirrors gti_fiscal_agent.rtc_auto_apply: the Hub sets
+a per-installation, per-capability mode DISABLED / DRY_RUN / APPROVAL_REQUIRED / ENABLED and delivers
+it inside an authenticated job (central.app.cfop_ippt_audit.job_authorizations). The Agent only
+exposes the restricted capability CFOP_IPPT_WRITE_V1; there is no local .env switch. Write allowlist:
+ALLOWED_FIELDS = ("CFOP", "IPPT") on C000025 - no free SQL, one short transaction per cell.
+
+Compare-before-write ("human override sempre vence"): every candidate carries the exact value the
+field_contract resolver saw when it was selected (`expected_before`). Immediately before the UPDATE
+the cell is re-read; if it no longer equals `expected_before` - a human (or any other process) changed
+it since - the write is ABORTED with STALE_VALUE_CHANGED_SINCE_DECISION and nothing is written; if it
+already equals the proposed value the write is skipped as NO_WRITE_REQUIRED (idempotent - a second run
+over the same candidates always writes 0). Every applied change is read back and verified; a mismatch
+never returns APPLIED. `rollback_one` restores the previous value only if the current value still
+equals what was written (a later human edit blocks the rollback too).
+"""
+from __future__ import annotations
+
+CAPABILITY = "CFOP_IPPT_WRITE_V1"
+ALLOWED_FIELDS = ("CFOP", "IPPT")
+MODES = ("DISABLED", "DRY_RUN", "APPROVAL_REQUIRED", "ENABLED")
+DEFAULT_MAX_PER_JOB = 500
+
+
+class CfopIpptApplyAborted(Exception):
+    """The cell is left untouched; `reason` is reported in the audit."""
+
+    def __init__(self, reason: str):
+        super().__init__(reason)
+        self.reason = reason
+
+
+def _text(value) -> str:
+    return "" if value is None else str(value).strip()
+
+
+def write_gate(*, mode: str, capability: bool = True, approved: bool = True) -> dict:
+    """Same shape/semantics as rtc_auto_apply.write_gate: a real write needs the capability AND an
+    authenticated Hub mode that permits writing (ENABLED, or APPROVAL_REQUIRED for approved cells)."""
+    mode = mode if mode in MODES else "DISABLED"
+    missing = []
+    if not capability:
+        missing.append("CAPABILITY")
+    if mode not in ("ENABLED", "APPROVAL_REQUIRED"):
+        missing.append("HUB_MODE_" + mode)
+    elif mode == "APPROVAL_REQUIRED" and not approved:
+        missing.append("HUB_APPROVAL")
+    return {"allowed": not missing, "missing": missing, "mode": mode, "capability": bool(capability)}
+
+
+def candidate_fingerprint(code: str, field: str, expected_before, proposed) -> str:
+    import hashlib
+    return hashlib.sha256("|".join((_text(code), field, _text(expected_before), _text(proposed))).encode("utf-8")).hexdigest()[:24]
+
+
+def select_candidates(results: dict, *, limit: int = DEFAULT_MAX_PER_JOB) -> tuple[list[dict], int]:
+    """`results` is {product_code: {field: {"state": ..., "current": ..., "proposed": ..., "write_eligible": ..., "reason": ...}}} -
+    the SAME per-field dict central.app.field_contract.resolve_product already returns for every product in the job's dry-run pass.
+    Only CFOP/IPPT cells with state=CHANGE and write_eligible=True are candidates - never a second, independent rule."""
+    from central.app.field_contract import CHANGE
+    ready = []
+    for code, fields in results.items():
+        for field in ALLOWED_FIELDS:
+            r = (fields or {}).get(field) or {}
+            if r.get("state") != CHANGE or not r.get("write_eligible"):
+                continue
+            expected_before, proposed = r.get("current"), r.get("proposed")
+            ready.append({"product_code": code, "field": field, "expected_before": expected_before, "proposed": proposed, "reason": r.get("reason"),
+                          "fingerprint": candidate_fingerprint(code, field, expected_before, proposed)})
+    return ready[:limit], len(ready)
+
+
+def apply_candidate(conn, candidate: dict, *, now: str, job_id=None) -> dict:
+    """One cell, one short transaction. Raises CfopIpptApplyAborted (nothing written)."""
+    field = candidate["field"]
+    assert field in ALLOWED_FIELDS
+    code, expected_before, proposed = candidate["product_code"], candidate["expected_before"], candidate["proposed"]
+    base = {"product_code": code, "field": field, "job_id": job_id, "timestamp": now, "action": "APPLY", "reason_code": candidate.get("reason")}
+    cur = conn.cursor()
+    try:
+        cur.execute(f"SELECT {field} FROM C000025 WHERE CODIGO = ?", (code,))
+        rows = cur.fetchall()
+        if len(rows) != 1:
+            raise CfopIpptApplyAborted("PRODUCT_ROW_NOT_UNIQUE" if rows else "PRODUCT_NOT_FOUND")
+        current = _text(rows[0][0]) or None
+        norm_expected = _text(expected_before) or None
+        if current == (_text(proposed) or None):
+            raise CfopIpptApplyAborted("NO_WRITE_REQUIRED")
+        if current != norm_expected:
+            raise CfopIpptApplyAborted("STALE_VALUE_CHANGED_SINCE_DECISION")
+        cur.execute(f"UPDATE C000025 SET {field} = ? WHERE CODIGO = ? AND (({field} IS NULL) OR (TRIM({field}) = ?))", (proposed, code, norm_expected or ""))
+        if cur.rowcount != 1:
+            conn.rollback()
+            raise CfopIpptApplyAborted("UPDATE_ROWCOUNT_NOT_ONE")
+        conn.commit()
+        cur = conn.cursor()
+        cur.execute(f"SELECT {field} FROM C000025 WHERE CODIGO = ?", (code,))
+        after = _text(cur.fetchone()[0]) or None
+        if after != (_text(proposed) or None):
+            raise RuntimeError(f"post-write verification failed for {code}.{field}: {after!r} != {proposed!r}")
+        return {**base, "outcome": "APPLIED", "old_value": current, "new_value": proposed, "rowcount": 1, "verified_after": True}
+    except CfopIpptApplyAborted:
+        try:
+            conn.rollback()
+        except Exception:  # noqa: BLE001
+            pass
+        raise
+    except Exception:
+        try:
+            conn.rollback()
+        except Exception:  # noqa: BLE001
+            pass
+        raise
+
+
+def apply_candidates(conn, candidates: list[dict], *, now: str, job_id=None, authorize=None) -> dict:
+    """Applies each candidate independently; a failure or an abort never stops the rest and never
+    leaves a partial change. `authorize(product_code, field)` (optional) re-asks the Hub, right before
+    each UPDATE, whether the central authorization still permits it."""
+    applied, aborted, reasons = [], [], {}
+    for candidate in candidates:
+        code, field = candidate["product_code"], candidate["field"]
+        if authorize is not None:
+            denied = authorize(code, field)
+            if denied:
+                aborted.append({"outcome": "ABORTED", "product_code": code, "field": field, "reason": denied, "job_id": job_id, "timestamp": now, "action": "APPLY"})
+                reasons[denied] = reasons.get(denied, 0) + 1
+                continue
+        try:
+            applied.append(apply_candidate(conn, candidate, now=now, job_id=job_id))
+        except CfopIpptApplyAborted as stop:
+            aborted.append({"outcome": "ABORTED", "product_code": code, "field": field, "reason": stop.reason, "job_id": job_id, "timestamp": now, "action": "APPLY"})
+            reasons[stop.reason] = reasons.get(stop.reason, 0) + 1
+        except Exception:  # noqa: BLE001 - a technical failure of ONE cell must not abort the batch
+            aborted.append({"outcome": "FAILED", "product_code": code, "field": field, "reason": "TECHNICAL_FAILURE", "job_id": job_id, "timestamp": now})
+            reasons["TECHNICAL_FAILURE"] = reasons.get("TECHNICAL_FAILURE", 0) + 1
+    return {"applied": applied, "aborted": aborted, "aborted_reasons": reasons, "write_attempts": len(candidates), "write_successes": len(applied)}
+
+
+def rollback_one(conn, item: dict, *, now: str, job_id=None) -> dict:
+    """Restores C000025.<field> to `old_value` for ONE previously applied audit record, only if the
+    current value still equals what was written (`new_value`) - a later human edit blocks the rollback."""
+    field = item.get("field")
+    if field not in ALLOWED_FIELDS:
+        return {"outcome": "ABORTED", "reason": "UNKNOWN_FIELD", "field": field, "job_id": job_id, "timestamp": now}
+    code, written, restore_to = _text(item.get("product_code")), _text(item.get("new_value")) or None, item.get("old_value")
+    base = {"product_code": code, "field": field, "job_id": job_id, "timestamp": now, "action": "ROLLBACK", "rollback_of": item.get("audit_id"), "old_value": written, "new_value": restore_to}
+    if not code:
+        return {**base, "outcome": "ABORTED", "reason": "ROLLBACK_REQUEST_INVALID"}
+    cur = conn.cursor()
+    try:
+        cur.execute(f"SELECT {field} FROM C000025 WHERE CODIGO = ?", (code,))
+        rows = cur.fetchall()
+        if len(rows) != 1:
+            conn.rollback()
+            return {**base, "outcome": "ABORTED", "reason": "PRODUCT_ROW_NOT_UNIQUE" if rows else "PRODUCT_NOT_FOUND"}
+        current = _text(rows[0][0]) or None
+        if current != written:
+            conn.rollback()
+            return {**base, "outcome": "ABORTED", "reason": "ROLLBACK_CONFLICT_VALUE_CHANGED"}
+        cur.execute(f"UPDATE C000025 SET {field} = ? WHERE CODIGO = ? AND (({field} IS NULL) OR (TRIM({field}) = ?))", (restore_to, code, written or ""))
+        if cur.rowcount != 1:
+            conn.rollback()
+            return {**base, "outcome": "ABORTED", "reason": "UPDATE_ROWCOUNT_NOT_ONE"}
+        conn.commit()
+        cur = conn.cursor()
+        cur.execute(f"SELECT {field} FROM C000025 WHERE CODIGO = ?", (code,))
+        after = _text(cur.fetchone()[0]) or None
+        if after != (_text(restore_to) or None):
+            raise RuntimeError("post-rollback verification failed")
+        return {**base, "outcome": "ROLLED_BACK", "rowcount": 1, "verified_after": True}
+    except Exception as error:  # noqa: BLE001
+        try:
+            conn.rollback()
+        except Exception:  # noqa: BLE001
+            pass
+        return {**base, "outcome": "FAILED", "reason": type(error).__name__}
+
+
+def rollback_records(conn, items: list[dict], *, now: str, job_id=None) -> dict:
+    results = [rollback_one(conn, item, now=now, job_id=job_id) for item in items]
+    return {"rollbacks": results, "rolled_back": sum(1 for r in results if r["outcome"] == "ROLLED_BACK")}