marius/Open_Ultra_SAST
1
0
Fork 0
mirror of https://github.com/norandom/OpenUltraSAST.git synced 2026-10-16 03:49:26 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(plane): ax field names verbatim; pins and bindings as stripped annotations

Browse Source 
Requirement 1.1 (Task/Workspace/Model use only fields ax documents) and
Requirement 1.4 (no ax field repurposed) were violated: manifests accepted
metadata.namespace and a Workspace git `commit`, and the reconciler dumped
every git field into what `ax apply` receives.

Facts from ax's source (pkg/apis/v1alpha1/ax.proto, internal/workspace/setup.go):
- ObjectMeta is `name` and `atespace` (plus server-set creation_timestamp);
  no `namespace`, no `annotations`.
- GitRepo is `name`, `repo`, `branch`, `dir`, `depth`; no `commit`.
- The API server decodes with protojson.UnmarshalOptions{}: an unknown field
  is an error at `ax apply` (confirmed on the live cluster: a rendered
  Workspace+Model applied cleanly; the same Workspace with `commit` was
  refused with `unknown field "commit"`).
- ValidateModel checks only metadata; any spec.provider string is accepted.

Changes:
- manifests: metadata takes `name`, `atespace`, `annotations` (our extension,
  never sent to ax); git entries take exactly ax's five fields (depth int);
  `commit` in spec.git is rejected with a pointer to the new Workspace
  annotation `openultrasast.io/git-commits: "<git name>=<40-hex>,..."`,
  validated for unknown git names and non-hex shas.
- reconciler: rendered docs carry only ax fields; each bound Workspace's pins
  reach the Task as env OUSAST_GIT_PINS = {"<workspace>/<git name>": "<sha>"}.
  A test walks every rendered doc against ax's field lists.
- runner: pins come from OUSAST_GIT_PINS; git is fetched as setup.go does
  (init/remote/fetch with --depth when depth > 0/checkout FETCH_HEAD, default
  branch main, cloneDestination's dir and placeholder-name rules), then the
  pin is checked out. A pinned entry without a branch fetches the commit.
- workspaces: emits the annotation, `dir: repo` and `depth: 1` instead of
  `commit`, so generated case workspaces keep <path>/repo at depth 1.
- design.md: ax field names verbatim; pins and model binding as stripped
  openultrasast.io/* annotations carried to the task by env.

Co-Authored-By: Claude Opus 5.5 <[email protected]>
Claude-Session: https://claude.ai/code/session_01EmM1j46qgKMh3zqD1QMwEo
This commit is contained in:
Marius Ciepluch
2026-09-29 18:36:29 +02:00
co-authored by Claude Opus 5.5
parent e6d5124ffc
commit 18b67cf942
9 changed files with 450 additions and 81 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/fixtures/plane/smoke-task.yaml
Vendored
+29
View File
@@ -0,0 +1,29 @@
# Minimal Task for the bring-up smoke test: default runner image, a files-only workspace, no MCP or skills.
apiVersion: ax.io/v1alpha1
kind: Task
metadata:
name: ousast-smoke
atespace: default
spec:
image: "gcr.io/ax-substrate/ate-images/ax-task-runner@sha256:3a0dea6ad8b55278685db58aca6e37dc4ba04056831d45bef3aaeafdca43cac6"
command: ["sh", "-c", "cat /workspace/hello.txt && echo smoke-ok"]
resources:
requests:
cpu: "100m"
memory: "256Mi"
limits:
cpu: "1"
memory: "1Gi"
workspaces:
- name: ousast-smoke-ws
path: "/workspace"
---
apiVersion: ax.io/v1alpha1
kind: Workspace
metadata:
name: ousast-smoke-ws
atespace: default
spec:
files:
- path: "hello.txt"
content: "hello from openultrasast\n"
tests/test_plane_manifests.py
+63 -6
View File
@@ -25,7 +25,7 @@ apiVersion: ax.io/v1alpha1
kind: Task
metadata:
name: verify
namespace: sast
atespace: sast
spec:
image: openultrasast:dev
command: ["verify", "--pass", "a"]
@@ -47,12 +47,15 @@ apiVersion: ax.io/v1alpha1
kind: Workspace
metadata:
name: flask-abc123
annotations:
openultrasast.io/git-commits: "repo=0123456789abcdef0123456789abcdef01234567"
spec:
git:
- name: repo
repo: https://github.com/pallets/flask.git
branch: main
commit: abc123
dir: flask
depth: 1
files:
- path: candidates.json
content: '{"sinks": []}'
@@ -143,7 +146,7 @@ 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.metadata.atespace == "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"
@@ -162,12 +165,66 @@ def test_task_minimal_defaults() -> None:
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.git[0].repo.endswith("flask.git")
assert workspace.git[0].dir == "flask" and workspace.git[0].depth == 1
assert workspace.pins == {"repo": "0123456789abcdef0123456789abcdef01234567"}
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"
SMOKE = Path(__file__).parent / "fixtures" / "plane" / "smoke-task.yaml"
def test_the_bring_up_smoke_manifest_loads() -> None:
"""``ops/ax/smoke-task.yaml`` (copied as a fixture) is what ``ax apply`` accepted; it must load here too."""
manifests = load_manifests([SMOKE])
assert manifests.tasks["ousast-smoke"].metadata.atespace == "default"
assert manifests.workspaces["ousast-smoke-ws"].metadata.atespace == "default"
def test_the_shipped_smoke_manifest_matches_the_fixture() -> None:
shipped = Path(__file__).parent.parent / "ops" / "ax" / "smoke-task.yaml"
if not shipped.is_file():
pytest.skip("ops/ax/smoke-task.yaml is not in this checkout")
assert load_manifests([shipped]).tasks.keys() == {"ousast-smoke"}
@pytest.mark.parametrize("key", ["namespace", "labels", "creation_timestamp"])
def test_metadata_takes_ax_names_only(key: str) -> None:
task = doc(TASK_YAML)
task["metadata"][key] = "sast"
rejected(task, f"Task/verify: metadata.{key} is not a field")
def test_git_commit_is_rejected_with_a_pointer_to_the_annotation() -> None:
workspace = doc(WORKSPACE_YAML)
workspace["spec"]["git"][0]["commit"] = "abc123"
message = rejected(workspace, "Workspace/flask-abc123: spec.git[0].commit is not an ax field")
assert "openultrasast.io/git-commits" in message
@pytest.mark.parametrize(
("value", "rule"),
[
("other=" + "a" * 40, '"other" names no spec.git entry'),
("repo=abc123", '"repo=abc123" is not a 40-hex commit'),
("repo", 'item "repo" must be'),
],
)
def test_git_commits_annotation_is_validated(value: str, rule: str) -> None:
workspace = doc(WORKSPACE_YAML)
workspace["metadata"]["annotations"]["openultrasast.io/git-commits"] = value
message = rejected(workspace, 'metadata.annotations["openultrasast.io/git-commits"]')
assert rule in message, message
def test_git_depth_must_be_an_integer() -> None:
workspace = doc(WORKSPACE_YAML)
workspace["spec"]["git"][0]["depth"] = "1"
rejected(workspace, "Workspace/flask-abc123: spec.git[0].depth must be a non-negative integer")
def test_model_loads_with_declared_extension() -> None:
model = parse_manifest(doc(MODEL_YAML))
assert isinstance(model, Model)
@@ -235,8 +292,8 @@ def test_unknown_nested_fields_are_named_with_index() -> None:
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")
workspace["spec"]["git"][0]["submodules"] = True
rejected(workspace, "Workspace/flask-abc123: spec.git[0].submodules 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")
tests/test_plane_reconciler.py
+63 -6
View File
@@ -24,6 +24,40 @@ from openultrasast.plane import reconciler
from openultrasast.plane.workspaces import workspaces
RECONCILER = Path("src/openultrasast/plane/reconciler.py")
PIN = "0123456789abcdef0123456789abcdef01234567"
# The fields ax's API server decodes (pkg/apis/v1alpha1/ax.proto, protojson: an unknown field is an error at
# ``ax apply``). ``None`` marks a value ax types as free-form (a Struct) or that this test does not descend into.
_RESOURCES = {"cpu": None, "memory": None}
AX_FIELDS: dict[str, dict] = {
"Task": {
"image": None,
"command": None,
"env": {"name": None, "value": None},
"resources": {"requests": _RESOURCES, "limits": _RESOURCES},
"workspaces": {"name": None, "path": None, "goal": None},
"debug": None,
},
"Workspace": {
"git": {"name": None, "repo": None, "branch": None, "dir": None, "depth": None},
"files": {"path": None, "content": None},
"mcp": {"registries": None, "servers": None},
"skills": {"registries": None, "path": None},
},
"Model": {"provider": None, "model": None, "secretKey": {"name": None, "key": None}, "parameters": None},
}
def non_ax_fields(value: object, schema: dict | None, path: str) -> list[str]:
"""Every key under ``value`` that ``schema`` does not name, as dotted paths."""
if schema is None:
return []
if isinstance(value, list):
return [bad for i, item in enumerate(value) for bad in non_ax_fields(item, schema, f"{path}[{i}]")]
assert isinstance(value, dict), f"{path} must be a mapping"
bad = [f"{path}.{k}" for k in value if k not in schema]
return bad + [b for k, v in value.items() if k in schema for b in non_ax_fields(v, schema[k], f"{path}.{k}")]
FAKE_AX = '''#!{python}
"""Fake ax CLI for the reconciler tests: argv log, scripted phases, scripted deliveries."""
@@ -90,8 +124,10 @@ apiVersion: ax.io/v1alpha1
kind: Workspace
metadata:
name: case-pin
atespace: default
annotations: {openultrasast.io/git-commits: "repo=0123456789abcdef0123456789abcdef01234567"}
spec:
git: [{name: repo, repo: https://example.invalid/repo.git, commit: abc123}]
git: [{name: repo, repo: https://example.invalid/repo.git, dir: src, depth: 1}]
---
apiVersion: ax.io/v1alpha1
kind: Task
@@ -242,6 +278,10 @@ def test_chain_runs_in_order_with_inputs_env_and_model(fake: Fake, tmp_path: Pat
assert inputs_ws["spec"]["files"] == [{"path": "repo-facts/facts.json", "content": json.dumps({"a.py": {"functions": ["f"]}})}]
assert verify["docs"][0]["spec"]["workspaces"][-1] == {"name": "chain-verify-inputs", "path": "/workspace/.ousast-in/verify"}
assert "annotations" not in verify["docs"][0]["metadata"], "ax's ObjectMeta has no annotations"
assert json.loads(env["OUSAST_GIT_PINS"]) == {"case-pin/repo": PIN}, "the Workspace annotation reaches the task as env"
assert verify["docs"][2]["spec"]["git"] == [{"name": "repo", "repo": "https://example.invalid/repo.git", "dir": "src", "depth": 1}]
assert verify["docs"][2]["metadata"] == {"name": "case-pin", "atespace": "default"}
assert "OUSAST_GIT_PINS" not in fake.applied("chain", "agree")["env"], "no bound workspace, no pins"
facts = fake.applied("chain", "repo-facts")
assert "OUSAST_MODEL" not in facts["env"] and "OUSAST_BUDGET_USD" not in facts["env"]
assert facts["docs"][0]["spec"]["image"] == "ousast-runner:dev"
@@ -440,8 +480,8 @@ def test_doctor_names_the_bring_up_script_when_nothing_is_reachable(monkeypatch:
def test_workspaces_one_manifest_per_case_pin(tmp_path: Path) -> None:
population = tmp_path / "population.toml"
population.write_text(
'[[case]]\nid = "demo-sqli"\nrepo = "https://github.com/x/y"\nvulnerable = "aaa"\nfixed = "bbb"\n'
'benign = { base = "ccc", tip = "ddd" }\n',
'[[case]]\nid = "demo-sqli"\nrepo = "https://github.com/x/y"\nvulnerable = "' + "a" * 40 + '"\nfixed = "' + "b" * 40 + '"\n'
'benign = { base = "' + "c" * 40 + '", tip = "' + "d" * 40 + '" }\n',
encoding="utf-8",
)
written = workspaces(population, tmp_path / "out")
@@ -451,12 +491,29 @@ def test_workspaces_one_manifest_per_case_pin(tmp_path: Path) -> None:
assert doc == {
"apiVersion": "ax.io/v1alpha1",
"kind": "Workspace",
"metadata": {"name": "demo-sqli-vulnerable"},
"spec": {"git": [{"name": "repo", "repo": "https://github.com/x/y", "commit": "aaa"}]},
"metadata": {"name": "demo-sqli-vulnerable", "annotations": {"openultrasast.io/git-commits": "repo=" + "a" * 40}},
"spec": {"git": [{"name": "repo", "repo": "https://github.com/x/y", "dir": "repo", "depth": 1}]},
}
from openultrasast.plane.manifests import load_manifests
assert set(load_manifests(written).workspaces) == {p.stem for p in written}
loaded = load_manifests(written).workspaces
assert set(loaded) == {p.stem for p in written}
assert loaded["demo-sqli-benign-tip"].pins == {"repo": "d" * 40}
def test_rendered_documents_carry_only_ax_fields(tmp_path: Path) -> None:
"""Req 1.1/1.4: whatever ``ax apply`` receives names only fields ax documents; annotations never leave."""
(tmp_path / "repo-facts").mkdir()
(tmp_path / "repo-facts" / "facts.json").write_text("{}", encoding="utf-8")
run, manifests = reconciler.load_run(write_run(tmp_path, "fields"))
docs = [d for entry in run.tasks for d in reconciler.render_task(run, entry, manifests, tmp_path, "http://h:1/")]
assert {d["kind"] for d in docs} == {"Task", "Workspace", "Model"}
for rendered in docs:
where = f"{rendered['kind']}/{rendered['metadata']['name']}"
assert set(rendered) == {"apiVersion", "kind", "metadata", "spec"}, where
assert set(rendered["metadata"]) <= {"name", "atespace"}, where
assert non_ax_fields(rendered["spec"], AX_FIELDS[rendered["kind"]], f"{where} spec") == []
assert "openultrasast.io/" not in yaml.safe_dump(rendered["metadata"]), where
def test_reconciler_is_under_500_lines_and_holds_no_pipeline_logic() -> None:
tests/test_plane_runner.py
+109 -10
View File
@@ -19,7 +19,15 @@ from pathlib import Path
import pytest
from openultrasast.plane import runner
from openultrasast.plane.runner import EXIT_BY_STATUS, DeliveryError, deliver, find_cached_checkout, start_health_server
from openultrasast.plane.manifests import GitSource
from openultrasast.plane.runner import (
EXIT_BY_STATUS,
DeliveryError,
clone_destination,
deliver,
find_cached_checkout,
start_health_server,
)
STUB = Path(__file__).parent / "plane_stub_task.py"
STUB_MODULE = f"{runner.TASK_PACKAGE}.stub_task"
@@ -313,14 +321,19 @@ def upstream(tmp_path: Path) -> tuple[Path, str]:
return repo, pin
def git_workspace_yaml(repo_url: str, pin: str) -> str:
def git_workspace_yaml(repo_url: str, extra: str = "") -> str:
"""ax's Workspace as ax hands it over: no annotations, no commit (the pin travels in ``OUSAST_GIT_PINS``)."""
return (
"apiVersion: ax.io/v1alpha1\nkind: Workspace\nmetadata:\n name: case\nspec:\n git:\n"
f" - name: repo\n repo: {repo_url}\n branch: main\n commit: {pin}\n"
f" - name: code\n repo: {repo_url}\n branch: main\n{extra}"
" files:\n - path: candidates.json\n content: '{\"sinks\": []}'\n"
)
def pins(pin: str, key: str = "case/code") -> str:
return json.dumps({key: pin})
def test_git_entry_is_exported_from_the_case_cache_at_the_pin(
env: dict[str, str], monkeypatch: pytest.MonkeyPatch, tmp_path: Path, upstream: tuple[Path, str]
) -> None:
@@ -332,22 +345,24 @@ def test_git_entry_is_exported_from_the_case_cache_at_the_pin(
git("remote", "set-url", "origin", url, cwd=case)
assert find_cached_checkout(cache, "https://example.invalid/acme/repo") == case
assert find_cached_checkout(cache, "https://example.invalid/other/repo.git") is None
code, output = run_main(monkeypatch, tmp_path, git_workspace_yaml(url, pin), OUSAST_CASE_CACHE=str(cache))
code, output = run_main(monkeypatch, tmp_path, git_workspace_yaml(url), OUSAST_CASE_CACHE=str(cache), OUSAST_GIT_PINS=pins(pin))
assert code == 0, (output / "summary.json").read_text()
checkout = tmp_path / "ws" / "repo"
checkout = tmp_path / "ws" / "code"
assert (checkout / "hello.py").read_text() == "print('v1')\n", "the pinned commit, not the branch tip"
assert not (checkout / ".git").exists(), "an export, as evaluate.export makes"
facts = json.loads((output / "facts.json").read_text())
assert facts["workspace_files"] == ["candidates.json", "repo/hello.py"]
assert facts["workspace_files"] == ["candidates.json", "code/hello.py"]
def test_git_entry_without_a_cache_is_shallow_cloned_at_the_pin(
env: dict[str, str], monkeypatch: pytest.MonkeyPatch, tmp_path: Path, upstream: tuple[Path, str]
) -> None:
repo, pin = upstream
code, output = run_main(monkeypatch, tmp_path, git_workspace_yaml(repo.as_uri(), pin), OUSAST_CASE_CACHE=str(tmp_path / "no-cache"))
code, output = run_main(
monkeypatch, tmp_path, git_workspace_yaml(repo.as_uri()), OUSAST_CASE_CACHE=str(tmp_path / "no-cache"), OUSAST_GIT_PINS=pins(pin)
)
assert code == 0, (output / "summary.json").read_text()
checkout = tmp_path / "ws" / "repo"
checkout = tmp_path / "ws" / "code"
assert (checkout / "hello.py").read_text() == "print('v1')\n"
assert git("rev-parse", "HEAD", cwd=checkout) == pin
@@ -356,9 +371,93 @@ def test_a_materialised_repo_survives_a_resume(
env: dict[str, str], monkeypatch: pytest.MonkeyPatch, tmp_path: Path, upstream: tuple[Path, str]
) -> None:
repo, pin = upstream
checkout = tmp_path / "ws" / "repo"
checkout = tmp_path / "ws" / "code"
checkout.mkdir(parents=True)
(checkout / "hello.py").write_text("kept\n")
code, _ = run_main(monkeypatch, tmp_path, git_workspace_yaml(repo.as_uri(), pin))
code, _ = run_main(monkeypatch, tmp_path, git_workspace_yaml(repo.as_uri()), OUSAST_GIT_PINS=pins(pin))
assert code == 0
assert (checkout / "hello.py").read_text() == "kept\n", "/workspace persists across suspend and resume; no re-clone"
# --- ax's GitRepo fields: dir, depth; the commit pin from OUSAST_GIT_PINS ------------------------------------------------
@pytest.mark.parametrize(
("name", "repo", "dir_", "expected"),
[
("code", "https://github.com/chalk/chalk.git", None, "/ws/code"),
("repo", "https://github.com/chalk/chalk.git", None, "/ws/chalk"),
("origin", "[email protected]:chalk/chalk", None, "/ws/chalk"),
("code", "https://github.com/chalk/chalk.git", "src", "/ws/src"),
("code", "https://github.com/chalk/chalk.git", ".", "/ws"),
("code", "https://github.com/chalk/chalk.git", "/elsewhere", "/elsewhere"),
],
)
def test_clone_destination_follows_ax_setup(name: str, repo: str, dir_: str | None, expected: str) -> None:
"""ax internal/workspace/setup.go cloneDestination: dir wins; placeholder names derive from the URL."""
assert clone_destination(GitSource(name, repo, dir=dir_), Path("/ws")) == Path(expected)
def test_git_dir_and_depth_are_honoured(
env: dict[str, str], monkeypatch: pytest.MonkeyPatch, tmp_path: Path, upstream: tuple[Path, str]
) -> None:
repo, _ = upstream
code, output = run_main(monkeypatch, tmp_path, git_workspace_yaml(repo.as_uri(), " dir: src\n depth: 1\n"))
assert code == 0, (output / "summary.json").read_text()
checkout = tmp_path / "ws" / "src"
assert (checkout / "hello.py").read_text() == "print('v2')\n", "no pin: the branch tip"
assert git("rev-parse", "--is-shallow-repository", cwd=checkout) == "true", "depth 1 is a shallow fetch"
assert not (tmp_path / "ws" / "code").exists()
def test_a_pin_naming_no_bound_git_entry_fails_the_task(
env: dict[str, str], monkeypatch: pytest.MonkeyPatch, tmp_path: Path, upstream: tuple[Path, str]
) -> None:
repo, pin = upstream
code, output = run_main(monkeypatch, tmp_path, git_workspace_yaml(repo.as_uri()), OUSAST_GIT_PINS=pins(pin, "case/other"))
assert code == EXIT_BY_STATUS["failed"]
assert "case/other" in json.loads((output / "summary.json").read_text())["reason"]
def test_unreadable_pins_fail_the_task(env: dict[str, str], monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
code, output = run_main(monkeypatch, tmp_path, files_workspace_yaml(), OUSAST_GIT_PINS='{"case/code": "abc"}')
assert code == EXIT_BY_STATUS["failed"]
assert "OUSAST_GIT_PINS" in json.loads((output / "summary.json").read_text())["reason"]
def test_git_commits_annotation_round_trips_to_the_checkout(
env: dict[str, str], monkeypatch: pytest.MonkeyPatch, tmp_path: Path, upstream: tuple[Path, str]
) -> None:
"""Workspace annotation -> reconciler -> ``OUSAST_GIT_PINS`` in the Task env -> the runner checks out the pin."""
import yaml
from openultrasast.plane import reconciler
repo, pin = upstream
output, workspace = tmp_path / "out", tmp_path / "ws"
manifest = tmp_path / "run.yaml"
manifest.write_text(
"apiVersion: ax.io/v1alpha1\nkind: Workspace\nmetadata:\n name: case\n"
f" annotations: {{openultrasast.io/git-commits: 'code={pin}'}}\n"
f"spec:\n git: [{{name: code, repo: '{repo.as_uri()}', branch: main}}]\n---\n"
"apiVersion: ax.io/v1alpha1\nkind: Task\nmetadata: {name: stub}\n"
f"spec: {{command: [stub-task], workspaces: [{{name: case, path: '{workspace}'}}]}}\n---\n"
"apiVersion: openultrasast.io/v1alpha1\nkind: Run\nmetadata: {name: rt}\nspec: {tasks: [{name: stub, task: stub}]}\n"
)
run, manifests = reconciler.load_run(manifest)
task_doc, *rest = reconciler.render_task(run, run.tasks[0], manifests, tmp_path, "unused")
assert rest[0]["metadata"] == {"name": "case"}, "annotations are stripped before ax"
assert rest[0]["spec"]["git"] == [{"name": "code", "repo": repo.as_uri(), "branch": "main"}], "no commit for ax"
overrides = {"OUSAST_OUTPUT_DIR": str(output)}
task_doc["spec"]["env"] = [
{"name": e["name"], "value": overrides.get(e["name"], e["value"])}
for e in task_doc["spec"]["env"]
if e["name"] != "OUSAST_ARTIFACT_URL"
]
assert json.loads({e["name"]: e["value"] for e in task_doc["spec"]["env"]}["OUSAST_GIT_PINS"]) == {"case/code": pin}
monkeypatch.setenv("AX_TASK_YAML", yaml.safe_dump(task_doc))
monkeypatch.setenv("AX_WORKSPACES_YAML", yaml.safe_dump_all([d for d in rest if d["kind"] == "Workspace"]))
assert runner.main() == 0, (output / "summary.json").read_text()
checkout = workspace / "code"
assert git("rev-parse", "HEAD", cwd=checkout) == pin, "the pinned commit, not the branch tip"
assert (checkout / "hello.py").read_text() == "print('v1')\n"