122 lines
4.7 KiB
Python
122 lines
4.7 KiB
Python
"""一炁事件模型与事件存储抽象。
|
||
|
||
字段对齐 docs/architecture/data-and-security-architecture.md §4。当前阶段不引入数据库（Alembic 迁移为独立工程），
|
||
EventStore 用 Protocol 抽象、提供内存实现；内存实现额外支持按 run 订阅，供 SSE 实时推送。
|
||
未来接入 SQLAlchemy + Alembic 时替换实现即可，调用方无需改动。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import fnmatch
|
||
import uuid
|
||
from datetime import UTC, datetime
|
||
from typing import Any, Literal, Protocol
|
||
|
||
from pydantic import BaseModel, ConfigDict, Field
|
||
|
||
Visibility = Literal["public", "internal", "audit"]
|
||
Sensitivity = Literal["none", "low", "medium", "high"]
|
||
|
||
|
||
def new_event_id() -> str:
|
||
"""生成带 ``event_`` 前缀的事件 ID（契约字典 ID 前缀表）。"""
|
||
return f"event_{uuid.uuid4().hex}"
|
||
|
||
|
||
def now_iso() -> str:
|
||
return datetime.now(UTC).isoformat()
|
||
|
||
|
||
class Event(BaseModel):
|
||
"""一炁事件。"""
|
||
|
||
model_config = ConfigDict(extra="forbid")
|
||
|
||
id: str = Field(default_factory=new_event_id)
|
||
type: str
|
||
source: str
|
||
payload: dict[str, Any] = Field(default_factory=dict)
|
||
visibility: Visibility = "internal"
|
||
sensitivity: Sensitivity = "low"
|
||
run_id: str | None = None
|
||
character_id: str | None = None
|
||
trace_id: str | None = None
|
||
parent_id: str | None = None
|
||
sequence: int | None = None
|
||
evidence_ids: list[str] = Field(default_factory=list)
|
||
created_at: str = Field(default_factory=now_iso)
|
||
|
||
|
||
class EventStore(Protocol):
|
||
"""事件存储接口：append 写入、list_by_run 检索、subscribe/unsubscribe 订阅实时事件。"""
|
||
|
||
async def append(self, event: Event, **kwargs: Any) -> Event: ...
|
||
|
||
async def list_by_run(self, run_id: str) -> list[Event]: ...
|
||
|
||
def subscribe(self, run_id: str) -> asyncio.Queue[Event]: ...
|
||
|
||
def unsubscribe(self, run_id: str, queue: asyncio.Queue[Event]) -> None: ...
|
||
|
||
def subscribe_by_pattern(self, pattern: str) -> asyncio.Queue[Event]:
|
||
"""按 event.type 的 glob pattern 订阅（如 'time.*' / '*' / 'time.reunion'）。
|
||
|
||
与 subscribe(run_id) 对称：append 时用 fnmatch 匹配命中 pattern 的订阅者并 put_nowait。
|
||
"""
|
||
...
|
||
|
||
def unsubscribe_by_pattern(self, pattern: str, queue: asyncio.Queue[Event]) -> None:
|
||
"""取消 pattern 订阅（与 subscribe_by_pattern 对称）。"""
|
||
...
|
||
|
||
|
||
class InMemoryEventStore:
|
||
"""内存事件存储，供本地与测试使用。支持按 run 订阅，append 时通知订阅者。"""
|
||
|
||
def __init__(self) -> None:
|
||
self._events: list[Event] = []
|
||
self._subscribers: dict[str, list[asyncio.Queue[Event]]] = {}
|
||
self._pattern_subscribers: dict[str, list[asyncio.Queue[Event]]] = {}
|
||
|
||
async def append(self, event: Event, **kwargs: Any) -> Event: # noqa: ARG002
|
||
# 为同一 run 的事件分配单调递增的 sequence（SSE 排序与断线对齐用，API §5.7）
|
||
if event.run_id and event.sequence is None:
|
||
count = sum(1 for e in self._events if e.run_id == event.run_id)
|
||
event = event.model_copy(update={"sequence": count + 1})
|
||
self._events.append(event)
|
||
for queue in self._subscribers.get(event.run_id or "", []):
|
||
queue.put_nowait(event)
|
||
# pattern 订阅：fnmatch 匹配 event.type，命中则 put_nowait（与 run_id 订阅对称的内存 pub/sub）
|
||
for pattern, queues in self._pattern_subscribers.items():
|
||
if fnmatch.fnmatch(event.type, pattern):
|
||
for queue in queues:
|
||
queue.put_nowait(event)
|
||
return event
|
||
|
||
async def list_by_run(self, run_id: str) -> list[Event]:
|
||
return [e for e in self._events if e.run_id == run_id]
|
||
|
||
def subscribe(self, run_id: str) -> asyncio.Queue[Event]:
|
||
queue: asyncio.Queue[Event] = asyncio.Queue()
|
||
self._subscribers.setdefault(run_id, []).append(queue)
|
||
return queue
|
||
|
||
def unsubscribe(self, run_id: str, queue: asyncio.Queue[Event]) -> None:
|
||
subscribers = self._subscribers.get(run_id)
|
||
if subscribers is not None:
|
||
self._subscribers[run_id] = [q for q in subscribers if q is not queue]
|
||
|
||
def subscribe_by_pattern(self, pattern: str) -> asyncio.Queue[Event]:
|
||
queue: asyncio.Queue[Event] = asyncio.Queue()
|
||
self._pattern_subscribers.setdefault(pattern, []).append(queue)
|
||
return queue
|
||
|
||
def unsubscribe_by_pattern(self, pattern: str, queue: asyncio.Queue[Event]) -> None:
|
||
subscribers = self._pattern_subscribers.get(pattern)
|
||
if subscribers is not None:
|
||
self._pattern_subscribers[pattern] = [q for q in subscribers if q is not queue]
|
||
|
||
def all(self) -> list[Event]:
|
||
return list(self._events)
|