feat(inventory): implement inter-project material transfer backend

- Add ProjectTransfer database model and migration
- Implement project transfer service with immediate stock deduction on creation
- Support 100% full delivery receipt logic across destination warehouses
- Enforce canonical_specs_hash matching for cross-project stock merge
- Add RBAC guards and status lock on received transfers (409 Conflict)
- Register API router for transfers with on-the-way integration

backend/app/db/init_db.py

@@ -432,3 +432,48 @@ async def ensure_audit_table(engine: AsyncEngine) -> None:
         for stmt in indexes:
             await conn.execute(text(stmt))
     logger.info("audit_log table ensured")
+
+
+async def ensure_project_transfers_table(engine: AsyncEngine) -> None:
+    """Create (if missing) the project_transfers table for inter-site transfers. Safe on any volume."""
+    ddl = """
+    CREATE TABLE IF NOT EXISTS project_transfers (
+        id                  UUID            NOT NULL DEFAULT gen_random_uuid() PRIMARY KEY,
+        from_project_id     UUID            NOT NULL REFERENCES projects(id) ON DELETE RESTRICT,
+        from_warehouse_id   UUID            NOT NULL REFERENCES warehouses(id) ON DELETE RESTRICT,
+        to_project_id       UUID            NOT NULL REFERENCES projects(id) ON DELETE RESTRICT,
+        material_id         UUID            NOT NULL REFERENCES materials(id) ON DELETE RESTRICT,
+        brand_id            UUID            REFERENCES brands(id) ON DELETE SET NULL,
+        unit_id             INTEGER         NOT NULL REFERENCES units(id) ON DELETE RESTRICT,
+        quantity            NUMERIC(18,6)   NOT NULL CHECK (quantity > 0),
+        specs               JSONB           NOT NULL DEFAULT '{}'::jsonb,
+        specs_hash          TEXT            NOT NULL DEFAULT 'default',
+        note                TEXT,
+        status              TEXT            NOT NULL DEFAULT 'in_transit' CHECK (status IN ('in_transit', 'received', 'cancelled')),
+        out_transaction_id  UUID            REFERENCES inventory_transactions(id) ON DELETE SET NULL,
+        transfer_date       TIMESTAMPTZ     NOT NULL DEFAULT now(),
+        transferred_by      TEXT,
+        received_at         TIMESTAMPTZ,
+        received_by         TEXT,
+        to_warehouse_id     UUID            REFERENCES warehouses(id) ON DELETE SET NULL,
+        receipt_allocations JSONB,
+        created_at          TIMESTAMPTZ     NOT NULL DEFAULT now(),
+        updated_at          TIMESTAMPTZ     NOT NULL DEFAULT now()
+    )
+    """
+    indexes = (
+        "CREATE INDEX IF NOT EXISTS idx_project_transfers_from ON project_transfers(from_project_id, status)",
+        "CREATE INDEX IF NOT EXISTS idx_project_transfers_to ON project_transfers(to_project_id, status)",
+        "DROP TRIGGER IF EXISTS trg_project_transfers_updated_at ON project_transfers",
+        """
+        CREATE TRIGGER trg_project_transfers_updated_at
+            BEFORE UPDATE ON project_transfers
+            FOR EACH ROW EXECUTE FUNCTION set_updated_at()
+        """,
+    )
+    async with engine.begin() as conn:
+        await conn.execute(text(ddl))
+        for stmt in indexes:
+            await conn.execute(text(stmt))
+    logger.info("project_transfers table ensured")
+

backend/app/main.py

@@ -31,6 +31,7 @@ from app.db.init_db import (
     ensure_dashboard_tables,
     ensure_material_topic_column,
     ensure_quote_projects,
+    ensure_project_transfers_table,
     ensure_v26,
 )
 from app.observability import RequestContextMiddleware, setup_logging
@@ -107,6 +108,7 @@ async def lifespan(_app: FastAPI):
     await ensure_v26(engine)
     await ensure_audit_table(engine)
     await ensure_drop_category_tables(engine)
+    await ensure_project_transfers_table(engine)
     task = asyncio.create_task(_auto_expire_loop()) if settings.AUTO_EXPIRE_ENABLED else None
     try:
         yield

backend/app/models/__init__.py

@@ -32,6 +32,8 @@ from app.models.procurement import (
     TermsheetScheduleStep,
 )
 
+from app.models.project_transfer import ProjectTransfer
+
 __all__ = [
     "Base",
     "Unit",
@@ -60,4 +62,5 @@ __all__ = [
     "ItemDecision",
     "ItemReceipt",
     "QuoteAttachment",
+    "ProjectTransfer",
 ]

backend/app/models/project_transfer.py

@@ -0,0 +1,87 @@
+import uuid
+from datetime import datetime
+from decimal import Decimal
+
+from sqlalchemy import (
+    DateTime,
+    ForeignKey,
+    Numeric,
+    Text,
+    func,
+    text,
+)
+from sqlalchemy.dialects.postgresql import JSONB
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+
+from app.db.base import Base
+from app.models.catalog import _timestamptz, _uuid_pk
+
+
+class ProjectTransfer(Base):
+    """project_transfers — Inter-project (inter-site) material transfer ledger.
+
+    Records the movement of materials from a source project warehouse to a
+    destination project. Stock is deducted from the source warehouse immediately
+    upon creation (status = 'in_transit') via out_transaction_id.
+    When the destination project PM confirms receipt in 'Arrivals', stock is
+    added to the destination warehouse (status = 'received').
+    While 'in_transit', the transfer can be edited or cancelled (restoring source stock).
+    Once 'received', the transfer is permanently locked from edit or cancellation.
+    """
+
+    __tablename__ = "project_transfers"
+    __mapper_args__ = {"eager_defaults": True}
+
+    id: Mapped[uuid.UUID] = _uuid_pk()
+    from_project_id: Mapped[uuid.UUID] = mapped_column(
+        ForeignKey("projects.id", ondelete="RESTRICT"), nullable=False
+    )
+    from_warehouse_id: Mapped[uuid.UUID] = mapped_column(
+        ForeignKey("warehouses.id", ondelete="RESTRICT"), nullable=False
+    )
+    to_project_id: Mapped[uuid.UUID] = mapped_column(
+        ForeignKey("projects.id", ondelete="RESTRICT"), nullable=False
+    )
+    material_id: Mapped[uuid.UUID] = mapped_column(
+        ForeignKey("materials.id", ondelete="RESTRICT"), nullable=False
+    )
+    brand_id: Mapped[uuid.UUID | None] = mapped_column(
+        ForeignKey("brands.id", ondelete="SET NULL")
+    )
+    unit_id: Mapped[int] = mapped_column(
+        ForeignKey("units.id", ondelete="RESTRICT"), nullable=False
+    )
+    quantity: Mapped[Decimal] = mapped_column(Numeric(18, 6), nullable=False)
+    specs: Mapped[dict] = mapped_column(JSONB, nullable=False, server_default=text("'{}'::jsonb"))
+    specs_hash: Mapped[str] = mapped_column(Text, nullable=False, server_default=text("'default'"))
+    note: Mapped[str | None] = mapped_column(Text)
+    status: Mapped[str] = mapped_column(Text, nullable=False, server_default=text("'in_transit'"))
+
+    out_transaction_id: Mapped[uuid.UUID | None] = mapped_column(
+        ForeignKey("inventory_transactions.id", ondelete="SET NULL")
+    )
+    transfer_date: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, server_default=func.now()
+    )
+    transferred_by: Mapped[str | None] = mapped_column(Text)
+    received_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+    received_by: Mapped[str | None] = mapped_column(Text)
+    to_warehouse_id: Mapped[uuid.UUID | None] = mapped_column(
+        ForeignKey("warehouses.id", ondelete="SET NULL")
+    )
+    receipt_allocations: Mapped[list | None] = mapped_column(JSONB)
+    created_at: Mapped[datetime] = _timestamptz()
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now()
+    )
+
+    from_project: Mapped["Project"] = relationship("Project", foreign_keys=[from_project_id], lazy="raise")
+    to_project: Mapped["Project"] = relationship("Project", foreign_keys=[to_project_id], lazy="raise")
+    from_warehouse: Mapped["Warehouse"] = relationship("Warehouse", foreign_keys=[from_warehouse_id], lazy="raise")
+    to_warehouse: Mapped["Warehouse | None"] = relationship("Warehouse", foreign_keys=[to_warehouse_id], lazy="raise")
+    material: Mapped["Material"] = relationship("Material", foreign_keys=[material_id], lazy="raise")
+    brand: Mapped["Brand | None"] = relationship("Brand", foreign_keys=[brand_id], lazy="raise")
+    unit: Mapped["Unit"] = relationship("Unit", foreign_keys=[unit_id], lazy="raise")
+    out_transaction: Mapped["InventoryTransaction | None"] = relationship(
+        "InventoryTransaction", foreign_keys=[out_transaction_id], lazy="raise"
+    )

backend/app/routers/__init__.py

@@ -14,6 +14,7 @@ from app.routers import (
     material_categories,
     materials,
     pm,
+    project_transfers,
     projects,
     quote_events,
     quotes,
@@ -50,5 +51,6 @@ ROUTERS = [
     standard_conversions.router,
     inventory.router,
     pm.router,
+    project_transfers.router,
     users.router,
 ]

backend/app/routers/project_transfers.py

@@ -0,0 +1,205 @@
+import uuid
+from typing import Literal
+
+from fastapi import APIRouter, Depends, Query, Request
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.db.engine import get_db
+from app.routers.pm import ArrivalScope, arrivals_scope
+from app.schemas.project_transfers import (
+    ProjectTransferCreate,
+    ProjectTransferOut,
+    ProjectTransferReceive,
+    ProjectTransferUpdate,
+)
+from app.services import project_transfer_service
+from app.utils.errors import ForbiddenError
+
+router = APIRouter(tags=["project-transfers"])
+
+
+def _to_transfer_out(t) -> dict:
+    return {
+        "id": t.id,
+        "from_project": (
+            {"id": t.from_project.id, "name": t.from_project.name}
+            if t.from_project
+            else {"id": t.from_project_id, "name": ""}
+        ),
+        "from_warehouse": (
+            {"id": t.from_warehouse.id, "name": t.from_warehouse.name}
+            if t.from_warehouse
+            else {"id": t.from_warehouse_id, "name": ""}
+        ),
+        "to_project": (
+            {"id": t.to_project.id, "name": t.to_project.name}
+            if t.to_project
+            else {"id": t.to_project_id, "name": ""}
+        ),
+        "to_warehouse": (
+            {"id": t.to_warehouse.id, "name": t.to_warehouse.name}
+            if t.to_warehouse
+            else None
+        ),
+        "material": (
+            {"id": t.material.id, "name": t.material.name}
+            if t.material
+            else {"id": t.material_id, "name": ""}
+        ),
+        "brand": (
+            {"id": t.brand.id, "name": t.brand.name}
+            if t.brand
+            else None
+        ),
+        "unit": (
+            {"id": t.unit.id, "code": t.unit.code, "name_fa": t.unit.name_fa}
+            if t.unit
+            else {"id": t.unit_id, "code": "", "name_fa": None}
+        ),
+        "quantity": t.quantity,
+        "specs": t.specs or {},
+        "specs_hash": t.specs_hash,
+        "note": t.note,
+        "status": t.status,
+        "transfer_date": t.transfer_date,
+        "transferred_by": t.transferred_by,
+        "received_at": t.received_at,
+        "received_by": t.received_by,
+        "receipt_allocations": t.receipt_allocations,
+        "created_at": t.created_at,
+        "updated_at": t.updated_at,
+    }
+
+
+# ---------------------------------------------------------------------------
+# POST /inventory/project-transfers — Create inter-project transfer
+# ---------------------------------------------------------------------------
+@router.post("/inventory/project-transfers", status_code=201, response_model=ProjectTransferOut)
+async def create_transfer(
+    payload: ProjectTransferCreate,
+    ctx: ArrivalScope = Depends(arrivals_scope),
+    db: AsyncSession = Depends(get_db),
+):
+    transfer = await project_transfer_service.create_project_transfer(
+        db=db,
+        payload=payload,
+        caller_project_id=ctx.project_id,
+        caller_role=ctx.role,
+        transferred_by=ctx.display_name,
+    )
+    return _to_transfer_out(transfer)
+
+
+# ---------------------------------------------------------------------------
+# GET /inventory/project-transfers — List transfers with direction and filters
+# ---------------------------------------------------------------------------
+@router.get("/inventory/project-transfers")
+async def list_transfers(
+    project_id: uuid.UUID | None = Query(default=None),
+    direction: Literal["outgoing", "incoming", "all"] = Query(default="all"),
+    status: str | None = Query(default=None),
+    ctx: ArrivalScope = Depends(arrivals_scope),
+    db: AsyncSession = Depends(get_db),
+):
+    target_pid = project_id or ctx.project_id
+    transfers = await project_transfer_service.list_project_transfers(
+        db=db,
+        project_id=target_pid,
+        direction=direction,
+        status=status,
+    )
+    return [_to_transfer_out(t) for t in transfers]
+
+
+# ---------------------------------------------------------------------------
+# GET /inventory/project-transfers/{id} — Single transfer details
+# ---------------------------------------------------------------------------
+@router.get("/inventory/project-transfers/{transfer_id}", response_model=ProjectTransferOut)
+async def get_transfer(
+    transfer_id: uuid.UUID,
+    ctx: ArrivalScope = Depends(arrivals_scope),
+    db: AsyncSession = Depends(get_db),
+):
+    transfer = await project_transfer_service.get_project_transfer_or_404(db, transfer_id)
+    if ctx.role == "pm" and ctx.project_id is not None:
+        if transfer.from_project_id != ctx.project_id and transfer.to_project_id != ctx.project_id:
+            raise ForbiddenError("شما به این رکورد انتقال دسترسی ندارید")
+    return _to_transfer_out(transfer)
+
+
+# ---------------------------------------------------------------------------
+# PATCH /inventory/project-transfers/{id} — Edit in-transit transfer (source PM only)
+# ---------------------------------------------------------------------------
+@router.patch("/inventory/project-transfers/{transfer_id}", response_model=ProjectTransferOut)
+async def update_transfer(
+    transfer_id: uuid.UUID,
+    payload: ProjectTransferUpdate,
+    ctx: ArrivalScope = Depends(arrivals_scope),
+    db: AsyncSession = Depends(get_db),
+):
+    transfer = await project_transfer_service.update_project_transfer(
+        db=db,
+        transfer_id=transfer_id,
+        payload=payload,
+        caller_project_id=ctx.project_id,
+        caller_role=ctx.role,
+    )
+    return _to_transfer_out(transfer)
+
+
+# ---------------------------------------------------------------------------
+# DELETE /inventory/project-transfers/{id} — Cancel in-transit transfer (source PM only)
+# ---------------------------------------------------------------------------
+@router.delete("/inventory/project-transfers/{transfer_id}")
+async def delete_transfer(
+    transfer_id: uuid.UUID,
+    ctx: ArrivalScope = Depends(arrivals_scope),
+    db: AsyncSession = Depends(get_db),
+):
+    return await project_transfer_service.delete_project_transfer(
+        db=db,
+        transfer_id=transfer_id,
+        caller_project_id=ctx.project_id,
+        caller_role=ctx.role,
+    )
+
+
+# ---------------------------------------------------------------------------
+# GET /pm/transfers/on-the-way — Incoming in-transit transfers for active project
+# ---------------------------------------------------------------------------
+@router.get("/pm/transfers/on-the-way")
+async def pm_incoming_transfers(
+    ctx: ArrivalScope = Depends(arrivals_scope),
+    db: AsyncSession = Depends(get_db),
+):
+    if ctx.role == "pm" and ctx.project_id is None:
+        raise ForbiddenError("شما به هیچ پروژه‌ای متصل نیستید")
+
+    transfers = await project_transfer_service.list_project_transfers(
+        db=db,
+        project_id=ctx.project_id,
+        direction="incoming",
+        status="in_transit",
+    )
+    return [_to_transfer_out(t) for t in transfers]
+
+
+# ---------------------------------------------------------------------------
+# POST /inventory/project-transfers/{id}/receive — Destination PM confirms arrival
+# ---------------------------------------------------------------------------
+@router.post("/inventory/project-transfers/{transfer_id}/receive", response_model=ProjectTransferOut)
+async def receive_transfer(
+    transfer_id: uuid.UUID,
+    payload: ProjectTransferReceive,
+    ctx: ArrivalScope = Depends(arrivals_scope),
+    db: AsyncSession = Depends(get_db),
+):
+    transfer = await project_transfer_service.receive_project_transfer(
+        db=db,
+        transfer_id=transfer_id,
+        payload=payload,
+        caller_project_id=ctx.project_id,
+        caller_role=ctx.role,
+        received_by=ctx.display_name,
+    )
+    return _to_transfer_out(transfer)

backend/app/schemas/project_transfers.py

@@ -0,0 +1,74 @@
+from datetime import datetime
+from decimal import Decimal
+from uuid import UUID
+
+from pydantic import Field
+
+from app.schemas.common import APIModel
+
+
+class WarehouseAllocationIn(APIModel):
+    warehouse_id: UUID
+    quantity: Decimal = Field(gt=0)
+
+
+class ProjectTransferCreate(APIModel):
+    from_warehouse_id: UUID
+    to_project_id: UUID
+    material_id: UUID
+    brand_id: UUID | None = None
+    unit_id: int
+    quantity: Decimal = Field(gt=0)
+    specs: dict = Field(default_factory=dict)
+    note: str | None = None
+    transfer_date: datetime | None = None
+
+
+class ProjectTransferUpdate(APIModel):
+    to_project_id: UUID | None = None
+    quantity: Decimal | None = Field(default=None, gt=0)
+    note: str | None = None
+    transfer_date: datetime | None = None
+
+
+class ProjectTransferReceive(APIModel):
+    allocations: list[WarehouseAllocationIn] = Field(
+        min_length=1,
+        description="Warehouse allocations in destination project (sum must equal transfer quantity)",
+    )
+    received_at: datetime | None = None
+    note: str | None = None
+
+
+class IdNameOut(APIModel):
+    id: UUID
+    name: str
+
+
+class UnitBriefOut(APIModel):
+    id: int
+    code: str
+    name_fa: str | None = None
+
+
+class ProjectTransferOut(APIModel):
+    id: UUID
+    from_project: IdNameOut
+    from_warehouse: IdNameOut
+    to_project: IdNameOut
+    to_warehouse: IdNameOut | None = None
+    material: IdNameOut
+    brand: IdNameOut | None = None
+    unit: UnitBriefOut
+    quantity: Decimal
+    specs: dict
+    specs_hash: str
+    note: str | None = None
+    status: str
+    transfer_date: datetime
+    transferred_by: str | None = None
+    received_at: datetime | None = None
+    received_by: str | None = None
+    receipt_allocations: list[dict] | None = None
+    created_at: datetime
+    updated_at: datetime

backend/app/services/project_transfer_service.py

@@ -0,0 +1,417 @@
+import uuid
+from datetime import datetime, timezone
+from decimal import Decimal
+
+from sqlalchemy import delete, select
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.orm import selectinload
+
+from app.models.catalog import Brand, Material, Unit
+from app.models.inventory import InventoryItem, InventoryTransaction, Warehouse
+from app.models.procurement import Project
+from app.models.project_transfer import ProjectTransfer
+from app.schemas.project_transfers import (
+    ProjectTransferCreate,
+    ProjectTransferReceive,
+    ProjectTransferUpdate,
+)
+from app.services.inventory_service import (
+    _find_inventory,
+    _get_or_create_inventory,
+    canonical_specs_hash,
+)
+from app.utils.errors import ConflictError, ForbiddenError, NotFoundError, ValidationError
+
+
+async def get_project_transfer_or_404(db: AsyncSession, transfer_id: uuid.UUID) -> ProjectTransfer:
+    stmt = (
+        select(ProjectTransfer)
+        .where(ProjectTransfer.id == transfer_id)
+        .options(
+            selectinload(ProjectTransfer.from_project),
+            selectinload(ProjectTransfer.to_project),
+            selectinload(ProjectTransfer.from_warehouse),
+            selectinload(ProjectTransfer.to_warehouse),
+            selectinload(ProjectTransfer.material),
+            selectinload(ProjectTransfer.brand),
+            selectinload(ProjectTransfer.unit),
+        )
+    )
+    transfer = await db.scalar(stmt)
+    if transfer is None:
+        raise NotFoundError("رکورد انتقال بین کارگاهی یافت نشد", field="transfer_id")
+    return transfer
+
+
+async def create_project_transfer(
+    db: AsyncSession,
+    payload: ProjectTransferCreate,
+    caller_project_id: uuid.UUID | None,
+    caller_role: str,
+    transferred_by: str | None = None,
+) -> ProjectTransfer:
+    tx = db.begin_nested() if db.in_transaction() else db.begin()
+    async with tx:
+        # Validate source warehouse
+        from_wh = await db.get(Warehouse, payload.from_warehouse_id)
+        if from_wh is None:
+            raise NotFoundError("انبار مبدأ یافت نشد", field="from_warehouse_id")
+
+        # RBAC: PM can only send from their own project
+        if caller_role == "pm" and caller_project_id is not None:
+            if from_wh.project_id != caller_project_id:
+                raise ForbiddenError("شما فقط مجاز به انتقال از انبار پروژه خود هستید")
+
+        from_proj = await db.get(Project, from_wh.project_id)
+        to_proj = await db.get(Project, payload.to_project_id)
+        if to_proj is None:
+            raise NotFoundError("پروژه مقصد یافت نشد", field="to_project_id")
+
+        if from_wh.project_id == payload.to_project_id:
+            raise ValidationError(
+                "پروژه مقصد نمی‌تواند همان پروژه مبدأ باشد (از انتقال بین‌انباری استفاده کنید)",
+                field="to_project_id",
+            )
+
+        # Validate material, brand, unit
+        mat = await db.get(Material, payload.material_id)
+        if mat is None:
+            raise NotFoundError("ماده اولیه یافت نشد", field="material_id")
+
+        if payload.brand_id is not None:
+            brand = await db.get(Brand, payload.brand_id)
+            if brand is None:
+                raise NotFoundError("برند یافت نشد", field="brand_id")
+
+        unit = await db.get(Unit, payload.unit_id)
+        if unit is None:
+            raise NotFoundError("واحد یافت نشد", field="unit_id")
+
+        # Requirement 2: Canonical specs hash from inventory_service
+        specs = payload.specs or {}
+        specs_hash = canonical_specs_hash(specs)
+
+        # Check stock in source warehouse with SELECT ... FOR UPDATE
+        source_item = await _find_inventory(
+            db,
+            payload.from_warehouse_id,
+            payload.material_id,
+            payload.brand_id,
+            specs_hash,
+            for_update=True,
+        )
+        if source_item is None or source_item.quantity < payload.quantity:
+            avail = source_item.quantity if source_item else Decimal("0")
+            raise ValidationError(
+                f"موجودی کافی در انبار مبدأ وجود ندارد (موجودی فعلی: {avail})",
+                field="quantity",
+            )
+
+        # Deduct from source warehouse immediately
+        source_item.quantity -= payload.quantity
+
+        transfer_pair_id = uuid.uuid4()
+        now_dt = payload.transfer_date or datetime.now(timezone.utc)
+        note_text = f"انتقال بین کارگاهی به پروژه {to_proj.name}"
+        if payload.note:
+            note_text += f" — {payload.note}"
+
+        out_tx = InventoryTransaction(
+            inventory_id=source_item.id,
+            warehouse_id=from_wh.id,
+            transaction_type="TRANSFER_OUT",
+            quantity=-payload.quantity,
+            transfer_pair_id=transfer_pair_id,
+            performed_by=transferred_by,
+            transaction_date=now_dt,
+            note=note_text,
+        )
+        db.add(out_tx)
+        await db.flush()
+
+        transfer = ProjectTransfer(
+            from_project_id=from_wh.project_id,
+            from_warehouse_id=from_wh.id,
+            to_project_id=payload.to_project_id,
+            material_id=payload.material_id,
+            brand_id=payload.brand_id,
+            unit_id=payload.unit_id,
+            quantity=payload.quantity,
+            specs=specs,
+            specs_hash=specs_hash,
+            note=payload.note,
+            status="in_transit",
+            out_transaction_id=out_tx.id,
+            transfer_date=now_dt,
+            transferred_by=transferred_by,
+        )
+        db.add(transfer)
+        await db.flush()
+
+    return await get_project_transfer_or_404(db, transfer.id)
+
+
+async def update_project_transfer(
+    db: AsyncSession,
+    transfer_id: uuid.UUID,
+    payload: ProjectTransferUpdate,
+    caller_project_id: uuid.UUID | None,
+    caller_role: str,
+) -> ProjectTransfer:
+    tx = db.begin_nested() if db.in_transaction() else db.begin()
+    async with tx:
+        transfer = await db.scalar(
+            select(ProjectTransfer).where(ProjectTransfer.id == transfer_id).with_for_update()
+        )
+        if transfer is None:
+            raise NotFoundError("رکورد انتقال یافت نشد", field="transfer_id")
+
+        # RBAC (Requirement 4)
+        if caller_role == "pm" and caller_project_id is not None:
+            if transfer.from_project_id != caller_project_id:
+                raise ForbiddenError("شما فقط مجاز به ویرایش انتقال‌های پروژه مبدأ خود هستید")
+
+        # Status Lock Guard (Requirement 4)
+        if transfer.status == "received":
+            raise ConflictError(
+                "این انتقال در پروژه مقصد تحویل و ثبت انبار شده است و امکان ویرایش آن وجود ندارد"
+            )
+        if transfer.status != "in_transit":
+            raise ConflictError("فقط انتقال‌های در راه (in_transit) قابل ویرایش هستند")
+
+        if payload.to_project_id is not None and payload.to_project_id != transfer.to_project_id:
+            new_to_proj = await db.get(Project, payload.to_project_id)
+            if new_to_proj is None:
+                raise NotFoundError("پروژه مقصد یافت نشد", field="to_project_id")
+            if payload.to_project_id == transfer.from_project_id:
+                raise ValidationError("پروژه مقصد نمی‌تواند همان پروژه مبدأ باشد", field="to_project_id")
+            transfer.to_project_id = payload.to_project_id
+
+        # Quantity adjustment (reconciles source stock difference)
+        if payload.quantity is not None and payload.quantity != transfer.quantity:
+            delta = payload.quantity - transfer.quantity
+            source_item = await _find_inventory(
+                db,
+                transfer.from_warehouse_id,
+                transfer.material_id,
+                transfer.brand_id,
+                transfer.specs_hash,
+                for_update=True,
+            )
+            if delta > 0:
+                if source_item is None or source_item.quantity < delta:
+                    avail = source_item.quantity if source_item else Decimal("0")
+                    raise ValidationError(
+                        f"موجودی انبار مبدأ برای افزایش انتقال کافی نیست (موجود: {avail})",
+                        field="quantity",
+                    )
+                source_item.quantity -= delta
+            else:
+                if source_item is not None:
+                    source_item.quantity += abs(delta)
+
+            transfer.quantity = payload.quantity
+
+            # Update out_transaction
+            if transfer.out_transaction_id is not None:
+                out_tx = await db.get(InventoryTransaction, transfer.out_transaction_id)
+                if out_tx is not None:
+                    out_tx.quantity = -payload.quantity
+
+        if payload.note is not None:
+            transfer.note = payload.note
+
+        if payload.transfer_date is not None:
+            transfer.transfer_date = payload.transfer_date
+            if transfer.out_transaction_id is not None:
+                out_tx = await db.get(InventoryTransaction, transfer.out_transaction_id)
+                if out_tx is not None:
+                    out_tx.transaction_date = payload.transfer_date
+
+    return await get_project_transfer_or_404(db, transfer_id)
+
+
+async def delete_project_transfer(
+    db: AsyncSession,
+    transfer_id: uuid.UUID,
+    caller_project_id: uuid.UUID | None,
+    caller_role: str,
+) -> dict:
+    tx = db.begin_nested() if db.in_transaction() else db.begin()
+    async with tx:
+        transfer = await db.scalar(
+            select(ProjectTransfer).where(ProjectTransfer.id == transfer_id).with_for_update()
+        )
+        if transfer is None:
+            raise NotFoundError("رکورد انتقال یافت نشد", field="transfer_id")
+
+        # RBAC (Requirement 4)
+        if caller_role == "pm" and caller_project_id is not None:
+            if transfer.from_project_id != caller_project_id:
+                raise ForbiddenError("شما فقط مجاز به لغو انتقال‌های پروژه مبدأ خود هستید")
+
+        # Status Lock Guard (Requirement 4)
+        if transfer.status == "received":
+            raise ConflictError(
+                "این انتقال در پروژه مقصد تحویل و ثبت انبار شده است و امکان لغو یا حذف آن وجود ندارد"
+            )
+
+        # Reversibility (Requirement 4): Restore stock to source warehouse
+        source_item = await _find_inventory(
+            db,
+            transfer.from_warehouse_id,
+            transfer.material_id,
+            transfer.brand_id,
+            transfer.specs_hash,
+            for_update=True,
+        )
+        if source_item is not None:
+            source_item.quantity += transfer.quantity
+
+        # Delete out_transaction
+        if transfer.out_transaction_id is not None:
+            out_tx = await db.get(InventoryTransaction, transfer.out_transaction_id)
+            if out_tx is not None:
+                await db.delete(out_tx)
+
+        await db.delete(transfer)
+
+    return {"ok": True, "message": "انتقال با موفقیت لغو شد و موجودی به انبار مبدأ بازگردانده شد"}
+
+
+async def receive_project_transfer(
+    db: AsyncSession,
+    transfer_id: uuid.UUID,
+    payload: ProjectTransferReceive,
+    caller_project_id: uuid.UUID | None,
+    caller_role: str,
+    received_by: str | None = None,
+) -> ProjectTransfer:
+    tx = db.begin_nested() if db.in_transaction() else db.begin()
+    async with tx:
+        transfer = await db.scalar(
+            select(ProjectTransfer).where(ProjectTransfer.id == transfer_id).with_for_update()
+        )
+        if transfer is None:
+            raise NotFoundError("رکورد انتقال یافت نشد", field="transfer_id")
+
+        # RBAC (Requirement 4)
+        if caller_role == "pm" and caller_project_id is not None:
+            if transfer.to_project_id != caller_project_id:
+                raise ForbiddenError("شما فقط مجاز به تأیید تحویل در پروژه مقصد خود هستید")
+
+        # Lock Guard
+        if transfer.status == "received":
+            raise ConflictError("این انتقال قبلاً تحویل گرفته و ثبت انبار شده است")
+        if transfer.status != "in_transit":
+            raise ConflictError("این انتقال در وضعیت در راه نیست")
+
+        # Requirement 1: 100% full delivery logic (no partial receipt)
+        alloc_sum = sum(Decimal(str(a.quantity)) for a in payload.allocations)
+        if alloc_sum != transfer.quantity:
+            raise ValidationError(
+                f"تخصیص انبارها باید دقیقاً برابر کل مقدار ارسالی ({transfer.quantity}) باشد. مجموع واردشده: {alloc_sum}",
+                field="allocations",
+            )
+
+        # Validate destination warehouses belong to destination project
+        wh_ids = [a.warehouse_id for a in payload.allocations]
+        wh_rows = (await db.execute(select(Warehouse).where(Warehouse.id.in_(wh_ids)))).scalars().all()
+        wh_map = {w.id: w for w in wh_rows}
+
+        for a in payload.allocations:
+            wh = wh_map.get(a.warehouse_id)
+            if wh is None or wh.project_id != transfer.to_project_id:
+                raise ValidationError(
+                    "انبار انتخاب‌شده معتبر نیست یا متعلق به پروژه مقصد نمی‌باشد",
+                    field="allocations.warehouse_id",
+                )
+
+        from_proj = await db.get(Project, transfer.from_project_id)
+        from_proj_name = from_proj.name if from_proj else "پروژه مبدأ"
+
+        out_tx = (
+            await db.get(InventoryTransaction, transfer.out_transaction_id)
+            if transfer.out_transaction_id
+            else None
+        )
+        pair_id = out_tx.transfer_pair_id if (out_tx and out_tx.transfer_pair_id) else uuid.uuid4()
+        now_dt = payload.received_at or datetime.now(timezone.utc)
+
+        # Add stock to destination warehouse(s) via TRANSFER_IN
+        for a in payload.allocations:
+            dest_item = await _get_or_create_inventory(
+                db,
+                a.warehouse_id,
+                transfer.material_id,
+                transfer.brand_id,
+                transfer.unit_id,
+                transfer.specs,
+                transfer.specs_hash,
+            )
+            dest_item.quantity += a.quantity
+
+            note_text = f"رسید انتقال بین کارگاهی از {from_proj_name}"
+            if payload.note:
+                note_text += f" — {payload.note}"
+
+            in_tx = InventoryTransaction(
+                inventory_id=dest_item.id,
+                warehouse_id=a.warehouse_id,
+                transaction_type="TRANSFER_IN",
+                quantity=a.quantity,
+                transfer_pair_id=pair_id,
+                performed_by=received_by,
+                transaction_date=now_dt,
+                note=note_text,
+            )
+            db.add(in_tx)
+
+        transfer.status = "received"
+        transfer.received_at = now_dt
+        transfer.received_by = received_by
+        transfer.to_warehouse_id = payload.allocations[0].warehouse_id
+        transfer.receipt_allocations = [
+            {"warehouse_id": str(a.warehouse_id), "quantity": str(a.quantity)}
+            for a in payload.allocations
+        ]
+
+    return await get_project_transfer_or_404(db, transfer_id)
+
+
+async def list_project_transfers(
+    db: AsyncSession,
+    project_id: uuid.UUID | None = None,
+    direction: str = "all",  # 'outgoing' | 'incoming' | 'all'
+    status: str | None = None,
+) -> list[ProjectTransfer]:
+    stmt = (
+        select(ProjectTransfer)
+        .options(
+            selectinload(ProjectTransfer.from_project),
+            selectinload(ProjectTransfer.to_project),
+            selectinload(ProjectTransfer.from_warehouse),
+            selectinload(ProjectTransfer.to_warehouse),
+            selectinload(ProjectTransfer.material),
+            selectinload(ProjectTransfer.brand),
+            selectinload(ProjectTransfer.unit),
+        )
+    )
+
+    if project_id is not None:
+        if direction == "outgoing":
+            stmt = stmt.where(ProjectTransfer.from_project_id == project_id)
+        elif direction == "incoming":
+            stmt = stmt.where(ProjectTransfer.to_project_id == project_id)
+        else:
+            stmt = stmt.where(
+                (ProjectTransfer.from_project_id == project_id)
+                | (ProjectTransfer.to_project_id == project_id)
+            )
+
+    if status:
+        stmt = stmt.where(ProjectTransfer.status == status)
+
+    stmt = stmt.order_by(ProjectTransfer.transfer_date.desc())
+    res = await db.execute(stmt)
+    return list(res.scalars().all())

sql/migration.sql

@@ -551,6 +551,35 @@ CREATE TABLE standard_conversions (
     CONSTRAINT uq_standard_conversion UNIQUE (from_unit, to_unit)
 );
 
+-- Inter-project material transfers
+CREATE TABLE project_transfers (
+    id                  UUID            NOT NULL DEFAULT gen_random_uuid() PRIMARY KEY,
+    from_project_id     UUID            NOT NULL REFERENCES projects(id) ON DELETE RESTRICT,
+    from_warehouse_id   UUID            NOT NULL REFERENCES warehouses(id) ON DELETE RESTRICT,
+    to_project_id       UUID            NOT NULL REFERENCES projects(id) ON DELETE RESTRICT,
+    material_id         UUID            NOT NULL REFERENCES materials(id) ON DELETE RESTRICT,
+    brand_id            UUID            REFERENCES brands(id) ON DELETE SET NULL,
+    unit_id             INTEGER         NOT NULL REFERENCES units(id) ON DELETE RESTRICT,
+    quantity            NUMERIC(18,6)   NOT NULL CHECK (quantity > 0),
+    specs               JSONB           NOT NULL DEFAULT '{}'::jsonb,
+    specs_hash          TEXT            NOT NULL DEFAULT 'default',
+    note                TEXT,
+    status              TEXT            NOT NULL DEFAULT 'in_transit' CHECK (status IN ('in_transit', 'received', 'cancelled')),
+    out_transaction_id  UUID            REFERENCES inventory_transactions(id) ON DELETE SET NULL,
+    transfer_date       TIMESTAMPTZ     NOT NULL DEFAULT now(),
+    transferred_by      TEXT,
+    received_at         TIMESTAMPTZ,
+    received_by         TEXT,
+    to_warehouse_id     UUID            REFERENCES warehouses(id) ON DELETE SET NULL,
+    receipt_allocations JSONB,
+    created_at          TIMESTAMPTZ     NOT NULL DEFAULT now(),
+    updated_at          TIMESTAMPTZ     NOT NULL DEFAULT now()
+);
+
+CREATE INDEX idx_project_transfers_from ON project_transfers(from_project_id, status);
+CREATE INDEX idx_project_transfers_to ON project_transfers(to_project_id, status);
+
+
 -- ============================================================================
 -- UPDATED_AT TRIGGER (auto-set updated_at on row modification)
 -- ============================================================================
@@ -579,6 +608,10 @@ CREATE TRIGGER trg_quote_attachments_updated_at
     BEFORE UPDATE ON quote_attachments
     FOR EACH ROW EXECUTE FUNCTION set_updated_at();
 
+CREATE TRIGGER trg_project_transfers_updated_at
+    BEFORE UPDATE ON project_transfers
+    FOR EACH ROW EXECUTE FUNCTION set_updated_at();
+
 -- ============================================================================
 -- AUTOMATION TRIGGERS (v2.5 + v2.6)
 -- ============================================================================