[FIX] Address Greptile: report all workflow skip reasons in one pass

- workflow: a workflow blocked by both a skipped tool and a skipped connector
  now surfaces both reasons instead of only the first (no extra re-run to learn
  the second blocker).
- tests: assert the tool-skip cascade emits its report warning; cover the
  dual-skip (tool + connector) case.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CsGrHbs5SWmQkKqiimg6CF

Author: chandrasekharan-zipstack

src/unstract/clone/phases/workflow.py

@@ -123,42 +123,37 @@ def _clone_one(
         name = src["workflow_name"]
         src_id = src["id"]
 
+        # Collect every blocking dependency before skipping, so a workflow
+        # held up by both a skipped tool and a skipped connector reports both
+        # reasons in one pass rather than one per re-run.
+        skip_reasons: list[str] = []
         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,
+            skip_reasons.append(
+                "its tool was skipped (frictionless adapter); wire equivalents "
+                "on target and re-run"
             )
-            with lock:
-                result.skipped += 1
-                result.warnings.append(
-                    f"workflow '{name}' skipped — its tool was skipped "
-                    "(frictionless adapter); wire equivalents on target and re-run"
-                )
-            return
-
         skipped_conns = (
             self._wf_to_connector_ids.get(src_id, set())
             & self.ctx.skipped_connector_ids
         )
         if skipped_conns:
+            skip_reasons.append(
+                "depends on un-clonable connector(s) "
+                f"({', '.join(sorted(skipped_conns))}); provision them on target "
+                "with the same name and re-run"
+            )
+        if skip_reasons:
             logger.warning(
-                "skipping workflow '%s' src=%s — endpoint connector(s) %s were "
-                "skipped (OAuth/redacted); cloning it would create pipelines that "
-                "fail every run",
+                "skipping workflow '%s' src=%s — %s",
                 name,
                 src_id,
-                ", ".join(sorted(skipped_conns)),
+                "; ".join(skip_reasons),
             )
             with lock:
                 result.skipped += 1
-                result.warnings.append(
-                    f"workflow '{name}' skipped — depends on un-clonable "
-                    "connector(s); provision them on target with the same name "
-                    "and re-run"
-                )
+                for reason in skip_reasons:
+                    result.warnings.append(f"workflow '{name}' skipped — {reason}")
             return
 
         try:

tests/clone/test_workflow_phase.py

@@ -195,6 +195,36 @@ def test_cascade_skip_when_workflow_tool_was_skipped():
     assert [p["workflow_name"] for p in tgt.posts] == ["OK WF"]
     assert ctx.remap.resolve("workflow", "wf-skipped") is None
     assert ctx.remap.resolve("workflow", "wf-ok") is not None
+    assert any("its tool was skipped" in w for w in result.warnings)
+
+
+def test_dual_skip_reports_both_tool_and_connector_reasons():
+    """A workflow blocked by both a skipped tool and a skipped connector
+    surfaces both reasons in one pass, not one per re-run.
+    """
+    skipped_reg = "skipped-registry-id"
+    skipped_conn = "oauth-conn-id"
+    src = FakeClient([_src("wf-both", "Blocked WF")])
+    src.tool_instances = [{"workflow": "wf-both", "tool_id": skipped_reg}]
+    src.endpoints = [
+        {
+            "workflow": "wf-both",
+            "endpoint_type": "SOURCE",
+            "connector_instance": {"id": skipped_conn},
+        }
+    ]
+    tgt = FakeClient()
+    ctx = _ctx(src, tgt)
+    ctx.skipped_custom_tool_registry_ids.add(skipped_reg)
+    ctx.skipped_connector_ids.add(skipped_conn)
+
+    result = WorkflowPhase(ctx).run(CloneReport())
+
+    assert result.skipped == 1
+    assert tgt.posts == []
+    assert ctx.remap.resolve("workflow", "wf-both") is None
+    assert any("its tool was skipped" in w for w in result.warnings)
+    assert any("un-clonable connector" in w for w in result.warnings)
 
 
 def test_cascade_skip_when_endpoint_connector_skipped():