diff --git a/backend/src/cyclone/handlers/__init__.py b/backend/src/cyclone/handlers/__init__.py
new file mode 100644
index 0000000..784179c
--- /dev/null
+++ b/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
diff --git a/backend/src/cyclone/handlers/_ack_id.py b/backend/src/cyclone/handlers/_ack_id.py
new file mode 100644
index 0000000..7fb08ee
--- /dev/null
+++ b/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'}"
diff --git a/backend/src/cyclone/handlers/handle_result.py b/backend/src/cyclone/handlers/handle_result.py
new file mode 100644
index 0000000..58dc24f
--- /dev/null
+++ b/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
diff --git a/backend/tests/test_ack_id_helpers.py b/backend/tests/test_ack_id_helpers.py
new file mode 100644
index 0000000..b8adc9a
--- /dev/null
+++ b/backend/tests/test_ack_id_helpers.py
@@ -0,0 +1,142 @@
+"""Tests for the dedup-ed ack ID helpers (SP27 Task 1).
+
+Locks the contract for the helpers that ``scheduler.py`` + ``api.py``
+will both import from one place in ``cyclone.handlers._ack_id``.
+"""
+from __future__ import annotations
+
+import hashlib
+from pathlib import Path
+
+from cyclone.handlers._ack_id import (
+    ack_count_summary,
+    ack_synthetic_source_batch_id,
+    two77ca_synthetic_source_batch_id,
+)
+from cyclone.parsers.parse_999 import parse_999_text
+
+
+ACCEPTED = Path(__file__).parent / "fixtures" / "minimal_999.txt"
+REJECTED = Path(__file__).parent / "fixtures" / "minimal_999_rejected.txt"
+
+
+def _parse(path: Path):
+    return parse_999_text(path.read_text(), input_file=path.name)
+
+
+def test_ack_count_summary_all_accepted():
+    result = _parse(ACCEPTED)
+    recv, acc, rej, code = ack_count_summary(result)
+    assert recv == 1
+    assert acc == 1
+    assert rej == 0
+    assert code == "A"
+
+
+def test_ack_count_summary_all_rejected():
+    result = _parse(REJECTED)
+    recv, acc, rej, code = ack_count_summary(result)
+    assert recv == 1
+    assert acc == 0
+    assert rej == 1
+    assert code == "R"
+
+
+def test_ack_count_summary_partial_uses_p_code():
+    # Synthesize a 2-set 999 inline with one AK5=A and one AK5=R.
+    # ISA is exactly 106 chars (positions 0-105); the parser slices
+    # positionally then splits the body on ``~``.
+    # SE count = ST + AK1 + AK2_1 + AK5_1 + AK2_2 + AK5_2 + AK9 + SE = 8.
+    partial_999 = (
+        "ISA*00*          *00*          *ZZ*SUBMITTERID    *ZZ*RECEIVERID     "
+        "*240101*1200*^*00501*000000001*0*P*:~"
+        "GS*FA*SUBMITTERID*RECEIVERID*20240101*1200*1*X*005010X231A1~"
+        "ST*999*0001*005010X231A1~"
+        "AK1*HC*0001*1*2~"
+        "AK2*837*1~"
+        "AK5*A~"
+        "AK2*837*2~"
+        "AK5*R~"
+        "AK9*P*2*1*1~"
+        "SE*8*0001~"
+        "GE*1*1~"
+        "IEA*1*000000001~"
+    )
+    result = parse_999_text(partial_999, input_file="partial.999")
+    recv, acc, rej, code = ack_count_summary(result)
+    assert recv == 2
+    assert acc == 1
+    assert rej == 1
+    assert code == "P"
+
+
+def test_ack_synthetic_with_pcn_and_filename_is_deterministic_and_prefix():
+    bsid1 = ack_synthetic_source_batch_id(
+        "000000001", pcn="PCN-12345", source_filename="tp_999.x12",
+    )
+    bsid2 = ack_synthetic_source_batch_id(
+        "000000001", pcn="PCN-12345", source_filename="tp_999.x12",
+    )
+    assert bsid1 == bsid2, "same inputs must produce same id"
+    assert bsid1.startswith("999-PCN-12345-")
+    # Hash is 8 hex chars
+    suffix = bsid1.split("-")[-1]
+    assert len(suffix) == 8 and all(
+        c in "0123456789abcdef" for c in suffix
+    ), suffix
+
+
+def test_ack_synthetic_with_pcn_strips_whitespace():
+    bsid = ack_synthetic_source_batch_id(
+        "000000001", pcn="  PCN  ", source_filename="f.x12",
+    )
+    assert bsid.startswith("999-PCN-")
+
+
+def test_ack_synthetic_without_pcn_uses_icn():
+    bsid = ack_synthetic_source_batch_id(
+        "777777777", pcn=None, source_filename="f.x12",
+    )
+    assert bsid.startswith("999-777777777-")
+    # <= 32 chars total (VARCHAR(32) constraint)
+    assert len(bsid) <= 32
+
+
+def test_ack_synthetic_default_icn_when_empty():
+    bsid = ack_synthetic_source_batch_id("", pcn=None, source_filename=None)
+    # No PCN, no filename → falls through to default ICN ``000000001``
+    # (no hash suffix without a filename).
+    assert bsid == "999-000000001"
+
+
+def test_two77ca_synthetic_uses_icn():
+    assert two77ca_synthetic_source_batch_id("000012345") == "277CA-000012345"
+
+
+def test_two77ca_synthetic_empty_falls_back_to_default():
+    assert two77ca_synthetic_source_batch_id("") == "277CA-000000001"
+    assert two77ca_synthetic_source_batch_id(None) == "277CA-000000001"
+
+
+def test_ack_count_summary_empty_set_responses_returns_zeros():
+    """A 999 envelope without any AK2/IK5 sets is malformed per the spec,
+    but the helper must still return a sane (0, 0, 0, 'A') tuple — the
+    scheduler adds it to result.errors and the caller decides."""
+    class _StubResult:
+        set_responses = []
+    recv, acc, rej, code = ack_count_summary(_StubResult())
+    assert recv == 0
+    assert acc == 0
+    assert rej == 0
+    # rejected == 0 → "A" (no rejection signal in this degenerate case)
+    assert code == "A"
+
+
+def test_ack_synthetic_uses_real_sha1_truncation():
+    """The hash should match hashlib.sha1(filename).hexdigest()[:8] so
+    a future migration to a different hash is intentional, not drift."""
+    expected = hashlib.sha1(b"my-file.x12").hexdigest()[:8]
+    bsid = ack_synthetic_source_batch_id(
+        "000000001", pcn=None, source_filename="my-file.x12",
+    )
+    assert bsid.endswith(f"-{expected}")