Related Links

• Part 1 — Setup Project. Dependencies Direction. CRUD
• Part 3 — External Integration. Users Engagement
• Part 4 — Publication Lifecycle. Editorial Review
• Repository — main branch
• Pull Request #1 — iteration-02 → main (this part)

1. Feature Request

A writer editing an article wants AI help on demand. Three actions to start with: summarize the article, improve the grammar of a paragraph they are working on, suggest a title. The suggestion can be accepted or dismissed. If the AI is slow or unavailable, the editor keeps writing.

The model provider lives on the other side of the network — this is an external capability. The domain change stays small by design. The opposite case (internal, domain-heavy) is Part 4.

1.1 API Contract

I start any new feature from the contract between Client and Server. The contract forces UX, error semantics, and protocol shape to be decided before any code is written. Once it is in place, every other layer has something concrete to satisfy.

• action is a query parameter
• values are summarize, improve_grammar, suggest_title
• for summarize and suggest_title, the body is empty — the handler reads the article body from storage
• for improve_grammar, input is required and carries the text being edited

One endpoint, multiple actions, dispatched in the handler.

A naming choice I would not repeat. /articles/{slug}/actions reads fine in isolation but ages poorly the moment the article gains other kinds of "actions" — submit, retract, review, approve, reject, publish. By Part 4 the lifecycle adds seven of them, and /actions collides with every one. Verb-named endpoints (/submit, /approve, …) survive that growth; a generic /actions dispatcher does not. I keep this endpoint as-is so the Part 2 → Part 4 contrast is visible — Part 4 revisits the decision and goes the other way.

1.2 UI Behaviour

2. Implementation & Examples

I walk outside-in. The contract is set, so the HTTP layer goes first: actions becomes a single dispatcher endpoint. Then the application layer composes three use cases against an abstraction the domain will own. Then the abstraction itself, the new domain errors, and the infrastructure binding that fulfils the abstraction. By the time the binding is written, the domain has already told two layers what it wants.

2.1 HTTP Contracts

# src/http/contracts/assistance.py

from pydantic import Field

from src.http._base import PublicModel


class ActionRequest(PublicModel):
    # `input` is meaningful only for actions that take user-supplied text
    # (currently: improve_grammar). Empty strings are rejected at the
    # contract level via `min_length=1`.
    input: str | None = Field(default=None, min_length=1)


class ActionPublic(PublicModel):
    suggestion: str
    confidence: float | None = None

confidence is in the response shape from day one so the contract can grow without a breaking change. The current adapter does not compute it yet.

2.2 HTTP Resources

The actions route is the dispatcher. FastAPI validates ?action against the AssistanceKind enum for free, so an unknown value is already a 422 before any code runs.

# src/http/resources/articles.py  (actions route shown; CRUD routes omitted)

from fastapi import APIRouter, HTTPException, status

from src.application import articles
from src.domain.cognitive_layer import AssistanceKind
from src.http.contracts.assistance import ActionPublic, ActionRequest
from src.infrastructure.database.repositories.articles import (
    SqlAlchemyArticlesRepository,
)
from src.infrastructure.pydantic_bindings import PydanticAICognitiveLayer


@router.post("/{slug}/actions", status_code=status.HTTP_200_OK)
async def article_actions(
    slug: str,
    action: AssistanceKind,
    body: ActionRequest | None = None,
) -> ActionPublic:
    repository = SqlAlchemyArticlesRepository()
    layer = PydanticAICognitiveLayer()

    match action:
        case AssistanceKind.SUMMARIZE:
            response = await articles.summarize_article(repository, layer, slug)

        case AssistanceKind.IMPROVE_GRAMMAR:
            if body is None or body.input is None:
                raise HTTPException(
                    status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
                    detail="`input` is required for improve_grammar",
                )
            response = await articles.improve_grammar(
                repository, layer, slug, body.input,
            )

        case AssistanceKind.SUGGEST_TITLE:
            response = await articles.suggest_title(repository, layer, slug)

    return ActionPublic.model_validate(response)

• FastAPI validates ?action against AssistanceKind — unknown values return 422 without reaching the handler
• the request body is optional (only improve_grammar consumes it)
• extra="forbid" on PublicModel rejects unknown fields, min_length=1 rejects empty strings
• no @transactional — editorial actions read the article and call an external model, they do not write to the database

2.3 Application Layer

The three actions are separate functions because they are separate operations on the article. The HTTP dispatcher hands the call to the right one; the application layer is where each operation actually lives.

# src/application/articles.py  (editorial actions shown; CRUD omitted)

from src.domain.articles import BookshelfRepository
from src.domain.articles.policies import find_stop_word
from src.domain.cognitive_layer import (
    AssistanceKind,
    CognitiveLayer,
    CognitiveRequest,
    CognitiveResponse,
)
from src.domain.errors import CognitiveOutputRefused


async def summarize_article(
    repository: BookshelfRepository,
    cognitive: CognitiveLayer,
    slug: str,
) -> CognitiveResponse:
    article = await repository.article(slug)
    return await _ask_and_enforce(
        cognitive,
        kind=AssistanceKind.SUMMARIZE,
        input_text=article.body,
    )


async def improve_grammar(
    repository: BookshelfRepository,
    cognitive: CognitiveLayer,
    slug: str,
    text: str,
) -> CognitiveResponse:
    article = await repository.article(slug)
    return await _ask_and_enforce(
        cognitive,
        kind=AssistanceKind.IMPROVE_GRAMMAR,
        input_text=text,
        context=article.body,
    )


async def suggest_title(
    repository: BookshelfRepository,
    cognitive: CognitiveLayer,
    slug: str,
) -> CognitiveResponse:
    article = await repository.article(slug)
    return await _ask_and_enforce(
        cognitive,
        kind=AssistanceKind.SUGGEST_TITLE,
        input_text=article.body,
    )


async def _ask_and_enforce(
    cognitive: CognitiveLayer,
    *,
    kind: AssistanceKind,
    input_text: str,
    context: str | None = None,
) -> CognitiveResponse:
    request = CognitiveRequest(kind=kind, input=input_text, context=context)
    response = await cognitive.ask(request)

    word = find_stop_word(response.suggestion)
    if word is not None:
        raise CognitiveOutputRefused(
            f"suggestion contains forbidden word {word!r}"
        )

    return response

Each function names what it does. _ask_and_enforce carries the common steps:

• build the CognitiveRequest
• call the cognitive layer
• run the article's forbidden-words policy on the suggestion

A model that returns "clickbait" is no more acceptable than a writer who types it.

Adding a fourth action is one new AssistanceKind value, one new prompt, one new application function, one extra branch in the HTTP dispatcher.

2.4 Domain Layer. Cognitive Layer

The domain only carries the abstraction and the editorial vocabulary it speaks in. The model is somebody else's problem — the infrastructure binding's, specifically.

# src/domain/cognitive_layer/entities.py

from enum import StrEnum

from src.domain.base import DomainModel


class AssistanceKind(StrEnum):
    SUMMARIZE = "summarize"
    IMPROVE_GRAMMAR = "improve_grammar"
    SUGGEST_TITLE = "suggest_title"


class CognitiveRequest(DomainModel):
    kind: AssistanceKind
    input: str
    context: str | None = None


class CognitiveResponse(DomainModel):
    suggestion: str
    confidence: float | None = None

# src/domain/cognitive_layer/layer.py

import abc

from .entities import CognitiveRequest, CognitiveResponse


class CognitiveLayer(abc.ABC):
    @abc.abstractmethod
    async def ask(self, request: CognitiveRequest) -> CognitiveResponse: ...

Prompts are domain knowledge — what the editor wants for each AssistanceKind. A plain dict so the infrastructure adapter indexes into it directly.

# src/domain/cognitive_layer/prompts.py

from .entities import AssistanceKind


PROMPTS: dict[AssistanceKind, str] = {
    AssistanceKind.SUMMARIZE: (
        "Summarize the following article in one or two sentences. "
        "Return only the summary."
    ),
    AssistanceKind.IMPROVE_GRAMMAR: (
        "Improve the grammar and clarity of the following text. "
        "Do not change the factual content. Return only the improved text."
    ),
    AssistanceKind.SUGGEST_TITLE: (
        "Suggest a short, clear title for the article below. "
        "Return only the title."
    ),
}

Changing the editorial voice is a one-file domain change. No model wiring moves.

2.5 Domain Errors

# src/domain/errors/__init__.py  (additions only)

class CognitiveOutputRefused(DomainError):
    def __init__(self, reason: str) -> None:
        super().__init__(f"Cognitive layer refused: {reason}")
        self.reason = reason


class CognitiveLayerUnavailable(DomainError):
    def __init__(self, reason: str) -> None:
        super().__init__(f"Cognitive layer unavailable: {reason}")
        self.reason = reason

• Refused → 422: the model said no (usage limit, content policy, empty response)
• Unavailable → 503: the model could not be reached (downtime, timeout, missing API key)

2.6 Infrastructure Layer. pydantic-ai

The only file that imports pydantic_ai. Provider exceptions are caught here and translated into the domain's failure vocabulary before they escape the module.

# src/infrastructure/pydantic_bindings.py

import os

from pydantic_ai import Agent
from pydantic_ai.exceptions import ModelHTTPError, UsageLimitExceeded

from src.domain.cognitive_layer import (
    PROMPTS,
    CognitiveLayer,
    CognitiveRequest,
    CognitiveResponse,
)
from src.domain.errors import CognitiveLayerUnavailable, CognitiveOutputRefused


class PydanticAICognitiveLayer(CognitiveLayer):
    default_model: str = "openai:gpt-4o-mini"
    model_env_var: str = "ASSIST_MODEL"

    def __init__(self, model: str | None = None) -> None:
        self._agent: Agent = Agent(
            model or os.getenv(self.model_env_var, self.default_model),
            output_type=str,
        )

    async def ask(self, request: CognitiveRequest) -> CognitiveResponse:
        prompt: str = self._render(request)

        try:
            result = await self._agent.run(prompt)
        except UsageLimitExceeded as exc:
            raise CognitiveOutputRefused(str(exc)) from exc
        except ModelHTTPError as exc:
            raise CognitiveLayerUnavailable(str(exc)) from exc

        suggestion: str = (result.output or "").strip()
        if not suggestion:
            raise CognitiveOutputRefused("empty response from the model")

        return CognitiveResponse(suggestion=suggestion)

    def _render(self, request: CognitiveRequest) -> str:
        parts: list[str] = [PROMPTS[request.kind], "\n\n", request.input]
        if request.context:
            parts.extend(
                ["\n---\nContext (surrounding article):\n", request.context]
            )
        return "".join(parts)

The default model identifier, the env-var name, and the prompt-rendering helper all live on the class — they are part of the same concept (the pydantic-ai adapter), so they sit inside it rather than as module-level constants.

2.7 Error Mapping

Two new entries in the registry; the route is unchanged.

# src/infrastructure/application/error_handlers.py  (additions only)

async def cognitive_output_refused_handler(
    _: Request, exc: CognitiveOutputRefused
) -> JSONResponse:
    return JSONResponse(
        status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
        content={"detail": str(exc), "reason": exc.reason},
    )


async def cognitive_layer_unavailable_handler(
    _: Request, exc: CognitiveLayerUnavailable
) -> JSONResponse:
    return JSONResponse(
        status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
        content={"detail": str(exc), "reason": exc.reason},
    )


ERROR_HANDLERS = (
    # ...existing entries from Part 1...
    (CognitiveOutputRefused, cognitive_output_refused_handler),
    (CognitiveLayerUnavailable, cognitive_layer_unavailable_handler),
)

2.8 pydantic-ai vs LangChain

I picked pydantic-ai over LangChain:

Swapping bindings means replacing one file. The CognitiveLayer ABC is what the rest of the system depends on; whether the implementation says pydantic_ai or langchain is invisible above.

2.9 Testing

The fake is the only new test scaffold.

# src/tests/fakes/cognitive.py

from src.domain.cognitive_layer import (
    CognitiveLayer,
    CognitiveRequest,
    CognitiveResponse,
)


class FakeCognitiveLayer(CognitiveLayer):
    def __init__(self, suggestion: str = "[fake suggestion]") -> None:
        self._suggestion = suggestion
        self.calls: list[CognitiveRequest] = []

    async def ask(self, request: CognitiveRequest) -> CognitiveResponse:
        self.calls.append(request)
        return CognitiveResponse(suggestion=self._suggestion)

One test per action verifies the use case wires the right kind and input. The grammar test additionally checks that the article body is forwarded as context.

# src/tests/unit/test_cognitive.py  (one test shown; full file in the repo)

@pytest.mark.asyncio
async def test_improve_grammar_uses_user_text_and_article_context(
    seeded_repository, cognitive
) -> None:
    await articles_use_cases.improve_grammar(
        seeded_repository,
        cognitive,
        slug="first-post",
        text="This is the paragraph to fix.",
    )

    assert cognitive.calls[0].kind == AssistanceKind.IMPROVE_GRAMMAR
    assert cognitive.calls[0].input == "This is the paragraph to fix."
    assert cognitive.calls[0].context == "Hello."

The full file also covers the missing-article path and a stop-word-on-AI-output case (the article's policy refusing a "clickbait" suggestion).

3. In Next Articles

Part 3 adds another external capability — pulling articles in from Medium and Reddit. The domain change stays small: one file inside the existing article aggregate. Infrastructure absorbs the two scrapers.

Part 4 flips the picture. A new internal feature — a seven-state article publication lifecycle with role-aware transitions, a split publication pipeline, and a supervisor review — forces the domain to grow.

A narrow contract is the only kind of contract that survives a second implementation.