Michael Chihlas
6e7c4afc7d
Three start_* functions creating L1WalkSession rows with appropriate session_kind and target id. Engineers acting in L1 mode get acting_as='l1_coverage' for audit; native l1_tech users get acting_as=None. step/notes (T13) and resolve/escalate (T14) extend this file next. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
94 lines
2.5 KiB
Python
94 lines
2.5 KiB
Python
"""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
|