313 lines
16 KiB
Python
313 lines
16 KiB
Python
"""Long-running Agent service: heartbeat + poll Hub jobs + execute pipeline.
|
|
|
|
startup -> load config -> validate Hub -> validate Firebird -> heartbeat
|
|
-> poll job -> claim -> execute -> report result -> back to polling.
|
|
|
|
No job available: sleep POLL_INTERVAL_SECONDS.
|
|
Hub unavailable: retry with simple capped backoff; the Agent stays alive.
|
|
Job fails internally (bad Firebird read, classify error, write refused):
|
|
the job is reported FAILED with a sanitized error and the Agent keeps polling.
|
|
SIGINT/SIGTERM: finish the current job (if any) and stop the loop cleanly.
|
|
|
|
Written to be runnable as-is today, and easy to later wrap as a Windows
|
|
Service (a service wrapper would just call main()); no installer here.
|
|
"""
|
|
import argparse
|
|
import importlib.util
|
|
import json
|
|
import logging
|
|
import logging.handlers
|
|
import os
|
|
import signal
|
|
import sys
|
|
import time
|
|
from pathlib import Path
|
|
|
|
AGENT_ROOT = Path(__file__).resolve().parents[1]
|
|
PROJECT_ROOT = AGENT_ROOT.parent
|
|
sys.path.insert(0, str(AGENT_ROOT / "src"))
|
|
sys.path.insert(0, str(PROJECT_ROOT)) # ships central/app engine modules read-only (central.app.*)
|
|
|
|
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.state import AgentState # noqa: E402
|
|
|
|
_MAX_BACKOFF_SECONDS = 60
|
|
|
|
logger = logging.getLogger("gti_fiscal_agent")
|
|
|
|
|
|
def setup_logging(log_path):
|
|
"""Rotating local log file. Never logs FB_PASSWORD or AGENT_TOKEN — every
|
|
log call in this module only references non-secret identifiers/counts."""
|
|
logger.setLevel(logging.INFO)
|
|
if logger.handlers:
|
|
return logger
|
|
Path(log_path).parent.mkdir(parents=True, exist_ok=True)
|
|
handler = logging.handlers.RotatingFileHandler(log_path, maxBytes=5_000_000, backupCount=5, encoding="utf-8")
|
|
handler.setFormatter(logging.Formatter("%(asctime)s %(levelname)s %(message)s"))
|
|
logger.addHandler(handler)
|
|
logger.addHandler(logging.StreamHandler())
|
|
return logger
|
|
|
|
|
|
def load_dotenv(path: Path) -> None:
|
|
if not path.exists():
|
|
return
|
|
for line in path.read_text(encoding="utf-8").splitlines():
|
|
line = line.strip()
|
|
if not line or line.startswith("#") or "=" not in line:
|
|
continue
|
|
key, _, value = line.partition("=")
|
|
os.environ.setdefault(key.strip(), value.strip())
|
|
|
|
|
|
def load_executor():
|
|
spec = importlib.util.spec_from_file_location("firebird_write_executor", PROJECT_ROOT / "scripts" / "firebird_write_executor.py")
|
|
module = importlib.util.module_from_spec(spec)
|
|
sys.modules[spec.name] = module
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
class Config:
|
|
def __init__(self, env):
|
|
self.hub_url = env.get("GTI_HUB_URL") or env.get("HUB_URL") or "http://localhost:8000"
|
|
self.agent_id = env.get("AGENT_ID", "")
|
|
self.agent_token = env.get("AGENT_TOKEN") or env.get("AGENT_DEV_TOKEN", "")
|
|
self.company_id = env.get("COMPANY_ID", "")
|
|
self.company_uf = env.get("COMPANY_UF", "")
|
|
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.
|
|
requested = env.get("FIREBIRD_WRITES", "0") == "1"
|
|
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"
|
|
self.poll_interval_seconds = float(env.get("POLL_INTERVAL_SECONDS", "5"))
|
|
self.agent_version = env.get("AGENT_VERSION", "1.0.0-rc1")
|
|
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
|
|
# a real client database. Left unset, both default to the dev-only values,
|
|
# so a real client path/schema is never silently authorized (see
|
|
# scripts/firebird_write_executor.py's assert_authorized_write_path).
|
|
self.authorized_write_database_path = env.get("AUTHORIZED_WRITE_DATABASE_PATH")
|
|
self.expected_schema_fingerprint = env.get("EXPECTED_SCHEMA_FINGERPRINT", REFERENCE_SCHEMA_V1)
|
|
self.log_path = env.get("AGENT_LOG_PATH", str(AGENT_ROOT / "logs" / "agent.log"))
|
|
|
|
|
|
class GracefulStop:
|
|
def __init__(self):
|
|
self.stop = False
|
|
signal.signal(signal.SIGINT, self._handle)
|
|
signal.signal(signal.SIGTERM, self._handle)
|
|
|
|
def _handle(self, signum, frame):
|
|
self.stop = True
|
|
|
|
|
|
def build_job_coverage(env, client, products, decisions):
|
|
"""Company-context-aware coverage over the job's products (read-only). Fails closed when the company context is unresolved."""
|
|
from gti_fiscal_agent.company_context import mask_cnpj, preflight_company_context
|
|
from gti_fiscal_agent.coverage import build_refs, compute_coverage
|
|
from gti_fiscal_agent.firebird_reader import read_evidence
|
|
ctx = preflight_company_context(env)
|
|
if ctx["status"] != "PASS":
|
|
raise RuntimeError("COMPANY_CONTEXT=FAIL: " + "; ".join(ctx.get("reasons") or ["unresolved"]))
|
|
evidence = read_evidence(env, ctx["company_code"])
|
|
refs = build_refs(client.get_json, products, ctx["company_uf"])
|
|
report = compute_coverage(products, decisions, {"uf": ctx["company_uf"], "crt": ctx["company_crt"]}, evidence.get("company_default_csosn"), refs, evidence)
|
|
report["company_context"] = {"COMPANY_CONTEXT_STATUS": "PASS", "classification": ctx["classification"], "company_code": ctx["company_code"], "company_cnpj": ctx["company_cnpj"], "company_uf": ctx["company_uf"],
|
|
"company_crt": ctx["company_crt"], "source_of_truth": ctx["source_of_truth"], "confidence": ctx["confidence"], "warnings": ctx["warnings"]}
|
|
return report
|
|
|
|
|
|
def open_write_connection(env, database_path, firebird_writes=True):
|
|
if not firebird_writes: # technical guarantee: no write-capable connection exists in DRY_RUN / FIREBIRD_WRITES=0
|
|
raise RuntimeError("write connection refused: writes are disabled (FISCAL_MODE=DRY_RUN or FIREBIRD_WRITES=0)")
|
|
import fdb
|
|
return fdb.connect(
|
|
database=database_path,
|
|
user=env["FB_USER"],
|
|
password=env["FB_PASSWORD"],
|
|
fb_library_name=env.get("FIREBIRD_CLIENT_LIBRARY"),
|
|
)
|
|
|
|
|
|
def execute_job(job, config, executor, state, env):
|
|
"""Run the pipeline for one claimed job. Never raises for job-domain
|
|
failures (bad Firebird read, classify error) — returns (result_summary, error)."""
|
|
kind = job["kind"]
|
|
payload = job.get("payload") or {}
|
|
try:
|
|
if config.firebird_writes:
|
|
# Layered, per-installation authorization: the path must match this
|
|
# installation's own explicitly configured authorized_write_database_path
|
|
# (defaults to the dev-only constant, so a real client path is refused
|
|
# unless the installer deliberately set it), AND the live schema
|
|
# fingerprint must match what this installation was configured to expect.
|
|
executor.assert_authorized_write_path(config.database_path, authorized_path=config.authorized_write_database_path)
|
|
executor.assert_schema_fingerprint_authorized(config.schema_fingerprint, config.expected_schema_fingerprint)
|
|
|
|
codigos = payload.get("codigos") if kind == "REPROCESS_PRODUCTS" else None
|
|
products = read_products(env, codigos=codigos)
|
|
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])
|
|
|
|
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)
|
|
finally:
|
|
if write_connection is not None:
|
|
write_connection.close()
|
|
|
|
coverage = build_job_coverage(env, client, products, decisions) if kind == "REPROCESS_ALL" else None
|
|
result_summary = {
|
|
"coverage": coverage,
|
|
"fiscal_mode": getattr(config, "fiscal_mode", None),
|
|
"release_version": env.get("AGENT_VERSION") or "1.0.0-rc1",
|
|
"products_processed": summary["processed"],
|
|
"writes_attempted": summary["writes_attempted"],
|
|
"writes_succeeded": summary["writes_succeeded"],
|
|
"writes_failed": summary["writes_failed"],
|
|
"no_ops": summary["no_ops"],
|
|
"firebird_writes_enabled": config.firebird_writes,
|
|
"agent_id": job["agent_id"],
|
|
"schema_fingerprint": config.schema_fingerprint,
|
|
}
|
|
logger.info(
|
|
"job %s finished: processed=%d writes_attempted=%d writes_succeeded=%d writes_failed=%d no_ops=%d",
|
|
job.get("job_id"), summary["processed"], summary["writes_attempted"], summary["writes_succeeded"], summary["writes_failed"], summary["no_ops"],
|
|
)
|
|
if summary["writes_failed"] > 0:
|
|
return result_summary, f"{summary['writes_failed']} write(s) failed guard checks"
|
|
return result_summary, None
|
|
except executor.WriteRefused as error:
|
|
logger.warning("job %s refused: %s", job.get("job_id"), error)
|
|
return None, f"write refused: {error}"
|
|
except Exception as error: # noqa: BLE001 — job-domain isolation boundary
|
|
logger.warning("job %s failed internally: %s: %s", job.get("job_id"), type(error).__name__, error)
|
|
return None, f"{type(error).__name__}: {error}"
|
|
|
|
|
|
def run(config, env, *, max_iterations=None, stopper=None):
|
|
"""`stopper` defaults to a signal-based GracefulStop (SIGINT/SIGTERM), which
|
|
only works from the main thread. A host that manages its own lifecycle
|
|
(e.g. the Windows Service wrapper, which stops the loop via the Service
|
|
Control Manager instead of OS signals) should pass its own stop object —
|
|
anything with a `.stop` boolean attribute — and flip it from any thread."""
|
|
executor = load_executor()
|
|
state = AgentState(config.state_db)
|
|
client = HubClient(config.hub_url, config.agent_token)
|
|
installation_id = state.installation_id() if hasattr(state, "installation_id") else (config.agent_id or "legacy-installation")
|
|
stopper = stopper if stopper is not None else GracefulStop()
|
|
|
|
backoff = 1
|
|
was_down = False
|
|
iterations = 0
|
|
last_job_id = None
|
|
while not stopper.stop:
|
|
if max_iterations is not None and iterations >= max_iterations:
|
|
break
|
|
iterations += 1
|
|
|
|
try:
|
|
heartbeat_payload = {
|
|
"installation_id": installation_id, "status": "IDLE", "schema_fingerprint": config.schema_fingerprint,
|
|
"agent_version": config.agent_version, "last_job_id": last_job_id, "display_name": env.get("AGENT_DISPLAY_NAME", ""),
|
|
"customer_code": env.get("CUSTOMER_CODE"), "hostname": env.get("AGENT_HOSTNAME") or os.environ.get("COMPUTERNAME"),
|
|
"windows_version": env.get("WINDOWS_VERSION"), "company_code": config.company_id, "company_uf": config.company_uf,
|
|
"database_path": config.database_path, "capabilities": {"heartbeat": True, "jobs": True, "local_logs": True},
|
|
"fiscal_engine_version": env.get("FISCAL_ENGINE_VERSION", config.agent_version), "dataset_versions": {},
|
|
}
|
|
if hasattr(client, "heartbeat_metadata"):
|
|
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)
|
|
if was_down:
|
|
logger.info("reconnected to Hub after %ss backoff", backoff)
|
|
was_down = False
|
|
backoff = 1
|
|
except Exception as error:
|
|
logger.warning("heartbeat failed, Hub unreachable: %s (retrying in %ss)", error, backoff)
|
|
was_down = True
|
|
time.sleep(backoff)
|
|
backoff = min(backoff * 2, _MAX_BACKOFF_SECONDS)
|
|
continue
|
|
|
|
try:
|
|
job = client.next_job()
|
|
except Exception as error:
|
|
logger.warning("poll failed, Hub unreachable: %s (retrying in %ss)", error, backoff)
|
|
was_down = True
|
|
time.sleep(backoff)
|
|
backoff = min(backoff * 2, _MAX_BACKOFF_SECONDS)
|
|
continue
|
|
|
|
if job is None:
|
|
time.sleep(config.poll_interval_seconds)
|
|
continue
|
|
|
|
last_job_id = job.get("job_id")
|
|
logger.info("job %s received: kind=%s", job.get("job_id"), job["kind"])
|
|
try:
|
|
if hasattr(client, "heartbeat_metadata"):
|
|
client.heartbeat_metadata({"installation_id": installation_id, "status": "BUSY", "schema_fingerprint": config.schema_fingerprint, "agent_version": config.agent_version, "last_job_id": last_job_id})
|
|
else:
|
|
client.heartbeat("BUSY", schema_fingerprint=config.schema_fingerprint, agent_version=config.agent_version, last_job_id=last_job_id)
|
|
except Exception:
|
|
pass # best-effort; the job itself still proceeds
|
|
|
|
result_summary, error = execute_job(job, config, executor, state, env)
|
|
try:
|
|
if error is None:
|
|
client.complete_job(job.get("job_id"), result_summary)
|
|
logger.info("job %s reported COMPLETED", job.get("job_id"))
|
|
else:
|
|
client.fail_job(job.get("job_id"), error)
|
|
logger.warning("job %s reported FAILED: %s", job.get("job_id"), error)
|
|
except Exception as report_error:
|
|
logger.warning("could not report job %s result to Hub: %s", job.get("job_id"), report_error)
|
|
|
|
|
|
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.0.0-rc1")
|
|
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":
|
|
logger.error("startup aborted: preflight failed (fail closed): %s", report["blockers"])
|
|
raise SystemExit(2)
|
|
return report
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--max-iterations", type=int, default=None, help="stop after N poll cycles (used by tests); omit to run forever")
|
|
args = parser.parse_args()
|
|
|
|
load_dotenv(Path(os.environ.get("GTI_AGENT_ENV_FILE") or PROJECT_ROOT / ".env"))
|
|
config = Config(os.environ)
|
|
setup_logging(config.log_path)
|
|
|
|
logger.info("startup: agent_id=%s company_id=%s agent_version=%s firebird_writes=%s", config.agent_id, config.company_id, config.agent_version, config.firebird_writes)
|
|
hub_ok = HubClient(config.hub_url, config.agent_token).health()
|
|
firebird_ok = check_connection(os.environ)
|
|
logger.info("startup checks: hub_ok=%s firebird_ok=%s", hub_ok, firebird_ok)
|
|
enforce_startup_preflight(os.environ)
|
|
|
|
run(config, os.environ, max_iterations=args.max_iterations)
|
|
logger.info("shutdown: stopped cleanly")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|