diff --git a/.github/workflows/python-tests.yml b/.github/workflows/python-tests.yml index a88a82470f..a4c4f703ba 100644 --- a/.github/workflows/python-tests.yml +++ b/.github/workflows/python-tests.yml @@ -423,7 +423,10 @@ jobs: needs: [changes, chat-bundle] if: needs.changes.outputs.python_tests == 'true' runs-on: ubuntu-latest - timeout-minutes: 45 + # All four shards were still progressing at 83-94% when the 45m job + # deadline cancelled run 37916979857. Reserve setup/reporting headroom + # without changing test-level deadlines, collection or worker count. + timeout-minutes: 60 strategy: fail-fast: false matrix: @@ -472,7 +475,7 @@ jobs: # Without timing history least_duration alternates equal-weight tests. # Each runner retains the measured two-worker pool. run: >- - python -m pytest -q -n 2 -m "not stage2c_e2e" + python -m pytest -v -n 2 -m "not stage2c_e2e" --splits 4 --group ${{ matrix.shard }} --splitting-algorithm least_duration --durations=25 --durations-min=1 diff --git a/docs/development/testing-and-quality.md b/docs/development/testing-and-quality.md index 1a54154e5f..8b58b8673e 100644 --- a/docs/development/testing-and-quality.md +++ b/docs/development/testing-and-quality.md @@ -555,6 +555,9 @@ Lint, type checks, and the CLI budget run separately. The required `pytest` check rejects failed/skipped shards and missing coverage artifacts, then uses `coverage combine` to enforce the existing 19.6% floor on the union, not on individual shards. Relative coverage paths make reports portable across runners. +Each shard has a 60-minute job ceiling for setup, the full partition and report +upload; individual test deadlines stay unchanged. Verbose progress names failed +cases before the suite finishes, so an interrupted run retains their identities. The reusable Sonar workflow consumes that same run's XML and never reruns pytest or reads cross-run artifacts. Missing Sonar tokens still skip analysis successfully; test jobs receive no Sonar secret. The trigger is the union of @@ -565,11 +568,13 @@ Linux 全套测试分到四台 hosted runner，每台保留两个 xdist worker 按完整 collection 分片；没有历史耗时时，等权测试交替分配。lint、类型检查和 CLI 预算独立执行。必需的 `pytest` 汇总检查会拒绝失败／跳过的分片和缺失的 coverage， 合并后再执行原有 19.6% 门槛；不要求单个分片达到全套覆盖率。coverage 使用相对路径， +每个分片的作业上限为 60 分钟，覆盖环境准备、完整分片和报告上传；单项测试超时 +保持不变。详细进度在套件结束前就记录失败用例名称，运行中断后仍可定位这些用例。 Sonar 只复用同一次 run 的 XML，不重复测试、不跨 run 取产物。缺少 token 仍成功跳过 Sonar，测试 job 不接收 Sonar secret。触发范围取原有两套 workflow 的并集；纯前端 PR 使用前述豁免，Sonar 配置变更仍全量运行，包括没有 token 的 fork。 -Reproduce one shard locally with `uv run --extra test python -m pytest -q -n 2 --splits 4 --group 1 +Reproduce one shard locally with `uv run --extra test python -m pytest -v -n 2 --splits 4 --group 1 --splitting-algorithm least_duration --cov=loopx`. Omit the split arguments to run the complete suite locally. 全量本地测试仍省略分片参数即可。 diff --git a/examples/shared-goal-authority-e2e/mutants.py b/examples/shared-goal-authority-e2e/mutants.py index c1b86f7a74..9efdf83dc3 100644 --- a/examples/shared-goal-authority-e2e/mutants.py +++ b/examples/shared-goal-authority-e2e/mutants.py @@ -287,8 +287,8 @@ def apply(source: str) -> str: "if False:")),), "tests/control_plane/test_shadow_drain_adversarial.py::test_state_file_override_cannot_attribute_an_unbound_source_to_active_lineage"), Case("bootstrap_unregistered_root", ((COORDINATION + "runtime_shadow.ts", replacement( - "if (registeredRoot !== request.runtime_root)", - "if (false && registeredRoot !== request.runtime_root)")),), + "if (!await sameSourceRuntimeRoot(registeredRoot, request.runtime_root))", + "if (false)")),), "tests/control_plane/test_runtime_shadow_bounded_e2e.py::test_controller_cannot_bind_the_registered_source_to_an_override_runtime_root"), Case("bootstrap_unregistered_state", ((COORDINATION + "runtime_shadow.ts", replacement( "if (resolve(String(snapshot.state_path)) !== resolve(String(snapshot.registered_state_path)))", diff --git a/loopx/capabilities/explore/README.md b/loopx/capabilities/explore/README.md index dae96268fc..69ae96cb6d 100644 --- a/loopx/capabilities/explore/README.md +++ b/loopx/capabilities/explore/README.md @@ -398,6 +398,20 @@ packet with: - the safety boundary that keeps the packet advisory rather than an replacement for `quota should-run`. +Both planners apply the existing Todo readiness rule before scheduling or +bundling current work. An open Todo with `resume_when` is selectable only when +its projected `resume_ready` is true. Unready Todos remain in rejected-candidate +diagnostics with their condition, but consume no verification width or resource +slot and carry no claim/lease suggestions. The bounded turn context preserves +these diagnostics and reports omitted candidates; read the full plan for detail. +If no branch is actionable, the plan asks the caller to inspect the conditions +and replan instead of suggesting execution. + +For `B.resume_when=todo_done:A`, handing A to another Agent does not release B; +completing A does. Successor lineage alone is not a completion dependency. Future +work remains visible for planning, and fresh quota, claim and lease checks still +govern execution after any plan, handoff or readiness change. + An advancement todo may opt into typed result diagnostics by attaching one or more explicit Explore node ids: diff --git a/loopx/capabilities/explore/todo_branch_plan.py b/loopx/capabilities/explore/todo_branch_plan.py index b1269d9616..e2fbb757cc 100644 --- a/loopx/capabilities/explore/todo_branch_plan.py +++ b/loopx/capabilities/explore/todo_branch_plan.py @@ -214,16 +214,27 @@ def _claim_bucket(item: Mapping[str, Any], *, agent_id: str | None) -> int: return 2 -def _monitor_lane_exclusion(candidate: Mapping[str, Any]) -> dict[str, Any] | None: - if candidate.get("task_class") != TODO_TASK_CLASS_MONITOR: - return None +def _candidate_readiness(item: Mapping[str, Any]) -> dict[str, Any]: + # Reuse the canonical read model; a planner does not resolve dependencies + # from ownership, lineage, or the other candidates selected in this batch. return { - **candidate, - "selection_status": "excluded_non_exploration_lane", - "exclusion_reason": "continuous_monitor_does_not_consume_exploration_budget", + "actionable_open": todo_item_is_actionable_open(dict(item)), + **{key: item[key] for key in ("status", "resume_when", "resume_ready") if key in item}, } +def _execution_exclusion(candidate: Mapping[str, Any]) -> dict[str, Any] | None: + if candidate.get("task_class") == TODO_TASK_CLASS_MONITOR: + return { + **candidate, + "selection_status": "excluded_non_exploration_lane", + "exclusion_reason": "continuous_monitor_does_not_consume_exploration_budget", + } + if not candidate["actionable_open"]: + return {**candidate, "selection_status": "not_actionable_open"} + return None + + def _lane_id(todo_id: str, *, index: int) -> str: return f"branch_{index}_{todo_id.removeprefix('todo_')}" @@ -366,6 +377,7 @@ def build_explore_todo_branch_plan( task_class = todo_item_task_class(dict(item)) required_capabilities = _required_capabilities(item) candidate = { + **_candidate_readiness(item), "todo_id": todo_id, "text": _compact_text(item.get("text") or item.get("title")), "priority": item.get("priority"), @@ -407,15 +419,16 @@ def build_explore_todo_branch_plan( ) ) - monitor_lanes = [ + excluded_candidates = [ exclusion for candidate in candidates - if (exclusion := _monitor_lane_exclusion(candidate)) is not None + if (exclusion := _execution_exclusion(candidate)) is not None ] exploration_candidates = [ candidate for candidate in candidates if candidate.get("task_class") != TODO_TASK_CLASS_MONITOR + if candidate["actionable_open"] ] schedulable = [ candidate @@ -552,7 +565,7 @@ def build_explore_todo_branch_plan( lane="todo_branch_plan", ) if not invalidated: - rejected = [*monitor_lanes, *dependency_rejections, *selection_rejections] + rejected = [*excluded_candidates, *dependency_rejections, *selection_rejections] selected = valid_selected break new_invalidated_ids = { @@ -562,7 +575,7 @@ def build_explore_todo_branch_plan( dependency_events.extend(events) invalidated_ids.update(new_invalidated_ids) if not new_invalidated_ids: - rejected = [*monitor_lanes, *dependency_rejections, *selection_rejections] + rejected = [*excluded_candidates, *dependency_rejections, *selection_rejections] selected = valid_selected break @@ -602,6 +615,7 @@ def build_explore_todo_branch_plan( "exploration_candidate_count": len(exploration_candidates), "selected_count": len(selected), "selected_branches": selected, + "rejected_candidate_count": len(rejected), "rejected_candidates": rejected[: max(0, normalized_width * 3)], "resource_portfolio": resource_portfolio_with_selection(resource_portfolio, selected), "hazard_model": { @@ -642,6 +656,9 @@ def build_explore_todo_branch_plan( "registered goal's spawn_policy (spawn_allowed, max_children) to receive " "suggested commands; execution stays in the normal LoopX lifecycle." if gate["state"] == GATE_STATE_ANALYSIS_ONLY + else "No actionable branch was selected. Inspect rejected candidates and their " + "resume conditions; replan when readiness changes." + if not selected else "Run the primary branch normally; hand selected speculative branches to side " "agents only after claiming/lease commands are accepted and workspace guards pass." if len(selected) > 1 diff --git a/loopx/capabilities/explore/worker_branch_plan.py b/loopx/capabilities/explore/worker_branch_plan.py index e3400d1599..8b1f6248e4 100644 --- a/loopx/capabilities/explore/worker_branch_plan.py +++ b/loopx/capabilities/explore/worker_branch_plan.py @@ -37,11 +37,12 @@ _branch_confidence, _branch_expected_evidence_units, _branch_score, + _candidate_readiness, _claim_command, _compact_text, _frontier_tokens, _lease_command, - _monitor_lane_exclusion, + _execution_exclusion, _required_capabilities, _required_write_scopes, _shared_dependency_capabilities, @@ -396,6 +397,7 @@ def _todo_candidate( task_class = todo_item_task_class(dict(item)) required_capabilities = _required_capabilities(item) candidate = { + **_candidate_readiness(item), "todo_id": todo_id, "text": _compact_text(item.get("text") or item.get("title")), "priority": item.get("priority"), @@ -640,21 +642,23 @@ def _build_worker_branch_candidates( ) ) - monitor_lanes = [ + excluded_candidates = [ exclusion for candidate in todo_candidates - if (exclusion := _monitor_lane_exclusion(candidate)) is not None + if (exclusion := _execution_exclusion(candidate)) is not None ] blocked = [ {**candidate, "selection_status": "blocked_claimed_by_other"} for candidate in todo_candidates if candidate.get("task_class") != TODO_TASK_CLASS_MONITOR + if candidate["actionable_open"] if candidate.get("claimed_by") and candidate.get("claimed_by") != normalized_agent ] eligible = [ candidate for candidate in todo_candidates if candidate.get("task_class") != TODO_TASK_CLASS_MONITOR + if candidate["actionable_open"] if not (candidate.get("claimed_by") and candidate.get("claimed_by") != normalized_agent) ] @@ -749,7 +753,7 @@ def _branch_sort_key(item: Mapping[str, Any]) -> tuple: ) branch_candidates.sort(key=_branch_sort_key) - return branch_candidates, [*monitor_lanes, *blocked] + return branch_candidates, [*excluded_candidates, *blocked] def _baseline_worker_lanes(branch_candidates: Sequence[Mapping[str, Any]], *, width: int) -> dict[str, Any]: @@ -1250,6 +1254,9 @@ def build_explore_worker_branch_plan( "registered goal's spawn_policy (spawn_allowed, max_children) to receive " "suggested commands; execution stays in the normal LoopX lifecycle." if gate["state"] == GATE_STATE_ANALYSIS_ONLY + else "No actionable worker branch was selected. Inspect rejected candidates and " + "their resume conditions; replan when readiness changes." + if not selected else "Use quota should-run, then execute each selected worker branch through " "normal LoopX claim, lease, todo execution, explore writeback, refresh-state, " "and quota spend-slot." diff --git a/loopx/control_plane/capabilities/explore_turn_context.ts b/loopx/control_plane/capabilities/explore_turn_context.ts index f1363abdd2..1970127a99 100644 --- a/loopx/control_plane/capabilities/explore_turn_context.ts +++ b/loopx/control_plane/capabilities/explore_turn_context.ts @@ -27,7 +27,8 @@ function evidenceCount(audit: JsonObject, kind: string, detailCount: number): nu return Math.max(detailCount, total); } function branchContext(row: JsonObject): JsonObject { - const result = compact(row, ["todo_id", "text", "branch_role", "score", "confidence"]); + const result = compact(row, ["todo_id", "text", "branch_role", "score", "confidence", + "selection_status", "status", "actionable_open", "resume_when", "resume_ready"]); if (row.typed_evidence_audit == null) return result; const audit = requireJsonObject(row.typed_evidence_audit, "Explore evidence audit"); const nodes = rows(audit.nodes), findings = rows(audit.findings), edges = rows(audit.relevant_edges); @@ -64,6 +65,7 @@ export function projectExploreTurnContext(params: JsonObject): JsonObject { String(b.last_updated_at ?? "").localeCompare(String(a.last_updated_at ?? ""))); const findings = rows(projection.findings); const branches = rows(plan.selected_branches); + const rejected = rows(plan.rejected_candidates); const frontier = rows(projection.frontier); const selectedRefs = new Set(harness ? branches.slice(0, 3).flatMap(branch => { if (branch.typed_evidence_audit == null) return []; @@ -155,6 +157,9 @@ export function projectExploreTurnContext(params: JsonObject): JsonObject { candidate_count: plan.candidate_count ?? 0, selected_branches: branches.slice(0, 3).map(branchContext), omitted_selected_branches: Math.max(0, branches.length - 3), + rejected_candidates: rejected.slice(0, 3).map(branchContext), + omitted_rejected_candidates: Math.max(0, + requireInteger(plan.rejected_candidate_count ?? rejected.length, "rejected_candidate_count") - 3), frontier: frontier.slice(0, 3).map(node => compact(node, ["node_id", "title", "status", "summary"])), omitted_frontier: Math.max(0, frontier.length - 3), // Expand the same Todo audit shown above, not a different worker-lane plan. diff --git a/tests/capabilities/test_explore_turn_context.py b/tests/capabilities/test_explore_turn_context.py index b89af5c87f..0bacfad5e8 100644 --- a/tests/capabilities/test_explore_turn_context.py +++ b/tests/capabilities/test_explore_turn_context.py @@ -543,6 +543,34 @@ def test_turn_context_audit_count_compatibility(counts, omitted): assert len(result["findings"]) == 3 +def test_turn_context_preserves_wait_conditions_and_total_omissions(): + from loopx.capabilities.explore.todo_branch_plan import build_explore_todo_branch_plan + + plan = build_explore_todo_branch_plan( + goal_id="research", agent_id="worker", width=3, + orchestration={"explore_harness": {"enabled": True}}, + todos=[{ + "todo_id": f"todo_waiting_{index}", "text": "Use the prepared artifact", + "status": "open", "task_class": "advancement_task", "priority": "P0", + "resume_when": "todo_done:todo_prepare", "resume_ready": False, + } for index in range(12)], + ) + packet = effect_runtime_result("explore.turn_context", { + "goal_id": "research", "agent_id": "worker", "route": ["loopx"], + "harness_gate": {"enabled": True}, "graph_enabled": False, + "projection": {}, "plan": plan, + }) + context = packet["harness"] + assert context["selected_branches"] == [] + assert len(context["rejected_candidates"]) == 3 + assert context["omitted_rejected_candidates"] == 9 + for row in context["rejected_candidates"]: + assert row["actionable_open"] is False + assert row["resume_ready"] is False + assert row["resume_when"] == "todo_done:todo_prepare" + assert "todo-branch-plan" in context["plan_command"] + + @pytest.mark.parametrize("kind", ["findings", "edges"]) def test_turn_context_rejects_negative_evidence_count(kind): with pytest.raises(EffectRuntimeRejected, match=f"Explore {kind} count must be nonnegative") as rejected: diff --git a/tests/capabilities/test_periodic_report_native_successor.py b/tests/capabilities/test_periodic_report_native_successor.py index d4fb487496..c3328274c9 100644 --- a/tests/capabilities/test_periodic_report_native_successor.py +++ b/tests/capabilities/test_periodic_report_native_successor.py @@ -41,7 +41,7 @@ def _native_successor( "--delivery-workspace-path", str(project), "--agent-id", AGENT_ID, "--delivery-batch-scale", "implementation", "--delivery-outcome", "outcome_progress", "--no-global-sync", - "--suppress-external-sinks", + "--suppress-external-sinks", "--available-capability", "network", ] closed = _run_cli([ *base, "--classification", "fixture_stage_closed", @@ -141,6 +141,7 @@ def test_native_successor_produces_only_an_enabled_bounded_intent( assert intent["requested_write_scope"] == [] assert intent["payload"]["generation_authorized"] is False assert intent["payload"]["external_delivery_authorized"] is False + assert intent["payload"]["available_capabilities"] == ["network"] stage = intent["payload"]["stage_completion"] assert stage["transition"] == "successor_frontier_settled" assert stage["acceptance"] == "validated" diff --git a/tests/cli_commands/test_periodic_report_intent_capability_chain.py b/tests/cli_commands/test_periodic_report_intent_capability_chain.py index ce8b04f5b9..9ad4ec67da 100644 --- a/tests/cli_commands/test_periodic_report_intent_capability_chain.py +++ b/tests/cli_commands/test_periodic_report_intent_capability_chain.py @@ -4,7 +4,6 @@ from pathlib import Path from loopx.capabilities.periodic_report.pending_intent import ( - consume_pending_periodic_report_intent, pending_periodic_report_intents, ) from settlement_capability_dispatch_fixture import ( @@ -18,13 +17,7 @@ def _write_unclaimed_frontier_state(project: Path) -> None: - """State whose progress snapshot has no agent-claimed items yet. - - The unclaimed advancement Todo keeps the successor frontier owned, so the - dispatch still freezes one stage receipt, while the durable progress - snapshot stays empty and the pending intent carries capability evidence - without a frozen project_progress projection. - """ + """Leave an active successor after the earlier stage has been superseded.""" project.joinpath("goal.md").write_text( f"""# Goal @@ -70,20 +63,7 @@ def _intent_sidecar(runtime: Path) -> dict[str, object]: return json.loads(sidecars[0].read_text(encoding="utf-8")) -def _editorial_fact_sources(registry: Path, runtime: Path) -> set[str]: - result = consume_pending_periodic_report_intent( - registry_path=registry, - runtime_root=runtime, - goal_id=GOAL_ID, - agent_id=AGENT_ID, - execute=True, - ) - assert result["status"] == "editorial_required", result - request = json.loads(Path(result["editorial_request_path"]).read_text("utf-8")) - return {str(fact.get("source_ref")) for fact in request["facts"]} - - -def test_pending_intent_fallback_uses_producer_capability_evidence( +def test_later_todo_completion_does_not_replay_superseded_successor_milestone( tmp_path: Path, ) -> None: captured, registry, runtime = complete_todo_via_cli( @@ -92,23 +72,19 @@ def test_pending_intent_fallback_uses_producer_capability_evidence( write_state=_write_unclaimed_frontier_state, ) - assert "available_capabilities" in captured + assert captured["available_capabilities"] == ["network"] intents = pending_periodic_report_intents( registry_path=registry, runtime_root=runtime, goal_id=GOAL_ID, agent_id=AGENT_ID, ) - assert len(intents) == 1 - payload = intents[0]["payload"] - assert "project_progress" not in payload - assert payload["available_capabilities"] == ["network"] - - _claim_gated_successors(registry.parent) - assert f"todo:{GATED_TODO}" in _editorial_fact_sources(registry, runtime) + assert intents == [] + sidecar = _intent_sidecar(runtime) + assert sidecar["intent"] is None -def test_pending_intent_fallback_fails_closed_without_producer_evidence( +def test_later_todo_completion_without_capabilities_does_not_replay_milestone( tmp_path: Path, ) -> None: _captured, registry, runtime = complete_todo_via_cli( @@ -117,13 +93,13 @@ def test_pending_intent_fallback_fails_closed_without_producer_evidence( write_state=_write_unclaimed_frontier_state, ) - sidecar = _intent_sidecar(runtime) - assert "project_progress" not in sidecar["intent"]["payload"] - assert "available_capabilities" not in sidecar["intent"]["payload"] - - _claim_gated_successors(registry.parent) - assert f"todo:{GATED_TODO}" not in _editorial_fact_sources(registry, runtime) - assert f"todo:{PLAIN_TODO}" in _editorial_fact_sources(registry, runtime) + assert _intent_sidecar(runtime)["intent"] is None + assert pending_periodic_report_intents( + registry_path=registry, + runtime_root=runtime, + goal_id=GOAL_ID, + agent_id=AGENT_ID, + ) == [] def test_post_writeback_frontier_and_progress_share_one_canonical_snapshot( @@ -175,8 +151,8 @@ def observe(**kwargs): agent_id=AGENT_ID, ) assert len(reads) == 1 - assert result["stage_completion"]["acceptance"] == "validated" - assert f"todo:{GATED_TODO}" in { - r["source_ref"] for r in result["project_progress"]["items"] - } + # The history contains a later active successor, while this ordinary + # post-writeback payload has no exact refresh identity. Canonical Todo + # loading still occurs once, but it cannot recover the older milestone. + assert result == {} assert not state.exists() diff --git a/tests/cli_commands/test_todo_complete_settlement_capabilities.py b/tests/cli_commands/test_todo_complete_settlement_capabilities.py index c4ce0e9554..f434eaa39c 100644 --- a/tests/cli_commands/test_todo_complete_settlement_capabilities.py +++ b/tests/cli_commands/test_todo_complete_settlement_capabilities.py @@ -49,20 +49,7 @@ def _write_stage_state(project: Path) -> None: ) -def _dispatched_next_action_source(captured: dict[str, object]) -> str: - intents = captured["post_writeback_hooks"]["intents"] - assert isinstance(intents, list) and len(intents) == 1 - items = intents[0]["payload"]["project_progress"]["items"] - matched = [ - item["source_ref"] - for item in items - if item.get("content_kind") == "next_action" - ] - assert len(matched) == 1 - return matched[0] - - -def test_todo_complete_dispatch_selects_gated_successor_with_settlement_evidence( +def test_todo_complete_does_not_replay_superseded_successor_milestone( tmp_path: Path, ) -> None: captured, _registry, _runtime = complete_todo_via_cli( @@ -72,10 +59,11 @@ def test_todo_complete_dispatch_selects_gated_successor_with_settlement_evidence ) assert captured["available_capabilities"] == ["network"] - assert _dispatched_next_action_source(captured) == f"todo:{GATED_TODO}" + assert captured["post_writeback_hooks"]["intent_count"] == 0 + assert captured["post_writeback_hooks"]["intents"] == [] -def test_todo_complete_dispatch_fails_closed_without_capability_evidence( +def test_todo_complete_without_capabilities_does_not_replay_milestone( tmp_path: Path, ) -> None: captured, _registry, _runtime = complete_todo_via_cli( @@ -85,7 +73,8 @@ def test_todo_complete_dispatch_fails_closed_without_capability_evidence( ) assert "available_capabilities" not in captured - assert _dispatched_next_action_source(captured) == f"todo:{PLAIN_TODO}" + assert captured["post_writeback_hooks"]["intent_count"] == 0 + assert captured["post_writeback_hooks"]["intents"] == [] def _write_source_registry(path: Path, runtime: Path) -> None: @@ -185,7 +174,7 @@ def hold_lifetime_lock() -> None: assert replay["changed"] is False assert replay["completion_receipt_id"] == completion_receipt_id assert Path(str(replay["state_file"])).read_bytes() == committed_state - assert replay["post_writeback_hooks"]["intent_count"] == 1 + assert replay["post_writeback_hooks"]["intent_count"] == 0 def test_todo_complete_isolates_admission_codec_failure_and_recovers_on_replay( @@ -241,7 +230,7 @@ def for_plan(**kwargs: object) -> FirstPartyHostGoalAdmission: assert replay["changed"] is False assert replay["completion_receipt_id"] == completion_receipt_id assert Path(str(replay["state_file"])).read_bytes() == committed_state - assert replay["post_writeback_hooks"]["intent_count"] == 1 + assert replay["post_writeback_hooks"]["intent_count"] == 0 def test_todo_complete_isolates_lifetime_transport_failure( diff --git a/tests/control_plane/test_canonical_dependency_execution.py b/tests/control_plane/test_canonical_dependency_execution.py index 0d334e6e3e..20c398eb53 100644 --- a/tests/control_plane/test_canonical_dependency_execution.py +++ b/tests/control_plane/test_canonical_dependency_execution.py @@ -24,6 +24,8 @@ def canonical_dependency(request, tmp_path, monkeypatch): registry.write_text(json.dumps({"common_runtime_root": str(runtime), "goals": [{ "id": GOAL, "repo": str(tmp_path), "state_file": state.name, "coordination": {"registered_agents": ["agent-a", "agent-b"]}, + "spawn_policy": {"spawn_allowed": True, "max_children": 4, + "explore_harness": {"enabled": True}}, }]})) todos = [{ "schema_version": "todo_item_v0", "todo_id": todo_id, "role": "agent", @@ -148,3 +150,56 @@ def test_wait_rejects_atomic_claim_without_changing_assignment(canonical_depende assert readback["claimed_by"] == "agent-a" and readback["resume_ready"] is False inspected = cli(["task-lease", "inspect"], WAITING) assert inspected["lease"]["status"] == "released" and inspected["lease"]["version"] == 1 + + +def test_explore_plans_follow_dependency_through_handoff_and_completion(canonical_dependency): + cli = canonical_dependency + first = _acquire(cli, WAITING, "agent-a", "waiting-original") + _arm_wait(cli, first["lease"]["version"]) + rejected = cli(["task-lease", "transfer"], WAITING, "--owner", "agent-a", + "--idempotency-key", "waiting-original", "--expected-version", "1", + "--new-owner", "agent-b", "--new-idempotency-key", "waiting-receiver", + "--ttl-seconds", "600", "--transfer-claim", expected_exit=1) + assert rejected["error_code"] == "todo_dependency_pending" + assert cli(["todo", "list"], WAITING)["todo"]["claimed_by"] == "agent-a" + cli(["task-lease", "release"], WAITING, "--owner", "agent-a", + "--idempotency-key", "waiting-original", "--expected-version", "1") + + def assert_plans(*, ready): + before = cli(["todo", "list"], WAITING)["todo"] + for command, selected_key, rejected_key in [ + ("todo-branch-plan", "selected_branches", "rejected_candidates"), + ("worker-branch-plan", "selected_worker_branches", "rejected_worker_branches"), + ]: + plan = cli(["explore", command], None, "--agent-id", "agent-a", "--scheduler-load", "0") + selected = {todo_id for row in plan[selected_key] + for todo_id in row.get("todo_ids", [row["todo_id"]])} + assert (WAITING in selected) is ready + assert FALLBACK in selected + if not ready: + waiting = next(row for row in plan[rejected_key] if row["todo_id"] == WAITING) + assert waiting["resume_ready"] is False + assert waiting["resume_when"] == f"todo_done:{PREREQUISITE}" + assert not waiting.get("suggested_commands") + context = cli(["explore", "turn-context"], None, "--agent-id", "agent-a")["harness"] + assert (WAITING in {row["todo_id"] for row in context["selected_branches"]}) is ready + if not ready: + waiting = next(row for row in context["rejected_candidates"] if row["todo_id"] == WAITING) + assert waiting["resume_ready"] is False + assert waiting["resume_when"] == f"todo_done:{PREREQUISITE}" + assert cli(["todo", "list"], WAITING)["todo"] == before + + assert_plans(ready=False) + prerequisite = _acquire(cli, PREREQUISITE, "agent-b", "prerequisite-original") + transferred = cli(["task-lease", "transfer"], PREREQUISITE, "--owner", "agent-b", + "--idempotency-key", "prerequisite-original", "--expected-version", str(prerequisite["lease"]["version"]), + "--new-owner", "agent-a", "--new-idempotency-key", "prerequisite-receiver", + "--ttl-seconds", "600", "--transfer-claim") + assert transferred["transferred"] + assert_plans(ready=False) + assert not _acquire(cli, WAITING, "agent-a", "waiting-next", version=1, expected_exit=1)["ok"] + assert _complete(cli, PREREQUISITE, "agent-a", "prerequisite-receiver", transferred["lease"]["version"])["completed"] + assert_plans(ready=True) + resumed = _acquire(cli, WAITING, "agent-a", "waiting-next", version=1) + assert resumed["acquired"] + assert _complete(cli, WAITING, "agent-a", "waiting-next", resumed["lease"]["version"])["completed"] diff --git a/tests/control_plane/test_shadow_mutant_locators.py b/tests/control_plane/test_shadow_mutant_locators.py new file mode 100644 index 0000000000..0914f62c9b --- /dev/null +++ b/tests/control_plane/test_shadow_mutant_locators.py @@ -0,0 +1,28 @@ +"""Every Stage 2C mutation must still target the current production source.""" +from __future__ import annotations + +import importlib.util +import sys +from pathlib import Path + +import pytest + + +ROOT = Path(__file__).resolve().parents[2] +SPEC = importlib.util.spec_from_file_location( + "stage2c_mutant_locator_contract", ROOT / "examples/shared-goal-authority-e2e/mutants.py", +) +assert SPEC is not None and SPEC.loader is not None +HARNESS = importlib.util.module_from_spec(SPEC) +sys.modules[SPEC.name] = HARNESS +SPEC.loader.exec_module(HARNESS) + + +@pytest.mark.parametrize("case", HARNESS.CASES, ids=lambda case: case.name) +def test_mutation_targets_current_production_source(case) -> None: + # Exercise each real edit rather than maintaining a second locator list. + # The full harness separately requires a green oracle and assertion kill. + for relative_path, edit in case.edits: + original = (ROOT / relative_path).read_text(encoding="utf-8") + mutated = edit(original) + assert mutated != original, f"{case.name}: mutation did not change {relative_path}" diff --git a/tests/test_explore_todo_readiness.py b/tests/test_explore_todo_readiness.py new file mode 100644 index 0000000000..215695298f --- /dev/null +++ b/tests/test_explore_todo_readiness.py @@ -0,0 +1,106 @@ +"""Current dispatch suggestions must obey the existing Todo readiness contract.""" +from copy import deepcopy + +import pytest + +from loopx.capabilities.explore.todo_branch_plan import build_explore_todo_branch_plan +from loopx.capabilities.explore.worker_branch_plan import build_explore_worker_branch_plan + + +ORCHESTRATION = { + "spawn_allowed": True, "max_children": 4, "explore_harness": {"enabled": True}, +} + + +def todo(todo_id, **fields): + return { + "todo_id": todo_id, "text": "Inspect a synthetic artifact", "status": "open", + "priority": "P0", "task_class": "advancement_task", "claimed_by": "worker", + "required_write_scopes": [f"artifacts/{todo_id}/**"], **fields, + } + + +@pytest.fixture(params=["todo", "worker"]) +def planner(request): + def build(todos, *, orchestration=ORCHESTRATION, **kwargs): + common = dict(goal_id="readiness", agent_id="worker", todos=todos, + orchestration=orchestration, scheduler_load=0, **kwargs) + if request.param == "todo": + plan = build_explore_todo_branch_plan(width=4, **common) + return plan, plan["selected_branches"], plan["rejected_candidates"] + plan = build_explore_worker_branch_plan(worker_width=4, **common) + return plan, plan["selected_worker_branches"], plan["rejected_worker_branches"] + return build + + +def selected_ids(branches): + return {todo_id for branch in branches + for todo_id in branch.get("todo_ids", [branch["todo_id"]])} + + +@pytest.mark.parametrize("condition", ["todo_done:todo_prepare", "monitor_changed:todo_monitor"]) +@pytest.mark.parametrize("ready", [False, None]) +def test_unready_todo_is_diagnostic_only_then_becomes_selectable(planner, condition, ready): + waiting = todo("todo_waiting", resume_when=condition) + if ready is not None: + waiting["resume_ready"] = ready + original = deepcopy(waiting) + plan, selected, rejected = planner([waiting]) + assert selected == [] + assert "No actionable" in plan["next_action"] + assert rejected[0]["todo_id"] == "todo_waiting" + assert rejected[0]["selection_status"] == "not_actionable_open" + assert rejected[0]["resume_when"] == condition + assert not rejected[0].get("suggested_commands") + assert waiting == original + + _, selected, rejected = planner([{**waiting, "resume_ready": True}]) + assert selected_ids(selected) == {"todo_waiting"} + assert selected[0]["suggested_commands"] + assert rejected == [] + + +@pytest.mark.parametrize("shared_scope", [False, True]) +def test_waiting_todo_uses_neither_worker_bundle_nor_resource_capacity(planner, shared_scope): + waiting = todo("todo_waiting", resume_when="todo_done:todo_prepare", resume_ready=False) + ready = todo("todo_ready", priority="P2") + if shared_scope: + waiting["required_write_scopes"] = ready["required_write_scopes"] + for item in (waiting, ready): + item["required_capabilities"] = ["resource_lane:compute"] + plan, selected, rejected = planner([waiting, ready], resource_capacities={"compute": 2}) + assert selected_ids(selected) == {"todo_ready"} + assert plan["resource_portfolio"]["selected_slot_count"] == 1 + assert plan["resource_portfolio"]["remaining_slot_count"] == 1 + assert next(row for row in rejected if row["todo_id"] == "todo_waiting")["resume_ready"] is False + + +@pytest.mark.parametrize("fields", [ + {"status": "done", "done": True}, + {"status": "blocked"}, + {"goal_acceptance_guard": {"allowed": False}}, +]) +def test_other_canonical_ineligibility_is_not_just_a_score_penalty(planner, fields): + _, selected, rejected = planner([todo("todo_unavailable", **fields)]) + assert selected == [] + assert rejected[0]["selection_status"] == "not_actionable_open" + + +def test_lineage_and_other_agent_ownership_do_not_become_completion_dependencies(planner): + successor = todo("todo_successor", successor_todo_id="todo_other") + foreign = todo("todo_other", claimed_by="peer") + _, selected, _ = planner([successor, foreign]) + assert selected_ids(selected) == {"todo_successor"} + + +def test_analysis_only_and_disabled_gates_keep_their_authority(planner): + waiting = todo("todo_waiting", resume_when="todo_done:todo_prepare", resume_ready=False) + ready = todo("todo_ready") + _, selected, rejected = planner( + [waiting, ready], orchestration={"explore_harness": {"enabled": True}}, + ) + assert selected_ids(selected) == {"todo_ready"} + assert all(not row.get("suggested_commands") for row in [*selected, *rejected]) + plan, selected, rejected = planner([waiting, ready], orchestration={}) + assert plan["enabled"] is False + assert selected == rejected == [] diff --git a/tests/test_python_ci_workflow.py b/tests/test_python_ci_workflow.py index b8484204b1..8f4174c052 100644 --- a/tests/test_python_ci_workflow.py +++ b/tests/test_python_ci_workflow.py @@ -6,6 +6,7 @@ import shlex import subprocess import sys +import time import xml.etree.ElementTree as ET from pathlib import Path @@ -382,6 +383,55 @@ def test_four_shards_execute_each_test_once_and_merge_portable_coverage( assert "--cov-fail-under" not in template +def test_shard_names_failed_cases_before_the_remaining_tests_finish(tmp_path: Path) -> None: + # Drive the real shard command: one worker fails while the other waits. + # A runner deadline must not leave only an anonymous F in the progress log. + step = WORKFLOW.split("name: Run test shard", 1)[1] + template = step.split("run: >-", 1)[1].split(" - name:", 1)[0] + args = shlex.split(template.replace("${{ matrix.shard }}", "1")) + args[0] = sys.executable + args[args.index("--cov=loopx")] = "--cov=ci_subject" + (tmp_path / "ci_subject.py").write_text("def value(case):\n return case\n") + (tmp_path / "test_subject.py").write_text( + "from pathlib import Path\nimport time\nimport pytest\n" + "from ci_subject import value\n" + "@pytest.mark.parametrize('case', range(8))\n" + "def test_outcome(case):\n" + " if case < 4:\n assert value(case) < 0, 'synthetic failure'\n" + " Path('waiting').touch()\n" + " while not Path('release').exists():\n time.sleep(0.01)\n" + " assert value(case) >= 4\n", + ) + env = {key: value for key, value in os.environ.items() + if not key.startswith(("COVERAGE", "COV_CORE", "PYTEST"))} + log = tmp_path / "progress.log" + with log.open("w") as output: + process = subprocess.Popen(args, cwd=tmp_path, env=env, stdout=output, stderr=subprocess.STDOUT) + try: + deadline = time.monotonic() + 15 + while time.monotonic() < deadline: + progress = log.read_text() + if (tmp_path / "waiting").exists() and re.search( + r"FAILED test_subject.py::test_outcome\[\d+\]", progress, + ): + break + time.sleep(0.05) + assert process.poll() is None, log.read_text() + assert (tmp_path / "waiting").exists(), log.read_text() + assert re.search(r"FAILED test_subject.py::test_outcome\[\d+\]", log.read_text()) + finally: + (tmp_path / "release").touch() + try: + process.wait(timeout=30) + except subprocess.TimeoutExpired: + process.kill() + process.wait() + assert process.returncode == 1, log.read_text() + cases = ET.parse(tmp_path / "junit.xml").findall(".//testcase") + assert len(cases) == 2 + assert len([case for case in cases if case.find("failure") is not None]) == 1 + + def test_backend_and_mixed_prs_require_the_browser_qualified_artifact() -> None: producer = WORKFLOW.split(" chat-bundle:\n", 1)[1].split(" chat-bundle-browser:\n", 1)[0] assert "needs.changes.outputs.core_tests == 'true'" in producer