Fix ephemeral scheduler pytest fixture cleanup

scripts/aoa_session_memory.py

@@ -196843,6 +196843,7 @@ def skill_session_evidence_probe(
     manifest_cache: dict[str, dict[str, Any]] = {}
     raw_preview_cache: dict[tuple[str, str, str, int], dict[str, Any]] = {}
     hits: dict[str, dict[str, Any]] = {}
+    unverifiable_hits: dict[str, dict[str, Any]] = {}
     admitted_read_entries: list[
         tuple[
             dict[str, Any],
@@ -196967,6 +196968,13 @@ def skill_session_evidence_probe(
             or observation_status in {"failed", "timed_out"}
         ):
             result_failures += 1
+            hit["entity_usage_role_override"] = "context"
+            hit["skill_evidence_state"] = "mentioned"
+            hit["skill_state_admission"] = (
+                "raw_exact_skill_artifact_read_without_"
+                "correlation_owned_nonfailure_result"
+            )
+            unverifiable_hits[str(hit.get("doc_id") or "")] = hit
             continue
         result_hit = session_query_source_hit(
             scan,
@@ -197329,6 +197337,7 @@ def skill_session_evidence_probe(
                 ]
             ),
             "_hits": list(hits.values()),
+            "_unverifiable_hits": list(unverifiable_hits.values()),
         }
     )
     return packet
@@ -200912,6 +200921,11 @@ def entity_usage_audit(
         for hit in skill_query_time_source_probe.pop("_hits", [])
         if isinstance(hit, dict)
     ] if normalized_kind == "skill" else []
+    skill_query_time_source_unverifiable_hits = [
+        hit
+        for hit in skill_query_time_source_probe.pop("_unverifiable_hits", [])
+        if isinstance(hit, dict)
+    ] if normalized_kind == "skill" else []
     if exact_instance_mode:
         mcp_tool_query_time_source_hits = [
             hit
@@ -200990,6 +201004,39 @@ def entity_usage_audit(
             source_hit,
             "session_scoped_query_time_exact_correlation_probe",
         )
+
+    unverifiable_skill_read_doc_ids = {
+        str(hit.get("doc_id") or "")
+        for hit in skill_query_time_source_unverifiable_hits
+        if str(hit.get("doc_id") or "")
+    }
+
+    def demote_unverifiable_skill_read_hit(hit: dict[str, Any]) -> None:
+        """Keep a failed exact read from becoming a heuristic skill read."""
+        if normalized_kind != "skill":
+            return
+        doc_id = str(hit.get("doc_id") or "")
+        if doc_id not in unverifiable_skill_read_doc_ids:
+            return
+        current_state = effective_skill_evidence_state_for_event(
+            compact_usage_event_from_search_hit(hit),
+            anchor=anchor,
+        )
+        if current_state != "skill_read":
+            return
+        hit["entity_usage_role_override"] = "context"
+        hit["skill_evidence_state"] = "mentioned"
+        hit["skill_state_admission"] = (
+            "raw_exact_skill_artifact_read_without_"
+            "correlation_owned_nonfailure_result"
+        )
+
+    def demote_unverifiable_skill_read_hits() -> None:
+        if normalized_kind != "skill":
+            return
+        for hit in merged.values():
+            demote_unverifiable_skill_read_hit(hit)
+
     route_result_summaries: list[dict[str, Any]] = []
     usage_role_fast_path_supported = bool(
         lookup_candidates
@@ -201797,6 +201844,7 @@ def bounded_route_results(
         return selected, truncated
 
     def merged_route_counts() -> tuple[int, int, int]:
+        demote_unverifiable_skill_read_hits()
         apply_mcp_tool_usage_invocation_admission()
         apply_tool_usage_invocation_admission()
         apply_mcp_usage_invocation_admission()
@@ -201968,6 +202016,7 @@ def merged_route_counts() -> tuple[int, int, int]:
                 skill_dispatch_entity_probe_result_count += len(dispatch_hits)
                 dispatch_candidates: list[tuple[dict[str, Any], dict[str, Any], str]] = []
                 for hit in dispatch_hits:
+                    demote_unverifiable_skill_read_hit(hit)
                     event = compact_usage_event_from_search_hit(hit)
                     state = effective_skill_evidence_state_for_event(
                         event,
@@ -202173,6 +202222,7 @@ def merged_route_counts() -> tuple[int, int, int]:
     route_evidence_satisfies_kind = route_usage_hit_count > 0 or (
         normalized_kind not in ENTITY_USAGE_DIRECT_TRACE_KINDS and route_evidence_hit_count > 0
     )
+    demote_unverifiable_skill_read_hits()
     skill_route_state_counts: Counter[str] = Counter()
     if normalized_kind == "skill":
         for route_hit in merged.values():
@@ -202214,6 +202264,7 @@ def merged_route_counts() -> tuple[int, int, int]:
         )
     if isinstance(skill_prompt_visibility, dict):
         skill_prompt_visibility.pop("_hits", None)
+    demote_unverifiable_skill_read_hits()
     skill_text_fallback_deferred = False
     if (
         not skip_text_search

scripts/pytest_scheduler_experiment.py

@@ -218,6 +218,7 @@ def _pytest_argv(
     method: Method,
     *,
     junit_path: Path,
+    basetemp: Path,
     nodeids: Sequence[str] | None = None,
     targets: Sequence[str] | None = None,
     collect_only: bool = False,
@@ -235,6 +236,10 @@ def _pytest_argv(
         str(REPO_ROOT),
         "--confcutdir",
         str(REPO_ROOT),
+        # Keep pytest fixture state inside this process's job artifact. xdist
+        # adds one popen-gwN child beneath this path for each worker.
+        "--basetemp",
+        str(basetemp),
         "-p",
         PROBE_MODULE,
     ]
@@ -534,6 +539,7 @@ def _run_static(
         _pytest_argv(
             method,
             junit_path=artifact_root / "unused.xml",
+            basetemp=artifact_root / "pytest-basetemp" / "collection",
             targets=targets,
             collect_only=True,
         ),
@@ -584,6 +590,7 @@ def run_shard(index: int, shard: list[str]) -> tuple[dict[str, Any], list[dict[s
             _pytest_argv(
                 method,
                 junit_path=artifact_root / f"{step_id}.junit.xml",
+                basetemp=artifact_root / "pytest-basetemp" / step_id,
                 nodeids=shard,
             ),
             env=shard_env,
@@ -647,7 +654,11 @@ def run_trial(args: argparse.Namespace) -> dict[str, Any]:
             ),
         )
     with tempfile.TemporaryDirectory(
-        prefix="aoa-session-memory-pytest-",
+        # Keep the scheduler's temporary path semantically neutral.  Session
+        # memory derives route/session facets from path text, so an owner name
+        # such as ``aoa-session-memory`` here can change the meaning of a
+        # fixture's synthetic command path.
+        prefix="pytest-scheduler-",
         dir=tempfile.gettempdir(),
     ) as temporary_root:
         return _run_trial(
@@ -667,6 +678,9 @@ def _run_trial(
     artifact_root.mkdir(parents=True, exist_ok=True)
     if any(artifact_root.iterdir()):
         raise ExperimentError(f"artifact root must start empty: {artifact_root}")
+    # Pytest creates each process-private basetemp below this shared parent;
+    # the parent must exist before pytest receives --basetemp.
+    (artifact_root / "pytest-basetemp").mkdir()
     experiment = receipt_path is not None
     started_at = dt.datetime.now(dt.UTC)
     started = time.monotonic()
@@ -716,6 +730,7 @@ def _run_trial(
                 _pytest_argv(
                     method,
                     junit_path=artifact_root / "pytest.junit.xml",
+                    basetemp=artifact_root / "pytest-basetemp" / "pytest",
                     targets=targets,
                 ),
                 env=run_env,

tests/test_session_memory.py

@@ -18414,7 +18414,10 @@ def test_skill_read_query_time_probe_requires_correlation_owned_nonfailure_resul
     tmp_path: Path,
 ) -> None:
     """A SKILL.md read survives stale/missing search only with its own result."""
-    workspace = tmp_path / "AbyssOS"
+    # Keep the fixture under an owner-named directory.  Query-time evidence
+    # must remain correct when a scheduler basetemp contributes
+    # ``aoa-session-memory`` path text to the synthetic command paths.
+    workspace = tmp_path / "aoa-session-memory-path-fixture" / "AbyssOS"
     aoa_root = workspace / ".aoa"
     skill_path = tmp_path / ".codex" / "skills" / "gold-route" / "SKILL.md"
     transcript = tmp_path / "rollout-skill-read-query-time.jsonl"

tests/test_validation_scheduler_experiment.py

@@ -105,6 +105,80 @@ def test_ordinary_route_uses_fresh_external_bytecode_prefix(tmp_path: Path) -> N
     assert pycache_root.is_dir()
 
 
+@pytest.mark.parametrize("failed_step", (None, "shard-1"))
+def test_ephemeral_scheduler_binds_and_reclaims_each_pytest_temp_root(
+    monkeypatch: pytest.MonkeyPatch,
+    failed_step: str | None,
+) -> None:
+    nodeids = ("tests/test_example.py::test_one", "tests/test_example.py::test_two")
+    calls: list[tuple[str, Path, Path]] = []
+    artifact_roots: list[Path] = []
+    fixture_was_created: list[bool] = []
+
+    def fake_run_process(
+        step_id: str,
+        argv: list[str],
+        *,
+        env: dict[str, str],
+        artifact_root: Path,
+        timeout_seconds: float,
+    ) -> dict[str, object]:
+        del timeout_seconds
+        basetemp = Path(argv[argv.index("--basetemp") + 1])
+        assert basetemp.parent.is_dir()
+        basetemp.mkdir()
+        fixture_path = basetemp / "fixture-marker"
+        fixture_path.mkdir()
+        probe_path = Path(env[pytest_scheduler_experiment.PROBE_LOG_ENV])
+        probe_path.parent.mkdir(parents=True, exist_ok=True)
+        if step_id == "collection":
+            events = [{"event": "collection", "worker": "controller", "nodeids": list(nodeids)}]
+        else:
+            junit_index = argv.index("--junitxml")
+            selected = argv[junit_index + 2 :]
+            events = [
+                {
+                    "event": "report",
+                    "nodeid": nodeid,
+                    "when": "call",
+                    "outcome": "passed",
+                    "duration_seconds": 0.001,
+                }
+                for nodeid in selected
+            ]
+        probe_path.write_text(
+            "".join(json.dumps(event) + "\n" for event in events),
+            encoding="utf-8",
+        )
+        artifact_roots.append(artifact_root)
+        calls.append((step_id, basetemp, fixture_path))
+        fixture_was_created.append(fixture_path.is_dir())
+        return {
+            "id": step_id,
+            "returncode": 1 if step_id == failed_step else 0,
+            "timed_out": False,
+            "stdout": {"path": str(artifact_root / f"{step_id}.stdout"), "tail": ""},
+            "stderr": {"path": str(artifact_root / f"{step_id}.stderr"), "tail": ""},
+        }
+
+    monkeypatch.setattr(pytest_scheduler_experiment, "source_test_targets", lambda: nodeids)
+    monkeypatch.setattr(pytest_scheduler_experiment, "_run_process", fake_run_process)
+    args = pytest_scheduler_experiment.build_parser().parse_args(["--method", "static2"])
+
+    result = pytest_scheduler_experiment.run_trial(args)
+
+    assert result["ok"] is (failed_step is None)
+    assert {step_id for step_id, _, _ in calls} == {"collection", "shard-1", "shard-2"}
+    assert len({basetemp for _, basetemp, _ in calls}) == 3
+    artifact_root = artifact_roots[0]
+    assert all(
+        basetemp.is_relative_to(artifact_root / "pytest-basetemp")
+        for _, basetemp, _ in calls
+    )
+    assert all(fixture_was_created)
+    assert artifact_roots and not artifact_roots[0].exists()
+
+
 def test_scheduler_plan_keeps_all_candidates_in_shadow() -> None:
     plan = validation_scheduler_experiment.candidate_plan()
     methods = {item["name"]: item for item in plan["methods"]}