Declare a governed aggregate

Standing up a governed aggregate by hand means assembling the same wiring every time: the document registry, the typed facade, the search index-on-write sync, the soft-delete read filter, the invariant checks inside the transaction, and the four-piece outbox dance. Each verb is one call; the composition between them is not.

AggregateKit collapses that composition into one typed declaration. It composes wiring, not models — you still write the four models — and emits the slice as separate artifacts, so the app and backend layers stay decoupled.

The runnable version lives at examples/recipes/aggregate_kit/ and runs on the in-memory mock — no infrastructure needed.

The four models¶

The models are yours — the only thing the kit cannot invent. Here Task extends the soft-delete mixin (the precondition for soft_delete=True) and emits an event when it is completed:

# The four models — the author's, and the only thing the kit cannot invent. `Task` extends the
# soft-delete mixin (the type-level precondition for `soft_delete=True`) and emits an event when it
# is completed; the emitter fires as the governed update persists.
class TaskCompleted(DomainEvent):
    aggregate_id: UUID


class Task(DocWithSoftDeletion, AggregateRoot):
    project_id: str
    title: str
    points: int = 1
    done: bool = False

    @event_emitter(fields={"done"})
    def _on_complete(before, after: Self, diff: JsonDict) -> DomainEvent | None:  # type: ignore[no-untyped-def]
        if after.done and not before.done:
            return TaskCompleted(aggregate_id=after.id)

        return None


class TaskCreate(CreateDocumentCmd):
    project_id: str
    title: str
    points: int = 1


class TaskUpdate(UpdateCmdWithSoftDeletion):
    title: str | None = None
    points: int | None = None
    done: bool | None = None


class TaskRead(ReadDocument):
    project_id: str
    title: str
    points: int = 1
    done: bool = False
    is_deleted: bool = False

One declaration¶

AggregateKit bundles the spec with its optional concerns. This one is persisted, soft-deletable, kept in an external search index on every write, guarded by a cross-record invariant, and event-relaying:

# One declaration → the governed slice: persisted + soft-deletable + searchable-and-synced +
# invariant-guarded + event-relaying. The kit composes the wiring; the models stay yours.
TASKS = AggregateKit(
    spec=TASK_SPEC,
    soft_delete=True,
    search=TASK_INDEX,
    invariants=(PROJECT_BUDGET,),
    outbox=OutboxEmit(
        spec=TASK_EVENTS,
        emits=(
            EmitMapping(
                event=TaskCompleted,
                event_type="task.completed",
                to_payload=lambda e: TaskEventPayload(task_id=str(e.aggregate_id)),
            ),
        ),
        relay=RelayBinding(queue_spec=TASKS_QUEUE),
    ),
)

Each concern is opt-in and independently useful on its own:

soft_delete — the generated LIST excludes soft-deleted rows and GET 404s one; you get delete/restore ops and an optional after-commit purge. With get_deleted="read", GET returns the deleted row with is_deleted set instead, so a reference to it still resolves. A model can name companion fields that change only in the write that deletes or restores the row (soft_delete_companions = frozenset({"deleted_with"})), such as a cascade marker the delete sets and the restore clears; any other update that touches one is refused. Declare hard_delete=False on the spec when no document operation may erase a row: the kit then generates no kill and refuses a handler that adds one.
search — the external index is kept in sync on every committed write (search stays consistent); a searchable aggregate that silently drifts is worse than no search. Delivery is after-commit best-effort by default (bounded in-place retry, then an at-most-once loss logged with reconcilable identity — the index stays stale for that row until its next write). Pass search_delivery=OutboxSearchSync() to route index maintenance through a dedicated transactional outbox instead: an identity-only marker (no row data, so nothing sensitive rides the outbox) is staged in the same transaction as the write, relayed at-least-once, and applied by a consumer that re-reads the row's committed state — idempotent and reorder-safe (a stale upsert marker delivered after a later delete re-reads the deleted state, so it can never resurrect a ghost), deduped by an inbox. Wire the reported search_sync_route (one name for its outbox, queue, and inbox) in the deps module; lifecycle_steps(tx_route=…) then carries the sync relay and consumer. Composed with soft_delete, the kit's search reads also exclude soft-deleted rows — the spec must declare is_deleted in facetable_fields (which ensure_index provisions as a filterable attribute on external indexes); the kit fails closed at construction otherwise.
versioned — correction lineage: the aggregate stops being overwritten and starts being corrected. See Correcting a fact instead of overwriting it.
temporal — effective dating: the aggregate gains a validity period, and the question "which row applied on day D" stops being a scan. See Reading what applied on a day.
invariants — each SystemInvariant is enforced preventively inside the write transaction, at the isolation floor it needs, so a write that would break the law is rolled back.
outbox — the transactional outbox wiring: the in-tx flush, the domain-event → outbox bridge, and the relay step, from one OutboxEmit.

Correcting a fact instead of overwriting it¶

A regulated ledger is not allowed to overwrite a fact. versioned=VersionedPolicy(...) makes the aggregate keep every assertion it ever made: the old value stays readable, the new one says what it replaced, and somebody's name is on the change.

AggregateKit(
    spec=readings,                                   # domain + update cmd on the versioning mixins
    versioned=VersionedPolicy(corrections=corrections_spec),
)

It adds five fields — root_id, version, supersedes_id, is_current, superseded_at — and three operations. correct(id, expected_version, patch, reason) retires the current version and inserts a successor carrying the predecessor's values with your patch over them, recording who and why in one transaction. history(root_id) returns the chain oldest-first; as_of(root_id, at) returns the version that was current at an instant. Every generated read stays on current versions, so an aggregate that opted in reads like one that never did.

The generated update stops being a way to change what a fact says. It refuses a patch touching any field the row asserts, and leaves exactly two writes reaching a version: the one that retires it, and soft deletion, which hides a row without contradicting it. The refusal is on the domain model rather than on the operation, so a repair script or a hand-written handler meets it too.

Three things are worth knowing before you declare it.

"Current" is a stored flag, not a derived anti-join. NOT EXISTS (SELECT 1 FROM t s WHERE s.supersedes_id = t.id) is the obvious way to express it and the wrong one: an index cannot see an anti-join and neither can a SystemInvariant, both of which read a column. The flag is redundant state on purpose, and the price is that every write must go through the kit's handlers.

The index that protects it is your migration to write. The kit refuses to build unless the spec declares both guarantees, and startup refuses a deployment whose indexes are missing, naming the DDL:

guarantees = (
    UniqueTogether(fields=("root_id",), where={"$values": {"is_current": True}}),
    UniqueTogether(fields=("supersedes_id",), skip_null=True),
)

One current version per fact, and one successor per predecessor. The second is what stops two concurrent corrections forking the chain, and it is not optional — without it both corrections commit and every chain walker picks whichever row it saw first.

On Mongo, sparse is not the index you want

The second guarantee exempts the first versions, whose supersedes_id is null — and a sparse unique index does not do that: it skips a document only when every indexed field is missing, and it still indexes an explicit null, so two first versions collide under it. The exemption is a partialFilterExpression naming what the field is ({supersedes_id: {$type: "string"}}), which is how a partial filter says "not null" given it admits no negation. Startup checks for that specifically and refuses a sparse index offered in its place, printing the statement that would work.

The kit declares the law, so your writes run serializable. single_current_head — at most one current version per fact — comes with versioned=, and it is the detective control for a path that reached the rows outside the kit's handlers. Enforcing it preventively is only correct at SERIALIZABLE, so the kit raises every write on the aggregate to that floor and fails closed if the wired transaction manager does not report it. That is a deployment requirement, not a default you can lower.

A correction writes two aggregates, so the corrections relation is yours to declare and wire: its route, its encryption policy and its retention are facts only you hold. Its create command must be CreateCorrectionCmd. Each correction records who made it, as an audit row does: actor_id is the principal, or for a delegated call the nearest actor in its chain; subject_id is the principal it acted for; actor_ids is the whole chain, nearest first. Its domain and read models must declare all three, and none may be lenient (read_conformity="lenient" makes every defaulted field lenient) or in write_omit_fields; otherwise the kit refuses the declaration rather than lose them on every write or read. A Postgres corrections table created before these existed needs them — until then every correction fails on the write:

ALTER TABLE <corrections> ADD COLUMN subject_id uuid;
ALTER TABLE <corrections> ADD COLUMN actor_ids uuid[] NOT NULL DEFAULT '{}';

Rows a delegated correction wrote on 0.9.0 or earlier hold the user it acted for in actor_id; rows from 0.10.0 on hold the agent. Those earlier rows have no subject_id and an empty actor_ids: the chain was never recorded, so it cannot be recovered for them.

Reading what applied on a day¶

Contracts, price lists, tax rates and consent records are one shape: a key, a period, a payload, and the rule that two periods for one key may not overlap. temporal=TemporalPolicy(...) declares it once.

AggregateKit(
    spec=contracts,                                  # domain + create cmd on the validity mixins
    temporal=TemporalPolicy(key=("employee_id",), bounds="[]"),
)

It adds valid_from and a nullable valid_to, and two reads. effective_on(key, on) returns the one row in force that day, or not_found. timeline(key, start, end) returns every row whose period meets the window, earliest first, as a page — a month is one query rather than thirty.

Three things are worth knowing before you declare it.

valid_to means what the declaration says it means, and nothing else. bounds="[]" puts both endpoints in force, which is what "valid through 31 March" means to the person who typed it; bounds="[)" excludes the end, which is what tiles cleanly at a day or instant grain. The same value is read by effective_on, by timeline and by the guarantee the store enforces, so they cannot disagree — and the domain model carries it too, as temporal_bounds, because an update that moves an end date carries no start date and only the model sees the period the write produces. The kit refuses to build when the two disagree.

None is open-ended, never a sentinel date. A 9999-12-31 simplifies one predicate and then lies in every export, report and column a human reads.

The non-overlap rule is the store's, and the constraint is your migration to write. The kit refuses to build unless the spec declares the guarantee, and startup refuses a deployment whose constraint is missing, naming the DDL:

guarantees = (
    NonOverlapping(key=("employee_id",), period=("valid_from", "valid_to"), bounds="[]"),
)

A read-then-insert check in the kit could not be correct: two writers each see no conflict and both insert. On Postgres the mechanism is an exclusion constraint over a range type, and startup checks the bounds it was built with as well as its columns — [] and [) agree about every day but the boundary, which is the day a guarantee is read for.

Mongo refuses an effective-dated aggregate, deliberately

Non-overlap is a pairwise comparison between rows, not a property of one row's fields, so unlike filtered uniqueness it is not expressible as a partial index. Mongo has no mechanism for it, so the guarantee is unsatisfiable there and wiring fails closed rather than serving an aggregate whose declared rule nothing keeps.

A period in force on no day is refused. Under bounds="[)", 1 March to 1 March covers zero days: nothing would read the row and nothing would conflict with it. Ending a record on the day it started means deleting it, so the write is refused rather than stored — on the domain model, so a repair script meets it too.

Composed with versioned, the guarantee is scoped to the current versions. That pair is the bitemporal case — when the fact applied, and when you asserted it — and it needs the filtered form, because a correction writes a successor carrying its predecessor's dates. Over every row those two overlap and the correction is refused; over the rows in force the property still says what it meant. Declare it with the filter and the kit wires both arms:

guarantees = (
    NonOverlapping(
        key=("employee_id",),
        period=("valid_from", "valid_to"),
        bounds="[]",
        where={"$values": {"is_current": True}},
    ),
    ...  # the two the versioned arm requires
)

The kit refuses the unfiltered form on such an aggregate rather than letting its first correction fail, and the dated reads scope themselves the same way — effective_on answers with the version in force, never one that has since been corrected. On Postgres the mechanism is EXCLUDE ... WHERE (...), and startup checks the predicate names the fields the filter selects on, exactly as it does for a partial unique index.

What it emits — separately¶

The kit never returns a coupled god-object. It emits the app-layer registry, the typed facade, and the domain-event bridges as distinct artifacts; which store backs it stays your call, wired into the deps module:

# The kit emits its pieces separately: the staging bridges go on the deps module, the typed facade
# and the frozen registry stay in the app layer. Which store backs it (backend config) stays yours.
def build_stack() -> tuple[
    ExecutionRuntime,
    OperationFacadeFactory[DocumentFacade[TaskRead, TaskCreate, TaskUpdate]],
]:
    module = MockDepsModule(domain_events=TASKS.domain_events())
    runtime = build_runtime(module)
    tasks = TASKS.facade(runtime, tx_route=_TX)
    return runtime, tasks

registry(tx_route=…) — the composed, frozen operation registry.
facade(runtime, tx_route=…) — a per-call, precisely-typed DocumentFacade over it (create/get/list/update keep your C/U/R types — no erosion). Pass the same tx_route you gave registry() and the routes emitter, so every surface executes the write ops on the one transaction route the deps module registers a manager under.
domain_events() — the outbox staging bridges, for the deps module.
lifecycle_steps(tx_route=…) — the outbox relay (plus, with durable search_delivery, the search-sync relay and consumer), for the runtime.

Field encryption declared on the spec flows through untouched. Backend config (rw_documents= / searches= / outboxes=) and HTTP routes stay yours — wire them over registry() with your deps module and the route generators — so the hexagonal layer split holds.

Shaping the writes¶

Three options change how the generated writes behave without replacing a handler:

kit = AggregateKit(
    spec=ORDER_SPEC,
    dtos=DocumentDTOs(read=OrderRead, create=OrderRequest, update=OrderUpdate),
    mappers=DocumentMappers(
        create=PydanticPipelineMapperFactory(
            in_=OrderRequest,
            out=CreateOrderCmd,
            step_factories=(NumberIdMappingStepFactory(spec=ORDER_NO, name_field="name"),),
        ),
    ),
    transactional_writes=True,
    update_returns="record",
)

mappers= / dtos= — how an inbound DTO becomes the domain command, and which DTOs the generated routes and tools advertise. As in build_document_registry, a slot left out disables its operation (DocumentDTOs(read=…, create=…) is an aggregate with no update). Your mappers are the base the other options compose on: soft_delete adds its exclusion after your list mapper; versioned seeds the first version through your create mapper, maps a correction through your update mapper as update does, and strips its lineage fields from what that mapper produces. List mappers apply to the document list operations only — not to get, search, or the versioned and temporal reads.
transactional_writes=True — create, update, kill and the soft-delete transitions run in a transaction on the registry's tx_route, so the document reads and writes a mapper makes commit or roll back with the write. A counter does not: counters allocate on their own connection, so a create that fails after its number-id step leaves a gap. Off by default.

A write option whose operation is not generated — a create mapper on a read-only spec, update_returns="record" with the update slot left empty — is refused when the kit is built. - update_returns="record" — the generated update returns the updated record instead of {data, diff}, and its route advertises the read model. The typed facade keeps the default's return type.

The escape hatch¶

The kit gives the governed-CRUD floor; a bespoke lifecycle comes through a first-class escape hatch, not a fall-off-the-cliff:

kit = AggregateKit(
    spec=TASK_SPEC,
    handlers={DocumentKernelOp.UPDATE: MyCustomUpdate},  # override a generated op's handler
    extra_ops=my_report_registry,                        # merge bespoke operations
)

handlers= replaces a generated op's handler while keeping the rest of the slice (routes/facade/relay) intact; extra_ops= merges custom operations (a lifecycle transition, a report) into the composed registry.

Verifiable by construction¶

Because the kit owns the composition, the invariant you declared isn't only enforced — it is verifiable. The same SystemInvariant compiles into a DST conformance oracle, so a deterministic simulation can prove the enforcement holds under concurrent interleaving, not just single-threaded. The only bridge is compile_oracle; the kit and its models are unchanged:

# The moat: the same law compiles into a DST oracle. A deterministic simulation drives the kit's
# operations under concurrent interleavings and proves the enforcement holds — no double-book of
# the sprint's capacity. `compile_oracle` is the only bridge; the kit and its models are unchanged.
_ORACLE = compile_oracle(SPRINT_CAPACITY)

simulation = Simulation(
    operations=TICKETS.registry(tx_route="mock"),
    deps=lambda: MockDepsModule(),
    observe=_ORACLE.observe,
    invariants=[*_ORACLE.invariants],
)

Run it — every operation raced another and the declared invariant still held:

forze dst run examples.recipes.aggregate_kit_dst.app:simulation   # ✓ no violation found

A bare registry with no enforcement double-books the capacity under the same schedule; the kit-composed slice holds. "Declare an aggregate, get a slice that is also DST-verifiable" is the payoff of a framework-owned composition — scattered generated code could not be uniformly verified. See Deterministic Simulation Testing for the full model.

Honest limits¶

It does not reduce the models. You still write Domain / Create / Update / Read + the DocumentSpec — Python has no Partial<T>/Pick<T>, and the four contracts genuinely diverge. The kit collapses wiring, never the models.
It does not invent a lifecycle. soft_delete=True gives the governed-CRUD floor; a status machine (like StoredFileKitSpec's pending → ready → failed) comes through the escape hatch.
A wholly lifecycle-driven aggregate is its own shape, not a kit with many overrides. The escape hatch is for overriding a few generated ops. When the entire operation surface is bespoke — StoredFileKitSpec's upload/download/delete are a create-then-upload lifecycle, a cross-port join, and a status-based delete, with no plain create/update at all — hand-wire it (as StoredFileKitSpec does); the kit's generated CRUD would only get in the way.
versioned= is not an audit log, and not bitemporal. A Correction records one aggregate's lineage — who changed this fact and why — not cross-aggregate who-did-what. And it versions assertions about a fact, not the period the fact applies to; validity over time is a separate declaration that composes with this one.
It does not couple to a backend. registry() / facade() are backend-agnostic; you wire the store yourself. No AggregateKit(...).build_everything(client).
storage= gives the blob ops, not the join. Declaring an object-storage bucket exposes upload/download/head/delete alongside the document ops (its name must differ from the document's), but correlating a row to its blob — a storage_key field, an upload-then-create lifecycle — is yours, through the escape hatch.

Projecting it¶

registry() is the backend-agnostic surface; wire the store into a deps module and the routes onto a FastAPI router. forze_fastapi.attach_aggregate_routes projects the whole slice — document, soft-delete, search, and (under a /blobs sub-path) storage — in one call:

from forze_fastapi.routes import attach_aggregate_routes

attach_aggregate_routes(router, TASKS, ctx_dep=ctx_dep, style="rest", tx_route="pg")

The routes execute through the composed registry, so tx_route must match the deps module. kit.backend_requirements(tx_route="pg") reports the document / search / search-sync / storage / outbox routes, the transaction route, and the crypto requirement (crypto_required — whether a keyring is needed) that module must provide — a checklist you can assert in a startup test (and check_wiring fails closed on anything missing).

Notes¶

The outbox flush is attached to update (the op that emits): @event_emitter fires as an update persists, so a generated create never stages. Emit on create/delete through the escape hatch.
Preventive invariants default to SERIALIZABLE — the isolation most predicate-over-a- read-set laws need to survive write skew. The kit opens the write transaction there.