"""Read legacy Agent/Markdown assets and emit guarded acceptance SQL. The caller owns the read-only snapshot and target transaction. Source prompt text is opaque data: it is preserved verbatim, never evaluated or logged here. Plans and SQL contain private source bodies; persist only safe_asset_summary. """ from __future__ import annotations import hashlib import json import re import sqlite3 import uuid from weknora_identity_etl import ADMIN_TARGET_ID, IDENTITY_LEDGER from weknora_model_etl import resolve_agent_model SOURCE = "yawan-workbench" TARGET_DATABASE = "yawan_native_acceptance_20260915" TARGET_TENANT = 10059 SOURCE_ORGANIZATION = 1 LEDGER = "yawan_legacy_asset_mappings" BINDING_LEDGER = "yawan_legacy_asset_bindings" NAMESPACE = uuid.UUID("8b8041ef-9cd6-4803-96ca-17fe215ae948") AGENT_COLUMNS = ("id", "name", "role", "desc", "content_tags", "kb_ids", "fusion_agent_ids", "model_provider_id", "model_name", "model_catalog_id", "prompt_template_id", "allowed_tools", "rules", "rules_version", "status", "builtin", "organization_id", "owner_user_id", "visibility", "created_at", "runtime_snapshot", "persona_json", "digital_human_id", "kb_category_ids") SKILL_COLUMNS = ("id", "name", "desc", "instructions", "enabled", "builtin", "version", "organization_id", "owner_user_id", "visibility", "created_at") TEMPLATE_COLUMNS = ("id", "name", "desc", "role_prompt", "task_template", "constraints", "examples", "output_schema", "style", "version", "enabled", "builtin", "organization_id", "owner_user_id", "visibility", "created_at", "updated_at") AGENT_JSON = {"content_tags": list, "kb_ids": list, "fusion_agent_ids": list, "allowed_tools": list, "runtime_snapshot": dict, "persona_json": dict, "kb_category_ids": dict} SOURCE_POLICY = "private_bodies_remain_private; no_legacy_grants_inferred" def asset_id(kind: str, source_id: int) -> str: if kind not in ("agent", "skill", "template") or type(source_id) is not int or source_id <= 0: raise ValueError("Invalid source asset identity") return str(uuid.uuid5(NAMESPACE, f"{SOURCE}:tenant:{TARGET_TENANT}:organization:1:{kind}:{source_id}")) def _read(db: sqlite3.Connection, table: str, columns: tuple[str, ...]) -> list[dict]: return [dict(zip(columns, row)) for row in db.execute(f"SELECT {','.join(columns)} FROM {table} ORDER BY id")] def _json(value, expected: type): parsed = expected() if value is None else json.loads(value) if not isinstance(parsed, expected): raise ValueError("Legacy asset JSON has an incompatible type") return parsed def _uuid(value) -> str: if not isinstance(value, str): raise ValueError("Target mapping requires a canonical UUID") try: if str(uuid.UUID(value)) != value: raise ValueError except ValueError: raise ValueError("Target mapping requires a canonical UUID") from None return value def _mapping(mapping: dict) -> dict[str, str]: if not isinstance(mapping, dict): raise ValueError("Expected a source ID to native ID mapping") result = {} for key, value in mapping.items(): normalized = str(key) if not normalized.isdigit() or int(normalized) <= 0 or normalized != str(int(normalized)): raise ValueError("Mapping contains an invalid source ID") if normalized in result: raise ValueError("Duplicate source mapping identity") result[normalized] = _uuid(value) if len(set(result.values())) != len(result): raise ValueError("Target mapping is not one-to-one") return result def _owner(source: dict, identity_mapping: dict) -> tuple[int, str]: if (source.get("organization_id") != SOURCE_ORGANIZATION or type(source.get("owner_user_id")) is not int or source["owner_user_id"] < 0): raise ValueError("Legacy asset owner or organization is invalid") source_owner = source["owner_user_id"] or 1 if str(source_owner) not in identity_mapping: raise ValueError("Legacy asset owner has no durable identity mapping") return source_owner, identity_mapping[str(source_owner)] def _source_flags(source: dict) -> None: if (source.get("visibility") not in ("private", "global") or source.get("builtin") not in (0, 1)): raise ValueError("Legacy asset visibility or flags need separate review") if (not isinstance(source.get("name"), str) or not source["name"].strip() or len(source["name"]) > 255 or "\x00" in source["name"]): raise ValueError("Legacy asset name is invalid") def _target_agent(source: dict, identities: dict, kbs: dict, models: dict, binding_ids: list[str]) -> tuple[dict, dict]: _source_flags(source) _, owner = _owner(source, identities) if source["status"] not in ("active", "disabled"): raise ValueError("Legacy Agent status needs separate review") if not isinstance(source["rules"], str) or "\x00" in source["rules"]: raise ValueError("Legacy Agent rules are invalid") for field, expected in AGENT_JSON.items(): if not isinstance(source[field], expected): raise ValueError("Legacy Agent configuration type differs") if source["role"] not in ("", "general", "coordinator", "production", "qc", "service", "training_map"): raise ValueError("Legacy Agent role is invalid") tags = source["content_tags"] if (len(tags) > 32 or any(not isinstance(tag, str) or not tag.strip() or len(tag) > 64 or "\x00" in tag for tag in tags) or len(set(tags)) != len(tags)): raise ValueError("Legacy Agent content tags are invalid") fusions = source["fusion_agent_ids"] if (len(fusions) > 8 or any(type(value) is not int or value <= 0 or value == source["id"] for value in fusions) or len(set(fusions)) != len(fusions) or (fusions and source["role"] != "production")): raise ValueError("Legacy Agent fusion configuration is invalid") if source["kb_category_ids"]: raise ValueError("Legacy Agent category restrictions require native mapping before execution") if len(binding_ids) > 32: raise ValueError("Legacy Agent exceeds native instruction skill binding limit") kb_ids = source["kb_ids"] if (any(type(k) is not int or k <= 0 or str(k) not in kbs for k in kb_ids) or len(set(kb_ids)) != len(kb_ids)): raise ValueError("Legacy Agent has an unmapped or duplicate knowledge base") routing = resolve_agent_model(models, source) # Native QA gives textual assets an execution path without granting legacy # runtime tools (handoff/memory/upstream reads have no equivalent contract). config = {"agent_mode": "quick-answer", "agent_type": "custom", "system_prompt": source["rules"] or "", "role": source["role"], "content_tags": tags, "fusion_agent_ids": [asset_id("agent", value) for value in fusions], "model_id": routing["target_model_id"], "kb_selection_mode": "selected" if kb_ids else "none", "knowledge_bases": [kbs[str(k)] for k in kb_ids], "instruction_skill_ids": binding_ids, "skills_selection_mode": "none", "selected_skills": [], "allowed_tools": [], "mcp_selection_mode": "none", "mcp_services": [], "sandbox_config_id": "", "web_search_enabled": False, "citation_enabled": True} return {"id": asset_id("agent", source["id"]), "tenant_id": TARGET_TENANT, "name": source["name"], "description": source["desc"] or "", "created_by": owner, "is_builtin": False, "visibility": source["visibility"], "enabled": source["status"] == "active", "config": config}, routing def _target_skill(source: dict, identities: dict) -> dict: _source_flags(source) _, owner = _owner(source, identities) if source["enabled"] not in (0, 1) or type(source["version"]) is not int or source["version"] < 1: raise ValueError("Legacy skill enabled/version flags are invalid") instructions = source["instructions"] if not isinstance(instructions, str) or not instructions.strip() or "\x00" in instructions: raise ValueError("Legacy skill instructions are invalid") if len(instructions.encode("utf-8")) > 256 * 1024: raise ValueError("Legacy skill exceeds the native import size limit") return {"id": asset_id("skill", source["id"]), "tenant_id": TARGET_TENANT, "name": source["name"], "description": source["desc"] or "", "created_by": owner, "version": str(source["version"]), "instructions": instructions, "instruction_only": True, "visibility": source["visibility"], "enabled": bool(source["enabled"]), "bundle_ref": "", "bundle_sha256": hashlib.sha256(instructions.encode()).hexdigest()} def _binding_review(agent: dict, skill: dict) -> list[str]: review = [] if not agent["enabled"]: review.append("agent_disabled_preserved") if not skill["enabled"]: review.append("skill_disabled_preserved") if skill["visibility"] == "private" and skill["created_by"] != agent["created_by"]: review.append("blocked_private_owner_mismatch") if skill["visibility"] == "private" and agent["visibility"] == "global": review.append("global_agent_private_skill_requires_user_access_check") return review def _binding_effective(agent_owner: str, skill: dict) -> bool: return skill["visibility"] == "global" or skill["created_by"] == agent_owner def asset_plan(db: sqlite3.Connection, identity_mapping: dict, kb_mapping: dict, model_mapping: dict) -> dict: """Build a body-bearing plan; mapping values must come from target ledgers. identity_mapping / kb_mapping are {legacy_id: native_uuid}. model_mapping is the safe model_plan result. No target writes or credential reads occur. """ identities, kbs = _mapping(identity_mapping), _mapping(kb_mapping) if identities.get("1") != ADMIN_TARGET_ID: raise ValueError("Legacy owner zero must map through the pinned administrator") # Shared grant semantics require a separate native mapping. Preserve no # inferred public access if the source changes while preparing this batch. grants = {t: _read(db, t, ("id", "agent_id" if t == "agentgrant" else "skill_id", "user_id", "permission", "enabled", "created_at")) for t in ("agentgrant", "agentskillgrant")} if any(grants.values()): raise ValueError("Source explicit grants require a reviewed native mapping") sources = _read(db, "agent", AGENT_COLUMNS) skills = [] for source in _read(db, "agentskill", SKILL_COLUMNS): skills.append({"source": source, "target": _target_skill(source, identities)}) skill_by_id = {x["source"]["id"]: x for x in skills} if len(skill_by_id) != len(skills): raise ValueError("Duplicate source skill identity") bindings = _read(db, "agentskillbinding", ("id", "agent_id", "skill_id")) by_agent, pairs, binding_source_ids = {}, set(), set() source_agents = {a["id"] for a in sources} source_owners = {a["id"]: _owner(a, identities)[1] for a in sources} for binding in bindings: pair = (binding["agent_id"], binding["skill_id"]) if (binding["agent_id"] not in source_agents or binding["skill_id"] not in skill_by_id or pair in pairs or binding["id"] in binding_source_ids): raise ValueError("Legacy binding is duplicated or refers to a missing asset") pairs.add(pair) binding_source_ids.add(binding["id"]) if _binding_effective(source_owners[binding["agent_id"]], skill_by_id[binding["skill_id"]]["target"]): by_agent.setdefault(binding["agent_id"], []).append(asset_id("skill", binding["skill_id"])) agents = [] for source in sources: for field, expected in AGENT_JSON.items(): source[field] = _json(source[field], expected) target, routing = _target_agent(source, identities, kbs, model_mapping, by_agent.get(source["id"], [])) agents.append({"source": source, "target": target, "model_resolution": routing, "compatibility": ["source_rules_verbatim_in_system_prompt", "native_quick_answer_execution", "legacy_runtime_orchestration_not_reimplemented", "legacy_builtin_becomes_managed_custom_asset"]}) agent_by_id = {x["source"]["id"]: x for x in agents} templates = _read(db, "agentprompttemplate", TEMPLATE_COLUMNS) for template in templates: _source_flags(template) _owner(template, identities) template_ids = {t["id"] for t in templates} for agent in agents: source = agent["source"] if source["prompt_template_id"] is not None and source["prompt_template_id"] not in template_ids: raise ValueError("Legacy Agent references a missing prompt template") if any(a not in source_agents for a in source["fusion_agent_ids"]): raise ValueError("Legacy Agent references a missing fusion Agent") if any(agent_by_id[a]["source"]["role"] != "production" for a in source["fusion_agent_ids"]): raise ValueError("Legacy Agent fusion target is not a production Agent") mapped_bindings = [] for source in bindings: agent, skill = agent_by_id[source["agent_id"]]["target"], skill_by_id[source["skill_id"]]["target"] mapped_bindings.append({"source": source, "agent_id": agent["id"], "skill_id": skill["id"], "effective": _binding_effective(agent["created_by"], skill), "review": _binding_review(agent, skill)}) # Later identity imports may include deleted users who own none of these assets. owners = {"1"} | {str(source["owner_user_id"] or 1) for source in sources + templates} owners.update(str(item["source"]["owner_user_id"] or 1) for item in skills) identities = {key: value for key, value in identities.items() if key in owners} result = {"schema_version": 1, "source": SOURCE, "target_database": TARGET_DATABASE, "target_tenant_id": TARGET_TENANT, "source_organization_id": SOURCE_ORGANIZATION, "identity_mapping": identities, "kb_mapping": kbs, "model_mapping": model_mapping, "agents": agents, "skills": skills, "bindings": mapped_bindings, "prompt_templates": templates, "grants": grants, "access_policy": SOURCE_POLICY, "repeat_policy": "exact_ledger_replay_preserves_native_edits_and_deletion", "unsupported_fields": ["prompt_template_id", "allowed_tools", "rules_version", "builtin", "created_at", "runtime_snapshot", "persona_json", "digital_human_id", "kb_category_ids"], "ledgers": {"assets": LEDGER, "bindings": BINDING_LEDGER}} _validate_plan(result) return result def _validate_plan(plan: dict) -> None: if (plan.get("schema_version") != 1 or plan.get("source") != SOURCE or plan.get("target_database") != TARGET_DATABASE or plan.get("target_tenant_id") != TARGET_TENANT or plan.get("source_organization_id") != SOURCE_ORGANIZATION or plan.get("access_policy") != SOURCE_POLICY): raise ValueError("Asset migration scope or access policy differs") identities, kbs = _mapping(plan["identity_mapping"]), _mapping(plan["kb_mapping"]) if identities.get("1") != ADMIN_TARGET_ID or any(plan["grants"].values()): raise ValueError("Asset migration identity or explicit grant policy differs") template_ids = set() for template in plan["prompt_templates"]: _source_flags(template) _owner(template, identities) if template["id"] in template_ids: raise ValueError("Duplicate legacy prompt template") template_ids.add(template["id"]) agents, skills, seen = {}, {}, set() for item in plan["skills"]: source, target = item["source"], item["target"] if target != _target_skill(source, identities) or target["id"] in seen: raise ValueError("Asset plan changes skill identity, body or access") seen.add(target["id"]) skills[source["id"]] = target source_owners = {item["source"]["id"]: _owner(item["source"], identities)[1] for item in plan["agents"]} by_agent, pairs, binding_ids = {}, set(), set() for binding in plan["bindings"]: source = binding["source"] pair = (source["agent_id"], source["skill_id"]) if (pair in pairs or source["id"] in binding_ids or source["agent_id"] not in source_owners or source["skill_id"] not in skills or binding["skill_id"] != asset_id("skill", source["skill_id"]) or binding["agent_id"] != asset_id("agent", source["agent_id"])): raise ValueError("Asset plan changes binding identity") pairs.add(pair) binding_ids.add(source["id"]) effective = _binding_effective(source_owners[source["agent_id"]], skills[source["skill_id"]]) if binding["effective"] is not effective: raise ValueError("Asset plan changes binding access disposition") if effective: by_agent.setdefault(source["agent_id"], []).append(binding["skill_id"]) for item in plan["agents"]: source, target = item["source"], item["target"] if source["prompt_template_id"] is not None and source["prompt_template_id"] not in template_ids: raise ValueError("Legacy Agent references a missing prompt template") expected, routing = _target_agent(source, identities, kbs, plan["model_mapping"], by_agent.get(source["id"], [])) if target != expected or item["model_resolution"] != routing or target["id"] in seen: raise ValueError("Asset plan changes Agent identity, body, access or routing") seen.add(target["id"]) agents[source["id"]] = target for item in plan["agents"]: if any(a not in agents for a in item["source"]["fusion_agent_ids"]): raise ValueError("Legacy Agent references a missing fusion Agent") if any(agents[a]["config"]["role"] != "production" for a in item["source"]["fusion_agent_ids"]): raise ValueError("Legacy Agent fusion target is not a production Agent") for binding in plan["bindings"]: source = binding["source"] if source["agent_id"] not in agents or binding["review"] != _binding_review(agents[source["agent_id"]], skills[source["skill_id"]]): raise ValueError("Asset plan changes binding access review") def safe_asset_summary(plan: dict) -> dict: """Return metadata only; no source body, SQL, credentials or snapshots.""" _validate_plan(plan) return {"target_database": TARGET_DATABASE, "target_tenant_id": TARGET_TENANT, "counts": {key: len(plan[key]) for key in ("agents", "skills", "bindings", "prompt_templates")}, "effective_bindings": sum(b["effective"] for b in plan["bindings"]), "blocked_bindings": sum(not b["effective"] for b in plan["bindings"]), "disabled_agent_ids": [x["source"]["id"] for x in plan["agents"] if not x["target"]["enabled"]], "disabled_skill_ids": [x["source"]["id"] for x in plan["skills"] if not x["target"]["enabled"]], "binding_reviews": [{"source_binding_id": b["source"]["id"], "review": b["review"]} for b in plan["bindings"] if b["review"]], "agent_map": {str(x["source"]["id"]): x["target"]["id"] for x in plan["agents"]}, "skill_map": {str(x["source"]["id"]): x["target"]["id"] for x in plan["skills"]}} def asset_sql(plan: dict) -> str: """Return guarded SQL without executing it or adding transaction boundaries.""" _validate_plan(plan) if not re.fullmatch(r"[a-f0-9]{64}", str(plan.get("source_snapshot_sha256", ""))): raise ValueError("Asset SQL requires a source snapshot fingerprint") payload = json.dumps(plan, ensure_ascii=True, sort_keys=True, separators=(",", ":")) literal = "'" + payload.replace("'", "''") + "'::jsonb" tag = "assets_" + hashlib.sha256(payload.encode()).hexdigest()[:16] return f"""SET LOCAL standard_conforming_strings = on; SET LOCAL log_statement = 'none'; SET LOCAL log_min_error_statement = 'panic'; DO ${tag}$ DECLARE migration jsonb := {literal}; item jsonb; target jsonb; source_data jsonb; mapping record; asset_kind text; BEGIN IF current_database() <> '{TARGET_DATABASE}' THEN RAISE EXCEPTION 'Asset target database mismatch'; END IF; PERFORM pg_advisory_xact_lock({TARGET_TENANT}, 11903); PERFORM 1 FROM tenants WHERE id={TARGET_TENANT} AND status='active' AND deleted_at IS NULL FOR SHARE; IF NOT FOUND THEN RAISE EXCEPTION 'Asset migration requires active target tenant'; END IF; LOCK TABLE custom_agents, tenant_skill_catalog IN SHARE ROW EXCLUSIVE MODE; CREATE TABLE IF NOT EXISTS {LEDGER} ( source_system text NOT NULL, source_organization_id bigint NOT NULL, source_kind text NOT NULL, source_id bigint NOT NULL, target_tenant_id integer NOT NULL, target_id varchar(36) NOT NULL, source_snapshot jsonb NOT NULL, native_snapshot jsonb NOT NULL, compatibility jsonb NOT NULL, source_snapshot_sha256 varchar(64) NOT NULL CHECK (source_snapshot_sha256 ~ '^[a-f0-9]{{64}}$'), created_at timestamptz NOT NULL DEFAULT now(), PRIMARY KEY(source_system,source_organization_id,source_kind,source_id), UNIQUE(target_tenant_id,source_kind,target_id) ); CREATE TABLE IF NOT EXISTS {BINDING_LEDGER} ( source_system text NOT NULL, source_organization_id bigint NOT NULL, source_binding_id bigint NOT NULL, target_tenant_id integer NOT NULL, target_agent_id varchar(36) NOT NULL, target_skill_id varchar(36) NOT NULL, source_snapshot jsonb NOT NULL, access_review jsonb NOT NULL, effective boolean NOT NULL, source_snapshot_sha256 varchar(64) NOT NULL CHECK (source_snapshot_sha256 ~ '^[a-f0-9]{{64}}$'), created_at timestamptz NOT NULL DEFAULT now(), PRIMARY KEY(source_system,source_organization_id,source_binding_id), UNIQUE(target_tenant_id,target_agent_id,target_skill_id) ); LOCK TABLE {LEDGER}, {BINDING_LEDGER} IN SHARE ROW EXCLUSIVE MODE; IF EXISTS (SELECT 1 FROM jsonb_each_text(migration->'identity_mapping') i LEFT JOIN {IDENTITY_LEDGER} m ON m.source_system='{SOURCE}' AND m.source_organization_id=1 AND m.source_user_id=i.key::bigint AND m.target_tenant_id={TARGET_TENANT} AND m.target_user_id=i.value LEFT JOIN users u ON u.id=i.value LEFT JOIN tenant_members tm ON tm.user_id=i.value AND tm.tenant_id={TARGET_TENANT} WHERE m.target_user_id IS NULL OR u.id IS NULL OR u.deleted_at IS NOT NULL OR tm.id IS NULL OR tm.deleted_at IS NOT NULL) THEN RAISE EXCEPTION 'Asset owner durable identity or membership mismatch'; END IF; IF EXISTS (SELECT 1 FROM jsonb_each_text(migration->'kb_mapping') i LEFT JOIN knowledge_bases k ON k.id=i.value AND k.tenant_id={TARGET_TENANT} AND k.deleted_at IS NULL WHERE k.id IS NULL) THEN RAISE EXCEPTION 'Asset knowledge base mapping missing or outside tenant'; END IF; IF EXISTS (SELECT 1 FROM jsonb_array_elements(migration->'agents') AS a(value) LEFT JOIN models m ON m.id=(a.value->'target'->'config'->>'model_id') AND m.tenant_id={TARGET_TENANT} WHERE m.id IS NULL OR m.deleted_at IS NOT NULL OR m.status <> 'active' OR m.type <> 'KnowledgeQA') THEN RAISE EXCEPTION 'Asset model mapping missing or not usable'; END IF; FOREACH asset_kind IN ARRAY ARRAY['skill','agent'] LOOP FOR item IN SELECT value FROM jsonb_array_elements(migration->(CASE asset_kind WHEN 'skill' THEN 'skills' ELSE 'agents' END)) LOOP target := item->'target'; source_data := item->'source'; SELECT * INTO mapping FROM {LEDGER} WHERE source_system='{SOURCE}' AND source_organization_id=1 AND source_kind=asset_kind AND source_id=(source_data->>'id')::bigint; IF FOUND THEN IF mapping.target_tenant_id IS DISTINCT FROM {TARGET_TENANT} OR mapping.target_id IS DISTINCT FROM target->>'id' OR mapping.source_snapshot IS DISTINCT FROM source_data OR (mapping.native_snapshot IS DISTINCT FROM target AND NOT ( asset_kind='agent' AND NOT (mapping.native_snapshot->'config' ? 'role') AND NOT (mapping.native_snapshot->'config' ? 'content_tags') AND mapping.native_snapshot = (target #- '{{config,role}}' #- '{{config,content_tags}}')) AND NOT ( asset_kind='agent' AND NOT (mapping.native_snapshot->'config' ? 'fusion_agent_ids') AND (mapping.native_snapshot = (target #- '{{config,fusion_agent_ids}}') OR ( NOT (mapping.native_snapshot->'config' ? 'role') AND NOT (mapping.native_snapshot->'config' ? 'content_tags') AND mapping.native_snapshot = (target #- '{{config,role}}' #- '{{config,content_tags}}' #- '{{config,fusion_agent_ids}}'))))) THEN RAISE EXCEPTION 'Asset durable mapping conflict'; END IF; -- A replay may not overwrite later edits or resurrect deleted resources. IF asset_kind='skill' THEN PERFORM 1 FROM tenant_skill_catalog WHERE id=target->>'id' AND tenant_id={TARGET_TENANT} AND created_by=target->>'created_by' AND instruction_only=true; ELSE PERFORM 1 FROM custom_agents WHERE id=target->>'id' AND tenant_id={TARGET_TENANT} AND created_by=target->>'created_by'; END IF; IF NOT FOUND THEN RAISE EXCEPTION 'Mapped native asset removed; refusing recreation'; END IF; ELSE IF asset_kind='skill' THEN IF EXISTS (SELECT 1 FROM tenant_skill_catalog WHERE id=target->>'id' OR (tenant_id={TARGET_TENANT} AND created_by=target->>'created_by' AND name=target->>'name' AND instruction_only=true)) THEN RAISE EXCEPTION 'Native skill ID or owner/name collision'; END IF; INSERT INTO tenant_skill_catalog(id,tenant_id,name,description,created_by,version,instructions,instruction_only, visibility,enabled,bundle_ref,bundle_sha256,created_at,updated_at) VALUES(target->>'id',{TARGET_TENANT},target->>'name',target->>'description',target->>'created_by',target->>'version', target->>'instructions',true,target->>'visibility',(target->>'enabled')::boolean,'',target->>'bundle_sha256',now(),now()); ELSE IF EXISTS (SELECT 1 FROM custom_agents WHERE id=target->>'id' OR (tenant_id={TARGET_TENANT} AND created_by=target->>'created_by' AND name=target->>'name')) THEN RAISE EXCEPTION 'Native Agent ID or owner/name collision'; END IF; INSERT INTO custom_agents(id,tenant_id,name,description,created_by,is_builtin,visibility,enabled,config,created_at,updated_at) VALUES(target->>'id',{TARGET_TENANT},target->>'name',target->>'description',target->>'created_by',false, target->>'visibility',(target->>'enabled')::boolean,target->'config',now(),now()); END IF; INSERT INTO {LEDGER}(source_system,source_organization_id,source_kind,source_id,target_tenant_id,target_id, source_snapshot,native_snapshot,compatibility,source_snapshot_sha256) VALUES('{SOURCE}',1,asset_kind,(source_data->>'id')::bigint,{TARGET_TENANT},target->>'id',source_data,target, jsonb_build_object('unsupported_fields',migration->'unsupported_fields','model_resolution',item->'model_resolution', 'access_policy',migration->'access_policy','execution',item->'compatibility'),migration->>'source_snapshot_sha256'); END IF; END LOOP; END LOOP; FOR item IN SELECT value FROM jsonb_array_elements(migration->'prompt_templates') LOOP SELECT * INTO mapping FROM {LEDGER} WHERE source_system='{SOURCE}' AND source_organization_id=1 AND source_kind='template' AND source_id=(item->>'id')::bigint; IF FOUND THEN IF mapping.target_tenant_id <> {TARGET_TENANT} OR mapping.source_snapshot IS DISTINCT FROM item THEN RAISE EXCEPTION 'Legacy prompt template provenance conflict'; END IF; ELSE INSERT INTO {LEDGER}(source_system,source_organization_id,source_kind,source_id,target_tenant_id,target_id, source_snapshot,native_snapshot,compatibility,source_snapshot_sha256) VALUES('{SOURCE}',1,'template',(item->>'id')::bigint,{TARGET_TENANT},'legacy-template-' || (item->>'id'),item,'{{}}', '{{"execution":"ledger_only_legacy_template_format_not_evaluated"}}',migration->>'source_snapshot_sha256'); END IF; END LOOP; FOR item IN SELECT value FROM jsonb_array_elements(migration->'bindings') LOOP SELECT * INTO mapping FROM {BINDING_LEDGER} WHERE source_system='{SOURCE}' AND source_organization_id=1 AND source_binding_id=(item->'source'->>'id')::bigint; IF FOUND THEN IF mapping.target_tenant_id <> {TARGET_TENANT} OR mapping.target_agent_id IS DISTINCT FROM item->>'agent_id' OR mapping.target_skill_id IS DISTINCT FROM item->>'skill_id' OR mapping.source_snapshot IS DISTINCT FROM item->'source' OR mapping.access_review IS DISTINCT FROM item->'review' OR mapping.effective IS DISTINCT FROM (item->>'effective')::boolean THEN RAISE EXCEPTION 'Legacy skill binding provenance conflict'; END IF; ELSE INSERT INTO {BINDING_LEDGER}(source_system,source_organization_id,source_binding_id,target_tenant_id,target_agent_id, target_skill_id,source_snapshot,access_review,effective,source_snapshot_sha256) VALUES('{SOURCE}',1,(item->'source'->>'id')::bigint,{TARGET_TENANT},item->>'agent_id',item->>'skill_id', item->'source',item->'review',(item->>'effective')::boolean,migration->>'source_snapshot_sha256'); END IF; END LOOP; END ${tag}$; """