diff --git a/agent/pyproject.toml b/agent/pyproject.toml
index 49f5aa9..ebbde13 100644
--- a/agent/pyproject.toml
+++ b/agent/pyproject.toml
@@ -1,6 +1,6 @@
[project]
name = "gti-fiscal-agent"
-version = "1.1.2"
+version = "1.1.3"
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 83988f4..5ab8d29 100644
--- a/agent/scripts/run_agent_service.py
+++ b/agent/scripts/run_agent_service.py
@@ -31,10 +31,12 @@ sys.path.insert(0, str(PROJECT_ROOT)) # ships central/app engine modules read-o
from gti_fiscal_agent.client import HubClient # noqa: E402
from gti_fiscal_agent.firebird_reader import check_connection, read_products # noqa: E402
from gti_fiscal_agent.pipeline import run_pipeline # noqa: E402
-from gti_fiscal_agent.runner import REFERENCE_SCHEMA_V1, to_hub_product # noqa: E402
+from gti_fiscal_agent.runner import REFERENCE_SCHEMA_V1, product_fingerprint, to_hub_product # noqa: E402
+from central.app.operational_state import classify_product_state, summarize_operational_states # noqa: E402
from gti_fiscal_agent.state import AgentState # noqa: E402
_MAX_BACKOFF_SECONDS = 60
+SYNC_JOB_KINDS = ("INITIAL_SYNC", "SYNC_INCREMENTAL", "REPROCESS_ALL") # read-only ERP jobs that refresh coverage
logger = logging.getLogger("gti_fiscal_agent")
@@ -87,6 +89,7 @@ class Config:
self.agent_token = self._resolve_agent_token(env)
self.company_id = env.get("COMPANY_ID", "")
self.company_uf = env.get("COMPANY_UF", "")
+ _resolve_firebird_credentials(env)
self.database_path = env["FIREBIRD_DATABASE_PATH"]
# FISCAL_MODE gates writes technically: DRY_RUN can never write, whatever FIREBIRD_WRITES says. Unset = legacy
# behaviour (FIREBIRD_WRITES decides). `firebird_writes` below is the EFFECTIVE value used by every write path.
@@ -94,10 +97,12 @@ class Config:
self.fiscal_mode = (env.get("FISCAL_MODE") or ("AUTO_APPLY" if requested else "DRY_RUN")).strip().upper()
self.requested_writes = requested
self.firebird_writes = requested and self.fiscal_mode == "AUTO_APPLY"
- jobs_enabled = (env.get("AGENT_JOBS_ENABLED", "1") or "1").strip().upper()
- self.agent_jobs_enabled = jobs_enabled not in ("0", "FALSE", "NO")
+ # Job execution is controlled CENTRALLY (Hub heartbeat: processing_enabled) — there is no
+ # local toggle, so no installation ever needs its agent.env edited to start/stop work.
+ # (A legacy AGENT_JOBS_ENABLED line in an old agent.env is deliberately ignored.)
+ self.agent_jobs_enabled = True
self.poll_interval_seconds = float(env.get("POLL_INTERVAL_SECONDS", "5"))
- self.agent_version = env.get("AGENT_VERSION", "1.1.2")
+ self.agent_version = env.get("AGENT_VERSION", "1.1.3")
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
@@ -125,6 +130,23 @@ class Config:
return ""
+def _resolve_firebird_credentials(env):
+ """FB_USER/FB_PASSWORD come from the DPAPI-protected firebird.bin written by Setup
+ (never from a hand-edited agent.env, never sent to the Hub). An explicit FB_PASSWORD in
+ the environment still wins (dev override)."""
+ if env.get("FB_PASSWORD"):
+ return
+ from gti_fiscal_agent.credential_store import read_credential
+
+ path = env.get("GTI_FIREBIRD_CREDENTIAL_FILE") or str(Path(os.environ.get("PROGRAMFILES", "C:/Program Files")) / "GTI" / "FiscalAgent" / "firebird.bin")
+ try:
+ data = json.loads(read_credential(path, env=env))
+ except Exception as exc: # noqa: BLE001 — absent/unreadable: preflight will fail closed with a clear blocker
+ logger.info("no Firebird credential file usable: %s", type(exc).__name__)
+ return
+ env["FB_USER"], env["FB_PASSWORD"] = data.get("user", ""), data.get("password", "")
+
+
class GracefulStop:
def __init__(self):
self.stop = False
@@ -183,23 +205,46 @@ def execute_job(job, config, executor, state, env):
logger.info("job %s: read %d product(s) from Firebird", job.get("job_id"), len(products))
client = HubClient(config.hub_url, config.agent_token)
- decisions = client.classify_batch([to_hub_product(p) for p in products])
+ sync_kind = kind in SYNC_JOB_KINDS
+ cache = sync_kind and hasattr(state, "remember_decision")
+ # SYNC_INCREMENTAL = delta: only products whose source fingerprint changed (or are new) are
+ # re-classified by the Hub; INITIAL_SYNC / REPROCESS_ALL / REPROCESS_PRODUCTS process everything they read.
+ if kind == "SYNC_INCREMENTAL" and cache:
+ to_process = [p for p in products if state.cached_fingerprint(p["CODIGO"]) != product_fingerprint(p)]
+ else:
+ to_process = products
+ decisions = client.classify_batch([to_hub_product(p) for p in to_process])
write_connection = None
if config.firebird_writes:
write_connection = open_write_connection(env, config.database_path)
try:
- summary = run_pipeline(products, decisions, state, config.firebird_writes, write_connection, executor, config.database_path, authorized_path=config.authorized_write_database_path)
+ summary = run_pipeline(to_process, decisions, state, config.firebird_writes, write_connection, executor, config.database_path, authorized_path=config.authorized_write_database_path)
finally:
if write_connection is not None:
write_connection.close()
- coverage = build_job_coverage(env, client, products, decisions) if kind == "REPROCESS_ALL" else None
+ operational = coverage = None
+ if sync_kind:
+ if cache:
+ for product, decision in zip(to_process, decisions):
+ state.remember_decision(product["CODIGO"], product_fingerprint(product), decision)
+ state.forget_missing([p["CODIGO"] for p in products])
+ all_decisions = [state.cached_decision(p["CODIGO"]) or {} for p in products]
+ else:
+ all_decisions = decisions
+ operational = summarize_operational_states([classify_product_state(p, d) for p, d in zip(products, all_decisions)])
+ coverage = build_job_coverage(env, client, products, all_decisions)
result_summary = {
"coverage": coverage,
+ "operational": operational,
"fiscal_mode": getattr(config, "fiscal_mode", None),
- "release_version": env.get("AGENT_VERSION") or "1.1.2",
+ "release_version": env.get("AGENT_VERSION") or "1.1.3",
+ "products_total": len(products),
+ "products_changed": len(to_process),
"products_processed": summary["processed"],
+ "blocked_products": summary["blocked_products"],
+ "resolved_products": summary["resolved_products"],
"writes_attempted": summary["writes_attempted"],
"writes_succeeded": summary["writes_succeeded"],
"writes_failed": summary["writes_failed"],
@@ -275,9 +320,10 @@ def run(config, env, *, max_iterations=None, stopper=None):
"fiscal_engine_version": env.get("FISCAL_ENGINE_VERSION", config.agent_version), "dataset_versions": {},
}
if hasattr(client, "heartbeat_metadata"):
- client.heartbeat_metadata(heartbeat_payload)
+ hb_response = client.heartbeat_metadata(heartbeat_payload)
else:
- client.heartbeat("IDLE", schema_fingerprint=config.schema_fingerprint, agent_version=config.agent_version, last_job_id=last_job_id)
+ hb_response = client.heartbeat("IDLE", schema_fingerprint=config.schema_fingerprint, agent_version=config.agent_version, last_job_id=last_job_id)
+ processing_enabled = not (isinstance(hb_response, dict) and hb_response.get("processing_enabled") is False)
if was_down:
logger.info("reconnected to Hub after %ss backoff", backoff)
was_down = False
@@ -289,7 +335,7 @@ def run(config, env, *, max_iterations=None, stopper=None):
backoff = min(backoff * 2, _MAX_BACKOFF_SECONDS)
continue
- if not config.agent_jobs_enabled:
+ if not processing_enabled: # central pause: heartbeat keeps flowing, no job is claimed or executed
time.sleep(config.poll_interval_seconds)
continue
@@ -331,7 +377,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.1.2")
+ report = run_preflight(env, agent_version=env.get("AGENT_VERSION") or "1.1.3")
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/preflight.py b/agent/src/gti_fiscal_agent/preflight.py
index a9f5284..8f8c5cc 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.1.2"
+AGENT_VERSION_DEFAULT = "1.1.3"
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/src/gti_fiscal_agent/state.py b/agent/src/gti_fiscal_agent/state.py
index 12bbc10..0f61868 100644
--- a/agent/src/gti_fiscal_agent/state.py
+++ b/agent/src/gti_fiscal_agent/state.py
@@ -1,4 +1,5 @@
"""Local SQLite outbox; it has no Firebird dependency or write path."""
+import json
import sqlite3
from pathlib import Path
import uuid
@@ -8,6 +9,9 @@ class AgentState:
self.connection = sqlite3.connect(Path(path)); self.connection.row_factory = sqlite3.Row
self.connection.executescript("CREATE TABLE IF NOT EXISTS product_state(product_id TEXT PRIMARY KEY, source_fingerprint TEXT NOT NULL, dataset_version TEXT NOT NULL, engine_version TEXT NOT NULL, status TEXT NOT NULL, last_processed_at TEXT, last_result_hash TEXT, reason_code TEXT); CREATE TABLE IF NOT EXISTS outbox(product_id TEXT PRIMARY KEY, source_fingerprint TEXT NOT NULL, status TEXT NOT NULL, retries INTEGER NOT NULL DEFAULT 0); CREATE TABLE IF NOT EXISTS ipi_write_log(product_id TEXT NOT NULL, source_fingerprint TEXT NOT NULL, decision_status TEXT, ipi_status TEXT, write_status TEXT NOT NULL, before_value TEXT, proposed_value TEXT, result TEXT, reason_code TEXT, timestamp TEXT NOT NULL);")
self.connection.execute("CREATE TABLE IF NOT EXISTS installation_identity (id INTEGER PRIMARY KEY CHECK(id=1), installation_id TEXT NOT NULL)")
+ # Last Hub decision per product + the source fingerprint it was computed for: what makes
+ # SYNC_INCREMENTAL a real delta (only new/changed products are re-classified).
+ self.connection.execute("CREATE TABLE IF NOT EXISTS decision_cache (product_id TEXT PRIMARY KEY, source_fingerprint TEXT NOT NULL, decision_json TEXT NOT NULL)")
self.connection.commit()
def stored_installation_id(self):
row = self.connection.execute("SELECT installation_id FROM installation_identity WHERE id=1").fetchone()
@@ -35,6 +39,23 @@ class AgentState:
if row is None or str(uuid.UUID(row["installation_id"])) != value:
raise ValueError("persisted installation_id conflicts with candidate")
return value
+ def cached_fingerprint(self, product_id):
+ row = self.connection.execute("SELECT source_fingerprint FROM decision_cache WHERE product_id=?", (product_id,)).fetchone()
+ return row["source_fingerprint"] if row else None
+ def cached_decision(self, product_id):
+ row = self.connection.execute("SELECT decision_json FROM decision_cache WHERE product_id=?", (product_id,)).fetchone()
+ return json.loads(row["decision_json"]) if row else None
+ def remember_decision(self, product_id, source_fingerprint, decision):
+ self.connection.execute("INSERT OR REPLACE INTO decision_cache(product_id,source_fingerprint,decision_json) VALUES(?,?,?)", (product_id, source_fingerprint, json.dumps(decision, default=str)))
+ self.connection.commit()
+ def forget_missing(self, present_product_ids):
+ """Products no longer in the ERP leave the LOCAL cache only; Hub-side history/blockers are never deleted by this."""
+ present = set(present_product_ids)
+ gone = [row["product_id"] for row in self.connection.execute("SELECT product_id FROM decision_cache") if row["product_id"] not in present]
+ for product_id in gone:
+ self.connection.execute("DELETE FROM decision_cache WHERE product_id=?", (product_id,))
+ self.connection.commit()
+ return len(gone)
def enqueue(self, product_id, source_fingerprint, dataset_version, engine_version):
row = self.connection.execute("SELECT * FROM product_state WHERE product_id=?", (product_id,)).fetchone()
if row and tuple(row[key] for key in ("source_fingerprint", "engine_version", "status")) == (source_fingerprint, engine_version, "RESULT_RECEIVED"): return False
diff --git a/agent/tests/test_run_agent_service.py b/agent/tests/test_run_agent_service.py
index 750a4ea..d012cf0 100644
--- a/agent/tests/test_run_agent_service.py
+++ b/agent/tests/test_run_agent_service.py
@@ -90,10 +90,12 @@ def test_config_defaults_from_env():
assert config.agent_jobs_enabled is True
-def test_config_can_disable_agent_job_polling():
+def test_local_agent_jobs_toggle_is_ignored_control_is_central():
+ # Zero-touch: a stale AGENT_JOBS_ENABLED=0 left in an old agent.env must not silence the Agent;
+ # only the Hub (heartbeat processing_enabled) can pause it.
config = service.Config({"FIREBIRD_DATABASE_PATH": AUTHORIZED, "AGENT_JOBS_ENABLED": "0"})
- assert config.agent_jobs_enabled is False
+ assert config.agent_jobs_enabled is True
def test_config_reads_gti_hub_url_and_writes_flag():
@@ -294,7 +296,7 @@ class _ExternalStopFlag:
self.stop = False
-def test_run_sends_heartbeat_but_never_polls_when_jobs_disabled(monkeypatch, tmp_path):
+def test_run_sends_heartbeat_but_never_polls_when_paused_centrally(monkeypatch, tmp_path):
stopper = _ExternalStopFlag()
fake_hub = _FakeHubForLoop()
installation_id = "00000000-0000-4000-8000-000000000222"
@@ -308,11 +310,11 @@ def test_run_sends_heartbeat_but_never_polls_when_jobs_disabled(monkeypatch, tmp
calls["heartbeat"] += 1
calls["heartbeat_installation_id"] = payload["installation_id"]
stopper.stop = True
- return {"agent_id": 2}
+ return {"agent_id": 2, "processing_enabled": False} # paused centrally by the Hub
def forbidden_next_job():
calls["next_job"] += 1
- raise AssertionError("jobs are disabled; next_job must not be called")
+ raise AssertionError("processing is paused centrally; next_job must not be called")
fake_hub.identity = identity
fake_hub.heartbeat_metadata = heartbeat_metadata
@@ -323,7 +325,6 @@ def test_run_sends_heartbeat_but_never_polls_when_jobs_disabled(monkeypatch, tmp
config = _config(
agent_id="2",
- agent_jobs_enabled=False,
poll_interval_seconds=0,
state_db=str(tmp_path / "agent.db"),
)
diff --git a/agent/tests/test_zero_touch_agent.py b/agent/tests/test_zero_touch_agent.py
new file mode 100644
index 0000000..64a7510
--- /dev/null
+++ b/agent/tests/test_zero_touch_agent.py
@@ -0,0 +1,127 @@
+"""Agent side of the zero-touch model: delta sync via fingerprint cache, read-only guarantee,
+blockers/operational summary in the job result, central pause, Firebird credential resolution."""
+import json
+
+import pytest
+
+from test_run_agent_service import AUTHORIZED, _FakeExecutor, _config, service
+
+
+class FakeHub:
+ def __init__(self):
+ self.classified = []
+
+ def classify_batch(self, products):
+ self.classified.append([p["product_id"] for p in products])
+ return [{"status": "EXACT_NCM" if p["current_ncm"] else "INVALID_NCM", "ipi_status": "IPI_RESOLVED", "classification": {}} for p in products]
+
+
+def product(code, ncm="03055100", name="X"):
+ return {"CODIGO": code, "PRODUTO": name, "EAN_GTIN": None, "CLASSIFICACAO_FISCAL": ncm, "NCM": None, "CFOP": "5102", "CST": None, "ALIQUOTA": None, "CEST": None, "CSOSN": "102", "IPI": None}
+
+
+@pytest.fixture
+def env_setup(monkeypatch, tmp_path):
+ hub = FakeHub()
+ monkeypatch.setattr(service, "HubClient", lambda *a, **k: hub)
+ monkeypatch.setattr(service, "build_job_coverage", lambda env, client, products, decisions: {"total_products": len(products)})
+ state = service.AgentState(tmp_path / "s.db")
+ return hub, state
+
+
+def run_job(kind, products, state, monkeypatch, config=None):
+ monkeypatch.setattr(service, "read_products", lambda env, codigos=None: products)
+ return service.execute_job({"job_id": 1, "kind": kind, "agent_id": 2, "payload": {}}, config or _config(), _FakeExecutor(), state, {})
+
+
+def test_initial_sync_processes_everything_and_reports_operational_summary(env_setup, monkeypatch):
+ hub, state = env_setup
+ products = [product("1"), product("2"), product("3", ncm="")]
+ result, error = run_job("INITIAL_SYNC", products, state, monkeypatch)
+ assert error is None
+ assert result["products_total"] == 3 and result["products_changed"] == 3 and hub.classified == [["1", "2", "3"]]
+ op = result["operational"]
+ assert op["TOTAL_PRODUCTS"] == 3 and op["COMPLETE"] == 2 and op["SAFE_BLOCKED"] == 1
+ assert op["COMPLETE"] + op["CORRECTED"] + op["SAFE_BLOCKED"] + op["FAILED_TECHNICAL"] == op["TOTAL_PRODUCTS"] # invariant closes
+ assert [b["product_code"] for b in result["blocked_products"]] == ["3"] and result["resolved_products"] == ["1", "2"]
+
+
+def test_incremental_sync_reclassifies_only_new_and_changed_products(env_setup, monkeypatch):
+ hub, state = env_setup
+ base = [product("1"), product("2"), product("3")]
+ run_job("INITIAL_SYNC", base, state, monkeypatch)
+ hub.classified.clear()
+
+ result, error = run_job("SYNC_INCREMENTAL", base, state, monkeypatch) # nothing changed
+ assert error is None and result["products_changed"] == 0 and hub.classified == [[]] or hub.classified == []
+ assert result["operational"]["TOTAL_PRODUCTS"] == 3 # coverage still complete, rebuilt from the cache
+
+ hub.classified.clear()
+ changed = [product("1"), product("2", ncm="", name="broken now"), product("3"), product("4")] # 2 changed, 4 new
+ result, _ = run_job("SYNC_INCREMENTAL", changed, state, monkeypatch)
+ assert sorted(hub.classified[0]) == ["2", "4"] and result["products_changed"] == 2
+ assert result["operational"]["TOTAL_PRODUCTS"] == 4 and result["operational"]["SAFE_BLOCKED"] == 1
+
+
+def test_removed_product_leaves_local_cache_but_is_not_reported_as_resolved(env_setup, monkeypatch):
+ hub, state = env_setup
+ run_job("INITIAL_SYNC", [product("1"), product("2")], state, monkeypatch)
+ result, _ = run_job("SYNC_INCREMENTAL", [product("1")], state, monkeypatch)
+ assert state.cached_decision("2") is None and result["operational"]["TOTAL_PRODUCTS"] == 1
+ assert result["resolved_products"] == [] # no history is auto-resolved/deleted for a product that merely disappeared
+
+
+def test_every_sync_kind_is_read_only(env_setup, monkeypatch):
+ hub, state = env_setup
+ for kind in ("INITIAL_SYNC", "SYNC_INCREMENTAL", "REPROCESS_ALL"):
+ result, error = run_job(kind, [product("1")], state, monkeypatch)
+ assert error is None and result["writes_attempted"] == 0 and result["writes_succeeded"] == 0 and result["firebird_writes_enabled"] is False
+ with pytest.raises(RuntimeError, match="writes are disabled"):
+ service.open_write_connection({}, AUTHORIZED, firebird_writes=False) # no write-capable connection can even be opened
+
+
+def test_hub_offline_during_job_fails_the_job_cleanly_no_crash(env_setup, monkeypatch):
+ hub, state = env_setup
+ def boom(products): raise ConnectionError("hub down")
+ hub.classify_batch = boom
+ result, error = run_job("SYNC_INCREMENTAL", [product("9")], state, monkeypatch)
+ assert result is None and "ConnectionError" in error # reported as a job failure, the service loop keeps running
+
+
+def test_firebird_offline_during_job_fails_the_job_cleanly(env_setup, monkeypatch):
+ hub, state = env_setup
+ def down(env, codigos=None): raise OSError("Firebird service not running")
+ monkeypatch.setattr(service, "read_products", down)
+ result, error = service.execute_job({"job_id": 1, "kind": "SYNC_INCREMENTAL", "agent_id": 2, "payload": {}}, _config(), _FakeExecutor(), state, {})
+ assert result is None and "OSError" in error
+
+
+def test_firebird_credentials_come_from_the_dpapi_file_not_agent_env(monkeypatch, tmp_path):
+ from gti_fiscal_agent import credential_store
+ path = tmp_path / "firebird.bin"
+ monkeypatch.setattr(credential_store, "read_credential", lambda p, env=None: json.dumps({"user": "SYSDBA", "password": "s3cret"}) if str(p) == str(path) else (_ for _ in ()).throw(FileNotFoundError()))
+ env = {"FIREBIRD_DATABASE_PATH": AUTHORIZED, "GTI_FIREBIRD_CREDENTIAL_FILE": str(path)}
+ service.Config(env)
+ assert env["FB_USER"] == "SYSDBA" and env["FB_PASSWORD"] == "s3cret"
+ explicit = {"FIREBIRD_DATABASE_PATH": AUTHORIZED, "FB_PASSWORD": "dev", "FB_USER": "u", "GTI_FIREBIRD_CREDENTIAL_FILE": str(path)}
+ service.Config(explicit)
+ assert explicit["FB_PASSWORD"] == "dev" # an explicit env value still wins (dev override)
+
+
+def test_central_resume_after_pause_makes_agent_poll_again(monkeypatch, tmp_path):
+ class Stopper: stop = False
+ stopper = Stopper()
+ states = iter([False, True]) # paused, then resumed
+ calls = {"next_job": 0}
+
+ class Hub:
+ def identity(self): return {"agent_id": 2, "installation_id": "00000000-0000-4000-8000-000000000222"}
+ def heartbeat_metadata(self, payload): return {"processing_enabled": next(states)}
+ def next_job(self):
+ calls["next_job"] += 1; stopper.stop = True; return None
+
+ monkeypatch.setattr(service, "HubClient", lambda *a, **k: Hub())
+ monkeypatch.setattr(service, "load_executor", lambda: _FakeExecutor())
+ monkeypatch.setattr(service.time, "sleep", lambda s: None)
+ service.run(_config(agent_id="2", poll_interval_seconds=0, state_db=str(tmp_path / "a.db")), {"AGENT_ID": "2"}, stopper=stopper)
+ assert calls["next_job"] == 1 # first cycle paused (no poll), second cycle resumed (polled)
diff --git a/central/alembic/versions/0026_central_processing_control.py b/central/alembic/versions/0026_central_processing_control.py
new file mode 100644
index 0000000..9e43c94
--- /dev/null
+++ b/central/alembic/versions/0026_central_processing_control.py
@@ -0,0 +1,24 @@
+"""Central processing control + sync bookkeeping per installation.
+
+Revision ID: 0026_central_processing_control
+Revises: 0025_tenant_external_cnpj
+"""
+from alembic import op
+import sqlalchemy as sa
+
+revision = "0026_central_processing_control"
+down_revision = "0025_tenant_external_cnpj"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+ op.add_column("installations", sa.Column("processing_enabled", sa.Boolean(), nullable=False, server_default="true"))
+ op.add_column("installations", sa.Column("sync_interval_seconds", sa.Integer(), nullable=False, server_default="1800"))
+ op.add_column("installations", sa.Column("last_sync_at", sa.DateTime(timezone=True), nullable=True))
+ op.add_column("installations", sa.Column("last_sync_changed", sa.Integer(), nullable=True))
+
+
+def downgrade():
+ for column in ("last_sync_changed", "last_sync_at", "sync_interval_seconds", "processing_enabled"):
+ op.drop_column("installations", column)
diff --git a/central/app/api.py b/central/app/api.py
index 50bc678..e9f4a7d 100644
--- a/central/app/api.py
+++ b/central/app/api.py
@@ -10,7 +10,7 @@ from .auth import authenticate, bootstrap
from .classification import ENGINE_VERSION, Classifier
from .config import settings
from .datasets.service import DatasetService
-from .jobs import claim_next_job, complete_job, create_job, fail_job
+from .jobs import SYNC_KINDS, claim_next_job, complete_job, create_job, fail_job, schedule_automatic_sync
from .blocked_products import list_blockers, resolve_blocker, resolve_product_blockers, upsert_blocker
from .context_keys import validate_entry
from .models import AgentSnapshot, BlockedProduct, FiscalContextEntry, ClassificationCache, ClassificationRequest, ClassificationResult, Dataset, Installation, Job
@@ -206,8 +206,13 @@ def router(session_factory, bootstrap_token):
session.add(agent); session.flush()
record_event(session, tenant_id=agent.tenant_id, installation_id=installation.id, agent_id=agent.id, severity="INFO", event_type="HEARTBEAT", message="Agent heartbeat", context={"status": agent.last_status, "agent_version": agent.agent_version})
session.commit()
+ # Central scheduler: exactly one INITIAL_SYNC after the first heartbeat, then
+ # one SYNC_INCREMENTAL per interval — idempotent, no local cron/toggle needed.
+ if agent.enabled:
+ schedule_automatic_sync(session, agent, installation, snapshot)
logger.info("heartbeat agent_id=%s status=%s last_job_id=%s", agent.id, agent.last_status, payload.get("last_job_id"))
- return {"agent_id": agent.id, "installation_id": installation.installation_id, "last_seen_at": agent.last_seen_at.isoformat(), "enabled": agent.enabled}
+ return {"agent_id": agent.id, "installation_id": installation.installation_id, "last_seen_at": agent.last_seen_at.isoformat(), "enabled": agent.enabled,
+ "processing_enabled": bool(installation.processing_enabled), "sync_interval_seconds": installation.sync_interval_seconds}
finally: session.close()
@routes.post("/agents/register")
@@ -332,6 +337,36 @@ def router(session_factory, bootstrap_token):
"attempts": job.attempts, "result_summary": job.result_summary, "error": job.error,
}
+ def _apply_job_outcome(session, agent, installation, job, result_summary, *, failed):
+ """After a job finishes: coverage snapshot (from the Agent's operational
+ summary), sync bookkeeping and an audit-friendly event. Never stores secrets."""
+ now = datetime.now(timezone.utc)
+ snapshot = session.scalar(select(AgentSnapshot).where(AgentSnapshot.installation_id == installation.id))
+ if snapshot is None:
+ snapshot = AgentSnapshot(installation_id=installation.id); session.add(snapshot)
+ snapshot.last_job_status = "FAILED" if failed else "COMPLETED"
+ snapshot.last_job_at = now
+ context = {"kind": job.kind}
+ operational = result_summary.get("operational") if not failed else None
+ if operational and job.kind in SYNC_KINDS:
+ total = int(operational.get("TOTAL_PRODUCTS", 0))
+ complete = int(operational.get("COMPLETE", 0)) + int(operational.get("CORRECTED", 0))
+ snapshot.total_products = total
+ snapshot.fiscally_complete = complete
+ snapshot.safe_blocked = int(operational.get("SAFE_BLOCKED", 0))
+ snapshot.failed_products = int(operational.get("FAILED_TECHNICAL", 0))
+ snapshot.coverage_percent = round(100.0 * complete / total, 2) if total else 0.0
+ snapshot.blockers = {state: int(count) for state, count in operational.items() if state.startswith("BLOCKED_") and count}
+ snapshot.last_successful_job_at = now
+ installation.last_sync_at = now
+ installation.last_sync_changed = int(result_summary.get("products_changed", result_summary.get("products_processed", 0)) or 0)
+ context.update({"total_products": total, "products_changed": installation.last_sync_changed, "products_processed": result_summary.get("products_processed")})
+ if failed:
+ context["error"] = result_summary.get("error")
+ record_event(session, tenant_id=agent.tenant_id, installation_id=installation.id, agent_id=agent.id, severity="ERROR" if failed else "INFO",
+ event_type="JOB_FAILED" if failed else "JOB_COMPLETED", message=f"Job {job.kind} {'failed' if failed else 'completed'}", context=context, job_id=job.id)
+ session.commit()
+
@routes.post("/jobs")
def create_job_route(payload: dict, identity=Depends(principal)):
session, agent = identity
@@ -353,6 +388,10 @@ def router(session_factory, bootstrap_token):
try:
job = claim_next_job(session, agent)
if job is None: return None
+ installation = session.scalar(select(Installation).where(Installation.agent_id == agent.id))
+ if installation is not None:
+ record_event(session, tenant_id=agent.tenant_id, installation_id=installation.id, agent_id=agent.id, severity="INFO", event_type="JOB_STARTED", message=f"Job {job.kind} started", context={"kind": job.kind}, job_id=job.id)
+ session.commit()
logger.info("job claimed job_id=%s agent_id=%s kind=%s", job.id, agent.id, job.kind)
return _job_view(job)
finally: session.close()
@@ -388,6 +427,8 @@ def router(session_factory, bootstrap_token):
for product_code in result_summary.get("resolved_products", []):
resolve_product_blockers(session, agent.tenant_id, product_code, installation_id=installation.id if installation else None)
if not complete_job(session, job, result_summary): raise HTTPException(409, "job is not RUNNING")
+ if installation is not None:
+ _apply_job_outcome(session, agent, installation, job, result_summary, failed=False)
logger.info("job completed job_id=%s agent_id=%s result=%s", job_id, agent.id, payload.get("result_summary", {}))
return _job_view(session.get(Job, job_id))
finally: session.close()
@@ -399,6 +440,9 @@ def router(session_factory, bootstrap_token):
job = session.get(Job, job_id)
if job is None or job.agent_id != agent.id: raise HTTPException(404, "job not found")
if not fail_job(session, job, payload.get("error", "unknown error")): raise HTTPException(409, "job is not RUNNING")
+ installation = session.scalar(select(Installation).where(Installation.agent_id == agent.id))
+ if installation is not None:
+ _apply_job_outcome(session, agent, installation, job, {"error": str(payload.get("error", "unknown error"))[:500]}, failed=True)
logger.warning("job failed job_id=%s agent_id=%s error=%s", job_id, agent.id, payload.get("error", "unknown error"))
return _job_view(session.get(Job, job_id))
finally: session.close()
diff --git a/central/app/jobs.py b/central/app/jobs.py
index 45ab8aa..5e5e899 100644
--- a/central/app/jobs.py
+++ b/central/app/jobs.py
@@ -1,31 +1,74 @@
-"""Minimal job queue for Agent orchestration: create, atomically claim, complete, fail."""
-from datetime import datetime, timezone
+"""Job queue for Agent orchestration: create (idempotent per kind), claim one job at
+a time per Agent, complete, fail. Also the central sync scheduler helpers.
+
+Every fiscal job is READ-ONLY on the ERP (INITIAL_SYNC, SYNC_INCREMENTAL,
+REPROCESS_ALL, REPROCESS_PRODUCTS); automation never authorizes Firebird writes.
+"""
+from datetime import datetime, timedelta, timezone
from sqlalchemy import select, update
-from .models import Job
+from .models import Installation, Job
-_VALID_KINDS = {"REPROCESS_ALL", "REPROCESS_PRODUCTS"}
+FULL_KINDS = ("INITIAL_SYNC", "REPROCESS_ALL") # read every product, reclassify all
+SYNC_KINDS = FULL_KINDS + ("SYNC_INCREMENTAL",) # jobs that count as "a sync happened"
+_VALID_KINDS = set(SYNC_KINDS) | {"REPROCESS_PRODUCTS"}
_MAX_ERROR_LENGTH = 2000
+STALE_RUNNING_AFTER = timedelta(hours=1) # an Agent that died mid-job must not block the queue forever
+
+
+def _aware(value):
+ if value is not None and value.tzinfo is None:
+ return value.replace(tzinfo=timezone.utc)
+ return value
def create_job(session, agent, kind, payload=None):
+ """Create a job. Sync-type jobs are idempotent: if this Agent already has a
+ PENDING/RUNNING job that covers the request, that job is returned instead
+ (no duplicates, never two full jobs at once)."""
if kind not in _VALID_KINDS:
raise ValueError(f"unsupported job kind: {kind}")
+ if kind in SYNC_KINDS:
+ # A full job subsumes an incremental one, not the other way round: asking for a
+ # full reprocess while only an incremental is queued must still queue the full.
+ blocking = FULL_KINDS if kind in FULL_KINDS else SYNC_KINDS
+ existing = session.scalar(select(Job).where(Job.agent_id == agent.id, Job.kind.in_(blocking), Job.status.in_(("PENDING", "RUNNING"))).order_by(Job.id).limit(1))
+ if existing is not None:
+ return existing
job = Job(agent_id=agent.id, kind=kind, status="PENDING", payload=payload or {})
session.add(job)
session.commit()
return job
+def recover_stale_running(session, agent, now=None):
+ now = now or datetime.now(timezone.utc)
+ cutoff = now - STALE_RUNNING_AFTER
+ stale = session.scalars(select(Job).where(Job.agent_id == agent.id, Job.status == "RUNNING")).all()
+ changed = False
+ for job in stale:
+ if job.started_at is not None and _aware(job.started_at) < cutoff:
+ job.status, job.finished_at, job.error = "FAILED", now, "stale: Agent stopped reporting while RUNNING"
+ changed = True
+ if changed:
+ session.commit()
+
+
def claim_next_job(session, agent):
- """Atomically claim the oldest PENDING job for this agent only.
- A single UPDATE...WHERE status='PENDING' guards against two concurrent
- pollers (this agent or any other) claiming the same job twice."""
+ """Atomically claim the oldest PENDING job for this agent only — and only when
+ processing is enabled centrally and no other job of this Agent is RUNNING
+ (one fiscal job at a time per Firebird)."""
+ installation = session.scalar(select(Installation).where(Installation.agent_id == agent.id))
+ if installation is not None and not installation.processing_enabled:
+ return None
+ recover_stale_running(session, agent)
+ if session.scalar(select(Job.id).where(Job.agent_id == agent.id, Job.status == "RUNNING").limit(1)) is not None:
+ return None
candidate = session.scalar(
select(Job.id)
.where(Job.agent_id == agent.id, Job.status == "PENDING")
- .order_by(Job.created_at.asc())
+ .order_by(Job.created_at.asc(), Job.id.asc())
.limit(1)
)
if candidate is None:
@@ -61,3 +104,25 @@ def fail_job(session, job, error):
)
session.commit()
return result.rowcount == 1
+
+
+def schedule_automatic_sync(session, agent, installation, snapshot, now=None):
+ """Called from the Agent heartbeat. Central scheduler, idempotent:
+ - first heartbeat of an installation with no initial coverage and no sync job
+ ever created -> exactly ONE INITIAL_SYNC;
+ - otherwise, if processing is enabled and the last sync attempt is older than
+ the installation's interval, and nothing is queued/running -> ONE SYNC_INCREMENTAL.
+ Returns the created job or None."""
+ if installation is None or not installation.processing_enabled:
+ return None
+ now = now or datetime.now(timezone.utc)
+ sync_jobs = session.scalars(select(Job).where(Job.agent_id == agent.id, Job.kind.in_(SYNC_KINDS)).order_by(Job.id.desc())).all()
+ if any(job.status in ("PENDING", "RUNNING") for job in sync_jobs):
+ return None
+ never_synced = not any(job.kind == "INITIAL_SYNC" or job.kind == "REPROCESS_ALL" for job in sync_jobs) and not int(getattr(snapshot, "total_products", 0) or 0)
+ if never_synced and not sync_jobs:
+ return create_job(session, agent, "INITIAL_SYNC", {"reason": "INITIAL_SYNC"})
+ last_attempt = max((_aware(job.finished_at or job.created_at) for job in sync_jobs if (job.finished_at or job.created_at)), default=None)
+ if last_attempt is not None and (now - last_attempt).total_seconds() >= installation.sync_interval_seconds:
+ return create_job(session, agent, "SYNC_INCREMENTAL", {"reason": "SCHEDULED"})
+ return None
diff --git a/central/app/models.py b/central/app/models.py
index 1306b5d..8043b2e 100644
--- a/central/app/models.py
+++ b/central/app/models.py
@@ -276,6 +276,12 @@ class Installation(Base):
started_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
last_seen_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True, index=True)
last_status: Mapped[str | None] = mapped_column(String(32), nullable=True)
+ # Central execution control (no local .env toggle): the Hub decides whether this
+ # installation's Agent executes jobs, and how often the automatic incremental sync runs.
+ processing_enabled: Mapped[bool] = mapped_column(Boolean, default=True, server_default="true")
+ sync_interval_seconds: Mapped[int] = mapped_column(Integer, default=1800, server_default="1800")
+ last_sync_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
+ last_sync_changed: Mapped[int | None] = mapped_column(Integer, nullable=True)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now())
diff --git a/central/app/version.py b/central/app/version.py
index 89bfd78..0a21bae 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.1.2"
+GTI_FISCAL_VERSION = "1.1.3"
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/app/web.py b/central/app/web.py
index f0a3f3e..207f0a3 100644
--- a/central/app/web.py
+++ b/central/app/web.py
@@ -279,9 +279,51 @@ def router(session_factory):
for event in events:
event["occurred_display"] = format_ui_datetime(datetime.fromisoformat(event["occurred_at"]) if event["occurred_at"] else None)
agent_blockers = list_blockers(session, tenant_id=installation.tenant_id, agent_id=agent_id, page=1, page_size=25)
- return render(request, "agent_detail.html", {"detail": detail, "tab": tab, "blockers": agent_blockers, "blocker_breakdown": blocker_filter_summary(session, tenant_id=installation.tenant_id, agent_id=agent_id, status="OPEN")["blocker_breakdown"], "jobs": list_jobs(session, agent_id=agent_id, page=1, page_size=25), "events": events, "event_page": max(1, event_page), "has_more": has_more, "severity": severity or "", "event_type": event_type or "", "job_id": job_id or "", "product_code": product_code or "", "since": since or "", "until": until or "", "event_query_base": query_base(tab=tab, severity=severity, event_type=event_type, job_id=job_id, product_code=product_code, since=since, until=until), "auth": auth_data, "csrf_token": request.cookies.get("gti_csrf", "")})
+ return render(request, "agent_detail.html", {"detail": detail, "tab": tab, "blockers": agent_blockers, "blocker_breakdown": blocker_filter_summary(session, tenant_id=installation.tenant_id, agent_id=agent_id, status="OPEN")["blocker_breakdown"], "jobs": list_jobs(session, agent_id=agent_id, page=1, page_size=25), "events": events, "event_page": max(1, event_page), "has_more": has_more, "severity": severity or "", "event_type": event_type or "", "job_id": job_id or "", "product_code": product_code or "", "since": since or "", "until": until or "", "event_query_base": query_base(tab=tab, severity=severity, event_type=event_type, job_id=job_id, product_code=product_code, since=since, until=until), "auth": auth_data, "csrf_token": request.cookies.get("gti_csrf", ""), "ops": _ops_view(session, installation, agent)})
finally: session.close()
+ def _ops_view(session, installation, agent):
+ """Processing status shown on the Agent detail (no clutter: 5 facts)."""
+ last_job = session.scalar(select(Job).where(Job.agent_id == agent.id).order_by(Job.id.desc()).limit(1))
+ last_sync = _aware(installation.last_sync_at)
+ next_sync = None
+ if installation.processing_enabled and last_sync is not None:
+ next_sync = last_sync + timedelta(seconds=installation.sync_interval_seconds)
+ return {"enabled": bool(installation.processing_enabled), "interval_minutes": installation.sync_interval_seconds // 60,
+ "last_sync": format_ui_datetime(last_sync) if last_sync else None, "next_sync": format_ui_datetime(next_sync) if next_sync else None,
+ "last_changed": installation.last_sync_changed, "last_job": {"kind": last_job.kind, "status": last_job.status, "id": last_job.id} if last_job else None}
+
+ async def _installation_action(request, agent_id, action):
+ 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"))
+ agent = session.get(Agent, agent_id)
+ installation = session.scalar(select(Installation).where(Installation.agent_id == agent_id)) if agent else None
+ if not agent or not installation: return HTMLResponse("Agent not found", status_code=404)
+ if action == "pause": installation.processing_enabled = False
+ elif action == "resume": installation.processing_enabled = True
+ elif action == "sync-now":
+ job = create_job(session, agent, "SYNC_INCREMENTAL", {"reason": "MANUAL_SYNC_NOW", "requested_by": admin.id})
+ record_audit(session, admin_user_id=admin.id, tenant_id=installation.tenant_id, action="SYNC_NOW", target_type="installation", target_id=installation.installation_id, context={"job_id": job.id, "kind": job.kind})
+ record_audit(session, admin_user_id=admin.id, tenant_id=installation.tenant_id, action=f"PROCESSING_{action.upper().replace('-', '_')}", target_type="installation", target_id=installation.installation_id, context={"processing_enabled": installation.processing_enabled}) if action in ("pause", "resume") else None
+ session.commit()
+ return RedirectResponse(f"/agents/{agent_id}?tab=jobs", status_code=303)
+ finally: session.close()
+
+ @routes.post("/agents/{agent_id}/sync-now")
+ async def sync_now(request: Request, agent_id: int):
+ return await _installation_action(request, agent_id, "sync-now")
+
+ @routes.post("/agents/{agent_id}/pause")
+ async def pause_processing(request: Request, agent_id: int):
+ return await _installation_action(request, agent_id, "pause")
+
+ @routes.post("/agents/{agent_id}/resume")
+ async def resume_processing(request: Request, agent_id: int):
+ return await _installation_action(request, agent_id, "resume")
+
@routes.get("/agents/{agent_id}/reprocess-confirmation", response_class=HTMLResponse)
def reprocess_confirmation(request: Request, agent_id: int):
session, auth_data, redirect = page_auth(request)
diff --git a/central/pyproject.toml b/central/pyproject.toml
index 6e6376c..bf81698 100644
--- a/central/pyproject.toml
+++ b/central/pyproject.toml
@@ -1,6 +1,6 @@
[project]
name = "gti-fiscal-hub"
-version = "1.1.2"
+version = "1.1.3"
requires-python = ">=3.10"
dependencies = [
"fastapi>=0.115,<1",
diff --git a/central/templates/agent_detail.html b/central/templates/agent_detail.html
index 2101ff4..17ac25d 100644
--- a/central/templates/agent_detail.html
+++ b/central/templates/agent_detail.html
@@ -4,6 +4,11 @@
{{ detail.client or "Cliente não identificado" }} / INSTALAÇÃO
{{ detail.display_name or detail.hostname or detail.installation_id }} {{ detail.hostname or "hostname indisponível" }}
Saúde técnica: {{ detail.technical_health_label }} Status fiscal: {{ detail.fiscal_status_label }}
Produtos {{ detail.total_products if detail.total_products is not none else "Não informado" }} Completos {{ detail.fiscally_complete if detail.fiscally_complete is not none else "Não informado" }} Bloqueados com segurança {{ detail.safe_blocked if detail.safe_blocked is not none else "Não informado" }} Falhas técnicas {{ detail.failed_products if detail.failed_products is not none else "Não informado" }} Não classificados {{ detail.unclassified_products if detail.unclassified_products is not none else "Não informado" }} Cobertura fiscal {% if detail.fiscal_completion_percent is none %}Não informado nenhum snapshot recebido {% else %}{{ "%.1f"|format(detail.fiscal_completion_percent)|replace(".", ",") }}% {{ detail.fiscally_complete }} / {{ detail.total_products }} completos {% endif %}Cobertura ainda não informada: nenhum snapshot recebido deste servidor.
{% elif detail.coverage_invariant_pass %}Cobertura reconciliada para este servidor.
{% else %}Inconsistência de cobertura: {{ detail.coverage_overcount_products }} acima do total{% if detail.coverage_invalid_counts %}; existem contagens negativas{% endif %}.
{% endif %}
+Processamento Controlado pelo Hub — nenhuma configuração local.
Processamento {{ "Ativo" if ops.enabled else "Pausado" }} · sincronização automática a cada {{ ops.interval_minutes }} minÚltima sincronização {{ ops.last_sync or "ainda não sincronizado" }}{% if ops.last_changed is not none %} · {{ ops.last_changed }} produto(s) alterado(s){% endif %} Sincronização agendada {{ ops.next_sync or ("aguardando o primeiro sync" if ops.enabled else "pausado") }} Último job {% if ops.last_job %}{{ ops.last_job.kind }} · {{ ops.last_job.status }}{% else %}nenhum{% endif %} {% for name,label in [("overview","Resumo"),("fiscal","Fiscal"),("blockers","Blockers"),("jobs","Jobs"),("events","Eventos")] %}{{ label }} {% endfor %}
Cliente {{ detail.client or "Não informado" }} Nome de exibição {{ detail.display_name or "Não informado" }} Hostname {{ detail.hostname or "Não informado" }} Empresa {{ detail.company_name or detail.company_code or "Não informado" }} UF {{ detail.company_uf or "Não informado" }} CRT {{ detail.company_crt or "Não informado" }} Versão do agent {{ detail.agent_version_display }} Último heartbeat {{ detail.last_seen_display }} · {{ detail.last_seen_age }} Firebird {{ detail.firebird_version or "Não informado" }} · ODS {{ detail.firebird_ods or "Não informado" }} IDs técnicos Agent {{ detail.technical_ids.agent_id }} · instalação {{ detail.technical_ids.installation_db_id }} {% for event in events[:8] %}{% else %}Nenhum evento.
{% endfor %} .exe` **once** and types the install
+code. Nothing else happens on the client server — no PowerShell, no `agent.env` editing,
+no `services.msc`, no per-client cron:
+
+1. **Firebird discovery (read-only).** Setup finds the Firebird service and candidate
+ `.FDB` files and probes each one (read-only connection) for `C000025` + `C000004`.
+ Exactly one SGV database -> selected automatically. Several, none, or credentials
+ other than the SGV default -> the wizard asks *during the installation* (list / file
+ picker + user/password, verified before use). It never guesses among several.
+2. **Enrollment** with the install code, then three protected files (DPAPI LocalMachine +
+ SYSTEM/Administrators ACL): `credentials.bin` (Hub token), `firebird.bin` (Firebird
+ user/password, **never sent to the Hub**), and a non-secret `agent.env`.
+3. **Service** `GTIFiscalAgent`: Automatic Delayed Start, restart-on-failure, **started by
+ Setup**; it waits for the first heartbeat and tells the operator "nenhuma outra ação".
+4. **First sync is automatic.** On the first heartbeat of an installation without
+ coverage the Hub queues exactly one `INITIAL_SYNC` (idempotent). Afterwards the Hub's
+ scheduler queues one `SYNC_INCREMENTAL` per interval (default 30 min,
+ `installations.sync_interval_seconds`), only when nothing is queued/running.
+ `SYNC_INCREMENTAL` re-classifies only new/changed products (per-product source
+ fingerprint cache in the Agent's SQLite) and rebuilds coverage from the cache.
+ `REPROCESS_ALL` runs only when an operator asks for it.
+5. **Central control** (Agent detail page, no shell, no arbitrary commands):
+ *Sincronizar agora*, *Reprocessar tudo*, *Pausar / Retomar processamento*, plus status
+ (Ativo/Pausado, última/agendada sincronização, produtos alterados, último job).
+ Pause = the heartbeat keeps flowing but the Hub hands out no job and schedules none.
+ The old local `AGENT_JOBS_ENABLED` is ignored — there is no local toggle.
+
+Guarantees: one fiscal job per Agent at a time (FIFO queue, a job stuck RUNNING > 1 h is
+failed so it cannot block the queue), a failed sync backs off for the interval instead of
+looping, a Hub or Firebird outage fails the job cleanly and the service keeps running.
+Every job kind is **read-only on the ERP** (`FIREBIRD_WRITES=0`, `FISCAL_MODE=DRY_RUN`).
+
+Upgrading an installed host: run the newer Setup again (no code needed). It replaces the
+binaries, keeps `agent_id`/`installation_id`/credentials, removes the legacy local toggle and
+restarts the service.
diff --git a/installer/firebird_discovery.py b/installer/firebird_discovery.py
new file mode 100644
index 0000000..31096a9
--- /dev/null
+++ b/installer/firebird_discovery.py
@@ -0,0 +1,166 @@
+"""Safe, read-only Firebird/SGV discovery for Setup.exe.
+
+Finds candidate .FDB files, and for each one checks (read-only connection, no
+writes, ever) that it is an SGV database: it must contain both C000025
+(products) and C000004 (company). Exactly one compatible database is selected
+automatically; several are NEVER chosen silently — the caller (wizard) asks the
+operator during installation. Credentials are only ever used locally.
+
+Pure/injectable on purpose (`connect`, `finder`) so it is unit-testable without
+Firebird or a Windows host.
+"""
+from __future__ import annotations
+
+import os
+import subprocess
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Callable
+
+# The SGV/Firebird default. Tried silently; anything else is asked for inside the wizard.
+DEFAULT_CREDENTIALS: tuple[tuple[str, str], ...] = (("SYSDBA", "masterkey"),)
+REQUIRED_TABLES = ("C000025", "C000004")
+MAX_DEPTH = 4
+MIN_BYTES = 200 * 1024
+MAX_FILES = 300
+SKIP_DIRS = {"windows", "program files", "program files (x86)", "programdata", "$recycle.bin", "system volume information",
+ "appdata", "node_modules", ".git", "winsxs", "recovery", "perflogs", "users"}
+SKIP_FILES = {"security2.fdb", "security3.fdb", "security4.fdb", "security.fdb"}
+
+
+@dataclass
+class Candidate:
+ path: str
+ status: str # OK | NOT_SGV | AUTH_FAILED | ERROR
+ user: str | None = None
+ password: str | None = None
+ detail: str = ""
+
+
+@dataclass
+class Discovery:
+ state: str # SELECTED | AMBIGUOUS | AUTH_REQUIRED | NONE
+ compatible: list[Candidate] = field(default_factory=list)
+ auth_failed: list[Candidate] = field(default_factory=list)
+ scanned: int = 0
+ firebird_service: str | None = None
+
+ @property
+ def selected(self) -> Candidate | None:
+ return self.compatible[0] if self.state == "SELECTED" else None
+
+
+def fixed_drives() -> list[str]:
+ try:
+ import ctypes
+ mask = ctypes.windll.kernel32.GetLogicalDrives()
+ drives = []
+ for i in range(26):
+ root = f"{chr(65 + i)}:\\"
+ if mask & (1 << i) and ctypes.windll.kernel32.GetDriveTypeW(root) == 3: # DRIVE_FIXED
+ drives.append(root)
+ return drives
+ except Exception: # noqa: BLE001
+ return ["C:\\"]
+
+
+def find_fdb_files(roots: list[str], max_depth: int = MAX_DEPTH, limit: int = MAX_FILES) -> list[str]:
+ found: list[str] = []
+ for root in roots:
+ root = str(root)
+ base_depth = root.rstrip("\\/").count(os.sep)
+ for current, dirs, files in os.walk(root, onerror=lambda _e: None):
+ depth = current.rstrip("\\/").count(os.sep) - base_depth
+ dirs[:] = [d for d in dirs if d.lower() not in SKIP_DIRS] if depth < max_depth else []
+ for name in files:
+ lower = name.lower()
+ if lower.endswith(".fdb") and lower not in SKIP_FILES:
+ full = os.path.join(current, name)
+ try:
+ if os.path.getsize(full) >= MIN_BYTES:
+ found.append(full)
+ except OSError:
+ continue
+ if len(found) >= limit:
+ return found
+ return found
+
+
+def default_roots() -> list[str]:
+ roots: list[str] = []
+ for var in ("ProgramFiles", "ProgramFiles(x86)"):
+ base = os.environ.get(var)
+ if base:
+ roots.extend(str(p) for p in Path(base).glob("Firebird*"))
+ roots.extend(fixed_drives())
+ return roots
+
+
+def firebird_service() -> str | None:
+ try:
+ out = subprocess.run(["sc.exe", "query", "type=", "service", "state=", "all"], capture_output=True, text=True, timeout=20).stdout
+ except Exception: # noqa: BLE001
+ return None
+ for line in out.splitlines():
+ if line.strip().upper().startswith("SERVICE_NAME") and "FIREBIRD" in line.upper():
+ return line.split(":", 1)[1].strip()
+ return None
+
+
+def _default_connect(path: str, user: str, password: str):
+ import fdb
+ return fdb.connect(database=path, user=user, password=password, isolation_level=fdb.ISOLATION_LEVEL_READ_COMMITED_RO)
+
+
+def probe(path: str, user: str, password: str, *, connect: Callable | None = None) -> Candidate:
+ connect = connect or _default_connect
+ try:
+ conn = connect(path, user, password)
+ except Exception as error: # noqa: BLE001
+ text = str(error).lower()
+ if "user name" in text or "password" in text or "not defined" in text:
+ return Candidate(path, "AUTH_FAILED", detail="credenciais recusadas")
+ return Candidate(path, "ERROR", detail=type(error).__name__)
+ try:
+ cursor = conn.cursor()
+ cursor.execute("SELECT TRIM(RDB$RELATION_NAME) FROM RDB$RELATIONS WHERE RDB$RELATION_NAME IN ('C000025','C000004')")
+ have = {row[0] for row in cursor.fetchall()}
+ except Exception as error: # noqa: BLE001
+ return Candidate(path, "ERROR", detail=type(error).__name__)
+ finally:
+ try:
+ conn.close()
+ except Exception: # noqa: BLE001
+ pass
+ if all(table in have for table in REQUIRED_TABLES):
+ return Candidate(path, "OK", user, password)
+ return Candidate(path, "NOT_SGV", detail="sem C000025/C000004")
+
+
+def discover(*, roots: list[str] | None = None, credentials=DEFAULT_CREDENTIALS, connect: Callable | None = None,
+ finder: Callable | None = None) -> Discovery:
+ finder = finder or find_fdb_files
+ files = finder(roots if roots is not None else default_roots())
+ compatible: list[Candidate] = []
+ auth_failed: list[Candidate] = []
+ for path in files:
+ result = None
+ for user, password in credentials:
+ result = probe(path, user, password, connect=connect)
+ if result.status != "AUTH_FAILED":
+ break
+ if result is None:
+ continue
+ if result.status == "OK":
+ compatible.append(result)
+ elif result.status == "AUTH_FAILED":
+ auth_failed.append(result)
+ if len(compatible) == 1:
+ state = "SELECTED"
+ elif len(compatible) > 1:
+ state = "AMBIGUOUS"
+ elif auth_failed:
+ state = "AUTH_REQUIRED"
+ else:
+ state = "NONE"
+ return Discovery(state, compatible, auth_failed, len(files), firebird_service())
diff --git a/installer/install_common.py b/installer/install_common.py
index 145db42..1d69fd1 100644
--- a/installer/install_common.py
+++ b/installer/install_common.py
@@ -13,9 +13,11 @@ Policy enforced here (mirrors CLAUDE.md):
"""
from __future__ import annotations
+import json
import os
import subprocess
import sys
+import time
import winreg
from pathlib import Path
@@ -64,7 +66,7 @@ def enroll(install_code: str, *, hub: str | None = None) -> dict:
return response.json()
-def write_agent_env(install_dir_path: Path, enroll_result: dict) -> None:
+def write_agent_env(install_dir_path: Path, enroll_result: dict, firebird_path: str = "") -> None:
"""Write the non-secret agent.env. The permanent credential goes to
credentials.bin (DPAPI), never here. FIREBIRD_WRITES=0 / DRY_RUN always."""
from gti_fiscal_agent.runner import REFERENCE_SCHEMA_V1
@@ -76,7 +78,8 @@ def write_agent_env(install_dir_path: Path, enroll_result: dict) -> None:
(f"INSTALLATION_ID={enroll_result.get('installation_id') or ''}"),
("COMPANY_ID="),
("COMPANY_UF="),
- ("FIREBIRD_DATABASE_PATH="),
+ (f"FIREBIRD_DATABASE_PATH={firebird_path}"),
+ # Firebird user/password live ONLY in the DPAPI-protected firebird.bin (never here, never sent to the Hub).
("FB_USER="),
("FB_PASSWORD="),
("FISCAL_MODE=DRY_RUN"),
@@ -103,6 +106,72 @@ def write_credential(install_dir_path: Path, agent_token: str) -> None:
_write(install / "credentials.bin", agent_token)
+def write_firebird_credential(install_dir_path: Path, user: str, password: str) -> None:
+ """Firebird user/password -> firebird.bin via DPAPI LocalMachine + ACL (fail closed).
+ Local only: this never reaches the Hub and never lands in agent.env."""
+ from gti_fiscal_agent.credential_store import write_credential as _write
+
+ install = Path(install_dir_path)
+ install.mkdir(parents=True, exist_ok=True)
+ _write(install / "firebird.bin", json.dumps({"user": user, "password": password}))
+
+
+def detect_existing(install_dir_path: Path) -> dict | None:
+ """An already-enrolled installation (identity + DPAPI credential present)."""
+ install = Path(install_dir_path)
+ env_file, cred = install / "agent.env", install / "credentials.bin"
+ if not env_file.exists() or not cred.exists():
+ return None
+ values = {}
+ for line in env_file.read_text(encoding="utf-8", errors="replace").splitlines():
+ if "=" in line and not line.lstrip().startswith("#"):
+ key, _, value = line.partition("=")
+ values[key.strip()] = value.strip()
+ if not values.get("AGENT_ID"):
+ return None
+ return {"agent_id": values["AGENT_ID"], "installation_id": values.get("INSTALLATION_ID"), "hub_url": values.get("GTI_HUB_URL") or hub_url(), "env": values}
+
+
+def upgrade_agent_env(install_dir_path: Path, agent_version: str) -> None:
+ """Migrate an existing agent.env to the central-control model: drop the legacy local
+ AGENT_JOBS_ENABLED toggle, refresh AGENT_VERSION, keep every other value (identity, Firebird path...)."""
+ env_file = Path(install_dir_path) / "agent.env"
+ kept, seen_version = [], False
+ for line in env_file.read_text(encoding="utf-8", errors="replace").splitlines():
+ key = line.partition("=")[0].strip()
+ if key == "AGENT_JOBS_ENABLED":
+ continue
+ if key == "AGENT_VERSION":
+ line, seen_version = f"AGENT_VERSION={agent_version}", True
+ kept.append(line)
+ if not seen_version:
+ kept.append(f"AGENT_VERSION={agent_version}")
+ env_file.write_text(chr(10).join(kept) + chr(10), encoding="utf-8")
+
+
+def wait_service_state(wanted: str, timeout: float = 45.0) -> bool:
+ deadline = time.time() + timeout
+ while time.time() < deadline:
+ if wanted in _sc(["query", SERVICE_NAME]).stdout:
+ return True
+ time.sleep(1.5)
+ return False
+
+
+def wait_for_heartbeat(agent_token: str, hub: str, timeout: float = 120.0) -> bool:
+ """True once the Hub sees this Agent ONLINE (first heartbeat after the service started)."""
+ deadline = time.time() + timeout
+ while time.time() < deadline:
+ try:
+ response = httpx.get((hub or hub_url()).rstrip("/") + "/v1/agents/me", headers={"Authorization": f"Bearer {agent_token}"}, timeout=10)
+ if response.status_code == 200 and response.json().get("status") == "ONLINE":
+ return True
+ except Exception: # noqa: BLE001
+ pass
+ time.sleep(4)
+ return False
+
+
def _sc(args: list[str]) -> subprocess.CompletedProcess[str]:
return subprocess.run(["sc.exe", *args], capture_output=True, text=True)
diff --git a/installer/setup_main.py b/installer/setup_main.py
index 1396cc5..1758609 100644
--- a/installer/setup_main.py
+++ b/installer/setup_main.py
@@ -1,26 +1,23 @@
-"""Setup.exe — universal Windows installer for the GTI Fiscal Agent.
+"""Setup.exe — universal, zero-touch Windows installer for the GTI Fiscal Agent.
-Two entry modes:
-- GUI (default): tkinter wizard with a single install-code field.
-- CLI (`--cli --code XXXX [--hub URL] [--install-dir PATH] [--no-service]`):
- scripted install used by the automated E2E and the build smoke test.
+The operator runs it ONCE. Nothing has to be edited, started or scheduled afterwards.
-Run elevated (PyInstaller --uac-admin). Enrollment is resolved by the Hub from
-the install-code hash alone — no enrollment token, no bundle.
+Flow (GUI wizard, or `--cli`):
+ 1. discover the SGV Firebird database (read-only): exactly one compatible
+ database is selected automatically; several/none/unknown credentials ->
+ the wizard asks the operator DURING the installation (never silently guessed)
+ 2. POST /v1/agents/enroll {install_code, hostname, agent_version, machine_fingerprint}
+ 3. agent.env (non-secret only) + credentials.bin (Hub token) + firebird.bin
+ (Firebird user/password) — DPAPI LocalMachine + SYSTEM/Administrators ACL
+ 4. extract the service host + Uninstall.exe + write executor
+ 5. register the service (Automatic Delayed Start + restart on failure) and START it
+ 6. wait for the first heartbeat; the Hub then creates the INITIAL_SYNC by itself
-Flow on success:
- 1. POST /v1/agents/enroll {install_code, hostname, agent_version, machine_fingerprint}
- 2. write agent.env (non-secret only) + credentials.bin (DPAPI LocalMachine + ACL)
- 3. extract the bundled Service host (onedir) + Uninstall.exe + write executor
- into the install directory
- 4. register service GTIFiscalAgent (Delayed Auto) + failure recovery
- 5. Firebird not configured -> report "serviço instalado; configure o arquivo
- de banco para ativar" and do NOT start the service (no error loop)
- 6. write ARP uninstall key
- 7. best-effort WINDOWS_INSTALL_COMPLETED audit
+Re-running Setup on an already-enrolled host (no code needed) UPGRADES it in place:
+same identity and credentials, legacy local toggles removed, service restarted.
-The install code is never persisted; agent.env never holds the token; no UI or
-flag enables Firebird/AI writes (always dry-run, writes disabled).
+The install code is never persisted; the Firebird password never goes to the Hub;
+no UI or flag enables Firebird/AI writes (always dry-run, writes disabled).
"""
from __future__ import annotations
@@ -31,22 +28,61 @@ import sys
import threading
import tkinter as tk
from pathlib import Path
-from tkinter import messagebox, ttk
+from tkinter import filedialog, messagebox, ttk
+from typing import Callable
import httpx
+import firebird_discovery as fbd
import install_common as common
MEIPASS = Path(getattr(sys, "_MEIPASS", Path(__file__).resolve().parent))
+Chooser = Callable[["fbd.Discovery"], "dict | None"]
-def install_flow(code: str, *, hub: str | None = None, install_dir_path: str | None = None, register_service: bool = True) -> dict:
- """Perform the full install. Returns the enroll result dict; raises RuntimeError
- with a user-facing message on any failure."""
+def setup_version() -> str:
+ for candidate in (MEIPASS / "VERSION", Path(__file__).resolve().parents[1] / "release" / "VERSION"):
+ if candidate.exists():
+ return candidate.read_text(encoding="utf-8").strip()
+ return os.environ.get("AGENT_VERSION", "")
+
+
+def resolve_firebird(*, explicit: dict | None, chooser: Chooser | None, progress: Callable[[str], None] = lambda _m: None) -> dict:
+ """Return {"path","user","password"} for a verified SGV database, or raise RuntimeError."""
+ if explicit:
+ checked = fbd.probe(explicit["path"], explicit["user"], explicit["password"])
+ if checked.status != "OK":
+ raise RuntimeError(f"banco Firebird informado não é um banco SGV utilizável ({checked.status}: {checked.detail})")
+ return explicit
+ progress("Procurando o banco de dados do SGV (Firebird)…")
+ found = fbd.discover()
+ if found.state == "SELECTED":
+ chosen = found.selected
+ return {"path": chosen.path, "user": chosen.user, "password": chosen.password}
+ if chooser is None:
+ detail = {"AMBIGUOUS": f"{len(found.compatible)} bancos SGV encontrados", "AUTH_REQUIRED": "usuário/senha do Firebird necessários", "NONE": "nenhum banco SGV encontrado"}[found.state]
+ raise RuntimeError(f"seleção do banco Firebird necessária ({detail}); informe --fdb, --fb-user e --fb-password")
+ choice = chooser(found)
+ if not choice:
+ raise RuntimeError("instalação cancelada: nenhum banco Firebird selecionado")
+ checked = fbd.probe(choice["path"], choice["user"], choice["password"])
+ if checked.status != "OK":
+ raise RuntimeError(f"banco selecionado não é utilizável ({checked.status}: {checked.detail})")
+ return choice
+
+
+def install_flow(code: str, *, hub: str | None = None, install_dir_path: str | None = None, register_service: bool = True,
+ firebird: dict | None = None, chooser: Chooser | None = None, require_firebird: bool = True,
+ wait_heartbeat: bool = True, progress: Callable[[str], None] = lambda _m: None) -> dict:
+ """Perform the full zero-touch install. Returns a result dict; raises RuntimeError with a
+ user-facing message on any failure (nothing is enrolled if Firebird selection fails)."""
if install_dir_path:
os.environ["GTI_INSTALL_DIR"] = install_dir_path
install = common.install_dir()
+ chosen = resolve_firebird(explicit=firebird, chooser=chooser, progress=progress) if require_firebird else None
+
+ progress("Enviando o código de instalação ao hub…")
try:
result = common.enroll(code, hub=hub)
except httpx.HTTPStatusError as error:
@@ -60,29 +96,71 @@ def install_flow(code: str, *, hub: str | None = None, install_dir_path: str | N
except Exception as error: # noqa: BLE001
raise RuntimeError(f"falha no enrollment: {error}") from error
- agent_version = str(result.get("agent_version") or os.environ.get("AGENT_VERSION") or "")
+ agent_version = str(result.get("agent_version") or os.environ.get("AGENT_VERSION") or setup_version())
os.environ["AGENT_VERSION"] = agent_version
- common.write_agent_env(install, result)
+ common.write_agent_env(install, result, firebird_path=chosen["path"] if chosen else "")
common.write_credential(install, result["agent_token"])
+ if chosen:
+ common.write_firebird_credential(install, chosen["user"], chosen["password"])
+ progress("Instalando o serviço…")
extract_payload(install)
+ started = heartbeat = False
if register_service:
common.install_service(install)
common.write_arp_uninstall(install)
- if common.service_running():
- common.start_service()
+ if chosen: # without a Firebird database the service would only fail preflight in a loop
+ progress("Iniciando o serviço…")
+ if not common.service_running():
+ common.start_service()
+ started = common.wait_service_state("RUNNING", timeout=45)
+ if started and wait_heartbeat:
+ progress("Aguardando o primeiro heartbeat…")
+ heartbeat = common.wait_for_heartbeat(result["agent_token"], result.get("hub_url") or hub, timeout=120)
try:
- common.audit_event(result["agent_token"], result.get("hub_url") or hub, "WINDOWS_INSTALL_COMPLETED", f"agent {result.get('agent_id')} installed")
+ common.audit_event(result["agent_token"], result.get("hub_url") or hub, "WINDOWS_INSTALL_COMPLETED", f"agent {result.get('agent_id')} installed; service_running={started} heartbeat={heartbeat}")
except Exception: # noqa: BLE001 — best effort
pass
- result["install_dir"] = str(install)
- result["agent_version"] = agent_version
+ result.update(install_dir=str(install), agent_version=agent_version, firebird_path=chosen["path"] if chosen else None,
+ service_running=started, heartbeat=heartbeat, mode="INSTALL")
return result
+def upgrade_flow(*, install_dir_path: str | None = None, progress: Callable[[str], None] = lambda _m: None) -> dict:
+ """Re-run on an enrolled host: replace the binaries in place, keep identity/credentials,
+ migrate agent.env to the central-control model, restart the service."""
+ if install_dir_path:
+ os.environ["GTI_INSTALL_DIR"] = install_dir_path
+ install = common.install_dir()
+ existing = common.detect_existing(install)
+ if existing is None:
+ raise RuntimeError("nenhuma instalação existente encontrada para atualizar")
+ version = setup_version()
+ progress("Atualizando o Agent (identidade e credenciais preservadas)…")
+ if common.service_installed():
+ common.stop_service()
+ common.wait_service_state("STOPPED", timeout=45)
+ extract_payload(install)
+ common.upgrade_agent_env(install, version)
+ common.install_service(install)
+ common.write_arp_uninstall(install)
+ os.environ["AGENT_VERSION"] = version
+ progress("Iniciando o serviço…")
+ common.start_service()
+ started = common.wait_service_state("RUNNING", timeout=45)
+ heartbeat = False
+ try:
+ from gti_fiscal_agent.credential_store import read_credential
+ heartbeat = started and common.wait_for_heartbeat(read_credential(install / "credentials.bin"), existing["hub_url"], timeout=120)
+ except Exception: # noqa: BLE001
+ pass
+ return {"agent_id": existing["agent_id"], "installation_id": existing["installation_id"], "install_dir": str(install), "agent_version": version,
+ "service_running": started, "heartbeat": heartbeat, "mode": "UPGRADE", "hub_url": existing["hub_url"]}
+
+
def extract_payload(install: Path) -> None:
"""Copy the bundled service host onedir, Uninstall.exe and the write
executor out of the onefile extraction dir into the install directory."""
@@ -103,72 +181,151 @@ def extract_payload(install: Path) -> None:
shutil.copy2(source, dest)
+def final_message(result: dict) -> str:
+ if result.get("service_running") and result.get("heartbeat"):
+ return ("Instalação concluída.\n\nO serviço GTI Fiscal Agent está em execução e já se comunicou com o painel. "
+ "A primeira sincronização começa automaticamente.\n\nNenhuma outra ação é necessária neste servidor.")
+ if result.get("service_running"):
+ return ("Instalação concluída e serviço em execução, mas o primeiro heartbeat ainda não foi confirmado.\n\n"
+ "O serviço continua tentando sozinho; acompanhe pelo painel. Nenhuma edição de arquivo é necessária.")
+ return ("Instalação concluída, mas o serviço não iniciou.\n\nVerifique o log em "
+ f"{result.get('install_dir')}\\logs\\agent.log ou contate o suporte.")
+
+
class SetupWizard:
def __init__(self, root: tk.Tk):
self.root = root
root.title("GTI Fiscal Agent")
root.resizable(False, False)
- root.geometry("720x250")
+ root.geometry("720x300")
+ self.existing = common.detect_existing(common.install_dir())
frame = ttk.Frame(root, padding=20)
frame.pack(fill="both", expand=True)
-
ttk.Label(frame, text="GTI Fiscal Agent — Instalação", font=("Segoe UI", 13, "bold")).pack(anchor="w")
- ttk.Label(frame, text="Digite o código de instalação gerado no painel GTI Fiscal.").pack(anchor="w", pady=(4, 2))
- ttk.Label(frame, text=f"Hub: {common.hub_url()}", foreground="#555").pack(anchor="w")
-
self.code_var = tk.StringVar()
- ttk.Entry(frame, textvariable=self.code_var, width=26, font=("Consolas", 14), justify="center").pack(anchor="w", pady=(14, 4))
-
- self.status = ttk.Label(frame, text="", foreground="#333")
+ if self.existing:
+ ttk.Label(frame, text="Este servidor já está instalado. Clique em Atualizar para aplicar a versão mais recente\n(identidade e credenciais são preservadas).").pack(anchor="w", pady=(6, 2))
+ else:
+ ttk.Label(frame, text="Digite o código de instalação gerado no painel GTI Fiscal.").pack(anchor="w", pady=(4, 2))
+ ttk.Label(frame, text=f"Hub: {common.hub_url()}", foreground="#555").pack(anchor="w")
+ ttk.Entry(frame, textvariable=self.code_var, width=26, font=("Consolas", 14), justify="center").pack(anchor="w", pady=(14, 4))
+ self.status = ttk.Label(frame, text="", foreground="#333", wraplength=660, justify="left")
self.status.pack(anchor="w", pady=(6, 0))
-
buttons = ttk.Frame(frame)
buttons.pack(anchor="e", pady=(10, 0))
- self.install_btn = ttk.Button(buttons, text="Instalar", command=self.on_install)
+ self.install_btn = ttk.Button(buttons, text="Atualizar" if self.existing else "Instalar", command=self.on_install)
self.install_btn.pack(side="left", padx=(0, 6))
ttk.Button(buttons, text="Fechar", command=root.destroy).pack(side="left")
-
root.bind("", lambda _event: self.on_install())
- def _set_busy(self, busy: bool, text: str = ""):
- self.status.config(text=text)
- self.install_btn.config(state="disabled" if busy else "normal")
+ def _say(self, text: str, color: str = "#333"):
+ self.root.after(0, lambda: self.status.config(text=text, foreground=color))
+
+ def _busy(self, busy: bool):
+ self.root.after(0, lambda: self.install_btn.config(state="disabled" if busy else "normal"))
def on_install(self):
code = self.code_var.get().strip()
- if not code:
+ if not self.existing and not code:
self.status.config(text="Informe o código de instalação.", foreground="#b00")
return
- self._set_busy(True, "Enviando código ao hub…")
+ self._busy(True)
threading.Thread(target=self._run, args=(code,), daemon=True).start()
+ def choose_firebird(self, found: "fbd.Discovery") -> dict | None:
+ """Runs on the worker thread; shows the selection dialog on the UI thread and waits."""
+ holder: dict = {}
+ done = threading.Event()
+ self.root.after(0, lambda: self._firebird_dialog(found, holder, done))
+ done.wait()
+ return holder.get("choice")
+
+ def _firebird_dialog(self, found: "fbd.Discovery", holder: dict, done: threading.Event):
+ win = tk.Toplevel(self.root)
+ win.title("Banco de dados do SGV")
+ win.geometry("640x330")
+ win.grab_set()
+ frame = ttk.Frame(win, padding=16)
+ frame.pack(fill="both", expand=True)
+ path_var, user_var, pass_var = tk.StringVar(), tk.StringVar(value="SYSDBA"), tk.StringVar()
+ msg = ttk.Label(frame, text="", foreground="#b00", wraplength=600)
+ if found.state == "AMBIGUOUS":
+ ttk.Label(frame, text="Foram encontrados vários bancos SGV. Selecione o correto:").pack(anchor="w")
+ box = tk.Listbox(frame, height=6)
+ for cand in found.compatible:
+ box.insert("end", cand.path)
+ box.pack(fill="x", pady=8)
+ box.selection_set(0)
+
+ def confirm():
+ if not box.curselection():
+ return
+ cand = found.compatible[box.curselection()[0]]
+ holder["choice"] = {"path": cand.path, "user": cand.user, "password": cand.password}
+ win.destroy()
+ done.set()
+ else:
+ why = "Usuário/senha do Firebird necessários." if found.state == "AUTH_REQUIRED" else "Nenhum banco SGV foi encontrado automaticamente."
+ ttk.Label(frame, text=f"{why} Informe o arquivo do banco e as credenciais:").pack(anchor="w")
+ row = ttk.Frame(frame)
+ row.pack(fill="x", pady=6)
+ ttk.Entry(row, textvariable=path_var, width=60).pack(side="left")
+ ttk.Button(row, text="Procurar…", command=lambda: path_var.set(filedialog.askopenfilename(filetypes=[("Firebird", "*.fdb *.FDB"), ("Todos", "*.*")]) or path_var.get())).pack(side="left", padx=6)
+ for label, var, secret in (("Usuário", user_var, False), ("Senha", pass_var, True)):
+ r = ttk.Frame(frame)
+ r.pack(fill="x", pady=2)
+ ttk.Label(r, text=label, width=9).pack(side="left")
+ ttk.Entry(r, textvariable=var, width=30, show="*" if secret else "").pack(side="left")
+ if found.auth_failed:
+ path_var.set(found.auth_failed[0].path)
+
+ def confirm():
+ cand = fbd.probe(path_var.get().strip(), user_var.get().strip(), pass_var.get())
+ if cand.status != "OK":
+ msg.config(text=f"Não foi possível usar este banco ({cand.status}: {cand.detail}). Verifique o arquivo e as credenciais.")
+ return
+ holder["choice"] = {"path": path_var.get().strip(), "user": user_var.get().strip(), "password": pass_var.get()}
+ win.destroy()
+ done.set()
+
+ def cancel():
+ win.destroy()
+ done.set()
+
+ msg.pack(anchor="w")
+ btns = ttk.Frame(frame)
+ btns.pack(anchor="e", pady=10)
+ ttk.Button(btns, text="Continuar", command=confirm).pack(side="left", padx=6)
+ ttk.Button(btns, text="Cancelar", command=cancel).pack(side="left")
+ win.protocol("WM_DELETE_WINDOW", cancel)
+
def _run(self, code: str):
try:
- result = install_flow(code)
+ if self.existing:
+ result = upgrade_flow(progress=self._say)
+ else:
+ result = install_flow(code, chooser=self.choose_firebird, progress=self._say)
except RuntimeError as error:
- self._set_busy(False)
- messagebox.showerror("GTI Fiscal Agent", str(error))
+ self._busy(False)
+ self._say(str(error), "#b00")
+ self.root.after(0, lambda: messagebox.showerror("GTI Fiscal Agent", str(error)))
return
- self._set_busy(False)
- install = result["install_dir"]
- messagebox.showinfo(
- "GTI Fiscal Agent",
- "Serviço GTI Fiscal Agent instalado com sucesso.\n\n"
- "Firebird não está configurado — o serviço não foi iniciado.\n"
- f"Local: {install}\n"
- "Configure o arquivo de banco em agent.env para ativar.",
- )
- self.root.destroy()
+ self._busy(False)
+ self.root.after(0, lambda: (messagebox.showinfo("GTI Fiscal Agent", final_message(result)), self.root.destroy()))
def cli() -> int:
parser = argparse.ArgumentParser(description="GTI Fiscal Agent Setup (headless)")
parser.add_argument("--cli", action="store_true", help="run headless (no GUI)")
- parser.add_argument("--code", required=True, help="XXXX-XXXX-XXXX-XXXX install code")
+ parser.add_argument("--code", default=None, help="XXXX-XXXX-XXXX-XXXX install code (omit to upgrade an existing install)")
parser.add_argument("--hub", default=None, help="Hub base URL (dev override; default prod)")
parser.add_argument("--install-dir", default=None, help="install directory (dev override)")
parser.add_argument("--no-service", action="store_true", help="skip service registration + ARP (E2E scratch runs)")
+ parser.add_argument("--fdb", default=None, help="Firebird database path (skips discovery)")
+ parser.add_argument("--fb-user", default="SYSDBA")
+ parser.add_argument("--fb-password", default=None)
+ parser.add_argument("--no-firebird", action="store_true", help="scratch E2E without any Firebird database")
args = parser.parse_args()
if args.hub:
@@ -176,8 +333,15 @@ def cli() -> int:
if args.install_dir:
os.environ["GTI_INSTALL_DIR"] = args.install_dir
- result = install_flow(args.code, hub=args.hub, install_dir_path=args.install_dir, register_service=not args.no_service)
- print(f"INSTALL_OK agent_id={result.get('agent_id')} install_dir={result.get('install_dir')} hub={result.get('hub_url')}")
+ if not args.code:
+ result = upgrade_flow(install_dir_path=args.install_dir, progress=print)
+ print(f"UPGRADE_OK agent_id={result['agent_id']} service_running={result['service_running']} heartbeat={result['heartbeat']}")
+ return 0
+ explicit = {"path": args.fdb, "user": args.fb_user, "password": args.fb_password or ""} if args.fdb else None
+ result = install_flow(args.code, hub=args.hub, install_dir_path=args.install_dir, register_service=not args.no_service,
+ firebird=explicit, require_firebird=not args.no_firebird, progress=print)
+ print(f"INSTALL_OK agent_id={result.get('agent_id')} install_dir={result.get('install_dir')} hub={result.get('hub_url')} "
+ f"service_running={result.get('service_running')} heartbeat={result.get('heartbeat')}")
return 0
@@ -195,4 +359,4 @@ def main():
if __name__ == "__main__":
- sys.exit(main())
\ No newline at end of file
+ sys.exit(main())
diff --git a/release/RELEASE_MANIFEST.json b/release/RELEASE_MANIFEST.json
index 9d7a0b8..c3eaa0b 100644
--- a/release/RELEASE_MANIFEST.json
+++ b/release/RELEASE_MANIFEST.json
@@ -1,14 +1,14 @@
{
"artifact": "RELEASE_MANIFEST",
- "GTI_FISCAL_VERSION": "1.1.2",
- "generated_at": "2026-09-24T11:12:47+00:00",
+ "GTI_FISCAL_VERSION": "1.1.3",
+ "generated_at": "2026-09-24T11:46:32+00:00",
"hub": {
- "version": "1.1.2",
+ "version": "1.1.3",
"api_compat": "1.0",
"engine_version": "0.3.0",
"decision_version": "mission-04c-ipi-ex-split"
},
- "agent_version": "1.1.2",
+ "agent_version": "1.1.3",
"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": "0025_tenant_external_cnpj",
- "applied_revision": "0025_tenant_external_cnpj",
+ "alembic_head_in_repo": "0026_central_processing_control",
+ "applied_revision": "0026_central_processing_control",
"applied_revision_source": "production Hub 10.0.10.117 (gti-fiscal-hub-fiscal-hub-1), verified 2026-09-24",
- "count": 25
+ "count": 26
},
"datasets_active": [
{
diff --git a/release/VERSION b/release/VERSION
index 45a1b3f..781dcb0 100644
--- a/release/VERSION
+++ b/release/VERSION
@@ -1 +1 @@
-1.1.2
+1.1.3
diff --git a/scripts/build_windows_installer.ps1 b/scripts/build_windows_installer.ps1
index eda078b..c75df1c 100644
--- a/scripts/build_windows_installer.ps1
+++ b/scripts/build_windows_installer.ps1
@@ -74,7 +74,7 @@ if (-not $SkipSetup) {
$ExecutorData = (Join-Path $Root "scripts\firebird_write_executor.py") + ";" + "scripts"
Invoke-Step "Build Setup.exe (onefile, uac-admin)" -Body {
& $py -m PyInstaller @Common @Paths @Hidden @Excludes "--onefile" "--uac-admin" "--name" "GTIFiscalAgent-Setup" `
- "--add-data" $ServiceData "--add-data" $UninstallData "--add-data" $ExecutorData `
+ "--add-data" $ServiceData "--add-data" $UninstallData "--add-data" $ExecutorData "--add-data" $VersionData `
(Join-Path $Root "installer\setup_main.py")
}
}
diff --git a/scripts/e2e_zero_touch_local.py b/scripts/e2e_zero_touch_local.py
new file mode 100644
index 0000000..b3de3f1
--- /dev/null
+++ b/scripts/e2e_zero_touch_local.py
@@ -0,0 +1,83 @@
+"""Real zero-touch runtime E2E against the local dev Hub + a real Firebird test database.
+
+No manual step between heartbeat and result: a brand-new installation identity sends its
+first heartbeat, the Hub queues exactly one INITIAL_SYNC, the Agent claims and executes it
+(READ-ONLY), coverage/blockers land on the Hub; then the central scheduler queues a
+SYNC_INCREMENTAL by itself (interval shortened to 1 s for the test) which must reprocess
+only the delta (nothing changed -> 0), and finally a central pause stops job execution while
+heartbeats continue.
+
+Usage (dev Hub on :8000, .env with FB credentials): python scripts/e2e_zero_touch_local.py
+"""
+import json
+import os
+import subprocess
+import sys
+import tempfile
+import time
+import uuid
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path[:0] = [str(ROOT), str(ROOT / "agent" / "src"), str(ROOT / "agent" / "scripts")]
+
+from dotenv import dotenv_values # noqa: E402
+
+env_file = dotenv_values(ROOT / ".env")
+HUB = "http://localhost:8000"
+
+
+def psql(sql):
+ out = subprocess.run(["docker", "exec", "projeto-postgres-1", "psql", "-U", "gti", "-d", "gti_fiscal", "-At", "-F", "|", "-c", sql], capture_output=True, text=True)
+ return out.stdout.strip()
+
+
+def main():
+ import run_agent_service as service
+
+ tmp = Path(tempfile.mkdtemp(prefix="gti_zt_"))
+ # A brand-new Agent (own tenant + token), like a freshly enrolled installation — never the shared dev agent.
+ reg = subprocess.run(["docker", "exec", "projeto-fiscal-hub-1", "python", "central/scripts/register_agent.py", "--company", f"e2e-zero-touch-{int(time.time())}", "--name", "zt"], capture_output=True, text=True).stdout
+ token = next(line.split("=", 1)[1] for line in reg.splitlines() if line.strip().startswith("AGENT_TOKEN="))
+ installation = str(uuid.uuid4())
+ os.environ.update({
+ "GTI_HUB_URL": HUB, "AGENT_TOKEN": token, "AGENT_ID": "dev", "INSTALLATION_ID": installation,
+ "FIREBIRD_DATABASE_PATH": env_file["FIREBIRD_DATABASE_PATH"], "FB_USER": env_file.get("FB_USER", "SYSDBA"), "FB_PASSWORD": env_file["FB_PASSWORD"],
+ "FIREBIRD_CLIENT_LIBRARY": env_file.get("FIREBIRD_CLIENT_LIBRARY", ""), "FIREBIRD_WRITES": "0", "FISCAL_MODE": "DRY_RUN",
+ "AGENT_STATE_DB": str(tmp / "state.sqlite"), "AGENT_LOG_PATH": str(tmp / "agent.log"), "POLL_INTERVAL_SECONDS": "0.2",
+ "AGENT_JOBS_ENABLED": "0", # legacy local toggle deliberately set to 0: must be ignored (control is central)
+ })
+ config = service.Config(os.environ)
+ service.setup_logging(config.log_path)
+ print("INSTALLATION_ID=", installation, "FIREBIRD_WRITES=", config.firebird_writes)
+
+ def run(n):
+ service.run(config, os.environ, max_iterations=n)
+
+ # 1) first heartbeat -> INITIAL_SYNC by the Hub -> claimed + executed in the same loop
+ run(4)
+ jobs = psql(f"select j.id,j.kind,j.status,j.result_summary->>'products_total',j.result_summary->>'products_changed' from jobs j join installations i on i.agent_id=j.agent_id where i.installation_id='{installation}' order by j.id")
+ print("AFTER_INITIAL:\n" + jobs)
+ snap = psql(f"select s.total_products,s.fiscally_complete,s.safe_blocked,s.failed_products,s.blockers from agent_snapshots s join installations i on i.id=s.installation_id where i.installation_id='{installation}'")
+ print("SNAPSHOT=", snap)
+ blockers = psql(f"select count(*) from blocked_products b join installations i on i.id=b.installation_id where i.installation_id='{installation}'")
+ print("BLOCKERS_TOTAL=", blockers)
+
+ # 2) central scheduler: shorten the interval, next heartbeat queues one SYNC_INCREMENTAL by itself
+ psql(f"update installations set sync_interval_seconds=1 where installation_id='{installation}'")
+ time.sleep(2)
+ run(4)
+ print("AFTER_INCREMENTAL:\n" + psql(f"select j.id,j.kind,j.status,j.payload->>'reason',j.result_summary->>'products_total',j.result_summary->>'products_changed' from jobs j join installations i on i.agent_id=j.agent_id where i.installation_id='{installation}' order by j.id"))
+
+ # 3) central pause: heartbeat continues, nothing is claimed
+ psql(f"update installations set processing_enabled=false where installation_id='{installation}'")
+ before = psql("select count(*) from jobs where status='RUNNING'")
+ time.sleep(2)
+ run(3)
+ print("PAUSED_RUNNING_JOBS=", before, "LAST_STATUS=", psql(f"select last_status from installations where installation_id='{installation}'"))
+ print("WRITES:", psql(f"select coalesce(sum((result_summary->>'writes_attempted')::int),0), coalesce(sum((result_summary->>'writes_succeeded')::int),0) from jobs j join installations i on i.agent_id=j.agent_id where i.installation_id='{installation}'"))
+ print("LOCAL_ipi_write_log_rows=", __import__("sqlite3").connect(tmp / "state.sqlite").execute("select count(*), sum(write_status='DRY_RUN_ONLY') from ipi_write_log").fetchone())
+
+
+if __name__ == "__main__":
+ main()
diff --git a/tests/test_firebird_discovery.py b/tests/test_firebird_discovery.py
new file mode 100644
index 0000000..9187b06
--- /dev/null
+++ b/tests/test_firebird_discovery.py
@@ -0,0 +1,71 @@
+"""Setup.exe's Firebird discovery: read-only SGV detection, auto-select only when unambiguous."""
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).parents[1] / "installer"))
+import firebird_discovery as fbd # noqa: E402
+
+
+class FakeConn:
+ def __init__(self, tables): self.tables = tables; self.closed = False; self.sql = []
+ def cursor(self):
+ outer = self
+ class C:
+ def execute(self, sql): outer.sql.append(sql)
+ def fetchall(self): return [(t,) for t in outer.tables]
+ return C()
+ def close(self): self.closed = True
+
+
+def make_connect(spec):
+ """spec: path -> tables | 'AUTH' | Exception"""
+ def connect(path, user, password):
+ value = spec[path]
+ if value == "AUTH" and (user, password) != ("SYSDBA", "masterkey"):
+ raise Exception("Your user name and password are not defined")
+ if value == "BROKEN": raise OSError("cannot open")
+ return FakeConn(["C000025", "C000004"] if value == "AUTH" else value)
+ return connect
+
+
+def run(spec, credentials=fbd.DEFAULT_CREDENTIALS):
+ return fbd.discover(roots=["X"], finder=lambda roots: list(spec), connect=make_connect(spec), credentials=credentials)
+
+
+def test_exactly_one_sgv_database_is_selected_automatically():
+ found = run({r"D:\SGV\loja.fdb": ["C000025", "C000004", "OTHER"], r"D:\other\x.fdb": ["FOO"]})
+ assert found.state == "SELECTED" and found.selected.path == r"D:\SGV\loja.fdb" and found.selected.user == "SYSDBA"
+
+
+def test_several_compatible_databases_are_never_chosen_silently():
+ found = run({r"D:\a.fdb": ["C000025", "C000004"], r"D:\b.fdb": ["C000025", "C000004"]})
+ assert found.state == "AMBIGUOUS" and len(found.compatible) == 2 and found.selected is None
+
+
+def test_database_needs_both_c000025_and_c000004():
+ assert run({r"D:\a.fdb": ["C000025"]}).state == "NONE"
+
+
+def test_unknown_credentials_ask_the_operator_inside_the_installer():
+ found = run({r"D:\a.fdb": "AUTH"}, credentials=(("SYSDBA", "wrong"),))
+ assert found.state == "AUTH_REQUIRED" and found.auth_failed[0].path == r"D:\a.fdb"
+
+
+def test_probe_is_read_only_and_closes_the_connection():
+ conn = FakeConn(["C000025", "C000004"])
+ cand = fbd.probe("x.fdb", "SYSDBA", "masterkey", connect=lambda *a: conn)
+ assert cand.status == "OK" and conn.closed and all(sql.lstrip().upper().startswith("SELECT") for sql in conn.sql)
+
+
+def test_broken_file_is_reported_not_raised():
+ assert fbd.probe("x.fdb", "u", "p", connect=lambda *a: (_ for _ in ()).throw(OSError("boom"))).status == "ERROR"
+
+
+def test_finder_skips_system_dirs_security_db_and_tiny_files(tmp_path):
+ big = b"0" * (fbd.MIN_BYTES + 1)
+ (tmp_path / "SGV").mkdir(); (tmp_path / "SGV" / "loja.FDB").write_bytes(big)
+ (tmp_path / "SGV" / "security2.fdb").write_bytes(big)
+ (tmp_path / "SGV" / "tiny.fdb").write_bytes(b"0")
+ (tmp_path / "Windows").mkdir(); (tmp_path / "Windows" / "sys.fdb").write_bytes(big)
+ found = fbd.find_fdb_files([str(tmp_path)])
+ assert [Path(p).name for p in found] == ["loja.FDB"]
diff --git a/tests/test_release_candidate.py b/tests/test_release_candidate.py
index b62e131..c64df3a 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.1.2"
+ assert v == GTI_FISCAL_VERSION == "1.1.3"
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.1.2"' in (ROOT / "agent" / "scripts" / "run_agent_service.py").read_text(encoding="utf-8")
+ assert '"AGENT_VERSION", "1.1.3"' 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"))
diff --git a/tests/test_zero_touch_installer.py b/tests/test_zero_touch_installer.py
new file mode 100644
index 0000000..9e74a82
--- /dev/null
+++ b/tests/test_zero_touch_installer.py
@@ -0,0 +1,118 @@
+"""Zero-touch Setup: discovery -> enroll -> secrets in DPAPI stores -> service registered AND started
+-> first heartbeat awaited; upgrade preserves identity and drops the legacy local toggle.
+The Windows/SCM/Hub edges are faked; the orchestration is real."""
+import importlib
+import sys
+from pathlib import Path
+
+import pytest
+
+ROOT = Path(__file__).parents[1]
+for extra in (ROOT, ROOT / "installer", ROOT / "agent" / "src"):
+ sys.path.insert(0, str(extra))
+
+setup_main = importlib.import_module("setup_main")
+common = setup_main.common
+fbd = setup_main.fbd
+
+ENROLL = {"agent_id": 7, "agent_token": "tok-never-in-env", "hub_url": "https://gtifiscal.gtigrupo.com.br", "agent_version": "1.1.3",
+ "installation_id": "00000000-0000-4000-8000-0000000000bb"}
+
+
+@pytest.fixture
+def world(tmp_path, monkeypatch):
+ calls = []
+ monkeypatch.setenv("GTI_INSTALL_DIR", str(tmp_path / "inst"))
+ monkeypatch.setattr(setup_main, "MEIPASS", tmp_path / "payload")
+ payload = tmp_path / "payload"
+ (payload / "GTIFiscalAgentService").mkdir(parents=True); (payload / "GTIFiscalAgentService" / "GTIFiscalAgentService.exe").write_bytes(b"x")
+ (payload / "Uninstall.exe").write_bytes(b"x"); (payload / "scripts").mkdir(); (payload / "scripts" / "firebird_write_executor.py").write_text("# x")
+ (payload / "VERSION").write_text("1.1.3\n")
+ monkeypatch.setattr(common, "enroll", lambda code, hub=None: calls.append(("enroll", code)) or dict(ENROLL))
+ monkeypatch.setattr(common, "write_credential", lambda d, t: calls.append(("hub_cred", t)))
+ monkeypatch.setattr(common, "write_firebird_credential", lambda d, u, p: calls.append(("fb_cred", u)))
+ monkeypatch.setattr(common, "install_service", lambda d: calls.append(("install_service",)))
+ monkeypatch.setattr(common, "write_arp_uninstall", lambda d: None)
+ monkeypatch.setattr(common, "service_running", lambda: False)
+ monkeypatch.setattr(common, "service_installed", lambda: True)
+ monkeypatch.setattr(common, "start_service", lambda: calls.append(("start",)))
+ monkeypatch.setattr(common, "stop_service", lambda: calls.append(("stop",)))
+ monkeypatch.setattr(common, "wait_service_state", lambda wanted, timeout=45: True)
+ monkeypatch.setattr(common, "wait_for_heartbeat", lambda token, hub, timeout=120: calls.append(("heartbeat_wait",)) or True)
+ monkeypatch.setattr(common, "audit_event", lambda *a, **k: None)
+ return calls, tmp_path / "inst"
+
+
+def one_db(monkeypatch):
+ cand = fbd.Candidate(r"D:\SGV\loja.fdb", "OK", "SYSDBA", "masterkey")
+ monkeypatch.setattr(fbd, "discover", lambda **k: fbd.Discovery("SELECTED", [cand]))
+
+
+def test_zero_touch_install_ends_with_service_started_and_heartbeat_confirmed(world, monkeypatch):
+ calls, inst = world
+ one_db(monkeypatch)
+ result = setup_main.install_flow("AAAA-BBBB-CCCC-DDDD")
+ names = [c[0] for c in calls]
+ assert names == ["enroll", "hub_cred", "fb_cred", "install_service", "start", "heartbeat_wait"]
+ assert result["service_running"] is True and result["heartbeat"] is True and result["firebird_path"] == r"D:\SGV\loja.fdb"
+ env = (inst / "agent.env").read_text(encoding="utf-8")
+ assert r"FIREBIRD_DATABASE_PATH=D:\SGV\loja.fdb" in env and "FIREBIRD_WRITES=0" in env and "FISCAL_MODE=DRY_RUN" in env
+ assert "tok-never-in-env" not in env and "masterkey" not in env and "AGENT_JOBS_ENABLED" not in env
+ assert "Nenhuma outra ação" in setup_main.final_message(result) # the operator is told nothing else is needed
+
+
+def test_ambiguous_databases_ask_the_operator_and_nothing_is_enrolled_if_cancelled(world, monkeypatch):
+ calls, _ = world
+ a, b = fbd.Candidate(r"D:\a.fdb", "OK", "SYSDBA", "masterkey"), fbd.Candidate(r"D:\b.fdb", "OK", "SYSDBA", "masterkey")
+ monkeypatch.setattr(fbd, "discover", lambda **k: fbd.Discovery("AMBIGUOUS", [a, b]))
+ with pytest.raises(RuntimeError, match="seleção do banco Firebird"):
+ setup_main.install_flow("AAAA-BBBB-CCCC-DDDD") # headless, no chooser: refuses to guess
+ assert calls == []
+ with pytest.raises(RuntimeError, match="cancelada"):
+ setup_main.install_flow("AAAA-BBBB-CCCC-DDDD", chooser=lambda found: None)
+ assert calls == [] # the install code was NOT consumed by an install that could not complete
+
+
+def test_operator_choice_is_verified_before_use(world, monkeypatch):
+ calls, inst = world
+ a, b = fbd.Candidate(r"D:\a.fdb", "OK", "SYSDBA", "masterkey"), fbd.Candidate(r"D:\b.fdb", "OK", "SYSDBA", "masterkey")
+ monkeypatch.setattr(fbd, "discover", lambda **k: fbd.Discovery("AMBIGUOUS", [a, b]))
+ monkeypatch.setattr(fbd, "probe", lambda path, u, p, **k: fbd.Candidate(path, "OK", u, p))
+ result = setup_main.install_flow("AAAA-BBBB-CCCC-DDDD", chooser=lambda found: {"path": r"D:\b.fdb", "user": "SYSDBA", "password": "masterkey"})
+ assert result["firebird_path"] == r"D:\b.fdb"
+ monkeypatch.setattr(fbd, "probe", lambda path, u, p, **k: fbd.Candidate(path, "AUTH_FAILED"))
+ with pytest.raises(RuntimeError, match="não é utilizável"):
+ setup_main.install_flow("AAAA-BBBB-CCCC-DDDD", chooser=lambda found: {"path": r"D:\b.fdb", "user": "x", "password": "y"})
+
+
+def test_upgrade_preserves_identity_drops_local_toggle_and_restarts(world):
+ calls, inst = world
+ inst.mkdir(parents=True)
+ (inst / "credentials.bin").write_bytes(b"blob")
+ (inst / "agent.env").write_text("GTI_HUB_URL=https://gtifiscal.gtigrupo.com.br\nAGENT_ID=2\nINSTALLATION_ID=bb50884a-5e78-4f26-8cac-804932dae8de\n"
+ "AGENT_JOBS_ENABLED=0\nAGENT_VERSION=1.1.1\nFIREBIRD_DATABASE_PATH=D:/SGV/x.fdb\nFISCAL_MODE=DRY_RUN\nFIREBIRD_WRITES=0\n", encoding="utf-8")
+ result = setup_main.upgrade_flow()
+ env = (inst / "agent.env").read_text(encoding="utf-8")
+ assert "AGENT_JOBS_ENABLED" not in env and "AGENT_VERSION=1.1.3" in env
+ assert "AGENT_ID=2" in env and "INSTALLATION_ID=bb50884a-5e78-4f26-8cac-804932dae8de" in env and "FIREBIRD_DATABASE_PATH=D:/SGV/x.fdb" in env
+ assert [c[0] for c in calls] == ["stop", "install_service", "start"]
+ assert "enroll" not in [c[0] for c in calls] # NEVER re-enrolls: no new Agent, no duplicate installation
+ assert result["mode"] == "UPGRADE" and result["installation_id"] == "bb50884a-5e78-4f26-8cac-804932dae8de" and result["service_running"]
+
+
+def test_second_run_on_enrolled_host_is_an_upgrade_not_a_new_install(world):
+ calls, inst = world
+ inst.mkdir(parents=True); (inst / "credentials.bin").write_bytes(b"b"); (inst / "agent.env").write_text("AGENT_ID=2\nINSTALLATION_ID=x\n")
+ assert common.detect_existing(inst)["agent_id"] == "2"
+ assert common.detect_existing(inst / "nope") is None
+
+
+def test_service_is_registered_delayed_auto_with_restart_recovery():
+ text = (ROOT / "installer" / "install_common.py").read_text(encoding="utf-8")
+ assert '"start=", "delayed-auto"' in text and 'SERVICE_FAILURE = "restart/60000/restart/60000/restart/60000"' in text
+
+
+def test_no_write_flags_anywhere_in_zero_touch_installer():
+ for name in ("setup_main.py", "firebird_discovery.py"):
+ text = (ROOT / "installer" / name).read_text(encoding="utf-8")
+ assert "FIREBIRD_WRITES=1" not in text and "AUTO_APPLY" not in text and "UPDATE " not in text.upper().replace("UPDATE_", "")