From 301aac112ef80e2f359bfa781dc57abcd3b624b6 Mon Sep 17 00:00:00 2001 From: gti Date: Tue, 22 Sep 2026 15:36:12 -0300 Subject: [PATCH] feat: add autonomous fiscal control panel --- agent/scripts/run_agent_service.py | 19 ++- agent/src/gti_fiscal_agent/client.py | 9 ++ agent/src/gti_fiscal_agent/state.py | 10 ++ central/alembic/env.py | 9 +- .../0015_ai_budget_and_product_cache.py | 2 +- .../versions/0017_control_plane_foundation.py | 101 ++++++++++++++ central/app/api.py | 65 ++++++++- central/app/audit.py | 11 ++ central/app/config.py | 9 ++ central/app/control_api.py | 125 ++++++++++++++++++ central/app/control_plane.py | 94 +++++++++++++ central/app/events.py | 74 +++++++++++ central/app/human_auth.py | 75 +++++++++++ central/app/installation_identity.py | 104 +++++++++++++++ central/app/main.py | 6 + central/app/models.py | 103 +++++++++++++++ central/app/web.py | 122 +++++++++++++++++ central/pyproject.toml | 1 + central/scripts/create_admin.py | 21 +++ .../scripts/generate_control_panel_fixture.py | 52 ++++++++ central/scripts/maintain_events.py | 22 +++ central/static/control-panel.css | 1 + central/static/control-panel.js | 4 + central/templates/agent_detail.html | 9 ++ central/templates/agents.html | 1 + central/templates/base.html | 8 ++ central/templates/blockers.html | 1 + central/templates/dashboard.html | 1 + central/templates/jobs.html | 1 + central/templates/login.html | 1 + central/tests/test_control_panel_fixture.py | 14 ++ central/tests/test_control_plane.py | 73 ++++++++++ docs/AI_PROJECT_STATE.md | 9 ++ docs/CONTROL_PANEL_V1.md | 104 +++++++++++++++ 34 files changed, 1252 insertions(+), 9 deletions(-) create mode 100644 central/alembic/versions/0017_control_plane_foundation.py create mode 100644 central/app/audit.py create mode 100644 central/app/control_api.py create mode 100644 central/app/control_plane.py create mode 100644 central/app/events.py create mode 100644 central/app/human_auth.py create mode 100644 central/app/installation_identity.py create mode 100644 central/app/web.py create mode 100644 central/scripts/create_admin.py create mode 100644 central/scripts/generate_control_panel_fixture.py create mode 100644 central/scripts/maintain_events.py create mode 100644 central/static/control-panel.css create mode 100644 central/static/control-panel.js create mode 100644 central/templates/agent_detail.html create mode 100644 central/templates/agents.html create mode 100644 central/templates/base.html create mode 100644 central/templates/blockers.html create mode 100644 central/templates/dashboard.html create mode 100644 central/templates/jobs.html create mode 100644 central/templates/login.html create mode 100644 central/tests/test_control_panel_fixture.py create mode 100644 central/tests/test_control_plane.py create mode 100644 docs/CONTROL_PANEL_V1.md diff --git a/agent/scripts/run_agent_service.py b/agent/scripts/run_agent_service.py index 0420345..235df82 100644 --- a/agent/scripts/run_agent_service.py +++ b/agent/scripts/run_agent_service.py @@ -206,6 +206,7 @@ def run(config, env, *, max_iterations=None, stopper=None): 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 @@ -218,7 +219,18 @@ def run(config, env, *, max_iterations=None, stopper=None): iterations += 1 try: - client.heartbeat("IDLE", schema_fingerprint=config.schema_fingerprint, agent_version=config.agent_version, last_job_id=last_job_id) + 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 @@ -246,7 +258,10 @@ def run(config, env, *, max_iterations=None, stopper=None): last_job_id = job.get("job_id") logger.info("job %s received: kind=%s", job.get("job_id"), job["kind"]) try: - client.heartbeat("BUSY", schema_fingerprint=config.schema_fingerprint, agent_version=config.agent_version, last_job_id=last_job_id) + 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 diff --git a/agent/src/gti_fiscal_agent/client.py b/agent/src/gti_fiscal_agent/client.py index ac2026a..8bb5025 100644 --- a/agent/src/gti_fiscal_agent/client.py +++ b/agent/src/gti_fiscal_agent/client.py @@ -20,6 +20,15 @@ class HubClient: payload = {"status": status, "schema_fingerprint": schema_fingerprint, "agent_version": agent_version, "last_job_id": last_job_id} response = httpx.post(f"{self.url}/v1/agents/heartbeat", json=payload, headers=self._headers(), timeout=15) response.raise_for_status(); return response.json() + def register(self, payload): + response = httpx.post(f"{self.url}/v1/agents/register", json=payload, headers=self._headers(), timeout=15) + response.raise_for_status(); return response.json() + def heartbeat_metadata(self, payload): + response = httpx.post(f"{self.url}/v1/agents/heartbeat", json=payload, headers=self._headers(), timeout=15) + response.raise_for_status(); return response.json() + def event(self, payload): + response = httpx.post(f"{self.url}/v1/agents/events", json=payload, headers=self._headers(), timeout=15) + response.raise_for_status(); return response.json() def next_job(self): response = httpx.get(f"{self.url}/v1/jobs/next", headers=self._headers(), timeout=15) response.raise_for_status(); return response.json() diff --git a/agent/src/gti_fiscal_agent/state.py b/agent/src/gti_fiscal_agent/state.py index fc40178..6847125 100644 --- a/agent/src/gti_fiscal_agent/state.py +++ b/agent/src/gti_fiscal_agent/state.py @@ -1,11 +1,21 @@ """Local SQLite outbox; it has no Firebird dependency or write path.""" import sqlite3 from pathlib import Path +import uuid class AgentState: def __init__(self, path): 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)") + self.connection.commit() + def installation_id(self): + row = self.connection.execute("SELECT installation_id FROM installation_identity WHERE id=1").fetchone() + if row: + return row["installation_id"] + value = str(uuid.uuid4()) + self.connection.execute("INSERT INTO installation_identity(id, installation_id) VALUES(1, ?)", (value,)); self.connection.commit() + return value 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/central/alembic/env.py b/central/alembic/env.py index 3999aa9..40be691 100644 --- a/central/alembic/env.py +++ b/central/alembic/env.py @@ -4,6 +4,10 @@ from central.app.models import Base from central.app.config import settings config = context.config target_metadata = Base.metadata +def run_migrations_offline(): + context.configure(url=settings.database_url, target_metadata=target_metadata, literal_binds=True, dialect_opts={"paramstyle": "named"}) + with context.begin_transaction(): + context.run_migrations() def run_migrations_online(): config.set_main_option("sqlalchemy.url", settings.database_url) engine = engine_from_config(config.get_section(config.config_ini_section), prefix="sqlalchemy.", poolclass=pool.NullPool) @@ -11,4 +15,7 @@ def run_migrations_online(): context.configure(connection=connection, target_metadata=target_metadata) with context.begin_transaction(): context.run_migrations() -run_migrations_online() +if context.is_offline_mode(): + run_migrations_offline() +else: + run_migrations_online() diff --git a/central/alembic/versions/0015_ai_budget_and_product_cache.py b/central/alembic/versions/0015_ai_budget_and_product_cache.py index d9abf92..c9d4983 100644 --- a/central/alembic/versions/0015_ai_budget_and_product_cache.py +++ b/central/alembic/versions/0015_ai_budget_and_product_cache.py @@ -5,7 +5,7 @@ import sqlalchemy as sa revision = "0015_ai_budget_and_product_cache" -down_revision = "0014_ai_eval_cache" +down_revision = "0014" branch_labels = None depends_on = None diff --git a/central/alembic/versions/0017_control_plane_foundation.py b/central/alembic/versions/0017_control_plane_foundation.py new file mode 100644 index 0000000..d7bd697 --- /dev/null +++ b/central/alembic/versions/0017_control_plane_foundation.py @@ -0,0 +1,101 @@ +"""Add the operational control-plane records.""" + +from alembic import op +import sqlalchemy as sa + +revision = "0017_control_plane_foundation" +down_revision = "0016_blocked_product_registry" +branch_labels = None +depends_on = None + + +def upgrade(): + op.create_table( + "installations", + sa.Column("id", sa.Integer(), primary_key=True), + sa.Column("agent_id", sa.Integer(), sa.ForeignKey("agents.id"), nullable=False), + sa.Column("tenant_id", sa.Integer(), sa.ForeignKey("tenants.id"), nullable=False), + sa.Column("installation_id", sa.String(36), nullable=False), + sa.Column("display_name", sa.String(160), nullable=False, server_default=""), + sa.Column("customer_code", sa.String(64)), sa.Column("notes", sa.Text()), + sa.Column("hostname", sa.String(255)), sa.Column("windows_version", sa.String(128)), + sa.Column("machine_fingerprint", sa.String(128)), sa.Column("company_code", sa.String(64)), + sa.Column("company_name", sa.String(160)), sa.Column("company_cnpj_masked", sa.String(24)), + sa.Column("company_uf", sa.String(2)), sa.Column("company_crt", sa.String(16)), + sa.Column("firebird_version", sa.String(64)), sa.Column("firebird_ods", sa.String(32)), + sa.Column("database_identifier", sa.String(255)), sa.Column("capabilities", sa.JSON(), nullable=False, server_default="{}"), + sa.Column("fiscal_engine_version", sa.String(64)), sa.Column("dataset_versions", sa.JSON(), nullable=False, server_default="{}"), + sa.Column("started_at", sa.DateTime(timezone=True)), sa.Column("last_seen_at", sa.DateTime(timezone=True)), + sa.Column("last_status", sa.String(32)), sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now()), + sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now()), + sa.UniqueConstraint("installation_id", name="uq_installation_identity"), + sa.UniqueConstraint("agent_id", name="uq_installation_agent"), + ) + op.create_index("ix_installations_agent_id", "installations", ["agent_id"]) + op.create_index("ix_installations_tenant_id", "installations", ["tenant_id"]) + op.create_index("ix_installations_hostname", "installations", ["hostname"]) + op.create_index("ix_installations_customer_code", "installations", ["customer_code"]) + op.create_index("ix_installations_last_seen_at", "installations", ["last_seen_at"]) + op.create_index("ix_installations_installation_id", "installations", ["installation_id"]) + + op.create_table( + "agent_snapshots", + sa.Column("id", sa.Integer(), primary_key=True), sa.Column("installation_id", sa.Integer(), sa.ForeignKey("installations.id"), nullable=False), + sa.Column("total_products", sa.Integer(), nullable=False, server_default="0"), sa.Column("fiscally_complete", sa.Integer(), nullable=False, server_default="0"), + sa.Column("safe_blocked", sa.Integer(), nullable=False, server_default="0"), sa.Column("failed_products", sa.Integer(), nullable=False, server_default="0"), + sa.Column("coverage_percent", sa.Float(), nullable=False, server_default="0"), sa.Column("blockers", sa.JSON(), nullable=False, server_default="{}"), + sa.Column("last_job_status", sa.String(16)), sa.Column("last_job_at", sa.DateTime(timezone=True)), sa.Column("last_successful_job_at", sa.DateTime(timezone=True)), + sa.Column("agent_health", sa.String(16), nullable=False, server_default="UNKNOWN"), sa.Column("health_reason", sa.Text()), + sa.Column("version_drift", sa.String(16), nullable=False, server_default="UNKNOWN"), sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now()), + sa.UniqueConstraint("installation_id", name="uq_agent_snapshot_installation"), + ) + for name, column in (("installation", "installation_id"), ("health", "agent_health"), ("version_drift", "version_drift"), ("updated", "updated_at")): + op.create_index(f"ix_agent_snapshots_{name}", "agent_snapshots", [column]) + + op.create_table( + "agent_events", + sa.Column("id", sa.Integer(), primary_key=True), sa.Column("tenant_id", sa.Integer(), sa.ForeignKey("tenants.id"), nullable=False), + sa.Column("installation_id", sa.Integer(), sa.ForeignKey("installations.id"), nullable=False), sa.Column("agent_id", sa.Integer(), sa.ForeignKey("agents.id"), nullable=False), + sa.Column("severity", sa.String(16), nullable=False), sa.Column("event_type", sa.String(48), nullable=False), sa.Column("message", sa.Text(), nullable=False), + sa.Column("structured_context", sa.JSON(), nullable=False, server_default="{}"), sa.Column("job_id", sa.Integer(), sa.ForeignKey("jobs.id")), + sa.Column("product_code", sa.String(64)), sa.Column("blocker_id", sa.Integer(), sa.ForeignKey("blocked_products.id")), sa.Column("is_audit", sa.Boolean(), nullable=False, server_default="false"), + sa.Column("occurred_at", sa.DateTime(timezone=True), server_default=sa.func.now()), + ) + for name, column in (("tenant", "tenant_id"), ("installation", "installation_id"), ("agent", "agent_id"), ("severity", "severity"), ("event_type", "event_type"), ("job", "job_id"), ("product", "product_code"), ("blocker", "blocker_id"), ("audit", "is_audit"), ("occurred", "occurred_at")): + op.create_index(f"ix_agent_events_{name}", "agent_events", [column]) + + op.create_table( + "admin_users", + sa.Column("id", sa.Integer(), primary_key=True), sa.Column("username", sa.String(128), nullable=False, unique=True), + sa.Column("password_hash", sa.String(255), nullable=False), sa.Column("is_active", sa.Boolean(), nullable=False, server_default="true"), + sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now()), sa.Column("last_login_at", sa.DateTime(timezone=True)), + ) + op.create_index("ix_admin_users_username", "admin_users", ["username"]) + op.create_table( + "web_sessions", + sa.Column("id", sa.Integer(), primary_key=True), sa.Column("admin_user_id", sa.Integer(), sa.ForeignKey("admin_users.id"), nullable=False), + sa.Column("token_hash", sa.String(64), nullable=False, unique=True), sa.Column("csrf_hash", sa.String(64), nullable=False), + sa.Column("expires_at", sa.DateTime(timezone=True), nullable=False), sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now()), + sa.Column("last_seen_at", sa.DateTime(timezone=True), server_default=sa.func.now()), sa.Column("user_agent", sa.String(512)), sa.Column("remote_addr", sa.String(64)), + ) + op.create_index("ix_web_sessions_admin_user_id", "web_sessions", ["admin_user_id"]) + op.create_index("ix_web_sessions_token_hash", "web_sessions", ["token_hash"]) + op.create_index("ix_web_sessions_expires_at", "web_sessions", ["expires_at"]) + op.create_table( + "audit_events", + sa.Column("id", sa.Integer(), primary_key=True), sa.Column("admin_user_id", sa.Integer(), sa.ForeignKey("admin_users.id")), + sa.Column("tenant_id", sa.Integer(), sa.ForeignKey("tenants.id")), sa.Column("action", sa.String(64), nullable=False), + sa.Column("target_type", sa.String(64), nullable=False), sa.Column("target_id", sa.String(128), nullable=False), sa.Column("context", sa.JSON(), nullable=False, server_default="{}"), + sa.Column("severity", sa.String(16), nullable=False, server_default="INFO"), sa.Column("occurred_at", sa.DateTime(timezone=True), server_default=sa.func.now()), + ) + for name, column in (("admin", "admin_user_id"), ("tenant", "tenant_id"), ("action", "action"), ("occurred", "occurred_at")): + op.create_index(f"ix_audit_events_{name}", "audit_events", [column]) + + +def downgrade(): + op.drop_table("audit_events") + op.drop_table("web_sessions") + op.drop_table("admin_users") + op.drop_table("agent_events") + op.drop_table("agent_snapshots") + op.drop_table("installations") diff --git a/central/app/api.py b/central/app/api.py index 52d0f27..bacb32e 100644 --- a/central/app/api.py +++ b/central/app/api.py @@ -1,5 +1,6 @@ """Authenticated version-one Hub routes.""" import hashlib, json, logging +import uuid from datetime import datetime, timezone from fastapi import APIRouter, Depends, HTTPException, Request, Response import csv @@ -12,7 +13,9 @@ from .datasets.service import DatasetService from .jobs import claim_next_job, complete_job, create_job, fail_job from .blocked_products import list_blockers, resolve_blocker, resolve_product_blockers, upsert_blocker from .context_keys import validate_entry -from .models import BlockedProduct, FiscalContextEntry, ClassificationCache, ClassificationRequest, ClassificationResult, Dataset, Job +from .models import AgentSnapshot, BlockedProduct, FiscalContextEntry, ClassificationCache, ClassificationRequest, ClassificationResult, Dataset, Installation, Job +from .events import record_event +from .installation_identity import compute_agent_health, compute_version_drift, sanitize_registration logger = logging.getLogger("gti_fiscal_hub") @@ -158,15 +161,67 @@ def router(session_factory, bootstrap_token): def heartbeat(payload: dict, identity=Depends(principal)): session, agent = identity try: - agent.last_seen_at = datetime.now(timezone.utc) - agent.last_status = payload.get("status") + try: + supplied = sanitize_registration(payload) + except (TypeError, ValueError) as error: + raise HTTPException(422, f"invalid registration metadata: {error}") from error + now = datetime.now(timezone.utc) + agent.last_seen_at = now + agent.last_status = supplied.get("status") or payload.get("status") if payload.get("schema_fingerprint"): agent.schema_fingerprint = payload["schema_fingerprint"] if payload.get("agent_version"): agent.agent_version = payload["agent_version"] - session.add(agent); session.commit() + installation_key = supplied.get("installation_id") or str(uuid.uuid4()) + installation = session.scalar(select(Installation).where(Installation.agent_id == agent.id)) + if installation is not None and installation.installation_id != installation_key: + raise HTTPException(409, "installation_id conflicts with registered Agent") + if installation is None: + installation = Installation(agent_id=agent.id, tenant_id=agent.tenant_id, installation_id=installation_key, display_name=supplied.get("display_name") or agent.name) + session.add(installation); session.flush() + fields = {"display_name": supplied.get("display_name"), "customer_code": supplied.get("customer_code"), "notes": supplied.get("notes"), "hostname": supplied.get("hostname"), "windows_version": supplied.get("windows_version"), "machine_fingerprint": supplied.get("machine_fingerprint"), "company_code": supplied.get("company_code"), "company_name": supplied.get("company_name"), "company_cnpj_masked": supplied.get("company_cnpj_masked"), "company_uf": supplied.get("company_uf"), "company_crt": supplied.get("company_crt"), "firebird_version": supplied.get("firebird_version"), "firebird_ods": supplied.get("firebird_ods"), "database_identifier": supplied.get("database_identifier"), "capabilities": supplied.get("capabilities"), "fiscal_engine_version": supplied.get("fiscal_engine_version"), "dataset_versions": supplied.get("dataset_versions"), "last_status": supplied.get("status") or payload.get("status"), "last_seen_at": now} + for key, value in fields.items(): + if value is not None: setattr(installation, key, value) + if supplied.get("started_at"): + try: + installation.started_at = datetime.fromisoformat(str(supplied["started_at"]).replace("Z", "+00:00")) + except ValueError: + raise HTTPException(422, "started_at must be ISO-8601") + snapshot = session.scalar(select(AgentSnapshot).where(AgentSnapshot.installation_id == installation.id)) + if snapshot is None: + snapshot = AgentSnapshot(installation_id=installation.id) + session.add(snapshot) + if payload.get("coverage"): + coverage = payload["coverage"] + snapshot.total_products = int(coverage.get("total_products", coverage.get("total", snapshot.total_products)) or 0) + snapshot.fiscally_complete = int(coverage.get("fiscally_complete", coverage.get("complete", snapshot.fiscally_complete)) or 0) + snapshot.safe_blocked = int(coverage.get("safe_blocked", snapshot.safe_blocked) or 0) + snapshot.failed_products = int(coverage.get("failed_products", snapshot.failed_products) or 0) + snapshot.coverage_percent = float(coverage.get("coverage_percent", snapshot.coverage_percent) or 0) + snapshot.blockers = coverage.get("blockers", snapshot.blockers) or {} + snapshot.version_drift = compute_version_drift(agent.agent_version) + snapshot.agent_health, snapshot.health_reason = compute_agent_health(installation, supplemental=snapshot) + 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() 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, "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} + finally: session.close() + + @routes.post("/agents/register") + def register_agent(payload: dict, identity=Depends(principal)): + return heartbeat({**payload, "status": payload.get("status", "IDLE")}, identity) + + @routes.post("/agents/events") + def agent_event(payload: dict, identity=Depends(principal)): + session, agent = identity + try: + installation = session.scalar(select(Installation).where(Installation.agent_id == agent.id)) + if installation is None: + raise HTTPException(409, "Agent must heartbeat/register before sending events") + event = record_event(session, tenant_id=agent.tenant_id, installation_id=installation.id, agent_id=agent.id, severity=payload.get("severity", "INFO"), event_type=payload.get("event_type", "HEARTBEAT"), message=payload.get("message", ""), context=payload.get("context") or {}, job_id=payload.get("job_id"), product_code=payload.get("product_code"), blocker_id=payload.get("blocker_id")) + session.commit() + return {"event_id": event.id} finally: session.close() @routes.get("/agents/me") diff --git a/central/app/audit.py b/central/app/audit.py new file mode 100644 index 0000000..b2a4f42 --- /dev/null +++ b/central/app/audit.py @@ -0,0 +1,11 @@ +"""Immutable audit events for human control-plane actions.""" + +from .events import sanitize_event +from .models import AuditEvent + + +def record_audit(session, *, admin_user_id: int | None, action: str, target_type: str, target_id: str, context: dict | None = None, tenant_id: int | None = None, severity: str = "INFO"): + _, safe_context = sanitize_event(action, context or {}) + event = AuditEvent(admin_user_id=admin_user_id, tenant_id=tenant_id, action=action, target_type=target_type, target_id=str(target_id), context=safe_context, severity=severity) + session.add(event) + return event diff --git a/central/app/config.py b/central/app/config.py index c43d4dc..cd27f4c 100644 --- a/central/app/config.py +++ b/central/app/config.py @@ -27,6 +27,15 @@ class Settings: base_url: str = os.getenv("BASE_URL", "http://localhost:8000") # An Agent is considered ONLINE if a heartbeat arrived within this many seconds. agent_online_threshold_seconds: int = int(os.getenv("AGENT_ONLINE_THRESHOLD_SECONDS", "120")) + agent_heartbeat_interval_seconds: int = int(os.getenv("AGENT_HEARTBEAT_INTERVAL_SECONDS", "40")) + agent_offline_after_seconds: int = int(os.getenv("AGENT_OFFLINE_AFTER_SECONDS", "0")) + agent_server_version: str = os.getenv("CURRENT_SERVER_VERSION", "1.0.0rc1") + agent_event_retention_days: int = int(os.getenv("AGENT_EVENT_RETENTION_DAYS", "90")) + web_session_seconds: int = int(os.getenv("WEB_SESSION_SECONDS", "28800")) + web_cookie_secure: bool = os.getenv("WEB_COOKIE_SECURE", "0") == "1" + web_cookie_name: str = os.getenv("WEB_COOKIE_NAME", "gti_admin_session") + web_max_event_message: int = int(os.getenv("WEB_MAX_EVENT_MESSAGE", "2000")) + web_max_event_context_bytes: int = int(os.getenv("WEB_MAX_EVENT_CONTEXT_BYTES", "12000")) ncm_dataset_url: str = os.getenv("NCM_DATASET_URL", "https://portalunico.siscomex.gov.br/classif/api/publico/nomenclatura/download/json") cest_dataset_url: str = os.getenv("CEST_DATASET_URL", "https://www.confaz.fazenda.gov.br/legislacao/convenios/2018/CV142_18") dataset_storage_path: str = os.getenv("DATASET_STORAGE_PATH", "./data/datasets") diff --git a/central/app/control_api.py b/central/app/control_api.py new file mode 100644 index 0000000..1819905 --- /dev/null +++ b/central/app/control_api.py @@ -0,0 +1,125 @@ +"""Human-authenticated JSON API for the control panel.""" + +from datetime import datetime, timezone + +from fastapi import APIRouter, Depends, HTTPException, Request +from sqlalchemy import select + +from .audit import record_audit +from .control_plane import dashboard_summary, installation_view, list_blockers, list_installations, list_jobs, list_panel_events +from .human_auth import require_csrf, require_human +from .models import Agent, AgentSnapshot, BlockedProduct, Installation, Tenant +from .jobs import create_job + + +def router(session_factory): + routes = APIRouter(prefix="/v1") + + def identity(request: Request): + session = session_factory() + try: + web_session, admin = require_human(request, session) + return session, web_session, admin + except Exception: + session.close(); raise + + @routes.get("/dashboard/summary") + def summary(identity=Depends(identity)): + session, _, _ = identity + try: return dashboard_summary(session) + finally: session.close() + + @routes.get("/agents") + def agents(page: int = 1, page_size: int = 50, client: str | None = None, health: str | None = None, version: str | None = None, search: str | None = None, last_seen_before: str | None = None, identity=Depends(identity)): + session, _, _ = identity + try: return list_installations(session, page=page, page_size=page_size, client=client, health=health, version=version, search=search, last_seen_before=datetime.fromisoformat(last_seen_before) if last_seen_before else None) + finally: session.close() + + def get_installation(session, agent_id: int): + row = session.execute(select(Installation, Agent, Tenant, AgentSnapshot).join(Agent, Agent.id == Installation.agent_id).join(Tenant, Tenant.id == Installation.tenant_id).outerjoin(AgentSnapshot, AgentSnapshot.installation_id == Installation.id).where(Agent.id == agent_id)).first() + if not row: raise HTTPException(404, "Agent installation not found") + return row + + @routes.get("/agents/{agent_id}") + def agent_detail(agent_id: int, identity=Depends(identity)): + session, _, _ = identity + try: + installation, agent, tenant, snapshot = get_installation(session, agent_id) + return installation_view(installation, agent, tenant, snapshot) + finally: session.close() + + @routes.get("/agents/{agent_id}/health") + def agent_health(agent_id: int, identity=Depends(identity)): + return agent_detail(agent_id, identity) + + @routes.get("/agents/{agent_id}/coverage") + def agent_coverage(agent_id: int, identity=Depends(identity)): + detail = agent_detail(agent_id, identity) + return {key: detail[key] for key in ("total_products", "fiscally_complete", "safe_blocked", "failed_products", "coverage_percent", "blockers")} + + @routes.get("/agents/{agent_id}/blockers") + def agent_blockers(agent_id: int, page: int = 1, page_size: int = 50, blocker_type: str | None = None, field: str | None = None, identity=Depends(identity)): + session, _, _ = identity + try: + installation, agent, _, _ = get_installation(session, agent_id) + return list_blockers(session, tenant_id=installation.tenant_id, page=page, page_size=page_size, blocker_type=blocker_type, field=field) + finally: session.close() + + @routes.get("/agents/{agent_id}/jobs") + def agent_jobs(agent_id: int, page: int = 1, page_size: int = 50, status: str | None = None, kind: str | None = None, identity=Depends(identity)): + session, _, _ = identity + try: return list_jobs(session, agent_id=agent_id, page=page, page_size=page_size, status=status, kind=kind) + finally: session.close() + + @routes.get("/agents/{agent_id}/events") + def agent_events(agent_id: int, page: int = 1, page_size: int = 50, severity: str | None = None, event_type: str | None = None, since: str | None = None, until: str | None = None, identity=Depends(identity)): + session, _, _ = identity + try: + installation, _, _, _ = get_installation(session, agent_id) + return {"page": max(1, page), "page_size": min(100, max(1, page_size)), "items": list_panel_events(session, installation.id, page=page, page_size=page_size, severity=severity, event_type=event_type, since=datetime.fromisoformat(since) if since else None, until=datetime.fromisoformat(until) if until else None)} + finally: session.close() + + @routes.get("/blockers/global") + def global_blockers(page: int = 1, page_size: int = 50, blocker_type: str | None = None, field: str | None = None, identity=Depends(identity)): + session, _, _ = identity + try: return list_blockers(session, page=page, page_size=page_size, blocker_type=blocker_type, field=field) + finally: session.close() + + @routes.get("/jobs/global") + def global_jobs(page: int = 1, page_size: int = 50, status: str | None = None, kind: str | None = None, since: str | None = None, until: str | None = None, identity=Depends(identity)): + session, _, _ = identity + try: return list_jobs(session, page=page, page_size=page_size, status=status, kind=kind, since=datetime.fromisoformat(since) if since else None, until=datetime.fromisoformat(until) if until else None) + finally: session.close() + + @routes.post("/agents/{agent_id}/actions/reprocess") + def reprocess(agent_id: int, payload: dict, request: Request, identity=Depends(identity)): + session, web_session, admin = identity + try: + require_csrf(web_session, payload.get("csrf_token")) + if payload.get("confirm") is not True: raise HTTPException(400, "explicit confirmation required") + installation, agent, _, _ = get_installation(session, agent_id) + kind = payload.get("kind", "REPROCESS_ALL") + if kind not in {"REPROCESS_ALL", "REPROCESS_PRODUCTS"}: raise HTTPException(400, "unsupported action") + job_payload = payload.get("payload") or {} + if kind == "REPROCESS_PRODUCTS" and not job_payload.get("codigos"): raise HTTPException(400, "codigos required") + job = create_job(session, agent, kind, job_payload) + record_audit(session, admin_user_id=admin.id, tenant_id=installation.tenant_id, action=kind, target_type="installation", target_id=installation.installation_id, context={"job_id": job.id, "request_ip": request.client.host if request.client else None}) + session.commit() + return {"job_id": job.id, "status": job.status} + finally: session.close() + + @routes.post("/blockers/{blocker_id}/resolve-human") + def resolve_blocker(blocker_id: int, payload: dict, request: Request, identity=Depends(identity)): + session, web_session, admin = identity + try: + require_csrf(web_session, payload.get("csrf_token")) + if payload.get("confirm") is not True or not str(payload.get("context", "")).strip(): raise HTTPException(400, "confirmation and explicit context required") + blocker = session.get(BlockedProduct, blocker_id) + if not blocker: raise HTTPException(404, "blocker not found") + blocker.resolved_at = datetime.now(timezone.utc); blocker.resolution_source = "HUMAN_PANEL" + record_audit(session, admin_user_id=admin.id, tenant_id=blocker.tenant_id, action="RESOLVE_BLOCKER", target_type="blocker", target_id=str(blocker_id), context={"context": payload["context"]}) + session.commit() + return {"resolved": True, "blocker_id": blocker_id} + finally: session.close() + + return routes diff --git a/central/app/control_plane.py b/central/app/control_plane.py new file mode 100644 index 0000000..3e77432 --- /dev/null +++ b/central/app/control_plane.py @@ -0,0 +1,94 @@ +"""Bounded control-panel queries backed by installation snapshots.""" + +from __future__ import annotations + +from datetime import datetime, timezone + +from sqlalchemy import func, or_, select + +from .config import settings +from .installation_identity import compute_agent_health +from .models import Agent, AgentEvent, AgentSnapshot, BlockedProduct, Installation, Job, Tenant + + +def _page(page: int, page_size: int): + return max(1, page), min(100, max(1, page_size)) + + +def _iso(value): + return value.isoformat() if value else None + + +def installation_view(installation, agent, tenant, snapshot=None): + snapshot = snapshot or AgentSnapshot(installation_id=installation.id) + health, reason = compute_agent_health(installation, supplemental=snapshot) + return { + "agent_id": agent.id, "installation_id": installation.installation_id, "installation_db_id": installation.id, + "client": tenant.name if tenant else None, "tenant_id": installation.tenant_id, "display_name": installation.display_name, + "hostname": installation.hostname, "company_name": installation.company_name, "company_code": installation.company_code, + "company_cnpj": installation.company_cnpj_masked, "company_uf": installation.company_uf, "company_crt": installation.company_crt, + "agent_version": agent.agent_version, "version_drift": snapshot.version_drift, "health": health, "health_reason": reason, + "last_seen_at": _iso(installation.last_seen_at), "started_at": _iso(installation.started_at), "last_status": installation.last_status, + "total_products": snapshot.total_products, "fiscally_complete": snapshot.fiscally_complete, "safe_blocked": snapshot.safe_blocked, + "failed_products": snapshot.failed_products, "coverage_percent": snapshot.coverage_percent, "blockers": snapshot.blockers or {}, + "last_job_status": snapshot.last_job_status, "last_job_at": _iso(snapshot.last_job_at), "last_successful_job_at": _iso(snapshot.last_successful_job_at), + "firebird_version": installation.firebird_version, "firebird_ods": installation.firebird_ods, "database_identifier": installation.database_identifier, + "capabilities": installation.capabilities or {}, "fiscal_engine_version": installation.fiscal_engine_version, "dataset_versions": installation.dataset_versions or {}, + } + + +def list_installations(session, *, page=1, page_size=50, client=None, health=None, version=None, search=None, last_seen_before=None): + page, page_size = _page(page, page_size) + query = select(Installation, Agent, Tenant, AgentSnapshot).join(Agent, Agent.id == Installation.agent_id).join(Tenant, Tenant.id == Installation.tenant_id).outerjoin(AgentSnapshot, AgentSnapshot.installation_id == Installation.id) + if client: query = query.where(or_(Tenant.name.ilike(f"%{client}%"), Installation.customer_code.ilike(f"%{client}%"))) + if version: query = query.where(Agent.agent_version == version) + if search: query = query.where(or_(Installation.display_name.ilike(f"%{search}%"), Installation.hostname.ilike(f"%{search}%"), Tenant.name.ilike(f"%{search}%"))) + if last_seen_before: query = query.where(or_(Installation.last_seen_at.is_(None), Installation.last_seen_at < last_seen_before)) + ordered = query.order_by(Installation.display_name.asc(), Installation.id.asc()) + rows = session.execute(ordered).all() if health else session.execute(ordered.offset((page - 1) * page_size).limit(page_size)).all() + items = [installation_view(i, a, t, s) for i, a, t, s in rows] + total = None + if health: + items = [item for item in items if item["health"] == health] + total = len(items) + items = items[(page - 1) * page_size:page * page_size] + return {"page": page, "page_size": page_size, "items": items, "total": total if total is not None else len(items), "health_filter_applied": health} + + +def dashboard_summary(session): + query = select(Installation, Agent, Tenant, AgentSnapshot).join(Agent, Agent.id == Installation.agent_id).join(Tenant, Tenant.id == Installation.tenant_id).outerjoin(AgentSnapshot, AgentSnapshot.installation_id == Installation.id).order_by(Installation.id.asc()) + items = [installation_view(i, a, t, s) for i, a, t, s in session.execute(query).all()] + counts = {state: sum(item["health"] == state for item in items) for state in ("ONLINE", "DEGRADED", "OFFLINE", "FAILED", "UNKNOWN")} + totals = {key: sum(int(item[key] or 0) for item in items) for key in ("total_products", "fiscally_complete", "safe_blocked", "failed_products")} + blocker_counts = {} + for item in items: + for key, value in (item["blockers"] or {}).items(): blocker_counts[key] = blocker_counts.get(key, 0) + int(value or 0) + return {"agents_total": len(items), **{key.lower(): value for key, value in counts.items()}, **totals, "blockers": blocker_counts, "stale_agents": [item for item in items if item["health"] == "OFFLINE"], "outdated_agents": [item for item in items if item["version_drift"] == "OUTDATED"]} + + +def list_jobs(session, *, agent_id=None, page=1, page_size=50, status=None, kind=None, since=None, until=None): + page, page_size = _page(page, page_size) + query = select(Job).order_by(Job.created_at.desc(), Job.id.desc()) + if agent_id: query = query.where(Job.agent_id == agent_id) + if status: query = query.where(Job.status == status) + if kind: query = query.where(Job.kind == kind) + if since: query = query.where(Job.created_at >= since) + if until: query = query.where(Job.created_at <= until) + rows = session.scalars(query.offset((page - 1) * page_size).limit(page_size)).all() + return {"page": page, "page_size": page_size, "items": [{"job_id": row.id, "agent_id": row.agent_id, "kind": row.kind, "status": row.status, "created_at": _iso(row.created_at), "started_at": _iso(row.started_at), "finished_at": _iso(row.finished_at), "attempts": row.attempts, "result_summary": row.result_summary, "error": row.error} for row in rows], "total": len(rows)} + + +def list_blockers(session, *, tenant_id=None, agent_id=None, page=1, page_size=50, blocker_type=None, field=None, resolved=False): + page, page_size = _page(page, page_size) + query = select(BlockedProduct, Tenant).join(Tenant, Tenant.id == BlockedProduct.tenant_id).where(BlockedProduct.resolved_at.is_not(None) if resolved else BlockedProduct.resolved_at.is_(None)).order_by(BlockedProduct.last_seen_at.desc(), BlockedProduct.id.desc()) + if tenant_id: query = query.where(BlockedProduct.tenant_id == tenant_id) + if blocker_type: query = query.where(BlockedProduct.blocker_type == blocker_type) + if field: query = query.where(BlockedProduct.field == field) + rows = session.execute(query.offset((page - 1) * page_size).limit(page_size)).all() + return {"page": page, "page_size": page_size, "items": [{"id": row.id, "tenant_id": row.tenant_id, "client": tenant.name, "product_code": row.product_code, "description": row.description, "field": row.field, "blocker_type": row.blocker_type, "reason": row.reason_code, "evidence_summary": row.evidence_summary, "first_seen_at": _iso(row.first_seen_at), "last_seen_at": _iso(row.last_seen_at), "retry_count": row.retry_count, "resolved_at": _iso(row.resolved_at)} for row, tenant in rows], "total": len(rows)} + + +def list_panel_events(session, installation_db_id, **filters): + from .events import list_events + rows = list_events(session, installation_db_id, **filters) + return [{"id": row.id, "severity": row.severity, "event_type": row.event_type, "message": row.message, "context": row.structured_context, "job_id": row.job_id, "product_code": row.product_code, "occurred_at": _iso(row.occurred_at)} for row in rows] diff --git a/central/app/events.py b/central/app/events.py new file mode 100644 index 0000000..96bffe6 --- /dev/null +++ b/central/app/events.py @@ -0,0 +1,74 @@ +"""Structured operational events with defensive redaction and retention.""" + +from __future__ import annotations + +import json +import re +from datetime import datetime, timedelta, timezone + +from sqlalchemy import delete, select + +from .config import settings +from .models import AgentEvent + +EVENT_TYPES = { + "AGENT_STARTED", "AGENT_STOPPED", "HEARTBEAT", "PREFLIGHT_PASS", "PREFLIGHT_FAIL", "JOB_CLAIMED", "JOB_STARTED", + "JOB_COMPLETED", "JOB_FAILED", "PRODUCT_BLOCKED", "BLOCKER_RESOLVED", "DATASET_CHANGED", "COMPANY_CONTEXT_CHANGED", + "SCHEMA_CHANGED", "WRITE_PLANNED", "WRITE_COMPLETED", "WRITE_FAILED", "CONNECTION_ERROR", "RETRY", "VERSION_CHANGED", + "ADMIN_ACTION", +} +SEVERITIES = {"INFO", "WARNING", "ERROR", "CRITICAL"} +SECRET_RE = re.compile(r"(?i)(password|fb_password|token|api[_-]?key|secret|authorization|connection[_-]?string)\s*[:=]\s*[^,;\s]+") + + +def _redact(value): + if isinstance(value, dict): + return {str(key): "[REDACTED]" if str(key).lower() in {"password", "fb_password", "token", "api_key", "secret", "authorization", "connection_string"} else _redact(item) for key, item in value.items()} + if isinstance(value, list): + return [_redact(item) for item in value[:100]] + if isinstance(value, str): + return SECRET_RE.sub(lambda match: f"{match.group(1)}=[REDACTED]", value) + return value + + +def sanitize_event(message: str, context: dict | None = None) -> tuple[str, dict]: + safe_message = _redact(str(message or ""))[: settings.web_max_event_message] + safe_context = _redact(context or {}) + encoded = json.dumps(safe_context, ensure_ascii=False, default=str) + if len(encoded.encode()) > settings.web_max_event_context_bytes: + encoded = encoded.encode()[: settings.web_max_event_context_bytes].decode("utf-8", errors="ignore") + safe_context = {"truncated": True, "payload": encoded} + return safe_message, safe_context + + +def record_event(session, *, tenant_id: int, installation_id: int, agent_id: int, severity: str, event_type: str, message: str, context: dict | None = None, job_id: int | None = None, product_code: str | None = None, blocker_id: int | None = None, is_audit: bool = False, occurred_at: datetime | None = None): + severity = severity.upper() + event_type = event_type.upper() + if severity not in SEVERITIES: + raise ValueError("invalid event severity") + if event_type not in EVENT_TYPES: + raise ValueError("invalid event type") + safe_message, safe_context = sanitize_event(message, context) + event = AgentEvent(tenant_id=tenant_id, installation_id=installation_id, agent_id=agent_id, severity=severity, event_type=event_type, message=safe_message, structured_context=safe_context, job_id=job_id, product_code=product_code, blocker_id=blocker_id, is_audit=is_audit, occurred_at=occurred_at or datetime.now(timezone.utc)) + session.add(event) + return event + + +def purge_expired_events(session, now: datetime | None = None, retention_days: int | None = None) -> int: + cutoff = (now or datetime.now(timezone.utc)) - timedelta(days=retention_days if retention_days is not None else settings.agent_event_retention_days) + result = session.execute(delete(AgentEvent).where(AgentEvent.occurred_at < cutoff, AgentEvent.is_audit.is_(False), AgentEvent.severity != "CRITICAL")) + session.commit() + return result.rowcount or 0 + + +def list_events(session, installation_db_id: int, *, page: int = 1, page_size: int = 50, severity: str | None = None, event_type: str | None = None, job_id: int | None = None, product_code: str | None = None, since: datetime | None = None, until: datetime | None = None): + page = max(1, page); page_size = min(100, max(1, page_size)) + query = select(AgentEvent).where(AgentEvent.installation_id == installation_db_id) + for column, value in ((AgentEvent.severity, severity), (AgentEvent.event_type, event_type), (AgentEvent.job_id, job_id), (AgentEvent.product_code, product_code)): + if value is not None: + query = query.where(column == value) + if since: query = query.where(AgentEvent.occurred_at >= since) + if until: query = query.where(AgentEvent.occurred_at <= until) + query = query.order_by(AgentEvent.occurred_at.desc(), AgentEvent.id.desc()) + rows = session.scalars(query.offset((page - 1) * page_size).limit(page_size)).all() + return rows diff --git a/central/app/human_auth.py b/central/app/human_auth.py new file mode 100644 index 0000000..1806a90 --- /dev/null +++ b/central/app/human_auth.py @@ -0,0 +1,75 @@ +"""Separate human authentication backed by hashed passwords and sessions.""" + +from __future__ import annotations + +import base64 +import hashlib +import hmac +import os +import secrets +from datetime import datetime, timedelta, timezone + +from fastapi import HTTPException, Request +from sqlalchemy import select + +from .config import settings +from .models import AdminUser, WebSession + + +def _digest(value: str) -> str: + return hashlib.sha256(value.encode()).hexdigest() + + +def hash_password(password: str) -> str: + salt = os.urandom(16) + encoded = hashlib.pbkdf2_hmac("sha256", password.encode(), salt, 240_000) + return "pbkdf2_sha256$240000$%s$%s" % (base64.urlsafe_b64encode(salt).decode(), base64.urlsafe_b64encode(encoded).decode()) + + +def verify_password(password: str, stored: str) -> bool: + try: + algorithm, rounds, salt, digest = stored.split("$", 3) + candidate = hashlib.pbkdf2_hmac(algorithm.removeprefix("pbkdf2_"), password.encode(), base64.urlsafe_b64decode(salt), int(rounds)) + return hmac.compare_digest(base64.urlsafe_b64encode(candidate).decode(), digest) + except (ValueError, TypeError): + return False + + +def create_admin(session, username: str, password: str): + username = username.strip().lower() + if len(username) < 3 or len(password) < 12: + raise ValueError("username must have 3 characters and password 12 characters") + if session.scalar(select(AdminUser).where(AdminUser.username == username)): + raise ValueError("username already exists") + admin = AdminUser(username=username, password_hash=hash_password(password)) + session.add(admin); session.commit(); return admin + + +def create_session(session, admin: AdminUser, request: Request): + raw_token = secrets.token_urlsafe(32) + raw_csrf = secrets.token_urlsafe(32) + now = datetime.now(timezone.utc) + row = WebSession(admin_user_id=admin.id, token_hash=_digest(raw_token), csrf_hash=_digest(raw_csrf), expires_at=now + timedelta(seconds=settings.web_session_seconds), user_agent=request.headers.get("user-agent", "")[:512], remote_addr=request.client.host if request.client else None) + session.add(row); admin.last_login_at = now; session.commit() + return raw_token, raw_csrf + + +def current_session(request: Request, session): + token = request.cookies.get(settings.web_cookie_name) + if not token: return None, None + row = session.scalar(select(WebSession).where(WebSession.token_hash == _digest(token))) + if not row or row.expires_at.replace(tzinfo=timezone.utc) < datetime.now(timezone.utc): return None, None + admin = session.get(AdminUser, row.admin_user_id) + if not admin or not admin.is_active: return None, None + return row, admin + + +def require_human(request: Request, session): + web_session, admin = current_session(request, session) + if not admin: raise HTTPException(401, "human login required") + return web_session, admin + + +def require_csrf(web_session: WebSession, csrf_token: str | None): + if not csrf_token or not hmac.compare_digest(web_session.csrf_hash, _digest(csrf_token)): + raise HTTPException(403, "invalid CSRF token") diff --git a/central/app/installation_identity.py b/central/app/installation_identity.py new file mode 100644 index 0000000..450291b --- /dev/null +++ b/central/app/installation_identity.py @@ -0,0 +1,104 @@ +"""Stable installation identity and safe operational metadata helpers.""" + +from __future__ import annotations + +import hashlib +import re +import uuid +from datetime import datetime, timezone +from pathlib import Path + +from .config import settings + +SECRET_KEYS = {"password", "fb_password", "token", "agent_token", "api_key", "secret", "authorization", "connection_string"} + + +def utcnow() -> datetime: + return datetime.now(timezone.utc) + + +def ensure_installation_id(state_path: str, *, new_installation: bool = False) -> str: + """Persist a UUID beside the local Agent state without using hostname/IP.""" + path = Path(state_path) + identity_path = path.with_suffix(path.suffix + ".installation_id") + if identity_path.exists() and not new_installation: + value = identity_path.read_text(encoding="utf-8").strip() + try: + return str(uuid.UUID(value)) + except ValueError: + pass + value = str(uuid.uuid4()) + identity_path.parent.mkdir(parents=True, exist_ok=True) + temporary = identity_path.with_suffix(identity_path.suffix + ".tmp") + temporary.write_text(value + "\n", encoding="utf-8") + temporary.replace(identity_path) + return value + + +def mask_cnpj(value: str | None) -> str | None: + digits = re.sub(r"\D", "", value or "") + if len(digits) != 14: + return None if not value else "***" + return f"{digits[:2]}.***.***/{digits[8:12]}-{digits[12:]}" + + +def sanitize_database_path(value: str | None) -> str | None: + if not value: + return None + normalized = str(value).replace("\\", "/") + return normalized.rsplit("/", 1)[-1][:255] + + +def _safe_value(key: str, value): + if key.lower() in SECRET_KEYS: + return "[REDACTED]" + if isinstance(value, dict): + return {str(k): _safe_value(str(k), v) for k, v in value.items() if str(k).lower() not in SECRET_KEYS} + if isinstance(value, list): + return [_safe_value(key, item) for item in value[:100]] + return value + + +def sanitize_registration(payload: dict) -> dict: + allowed = { + "installation_id", "display_name", "customer_code", "notes", "hostname", "windows_version", "machine_fingerprint", + "company_code", "company_name", "company_cnpj", "company_uf", "company_crt", "firebird_version", "firebird_ods", + "database_identifier", "database_path", "capabilities", "fiscal_engine_version", "dataset_versions", "started_at", + "status", "schema_fingerprint", "agent_version", "last_job_id", "coverage", "health", "event_type", "message", "context", + } + result = {key: _safe_value(key, value) for key, value in payload.items() if key in allowed} + if "company_cnpj" in result: + result["company_cnpj_masked"] = mask_cnpj(result.pop("company_cnpj")) + if "database_path" in result: + result["database_identifier"] = sanitize_database_path(result.pop("database_path")) + if result.get("installation_id"): + result["installation_id"] = str(uuid.UUID(str(result["installation_id"]))) + return result + + +def compute_agent_health(snapshot, now: datetime | None = None, *, supplemental=None) -> tuple[str, str]: + now = now or utcnow() + last_seen = getattr(snapshot, "last_seen_at", None) + if last_seen is None: + return "UNKNOWN", "no heartbeat received" + if last_seen.tzinfo is None: + last_seen = last_seen.replace(tzinfo=timezone.utc) + offline_after = settings.agent_offline_after_seconds or settings.agent_heartbeat_interval_seconds * 3 + if (now - last_seen).total_seconds() > offline_after: + return "OFFLINE", f"last heartbeat exceeds {offline_after}s" + source = supplemental or snapshot + if any(getattr(candidate, "agent_health", None) == "FAILED" or getattr(candidate, "last_status", None) == "FAILED" for candidate in (snapshot, supplemental) if candidate is not None): + return "FAILED", "persistent technical failure" + if getattr(source, "failed_products", 0) or getattr(source, "last_job_status", "") == "FAILED": + return "DEGRADED", "recent technical or job failure" + return "ONLINE", "recent heartbeat without critical technical failure" + + +def compute_version_drift(agent_version: str | None) -> str: + if not agent_version: + return "UNKNOWN" + return "CURRENT" if agent_version == settings.agent_server_version else "OUTDATED" + + +def machine_fingerprint(parts: list[str]) -> str: + return hashlib.sha256("|".join(parts).encode()).hexdigest() diff --git a/central/app/main.py b/central/app/main.py index 7308cc4..0c04a3e 100644 --- a/central/app/main.py +++ b/central/app/main.py @@ -4,9 +4,12 @@ import logging from collections.abc import Callable from fastapi import FastAPI, HTTPException +from fastapi.staticfiles import StaticFiles from .config import settings from .db import SessionLocal, database_is_available from .api import router +from .control_api import router as control_api_router +from .web import router as web_router from .version import GTI_FISCAL_VERSION logging.basicConfig(level=getattr(logging, settings.log_level.upper(), logging.INFO)) @@ -33,6 +36,9 @@ def create_app(database_probe: Callable[[], bool] = database_is_available, sessi return {"version": GTI_FISCAL_VERSION, "api_compat": API_COMPAT_MAJOR_MINOR, "engine_version": ENGINE_VERSION, "decision_version": DECISION_VERSION} app.include_router(router(session_factory, settings.agent_dev_token if bootstrap_token is None else bootstrap_token)) + app.include_router(control_api_router(session_factory)) + app.include_router(web_router(session_factory)) + app.mount("/static", StaticFiles(directory="central/static"), name="static") return app diff --git a/central/app/models.py b/central/app/models.py index 11e2732..07291e3 100644 --- a/central/app/models.py +++ b/central/app/models.py @@ -235,6 +235,109 @@ class Agent(Base): updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now()) +class Installation(Base): + __tablename__ = "installations" + __table_args__ = (UniqueConstraint("installation_id", name="uq_installation_identity"), UniqueConstraint("agent_id", name="uq_installation_agent")) + id: Mapped[int] = mapped_column(primary_key=True) + agent_id: Mapped[int] = mapped_column(ForeignKey("agents.id"), index=True) + tenant_id: Mapped[int] = mapped_column(ForeignKey("tenants.id"), index=True) + installation_id: Mapped[str] = mapped_column(String(36), index=True) + display_name: Mapped[str] = mapped_column(String(160), default="", server_default="") + customer_code: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True) + notes: Mapped[str | None] = mapped_column(Text, nullable=True) + hostname: Mapped[str | None] = mapped_column(String(255), nullable=True, index=True) + windows_version: Mapped[str | None] = mapped_column(String(128), nullable=True) + machine_fingerprint: Mapped[str | None] = mapped_column(String(128), nullable=True) + company_code: Mapped[str | None] = mapped_column(String(64), nullable=True) + company_name: Mapped[str | None] = mapped_column(String(160), nullable=True) + company_cnpj_masked: Mapped[str | None] = mapped_column(String(24), nullable=True) + company_uf: Mapped[str | None] = mapped_column(String(2), nullable=True) + company_crt: Mapped[str | None] = mapped_column(String(16), nullable=True) + firebird_version: Mapped[str | None] = mapped_column(String(64), nullable=True) + firebird_ods: Mapped[str | None] = mapped_column(String(32), nullable=True) + database_identifier: Mapped[str | None] = mapped_column(String(255), nullable=True) + capabilities: Mapped[dict] = mapped_column(JSON, default=dict, server_default="{}") + fiscal_engine_version: Mapped[str | None] = mapped_column(String(64), nullable=True) + dataset_versions: Mapped[dict] = mapped_column(JSON, default=dict, server_default="{}") + 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) + 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()) + + +class AgentSnapshot(Base): + __tablename__ = "agent_snapshots" + id: Mapped[int] = mapped_column(primary_key=True) + installation_id: Mapped[int] = mapped_column(ForeignKey("installations.id"), unique=True, index=True) + total_products: Mapped[int] = mapped_column(Integer, default=0, server_default="0") + fiscally_complete: Mapped[int] = mapped_column(Integer, default=0, server_default="0") + safe_blocked: Mapped[int] = mapped_column(Integer, default=0, server_default="0") + failed_products: Mapped[int] = mapped_column(Integer, default=0, server_default="0") + coverage_percent: Mapped[float] = mapped_column(default=0.0, server_default="0") + blockers: Mapped[dict] = mapped_column(JSON, default=dict, server_default="{}") + last_job_status: Mapped[str | None] = mapped_column(String(16), nullable=True) + last_job_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + last_successful_job_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + agent_health: Mapped[str] = mapped_column(String(16), default="UNKNOWN", server_default="UNKNOWN", index=True) + health_reason: Mapped[str | None] = mapped_column(Text, nullable=True) + version_drift: Mapped[str] = mapped_column(String(16), default="UNKNOWN", server_default="UNKNOWN", index=True) + updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now(), index=True) + + +class AgentEvent(Base): + __tablename__ = "agent_events" + id: Mapped[int] = mapped_column(primary_key=True) + tenant_id: Mapped[int] = mapped_column(ForeignKey("tenants.id"), index=True) + installation_id: Mapped[int] = mapped_column(ForeignKey("installations.id"), index=True) + agent_id: Mapped[int] = mapped_column(ForeignKey("agents.id"), index=True) + severity: Mapped[str] = mapped_column(String(16), index=True) + event_type: Mapped[str] = mapped_column(String(48), index=True) + message: Mapped[str] = mapped_column(Text) + structured_context: Mapped[dict] = mapped_column(JSON, default=dict) + job_id: Mapped[int | None] = mapped_column(ForeignKey("jobs.id"), nullable=True, index=True) + product_code: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True) + blocker_id: Mapped[int | None] = mapped_column(ForeignKey("blocked_products.id"), nullable=True, index=True) + is_audit: Mapped[bool] = mapped_column(Boolean, default=False, server_default="false", index=True) + occurred_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now(), index=True) + + +class AdminUser(Base): + __tablename__ = "admin_users" + id: Mapped[int] = mapped_column(primary_key=True) + username: Mapped[str] = mapped_column(String(128), unique=True, index=True) + password_hash: Mapped[str] = mapped_column(String(255)) + is_active: Mapped[bool] = mapped_column(Boolean, default=True, server_default="true") + created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now()) + last_login_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + + +class WebSession(Base): + __tablename__ = "web_sessions" + id: Mapped[int] = mapped_column(primary_key=True) + admin_user_id: Mapped[int] = mapped_column(ForeignKey("admin_users.id"), index=True) + token_hash: Mapped[str] = mapped_column(String(64), unique=True, index=True) + csrf_hash: Mapped[str] = mapped_column(String(64)) + expires_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), index=True) + created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now()) + last_seen_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now()) + user_agent: Mapped[str | None] = mapped_column(String(512), nullable=True) + remote_addr: Mapped[str | None] = mapped_column(String(64), nullable=True) + + +class AuditEvent(Base): + __tablename__ = "audit_events" + id: Mapped[int] = mapped_column(primary_key=True) + admin_user_id: Mapped[int | None] = mapped_column(ForeignKey("admin_users.id"), nullable=True, index=True) + tenant_id: Mapped[int | None] = mapped_column(ForeignKey("tenants.id"), nullable=True, index=True) + action: Mapped[str] = mapped_column(String(64), index=True) + target_type: Mapped[str] = mapped_column(String(64)) + target_id: Mapped[str] = mapped_column(String(128)) + context: Mapped[dict] = mapped_column(JSON, default=dict) + severity: Mapped[str] = mapped_column(String(16), default="INFO", server_default="INFO") + occurred_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now(), index=True) + + class Job(Base): __tablename__ = "jobs" id: Mapped[int] = mapped_column(primary_key=True) diff --git a/central/app/web.py b/central/app/web.py new file mode 100644 index 0000000..9c175e4 --- /dev/null +++ b/central/app/web.py @@ -0,0 +1,122 @@ +"""Server-rendered operator pages for the GTI Fiscal control panel.""" + +from urllib.parse import parse_qs + +from fastapi import APIRouter, Request +from fastapi.responses import HTMLResponse, RedirectResponse +from fastapi.templating import Jinja2Templates +from sqlalchemy import select + +from .config import settings +from .control_plane import dashboard_summary, list_blockers, list_installations, list_jobs, list_panel_events +from .human_auth import create_session, current_session, require_csrf, verify_password +from .models import AdminUser, Installation, Agent +from .jobs import create_job +from .audit import record_audit + + +def router(session_factory): + routes = APIRouter() + templates = Jinja2Templates(directory="central/templates") + + def auth(request, session): + return current_session(request, session) + + def render(request, template, context=None, status_code=200): + return templates.TemplateResponse(template, {"request": request, **(context or {})}, status_code=status_code) + + @routes.get("/login", response_class=HTMLResponse) + def login_page(request: Request): + return render(request, "login.html", {"error": None}) + + @routes.post("/login") + async def login(request: Request): + values = {key: items[0] for key, items in parse_qs((await request.body()).decode()).items() if items} + with session_factory() as session: + admin = session.scalar(select(AdminUser).where(AdminUser.username == values.get("username", "").strip().lower())) + if not admin or not admin.is_active or not verify_password(values.get("password", ""), admin.password_hash): + return render(request, "login.html", {"error": "Credenciais inválidas"}, 401) + token, csrf = create_session(session, admin, request) + response = RedirectResponse("/", status_code=303) + response.set_cookie(settings.web_cookie_name, token, max_age=settings.web_session_seconds, httponly=True, secure=settings.web_cookie_secure, samesite="lax") + response.set_cookie("gti_csrf", csrf, max_age=settings.web_session_seconds, httponly=False, secure=settings.web_cookie_secure, samesite="lax") + return response + + @routes.post("/logout") + def logout(request: Request): + with session_factory() as session: + web_session, admin = current_session(request, session) + if web_session: + session.delete(web_session); session.commit() + response = RedirectResponse("/login", status_code=303) + response.delete_cookie(settings.web_cookie_name); response.delete_cookie("gti_csrf") + return response + + def page_auth(request): + session = session_factory() + web_session, admin = current_session(request, session) + if not admin: + session.close(); return None, None, RedirectResponse("/login", status_code=303) + return session, (web_session, admin), None + + async def form_values(request): + return {key: items[0] for key, items in parse_qs((await request.body()).decode()).items() if items} + + @routes.get("/", response_class=HTMLResponse) + def dashboard(request: Request): + session, auth_data, redirect = page_auth(request) + if redirect: return redirect + try: return render(request, "dashboard.html", {"summary": dashboard_summary(session), "auth": auth_data}) + finally: session.close() + + @routes.get("/agents", response_class=HTMLResponse) + def agents(request: Request, page: int = 1, search: str | None = None, health: str | None = None): + session, auth_data, redirect = page_auth(request) + if redirect: return redirect + try: return render(request, "agents.html", {"result": list_installations(session, page=page, page_size=50, search=search, health=health), "auth": auth_data, "search": search or "", "health_filter": health or ""}) + finally: session.close() + + @routes.get("/agents/{agent_id}", response_class=HTMLResponse) + def agent_detail(request: Request, agent_id: int, tab: str = "overview"): + session, auth_data, redirect = page_auth(request) + if redirect: return redirect + try: + result = list_installations(session, page=1, page_size=100) + detail = next((item for item in result["items"] if item["agent_id"] == agent_id), None) + if detail is None: return HTMLResponse("Agent not found", status_code=404) + installation = session.scalar(select(Installation).where(Installation.agent_id == agent_id)) + return render(request, "agent_detail.html", {"detail": detail, "tab": tab, "blockers": list_blockers(session, tenant_id=installation.tenant_id, page=1, page_size=25), "jobs": list_jobs(session, agent_id=agent_id, page=1, page_size=25), "events": list_panel_events(session, installation.id, page=1, page_size=25), "auth": auth_data, "csrf_token": request.cookies.get("gti_csrf", "")}) + finally: session.close() + + @routes.post("/agents/{agent_id}/reprocess") + async def reprocess_agent(request: Request, agent_id: int): + session, auth_data, redirect = page_auth(request) + if redirect: return redirect + try: + values = await form_values(request); web_session, admin = auth_data + require_csrf(web_session, values.get("csrf_token")) + if values.get("confirm") != "yes": return HTMLResponse("confirmation required", status_code=400) + agent = session.get(Agent, agent_id) + installation = session.scalar(select(Installation).where(Installation.agent_id == agent_id)) + if not agent or not installation: return HTMLResponse("Agent not found", status_code=404) + job = create_job(session, agent, "REPROCESS_ALL", {}) + record_audit(session, admin_user_id=admin.id, tenant_id=installation.tenant_id, action="REPROCESS_ALL", target_type="installation", target_id=installation.installation_id, context={"job_id": job.id}) + session.commit() + return RedirectResponse(f"/agents/{agent_id}?tab=jobs", status_code=303) + finally: session.close() + + @routes.get("/blockers", response_class=HTMLResponse) + def blockers(request: Request, page: int = 1, blocker_type: str | None = None): + session, auth_data, redirect = page_auth(request) + if redirect: return redirect + try: return render(request, "blockers.html", {"result": list_blockers(session, page=page, page_size=50, blocker_type=blocker_type), "auth": auth_data}) + finally: session.close() + + @routes.get("/jobs", response_class=HTMLResponse) + def jobs(request: Request, page: int = 1, status: str | None = None): + session, auth_data, redirect = page_auth(request) + if redirect: return redirect + try: return render(request, "jobs.html", {"result": list_jobs(session, page=page, page_size=50, status=status), "auth": auth_data}) + finally: session.close() + + return routes diff --git a/central/pyproject.toml b/central/pyproject.toml index ef4e591..4bee8a8 100644 --- a/central/pyproject.toml +++ b/central/pyproject.toml @@ -4,6 +4,7 @@ version = "1.0.0rc1" requires-python = ">=3.10" dependencies = [ "fastapi>=0.115,<1", + "jinja2>=3.1,<4", "uvicorn[standard]>=0.30,<1", "sqlalchemy>=2.0,<3", "psycopg[binary]>=3.2,<4", diff --git a/central/scripts/create_admin.py b/central/scripts/create_admin.py new file mode 100644 index 0000000..ff85b55 --- /dev/null +++ b/central/scripts/create_admin.py @@ -0,0 +1,21 @@ +"""Create a human panel administrator without ever printing the password.""" + +import getpass + +from central.app.db import SessionLocal +from central.app.human_auth import create_admin + + +def main(): + username = input("Admin username: ").strip() + password = getpass.getpass("Admin password (12+ chars): ") + confirmation = getpass.getpass("Confirm password: ") + if password != confirmation: + raise SystemExit("passwords do not match") + with SessionLocal() as session: + admin = create_admin(session, username, password) + print(f"created admin {admin.username}") + + +if __name__ == "__main__": + main() diff --git a/central/scripts/generate_control_panel_fixture.py b/central/scripts/generate_control_panel_fixture.py new file mode 100644 index 0000000..b4a7bfe --- /dev/null +++ b/central/scripts/generate_control_panel_fixture.py @@ -0,0 +1,52 @@ +"""Generate explicit, deterministic non-production control-panel data.""" + +import argparse +import uuid +from datetime import datetime, timedelta, timezone + +from central.app.config import settings +from central.app.db import SessionLocal +from central.app.models import Agent, AgentEvent, AgentSnapshot, BlockedProduct, Installation, Job, Tenant +from sqlalchemy import create_engine +from sqlalchemy.orm import sessionmaker + + +def generate_fixture(session, agent_count=100, seed=7): + now = datetime.now(timezone.utc) + states = ["ONLINE"] * 55 + ["DEGRADED"] * 20 + ["OFFLINE"] * 15 + ["FAILED"] * 10 + created = {"agents": 0, "events": 0, "blockers": 0, "jobs": 0} + for index in range(agent_count): + tenant = Tenant(name=f"Fixture Client {index:03d}"); session.add(tenant); session.flush() + agent = Agent(tenant_id=tenant.id, name=f"fixture-agent-{index:03d}", token_hash=(f"fixture-{index:03d}".ljust(64, "0"))[:64], agent_version="1.0.0rc1"); session.add(agent); session.flush() + state = states[index % len(states)] + last_seen = now - timedelta(seconds=10 if state in {"ONLINE", "DEGRADED"} else 10_000) + installation = Installation(agent_id=agent.id, tenant_id=tenant.id, installation_id=str(uuid.uuid5(uuid.NAMESPACE_DNS, f"fixture-{index}")), display_name=f"{tenant.name} - Servidor Principal", customer_code=f"F{index:03d}", hostname=f"srv-{index:03d}", company_name=f"Empresa {index:03d}", company_uf="BA", company_crt="1", last_seen_at=last_seen, last_status="IDLE") + session.add(installation); session.flush() + snapshot = AgentSnapshot(installation_id=installation.id, total_products=5000 + index, fiscally_complete=4900 + index, safe_blocked=index % 9, failed_products=1 if state == "FAILED" else 0, coverage_percent=98.0, blockers={"CFOP": index % 7, "ST": index % 5}, agent_health=state, last_job_status="FAILED" if state in {"DEGRADED", "FAILED"} else "COMPLETED") + session.add(snapshot); created["agents"] += 1 + for event_index in range(5): + session.add(AgentEvent(tenant_id=tenant.id, installation_id=installation.id, agent_id=agent.id, severity="ERROR" if state == "FAILED" and event_index == 0 else "INFO", event_type="JOB_FAILED" if state in {"DEGRADED", "FAILED"} and event_index == 0 else "HEARTBEAT", message="Fixture operational event", structured_context={"fixture": True}, occurred_at=now - timedelta(minutes=event_index))) + created["events"] += 1 + session.add(Job(agent_id=agent.id, kind="REPROCESS_ALL", status="FAILED" if state in {"DEGRADED", "FAILED"} else "COMPLETED", payload={}, result_summary={"fixture": True}, error="fixture technical failure" if state in {"DEGRADED", "FAILED"} else None)) + created["jobs"] += 1 + if index % 2 == 0: + session.add(BlockedProduct(tenant_id=tenant.id, company=tenant.name, product_code=f"P{index:06d}", description="Fixture product", blocker_type="CFOP", field="CFOP", reason_code="INSUFFICIENT_EVIDENCE", evidence_summary={"fixture": True}, input_fingerprint=f"{index:064d}", blocker_fingerprint=f"{index+1000:064d}")) + created["blockers"] += 1 + session.commit() + return created + + +def main(): + parser = argparse.ArgumentParser(); parser.add_argument("--database-url", default=None); parser.add_argument("--agents", type=int, default=100); parser.add_argument("--seed", type=int, default=7); parser.add_argument("--confirm-non-production", action="store_true") + args = parser.parse_args() + if settings.environment == "production" or not args.confirm_non_production: + raise SystemExit("refusing fixture generation without --confirm-non-production in non-production") + session_factory = SessionLocal + if args.database_url: + session_factory = sessionmaker(bind=create_engine(args.database_url)) + with session_factory() as session: + print(generate_fixture(session, args.agents, args.seed)) + + +if __name__ == "__main__": + main() diff --git a/central/scripts/maintain_events.py b/central/scripts/maintain_events.py new file mode 100644 index 0000000..6dc30fc --- /dev/null +++ b/central/scripts/maintain_events.py @@ -0,0 +1,22 @@ +"""Expire eligible operational events; audit and critical events are preserved.""" + +import argparse + +from central.app.db import SessionLocal +from central.app.events import purge_expired_events + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument("--dry-run", action="store_true") + parser.add_argument("--retention-days", type=int, default=None) + args = parser.parse_args() + if args.dry_run: + print("dry-run: no events deleted") + return + with SessionLocal() as session: + print(f"deleted_events={purge_expired_events(session, retention_days=args.retention_days)}") + + +if __name__ == "__main__": + main() diff --git a/central/static/control-panel.css b/central/static/control-panel.css new file mode 100644 index 0000000..72a949e --- /dev/null +++ b/central/static/control-panel.css @@ -0,0 +1 @@ +:root{--ink:#172126;--muted:#6d7c82;--paper:#f5f7f6;--card:#fff;--line:#dce4e1;--accent:#0e766e;--accent-soft:#dff1ec;--danger:#b64242;--warn:#a86b18;--shadow:0 10px 30px #183b3210}*{box-sizing:border-box}body{margin:0;background:var(--paper);color:var(--ink);font:14px/1.5 Inter,ui-sans-serif,system-ui,sans-serif}.topbar{height:68px;background:#112c2b;color:#f5fffb;display:flex;align-items:center;padding:0 clamp(18px,4vw,64px);gap:42px}.brand{font-weight:900;letter-spacing:.1em;color:#fff;text-decoration:none;font-size:17px}.brand span{color:#8ed6bd}.topbar nav{display:flex;gap:22px;flex:1}.topbar nav a{color:#c2d9d2;text-decoration:none}.topbar nav a:hover{color:#fff}.ghost{background:transparent;border:1px solid #ffffff44;color:#fff;padding:7px 13px;border-radius:8px}.shell{max-width:1440px;margin:0 auto;padding:38px clamp(18px,4vw,64px) 70px}.page-heading,.detail-head{display:flex;justify-content:space-between;align-items:flex-end;gap:20px;margin-bottom:26px}h1,h2,p{margin-top:0}h1{font-size:clamp(28px,3vw,42px);line-height:1.05;margin-bottom:8px;letter-spacing:-.04em}h2{font-size:17px;margin-bottom:16px}.eyebrow{color:var(--accent);font-size:11px;font-weight:800;letter-spacing:.14em;margin-bottom:9px}.muted,small{color:var(--muted)}small{display:block;font-size:12px}.live-dot{color:var(--accent);font-size:12px}.cards{display:grid;grid-template-columns:repeat(6,1fr);gap:12px;margin-bottom:20px}.cards.compact{grid-template-columns:repeat(4,1fr)}.metric,.panel{background:var(--card);border:1px solid var(--line);box-shadow:var(--shadow);border-radius:14px}.metric{padding:17px}.metric span{color:var(--muted);font-size:12px}.metric strong{font-size:29px;display:block;margin-top:8px;letter-spacing:-.04em}.split{display:grid;grid-template-columns:1fr 1fr;gap:20px;margin-bottom:20px}.panel{padding:22px;margin-bottom:20px}.panel-title{display:flex;justify-content:space-between;align-items:center;gap:12px}.panel-title a{color:var(--accent);font-size:12px}.coverage{display:flex;gap:clamp(20px,5vw,70px)}.coverage strong{font-size:28px;display:block}.coverage span{color:var(--muted);font-size:12px}.bar-row,.list-row,.event-row{display:flex;align-items:center;justify-content:space-between;border-top:1px solid var(--line);padding:11px 0}.bar-row:first-of-type,.list-row:first-of-type{border-top:0}.list-row{color:inherit;text-decoration:none}.list-row span{min-width:0}.list-row b,.list-row small{overflow:hidden;text-overflow:ellipsis;white-space:nowrap}.badge{display:inline-block;padding:4px 8px;border-radius:999px;font-size:10px;font-weight:800;letter-spacing:.04em;background:#e9eeee;color:#526166}.badge.large{padding:8px 12px}.badge.online,.badge.completed{background:var(--accent-soft);color:#126d5b}.badge.degraded,.badge.warning,.badge.running{background:#fff0d6;color:var(--warn)}.badge.offline,.badge.failed,.badge.error{background:#ffe3e3;color:var(--danger)}.badge.blocked{background:#eee8fa;color:#7553a3}.table-wrap{overflow:auto;padding:0}table{width:100%;border-collapse:collapse;min-width:840px}th,td{text-align:left;padding:14px 18px;border-bottom:1px solid var(--line);vertical-align:top}th{font-size:11px;text-transform:uppercase;letter-spacing:.08em;color:var(--muted);background:#f9fbfa}td a{color:var(--ink);text-decoration:none}.filters{display:flex;gap:8px;flex-wrap:wrap}input,select{border:1px solid var(--line);border-radius:8px;padding:9px 11px;background:#fff;color:var(--ink)}button{cursor:pointer;border:0;border-radius:8px;padding:9px 14px;font-weight:700}.primary{background:var(--accent);color:white}.secondary{background:var(--accent-soft);color:#126d5b}.detail-head{align-items:center}.tabs{display:flex;gap:5px;margin-bottom:20px;border-bottom:1px solid var(--line)}.tabs a{padding:10px 14px;text-decoration:none;color:var(--muted);text-transform:capitalize}.tabs a.active{color:var(--accent);border-bottom:2px solid var(--accent);font-weight:800}.facts{display:grid;grid-template-columns:130px 1fr;gap:10px;margin:0}.facts dt{color:var(--muted)}.facts dd{margin:0;font-weight:600}.event-row{gap:12px;justify-content:flex-start}.event-time{color:var(--muted);font-size:11px;min-width:150px}.severity{font-size:10px;font-weight:800;margin-left:auto}.severity.error,.severity.critical{color:var(--danger)}.login-page{display:grid;place-items:center;min-height:100vh;background:#112c2b}.login-card{background:#fff;padding:36px;border-radius:18px;width:min(410px,calc(100% - 32px));box-shadow:var(--shadow)}.login-card label{display:block;margin:16px 0 6px;font-weight:700}.login-card input{width:100%;margin-top:6px}.login-card button{width:100%;margin-top:20px}.alert{padding:10px;border-radius:8px}.alert.error{color:var(--danger);background:#ffe3e3}.error-text{color:var(--danger)}@media(max-width:900px){.cards{grid-template-columns:repeat(3,1fr)}.split{grid-template-columns:1fr}.topbar{gap:16px;padding:0 18px}.topbar nav{gap:11px;overflow:auto}.topbar nav a{white-space:nowrap}}@media(max-width:560px){.cards,.cards.compact{grid-template-columns:repeat(2,1fr)}.page-heading,.detail-head{display:block}.page-heading .filters{margin-top:18px}.filters input,.filters select,.filters button{width:100%}.shell{padding-top:26px}.panel{padding:16px}.event-time{min-width:0}.event-row{display:grid;grid-template-columns:1fr auto}.event-row span:nth-child(2){grid-column:1 / -1}} diff --git a/central/static/control-panel.js b/central/static/control-panel.js new file mode 100644 index 0000000..6543b54 --- /dev/null +++ b/central/static/control-panel.js @@ -0,0 +1,4 @@ +document.addEventListener("click", event => { + const action = event.target.closest("[data-confirm]"); + if (action && !window.confirm(action.dataset.confirm)) event.preventDefault(); +}); diff --git a/central/templates/agent_detail.html b/central/templates/agent_detail.html new file mode 100644 index 0000000..38e6b5e --- /dev/null +++ b/central/templates/agent_detail.html @@ -0,0 +1,9 @@ +{% extends "base.html" %} +{% block title %}{{ detail.display_name }} | GTI Fiscal{% endblock %} +{% block content %} +

{{ detail.client }} / INSTALAÇÃO

{{ detail.display_name or detail.hostname or detail.installation_id }}

{{ detail.hostname or "hostname indisponível" }} · Agent {{ detail.agent_id }} · Installation {{ detail.installation_id }}

{{ detail.health }}
+
Produtos{{ detail.total_products }}
Completos{{ detail.fiscally_complete }}
Safe blocked{{ detail.safe_blocked }}
Coverage{{ "%.1f"|format(detail.coverage_percent) }}%
+ +

Contexto da empresa

Empresa
{{ detail.company_name or "-" }}
UF / CRT
{{ detail.company_uf or "-" }} / {{ detail.company_crt or "-" }}
Firebird
{{ detail.firebird_version or "-" }} · ODS {{ detail.firebird_ods or "-" }}
Último heartbeat
{{ detail.last_seen_at or "-" }}
Último job
{{ detail.last_job_status or "-" }} · {{ detail.last_job_at or "-" }}

Eventos recentes

{% for event in events[:8] %}
{{ event.occurred_at }}{{ event.event_type }}{{ event.message }}{{ event.severity }}
{% else %}

Nenhum evento.

{% endfor %}
+

Blockers abertos

Ver todos
{% for item in blockers.items[:8] %}
{{ item.product_code }} · {{ item.field }}{{ item.reason }}{{ item.blocker_type }}
{% else %}

Nenhum blocker aberto.

{% endfor %}
+{% endblock %} diff --git a/central/templates/agents.html b/central/templates/agents.html new file mode 100644 index 0000000..70905ed --- /dev/null +++ b/central/templates/agents.html @@ -0,0 +1 @@ +{% extends "base.html" %}{% block title %}Agents | GTI Fiscal{% endblock %}{% block content %}

INVENTÁRIO OPERACIONAL

Agents / Servidores

{% for item in result.items %}{% else %}{% endfor %}
Cliente / instalaçãoServidorEmpresaHealthVersãoProdutosCompleteBlockedLast seen
{{ item.client }}{{ item.display_name }}{{ item.hostname or "-" }}{{ item.company_name or item.company_code or "-" }}{{ item.company_uf or "" }} {{ item.company_crt or "" }}{{ item.health }}{{ item.agent_version or "UNKNOWN" }}{{ item.version_drift }}{{ item.total_products }}{{ item.fiscally_complete }}{{ item.safe_blocked }}{{ item.last_seen_at or "Never" }}
Nenhuma instalação encontrada.
{% endblock %} diff --git a/central/templates/base.html b/central/templates/base.html new file mode 100644 index 0000000..7ee7ad2 --- /dev/null +++ b/central/templates/base.html @@ -0,0 +1,8 @@ + + +{% block title %}GTI Fiscal Control Panel{% endblock %} + +
GTI FISCAL{% if auth %}
{% endif %}
+
{% block content %}{% endblock %}
+ + diff --git a/central/templates/blockers.html b/central/templates/blockers.html new file mode 100644 index 0000000..5e00dec --- /dev/null +++ b/central/templates/blockers.html @@ -0,0 +1 @@ +{% extends "base.html" %}{% block title %}Blockers | GTI Fiscal{% endblock %}{% block content %}

FILA FISCAL

Blockers globais

{% for item in result.items %}{% else %}{% endfor %}
ClienteProdutoCampoTipoMotivoÚltima ocorrênciaRetries
{{ item.client or item.tenant_id }}{{ item.product_code }}{{ item.description or "" }}{{ item.field }}{{ item.blocker_type }}{{ item.reason }}{{ item.last_seen_at }}{{ item.retry_count }}
Nenhum blocker aberto.
{% endblock %} diff --git a/central/templates/dashboard.html b/central/templates/dashboard.html new file mode 100644 index 0000000..d9c8354 --- /dev/null +++ b/central/templates/dashboard.html @@ -0,0 +1 @@ +{% extends "base.html" %}{% block title %}Resumo | GTI Fiscal{% endblock %}{% block content %}

OPERAÇÃO CENTRAL

Qual cliente precisa de atenção?

● Atualizado pelo snapshot
{% for key,label in [("agents_total","Agents total"),("online","Online"),("degraded","Degraded"),("offline","Offline"),("failed","Failed"),("safe_blocked","Safe blocked")] %}
{{ label }}{{ summary[key] }}
{% endfor %}

Cobertura fiscal

Ver instalações
{{ summary.total_products }}Produtos
{{ summary.fiscally_complete }}Completos
{{ summary.failed_products }}Falhas técnicas

Principais blockers

Abrir fila
{% for key,value in summary.blockers.items() %}
{{ key }}{{ value }}
{% else %}

Nenhum blocker aberto.

{% endfor %}
{% if summary.stale_agents %}

Heartbeat atrasado

Filtrar Agents
{% for item in summary.stale_agents[:8] %}{{ item.client }} · {{ item.display_name }}{{ item.hostname or "hostname indisponível" }}OFFLINE{% endfor %}
{% endif %}{% endblock %} diff --git a/central/templates/jobs.html b/central/templates/jobs.html new file mode 100644 index 0000000..447246b --- /dev/null +++ b/central/templates/jobs.html @@ -0,0 +1 @@ +{% extends "base.html" %}{% block title %}Jobs | GTI Fiscal{% endblock %}{% block content %}

EXECUÇÕES

Jobs globais

{% for item in result.items %}{% else %}{% endfor %}
JobAgentKindStatusCreatedFinishedError
#{{ item.job_id }}{{ item.agent_id }}{{ item.kind }}{{ item.status }}{{ item.created_at }}{{ item.finished_at or "-" }}{{ item.error or "-" }}
Nenhum job encontrado.
{% endblock %} diff --git a/central/templates/login.html b/central/templates/login.html new file mode 100644 index 0000000..9f3cc0e --- /dev/null +++ b/central/templates/login.html @@ -0,0 +1 @@ +Login | GTI Fiscal

GTI FISCAL / CONTROL PANEL

Entrar no painel

{% if error %}
{{ error }}
{% endif %}
diff --git a/central/tests/test_control_panel_fixture.py b/central/tests/test_control_panel_fixture.py new file mode 100644 index 0000000..ff812bb --- /dev/null +++ b/central/tests/test_control_panel_fixture.py @@ -0,0 +1,14 @@ +from central.tests.test_control_plane import setup_db +from central.scripts.generate_control_panel_fixture import generate_fixture +from central.app.models import AgentEvent, AgentSnapshot, Installation, Job + + +def test_fixture_creates_one_hundred_varied_installations_and_events(): + session_factory = setup_db() + with session_factory() as session: + result = generate_fixture(session, agent_count=100, seed=11) + assert result == {"agents": 100, "events": 500, "blockers": 50, "jobs": 100} + assert session.query(Installation).count() == 100 + assert session.query(AgentEvent).count() == 500 + assert session.query(Job).count() == 100 + assert {row.agent_health for row in session.query(AgentSnapshot).all()} == {"ONLINE", "DEGRADED", "OFFLINE", "FAILED"} diff --git a/central/tests/test_control_plane.py b/central/tests/test_control_plane.py new file mode 100644 index 0000000..61a2146 --- /dev/null +++ b/central/tests/test_control_plane.py @@ -0,0 +1,73 @@ +from datetime import datetime, timedelta, timezone + +from fastapi.testclient import TestClient +from sqlalchemy import create_engine +from sqlalchemy.pool import StaticPool +from sqlalchemy.orm import sessionmaker + +from central.app.human_auth import create_admin, create_session, hash_password, verify_password +from central.app.main import create_app +from central.app.models import Agent, AgentEvent, AgentSnapshot, Base, Installation, Tenant +from central.app.events import purge_expired_events, record_event, sanitize_event +from central.app.installation_identity import compute_agent_health +from types import SimpleNamespace + + +def setup_db(): + engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool) + Base.metadata.create_all(engine) + return sessionmaker(bind=engine, expire_on_commit=False) + + +def seed(session_factory): + with session_factory() as session: + tenant = Tenant(name="Acme"); session.add(tenant); session.flush() + agent = Agent(tenant_id=tenant.id, name="agent", token_hash="x" * 64); session.add(agent); session.flush() + installation = Installation(agent_id=agent.id, tenant_id=tenant.id, installation_id="11111111-1111-4111-8111-111111111111", display_name="Acme ERP", hostname="srv-01", last_seen_at=datetime.now(timezone.utc)); session.add(installation); session.flush() + session.add(AgentSnapshot(installation_id=installation.id, total_products=10, fiscally_complete=8, safe_blocked=2, blockers={"CFOP": 2})) + admin = create_admin(session, "operator", "a-secure-password") + return agent.id, admin.id + + +def test_password_and_event_redaction_and_retention(): + assert verify_password("a-secure-password", hash_password("a-secure-password")) + message, context = sanitize_event("password=bad token=secret", {"password": "bad", "nested": {"api_key": "key"}}) + assert "bad" not in message and context["password"] == "[REDACTED]" and context["nested"]["api_key"] == "[REDACTED]" + session_factory = setup_db() + with session_factory() as session: + agent_id, _ = seed(session_factory) + installation = session.query(Installation).first() + old = record_event(session, tenant_id=installation.tenant_id, installation_id=installation.id, agent_id=agent_id, severity="INFO", event_type="HEARTBEAT", message="old", occurred_at=datetime.now(timezone.utc) - timedelta(days=100)) + keep = record_event(session, tenant_id=installation.tenant_id, installation_id=installation.id, agent_id=agent_id, severity="CRITICAL", event_type="CONNECTION_ERROR", message="keep", occurred_at=datetime.now(timezone.utc) - timedelta(days=100), is_audit=True) + session.commit() + assert purge_expired_events(session, retention_days=90) == 1 + assert session.get(AgentEvent, old.id) is None and session.get(AgentEvent, keep.id) is not None + + +def test_health_model_separates_safe_blocked_from_technical_failure(): + now = datetime.now(timezone.utc) + recent = SimpleNamespace(last_seen_at=now, failed_products=0, last_job_status="COMPLETED") + blocked = SimpleNamespace(last_seen_at=now, failed_products=0, last_job_status="COMPLETED") + failed = SimpleNamespace(last_seen_at=now, failed_products=1, last_job_status="FAILED", agent_health="FAILED") + stale = SimpleNamespace(last_seen_at=now - timedelta(hours=2), failed_products=0, last_job_status="COMPLETED") + assert compute_agent_health(recent)[0] == "ONLINE" + assert compute_agent_health(blocked, supplemental=blocked)[0] == "ONLINE" + assert compute_agent_health(failed)[0] == "FAILED" + assert compute_agent_health(stale)[0] == "OFFLINE" + assert compute_agent_health(SimpleNamespace(last_seen_at=None))[0] == "UNKNOWN" + + +def test_human_login_and_dashboard_are_separate_from_agent_bearer(): + session_factory = setup_db(); seed(session_factory) + client = TestClient(create_app(database_probe=lambda: True, session_factory=session_factory, bootstrap_token=None)) + assert client.get("/", follow_redirects=False).status_code == 303 + response = client.post("/login", data={"username": "operator", "password": "a-secure-password"}, follow_redirects=False) + assert response.status_code == 303 + assert client.get("/v1/dashboard/summary").status_code == 200 + assert TestClient(create_app(database_probe=lambda: True, session_factory=session_factory, bootstrap_token=None)).get("/v1/dashboard/summary", headers={"Authorization": "Bearer x"}).status_code == 401 + assert client.get("/v1/agents").json()["items"][0]["installation_id"] + csrf = client.cookies.get("gti_csrf") + action = client.post("/v1/agents/1/actions/reprocess", json={"confirm": True, "csrf_token": csrf, "kind": "REPROCESS_ALL"}) + assert action.status_code == 200 and action.json()["status"] == "PENDING" + denied = client.post("/v1/agents/1/actions/reprocess", json={"confirm": True, "csrf_token": "wrong", "kind": "REPROCESS_ALL"}) + assert denied.status_code == 403 diff --git a/docs/AI_PROJECT_STATE.md b/docs/AI_PROJECT_STATE.md index 228a017..02ff861 100644 --- a/docs/AI_PROJECT_STATE.md +++ b/docs/AI_PROJECT_STATE.md @@ -9,6 +9,15 @@ after a later successful evaluation. Partial processing keeps heartbeat and new jobs healthy. `AUTONOMOUSLY_PROCESSABLE_PRODUCTS=35066` despite `SAFE_BLOCKED_PRODUCTS=194`; no AI or Firebird write is required for operation. +## Control panel V1 foundation (2026-09-22) + +`CONTROL_PANEL_FOUNDATION_IMPLEMENTED` — the Hub now has persistent installation +identity, enriched metadata, snapshots, deterministic health/version drift, +redacted structured events with retention, separate human admin sessions with +CSRF, paginated control-plane APIs, and Jinja2 operator screens. The panel does +not expose Agent tokens as human sessions and cannot enable AI/Firebird writes, +execute SQL, or run shell commands. + ## Autonomous fiscal completion V1 (2026-09-22) `AUTONOMOUS_FISCAL_COMPLETION_V1_MEASURED` — requiredness matrix and pure diff --git a/docs/CONTROL_PANEL_V1.md b/docs/CONTROL_PANEL_V1.md new file mode 100644 index 0000000..bb9642e --- /dev/null +++ b/docs/CONTROL_PANEL_V1.md @@ -0,0 +1,104 @@ +# GTI Fiscal Control Panel V1 + +## Architecture + +The existing FastAPI Hub is the single control plane. The web interface is +server-rendered with Jinja2, project CSS, and small vanilla JavaScript. There +is no SPA, React/Vue application, Node build chain, raw log upload, or remote +shell path. + +The operational hierarchy is: + +```text +client -> installation -> agent -> company -> Firebird -> health -> coverage -> blockers -> jobs -> events +``` + +## Identity + +Every Agent state database receives one UUID in `installation_identity`. +`installation_id` is sent on heartbeat/registration and remains stable across +restarts. `agent_id` is the Hub Agent record; hostname, display name and IP are +metadata only. A conflicting installation UUID is rejected rather than silently +reassigned. + +Registration metadata includes display name, customer code, hostname, Agent and +Windows version, machine fingerprint, company context, masked CNPJ, Firebird +version/ODS, sanitized database identifier, capabilities, fiscal engine, +dataset versions and lifecycle times. Passwords, tokens, API keys and full +connection strings are never persisted. + +## Authentication + +Agents use the existing `Authorization: Bearer ` flow. Human users +use `/login` and a separate `gti_admin_session` HttpOnly cookie backed by a +server-side session hash. The CSRF value is separate and checked for every +mutating human action. Create the first admin locally with: + +```text +python -m central.scripts.create_admin +``` + +Passwords are PBKDF2-HMAC-SHA256 hashes with random salts. Agent tokens are not +accepted as web sessions. + +## Health And Snapshots + +The Hub stores one `agent_snapshots` row per installation, containing product +counts, completion, safe blockers, technical failures, coverage, blocker +aggregates, job timestamps, health and version drift. The dashboard reads this +snapshot instead of scanning product tables. + +- `ONLINE`: recent heartbeat and no technical failure. +- `DEGRADED`: recent technical/job failure while processing remains possible. +- `OFFLINE`: heartbeat older than `AGENT_OFFLINE_AFTER_SECONDS`, or three + heartbeat intervals when that value is unset. +- `FAILED`: persistent technical failure that prevents processing. +- `UNKNOWN`: no usable heartbeat or snapshot. + +`SAFE_BLOCKED` is fiscal work isolation and does not itself make an Agent +degraded or failed. Version drift is `CURRENT`, `OUTDATED` or `UNKNOWN` against +`CURRENT_SERVER_VERSION`. + +## Events And Retention + +Agents submit structured events at `POST /v1/agents/events`. Event types are +allowlisted and messages/nested JSON are redacted and size-limited before +storage. Raw log files are never uploaded. + +The default retention is 90 days (`AGENT_EVENT_RETENTION_DAYS`). Run +`python -m central.scripts.maintain_events` for cleanup, or use `--dry-run` to +inspect the command without deletion. Audit events, blocker history, critical +events and write audit records are preserved. + +## APIs And Screens + +Agent routes remain Bearer authenticated: `POST /v1/agents/register`, +`POST /v1/agents/heartbeat`, and `POST /v1/agents/events`. + +Human panel APIs provide `/v1/dashboard/summary`, `/v1/agents`, +`/v1/agents/{agent_id}` plus health, coverage, blockers, jobs and events, +global `/v1/blockers/global` and `/v1/jobs/global`, and confirmed reprocess or +blocker-resolution actions. Lists use bounded pagination and filters. + +The UI pages are `/`, `/agents`, `/agents/{agent_id}`, `/blockers`, `/jobs` and +`/login`. Primary rows show client, display name and hostname; UUIDs remain in +technical detail. + +## Safe Actions + +Human actions require a session, CSRF, explicit confirmation and an +`audit_events` row. The panel can create reprocess jobs and resolve blockers +only with explicit context. It cannot enable Firebird/AI writes, alter +credentials/path/CRT, execute SQL or shell commands, restart Windows services, +or update Agents automatically. + +## Fixture + +The explicit non-production fixture command is: + +```text +python -m central.scripts.generate_control_panel_fixture --agents 100 --confirm-non-production +``` + +It creates varied health states, snapshots, blockers, jobs and events, and is +never invoked at startup or migration time.