fix(control-plane): bind wait exemptions to exact current targets

Signed-off-by: huangruiteng <huangrt01@163.com>

Author: huangruiteng

examples/shared-goal-authority-e2e/mutants.py

@@ -47,6 +47,13 @@ def command(self) -> list[str]:
 
 
 CASES = [
+    Case('delivery_wait_target_unbound', (('loopx/control_plane/todos/resume_condition.ts', replacement(
+        'condition.target_todo_id !== spec.target || ', '')),),
+         'tests/control_plane_ts/delivery_response.test.ts', 'exact dependency identity'),
+    Case('delivery_wait_unknown_class', (('loopx/control_plane/todos/resume_condition.ts', replacement(
+        '["advancement_task", "user_gate", "user_action", "blocker"].includes(String(condition.target_task_class))',
+        'true')),),
+         'tests/control_plane_ts/delivery_response.test.ts', 'exact dependency identity'),
     Case('rollout_cwd_root', (('loopx/cli_rollout.py', replacement(
         'resolve_runtime_root(registry, runtime_root_arg, registry_path=registry_path)',
         'resolve_runtime_root(registry, runtime_root_arg)')),),

loopx/control_plane/todos/resume_condition.ts

@@ -432,22 +432,35 @@ function diagnosedCondition(condition: JsonObject, waitingTodoId: string): JsonO
 
 /** Positive wait proof for consumers that may relax supervision. Historical
  * absence of an invalid marker is not proof of a valid, identified target. */
-export function resumeConditionHasKnownPendingTarget(condition: JsonObject): boolean {
+export function resumeConditionHasKnownPendingTarget(condition: JsonObject, waitingTodo: JsonObject): boolean {
+  const spec = parseResumeWhen(waitingTodo.resume_when);
+  if (!spec || condition.resume_when !== spec.normalized || condition.kind !== spec.kind
+    || (condition.target !== undefined && condition.target !== spec.target)) return false;
   if (condition.schema_version !== "todo_resume_condition_v0" || condition.satisfied !== false
-    || resumeAvailabilityReason(condition) !== "resume_condition_pending") return false;
-  switch (condition.kind) {
+    || diagnoseTodoResumeCondition(condition, String(waitingTodo.todo_id)).state !== "pending") return false;
+  if ((spec.kind === "todo_done" || spec.kind === "monitor_changed")
+    && (condition.target_todo_id !== spec.target || spec.target === waitingTodo.todo_id)) return false;
+  switch (spec.kind) {
     case "todo_done":
       return ["open", "deferred"].includes(String(condition.target_status))
-        && typeof condition.target_task_class === "string" && condition.target_task_class !== "continuous_monitor"
+        && ["advancement_task", "user_gate", "user_action", "blocker"].includes(String(condition.target_task_class))
         && (condition.target_archive_state === null || condition.target_archive_state === "active");
     case "monitor_changed":
       return condition.target_task_class === "continuous_monitor" && condition.target_status === "open"
-        && typeof condition.baseline_generation === "number" && condition.baseline_generation >= 0
+        && typeof condition.baseline_generation === "number" && Number.isSafeInteger(condition.baseline_generation)
+        && condition.baseline_generation >= 0 && condition.baseline_generation === waitingTodo.resume_monitor_generation
         && typeof condition.material_change_generation === "number"
+        && Number.isSafeInteger(condition.material_change_generation) && condition.material_change_generation >= 0
         && condition.material_change_generation <= condition.baseline_generation;
-    case "capacity_available": return condition.provider_required === false;
-    case "pr_merged": return typeof condition.pr_repo === "string" && condition.pr_repo.length > 0
-      && condition.repository_binding_state !== "ambiguous";
+    case "capacity_available": return condition.provider_required === false
+      && condition.provider === "runtime_available_capabilities" && condition.capability === spec.target;
+    case "pr_merged": {
+      const ref = normalizedPrRef(spec.target);
+      const repository = ref?.repo ?? githubRepository(waitingTodo.task_repository);
+      return ref !== null && repository !== null && condition.pr_repo === repository
+        && condition.pr_number === ref.number && condition.repository_binding_state !== "ambiguous"
+        && condition.repository_binding_source === (ref.repo ? "qualified_resume_when" : "task_repository");
+    }
     default: return false;
   }
 }

loopx/control_plane/work_items/delivery_history.py

@@ -93,14 +93,16 @@ def project_delivery_response(
     source = next((item for item in todo_planning_source_items(summary, include_terminal=True)
                    if item.get("todo_id") == run.get("todo_id")), None) if summary else None
     fields = ("todo_id", "role", "status", "task_class", "archive_state", "claimed_by",
-              "excluded_agents", "resume_when", "resume_ready", "resume_condition")
+              "excluded_agents", "resume_when", "resume_ready", "resume_condition",
+              "resume_monitor_generation", "task_repository")
     todo = {key: source[key] for key in fields if key in source} if source else None
     if todo and isinstance(todo.get("resume_condition"), dict):
         condition = todo["resume_condition"]
         todo["resume_condition"] = {key: condition[key] for key in (
             "schema_version", "resume_when", "satisfied", "invalid_target", "invalid_state",
-            "kind", "target_todo_id", "target_status", "target_task_class", "target_archive_state",
-            "baseline_generation", "material_change_generation", "provider_required", "pr_repo", "repository_binding_state",
+            "kind", "target", "target_todo_id", "target_status", "target_task_class", "target_archive_state",
+            "baseline_generation", "material_change_generation", "provider_required", "provider", "capability",
+            "pr_repo", "pr_number", "repository_binding_state", "repository_binding_source",
         ) if key in condition}
     result = effect_runtime_result("work_item.delivery_response.project", {
         "run": _run_facts(run), "todo": todo,

loopx/control_plane/work_items/delivery_history.ts

@@ -118,7 +118,7 @@ export function projectDeliveryResponse(value: unknown): JsonObject {
     runs: [input.run], outcome_floor_configured: true }).runs as DeliverySignal[])[0];
   const todo = input.todo === null ? null : requireJsonObject(input.todo, "bound Todo");
   const runAgent = typeof input.run_agent_id === "string" ? input.run_agent_id.trim() : "";
-  const agentId = typeof input.agent_id === "string" ? input.agent_id.trim() : runAgent;
+  const agentId = typeof input.agent_id === "string" ? input.agent_id.trim() : "";
   const owner = typeof todo?.claimed_by === "string" ? todo.claimed_by.trim() : "";
   const excluded = Array.isArray(todo?.excluded_agents) ? todo.excluded_agents : [];
   const condition = todo?.resume_condition && typeof todo.resume_condition === "object"
@@ -131,12 +131,10 @@ export function projectDeliveryResponse(value: unknown): JsonObject {
     && todo.role === "agent" && todo.task_class === "advancement_task"
     && ["open", "deferred"].includes(String(todo.status))
     && (todo.archive_state === undefined || todo.archive_state === "active")
-    && agentId === runAgent && (!owner || owner === agentId) && !excluded.includes(agentId)
+    && Boolean(agentId) && agentId === runAgent && (!owner || owner === agentId) && !excluded.includes(agentId)
     && condition?.schema_version === "todo_resume_condition_v0"
-    && condition.resume_when === todo.resume_when && Boolean(todo.resume_when)
-    && condition.target_todo_id !== todo.todo_id
     && condition.satisfied === false && todo.resume_ready !== true
-    && resumeConditionHasKnownPendingTarget(condition);
+    && resumeConditionHasKnownPendingTarget(condition, todo);
   return {
     schema_version: "delivery_response_v0",
     outcome_floor_applicable: !canonicalWait,

tests/control_plane/test_delivery_response.py

@@ -1,9 +1,12 @@
 """Current canonical waits constrain historical supervision, not vice versa."""
 from copy import deepcopy
+import pytest
 
 from loopx.control_plane.handoff.delivery_contract import handoff_delivery_contract
 from loopx.control_plane.testing.quota_fixtures import quota_status_payload, quota_todo_item, quota_todo_summary
 from loopx.control_plane.work_items.delivery_history import project_delivery_response
+from loopx.control_plane.todos.summary_item import TODO_PLANNING_SOURCE_KEYS
+from loopx.control_plane.todos.resume_condition import evaluate_todo_resume_conditions
 from loopx.control_plane.work_items.work_lane_context import build_work_lane_context_contract
 from loopx.quota import build_quota_should_run, quota_with_handoff_outcome_floor
 from loopx.status import project_post_handoff_history
@@ -23,14 +26,41 @@ def waiting_todo():
     return quota_todo_item(todo_id="todo_delivery", title="Wait for the dependency", status="deferred",
         claimed_by="agent-a", resume_when="todo_done:todo_dependency", resume_ready=False,
         resume_condition={"schema_version": "todo_resume_condition_v0", "resume_when": "todo_done:todo_dependency",
-                          "satisfied": False, "kind": "todo_done", "target_status": "open",
+                          "satisfied": False, "kind": "todo_done", "target_todo_id": "todo_dependency", "target_status": "open",
                           "target_task_class": "advancement_task", "target_archive_state": "active"})
 
 
 def dependency_todo():
     return quota_todo_item(todo_id="todo_dependency", title="Complete prerequisite", claimed_by="agent-b")
 
 
+@pytest.mark.parametrize("patch", [
+    {"target_todo_id": "todo_other"}, {"target_todo_id": None},
+    {"target_task_class": "unknown_class"}, {"target_task_class": ""},
+    {"kind": "monitor_changed"},
+])
+def test_stale_or_malformed_target_cannot_suppress_any_consumer(patch):
+    todo = waiting_todo()
+    extra = quota_todo_item(todo_id="todo_alternative", title="Continue independent work", claimed_by="agent-a")
+    summary = quota_todo_summary([todo, dependency_todo(), extra], claim_scope_agent_id="agent-a")
+    # Mutate the evaluated snapshot, not the authoring input: summary building
+    # legitimately recomputes supplied conditions from its canonical source.
+    for key in TODO_PLANNING_SOURCE_KEYS:
+        for row in summary.get(key, []):
+            if row["todo_id"] == "todo_delivery":
+                row["resume_condition"].update(patch)
+    readiness = project_post_handoff_history([blocked_run()] * 3, PROFILE)
+    asset = {"execution_profile": PROFILE, "agent_todos": summary}
+    item = {"handoff_readiness": readiness, "agent_todos": summary, "project_asset": asset}
+    before = deepcopy(item)
+    assert project_delivery_response(blocked_run(), summary)["reason"] == "history_supervision"
+    assert quota_with_handoff_outcome_floor({"state": "eligible"}, waiting_on="codex",
+        project_asset=asset, handoff_readiness=readiness)["state"] == "focus_wait"
+    assert handoff_delivery_contract(item) is not None
+    assert build_work_lane_context_contract(item, agent_todo_summary=summary)["must_attempt_work"] is True
+    assert item == before
+
+
 def test_status_compaction_preserves_binding_and_all_consumers_defer_to_current_wait():
     readiness = project_post_handoff_history([blocked_run()] * 3, PROFILE)
     summary = quota_todo_summary([waiting_todo(), dependency_todo()], claim_scope_agent_id="agent-a")
@@ -55,6 +85,26 @@ def test_status_compaction_preserves_binding_and_all_consumers_defer_to_current_
     assert "outcome_followthrough" not in lane
 
 
+@pytest.mark.parametrize("resume, patch", [
+    ("monitor_changed:todo_dependency", {"baseline_generation": 1}),
+    ("capacity_available:network", {"capability": "other"}),
+    ("pr_merged:#1", {"pr_number": 2}),
+])
+def test_real_resume_projection_identity_survives_python_transport(resume, patch):
+    todo = {**waiting_todo(), "resume_when": resume, "resume_monitor_generation": 0,
+            "task_repository": "git:github.com/example/project"}
+    dependency = {**dependency_todo(), "task_class": "continuous_monitor", "material_change_generation": 0}
+    condition = evaluate_todo_resume_conditions([todo], source_items=[dependency], available_capabilities=[])[todo["todo_id"]]
+    summary = quota_todo_summary([todo, dependency], claim_scope_agent_id="agent-a")
+    for key in TODO_PLANNING_SOURCE_KEYS:
+        for row in summary.get(key, []):
+            if row["todo_id"] == todo["todo_id"]:
+                row["resume_condition"] = condition
+    assert project_delivery_response(blocked_run(), summary)["reason"] == "canonical_todo_wait"
+    condition.update(patch)
+    assert project_delivery_response(blocked_run(), summary)["reason"] == "history_supervision"
+
+
 def test_surface_supervision_remains_and_invalid_wait_does_not_clear_floor():
     summary = quota_todo_summary([waiting_todo()], claim_scope_agent_id="agent-a")
     asset = {"execution_profile": PROFILE, "agent_todos": summary}

tests/control_plane_ts/delivery_response.test.ts

@@ -13,7 +13,7 @@ const waiting = { todo_id: "todo_delivery", role: "agent", status: "deferred",
   task_class: "advancement_task", claimed_by: "agent-a", resume_when: "todo_done:todo_dependency",
   resume_ready: false, resume_condition: { schema_version: "todo_resume_condition_v0",
     resume_when: "todo_done:todo_dependency", satisfied: false, kind: "todo_done",
-    target_status: "open", target_task_class: "advancement_task", target_archive_state: "active" } };
+    target_todo_id: "todo_dependency", target_status: "open", target_task_class: "advancement_task", target_archive_state: "active" } };
 const input = { run, todo: waiting, run_agent_id: "agent-a", agent_id: "agent-a" };
 
 test("wait proof consumes the real resume evaluator for all four condition kinds", () => {
@@ -34,9 +34,37 @@ test("wait proof consumes the real resume evaluator for all four condition kinds
     const condition = (evaluated.conditions as JsonObject[])[0].condition;
     assert.equal(projectDeliveryResponse({ ...input, todo: { ...todo, resume_condition: condition } }).outcome_floor_applicable,
       !valid, resume + JSON.stringify(source));
+    if (valid) {
+      const proof = condition as JsonObject;
+      const mutations: JsonObject[] = [{ kind: "unknown" }, { target: "other_target" }];
+      if (proof.kind === "todo_done" || proof.kind === "monitor_changed") {
+        mutations.push({ target_todo_id: "todo_other" }, { target_todo_id: null }, { target_task_class: "unknown" });
+      }
+      if (proof.kind === "monitor_changed") mutations.push(
+        { baseline_generation: 1 }, { material_change_generation: -1 }, { material_change_generation: 0.5 });
+      if (proof.kind === "capacity_available") mutations.push({ capability: "other" }, { provider_required: true });
+      if (proof.kind === "pr_merged") mutations.push({ pr_number: 2 }, { pr_repo: "example/other" });
+      for (const patch of mutations) assert.equal(projectDeliveryResponse({ ...input,
+        todo: { ...todo, resume_condition: { ...proof, ...patch } } }).reason,
+      "history_supervision", resume + JSON.stringify(patch));
+    }
   }
 });
 
+test("known non-monitor completion classes and repository-bound PR waits remain supported", () => {
+  for (const task_class of ["advancement_task", "user_gate", "user_action", "blocker"]) {
+    assert.equal(projectDeliveryResponse({ ...input, todo: { ...waiting,
+      resume_condition: { ...waiting.resume_condition, target_task_class: task_class } } }).reason, "canonical_todo_wait");
+  }
+  const todo = { ...waiting, resume_when: "pr_merged:#1", task_repository: "git:github.com/example/project" };
+  const evaluated = evaluateTodoResumeConditions({ schema_version: TODO_RESUME_EVALUATION_REQUEST_SCHEMA_VERSION,
+    items: [todo], source_items: [], rollout_events: [], available_capabilities: [] });
+  const condition = (evaluated.conditions as JsonObject[])[0].condition;
+  assert.equal(projectDeliveryResponse({ ...input, todo: { ...todo, resume_condition: condition } }).reason, "canonical_todo_wait");
+  assert.equal(projectDeliveryResponse({ ...input, todo: { ...todo, task_repository: "git:github.com/example/other",
+    resume_condition: condition } }).reason, "history_supervision");
+});
+
 test("a bound blocked observation delegates a current legal wait to canonical planning", () => {
   const before = structuredClone(input);
   const result = projectDeliveryResponse(input);
@@ -46,9 +74,24 @@ test("a bound blocked observation delegates a current legal wait to canonical pl
   assert.deepEqual(input, before);
 });
 
+test("exact dependency identity and a supported completion class are required", () => {
+  for (const patch of [
+    { target_todo_id: "todo_other" }, { target_todo_id: null },
+    { target: "todo_other" }, { kind: "capacity_available" },
+    { target_task_class: "unknown_class" }, { target_task_class: "" },
+  ]) {
+    const result = projectDeliveryResponse({ ...input, todo: { ...waiting,
+      resume_condition: { ...waiting.resume_condition, ...patch } } });
+    assert.equal(result.reason, "history_supervision", JSON.stringify(patch));
+    assert.equal(result.outcome_floor_applicable, true);
+    assert.equal((result.outcome_followthrough as JsonObject).required, true);
+  }
+});
+
 test("history alone, missing source, invalid wait, and other actors cannot exempt the floor", () => {
   for (const patch of [
     { todo: null }, { agent_id: "agent-b" }, { run_agent_id: "agent-b" },
+    { agent_id: null }, { run_agent_id: null }, { agent_id: "", run_agent_id: "" },
     { todo: { ...waiting, todo_id: "todo_other" } },
     { todo: { ...waiting, claimed_by: "agent-b" } },
     { todo: { ...waiting, excluded_agents: ["agent-a"] } },