feat: project management module — create, log, read, list, server persistence

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

src/workbench/project.py

@@ -0,0 +1,156 @@
+"""Project directory management — create, log, read, list, server persistence."""
+
+from __future__ import annotations
+
+import json
+import os
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Optional
+
+WORKBENCH_DIR = Path.home() / "workbench"
+SCAFFOLD_HTML = Path(__file__).parent / "scaffold.html"
+
+
+def _now_iso() -> str:
+    return datetime.now(timezone.utc).astimezone().isoformat(timespec="seconds")
+
+
+def project_path(name: str, workbench_dir: Path = WORKBENCH_DIR) -> Path:
+    return workbench_dir / name
+
+
+def project_exists(name: str, workbench_dir: Path = WORKBENCH_DIR) -> bool:
+    return project_path(name, workbench_dir).is_dir()
+
+
+def create_project(
+    name: str,
+    title: str,
+    description: str = "",
+    workbench_dir: Path = WORKBENCH_DIR,
+) -> Path:
+    """Create a project directory with scaffold HTML and log files. Idempotent."""
+    pdir = project_path(name, workbench_dir)
+    pdir.mkdir(parents=True, exist_ok=True)
+    (pdir / "assets").mkdir(exist_ok=True)
+
+    if SCAFFOLD_HTML.exists():
+        template = SCAFFOLD_HTML.read_text()
+        html = template.replace("{{TITLE}}", title)
+        html = html.replace("{{DESCRIPTION}}", description)
+        (pdir / "index.html").write_text(html)
+
+    md_path = pdir / "session.md"
+    if not md_path.exists():
+        md_path.write_text(f"# {title} — Session Log\n\nStarted: {_now_iso()}\n\n")
+
+    jsonl_path = pdir / "session.jsonl"
+    if not jsonl_path.exists():
+        jsonl_path.write_text("")
+
+    cost_path = pdir / "cost-log.jsonl"
+    if not cost_path.exists():
+        cost_path.write_text("")
+
+    state_path = pdir / "state.json"
+    if not state_path.exists():
+        state_path.write_text("{}")
+
+    return pdir
+
+
+def append_log(
+    name: str,
+    entry: str,
+    data: Optional[dict] = None,
+    workbench_dir: Path = WORKBENCH_DIR,
+) -> dict:
+    """Append to session.md and session.jsonl."""
+    pdir = project_path(name, workbench_dir)
+    ts = _now_iso()
+
+    with open(pdir / "session.md", "a") as f:
+        f.write(f"\n### {ts}\n{entry}\n")
+
+    obj = {"ts": ts, "entry": entry}
+    if data:
+        obj.update(data)
+    with open(pdir / "session.jsonl", "a") as f:
+        f.write(json.dumps(obj) + "\n")
+
+    return obj
+
+
+def read_log(
+    name: str, tail: int = 20, workbench_dir: Path = WORKBENCH_DIR
+) -> list[dict]:
+    """Read recent log entries from session.jsonl."""
+    jsonl_path = project_path(name, workbench_dir) / "session.jsonl"
+    if not jsonl_path.exists():
+        return []
+
+    lines = jsonl_path.read_text().strip().split("\n")
+    lines = [l for l in lines if l.strip()]
+    recent = lines[-tail:] if len(lines) > tail else lines
+
+    entries = []
+    for line in recent:
+        try:
+            entries.append(json.loads(line))
+        except json.JSONDecodeError:
+            entries.append({"raw": line})
+    return entries
+
+
+def list_projects(workbench_dir: Path = WORKBENCH_DIR) -> list[dict]:
+    """List all project directories."""
+    if not workbench_dir.exists():
+        return []
+    projects = []
+    for d in sorted(workbench_dir.iterdir()):
+        if d.is_dir() and not d.name.startswith("."):
+            projects.append({"name": d.name})
+    return projects
+
+
+def log_session_event(
+    name: str, event: str, workbench_dir: Path = WORKBENCH_DIR, **extra
+) -> None:
+    """Append to cost-log.jsonl."""
+    pdir = project_path(name, workbench_dir)
+    obj = {"ts": _now_iso(), "event": event, "project": name}
+    obj.update(extra)
+    with open(pdir / "cost-log.jsonl", "a") as f:
+        f.write(json.dumps(obj) + "\n")
+
+
+def write_server_info(
+    name: str, pid: int, port: int, workbench_dir: Path = WORKBENCH_DIR
+) -> None:
+    """Write .server.json with running server info."""
+    pdir = project_path(name, workbench_dir)
+    info = {"pid": pid, "port": port, "started": _now_iso()}
+    (pdir / ".server.json").write_text(json.dumps(info))
+
+
+def read_server_info(
+    name: str, workbench_dir: Path = WORKBENCH_DIR
+) -> Optional[dict]:
+    """Read .server.json. Returns None if not found."""
+    server_file = project_path(name, workbench_dir) / ".server.json"
+    if not server_file.exists():
+        return None
+    try:
+        return json.loads(server_file.read_text())
+    except (json.JSONDecodeError, OSError):
+        return None
+
+
+def clear_server_info(
+    name: str, workbench_dir: Path = WORKBENCH_DIR
+) -> None:
+    """Delete .server.json."""
+    server_file = project_path(name, workbench_dir) / ".server.json"
+    if server_file.exists():
+        server_file.unlink()