feat: mid-session status updates — ticket notes, client updates, email drafts

Engineers can now generate AI-powered status updates during active FlowPilot
sessions and after resolve/escalate. Three audiences (Ticket Notes, Client
Update, Email Draft) with Quick/Detailed length options. Copy to clipboard
with one click. Client names auto-inserted from intake/PSA context.

Backend: new endpoint POST /ai-sessions/{id}/status-update with audience-aware
system prompts. Frontend: StatusUpdateModal with 2-step selection flow,
Share Update button in action bar, Share Resolution/Escalation on completed
sessions. Also updates Solutions Library spec with Community tier design.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-03-23 06:26:32 +00:00
parent 0d78410dea
commit fab25456a5
13 changed files with 1560 additions and 5 deletions

View File

@@ -36,6 +36,8 @@ from app.schemas.ai_session import (
SessionCloseResponse,
SessionDocumentation,
RateSessionRequest,
StatusUpdateRequest,
StatusUpdateResponse,
PickupSessionRequest,
LinkTicketRequest,
AISessionSummary,
@@ -733,6 +735,31 @@ async def get_documentation(
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=str(e))
# ── Status Update ──
@router.post("/{session_id}/status-update", response_model=StatusUpdateResponse)
@limiter.limit("20/minute")
async def create_status_update(
request: Request,
session_id: UUID,
data: StatusUpdateRequest,
current_user: Annotated[User, Depends(get_current_active_user)],
db: Annotated[AsyncSession, Depends(get_db)],
):
"""Generate a status update for ticket notes, client, or email."""
try:
return await flowpilot_engine.generate_status_update(
session_id=session_id,
request=data,
user_id=current_user.id,
db=db,
)
except ValueError as e:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(e))
except PermissionError as e:
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=str(e))
# ── Rate ──
@router.post("/{session_id}/rate", status_code=204)

View File

@@ -137,6 +137,38 @@ class SessionCloseResponse(BaseModel):
member_mapping_warning: str | None = None
class StatusUpdateRequest(BaseModel):
"""Generate a mid-session or post-session status update."""
audience: str = Field(
...,
pattern="^(ticket_notes|client_update|email_draft)$",
description="Who is this update for?",
)
length: str = Field(
"detailed",
pattern="^(quick|detailed)$",
description="Quick (1-2 sentences) or detailed breakdown",
)
context: str = Field(
"status",
pattern="^(status|resolution|escalation)$",
description="What type of communication: mid-session status, resolution close-out, or escalation handoff",
)
class StatusUpdateResponse(BaseModel):
"""Generated status update content."""
content: str
audience: str
length: str
context: str
session_status: str
steps_completed: int
time_spent_display: str | None = None
client_name: str | None = None
generated_at: datetime
class RateSessionRequest(BaseModel):
"""Submit post-session rating."""
rating: int = Field(..., ge=1, le=5)

View File

@@ -34,6 +34,8 @@ from app.schemas.ai_session import (
SessionCloseResponse,
SessionDocumentation,
DocumentationStep,
StatusUpdateRequest,
StatusUpdateResponse,
)
logger = logging.getLogger(__name__)
@@ -840,6 +842,237 @@ async def get_session_documentation(
return _generate_documentation(session)
async def generate_status_update(
session_id: UUID,
request: StatusUpdateRequest,
user_id: UUID,
db: AsyncSession,
) -> StatusUpdateResponse:
"""Generate a status update for ticket notes, client communication, or email draft."""
session = await _load_session(session_id, user_id, db)
# Build conversation summary from session steps
steps_summary = []
for step in sorted(session.steps, key=lambda s: s.step_order):
content = step.content or {}
text = content.get("text", "")
response = step.free_text_input or step.selected_option or ("Skipped" if step.was_skipped else None)
outcome = None
if step.action_result:
outcome = "Succeeded" if step.action_result.get("success") else "Did not resolve"
entry = f"Step {step.step_order + 1}: {text}"
if response:
entry += f"\n Engineer response: {response}"
if outcome:
entry += f"\n Outcome: {outcome}"
steps_summary.append(entry)
steps_text = "\n".join(steps_summary) if steps_summary else "No diagnostic steps yet."
# Calculate time spent
now = datetime.now(timezone.utc)
ref_time = session.resolved_at or now
delta = ref_time - session.created_at
total_minutes = int(delta.total_seconds() / 60)
if total_minutes < 60:
time_display = f"{total_minutes} minutes"
else:
hours = total_minutes // 60
remaining = total_minutes % 60
time_display = f"{hours}h {remaining}m"
# Extract client name from intake or ticket data
client_name = None
intake = session.intake_content or {}
if session.ticket_data:
company = session.ticket_data.get("company", {})
client_name = company.get("name") if isinstance(company, dict) else None
if not client_name:
client_name = intake.get("client_name") or intake.get("company_name")
# Get engineer name for sign-off
engineer_name = session.user.name if session.user and session.user.name else "Your support team"
# Build system prompt based on audience and context
system_prompt = _build_status_update_prompt(
audience=request.audience,
length=request.length,
context=request.context,
client_name=client_name,
engineer_name=engineer_name,
)
# Build user message with full session context
user_message = _build_status_update_context(
session=session,
steps_text=steps_text,
time_display=time_display,
context=request.context,
client_name=client_name,
)
provider = get_ai_provider(settings.get_model_for_action("quick_action"))
raw_response, input_tokens, output_tokens = await provider.generate_text(
system_prompt=system_prompt,
messages=[{"role": "user", "content": user_message}],
max_tokens=1500,
)
# Track token usage
session.total_input_tokens = (session.total_input_tokens or 0) + input_tokens
session.total_output_tokens = (session.total_output_tokens or 0) + output_tokens
# Store as a session step
step = AISessionStep(
id=uuid.uuid4(),
session_id=session.id,
step_order=session.step_count,
step_type="status_update",
content={
"audience": request.audience,
"length": request.length,
"context": request.context,
"generated_content": raw_response.strip(),
"client_name": client_name,
},
confidence_score=1.0,
confidence_tier="high",
)
db.add(step)
session.step_count += 1
await db.flush()
return StatusUpdateResponse(
content=raw_response.strip(),
audience=request.audience,
length=request.length,
context=request.context,
session_status=session.status,
steps_completed=len(steps_summary),
time_spent_display=time_display,
client_name=client_name,
generated_at=now,
)
def _build_status_update_prompt(
audience: str,
length: str,
context: str,
client_name: str | None,
engineer_name: str,
) -> str:
"""Build the system prompt for status update generation."""
length_instruction = (
"Keep it to 1-2 sentences maximum. Just the essentials."
if length == "quick"
else "Provide a full breakdown with steps completed, findings, and next steps."
)
context_labels = {
"status": "mid-session progress update",
"resolution": "resolution close-out summary",
"escalation": "escalation handoff note",
}
context_label = context_labels.get(context, "status update")
if audience == "ticket_notes":
return f"""You are generating an internal {context_label} for a PSA ticket note.
Rules:
- Be technical, concise, and factual
- Use markdown formatting (bold headers, bullet lists)
- Include: current status, steps completed, findings, what's been ruled out, next steps
- Do NOT soften language or add pleasantries
- Do NOT include greetings or sign-offs
- {length_instruction}
{"- Include root cause and resolution details since this is a close-out note" if context == "resolution" else ""}
{"- Include what was tried, what failed, and why this is being escalated" if context == "escalation" else ""}
Output ONLY the update text. No JSON, no markdown code fences, no preamble."""
elif audience == "client_update":
client_greeting = f"Address the client as '{client_name}'" if client_name else "Use a generic greeting like 'Hi'"
return f"""You are generating a client-facing {context_label}.
Rules:
- Be professional, reassuring, and non-technical
- NEVER use technical jargon (no "transport rules", "MX records", "DNS", "registry", "GPO", etc.)
- NEVER include server names, IP addresses, internal tool names, or technical identifiers
- Explain findings in plain language a non-technical business owner would understand
- {client_greeting}
- Sign off with: {engineer_name}
- {length_instruction}
{"- This is good news — the issue is resolved. Summarize what was wrong and what was done in plain language." if context == "resolution" else ""}
{"- Be reassuring — explain that a specialist is being brought in, not that something failed." if context == "escalation" else ""}
Output ONLY the update text. No JSON, no markdown code fences, no preamble."""
else: # email_draft
client_greeting = f"Address the client as '{client_name}'" if client_name else "Use a generic greeting like 'Hi'"
subject_hints = {
"status": "Update: [brief issue description]",
"resolution": "Resolved: [brief issue description]",
"escalation": "Update: [brief issue description] — Specialist Review",
}
return f"""You are generating a complete email draft for client communication.
Rules:
- Include a Subject: line at the very top
- Subject format: {subject_hints.get(context, "Update: [issue]")}
- {client_greeting}
- Be professional, reassuring, and non-technical
- NEVER use technical jargon, server names, IP addresses, or internal tool names
- Include a professional sign-off with:
{engineer_name}
- {length_instruction}
{"- This is good news — the issue is resolved." if context == "resolution" else ""}
{"- Be reassuring — explain that a specialist is being brought in." if context == "escalation" else ""}
Output ONLY the email text (Subject + body). No JSON, no markdown code fences, no preamble."""
def _build_status_update_context(
session: AISession,
steps_text: str,
time_display: str,
context: str,
client_name: str | None,
) -> str:
"""Build the user message containing full session context for the AI."""
parts = [
f"Session status: {session.status}",
f"Time spent: {time_display}",
f"Problem summary: {session.problem_summary or 'Not yet determined'}",
]
if session.problem_domain:
parts.append(f"Problem domain: {session.problem_domain}")
if client_name:
parts.append(f"Client: {client_name}")
if session.psa_ticket_id:
parts.append(f"Ticket ID: {session.psa_ticket_id}")
parts.append(f"\nDiagnostic steps:\n{steps_text}")
if context == "resolution" and session.resolution_summary:
parts.append(f"\nResolution: {session.resolution_summary}")
if context == "escalation" and session.escalation_reason:
parts.append(f"\nEscalation reason: {session.escalation_reason}")
# Include recent conversation messages for richer context
messages = session.conversation_messages or []
if messages:
recent = messages[-10:] # Last 10 messages
convo_text = "\n".join(
f"{'Engineer' if m['role'] == 'user' else 'FlowPilot'}: {m['content'][:300]}"
for m in recent
if isinstance(m, dict) and "role" in m and "content" in m
)
parts.append(f"\nRecent conversation:\n{convo_text}")
return "\n".join(parts)
# ── Internal helpers ──
async def _load_session(