diff --git a/tests/architecture/test_content_digest_single_owner.py b/tests/architecture/test_content_digest_single_owner.py index dbad12cf89..45f4b27284 100644 --- a/tests/architecture/test_content_digest_single_owner.py +++ b/tests/architecture/test_content_digest_single_owner.py @@ -721,15 +721,22 @@ def _owner_text_reads(tree: ast.AST) -> list[int]: def _analyse(source: str, name: str = "") -> dict[str, Any]: tree = ast.parse(source, filename=name) root = _collect_scopes(tree) - return { - "tree": tree, - "root": root, - "statements": _shape_statements(tree, root), - "constructions": _regex_constructions(tree, root), - "owner_imports": _owner_import_map(tree), - "owner_text_reads": _owner_text_reads(tree), - "loaded": _loaded_names(tree), - } + try: + return { + "statements": _shape_statements(tree, root), + "constructions": _regex_constructions(tree, root), + "owner_imports": _owner_import_map(tree), + "owner_text_reads": _owner_text_reads(tree), + "loaded": _loaded_names(tree), + } + finally: + # Readers are finished. Break AST -> scope -> binding -> AST cycles; + # retaining them makes later collections traverse whole source trees. + for node in ast.walk(tree): + scope = getattr(node, _SCOPE_ATTRIBUTE, None) + if scope is not None: + scope.names.clear() + delattr(node, _SCOPE_ATTRIBUTE) _REPO_SCAN: list[dict[str, Any]] | None = None @@ -738,8 +745,8 @@ def _analyse(source: str, name: str = "") -> dict[str, Any]: def _repo_scan() -> list[dict[str, Any]]: """One pass over every module under `loopx/`, shared by all four layers. - 1,167 modules parse and analyse in about four seconds, so this runs once per test - session rather than once per layer, and no layer is given a different view of the tree. + Cache the consumed read model once per session, not the temporary syntax trees, + scopes or raw source. Every layer still sees the same complete analysed snapshot. """ global _REPO_SCAN @@ -753,7 +760,6 @@ def _repo_scan() -> list[dict[str, Any]]: source = path.read_text(encoding="utf-8") entry = _analyse(source, str(path)) entry["path"] = path - entry["source"] = source entry["relative"] = f"loopx/{path.relative_to(PACKAGE_ROOT).as_posix()}" parts = list(path.relative_to(PACKAGE_ROOT.parent).parts) parts[-1] = parts[-1][: -len(".py")] diff --git a/tests/architecture/test_reward_memory_field_vocabulary_owner.py b/tests/architecture/test_reward_memory_field_vocabulary_owner.py index 915b8f766f..91cec1c923 100644 --- a/tests/architecture/test_reward_memory_field_vocabulary_owner.py +++ b/tests/architecture/test_reward_memory_field_vocabulary_owner.py @@ -167,84 +167,113 @@ def folded_value(node: ast.expr) -> tuple[str, object] | None: return None +def _analyse(source: str): + """Derive the consumed sites once; syntax trees remain temporary.""" + + tree = ast.parse(source) + bindings = [] + for node in tree.body: + if isinstance(node, ast.Assign): + names = frozenset(t.id for t in node.targets if isinstance(t, ast.Name)) + bindings.append((node.lineno, names, folded_value(node.value))) + members, strings = [], [] + for node in ast.walk(tree): + value = collection_members(node) + if value is not None: + members.append((node.lineno, value)) + if isinstance(node, ast.Constant) and isinstance(node.value, str): + strings.append((node.lineno, node.value)) + return bindings, members, strings + + +def _binding_lines(bindings, name, kind, payload): + return [line for line, names, value in bindings + if name in names and value == (kind, payload)] + + +def _sites(analysis, name, kind, payload): + bindings, members, strings = analysis + named = _binding_lines(bindings, name, kind, payload) + inline = ([line for line, value in members if value == payload] if kind == "members" + else [line for line, value in strings if payload in value]) + return named, inline + + def binding_sites(source: str, name: str, kind: str, payload: object) -> list[int]: """Module-level assignments that bind this name to the owner's folded value.""" - found: list[int] = [] - for node in ast.parse(source).body: - if not isinstance(node, ast.Assign): - continue - if not any(isinstance(t, ast.Name) and t.id == name for t in node.targets): - continue - folded = folded_value(node.value) - if folded is not None and folded == (kind, payload): - found.append(node.lineno) - return found + return _binding_lines(_analyse(source)[0], name, kind, payload) def member_literal_sites(source: str, members: frozenset[str]) -> list[int]: - tree = ast.parse(source) - return [ - node.lineno - for node in ast.walk(tree) - if folded_value(node) == ("members", members) - ] + return _sites(_analyse(source), "", "members", members)[1] def body_literal_sites(source: str, body: str) -> list[int]: - tree = ast.parse(source) - return [ - node.lineno - for node in ast.walk(tree) - if isinstance(node, ast.Constant) and isinstance(node.value, str) and body in node.value - ] - + return _sites(_analyse(source), "", "body", body)[1] -def scan_sets(name: str) -> tuple[list[str], list[str]]: - members, owner_path, _module, _consumers = SET_DECISIONS[name] - return _scan(name, "members", members, owner_path, member_literal_sites, {}) - -def scan_patterns(name: str) -> tuple[list[str], list[str]]: - body, owner_path, _module, _consumers = PATTERN_DECISIONS[name] - declared = DECLARED_EMBEDDED_BODIES.get(body, {}) - return _scan(name, "body", body, owner_path, lambda s, v: body_literal_sites(s, v), declared) - - -def _scan(name, kind, payload, owner_path, value_sites, declared) -> tuple[list[str], list[str]]: - bindings: list[str] = [] - per_file: dict[str, list[int]] = {} +def _repository_scan(): + queries = [(name, "members", spec[0], spec[1], {}) + for name, spec in SET_DECISIONS.items()] + queries += [(name, "body", spec[0], spec[1], DECLARED_EMBEDDED_BODIES.get(spec[0], {})) + for name, spec in PATTERN_DECISIONS.items()] + bindings = {name: [] for name, *_ in queries} + sites = {name: {} for name, *_ in queries} for path in sorted(PACKAGE_ROOT.rglob("*.py")): rel = path.relative_to(REPO_ROOT).as_posix() - source = path.read_text(encoding="utf-8") - if rel != owner_path: - for lineno in binding_sites(source, name, kind, payload): - bindings.append(f"{rel}:{lineno} rebinds {name} to the same {kind}") - for lineno in value_sites(source, payload): - per_file.setdefault(rel, []).append(lineno) - unexplained = [ - f"{rel}:{lineno} states the same {kind} inline" - for rel, lines in per_file.items() - if rel not in declared - for lineno in lines - ] - for rel, count in declared.items(): - seen = len(per_file.get(rel, [])) - if seen != count: - unexplained.append(f"{rel} declares {count} embedded site(s), found {seen}") - return bindings, unexplained + analysis = _analyse(path.read_text(encoding="utf-8")) + for name, kind, payload, owner_path, _declared in queries: + if rel == owner_path: + continue + named, inline = _sites(analysis, name, kind, payload) + bindings[name].extend(f"{rel}:{line} rebinds {name} to the same {kind}" + for line in named) + if inline: + sites[name][rel] = inline + result = {} + for name, kind, _payload, _owner, declared in queries: + unexplained = [f"{rel}:{line} states the same {kind} inline" + for rel, lines in sites[name].items() if rel not in declared + for line in lines] + for rel, count in declared.items(): + seen = len(sites[name].get(rel, [])) + if seen != count: + unexplained.append(f"{rel} declares {count} embedded site(s), found {seen}") + result[name] = bindings[name], unexplained + return result + + +def _repository_reader(): + # Source is fixed for a module's tests. Retain only a complete findings view; + # failures leave the reader empty so a repaired source can be retried. + result = None + + def read(name): + nonlocal result + if result is None: + result = _repository_scan() + bindings, inline = result[name] + return list(bindings), list(inline) + + return read + + +@pytest.fixture(scope="module") +def repository_findings(): + return _repository_reader() @pytest.mark.parametrize("name", sorted(SET_DECISIONS)) -def test_only_the_owner_module_binds_each_field_vocabulary(name: str) -> None: - bindings, inline = scan_sets(name) +def test_only_the_owner_module_binds_each_field_vocabulary(name: str, repository_findings) -> None: + bindings, inline = repository_findings(name) assert bindings == [], "\n".join(bindings) assert inline == [], f"{name} is still spelled out somewhere: " + "\n".join(inline) @pytest.mark.parametrize("name", sorted(PATTERN_DECISIONS)) -def test_only_the_owner_module_compiles_each_pattern(name: str) -> None: - bindings, embedded = scan_patterns(name) +def test_only_the_owner_module_compiles_each_pattern(name: str, repository_findings) -> None: + bindings, embedded = repository_findings(name) assert bindings == [], "\n".join(bindings) assert embedded == [], f"{name} body appears undeclared: " + "\n".join(embedded) @@ -412,3 +441,39 @@ def test_the_two_scan_layers_see_exactly_what_they_should(label: str, source: st assert hits_body, label else: assert not hits_members and not hits_body, label + + +@pytest.mark.parametrize("source,owner,named,inline", [ + ('_SOURCE_FIELDS = ["source_kind", "source_ref", "actor_ref", "actor_role"]\n', False, 1, 1), + ('ALLOWED = ["source_kind", "source_ref", "actor_ref", "actor_role"]\n', False, 0, 1), + ('_SOURCE_FIELDS = ["other"]\n', False, 0, 0), + ('_SOURCE_FIELDS = ["source_kind", "source_ref", "actor_ref", "actor_role"]\n', True, 0, 0), +]) +def test_repository_findings_preserve_refusals_and_retry(tmp_path, monkeypatch, source, owner, named, inline): + monkeypatch.setitem(globals(), "REPO_ROOT", tmp_path) + monkeypatch.setitem(globals(), "PACKAGE_ROOT", tmp_path / "loopx") + declared = tmp_path / "loopx/capabilities/issue_fix/pr_description.py" + declared.parent.mkdir(parents=True) + body = PATTERN_DECISIONS["_REPO_PATTERN"][0] + declared.write_text("\n".join(f"P{index} = {body!r}" for index in range(3))) + path = tmp_path / (SET_DECISIONS["_SOURCE_FIELDS"][1] if owner else "loopx/extra.py") + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text("BROKEN = (\n") + read = _repository_reader() + with pytest.raises(SyntaxError): + read("_SOURCE_FIELDS") + path.write_text(source) + bindings, sites = read("_SOURCE_FIELDS") + assert len(bindings) == named + assert len(sites) == inline + bindings.clear() + sites.clear() + assert tuple(map(len, read("_SOURCE_FIELDS"))) == (named, inline) + # Other decisions and the declared embedded sites are part of this same scan. + assert read("_REPO_PATTERN") == ([], []) + if owner: + declared.write_text("ONE = " + repr(body)) + _, rejected = _repository_reader()("_REPO_PATTERN") + assert rejected == [ + "loopx/capabilities/issue_fix/pr_description.py declares 3 embedded site(s), found 1" + ] diff --git a/tests/architecture/test_wire_schema_version_owners.py b/tests/architecture/test_wire_schema_version_owners.py index 7e8c9ac0f0..e83a7b379d 100644 --- a/tests/architecture/test_wire_schema_version_owners.py +++ b/tests/architecture/test_wire_schema_version_owners.py @@ -133,6 +133,33 @@ def module_source(rel_path: str) -> str: return (REPO_ROOT / rel_path).read_text(encoding="utf-8") +def _analyse(source: str, names: frozenset[str]): + """Derive named bindings and string sites without retaining syntax trees.""" + + tree = ast.parse(source) + bindings = [] + for node in tree.body: + if isinstance(node, ast.Assign): + targets = frozenset(t.id for t in node.targets if isinstance(t, ast.Name)) + if targets & names: + bindings.append((node.lineno, targets, ast.unparse(node.value))) + strings = [(node.lineno, node.value) for node in ast.walk(tree) + if isinstance(node, ast.Constant) and isinstance(node.value, str)] + return bindings, strings + + +def _binding_lines(bindings, name: str, value: str | None): + return [line for line, names, expression in bindings + if name in names and (value is None or expression == repr(value))] + + +def _sites(analysis, name: str, value: str | None): + bindings, strings = analysis + named = _binding_lines(bindings, name, value) + inline = [] if value is None else [line for line, text in strings if value in text] + return named, inline + + def module_level_binds(source: str, name: str, value: str | None = None) -> list[int]: """Lines where this module restates the owner's *value* under this name. @@ -143,16 +170,7 @@ def module_level_binds(source: str, name: str, value: str | None = None) -> list name-only view used by the neighbour probes. """ - tree = ast.parse(source) - found: list[int] = [] - for node in tree.body: - if not isinstance(node, ast.Assign): - continue - if not any(isinstance(target, ast.Name) and target.id == name for target in node.targets): - continue - if value is None or ast.unparse(node.value) == repr(value): - found.append(node.lineno) - return found + return _binding_lines(_analyse(source, frozenset({name}))[0], name, value) def literal_sites(source: str, value: str) -> list[int]: @@ -163,53 +181,65 @@ def literal_sites(source: str, value: str) -> list[int]: into an f-string, and an f-string interpolation is not a literal at all. """ - tree = ast.parse(source) - return [ - node.lineno - for node in ast.walk(tree) - if isinstance(node, ast.Constant) and isinstance(node.value, str) and value in node.value - ] - + return _sites(_analyse(source, frozenset()), "", value)[1] -def scan(name: str) -> tuple[list[str], list[str]]: - """Return (undefined-owner findings, unexplained literal findings) for one decision.""" - owner_path, _, value, _ = DECISIONS[name] - declared = DECLARED_LITERAL_SITES.get(value, {}) - bindings_found: list[str] = [] - per_file: dict[str, list[int]] = {} +def _repository_scan(): + bindings = {name: [] for name in DECISIONS} + sites = {name: {} for name in DECISIONS} + names = frozenset(DECISIONS) for path in sorted(PACKAGE_ROOT.rglob("*.py")): rel = path.relative_to(REPO_ROOT).as_posix() - if rel == owner_path: - continue - source = path.read_text(encoding="utf-8") - for lineno in module_level_binds(source, name, value): - bindings_found.append(f"{rel}:{lineno} restates {name}={value!r}") - lines = literal_sites(source, value) - if lines: - per_file[rel] = lines - unexplained = [ - f"{rel}:{lineno} spells {value!r}" - for rel, lines in per_file.items() - if rel not in declared - for lineno in lines - ] - for rel, count in declared.items(): - seen = len(per_file.get(rel, [])) - if seen != count: - unexplained.append(f"{rel} declares {count} site(s), found {seen}") - return bindings_found, unexplained + analysis = _analyse(path.read_text(encoding="utf-8"), names) + for name, (owner_path, _module, value, _consumers) in DECISIONS.items(): + if rel == owner_path: + continue + named, inline = _sites(analysis, name, value) + bindings[name].extend(f"{rel}:{line} restates {name}={value!r}" for line in named) + if inline: + sites[name][rel] = inline + result = {} + for name, (_owner, _module, value, _consumers) in DECISIONS.items(): + declared = DECLARED_LITERAL_SITES.get(value, {}) + unexplained = [f"{rel}:{line} spells {value!r}" + for rel, lines in sites[name].items() if rel not in declared + for line in lines] + for rel, count in declared.items(): + seen = len(sites[name].get(rel, [])) + if seen != count: + unexplained.append(f"{rel} declares {count} site(s), found {seen}") + result[name] = bindings[name], unexplained + return result + + +def _repository_reader(): + # Cache only the completed module-test findings, never ASTs or raw source. + result = None + + def read(name): + nonlocal result + if result is None: + result = _repository_scan() + bindings, inline = result[name] + return list(bindings), list(inline) + + return read + + +@pytest.fixture(scope="module") +def repository_findings(): + return _repository_reader() @pytest.mark.parametrize("name", sorted(DECISIONS)) -def test_only_the_owner_module_defines_the_name(name: str) -> None: - bindings_found, _ = scan(name) +def test_only_the_owner_module_defines_the_name(name: str, repository_findings) -> None: + bindings_found, _ = repository_findings(name) assert bindings_found == [], "\n".join(bindings_found) @pytest.mark.parametrize("name", sorted(DECISIONS)) -def test_no_other_module_spells_the_value_out(name: str) -> None: - _, literals = scan(name) +def test_no_other_module_spells_the_value_out(name: str, repository_findings) -> None: + _, literals = repository_findings(name) assert literals == [], "\n".join(literals) @@ -401,3 +431,37 @@ def test_the_scan_leaves_a_folded_consumer_alone(label: str, source: str) -> Non assert [ name for name, (_o, _m, folded, _c) in DECISIONS.items() if literal_sites(source, folded) ] == [], label + + +@pytest.mark.parametrize("source,owner,named,inline", [ + ('REQUEST_SCHEMA = "semantic_preference_provider_request_v0"\n', False, 1, 1), + ('help = "Public-safe semantic_preference_provider_request_v0 here"\n', False, 0, 1), + ('REQUEST_SCHEMA = "a_different_document_v0"\n', False, 0, 0), + ('REQUEST_SCHEMA = "semantic_preference_provider_request_v0"\n', True, 0, 0), +]) +def test_repository_findings_preserve_refusals_and_retry(tmp_path, monkeypatch, source, owner, named, inline): + monkeypatch.setitem(globals(), "REPO_ROOT", tmp_path) + monkeypatch.setitem(globals(), "PACKAGE_ROOT", tmp_path / "loopx") + declared = tmp_path / "loopx/capabilities/reward_memory/registry.py" + declared.parent.mkdir(parents=True) + declared.write_text(f"SCHEMA = {DECISIONS['RESPONSE_SCHEMA'][2]!r}\n") + path = tmp_path / (DECISIONS["REQUEST_SCHEMA"][0] if owner else "loopx/extra.py") + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text("BROKEN = (\n") + read = _repository_reader() + with pytest.raises(SyntaxError): + read("REQUEST_SCHEMA") + path.write_text(source) + bindings, sites = read("REQUEST_SCHEMA") + assert len(bindings) == named + assert len(sites) == inline + bindings.clear() + sites.clear() + assert tuple(map(len, read("REQUEST_SCHEMA"))) == (named, inline) + assert read("RESPONSE_SCHEMA") == ([], []) + if owner: + declared.write_text("") + _, rejected = _repository_reader()("RESPONSE_SCHEMA") + assert rejected == [ + "loopx/capabilities/reward_memory/registry.py declares 1 site(s), found 0" + ] diff --git a/tests/control_plane/test_canonical_lease_inspection.py b/tests/control_plane/test_canonical_lease_inspection.py index ba53566605..88ca782337 100644 --- a/tests/control_plane/test_canonical_lease_inspection.py +++ b/tests/control_plane/test_canonical_lease_inspection.py @@ -186,10 +186,10 @@ def test_inactive_legacy_inspection_does_not_project_todos(tmp_path, monkeypatch state = tmp_path / "state.md" state.write_text("---\nhandoff_mode: hard_lease\n---\n# Synthetic state\n") registry.write_text(json.dumps({"goals": [{"id": GOAL, "repo": str(tmp_path), "state_file": str(state)}]})) - import loopx.todos + from loopx.control_plane.todos import list_readback def forbidden_projection(**_): raise AssertionError("inactive inspection must not parse Todo history") - monkeypatch.setattr(loopx.todos, "list_goal_todos", forbidden_projection) + monkeypatch.setattr(list_readback, "list_goal_todos", forbidden_projection) runtime = tmp_path / "runtime" missing = _inspect(registry, runtime) assert missing["active"] is False and missing["lease"] is None @@ -211,13 +211,13 @@ def test_legacy_inspection_reloads_lease_after_demanding_todo_facts(tmp_path, mo lease_path.parent.mkdir(parents=True) lease = {"schema_version": "task_lease_v0", "status": "active", "owner": "agent-a", "expires_at": "2099-01-01T00:00:00Z"} lease_path.write_text(json.dumps(lease)) - import loopx.todos + from loopx.control_plane.todos import list_readback calls = [] def release_during_projection(**_): calls.append(True) lease_path.write_text(json.dumps({**lease, "status": "released"})) return {"todos": [{"todo_id": TODO, "status": "open", "claimed_by": None, "excluded_agents": []}]} - monkeypatch.setattr(loopx.todos, "list_goal_todos", release_during_projection) + monkeypatch.setattr(list_readback, "list_goal_todos", release_during_projection) result = _inspect(registry, runtime) assert len(calls) == 1 assert result["active"] is False diff --git a/tests/control_plane/test_canonical_todo_writer_isolation.py b/tests/control_plane/test_canonical_todo_writer_isolation.py index 546abad098..5fc6cd690a 100644 --- a/tests/control_plane/test_canonical_todo_writer_isolation.py +++ b/tests/control_plane/test_canonical_todo_writer_isolation.py @@ -363,11 +363,13 @@ def bootstrap(): @pytest.mark.parametrize("provider", ["file", "sqlite"]) def test_leased_delivery_settlement_and_recovery_without_source_todo_writers(tmp_path, provider, without_source_todo_writers): - project, runtime, registry, display, _ = _source( + project, runtime, registry, display, prefix = _source( tmp_path, provider=provider, handoff_mode="hard_lease", extra=f"claimed_by={settlement.AGENT_ID}", ) - display.unlink() + # The isolated package removes the writers; erase only the Todo display, + # not the Goal objective and constraints needed for delivery admission. + display.write_text(prefix) def run(*args, succeeds=True): result = subprocess.run( diff --git a/tests/control_plane/test_quota_authority_settlement_journey.py b/tests/control_plane/test_quota_authority_settlement_journey.py index 6a02a1ef84..8810789fb9 100644 --- a/tests/control_plane/test_quota_authority_settlement_journey.py +++ b/tests/control_plane/test_quota_authority_settlement_journey.py @@ -78,9 +78,11 @@ def test_stale_markdown_cannot_admit_terminal_or_blocked_work(tmp_path, provider @pytest.mark.parametrize("provider", ["legacy", "file", "sqlite"]) def test_exact_selection_reaches_work_beyond_display_limits(tmp_path, provider): - project, runtime, registry, path, _ = _source(tmp_path, provider=provider) + project, runtime, registry, path, prefix = _source(tmp_path, provider=provider) if provider != "legacy": - path.unlink() # Display is not a prerequisite for promoted admission. + # Canonical Todo admission does not need displayed rows. Goal intent + # and constraints remain required independently of the Todo provider. + path.write_text(prefix) code, guard = _guard(project, runtime, registry) assert code == 0, guard assert guard["decision"] == "run" @@ -88,6 +90,27 @@ def test_exact_selection_reaches_work_beyond_display_limits(tmp_path, provider): assert guard["heartbeat_receipt"]["settlement_identity"]["todo_id"] == cli.TODO_ID +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_canonical_todos_cannot_replace_missing_goal_intent(tmp_path, provider): + project, runtime, registry, path, _ = _source(tmp_path, provider=provider) + before = list_goal_todos(registry_path=registry, goal_id=cli.GOAL_ID, + runtime_root_arg=str(runtime), todo_id=cli.TODO_ID) + path.unlink() + # A read-only canonical Todo lookup still works; it cannot authorize a + # delivery whose Goal objective and constraints are unavailable. + assert list_goal_todos(registry_path=registry, goal_id=cli.GOAL_ID, + runtime_root_arg=str(runtime), todo_id=cli.TODO_ID) == before + code, guard = _guard(project, runtime, registry) + assert code == 1 + assert guard["should_run"] is False + assert not guard.get("selected_todo") + assert cli._heartbeat_receipt_count(runtime, cli.TURN_ID) == 0 + assert cli._spend_run_count(runtime) == 0 + assert not path.exists() + assert list_goal_todos(registry_path=registry, goal_id=cli.GOAL_ID, + runtime_root_arg=str(runtime), todo_id=cli.TODO_ID) == before + + @pytest.mark.parametrize("provider", ["file", "sqlite"]) @pytest.mark.parametrize("conflicting_target", [False, True]) def test_canonical_decision_gate_cannot_be_bypassed_by_exact_selection( @@ -245,7 +268,7 @@ def test_semantic_progress_replay_keeps_original_turn_after_todo_wait(tmp_path, @pytest.mark.parametrize("provider", ["file", "sqlite"]) def test_migrated_hard_lease_can_write_settle_retry_and_continue(tmp_path, provider): - project, runtime, registry, path, _ = _source( + project, runtime, registry, path, prefix = _source( tmp_path, provider=provider, handoff_mode="legacy", extra=f"claimed_by={cli.AGENT_ID}", ) @@ -263,7 +286,8 @@ def run(*args): assert code == 0 and migrated["status"] == "applied", migrated assert Path(migrated["backup_path"]).is_file() assert migrated["execution_authority_granted"] is False - path.unlink() # The migrated journey cannot use the old Markdown writer. + # Remove Todo display while retaining the independent Goal frame. + path.write_text(prefix) code, guard = _guard(project, runtime, registry) assert code == 0 and guard["decision"] == "run", guard diff --git a/tests/control_plane/test_quota_settlement_cli.py b/tests/control_plane/test_quota_settlement_cli.py index 792582f8f6..c8ed2eb087 100644 --- a/tests/control_plane/test_quota_settlement_cli.py +++ b/tests/control_plane/test_quota_settlement_cli.py @@ -2697,8 +2697,13 @@ def test_prior_host_closeout_reads_archived_exact_todo( def test_standard_codex_app_settlement_is_receipted_and_idempotent( - tmp_path: Path, + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: + from examples.control_plane.quota_plan_fixtures import write_codex_heartbeat_fixture + + codex_home = tmp_path / "codex-home" + monkeypatch.setenv("CODEX_HOME", str(codex_home)) + monkeypatch.setenv("CODEX_THREAD_ID", "scheduler-fixture") project, runtime, registry_path = _write_fixture(tmp_path) _configure_read_only_todo(project) binding = ( @@ -2886,6 +2891,12 @@ def test_standard_codex_app_settlement_is_receipted_and_idempotent( TURN_ID, "--execute", ] + # Matching Host facts cannot revive the superseded receipt; the fresh + # Turn below must still acknowledge this real isolated host observation. + write_codex_heartbeat_fixture( + codex_home, goal_id=GOAL_ID, agent_id=AGENT_ID, + rrule="FREQ=MINUTELY;INTERVAL=3", + ) ack_rc, ack = _run_cli( registry_path, runtime, @@ -5589,8 +5600,13 @@ def test_pending_action_selection_does_not_commit_after_new_user_gate( @pytest.mark.parametrize("write_vision", [True, False], ids=["qualified-vision", "missing-vision"]) def test_todoless_autonomous_replan_settles_quota_refresh_spend_chain( - tmp_path: Path, write_vision: bool, + tmp_path: Path, write_vision: bool, monkeypatch: pytest.MonkeyPatch, ) -> None: + from examples.control_plane.quota_plan_fixtures import write_codex_heartbeat_fixture + + codex_home = tmp_path / "codex-home" + monkeypatch.setenv("CODEX_HOME", str(codex_home)) + monkeypatch.setenv("CODEX_THREAD_ID", "scheduler-fixture") project, runtime, registry_path = _write_fixture(tmp_path) _configure_autonomous_replan_fixture(project, runtime, registry_path) registry = json.loads(registry_path.read_text(encoding="utf-8")) @@ -5761,6 +5777,28 @@ def test_todoless_autonomous_replan_settles_quota_refresh_spend_chain( ] == "autonomous_replan" assert _spend_run_count(runtime) == 1 + # A settled receipt retains the original recipe, not authority to invent + # current Host facts. Missing and different cadences cannot mutate state. + for observed in (None, "FREQ=MINUTELY;INTERVAL=6"): + if observed is not None: + write_codex_heartbeat_fixture( + codex_home, goal_id=GOAL_ID, agent_id=AGENT_ID, rrule=observed, + ) + refused_rc, refused = _run_cli( + registry_path, runtime, *original_scheduler_ack_args, + ) + assert refused_rc == 1, refused + assert refused["scheduler_commit"]["reason_code"] == "rrule_mismatch" + assert refused["scheduler_commit"]["written"] is False + assert not Path(refused["scheduler_commit"]["path"]).exists() + assert _spend_run_count(runtime) == 1 + + # Simulate the host update through the existing isolated manifest fixture; + # command-line hints alone do not prove the actual Codex Host readback. + write_codex_heartbeat_fixture( + codex_home, goal_id=GOAL_ID, agent_id=AGENT_ID, + rrule="FREQ=MINUTELY;INTERVAL=3", + ) ack_rc, ack = _run_cli( registry_path, runtime, diff --git a/tests/control_plane/test_retired_todo_event_source.py b/tests/control_plane/test_retired_todo_event_source.py index c2ac09760d..d1d74043f8 100644 --- a/tests/control_plane/test_retired_todo_event_source.py +++ b/tests/control_plane/test_retired_todo_event_source.py @@ -234,7 +234,9 @@ def readback(expected, lease_status=None, *, agent="agent-a", target=None): payload = cli("status") indexed = next(row for row in payload["todo_index"]["items"] if row["todo_id"] == target) assert indexed["status"] == expected - directory = cli("agent-directory", "--agent-id", agent) + # This journey owns the synthetic project, not a repeated boundary scan + # of the checkout that happens to be the CLI harness's working directory. + directory = cli("agent-directory", "--agent-id", agent, "--scan-path", str(tmp_path)) work = next(row for row in directory["rows"] if row["agent_id"] == agent)["work"] if expected == "done": assert work is None @@ -280,8 +282,11 @@ def readback(expected, lease_status=None, *, agent="agent-a", target=None): unavailable = runtime / "unavailable-provider" authority.rename(unavailable) for command in ("status", "agent-directory"): - failure = subprocess.run([sys.executable, "-m", "loopx.cli", "--registry", str(registry), - "--runtime-root", str(runtime), "--format", "json", command, "--goal-id", "goal-a"], + arguments = [sys.executable, "-m", "loopx.cli", "--registry", str(registry), + "--runtime-root", str(runtime), "--format", "json", command, "--goal-id", "goal-a"] + if command == "agent-directory": + arguments.extend(["--scan-path", str(tmp_path)]) + failure = subprocess.run(arguments, capture_output=True, text=True, timeout=60) if failure.stdout.strip(): failed = json.loads(failure.stdout) diff --git a/tests/control_plane/test_validated_create_recovery.py b/tests/control_plane/test_validated_create_recovery.py index 9cd184cfdb..a89b9f7b58 100644 --- a/tests/control_plane/test_validated_create_recovery.py +++ b/tests/control_plane/test_validated_create_recovery.py @@ -216,11 +216,25 @@ def lose_response(method, request, **kwargs): real(method, request, **kwargs) raise EffectRuntimeResponseAmbiguous(method, timeout=1) monkeypatch.setattr(provider_update, "effect_runtime_result", lose_response) - with pytest.raises(EffectRuntimeResponseAmbiguous): + with pytest.raises(LocalCoordinationAuthorityUnavailable) as error: update_goal_todo(registry_path=registry, goal_id="goal-a", todo_id=created["todo_id"], role="agent", agent_id="agent-a", update_operation_id="revise-after-loss", update_expected_provider_revision=read(runtime)["provider_revision"], validation_command_json=json.dumps([sys.executable, "-c", "raise SystemExit(3)"])) + assert error.value.code == "runtime_response_ambiguous" + assert error.value.payload["todo_id"] == created["todo_id"] + assert error.value.payload["operation_id"] == "revise-after-loss" + # The public boundary supplies a read of the original operation, never a + # new write or a claim that its missing response proves writer shutdown. + command = error.value.payload["recovery_command"] + assert command[0] == "loopx" + committed = read(runtime) + recovered = subprocess.run([sys.executable, "-m", "loopx.cli", *command[1:]], + capture_output=True, text=True, timeout=60) + assert recovered.returncode == 0, recovered.stdout + recovered.stderr + receipt = json.loads(recovered.stdout) + assert receipt["status"] == "found" and receipt["operation_id"] == "revise-after-loss" + assert read(runtime) == committed project(registry, runtime) canonical = read(runtime)["todos"][0] stored = read_completion_validation_declaration(runtime_root=runtime, goal_id="goal-a", diff --git a/tests/test_goal_attention.py b/tests/test_goal_attention.py index 18a3f83ce8..6aecc3f4e4 100644 --- a/tests/test_goal_attention.py +++ b/tests/test_goal_attention.py @@ -17,7 +17,7 @@ def test_steward_turn_receives_blocker_and_decision_without_channel(tmp_path, mo blocker = {"todo_id": "todo_work", "role": "agent", "task_class": "advancement_task", "status": "blocked", "text": "Verify the primary result", "reason": "Required input has not arrived"} - monkeypatch.setattr("loopx.todos.list_goal_todos", lambda **_: { + monkeypatch.setattr("loopx.control_plane.todos.list_readback.list_goal_todos", lambda **_: { "ok": True, "todos": [gate, blocker], "source": "canonical_fixture", }) monkeypatch.setattr(portfolio, "collect_status", lambda **_: { diff --git a/tests/test_sonarcloud_workflow.py b/tests/test_sonarcloud_workflow.py index bfdca6eeb2..2c2d026861 100644 --- a/tests/test_sonarcloud_workflow.py +++ b/tests/test_sonarcloud_workflow.py @@ -2,6 +2,7 @@ import re from pathlib import Path +from types import SimpleNamespace import pytest import yaml @@ -61,23 +62,66 @@ def test_each_missing_analysis_guard_is_rejected() -> None: _assert_token_guards(mutant) -def test_sonar_reuses_same_run_coverage_without_a_privileged_trigger() -> None: +def _assert_analysis_boundary(workflow: dict, caller: dict) -> None: + # A trusted metadata-only cleanup trigger belongs to the caller, not to + # Sonar. Analysis must still depend on successful coverage in this run. + assert set(workflow["on"]) == {"workflow_call"} + sonar = caller["jobs"]["sonar"] + assert sonar["needs"] == "pytest" + assert sonar["uses"] == "./.github/workflows/sonarcloud.yml" + # Without a status function GitHub retains the default success() check. + assert sonar.get("if") == "github.event_name != 'merge_group'" + for action in ("closed", "synchronize"): + github = SimpleNamespace(event_name="pull_request_target", + event=SimpleNamespace(action=action)) + context = {"github": github, "always": lambda: True} + + def condition(expression: str) -> bool: + # These two shipped conditions use string comparisons and Boolean + # operators only, as in the existing executable CI gate tests. + return bool(eval(expression.replace("&&", "and").replace("||", "or"), + {"__builtins__": {}}, context)) + + classified = condition(caller["jobs"]["changes"]["if"]) + # If classification were reachable, assume its most permissive output. + # Skipped jobs cannot supply that output to this run's dependencies. + context["needs"] = SimpleNamespace(changes=SimpleNamespace( + outputs=SimpleNamespace(python_tests="true" if classified else ""))) + assert not condition(caller["jobs"]["pytest"]["if"]) + + +def test_sonar_reuses_same_run_coverage_and_isolated_trigger() -> None: workflow = WORKFLOW.read_text(encoding="utf-8") caller = WORKFLOW.with_name("python-tests.yml").read_text(encoding="utf-8") - assert "workflow_call:" in workflow - assert "workflow_run:" not in workflow - assert "pull_request_target:" not in workflow + caller + _assert_analysis_boundary(yaml.load(workflow, Loader=yaml.BaseLoader), + yaml.load(caller, Loader=yaml.BaseLoader)) assert "python -m pytest" not in workflow assert "name: python-coverage-xml" in workflow assert "run-id:" not in workflow assert "github-token:" not in workflow - sonar = yaml.safe_load(caller)["jobs"]["sonar"] - # Coverage comes from this run's pytest job, handed to the local reusable - # workflow. The only condition may skip merge-queue refs; a status function - # such as always() would let analysis run without that coverage. - assert sonar["needs"] == "pytest" - assert sonar["uses"] == "./.github/workflows/sonarcloud.yml" - assert sonar.get("if") == "github.event_name != 'merge_group'" assert '"apps/**"' in caller assert '"sonar-project.properties"' in caller + + +@pytest.mark.parametrize("unsafe", ["trigger", "classification", "coverage", "status"]) +def test_reachable_privileged_analysis_or_coverage_bypass_is_rejected(unsafe) -> None: + workflow = yaml.load(WORKFLOW.read_text(), Loader=yaml.BaseLoader) + caller = yaml.load(WORKFLOW.with_name("python-tests.yml").read_text(), Loader=yaml.BaseLoader) + if unsafe == "trigger": + workflow["on"]["pull_request_target"] = {} + elif unsafe == "classification": + caller["jobs"]["changes"]["if"] = "always()" + elif unsafe == "coverage": + caller["jobs"]["pytest"]["if"] = "always()" + else: + caller["jobs"]["sonar"]["if"] = "always()" + with pytest.raises(AssertionError): + _assert_analysis_boundary(workflow, caller) + + +def test_trusted_cleanup_event_extension_does_not_activate_analysis() -> None: + workflow = yaml.load(WORKFLOW.read_text(), Loader=yaml.BaseLoader) + caller = yaml.load(WORKFLOW.with_name("python-tests.yml").read_text(), Loader=yaml.BaseLoader) + caller["on"]["pull_request_target"]["types"] = ["closed", "synchronize"] + _assert_analysis_boundary(workflow, caller) diff --git a/tests/test_state_refresh_projections.py b/tests/test_state_refresh_projections.py index b411ac344d..8c535e2d13 100644 --- a/tests/test_state_refresh_projections.py +++ b/tests/test_state_refresh_projections.py @@ -70,7 +70,7 @@ def test_refresh_output_projections_preserve_optional_run_contracts() -> None: "frontmatter": {"updated_at": "2026-07-30T23:59:00+00:00"}, } assert index_record["requested_classification"] == "repair_completed" - assert index_record["agent_vision"] == { + expected_vision = { key: agent_vision[key] for key in ( "schema_version", @@ -83,6 +83,10 @@ def test_refresh_output_projections_preserve_optional_run_contracts() -> None: "path_delta", ) } + # The run-index contract carries an explicit null for the optional, + # default-off stride decision. All original fields remain unchanged. + expected_vision["stride_decision"] = None + assert index_record["agent_vision"] == expected_vision assert "validation" not in index_record["agent_vision"] for field in ( "delivery_batch_scale",