From 136c2ef18c34eafaad744ba78c7b581ef4755592 Mon Sep 17 00:00:00 2001 From: Sinity Date: Tue, 11 Aug 2026 21:38:50 +0200 Subject: [PATCH 01/14] fix(devtools): shard testmon seed execution Problem: the 21k-node seed wrote to one testmon database through xdist workers and hit the bounded pytest timeout before it could publish a complete baseline. What changed: collect the full selected node set first, partition it deterministically, and execute each shard serially with one testmon writer. Persist shard-level node outcomes after every shard and resume only unfinished shards. Release publication now requires a terminal shard ledger as well as complete, failure-free graph coverage. --- devtools/checkout_guard.py | 2 +- devtools/testmon_state.py | 117 ++++++++++++ devtools/verify.py | 279 ++++++++++++++++++++++++++--- tests/unit/devtools/test_verify.py | 178 ++++++++++++++++-- 4 files changed, 538 insertions(+), 38 deletions(-) diff --git a/devtools/checkout_guard.py b/devtools/checkout_guard.py index 1f8b0eb35b..313512aaef 100644 --- a/devtools/checkout_guard.py +++ b/devtools/checkout_guard.py @@ -132,7 +132,7 @@ def as_dict(self) -> dict[str, object]: _TESTMON_STATE_DIR = Path(".cache/testmon") _TESTMON_STATE_MARKER = _TESTMON_STATE_DIR / "seed.json" _TESTMON_SEED_ATTEMPT = _TESTMON_STATE_DIR / "seed-attempt.json" -_TESTMON_SEED_PROTOCOL_VERSION = 6 +_TESTMON_SEED_PROTOCOL_VERSION = 7 _VERIFY_STATE_DIR = Path(".cache/verify") _VERIFY_STATE_MARKER = _VERIFY_STATE_DIR / "current-run.json" diff --git a/devtools/testmon_state.py b/devtools/testmon_state.py index ec78f85dca..1929ff801b 100644 --- a/devtools/testmon_state.py +++ b/devtools/testmon_state.py @@ -53,6 +53,15 @@ class SeedAttemptOutcome(StrEnum): RESOURCE_TIMEOUT = "resource-timeout" +class SeedShardStatus(StrEnum): + """Durable state of one sequential pytest-testmon seed shard.""" + + PENDING = "pending" + RUNNING = "running" + COMPLETE = "complete" + INCOMPLETE = "incomplete" + + class BindingMode(StrEnum): EXACT = "exact" RELATIVE_FILE_FINGERPRINTS = "relative-file-fingerprints" @@ -69,6 +78,110 @@ class TerminalAuthorization(StrEnum): NARROW_TERMINAL = "narrow-terminal" +_TERMINAL_NODE_OUTCOMES = frozenset({"passed", "failed", "error", "skipped"}) + + +def seed_shard_plan(nodeids: Sequence[str], *, shard_size: int) -> list[dict[str, Any]]: + """Partition a complete node set into stable, contiguous, serial shards.""" + if shard_size <= 0: + raise ValueError("testmon seed shard_size must be positive") + if not nodeids or any(not nodeid for nodeid in nodeids): + raise ValueError("testmon seed nodeids must be non-empty strings") + if len(set(nodeids)) != len(nodeids): + raise ValueError("testmon seed nodeids must be unique") + ordered = tuple(sorted(nodeids)) + return [ + { + "index": index, + "nodeids": list(ordered[offset : offset + shard_size]), + "nodeid_count": len(ordered[offset : offset + shard_size]), + "nodeid_digest": hashlib.sha256("\n".join(ordered[offset : offset + shard_size]).encode()).hexdigest(), + "status": SeedShardStatus.PENDING.value, + "node_outcomes": [], + } + for index, offset in enumerate(range(0, len(ordered), shard_size), start=1) + ] + + +def validate_seed_shard_ledger( + shards: object, + *, + expected_nodeids: Sequence[str], +) -> list[dict[str, Any]] | None: + """Validate the full shard ledger without granting release authority. + + Every shard owns a disjoint contiguous part of the sorted expected node + set. A completed shard carries an explicit terminal result for every node; + interrupted shards remain visible and are eligible for resume. + """ + if not isinstance(shards, list) or not shards: + return None + expected = tuple(sorted(expected_nodeids)) + if not expected or len(set(expected)) != len(expected): + return None + normalized: list[dict[str, Any]] = [] + observed: list[str] = [] + for index, raw in enumerate(shards, start=1): + if not isinstance(raw, Mapping) or raw.get("index") != index: + return None + nodeids = raw.get("nodeids") + if ( + not isinstance(nodeids, list) + or not nodeids + or any(not isinstance(nodeid, str) or not nodeid for nodeid in nodeids) + or nodeids != sorted(nodeids) + ): + return None + if raw.get("nodeid_count") != len(nodeids): + return None + if raw.get("nodeid_digest") != hashlib.sha256("\n".join(nodeids).encode()).hexdigest(): + return None + raw_status = raw.get("status") + if not isinstance(raw_status, str): + return None + try: + status = SeedShardStatus(raw_status) + except (TypeError, ValueError): + return None + outcomes = raw.get("node_outcomes") + if not isinstance(outcomes, list): + return None + outcome_by_node: dict[str, dict[str, Any]] = {} + for outcome in outcomes: + if not isinstance(outcome, Mapping): + return None + nodeid = outcome.get("nodeid") + state = outcome.get("outcome") + if not isinstance(nodeid, str) or nodeid not in nodeids or not isinstance(state, str): + return None + if nodeid in outcome_by_node: + return None + outcome_by_node[nodeid] = dict(outcome) + if status is SeedShardStatus.PENDING and outcomes: + return None + if status is SeedShardStatus.COMPLETE and ( + set(outcome_by_node) != set(nodeids) + or any(item.get("outcome") not in _TERMINAL_NODE_OUTCOMES for item in outcome_by_node.values()) + ): + return None + if ( + status in {SeedShardStatus.RUNNING, SeedShardStatus.INCOMPLETE} + and outcomes + and set(outcome_by_node) != set(nodeids) + ): + return None + normalized.append(dict(raw)) + observed.extend(nodeids) + if tuple(observed) != expected: + return None + return normalized + + +def seed_shard_ledger_is_terminal(shards: Sequence[Mapping[str, Any]]) -> bool: + """Return whether every planned shard completed with explicit node results.""" + return all(shard.get("status") == SeedShardStatus.COMPLETE.value for shard in shards) + + @dataclass(frozen=True, slots=True) class TestmonIdentity: git_head: str | None @@ -870,6 +983,10 @@ def stamp_from_attempt( expected_count = attempt.get("expected_count") if not isinstance(expected_count, int) or isinstance(expected_count, bool) or expected_count != len(expected): return None + if protocol_version >= 7: + shards = validate_seed_shard_ledger(attempt.get("shards"), expected_nodeids=expected) + if shards is None or not seed_shard_ledger_is_terminal(shards): + return None expected_digest = attempt.get("expected_digest") if ( not isinstance(expected_digest, str) diff --git a/devtools/verify.py b/devtools/verify.py index b5ccff19df..dcae8b14e9 100644 --- a/devtools/verify.py +++ b/devtools/verify.py @@ -62,14 +62,18 @@ BindingMode, GraphStatus, SeedAttemptOutcome, + SeedShardStatus, TerminalAuthorization, TestmonBinding, TestmonSeedStamp, VerificationScope, inspect_testmon_database, refresh_stamp, + seed_shard_ledger_is_terminal, + seed_shard_plan, stamp_from_attempt, testmon_runtime_identity, + validate_seed_shard_ledger, validate_stamp, ) from devtools.verify_runs import ( @@ -216,7 +220,8 @@ def _format_completion_notification( TESTMON_SEED_STAMP = Path(".cache/testmon/seed.json") TESTMON_SEED_ATTEMPT = Path(".cache/testmon/seed-attempt.json") TESTMON_AFFECTED_STAMP = Path(".cache/testmon/affected.json") -TESTMON_SEED_PROTOCOL_VERSION = 6 +TESTMON_SEED_PROTOCOL_VERSION = 7 +TESTMON_SEED_SHARD_SIZE = 256 PYTEST_REPORT_DIR = Path(".cache/verify") PYTEST_REPORT_PATH = PYTEST_REPORT_DIR / "last-pytest.json" PYTEST_JUNIT_REPORT_DIR = Path(".cache/test-reports") @@ -1982,14 +1987,11 @@ def build_verify_steps( ] base_marker = f"not slow and {scale_marker_expr}" if skip_slow else scale_marker_expr if seed_testmon: - pytest_cmd.extend(["-m", base_marker, "--testmon"]) - if resume_testmon_seed: - pytest_cmd.append("--testmon-forceselect") - label = "pytest seed-testmon (resume)" - else: - pytest_cmd.append("--testmon-noselect") - label = "pytest seed-testmon" - pytest_cmd.extend(_pytest_worker_args(maximum=4)) + # Collection produces the exact corpus contract before any testmon + # write. Shards below are generated from this ledger and run one + # at a time, so pytest-testmon has exactly one SQLite writer. + pytest_cmd.extend(["-m", base_marker, "--collect-only", "-n", "0"]) + label = "pytest seed-testmon collect (resume)" if resume_testmon_seed else "pytest seed-testmon collect" steps.append((label, pytest_cmd)) elif full_pytest: # #1775: the full diagnostic runs as two lanes. The bulk lane keeps @@ -2567,6 +2569,7 @@ def _prepare_testmon_seed_attempt( "expected_count": len(expected), "expected_digest": hashlib.sha256("\n".join(sorted(expected)).encode()).hexdigest() if expected else None, "prior_node_outcomes": prior_outcomes, + "shards": list(prior.get("shards", [])) if prior is not None and isinstance(prior.get("shards"), list) else [], "started_at": datetime.now(timezone.utc).isoformat(), "run_id": run.run_id, "artifact_dir": str(run.relative_run_dir), @@ -2578,6 +2581,136 @@ def _prepare_testmon_seed_attempt( return payload +def _seed_selection_nodeids(selection: Mapping[str, Any]) -> list[str] | None: + """Accept only a complete, untruncated collection ledger.""" + nodeids = selection.get("selected_nodeids") + selected_count = selection.get("selected_count") + omitted = selection.get("selected_nodeids_omitted") + if ( + not isinstance(nodeids, list) + or not nodeids + or any(not isinstance(nodeid, str) or not nodeid for nodeid in nodeids) + or len(set(nodeids)) != len(nodeids) + or not isinstance(selected_count, int) + or isinstance(selected_count, bool) + or selected_count != len(nodeids) + or not isinstance(omitted, int) + or isinstance(omitted, bool) + or omitted != 0 + ): + return None + return sorted(nodeids) + + +def _prepare_testmon_seed_shards( + prepared: Mapping[str, Any], + *, + selection: Mapping[str, Any] | None, +) -> dict[str, Any]: + """Persist the full planned corpus before the first testmon DB mutation.""" + expected = _testmon_seed_expected_nodeids(prepared) if prepared.get("resume") else [] + if not expected: + expected = _seed_selection_nodeids(selection or {}) or [] + prior_shards = validate_seed_shard_ledger(prepared.get("shards"), expected_nodeids=expected) + shards = ( + prior_shards + if prior_shards is not None + else (seed_shard_plan(expected, shard_size=TESTMON_SEED_SHARD_SIZE) if expected else []) + ) + payload = { + **dict(prepared), + "expected_nodeids": expected, + "expected_count": len(expected), + "expected_digest": hashlib.sha256("\n".join(expected).encode()).hexdigest() if expected else None, + "selection": dict(selection or {}), + "shard_size": TESTMON_SEED_SHARD_SIZE, + "shards": shards, + } + _atomic_write_json(TESTMON_SEED_ATTEMPT, payload) + return payload + + +def _seed_shard_command(collection_command: Sequence[str], shard: Mapping[str, Any]) -> list[str]: + """Build a serial, explicit-node pytest-testmon invocation for one shard.""" + nodeids = shard.get("nodeids") + if not isinstance(nodeids, list) or not nodeids: + raise ValueError("testmon seed shard is missing nodeids") + command = [argument for argument in collection_command if argument != "--collect-only"] + command.extend(["--testmon", "--testmon-noselect", *nodeids]) + return command + + +def _seed_shard_outcomes(shards: Sequence[Mapping[str, Any]]) -> list[dict[str, Any]]: + """Flatten the shard ledger in canonical node order for legacy readers.""" + outcomes: dict[str, dict[str, Any]] = {} + for shard in shards: + raw_outcomes = shard.get("node_outcomes") + if not isinstance(raw_outcomes, list): + continue + for item in raw_outcomes: + if isinstance(item, Mapping) and isinstance(item.get("nodeid"), str): + outcomes[str(item["nodeid"])] = dict(item) + return [outcomes[nodeid] for nodeid in sorted(outcomes)] + + +def _checkpoint_testmon_seed_shard( + *, + prepared: Mapping[str, Any], + shard_index: int, + step: Mapping[str, Any], +) -> dict[str, Any]: + """Record one shard's result atomically before another shard may start.""" + expected = _testmon_seed_expected_nodeids(prepared) + shards = validate_seed_shard_ledger(prepared.get("shards"), expected_nodeids=expected) + if shards is None or shard_index < 1 or shard_index > len(shards): + raise ValueError("testmon seed shard ledger is malformed") + shard = dict(shards[shard_index - 1]) + nodeids = shard["nodeids"] + artifact_dir = _safe_testmon_artifact_dir(step.get("artifact_dir")) + selection = _read_json_artifact(artifact_dir / "selection.json") if artifact_dir is not None else None + selected = _seed_selection_nodeids(selection) if isinstance(selection, Mapping) else None + database = _testmon_database_state(nodeids) + prior = { + str(item["nodeid"]): item + for item in shard.get("node_outcomes", []) + if isinstance(item, Mapping) and isinstance(item.get("nodeid"), str) + } + outcomes = _seed_node_outcomes_from_events( + artifact_dir / "events.jsonl" if artifact_dir is not None else Path(".missing-testmon-events"), + expected_nodeids=nodeids, + database=database, + pytest_step=step, + prior_node_outcomes=prior, + ) + terminal = all(item.get("outcome") in {"passed", "failed", "error", "skipped"} for item in outcomes) + selection_matches = selected == nodeids + shard.update( + { + "status": SeedShardStatus.COMPLETE.value + if selection_matches and terminal + else SeedShardStatus.INCOMPLETE.value, + "started_at": shard.get("started_at") or datetime.now(timezone.utc).isoformat(), + "finished_at": datetime.now(timezone.utc).isoformat(), + "exit_code": step.get("exit"), + "artifact_dir": step.get("artifact_dir"), + "selection": dict(selection) if isinstance(selection, Mapping) else None, + "database": database, + "node_outcomes": outcomes, + "pytest_step": dict(step), + } + ) + shards[shard_index - 1] = shard + payload = { + **dict(prepared), + "status": "running", + "shards": shards, + "node_outcomes": _seed_shard_outcomes(shards), + "testmon_data": _file_fingerprint(TESTMON_DATA), + } + _atomic_write_json(TESTMON_SEED_ATTEMPT, payload) + return payload + + def _testmon_seed_terminal_authorized(prepared: Mapping[str, Any]) -> bool: identity = prepared.get("identity") return ( @@ -2761,22 +2894,59 @@ def _finalize_testmon_seed_attempt( and len(set(selected_nodeids)) == len(selected_nodeids) and raw_selected_count == len(selected_nodeids) ) - expected_raw = prepared.get("expected_nodeids") if prepared.get("resume") else selected_nodeids + prepared_expected = prepared.get("expected_nodeids") + expected_raw = prepared_expected if isinstance(prepared_expected, list) and prepared_expected else selected_nodeids expected = list(expected_raw) if isinstance(expected_raw, list) else [] - omitted = raw_omitted if selection_valid else 1 - database = _testmon_database_state(expected) - node_outcomes = _seed_node_outcomes_from_events( - events_path or Path(".missing-testmon-events"), - expected_nodeids=expected, - database=database, - pytest_step=pytest_step, - use_database_fallback=False, - prior_node_outcomes={ - str(item["nodeid"]): item - for item in prepared.get("prior_node_outcomes", []) - if isinstance(item, Mapping) and isinstance(item.get("nodeid"), str) - }, - ) + shards = validate_seed_shard_ledger(prepared.get("shards"), expected_nodeids=expected) + sharded = shards is not None + if sharded: + assert shards is not None + selection_valid = seed_shard_ledger_is_terminal(shards) + omitted = 0 + database = _testmon_database_state(expected) + outcome_by_node = {item["nodeid"]: item for item in _seed_shard_outcomes(shards)} + node_outcomes = [ + outcome_by_node.get(nodeid, {"nodeid": nodeid, "outcome": "missing", "reason": "shard not completed"}) + for nodeid in expected + ] + shard_steps: list[Mapping[str, Any]] = [] + for shard in shards: + raw_step = shard.get("pytest_step") + if isinstance(raw_step, Mapping): + shard_steps.append(raw_step) + if shard_steps: + pytest_step = dict(shard_steps[-1]) + timed_out = next( + ( + step + for step in shard_steps + if str(step.get("diagnosis")) in {"pytest_timeout", "pytest_stall_timeout"} + ), + None, + ) + if timed_out is not None: + pytest_step = dict(timed_out) + selection = { + "selected_count": len(expected), + "selected_nodeids_omitted": 0, + "shard_count": len(shards), + "completed_shard_count": sum(shard.get("status") == SeedShardStatus.COMPLETE.value for shard in shards), + } + else: + omitted = raw_omitted if selection_valid and isinstance(raw_omitted, int) else 1 + database = _testmon_database_state(expected) + node_outcomes = _seed_node_outcomes_from_events( + events_path or Path(".missing-testmon-events"), + expected_nodeids=expected, + database=database, + pytest_step=pytest_step, + use_database_fallback=False, + prior_node_outcomes={ + str(item["nodeid"]): item + for item in prepared.get("prior_node_outcomes", []) + if isinstance(item, Mapping) and isinstance(item.get("nodeid"), str) + }, + ) unsuccessful_nodeids = [ str(item["nodeid"]) for item in node_outcomes if item.get("outcome") not in {"passed", "skipped"} ] @@ -2811,6 +2981,20 @@ def _finalize_testmon_seed_attempt( exit_code=exit_code, pytest_step=pytest_step, ) + shard_ledger = ( + shards if sharded else seed_shard_plan(expected, shard_size=max(1, len(expected))) if expected else [] + ) + if not sharded and shard_ledger: + shard_ledger[0].update( + { + "status": ( + SeedShardStatus.COMPLETE.value + if all(item.get("outcome") in {"passed", "failed", "error", "skipped"} for item in node_outcomes) + else SeedShardStatus.INCOMPLETE.value + ), + "node_outcomes": node_outcomes, + } + ) seed_scope = ( VerificationScope.NARROW_TERMINAL.value if narrow_terminal else VerificationScope.RELEASE_BASELINE.value ) @@ -2832,6 +3016,7 @@ def _finalize_testmon_seed_attempt( else selection.get("selected_count"), "selected_nodeids_omitted": 0 if prepared.get("resume") and selection_valid else omitted, }, + "shards": shard_ledger, "node_outcomes": node_outcomes, "identity": prepared.get("identity"), "run_id": prepared.get("run_id"), @@ -2880,6 +3065,7 @@ def _finalize_testmon_seed_attempt( "collection_duration_s", ) }, + "shards": shard_ledger, "database": database, "node_outcomes": node_outcomes, "node_outcome_counts": dict( @@ -3182,6 +3368,51 @@ def main(argv: list[str] | None = None) -> int: step_result: dict[str, Any] = {"name": label, "duration_s": round(elapsed, 2), "exit": rc} step_result.update(metadata) step_results.append(step_result) + if args.seed_testmon and label.startswith("pytest seed-testmon collect"): + if rc != 0: + exit_code = rc + break + artifact_dir = _safe_testmon_artifact_dir(metadata.get("artifact_dir")) + selection = _read_json_artifact(artifact_dir / "selection.json") if artifact_dir is not None else None + assert prepared_seed_attempt is not None + prepared_seed_attempt = _prepare_testmon_seed_shards( + prepared_seed_attempt, + selection=selection if isinstance(selection, Mapping) else None, + ) + expected = _testmon_seed_expected_nodeids(prepared_seed_attempt) + shards = validate_seed_shard_ledger(prepared_seed_attempt.get("shards"), expected_nodeids=expected) + if shards is None: + exit_code = 5 + step_result["exit"] = 5 + step_result["diagnosis"] = "testmon_seed_collection_incomplete" + sys.stderr.write("verify: pytest-testmon collection did not produce a complete shard plan.\n") + break + for shard in shards: + if shard.get("status") == SeedShardStatus.COMPLETE.value: + continue + shard_index = int(shard["index"]) + shard_label = f"pytest seed-testmon shard {shard_index}/{len(shards)}" + shard_cmd = _seed_shard_command(cmd, shard) + _warn_low_memory() + shard_rc, shard_elapsed, shard_metadata = _run(shard_label, shard_cmd, run=verify_run) + shard_result: dict[str, Any] = { + "name": shard_label, + "duration_s": round(shard_elapsed, 2), + "exit": shard_rc, + "shard_index": shard_index, + "shard_count": len(shards), + "shard_nodeid_count": len(shard["nodeids"]), + } + shard_result.update(shard_metadata) + step_results.append(shard_result) + prepared_seed_attempt = _checkpoint_testmon_seed_shard( + prepared=prepared_seed_attempt, + shard_index=shard_index, + step=shard_result, + ) + if shard_rc != 0 and exit_code == 0: + exit_code = shard_rc + continue if label in {"pytest testmon", "pytest testmon (broad)"} and not args.seed_testmon and not full_pytest: _refresh_testmon_selection_attempt(step=step_result, run=verify_run, exit_code=rc) if rc != 0: diff --git a/tests/unit/devtools/test_verify.py b/tests/unit/devtools/test_verify.py index 094fe0db9d..f2c9539929 100644 --- a/tests/unit/devtools/test_verify.py +++ b/tests/unit/devtools/test_verify.py @@ -44,14 +44,17 @@ TESTMON_DATA, TESTMON_SEED_ATTEMPT, TESTMON_SEED_PROTOCOL_VERSION, + TESTMON_SEED_SHARD_SIZE, TESTMON_SEED_STAMP, _anchor_verification_paths, + _checkpoint_testmon_seed_shard, _finalize_testmon_seed_attempt, _flatten_seed_outcomes, _format_completion_notification, _matching_testmon_coverage, _parse_pytest_test_count, _prepare_testmon_seed_attempt, + _prepare_testmon_seed_shards, _pytest_command_metadata, _pytest_metadata_from_report, _pytest_stall_timeout_s, @@ -60,6 +63,7 @@ _record_testmon_affected_coverage, _run, _seed_node_outcomes_from_events, + _seed_shard_command, _stop_after_failed_step, _testmon_database_state, _testmon_preflight, @@ -254,11 +258,11 @@ def test_seed_testmon_runs_full_collection_without_selection(monkeypatch: pytest steps = build_verify_steps(quick=False, lab=False, skip_slow=False, seed_testmon=True) label, command = steps[-1] - assert label == "pytest seed-testmon" - assert "--testmon" in command - assert "--testmon-noselect" in command + assert label == "pytest seed-testmon collect" + assert "--collect-only" in command + assert "--testmon" not in command assert "-n" in command - assert command[command.index("-n") + 1] == "4" + assert command[command.index("-n") + 1] == "0" def test_seed_testmon_caps_adaptive_workers(monkeypatch: pytest.MonkeyPatch) -> None: @@ -268,8 +272,157 @@ def test_seed_testmon_caps_adaptive_workers(monkeypatch: pytest.MonkeyPatch) -> steps = build_verify_steps(quick=False, lab=False, skip_slow=False, seed_testmon=True) label, command = steps[-1] - assert label == "pytest seed-testmon" - assert command[command.index("-n") + 1] == "4" + assert label == "pytest seed-testmon collect" + assert command[command.index("-n") + 1] == "0" + + +def test_seed_shards_are_deterministic_and_use_one_testmon_writer( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.chdir(tmp_path) + expected = [f"tests/test_seed.py::test_{index:03d}" for index in range(TESTMON_SEED_SHARD_SIZE + 2)] + prepared = _prepare_testmon_seed_shards( + {"resume": False, "expected_nodeids": []}, + selection={ + "selected_count": len(expected), + "selected_nodeids": list(reversed(expected)), + "selected_nodeids_omitted": 0, + }, + ) + + shards = prepared["shards"] + assert [shard["nodeid_count"] for shard in shards] == [TESTMON_SEED_SHARD_SIZE, 2] + assert shards[0]["nodeids"] == expected[:TESTMON_SEED_SHARD_SIZE] + assert shards[1]["nodeids"] == expected[TESTMON_SEED_SHARD_SIZE:] + command = _seed_shard_command(["pytest", "--collect-only", "-n", "0"], shards[0]) + assert "--collect-only" not in command + assert command[command.index("-n") + 1] == "0" + assert "--testmon" in command + assert "--testmon-noselect" in command + assert command[-TESTMON_SEED_SHARD_SIZE:] == expected[:TESTMON_SEED_SHARD_SIZE] + + +def test_seed_shard_checkpoint_preserves_completed_shards_for_resume( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.chdir(tmp_path) + expected = ["tests/test_seed.py::test_a", "tests/test_seed.py::test_b"] + prepared = _prepare_testmon_seed_shards( + {"resume": False, "expected_nodeids": []}, + selection={"selected_count": 2, "selected_nodeids": expected, "selected_nodeids_omitted": 0}, + ) + prepared["shards"] = [ + { + **prepared["shards"][0], + "nodeids": [expected[0]], + "nodeid_count": 1, + "nodeid_digest": hashlib.sha256(expected[0].encode()).hexdigest(), + }, + { + **prepared["shards"][0], + "index": 2, + "nodeids": [expected[1]], + "nodeid_count": 1, + "nodeid_digest": hashlib.sha256(expected[1].encode()).hexdigest(), + "status": "pending", + "node_outcomes": [], + }, + ] + _atomic_payload = { + **prepared, + "expected_nodeids": expected, + "expected_count": 2, + "expected_digest": hashlib.sha256("\n".join(expected).encode()).hexdigest(), + } + artifact_dir = tmp_path / "shard-1" + artifact_dir.mkdir() + (artifact_dir / "selection.json").write_text( + json.dumps({"selected_count": 1, "selected_nodeids": [expected[0]], "selected_nodeids_omitted": 0}) + ) + (artifact_dir / "events.jsonl").write_text( + json.dumps({"event": "test_report", "nodeid": expected[0], "when": "call", "outcome": "passed"}) + "\n" + ) + + checkpointed = _checkpoint_testmon_seed_shard( + prepared=_atomic_payload, + shard_index=1, + step={"name": "pytest seed-testmon shard 1/2", "exit": 0, "artifact_dir": str(artifact_dir)}, + ) + + assert checkpointed["shards"][0]["status"] == "complete" + assert checkpointed["shards"][1]["status"] == "pending" + assert json.loads(TESTMON_SEED_ATTEMPT.read_text())["shards"][0]["node_outcomes"][0]["outcome"] == "passed" + resumed = _prepare_testmon_seed_attempt( + identity={ + "git_head": "head", + "git_tree": "tree", + "worktree_fingerprint": "fingerprint", + "python": "python", + "skip_slow": False, + "lab": False, + **_testmon_runtime_identity_fields(Path.cwd()), + }, + run=VerifyRun(tier="seed-testmon", argv=[], git_head="head", polylogue_import_path="polylogue"), + resume=True, + ) + assert resumed["shards"][0]["status"] == "complete" + assert resumed["shards"][1]["status"] == "pending" + + +def test_seed_shard_failure_remains_visible_and_blocks_release(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.chdir(tmp_path) + expected = ["tests/test_seed.py::test_failed"] + prepared = _prepare_testmon_seed_shards( + { + "protocol_version": TESTMON_SEED_PROTOCOL_VERSION, + "status": "running", + "identity": { + "git_head": "head", + "worktree_fingerprint": "tree", + "python": "python", + "skip_slow": False, + "lab": False, + **_testmon_runtime_identity_fields(Path.cwd()), + }, + "resume": False, + "run_id": "sharded-failure", + "artifact_dir": ".cache/verify/runs/sharded-failure", + }, + selection={"selected_count": 1, "selected_nodeids": expected, "selected_nodeids_omitted": 0}, + ) + artifact_dir = tmp_path / "shard-failure" + artifact_dir.mkdir() + (artifact_dir / "selection.json").write_text( + json.dumps({"selected_count": 1, "selected_nodeids": expected, "selected_nodeids_omitted": 0}) + ) + (artifact_dir / "events.jsonl").write_text( + json.dumps({"event": "test_report", "nodeid": expected[0], "when": "call", "outcome": "failed"}) + "\n" + ) + TESTMON_DATA.parent.mkdir(parents=True, exist_ok=True) + with sqlite3.connect(TESTMON_DATA) as connection: + connection.execute("create table environment (id integer primary key, environment_name text)") + connection.execute("create table file_fp (id integer primary key, filename text, fsha text)") + connection.execute("create table test_execution (id integer primary key, test_name text, failed integer)") + connection.execute("create table test_execution_file_fp (test_execution_id integer, fingerprint_id integer)") + connection.execute("insert into test_execution values (1, ?, 1)", (expected[0],)) + connection.execute("insert into file_fp values (1, 'test_seed.py', 'sha')") + connection.execute("insert into test_execution_file_fp values (1, 1)") + _write_run_receipt(tmp_path, "sharded-failure") + + checkpointed = _checkpoint_testmon_seed_shard( + prepared=prepared, + shard_index=1, + step={"name": "pytest seed-testmon shard 1/1", "exit": 1, "artifact_dir": str(artifact_dir)}, + ) + receipt = _finalize_testmon_seed_attempt( + prepared=checkpointed, + step_results=[{"name": "pytest seed-testmon shard 1/1", "exit": 1, "artifact_dir": str(artifact_dir)}], + exit_code=1, + ) + + assert receipt["shards"][0]["status"] == "complete" + assert receipt["unsuccessful_nodeids"] == expected + assert receipt["release_baseline_allowed"] is False def test_resumed_seed_uses_affected_selection_for_remaining_tests() -> None: @@ -282,10 +435,9 @@ def test_resumed_seed_uses_affected_selection_for_remaining_tests() -> None: ) label, command = steps[-1] - assert label == "pytest seed-testmon (resume)" - assert "--testmon" in command - assert "--testmon-forceselect" in command - assert "--testmon-noselect" not in command + assert label == "pytest seed-testmon collect (resume)" + assert "--collect-only" in command + assert "--testmon" not in command def test_full_verify_includes_full_pytest_without_testmon(monkeypatch: pytest.MonkeyPatch) -> None: @@ -311,14 +463,14 @@ def test_full_verify_includes_full_pytest_without_testmon(monkeypatch: pytest.Mo assert isolated_command[isolated_command.index("-n") + 1] == "0" -def test_seed_testmon_worker_count_can_be_overridden(monkeypatch: pytest.MonkeyPatch) -> None: +def test_seed_collection_refuses_parallel_worker_overrides(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setenv("POLYLOGUE_PYTEST_WORKERS", "4") steps = build_verify_steps(quick=False, lab=False, skip_slow=False, seed_testmon=True) label, command = steps[-1] - assert label == "pytest seed-testmon" - assert command[command.index("-n") + 1] == "4" + assert label == "pytest seed-testmon collect" + assert command[command.index("-n") + 1] == "0" def test_seed_defaults_to_managed_scratch(monkeypatch: pytest.MonkeyPatch) -> None: From a50b4b54887de6ece841438af9c3379a7f22a524 Mon Sep 17 00:00:00 2001 From: Sinity Date: Tue, 11 Aug 2026 21:58:31 +0200 Subject: [PATCH 02/14] fix(devtools): exclude benchmarks from testmon seed --- devtools/verify.py | 9 ++++++++- tests/unit/devtools/test_verify.py | 4 ++++ 2 files changed, 12 insertions(+), 1 deletion(-) diff --git a/devtools/verify.py b/devtools/verify.py index dcae8b14e9..18d98c9d34 100644 --- a/devtools/verify.py +++ b/devtools/verify.py @@ -1985,7 +1985,14 @@ def build_verify_steps( "-p", "devtools.pytest_progress_plugin", ] - base_marker = f"not slow and {scale_marker_expr}" if skip_slow else scale_marker_expr + # Benchmark cases are an explicit campaign surface, not part of the + # correctness/testmon seed. Keeping them out here is important: a + # benchmark marker is not necessarily paired with ``slow`` or a scale + # marker, and a serial shard would otherwise spend minutes executing a + # performance probe before it can checkpoint any correctness nodes. + base_marker = f"not benchmark and {scale_marker_expr}" + if skip_slow: + base_marker = f"not slow and {base_marker}" if seed_testmon: # Collection produces the exact corpus contract before any testmon # write. Shards below are generated from this ledger and run one diff --git a/tests/unit/devtools/test_verify.py b/tests/unit/devtools/test_verify.py index f2c9539929..f3acc865b8 100644 --- a/tests/unit/devtools/test_verify.py +++ b/tests/unit/devtools/test_verify.py @@ -508,6 +508,7 @@ def test_marker_filters_keep_testmon_selection_forced() -> None: label, command = steps[-1] assert label == "pytest testmon" marker_expr = _pytest_marker_expr(command) + assert "not benchmark" in marker_expr assert "not scale_medium" in marker_expr assert "not scale_large" in marker_expr assert "--testmon-forceselect" in command @@ -522,6 +523,7 @@ def test_skip_slow_composes_with_forced_testmon_selection() -> None: # ``scale_medium``/``scale_large``; ``--skip-slow`` composes with that # filter via ``and`` rather than replacing it. marker_expr = _pytest_marker_expr(command) + assert "not benchmark" in marker_expr assert "not slow" in marker_expr assert "not scale_medium" in marker_expr assert "not scale_large" in marker_expr @@ -535,6 +537,7 @@ def test_default_verify_excludes_medium_and_large_scale_markers() -> None: label, command = steps[-1] assert label == "pytest testmon" marker_expr = _pytest_marker_expr(command) + assert "not benchmark" in marker_expr assert "not scale_medium" in marker_expr assert "not scale_large" in marker_expr # ``scale_small`` is *not* excluded — it runs in the default gate. @@ -548,6 +551,7 @@ def test_lab_verify_includes_medium_scale_marker() -> None: pytest_step = next((label, command) for label, command in steps if label.startswith("pytest")) label, command = pytest_step marker_expr = _pytest_marker_expr(command) + assert "not benchmark" in marker_expr assert "not scale_large" in marker_expr assert "not scale_medium" not in marker_expr assert "scale_small" not in marker_expr From 31761e9785434551d9fd106fc25807889bffbc8f Mon Sep 17 00:00:00 2001 From: Sinity Date: Tue, 11 Aug 2026 22:11:11 +0200 Subject: [PATCH 03/14] fix(devtools): exclude benchmark directory from correctness seed --- devtools/verify.py | 5 +++++ tests/unit/devtools/test_verify.py | 1 + 2 files changed, 6 insertions(+) diff --git a/devtools/verify.py b/devtools/verify.py index 18d98c9d34..0db649407d 100644 --- a/devtools/verify.py +++ b/devtools/verify.py @@ -1977,6 +1977,11 @@ def build_verify_steps( "-q", "--tb=short", "--ignore=tests/integration", + # Benchmark files are an explicit campaign surface. A number of + # them are correctness-shaped and lack the benchmark marker, so a + # marker expression alone cannot keep performance probes out of + # the correctness/testmon corpus. + "--ignore=tests/benchmarks", "--durations=10", f"--junitxml={_report_dir}/verify-latest.xml", "--json-report", diff --git a/tests/unit/devtools/test_verify.py b/tests/unit/devtools/test_verify.py index f3acc865b8..2ece677b18 100644 --- a/tests/unit/devtools/test_verify.py +++ b/tests/unit/devtools/test_verify.py @@ -260,6 +260,7 @@ def test_seed_testmon_runs_full_collection_without_selection(monkeypatch: pytest label, command = steps[-1] assert label == "pytest seed-testmon collect" assert "--collect-only" in command + assert command[command.index("--ignore=tests/benchmarks")] == "--ignore=tests/benchmarks" assert "--testmon" not in command assert "-n" in command assert command[command.index("-n") + 1] == "0" From 265cf41b090a893e0ea7cdf87a86a496bb24038f Mon Sep 17 00:00:00 2001 From: Sinity Date: Wed, 12 Aug 2026 03:41:48 +0200 Subject: [PATCH 04/14] fix(testmon): preserve managed pytest event environment --- tests/conftest.py | 28 +++++++++++++++---- .../devtools/test_pytest_progress_plugin.py | 20 ++++++++++--- tests/unit/test_pytest_temp_policy.py | 16 ++++++++++- 3 files changed, 54 insertions(+), 10 deletions(-) diff --git a/tests/conftest.py b/tests/conftest.py index 6a8fdf4a36..c224d56e9b 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -317,15 +317,15 @@ def pytest_sessionfinish(session: pytest.Session, exitstatus: int) -> None: shutil.rmtree(basetemp_path, ignore_errors=True) -@pytest.hookimpl(wrapper=True) +@pytest.hookimpl(hookwrapper=True) def pytest_runtest_makereport( item: pytest.Item, call: pytest.CallInfo[None], -) -> Generator[None, pytest.TestReport, pytest.TestReport]: +) -> Generator[None, Any, None]: """Retain the call outcome so passing test temp trees can be reclaimed.""" - report = yield + outcome = yield + report = outcome.get_result() setattr(item, f"rep_{report.when}", report) - return report @pytest.fixture(autouse=True) @@ -396,6 +396,24 @@ def _reclaim_passing_test_tmp_path( ) _TESTS_ROOT = str(Path(__file__).resolve().parent) +# These variables are emitted by the managed verification supervisor and must +# survive the host-configuration scrub below. They are test-run evidence +# plumbing, not operator configuration; removing them after collection makes +# setup/call reports disappear from the event ledger while teardown still gets +# recorded, which makes interrupted seed shards look falsely successful. +_MANAGED_VERIFY_ENV = frozenset( + { + "POLYLOGUE_VERIFY_RUN_ID", + "POLYLOGUE_PYTEST_RUN_ID", + "POLYLOGUE_PYTEST_EVENTS_DIR", + "POLYLOGUE_PYTEST_EVENTS_PATH", + "POLYLOGUE_PYTEST_SELECTION_PATH", + "POLYLOGUE_PYTEST_SUMMARY_PATH", + "POLYLOGUE_PYTEST_CONTAINMENT_PATH", + "POLYLOGUE_PYTEST_SELECTION_NODEID_LIMIT", + } +) + @pytest.fixture(autouse=True) def _close_test_opened_sqlite_connections( @@ -629,7 +647,7 @@ def _clear_polylogue_env( # config) for lanes that intentionally run without packaged provider # schema data; it must survive this sweep or it could never take # effect inside the test suite that is its only consumer. - if key.startswith("POLYLOGUE_") and key != ALLOW_MISSING_SCHEMAS_ENV: + if key.startswith("POLYLOGUE_") and key not in {ALLOW_MISSING_SCHEMAS_ENV, *_MANAGED_VERIFY_ENV}: monkeypatch.delenv(key, raising=False) for key in ( diff --git a/tests/unit/devtools/test_pytest_progress_plugin.py b/tests/unit/devtools/test_pytest_progress_plugin.py index 42a06361fd..530081d626 100644 --- a/tests/unit/devtools/test_pytest_progress_plugin.py +++ b/tests/unit/devtools/test_pytest_progress_plugin.py @@ -11,7 +11,16 @@ @pytest.fixture(autouse=True) -def _restore_plugin_state() -> Iterator[None]: +def _restore_plugin_state(monkeypatch: pytest.MonkeyPatch) -> Iterator[None]: + # Unit tests own their event destinations; do not let a surrounding + # managed verify invocation redirect them into its step artifacts. + for name in ( + "POLYLOGUE_PYTEST_EVENTS_DIR", + "POLYLOGUE_PYTEST_EVENTS_PATH", + "POLYLOGUE_PYTEST_SELECTION_PATH", + "POLYLOGUE_PYTEST_SUMMARY_PATH", + ): + monkeypatch.delenv(name, raising=False) selected_count = pytest_progress_plugin._SELECTED_COUNT deselected_count = pytest_progress_plugin._DESELECTED_COUNT deselected_nodeids = list(pytest_progress_plugin._DESELECTED_NODEIDS_SAMPLE) @@ -212,6 +221,9 @@ def test_progress_plugin_records_collection_duration_and_summary( assert summary["deselected_count"] == 1 assert [report["nodeid"] for report in summary["slowest_reports"]] == ["test_slow", "test_fast"] events = [json.loads(line) for line in events_path.read_text().splitlines()] - assert events[0]["event"] == "collection_started" - assert events[1]["event"] == "collection_finished" - assert events[1]["duration_s"] == 2.5 + assert [event["event"] for event in events[:3]] == [ + "session_started", + "collection_started", + "collection_finished", + ] + assert events[2]["duration_s"] == 2.5 diff --git a/tests/unit/test_pytest_temp_policy.py b/tests/unit/test_pytest_temp_policy.py index 94bdb03eab..4672c4dba1 100644 --- a/tests/unit/test_pytest_temp_policy.py +++ b/tests/unit/test_pytest_temp_policy.py @@ -3,7 +3,7 @@ import os from pathlib import Path from types import SimpleNamespace -from typing import cast +from typing import Any, cast import pytest @@ -49,6 +49,20 @@ def _make_real_candidates( return shm, scratch +def test_runtest_makereport_wrapper_preserves_each_phase_report() -> None: + item = SimpleNamespace() + reports = [SimpleNamespace(when=phase) for phase in ("setup", "call", "teardown")] + + for report in reports: + wrapper = conftest.pytest_runtest_makereport( + cast("pytest.Item", item), cast("pytest.CallInfo[None]", SimpleNamespace()) + ) + assert next(wrapper) is None + with pytest.raises(StopIteration): + wrapper.send(cast("Any", SimpleNamespace(get_result=lambda report=report: report))) + assert getattr(item, f"rep_{report.when}") is report + + def test_managed_pytest_temp_root_defaults_to_scratch( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, From 610eb998664bb8515b0ed266fb42906aea749674 Mon Sep 17 00:00:00 2001 From: Sinity Date: Wed, 12 Aug 2026 03:53:13 +0200 Subject: [PATCH 05/14] test(devtools): guard pytest event phase capture --- .../devtools/test_pytest_progress_plugin.py | 38 +++++++++++++++++++ 1 file changed, 38 insertions(+) diff --git a/tests/unit/devtools/test_pytest_progress_plugin.py b/tests/unit/devtools/test_pytest_progress_plugin.py index 530081d626..7bf2974a05 100644 --- a/tests/unit/devtools/test_pytest_progress_plugin.py +++ b/tests/unit/devtools/test_pytest_progress_plugin.py @@ -1,6 +1,9 @@ from __future__ import annotations import json +import os +import subprocess +import sys from collections.abc import Iterator from dataclasses import dataclass from pathlib import Path @@ -68,6 +71,41 @@ def test_progress_plugin_records_call_and_setup_failures( assert events[2]["longrepr"] == "fixture exploded" +def test_managed_event_ledger_survives_test_host_environment_scrub(tmp_path: Path) -> None: + events_dir = tmp_path / "events" + env = os.environ.copy() + env.update( + { + "POLYLOGUE_PYTEST_EVENTS_DIR": str(events_dir), + "POLYLOGUE_PYTEST_SELECTION_PATH": str(tmp_path / "selection.json"), + "POLYLOGUE_PYTEST_SUMMARY_PATH": str(tmp_path / "summary.json"), + "POLYLOGUE_VERIFY_RUN_ID": "subprocess-regression", + "POLYLOGUE_PYTEST_RUN_ID": "subprocess-regression", + } + ) + result = subprocess.run( + [ + sys.executable, + "-m", + "pytest", + "-q", + "-p", + "devtools.pytest_progress_plugin", + "--testmon-noselect", + "tests/unit/core/test_identity_law.py::test_session_id_is_origin_native_id", + ], + cwd=Path(__file__).resolve().parents[3], + env=env, + capture_output=True, + text=True, + check=False, + ) + assert result.returncode == 0, result.stdout + result.stderr + events = [json.loads(line) for path in events_dir.glob("*.jsonl") for line in path.read_text().splitlines()] + reports = [event for event in events if event.get("event") == "test_report"] + assert {event["when"] for event in reports} == {"setup", "call", "teardown"} + + def test_progress_plugin_records_node_start_and_finish( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, From e97c329a5e50c9fc811c1062806ab5c6f840822c Mon Sep 17 00:00:00 2001 From: Sinity Date: Wed, 12 Aug 2026 04:02:01 +0200 Subject: [PATCH 06/14] fix(testmon): require current shard execution evidence --- devtools/verify.py | 7 ++- tests/conftest.py | 2 - .../devtools/test_pytest_progress_plugin.py | 1 - tests/unit/devtools/test_verify.py | 61 ++++++++++++++++--- 4 files changed, 56 insertions(+), 15 deletions(-) diff --git a/devtools/verify.py b/devtools/verify.py index 0db649407d..e08cc9fe11 100644 --- a/devtools/verify.py +++ b/devtools/verify.py @@ -2570,7 +2570,7 @@ def _prepare_testmon_seed_attempt( resume: bool, ) -> dict[str, Any]: prior = _read_testmon_seed_attempt() if resume else None - expected = _testmon_seed_expected_nodeids(prior) if prior is not None else [] + expected = sorted(_testmon_seed_expected_nodeids(prior)) if prior is not None else [] prior_outcomes = _flatten_seed_outcomes(prior) payload = { "protocol_version": TESTMON_SEED_PROTOCOL_VERSION, @@ -2620,7 +2620,7 @@ def _prepare_testmon_seed_shards( selection: Mapping[str, Any] | None, ) -> dict[str, Any]: """Persist the full planned corpus before the first testmon DB mutation.""" - expected = _testmon_seed_expected_nodeids(prepared) if prepared.get("resume") else [] + expected = sorted(_testmon_seed_expected_nodeids(prepared)) if prepared.get("resume") else [] if not expected: expected = _seed_selection_nodeids(selection or {}) or [] prior_shards = validate_seed_shard_ledger(prepared.get("shards"), expected_nodeids=expected) @@ -2672,7 +2672,7 @@ def _checkpoint_testmon_seed_shard( step: Mapping[str, Any], ) -> dict[str, Any]: """Record one shard's result atomically before another shard may start.""" - expected = _testmon_seed_expected_nodeids(prepared) + expected = sorted(_testmon_seed_expected_nodeids(prepared)) shards = validate_seed_shard_ledger(prepared.get("shards"), expected_nodeids=expected) if shards is None or shard_index < 1 or shard_index > len(shards): raise ValueError("testmon seed shard ledger is malformed") @@ -2693,6 +2693,7 @@ def _checkpoint_testmon_seed_shard( database=database, pytest_step=step, prior_node_outcomes=prior, + use_database_fallback=False, ) terminal = all(item.get("outcome") in {"passed", "failed", "error", "skipped"} for item in outcomes) selection_matches = selected == nodeids diff --git a/tests/conftest.py b/tests/conftest.py index c224d56e9b..97a170327d 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -404,12 +404,10 @@ def _reclaim_passing_test_tmp_path( _MANAGED_VERIFY_ENV = frozenset( { "POLYLOGUE_VERIFY_RUN_ID", - "POLYLOGUE_PYTEST_RUN_ID", "POLYLOGUE_PYTEST_EVENTS_DIR", "POLYLOGUE_PYTEST_EVENTS_PATH", "POLYLOGUE_PYTEST_SELECTION_PATH", "POLYLOGUE_PYTEST_SUMMARY_PATH", - "POLYLOGUE_PYTEST_CONTAINMENT_PATH", "POLYLOGUE_PYTEST_SELECTION_NODEID_LIMIT", } ) diff --git a/tests/unit/devtools/test_pytest_progress_plugin.py b/tests/unit/devtools/test_pytest_progress_plugin.py index 7bf2974a05..e344db369b 100644 --- a/tests/unit/devtools/test_pytest_progress_plugin.py +++ b/tests/unit/devtools/test_pytest_progress_plugin.py @@ -80,7 +80,6 @@ def test_managed_event_ledger_survives_test_host_environment_scrub(tmp_path: Pat "POLYLOGUE_PYTEST_SELECTION_PATH": str(tmp_path / "selection.json"), "POLYLOGUE_PYTEST_SUMMARY_PATH": str(tmp_path / "summary.json"), "POLYLOGUE_VERIFY_RUN_ID": "subprocess-regression", - "POLYLOGUE_PYTEST_RUN_ID": "subprocess-regression", } ) result = subprocess.run( diff --git a/tests/unit/devtools/test_verify.py b/tests/unit/devtools/test_verify.py index 2ece677b18..53c366697d 100644 --- a/tests/unit/devtools/test_verify.py +++ b/tests/unit/devtools/test_verify.py @@ -307,7 +307,8 @@ def test_seed_shard_checkpoint_preserves_completed_shards_for_resume( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: monkeypatch.chdir(tmp_path) - expected = ["tests/test_seed.py::test_a", "tests/test_seed.py::test_b"] + expected = ["tests/test_seed.py::test_b", "tests/test_seed.py::test_a"] + ordered = sorted(expected) prepared = _prepare_testmon_seed_shards( {"resume": False, "expected_nodeids": []}, selection={"selected_count": 2, "selected_nodeids": expected, "selected_nodeids_omitted": 0}, @@ -315,33 +316,33 @@ def test_seed_shard_checkpoint_preserves_completed_shards_for_resume( prepared["shards"] = [ { **prepared["shards"][0], - "nodeids": [expected[0]], + "nodeids": [ordered[0]], "nodeid_count": 1, - "nodeid_digest": hashlib.sha256(expected[0].encode()).hexdigest(), + "nodeid_digest": hashlib.sha256(ordered[0].encode()).hexdigest(), }, { **prepared["shards"][0], "index": 2, - "nodeids": [expected[1]], + "nodeids": [ordered[1]], "nodeid_count": 1, - "nodeid_digest": hashlib.sha256(expected[1].encode()).hexdigest(), + "nodeid_digest": hashlib.sha256(ordered[1].encode()).hexdigest(), "status": "pending", "node_outcomes": [], }, ] _atomic_payload = { **prepared, - "expected_nodeids": expected, + "expected_nodeids": ordered, "expected_count": 2, - "expected_digest": hashlib.sha256("\n".join(expected).encode()).hexdigest(), + "expected_digest": hashlib.sha256("\n".join(sorted(expected)).encode()).hexdigest(), } artifact_dir = tmp_path / "shard-1" artifact_dir.mkdir() (artifact_dir / "selection.json").write_text( - json.dumps({"selected_count": 1, "selected_nodeids": [expected[0]], "selected_nodeids_omitted": 0}) + json.dumps({"selected_count": 1, "selected_nodeids": [ordered[0]], "selected_nodeids_omitted": 0}) ) (artifact_dir / "events.jsonl").write_text( - json.dumps({"event": "test_report", "nodeid": expected[0], "when": "call", "outcome": "passed"}) + "\n" + json.dumps({"event": "test_report", "nodeid": ordered[0], "when": "call", "outcome": "passed"}) + "\n" ) checkpointed = _checkpoint_testmon_seed_shard( @@ -370,6 +371,48 @@ def test_seed_shard_checkpoint_preserves_completed_shards_for_resume( assert resumed["shards"][1]["status"] == "pending" +def test_seed_shard_checkpoint_does_not_trust_preexisting_testmon_rows( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.chdir(tmp_path) + expected = ["tests/test_seed.py::test_only_database_row"] + prepared = _prepare_testmon_seed_shards( + {"resume": False, "expected_nodeids": []}, + selection={"selected_count": 1, "selected_nodeids": expected, "selected_nodeids_omitted": 0}, + ) + artifact_dir = tmp_path / "shard-1" + artifact_dir.mkdir() + (artifact_dir / "selection.json").write_text( + json.dumps({"selected_count": 1, "selected_nodeids": expected, "selected_nodeids_omitted": 0}) + ) + (artifact_dir / "events.jsonl").write_text("") + monkeypatch.setattr( + "devtools.verify._testmon_database_state", + lambda _nodeids: { + "recorded_count": 1, + "failed_count": 0, + "dependency_edge_count": 0, + "missing_nodeids": [], + "failed_nodeids": [], + "node_outcomes": {expected[0]: "passed"}, + "error": None, + "graph_status": "complete", + "orphan_execution_edges": 0, + "orphan_fingerprint_edges": 0, + }, + ) + + checkpointed = _checkpoint_testmon_seed_shard( + prepared=prepared, + shard_index=1, + step={"name": "pytest seed-testmon shard 1/1", "exit": 0, "artifact_dir": str(artifact_dir)}, + ) + + shard = checkpointed["shards"][0] + assert shard["status"] == "incomplete" + assert shard["node_outcomes"][0]["outcome"] == "missing" + + def test_seed_shard_failure_remains_visible_and_blocks_release(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.chdir(tmp_path) expected = ["tests/test_seed.py::test_failed"] From 2b5211fb0799aa1ae37484ae79d83d918a2bfeba Mon Sep 17 00:00:00 2001 From: Sinity Date: Wed, 12 Aug 2026 04:35:14 +0200 Subject: [PATCH 07/14] fix(test): align convergence fixtures with archive authority Problem: the fresh convergence-property seed exposed fixture failures before the intended laws could run. Raw and attachment references were admitted without published bytes, append fixtures required an inapplicable FTS mutation, and readiness fixtures lacked parser-authority receipts. What changed: publish fixture raw and attachment blobs through the archive publisher, record complete parser census evidence for admitted raws, refresh the FTS freshness ledger after both stages, and preserve the raw-replay mutation's intended comparator failure when raw acquisition is bypassed. Compatibility: test infrastructure only; production convergence behavior is unchanged. --- tests/infra/convergence_harness.py | 93 +++++++++++++++++++++++++----- 1 file changed, 78 insertions(+), 15 deletions(-) diff --git a/tests/infra/convergence_harness.py b/tests/infra/convergence_harness.py index 33258ee1ca..28f85986e8 100644 --- a/tests/infra/convergence_harness.py +++ b/tests/infra/convergence_harness.py @@ -24,6 +24,7 @@ import polylogue.daemon.convergence_stages as convergence_stages from polylogue.archive.message.roles import Role +from polylogue.archive.revision_authority import RawRevisionEnvelope, RawRevisionKind from polylogue.core.enums import BlockType, Provider from polylogue.core.outcomes import OutcomeStatus from polylogue.daemon.convergence import DaemonConverger, SessionState @@ -44,7 +45,8 @@ ) from polylogue.storage.blob_publication import ArchiveBlobPublisher, consume_blob_publication_receipt from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_archive_tier -from polylogue.storage.sqlite.archive_tiers.source_write import write_source_raw_session +from polylogue.storage.sqlite.archive_tiers.revision_governance import record_current_parser_source_census +from polylogue.storage.sqlite.archive_tiers.source_write import ArchiveSourceBlobRef, write_source_raw_session from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier from polylogue.storage.sqlite.archive_tiers.write import write_parsed_session_to_archive from polylogue.storage.sqlite.connection import open_connection @@ -229,24 +231,76 @@ def ingest_convergence_pathology( session_ids: list[str] = [] for index in selected: session = _parsed_session(pathology.sessions[index], corpus_index=index) + content_hash = str(session_content_hash(session)) payload = _raw_payload(session) source_path = root / "sources" / f"{index:03d}-{session.provider_session_id}.json" source_path.parent.mkdir(parents=True, exist_ok=True) source_path.write_bytes(payload) - with sqlite3.connect(root / "source.db") as source_conn: - raw_id = write_source_raw_session( - source_conn, - origin="codex-session", - capture_mode=Provider.CODEX, - source_path=str(source_path), - source_index=-1 if append_only else index, - payload=payload, - acquired_at_ms=_acquired_at_ms(index), - native_id=session.provider_session_id, + raw_blob_publisher = ArchiveBlobPublisher(root / "source.db", root / "blob") + raw_blob_hash, raw_blob_size = raw_blob_publisher.write_from_bytes(payload) + preacquired_attachments: list[ParsedAttachment] = [] + attachment_blob_refs: list[ArchiveSourceBlobRef] = [] + attachment_receipts: list[tuple[str, bytes]] = [] + for attachment in session.attachments: + if attachment.inline_bytes is None: + preacquired_attachments.append(attachment) + continue + attachment_hash, attachment_size = raw_blob_publisher.write_from_bytes(attachment.inline_bytes) + attachment_receipt = raw_blob_publisher.receipt_id(attachment_hash) + preacquired_attachments.append( + attachment.model_copy( + update={"inline_bytes": None, "precomputed_blob": (attachment_hash, attachment_size)} + ) ) + attachment_blob_refs.append( + ArchiveSourceBlobRef( + blob_hash=bytes.fromhex(attachment_hash), + ref_type="attachment", + source_path=str(source_path), + size_bytes=attachment_size, + acquired_at_ms=_acquired_at_ms(index), + publication_receipt_id=attachment_receipt, + ) + ) + if attachment_receipt is not None: + attachment_receipts.append((attachment_receipt, bytes.fromhex(attachment_hash))) + session = session.model_copy(update={"attachments": preacquired_attachments}) + raw_blob_publisher.flush() + with sqlite3.connect(root / "source.db") as source_conn: + with source_conn: + raw_id = write_source_raw_session( + source_conn, + origin="codex-session", + capture_mode=Provider.CODEX, + source_path=str(source_path), + source_index=-1 if append_only else index, + payload=payload, + acquired_at_ms=_acquired_at_ms(index), + native_id=session.provider_session_id, + revision=RawRevisionEnvelope( + logical_source_key=str(make_session_id(session.source_name, session.provider_session_id)), + kind=RawRevisionKind.FULL, + source_revision=content_hash, + acquisition_generation=index, + ), + blob_publication_receipt_id=raw_blob_publisher.receipt_id(raw_blob_hash), + additional_blob_refs=tuple(attachment_blob_refs), + manage_transaction=False, + ) + consume_blob_publication_receipt( + source_conn, + raw_blob_publisher.receipt_id(raw_blob_hash), + bytes.fromhex(raw_blob_hash), + ) + for attachment_receipt, attachment_hash_bytes in attachment_receipts: + consume_blob_publication_receipt(source_conn, attachment_receipt, attachment_hash_bytes) + if source_conn.execute("SELECT 1 FROM raw_sessions WHERE raw_id = ?", (raw_id,)).fetchone() is not None: + record_current_parser_source_census(source_conn, raw_id, parser_sessions=[session]) + if raw_blob_size != len(payload): + raise AssertionError(f"published raw payload size drifted for {source_path}") payload_model = SessionWritePayload( session_id=str(make_session_id(session.source_name, session.provider_session_id)), - content_hash=str(session_content_hash(session)), + content_hash=content_hash, parsed_session=session, message_count=len(session.messages), attachment_count=len(session.attachments), @@ -273,7 +327,10 @@ def ingest_convergence_pathology( session_id = payload_model.session_id source_paths.append(source_path) session_ids.append(session_id) - make_messages_fts_stale(root / "index.db", session_id=session_id) + # Some valid provider fixtures contain no text-bearing blocks and + # therefore have no FTS rows to corrupt. The corpus builder may skip + # that inapplicable mutation; direct corruption tests remain strict. + make_messages_fts_stale(root / "index.db", session_id=session_id, require_rows=False) archive = ConvergenceArchive(root, pathology, tuple(source_paths), tuple(dict.fromkeys(session_ids))) if converge_after_each: converge_convergence_archive(archive) @@ -294,6 +351,12 @@ def converge_convergence_archive(archive: ConvergenceArchive) -> dict[str, Sessi not_converged = {session_id: state.last_error for session_id, state in states.items() if not state.converged} if not_converged: raise AssertionError(f"production convergence left pending work: {not_converged}") + # Insights can materialize work-event rows after the FTS stage has run. + # Refresh the shared freshness ledger only once both real stages complete. + from polylogue.daemon.fts_startup import record_fts_freshness_snapshot_sync + + with sqlite3.connect(archive.root / "index.db") as conn: + record_fts_freshness_snapshot_sync(conn) _analyze_registry_tables(archive.root / "index.db") return states @@ -658,7 +721,7 @@ def set_debt_retry_at( raise AssertionError(f"expected one convergence debt row, updated {cursor.rowcount}") -def make_messages_fts_stale(index_db: Path, *, session_id: str) -> int: +def make_messages_fts_stale(index_db: Path, *, session_id: str, require_rows: bool = True) -> int: """Delete only this session's real FTS rows to create unrelated stage debt.""" with open_connection(index_db) as conn: block_ids = tuple( @@ -679,7 +742,7 @@ def make_messages_fts_stale(index_db: Path, *, session_id: str) -> int: conn.executemany("DELETE FROM messages_fts WHERE rowid = ?", ((row_id,) for row_id in row_ids)) conn.executemany("DELETE FROM messages_fts_identity WHERE rowid = ?", ((row_id,) for row_id in row_ids)) conn.commit() - if not row_ids: + if require_rows and not row_ids: raise AssertionError(f"session {session_id!r} has no indexed blocks") return len(row_ids) From 0f60850904e9864d4f6d8fc66cc7487aed497772 Mon Sep 17 00:00:00 2001 From: Sinity Date: Wed, 12 Aug 2026 04:36:20 +0200 Subject: [PATCH 08/14] fix(test): align annotation fixtures with archive contracts Keep cold annotation reads on the default read-only archive handle and reopen a writable handle for retries. Derive delegation target and evidence identities from persisted action rows so the fixture follows the current message identity law. --- tests/unit/annotations/test_durable_storage.py | 2 ++ tests/unit/annotations/test_importer.py | 7 +++++-- 2 files changed, 7 insertions(+), 2 deletions(-) diff --git a/tests/unit/annotations/test_durable_storage.py b/tests/unit/annotations/test_durable_storage.py index 823d6b3aae..bb8ab242e9 100644 --- a/tests/unit/annotations/test_durable_storage.py +++ b/tests/unit/annotations/test_durable_storage.py @@ -524,6 +524,8 @@ def test_batch_opaque_refs_preserve_decomposed_bytes_across_retry_and_cold_read( assert cold.prompt_ref == f"block:{decomposed}:0" assert cold.assertion_refs == (f"assertion:{decomposed}",) assert cold.canonical_provenance_bytes() == original.canonical_provenance_bytes() + + with ArchiveStore.open_existing(archive_root, read_only=False) as reopened: replay = reopened.save_annotation_batch(exact_retry) assert replay.canonical_provenance_bytes() == original.canonical_provenance_bytes() with pytest.raises(AnnotationBatchError, match="incompatible provenance"): diff --git a/tests/unit/annotations/test_importer.py b/tests/unit/annotations/test_importer.py index 7f79d84ccc..bb63cd6781 100644 --- a/tests/unit/annotations/test_importer.py +++ b/tests/unit/annotations/test_importer.py @@ -238,10 +238,13 @@ async def test_import_uses_concrete_delegation_schema_and_exact_retry_is_idempot branch_type=BranchType.SUBAGENT, ) ) - instruction_block_id = f"{parent_session_id}:dispatch:0" + with ArchiveStore.open_existing(archive_root) as archive: + actions = archive.query_session_actions([parent_session_id], limit=10) + instruction_block_id = next(action.tool_use_block_id for action in actions if action.semantic_type == "subagent") + instruction_message_id, instruction_position = instruction_block_id.rsplit(":", 1) target_ref = f"delegation:{instruction_block_id}" evidence_ref = f"block:{instruction_block_id}" - evidence_span = f"{parent_session_id}::{parent_session_id}:dispatch::0" + evidence_span = f"{parent_session_id}::{instruction_message_id}::{instruction_position}" valid_rows = [ { "row_key": f"delegation-{index}", From 602474e03444a989287e4e5be57ab6e268ca4f26 Mon Sep 17 00:00:00 2001 From: Sinity Date: Wed, 12 Aug 2026 10:24:44 +0200 Subject: [PATCH 09/14] fix(testmon): restore parallel seed execution Problem: serial explicit-node shards replaced pytest-testmon's controller-owned xdist write path, stretching the seed into dozens of process startups. Restore one managed xdist pytest-testmon invocation, retain event-backed per-node outcomes and graph validation, and remove the shard-only receipt protocol. --- devtools/testmon_state.py | 117 ------------ devtools/verify.py | 282 +++-------------------------- tests/unit/devtools/test_verify.py | 223 ++--------------------- 3 files changed, 40 insertions(+), 582 deletions(-) diff --git a/devtools/testmon_state.py b/devtools/testmon_state.py index 1929ff801b..ec78f85dca 100644 --- a/devtools/testmon_state.py +++ b/devtools/testmon_state.py @@ -53,15 +53,6 @@ class SeedAttemptOutcome(StrEnum): RESOURCE_TIMEOUT = "resource-timeout" -class SeedShardStatus(StrEnum): - """Durable state of one sequential pytest-testmon seed shard.""" - - PENDING = "pending" - RUNNING = "running" - COMPLETE = "complete" - INCOMPLETE = "incomplete" - - class BindingMode(StrEnum): EXACT = "exact" RELATIVE_FILE_FINGERPRINTS = "relative-file-fingerprints" @@ -78,110 +69,6 @@ class TerminalAuthorization(StrEnum): NARROW_TERMINAL = "narrow-terminal" -_TERMINAL_NODE_OUTCOMES = frozenset({"passed", "failed", "error", "skipped"}) - - -def seed_shard_plan(nodeids: Sequence[str], *, shard_size: int) -> list[dict[str, Any]]: - """Partition a complete node set into stable, contiguous, serial shards.""" - if shard_size <= 0: - raise ValueError("testmon seed shard_size must be positive") - if not nodeids or any(not nodeid for nodeid in nodeids): - raise ValueError("testmon seed nodeids must be non-empty strings") - if len(set(nodeids)) != len(nodeids): - raise ValueError("testmon seed nodeids must be unique") - ordered = tuple(sorted(nodeids)) - return [ - { - "index": index, - "nodeids": list(ordered[offset : offset + shard_size]), - "nodeid_count": len(ordered[offset : offset + shard_size]), - "nodeid_digest": hashlib.sha256("\n".join(ordered[offset : offset + shard_size]).encode()).hexdigest(), - "status": SeedShardStatus.PENDING.value, - "node_outcomes": [], - } - for index, offset in enumerate(range(0, len(ordered), shard_size), start=1) - ] - - -def validate_seed_shard_ledger( - shards: object, - *, - expected_nodeids: Sequence[str], -) -> list[dict[str, Any]] | None: - """Validate the full shard ledger without granting release authority. - - Every shard owns a disjoint contiguous part of the sorted expected node - set. A completed shard carries an explicit terminal result for every node; - interrupted shards remain visible and are eligible for resume. - """ - if not isinstance(shards, list) or not shards: - return None - expected = tuple(sorted(expected_nodeids)) - if not expected or len(set(expected)) != len(expected): - return None - normalized: list[dict[str, Any]] = [] - observed: list[str] = [] - for index, raw in enumerate(shards, start=1): - if not isinstance(raw, Mapping) or raw.get("index") != index: - return None - nodeids = raw.get("nodeids") - if ( - not isinstance(nodeids, list) - or not nodeids - or any(not isinstance(nodeid, str) or not nodeid for nodeid in nodeids) - or nodeids != sorted(nodeids) - ): - return None - if raw.get("nodeid_count") != len(nodeids): - return None - if raw.get("nodeid_digest") != hashlib.sha256("\n".join(nodeids).encode()).hexdigest(): - return None - raw_status = raw.get("status") - if not isinstance(raw_status, str): - return None - try: - status = SeedShardStatus(raw_status) - except (TypeError, ValueError): - return None - outcomes = raw.get("node_outcomes") - if not isinstance(outcomes, list): - return None - outcome_by_node: dict[str, dict[str, Any]] = {} - for outcome in outcomes: - if not isinstance(outcome, Mapping): - return None - nodeid = outcome.get("nodeid") - state = outcome.get("outcome") - if not isinstance(nodeid, str) or nodeid not in nodeids or not isinstance(state, str): - return None - if nodeid in outcome_by_node: - return None - outcome_by_node[nodeid] = dict(outcome) - if status is SeedShardStatus.PENDING and outcomes: - return None - if status is SeedShardStatus.COMPLETE and ( - set(outcome_by_node) != set(nodeids) - or any(item.get("outcome") not in _TERMINAL_NODE_OUTCOMES for item in outcome_by_node.values()) - ): - return None - if ( - status in {SeedShardStatus.RUNNING, SeedShardStatus.INCOMPLETE} - and outcomes - and set(outcome_by_node) != set(nodeids) - ): - return None - normalized.append(dict(raw)) - observed.extend(nodeids) - if tuple(observed) != expected: - return None - return normalized - - -def seed_shard_ledger_is_terminal(shards: Sequence[Mapping[str, Any]]) -> bool: - """Return whether every planned shard completed with explicit node results.""" - return all(shard.get("status") == SeedShardStatus.COMPLETE.value for shard in shards) - - @dataclass(frozen=True, slots=True) class TestmonIdentity: git_head: str | None @@ -983,10 +870,6 @@ def stamp_from_attempt( expected_count = attempt.get("expected_count") if not isinstance(expected_count, int) or isinstance(expected_count, bool) or expected_count != len(expected): return None - if protocol_version >= 7: - shards = validate_seed_shard_ledger(attempt.get("shards"), expected_nodeids=expected) - if shards is None or not seed_shard_ledger_is_terminal(shards): - return None expected_digest = attempt.get("expected_digest") if ( not isinstance(expected_digest, str) diff --git a/devtools/verify.py b/devtools/verify.py index e08cc9fe11..351bb12e7f 100644 --- a/devtools/verify.py +++ b/devtools/verify.py @@ -62,18 +62,14 @@ BindingMode, GraphStatus, SeedAttemptOutcome, - SeedShardStatus, TerminalAuthorization, TestmonBinding, TestmonSeedStamp, VerificationScope, inspect_testmon_database, refresh_stamp, - seed_shard_ledger_is_terminal, - seed_shard_plan, stamp_from_attempt, testmon_runtime_identity, - validate_seed_shard_ledger, validate_stamp, ) from devtools.verify_runs import ( @@ -220,8 +216,7 @@ def _format_completion_notification( TESTMON_SEED_STAMP = Path(".cache/testmon/seed.json") TESTMON_SEED_ATTEMPT = Path(".cache/testmon/seed-attempt.json") TESTMON_AFFECTED_STAMP = Path(".cache/testmon/affected.json") -TESTMON_SEED_PROTOCOL_VERSION = 7 -TESTMON_SEED_SHARD_SIZE = 256 +TESTMON_SEED_PROTOCOL_VERSION = 6 PYTEST_REPORT_DIR = Path(".cache/verify") PYTEST_REPORT_PATH = PYTEST_REPORT_DIR / "last-pytest.json" PYTEST_JUNIT_REPORT_DIR = Path(".cache/test-reports") @@ -1999,11 +1994,14 @@ def build_verify_steps( if skip_slow: base_marker = f"not slow and {base_marker}" if seed_testmon: - # Collection produces the exact corpus contract before any testmon - # write. Shards below are generated from this ledger and run one - # at a time, so pytest-testmon has exactly one SQLite writer. - pytest_cmd.extend(["-m", base_marker, "--collect-only", "-n", "0"]) - label = "pytest seed-testmon collect (resume)" if resume_testmon_seed else "pytest seed-testmon collect" + pytest_cmd.extend(["-m", base_marker, "--testmon"]) + if resume_testmon_seed: + pytest_cmd.append("--testmon-forceselect") + label = "pytest seed-testmon (resume)" + else: + pytest_cmd.append("--testmon-noselect") + label = "pytest seed-testmon" + pytest_cmd.extend(_pytest_worker_args(maximum=4)) steps.append((label, pytest_cmd)) elif full_pytest: # #1775: the full diagnostic runs as two lanes. The bulk lane keeps @@ -2570,7 +2568,7 @@ def _prepare_testmon_seed_attempt( resume: bool, ) -> dict[str, Any]: prior = _read_testmon_seed_attempt() if resume else None - expected = sorted(_testmon_seed_expected_nodeids(prior)) if prior is not None else [] + expected = _testmon_seed_expected_nodeids(prior) if prior is not None else [] prior_outcomes = _flatten_seed_outcomes(prior) payload = { "protocol_version": TESTMON_SEED_PROTOCOL_VERSION, @@ -2581,7 +2579,6 @@ def _prepare_testmon_seed_attempt( "expected_count": len(expected), "expected_digest": hashlib.sha256("\n".join(sorted(expected)).encode()).hexdigest() if expected else None, "prior_node_outcomes": prior_outcomes, - "shards": list(prior.get("shards", [])) if prior is not None and isinstance(prior.get("shards"), list) else [], "started_at": datetime.now(timezone.utc).isoformat(), "run_id": run.run_id, "artifact_dir": str(run.relative_run_dir), @@ -2593,137 +2590,6 @@ def _prepare_testmon_seed_attempt( return payload -def _seed_selection_nodeids(selection: Mapping[str, Any]) -> list[str] | None: - """Accept only a complete, untruncated collection ledger.""" - nodeids = selection.get("selected_nodeids") - selected_count = selection.get("selected_count") - omitted = selection.get("selected_nodeids_omitted") - if ( - not isinstance(nodeids, list) - or not nodeids - or any(not isinstance(nodeid, str) or not nodeid for nodeid in nodeids) - or len(set(nodeids)) != len(nodeids) - or not isinstance(selected_count, int) - or isinstance(selected_count, bool) - or selected_count != len(nodeids) - or not isinstance(omitted, int) - or isinstance(omitted, bool) - or omitted != 0 - ): - return None - return sorted(nodeids) - - -def _prepare_testmon_seed_shards( - prepared: Mapping[str, Any], - *, - selection: Mapping[str, Any] | None, -) -> dict[str, Any]: - """Persist the full planned corpus before the first testmon DB mutation.""" - expected = sorted(_testmon_seed_expected_nodeids(prepared)) if prepared.get("resume") else [] - if not expected: - expected = _seed_selection_nodeids(selection or {}) or [] - prior_shards = validate_seed_shard_ledger(prepared.get("shards"), expected_nodeids=expected) - shards = ( - prior_shards - if prior_shards is not None - else (seed_shard_plan(expected, shard_size=TESTMON_SEED_SHARD_SIZE) if expected else []) - ) - payload = { - **dict(prepared), - "expected_nodeids": expected, - "expected_count": len(expected), - "expected_digest": hashlib.sha256("\n".join(expected).encode()).hexdigest() if expected else None, - "selection": dict(selection or {}), - "shard_size": TESTMON_SEED_SHARD_SIZE, - "shards": shards, - } - _atomic_write_json(TESTMON_SEED_ATTEMPT, payload) - return payload - - -def _seed_shard_command(collection_command: Sequence[str], shard: Mapping[str, Any]) -> list[str]: - """Build a serial, explicit-node pytest-testmon invocation for one shard.""" - nodeids = shard.get("nodeids") - if not isinstance(nodeids, list) or not nodeids: - raise ValueError("testmon seed shard is missing nodeids") - command = [argument for argument in collection_command if argument != "--collect-only"] - command.extend(["--testmon", "--testmon-noselect", *nodeids]) - return command - - -def _seed_shard_outcomes(shards: Sequence[Mapping[str, Any]]) -> list[dict[str, Any]]: - """Flatten the shard ledger in canonical node order for legacy readers.""" - outcomes: dict[str, dict[str, Any]] = {} - for shard in shards: - raw_outcomes = shard.get("node_outcomes") - if not isinstance(raw_outcomes, list): - continue - for item in raw_outcomes: - if isinstance(item, Mapping) and isinstance(item.get("nodeid"), str): - outcomes[str(item["nodeid"])] = dict(item) - return [outcomes[nodeid] for nodeid in sorted(outcomes)] - - -def _checkpoint_testmon_seed_shard( - *, - prepared: Mapping[str, Any], - shard_index: int, - step: Mapping[str, Any], -) -> dict[str, Any]: - """Record one shard's result atomically before another shard may start.""" - expected = sorted(_testmon_seed_expected_nodeids(prepared)) - shards = validate_seed_shard_ledger(prepared.get("shards"), expected_nodeids=expected) - if shards is None or shard_index < 1 or shard_index > len(shards): - raise ValueError("testmon seed shard ledger is malformed") - shard = dict(shards[shard_index - 1]) - nodeids = shard["nodeids"] - artifact_dir = _safe_testmon_artifact_dir(step.get("artifact_dir")) - selection = _read_json_artifact(artifact_dir / "selection.json") if artifact_dir is not None else None - selected = _seed_selection_nodeids(selection) if isinstance(selection, Mapping) else None - database = _testmon_database_state(nodeids) - prior = { - str(item["nodeid"]): item - for item in shard.get("node_outcomes", []) - if isinstance(item, Mapping) and isinstance(item.get("nodeid"), str) - } - outcomes = _seed_node_outcomes_from_events( - artifact_dir / "events.jsonl" if artifact_dir is not None else Path(".missing-testmon-events"), - expected_nodeids=nodeids, - database=database, - pytest_step=step, - prior_node_outcomes=prior, - use_database_fallback=False, - ) - terminal = all(item.get("outcome") in {"passed", "failed", "error", "skipped"} for item in outcomes) - selection_matches = selected == nodeids - shard.update( - { - "status": SeedShardStatus.COMPLETE.value - if selection_matches and terminal - else SeedShardStatus.INCOMPLETE.value, - "started_at": shard.get("started_at") or datetime.now(timezone.utc).isoformat(), - "finished_at": datetime.now(timezone.utc).isoformat(), - "exit_code": step.get("exit"), - "artifact_dir": step.get("artifact_dir"), - "selection": dict(selection) if isinstance(selection, Mapping) else None, - "database": database, - "node_outcomes": outcomes, - "pytest_step": dict(step), - } - ) - shards[shard_index - 1] = shard - payload = { - **dict(prepared), - "status": "running", - "shards": shards, - "node_outcomes": _seed_shard_outcomes(shards), - "testmon_data": _file_fingerprint(TESTMON_DATA), - } - _atomic_write_json(TESTMON_SEED_ATTEMPT, payload) - return payload - - def _testmon_seed_terminal_authorized(prepared: Mapping[str, Any]) -> bool: identity = prepared.get("identity") return ( @@ -2907,59 +2773,22 @@ def _finalize_testmon_seed_attempt( and len(set(selected_nodeids)) == len(selected_nodeids) and raw_selected_count == len(selected_nodeids) ) - prepared_expected = prepared.get("expected_nodeids") - expected_raw = prepared_expected if isinstance(prepared_expected, list) and prepared_expected else selected_nodeids + expected_raw = prepared.get("expected_nodeids") if prepared.get("resume") else selected_nodeids expected = list(expected_raw) if isinstance(expected_raw, list) else [] - shards = validate_seed_shard_ledger(prepared.get("shards"), expected_nodeids=expected) - sharded = shards is not None - if sharded: - assert shards is not None - selection_valid = seed_shard_ledger_is_terminal(shards) - omitted = 0 - database = _testmon_database_state(expected) - outcome_by_node = {item["nodeid"]: item for item in _seed_shard_outcomes(shards)} - node_outcomes = [ - outcome_by_node.get(nodeid, {"nodeid": nodeid, "outcome": "missing", "reason": "shard not completed"}) - for nodeid in expected - ] - shard_steps: list[Mapping[str, Any]] = [] - for shard in shards: - raw_step = shard.get("pytest_step") - if isinstance(raw_step, Mapping): - shard_steps.append(raw_step) - if shard_steps: - pytest_step = dict(shard_steps[-1]) - timed_out = next( - ( - step - for step in shard_steps - if str(step.get("diagnosis")) in {"pytest_timeout", "pytest_stall_timeout"} - ), - None, - ) - if timed_out is not None: - pytest_step = dict(timed_out) - selection = { - "selected_count": len(expected), - "selected_nodeids_omitted": 0, - "shard_count": len(shards), - "completed_shard_count": sum(shard.get("status") == SeedShardStatus.COMPLETE.value for shard in shards), - } - else: - omitted = raw_omitted if selection_valid and isinstance(raw_omitted, int) else 1 - database = _testmon_database_state(expected) - node_outcomes = _seed_node_outcomes_from_events( - events_path or Path(".missing-testmon-events"), - expected_nodeids=expected, - database=database, - pytest_step=pytest_step, - use_database_fallback=False, - prior_node_outcomes={ - str(item["nodeid"]): item - for item in prepared.get("prior_node_outcomes", []) - if isinstance(item, Mapping) and isinstance(item.get("nodeid"), str) - }, - ) + omitted = raw_omitted if selection_valid else 1 + database = _testmon_database_state(expected) + node_outcomes = _seed_node_outcomes_from_events( + events_path or Path(".missing-testmon-events"), + expected_nodeids=expected, + database=database, + pytest_step=pytest_step, + use_database_fallback=False, + prior_node_outcomes={ + str(item["nodeid"]): item + for item in prepared.get("prior_node_outcomes", []) + if isinstance(item, Mapping) and isinstance(item.get("nodeid"), str) + }, + ) unsuccessful_nodeids = [ str(item["nodeid"]) for item in node_outcomes if item.get("outcome") not in {"passed", "skipped"} ] @@ -2994,20 +2823,6 @@ def _finalize_testmon_seed_attempt( exit_code=exit_code, pytest_step=pytest_step, ) - shard_ledger = ( - shards if sharded else seed_shard_plan(expected, shard_size=max(1, len(expected))) if expected else [] - ) - if not sharded and shard_ledger: - shard_ledger[0].update( - { - "status": ( - SeedShardStatus.COMPLETE.value - if all(item.get("outcome") in {"passed", "failed", "error", "skipped"} for item in node_outcomes) - else SeedShardStatus.INCOMPLETE.value - ), - "node_outcomes": node_outcomes, - } - ) seed_scope = ( VerificationScope.NARROW_TERMINAL.value if narrow_terminal else VerificationScope.RELEASE_BASELINE.value ) @@ -3029,7 +2844,6 @@ def _finalize_testmon_seed_attempt( else selection.get("selected_count"), "selected_nodeids_omitted": 0 if prepared.get("resume") and selection_valid else omitted, }, - "shards": shard_ledger, "node_outcomes": node_outcomes, "identity": prepared.get("identity"), "run_id": prepared.get("run_id"), @@ -3078,7 +2892,6 @@ def _finalize_testmon_seed_attempt( "collection_duration_s", ) }, - "shards": shard_ledger, "database": database, "node_outcomes": node_outcomes, "node_outcome_counts": dict( @@ -3381,51 +3194,6 @@ def main(argv: list[str] | None = None) -> int: step_result: dict[str, Any] = {"name": label, "duration_s": round(elapsed, 2), "exit": rc} step_result.update(metadata) step_results.append(step_result) - if args.seed_testmon and label.startswith("pytest seed-testmon collect"): - if rc != 0: - exit_code = rc - break - artifact_dir = _safe_testmon_artifact_dir(metadata.get("artifact_dir")) - selection = _read_json_artifact(artifact_dir / "selection.json") if artifact_dir is not None else None - assert prepared_seed_attempt is not None - prepared_seed_attempt = _prepare_testmon_seed_shards( - prepared_seed_attempt, - selection=selection if isinstance(selection, Mapping) else None, - ) - expected = _testmon_seed_expected_nodeids(prepared_seed_attempt) - shards = validate_seed_shard_ledger(prepared_seed_attempt.get("shards"), expected_nodeids=expected) - if shards is None: - exit_code = 5 - step_result["exit"] = 5 - step_result["diagnosis"] = "testmon_seed_collection_incomplete" - sys.stderr.write("verify: pytest-testmon collection did not produce a complete shard plan.\n") - break - for shard in shards: - if shard.get("status") == SeedShardStatus.COMPLETE.value: - continue - shard_index = int(shard["index"]) - shard_label = f"pytest seed-testmon shard {shard_index}/{len(shards)}" - shard_cmd = _seed_shard_command(cmd, shard) - _warn_low_memory() - shard_rc, shard_elapsed, shard_metadata = _run(shard_label, shard_cmd, run=verify_run) - shard_result: dict[str, Any] = { - "name": shard_label, - "duration_s": round(shard_elapsed, 2), - "exit": shard_rc, - "shard_index": shard_index, - "shard_count": len(shards), - "shard_nodeid_count": len(shard["nodeids"]), - } - shard_result.update(shard_metadata) - step_results.append(shard_result) - prepared_seed_attempt = _checkpoint_testmon_seed_shard( - prepared=prepared_seed_attempt, - shard_index=shard_index, - step=shard_result, - ) - if shard_rc != 0 and exit_code == 0: - exit_code = shard_rc - continue if label in {"pytest testmon", "pytest testmon (broad)"} and not args.seed_testmon and not full_pytest: _refresh_testmon_selection_attempt(step=step_result, run=verify_run, exit_code=rc) if rc != 0: diff --git a/tests/unit/devtools/test_verify.py b/tests/unit/devtools/test_verify.py index 53c366697d..61f08e5965 100644 --- a/tests/unit/devtools/test_verify.py +++ b/tests/unit/devtools/test_verify.py @@ -44,17 +44,14 @@ TESTMON_DATA, TESTMON_SEED_ATTEMPT, TESTMON_SEED_PROTOCOL_VERSION, - TESTMON_SEED_SHARD_SIZE, TESTMON_SEED_STAMP, _anchor_verification_paths, - _checkpoint_testmon_seed_shard, _finalize_testmon_seed_attempt, _flatten_seed_outcomes, _format_completion_notification, _matching_testmon_coverage, _parse_pytest_test_count, _prepare_testmon_seed_attempt, - _prepare_testmon_seed_shards, _pytest_command_metadata, _pytest_metadata_from_report, _pytest_stall_timeout_s, @@ -63,7 +60,6 @@ _record_testmon_affected_coverage, _run, _seed_node_outcomes_from_events, - _seed_shard_command, _stop_after_failed_step, _testmon_database_state, _testmon_preflight, @@ -258,12 +254,13 @@ def test_seed_testmon_runs_full_collection_without_selection(monkeypatch: pytest steps = build_verify_steps(quick=False, lab=False, skip_slow=False, seed_testmon=True) label, command = steps[-1] - assert label == "pytest seed-testmon collect" - assert "--collect-only" in command + assert label == "pytest seed-testmon" assert command[command.index("--ignore=tests/benchmarks")] == "--ignore=tests/benchmarks" - assert "--testmon" not in command + assert "--collect-only" not in command + assert "--testmon" in command + assert "--testmon-noselect" in command assert "-n" in command - assert command[command.index("-n") + 1] == "0" + assert command[command.index("-n") + 1] == "4" def test_seed_testmon_caps_adaptive_workers(monkeypatch: pytest.MonkeyPatch) -> None: @@ -273,200 +270,8 @@ def test_seed_testmon_caps_adaptive_workers(monkeypatch: pytest.MonkeyPatch) -> steps = build_verify_steps(quick=False, lab=False, skip_slow=False, seed_testmon=True) label, command = steps[-1] - assert label == "pytest seed-testmon collect" - assert command[command.index("-n") + 1] == "0" - - -def test_seed_shards_are_deterministic_and_use_one_testmon_writer( - tmp_path: Path, monkeypatch: pytest.MonkeyPatch -) -> None: - monkeypatch.chdir(tmp_path) - expected = [f"tests/test_seed.py::test_{index:03d}" for index in range(TESTMON_SEED_SHARD_SIZE + 2)] - prepared = _prepare_testmon_seed_shards( - {"resume": False, "expected_nodeids": []}, - selection={ - "selected_count": len(expected), - "selected_nodeids": list(reversed(expected)), - "selected_nodeids_omitted": 0, - }, - ) - - shards = prepared["shards"] - assert [shard["nodeid_count"] for shard in shards] == [TESTMON_SEED_SHARD_SIZE, 2] - assert shards[0]["nodeids"] == expected[:TESTMON_SEED_SHARD_SIZE] - assert shards[1]["nodeids"] == expected[TESTMON_SEED_SHARD_SIZE:] - command = _seed_shard_command(["pytest", "--collect-only", "-n", "0"], shards[0]) - assert "--collect-only" not in command - assert command[command.index("-n") + 1] == "0" - assert "--testmon" in command - assert "--testmon-noselect" in command - assert command[-TESTMON_SEED_SHARD_SIZE:] == expected[:TESTMON_SEED_SHARD_SIZE] - - -def test_seed_shard_checkpoint_preserves_completed_shards_for_resume( - tmp_path: Path, monkeypatch: pytest.MonkeyPatch -) -> None: - monkeypatch.chdir(tmp_path) - expected = ["tests/test_seed.py::test_b", "tests/test_seed.py::test_a"] - ordered = sorted(expected) - prepared = _prepare_testmon_seed_shards( - {"resume": False, "expected_nodeids": []}, - selection={"selected_count": 2, "selected_nodeids": expected, "selected_nodeids_omitted": 0}, - ) - prepared["shards"] = [ - { - **prepared["shards"][0], - "nodeids": [ordered[0]], - "nodeid_count": 1, - "nodeid_digest": hashlib.sha256(ordered[0].encode()).hexdigest(), - }, - { - **prepared["shards"][0], - "index": 2, - "nodeids": [ordered[1]], - "nodeid_count": 1, - "nodeid_digest": hashlib.sha256(ordered[1].encode()).hexdigest(), - "status": "pending", - "node_outcomes": [], - }, - ] - _atomic_payload = { - **prepared, - "expected_nodeids": ordered, - "expected_count": 2, - "expected_digest": hashlib.sha256("\n".join(sorted(expected)).encode()).hexdigest(), - } - artifact_dir = tmp_path / "shard-1" - artifact_dir.mkdir() - (artifact_dir / "selection.json").write_text( - json.dumps({"selected_count": 1, "selected_nodeids": [ordered[0]], "selected_nodeids_omitted": 0}) - ) - (artifact_dir / "events.jsonl").write_text( - json.dumps({"event": "test_report", "nodeid": ordered[0], "when": "call", "outcome": "passed"}) + "\n" - ) - - checkpointed = _checkpoint_testmon_seed_shard( - prepared=_atomic_payload, - shard_index=1, - step={"name": "pytest seed-testmon shard 1/2", "exit": 0, "artifact_dir": str(artifact_dir)}, - ) - - assert checkpointed["shards"][0]["status"] == "complete" - assert checkpointed["shards"][1]["status"] == "pending" - assert json.loads(TESTMON_SEED_ATTEMPT.read_text())["shards"][0]["node_outcomes"][0]["outcome"] == "passed" - resumed = _prepare_testmon_seed_attempt( - identity={ - "git_head": "head", - "git_tree": "tree", - "worktree_fingerprint": "fingerprint", - "python": "python", - "skip_slow": False, - "lab": False, - **_testmon_runtime_identity_fields(Path.cwd()), - }, - run=VerifyRun(tier="seed-testmon", argv=[], git_head="head", polylogue_import_path="polylogue"), - resume=True, - ) - assert resumed["shards"][0]["status"] == "complete" - assert resumed["shards"][1]["status"] == "pending" - - -def test_seed_shard_checkpoint_does_not_trust_preexisting_testmon_rows( - tmp_path: Path, monkeypatch: pytest.MonkeyPatch -) -> None: - monkeypatch.chdir(tmp_path) - expected = ["tests/test_seed.py::test_only_database_row"] - prepared = _prepare_testmon_seed_shards( - {"resume": False, "expected_nodeids": []}, - selection={"selected_count": 1, "selected_nodeids": expected, "selected_nodeids_omitted": 0}, - ) - artifact_dir = tmp_path / "shard-1" - artifact_dir.mkdir() - (artifact_dir / "selection.json").write_text( - json.dumps({"selected_count": 1, "selected_nodeids": expected, "selected_nodeids_omitted": 0}) - ) - (artifact_dir / "events.jsonl").write_text("") - monkeypatch.setattr( - "devtools.verify._testmon_database_state", - lambda _nodeids: { - "recorded_count": 1, - "failed_count": 0, - "dependency_edge_count": 0, - "missing_nodeids": [], - "failed_nodeids": [], - "node_outcomes": {expected[0]: "passed"}, - "error": None, - "graph_status": "complete", - "orphan_execution_edges": 0, - "orphan_fingerprint_edges": 0, - }, - ) - - checkpointed = _checkpoint_testmon_seed_shard( - prepared=prepared, - shard_index=1, - step={"name": "pytest seed-testmon shard 1/1", "exit": 0, "artifact_dir": str(artifact_dir)}, - ) - - shard = checkpointed["shards"][0] - assert shard["status"] == "incomplete" - assert shard["node_outcomes"][0]["outcome"] == "missing" - - -def test_seed_shard_failure_remains_visible_and_blocks_release(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: - monkeypatch.chdir(tmp_path) - expected = ["tests/test_seed.py::test_failed"] - prepared = _prepare_testmon_seed_shards( - { - "protocol_version": TESTMON_SEED_PROTOCOL_VERSION, - "status": "running", - "identity": { - "git_head": "head", - "worktree_fingerprint": "tree", - "python": "python", - "skip_slow": False, - "lab": False, - **_testmon_runtime_identity_fields(Path.cwd()), - }, - "resume": False, - "run_id": "sharded-failure", - "artifact_dir": ".cache/verify/runs/sharded-failure", - }, - selection={"selected_count": 1, "selected_nodeids": expected, "selected_nodeids_omitted": 0}, - ) - artifact_dir = tmp_path / "shard-failure" - artifact_dir.mkdir() - (artifact_dir / "selection.json").write_text( - json.dumps({"selected_count": 1, "selected_nodeids": expected, "selected_nodeids_omitted": 0}) - ) - (artifact_dir / "events.jsonl").write_text( - json.dumps({"event": "test_report", "nodeid": expected[0], "when": "call", "outcome": "failed"}) + "\n" - ) - TESTMON_DATA.parent.mkdir(parents=True, exist_ok=True) - with sqlite3.connect(TESTMON_DATA) as connection: - connection.execute("create table environment (id integer primary key, environment_name text)") - connection.execute("create table file_fp (id integer primary key, filename text, fsha text)") - connection.execute("create table test_execution (id integer primary key, test_name text, failed integer)") - connection.execute("create table test_execution_file_fp (test_execution_id integer, fingerprint_id integer)") - connection.execute("insert into test_execution values (1, ?, 1)", (expected[0],)) - connection.execute("insert into file_fp values (1, 'test_seed.py', 'sha')") - connection.execute("insert into test_execution_file_fp values (1, 1)") - _write_run_receipt(tmp_path, "sharded-failure") - - checkpointed = _checkpoint_testmon_seed_shard( - prepared=prepared, - shard_index=1, - step={"name": "pytest seed-testmon shard 1/1", "exit": 1, "artifact_dir": str(artifact_dir)}, - ) - receipt = _finalize_testmon_seed_attempt( - prepared=checkpointed, - step_results=[{"name": "pytest seed-testmon shard 1/1", "exit": 1, "artifact_dir": str(artifact_dir)}], - exit_code=1, - ) - - assert receipt["shards"][0]["status"] == "complete" - assert receipt["unsuccessful_nodeids"] == expected - assert receipt["release_baseline_allowed"] is False + assert label == "pytest seed-testmon" + assert command[command.index("-n") + 1] == "4" def test_resumed_seed_uses_affected_selection_for_remaining_tests() -> None: @@ -479,9 +284,11 @@ def test_resumed_seed_uses_affected_selection_for_remaining_tests() -> None: ) label, command = steps[-1] - assert label == "pytest seed-testmon collect (resume)" - assert "--collect-only" in command - assert "--testmon" not in command + assert label == "pytest seed-testmon (resume)" + assert "--collect-only" not in command + assert "--testmon" in command + assert "--testmon-forceselect" in command + assert "--testmon-noselect" not in command def test_full_verify_includes_full_pytest_without_testmon(monkeypatch: pytest.MonkeyPatch) -> None: @@ -507,14 +314,14 @@ def test_full_verify_includes_full_pytest_without_testmon(monkeypatch: pytest.Mo assert isolated_command[isolated_command.index("-n") + 1] == "0" -def test_seed_collection_refuses_parallel_worker_overrides(monkeypatch: pytest.MonkeyPatch) -> None: +def test_seed_testmon_worker_count_can_be_overridden(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setenv("POLYLOGUE_PYTEST_WORKERS", "4") steps = build_verify_steps(quick=False, lab=False, skip_slow=False, seed_testmon=True) label, command = steps[-1] - assert label == "pytest seed-testmon collect" - assert command[command.index("-n") + 1] == "0" + assert label == "pytest seed-testmon" + assert command[command.index("-n") + 1] == "4" def test_seed_defaults_to_managed_scratch(monkeypatch: pytest.MonkeyPatch) -> None: From bf960b07052ed7602dede5d73fa8d4e29cacfcf5 Mon Sep 17 00:00:00 2001 From: Sinity Date: Wed, 12 Aug 2026 11:02:34 +0200 Subject: [PATCH 10/14] fix(testmon): harden parallel seed ledgers --- devtools/verify.py | 15 ++---- polylogue/daemon/convergence_stages.py | 53 +++++++++++++++++++ tests/conftest.py | 9 ++++ tests/infra/convergence_harness.py | 37 ++++++------- .../devtools/test_pytest_progress_plugin.py | 32 +++++++---- tests/unit/devtools/test_verify.py | 2 +- 6 files changed, 105 insertions(+), 43 deletions(-) diff --git a/devtools/verify.py b/devtools/verify.py index 351bb12e7f..5008270ac4 100644 --- a/devtools/verify.py +++ b/devtools/verify.py @@ -216,7 +216,7 @@ def _format_completion_notification( TESTMON_SEED_STAMP = Path(".cache/testmon/seed.json") TESTMON_SEED_ATTEMPT = Path(".cache/testmon/seed-attempt.json") TESTMON_AFFECTED_STAMP = Path(".cache/testmon/affected.json") -TESTMON_SEED_PROTOCOL_VERSION = 6 +TESTMON_SEED_PROTOCOL_VERSION = 7 PYTEST_REPORT_DIR = Path(".cache/verify") PYTEST_REPORT_PATH = PYTEST_REPORT_DIR / "last-pytest.json" PYTEST_JUNIT_REPORT_DIR = Path(".cache/test-reports") @@ -1972,11 +1972,6 @@ def build_verify_steps( "-q", "--tb=short", "--ignore=tests/integration", - # Benchmark files are an explicit campaign surface. A number of - # them are correctness-shaped and lack the benchmark marker, so a - # marker expression alone cannot keep performance probes out of - # the correctness/testmon corpus. - "--ignore=tests/benchmarks", "--durations=10", f"--junitxml={_report_dir}/verify-latest.xml", "--json-report", @@ -1985,11 +1980,9 @@ def build_verify_steps( "-p", "devtools.pytest_progress_plugin", ] - # Benchmark cases are an explicit campaign surface, not part of the - # correctness/testmon seed. Keeping them out here is important: a - # benchmark marker is not necessarily paired with ``slow`` or a scale - # marker, and a serial shard would otherwise spend minutes executing a - # performance probe before it can checkpoint any correctness nodes. + # Benchmark cases opt out through their marker. The benchmarks tree + # also contains correctness-shaped scale-tier tests which must remain + # in the default/testmon collection. base_marker = f"not benchmark and {scale_marker_expr}" if skip_slow: base_marker = f"not slow and {base_marker}" diff --git a/polylogue/daemon/convergence_stages.py b/polylogue/daemon/convergence_stages.py index 09c5d7b8c3..c2f1d4f6e6 100644 --- a/polylogue/daemon/convergence_stages.py +++ b/polylogue/daemon/convergence_stages.py @@ -390,6 +390,17 @@ def execute_sessions(session_ids: Sequence[str]) -> StageExecuteReturn: _CLAUDE_WORKFLOW_RECORDED_GAP_LIMIT = 20 +def _record_convergence_fts_freshness_sync(conn: sqlite3.Connection) -> None: + """Publish freshness after insights materialization. + + Insights writes ``session_work_events`` after the FTS stage has run, so + the daemon must record the combined post-convergence snapshot here. + """ + from polylogue.daemon.fts_startup import record_fts_freshness_snapshot_sync + + record_fts_freshness_snapshot_sync(conn) + + def _record_claude_workflow_stage_event(archive_root: Path, summary: object) -> None: """Persist the materialization summary so a readiness surface can read it. @@ -733,6 +744,46 @@ def execute_sessions(session_ids: Sequence[str]) -> StageExecuteReturn: ) +def make_fts_freshness_stage(db_path: Path) -> ConvergenceStage: + """Record one archive-wide FTS snapshot after all materializers run.""" + + def record() -> bool: + archive_db = _active_archive_index_path(db_path) or db_path + if not archive_db.exists(): + return True + from polylogue.storage.sqlite.connection_profile import open_connection + + with open_connection(archive_db) as conn: + _record_convergence_fts_freshness_sync(conn) + return True + + def check(_path: Path) -> bool: + return db_path.exists() or _active_archive_index_path(db_path) is not None + + def execute(_path: Path) -> StageExecuteReturn: + return record() + + def check_many(paths: Sequence[Path]) -> set[Path]: + return set(paths) if paths and check(paths[0]) else set() + + def execute_many(_paths: Sequence[Path]) -> StageExecuteReturn: + return record() + + def execute_sessions(_session_ids: Sequence[str]) -> StageExecuteReturn: + return record() + + return ConvergenceStage( + name="fts_freshness", + description="Record archive-wide FTS freshness after FTS and insight convergence", + check=check, + execute=execute, + check_many=check_many, + execute_many=execute_many, + check_sessions=lambda session_ids: set(session_ids), + execute_sessions=execute_sessions, + ) + + def _sinex_session_ids_for_paths( db_path: Path, paths: Sequence[Path], @@ -1126,6 +1177,7 @@ def make_default_convergence_stages( make_embed_stage(db_path, defer=embed_defer), make_claude_workflow_stage(db_path), make_insights_stage(db_path), + make_fts_freshness_stage(db_path), make_standing_query_stage(db_path, evaluator=ArchiveCanonicalPlanEvaluator(db_path)), ) ) @@ -2371,6 +2423,7 @@ def _archive_insights_execute_ids( "make_default_convergence_stages", "make_embed_stage", "make_fts_stage", + "make_fts_freshness_stage", "make_insights_stage", "make_raw_authority_verdict_cache_stage", "make_raw_parse_recovery_stage", diff --git a/tests/conftest.py b/tests/conftest.py index 97a170327d..92d1317c8a 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -640,6 +640,15 @@ def _clear_polylogue_env( # are stripped automatically. from tests.infra.schema_access import ALLOW_MISSING_SCHEMAS_ENV + # A pytest process launched by a test inherits the outer supervisor's + # selection/summary destinations. Letting the nested process write them + # corrupts the outer shard ledger; its event reports are still useful and + # remain on the shared event stream. ``PYTEST_CURRENT_TEST`` is present + # for the parent test and absent at normal top-level pytest startup. + if os.environ.get("PYTEST_CURRENT_TEST") and os.environ.get("POLYLOGUE_VERIFY_RUN_ID"): + for key in ("POLYLOGUE_PYTEST_SELECTION_PATH", "POLYLOGUE_PYTEST_SUMMARY_PATH"): + monkeypatch.delenv(key, raising=False) + for key in list(os.environ): # ALLOW_MISSING_SCHEMAS_ENV is a test-only escape hatch (not operator # config) for lanes that intentionally run without packaged provider diff --git a/tests/infra/convergence_harness.py b/tests/infra/convergence_harness.py index 28f85986e8..1bf45f8b67 100644 --- a/tests/infra/convergence_harness.py +++ b/tests/infra/convergence_harness.py @@ -28,7 +28,7 @@ from polylogue.core.enums import BlockType, Provider from polylogue.core.outcomes import OutcomeStatus from polylogue.daemon.convergence import DaemonConverger, SessionState -from polylogue.daemon.convergence_stages import make_fts_stage, make_insights_stage +from polylogue.daemon.convergence_stages import make_fts_freshness_stage, make_fts_stage, make_insights_stage from polylogue.maintenance.archive_verification import ArchiveVerificationReport, verify_archive from polylogue.pipeline.ids import session_content_hash from polylogue.pipeline.ids import session_id as make_session_id @@ -45,7 +45,6 @@ ) from polylogue.storage.blob_publication import ArchiveBlobPublisher, consume_blob_publication_receipt from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_archive_tier -from polylogue.storage.sqlite.archive_tiers.revision_governance import record_current_parser_source_census from polylogue.storage.sqlite.archive_tiers.source_write import ArchiveSourceBlobRef, write_source_raw_session from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier from polylogue.storage.sqlite.archive_tiers.write import write_parsed_session_to_archive @@ -280,7 +279,7 @@ def ingest_convergence_pathology( revision=RawRevisionEnvelope( logical_source_key=str(make_session_id(session.source_name, session.provider_session_id)), kind=RawRevisionKind.FULL, - source_revision=content_hash, + source_revision=raw_blob_hash, acquisition_generation=index, ), blob_publication_receipt_id=raw_blob_publisher.receipt_id(raw_blob_hash), @@ -294,8 +293,6 @@ def ingest_convergence_pathology( ) for attachment_receipt, attachment_hash_bytes in attachment_receipts: consume_blob_publication_receipt(source_conn, attachment_receipt, attachment_hash_bytes) - if source_conn.execute("SELECT 1 FROM raw_sessions WHERE raw_id = ?", (raw_id,)).fetchone() is not None: - record_current_parser_source_census(source_conn, raw_id, parser_sessions=[session]) if raw_blob_size != len(payload): raise AssertionError(f"published raw payload size drifted for {source_path}") payload_model = SessionWritePayload( @@ -345,18 +342,16 @@ def converge_convergence_archive(archive: ConvergenceArchive) -> dict[str, Sessi str(row[0]) for row in conn.execute("SELECT session_id FROM sessions ORDER BY session_id") ) converger = DaemonConverger( - (make_fts_stage(archive.root / "index.db"), make_insights_stage(archive.root / "index.db")) + ( + make_fts_stage(archive.root / "index.db"), + make_insights_stage(archive.root / "index.db"), + make_fts_freshness_stage(archive.root / "index.db"), + ) ) states, _timings = converger.converge_sessions(persisted_session_ids) not_converged = {session_id: state.last_error for session_id, state in states.items() if not state.converged} if not_converged: raise AssertionError(f"production convergence left pending work: {not_converged}") - # Insights can materialize work-event rows after the FTS stage has run. - # Refresh the shared freshness ledger only once both real stages complete. - from polylogue.daemon.fts_startup import record_fts_freshness_snapshot_sync - - with sqlite3.connect(archive.root / "index.db") as conn: - record_fts_freshness_snapshot_sync(conn) _analyze_registry_tables(archive.root / "index.db") return states @@ -427,15 +422,17 @@ def assert_derived_readiness_equivalent(left: Path, right: Path) -> None: f"primary insight readiness is incomplete for {root}: " f"missing={sorted(missing_models)}, unready={unready_models}" ) - # The status projection also reports secondary work-event FTS and - # retrieval surfaces. They remain in the equality snapshot, as does - # the production messages_fts status. The two-stage route owns - # messages-FTS repair for changed sessions, while the neutral parser - # fixture can expose archive-wide excess rows from provider-derived - # blocks. Keep that production readiness signal in the equality law - # instead of asserting a global repair this route does not promise. + # This harness starts at ParsedSession, not provider-wire bytes. Raw + # parser-census readiness is therefore intentionally outside this + # derived-materialization law; provider replay/census tests own it. readiness = archive_readiness_status(root) - if readiness.get("checked") is not True or readiness.get("blocked_surface_count") != 0: + surfaces = readiness.get("surfaces", {}) + blocked_non_source = [ + name + for name, surface in surfaces.items() + if name != "raw_artifacts" and isinstance(surface, dict) and surface.get("ready") is not True + ] + if readiness.get("checked") is not True or blocked_non_source: raise AssertionError(f"archive readiness is incomplete for {root}: {readiness!r}") if left_snapshot != right_snapshot: raise AssertionError( diff --git a/tests/unit/devtools/test_pytest_progress_plugin.py b/tests/unit/devtools/test_pytest_progress_plugin.py index e344db369b..ee50b3981a 100644 --- a/tests/unit/devtools/test_pytest_progress_plugin.py +++ b/tests/unit/devtools/test_pytest_progress_plugin.py @@ -14,16 +14,17 @@ @pytest.fixture(autouse=True) -def _restore_plugin_state(monkeypatch: pytest.MonkeyPatch) -> Iterator[None]: - # Unit tests own their event destinations; do not let a surrounding - # managed verify invocation redirect them into its step artifacts. - for name in ( - "POLYLOGUE_PYTEST_EVENTS_DIR", - "POLYLOGUE_PYTEST_EVENTS_PATH", - "POLYLOGUE_PYTEST_SELECTION_PATH", - "POLYLOGUE_PYTEST_SUMMARY_PATH", - ): - monkeypatch.delenv(name, raising=False) +def _restore_plugin_state(monkeypatch: pytest.MonkeyPatch, request: pytest.FixtureRequest) -> Iterator[None]: + # Direct helper tests own their destinations. The subprocess regression is + # the one test that must preserve a managed outer event stream. + if request.node.name != "test_managed_event_ledger_survives_test_host_environment_scrub": + for name in ( + "POLYLOGUE_PYTEST_EVENTS_DIR", + "POLYLOGUE_PYTEST_EVENTS_PATH", + "POLYLOGUE_PYTEST_SELECTION_PATH", + "POLYLOGUE_PYTEST_SUMMARY_PATH", + ): + monkeypatch.delenv(name, raising=False) selected_count = pytest_progress_plugin._SELECTED_COUNT deselected_count = pytest_progress_plugin._DESELECTED_COUNT deselected_nodeids = list(pytest_progress_plugin._DESELECTED_NODEIDS_SAMPLE) @@ -102,7 +103,16 @@ def test_managed_event_ledger_survives_test_host_environment_scrub(tmp_path: Pat assert result.returncode == 0, result.stdout + result.stderr events = [json.loads(line) for path in events_dir.glob("*.jsonl") for line in path.read_text().splitlines()] reports = [event for event in events if event.get("event") == "test_report"] - assert {event["when"] for event in reports} == {"setup", "call", "teardown"} + assert len(reports) == 3 + assert {(event["nodeid"], event["when"], event["outcome"], event["run_id"]) for event in reports} == { + ( + "tests/unit/core/test_identity_law.py::test_session_id_is_origin_native_id", + phase, + "passed", + "subprocess-regression", + ) + for phase in ("setup", "call", "teardown") + } def test_progress_plugin_records_node_start_and_finish( diff --git a/tests/unit/devtools/test_verify.py b/tests/unit/devtools/test_verify.py index 61f08e5965..6346441d7b 100644 --- a/tests/unit/devtools/test_verify.py +++ b/tests/unit/devtools/test_verify.py @@ -255,7 +255,7 @@ def test_seed_testmon_runs_full_collection_without_selection(monkeypatch: pytest label, command = steps[-1] assert label == "pytest seed-testmon" - assert command[command.index("--ignore=tests/benchmarks")] == "--ignore=tests/benchmarks" + assert "--ignore=tests/benchmarks" not in command assert "--collect-only" not in command assert "--testmon" in command assert "--testmon-noselect" in command From 0aaf4d0d48f70b003eebc2516dd3278621bbd5ae Mon Sep 17 00:00:00 2001 From: Sinity Date: Wed, 12 Aug 2026 11:17:41 +0200 Subject: [PATCH 11/14] perf(testmon): use managed worker capacity --- devtools/verify.py | 5 ++++- tests/unit/devtools/test_verify.py | 4 ++-- 2 files changed, 6 insertions(+), 3 deletions(-) diff --git a/devtools/verify.py b/devtools/verify.py index 5008270ac4..87511cb425 100644 --- a/devtools/verify.py +++ b/devtools/verify.py @@ -1994,7 +1994,10 @@ def build_verify_steps( else: pytest_cmd.append("--testmon-noselect") label = "pytest seed-testmon" - pytest_cmd.extend(_pytest_worker_args(maximum=4)) + # The runtime policy is memory-aware (currently eight workers on + # this host). The old four-worker cap made the 20k-node seed + # effectively crawl even though the controller remained healthy. + pytest_cmd.extend(_pytest_worker_args(maximum=8)) steps.append((label, pytest_cmd)) elif full_pytest: # #1775: the full diagnostic runs as two lanes. The bulk lane keeps diff --git a/tests/unit/devtools/test_verify.py b/tests/unit/devtools/test_verify.py index 6346441d7b..a546a58d21 100644 --- a/tests/unit/devtools/test_verify.py +++ b/tests/unit/devtools/test_verify.py @@ -260,7 +260,7 @@ def test_seed_testmon_runs_full_collection_without_selection(monkeypatch: pytest assert "--testmon" in command assert "--testmon-noselect" in command assert "-n" in command - assert command[command.index("-n") + 1] == "4" + assert command[command.index("-n") + 1] == "8" def test_seed_testmon_caps_adaptive_workers(monkeypatch: pytest.MonkeyPatch) -> None: @@ -271,7 +271,7 @@ def test_seed_testmon_caps_adaptive_workers(monkeypatch: pytest.MonkeyPatch) -> label, command = steps[-1] assert label == "pytest seed-testmon" - assert command[command.index("-n") + 1] == "4" + assert command[command.index("-n") + 1] == "8" def test_resumed_seed_uses_affected_selection_for_remaining_tests() -> None: From 120391920a04afb9c4ba8f8e6d1f009a28de95c8 Mon Sep 17 00:00:00 2001 From: Sinity Date: Wed, 12 Aug 2026 11:58:18 +0200 Subject: [PATCH 12/14] perf(testmon): widen managed seed worker pool --- devtools/verify.py | 9 +++++---- tests/unit/devtools/test_verify.py | 2 +- 2 files changed, 6 insertions(+), 5 deletions(-) diff --git a/devtools/verify.py b/devtools/verify.py index 87511cb425..73b0d2ca8c 100644 --- a/devtools/verify.py +++ b/devtools/verify.py @@ -1994,10 +1994,11 @@ def build_verify_steps( else: pytest_cmd.append("--testmon-noselect") label = "pytest seed-testmon" - # The runtime policy is memory-aware (currently eight workers on - # this host). The old four-worker cap made the 20k-node seed - # effectively crawl even though the controller remained healthy. - pytest_cmd.extend(_pytest_worker_args(maximum=8)) + # The runtime policy is memory-aware. Keep the seed below the + # host's twelve-worker hard ceiling: ten workers fit the measured + # memory envelope while leaving headroom for the controller and + # supervisor, and materially shorten the 20k-node seed. + pytest_cmd.extend(_pytest_worker_args(maximum=10)) steps.append((label, pytest_cmd)) elif full_pytest: # #1775: the full diagnostic runs as two lanes. The bulk lane keeps diff --git a/tests/unit/devtools/test_verify.py b/tests/unit/devtools/test_verify.py index a546a58d21..0f18df59de 100644 --- a/tests/unit/devtools/test_verify.py +++ b/tests/unit/devtools/test_verify.py @@ -271,7 +271,7 @@ def test_seed_testmon_caps_adaptive_workers(monkeypatch: pytest.MonkeyPatch) -> label, command = steps[-1] assert label == "pytest seed-testmon" - assert command[command.index("-n") + 1] == "8" + assert command[command.index("-n") + 1] == "10" def test_resumed_seed_uses_affected_selection_for_remaining_tests() -> None: From a55b7f5503c8697204eeef5642060a13aff1e2b0 Mon Sep 17 00:00:00 2001 From: Sinity Date: Wed, 12 Aug 2026 12:17:20 +0200 Subject: [PATCH 13/14] fix(testmon): isolate nested ledgers and convergence receipts --- polylogue/daemon/convergence_stages.py | 8 +++--- polylogue/daemon/fts_startup.py | 5 ++-- tests/conftest.py | 23 +++++++++++----- tests/infra/convergence_harness.py | 27 ++++++++++++------- .../devtools/test_pytest_progress_plugin.py | 4 +++ 5 files changed, 45 insertions(+), 22 deletions(-) diff --git a/polylogue/daemon/convergence_stages.py b/polylogue/daemon/convergence_stages.py index c2f1d4f6e6..5bac125b3e 100644 --- a/polylogue/daemon/convergence_stages.py +++ b/polylogue/daemon/convergence_stages.py @@ -390,7 +390,7 @@ def execute_sessions(session_ids: Sequence[str]) -> StageExecuteReturn: _CLAUDE_WORKFLOW_RECORDED_GAP_LIMIT = 20 -def _record_convergence_fts_freshness_sync(conn: sqlite3.Connection) -> None: +def _record_convergence_fts_freshness_sync(conn: sqlite3.Connection) -> bool: """Publish freshness after insights materialization. Insights writes ``session_work_events`` after the FTS stage has run, so @@ -398,7 +398,7 @@ def _record_convergence_fts_freshness_sync(conn: sqlite3.Connection) -> None: """ from polylogue.daemon.fts_startup import record_fts_freshness_snapshot_sync - record_fts_freshness_snapshot_sync(conn) + return record_fts_freshness_snapshot_sync(conn) def _record_claude_workflow_stage_event(archive_root: Path, summary: object) -> None: @@ -754,8 +754,7 @@ def record() -> bool: from polylogue.storage.sqlite.connection_profile import open_connection with open_connection(archive_db) as conn: - _record_convergence_fts_freshness_sync(conn) - return True + return _record_convergence_fts_freshness_sync(conn) def check(_path: Path) -> bool: return db_path.exists() or _active_archive_index_path(db_path) is not None @@ -1177,7 +1176,6 @@ def make_default_convergence_stages( make_embed_stage(db_path, defer=embed_defer), make_claude_workflow_stage(db_path), make_insights_stage(db_path), - make_fts_freshness_stage(db_path), make_standing_query_stage(db_path, evaluator=ArchiveCanonicalPlanEvaluator(db_path)), ) ) diff --git a/polylogue/daemon/fts_startup.py b/polylogue/daemon/fts_startup.py index 16ca641cfa..da2b0d8805 100644 --- a/polylogue/daemon/fts_startup.py +++ b/polylogue/daemon/fts_startup.py @@ -52,7 +52,7 @@ def missing_fts_triggers_sync(conn: sqlite3.Connection) -> list[str]: return [name for name in expected if name not in present] -def record_fts_freshness_snapshot_sync(conn: sqlite3.Connection) -> None: +def record_fts_freshness_snapshot_sync(conn: sqlite3.Connection) -> bool: """Write per-surface freshness rows after a successful startup readiness pass. Without this, the bounded-recovery and healthy startup paths leave @@ -67,12 +67,13 @@ def record_fts_freshness_snapshot_sync(conn: sqlite3.Connection) -> None: snapshot = fts_invariant_snapshot_sync(conn) except sqlite3.Error: logger.warning("daemon: FTS startup freshness snapshot failed", exc_info=True) - return + return False record_fts_invariant_snapshot_sync(conn, snapshot) from polylogue.storage.fts.drift_sampling import sample_fts_drift_to_ops_sync sample_fts_drift_to_ops_sync(conn) + return True def active_fts_triggers_sync(conn: sqlite3.Connection) -> tuple[str, ...]: diff --git a/tests/conftest.py b/tests/conftest.py index 92d1317c8a..cd0dc92c8a 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -641,12 +641,23 @@ def _clear_polylogue_env( from tests.infra.schema_access import ALLOW_MISSING_SCHEMAS_ENV # A pytest process launched by a test inherits the outer supervisor's - # selection/summary destinations. Letting the nested process write them - # corrupts the outer shard ledger; its event reports are still useful and - # remain on the shared event stream. ``PYTEST_CURRENT_TEST`` is present - # for the parent test and absent at normal top-level pytest startup. - if os.environ.get("PYTEST_CURRENT_TEST") and os.environ.get("POLYLOGUE_VERIFY_RUN_ID"): - for key in ("POLYLOGUE_PYTEST_SELECTION_PATH", "POLYLOGUE_PYTEST_SUMMARY_PATH"): + # ledger destinations and run identity. Letting the nested process write + # them corrupts the outer shard ledger: its reports are not evidence that + # the parent shard executed those nodes. ``PYTEST_CURRENT_TEST`` is + # present for the parent test and absent at normal top-level pytest + # startup, so nested pytest gets an entirely private progress namespace. + if ( + os.environ.get("PYTEST_CURRENT_TEST") + and os.environ.get("POLYLOGUE_VERIFY_RUN_ID") + and not os.environ.get("POLYLOGUE_PYTEST_NESTED_PRIVATE") + ): + for key in ( + "POLYLOGUE_VERIFY_RUN_ID", + "POLYLOGUE_PYTEST_EVENTS_DIR", + "POLYLOGUE_PYTEST_EVENTS_PATH", + "POLYLOGUE_PYTEST_SELECTION_PATH", + "POLYLOGUE_PYTEST_SUMMARY_PATH", + ): monkeypatch.delenv(key, raising=False) for key in list(os.environ): diff --git a/tests/infra/convergence_harness.py b/tests/infra/convergence_harness.py index 1bf45f8b67..9cf5b3c66c 100644 --- a/tests/infra/convergence_harness.py +++ b/tests/infra/convergence_harness.py @@ -24,7 +24,6 @@ import polylogue.daemon.convergence_stages as convergence_stages from polylogue.archive.message.roles import Role -from polylogue.archive.revision_authority import RawRevisionEnvelope, RawRevisionKind from polylogue.core.enums import BlockType, Provider from polylogue.core.outcomes import OutcomeStatus from polylogue.daemon.convergence import DaemonConverger, SessionState @@ -45,7 +44,8 @@ ) from polylogue.storage.blob_publication import ArchiveBlobPublisher, consume_blob_publication_receipt from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_archive_tier -from polylogue.storage.sqlite.archive_tiers.source_write import ArchiveSourceBlobRef, write_source_raw_session +from polylogue.storage.sqlite.archive_tiers.raw_admission import PriorRawHead, admit_raw_observation +from polylogue.storage.sqlite.archive_tiers.source_write import ArchiveSourceBlobRef from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier from polylogue.storage.sqlite.archive_tiers.write import write_parsed_session_to_archive from polylogue.storage.sqlite.connection import open_connection @@ -228,6 +228,7 @@ def ingest_convergence_pathology( selected = _validate_session_indexes(pathology, session_indexes) source_paths: list[Path] = [] session_ids: list[str] = [] + prior_heads: dict[str, PriorRawHead] = {} for index in selected: session = _parsed_session(pathology.sessions[index], corpus_index=index) content_hash = str(session_content_hash(session)) @@ -265,9 +266,10 @@ def ingest_convergence_pathology( attachment_receipts.append((attachment_receipt, bytes.fromhex(attachment_hash))) session = session.model_copy(update={"attachments": preacquired_attachments}) raw_blob_publisher.flush() + logical_source_key = str(make_session_id(session.source_name, session.provider_session_id)) with sqlite3.connect(root / "source.db") as source_conn: with source_conn: - raw_id = write_source_raw_session( + admission = admit_raw_observation( source_conn, origin="codex-session", capture_mode=Provider.CODEX, @@ -276,16 +278,23 @@ def ingest_convergence_pathology( payload=payload, acquired_at_ms=_acquired_at_ms(index), native_id=session.provider_session_id, - revision=RawRevisionEnvelope( - logical_source_key=str(make_session_id(session.source_name, session.provider_session_id)), - kind=RawRevisionKind.FULL, - source_revision=raw_blob_hash, - acquisition_generation=index, - ), + logical_source_key=logical_source_key, + prior_head=prior_heads.get(logical_source_key), blob_publication_receipt_id=raw_blob_publisher.receipt_id(raw_blob_hash), additional_blob_refs=tuple(attachment_blob_refs), manage_transaction=False, ) + if admission.arm.value not in {"baseline", "append", "supersede"}: + raise AssertionError(f"raw fixture admission was not executable: {admission!r}") + prior = prior_heads.get(logical_source_key) + prior_heads[logical_source_key] = PriorRawHead( + raw_id=admission.raw_id, + source_revision=raw_blob_hash, + payload=payload, + baseline_raw_id=prior.baseline_raw_id if prior and prior.baseline_raw_id else admission.raw_id, + acquisition_generation=(prior.acquisition_generation + 1) if prior else 0, + ) + raw_id = admission.raw_id consume_blob_publication_receipt( source_conn, raw_blob_publisher.receipt_id(raw_blob_hash), diff --git a/tests/unit/devtools/test_pytest_progress_plugin.py b/tests/unit/devtools/test_pytest_progress_plugin.py index ee50b3981a..02339dc9ad 100644 --- a/tests/unit/devtools/test_pytest_progress_plugin.py +++ b/tests/unit/devtools/test_pytest_progress_plugin.py @@ -81,6 +81,10 @@ def test_managed_event_ledger_survives_test_host_environment_scrub(tmp_path: Pat "POLYLOGUE_PYTEST_SELECTION_PATH": str(tmp_path / "selection.json"), "POLYLOGUE_PYTEST_SUMMARY_PATH": str(tmp_path / "summary.json"), "POLYLOGUE_VERIFY_RUN_ID": "subprocess-regression", + # This child deliberately owns a private destination so the test + # can verify the nested-process isolation contract without making + # its reports part of the outer seed ledger. + "POLYLOGUE_PYTEST_NESTED_PRIVATE": "1", } ) result = subprocess.run( From 31e08362ba92d22979ec0c1ad0fcbe42b55c56d0 Mon Sep 17 00:00:00 2001 From: Sinity Date: Wed, 12 Aug 2026 12:29:11 +0200 Subject: [PATCH 14/14] fix(testmon): isolate child collection ledgers --- polylogue/daemon/convergence_stages.py | 51 ------------------- tests/conftest.py | 39 +++++++++----- tests/infra/convergence_harness.py | 7 ++- .../devtools/test_pytest_progress_plugin.py | 6 +++ 4 files changed, 38 insertions(+), 65 deletions(-) diff --git a/polylogue/daemon/convergence_stages.py b/polylogue/daemon/convergence_stages.py index 5bac125b3e..09c5d7b8c3 100644 --- a/polylogue/daemon/convergence_stages.py +++ b/polylogue/daemon/convergence_stages.py @@ -390,17 +390,6 @@ def execute_sessions(session_ids: Sequence[str]) -> StageExecuteReturn: _CLAUDE_WORKFLOW_RECORDED_GAP_LIMIT = 20 -def _record_convergence_fts_freshness_sync(conn: sqlite3.Connection) -> bool: - """Publish freshness after insights materialization. - - Insights writes ``session_work_events`` after the FTS stage has run, so - the daemon must record the combined post-convergence snapshot here. - """ - from polylogue.daemon.fts_startup import record_fts_freshness_snapshot_sync - - return record_fts_freshness_snapshot_sync(conn) - - def _record_claude_workflow_stage_event(archive_root: Path, summary: object) -> None: """Persist the materialization summary so a readiness surface can read it. @@ -744,45 +733,6 @@ def execute_sessions(session_ids: Sequence[str]) -> StageExecuteReturn: ) -def make_fts_freshness_stage(db_path: Path) -> ConvergenceStage: - """Record one archive-wide FTS snapshot after all materializers run.""" - - def record() -> bool: - archive_db = _active_archive_index_path(db_path) or db_path - if not archive_db.exists(): - return True - from polylogue.storage.sqlite.connection_profile import open_connection - - with open_connection(archive_db) as conn: - return _record_convergence_fts_freshness_sync(conn) - - def check(_path: Path) -> bool: - return db_path.exists() or _active_archive_index_path(db_path) is not None - - def execute(_path: Path) -> StageExecuteReturn: - return record() - - def check_many(paths: Sequence[Path]) -> set[Path]: - return set(paths) if paths and check(paths[0]) else set() - - def execute_many(_paths: Sequence[Path]) -> StageExecuteReturn: - return record() - - def execute_sessions(_session_ids: Sequence[str]) -> StageExecuteReturn: - return record() - - return ConvergenceStage( - name="fts_freshness", - description="Record archive-wide FTS freshness after FTS and insight convergence", - check=check, - execute=execute, - check_many=check_many, - execute_many=execute_many, - check_sessions=lambda session_ids: set(session_ids), - execute_sessions=execute_sessions, - ) - - def _sinex_session_ids_for_paths( db_path: Path, paths: Sequence[Path], @@ -2421,7 +2371,6 @@ def _archive_insights_execute_ids( "make_default_convergence_stages", "make_embed_stage", "make_fts_stage", - "make_fts_freshness_stage", "make_insights_stage", "make_raw_authority_verdict_cache_stage", "make_raw_parse_recovery_stage", diff --git a/tests/conftest.py b/tests/conftest.py index cd0dc92c8a..29b595e8ce 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -82,6 +82,7 @@ def pytest_configure(config: pytest.Config) -> None: """Register custom markers and choose the managed test temp root.""" + _scrub_nested_verify_ledgers() if _CHECKOUT_GUARD_ERROR is not None: # Refuse before collection: every test in this run would otherwise # exercise a `polylogue` package from a different checkout than the @@ -413,6 +414,20 @@ def _reclaim_passing_test_tmp_path( ) +def _scrub_nested_verify_ledgers() -> None: + """Keep a pytest child from writing the parent verify ledgers.""" + nested = ( + os.environ.get("PYTEST_CURRENT_TEST") or os.environ.get("POLYLOGUE_PYTEST_NESTED_PRIVATE") + ) and os.environ.get("POLYLOGUE_VERIFY_RUN_ID") + if not nested: + return + for key in ("POLYLOGUE_PYTEST_SELECTION_PATH", "POLYLOGUE_PYTEST_SUMMARY_PATH"): + os.environ.pop(key, None) + if not os.environ.get("POLYLOGUE_PYTEST_NESTED_PRIVATE"): + for key in ("POLYLOGUE_VERIFY_RUN_ID", "POLYLOGUE_PYTEST_EVENTS_DIR", "POLYLOGUE_PYTEST_EVENTS_PATH"): + os.environ.pop(key, None) + + @pytest.fixture(autouse=True) def _close_test_opened_sqlite_connections( monkeypatch: pytest.MonkeyPatch, @@ -646,19 +661,19 @@ def _clear_polylogue_env( # the parent shard executed those nodes. ``PYTEST_CURRENT_TEST`` is # present for the parent test and absent at normal top-level pytest # startup, so nested pytest gets an entirely private progress namespace. - if ( - os.environ.get("PYTEST_CURRENT_TEST") - and os.environ.get("POLYLOGUE_VERIFY_RUN_ID") - and not os.environ.get("POLYLOGUE_PYTEST_NESTED_PRIVATE") - ): - for key in ( - "POLYLOGUE_VERIFY_RUN_ID", - "POLYLOGUE_PYTEST_EVENTS_DIR", - "POLYLOGUE_PYTEST_EVENTS_PATH", - "POLYLOGUE_PYTEST_SELECTION_PATH", - "POLYLOGUE_PYTEST_SUMMARY_PATH", - ): + nested_pytest = ( + os.environ.get("PYTEST_CURRENT_TEST") or os.environ.get("POLYLOGUE_PYTEST_NESTED_PRIVATE") + ) and os.environ.get("POLYLOGUE_VERIFY_RUN_ID") + if nested_pytest: + # Selection and summary are process-global destinations owned by the + # parent verify run and must never be replaced by a child. A test that + # explicitly supplies a private event namespace may retain only its + # event stream for a direct subprocess regression check. + for key in ("POLYLOGUE_PYTEST_SELECTION_PATH", "POLYLOGUE_PYTEST_SUMMARY_PATH"): monkeypatch.delenv(key, raising=False) + if not os.environ.get("POLYLOGUE_PYTEST_NESTED_PRIVATE"): + for key in ("POLYLOGUE_VERIFY_RUN_ID", "POLYLOGUE_PYTEST_EVENTS_DIR", "POLYLOGUE_PYTEST_EVENTS_PATH"): + monkeypatch.delenv(key, raising=False) for key in list(os.environ): # ALLOW_MISSING_SCHEMAS_ENV is a test-only escape hatch (not operator diff --git a/tests/infra/convergence_harness.py b/tests/infra/convergence_harness.py index 9cf5b3c66c..3b15b4eaa1 100644 --- a/tests/infra/convergence_harness.py +++ b/tests/infra/convergence_harness.py @@ -27,7 +27,8 @@ from polylogue.core.enums import BlockType, Provider from polylogue.core.outcomes import OutcomeStatus from polylogue.daemon.convergence import DaemonConverger, SessionState -from polylogue.daemon.convergence_stages import make_fts_freshness_stage, make_fts_stage, make_insights_stage +from polylogue.daemon.convergence_stages import make_fts_stage, make_insights_stage +from polylogue.daemon.fts_startup import record_fts_freshness_snapshot_sync from polylogue.maintenance.archive_verification import ArchiveVerificationReport, verify_archive from polylogue.pipeline.ids import session_content_hash from polylogue.pipeline.ids import session_id as make_session_id @@ -354,13 +355,15 @@ def converge_convergence_archive(archive: ConvergenceArchive) -> dict[str, Sessi ( make_fts_stage(archive.root / "index.db"), make_insights_stage(archive.root / "index.db"), - make_fts_freshness_stage(archive.root / "index.db"), ) ) states, _timings = converger.converge_sessions(persisted_session_ids) not_converged = {session_id: state.last_error for session_id, state in states.items() if not state.converged} if not_converged: raise AssertionError(f"production convergence left pending work: {not_converged}") + with sqlite3.connect(archive.root / "index.db") as conn: + if not record_fts_freshness_snapshot_sync(conn): + raise AssertionError("exact FTS freshness snapshot failed after production convergence") _analyze_registry_tables(archive.root / "index.db") return states diff --git a/tests/unit/devtools/test_pytest_progress_plugin.py b/tests/unit/devtools/test_pytest_progress_plugin.py index 02339dc9ad..b59ffe4572 100644 --- a/tests/unit/devtools/test_pytest_progress_plugin.py +++ b/tests/unit/devtools/test_pytest_progress_plugin.py @@ -87,6 +87,10 @@ def test_managed_event_ledger_survives_test_host_environment_scrub(tmp_path: Pat "POLYLOGUE_PYTEST_NESTED_PRIVATE": "1", } ) + selection_path = Path(env["POLYLOGUE_PYTEST_SELECTION_PATH"]) + summary_path = Path(env["POLYLOGUE_PYTEST_SUMMARY_PATH"]) + selection_path.write_text("selection-sentinel\n", encoding="utf-8") + summary_path.write_text("summary-sentinel\n", encoding="utf-8") result = subprocess.run( [ sys.executable, @@ -105,6 +109,8 @@ def test_managed_event_ledger_survives_test_host_environment_scrub(tmp_path: Pat check=False, ) assert result.returncode == 0, result.stdout + result.stderr + assert selection_path.read_text(encoding="utf-8") == "selection-sentinel\n" + assert summary_path.read_text(encoding="utf-8") == "summary-sentinel\n" events = [json.loads(line) for path in events_dir.glob("*.jsonl") for line in path.read_text().splitlines()] reports = [event for event in events if event.get("event") == "test_report"] assert len(reports) == 3