453 lines
17 KiB
Python
453 lines
17 KiB
Python
"""PSA Documentation Push Service.
|
|
|
|
Generates structured documentation from FlowPilot AI sessions and pushes
|
|
it back to ConnectWise as internal notes + optional time entries.
|
|
"""
|
|
import logging
|
|
import math
|
|
import uuid
|
|
from datetime import datetime, timezone, timedelta
|
|
from typing import Optional, Any
|
|
from uuid import UUID
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.ai_session import AISession
|
|
from app.models.psa_activity_log import PsaActivityLog
|
|
from app.models.psa_connection import PsaConnection
|
|
from app.models.psa_member_mapping import PsaMemberMapping
|
|
from app.models.psa_post_log import PsaPostLog
|
|
from app.services.psa.registry import get_provider_for_connection
|
|
from app.services.psa.types import NoteType
|
|
from app.services.redaction_service import apply_redaction_to_text
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Default flowpilot_settings values
|
|
DEFAULT_SETTINGS = {
|
|
"auto_push": True,
|
|
"auto_time_entry": True,
|
|
"time_rounding": "15min", # "15min", "30min", "exact", "none"
|
|
"note_visibility": "internal", # "internal", "both"
|
|
"include_diagnostic_steps": True,
|
|
}
|
|
|
|
|
|
def _get_setting(connection: PsaConnection, key: str) -> Any:
|
|
"""Get a flowpilot setting with default fallback."""
|
|
settings = connection.flowpilot_settings or {}
|
|
return settings.get(key, DEFAULT_SETTINGS.get(key))
|
|
|
|
|
|
def _round_hours(hours: float, rounding: str) -> float:
|
|
"""Round hours according to the rounding setting."""
|
|
if rounding == "exact":
|
|
return round(hours, 2)
|
|
elif rounding == "30min":
|
|
return math.ceil(hours * 2) / 2
|
|
else: # default 15min
|
|
return math.ceil(hours * 4) / 4
|
|
|
|
|
|
def _format_datetime(dt: datetime | None) -> str:
|
|
"""Format a datetime for display in notes."""
|
|
if not dt:
|
|
return "N/A"
|
|
return dt.strftime("%Y-%m-%d %I:%M %p UTC")
|
|
|
|
|
|
def _get_engineer_response(step) -> str | None:
|
|
"""Extract the engineer's response label from a step."""
|
|
if step.was_skipped:
|
|
return "Skipped"
|
|
if step.selected_option and step.options_presented:
|
|
for opt in step.options_presented:
|
|
if opt.get("value") == step.selected_option:
|
|
return opt.get("label", step.selected_option)
|
|
return step.selected_option
|
|
if step.selected_option:
|
|
return step.selected_option
|
|
if step.free_text_input:
|
|
return step.free_text_input
|
|
return None
|
|
|
|
|
|
def format_resolution_note(session: AISession, include_steps: bool = True) -> str:
|
|
"""Format a resolved session as a plain-text note for CW."""
|
|
engineer_name = getattr(session, 'user', None)
|
|
engineer_display = engineer_name.name if engineer_name and hasattr(engineer_name, 'name') else "Unknown"
|
|
|
|
duration_str = ""
|
|
if session.resolved_at and session.created_at:
|
|
delta = session.resolved_at - session.created_at
|
|
total_hrs = round(delta.total_seconds() / 3600, 2)
|
|
duration_str = f" — {total_hrs} hrs"
|
|
|
|
lines = [
|
|
f"FlowPilot Session — {engineer_display}{duration_str}",
|
|
f"Problem: {session.problem_summary or 'No summary available'}",
|
|
]
|
|
|
|
# Diagnostic steps
|
|
if include_steps and session.steps:
|
|
lines.append("")
|
|
lines.append("Steps:")
|
|
for step in session.steps:
|
|
content = step.content or {}
|
|
step_type = content.get("type", "")
|
|
if step_type == "resolution_suggestion":
|
|
continue # Not a diagnostic step
|
|
description = content.get("text", "").strip()
|
|
if not description:
|
|
continue
|
|
response = _get_engineer_response(step)
|
|
line = f"{step.step_order + 1}. {description}"
|
|
if response and response != "Skipped":
|
|
line += f" — {response}"
|
|
elif response == "Skipped":
|
|
line += " (skipped)"
|
|
lines.append(line)
|
|
|
|
# Resolution
|
|
lines.append("")
|
|
lines.append(f"Resolution: {session.resolution_summary or 'No resolution summary'}")
|
|
if session.resolution_action:
|
|
lines.append(session.resolution_action)
|
|
|
|
# Follow-up recommendations from resolution suggestion step
|
|
follow_ups: list[str] = []
|
|
for step in session.steps:
|
|
content = step.content or {}
|
|
if content.get("type") == "resolution_suggestion":
|
|
recs = content.get("follow_up_recommendations", [])
|
|
if isinstance(recs, list):
|
|
follow_ups.extend(recs)
|
|
if follow_ups:
|
|
lines.append("")
|
|
lines.append("Follow-up:")
|
|
for rec in follow_ups:
|
|
lines.append(f"- {rec}")
|
|
|
|
# Timing
|
|
lines.append("")
|
|
lines.append(f"Start: {_format_datetime(session.created_at)}")
|
|
lines.append(f"End: {_format_datetime(session.resolved_at)}")
|
|
if session.resolved_at and session.created_at:
|
|
delta = session.resolved_at - session.created_at
|
|
total_hrs = round(delta.total_seconds() / 3600, 2)
|
|
lines.append(f"Total: {total_hrs} hrs")
|
|
|
|
lines.append("")
|
|
lines.append("Generated by ResolutionFlow")
|
|
|
|
return "\n".join(lines)
|
|
|
|
|
|
def _derive_what_we_know(session: AISession) -> tuple[list[str], list[str], list[str]]:
|
|
"""Return (confirmed, ruled_out, pending) findings.
|
|
|
|
Uses session.evidence_items when the cockpit branch is merged; falls back
|
|
to deriving from completed diagnostic steps.
|
|
"""
|
|
evidence_items = getattr(session, 'evidence_items', None)
|
|
if evidence_items:
|
|
confirmed = [e['text'] for e in evidence_items if e.get('status') == 'confirmed']
|
|
ruled_out = [e['text'] for e in evidence_items if e.get('status') == 'ruled_out']
|
|
pending = [e['text'] for e in evidence_items if e.get('status') == 'pending']
|
|
return confirmed, ruled_out, pending
|
|
|
|
# Derive from completed steps — all answered steps become findings
|
|
findings = []
|
|
for step in sorted(session.steps or [], key=lambda s: s.step_order):
|
|
content = step.content or {}
|
|
if content.get("type") in ("resolution_suggestion", "briefing", "status_update"):
|
|
continue
|
|
description = content.get("text", "").strip()
|
|
if not description or step.was_skipped:
|
|
continue
|
|
response = _get_engineer_response(step)
|
|
if response:
|
|
findings.append(f"{description} — {response}")
|
|
return findings, [], []
|
|
|
|
|
|
def format_escalation_note(session: AISession, include_steps: bool = True) -> str:
|
|
"""Format an escalated session as a plain-text note for CW."""
|
|
engineer_obj = getattr(session, 'user', None)
|
|
engineer_display = engineer_obj.name if engineer_obj and hasattr(engineer_obj, 'name') else "Unknown"
|
|
|
|
escalated_to_obj = getattr(session, 'escalated_to', None)
|
|
escalated_to_display = escalated_to_obj.name if escalated_to_obj and hasattr(escalated_to_obj, 'name') else None
|
|
|
|
escalated_at = session.resolved_at or datetime.now(timezone.utc)
|
|
duration_str = ""
|
|
if session.created_at:
|
|
delta = escalated_at - session.created_at
|
|
total_hrs = round(delta.total_seconds() / 3600, 2)
|
|
duration_str = f" — {total_hrs} hrs"
|
|
|
|
header = f"FlowPilot Escalation — {engineer_display}{duration_str}"
|
|
if escalated_to_display:
|
|
header += f" → {escalated_to_display}"
|
|
lines = [
|
|
header,
|
|
f"Problem: {session.problem_summary or 'No summary available'}",
|
|
]
|
|
|
|
# Work completed with responses
|
|
if include_steps and session.steps:
|
|
lines.append("")
|
|
lines.append("Work completed:")
|
|
for step in sorted(session.steps, key=lambda s: s.step_order):
|
|
content = step.content or {}
|
|
if content.get("type") in ("resolution_suggestion", "briefing", "status_update"):
|
|
continue
|
|
description = content.get("text", "").strip()
|
|
if not description:
|
|
continue
|
|
response = _get_engineer_response(step)
|
|
line = f"{step.step_order + 1}. {description}"
|
|
if response and response != "Skipped":
|
|
line += f" — {response}"
|
|
elif response == "Skipped":
|
|
line += " (skipped)"
|
|
lines.append(line)
|
|
|
|
# What We Know
|
|
confirmed, ruled_out, pending = _derive_what_we_know(session)
|
|
if confirmed or ruled_out or pending:
|
|
lines.append("")
|
|
lines.append("What we know:")
|
|
for f in confirmed:
|
|
lines.append(f" ✓ {f}")
|
|
for f in ruled_out:
|
|
lines.append(f" ✗ {f}")
|
|
for f in pending:
|
|
lines.append(f" ? {f}")
|
|
|
|
# Escalation reason
|
|
lines.append("")
|
|
lines.append(f"Escalation reason: {session.escalation_reason or 'No reason provided'}")
|
|
|
|
# Suggested next steps from escalation package
|
|
pkg = session.escalation_package or {}
|
|
if suggestions := pkg.get("suggested_next_steps"):
|
|
lines.append("")
|
|
lines.append("Suggested next steps:")
|
|
items = suggestions if isinstance(suggestions, list) else [str(suggestions)]
|
|
for s in items:
|
|
lines.append(f"- {s}")
|
|
|
|
# Timing
|
|
lines.append("")
|
|
lines.append(f"Start: {_format_datetime(session.created_at)}")
|
|
lines.append(f"Escalated: {_format_datetime(escalated_at)}")
|
|
if session.created_at:
|
|
delta = escalated_at - session.created_at
|
|
total_hrs = round(delta.total_seconds() / 3600, 2)
|
|
lines.append(f"Total: {total_hrs} hrs")
|
|
|
|
lines.append("")
|
|
lines.append("Generated by ResolutionFlow")
|
|
|
|
return "\n".join(lines)
|
|
|
|
|
|
async def push_documentation(
|
|
session: AISession,
|
|
user_id: UUID,
|
|
db: AsyncSession,
|
|
) -> dict[str, Any]:
|
|
"""Push session documentation to PSA ticket.
|
|
|
|
Returns:
|
|
{
|
|
"psa_push_status": "sent" | "pending_retry" | "failed" | "no_psa",
|
|
"psa_push_error": str | None,
|
|
"member_mapping_warning": str | None,
|
|
}
|
|
"""
|
|
if not session.psa_ticket_id or not session.psa_connection_id:
|
|
return {"psa_push_status": "no_psa", "psa_push_error": None, "member_mapping_warning": None}
|
|
|
|
# Load connection and check settings
|
|
result = await db.execute(
|
|
select(PsaConnection).where(PsaConnection.id == session.psa_connection_id)
|
|
)
|
|
connection = result.scalar_one_or_none()
|
|
if not connection:
|
|
return {"psa_push_status": "failed", "psa_push_error": "PSA connection not found", "member_mapping_warning": None}
|
|
|
|
if not _get_setting(connection, "auto_push"):
|
|
return {"psa_push_status": "no_psa", "psa_push_error": None, "member_mapping_warning": None}
|
|
|
|
# Format the note
|
|
include_steps = _get_setting(connection, "include_diagnostic_steps")
|
|
if session.status == "resolved":
|
|
note_text = format_resolution_note(session, include_steps=include_steps)
|
|
else:
|
|
note_text = format_escalation_note(session, include_steps=include_steps)
|
|
|
|
# Redact sensitive data
|
|
note_text, _ = apply_redaction_to_text(note_text)
|
|
|
|
# Determine note type
|
|
visibility = _get_setting(connection, "note_visibility")
|
|
note_type = NoteType.INTERNAL_ANALYSIS if visibility == "internal" else NoteType.DESCRIPTION
|
|
|
|
# Check member mapping for time entry
|
|
member_mapping_warning = None
|
|
member_mapping = None
|
|
if _get_setting(connection, "auto_time_entry") and _get_setting(connection, "time_rounding") != "none":
|
|
mapping_result = await db.execute(
|
|
select(PsaMemberMapping).where(
|
|
PsaMemberMapping.psa_connection_id == session.psa_connection_id,
|
|
PsaMemberMapping.user_id == user_id,
|
|
)
|
|
)
|
|
member_mapping = mapping_result.scalar_one_or_none()
|
|
if not member_mapping:
|
|
member_mapping_warning = "Map your CW account in Settings → Integrations to enable auto-logged time entries."
|
|
|
|
# Push to PSA
|
|
try:
|
|
provider = await get_provider_for_connection(session.psa_connection_id, db)
|
|
|
|
# Post the note
|
|
posted_note = await provider.post_note(
|
|
ticket_id=session.psa_ticket_id,
|
|
text=note_text,
|
|
note_type=note_type,
|
|
)
|
|
|
|
# Create time entry if member mapping exists
|
|
time_entry_hours: Optional[float] = None
|
|
if member_mapping and session.resolved_at and session.created_at:
|
|
try:
|
|
delta = session.resolved_at - session.created_at
|
|
hours = delta.total_seconds() / 3600
|
|
rounding = _get_setting(connection, "time_rounding")
|
|
rounded_hours = _round_hours(hours, rounding)
|
|
if rounded_hours > 0:
|
|
await provider.create_time_entry(
|
|
ticket_id=session.psa_ticket_id,
|
|
member_id=member_mapping.external_member_id,
|
|
hours=rounded_hours,
|
|
notes=f"FlowPilot session: {session.problem_summary or 'Troubleshooting'}",
|
|
)
|
|
time_entry_hours = rounded_hours
|
|
except Exception as e:
|
|
logger.warning("Failed to create time entry for session %s: %s", session.id, e)
|
|
# Don't fail the note push just because time entry failed
|
|
|
|
# Log PSA activity — note posted
|
|
try:
|
|
note_activity = PsaActivityLog(
|
|
account_id=session.account_id,
|
|
session_id=session.id,
|
|
activity_type="note_posted",
|
|
hours_logged=None,
|
|
psa_ticket_id=session.psa_ticket_id,
|
|
)
|
|
db.add(note_activity)
|
|
except Exception as e:
|
|
logger.warning("Failed to log PSA note activity for session %s: %s", session.id, e)
|
|
|
|
# Log time entry activity if one was created
|
|
if time_entry_hours is not None:
|
|
try:
|
|
time_activity = PsaActivityLog(
|
|
account_id=session.account_id,
|
|
session_id=session.id,
|
|
activity_type="time_entry_posted",
|
|
hours_logged=time_entry_hours,
|
|
psa_ticket_id=session.psa_ticket_id,
|
|
)
|
|
db.add(time_activity)
|
|
except Exception as e:
|
|
logger.warning("Failed to log PSA time entry activity for session %s: %s", session.id, e)
|
|
|
|
# Log success
|
|
log_entry = PsaPostLog(
|
|
id=uuid.uuid4(),
|
|
account_id=session.account_id,
|
|
ai_session_id=session.id,
|
|
psa_connection_id=session.psa_connection_id,
|
|
ticket_id=session.psa_ticket_id,
|
|
note_type=note_type,
|
|
content_posted=note_text[:10000], # Truncate for storage
|
|
external_note_id=posted_note.id,
|
|
status="success",
|
|
posted_by=user_id,
|
|
)
|
|
db.add(log_entry)
|
|
|
|
return {
|
|
"psa_push_status": "sent",
|
|
"psa_push_error": None,
|
|
"member_mapping_warning": member_mapping_warning,
|
|
}
|
|
|
|
except Exception as e:
|
|
logger.warning("PSA push failed for session %s: %s", session.id, e)
|
|
|
|
# Log failure with retry scheduling
|
|
log_entry = PsaPostLog(
|
|
id=uuid.uuid4(),
|
|
account_id=session.account_id,
|
|
ai_session_id=session.id,
|
|
psa_connection_id=session.psa_connection_id,
|
|
ticket_id=session.psa_ticket_id,
|
|
note_type=note_type,
|
|
content_posted=note_text[:10000],
|
|
status="pending_retry",
|
|
error_message=str(e)[:500],
|
|
retry_count=0,
|
|
next_retry_at=datetime.now(timezone.utc) + timedelta(minutes=5),
|
|
posted_by=user_id,
|
|
)
|
|
db.add(log_entry)
|
|
|
|
return {
|
|
"psa_push_status": "pending_retry",
|
|
"psa_push_error": str(e)[:200],
|
|
"member_mapping_warning": member_mapping_warning,
|
|
}
|
|
|
|
|
|
async def retry_failed_push(
|
|
log_entry: PsaPostLog,
|
|
db: AsyncSession,
|
|
) -> bool:
|
|
"""Retry a failed PSA push. Returns True on success."""
|
|
try:
|
|
provider = await get_provider_for_connection(log_entry.psa_connection_id, db)
|
|
posted_note = await provider.post_note(
|
|
ticket_id=log_entry.ticket_id,
|
|
text=log_entry.content_posted,
|
|
note_type=log_entry.note_type,
|
|
)
|
|
log_entry.status = "success"
|
|
log_entry.external_note_id = posted_note.id
|
|
log_entry.error_message = None
|
|
log_entry.next_retry_at = None
|
|
return True
|
|
except Exception as e:
|
|
log_entry.retry_count += 1
|
|
log_entry.error_message = str(e)[:500]
|
|
|
|
if log_entry.retry_count >= 3:
|
|
log_entry.status = "failed"
|
|
log_entry.next_retry_at = None
|
|
else:
|
|
# Exponential backoff: 5min, 15min, 45min
|
|
backoff_minutes = 5 * (3 ** log_entry.retry_count)
|
|
log_entry.next_retry_at = datetime.now(timezone.utc) + timedelta(minutes=backoff_minutes)
|
|
|
|
logger.warning(
|
|
"PSA retry %d failed for log %s: %s",
|
|
log_entry.retry_count, log_entry.id, e,
|
|
)
|
|
return False
|