feat: close autonomous operations for blocked products

central/app/api.py

@@ -1,15 +1,18 @@
 """Authenticated version-one Hub routes."""
 import hashlib, json, logging
 from datetime import datetime, timezone
-from fastapi import APIRouter, Depends, HTTPException, Request
+from fastapi import APIRouter, Depends, HTTPException, Request, Response
+import csv
+import io
 from sqlalchemy import select
 from .auth import authenticate, bootstrap
 from .classification import ENGINE_VERSION, Classifier
 from .config import settings
 from .datasets.service import DatasetService
 from .jobs import claim_next_job, complete_job, create_job, fail_job
+from .blocked_products import list_blockers, resolve_blocker, resolve_product_blockers, upsert_blocker
 from .context_keys import validate_entry
-from .models import FiscalContextEntry, ClassificationCache, ClassificationRequest, ClassificationResult, Dataset, Job
+from .models import BlockedProduct, FiscalContextEntry, ClassificationCache, ClassificationRequest, ClassificationResult, Dataset, Job
 
 logger = logging.getLogger("gti_fiscal_hub")
 
@@ -182,6 +185,63 @@ def router(session_factory, bootstrap_token):
             }
         finally: session.close()
 
+    def _blocked_view(row):
+        return {
+            "id": row.id, "tenant_id": row.tenant_id, "company": row.company, "product_code": row.product_code,
+            "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,
+            "next_retry_at": row.next_retry_at.isoformat() if row.next_retry_at else None, "resolved_at": row.resolved_at.isoformat() if row.resolved_at else None,
+            "resolution_source": row.resolution_source,
+        }
+
+    @routes.get("/blockers")
+    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)]}
+        finally:
+            session.close()
+
+    @routes.get("/blockers/export.csv")
+    def blockers_export(identity=Depends(principal)):
+        session, agent = identity
+        try:
+            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):
+                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:
+            session.close()
+
+    @routes.post("/blockers/reprocess")
+    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"))
+            codes = [row.product_code for row in rows if row.product_code]
+            if not codes:
+                return {"created": False, "reason": "NO_DUE_BLOCKERS"}
+            job = create_job(session, agent, "REPROCESS_PRODUCTS", {"codigos": codes, "reason": "BLOCKER_REPROCESS"})
+            return {"created": True, "job_id": job.id, "products": len(codes)}
+        finally:
+            session.close()
+
+    @routes.post("/blockers/{blocker_id}/resolve")
+    def blockers_resolve(blocker_id: int, payload: dict, identity=Depends(principal)):
+        session, agent = identity
+        try:
+            row = session.get(BlockedProduct, blocker_id)
+            if row is None or row.tenant_id != agent.tenant_id:
+                raise HTTPException(404, "blocker not found")
+            if not resolve_blocker(session, blocker_id, payload.get("resolution_source", "OPERATOR")):
+                raise HTTPException(404, "blocker not found")
+            return {"resolved": True, "blocker_id": blocker_id}
+        finally:
+            session.close()
+
     def _job_view(job):
         return {
             "job_id": job.id, "agent_id": job.agent_id, "kind": job.kind, "status": job.status,
@@ -231,7 +291,12 @@ def router(session_factory, bootstrap_token):
         try:
             job = session.get(Job, job_id)
             if job is None or job.agent_id != agent.id: raise HTTPException(404, "job not found")
-            if not complete_job(session, job, payload.get("result_summary", {})): raise HTTPException(409, "job is not RUNNING")
+            result_summary = payload.get("result_summary", {})
+            for blocker in result_summary.get("blocked_products", []):
+                upsert_blocker(session, agent.tenant_id, blocker)
+            for product_code in result_summary.get("resolved_products", []):
+                resolve_product_blockers(session, agent.tenant_id, product_code)
+            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))
         finally: session.close()

central/app/blocked_products.py

@@ -0,0 +1,62 @@
+"""Central blocked-product registry with fingerprint deduplication."""
+
+from datetime import datetime, timezone
+from sqlalchemy import select
+
+from .models import BlockedProduct
+from .operational_state import blocker_fingerprint
+
+
+def upsert_blocker(session, tenant_id: int, payload: dict) -> 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)
+        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)
+        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):
+    query = select(BlockedProduct).where(BlockedProduct.tenant_id == tenant_id, BlockedProduct.resolved_at.is_(None))
+    if input_fingerprint is not None:
+        query = query.where(BlockedProduct.input_fingerprint != input_fingerprint)
+    if blocker_type:
+        query = query.where(BlockedProduct.blocker_type == blocker_type)
+    if field:
+        query = query.where(BlockedProduct.field == field)
+    return list(session.scalars(query).all())
+
+
+def retry_due(row: BlockedProduct, current_input_fingerprint: str, now: datetime | None = None) -> bool:
+    now = now or datetime.now(timezone.utc)
+    if row.input_fingerprint != current_input_fingerprint:
+        return True
+    return row.next_retry_at is not None and row.next_retry_at <= now
+
+
+def resolve_blocker(session, blocker_id: int, source: str) -> bool:
+    row = session.get(BlockedProduct, blocker_id)
+    if row is None:
+        return False
+    row.resolved_at = datetime.now(timezone.utc)
+    row.resolution_source = source
+    session.commit()
+    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()
+    now = datetime.now(timezone.utc)
+    for row in rows:
+        row.resolved_at = now
+        row.resolution_source = source
+    if rows:
+        session.commit()
+    return len(rows)

central/app/models.py

@@ -250,6 +250,30 @@ class Job(Base):
     error: Mapped[str | None] = mapped_column(Text, nullable=True)
 
 
+class BlockedProduct(Base):
+    """Deduplicated safe exception for one tenant/product/field/input."""
+    __tablename__ = "blocked_products"
+    __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)
+    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)
+    description: Mapped[str | None] = mapped_column(Text, nullable=True)
+    blocker_type: Mapped[str] = mapped_column(String(48), index=True)
+    field: Mapped[str] = mapped_column(String(64), index=True)
+    reason_code: Mapped[str] = mapped_column(String(96))
+    evidence_summary: Mapped[dict] = mapped_column(JSON, default=dict)
+    input_fingerprint: Mapped[str] = mapped_column(String(64), index=True)
+    blocker_fingerprint: Mapped[str] = mapped_column(String(64))
+    first_seen_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())
+    retry_count: Mapped[int] = mapped_column(Integer, default=0, server_default="0")
+    next_retry_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
+    resolved_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
+    resolution_source: Mapped[str | None] = mapped_column(String(48), nullable=True)
+
+
 class ClassificationRequest(Base):
     __tablename__ = "classification_requests"
     id: Mapped[int] = mapped_column(primary_key=True)

central/app/operational_state.py

@@ -0,0 +1,48 @@
+"""Per-product operational state and safe partial-processing metrics."""
+
+from collections import Counter
+import hashlib
+import json
+
+STATES = {
+    "COMPLETE", "CORRECTED", "BLOCKED_MISSING_EVIDENCE", "BLOCKED_CONFLICT",
+    "BLOCKED_COMPANY_CONTEXT", "BLOCKED_SCHEMA", "FAILED_TECHNICAL",
+}
+
+
+def blocker_fingerprint(tenant: str, company: str, product_code: str, field: str, reason_code: str, input_fingerprint: str) -> str:
+    raw = {"tenant": tenant, "company": company, "product_code": product_code, "field": field, "reason_code": reason_code, "input_fingerprint": input_fingerprint}
+    return hashlib.sha256(json.dumps(raw, sort_keys=True, separators=(",", ":")).encode()).hexdigest()
+
+
+def classify_product_state(product: dict, decision: dict, *, write_status: str | None = None) -> dict:
+    status = str(decision.get("status") or "").upper()
+    if write_status == "FAILED":
+        state = "FAILED_TECHNICAL"
+    elif status in {"AMBIGUOUS", "CONFLICT", "NCM_CONFLICT"}:
+        state = "BLOCKED_CONFLICT"
+    elif status in {"INVALID_NCM", "LEGACY_NCM", "NOT_FOUND"}:
+        state = "BLOCKED_MISSING_EVIDENCE"
+    elif status in {"COMPANY_CONTEXT_REQUIRED", "COMPANY_CONTEXT_FAIL"}:
+        state = "BLOCKED_COMPANY_CONTEXT"
+    elif status in {"ERP_SCHEMA_LIMITATION", "UNSUPPORTED_SCHEMA"}:
+        state = "BLOCKED_SCHEMA"
+    elif write_status == "SUCCEEDED":
+        state = "CORRECTED"
+    else:
+        state = "COMPLETE"
+    return {"product_code": str(product.get("CODIGO") or ""), "state": state, "decision_status": status}
+
+
+def should_retry_blocker(previous_input_fingerprint: str | None, current_input_fingerprint: str) -> bool:
+    return previous_input_fingerprint != current_input_fingerprint
+
+
+def summarize_operational_states(states: list[dict]) -> dict:
+    counts = Counter(item.get("state") for item in states)
+    result = {state: counts.get(state, 0) for state in sorted(STATES)}
+    result["TOTAL_PRODUCTS"] = len(states)
+    result["SAFE_BLOCKED"] = sum(result[state] for state in STATES if state.startswith("BLOCKED_"))
+    result["AUTONOMOUSLY_PROCESSABLE_PRODUCTS"] = result["TOTAL_PRODUCTS"] - result["FAILED_TECHNICAL"]
+    result["AGENT_HEALTH"] = "HEALTHY" if result["FAILED_TECHNICAL"] == 0 else "DEGRADED"
+    return result