Capability Config Store

Persistent storage for per-capability configuration (with enabled flag)

CapabilityConfigRecord

Folded shape that pairs a capability’s config dict with its enabled flag. The pairing (per CR-2’s enable/disable design) lives in one record so the substrate can persist and restore both in a single round-trip.


CapabilityConfigRecord


def CapabilityConfigRecord(
    config:Dict=<factory>, enabled:bool=True, updated_at:float=0.0
)->None:

Persisted state for a capability: config dict + enabled flag.

CapabilityConfigStore Protocol

Interface that any store implementation must satisfy. Substrate ships LocalCapabilityConfigStore as the default cross-session, single-user backend; the future cjm-workflow-state-backed WorkflowCapabilityConfigStore (CR-2) implements the same Protocol so hosts can swap stores without code changes.

runtime_checkable so consumers can isinstance(x, CapabilityConfigStore) for duck-typing, while still relying on static-type-checker enforcement.


CapabilityConfigStore


def CapabilityConfigStore(
    args:VAR_POSITIONAL, kwargs:VAR_KEYWORD
):

Protocol for persisting per-capability CapabilityConfigRecord across sessions.

LocalCapabilityConfigStore

Substrate-shipped default. SQLite at ~/.cjm/capability_configs.db (or a caller-provided path). Cross-session, single-user — suitable for CLI tools and single-user desktop hosts. Workflow-scoped or multi-user hosts plug a different CapabilityConfigStore implementation in via dependency injection.


LocalCapabilityConfigStore


def LocalCapabilityConfigStore(
    db_path:Optional=None
):

SQLite-backed default implementation of CapabilityConfigStore.

The DB is created lazily on first write. Reads against a non-existent DB return empty results rather than raising, so hosts can call .get() on a fresh install without preparing the file first.


LocalCapabilityConfigStore.get


def get(
    capability_name:str, # Capability to look up
)->Optional: # Persisted record or None if absent

Fetch the record for a capability.


LocalCapabilityConfigStore.set


def set(
    capability_name:str, # Capability to write
    record:CapabilityConfigRecord, # New record (updated_at overwritten with current time)
)->None:

Persist a record. Stamps updated_at to the current time.


LocalCapabilityConfigStore.delete


def delete(
    capability_name:str, # Capability to remove
)->bool: # True if a row was deleted

Remove the record for a capability.


LocalCapabilityConfigStore.list_all


def list_all(
    
)->Dict: # capability_name -> record

Return all stored records keyed by capability name.

# SG-22 regression: LocalCapabilityConfigStore satisfies the CapabilityConfigStore
# Protocol and round-trips records cleanly across set/get/delete/list_all.
import tempfile
import os

fd, db_path = tempfile.mkstemp(suffix=".db")
os.close(fd)
os.unlink(db_path)  # Start from a fresh non-existent file

try:
    store = LocalCapabilityConfigStore(Path(db_path))
    
    # Protocol satisfaction (runtime_checkable enables isinstance)
    assert isinstance(store, CapabilityConfigStore), \
        "LocalCapabilityConfigStore should satisfy CapabilityConfigStore Protocol"
    
    # Empty store: missing reads return None and {}
    assert store.get("whisper") is None
    assert store.list_all() == {}
    assert store.delete("whisper") is False
    
    # Round-trip a record
    rec = CapabilityConfigRecord(config={"model": "large-v3"}, enabled=False)
    store.set("whisper", rec)
    
    out = store.get("whisper")
    assert out is not None
    assert out.config == {"model": "large-v3"}
    assert out.enabled is False
    assert out.updated_at > 0
    
    # Overwrite + list_all
    store.set("whisper", CapabilityConfigRecord(config={"model": "tiny"}, enabled=True))
    store.set("gemini", CapabilityConfigRecord(config={"api_key": "x"}, enabled=True))
    all_records = store.list_all()
    assert set(all_records.keys()) == {"whisper", "gemini"}
    assert all_records["whisper"].config == {"model": "tiny"}
    assert all_records["whisper"].enabled is True
    
    # Delete returns True once + False on second call
    assert store.delete("whisper") is True
    assert store.delete("whisper") is False
    assert store.get("whisper") is None
    assert set(store.list_all().keys()) == {"gemini"}
    
    print("SG-22 LocalCapabilityConfigStore round-trip: PASS")
finally:
    try:
        os.unlink(db_path)
    except OSError:
        pass
SG-22 LocalCapabilityConfigStore round-trip: PASS