feat(taskservice): REST API, liveness protocol, artifact store, MCP contract

Slice 1, PR 3 of 4 — the task service API contract.

- panopticon.taskservice.service.TaskService: deterministic orchestration over the
  repository — tasks, transition enforcement (via the engine), slug, artifacts, and
  ephemeral liveness registrations. Injectable clock + id factory for test determinism.
- panopticon.taskservice.api: FastAPI app factory exposing repos, tasks, transitions
  (with per-responsibility settlement on the wire), slug, artifacts, liveness, and
  discovery, with domain exceptions mapped to HTTP status codes.
- panopticon.core.artifacts + taskservice.artifacts_fs: the artifact-store interface, the
  filesystem adapter, and the shared id->path->MCP-URI resolver (ADR 0003).
- panopticon.taskservice.mcp: the MCP surface contract (resource URI scheme + tool
  specs), defined now; the server is wired when real containers connect (Slice 2).

Adds fastapi + uvicorn (runtime) and httpx (dev, for TestClient).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: tildesrc

docs/BACKLOG.md

@@ -41,6 +41,9 @@ in the ADRs; this file is for the smaller stuff that doesn't have a home there y
   _(Slice 1, P2)_
 - [ ] **Registrations are in-memory** — lost on task-service restart; no reconciliation with
   live containers on reconnect (relates to ADR 0008 failure-handling). _(Slice 1, P2)_
+- [ ] **Slug-addressable artifacts** — once a task has a `slug`, both
+  `tasks/{task_id}/artifacts/{name}` and `tasks/{slug}/artifacts/{name}` should resolve to the
+  same artifact (slug as an alias for the id on the artifact routes). _(Slice 1, P3)_
 
 ## Tracked elsewhere (pointers, do not duplicate)
 

pyproject.toml

@@ -4,13 +4,16 @@ version = "0.0.1"
 description = "Orchestrate multiple coding agents across isolated tasks and configurable workflows."
 requires-python = ">=3.11"
 dependencies = [
+    "fastapi>=0.110",
     "sqlalchemy>=2.0.50",
+    "uvicorn>=0.29",
 ]
 
 [dependency-groups]
 dev = [
     "pytest>=8",
     "mypy>=1.11",
+    "httpx>=0.27",
 ]
 
 [build-system]

src/panopticon/core/artifacts.py

@@ -0,0 +1,55 @@
+"""The artifact-store interface + the shared id→path→URI resolver (ADR 0003).
+
+Freeform per-task files (plan, notes) are file-backed, not in the DB. The same bytes are
+reachable via the filesystem, the dashboard, and MCP; this module owns the single resolver
+that maps ``(task_id, name)`` to a path and an MCP URI so every surface agrees.
+"""
+
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+
+MCP_URI_SCHEME = "panopticon"
+
+
+class ArtifactError(Exception):
+    """Base class for artifact-store failures."""
+
+
+class InvalidArtifactName(ArtifactError):
+    """Raised for an artifact name (or task id) that could escape its directory."""
+
+
+def validate_segment(segment: str) -> None:
+    """Reject names/ids that contain path separators, dot-segments, or are empty."""
+    if (
+        not segment
+        or "/" in segment
+        or "\\" in segment
+        or segment in (".", "..")
+        or segment.startswith(".")
+    ):
+        raise InvalidArtifactName(f"invalid artifact segment: {segment!r}")
+
+
+def mcp_uri(task_id: str, name: str) -> str:
+    """The canonical MCP resource URI for an artifact (the shared resolver)."""
+    validate_segment(task_id)
+    validate_segment(name)
+    return f"{MCP_URI_SCHEME}://tasks/{task_id}/artifacts/{name}"
+
+
+class ArtifactStore(ABC):
+    """Read/write per-task artifact files."""
+
+    @abstractmethod
+    def put(self, task_id: str, name: str, content: bytes) -> None:
+        """Create or overwrite an artifact."""
+
+    @abstractmethod
+    def get(self, task_id: str, name: str) -> bytes | None:
+        """Return artifact bytes, or ``None`` if it does not exist."""
+
+    @abstractmethod
+    def list(self, task_id: str) -> list[str]:
+        """Return the names of a task's artifacts (empty if none)."""

src/panopticon/taskservice/api.py

@@ -0,0 +1,252 @@
+"""The task service REST API (FastAPI).
+
+The dashboard, the runner, and in-container skills are clients of this API. (Agents also
+reach artifacts/tools over MCP — see :mod:`panopticon.taskservice.mcp` — but the walking
+skeleton uses REST.) ``create_app`` builds an app around an injected
+:class:`~panopticon.taskservice.service.TaskService`, so tests can wire a deterministic one.
+"""
+
+from __future__ import annotations
+
+from fastapi import FastAPI, HTTPException, Request, Response
+from fastapi.responses import JSONResponse
+from pydantic import BaseModel, ConfigDict
+
+from panopticon.core.artifacts import ArtifactError
+from panopticon.core.models import Actor, Repo, Status
+from panopticon.core.store import AlreadyExists, NotFound, StoreError
+from panopticon.core.workflow import IllegalTransition, InvalidWorkflow, ResponsibilitiesNotMet
+from panopticon.taskservice.service import TaskService, UnknownWorkflow
+
+# -- wire schemas -------------------------------------------------------------------
+
+
+# ``*Out`` models read straight off the domain objects (`model_validate`): their fields match
+# the domain attribute names, so `from_attributes=True` does the conversion — incl. nested
+# Task -> History -> Responsibility — with no hand-written copying.
+
+
+class ResponsibilityOut(BaseModel):
+    model_config = ConfigDict(from_attributes=True)
+
+    key: str
+    description: str
+    status: Status
+    comment: str | None = None
+
+
+class HistoryOut(BaseModel):
+    model_config = ConfigDict(from_attributes=True)
+
+    at: str
+    from_state: str | None
+    to_state: str
+    trigger: str | None = None
+    note: str | None = None
+    responsibilities: list[ResponsibilityOut] = []
+
+
+class TaskOut(BaseModel):
+    model_config = ConfigDict(from_attributes=True)
+
+    id: str
+    repo_id: str
+    workflow: str
+    state: str
+    turn: Actor
+    slug: str | None
+    history: list[HistoryOut]
+
+
+class RepoIn(BaseModel):
+    id: str
+    name: str
+    git_url: str
+    default_base: str = "main"
+
+
+class RepoOut(BaseModel):
+    model_config = ConfigDict(from_attributes=True)
+
+    id: str
+    name: str
+    git_url: str
+    default_base: str
+
+
+class CreateTaskIn(BaseModel):
+    repo_id: str
+    workflow: str
+
+
+class ResponsibilityIn(BaseModel):
+    key: str
+    status: Status
+    comment: str | None = None
+
+
+class TransitionIn(BaseModel):
+    to_state: str
+    trigger: str | None = None
+    note: str | None = None
+
+
+class SlugIn(BaseModel):
+    slug: str
+
+
+class RegisterIn(BaseModel):
+    container_id: str
+    runner_id: str | None = None
+
+
+class RegistrationOut(BaseModel):
+    model_config = ConfigDict(from_attributes=True)
+
+    id: str
+    task_id: str
+    container_id: str
+    runner_id: str | None
+    registered_at: str
+    last_seen: str
+
+
+def create_app(service: TaskService) -> FastAPI:
+    app = FastAPI(title="panopticon task service", version="0.0.1")
+
+    # -- error mapping: domain exceptions -> HTTP status --------------------------
+
+    @app.exception_handler(NotFound)
+    async def _not_found(_: Request, exc: NotFound) -> JSONResponse:
+        return JSONResponse(status_code=404, content={"detail": str(exc)})
+
+    @app.exception_handler(AlreadyExists)
+    async def _conflict(_: Request, exc: AlreadyExists) -> JSONResponse:
+        return JSONResponse(status_code=409, content={"detail": str(exc)})
+
+    @app.exception_handler(IllegalTransition)
+    async def _illegal(_: Request, exc: IllegalTransition) -> JSONResponse:
+        return JSONResponse(status_code=409, content={"detail": str(exc)})
+
+    @app.exception_handler(ResponsibilitiesNotMet)
+    async def _responsibilities(_: Request, exc: ResponsibilitiesNotMet) -> JSONResponse:
+        return JSONResponse(status_code=409, content={"detail": str(exc)})
+
+    @app.exception_handler(UnknownWorkflow)
+    async def _unknown_wf(_: Request, exc: UnknownWorkflow) -> JSONResponse:
+        return JSONResponse(status_code=400, content={"detail": str(exc)})
+
+    @app.exception_handler(InvalidWorkflow)
+    async def _invalid_wf(_: Request, exc: InvalidWorkflow) -> JSONResponse:
+        return JSONResponse(status_code=400, content={"detail": str(exc)})
+
+    @app.exception_handler(ArtifactError)
+    async def _artifact(_: Request, exc: ArtifactError) -> JSONResponse:
+        return JSONResponse(status_code=400, content={"detail": str(exc)})
+
+    @app.exception_handler(StoreError)
+    async def _store_error(_: Request, exc: StoreError) -> JSONResponse:
+        return JSONResponse(status_code=409, content={"detail": str(exc)})
+
+    # -- health & discovery -------------------------------------------------------
+
+    @app.get("/healthz")
+    async def healthz() -> dict[str, str]:
+        return {"status": "ok"}
+
+    @app.get("/workflows")
+    async def list_workflows() -> list[str]:
+        return service.workflow_names()
+
+    # -- repos --------------------------------------------------------------------
+
+    @app.post("/repos", status_code=201)
+    async def create_repo(body: RepoIn) -> RepoOut:
+        repo = service.create_repo(Repo(**body.model_dump()))
+        return RepoOut.model_validate(repo)
+
+    @app.get("/repos")
+    async def list_repos() -> list[RepoOut]:
+        return [RepoOut.model_validate(r) for r in service.list_repos()]
+
+    @app.get("/repos/{repo_id}")
+    async def get_repo(repo_id: str) -> RepoOut:
+        return RepoOut.model_validate(service.get_repo(repo_id))
+
+    # -- tasks --------------------------------------------------------------------
+
+    @app.post("/tasks", status_code=201)
+    async def create_task(body: CreateTaskIn) -> TaskOut:
+        return TaskOut.model_validate(service.create_task(body.repo_id, body.workflow))
+
+    @app.get("/tasks")
+    async def list_tasks() -> list[TaskOut]:
+        return [TaskOut.model_validate(t) for t in service.list_tasks()]
+
+    @app.get("/tasks/{task_id}")
+    async def get_task(task_id: str) -> TaskOut:
+        return TaskOut.model_validate(service.get_task(task_id))
+
+    @app.post("/tasks/{task_id}/transition")
+    async def transition(task_id: str, body: TransitionIn) -> TaskOut:
+        return TaskOut.model_validate(
+            service.request_transition(
+                task_id, body.to_state, trigger=body.trigger, note=body.note
+            )
+        )
+
+    @app.post("/tasks/{task_id}/responsibilities")
+    async def resolve_responsibility(task_id: str, body: ResponsibilityIn) -> TaskOut:
+        try:
+            task = service.resolve_responsibility(
+                task_id, body.key, status=body.status, comment=body.comment
+            )
+        except ValueError as exc:  # unknown key / PENDING / FAILED without a comment
+            raise HTTPException(status_code=400, detail=str(exc)) from exc
+        return TaskOut.model_validate(task)
+
+    @app.put("/tasks/{task_id}/slug")
+    async def set_slug(task_id: str, body: SlugIn) -> TaskOut:
+        return TaskOut.model_validate(service.set_slug(task_id, body.slug))
+
+    # -- artifacts ----------------------------------------------------------------
+
+    @app.put("/tasks/{task_id}/artifacts/{name}", status_code=204)
+    async def put_artifact(task_id: str, name: str, request: Request) -> Response:
+        service.put_artifact(task_id, name, await request.body())
+        return Response(status_code=204)
+
+    @app.get("/tasks/{task_id}/artifacts")
+    async def list_artifacts(task_id: str) -> list[str]:
+        return service.list_artifacts(task_id)
+
+    @app.get("/tasks/{task_id}/artifacts/{name}")
+    async def get_artifact(task_id: str, name: str) -> Response:
+        content = service.get_artifact(task_id, name)
+        if content is None:
+            raise HTTPException(status_code=404, detail=f"artifact {name!r} not found")
+        return Response(content=content, media_type="application/octet-stream")
+
+    # -- liveness -----------------------------------------------------------------
+
+    @app.post("/tasks/{task_id}/registrations", status_code=201)
+    async def register(task_id: str, body: RegisterIn) -> RegistrationOut:
+        return RegistrationOut.model_validate(
+            service.register(task_id, body.container_id, body.runner_id)
+        )
+
+    @app.get("/tasks/{task_id}/registrations")
+    async def list_registrations(task_id: str) -> list[RegistrationOut]:
+        service.get_task(task_id)  # 404 if the task is unknown
+        return [RegistrationOut.model_validate(r) for r in service.registrations(task_id)]
+
+    @app.post("/registrations/{registration_id}/heartbeat")
+    async def heartbeat(registration_id: str) -> RegistrationOut:
+        return RegistrationOut.model_validate(service.heartbeat(registration_id))
+
+    @app.delete("/registrations/{registration_id}", status_code=204)
+    async def deregister(registration_id: str) -> Response:
+        service.deregister(registration_id)
+        return Response(status_code=204)
+
+    return app

src/panopticon/taskservice/artifacts_fs.py

@@ -0,0 +1,39 @@
+"""Filesystem artifact-store adapter (ADR 0003: local filesystem first).
+
+Layout: ``<root>/tasks/<task_id>/<name>``. The same files are openable in an editor and,
+later, served over MCP using the resolver in :mod:`panopticon.core.artifacts`.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+from panopticon.core.artifacts import ArtifactStore, validate_segment
+
+
+class FilesystemArtifactStore(ArtifactStore):
+    """Store artifacts as plain files under a root directory."""
+
+    def __init__(self, root: str | Path) -> None:
+        self._root = Path(root)
+
+    def _task_dir(self, task_id: str) -> Path:
+        validate_segment(task_id)
+        return self._root / "tasks" / task_id
+
+    def put(self, task_id: str, name: str, content: bytes) -> None:
+        validate_segment(name)
+        task_dir = self._task_dir(task_id)
+        task_dir.mkdir(parents=True, exist_ok=True)
+        (task_dir / name).write_bytes(content)
+
+    def get(self, task_id: str, name: str) -> bytes | None:
+        validate_segment(name)
+        path = self._task_dir(task_id) / name
+        return path.read_bytes() if path.is_file() else None
+
+    def list(self, task_id: str) -> list[str]:
+        task_dir = self._task_dir(task_id)
+        if not task_dir.is_dir():
+            return []
+        return sorted(p.name for p in task_dir.iterdir() if p.is_file())