feat(poc): rewind MCP, skill, and recover-ladder docs
Initial PoC: find_protection, tagged checkpoints, FLR, mutating catalog. Lab 10.9 status enums (0=Initializing, 1=MeetingSLA). Credentials stay in gitignored config.json.
This commit is contained in:
@@ -0,0 +1,73 @@
|
||||
"""Mutating catalog: opt-in list of tools that must be guarded."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from dataclasses import asdict, dataclass
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
|
||||
@dataclass
|
||||
class CatalogEntry:
|
||||
server: str
|
||||
tool: str
|
||||
vm_arg: str
|
||||
notes: str = ""
|
||||
|
||||
def key(self) -> tuple[str, str]:
|
||||
return (self.server.lower(), self.tool.lower())
|
||||
|
||||
def as_dict(self) -> dict[str, str]:
|
||||
return asdict(self)
|
||||
|
||||
|
||||
def entry_from_dict(raw: dict[str, Any]) -> CatalogEntry:
|
||||
server = str(raw.get("server") or "").strip()
|
||||
tool = str(raw.get("tool") or "").strip()
|
||||
vm_arg = str(raw.get("vm_arg") or "").strip()
|
||||
if not server or not tool or not vm_arg:
|
||||
raise ValueError("catalog entry needs server, tool, and vm_arg")
|
||||
return CatalogEntry(
|
||||
server=server,
|
||||
tool=tool,
|
||||
vm_arg=vm_arg,
|
||||
notes=str(raw.get("notes") or ""),
|
||||
)
|
||||
|
||||
|
||||
class MutatingCatalog:
|
||||
def __init__(self, entries: list[CatalogEntry] | None = None, path: Path | None = None):
|
||||
self._entries: dict[tuple[str, str], CatalogEntry] = {}
|
||||
self.path = path
|
||||
for entry in entries or []:
|
||||
self._entries[entry.key()] = entry
|
||||
|
||||
def list(self) -> list[CatalogEntry]:
|
||||
return sorted(self._entries.values(), key=lambda e: e.key())
|
||||
|
||||
def get(self, server: str, tool: str) -> CatalogEntry | None:
|
||||
return self._entries.get((server.lower(), tool.lower()))
|
||||
|
||||
def add(self, entry: CatalogEntry) -> CatalogEntry:
|
||||
self._entries[entry.key()] = entry
|
||||
self.save()
|
||||
return entry
|
||||
|
||||
def save(self) -> None:
|
||||
if self.path is None:
|
||||
return
|
||||
payload = {"mutating_tools": [e.as_dict() for e in self.list()]}
|
||||
self.path.parent.mkdir(parents=True, exist_ok=True)
|
||||
if self.path.suffix == ".json" and self.path.exists():
|
||||
existing = json.loads(self.path.read_text(encoding="utf-8"))
|
||||
if isinstance(existing, dict):
|
||||
existing["mutating_tools"] = payload["mutating_tools"]
|
||||
payload = existing
|
||||
self.path.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8")
|
||||
|
||||
@classmethod
|
||||
def from_config(cls, data: dict[str, Any], path: Path | None = None) -> MutatingCatalog:
|
||||
raw = data.get("mutating_tools") or []
|
||||
entries = [entry_from_dict(item) for item in raw]
|
||||
return cls(entries, path=path)
|
||||
Reference in New Issue
Block a user