butler/gateway/item_events.py

Rama main · réplica de lectura del repositorio original.

"""App-server style Thread/Turn/Item outbound events (Codex protocol subset)."""

from __future__ import annotations

import time
import uuid
from dataclasses import dataclass, field
from typing import Any

from butler.gateway.outbound_events import OutboundEvent

SCHEMA = "butler.app-server/1"


@dataclass(frozen=True)
class ThreadItemSpec:
    thread_id: str = ""
    turn_id: str = ""
    item_id: str = ""
    item_type: str = ""
    phase: str = ""
    payload: dict[str, Any] = field(default_factory=dict)

    def to_dict(self) -> dict[str, Any]:
        return {
            "schema": SCHEMA,
            "thread_id": self.thread_id,
            "turn_id": self.turn_id,
            "item_id": self.item_id,
            "item_type": self.item_type,
            "phase": self.phase,
            "payload": dict(self.payload),
        }


def _new_item_id(prefix: str = "item") -> str:
    return f"{prefix}_{uuid.uuid4().hex[:12]}"


def thread_item_event(
    item_type: str,
    *,
    phase: str,
    thread_id: str = "",
    turn_id: str = "",
    item_id: str = "",
    payload: dict[str, Any] | None = None,
) -> OutboundEvent:
    spec = ThreadItemSpec(
        thread_id=str(thread_id or "").strip(),
        turn_id=str(turn_id or "").strip(),
        item_id=item_id or _new_item_id(item_type[:8] or "item"),
        item_type=str(item_type or "").strip(),
        phase=str(phase or "").strip(),
        payload=dict(payload or {}),
    )
    return OutboundEvent(
        kind="thread_item",
        phase=spec.phase,
        name=spec.item_type,
        detail=(spec.payload.get("detail") or "")[:200],
        extra=spec.to_dict(),
        monotonic=time.monotonic(),
    )


def context_compaction_item(
    *,
    phase: str,
    thread_id: str = "",
    turn_id: str = "",
    item_id: str = "",
    tokens_before: int = 0,
    tokens_after: int = 0,
    messages_before: int = 0,
    messages_after: int = 0,
    source: str = "context",
    remote: bool = False,
) -> OutboundEvent:
    return thread_item_event(
        "context_compaction",
        phase=phase,
        thread_id=thread_id,
        turn_id=turn_id,
        item_id=item_id,
        payload={
            "source": source[:32],
            "tokens_before": max(0, int(tokens_before)),
            "tokens_after": max(0, int(tokens_after)),
            "messages_before": max(0, int(messages_before)),
            "messages_after": max(0, int(messages_after)),
            "remote": bool(remote),
        },
    )


def emit_thread_item(event: OutboundEvent) -> None:
    """Best-effort push to global item sink (health / diagnostics)."""
    from butler.gateway.item_events_ops import record_thread_item_safe

    record_thread_item_safe(event)