agent-runtime-framework/tests/invariants/test_workflow_script_composite_resolves.py

"""Workflow-script harness composition invariant (spec/typed-workflows.md WF-COMP-31
+ spec/workflow-variables.md WV-19 + spec/harness.md HD-14).

The two new framework layers — ``harnesses/contexts/shared-variables/v1/`` and
``harnesses/composites/workflow-script/v1/`` — must exist on the tree, must parse,
and the composite must resolve every layer it references to a real file on disk.
This test guards against the most common regression: a missing or renamed layer
that would only surface at the dispatcher's ``compose_harness`` call, where it
costs an entire task's worth of wasted dispatch round-trip.

Spec: /home/paul/dev/claude/projects/agent-runtimes/spec/workflow-variables.md
      (WV-19, WV-20, WF-COMP-31), spec/harness.md (HD-14).

Not committed by the authoring agent per its instructions — written here so it
can run locally against the current tree.
"""
from __future__ import annotations

from pathlib import Path

import pytest
import yaml

_HARNESSES_ROOT = Path(__file__).resolve().parents[2] / "harnesses"

# Composite layer entries use the singular kind as the dict key. The kind maps
# to a top-level directory by adding "s".
_KIND_TO_DIR = {
    "context": "contexts",
    "capability": "capabilities",
    "composite": "composites",
}


def _load_layer(kind_dir: str, name: str) -> dict:
    p = _HARNESSES_ROOT / kind_dir / name / "v1" / "harness.yaml"
    return yaml.safe_load(p.read_text(encoding="utf-8"))


def test_shared_variables_layer_exists_and_parses():
    """``harnesses/contexts/shared-variables/v1/harness.yaml`` must exist,
    parse as YAML, and carry ``provides: [shared-variables]`` (the HD-14 tag).

    Spec: spec/workflow-variables.md WV-19; spec/harness.md HD-14.
    """
    parsed = _load_layer("contexts", "shared-variables")
    assert parsed["kind"] == "context"
    assert parsed["name"] == "shared-variables"
    assert parsed["version"] == 1
    assert parsed["provides"] == ["shared-variables"], (
        "WV-19: provides MUST contain the 'shared-variables' tag — HD-14's "
        "extra_layers dedup key, and the tag the dispatcher's payload-marker "
        "injection matches against."
    )
    # WV-20: this layer's OWN env: declaration sets the two WORKFLOW_VARIABLES_*
    # vars (NOT the entrypoint, NOT the container backend).
    env = parsed["env"]
    assert env["WORKFLOW_VARIABLES_ROOT"] == "/variables"
    assert "WORKFLOW_VARIABLES_SELF" in env, (
        "WV-20: WORKFLOW_VARIABLES_SELF must be set by the layer's env: "
        "(templated per-node at harness-composition time on the dispatcher)"
    )
    # Agent-facing context: a CLAUDE.md must be mounted under /opt/harness/context/.
    files = parsed.get("context_files", [])
    targets = [f["target"] for f in files]
    assert any(t.startswith("/opt/harness/context/shared-variables/") for t in targets), (
        "WV-19: agent-facing context must be mounted under /opt/harness/context/"
        "(the HC-7 namespacing point) so the agent sees it at startup."
    )


def test_shared_variables_claude_md_present_and_substantial():
    """The agent-facing CLAUDE.md must be a non-trivial markdown document
    (HC-7 says it is read by the agent, not a placeholder).

    Spec: spec/workflow-variables.md WV-19; spec/harness.md HC-7.
    """
    md = _HARNESSES_ROOT / "contexts" / "shared-variables" / "v1" / "CLAUDE.md"
    assert md.exists(), "WV-19: shared-variables/v1 must ship a CLAUDE.md"
    text = md.read_text(encoding="utf-8")
    assert text.count("\n") >= 30, (
        f"WV-19: the CLAUDE.md must explain the /variables/ contract to the "
        f"agent — got {text.count(chr(10))} lines, expected >=30."
    )
    # Spot-check the four high-value pieces of content the agent MUST see:
    assert "/variables/" in text, "must explain the filesystem root"
    assert "WORKFLOW_VARIABLES_SELF" in text, "must document the env vars"
    assert "wfvar" in text, "must point at the CLI helper (WV-21)"
    assert "reserved" in text.lower(), "must list the reserved-name families (WV-6)"


def test_workflow_script_composite_resolves():
    """``harnesses/composites/workflow-script/v1/harness.yaml`` must declare
    exactly the two layers the spec names (WF-COMP-31) and every referenced
    file must exist on disk.

    Spec: spec/typed-workflows.md WF-COMP-31 ("Internal composite layered from
    ``shared-variables/v1`` + ``cp-service-token/v1``").
    """
    parsed = _load_layer("composites", "workflow-script")
    assert parsed["kind"] == "composite"
    assert parsed["name"] == "workflow-script"
    assert parsed["version"] == 1
    assert isinstance(parsed["layers"], list) and len(parsed["layers"]) >= 2, (
        "WF-COMP-31: the composite must layer at least shared-variables/v1 + "
        "cp-service-token/v1; missing one breaks the credential env seam."
    )

    refs = []
    for layer in parsed["layers"]:
        assert isinstance(layer, dict) and len(layer) == 1, (
            f"each layer entry must be a single-key dict (kind: ref), got {layer}"
        )
        kind_name, ref = next(iter(layer.items()))
        assert kind_name in _KIND_TO_DIR, f"unknown layer kind {kind_name!r}"
        assert "/" in ref, f"layer ref must be '<name>/v<N>', got {ref!r}"
        name, ver = ref.split("/")
        assert ver.startswith("v") and ver[1:].isdigit(), (
            f"layer ref version must match v<N>, got {ver!r}"
        )
        kind_dir = _KIND_TO_DIR[kind_name]
        layer_path = _HARNESSES_ROOT / kind_dir / name / ver / "harness.yaml"
        assert layer_path.exists(), (
            f"WF-COMP-31: composite references a layer that does not exist on "
            f"the tree: {layer_path.relative_to(_HARNESSES_ROOT)}"
        )
        layer_data = yaml.safe_load(layer_path.read_text(encoding="utf-8"))
        assert layer_data["name"] == name, (
            f"name mismatch in {layer_path}: file says {layer_data['name']!r}, "
            f"ref says {name!r}"
        )
        refs.append((kind_name, name, ver, layer_data))

    # WF-COMP-31: the two SPECIFIED layers must both be present.
    ref_names = {(kind, name) for kind, name, _, _ in refs}
    assert ("context", "shared-variables") in ref_names, (
        "WF-COMP-31: workflow-script/v1 MUST layer shared-variables/v1 "
        "(WV-19 — the agent-facing /variables/ contract)"
    )
    assert ("context", "cp-service-token") in ref_names, (
        "WF-COMP-31 / AU-51 credential env seam: workflow-script/v1 MUST "
        "layer cp-service-token/v1 so CP_SERVICE_TOKEN_FILE is in the "
        "entrypoint process env before shell_runner runs."
    )


def test_workflow_script_composite_brings_no_model_or_endpoint_layer():
    """``kind: script`` nodes run via the shell runner — the composite MUST
    NOT bring agent-only layers (no model, endpoint, CLAUDE.md context that
    drives the agent runner, etc.). This is a structural regression gate: a
    future refactor that adds e.g. ``anthropic-cloud-paul-oauth/v1`` here
    would silently widen the script-node attack surface.

    Spec: spec/typed-workflows.md WF-COMP-31 ("the flag-on kind: script
    harness ... MUST NOT bring agent-only layers").
    """
    parsed = _load_layer("composites", "workflow-script")
    forbidden_substrings = (
        # cloud-auth contexts the agent runner needs
        "anthropic-cloud", "minimax", "qwen-code", "deepseek-code",
        "alibaba-model-studio", "fable-low-effort", "z-ai",
        # the model-bearing capability / context layers
        "code-methodology", "spec-writing", "planning", "integration",
        # harness composites that bundle a model
        "code-", "planning-", "spec-writing-", "best-practices-",
        "sonnet-", "opus-", "minimax-", "haiku-", "airouter-", "fable-",
    )
    layer_names = [
        next(iter(layer.values())).split("/")[0]
        for layer in parsed["layers"]
    ]
    bad = [
        n for n in layer_names
        if any(sub in n for sub in forbidden_substrings)
    ]
    assert not bad, (
        f"WF-COMP-31: workflow-script/v1 MUST NOT bring agent-only layers; "
        f"found: {bad}"
    )


def test_cp_service_token_layer_already_sets_cp_service_token_file_env():
    """WF-COMP-31 'AU-51 credential env seam': cp-service-token/v1 must set
    ``CP_SERVICE_TOKEN_FILE`` in its env: declaration (the layer's init.sh
    restages the ESO-mounted secret to an agent-readable 0600 copy at the
    path CP_SERVICE_TOKEN_FILE points at). Verified here at the YAML level
    rather than at runtime — the spec's claim is structural.

    Spec: spec/typed-workflows.md WF-COMP-31; spec/harness.md AU-51/AU-51b
    (bug b3a96acc).
    """
    parsed = _load_layer("contexts", "cp-service-token")
    env = parsed["env"]
    assert "CP_SERVICE_TOKEN_FILE" in env, (
        "WF-COMP-31 'AU-51 credential env seam': cp-service-token/v1 must "
        "declare CP_SERVICE_TOKEN_FILE in its env: block; without it the "
        "shell_runner subprocess does not inherit the AU-51 credential."
    )
    fpath = env["CP_SERVICE_TOKEN_FILE"]
    assert fpath.endswith("/token"), (
        f"CP_SERVICE_TOKEN_FILE must point at the restaged 0600 file, not "
        f"the raw 0400 ESO mount — got {fpath!r}"
    )


def test_workflow_script_composite_forge_credentials_are_read_only():
    """Any layer in workflow-script/v1 that carries a forge credential MUST
    declare ``write_targets: []`` (HC-11): a script node may read the project
    repo but never push. The write-capable gitea-ssh/v1 layer must not be
    present (bug 45a48685).

    Spec: spec/harness.md HC-11, spec/typed-workflows.md WF-COMP-34.
    """
    parsed = _load_layer("composites", "workflow-script")
    from tests.invariants.forge_credentials import _layer_has_forge_credentials

    for layer in parsed["layers"]:
        kind_name, ref = next(iter(layer.items()))
        name = ref.split("/")[0]
        assert name != "gitea-ssh", "workflow-script/v1 must not layer the write-capable gitea-ssh/v1 (bug 45a48685)"
        kind_dir = _KIND_TO_DIR[kind_name]
        sub = _load_layer(kind_dir, name)
        if _layer_has_forge_credentials(sub):
            assert sub.get("write_targets") == [], (
                f"workflow-script/v1's layer {name} carries a forge credential "
                f"but does not declare write_targets: [] (HC-11 read-only)."
            )