feat(procurement): support multi-project line item split allocation and merge on reopen

- Add ProjectAllocationIn and allocations field to ItemDecisionIn schema
- Implement line item split allocation across projects in accept_quote
- Replicate all ItemTermsheet options and LineItemSpec records to split rows
- Ensure QuoteProjects and LineItemProjects relations are updated
- Add validation against duplicate project selection and decimal sum mismatches
- Implement _merge_split_line_items_on_reopen to consolidate rows when reverting status to offered

backend/app/schemas/procurement.py

@@ -135,6 +135,11 @@ class AcceptTermsheetIn(APIModel):
     schedule_steps: list[ScheduleStepIn] = Field(min_length=1)
 
 
+class ProjectAllocationIn(APIModel):
+    project_id: UUID
+    quantity: Decimal = Field(gt=0, description="Allocated quantity for this project")
+
+
 class ItemDecisionIn(APIModel):
     line_item_id: UUID
     status: Literal["chosen", "rejected"]
@@ -142,6 +147,7 @@ class ItemDecisionIn(APIModel):
     adjusted_unit_price: Decimal | None = Field(default=None, gt=0)
     adjusted_notes: str | None = None
     project_id: UUID | None = Field(default=None, description="The ONE purchased project (v2.6 — required when the offer spans several projects)")
+    allocations: list[ProjectAllocationIn] = Field(default_factory=list, description="Multi-project split allocations for this line item")
 
 
 class ItemProjectsIn(APIModel):

backend/app/services/quote_service.py

@@ -7,6 +7,7 @@ same transaction.
 import asyncio
 import logging
 from datetime import date, datetime, timezone
+from decimal import Decimal
 from uuid import UUID
 
 from sqlalchemy import delete, func, select, update
@@ -18,6 +19,7 @@ from app.models.procurement import (
     ItemTermsheet,
     LineItem,
     LineItemProjects,
+    LineItemSpec,
     Project,
     Quote,
     QuoteEvent,
@@ -418,6 +420,78 @@ async def replace_quote_children(
 
 
 
+async def _ensure_quote_project(db: AsyncSession, quote_id: UUID, project_id: UUID) -> None:
+    """Ensure (quote_id, project_id) exists in quote_projects junction."""
+    exists = await db.scalar(
+        select(QuoteProjects).where(
+            QuoteProjects.quote_id == quote_id,
+            QuoteProjects.project_id == project_id,
+        )
+    )
+    if not exists:
+        db.add(QuoteProjects(quote_id=quote_id, project_id=project_id))
+
+
+async def _merge_split_line_items_on_reopen(db: AsyncSession, quote_id: UUID) -> None:
+    """Consolidate split line items when reverting a quote back to offered.
+
+    Line items sharing the exact same material, brand_material, unit, notes,
+    and identical specs are merged into a single consolidated row whose quantity
+    is the sum of all split rows. Secondary split rows are safely deleted.
+    """
+    line_items = (
+        await db.execute(
+            select(LineItem).where(LineItem.quote_id == quote_id).order_by(LineItem.created_at.asc(), LineItem.id.asc())
+        )
+    ).scalars().all()
+    if len(line_items) <= 1:
+        return
+
+    # Load all specs for these line items to compare canonical signatures
+    specs_rows = (
+        await db.execute(
+            select(LineItemSpec).where(LineItemSpec.line_item_id.in_([li.id for li in line_items]))
+        )
+    ).scalars().all()
+    specs_by_li: dict[UUID, list[tuple]] = {}
+    for sp in specs_rows:
+        sig = (
+            str(sp.attribute_definition_id),
+            str(sp.value_number) if sp.value_number is not None else "",
+            (sp.value_string or "").strip(),
+            str(sp.value_boolean) if sp.value_boolean is not None else "",
+            str(sp.value_date) if sp.value_date is not None else "",
+        )
+        specs_by_li.setdefault(sp.line_item_id, []).append(sig)
+
+    # Group line items by identity
+    groups: dict[tuple, list[LineItem]] = {}
+    for li in line_items:
+        specs_sig = tuple(sorted(specs_by_li.get(li.id, [])))
+        key = (
+            li.material_id,
+            li.brand_material_id,
+            li.unit_id,
+            (li.notes or "").strip(),
+            specs_sig,
+        )
+        groups.setdefault(key, []).append(li)
+
+    for key, items in groups.items():
+        if len(items) > 1:
+            primary = items[0]
+            total_qty = sum(Decimal(str(it.quantity)) for it in items)
+            primary.quantity = total_qty
+
+            secondary_ids = [it.id for it in items[1:]]
+            await db.execute(delete(ItemDecision).where(ItemDecision.line_item_id.in_(secondary_ids)))
+            await db.execute(delete(ItemTermsheet).where(ItemTermsheet.line_item_id.in_(secondary_ids)))
+            await db.execute(delete(LineItemProjects).where(LineItemProjects.line_item_id.in_(secondary_ids)))
+            await db.execute(delete(LineItemSpec).where(LineItemSpec.line_item_id.in_(secondary_ids)))
+            await db.execute(delete(LineItem).where(LineItem.id.in_(secondary_ids)))
+            await db.flush()
+
+
 async def _apply_status_transition(db: AsyncSession, quote: Quote, new_status: str) -> None:
     """Backward transitions only; reverts bookkeeping when going back to offered."""
     current = quote.status
@@ -454,6 +528,8 @@ async def _apply_status_transition(db: AsyncSession, quote: Quote, new_status: s
             )
             .execution_options(synchronize_session=False)
         )
+        # Requirement 2: Merge split line items back into single rows
+        await _merge_split_line_items_on_reopen(db, quote.id)
         # LineItemProjects (candidate projects) are preserved untouched!
     elif new_status == "accepted":  # arrived -> accepted
         quote.arrived_date = None
@@ -557,6 +633,125 @@ async def accept_quote(db: AsyncSession, quote_id: UUID, payload: AcceptQuoteIn,
                         "برای قلم انتخاب‌شده adjusted_unit_price الزامی است", field="item_decisions"
                     )
                 termsheet_id = ts_ids_by_index[index]
+
+                li_row = await db.scalar(
+                    select(LineItem).where(LineItem.id == decision_in.line_item_id).with_for_update()
+                )
+                if li_row is None:
+                    raise NotFoundError("خط آیتم یافت نشد", field="item_decisions.line_item_id")
+
+                allocations = decision_in.allocations or []
+                if allocations:
+                    # Requirement 3: Validate no duplicate projects within allocations of this item
+                    alloc_pids = [a.project_id for a in allocations]
+                    if len(alloc_pids) != len(set(alloc_pids)):
+                        raise ValidationError(
+                            "پروژه تکراری برای یک قلم کالا در ردیف‌های تخصیص انتخاب شده است",
+                            field="item_decisions.allocations",
+                        )
+                    # Requirement 5: Decimal precision check for sum of allocation quantities
+                    alloc_sum = sum(Decimal(str(a.quantity)) for a in allocations)
+                    target_qty = Decimal(str(li_row.quantity))
+                    if alloc_sum != target_qty:
+                        raise ValidationError(
+                            f"مجموع مقادیر تخصیص‌یافته ({alloc_sum}) با مقدار کل قلم ({target_qty}) برابر نیست",
+                            field="item_decisions.allocations",
+                        )
+                    # Validate all projects exist
+                    for a in allocations:
+                        if await db.get(Project, a.project_id) is None:
+                            raise NotFoundError("پروژه یافت نشد", field="item_decisions.allocations.project_id")
+
+                    if len(allocations) >= 2:
+                        # 1. Update the original line item to the first allocation quantity
+                        first_alloc = allocations[0]
+                        li_row.quantity = first_alloc.quantity
+                        decision_in.project_id = first_alloc.project_id
+
+                        # Query all existing item_termsheets and line_item_specs for copying
+                        orig_termsheets = (
+                            await db.execute(
+                                select(ItemTermsheet).where(ItemTermsheet.line_item_id == li_row.id)
+                            )
+                        ).scalars().all()
+                        orig_specs = (
+                            await db.execute(
+                                select(LineItemSpec).where(LineItemSpec.line_item_id == li_row.id)
+                            )
+                        ).scalars().all()
+
+                        # 2. For each subsequent allocation, create a new LineItem and copy specs + ALL termsheets
+                        for split_alloc in allocations[1:]:
+                            new_li = LineItem(
+                                quote_id=quote.id,
+                                material_id=li_row.material_id,
+                                brand_material_id=li_row.brand_material_id,
+                                unit_id=li_row.unit_id,
+                                quantity=split_alloc.quantity,
+                                notes=li_row.notes,
+                            )
+                            db.add(new_li)
+                            await db.flush()
+
+                            # Copy LineItemSpec
+                            for sp in orig_specs:
+                                db.add(
+                                    LineItemSpec(
+                                        line_item_id=new_li.id,
+                                        attribute_definition_id=sp.attribute_definition_id,
+                                        value_number=sp.value_number,
+                                        value_string=sp.value_string,
+                                        value_boolean=sp.value_boolean,
+                                        value_date=sp.value_date,
+                                        unit_id_at_time=sp.unit_id_at_time,
+                                    )
+                                )
+
+                            # Requirement 1: Copy ALL payment options (ItemTermsheet)
+                            chosen_new_it: ItemTermsheet | None = None
+                            for it_orig in orig_termsheets:
+                                price = (
+                                    decision_in.adjusted_unit_price
+                                    if (it_orig.termsheet_id == termsheet_id and decision_in.adjusted_unit_price is not None)
+                                    else it_orig.unit_price
+                                )
+                                copied_it = ItemTermsheet(
+                                    line_item_id=new_li.id,
+                                    termsheet_id=it_orig.termsheet_id,
+                                    unit_price=price,
+                                    min_order_qty=it_orig.min_order_qty,
+                                )
+                                db.add(copied_it)
+                                if it_orig.termsheet_id == termsheet_id:
+                                    chosen_new_it = copied_it
+
+                            if chosen_new_it is None:
+                                chosen_new_it = ItemTermsheet(
+                                    line_item_id=new_li.id,
+                                    termsheet_id=termsheet_id,
+                                    unit_price=decision_in.adjusted_unit_price,
+                                )
+                                db.add(chosen_new_it)
+                            await db.flush()
+
+                            # Create ItemDecision for the split line item
+                            new_decision = ItemDecision(
+                                line_item_id=new_li.id,
+                                status="chosen",
+                                chosen_item_termsheet_id=chosen_new_it.id,
+                                adjusted_unit_price=decision_in.adjusted_unit_price,
+                                adjusted_notes=decision_in.adjusted_notes,
+                                project_id=split_alloc.project_id,
+                                decided_at=utcnow(),
+                            )
+                            db.add(new_decision)
+
+                            # Requirement 4: sync QuoteProjects & LineItemProjects
+                            await _ensure_quote_project(db, quote.id, split_alloc.project_id)
+                            db.add(LineItemProjects(line_item_id=new_li.id, project_id=split_alloc.project_id))
+                    elif len(allocations) == 1:
+                        decision_in.project_id = allocations[0].project_id
+
                 it = await db.scalar(
                     select(ItemTermsheet).where(
                         ItemTermsheet.line_item_id == decision_in.line_item_id,
@@ -610,6 +805,8 @@ async def accept_quote(db: AsyncSession, quote_id: UUID, payload: AcceptQuoteIn,
                 if decision.project_id is not None:
                     if await db.get(Project, decision.project_id) is None:
                         raise NotFoundError("پروژه یافت نشد", field="item_decisions.project_id")
+                    # Requirement 4: sync QuoteProjects
+                    await _ensure_quote_project(db, quote.id, decision.project_id)
                 chosen_ts_ids.add(termsheet_id)
             else:  # rejected
                 if decision is None: