"""Phase 2 TaskSession: in-memory + JSON file. Not a TaskInstance."""
|
|
from __future__ import annotations
|
|
import asyncio
|
import json
|
import os
|
import time
|
import uuid
|
from dataclasses import asdict, dataclass, field
|
from typing import Any, Dict, List, Optional, TYPE_CHECKING
|
|
from config.config_loader import get_project_dir
|
from config.logger import setup_logging
|
|
if TYPE_CHECKING:
|
from core.connection import ConnectionHandler
|
|
TAG = __name__
|
logger = setup_logging()
|
|
SKILL_EVENT_PREFIX = "[skill_event]"
|
DEFER_ASR_REPORT_MODES = ("PLAY_AUDIO", "SHADOWING")
|
|
|
def _now_ms() -> int:
|
return int(time.time() * 1000)
|
|
|
def session_dir() -> str:
|
return os.path.join(get_project_dir(), "data", "task_sessions")
|
|
|
def parse_skill_event(text: str) -> Optional[Dict[str, str]]:
|
"""Parse `[skill_event]audio.completed|app_id|task_id` into a dict."""
|
if not text:
|
return None
|
raw = str(text).strip()
|
if not raw.startswith(SKILL_EVENT_PREFIX):
|
return None
|
payload = raw[len(SKILL_EVENT_PREFIX) :].strip()
|
parts = [p.strip() for p in payload.split("|")]
|
name = parts[0] if parts else ""
|
if not name:
|
return None
|
return {
|
"name": name,
|
"app_id": parts[1] if len(parts) > 1 else "",
|
"task_id": parts[2] if len(parts) > 2 else "",
|
"duration_ms": parts[3] if len(parts) > 3 else "",
|
}
|
|
|
def should_report_asr_now(conn: Any) -> bool:
|
"""PLAY_AUDIO / SHADOWING defer report so echo and ignored uplink are not archived."""
|
mode = getattr(conn, "interaction_mode", "CHAT")
|
return mode not in DEFER_ASR_REPORT_MODES
|
|
|
def stash_asr_audio(conn: Any, audio_snapshot: Any) -> None:
|
conn.last_asr_audio_snapshot = audio_snapshot
|
|
|
@dataclass
|
class TaskSession:
|
task_session_id: str
|
ws_session_id: str = ""
|
device_id: str = ""
|
agent_id: str = ""
|
app_id: str = ""
|
task_id: str = ""
|
source: str = "voice"
|
interaction_mode: str = ""
|
started_at: int = 0
|
ended_at: int = 0
|
status: str = "running"
|
end_reason: str = ""
|
turns: List[Dict[str, Any]] = field(default_factory=list)
|
events: List[Dict[str, Any]] = field(default_factory=list)
|
|
def find_turn(self, index: int) -> Optional[Dict[str, Any]]:
|
for turn in self.turns:
|
if turn.get("index") == index:
|
return turn
|
return None
|
|
def upsert_turn(
|
self,
|
index: int,
|
prompt_text: str = "",
|
asr_text: Optional[str] = None,
|
echo_rejected: bool = False,
|
expected_text: Optional[str] = None,
|
metrics: Optional[List[Dict[str, Any]]] = None,
|
assessed_at: Optional[int] = None,
|
audio_id: Optional[str] = None,
|
) -> Dict[str, Any]:
|
"""Same step_index stays one row; last ASR wins. Empty metrics do not wipe old."""
|
existing = self.find_turn(index)
|
if existing is None:
|
turn: Dict[str, Any] = {
|
"index": index,
|
"prompt_text": prompt_text or "",
|
"asr_text": asr_text or "",
|
"echo_rejected": bool(echo_rejected),
|
"recorded_at": _now_ms(),
|
}
|
if expected_text:
|
turn["expected_text"] = expected_text
|
if metrics:
|
turn["metrics"] = metrics
|
if assessed_at is not None:
|
turn["assessed_at"] = assessed_at
|
if audio_id:
|
turn["audio_id"] = audio_id
|
self.turns.append(turn)
|
return turn
|
if prompt_text:
|
existing["prompt_text"] = prompt_text
|
if asr_text is not None:
|
existing["asr_text"] = asr_text
|
existing["echo_rejected"] = bool(echo_rejected)
|
existing["recorded_at"] = _now_ms()
|
if expected_text:
|
existing["expected_text"] = expected_text
|
if metrics:
|
existing["metrics"] = metrics
|
if assessed_at is not None:
|
existing["assessed_at"] = assessed_at
|
if audio_id:
|
existing["audio_id"] = audio_id
|
return existing
|
|
def add_turn(
|
self,
|
index: int,
|
prompt_text: str,
|
asr_text: str,
|
echo_rejected: bool = False,
|
) -> None:
|
self.upsert_turn(
|
index=index,
|
prompt_text=prompt_text,
|
asr_text=asr_text,
|
echo_rejected=echo_rejected,
|
)
|
|
def add_event(self, name: str, app_id: str = "", task_id: str = "") -> None:
|
self.events.append(
|
{
|
"name": name,
|
"app_id": app_id or self.app_id,
|
"task_id": task_id or self.task_id,
|
"at": _now_ms(),
|
}
|
)
|
|
def finish(self, reason: str) -> None:
|
if self.status != "running":
|
return
|
self.status = "completed" if reason in ("finished", "play_completed") else "cancelled"
|
self.end_reason = reason
|
self.ended_at = _now_ms()
|
|
def to_dict(self) -> Dict[str, Any]:
|
return asdict(self)
|
|
def save(self) -> None:
|
try:
|
directory = session_dir()
|
os.makedirs(directory, exist_ok=True)
|
path = os.path.join(directory, f"{self.task_session_id}.json")
|
with open(path, "w", encoding="utf-8") as handle:
|
json.dump(self.to_dict(), handle, ensure_ascii=False, indent=2)
|
except Exception as exc:
|
logger.bind(tag=TAG).warning(f"保存 TaskSession 失败: {exc}")
|
return
|
if getattr(self, "_skip_upsert", False):
|
return
|
try:
|
_schedule_upsert(self.to_dict(), self)
|
except Exception as exc:
|
logger.bind(tag=TAG).warning(f"TaskInstance 上报调度失败: {exc}")
|
|
@classmethod
|
def load(cls, task_session_id: str) -> Optional["TaskSession"]:
|
path = os.path.join(session_dir(), f"{task_session_id}.json")
|
if not os.path.isfile(path):
|
return None
|
with open(path, "r", encoding="utf-8") as handle:
|
data = json.load(handle)
|
return cls(
|
task_session_id=data.get("task_session_id") or task_session_id,
|
ws_session_id=data.get("ws_session_id") or "",
|
device_id=data.get("device_id") or "",
|
agent_id=data.get("agent_id") or "",
|
app_id=data.get("app_id") or "",
|
task_id=data.get("task_id") or "",
|
source=data.get("source") or "voice",
|
interaction_mode=data.get("interaction_mode") or "",
|
started_at=int(data.get("started_at") or 0),
|
ended_at=int(data.get("ended_at") or 0),
|
status=data.get("status") or "running",
|
end_reason=data.get("end_reason") or "",
|
turns=list(data.get("turns") or []),
|
events=list(data.get("events") or []),
|
)
|
|
|
def begin_session(conn: "ConnectionHandler", app_id: str, task_id: str, source: str, mode: str) -> TaskSession:
|
existing = getattr(conn, "task_session", None)
|
if existing and getattr(existing, "status", "") == "running":
|
end_session(conn, "preempted")
|
session = TaskSession(
|
task_session_id=uuid.uuid4().hex,
|
ws_session_id=str(getattr(conn, "session_id", "") or ""),
|
device_id=str(getattr(conn, "device_id", "") or ""),
|
agent_id=str(getattr(conn, "agent_id", "") or ""),
|
app_id=app_id,
|
task_id=task_id,
|
source=source or "voice",
|
interaction_mode=mode or "",
|
started_at=_now_ms(),
|
status="running",
|
)
|
conn.task_session = session
|
session.save()
|
logger.bind(tag=TAG).info(
|
f"TaskSession 开始 {session.task_session_id} {app_id}/{task_id} mode={mode}"
|
)
|
return session
|
|
|
def end_session(conn: "ConnectionHandler", reason: str) -> None:
|
session = getattr(conn, "task_session", None)
|
if not session or getattr(session, "status", "") != "running":
|
return
|
session.finish(reason)
|
session.save()
|
logger.bind(tag=TAG).info(
|
f"TaskSession 结束 {session.task_session_id} status={session.status} reason={reason}"
|
)
|
_schedule_close_parent_matching(conn, session, reason)
|
|
|
def current_session(conn: "ConnectionHandler") -> Optional[TaskSession]:
|
session = getattr(conn, "task_session", None)
|
if session and getattr(session, "status", "") == "running":
|
return session
|
return None
|
|
|
def schedule_close_parent_except(
|
conn: "ConnectionHandler",
|
reason: str,
|
except_app: str = "",
|
except_task: str = "",
|
) -> None:
|
device_id = str(getattr(conn, "device_id", "") or "")
|
if not device_id:
|
return
|
payload: Dict[str, Any] = {
|
"device_id": device_id,
|
"reason": reason or "server",
|
"except_app_id": except_app or "",
|
"except_task_id": except_task or "",
|
}
|
_schedule_close_parent(payload)
|
|
|
def _schedule_close_parent_matching(
|
conn: "ConnectionHandler",
|
session: TaskSession,
|
reason: str,
|
) -> None:
|
device_id = str(getattr(conn, "device_id", "") or getattr(session, "device_id", "") or "")
|
if not device_id:
|
return
|
status = "done" if getattr(session, "status", "") == "completed" else "cancelled"
|
payload: Dict[str, Any] = {
|
"device_id": device_id,
|
"reason": reason or "",
|
"status": status,
|
"match_app_id": session.app_id or "",
|
"match_task_id": session.task_id or "",
|
}
|
_schedule_close_parent(payload)
|
|
|
def _schedule_close_parent(payload: Dict[str, Any]) -> None:
|
async def _run():
|
try:
|
from config.manage_api_client import close_parent_immediate_tasks
|
|
await close_parent_immediate_tasks(payload)
|
except Exception as exc:
|
logger.bind(tag=TAG).warning(f"关闭家长即时任务失败: {exc}")
|
|
try:
|
loop = asyncio.get_running_loop()
|
except RuntimeError:
|
return
|
loop.create_task(_run())
|
|
|
async def _upsert_task_instance(payload: Dict[str, Any]) -> bool:
|
from config.manage_api_client import upsert_task_instance
|
|
data = await upsert_task_instance(payload)
|
return data is not None
|
|
|
def _schedule_upsert(payload: Dict[str, Any], session: Optional["TaskSession"] = None) -> None:
|
async def _run():
|
try:
|
ok = await _upsert_task_instance(payload)
|
if not ok and session is not None:
|
session._skip_upsert = True
|
logger.bind(tag=TAG).warning("TaskInstance 上报失败，本会话不再重复上报")
|
except Exception as exc:
|
if session is not None:
|
session._skip_upsert = True
|
logger.bind(tag=TAG).warning(f"TaskInstance 上报失败: {exc}")
|
|
try:
|
loop = asyncio.get_running_loop()
|
except RuntimeError:
|
return
|
loop.create_task(_run())
|
