diff --git a/backend/app/services/l1_session_service.py b/backend/app/services/l1_session_service.py new file mode 100644 index 00000000..5a33c5e3 --- /dev/null +++ b/backend/app/services/l1_session_service.py @@ -0,0 +1,93 @@ +"""L1 session lifecycle: start (flow/proposal/adhoc), step, notes, resolve, escalate. + +start_* functions live in T12; step/notes are T13; resolve/escalate are T14. +""" +from typing import Optional +from uuid import UUID + +from sqlalchemy.ext.asyncio import AsyncSession + +from app.models.l1_walk_session import L1WalkSession +from app.models.user import User + + +def _resolve_acting_as(user: User) -> Optional[str]: + """An engineer (whether covering or not) gets tagged for audit when using L1 surface. + + Returns 'l1_coverage' for engineers (only engineers WITH the coverage flag should + reach this code path — the require_l1_or_coverage dep gates that). For native + l1_tech users, returns None (no special tag — they ARE l1). + """ + if user.account_role == "engineer": + return "l1_coverage" + return None + + +async def start_flow_session( + db: AsyncSession, + *, + account_id: UUID, + user: User, + flow_id: UUID, + ticket_id: str, + ticket_kind: str, # 'psa' | 'internal' +) -> L1WalkSession: + """Start a session walking an authored flow.""" + session = L1WalkSession( + account_id=account_id, + created_by_user_id=user.id, + acting_as=_resolve_acting_as(user), + ticket_id=ticket_id, + ticket_kind=ticket_kind, + session_kind="flow", + flow_id=flow_id, + ) + db.add(session) + await db.flush() + return session + + +async def start_proposal_session( + db: AsyncSession, + *, + account_id: UUID, + user: User, + flow_proposal_id: UUID, + ticket_id: str, + ticket_kind: str, +) -> L1WalkSession: + """Start a session walking an AI-built FlowProposal.""" + session = L1WalkSession( + account_id=account_id, + created_by_user_id=user.id, + acting_as=_resolve_acting_as(user), + ticket_id=ticket_id, + ticket_kind=ticket_kind, + session_kind="proposal", + flow_proposal_id=flow_proposal_id, + ) + db.add(session) + await db.flush() + return session + + +async def start_adhoc_session( + db: AsyncSession, + *, + account_id: UUID, + user: User, + ticket_id: str, + ticket_kind: str, +) -> L1WalkSession: + """Start an ad-hoc session with no tree (free-form note-taking only).""" + session = L1WalkSession( + account_id=account_id, + created_by_user_id=user.id, + acting_as=_resolve_acting_as(user), + ticket_id=ticket_id, + ticket_kind=ticket_kind, + session_kind="adhoc", + ) + db.add(session) + await db.flush() + return session diff --git a/backend/tests/test_l1_session_service.py b/backend/tests/test_l1_session_service.py new file mode 100644 index 00000000..d6bd687d --- /dev/null +++ b/backend/tests/test_l1_session_service.py @@ -0,0 +1,221 @@ +"""Tests for l1_session_service start_* functions (T12).""" +import uuid +import pytest +from sqlalchemy.ext.asyncio import AsyncSession + +from app.models.account import Account +from app.models.user import User +from app.models.tree import Tree +from app.models.ai_session import AISession +from app.models.flow_proposal import FlowProposal +from app.services.l1_session_service import ( + start_flow_session, + start_proposal_session, + start_adhoc_session, + _resolve_acting_as, +) + + +# --------------------------------------------------------------------------- +# 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].upper(), + ) + db.add(account) + await db.flush() + return account + + +async def _make_user( + db: AsyncSession, + *, + account_id: uuid.UUID, + account_role: str = "l1_tech", + can_cover_l1: bool = False, +) -> User: + user = User( + id=uuid.uuid4(), + email=f"user-{uuid.uuid4()}@example.com", + name="Test User", + account_id=account_id, + account_role=account_role, + role="engineer", + is_active=True, + can_cover_l1=can_cover_l1, + ) + db.add(user) + await db.flush() + return user + + +async def _make_tree(db: AsyncSession, *, account_id: uuid.UUID, author_id: uuid.UUID) -> Tree: + tree = Tree( + id=uuid.uuid4(), + name="Test Flow", + account_id=account_id, + author_id=author_id, + tree_type="troubleshooting", + tree_structure={"nodes": [], "edges": []}, + visibility="team", + status="published", + ) + db.add(tree) + await db.flush() + return tree + + +async def _make_ai_session(db: AsyncSession, *, user_id: uuid.UUID, account_id: uuid.UUID) -> AISession: + ai_session = AISession( + id=uuid.uuid4(), + user_id=user_id, + account_id=account_id, + session_type="chat", + intake_type="free_text", + intake_content={"text": "test"}, + status="active", + confidence_tier="discovery", + conversation_messages=[], + ) + db.add(ai_session) + await db.flush() + return ai_session + + +async def _make_proposal( + db: AsyncSession, + *, + account_id: uuid.UUID, + source_session_id: uuid.UUID, +) -> FlowProposal: + proposal = FlowProposal( + id=uuid.uuid4(), + account_id=account_id, + source_session_id=source_session_id, + proposal_type="new_flow", + title="Test Proposal", + proposed_flow_data={"nodes": [], "edges": []}, + source="manual_draft", + status="pending", + ) + db.add(proposal) + await db.flush() + return proposal + + +# --------------------------------------------------------------------------- +# Unit tests for _resolve_acting_as (no DB needed) +# --------------------------------------------------------------------------- + +@pytest.mark.asyncio +async def test_resolve_acting_as_for_engineer_returns_coverage_tag(): + user = User(account_role="engineer") + assert _resolve_acting_as(user) == "l1_coverage" + + +@pytest.mark.asyncio +async def test_resolve_acting_as_for_l1_tech_returns_none(): + user = User(account_role="l1_tech") + assert _resolve_acting_as(user) is None + + +@pytest.mark.asyncio +async def test_resolve_acting_as_for_owner_returns_none(): + user = User(account_role="owner") + assert _resolve_acting_as(user) is None + + +# --------------------------------------------------------------------------- +# Integration tests (real DB) +# --------------------------------------------------------------------------- + +@pytest.mark.asyncio +async def test_start_flow_session_creates_active_flow_session(test_db: AsyncSession): + account = await _make_account(test_db) + l1 = await _make_user(test_db, account_id=account.id) + tree = await _make_tree(test_db, account_id=account.id, author_id=l1.id) + + session = await start_flow_session( + test_db, + account_id=account.id, + user=l1, + flow_id=tree.id, + ticket_id="ticket-abc", + ticket_kind="internal", + ) + assert session.session_kind == "flow" + assert session.flow_id == tree.id + assert session.flow_proposal_id is None + assert session.status == "active" + assert session.walked_path == [] + assert session.walk_notes == [] + assert session.acting_as is None # l1_tech native + assert session.ticket_id == "ticket-abc" + assert session.ticket_kind == "internal" + + +@pytest.mark.asyncio +async def test_start_proposal_session_creates_active_proposal_session(test_db: AsyncSession): + account = await _make_account(test_db) + l1 = await _make_user(test_db, account_id=account.id) + ai_session = await _make_ai_session(test_db, user_id=l1.id, account_id=account.id) + proposal = await _make_proposal( + test_db, + account_id=account.id, + source_session_id=ai_session.id, + ) + + session = await start_proposal_session( + test_db, + account_id=account.id, + user=l1, + flow_proposal_id=proposal.id, + ticket_id="ticket-xyz", + ticket_kind="psa", + ) + assert session.session_kind == "proposal" + assert session.flow_proposal_id == proposal.id + assert session.flow_id is None + assert session.status == "active" + + +@pytest.mark.asyncio +async def test_start_adhoc_session_no_flow_no_proposal(test_db: AsyncSession): + account = await _make_account(test_db) + l1 = await _make_user(test_db, account_id=account.id) + session = await start_adhoc_session( + test_db, + account_id=account.id, + user=l1, + ticket_id="ticket-adhoc", + ticket_kind="internal", + ) + assert session.session_kind == "adhoc" + assert session.flow_id is None + assert session.flow_proposal_id is None + assert session.walked_path == [] + assert session.walk_notes == [] + + +@pytest.mark.asyncio +async def test_engineer_with_coverage_gets_acting_as_tag(test_db: AsyncSession): + account = await _make_account(test_db) + eng = await _make_user( + test_db, + account_id=account.id, + account_role="engineer", + can_cover_l1=True, + ) + session = await start_adhoc_session( + test_db, + account_id=account.id, + user=eng, + ticket_id="t", + ticket_kind="internal", + ) + assert session.acting_as == "l1_coverage"