feat: Implement linked file management in FileService

- Added methods to create, register, and manage linked-input aliases for files across sessions.
- Introduced functionality to check for existing linked references and handle read-only constraints during file updates.
- Enhanced the orchestrator to support linking files into sessions, ensuring proper handling of explicit and linked files during execution.
- Updated unit tests to cover new linked file behaviors and ensure correct functionality.

Author: usnavy13

src/services/file.py

@@ -75,6 +75,62 @@ def _get_session_files_key(self, session_id: str) -> str:
         """Generate Redis key for session file list."""
         return f"session_files:{session_id}"
 
+    def _get_file_links_key(self, session_id: str, file_id: str) -> str:
+        """Generate Redis key for aliases that reference a source file."""
+        return f"file_links:{session_id}:{file_id}"
+
+    async def _register_link_reference(
+        self,
+        source_session_id: str,
+        source_file_id: str,
+        linked_session_id: str,
+        linked_file_id: str,
+    ) -> None:
+        """Track a linked-input alias for cleanup safety."""
+        links_key = self._get_file_links_key(source_session_id, source_file_id)
+        ttl_seconds = settings.get_session_ttl_minutes() * 60
+        await self.redis_client.sadd(links_key, f"{linked_session_id}:{linked_file_id}")
+        await self.redis_client.expire(links_key, ttl_seconds)
+
+    async def _remove_link_reference(
+        self,
+        source_session_id: str,
+        source_file_id: str,
+        linked_session_id: str,
+        linked_file_id: str,
+    ) -> None:
+        """Remove a linked-input alias reference."""
+        links_key = self._get_file_links_key(source_session_id, source_file_id)
+        await self.redis_client.srem(links_key, f"{linked_session_id}:{linked_file_id}")
+
+    async def _has_link_references(self, session_id: str, file_id: str) -> bool:
+        """Return True when other session aliases still reference a file."""
+        links_key = self._get_file_links_key(session_id, file_id)
+        return bool(await self.redis_client.smembers(links_key))
+
+    async def _find_linked_file(
+        self, target_session_id: str, source_session_id: str, source_file_id: str
+    ) -> Optional[str]:
+        """Return an existing linked-input alias for the given source file."""
+        session_files_key = self._get_session_files_key(target_session_id)
+        file_ids = await self.redis_client.smembers(session_files_key)
+
+        for candidate_file_id in file_ids:
+            metadata = await self.get_file_metadata(
+                target_session_id, candidate_file_id
+            )
+            if not metadata:
+                continue
+
+            if (
+                metadata.get("type") == "linked_input"
+                and metadata.get("source_session_id") == source_session_id
+                and metadata.get("source_file_id") == source_file_id
+            ):
+                return candidate_file_id
+
+        return None
+
     async def _store_file_metadata(
         self, session_id: str, file_id: str, metadata: Dict[str, Any]
     ) -> None:
@@ -317,6 +373,69 @@ async def list_files(self, session_id: str) -> List[FileInfo]:
             logger.error("Failed to list files", error=str(e), session_id=session_id)
             return []
 
+    async def link_file_into_session(
+        self, target_session_id: str, source_session_id: str, source_file_id: str
+    ) -> Optional[FileInfo]:
+        """Create or reuse a read-only linked alias in the target session."""
+        source_metadata = await self.get_file_metadata(
+            source_session_id, source_file_id
+        )
+        if not source_metadata:
+            logger.warning(
+                "Cannot link missing source file",
+                source_session_id=source_session_id,
+                source_file_id=source_file_id,
+                target_session_id=target_session_id,
+            )
+            return None
+
+        existing_linked_file_id = await self._find_linked_file(
+            target_session_id, source_session_id, source_file_id
+        )
+        if existing_linked_file_id:
+            return await self.get_file_info(target_session_id, existing_linked_file_id)
+
+        linked_file_id = generate_file_id()
+        metadata = {
+            "file_id": linked_file_id,
+            "filename": source_metadata["filename"],
+            "content_type": source_metadata["content_type"],
+            "object_key": source_metadata["object_key"],
+            "session_id": target_session_id,
+            "created_at": datetime.utcnow().isoformat(),
+            "size": source_metadata["size"],
+            "path": source_metadata["path"],
+            "type": "linked_input",
+            "source_session_id": source_session_id,
+            "source_file_id": source_file_id,
+            "is_read_only": "1",
+        }
+
+        await self._store_file_metadata(target_session_id, linked_file_id, metadata)
+        await self._register_link_reference(
+            source_session_id,
+            source_file_id,
+            target_session_id,
+            linked_file_id,
+        )
+
+        logger.debug(
+            "Linked file into session",
+            target_session_id=target_session_id,
+            linked_file_id=linked_file_id,
+            source_session_id=source_session_id,
+            source_file_id=source_file_id,
+        )
+
+        return FileInfo(
+            file_id=linked_file_id,
+            filename=metadata["filename"],
+            size=metadata["size"],
+            content_type=metadata["content_type"],
+            created_at=datetime.fromisoformat(metadata["created_at"]),
+            path=metadata["path"],
+        )
+
     async def download_file(self, session_id: str, file_id: str) -> Optional[str]:
         """Generate download URL for a file."""
         metadata = await self.get_file_metadata(session_id, file_id)
@@ -353,6 +472,30 @@ async def delete_file(self, session_id: str, file_id: str) -> bool:
         if not metadata:
             return False
 
+        if metadata.get("type") == "linked_input":
+            await self._delete_file_metadata(session_id, file_id)
+            await self._remove_link_reference(
+                metadata["source_session_id"],
+                metadata["source_file_id"],
+                session_id,
+                file_id,
+            )
+            logger.debug(
+                "Deleted linked file alias",
+                session_id=session_id,
+                file_id=file_id,
+            )
+            return True
+
+        if await self._has_link_references(session_id, file_id):
+            await self._delete_file_metadata(session_id, file_id)
+            logger.debug(
+                "Deleted file metadata but retained shared object",
+                session_id=session_id,
+                file_id=file_id,
+            )
+            return True
+
         object_key = metadata["object_key"]
 
         try:
@@ -753,6 +896,12 @@ async def cleanup_orphan_objects(self, batch_limit: int = 1000) -> int:
                 if object_session_id in active_session_ids:
                     continue
 
+                source_file_id = parts[3] if len(parts) >= 4 else None
+                if source_file_id and await self._has_link_references(
+                    object_session_id, source_file_id
+                ):
+                    continue
+
                 # Double-check via Redis existence in case index is stale
                 if object_session_id not in checked_missing_sessions:
                     try:
@@ -829,6 +978,14 @@ async def update_file_content(
                 )
                 return False
 
+            if metadata.get("is_read_only") == "1":
+                logger.debug(
+                    "Skipping update for read-only file",
+                    session_id=session_id[:12],
+                    file_id=file_id,
+                )
+                return False
+
             object_key = metadata.get("object_key")
             if not object_key:
                 logger.warning(

src/services/interfaces.py

@@ -83,6 +83,13 @@ async def list_executions(
 class FileServiceInterface(ABC):
     """Interface for file management service."""
 
+    @abstractmethod
+    async def get_file_metadata(
+        self, session_id: str, file_id: str
+    ) -> Optional[Dict[str, Any]]:
+        """Retrieve raw file metadata."""
+        pass
+
     @abstractmethod
     async def upload_file(
         self, session_id: str, request: FileUploadRequest
@@ -122,6 +129,13 @@ async def stream_file_to_path(
         """Stream file content directly to a local file path."""
         pass
 
+    @abstractmethod
+    async def link_file_into_session(
+        self, target_session_id: str, source_session_id: str, source_file_id: str
+    ) -> Optional[FileInfo]:
+        """Create or reuse a read-only alias for a source file in another session."""
+        pass
+
     @abstractmethod
     async def delete_file(self, session_id: str, file_id: str) -> bool:
         """Delete a file from storage."""

src/services/orchestrator.py

@@ -40,6 +40,7 @@
     ExecutionServiceInterface,
     FileServiceInterface,
 )
+from .execution.output import OutputProcessor
 from .state import StateService
 from .state_archival import StateArchivalService
 
@@ -317,19 +318,55 @@ async def _mount_files(self, ctx: ExecutionContext) -> List[Dict[str, Any]]:
         """Mount files for code execution.
 
         Behavior:
-        1. If request.files[] is provided, mount those files (explicit mounting)
-        2. If no request.files[] but session_id exists, auto-mount ALL session files
-        3. If neither, return empty list
+        1. Mount explicit file references from request.files[]
+        2. Also auto-mount files already tracked in the current session
+        3. Deduplicate by mounted filename with precedence:
+           explicit refs > native current-session files > linked-input aliases
         """
-        # If explicit files provided, mount those (existing behavior)
+        explicit_files = []
         if ctx.request.files:
-            return await self._mount_explicit_files(ctx)
+            explicit_files = await self._mount_explicit_files(ctx)
 
-        # Auto-mount all session files when session_id exists but no explicit files
+        session_files = []
         if ctx.session_id:
-            return await self._auto_mount_session_files(ctx)
+            session_files = await self._auto_mount_session_files(ctx)
+
+        native_session_files = [
+            file_info
+            for file_info in session_files
+            if not file_info.get("is_linked_input")
+        ]
+        linked_session_files = [
+            file_info for file_info in session_files if file_info.get("is_linked_input")
+        ]
+
+        return self._merge_mounted_files(
+            explicit_files,
+            native_session_files,
+            linked_session_files,
+        )
+
+    def _mount_dedupe_key(self, file_info: Dict[str, Any]) -> str:
+        """Return the normalized filename key used for mount precedence."""
+        return OutputProcessor.sanitize_filename(file_info.get("filename", ""))
+
+    def _merge_mounted_files(
+        self, *groups: List[Dict[str, Any]]
+    ) -> List[Dict[str, Any]]:
+        """Merge mounted file groups using filename-based precedence."""
+        merged: List[Dict[str, Any]] = []
+        mounted_names = set()
+
+        for group in groups:
+            for file_info in group:
+                dedupe_key = self._mount_dedupe_key(file_info)
+                if not dedupe_key or dedupe_key in mounted_names:
+                    continue
 
-        return []
+                merged.append(file_info)
+                mounted_names.add(dedupe_key)
+
+        return merged
 
     async def _mount_explicit_files(
         self, ctx: ExecutionContext
@@ -363,13 +400,21 @@ async def _mount_explicit_files(
             if key in mounted_ids:
                 continue
 
+            if ctx.session_id and file_ref.session_id != ctx.session_id:
+                await self.file_service.link_file_into_session(
+                    ctx.session_id,
+                    file_ref.session_id,
+                    file_info.file_id,
+                )
+
             mounted.append(
                 {
                     "file_id": file_info.file_id,
                     "filename": file_info.filename,
                     "path": file_info.path,
                     "size": file_info.size,
                     "session_id": file_ref.session_id,
+                    "is_linked_input": False,
                 }
             )
             mounted_ids.add(key)
@@ -399,6 +444,13 @@ async def _auto_mount_session_files(
         session_files = await self.file_service.list_files(ctx.session_id)
 
         for file_info in session_files:
+            file_metadata = await self.file_service.get_file_metadata(
+                ctx.session_id, file_info.file_id
+            )
+            is_linked_input = (
+                file_metadata.get("type") == "linked_input" if file_metadata else False
+            )
+
             # Skip duplicates (shouldn't happen, but defensive)
             key = (ctx.session_id, file_info.file_id)
             if key in mounted_ids:
@@ -411,6 +463,7 @@ async def _auto_mount_session_files(
                     "path": file_info.path,
                     "size": file_info.size,
                     "session_id": ctx.session_id,
+                    "is_linked_input": is_linked_input,
                 }
             )
             mounted_ids.add(key)
@@ -600,6 +653,14 @@ async def _update_mounted_files_content(self, ctx: ExecutionContext) -> None:
                     )
                     continue
 
+                if file_metadata and file_metadata.get("is_read_only") == "1":
+                    logger.debug(
+                        "Skipping update for read-only linked file",
+                        filename=filename,
+                        file_id=file_id,
+                    )
+                    continue
+
                 # Read current content from container
                 file_path = f"/mnt/data/{filename}"
                 content = sandbox_manager.get_file_content_from_sandbox(