diff --git a/backend/app/api/endpoints/sessions.py b/backend/app/api/endpoints/sessions.py
index 951489a7..c560dba2 100644
--- a/backend/app/api/endpoints/sessions.py
+++ b/backend/app/api/endpoints/sessions.py
@@ -23,8 +23,11 @@ from app.schemas.session import (
     SessionComplete,
     SessionVariablesUpdate,
     PrepareSessionRequest,
+    TicketLinkRequest,
+    TicketLinkResponse,
+    PSATicketResponse,
 )
-from app.api.deps import get_current_active_user
+from app.api.deps import get_current_active_user, require_engineer_or_admin
 from app.core.permissions import can_access_tree
 from app.services.export_service import generate_markdown_export, generate_text_export, generate_html_export, generate_psa_export
 
@@ -738,3 +741,101 @@ async def batch_launch_sessions(
             for s in created_sessions
         ],
     )
+
+
+# ── PSA Ticket Link ─────────────────────────────────────────────────
+
+
+@router.patch("/{session_id}/ticket-link", response_model=TicketLinkResponse)
+async def link_ticket(
+    session_id: UUID,
+    data: TicketLinkRequest,
+    current_user: Annotated[User, Depends(require_engineer_or_admin)],
+    db: Annotated[AsyncSession, Depends(get_db)],
+):
+    """Link or unlink a PSA ticket to/from a session."""
+    from app.models.psa_connection import PsaConnection
+    from app.services.psa.registry import get_provider_for_account
+    from app.services.psa.exceptions import PSANotFoundError, PSAError
+
+    # Look up session
+    result = await db.execute(select(Session).where(Session.id == session_id))
+    session = result.scalar_one_or_none()
+    if not session:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="Session not found",
+        )
+
+    # Verify ownership or admin
+    if session.user_id != current_user.id and session.assigned_to_id != current_user.id:
+        if not current_user.is_super_admin:
+            raise HTTPException(
+                status_code=status.HTTP_403_FORBIDDEN,
+                detail="You don't have access to this session",
+            )
+
+    # Unlink
+    if data.psa_ticket_id is None:
+        session.psa_ticket_id = None
+        session.psa_connection_id = None
+        await db.commit()
+        return TicketLinkResponse(
+            session_id=str(session.id),
+            psa_ticket_id=None,
+            ticket=None,
+        )
+
+    # Link — validate ticket exists in CW
+    if not current_user.account_id:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="No account associated with your user",
+        )
+
+    try:
+        provider = await get_provider_for_account(current_user.account_id, db)
+    except PSAError as exc:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail=str(exc),
+        )
+
+    # Fetch the connection to store its ID
+    conn_result = await db.execute(
+        select(PsaConnection).where(
+            PsaConnection.account_id == current_user.account_id,
+            PsaConnection.is_active.is_(True),
+        )
+    )
+    psa_connection = conn_result.scalar_one_or_none()
+
+    try:
+        ticket = await provider.get_ticket(data.psa_ticket_id)
+    except PSANotFoundError:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="Ticket not found in ConnectWise",
+        )
+    except PSAError as exc:
+        raise HTTPException(
+            status_code=status.HTTP_502_BAD_GATEWAY,
+            detail=f"PSA error: {exc}",
+        )
+
+    session.psa_ticket_id = ticket.id
+    session.psa_connection_id = psa_connection.id if psa_connection else None
+    await db.commit()
+
+    return TicketLinkResponse(
+        session_id=str(session.id),
+        psa_ticket_id=ticket.id,
+        ticket=PSATicketResponse(
+            id=ticket.id,
+            summary=ticket.summary,
+            company_name=ticket.company_name,
+            board_name=ticket.board_name,
+            status_name=ticket.status_name,
+            priority_name=ticket.priority_name,
+        ),
+    )