Instance Storage Patterns Agent Skill

~/.my-tool/
  instances/{instance_id}/
    instance.json         — lifecycle metadata (status, timestamps)
    workspace/            — code, git repo, tests
    logs/                 — pipeline node logs, checkpoint.json
    events/               — events.jsonl, state.json, input-requests/
    sidecar-data/         — sidecar service data (databases, repos)
    output/               — artifacts (branch_name.txt, pr_url.txt)

def get_instance_dir(instance_id: str) -> Path:
    d = get_data_root() / "instances" / instance_id
    d.mkdir(parents=True, exist_ok=True)
    return d

# Subdirectory names used by other modules:
WORKSPACE_DIR = "workspace"
LOGS_DIR = "logs"
EVENTS_DIR = "events"

def get_data_root() -> Path:
    env_dir = os.environ.get("MY_TOOL_DATA_DIR")
    root = Path(env_dir) if env_dir else Path.home() / ".my-tool"
    root.mkdir(parents=True, exist_ok=True)
    return root

class InstanceStore:
    def __init__(self, root: str | Path) -> None:
        self._root = Path(root)
        # Note: locks are never pruned — one Lock (~100 bytes) per instance_id seen.
        self._locks: defaultdict[str, threading.Lock] = defaultdict(threading.Lock)

    def create_instance(self, instance_id, name, params) -> dict:
        with self._locks[instance_id]:
            ...

    def update_instance(self, instance_id, **changes) -> dict | None:
        with self._locks[instance_id]:
            data = self._read_instance(instance_id)
            ...

def _write_instance(self, instance_id, data):
    path = self._instance_path(instance_id)
    path.parent.mkdir(parents=True, exist_ok=True)
    content = json.dumps(data, ensure_ascii=False, default=str)
    fd, tmp_path = tempfile.mkstemp(dir=path.parent, suffix=".tmp")
    try:
        os.write(fd, content.encode("utf-8"))
        os.close(fd)
        Path(tmp_path).replace(path)        # atomic
    except BaseException:
        with contextlib.suppress(OSError):
            os.close(fd)
        Path(tmp_path).unlink(missing_ok=True)
        raise

def empty_instance() -> dict:
    """Each call returns an independent object — no shared mutables."""
    return {"status": "pending", "created_at": None, "metadata": {}}

# storage.py — filesystem-backed instance store
from collections import defaultdict
import contextlib, json, os, tempfile, threading
from pathlib import Path

class InstanceStore:
    def __init__(self, root: Path):
        self._root = root
        self._locks: defaultdict[str, threading.Lock] = defaultdict(threading.Lock)

    def _path(self, instance_id: str) -> Path:
        return self._root / "instances" / instance_id / "instance.json"

    def _read(self, instance_id: str) -> dict | None:
        path = self._path(instance_id)
        if not path.exists():
            return None
        return json.loads(path.read_text())

    def _write(self, instance_id: str, data: dict) -> None:
        path = self._path(instance_id)
        path.parent.mkdir(parents=True, exist_ok=True)
        fd, tmp = tempfile.mkstemp(dir=path.parent, suffix=".tmp")
        try:
            os.write(fd, json.dumps(data, default=str).encode())
            os.close(fd)
            Path(tmp).replace(path)
        except BaseException:
            with contextlib.suppress(OSError):
                os.close(fd)
            Path(tmp).unlink(missing_ok=True)
            raise

    def create(self, instance_id: str, **kwargs) -> dict:
        with self._locks[instance_id]:
            from datetime import UTC, datetime
            now = datetime.now(UTC).isoformat()
            data = {"instance_id": instance_id, "status": "starting",
                    "created_at": now, "updated_at": now, **kwargs}
            self._write(instance_id, data)
            return data

    def update(self, instance_id: str, **changes) -> dict | None:
        with self._locks[instance_id]:
            data = self._read(instance_id)
            if data is None:
                return None
            data.update(changes)
            from datetime import UTC, datetime
            data["updated_at"] = datetime.now(UTC).isoformat()
            self._write(instance_id, data)
            return data

    def list_all(self) -> list[dict]:
        instances_dir = self._root / "instances"
        if not instances_dir.exists():
            return []
        results = []
        for d in instances_dir.iterdir():
            if d.is_dir():
                data = self._read(d.name)
                if data:
                    results.append(data)
        results.sort(key=lambda i: i.get("created_at", ""), reverse=True)
        return results

    def delete(self, instance_id: str) -> bool:
        with self._locks[instance_id]:
            path = self._root / "instances" / instance_id / "instance.json"
            if not path.exists():
                return False
            import shutil
            shutil.rmtree(self._root / "instances" / instance_id)
            return True