From 547b2d14b6e966634a1c9cc4bf37faf4ea94fc0d Mon Sep 17 00:00:00 2001 From: 8Dionysus Date: Sat, 5 Sep 2026 22:46:14 -0600 Subject: [PATCH 1/2] fix(validation): bind pytest fixtures to scheduler artifacts --- scripts/pytest_scheduler_experiment.py | 11 +++ tests/test_validation_scheduler_experiment.py | 74 +++++++++++++++++++ 2 files changed, 85 insertions(+) diff --git a/scripts/pytest_scheduler_experiment.py b/scripts/pytest_scheduler_experiment.py index cadf0238..d7ee7a53 100644 --- a/scripts/pytest_scheduler_experiment.py +++ b/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, @@ -667,6 +674,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 +726,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, diff --git a/tests/test_validation_scheduler_experiment.py b/tests/test_validation_scheduler_experiment.py index 3373a7f9..0a37467c 100644 --- a/tests/test_validation_scheduler_experiment.py +++ b/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"]} From fa77290882c12d6cff8e73d9b92f92fedfbffcea Mon Sep 17 00:00:00 2001 From: 8Dionysus Date: Sat, 5 Sep 2026 23:25:21 -0600 Subject: [PATCH 2/2] fix(validation): keep failed skill reads out of audit state --- scripts/aoa_session_memory.py | 51 ++++++++++++++++++++++++++ scripts/pytest_scheduler_experiment.py | 6 ++- tests/test_session_memory.py | 5 ++- 3 files changed, 60 insertions(+), 2 deletions(-) diff --git a/scripts/aoa_session_memory.py b/scripts/aoa_session_memory.py index 424a393e..ba52920e 100755 --- a/scripts/aoa_session_memory.py +++ b/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 diff --git a/scripts/pytest_scheduler_experiment.py b/scripts/pytest_scheduler_experiment.py index d7ee7a53..2bf5e165 100644 --- a/scripts/pytest_scheduler_experiment.py +++ b/scripts/pytest_scheduler_experiment.py @@ -654,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( diff --git a/tests/test_session_memory.py b/tests/test_session_memory.py index 23c02704..640e8656 100644 --- a/tests/test_session_memory.py +++ b/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"