File size: 3,767 Bytes
5df55ff 8fab536 5df55ff 8fab536 5df55ff 8fab536 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 | """In-memory session manager for pitch battles."""
from __future__ import annotations
import re
import uuid
from typing import Any
SESSIONS: dict[str, dict[str, Any]] = {}
def create_session(
startup: dict,
persona: str,
difficulty: str,
input_mode: str,
) -> dict[str, Any]:
"""Create a new pitch battle session."""
session_id = str(uuid.uuid4())
session = {
"session_id": session_id,
"startup": startup,
"persona": persona,
"difficulty": difficulty,
"input_mode": input_mode,
"round": 1,
"history": [],
"voice_pitch": None,
"pending_voice_turns": {},
"confirmed_voice_turns": [],
}
SESSIONS[session_id] = session
return session
def get_session(session_id: str) -> dict[str, Any] | None:
"""Return a session by id, or None if missing."""
return SESSIONS.get(session_id)
def append_user_message(session_id: str, message: str) -> None:
"""Append a user message to session history."""
session = SESSIONS.get(session_id)
if not session:
return
session["history"].append({"role": "user", "content": message})
def append_ai_message(session_id: str, message: str, attack_tag: str) -> None:
"""Append an AI opponent message to session history."""
session = SESSIONS.get(session_id)
if not session:
return
session["history"].append(
{"role": "assistant", "content": message, "attack_tag": attack_tag}
)
def increment_round(session_id: str) -> int:
"""Increment and return the current round number."""
session = SESSIONS.get(session_id)
if not session:
return 0
session["round"] = session.get("round", 1) + 1
return session["round"]
def get_history(session_id: str) -> list[dict[str, Any]]:
"""Return conversation history for a session."""
session = SESSIONS.get(session_id)
if not session:
return []
return list(session.get("history", []))
def reset_session(session_id: str) -> bool:
"""Delete a session. Returns True if it existed."""
if session_id in SESSIONS:
del SESSIONS[session_id]
return True
return False
def set_voice_pitch(session_id: str, voice_pitch: dict[str, Any]) -> None:
"""Store opening voice pitch metadata on session."""
session = SESSIONS.get(session_id)
if session:
session["voice_pitch"] = voice_pitch
def store_pending_voice_turn(session_id: str, turn_record: dict[str, Any]) -> None:
"""Store a pending (unconfirmed) voice turn."""
session = SESSIONS.get(session_id)
if not session:
return
pending = session.setdefault("pending_voice_turns", {})
vid = turn_record.get("voice_turn_id", "")
if vid:
pending[vid] = turn_record
def confirm_voice_turn(
session_id: str,
voice_turn_id: str,
final_transcript: str,
) -> bool:
"""Confirm a pending voice turn and move it to confirmed_voice_turns."""
session = SESSIONS.get(session_id)
if not session:
return False
pending = session.get("pending_voice_turns") or {}
turn = pending.get(voice_turn_id)
if not turn:
return False
turn = dict(turn)
turn["transcript"] = str(final_transcript).strip()
turn["confirmed"] = True
turn["word_count"] = len(re.findall(r"\b\w+\b", turn["transcript"]))
session.setdefault("confirmed_voice_turns", []).append(turn)
del pending[voice_turn_id]
return True
def get_pending_voice_turn(session_id: str, voice_turn_id: str) -> dict[str, Any] | None:
"""Return a pending voice turn by id."""
session = SESSIONS.get(session_id)
if not session:
return None
return (session.get("pending_voice_turns") or {}).get(voice_turn_id)
|