[subscriber] add kvcache event subscriber

Author: Martin7-1

subscriber/.gitignore

@@ -0,0 +1,53 @@
+# === Qoder / local agent state ===
+.qoder/
+
+# === 本地环境信息 ===
+env.txt
+
+# === Python & uv ===
+__pycache__/
+*.py[cod]
+*$py.class
+*.so
+*.egg-info/
+*.egg
+dist/
+build/
+.eggs/
+
+# uv 虚拟环境与锁文件缓存
+.venv/
+.python-version
+
+# 测试与覆盖率
+.pytest_cache/
+.coverage
+htmlcov/
+.tox/
+.nox/
+
+# mypy / ruff / linting 缓存
+.mypy_cache/
+.ruff_cache/
+.dmypy.json
+
+# Jupyter Notebook
+.ipynb_checkpoints/
+
+# === PyCharm / JetBrains IDE ===
+.idea/
+*.iml
+*.iws
+*.ipr
+out/
+
+# === OS ===
+.DS_Store
+Thumbs.db
+
+# === 环境变量与密钥（切勿提交）===
+.env
+.env.*
+!.env.example
+
+*.log

subscriber/AGENTS.md

@@ -0,0 +1,146 @@
+# AGENTS.md — subscriber
+
+与 vLLM 同机部署的 Python 进程。源码包在仓库根目录下的 `subscriber/`，独立管理依赖，不依赖父项目构建系统。
+
+## 开发命令
+
+所有命令在仓库根目录（`pyproject.toml` 所在目录）下执行，使用 `uv` 管理环境。
+
+```bash
+# 初始化 / 同步依赖
+uv sync --dev
+
+# 运行测试
+uv run pytest
+uv run pytest -v                              # 详细输出
+uv run pytest tests/engine/vllm/test_incremental.py  # 单文件
+
+# Lint（ruff，line-length=88，规则集 E/F/I/UP/B）
+uv run ruff check subscriber/ tests/
+uv run ruff check --fix subscriber/ tests/    # 自动修复
+
+# 格式化
+uv run ruff format subscriber/ tests/
+uv run ruff format --check subscriber/ tests/ # 仅检查，不修改
+
+# 类型检查（mypy strict）
+uv run mypy subscriber/
+
+# 启动进程（默认连接本机 vLLM）
+uv run python -m subscriber
+
+# 指定参数启动
+uv run python -m subscriber \
+  --zmq-pub-endpoint tcp://localhost:5557 \
+  --zmq-replay-endpoint tcp://localhost:5558 \
+  --kvcm-addr 10.0.0.1:50051 \
+  --engine-type vllm
+
+# 使用配置文件
+uv run python -m subscriber --config config.yaml
+```
+
+## 质量门禁（Quality Gates）
+
+- **pre-commit 钩子（本地强制）**：`.pre-commit-config.yaml` 定义了 ruff check /
+  ruff format --check / mypy 三项检查，通过 `uv run pre-commit install` 安装到
+  `.git/hooks/pre-commit` 后在每次 commit 时强制执行。clone 仓库后必须先执行
+  一次 `uv run pre-commit install`。
+- **pytest 门禁（显式手动策略）**：本仓库当前没有 CI，也不使用 pre-push 钩子
+  跑测试。`uv run pytest` 的执行点是**手动策略**：每次代码修改完成后、以及
+  push 之前，必须手动运行 `uv run pytest` 且全部通过（见「约束」一节的
+  「修改后必须通过检查」）。测试未通过的修改视为未完成，不得提交/推送。
+
+## Protobuf 兼容代码维护
+
+生产环境固定使用 `protobuf==3.20.3`。当前 `grpc-tools` 生成的 Python
+代码会导入 `google.protobuf.runtime_version` / `_builder` 并校验较新的
+protobuf runtime，无法在生产环境加载。因此 `subscriber/proto/` 下的
+Python pb 文件必须 hand-write / 手工维护，**禁止运行
+`grpc_tools.protoc` 后直接提交生成结果**。
+
+`subscriber/proto/kv_cache_group_metadata.proto` 仍是 authoritative wire
+schema。修改 `.proto` 后：
+
+1. 手工同步 `kv_cache_group_metadata_pb2.py` 中的 descriptor
+   message/field/service 定义，继续使用兼容 protobuf 3.20.3 的
+   `descriptor_pb2` / `message_factory` 实现；不得引入
+   `runtime_version` 或 `_builder`。
+2. 手工同步 `kv_cache_group_metadata_pb2.pyi` 的字段、类型和构造函数。
+3. 只有 RPC service、method 或 request/response type 发生变化时，才
+   手工同步 `kv_cache_group_metadata_pb2_grpc.py`；仅新增 message field
+   时该文件不应产生 diff。
+4. 保持 authoritative field number、无 `package` 的 service identity 和
+   完整 RPC path 不变，并增加 descriptor field-number、raw wire parse 和
+   RPC path 测试。
+
+上述 pb 文件均提交到仓库，构建时不会自动生成。修改后至少运行对应
+proto/client 测试和 `uv run mypy subscriber/`，确认 protobuf 3.20.3
+环境可以正常 import。
+
+## 约束
+
+- **asyncio 全异步**：所有 IO 必须使用 `zmq.asyncio.Socket`，禁止在 async 函数中调用同步 ZMQ socket 方法。
+- **引擎适配器隔离**：ZMQ 订阅和 replay 逻辑封装在 `AbstractEngineAdapter` 子类中，`kv_event_loop` 不感知具体引擎实现。
+- **引擎存活探测隔离**：底层连接状态（如 ZMQ socket monitor）由 `AbstractEngineAdapter.watch_liveness()` 暴露为引擎无关事件，判死、epoch、reset 策略集中在 health coordinator。
+- **冷启动 reset 语义**：subscriber 首次连接引擎前不发送 `AllBlocksCleared`；只有曾经进入可发送代际后再断线并达到重试阈值，才发送 reset。
+- **发送门控**：向 kvcm 转发实时或 replay batch 前必须通过 `EngineHealthCoordinator.wait_ready_epoch()` 获取当前可发送 epoch。
+- **分阶段 span 计时**：从 event 到手到发送 kvcm 的每个阶段用 span 埋点，可拓展。载体 `EngineEventBatch(batches, timer)` 由 adapter yield：timer 起源于 adapter 开始处理该 event，adapter 标记自身阶段（live=`decode`、replay=`replay_fetch`），`send_incremental_events` 在同一 timer 上续标 `queue_wait`/`gate_wait`/`block_filter`/`metadata_learn`/`kvcm_send`。加新阶段只需在相应位置多一次 `timer.mark("stage")`。span 上报是 best-effort，`StageTimer.mark/spans` 与 `report()` 契约永不抛异常，绝不阻断转发主路径。
+- **Metrics 上报规范**：所有指标通过 `subscriber/metrics/` 包（`_base.py` 中的 `_dashlog_counter` / `_dashlog_gauge`）统一出口上报，自动添加 `kvcache_subscriber_` 前缀。维度信息（endpoint、status、reason 等）通过 `tags` 参数传递（对应 Prometheus labels），不要编码进 metric name。dashlog 本地不可编译时所有 report 函数为 no-op（`try: import dashlog / except ImportError` fallback）。新增或修改指标后必须同步更新 `docs/metrics.json` 指标目录。
+- **Replay 语义**：检测到 seq gap 时，先聚合所有 replay batch 一次性 yield，再 yield 当前实时消息。replay 使用 DEALER socket（非 REQ），避免 async 下严格一问一答的阻塞问题。
+- **配置优先级**：CLI 显式参数 > `--config yaml` 文件 > dataclass 默认值。
+- **Lint/类型检查禁用规则**：禁止通过内联注释（如 `# noqa`、`# type: ignore`）禁用 ruff 或 mypy 检查。自动生成的代码（如 protobuf 生成文件）应在 `pyproject.toml` 或 `ruff.toml` 中通过 `exclude` / `per-file-ignores` 排除整个文件；只有确实无法通过修改代码解决的极少数情况，才允许使用注释禁用，且必须在注释中说明原因。
+- **Docstring 规范**：通用接口/模块定义（后续要实现某个接口的 placeholder）、抽象基类的 public 方法、复杂实现逻辑必须有 docstring，说明契约、默认行为和关键不变量。简单的内部 helper 和自解释代码不需要。
+- **设计文档落盘目录**：所有 spec / 设计文档写入 `docs/superpowers/specs/`，命名格式为 `YYYY-MM-DD_slug.md`（如 `2026-07-21_engine-node-metadata-rpc.md`）。不要在项目根目录或其他位置创建 `specs/` 目录。
+- **仓库根目录不落运行时日志**：运行时日志不得写入仓库根目录；review / 报告类产物归档到 `docs/superpowers/reviews/`。
+- **修改后必须通过检查**：所有代码修改完成后，必须确保 `uv run pytest`、`uv run ruff check subscriber/ tests/` 和 `uv run mypy subscriber/` 全部通过，否则修改视为未完成。ruff/mypy 由 pre-commit 钩子在 commit 时强制；pytest 为手动门禁（见「质量门禁」一节）。
+
+## Health Integration 开发原则
+
+subscriber 与同机 DashServing "同生同死"：subscriber 未 active 时 DashServing 不导流，subscriber 异常时 DashServing 暴露健康失败。详细 spec 见 `docs/superpowers/specs/2026-07-22_subscriber-health-dashserving-integration.md`。
+
+**核心不变量（修改代码前必须理解）：**
+
+- **引擎健康只由 `GetWorkerStatus.alive` 驱动**：不用 HTTP `/readiness` 做引擎健康判断，不用 DashServing 探针反馈 engine DEAD。
+- **KVCM 是 lossy 旁路，不是门控**：KVCM 不可用不阻塞转发、不影响探针、不改变 subscriber 状态。发送失败丢弃并计数，恢复后自动续传。不要为 KVCM 加 generation/snapshot/condition 等复杂机制。
+- **seq_id 线性化**：状态上报靠单调递增 seq_id 防乱序。低 seq 的 active 不能复活已接受的 inactive。不要加 session_id 除非需要支持独立重启 subscriber。
+- **HostDown 幂等**：`AllBlocksCleared` 每个 sendable epoch 最多发一次。冷启动失败不发。
+- **同生同死探针语义**：starting → readiness 503 / liveness 200（startup grace）；active → 200/200；inactive/failed/TTL expired → 503/503。
+- **环境变量**：`DS_LLM_LAUNCH_KV_EVENT_SUBSCRIBER=1` 时 DashServing 启用 subscriber 健康检查，否则探针行为不变。
+
+**不要做的事：**
+
+- 不要为引擎重启加 metadata 重新验证（拓扑不变，Pod 替换兜底）
+- 不要轮询 DashServing readiness 来控制转发（`wait_ready_epoch()` 已是正确门控）
+- 不要在 KVCM transport 失败时改 `is_registered`（heartbeat loop 自己管重连）
+- 不要把 learn mode 加回来（已删除）
+- 不要把 transient 启动错误报为 failed（只有 protocol/unsupported 才是 fatal）
+
+## 添加新引擎适配器（参考现有 SGLang 实现）
+
+1. 参考现有 `subscriber/engine/sglang/` 包布局：`adapter.py` 实现 `AbstractEngineAdapter` 并加注册装饰器，`__init__.py` re-export 适配器类：
+
+```python
+# subscriber/engine/<engine>/adapter.py
+from subscriber.engine.base import AbstractEngineAdapter
+
+@AbstractEngineAdapter.register("sglang")
+class SGLangAdapter(AbstractEngineAdapter):
+    async def subscribe_kv_events(self):
+        ...  # 实现事件订阅逻辑
+
+    async def watch_liveness(self):
+        ...  # 实现引擎无关存活事件流
+```
+
+完整可运行示例见 `subscriber/engine/sglang/adapter.py`（placeholder 适配器，含全部必须实现的方法）。
+
+2. 启动时传入 `--engine-type sglang`，`AbstractEngineAdapter.create()` 会自动 lazy import 并实例化。
+
+无需修改 `main.py`、`kv_event_loop` 或任何 client 代码。
+
+## Stub 待接入项
+
+| 编号 | 内容 | 文件 | 阶段 |
+|---|---|---|---|
+| S-4 | subscriber 不可用恢复策略：kvcm SDK 心跳、TTL、session/checkpoint | kvcm SDK / kvcm 服务端 / subscriber | DONE（heartbeat + TTL + state reporter 已实现；session/checkpoint 见 TODO(independent-restart)） |

subscriber/README.md

@@ -0,0 +1,53 @@
+# KVCacheEventSubscriber
+
+A process that subscribes to inference engine's KV cache events and forwards them to the kvcm service.
+
+## Installation
+
+```bash
+uv sync
+```
+
+## Running
+
+```bash
+# With defaults
+uv run python -m subscriber
+
+# With CLI args
+uv run python -m subscriber --zmq-pub-endpoint tcp://localhost:5557 --kvcm-addr 10.0.0.1:50051
+
+# With config file
+uv run python -m subscriber --config config.yaml
+```
+
+## Development
+
+```bash
+# Run tests
+uv run pytest
+
+# Lint
+uv run ruff check subscriber/ tests/
+
+# Type check
+uv run mypy subscriber/
+```
+
+## Protobuf Compatibility Maintenance
+
+Production runs `protobuf==3.20.3`, and code emitted by `grpc_tools.protoc` imports
+`google.protobuf.runtime_version` / `_builder`, which cannot load on that runtime.
+The Python pb files in `subscriber/proto/` (`_pb2.py`, `_pb2_grpc.py`, `_pb2.pyi`) are
+therefore **hand-maintained** — do **not** run `grpc_tools.protoc` and commit its output.
+
+`subscriber/proto/kv_cache_group_metadata.proto` remains the authoritative wire schema.
+After changing the `.proto`, manually sync the pb files following the procedure in
+[AGENTS.md](AGENTS.md) (section "Protobuf 兼容代码维护"): keep the
+`descriptor_pb2` / `message_factory` implementation compatible with protobuf 3.20.3,
+never introduce `runtime_version` or `_builder`, and only touch `_pb2_grpc.py` when the
+RPC service/method/request/response types change.
+
+The pb files are committed to the repository and are never generated at build time.
+After any change, run the proto/client tests and `uv run mypy subscriber/` to confirm
+they import cleanly under protobuf 3.20.3.