fix: isolate per-organization aggregation failures

reflexio/server/services/playbook/README.md

@@ -112,6 +112,13 @@ scheduled cycle; new durable signals preserve that due time so changes coalesce,
 while a remaining backlog continues in bounded follow-up units. The manual
 `/api/run_playbook_aggregation` route remains a fenced administrative full rerun.
 
+The scheduler processes organizations sequentially. An exception escaping an
+organization's execution defers that organization for five minutes using
+monotonic time; later organizations continue after the failing call returns.
+Failed repair attempts also respect the five-minute repair interval. Logs
+identify the organization and failing stage, and provider enumeration failures
+remain separate tick-level errors. Durable retries and lease fencing are unchanged.
+
 Incremental work is isolated by `agent_version`:
 
 1. Admit only the newest configured window of CURRENT, nonempty rows that have

reflexio/server/services/playbook/aggregation_scheduler.py

@@ -112,8 +112,12 @@ def __init__(
         self._poll_interval_seconds = poll_interval_seconds
         self._worker_id = worker_id or uuid.uuid4().hex
         self._last_repair_at: dict[str, float] = {}
+        self._retry_after: dict[str, float] = {}
+        # Organization execution is sequential on the scheduler thread.
+        self._active_stage = "configuration"
 
     def _run_context(self, context: RequestContext) -> None:
+        self._active_stage = "configuration"
         storage = context.storage
         playbook_config = getattr(
             context.configurator.get_config(), "user_playbook_extractor_config", None
@@ -140,15 +144,17 @@ def _run_context(self, context: RequestContext) -> None:
             last_repair_at is None
             or repair_now - last_repair_at >= _REPAIR_INTERVAL_SECONDS
         ):
-            repaired = storage.repair_playbook_aggregation_pending_state()
             self._last_repair_at[context.org_id] = repair_now
+            self._active_stage = "repair"
+            repaired = storage.repair_playbook_aggregation_pending_state()
             for agent_version in repaired:
                 logger.info(
                     "event=playbook_aggregation_progress state=scheduled org_id=%s "
                     "agent_version=%s reason=discovery_repair pending=true",
                     context.org_id,
                     agent_version,
                 )
+        self._active_stage = "claim"
         claim = storage.claim_due_playbook_aggregation(
             owner=f"aggregation:{self._worker_id}:{context.org_id}",
             lease_seconds=AGGREGATION_LEASE_SECONDS,
@@ -163,6 +169,7 @@ def _run_context(self, context: RequestContext) -> None:
             claim.agent_version,
             claim.fence,
         )
+        self._active_stage = "aggregation"
         heartbeat = AggregationLeaseHeartbeat(storage, claim)
         heartbeat.start()
         success = False
@@ -239,6 +246,7 @@ def _run_context(self, context: RequestContext) -> None:
                 claim.fence,
             )
         finally:
+            self._active_stage = "finalization"
             heartbeat.stop()
             active_claim = heartbeat.claim
             finished = storage.finish_playbook_aggregation_claim(
@@ -285,9 +293,28 @@ def _run_once(self) -> float:
             for context in self._context_provider():
                 if self._stop_event.is_set():
                     break
-                self._run_context(context)
+                org_id = context.org_id
+                if time.monotonic() < self._retry_after.get(org_id, 0):
+                    continue
+                try:
+                    self._run_context(context)
+                except Exception:
+                    self._retry_after[org_id] = (
+                        time.monotonic() + _REPAIR_INTERVAL_SECONDS
+                    )
+                    logger.exception(
+                        "event=playbook_aggregation_scheduler_org_failed org_id=%s "
+                        "stage=%s retry_after_seconds=%s",
+                        org_id,
+                        self._active_stage,
+                        _REPAIR_INTERVAL_SECONDS,
+                    )
+                else:
+                    self._retry_after.pop(org_id, None)
         except Exception:
-            logger.exception("event=playbook_aggregation_scheduler_tick_failed")
+            logger.exception(
+                "event=playbook_aggregation_scheduler_tick_failed stage=context_provider"
+            )
         return self._poll_interval_seconds
 
     def _on_started(self) -> None:

tests/server/services/playbook/test_aggregation_scheduler.py

@@ -190,6 +190,104 @@ def test_lease_heartbeat_marks_renewal_exception_as_lost() -> None:
         heartbeat.require_live()
 
 
+@pytest.mark.parametrize(
+    "failing_operation",
+    [
+        "repair_playbook_aggregation_pending_state",
+        "claim_due_playbook_aggregation",
+        "finish_playbook_aggregation_claim",
+    ],
+)
+def test_org_failure_defers_only_that_org_and_recovers(
+    failing_operation, monkeypatch, caplog
+) -> None:
+    clock = [1000.0]
+    monkeypatch.setattr(aggregation_scheduler.time, "monotonic", lambda: clock[0])
+    broken = MagicMock(supports_incremental_playbook_aggregation=True)
+    healthy = MagicMock(supports_incremental_playbook_aggregation=True)
+    for storage in (broken, healthy):
+        storage.repair_playbook_aggregation_pending_state.return_value = []
+        storage.claim_due_playbook_aggregation.return_value = None
+    healthy.claim_due_playbook_aggregation.return_value = PlaybookAggregationClaim(
+        "v1", "healthy-owner", 1, 0, 2000
+    )
+    healthy.get_playbook_aggregation_invalidations.return_value = [
+        SimpleNamespace(invalidation_id=i)
+        for i in range(aggregation_scheduler.AGGREGATION_INVALIDATION_BATCH_SIZE + 1)
+    ]
+    monkeypatch.setattr(
+        aggregation_scheduler.AggregationLeaseHeartbeat, "start", lambda _: None
+    )
+    if failing_operation == "finish_playbook_aggregation_claim":
+        broken.claim_due_playbook_aggregation.return_value = PlaybookAggregationClaim(
+            "v1", "owner", 1, 0, 2000
+        )
+        broken.get_playbook_aggregation_invalidations.return_value = [
+            SimpleNamespace(invalidation_id=i)
+            for i in range(
+                aggregation_scheduler.AGGREGATION_INVALIDATION_BATCH_SIZE + 1
+            )
+        ]
+    failure = getattr(broken, failing_operation)
+    failure.side_effect = RuntimeError("database unavailable")
+    contexts = [_context(broken), _context(healthy)]
+    contexts[1].org_id = "org-2"
+    scheduler = aggregation_scheduler.PlaybookAggregationScheduler(
+        context_provider=lambda: contexts
+    )
+
+    scheduler._run_once()
+    assert failure.call_count == 1
+    assert healthy.claim_due_playbook_aggregation.call_count == 1
+    assert healthy.finish_playbook_aggregation_claim.call_count == 1
+    assert healthy.finish_playbook_aggregation_claim.call_args.kwargs["success"] is True
+    assert "org_id=org-1" in caplog.text
+    expected_stage = {
+        "repair_playbook_aggregation_pending_state": "repair",
+        "claim_due_playbook_aggregation": "claim",
+        "finish_playbook_aggregation_claim": "finalization",
+    }[failing_operation]
+    assert f"stage={expected_stage}" in caplog.text
+    clock[0] += 299
+    scheduler._run_once()
+    assert failure.call_count == 1
+    assert healthy.claim_due_playbook_aggregation.call_count == 2
+    assert healthy.finish_playbook_aggregation_claim.call_count == 2
+
+    failure.side_effect = None
+    broken.claim_due_playbook_aggregation.return_value = None
+    clock[0] += 1
+    scheduler._run_once()
+    assert "org-1" not in scheduler._retry_after
+    assert healthy.claim_due_playbook_aggregation.call_count == 3
+    assert healthy.finish_playbook_aggregation_claim.call_count == 3
+
+
+def test_failed_repair_attempt_is_throttled(monkeypatch) -> None:
+    storage = MagicMock(supports_incremental_playbook_aggregation=True)
+    storage.repair_playbook_aggregation_pending_state.side_effect = RuntimeError("down")
+    storage.claim_due_playbook_aggregation.return_value = None
+    monkeypatch.setattr(aggregation_scheduler.time, "monotonic", lambda: 1000.0)
+    scheduler = aggregation_scheduler.PlaybookAggregationScheduler(
+        context_provider=lambda: []
+    )
+    with pytest.raises(RuntimeError, match="down"):
+        scheduler._run_context(_context(storage))
+    scheduler._run_context(_context(storage))
+    storage.repair_playbook_aggregation_pending_state.assert_called_once()
+
+
+def test_context_provider_failure_remains_recoverable(caplog) -> None:
+    provider = MagicMock(side_effect=[RuntimeError("repository unavailable"), []])
+    scheduler = aggregation_scheduler.PlaybookAggregationScheduler(
+        context_provider=provider
+    )
+    scheduler._run_once()
+    scheduler._run_once()
+    assert provider.call_count == 2
+    assert "stage=context_provider" in caplog.text
+
+
 def test_stopped_local_scheduler_drops_captured_context(monkeypatch) -> None:
     monkeypatch.delenv("DEPLOYMENT_MODE", raising=False)
     aggregation_scheduler._LOCAL_SCHEDULERS.clear()