feat(B01,B03): 프로젝트 생성 전 임시 보관함 (temp upload)

라이다 원본은 업로드에 오래 걸려 프로젝트 정보 확정 전에 미리 올릴 수 있어야 한다.
계정에 묶인 임시 보관함을 만들고, 나중에 만든 프로젝트로 자료를 옮겨 쓴다.

저장·DB
- storage/tmp/{user_id}/{batch_id}/ 아래에 프로젝트 저장소와 동일한 구조를 써서
  청크 저장·병합 엔진(resolve_upload_destination/merge_upload_chunks)을 그대로 재사용
- 010_temp_upload.sql: temp_upload_batches / temp_upload_files 신설,
  upload_sessions.project_id NULL 허용 + temp_batch_id 추가(FK명 조회 후 재생성)
- config: TEMP_UPLOAD_DIR_NAME / TEMP_UPLOAD_RETENTION_DAYS(30) /
  TEMP_UPLOAD_CLEANUP_INTERVAL_HOURS(6)

백엔드
- B03_FileInput_Router_Temp.py: 묶음 생성·목록·삭제, 일반/청크 업로드, finalize,
  이어올리기 상태 조회, 프로젝트 연결(attach)
- attach: 파일 이동 후 input_files 등록, stage 0 완료, WF1·자동 설계 체인 트리거
- common_util_temp_cleanup.py: 완료 시각 기준 만료분 주기 삭제(서버 시작 시 1회 포함)

프론트엔드
- B01 대시보드 임시 보관함 섹션: 프로젝트 등록과 같은 폼 + 보관 목록.
  진행률은 모달이 아니라 리스트 행에 표시, 새로고침 후 이어올리기 지원
- B03 업로드 컨테이너 내부 불러오기 버튼과 선택 모달.
  완료된 묶음만 노출하고, 선택 후 업로드를 누르면 이동과 분석으로 이어짐

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

B03_FileInput/B03_FileInput_Router_Temp.py

@@ -0,0 +1,659 @@
+"""임시 보관함 라우터 — 프로젝트 생성 전에 계정에 묶어 자료를 올려 둔다.
+
+라이다 원본은 업로드에 오래 걸려서 프로젝트 정보가 확정되기 전에 미리 올릴 수 있어야
+한다(2026-08-08 사용자 지시). 저장 구조를 프로젝트 저장소와 똑같이 맞춰 두어 청크
+업로드·병합 엔진을 그대로 재사용하고, 프로젝트로 옮길 때도 같은 상대 경로로 붙인다.
+"""
+
+import asyncio
+import logging
+import shutil
+from pathlib import Path
+from typing import Any
+from uuid import UUID, uuid4
+
+from fastapi import APIRouter, Depends, File, Form, UploadFile
+from fastapi.responses import JSONResponse
+
+from B03_FileInput.B03_FileInput_Engine import (
+    merge_upload_chunks,
+    remove_chunk_session,
+    resolve_chunk_session_dir,
+    resolve_upload_destination,
+    save_upload_chunk,
+    save_upload_stream,
+)
+from B03_FileInput.B03_FileInput_Engine_Analyze import analyze_input_metadata
+from B03_FileInput.B03_FileInput_Repository import (
+    create_input_file,
+    get_project_storage_relative_path,
+    list_completed_chunk_indexes,
+    mark_upload_session_completed,
+    mark_upload_session_failed,
+    upsert_upload_chunk,
+)
+from B03_FileInput.B03_FileInput_Repository_Temp import (
+    create_temp_batch,
+    create_temp_upload_session,
+    delete_temp_batch,
+    get_temp_batch,
+    get_temp_batch_file_types,
+    get_temp_upload_session,
+    is_batch_required_complete,
+    list_temp_batch_files,
+    list_temp_batch_sessions,
+    list_temp_batches,
+    mark_temp_batch_completed,
+    mark_temp_batch_linked,
+    upsert_temp_batch_file,
+)
+from B03_FileInput.B03_FileInput_Schema import (
+    ChunkSessionCreateRequest,
+    ChunkSessionCreateResponse,
+    ChunkUploadResponse,
+    FileUploadDescriptor,
+    UploadFinalizeRequest,
+    UploadStatusResponse,
+)
+from B03_FileInput.B03_FileInput_Schema_Temp import (
+    TempBatchAttachResponse,
+    TempBatchCreateRequest,
+    TempBatchCreateResponse,
+    TempBatchFile,
+    TempBatchItem,
+    TempBatchListResponse,
+    TempBatchPendingSession,
+    TempFileUploadResponse,
+    TempFileUploadResult,
+)
+from B03_FileInput.B03_FileInput_Service_WF1 import trigger_wf1_analysis_and_email
+from common_util.common_util_auth import verify_session
+from common_util.common_util_storage import (
+    resolve_stored_project_path,
+    resolve_temp_batch_path,
+)
+from common_util.common_util_workflow_state import complete_stage
+from config.config_db import get_db_pool
+from config.config_system import (
+    TEMP_UPLOAD_RETENTION_DAYS,
+    UPLOAD_CHUNK_SIZE_BYTES,
+    UPLOAD_MAX_FILES,
+)
+
+logger = logging.getLogger(__name__)
+router = APIRouter(prefix="/api/temp-uploads", tags=["B03 Temp Upload"])
+attach_router = APIRouter(prefix="/api/projects", tags=["B03 Temp Upload"])
+
+_POINT_CLOUD_FILE_TYPES = frozenset({"las", "laz"})
+
+
+def _total_chunks(size_bytes: int, chunk_size_bytes: int) -> int:
+    return max(1, (size_bytes + chunk_size_bytes - 1) // chunk_size_bytes)
+
+
+def _iso(value: Any) -> str | None:
+    return value.isoformat() if value is not None and hasattr(value, "isoformat") else None
+
+
+async def _batch_root(connection: Any, *, batch_id: str, user_id: int) -> Path:
+    """소유권을 확인하고 묶음 폴더를 돌려준다."""
+    await get_temp_batch(connection, batch_id=batch_id, user_id=user_id)
+    return Path(resolve_temp_batch_path(user_id, batch_id))
+
+
+async def _refresh_batch_status(connection: Any, *, batch_id: str) -> bool:
+    """필수 파일이 다 찼으면 완료로 올린다. 완료 여부를 돌려준다."""
+    file_types = await get_temp_batch_file_types(connection, batch_id=batch_id)
+    complete = is_batch_required_complete(file_types)
+    if complete:
+        await mark_temp_batch_completed(connection, batch_id=batch_id)
+    return complete
+
+
+@router.post("", response_model=TempBatchCreateResponse)
+async def create_batch(
+    payload: TempBatchCreateRequest,
+    session: dict[str, Any] = Depends(verify_session),
+) -> TempBatchCreateResponse | JSONResponse:
+    """보관함 묶음(파일 한 세트)을 만든다."""
+    batch_id = str(uuid4())
+    user_id = int(session["user_id"])
+    pool = get_db_pool()
+    try:
+        async with pool.acquire() as connection:
+            await create_temp_batch(
+                connection,
+                batch_id=batch_id,
+                user_id=user_id,
+                name=payload.name,
+                memo=payload.memo,
+            )
+            await connection.commit()
+        resolve_temp_batch_path(user_id, batch_id)
+        return TempBatchCreateResponse(batch_id=batch_id, name=payload.name)
+    except Exception:
+        logger.exception("임시 보관함 묶음 생성 실패: user_id=%s", user_id)
+        return JSONResponse(
+            status_code=500,
+            content={"status": "error", "message": "보관함 생성 중 오류가 발생했습니다."},
+        )
+
+
+@router.get("", response_model=TempBatchListResponse)
+async def list_batches(
+    session: dict[str, Any] = Depends(verify_session),
+) -> TempBatchListResponse | JSONResponse:
+    """내 보관함 목록 — 파일 목록과 진행 중 세션 진행률을 함께 준다."""
+    user_id = int(session["user_id"])
+    pool = get_db_pool()
+    try:
+        async with pool.acquire() as connection:
+            batches = await list_temp_batches(connection, user_id=user_id)
+            ids = [str(batch["id"]) for batch in batches]
+            files_by_batch = await list_temp_batch_files(connection, batch_ids=ids)
+            sessions_by_batch = await list_temp_batch_sessions(connection, batch_ids=ids)
+
+        items: list[TempBatchItem] = []
+        for batch in batches:
+            batch_id = str(batch["id"])
+            files = files_by_batch.get(batch_id, [])
+            sessions = sessions_by_batch.get(batch_id, [])
+            file_types = {str(item["file_type"]).lower() for item in files}
+            items.append(
+                TempBatchItem(
+                    batch_id=batch_id,
+                    name=str(batch["name"]),
+                    memo=batch.get("memo"),
+                    status=str(batch["status"]),
+                    files=[
+                        TempBatchFile(
+                            file_type=str(item["file_type"]),
+                            original_filename=str(item["original_filename"]),
+                            file_size_bytes=int(item["file_size_bytes"]),
+                            crs_epsg=item.get("crs_epsg"),
+                        )
+                        for item in files
+                    ],
+                    pending_sessions=[
+                        TempBatchPendingSession(
+                            upload_session_id=str(item["id"]),
+                            original_filename=str(item["original_filename"]),
+                            file_size_bytes=int(item["file_size_bytes"]),
+                            total_chunks=int(item["total_chunks"]),
+                            completed_chunks=int(item["completed_chunks"]),
+                            progress_percent=round(
+                                int(item["completed_chunks"])
+                                / max(1, int(item["total_chunks"]))
+                                * 100,
+                                1,
+                            ),
+                        )
+                        for item in sessions
+                    ],
+                    total_size_bytes=sum(int(item["file_size_bytes"]) for item in files),
+                    required_complete=is_batch_required_complete(file_types),
+                    completed_at=_iso(batch.get("completed_at")),
+                    expires_at=_iso(batch.get("expires_at")),
+                    linked_project_id=(
+                        str(batch["linked_project_id"]) if batch.get("linked_project_id") else None
+                    ),
+                    created_at=_iso(batch.get("created_at")),
+                )
+            )
+        return TempBatchListResponse(batches=items, retention_days=TEMP_UPLOAD_RETENTION_DAYS)
+    except Exception:
+        logger.exception("임시 보관함 목록 조회 실패: user_id=%s", user_id)
+        return JSONResponse(
+            status_code=500,
+            content={"status": "error", "message": "보관함 조회 중 오류가 발생했습니다."},
+        )
+
+
+@router.delete("/{batch_id}")
+async def remove_batch(
+    batch_id: str,
+    session: dict[str, Any] = Depends(verify_session),
+) -> JSONResponse:
+    """보관함 묶음과 저장 파일을 지운다."""
+    user_id = int(session["user_id"])
+    pool = get_db_pool()
+    try:
+        async with pool.acquire() as connection:
+            await get_temp_batch(connection, batch_id=batch_id, user_id=user_id)
+            await delete_temp_batch(connection, batch_id=batch_id, user_id=user_id)
+            await connection.commit()
+        batch_root = Path(resolve_temp_batch_path(user_id, batch_id, create=False))
+        shutil.rmtree(batch_root, ignore_errors=True)
+        return JSONResponse(content={"status": "success", "batch_id": batch_id})
+    except LookupError as exc:
+        return JSONResponse(status_code=404, content={"status": "error", "message": str(exc)})
+    except Exception:
+        logger.exception("임시 보관함 삭제 실패: batch_id=%s", batch_id)
+        return JSONResponse(
+            status_code=500,
+            content={"status": "error", "message": "보관함 삭제 중 오류가 발생했습니다."},
+        )
+
+
+@router.post("/{batch_id}/files", response_model=TempFileUploadResponse)
+async def upload_batch_files(
+    batch_id: str,
+    files: list[UploadFile] = File(...),
+    session: dict[str, Any] = Depends(verify_session),
+) -> TempFileUploadResponse | JSONResponse:
+    """작은 파일(csv·prj·tfw·tif)을 보관함에 바로 저장한다."""
+    user_id = int(session["user_id"])
+    if not files or len(files) > UPLOAD_MAX_FILES:
+        message = f"파일은 1~{UPLOAD_MAX_FILES}개까지 가능합니다."
+        return JSONResponse(status_code=400, content={"status": "error", "message": message})
+    pool = get_db_pool()
+    try:
+        async with pool.acquire() as connection:
+            batch_root = await _batch_root(connection, batch_id=batch_id, user_id=user_id)
+            results: list[TempFileUploadResult] = []
+            for upload in files:
+                descriptor = FileUploadDescriptor(
+                    original_filename=upload.filename or "",
+                    size_bytes=max(1, upload.size or 1),
+                )
+                destination = resolve_upload_destination(batch_root, descriptor)
+                written_bytes = await save_upload_stream(upload, destination)
+                metadata = await asyncio.to_thread(analyze_input_metadata, destination)
+                relative_path = destination.relative_to(batch_root).as_posix()
+                file_type = destination.suffix.lower().lstrip(".")
+                crs_epsg = metadata.get("epsg")
+                await upsert_temp_batch_file(
+                    connection,
+                    batch_id=batch_id,
+                    file_type=file_type,
+                    original_filename=descriptor.original_filename,
+                    relative_path=relative_path,
+                    file_size_bytes=written_bytes,
+                    crs_epsg=int(crs_epsg) if crs_epsg is not None else None,
+                    metadata=metadata,
+                )
+                results.append(
+                    TempFileUploadResult(
+                        batch_id=batch_id,
+                        file_type=file_type,
+                        original_filename=descriptor.original_filename,
+                        relative_path=relative_path,
+                        size_bytes=written_bytes,
+                        metadata=metadata,
+                    )
+                )
+            required_complete = await _refresh_batch_status(connection, batch_id=batch_id)
+            await connection.commit()
+        return TempFileUploadResponse(
+            batch_id=batch_id, files=results, required_complete=required_complete
+        )
+    except LookupError as exc:
+        return JSONResponse(status_code=404, content={"status": "error", "message": str(exc)})
+    except (OSError, ValueError) as exc:
+        return JSONResponse(status_code=400, content={"status": "error", "message": str(exc)})
+    except Exception:
+        logger.exception("임시 보관함 파일 저장 실패: batch_id=%s", batch_id)
+        return JSONResponse(
+            status_code=500,
+            content={"status": "error", "message": "파일 저장 중 오류가 발생했습니다."},
+        )
+    finally:
+        for upload in files:
+            await upload.close()
+
+
+@router.post("/{batch_id}/upload-sessions", response_model=ChunkSessionCreateResponse)
+async def create_batch_upload_session(
+    batch_id: str,
+    payload: ChunkSessionCreateRequest,
+    session: dict[str, Any] = Depends(verify_session),
+) -> ChunkSessionCreateResponse | JSONResponse:
+    """대용량 파일(LAS/LAZ) 청크 세션을 만든다 — 프로젝트 업로드와 같은 규칙."""
+    user_id = int(session["user_id"])
+    chunk_size_bytes = min(payload.chunk_size_bytes, UPLOAD_CHUNK_SIZE_BYTES)
+    total_chunks = _total_chunks(payload.size_bytes, chunk_size_bytes)
+    session_id = str(uuid4())
+    pool = get_db_pool()
+    try:
+        async with pool.acquire() as connection:
+            await _batch_root(connection, batch_id=batch_id, user_id=user_id)
+            await create_temp_upload_session(
+                connection,
+                session_id=session_id,
+                batch_id=batch_id,
+                original_filename=payload.original_filename,
+                file_size_bytes=payload.size_bytes,
+                chunk_size_bytes=chunk_size_bytes,
+                total_chunks=total_chunks,
+            )
+            await connection.commit()
+        return ChunkSessionCreateResponse(
+            project_id=batch_id,
+            upload_session_id=session_id,
+            original_filename=payload.original_filename,
+            file_size_bytes=payload.size_bytes,
+            chunk_size_bytes=chunk_size_bytes,
+            total_chunks=total_chunks,
+        )
+    except LookupError as exc:
+        return JSONResponse(status_code=404, content={"status": "error", "message": str(exc)})
+    except (OSError, ValueError) as exc:
+        return JSONResponse(status_code=400, content={"status": "error", "message": str(exc)})
+    except Exception:
+        logger.exception("임시 보관함 청크 세션 생성 실패: batch_id=%s", batch_id)
+        return JSONResponse(
+            status_code=500,
+            content={"status": "error", "message": "업로드 세션 생성 중 오류가 발생했습니다."},
+        )
+
+
+@router.post("/{batch_id}/chunks", response_model=ChunkUploadResponse)
+async def upload_batch_chunk(
+    batch_id: str,
+    session_id: str = Form(...),
+    chunk_index: int = Form(...),
+    chunk_data: UploadFile = File(...),
+    session: dict[str, Any] = Depends(verify_session),
+) -> ChunkUploadResponse | JSONResponse:
+    """청크 한 조각을 보관함 묶음 폴더에 저장한다."""
+    user_id = int(session["user_id"])
+    pool = get_db_pool()
+    try:
+        async with pool.acquire() as connection:
+            batch_root = await _batch_root(connection, batch_id=batch_id, user_id=user_id)
+            upload_session = await get_temp_upload_session(
+                connection, batch_id=batch_id, session_id=session_id
+            )
+            if chunk_index < 0 or chunk_index >= int(upload_session["total_chunks"]):
+                return JSONResponse(
+                    status_code=400,
+                    content={"status": "error", "message": "청크 인덱스가 범위를 벗어났습니다."},
+                )
+            session_dir = resolve_chunk_session_dir(batch_root, session_id)
+            chunk_path, size_bytes, chunk_hash = await save_upload_chunk(
+                chunk_data,
+                session_dir,
+                chunk_index,
+                expected_max_bytes=int(upload_session["chunk_size_bytes"]),
+            )
+            completed_chunks = await upsert_upload_chunk(
+                connection,
+                session_id=session_id,
+                chunk_index=chunk_index,
+                chunk_hash=chunk_hash,
+                size_bytes=size_bytes,
+                stored_at=chunk_path.relative_to(batch_root).as_posix(),
+            )
+        return ChunkUploadResponse(
+            upload_session_id=session_id,
+            chunk_index=chunk_index,
+            completed_chunks=completed_chunks,
+            total_chunks=int(upload_session["total_chunks"]),
+            chunk_hash=chunk_hash,
+        )
+    except LookupError as exc:
+        return JSONResponse(status_code=404, content={"status": "error", "message": str(exc)})
+    except (OSError, ValueError) as exc:
+        return JSONResponse(status_code=400, content={"status": "error", "message": str(exc)})
+    except Exception:
+        logger.exception("임시 보관함 청크 업로드 실패: batch_id=%s", batch_id)
+        return JSONResponse(
+            status_code=500,
+            content={"status": "error", "message": "청크 업로드 중 오류가 발생했습니다."},
+        )
+    finally:
+        await chunk_data.close()
+
+
+@router.get("/{batch_id}/upload-status/{session_id}", response_model=UploadStatusResponse)
+async def get_batch_upload_status(
+    batch_id: str,
+    session_id: str,
+    session: dict[str, Any] = Depends(verify_session),
+) -> UploadStatusResponse | JSONResponse:
+    """이어올리기용 — 이미 올라간 청크 번호를 돌려준다."""
+    user_id = int(session["user_id"])
+    pool = get_db_pool()
+    try:
+        async with pool.acquire() as connection:
+            await get_temp_batch(connection, batch_id=batch_id, user_id=user_id)
+            upload_session = await get_temp_upload_session(
+                connection, batch_id=batch_id, session_id=session_id
+            )
+            completed_indexes = await list_completed_chunk_indexes(
+                connection, session_id=session_id
+            )
+        return UploadStatusResponse(
+            upload_session_id=session_id,
+            upload_status=str(upload_session["status"]),
+            original_filename=str(upload_session["original_filename"]),
+            file_size_bytes=int(upload_session["file_size_bytes"]),
+            chunk_size_bytes=int(upload_session["chunk_size_bytes"]),
+            total_chunks=int(upload_session["total_chunks"]),
+            completed_chunks=len(completed_indexes),
+            completed_chunk_indexes=completed_indexes,
+        )
+    except LookupError as exc:
+        return JSONResponse(status_code=404, content={"status": "error", "message": str(exc)})
+    except Exception:
+        logger.exception("임시 보관함 업로드 상태 조회 실패: batch_id=%s", batch_id)
+        return JSONResponse(
+            status_code=500,
+            content={"status": "error", "message": "업로드 상태 조회 중 오류가 발생했습니다."},
+        )
+
+
+@router.post("/{batch_id}/finalize", response_model=TempFileUploadResponse)
+async def finalize_batch_upload(
+    batch_id: str,
+    payload: UploadFinalizeRequest,
+    session: dict[str, Any] = Depends(verify_session),
+) -> TempFileUploadResponse | JSONResponse:
+    """청크를 병합해 보관함에 저장하고, 필수 파일이 다 차면 완료로 올린다."""
+    user_id = int(session["user_id"])
+    pool = get_db_pool()
+    final_path: Path | None = None
+    try:
+        async with pool.acquire() as connection:
+            batch_root = await _batch_root(connection, batch_id=batch_id, user_id=user_id)
+            upload_session = await get_temp_upload_session(
+                connection, batch_id=batch_id, session_id=payload.session_id
+            )
+            if int(upload_session["total_chunks"]) != payload.total_chunks:
+                return JSONResponse(
+                    status_code=400,
+                    content={"status": "error", "message": "세션 청크 개수가 일치하지 않습니다."},
+                )
+            completed_indexes = await list_completed_chunk_indexes(
+                connection, session_id=payload.session_id
+            )
+            if completed_indexes != list(range(payload.total_chunks)):
+                return JSONResponse(
+                    status_code=400,
+                    content={"status": "error", "message": "아직 업로드되지 않은 청크가 있습니다."},
+                )
+
+            descriptor = FileUploadDescriptor(
+                original_filename=str(upload_session["original_filename"]),
+                size_bytes=int(upload_session["file_size_bytes"]),
+            )
+            final_path = merge_upload_chunks(
+                batch_root, descriptor, payload.session_id, payload.total_chunks
+            )
+            metadata = await asyncio.to_thread(analyze_input_metadata, final_path)
+            relative_path = final_path.relative_to(batch_root).as_posix()
+            file_type = final_path.suffix.lower().lstrip(".")
+            crs_epsg = metadata.get("epsg")
+
+            await connection.begin()
+            try:
+                await upsert_temp_batch_file(
+                    connection,
+                    batch_id=batch_id,
+                    file_type=file_type,
+                    original_filename=descriptor.original_filename,
+                    relative_path=relative_path,
+                    file_size_bytes=int(upload_session["file_size_bytes"]),
+                    crs_epsg=int(crs_epsg) if crs_epsg is not None else None,
+                    metadata=metadata,
+                )
+                await mark_upload_session_completed(connection, session_id=payload.session_id)
+                required_complete = await _refresh_batch_status(connection, batch_id=batch_id)
+                await connection.commit()
+            except Exception:
+                await connection.rollback()
+                await mark_upload_session_failed(connection, session_id=payload.session_id)
+                raise
+
+        remove_chunk_session(batch_root, payload.session_id)
+        return TempFileUploadResponse(
+            batch_id=batch_id,
+            files=[
+                TempFileUploadResult(
+                    batch_id=batch_id,
+                    file_type=file_type,
+                    original_filename=descriptor.original_filename,
+                    relative_path=relative_path,
+                    size_bytes=int(upload_session["file_size_bytes"]),
+                    metadata=metadata,
+                )
+            ],
+            required_complete=required_complete,
+        )
+    except LookupError as exc:
+        return JSONResponse(status_code=404, content={"status": "error", "message": str(exc)})
+    except (OSError, ValueError) as exc:
+        if final_path is not None:
+            final_path.unlink(missing_ok=True)
+        return JSONResponse(status_code=400, content={"status": "error", "message": str(exc)})
+    except Exception:
+        if final_path is not None:
+            final_path.unlink(missing_ok=True)
+        logger.exception("임시 보관함 병합 실패: batch_id=%s", batch_id)
+        return JSONResponse(
+            status_code=500,
+            content={"status": "error", "message": "업로드 최종 처리 중 오류가 발생했습니다."},
+        )
+
+
+@attach_router.post(
+    "/{project_id}/temp-uploads/{batch_id}/attach", response_model=TempBatchAttachResponse
+)
+async def attach_temp_batch(
+    project_id: UUID,
+    batch_id: str,
+    session: dict[str, Any] = Depends(verify_session),
+) -> TempBatchAttachResponse | JSONResponse:
+    """보관함 자료를 프로젝트 영구저장소로 옮기고 초기 분석을 시작한다.
+
+    파일 이동 → `input_files` 등록 → stage 0 완료 → WF1·자동 설계 체인까지, B03에서
+    직접 업로드했을 때와 같은 흐름을 탄다.
+    """
+    from B03_FileInput.B03_FileInput_Router import _schedule_background_task
+
+    user_id = int(session["user_id"])
+    pool = get_db_pool()
+    try:
+        async with pool.acquire() as connection:
+            batch = await get_temp_batch(connection, batch_id=batch_id, user_id=user_id)
+            if str(batch["status"]) == "linked":
+                return JSONResponse(
+                    status_code=400,
+                    content={"status": "error", "message": "이미 프로젝트에 연결된 보관함입니다."},
+                )
+            files = (await list_temp_batch_files(connection, batch_ids=[batch_id])).get(
+                batch_id, []
+            )
+            file_types = {str(item["file_type"]).lower() for item in files}
+            if not is_batch_required_complete(file_types):
+                return JSONResponse(
+                    status_code=400,
+                    content={
+                        "status": "error",
+                        "message": "필수 파일이 모두 갖춰진 보관함만 연결할 수 있습니다.",
+                    },
+                )
+            stored_path = await get_project_storage_relative_path(connection, project_id)
+
+        batch_root = Path(resolve_temp_batch_path(user_id, batch_id, create=False))
+        project_root = Path(resolve_stored_project_path(stored_path))
+
+        moved: list[dict[str, Any]] = []
+        for item in files:
+            source = batch_root / str(item["relative_path"])
+            if not source.is_file():
+                name = item["original_filename"]
+                raise FileNotFoundError(f"보관함 파일을 찾을 수 없습니다: {name}")
+            destination = project_root / str(item["relative_path"])
+            destination.parent.mkdir(parents=True, exist_ok=True)
+            await asyncio.to_thread(shutil.move, str(source), str(destination))
+            moved.append({**item, "destination": destination})
+
+        point_cloud_input_id: int | None = None
+        async with pool.acquire() as connection:
+            await connection.begin()
+            try:
+                for item in moved:
+                    metadata = item.get("metadata")
+                    if isinstance(metadata, str):
+                        import json as _json
+
+                        metadata = _json.loads(metadata)
+                    input_file_id = await create_input_file(
+                        connection,
+                        project_id=project_id,
+                        file_type=str(item["file_type"]),
+                        original_filename=str(item["original_filename"]),
+                        relative_path=str(item["relative_path"]),
+                        file_size_bytes=int(item["file_size_bytes"]),
+                        upload_by=user_id,
+                        crs_epsg=item.get("crs_epsg"),
+                        metadata=metadata or {},
+                    )
+                    if str(item["file_type"]).lower() in _POINT_CLOUD_FILE_TYPES:
+                        point_cloud_input_id = input_file_id
+                async with connection.cursor() as cursor:
+                    await complete_stage(cursor, str(project_id), 0)
+                await mark_temp_batch_linked(
+                    connection, batch_id=batch_id, project_id=str(project_id)
+                )
+                await connection.commit()
+            except Exception:
+                await connection.rollback()
+                raise
+
+        shutil.rmtree(batch_root, ignore_errors=True)
+
+        analysis_started = point_cloud_input_id is not None
+        if analysis_started:
+            _schedule_background_task(
+                trigger_wf1_analysis_and_email(
+                    project_id=project_id,
+                    input_file_id=point_cloud_input_id,
+                    user_role=str(session["role"]),
+                ),
+                task_name=f"b04-preprocess-auto-{project_id}",
+            )
+        logger.info(
+            "보관함 연결 완료: project_id=%s batch_id=%s 파일=%d건 분석시작=%s",
+            project_id,
+            batch_id,
+            len(moved),
+            analysis_started,
+        )
+        return TempBatchAttachResponse(
+            project_id=str(project_id),
+            batch_id=batch_id,
+            moved_files=len(moved),
+            analysis_started=analysis_started,
+        )
+    except LookupError as exc:
+        return JSONResponse(status_code=404, content={"status": "error", "message": str(exc)})
+    except (OSError, ValueError) as exc:
+        return JSONResponse(status_code=400, content={"status": "error", "message": str(exc)})
+    except Exception:
+        logger.exception("보관함 연결 실패: project_id=%s batch_id=%s", project_id, batch_id)
+        return JSONResponse(
+            status_code=500,
+            content={"status": "error", "message": "보관함 연결 중 오류가 발생했습니다."},
+        )