feat(handoffs): publish owner task interfaces

Assistant: codex
Assistant-Model: gpt-5.6-sol
Assistant-Session: 01a023c0-a0a3-7c03-b395-5a0d2757214d

src/repo_manager/cli.py

@@ -382,6 +382,21 @@ def main(argv: list[str] | None = None) -> int:
     p_workload_resolve.add_argument("--name", required=True)
     p_workload_resolve.add_argument("--deployable", default=None)
 
+    p_owner_interface = sub.add_parser(
+        "owner-interface",
+        help="Inspect validated owner-consumable task interfaces",
+    )
+    owner_interface_sub = p_owner_interface.add_subparsers(dest="owner_interface_command")
+    p_owner_interface_validate = owner_interface_sub.add_parser(
+        "validate", help="Validate and return owner task interfaces"
+    )
+    p_owner_interface_validate.add_argument(
+        "--path", default="interfaces", help="One interface YAML or an interface directory"
+    )
+    p_owner_interface_validate.add_argument(
+        "--owner", default=None, help="Return interfaces for this target repo or owner agent"
+    )
+
     args = parser.parse_args(argv)
 
     if args.version or args.command in (None, "version"):
@@ -847,6 +862,16 @@ def main(argv: list[str] | None = None) -> int:
         print(json.dumps(result, indent=2))
         return 0 if result.get("ok") else 1
 
+    if args.command == "owner-interface":
+        if not args.owner_interface_command:
+            p_owner_interface.print_help()
+            return 2
+        from repo_manager.owner_interfaces import validate_owner_interfaces
+
+        result = validate_owner_interfaces(Path(args.path), owner=args.owner)
+        print(json.dumps(result, indent=2))
+        return 0 if result.get("ok") else 1
+
     parser.print_help()
     return 0
 

src/repo_manager/owner_interfaces.py

@@ -0,0 +1,177 @@
+"""Validation and rendering for owner-consumable task interfaces."""
+
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any
+
+import yaml
+
+API_VERSION = "helixforge.repo-manager/v1"
+KIND = "OwnerTaskInterface"
+DISPOSITIONS = ("approved", "amended", "rejected")
+PRIORITIES = frozenset({"high", "medium", "low"})
+
+
+def _error(code: str, path: Path, message: str) -> dict[str, str]:
+    return {"code": code, "path": str(path), "message": message}
+
+
+def _text(value: Any) -> str | None:
+    if not isinstance(value, str):
+        return None
+    value = value.strip()
+    return value or None
+
+
+def _string_list(value: Any) -> list[str] | None:
+    if not isinstance(value, list) or not value:
+        return None
+    values = [_text(item) for item in value]
+    if any(item is None for item in values):
+        return None
+    return [item for item in values if item is not None]
+
+
+def _interface_paths(path: Path) -> list[Path]:
+    path = path.expanduser().resolve()
+    if path.is_file():
+        return [path]
+    return sorted(path.glob("*.yaml"))
+
+
+def validate_owner_interfaces(path: Path, *, owner: str | None = None) -> dict[str, Any]:
+    """Validate task interfaces and return their directly consumable content."""
+    path = path.expanduser().resolve()
+    errors: list[dict[str, str]] = []
+    interfaces: list[dict[str, Any]] = []
+    seen_ids: set[str] = set()
+
+    paths = _interface_paths(path)
+    if not paths:
+        errors.append(_error("interface_not_found", path, "no interface YAML files found"))
+
+    for interface_path in paths:
+        try:
+            raw = yaml.safe_load(interface_path.read_text(encoding="utf-8")) or {}
+        except (OSError, yaml.YAMLError) as exc:
+            errors.append(_error("interface_invalid", interface_path, str(exc)))
+            continue
+        if not isinstance(raw, dict):
+            errors.append(_error("interface_invalid", interface_path, "root must be a mapping"))
+            continue
+        if raw.get("apiVersion") != API_VERSION or raw.get("kind") != KIND:
+            errors.append(
+                _error(
+                    "interface_kind_invalid",
+                    interface_path,
+                    f"expected apiVersion {API_VERSION!r} and kind {KIND!r}",
+                )
+            )
+            continue
+
+        metadata = raw.get("metadata")
+        source = raw.get("source")
+        target = raw.get("target")
+        approval = raw.get("approval")
+        task = raw.get("task")
+        if not all(isinstance(item, dict) for item in (metadata, source, target, approval, task)):
+            errors.append(
+                _error(
+                    "interface_shape_invalid",
+                    interface_path,
+                    "metadata, source, target, approval, and task must be mappings",
+                )
+            )
+            continue
+
+        interface_id = _text(metadata.get("id"))
+        title = _text(metadata.get("title"))
+        source_repo = _text(source.get("repo"))
+        source_task = _text(source.get("workplan_task"))
+        target_repo = _text(target.get("repo"))
+        owner_agent = _text(target.get("owner_agent"))
+        task_title = _text(task.get("title"))
+        objective = _text(task.get("objective"))
+        priority = _text(task.get("priority"))
+        deliverables = _string_list(task.get("deliverables"))
+        acceptance = _string_list(task.get("acceptance"))
+        dispositions = approval.get("dispositions")
+
+        required = {
+            "metadata.id": interface_id,
+            "metadata.title": title,
+            "source.repo": source_repo,
+            "source.workplan_task": source_task,
+            "target.repo": target_repo,
+            "target.owner_agent": owner_agent,
+            "task.title": task_title,
+            "task.objective": objective,
+        }
+        missing = [name for name, value in required.items() if value is None]
+        if missing:
+            errors.append(
+                _error(
+                    "interface_required_field_missing",
+                    interface_path,
+                    "missing non-empty fields: " + ", ".join(missing),
+                )
+            )
+            continue
+        if priority not in PRIORITIES:
+            errors.append(
+                _error(
+                    "interface_priority_invalid",
+                    interface_path,
+                    "task.priority must be high, medium, or low",
+                )
+            )
+            continue
+        if deliverables is None or acceptance is None:
+            errors.append(
+                _error(
+                    "interface_task_invalid",
+                    interface_path,
+                    "task.deliverables and task.acceptance must be non-empty string lists",
+                )
+            )
+            continue
+        if dispositions != list(DISPOSITIONS):
+            errors.append(
+                _error(
+                    "interface_approval_invalid",
+                    interface_path,
+                    "approval.dispositions must be [approved, amended, rejected]",
+                )
+            )
+            continue
+        if "status" in raw or "status" in metadata:
+            errors.append(
+                _error(
+                    "interface_lifecycle_forbidden",
+                    interface_path,
+                    "an interface is not a work record and must not carry lifecycle status",
+                )
+            )
+            continue
+        if interface_id in seen_ids:
+            errors.append(
+                _error("interface_id_duplicate", interface_path, f"duplicate id {interface_id!r}")
+            )
+            continue
+        seen_ids.add(interface_id)
+
+        if owner is None or owner in {target_repo, owner_agent}:
+            document = dict(raw)
+            document["path"] = str(interface_path)
+            interfaces.append(document)
+
+    return {
+        "ok": not errors,
+        "apiVersion": API_VERSION,
+        "kind": KIND,
+        "root": str(path),
+        "interface_count": len(interfaces),
+        "interfaces": interfaces,
+        "errors": errors,
+    }