engine/plugins/logging.py

"""턴 로깅 Plugin — 횡종단 관심사 (ADK Plugin).

이벤트 스트림에서 유저 입력/가드 판정/캐릭터 응답/디렉터 도구 호출을
포켓 세션 단위로 기록한다. 연구 데이터 겸용 — 삭제 금지.
"""

from __future__ import annotations

from typing import Optional

from google.adk.agents.invocation_context import InvocationContext
from google.adk.events import Event
from google.adk.plugins import BasePlugin

from engine.repositories.playlog import PlaylogRepository


class TurnLoggingPlugin(BasePlugin):
    def __init__(self, playlog: PlaylogRepository | None = None):
        super().__init__(name="turn_logging")
        self.playlog = playlog or PlaylogRepository()

    async def on_user_message_callback(self, *, invocation_context: InvocationContext,
                                       user_message) -> None:
        text = ""
        if user_message and user_message.parts:
            text = " ".join(p.text or "" for p in user_message.parts)
        self.playlog.append({
            "kind": "user_input",
            "session": invocation_context.session.id,
            "card": invocation_context.session.state.get("pocket_card_id"),
            "text": text,
        })
        return None

    async def on_event_callback(self, *, invocation_context: InvocationContext,
                                event: Event) -> Optional[Event]:
        state = invocation_context.session.state
        record = {
            "kind": "event",
            "session": invocation_context.session.id,
            "card": state.get("pocket_card_id"),
            "author": event.author,
        }
        if event.content and event.content.parts:
            texts = [p.text for p in event.content.parts if p.text]
            calls = [{"tool": p.function_call.name, "args": dict(p.function_call.args or {})}
                     for p in event.content.parts if p.function_call]
            if texts:
                record["text"] = "".join(texts)[:500]
            if calls:
                record["tool_calls"] = calls
        if event.actions and event.actions.state_delta:
            delta = event.actions.state_delta
            record["state_delta_keys"] = sorted(delta.keys())
            if "guard_flag" in delta:
                record["guard_flag"] = delta["guard_flag"]
        self.playlog.append(record)
        return None

    async def after_run_callback(self, *, invocation_context: InvocationContext) -> None:
        s = invocation_context.session.state
        self.playlog.append({
            "kind": "turn_summary",
            "session": invocation_context.session.id,
            "card": s.get("pocket_card_id"),
            "hidden_trust": s.get("hidden_trust"),
            "hidden_doubt": s.get("hidden_doubt"),
            "beats_hit": s.get("pocket_beats_hit"),
            "converged": s.get("pocket_converged"),
        })