feat: zero-touch Agent — central scheduler, central pause, delta sync, discovery-driven installer

Hub: installations.processing_enabled / sync_interval_seconds / last_sync_* (migration 0026).
The first heartbeat of an installation without coverage queues exactly one INITIAL_SYNC;
afterwards one SYNC_INCREMENTAL per interval (idempotent, backs off on failure). One fiscal
job at a time per Agent (FIFO, stale RUNNING recovery). Job completion now updates the
coverage snapshot, blockers and JOB_STARTED/COMPLETED/FAILED events. Panel: Sincronizar
agora / Reprocessar tudo / Pausar / Retomar + status; pause is enforced Hub-side.
Agent: local AGENT_JOBS_ENABLED is ignored (central control); SYNC_INCREMENTAL re-classifies
only new/changed products via a per-product fingerprint+decision cache and rebuilds coverage
from it; results carry operational summary and blockers; Firebird credentials come from a
DPAPI firebird.bin. Setup: read-only SGV Firebird discovery (auto-select only when
unambiguous, wizard chooser otherwise), starts the service, waits for the first heartbeat,
upgrade-in-place mode. All jobs stay read-only (FIREBIRD_WRITES=0). Version 1.1.3.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

central/app/jobs.py

@@ -1,31 +1,74 @@
-"""Minimal job queue for Agent orchestration: create, atomically claim, complete, fail."""
-from datetime import datetime, timezone
+"""Job queue for Agent orchestration: create (idempotent per kind), claim one job at
+a time per Agent, complete, fail. Also the central sync scheduler helpers.
+
+Every fiscal job is READ-ONLY on the ERP (INITIAL_SYNC, SYNC_INCREMENTAL,
+REPROCESS_ALL, REPROCESS_PRODUCTS); automation never authorizes Firebird writes.
+"""
+from datetime import datetime, timedelta, timezone
 
 from sqlalchemy import select, update
 
-from .models import Job
+from .models import Installation, Job
 
-_VALID_KINDS = {"REPROCESS_ALL", "REPROCESS_PRODUCTS"}
+FULL_KINDS = ("INITIAL_SYNC", "REPROCESS_ALL")           # read every product, reclassify all
+SYNC_KINDS = FULL_KINDS + ("SYNC_INCREMENTAL",)          # jobs that count as "a sync happened"
+_VALID_KINDS = set(SYNC_KINDS) | {"REPROCESS_PRODUCTS"}
 _MAX_ERROR_LENGTH = 2000
+STALE_RUNNING_AFTER = timedelta(hours=1)                  # an Agent that died mid-job must not block the queue forever
+
+
+def _aware(value):
+    if value is not None and value.tzinfo is None:
+        return value.replace(tzinfo=timezone.utc)
+    return value
 
 
 def create_job(session, agent, kind, payload=None):
+    """Create a job. Sync-type jobs are idempotent: if this Agent already has a
+    PENDING/RUNNING job that covers the request, that job is returned instead
+    (no duplicates, never two full jobs at once)."""
     if kind not in _VALID_KINDS:
         raise ValueError(f"unsupported job kind: {kind}")
+    if kind in SYNC_KINDS:
+        # A full job subsumes an incremental one, not the other way round: asking for a
+        # full reprocess while only an incremental is queued must still queue the full.
+        blocking = FULL_KINDS if kind in FULL_KINDS else SYNC_KINDS
+        existing = session.scalar(select(Job).where(Job.agent_id == agent.id, Job.kind.in_(blocking), Job.status.in_(("PENDING", "RUNNING"))).order_by(Job.id).limit(1))
+        if existing is not None:
+            return existing
     job = Job(agent_id=agent.id, kind=kind, status="PENDING", payload=payload or {})
     session.add(job)
     session.commit()
     return job
 
 
+def recover_stale_running(session, agent, now=None):
+    now = now or datetime.now(timezone.utc)
+    cutoff = now - STALE_RUNNING_AFTER
+    stale = session.scalars(select(Job).where(Job.agent_id == agent.id, Job.status == "RUNNING")).all()
+    changed = False
+    for job in stale:
+        if job.started_at is not None and _aware(job.started_at) < cutoff:
+            job.status, job.finished_at, job.error = "FAILED", now, "stale: Agent stopped reporting while RUNNING"
+            changed = True
+    if changed:
+        session.commit()
+
+
 def claim_next_job(session, agent):
-    """Atomically claim the oldest PENDING job for this agent only.
-    A single UPDATE...WHERE status='PENDING' guards against two concurrent
-    pollers (this agent or any other) claiming the same job twice."""
+    """Atomically claim the oldest PENDING job for this agent only — and only when
+    processing is enabled centrally and no other job of this Agent is RUNNING
+    (one fiscal job at a time per Firebird)."""
+    installation = session.scalar(select(Installation).where(Installation.agent_id == agent.id))
+    if installation is not None and not installation.processing_enabled:
+        return None
+    recover_stale_running(session, agent)
+    if session.scalar(select(Job.id).where(Job.agent_id == agent.id, Job.status == "RUNNING").limit(1)) is not None:
+        return None
     candidate = session.scalar(
         select(Job.id)
         .where(Job.agent_id == agent.id, Job.status == "PENDING")
-        .order_by(Job.created_at.asc())
+        .order_by(Job.created_at.asc(), Job.id.asc())
         .limit(1)
     )
     if candidate is None:
@@ -61,3 +104,25 @@ def fail_job(session, job, error):
     )
     session.commit()
     return result.rowcount == 1
+
+
+def schedule_automatic_sync(session, agent, installation, snapshot, now=None):
+    """Called from the Agent heartbeat. Central scheduler, idempotent:
+    - first heartbeat of an installation with no initial coverage and no sync job
+      ever created -> exactly ONE INITIAL_SYNC;
+    - otherwise, if processing is enabled and the last sync attempt is older than
+      the installation's interval, and nothing is queued/running -> ONE SYNC_INCREMENTAL.
+    Returns the created job or None."""
+    if installation is None or not installation.processing_enabled:
+        return None
+    now = now or datetime.now(timezone.utc)
+    sync_jobs = session.scalars(select(Job).where(Job.agent_id == agent.id, Job.kind.in_(SYNC_KINDS)).order_by(Job.id.desc())).all()
+    if any(job.status in ("PENDING", "RUNNING") for job in sync_jobs):
+        return None
+    never_synced = not any(job.kind == "INITIAL_SYNC" or job.kind == "REPROCESS_ALL" for job in sync_jobs) and not int(getattr(snapshot, "total_products", 0) or 0)
+    if never_synced and not sync_jobs:
+        return create_job(session, agent, "INITIAL_SYNC", {"reason": "INITIAL_SYNC"})
+    last_attempt = max((_aware(job.finished_at or job.created_at) for job in sync_jobs if (job.finished_at or job.created_at)), default=None)
+    if last_attempt is not None and (now - last_attempt).total_seconds() >= installation.sync_interval_seconds:
+        return create_job(session, agent, "SYNC_INCREMENTAL", {"reason": "SCHEDULED"})
+    return None