ThothII/docs/superpowers/plans/2026-07-11-adapter-foundations.md

Adapter Foundations Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: Route all DWH and vector operations through stable typed contracts while preserving current direct/REST behavior.

Architecture: Define small Python protocols and capability records, implement adapters around existing modules, and centralize construction in one factory. Introduce a discriminated workspace schema with a compatibility translator for current YAML.

Tech Stack: Python 3.11+, typing.Protocol, Pydantic 2, SQLAlchemy, psycopg2, requests, pytest, Typer.

Global Constraints

• Do not change workflow phases or CLI output formats.
• PostgreSQL direct and current Thoth/PostgREST are the only supported DWH transports in this plan.
• Preserve reader/writer vector credential separation.
• All migrations must accept existing database.transport, rest, vector_db, vector_rest, and vector_write_rest fields.
• JSON stdout stays pristine; diagnostics go to stderr unless part of the JSON result.

---

Task 1: Define DWH ports and capabilities

Files:

• Create: harness/tht/ports/__init__.py
• Create: harness/tht/ports/dwh.py
• Test: harness/tests/test_dwh_port_contract.py

Interfaces:

• Produces: DwhCapabilities, DwhAdapter, DwhHealth, DistinctValues, and UnsupportedCapability.

• Consumes: existing catalog models from tht.db.introspect and execution result types from tht.db.execute.

• Step 1: Write the failing protocol-shape test

def test_fake_adapter_satisfies_runtime_protocol():
    adapter = FakeDwhAdapter()
    assert isinstance(adapter, DwhAdapter)
    assert adapter.capabilities.explain is True
    assert adapter.health().ok is True

• Step 2: Run the focused test and confirm failure

Run: cd harness && .venv/bin/pytest tests/test_dwh_port_contract.py -q
Expected: FAIL because tht.ports.dwh does not exist.

• Step 3: Add the minimal public contract

@dataclass(frozen=True)
class DwhCapabilities:
    introspection: bool = True
    explain: bool = True
    sampling: bool = True
    distinct_values: bool = True

@dataclass(frozen=True)
class DistinctValues:
    values: list[object]
    truncated: bool

@runtime_checkable
class DwhAdapter(Protocol):
    @property
    def capabilities(self) -> DwhCapabilities: ...
    def health(self) -> DwhHealth: ...
    def introspect(self) -> PhysicalSchema: ...
    def run_query(self, sql: str, *, limit: int) -> ExecResult: ...
    def explain(self, sql: str) -> PlanSummary: ...
    def sample_column(self, table: str, column: str, *, limit: int) -> list[object]: ...
    def distinct_values(self, table: str, column: str) -> DistinctValues: ...

• Step 4: Run contract test and type-oriented import smoke test

Run: cd harness && .venv/bin/pytest tests/test_dwh_port_contract.py -q
Expected: PASS.

• Step 5: Commit

git add harness/tht/ports harness/tests/test_dwh_port_contract.py
git commit -m "refactor(dwh): define adapter contract"

Task 2: Wrap direct and REST DWH implementations

Files:

• Create: harness/tht/adapters/dwh/__init__.py
• Create: harness/tht/adapters/dwh/postgres.py
• Create: harness/tht/adapters/dwh/thoth_rest.py
• Test: harness/tests/test_dwh_adapters.py
• Test: harness/tests/test_dwh_port_contract.py
• Test: harness/tests/l0/test_db_sampling.py
• Modify: harness/tht/ports/__init__.py
• Modify: harness/tht/ports/dwh.py
• Modify: harness/tht/execute/__init__.py
• Modify: harness/tht/db/execute.py
• Modify: harness/tht/db/sampling.py
• Modify: harness/tht/rest/execute.py
• Modify: docs/superpowers/plans/2026-07-11-adapter-foundations.md

Interfaces:

• Consumes: DwhAdapter from Task 1; existing DatabaseConfig, RestConfig, catalog, sampling, execute, and explain functions.

• Produces: PostgresDwhAdapter(config) and ThothRestDwhAdapter(database, rest).

• Step 1: Add parametrized contract tests for both wrappers

@pytest.mark.parametrize("factory", [postgres_factory, rest_factory])
def test_adapter_rejects_write_sql(factory):
    with pytest.raises(ExecutionError):
        factory().run_query("delete from fact_sales", limit=10)

• Step 2: Verify failure

Run: cd harness && .venv/bin/pytest tests/test_dwh_adapters.py -q
Expected: FAIL because the adapter classes are absent.

• Step 3: Implement thin wrappers, without duplicating transport logic

class PostgresDwhAdapter:
    capabilities = DwhCapabilities()
    def __init__(self, config: DatabaseConfig):
        self._config = config
        self._engine = make_engine(config)
    def run_query(self, sql: str, *, limit: int) -> ExecResult:
        return run_query(self._engine, sql, limit=limit)

Implement the analogous REST wrapper by delegating to tht.rest.*; translate transport-specific errors only at the adapter boundary. Both wrappers delegate frequency-ranked, distinct sampling to the paired implementations in tht.db.sampling. Query and sampling limits must be runtime-positive integers (booleans and floats are rejected), and distinct_values reports any cap through DistinctValues.truncated.

• Step 4: Run adapter, read-only, sampling, and REST tests

Run: cd harness && .venv/bin/pytest tests/test_dwh_adapters.py tests/test_readonly_guard.py tests/test_rest_client.py tests/l0/test_db_sampling.py -q
Expected: PASS; L0 may deselect when Docker is unavailable.

• Step 5: Commit

git add docs/superpowers/plans/2026-07-11-adapter-foundations.md \
  harness/tht/ports harness/tht/adapters/dwh harness/tht/execute/__init__.py \
  harness/tht/db/execute.py harness/tht/db/sampling.py harness/tht/rest/execute.py \
  harness/tests/test_dwh_port_contract.py harness/tests/test_dwh_adapters.py \
  harness/tests/l0/test_db_sampling.py
git commit -m "refactor(dwh): adapt direct and REST transports"

Task 3: Define vector port and wrappers

Files:

• Create: harness/tht/ports/vector.py
• Create: harness/tht/adapters/vector/__init__.py
• Create: harness/tht/adapters/vector/thoth_http.py
• Create: harness/tht/adapters/vector/legacy_direct.py
• Test: harness/tests/test_vector_port_contract.py
• Modify: harness/tht/vectorstore/reader.py

Interfaces:

• Produces: VectorStore, VectorCapabilities, VectorHealth, VectorRecord, VectorWriteRecord, VectorHit, ThothHttpVectorStore.

• Preserves: current VectorRestClient, DirectSearcher, and RestSearcher behavior behind wrappers.

• Step 1: Write read/write capability and dual-credential tests

def test_http_store_reports_reader_without_writer():
    store = ThothHttpVectorStore(reader=reader, writer=None)
    assert store.capabilities.search is True
    assert store.capabilities.upsert is False
    with pytest.raises(VectorWriteUnavailable):
        store.upsert("memory", [])

• Step 2: Verify failure

Run: cd harness && .venv/bin/pytest tests/test_vector_port_contract.py -q
Expected: FAIL because the vector port is absent.

• Step 3: Implement the vector contract and wrappers

@runtime_checkable
class VectorStore(Protocol):
    @property
    def capabilities(self) -> VectorCapabilities: ...
    def health(self) -> VectorHealth: ...
    def search(self, collections: list[str], embedding: list[float], *, limit: int,
               kinds: list[str] | None = None) -> list[VectorHit]: ...
    def existing_hashes(self, collection: str, kinds: list[str]) -> dict[str, str]: ...
    def upsert(self, collection: str, records: list[VectorWriteRecord]) -> int: ...

VectorWriteRecord is the transport-neutral write envelope: it contains the canonical VectorRecord, a precomputed embedding, and a content hash. Adapters must preserve VectorRecord.metadata unchanged, including semantic keys named embedding or content_hash.

• Step 4: Run vector regression tests

Run: cd harness && .venv/bin/pytest tests/test_vector_port_contract.py tests/test_vector_dual_key.py tests/test_search_similar_kinds.py tests/test_memory_save_one.py tests/test_solved_question.py -q
Expected: PASS.

• Step 5: Commit

git add harness/tht/ports/vector.py harness/tht/adapters/vector harness/tht/vectorstore/reader.py harness/tests/test_vector_port_contract.py
git commit -m "refactor(vector): define store contract"

Task 4: Introduce discriminated resource configuration with legacy translation

Files:

• Create: harness/tht/config_compat.py
• Modify: harness/tht/config.py
• Modify: harness/workspaces/tht.example.yaml
• Test: harness/tests/test_config_resources.py
• Test: harness/tests/test_config_legacy_compat.py

Interfaces:

• Produces: DwhResourceConfig, VectorResourceConfig, WorkspaceRoots and translate_legacy_config(raw).

• Consumes: existing YAML environment expansion and ConfigError behavior.

• Step 1: Write new-schema and legacy-equivalence tests

def test_legacy_rest_workspace_equals_new_resource_schema(tmp_path):
    old = load_config(write_old_workspace(tmp_path))
    new = load_config(write_new_workspace(tmp_path))
    assert old.dwh.model_dump() == new.dwh.model_dump()

• Step 2: Verify failure

Run: cd harness && .venv/bin/pytest tests/test_config_resources.py tests/test_config_legacy_compat.py -q
Expected: FAIL because dwh and discriminated vector resources are absent.

• Step 3: Add discriminated models and an isolated translator

class PostgresDwhConfig(BaseModel):
    type: Literal["postgres_direct"]
    connection: DatabaseConfig

class ThothRestDwhConfig(BaseModel):
    type: Literal["thoth_rest"]
    database: DatabaseIdentityConfig
    endpoint: RestConfig

DwhResourceConfig = Annotated[
    PostgresDwhConfig | ThothRestDwhConfig,
    Field(discriminator="type"),
]

Translate legacy keys before Pydantic validation and emit one deprecation warning to stderr, never stdout.

• Step 4: Run all config and workspace tests

Run: cd harness && .venv/bin/pytest tests/test_config_resources.py tests/test_config_legacy_compat.py tests/test_workspace.py tests/test_vector_dual_key.py -q
Expected: PASS.

• Step 5: Commit

git add harness/tht/config.py harness/tht/config_compat.py harness/workspaces/tht.example.yaml harness/tests/test_config_resources.py harness/tests/test_config_legacy_compat.py
git commit -m "feat(config): add typed resource schema"

Task 5: Centralize construction and migrate command call sites

Files:

• Create: harness/tht/adapters/factory.py
• Modify: harness/tht/cli/db_cmd.py
• Modify: harness/tht/cli/schema_cmd.py
• Modify: harness/tht/cli/sql_cmd.py
• Modify: harness/tht/cli/lsh_cmd.py
• Modify: harness/tht/cli/search_cmd.py
• Modify: harness/tht/cli/vector_cmd.py
• Modify: harness/tht/cli/memory_cmd.py
• Test: harness/tests/test_adapter_factory.py

Interfaces:

• Produces: build_dwh(cfg: Config) -> DwhAdapter and build_vector_store(cfg: Config, *, require_write: bool = False) -> VectorStore.
• Consumes: resource configs from Task 4 and wrappers from Tasks 2-3.

Correction: DwhAdapter.distinct_values(table, column, *, limit) requires an explicit positive limit, and direct DWH construction injects cfg.execution.statement_timeout_ms. Targeted vector writes consume the factory-returned VectorStore and pass VectorWriteRecord objects to upsert.

Transitional exception: build_vector_loader remains solely for bulk collection sync (vector init/rebuild/index flows). It may still construct the legacy table-scoped writer directly until docs/superpowers/plans/2026-07-11-local-pgvector-profile.md migrates the local pgvector/vector schema and bulk-sync path. Interactive and targeted writes (memory save-one and solved-question indexing) are not covered by this exception and must continue through build_vector_store(..., require_write=True) and the public vector port.

• Step 1: Write exact factory selection and missing-writer tests

def test_factory_selects_http_vector_and_requires_writer(config):
    assert isinstance(build_vector_store(config), ThothHttpVectorStore)
    with pytest.raises(ConfigError, match="writer"):
        build_vector_store(config, require_write=True)

• Step 2: Verify failure

Run: cd harness && .venv/bin/pytest tests/test_adapter_factory.py -q
Expected: FAIL because the factory does not exist.

• Step 3: Implement factory and replace per-command branching

def build_dwh(cfg: Config) -> DwhAdapter:
    match cfg.dwh.type:
        case "postgres_direct": return PostgresDwhAdapter(cfg.dwh.connection)
        case "thoth_rest": return ThothRestDwhAdapter(cfg.dwh.database, cfg.dwh.endpoint)
        case other: raise ConfigError(f"Adapter DWH non supportato: {other}")

Delete transport checks from migrated commands; keep CLI wording and exit codes stable.

• Step 4: Run focused command suites, full harness suite, and ruff

Run: cd harness && .venv/bin/pytest tests/test_adapter_factory.py tests/test_schema_introspect_guard.py tests/test_sql_preview_json.py tests/test_search_pack.py tests/integration/test_gate_cli_signatures.py -q
Expected: PASS.
Run: cd harness && .venv/bin/pytest -q
Expected: all non-L2 tests PASS.
Run: cd harness && .venv/bin/ruff check tht tests/test_adapter_factory.py tests/test_dwh_port_contract.py tests/test_dwh_adapters.py tests/test_vector_port_contract.py tests/test_config_resources.py tests/test_config_legacy_compat.py
Expected: no errors.

• Step 5: Commit

git add harness/tht harness/tests harness/workspaces/tht.example.yaml
git commit -m "refactor(core): route integrations through adapter factory"