feat(l1): internal_ticket_service with CRUD + status transitions

create_ticket, update_status (sets resolved_at on resolve), get_ticket,
list_tickets_for_account (status filter, account-scoped), promote_to_psa.
Used by L1 intake when account has no PSA integration configured.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

backend/app/services/internal_ticket_service.py

@@ -0,0 +1,90 @@
+"""CRUD + status transitions for internal_tickets (the no-PSA fallback ticket model)."""
+from datetime import datetime, timezone
+from typing import Optional
+from uuid import UUID
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.models.internal_ticket import InternalTicket
+
+
+async def create_ticket(
+    db: AsyncSession,
+    *,
+    account_id: UUID,
+    created_by_user_id: UUID,
+    problem_statement: str,
+    customer_name: Optional[str] = None,
+    customer_contact: Optional[str] = None,
+) -> InternalTicket:
+    """Create a new internal ticket in 'open' status."""
+    ticket = InternalTicket(
+        account_id=account_id,
+        created_by_user_id=created_by_user_id,
+        problem_statement=problem_statement,
+        customer_name=customer_name,
+        customer_contact=customer_contact,
+    )
+    db.add(ticket)
+    await db.flush()
+    return ticket
+
+
+async def update_status(
+    db: AsyncSession,
+    *,
+    ticket_id: UUID,
+    status: str,
+    resolution_notes: Optional[str] = None,
+    assigned_user_id: Optional[UUID] = None,
+) -> InternalTicket:
+    """Transition a ticket to a new status. Sets resolved_at when status='resolved'."""
+    ticket = await db.get(InternalTicket, ticket_id)
+    if not ticket:
+        raise ValueError(f"InternalTicket {ticket_id} not found")
+    ticket.status = status
+    if status == 'resolved':
+        ticket.resolved_at = datetime.now(timezone.utc)
+        if resolution_notes is not None:
+            ticket.resolution_notes = resolution_notes
+    if assigned_user_id is not None:
+        ticket.assigned_user_id = assigned_user_id
+    await db.flush()
+    return ticket
+
+
+async def get_ticket(db: AsyncSession, ticket_id: UUID) -> Optional[InternalTicket]:
+    """Fetch a ticket by ID. Returns None if not found."""
+    return await db.get(InternalTicket, ticket_id)
+
+
+async def list_tickets_for_account(
+    db: AsyncSession,
+    *,
+    account_id: UUID,
+    status: Optional[str] = None,
+    limit: int = 100,
+) -> list[InternalTicket]:
+    """List tickets for an account, optionally filtered by status, newest first."""
+    stmt = select(InternalTicket).where(InternalTicket.account_id == account_id)
+    if status:
+        stmt = stmt.where(InternalTicket.status == status)
+    stmt = stmt.order_by(InternalTicket.created_at.desc()).limit(limit)
+    result = await db.execute(stmt)
+    return list(result.scalars())
+
+
+async def promote_to_psa(
+    db: AsyncSession,
+    *,
+    ticket_id: UUID,
+    psa_ticket_id: str,
+) -> InternalTicket:
+    """Mark an internal ticket as promoted to PSA."""
+    ticket = await db.get(InternalTicket, ticket_id)
+    if not ticket:
+        raise ValueError(f"InternalTicket {ticket_id} not found")
+    ticket.psa_promoted_ticket_id = psa_ticket_id
+    await db.flush()
+    return ticket

backend/tests/test_internal_ticket_service.py

@@ -0,0 +1,182 @@
+"""Unit + integration tests for internal_ticket_service."""
+import uuid
+import pytest
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.models.account import Account
+from app.models.user import User
+from app.services.internal_ticket_service import (
+    create_ticket, update_status, get_ticket,
+    list_tickets_for_account, promote_to_psa,
+)
+
+
+# ---------------------------------------------------------------------------
+# Test helpers
+# ---------------------------------------------------------------------------
+
+async def _make_account(db: AsyncSession) -> Account:
+    s = str(uuid.uuid4())[:8]
+    account = Account(
+        id=uuid.uuid4(),
+        name=f"Test Account {s}",
+        display_code=s[:8],
+    )
+    db.add(account)
+    await db.flush()
+    return account
+
+
+async def _make_user(
+    db: AsyncSession,
+    *,
+    account_id: uuid.UUID,
+    role: str = "l1_tech",
+) -> User:
+    s = str(uuid.uuid4())[:8]
+    user = User(
+        id=uuid.uuid4(),
+        email=f"user-{s}@example.com",
+        name=f"User {s}",
+        account_id=account_id,
+        account_role=role,
+        role="engineer",
+        is_active=True,
+    )
+    db.add(user)
+    await db.flush()
+    return user
+
+
+# ---------------------------------------------------------------------------
+# Tests
+# ---------------------------------------------------------------------------
+
+@pytest.mark.asyncio
+async def test_create_ticket_sets_status_open(test_db: AsyncSession):
+    account = await _make_account(test_db)
+    l1 = await _make_user(test_db, account_id=account.id)
+    ticket = await create_ticket(
+        test_db,
+        account_id=account.id,
+        created_by_user_id=l1.id,
+        problem_statement="Outlook can't connect",
+        customer_name="Alice",
+    )
+    assert ticket.status == 'open'
+    assert ticket.account_id == account.id
+    assert ticket.customer_name == "Alice"
+    assert ticket.created_by_user_id == l1.id
+
+
+@pytest.mark.asyncio
+async def test_update_status_to_resolved_sets_resolved_at(test_db: AsyncSession):
+    account = await _make_account(test_db)
+    l1 = await _make_user(test_db, account_id=account.id)
+    ticket = await create_ticket(
+        test_db,
+        account_id=account.id,
+        created_by_user_id=l1.id,
+        problem_statement="Test",
+    )
+    assert ticket.resolved_at is None
+    updated = await update_status(
+        test_db,
+        ticket_id=ticket.id,
+        status='resolved',
+        resolution_notes="Fixed via reboot",
+    )
+    assert updated.status == 'resolved'
+    assert updated.resolved_at is not None
+    assert updated.resolution_notes == "Fixed via reboot"
+
+
+@pytest.mark.asyncio
+async def test_update_status_to_escalated_does_not_set_resolved_at(test_db: AsyncSession):
+    account = await _make_account(test_db)
+    l1 = await _make_user(test_db, account_id=account.id)
+    ticket = await create_ticket(
+        test_db, account_id=account.id, created_by_user_id=l1.id,
+        problem_statement="x",
+    )
+    updated = await update_status(test_db, ticket_id=ticket.id, status='escalated')
+    assert updated.status == 'escalated'
+    assert updated.resolved_at is None
+
+
+@pytest.mark.asyncio
+async def test_update_status_assigns_user(test_db: AsyncSession):
+    account = await _make_account(test_db)
+    l1 = await _make_user(test_db, account_id=account.id)
+    engineer = await _make_user(test_db, account_id=account.id, role="engineer")
+    ticket = await create_ticket(
+        test_db, account_id=account.id, created_by_user_id=l1.id,
+        problem_statement="x",
+    )
+    updated = await update_status(
+        test_db, ticket_id=ticket.id, status='escalated',
+        assigned_user_id=engineer.id,
+    )
+    assert updated.assigned_user_id == engineer.id
+
+
+@pytest.mark.asyncio
+async def test_get_ticket_returns_none_for_missing_id(test_db: AsyncSession):
+    result = await get_ticket(test_db, ticket_id=uuid.uuid4())
+    assert result is None
+
+
+@pytest.mark.asyncio
+async def test_list_tickets_filters_by_account(test_db: AsyncSession):
+    account_a = await _make_account(test_db)
+    account_b = await _make_account(test_db)
+    l1_a = await _make_user(test_db, account_id=account_a.id)
+    l1_b = await _make_user(test_db, account_id=account_b.id)
+    ticket_a = await create_ticket(
+        test_db, account_id=account_a.id, created_by_user_id=l1_a.id,
+        problem_statement="A",
+    )
+    ticket_b = await create_ticket(
+        test_db, account_id=account_b.id, created_by_user_id=l1_b.id,
+        problem_statement="B",
+    )
+    rows = await list_tickets_for_account(test_db, account_id=account_a.id)
+    ids = [r.id for r in rows]
+    assert ticket_a.id in ids
+    assert ticket_b.id not in ids
+
+
+@pytest.mark.asyncio
+async def test_list_tickets_filters_by_status(test_db: AsyncSession):
+    account = await _make_account(test_db)
+    l1 = await _make_user(test_db, account_id=account.id)
+    open_t = await create_ticket(
+        test_db, account_id=account.id, created_by_user_id=l1.id,
+        problem_statement="open",
+    )
+    resolved_t = await create_ticket(
+        test_db, account_id=account.id, created_by_user_id=l1.id,
+        problem_statement="r",
+    )
+    await update_status(test_db, ticket_id=resolved_t.id, status='resolved')
+    open_rows = await list_tickets_for_account(test_db, account_id=account.id, status='open')
+    assert open_t.id in [r.id for r in open_rows]
+    assert resolved_t.id not in [r.id for r in open_rows]
+
+
+@pytest.mark.asyncio
+async def test_promote_to_psa_sets_external_id(test_db: AsyncSession):
+    account = await _make_account(test_db)
+    l1 = await _make_user(test_db, account_id=account.id)
+    ticket = await create_ticket(
+        test_db, account_id=account.id, created_by_user_id=l1.id,
+        problem_statement="x",
+    )
+    updated = await promote_to_psa(test_db, ticket_id=ticket.id, psa_ticket_id="CW-12345")
+    assert updated.psa_promoted_ticket_id == "CW-12345"
+
+
+@pytest.mark.asyncio
+async def test_update_status_raises_for_missing_ticket(test_db: AsyncSession):
+    with pytest.raises(ValueError, match="not found"):
+        await update_status(test_db, ticket_id=uuid.uuid4(), status='resolved')