fix(console): 日志页实例筛选在集群档静默失效——前端只传 instance_id，后端需 service_type+instance_id 才收窄，界面也不显示 instance_id

Start in plaita-console/backend/api/logs.py: the key selection in list_logs (around lines 125-134) and the per-entry filtering after LogEntry is built (around line 168), then stream_logs (around lines 381-407). Reproduce with FakeRedis holding worker-A and worker-B streams, then call GET /api/logs?instance_id=worker-A. Done when only worker-A rows come back, the same result as service_type+instance_id, and SSE drops worker-B pubsub messages. Add the test under plaita-console/backend/tests/console/.

优先级 P2 · 依赖：无

基线 HEAD = 65f48f2

Summary

集群档（Redis 日志后端）的日志页实例筛选是静默空操作：前端唯一能表达的查询形态是「只有 instance_id」，而后端 /logs 只在 service_type 存在时才收窄 Redis key，收窄后也从不按条目 instance_id 过滤。以下行号均为本次重读所得。

① 前端两条取数路径都只传 instance_id（SSE）

plaita-console/frontend/src/pages/Logs.tsx:32-35

    const params = new URLSearchParams()
    if (levelFilter) params.set('level', levelFilter)
    if (instanceFilter) params.set('instance_id', instanceFilter)

② 轮询路径同样只传 instance_id

plaita-console/frontend/src/pages/Logs.tsx:101-110

  const { data, isLoading, refetch } = useQuery({
    queryKey: ['logs', levelFilter, instanceFilter],
    queryFn: () =>
      api.getLogs({
        level: levelFilter || undefined,
        instance_id: instanceFilter || undefined,
        limit: 200,
      }),
    refetchInterval: isStreaming && !useSSE ? 2000 : false,
  })

③ 反证：Logs.tsx 全文没有 service_type 传参，只有渲染

$ grep -n "service_type" plaita-console/frontend/src/pages/Logs.tsx
276:function LogLine({ log }: { log: { timestamp: string; level: string; message: string; service_type?: string } }) {
292:      {log.service_type && (
294:          {log.service_type}

客户端封装本身是支持 service_type 的（plaita-console/frontend/src/services/api.ts:381-391，if (params.service_type) searchParams.set('service_type', params.service_type)），即缺口在调用点而非协议层。

④ 后端 key 收窄以 service_type 为唯一前提，缺失即退化为全量键

plaita-console/backend/api/logs.py:125-134

    # 构建 Stream key 模式
    if service_type and instance_id:
        pattern = f"plaita:logs:{service_type}:{instance_id}"
        keys = [pattern]
    elif service_type:
        pattern = f"plaita:logs:{service_type}:*"
        keys = redis.keys(pattern)
    else:
        pattern = "plaita:logs:*"
        keys = redis.keys(pattern)

⑤ 构造 LogEntry 后只比对 level，从不比对 instance_id

plaita-console/backend/api/logs.py:158-169

                    log_entry = LogEntry(
                        timestamp=entry_data.get("timestamp", entry_id),
                        level=entry_data.get("level", "INFO"),
                        service_type=entry_data.get("service_type"),
                        instance_id=entry_data.get("instance_id"),
                        message=entry_data.get("message", ""),
                        context=json.loads(entry_data.get("context", "{}"))
                    )
                    
                    # 级别筛选
                    if level and log_entry.level != level:
                        continue

⑥ SSE 侧同一套选择逻辑：仅给 instance_id 时订阅通配频道

plaita-console/backend/api/logs.py:381-391

        if service_type and instance_id:
            channel = f"plaita:logs:stream:{service_type}:{instance_id}"
        elif service_type:
            channel = f"plaita:logs:stream:{service_type}:*"
        else:
            channel = "plaita:logs:stream:*"
        
        if "*" in channel:
            pubsub.psubscribe(channel)
        else:
            pubsub.subscribe(channel)

⑦ SSE 事件侧只按 level 过滤

plaita-console/backend/api/logs.py:400-407

                        # 级别筛选
                        if level and data.get("level") != level:
                            continue
                        
                        yield {
                            "event": "log",
                            "data": json.dumps(data)
                        }

⑧ 契约漂移：同文件专属端点是按条目 instance_id 过滤的

plaita-console/backend/api/logs.py:244-248

            for entry_id, entry_data in entries:
                try:
                    entry_instance_id = entry_data.get("instance_id")
                    if entry_instance_id != instance_id:
                        continue

说明 /logs 缺省不过滤并非有意设计，而是 /logs/instance/{instance_id} 之外的另一套语义。

⑨ 发布侧通道确实带 service_type，所以实例级订阅本可命中

plaita/server/log_handler.py:55-58

        # Stream key
        self.stream_key = f"plaita:logs:{service_type}:{instance_id}"
        # Pub/Sub 通道
        self.pubsub_channel = f"{self.STREAM_CHANNEL_PREFIX}:{service_type}:{instance_id}"

条目内本身带 instance_id（plaita/server/log_handler.py:116 "instance_id": self.instance_id,），因此后端与前端两侧都存在可用的过滤依据，只是都没用。

⑩ 前端收到行后也不按 instance 过滤，且行内不渲染 instance_id

plaita-console/frontend/src/pages/Logs.tsx:114-119

  const filteredLogs = baseLogs.filter((log) => {
    if (searchText && !log.message.toLowerCase().includes(searchText.toLowerCase())) {
      return false
    }
    return true
  })

plaita-console/frontend/src/pages/Logs.tsx:276-296

function LogLine({ log }: { log: { timestamp: string; level: string; message: string; service_type?: string } }) {
  ...
      {log.service_type && (
        <span className="text-ink-faint w-24 truncate shrink-0">
          {log.service_type}
        </span>
      )}

⑪ 可达入口：Cluster 实例卡的日志弹窗有只带 instance_id 的深链

plaita-console/frontend/src/pages/Cluster.tsx:368-369

          <a 
            href={`/logs?instance_id=${instanceId}`}

而同页弹窗走的是正确端点（plaita-console/frontend/src/pages/Cluster.tsx:282 queryFn: () => api.getInstanceLogs(instanceId, { level: level || undefined, limit: 100 }),），两条路径语义不一致。

⑫ 实跑复现（基线 HEAD 上直跑）

夹具：app = FastAPI(); app.state.local_mode = False; app.include_router(logs_api.router, prefix="/api")，app.state.redis 用 FakeRedis——keys("plaita:logs:*") 返回 plaita:logs:flow_worker:worker-A 与 plaita:logs:flow_worker:worker-B 两个 key，xrevrange 分别返回 instance_id=worker-A（message A-error-1）与 instance_id=worker-B（message B-error-9）各一条，再用 starlette.testclient.TestClient 直打 /api/logs。

$ python repro_logs_instance.py
instance_id=worker-A (what the Logs page sends)  -> [["worker-B", "B-error-9"], ["worker-A", "A-error-1"]]
service_type+instance_id                         -> [["worker-A", "A-error-1"]]
GET /api/logs/instance/worker-A                -> [["worker-A", "A-error-1"], ["worker-A", "A-error-1"]]

即：前端唯一能发出的参数组合下筛选完全失效（返回了 worker-B 的行），补上 service_type 才收窄；专属端点则正确。

影响

命中场景是集群档部署（request.app.state.local_mode == False，日志落 Redis Stream 并由 plaita/server/log_handler.py 发布）下的任何人：

• 运维/开发从 Cluster 页实例卡的「在日志页面查看更多 →」点进来（Cluster.tsx:369），或在 URL 上带 ?instance_id=<某实例>；
• 日志页会显示实例筛选 chip（Logs.tsx:209-216）、不报错、条数正常，但列表里混着同 service_type 下其他实例的日志；轮询与 SSE 两种模式都一样失效。

代价是排障被误导：日志行只渲染 service_type、不渲染 instance_id（Logs.tsx:292-296），混入行与目标实例的行在视觉上不可区分，容易把 A 实例的 ERROR 当成 B 实例的故障，或误判「该实例没有报错」。加上 limit 是跨全部实例 key 拼装后截断（logs.py:179-182），目标实例的日志还可能被其他实例的日志挤出 200 条窗口，表现为「实例明明在报错但页面空白」。

不做的话，实例级排障这一入口会长期静默失效——它不崩溃、不告警，只是返回错误集合。本地单机模式不受影响：logs.py:107-123 的 is_local 分支把 instance_id 当作执行 ID 传给 list_local_logs(execution_id=...)，筛选是生效的，因此本缺陷只在集群档可达。

建议

• 后端 list_logs（plaita-console/backend/api/logs.py）：在 plaita-console/backend/api/logs.py:168 的级别过滤旁，按条目字段补实例过滤，与 /logs/instance/{instance_id} 语义对齐：

  if instance_id and log_entry.instance_id != instance_id:
      continue
  

  并把 key 择取改为「instance_id 一律不参与 key 拼接、只作条目过滤」，避免依赖调用方同时给出 service_type。
• 后端 SSE stream_logs（plaita-console/backend/api/logs.py）：仅给 instance_id 时保持 plaita:logs:stream:* 订阅，在事件侧补 if instance_id and data.get("instance_id") != instance_id: continue；service_type 与 instance_id 同时给出时保留现有的精确 subscribe。
• 前端（可选加固，非唯一修法）：Cluster.tsx:369 深链带上实例的 service_type，Logs.tsx 一路透传到 params/api.getLogs，让 SSE 走精确频道；但必须容忍只有 instance_id 的 URL（历史链接、手工输入），所以后端过滤是必需项。
• 验收条件：
  1. 新增集群分支测试（建议落在 plaita-console/backend/tests/console/），以 FakeRedis 提供 worker-A/worker-B 两条 stream，断言 GET /api/logs?instance_id=worker-A 只返回 worker-A 的行；再断言 GET /api/logs?service_type=flow_worker&instance_id=worker-A 结果与之一致。现有 plaita-console/backend/tests/console/test_local_scheduler.py:181-194 只覆盖 is_local 分支，无法回归本缺陷。
  2. SSE 断言：GET /api/logs/stream?instance_id=worker-A 收到 worker-B 的 pubsub 消息时不产出 event: log。
  3. 上述夹具重跑应变为 instance_id=worker-A (what the Logs page sends) -> [["worker-A", "A-error-1"]]，与 service_type+instance_id 一行一致。

边界

• 不处理 /logs 集群分支 redis.keys("plaita:logs:*") 全命名空间扫描的规模问题，也不做日志索引/统一查询层重构（规模问题见 #40）。
• 不改日志采集与存储协议（plaita/server/log_handler.py 的 key/通道格式是对的，不动）；不把日志行 UI 扩成通用多列展示，只要求混入行可辨识。
• 不触碰 local_mode 分支（logs.py:107-123），它在本地单机档已正确按执行 ID 过滤。
• 与 #46 不同根：那是「日志根本不产生」（服务启动器缺 logging 配置）；本单是「日志产生了，但实例筛选不生效」。
• 与 #25 不同根：那里 SSE 频道缺发布者；本单 plaita:logs:stream:{service_type}:{instance_id} 的发布侧存在（plaita/server/log_handler.py:58、:167-168），只是订阅侧未按实例过滤。
• 查重结论：gh issue list -R jeffkit/plaita --state all --limit 100 全量 33 条中无同根单，无需并入既有 issue。