diff --git a/central/alembic/versions/0018_blocker_installation_identity.py b/central/alembic/versions/0018_blocker_installation_identity.py new file mode 100644 index 0000000..5fcec5e --- /dev/null +++ b/central/alembic/versions/0018_blocker_installation_identity.py @@ -0,0 +1,20 @@ +"""Associate new blocker rows with their authenticated installation.""" + +from alembic import op +import sqlalchemy as sa + + +revision = "0018_blocker_installation_identity" +down_revision = "0017_control_plane_foundation" +branch_labels = None +depends_on = None + + +def upgrade(): + op.add_column("blocked_products", sa.Column("installation_id", sa.Integer(), sa.ForeignKey("installations.id"), nullable=True)) + op.create_index("ix_blocked_products_installation_id", "blocked_products", ["installation_id"]) + + +def downgrade(): + op.drop_index("ix_blocked_products_installation_id", table_name="blocked_products") + op.drop_column("blocked_products", "installation_id") diff --git a/central/app/api.py b/central/app/api.py index c1283b3..df1a181 100644 --- a/central/app/api.py +++ b/central/app/api.py @@ -245,6 +245,7 @@ def router(session_factory, bootstrap_token): def _blocked_view(row): return { "id": row.id, "tenant_id": row.tenant_id, "company": row.company, "product_code": row.product_code, + "installation_db_id": row.installation_id, "gtin": row.gtin, "description": row.description, "blocker_type": row.blocker_type, "field": row.field, "reason_code": row.reason_code, "evidence_summary": row.evidence_summary, "first_seen_at": row.first_seen_at.isoformat() if row.first_seen_at else None, "last_seen_at": row.last_seen_at.isoformat() if row.last_seen_at else None, "retry_count": row.retry_count, @@ -256,7 +257,8 @@ def router(session_factory, bootstrap_token): def blockers_list(blocker_type: str | None = None, field: str | None = None, identity=Depends(principal)): session, agent = identity try: - return {"blockers": [_blocked_view(row) for row in list_blockers(session, agent.tenant_id, blocker_type=blocker_type, field=field)]} + installation = session.scalar(select(Installation).where(Installation.agent_id == agent.id)) + return {"blockers": [_blocked_view(row) for row in list_blockers(session, agent.tenant_id, blocker_type=blocker_type, field=field, installation_id=installation.id if installation else None)]} finally: session.close() @@ -264,10 +266,11 @@ def router(session_factory, bootstrap_token): def blockers_export(identity=Depends(principal)): session, agent = identity try: + installation = session.scalar(select(Installation).where(Installation.agent_id == agent.id)) output = io.StringIO() writer = csv.writer(output) writer.writerow(["id", "company", "product_code", "gtin", "blocker_type", "field", "reason_code", "retry_count", "first_seen_at", "last_seen_at"]) - for row in list_blockers(session, agent.tenant_id): + for row in list_blockers(session, agent.tenant_id, installation_id=installation.id if installation else None): writer.writerow([row.id, row.company, row.product_code, row.gtin or "", row.blocker_type, row.field, row.reason_code, row.retry_count, row.first_seen_at.isoformat() if row.first_seen_at else "", row.last_seen_at.isoformat() if row.last_seen_at else ""]) return Response(output.getvalue(), media_type="text/csv", headers={"Content-Disposition": "attachment; filename=blocked-products.csv"}) finally: @@ -277,7 +280,8 @@ def router(session_factory, bootstrap_token): def blockers_reprocess(payload: dict, identity=Depends(principal)): session, agent = identity try: - rows = list_blockers(session, agent.tenant_id, blocker_type=payload.get("blocker_type"), field=payload.get("field")) + installation = session.scalar(select(Installation).where(Installation.agent_id == agent.id)) + rows = list_blockers(session, agent.tenant_id, blocker_type=payload.get("blocker_type"), field=payload.get("field"), installation_id=installation.id if installation else None) codes = [row.product_code for row in rows if row.product_code] if not codes: return {"created": False, "reason": "NO_DUE_BLOCKERS"} @@ -358,10 +362,11 @@ def router(session_factory, bootstrap_token): job = session.get(Job, job_id) if job is None or job.agent_id != agent.id: raise HTTPException(404, "job not found") result_summary = payload.get("result_summary", {}) + installation = session.scalar(select(Installation).where(Installation.agent_id == agent.id)) for blocker in result_summary.get("blocked_products", []): - upsert_blocker(session, agent.tenant_id, blocker) + upsert_blocker(session, agent.tenant_id, blocker, installation_id=installation.id if installation else None) for product_code in result_summary.get("resolved_products", []): - resolve_product_blockers(session, agent.tenant_id, product_code) + resolve_product_blockers(session, agent.tenant_id, product_code, installation_id=installation.id if installation else None) if not complete_job(session, job, result_summary): raise HTTPException(409, "job is not RUNNING") logger.info("job completed job_id=%s agent_id=%s result=%s", job_id, agent.id, payload.get("result_summary", {})) return _job_view(session.get(Job, job_id)) diff --git a/central/app/blocked_products.py b/central/app/blocked_products.py index 3c3e8b7..ba60a60 100644 --- a/central/app/blocked_products.py +++ b/central/app/blocked_products.py @@ -7,24 +7,28 @@ from .models import BlockedProduct from .operational_state import blocker_fingerprint -def upsert_blocker(session, tenant_id: int, payload: dict) -> BlockedProduct: +def upsert_blocker(session, tenant_id: int, payload: dict, *, installation_id: int | None = None) -> BlockedProduct: fingerprint = payload.get("blocker_fingerprint") or blocker_fingerprint(payload.get("tenant", str(tenant_id)), payload.get("company", ""), payload.get("product_code", ""), payload.get("field", ""), payload.get("reason_code", ""), payload.get("input_fingerprint", "")) row = session.scalar(select(BlockedProduct).where(BlockedProduct.blocker_fingerprint == fingerprint)) now = datetime.now(timezone.utc) if row is None: - row = BlockedProduct(tenant_id=tenant_id, company=payload.get("company", ""), product_code=str(payload.get("product_code", "")), gtin=payload.get("gtin"), description=payload.get("description"), blocker_type=payload.get("blocker_type", "BLOCKED_MISSING_EVIDENCE"), field=payload.get("field", ""), reason_code=payload.get("reason_code", ""), evidence_summary=payload.get("evidence_summary") or {}, input_fingerprint=payload.get("input_fingerprint", ""), blocker_fingerprint=fingerprint) + row = BlockedProduct(tenant_id=tenant_id, installation_id=installation_id, company=payload.get("company", ""), product_code=str(payload.get("product_code", "")), gtin=payload.get("gtin"), description=payload.get("description"), blocker_type=payload.get("blocker_type", "BLOCKED_MISSING_EVIDENCE"), field=payload.get("field", ""), reason_code=payload.get("reason_code", ""), evidence_summary=payload.get("evidence_summary") or {}, input_fingerprint=payload.get("input_fingerprint", ""), blocker_fingerprint=fingerprint) session.add(row) else: row.last_seen_at = now row.evidence_summary = payload.get("evidence_summary") or row.evidence_summary row.input_fingerprint = payload.get("input_fingerprint", row.input_fingerprint) + if row.installation_id is None and installation_id is not None: + row.installation_id = installation_id row.resolved_at = None session.commit() return row -def list_blockers(session, tenant_id: int, *, input_fingerprint: str | None = None, blocker_type: str | None = None, field: str | None = None): +def list_blockers(session, tenant_id: int, *, input_fingerprint: str | None = None, blocker_type: str | None = None, field: str | None = None, installation_id: int | None = None): query = select(BlockedProduct).where(BlockedProduct.tenant_id == tenant_id, BlockedProduct.resolved_at.is_(None)) + if installation_id is not None: + query = query.where((BlockedProduct.installation_id == installation_id) | BlockedProduct.installation_id.is_(None)) if input_fingerprint is not None: query = query.where(BlockedProduct.input_fingerprint != input_fingerprint) if blocker_type: @@ -51,8 +55,11 @@ def resolve_blocker(session, blocker_id: int, source: str) -> bool: return True -def resolve_product_blockers(session, tenant_id: int, product_code: str, source: str = "AUTO_REEVALUATION") -> int: - rows = session.scalars(select(BlockedProduct).where(BlockedProduct.tenant_id == tenant_id, BlockedProduct.product_code == str(product_code), BlockedProduct.resolved_at.is_(None))).all() +def resolve_product_blockers(session, tenant_id: int, product_code: str, source: str = "AUTO_REEVALUATION", *, installation_id: int | None = None) -> int: + query = select(BlockedProduct).where(BlockedProduct.tenant_id == tenant_id, BlockedProduct.product_code == str(product_code), BlockedProduct.resolved_at.is_(None)) + if installation_id is not None: + query = query.where((BlockedProduct.installation_id == installation_id) | BlockedProduct.installation_id.is_(None)) + rows = session.scalars(query).all() now = datetime.now(timezone.utc) for row in rows: row.resolved_at = now diff --git a/central/app/control_api.py b/central/app/control_api.py index 0834c13..b5df1e2 100644 --- a/central/app/control_api.py +++ b/central/app/control_api.py @@ -62,7 +62,7 @@ def router(session_factory): 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) + return list_blockers(session, tenant_id=installation.tenant_id, agent_id=agent_id, page=page, page_size=page_size, blocker_type=blocker_type, field=field) finally: session.close() @routes.get("/agents/{agent_id}/jobs") diff --git a/central/app/control_plane.py b/central/app/control_plane.py index df42839..be65a74 100644 --- a/central/app/control_plane.py +++ b/central/app/control_plane.py @@ -4,9 +4,9 @@ from __future__ import annotations from datetime import datetime, timedelta, timezone -from sqlalchemy import exists, func, or_, select -from sqlalchemy.orm import aliased +from sqlalchemy import and_, case, func, or_, select +from .config import settings from .events import list_events from .installation_identity import compute_agent_health, compute_version_drift from .models import Agent, AgentEvent, AgentSnapshot, AuditEvent, BlockedProduct, Installation, Job, Tenant @@ -15,6 +15,7 @@ from .panel_presentation import format_age, format_duration, format_ui_datetime, HEALTH_STATES = ("ONLINE", "DEGRADED", "OFFLINE", "FAILED", "UNKNOWN") BLOCKER_FIELDS = ("CFOP", "ST", "ICMS", "CSOSN", "NCM", "OTHER") +ATTENTION_CANDIDATE_LIMIT = 10 def _page(page: int, page_size: int): @@ -25,12 +26,36 @@ def _iso(value): return value.isoformat() if value else None +def _utc_now(value=None): + value = value or datetime.now(timezone.utc) + return value.replace(tzinfo=timezone.utc) if value.tzinfo is None else value + + +def _health_expression(now): + offline_after = settings.agent_offline_after_seconds or settings.agent_heartbeat_interval_seconds * 3 + cutoff = _utc_now(now) - timedelta(seconds=offline_after) + return case( + (Installation.last_seen_at.is_(None), "UNKNOWN"), + (or_(AgentSnapshot.agent_health == "FAILED", Installation.last_status == "FAILED"), "FAILED"), + (Installation.last_seen_at < cutoff, "OFFLINE"), + (or_(AgentSnapshot.failed_products > 0, AgentSnapshot.last_job_status == "FAILED"), "DEGRADED"), + else_="ONLINE", + ) + + +def _fiscal_completion_expression(): + total = func.coalesce(AgentSnapshot.total_products, 0) + complete = func.coalesce(AgentSnapshot.fiscally_complete, 0) + raw = case((total > 0, complete * 100.0 / total), else_=func.coalesce(AgentSnapshot.coverage_percent, 0)) + return func.round(case((raw < 0, 0), (raw > 100, 100), else_=raw), 1) + + def _fiscal_completion(snapshot) -> float: total = int(getattr(snapshot, "total_products", 0) or 0) complete = int(getattr(snapshot, "fiscally_complete", 0) or 0) if total: return round(max(0, min(total, complete)) * 100 / total, 1) - return round(float(getattr(snapshot, "coverage_percent", 0) or 0), 1) + return round(max(0, min(100, float(getattr(snapshot, "coverage_percent", 0) or 0))), 1) def _fiscal_status(snapshot) -> str: @@ -134,21 +159,20 @@ def _installation_query_filters(query, *, client=None, version=None, search=None def list_installations(session, *, page=1, page_size=50, client=None, health=None, version=None, search=None, last_seen_before=None, sort="display_name", now=None): page, page_size = _page(page, page_size) + now = _utc_now(now) 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) query = _installation_query_filters(query, client=client, version=version, search=search, last_seen_before=last_seen_before) + health_expression = _health_expression(now) + if health: + query = query.where(health_expression == health) if sort in {"worst_coverage", "coverage", "coverage_asc", "fiscal_completion_percent"}: - ordered = query.order_by(func.coalesce(AgentSnapshot.coverage_percent, 0).asc(), Installation.display_name.asc(), Installation.id.asc()) + ordered = query.order_by(_fiscal_completion_expression().asc(), Installation.display_name.asc(), Installation.id.asc()) else: 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() + rows = session.execute(ordered.offset((page - 1) * page_size).limit(page_size)).all() items = [installation_view(i, a, t, s, now=now) for i, a, t, s in rows] - if health: - items = [item for item in items if item["health"] == health] - total = len(items) - items = items[(page - 1) * page_size:page * page_size] - else: - count_query = select(func.count()).select_from(query.with_only_columns(Installation.id).order_by(None).subquery()) - total = session.scalar(count_query) or 0 + count_query = select(func.count()).select_from(query.with_only_columns(Installation.id).order_by(None).subquery()) + total = session.scalar(count_query) or 0 return {"page": page, "page_size": page_size, "items": items, "total": total, "health_filter_applied": health} @@ -199,9 +223,7 @@ def _parse_age(value): def _blockers_query(session, *, tenant_id=None, agent_id=None, blocker_type=None, field=None, status="OPEN", client=None, installation=None, search=None, age=None): - installation_filter = aliased(Installation) - representative = select(func.min(Installation.id).label("installation_db_id"), Installation.tenant_id.label("installation_tenant_id")).group_by(Installation.tenant_id).subquery() - query = select(BlockedProduct, Tenant, Installation, Agent).join(Tenant, Tenant.id == BlockedProduct.tenant_id).outerjoin(representative, representative.c.installation_tenant_id == BlockedProduct.tenant_id).outerjoin(Installation, Installation.id == representative.c.installation_db_id).outerjoin(Agent, Agent.id == Installation.agent_id) + query = select(BlockedProduct, Tenant, Installation, Agent).join(Tenant, Tenant.id == BlockedProduct.tenant_id).outerjoin(Installation, Installation.id == BlockedProduct.installation_id).outerjoin(Agent, Agent.id == Installation.agent_id) status = (status or "OPEN").upper() if status == "OPEN": query = query.where(BlockedProduct.resolved_at.is_(None)) @@ -210,7 +232,7 @@ def _blockers_query(session, *, tenant_id=None, agent_id=None, blocker_type=None if tenant_id is not None: query = query.where(BlockedProduct.tenant_id == tenant_id) if agent_id is not None: - query = query.where(exists(select(installation_filter.id).where(installation_filter.tenant_id == BlockedProduct.tenant_id, installation_filter.agent_id == agent_id))) + query = query.where(Agent.id == agent_id) if blocker_type: query = query.where(BlockedProduct.blocker_type == blocker_type) if field: @@ -218,18 +240,7 @@ def _blockers_query(session, *, tenant_id=None, agent_id=None, blocker_type=None if client: query = query.where(or_(Tenant.name.ilike(f"%{client}%"), BlockedProduct.company.ilike(f"%{client}%"))) if installation: - query = query.where( - exists( - select(installation_filter.id).where( - installation_filter.tenant_id == BlockedProduct.tenant_id, - or_( - installation_filter.installation_id.ilike(f"%{installation}%"), - installation_filter.display_name.ilike(f"%{installation}%"), - installation_filter.hostname.ilike(f"%{installation}%"), - ), - ) - ) - ) + query = query.where(or_(Installation.installation_id.ilike(f"%{installation}%"), Installation.display_name.ilike(f"%{installation}%"), Installation.hostname.ilike(f"%{installation}%"))) if search: term = f"%{search}%" query = query.where(or_(BlockedProduct.product_code.ilike(term), BlockedProduct.description.ilike(term), BlockedProduct.field.ilike(term), BlockedProduct.blocker_type.ilike(term), BlockedProduct.reason_code.ilike(term), Tenant.name.ilike(term))) @@ -300,50 +311,80 @@ def list_jobs(session, *, agent_id=None, page=1, page_size=50, status=None, kind def dashboard_summary(session, now=None): - now = now or datetime.now(timezone.utc) - 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, now=now) for i, a, t, s in session.execute(query).all()] - counts = {state: sum(item["technical_health"] == state for item in items) for state in HEALTH_STATES} - 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) + now = _utc_now(now) + installation_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) + health_expression = _health_expression(now) + health_rows = session.execute( + select(health_expression.label("health"), func.count(Installation.id)) + .select_from(Installation) + .join(Agent, Agent.id == Installation.agent_id) + .join(Tenant, Tenant.id == Installation.tenant_id) + .outerjoin(AgentSnapshot, AgentSnapshot.installation_id == Installation.id) + .group_by(health_expression) + ).all() + counts = {state: 0 for state in HEALTH_STATES} + for state, count in health_rows: + counts[state] = int(count) - blocker_rows = session.scalars(select(BlockedProduct).where(BlockedProduct.resolved_at.is_(None))).all() + total_row = session.execute( + select( + func.coalesce(func.sum(AgentSnapshot.total_products), 0), + func.coalesce(func.sum(AgentSnapshot.fiscally_complete), 0), + func.coalesce(func.sum(AgentSnapshot.safe_blocked), 0), + func.coalesce(func.sum(AgentSnapshot.failed_products), 0), + ).select_from(Installation).outerjoin(AgentSnapshot, AgentSnapshot.installation_id == Installation.id) + ).one() + totals = dict(zip(("total_products", "fiscally_complete", "safe_blocked", "failed_products"), (int(value or 0) for value in total_row))) + + open_filter = BlockedProduct.resolved_at.is_(None) + open_blockers = session.scalar(select(func.count(BlockedProduct.id)).where(open_filter)) or 0 + breakdown_expression = func.coalesce(BlockedProduct.field, "OTHER") + breakdown_rows = session.execute(select(breakdown_expression, func.count(BlockedProduct.id)).where(open_filter).group_by(breakdown_expression)).all() open_blocker_breakdown = {field: 0 for field in BLOCKER_FIELDS} - product_pairs = set() - for row in blocker_rows: - product_pairs.add((row.tenant_id, row.product_code)) - key = str(row.field or "").upper() - open_blocker_breakdown[key if key in open_blocker_breakdown else "OTHER"] += 1 + for field, count in breakdown_rows: + key = str(field or "").upper() + open_blocker_breakdown[key if key in open_blocker_breakdown else "OTHER"] += int(count) + product_pairs = select(BlockedProduct.tenant_id, BlockedProduct.product_code).where(open_filter).distinct().subquery() + safe_blocked_products = session.scalar(select(func.count()).select_from(product_pairs)) or 0 + blocker_counts = {field: count for field, count in open_blocker_breakdown.items() if count} + + def installation_candidates(health): + query = installation_query.where(health_expression == health).order_by(Installation.id.asc()).limit(ATTENTION_CANDIDATE_LIMIT) + return [installation_view(i, a, t, s, now=now) for i, a, t, s in session.execute(query).all()] + + health_candidates = {health: installation_candidates(health) for health in ("FAILED", "DEGRADED", "OFFLINE", "UNKNOWN")} + version_filter = or_(AgentSnapshot.version_drift == "OUTDATED", and_(Agent.agent_version.is_not(None), Agent.agent_version != settings.agent_server_version)) + outdated_query = installation_query.where(version_filter).order_by(Installation.id.asc()).limit(ATTENTION_CANDIDATE_LIMIT) + outdated_agents = [installation_view(i, a, t, s, now=now) for i, a, t, s in session.execute(outdated_query).all()] + + blocker_query = select(BlockedProduct, Tenant, Installation, Agent, AgentSnapshot).join(Tenant, Tenant.id == BlockedProduct.tenant_id).outerjoin(Installation, Installation.id == BlockedProduct.installation_id).outerjoin(Agent, Agent.id == Installation.agent_id).outerjoin(AgentSnapshot, AgentSnapshot.installation_id == Installation.id).where(open_filter, BlockedProduct.field.in_(BLOCKER_FIELDS[:-1])).order_by(BlockedProduct.last_seen_at.desc(), BlockedProduct.id.desc()).limit(ATTENTION_CANDIDATE_LIMIT) + blocker_candidates = session.execute(blocker_query).all() attention = [] - for item in items: - if item["technical_health"] in {"FAILED", "DEGRADED", "OFFLINE", "UNKNOWN"}: - rank = {"FAILED": 0, "DEGRADED": 1, "OFFLINE": 2, "UNKNOWN": 3}[item["technical_health"]] - attention.append((_attention_item(item, item["health_reason"], _technical_timestamp(item), rank=rank, now=now), rank, item["agent_id"])) - item_by_tenant = {item["tenant_id"]: item for item in items} - for row in blocker_rows: - item = item_by_tenant.get(row.tenant_id) - if item and (row.field or "").upper() in BLOCKER_FIELDS[:-1]: - attention.append((_attention_item(item, f"{row.field or row.blocker_type}: {row.reason_code}", row.last_seen_at, rank=4, link=f"/blockers/{row.id}", now=now), 4, row.id)) - for item in items: - if item["version_drift"] == "OUTDATED": - attention.append((_attention_item(item, "Versão do agent desatualizada", item.get("last_seen_at"), rank=5, now=now), 5, item["agent_id"])) + for health in ("FAILED", "DEGRADED", "OFFLINE", "UNKNOWN"): + rank = {"FAILED": 0, "DEGRADED": 1, "OFFLINE": 2, "UNKNOWN": 3}[health] + attention.extend((_attention_item(item, item["health_reason"], _technical_timestamp(item), rank=rank, now=now), rank, item["agent_id"]) for item in health_candidates[health]) + for blocker, tenant, installation, agent, snapshot in blocker_candidates: + if installation and agent: + item = installation_view(installation, agent, tenant, snapshot, now=now) + else: + item = {"client": tenant.name, "display_name": None, "hostname": None, "agent_id": None, "technical_health": "UNKNOWN", "health_reason": "tenant-level blocker", "last_seen_at": None, "last_job_at": None} + attention.append((_attention_item(item, f"{blocker.field or blocker.blocker_type}: {blocker.reason_code}", blocker.last_seen_at, rank=4, link=f"/blockers/{blocker.id}", now=now), 4, blocker.id)) + for item in outdated_agents: + attention.append((_attention_item(item, "Versão do agent desatualizada", item.get("last_seen_at"), rank=5, now=now), 5, item["agent_id"])) attention.sort(key=lambda entry: (entry[1], -(entry[0]["timestamp"] is not None), entry[2])) failed_jobs = session.scalar(select(func.count(Job.id)).where(Job.status == "FAILED")) or 0 return { - "agents_total": len(items), + "agents_total": sum(counts.values()), **{key.lower(): value for key, value in counts.items()}, "health_counts": counts, **totals, "blockers": blocker_counts, - "stale_agents": [item for item in items if item["technical_health"] == "OFFLINE"], - "outdated_agents": [item for item in items if item["version_drift"] == "OUTDATED"], - "safe_blocked_products": len(product_pairs), - "open_blockers": len(blocker_rows), + "stale_agents": health_candidates["OFFLINE"], + "outdated_agents": outdated_agents, + "safe_blocked_products": int(safe_blocked_products), + "open_blockers": int(open_blockers), "open_blocker_breakdown": open_blocker_breakdown, "attention_items": [entry[0] for entry in attention[:10]], "failed_jobs": failed_jobs, diff --git a/central/app/models.py b/central/app/models.py index 07291e3..8bf1f4d 100644 --- a/central/app/models.py +++ b/central/app/models.py @@ -359,6 +359,7 @@ class BlockedProduct(Base): __table_args__ = (UniqueConstraint("blocker_fingerprint", name="uq_blocked_product_fingerprint"),) id: Mapped[int] = mapped_column(primary_key=True) tenant_id: Mapped[int] = mapped_column(ForeignKey("tenants.id"), index=True) + installation_id: Mapped[int | None] = mapped_column(ForeignKey("installations.id"), nullable=True, index=True) company: Mapped[str] = mapped_column(String(128), index=True) product_code: Mapped[str] = mapped_column(String(64), index=True) gtin: Mapped[str | None] = mapped_column(String(16), nullable=True, index=True) diff --git a/central/app/web.py b/central/app/web.py index 687b0a1..1d4c0b4 100644 --- a/central/app/web.py +++ b/central/app/web.py @@ -88,7 +88,7 @@ def router(session_factory): 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", "")}) + return render(request, "agent_detail.html", {"detail": detail, "tab": tab, "blockers": list_blockers(session, tenant_id=installation.tenant_id, agent_id=agent_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") diff --git a/central/scripts/generate_control_panel_fixture.py b/central/scripts/generate_control_panel_fixture.py index 1c238d9..50e8b89 100644 --- a/central/scripts/generate_control_panel_fixture.py +++ b/central/scripts/generate_control_panel_fixture.py @@ -31,7 +31,7 @@ def generate_fixture(session, agent_count=100, seed=7): 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}")) + session.add(BlockedProduct(tenant_id=tenant.id, installation_id=installation.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 diff --git a/central/tests/test_jobs.py b/central/tests/test_jobs.py index d878d70..f2e9b34 100644 --- a/central/tests/test_jobs.py +++ b/central/tests/test_jobs.py @@ -5,7 +5,7 @@ from sqlalchemy.pool import StaticPool from central.app.jobs import claim_next_job, complete_job, create_job, fail_job from central.app.main import create_app -from central.app.models import Agent, Base, Job, Tenant +from central.app.models import Agent, Base, BlockedProduct, Job, Tenant def _make_agent(session): @@ -133,8 +133,9 @@ def test_reprocess_products_requires_codigos(): def test_blocker_registry_api_deduplicates_and_resolves(): - client, _ = _client() + client, engine = _client() headers = {"Authorization": "Bearer test-token"} + assert client.post("/v1/agents/heartbeat", json={"status": "IDLE", "installation_id": "00000000-0000-4000-8000-000000000001"}, headers=headers).status_code == 200 summary = {"blocked_products": [{"company": "C", "product_code": "P1", "blocker_type": "BLOCKED_MISSING_EVIDENCE", "field": "CFOP", "reason_code": "MISSING", "input_fingerprint": "v1", "evidence_summary": {"source": "ERP"}}]} created = client.post("/v1/jobs", json={"kind": "REPROCESS_PRODUCTS", "payload": {"codigos": ["P1"]}}, headers=headers) job_id = created.json()["job_id"] @@ -143,6 +144,8 @@ def test_blocker_registry_api_deduplicates_and_resolves(): listed = client.get("/v1/blockers?field=CFOP", headers=headers) assert listed.status_code == 200 assert len(listed.json()["blockers"]) == 1 + with Session(engine) as session: + assert session.query(BlockedProduct).one().installation_id is not None blocker_id = listed.json()["blockers"][0]["id"] resolved = client.post(f"/v1/blockers/{blocker_id}/resolve", json={"resolution_source": "TEST"}, headers=headers) assert resolved.json()["resolved"] is True diff --git a/central/tests/test_operational_view_models.py b/central/tests/test_operational_view_models.py index 7a591bb..907dadd 100644 --- a/central/tests/test_operational_view_models.py +++ b/central/tests/test_operational_view_models.py @@ -1,6 +1,9 @@ from datetime import datetime, timedelta, timezone +from sqlalchemy import event + from central.app.audit import record_audit +from central.app.blocked_products import upsert_blocker from central.app.control_plane import dashboard_summary, list_blockers, list_installations, list_jobs from central.app.events import record_event from central.app.models import Agent, AgentSnapshot, AuditEvent, BlockedProduct, Installation, Job, Tenant @@ -51,9 +54,10 @@ def _installation(session, index, *, health="ONLINE", coverage=80, blocked=0, ve return tenant, agent, installation, snapshot -def _blocker(session, tenant, *, product, field="CFOP", blocker_type="BLOCKED_MISSING_EVIDENCE", index=0): +def _blocker(session, tenant, *, product, field="CFOP", blocker_type="BLOCKED_MISSING_EVIDENCE", index=0, installation_id=None): row = BlockedProduct( tenant_id=tenant.id, + installation_id=installation_id, company=tenant.name, product_code=product, description=f"Product {product}", @@ -153,8 +157,8 @@ def test_blockers_filter_with_priority_age_and_stable_pagination(): session_factory = setup_db() with session_factory() as session: tenant, _, installation, _ = _installation(session, 1) - critical = _blocker(session, tenant, product="P1", field="", blocker_type="TECHNICAL_FAILURE", index=1) - _blocker(session, tenant, product="P2", field="CFOP", index=2) + critical = _blocker(session, tenant, product="P1", field="", blocker_type="TECHNICAL_FAILURE", index=1, installation_id=installation.id) + _blocker(session, tenant, product="P2", field="CFOP", index=2, installation_id=installation.id) session.commit() result = list_blockers(session, client="Client 1", installation=installation.display_name, status="OPEN", search="P1", page=1, page_size=1) @@ -170,7 +174,7 @@ def test_blocker_detail_contains_context_events_jobs_and_resolution_audit(): session_factory = setup_db() with session_factory() as session: tenant, agent, installation, _ = _installation(session, 1) - blocker = _blocker(session, tenant, product="P1", index=1) + blocker = _blocker(session, tenant, product="P1", index=1, installation_id=installation.id) job = Job(agent_id=agent.id, kind="REPROCESS_ALL", status="FAILED", created_at=NOW, error="failed") session.add(job) session.flush() @@ -188,3 +192,91 @@ def test_blocker_detail_contains_context_events_jobs_and_resolution_audit(): assert detail["related_events"][0]["job_id"] == job.id assert detail["related_jobs"][0]["job_id"] == job.id assert detail["resolution_audit"][0]["action"] == "RESOLVE_BLOCKER" + + +def test_blockers_keep_exact_installation_identity_for_same_tenant(): + session_factory = setup_db() + with session_factory() as session: + tenant = Tenant(name="Shared Client") + session.add(tenant) + session.flush() + agent_a = Agent(tenant_id=tenant.id, name="agent-a", token_hash="token-a".ljust(64, "x")) + agent_b = Agent(tenant_id=tenant.id, name="agent-b", token_hash="token-b".ljust(64, "x")) + session.add_all([agent_a, agent_b]) + session.flush() + installation_a = Installation(agent_id=agent_a.id, tenant_id=tenant.id, installation_id="00000000-0000-4000-8000-000000000101", display_name="Store A", hostname="server-a") + installation_b = Installation(agent_id=agent_b.id, tenant_id=tenant.id, installation_id="00000000-0000-4000-8000-000000000102", display_name="Store B", hostname="server-b") + session.add_all([installation_a, installation_b]) + session.flush() + row_a = upsert_blocker(session, tenant.id, {"company": tenant.name, "product_code": "P-A", "field": "CFOP", "reason_code": "MISSING", "input_fingerprint": "a"}, installation_id=installation_a.id) + row_b = upsert_blocker(session, tenant.id, {"company": tenant.name, "product_code": "P-B", "field": "CFOP", "reason_code": "MISSING", "input_fingerprint": "b"}, installation_id=installation_b.id) + + result_a = list_blockers(session, tenant_id=tenant.id, installation="Store A") + result_b = list_blockers(session, tenant_id=tenant.id, installation="Store B") + + assert result_a["total"] == 1 and result_a["items"][0]["id"] == row_a.id + assert result_a["items"][0]["installation_id"] == installation_a.installation_id + assert result_b["total"] == 1 and result_b["items"][0]["id"] == row_b.id + assert result_b["items"][0]["installation_id"] == installation_b.installation_id + + +def test_legacy_tenant_blocker_has_no_fabricated_installation_context(): + session_factory = setup_db() + with session_factory() as session: + tenant, _, installation, _ = _installation(session, 1) + row = _blocker(session, tenant, product="LEGACY", index=99) + result = list_blockers(session, tenant_id=tenant.id) + + assert result["items"][0]["id"] == row.id + assert result["items"][0]["installation_id"] is None + from central.app.control_plane import blocker_detail + + detail = blocker_detail(session, row.id) + assert detail["installation"]["installation_id"] is None + assert detail["server"]["hostname"] is None + assert installation.hostname == "server-1" + + +def test_dashboard_uses_sql_aggregates_for_operational_totals(): + session_factory = setup_db() + with session_factory() as session: + tenant, _, _, _ = _installation(session, 1) + _blocker(session, tenant, product="P1", index=101) + statements = [] + event.listen(session.bind, "before_cursor_execute", lambda conn, cursor, statement, parameters, context, executemany: statements.append(statement)) + + dashboard_summary(session, now=NOW) + + assert any("COUNT" in statement.upper() and "BLOCKED_PRODUCTS" in statement.upper() for statement in statements) + assert any("DISTINCT" in statement.upper() and "BLOCKED_PRODUCTS" in statement.upper() for statement in statements) + + +def test_health_filter_is_paginated_in_sql_and_counts_all_matches(): + session_factory = setup_db() + with session_factory() as session: + for index in range(1, 4): + _installation(session, index, health="ONLINE") + statements = [] + event.listen(session.bind, "before_cursor_execute", lambda conn, cursor, statement, parameters, context, executemany: statements.append(statement)) + + result = list_installations(session, health="ONLINE", page=2, page_size=1, now=NOW) + + assert result["total"] == 3 + assert len(result["items"]) == 1 + assert any("LIMIT" in statement.upper() and "INSTALLATIONS" in statement.upper() for statement in statements) + + +def test_worst_coverage_sort_uses_the_displayed_derived_percentage(): + session_factory = setup_db() + with session_factory() as session: + _, _, first, first_snapshot = _installation(session, 1, coverage=80) + _, _, second, second_snapshot = _installation(session, 2, coverage=20) + first_snapshot.coverage_percent = 10 + second_snapshot.coverage_percent = 90 + session.commit() + + result = list_installations(session, sort="worst_coverage", page_size=10, now=NOW) + + assert result["items"][0]["installation_id"] == second.installation_id + assert result["items"][0]["fiscal_completion_percent"] == 20 + assert first.installation_id != second.installation_id