Application Core and CLI Adapter

software-engineering
full-stack
agents
application-architecture
cli
Build one application use case around a replaceable agent runner, then expose it through a thin command-line adapter.

A full-stack application needs one place where a user action becomes business behavior. If the CLI, HTTP route, and WebSocket handler each orchestrate the agent and database independently, they will eventually persist different events or handle failures differently. This chapter builds AutocodeApplication as the shared use-case layer, gives it a replaceable AgentRunner, and keeps the CLI as the first thin adapter.

The application service is the center

AutocodeApplication.stream_message owns the vertical use case. It validates the session, serializes concurrent runs for that session, journals and projects the user message, consumes transport-neutral runner events, persists them, and publishes each durable event to observers. It knows nothing about command-line parsing, HTTP status codes, WebSocket frames, or DOM elements.

This boundary gives every transport the same answer to a basic question: which durable events does one submitted message create?

from tempfile import TemporaryDirectory

from autocode.application import AutocodeApplication
from autocode.runner import DemoAgentRunner
from autocode.store.repository import SessionRepository

with TemporaryDirectory() as directory:
    application = AutocodeApplication(
        SessionRepository(f"{directory}/sessions.db"),
        DemoAgentRunner(),
    )
    session = application.create_session("first vertical slice")
    streamed = [
        event
        async for event in application.stream_message(
            session.session_id, "trace the request path"
        )
    ]
    restored = application.get_session(session.session_id)

assert [event.event_id for event in streamed] == [event.event_id for event in restored.events]
assert streamed[0].kind == "user_message"
assert streamed[-1].kind == "run_finished"
print("durable event kinds:", [event.kind for event in streamed])
durable event kinds: ['user_message', 'run_started', 'text_delta', 'text_delta', 'text_delta', 'text_delta', 'text_delta', 'text_delta', 'text_delta', 'assistant_message', 'run_finished']

The streamed and restored event identifiers are equal because publication follows persistence. A browser may disconnect after any event, but it can ask the repository for the suffix after its last cursor. The deterministic runner still crosses the complete application and database path; it only replaces the external model call.

A port separates orchestration from the model

AgentRunner is a structural interface with one operation: stream product-neutral updates for a session and message. DemoAgentRunner makes the course reproducible. HarnessAgentRunner translates the preceding course’s AgentEvent values into the product vocabulary. Neither runner writes SQLite or sends a WebSocket frame.

This is dependency inversion in practical form. The application owns the interface it needs; model-specific code implements that interface at the edge.

from autocode.runner import DemoAgentRunner

runner = DemoAgentRunner()
updates = [update async for update in runner.stream("session-1", "inspect the project")]

assert updates[0].kind == "run_started"
assert updates[-1].kind == "run_finished"
assert "".join(
    update.payload.get("content", "")
    for update in updates
    if update.kind == "text_delta"
).startswith("Demo agent received")
print("runner boundary:", [update.kind for update in updates])
runner boundary: ['run_started', 'text_delta', 'text_delta', 'text_delta', 'text_delta', 'text_delta', 'text_delta', 'text_delta', 'assistant_message', 'run_finished']

The runner sequence has no event identifiers or database cursors. Those are product concerns added by AutocodeApplication when it records each update. Keeping the distinction prevents a model SDK from becoming the durable source of truth and lets the same scripted runner drive CLI, HTTP, and WebSocket tests.

The CLI is an adapter, not the application

The installed autocode command parses arguments, selects a database path and runner mode, calls the application service, and formats the resulting durable events. JSON Lines remains useful for shell pipelines, while the interactive path prints a session identifier and final answer. doctor and config avoid model or network work so installation failures stay distinguishable from agent failures.

import json
import subprocess
import sys
from tempfile import TemporaryDirectory

from autocode.cli import VERSION
from autocode.store.repository import SessionRepository

with TemporaryDirectory() as directory:
    database = f"{directory}/sessions.db"
    completed = subprocess.run(
        [
            sys.executable,
            "-m",
            "autocode.cli",
            "run",
            "explain adapters",
            "--db",
            database,
            "--json",
        ],
        check=False,
        capture_output=True,
        text=True,
    )
    emitted = [json.loads(line) for line in completed.stdout.splitlines()]
    restored = SessionRepository(database).get(emitted[0]["session_id"])

assert VERSION == "0.2.0"
assert completed.returncode == 0
assert restored is not None
assert [event.event_id for event in restored.events] == [event["event_id"] for event in emitted]
print("CLI and repository agree on", len(emitted), "events")
CLI and repository agree on 10 events

The CLI output and repository contain the same event identifiers because the command did not reimplement the run. Chapter 02 adds REST and browser adapters around this service. Chapter 03 adds WebSocket commands and streamed publication without changing the use case established here.

Exercises

Design a second adapter without moving business behavior into it. Identify what the adapter validates and formats, what AutocodeApplication still owns, and how one test proves that CLI and web submissions persist the same event vocabulary.

[P01.1] Keep an HTTP route thin

Specify a POST /api/sessions/{id}/messages adapter around AutocodeApplication.stream_message. Name the request validation, response or streaming behavior, status codes, and the assertion that proves it did not create a second run implementation.

Back to top