feat(sp27): create handlers/ package skeleton + dedup ack ID helpers

CycloneStore split precedent (SP21) — lift three helpers from
scheduler.py + api.py into one module. Both callers will switch
to this in Task 6. The new package also defines HandleResult
and the HANDLERS registry; handle_999 / handle_ta1 / handle_277ca
/ handle_835 fill in over Tasks 2-5.

The registry is lazy + best-effort: register_handlers() catches
broad Exception so a partial-modification SyntaxError in one
in-flight handler module can't break scheduler or API import.

11 tests added; existing suite unchanged (1 failed + 2 errors
pre-existing in test_provider_extended_response.py are not
introduced by this commit).

backend/src/cyclone/handlers/__init__.py

@@ -0,0 +1,113 @@
+"""File-type handlers for inbound MFT files (SP27).
+
+Each handler is a pure function that parses + persists + dispatches
+events for one file type. The scheduler and the FastAPI endpoints
+both delegate here; the ``HANDLERS`` registry maps ``file_type`` →
+handler function.
+
+Public API:
+    HandleResult          — dataclass returned by every handler
+    HANDLERS              — ``{"999": handle_999, "835": handle_835, ...}``
+    handle_999, handle_ta1, handle_277ca, handle_835
+        — call signatures: ``handle(text: str, source_file: str) -> HandleResult``
+
+The handlers own their own DB session lifecycle. They emit pubsub
+events via the optional ``event_bus`` parameter (the FastAPI endpoint
+injects ``app.state.event_bus``; the scheduler passes ``None``).
+They never raise on per-segment problems; per-segment issues are
+logged and folded into the result. Whole-document failures
+(missing ISA, bad encoding) surface as ``CycloneParseError``, which
+the caller catches and records as ``STATUS_ERROR``.
+"""
+from __future__ import annotations
+
+import importlib
+import logging
+
+from ._ack_id import (
+    ack_count_summary,
+    ack_synthetic_source_batch_id,
+    two77ca_synthetic_source_batch_id,
+)
+from .handle_result import HandleResult
+
+log = logging.getLogger(__name__)
+
+# Module-level HANDLERS dict populated lazily once handler modules
+# ship. Keys are file_type strings, values are the ``handle``
+# callable for that type.
+HANDLERS: dict[str, object] = {}
+
+
+# Candidate handlers, in registration order. Each tuple is
+# (module-path, file_type). The 277/277CA mapping is added explicitly
+# after registration when the 277CA handler is present.
+_CANDIDATES: list[tuple[str, str]] = [
+    ("cyclone.handlers.handle_999", "999"),
+    ("cyclone.handlers.handle_ta1", "TA1"),
+    ("cyclone.handlers.handle_277ca", "277CA"),
+    ("cyclone.handlers.handle_835", "835"),
+]
+
+
+def register_handlers() -> None:
+    """Populate ``HANDLERS`` from the per-type handler modules.
+
+    Tolerates missing or broken handler modules so the package can be
+    imported incrementally as each handler ships (Tasks 2-5), and so
+    a partial-modification error in one handler module can't break
+    scheduler / API import. Safe to call multiple times.
+    """
+    if HANDLERS:
+        return
+    for mod_path, file_type in _CANDIDATES:
+        try:
+            mod = importlib.import_module(mod_path)
+            fn = getattr(mod, "handle")
+        except Exception as exc:  # noqa: BLE001 — best-effort registry
+            log.debug(
+                "handler %s unavailable: %s",
+                mod_path, exc,
+            )
+            continue
+        HANDLERS[file_type] = fn
+        # The 277 filename maps to the same 277CA handler.
+        if file_type == "277CA":
+            HANDLERS["277"] = fn
+
+
+register_handlers()
+
+
+# Re-export handler functions on this package so callers can use the
+# flat import (``from cyclone.handlers import handle_999``) once each
+# module ships. Set after registration so we know what's present.
+def _reexport_handlers() -> None:
+    """Re-export each handler's ``handle`` fn as ``handle_<type>``.
+
+    No-op for absent handlers. Re-run on each import so freshly-
+    installed handler modules (e.g. Tasks 2-5 commits) are visible
+    after ``register_handlers()`` without a process restart.
+    """
+    for file_type, fn in list(HANDLERS.items()):
+        if file_type in ("277",):  # alias of 277CA; don't re-export twice
+            continue
+        globals()[f"handle_{file_type.lower()}"] = fn
+
+
+_reexport_handlers()
+
+
+__all__ = [
+    "HANDLERS",
+    "HandleResult",
+    "register_handlers",
+    "ack_count_summary",
+    "ack_synthetic_source_batch_id",
+    "two77ca_synthetic_source_batch_id",
+    # handler_* names are added at module load via _reexport_handlers
+]
+# Populate __all__ with the present handler symbols.
+for _h in ("handle_999", "handle_ta1", "handle_277ca", "handle_835"):
+    if _h in globals():
+        __all__.append(_h)  # noqa: PYI056

backend/src/cyclone/handlers/_ack_id.py

@@ -0,0 +1,87 @@
+"""Ack ID helpers shared between the scheduler and the FastAPI
+endpoints (SP27 Task 1).
+
+Lifted from ``scheduler.py:_ack_count_summary`` +
+``_ack_synthetic_source_batch_id`` + ``_277ca_synthetic_source_batch_id`` —
+all three had inline copies in both ``scheduler.py`` and ``api.py``
+prior to this SP. Both callers now import from this module.
+
+Helpers
+-------
+``ack_count_summary(result)``
+    Aggregate ``(received, accepted, rejected, ack_code)`` from a
+    parsed ``ParseResult999``, trusting set-level IK5 over the
+    functional-group AK9. Gainwell's MFT ships AK9 segments that
+    contradict the per-set IK5 (e.g. ``AK9*A*1*1*1`` with
+    ``IK5*A``), so trusting AK9's rejected count over-reports.
+    See scheduler commit ``6507a8c`` for the operational context.
+
+``ack_synthetic_source_batch_id(icn, *, pcn, source_filename)``
+    Build a unique-per-file ``batches.id`` for a 999 that ships
+    without its own source batch. Falls back to a hash suffix of
+    the filename so daily pulls don't all collapse onto the same
+    ICN. Gainwell's MFT ships every 999 with the default ICN
+    ``000000001`` (per the scheduler docstring).
+
+``two77ca_synthetic_source_batch_id(icn)``
+    Same idea for a 277CA without its own source batch.
+"""
+from __future__ import annotations
+
+import hashlib
+from typing import Any
+
+
+def ack_count_summary(result: Any) -> tuple[int, int, int, str]:
+    """Aggregate ``(received, accepted, rejected, ack_code)`` from
+    a ``ParseResult999``.
+
+    Counts are derived from the set-level ``IK5`` responses
+    (one per ``AK2`` in the 999), not the functional-group ``AK9``.
+    Gainwell's MFT ships contradictory AK9 segments; the per-set
+    IK5 is the authoritative per-claim accept/reject signal.
+    """
+    sets = result.set_responses
+    received = len(sets)
+    accepted = sum(1 for s in sets if s.set_accept_reject.code == "A")
+    rejected = received - accepted
+    if rejected == 0:
+        code = "A"
+    elif accepted == 0:
+        code = "R"
+    else:
+        code = "P"
+    return (received, accepted, rejected, code)
+
+
+def ack_synthetic_source_batch_id(
+    interchange_control_number: str,
+    *,
+    pcn: str | None = None,
+    source_filename: str | None = None,
+) -> str:
+    """Synthetic ``batches.id`` for a received 999 with no source batch.
+
+    Precedence for the human-readable part of the id (column is
+    ``VARCHAR(32)``):
+
+      1. ``999-{pcn}-{hash8}`` if ``AK2`` set_control_number is
+         present (the common case). 4 + 9 + 1 + 8 = 22 chars max.
+      2. ``999-{icn}-{hash8}`` if no AK2 (envelope-only 999).
+      3. ``999-{hash12}`` if no filename either (shouldn't happen
+         in production).
+    """
+    short_hash = ""
+    if source_filename:
+        short_hash = hashlib.sha1(source_filename.encode("utf-8")).hexdigest()[:8]
+    if pcn and pcn.strip():
+        return f"999-{pcn.strip()}-{short_hash}"
+    icn = (interchange_control_number or "").strip() or "000000001"
+    if short_hash:
+        return f"999-{icn}-{short_hash}"
+    return f"999-{short_hash or icn}"
+
+
+def two77ca_synthetic_source_batch_id(interchange_control_number: str) -> str:
+    """Synthetic ``batches.id`` for a received 277CA with no source batch."""
+    return f"277CA-{(interchange_control_number or '').strip() or '000000001'}"

backend/src/cyclone/handlers/handle_result.py

@@ -0,0 +1,37 @@
+"""Shared ``HandleResult`` dataclass for handlers (SP27).
+
+Every per-file-type handler returns the same shape so the
+scheduler's ``_handle_one`` and the FastAPI parse endpoints can
+process them uniformly.
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+
+@dataclass
+class HandleResult:
+    """Outcome of one handler invocation.
+
+    Attributes:
+        parser_used: The parser name written to
+            ``processed_inbound_files.parser_used`` (e.g. ``"parse_999"``)
+            and surfaced in the UI.
+        claim_count: The number of claim rows persisted (or batch
+            records, depending on the file type). For 999/TA1 this
+            is the receipt count; for 835 this is the per-claim
+            remittance count; for 277CA this is the per-claim
+            status count.
+        batch_id: The persisted batch id (when the handler creates a
+            row in ``batches``). ``None`` for handlers that persist
+            into per-file-type ack tables (999/TA1/277CA) rather
+            than into the unified ``batches`` table.
+        matched_count: For 835, the number of remits that were
+            matched to an existing claim by ``reconcile.match``.
+            Zero for other handlers.
+    """
+
+    parser_used: str
+    claim_count: int
+    batch_id: str | None = None
+    matched_count: int = 0