Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 14 additions & 0 deletions .console/log.md
Original file line number Diff line number Diff line change
@@ -1,5 +1,19 @@
# Log

## 2026-06-16 — feat: Fleet Capabilities section in startup context (capability plane Phase 1)

First consumer of the capability registry. `build_resume_prompt` now appends a
**Fleet Capabilities** section to the compiled `.console/.context`, so an
operator/agent sees what the fleet can do (owner, scope, risk, lane) before
acting. Reads PlatformManifest's `capabilities.yaml` **directly via PyYAML** —
no `platform_manifest`/`repograph` import, no RepoGraph compile — because the
flat authoring YAML already carries owner_repo_id/target_scope/risk/routing as
fields. Located in-repo when anchored at PlatformManifest, else a sibling
PlatformManifest checkout. Fail-soft: missing/malformed registry → section
omitted, context compilation never blocked. Private (`visibility != public`)
capabilities are never surfaced. This is the read-model legibility consumer the
plane was built for; routing stays descriptive (no execution wiring).

## 2026-06-15 — chore: cwd-safe ContextGuard hook command

Hardened `.claude/settings.json` hook commands to
Expand Down
98 changes: 98 additions & 0 deletions src/operator_console/bootstrap.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,8 @@
from datetime import datetime
from pathlib import Path

import yaml

# Ordered sections in the context — label maps to filename
CONTEXT_SECTIONS = [
("task.md", "Task"),
Expand All @@ -18,6 +20,17 @@
("log.md", "Log"),
]

# Capability registry (read-model) — the fleet's "what can it DO" catalog lives
# in PlatformManifest. The console surfaces it into startup context so an
# operator/agent sees the fleet's owned capabilities before acting. We read the
# flat authoring YAML DIRECTLY (PyYAML, no platform_manifest/repograph import) —
# the file carries owner_repo_id / target_scope / risk / routing.preferred_lane
# as authored fields, so no graph compile is needed. Located relative to the
# anchored repo: the repo itself when anchored at PlatformManifest, else a
# sibling PlatformManifest checkout. Read-only legibility; never executes.
_CAPABILITIES_REL = Path("src/platform_manifest/data/capabilities.yaml")
_PLATFORM_MANIFEST_DIRNAME = "PlatformManifest"

# Hot-trim (tiered-memory spec §5): the log grows without bound, so compiling it
# whole bloats the always-loaded startup blob (seen at 3k–6k lines in active
# repos). Compile only the most-recent entries here; the full history stays in
Expand Down Expand Up @@ -105,6 +118,87 @@ def _get_branch(repo_root: Path) -> str:
return "unknown"


def _find_capabilities_file(repo_root: Path) -> Path | None:
"""Locate the capability registry YAML. Anchored at PlatformManifest the file
is in-repo; anchored elsewhere look for a sibling PlatformManifest checkout.
Returns None when neither exists (fail-soft)."""
for path in (
repo_root / _CAPABILITIES_REL,
repo_root.parent / _PLATFORM_MANIFEST_DIRNAME / _CAPABILITIES_REL,
):
if path.is_file():
return path
return None


def _format_capability_scope(scope: dict) -> str:
"""Render a capability target_scope trichotomy compactly (repo / repo_set /
fleet) — mirrors the registry's locked target_scope contract."""
kind = (scope or {}).get("kind", "?")
if kind == "repo":
return f"repo({scope.get('repo_id', '?')})"
if kind == "repo_set":
sel = scope.get("selector") or {}
inner = ", ".join(f"{k}={v}" for k, v in sorted(sel.items())) or "*"
return f"repo_set({inner})"
return str(kind) # fleet (or unknown) — no id/selector by contract


def _render_capabilities_section(repo_root: Path) -> str | None:
"""Compile the PUBLIC capability catalog into a context section, grouped by
owning repo. Fail-soft: any missing file / parse error / unexpected shape
returns None so context compilation is never blocked.

Format contract (stable — tests rely on it):
## Fleet Capabilities

_What the fleet can do ..._

**<owner_repo_id>**
- <Name> — <scope> · <risk>[ · lane: <lane>]
"""
path = _find_capabilities_file(repo_root)
if path is None:
return None
try:
data = yaml.safe_load(path.read_text(encoding="utf-8")) or {}
caps = data.get("capabilities") or []
if not isinstance(caps, list):
return None
except Exception:
return None

by_owner: dict[str, list[str]] = {}
for cap in caps:
if not isinstance(cap, dict):
continue
if cap.get("visibility", "public") != "public":
continue # never surface private capabilities in console context
owner = cap.get("owner_repo_id", "unknown")
name = cap.get("name") or cap.get("action_id") or "unnamed"
scope = _format_capability_scope(cap.get("target_scope") or {})
risk = cap.get("risk", "unknown")
lane = (cap.get("routing") or {}).get("preferred_lane")
line = f"- {name} — {scope} · {risk}"
if lane:
line += f" · lane: {lane}"
by_owner.setdefault(owner, []).append(line)

if not by_owner:
return None

blocks = [
"**{}**\n{}".format(owner, "\n".join(sorted(by_owner[owner])))
for owner in sorted(by_owner)
]
return (
"## Fleet Capabilities\n\n"
"_What the fleet can do — read-model, CAP1-enforced. "
"Source: PlatformManifest capabilities.yaml. Read-only legibility._\n\n"
+ "\n\n".join(blocks)
)


def build_resume_prompt(
repo_root: Path,
files: list[str] | None = None,
Expand All @@ -131,6 +225,10 @@ def build_resume_prompt(
if content:
sections.append(f"## {label}\n\n{content}")

caps_section = _render_capabilities_section(repo_root)
if caps_section:
sections.append(caps_section)

if peer_roots:
for peer_name, peer_root in peer_roots:
peer_console = peer_root / ".console"
Expand Down
158 changes: 158 additions & 0 deletions tests/test_bootstrap_capabilities.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,158 @@
# SPDX-License-Identifier: AGPL-3.0-or-later
# Copyright (C) 2026 ProtocolWarden
"""Tests for the Fleet Capabilities context section (bootstrap capability
read-model consumer). The console reads PlatformManifest's capabilities.yaml
directly (PyYAML, no platform_manifest/repograph import) and renders a grouped,
fail-soft legibility section into the compiled startup context."""

from __future__ import annotations

from pathlib import Path

from operator_console.bootstrap import (
_find_capabilities_file,
_format_capability_scope,
_render_capabilities_section,
build_resume_prompt,
)

_REL = Path("src/platform_manifest/data/capabilities.yaml")

_REGISTRY = """schema_kind: capabilities
schema_version: 1.0.0
capabilities:
- action_id: repo_health_audit
name: Repo Health Audit
owner_repo_id: custodian
target_scope:
kind: repo_set
selector:
visibility: public
risk: read_only
visibility: public
- action_id: board_unblock
name: Board Unblock
owner_repo_id: operations_center
target_scope:
kind: fleet
risk: mutates_fleet
routing:
preferred_lane: maintenance
visibility: public
"""


def _write_registry(root: Path, text: str = _REGISTRY) -> Path:
path = root / _REL
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(text, encoding="utf-8")
return path


# --- locator -----------------------------------------------------------------

def test_finds_registry_in_repo(tmp_path: Path) -> None:
_write_registry(tmp_path)
assert _find_capabilities_file(tmp_path) == tmp_path / _REL


def test_finds_registry_in_sibling_platform_manifest(tmp_path: Path) -> None:
# Anchored at some OTHER repo; PlatformManifest is a sibling checkout.
pm = tmp_path / "PlatformManifest"
_write_registry(pm)
other = tmp_path / "OperatorConsole"
other.mkdir()
assert _find_capabilities_file(other) == pm / _REL


def test_missing_registry_returns_none(tmp_path: Path) -> None:
assert _find_capabilities_file(tmp_path) is None


# --- scope rendering (locked trichotomy) -------------------------------------

def test_scope_repo() -> None:
assert _format_capability_scope({"kind": "repo", "repo_id": "custodian"}) == "repo(custodian)"


def test_scope_repo_set() -> None:
assert _format_capability_scope(
{"kind": "repo_set", "selector": {"visibility": "public"}}
) == "repo_set(visibility=public)"


def test_scope_fleet() -> None:
assert _format_capability_scope({"kind": "fleet"}) == "fleet"


# --- section rendering -------------------------------------------------------

def test_section_groups_by_owner_sorted_with_optional_lane(tmp_path: Path) -> None:
_write_registry(tmp_path)
section = _render_capabilities_section(tmp_path)
assert section is not None
assert "## Fleet Capabilities" in section
# grouped by owner, owners sorted (custodian before operations_center)
assert section.index("**custodian**") < section.index("**operations_center**")
# read_only cap has no routing → no lane suffix
assert "Repo Health Audit — repo_set(visibility=public) · read_only" in section
assert "Repo Health Audit — repo_set(visibility=public) · read_only · lane" not in section
# fleet cap with routing → lane shown
assert "Board Unblock — fleet · mutates_fleet · lane: maintenance" in section


def test_private_capability_excluded(tmp_path: Path) -> None:
_write_registry(tmp_path, _REGISTRY + """ - action_id: secret_op
name: Secret Op
owner_repo_id: video_foundry
target_scope:
kind: fleet
risk: read_only
visibility: private
""")
section = _render_capabilities_section(tmp_path)
assert section is not None
assert "Secret Op" not in section
assert "video_foundry" not in section


# --- fail-soft (never block context compilation) -----------------------------

def test_no_registry_returns_none(tmp_path: Path) -> None:
assert _render_capabilities_section(tmp_path) is None


def test_malformed_yaml_returns_none(tmp_path: Path) -> None:
_write_registry(tmp_path, "capabilities: [unclosed\n - : :")
assert _render_capabilities_section(tmp_path) is None


def test_capabilities_not_a_list_returns_none(tmp_path: Path) -> None:
_write_registry(tmp_path, "schema_kind: capabilities\ncapabilities: not-a-list\n")
assert _render_capabilities_section(tmp_path) is None


def test_empty_capabilities_returns_none(tmp_path: Path) -> None:
_write_registry(tmp_path, "schema_kind: capabilities\ncapabilities: []\n")
assert _render_capabilities_section(tmp_path) is None


# --- integration with the full context build ---------------------------------

def test_build_resume_prompt_includes_capabilities(tmp_path: Path) -> None:
console = tmp_path / ".console"
console.mkdir()
(console / "task.md").write_text("Do the thing", encoding="utf-8")
_write_registry(tmp_path)
prompt = build_resume_prompt(tmp_path)
assert "## Fleet Capabilities" in prompt
assert "Board Unblock" in prompt


def test_build_resume_prompt_safe_without_registry(tmp_path: Path) -> None:
console = tmp_path / ".console"
console.mkdir()
(console / "task.md").write_text("Do the thing", encoding="utf-8")
prompt = build_resume_prompt(tmp_path)
assert "## Fleet Capabilities" not in prompt
assert "Do the thing" in prompt
Loading