Plugin Config Store

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

PluginConfigRecord

Folded shape that pairs a plugin’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.

---

PluginConfigRecord

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

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

PluginConfigStore Protocol

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

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

---

PluginConfigStore

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

Protocol for persisting per-plugin PluginConfigRecord across sessions.

LocalPluginConfigStore

Substrate-shipped default. SQLite at ~/.cjm/plugin_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 PluginConfigStore implementation in via dependency injection.

---

LocalPluginConfigStore

def LocalPluginConfigStore(
    db_path:Optional=None
):

SQLite-backed default implementation of PluginConfigStore.

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.

---

LocalPluginConfigStore.get

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

Fetch the record for a plugin.

---

LocalPluginConfigStore.set

def set(
    plugin_name:str, # Plugin to write
    record:PluginConfigRecord, # New record (updated_at overwritten with current time)
)->None:

Persist a record. Stamps updated_at to the current time.

---

LocalPluginConfigStore.delete

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

Remove the record for a plugin.

---

LocalPluginConfigStore.list_all

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

Return all stored records keyed by plugin name.

# SG-22 regression: LocalPluginConfigStore satisfies the PluginConfigStore
# 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 = LocalPluginConfigStore(Path(db_path))
    
    # Protocol satisfaction (runtime_checkable enables isinstance)
    assert isinstance(store, PluginConfigStore), \
        "LocalPluginConfigStore should satisfy PluginConfigStore 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 = PluginConfigRecord(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", PluginConfigRecord(config={"model": "tiny"}, enabled=True))
    store.set("gemini", PluginConfigRecord(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 LocalPluginConfigStore round-trip: PASS")
finally:
    try:
        os.unlink(db_path)
    except OSError:
        pass

SG-22 LocalPluginConfigStore round-trip: PASS