quarry/quarry_core.py

"""Quarry core: a small task board for agents. JSON file storage, no dependencies.

Tasks have a title, detail, status (open | doing | done), an owner (the agent that created it),
an optional assignee (a known agent id) and comments. Write rules:
  * any agent may create a task and comment on any task;
  * the owner may edit everything on its task (title, detail, status, assignee) and delete it;
  * the assignee may move the task (status) but not edit or reassign it;
  * people (browser, read access) only read.
"""
from __future__ import annotations

import json
import os
import secrets
import threading
from datetime import datetime, timezone
from pathlib import Path

STATUSES = ("open", "doing", "done")
TITLE_MAX, DETAIL_MAX, COMMENT_MAX = 120, 2000, 2000


def now() -> str:
    return datetime.now(timezone.utc).replace(microsecond=0).isoformat()


def _clean(value, limit: int, required: bool) -> str | None:
    if value is None:
        value = ""
    if not isinstance(value, str):
        return None
    value = value.strip()
    if (required and not value) or len(value) > limit:
        return None
    return value


class Board:
    def __init__(self, path: str | os.PathLike, agents: dict[str, str]):
        self.path = Path(path)
        self.agents = dict(agents)  # id -> display name
        self.lock = threading.RLock()

    # storage
    def _load(self) -> dict:
        if not self.path.is_file():
            return {"tasks": []}
        data = json.loads(self.path.read_text(encoding="utf-8"))
        if not isinstance(data, dict) or not isinstance(data.get("tasks"), list):
            return {"tasks": []}
        return data

    def _save(self, data: dict) -> None:
        self.path.parent.mkdir(parents=True, exist_ok=True)
        tmp = self.path.with_suffix(".tmp")
        tmp.write_text(json.dumps(data, indent=2) + "\n", encoding="utf-8")
        os.chmod(tmp, 0o600)
        os.replace(tmp, self.path)

    # views
    def public(self, t: dict) -> dict:
        a = t.get("assignee") or None
        return {
            "id": t["id"], "title": t.get("title", ""), "detail": t.get("detail", ""),
            "status": t.get("status", "open"), "owner_id": t.get("owner_id", ""),
            "owner_name": self.agents.get(t.get("owner_id", ""), t.get("owner_id", "")),
            "assignee": a, "assignee_name": self.agents.get(a, a) if a else None,
            "created_at": t.get("created_at"), "updated_at": t.get("updated_at"),
            "comments": [
                {"author_id": c.get("author_id", ""), "author_name": self.agents.get(c.get("author_id", ""), c.get("author_id", "")),
                 "text": c.get("text", ""), "at": c.get("at")}
                for c in t.get("comments") or [] if isinstance(c, dict)
            ],
        }

    def list(self) -> list[dict]:
        tasks = [self.public(t) for t in self._load()["tasks"] if isinstance(t, dict)]
        tasks.sort(key=lambda t: t.get("updated_at") or "", reverse=True)
        return tasks

    def get(self, task_id: str) -> dict | None:
        t = self._find(self._load(), task_id)
        return self.public(t) if t else None

    @staticmethod
    def _find(data: dict, task_id: str) -> dict | None:
        return next((t for t in data["tasks"] if isinstance(t, dict) and t.get("id") == task_id), None)

    def _assignee(self, value) -> tuple[bool, str | None]:
        if value in (None, ""):
            return True, None
        if isinstance(value, str) and value.strip() in self.agents:
            return True, value.strip()
        return False, None

    # writes (agent is an authenticated agent id)
    def create(self, agent: str, payload: dict) -> tuple[int, dict]:
        title = _clean(payload.get("title"), TITLE_MAX, True)
        detail = _clean(payload.get("detail", ""), DETAIL_MAX, False)
        status = payload.get("status") or "open"
        if title is None:
            return 400, {"error": "invalid_title"}
        if detail is None:
            return 400, {"error": "invalid_detail"}
        if status not in STATUSES:
            return 400, {"error": "invalid_status"}
        ok, assignee = self._assignee(payload.get("assignee"))
        if not ok:
            return 400, {"error": "invalid_assignee", "message": "assignee must be a known agent id or null"}
        ts = now()
        task = {"id": secrets.token_hex(4), "title": title, "detail": detail, "status": status,
                "owner_id": agent, "assignee": assignee, "created_at": ts, "updated_at": ts, "comments": []}
        with self.lock:
            data = self._load()
            data["tasks"].append(task)
            self._save(data)
        return 201, {"task": self.public(task)}

    def update(self, agent: str, task_id: str, payload: dict) -> tuple[int, dict]:
        with self.lock:
            data = self._load()
            t = self._find(data, task_id)
            if t is None:
                return 404, {"error": "not_found"}
            owner = t.get("owner_id") == agent
            fields = set(payload) & {"title", "detail", "status", "assignee"}
            if not fields:
                return 400, {"error": "nothing_to_change"}
            if not owner:
                if t.get("assignee") == agent and fields == {"status"}:
                    pass
                else:
                    return 403, {"error": "not_owner", "message": "only the owner edits a task; the assignee may change its status"}
            if "title" in fields:
                v = _clean(payload["title"], TITLE_MAX, True)
                if v is None:
                    return 400, {"error": "invalid_title"}
                t["title"] = v
            if "detail" in fields:
                v = _clean(payload["detail"], DETAIL_MAX, False)
                if v is None:
                    return 400, {"error": "invalid_detail"}
                t["detail"] = v
            if "status" in fields:
                if payload["status"] not in STATUSES:
                    return 400, {"error": "invalid_status"}
                t["status"] = payload["status"]
            if "assignee" in fields:
                ok, a = self._assignee(payload["assignee"])
                if not ok:
                    return 400, {"error": "invalid_assignee"}
                t["assignee"] = a
            t["updated_at"] = now()
            t["updated_by"] = agent
            self._save(data)
            return 200, {"task": self.public(t)}

    def comment(self, agent: str, task_id: str, payload: dict) -> tuple[int, dict]:
        text = _clean(payload.get("text"), COMMENT_MAX, True)
        if text is None:
            return 400, {"error": "invalid_text"}
        with self.lock:
            data = self._load()
            t = self._find(data, task_id)
            if t is None:
                return 404, {"error": "not_found"}
            c = {"author_id": agent, "text": text, "at": now()}
            t.setdefault("comments", []).append(c)
            t["updated_at"] = c["at"]
            self._save(data)
            return 201, {"task": self.public(t)}

    def delete(self, agent: str, task_id: str) -> tuple[int, dict]:
        with self.lock:
            data = self._load()
            t = self._find(data, task_id)
            if t is None:
                return 404, {"error": "not_found"}
            if t.get("owner_id") != agent:
                return 403, {"error": "not_owner"}
            data["tasks"].remove(t)
            self._save(data)
            return 200, {"deleted": task_id}