How to write a source adapter¶

The contract every adapter that implements WritableSourcePort must honor. Verified by packages/sdk-py/tests/test_port_contract.py, parametrized over [FilesystemWritableSource, SqlAlchemySource[sqlite], SqlAlchemySource[postgres]].

The raw SQL adapters are retired

The raw SqliteSource / PostgresSource were removed (s-retire-raw-sql-adapters): SqlAlchemySource is the only SQL source in the Python SDK — same tables, both dialects, zero data migration (see the migration note in the changelog).

A new adapter is considered production-ready only when its row in the contract test suite is fully green (all tests pass or skip explicitly with a documented reason — never fail).

Mandatory contract (every WritableSource)¶

Capability Protocols (dna.kernel.capabilities)¶

Capability	What it covers
`KernelAttachable`	`attach_kernel(kernel)` — accept post-init wiring of `_writers` + `_readers` from the kernel. Idempotent. Raises `TypeError` if `kernel` is not a `Kernel` instance.
`BundleEntryReadable`	`fetch_bundle_entry(scope, container, name, entry, *, tenant) -> bytes \\| Awaitable[bytes]`. Returns single bundle entry bytes; raises `FileNotFoundError` on miss. Honors tenant overlay.
`Versionable`	`get_version(scope, kind, name, version_id) -> dict`. Per-Kind semver versioning (Phase 10 catalog flow). Adapters that don't support this should omit the method; callers should degrade gracefully (report the capability as unsupported) instead of crashing.

When to add a new capability¶

Adding MyCapability is a 4-step process:

Define MyCapability(Protocol) with @runtime_checkable in packages/sdk-py/dna/kernel/capabilities.py.
Replace any hasattr(adapter, "method") in the kernel with isinstance(adapter, MyCapability).
Instance the capability here.
Cover it in packages/sdk-py/tests/test_port_contract.py so adapters either implement it or get explicitly skipped.

Round-trip¶

Operation	Acceptance
`save_instance(scope, "Genome", scope, raw)` then `publish(...)`	The root Genome appears in `mi.root` after `kernel.instance_async(scope)`.
`kernel.write_instance(scope, "Skill", name, raw)` then `publish(...)`	Skill appears in `mi.instances` / `kernel.query(scope, "Skill")`. Bundle entries (e.g. `SKILL.md`) persisted via the source's backing store.
`kernel.fetch_bundle_entry_async(scope, kind, name, entry)`	Returns `bytes` for existing entries; `FileNotFoundError` for missing entries (consistent across all adapters).

Boot-time validation (kernel-level, propagates uniformly)¶

Trigger	Outcome
Two `KindPort`s with the same `(api_version, kind)` tuple	`KindRegistrationError` from `kernel.kind(port)`
Two `KindPort`s with the same `alias`	`KindRegistrationError` from `kernel.kind(port)`
Two BUNDLE-pattern Kinds with the same `(storage.container, storage.marker)` and neither has `marker_shared_allowed = True`	`KindRegistrationError` from `kernel.kind(port)`
Reader missing `detect()` or `read()`	`ReaderRegistrationError` from `kernel.reader(r)`
Writer missing `can_write()` or `write()`	`WriterRegistrationError` from `kernel.writer(w)`
Extension missing `register()` callable	`ExtensionLoadError` from `kernel.load(ext)`

Optional capabilities (declare or skip explicitly)¶

These are NOT required for v1.0. An adapter may declare it doesn't support them by raising NotImplementedError on the corresponding method — the contract test then skips the case.

Capability	Adapter coverage today
Per-tenant layer overlay	Filesystem ✓, SQLAlchemy[postgres] ✓; SQLAlchemy[sqlite] inherits the i-092 PK debt (overlay publish clobbers base — strict xfail in the kit)
Versioning + immutable releases	Filesystem ✓, SQLAlchemy ✓ (both dialects)
Lockfile install/update flow	Filesystem ✓, SQLAlchemy[postgres] ✓ (Phase 10)
Cross-process cache invalidation (LISTEN/NOTIFY)	SQLAlchemy[postgres] ✓ (Phase 15.1 outbox + `kernel_writes` channel); FS uses in-process events; the sqlite dialect is single-process

Using the SQLAlchemy adapter (s-sqlalchemy-source-production)¶

SqlAlchemySource (dna/adapters/sqlalchemy_/) is ONE adapter over SQLAlchemy Core 2.x async that speaks BOTH SQL dialects and binds to the exact same tables the retired raw adapters owned. Since i-038 the schema is applied by Alembic (dna/adapters/sqlalchemy_/alembic/), against the ONE table model in dna/adapters/sqlalchemy_/schema.py:

dialect	driver	tables	control table
`sqlite+aiosqlite`	aiosqlite	`instances` / `versions` / `bundle_entries` / `layer_instances`	`alembic_version`
`postgresql+asyncpg`	asyncpg	`{schema}.dna_instances` / `dna_versions` / `dna_bundle_entries` / `dna_layer_instances` / `dna_outbox` / `dna_versions_seq`	`{schema}.alembic_version`

The tables are unchanged — the Alembic baseline revision reproduces the retired ladder's final schema exactly (verified against pg_dump --schema-only and SQLAlchemy inspect() on both dialects). The historical control tables (schema_migrations, {schema}.dna_schema_migrations) are retired: a database carrying one is detected at boot, stamped at the baseline revision and the old table dropped, with no DDL replayed (locked by tests/test_alembic_legacy_baseline.py). A database at a partial ladder version is refused loudly — bring it to the ladder head on dna-sdk <= 0.20.0 first.

Install¶

pip install "dna-sdk[sql]"   # sqlalchemy[asyncio] + aiosqlite + asyncpg

Nothing in the default install imports sqlalchemy (guard: tests/test_sqlalchemy_source.py::test_default_import_never_pulls_sqlalchemy).

Wiring¶

from dna.adapters.sqlalchemy_ import SqlAlchemySource
from dna.kernel import Kernel

# SQLite — replaces the retired SqliteSource(db_path=...):
src = SqlAlchemySource("sqlite+aiosqlite:///path/to.db")

# Postgres — replaces the retired PostgresSource(pool, schema=...)
# (a URL instead of an asyncpg pool; same schema kwarg):
src = SqlAlchemySource(
    "postgresql+asyncpg://user:pass@host:5432/db", schema="public",
)

await src.connect()               # runs the dialect's migrations (idempotent)
kernel = Kernel.auto(source=src)  # or kernel.source(src) on an existing kernel

Behavior parity notes¶

Eventbus (pg dialect): every write emits the Phase 15.1 outbox row
dna_versions_seq checkpoint + pg_notify('kernel_writes', …) in the same transaction as the data write. The payload is produced by dna.kernel.eventbus.build_notify_payload — the same wire contract PostgresEventBus subscribes to, so subscribers work unchanged. supports_cross_process_invalidation is True on pg, False on sqlite.
View cache: load_all/load_layer(tenant) are memoized per (scope, tenant) with deep-copy returns — same as raw PG — and invalidated on local writes AND via kernel.on_write (attach_kernel).
Auto-publish: save_instance is the publish point — the doc is visible in load_all immediately; publish() remains available for the explicit draft→publish flow.
Known inherited limit: the sqlite dialect inherits i-092 (instances PK lacks tenant → a tenant overlay publish clobbers the base row) — it binds to the existing schema by design. The pg dialect passes the same case (tenant-aware PK): schema debt, not adapter debt.
Native COUNT (pg dialect): count() aggregates in SQL — only aggregates travel back, never rows (F2 D2, inherited from the raw PG adapter). The sqlite dialect rides query() via the shared helper.
Perf: packages/sdk-py/scripts/bench_sources.py prints save/load_all/query timings per dialect (temp sqlite file always; DATABASE_URL adds the pg row).

Running the suite¶

# Filesystem + the sqlite dialect (always available)
cd packages/sdk-py && uv run pytest tests/test_port_contract.py -v

# Add the postgres dialect (requires running DB)
DATABASE_URL=postgresql://dna:dna@localhost:5432/dna \
  uv run pytest tests/test_port_contract.py -v

Expected result: all green for FS + sqlite. Postgres tests skip when DATABASE_URL unset; otherwise they must be green too.

CI gate: any PR that breaks the contract test for any adapter is blocked from merge — that's the structural change that gives confidence to ship the SDK.

Adding a new adapter¶

Implement WritableSourcePort (mandatory) + KernelAttachable + BundleEntryReadable (capability Protocols).
Add a builder in packages/sdk-py/tests/test_port_contract.py next to _build_fs_source and the _build_sqlalchemy_* builders.
Run uv run pytest tests/test_port_contract.py -v — every test should pass or skip explicitly.
If a test fails, fix the adapter, NOT the test. The test encodes the contract; mutating it amounts to lying about what the adapter supports.

Writing a Source adapter (s-dna-source-conformance-kit)¶

The end-to-end recipe for a NEW Source adapter — the conformance kit is your safety net at every step.

1. Declare the contract in the class statement¶

Python adapters subclass the port Protocol explicitly (chosen pattern — readable, statically checkable, mirrors the TS implements clause):

from dna.kernel.protocols import SourcePort, WritableSourcePort

class MyReadOnlySource(SourcePort): ...
class MySource(WritableSourcePort): ...

Caveat to keep straight: inheriting a Protocol makes isinstance pass nominally and inherits the Protocol's no-op method stubs — it cannot prove behavior. That's the kit's job (below). The one deliberate exception is AsyncSourceAdapter: it's a transparent __getattr__ proxy that mirrors whatever it wraps, so inheriting would both shadow the forwarding and overclaim; its conformance is structural.

If your storage client is synchronous (a blocking SDK like boto3), keep the adapter sync and hand the kernel AsyncSourceAdapter(your_source) — never the raw sync object.

2. Declare `capabilities()` honestly¶

Return a literal SourceCapabilities (see kernel/capabilities.py). The kit's capabilities_declared_honestly case asserts your declaration matches the reflection oracle (derive_capabilities) — a declaration that overclaims or underclaims fails.

One capability is about your STORAGE MODEL rather than a method you implement: api_version_identity. A Kind is identified by (apiVersion, kind), so two workspaces may each declare a Deal under their own namespace. Declare it when your row key carries that identity — (scope, kind, apiVersion, name) — so both instances can exist at once and a caller can ask for one of them; the oracle derives it from load_one accepting api_version. The built-in SQL adapter declares it. The filesystem adapter does NOT: an instance's path is <container>/<name>, and the container comes from the kernel's StorageDescriptor registry, so two Kinds are distinct on disk only insofar as the registry gives them distinct containers. Neither answer is wrong; declaring it is what keeps the difference visible, and the kit's two_kinds_one_name_stay_apart case runs only for adapters that claim it.

3. Pass the boot gate¶

kernel.source(src) validates the CORE surface by name (supports_readers, load_bootstrap_docs, load_all, resolve_ref, load_layer, close) and raises SourceRegistrationError naming what's missing. The capability-mediated members (list_doc_refs, load_one, query, count) may be absent — the kernel serves them via load_all fallbacks and logs a warning; implement them for production workloads. The gate checks NAMES only (runtime_checkable semantics) — passing it does not mean the adapter works.

For test doubles: subclass dna.testing.CoreSourceStub so your fake passes the gate without hand-rolling the core surface.

4. Run the conformance kit — the real safety net¶

The kit ships in the package (dna.testing), not in this repo's tests — external adapter authors run the exact same battery:

import pytest
from dna.testing import source_conformance_suite, FIXTURE_SCOPE, fixture_docs

async def my_factory():
    src = MySource(...)          # fresh instance, isolated env
    async def cleanup():
        await src.close()
    return src, cleanup

CASES = source_conformance_suite(my_factory)

@pytest.mark.asyncio
@pytest.mark.parametrize("case", CASES, ids=lambda c: c.name)
async def test_my_source_conformance(case):
    await case.run()

Rules of the kit:

The factory is called once PER CASE (fresh adapter each time) and owns the environment: temp dirs / schemas, kernel wiring (Kernel.auto(source=src)) if the adapter needs it, and — for READ-ONLY adapters — pre-seeding fixture_docs() under FIXTURE_SCOPE in the native storage. Writable adapters are seeded by the kit through their own save_instance/publish (part of the test).
Cases are capability-aware: they SKIP (CaseNotApplicable) what you don't declare and FAIL what you declare but don't honor.
Non-pytest consumption: await run_source_conformance(my_factory) returns a ConformanceReport (.ok, .failed, .raise_if_failed()).

5. Add your adapter to the in-repo matrix¶

packages/sdk-py/tests/test_source_conformance_kit.py runs the kit over ALL in-repo adapters (FS read-only, FS writable, Composite, AsyncSourceAdapter, SqlAlchemySource × both dialects). A new in-repo adapter adds a factory there; known divergences get an explicit xfail/skip with an Issue id — never a silent green.

Schema migrations (SQL-backed adapters, s-dna-migration-contract)¶

An adapter that owns SQL storage manages its own schema through the shared forward-only runner (dna/adapters/_migrations.py::run_migrations). The contract:

Forward-only, numbered. Migrations are a Mapping[int, payload] keyed by positive integer version, applied in ascending numeric order. Downgrade is not supported — recovery from a bad migration is backup/re-seed, never a down() script.
Append-only. A migration that has ever shipped is frozen: never edit an applied version's payload — append a new version that fixes forward (see SQLite v8 rebuilding the v3 table, or Postgres v9 adding the column v-earlier forgot). The control table records what ran; editing history would desynchronize every existing DB.
Automatic upgrade on boot. SqlAlchemySource.connect() runs pending migrations before serving — deploying new code upgrades the store, no separate migrate step. Booting an up-to-date store applies nothing (idempotent re-boot).
Control table. The built-in SQL adapter uses Alembic's standard alembic_version (in the DNA schema on Postgres, so several DNA schemas in one database track independently). Third-party adapters using the numbered runner below own their own control table.
Older binary vs newer store doesn't crash: unknown recorded versions are left untouched and logged as a warning.
Atomicity is the adapter's. The runner owns ordering/skip/ reporting; the adapter's apply_version callable owns "apply + record" with its dialect's semantics — Postgres wraps each version's statements + control-table insert in ONE transaction; SQLite runs executescript (which self-commits) then records.

Adopting the runner in a third-party adapter¶

Give the runner three async callables bound to your storage and expose the public entrypoint:

from dna.adapters._migrations import run_migrations

MIGRATIONS: dict[int, str] = {1: "CREATE TABLE ...", 2: "ALTER TABLE ..."}

class MySource(WritableSourcePort):
    async def run_schema_migrations(self) -> list[str]:
        """Public entrypoint — also call it from your boot path.

        The conformance kit wants revision IDENTIFIERS as strings; if you
        number your migrations, return ``[str(v) for v in applied]``.
        """
        return await run_migrations(
            MIGRATIONS,
            ensure_control_table=self._ensure_control_table,  # CREATE TABLE IF NOT EXISTS my_control(version, applied_at)
            fetch_applied=self._fetch_applied,                # -> versions already recorded
            apply_version=self._apply_one,                    # apply payload + record version, atomically for YOUR dialect
            dialect="MyStore",
        )

The conformance kit picks the capability up by duck-typing (same pattern as list_scopes): an adapter exposing run_schema_migrations() gets the schema_migrations_idempotent case — after boot, a re-run must return [] (the control table survived in the backing store), and the first run must return list[str] of revision ids. Adapters without SQL storage simply don't implement the method and the case skips.

One writer per schema¶

The migration stream above is owned by this runtime. Point a second implementation at the same Postgres schema and both will try to "complete" the other's history with conflicting DDL. One schema belongs to one writer.