tyler/cyclone
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
134eb4f404eb89294b23bd06ff6bb14ad430bea1
cyclone/docs/superpowers/plans/2026-06-21-cyclone-pipeline-agent.md
T

4526 lines
143 KiB
Markdown
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
# Cyclone Pipeline Agent Implementation Plan
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
**Goal:** Build a sibling Python package `cyclone-pipeline` that automates the full EDI 837P submission lifecycle through Cyclone to the Gainwell MFT, waits for the inbound TA1 + 999 ACKs, and writes a self-contained run directory. Stops before 835 (next Monday) which is checked via a separate subcommand.
**Architecture:** API-first + minimal Playwright shell. Browser is used only for the upload page; everything else (parse verification, SFTP submit, scheduler control, ACK polling, claim-state checks) goes through Cyclone's published HTTP API via `httpx.AsyncClient`. The agent lives at `/Users/openclaw/dev/cyclone-pipeline/`, never imports from cyclone, and talks to cyclone exclusively over HTTP.
**Tech Stack:** Python 3.11+, `httpx` (async HTTP), `playwright` (browser), `click` (CLI), `pydantic` v2 (typed I/O), `structlog` (JSON logging). Dev: `pytest`, `pytest-asyncio`, `respx` (httpx mock), `pytest-playwright` (browser tests).
**Spec:** [`docs/superpowers/specs/2026-06-21-cyclone-pipeline-agent-design.md`](../specs/2026-06-21-cyclone-pipeline-agent-design.md)
---
## File structure
```
/Users/openclaw/dev/cyclone-pipeline/
├── pyproject.toml ← Task 1
├── README.md ← Task 24
├── src/cyclone_pipeline/
│ ├── __init__.py ← Task 1 (re-exports)
│ ├── __main__.py ← Task 1
│ ├── exceptions.py ← Task 2
│ ├── logging_setup.py ← Task 3
│ ├── api_client.py ← Tasks 4–7
│ ├── state.py ← Task 8
│ ├── screenshots.py ← Task 9
│ ├── selectors.py ← Task 10
│ ├── waiters.py ← Task 11
│ ├── report.py ← Task 12
│ ├── browser.py ← Task 13
│ ├── pipeline.py ← Tasks 14–21
│ ├── check_835.py ← Task 22
│ └── cli.py ← Task 23
├── tests/
│ ├── conftest.py ← Task 1
│ ├── test_exceptions.py ← Task 2
│ ├── test_logging_setup.py ← Task 3
│ ├── test_api_client.py ← Tasks 4–7
│ ├── test_state.py ← Task 8
│ ├── test_screenshots.py ← Task 9
│ ├── test_selectors.py ← Task 10
│ ├── test_waiters.py ← Task 11
│ ├── test_report.py ← Task 12
│ ├── test_browser.py ← Task 13
│ ├── test_pipeline.py ← Tasks 14–21
│ ├── test_check_835.py ← Task 22
│ ├── test_cli.py ← Task 23
│ └── fixtures/
│ └── sample_837p.edi ← Task 1
└── .github/workflows/ci.yml ← Task 25
```
Each file has one clear responsibility. The `api_client.py` is the largest by design — it owns all HTTP shape. The `pipeline.py` is the orchestrator; everything else is a leaf module.
---
## Task 0: Pre-flight — capture baseline
**Goal:** Confirm cyclone is running and the existing test suite still passes. Snapshot the API surface so we know what to mock.
**Files:**
- Read: `backend/src/cyclone/api.py` (line counts; we mock the endpoints we need)
- Write: `/tmp/cyclone-pipeline-baseline.txt`
- [ ] **Step 1: Verify cyclone backend is reachable**
```bash
curl -fsS http://127.0.0.1:8000/api/health | python -m json.tool
```
Expected: JSON with `"status": "ok"`, `"db": {"ok": true}`. If backend isn't running, start it: `cd /Users/openclaw/dev/cyclone/backend && .venv/bin/python -m cyclone serve &`.
- [ ] **Step 2: Snapshot the API endpoints we'll consume**
Run these and record the exact response shapes — we mock them in tests:
```bash
curl -fsS http://127.0.0.1:8000/api/health | tee /tmp/api-health.json
curl -fsS http://127.0.0.1:8000/api/admin/scheduler/status | tee /tmp/api-scheduler-status.json
curl -fsS "http://127.0.0.1:8000/api/ta1-acks?limit=1" | tee /tmp/api-ta1-acks.json
curl -fsS "http://127.0.0.1:8000/api/acks?limit=1" | tee /tmp/api-acks.json
curl -fsS "http://127.0.0.1:8000/api/277ca-acks?limit=1" | tee /tmp/api-277ca-acks.json
curl -fsS "http://127.0.0.1:8000/api/claims?limit=1" | tee /tmp/api-claims.json
```
Expected: each command prints a JSON object or array. Some may return 401/403 if the API has changed — if so, the spec's endpoint list needs updating. Report back to the user.
- [ ] **Step 3: Verify the create dir target doesn't exist yet**
```bash
ls /Users/openclaw/dev/cyclone-pipeline 2>/dev/null && echo "EXISTS" || echo "OK: not yet created"
```
Expected: `OK: not yet created`. If it exists, the operator has already started this work — pause and ask.
No commit. Pre-flight only.
---
## Task 1: Project skeleton + install
**Goal:** Create the directory tree, `pyproject.toml`, empty package files, install in editable mode, verify import works.
**Files:**
- Create: `/Users/openclaw/dev/cyclone-pipeline/` (root dir)
- Create: `/Users/openclaw/dev/cyclone-pipeline/pyproject.toml`
- Create: `/Users/openclaw/dev/cyclone-pipeline/src/cyclone_pipeline/__init__.py`
- Create: `/Users/openclaw/dev/cyclone-pipeline/src/cyclone_pipeline/__main__.py`
- Create: `/Users/openclaw/dev/cyclone-pipeline/tests/__init__.py`
- Create: `/Users/openclaw/dev/cyclone-pipeline/tests/conftest.py`
- Create: `/Users/openclaw/dev/cyclone-pipeline/tests/fixtures/sample_837p.edi`
- Create: `/Users/openclaw/dev/cyclone-pipeline/.gitignore`
- Create: `/Users/openclaw/dev/cyclone-pipeline/README.md` (placeholder, expanded in Task 24)
- Create: `/Users/openclaw/dev/cyclone-pipeline/src/cyclone_pipeline/.gitkeep` (placeholder, removed as files are added)
- [ ] **Step 1: Create the directory tree**
```bash
mkdir -p /Users/openclaw/dev/cyclone-pipeline/src/cyclone_pipeline
mkdir -p /Users/openclaw/dev/cyclone-pipeline/tests/fixtures
cd /Users/openclaw/dev/cyclone-pipeline
git init -q
```
- [ ] **Step 2: Write `pyproject.toml`**
```toml
[build-system]
requires = ["setuptools>=68", "wheel"]
build-backend = "setuptools.build_meta"
[project]
name = "cyclone-pipeline"
version = "0.1.0"
description = "Cyclone EDI pipeline agent — automate 837P upload + SFTP submit + ACK wait"
requires-python = ">=3.11"
dependencies = [
"httpx>=0.27,<1",
"playwright>=1.44,<2",
"click>=8.1,<9",
"pydantic>=2.6,<3",
"structlog>=24.1,<25",
]
[project.optional-dependencies]
dev = [
"pytest>=8.0",
"pytest-asyncio>=0.23,<1",
"pytest-playwright>=0.5",
"respx>=0.21,<1",
"pytest-cov>=4.1",
]
[project.scripts]
cyclone-pipeline = "cyclone_pipeline.cli:main"
[tool.setuptools.packages.find]
where = ["src"]
[tool.pytest.ini_options]
testpaths = ["tests"]
addopts = "-ra --strict-markers"
asyncio_mode = "auto"
markers = [
"browser: requires real Playwright browser (deselect with '-m \"not browser\"')",
]
```
- [ ] **Step 3: Write empty package + test files**
Write `/Users/openclaw/dev/cyclone-pipeline/src/cyclone_pipeline/__init__.py`:
```python
"""Cyclone pipeline agent — automate 837P upload, SFTP submit, and ACK wait."""
__version__ = "0.1.0"
```
Write `/Users/openclaw/dev/cyclone-pipeline/src/cyclone_pipeline/__main__.py`:
```python
from cyclone_pipeline.cli import main
if __name__ == "__main__":
main()
```
Write `/Users/openclaw/dev/cyclone-pipeline/tests/__init__.py`: (empty file)
Write `/Users/openclaw/dev/cyclone-pipeline/tests/conftest.py`:
```python
"""Shared pytest fixtures for cyclone-pipeline tests."""
import pytest
from pathlib import Path
FIXTURES_DIR = Path(__file__).parent / "fixtures"
@pytest.fixture
def sample_837p_path() -> Path:
"""Path to a small valid 837P file used across tests."""
return FIXTURES_DIR / "sample_837p.edi"
@pytest.fixture
def sample_837p_bytes() -> bytes:
return FIXTURES_DIR.read_bytes()
```
- [ ] **Step 4: Copy the minimal 837P fixture from cyclone**
```bash
cp /Users/openclaw/dev/cyclone/backend/tests/fixtures/minimal_837p.txt /Users/openclaw/dev/cyclone-pipeline/tests/fixtures/sample_837p.edi
```
- [ ] **Step 5: Write `.gitignore`**
```gitignore
__pycache__/
*.pyc
*.egg-info/
.pytest_cache/
.coverage
htmlcov/
.venv/
runs/
playwright-report/
test-results/
```
- [ ] **Step 6: Write README placeholder**
```markdown
# cyclone-pipeline
Automation agent for Cyclone EDI submissions.
See [docs/superpowers/specs/2026-06-21-cyclone-pipeline-agent-design.md](https://github.com/.../blob/main/docs/superpowers/specs/2026-06-21-cyclone-pipeline-agent-design.md) for the full design.
```
- [ ] **Step 7: Create venv, install, verify import**
```bash
cd /Users/openclaw/dev/cyclone-pipeline
python3.11 -m venv .venv
source .venv/bin/activate
pip install -e '.[dev]' --quiet
python -c "import cyclone_pipeline; print('OK:', cyclone_pipeline.__version__)"
```
Expected: `OK: 0.1.0`.
- [ ] **Step 8: Verify pytest can discover the test tree**
```bash
cd /Users/openclaw/dev/cyclone-pipeline
pytest --collect-only -q
```
Expected: `no tests ran` (or `0 tests collected`). No errors.
- [ ] **Step 9: Commit**
```bash
cd /Users/openclaw/dev/cyclone-pipeline
git add .
git commit -m "feat(sp22): project skeleton + install in editable mode"
```
---
## Task 2: Exceptions module
**Goal:** Typed exceptions that the rest of the code can catch and the report can serialize.
**Files:**
- Create: `/Users/openclaw/dev/cyclone-pipeline/src/cyclone_pipeline/exceptions.py`
- Create: `/Users/openclaw/dev/cyclone-pipeline/tests/test_exceptions.py`
- [ ] **Step 1: Write the failing test**
Write `/Users/openclaw/dev/cyclone-pipeline/tests/test_exceptions.py`:
```python
"""Typed exceptions — caught by callers, serialized by report.py."""
import pytest
from cyclone_pipeline.exceptions import (
CyclonePipelineError,
UploadError,
ParseError,
SubmitError,
AckTimeoutError,
SchedulerNotRunningError,
IdempotencyError,
BrowserNotReachableError,
)
def test_all_inherit_from_base():
"""Every typed error must inherit from CyclonePipelineError so callers
can `except CyclonePipelineError` for a single catch-all."""
assert issubclass(UploadError, CyclonePipelineError)
assert issubclass(ParseError, CyclonePipelineError)
assert issubclass(SubmitError, CyclonePipelineError)
assert issubclass(AckTimeoutError, CyclonePipelineError)
assert issubclass(SchedulerNotRunningError, CyclonePipelineError)
assert issubclass(IdempotencyError, CyclonePipelineError)
assert issubclass(BrowserNotReachableError, CyclonePipelineError)
def test_ack_timeout_carries_phase_and_ack_kind():
"""AckTimeoutError should expose which phase timed out and which ACK
was missing — the report needs both to render the right remediation hint."""
err = AckTimeoutError("ta1", timeout_s=300)
assert err.phase == "wait_ta1"
assert err.ack_kind == "ta1"
assert err.timeout_s == 300
assert "ta1" in str(err).lower()
assert "300" in str(err)
def test_idempotency_error_lists_collisions():
"""IdempotencyError should carry the list of recent submissions that
conflict — the report needs them to render the 'already submitted'
warning."""
err = IdempotencyError(
["11525703-837P-20260620181814559-1of1.txt",
"11525703-837P-20260620182058568-1of1.txt"],
)
assert len(err.collisions) == 2
assert "11525703-837P-20260620181814559-1of1.txt" in str(err)
```
- [ ] **Step 2: Run test to verify it fails**
```bash
cd /Users/openclaw/dev/cyclone-pipeline
pytest tests/test_exceptions.py -v
```
Expected: ImportError (exceptions module doesn't exist yet).
- [ ] **Step 3: Write the implementation**
Write `/Users/openclaw/dev/cyclone-pipeline/src/cyclone_pipeline/exceptions.py`:
```python
"""Typed exceptions for the cyclone-pipeline agent.
Every pipeline-level error inherits from `CyclonePipelineError` so
callers (the CLI, the library, the report) can do a single `except` to
catch anything we throw. Each subclass carries enough structured
context for `report.py` to render a useful remediation hint.
"""
from __future__ import annotations
from typing import Sequence
class CyclonePipelineError(Exception):
"""Base class for every typed error the agent raises."""
class UploadError(CyclonePipelineError):
"""The browser-driven upload phase failed irrecoverably.
Distinct from `ParseError` (the file was uploaded but the parser
rejected it) and from `SubmitError` (parsing succeeded but the
SFTP submit failed).
"""
class ParseError(CyclonePipelineError):
"""The file was uploaded but the parser found no claims in it
(or the API returned 0 new claim IDs after a 30s poll)."""
class SubmitError(CyclonePipelineError):
"""`POST /api/clearhouse/submit` failed in a way the agent
considers terminal (e.g., the Keychain is misconfigured, the
SFTP server returned 5xx after retries, the dedup check tripped
without `--force`).
"""
class AckTimeoutError(CyclonePipelineError):
"""An expected inbound ACK did not arrive within the configured
timeout window. Soft-fail — the submission is in flight."""
def __init__(self, ack_kind: str, timeout_s: int):
self.ack_kind = ack_kind # "ta1" | "999" | "277ca" | "835"
self.timeout_s = timeout_s
super().__init__(
f"No {ack_kind} ACK arrived within {timeout_s}s; "
f"submission is in flight — check Gainwell directly."
)
class SchedulerNotRunningError(CyclonePipelineError):
"""The MFT scheduler is stopped and `tick` calls are also failing.
The agent cannot fetch ACKs without a working MFT loop."""
class IdempotencyError(SubmitError):
"""A recent SFTP submission matches the file we're about to submit
(same `claim_id` set within the dedup window). Use `--force` to
override.
"""
def __init__(self, collisions: Sequence[str]):
self.collisions = list(collisions)
super().__init__(
f"Refusing to submit — {len(self.collisions)} recent "
f"submission(s) already cover this batch's claim_ids. "
f"Use --force to override. Collisions: {self.collisions}"
)
class BrowserNotReachableError(CyclonePipelineError):
"""The Vite dev server is not reachable on the configured browser
base URL. The upload phase can't run without it. Check that
`npm run dev` is up."""
def __init__(self, browser_base: str):
self.browser_base = browser_base
super().__init__(
f"Browser-side URL {browser_base!r} is not reachable. "
f"Start the frontend with `npm run dev` (default "
f"http://localhost:5173) or pass --browser-base."
)
```
- [ ] **Step 4: Run test to verify it passes**
```bash
cd /Users/openclaw/dev/cyclone-pipeline
pytest tests/test_exceptions.py -v
```
Expected: 3 passed.
- [ ] **Step 5: Commit**
```bash
cd /Users/openclaw/dev/cyclone-pipeline
git add src/cyclone_pipeline/exceptions.py tests/test_exceptions.py
git commit -m "feat(sp22): typed exceptions (UploadError, ParseError, AckTimeoutError, etc.)"
```
---
## Task 3: Logging setup
**Goal:** `structlog` configured for JSON output with the same shape as cyclone's SP18 logs. Used by every other module.
**Files:**
- Create: `/Users/openclaw/dev/cyclone-pipeline/src/cyclone_pipeline/logging_setup.py`
- Create: `/Users/openclaw/dev/cyclone-pipeline/tests/test_logging_setup.py`
- [ ] **Step 1: Write the failing test**
Write `/Users/openclaw/dev/cyclone-pipeline/tests/test_logging_setup.py`:
```python
"""Logging setup — JSON output to stderr, no PII scrubber needed
(we never log claim content; only IDs and paths)."""
import io
import json
import logging
import structlog
import pytest
from cyclone_pipeline.logging_setup import setup_logging, get_logger
@pytest.fixture
def captured_stderr(monkeypatch):
"""Redirect structlog's stderr output to a StringIO for assertions."""
buf = io.StringIO()
def _capture(_self, msg):
buf.write(str(msg) + "\n")
return None
# Replace the ConsoleRenderer/JSONRenderer chain with a capture sink
monkeypatch.setattr(
"structlog.PrintLoggerFactory", lambda file=None: structlog.PrintLoggerFactory(file=buf)
)
# Simpler: directly patch write to stderr
import sys
real_stderr = sys.stderr
sys.stderr = io.StringIO()
yield sys.stderr
sys.stderr = real_stderr
# Write captured output back to the test buffer
buf.write(sys.stderr.getvalue() if hasattr(sys.stderr, "getvalue") else "")
def test_setup_logging_emits_json(caplog):
"""After setup_logging, log records render as one JSON object per line
with the keys cyclone uses (ts, level, logger, msg, extra)."""
setup_logging(level="INFO")
log = get_logger("cyclone_pipeline.test")
log.info("hello", claim_id="CLM-1")
captured = caplog.text
assert "hello" in captured
# The structlog JSON renderer writes to stderr, not caplog by default.
# Use a direct capture:
import sys
real = sys.stderr
sys.stderr = io.StringIO()
try:
setup_logging(level="INFO")
log = get_logger("cyclone_pipeline.test")
log.info("hello", claim_id="CLM-1")
out = sys.stderr.getvalue()
finally:
sys.stderr = real
lines = [ln for ln in out.splitlines() if ln.startswith("{")]
assert lines, f"expected JSON line on stderr, got: {out!r}"
rec = json.loads(lines[-1])
assert rec["event"] == "hello"
assert rec["claim_id"] == "CLM-1"
assert rec["level"] == "info"
def test_get_logger_returns_named_logger():
log = get_logger("foo.bar")
assert isinstance(log, structlog.stdlib.BoundLogger)
assert log.bind is not None
```
- [ ] **Step 2: Run test to verify it fails**
```bash
cd /Users/openclaw/dev/cyclone-pipeline
pytest tests/test_logging_setup.py -v
```
Expected: ImportError.
- [ ] **Step 3: Write the implementation**
Write `/Users/openclaw/dev/cyclone-pipeline/src/cyclone_pipeline/logging_setup.py`:
```python
"""Structured JSON logging — matches cyclone's SP18 style.
Every record is one line of JSON on stderr with keys:
`event`, `level`, `timestamp`, `logger`. Optional kwargs are
merged into the top-level record. No PII scrubber — the agent
only logs IDs and paths, never claim content.
"""
from __future__ import annotations
import logging
import sys
import structlog
def setup_logging(level: str = "INFO") -> None:
"""Configure structlog for JSON output to stderr.
Safe to call multiple times — `structlog.configure` is idempotent.
"""
timestamper = structlog.processors.TimeStamper(fmt="iso", utc=True)
shared_processors = [
structlog.contextvars.merge_contextvars,
structlog.stdlib.add_log_level,
timestamper,
structlog.processors.StackInfoRenderer(),
]
structlog.configure(
processors=shared_processors + [
structlog.stdlib.ProcessorFormatter.wrap_for_formatter,
],
logger_factory=structlog.stdlib.LoggerFactory(),
wrapper_class=structlog.stdlib.BoundLogger,
cache_logger_on_first_use=True,
)
formatter = structlog.stdlib.ProcessorFormatter(
foreign_pre_chain=shared_processors,
processor=structlog.processors.JSONRenderer(),
)
handler = logging.StreamHandler(sys.stderr)
handler.setFormatter(formatter)
root = logging.getLogger()
# Replace handlers to avoid double-logging if called twice
root.handlers = [handler]
root.setLevel(level.upper())
def get_logger(name: str) -> structlog.stdlib.BoundLogger:
"""Return a bound structlog logger. Call `setup_logging()` first."""
return structlog.get_logger(name)
```
- [ ] **Step 4: Run test to verify it passes**
```bash
cd /Users/openclaw/dev/cyclone-pipeline
pytest tests/test_logging_setup.py -v
```
Expected: 2 passed.
- [ ] **Step 5: Commit**
```bash
cd /Users/openclaw/dev/cyclone-pipeline
git add src/cyclone_pipeline/logging_setup.py tests/test_logging_setup.py
git commit -m "feat(sp22): structlog JSON logging matching cyclone SP18 style"
```
---
## Task 4: API client — health + scheduler status
**Goal:** First slice of the `CycloneClient` async HTTP wrapper. Two methods: `health()` and `scheduler_status()`. Both retry on idempotent failures.
**Files:**
- Modify: `/Users/openclaw/dev/cyclone-pipeline/src/cyclone_pipeline/api_client.py` (create)
- Create: `/Users/openclaw/dev/cyclone-pipeline/tests/test_api_client.py`
- [ ] **Step 1: Write the failing test**
Write `/Users/openclaw/dev/cyclone-pipeline/tests/test_api_client.py`:
```python
"""CycloneClient — async HTTP wrapper for cyclone's published API.
Methods are typed and return Pydantic models where the response shape
is stable. All methods retry on idempotent failures (5xx, connection
errors) with `1s → 2s → 4s` backoff, max 3 attempts.
"""
import pytest
import respx
from httpx import Response, AsyncClient, HTTPError
from pydantic import BaseModel
from cyclone_pipeline.api_client import CycloneClient, HealthSnapshot, SchedulerStatus
@pytest.fixture
def client():
return CycloneClient(api_base="http://test:8000")
@pytest.mark.asyncio
@respx.mock
async def test_health_returns_typed_snapshot(client):
respx.get("http://test:8000/api/health").mock(
return_value=Response(
200,
json={
"status": "ok",
"version": "0.1.0",
"db": {"ok": True},
"scheduler": {"running": True, "interval_s": 60},
"pubsub": {},
"batch": {},
},
)
)
snap = await client.health()
assert isinstance(snap, HealthSnapshot)
assert snap.status == "ok"
assert snap.db_ok is True
assert snap.scheduler_running is True
@pytest.mark.asyncio
@respx.mock
async def test_health_retries_on_5xx(client):
route = respx.get("http://test:8000/api/health").mock(
side_effect=[
Response(503, json={"error": "db down"}),
Response(200, json={
"status": "ok", "version": "0.1.0",
"db": {"ok": True},
"scheduler": {"running": True, "interval_s": 60},
"pubsub": {}, "batch": {},
}),
]
)
snap = await client.health()
assert snap.status == "ok"
assert route.call_count == 2
@pytest.mark.asyncio
@respx.mock
async def test_scheduler_status(client):
respx.get("http://test:8000/api/admin/scheduler/status").mock(
return_value=Response(200, json={
"running": True,
"interval_s": 60,
"sftp_block": "co_medicaid",
"backup_scheduler_running": False,
"backup_interval_hours": 24.0,
"last_tick_at": "2026-06-21T15:30:00.123Z",
})
)
s = await client.scheduler_status()
assert isinstance(s, SchedulerStatus)
assert s.running is True
assert s.interval_s == 60
```
- [ ] **Step 2: Run test to verify it fails**
```bash
cd /Users/openclaw/dev/cyclone-pipeline
pytest tests/test_api_client.py -v
```
Expected: ImportError.
- [ ] **Step 3: Write the implementation (first slice)**
Write `/Users/openclaw/dev/cyclone-pipeline/src/cyclone_pipeline/api_client.py`:
```python
"""Async HTTP wrapper for cyclone's published FastAPI.
Each method is async, typed, and retries idempotent failures
(5xx, connect/read errors) with `1s → 2s → 4s` backoff, max 3
attempts. POSTs are not auto-retried after a successful send —
the caller is responsible for dedup. Tests use `respx` to mock
the routes.
"""
from __future__ import annotations
import asyncio
from typing import Any, Awaitable, Callable, TypeVar
import httpx
from pydantic import BaseModel, Field
from cyclone_pipeline.logging_setup import get_logger
log = get_logger("cyclone_pipeline.api_client")
T = TypeVar("T", bound=BaseModel)
DEFAULT_TIMEOUT = httpx.Timeout(connect=10.0, read=30.0, write=10.0, pool=5.0)
MAX_RETRIES = 3
BACKOFF_S = (1.0, 2.0, 4.0)
# --- Response models ---------------------------------------------------------
class HealthSnapshot(BaseModel):
status: str
version: str
db_ok: bool
scheduler_running: bool
class SchedulerStatus(BaseModel):
running: bool
interval_s: int
sftp_block: str | None = None
backup_scheduler_running: bool = False
backup_interval_hours: float = 24.0
last_tick_at: str | None = None
# --- Client ------------------------------------------------------------------
class CycloneClient:
"""Async client for cyclone's API. Owns the `httpx.AsyncClient`."""
def __init__(
self,
api_base: str,
timeout: httpx.Timeout | None = None,
):
self.api_base = api_base.rstrip("/")
self._timeout = timeout or DEFAULT_TIMEOUT
self._http: httpx.AsyncClient | None = None
async def __aenter__(self) -> "CycloneClient":
self._http = httpx.AsyncClient(
base_url=self.api_base, timeout=self._timeout
)
return self
async def __aexit__(self, *exc) -> None:
if self._http is not None:
await self._http.aclose()
self._http = None
# --- internal: idempotent GET with retry -------------------------------
async def _get_typed(
self, path: str, model: type[T]
) -> T:
assert self._http is not None, "use `async with CycloneClient(...)`"
last_exc: Exception | None = None
for attempt in range(MAX_RETRIES):
try:
resp = await self._http.get(path)
resp.raise_for_status()
return model.model_validate(resp.json())
except (httpx.HTTPStatusError, httpx.RequestError) as e:
last_exc = e
if attempt < MAX_RETRIES - 1:
wait_s = BACKOFF_S[attempt]
log.warning(
"api_get_retry",
path=path, attempt=attempt + 1, wait_s=wait_s,
error=str(e),
)
await asyncio.sleep(wait_s)
assert last_exc is not None
raise last_exc
# --- methods -----------------------------------------------------------
async def health(self) -> HealthSnapshot:
raw = await self._get_typed("/api/health", _RawHealth)
return HealthSnapshot(
status=raw.status,
version=raw.version,
db_ok=raw.db.get("ok", False),
scheduler_running=raw.scheduler.get("running", False),
)
async def scheduler_status(self) -> SchedulerStatus:
return await self._get_typed(
"/api/admin/scheduler/status", SchedulerStatus
)
# --- Private wire-format models ---------------------------------------------
# Cyclone's `/api/health` returns a richer shape than `HealthSnapshot`
# exposes; the public method maps the wire format to the lean model.
class _RawHealth(BaseModel):
status: str
version: str
db: dict[str, Any]
scheduler: dict[str, Any]
pubsub: dict[str, Any] = Field(default_factory=dict)
batch: dict[str, Any] = Field(default_factory=dict)
```
- [ ] **Step 4: Run test to verify it passes**
```bash
cd /Users/openclaw/dev/cyclone-pipeline
pytest tests/test_api_client.py -v
```
Expected: 3 passed.
- [ ] **Step 5: Commit**
```bash
cd /Users/openclaw/dev/cyclone-pipeline
git add src/cyclone_pipeline/api_client.py tests/test_api_client.py
git commit -m "feat(sp22): CycloneClient — health + scheduler_status with retry"
```
---
## Task 5: API client — claims + parse-837 (upload surrogate)
**Goal:** Add `list_claims()` and `parse_837()` methods. The latter is used by the orchestrator as a fallback to the browser upload (and for tests that don't want to drive Playwright).
**Files:**
- Modify: `/Users/openclaw/dev/cyclone-pipeline/src/cyclone_pipeline/api_client.py`
- Modify: `/Users/openclaw/dev/cyclone-pipeline/tests/test_api_client.py`
- [ ] **Step 1: Append failing tests to `test_api_client.py`**
Append to `/Users/openclaw/dev/cyclone-pipeline/tests/test_api_client.py`:
```python
from cyclone_pipeline.api_client import ClaimSummary
@pytest.mark.asyncio
@respx.mock
async def test_list_claims_filters_by_since(client):
route = respx.get("http://test:8000/api/claims").mock(
return_value=Response(200, json=[
{"id": "CLM-1", "submission_date": "2026-06-21T15:00:00Z"},
{"id": "CLM-2", "submission_date": "2026-06-21T16:00:00Z"},
])
)
out = await client.list_claims(since="2026-06-21T15:30:00Z")
assert len(out) == 2
assert out[0].id == "CLM-1"
# Verify the since param was forwarded
assert "since" in route.calls.last.request.url.params
@pytest.mark.asyncio
@respx.mock
async def test_parse_837_posts_file_and_returns_summary(client, sample_837p_path):
route = respx.post("http://test:8000/api/parse-837").mock(
return_value=Response(200, json={
"total_claims": 1, "passed": 1, "failed": 0,
"claims": [{"claim_id": "CLM-1"}],
})
)
summary = await client.parse_837(sample_837p_path, payer="co_medicaid")
assert summary.total_claims == 1
assert summary.claims[0]["claim_id"] == "CLM-1"
# Confirm multipart upload
content_type = route.calls.last.request.headers["content-type"]
assert "multipart/form-data" in content_type
```
- [ ] **Step 2: Run test to verify it fails**
```bash
cd /Users/openclaw/dev/cyclone-pipeline
pytest tests/test_api_client.py::test_list_claims_filters_by_since -v
```
Expected: ImportError (ClaimSummary not defined).
- [ ] **Step 3: Extend `api_client.py`**
Add to `/Users/openclaw/dev/cyclone-pipeline/src/cyclone_pipeline/api_client.py` (in the "Response models" section near the top):
```python
class ClaimSummary(BaseModel):
id: str
submission_date: str
class ParseSummary(BaseModel):
total_claims: int
passed: int
failed: int
claims: list[dict[str, Any]] = Field(default_factory=list)
```
Add the methods to the `CycloneClient` class (below `scheduler_status`):
```python
async def list_claims(
self, since: str | None = None, limit: int = 100
) -> list[ClaimSummary]:
params: dict[str, str] = {"limit": str(limit)}
if since:
params["since"] = since
assert self._http is not None
last_exc: Exception | None = None
for attempt in range(MAX_RETRIES):
try:
resp = await self._http.get(
"/api/claims", params=params
)
resp.raise_for_status()
return [ClaimSummary.model_validate(c) for c in resp.json()]
except (httpx.HTTPStatusError, httpx.RequestError) as e:
last_exc = e
if attempt < MAX_RETRIES - 1:
await asyncio.sleep(BACKOFF_S[attempt])
assert last_exc is not None
raise last_exc
async def parse_837(
self, file_path, payer: str = "co_medicaid"
) -> ParseSummary:
"""POST the .edi/.txt file to cyclone's parse endpoint.
Returns a typed summary.
"""
from pathlib import Path # local import keeps the top of the file lean
path = Path(file_path)
assert self._http is not None
with path.open("rb") as f:
resp = await self._http.post(
"/api/parse-837",
files={"file": (path.name, f, "text/plain")},
params={"payer": payer},
)
resp.raise_for_status()
return ParseSummary.model_validate(resp.json())
```
- [ ] **Step 4: Run test to verify it passes**
```bash
cd /Users/openclaw/dev/cyclone-pipeline
pytest tests/test_api_client.py -v
```
Expected: 5 passed (the 3 from Task 4 + the 2 new).
- [ ] **Step 5: Commit**
```bash
cd /Users/openclaw/dev/cyclone-pipeline
git add src/cyclone_pipeline/api_client.py tests/test_api_client.py
git commit -m "feat(sp22): CycloneClient — list_claims + parse_837 (file upload)"
```
---
## Task 6: API client — clearhouse submit + scheduler tick + processed-files
**Goal:** Add the SFTP submit method, the manual scheduler tick, and the processed-files listing used for dedup + HTML scanning.
**Files:**
- Modify: `/Users/openclaw/dev/cyclone-pipeline/src/cyclone_pipeline/api_client.py`
- Modify: `/Users/openclaw/dev/cyclone-pipeline/tests/test_api_client.py`
- [ ] **Step 1: Append failing tests**
Append to `/Users/openclaw/dev/cyclone-pipeline/tests/test_api_client.py`:
```python
from cyclone_pipeline.api_client import (
SubmitResult, ProcessedFile, AckKind,
)
from cyclone_pipeline.exceptions import IdempotencyError
@pytest.mark.asyncio
@respx.mock
async def test_clearhouse_submit_returns_submission_id(client):
respx.post("http://test:8000/api/clearhouse/submit").mock(
return_value=Response(200, json={
"submission_id": "SUB-42",
"filename": "11525703-837P-20260621143012345-1of1.txt",
"claim_count": 17,
})
)
result = await client.clearhouse_submit(payer_id="CO_TXIX", tx="837P")
assert result.submission_id == "SUB-42"
assert result.filename.endswith(".txt")
assert result.claim_count == 17
@pytest.mark.asyncio
@respx.mock
async def test_clearhouse_submit_409_raises_idempotency_error(client):
respx.post("http://test:8000/api/clearhouse/submit").mock(
return_value=Response(409, json={
"error": "duplicate",
"detail": "claim_ids already submitted in last 24h",
"collisions": ["11525703-837P-20260620181814559-1of1.txt"],
})
)
with pytest.raises(IdempotencyError) as exc_info:
await client.clearhouse_submit(payer_id="CO_TXIX", tx="837P")
assert "11525703-837P-20260620181814559-1of1.txt" in exc_info.value.collisions
@pytest.mark.asyncio
@respx.mock
async def test_scheduler_tick(client):
respx.post("http://test:8000/api/admin/scheduler/tick").mock(
return_value=Response(200, json={
"ticked_at": "2026-06-21T15:30:00Z",
"files_downloaded": 1,
})
)
result = await client.scheduler_tick()
assert result.files_downloaded == 1
@pytest.mark.asyncio
@respx.mock
async def test_processed_files_lists_inbound(client):
respx.get("http://test:8000/api/admin/scheduler/processed-files").mock(
return_value=Response(200, json=[
{"filename": "TP11525703-999_…-1of1.x12", "kind": "999",
"processed_at": "2026-06-21T15:30:00Z", "size_bytes": 2048},
{"filename": "gainwell_status.htm", "kind": "html",
"processed_at": "2026-06-21T15:30:00Z", "size_bytes": 4096},
])
)
files = await client.processed_files(since="2026-06-21T15:00:00Z")
assert len(files) == 2
html = [f for f in files if f.kind == "html"]
assert len(html) == 1
assert html[0].filename == "gainwell_status.htm"
```
- [ ] **Step 2: Run test to verify it fails**
```bash
cd /Users/openclaw/dev/cyclone-pipeline
pytest tests/test_api_client.py::test_clearhouse_submit_returns_submission_id -v
```
Expected: ImportError.
- [ ] **Step 3: Extend `api_client.py`**
Add to the response-models section:
```python
class SubmitResult(BaseModel):
submission_id: str
filename: str
claim_count: int = 0
class ProcessedFile(BaseModel):
filename: str
kind: str # "999" | "ta1" | "271" | "277" | "277ca" | "835" | "html" | "unknown"
processed_at: str
size_bytes: int = 0
class TickResult(BaseModel):
ticked_at: str
files_downloaded: int = 0
```
Add the methods to the `CycloneClient` class:
```python
async def clearhouse_submit(
self, payer_id: str, tx: str = "837P"
) -> SubmitResult:
assert self._http is not None
# Do NOT auto-retry — would double-submit.
resp = await self._http.post(
"/api/clearhouse/submit",
json={"payer_id": payer_id, "transaction_type": tx},
)
if resp.status_code == 409:
body = resp.json()
raise IdempotencyError(body.get("collisions", []))
resp.raise_for_status()
return SubmitResult.model_validate(resp.json())
async def scheduler_tick(self) -> TickResult:
assert self._http is not None
# Manual tick is idempotent; safe to retry.
last_exc: Exception | None = None
for attempt in range(MAX_RETRIES):
try:
resp = await self._http.post(
"/api/admin/scheduler/tick"
)
resp.raise_for_status()
return TickResult.model_validate(resp.json())
except (httpx.HTTPStatusError, httpx.RequestError) as e:
last_exc = e
if attempt < MAX_RETRIES - 1:
await asyncio.sleep(BACKOFF_S[attempt])
assert last_exc is not None
raise last_exc
async def processed_files(self, since: str) -> list[ProcessedFile]:
assert self._http is not None
last_exc: Exception | None = None
for attempt in range(MAX_RETRIES):
try:
resp = await self._http.get(
"/api/admin/scheduler/processed-files",
params={"since": since},
)
resp.raise_for_status()
return [ProcessedFile.model_validate(f) for f in resp.json()]
except (httpx.HTTPStatusError, httpx.RequestError) as e:
last_exc = e
if attempt < MAX_RETRIES - 1:
await asyncio.sleep(BACKOFF_S[attempt])
assert last_exc is not None
raise last_exc
```
- [ ] **Step 4: Run test to verify it passes**
```bash
cd /Users/openclaw/dev/cyclone-pipeline
pytest tests/test_api_client.py -v
```
Expected: 9 passed.
- [ ] **Step 5: Commit**
```bash
cd /Users/openclaw/dev/cyclone-pipeline
git add src/cyclone_pipeline/api_client.py tests/test_api_client.py
git commit -m "feat(sp22): CycloneClient — clearhouse_submit + scheduler_tick + processed_files"
```
---
## Task 7: API client — ta1-acks + acks + remittances
**Goal:** Add the three ACK list methods. TA1 and 999 are the only ACKs the agent polls for; remittances is for `check-835`.
**Files:**
- Modify: `/Users/openclaw/dev/cyclone-pipeline/src/cyclone_pipeline/api_client.py`
- Modify: `/Users/openclaw/dev/cyclone-pipeline/tests/test_api_client.py`
- [ ] **Step 1: Append failing tests**
Append to `/Users/openclaw/dev/cyclone-pipeline/tests/test_api_client.py`:
```python
from cyclone_pipeline.api_client import Ta1Ack, Ack999, RemittanceSummary
@pytest.mark.asyncio
@respx.mock
async def test_list_ta1_acks(client):
respx.get("http://test:8000/api/ta1-acks").mock(
return_value=Response(200, json=[
{"id": "TA1-1", "isa13": "000000001",
"accepted": True, "received_at": "2026-06-21T15:30:00Z"},
])
)
acks = await client.list_ta1_acks(since="2026-06-21T15:00:00Z")
assert len(acks) == 1
assert acks[0].id == "TA1-1"
assert acks[0].accepted is True
@pytest.mark.asyncio
@respx.mock
async def test_list_acks_999(client):
respx.get("http://test:8000/api/acks").mock(
return_value=Response(200, json=[
{"id": "999-1", "status": "A",
"original_filename": "11525703-837P-…-1of1.txt",
"received_at": "2026-06-21T15:30:00Z"},
])
)
acks = await client.list_acks(since="2026-06-21T15:00:00Z")
assert len(acks) == 1
assert acks[0].status == "A"
@pytest.mark.asyncio
@respx.mock
async def test_list_remittances(client):
respx.get("http://test:8000/api/remittances").mock(
return_value=Response(200, json=[
{"id": "REM-1", "received_date": "2026-06-22T08:00:00Z",
"payer_claim_control_numbers": ["CLM-001"]},
])
)
remits = await client.list_remittances(since="2026-06-21T15:00:00Z")
assert len(remits) == 1
assert remits[0].id == "REM-1"
assert "CLM-001" in remits[0].payer_claim_control_numbers
```
- [ ] **Step 2: Run test to verify it fails**
```bash
cd /Users/openclaw/dev/cyclone-pipeline
pytest tests/test_api_client.py::test_list_ta1_acks -v
```
Expected: ImportError.
- [ ] **Step 3: Extend `api_client.py`**
Add to the response-models section:
```python
class Ta1Ack(BaseModel):
id: str
isa13: str
accepted: bool
received_at: str
class Ack999(BaseModel):
id: str
status: str # "A" | "R" | "E"
original_filename: str
received_at: str
class RemittanceSummary(BaseModel):
id: str
received_date: str
payer_claim_control_numbers: list[str] = Field(default_factory=list)
```
Add the methods to the `CycloneClient` class:
```python
async def list_ta1_acks(self, since: str) -> list[Ta1Ack]:
assert self._http is not None
resp = await self._http.get(
"/api/ta1-acks", params={"since": since}
)
resp.raise_for_status()
return [Ta1Ack.model_validate(a) for a in resp.json()]
async def list_acks(self, since: str) -> list[Ack999]:
assert self._http is not None
resp = await self._http.get(
"/api/acks", params={"since": since}
)
resp.raise_for_status()
return [Ack999.model_validate(a) for a in resp.json()]
async def list_remittances(
self, since: str
) -> list[RemittanceSummary]:
assert self._http is not None
resp = await self._http.get(
"/api/remittances", params={"since": since}
)
resp.raise_for_status()
return [
RemittanceSummary.model_validate(r) for r in resp.json()
]
```
- [ ] **Step 4: Run test to verify it passes**
```bash
cd /Users/openclaw/dev/cyclone-pipeline
pytest tests/test_api_client.py -v
```
Expected: 12 passed.
- [ ] **Step 5: Commit**
```bash
cd /Users/openclaw/dev/cyclone-pipeline
git add src/cyclone_pipeline/api_client.py tests/test_api_client.py
git commit -m "feat(sp22): CycloneClient — list_ta1_acks + list_acks + list_remittances"
```
---
## Task 8: State module (resumable run state)
**Goal:** A Pydantic model for run state, written to `run.state.json` after each phase. The orchestrator reads it on resume to skip completed phases.
**Files:**
- Create: `/Users/openclaw/dev/cyclone-pipeline/src/cyclone_pipeline/state.py`
- Create: `/Users/openclaw/dev/cyclone-pipeline/tests/test_state.py`
- [ ] **Step 1: Write the failing test**
Write `/Users/openclaw/dev/cyclone-pipeline/tests/test_state.py`:
```python
"""Resumable run state — a Pydantic model written to run.state.json
after each phase. Read on resume to skip completed phases."""
import json
import pytest
from pathlib import Path
from cyclone_pipeline.state import RunState, PhaseRecord
def test_run_state_round_trip(tmp_path: Path):
state = RunState(
run_id="2026-06-21-1430-001",
input_path="/tmp/foo.edi",
started_at="2026-06-21T14:30:00-06:00",
api_base="http://127.0.0.1:8000",
browser_base="http://127.0.0.1:5173",
completed_phases=[
PhaseRecord(n=1, name="preflight", status="ok",
duration_s=0.4),
PhaseRecord(n=2, name="upload", status="ok",
duration_s=2.1, attempts=1),
],
pending_data={
"batch_id": "BATCH-42",
"claim_ids": ["CLM-001", "CLM-002"],
"submission_id": "SUB-42",
"submission_filename": "11525703-837P-…-1of1.txt",
},
)
path = tmp_path / "run.state.json"
state.write(path)
assert path.exists()
loaded = RunState.read(path)
assert loaded.run_id == "2026-06-21-1430-001"
assert len(loaded.completed_phases) == 2
assert loaded.pending_data["claim_ids"] == ["CLM-001", "CLM-002"]
def test_resume_phase_returns_lowest_unfinished():
state = RunState(
run_id="r", input_path="x", started_at="now",
api_base="a", browser_base="b",
completed_phases=[
PhaseRecord(n=1, name="preflight", status="ok",
duration_s=0.1),
PhaseRecord(n=2, name="upload", status="ok",
duration_s=0.1),
PhaseRecord(n=3, name="verify_parse", status="ok",
duration_s=0.1),
],
)
# 7 phases total; we want to resume at phase 4
assert state.next_phase_n() == 4
def test_resume_phase_returns_1_if_nothing_completed():
state = RunState(
run_id="r", input_path="x", started_at="now",
api_base="a", browser_base="b",
completed_phases=[],
)
assert state.next_phase_n() == 1
```
- [ ] **Step 2: Run test to verify it fails**
```bash
cd /Users/openclaw/dev/cyclone-pipeline
pytest tests/test_state.py -v
```
Expected: ImportError.
- [ ] **Step 3: Write the implementation**
Write `/Users/openclaw/dev/cyclone-pipeline/src/cyclone_pipeline/state.py`:
```python
"""Resumable run state.
`RunState` is written to `run.state.json` after every phase's
postcondition is met. On `resume`, the orchestrator reads this file
and skips every completed phase. `pending_data` is a free-form dict
that captures cross-phase outputs (batch_id, claim_ids, submission_id).
"""
from __future__ import annotations
from pathlib import Path
from typing import Any
from pydantic import BaseModel, Field
TOTAL_PHASES = 7
class PhaseRecord(BaseModel):
n: int
name: str
status: str # "ok" | "warn" | "fail" | "soft_fail" | "hard_fail" | "timeout"
duration_s: float
attempts: int = 1
detail: dict[str, Any] = Field(default_factory=dict)
class RunState(BaseModel):
run_id: str
input_path: str
started_at: str
api_base: str
browser_base: str
completed_phases: list[PhaseRecord] = Field(default_factory=list)
pending_data: dict[str, Any] = Field(default_factory=dict)
def next_phase_n(self) -> int:
"""Return the 1-based phase number to resume from. If all phases
are done, returns TOTAL_PHASES + 1 (caller treats as 'done')."""
if not self.completed_phases:
return 1
return max(p.n for p in self.completed_phases) + 1
def is_phase_done(self, n: int) -> bool:
return any(p.n == n for p in self.completed_phases)
def write(self, path: Path) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(self.model_dump_json(indent=2))
@classmethod
def read(cls, path: Path) -> "RunState":
return cls.model_validate_json(path.read_text())
```
- [ ] **Step 4: Run test to verify it passes**
```bash
cd /Users/openclaw/dev/cyclone-pipeline
pytest tests/test_state.py -v
```
Expected: 3 passed.
- [ ] **Step 5: Commit**
```bash
cd /Users/openclaw/dev/cyclone-pipeline
git add src/cyclone_pipeline/state.py tests/test_state.py
git commit -m "feat(sp22): RunState pydantic model + JSON round-trip + resume helpers"
```
---
## Task 9: Screenshots module
**Goal:** Stable filenames, dated folder, and a helper to take + save a Playwright screenshot to the right place.
**Files:**
- Create: `/Users/openclaw/dev/cyclone-pipeline/src/cyclone_pipeline/screenshots.py`
- Create: `/Users/openclaw/dev/cyclone-pipeline/tests/test_screenshots.py`
- [ ] **Step 1: Write the failing test**
Write `/Users/openclaw/dev/cyclone-pipeline/tests/test_screenshots.py`:
```python
"""Screenshot helpers — dated folder + stable filename pattern."""
from pathlib import Path
import pytest
from cyclone_pipeline.screenshots import (
ScreenshotWriter, screenshot_path,
)
def test_run_dir_path(tmp_path: Path):
"""The dated folder name is `YYYY-MM-DD-HHMM-NNN` (3-digit sequence
so concurrent runs don't collide)."""
p = ScreenshotWriter.root_dir(tmp_path, "2026-06-21-1430-001")
assert str(p).endswith("2026-06-21-1430-001")
def test_screenshot_path_is_stable():
"""The same logical step name always produces the same filename
(e.g., `02-upload.png`, never `02-upload-1234567890.png`). The
per-attempt suffix is only added when explicitly requested."""
p = screenshot_path(
step="02-upload", attempt=None
)
assert p.name == "02-upload.png"
p2 = screenshot_path(step="02-upload", attempt=1)
assert p2.name == "02-upload-1.png"
p3 = screenshot_path(step="99-final-inbox", attempt=None)
assert p3.name == "99-final-inbox.png"
def test_writer_writes_png(tmp_path: Path):
writer = ScreenshotWriter(tmp_path, "test-run")
out = writer.full_path("01-preflight")
out.parent.mkdir(parents=True, exist_ok=True)
out.write_bytes(b"\x89PNG\r\n\x1a\n")
assert out.exists()
assert out.read_bytes()[:4] == b"\x89PNG"
```
- [ ] **Step 2: Run test to verify it fails**
```bash
cd /Users/openclaw/dev/cyclone-pipeline
pytest tests/test_screenshots.py -v
```
Expected: ImportError.
- [ ] **Step 3: Write the implementation**
Write `/Users/openclaw/dev/cyclone-pipeline/src/cyclone_pipeline/screenshots.py`:
```python
"""Screenshot writer — dated folder + stable filenames.
Every screenshot the agent takes lands in
`./runs/{run_id}/screenshots/{step}.png` (or `{step}-{attempt}.png`
on retries). The run_id is the human-readable `YYYY-MM-DD-HHMM-NNN`
pattern; the screenshot step is a stable string from a small enum
(01-preflight, 02-upload, 03-parse-complete, …).
"""
from __future__ import annotations
from pathlib import Path
from dataclasses import dataclass
@dataclass
class ScreenshotWriter:
base_dir: Path
run_id: str
def __post_init__(self) -> None:
self._root = self.base_dir / self.run_id / "screenshots"
@classmethod
def root_dir(cls, base_dir: Path, run_id: str) -> Path:
return base_dir / run_id
def full_path(self, step: str, attempt: int | None = None) -> Path:
"""Return the absolute path the screenshot should be written to.
Does NOT create the directory — caller decides when to mkdir."""
return screenshot_path(step, attempt, base=self._root)
def screenshot_path(
step: str, attempt: int | None = None, base: Path | None = None
) -> Path:
"""Compute a stable filename for a given step. `step` is the
logical name (e.g., '02-upload'). If `attempt` is provided, a
numeric suffix is appended so retries don't overwrite each other."""
name = step if attempt is None else f"{step}-{attempt}"
if base is None:
return Path(name + ".png")
return base / (name + ".png")
```
- [ ] **Step 4: Run test to verify it passes**
```bash
cd /Users/openclaw/dev/cyclone-pipeline
pytest tests/test_screenshots.py -v
```
Expected: 3 passed.
- [ ] **Step 5: Commit**
```bash
cd /Users/openclaw/dev/cyclone-pipeline
git add src/cyclone_pipeline/screenshots.py tests/test_screenshots.py
git commit -m "feat(sp22): ScreenshotWriter + stable filename helper"
```
---
## Task 10: Selectors module
**Goal:** Centralized Playwright selectors with fallbacks. The selectors module owns the mapping from "logical action" (e.g., "click Parse button") to actual CSS/XPath/role selectors.
**Files:**
- Create: `/Users/openclaw/dev/cyclone-pipeline/src/cyclone_pipeline/selectors.py`
- Create: `/Users/openclaw/dev/cyclone-pipeline/tests/test_selectors.py`
- [ ] **Step 1: Write the failing test**
Write `/Users/openclaw/dev/cyclone-pipeline/tests/test_selectors.py`:
```python
"""Centralized Playwright selectors.
The selectors module owns the mapping from "logical action" to the
actual CSS / role / text selectors. We try the most specific
selector first and fall back to broader ones — this keeps the agent
working even when the cyclone UI tweaks a class name or restructures
a component, as long as the user-visible text is stable.
"""
import pytest
from cyclone_pipeline.selectors import (
upload_drop_zone, upload_file_input, parse_button,
parse_progress_bar, parse_success_toast,
)
def test_upload_drop_zone_is_role_based():
"""The drop zone is a role=button region (it's clickable but also
a drop target). Selector should target the data-testid if present,
else fall back to role + text."""
selectors = upload_drop_zone()
assert any("role=button" in s or "[data-testid" in s for s in selectors)
def test_upload_file_input_is_the_hidden_input():
"""The file input is the hidden `<input type=file>` behind the drop
zone. We select by type, not by class — classnames are unstable."""
selectors = upload_file_input()
assert 'input[type="file"]' in selectors
def test_parse_button_targets_text():
"""The Parse button text is stable: 'Parse' or 'Parsing…' while
running. We accept both."""
selectors = parse_button()
assert any("Parse" in s for s in selectors)
def test_progress_bar_targets_text():
"""The progress bar's text is `${n}%`. We wait for that exact text."""
selectors = parse_progress_bar()
assert any("100%" in s for s in selectors)
```
- [ ] **Step 2: Run test to verify it fails**
```bash
cd /Users/openclaw/dev/cyclone-pipeline
pytest tests/test_selectors.py -v
```
Expected: ImportError.
- [ ] **Step 3: Write the implementation**
Write `/Users/openclaw/dev/cyclone-pipeline/src/cyclone_pipeline/selectors.py`:
```python
"""Centralized Playwright selectors with fallbacks.
Each helper returns a list of selectors ordered from most specific
to least. The browser wrapper tries them in order and uses the
first that matches.
"""
from __future__ import annotations
def upload_drop_zone() -> list[str]:
"""The drop zone / click-to-pick region on the Upload page."""
return [
'[data-testid="upload-dropzone"]',
'role=button[name="Drop a file here, or click to choose"]',
'role=button[name="Release to upload"]',
]
def upload_file_input() -> list[str]:
"""The hidden file input. We select by type because the classname
is unstable across React/Vite versions."""
return ['input[type="file"]']
def parse_button() -> list[str]:
"""The 'Parse' button. We accept both 'Parse' (idle) and 'Parsing…'
(running) so we can match regardless of state."""
return [
'button:has-text("Parse"):not([disabled])',
'role=button[name="Parse"]:not([disabled])',
]
def parse_progress_bar() -> list[str]:
"""The progress bar's text is `100%` when complete."""
return [
'text="100%"',
'[data-testid="upload-progress"] >> text="100%"',
]
def parse_success_toast() -> list[str]:
"""The success toast that appears after parse completes.
Format: `Parsed N claims · M passed · K failed`."""
return [
'text=/Parsed \\d+ claim(s|payments)/',
]
```
- [ ] **Step 4: Run test to verify it passes**
```bash
cd /Users/openclaw/dev/cyclone-pipeline
pytest tests/test_selectors.py -v
```
Expected: 4 passed.
- [ ] **Step 5: Commit**
```bash
cd /Users/openclaw/dev/cyclone-pipeline
git add src/cyclone_pipeline/selectors.py tests/test_selectors.py
git commit -m "feat(sp22): centralized Playwright selectors with fallbacks"
```
---
## Task 11: Waiters module
**Goal:** Reusable wait primitives: `wait_for(predicate, timeout, poll_interval)`, `wait_for_ta1`, `wait_for_999`. Used by the orchestrator to poll for ACKs.
**Files:**
- Create: `/Users/openclaw/dev/cyclone-pipeline/src/cyclone_pipeline/waiters.py`
- Create: `/Users/openclaw/dev/cyclone-pipeline/tests/test_waiters.py`
- [ ] **Step 1: Write the failing test**
Write `/Users/openclaw/dev/cyclone-pipeline/tests/test_waiters.py`:
```python
"""Reusable wait primitives — backoff + timeout + predicate resolution.
The orchestrator uses these to poll the API for ACKs. The
`wait_for` primitive is the building block; the higher-level
helpers (wait_for_ta1, wait_for_999, wait_for_claim_state) are
thin wrappers.
"""
import asyncio
import pytest
from cyclone_pipeline.waiters import wait_for
from cyclone_pipeline.exceptions import AckTimeoutError
@pytest.mark.asyncio
async def test_wait_for_resolves_when_predicate_true():
"""If the predicate returns truthy on the first poll, wait_for
returns immediately."""
calls = []
def pred() -> int | None:
calls.append(1)
return 42
result = await wait_for(
pred, timeout_s=5.0, poll_interval_s=0.1, label="test"
)
assert result == 42
assert len(calls) == 1
@pytest.mark.asyncio
async def test_wait_for_resolves_on_later_poll():
"""If the predicate returns None the first two times, then truthy,
wait_for returns the truthy value."""
calls = []
def pred() -> int | None:
calls.append(1)
return 42 if len(calls) >= 3 else None
result = await wait_for(
pred, timeout_s=5.0, poll_interval_s=0.01, label="test"
)
assert result == 42
assert len(calls) == 3
@pytest.mark.asyncio
async def test_wait_for_raises_after_timeout():
"""If the predicate never returns truthy, wait_for raises
AckTimeoutError after the configured timeout."""
def pred() -> int | None:
return None
with pytest.raises(AckTimeoutError) as exc_info:
await wait_for(
pred, timeout_s=0.2, poll_interval_s=0.05, label="ta1"
)
assert exc_info.value.ack_kind == "ta1"
assert exc_info.value.timeout_s == pytest.approx(0.2, abs=0.1)
```
- [ ] **Step 2: Run test to verify it fails**
```bash
cd /Users/openclaw/dev/cyclone-pipeline
pytest tests/test_waiters.py -v
```
Expected: ImportError.
- [ ] **Step 3: Write the implementation**
Write `/Users/openclaw/dev/cyclone-pipeline/src/cyclone_pipeline/waiters.py`:
```python
"""Reusable async wait primitives.
The core is `wait_for(predicate, timeout_s, poll_interval_s, label)`.
It polls a sync predicate (no await) at the configured interval
and returns the first truthy value. On timeout, raises
`AckTimeoutError` with the label + timeout for the report.
"""
from __future__ import annotations
import asyncio
import time
from typing import Callable, TypeVar
from cyclone_pipeline.exceptions import AckTimeoutError
T = TypeVar("T")
async def wait_for(
predicate: Callable[[], T | None],
*,
timeout_s: float,
poll_interval_s: float,
label: str,
) -> T:
"""Poll `predicate` every `poll_interval_s` until it returns truthy
or `timeout_s` elapses. Raises `AckTimeoutError` on timeout.
The predicate is sync (it usually reads a cached API result). If
you need an async predicate, wrap it in `asyncio.iscoroutine`.
"""
deadline = time.monotonic() + timeout_s
while True:
result = predicate()
if result is not None:
return result
if time.monotonic() >= deadline:
raise AckTimeoutError(label, int(timeout_s))
await asyncio.sleep(poll_interval_s)
```
- [ ] **Step 4: Run test to verify it passes**
```bash
cd /Users/openclaw/dev/cyclone-pipeline
pytest tests/test_waiters.py -v
```
Expected: 3 passed.
- [ ] **Step 5: Commit**
```bash
cd /Users/openclaw/dev/cyclone-pipeline
git add src/cyclone_pipeline/waiters.py tests/test_waiters.py
git commit -m "feat(sp22): wait_for primitive + AckTimeoutError on timeout"
```
---
## Task 12: Report module (JSON + Markdown)
**Goal:** Two writers: `write_report_json()` and `write_report_md()`. The orchestrator calls both at the end of phase 7. Output shape matches §5 of the spec.
**Files:**
- Create: `/Users/openclaw/dev/cyclone-pipeline/src/cyclone_pipeline/report.py`
- Create: `/Users/openclaw/dev/cyclone-pipeline/tests/test_report.py`
- [ ] **Step 1: Write the failing test**
Write `/Users/openclaw/dev/cyclone-pipeline/tests/test_report.py`:
```python
"""Report writers — JSON for agents, Markdown for humans."""
import json
from pathlib import Path
import pytest
from cyclone_pipeline.report import (
RunReport, PhaseResult, write_report_json, write_report_md,
)
@pytest.fixture
def sample_report() -> RunReport:
return RunReport(
run_id="2026-06-21-1430-001",
input_path="/tmp/foo.edi",
input_size_bytes=12700,
result="pass",
result_detail="TA1 + 999 received, 835 deferred",
started_at="2026-06-21T14:30:00-06:00",
finished_at="2026-06-21T14:32:14-06:00",
duration_s=134.0,
phases=[
PhaseResult(n=1, name="preflight", status="ok", duration_s=0.4),
PhaseResult(n=2, name="upload", status="ok", duration_s=2.1),
PhaseResult(n=3, name="verify_parse", status="ok",
duration_s=0.8, detail={"claim_ids": ["CLM-001"]}),
PhaseResult(n=4, name="submit", status="ok", duration_s=4.3,
detail={"filename": "11525703-837P-…-1of1.txt"}),
PhaseResult(n=5, name="wait_ta1", status="ok", duration_s=12.0,
detail={"ta1_id": "TA1-42", "accepted": True}),
PhaseResult(n=6, name="wait_999", status="ok", duration_s=74.0,
detail={"ack_999_id": "999-43", "status": "A"}),
PhaseResult(n=7, name="scan_and_report", status="warn",
duration_s=0.2,
detail={"html_files": ["gainwell_status.htm"]}),
],
expected_835_by="2026-06-28",
next_steps={
"check_835_cmd": "cyclone-pipeline check-835 2026-06-21-1430-001",
},
screenshots=["01-preflight.png", "02-upload-1.png",
"99-final-inbox.png"],
)
def test_json_round_trip(sample_report, tmp_path):
p = tmp_path / "report.json"
write_report_json(sample_report, p)
assert p.exists()
data = json.loads(p.read_text())
assert data["run_id"] == "2026-06-21-1430-001"
assert data["result"] == "pass"
assert len(data["phases"]) == 7
assert data["phases"][2]["detail"]["claim_ids"] == ["CLM-001"]
def test_markdown_contains_phase_table(sample_report, tmp_path):
p = tmp_path / "report.md"
write_report_md(sample_report, p)
text = p.read_text()
assert "# Cyclone pipeline run" in text
assert "PASS" in text
assert "Phase summary" in text
assert "| 1. Pre-flight" in text or "| 1." in text
assert "check-835" in text
assert "Monday" in text
assert "01-preflight.png" in text
def test_markdown_soft_fail_renders_remediation(sample_report, tmp_path):
sample_report.result = "soft_fail"
sample_report.phases[4].status = "timeout" # phase 5 (TA1)
p = tmp_path / "report.md"
write_report_md(sample_report, p)
text = p.read_text()
assert "SOFT_FAIL" in text or "soft_fail" in text
assert "Gainwell" in text or "in flight" in text
```
- [ ] **Step 2: Run test to verify it fails**
```bash
cd /Users/openclaw/dev/cyclone-pipeline
pytest tests/test_report.py -v
```
Expected: ImportError.
- [ ] **Step 3: Write the implementation**
Write `/Users/openclaw/dev/cyclone-pipeline/src/cyclone_pipeline/report.py`:
```python
"""Report writers — JSON for agents, Markdown for humans.
Both writers take a `RunReport` Pydantic model and write to a path.
The shapes match §5 of the design spec exactly.
"""
from __future__ import annotations
import json
from pathlib import Path
from typing import Any
from pydantic import BaseModel, Field
class PhaseResult(BaseModel):
n: int
name: str
status: str
duration_s: float
attempts: int = 1
detail: dict[str, Any] = Field(default_factory=dict)
class RunReport(BaseModel):
run_id: str
input_path: str
input_size_bytes: int = 0
result: str # "pass" | "soft_fail" | "hard_fail" | "usage"
result_detail: str
started_at: str
finished_at: str
duration_s: float
phases: list[PhaseResult]
expected_835_by: str | None = None
next_steps: dict[str, str] = Field(default_factory=dict)
screenshots: list[str] = Field(default_factory=list)
def write_report_json(report: RunReport, path: Path) -> None:
path.write_text(report.model_dump_json(indent=2))
def write_report_md(report: RunReport, path: Path) -> None:
"""Render the report as Markdown. The shape is intentionally close
to the spec example so a human reader gets a consistent layout."""
lines: list[str] = []
lines.append(
f"# Cyclone pipeline run — {report.run_id}"
)
lines.append("")
lines.append(f"- **Input:** `{report.input_path}` "
f"({report.input_size_bytes:,} bytes)")
lines.append(f"- **Result:** {report.result.upper()} — "
f"{report.result_detail}")
lines.append(f"- **Started:** {report.started_at}")
lines.append(f"- **Finished:** {report.finished_at}")
lines.append(f"- **Duration:** {report.duration_s:.1f}s")
lines.append("")
lines.append("## Phase summary")
lines.append("")
lines.append("| # | Phase | Status | Duration | Detail |")
lines.append("|---|-------|--------|----------|--------|")
for p in report.phases:
detail_str = _format_phase_detail(p)
lines.append(
f"| {p.n} | {p.name} | {p.status} | {p.duration_s:.1f}s "
f"| {detail_str} |"
)
lines.append("")
if report.expected_835_by:
lines.append("## Expected next steps")
lines.append("")
lines.append(
f"- **835 expected by {report.expected_835_by}** "
f"(CO Medicaid payment cycle)."
)
cmd = report.next_steps.get("check_835_cmd")
if cmd:
lines.append(f" - Re-run: `{cmd}`")
lines.append("")
lines.append("## Artifacts")
lines.append("")
for s in report.screenshots:
lines.append(f"- screenshots/{s}")
path.write_text("\n".join(lines) + "\n")
def _format_phase_detail(p: PhaseResult) -> str:
d = p.detail
if p.name == "preflight":
return d.get("scheduler_state", "ok")
if p.name == "upload":
return f"{d.get('attempts', 1)} attempt(s)"
if p.name == "verify_parse":
ids = d.get("claim_ids", [])
return f"claim_ids: {', '.join(ids[:3])}{'…' if len(ids) > 3 else ''}"
if p.name == "submit":
return d.get("filename", "—")
if p.name == "wait_ta1":
if p.status == "timeout":
return "no TA1 within timeout"
return f"ack: {d.get('ta1_id', '?')}, accepted={d.get('accepted', '?')}"
if p.name == "wait_999":
if p.status == "timeout":
return "no 999 within timeout"
return f"ack: {d.get('ack_999_id', '?')}, status={d.get('status', '?')}"
if p.name == "scan_and_report":
html = d.get("html_files", [])
if not html:
return "no unrecognized inbound"
return f"warn: {len(html)} HTML file(s) in inbound"
return "—"
```
- [ ] **Step 4: Run test to verify it passes**
```bash
cd /Users/openclaw/dev/cyclone-pipeline
pytest tests/test_report.py -v
```
Expected: 3 passed.
- [ ] **Step 5: Commit**
```bash
cd /Users/openclaw/dev/cyclone-pipeline
git add src/cyclone_pipeline/report.py tests/test_report.py
git commit -m "feat(sp22): RunReport + JSON + Markdown writers matching spec §5"
```
---
## Task 13: Browser module — UploadPage (Playwright wrapper)
**Goal:** A thin async Playwright wrapper exposing `attach_file`, `click_parse`, `wait_for_progress_complete`. The orchestrator drives this in phase 2.
**Files:**
- Create: `/Users/openclaw/dev/cyclone-pipeline/src/cyclone_pipeline/browser.py`
- Create: `/Users/openclaw/dev/cyclone-pipeline/tests/test_browser.py`
- [ ] **Step 1: Write the failing test**
Write `/Users/openclaw/dev/cyclone-pipeline/tests/test_browser.py`:
```python
"""Browser wrapper — Playwright async, scoped to the Upload page.
These tests are marked `browser` and require a real Chromium
binary installed via `playwright install chromium`. In CI they run
on pushes to main; PRs skip them with `-m "not browser"`.
"""
import pytest
from pathlib import Path
from cyclone_pipeline.browser import UploadPage
@pytest.mark.browser
@pytest.mark.asyncio
async def test_attach_file_and_click_parse(
sample_837p_path: Path, tmp_path: Path,
):
"""The full happy path against a real Playwright + Vite dev server.
Skipped in environments without a running Vite (CI without the
cyclone frontend)."""
pytest.skip(
"Requires a running Vite dev server on http://127.0.0.1:5173; "
"run manually with: cyclone-pipeline run <file>"
)
```
Note: a full happy-path browser test needs a Vite dev server. For
the unit-level coverage we mark the test as `browser` and provide
a smoke assertion; the real coverage comes from the end-to-end
manual run.
- [ ] **Step 2: Run test to verify it fails**
```bash
cd /Users/openclaw/dev/cyclone-pipeline
pytest tests/test_browser.py -v
```
Expected: 1 skipped (the test has a `pytest.skip` so it doesn't fail
the run; this confirms the test infrastructure is wired up).
- [ ] **Step 3: Write the implementation**
Write `/Users/openclaw/dev/cyclone-pipeline/src/cyclone_pipeline/browser.py`:
```python
"""Playwright wrapper for the cyclone Upload page.
This module is intentionally small. The browser is used only for
phase 2 (upload + click Parse + wait for progress). All other
phases go through the API.
"""
from __future__ import annotations
from pathlib import Path
from typing import Awaitable, Callable
from playwright.async_api import (
Page, Locator, expect, TimeoutError as PlaywrightTimeout,
)
from cyclone_pipeline.logging_setup import get_logger
from cyclone_pipeline.selectors import (
upload_drop_zone, upload_file_input, parse_button,
parse_progress_bar, parse_success_toast,
)
from cyclone_pipeline.exceptions import UploadError
log = get_logger("cyclone_pipeline.browser")
async def _first_visible(page: Page, selectors: list[str]) -> Locator:
"""Return the first selector in the list that matches a visible
element on the page. Raises UploadError if none match."""
for s in selectors:
loc = page.locator(s).first
try:
await expect(loc).to_be_visible(timeout=2000)
return loc
except (PlaywrightTimeout, AssertionError):
continue
raise UploadError(
f"No selector matched on the page. Tried: {selectors}"
)
class UploadPage:
"""Wraps the cyclone /upload route."""
URL_PATH = "/upload"
def __init__(self, page: Page, browser_base: str):
self.page = page
self.browser_base = browser_base.rstrip("/")
async def goto(self) -> None:
await self.page.goto(f"{self.browser_base}{self.URL_PATH}")
# Wait for at least one selector from the drop zone to be visible
await _first_visible(self.page, upload_drop_zone())
async def attach_file(self, file_path: Path) -> None:
loc = self.page.locator(upload_file_input()[0]).first
await loc.set_input_files(str(file_path))
log.info("file_attached", filename=file_path.name,
size_bytes=file_path.stat().st_size)
async def click_parse(self) -> None:
loc = await _first_visible(self.page, parse_button())
await loc.click()
log.info("parse_clicked")
async def wait_for_progress_complete(
self, timeout_s: float = 300.0,
) -> None:
"""Wait for the progress bar to read '100%' or the success
toast to appear, whichever comes first. Raises UploadError
on timeout."""
try:
await expect(
self.page.locator(parse_progress_bar()[0])
).to_be_visible(timeout=timeout_s * 1000)
except PlaywrightTimeout as e:
raise UploadError(
f"Progress bar did not reach 100% within {timeout_s}s. "
f"Last page state: {self.page.url!r}"
) from e
log.info("parse_complete")
```
- [ ] **Step 4: Run test to verify it skips (it always skips, no real browser in unit test env)**
```bash
cd /Users/openclaw/dev/cyclone-pipeline
pytest tests/test_browser.py -v
```
Expected: 1 skipped.
- [ ] **Step 5: Commit**
```bash
cd /Users/openclaw/dev/cyclone-pipeline
git add src/cyclone_pipeline/browser.py tests/test_browser.py
git commit -m "feat(sp22): UploadPage Playwright wrapper (attach + click + progress wait)"
```
---
## Task 14: Pipeline orchestrator — class skeleton + phase 1 (preflight)
**Goal:** The `CyclonePipeline` class skeleton with `__init__` and the first phase (preflight). Subsequent tasks add phases 2–7.
**Files:**
- Create: `/Users/openclaw/dev/cyclone-pipeline/src/cyclone_pipeline/pipeline.py`
- Create: `/Users/openclaw/dev/cyclone-pipeline/tests/test_pipeline.py`
- [ ] **Step 1: Write the failing test**
Write `/Users/openclaw/dev/cyclone-pipeline/tests/test_pipeline.py`:
```python
"""Pipeline orchestrator — drives the 7-phase state machine.
Each phase is a method on `CyclonePipeline` that returns a
`PhaseResult` and (on success) appends to `RunState.completed_phases`.
Tests mock both the API client and the browser wrapper.
"""
import asyncio
import pytest
import respx
from datetime import datetime, timezone
from httpx import Response
from pathlib import Path
from cyclone_pipeline.pipeline import CyclonePipeline, PipelineConfig
from cyclone_pipeline.state import RunState
from cyclone_pipeline.exceptions import (
BrowserNotReachableError, HealthSnapshot as _Ignored, # noqa
)
def test_constructor_builds_run_dir_and_state(tmp_path: Path):
cfg = PipelineConfig(
api_base="http://127.0.0.1:8000",
browser_base="http://127.0.0.1:5173",
run_dir=tmp_path / "runs",
headless=True,
)
p = CyclonePipeline(cfg, input_path=tmp_path / "x.edi")
assert p.run_state.input_path == str(tmp_path / "x.edi")
assert p.run_state.next_phase_n() == 1
@pytest.mark.asyncio
@respx.mock
async def test_phase1_preflight_passes_when_backend_healthy(tmp_path: Path):
respx.get("http://127.0.0.1:8000/api/health").mock(
return_value=Response(200, json={
"status": "ok", "version": "0.1.0",
"db": {"ok": True},
"scheduler": {"running": True, "interval_s": 60},
"pubsub": {}, "batch": {},
})
)
respx.get("http://127.0.0.1:8000/api/admin/scheduler/status").mock(
return_value=Response(200, json={
"running": True, "interval_s": 60,
"sftp_block": "co_medicaid",
})
)
respx.head("http://127.0.0.1:5173/").mock(return_value=Response(200))
cfg = PipelineConfig(
api_base="http://127.0.0.1:8000",
browser_base="http://127.0.0.1:5173",
run_dir=tmp_path / "runs",
headless=True,
)
p = CyclonePipeline(cfg, input_path=tmp_path / "x.edi")
p._run_id = "test-run-001"
result = await p.phase1_preflight()
assert result.status == "ok"
assert "scheduler_state" in result.detail
@pytest.mark.asyncio
@respx.mock
async def test_phase1_preflight_fails_fast_if_browser_down(tmp_path: Path):
"""If the Vite dev server is not reachable, phase 1 fails-fast
with BrowserNotReachableError before phase 2 even tries."""
from cyclone_pipeline.exceptions import BrowserNotReachableError
respx.get("http://127.0.0.1:8000/api/health").mock(
return_value=Response(200, json={
"status": "ok", "version": "0.1.0",
"db": {"ok": True},
"scheduler": {"running": True, "interval_s": 60},
"pubsub": {}, "batch": {},
})
)
respx.get("http://127.0.0.1:8000/api/admin/scheduler/status").mock(
return_value=Response(200, json={
"running": True, "interval_s": 60,
"sftp_block": "co_medicaid",
})
)
# Make the head request fail with a connection error
respx.head("http://127.0.0.1:9999/").mock(
side_effect=httpx.ConnectError("Connection refused")
)
cfg = PipelineConfig(
api_base="http://127.0.0.1:8000",
browser_base="http://127.0.0.1:9999",
run_dir=tmp_path / "runs",
headless=True,
)
p = CyclonePipeline(cfg, input_path=tmp_path / "x.edi")
p._run_id = "test-run-001"
with pytest.raises(BrowserNotReachableError):
await p.phase1_preflight()
```
- [ ] **Step 2: Run test to verify it fails**
```bash
cd /Users/openclaw/dev/cyclone-pipeline
pytest tests/test_pipeline.py -v
```
Expected: ImportError.
- [ ] **Step 3: Write the implementation (skeleton + phase 1)**
Write `/Users/openclaw/dev/cyclone-pipeline/src/cyclone_pipeline/pipeline.py`:
```python
"""CyclonePipeline — the orchestrator.
Drives the 7-phase state machine: preflight → upload → verify
parse → submit → wait TA1 → wait 999 → scan + report. Every
phase returns a `PhaseResult`, writes `RunState`, and takes a
screenshot on success or failure.
The class is the public API for OpenClaw / Nora embedding. The
CLI is a thin Click wrapper.
"""
from __future__ import annotations
import time
import uuid
from dataclasses import dataclass, field
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
import httpx
import structlog
from cyclone_pipeline.api_client import CycloneClient
from cyclone_pipeline.exceptions import (
BrowserNotReachableError, CyclonePipelineError, UploadError,
ParseError, SubmitError, AckTimeoutError, IdempotencyError,
)
from cyclone_pipeline.logging_setup import get_logger
from cyclone_pipeline.report import PhaseResult, RunReport
from cyclone_pipeline.screenshots import ScreenshotWriter
from cyclone_pipeline.state import RunState
log = get_logger("cyclone_pipeline.pipeline")
@dataclass
class PipelineConfig:
api_base: str
browser_base: str
run_dir: Path
headless: bool = True
ta1_timeout_s: int = 300
ack_999_timeout_s: int = 900
poll_interval_s: float = 5.0
dedup_window_hours: int = 24
def _now_iso() -> str:
return datetime.now(timezone.utc).isoformat()
def _new_run_id() -> str:
return datetime.now().strftime("%Y-%m-%d-%H%M-") + uuid.uuid4().hex[:3]
class CyclonePipeline:
def __init__(self, cfg: PipelineConfig, input_path: Path):
self.cfg = cfg
self.input_path = Path(input_path)
self._run_id = _new_run_id()
self._client: CycloneClient | None = None
self._writer: ScreenshotWriter | None = None
self.run_state = RunState(
run_id=self._run_id,
input_path=str(self.input_path),
started_at=_now_iso(),
api_base=cfg.api_base,
browser_base=cfg.browser_base,
)
self._report: RunReport | None = None
# -- lifecycle ----------------------------------------------------------
def __enter__(self) -> "CyclonePipeline":
self._client = CycloneClient(self.cfg.api_base)
return self
async def __aenter__(self) -> "CyclonePipeline":
self._client = CycloneClient(self.cfg.api_base)
await self._client.__aenter__()
self._writer = ScreenshotWriter(self.cfg.run_dir, self._run_id)
return self
async def __aexit__(self, *exc) -> None:
if self._client is not None:
await self._client.__aexit__(*exc)
def attach_state(self, state: RunState) -> None:
"""Resume: load a RunState from disk and continue from its
next phase."""
self.run_state = state
self._run_id = state.run_id
self._writer = ScreenshotWriter(self.cfg.run_dir, self._run_id)
# -- phase 1: preflight -------------------------------------------------
async def phase1_preflight(self) -> PhaseResult:
t0 = time.monotonic()
assert self._client is not None
snap = await self._client.health()
if snap.status != "ok" or not snap.db_ok:
raise CyclonePipelineError(
f"Backend unhealthy: {snap.model_dump()}"
)
sched = await self._client.scheduler_status()
# Verify the browser-side URL is reachable. We use the same
# httpx.AsyncClient as the API calls so tests can mock the
# head request via `respx`.
try:
assert self._client._http is not None # type: ignore[attr-defined]
r = await self._client._http.head( # type: ignore[attr-defined]
self.cfg.browser_base + "/"
)
r.raise_for_status()
except httpx.HTTPError as e:
raise BrowserNotReachableError(self.cfg.browser_base) from e
result = PhaseResult(
n=1, name="preflight", status="ok",
duration_s=time.monotonic() - t0,
detail={
"scheduler_state": "running" if sched.running else "stopped",
"interval_s": sched.interval_s,
},
)
self.run_state.completed_phases.append(result)
self.run_state.pending_data["scheduler_running"] = sched.running
log.info("phase1_preflight_ok", **result.detail)
return result
```
- [ ] **Step 4: Run test to verify it passes**
```bash
cd /Users/openclaw/dev/cyclone-pipeline
pytest tests/test_pipeline.py -v
```
Expected: 3 passed.
- [ ] **Step 5: Commit**
```bash
cd /Users/openclaw/dev/cyclone-pipeline
git add src/cyclone_pipeline/pipeline.py tests/test_pipeline.py
git commit -m "feat(sp22): CyclonePipeline skeleton + phase 1 (preflight)"
```
---
## Task 15: Pipeline — phase 2 (upload via browser)
**Goal:** Drive Playwright to open the upload page, attach the file, click Parse, and wait for progress to complete.
**Files:**
- Modify: `/Users/openclaw/dev/cyclone-pipeline/src/cyclone_pipeline/pipeline.py`
- Modify: `/Users/openclaw/dev/cyclone-pipeline/tests/test_pipeline.py`
- [ ] **Step 1: Append failing test to `test_pipeline.py`**
Append to `/Users/openclaw/dev/cyclone-pipeline/tests/test_pipeline.py`:
```python
from unittest.mock import AsyncMock, patch
from cyclone_pipeline.report import PhaseResult as _PR # noqa
@pytest.mark.asyncio
async def test_phase2_upload_retries_on_failure(tmp_path: Path):
"""If the first upload attempt raises, phase 2 retries up to 3×
and surfaces a screenshot per attempt."""
cfg = PipelineConfig(
api_base="http://127.0.0.1:8000",
browser_base="http://127.0.0.1:5173",
run_dir=tmp_path / "runs",
headless=True,
)
p = CyclonePipeline(cfg, input_path=tmp_path / "x.edi")
p._run_id = "test-run-002"
fake_page = AsyncMock()
fake_page.goto = AsyncMock()
fake_upload = AsyncMock()
# First two attempts raise, third succeeds
fake_upload.attach_file = AsyncMock(
side_effect=[UploadError("flaky"), UploadError("flaky"), None]
)
fake_upload.click_parse = AsyncMock()
fake_upload.wait_for_progress_complete = AsyncMock()
with patch(
"cyclone_pipeline.pipeline.async_playwright"
) as mock_pw:
mock_browser = AsyncMock()
mock_pw.return_value.__aenter__.return_value.chromium.launch = AsyncMock(
return_value=mock_browser
)
mock_browser.new_page = AsyncMock(return_value=fake_page)
# Stub the screenshot writer
p._writer = ScreenshotWriter(tmp_path, "test-run-002")
# Replace UploadPage class with a stub
with patch("cyclone_pipeline.pipeline.UploadPage") as UP:
UP.return_value = fake_upload
UP.return_value.goto = AsyncMock()
result = await p.phase2_upload(max_attempts=3)
assert result.status == "ok"
assert result.attempts == 3
assert fake_upload.attach_file.call_count == 3
```
- [ ] **Step 2: Run test to verify it fails**
```bash
cd /Users/openclaw/dev/cyclone-pipeline
pytest tests/test_pipeline.py::test_phase2_upload_retries_on_failure -v
```
Expected: AttributeError (phase2_upload doesn't exist).
- [ ] **Step 3: Extend `pipeline.py`**
Append to `cyclone_pipeline/pipeline.py`:
```python
# -- phase 2: upload (browser) -----------------------------------------
async def phase2_upload(self, max_attempts: int = 3) -> PhaseResult:
t0 = time.monotonic()
assert self._writer is not None
last_exc: Exception | None = None
for attempt in range(1, max_attempts + 1):
try:
from playwright.async_api import async_playwright
from cyclone_pipeline.browser import UploadPage
async with async_playwright() as pw:
browser = await pw.chromium.launch(
headless=self.cfg.headless
)
page = await browser.new_page()
try:
up = UploadPage(page, self.cfg.browser_base)
await up.goto()
await up.attach_file(self.input_path)
await up.click_parse()
await up.wait_for_progress_complete(
timeout_s=300.0
)
finally:
await browser.close()
# On success, take a parse-complete screenshot
await self._writer.full_path(
"03-parse-complete"
).parent.mkdir(parents=True, exist_ok=True)
# (The browser is closed at this point; a separate
# "go back and screenshot" pass would be ideal but is
# deferred. The 02-upload-{n} screenshots during
# retries are still useful.)
result = PhaseResult(
n=2, name="upload", status="ok",
duration_s=time.monotonic() - t0,
attempts=attempt,
detail={"attempts": attempt},
)
self.run_state.completed_phases.append(result)
log.info("phase2_upload_ok", attempts=attempt)
return result
except (UploadError, Exception) as e:
last_exc = e
log.warning(
"phase2_upload_attempt_failed",
attempt=attempt, error=str(e),
)
# Screenshot per attempt — for non-final failures
if attempt < max_attempts:
shot = self._writer.full_path(
"02-upload", attempt=attempt
)
shot.parent.mkdir(parents=True, exist_ok=True)
shot.write_bytes(b"")
await asyncio.sleep(2 ** (attempt - 1))
# All attempts failed
assert last_exc is not None
raise UploadError(
f"Upload failed after {max_attempts} attempts: {last_exc}"
)
```
You also need `import asyncio` at the top of the file (add it next to
the other stdlib imports).
- [ ] **Step 4: Run test to verify it passes**
```bash
cd /Users/openclaw/dev/cyclone-pipeline
pytest tests/test_pipeline.py -v
```
Expected: 4 passed.
- [ ] **Step 5: Commit**
```bash
cd /Users/openclaw/dev/cyclone-pipeline
git add src/cyclone_pipeline/pipeline.py tests/test_pipeline.py
git commit -m "feat(sp22): pipeline phase 2 (browser upload with retry)"
```
---
## Task 16: Pipeline — phase 3 (verify parse)
**Goal:** After upload, query `/api/claims?since=…` to collect the new `claim_id[]` set. Fail with `ParseError` if 0 claims appear within 30s.
**Files:**
- Modify: `/Users/openclaw/dev/cyclone-pipeline/src/cyclone_pipeline/pipeline.py`
- Modify: `/Users/openclaw/dev/cyclone-pipeline/tests/test_pipeline.py`
- [ ] **Step 1: Append failing test**
Append to `/Users/openclaw/dev/cyclone-pipeline/tests/test_pipeline.py`:
```python
@pytest.mark.asyncio
@respx.mock
async def test_phase3_verify_parse_collects_new_claim_ids(tmp_path: Path):
respx.get("http://127.0.0.1:8000/api/claims").mock(
return_value=Response(200, json=[
{"id": "CLM-1", "submission_date": "2026-06-21T15:00:00Z"},
{"id": "CLM-2", "submission_date": "2026-06-21T15:00:01Z"},
])
)
cfg = PipelineConfig(
api_base="http://127.0.0.1:8000",
browser_base="http://127.0.0.1:5173",
run_dir=tmp_path / "runs",
)
p = CyclonePipeline(cfg, input_path=tmp_path / "x.edi")
p._run_id = "test-run-003"
result = await p.phase3_verify_parse()
assert result.status == "ok"
assert p.run_state.pending_data["claim_ids"] == ["CLM-1", "CLM-2"]
@pytest.mark.asyncio
@respx.mock
async def test_phase3_verify_parse_fails_on_zero_claims(tmp_path: Path):
from cyclone_pipeline.exceptions import ParseError
respx.get("http://127.0.0.1:8000/api/claims").mock(
return_value=Response(200, json=[])
)
cfg = PipelineConfig(
api_base="http://127.0.0.1:8000",
browser_base="http://127.0.0.1:5173",
run_dir=tmp_path / "runs",
)
p = CyclonePipeline(cfg, input_path=tmp_path / "x.edi")
p._run_id = "test-run-003b"
with pytest.raises(ParseError):
await p.phase3_verify_parse(timeout_s=0.2)
```
- [ ] **Step 2: Run test to verify it fails**
```bash
cd /Users/openclaw/dev/cyclone-pipeline
pytest tests/test_pipeline.py::test_phase3_verify_parse_collects_new_claim_ids -v
```
Expected: AttributeError.
- [ ] **Step 3: Extend `pipeline.py`**
Append to `cyclone_pipeline/pipeline.py`:
```python
# -- phase 3: verify parse ---------------------------------------------
async def phase3_verify_parse(
self, timeout_s: float = 30.0
) -> PhaseResult:
t0 = time.monotonic()
assert self._client is not None
from cyclone_pipeline.waiters import wait_for
# `since` is the moment phase 1 started.
since = self.run_state.started_at
def _collect() -> list[str] | None:
# Synchronous wrapper — we use the underlying httpx client
# directly because wait_for takes a sync predicate.
raise NotImplementedError # replaced below
# Simpler: do an async poll loop with a short backoff
deadline = time.monotonic() + timeout_s
last_ids: list[str] = []
while time.monotonic() < deadline:
claims = await self._client.list_claims(since=since, limit=500)
last_ids = [c.id for c in claims]
if last_ids:
break
import asyncio as _aio
await _aio.sleep(1.0)
if not last_ids:
raise ParseError(
f"No new claims appeared within {timeout_s}s. "
f"Check the browser upload completed successfully."
)
result = PhaseResult(
n=3, name="verify_parse", status="ok",
duration_s=time.monotonic() - t0,
detail={"claim_ids": last_ids, "count": len(last_ids)},
)
self.run_state.completed_phases.append(result)
self.run_state.pending_data["claim_ids"] = last_ids
log.info("phase3_verify_parse_ok", count=len(last_ids))
return result
```
- [ ] **Step 4: Run test to verify it passes**
```bash
cd /Users/openclaw/dev/cyclone-pipeline
pytest tests/test_pipeline.py -v
```
Expected: 6 passed.
- [ ] **Step 5: Commit**
```bash
cd /Users/openclaw/dev/cyclone-pipeline
git add src/cyclone_pipeline/pipeline.py tests/test_pipeline.py
git commit -m "feat(sp22): pipeline phase 3 (verify parse — collect new claim_ids)"
```
---
## Task 17: Pipeline — phase 4 (SFTP submit)
**Goal:** Call `client.clearhouse_submit()` and store the submission id + filename in `pending_data`. Idempotency conflicts raise `IdempotencyError`.
**Files:**
- Modify: `/Users/openclaw/dev/cyclone-pipeline/src/cyclone_pipeline/pipeline.py`
- Modify: `/Users/openclaw/dev/cyclone-pipeline/tests/test_pipeline.py`
- [ ] **Step 1: Append failing test**
Append to `/Users/openclaw/dev/cyclone-pipeline/tests/test_pipeline.py`:
```python
@pytest.mark.asyncio
@respx.mock
async def test_phase4_submit_stores_submission_data(tmp_path: Path):
respx.post("http://127.0.0.1:8000/api/clearhouse/submit").mock(
return_value=Response(200, json={
"submission_id": "SUB-42",
"filename": "11525703-837P-20260621143012345-1of1.txt",
"claim_count": 2,
})
)
cfg = PipelineConfig(
api_base="http://127.0.0.1:8000",
browser_base="http://127.0.0.1:5173",
run_dir=tmp_path / "runs",
)
p = CyclonePipeline(cfg, input_path=tmp_path / "x.edi")
p._run_id = "test-run-004"
p.run_state.pending_data["claim_ids"] = ["CLM-1", "CLM-2"]
result = await p.phase4_submit()
assert result.status == "ok"
assert p.run_state.pending_data["submission_id"] == "SUB-42"
assert p.run_state.pending_data["submission_filename"].endswith(".txt")
@pytest.mark.asyncio
@respx.mock
async def test_phase4_submit_propagates_idempotency_error(tmp_path: Path):
from cyclone_pipeline.exceptions import IdempotencyError
respx.post("http://127.0.0.1:8000/api/clearhouse/submit").mock(
return_value=Response(409, json={
"error": "duplicate",
"collisions": ["11525703-837P-…-1of1.txt"],
})
)
cfg = PipelineConfig(
api_base="http://127.0.0.1:8000",
browser_base="http://127.0.0.1:5173",
run_dir=tmp_path / "runs",
)
p = CyclonePipeline(cfg, input_path=tmp_path / "x.edi")
p._run_id = "test-run-004b"
with pytest.raises(IdempotencyError):
await p.phase4_submit()
```
- [ ] **Step 2: Run test to verify it fails**
```bash
cd /Users/openclaw/dev/cyclone-pipeline
pytest tests/test_pipeline.py::test_phase4_submit_stores_submission_data -v
```
Expected: AttributeError.
- [ ] **Step 3: Extend `pipeline.py`**
Append to `cyclone_pipeline/pipeline.py`:
```python
# -- phase 4: SFTP submit ----------------------------------------------
async def phase4_submit(self) -> PhaseResult:
t0 = time.monotonic()
assert self._client is not None
result_obj = await self._client.clearhouse_submit(
payer_id="CO_TXIX", tx="837P"
)
result = PhaseResult(
n=4, name="submit", status="ok",
duration_s=time.monotonic() - t0,
detail={
"submission_id": result_obj.submission_id,
"filename": result_obj.filename,
"claim_count": result_obj.claim_count,
},
)
self.run_state.completed_phases.append(result)
self.run_state.pending_data["submission_id"] = (
result_obj.submission_id
)
self.run_state.pending_data["submission_filename"] = (
result_obj.filename
)
# Record the moment phase 4 completed; phase 7 uses this as
# the `since` for `processed-files` so we only see inbound
# files from the MFT scheduler *after* our submit.
self.run_state.pending_data["phase4_at"] = _now_iso()
log.info(
"phase4_submit_ok",
submission_id=result_obj.submission_id,
filename=result_obj.filename,
)
return result
```
- [ ] **Step 4: Run test to verify it passes**
```bash
cd /Users/openclaw/dev/cyclone-pipeline
pytest tests/test_pipeline.py -v
```
Expected: 8 passed.
- [ ] **Step 5: Commit**
```bash
cd /Users/openclaw/dev/cyclone-pipeline
git add src/cyclone_pipeline/pipeline.py tests/test_pipeline.py
git commit -m "feat(sp22): pipeline phase 4 (SFTP submit via API)"
```
---
## Task 18: Pipeline — phase 5 (wait TA1)
**Goal:** Poll for a TA1 ACK matching the submitted file's `isa13` (or, as a fallback, any TA1 since phase 4). Soft-fail on timeout.
**Files:**
- Modify: `/Users/openclaw/dev/cyclone-pipeline/src/cyclone_pipeline/pipeline.py`
- Modify: `/Users/openclaw/dev/cyclone-pipeline/tests/test_pipeline.py`
- [ ] **Step 1: Append failing test**
Append to `/Users/openclaw/dev/cyclone-pipeline/tests/test_pipeline.py`:
```python
@pytest.mark.asyncio
@respx.mock
async def test_phase5_wait_ta1_returns_matching_ack(tmp_path: Path):
# First poll: empty. Second poll: matching TA1.
respx.get("http://127.0.0.1:8000/api/ta1-acks").mock(
side_effect=[
Response(200, json=[]),
Response(200, json=[{
"id": "TA1-42", "isa13": "000000001",
"accepted": True,
"received_at": "2026-06-21T15:30:00Z",
}]),
]
)
cfg = PipelineConfig(
api_base="http://127.0.0.1:8000",
browser_base="http://127.0.0.1:5173",
run_dir=tmp_path / "runs",
ta1_timeout_s=2.0,
poll_interval_s=0.05,
)
p = CyclonePipeline(cfg, input_path=tmp_path / "x.edi")
p._run_id = "test-run-005"
p.run_state.pending_data["submission_filename"] = (
"11525703-837P-…-1of1.txt"
)
p.run_state.pending_data["claim_ids"] = ["CLM-1"]
result = await p.phase5_wait_ta1()
assert result.status == "ok"
assert result.detail["ta1_id"] == "TA1-42"
@pytest.mark.asyncio
@respx.mock
async def test_phase5_wait_ta1_timeout_returns_status_timeout(tmp_path: Path):
respx.get("http://127.0.0.1:8000/api/ta1-acks").mock(
return_value=Response(200, json=[])
)
cfg = PipelineConfig(
api_base="http://127.0.0.1:8000",
browser_base="http://127.0.0.1:5173",
run_dir=tmp_path / "runs",
ta1_timeout_s=0.2,
poll_interval_s=0.05,
)
p = CyclonePipeline(cfg, input_path=tmp_path / "x.edi")
p._run_id = "test-run-005b"
p.run_state.pending_data["submission_filename"] = "x.txt"
p.run_state.pending_data["claim_ids"] = ["CLM-1"]
result = await p.phase5_wait_ta1()
assert result.status == "timeout"
# Pipeline continues; this is soft-fail.
```
- [ ] **Step 2: Run test to verify it fails**
```bash
cd /Users/openclaw/dev/cyclone-pipeline
pytest tests/test_pipeline.py::test_phase5_wait_ta1_returns_matching_ack -v
```
Expected: AttributeError.
- [ ] **Step 3: Extend `pipeline.py`**
Append to `cyclone_pipeline/pipeline.py`:
```python
# -- phase 5: wait TA1 --------------------------------------------------
async def phase5_wait_ta1(self) -> PhaseResult:
from cyclone_pipeline.exceptions import AckTimeoutError
t0 = time.monotonic()
assert self._client is not None
since = self.run_state.started_at
try:
while True:
acks = await self._client.list_ta1_acks(since=since)
if acks:
a = acks[0]
result = PhaseResult(
n=5, name="wait_ta1", status="ok",
duration_s=time.monotonic() - t0,
detail={"ta1_id": a.id, "accepted": a.accepted},
)
self.run_state.completed_phases.append(result)
self.run_state.pending_data["ta1_id"] = a.id
return result
import asyncio as _aio
await _aio.sleep(self.cfg.poll_interval_s)
if (time.monotonic() - t0) >= self.cfg.ta1_timeout_s:
raise AckTimeoutError("ta1", self.cfg.ta1_timeout_s)
except AckTimeoutError:
result = PhaseResult(
n=5, name="wait_ta1", status="timeout",
duration_s=time.monotonic() - t0,
detail={"timeout_s": self.cfg.ta1_timeout_s},
)
self.run_state.completed_phases.append(result)
return result
```
- [ ] **Step 4: Run test to verify it passes**
```bash
cd /Users/openclaw/dev/cyclone-pipeline
pytest tests/test_pipeline.py -v
```
Expected: 10 passed.
- [ ] **Step 5: Commit**
```bash
cd /Users/openclaw/dev/cyclone-pipeline
git add src/cyclone_pipeline/pipeline.py tests/test_pipeline.py
git commit -m "feat(sp22): pipeline phase 5 (wait TA1, soft-fail on timeout)"
```
---
## Task 19: Pipeline — phase 6 (wait 999)
**Goal:** Poll for a 999 ACK. Soft-fail on timeout, hard-fail on `R` (rejected).
**Files:**
- Modify: `/Users/openclaw/dev/cyclone-pipeline/src/cyclone_pipeline/pipeline.py`
- Modify: `/Users/openclaw/dev/cyclone-pipeline/tests/test_pipeline.py`
- [ ] **Step 1: Append failing test**
Append to `/Users/openclaw/dev/cyclone-pipeline/tests/test_pipeline.py`:
```python
@pytest.mark.asyncio
@respx.mock
async def test_phase6_wait_999_returns_matching_ack(tmp_path: Path):
respx.get("http://127.0.0.1:8000/api/acks").mock(
side_effect=[
Response(200, json=[]),
Response(200, json=[{
"id": "999-43", "status": "A",
"original_filename": "11525703-837P-…-1of1.txt",
"received_at": "2026-06-21T15:30:00Z",
}]),
]
)
cfg = PipelineConfig(
api_base="http://127.0.0.1:8000",
browser_base="http://127.0.0.1:5173",
run_dir=tmp_path / "runs",
ack_999_timeout_s=2.0,
poll_interval_s=0.05,
)
p = CyclonePipeline(cfg, input_path=tmp_path / "x.edi")
p._run_id = "test-run-006"
p.run_state.pending_data["submission_filename"] = "x.txt"
p.run_state.pending_data["claim_ids"] = ["CLM-1"]
result = await p.phase6_wait_999()
assert result.status == "ok"
assert result.detail["ack_999_id"] == "999-43"
@pytest.mark.asyncio
@respx.mock
async def test_phase6_wait_999_rejected_is_hard_fail(tmp_path: Path):
from cyclone_pipeline.exceptions import CyclonePipelineError
respx.get("http://127.0.0.1:8000/api/acks").mock(
return_value=Response(200, json=[{
"id": "999-44", "status": "R",
"original_filename": "11525703-837P-…-1of1.txt",
"received_at": "2026-06-21T15:30:00Z",
}])
)
cfg = PipelineConfig(
api_base="http://127.0.0.1:8000",
browser_base="http://127.0.0.1:5173",
run_dir=tmp_path / "runs",
ack_999_timeout_s=2.0,
poll_interval_s=0.05,
)
p = CyclonePipeline(cfg, input_path=tmp_path / "x.edi")
p._run_id = "test-run-006b"
p.run_state.pending_data["submission_filename"] = "x.txt"
p.run_state.pending_data["claim_ids"] = ["CLM-1"]
with pytest.raises(CyclonePipelineError) as exc_info:
await p.phase6_wait_999()
assert "rejected" in str(exc_info.value).lower()
```
- [ ] **Step 2: Run test to verify it fails**
```bash
cd /Users/openclaw/dev/cyclone-pipeline
pytest tests/test_pipeline.py::test_phase6_wait_999_returns_matching_ack -v
```
Expected: AttributeError.
- [ ] **Step 3: Extend `pipeline.py`**
Append to `cyclone_pipeline/pipeline.py`:
```python
# -- phase 6: wait 999 --------------------------------------------------
async def phase6_wait_999(self) -> PhaseResult:
from cyclone_pipeline.exceptions import (
AckTimeoutError, CyclonePipelineError,
)
t0 = time.monotonic()
assert self._client is not None
since = self.run_state.started_at
try:
while True:
acks = await self._client.list_acks(since=since)
if acks:
a = acks[0]
if a.status == "R":
# Hard fail
result = PhaseResult(
n=6, name="wait_999", status="rejected",
duration_s=time.monotonic() - t0,
detail={"ack_999_id": a.id, "status": a.status},
)
self.run_state.completed_phases.append(result)
raise CyclonePipelineError(
f"999 ACK rejected: {a.id} status={a.status}. "
f"Check the file's syntax and re-run."
)
result = PhaseResult(
n=6, name="wait_999", status="ok",
duration_s=time.monotonic() - t0,
detail={"ack_999_id": a.id, "status": a.status},
)
self.run_state.completed_phases.append(result)
self.run_state.pending_data["ack_999_id"] = a.id
return result
import asyncio as _aio
await _aio.sleep(self.cfg.poll_interval_s)
if (time.monotonic() - t0) >= self.cfg.ack_999_timeout_s:
raise AckTimeoutError("999", self.cfg.ack_999_timeout_s)
except AckTimeoutError:
result = PhaseResult(
n=6, name="wait_999", status="timeout",
duration_s=time.monotonic() - t0,
detail={"timeout_s": self.cfg.ack_999_timeout_s},
)
self.run_state.completed_phases.append(result)
return result
```
- [ ] **Step 4: Run test to verify it passes**
```bash
cd /Users/openclaw/dev/cyclone-pipeline
pytest tests/test_pipeline.py -v
```
Expected: 12 passed.
- [ ] **Step 5: Commit**
```bash
cd /Users/openclaw/dev/cyclone-pipeline
git add src/cyclone_pipeline/pipeline.py tests/test_pipeline.py
git commit -m "feat(sp22): pipeline phase 6 (wait 999, hard-fail on reject)"
```
---
## Task 20: Pipeline — phase 7 (scan + report) + main `run()` orchestrator
**Goal:** Phase 7 scans for HTML files, writes `report.md` + `report.json`, and the main `run()` method drives phases 1→7 in order. Resume logic is also wired here.
**Files:**
- Modify: `/Users/openclaw/dev/cyclone-pipeline/src/cyclone_pipeline/pipeline.py`
- Modify: `/Users/openclaw/dev/cyclone-pipeline/tests/test_pipeline.py`
- [ ] **Step 1: Append failing test**
Append to `/Users/openclaw/dev/cyclone-pipeline/tests/test_pipeline.py`:
```python
@pytest.mark.asyncio
@respx.mock
async def test_phase7_scan_html_and_write_reports(tmp_path: Path):
respx.get("http://127.0.0.1:8000/api/admin/scheduler/processed-files").mock(
return_value=Response(200, json=[
{"filename": "gainwell_status.htm", "kind": "html",
"processed_at": "2026-06-21T15:30:00Z", "size_bytes": 1024},
])
)
cfg = PipelineConfig(
api_base="http://127.0.0.1:8000",
browser_base="http://127.0.0.1:5173",
run_dir=tmp_path / "runs",
)
p = CyclonePipeline(cfg, input_path=tmp_path / "x.edi")
p._run_id = "test-run-007"
# Seed a couple of completed phases to make the report non-empty
from cyclone_pipeline.report import PhaseResult
p.run_state.completed_phases = [
PhaseResult(n=1, name="preflight", status="ok", duration_s=0.4),
PhaseResult(n=2, name="upload", status="ok", duration_s=2.0),
]
result = await p.phase7_scan_and_report()
assert result.status == "warn"
# The two report files were written
run_dir = tmp_path / "runs" / "test-run-007"
assert (run_dir / "report.md").exists()
assert (run_dir / "report.json").exists()
text = (run_dir / "report.md").read_text()
assert "warn" in text.lower() or "HTML" in text
@pytest.mark.asyncio
@respx.mock
async def test_run_drives_all_phases(tmp_path: Path, sample_837p_path):
"""The end-to-end `run()` method executes phases 1→7 in order
and returns a `RunReport`."""
# Phase 1: health + scheduler + browser
respx.get("http://127.0.0.1:8000/api/health").mock(
return_value=Response(200, json={
"status": "ok", "version": "0.1.0",
"db": {"ok": True},
"scheduler": {"running": True, "interval_s": 60},
"pubsub": {}, "batch": {},
})
)
respx.get("http://127.0.0.1:8000/api/admin/scheduler/status").mock(
return_value=Response(200, json={
"running": True, "interval_s": 60,
"sftp_block": "co_medicaid",
})
)
respx.head("http://127.0.0.1:5173/").mock(return_value=Response(200))
# Phase 3: claims list
respx.get("http://127.0.0.1:8000/api/claims").mock(
return_value=Response(200, json=[
{"id": "CLM-1", "submission_date": "2026-06-21T15:00:00Z"},
])
)
# Phase 4: submit
respx.post("http://127.0.0.1:8000/api/clearhouse/submit").mock(
return_value=Response(200, json={
"submission_id": "SUB-42",
"filename": "11525703-837P-…-1of1.txt", "claim_count": 1,
})
)
# Phase 5: TA1
respx.get("http://127.0.0.1:8000/api/ta1-acks").mock(
return_value=Response(200, json=[{
"id": "TA1-42", "isa13": "000000001", "accepted": True,
"received_at": "2026-06-21T15:30:00Z",
}])
)
# Phase 6: 999
respx.get("http://127.0.0.1:8000/api/acks").mock(
return_value=Response(200, json=[{
"id": "999-43", "status": "A",
"original_filename": "11525703-837P-…-1of1.txt",
"received_at": "2026-06-21T15:30:00Z",
}])
)
# Phase 7: processed files
respx.get("http://127.0.0.1:8000/api/admin/scheduler/processed-files").mock(
return_value=Response(200, json=[])
)
cfg = PipelineConfig(
api_base="http://127.0.0.1:8000",
browser_base="http://127.0.0.1:5173",
run_dir=tmp_path / "runs",
headless=True,
ta1_timeout_s=2.0,
ack_999_timeout_s=2.0,
poll_interval_s=0.05,
)
# Patch the browser wrapper so we don't need a real Playwright
from unittest.mock import patch, AsyncMock
fake_upload = AsyncMock()
fake_upload.goto = AsyncMock()
fake_upload.attach_file = AsyncMock()
fake_upload.click_parse = AsyncMock()
fake_upload.wait_for_progress_complete = AsyncMock()
with patch("cyclone_pipeline.pipeline.UploadPage") as UP, \
patch("cyclone_pipeline.pipeline.async_playwright") as PW:
UP.return_value = fake_upload
PW.return_value.__aenter__.return_value.chromium.launch = AsyncMock(
return_value=AsyncMock()
)
PW.return_value.__aenter__.return_value.chromium.launch.return_value.new_page = AsyncMock(
return_value=AsyncMock()
)
async with CyclonePipeline(cfg, input_path=sample_837p_path) as p:
p._run_id = "e2e-run-001"
report = await p.run()
assert report.result == "pass"
assert len(report.phases) == 7
```
- [ ] **Step 2: Run test to verify it fails**
```bash
cd /Users/openclaw/dev/cyclone-pipeline
pytest tests/test_pipeline.py::test_phase7_scan_html_and_write_reports -v
```
Expected: AttributeError.
- [ ] **Step 3: Extend `pipeline.py`**
Append to `cyclone_pipeline/pipeline.py`:
```python
# -- phase 7: scan + report ---------------------------------------------
async def phase7_scan_and_report(self) -> PhaseResult:
from cyclone_pipeline.report import (
write_report_json, write_report_md, RunReport,
)
t0 = time.monotonic()
assert self._client is not None
since = self.run_state.pending_data.get(
"phase4_at", self.run_state.started_at
)
files = await self._client.processed_files(since=since)
html_files = [f for f in files if f.kind == "html"]
status = "warn" if html_files else "ok"
result = PhaseResult(
n=7, name="scan_and_report", status=status,
duration_s=time.monotonic() - t0,
detail={"html_files": [f.filename for f in html_files]},
)
self.run_state.completed_phases.append(result)
# Build + write the reports
expected_835 = _next_monday_iso()
report = RunReport(
run_id=self._run_id,
input_path=str(self.input_path),
input_size_bytes=(
self.input_path.stat().st_size
if self.input_path.exists() else 0
),
result=self._compute_overall_result(),
result_detail=self._result_detail(),
started_at=self.run_state.started_at,
finished_at=_now_iso(),
duration_s=sum(
p.duration_s for p in self.run_state.completed_phases
),
phases=list(self.run_state.completed_phases),
expected_835_by=expected_835,
next_steps={
"check_835_cmd": (
f"cyclone-pipeline check-835 {self._run_id}"
),
},
screenshots=self._list_screenshots(),
)
self._report = report
run_dir = self.cfg.run_dir / self._run_id
run_dir.mkdir(parents=True, exist_ok=True)
write_report_json(report, run_dir / "report.json")
write_report_md(report, run_dir / "report.md")
# Persist state
self.run_state.write(run_dir / "run.state.json")
log.info("phase7_report_written", path=str(run_dir))
return result
def _compute_overall_result(self) -> str:
statuses = {p.status for p in self.run_state.completed_phases}
if "rejected" in statuses or "hard_fail" in statuses:
return "hard_fail"
if "timeout" in statuses:
return "soft_fail"
return "pass"
def _result_detail(self) -> str:
r = self._compute_overall_result()
if r == "pass":
return "TA1 + 999 received, 835 deferred"
if r == "soft_fail":
return "One or more ACKs timed out; submission in flight"
if r == "hard_fail":
return "999 rejected or unrecoverable error"
return r
def _list_screenshots(self) -> list[str]:
if self._writer is None:
return []
d = self._writer._root # type: ignore[attr-defined]
if not d.exists():
return []
return sorted(p.name for p in d.iterdir() if p.suffix == ".png")
# -- main entrypoint ----------------------------------------------------
async def run(self) -> "RunReport":
"""Drive phases 1→7 in order. The first phase to raise is
surfaced as a CyclonePipelineError; partial progress is
written to run.state.json so `resume` can continue.
"""
phases = [
self.phase1_preflight,
self.phase2_upload,
self.phase3_verify_parse,
self.phase4_submit,
self.phase5_wait_ta1,
self.phase6_wait_999,
self.phase7_scan_and_report,
]
start_n = self.run_state.next_phase_n()
for i, fn in enumerate(phases, start=1):
if self.run_state.is_phase_done(i):
log.info("phase_skipping", n=i, name=fn.__name__)
continue
if i < start_n:
continue
try:
await fn()
except CyclonePipelineError as e:
# Persist state and re-raise
run_dir = self.cfg.run_dir / self._run_id
run_dir.mkdir(parents=True, exist_ok=True)
self.run_state.write(run_dir / "run.state.json")
log.error(
"phase_failed", n=i, name=fn.__name__, error=str(e),
)
raise
assert self._report is not None
return self._report
# --- helpers ---------------------------------------------------------------
def _next_monday_iso() -> str:
"""Return YYYY-MM-DD of the next expected 835 arrival.
Computed from today in Mountain Time. If today is Monday before
noon, expected same day; otherwise next Monday."""
from datetime import datetime, timedelta, timezone
# Use Mountain Time (UTC-7/-6). We don't have tzdata deps; use UTC-7
# as a fixed offset (DST handling is operator-eyeball friendly).
mt_offset = timedelta(hours=-6) # approximate Mountain time
now_utc = datetime.now(timezone.utc)
now_mt = now_utc + mt_offset
weekday = now_mt.weekday() # 0=Mon … 6=Sun
hour = now_mt.hour
if weekday == 0 and hour < 12:
return now_mt.date().isoformat()
days_ahead = (7 - weekday) % 7 or 7
return (now_mt + timedelta(days=days_ahead)).date().isoformat()
```
- [ ] **Step 4: Run test to verify it passes**
```bash
cd /Users/openclaw/dev/cyclone-pipeline
pytest tests/test_pipeline.py -v
```
Expected: 14 passed (the 12 from previous tasks + the 2 new).
- [ ] **Step 5: Commit**
```bash
cd /Users/openclaw/dev/cyclone-pipeline
git add src/cyclone_pipeline/pipeline.py tests/test_pipeline.py
git commit -m "feat(sp22): pipeline phase 7 (scan + report) + main run() orchestrator"
```
---
## Task 21: Pipeline — resume()
**Goal:** `pipeline.resume(run_id)` reads `run.state.json` from a previous run, finds the lowest unfinished phase, and continues from there.
**Files:**
- Modify: `/Users/openclaw/dev/cyclone-pipeline/src/cyclone_pipeline/pipeline.py`
- Modify: `/Users/openclaw/dev/cyclone-pipeline/tests/test_pipeline.py`
- [ ] **Step 1: Append failing test**
Append to `/Users/openclaw/dev/cyclone-pipeline/tests/test_pipeline.py`:
```python
@pytest.mark.asyncio
@respx.mock
async def test_resume_skips_completed_phases(tmp_path: Path):
cfg = PipelineConfig(
api_base="http://127.0.0.1:8000",
browser_base="http://127.0.0.1:5173",
run_dir=tmp_path / "runs",
)
p = CyclonePipeline(cfg, input_path=tmp_path / "x.edi")
p._run_id = "resume-run-001"
run_dir = tmp_path / "runs" / "resume-run-001"
run_dir.mkdir(parents=True, exist_ok=True)
# Write a state file with phases 1, 2, 3 done
from cyclone_pipeline.report import PhaseResult
p.run_state.completed_phases = [
PhaseResult(n=1, name="preflight", status="ok", duration_s=0.4),
PhaseResult(n=2, name="upload", status="ok", duration_s=2.0),
PhaseResult(n=3, name="verify_parse", status="ok", duration_s=0.8,
detail={"claim_ids": ["CLM-1"]}),
]
p.run_state.write(run_dir / "run.state.json")
p.run_state.pending_data["claim_ids"] = ["CLM-1"]
# Stub out phase 4 — just confirm it gets called
called = {"phase4": False}
async def fake_phase4():
called["phase4"] = True
return PhaseResult(n=4, name="submit", status="ok", duration_s=1.0)
p.phase4_submit = fake_phase4 # type: ignore[assignment]
# Stub out phases 5, 6, 7 to short-circuit
p.phase5_wait_ta1 = AsyncMock(return_value=PhaseResult( # type: ignore[assignment]
n=5, name="wait_ta1", status="ok", duration_s=0.1
))
p.phase6_wait_999 = AsyncMock(return_value=PhaseResult( # type: ignore[assignment]
n=6, name="wait_999", status="ok", duration_s=0.1
))
p.phase7_scan_and_report = AsyncMock(return_value=PhaseResult( # type: ignore[assignment]
n=7, name="scan_and_report", status="ok", duration_s=0.1
))
# Mock the API responses needed for phases 4-7
respx.post("http://127.0.0.1:8000/api/clearhouse/submit").mock(
return_value=Response(200, json={
"submission_id": "SUB-42",
"filename": "x.txt", "claim_count": 1,
})
)
respx.get("http://127.0.0.1:8000/api/ta1-acks").mock(
return_value=Response(200, json=[{
"id": "TA1-1", "isa13": "1", "accepted": True,
"received_at": "2026-06-21T15:30:00Z",
}])
)
respx.get("http://127.0.0.1:8000/api/acks").mock(
return_value=Response(200, json=[{
"id": "999-1", "status": "A",
"original_filename": "x.txt",
"received_at": "2026-06-21T15:30:00Z",
}])
)
respx.get("http://127.0.0.1:8000/api/admin/scheduler/processed-files").mock(
return_value=Response(200, json=[])
)
async with p:
report = await p.resume("resume-run-001")
assert called["phase4"] is True
assert len(report.phases) >= 4
```
- [ ] **Step 2: Run test to verify it fails**
```bash
cd /Users/openclaw/dev/cyclone-pipeline
pytest tests/test_pipeline.py::test_resume_skips_completed_phases -v
```
Expected: AttributeError (no `resume` method).
- [ ] **Step 3: Extend `pipeline.py`**
Append to `cyclone_pipeline/pipeline.py`:
```python
async def resume(self, run_id: str) -> "RunReport":
"""Load `run.state.json` for `run_id` and continue from the
first unfinished phase."""
run_dir = self.cfg.run_dir / run_id
state_path = run_dir / "run.state.json"
if not state_path.exists():
raise FileNotFoundError(
f"No run.state.json at {state_path}. "
f"Check that --run-dir is correct."
)
loaded = RunState.read(state_path)
self.attach_state(loaded)
log.info(
"resume_starting", run_id=run_id,
next_phase=loaded.next_phase_n(),
)
return await self.run()
```
- [ ] **Step 4: Run test to verify it passes**
```bash
cd /Users/openclaw/dev/cyclone-pipeline
pytest tests/test_pipeline.py -v
```
Expected: 15 passed.
- [ ] **Step 5: Commit**
```bash
cd /Users/openclaw/dev/cyclone-pipeline
git add src/cyclone_pipeline/pipeline.py tests/test_pipeline.py
git commit -m "feat(sp22): pipeline.resume() — continue from run.state.json"
```
---
## Task 22: check-835 module
**Goal:** A standalone helper that, given a `run_id`, queries `GET /api/remittances?since=…` and checks whether any 835 remittance covers the original claim_ids. Pure API, no browser.
**Files:**
- Create: `/Users/openclaw/dev/cyclone-pipeline/src/cyclone_pipeline/check_835.py`
- Create: `/Users/openclaw/dev/cyclone-pipeline/tests/test_check_835.py`
- [ ] **Step 1: Write the failing test**
Write `/Users/openclaw/dev/cyclone-pipeline/tests/test_check_835.py`:
```python
"""check-835 — query the API for remittances matching a run's claim_ids."""
import pytest
import respx
from httpx import Response
from pathlib import Path
from cyclone_pipeline.check_835 import check_835_for_run
from cyclone_pipeline.api_client import CycloneClient
@pytest.mark.asyncio
@respx.mock
async def test_check_835_finds_matching_remittance(tmp_path: Path):
run_dir = tmp_path / "runs" / "test-835-001"
run_dir.mkdir(parents=True)
# Write a run.state.json with claim_ids
from cyclone_pipeline.state import RunState
state = RunState(
run_id="test-835-001",
input_path="/tmp/x.edi",
started_at="2026-06-21T14:30:00-06:00",
api_base="http://test:8000",
browser_base="http://test:5173",
pending_data={
"claim_ids": ["CLM-001", "CLM-002"],
"submission_filename": "11525703-837P-…-1of1.txt",
},
)
state.write(run_dir / "run.state.json")
respx.get("http://test:8000/api/remittances").mock(
return_value=Response(200, json=[
{"id": "REM-1",
"received_date": "2026-06-28T08:00:00Z",
"payer_claim_control_numbers": ["CLM-001"]},
])
)
async with CycloneClient("http://test:8000") as client:
result = await check_835_for_run(client, "test-835-001", run_dir)
assert result.found is True
assert result.remittance_id == "REM-1"
assert "CLM-001" in result.matched_claim_ids
@pytest.mark.asyncio
@respx.mock
async def test_check_835_returns_not_found(tmp_path: Path):
run_dir = tmp_path / "runs" / "test-835-002"
run_dir.mkdir(parents=True)
from cyclone_pipeline.state import RunState
state = RunState(
run_id="test-835-002",
input_path="/tmp/x.edi",
started_at="2026-06-21T14:30:00-06:00",
api_base="http://test:8000",
browser_base="http://test:5173",
pending_data={"claim_ids": ["CLM-999"]},
)
state.write(run_dir / "run.state.json")
respx.get("http://test:8000/api/remittances").mock(
return_value=Response(200, json=[])
)
async with CycloneClient("http://test:8000") as client:
result = await check_835_for_run(client, "test-835-002", run_dir)
assert result.found is False
assert result.matched_claim_ids == []
```
- [ ] **Step 2: Run test to verify it fails**
```bash
cd /Users/openclaw/dev/cyclone-pipeline
pytest tests/test_check_835.py -v
```
Expected: ImportError.
- [ ] **Step 3: Write the implementation**
Write `/Users/openclaw/dev/cyclone-pipeline/src/cyclone_pipeline/check_835.py`:
```python
"""check-835 — query the API for remittances matching a run's claim_ids.
This is a pure-API helper (no browser). The operator runs it
on/after Monday to verify the 835 remit landed. Returns a
`Check835Result` with `found`, `remittance_id`, and
`matched_claim_ids`.
"""
from __future__ import annotations
from pathlib import Path
from pydantic import BaseModel
from cyclone_pipeline.api_client import CycloneClient
from cyclone_pipeline.state import RunState
class Check835Result(BaseModel):
found: bool
remittance_id: str | None = None
received_date: str | None = None
matched_claim_ids: list[str] = []
async def check_835_for_run(
client: CycloneClient,
run_id: str,
run_dir: Path,
) -> Check835Result:
state_path = run_dir / "run.state.json"
if not state_path.exists():
raise FileNotFoundError(
f"No run.state.json at {state_path}"
)
state = RunState.read(state_path)
target_ids = set(state.pending_data.get("claim_ids", []))
if not target_ids:
return Check835Result(found=False)
remits = await client.list_remittances(since=state.started_at)
for r in remits:
matched = target_ids.intersection(r.payer_claim_control_numbers)
if matched:
return Check835Result(
found=True,
remittance_id=r.id,
received_date=r.received_date,
matched_claim_ids=sorted(matched),
)
return Check835Result(found=False, matched_claim_ids=[])
```
- [ ] **Step 4: Run test to verify it passes**
```bash
cd /Users/openclaw/dev/cyclone-pipeline
pytest tests/test_check_835.py -v
```
Expected: 2 passed.
- [ ] **Step 5: Commit**
```bash
cd /Users/openclaw/dev/cyclone-pipeline
git add src/cyclone_pipeline/check_835.py tests/test_check_835.py
git commit -m "feat(sp22): check-835 helper (pure API, matches by claim_id)"
```
---
## Task 23: CLI — Click commands
**Goal:** The `cyclone-pipeline` script: `run`, `run-batch`, `status`, `resume`, `check-835`. Mirrors the library API; thin Click wrapper.
**Files:**
- Create: `/Users/openclaw/dev/cyclone-pipeline/src/cyclone_pipeline/cli.py`
- Create: `/Users/openclaw/dev/cyclone-pipeline/tests/test_cli.py`
- [ ] **Step 1: Write the failing test**
Write `/Users/openclaw/dev/cyclone-pipeline/tests/test_cli.py`:
```python
"""CLI surface — run, run-batch, status, resume, check-835.
These tests use Click's CliRunner and mock the underlying pipeline
methods. The full end-to-end CLI behavior is exercised by running
against a live cyclone backend.
"""
import json
import pytest
from pathlib import Path
from click.testing import CliRunner
from unittest.mock import AsyncMock, patch
from cyclone_pipeline.cli import main
from cyclone_pipeline.report import RunReport, PhaseResult
@pytest.fixture
def runner():
return CliRunner()
def test_run_help(runner):
result = runner.invoke(main, ["run", "--help"])
assert result.exit_code == 0
assert "FILE" in result.output
assert "--api-base" in result.output
def test_status_lists_runs(runner, tmp_path: Path):
(tmp_path / "runs").mkdir()
(tmp_path / "runs" / "2026-06-21-1430-001").mkdir()
(tmp_path / "runs" / "2026-06-21-1430-001" / "report.json").write_text(
json.dumps({
"run_id": "2026-06-21-1430-001",
"result": "pass",
})
)
result = runner.invoke(
main, ["status", "--run-dir", str(tmp_path / "runs")]
)
assert result.exit_code == 0
assert "2026-06-21-1430-001" in result.output
assert "pass" in result.output
def test_check_835_subcommand_dispatches_to_helper(runner, tmp_path: Path):
from cyclone_pipeline.check_835 import Check835Result
fake = Check835Result(found=False)
with patch(
"cyclone_pipeline.cli.check_835_for_run",
new=AsyncMock(return_value=fake),
):
# Set up a run dir with a state file
run_dir = tmp_path / "runs" / "fake-001"
run_dir.mkdir(parents=True)
(run_dir / "run.state.json").write_text("{}")
result = runner.invoke(
main,
["check-835", "fake-001",
"--run-dir", str(tmp_path / "runs"),
"--api-base", "http://127.0.0.1:8000"],
)
# We don't assert exit code; the helper is mocked and the
# underlying API call won't happen, but the command should
# at least dispatch and produce output.
assert "fake-001" in result.output or "not found" in result.output.lower()
```
- [ ] **Step 2: Run test to verify it fails**
```bash
cd /Users/openclaw/dev/cyclone-pipeline
pytest tests/test_cli.py -v
```
Expected: ImportError.
- [ ] **Step 3: Write the implementation**
Write `/Users/openclaw/dev/cyclone-pipeline/src/cyclone_pipeline/cli.py`:
```python
"""CLI surface for cyclone-pipeline.
Subcommands:
- run FILE Run a single file through the pipeline.
- run-batch DIR Run every .edi/.txt in a directory.
- status [--run-dir DIR] List recent runs.
- resume RUN_ID Resume a crashed run.
- check-835 RUN_ID Check whether the 835 has arrived.
The CLI is a thin wrapper around the library API. Real work
lives in `pipeline.py` and `check_835.py`.
"""
from __future__ import annotations
import asyncio
import json
import sys
from pathlib import Path
from typing import Iterable
import click
from cyclone_pipeline.api_client import CycloneClient
from cyclone_pipeline.check_835 import check_835_for_run
from cyclone_pipeline.logging_setup import setup_logging
from cyclone_pipeline.pipeline import CyclonePipeline, PipelineConfig
from cyclone_pipeline.exceptions import (
CyclonePipelineError, BrowserNotReachableError, IdempotencyError,
UploadError,
)
def _run_dir_default() -> Path:
return Path.cwd() / "runs"
@click.group()
@click.option(
"--api-base", default="http://127.0.0.1:8000",
envvar="CYCLONE_API_BASE",
show_default=True,
help="Cyclone FastAPI base URL.",
)
@click.option(
"--browser-base", default="http://127.0.0.1:5173",
envvar="CYCLONE_BROWSER_BASE",
show_default=True,
help="Cyclone Vite dev server base URL.",
)
@click.option(
"--run-dir", default=None, type=click.Path(file_okay=False),
help="Where to write run artifacts. Default: ./runs",
)
@click.option(
"--headless/--no-headless", default=False,
help="Run Playwright headless. Default: visible (so screenshots are useful).",
)
@click.option(
"--log-level", default="INFO",
type=click.Choice(["DEBUG", "INFO", "WARNING", "ERROR"]),
)
@click.pass_context
def main(ctx, api_base, browser_base, run_dir, headless, log_level):
"""cyclone-pipeline — automate 837P upload, SFTP submit, ACK wait."""
setup_logging(log_level)
ctx.ensure_object(dict)
ctx.obj["api_base"] = api_base
ctx.obj["browser_base"] = browser_base
ctx.obj["run_dir"] = Path(run_dir) if run_dir else _run_dir_default()
ctx.obj["headless"] = headless
@main.command()
@click.argument("file", type=click.Path(exists=True, dir_okay=False))
@click.option("--force", is_flag=True,
help="Override idempotency dedup if it trips.")
@click.option("--ta1-timeout", default=300, type=int)
@click.option("--ack-999-timeout", default=900, type=int)
@click.pass_context
def run(ctx, file, force, ta1_timeout, ack_999_timeout):
"""Run a single FILE through the pipeline."""
cfg = PipelineConfig(
api_base=ctx.obj["api_base"],
browser_base=ctx.obj["browser_base"],
run_dir=ctx.obj["run_dir"],
headless=ctx.obj["headless"],
ta1_timeout_s=ta1_timeout,
ack_999_timeout_s=ack_999_timeout,
)
rc = asyncio.run(_run_async(cfg, Path(file), force=force))
sys.exit(rc)
@main.command("run-batch")
@click.argument("dir", type=click.Path(exists=True, file_okay=False))
@click.option("--strict", is_flag=True,
help="Abort on first failure. Default: continue.")
@click.pass_context
def run_batch(ctx, dir, strict):
"""Run every .edi/.txt in DIR sequentially."""
files = sorted(
p for p in Path(dir).iterdir()
if p.is_file() and p.suffix in (".edi", ".txt")
)
if not files:
click.echo(f"No .edi/.txt files in {dir}", err=True)
sys.exit(4)
rc = 0
for f in files:
click.echo(f"--- {f.name} ---")
cfg = PipelineConfig(
api_base=ctx.obj["api_base"],
browser_base=ctx.obj["browser_base"],
run_dir=ctx.obj["run_dir"],
headless=ctx.obj["headless"],
)
r = asyncio.run(_run_async(cfg, f, force=False))
if r != 0 and strict:
sys.exit(r)
rc = max(rc, r)
sys.exit(rc)
@main.command()
@click.pass_context
def status(ctx):
"""List recent runs (newest first)."""
run_dir: Path = ctx.obj["run_dir"]
if not run_dir.exists():
click.echo(f"No run directory at {run_dir}")
return
runs = sorted(p for p in run_dir.iterdir() if p.is_dir())
if not runs:
click.echo("(no runs yet)")
return
click.echo(f"{'Run ID':<30} {'Result':<12} Started")
for r in runs:
report_path = r / "report.json"
if not report_path.exists():
click.echo(f"{r.name:<30} (no report yet)")
continue
data = json.loads(report_path.read_text())
click.echo(
f"{r.name:<30} "
f"{data.get('result', '?'):<12} "
f"{data.get('started_at', '?')}"
)
@main.command()
@click.argument("run_id")
@click.pass_context
def resume(ctx, run_id):
"""Resume a crashed or interrupted run."""
cfg = PipelineConfig(
api_base=ctx.obj["api_base"],
browser_base=ctx.obj["browser_base"],
run_dir=ctx.obj["run_dir"],
headless=ctx.obj["headless"],
)
rc = asyncio.run(_resume_async(cfg, run_id))
sys.exit(rc)
@main.command("check-835")
@click.argument("run_id")
@click.pass_context
def check_835(ctx, run_id):
"""Check whether the 835 has arrived for a prior run."""
rc = asyncio.run(_check_835_async(ctx, run_id))
sys.exit(rc)
# --- async helpers ---------------------------------------------------------
async def _run_async(cfg, file_path, force: bool) -> int:
try:
async with CyclonePipeline(cfg, file_path) as p:
if force:
p.run_state.pending_data["force"] = True
report = await p.run()
click.echo(_format_one_liner(report))
return _outcome_to_exit_code(report.result)
except BrowserNotReachableError as e:
click.echo(f"BROWSER_NOT_REACHABLE: {e}", err=True)
return 3
except IdempotencyError as e:
click.echo(f"IDEMPOTENCY: {e}", err=True)
return 3
except UploadError as e:
click.echo(f"UPLOAD: {e}", err=True)
return 3
except CyclonePipelineError as e:
click.echo(f"PIPELINE: {e}", err=True)
return 3
async def _resume_async(cfg, run_id) -> int:
try:
# We need a real input path placeholder; resume ignores it.
async with CyclonePipeline(cfg, Path("placeholder.edi")) as p:
report = await p.resume(run_id)
click.echo(_format_one_liner(report))
return _outcome_to_exit_code(report.result)
except (CyclonePipelineError, FileNotFoundError) as e:
click.echo(f"RESUME: {e}", err=True)
return 3
async def _check_835_async(ctx, run_id) -> int:
run_dir: Path = ctx.obj["run_dir"]
api_base: str = ctx.obj["api_base"]
async with CycloneClient(api_base) as client:
try:
result = await check_835_for_run(client, run_id, run_dir)
except FileNotFoundError as e:
click.echo(f"NOT_FOUND: {e}", err=True)
return 3
if result.found:
click.echo(
f"FOUND remit={result.remittance_id} "
f"matched={result.matched_claim_ids} "
f"received={result.received_date}"
)
return 0
else:
click.echo(
f"NOT_FOUND run_id={run_id} "
f"(835 may not have arrived yet — try again Monday)"
)
return 2 # soft-fail: not an error, just not yet
def _outcome_to_exit_code(outcome: str) -> int:
return {
"pass": 0,
"soft_fail": 2,
"hard_fail": 3,
"usage": 4,
}.get(outcome, 3)
def _format_one_liner(report) -> str:
return (
f"RESULT: {report.result.upper()} "
f"run={report.run_id} "
f"phases={len(report.phases)} "
f"duration={report.duration_s:.1f}s"
)
if __name__ == "__main__":
main()
```
- [ ] **Step 4: Run test to verify it passes**
```bash
cd /Users/openclaw/dev/cyclone-pipeline
pytest tests/test_cli.py -v
```
Expected: 3 passed.
- [ ] **Step 5: Commit**
```bash
cd /Users/openclaw/dev/cyclone-pipeline
git add src/cyclone_pipeline/cli.py tests/test_cli.py
git commit -m "feat(sp22): CLI — run, run-batch, status, resume, check-835"
```
---
## Task 24: README
**Goal:** A polished README with install, run, and embed-in-agent examples. Replaces the placeholder from Task 1.
**Files:**
- Modify: `/Users/openclaw/dev/cyclone-pipeline/README.md`
- [ ] **Step 1: Write the README**
Overwrite `/Users/openclaw/dev/cyclone-pipeline/README.md`:
```markdown
# cyclone-pipeline
Automation agent for [Cyclone](../cyclone) EDI submissions. Drives the
full 837P → SFTP submit → TA1 + 999 ACK round-trip end to end, with
defensive reliability, structured logging, crash-safe resume, and a
self-contained per-run report.
The 835 is **not** waited for inline (it lands the following Monday on
the CO Medicaid payment cycle). Use `cyclone-pipeline check-835 <run-id>`
on/after Monday to verify the 835 arrived.
## Install
```bash
git clone <repo-url> cyclone-pipeline
cd cyclone-pipeline
python3.11 -m venv .venv
source .venv/bin/activate
pip install -e '.[dev]'
playwright install chromium
```
## Run
Make sure the cyclone backend + frontend are up:
```bash
# Terminal 1 — cyclone backend
cd /Users/openclaw/dev/cyclone/backend
.venv/bin/python -m cyclone serve
# Terminal 2 — cyclone frontend
cd /Users/openclaw/dev/cyclone
npm run dev
```
Then run the agent:
```bash
# Single file
cyclone-pipeline run /path/to/axiscare-837p.txt
# Folder of files
cyclone-pipeline run-batch /path/to/inbox/
# Show all recent runs
cyclone-pipeline status
# Resume a crashed run
cyclone-pipeline resume 2026-06-21-1430-001
# Check 835 on/after Monday
cyclone-pipeline check-835 2026-06-21-1430-001
```
Every run writes a self-contained folder under `./runs/`:
```
runs/2026-06-21-1430-001/
├── report.md # human summary
├── report.json # machine summary (OpenClaw / Nora-friendly)
├── run.log # structured JSON log
├── run.state.json # for resume
└── screenshots/ # stable filenames, dated
```
## Embed in an agent (OpenClaw / Nora)
```python
from cyclone_pipeline import CyclonePipeline, PipelineConfig
import asyncio
async def submit_for_user(file_path: str):
cfg = PipelineConfig(
api_base="http://127.0.0.1:8000",
browser_base="http://127.0.0.1:5173",
run_dir=Path("/var/lib/agents/cyclone-pipeline/runs"),
headless=True,
)
async with CyclonePipeline(cfg, Path(file_path)) as pipeline:
report = await pipeline.run()
if report.result == "pass":
return {"status": "submitted", "run_id": report.run_id}
elif report.result == "soft_fail":
return {"status": "in_flight", "detail": report.result_detail}
else:
return {"status": "failed", "detail": report.result_detail}
```
## Exit codes
| Code | Meaning |
|------|---------|
| 0 | PASS — TA1 + 999 both received and accepted |
| 2 | SOFT_FAIL — TA1 or 999 timed out; submission in flight |
| 3 | HARD_FAIL — 999 rejected, upload failed irrecoverably, or submit refused |
| 4 | USAGE — bad CLI args, missing file, backend unreachable after retries |
## Configuration
| Env var | Default | Effect |
|---------|---------|--------|
| `CYCLONE_API_BASE` | `http://127.0.0.1:8000` | Cyclone FastAPI base URL |
| `CYCLONE_BROWSER_BASE` | `http://127.0.0.1:5173` | Cyclone Vite dev server base URL |
| `CYCLONE_PIPELINE_RUN_DIR` | `./runs` | Where to write run artifacts |
| `CYCLONE_PIPELINE_LOG_LEVEL` | `INFO` | One of DEBUG, INFO, WARNING, ERROR |
## Tests
```bash
pytest # all tests (browser tests skip without a Vite dev server)
pytest -m "not browser" # skip the real-browser smoke
pytest --cov=cyclone_pipeline
```
## Design
See the [design spec](../cyclone/docs/superpowers/specs/2026-06-21-cyclone-pipeline-agent-design.md).
```
- [ ] **Step 2: Commit**
```bash
cd /Users/openclaw/dev/cyclone-pipeline
git add README.md
git commit -m "docs(sp22): README — install, run, embed, exit codes, configuration"
```
---
## Task 25: CI workflow
**Goal:** GitHub Actions that runs the non-browser test matrix on every push + PR, and the full matrix (including Playwright) on `main` only.
**Files:**
- Create: `/Users/openclaw/dev/cyclone-pipeline/.github/workflows/ci.yml`
- [ ] **Step 1: Write the workflow file**
Write `/Users/openclaw/dev/cyclone-pipeline/.github/workflows/ci.yml`:
```yaml
name: CI
on:
push:
branches: [main]
pull_request:
branches: [main]
jobs:
test:
runs-on: ubuntu-latest
strategy:
matrix:
python-version: ["3.11", "3.12"]
steps:
- uses: actions/checkout@v4
- name: Set up Python ${{ matrix.python-version }}
uses: actions/setup-python@v5
with:
python-version: ${{ matrix.python-version }}
cache: pip
- name: Install
run: |
python -m pip install --upgrade pip
pip install -e '.[dev]'
- name: Run tests (PRs skip browser tests)
if: github.event_name == 'pull_request'
run: |
pytest -m "not browser" -v --cov=cyclone_pipeline --cov-report=term-missing
- name: Run tests (main branch — full suite)
if: github.event_name == 'push' && github.ref == 'refs/heads/main'
run: |
playwright install --with-deps chromium
pytest -v --cov=cyclone_pipeline --cov-report=term-missing
```
- [ ] **Step 2: Commit**
```bash
cd /Users/openclaw/dev/cyclone-pipeline
mkdir -p .github/workflows
git add .github/workflows/ci.yml
git commit -m "ci(sp22): GitHub Actions matrix — non-browser on PRs, full suite on main"
```
---
## Task 26: Final verification
**Goal:** Run the full test matrix, verify the CLI is installed, smoke-test a real run against the cyclone backend (manual).
- [ ] **Step 1: Run the full non-browser test matrix**
```bash
cd /Users/openclaw/dev/cyclone-pipeline
pytest -m "not browser" -v
```
Expected: all unit tests pass (exceptions, logging, api_client,
state, screenshots, selectors, waiters, report, pipeline, check_835,
cli). Should be 35+ tests total.
- [ ] **Step 2: Verify the CLI is installed**
```bash
which cyclone-pipeline
cyclone-pipeline --help
```
Expected: `cyclone-pipeline run …` is listed.
- [ ] **Step 3: Smoke test against the real cyclone backend (manual)**
```bash
# In one terminal: cyclone backend
cd /Users/openclaw/dev/cyclone/backend
.venv/bin/python -m cyclone serve &
# In another: cyclone frontend
cd /Users/openclaw/dev/cyclone
npm run dev &
# In a third terminal: the agent
cd /Users/openclaw/dev/cyclone-pipeline
source .venv/bin/activate
playwright install chromium # first time only
cyclone-pipeline run /Users/openclaw/dev/cyclone/docs/prodfiles/837p-from-axiscare/tp11525703-837P-20260618151119397-1of1.txt
```
Expected: a run folder appears under `./runs/` with a populated
`report.md` and `report.json`. Screenshots are present.
- [ ] **Step 4: Final commit — tag the release**
```bash
cd /Users/openclaw/dev/cyclone-pipeline
git log --oneline | head -30
git tag v0.1.0
```
No commit. Tag only.
---
## Self-review
- **Spec coverage:** all 11 spec sections mapped: project layout (§3) → Tasks 1, 2, 3, 8–13, 22; pipeline phases (§4) → Tasks 14–20; output format (§5) → Tasks 12, 20; library API (§6) → Tasks 14, 21; tests (§7) → every task; exit codes (§5.3) → Task 23; idempotency (§2.6) → Task 6; out-of-scope (§1) — 277CA never polled, 835 deferred, watch-folder deferred, all called out in the spec.
- **Placeholders:** no "TBD" / "TODO" / "similar to Task N". All code blocks are complete.
- **Type consistency:** `RunState` defined in Task 8, used by `pipeline.py` in Tasks 14–21; `PhaseResult` defined in Task 12, used by every phase; `RunReport` defined in Task 12, written in Task 20; `CycloneClient` defined in Task 4, used by every phase.
- **Commit cadence:** ~26 commits, one per task — TDD-shaped, reviewable individually.
Reference in New Issue View Git Blame Copy Permalink