feat(migration): add generic legacy extension support pipeline

core/extension_support_pipeline.py

@@ -0,0 +1,185 @@
+"""Forward-engineered support pipeline for legacy Joomla extensions.
+
+Unsupported component/plugin/module/package families are advanced through a
+bounded adapter pipeline:
+
+    observe -> normalize -> derive -> install -> native-upgrade -> prove -> hydrate
+
+Vendor archives remain immutable. Derived candidates are never called compatible
+until the selected target proves the required functional contract. Operator
+policy controls how far automation may proceed.
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass, asdict
+from typing import Any
+import re
+
+from . import compat_bridge_forge as bridge
+from . import unified_migration as um
+
+MODES = {"strict", "assisted", "auto"}
+
+
+@dataclass(frozen=True)
+class SupportProfile:
+    element: str
+    name: str
+    family_stems: tuple[str, ...] = ()
+    required_intents: tuple[str, ...] = ("http.route.dispatch",)
+    notes: tuple[str, ...] = ()
+
+
+PROFILES: dict[str, SupportProfile] = {
+    "com_easyblog": SupportProfile(
+        element="com_easyblog",
+        name="EasyBlog",
+        family_stems=("easyblog",),
+        required_intents=("http.route.dispatch", "administrator.route.dispatch"),
+        notes=(
+            "Treat the captured vendor package as immutable input.",
+            "On a Joomla 6 target, prefer a derived compatibility candidate plus runtime proof over raw table hydration.",
+            "EasyBlog data remains quarantined until target code/schema is proven ready.",
+        ),
+    ),
+}
+
+
+def normalize_mode(value: Any) -> str:
+    mode = str(value or "assisted").strip().lower()
+    return mode if mode in MODES else "assisted"
+
+
+def _version_tuple(raw: str) -> tuple[int, ...]:
+    nums = re.findall(r"\d+", str(raw or ""))
+    return tuple(int(x) for x in nums[:4]) or (0,)
+
+
+def _major(raw: str) -> int:
+    return _version_tuple(raw)[0]
+
+
+def _extension_row(site: str, element: str) -> dict | None:
+    rows, _ = um._extension_rows(site)
+    lowered = str(element or "").lower()
+    for row in rows:
+        if str(row.get("element") or "").lower() == lowered:
+            manifest = um._manifest(row.get("manifest_cache"))
+            return {
+                "element": row.get("element") or "",
+                "name": row.get("name") or row.get("element") or "",
+                "type": row.get("type") or "",
+                "folder": row.get("folder") or "",
+                "version": str(manifest.get("version") or ""),
+                "author": str(manifest.get("author") or ""),
+            }
+    return None
+
+
+def coordinate(source: str, target: str, element: str) -> dict:
+    src_rows, _ = um._extension_rows(source)
+    dst_rows, _ = um._extension_rows(target)
+    return {
+        "source": source,
+        "target": target,
+        "element": element,
+        "source_joomla": um._joomla_version(src_rows),
+        "target_joomla": um._joomla_version(dst_rows),
+        "source_extension": _extension_row(source, element),
+        "target_extension": _extension_row(target, element),
+    }
+
+
+def profile_for(element: str) -> dict:
+    profile = PROFILES.get(str(element or "").lower())
+    if profile:
+        return asdict(profile)
+    stem = re.sub(r"^(?:com_|plg_|mod_|pkg_|file_)", "", str(element or "").lower())
+    return asdict(SupportProfile(element=str(element or ""), name=str(element or "extension"), family_stems=(stem,) if stem else ()))
+
+
+def plan(source: str, target: str, element: str, *, mode: str = "assisted") -> dict:
+    mode = normalize_mode(mode)
+    coord = coordinate(source, target, element)
+    profile = profile_for(element)
+    src = coord.get("source_extension") or {}
+    dst = coord.get("target_extension") or {}
+    target_major = _major(coord.get("target_joomla") or "")
+    try:
+        forge_plan = bridge.plan(source, target, element)
+    except Exception as exc:
+        forge_plan = {"can_forge": False, "verified": False, "known_breaks": 0, "automatic_candidates": [], "semantic_ports": [], "error": f"{type(exc).__name__}: {str(exc)[:320]}"}
+
+    target_present = bool(dst)
+    source_present = bool(src)
+    verified = bool(forge_plan.get("verified"))
+    automatic = list(forge_plan.get("automatic_candidates") or [])
+    semantic = list(forge_plan.get("semantic_ports") or [])
+    captured = bool((forge_plan.get("package") or {}).get("available"))
+
+    if verified:
+        state, next_action = "verified", "release_data_family"
+    elif target_present and not forge_plan.get("known_breaks"):
+        state, next_action = "target_present_unproven", "runtime_verify"
+    elif semantic:
+        state, next_action = "semantic_port_required", "operator_or_adapter_port"
+    elif captured and automatic:
+        state, next_action = "bridge_ready", "forge_candidate"
+    elif captured and target_major >= 6:
+        state, next_action = "runtime_proof_required", "scan_and_probe"
+    elif not captured:
+        state, next_action = "package_required", "capture_authorized_package"
+    else:
+        state, next_action = "inspect", "inspect_runtime"
+
+    if str(element).lower() == "com_easyblog" and target_major >= 6 and captured and not semantic and not verified:
+        next_action = "forge_install_upgrade_verify"
+
+    allowed = {
+        "scan": True,
+        "deterministic_transform": mode in {"assisted", "auto"},
+        "install_derived_candidate": mode == "auto",
+        "native_upgrade_hooks": mode == "auto",
+        "release_data_family": verified,
+    }
+    approvals = []
+    if mode == "strict" and next_action in {"forge_candidate", "forge_install_upgrade_verify"}:
+        approvals.append("allow deterministic compatibility transforms")
+    if mode in {"strict", "assisted"} and next_action in {"forge_candidate", "runtime_verify", "forge_install_upgrade_verify", "scan_and_probe"}:
+        approvals.append("approve target installation/runtime verification")
+    if semantic:
+        approvals.append("semantic port required; no automatic source rewrite")
+
+    stages = [
+        {"id": "observe", "status": "done" if source_present else "blocked", "detail": "source extension/runtime inventory"},
+        {"id": "normalize", "status": "ready" if source_present else "blocked", "detail": "normalize extension family + migration intent"},
+        {"id": "derive", "status": "done" if verified else ("ready" if captured else "blocked"), "detail": "preserve vendor archive; derive compatibility candidate only"},
+        {"id": "install", "status": "done" if verified else ("ready" if allowed["install_derived_candidate"] else "approval"), "detail": "install through Joomla administrator installer"},
+        {"id": "upgrade", "status": "done" if verified else ("ready" if allowed["native_upgrade_hooks"] else "approval"), "detail": "allow extension/Joomla installer schema and upgrade hooks to establish target-owned structure"},
+        {"id": "prove", "status": "done" if verified else "pending", "detail": "frontend/admin runtime probes on selected target"},
+        {"id": "hydrate", "status": "ready" if verified else "blocked", "detail": "release extension data only after target support proof"},
+    ]
+    return {"coordinate": coord, "profile": profile, "mode": mode, "state": state, "next_action": next_action, "allowed": allowed, "approvals_required": approvals, "stages": stages, "forge": forge_plan, "contract": "observe -> normalize -> derive -> install -> native-upgrade -> prove -> hydrate"}
+
+
+def execute(source: str, target: str, element: str, *, mode: str = "assisted", source_username: str = "", source_password: str = "", target_username: str = "", target_password: str = "") -> dict:
+    p = plan(source, target, element, mode=mode)
+    mode = p["mode"]
+    if p["state"] == "verified":
+        return {**p, "executed": [], "ready_for_data": True}
+    if p.get("forge", {}).get("semantic_ports"):
+        return {**p, "executed": [], "ready_for_data": False, "stop_reason": "semantic_port_required"}
+    if not p.get("forge", {}).get("package", {}).get("available"):
+        return {**p, "executed": [], "ready_for_data": False, "stop_reason": "authorized_source_package_required"}
+    executed = []
+    if p["allowed"]["deterministic_transform"]:
+        candidate = bridge.forge(source, target, element)
+        executed.append({"stage": "derive", "result": candidate})
+    else:
+        return {**p, "executed": executed, "ready_for_data": False, "stop_reason": "operator_approval_required"}
+    if not p["allowed"]["install_derived_candidate"]:
+        return {**plan(source, target, element, mode=mode), "executed": executed, "ready_for_data": False, "stop_reason": "target_install_approval_required"}
+    proof = bridge.verify(source, target, element, source_username=source_username, source_password=source_password, target_username=target_username, target_password=target_password)
+    executed.append({"stage": "install_upgrade_prove", "result": proof})
+    final = plan(source, target, element, mode=mode)
+    return {**final, "executed": executed, "ready_for_data": bool(proof.get("verified"))}