Get Started

Your First Feature

This walkthrough describes a feature built end-to-end following the vertical slice pattern used throughout Spectral. It touches every layer — domain, application, infrastructure, API — wired together through the composition root.

The examples reference the rule_candidates feature inside spectral.worlds as the worked target shape — an operator-facing surface where the Operations app can propose a new rule candidate against a deployed action, the Evolution Loop’s gates verify it, and the operator enshrines it. Adapt the patterns to your own context.

Understand the context structure

Each context — spectral.worlds or spectral.platform — spans three Clean Architecture layers inside src/spectral/<context>/ and surfaces through a router in apps/api/src/spectral_api/routers/:

src/spectral/<context>/
├── domain/
│   ├── {feature}/
│   │   ├── models.py      # Frozen Pydantic entities + pure domain functions
│   │   └── exceptions.py  # Feature-specific error types
│   └── ...
├── application/
│   ├── {feature}/
│   │   ├── repositories.py  # Repository Protocol definitions
│   │   └── lifecycle.py     # Application services (orchestration)
│   └── ...
├── infrastructure/
│   └── {feature}/
│       └── repositories.py  # Concrete repo implementations (SQL)
└── contracts/                # Producer-owned public surface (per ADR-065)
    ├── events/               # Typed event payloads this context publishes
    └── protocols/            # Callee-owned OHS Protocols this context publishes

apps/api/src/spectral_api/
├── routers/operator/{feature}.py   # Operator HTTP endpoints
├── dependencies.py                 # Composition root — wires infra → app
└── models/                         # Request/response Pydantic models

Step 1: Domain model

Add your entity to domain/{feature}/models.py. Domain models are frozen Pydantic models — immutable value objects with no I/O and no framework dependencies.

The example below is illustrative — it shows the shape a feature’s domain model takes, not a literal current file. (The real authoring domain lives under src/spectral/worlds/domain/authoring/, .../pending_candidates/, and .../enshrinement/; action assignment is membership and can happen before or after approval, while publish/deploy only gather assigned enshrined rules.) A frozen-Pydantic candidate model looks like:

from __future__ import annotations

import uuid
from datetime import UTC, datetime
from typing import Literal

from pydantic import BaseModel, ConfigDict, Field


RuleStatus = Literal["candidate", "enshrined", "retired"]


class RuleCandidate(BaseModel):
    """A proposed rule under review, not yet enshrined."""

    model_config = ConfigDict(frozen=True)

    id: uuid.UUID = Field(default_factory=uuid.uuid4)
    org_id: str
    domain_id: str
    action: str                                    # e.g. "wire_transfer.release"
    natural_language: str                          # the operator-authored or distilled form
    tier: Literal["T1", "T2", "T3"]
    outcome: Literal["GREEN", "GREEN-SKIP", "YELLOW", "RED"]
    status: RuleStatus = "candidate"
    applies_when_source: str | None = None
    generated_code: str | None = None              # populated after World Agent code-gen
    inline_tests: str | None = None
    created_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
    updated_at: datetime = Field(default_factory=lambda: datetime.now(UTC))

Conventions to follow:

model_config = ConfigDict(frozen=True) — all domain models are immutable.
Use Field(default_factory=...) for mutable defaults and auto-generated IDs.
State transitions are pure functions that return a new instance via model_copy(update={...}).
Domain functions live in the same models.py file alongside the entities they operate on.

Module and package conventions:

No underscore-prefixed module files — use helpers.py, not _helpers.py. The underscore prefix is reserved for private names within a module, not for module filenames.
Public API is defined by __init__.py re-exports using the self-alias pattern:
domain/rule_candidates/__init__.py
```
from .models import RuleCandidate as RuleCandidate
from .models import transition_status as transition_status
```
The as Name form signals to type checkers and readers that this is an intentional public re-export. Do not use __all__.
Consumers should import from the package (spectral.worlds.domain.rule_candidates) rather than reaching into internal modules (spectral.worlds.domain.rule_candidates.models) when possible.

If your context needs specific error types, add them to domain/{feature}/exceptions.py, inheriting from the shared hierarchy in spectral.core.errors:

from spectral.core.errors import InvalidTransitionError

class RuleCandidateTransitionError(InvalidTransitionError):
    def __init__(
        self, from_state: str, to_state: str, valid_targets: list[str] | None = None,
    ) -> None:
        super().__init__("RuleCandidate", from_state, to_state, valid_targets)

See Domain Model for the full entity reference.

Step 2: Repository protocol

Define the data-access contract in application/{feature}/repositories.py. Protocols live in the application layer (not domain) because they describe what the use cases need, not what the domain is.

From src/spectral/worlds/application/rule_candidates/repositories.py:

from __future__ import annotations

from typing import TYPE_CHECKING, Protocol, runtime_checkable

if TYPE_CHECKING:
    import uuid
    from spectral.worlds.domain.rule_candidates.models import RuleCandidate, RuleStatus

@runtime_checkable
class RuleCandidateRepo(Protocol):
    """Protocol for RuleCandidate data access."""

    def create(self, candidate: RuleCandidate) -> RuleCandidate: ...
    def get_by_id(self, candidate_id: uuid.UUID) -> RuleCandidate | None: ...
    def list_by_action(
        self,
        org_id: str,
        domain_id: str,
        action: str,
        *,
        status: RuleStatus | None = None,
        limit: int = 50,
    ) -> list[RuleCandidate]: ...
    def update_status(self, candidate_id: uuid.UUID, new_status: RuleStatus) -> None: ...

Key patterns:

@runtime_checkable — allows isinstance() checks if needed.
TYPE_CHECKING guard — domain imports only happen at type-check time, keeping the module lightweight at runtime and satisfying the architecture validator.
Method signatures use domain types, not raw dicts or ORM objects.

Step 3: Application service

Create or extend a service in application/{feature}/. Services orchestrate domain logic and repository calls. They receive repositories through constructor injection and return Result[T] for expected failures.

From src/spectral/worlds/application/rule_candidates/lifecycle.py:

import uuid

from spectral.core.errors import NotFoundError
from spectral.core.result import Failure, Result, Success
from spectral.worlds.domain.rule_candidates.exceptions import (
    RuleCandidateTransitionError,
)
from spectral.worlds.domain.rule_candidates.models import (
    RuleCandidate, RuleStatus, transition_status,
)


class RuleCandidateLifecycleService:
    """Application service for RuleCandidate lifecycle operations."""

    def __init__(self, candidate_repo, approval_decision_repo):
        self._cand_repo = candidate_repo
        self._approval_repo = approval_decision_repo

    def approve_candidate(
        self,
        candidate_id: uuid.UUID,
        *,
        operator_id: uuid.UUID,
        correlation_id: str,
    ) -> Result[RuleCandidate]:
        candidate = self._cand_repo.get_by_id(candidate_id)
        if not candidate:
            return Failure(NotFoundError("RuleCandidate", candidate_id))

        try:
            enshrined = transition_status(candidate, "enshrined")
        except RuleCandidateTransitionError as e:
            return Failure(e)

        self._cand_repo.update_status(candidate_id, "enshrined")
        self._approval_repo.record(
            candidate_id=candidate_id,
            operator_id=operator_id,
            correlation_id=correlation_id,
        )
        return Success(enshrined)

The Result pattern:

Success(value) for happy paths.
Failure(DomainError) for expected business failures (not found, invalid transition, etc.).
Programming errors (invariant violations) still raise exceptions.
The API layer pattern-matches on the result to produce the right HTTP response.

# The Result type is defined in spectral.core.result
type Result[T] = Success[T] | Failure

Step 4: Infrastructure repository

Implement the protocol in infrastructure/{feature}/repositories.py. This is where SQL, external API calls, and other I/O live.

From src/spectral/worlds/infrastructure/rule_candidates/repositories.py:

import uuid

from spectral.worlds.domain.rule_candidates.models import RuleCandidate, RuleStatus


class RuleCandidateRepository:
    """Data access for RuleCandidate lifecycle operations."""

    def __init__(self, conn):
        self._conn = conn

    def get_by_id(self, candidate_id: uuid.UUID) -> RuleCandidate | None:
        with self._conn.cursor() as cur:
            cur.execute(
                "SELECT id, org_id, domain_id, action, natural_language, tier, "
                "outcome, status, applies_when_source, generated_code, "
                "inline_tests, created_at, updated_at "
                "FROM worlds.rule_candidates WHERE id = %s",
                (str(candidate_id),),
            )
            row = cur.fetchone()
        if not row:
            return None
        return self._row_to_candidate(row)

    def _row_to_candidate(self, row) -> RuleCandidate:
        return RuleCandidate(
            id=uuid.UUID(str(row[0])),
            org_id=row[1],
            domain_id=row[2],
            action=row[3],
            natural_language=row[4],
            tier=row[5],
            outcome=row[6],
            status=row[7],
            applies_when_source=row[8],
            generated_code=row[9],
            inline_tests=row[10],
            created_at=row[11],
            updated_at=row[12],
        )

Conventions:

Constructor takes a raw conn (psycopg connection) — no ORM.
Private _row_to_{entity} mapper converts database rows to domain models.
Tenancy filtering keys on org_id + domain_id per ADR-086 D6.
The class does not declare that it implements the protocol — Python structural typing handles this automatically. As long as the methods match, it satisfies the protocol.

Step 5: API router

Add endpoints in apps/api/src/spectral_api/routers/operator/{feature}.py. Routers are thin — they parse HTTP, call the service, and format the response.

From apps/api/src/spectral_api/routers/operator/rule_candidates.py:

import uuid

from fastapi import APIRouter, Depends

from spectral.core.result import Failure, Success
from spectral.worlds.application.rule_candidates.lifecycle import (
    RuleCandidateLifecycleService,
)
from spectral_api.dependencies import (
    get_rule_candidate_service,
    get_db_connection,
)
from spectral_api.middleware.auth_v2 import AuthContext, require_operations_scope
from spectral_api.problem_details import problem_details_response

router = APIRouter(tags=["operator", "rule_candidates"])


@router.post(
    "/api/operator/rule-candidates/{candidate_id}/approve",
    response_model=RuleCandidateDetailResponse,
)
def approve_candidate(
    candidate_id: str,
    body: ApproveCandidateRequest,
    auth: AuthContext = Depends(require_operations_scope("approve:modules")),
    svc: RuleCandidateLifecycleService = Depends(get_rule_candidate_service),
):
    match svc.approve_candidate(
        uuid.UUID(candidate_id),
        operator_id=auth.user_id,
        correlation_id=body.correlation_id,
    ):
        case Failure(error):
            return problem_details_response(error)
        case Success(candidate):
            return RuleCandidateDetailResponse.from_domain(candidate)

Patterns to follow:

Depends(require_operations_scope("scope")) for operator-side RBAC on every route.
Depends(get_rule_candidate_service) injects the fully-wired service from the composition root.
match/case on Result — Failure maps to problem_details_response() (RFC 9457), Success continues to build the response.
Response models are separate Pydantic classes defined in the router file or in spectral_api/models/.

Step 6: Composition root

Wire your new service in apps/api/src/spectral_api/dependencies.py. This is the only file that imports from both spectral.<context>.application and spectral.<context>.infrastructure.

def get_rule_candidate_service(db=Depends(get_db_connection)):
    """Provide a fully-wired RuleCandidateLifecycleService."""
    from spectral.worlds.application.rule_candidates.lifecycle import (
        RuleCandidateLifecycleService,
    )
    from spectral.worlds.infrastructure.rule_candidates.repositories import (
        ApprovalDecisionRepository,
        RuleCandidateRepository,
    )

    return RuleCandidateLifecycleService(
        candidate_repo=RuleCandidateRepository(db),
        approval_decision_repo=ApprovalDecisionRepository(db),
    )

Why lazy imports inside the function? The composition root is the boundary — by importing inside the function body, the module-level import graph stays clean, and the architecture validator sees that only dependencies.py bridges the application/infrastructure gap.

Then register your router in apps/api/src/spectral_api/main.py:

from spectral_api.routers.operator import rule_candidates
app.include_router(rule_candidates.router)

Step 7: Tests

Spectral uses layered test markers. For a new feature, write at minimum:

Unit test — domain logic

Test pure domain functions with no mocks, no I/O. These run in milliseconds.

import pytest
from spectral.worlds.domain.rule_candidates.models import (
    RuleCandidate, transition_status,
)
from spectral.worlds.domain.rule_candidates.exceptions import (
    RuleCandidateTransitionError,
)


class TestTransitionStatus:
    def test_candidate_to_enshrined(self, rule_candidate_factory):
        candidate = rule_candidate_factory(status="candidate")
        result = transition_status(candidate, "enshrined")
        assert result.status == "enshrined"

    def test_invalid_transition_raises(self, rule_candidate_factory):
        candidate = rule_candidate_factory(status="retired")
        with pytest.raises(RuleCandidateTransitionError):
            transition_status(candidate, "enshrined")

Integration test — paths between contexts

If your feature touches more than one context, integration tests are non-negotiable per AGENTS.md — exercising the full path (producer → substrate → consumer or caller → bridge → callee) against real infrastructure. Isolated unit tests do not satisfy this AC; each epic’s Definition of Done lists the integration test as a load-bearing requirement for work that spans contexts.

The two flow shapes — notification (events + ACL) and call (callee-owned OHS Protocol + bridge tool in apps/*) — are detailed in Contract Surfaces. For the bilateral contract test pattern that pins event drift between producer and consumer, see Testing — Bilateral contract tests.

If your feature is single-context and does not cross any of these seams, contract and unit tests are sufficient.

Contract test — API router

Test the HTTP boundary with a real (or test) database. These verify auth, status codes, and response shapes.

import pytest

@pytest.mark.contract
class TestApproveCandidate:
    def test_requires_auth(self, client):
        resp = client.post(
            "/api/operator/rule-candidates/123/approve",
            json={"correlation_id": "test-123"},
        )
        assert resp.status_code == 401

    def test_requires_operations_scope(self, customer_authenticated_client, candidate_id):
        resp = customer_authenticated_client.post(
            f"/api/operator/rule-candidates/{candidate_id}/approve",
            json={"correlation_id": "test-123"},
        )
        assert resp.status_code == 403

    def test_approves_pending_candidate(self, operator_client, pending_candidate_id):
        resp = operator_client.post(
            f"/api/operator/rule-candidates/{pending_candidate_id}/approve",
            json={"correlation_id": "test-123"},
        )
        assert resp.status_code == 200
        assert resp.json()["status"] == "enshrined"

Verification checklist

Before pushing, run the full quality gate:

# Individual checks
uv run ruff check                                       # Lint — 0 errors (incl. ANN family)
uv run ruff format --check                              # Format — 0 diffs
uv run ty check                                         # Type check — 0 diagnostics (strict-max)
uv run pytest -m "unit or contract" -q                  # Fast tests
uv run python tools/quality/validate_architecture.py    # Import boundaries
uv run python tools/quality/check_migration_naming.py   # Migration naming
uv run python tools/quality/check_migration_compat.py   # Migration expand/contract safety
uv run python tools/quality/check_deploy_manifest_coverage.py

# Or all at once (recommended)
bash tools/dev/precheck.sh

ty is the canonical Python type-check gate per type-checking (ADR-051); mypy still runs as an informational warning inside precheck.sh during the transition. The architecture validator will catch any import boundary violations — for example, if your application service accidentally imports from infrastructure, or your domain model imports from application. Fix these before pushing; they block CI.

Local dev orchestration

bash tools/dev/start.sh brings up the full local stack — Supabase (Postgres + Auth + Storage + Realtime) plus the API and workers. The script is the canonical local entrypoint; CONTRIBUTING.md → Quick start in the repo root is the source of truth for prerequisites and one-shot setup.

bash tools/dev/setup.sh   # one-shot environment bootstrap (first run only)
bash tools/dev/start.sh   # boot the local stack

Summary

Every feature follows the same vertical path through the codebase:

Step	Layer	File	What you add
1	Domain	`domain/{feature}/models.py`	Frozen Pydantic entity + pure functions
2	Application	`application/{feature}/repositories.py`	`Protocol` for data access
3	Application	`application/{feature}/{service}.py`	Service with constructor injection + `Result`
4	Infrastructure	`infrastructure/{feature}/repositories.py`	SQL implementation + row mapper
5	API	`spectral_api/routers/.../{feature}.py`	Endpoint with `Depends()` + `match`/`case`
6	Composition	`spectral_api/dependencies.py`	Factory function wiring infra to app
7	Tests	`tests/unit/...`, `tests/contract/...`	Domain unit tests + API contract tests

The context structure keeps each feature self-contained. When you open a context directory, everything related to that concept is right there — models, services, repos, exceptions — across all three layers.

Previous
Operator Walkthrough Next
Overview