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.
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.
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.
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
defset( 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 tempfileimport osfd, db_path = tempfile.mkstemp(suffix=".db")os.close(fd)os.unlink(db_path) # Start from a fresh non-existent filetry: store = LocalCapabilityConfigStore(Path(db_path))# Protocol satisfaction (runtime_checkable enables isinstance)assertisinstance(store, CapabilityConfigStore), \"LocalCapabilityConfigStore should satisfy CapabilityConfigStore Protocol"# Empty store: missing reads return None and {}assert store.get("whisper") isNoneassert store.list_all() == {}assert store.delete("whisper") isFalse# Round-trip a record rec = CapabilityConfigRecord(config={"model": "large-v3"}, enabled=False) store.set("whisper", rec) out = store.get("whisper")assert out isnotNoneassert out.config == {"model": "large-v3"}assert out.enabled isFalseassert 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()assertset(all_records.keys()) == {"whisper", "gemini"}assert all_records["whisper"].config == {"model": "tiny"}assert all_records["whisper"].enabled isTrue# Delete returns True once + False on second callassert store.delete("whisper") isTrueassert store.delete("whisper") isFalseassert store.get("whisper") isNoneassertset(store.list_all().keys()) == {"gemini"}print("SG-22 LocalCapabilityConfigStore round-trip: PASS")finally:try: os.unlink(db_path)exceptOSError:pass