UN-3479 [FIX] Address Greptile: HITL auth errors, lookup staging, agentic re-run idempotency

- client.get_review_settings: only suppress the backend's DoesNotExist
  (500); re-raise 401/403/429 so an auth error can't silently drop a
  configured HITLSettings row.
- lookups._replay_one_version: return after a template/adapter staging
  failure instead of publishing a version with stale content.
- agentic _clone_prompt_versions / _clone_schemas: adopt children already
  on target (keyed by version) so a re-run against the same pair doesn't
  re-create duplicates — mirrors the lookups version-replay guard.

Tests added for each. The draft_cache TOCTOU (P2) is already mitigated:
the `is None` write guard guarantees a valid draft id is never overwritten
by a peer's failure; only a benign extra GET can occur.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: 4f5ab436-eabb-4464-9a96-f024724bc818

Author: chandrasekharan-zipstack

src/unstract/clone/DESIGN_cloud_entity_support.md

@@ -775,15 +775,18 @@ def get_review_settings(self, workflow_id: str) -> dict[str, Any] | None:
         """Fetch a workflow's HITLSettings row, or ``None`` if absent.
 
         The backend's ``settings_using_workflow`` raises ``DoesNotExist``
-        (→ 500) when no row exists rather than 404, so any error here is
-        treated as "no settings to clone".
+        (→ 500) when no row exists rather than 404, so only a 500 is treated
+        as "no settings to clone". Other errors (401/403/429) must surface —
+        suppressing them would silently drop a configured HITLSettings row.
         """
         try:
             body = self._request(
                 "GET", f"manual_review/settings/workflow/{workflow_id}/"
             )
-        except PlatformAPIError:
-            return None
+        except PlatformAPIError as e:
+            if e.status_code == 500:
+                return None
+            raise
         return (body or {}).get("data")
 
     def create_review_settings(self, payload: dict[str, Any]) -> dict[str, Any]:

src/unstract/clone/client.py

@@ -251,6 +251,22 @@ def _clone_prompt_versions(
                 result.errors.append(f"agentic {name} list prompt-versions: {e}")
             return
 
+        # Adopt versions already on target (keyed by version number) so a
+        # re-run against the same pair doesn't re-create duplicates.
+        try:
+            tgt_versions = self.ctx.target.list_agentic_prompt_versions(
+                project_id=tgt_project_id
+            )
+        except Exception as e:
+            logger.warning(
+                "agentic '%s': target prompt-version listing failed "
+                "(re-run may duplicate): %s",
+                name,
+                e,
+            )
+            tgt_versions = []
+        tgt_by_version = {v.get("version"): v for v in tgt_versions}
+
         # parent_version is a self-FK: clone roots (no parent) first so a child's
         # parent already resolves. Sort by version ascending as a stable order.
         ordered = sorted(
@@ -259,18 +275,28 @@ def _clone_prompt_versions(
         )
         for src in ordered:
             self._clone_one_prompt_version(
-                name, src, tgt_project_id, result, lock
+                name, src, tgt_project_id, tgt_by_version, result, lock
             )
 
     def _clone_one_prompt_version(
         self,
         name: str,
         src: dict[str, Any],
         tgt_project_id: str,
+        tgt_by_version: dict[Any, dict[str, Any]],
         result: PhaseResult,
         lock: threading.Lock,
     ) -> None:
         src_vid = src["id"]
+
+        existing = tgt_by_version.get(src.get("version"))
+        if existing is not None:
+            with lock:
+                result.adopted += 1
+                self.ctx.remap.record(
+                    "agentic_prompt_version", src_vid, existing["id"]
+                )
+            return
         payload = {
             k: v
             for k, v in src.items()
@@ -349,7 +375,27 @@ def _clone_schemas(
                 result.errors.append(f"agentic {name} list schemas: {e}")
             return
 
+        # Adopt schemas already on target (keyed by version) so a re-run
+        # doesn't re-create duplicates.
+        try:
+            tgt_schemas = self.ctx.target.list_agentic_schemas(
+                project_id=tgt_project_id
+            )
+        except Exception as e:
+            logger.warning(
+                "agentic '%s': target schema listing failed (re-run may "
+                "duplicate): %s",
+                name,
+                e,
+            )
+            tgt_schemas = []
+        existing_versions = {s.get("version") for s in tgt_schemas}
+
         for src in src_schemas:
+            if src.get("version") in existing_versions:
+                with lock:
+                    result.adopted += 1
+                continue
             payload = {
                 k: v
                 for k, v in src.items()

src/unstract/clone/phases/agentic_studio.py

@@ -403,6 +403,9 @@ def _replay_one_version(
                     result.errors.append(
                         f"lookup {name} version {version_name} template: {e}"
                     )
+                # Don't publish — a stale draft would freeze wrong content
+                # into the named version.
+                return
         tgt_adapters = self._remap_adapters(
             name, detail.get("adapters") or {}, result, lock
         )
@@ -423,6 +426,8 @@ def _replay_one_version(
                     result.errors.append(
                         f"lookup {name} version {version_name} adapters: {e}"
                     )
+                # Don't publish a version with unmapped adapters.
+                return
 
         self._replay_version_files(
             name, src_lookup_id, tgt_lookup_id, src_version_id, detail, result, lock

src/unstract/clone/phases/lookups.py

@@ -299,6 +299,58 @@ def test_schema_clone_bound_to_target_project():
     assert schema["json_schema"] == '{"type":"object"}'
 
 
+def test_rerun_adopts_existing_prompt_versions_and_schemas():
+    # Re-run against a pair whose project + children already exist on target
+    # must adopt them (no duplicate child creates).
+    src = FakeClient(
+        projects=[_src_project("src-p", "Receipts")],
+        versions={
+            "src-p": [
+                {
+                    "id": "src-v1",
+                    "project": "src-p",
+                    "version": 1,
+                    "prompt_text": "v1",
+                    "parent_version": None,
+                }
+            ]
+        },
+        schemas={
+            "src-p": [
+                {
+                    "id": "src-s1",
+                    "project": "src-p",
+                    "json_schema": '{"type":"object"}',
+                    "version": 1,
+                }
+            ]
+        },
+    )
+    tgt = FakeClient(
+        projects=[{"id": "tgt-existing", "name": "Receipts"}],
+        versions={
+            "tgt-existing": [
+                {"id": "tgt-v1", "project": "tgt-existing", "version": 1}
+            ]
+        },
+        schemas={
+            "tgt-existing": [
+                {"id": "tgt-s1", "project": "tgt-existing", "version": 1}
+            ]
+        },
+    )
+    ctx = _ctx(src, tgt, on_name_conflict="adopt")
+
+    result = AgenticStudioPhase(ctx).run(CloneReport())
+
+    # No duplicate children created on the re-run.
+    assert tgt.created_versions == []
+    assert tgt.created_schemas == []
+    assert result.created == 0
+    # Existing version adopted + remap recorded so child parents still resolve.
+    assert ctx.remap.resolve("agentic_prompt_version", "src-v1") == "tgt-v1"
+
+
 def test_registry_republished_and_remapped():
     src = FakeClient(
         projects=[_src_project("src-p", "Receipts")],

tests/clone/test_agentic_studio_phase.py

@@ -143,3 +143,17 @@ def test_options_response_with_null_body_still_yields_empty_schema():
     # Some deployments return 200 with no body on OPTIONS.
     client, _ = _client_with_mock(payload=None, text="")
     assert client.get_post_schema("pipeline/") == frozenset()
+
+
+def test_get_review_settings_500_treated_as_absent():
+    # Backend raises DoesNotExist (-> 500) when no HITLSettings row exists.
+    client, _ = _client_with_mock(status=500, text="DoesNotExist")
+    assert client.get_review_settings("wf-1") is None
+
+
+def test_get_review_settings_reraises_non_500():
+    # Auth / rate-limit errors must surface, not masquerade as "no settings".
+    client, _ = _client_with_mock(status=403, text="forbidden")
+    with pytest.raises(PlatformAPIError) as exc_info:
+        client.get_review_settings("wf-1")
+    assert exc_info.value.status_code == 403

tests/clone/test_client.py

@@ -460,6 +460,50 @@ def test_published_version_replayed_publishes_in_order_and_records_remap():
     assert remap.resolve("lookup_version", "src-v2") is not None
 
 
+def test_staging_template_failure_skips_publish():
+    # If staging a version's template onto the draft fails, the version must
+    # NOT be published — else stale content freezes into a named version.
+    src = FakeClient(
+        lookups=[_src_lookup("src-lk", "Vendors")],
+        details={"src-lk": _src_detail("Current draft", llm="src-llm")},
+        versions={
+            "src-lk": [
+                {
+                    "version_id": "src-v1",
+                    "is_draft": False,
+                    "version_name": "v1",
+                    "version_number": 1,
+                },
+            ]
+        },
+        version_details={
+            "src-v1": {
+                "version_id": "src-v1",
+                "is_draft": False,
+                "version_name": "v1",
+                "version_number": 1,
+                "prompt_template": "Frozen v1",
+                "adapters": {"llm": "src-llm", "x2text": None},
+                "files": [],
+            },
+        },
+    )
+    tgt = FakeClient()
+
+    def _boom(lookup_id, prompt_template):
+        raise RuntimeError("template too long")
+
+    tgt.update_lookup_draft_template = _boom
+    remap = RemapTable()
+    remap.record("adapter", "src-llm", "tgt-llm")
+    ctx = _ctx(src, tgt, remap=remap)
+
+    result = LookupsPhase(ctx).run(CloneReport())
+
+    assert tgt.published_versions == []  # staging failed -> no publish
+    assert result.failed >= 1
+
+
 def test_published_version_adopted_on_rerun_no_republish():
     # Target already has a same-name definition with the same published version
     # names: re-run must adopt them (no re-publish) and still record remaps.