feat(clone): clone agentic verified-data; honour skip strategy for docs

Agentic verified-data ("ground truth manually verified by user") is curated
input, not regenerable output, so it must be cloned. AgenticStudioPhase now
re-points each source verified-data row to the cloned target document by
filename and recreates it (skipping docs absent on target and rows already
present). Extracted/comparison data stays uncloned — both regenerate on a
re-run + re-verify.

Also honour file_strategy="skip" in the new document path, matching the files
and lookups phases: under skip, agentic documents are listed and counted as
skipped (operator re-uploads), not transferred.

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

Author: chandrasekharan-zipstack

src/unstract/clone/client.py

@@ -963,3 +963,20 @@ def upload_agentic_document(
         return self._request(
             "POST", f"agentic/projects/{project_id}/documents/upload/", files=files
         )
+
+    def list_agentic_verified_data(self, project_id: str) -> list[dict[str, Any]]:
+        """List a project's verified (ground-truth) data rows. Each carries
+        ``document_name``, ``document``, and ``data`` (the curated JSON).
+        """
+        result = self._request(
+            "GET", "agentic/verified-data/", params={"project_id": project_id}
+        )
+        return result if isinstance(result, list) else (result or {}).get("results", [])
+
+    def create_agentic_verified_data(self, payload: dict[str, Any]) -> dict[str, Any]:
+        """Create a verified-data row (``project``, ``document``, ``data``).
+
+        One row per (project, document); callers pre-check to avoid a
+        uniqueness clash on re-runs.
+        """
+        return self._request("POST", "agentic/verified-data/", json=payload)

src/unstract/clone/phases/agentic_studio.py

@@ -22,6 +22,11 @@
      per-run ``agentic_prompt_version`` table and the child's ``parent_version``
      resolves through it. ``project`` is bound to the target id.
    - **schemas**: bound to the target ``project`` and created.
+   - **documents**: source uploads re-uploaded to the target project (skipping
+     filenames already present); they live in their own store, not the Prompt
+     Studio document table.
+   - **verified data**: ground-truth rows re-pointed to the cloned document by
+     filename. Extracted/comparison data is regenerable and not cloned.
 3. **Registry**: if the project has an active schema + prompt, republish its
    registry entry via the ``export`` action (mirror of custom_tool) and
    record an ``agentic_studio_registry`` remap. Projects with no source
@@ -172,6 +177,7 @@ def _clone_project(
         self._clone_prompt_versions(name, src_project_id, tgt_project_id, result, lock)
         self._clone_schemas(name, src_project_id, tgt_project_id, result, lock)
         self._clone_documents(name, src_project_id, tgt_project_id, result, lock)
+        self._clone_verified_data(name, src_project_id, tgt_project_id, result, lock)
         self._republish_registry(name, src_project_id, tgt_project_id, result, lock)
 
     def _build_project_payload(
@@ -445,6 +451,17 @@ def _clone_documents(
         if not src_docs:
             return
 
+        # Honour the file strategy: 'skip' means no binary transfer, like the
+        # files and lookups phases. Operator re-uploads on target.
+        if self.ctx.options.file_strategy == "skip":
+            with lock:
+                result.skipped += len(src_docs)
+                result.warnings.append(
+                    f"agentic {name}: {len(src_docs)} document(s) not copied "
+                    "(file_strategy=skip) — upload them manually on target"
+                )
+            return
+
         try:
             tgt_docs = self.ctx.target.list_agentic_documents(tgt_project_id)
         except Exception as e:
@@ -521,6 +538,82 @@ def _clone_one_document(
             result.created += 1
         logger.info("agentic '%s': uploaded document '%s'", name, file_name)
 
+    # ----- verified (ground-truth) data -----
+
+    def _clone_verified_data(
+        self,
+        name: str,
+        src_project_id: str,
+        tgt_project_id: str,
+        result: PhaseResult,
+        lock: threading.Lock,
+    ) -> None:
+        try:
+            src_rows = self.ctx.source.list_agentic_verified_data(src_project_id)
+        except Exception as e:
+            logger.exception("agentic '%s': verified-data listing failed: %s", name, e)
+            with lock:
+                result.failed += 1
+                result.errors.append(f"agentic {name} list verified-data: {e}")
+            return
+        if not src_rows:
+            return
+
+        # Verified data FKs a document; map source rows to target docs by
+        # filename, the only identity stable across orgs.
+        try:
+            tgt_docs = self.ctx.target.list_agentic_documents(tgt_project_id)
+            tgt_existing = self.ctx.target.list_agentic_verified_data(tgt_project_id)
+        except Exception as e:
+            logger.warning(
+                "agentic '%s': target verified-data lookup failed "
+                "(re-run may duplicate): %s",
+                name,
+                e,
+            )
+            tgt_docs, tgt_existing = [], []
+        doc_id_by_name = {d.get("original_filename"): d.get("id") for d in tgt_docs}
+        verified_doc_ids = {r.get("document") for r in tgt_existing}
+
+        for src in src_rows:
+            file_name = src.get("document_name")
+            tgt_doc_id = doc_id_by_name.get(file_name)
+            if not tgt_doc_id:
+                with lock:
+                    result.skipped += 1
+                    result.warnings.append(
+                        f"agentic {name}: verified data for '{file_name}' skipped — "
+                        "document not on target"
+                    )
+                continue
+            if tgt_doc_id in verified_doc_ids:
+                with lock:
+                    result.skipped += 1
+                continue
+            payload = {
+                "project": tgt_project_id,
+                "document": tgt_doc_id,
+                "data": src.get("data"),
+            }
+            try:
+                self.ctx.target.create_agentic_verified_data(payload)
+            except Exception as e:
+                logger.exception(
+                    "agentic '%s': verified data for '%s' create failed: %s",
+                    name,
+                    file_name,
+                    e,
+                )
+                with lock:
+                    result.failed += 1
+                    result.errors.append(
+                        f"agentic {name} create verified-data {file_name}: {e}"
+                    )
+                continue
+            with lock:
+                result.created += 1
+            logger.info("agentic '%s': cloned verified data for '%s'", name, file_name)
+
     # ----- registry -----
 
     def _republish_registry(
@@ -626,12 +719,21 @@ def _plan_children(
             src_docs = self.ctx.source.list_agentic_documents(src_project_id)
         except Exception:
             src_docs = []
+        try:
+            src_verified = self.ctx.source.list_agentic_verified_data(src_project_id)
+        except Exception:
+            src_verified = []
         with lock:
             for v in src_versions:
                 self.ctx.remap.record_planned("agentic_prompt_version", v["id"])
                 result.created += 1
             result.created += len(src_schemas)
-            result.created += len(src_docs)
+            # Documents move only when the file strategy copies binaries.
+            if self.ctx.options.file_strategy == "skip":
+                result.skipped += len(src_docs)
+            else:
+                result.created += len(src_docs)
+            result.created += len(src_verified)
 
     # ----- settings -----
 

tests/clone/test_agentic_studio_phase.py

@@ -41,6 +41,7 @@ def __init__(
         users=None,
         documents=None,
         document_blobs=None,
+        verified_data=None,
     ):
         self.projects = list(projects or [])
         self.users = list(users or [])
@@ -56,6 +57,8 @@ def __init__(
         self.documents = {k: list(v) for k, v in (documents or {}).items()}
         # document_id -> raw bytes
         self.document_blobs = dict(document_blobs or {})
+        # project_id -> list of verified-data rows
+        self.verified_data = {k: list(v) for k, v in (verified_data or {}).items()}
 
         self.created_projects: list[dict] = []
         self.created_versions: list[dict] = []
@@ -64,6 +67,7 @@ def __init__(
         self.updated_settings: list[tuple[str, dict]] = []
         self.exported_projects: list[str] = []
         self.uploaded_documents: list[tuple[str, str, bytes]] = []
+        self.created_verified_data: list[dict] = []
         self._next_id = 1
 
     def _mint(self, prefix: str) -> str:
@@ -162,6 +166,20 @@ def upload_agentic_document(self, project_id, file_name, data, mime_type):
         self.documents.setdefault(project_id, []).append(row)
         return {"data": [{"document_name": file_name}]}
 
+    # ----- verified data -----
+
+    def list_agentic_verified_data(self, project_id):
+        return list(self.verified_data.get(project_id, []))
+
+    def create_agentic_verified_data(self, payload):
+        new = dict(payload)
+        new["id"] = self._mint("tgt-vd")
+        self.created_verified_data.append(new)
+        self.verified_data.setdefault(payload["project"], []).append(
+            {"document": payload["document"], "data": payload["data"]}
+        )
+        return new
+
 
 def _ctx(source, target, *, remap=None, **opt_overrides):
     return CloneContext(
@@ -573,3 +591,86 @@ def test_dry_run_counts_documents_without_uploading():
     assert tgt.uploaded_documents == []
     # 1 project + 2 documents predicted as creates.
     assert result.created == 3
+
+
+def test_verified_data_cloned_mapped_to_target_doc_by_filename():
+    src = FakeClient(
+        projects=[_src_project("src-p", "Receipts")],
+        documents={"src-p": [{"id": "d1", "original_filename": "report.pdf"}]},
+        document_blobs={"d1": b"%PDF-r"},
+        verified_data={
+            "src-p": [
+                {"document_name": "report.pdf", "document": "d1", "data": {"x": 1}},
+                # Ground truth for a doc that won't exist on target → skipped.
+                {"document_name": "gone.pdf", "document": "d9", "data": {"y": 2}},
+            ]
+        },
+    )
+    tgt = FakeClient()
+    ctx = _ctx(src, tgt)
+
+    result = AgenticStudioPhase(ctx).run(CloneReport())
+
+    assert result.failed == 0
+    assert len(tgt.created_verified_data) == 1
+    row = tgt.created_verified_data[0]
+    # Re-pointed to the freshly uploaded target document, not the source id.
+    uploaded_doc_id = tgt.documents[row["project"]][0]["id"]
+    assert row["document"] == uploaded_doc_id
+    assert row["data"] == {"x": 1}
+    # 'gone.pdf' ground truth skipped (no matching target document).
+    assert any("gone.pdf" in w for w in result.warnings)
+
+
+def test_verified_data_skipped_when_already_on_target():
+    src = FakeClient(
+        projects=[_src_project("src-p", "Receipts")],
+        documents={"src-p": [{"id": "d1", "original_filename": "report.pdf"}]},
+        verified_data={
+            "src-p": [
+                {"document_name": "report.pdf", "document": "d1", "data": {"x": 1}}
+            ]
+        },
+    )
+    tgt = FakeClient()
+    # Adopt an existing target project pre-seeded with the doc + its verified row.
+    tgt.projects.append({"id": "tgt-proj-0001", "name": "Receipts"})
+    tgt.documents["tgt-proj-0001"] = [{"id": "t-d1", "original_filename": "report.pdf"}]
+    tgt.verified_data["tgt-proj-0001"] = [{"document": "t-d1", "data": {"x": 1}}]
+    ctx = _ctx(src, tgt)
+
+    AgenticStudioPhase(ctx).run(CloneReport())
+
+    assert tgt.created_verified_data == []
+
+
+def test_dry_run_counts_verified_data_without_writing():
+    src = FakeClient(
+        projects=[_src_project("src-p", "Receipts")],
+        verified_data={
+            "src-p": [{"document_name": "a.pdf", "document": "d1", "data": {}}]
+        },
+    )
+    tgt = FakeClient()
+    ctx = _ctx(src, tgt, dry_run=True)
+
+    result = AgenticStudioPhase(ctx).run(CloneReport())
+
+    assert tgt.created_verified_data == []
+    # 1 project + 1 verified-data row predicted as creates.
+    assert result.created == 2
+
+
+def test_documents_not_copied_under_file_strategy_skip():
+    src = FakeClient(
+        projects=[_src_project("src-p", "Receipts")],
+        documents={"src-p": [{"id": "d1", "original_filename": "report.pdf"}]},
+        document_blobs={"d1": b"%PDF-r"},
+    )
+    tgt = FakeClient()
+    ctx = _ctx(src, tgt, file_strategy="skip")
+
+    result = AgenticStudioPhase(ctx).run(CloneReport())
+
+    assert tgt.uploaded_documents == []
+    assert result.skipped == 1