VM_CONTACT: accept the preserved workspace root in U4

backend/deploy/VM_CONTACT.md

@@ -405,7 +405,10 @@ require(all((target[k].startswith('/') for k in ('root', 'admin_socket', 'token_
 root = pathlib.Path(target["root"]).resolve()
 install = pathlib.Path("/opt/vm-mcp").resolve()
 require(root != pathlib.Path('/') and root != install and (root not in install.parents) and (install not in root.parents), 'workspace overlaps install')
-require(root.is_dir() and any((p.name != '.vm-mcp-home' for p in root.iterdir())), 'workspace must exist and be nonempty')
+# A preserved root (the inventoried one) may be empty; a newly chosen root must hold data.
+preserved = root == pathlib.Path(state['home']).parent.resolve()
+require(root.is_dir(), 'workspace must exist and be a directory')
+require(preserved or any((p.name != '.vm-mcp-home' for p in root.iterdir())), 'a changed workspace root must exist and be nonempty')
 require(not re.search('[\\s%\\\\"\']', target['root']), 'hmmm: pinned installer cannot represent this workspace drop-in safely')
 # Verify the completed archive without touching the live deployment.
 h = hashlib.sha256()
@@ -463,6 +466,11 @@ PY
 )
 ```
 
+The workspace guard accepts the preserved (inventoried) root whether or not it holds
+data; it must still exist and be a directory. Only a root changed in `target.json`
+must also be nonempty, so an edit cannot silently point the service at an empty one.
+A root counts as preserved when it resolves to the parent of the U1 state's service-user home.
+
 Expect `rc=0`. The Python result is initialized by the actual invocation, so an
 interactive shell's old `rc` cannot turn success into failure. On any failure after
 stopping, keep using the human recovery path and use U6 or fix forward. Do not infer
@@ -521,7 +529,10 @@ def check(label, passed):
 for key, field in (("VM_MCP_ROOT", "root"), ("VM_MCP_PROFILE", "profile"), ("VM_MCP_PORT", "port"), ("VM_MCP_ADMIN_SOCKET", "admin_socket")):
     check(key + " preserved", env.get(key) == target[field])
 check("target service home", pwd.getpwnam("vmmcp").pw_dir == target["root"].rstrip("/") + "/.vm-mcp-home")
-check("workspace not empty", any(p.name != ".vm-mcp-home" for p in pathlib.Path(target["root"]).iterdir()))
+state = json.loads((B / "state.json").read_text())
+preserved = pathlib.Path(target["root"]).resolve() == pathlib.Path(state["home"]).parent.resolve()
+check("workspace exists" + (" (preserved root, may be empty)" if preserved else " and is not empty"),
+      pathlib.Path(target["root"]).is_dir() and (preserved or any(p.name != ".vm-mcp-home" for p in pathlib.Path(target["root"]).iterdir())))
 unit_result = subprocess.run(["systemctl", "show", "vm-mcp.service",
                               "--property=ReadWritePaths,ProtectSystem"], check=True, text=True, capture_output=True)
 unit = dict(line.split("=", 1) for line in unit_result.stdout.splitlines() if "=" in line)

backend/tests/test_vm_contact_upgrade.py

@@ -198,6 +198,11 @@ def __init__(self, *parts):
                 if type(Path()).__init__ is not object.__init__:
                     super().__init__(vm.path(Path(*parts)))
 
+            def with_segments(self, *parts):
+                # 3.12+ derives child/glob paths through this hook; they are already
+                # sandbox-resolved, so return plain paths (matching 3.11 behaviour).
+                return type(Path())(*parts)
+
             def open(self, *args, **kwargs):
                 return vm.path(self).open(*args, **kwargs)
 
@@ -312,6 +317,74 @@ def test_failed_new_snapshot_keeps_existing_latest(self):
         self.assertEqual(self.vm.backup(), old)
         self.assertEqual((old / "COMPLETE").read_text(), "complete\n")
 
+    def set_target_root(self, root):
+        target = self.vm.target()
+        target["root"] = root
+        (self.vm.backup() / "target.json").write_text(json.dumps(target))
+        return target
+
+    def installed(self):
+        return any(args[0] == "bash" for args, _ in self.vm.commands)
+
+    def test_u4_accepts_preserved_empty_root(self):
+        self.snapshot_target()
+        self.vm.path("/srv/existing/data.txt").unlink()
+        self.vm.commands.clear()
+        self.vm.execute("U4")
+        self.assertTrue(self.installed())
+        self.assertEqual(self.vm.installer_env["VM_MCP_ROOT"], "/srv/existing")
+
+    def test_u4_refuses_preserved_missing_root(self):
+        self.snapshot_target()
+        shutil.rmtree(self.vm.path("/srv/existing"))
+        self.vm.commands.clear()
+        with self.assertRaisesRegex(AssertionError, "must exist and be a directory"):
+            self.vm.execute("U4")
+        self.assertEqual(self.destructive(), [])
+
+    def test_u4_refuses_preserved_root_that_is_a_file(self):
+        self.snapshot_target()
+        shutil.rmtree(self.vm.path("/srv/existing"))
+        self.vm.write("/srv/existing", "not a directory")
+        self.vm.commands.clear()
+        with self.assertRaisesRegex(AssertionError, "must exist and be a directory"):
+            self.vm.execute("U4")
+        self.assertEqual(self.destructive(), [])
+
+    def test_u4_refuses_changed_empty_root(self):
+        self.snapshot_target()
+        self.vm.path("/srv/other").mkdir(parents=True)
+        self.set_target_root("/srv/other")
+        self.vm.commands.clear()
+        with self.assertRaisesRegex(AssertionError, "changed workspace root must exist and be nonempty"):
+            self.vm.execute("U4")
+        self.assertEqual(self.destructive(), [])
+
+    def test_u4_changed_nonempty_root_passes_the_workspace_guard(self):
+        self.snapshot_target()
+        self.vm.write("/srv/other/data.txt", "fixture")
+        self.set_target_root("/srv/other")
+        self.vm.commands.clear()
+        self.vm.execute("U4")
+        self.assertEqual(self.vm.installer_env["VM_MCP_ROOT"], "/srv/other")
+
+    def test_u5_fails_on_changed_empty_root(self):
+        self.snapshot_target()
+        self.vm.path("/srv/other").mkdir(parents=True)
+        target = self.set_target_root("/srv/other")
+        self.vm.write("/etc/vm-mcp.env", "".join(key + "=" + shlex.quote(target[field]) + "\n" for key, field in
+            (("VM_MCP_ROOT", "root"), ("VM_MCP_PROFILE", "profile"),
+             ("VM_MCP_PORT", "port"), ("VM_MCP_ADMIN_SOCKET", "admin_socket"))))
+        self.vm.home = "/srv/other/.vm-mcp-home"
+        self.vm.write_paths = "/srv/other"
+        with self.assertRaises(SystemExit) as failure:
+            self.vm.execute("U5")
+        self.assertEqual(failure.exception.code, 1)
+        self.vm.write("/srv/other/data.txt", "fixture")
+        with self.assertRaises(SystemExit) as ok:
+            self.vm.execute("U5")
+        self.assertEqual(ok.exception.code, 0)
+
     def test_shell_metacharacters_and_spaces_roundtrip_without_evaluation(self):
         root = '/srv/a workspace;$(touch /tmp/never-created)`id`'
         self.vm.home = root + "/.vm-mcp-home"
@@ -416,7 +489,12 @@ def environment():
             self.vm.execute("U5")
         self.assertEqual(failure.exception.code, 1)
         self.vm.home = target["root"] + "/.vm-mcp-home"
+        # The preserved root may be empty, but it must still exist.
         self.vm.path("/srv/existing/data.txt").unlink()
+        with self.assertRaises(SystemExit) as ok:
+            self.vm.execute("U5")
+        self.assertEqual(ok.exception.code, 0)
+        shutil.rmtree(self.vm.path("/srv/existing"))
         with self.assertRaises(SystemExit) as failure:
             self.vm.execute("U5")
         self.assertEqual(failure.exception.code, 1)