feat(server): 日志模块——任务 / 定时与自动 / 登录 / 审计合并查询与 SSE 流，登录失败留审计

Co-Authored-By: Claude Opus 5.5 <[email protected]>
Claude-Session: https://claude.ai/code/session_01Cxwgnn2pbzQ7rXv3rq7fei

apps/server/src/modules/auth-settings/auth-routes.ts

@@ -135,7 +135,23 @@ export function registerAuthRoutes(app: ModuleApp, ctx: ServerContext) {
   });
 
   endpoint(app, D.login, async ({ body, c }) => {
-    const r = await verifyLogin(deps(), body, clientIp(c));
+    const r = await verifyLogin(deps(), body, clientIp(c)).catch(async (err) => {
+      // 日志模块的「登录」类:用户名 / 密码 / 验证码错误留一条(被限速拦下的不写,见 AuditAction 注释)
+      if (err instanceof ApiException && err.code === "unauthorized") {
+        await audit(
+          ctx,
+          undefined,
+          "admin.login_failed",
+          { type: "admin_user" },
+          {
+            username: body.username,
+            ip: clientIp(c),
+            reason: err.message,
+          },
+        );
+      }
+      throw err;
+    });
     if (r.status === "totp_required") return { status: "totp_required" as const };
     const s = await startSession(c, r.admin.id);
     const actor = adminActor(r.admin);

apps/server/src/modules/index.ts

@@ -5,6 +5,7 @@ import { browserModule } from "./browser/index.ts";
 import { cacheModule } from "./cache/index.ts";
 import { desktopModule } from "./desktop/index.ts";
 import { fileManagerModule } from "./file-manager/index.ts";
+import { logsModule } from "./logs/index.ts";
 import { mcpModule } from "./mcp/index.ts";
 import { piModule } from "./pi/index.ts";
 import { schedulesModule } from "./schedules/index.ts";
@@ -37,4 +38,5 @@ export const SERVER_MODULES: readonly ServerModule[] = [
   webSearchModule,
   cacheModule,
   systemModule,
+  logsModule,
 ];

apps/server/src/modules/logs/index.ts

@@ -0,0 +1,130 @@
+import { createRoute } from "@hono/zod-openapi";
+import { Api, apiResponse, LOG_STREAM_EVENT, type LogFilter, LogStreamQuery } from "@umt/contracts";
+import type { Context } from "hono";
+import { streamSSE } from "hono/streaming";
+import { requireScope } from "../../http/auth.ts";
+import { ApiException } from "../../http/errors.ts";
+import type { ServerModule } from "../types.ts";
+import { type LogQueries, logQueries } from "./query.ts";
+import { createLogStreamPoller, STREAM_POLL_MS } from "./stream.ts";
+
+export { decodeCursor, encodeCursor, logQueries } from "./query.ts";
+export { createLogStreamPoller } from "./stream.ts";
+
+/**
+ * 日志模块(界面第二轮 V 线,docs/runbook/logs.md):
+ * - GET /logs:任务 / 定时与自动 / 登录 / 审计合并成一份列表(查询时 union,不新建表),仅 admin;
+ * - GET /logs/stream:同一组筛选的 SSE 流,事件名 log.entry(新日志或状态变化,客户端按 id 覆盖)。
+ */
+export interface LogsModuleOptions {
+  /** 引擎 schema(测试覆盖);缺省 openworkflow */
+  engineSchema?: string;
+  /** 流的轮询间隔,缺省 2 秒 */
+  pollMs?: number;
+  keepaliveMs?: number;
+}
+
+export function createLogsModule(opts: LogsModuleOptions = {}): ServerModule {
+  return {
+    name: "logs",
+    registerRoutes(app, ctx) {
+      let q: LogQueries | null = null;
+      const queries = () => {
+        q ??= logQueries(ctx.db, { engineSchema: opts.engineSchema });
+        return q;
+      };
+
+      const list = Api.logs.list;
+      app.openapi(
+        createRoute({
+          method: list.method,
+          path: list.path,
+          summary: list.summary,
+          tags: ["logs"],
+          middleware: [requireScope(list.scope)] as never,
+          request: { query: list.query as never },
+          responses: {
+            200: { description: list.summary, content: { "application/json": { schema: apiResponse(list.response) } } },
+          },
+        }),
+        // biome-ignore lint/suspicious/noExplicitAny: 入参已由 zod-openapi 校验
+        (async (c: any) => {
+          const query = c.req.valid("query");
+          const data = list.response.parse(await queries().list(query));
+          return c.json({ ok: true, data }, 200);
+          // biome-ignore lint/suspicious/noExplicitAny: 同上
+        }) as any,
+      );
+
+      const stream = Api.logs.stream;
+      app.openapi(
+        createRoute({
+          method: stream.method,
+          path: stream.path,
+          summary: stream.summary,
+          tags: ["logs"],
+          middleware: [requireScope(stream.scope)] as never,
+          responses: {
+            200: { description: stream.summary, content: { "text/event-stream": { schema: { type: "string" } } } },
+          },
+        }),
+        // SSE 响应不走 JSON 信封
+        (async (c: Context) => {
+          const parsed = LogStreamQuery.safeParse(c.req.query());
+          if (!parsed.success) throw new ApiException("bad_request", parsed.error.issues[0]?.message ?? "筛选参数无效");
+          const { since, ...filter } = parsed.data;
+          const poller = createLogStreamPoller(queries(), filter as LogFilter, {
+            ...(since ? { since: new Date(since) } : {}),
+          });
+          const log = ctx.logger.child({ component: "logs-stream" });
+          return streamSSE(c, async (s) => {
+            let closed = false;
+            let wake: (() => void) | null = null;
+            s.onAbort(() => {
+              closed = true;
+              wake?.();
+            });
+            const keepalive = setInterval(() => {
+              void s.write(": keepalive\n\n").catch(() => {
+                closed = true;
+              });
+            }, opts.keepaliveMs ?? 15_000);
+            await s.writeSSE({ event: "ready", data: JSON.stringify({ at: new Date().toISOString() }) });
+            try {
+              while (!closed) {
+                let items: Awaited<ReturnType<typeof poller.poll>> = [];
+                try {
+                  items = await poller.poll();
+                } catch (err) {
+                  // 库暂时不可达:这一轮跳过,下一轮再查
+                  log.debug({ err }, "日志流查询失败");
+                }
+                for (const e of items) {
+                  if (closed) break;
+                  await s.writeSSE({ id: e.id, event: LOG_STREAM_EVENT, data: JSON.stringify(e) }).catch(() => {
+                    closed = true;
+                  });
+                }
+                if (closed) break;
+                await new Promise<void>((r) => {
+                  const t = setTimeout(r, opts.pollMs ?? STREAM_POLL_MS);
+                  wake = () => {
+                    clearTimeout(t);
+                    r();
+                  };
+                });
+                wake = null;
+              }
+            } finally {
+              clearInterval(keepalive);
+            }
+          });
+          // biome-ignore lint/suspicious/noExplicitAny: 同上
+        }) as any,
+      );
+    },
+    registerTasks() {},
+  };
+}
+
+export const logsModule = createLogsModule();

apps/server/src/modules/logs/query.ts

@@ -0,0 +1,311 @@
+import type { LogEntry, LogFilter, LogLevel, LogPage, LogSource, LogType } from "@umt/contracts";
+import { type DB, type Kysely, sql } from "@umt/db";
+import type { RawBuilder } from "kysely";
+import { redact } from "../../logger.ts";
+import { BUSINESS_EXPR } from "../auth-settings/engine-business.ts";
+
+// 日志聚合查询(界面第二轮 V 线):四个来源各自出同一组列,按 (at, id) 倒序各取 limit+1 条,union 后再取前 limit 条。
+// 每个来源单独 order + limit,PG 能沿各自的时间索引倒着扫,不需要把全部来源物化再排序。
+//
+// 来源与口径(docs/runbook/logs.md):
+// - engine_run  → task:引擎一次运行;时间 = 创建时间;标题 = 业务对象名(没有时为工作流名);对象 = 业务对象
+// - schedule_run → auto:定时任务运行记录(含智能体巡检);时间 = 开始时间;对象 = 定时任务
+// - agent_run(trigger=event,非子会话)→ auto:智能体后台调查;对象 = 会话
+// - audit_event → auth(admin.* / api_key.*)或 audit(其余);对象 = 审计目标
+// 例行记录(routine,缺省不列,system=1 或按对象筛选时列出):没有业务对象的引擎运行、定时任务触发且成功的运行、
+// 巡检 / 后台调查对应的引擎运行(已在 auto 里各有一条)、非巡检定时任务的成功 / 跳过运行。
+
+export type Cursor = { us: number; id: string };
+
+export function encodeCursor(c: Cursor): string {
+  return btoa(JSON.stringify([c.us, c.id])).replace(/=+$/, "");
+}
+
+export function decodeCursor(s: string | undefined | null): Cursor | null {
+  if (!s) return null;
+  try {
+    const [us, id] = JSON.parse(atob(s)) as [unknown, unknown];
+    if (typeof us === "number" && Number.isFinite(us) && typeof id === "string") return { us, id };
+  } catch {
+    // 无效游标当作第一页
+  }
+  return null;
+}
+
+/** 业务对象 kind → 日志对象类型(与审计 target_type 同名,对象筛选两边通用) */
+export const BUSINESS_OBJECT_TYPE: Record<string, string> = {
+  download: "download_task",
+  organization: "organization_record",
+  agent: "agent_run",
+  subscription: "subscription",
+  feed: "feed_source",
+  site: "site",
+  decision: "decision",
+  media: "media_entity",
+  schedule: "scheduled_job",
+};
+
+const SOURCE_TYPES: Record<LogSource, LogType[]> = {
+  engine_run: ["task"],
+  schedule_run: ["auto"],
+  agent_run: ["auto"],
+  audit_event: ["auth", "audit"],
+};
+
+const UUID_RE = "^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$";
+
+function engineSource(schema: string): RawBuilder<unknown> {
+  if (!/^[a-z_][a-z0-9_]*$/.test(schema)) throw new Error(`非法 schema 名 ${schema}`);
+  const t = sql.raw(`"${schema}".workflow_runs`);
+  const kind = sql`(biz.b->>'kind')`;
+  const ref = sql`(biz.b->>'refId')`;
+  const uuidRef = sql`(case when ${ref} ~* ${UUID_RE} then ${ref}::uuid end)`;
+  const objectType = sql`(case ${kind} ${sql.join(
+    Object.entries(BUSINESS_OBJECT_TYPE).map(([k, v]) => sql`when ${sql.lit(k)} then ${sql.lit(v)}`),
+    sql` `,
+  )} end)`;
+  const title = sql`(case ${kind}
+    when 'schedule' then (select j.name from scheduled_job j where j.id = ${ref})
+    when 'agent' then (select a.goal from agent_run a where a.id = ${uuidRef})
+    when 'download' then (select d.name from download_task d where d.id = ${uuidRef})
+    when 'organization' then (select coalesce(e.title, r.target_path, r.source_path) from organization_record r
+      left join media_entity e on e.id = r.media_entity_id where r.id = ${uuidRef})
+    when 'subscription' then (select coalesce(e.title, s.title_query) from subscription s
+      left join media_entity e on e.id = s.media_entity_id where s.id = ${uuidRef})
+    when 'feed' then (select f.name from feed_source f where f.id = ${uuidRef})
+    when 'site' then (select s.name from site s where s.id = ${uuidRef})
+    when 'decision' then (select d.title from decision d where d.id = ${uuidRef})
+    when 'media' then (select e.title from media_entity e where e.id = ${uuidRef})
+  end)`;
+  const agent = sql`(case when ${kind} = 'agent' then
+    (select jsonb_build_object('status', a.status, 'trigger', a.trigger) from agent_run a where a.id = ${uuidRef}) end)`;
+  const st = sql.raw("workflow_runs.status");
+  return sql`select
+      'engine_run:' || workflow_runs.id as id,
+      'task'::text as type,
+      'engine_run'::text as source,
+      workflow_runs.id::text as ref_id,
+      workflow_runs.created_at as at,
+      workflow_runs.finished_at as finished_at,
+      (case when workflow_runs.finished_at is not null and workflow_runs.started_at is not null
+        then extract(epoch from workflow_runs.finished_at - workflow_runs.started_at) * 1000 end)::float8 as duration_ms,
+      (case
+        when ${st} = 'failed' then 'error'
+        when ${st} = 'canceled' then 'warn'
+        when ${st} in ('completed', 'succeeded') then (case ag.a->>'status'
+          when 'failed' then 'error' when 'budget_exceeded' then 'warn' when 'cancelled' then 'warn' else 'success' end)
+        else 'running' end) as level,
+      ${st}::text as status,
+      coalesce(bt.title, workflow_runs.workflow_name) as title,
+      (case when ${st} = 'failed' then left(coalesce(workflow_runs.error->>'message', workflow_runs.error::text), 500) end)
+        as message,
+      ${objectType} as object_type,
+      ${ref} as object_id,
+      bt.title as object_label,
+      null::text as actor_type,
+      null::text as actor_id,
+      jsonb_build_object('workflowName', workflow_runs.workflow_name, 'businessKind', ${kind},
+        'businessRefId', ${ref}, 'businessTitle', bt.title, 'outcome', ag.a->>'status',
+        'agentTrigger', ag.a->>'trigger', 'attempts', workflow_runs.attempts) as extra,
+      (biz.b is null
+        or coalesce(${kind} = 'schedule' and ${st} in ('completed', 'succeeded'), false)
+        or coalesce((ag.a->>'trigger') in ('event', 'schedule'), false)) as routine,
+      concat_ws(' ', workflow_runs.workflow_name, workflow_runs.id, bt.title, ${ref},
+        left(workflow_runs.error::text, 500)) as search
+    from ${t}
+    cross join lateral (select ${BUSINESS_EXPR} as b) biz
+    cross join lateral (select ${title} as title) bt
+    cross join lateral (select ${agent} as a) ag
+    where workflow_runs.namespace_id = 'default'`;
+}
+
+const SCHEDULE_SOURCE = sql`select
+    'schedule_run:' || r.id::text as id,
+    'auto'::text as type,
+    'schedule_run'::text as source,
+    r.id::text as ref_id,
+    r.started_at as at,
+    r.finished_at as finished_at,
+    r.duration_ms::float8 as duration_ms,
+    (case
+      when r.status = 'failed' or r.agent_status = 'failed' then 'error'
+      when r.agent_status in ('budget_exceeded', 'cancelled') then 'warn'
+      when r.status = 'ok' then 'success'
+      when r.status = 'skipped' then 'info'
+      else 'running' end) as level,
+    r.status::text as status,
+    coalesce(j.name, r.job_id) as title,
+    coalesce(r.error, r.summary) as message,
+    'scheduled_job'::text as object_type,
+    r.job_id as object_id,
+    j.name as object_label,
+    r.trigger::text as actor_type,
+    null::text as actor_id,
+    jsonb_build_object('jobId', r.job_id, 'jobKind', j.kind, 'trigger', r.trigger, 'agentRunId', r.agent_run_id,
+      'agentStatus', r.agent_status, 'summary', r.summary, 'error', r.error) as extra,
+    (r.status in ('ok', 'skipped') and coalesce(j.kind::text, 'task') <> 'agent_check'
+      and r.agent_status is distinct from 'failed') as routine,
+    concat_ws(' ', r.job_id, j.name, r.summary, r.error, r.agent_run_id::text) as search
+  from scheduled_job_run r left join scheduled_job j on j.id = r.job_id`;
+
+const AGENT_SOURCE = sql`select
+    'agent_run:' || a.id::text as id,
+    'auto'::text as type,
+    'agent_run'::text as source,
+    a.id::text as ref_id,
+    a.created_at as at,
+    a.finished_at as finished_at,
+    (case when a.finished_at is not null and a.started_at is not null
+      then extract(epoch from a.finished_at - a.started_at) * 1000 end)::float8 as duration_ms,
+    (case a.status
+      when 'succeeded' then 'success' when 'failed' then 'error'
+      when 'cancelled' then 'warn' when 'budget_exceeded' then 'warn' else 'running' end) as level,
+    a.status::text as status,
+    a.goal as title,
+    coalesce(a.summary, a.error->>'message', a.spawn_reason) as message,
+    'agent_run'::text as object_type,
+    a.id::text as object_id,
+    null::text as object_label,
+    'event'::text as actor_type,
+    a.created_by as actor_id,
+    jsonb_build_object('trigger', a.trigger, 'mode', a.mode, 'toolCalls', a.tool_call_count,
+      'spawnReason', a.spawn_reason, 'model', a.model) as extra,
+    false as routine,
+    concat_ws(' ', a.goal, a.summary, a.id::text, left(a.error::text, 500)) as search
+  from agent_run a where a.trigger = 'event' and a.parent_run_id is null`;
+
+const AUDIT_SOURCE = sql`select
+    'audit_event:' || e.id::text as id,
+    (case when e.action like 'admin.%' or e.action like 'api_key.%' then 'auth' else 'audit' end) as type,
+    'audit_event'::text as source,
+    e.id::text as ref_id,
+    e.at as at,
+    null::timestamptz as finished_at,
+    null::float8 as duration_ms,
+    (case
+      when e.action in ('admin.login_failed', 'capability.denied') then 'warn'
+      when e.action in ('admin.login_success', 'admin.passkey_login', 'admin.setup') then 'success'
+      else 'info' end) as level,
+    e.action as status,
+    e.action as title,
+    (case when e.action like 'admin.%' then concat_ws(' · ', e.payload->>'username', e.payload->>'ip')
+      else nullif(concat_ws(' · ', e.payload->>'reason', e.payload->>'error'), '') end) as message,
+    e.target_type as object_type,
+    e.target_id as object_id,
+    null::text as object_label,
+    e.actor_type as actor_type,
+    e.actor_id as actor_id,
+    jsonb_build_object('action', e.action, 'traceId', e.trace_id, 'payload', e.payload) as extra,
+    false as routine,
+    concat_ws(' ', e.action, e.target_type, e.target_id, e.actor_id, left(e.payload::text, 2000)) as search
+  from audit_event e`;
+
+type Row = {
+  id: string;
+  type: LogType;
+  source: LogSource;
+  ref_id: string;
+  at: Date;
+  finished_at: Date | null;
+  duration_ms: number | null;
+  level: LogLevel;
+  status: string | null;
+  title: string | null;
+  message: string | null;
+  object_type: string | null;
+  object_id: string | null;
+  object_label: string | null;
+  actor_type: string | null;
+  actor_id: string | null;
+  extra: Record<string, unknown> | null;
+  cursor_us: string | number;
+};
+
+export interface LogQueryInput extends LogFilter {
+  cursor?: string | null;
+  limit: number;
+}
+
+const escapeLike = (s: string) => s.replace(/[\\%_]/g, (c) => `\\${c}`);
+
+export interface LogQueries {
+  list(q: LogQueryInput): Promise<LogPage>;
+}
+
+export function logQueries(db: Kysely<DB>, opts: { engineSchema?: string } = {}): LogQueries {
+  const engineSchema = opts.engineSchema ?? "openworkflow";
+  let engineReady = false;
+  const engineExists = async () => {
+    if (engineReady) return true;
+    const r = await sql<{
+      t: string | null;
+    }>`select to_regclass(${`"${engineSchema}".workflow_runs`})::text as t`.execute(db);
+    engineReady = !!r.rows[0]?.t;
+    return engineReady;
+  };
+
+  return {
+    async list(q) {
+      const types = q.type?.length ? new Set(q.type) : null;
+      const sources: [LogSource, RawBuilder<unknown>][] = [];
+      const want = (s: LogSource) => !types || SOURCE_TYPES[s].some((t) => types.has(t));
+      if (want("engine_run") && (await engineExists())) sources.push(["engine_run", engineSource(engineSchema)]);
+      if (want("schedule_run")) sources.push(["schedule_run", SCHEDULE_SOURCE]);
+      if (want("agent_run")) sources.push(["agent_run", AGENT_SOURCE]);
+      if (want("audit_event")) sources.push(["audit_event", AUDIT_SOURCE]);
+      if (!sources.length) return { items: [], nextCursor: null };
+
+      const conds: RawBuilder<unknown>[] = [];
+      if (types) conds.push(sql`s.type = any(${[...types]}::text[])`);
+      if (q.level?.length) conds.push(sql`s.level = any(${q.level}::text[])`);
+      if (q.from) conds.push(sql`s.at >= ${new Date(q.from)}`);
+      if (q.to) conds.push(sql`s.at < ${new Date(q.to)}`);
+      if (q.q) conds.push(sql`s.search ilike ${`%${escapeLike(q.q)}%`}`);
+      if (q.objectType) conds.push(sql`s.object_type = ${q.objectType}`);
+      if (q.objectId) conds.push(sql`s.object_id = ${q.objectId}`);
+      if (q.system !== "1" && !q.objectId) conds.push(sql`not s.routine`);
+      const cur = decodeCursor(q.cursor);
+      if (cur) {
+        conds.push(sql`(s.at, s.id) < (timestamptz 'epoch' + ${cur.us} * interval '1 microsecond', ${cur.id})`);
+      }
+      const where = conds.length ? sql`where ${sql.join(conds, sql` and `)}` : sql``;
+      const n = q.limit + 1;
+      const parts = sources.map(
+        ([, src]) => sql`(select s.* from (${src}) s ${where} order by s.at desc, s.id desc limit ${n})`,
+      );
+      const rows = await sql<Row>`select u.*, (extract(epoch from u.at) * 1000000)::bigint as cursor_us
+        from (${sql.join(parts, sql` union all `)}) u
+        order by u.at desc, u.id desc limit ${n}`.execute(db);
+      const items = rows.rows.slice(0, q.limit);
+      const last = items[items.length - 1];
+      return {
+        items: items.map(toEntry),
+        nextCursor:
+          rows.rows.length > q.limit && last ? encodeCursor({ us: Number(last.cursor_us), id: last.id }) : null,
+      };
+    },
+  };
+}
+
+const iso = (d: Date | string | null) => (d == null ? null : (d instanceof Date ? d : new Date(d)).toISOString());
+
+export function toEntry(r: Row): LogEntry {
+  const extra = (r.extra ?? {}) as Record<string, unknown>;
+  return {
+    id: r.id,
+    type: r.type,
+    source: r.source,
+    refId: r.ref_id,
+    at: iso(r.at)!,
+    finishedAt: iso(r.finished_at),
+    durationMs: r.duration_ms == null ? null : Math.max(0, Math.round(Number(r.duration_ms))),
+    level: r.level,
+    status: r.status,
+    title: r.title || r.ref_id,
+    message: r.message || null,
+    object: r.object_type ? { type: r.object_type, id: r.object_id, label: r.object_label } : null,
+    actor: r.actor_type ? { type: r.actor_type, id: r.actor_id } : null,
+    // 审计负载可能带敏感字段:与审计接口一样脱敏后再给
+    extra: r.source === "audit_event" ? (redact(extra) as Record<string, unknown>) : extra,
+  };
+}

apps/server/src/modules/logs/stream.ts

@@ -0,0 +1,56 @@
+import type { LogEntry, LogFilter } from "@umt/contracts";
+import type { LogQueries } from "./query.ts";
+
+// 日志流(GET /logs/stream):按同一组筛选每隔 pollMs 查一次「回看窗口」内的最新日志,
+// 没推过的推出去(新日志),推过但级别 / 状态 / 摘要变了的再推一次(客户端按 id 覆盖,如任务从进行中变成完成)。
+// 回看窗口 = 最近 lookbackMs(首次从 since 起),避免事务晚提交的记录(时间早于已推的最新一条)被漏掉。
+
+export const STREAM_POLL_MS = 2_000;
+export const STREAM_LOOKBACK_MS = 10 * 60_000;
+const STREAM_BATCH = 200;
+
+const signature = (e: LogEntry) => `${e.level}|${e.status ?? ""}|${e.message ?? ""}|${e.title}|${e.finishedAt ?? ""}`;
+
+export interface LogStreamPoller {
+  /** 查一次,返回需要推送的条目(按时间正序,新的在后) */
+  poll(): Promise<LogEntry[]>;
+}
+
+export function createLogStreamPoller(
+  queries: LogQueries,
+  filter: LogFilter,
+  o: { since?: Date; now?: () => number; lookbackMs?: number } = {},
+): LogStreamPoller {
+  const now = o.now ?? Date.now;
+  const lookbackMs = o.lookbackMs ?? STREAM_LOOKBACK_MS;
+  const since = (o.since ?? new Date(now())).getTime();
+  /** id → 上次推送时的签名;窗口外的定期丢掉 */
+  const seen = new Map<string, { sig: string; at: number }>();
+  let first = true;
+
+  return {
+    async poll() {
+      // 首次从 since 起(客户端列表最新一条之后的都要补上),之后只看回看窗口
+      const windowStart = first ? Math.min(since, now() - lookbackMs) : now() - lookbackMs;
+      const from = filter.from ? Math.max(Date.parse(filter.from), windowStart) : windowStart;
+      const page = await queries.list({ ...filter, from: new Date(from).toISOString(), limit: STREAM_BATCH });
+      const out: LogEntry[] = [];
+      for (const e of page.items) {
+        const at = Date.parse(e.at);
+        const sig = signature(e);
+        const prev = seen.get(e.id);
+        seen.set(e.id, { sig, at });
+        // 首次查询:连接之前(since 之前)的日志客户端已有,只记签名不推
+        if (!prev) {
+          if (first && at <= since) continue;
+          out.push(e);
+        } else if (prev.sig !== sig) out.push(e);
+      }
+      first = false;
+      // 窗口外的不会再查到,签名丢掉
+      const cutoff = now() - lookbackMs;
+      for (const [id, v] of seen) if (v.at < cutoff) seen.delete(id);
+      return out.reverse();
+    },
+  };
+}