feat(sp6): apply_999_rejections state helper

T3 — moves claims to ClaimState.REJECTED on 999 AK5 R/E/X.

- New module: src/cyclone/inbox_state.py
- New tests: tests/test_inbox_state.py (4 cases)
- Adds ORM mappings for rejection_reason / rejected_at / resubmit_count
  / state_changed_at on Claim (db.py) — these were added to the
  schema by migration 0004 but not yet exposed to the ORM.

backend/src/cyclone/db.py

@@ -195,9 +195,19 @@ class Claim(Base):
     state: Mapped[ClaimState] = mapped_column(
         Enum(ClaimState, native_enum=False), nullable=False, default=ClaimState.SUBMITTED
     )
+    state_changed_at: Mapped[Optional[datetime]] = mapped_column(
+        DateTime(timezone=True), nullable=True
+    )
     state_before_reversal: Mapped[Optional[ClaimState]] = mapped_column(
         Enum(ClaimState, native_enum=False), nullable=True
     )
+    rejection_reason: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
+    rejected_at: Mapped[Optional[datetime]] = mapped_column(
+        DateTime(timezone=True), nullable=True
+    )
+    resubmit_count: Mapped[int] = mapped_column(
+        Integer, nullable=False, default=0, server_default=text("0")
+    )
     matched_remittance_id: Mapped[Optional[str]] = mapped_column(
         # ORM policy: SET NULL on remittance delete. The claim may outlive its
         # remittance during reversal/reimport flows; without this, deleting a

backend/src/cyclone/inbox_state.py

@@ -0,0 +1,76 @@
+"""999 ACK → claim state transitions for the Inbox Rejected lane.
+
+SP6 T3: apply_999_rejections.
+For each set response with code R or E, look up the matching claim and
+move it to REJECTED. Idempotent on already-rejected claims.
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from datetime import datetime, timezone
+from typing import Callable
+
+from sqlalchemy.orm import Session
+
+from cyclone.db import Claim, ClaimState
+
+
+@dataclass
+class Apply999Result:
+    matched: list[str] = field(default_factory=list)   # claim ids transitioned
+    orphans: list[str] = field(default_factory=list)   # PCNs we couldn't resolve
+
+
+def _build_reason(code: str, segment_error_count: int) -> str:
+    parts = [f"999 AK5 set-level {code}"]
+    if segment_error_count:
+        parts.append(f"{segment_error_count} segment error(s)")
+    return "; ".join(parts)
+
+
+def apply_999_rejections(
+    session: Session,
+    parsed_999,
+    *,
+    claim_lookup: Callable[[str], Claim | None],
+) -> Apply999Result:
+    """For each set response with code R or E, look up the matching claim and
+    move it to REJECTED. Idempotent on already-rejected claims.
+
+    Args:
+        session: SQLAlchemy session.
+        parsed_999: a ParseResult999 (or any object with .set_responses).
+        claim_lookup: callable from patient_control_number → Claim or None.
+
+    Returns:
+        Apply999Result with lists of matched claim ids and orphan PCNs.
+    """
+    result = Apply999Result()
+    now = datetime.now(timezone.utc)
+
+    for sr in parsed_999.set_responses:
+        code = sr.set_accept_reject.code
+        if code not in ("R", "E", "X"):
+            continue
+
+        claim = claim_lookup(sr.set_control_number)
+        if claim is None:
+            result.orphans.append(sr.set_control_number)
+            continue
+
+        if claim.state == ClaimState.REJECTED:
+            # Idempotent: don't double-mutate.
+            continue
+
+        claim.state = ClaimState.REJECTED
+        claim.state_changed_at = now
+        claim.rejected_at = now
+        claim.rejection_reason = _build_reason(
+            code, len(sr.segment_errors or [])
+        )
+        result.matched.append(claim.id)
+
+    if result.matched or result.orphans:
+        session.commit()
+
+    return result

backend/tests/test_inbox_state.py

@@ -0,0 +1,166 @@
+"""TDD: apply_999_rejections moves claims to REJECTED on 999 AK5 R/E.
+
+Plan: docs/superpowers/plans/2026-06-20-cyclone-workflow-automation.md
+Task 3 (T3).
+"""
+from __future__ import annotations
+
+from datetime import date, datetime, timezone
+
+import pytest
+
+from cyclone import db
+from cyclone.db import Batch, Claim, ClaimState
+from cyclone.inbox_state import apply_999_rejections
+from cyclone.parsers.models import BatchSummary, Envelope
+from cyclone.parsers.models_999 import (
+    AcknowledgmentHeader,
+    FunctionalGroupAck,
+    ParseResult999,
+    SetAcceptReject,
+    SetFunctionalGroupResponse,
+)
+
+
+@pytest.fixture(autouse=True)
+def _setup(tmp_path, monkeypatch):
+    monkeypatch.setenv("CYCLONE_DB_URL", f"sqlite:///{tmp_path}/inbox.db")
+    db._reset_for_tests()
+    db.init_db()
+
+
+def _seed_batch() -> None:
+    with db.SessionLocal()() as s:
+        s.add(Batch(
+            id="B-1", kind="837p", input_filename="x.txt",
+            parsed_at=datetime.now(timezone.utc),
+            totals_json={"total_claims": 1},
+            validation_json={"passed": True, "warnings": [], "errors": []},
+            raw_result_json={"_": "stub"},
+        ))
+        s.commit()
+
+
+def _add_claim(*, claim_id: str, pcn: str, state: ClaimState = ClaimState.SUBMITTED) -> None:
+    _seed_batch()
+    with db.SessionLocal()() as s:
+        s.add(Claim(
+            id=claim_id, batch_id="B-1",
+            patient_control_number=pcn, state=state,
+        ))
+        s.commit()
+
+
+def _make_999(*, claim_refs: list[tuple[str, str]]) -> ParseResult999:
+    """claim_refs: list of (set_control_number, accept_code)."""
+    set_responses = [
+        SetFunctionalGroupResponse(
+            ak2=AcknowledgmentHeader(
+                functional_id_code="837", group_control_number=ref,
+            ),
+            set_control_number=ref,
+            transaction_set_identifier="837",
+            segment_errors=[],
+            set_accept_reject=SetAcceptReject(code=code),  # type: ignore[arg-type]
+        )
+        for ref, code in claim_refs
+    ]
+    return ParseResult999(
+        envelope=Envelope(
+            sender_id="RECEIVER", receiver_id="SUBMITTER",
+            control_number="000000001", transaction_date=date(2024, 1, 1),
+            implementation_guide="005010X231A1",
+        ),
+        functional_group_acks=[
+            FunctionalGroupAck(
+                ak1=AcknowledgmentHeader(functional_id_code="HC", group_control_number="0001"),
+                received_count=len(set_responses), accepted_count=0,
+                rejected_count=len(set_responses), ack_code="R",
+            ),
+        ],
+        set_responses=set_responses,
+        summary=BatchSummary(
+            input_file="x.txt", total_claims=len(set_responses),
+            passed=0, failed=len(set_responses),
+        ),
+    )
+
+
+def _lookup_factory(session):
+    """Build a claim_lookup closure that uses the given session.
+
+    The session must be the same one passed to apply_999_rejections so
+    that the returned Claim is attached to that session's transaction.
+    """
+    def _lookup(pcn: str):
+        return session.query(Claim).filter_by(patient_control_number=pcn).first()
+    return _lookup
+
+
+def test_rejection_moves_submitted_claim_to_rejected_state():
+    _add_claim(claim_id="CLP-1", pcn="PCN-001")
+
+    with db.SessionLocal()() as s:
+        result = apply_999_rejections(
+            s,
+            _make_999(claim_refs=[("PCN-001", "R")]),
+            claim_lookup=_lookup_factory(s),
+        )
+        s.commit()
+
+    assert result.matched == ["CLP-1"]
+    assert result.orphans == []
+    with db.SessionLocal()() as s:
+        c = s.get(Claim, "CLP-1")
+        assert c.state == ClaimState.REJECTED
+        assert c.rejected_at is not None
+        assert "999" in (c.rejection_reason or "")
+
+
+def test_accepted_set_leaves_claim_alone():
+    _add_claim(claim_id="CLP-1", pcn="PCN-002")
+
+    with db.SessionLocal()() as s:
+        result = apply_999_rejections(
+            s,
+            _make_999(claim_refs=[("PCN-002", "A")]),
+            claim_lookup=_lookup_factory(s),
+        )
+        s.commit()
+
+    assert result.matched == []
+    with db.SessionLocal()() as s:
+        c = s.get(Claim, "CLP-1")
+        assert c.state == ClaimState.SUBMITTED
+        assert c.rejected_at is None
+        assert c.rejection_reason is None
+
+
+def test_unknown_pcn_recorded_as_orphan_not_error():
+    with db.SessionLocal()() as s:
+        result = apply_999_rejections(
+            s,
+            _make_999(claim_refs=[("PCN-NONEXISTENT", "R")]),
+            claim_lookup=_lookup_factory(s),
+        )
+        s.commit()
+
+    assert result.matched == []
+    assert result.orphans == ["PCN-NONEXISTENT"]
+
+
+def test_already_rejected_claim_is_idempotent():
+    _add_claim(claim_id="CLP-1", pcn="PCN-003", state=ClaimState.REJECTED)
+
+    with db.SessionLocal()() as s:
+        result = apply_999_rejections(
+            s,
+            _make_999(claim_refs=[("PCN-003", "R")]),
+            claim_lookup=_lookup_factory(s),
+        )
+        s.commit()
+
+    assert result.matched == []   # no-op
+    with db.SessionLocal()() as s:
+        c = s.get(Claim, "CLP-1")
+        assert c.state == ClaimState.REJECTED