UN-3479 [FIX] Address Greptile: re-run idempotency for lookups versions + MR api keys

- lookups: adopt existing target published versions by name instead of
  re-publishing (fixes spurious failures / duplicate version history on
  re-runs and adopted definitions)
- lookups: guard draft_cache so a failed fetch can't clobber a peer's
  valid draft id (TOCTOU)
- manual_review: adopt ReviewApiKey by (class_name, description) so
  re-runs don't create duplicate keys; only warn for keys actually minted
- tests for both adopt paths

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

Author: chandrasekharan-zipstack

src/unstract/clone/phases/lookups.py

@@ -303,7 +303,9 @@ def _replay_versions(
         """Reproduce the source's published versions on the target in
         ``version_number`` order: set the target draft to each version's
         content + files, then publish. Records a ``lookup_version`` remap per
-        published version so its pinned assignments resolve.
+        published version so its pinned assignments resolve. A target version
+        with the same ``version_name`` is adopted (no re-publish) so re-runs
+        and adopted definitions stay idempotent.
         """
         try:
             versions = self.ctx.source.list_lookup_versions(src_lookup_id)
@@ -316,11 +318,41 @@ def _replay_versions(
                 )
             return
 
+        # Existing target versions — re-publishing a name that already exists
+        # would error or duplicate history; adopt those instead.
+        try:
+            tgt_versions = self.ctx.target.list_lookup_versions(tgt_lookup_id)
+        except Exception as e:
+            logger.warning(
+                "lookup '%s': target version listing failed: %s", name, e
+            )
+            tgt_versions = []
+        tgt_by_name: dict[str, str] = {
+            v["version_name"]: str(v["version_id"])
+            for v in tgt_versions
+            if v.get("version_name") and v.get("version_id")
+        }
+
         published = sorted(
             (v for v in versions if not v.get("is_draft")),
             key=lambda v: v.get("version_number") or 0,
         )
         for v in published:
+            existing_id = tgt_by_name.get(v.get("version_name"))
+            if existing_id is not None:
+                src_version_id = v.get("version_id")
+                with lock:
+                    result.adopted += 1
+                    if src_version_id:
+                        self.ctx.remap.record(
+                            "lookup_version", str(src_version_id), existing_id
+                        )
+                logger.info(
+                    "lookup '%s': adopted published version '%s' (already present)",
+                    name,
+                    v.get("version_name"),
+                )
+                continue
             self._replay_one_version(
                 name, src_lookup_id, tgt_lookup_id, v, result, lock
             )
@@ -825,5 +857,8 @@ def _target_draft_version(
             logger.warning("target lookup %s draft fetch failed: %s", tgt_lookup_id, e)
             draft_id = None
         with lock:
-            draft_cache[tgt_lookup_id] = draft_id
-        return draft_id
+            # A racing peer may have already cached a valid id; the GET is
+            # read-only so a real id always wins over this thread's failure.
+            if draft_cache.get(tgt_lookup_id) is None:
+                draft_cache[tgt_lookup_id] = draft_id
+            return draft_cache[tgt_lookup_id]

src/unstract/clone/phases/manual_review.py

@@ -401,14 +401,30 @@ def _clone_api_keys(self, result: PhaseResult) -> None:
             logger.info("[dry-run] would recreate %d review api key(s)", len(src_keys))
             return
 
-        # The api_key secret is server-minted and non-copyable; recreating
-        # yields a NEW secret, so external consumers must be re-wired.
-        result.warnings.append(
-            f"{len(src_keys)} review API key(s) recreated with freshly minted "
-            "secrets — the original key values cannot be copied; re-wire any "
-            "external consumers to the new keys"
-        )
+        # ReviewApiKey has no cross-org natural key (api_key is server-minted),
+        # so adopt by (class_name, description) to stay idempotent on re-runs.
+        try:
+            tgt_keys = self.ctx.target.list_review_api_keys()
+        except Exception as e:
+            logger.exception(
+                "manual_review: failed to list target review api keys: %s", e
+            )
+            result.failed += 1
+            result.errors.append(f"list target review_api_keys: {e}")
+            return
+        existing = {
+            (k.get("class_name"), k.get("description")) for k in tgt_keys
+        }
+
+        recreated = 0
         for src in src_keys:
+            if (src.get("class_name"), src.get("description")) in existing:
+                result.adopted += 1
+                logger.info(
+                    "adopted review api key class=%s (already present)",
+                    src.get("class_name"),
+                )
+                continue
             payload = {
                 k: src[k]
                 for k in ("class_name", "description", "is_active")
@@ -424,4 +440,14 @@ def _clone_api_keys(self, result: PhaseResult) -> None:
                 result.errors.append(f"create review_api_key: {e}")
                 continue
             result.created += 1
-        logger.info("recreated %d review api key(s) on target", len(src_keys))
+            recreated += 1
+
+        # The api_key secret is server-minted and non-copyable; newly created
+        # keys get a NEW secret, so external consumers must be re-wired.
+        if recreated:
+            result.warnings.append(
+                f"{recreated} review API key(s) recreated with freshly minted "
+                "secrets — the original key values cannot be copied; re-wire any "
+                "external consumers to the new keys"
+            )
+        logger.info("recreated %d review api key(s) on target", recreated)

tests/clone/test_lookups_phase.py

@@ -460,6 +460,66 @@ def test_published_version_replayed_publishes_in_order_and_records_remap():
     assert remap.resolve("lookup_version", "src-v2") is not None
 
 
+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.
+    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(
+        lookups=[{"lookup_id": "tgt-lk", "name": "Vendors"}],
+        details={
+            "tgt-lk": {
+                "prompt_template": "",
+                "draft_version_id": "tgt-draft-lk",
+                "adapters": {"llm": None, "x2text": None},
+            }
+        },
+        versions={
+            "tgt-lk": [
+                {
+                    "version_id": "tgt-v1",
+                    "is_draft": False,
+                    "version_name": "v1",
+                    "version_number": 1,
+                },
+            ]
+        },
+    )
+    remap = RemapTable()
+    remap.record("adapter", "src-llm", "tgt-llm")
+    ctx = _ctx(src, tgt, remap=remap)
+    report = CloneReport()
+
+    LookupsPhase(ctx).run(report)
+
+    # Nothing re-published; remap points at the existing target version.
+    assert tgt.published_versions == []
+    assert remap.resolve("lookup_version", "src-v1") == "tgt-v1"
+
+
 def test_published_pinned_assignment_resolves_via_version_remap():
     src = _src_published_lookup()
     tgt = FakeClient()

tests/clone/test_manual_review_phase.py

@@ -249,6 +249,22 @@ def test_review_api_key_recreated_with_warning():
     assert any("re-wire any external consumers" in w for w in result.warnings)
 
 
+def test_review_api_key_adopted_on_rerun():
+    # Target already carries a key with the same (class_name, description):
+    # re-run must adopt, not create a duplicate or re-warn.
+    key = {"class_name": "invoices", "description": "d", "is_active": True}
+    src = FakeClient(workflows=[], api_keys=[{"id": "k1", "api_key": "s", **key}])
+    tgt = FakeClient(api_keys=[{"id": "t1", "api_key": "other", **key}])
+    ctx = _ctx(src, tgt, remap=RemapTable())
+    report = CloneReport()
+
+    result = ManualReviewPhase(ctx).run(report)
+
+    assert tgt.created_api_keys == []
+    assert result.adopted >= 1
+    assert not any("re-wire any external consumers" in w for w in result.warnings)
+
+
 def test_dry_run_plans_without_writing():
     src = FakeClient(
         workflows=[{"id": "src-wf", "workflow_name": "WF"}],