feat(plane): manifest schemas and validation (task 1)

Add openultrasast.plane.manifests: frozen dataclasses and loaders for ax's
Task, Workspace and Model kinds (ax.io/v1alpha1, ax's fields only) and the
Run kind (openultrasast.io/v1alpha1: tasks, dependsOn, inputs, outputs,
budget, serialize). Every violation raises ManifestError naming the kind,
metadata.name and field path: unknown fields at any depth, wrong apiVersion
or kind, a provider outside google/anthropic unless deepseek is declared via
the openultrasast.io/provider-extension annotation, a Run with a dependency
cycle, an input naming no task, and a Run task naming no loaded Task.
load_manifests reads multi-document YAML files and indexes by kind and name.

PyYAML becomes the one declared core dependency; the two zero-dependency
guards now pin exactly that, and the module audit registers the plane
package under the ai-service-plane spec.

Co-Authored-By: Claude Fable 5.1 <[email protected]>
Claude-Session: https://claude.ai/code/session_01EmM1j46qgKMh3zqD1QMwEo

benchmarks/measurements/2026-09-08-module-audit.json

@@ -1,7 +1,7 @@
 {
-  "modules": 118,
+  "modules": 120,
   "counts": {
-    "load_bearing": 109,
+    "load_bearing": 111,
     "standalone_capability": 9
   },
   "orphaned": [],
@@ -645,6 +645,18 @@
         "pair_gate"
       ]
     },
+    {
+      "module": "plane",
+      "classification": "load_bearing",
+      "reason": "ai-service-plane Req 1: the model-level service plane package; reconciler and runner land in tasks 3-4",
+      "importers": []
+    },
+    {
+      "module": "plane.manifests",
+      "classification": "load_bearing",
+      "reason": "ai-service-plane Req 1.1-1.4: ax-shaped Task/Workspace/Model and the Run kind, validated before anything runs",
+      "importers": []
+    },
     {
       "module": "policy",
       "classification": "load_bearing",

pyproject.toml

@@ -8,13 +8,13 @@ version = "1.2.0a1"
 description = "OpenCode-oriented SAST harness for evidence-gated security analysis"
 readme = "README.md"
 requires-python = ">=3.11"
-dependencies = []
+dependencies = ["pyyaml>=6.0"]
 
 [project.optional-dependencies]
 # HarnessX is the optional agentic plane (LLM hunter/verifier + self-improvement).
 # SHA-pinned for reproducibility; the heavy transitive footprint (see
 # .kiro/specs/harnessx-self-improving-rulesets/dependency-audit.md) lands ONLY when
-# this extra is installed. The core install stays zero-dependency and every HarnessX
+# this extra is installed. The core install declares only PyYAML (plane manifests) and every HarnessX
 # import is guarded by openultrasast.harness_ext (lazy). pip does not pull the
 # recipe/verl_harnessX submodule (it is outside the packaged harnessx* modules).
 harnessx = [
@@ -73,6 +73,11 @@ ignore_missing_imports = true
 module = "tree_sitter"
 ignore_missing_imports = true
 
+# PyYAML ships no inline types and types-PyYAML is not a dev dependency.
+[[tool.mypy.overrides]]
+module = "yaml"
+ignore_missing_imports = true
+
 [dependency-groups]
 dev = [
     "dagger-io>=0.21.7",

src/openultrasast/model/audit.py

@@ -54,6 +54,8 @@ SPEC_OWNED = {
     "cpg": "model-grounded-detection Req 4: the CPG seam package",
     "cpg.backend": "model-grounded-detection Req 4.1/4.5: the single Joern subprocess boundary",
     "cpg.capability": "model-grounded-detection Req 4.2/4.3: the engine probe that degrades to suspicion",
+    "plane": "ai-service-plane Req 1: the model-level service plane package; reconciler and runner land in tasks 3-4",
+    "plane.manifests": "ai-service-plane Req 1.1-1.4: ax-shaped Task/Workspace/Model and the Run kind, validated before anything runs",
 }
 
 STANDALONE = {

src/openultrasast/plane/__init__.py

@@ -0,0 +1 @@
+"""Model-level service plane: ax-shaped manifests, a local reconciler and the runner contract."""

src/openultrasast/plane/manifests.py

@@ -0,0 +1,542 @@
+"""Manifests of the model-level service plane.
+
+``Task``, ``Workspace`` and ``Model`` follow google/ax (``apiVersion: ax.io/v1alpha1``) and carry only the
+fields ax documents. ``Run`` (``apiVersion: openultrasast.io/v1alpha1``) is this project's kind for the task
+graph, artifacts and budgets; it holds nothing a task needs to execute. Every schema violation raises
+:class:`ManifestError`, whose message names the kind, the ``metadata.name`` and the offending field path
+(``Task/verify: spec.foo is not a field``), before anything runs.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Iterable, Mapping
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any
+
+import yaml
+
+__all__ = [
+    "AX_API_VERSION",
+    "AX_PROVIDERS",
+    "EXTENSION_PROVIDERS",
+    "PROVIDER_EXTENSION_ANNOTATION",
+    "RUN_API_VERSION",
+    "Budget",
+    "EnvVar",
+    "FileEntry",
+    "GitSource",
+    "ManifestError",
+    "Manifests",
+    "McpSpec",
+    "Metadata",
+    "Model",
+    "ResourceSpec",
+    "Resources",
+    "Run",
+    "RunTask",
+    "SecretKey",
+    "SkillsSpec",
+    "Task",
+    "Workspace",
+    "WorkspaceBinding",
+    "check_run_tasks",
+    "load_manifests",
+    "parse_manifest",
+]
+
+AX_API_VERSION = "ax.io/v1alpha1"
+RUN_API_VERSION = "openultrasast.io/v1alpha1"
+PROVIDER_EXTENSION_ANNOTATION = "openultrasast.io/provider-extension"
+AX_PROVIDERS = frozenset({"google", "anthropic"})
+EXTENSION_PROVIDERS = frozenset({"deepseek"})
+
+_API_VERSIONS = {"Task": AX_API_VERSION, "Workspace": AX_API_VERSION, "Model": AX_API_VERSION, "Run": RUN_API_VERSION}
+
+
+class ManifestError(ValueError):
+    """A manifest violates its schema; the message names kind, ``metadata.name`` and the field."""
+
+
+# --- dataclasses -----------------------------------------------------------------------------------------------
+
+
+@dataclass(frozen=True)
+class Metadata:
+    name: str
+    namespace: str | None = None
+    annotations: Mapping[str, str] = field(default_factory=dict)
+
+
+@dataclass(frozen=True)
+class EnvVar:
+    name: str
+    value: str
+
+
+@dataclass(frozen=True)
+class ResourceSpec:
+    cpu: str | None = None
+    memory: str | None = None
+
+
+@dataclass(frozen=True)
+class Resources:
+    requests: ResourceSpec | None = None
+    limits: ResourceSpec | None = None
+
+
+@dataclass(frozen=True)
+class WorkspaceBinding:
+    name: str
+    path: str
+    goal: str | None = None
+
+
+@dataclass(frozen=True)
+class Task:
+    metadata: Metadata
+    command: tuple[str, ...]
+    image: str | None = None
+    env: tuple[EnvVar, ...] = ()
+    resources: Resources | None = None
+    workspaces: tuple[WorkspaceBinding, ...] = ()
+    debug: bool = False
+
+
+@dataclass(frozen=True)
+class GitSource:
+    name: str
+    repo: str
+    branch: str | None = None
+    commit: str | None = None
+
+
+@dataclass(frozen=True)
+class FileEntry:
+    path: str
+    content: str
+
+
+@dataclass(frozen=True)
+class McpSpec:
+    registries: tuple[Any, ...] = ()
+    servers: tuple[Any, ...] = ()
+
+
+@dataclass(frozen=True)
+class SkillsSpec:
+    registries: tuple[Any, ...] = ()
+    path: str | None = None
+
+
+@dataclass(frozen=True)
+class Workspace:
+    metadata: Metadata
+    git: tuple[GitSource, ...] = ()
+    files: tuple[FileEntry, ...] = ()
+    mcp: McpSpec | None = None
+    skills: SkillsSpec | None = None
+
+
+@dataclass(frozen=True)
+class SecretKey:
+    name: str
+    key: str
+
+
+@dataclass(frozen=True)
+class Model:
+    metadata: Metadata
+    provider: str
+    model: str
+    secret_key: SecretKey | None = None
+    parameters: Mapping[str, Any] = field(default_factory=dict)
+
+
+@dataclass(frozen=True)
+class Budget:
+    usd: float | None = None
+    calls: int | None = None
+
+
+@dataclass(frozen=True)
+class RunTask:
+    name: str
+    task: str
+    depends_on: tuple[str, ...] = ()
+    inputs: Mapping[str, str] = field(default_factory=dict)
+    outputs: tuple[str, ...] = ()
+    budget: Budget | None = None
+    serialize: str | None = None
+
+    @property
+    def producers(self) -> tuple[str, ...]:
+        """Tasks this task waits for: ``dependsOn`` plus the producer of every input, in order, unique."""
+        seen: dict[str, None] = dict.fromkeys(self.depends_on)
+        for ref in self.inputs.values():
+            seen.setdefault(ref.split("/", 1)[0])
+        return tuple(seen)
+
+
+@dataclass(frozen=True)
+class Run:
+    metadata: Metadata
+    tasks: tuple[RunTask, ...]
+    artifacts: str | None = None
+
+    def task(self, name: str) -> RunTask:
+        for entry in self.tasks:
+            if entry.name == name:
+                return entry
+        raise KeyError(name)
+
+
+@dataclass(frozen=True)
+class Manifests:
+    """Loaded manifests by kind, each keyed by ``metadata.name``."""
+
+    tasks: dict[str, Task] = field(default_factory=dict)
+    workspaces: dict[str, Workspace] = field(default_factory=dict)
+    models: dict[str, Model] = field(default_factory=dict)
+    runs: dict[str, Run] = field(default_factory=dict)
+
+
+# --- field access with named failures ----------------------------------------------------------------------------
+
+
+def _join(path: str, key: str) -> str:
+    return key if not path else f"{path}.{key}"
+
+
+class _Check:
+    """Field access on one document; every failure names ``Kind/name: path rule``."""
+
+    def __init__(self, kind: str, name: str) -> None:
+        self.prefix = f"{kind}/{name}: "
+
+    def fail(self, path: str, rule: str) -> ManifestError:
+        return ManifestError(f"{self.prefix}{path} {rule}")
+
+    def mapping(self, value: object, path: str, allowed: Iterable[str]) -> dict[str, Any]:
+        if not isinstance(value, dict):
+            raise self.fail(path or "document", "must be a mapping")
+        allowed_set = frozenset(allowed)
+        for key in value:
+            if key not in allowed_set:
+                raise self.fail(_join(path, str(key)), "is not a field")
+        return value
+
+    def optional_mapping(self, parent: Mapping[str, Any], path: str, key: str, allowed: Iterable[str]) -> dict[str, Any] | None:
+        value = parent.get(key)
+        return None if value is None else self.mapping(value, _join(path, key), allowed)
+
+    def string(self, parent: Mapping[str, Any], path: str, key: str) -> str:
+        value = parent.get(key)
+        if value is None:
+            raise self.fail(_join(path, key), "is required")
+        if not isinstance(value, str) or not value:
+            raise self.fail(_join(path, key), "must be a non-empty string")
+        return value
+
+    def optional_string(self, parent: Mapping[str, Any], path: str, key: str) -> str | None:
+        return None if parent.get(key) is None else self.string(parent, path, key)
+
+    def sequence(self, parent: Mapping[str, Any], path: str, key: str) -> list[Any]:
+        value = parent.get(key)
+        if value is None:
+            return []
+        if not isinstance(value, list):
+            raise self.fail(_join(path, key), "must be a list")
+        return value
+
+    def strings(self, parent: Mapping[str, Any], path: str, key: str) -> tuple[str, ...]:
+        items = self.sequence(parent, path, key)
+        for index, item in enumerate(items):
+            if not isinstance(item, str) or not item:
+                raise self.fail(f"{_join(path, key)}[{index}]", "must be a non-empty string")
+        return tuple(items)
+
+    def mappings(self, parent: Mapping[str, Any], path: str, key: str, allowed: Iterable[str]) -> list[tuple[str, dict[str, Any]]]:
+        allowed_set = frozenset(allowed)
+        base = _join(path, key)
+        return [
+            (f"{base}[{i}]", self.mapping(item, f"{base}[{i}]", allowed_set)) for i, item in enumerate(self.sequence(parent, path, key))
+        ]
+
+    def string_map(self, parent: Mapping[str, Any], path: str, key: str) -> dict[str, str]:
+        value = parent.get(key)
+        if value is None:
+            return {}
+        if not isinstance(value, dict):
+            raise self.fail(_join(path, key), "must be a mapping")
+        for item_key, item in value.items():
+            if not isinstance(item_key, str) or not isinstance(item, str) or not item:
+                raise self.fail(_join(_join(path, key), str(item_key)), "must map a string to a non-empty string")
+        return value
+
+    def boolean(self, parent: Mapping[str, Any], path: str, key: str) -> bool:
+        value = parent.get(key, False)
+        if not isinstance(value, bool):
+            raise self.fail(_join(path, key), "must be true or false")
+        return value
+
+
+# --- per-kind parsers ----------------------------------------------------------------------------------------------
+
+
+def _parse_metadata(check: _Check, value: object) -> Metadata:
+    meta = check.mapping(value, "metadata", ("name", "namespace", "annotations"))
+    return Metadata(
+        name=check.string(meta, "metadata", "name"),
+        namespace=check.optional_string(meta, "metadata", "namespace"),
+        annotations=check.string_map(meta, "metadata", "annotations"),
+    )
+
+
+def _parse_resource_spec(check: _Check, spec: Mapping[str, Any], path: str, key: str) -> ResourceSpec | None:
+    entry = check.optional_mapping(spec, path, key, ("cpu", "memory"))
+    if entry is None:
+        return None
+    sub = _join(path, key)
+    return ResourceSpec(cpu=check.optional_string(entry, sub, "cpu"), memory=check.optional_string(entry, sub, "memory"))
+
+
+def _parse_task(check: _Check, metadata: Metadata, spec: Mapping[str, Any]) -> Task:
+    command = check.strings(spec, "spec", "command")
+    if not command:
+        raise check.fail("spec.command", "is required")
+    resources = check.optional_mapping(spec, "spec", "resources", ("requests", "limits"))
+    return Task(
+        metadata=metadata,
+        command=command,
+        image=check.optional_string(spec, "spec", "image"),
+        env=tuple(
+            EnvVar(check.string(e, p, "name"), check.string(e, p, "value"))
+            for p, e in check.mappings(spec, "spec", "env", ("name", "value"))
+        ),
+        resources=None
+        if resources is None
+        else Resources(
+            requests=_parse_resource_spec(check, resources, "spec.resources", "requests"),
+            limits=_parse_resource_spec(check, resources, "spec.resources", "limits"),
+        ),
+        workspaces=tuple(
+            WorkspaceBinding(check.string(w, p, "name"), check.string(w, p, "path"), check.optional_string(w, p, "goal"))
+            for p, w in check.mappings(spec, "spec", "workspaces", ("name", "path", "goal"))
+        ),
+        debug=check.boolean(spec, "spec", "debug"),
+    )
+
+
+def _parse_workspace(check: _Check, metadata: Metadata, spec: Mapping[str, Any]) -> Workspace:
+    mcp = check.optional_mapping(spec, "spec", "mcp", ("registries", "servers"))
+    skills = check.optional_mapping(spec, "spec", "skills", ("registries", "path"))
+    return Workspace(
+        metadata=metadata,
+        git=tuple(
+            GitSource(
+                check.string(g, p, "name"),
+                check.string(g, p, "repo"),
+                check.optional_string(g, p, "branch"),
+                check.optional_string(g, p, "commit"),
+            )
+            for p, g in check.mappings(spec, "spec", "git", ("name", "repo", "branch", "commit"))
+        ),
+        files=tuple(
+            FileEntry(check.string(f, p, "path"), check.string(f, p, "content"))
+            for p, f in check.mappings(spec, "spec", "files", ("path", "content"))
+        ),
+        mcp=None
+        if mcp is None
+        else McpSpec(tuple(check.sequence(mcp, "spec.mcp", "registries")), tuple(check.sequence(mcp, "spec.mcp", "servers"))),
+        skills=None
+        if skills is None
+        else SkillsSpec(tuple(check.sequence(skills, "spec.skills", "registries")), check.optional_string(skills, "spec.skills", "path")),
+    )
+
+
+def _parse_model(check: _Check, metadata: Metadata, spec: Mapping[str, Any]) -> Model:
+    provider = check.string(spec, "spec", "provider")
+    if provider in EXTENSION_PROVIDERS:
+        if metadata.annotations.get(PROVIDER_EXTENSION_ANNOTATION) != "true":
+            raise check.fail(
+                "spec.provider",
+                f'"{provider}" is not an ax provider; requires metadata.annotations["{PROVIDER_EXTENSION_ANNOTATION}"] = "true"',
+            )
+    elif provider not in AX_PROVIDERS:
+        raise check.fail("spec.provider", f'"{provider}" is not one of {", ".join(sorted(AX_PROVIDERS | EXTENSION_PROVIDERS))}')
+    secret = check.optional_mapping(spec, "spec", "secretKey", ("name", "key"))
+    parameters = spec.get("parameters") or {}
+    if not isinstance(parameters, dict):
+        raise check.fail("spec.parameters", "must be a mapping")
+    return Model(
+        metadata=metadata,
+        provider=provider,
+        model=check.string(spec, "spec", "model"),
+        secret_key=None
+        if secret is None
+        else SecretKey(check.string(secret, "spec.secretKey", "name"), check.string(secret, "spec.secretKey", "key")),
+        parameters=parameters,
+    )
+
+
+def _parse_budget(check: _Check, entry: Mapping[str, Any], path: str) -> Budget | None:
+    budget = check.optional_mapping(entry, path, "budget", ("usd", "calls"))
+    if budget is None:
+        return None
+    usd, calls = budget.get("usd"), budget.get("calls")
+    if usd is not None and (isinstance(usd, bool) or not isinstance(usd, int | float) or usd < 0):
+        raise check.fail(f"{path}.budget.usd", "must be a non-negative number")
+    if calls is not None and (isinstance(calls, bool) or not isinstance(calls, int) or calls < 0):
+        raise check.fail(f"{path}.budget.calls", "must be a non-negative integer")
+    return Budget(usd=None if usd is None else float(usd), calls=calls)
+
+
+def _parse_run(check: _Check, metadata: Metadata, spec: Mapping[str, Any]) -> Run:
+    entries = check.mappings(spec, "spec", "tasks", ("name", "task", "dependsOn", "inputs", "outputs", "budget", "serialize"))
+    if not entries:
+        raise check.fail("spec.tasks", "is required")
+    tasks: list[RunTask] = []
+    for path, entry in entries:
+        inputs = check.string_map(entry, path, "inputs")
+        for input_name, ref in inputs.items():
+            producer, _, artifact = ref.partition("/")
+            if not producer or not artifact:
+                raise check.fail(f"{path}.inputs.{input_name}", f'"{ref}" must be "<task>/<artifact>"')
+        tasks.append(
+            RunTask(
+                name=check.string(entry, path, "name"),
+                task=check.string(entry, path, "task"),
+                depends_on=check.strings(entry, path, "dependsOn"),
+                inputs=inputs,
+                outputs=check.strings(entry, path, "outputs"),
+                budget=_parse_budget(check, entry, path),
+                serialize=check.optional_string(entry, path, "serialize"),
+            )
+        )
+    run = Run(metadata=metadata, tasks=tuple(tasks), artifacts=check.optional_string(spec, "spec", "artifacts"))
+    _check_run_graph(check, run)
+    return run
+
+
+def _check_run_graph(check: _Check, run: Run) -> None:
+    names: dict[str, int] = {}
+    for index, entry in enumerate(run.tasks):
+        if entry.name in names:
+            raise check.fail(f"spec.tasks[{index}].name", f'"{entry.name}" is defined twice')
+        names[entry.name] = index
+    for index, entry in enumerate(run.tasks):
+        for dep_index, dep in enumerate(entry.depends_on):
+            if dep not in names:
+                raise check.fail(f"spec.tasks[{index}].dependsOn[{dep_index}]", f'"{dep}" names no task in the Run')
+        for input_name, ref in entry.inputs.items():
+            if ref.split("/", 1)[0] not in names:
+                raise check.fail(f"spec.tasks[{index}].inputs.{input_name}", f'"{ref}" names no task in the Run')
+    cycle = _find_cycle(run)
+    if cycle is not None:
+        raise check.fail(f"spec.tasks[{names[cycle[0]]}].dependsOn", "closes a cycle: " + " -> ".join(cycle))
+
+
+def _find_cycle(run: Run) -> list[str] | None:
+    """Return one dependency cycle as ``[a, b, ..., a]`` over ``dependsOn`` and input producers, or None."""
+    state: dict[str, int] = {}  # 1 = on the current path, 2 = finished
+    path: list[str] = []
+
+    def visit(name: str) -> list[str] | None:
+        state[name] = 1
+        path.append(name)
+        for dep in run.task(name).producers:
+            mark = state.get(dep)
+            if mark == 1:
+                return path[path.index(dep) :] + [dep]
+            if mark is None:
+                found = visit(dep)
+                if found is not None:
+                    return found
+        path.pop()
+        state[name] = 2
+        return None
+
+    for entry in run.tasks:
+        if entry.name not in state:
+            found = visit(entry.name)
+            if found is not None:
+                return found
+    return None
+
+
+def check_run_tasks(run: Run, tasks: Mapping[str, Task]) -> None:
+    """Reject a Run whose entries name a Task that is not among ``tasks``."""
+    check = _Check("Run", run.metadata.name)
+    for index, entry in enumerate(run.tasks):
+        if entry.task not in tasks:
+            raise check.fail(f"spec.tasks[{index}].task", f'"{entry.task}" names no loaded Task')
+
+
+_PARSERS: dict[str, Any] = {"Task": _parse_task, "Workspace": _parse_workspace, "Model": _parse_model, "Run": _parse_run}
+_SPEC_FIELDS = {
+    "Task": ("image", "command", "env", "resources", "workspaces", "debug"),
+    "Workspace": ("git", "files", "mcp", "skills"),
+    "Model": ("provider", "model", "secretKey", "parameters"),
+    "Run": ("tasks", "artifacts"),
+}
+
+
+def parse_manifest(document: object, *, source: str = "<document>") -> Task | Workspace | Model | Run:
+    """Validate one decoded YAML document and return its manifest; raise :class:`ManifestError` naming the field."""
+    if not isinstance(document, dict):
+        raise ManifestError(f"{source}: a manifest must be a mapping")
+    kind = document.get("kind")
+    if not isinstance(kind, str) or kind not in _PARSERS:
+        raise ManifestError(f"{source}: kind {kind!r} is not one of Task, Workspace, Model, Run")
+    raw_meta = document.get("metadata")
+    raw_name = raw_meta.get("name") if isinstance(raw_meta, dict) else None
+    check = _Check(kind, raw_name if isinstance(raw_name, str) and raw_name else "?")
+    check.mapping(document, "", ("apiVersion", "kind", "metadata", "spec"))
+    api_version = document.get("apiVersion")
+    if api_version != _API_VERSIONS[kind]:
+        raise check.fail("apiVersion", f"{api_version!r} must be {_API_VERSIONS[kind]!r}")
+    metadata = _parse_metadata(check, raw_meta)
+    spec = check.mapping(document.get("spec"), "spec", _SPEC_FIELDS[kind])
+    manifest: Task | Workspace | Model | Run = _PARSERS[kind](check, metadata, spec)
+    return manifest
+
+
+def _documents(path: Path) -> list[Any]:
+    try:
+        text = path.read_text(encoding="utf-8")
+    except OSError as exc:
+        raise ManifestError(f"{path}: cannot be read: {exc}") from exc
+    try:
+        return list(yaml.safe_load_all(text))
+    except yaml.YAMLError as exc:
+        raise ManifestError(f"{path}: not valid YAML: {exc}") from exc
+
+
+def load_manifests(paths: Iterable[str | Path]) -> Manifests:
+    """Read one or more YAML files (multi-document allowed), validate each manifest and index them by kind and name.
+
+    Every Run is checked against the Tasks loaded in the same call; a name defined twice within a kind is rejected.
+    """
+    manifests = Manifests()
+    buckets: dict[type, dict[str, Any]] = {
+        Task: manifests.tasks,
+        Workspace: manifests.workspaces,
+        Model: manifests.models,
+        Run: manifests.runs,
+    }
+    for raw_path in paths:
+        path = Path(raw_path)
+        for index, document in enumerate(_documents(path)):
+            if document is None:
+                continue
+            manifest = parse_manifest(document, source=f"{path}[{index}]")
+            bucket = buckets[type(manifest)]
+            name = manifest.metadata.name
+            if name in bucket:
+                raise ManifestError(f"{type(manifest).__name__}/{name}: metadata.name is defined twice ({path})")
+            bucket[name] = manifest
+    for run in manifests.runs.values():
+        check_run_tasks(run, manifests.tasks)
+    return manifests

tests/test_harness_ext.py

@@ -18,8 +18,8 @@ from openultrasast.harness_ext import (
 PYPROJECT = tomllib.loads(Path("pyproject.toml").read_text())
 
 
-def test_core_dependencies_stay_empty() -> None:
+def test_core_dependencies_are_pyyaml_only() -> None:
-    assert PYPROJECT["project"]["dependencies"] == []
+    assert PYPROJECT["project"]["dependencies"] == ["pyyaml>=6.0"]  # ai-service-plane task 1: manifests are YAML; HarnessX stays an extra
 
 
 def test_harnessx_is_a_sha_pinned_optional_extra() -> None:

tests/test_plane_manifests.py

@@ -0,0 +1,419 @@
+"""Manifest schemas of the service plane: ax's kinds load, a Run loads, and every violation names its field."""
+
+from __future__ import annotations
+
+import copy
+from pathlib import Path
+from typing import Any
+
+import pytest
+import yaml
+
+from openultrasast.plane.manifests import (
+    ManifestError,
+    Model,
+    Run,
+    Task,
+    Workspace,
+    check_run_tasks,
+    load_manifests,
+    parse_manifest,
+)
+
+TASK_YAML = """
+apiVersion: ax.io/v1alpha1
+kind: Task
+metadata:
+  name: verify
+  namespace: sast
+spec:
+  image: openultrasast:dev
+  command: ["verify", "--pass", "a"]
+  env:
+    - name: OUSAST_PASS
+      value: "a"
+  resources:
+    requests: {cpu: "500m", memory: "1Gi"}
+    limits: {cpu: "2", memory: "4Gi"}
+  workspaces:
+    - name: case-pin
+      path: /workspace
+      goal: the repository under test at its vulnerable pin
+  debug: true
+"""
+
+WORKSPACE_YAML = """
+apiVersion: ax.io/v1alpha1
+kind: Workspace
+metadata:
+  name: flask-abc123
+spec:
+  git:
+    - name: repo
+      repo: https://github.com/pallets/flask.git
+      branch: main
+      commit: abc123
+  files:
+    - path: candidates.json
+      content: '{"sinks": []}'
+  mcp:
+    registries: [https://registry.example/mcp]
+    servers: [{name: fs}]
+  skills:
+    registries: [https://registry.example/skills]
+    path: /skills
+"""
+
+MODEL_YAML = """
+apiVersion: ax.io/v1alpha1
+kind: Model
+metadata:
+  name: deepseek-flash
+  annotations:
+    openultrasast.io/provider-extension: "true"
+spec:
+  provider: deepseek
+  model: deepseek-chat
+  secretKey:
+    name: deepseek-api
+    key: token
+  parameters:
+    price_prompt_usd_per_m: 0.27
+    price_completion_usd_per_m: 1.1
+"""
+
+RUN_YAML = """
+apiVersion: openultrasast.io/v1alpha1
+kind: Run
+metadata:
+  name: validation-46
+spec:
+  artifacts: ~/ousast-results/plane
+  tasks:
+    - name: repo-facts
+      task: repo-facts
+      outputs: [facts.json]
+      serialize: engine
+    - name: verify-a
+      task: verify
+      dependsOn: [repo-facts]
+      inputs: {facts: repo-facts/facts.json}
+      outputs: [units.jsonl]
+      budget: {usd: 1.5, calls: 400}
+    - name: verify-b
+      task: verify
+      dependsOn: [repo-facts]
+      inputs: {facts: repo-facts/facts.json}
+      outputs: [units.jsonl]
+      budget: {usd: 1.5}
+    - name: agree
+      task: agree
+      dependsOn: [verify-a, verify-b]
+      inputs: {a: verify-a/units.jsonl, b: verify-b/units.jsonl}
+      outputs: [agreed.json, disputed.json]
+"""
+
+TASK_STUB = """
+apiVersion: ax.io/v1alpha1
+kind: Task
+metadata: {name: %s}
+spec: {command: [%s]}
+"""
+
+
+def doc(text: str) -> dict[str, Any]:
+    loaded = yaml.safe_load(text)
+    assert isinstance(loaded, dict)
+    return loaded
+
+
+def rejected(document: object, field: str) -> str:
+    """Parse and assert a ManifestError whose message names ``field``; return the message."""
+    with pytest.raises(ManifestError) as info:
+        parse_manifest(document)
+    message = str(info.value)
+    assert field in message, message
+    return message
+
+
+# --- valid manifests -----------------------------------------------------------------------------------------------
+
+
+def test_task_loads_every_ax_field() -> None:
+    task = parse_manifest(doc(TASK_YAML))
+    assert isinstance(task, Task)
+    assert task.metadata.name == "verify"
+    assert task.metadata.namespace == "sast"
+    assert task.image == "openultrasast:dev"
+    assert task.command == ("verify", "--pass", "a")
+    assert task.env[0].name == "OUSAST_PASS" and task.env[0].value == "a"
+    assert task.resources is not None and task.resources.requests is not None and task.resources.limits is not None
+    assert task.resources.requests.cpu == "500m" and task.resources.limits.memory == "4Gi"
+    assert task.workspaces[0].path == "/workspace" and task.workspaces[0].goal is not None
+    assert task.debug is True
+
+
+def test_task_minimal_defaults() -> None:
+    task = parse_manifest(doc(TASK_STUB % ("repo-facts", "repo-facts")))
+    assert isinstance(task, Task)
+    assert task.image is None and task.env == () and task.resources is None and task.workspaces == () and task.debug is False
+
+
+def test_workspace_loads() -> None:
+    workspace = parse_manifest(doc(WORKSPACE_YAML))
+    assert isinstance(workspace, Workspace)
+    assert workspace.git[0].repo.endswith("flask.git") and workspace.git[0].commit == "abc123"
+    assert workspace.files[0].path == "candidates.json"
+    assert workspace.mcp is not None and len(workspace.mcp.registries) == 1 and len(workspace.mcp.servers) == 1
+    assert workspace.skills is not None and workspace.skills.path == "/skills"
+
+
+def test_model_loads_with_declared_extension() -> None:
+    model = parse_manifest(doc(MODEL_YAML))
+    assert isinstance(model, Model)
+    assert model.provider == "deepseek" and model.model == "deepseek-chat"
+    assert model.secret_key is not None and model.secret_key.key == "token"
+    assert model.parameters["price_prompt_usd_per_m"] == 0.27
+
+
+@pytest.mark.parametrize("provider", ["google", "anthropic"])
+def test_model_ax_providers_need_no_annotation(provider: str) -> None:
+    document = doc(MODEL_YAML)
+    del document["metadata"]["annotations"]
+    document["spec"]["provider"] = provider
+    model = parse_manifest(document)
+    assert isinstance(model, Model) and model.provider == provider
+
+
+def test_run_loads_graph() -> None:
+    run = parse_manifest(doc(RUN_YAML))
+    assert isinstance(run, Run)
+    assert run.artifacts == "~/ousast-results/plane"
+    assert [t.name for t in run.tasks] == ["repo-facts", "verify-a", "verify-b", "agree"]
+    verify_a = run.task("verify-a")
+    assert verify_a.task == "verify" and verify_a.depends_on == ("repo-facts",)
+    assert verify_a.inputs == {"facts": "repo-facts/facts.json"}
+    assert verify_a.budget is not None and verify_a.budget.usd == 1.5 and verify_a.budget.calls == 400
+    assert run.task("verify-b").budget is not None and run.task("verify-b").budget.calls is None
+    assert run.task("repo-facts").serialize == "engine" and run.task("repo-facts").budget is None
+    assert run.task("agree").producers == ("verify-a", "verify-b")
+
+
+def test_load_manifests_reads_multi_document_files(tmp_path: Path) -> None:
+    (tmp_path / "kinds.yaml").write_text("---".join([TASK_YAML, WORKSPACE_YAML, MODEL_YAML]) + "\n---\n", encoding="utf-8")
+    (tmp_path / "tasks.yaml").write_text(
+        "---".join([TASK_STUB % ("repo-facts", "repo-facts"), TASK_STUB % ("agree", "agree")]), encoding="utf-8"
+    )
+    (tmp_path / "run.yaml").write_text(RUN_YAML, encoding="utf-8")
+    manifests = load_manifests([tmp_path / "kinds.yaml", str(tmp_path / "tasks.yaml"), tmp_path / "run.yaml"])
+    assert set(manifests.tasks) == {"verify", "repo-facts", "agree"}
+    assert set(manifests.workspaces) == {"flask-abc123"}
+    assert set(manifests.models) == {"deepseek-flash"}
+    assert set(manifests.runs) == {"validation-46"}
+
+
+# --- rejections: unknown fields, apiVersion, kind ---------------------------------------------------------------------
+
+
+def test_unknown_top_level_field_is_named() -> None:
+    document = doc(TASK_YAML)
+    document["status"] = {}
+    assert rejected(document, "status is not a field").startswith("Task/verify: ")
+
+
+def test_unknown_spec_field_is_named() -> None:
+    document = doc(TASK_YAML)
+    document["spec"]["foo"] = 1
+    assert rejected(document, "spec.foo is not a field") == "Task/verify: spec.foo is not a field"
+
+
+def test_unknown_nested_fields_are_named_with_index() -> None:
+    task = doc(TASK_YAML)
+    task["spec"]["env"][0]["valueFrom"] = {}
+    rejected(task, "spec.env[0].valueFrom is not a field")
+    task = doc(TASK_YAML)
+    task["spec"]["resources"]["requests"]["gpu"] = "1"
+    rejected(task, "spec.resources.requests.gpu is not a field")
+    workspace = doc(WORKSPACE_YAML)
+    workspace["spec"]["git"][0]["depth"] = 1
+    rejected(workspace, "Workspace/flask-abc123: spec.git[0].depth is not a field")
+    model = doc(MODEL_YAML)
+    model["spec"]["secretKey"]["namespace"] = "x"
+    rejected(model, "Model/deepseek-flash: spec.secretKey.namespace is not a field")
+    run = doc(RUN_YAML)
+    run["spec"]["tasks"][1]["image"] = "x"
+    rejected(run, "Run/validation-46: spec.tasks[1].image is not a field")
+    run = doc(RUN_YAML)
+    run["spec"]["tasks"][1]["budget"]["tokens"] = 1
+    rejected(run, "spec.tasks[1].budget.tokens is not a field")
+
+
+def test_unknown_metadata_field_is_named() -> None:
+    document = doc(TASK_YAML)
+    document["metadata"]["labels"] = {"a": "b"}
+    rejected(document, "metadata.labels is not a field")
+
+
+def test_wrong_api_version_is_named() -> None:
+    task = doc(TASK_YAML)
+    task["apiVersion"] = "ax.io/v1"
+    assert rejected(task, "apiVersion").startswith("Task/verify: apiVersion 'ax.io/v1' must be 'ax.io/v1alpha1'")
+    run = doc(RUN_YAML)
+    run["apiVersion"] = "ax.io/v1alpha1"
+    rejected(run, "Run/validation-46: apiVersion 'ax.io/v1alpha1' must be 'openultrasast.io/v1alpha1'")
+
+
+def test_unknown_kind_and_non_mapping_are_rejected() -> None:
+    document = doc(TASK_YAML)
+    document["kind"] = "Pod"
+    with pytest.raises(ManifestError, match="kind 'Pod' is not one of Task, Workspace, Model, Run"):
+        parse_manifest(document)
+    with pytest.raises(ManifestError, match="must be a mapping"):
+        parse_manifest(["not", "a", "manifest"])
+
+
+def test_missing_name_is_named() -> None:
+    document = doc(TASK_YAML)
+    del document["metadata"]["name"]
+    assert rejected(document, "metadata.name is required") == "Task/?: metadata.name is required"
+
+
+def test_task_type_errors_are_named() -> None:
+    task = doc(TASK_YAML)
+    task["spec"]["command"] = "verify"
+    rejected(task, "spec.command must be a list")
+    task = doc(TASK_YAML)
+    task["spec"]["command"] = ["verify", 3]
+    rejected(task, "spec.command[1] must be a non-empty string")
+    task = doc(TASK_YAML)
+    del task["spec"]["command"]
+    rejected(task, "spec.command is required")
+    task = doc(TASK_YAML)
+    task["spec"]["debug"] = "yes"
+    rejected(task, "spec.debug must be true or false")
+    task = doc(TASK_YAML)
+    del task["spec"]["workspaces"][0]["path"]
+    rejected(task, "spec.workspaces[0].path is required")
+
+
+# --- rejections: Model provider --------------------------------------------------------------------------------------
+
+
+def test_deepseek_without_extension_annotation_is_rejected() -> None:
+    document = doc(MODEL_YAML)
+    del document["metadata"]["annotations"]
+    message = rejected(document, "spec.provider")
+    assert message.startswith("Model/deepseek-flash: spec.provider")
+    assert 'metadata.annotations["openultrasast.io/provider-extension"] = "true"' in message
+    document = doc(MODEL_YAML)
+    document["metadata"]["annotations"]["openultrasast.io/provider-extension"] = "yes"
+    rejected(document, "spec.provider")
+
+
+def test_unknown_provider_is_rejected_even_with_annotation() -> None:
+    document = doc(MODEL_YAML)
+    document["spec"]["provider"] = "openai"
+    assert rejected(document, 'spec.provider "openai" is not one of anthropic, deepseek, google')
+
+
+def test_model_requires_model_name() -> None:
+    document = doc(MODEL_YAML)
+    del document["spec"]["model"]
+    rejected(document, "spec.model is required")
+
+
+# --- rejections: Run graph --------------------------------------------------------------------------------------------
+
+
+def test_run_cycle_is_named() -> None:
+    document = doc(RUN_YAML)
+    document["spec"]["tasks"][0]["dependsOn"] = ["agree"]
+    message = rejected(document, "dependsOn closes a cycle")
+    assert message.startswith("Run/validation-46: spec.tasks[0].dependsOn closes a cycle: repo-facts -> ")
+    assert message.endswith("-> repo-facts")
+
+
+def test_run_self_dependency_is_a_cycle() -> None:
+    document = doc(RUN_YAML)
+    document["spec"]["tasks"][3]["dependsOn"] = ["agree"]
+    rejected(document, "spec.tasks[3].dependsOn closes a cycle: agree -> agree")
+
+
+def test_run_cycle_through_inputs_is_named() -> None:
+    document = doc(RUN_YAML)
+    document["spec"]["tasks"][0]["inputs"] = {"back": "verify-a/units.jsonl"}
+    rejected(document, "spec.tasks[0].dependsOn closes a cycle: repo-facts -> verify-a -> repo-facts")
+
+
+def test_run_input_naming_unknown_task_is_named() -> None:
+    document = doc(RUN_YAML)
+    document["spec"]["tasks"][1]["inputs"]["facts"] = "classify/facts.json"
+    rejected(document, 'spec.tasks[1].inputs.facts "classify/facts.json" names no task in the Run')
+
+
+def test_run_input_must_be_task_slash_artifact() -> None:
+    document = doc(RUN_YAML)
+    document["spec"]["tasks"][1]["inputs"]["facts"] = "facts.json"
+    rejected(document, 'spec.tasks[1].inputs.facts "facts.json" must be "<task>/<artifact>"')
+
+
+def test_run_depends_on_unknown_task_is_named() -> None:
+    document = doc(RUN_YAML)
+    document["spec"]["tasks"][3]["dependsOn"] = ["verify-a", "verify-c"]
+    rejected(document, 'spec.tasks[3].dependsOn[1] "verify-c" names no task in the Run')
+
+
+def test_run_duplicate_task_name_is_named() -> None:
+    document = doc(RUN_YAML)
+    document["spec"]["tasks"][2]["name"] = "verify-a"
+    rejected(document, 'spec.tasks[2].name "verify-a" is defined twice')
+
+
+def test_run_budget_types_are_named() -> None:
+    document = doc(RUN_YAML)
+    document["spec"]["tasks"][1]["budget"]["usd"] = "1.5"
+    rejected(document, "spec.tasks[1].budget.usd must be a non-negative number")
+    document = doc(RUN_YAML)
+    document["spec"]["tasks"][1]["budget"]["calls"] = 2.5
+    rejected(document, "spec.tasks[1].budget.calls must be a non-negative integer")
+    document = doc(RUN_YAML)
+    document["spec"]["tasks"][1]["budget"]["calls"] = -1
+    rejected(document, "spec.tasks[1].budget.calls must be a non-negative integer")
+
+
+def test_run_without_tasks_is_rejected() -> None:
+    document = doc(RUN_YAML)
+    document["spec"]["tasks"] = []
+    rejected(document, "spec.tasks is required")
+
+
+def test_run_task_naming_no_loaded_task_is_named(tmp_path: Path) -> None:
+    run = parse_manifest(doc(RUN_YAML))
+    assert isinstance(run, Run)
+    verify = parse_manifest(doc(TASK_YAML))
+    assert isinstance(verify, Task)
+    with pytest.raises(ManifestError, match=r'Run/validation-46: spec.tasks\[0\].task "repo-facts" names no loaded Task'):
+        check_run_tasks(run, {"verify": verify})
+    (tmp_path / "all.yaml").write_text("---".join([TASK_YAML, TASK_STUB % ("repo-facts", "repo-facts"), RUN_YAML]), encoding="utf-8")
+    with pytest.raises(ManifestError, match=r'spec.tasks\[3\].task "agree" names no loaded Task'):
+        load_manifests([tmp_path / "all.yaml"])
+
+
+# --- loader errors ---------------------------------------------------------------------------------------------------
+
+
+def test_load_manifests_rejects_duplicate_names_and_bad_files(tmp_path: Path) -> None:
+    (tmp_path / "dup.yaml").write_text("---".join([TASK_YAML, TASK_YAML]), encoding="utf-8")
+    with pytest.raises(ManifestError, match="Task/verify: metadata.name is defined twice"):
+        load_manifests([tmp_path / "dup.yaml"])
+    (tmp_path / "bad.yaml").write_text("kind: Task\nspec: [unclosed\n", encoding="utf-8")
+    with pytest.raises(ManifestError, match="not valid YAML"):
+        load_manifests([tmp_path / "bad.yaml"])
+    with pytest.raises(ManifestError, match="cannot be read"):
+        load_manifests([tmp_path / "missing.yaml"])
+
+
+def test_manifests_are_frozen() -> None:
+    task = parse_manifest(doc(TASK_YAML))
+    with pytest.raises(AttributeError):
+        task.metadata = copy.copy(task.metadata)  # type: ignore[misc]

tests/test_semantic_extra.py

@@ -13,9 +13,9 @@ from openultrasast.semantic.extra import grammar_for, has_semantic_extra
 from openultrasast.semantic.ir import parse_file
 
 
-def test_core_dependencies_stay_empty_and_semantic_is_optional() -> None:
+def test_core_dependencies_are_pyyaml_only_and_semantic_is_optional() -> None:
     payload = tomllib.loads(Path("pyproject.toml").read_text())
-    assert payload["project"]["dependencies"] == []
+    assert payload["project"]["dependencies"] == ["pyyaml>=6.0"]  # ai-service-plane task 1: manifests are YAML
     extra = payload["project"]["optional-dependencies"]["semantic"]
     joined = " ".join(extra)
     assert "tree-sitter" in joined