fix(ovpack): restore protected scope roots

openviking/storage/ovpack/operations.py

@@ -116,10 +116,44 @@ async def _ensure_parent_exists(viking_fs, parent: str, ctx: RequestContext) ->
         await viking_fs.mkdir(parent, ctx=ctx)
 
 
+_PROTECTED_SCOPE_ROOTS = frozenset({"viking://agent", "viking://user"})
+
+
+def _is_protected_scope_root(uri: str) -> bool:
+    return uri.rstrip("/") in _PROTECTED_SCOPE_ROOTS
+
+
+async def _clear_protected_root(viking_fs, root_uri: str, ctx: RequestContext) -> None:
+    """Remove a protected namespace's children while preserving its virtual root."""
+    while True:
+        try:
+            entries = await viking_fs.ls(root_uri, show_all_hidden=True, ctx=ctx)
+        except (NotFoundError, FileNotFoundError):
+            return
+        if not entries:
+            return
+
+        children: list[str] = []
+        prefix = f"{root_uri.rstrip('/')}/"
+        for entry in entries:
+            child_uri = entry.get("uri") if isinstance(entry, dict) else None
+            if not isinstance(child_uri, str) or not child_uri.startswith(prefix):
+                raise InvalidArgumentError(
+                    "Protected scope listing returned an invalid child",
+                    details={"root": root_uri, "child": child_uri},
+                )
+            children.append(child_uri)
+        for child_uri in children:
+            await viking_fs.rm(child_uri, recursive=True, ctx=ctx)
+
+
 async def _remove_existing_root(viking_fs, root_uri: str, ctx: RequestContext) -> None:
     if not hasattr(viking_fs, "rm"):
         logger.warning(f"[ovpack] Cannot remove existing resource without rm(): {root_uri}")
         return
+    if _is_protected_scope_root(root_uri):
+        await _clear_protected_root(viking_fs, root_uri, ctx)
+        return
     try:
         await viking_fs.rm(root_uri, recursive=True, ctx=ctx)
     except NotFoundError:
@@ -679,7 +713,9 @@ async def restore_ovpack(
             if kind in {"manifest", "internal"} or rel_path == "":
                 continue
             if kind == "directory":
-                await viking_fs.mkdir(join_uri(root_uri, rel_path), exist_ok=True, ctx=ctx)
+                target_uri = join_uri(root_uri, rel_path)
+                if not _is_protected_scope_root(target_uri):
+                    await viking_fs.mkdir(target_uri, exist_ok=True, ctx=ctx)
                 continue
 
             data = zf.read(safe_zip_path)

tests/misc/test_ovpack_import_policy.py

@@ -54,6 +54,33 @@ async def read_file(self, uri: str, ctx=None):
         raise FileNotFoundError(uri)
 
 
+class FakeInitializedRestoreVikingFS(FakeVikingFS):
+    """Model the public roots and protected user namespace of an initialized server."""
+
+    def __init__(self) -> None:
+        super().__init__()
+        self.removed: list[str] = []
+        self.children = {
+            "viking://resources": [{"uri": "viking://resources/old.md"}],
+            "viking://user": [{"uri": "viking://user/old-user"}],
+        }
+
+    async def ls(self, uri: str, show_all_hidden: bool = False, ctx=None):
+        if uri in self.children:
+            return list(self.children[uri])
+        raise NotFoundError(uri, "file")
+
+    async def mkdir(self, uri: str, exist_ok: bool = False, ctx=None):
+        assert uri != "viking://user"
+        await super().mkdir(uri, exist_ok=exist_ok, ctx=ctx)
+
+    async def rm(self, uri: str, recursive: bool = False, ctx=None):
+        assert uri != "viking://user"
+        self.removed.append(uri)
+        for entries in self.children.values():
+            entries[:] = [entry for entry in entries if entry["uri"] != uri]
+
+
 class FakeExportVikingFS:
     def __init__(self) -> None:
         self.binary_files = {
@@ -621,6 +648,30 @@ async def test_backup_restore_contract(temp_ovpack_path: Path, request_ctx: Requ
     assert fake_fs.tree_calls == ["viking://resources", "viking://user"]
 
 
+@pytest.mark.asyncio
+async def test_backup_restore_overwrites_initialized_protected_scope(
+    temp_ovpack_path: Path, request_ctx: RequestContext
+):
+    await backup_ovpack(FakeBackupVikingFS(), str(temp_ovpack_path), ctx=request_ctx)
+    fake_fs = FakeInitializedRestoreVikingFS()
+
+    assert (
+        await restore_ovpack(
+            fake_fs,
+            str(temp_ovpack_path),
+            request_ctx,
+            on_conflict="overwrite",
+        )
+        == "viking://"
+    )
+    assert fake_fs.removed == ["viking://resources", "viking://user/old-user"]
+    assert "viking://user" not in fake_fs.created_dirs
+    assert fake_fs.written_files == [
+        "viking://resources/README.md",
+        "viking://user/alice/sessions/sess_1/.meta.json",
+    ]
+
+
 @pytest.mark.asyncio
 async def test_backup_skips_missing_semantic_sidecars(
     temp_ovpack_path: Path,