권한은 딱 한 번 옮겨

    async def import_existing(self):
        """One-time transfer of authority; subsequent calls only audit copies."""
        ownership = await self.io("list", "ownership")
        if ownership:
            if ownership[0]["state"] == "canonical":
                return {"imported": [], "ownership": ownership[0], "copies": await self.copy_status()}
            return await self.finish_adoption(ownership[0])
        registry = await self.upstream.legacy_catalog()
        imported = []
        async with self.intent_lock:
            current_sets = await self.io("list", "set")
            for source in registry["sets"]:
                current = next((s for s in current_sets if (s["soul_pk"], s["channel"], s["name"]) == (source["soul_pk"], source["channel"], source["name"])), None)
                if current:
                    continue  # importing never resets Dad's current selections
                frames = {}
                for frame, item in source["frames"].items():
                    path = Path(item["path"]).resolve()
                    path.relative_to((Path(__file__).resolve().parents[2] / "cwkPippa/avatars").resolve())
                    data = await asyncio.to_thread(path.read_bytes)
                    if hashlib.sha256(data).hexdigest() != item["sha256"]:
                        raise Conflict("An existing avatar changed during import. Retry the import.")
                    asset = await self.io("save_asset", data, label=f"{source['name']} · {frame}", subjects=[source["soul_pk"]], origin="imported", source_path=str(path))
                    frames[frame] = asset["id"]
                record = await self.io("add", "set", soul_pk=source["soul_pk"], channel=source["channel"], name=source["name"], frames=list(frames), published=frames, selection=frames, live_revision=source["revision"], archived=False)
                imported.append(record)
        collections = await self.io("list", "set")
        legacy = [s["id"] for s in collections if s.get("published") and not await self.io("list", "publication", s["id"])]
        ownership = await self.io("create_intent", "legacy-avatar-authority-v1", {"schema": 1}, "ownership", {"state": "adopting", "owner": "cwkAvatars", "legacy_sets": legacy, "started_at": now()})
        result = await self.finish_adoption(ownership)
        return {**result, "imported": imported}

    async def finish_adoption(self, ownership):
        for id in ownership["legacy_sets"]:
            request_id = f"adopt-{id}"
            records = await self.io("list", "publication", id)
            existing = next((p for p in records if p.get("payload", {}).get("request_id") == request_id), None)
            if existing:
                if existing["state"] != "applied":
                    receipt = await self.upstream.publish(existing["payload"])
                    await self.io("complete_publication", existing["id"], receipt)
                continue
            collection = await self.io("get", id, "set")
            await self.publish(id, request_id, collection["version"], collection["live_revision"], mode="adopt")
        ownership = await self.io("update", ownership["id"], state="canonical", completed_at=now())
        return {"ownership": ownership, "imported": []}

    async def copy_status(self):
        """Compare installed copies without ever importing their changes."""
        collections = [c for c in await self.io("list", "set") if c.get("published")]
        ids = list(dict.fromkeys(id for c in collections for id in c["published"].values()))
        lineage = await self.io("asset_lineages", ids) if ids else {}
        # One registry round trip covers every published subject; the policy
        # verdict is still the registry's, taken now.
        await self.subjects(sorted({s for a in lineage.values() for s in a["subjects"]}))
        answer = []
        for collection in collections:
            hashes = {frame: lineage[id]["sha256"] for frame, id in sorted(collection["published"].items())}
            revision = hashlib.sha256(json.dumps(hashes, sort_keys=True).encode()).hexdigest()
            actual = await self.upstream.revision(collection["soul_pk"], collection["channel"], collection["name"])
            answer.append({"set_id": collection["id"], "soul_pk": collection["soul_pk"], "channel": collection["channel"], "name": collection["name"], "expected_revision": revision, "actual_revision": actual, "matches": actual == revision})
        return answer

        if body.get("mode") == "adopt":
            actual = {frame: digest(path) for frame, path in frame_files(target).items()}
            if actual != {frame: item["sha256"] for frame, item in files.items()}:
                raise ValueError("Legacy copies do not match the Avatars originals")
        backup = REPO / "avatar-backups" / "studio" / request_id
        backup.mkdir(parents=True, exist_ok=True)
        if target.exists():
            if not (backup / "previous").exists():
                shutil.copytree(target, backup / "previous")
            if revision(frame_files(backup / "previous")) != previous:
                raise ValueError("Backup verification failed")
        if body.get("mode") == "adopt":
            manifest["revision"] = previous
            receipt = {"request_id": request_id, "fingerprint": fingerprint, "status": "prepared", "mode": "adopt", "provenance": "legacy-adoption", "soul_pk": body["soul_pk"], "channel": body["channel"], "name": body["name"], "previous_revision": previous, "revision": previous, "backup": str(backup), "target": str(target), "manifest": manifest}
            atomic_write_text(receipt_path, json.dumps(receipt, indent=2))
            return finish_publication(receipt, receipt_path)