feat(sp27): surface consecutive_failures + last_error in Scheduler.status()

backend/src/cyclone/scheduler.py

@@ -95,6 +95,12 @@ class TickResult:
     files_skipped: int = 0
     files_errored: int = 0
     errors: list[str] = field(default_factory=list)
+    # SP27 Task 9: ``sftp_failed`` is the discriminator that separates
+    # SFTP-side failures (the listing step) from per-file processing
+    # errors. ``tick()`` keys off this flag — NOT ``result.errors`` —
+    # so a single malformed 999 cannot flip the operator's destructive
+    # pill. Set in ``_tick_impl``'s listing-error branches only.
+    sftp_failed: bool = False
 
     def as_dict(self) -> dict[str, Any]:
         return {
@@ -107,12 +113,22 @@ class TickResult:
             "files_skipped": self.files_skipped,
             "files_errored": self.files_errored,
             "errors": list(self.errors),
+            "sftp_failed": self.sftp_failed,
         }
 
 
 @dataclass
 class SchedulerStatus:
-    """Snapshot of the scheduler's runtime state."""
+    """Snapshot of the scheduler's runtime state.
+
+    SP27 Task 9 added the four SFTP-error fields
+    (``consecutive_failures``, ``last_error``, ``last_error_at``,
+    ``last_sftp_attempt_at``) so the operator's UI can tell "all
+    quiet on the MFT front" from "we've been unable to reach the
+    MFT server for 3 hours". The previous 06/25 incident went
+    unnoticed because the status dict had no signal for a hung
+    SFTP poll.
+    """
 
     running: bool
     poll_interval_seconds: int
@@ -123,6 +139,13 @@ class SchedulerStatus:
     total_skipped: int
     total_errored: int
     last_tick: Optional[TickResult] = None
+    # SP27 Task 9: SFTP-error surfacing. The UI flips to a destructive
+    # pill when ``consecutive_failures >= 3`` (``CYCLONE_SCHEDULER_FAIL_THRESHOLD``
+    # in the operator pill config; not enforced server-side).
+    consecutive_failures: int = 0
+    last_error: Optional[str] = None
+    last_error_at: Optional[datetime] = None
+    last_sftp_attempt_at: Optional[datetime] = None
 
     def as_dict(self) -> dict[str, Any]:
         return {
@@ -137,6 +160,15 @@ class SchedulerStatus:
             "total_skipped": self.total_skipped,
             "total_errored": self.total_errored,
             "last_tick": self.last_tick.as_dict() if self.last_tick else None,
+            "consecutive_failures": self.consecutive_failures,
+            "last_error": self.last_error,
+            "last_error_at": (
+                self.last_error_at.isoformat() if self.last_error_at else None
+            ),
+            "last_sftp_attempt_at": (
+                self.last_sftp_attempt_at.isoformat()
+                if self.last_sftp_attempt_at else None
+            ),
         }
 
 
@@ -231,6 +263,40 @@ class Scheduler:
         # ticks stack up; the next tick fires only after the previous
         # one finishes).
         self._tick_in_progress = False
+        # SP27 Task 9: SFTP-error surfacing. ``_consecutive_failures``
+        # counts back-to-back tick failures (cleared on the next
+        # successful tick). The other three fields are the most
+        # recent failure for the operator's pill — preserved across
+        # successes so the audit trail doesn't disappear the moment
+        # the MFT server recovers.
+        self._consecutive_failures = 0
+        self._last_error: Optional[str] = None
+        self._last_error_at: Optional[datetime] = None
+        self._last_sftp_attempt_at: Optional[datetime] = None
+
+    def _record_sftp_outcome(
+        self, *, success: bool, error: Optional[str] = None,
+    ) -> None:
+        """Update the SFTP-error state after a tick.
+
+        ``success=True`` resets the consecutive-failure counter; the
+        last-error fields are preserved as an audit trail (the
+        operator's UI can still surface "last failure: 2 hours ago"
+        after a recovery).
+
+        ``success=False`` bumps the counter and records the error
+        message + timestamp. ``last_sftp_attempt_at`` is bumped in
+        both cases so the operator can tell when the scheduler last
+        *tried* to reach the MFT (not just when it last failed).
+        """
+        now = datetime.now(timezone.utc)
+        self._last_sftp_attempt_at = now
+        if success:
+            self._consecutive_failures = 0
+            return
+        self._consecutive_failures += 1
+        self._last_error = error
+        self._last_error_at = now
 
     # ---- Public API -------------------------------------------------------
 
@@ -278,6 +344,10 @@ class Scheduler:
             total_skipped=self._total_skipped,
             total_errored=self._total_errored,
             last_tick=self._last_tick,
+            consecutive_failures=self._consecutive_failures,
+            last_error=self._last_error,
+            last_error_at=self._last_error_at,
+            last_sftp_attempt_at=self._last_sftp_attempt_at,
         )
 
     def is_running(self) -> bool:
@@ -291,6 +361,13 @@ class Scheduler:
         SFTP server from a stampede when the operator hits
         ``/api/admin/scheduler/tick`` while a scheduled tick is
         already running.
+
+        SP27 Task 9: the tick outcome is also recorded on the
+        scheduler's SFTP-error state via
+        :meth:`_record_sftp_outcome`. The discriminator is
+        ``TickResult.sftp_failed`` (set by ``_tick_impl``'s
+        listing-error branches only), not ``TickResult.errors``
+        which is also populated by per-file processing errors.
         """
         while self._tick_in_progress:
             await asyncio.sleep(0.05)
@@ -303,6 +380,20 @@ class Scheduler:
             self._total_processed += result.files_processed
             self._total_skipped += result.files_skipped
             self._total_errored += result.files_errored
+            # SFTP-error surfacing: discriminator is ``result.sftp_failed``,
+            # NOT ``result.errors``. ``sftp_failed`` is set by
+            # ``_tick_impl``'s listing-error branches (timeout, connect
+            # refused). Per-file processing errors append to
+            # ``result.errors`` but do NOT set ``sftp_failed`` — a
+            # single malformed 999 in a 5-file batch must not flip
+            # the operator's pill to destructive.
+            if result.sftp_failed:
+                self._record_sftp_outcome(
+                    success=False,
+                    error="; ".join(result.errors),
+                )
+            else:
+                self._record_sftp_outcome(success=True)
             return result
         finally:
             self._tick_in_progress = False
@@ -323,6 +414,11 @@ class Scheduler:
         Honors ``_stop_event`` between files. Concurrent calls are
         coalesced the same way :meth:`tick` does, so a slow handler
         can't be stampeded by a parallel admin invocation.
+
+        SP27 Task 9: deliberately does NOT update the SFTP-error
+        state (no listing, no SFTP). The consecutive-failure counter
+        is the signal for scheduled SFTP polling; operator-initiated
+        batches shouldn't be able to flip it.
         """
         while self._tick_in_progress:
             await asyncio.sleep(0.05)
@@ -386,11 +482,13 @@ class Scheduler:
         except asyncio.TimeoutError:
             log.exception("SFTP list_inbound timed out")
             result.errors.append("list_inbound: timeout")
+            result.sftp_failed = True
             result.finished_at = datetime.now(timezone.utc)
             return result
         except Exception as exc:  # noqa: BLE001
             log.exception("SFTP list_inbound failed")
             result.errors.append(f"list_inbound: {exc}")
+            result.sftp_failed = True
             result.finished_at = datetime.now(timezone.utc)
             return result
 

backend/tests/test_scheduler_status_errors.py

@@ -0,0 +1,229 @@
+"""SP27 Task 9: scheduler status surfaces SFTP failures.
+
+The 06/25 silent hang exposed a gap: ``Scheduler.status()`` had no
+signal that polling had stalled. The operator's UI couldn't tell
+"all quiet on the MFT front" from "we've been unable to reach the
+MFT server for 3 hours". This pins the fix — the status dict
+carries ``consecutive_failures``, ``last_error``, ``last_error_at``,
+and ``last_sftp_attempt_at`` so the operator pill can flip to
+destructive after 3 (or however many) consecutive failures.
+
+Discriminator under test: ``TickResult.sftp_failed`` is the flag
+``tick()`` keys off — NOT ``TickResult.errors``. Per-file processing
+errors append to ``errors`` but must NOT bump ``consecutive_failures``.
+The test suite pins both halves of the discriminator (SFTP failure
+bumps; per-file failure doesn't).
+"""
+from __future__ import annotations
+
+import asyncio
+from datetime import datetime, timezone
+
+import pytest
+
+from cyclone.clearhouse import InboundFile
+from cyclone.providers import SftpBlock
+from cyclone.scheduler import Scheduler
+
+
+def _block() -> SftpBlock:
+    return SftpBlock(
+        host="mft.example.com",
+        port=22,
+        username="user",
+        auth={"password_keychain_account": "x"},
+        paths={"inbound": "/inbound", "outbound": "/outbound"},
+        stub=True,
+    )
+
+
+# ---- pre-state: a fresh scheduler has no errors -------------------------
+
+
+def test_status_starts_with_zero_failures_and_no_error():
+    """A fresh scheduler must report no errors — pins that the new
+    fields default cleanly and don't trip on first read."""
+    sched = Scheduler(_block())
+    status = sched.status()
+    assert status.consecutive_failures == 0
+    assert status.last_error is None
+    assert status.last_error_at is None
+    assert status.last_sftp_attempt_at is None
+
+
+# ---- a failed SFTP call bumps consecutive_failures -----------------------
+
+
+@pytest.mark.asyncio
+async def test_status_records_last_error_after_failure():
+    """A failed tick bumps consecutive_failures and records the error.
+
+    Simulates a broken SFTP client. Pins the post-Task-8 behavior
+    where ``_tick_impl`` catches the error and ``tick()`` records
+    it on the scheduler's status.
+    """
+    class BrokenClient:
+        def list_inbound(self):
+            raise RuntimeError("simulated outage")
+
+        # SP27 Task 8: scheduler calls async_list_inbound() (the
+        # wait_for-wrapped variant). The broken stub raises before
+        # the wait_for wrapper runs — that's a real SFTP failure
+        # either way.
+        async def async_list_inbound(self):
+            raise RuntimeError("simulated outage")
+
+    sched = Scheduler(_block(), sftp_client_factory=lambda b: BrokenClient())
+
+    before = datetime.now(timezone.utc)
+    await sched.tick()
+    after = datetime.now(timezone.utc)
+
+    status = sched.status()
+    assert status.consecutive_failures == 1
+    assert status.last_error is not None
+    assert "simulated outage" in status.last_error
+    # last_error_at is set to roughly the time of the failed tick
+    assert status.last_error_at is not None
+    assert before <= status.last_error_at <= after
+    # last_sftp_attempt_at is also bumped — separate from last_error_at
+    # when the failure was an SFTP-side issue (not a parse error).
+    assert status.last_sftp_attempt_at is not None
+    assert before <= status.last_sftp_attempt_at <= after
+
+
+# ---- a successful tick clears consecutive_failures -----------------------
+
+
+@pytest.mark.asyncio
+async def test_successful_tick_clears_consecutive_failures():
+    """A tick that succeeds (or skips cleanly) resets the counter.
+
+    Without this, a transient blip would lock the destructive pill
+    on forever. Pins the recovery path.
+    """
+    class BrokenClient:
+        def list_inbound(self):
+            raise RuntimeError("transient")
+
+        async def async_list_inbound(self):
+            raise RuntimeError("transient")
+
+    class HealthyClient:
+        def list_inbound(self):
+            return []
+
+        async def async_list_inbound(self):
+            return []
+
+    # First: two failing ticks to bump the counter
+    sched = Scheduler(_block(), sftp_client_factory=lambda b: BrokenClient())
+    await sched.tick()
+    await sched.tick()
+    assert sched.status().consecutive_failures == 2
+
+    # Then: switch to a healthy client and tick again
+    sched._sftp_client_factory = lambda b: HealthyClient()
+    await sched.tick()
+    status = sched.status()
+    assert status.consecutive_failures == 0
+    # last_error is preserved (audit trail) but last_error_at is from
+    # the last failure, not the success.
+    assert status.last_error is not None  # the historical error stays
+    assert "transient" in status.last_error
+
+
+# ---- last_sftp_attempt_at moves on every attempt, not just failures ---
+
+
+@pytest.mark.asyncio
+async def test_last_sftp_attempt_at_advances_on_every_tick():
+    """Even a successful tick should bump last_sftp_attempt_at.
+
+    Without this, the operator can't tell when the scheduler last
+    *tried* to reach the MFT — only when it last failed.
+    """
+    sched = Scheduler(_block())  # default stub returns []
+    before = datetime.now(timezone.utc)
+    await sched.tick()
+    after = datetime.now(timezone.utc)
+    first = sched.status().last_sftp_attempt_at
+    assert first is not None
+    assert before <= first <= after
+
+    # Wait a moment, tick again — last_sftp_attempt_at should move forward.
+    await asyncio.sleep(0.01)
+    await sched.tick()
+    second = sched.status().last_sftp_attempt_at
+    assert second is not None
+    assert second > first
+
+
+# ---- as_dict surfaces the new fields for the API -----------------------
+
+
+def test_status_as_dict_includes_new_fields():
+    """The API endpoint (/api/health or /api/admin/scheduler) reads
+    status.as_dict() — the new fields must be present and serialized."""
+    sched = Scheduler(_block())
+    d = sched.status().as_dict()
+    assert "consecutive_failures" in d
+    assert d["consecutive_failures"] == 0
+    assert "last_error" in d
+    assert d["last_error"] is None
+    assert "last_error_at" in d
+    assert d["last_error_at"] is None
+    assert "last_sftp_attempt_at" in d
+    assert d["last_sftp_attempt_at"] is None
+
+
+# ---- per-file processing errors must NOT bump consecutive_failures ------
+#
+# This pins the SP27 Task 9 discriminator (``TickResult.sftp_failed``
+# vs. ``TickResult.errors``). Without this test, a future refactor
+# that broadens ``result.errors`` semantics — or rewires `tick()` to
+# key off it again — would silently re-break the discrimination.
+
+
+@pytest.mark.asyncio
+async def test_per_file_error_does_not_bump_consecutive_failures(monkeypatch):
+    """Per-file processing errors must NOT flip the destructive pill.
+
+    The listing step succeeds but ``_handle_one`` simulates the
+    real handler's error path: it appends to ``result.errors``
+    (the way ``_handle_one`` does on parser exceptions) but does
+    NOT touch ``result.sftp_failed``. The pill must stay clean.
+    """
+
+    class FilesClient:
+        async def async_list_inbound(self):
+            return [
+                InboundFile(
+                    name="tp123-999_MTRACKING-20260525001606060-1of1.x12",
+                    size=100,
+                    modified_at=datetime.now(timezone.utc),
+                    # ``local_path`` is None because fake_handle_one
+                    # never reaches the download step.
+                    local_path=None,
+                )
+            ]
+
+    sched = Scheduler(_block(), sftp_client_factory=lambda b: FilesClient())
+
+    async def fake_handle_one(f, result):
+        # Mirrors the real ``_handle_one`` parser-failure path:
+        # appends to result.errors, does NOT set sftp_failed.
+        result.errors.append(f"{f.name}: ValueError: garbage")
+
+    monkeypatch.setattr(sched, "_handle_one", fake_handle_one)
+
+    await sched.tick()
+    status = sched.status()
+
+    # Counter stays clean — per-file error is not an SFTP outage.
+    assert status.consecutive_failures == 0
+    # ``last_error`` is reserved for SFTP-side failures.
+    assert status.last_error is None
+    # But ``last_sftp_attempt_at`` moves forward regardless — the
+    # scheduler did try, it just failed to ingest the file.
+    assert status.last_sftp_attempt_at is not None