Add state-hub v0.1 — local-first state service for the Custodian

Implements the first live layer of the Custodian cognitive infrastructure: PostgreSQL schema, FastAPI REST API, FastMCP stdio server, and Observable Framework telemetry dashboard. - state-hub/: full stack (docker-compose, FastAPI, Alembic, MCP server, dashboard) - 5 DB tables: topics, workstreams, tasks, decisions, progress_events - 11 MCP tools + 5 resources registered in .mcp.json - Observable dashboard: Overview, Workstreams, Decisions, Progress pages - CLAUDE.md: session protocol (get_state_summary / add_progress_event ritual) - ~/.claude/CLAUDE.md: global cross-project reference to the hub - scripts/pull_image.py: WSL2 TLS-resilient Docker image downloader Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-02-24 17:47:49 +01:00
commit 0ea2788943
48 changed files with 8567 additions and 0 deletions
--- a/.env.example
+++ b/.env.example
@@ -0,0 +1,13 @@
+# Copy to .env and fill in values before running
+POSTGRES_DB=custodian
+POSTGRES_USER=custodian
+POSTGRES_PASSWORD=changeme
+
+DATABASE_URL=postgresql+asyncpg://custodian:changeme@127.0.0.1:5432/custodian
+
+# pgAdmin (optional, only used with --profile tools)
+PGADMIN_EMAIL=admin@local.dev
+PGADMIN_PASSWORD=admin
+
+# API
+API_BASE=http://127.0.0.1:8000
--- a/35
+++ b/35
@@ -0,0 +1,35 @@
+.PHONY: install db db-tools migrate seed api dashboard check start clean
+
+COMPOSE = docker compose -f infra/docker-compose.yml --env-file .env
+
+install:
+	uv sync
+
+db:
+	$(COMPOSE) up -d postgres
+
+db-tools:
+	$(COMPOSE) --profile tools up -d
+
+migrate:
+	uv run alembic upgrade head
+
+seed:
+	uv run python scripts/seed.py
+
+api:
+	uv run uvicorn api.main:app --reload --host 127.0.0.1 --port 8000
+
+dashboard:
+	cd dashboard && npm run dev
+
+check:
+	curl -sf http://127.0.0.1:8000/state/health | python3 -m json.tool
+
+start: db
+	sleep 3
+	$(MAKE) migrate
+	$(MAKE) api
+
+clean:
+	$(COMPOSE) down -v
--- a/alembic.ini
+++ b/alembic.ini
@@ -0,0 +1,39 @@
+[alembic]
+script_location = migrations
+prepend_sys_path = .
+version_path_separator = os
+sqlalchemy.url = postgresql+psycopg2://custodian:changeme@127.0.0.1:5432/custodian
+
+[loggers]
+keys = root,sqlalchemy,alembic
+
+[handlers]
+keys = console
+
+[formatters]
+keys = generic
+
+[logger_root]
+level = WARN
+handlers = console
+qualname =
+
+[logger_sqlalchemy]
+level = WARN
+handlers =
+qualname = sqlalchemy.engine
+
+[logger_alembic]
+level = INFO
+handlers =
+qualname = alembic
+
+[handler_console]
+class = StreamHandler
+args = (sys.stderr,)
+level = NOTSET
+formatter = generic
+
+[formatter_generic]
+format = %(levelname)-5.5s [%(name)s] %(message)s
+datefmt = %H:%M:%S
--- a/api/config.py
+++ b/api/config.py
@@ -0,0 +1,16 @@
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+class Settings(BaseSettings):
+    model_config = SettingsConfigDict(
+        env_file=".env",
+        env_file_encoding="utf-8",
+        extra="ignore",
+    )
+
+    database_url: str = "postgresql+asyncpg://custodian:changeme@127.0.0.1:5432/custodian"
+    api_base: str = "http://127.0.0.1:8000"
+    debug: bool = False
+
+
+settings = Settings()
--- a/api/database.py
+++ b/api/database.py
@@ -0,0 +1,24 @@
+from collections.abc import AsyncGenerator
+
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
+
+from api.config import settings
+
+engine = create_async_engine(
+    settings.database_url,
+    echo=settings.debug,
+    pool_pre_ping=True,
+    pool_size=10,
+    max_overflow=20,
+)
+
+async_session_factory = async_sessionmaker(
+    engine,
+    class_=AsyncSession,
+    expire_on_commit=False,
+)
+
+
+async def get_session() -> AsyncGenerator[AsyncSession, None]:
+    async with async_session_factory() as session:
+        yield session
--- a/api/main.py
+++ b/api/main.py
@@ -0,0 +1,32 @@
+from contextlib import asynccontextmanager
+
+from fastapi import FastAPI
+
+from api.database import engine
+from api.routers import decisions, progress, state, tasks, topics, workstreams
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    yield
+    await engine.dispose()
+
+
+app = FastAPI(
+    title="Custodian State Hub",
+    description="Local-first state API for the Custodian agent system.",
+    version="0.1.0",
+    lifespan=lifespan,
+)
+
+app.include_router(topics.router)
+app.include_router(workstreams.router)
+app.include_router(tasks.router)
+app.include_router(decisions.router)
+app.include_router(progress.router)
+app.include_router(state.router)
+
+
+@app.get("/", include_in_schema=False)
+async def root():
+    return {"service": "state-hub", "docs": "/docs"}
--- a/api/models/init.py
+++ b/api/models/init.py
@@ -0,0 +1,15 @@
+from api.models.base import Base
+from api.models.topic import Topic, TopicStatus, Domain
+from api.models.workstream import Workstream, WorkstreamStatus
+from api.models.task import Task, TaskStatus, TaskPriority
+from api.models.decision import Decision, DecisionType, DecisionStatus
+from api.models.progress_event import ProgressEvent
+
+__all__ = [
+    "Base",
+    "Topic", "TopicStatus", "Domain",
+    "Workstream", "WorkstreamStatus",
+    "Task", "TaskStatus", "TaskPriority",
+    "Decision", "DecisionType", "DecisionStatus",
+    "ProgressEvent",
+]
--- a/api/models/base.py
+++ b/api/models/base.py
@@ -0,0 +1,26 @@
+import uuid
+from datetime import datetime
+
+from sqlalchemy import DateTime, func
+from sqlalchemy.dialects.postgresql import UUID
+from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
+
+
+class Base(DeclarativeBase):
+    pass
+
+
+class TimestampMixin:
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now(), nullable=False
+    )
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True),
+        server_default=func.now(),
+        onupdate=func.now(),
+        nullable=False,
+    )
+
+
+def new_uuid() -> uuid.UUID:
+    return uuid.uuid4()
--- a/api/models/decision.py
+++ b/api/models/decision.py
@@ -0,0 +1,63 @@
+import enum
+import uuid
+from datetime import datetime
+
+from sqlalchemy import CheckConstraint, DateTime, Enum, ForeignKey, String, Text
+from sqlalchemy.dialects.postgresql import UUID
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+
+from api.models.base import Base, TimestampMixin, new_uuid
+
+
+class DecisionType(str, enum.Enum):
+    made = "made"
+    pending = "pending"
+
+
+class DecisionStatus(str, enum.Enum):
+    open = "open"
+    resolved = "resolved"
+    escalated = "escalated"
+    superseded = "superseded"
+
+
+class Decision(Base, TimestampMixin):
+    __tablename__ = "decisions"
+    __table_args__ = (
+        CheckConstraint(
+            "topic_id IS NOT NULL OR workstream_id IS NOT NULL",
+            name="ck_decisions_topic_or_workstream",
+        ),
+    )
+
+    id: Mapped[uuid.UUID] = mapped_column(
+        UUID(as_uuid=True), primary_key=True, default=new_uuid
+    )
+    topic_id: Mapped[uuid.UUID | None] = mapped_column(
+        UUID(as_uuid=True), ForeignKey("topics.id", ondelete="RESTRICT"), nullable=True, index=True
+    )
+    workstream_id: Mapped[uuid.UUID | None] = mapped_column(
+        UUID(as_uuid=True), ForeignKey("workstreams.id", ondelete="RESTRICT"), nullable=True, index=True
+    )
+    title: Mapped[str] = mapped_column(String(255), nullable=False)
+    description: Mapped[str | None] = mapped_column(Text, nullable=True)
+    decision_type: Mapped[DecisionType] = mapped_column(
+        Enum(DecisionType), nullable=False, default=DecisionType.pending
+    )
+    status: Mapped[DecisionStatus] = mapped_column(
+        Enum(DecisionStatus), nullable=False, default=DecisionStatus.open
+    )
+    rationale: Mapped[str | None] = mapped_column(Text, nullable=True)
+    decided_by: Mapped[str | None] = mapped_column(String(100), nullable=True)
+    decided_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
+    deadline: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
+    escalation_note: Mapped[str | None] = mapped_column(Text, nullable=True)
+    superseded_by: Mapped[uuid.UUID | None] = mapped_column(
+        UUID(as_uuid=True), ForeignKey("decisions.id", ondelete="SET NULL"), nullable=True
+    )
+
+    topic: Mapped["Topic | None"] = relationship("Topic", back_populates="decisions")  # noqa: F821
+    workstream: Mapped["Workstream | None"] = relationship("Workstream", back_populates="decisions")  # noqa: F821
+    progress_events: Mapped[list["ProgressEvent"]] = relationship(  # noqa: F821
+        "ProgressEvent", back_populates="decision", lazy="selectin"
+    )
--- a/api/models/progress_event.py
+++ b/api/models/progress_event.py
@@ -0,0 +1,43 @@
+import uuid
+from datetime import datetime
+
+from sqlalchemy import DateTime, ForeignKey, String, Text, func
+from sqlalchemy.dialects.postgresql import JSONB, UUID
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+
+from api.models.base import Base, new_uuid
+
+
+class ProgressEvent(Base):
+    """Append-only event log. No updated_at. No DELETE endpoint (constitution §5)."""
+
+    __tablename__ = "progress_events"
+
+    id: Mapped[uuid.UUID] = mapped_column(
+        UUID(as_uuid=True), primary_key=True, default=new_uuid
+    )
+    topic_id: Mapped[uuid.UUID | None] = mapped_column(
+        UUID(as_uuid=True), ForeignKey("topics.id", ondelete="RESTRICT"), nullable=True, index=True
+    )
+    workstream_id: Mapped[uuid.UUID | None] = mapped_column(
+        UUID(as_uuid=True), ForeignKey("workstreams.id", ondelete="RESTRICT"), nullable=True, index=True
+    )
+    task_id: Mapped[uuid.UUID | None] = mapped_column(
+        UUID(as_uuid=True), ForeignKey("tasks.id", ondelete="RESTRICT"), nullable=True, index=True
+    )
+    decision_id: Mapped[uuid.UUID | None] = mapped_column(
+        UUID(as_uuid=True), ForeignKey("decisions.id", ondelete="RESTRICT"), nullable=True, index=True
+    )
+    event_type: Mapped[str] = mapped_column(String(50), nullable=False, index=True)
+    summary: Mapped[str] = mapped_column(Text, nullable=False)
+    detail: Mapped[dict | None] = mapped_column(JSONB, nullable=True)
+    author: Mapped[str | None] = mapped_column(String(100), nullable=True)
+    session_id: Mapped[str | None] = mapped_column(String(100), nullable=True)
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now(), nullable=False, index=True
+    )
+
+    topic: Mapped["Topic | None"] = relationship("Topic", back_populates="progress_events")  # noqa: F821
+    workstream: Mapped["Workstream | None"] = relationship("Workstream", back_populates="progress_events")  # noqa: F821
+    task: Mapped["Task | None"] = relationship("Task", back_populates="progress_events")  # noqa: F821
+    decision: Mapped["Decision | None"] = relationship("Decision", back_populates="progress_events")  # noqa: F821
--- a/api/models/task.py
+++ b/api/models/task.py
@@ -0,0 +1,57 @@
+import enum
+import uuid
+from datetime import date
+
+from sqlalchemy import Date, Enum, ForeignKey, String, Text
+from sqlalchemy.dialects.postgresql import UUID
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+
+from api.models.base import Base, TimestampMixin, new_uuid
+
+
+class TaskStatus(str, enum.Enum):
+    todo = "todo"
+    in_progress = "in_progress"
+    blocked = "blocked"
+    done = "done"
+    cancelled = "cancelled"
+
+
+class TaskPriority(str, enum.Enum):
+    low = "low"
+    medium = "medium"
+    high = "high"
+    critical = "critical"
+
+
+class Task(Base, TimestampMixin):
+    __tablename__ = "tasks"
+
+    id: Mapped[uuid.UUID] = mapped_column(
+        UUID(as_uuid=True), primary_key=True, default=new_uuid
+    )
+    workstream_id: Mapped[uuid.UUID] = mapped_column(
+        UUID(as_uuid=True), ForeignKey("workstreams.id", ondelete="RESTRICT"), nullable=False, index=True
+    )
+    title: Mapped[str] = mapped_column(String(255), nullable=False)
+    description: Mapped[str | None] = mapped_column(Text, nullable=True)
+    status: Mapped[TaskStatus] = mapped_column(
+        Enum(TaskStatus), nullable=False, default=TaskStatus.todo
+    )
+    priority: Mapped[TaskPriority] = mapped_column(
+        Enum(TaskPriority), nullable=False, default=TaskPriority.medium
+    )
+    assignee: Mapped[str | None] = mapped_column(String(100), nullable=True)
+    due_date: Mapped[date | None] = mapped_column(Date, nullable=True)
+    blocking_reason: Mapped[str | None] = mapped_column(Text, nullable=True)
+    parent_task_id: Mapped[uuid.UUID | None] = mapped_column(
+        UUID(as_uuid=True), ForeignKey("tasks.id", ondelete="SET NULL"), nullable=True
+    )
+
+    workstream: Mapped["Workstream"] = relationship("Workstream", back_populates="tasks")  # noqa: F821
+    subtasks: Mapped[list["Task"]] = relationship(
+        "Task", foreign_keys=[parent_task_id], lazy="selectin"
+    )
+    progress_events: Mapped[list["ProgressEvent"]] = relationship(  # noqa: F821
+        "ProgressEvent", back_populates="task", lazy="selectin"
+    )
--- a/api/models/topic.py
+++ b/api/models/topic.py
@@ -0,0 +1,48 @@
+import enum
+import uuid
+
+from sqlalchemy import Enum, String, Text
+from sqlalchemy.dialects.postgresql import UUID
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+
+from api.models.base import Base, TimestampMixin, new_uuid
+
+
+class TopicStatus(str, enum.Enum):
+    active = "active"
+    paused = "paused"
+    archived = "archived"
+
+
+class Domain(str, enum.Enum):
+    custodian = "custodian"
+    railiance = "railiance"
+    markitect = "markitect"
+    coulomb_social = "coulomb_social"
+    personhood = "personhood"
+    foerster_capabilities = "foerster_capabilities"
+
+
+class Topic(Base, TimestampMixin):
+    __tablename__ = "topics"
+
+    id: Mapped[uuid.UUID] = mapped_column(
+        UUID(as_uuid=True), primary_key=True, default=new_uuid
+    )
+    slug: Mapped[str] = mapped_column(String(100), unique=True, nullable=False, index=True)
+    title: Mapped[str] = mapped_column(String(255), nullable=False)
+    description: Mapped[str | None] = mapped_column(Text, nullable=True)
+    domain: Mapped[Domain] = mapped_column(Enum(Domain), nullable=False)
+    status: Mapped[TopicStatus] = mapped_column(
+        Enum(TopicStatus), nullable=False, default=TopicStatus.active
+    )
+
+    workstreams: Mapped[list["Workstream"]] = relationship(  # noqa: F821
+        "Workstream", back_populates="topic", lazy="selectin"
+    )
+    decisions: Mapped[list["Decision"]] = relationship(  # noqa: F821
+        "Decision", back_populates="topic", lazy="selectin"
+    )
+    progress_events: Mapped[list["ProgressEvent"]] = relationship(  # noqa: F821
+        "ProgressEvent", back_populates="topic", lazy="selectin"
+    )
--- a/api/models/workstream.py
+++ b/api/models/workstream.py
@@ -0,0 +1,46 @@
+import enum
+import uuid
+from datetime import date
+
+from sqlalchemy import Date, Enum, ForeignKey, String, Text
+from sqlalchemy.dialects.postgresql import UUID
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+
+from api.models.base import Base, TimestampMixin, new_uuid
+
+
+class WorkstreamStatus(str, enum.Enum):
+    active = "active"
+    blocked = "blocked"
+    completed = "completed"
+    archived = "archived"
+
+
+class Workstream(Base, TimestampMixin):
+    __tablename__ = "workstreams"
+
+    id: Mapped[uuid.UUID] = mapped_column(
+        UUID(as_uuid=True), primary_key=True, default=new_uuid
+    )
+    topic_id: Mapped[uuid.UUID] = mapped_column(
+        UUID(as_uuid=True), ForeignKey("topics.id", ondelete="RESTRICT"), nullable=False, index=True
+    )
+    slug: Mapped[str] = mapped_column(String(100), unique=True, nullable=False, index=True)
+    title: Mapped[str] = mapped_column(String(255), nullable=False)
+    description: Mapped[str | None] = mapped_column(Text, nullable=True)
+    status: Mapped[WorkstreamStatus] = mapped_column(
+        Enum(WorkstreamStatus), nullable=False, default=WorkstreamStatus.active
+    )
+    owner: Mapped[str | None] = mapped_column(String(100), nullable=True)
+    due_date: Mapped[date | None] = mapped_column(Date, nullable=True)
+
+    topic: Mapped["Topic"] = relationship("Topic", back_populates="workstreams")  # noqa: F821
+    tasks: Mapped[list["Task"]] = relationship(  # noqa: F821
+        "Task", back_populates="workstream", lazy="selectin"
+    )
+    decisions: Mapped[list["Decision"]] = relationship(  # noqa: F821
+        "Decision", back_populates="workstream", lazy="selectin"
+    )
+    progress_events: Mapped[list["ProgressEvent"]] = relationship(  # noqa: F821
+        "ProgressEvent", back_populates="workstream", lazy="selectin"
+    )
--- a/api/routers/init.py
+++ b/api/routers/init.py
--- a/api/routers/decisions.py
+++ b/api/routers/decisions.py
@@ -0,0 +1,110 @@
+import uuid
+from datetime import datetime, timezone
+
+from fastapi import APIRouter, Depends, HTTPException, status
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from api.database import get_session
+from api.models.decision import Decision, DecisionStatus, DecisionType
+from api.schemas.decision import DecisionCreate, DecisionRead, DecisionUpdate
+
+router = APIRouter(prefix="/decisions", tags=["decisions"])
+
+_FINANCIAL_LEGAL_KEYWORDS = (
+    "financ", "legal", "payment", "purchas", "contract", "commit",
+    "obligation", "external representation",
+)
+
+
+def _needs_escalation(body: DecisionCreate) -> str | None:
+    if body.decision_type != DecisionType.pending:
+        return None
+    text = f"{body.title} {body.description or ''}".lower()
+    for kw in _FINANCIAL_LEGAL_KEYWORDS:
+        if kw in text:
+            return (
+                "Auto-escalated per constitution §4: this pending decision touches "
+                "financial or legal territory and requires explicit human approval before action."
+            )
+    return None
+
+
+@router.get("/", response_model=list[DecisionRead])
+async def list_decisions(
+    topic_id: uuid.UUID | None = None,
+    workstream_id: uuid.UUID | None = None,
+    status: DecisionStatus | None = None,
+    decision_type: DecisionType | None = None,
+    session: AsyncSession = Depends(get_session),
+) -> list[Decision]:
+    q = select(Decision)
+    if topic_id:
+        q = q.where(Decision.topic_id == topic_id)
+    if workstream_id:
+        q = q.where(Decision.workstream_id == workstream_id)
+    if status:
+        q = q.where(Decision.status == status)
+    if decision_type:
+        q = q.where(Decision.decision_type == decision_type)
+    q = q.order_by(Decision.created_at)
+    result = await session.execute(q)
+    return list(result.scalars().all())
+
+
+@router.post("/", response_model=DecisionRead, status_code=status.HTTP_201_CREATED)
+async def create_decision(
+    body: DecisionCreate,
+    session: AsyncSession = Depends(get_session),
+) -> Decision:
+    data = body.model_dump()
+    note = _needs_escalation(body)
+    if note:
+        data["escalation_note"] = note
+        data["status"] = DecisionStatus.escalated
+    decision = Decision(**data)
+    session.add(decision)
+    await session.commit()
+    await session.refresh(decision)
+    return decision
+
+
+@router.get("/{decision_id}", response_model=DecisionRead)
+async def get_decision(
+    decision_id: uuid.UUID,
+    session: AsyncSession = Depends(get_session),
+) -> Decision:
+    decision = await session.get(Decision, decision_id)
+    if decision is None:
+        raise HTTPException(status_code=404, detail="Decision not found")
+    return decision
+
+
+@router.patch("/{decision_id}", response_model=DecisionRead)
+async def update_decision(
+    decision_id: uuid.UUID,
+    body: DecisionUpdate,
+    session: AsyncSession = Depends(get_session),
+) -> Decision:
+    decision = await session.get(Decision, decision_id)
+    if decision is None:
+        raise HTTPException(status_code=404, detail="Decision not found")
+    for field, value in body.model_dump(exclude_unset=True).items():
+        setattr(decision, field, value)
+    await session.commit()
+    await session.refresh(decision)
+    return decision
+
+
+@router.delete("/{decision_id}", response_model=DecisionRead)
+async def supersede_decision(
+    decision_id: uuid.UUID,
+    session: AsyncSession = Depends(get_session),
+) -> Decision:
+    decision = await session.get(Decision, decision_id)
+    if decision is None:
+        raise HTTPException(status_code=404, detail="Decision not found")
+    decision.status = DecisionStatus.superseded
+    await session.commit()
+    await session.refresh(decision)
+    return decision
--- a/api/routers/progress.py
+++ b/api/routers/progress.py
@@ -0,0 +1,50 @@
+import uuid
+from datetime import datetime
+
+from fastapi import APIRouter, Depends, status
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from api.database import get_session
+from api.models.progress_event import ProgressEvent
+from api.schemas.progress_event import ProgressEventCreate, ProgressEventRead
+
+router = APIRouter(prefix="/progress", tags=["progress"])
+
+
+@router.get("/", response_model=list[ProgressEventRead])
+async def list_progress(
+    topic_id: uuid.UUID | None = None,
+    workstream_id: uuid.UUID | None = None,
+    task_id: uuid.UUID | None = None,
+    event_type: str | None = None,
+    since: datetime | None = None,
+    limit: int = 100,
+    session: AsyncSession = Depends(get_session),
+) -> list[ProgressEvent]:
+    q = select(ProgressEvent)
+    if topic_id:
+        q = q.where(ProgressEvent.topic_id == topic_id)
+    if workstream_id:
+        q = q.where(ProgressEvent.workstream_id == workstream_id)
+    if task_id:
+        q = q.where(ProgressEvent.task_id == task_id)
+    if event_type:
+        q = q.where(ProgressEvent.event_type == event_type)
+    if since:
+        q = q.where(ProgressEvent.created_at >= since)
+    q = q.order_by(ProgressEvent.created_at.desc()).limit(limit)
+    result = await session.execute(q)
+    return list(result.scalars().all())
+
+
+@router.post("/", response_model=ProgressEventRead, status_code=status.HTTP_201_CREATED)
+async def append_progress(
+    body: ProgressEventCreate,
+    session: AsyncSession = Depends(get_session),
+) -> ProgressEvent:
+    event = ProgressEvent(**body.model_dump())
+    session.add(event)
+    await session.commit()
+    await session.refresh(event)
+    return event
--- a/api/routers/state.py
+++ b/api/routers/state.py
@@ -0,0 +1,132 @@
+from datetime import datetime, timezone
+
+from fastapi import APIRouter, Depends
+from fastapi.responses import JSONResponse
+from sqlalchemy import func, select, text
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from api.database import get_session, engine
+from api.models.decision import Decision, DecisionStatus, DecisionType
+from api.models.progress_event import ProgressEvent
+from api.models.task import Task, TaskStatus
+from api.models.topic import Topic, TopicStatus
+from api.models.workstream import Workstream, WorkstreamStatus
+from api.schemas.decision import DecisionRead
+from api.schemas.progress_event import ProgressEventRead
+from api.schemas.state import (
+    DecisionTotals,
+    StateSummary,
+    TaskTotals,
+    Totals,
+    TopicTotals,
+    WorkstreamTotals,
+)
+from api.schemas.task import TaskRead
+from api.schemas.topic import TopicWithWorkstreams
+from api.schemas.workstream import WorkstreamRead
+
+router = APIRouter(prefix="/state", tags=["state"])
+
+
+@router.get("/summary", response_model=StateSummary)
+async def get_summary(session: AsyncSession = Depends(get_session)) -> StateSummary:
+    # Run all queries sequentially on one session.
+    # AsyncSession does not support concurrent operations (no gather on same session).
+
+    topics_rows = await session.execute(
+        select(Topic).where(Topic.status != TopicStatus.archived).order_by(Topic.created_at)
+    )
+    topics = list(topics_rows.scalars().all())
+
+    blocking_rows = await session.execute(
+        select(Decision)
+        .where(Decision.decision_type == DecisionType.pending)
+        .where(Decision.status.in_([DecisionStatus.open, DecisionStatus.escalated]))
+        .order_by(Decision.deadline.asc().nullslast(), Decision.created_at)
+    )
+    blocking = list(blocking_rows.scalars().all())
+
+    blocked_rows = await session.execute(
+        select(Task).where(Task.status == TaskStatus.blocked).order_by(Task.created_at)
+    )
+    blocked = list(blocked_rows.scalars().all())
+
+    recent_rows = await session.execute(
+        select(ProgressEvent).order_by(ProgressEvent.created_at.desc()).limit(20)
+    )
+    recent = list(recent_rows.scalars().all())
+
+    open_ws_rows = await session.execute(
+        select(Workstream)
+        .where(Workstream.status.in_([WorkstreamStatus.active, WorkstreamStatus.blocked]))
+        .order_by(Workstream.due_date.asc().nullslast(), Workstream.created_at)
+    )
+    open_ws = list(open_ws_rows.scalars().all())
+
+    # Totals — one GROUP BY per table
+    topic_counts = {r[0]: r[1] for r in await session.execute(
+        select(Topic.status, func.count()).group_by(Topic.status)
+    )}
+    ws_counts = {r[0]: r[1] for r in await session.execute(
+        select(Workstream.status, func.count()).group_by(Workstream.status)
+    )}
+    task_counts = {r[0]: r[1] for r in await session.execute(
+        select(Task.status, func.count()).group_by(Task.status)
+    )}
+    dec_counts = {r[0]: r[1] for r in await session.execute(
+        select(Decision.status, func.count()).group_by(Decision.status)
+    )}
+
+    totals = Totals(
+        topics=TopicTotals(
+            active=topic_counts.get(TopicStatus.active, 0),
+            paused=topic_counts.get(TopicStatus.paused, 0),
+            archived=topic_counts.get(TopicStatus.archived, 0),
+            total=sum(topic_counts.values()),
+        ),
+        workstreams=WorkstreamTotals(
+            active=ws_counts.get(WorkstreamStatus.active, 0),
+            blocked=ws_counts.get(WorkstreamStatus.blocked, 0),
+            completed=ws_counts.get(WorkstreamStatus.completed, 0),
+            archived=ws_counts.get(WorkstreamStatus.archived, 0),
+            total=sum(ws_counts.values()),
+        ),
+        tasks=TaskTotals(
+            todo=task_counts.get(TaskStatus.todo, 0),
+            in_progress=task_counts.get(TaskStatus.in_progress, 0),
+            blocked=task_counts.get(TaskStatus.blocked, 0),
+            done=task_counts.get(TaskStatus.done, 0),
+            cancelled=task_counts.get(TaskStatus.cancelled, 0),
+            total=sum(task_counts.values()),
+        ),
+        decisions=DecisionTotals(
+            open=dec_counts.get(DecisionStatus.open, 0),
+            resolved=dec_counts.get(DecisionStatus.resolved, 0),
+            escalated=dec_counts.get(DecisionStatus.escalated, 0),
+            superseded=dec_counts.get(DecisionStatus.superseded, 0),
+            total=sum(dec_counts.values()),
+        ),
+    )
+
+    return StateSummary(
+        generated_at=datetime.now(tz=timezone.utc),
+        totals=totals,
+        topics=[TopicWithWorkstreams.model_validate(t) for t in topics],
+        blocking_decisions=[DecisionRead.model_validate(d) for d in blocking],
+        blocked_tasks=[TaskRead.model_validate(t) for t in blocked],
+        recent_progress=[ProgressEventRead.model_validate(e) for e in recent],
+        open_workstreams=[WorkstreamRead.model_validate(w) for w in open_ws],
+    )
+
+
+@router.get("/health")
+async def health_check() -> dict:
+    try:
+        async with engine.connect() as conn:
+            await conn.execute(text("SELECT 1"))
+        return {"status": "ok", "db": "connected"}
+    except Exception as exc:
+        return JSONResponse(
+            status_code=503,
+            content={"status": "error", "db": str(exc)},
+        )
--- a/api/routers/tasks.py
+++ b/api/routers/tasks.py
@@ -0,0 +1,83 @@
+import uuid
+
+from fastapi import APIRouter, Depends, HTTPException, status
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from api.database import get_session
+from api.models.task import Task, TaskStatus
+from api.schemas.task import TaskCreate, TaskRead, TaskUpdate
+
+router = APIRouter(prefix="/tasks", tags=["tasks"])
+
+
+@router.get("/", response_model=list[TaskRead])
+async def list_tasks(
+    workstream_id: uuid.UUID | None = None,
+    status: TaskStatus | None = None,
+    assignee: str | None = None,
+    session: AsyncSession = Depends(get_session),
+) -> list[Task]:
+    q = select(Task)
+    if workstream_id:
+        q = q.where(Task.workstream_id == workstream_id)
+    if status:
+        q = q.where(Task.status == status)
+    if assignee:
+        q = q.where(Task.assignee == assignee)
+    q = q.order_by(Task.created_at)
+    result = await session.execute(q)
+    return list(result.scalars().all())
+
+
+@router.post("/", response_model=TaskRead, status_code=status.HTTP_201_CREATED)
+async def create_task(
+    body: TaskCreate,
+    session: AsyncSession = Depends(get_session),
+) -> Task:
+    task = Task(**body.model_dump())
+    session.add(task)
+    await session.commit()
+    await session.refresh(task)
+    return task
+
+
+@router.get("/{task_id}", response_model=TaskRead)
+async def get_task(
+    task_id: uuid.UUID,
+    session: AsyncSession = Depends(get_session),
+) -> Task:
+    task = await session.get(Task, task_id)
+    if task is None:
+        raise HTTPException(status_code=404, detail="Task not found")
+    return task
+
+
+@router.patch("/{task_id}", response_model=TaskRead)
+async def update_task(
+    task_id: uuid.UUID,
+    body: TaskUpdate,
+    session: AsyncSession = Depends(get_session),
+) -> Task:
+    task = await session.get(Task, task_id)
+    if task is None:
+        raise HTTPException(status_code=404, detail="Task not found")
+    for field, value in body.model_dump(exclude_unset=True).items():
+        setattr(task, field, value)
+    await session.commit()
+    await session.refresh(task)
+    return task
+
+
+@router.delete("/{task_id}", response_model=TaskRead)
+async def cancel_task(
+    task_id: uuid.UUID,
+    session: AsyncSession = Depends(get_session),
+) -> Task:
+    task = await session.get(Task, task_id)
+    if task is None:
+        raise HTTPException(status_code=404, detail="Task not found")
+    task.status = TaskStatus.cancelled
+    await session.commit()
+    await session.refresh(task)
+    return task
--- a/api/routers/topics.py
+++ b/api/routers/topics.py
@@ -0,0 +1,77 @@
+import uuid
+
+from fastapi import APIRouter, Depends, HTTPException, status
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from api.database import get_session
+from api.models.topic import Topic, TopicStatus
+from api.schemas.topic import TopicCreate, TopicRead, TopicUpdate, TopicWithWorkstreams
+
+router = APIRouter(prefix="/topics", tags=["topics"])
+
+
+@router.get("/", response_model=list[TopicRead])
+async def list_topics(
+    status: TopicStatus | None = None,
+    session: AsyncSession = Depends(get_session),
+) -> list[Topic]:
+    q = select(Topic)
+    if status:
+        q = q.where(Topic.status == status)
+    q = q.order_by(Topic.created_at)
+    result = await session.execute(q)
+    return list(result.scalars().all())
+
+
+@router.post("/", response_model=TopicRead, status_code=status.HTTP_201_CREATED)
+async def create_topic(
+    body: TopicCreate,
+    session: AsyncSession = Depends(get_session),
+) -> Topic:
+    topic = Topic(**body.model_dump())
+    session.add(topic)
+    await session.commit()
+    await session.refresh(topic)
+    return topic
+
+
+@router.get("/{topic_id}", response_model=TopicWithWorkstreams)
+async def get_topic(
+    topic_id: uuid.UUID,
+    session: AsyncSession = Depends(get_session),
+) -> Topic:
+    topic = await session.get(Topic, topic_id)
+    if topic is None:
+        raise HTTPException(status_code=404, detail="Topic not found")
+    return topic
+
+
+@router.patch("/{topic_id}", response_model=TopicRead)
+async def update_topic(
+    topic_id: uuid.UUID,
+    body: TopicUpdate,
+    session: AsyncSession = Depends(get_session),
+) -> Topic:
+    topic = await session.get(Topic, topic_id)
+    if topic is None:
+        raise HTTPException(status_code=404, detail="Topic not found")
+    for field, value in body.model_dump(exclude_unset=True).items():
+        setattr(topic, field, value)
+    await session.commit()
+    await session.refresh(topic)
+    return topic
+
+
+@router.delete("/{topic_id}", response_model=TopicRead)
+async def archive_topic(
+    topic_id: uuid.UUID,
+    session: AsyncSession = Depends(get_session),
+) -> Topic:
+    topic = await session.get(Topic, topic_id)
+    if topic is None:
+        raise HTTPException(status_code=404, detail="Topic not found")
+    topic.status = TopicStatus.archived
+    await session.commit()
+    await session.refresh(topic)
+    return topic
--- a/api/routers/workstreams.py
+++ b/api/routers/workstreams.py
@@ -0,0 +1,80 @@
+import uuid
+
+from fastapi import APIRouter, Depends, HTTPException, status
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from api.database import get_session
+from api.models.workstream import Workstream, WorkstreamStatus
+from api.schemas.workstream import WorkstreamCreate, WorkstreamRead, WorkstreamUpdate
+
+router = APIRouter(prefix="/workstreams", tags=["workstreams"])
+
+
+@router.get("/", response_model=list[WorkstreamRead])
+async def list_workstreams(
+    topic_id: uuid.UUID | None = None,
+    status: WorkstreamStatus | None = None,
+    session: AsyncSession = Depends(get_session),
+) -> list[Workstream]:
+    q = select(Workstream)
+    if topic_id:
+        q = q.where(Workstream.topic_id == topic_id)
+    if status:
+        q = q.where(Workstream.status == status)
+    q = q.order_by(Workstream.created_at)
+    result = await session.execute(q)
+    return list(result.scalars().all())
+
+
+@router.post("/", response_model=WorkstreamRead, status_code=status.HTTP_201_CREATED)
+async def create_workstream(
+    body: WorkstreamCreate,
+    session: AsyncSession = Depends(get_session),
+) -> Workstream:
+    ws = Workstream(**body.model_dump())
+    session.add(ws)
+    await session.commit()
+    await session.refresh(ws)
+    return ws
+
+
+@router.get("/{workstream_id}", response_model=WorkstreamRead)
+async def get_workstream(
+    workstream_id: uuid.UUID,
+    session: AsyncSession = Depends(get_session),
+) -> Workstream:
+    ws = await session.get(Workstream, workstream_id)
+    if ws is None:
+        raise HTTPException(status_code=404, detail="Workstream not found")
+    return ws
+
+
+@router.patch("/{workstream_id}", response_model=WorkstreamRead)
+async def update_workstream(
+    workstream_id: uuid.UUID,
+    body: WorkstreamUpdate,
+    session: AsyncSession = Depends(get_session),
+) -> Workstream:
+    ws = await session.get(Workstream, workstream_id)
+    if ws is None:
+        raise HTTPException(status_code=404, detail="Workstream not found")
+    for field, value in body.model_dump(exclude_unset=True).items():
+        setattr(ws, field, value)
+    await session.commit()
+    await session.refresh(ws)
+    return ws
+
+
+@router.delete("/{workstream_id}", response_model=WorkstreamRead)
+async def archive_workstream(
+    workstream_id: uuid.UUID,
+    session: AsyncSession = Depends(get_session),
+) -> Workstream:
+    ws = await session.get(Workstream, workstream_id)
+    if ws is None:
+        raise HTTPException(status_code=404, detail="Workstream not found")
+    ws.status = WorkstreamStatus.archived
+    await session.commit()
+    await session.refresh(ws)
+    return ws
--- a/api/schemas/init.py
+++ b/api/schemas/init.py
@@ -0,0 +1,15 @@
+from api.schemas.topic import TopicCreate, TopicUpdate, TopicRead, TopicWithWorkstreams
+from api.schemas.workstream import WorkstreamCreate, WorkstreamUpdate, WorkstreamRead
+from api.schemas.task import TaskCreate, TaskUpdate, TaskRead
+from api.schemas.decision import DecisionCreate, DecisionUpdate, DecisionRead
+from api.schemas.progress_event import ProgressEventCreate, ProgressEventRead
+from api.schemas.state import StateSummary, Totals, TopicTotals, WorkstreamTotals, TaskTotals, DecisionTotals
+
+__all__ = [
+    "TopicCreate", "TopicUpdate", "TopicRead", "TopicWithWorkstreams",
+    "WorkstreamCreate", "WorkstreamUpdate", "WorkstreamRead",
+    "TaskCreate", "TaskUpdate", "TaskRead",
+    "DecisionCreate", "DecisionUpdate", "DecisionRead",
+    "ProgressEventCreate", "ProgressEventRead",
+    "StateSummary", "Totals", "TopicTotals", "WorkstreamTotals", "TaskTotals", "DecisionTotals",
+]
--- a/api/schemas/decision.py
+++ b/api/schemas/decision.py
@@ -0,0 +1,58 @@
+import uuid
+from datetime import datetime
+
+from pydantic import BaseModel, ConfigDict, model_validator
+
+from api.models.decision import DecisionStatus, DecisionType
+
+
+class DecisionCreate(BaseModel):
+    topic_id: uuid.UUID | None = None
+    workstream_id: uuid.UUID | None = None
+    title: str
+    description: str | None = None
+    decision_type: DecisionType = DecisionType.pending
+    status: DecisionStatus = DecisionStatus.open
+    rationale: str | None = None
+    decided_by: str | None = None
+    decided_at: datetime | None = None
+    deadline: datetime | None = None
+    escalation_note: str | None = None
+
+    @model_validator(mode="after")
+    def topic_or_workstream_required(self) -> "DecisionCreate":
+        if self.topic_id is None and self.workstream_id is None:
+            raise ValueError("At least one of topic_id or workstream_id must be set")
+        return self
+
+
+class DecisionUpdate(BaseModel):
+    title: str | None = None
+    description: str | None = None
+    decision_type: DecisionType | None = None
+    status: DecisionStatus | None = None
+    rationale: str | None = None
+    decided_by: str | None = None
+    decided_at: datetime | None = None
+    deadline: datetime | None = None
+    escalation_note: str | None = None
+    superseded_by: uuid.UUID | None = None
+
+
+class DecisionRead(BaseModel):
+    model_config = ConfigDict(from_attributes=True)
+    id: uuid.UUID
+    topic_id: uuid.UUID | None = None
+    workstream_id: uuid.UUID | None = None
+    title: str
+    description: str | None = None
+    decision_type: DecisionType
+    status: DecisionStatus
+    rationale: str | None = None
+    decided_by: str | None = None
+    decided_at: datetime | None = None
+    deadline: datetime | None = None
+    escalation_note: str | None = None
+    superseded_by: uuid.UUID | None = None
+    created_at: datetime
+    updated_at: datetime
--- a/api/schemas/progress_event.py
+++ b/api/schemas/progress_event.py
@@ -0,0 +1,32 @@
+import uuid
+from datetime import datetime
+from typing import Any
+
+from pydantic import BaseModel, ConfigDict
+
+
+class ProgressEventCreate(BaseModel):
+    topic_id: uuid.UUID | None = None
+    workstream_id: uuid.UUID | None = None
+    task_id: uuid.UUID | None = None
+    decision_id: uuid.UUID | None = None
+    event_type: str
+    summary: str
+    detail: dict[str, Any] | None = None
+    author: str | None = None
+    session_id: str | None = None
+
+
+class ProgressEventRead(BaseModel):
+    model_config = ConfigDict(from_attributes=True)
+    id: uuid.UUID
+    topic_id: uuid.UUID | None = None
+    workstream_id: uuid.UUID | None = None
+    task_id: uuid.UUID | None = None
+    decision_id: uuid.UUID | None = None
+    event_type: str
+    summary: str
+    detail: dict[str, Any] | None = None
+    author: str | None = None
+    session_id: str | None = None
+    created_at: datetime
--- a/api/schemas/state.py
+++ b/api/schemas/state.py
@@ -0,0 +1,58 @@
+from datetime import datetime
+
+from pydantic import BaseModel
+
+from api.schemas.decision import DecisionRead
+from api.schemas.progress_event import ProgressEventRead
+from api.schemas.task import TaskRead
+from api.schemas.topic import TopicWithWorkstreams
+from api.schemas.workstream import WorkstreamRead
+
+
+class TopicTotals(BaseModel):
+    active: int = 0
+    paused: int = 0
+    archived: int = 0
+    total: int = 0
+
+
+class WorkstreamTotals(BaseModel):
+    active: int = 0
+    blocked: int = 0
+    completed: int = 0
+    archived: int = 0
+    total: int = 0
+
+
+class TaskTotals(BaseModel):
+    todo: int = 0
+    in_progress: int = 0
+    blocked: int = 0
+    done: int = 0
+    cancelled: int = 0
+    total: int = 0
+
+
+class DecisionTotals(BaseModel):
+    open: int = 0
+    resolved: int = 0
+    escalated: int = 0
+    superseded: int = 0
+    total: int = 0
+
+
+class Totals(BaseModel):
+    topics: TopicTotals
+    workstreams: WorkstreamTotals
+    tasks: TaskTotals
+    decisions: DecisionTotals
+
+
+class StateSummary(BaseModel):
+    generated_at: datetime
+    totals: Totals
+    topics: list[TopicWithWorkstreams]
+    blocking_decisions: list[DecisionRead]
+    blocked_tasks: list[TaskRead]
+    recent_progress: list[ProgressEventRead]
+    open_workstreams: list[WorkstreamRead]
--- a/api/schemas/task.py
+++ b/api/schemas/task.py
@@ -0,0 +1,51 @@
+import uuid
+from datetime import date, datetime
+
+from pydantic import BaseModel, ConfigDict, model_validator
+
+from api.models.task import TaskPriority, TaskStatus
+
+
+class TaskCreate(BaseModel):
+    workstream_id: uuid.UUID
+    title: str
+    description: str | None = None
+    status: TaskStatus = TaskStatus.todo
+    priority: TaskPriority = TaskPriority.medium
+    assignee: str | None = None
+    due_date: date | None = None
+    blocking_reason: str | None = None
+    parent_task_id: uuid.UUID | None = None
+
+
+class TaskUpdate(BaseModel):
+    title: str | None = None
+    description: str | None = None
+    status: TaskStatus | None = None
+    priority: TaskPriority | None = None
+    assignee: str | None = None
+    due_date: date | None = None
+    blocking_reason: str | None = None
+    parent_task_id: uuid.UUID | None = None
+
+    @model_validator(mode="after")
+    def blocking_reason_required_when_blocked(self) -> "TaskUpdate":
+        if self.status == TaskStatus.blocked and not self.blocking_reason:
+            raise ValueError("blocking_reason is required when status is blocked")
+        return self
+
+
+class TaskRead(BaseModel):
+    model_config = ConfigDict(from_attributes=True)
+    id: uuid.UUID
+    workstream_id: uuid.UUID
+    title: str
+    description: str | None = None
+    status: TaskStatus
+    priority: TaskPriority
+    assignee: str | None = None
+    due_date: date | None = None
+    blocking_reason: str | None = None
+    parent_task_id: uuid.UUID | None = None
+    created_at: datetime
+    updated_at: datetime
--- a/api/schemas/topic.py
+++ b/api/schemas/topic.py
@@ -0,0 +1,47 @@
+import uuid
+from datetime import datetime
+
+from pydantic import BaseModel, ConfigDict
+
+from api.models.topic import Domain, TopicStatus
+
+
+class TopicCreate(BaseModel):
+    slug: str
+    title: str
+    description: str | None = None
+    domain: Domain
+    status: TopicStatus = TopicStatus.active
+
+
+class TopicUpdate(BaseModel):
+    title: str | None = None
+    description: str | None = None
+    domain: Domain | None = None
+    status: TopicStatus | None = None
+
+
+class WorkstreamStub(BaseModel):
+    model_config = ConfigDict(from_attributes=True)
+    id: uuid.UUID
+    slug: str
+    title: str
+    status: str
+    owner: str | None = None
+    due_date: datetime | None = None
+
+
+class TopicRead(BaseModel):
+    model_config = ConfigDict(from_attributes=True)
+    id: uuid.UUID
+    slug: str
+    title: str
+    description: str | None = None
+    domain: Domain
+    status: TopicStatus
+    created_at: datetime
+    updated_at: datetime
+
+
+class TopicWithWorkstreams(TopicRead):
+    workstreams: list[WorkstreamStub] = []
--- a/api/schemas/workstream.py
+++ b/api/schemas/workstream.py
@@ -0,0 +1,38 @@
+import uuid
+from datetime import date, datetime
+
+from pydantic import BaseModel, ConfigDict
+
+from api.models.workstream import WorkstreamStatus
+
+
+class WorkstreamCreate(BaseModel):
+    topic_id: uuid.UUID
+    slug: str
+    title: str
+    description: str | None = None
+    status: WorkstreamStatus = WorkstreamStatus.active
+    owner: str | None = None
+    due_date: date | None = None
+
+
+class WorkstreamUpdate(BaseModel):
+    title: str | None = None
+    description: str | None = None
+    status: WorkstreamStatus | None = None
+    owner: str | None = None
+    due_date: date | None = None
+
+
+class WorkstreamRead(BaseModel):
+    model_config = ConfigDict(from_attributes=True)
+    id: uuid.UUID
+    topic_id: uuid.UUID
+    slug: str
+    title: str
+    description: str | None = None
+    status: WorkstreamStatus
+    owner: str | None = None
+    due_date: date | None = None
+    created_at: datetime
+    updated_at: datetime
--- a/dashboard/observablehq.config.js
+++ b/dashboard/observablehq.config.js
@@ -0,0 +1,13 @@
+export default {
+  root: "src",
+  title: "Custodian State Hub",
+  pages: [
+    { name: "Overview", path: "/" },
+    { name: "Workstreams", path: "/workstreams" },
+    { name: "Decisions", path: "/decisions" },
+    { name: "Progress", path: "/progress" },
+  ],
+  theme: ["air", "near-midnight"],
+  head: `<link rel="icon" href="data:image/svg+xml,<svg xmlns=%22http://www.w3.org/2000/svg%22 viewBox=%220 0 100 100%22><text y=%22.9em%22 font-size=%2290%22>🗄️</text></svg>">`,
+  footer: "Custodian State Hub — local-first, append-only, sovereignty-preserving.",
+};
--- a/dashboard/package-lock.json
+++ b/dashboard/package-lock.json
--- a/dashboard/package.json
+++ b/dashboard/package.json
@@ -0,0 +1,14 @@
+{
+  "name": "custodian-state-hub-dashboard",
+  "version": "0.1.0",
+  "private": true,
+  "type": "module",
+  "scripts": {
+    "dev": "observable preview",
+    "build": "observable build",
+    "clean": "rm -rf dist"
+  },
+  "dependencies": {
+    "@observablehq/framework": "^1.13.3"
+  }
+}
--- a/dashboard/src/data/decisions.json.py
+++ b/dashboard/src/data/decisions.json.py
@@ -0,0 +1,15 @@
+#!/usr/bin/env python3
+"""Observable data loader: all decisions."""
+import json
+import os
+import urllib.request
+import urllib.error
+
+API_BASE = os.environ.get("API_BASE", "http://127.0.0.1:8000").rstrip("/")
+
+try:
+    with urllib.request.urlopen(f"{API_BASE}/decisions", timeout=10) as resp:
+        data = json.loads(resp.read())
+    print(json.dumps(data))
+except urllib.error.URLError as e:
+    print(json.dumps({"error": str(e), "decisions": []}))
--- a/dashboard/src/data/progress.json.py
+++ b/dashboard/src/data/progress.json.py
@@ -0,0 +1,16 @@
+#!/usr/bin/env python3
+"""Observable data loader: recent progress events (last 200)."""
+import json
+import os
+import urllib.request
+import urllib.error
+
+API_BASE = os.environ.get("API_BASE", "http://127.0.0.1:8000").rstrip("/")
+
+try:
+    url = f"{API_BASE}/progress?limit=200"
+    with urllib.request.urlopen(url, timeout=10) as resp:
+        data = json.loads(resp.read())
+    print(json.dumps(data))
+except urllib.error.URLError as e:
+    print(json.dumps({"error": str(e), "events": []}))
--- a/dashboard/src/data/summary.json.py
+++ b/dashboard/src/data/summary.json.py
@@ -0,0 +1,31 @@
+#!/usr/bin/env python3
+"""Observable data loader: fetches /state/summary from the API."""
+import json
+import os
+import sys
+import urllib.request
+import urllib.error
+
+API_BASE = os.environ.get("API_BASE", "http://127.0.0.1:8000").rstrip("/")
+
+try:
+    with urllib.request.urlopen(f"{API_BASE}/state/summary", timeout=10) as resp:
+        data = json.loads(resp.read())
+    print(json.dumps(data))
+except urllib.error.URLError as e:
+    # Return empty structure so the dashboard can show an error state
+    print(json.dumps({
+        "error": str(e),
+        "generated_at": None,
+        "totals": {
+            "topics": {"active": 0, "paused": 0, "archived": 0, "total": 0},
+            "workstreams": {"active": 0, "blocked": 0, "completed": 0, "archived": 0, "total": 0},
+            "tasks": {"todo": 0, "in_progress": 0, "blocked": 0, "done": 0, "cancelled": 0, "total": 0},
+            "decisions": {"open": 0, "resolved": 0, "escalated": 0, "superseded": 0, "total": 0},
+        },
+        "topics": [],
+        "blocking_decisions": [],
+        "blocked_tasks": [],
+        "recent_progress": [],
+        "open_workstreams": [],
+    }))
--- a/dashboard/src/data/workstreams.json.py
+++ b/dashboard/src/data/workstreams.json.py
@@ -0,0 +1,15 @@
+#!/usr/bin/env python3
+"""Observable data loader: all workstreams."""
+import json
+import os
+import urllib.request
+import urllib.error
+
+API_BASE = os.environ.get("API_BASE", "http://127.0.0.1:8000").rstrip("/")
+
+try:
+    with urllib.request.urlopen(f"{API_BASE}/workstreams", timeout=10) as resp:
+        data = json.loads(resp.read())
+    print(json.dumps(data))
+except urllib.error.URLError as e:
+    print(json.dumps({"error": str(e), "workstreams": []}))
--- a/dashboard/src/decisions.md
+++ b/dashboard/src/decisions.md
@@ -0,0 +1,70 @@
+---
+title: Decisions
+---
+
+# Decisions
+
+```js
+const decisions = await FileAttachment("data/decisions.json").json();
+const data = Array.isArray(decisions) ? decisions : [];
+const pending = data.filter(d => d.decision_type === "pending");
+const made = data.filter(d => d.decision_type === "made");
+```
+
+```js
+const tab = view(Inputs.select(["Pending", "Made"], { label: "View" }));
+```
+
+```js
+const shown = tab === "Pending" ? pending : made;
+
+display(Inputs.table(shown.map(d => ({
+  Title: d.title,
+  Status: d.status + (d.escalation_note ? " ⚠️" : ""),
+  Decided_by: d.decided_by ?? "—",
+  Deadline: d.deadline ? new Date(d.deadline).toLocaleDateString() : "—",
+  Rationale: (d.rationale ?? "").slice(0, 80),
+  Updated: new Date(d.updated_at).toLocaleDateString(),
+})), { rows: 30 }));
+```
+
+## Resolution Velocity
+
+```js
+import * as Plot from "npm:@observablehq/plot";
+
+const resolved = made.filter(d => d.decided_at);
+const byMonth = resolved.reduce((acc, d) => {
+  const m = d.decided_at.slice(0, 7);
+  acc[m] = (acc[m] ?? 0) + 1;
+  return acc;
+}, {});
+
+display(Plot.plot({
+  title: "Decisions Resolved per Month",
+  x: { label: "Month", tickRotate: -30 },
+  y: { label: "Count", grid: true },
+  marks: [
+    Plot.barY(
+      Object.entries(byMonth).map(([month, count]) => ({ month, count })),
+      { x: "month", y: "count", fill: "steelblue", tip: true }
+    ),
+    Plot.ruleY([0]),
+  ],
+  marginBottom: 60,
+  width: 700,
+}));
+```
+
+```js
+if (tab === "Pending" && pending.filter(d => d.escalation_note).length > 0) {
+  display(html`<div class="escalation-box">
+    <strong>⚠️ Escalated decisions require human approval before any action is taken (constitution §4).</strong>
+    <ul>${pending.filter(d => d.escalation_note).map(d => html`<li><b>${d.title}</b>: ${d.escalation_note}</li>`)}</ul>
+  </div>`);
+}
+```
+
+<style>
+.escalation-box { background: #fff3cd; border: 2px solid orange; border-radius: 8px; padding: 1rem; margin-top: 1rem; }
+</style>
--- a/dashboard/src/index.md
+++ b/dashboard/src/index.md
@@ -0,0 +1,122 @@
+---
+title: Overview
+---
+
+# Custodian State Hub
+
+```js
+const summary = await FileAttachment("data/summary.json").json();
+const totals = summary.totals ?? {};
+const ws = totals.workstreams ?? {};
+const tasks = totals.tasks ?? {};
+const decisions = totals.decisions ?? {};
+const topics = totals.topics ?? {};
+```
+
+```js
+if (summary.error) display(html`<div class="warning">⚠️ API unreachable: ${summary.error}. Run <code>make api</code>.</div>`);
+```
+
+## Status
+
+```js
+display(html`<div class="grid grid-cols-4" style="gap:1rem; margin-bottom:1.5rem;">
+  <div class="card">
+    <h3>Active Workstreams</h3>
+    <p class="big-number">${ws.active ?? 0}</p>
+    <small>${ws.blocked ?? 0} blocked</small>
+  </div>
+  <div class="card ${(decisions.open + decisions.escalated) > 0 ? 'warn' : ''}">
+    <h3>Blocking Decisions</h3>
+    <p class="big-number">${(decisions.open ?? 0) + (decisions.escalated ?? 0)}</p>
+    <small>${decisions.escalated ?? 0} escalated</small>
+  </div>
+  <div class="card ${(tasks.blocked ?? 0) > 0 ? 'warn' : ''}">
+    <h3>Blocked Tasks</h3>
+    <p class="big-number">${tasks.blocked ?? 0}</p>
+    <small>of ${tasks.total ?? 0} total</small>
+  </div>
+  <div class="card">
+    <h3>Progress Events Today</h3>
+    <p class="big-number">${(summary.recent_progress ?? []).filter(e => e.created_at?.startsWith(new Date().toISOString().slice(0,10))).length}</p>
+    <small>last 20 shown below</small>
+  </div>
+</div>`);
+```
+
+## Tasks by Domain
+
+```js
+import * as Plot from "npm:@observablehq/plot";
+
+const tasksByDomain = [];
+for (const topic of (summary.topics ?? [])) {
+  for (const ws of (topic.workstreams ?? [])) {
+    // workstream stubs don't include tasks in summary — show per-topic WS count as proxy
+  }
+  tasksByDomain.push({ domain: topic.domain, status: topic.status, count: (topic.workstreams ?? []).length });
+}
+
+display(Plot.plot({
+  title: "Open Workstreams by Domain",
+  x: { label: "Domain" },
+  y: { label: "Count", grid: true },
+  marks: [
+    Plot.barY(tasksByDomain, { x: "domain", y: "count", fill: "domain", tip: true }),
+    Plot.ruleY([0]),
+  ],
+  marginBottom: 80,
+  width: 700,
+}));
+```
+
+## Blocking Decisions
+
+```js
+const blocking = summary.blocking_decisions ?? [];
+if (blocking.length === 0) {
+  display(html`<p style="color:green">✓ No blocking decisions.</p>`);
+} else {
+  display(Inputs.table(blocking.map(d => ({
+    Title: d.title,
+    Status: d.status,
+    Deadline: d.deadline ? new Date(d.deadline).toLocaleDateString() : "—",
+    Escalated: d.escalation_note ? "⚠️" : "",
+  }))));
+}
+```
+
+## Decisions Due Within 7 Days
+
+```js
+const now = new Date();
+const in7 = new Date(now.getTime() + 7*24*60*60*1000);
+const due = (summary.blocking_decisions ?? []).filter(d => d.deadline && new Date(d.deadline) <= in7);
+if (due.length === 0) {
+  display(html`<p>No decisions due in next 7 days.</p>`);
+} else {
+  display(Inputs.table(due.map(d => ({
+    Title: d.title,
+    Deadline: new Date(d.deadline).toLocaleString(),
+    Status: d.status,
+  }))));
+}
+```
+
+## Recent Activity
+
+```js
+display(Inputs.table((summary.recent_progress ?? []).map(e => ({
+  Time: new Date(e.created_at).toLocaleString(),
+  Type: e.event_type,
+  Author: e.author ?? "—",
+  Summary: e.summary,
+})), { maxWidth: 900 }));
+```
+
+<style>
+.card { background: var(--theme-background-alt); border-radius: 8px; padding: 1rem; }
+.card.warn { border: 2px solid orange; }
+.big-number { font-size: 2.5rem; font-weight: bold; margin: 0.25rem 0; }
+.warning { background: #fff3cd; border: 1px solid #ffc107; border-radius: 4px; padding: 0.75rem; }
+</style>
--- a/dashboard/src/progress.md
+++ b/dashboard/src/progress.md
@@ -0,0 +1,77 @@
+---
+title: Progress
+---
+
+# Progress Log
+
+*Append-only per constitution §5 — no deletions.*
+
+```js
+const events = await FileAttachment("data/progress.json").json();
+const data = Array.isArray(events) ? events : [];
+```
+
+```js
+const authorFilter = view(Inputs.select(
+  ["(all)", ...new Set(data.map(e => e.author ?? "unknown"))],
+  { label: "Author" }
+));
+const typeFilter = view(Inputs.select(
+  ["(all)", ...new Set(data.map(e => e.event_type))],
+  { label: "Event type" }
+));
+const sinceFilter = view(Inputs.date({ label: "Since" }));
+```
+
+```js
+const filtered = data.filter(e =>
+  (authorFilter === "(all)" || (e.author ?? "unknown") === authorFilter) &&
+  (typeFilter === "(all)" || e.event_type === typeFilter) &&
+  (!sinceFilter || new Date(e.created_at) >= sinceFilter)
+);
+
+display(html`<p><strong>${filtered.length}</strong> events shown (append-only, no deletions).</p>`);
+
+display(Inputs.table(filtered.map(e => ({
+  Time: new Date(e.created_at).toLocaleString(),
+  Type: e.event_type,
+  Author: e.author ?? "—",
+  Summary: e.summary,
+})), { rows: 50 }));
+```
+
+## Event Volume (Last 30 Days)
+
+```js
+import * as Plot from "npm:@observablehq/plot";
+
+const cutoff = new Date();
+cutoff.setDate(cutoff.getDate() - 30);
+
+const byDay = data
+  .filter(e => new Date(e.created_at) >= cutoff)
+  .reduce((acc, e) => {
+    const day = e.created_at.slice(0, 10);
+    acc[day] = (acc[day] ?? 0) + 1;
+    return acc;
+  }, {});
+
+display(Plot.plot({
+  title: "Progress Events per Day (30-day window)",
+  x: { label: "Date", tickRotate: -30 },
+  y: { label: "Events", grid: true },
+  marks: [
+    Plot.areaY(
+      Object.entries(byDay).sort().map(([day, count]) => ({ day, count })),
+      { x: "day", y: "count", fill: "steelblue", fillOpacity: 0.3 }
+    ),
+    Plot.lineY(
+      Object.entries(byDay).sort().map(([day, count]) => ({ day, count })),
+      { x: "day", y: "count", stroke: "steelblue" }
+    ),
+    Plot.ruleY([0]),
+  ],
+  marginBottom: 60,
+  width: 750,
+}));
+```
--- a/dashboard/src/workstreams.md
+++ b/dashboard/src/workstreams.md
@@ -0,0 +1,67 @@
+---
+title: Workstreams
+---
+
+# Workstreams
+
+```js
+const workstreams = await FileAttachment("data/workstreams.json").json();
+const data = Array.isArray(workstreams) ? workstreams : [];
+```
+
+```js
+const domainFilter = view(Inputs.select(
+  ["(all)", ...new Set(data.map(w => w.domain ?? "unknown"))],
+  { label: "Domain" }
+));
+const statusFilter = view(Inputs.select(
+  ["(all)", "active", "blocked", "completed", "archived"],
+  { label: "Status" }
+));
+const ownerFilter = view(Inputs.text({ label: "Owner contains" }));
+```
+
+```js
+const filtered = data.filter(w =>
+  (domainFilter === "(all)" || w.domain === domainFilter) &&
+  (statusFilter === "(all)" || w.status === statusFilter) &&
+  (!ownerFilter || (w.owner ?? "").toLowerCase().includes(ownerFilter.toLowerCase()))
+);
+
+const STATUS_COLOR = {
+  active: "green",
+  blocked: "orange",
+  completed: "blue",
+  archived: "gray",
+};
+
+display(Inputs.table(filtered.map(w => ({
+  Title: w.title,
+  Domain: w.domain,
+  Status: w.status,
+  Owner: w.owner ?? "—",
+  "Due": w.due_date ?? "—",
+  "Updated": new Date(w.updated_at).toLocaleDateString(),
+})), {
+  rows: 20,
+}));
+```
+
+```js
+import * as Plot from "npm:@observablehq/plot";
+
+display(Plot.plot({
+  title: "Workstream Status Distribution",
+  marks: [
+    Plot.barX(
+      Object.entries(
+        filtered.reduce((acc, w) => { acc[w.status] = (acc[w.status] ?? 0) + 1; return acc; }, {})
+      ).map(([status, count]) => ({ status, count })),
+      { y: "status", x: "count", fill: "status", tip: true }
+    ),
+    Plot.ruleX([0]),
+  ],
+  marginLeft: 80,
+  width: 500,
+}));
+```
--- a/infra/docker-compose.yml
+++ b/infra/docker-compose.yml
@@ -0,0 +1,33 @@
+services:
+  postgres:
+    image: postgres:16-alpine
+    restart: unless-stopped
+    ports:
+      - "127.0.0.1:5432:5432"
+    environment:
+      POSTGRES_DB: ${POSTGRES_DB:-custodian}
+      POSTGRES_USER: ${POSTGRES_USER:-custodian}
+      POSTGRES_PASSWORD: ${POSTGRES_PASSWORD}
+    volumes:
+      - pg_data:/var/lib/postgresql/data
+    healthcheck:
+      test: ["CMD-SHELL", "pg_isready -U ${POSTGRES_USER:-custodian} -d ${POSTGRES_DB:-custodian}"]
+      interval: 5s
+      timeout: 5s
+      retries: 5
+
+  pgadmin:
+    image: dpage/pgadmin4:latest
+    profiles: ["tools"]
+    restart: unless-stopped
+    ports:
+      - "127.0.0.1:5050:80"
+    environment:
+      PGADMIN_DEFAULT_EMAIL: ${PGADMIN_EMAIL:-admin@local.dev}
+      PGADMIN_DEFAULT_PASSWORD: ${PGADMIN_PASSWORD:-admin}
+    depends_on:
+      postgres:
+        condition: service_healthy
+
+volumes:
+  pg_data:
--- a/mcp_server/init.py
+++ b/mcp_server/init.py
--- a/mcp_server/server.py
+++ b/mcp_server/server.py
@@ -0,0 +1,360 @@
+"""Custodian State Hub MCP Server (stdio).
+
+Thin HTTP client over the FastAPI service — no direct DB access.
+All business logic stays in the API; this layer is stateless.
+"""
+from __future__ import annotations
+
+import json
+import os
+import sys
+from datetime import datetime
+from typing import Any
+from uuid import UUID
+
+import httpx
+from fastmcp import FastMCP
+
+API_BASE = os.environ.get("API_BASE", "http://127.0.0.1:8000").rstrip("/")
+
+mcp = FastMCP(
+    name="state-hub",
+    instructions=(
+        "Custodian State Hub: tracks topics, workstreams, tasks, decisions, and progress events. "
+        "Start every session with get_state_summary() for orientation. "
+        "All writes emit a progress_event automatically."
+    ),
+)
+
+# ---------------------------------------------------------------------------
+# HTTP helpers
+# ---------------------------------------------------------------------------
+
+def _client() -> httpx.Client:
+    return httpx.Client(base_url=API_BASE, timeout=30.0)
+
+
+def _get(path: str, params: dict | None = None) -> Any:
+    with _client() as c:
+        r = c.get(path, params={k: v for k, v in (params or {}).items() if v is not None})
+        r.raise_for_status()
+        return r.json()
+
+
+def _post(path: str, body: dict) -> Any:
+    if not path.endswith("/"):
+        path = path + "/"
+    with _client() as c:
+        r = c.post(path, json={k: v for k, v in body.items() if v is not None})
+        r.raise_for_status()
+        return r.json()
+
+
+def _patch(path: str, body: dict) -> Any:
+    with _client() as c:
+        r = c.patch(path, json={k: v for k, v in body.items() if v is not None})
+        r.raise_for_status()
+        return r.json()
+
+
+# ---------------------------------------------------------------------------
+# Resources
+# ---------------------------------------------------------------------------
+
+@mcp.resource("state://summary")
+def resource_summary() -> str:
+    """Full StateSummary JSON — primary orientation resource."""
+    return json.dumps(_get("/state/summary"), indent=2)
+
+
+@mcp.resource("state://topics")
+def resource_topics() -> str:
+    """Active topics list."""
+    return json.dumps(_get("/topics", {"status": "active"}), indent=2)
+
+
+@mcp.resource("state://workstreams/{topic_slug}")
+def resource_workstreams(topic_slug: str) -> str:
+    """Workstreams for a topic (by slug)."""
+    topics = _get("/topics", {"status": "active"})
+    match = next((t for t in topics if t["slug"] == topic_slug), None)
+    if not match:
+        return json.dumps({"error": f"Topic '{topic_slug}' not found"})
+    return json.dumps(_get("/workstreams", {"topic_id": match["id"]}), indent=2)
+
+
+@mcp.resource("state://decisions/blocking")
+def resource_blocking_decisions() -> str:
+    """All pending/escalated decisions."""
+    return json.dumps(
+        _get("/decisions", {"decision_type": "pending", "status": "open"}),
+        indent=2,
+    )
+
+
+@mcp.resource("state://tasks/blocked")
+def resource_blocked_tasks() -> str:
+    """All tasks with status=blocked."""
+    return json.dumps(_get("/tasks", {"status": "blocked"}), indent=2)
+
+
+# ---------------------------------------------------------------------------
+# Query tools
+# ---------------------------------------------------------------------------
+
+@mcp.tool()
+def get_state_summary() -> str:
+    """Primary orientation tool. Call at the start of every session.
+
+    Returns a full snapshot: topic/workstream/task/decision totals, blocking
+    decisions, blocked tasks, open workstreams, and the 20 most recent events.
+    """
+    return json.dumps(_get("/state/summary"), indent=2)
+
+
+@mcp.tool()
+def get_topic(slug: str) -> str:
+    """Return a topic (with workstreams) by slug, plus its recent progress events."""
+    topics = _get("/topics")
+    match = next((t for t in topics if t["slug"] == slug), None)
+    if not match:
+        return json.dumps({"error": f"Topic '{slug}' not found"})
+    topic_detail = _get(f"/topics/{match['id']}")
+    recent = _get("/progress", {"topic_id": match["id"], "limit": 10})
+    return json.dumps({"topic": topic_detail, "recent_progress": recent}, indent=2)
+
+
+@mcp.tool()
+def list_blocked_tasks(workstream_id: str | None = None) -> str:
+    """List all tasks with status=blocked, optionally filtered by workstream_id."""
+    return json.dumps(_get("/tasks", {"status": "blocked", "workstream_id": workstream_id}), indent=2)
+
+
+@mcp.tool()
+def list_pending_decisions(topic_id: str | None = None) -> str:
+    """List pending decisions sorted by deadline (nulls last).
+
+    Optionally filter by topic_id. Escalated decisions are included and
+    highlighted by their escalation_note.
+    """
+    results = _get("/decisions", {"decision_type": "pending", "topic_id": topic_id})
+    return json.dumps(results, indent=2)
+
+
+@mcp.tool()
+def get_recent_progress(limit: int = 20, since: str | None = None) -> str:
+    """Retrieve recent progress events to reconstruct session history.
+
+    Args:
+        limit: max events to return (default 20)
+        since: ISO datetime string — only events after this timestamp
+    """
+    return json.dumps(_get("/progress", {"limit": limit, "since": since}), indent=2)
+
+
+# ---------------------------------------------------------------------------
+# Mutate tools
+# ---------------------------------------------------------------------------
+
+@mcp.tool()
+def create_task(
+    workstream_id: str,
+    title: str,
+    priority: str = "medium",
+    description: str | None = None,
+    assignee: str | None = None,
+    due_date: str | None = None,
+) -> str:
+    """Create a new task and emit a progress_event.
+
+    Args:
+        workstream_id: UUID of the parent workstream
+        title: task title
+        priority: low | medium | high | critical
+        description: optional longer description
+        assignee: optional assignee name
+        due_date: optional ISO date string (YYYY-MM-DD)
+    """
+    task = _post("/tasks", {
+        "workstream_id": workstream_id,
+        "title": title,
+        "priority": priority,
+        "description": description,
+        "assignee": assignee,
+        "due_date": due_date,
+    })
+    _post("/progress", {
+        "workstream_id": workstream_id,
+        "task_id": task["id"],
+        "event_type": "task_created",
+        "summary": f"Task created: {title}",
+        "author": "custodian",
+        "detail": {"priority": priority, "assignee": assignee},
+    })
+    return json.dumps(task, indent=2)
+
+
+@mcp.tool()
+def update_task_status(
+    task_id: str,
+    status: str,
+    blocking_reason: str | None = None,
+) -> str:
+    """Update a task's status. blocking_reason is required when status='blocked'.
+
+    Args:
+        task_id: UUID of the task
+        status: todo | in_progress | blocked | done | cancelled
+        blocking_reason: required when status=blocked
+    """
+    body: dict[str, Any] = {"status": status}
+    if blocking_reason:
+        body["blocking_reason"] = blocking_reason
+    task = _patch(f"/tasks/{task_id}", body)
+    _post("/progress", {
+        "task_id": task_id,
+        "workstream_id": task.get("workstream_id"),
+        "event_type": "task_status_changed",
+        "summary": f"Task status → {status}: {task['title']}",
+        "author": "custodian",
+        "detail": {"blocking_reason": blocking_reason},
+    })
+    return json.dumps(task, indent=2)
+
+
+@mcp.tool()
+def record_decision(
+    title: str,
+    decision_type: str = "pending",
+    topic_id: str | None = None,
+    workstream_id: str | None = None,
+    description: str | None = None,
+    rationale: str | None = None,
+    decided_by: str | None = None,
+    deadline: str | None = None,
+) -> str:
+    """Record a decision (made or pending).
+
+    Pending decisions touching financial/legal topics are auto-escalated per
+    constitution §4.
+
+    Args:
+        title: decision title
+        decision_type: made | pending
+        topic_id: optional topic UUID
+        workstream_id: optional workstream UUID (at least one required)
+        description: optional context
+        rationale: reasoning behind the decision
+        decided_by: person/agent who decided
+        deadline: ISO datetime string for when decision is needed
+    """
+    decision = _post("/decisions", {
+        "title": title,
+        "decision_type": decision_type,
+        "topic_id": topic_id,
+        "workstream_id": workstream_id,
+        "description": description,
+        "rationale": rationale,
+        "decided_by": decided_by,
+        "deadline": deadline,
+    })
+    _post("/progress", {
+        "topic_id": topic_id,
+        "workstream_id": workstream_id,
+        "decision_id": decision["id"],
+        "event_type": "decision_recorded",
+        "summary": f"Decision recorded ({decision_type}): {title}",
+        "author": "custodian",
+        "detail": {"status": decision.get("status"), "escalation_note": decision.get("escalation_note")},
+    })
+    return json.dumps(decision, indent=2)
+
+
+@mcp.tool()
+def resolve_decision(
+    decision_id: str,
+    rationale: str,
+    decided_by: str,
+) -> str:
+    """Mark a decision as resolved.
+
+    Args:
+        decision_id: UUID of the decision
+        rationale: final reasoning/outcome
+        decided_by: who resolved it
+    """
+    decision = _patch(f"/decisions/{decision_id}", {
+        "status": "resolved",
+        "decision_type": "made",
+        "rationale": rationale,
+        "decided_by": decided_by,
+        "decided_at": datetime.utcnow().isoformat() + "Z",
+    })
+    _post("/progress", {
+        "topic_id": decision.get("topic_id"),
+        "workstream_id": decision.get("workstream_id"),
+        "decision_id": decision_id,
+        "event_type": "decision_resolved",
+        "summary": f"Decision resolved by {decided_by}: {decision['title']}",
+        "author": "custodian",
+        "detail": {"rationale": rationale},
+    })
+    return json.dumps(decision, indent=2)
+
+
+@mcp.tool()
+def add_progress_event(
+    summary: str,
+    event_type: str = "note",
+    topic_id: str | None = None,
+    workstream_id: str | None = None,
+    task_id: str | None = None,
+    detail: dict | None = None,
+) -> str:
+    """Append a progress event to the log.
+
+    Args:
+        summary: human-readable summary of what happened
+        event_type: free-form label, e.g. note | milestone | blocker | insight
+        topic_id: optional topic UUID
+        workstream_id: optional workstream UUID
+        task_id: optional task UUID
+        detail: optional structured data (JSONB)
+    """
+    event = _post("/progress", {
+        "topic_id": topic_id,
+        "workstream_id": workstream_id,
+        "task_id": task_id,
+        "event_type": event_type,
+        "summary": summary,
+        "author": "custodian",
+        "detail": detail,
+    })
+    return json.dumps(event, indent=2)
+
+
+@mcp.tool()
+def update_workstream_status(workstream_id: str, status: str) -> str:
+    """Update a workstream's status.
+
+    Args:
+        workstream_id: UUID of the workstream
+        status: active | blocked | completed | archived
+    """
+    ws = _patch(f"/workstreams/{workstream_id}", {"status": status})
+    _post("/progress", {
+        "workstream_id": workstream_id,
+        "topic_id": ws.get("topic_id"),
+        "event_type": "workstream_status_changed",
+        "summary": f"Workstream status → {status}: {ws['title']}",
+        "author": "custodian",
+    })
+    return json.dumps(ws, indent=2)
+
+
+# ---------------------------------------------------------------------------
+# Entry point
+# ---------------------------------------------------------------------------
+
+if __name__ == "__main__":
+    mcp.run(transport="stdio")
--- a/migrations/init.py
+++ b/migrations/init.py
--- a/migrations/env.py
+++ b/migrations/env.py
@@ -0,0 +1,56 @@
+import os
+from logging.config import fileConfig
+
+from alembic import context
+from sqlalchemy import engine_from_config, pool
+from sqlalchemy.ext.asyncio import AsyncEngine
+
+# Import all models so Alembic can detect them
+from api.models import Base  # noqa: F401 — registers all ORM classes
+from api.models import (  # noqa: F401
+    Decision, ProgressEvent, Task, Topic, Workstream,
+)
+
+config = context.config
+
+if config.config_file_name is not None:
+    fileConfig(config.config_file_name)
+
+target_metadata = Base.metadata
+
+# Allow DATABASE_URL to override alembic.ini
+db_url = os.environ.get("DATABASE_URL")
+if db_url:
+    # Alembic sync driver: replace asyncpg with psycopg2
+    sync_url = db_url.replace("postgresql+asyncpg://", "postgresql+psycopg2://")
+    config.set_main_option("sqlalchemy.url", sync_url)
+
+
+def run_migrations_offline() -> None:
+    url = config.get_main_option("sqlalchemy.url")
+    context.configure(
+        url=url,
+        target_metadata=target_metadata,
+        literal_binds=True,
+        dialect_opts={"paramstyle": "named"},
+    )
+    with context.begin_transaction():
+        context.run_migrations()
+
+
+def run_migrations_online() -> None:
+    connectable = engine_from_config(
+        config.get_section(config.config_ini_section, {}),
+        prefix="sqlalchemy.",
+        poolclass=pool.NullPool,
+    )
+    with connectable.connect() as connection:
+        context.configure(connection=connection, target_metadata=target_metadata)
+        with context.begin_transaction():
+            context.run_migrations()
+
+
+if context.is_offline_mode():
+    run_migrations_offline()
+else:
+    run_migrations_online()
--- a/migrations/versions/0001_initial_schema.py
+++ b/migrations/versions/0001_initial_schema.py
@@ -0,0 +1,248 @@
+"""initial schema
+
+Revision ID: 0001
+Revises:
+Create Date: 2026-02-24 00:00:00.000000
+"""
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy.dialects import postgresql
+
+revision: str = "0001"
+down_revision: Union[str, None] = None
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # Enums
+    topic_status = postgresql.ENUM(
+        "active", "paused", "archived", name="topicstatus", create_type=True
+    )
+    domain = postgresql.ENUM(
+        "custodian", "railiance", "markitect", "coulomb_social", "personhood",
+        "foerster_capabilities", name="domain", create_type=True
+    )
+    workstream_status = postgresql.ENUM(
+        "active", "blocked", "completed", "archived", name="workstreamstatus", create_type=True
+    )
+    task_status = postgresql.ENUM(
+        "todo", "in_progress", "blocked", "done", "cancelled", name="taskstatus", create_type=True
+    )
+    task_priority = postgresql.ENUM(
+        "low", "medium", "high", "critical", name="taskpriority", create_type=True
+    )
+    decision_type = postgresql.ENUM(
+        "made", "pending", name="decisiontype", create_type=True
+    )
+    decision_status = postgresql.ENUM(
+        "open", "resolved", "escalated", "superseded", name="decisionstatus", create_type=True
+    )
+
+    # topics
+    op.create_table(
+        "topics",
+        sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True),
+        sa.Column("slug", sa.String(100), nullable=False, unique=True),
+        sa.Column("title", sa.String(255), nullable=False),
+        sa.Column("description", sa.Text, nullable=True),
+        sa.Column("domain", domain, nullable=False),
+        sa.Column("status", topic_status, nullable=False, server_default="active"),
+        sa.Column(
+            "created_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.text("now()"),
+            nullable=False,
+        ),
+        sa.Column(
+            "updated_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.text("now()"),
+            nullable=False,
+        ),
+    )
+    op.create_index("ix_topics_slug", "topics", ["slug"])
+
+    # workstreams
+    op.create_table(
+        "workstreams",
+        sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True),
+        sa.Column(
+            "topic_id",
+            postgresql.UUID(as_uuid=True),
+            sa.ForeignKey("topics.id", ondelete="RESTRICT"),
+            nullable=False,
+        ),
+        sa.Column("slug", sa.String(100), nullable=False, unique=True),
+        sa.Column("title", sa.String(255), nullable=False),
+        sa.Column("description", sa.Text, nullable=True),
+        sa.Column("status", workstream_status, nullable=False, server_default="active"),
+        sa.Column("owner", sa.String(100), nullable=True),
+        sa.Column("due_date", sa.Date, nullable=True),
+        sa.Column(
+            "created_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.text("now()"),
+            nullable=False,
+        ),
+        sa.Column(
+            "updated_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.text("now()"),
+            nullable=False,
+        ),
+    )
+    op.create_index("ix_workstreams_slug", "workstreams", ["slug"])
+    op.create_index("ix_workstreams_topic_id", "workstreams", ["topic_id"])
+
+    # tasks
+    op.create_table(
+        "tasks",
+        sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True),
+        sa.Column(
+            "workstream_id",
+            postgresql.UUID(as_uuid=True),
+            sa.ForeignKey("workstreams.id", ondelete="RESTRICT"),
+            nullable=False,
+        ),
+        sa.Column("title", sa.String(255), nullable=False),
+        sa.Column("description", sa.Text, nullable=True),
+        sa.Column("status", task_status, nullable=False, server_default="todo"),
+        sa.Column("priority", task_priority, nullable=False, server_default="medium"),
+        sa.Column("assignee", sa.String(100), nullable=True),
+        sa.Column("due_date", sa.Date, nullable=True),
+        sa.Column("blocking_reason", sa.Text, nullable=True),
+        sa.Column(
+            "parent_task_id",
+            postgresql.UUID(as_uuid=True),
+            sa.ForeignKey("tasks.id", ondelete="SET NULL"),
+            nullable=True,
+        ),
+        sa.Column(
+            "created_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.text("now()"),
+            nullable=False,
+        ),
+        sa.Column(
+            "updated_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.text("now()"),
+            nullable=False,
+        ),
+    )
+    op.create_index("ix_tasks_workstream_id", "tasks", ["workstream_id"])
+
+    # decisions
+    op.create_table(
+        "decisions",
+        sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True),
+        sa.Column(
+            "topic_id",
+            postgresql.UUID(as_uuid=True),
+            sa.ForeignKey("topics.id", ondelete="RESTRICT"),
+            nullable=True,
+        ),
+        sa.Column(
+            "workstream_id",
+            postgresql.UUID(as_uuid=True),
+            sa.ForeignKey("workstreams.id", ondelete="RESTRICT"),
+            nullable=True,
+        ),
+        sa.Column("title", sa.String(255), nullable=False),
+        sa.Column("description", sa.Text, nullable=True),
+        sa.Column("decision_type", decision_type, nullable=False, server_default="pending"),
+        sa.Column("status", decision_status, nullable=False, server_default="open"),
+        sa.Column("rationale", sa.Text, nullable=True),
+        sa.Column("decided_by", sa.String(100), nullable=True),
+        sa.Column("decided_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("deadline", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("escalation_note", sa.Text, nullable=True),
+        sa.Column(
+            "superseded_by",
+            postgresql.UUID(as_uuid=True),
+            sa.ForeignKey("decisions.id", ondelete="SET NULL"),
+            nullable=True,
+        ),
+        sa.Column(
+            "created_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.text("now()"),
+            nullable=False,
+        ),
+        sa.Column(
+            "updated_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.text("now()"),
+            nullable=False,
+        ),
+        sa.CheckConstraint(
+            "topic_id IS NOT NULL OR workstream_id IS NOT NULL",
+            name="ck_decisions_topic_or_workstream",
+        ),
+    )
+    op.create_index("ix_decisions_topic_id", "decisions", ["topic_id"])
+    op.create_index("ix_decisions_workstream_id", "decisions", ["workstream_id"])
+
+    # progress_events
+    op.create_table(
+        "progress_events",
+        sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True),
+        sa.Column(
+            "topic_id",
+            postgresql.UUID(as_uuid=True),
+            sa.ForeignKey("topics.id", ondelete="RESTRICT"),
+            nullable=True,
+        ),
+        sa.Column(
+            "workstream_id",
+            postgresql.UUID(as_uuid=True),
+            sa.ForeignKey("workstreams.id", ondelete="RESTRICT"),
+            nullable=True,
+        ),
+        sa.Column(
+            "task_id",
+            postgresql.UUID(as_uuid=True),
+            sa.ForeignKey("tasks.id", ondelete="RESTRICT"),
+            nullable=True,
+        ),
+        sa.Column(
+            "decision_id",
+            postgresql.UUID(as_uuid=True),
+            sa.ForeignKey("decisions.id", ondelete="RESTRICT"),
+            nullable=True,
+        ),
+        sa.Column("event_type", sa.String(50), nullable=False),
+        sa.Column("summary", sa.Text, nullable=False),
+        sa.Column("detail", postgresql.JSONB, nullable=True),
+        sa.Column("author", sa.String(100), nullable=True),
+        sa.Column("session_id", sa.String(100), nullable=True),
+        sa.Column(
+            "created_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.text("now()"),
+            nullable=False,
+        ),
+    )
+    op.create_index("ix_progress_events_topic_id", "progress_events", ["topic_id"])
+    op.create_index("ix_progress_events_workstream_id", "progress_events", ["workstream_id"])
+    op.create_index("ix_progress_events_task_id", "progress_events", ["task_id"])
+    op.create_index("ix_progress_events_decision_id", "progress_events", ["decision_id"])
+    op.create_index("ix_progress_events_event_type", "progress_events", ["event_type"])
+    op.create_index("ix_progress_events_created_at", "progress_events", ["created_at"])
+
+
+def downgrade() -> None:
+    op.drop_table("progress_events")
+    op.drop_table("decisions")
+    op.drop_table("tasks")
+    op.drop_table("workstreams")
+    op.drop_table("topics")
+
+    for name in [
+        "topicstatus", "domain", "workstreamstatus",
+        "taskstatus", "taskpriority", "decisiontype", "decisionstatus",
+    ]:
+        op.execute(f"DROP TYPE IF EXISTS {name}")
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -0,0 +1,32 @@
+[project]
+name = "state-hub"
+version = "0.1.0"
+description = "Custodian State Hub — PostgreSQL + FastAPI + MCP server"
+requires-python = ">=3.12"
+dependencies = [
+    "fastapi>=0.115.0",
+    "uvicorn[standard]>=0.32.0",
+    "sqlalchemy[asyncio]>=2.0.0",
+    "asyncpg>=0.30.0",
+    "alembic>=1.14.0",
+    "pydantic>=2.10.0",
+    "pydantic-settings>=2.7.0",
+    "httpx>=0.28.0",
+    "fastmcp>=2.0.0",
+    "python-dotenv>=1.0.0",
+    "psycopg2-binary>=2.9.0",
+]
+
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
+[tool.hatch.build.targets.wheel]
+packages = ["api", "mcp_server"]
+
+[tool.uv]
+dev-dependencies = [
+    "pytest>=8.0.0",
+    "pytest-asyncio>=0.24.0",
+    "httpx>=0.28.0",
+]
--- a/scripts/pull_image.py
+++ b/scripts/pull_image.py
@@ -0,0 +1,185 @@
+#!/usr/bin/env python3
+"""
+Pull a Docker Hub image via the registry v2 API using Python's SSL (OpenSSL),
+then import it via `docker load`. Bypasses Docker's Go TLS client entirely.
+
+Usage: python pull_image.py <image:tag> [output.tar]
+  e.g: python pull_image.py postgres:16-alpine postgres.tar
+"""
+import json
+import os
+import sys
+import tarfile
+import tempfile
+import urllib.request
+import urllib.error
+import hashlib
+
+
+def get_token(repo: str) -> str:
+    url = f"https://auth.docker.io/token?service=registry.docker.io&scope=repository:{repo}:pull"
+    with urllib.request.urlopen(url, timeout=30) as r:
+        return json.loads(r.read())["token"]
+
+
+class _StripAuthOnRedirect(urllib.request.HTTPRedirectHandler):
+    """Follow redirects but strip Authorization; keep Range and other headers."""
+    def redirect_request(self, req, fp, code, msg, headers, newurl):
+        new_req = urllib.request.Request(newurl)
+        # Forward Range header (needed for chunked downloads) but NOT Authorization
+        for hdr in ("Range",):
+            val = req.get_header(hdr.capitalize())
+            if val:
+                new_req.add_header(hdr, val)
+        return new_req
+
+
+def _opener():
+    return urllib.request.build_opener(_StripAuthOnRedirect())
+
+
+def registry_get(url: str, token: str, headers: dict | None = None) -> bytes:
+    """GET with Bearer auth; follows redirects WITHOUT auth (for S3/CDN blobs)."""
+    req = urllib.request.Request(url, headers={"Authorization": f"Bearer {token}", **(headers or {})})
+    with _opener().open(req, timeout=60) as r:
+        return r.read()
+
+
+def pull_image(image: str, output_tar: str) -> None:
+    if ":" in image:
+        repo_name, tag = image.rsplit(":", 1)
+    else:
+        repo_name, tag = image, "latest"
+
+    if "/" not in repo_name:
+        repo_name = f"library/{repo_name}"
+
+    print(f"Authenticating for {repo_name}:{tag} ...")
+    token = get_token(repo_name)
+
+    # Fetch manifest (prefer OCI, fall back to v2 schema2)
+    print("Fetching manifest ...")
+    manifest_url = f"https://registry-1.docker.io/v2/{repo_name}/manifests/{tag}"
+    manifest_bytes = registry_get(
+        manifest_url,
+        token,
+        headers={"Accept": "application/vnd.docker.distribution.manifest.v2+json,"
+                            "application/vnd.oci.image.manifest.v1+json"},
+    )
+    manifest = json.loads(manifest_bytes)
+
+    # Handle manifest list (multi-arch) — pick linux/amd64
+    media_type = manifest.get("mediaType", "") or manifest.get("schemaVersion", "")
+    if "list" in str(media_type) or manifest.get("manifests"):
+        print("Manifest list detected — selecting linux/amd64 ...")
+        for m in manifest["manifests"]:
+            plat = m.get("platform", {})
+            if plat.get("os") == "linux" and plat.get("architecture") == "amd64":
+                digest = m["digest"]
+                manifest_bytes = registry_get(
+                    f"https://registry-1.docker.io/v2/{repo_name}/manifests/{digest}",
+                    token,
+                    headers={"Accept": "application/vnd.docker.distribution.manifest.v2+json"},
+                )
+                manifest = json.loads(manifest_bytes)
+                break
+
+    config_digest = manifest["config"]["digest"]
+    layers = manifest["layers"]
+
+    with tempfile.TemporaryDirectory() as tmpdir:
+        # Download config blob
+        print("Downloading config ...")
+        config_data = registry_get(
+            f"https://registry-1.docker.io/v2/{repo_name}/blobs/{config_digest}",
+            token,
+        )
+        config_filename = config_digest.replace("sha256:", "") + ".json"
+        config_path = os.path.join(tmpdir, config_filename)
+        with open(config_path, "wb") as f:
+            f.write(config_data)
+
+        # Download each layer
+        layer_dirs = []
+        for i, layer in enumerate(layers):
+            digest = layer["digest"]
+            size = layer["size"]
+            short = digest[7:19]
+            print(f"Downloading layer {i+1}/{len(layers)} ({short}..., {size//1024//1024}MB) ...")
+
+            blob_url = f"https://registry-1.docker.io/v2/{repo_name}/blobs/{digest}"
+            req = urllib.request.Request(blob_url, headers={"Authorization": f"Bearer {token}"})
+
+            layer_dir = os.path.join(tmpdir, f"layer_{i}")
+            os.makedirs(layer_dir)
+            layer_tar = os.path.join(layer_dir, "layer.tar")
+            version_file = os.path.join(layer_dir, "VERSION")
+            json_file = os.path.join(layer_dir, "json")
+
+            # Stream download with Range-request chunking so a TCP corruption
+            # only loses one 2MB chunk, not the whole download.
+            CHUNK_SIZE = 2 * 1024 * 1024  # 2MB per Range request
+            downloaded = 0
+            with open(layer_tar, "wb") as f:
+                while downloaded < size:
+                    end = min(downloaded + CHUNK_SIZE - 1, size - 1)
+                    while True:
+                        try:
+                            range_req = urllib.request.Request(
+                                blob_url,
+                                headers={
+                                    "Authorization": f"Bearer {token}",
+                                    "Range": f"bytes={downloaded}-{end}",
+                                },
+                            )
+                            with _opener().open(range_req, timeout=60) as resp:
+                                data = resp.read()
+                            break
+                        except Exception as exc:
+                            print(f"\r  retry at {downloaded//1024//1024}MB ({exc})...", end="", flush=True)
+                            import time; time.sleep(1)
+                    f.write(data)
+                    downloaded += len(data)
+                    pct = downloaded * 100 // size if size else 0
+                    print(f"\r  {downloaded//1024//1024}MB / {size//1024//1024}MB ({pct}%)", end="", flush=True)
+            print()
+
+            with open(version_file, "w") as f:
+                f.write("1.0")
+            with open(json_file, "w") as f:
+                json.dump({"id": digest.replace("sha256:", "")}, f)
+
+            layer_dirs.append(f"layer_{i}/layer.tar")
+
+        # Write manifest.json
+        manifest_json = [
+            {
+                "Config": config_filename,
+                "RepoTags": [f"{repo_name.replace('library/', '')}:{tag}"],
+                "Layers": layer_dirs,
+            }
+        ]
+        manifest_path = os.path.join(tmpdir, "manifest.json")
+        with open(manifest_path, "w") as f:
+            json.dump(manifest_json, f)
+
+        # Bundle into tar
+        print(f"Building {output_tar} ...")
+        with tarfile.open(output_tar, "w") as tar:
+            for name in [config_filename, "manifest.json"]:
+                tar.add(os.path.join(tmpdir, name), arcname=name)
+            for i in range(len(layers)):
+                for fname in ["layer.tar", "VERSION", "json"]:
+                    path = os.path.join(tmpdir, f"layer_{i}", fname)
+                    tar.add(path, arcname=f"layer_{i}/{fname}")
+
+    print(f"Done. Load with: docker load -i {output_tar}")
+
+
+if __name__ == "__main__":
+    if len(sys.argv) < 2:
+        print("Usage: pull_image.py <image:tag> [output.tar]")
+        sys.exit(1)
+    image = sys.argv[1]
+    output = sys.argv[2] if len(sys.argv) > 2 else image.replace(":", "_").replace("/", "_") + ".tar"
+    pull_image(image, output)
--- a/scripts/seed.py
+++ b/scripts/seed.py
@@ -0,0 +1,97 @@
+"""Seed the 6 canonical topics from canon/projects/."""
+import asyncio
+import sys
+from pathlib import Path
+
+# Allow running from state-hub/ root
+sys.path.insert(0, str(Path(__file__).parent.parent))
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from api.database import async_session_factory, engine
+from api.models.topic import Domain, Topic, TopicStatus
+
+TOPICS = [
+    {
+        "slug": "custodian",
+        "title": "The Custodian",
+        "description": (
+            "Master agent system: transgenerational cognitive infrastructure for "
+            "co-creating and stewarding knowledge across all domains."
+        ),
+        "domain": Domain.custodian,
+    },
+    {
+        "slug": "railiance",
+        "title": "Railiance",
+        "description": (
+            "DevOps & infrastructure reliability. Dependency for all other projects; "
+            "provides the deployment and operational backbone."
+        ),
+        "domain": Domain.railiance,
+    },
+    {
+        "slug": "markitect",
+        "title": "Markitect",
+        "description": (
+            "Knowledge artifact management: structured authoring, versioning, and "
+            "retrieval of canonical documents."
+        ),
+        "domain": Domain.markitect,
+    },
+    {
+        "slug": "coulomb-social",
+        "title": "Coulomb.social",
+        "description": (
+            "Co-creation marketplace experiment: connecting people around shared "
+            "projects and complementary capabilities."
+        ),
+        "domain": Domain.coulomb_social,
+    },
+    {
+        "slug": "personhood",
+        "title": "Personhood",
+        "description": (
+            "Rights and obligations framework: defining digital personhood, consent "
+            "models, and data sovereignty."
+        ),
+        "domain": Domain.personhood,
+    },
+    {
+        "slug": "foerster-capabilities",
+        "title": "Foerster Capabilities",
+        "description": (
+            "Agency capability taxonomy inspired by Heinz von Foerster: mapping the "
+            "space of possible cognitive and social actions."
+        ),
+        "domain": Domain.foerster_capabilities,
+    },
+]
+
+
+async def seed() -> None:
+    async with async_session_factory() as session:
+        for data in TOPICS:
+            existing = await session.execute(
+                select(Topic).where(Topic.slug == data["slug"])
+            )
+            if existing.scalar_one_or_none() is not None:
+                print(f"  skip (already exists): {data['slug']}")
+                continue
+            topic = Topic(
+                slug=data["slug"],
+                title=data["title"],
+                description=data["description"],
+                domain=data["domain"],
+                status=TopicStatus.active,
+            )
+            session.add(topic)
+            print(f"  insert: {data['slug']}")
+        await session.commit()
+    await engine.dispose()
+    print("Seed complete.")
+
+
+if __name__ == "__main__":
+    asyncio.run(seed())
--- a/uv.lock
+++ b/uv.lock