diff --git a/backend/src/cyclone/claim_acks.py b/backend/src/cyclone/claim_acks.py
new file mode 100644
index 0000000..022a9ba
--- /dev/null
+++ b/backend/src/cyclone/claim_acks.py
@@ -0,0 +1,478 @@
+"""SP28: per-ACK auto-linker.
+
+Three helpers, one per ACK kind, all run inside the same DB
+transaction that persists the Ack row. Each returns a slice of
+``ClaimAckLinkResult`` describing what was linked / orphaned / skipped.
+Plus :func:`lookup_claims_for_ack_set_response` (the two-pass join from
+spec D10) and :func:`link_manual` (the manual-fallback used by
+``/api/acks/{kind}/{id}/match-claim``).
+
+The module is pure (no DB session ownership — callers pass in the
+session, just like ``cyclone.inbox_state.apply_999_rejections``). The
+helpers emit ``ClaimAck`` ORM rows via ``add`` + ``flush``, and the
+caller commits.
+
+See ``docs/superpowers/specs/2026-07-02-cyclone-ack-claim-auto-link-design.md``
+§3 for the per-AK2 granularity, the two-pass join (ST02 via batch
+envelope index primary + Claim.patient_control_number fallback), and
+the idempotency contract enforced by ``ux_claim_acks_dedup``.
+"""
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass, field
+from datetime import datetime, timezone
+from typing import Callable, Iterable, Optional
+
+from sqlalchemy.orm import Session
+
+from cyclone.db import Batch, Claim, ClaimAck
+
+
+log = logging.getLogger(__name__)
+
+
+@dataclass
+class ClaimAckLinkResult:
+    """Outcome of a single helper call.
+
+    ``linked`` is a list of ``(claim_id, ak2_index)`` tuples — the link
+    rows actually written. ``ak2_index`` is ``None`` for 277CA / TA1
+    links (no per-segment index in those ack kinds) and an int for
+    999. ``orphans`` is a list of ``set_control_number`` strings the
+    join couldn't resolve to any claim (or, for TA1, the ICN when no
+    matching batch was found).
+    """
+
+    linked: list[tuple[Optional[str], Optional[int]]] = field(default_factory=list)
+    orphans: list[str] = field(default_factory=list)
+
+
+# ---------------------------------------------------------------------------
+# D10 two-pass join
+# ---------------------------------------------------------------------------
+
+
+def lookup_claims_for_ack_set_response(
+    session: Session,
+    set_control_number: str,
+    *,
+    batch_envelope_index: Optional[Callable[[str], Optional[str]]] = None,
+    pc_claim_lookup: Optional[Callable[[str], Optional[Claim]]] = None,
+) -> list[Claim]:
+    """Two-pass join for a single AK2 set_response / 277CA claim_status.
+
+    D10 (spec): for a 999 AK2-2 ``set_control_number`` or a 277CA REF*1K
+    ``payer_claim_control_number``, return every claim this ack
+    acknowledges. The primary join is
+    ``Batch.envelope.control_number == set_control_number`` (== source
+    837's ST02 on Gainwell batches); the fallback is
+    ``Claim.patient_control_number == set_control_number``.
+
+    Returns 0..N matching claims (one-ack-to-many when one 837 batch
+    shipped multiple claims under one ST02).
+
+    Args:
+        session: SQLAlchemy session the caller owns.
+        set_control_number: the AK2-2 / REF*1K value to resolve.
+        batch_envelope_index: optional pre-built index that maps
+            ``Batch.envelope.control_number`` → ``batch.id`` (built
+            once per ingest via ``store.batch_envelope_index``).
+            Pass to skip the per-set-response ``Batch`` scan in Pass 1.
+        pc_claim_lookup: optional pre-built callable that maps a PCN
+            to a single claim. Falls back to a session-wide query
+            when not supplied.
+
+    Pass 1 wins. The two paths cannot both fire for the same
+    ``set_control_number`` — if Pass 1 returns one or more claims,
+    Pass 2 is skipped. This is the false-positive guard from
+    spec §7.
+    """
+    if not set_control_number:
+        return []
+
+    # -- Pass 1: Batch.envelope.control_number primary --------------
+    matched_ids: list[str] = []
+    if batch_envelope_index is not None:
+        batch_id = batch_envelope_index(set_control_number)
+        if batch_id is not None:
+            matched_ids = [
+                cid for (cid,) in (
+                    session.query(Claim.id)
+                    .filter(Claim.batch_id == batch_id)
+                    .all()
+                )
+            ]
+    else:
+        # Fallback: scan all batches once per call. Slow but correct;
+        # callers SHOULD pass the index.
+        rows = (
+            session.query(Batch.id, Batch.raw_result_json)
+            .filter(Batch.kind == "837")
+            .all()
+        )
+        for bid, raw in rows:
+            env = (raw or {}).get("envelope") or {}
+            if env.get("control_number") == set_control_number:
+                matched_ids = [
+                    cid for (cid,) in (
+                        session.query(Claim.id)
+                        .filter(Claim.batch_id == bid)
+                        .all()
+                    )
+                ]
+                break
+
+    if matched_ids:
+        claims = (
+            session.query(Claim)
+            .filter(Claim.id.in_(matched_ids))
+            .all()
+        )
+        if claims:
+            return list(claims)
+
+    # -- Pass 2: Claim.patient_control_number fallback ---------------
+    if pc_claim_lookup is not None:
+        single = pc_claim_lookup(set_control_number)
+        if single is not None:
+            return [single]
+        return []
+
+    matches = (
+        session.query(Claim)
+        .filter(Claim.patient_control_number == set_control_number)
+        .all()
+    )
+    return list(matches)
+
+
+# ---------------------------------------------------------------------------
+# Per-ACK helpers — walk the parsed result and create one ClaimAck per
+# matched claim. Idempotent via dedup pre-check + ux_claim_acks_dedup.
+# ---------------------------------------------------------------------------
+
+
+def _link_exists(
+    session: Session,
+    *,
+    claim_id: str,
+    ack_kind: str,
+    ack_id: int,
+    ak2_index: Optional[int],
+) -> bool:
+    """True when a link row already exists for this dedup key.
+
+    Mirrors the partial unique index
+    ``ux_claim_acks_dedup(claim_id, ack_kind, ack_id, ak2_index)
+    WHERE claim_id IS NOT NULL AND ak2_index IS NOT NULL``. For TA1 /
+    277CA where ``ak2_index`` is NULL we still want to dedup at
+    ``(claim_id, ack_kind, ack_id)`` (the Inbox ack-orphans lane
+    shouldn't show the same PCN twice); a second narrower query
+    covers that case so a re-ingest of the same file is a no-op.
+    """
+    if ak2_index is not None:
+        return (
+            session.query(ClaimAck.id)
+            .filter(
+                ClaimAck.claim_id == claim_id,
+                ClaimAck.ack_kind == ack_kind,
+                ClaimAck.ack_id == ack_id,
+                ClaimAck.ak2_index == ak2_index,
+            )
+            .first()
+            is not None
+        )
+    return (
+        session.query(ClaimAck.id)
+        .filter(
+            ClaimAck.claim_id == claim_id,
+            ClaimAck.ack_kind == ack_kind,
+            ClaimAck.ack_id == ack_id,
+            ClaimAck.ak2_index.is_(None),
+        )
+        .first()
+        is not None
+    )
+
+
+def apply_999_acceptances(
+    session: Session,
+    parsed_999,
+    *,
+    ack_id: int,
+    batch_envelope_index: Optional[Callable[[str], Optional[str]]] = None,
+    pc_claim_lookup: Optional[Callable[[str], Optional[Claim]]] = None,
+    now: Optional[datetime] = None,
+) -> ClaimAckLinkResult:
+    """For every AK2 set-response, create one ``ClaimAck`` row per matched claim.
+
+    Both accepted AND rejected AK2s get a link row (so the
+    ClaimDrawer panel can show the rejection inline via the
+    ``set_accept_reject_code`` color-coded chip). Orphans are returned
+    but not linked.
+
+    Idempotent: if a ``(claim_id, '999', ack_id, ak2_index=i)`` row
+    already exists, the duplicate is skipped silently — DB-level
+    enforcement via ``ux_claim_acks_dedup``; the pre-check is here to
+    avoid IntegrityError log noise.
+
+    One AK2 can produce multiple ``claim_acks`` rows when the source
+    837 batch carried more than one claim under a shared ST02 (rare
+    on this codebase but supported by D10 / the schema).
+    """
+    result = ClaimAckLinkResult()
+    ts = now or datetime.now(timezone.utc)
+    set_responses = getattr(parsed_999, "set_responses", None) or []
+
+    for idx, sr in enumerate(set_responses):
+        scn = getattr(sr, "set_control_number", None) or ""
+        code = getattr(sr.set_accept_reject, "code", None) or ""
+        claims = lookup_claims_for_ack_set_response(
+            session, scn,
+            batch_envelope_index=batch_envelope_index,
+            pc_claim_lookup=pc_claim_lookup,
+        )
+        if not claims:
+            result.orphans.append(scn)
+            continue
+        for claim in claims:
+            if _link_exists(
+                session,
+                claim_id=claim.id,
+                ack_kind="999",
+                ack_id=ack_id,
+                ak2_index=idx,
+            ):
+                continue
+            session.add(ClaimAck(
+                claim_id=claim.id,
+                batch_id=None,
+                ack_id=ack_id,
+                ack_kind="999",
+                ak2_index=idx,
+                set_control_number=scn,
+                set_accept_reject_code=code,
+                linked_at=ts,
+                linked_by="auto",
+            ))
+            result.linked.append((claim.id, idx))
+
+    if result.linked:
+        session.flush()
+    return result
+
+
+def apply_277ca_acks(
+    session: Session,
+    parsed_277ca,
+    *,
+    ack_id: int,
+    batch_envelope_index: Optional[Callable[[str], Optional[str]]] = None,
+    pc_claim_lookup: Optional[Callable[[str], Optional[Claim]]] = None,
+    now: Optional[datetime] = None,
+) -> ClaimAckLinkResult:
+    """For every ClaimStatus with a payer_claim_control_number, create a ClaimAck.
+
+    Accepted AND rejected ClaimStatuses both link — the
+    ``set_accept_reject_code`` carries the STC category code. The
+    ``claim_acks`` row is independent of the existing
+    ``Claim.payer_rejected_at`` mutation from
+    :func:`cyclone.inbox_state_277ca.apply_277ca_rejections` (which
+    fires before this helper in the handler).
+
+    Idempotent via dedup key ``(claim_id, '277ca', ack_id, NULL)``.
+
+    Args:
+        parsed_277ca: a :class:`cyclone.parsers.models_277ca.ParseResult277CA`.
+    """
+    result = ClaimAckLinkResult()
+    ts = now or datetime.now(timezone.utc)
+    statuses = getattr(parsed_277ca, "claim_statuses", None) or []
+
+    for status in statuses:
+        scn = getattr(status, "payer_claim_control_number", None) or ""
+        code = (
+            getattr(getattr(status, "status_code", None), "__class__", None)
+            and (status.status_code or "")
+        )
+        # ``status_code`` may be a category like "A6:19:PR" — keep the
+        # whole STC composite so the UI can render the category without
+        # re-parsing raw_json. Truncate to 8 chars (column width).
+        code = (code or "")[:8]
+        if not scn:
+            # No REF*1K — orphan. Surface the STC composite so the
+            # operator can correlate via the ack's raw_json.
+            orphan_key = code or "(no REF*1K)"
+            result.orphans.append(orphan_key)
+            continue
+        claims = lookup_claims_for_ack_set_response(
+            session, scn,
+            batch_envelope_index=batch_envelope_index,
+            pc_claim_lookup=pc_claim_lookup,
+        )
+        if not claims:
+            result.orphans.append(scn)
+            continue
+        for claim in claims:
+            if _link_exists(
+                session,
+                claim_id=claim.id,
+                ack_kind="277ca",
+                ack_id=ack_id,
+                ak2_index=None,
+            ):
+                continue
+            session.add(ClaimAck(
+                claim_id=claim.id,
+                batch_id=None,
+                ack_id=ack_id,
+                ack_kind="277ca",
+                ak2_index=None,
+                set_control_number=scn,
+                set_accept_reject_code=code or None,
+                linked_at=ts,
+                linked_by="auto",
+            ))
+            result.linked.append((claim.id, None))
+
+    if result.linked:
+        session.flush()
+    return result
+
+
+def apply_ta1_envelope_link(
+    session: Session,
+    parsed_ta1,
+    *,
+    ack_id: int,
+    batch_lookup: Callable[[str, str], Optional[Batch]],
+    now: Optional[datetime] = None,
+) -> ClaimAckLinkResult:
+    """Link a TA1 to the most-recent Batch whose sender_id/receiver_id match.
+
+    TA1 is envelope-level only (ISA/IEA, no per-claim granularity).
+    The link row has ``claim_id IS NULL`` and ``batch_id`` populated.
+
+    Args:
+        parsed_ta1: a :class:`cyclone.parsers.models_ta1.ParseResultTa1`.
+        batch_lookup: ``(sender_id, receiver_id) -> Batch | None``.
+            The handler supplies a closure that calls
+            ``session.query(Batch).order_by(parsed_at.desc()).first()``.
+            Returning ``None`` produces an orphan (no batch match).
+
+    Idempotent via dedup key ``(None, 'ta1', ack_id, NULL)`` — since
+    ``claim_id IS NULL`` the partial unique index does not catch it
+    and we have to fall back to a Python-level check. Because TA1 is
+    a singleton per file this is cheap.
+    """
+    result = ClaimAckLinkResult()
+    ts = now or datetime.now(timezone.utc)
+    envelope = getattr(parsed_ta1, "envelope", None)
+    if envelope is None:
+        return result
+    batch = batch_lookup(envelope.sender_id or "", envelope.receiver_id or "")
+    ta1_obj = getattr(parsed_ta1, "ta1", None)
+    ack_code = getattr(ta1_obj, "ack_code", None) or ""
+
+    # Dedup: same (ack_kind, ack_id) → at most one TA1 envelope link.
+    existing = (
+        session.query(ClaimAck.id)
+        .filter(
+            ClaimAck.ack_kind == "ta1",
+            ClaimAck.ack_id == ack_id,
+            ClaimAck.claim_id.is_(None),
+        )
+        .first()
+    )
+    if existing is not None:
+        return result
+
+    if batch is None:
+        orphan_key = (
+            getattr(ta1_obj, "control_number", None)
+            or envelope.control_number
+            or ""
+        )
+        result.orphans.append(orphan_key)
+        return result
+
+    session.add(ClaimAck(
+        claim_id=None,
+        batch_id=batch.id,
+        ack_id=ack_id,
+        ack_kind="ta1",
+        ak2_index=None,
+        set_control_number=None,
+        set_accept_reject_code=ack_code,
+        linked_at=ts,
+        linked_by="auto",
+    ))
+    session.flush()
+    result.linked.append((None, None))
+    return result
+
+
+def link_manual(
+    session: Session,
+    *,
+    claim_id: str,
+    ack_kind: str,
+    ack_id: int,
+    set_control_number: Optional[str] = None,
+    set_accept_reject_code: Optional[str] = None,
+    ak2_index: Optional[int] = None,
+    now: Optional[datetime] = None,
+) -> tuple[ClaimAck, bool]:
+    """Create one manual link row, or return the existing one if already linked.
+
+    Used by ``POST /api/acks/{kind}/{ack_id}/match-claim``. Returns
+    ``(row, created)`` where ``created=False`` means the dedup check
+    found an existing link (idempotent re-call → 200, not 201).
+
+    Raises ``LookupError`` when the referenced claim doesn't exist
+    (the caller maps that to 404).
+    """
+    if ack_kind not in ("999", "277ca", "ta1"):
+        raise ValueError(f"link_manual: unknown ack_kind={ack_kind!r}")
+    claim = session.get(Claim, claim_id)
+    if claim is None:
+        raise LookupError(f"claim {claim_id} not found")
+
+    existing = (
+        session.query(ClaimAck)
+        .filter(
+            ClaimAck.ack_kind == ack_kind,
+            ClaimAck.ack_id == ack_id,
+            ClaimAck.claim_id == claim_id,
+        )
+        .first()
+    )
+    if existing is not None:
+        return existing, False
+
+    ts = now or datetime.now(timezone.utc)
+    row = ClaimAck(
+        claim_id=claim_id,
+        batch_id=None,
+        ack_id=ack_id,
+        ack_kind=ack_kind,
+        ak2_index=ak2_index,
+        set_control_number=set_control_number,
+        set_accept_reject_code=set_accept_reject_code,
+        linked_at=ts,
+        linked_by="manual",
+    )
+    session.add(row)
+    session.flush()
+    return row, True
+
+
+__all__ = [
+    "ClaimAckLinkResult",
+    "apply_999_acceptances",
+    "apply_277ca_acks",
+    "apply_ta1_envelope_link",
+    "link_manual",
+    "lookup_claims_for_ack_set_response",
+]
diff --git a/backend/tests/test_apply_claim_ack_links.py b/backend/tests/test_apply_claim_ack_links.py
new file mode 100644
index 0000000..ef3034b
--- /dev/null
+++ b/backend/tests/test_apply_claim_ack_links.py
@@ -0,0 +1,787 @@
+"""Tests for :mod:`cyclone.claim_acks` (SP28 auto-linker).
+
+Two flavours:
+* Pure-unit tests (no DB) for the walk-through logic of
+  ``apply_999_acceptances`` etc. via stubs.
+* Integration tests against the test DB (autouse conftest) that
+  exercise ``lookup_claims_for_ack_set_response``'s two-pass D10 join,
+  per-AK2 granularity, idempotent re-ingest, and the manual link.
+
+The 8 named tests from spec §6 are here as
+``test_999_*`` / ``test_277ca_*`` / ``test_ta1_*`` /
+``test_reingest_*`` / ``test_manual_link_*`` /
+``test_unlink_does_not_revert_claim_state``.
+
+Additional tests cover the D10 two-pass join (Step 2.5 of the plan)
+and the store facade wiring.
+"""
+from __future__ import annotations
+
+from datetime import date, datetime, timezone
+from decimal import Decimal
+
+import pytest
+
+from cyclone import claim_acks as ca
+from cyclone import db
+from cyclone.claim_acks import (
+    ClaimAckLinkResult,
+    apply_277ca_acks,
+    apply_999_acceptances,
+    apply_ta1_envelope_link,
+    link_manual,
+    lookup_claims_for_ack_set_response,
+)
+from cyclone.db import (
+    Batch,
+    Claim,
+    ClaimState,
+    ClaimAck,
+)
+from cyclone.parsers.models import (
+    BatchSummary,
+    ClaimHeader,
+    Envelope,
+    ParseResult,
+)
+from cyclone.parsers.models_277ca import (
+    AcknowledgmentHeader,
+    ClaimStatus,
+    ParseResult277CA,
+)
+from cyclone.parsers.models_999 import (
+    AcknowledgmentHeader as AckHeader999,
+    ParseResult999,
+    SetAcceptReject,
+    SetFunctionalGroupResponse,
+)
+from cyclone.parsers.models_ta1 import ParseResultTa1, Ta1Ack
+from cyclone.store import store
+
+
+# ---------------------------------------------------------------------------
+# Fixture seeds
+# ---------------------------------------------------------------------------
+
+
+def _seed_batch(
+    session,
+    *,
+    batch_id: str = "B1",
+    envelope_control: str = "991102989",
+    sender_id: str = "SENDER",
+    receiver_id: str = "RECEIVER",
+):
+    """Insert one Batch row whose raw_result_json envelope carries
+    ``envelope_control`` so D10 Pass 1 can resolve it via
+    ``batch_envelope_index``."""
+    b = Batch(
+        id=batch_id,
+        kind="837",
+        input_filename=f"{batch_id}.txt",
+        parsed_at=datetime(2026, 7, 2, 12, 0, tzinfo=timezone.utc),
+        raw_result_json={
+            "envelope": {
+                "sender_id": sender_id,
+                "receiver_id": receiver_id,
+                "control_number": envelope_control,
+                "transaction_date": "2026-07-02",
+            },
+        },
+    )
+    session.add(b)
+    session.commit()
+    session.refresh(b)
+    return b
+
+
+def _seed_claim(
+    session,
+    *,
+    claim_id: str,
+    batch_id: str,
+    patient_control_number: str | None = None,
+    state: ClaimState = ClaimState.SUBMITTED,
+):
+    c = Claim(
+        id=claim_id,
+        batch_id=batch_id,
+        patient_control_number=patient_control_number or claim_id,
+        charge_amount=Decimal("100.00"),
+        state=state,
+    )
+    session.add(c)
+    session.commit()
+    session.refresh(c)
+    return c
+
+
+def _seed_ack_link_index(*, envelope_control_to_batch: dict[str, str]):
+    """Build a closure compatible with ``lookup_claims_for_ack_set_response``."""
+
+    def _lookup(scn: str) -> str | None:
+        return envelope_control_to_batch.get(scn)
+
+    return _lookup
+
+
+def _parse_999_two_ak2s() -> ParseResult999:
+    """Two-AK2 999 — one accepted, one rejected."""
+    sr_accepted = SetFunctionalGroupResponse(
+        ak2=AckHeader999(functional_id_code="837", group_control_number="991102989"),
+        set_control_number="991102989",
+        transaction_set_identifier="837",
+        segment_errors=[],
+        set_accept_reject=SetAcceptReject(code="A"),
+    )
+    sr_rejected = SetFunctionalGroupResponse(
+        ak2=AckHeader999(functional_id_code="837", group_control_number="991102990"),
+        set_control_number="991102990",
+        transaction_set_identifier="837",
+        segment_errors=[],
+        set_accept_reject=SetAcceptReject(code="R"),
+    )
+    return ParseResult999(
+        envelope=Envelope(
+            sender_id="PAYER", receiver_id="SUBMITTER",
+            control_number="000000001", transaction_date=date(2026, 7, 2),
+        ),
+        functional_group_acks=[],
+        set_responses=[sr_accepted, sr_rejected],
+        summary=BatchSummary(
+            input_file="two_ak2.txt", control_number="000000001",
+            transaction_date=date(2026, 7, 2),
+            total_claims=2, passed=1, failed=1,
+        ),
+    )
+
+
+def _parse_277ca_one_accepted() -> ParseResult277CA:
+    return ParseResult277CA(
+        envelope=Envelope(
+            sender_id="PAYER", receiver_id="SUBMITTER",
+            control_number="000000077", transaction_date=date(2026, 7, 2),
+        ),
+        bht=AcknowledgmentHeader(
+            hierarchical_structure_code="0085",
+            transaction_set_purpose_code="08",
+            reference_identification="REFNUM001",
+        ),
+        summary=BatchSummary(
+            input_file="277_one_accepted.txt", control_number="000000077",
+            transaction_date=date(2026, 7, 2),
+            total_claims=1, passed=1, failed=0,
+        ),
+        claim_statuses=[
+            ClaimStatus(
+                status_code="A3:19:PR",
+                classification="accepted",
+                payer_claim_control_number="CLAIM001",
+            ),
+        ],
+    )
+
+
+def _parse_277ca_one_rejected() -> ParseResult277CA:
+    return ParseResult277CA(
+        envelope=Envelope(
+            sender_id="PAYER", receiver_id="SUBMITTER",
+            control_number="000000078", transaction_date=date(2026, 7, 2),
+        ),
+        bht=AcknowledgmentHeader(
+            hierarchical_structure_code="0085",
+            transaction_set_purpose_code="08",
+            reference_identification="REFNUM002",
+        ),
+        summary=BatchSummary(
+            input_file="277_one_rejected.txt", control_number="000000078",
+            transaction_date=date(2026, 7, 2),
+            total_claims=1, passed=0, failed=1,
+        ),
+        claim_statuses=[
+            ClaimStatus(
+                status_code="A6:19:PR",
+                classification="rejected",
+                payer_claim_control_number="CLAIM002",
+            ),
+        ],
+    )
+
+
+def _parse_ta1() -> ParseResultTa1:
+    return ParseResultTa1(
+        envelope=Envelope(
+            sender_id="RECEIVER", receiver_id="SENDER",
+            control_number="000000001", transaction_date=date(2026, 7, 2),
+        ),
+        ta1=Ta1Ack(
+            control_number="000000001",
+            interchange_date=date(2026, 7, 2),
+            interchange_time="1200",
+            ack_code="A",
+            note_code="000",
+            ack_generated_date=date(2026, 7, 2),
+        ),
+        source_batch_id="TA1-000000001",
+    )
+
+
+# ---------------------------------------------------------------------------
+# Spec §6 tests
+# ---------------------------------------------------------------------------
+
+
+def test_999_auto_creates_per_ak2_link_rows():
+    """Step 2.5 (named after spec §6).
+
+    Two AK2s in one 999 — one accepted, one rejected. They reference
+    two distinct batches (one claim per batch), so per-AK2 granularity
+    produces two link rows with two different set_accept_reject codes.
+    """
+    with db.SessionLocal()() as s:
+        _seed_batch(s, batch_id="B-999A", envelope_control="991102989")
+        _seed_claim(s, claim_id="CLM-A", batch_id="B-999A",
+                    patient_control_number="t991102989o1cA")
+        _seed_batch(s, batch_id="B-999B", envelope_control="991102990")
+        _seed_claim(s, claim_id="CLM-B", batch_id="B-999B",
+                    patient_control_number="t991102990o1cB")
+        # Index keyed by set_control_number -> batch_id (D10 batch
+        # envelope index).
+        idx = _seed_ack_link_index(envelope_control_to_batch={
+            "991102989": "B-999A",
+            "991102990": "B-999B",
+        })
+
+        parsed = _parse_999_two_ak2s()
+        out = apply_999_acceptances(s, parsed, ack_id=42,
+                                     batch_envelope_index=idx)
+        s.commit()
+
+    assert sorted(out.linked) == sorted([("CLM-A", 0), ("CLM-B", 1)])
+    assert out.orphans == []
+
+    with db.SessionLocal()() as s:
+        rows = (
+            s.query(ClaimAck)
+            .filter(ClaimAck.ack_kind == "999", ClaimAck.ack_id == 42)
+            .order_by(ClaimAck.id.asc())
+            .all()
+        )
+        assert len(rows) == 2
+        codes = {r.set_accept_reject_code for r in rows}
+        assert codes == {"A", "R"}
+        assert all(r.linked_by == "auto" for r in rows)
+
+
+def test_999_orphan_does_not_create_link():
+    """An AK2 whose set_control_number doesn't resolve stays orphan."""
+    with db.SessionLocal()() as s:
+        _seed_batch(s, batch_id="B-999-ORPHAN", envelope_control="0001")
+        parsed = _parse_999_two_ak2s()
+        idx = _seed_ack_link_index(envelope_control_to_batch={})
+        out = apply_999_acceptances(s, parsed, ack_id=99,
+                                     batch_envelope_index=idx)
+        s.commit()
+
+    assert out.linked == []
+    assert sorted(out.orphans) == ["991102989", "991102990"]
+
+    with db.SessionLocal()() as s:
+        n = s.query(ClaimAck).filter(ClaimAck.ack_kind == "999").count()
+        assert n == 0
+
+
+def test_277ca_accepted_creates_link_with_no_state_mutation():
+    """Accepted 277CA → link row, no payer_rejected stamp."""
+    with db.SessionLocal()() as s:
+        _seed_batch(s, batch_id="B-277A", envelope_control="991102989")
+        claim = _seed_claim(s, claim_id="CLM-Z", batch_id="B-277A",
+                            patient_control_number="CLAIM001")
+        idx = _seed_ack_link_index(envelope_control_to_batch={})
+
+        parsed = _parse_277ca_one_accepted()
+
+        def _pcn_lookup(pcn: str):
+            return (s.query(Claim)
+                     .filter_by(patient_control_number=pcn)
+                     .first())
+
+        out = apply_277ca_acks(s, parsed, ack_id=7,
+                                batch_envelope_index=idx,
+                                pc_claim_lookup=_pcn_lookup)
+        s.commit()
+
+    assert out.linked == [("CLM-Z", None)]
+    with db.SessionLocal()() as s:
+        c = s.get(Claim, "CLM-Z")
+        assert c.payer_rejected_at is None
+        link = s.query(ClaimAck).filter_by(claim_id="CLM-Z").one()
+        assert link.set_accept_reject_code.startswith("A3")
+        assert link.linked_by == "auto"
+
+
+def test_277ca_rejected_creates_link_and_mutates_state():
+    """Rejected 277CA → link row + payer_rejected_at stamp (mirrors
+    existing apply_277ca_rejections test)."""
+    from cyclone.inbox_state_277ca import apply_277ca_rejections
+
+    with db.SessionLocal()() as s:
+        _seed_batch(s, batch_id="B-277R", envelope_control="991102989")
+        _seed_claim(s, claim_id="CLM-Y", batch_id="B-277R",
+                    patient_control_number="CLAIM002")
+        idx = _seed_ack_link_index(envelope_control_to_batch={})
+        parsed = _parse_277ca_one_rejected()
+
+        # First, run the existing rejection mutator (it owns
+        # payer_rejected_at + state). Then run the auto-linker.
+        apply_277ca_rejections(
+            s, parsed,
+            claim_lookup=lambda pcn: (s.query(Claim)
+                                       .filter_by(patient_control_number=pcn)
+                                       .first()),
+            two77ca_id=7,
+        )
+
+        def _pcn_lookup(pcn: str):
+            return (s.query(Claim)
+                     .filter_by(patient_control_number=pcn)
+                     .first())
+
+        out = apply_277ca_acks(s, parsed, ack_id=7,
+                                batch_envelope_index=idx,
+                                pc_claim_lookup=_pcn_lookup)
+        s.commit()
+
+    assert out.linked == [("CLM-Y", None)]
+    with db.SessionLocal()() as s:
+        c = s.get(Claim, "CLM-Y")
+        assert c.payer_rejected_at is not None
+        assert c.payer_rejected_status_code and c.payer_rejected_status_code.startswith("A6")
+        link = s.query(ClaimAck).filter_by(claim_id="CLM-Y").one()
+        assert link.set_accept_reject_code.startswith("A6")
+
+
+def test_ta1_links_to_most_recent_matching_batch():
+    """TA1's batch_lookup closure picks the most-recent batch whose
+    envelope.sender_id/receiver_id matches the TA1 envelope."""
+    ta1 = _parse_ta1()
+    # TA1 carries the swapped ISA sender/receiver (the receiving side
+    # is sending the ACK back to the original submitter).
+    with db.SessionLocal()() as s:
+        # Two older batches whose envelope sender/receiver do NOT
+        # match the TA1.
+        _seed_batch(s, batch_id="T1-OLD", envelope_control="0001",
+                    sender_id="SENDER", receiver_id="RECEIVER")
+        _seed_batch(s, batch_id="T2-NEW", envelope_control="0002",
+                    sender_id="SENDER", receiver_id="RECEIVER")
+        # The matching batch has the swapped sender/receiver; ordered
+        # by parsed_at DESC and prefixed T3-MATCH so the most-recent
+        # one wins.
+        _seed_batch(s, batch_id="T3-MATCH", envelope_control="0003",
+                    sender_id="RECEIVER", receiver_id="SENDER")
+
+        def _batch_lookup(sender_id, receiver_id):
+            rows = (
+                s.query(Batch)
+                .filter(Batch.kind == "837")
+                .order_by(Batch.parsed_at.desc())
+                .all()
+            )
+            for row in rows:
+                env = (row.raw_result_json or {}).get("envelope") or {}
+                if (
+                    env.get("sender_id") == sender_id
+                    and env.get("receiver_id") == receiver_id
+                ):
+                    return row
+            return None
+
+        out = apply_ta1_envelope_link(s, ta1, ack_id=11,
+                                       batch_lookup=_batch_lookup)
+        s.commit()
+
+    assert out.linked == [(None, None)]
+    with db.SessionLocal()() as s:
+        link = (
+            s.query(ClaimAck)
+            .filter_by(ack_kind="ta1", ack_id=11)
+            .one()
+        )
+        assert link.claim_id is None
+        assert link.batch_id == "T3-MATCH"
+        assert link.set_accept_reject_code == "A"
+
+
+def test_reingest_same_999_is_idempotent():
+    """Submitting the same 999 twice produces exactly one ClaimAck
+    row per AK2 (the partial unique index enforces this)."""
+    with db.SessionLocal()() as s:
+        _seed_batch(s, batch_id="B-IDEMP", envelope_control="991102989")
+        _seed_claim(s, claim_id="CLM-I", batch_id="B-IDEMP",
+                    patient_control_number="t991102989o1cI")
+        idx = _seed_ack_link_index(envelope_control_to_batch={
+            "991102989": "B-IDEMP",
+            "991102990": "B-IDEMP",
+        })
+        parsed = _parse_999_two_ak2s()
+
+        out1 = apply_999_acceptances(s, parsed, ack_id=101,
+                                      batch_envelope_index=idx)
+        s.commit()
+    with db.SessionLocal()() as s:
+        out2 = apply_999_acceptances(s, parsed, ack_id=101,
+                                      batch_envelope_index=idx)
+        s.commit()
+
+    # Each call returns the freshly linked rows. Second call finds
+    # existing rows so linked is empty.
+    assert len(out1.linked) == 2
+    assert out2.linked == []  # idempotent — no duplicate rows
+    with db.SessionLocal()() as s:
+        n = s.query(ClaimAck).filter_by(ack_id=101).count()
+        assert n == 2
+
+
+def test_manual_link_endpoint_idempotent():
+    """link_manual returns the existing row when called twice."""
+    with db.SessionLocal()() as s:
+        _seed_batch(s, batch_id="B-MAN", envelope_control="991102989")
+        _seed_claim(s, claim_id="CLM-M", batch_id="B-MAN")
+
+        row1, created1 = link_manual(s, claim_id="CLM-M", ack_kind="999",
+                                      ack_id=200)
+        s.commit()
+    with db.SessionLocal()() as s:
+        row2, created2 = link_manual(s, claim_id="CLM-M", ack_kind="999",
+                                      ack_id=200)
+        s.commit()
+
+    assert created1 is True
+    assert created2 is False
+    assert row1.id == row2.id
+
+
+def test_manual_link_any_user_succeeds():
+    """Manual link does NOT raise (any-logged-in user posture per D5/D9)."""
+    # The helper itself is identity-independent; the auth posture is
+    # enforced at the API layer. Here we assert the helper runs to
+    # completion (no permission check) when called by any caller.
+    with db.SessionLocal()() as s:
+        _seed_batch(s, batch_id="B-AUTH", envelope_control="991102989")
+        _seed_claim(s, claim_id="CLM-N", batch_id="B-AUTH")
+        row, created = link_manual(s, claim_id="CLM-N", ack_kind="999",
+                                    ack_id=300)
+        s.commit()
+    assert created is True
+    assert row.linked_by == "manual"
+
+
+def test_manual_link_rejects_terminal_claim():
+    """A claim in state=REVERSED should NOT be link-able (the API
+    layer maps that to 409; the helper itself lets the API pick the
+    claim up and surfaces the state). Here we only assert that
+    ``link_manual`` does NOT raise — the API is responsible for the
+    409 decision via :class:`cyclone.db.ClaimState`."""
+    with db.SessionLocal()() as s:
+        _seed_batch(s, batch_id="B-TERM", envelope_control="991102989")
+        claim = _seed_claim(
+            s, claim_id="CLM-R", batch_id="B-TERM",
+            state=ClaimState.REVERSED,
+        )
+        # Helper itself does no state check — the API maps state to
+        # 409. We just exercise the helper so it's clear the door is
+        # open; the API test asserts the 409.
+        row, created = link_manual(s, claim_id="CLM-R", ack_kind="999",
+                                    ack_id=400)
+        s.commit()
+
+    assert created is True
+    assert row.claim_id == "CLM-R"
+
+
+def test_unlink_does_not_revert_claim_state():
+    """After unlink via remove_claim_ack, the claim retains its state
+    — unlinking only removes the link row, not the state mutation
+    from the original 999/277CA."""
+    with db.SessionLocal()() as s:
+        _seed_batch(s, batch_id="B-UL", envelope_control="991102989")
+        _seed_claim(s, claim_id="CLM-UL", batch_id="B-UL",
+                    state=ClaimState.REJECTED)
+        # Only the first AK2 resolves (single batch, single claim);
+        # the second becomes an orphan.
+        idx = _seed_ack_link_index(envelope_control_to_batch={
+            "991102989": "B-UL",
+        })
+        # Submit a 999 with one AK2 so we get one link row.
+        sr = SetFunctionalGroupResponse(
+            ak2=AckHeader999(functional_id_code="837",
+                             group_control_number="991102989"),
+            set_control_number="991102989",
+            transaction_set_identifier="837",
+            segment_errors=[],
+            set_accept_reject=SetAcceptReject(code="R"),
+        )
+        parsed = ParseResult999(
+            envelope=Envelope(
+                sender_id="P", receiver_id="S",
+                control_number="0001", transaction_date=date(2026, 7, 2),
+            ),
+            functional_group_acks=[],
+            set_responses=[sr],
+            summary=BatchSummary(
+                input_file="one_ak2.txt", control_number="0001",
+                transaction_date=date(2026, 7, 2),
+                total_claims=1, passed=0, failed=1,
+            ),
+        )
+        apply_999_acceptances(s, parsed, ack_id=500,
+                               batch_envelope_index=idx)
+        s.commit()
+    with db.SessionLocal()() as s:
+        links = (
+            s.query(ClaimAck)
+            .filter_by(claim_id="CLM-UL")
+            .all()
+        )
+        assert len(links) == 1
+        s.delete(links[0])
+        s.commit()
+    with db.SessionLocal()() as s:
+        claim = s.get(Claim, "CLM-UL")
+        assert claim.state == ClaimState.REJECTED
+        assert (
+            s.query(ClaimAck).filter_by(claim_id="CLM-UL").first() is None
+        )
+
+
+# ---------------------------------------------------------------------------
+# Step 2.5 — D10 two-pass join + multi-claim batch coverage
+# ---------------------------------------------------------------------------
+
+
+def test_lookup_claims_two_pass_join():
+    """D10: Pass 1 (batch.envelope.control_number) + Pass 2 (PCN)."""
+    with db.SessionLocal()() as s:
+        _seed_batch(s, batch_id="B-TP-A", envelope_control="991102989")
+        _seed_claim(s, claim_id="C-TP-A", batch_id="B-TP-A",
+                    patient_control_number="t991102989o1cA")
+        # No matching batch — Pass 2 alone resolves it.
+        _seed_batch(s, batch_id="B-TP-B", envelope_control="OTHER")
+        _seed_claim(s, claim_id="C-TP-B", batch_id="B-TP-B",
+                    patient_control_number="991102987")
+
+        # Pass 1: "991102989" -> B-TP-A -> [C-TP-A]
+        idx = _seed_ack_link_index(envelope_control_to_batch={
+            "991102989": "B-TP-A",
+            "991102990": "B-TP-A",
+        })
+
+        # PCN-only claim (no batch match). Pass 2 must find it.
+        def _pcn_lookup(pcn: str):
+            return s.query(Claim).filter_by(patient_control_number=pcn).first()
+
+        pass1 = lookup_claims_for_ack_set_response(
+            s, "991102989",
+            batch_envelope_index=idx,
+            pc_claim_lookup=_pcn_lookup,
+        )
+        pass2 = lookup_claims_for_ack_set_response(
+            s, "991102987",
+            batch_envelope_index=idx,
+            pc_claim_lookup=_pcn_lookup,
+        )
+        miss = lookup_claims_for_ack_set_response(
+            s, "0001",
+            batch_envelope_index=idx,
+            pc_claim_lookup=_pcn_lookup,
+        )
+
+    assert [c.id for c in pass1] == ["C-TP-A"]
+    assert [c.id for c in pass2] == ["C-TP-B"]
+    assert miss == []
+
+
+def test_lookup_claims_pass1_wins_over_pass2():
+    """If a PCN matches a claim in a DIFFERENT batch than the Pass 1
+    batch, only the Pass 1 result is returned (no double-fire)."""
+    with db.SessionLocal()() as s:
+        # Claim A lives in B1 (whose ST02 == '991102989'). Pass 1 hits.
+        _seed_batch(s, batch_id="B-PP-1", envelope_control="991102989")
+        _seed_claim(s, claim_id="C-PP-A", batch_id="B-PP-1",
+                    patient_control_number="991102989")
+        # Claim B lives in B2 (ST02 == 'OTHER') but has the same PCN
+        # by coincidence. Pass 1 misses for '991102989' on B2 so Pass 2
+        # SHOULD match it — but only when Pass 1 returns nothing.
+        _seed_batch(s, batch_id="B-PP-2", envelope_control="OTHER")
+        _seed_claim(s, claim_id="C-PP-B", batch_id="B-PP-2",
+                    patient_control_number="991102989")
+
+        idx = _seed_ack_link_index(envelope_control_to_batch={
+            "991102989": "B-PP-1",
+        })
+        rows = lookup_claims_for_ack_set_response(
+            s, "991102989",
+            batch_envelope_index=idx,
+            pc_claim_lookup=lambda pcn: s.query(Claim)
+                                          .filter_by(patient_control_number=pcn)
+                                          .first(),
+        )
+
+    # Pass 1 must win and skip Pass 2.
+    assert [c.id for c in rows] == ["C-PP-A"]
+
+
+def test_999_linker_emits_one_row_per_claim_in_multi_claim_batch():
+    """One 999 AK2 + N claims in one batch → N ClaimAck rows for that
+    single AK2 (one-ack-to-many)."""
+    with db.SessionLocal()() as s:
+        _seed_batch(s, batch_id="B-MC", envelope_control="991102989")
+        _seed_claim(s, claim_id="C-MC-1", batch_id="B-MC",
+                    patient_control_number="t991102989o1c1")
+        _seed_claim(s, claim_id="C-MC-2", batch_id="B-MC",
+                    patient_control_number="t991102989o1c2")
+        idx = _seed_ack_link_index(envelope_control_to_batch={
+            "991102989": "B-MC",
+        })
+
+        # 999 with one AK2 whose set_control_number matches the batch.
+        sr = SetFunctionalGroupResponse(
+            ak2=AckHeader999(functional_id_code="837",
+                             group_control_number="991102989"),
+            set_control_number="991102989",
+            transaction_set_identifier="837",
+            segment_errors=[],
+            set_accept_reject=SetAcceptReject(code="A"),
+        )
+        parsed = ParseResult999(
+            envelope=Envelope(
+                sender_id="P", receiver_id="S",
+                control_number="0001", transaction_date=date(2026, 7, 2),
+            ),
+            functional_group_acks=[],
+            set_responses=[sr],
+            summary=BatchSummary(
+                input_file="one_ak2.txt", control_number="0001",
+                transaction_date=date(2026, 7, 2),
+                total_claims=1, passed=1, failed=0,
+            ),
+        )
+        out = apply_999_acceptances(s, parsed, ack_id=42,
+                                     batch_envelope_index=idx)
+        s.commit()
+
+    assert sorted(out.linked) == sorted([("C-MC-1", 0), ("C-MC-2", 0)])
+    with db.SessionLocal()() as s:
+        n = s.query(ClaimAck).filter_by(ack_kind="999").count()
+        assert n == 2
+
+
+# ---------------------------------------------------------------------------
+# Step 3.4 — facade wiring test (placeholder; full Phase 3 adds the
+# store methods). The presence-only check verifies the public surface
+# later exposes the same names — kept here so the plan-step stays
+# trackable.
+# ---------------------------------------------------------------------------
+
+
+def test_store_facade_exposes_claim_ack_methods():
+    """Step 3.4: facade must expose add_claim_ack + 4 read methods."""
+    expected = [
+        "add_claim_ack",
+        "list_acks_for_claim",
+        "list_claims_for_ack",
+        "find_ack_orphans",
+        "remove_claim_ack",
+        "batch_envelope_index",
+    ]
+    for name in expected:
+        assert hasattr(store, name), f"missing CycloneStore.{name}"
+
+
+# ---------------------------------------------------------------------------
+# Step 2.1 — pure walk-through unit test (no DB) for the orphan tracking
+# ---------------------------------------------------------------------------
+
+
+def test_999_linker_walks_set_responses():
+    """Stub-based: one AK2 resolves, one orphan, no DB."""
+    # Mock claim_lookup: "AAA" -> Claim-like, otherwise None.
+    class _StubClaim:
+        def __init__(self, cid):
+            self.id = cid
+
+    from cyclone.parsers.models_999 import (
+        ParseResult999, SetAcceptReject, SetFunctionalGroupResponse,
+        AcknowledgmentHeader as AckHdr,
+    )
+
+    parse_result = ParseResult999.model_construct()
+    parse_result.envelope = Envelope.model_construct(
+        sender_id="", receiver_id="", control_number="",
+        transaction_date=date(2026, 7, 2),
+    )
+    parse_result.functional_group_acks = []
+    parse_result.summary = BatchSummary(
+        input_file="stub.txt", control_number="",
+        transaction_date=date(2026, 7, 2),
+        total_claims=2, passed=1, failed=1,
+    )
+    parse_result.set_responses = [
+        SetFunctionalGroupResponse.model_construct(
+            ak2=AckHdr(functional_id_code="837", group_control_number="AAA"),
+            set_control_number="AAA",
+            transaction_set_identifier="837",
+            segment_errors=[],
+            set_accept_reject=SetAcceptReject.model_construct(code="A"),
+        ),
+        SetFunctionalGroupResponse.model_construct(
+            ak2=AckHdr(functional_id_code="837", group_control_number="BBB"),
+            set_control_number="BBB",
+            transaction_set_identifier="837",
+            segment_errors=[],
+            set_accept_reject=SetAcceptReject.model_construct(code="R"),
+        ),
+    ]
+
+    seen: list[str] = []
+
+    def _pcn_lookup(pcn):
+        seen.append(pcn)
+        return _StubClaim(f"CLM-{pcn}") if pcn == "AAA" else None
+
+    class _StubQuery:
+        def filter(self, *args, **kwargs):
+            return self
+        def first(self):
+            return None  # no existing link rows in stub
+
+    class _StubSession:
+        # Pretend to be a Session; capture adds and answer the dedup
+        # query with "no existing rows" so the stub stays pure-unit.
+        def __init__(self):
+            self.added: list[ClaimAck] = []
+
+        def add(self, obj):  # noqa: D401
+            self.added.append(obj)
+
+        def flush(self):  # noqa: D401
+            pass
+
+        def query(self, *args, **kwargs):
+            return _StubQuery()
+
+    s = _StubSession()
+    out = apply_999_acceptances(
+        s, parse_result,
+        ack_id=1,
+        batch_envelope_index=lambda scn: None,  # Pass 1 misses → Pass 2 fires
+        pc_claim_lookup=_pcn_lookup,
+    )
+
+    assert seen == ["AAA", "BBB"]
+    assert [(cid, idx) for (cid, idx) in out.linked] == [("CLM-AAA", 0)]
+    assert out.orphans == ["BBB"]
+    assert len(s.added) == 1
+    assert s.added[0].claim_id == "CLM-AAA"
+    assert s.added[0].ack_kind == "999"
+    assert s.added[0].set_accept_reject_code == "A"