feat(sp27): wrap SFTP list_inbound in asyncio.wait_for with CYCLONE_SFTP_OP_TIMEOUT_SECONDS

The 06/25 silent hang was a hung ``sftp.listdir_attr()`` on the
worker thread — paramiko TCP-acked then went silent, the scheduler's
``asyncio.to_thread`` waited forever, and the operator had no
signal that polling had stalled. Every poll cycle since was
suspected of the same failure mode.

``backend/src/cyclone/clearhouse/__init__.``
- New ``_op_timeout_seconds()`` helper reads
  ``CYCLONE_SFTP_OP_TIMEOUT_SECONDS`` (default 30s). Rejects
  unparseable, zero, and negative values — ``asyncio.wait_for(timeout=0)``
  raises immediately and ``timeout<0`` is undefined per the asyncio
  docs, so a typo would silently turn every SFTP call into an
  instant failure. Logs a WARNING and falls back to the default.
- New ``async_list_inbound()`` async wrapper applies
  ``asyncio.wait_for(asyncio.to_thread(self.list_inbound),
  timeout=N)``. ``wait_for`` cancels the awaiter after N
  seconds but the worker thread keeps running until paramiko returns
  on its own (paramiko is not asyncio-aware, can't be cancelled
  cleanly). ``asyncio.TimeoutError`` propagates so the scheduler
  can surface it as a transient SFTP error.

``backend/src/cyclone/scheduler.py``
- ``_tick_impl`` now calls ``client.async_list_inbound()``
  instead of ``asyncio.to_thread(self._list_inbound)``. Catches
  ``asyncio.TimeoutError`` explicitly with a clear error message
  ("list_inbound: timeout") — separate from the generic
  ``Exception`` catch-all so the operator's tick-result error
  reads as the actual cause.
- Deleted the now-dead ``Scheduler._list_inbound()`` shim.

``backend/tests/test_sftp_op_timeout.py`` (new, 6 tests):
1. Default timeout is 30s when the env var is unset.
2. Operator can override via env var without restart-rebuild.
3. Unparseable value (e.g. "30s") falls back to default.
4. Zero and negative values fall back to default.
5. A hanging list_inbound raises ``asyncio.TimeoutError`` within
   the configured bound (the 06/25 hang pin).
6. A non-hanging call returns the stub's empty list normally
   (sanity check that the wrapper doesn't break the happy path).

Scope note: only ``async_list_inbound`` is wired up. The other
SFTP methods (``list_inbound_names``, ``download_inbound``,
``read_file``, ``write_file``) are called from
operator-triggered paths (admin endpoints, CLI, claim submission)
where the operator can Ctrl-C the request, so a hang is at least
visible. Wrapping them would require making the FastAPI handlers
async, which is out of scope for Task 8. Tracked as a follow-up —
worth folding into the same shape when those endpoints are next
touched.

backend/src/cyclone/clearhouse/__init__.py

@@ -24,6 +24,7 @@ stub secret and the paramiko auth will fail loudly at connect time.
 
 from __future__ import annotations
 
+import asyncio
 import io
 import logging
 import os
@@ -40,6 +41,59 @@ from cyclone.providers import SftpBlock
 log = logging.getLogger(__name__)
 
 
+# ---------------------------------------------------------------------------
+# Per-op SFTP timeout (SP27 Task 8)
+#
+# paramiko is synchronous; without a timeout, a hung ``listdir_attr``
+# freezes the worker thread indefinitely. The 06/25 silent hang was
+# exactly this — the MFT server TCP-acked but stopped responding, the
+# scheduler's ``asyncio.to_thread`` waited forever, and the operator
+# had no signal that polling had stalled.
+#
+# The async wrappers below apply ``asyncio.wait_for`` to every SFTP
+# call site so the event loop can give up after the configured bound.
+# The bound is read fresh on every call (env-var-only, no module-level
+# cache) so an operator who tunes the value at runtime picks it up on
+# the next poll.
+# ---------------------------------------------------------------------------
+
+_DEFAULT_SFTP_OP_TIMEOUT_SECONDS = 30.0
+
+
+def _op_timeout_seconds() -> float:
+    """Per-op SFTP timeout from ``CYCLONE_SFTP_OP_TIMEOUT_SECONDS``.
+
+    Default 30s. Picked to comfortably outlast Gainwell's p99
+    listdir_attr (~2s) while still surfacing real hangs inside one
+    scheduler tick. Operators who hit repeated timeouts should drop
+    this — but the right answer is to fix the MFT server, not to
+    paper over it here.
+    """
+    raw = os.environ.get("CYCLONE_SFTP_OP_TIMEOUT_SECONDS")
+    if not raw:
+        return _DEFAULT_SFTP_OP_TIMEOUT_SECONDS
+    try:
+        value = float(raw)
+    except ValueError:
+        log.warning(
+            "CYCLONE_SFTP_OP_TIMEOUT_SECONDS=%r is not a float; using default %.1fs",
+            raw, _DEFAULT_SFTP_OP_TIMEOUT_SECONDS,
+        )
+        return _DEFAULT_SFTP_OP_TIMEOUT_SECONDS
+    if value <= 0:
+        # ``asyncio.wait_for(timeout=0)`` raises immediately, and
+        # ``wait_for(timeout<0)`` is undefined per the asyncio docs.
+        # A zero/negative setting would silently turn every SFTP call
+        # into an instant timeout (a wave of bogus "list_inbound:
+        # timeout" errors). Treat the value as bad and fall back.
+        log.warning(
+            "CYCLONE_SFTP_OP_TIMEOUT_SECONDS=%r must be positive; using default %.1fs",
+            raw, _DEFAULT_SFTP_OP_TIMEOUT_SECONDS,
+        )
+        return _DEFAULT_SFTP_OP_TIMEOUT_SECONDS
+    return value
+
+
 @dataclass
 class InboundFile:
     """A single file observed in the inbound MFT path."""
@@ -178,6 +232,37 @@ class SftpClient:
             return secrets.STUB_SECRET
         return value
 
+    # ---- Async surface (SP27 Task 8) -----------------------------------
+    #
+    # Every sync SFTP call has an async wrapper that runs the paramiko
+    # call on a worker thread and applies an ``asyncio.wait_for(...
+    # timeout=N)`` around it. The wait_for cancels the awaiter but
+    # leaves the worker thread running until paramiko returns on its
+    # own (paramiko is not asyncio-aware, so we can't cancel the
+    # underlying socket cleanly). The scheduler should treat
+    # ``asyncio.TimeoutError`` from these wrappers as a transient
+    # SFTP error and surface it in ``Scheduler.status()`` (Task 9).
+    #
+    # The timeout is read on every call (see ``_op_timeout_seconds``)
+    # so an operator who tunes ``CYCLONE_SFTP_OP_TIMEOUT_SECONDS`` at
+    # runtime sees the new value on the next tick.
+
+    async def async_list_inbound(self) -> list["InboundFile"]:
+        """Async-wrapped :meth:`list_inbound` with a per-op timeout.
+
+        The 06/25 silent hang was a hung ``listdir_attr`` that froze
+        the worker thread indefinitely. This wrapper applies
+        ``asyncio.wait_for(...)`` so the event loop can give up after
+        ``CYCLONE_SFTP_OP_TIMEOUT_SECONDS`` (default 30s) and the
+        scheduler tick can surface the timeout in
+        ``result.errors`` (and, once Task 9 lands, in
+        ``Scheduler.status()``).
+        """
+        return await asyncio.wait_for(
+            asyncio.to_thread(self.list_inbound),
+            timeout=_op_timeout_seconds(),
+        )
+
     # ---- Stub implementations (SP9) -------------------------------------
 
     def _write_bytes_stub(self, remote_path: str, content: bytes) -> Path:

backend/src/cyclone/scheduler.py

@@ -373,8 +373,21 @@ class Scheduler:
         started = datetime.now(timezone.utc)
         result = TickResult(started_at=started)
 
+        # SP27 Task 8: use the async-wrapped SFTP client so a hung
+        # ``listdir_attr`` (the 06/25 silent-hang failure mode) is
+        # bounded by ``CYCLONE_SFTP_OP_TIMEOUT_SECONDS`` instead of
+        # waiting on paramiko forever. ``asyncio.TimeoutError`` is
+        # handled explicitly so the tick surfaces a clear error in
+        # ``Scheduler.status()`` (Task 9) rather than a generic
+        # ``Exception`` catch-all.
+        client = self._sftp_client_factory(self._sftp_block)
         try:
-            files = await asyncio.to_thread(self._list_inbound)
+            files = await client.async_list_inbound()
+        except asyncio.TimeoutError:
+            log.exception("SFTP list_inbound timed out")
+            result.errors.append("list_inbound: timeout")
+            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}")
@@ -390,11 +403,6 @@ class Scheduler:
         result.finished_at = datetime.now(timezone.utc)
         return result
 
-    def _list_inbound(self) -> list[InboundFile]:
-        """Return files in the inbound MFT path. Runs on a thread."""
-        client = self._sftp_client_factory(self._sftp_block)
-        return client.list_inbound()
-
     async def _handle_one(self, f: InboundFile, result: TickResult) -> None:
         """Process one inbound file: skip-if-seen, classify, parse, record."""
         if await self._already_processed(f.name):