yawan
f7fc992f57
Anydoc / anydoc parser engine (push) Canceled after 0s
App / Format, vet, test, and build (push) Canceled after 0s
CLI / test (macos-latest / go 1.26) (push) Canceled after 0s
CLI / test (ubuntu-latest / go 1.26) (push) Canceled after 0s
CLI / test (windows-latest / go 1.26) (push) Canceled after 0s
Build and Push Docker Image / build-ui (push) Canceled after 0s
Build and Push Docker Image / build-docreader (push) Canceled after 0s
Build and Push Docker Image / build-sandbox (push) Canceled after 0s
Build and Push Docker Image / build-app (amd64, linux/amd64, ubuntu-latest) (push) Canceled after 0s
Build and Push Docker Image / build-app (arm64, linux/arm64, ubuntu-24.04-arm) (push) Canceled after 0s
DocReader / Python tests and Go client tests (push) Canceled after 0s
Build & Publish (dsh-weknora) / Typecheck, test, pack (push) Canceled after 0s
Build & Publish (dsh-weknora) / End-to-end inside dsh (pinned) (push) Canceled after 0s
Build & Publish (dsh-weknora) / End-to-end inside dsh (latest, advisory) (push) Canceled after 0s
Frontend / Test, type-check, and build (push) Canceled after 0s
Go Lint Cache / warm lint cache (.) (push) Canceled after 0s
Go Lint Cache / warm lint cache (cli) (push) Canceled after 0s
Go Lint Cache / warm lint cache (client) (push) Canceled after 0s
Build & Publish (mcp-server) / Test (Python 3.10) (push) Canceled after 0s
Build & Publish (mcp-server) / Test (Python 3.11) (push) Canceled after 0s
Build & Publish (mcp-server) / Test (Python 3.12) (push) Canceled after 0s
Build & Publish (mcp-server) / Test (Python 3.13) (push) Canceled after 0s
Build and Push Docker Image / merge (push) Canceled after 0s
Build & Publish (dsh-weknora) / Publish to npm (push) Canceled after 0s
Build & Publish (mcp-server) / Build sdist + wheel (push) Canceled after 0s
Build & Publish (mcp-server) / Publish to PyPI (push) Canceled after 0s
470 lines
28 KiB
Python
470 lines
28 KiB
Python
"""Plan historical workbench resources and emit acceptance-only, insert-only SQL.
|
|
|
|
The caller supplies a consistent_snapshot() connection and durable identity,
|
|
Agent and knowledge-base mappings. No target SQL or model calls are executed.
|
|
Plans contain private historical text. Only safe_business_summary is reportable.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import copy
|
|
import hashlib
|
|
import json
|
|
import re
|
|
import sqlite3
|
|
import uuid
|
|
from pathlib import Path
|
|
|
|
from weknora_asset_etl import LEDGER as ASSET_LEDGER, _mapping
|
|
from weknora_identity_etl import ADMIN_TARGET_ID, IDENTITY_LEDGER
|
|
|
|
SOURCE = "yawan-workbench"
|
|
TARGET_DATABASE = "yawan_native_acceptance_20260915"
|
|
TARGET_TENANT = 10059
|
|
LEDGER = "yawan_legacy_business_mappings"
|
|
NAMESPACE = uuid.UUID("5eb4773e-f972-4cfa-b6aa-84118b82e7c2")
|
|
RESOURCE_KINDS = {"canvas": "flows", "trainingmap": "training-maps", "campaign": "content-tasks"}
|
|
HISTORY_TABLES = ("segment", "qcreport", "campaignrun", "runevent", "agentsession", "artifact", "llmcall",
|
|
"taskqueueitem", "canvasgrant", "resourcerevision", "resourcefork")
|
|
JSON_FIELDS = {
|
|
"canvas": ("nodes", "edges"),
|
|
"trainingmap": ("kb_ids", "kb_category_ids", "plan", "evidence", "coverage", "maps"),
|
|
"campaign": ("canvas_snapshot", "task_spec", "pending_confirms", "notes"),
|
|
"segment": ("generation_history", "fusion_agent_ids", "runtime_snapshot", "anchor_card", "qc_reports",
|
|
"failure", "references", "kb_category_ids", "kb_category_recommendation"),
|
|
"qcreport": ("checks", "runtime_snapshot"),
|
|
"campaignrun": ("checkpoint", "config_snapshot"), "runevent": ("payload",),
|
|
"agentsession": ("messages", "tool_state", "result_meta"),
|
|
"artifact": ("compose_report", "runtime_snapshot"),
|
|
"taskqueueitem": ("payload",), "resourcerevision": ("payload",),
|
|
}
|
|
POLICY = "historical_read_only_no_jobs_no_events_no_inferred_grants"
|
|
|
|
|
|
def business_id(kind: str, source_id: int, suffix: str = "") -> str:
|
|
if kind not in (*RESOURCE_KINDS, *HISTORY_TABLES) or type(source_id) is not int or source_id <= 0:
|
|
raise ValueError("Invalid historical business identity")
|
|
return str(uuid.uuid5(NAMESPACE, f"{SOURCE}:tenant:{TARGET_TENANT}:organization:1:{kind}:{source_id}:{suffix}"))
|
|
|
|
|
|
def _read(db: sqlite3.Connection, table: str) -> list[dict]:
|
|
if table not in (*RESOURCE_KINDS, *HISTORY_TABLES):
|
|
raise ValueError("Unreviewed source table")
|
|
query = db.execute(f'SELECT * FROM "{table}" ORDER BY id')
|
|
columns = [c[0] for c in query.description]
|
|
result = []
|
|
for values in query:
|
|
row = dict(zip(columns, values))
|
|
for field in JSON_FIELDS.get(table, ()):
|
|
if field in row and row[field] is not None:
|
|
row[field] = json.loads(row[field])
|
|
if type(row.get("id")) is not int or row["id"] <= 0:
|
|
raise ValueError("Invalid source row identity")
|
|
result.append(row)
|
|
return result
|
|
|
|
|
|
def _canonical(value) -> str:
|
|
return json.dumps(value, ensure_ascii=True, sort_keys=True, separators=(",", ":"), allow_nan=False)
|
|
|
|
|
|
def archive_text(value):
|
|
"""Losslessly retain SQLite text that PostgreSQL JSONB cannot represent."""
|
|
if isinstance(value, str) and any(c == "\x00" or 0xd800 <= ord(c) <= 0xdfff for c in value):
|
|
return {"legacy_text_encoding": "utf-8-surrogatepass/base64",
|
|
"body": base64.b64encode(value.encode("utf-8", errors="surrogatepass")).decode("ascii")}
|
|
if isinstance(value, dict):
|
|
return {k: archive_text(v) for k, v in value.items()}
|
|
if isinstance(value, list):
|
|
return [archive_text(v) for v in value]
|
|
return value
|
|
|
|
|
|
def postgres_plan(plan):
|
|
result = copy.deepcopy(plan)
|
|
for item in [*result["resources"], *result["history"]]:
|
|
item["source"] = archive_text(item["source"])
|
|
if "target" in item:
|
|
data = item["target"]["data"]
|
|
data["legacy_history"] = archive_text(data["legacy_history"])
|
|
return result
|
|
|
|
|
|
def _owner(source: dict, identities: dict) -> str:
|
|
if source.get("organization_id") != 1 or type(source.get("owner_user_id")) is not int:
|
|
raise ValueError("Business owner organization differs")
|
|
owner = source["owner_user_id"] or 1
|
|
if str(owner) not in identities:
|
|
raise ValueError(f"Business owner {owner} has no durable identity mapping")
|
|
return identities[str(owner)]
|
|
|
|
|
|
def _resolve(mapping: dict, source_id, missing: list[str], field: str) -> str:
|
|
if source_id in (0, None, ""):
|
|
missing.append(field)
|
|
return ""
|
|
if type(source_id) is not int or source_id < 1:
|
|
raise ValueError("Malformed historical foreign identity")
|
|
if str(source_id) not in mapping:
|
|
missing.append(field)
|
|
return ""
|
|
return mapping[str(source_id)]
|
|
|
|
|
|
def _marker(kind: str, source: dict, missing: list[str]) -> dict:
|
|
return {"source_system": SOURCE, "source_kind": kind, "source_id": source["id"],
|
|
"source_status": source.get("status", ""), "source_owner_user_id": source["owner_user_id"],
|
|
"source_visibility": source.get("visibility", "private"),
|
|
"read_only": True, "requires_review": True, "needs_configuration": bool(missing),
|
|
"missing_configuration": sorted(set(missing)),
|
|
"review_reasons": ["legacy_execution_contract_not_equivalent", "historical_import_does_not_resume_ai"]}
|
|
|
|
|
|
def _file(path: str, root: Path) -> dict:
|
|
resolved = Path(path).resolve()
|
|
if not resolved.is_relative_to(root.resolve()) or resolved.suffix.lower() not in (".md", ".docx"):
|
|
raise ValueError("Historical artifact path is outside the allowed artifacts directory")
|
|
if not resolved.is_file():
|
|
return {"source_path": path, "status": "missing"}
|
|
before = resolved.stat()
|
|
if before.st_size > 20 * 1024 * 1024:
|
|
raise ValueError("Historical artifact exceeds reviewed size limit")
|
|
body = resolved.read_bytes()
|
|
after = resolved.stat()
|
|
if (before.st_size, before.st_mtime_ns) != (after.st_size, after.st_mtime_ns):
|
|
raise ValueError("Historical artifact changed during read")
|
|
encoding = "utf-8" if resolved.suffix.lower() == ".md" else "base64"
|
|
return {"source_path": path, "status": "preserved", "size": len(body), "sha256": hashlib.sha256(body).hexdigest(),
|
|
"encoding": encoding, "body": body.decode("utf-8") if encoding == "utf-8" else base64.b64encode(body).decode("ascii")}
|
|
|
|
|
|
def _tree(source: dict, source_id: int, path: str) -> dict:
|
|
if not isinstance(source, dict) or not isinstance(source.get("title"), str):
|
|
raise ValueError("Historical map tree is malformed")
|
|
children = source.get("children", [])
|
|
if not isinstance(children, list):
|
|
raise ValueError("Historical map children are malformed")
|
|
return {"id": business_id("trainingmap", source_id, path), "label": source["title"],
|
|
"children": [_tree(n, source_id, f"{path}.{i}") for i, n in enumerate(children)]}
|
|
|
|
|
|
def _framework(source: dict, depth=1) -> str:
|
|
parts = ["#" * min(depth, 6) + " " + source["title"]]
|
|
for field in ("content", "hook"):
|
|
if source.get(field):
|
|
parts.append(source[field])
|
|
parts.extend(_framework(child, depth + 1) for child in source.get("children", []))
|
|
return "\n\n".join(parts)
|
|
|
|
|
|
def _review(reports: list[dict]) -> str:
|
|
# Formatting exposes original verdicts and checks; it does not run or pass QC.
|
|
return "\n\n".join("```json\n" + json.dumps({k: r.get(k) for k in ("id", "gate", "verdict", "checks", "created_at")},
|
|
ensure_ascii=False, indent=2) + "\n```" for r in reports)
|
|
|
|
|
|
def _target(kind: str, source: dict, plan: dict) -> dict:
|
|
identities, agents, kbs = (plan[k] for k in ("identity_mapping", "agent_mapping", "kb_mapping"))
|
|
missing = []
|
|
if kind == "canvas":
|
|
nodes, edges = [], []
|
|
seen = set()
|
|
for node in source["nodes"]:
|
|
nid = node.get("node_id")
|
|
if not isinstance(nid, str) or not nid or nid in seen:
|
|
raise ValueError("Canvas node identity is missing or duplicated")
|
|
seen.add(nid)
|
|
nodes.append({"id": nid, "agent_id": _resolve(agents, node.get("agent_id"), missing, "node_agent"),
|
|
"label": node.get("label", nid)})
|
|
for edge in source["edges"]:
|
|
start, end = edge.get("from", edge.get("from_node")), edge.get("to", edge.get("to_node"))
|
|
if start not in seen or end not in seen:
|
|
raise ValueError("Canvas edge has a missing node")
|
|
edges.append({"source": start, "target": end})
|
|
if not nodes:
|
|
missing.append("flow_nodes")
|
|
data = {"nodes": nodes, "edges": edges, "description": "",
|
|
"legacy_history": {"nodes": source["nodes"], "edges": source["edges"],
|
|
"supervisor_agent_id": source.get("supervisor_agent_id"),
|
|
"supervisor_mode": source.get("supervisor_mode")}}
|
|
elif kind == "trainingmap":
|
|
days = []
|
|
candidates = source["plan"].get("days", [])
|
|
if not isinstance(candidates, list):
|
|
raise ValueError("Historical training plan days must be an array")
|
|
for candidate in candidates:
|
|
day = candidate["day"]
|
|
if type(day) is not int or day != len(days) + 1:
|
|
raise ValueError("Historical training plan day ordering differs")
|
|
generated = source["maps"].get(str(day))
|
|
nodes = [_tree(n, source["id"], f"{day}.{i}") for i, n in enumerate(generated.get("children", []))] if generated else []
|
|
days.append({"day": day, "title": candidate["title"], "prerequisite": candidate.get("prerequisite", ""),
|
|
"hook": generated.get("hook", "") if generated else "", "nodes": nodes,
|
|
"framework": _framework(generated) if generated else "", "confirmed": False})
|
|
count = source["plan"].get("recommended_days", source.get("requested_days") or 0)
|
|
if count != len(days):
|
|
raise ValueError("Historical training plan day count differs")
|
|
kb_ids = [_resolve(kbs, k, missing, "knowledge_base") for k in source["kb_ids"]]
|
|
if source["kb_category_ids"]:
|
|
missing.append("knowledge_category_scope")
|
|
data = {"topic": source["topic"], "days_count": count,
|
|
"agent_id": _resolve(agents, source.get("agent_id"), missing, "agent_id"),
|
|
"knowledge_base_ids": [k for k in kb_ids if k], "knowledge_ids": [], "days": days,
|
|
"legacy_history": {k: source[k] for k in ("plan", "maps", "evidence", "coverage", "error", "confirmed_version")}}
|
|
else:
|
|
# Legacy campaigns contain a frozen canvas and free-form map text, not
|
|
# reliable foreign keys to a map/day/live canvas. Never infer by title.
|
|
missing.extend(["map_id", "map_day", "flow_id", "task_agent_id"])
|
|
segments = []
|
|
for item in plan["history"]:
|
|
s = item["source"]
|
|
if item["source_kind"] != "segment" or s["campaign_id"] != source["id"]:
|
|
continue
|
|
reports = [r["source"] for r in plan["history"] if r["source_kind"] == "qcreport" and r["source"]["segment_id"] == s["id"] and not r["relation_review"]]
|
|
artifact = plan["artifacts"].get(s["artifact_path"], {})
|
|
if s["artifact_path"] and artifact.get("status") != "preserved":
|
|
missing.append("segment_artifact")
|
|
segments.append({"id": business_id("segment", s["id"]), "title": s["title"],
|
|
"agent_id": _resolve(agents, s["agent_id"], missing, "segment_agent"),
|
|
"status": s["status"], "content": artifact.get("body", "") if artifact.get("encoding") == "utf-8" else "",
|
|
"review": _review(reports), "source_ids": []})
|
|
cross = [r["source"] for r in plan["history"] if r["source_kind"] == "qcreport"
|
|
and r["parent_resource_id"] == business_id(kind, source["id"]) and r["source"]["gate"] == "cross" and not r["relation_review"]]
|
|
data = {"map_id": "", "day": 0, "flow_id": "", "agent_id": "", "knowledge_base_ids": [],
|
|
"knowledge_ids": [], "segments": segments, "review": _review(cross), "composed": "",
|
|
"legacy_history": {"mindmap_text": source["mindmap_text"], "canvas_snapshot": source["canvas_snapshot"],
|
|
"canvas_name": source["canvas_name"], "task_spec": source["task_spec"],
|
|
"pending_confirms": source["pending_confirms"], "notes": source["notes"],
|
|
"segments": [{k: s["source"][k] for k in ("id", "node_id", "time_range", "status", "round", "references", "failure", "generation_history")}
|
|
for s in plan["history"] if s["source_kind"] == "segment" and s["source"]["campaign_id"] == source["id"]]}}
|
|
data["legacy_import"] = _marker(kind, source, missing)
|
|
name = source["topic"] if kind == "trainingmap" else source["name"]
|
|
if not isinstance(name, str) or not name.strip() or "\x00" in name:
|
|
raise ValueError("Historical resource name is invalid")
|
|
return {"id": business_id(kind, source["id"]), "tenant_id": TARGET_TENANT,
|
|
"owner_id": _owner(source, identities), "owner_name": "", "kind": RESOURCE_KINDS[kind],
|
|
"name": name, "status": source.get("status", "draft"), "version": source.get("version", 1),
|
|
"created_at": source.get("created_at") or None, "updated_at": source.get("finished_at") or source.get("created_at") or None,
|
|
"data": data}
|
|
|
|
|
|
def business_plan(db: sqlite3.Connection, identity_mapping: dict, agent_mapping: dict, kb_mapping: dict,
|
|
artifact_root: Path) -> dict:
|
|
"""Read an online-backup snapshot; never pass the live SQLite connection."""
|
|
if db.execute("PRAGMA query_only").fetchone()[0] != 1:
|
|
raise ValueError("Business plan requires the read-only consistent snapshot")
|
|
identities, agents, kbs = (_mapping(m) for m in (identity_mapping, agent_mapping, kb_mapping))
|
|
if identities.get("1") != ADMIN_TARGET_ID:
|
|
raise ValueError("Pinned source administrator identity differs")
|
|
sources = {table: _read(db, table) for table in RESOURCE_KINDS}
|
|
campaigns = {r["id"] for r in sources["campaign"]}
|
|
canvases = {r["id"] for r in sources["canvas"]}
|
|
history = {table: _read(db, table) for table in HISTORY_TABLES}
|
|
segments = {r["id"]: r["campaign_id"] for r in history["segment"]}
|
|
if any(cid not in campaigns for cid in segments.values()):
|
|
raise ValueError("Historical segment has a missing campaign")
|
|
runs = {r["id"]: r["campaign_id"] for r in history["campaignrun"]}
|
|
plan = {"schema_version": 1, "source": SOURCE, "target_database": TARGET_DATABASE,
|
|
"target_tenant_id": TARGET_TENANT, "identity_mapping": identities, "agent_mapping": agents,
|
|
"kb_mapping": kbs, "resources": [], "history": [], "artifacts": {}, "policy": POLICY,
|
|
"excluded_history_counts": {},
|
|
"timestamp_policy": "source_timestamp_else_import_time", "mapping_ledger": LEDGER}
|
|
for table, rows in history.items():
|
|
for row in rows:
|
|
parent = ""
|
|
relation_review = []
|
|
if table in ("canvasgrant", "resourcerevision", "resourcefork"):
|
|
if table != "canvasgrant" and row["resource_type"] != "canvas":
|
|
continue
|
|
cid = row.get("canvas_id", row.get("resource_id", row.get("source_id")))
|
|
if cid not in canvases:
|
|
raise ValueError("Canvas history has no source resource")
|
|
parent = business_id("canvas", cid)
|
|
else:
|
|
cid = row.get("campaign_id")
|
|
if table == "qcreport":
|
|
cid = segments.get(row["segment_id"])
|
|
if table == "taskqueueitem":
|
|
payload = row["payload"]
|
|
cid = payload.get("campaign_id") or runs.get(payload.get("run_id"))
|
|
if not cid:
|
|
continue
|
|
if not cid and row.get("run_id"):
|
|
cid = runs.get(row["run_id"])
|
|
if cid not in campaigns:
|
|
if table in ("llmcall", "agentsession", "taskqueueitem"):
|
|
plan["excluded_history_counts"][table] = plan["excluded_history_counts"].get(table, 0) + 1
|
|
continue
|
|
raise ValueError(f"Historical {table} {row['id']} has no source campaign")
|
|
if row.get("segment_id") and segments.get(row["segment_id"]) != cid:
|
|
relation_review.append("source_segment_campaign_mismatch_ledger_only")
|
|
if row.get("run_id") and runs.get(row["run_id"]) != cid:
|
|
relation_review.append("source_run_campaign_mismatch_ledger_only")
|
|
parent = business_id("campaign", cid)
|
|
if row.get("organization_id", 1) != 1:
|
|
raise ValueError("Historical row belongs to another organization")
|
|
plan["history"].append({"source_kind": table, "source_id": row["id"], "source": row,
|
|
"target_id": business_id(table, row["id"]), "parent_resource_id": parent,
|
|
"relation_review": relation_review})
|
|
paths = []
|
|
if table == "segment":
|
|
paths = [row["artifact_path"], *(v.get("artifact_path", "") for v in row["generation_history"])]
|
|
elif table == "artifact":
|
|
paths = [row["docx_path"]]
|
|
for path in paths:
|
|
if path and path not in plan["artifacts"]:
|
|
plan["artifacts"][path] = _file(path, artifact_root)
|
|
for kind, rows in sources.items():
|
|
for source in rows:
|
|
plan["resources"].append({"source_kind": kind, "source_id": source["id"], "source": source,
|
|
"target": _target(kind, source, plan)})
|
|
_validate_plan(plan)
|
|
return plan
|
|
|
|
|
|
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("policy") != POLICY or plan.get("mapping_ledger") != LEDGER):
|
|
raise ValueError("Historical migration scope differs")
|
|
for key in ("identity_mapping", "agent_mapping", "kb_mapping"):
|
|
_mapping(plan[key])
|
|
if plan["identity_mapping"].get("1") != ADMIN_TARGET_ID:
|
|
raise ValueError("Pinned source administrator identity differs")
|
|
parents = {business_id(r["source_kind"], r["source_id"]) for r in plan["resources"]}
|
|
seen = set()
|
|
for item in [*plan["resources"], *plan["history"]]:
|
|
key = (item["source_kind"], item["source_id"])
|
|
if key in seen or item["source"]["id"] != item["source_id"]:
|
|
raise ValueError("Historical identity is duplicated or changed")
|
|
seen.add(key)
|
|
if "target" in item:
|
|
if item["target"] != _target(item["source_kind"], item["source"], plan):
|
|
raise ValueError("Historical native projection differs")
|
|
elif (item["target_id"] != business_id(*key) or item["parent_resource_id"] not in parents):
|
|
raise ValueError("Historical parent mapping differs")
|
|
for path, artifact in plan["artifacts"].items():
|
|
if artifact["source_path"] != path or artifact["status"] not in ("preserved", "missing"):
|
|
raise ValueError("Historical artifact identity differs")
|
|
if artifact["status"] == "preserved":
|
|
raw = artifact["body"].encode("utf-8") if artifact["encoding"] == "utf-8" else base64.b64decode(artifact["body"], validate=True)
|
|
if len(raw) != artifact["size"] or hashlib.sha256(raw).hexdigest() != artifact["sha256"]:
|
|
raise ValueError("Historical artifact fingerprint differs")
|
|
|
|
|
|
def safe_business_summary(plan: dict) -> dict:
|
|
_validate_plan(plan)
|
|
return {"target_database": TARGET_DATABASE, "target_tenant_id": TARGET_TENANT,
|
|
"counts": {kind: sum(i["source_kind"] == kind for i in [*plan["resources"], *plan["history"]])
|
|
for kind in (*RESOURCE_KINDS, *HISTORY_TABLES)},
|
|
"resource_map": {kind: {str(i["source_id"]): i["target"]["id"] for i in plan["resources"] if i["source_kind"] == kind}
|
|
for kind in RESOURCE_KINDS},
|
|
"artifacts_preserved": sum(f["status"] == "preserved" for f in plan["artifacts"].values()),
|
|
"artifacts_missing": sum(f["status"] == "missing" for f in plan["artifacts"].values()),
|
|
"historical_read_only": len(plan["resources"]), "jobs_created": 0, "events_created": 0,
|
|
"excluded_outside_campaign_scope": plan["excluded_history_counts"],
|
|
"history_relation_reviews": [{"kind": i["source_kind"], "source_id": i["source_id"], "review": i["relation_review"]}
|
|
for i in plan["history"] if i["relation_review"]],
|
|
"needs_configuration": [{"kind": i["source_kind"], "source_id": i["source_id"],
|
|
"missing": i["target"]["data"]["legacy_import"]["missing_configuration"]}
|
|
for i in plan["resources"] if i["target"]["data"]["legacy_import"]["needs_configuration"]]}
|
|
|
|
|
|
def business_sql(plan: dict) -> str:
|
|
"""Return transaction-body SQL; no target writes happen in this function."""
|
|
_validate_plan(plan)
|
|
if not re.fullmatch(r"[a-f0-9]{64}", str(plan.get("source_snapshot_sha256", ""))):
|
|
raise ValueError("Business SQL requires a source snapshot fingerprint")
|
|
payload = _canonical(postgres_plan(plan))
|
|
literal = "'" + payload.replace("'", "''") + "'::jsonb"
|
|
tag = "business_" + 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; saved record; artifact jsonb;
|
|
native_id text; parent_id text; owner_name text;
|
|
BEGIN
|
|
IF current_database() <> '{TARGET_DATABASE}' THEN RAISE EXCEPTION 'Business target database mismatch'; END IF;
|
|
PERFORM pg_advisory_xact_lock({TARGET_TENANT},11904);
|
|
PERFORM 1 FROM tenants WHERE id={TARGET_TENANT} AND status='active' AND deleted_at IS NULL FOR SHARE;
|
|
IF NOT FOUND THEN RAISE EXCEPTION 'Business migration requires active target tenant'; END IF;
|
|
LOCK TABLE workbench_resources 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,
|
|
parent_resource_id varchar(36), source_snapshot jsonb NOT NULL, native_snapshot jsonb NOT NULL,
|
|
relation_review jsonb NOT NULL DEFAULT '[]',
|
|
artifact_snapshot jsonb NOT NULL DEFAULT '{{}}', source_snapshot_sha256 varchar(64) NOT NULL,
|
|
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),
|
|
CHECK(source_snapshot_sha256 ~ '^[a-f0-9]{{64}}$')
|
|
);
|
|
LOCK TABLE {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 'Business owner durable identity or membership mismatch'; END IF;
|
|
IF EXISTS (SELECT 1 FROM jsonb_each_text(migration->'agent_mapping') i
|
|
LEFT JOIN {ASSET_LEDGER} m ON m.source_system='{SOURCE}' AND m.source_organization_id=1 AND m.source_kind='agent'
|
|
AND m.source_id=i.key::bigint AND m.target_tenant_id={TARGET_TENANT} AND m.target_id=i.value
|
|
LEFT JOIN custom_agents a ON a.id=i.value AND a.tenant_id={TARGET_TENANT}
|
|
WHERE m.target_id IS NULL OR a.id IS NULL)
|
|
THEN RAISE EXCEPTION 'Business Agent durable mapping 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 'Business knowledge base scope mismatch'; END IF;
|
|
FOR item IN SELECT value FROM jsonb_array_elements((migration->'resources') || (migration->'history')) LOOP
|
|
source_data := item->'source'; target := coalesce(item->'target','{{}}');
|
|
native_id := coalesce(target->>'id',item->>'target_id'); parent_id := item->>'parent_resource_id';
|
|
artifact := '{{}}';
|
|
IF item->>'source_kind'='segment' THEN
|
|
SELECT coalesce(jsonb_object_agg(a.key,a.value),'{{}}') INTO artifact FROM jsonb_each(migration->'artifacts') a
|
|
WHERE a.key=source_data->>'artifact_path' OR EXISTS
|
|
(SELECT 1 FROM jsonb_array_elements(source_data->'generation_history') h WHERE h->>'artifact_path'=a.key);
|
|
ELSIF item->>'source_kind'='artifact' THEN
|
|
artifact := coalesce(migration->'artifacts'->(source_data->>'docx_path'),'{{}}');
|
|
END IF;
|
|
SELECT * INTO saved FROM {LEDGER} WHERE source_system='{SOURCE}' AND source_organization_id=1
|
|
AND source_kind=item->>'source_kind' AND source_id=(item->>'source_id')::bigint;
|
|
IF FOUND THEN
|
|
IF saved.target_tenant_id IS DISTINCT FROM {TARGET_TENANT} OR saved.target_id IS DISTINCT FROM native_id
|
|
OR saved.parent_resource_id IS DISTINCT FROM parent_id OR saved.source_snapshot IS DISTINCT FROM source_data
|
|
OR saved.native_snapshot IS DISTINCT FROM target OR saved.artifact_snapshot IS DISTINCT FROM artifact
|
|
OR saved.relation_review IS DISTINCT FROM coalesce(item->'relation_review','[]')
|
|
THEN RAISE EXCEPTION 'Business durable mapping conflict'; END IF;
|
|
IF item ? 'target' THEN
|
|
PERFORM 1 FROM workbench_resources WHERE id=native_id AND tenant_id={TARGET_TENANT}
|
|
AND owner_id=target->>'owner_id' AND kind=target->>'kind';
|
|
IF NOT FOUND THEN RAISE EXCEPTION 'Mapped business resource removed; refusing recreation'; END IF;
|
|
END IF;
|
|
ELSE
|
|
IF item ? 'target' THEN
|
|
IF EXISTS (SELECT 1 FROM workbench_resources r WHERE r.id=native_id OR
|
|
(r.tenant_id={TARGET_TENANT} AND r.owner_id=target->>'owner_id' AND r.kind=target->>'kind' AND r.name=target->>'name'
|
|
AND NOT EXISTS (SELECT 1 FROM jsonb_array_elements(migration->'resources') p WHERE p->'target'->>'id'=r.id)))
|
|
THEN RAISE EXCEPTION 'Native business ID or owner/name collision'; END IF;
|
|
SELECT username INTO owner_name FROM users WHERE id=target->>'owner_id';
|
|
INSERT INTO workbench_resources(id,tenant_id,owner_id,owner_name,kind,name,status,version,data,created_at,updated_at)
|
|
VALUES(native_id,{TARGET_TENANT},target->>'owner_id',owner_name,target->>'kind',target->>'name',target->>'status',
|
|
(target->>'version')::integer,target->'data',coalesce((target->>'created_at')::timestamptz,now()),
|
|
coalesce((target->>'updated_at')::timestamptz,now()));
|
|
ELSE
|
|
PERFORM 1 FROM workbench_resources WHERE id=parent_id AND tenant_id={TARGET_TENANT};
|
|
IF NOT FOUND THEN RAISE EXCEPTION 'Historical parent resource missing'; END IF;
|
|
END IF;
|
|
INSERT INTO {LEDGER}(source_system,source_organization_id,source_kind,source_id,target_tenant_id,target_id,
|
|
parent_resource_id,source_snapshot,native_snapshot,artifact_snapshot,relation_review,source_snapshot_sha256)
|
|
VALUES('{SOURCE}',1,item->>'source_kind',(item->>'source_id')::bigint,{TARGET_TENANT},native_id,
|
|
parent_id,source_data,target,artifact,coalesce(item->'relation_review','[]'),migration->>'source_snapshot_sha256');
|
|
END IF;
|
|
END LOOP;
|
|
END ${tag}$;
|
|
"""
|