fix(todos): allow an agent-bound user gate's initial unblock link

docs/project-agent-todo-contract.md

@@ -293,7 +293,30 @@ use `--clear-blocks-agent --global-gate`. Merely clearing scope in a multi-agent
 Goal is rejected; it must not turn an ambiguous gate into a global one.
 
 When a user gate only blocks one concrete action, add the blocked todo id with
-`unblocks_todo_id=<todo_id>`. When multiple todos share the same broad
+`unblocks_todo_id=<todo_id>`. If the gate was created first, its registered bound
+agent can attach the initial link while the gate is open:
+
+```bash
+loopx todo update --goal-id <goal-id> --role user --todo-id <gate-id> \
+  --agent-id <bound-agent> --unblocks-todo-id <agent-todo-id>
+```
+
+This is an administrative provider-CAS edit, including in `hard_lease`; it does
+not claim work or approve the gate. Optional note/evidence/reason may accompany
+the link. The target must already exist as an Agent Todo in the same provider
+snapshot and Goal, with `claimed_by` matching the gate's `bound_agent` /
+`blocks_agent`. Preview and write reject missing or out-of-scope targets without
+writing or consuming the first-link opportunity; the same gate can then be
+linked to a valid target. Existing links, gate scope, actor bindings and explicit
+or unreleased lease proofs retain their ordinary admission. Complete the linked
+gate with the actual decision and evidence to consume the target's covered
+decision scopes.
+`todo_done:<gate-id>` checks dependency readiness only; it does not consume those
+scopes. If an unlinked gate was already completed, reconcile its decision through
+an explicit linked gate using the original owner evidence rather than treating
+dependency completion as approval.
+
+When multiple todos share the same broad
 `action_kind`, use the schema-backed decision-scope fields instead of relying
 on title/body token overlap:
 

loopx/control_plane/coordination/todo_update_admission.ts

@@ -219,14 +219,30 @@ export function todoUpdateAdmissionRejection(
     const repositoryRejection = leaseRepositoryRejection(todo, lease);
     return repositoryRejection === null ? null : reject(repositoryRejection, "Evidence association must retain the completed work repository");
   }
-  // User reminders use an administrative provider-CAS edit. Retained released
+  // An initial gate link is planning, not execution or an owner decision.
+  const boundUserGateLink = todo.role === "user" && todo.task_class === "user_gate" &&
+    todo.status === "open" && todo.claimed_by == null && todo.unblocks_todo_id == null &&
+    input.actor_agent_id !== null && todo.bound_agent === input.actor_agent_id &&
+    todo.blocks_agent === input.actor_agent_id &&
+    typeof intent.unblocks_todo_id === "string" && intent.unblocks_todo_id.length > 0 &&
+    Object.keys(input.patch).every(field => field === "note") &&
+    input.clear_fields.every(field => field === "note") &&
+    Object.keys(intent).every(field => ["unblocks_todo_id", "evidence", "reason"].includes(field));
+  // User planning uses an administrative provider-CAS edit. Retained released
   // history is not an execution grant; active/expired history needs exact
   // release first. Explicit or partial proof never grants this exception.
   const boundUserActionMetadata = isBoundUserActionMetadataUpdate(todo, input);
-  if (boundUserActionMetadata && (lease === undefined || lease.status === "released") &&
+  if ((boundUserActionMetadata || boundUserGateLink) && (lease === undefined || lease.status === "released") &&
       input.lease_idempotency_key == null && input.lease_expected_version == null) {
     try {
       if (lease !== undefined) canonicalTaskLease(lease, input.goal_id, input.todo_id);
+      if (boundUserGateLink) {
+        const target = indexCoordinationProjection(head, input.goal_id).todos.get(intent.unblocks_todo_id as string);
+        if (target?.role !== "agent" || target.claimed_by !== todo.blocks_agent) {
+          return reject("user_gate_link_target_invalid",
+            "Initial gate link requires an existing Agent Todo in this Goal claimed by the gate's bound/blocked agent");
+        }
+      }
     } catch (error) {
       return reject("invalid_coordination_projection",
         error instanceof Error ? error.message : "invalid retained lease facts");

tests/control_plane/test_settled_monitor_user_gate.py

@@ -105,6 +105,70 @@ def call(*args):
     assert not any(row["classification"] in {"state_refreshed", "quota_slot_spent"} for row in rows)
 
 
+def test_file_gate_can_complete_before_primary_monitor_poll(tmp_path, monkeypatch):
+    isolate_sqlite_runtime(tmp_path, monkeypatch)
+    registry, runtime, _state = _write_fixture(tmp_path)
+    scope = "resource:action:prices_file"
+    gated = add_goal_todo(registry_path=registry, goal_id=GOAL_ID, role="agent",
+        text="Implement inventory value using the supplied prices", task_class="advancement_task",
+        status="blocked", claimed_by=AGENT_ID, agent_id=AGENT_ID,
+        required_decision_scopes=[scope])
+    gate = add_goal_todo(registry_path=registry, goal_id=GOAL_ID, role="user",
+        text="Supply inputs/prices.json", task_class="user_gate", bound_agent=AGENT_ID,
+        blocks_agent=AGENT_ID, unblocks_todo_id=gated["todo_id"],
+        decision_scope=scope, agent_id=AGENT_ID)
+    monitor = _add_monitor(registry, text="Observe the authorized prices file",
+        target_key="prices-file", next_due_at="2000-01-01T00:00:00Z")
+    turn = "file-gate-monitor-turn"
+    common = ["--goal-id", GOAL_ID, "--agent-id", AGENT_ID]
+    turn_args = [*common, "--runtime-profile", "generic_cli", "--turn-instance-id", turn,
+        "--available-capability", "network", "--available-capability", "external_evidence_poll"]
+    def call(*args):
+        return run_json_cli(*args, registry_path=registry, runtime_root=runtime)
+    admitted = call("quota", "should-run", *turn_args)
+    assert admitted["selected_todo"]["todo_id"] == monitor["todo_id"]
+    binding = admitted["heartbeat_receipt"]["settlement_identity"]
+
+    prices = registry.parent.parent / "inputs" / "prices.json"
+    prices.parent.mkdir()
+    prices.write_text('{"A":199}', encoding="utf-8")
+    observed = prices.read_text(encoding="utf-8")
+    assert json.loads(observed) == {"A": 199}
+    # Gate control is not completion of the monitor's Turn settlement Todo.
+    completed = call("todo", "complete", *common, "--todo-id", gate["todo_id"],
+        "--role", "user", "--decision-outcome", "approve",
+        "--evidence", f"inputs/prices.json supplied and validated: {observed}")
+    assert completed["completed"] is True
+    assert completed["decision_scope_resolution"]["state"] == "resolved"
+    assert completed["unblock_resume"]["state"] == "resumed"
+    rows = {row["todo_id"]: row for row in call("todo", "list", "--goal-id", GOAL_ID)["todos"]}
+    assert rows[gate["todo_id"]]["status"] == "done"
+    assert rows[gated["todo_id"]]["status"] == "open"
+    assert not rows[gated["todo_id"]].get("required_decision_scopes")
+    before_poll = call("quota", "should-run", *turn_args)
+    assert before_poll["heartbeat_receipt"]["settlement_identity"] == binding
+    assert before_poll["selected_todo"]["todo_id"] == monitor["todo_id"]
+
+    poll = call("quota", "monitor-poll", *turn_args, "--todo-id", monitor["todo_id"],
+        "--target-key", "prices-file", "--result-hash", observed, "--material-change", "--execute")
+    assert poll["turn_continuation"]["current_turn_settled"] is True
+    assert poll["turn_continuation"]["same_turn_independent_settlement_allowed"] is False
+    assert poll["turn_continuation"]["next_turn_required"] is True
+    replay = call("quota", "should-run", *turn_args)
+    assert replay["heartbeat_receipt"]["settlement_identity"] == binding
+    assert replay["selected_todo"]["todo_id"] == monitor["todo_id"]
+    assert replay["should_run"] is False
+    assert replay["interaction_contract"]["agent_channel"]["delivery_allowed"] is False
+    fresh_args = ["file-gate-implementation-turn" if arg == turn else arg for arg in turn_args]
+    fresh = call("quota", "should-run", *fresh_args)
+    assert fresh["selected_todo"]["todo_id"] == gated["todo_id"]
+    assert fresh["interaction_contract"]["agent_channel"]["delivery_allowed"] is True
+    index = runtime / "goals" / GOAL_ID / "runs" / "index.jsonl"
+    events = [json.loads(line) for line in index.read_text(encoding="utf-8").splitlines()]
+    assert sum(row["classification"] == "quota_monitor_poll" for row in events) == 1
+    assert not any(row["classification"] in {"state_refreshed", "quota_slot_spent"} for row in events)
+
+
 def _primary_monitor_fixture(tmp_path, monkeypatch, provider, route="source"):
     isolate_sqlite_runtime(tmp_path, monkeypatch)
     source, runtime, _state = _write_fixture(tmp_path, provider=provider)

tests/control_plane/test_user_gate_resume_cli.py

@@ -0,0 +1,127 @@
+"""A file gate created before its dependent is repairable through the public CLI."""
+from __future__ import annotations
+
+import json
+
+import pytest
+
+from canonical_authority_fixture import isolate_sqlite_runtime, promoted_create_fixture
+from loopx.control_plane.coordination.local_authority import read_canonical_todos_if_promoted
+from loopx.control_plane.testing.canary_harness import run_json_cli, run_json_cli_result
+
+
+@pytest.fixture
+def tmp_path(tmp_path_factory):
+    # The hashed SQLite filename must fit under the native Windows path limit.
+    return tmp_path_factory.mktemp("gate")
+
+
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+@pytest.mark.parametrize("dry_run", [True, False], ids=["preview", "write"])
+@pytest.mark.parametrize("invalid_target", ["missing", "user", "other_goal", "other_lane", "unclaimed"])
+def test_gate_first_link_rejects_invalid_target_without_consuming_link(
+    tmp_path, monkeypatch, provider, dry_run, invalid_target,
+):
+    isolate_sqlite_runtime(tmp_path, monkeypatch)
+    registry, runtime, state = promoted_create_fixture(
+        tmp_path, provider=provider, handoff_mode="hard_lease", registered_agents=["agent-a", "agent-b"])
+
+    def cli(*args):
+        return run_json_cli(*args, "--goal-id", "goal-a", registry_path=registry, runtime_root=runtime)
+
+    def snapshot():
+        return read_canonical_todos_if_promoted(runtime_root=runtime, goal_id="goal-a", include_leases=True)
+
+    gate = cli("todo", "add", "--role", "user", "--task-class", "user_gate", "--agent-id", "agent-a",
+        "--decision-scope", "resource:action:prices_file", "--text", "Provide prices")
+    target = cli("todo", "add", "--role", "agent", "--task-class", "advancement_task", "--claimed-by", "agent-a",
+        "--status", "blocked", "--required-decision-scope", "resource:action:prices_file",
+        "--resume-when", f"todo_done:{gate['todo_id']}", "--text", "Use the supplied prices")
+    invalid_id = "todo_missing"
+    if invalid_target == "user":
+        invalid_id = gate["todo_id"]
+    elif invalid_target == "other_goal":
+        other_registry, other_runtime, _ = promoted_create_fixture(
+            tmp_path / "x", goal_id="goal-b", provider=provider, handoff_mode="hard_lease")
+        other = run_json_cli("todo", "add", "--goal-id", "goal-b", "--role", "agent",
+            "--task-class", "advancement_task", "--claimed-by", "agent-a", "--text", "Other Goal work",
+            registry_path=other_registry, runtime_root=other_runtime)
+        invalid_id = other["todo_id"]
+    elif invalid_target in {"other_lane", "unclaimed"}:
+        claim = ("--claimed-by", "agent-b") if invalid_target == "other_lane" else ()
+        invalid_id = cli("todo", "add", "--role", "agent", "--task-class", "advancement_task",
+            *claim, "--text", "Work outside the gate's lane")["todo_id"]
+
+    before, state_before = snapshot(), state.read_bytes()
+    link_args = ("todo", "update", "--role", "user", "--todo-id", gate["todo_id"], "--agent-id", "agent-a",
+        "--update-operation-id", "link-file-gate", "--update-expected-provider-revision", before["provider_revision"])
+    code, rejected = run_json_cli_result(*link_args, "--goal-id", "goal-a", "--unblocks-todo-id", invalid_id,
+        *(("--dry-run",) if dry_run else ()), registry_path=registry, runtime_root=runtime)
+    assert code != 0, rejected
+    assert rejected["reason_code"] == "user_gate_link_target_invalid"
+    assert snapshot() == before
+    assert state.read_bytes() == state_before
+
+    valid_link = (*link_args, "--unblocks-todo-id", target["todo_id"])
+    assert cli(*valid_link, "--dry-run")["status"] == "planned"
+    assert snapshot() == before
+    assert cli(*valid_link)["status"] == "applied"
+    assert cli(*valid_link)["status"] == "replayed"
+    linked = cli("todo", "list", "--todo-id", gate["todo_id"])["todo"]
+    assert linked["unblocks_todo_id"] == target["todo_id"]
+    assert linked["status"] == "open"
+    dependent = cli("todo", "list", "--todo-id", target["todo_id"])["todo"]
+    assert dependent["status"] == "blocked"
+    assert dependent["required_decision_scopes"]
+
+
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+def test_gate_first_link_consumes_scope_and_resumes_dependent(tmp_path, monkeypatch, provider):
+    isolate_sqlite_runtime(tmp_path, monkeypatch)
+    registry, runtime, state = promoted_create_fixture(
+        tmp_path, provider=provider, handoff_mode="hard_lease", registered_agents=["agent-a", "agent-b"])
+
+    def cli(*args):
+        return run_json_cli(*args, "--goal-id", "goal-a", registry_path=registry, runtime_root=runtime)
+
+    def snapshot():
+        return read_canonical_todos_if_promoted(runtime_root=runtime, goal_id="goal-a", include_leases=True)
+
+    gate = cli("todo", "add", "--role", "user", "--task-class", "user_gate", "--agent-id", "agent-a",
+        "--decision-scope", "resource:action:prices_file", "--text", "Provide inputs/prices.json")
+    target = cli("todo", "add", "--role", "agent", "--task-class", "advancement_task", "--claimed-by", "agent-a",
+        "--status", "blocked", "--required-decision-scope", "resource:action:prices_file",
+        "--resume-when", f"todo_done:{gate['todo_id']}", "--text", "Calculate inventory value with supplied prices")
+    before = snapshot()
+    link_args = ("todo", "update", "--role", "user", "--todo-id", gate["todo_id"], "--agent-id", "agent-a",
+        "--unblocks-todo-id", target["todo_id"], "--update-operation-id", "link-file-gate",
+        "--update-expected-provider-revision", before["provider_revision"])
+    assert cli(*link_args, "--dry-run")["status"] == "planned"
+    assert snapshot() == before
+    assert cli(*link_args)["status"] == "applied"
+    assert cli(*link_args)["status"] == "replayed"
+    linked = snapshot()
+    assert linked["leases"] == before["leases"]
+    assert cli("todo", "list", "--todo-id", target["todo_id"])["todo"]["required_decision_scopes"]
+
+    prices = state.parents[3] / "inputs" / "prices.json"
+    prices.parent.mkdir()
+    prices.write_text('{"item": 125}\n', encoding="utf-8")
+    assert json.loads(prices.read_text(encoding="utf-8")) == {"item": 125}
+    completed = cli("todo", "complete", "--role", "user", "--todo-id", gate["todo_id"], "--agent-id", "agent-a",
+        "--decision-outcome", "approve", "--evidence", "Requested prices file is present with integer prices")
+    assert completed["decision_scope_resolution"]["remaining_required_decision_scopes"] == []
+    assert completed["unblock_resume"]["state"] == "resumed"
+    resumed = cli("todo", "list", "--todo-id", target["todo_id"])["todo"]
+    assert resumed["status"] == "open"
+    assert resumed["claimed_by"] == "agent-a"
+    assert resumed["required_decision_scopes"] == []
+    assert resumed["resume_condition"]["satisfied"] is True
+    claim = cli("todo", "claim", "--todo-id", target["todo_id"], "--agent-id", "agent-a", "--claimed-by", "agent-a",
+        "--claim-operation-id", "resume-value-work")
+    assert claim["execution_managed"] is True
+
+    denied, _ = run_json_cli_result("todo", "update", "--role", "user", "--goal-id", "goal-a",
+        "--todo-id", gate["todo_id"], "--agent-id", "agent-a", "--unblocks-todo-id", "todo_other",
+        registry_path=registry, runtime_root=runtime)
+    assert denied != 0