feat(sp27): extract handle_ta1 from scheduler.py into handlers/

backend/src/cyclone/handlers/handle_ta1.py

@@ -0,0 +1,89 @@
+"""Handle a TA1 Interchange Acknowledgment file (SP27 Task 3).
+
+Lifted verbatim from ``scheduler.py:_handle_ta1``. The handler owns
+its own DB session, dispatches to ``parse_ta1_text``, persists the
+interchange ack row in ``ta1_acks``, and returns a
+``(parser_used, claim_count)`` tuple.
+
+Unlike the 999 handler, TA1 is envelope-only — there is one TA1 per
+ISA/IEA interchange, no set-level (AK2) or claim-level matching.
+The claim_count is always 1 (one TA1 ack row per file).
+
+The actor tag is implicit (no audit event here — TA1 doesn't tag
+anything in the activity log; it's an infrastructure-level ack).
+"""
+from __future__ import annotations
+
+import json
+import logging
+from typing import Optional
+
+from cyclone import db
+from cyclone.parsers.exceptions import CycloneParseError
+from cyclone.parsers.parse_ta1 import parse_ta1_text
+from cyclone.store import store as cycl_store
+
+log = logging.getLogger(__name__)
+
+
+def handle(
+    text: str,
+    source_file: str,
+    *,
+    event_bus: Optional[object] = None,
+) -> tuple[str, int]:
+    """Parse a TA1, persist the interchange ack row.
+
+    Args:
+        text: Raw TA1 document bytes (decoded).
+        source_file: Filename the TA1 came from. Used for audit
+            attribution; the ``batches.id`` for TA1 rows is derived
+            internally from the parsed envelope's control number
+            (``TA1-{ICN}``).
+        event_bus: Optional pubsub handle. Reserved for the
+            FastAPI-endpoint migration in Task 6 — the scheduler
+            doesn't pass one.
+
+    Returns:
+        ``(parser_used, claim_count)`` tuple. TA1 always returns
+        ``claim_count=1`` (one TA1 ack row per interchange file).
+
+    Raises:
+        ValueError: on parser-level failure (wraps CycloneParseError).
+    """
+    try:
+        result = parse_ta1_text(text, input_file=source_file)
+    except CycloneParseError as exc:
+        raise ValueError(f"TA1 parse error: {exc}") from exc
+
+    with db.SessionLocal()() as session:
+        cycl_store.add_ta1_ack(
+            source_batch_id=result.source_batch_id,
+            control_number=result.ta1.control_number,
+            interchange_date=result.ta1.interchange_date,
+            interchange_time=result.ta1.interchange_time,
+            ack_code=result.ta1.ack_code,
+            note_code=result.ta1.note_code,
+            ack_generated_date=result.ta1.ack_generated_date,
+            sender_id=result.envelope.sender_id,
+            receiver_id=result.envelope.receiver_id,
+            raw_json=json.loads(result.model_dump_json()),
+        )
+        session.commit()
+
+    # TODO(sp27-task-6): bridge async EventBus.publish → sync caller
+    # (same gap as handle_999; both fixed when the FastAPI endpoints
+    # migrate to call these handlers instead of inline code).
+    if event_bus is not None:
+        publish = getattr(event_bus, "publish", None)
+        if callable(publish):
+            try:
+                publish("ack_received", {
+                    "source_batch_id": result.source_batch_id,
+                    "ack_code": result.ta1.ack_code,
+                    "kind": "TA1",
+                })
+            except Exception as exc:  # noqa: BLE001
+                log.warning("event_bus publish failed: %s", exc)
+
+    return ("parse_ta1", 1)

backend/src/cyclone/scheduler.py

@@ -56,6 +56,7 @@ from cyclone.clearhouse import InboundFile, SftpClient
 from cyclone.db import ProcessedInboundFile
 from cyclone.edi.filenames import parse_inbound_filename
 from cyclone.handlers import handle_999 as _handle_999
+from cyclone.handlers import handle_ta1 as _handle_ta1
 from cyclone.handlers._ack_id import (
     two77ca_synthetic_source_batch_id as _277ca_synthetic_source_batch_id,
 )
@@ -145,10 +146,10 @@ class SchedulerStatus:
 # persists its own DB rows. The scheduler records the outcome.
 # ---------------------------------------------------------------------------
 #
-# The 999 handler now lives in ``cyclone.handlers.handle_999`` (SP27
-# Task 2); the 277CA / TA1 / 835 handlers still live inline here and
-# are lifted in Tasks 3-5. The dict literal below references the
-# alias imports so the wiring stays identical.
+# 999 (SP27 Task 2) and TA1 (Task 3) now live in ``cyclone.handlers``;
+# the 277CA and 835 handlers stay inline and are lifted in Tasks 4-5.
+# The HANDLERS dict literal below references the alias imports so the
+# wiring stays identical.
 
 
 def _handle_835(text: str, source_file: str) -> tuple[str, int]:
@@ -252,33 +253,6 @@ def _handle_277ca(text: str, source_file: str) -> tuple[str, int]:
     return "parse_277ca", len(result.claim_statuses)
 
 
-def _handle_ta1(text: str, source_file: str) -> tuple[str, int]:
-    """Parse a TA1, persist the interchange ack row."""
-    from cyclone.parsers.parse_ta1 import parse_ta1_text
-    from cyclone.parsers.exceptions import CycloneParseError
-
-    try:
-        result = parse_ta1_text(text, input_file=source_file)
-    except CycloneParseError as exc:
-        raise ValueError(f"TA1 parse error: {exc}") from exc
-
-    with db.SessionLocal()() as session:
-        cycl_store.add_ta1_ack(
-            source_batch_id=result.source_batch_id,
-            control_number=result.ta1.control_number,
-            interchange_date=result.ta1.interchange_date,
-            interchange_time=result.ta1.interchange_time,
-            ack_code=result.ta1.ack_code,
-            note_code=result.ta1.note_code,
-            ack_generated_date=result.ta1.ack_generated_date,
-            sender_id=result.envelope.sender_id,
-            receiver_id=result.envelope.receiver_id,
-            raw_json=json.loads(result.model_dump_json()),
-        )
-        session.commit()
-    return "parse_ta1", 1
-
-
 # Map file_type → handler. Mirrors ROUTED_FILE_TYPES.
 HANDLERS: dict[str, Callable[[str, str], tuple[str, int]]] = {
     "999": _handle_999,
@@ -302,6 +276,10 @@ def _277ca_synthetic_source_batch_id(interchange_control_number: str) -> str:
     still inline (SP27 Task 4 will move it next to ``handle_999`` in
     ``cyclone/handlers/``). For the 999 side, ``_handle_999`` is
     already imported from ``cyclone.handlers.handle_999`` (Task 2).
+
+    TODO(sp27-task-4): delete this duplicate when ``_handle_277ca``
+    moves. The canonical copy now lives in
+    ``cyclone.handlers._ack_id.two77ca_synthetic_source_batch_id``.
     """
     return f"277CA-{(interchange_control_number or '').strip() or '000000001'}"
 

backend/tests/test_handlers_ta1.py

@@ -0,0 +1,99 @@
+"""Direct tests for the ``handle_ta1`` handler (SP27 Task 3).
+
+Locks the handler's contract independent of the scheduler lifecycle
+so a regression in the scheduler wiring doesn't hide a regression
+in the handler.
+"""
+from __future__ import annotations
+
+import pytest
+
+from cyclone import db
+from cyclone.handlers.handle_ta1 import handle
+from cyclone.parsers.exceptions import CycloneParseError
+
+
+ACCEPTED = (
+    "ISA*00*          *00*          *ZZ*SENDER         *ZZ*RECEIVER       "
+    "*260520*1750*^*00501*000000001*0*P*:~"
+    "TA1*000000001*20260520*1750*A*000*20260520~"
+    "IEA*1*000000001~"
+)
+
+REJECTED = (
+    "ISA*00*          *00*          *ZZ*COMEDASSISTPROG*ZZ*TP11525703     "
+    "*260520*1750*^*00501*000000001*0*P*:~"
+    "TA1*320293557*260520*2338*R*006~"
+    "IEA*0*000000001~"
+)
+
+
+def test_handle_ta1_persists_row_and_returns_count():
+    """Accepted TA1 → parse_ta1, claim_count=1, persists ack row with ack_code=A."""
+    parser_used, claim_count = handle(ACCEPTED, source_file="accepted.ta1")
+    assert parser_used == "parse_ta1"
+    assert claim_count == 1
+
+    # Lock the persistence half: a row was added to ta1_acks.
+    with db.SessionLocal()() as session:
+        rows = (
+            session.query(db.Ta1Ack)
+            .filter_by(source_batch_id="TA1-000000001")
+            .all()
+        )
+        assert len(rows) == 1
+        assert rows[0].ack_code == "A"
+        assert rows[0].note_code == "000"
+        assert rows[0].control_number == "000000001"
+
+
+def test_handle_ta1_rejected_persists_with_rejected_ack_code():
+    """Rejected TA1 (R ack_code, non-zero note_code) → persists row, R ack_code.
+
+    Note: source_batch_id is derived from the ISA control number
+    (``TA1-<ISA13>``), not the TA1 segment's internal ICN. Both
+    fixtures share ISA13=000000001, so source_batch_id is the same
+    string — but ack_code + note_code differentiate the two rows.
+    """
+    parser_used, claim_count = handle(REJECTED, source_file="rejected.ta1")
+    assert parser_used == "parse_ta1"
+    assert claim_count == 1
+
+    # Lock the rejection half: ack_code="R" + note_code="006" persisted.
+    # (DB is reset per-test, so this fixture owns exactly one row.)
+    with db.SessionLocal()() as session:
+        rows = (
+            session.query(db.Ta1Ack)
+            .filter_by(source_batch_id="TA1-000000001")
+            .all()
+        )
+        assert len(rows) == 1
+        assert rows[0].ack_code == "R"
+        assert rows[0].note_code == "006"
+
+
+def test_handle_ta1_missing_segment_raises():
+    """No TA1 segment → handler raises ValueError (wraps CycloneParseError)."""
+    text = (
+        "ISA*00*          *00*          *ZZ*SENDER         *ZZ*RECEIVER       "
+        "*260520*1750*^*00501*000000001*0*P*:~"
+        "IEA*0*000000001~"
+    )
+    with pytest.raises((CycloneParseError, ValueError)):
+        handle(text, source_file="empty.ta1")
+
+
+def test_handle_ta1_does_not_create_batches_row():
+    """TA1 is an envelope-only ack; no ``batches`` row should be created
+    (only ``ta1_acks``)."""
+    parser_used, claim_count = handle(ACCEPTED, source_file="accepted.ta1")
+    with db.SessionLocal()() as session:
+        # Confirm no batch was created. Source_batch_id starts with
+        # "TA1-" — search the batches table (which uses UUIDs) for any
+        # row that starts with "TA1-".
+        from sqlalchemy import select
+        stmt = select(db.Batch.__table__.c.id).where(
+            db.Batch.__table__.c.id.like("TA1-%")
+        )
+        rows = session.execute(stmt).all()
+        assert rows == []