feat(capability-registry): CUST-WP-0031 domain capability registry

- Migration p3k4l5m6n7o8: nullable repo_id FK on capability_catalog
- PATCH /capability-catalog/{id} endpoint for back-filling repo attribution
- register_capability MCP tool accepts optional repo_slug
- get_domain_summary now includes compact capabilities list (type+title+repo_slug)
- New get_capability_profile MCP tool: domain → repos → capabilities tree
- 6 repo descriptions populated; 25 catalog entries attributed to repos
- 9 new capabilities registered for personhood, foerster_capabilities, coulomb_social
- TOOLS.md: Capability Catalog & Requests section with full tool reference

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

state-hub/api/models/capability_catalog.py

@@ -22,6 +22,12 @@ class CapabilityCatalog(Base, TimestampMixin):
         nullable=False,
         index=True,
     )
+    repo_id: Mapped[uuid.UUID | None] = mapped_column(
+        UUID(as_uuid=True),
+        ForeignKey("managed_repos.id", ondelete="SET NULL"),
+        nullable=True,
+        index=True,
+    )
     capability_type: Mapped[str] = mapped_column(String(50), nullable=False)
     title: Mapped[str] = mapped_column(String(255), nullable=False)
     description: Mapped[str | None] = mapped_column(Text, nullable=True)
@@ -33,7 +39,12 @@ class CapabilityCatalog(Base, TimestampMixin):
     )
 
     domain: Mapped["Domain"] = relationship("Domain", lazy="selectin")  # noqa: F821
+    repo: Mapped["ManagedRepo | None"] = relationship("ManagedRepo", lazy="selectin")  # noqa: F821
 
     @property
     def domain_slug(self) -> str:
         return self.domain.slug if self.domain is not None else ""
+
+    @property
+    def repo_slug(self) -> str | None:
+        return self.repo.slug if self.repo is not None else None

state-hub/api/routers/capability_requests.py

@@ -11,9 +11,11 @@ from api.models.agent_message import AgentMessage
 from api.models.capability_catalog import CapabilityCatalog
 from api.models.capability_request import CapabilityRequest
 from api.models.domain import Domain
+from api.models.managed_repo import ManagedRepo
 from api.models.task import Task
 from api.schemas.capability_request import (
     CatalogCreate,
+    CatalogPatch,
     CatalogRead,
     CapabilityRequestAccept,
     CapabilityRequestCreate,
@@ -52,8 +54,15 @@ async def create_catalog_entry(
     session: AsyncSession = Depends(get_session),
 ) -> CapabilityCatalog:
     domain = await _resolve_domain(body.domain, session)
+
+    repo_id = None
+    if body.repo_slug:
+        repo = await _resolve_repo(body.repo_slug, session)
+        repo_id = repo.id
+
     entry = CapabilityCatalog(
         domain_id=domain.id,
+        repo_id=repo_id,
         capability_type=body.capability_type,
         title=body.title,
         description=body.description,
@@ -72,6 +81,31 @@ async def create_catalog_entry(
     return entry
 
 
+@router.patch("/capability-catalog/{entry_id}", response_model=CatalogRead)
+async def patch_catalog_entry(
+    entry_id: uuid.UUID,
+    body: CatalogPatch,
+    session: AsyncSession = Depends(get_session),
+) -> CapabilityCatalog:
+    entry = await session.get(CapabilityCatalog, entry_id)
+    if entry is None:
+        raise HTTPException(status_code=404, detail=f"Catalog entry '{entry_id}' not found")
+
+    if body.repo_slug is not None:
+        repo = await _resolve_repo(body.repo_slug, session)
+        entry.repo_id = repo.id
+    if body.description is not None:
+        entry.description = body.description
+    if body.keywords is not None:
+        entry.keywords = body.keywords
+    if body.status is not None:
+        entry.status = body.status
+
+    await session.commit()
+    await session.refresh(entry)
+    return entry
+
+
 @router.get("/capability-catalog/", response_model=list[CatalogRead])
 async def list_catalog(
     domain: str | None = Query(None),
@@ -552,6 +586,14 @@ async def _resolve_domain(slug: str, session: AsyncSession) -> Domain:
     return domain
 
 
+async def _resolve_repo(slug: str, session: AsyncSession) -> ManagedRepo:
+    result = await session.execute(select(ManagedRepo).where(ManagedRepo.slug == slug))
+    repo = result.scalar_one_or_none()
+    if repo is None:
+        raise HTTPException(status_code=404, detail=f"Repo '{slug}' not found")
+    return repo
+
+
 async def _get_request_or_404(request_id: uuid.UUID, session: AsyncSession) -> CapabilityRequest:
     req = await session.get(CapabilityRequest, request_id)
     if req is None:

state-hub/api/schemas/capability_request.py

@@ -14,6 +14,14 @@ class CatalogCreate(BaseModel):
     title: str
     description: str | None = None
     keywords: list[str] = []
+    repo_slug: str | None = None  # optional repo attribution
+
+
+class CatalogPatch(BaseModel):
+    repo_slug: str | None = None
+    description: str | None = None
+    keywords: list[str] | None = None
+    status: str | None = None
 
 
 class CatalogRead(BaseModel):
@@ -21,6 +29,8 @@ class CatalogRead(BaseModel):
 
     id: uuid.UUID
     domain_slug: str
+    repo_id: uuid.UUID | None = None
+    repo_slug: str | None = None
     capability_type: str
     title: str
     description: str | None = None

state-hub/mcp_server/TOOLS.md

@@ -24,13 +24,15 @@ Do not use them as a substitute for formal work definition inside the domain rep
 
 | Tool | Key Args | When to use |
 |------|----------|-------------|
-| `get_domain_summary(domain_slug)` | `domain_slug`: e.g. `"railiance"` | **Domain session start.** Scoped snapshot: active workstreams, blocking decisions, last 5 events, repo SBOM status — ~10% of get_state_summary() token cost. |
+| `get_domain_summary(domain_slug)` | `domain_slug`: e.g. `"railiance"` | **Domain session start.** Scoped snapshot: active workstreams, blocking decisions, last 5 events, repo SBOM status, compact capabilities list — ~10% of get_state_summary() token cost. |
 | `get_state_summary()` | — | **Cross-domain work / custodian sessions.** Full snapshot: totals, all blocking decisions, all blocked tasks, all open workstreams, last 20 events. Large (~10k tokens). |
 | `get_topic(slug)` | `slug`: e.g. `"markitect"` | Deep-dive on one topic + its workstreams + recent events. |
 | `list_tasks(workstream_id, status?)` | `workstream_id`: UUID (required); `status?`: todo/in_progress/blocked/done/cancelled | List all tasks in a workstream. Use this to look up task UUIDs before calling `update_task_status`, or to verify which workplan tasks are already synced to the DB. |
 | `list_blocked_tasks(workstream_id?)` | optional filter | Surface all impediments, optionally scoped to one workstream. |
 | `list_pending_decisions(topic_id?)` | optional filter | Decisions holding up work, sorted by deadline. |
 | `get_recent_progress(limit, since?)` | `limit` default 20; `since` ISO datetime | Reconstruct recent session history. |
+| `get_capability_profile(domain_slug?)` | `domain_slug`: optional domain slug | **Capability deep-dive.** Returns repos → capabilities tree for one domain or all active domains. Includes descriptions and keywords. For cross-domain architectural discussion or when a worker needs to understand what a domain provides without checking out its repos. |
+| `list_capabilities(domain?, capability_type?)` | optional filters | Browse the capability catalog entries. Returns full records including keywords. |
 
 ---
 
@@ -222,6 +224,36 @@ curl -X POST http://127.0.0.1:8000/repos/marki-docx/paths/ \
 
 ---
 
+## Capability Catalog & Requests
+
+Capabilities describe what each domain/repo can provide. Use the catalog for routing; use requests for cross-domain work coordination.
+
+### Catalog tools
+
+| Tool | Key Args | Notes |
+|------|----------|-------|
+| `register_capability(domain, capability_type, title, ...)` | `domain`: slug; `capability_type`: infrastructure/api/data/security/governance/documentation; `keywords?`; `description?`; `repo_slug?` | Add a capability to the catalog. Provide `repo_slug` to attribute it to a specific repo within the domain. |
+| `list_capabilities(domain?, capability_type?)` | optional filters | Browse active catalog entries with full detail (description + keywords). |
+| `get_capability_profile(domain_slug?)` | optional domain slug | **Deep-dive.** Returns domain → repos → capabilities tree with descriptions and keywords. Single domain or all active domains. Use when a worker needs to understand what a domain provides without checking out its repos. |
+
+### Request tools
+
+| Tool | Key Args | When to use |
+|------|----------|-------------|
+| `request_capability(title, capability_type, requesting_domain, requesting_agent, ...)` | `priority?`; `description?`; `requesting_workstream_id?`; `blocking_task_id?` | Ask another domain to provide a capability. Auto-routes to best matching catalog entry. |
+| `list_capability_requests(domain?, status?, capability_type?)` | optional filters | List open requests — filter by your domain to see what you must fulfill. |
+| `get_capability_request(request_id)` | `request_id`: UUID | Full detail on a single request. |
+| `accept_capability_request(request_id, fulfilling_agent, ...)` | `fulfilling_workstream_id?` | Accept a request routed to your domain. Notifies requester. |
+| `update_capability_request_status(request_id, status, note?)` | `status`: in_progress/ready_for_review/completed/rejected/withdrawn | Advance request lifecycle. `completed` auto-unblocks the linked task. |
+| `dispute_capability_routing(request_id, reason, disputed_by, suggested_domain?)` | all required except `suggested_domain` | Flag incorrect routing. Notifies custodian for re-routing. |
+| `reroute_capability_request(request_id, rerouted_by, note, domain?, catalog_entry_id?)` | one of `domain`/`catalog_entry_id` required | Re-route a disputed request to the correct domain. |
+| `patch_capability_request(request_id, ...)` | `catalog_entry_id?`; `priority?`; `blocking_task_id?`; `fulfilling_workstream_id?` | Correct mutable metadata on a request. |
+
+**Request status flow:** `requested` → `accepted` → `in_progress` → `ready_for_review` → `completed`
+Dispute path: `requested` → `routing_disputed` → `requested` (after re-route)
+
+---
+
 ## Domain Slugs
 
 Run `list_domains()` to get the live list. Default 6: `custodian` · `railiance` · `markitect` · `coulomb_social` · `personhood` · `foerster_capabilities`

state-hub/mcp_server/server.py

@@ -258,6 +258,18 @@ def get_domain_summary(domain_slug: str) -> str:
     }
     if goal_guidance:
         result["goal_guidance"] = goal_guidance
+
+    # Compact capabilities list (type + title + repo_slug only, capped at 20)
+    caps_raw = _get("/capability-catalog/", {"domain": domain_slug, "status": "active"})
+    if isinstance(caps_raw, list):
+        compact_caps = [
+            {"type": c["capability_type"], "title": c["title"], "repo_slug": c.get("repo_slug")}
+            for c in caps_raw[:20]
+        ]
+        result["capabilities"] = compact_caps
+        if len(caps_raw) > 20:
+            result["capabilities_truncated"] = True
+
     return json.dumps(result, indent=2)
 
 
@@ -1818,6 +1830,7 @@ def register_capability(
     title: str,
     description: str | None = None,
     keywords: list[str] | None = None,
+    repo_slug: str | None = None,
 ) -> str:
     """Register a capability that a domain can provide. Used for routing requests.
 
@@ -1827,6 +1840,7 @@ def register_capability(
         title: Short title for this capability
         description: Longer description (optional)
         keywords: List of keywords for routing (e.g. ['cluster', 'k8s', 'privacy'])
+        repo_slug: Optional repo slug to attribute this capability to a specific repo
     """
     entry = _post("/capability-catalog", {
         "domain": domain,
@@ -1834,6 +1848,7 @@ def register_capability(
         "title": title,
         "description": description,
         "keywords": keywords or [],
+        "repo_slug": repo_slug,
     })
     return json.dumps(entry, indent=2)
 
@@ -1855,6 +1870,87 @@ def list_capabilities(
     }), indent=2)
 
 
+@mcp.tool()
+def get_capability_profile(domain_slug: str | None = None) -> str:
+    """Full capability registry: domain → repos (with description) → capabilities.
+
+    Designed for deep-dive or cross-domain architectural discussion.
+
+    Args:
+        domain_slug: If provided, return profile for that one domain only.
+                     If omitted, return profiles for all active domains.
+
+    Returns a structured dict with repos nested under each domain, and
+    capabilities nested under each repo. Domain-level capabilities
+    (no repo assigned) appear under a synthetic entry with slug=null.
+    """
+    if domain_slug:
+        domain_slugs = [domain_slug]
+    else:
+        domains_raw = _get("/domains/")
+        if isinstance(domains_raw, dict) and "error" in domains_raw:
+            return json.dumps(domains_raw, indent=2)
+        domain_slugs = [d["slug"] for d in domains_raw if d.get("status") == "active"]
+
+    # Fetch topics once for title lookup
+    topics_raw = _get("/topics/")
+
+    profiles = []
+    for slug in domain_slugs:
+        repos_raw = _get("/repos/", {"domain": slug})
+        caps_raw = _get("/capability-catalog/", {"domain": slug, "status": "active"})
+        if isinstance(caps_raw, dict) and "error" in caps_raw:
+            caps_raw = []
+
+        # Index capabilities by repo_slug (None → domain-level)
+        caps_by_repo_slug: dict[str | None, list] = {}
+        for cap in caps_raw:
+            r_slug = cap.get("repo_slug")
+            caps_by_repo_slug.setdefault(r_slug, []).append({
+                "type": cap["capability_type"],
+                "title": cap["title"],
+                "description": cap.get("description"),
+                "keywords": cap.get("keywords", []),
+            })
+
+        repo_entries = []
+        if isinstance(repos_raw, list):
+            for repo in repos_raw:
+                repo_entries.append({
+                    "slug": repo["slug"],
+                    "name": repo["name"],
+                    "description": repo.get("description"),
+                    "capabilities": caps_by_repo_slug.get(repo["slug"], []),
+                })
+
+        # Domain-level caps (no repo assigned)
+        domain_level = caps_by_repo_slug.get(None, [])
+        if domain_level:
+            repo_entries.append({
+                "slug": None,
+                "name": "(domain-level)",
+                "description": None,
+                "capabilities": domain_level,
+            })
+
+        # Get topic title
+        topic_title = ""
+        if isinstance(topics_raw, list):
+            topic = next((t for t in topics_raw if t.get("domain_slug") == slug), None)
+            if topic:
+                topic_title = topic.get("title", "")
+
+        profiles.append({
+            "slug": slug,
+            "title": topic_title,
+            "repos": repo_entries,
+        })
+
+    if domain_slug and profiles:
+        return json.dumps(profiles[0], indent=2)
+    return json.dumps(profiles, indent=2)
+
+
 @mcp.tool()
 def request_capability(
     title: str,

state-hub/migrations/versions/p3k4l5m6n7o8_add_repo_id_to_capability_catalog.py

@@ -0,0 +1,28 @@
+"""add repo_id to capability_catalog
+
+Revision ID: p3k4l5m6n7o8
+Revises: o2j3k4l5m6n7
+Create Date: 2026-03-31
+
+"""
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects.postgresql import UUID
+
+revision = "p3k4l5m6n7o8"
+down_revision = "o2j3k4l5m6n7"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "capability_catalog",
+        sa.Column("repo_id", UUID(as_uuid=True), sa.ForeignKey("managed_repos.id", ondelete="SET NULL"), nullable=True),
+    )
+    op.create_index("ix_capability_catalog_repo_id", "capability_catalog", ["repo_id"])
+
+
+def downgrade() -> None:
+    op.drop_index("ix_capability_catalog_repo_id", table_name="capability_catalog")
+    op.drop_column("capability_catalog", "repo_id")

workplans/CUST-WP-0031-capability-registry.md

@@ -4,7 +4,7 @@ type: workplan
 title: "Domain Capability Registry"
 domain: custodian
 repo: the-custodian
-status: active
+status: done
 owner: custodian
 topic_slug: custodian
 created: "2026-03-31"
@@ -81,7 +81,7 @@ existing catalog entries; register capabilities for the three empty domains.
 ```task
 id: T01
 title: "Add repo_id FK to CapabilityCatalog"
-status: todo
+status: done
 priority: high
 description: >
   1. Write Alembic migration: add `repo_id` UUID nullable FK column to
@@ -103,7 +103,7 @@ state_hub_task_id: "8d5e3e37-c753-4cdc-9211-83ee39f6b0f2"
 ```task
 id: T02
 title: "Include compact capabilities in get_domain_summary"
-status: todo
+status: done
 priority: high
 description: >
   Extend the `get_domain_summary` MCP tool response to include a
@@ -122,7 +122,7 @@ state_hub_task_id: "ac47994c-efe9-404f-8065-9adf2e923d4c"
 ```task
 id: T03
 title: "New MCP tool: get_capability_profile"
-status: todo
+status: done
 priority: high
 description: >
   Add a new MCP tool `get_capability_profile(domain_slug: str | None = None)`
@@ -159,7 +159,7 @@ state_hub_task_id: "b380fd2b-28fa-4c47-96d6-4c65a0300c44"
 ```task
 id: T04
 title: "Populate missing repo descriptions"
-status: todo
+status: done
 priority: medium
 description: >
   Use PATCH /repos/{slug}/ to set `description` for the 8 repos that
@@ -184,7 +184,7 @@ state_hub_task_id: "00d44110-fcb4-45cc-8bc8-454af2629d2f"
 ```task
 id: T05
 title: "Back-fill repo_id on existing capability catalog entries"
-status: todo
+status: done
 priority: medium
 description: >
   After T01 lands, update the 25 existing catalog entries to set repo_id
@@ -221,7 +221,7 @@ state_hub_task_id: "7f0748c7-bdee-4801-b870-d4940a5a2e63"
 ```task
 id: T06
 title: "Register capabilities for personhood, foerster_capabilities, coulomb_social"
-status: todo
+status: done
 priority: medium
 description: >
   Register at least 3 capabilities per missing domain using
@@ -250,7 +250,7 @@ state_hub_task_id: "c6e1be52-961c-4e34-96b1-e450d64298df"
 ```task
 id: T07
 title: "Consistency gate and smoke test"
-status: todo
+status: done
 priority: low
 description: >
   1. Run `make test` — all existing tests must pass.