feat(testing): make run-plan dependency aware

Author: kvinwang

docs/testing/dstack-test-methodology.md

@@ -82,7 +82,15 @@ Review the plan for change coverage, regression breadth, compatibility matrices,
 
 ### 3.5 Execute
 
-An AI executor must:
+The `run-plan` orchestration agent must first read the guide, index, and every
+case specification. It processes cases in index order, starts an independent
+case-agent session for each runnable case, and reads the completed result before
+deciding about later cases. It may mark a later case `SKIPPED` without launching
+it only when a recorded earlier non-PASS result demonstrably makes the later
+case's prerequisite false or its result meaningless. Similarity, expected cost,
+or a mere possibility of failure is not sufficient. Independent cases continue.
+
+Each case executor must:
 
 1. read the plan `README.md` and `index.json`;
 2. execute cases in index order unless the guide explicitly permits parallelism;

docs/testing/test-report-output-spec.md

@@ -8,7 +8,8 @@ This document defines the normative format for AI sessions, case summaries, run
 <a id="report-principles"></a>
 ## 1. Principles
 
-1. Each case runs in an independent Codex or Claude session.
+1. `run-plan` uses an AI scheduling session before each case, and each case it
+   executes runs in an independent Codex or Claude session.
 2. The agent's native JSONL is the primary source for commands, tool calls, and raw output.
 3. The agent writes one shallow `result.json`; it does not copy command output into that file.
 4. The runner generates `runner.json`, timestamps, exit code, checksums, and run aggregates.
@@ -48,6 +49,11 @@ dstack-test run-case \
 
 The runner supplies the plan guide, `case.md`, output location, status rules, and result schema in the prompt. The agent must read the guide before the case and must not modify plan specifications.
 
+For a dependency-driven skip, the orchestrator creates a synthetic one-event
+case session containing the reason and causal earlier case IDs. It does not
+pretend that the skipped case was executed. The full decision process remains
+available under the run-level `decisions/` directory.
+
 <a id="report-layout"></a>
 ## 4. Result layout
 

tools/dstack-test/README.md

@@ -42,8 +42,16 @@ tools/dstack-test/dstack-test run-plan \
   -- "Follow the environment restrictions in README.md"
 ```
 
-Every case receives an independent agent session. `run-plan` continues after an
-agent/runner failure and finalizes an `INCOMPLETE` run when necessary.
+Before each case, `run-plan` starts a scheduling-agent session that reads the
+guide, index, all case specifications, and prior results. The runner—not the
+scheduling agent—then invokes `run-case` in a fresh session when the decision is
+`RUN`. When a
+prior non-PASS result demonstrably invalidates a later prerequisite, it records
+that later case as `SKIPPED` with the causal case IDs instead of starting an
+unnecessary case session. It must not skip independent or merely expensive
+cases. The complete orchestration session is stored at
+`<plan>/results/<run-id>/decisions/`. Keeping scheduling and case execution in
+separate processes avoids nested agent CLIs and preserves trust boundaries.
 
 ## Finalize and validate
 

tools/dstack-test/dstack-test

@@ -319,6 +319,100 @@ def run_case(
     return {"case": case.id, "status": result["status"], "result_dir": str(result_dir)}
 
 
+def skip_case(
+    case: render.CaseEntry,
+    run_id: str,
+    reason: str,
+    blocked_by: list[str],
+    model: str | None = None,
+) -> dict[str, Any]:
+    result_dir = case.path / "results" / run_id
+    if result_dir.exists() and any(result_dir.iterdir()):
+        raise DstackTestError(f"result directory is not empty: {result_dir}")
+    result_dir.mkdir(parents=True, exist_ok=True)
+    now = utc_now()
+    event = {
+        "type": "orchestrator.skip",
+        "case_id": case.id,
+        "reason": reason,
+        "blocked_by": blocked_by,
+    }
+    (result_dir / "session.jsonl").write_text(
+        json.dumps(event, ensure_ascii=False) + "\n", encoding="utf-8"
+    )
+    result = {
+        "schema_version": "1.0",
+        "case_id": case.id,
+        "status": "SKIPPED",
+        "summary": reason,
+        "steps": [],
+        "artifacts": [],
+        "remarks": "Skipped by the run-plan orchestrator after evaluating prior results.",
+    }
+    atomic_json(result_dir / "result.json", result)
+    atomic_json(
+        result_dir / "runner.json",
+        {
+            "schema_version": "1.0",
+            "run_id": run_id,
+            "case_id": case.id,
+            "agent": {
+                "type": "run-plan-scheduler",
+                "model": model or "unknown",
+            },
+            "session": {
+                "format": "orchestrator-decision-jsonl",
+                "path": "session.jsonl",
+                "events": 1,
+            },
+            "result_path": "result.json",
+            "started_at": now,
+            "finished_at": now,
+            "duration_ms": 0,
+            "exit_code": 0,
+            "result_valid": True,
+            "result_error": None,
+        },
+    )
+    write_sha256s(result_dir)
+    return {"case": case.id, "status": "SKIPPED", "result_dir": str(result_dir)}
+
+
+def orchestration_prompt(
+    plan: render.Plan,
+    case: render.CaseEntry,
+    prior: list[dict[str, Any]],
+    decision_path: Path,
+    extra_prompt: str,
+) -> str:
+    prior_json = json.dumps(prior, ensure_ascii=False, indent=2)
+    return f"""You are the dstack test-plan scheduling agent. Decide whether the next case should be executed; do not execute the case yourself.
+
+Read the plan guide at {plan.guide_path}, index at {plan.root / "index.json"}, every case specification, and the prior results below. The next case in authoritative order is {case.id} at {case.spec_path}.
+
+Prior completed results:
+{prior_json}
+
+Decision rules:
+- Choose RUN unless a recorded prior non-PASS result demonstrably makes this case's prerequisite false or makes its result meaningless.
+- Choose SKIP only for such a proven dependency consequence. Similarity, cost, likely failure, or an independently checkable prerequisite is not sufficient.
+- A SKIP decision must name one or more causal earlier case IDs in blocked_by and explain the exact dependency.
+- Never reinterpret a product failure as a scheduling failure. Independent cases continue after FAIL, BLOCKED, or SKIPPED.
+- Do not modify the plan or any case result.
+
+Atomically write exactly this shallow JSON object to {decision_path}:
+{{
+  "decision": "RUN|SKIP",
+  "reason": "concise evidence-based reason",
+  "blocked_by": ["prior-case-id"]
+}}
+For RUN, blocked_by must be empty. Finish after writing the decision.
+
+Additional caller constraints:
+{extra_prompt or "None"}
+"""
+
+
 def write_sha256s(root: Path) -> None:
     entries: list[str] = []
     for path in sorted(
@@ -609,37 +703,109 @@ def main(argv: list[str] | None = None) -> int:
             args.overwrite,
         )
     elif args.subcommand == "run-plan":
-        values = []
-        total = len(plan.cases)
-        for index, case in enumerate(plan.cases, 1):
-            print(f"[{index}/{total}] starting {case.id}", file=sys.stderr, flush=True)
-            try:
-                case_value = run_case(
+        run_dir = plan.root / "results" / args.run_id
+        existing = run_dir.exists() or any(
+            (case.path / "results" / args.run_id).exists() for case in plan.cases
+        )
+        if existing and not args.overwrite:
+            raise DstackTestError(
+                f"run {args.run_id!r} already exists; pass --overwrite to replace it"
+            )
+        if args.overwrite:
+            shutil.rmtree(run_dir, ignore_errors=True)
+            for case in plan.cases:
+                shutil.rmtree(case.path / "results" / args.run_id, ignore_errors=True)
+        decisions_dir = run_dir / "decisions"
+        decisions_dir.mkdir(parents=True, exist_ok=True)
+        effective_model = resolve_model(args.agent, args.model)
+        prior: list[dict[str, Any]] = []
+        for number, case in enumerate(plan.cases, 1):
+            print(
+                f"[{number}/{len(plan.cases)}] scheduling {case.id}",
+                file=sys.stderr,
+                flush=True,
+            )
+            decision_path = decisions_dir / f"{case.id}.json"
+            prompt = orchestration_prompt(
+                plan, case, prior, decision_path, prompt_text(args)
+            )
+            (decisions_dir / f"{case.id}.prompt.md").write_text(
+                prompt, encoding="utf-8"
+            )
+            command = agent_command(
+                args.agent,
+                effective_model,
+                args.workdir.resolve(),
+                prompt,
+                args.agent_arg,
+            )
+            with (
+                (decisions_dir / f"{case.id}.session.jsonl").open("wb") as stdout,
+                (decisions_dir / f"{case.id}.stderr.log").open("wb") as stderr,
+            ):
+                completed = subprocess.run(
+                    command, cwd=args.workdir, stdout=stdout, stderr=stderr, check=False
+                )
+            validate_session(decisions_dir / f"{case.id}.session.jsonl")
+            if completed.returncode != 0:
+                raise DstackTestError(
+                    f"scheduler failed for {case.id} with exit code {completed.returncode}"
+                )
+            decision = render.load_json(decision_path)
+            action = decision.get("decision")
+            reason = decision.get("reason")
+            blocked_by = decision.get("blocked_by", [])
+            if (
+                action not in ("RUN", "SKIP")
+                or not isinstance(reason, str)
+                or not reason.strip()
+                or not isinstance(blocked_by, list)
+            ):
+                raise DstackTestError(f"invalid scheduling decision for {case.id}")
+            prior_by_id = {item["case_id"]: item for item in prior}
+            if action == "SKIP":
+                if not blocked_by or any(
+                    ref not in prior_by_id or prior_by_id[ref]["status"] == "PASS"
+                    for ref in blocked_by
+                ):
+                    raise DstackTestError(
+                        f"invalid blocked_by references for {case.id}"
+                    )
+                skip_case(case, args.run_id, reason, blocked_by, effective_model)
+                result = {"case_id": case.id, "status": "SKIPPED", "summary": reason}
+            else:
+                if blocked_by:
+                    raise DstackTestError(
+                        f"RUN decision has blocked_by entries for {case.id}"
+                    )
+                run_case(
                     plan,
                     case,
                     args.run_id,
                     args.agent,
-                    args.model,
+                    effective_model,
                     args.workdir.resolve(),
                     prompt_text(args),
                     args.agent_arg,
-                    args.overwrite,
-                )
-                values.append(case_value)
-                print(
-                    f"[{index}/{total}] finished {case.id}: {case_value['status']}",
-                    file=sys.stderr,
-                    flush=True,
+                    False,
                 )
-            except DstackTestError as error:
-                values.append({"case": case.id, "runner_error": str(error)})
-                print(
-                    f"[{index}/{total}] runner error for {case.id}: {error}",
-                    file=sys.stderr,
-                    flush=True,
+                result = validate_summary(
+                    case, case.path / "results" / args.run_id / "result.json"
                 )
+            prior.append(
+                {
+                    "case_id": case.id,
+                    "status": result["status"],
+                    "summary": result["summary"],
+                }
+            )
+            print(
+                f"[{number}/{len(plan.cases)}] {case.id}: {result['status']}",
+                file=sys.stderr,
+                flush=True,
+            )
         run = finalize_run(plan, args.run_id, context_value(args.context))
-        value = {"cases": values, "run": run}
+        value = {"decisions": prior, "run": run}
     elif args.subcommand == "finalize":
         value = finalize_run(plan, args.run_id, context_value(args.context))
     elif args.subcommand == "validate":

tools/dstack-test/tests/test_dstack_test.py

@@ -56,6 +56,20 @@ def test_codex_model_is_read_from_config(self) -> None:
                     dstack_test.resolve_model("codex", None), "test-codex-model"
                 )
 
+    def test_orchestrator_can_record_dependency_skip(self) -> None:
+        with tempfile.TemporaryDirectory() as temporary:
+            plan_path = self.copy_fixture(Path(temporary))
+            plan = render.load_plan(plan_path)
+            case = plan.cases[0]
+            value = dstack_test.skip_case(
+                case, "run-skip", "prerequisite case failed", ["tc-prereq-001"]
+            )
+            self.assertEqual(value["status"], "SKIPPED")
+            result_dir = case.path / "results" / "run-skip"
+            result = dstack_test.validate_summary(case, result_dir / "result.json")
+            self.assertEqual(result["status"], "SKIPPED")
+            self.assertIn("tc-prereq-001", (result_dir / "session.jsonl").read_text())
+
     def test_validate_and_render_fixture(self) -> None:
         plan = render.load_plan(FIXTURE)
         valid = dstack_test.validate_run(plan, "run-demo")