fix(explore): exclude unready todos from dispatch suggestions

.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

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. 全量本地测试仍省略分片参数即可。
 

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)))",

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:
 

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

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."

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.

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:

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"