feat(psa): add PSA post preview, post-to-ticket, and post history endpoints

Three new session-scoped endpoints for posting session documentation
to linked PSA tickets:

- GET /sessions/{id}/psa-post/preview — generates PSA-formatted content,
  fetches ticket details and available statuses, counts previous posts
- POST /sessions/{id}/psa-post — posts note to CW ticket with optional
  status update, logs all actions in psa_post_log audit trail
- GET /sessions/{id}/psa-posts — returns post history for the session

New schemas: PsaPostRequest, PsaPostResponse, PsaPreviewResponse,
PsaPostLogResponse.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

backend/app/api/endpoints/sessions.py

@@ -27,6 +27,7 @@ from app.schemas.session import (
     TicketLinkResponse,
     PSATicketResponse,
 )
+from app.schemas.psa_connection import PsaPostRequest
 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
@@ -839,3 +840,268 @@ async def link_ticket(
             priority_name=ticket.priority_name,
         ),
     )
+
+
+# ── PSA Post to Ticket ────────────────────────────────────────────
+
+
+@router.get("/{session_id}/psa-post/preview")
+async def psa_post_preview(
+    session_id: UUID,
+    current_user: Annotated[User, Depends(require_engineer_or_admin)],
+    db: Annotated[AsyncSession, Depends(get_db)],
+):
+    """Preview the content that will be posted to the linked PSA ticket.
+
+    Generates session documentation in PSA format, fetches current ticket
+    details and available statuses, and counts previous posts.
+    """
+    from app.models.psa_post_log import PsaPostLog
+    from app.services.psa.registry import get_provider_for_account
+    from app.services.psa.exceptions import PSAError
+    from app.schemas.psa_connection import (
+        PsaPreviewResponse,
+        PSATicketSearchResult,
+        PSATicketStatusItem,
+    )
+    from sqlalchemy import func as sa_func
+
+    # Load 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=404, detail="Session not found")
+
+    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=403, detail="You don't have access to this session")
+
+    if not session.psa_ticket_id:
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+            detail="Session has no linked PSA ticket. Link a ticket first.",
+        )
+
+    if not current_user.account_id:
+        raise HTTPException(status_code=400, detail="No account associated with your user")
+
+    # Generate PSA export content
+    export_options = SessionExport(
+        format="psa",
+        include_timestamps=True,
+        include_tree_info=True,
+        include_outcome_notes=True,
+        include_next_steps=True,
+        include_summary=True,
+    )
+    content = generate_psa_export(session, export_options)
+
+    # Resolve session variables in content
+    session_vars = getattr(session, "session_variables", None) or {}
+    if session_vars:
+        from app.services.variable_service import resolve_variables
+        content = resolve_variables(content, session_vars)
+
+    # Fetch ticket details and statuses from CW
+    try:
+        provider = await get_provider_for_account(current_user.account_id, db)
+        ticket = await provider.get_ticket(session.psa_ticket_id)
+        available_statuses: list[PSATicketStatusItem] = []
+        if ticket.board_id:
+            statuses = await provider.get_ticket_statuses(ticket.board_id)
+            available_statuses = [
+                PSATicketStatusItem(id=s.id, name=s.name, is_closed=s.is_closed)
+                for s in statuses
+            ]
+    except PSAError as e:
+        raise HTTPException(status_code=502, detail=f"PSA error: {e}")
+
+    # Count previous posts
+    count_result = await db.execute(
+        select(sa_func.count(PsaPostLog.id)).where(
+            PsaPostLog.session_id == session_id
+        )
+    )
+    previous_posts = count_result.scalar_one()
+
+    return PsaPreviewResponse(
+        content=content,
+        ticket=PSATicketSearchResult(
+            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,
+            closed=ticket.closed,
+        ),
+        available_statuses=available_statuses,
+        character_count=len(content),
+        previous_posts=previous_posts,
+    )
+
+
+@router.post("/{session_id}/psa-post")
+async def psa_post_to_ticket(
+    session_id: UUID,
+    data: PsaPostRequest,
+    current_user: Annotated[User, Depends(require_engineer_or_admin)],
+    db: Annotated[AsyncSession, Depends(get_db)],
+):
+    """Post session documentation as a note to the linked PSA ticket.
+
+    Optionally updates the ticket status if update_status_id is provided.
+    All actions are logged in psa_post_log for audit trail.
+    """
+    from app.models.psa_connection import PsaConnection
+    from app.models.psa_post_log import PsaPostLog
+    from app.services.psa.registry import get_provider_for_account
+    from app.services.psa.exceptions import PSAError
+    from app.schemas.psa_connection import PsaPostResponse
+
+    # Load 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=404, detail="Session not found")
+
+    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=403, detail="You don't have access to this session")
+
+    if not session.psa_ticket_id:
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+            detail="Session has no linked PSA ticket. Link a ticket first.",
+        )
+
+    if not current_user.account_id:
+        raise HTTPException(status_code=400, detail="No account associated with your user")
+
+    # Get PSA connection ID for audit
+    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()
+
+    # Post note
+    try:
+        provider = await get_provider_for_account(current_user.account_id, db)
+        note_result = await provider.post_note(
+            ticket_id=session.psa_ticket_id,
+            text=data.content,
+            note_type=data.note_type,
+        )
+        note_status = "success"
+        external_note_id = note_result.id
+        error_message = None
+    except PSAError as e:
+        note_status = "failed"
+        external_note_id = None
+        error_message = str(e)
+
+    # Optionally update ticket status
+    status_changed_from = None
+    status_changed_to = None
+    if data.update_status_id and note_status == "success":
+        try:
+            # Get current status before update
+            current_ticket = await provider.get_ticket(session.psa_ticket_id)
+            status_changed_from = current_ticket.status_name
+
+            if current_ticket.status_id != data.update_status_id:
+                updated_ticket = await provider.update_ticket_status(
+                    session.psa_ticket_id, data.update_status_id
+                )
+                status_changed_to = updated_ticket.status_name
+        except PSAError as e:
+            # Log the status update failure but don't fail the whole request
+            # since the note was already posted successfully
+            if error_message:
+                error_message += f"; Status update failed: {e}"
+            else:
+                error_message = f"Note posted successfully but status update failed: {e}"
+
+    # Log to audit trail
+    log_entry = PsaPostLog(
+        session_id=session.id,
+        psa_connection_id=psa_connection.id if psa_connection else None,
+        ticket_id=session.psa_ticket_id,
+        note_type=data.note_type,
+        content_posted=data.content,
+        external_note_id=external_note_id,
+        status=note_status,
+        error_message=error_message,
+        status_changed_from=status_changed_from,
+        status_changed_to=status_changed_to,
+        posted_by=current_user.id,
+    )
+    db.add(log_entry)
+    await db.commit()
+    await db.refresh(log_entry)
+
+    if note_status == "failed":
+        raise HTTPException(
+            status_code=502,
+            detail=error_message or "Failed to post note to PSA",
+        )
+
+    return PsaPostResponse(
+        id=str(log_entry.id),
+        session_id=str(session.id),
+        ticket_id=session.psa_ticket_id,
+        note_type=data.note_type,
+        status=note_status,
+        external_note_id=external_note_id,
+        error_message=error_message,
+        status_changed_from=status_changed_from,
+        status_changed_to=status_changed_to,
+        posted_at=log_entry.posted_at.isoformat(),
+    )
+
+
+@router.get("/{session_id}/psa-posts")
+async def list_psa_posts(
+    session_id: UUID,
+    current_user: Annotated[User, Depends(require_engineer_or_admin)],
+    db: Annotated[AsyncSession, Depends(get_db)],
+):
+    """List all PSA post history for a session, ordered by most recent first."""
+    from app.models.psa_post_log import PsaPostLog
+    from app.schemas.psa_connection import PsaPostLogResponse
+
+    # Verify session access
+    result = await db.execute(select(Session).where(Session.id == session_id))
+    session = result.scalar_one_or_none()
+    if not session:
+        raise HTTPException(status_code=404, detail="Session not found")
+
+    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=403, detail="You don't have access to this session")
+
+    # Query post log
+    log_result = await db.execute(
+        select(PsaPostLog)
+        .where(PsaPostLog.session_id == session_id)
+        .order_by(PsaPostLog.posted_at.desc())
+    )
+    logs = log_result.scalars().all()
+
+    return [
+        PsaPostLogResponse(
+            id=str(log.id),
+            ticket_id=log.ticket_id,
+            note_type=log.note_type,
+            status=log.status,
+            error_message=log.error_message,
+            status_changed_from=log.status_changed_from,
+            status_changed_to=log.status_changed_to,
+            posted_at=log.posted_at.isoformat(),
+            content_preview=log.content_posted[:200],
+        )
+        for log in logs
+    ]

backend/app/schemas/__init__.py

@@ -18,6 +18,7 @@ from .script_template import (
 from .psa_connection import (
     PsaConnectionCreate, PsaConnectionUpdate, PsaConnectionResponse, PsaConnectionTestResponse,
     PSATicketSearchResult, PSATicketStatusItem,
+    PsaPostRequest, PsaPostResponse, PsaPreviewResponse, PsaPostLogResponse,
 )
 
 __all__ = [
@@ -46,4 +47,5 @@ __all__ = [
     # PSA Connection
     "PsaConnectionCreate", "PsaConnectionUpdate", "PsaConnectionResponse", "PsaConnectionTestResponse",
     "PSATicketSearchResult", "PSATicketStatusItem",
+    "PsaPostRequest", "PsaPostResponse", "PsaPreviewResponse", "PsaPostLogResponse",
 ]

backend/app/schemas/psa_connection.py

@@ -64,3 +64,45 @@ class PSATicketStatusItem(BaseModel):
     id: int
     name: str
     is_closed: bool = False
+
+
+# ── PSA post (note posting) schemas ──────────────────────────────
+
+
+class PsaPostRequest(BaseModel):
+    note_type: str = Field(pattern="^(internal_analysis|resolution|description)$")
+    content: str = Field(min_length=1)
+    update_status_id: int | None = None
+
+
+class PsaPostResponse(BaseModel):
+    id: str
+    session_id: str
+    ticket_id: str
+    note_type: str
+    status: str
+    external_note_id: str | None = None
+    error_message: str | None = None
+    status_changed_from: str | None = None
+    status_changed_to: str | None = None
+    posted_at: str
+
+
+class PsaPreviewResponse(BaseModel):
+    content: str
+    ticket: PSATicketSearchResult
+    available_statuses: list[PSATicketStatusItem]
+    character_count: int
+    previous_posts: int
+
+
+class PsaPostLogResponse(BaseModel):
+    id: str
+    ticket_id: str
+    note_type: str
+    status: str
+    error_message: str | None = None
+    status_changed_from: str | None = None
+    status_changed_to: str | None = None
+    posted_at: str
+    content_preview: str  # first 200 chars