Ports and adapters

application/ and infrastructure/ are fully implemented (milestone M2). This page documents the actual port shapes and their concrete adapters — for the design rationale behind choosing this pattern, see ../plans/architecture-and-roadmap.md and ADR 0001.

Why ports are Protocol, not ABC

application/ports.py defines every port as a typing.Protocol, not an abstract base class. Two reasons:

1. Structural typing keeps application/ free of any import from infrastructure/. A Protocol describes a shape; nothing has to inherit from it. import-linter's "infrastructure only reachable from bootstrap" contract would reject an ABC-based design the moment a concrete adapter needed to from claudeloop.application.ports import AgentGatewayBase — with Protocol, the adapter simply implements the right methods and duck-types into place.
2. Fakes for tests are trivial. FakeAgentGateway and friends in tests/application/fakes.py don't inherit from anything either; they just need the right method signatures, which mypy --strict verifies against the Protocol at type-check time — see test_runner.py for the fakes exercising the real AutonomousRunner, including the credit-top-up scenario end to end.

Ports (application/ports.py) and their adapters

Port	Responsibility	Real adapter	Fake (tests/application/fakes.py)
AgentGateway	Drive a live ClaudeSDKClient session — never query(), which raises after an error result and exits the process (see ADR 0002)	infrastructure/agent/gateway.py::ClaudeAgentGateway	FakeAgentGateway — replays a scripted list of TurnOutcomes
CapacityProbe	Run the cheap, throwaway turn used while waiting	infrastructure/agent/gateway.py::ClaudeCapacityProbe	FakeCapacityProbe — replays scripted TurnSignals
SessionCatalog	List/resolve sessions via the SDK's list_sessions()	infrastructure/agent/catalog.py::SdkSessionCatalog	a plain stub class in tests/application/test_usecases.py
ApiGateway	Invoke a generated REST command against anthropic.Anthropic()	infrastructure/api/gateway.py::AnthropicApiGateway	—
Clock	now()	infrastructure/clock.py::SystemClock	FakeClock — settable, doesn't advance on its own
Sleeper	await sleep_until(instant)	infrastructure/clock.py::AnyioSleeper	FakeSleeper — advances the paired FakeClock instantly, records what it was asked to wait for
ProgressReporter	Human-readable progress to the console	infrastructure/progress.py::ConsoleProgressReporter	FakeProgressReporter — records calls
AuditLog	Append-only JSONL of every recorded event	infrastructure/audit.py::JsonlAuditLog	FakeAuditLog — in-memory list
Notifier	Tell a human something needs attention	infrastructure/notify.py::StderrNotifier	not yet used by any test
RunStateStore	Persist run state so a killed run is resumable	infrastructure/state.py::FileRunStateStore	not yet wired into AutonomousRunner
SessionLock	Advisory file lock preventing two runners driving one session	infrastructure/lock.py::FileSessionLock	not yet wired into AutonomousRunner

RunStateStore and SessionLock have concrete adapters but AutonomousRunner doesn't call them yet — resuming a killed run and preventing concurrent drivers of one session are both still open work, not just documentation gaps.

There's also a narrower DoctorEnvironment protocol, local to application/usecases/doctor.py rather than ports.py, since doctor is deliberately cheap to run and shouldn't need a full AgentGateway. Its real adapter is infrastructure/doctor_env.py::RealDoctorEnvironment, which shells out to claude auth status and claude mcp list — the authoritative sources, rather than re-parsing config files whose format the docs warn changes between releases.

Why FakeClock/FakeSleeper, not unittest.mock or real sleeping

The waiting policy in domain/waiting.py is designed around instants, not durations, precisely so the application-layer tests never need to sleep for real. FakeClock holds a settable "now"; FakeSleeper.sleep_until(instant) jumps the paired clock straight to instant instead of blocking. This is what lets tests/application/test_runner.py exercise a full credit-top-up sequence — several failed probes, then a resumed run — in milliseconds of real wall-clock test time. See ../contributing/testing.md for why unittest.mock.patch("time.sleep") was rejected in favor of this (a real port + fake beats patching a stdlib call every test file has to remember to do consistently).

The composition root (bootstrap.py)

bootstrap.py is the only module permitted to import from every layer at once. Its actual wiring functions:

def build_runner(*, cwd: Path, config: RunnerConfig, ...) -> RunnerContext:
    """Wires ClaudeAgentGateway, ClaudeCapacityProbe, SystemClock, AnyioSleeper,
    JsonlAuditLog, and ConsoleProgressReporter into an AutonomousRunner."""

def build_session_catalog() -> SdkSessionCatalog: ...
def build_doctor_environment() -> DoctorEnvironment: ...

cli/ commands call into these functions and then drive the returned objects — they never construct an infrastructure.agent.gateway.ClaudeAgentGateway directly. This is the seam that makes it possible to swap an adapter (say, a different Notifier backend) without touching application/ or cli/ at all.