feat(clone): skip unmigratable resources upfront + failures summary

- Detect frictionless adapter dependence in custom_tool phase (source
  service-account view hides frictionless adapters; profile names not
  in that view can't be migrated). Skip the tool and cascade-skip its
  workflow via a registry-id set on CloneContext.
- Detect OAuth-backed connectors (metadata carries access_token /
  refresh_token) and skip ahead of POST. Avoids the backend's
  OAuthTimeOut 408 and surfaces a re-auth instruction for the operator.
- CloneReport now prints a "Failures" section before the table with
  one truncated line per recorded error (capped at 30 rows). Entity
  labels in the section are bold-cyan to match the phase column.
- Drop "workflow_owner" from POST payloads (server-managed).

Author: chandrasekharan-zipstack

src/unstract/clone/context.py

@@ -102,3 +102,6 @@ class CloneContext:
     target: PlatformClient
     options: CloneOptions
     remap: RemapTable = field(default_factory=RemapTable)
+    # Source prompt_registry_ids whose CustomTool was skipped; used to
+    # cascade-skip dependent workflows downstream.
+    skipped_custom_tool_registry_ids: set[str] = field(default_factory=set)

src/unstract/clone/phases/base.py

@@ -33,6 +33,7 @@
         "created_at",
         "modified_at",
         "shared_users",
+        "workflow_owner",
     }
 )
 

src/unstract/clone/phases/connector.py

@@ -31,6 +31,15 @@
 
 CONNECTOR_PATH = "connector/"
 
+# Backend POST serializer trips on these keys (connector_v2/serializers.py)
+# by trying to refresh against the source user's social auth — guaranteed
+# OAuthTimeOut on target. Detect here and skip ahead of POST.
+_OAUTH_TOKEN_KEYS: frozenset[str] = frozenset({"access_token", "refresh_token"})
+
+
+def _has_oauth_tokens(metadata: dict[str, Any]) -> bool:
+    return any(metadata.get(k) for k in _OAUTH_TOKEN_KEYS)
+
 
 class ConnectorPhase(Phase):
     name = "connector"
@@ -74,7 +83,8 @@ def _clone_one(
                 result.errors.append(f"GET source detail {name}: {e}")
             return
 
-        if not src.get("connector_metadata"):
+        metadata = src.get("connector_metadata") or {}
+        if not metadata:
             logger.info(
                 "skipping connector '%s' (src=%s, catalog=%s) — source returned no metadata",
                 name,
@@ -85,6 +95,19 @@ def _clone_one(
                 result.skipped += 1
             return
 
+        if _has_oauth_tokens(metadata):
+            logger.warning(
+                "skipping connector '%s' (src=%s, catalog=%s) — OAuth-backed; "
+                "re-authorise on target after the clone, then re-run to wire "
+                "dependent workflow endpoints.",
+                name,
+                src_id,
+                src.get("connector_id"),
+            )
+            with lock:
+                result.skipped += 1
+            return
+
         try:
             existing = self.ctx.target.list_connectors(name=name)
         except Exception as e:

src/unstract/clone/phases/custom_tool.py

@@ -79,6 +79,18 @@ def run(self, report: CloneReport) -> PhaseResult:
             result.errors.append(f"list target tools: {e}")
             return result
 
+        # Source's service-account view hides frictionless adapters; a
+        # profile-referenced name missing here flags a tool we can't migrate.
+        try:
+            self._src_adapter_names = {
+                a["adapter_name"] for a in self.ctx.source.list_adapters()
+            }
+        except Exception as e:
+            logger.exception("Failed to list source adapters: %s", e)
+            result.failed += 1
+            result.errors.append(f"list source adapters for visibility check: {e}")
+            return result
+
         # Updated under lock when a fresh create lands so duplicate
         # same-name source rows adopt instead of recreating.
         target_by_name: dict[str, dict[str, Any]] = {
@@ -235,7 +247,25 @@ def _create_fresh(
             )
             return None
 
-        adapter_ids = self._resolve_target_adapter_ids(src_tool_id, tool_name)
+        default_profile = self._source_default_profile(src_tool_id, tool_name)
+        if default_profile is None:
+            with lock:
+                result.failed += 1
+                result.errors.append(
+                    f"import {tool_name}: no default profile on source"
+                )
+            return None
+
+        invisible = self._invisible_source_adapter_names(default_profile)
+        if invisible:
+            self._register_frictionless_skip(
+                src_tool_id, tool_name, invisible, result, lock
+            )
+            return None
+
+        adapter_ids = self._resolve_target_adapter_ids(
+            default_profile, tool_name
+        )
         if adapter_ids is None:
             with lock:
                 result.failed += 1
@@ -265,17 +295,9 @@ def _create_fresh(
         )
         return tgt_tool_id
 
-    def _resolve_target_adapter_ids(
+    def _source_default_profile(
         self, src_tool_id: str, tool_name: str
-    ) -> dict[str, str] | None:
-        """Source profile carries adapter NAMES (per serializer); resolve
-        each name to a target adapter UUID via ``list_adapters(name=...)``.
-
-        Returns ``None`` if any of the four required adapters can't be
-        found on target — caller fails the tool. AdapterPhase preserves
-        names across orgs so this lookup should always hit when the
-        adapter clone ran cleanly.
-        """
+    ) -> dict[str, Any] | None:
         try:
             src_profiles = self.ctx.source.list_profiles(src_tool_id)
         except Exception as e:
@@ -293,11 +315,71 @@ def _resolve_target_adapter_ids(
                 "source tool '%s' has no profiles to derive adapter ids from",
                 tool_name,
             )
-            return None
+        return default
+
+    def _invisible_source_adapter_names(
+        self, default_profile: dict[str, Any]
+    ) -> list[str]:
+        """Profile adapter names not in the source's visible adapter set
+        (typically frictionless) — these can't be migrated.
+        """
+        missing: list[str] = []
+        for src_field, _ in _PROFILE_ADAPTER_FIELDS:
+            adapter_name = _extract_adapter_name(default_profile.get(src_field))
+            if adapter_name and adapter_name not in self._src_adapter_names:
+                missing.append(adapter_name)
+        return missing
+
+    def _register_frictionless_skip(
+        self,
+        src_tool_id: str,
+        tool_name: str,
+        missing_adapters: list[str],
+        result: PhaseResult,
+        lock: threading.Lock,
+    ) -> None:
+        """Record the skip + source registry id so dependent workflows
+        cascade-skip downstream.
+        """
+        logger.warning(
+            "skipping tool '%s' src=%s — default profile references adapters "
+            "not visible to the source service account (frictionless?): %s. "
+            "Wire equivalents on target and re-run.",
+            tool_name,
+            src_tool_id,
+            missing_adapters,
+        )
+        try:
+            src_regs = self.ctx.source.list_registries(custom_tool=src_tool_id)
+        except Exception as e:
+            logger.warning(
+                "registry lookup failed for skipped tool '%s' — "
+                "downstream cascade-skip may not fire: %s",
+                tool_name,
+                e,
+            )
+            src_regs = []
+        with lock:
+            result.skipped += 1
+            for reg in src_regs:
+                reg_id = reg.get("prompt_registry_id")
+                if reg_id:
+                    self.ctx.skipped_custom_tool_registry_ids.add(reg_id)
+
+    def _resolve_target_adapter_ids(
+        self, default_profile: dict[str, Any], tool_name: str
+    ) -> dict[str, str] | None:
+        """Source profile carries adapter NAMES (per serializer); resolve
+        each name to a target adapter UUID via ``list_adapters(name=...)``.
 
+        Returns ``None`` if any of the four required adapters can't be
+        found on target — caller fails the tool. AdapterPhase preserves
+        names across orgs so this lookup should always hit when the
+        adapter clone ran cleanly.
+        """
         resolved: dict[str, str] = {}
         for src_field, form_field in _PROFILE_ADAPTER_FIELDS:
-            adapter_name = _extract_adapter_name(default.get(src_field))
+            adapter_name = _extract_adapter_name(default_profile.get(src_field))
             if not adapter_name:
                 logger.warning(
                     "source default profile for tool '%s' missing adapter '%s'",

src/unstract/clone/phases/workflow.py

@@ -50,19 +50,57 @@ def run(self, report: CloneReport) -> PhaseResult:
             result.errors.append(f"list source workflows: {e}")
             return result
 
+        # Built once so per-workflow cascade-skip checks stay O(1).
+        self._wf_to_src_tool_id = self._collect_wf_tool_map(result)
+
         logger.info("Found %d workflow(s) in source org", len(src_workflows))
         self.parallel_map(
             src_workflows,
             lambda src, lock: self._clone_one(src, result, lock),
         )
         return result
 
+    def _collect_wf_tool_map(self, result: PhaseResult) -> dict[str, str]:
+        """Map source workflow_id to its ToolInstance.tool_id; listed once
+        to avoid N+1 fetches.
+        """
+        if not self.ctx.skipped_custom_tool_registry_ids:
+            return {}
+        try:
+            tis = self.ctx.source.list_tool_instances()
+        except Exception as e:
+            logger.warning(
+                "workflow phase: failed to list source tool_instances for "
+                "cascade-skip lookup (%s); proceeding without cascade",
+                e,
+            )
+            return {}
+        mapping: dict[str, str] = {}
+        for ti in tis:
+            wf_id = ti.get("workflow")
+            tool_id = ti.get("tool_id")
+            if wf_id and tool_id:
+                mapping[wf_id] = tool_id
+        return mapping
+
     def _clone_one(
         self, src: dict[str, Any], result: PhaseResult, lock: threading.Lock
     ) -> None:
         name = src["workflow_name"]
         src_id = src["id"]
 
+        src_tool_id = self._wf_to_src_tool_id.get(src_id)
+        if src_tool_id and src_tool_id in self.ctx.skipped_custom_tool_registry_ids:
+            logger.warning(
+                "skipping workflow '%s' src=%s — its tool was skipped in "
+                "custom_tool phase (frictionless adapter dependence)",
+                name,
+                src_id,
+            )
+            with lock:
+                result.skipped += 1
+            return
+
         try:
             existing = self.ctx.target.list_workflows(name=name)
         except Exception as e:

src/unstract/clone/report.py

@@ -75,6 +75,8 @@ def render(self) -> str:
         console = Console(
             file=buf, force_terminal=True, color_system="truecolor", width=100
         )
+        # Actionable summary first so it doesn't scroll past the table.
+        self._render_failures_summary(console_print=console.print, rich=True)
         self._render_endpoints(console.print)
         table = Table(title="Clone Report", header_style="bold cyan")
         table.add_column("Phase", style="bold", justify="left")
@@ -151,7 +153,9 @@ def _fmt_duration_plain(seconds: float) -> str:
         return f"{int(mins)}m{secs:.0f}s"
 
     def _render_plain(self) -> str:
-        lines = ["Clone Report", "=" * 60]
+        lines: list[str] = []
+        self._render_failures_summary(console_print=lines.append, rich=False)
+        lines.extend(["Clone Report", "=" * 60])
         self._render_endpoints(lines.append)
         header = (
             f"{'Phase':<24}{'Created':>10}{'Adopted':>10}"
@@ -281,6 +285,47 @@ def _files_sections_plain(self) -> list[str]:
                 lines.append(f"  - {self._describe_file_row(row)}")
         return lines
 
+    # Caps so a long traceback or many failures don't dominate the report.
+    _FAILURE_LINE_MAX_CHARS = 200
+    _FAILURE_MAX_ROWS = 30
+
+    def _render_failures_summary(self, console_print: Any, rich: bool) -> None:
+        rows: list[tuple[str, str]] = []
+        for p in self.phases:
+            for err in p.errors:
+                rows.append((p.name, err))
+        if not rows:
+            return
+        header = "Failures (see WARNING/ERROR log lines above for full detail)"
+        if rich:
+            console_print(f"[red]{header}:[/red]")
+        else:
+            console_print(f"{header}:")
+        shown = rows[: self._FAILURE_MAX_ROWS]
+        for phase_name, err in shown:
+            truncated = self._truncate(err, self._FAILURE_LINE_MAX_CHARS)
+            if rich:
+                console_print(
+                    f"  - [bold cyan]{phase_name}[/bold cyan]: {truncated}",
+                    highlight=False,
+                )
+            else:
+                console_print(f"  - {phase_name}: {truncated}")
+        remaining = len(rows) - len(shown)
+        if remaining > 0:
+            tail = f"  ... +{remaining} more — see logs"
+            if rich:
+                console_print(f"[dim]{tail}[/dim]")
+            else:
+                console_print(tail)
+
+    @staticmethod
+    def _truncate(text: str, limit: int) -> str:
+        text = text.replace("\n", " ")
+        if len(text) <= limit:
+            return text
+        return text[: limit - 1] + "…"
+
     @staticmethod
     def _describe_file_row(row: dict[str, Any]) -> str:
         tool = row.get("tool_name") or row.get("tool_id") or "?"

tests/clone/test_connector_phase.py

@@ -118,6 +118,31 @@ def test_redacted_metadata_connector_skipped():
     assert ctx.remap.resolve("connector", "src-ucs") is None
 
 
+def test_oauth_connector_skipped_before_post():
+    """OAuth-backed connectors (metadata carries access_token/refresh_token)
+    would fail target POST with OAuthTimeOut — skip ahead of POST so the
+    operator re-authorises post-clone.
+    """
+    oauth = _src("src-gdrive", "Unstract's google drive")
+    oauth["connector_metadata"] = {
+        "provider": "google-oauth2",
+        "uid": "src-user",
+        "access_token": "ya29.src-access",
+        "refresh_token": "1//src-refresh",
+    }
+    src = FakeClient([oauth])
+    tgt = FakeClient()
+    ctx = _ctx(src, tgt)
+
+    result = ConnectorPhase(ctx).run(CloneReport())
+
+    assert result.skipped == 1
+    assert result.created == 0
+    assert result.failed == 0
+    assert tgt.posts == []
+    assert ctx.remap.resolve("connector", "src-gdrive") is None
+
+
 def test_idempotency_zero_creates_on_rerun():
     src = FakeClient([_src("src-a", "Prod PG")])
     tgt = FakeClient(