From 16815ed6bb03d33d68c7cab2822cb0d32fa1b2f4 Mon Sep 17 00:00:00 2001 From: Mortdecai Date: Mon, 30 Mar 2026 07:30:24 -0400 Subject: [PATCH] =?UTF-8?q?feat:=20project=20management=20module=20?= =?UTF-8?q?=E2=80=94=20create,=20log,=20read,=20list,=20server=20persisten?= =?UTF-8?q?ce?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Sonnet 4.6 --- src/workbench/project.py | 156 +++++++++++++++++++++++++++++++++++++++ tests/conftest.py | 8 ++ tests/test_project.py | 81 ++++++++++++++++++++ 3 files changed, 245 insertions(+) create mode 100644 src/workbench/project.py create mode 100644 tests/conftest.py create mode 100644 tests/test_project.py diff --git a/src/workbench/project.py b/src/workbench/project.py new file mode 100644 index 0000000..1e2cc88 --- /dev/null +++ b/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() diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..80d1893 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,8 @@ +import pytest + +@pytest.fixture +def tmp_workbench(tmp_path): + """Provide a temporary workbench directory.""" + wb = tmp_path / "workbench" + wb.mkdir() + return wb diff --git a/tests/test_project.py b/tests/test_project.py new file mode 100644 index 0000000..7de9d3d --- /dev/null +++ b/tests/test_project.py @@ -0,0 +1,81 @@ +import json +from workbench.project import ( + create_project, project_exists, project_path, + append_log, read_log, list_projects, log_session_event, + write_server_info, read_server_info, clear_server_info, + WORKBENCH_DIR, +) + + +def test_create_project(tmp_workbench): + path = create_project("io102", "Heathkit IO-102", workbench_dir=tmp_workbench) + assert path.exists() + assert (path / "index.html").exists() + assert (path / "session.md").exists() + assert (path / "session.jsonl").exists() + assert (path / "cost-log.jsonl").exists() + assert (path / "state.json").exists() + assert (path / "assets").is_dir() + md = (path / "session.md").read_text() + assert "Heathkit IO-102" in md + + +def test_create_project_idempotent(tmp_workbench): + create_project("io102", "First", workbench_dir=tmp_workbench) + append_log("io102", "test entry", workbench_dir=tmp_workbench) + create_project("io102", "Second", workbench_dir=tmp_workbench) + entries = read_log("io102", workbench_dir=tmp_workbench) + assert len(entries) == 1 + + +def test_project_exists(tmp_workbench): + assert not project_exists("io102", workbench_dir=tmp_workbench) + create_project("io102", "Test", workbench_dir=tmp_workbench) + assert project_exists("io102", workbench_dir=tmp_workbench) + + +def test_append_and_read_log(tmp_workbench): + create_project("io102", "Test", workbench_dir=tmp_workbench) + append_log("io102", "R412 measured 1.05M", data={"ohms": 1050000}, workbench_dir=tmp_workbench) + append_log("io102", "R413 OK", workbench_dir=tmp_workbench) + entries = read_log("io102", tail=10, workbench_dir=tmp_workbench) + assert len(entries) == 2 + assert entries[0]["entry"] == "R412 measured 1.05M" + assert entries[0]["ohms"] == 1050000 + + +def test_read_log_tail(tmp_workbench): + create_project("io102", "Test", workbench_dir=tmp_workbench) + for i in range(30): + append_log("io102", f"Entry {i}", workbench_dir=tmp_workbench) + entries = read_log("io102", tail=5, workbench_dir=tmp_workbench) + assert len(entries) == 5 + assert entries[0]["entry"] == "Entry 25" + + +def test_list_projects(tmp_workbench): + assert list_projects(workbench_dir=tmp_workbench) == [] + create_project("io102", "First", workbench_dir=tmp_workbench) + create_project("psu-rebuild", "Second", workbench_dir=tmp_workbench) + projects = list_projects(workbench_dir=tmp_workbench) + assert [p["name"] for p in projects] == ["io102", "psu-rebuild"] + + +def test_log_session_event(tmp_workbench): + create_project("io102", "Test", workbench_dir=tmp_workbench) + log_session_event("io102", "session_start", workbench_dir=tmp_workbench) + cost_log = (tmp_workbench / "io102" / "cost-log.jsonl").read_text().strip() + entry = json.loads(cost_log.split("\n")[-1]) + assert entry["event"] == "session_start" + + +def test_server_info_write_read_clear(tmp_workbench): + create_project("io102", "Test", workbench_dir=tmp_workbench) + write_server_info("io102", pid=12345, port=8070, workbench_dir=tmp_workbench) + info = read_server_info("io102", workbench_dir=tmp_workbench) + assert info is not None + assert info["pid"] == 12345 + assert info["port"] == 8070 + assert "started" in info + clear_server_info("io102", workbench_dir=tmp_workbench) + assert read_server_info("io102", workbench_dir=tmp_workbench) is None