fix(collaboration): derive worker workspace from its pinned binding

Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com>

Author: huangruiteng

loopx/collaboration_mcp.py

@@ -177,7 +177,6 @@ def _spawn(self, operation_id: str) -> None:
             sys.executable, "-m", "loopx.collaboration_mcp", "--delegation-action", "worker", "--runtime-root", str(self.root),
             "--registry", str(self.registry), "--goal-id", self.goal_id,
             "--agent-id", self.agent_id, "--execution-config", str(self.config), "--operation-id=" + operation_id,
-            "--workspace", _read(self.path(operation_id))["identity"]["binding"]["workspace"],
         ], stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
             start_new_session=True, close_fds=True)
 
@@ -305,6 +304,7 @@ def _execute(self, path: Path, row: dict, binding: dict) -> None:
         request_id = row["identity"]["request_id"]
         common = ["--goal-id", self.goal_id, "--agent-id", binding["agent_id"]]
         host = binding["host_args"]
+        # Preserve the journaled validator argv so existing Turns retain their resume identity.
         validator = [sys.executable, "-m", "loopx.collaboration_mcp", "--delegation-action", "validate", "--runtime-root", str(self.root),
                      "--registry", str(self.registry), "--goal-id", self.goal_id,
                      "--agent-id", self.agent_id, "--execution-config", str(self.config),
@@ -407,7 +407,7 @@ def main():
     parser.add_argument("--registry", type=Path, required=True)
     parser.add_argument("--goal-id", required=True)
     parser.add_argument("--agent-id", required=True)
-    parser.add_argument("--workspace", type=Path, required=True)
+    parser.add_argument("--workspace", type=Path, help="Required when serving MCP; workers use their pinned binding")
     parser.add_argument("--execution-config", type=Path, help="Explicit operator-owned local execution bindings")
     parser.add_argument("--delegation-action", choices=["worker", "validate"],
                         help="Run a host-owned delegation action instead of serving MCP")
@@ -426,6 +426,8 @@ def main():
             except LockAcquireTimeoutError:
                 pass  # Another worker still owns the operation after the bounded wait.
         return
+    if args.workspace is None:
+        parser.error("--workspace is required when serving MCP")
     create_server(
         args.runtime_root.resolve(),
         args.registry.resolve(),

loopx/control_plane/turn_driver/host_candidate.py

@@ -253,7 +253,7 @@ def build_result(
         if field == "vision_unchanged_reason":
             continue
         value = candidate.get(field)
-        text = _bounded(value, limit=limit) if value else ""
+        text = _bounded(value, limit=limit)
         if text:
             result[field] = text
 

tests/test_collaboration_mcp.py

@@ -2,12 +2,23 @@
 
 import asyncio
 import json
+import subprocess
 import sys
 
 from mcp import ClientSession, StdioServerParameters
 from mcp.client.stdio import stdio_client
 
 
+def test_serving_mcp_still_requires_an_explicit_workspace(tmp_path):
+    result = subprocess.run([
+        sys.executable, "-m", "loopx.collaboration_mcp",
+        "--runtime-root", str(tmp_path), "--registry", str(tmp_path / "registry.json"),
+        "--goal-id", "delivery", "--agent-id", "builder",
+    ], capture_output=True, text=True, timeout=10)
+    assert result.returncode == 2
+    assert "--workspace is required when serving MCP" in result.stderr
+
+
 def test_scoped_stdio_tools_do_not_offer_shell_or_sender_override(tmp_path):
     registry = tmp_path / "registry.json"
     config = {

tests/test_local_delegation.py

@@ -158,18 +158,22 @@ async def disconnect_requester():
     assert (root / "analyst" / "initial" / "host-invocations").read_text() == "1"
     assert demo.canonical_tasks(root)["todo_analyst-initial"]["done"]
     returned = returns(original.root, original.goal_id, "lead")["items"]
-    assert len(returned) == 1 and returned[0]["decision"] == "adopt"
+    assert len(returned) == 1
+    assert returned[0]["decision"] == "adopt"
     assert wait(reconnected)["artifacts"] == result["artifacts"]
+    changed_brief = {**brief(), "purpose": "Changed instruction"}
     with pytest.raises(ValueError, match="identity conflict"):
-        reconnected.start("analysis", "analysis-1", {**brief(), "purpose": "Changed instruction"})
+        reconnected.start("analysis", "analysis-1", changed_brief)
+    ungranted = Delegations(original.root, original.registry, original.goal_id, "reviewer", original.config)
+    original_brief = brief()
     with pytest.raises(Exception, match="no delegation grant"):
-        Delegations(original.root, original.registry, original.goal_id, "reviewer", original.config).start("analysis", "other", brief())
+        ungranted.start("analysis", "other", original_brief)
     registry = json.loads(original.registry.read_text())
     registry["goals"][0]["status"] = "stopped"
     original.registry.write_text(json.dumps(registry))
     assert reconnected.read("analysis-1")["status"] == "accepted"
     with pytest.raises(ValueError, match="stopped"):
-        reconnected.start("analysis", "new-operation", brief())
+        reconnected.start("analysis", "new-operation", original_brief)
     registry["goals"][0]["status"] = "active"
     original.registry.write_text(json.dumps(registry))
     output = root / "analyst" / "initial" / "output.json"
@@ -183,7 +187,8 @@ def test_model_success_without_receiver_adoption_cannot_complete(service):
     (root / "skip-adoption").touch()
     runner.start("analysis", "analysis-1", brief())
     result = wait(runner)
-    assert result["status"] == "rejected" and "did not adopt" in result["error"]
+    assert result["status"] == "rejected"
+    assert "did not adopt" in result["error"]
     assert not demo.canonical_tasks(root)["todo_analyst-initial"]["done"]
 
 

tests/test_managed_research_team.py

@@ -82,10 +82,11 @@ def test_canonical_delivery_requires_completed_current_dependencies(team, monkey
              "--todo-id", "todo_reviewer-initial", "--text", "Independently analyze the initial filing and sources")
     with pytest.raises(RuntimeError, match="goal_acceptance_stale"):
         demo.complete(root, "reviewer", "initial")
+    document = json.loads((root / "bootstrap.json").read_text())["document"]
+    provider_revision = inspect_goal_acceptance(**route)["provider_revision"]
     with pytest.raises(ValueError):
-        configure_goal_acceptance(**route, document=json.loads((root / "bootstrap.json").read_text())["document"],
-                                  agent_id="lead", expected_provider_revision=inspect_goal_acceptance(**route)["provider_revision"],
-                                  execute=True)
+        configure_goal_acceptance(**route, document=document, agent_id="lead",
+                                  expected_provider_revision=provider_revision, execute=True)
     held = plan(root, "reviewer", "initial")
     assert not held.get("turn_envelope", {}).get("action", {}).get("delivery_allowed", False), held
     # Explicit owner amendment for this negative fixture, never done by delegate().