Bound Jira attachment transfers and recover missing native files (JVNAUTOSCI-2737)

docs/engineering/workflow_first_jira_to_von_migration_jvnautosci_1223.md

@@ -44,6 +44,13 @@ For attachments above the Jira proxy's default 10 MiB retrieval limit, set
 `VON_INTERNAL_MCP_JIRA_ATTACHMENT_MAX_SIZE_BYTES=104857600` in the migration
 process environment. This explicitly permits retrieval up to 100 MiB; larger
 files remain reported gaps requiring a separate bounded capture.
+Attachment transfers use a private directory created by the calling proxy. The
+Jira helper streams into that directory, enforces the byte limit while reading,
+and returns a small size/hash receipt through MCP. The proxy verifies the file
+before hydrating the existing archive payload and removes the temporary directory
+on return or failure. Source filenames cannot select local paths. Binary reads
+have a separate 360-second outer budget for the existing three 60-second HTTP
+attempts and transfer overhead; caller cancellation still takes precedence.
 
 Completed batches are checkpointed atomically. Repeating the command resumes a
 matching source snapshot. If Jira has changed, use a fresh report path for the
@@ -52,6 +59,10 @@ using the previous import projection, report conflicts, preserve prior source
 archives, and deduplicate source activity. Unresolved hierarchy/link targets
 can be repaired from retained import rows after all projects have imported.
 Never infer deletion authority from a missing source issue.
+For checkpointed attachments, reruns also check whether the native attachment
+has a retained file. Previously unavailable bytes fill the existing attachment
+when they become available, preserving its identity, annotations and original
+timestamp. A file already bound by another writer is preserved.
 
 Start large recovery runs with one importer process. Keep raw payloads in private
 files and return compact progress summaries to the calling agent. Monitor host

src/backend/integrations/internal_mcp/jira_proxy_mcp.py

@@ -7,11 +7,13 @@
 
 from __future__ import annotations
 
+import base64
 import logging
 import os
 import sys
+import tempfile
 import threading
-from dataclasses import dataclass
+from dataclasses import dataclass, replace
 from hashlib import sha256
 from pathlib import Path
 from typing import Any, Dict, Mapping, Optional, cast
@@ -203,18 +205,17 @@ def __init__(self, config: JiraProxyConfig):
             log_tag=_LOG_TAG,
         )
         self._client = MCPStdIOClient(client_config)
-        # Export downloads use the existing three 60-second HTTP attempts.
+        # Binary downloads use the existing three 60-second HTTP attempts.
         # Allow twice that declared inner bound for transfer and JSON encoding;
         # ordinary issue reads retain their current operation budget.
-        self._export_client = MCPStdIOClient(
-            MCPServerConfig(
-                command=config.command,
-                args=config.args,
-                env=config.env,
-                timeout_sec=360.0,
-                log_tag=_LOG_TAG,
-            )
+        self._binary_config = MCPServerConfig(
+            command=config.command,
+            args=config.args,
+            env=config.env,
+            timeout_sec=360.0,
+            log_tag=_LOG_TAG,
         )
+        self._binary_client = MCPStdIOClient(self._binary_config)
 
     def source_capture_session(self):
         """Keep one helper for a source capture; authority is checked per read."""
@@ -226,6 +227,7 @@ async def _call(
         arguments: Dict[str, Any],
         *,
         resource_id: str | None = None,
+        client_override: MCPStdIOClient | None = None,
     ) -> Any:
         authority = None
         if tool_name == "jira_search" or tool_name.startswith("jira_get_"):
@@ -236,11 +238,14 @@ async def _call(
             )
         try:
             client = (
-                self._export_client
-                if tool_name == "jira_get_migration_export"
+                self._binary_client
+                if tool_name in {
+                    "jira_get_migration_export",
+                    "jira_get_attachment_content",
+                }
                 else self._client
             )
-            result = await client.call_tool(tool_name, arguments)
+            result = await (client_override or client).call_tool(tool_name, arguments)
             if authority is not None and isinstance(result, dict):
                 result = {**result, "authority": authority}
             return result
@@ -351,7 +356,50 @@ async def get_attachment_content(
         arguments: Dict[str, Any] = {"attachment_id": attachment_id}
         if isinstance(max_size_bytes, int):
             arguments["max_size_bytes"] = max_size_bytes
-        return await self._call("jira_get_attachment_content", arguments)
+        limit = max_size_bytes if isinstance(max_size_bytes, int) else 10 * 1024 * 1024
+        # The helper and caller share a private, caller-created directory. Jira
+        # content cannot select a filesystem destination. Keep large base64 out
+        # of MCP framing; hydrate the established return shape only after read-back.
+        with tempfile.TemporaryDirectory(prefix="von-jira-attachment-") as directory:
+            client = MCPStdIOClient(
+                replace(
+                    self._binary_config,
+                    env={
+                        **(self._binary_config.env or {}),
+                        "VON_JIRA_ATTACHMENT_STAGING_DIR": directory,
+                    },
+                )
+            )
+            result = await self._call(
+                "jira_get_attachment_content", arguments, client_override=client
+            )
+            if not isinstance(result, dict) or result.get("success") is not True:
+                return result
+            staged = result.get("staged_file") or {}
+            name = staged.get("name", "")
+            if (
+                not isinstance(name, str)
+                or len(name) != 36
+                or not name.endswith(".bin")
+                or any(char not in "0123456789abcdef" for char in name[:-4])
+            ):
+                raise JiraProxyError("Invalid attachment staging receipt")
+            path = Path(directory) / name
+            if path.is_symlink() or not path.is_file() or path.stat().st_size > limit:
+                raise JiraProxyError("Invalid attachment staging file")
+            with path.open("rb") as stream:
+                data = stream.read(limit + 1)
+            if (
+                len(data) > limit
+                or len(data) != staged.get("size_bytes")
+                or len(data) != result.get("size_bytes")
+                or sha256(data).hexdigest() != staged.get("sha256")
+            ):
+                raise JiraProxyError("Attachment staging integrity mismatch")
+            result = dict(result)
+            result.pop("staged_file")
+            result["content_base64"] = base64.b64encode(data).decode("ascii")
+            return result
 
     async def get_transitions(self, *, issue_key: str) -> Dict[str, Any]:
         return await self._call("jira_get_transitions", {"issue_key": issue_key})

src/backend/mcp_server/mcp_server.py

@@ -4,8 +4,10 @@
 import re
 import time
 import sys
+from hashlib import sha256
 from pathlib import Path
 from typing import Any, Dict, List, Sequence
+from uuid import uuid4
 
 import anyio
 import requests
@@ -173,11 +175,15 @@ def _request_json(
     return {"success": False, "error": "Jira request failed after retries"}
 
 
-def _request_bytes(url: str) -> Dict[str, Any]:
+def _request_bytes(
+    url: str, *, destination: Path | None = None, max_size_bytes: int | None = None
+) -> Dict[str, Any]:
     max_attempts = 3
     delay_sec = 0.5
 
     for attempt in range(1, max_attempts + 1):
+        resp = None
+        staged_complete = False
         try:
             resp = requests.get(
                 url,
@@ -186,9 +192,30 @@ def _request_bytes(url: str) -> Dict[str, Any]:
                     "Accept": "*/*",
                 },
                 timeout=60,
+                stream=destination is not None,
             )
 
             if resp.ok:
+                if destination is not None:
+                    size = 0
+                    digest = sha256()
+                    with destination.open("wb") as output:
+                        for chunk in resp.iter_content(chunk_size=1024 * 1024):
+                            size += len(chunk)
+                            if max_size_bytes is not None and size > max_size_bytes:
+                                return {"success": False, "error": "attachment_too_large"}
+                            output.write(chunk)
+                            digest.update(chunk)
+                    staged_complete = True
+                    return {
+                        "success": True,
+                        "content_type": resp.headers.get("Content-Type"),
+                        "staged_file": {
+                            "name": destination.name,
+                            "size_bytes": size,
+                            "sha256": digest.hexdigest(),
+                        },
+                    }
                 return {
                     "success": True,
                     "status_code": resp.status_code,
@@ -234,6 +261,11 @@ def _request_bytes(url: str) -> Dict[str, Any]:
                 delay_sec = min(delay_sec * 2.0, 8.0)
                 continue
             return {"success": False, "error": f"Jira request failed: {exc}"}
+        finally:
+            if resp is not None:
+                resp.close()
+            if destination is not None and not staged_complete:
+                destination.unlink(missing_ok=True)
 
     return {"success": False, "error": "Jira request failed after retries"}
 
@@ -504,8 +536,18 @@ def jira_get_attachment_content(
             "metadata": metadata,
         }
 
+    # Only the trusted parent process sets this private staging directory. It
+    # is never a tool argument or a path derived from Jira metadata.
+    staging_directory = os.environ.get("VON_JIRA_ATTACHMENT_STAGING_DIR")
+    destination = (
+        Path(staging_directory) / f"{uuid4().hex}.bin"
+        if staging_directory
+        else None
+    )
     content_result = _request_bytes(
-        f"{JIRA_BASE_URL}/rest/api/3/attachment/content/{attachment_id}"
+        f"{JIRA_BASE_URL}/rest/api/3/attachment/content/{attachment_id}",
+        destination=destination,
+        max_size_bytes=max_size_bytes,
     )
     if content_result.get("success") is not True:
         return {
@@ -515,6 +557,19 @@ def jira_get_attachment_content(
             "metadata": metadata,
         }
 
+    if destination is not None:
+        staged = content_result["staged_file"]
+        return {
+            "success": True,
+            "attachment_id": attachment_id,
+            "filename": metadata.get("filename"),
+            "size_bytes": size_bytes if size_bytes is not None else staged["size_bytes"],
+            "content_type": metadata.get("mimeType")
+            or content_result.get("content_type"),
+            "staged_file": staged,
+            "metadata": metadata,
+        }
+
     raw_content = content_result.get("content")
     if not isinstance(raw_content, (bytes, bytearray)):
         return {

src/backend/services/jira_task_import_service.py

@@ -47,6 +47,7 @@
     find_task_by_external_reference,
     get_task,
     link_tasks,
+    list_task_attachments,
     record_task_history_event,
     set_task_epic,
     set_task_parent,
@@ -1296,6 +1297,27 @@ def _resolve_participant(raw_participant: Mapping[str, Any] | None) -> dict[str,
         mapped_fields.append("comments")
 
     attachments = _extract_issue_attachments(raw_issue)
+    retry_attachment_ids = {
+        item["attachment_id"]
+        for item in attachments
+        if item["attachment_id"] in existing_activity_checkpoint["attachment_ids"]
+        and item.get("content_base64")
+    }
+    if retry_attachment_ids:
+        # A checkpoint may represent metadata retained before its bytes were
+        # available. Reconcile actual native files before skipping those IDs.
+        offset = 0
+        while True:
+            page = list_task_attachments(task_concept_id, limit=500, offset=offset)
+            for stored in page["attachments"]:
+                source = stored.get("source") or {}
+                if source.get("source_system") == "jira" and stored.get(
+                    "file_copy_concept_id"
+                ):
+                    retry_attachment_ids.discard(source.get("external_id"))
+            offset += len(page["attachments"])
+            if offset >= page["total"] or not page["attachments"]:
+                break
     imported_attachment = False
     for attachment in attachments:
         raw_attachment_id = attachment.get("attachment_id")
@@ -1309,11 +1331,15 @@ def _resolve_participant(raw_participant: Mapping[str, Any] | None) -> dict[str,
                 attachment.get("created_at"),
                 attachment.get("size_bytes"),
             )
-        if attachment_id in existing_activity_checkpoint["attachment_ids"]:
+        if (
+            attachment_id in existing_activity_checkpoint["attachment_ids"]
+            and attachment_id not in retry_attachment_ids
+        ):
             continue
         imported_attachment = True
         if dry_run:
-            imported_activity["attachment_ids"].append(attachment_id)
+            if attachment_id not in imported_activity["attachment_ids"]:
+                imported_activity["attachment_ids"].append(attachment_id)
             continue
         author_resolution = _resolve_participant(
             attachment.get("author")
@@ -1423,7 +1449,8 @@ def _resolve_participant(raw_participant: Mapping[str, Any] | None) -> dict[str,
                 ),
                 file_copy_concept_id=file_copy_concept_id,
             )
-            imported_activity["attachment_ids"].append(attachment_id)
+            if attachment_id not in imported_activity["attachment_ids"]:
+                imported_activity["attachment_ids"].append(attachment_id)
         except Exception as exc:
             dropped_fields.append(
                 {

src/backend/services/task_management_service.py

@@ -4509,6 +4509,45 @@ def add_task_attachment(
         entry=attachment,
     )
     if stored_entry != attachment:
+        if normalised_file_copy_concept_id and not stored_entry.get(
+            "file_copy_concept_id"
+        ):
+            # Complete an earlier metadata-only import without duplicating its
+            # attachment or replacing a file another writer has already bound.
+            _, before = _get_task_doc(task_concept_id)
+            entries = _list_metadata_items(before, TASK_METADATA_KEY_ATTACHMENTS)
+            index = next(
+                (
+                    index
+                    for index, item in enumerate(entries)
+                    if item.get("attachment_id") == stored_entry["attachment_id"]
+                ),
+                None,
+            )
+            if index is None:
+                raise TaskManagementError("Attachment disappeared before file binding")
+            prefix = f"metadata.{TASK_METADATA_KEY_ATTACHMENTS}.{index}"
+            ConceptsRepository.update_one(
+                {
+                    "concept_id": task_concept_id,
+                    f"{prefix}.attachment_id": stored_entry["attachment_id"],
+                    f"{prefix}.file_copy_concept_id": {"$in": [None, ""]},
+                },
+                {
+                    "$set": {
+                        f"{prefix}.file_copy_concept_id": normalised_file_copy_concept_id,
+                        f"{prefix}.uri": attachment["uri"],
+                        "updated_at": _now(),
+                    }
+                },
+            )
+            _, current = _get_task_doc(task_concept_id)
+            for item in _list_metadata_items(current, TASK_METADATA_KEY_ATTACHMENTS):
+                if item.get("attachment_id") == stored_entry["attachment_id"]:
+                    if item.get("file_copy_concept_id"):
+                        return item
+                    break
+            raise TaskManagementError("Attachment file binding failed read-back")
         return stored_entry
     try:
         _append_task_history_event(