merge: fix/logs-perf

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

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

@@ -5,7 +5,8 @@ import { redact } from "../../logger.ts";
 import { BUSINESS_EXPR } from "../auth-settings/engine-business.ts";
 
 // 日志聚合查询(界面第二轮 V 线):四个来源各自出同一组列,按 (at, id) 倒序各取 limit+1 条,union 后再取前 limit 条。
-// 每个来源单独 order + limit,PG 能沿各自的时间索引倒着扫,不需要把全部来源物化再排序。
+// 每个来源单独 order + limit,PG 能沿各自的时间索引倒着扫,不需要把全部来源物化再排序;
+// 筛选只用原始列,标题等昂贵列只对取出的行算(见 SourceBuilder)。
 //
 // 来源与口径(docs/runbook/logs.md):
 // - engine_run  → task:引擎一次运行;时间 = 创建时间;标题 = 业务对象名(没有时为工作流名);对象 = 业务对象
@@ -54,11 +55,73 @@ const SOURCE_TYPES: Record<LogSource, LogType[]> = {
 
 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> {
+/** 关键词命中的业务对象每类最多取这么多个(只影响「按业务标题搜任务」,标题极泛的词会漏掉更早的对象) */
+export const KEYWORD_OBJECT_CAP = 2000;
+
+/** 各来源共用的已解析筛选 */
+interface Filter {
+  types: Set<LogType> | null;
+  levels: LogLevel[] | null;
+  from: Date | null;
+  to: Date | null;
+  /** ilike 模式(已转义,带 %) */
+  like: string | null;
+  /** 关键词命中的业务对象 "kind:id"(只给引擎来源用) */
+  matched: string[];
+  objectType: string | null;
+  objectId: string | null;
+  hideRoutine: boolean;
+  cursor: Cursor | null;
+}
+
+/**
+ * 每个来源分两段:
+ * 1. 只用原始列(及一次性算好的业务引用)筛选,`order by 时间 desc, 日志 id desc limit n`,PG 沿时间索引倒着扫、凑够 n 条就停;
+ * 2. 只对这 n 行算标题、对象标签、extra 等昂贵列。
+ * 以前是先构造带派生列(标题子查询、search 拼接)的子查询再在外层筛选,带筛选时要对全部引擎运行算一遍派生列。
+ */
+type SourceBuilder = (f: Filter, n: number) => RawBuilder<unknown>;
+
+/** 时间范围与游标条件;at、logId 是该来源的时间列与日志 id 表达式 */
+function commonConds(f: Filter, at: RawBuilder<unknown>, logId: RawBuilder<unknown>): RawBuilder<unknown>[] {
+  const out: RawBuilder<unknown>[] = [];
+  if (f.from) out.push(sql`${at} >= ${f.from}`);
+  if (f.to) out.push(sql`${at} < ${f.to}`);
+  if (f.cursor) {
+    const ts = sql`(timestamptz 'epoch' + ${f.cursor.us} * interval '1 microsecond')`;
+    // 拆成 at <= ts(可走索引)+ 同一时刻再比日志 id,与 (at, id) < (ts, id) 等价
+    out.push(sql`${at} <= ${ts} and (${at} < ${ts} or ${logId} < ${f.cursor.id})`);
+  }
+  return out;
+}
+
+const andAll = (conds: RawBuilder<unknown>[]) => (conds.length ? sql.join(conds, sql` and `) : sql`true`);
+const levelIn = (expr: RawBuilder<unknown>, f: Filter) => (f.levels ? [sql`${expr} = any(${f.levels}::text[])`] : []);
+
+function engineSource(schema: string): SourceBuilder {
   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 st = sql.raw("workflow_runs.status");
-  const ref = sql`(biz.b->>'refId')`;
+  const logId = sql`('engine_run:' || workflow_runs.id)`;
+  // 第一段:业务引用 BUSINESS_EXPR 每行只算一次(offset 0 阻止子查询上拉,否则每处引用都会把整个表达式展开重算)。
+  // 智能体结果(级别、例行要用)的 lateral 也加 offset 0:保持逐行嵌套循环,凑够 n 条就停;
+  // 不加时规划器会改成与 agent_run 的哈希连接,要先扫完全部运行。
+  const kind1 = sql`(biz.b->>'kind')`;
+  const ref1 = sql`(biz.b->>'refId')`;
+  const uuid1 = sql`(case when ${ref1} ~* ${UUID_RE} then ${ref1}::uuid end)`;
+  const level = sql`(case
+      when ${st} = 'failed' then 'error'
+      when ${st} = 'canceled' then 'warn'
+      when ${st} in ('completed', 'succeeded') then (case ag.status
+        when 'failed' then 'error' when 'budget_exceeded' then 'warn' when 'cancelled' then 'warn' else 'success' end)
+      else 'running' end)`;
+  const routine = sql`(biz.b is null
+      or coalesce(${kind1} = 'schedule' and ${st} in ('completed', 'succeeded'), false)
+      or coalesce(ag.trigger in ('event', 'schedule'), false))`;
+
+  // 第二段:p.kind / p.ref 是第一段带出来的列
+  const kind = sql.raw("p.kind");
+  const ref = sql.raw("p.ref");
   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)}`),
@@ -77,62 +140,94 @@ function engineSource(schema: string): RawBuilder<unknown> {
     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)`;
+  return (f, n) => {
-  const st = sql.raw("workflow_runs.status");
+    const conds: RawBuilder<unknown>[] = [
+      sql`workflow_runs.namespace_id = 'default'`,
+      ...commonConds(f, sql`workflow_runs.created_at`, logId),
+    ];
+    conds.push(...levelIn(level, f));
+    if (f.hideRoutine) conds.push(sql`not ${routine}`);
+    if (f.like) {
+      // 标题命中 → 业务引用在命中集合里;其余按原始列(工作流名、运行 id、业务对象 id、错误)
+      conds.push(sql`(${f.matched.length ? sql`(${kind1} || ':' || ${ref1}) = any(${f.matched}::text[]) or ` : sql``}
+        workflow_runs.workflow_name ilike ${f.like} or workflow_runs.id ilike ${f.like} or ${ref1} ilike ${f.like}
+        or left(workflow_runs.error::text, 500) ilike ${f.like})`);
+    }
+    if (f.objectType) {
+      const kinds = Object.entries(BUSINESS_OBJECT_TYPE)
+        .filter(([, v]) => v === f.objectType)
+        .map(([k]) => k);
+      conds.push(kinds.length ? sql`${kind1} = any(${kinds}::text[])` : sql`false`);
+    }
+    if (f.objectId) conds.push(sql`${ref1} = ${f.objectId}`);
     return sql`select
-      'engine_run:' || workflow_runs.id as id,
+        'engine_run:' || p.id as id,
         'task'::text as type,
         'engine_run'::text as source,
-      workflow_runs.id::text as ref_id,
+        p.id::text as ref_id,
-      workflow_runs.created_at as at,
+        p.created_at as at,
-      workflow_runs.finished_at as finished_at,
+        p.finished_at as finished_at,
-      (case when workflow_runs.finished_at is not null and workflow_runs.started_at is not null
+        (case when p.finished_at is not null and p.started_at is not null
-        then extract(epoch from workflow_runs.finished_at - workflow_runs.started_at) * 1000 end)::float8 as duration_ms,
+          then extract(epoch from p.finished_at - p.started_at) * 1000 end)::float8 as duration_ms,
-      (case
+        p.level,
-        when ${st} = 'failed' then 'error'
+        p.status::text as status,
-        when ${st} = 'canceled' then 'warn'
+        coalesce(bt.title, p.workflow_name) as title,
-        when ${st} in ('completed', 'succeeded') then (case ag.a->>'status'
+        (case when p.status = 'failed' then left(coalesce(p.error->>'message', p.error::text), 500) end) as message,
-          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},
+        jsonb_build_object('workflowName', p.workflow_name, 'businessKind', ${kind},
-        'businessRefId', ${ref}, 'businessTitle', bt.title, 'outcome', ag.a->>'status',
+          'businessRefId', ${ref}, 'businessTitle', bt.title, 'outcome', p.agent_status,
-        'agentTrigger', ag.a->>'trigger', 'attempts', workflow_runs.attempts) as extra,
+          'agentTrigger', p.agent_trigger, 'attempts', p.attempts) as extra
-      (biz.b is null
+      from (
-        or coalesce(${kind} = 'schedule' and ${st} in ('completed', 'succeeded'), false)
+        select workflow_runs.id, workflow_runs.created_at, workflow_runs.started_at, workflow_runs.finished_at,
-        or coalesce((ag.a->>'trigger') in ('event', 'schedule'), false)) as routine,
+          workflow_runs.status, workflow_runs.workflow_name, workflow_runs.error, workflow_runs.attempts,
-      concat_ws(' ', workflow_runs.workflow_name, workflow_runs.id, bt.title, ${ref},
+          ${kind1} as kind, ${ref1} as ref, ag.status as agent_status, ag.trigger as agent_trigger, ${level} as level
-        left(workflow_runs.error::text, 500)) as search
         from ${t}
-    cross join lateral (select ${BUSINESS_EXPR} as b) biz
+        cross join lateral (select ${BUSINESS_EXPR} as b offset 0) biz
-    cross join lateral (select ${title} as title) bt
+        left join lateral (select a.status::text as status, a.trigger::text as trigger from agent_run a
-    cross join lateral (select ${agent} as a) ag
+          where ${kind1} = 'agent' and a.id = ${uuid1} offset 0) ag on true
-    where workflow_runs.namespace_id = 'default'`;
+        where ${andAll(conds)}
+        order by workflow_runs.created_at desc, ${logId} desc
+        limit ${n}
+      ) p
+      cross join lateral (select ${title} as title offset 0) bt`;
+  };
 }
 
-const SCHEDULE_SOURCE = sql`select
+// 定时任务运行、智能体后台调查、审计:行数少、列都在本行(定时任务只多一个小表 join),两段合成一条查询:
-    'schedule_run:' || r.id::text as id,
+// 级别 / 类型 / 例行 / 关键词都是本行列上的表达式,直接放进 where,再按时间倒序取 n 条。
+
+const scheduleSource: SourceBuilder = (f, n) => {
+  const logId = sql`('schedule_run:' || r.id::text)`;
+  const level = sql`(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)`;
+  const conds = [...commonConds(f, sql`r.started_at`, logId), ...levelIn(level, f)];
+  if (f.hideRoutine) {
+    conds.push(sql`not (r.status in ('ok', 'skipped') and coalesce(j.kind::text, 'task') <> 'agent_check'
+      and r.agent_status is distinct from 'failed')`);
+  }
+  if (f.like) {
+    conds.push(sql`concat_ws(' ', r.job_id, j.name, r.summary, r.error, r.agent_run_id::text) ilike ${f.like}`);
+  }
+  if (f.objectType && f.objectType !== "scheduled_job") conds.push(sql`false`);
+  if (f.objectId) conds.push(sql`r.job_id = ${f.objectId}`);
+  return sql`select
+      ${logId} 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
+      ${level} as level,
-      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,
@@ -142,14 +237,30 @@ const SCHEDULE_SOURCE = sql`select
       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,
+        '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'
+    from scheduled_job_run r left join scheduled_job j on j.id = r.job_id
-      and r.agent_status is distinct from 'failed') as routine,
+    where ${andAll(conds)}
-    concat_ws(' ', r.job_id, j.name, r.summary, r.error, r.agent_run_id::text) as search
+    order by r.started_at desc, ${logId} desc
-  from scheduled_job_run r left join scheduled_job j on j.id = r.job_id`;
+    limit ${n}`;
+};
 
-const AGENT_SOURCE = sql`select
+const agentSource: SourceBuilder = (f, n) => {
-    'agent_run:' || a.id::text as id,
+  const logId = sql`('agent_run:' || a.id::text)`;
+  const level = sql`(case a.status
+      when 'succeeded' then 'success' when 'failed' then 'error'
+      when 'cancelled' then 'warn' when 'budget_exceeded' then 'warn' else 'running' end)`;
+  const conds = [
+    sql`a.trigger = 'event' and a.parent_run_id is null`,
+    ...commonConds(f, sql`a.created_at`, logId),
+    ...levelIn(level, f),
+  ];
+  if (f.like) {
+    conds.push(sql`concat_ws(' ', a.goal, a.summary, a.id::text, left(a.error::text, 500)) ilike ${f.like}`);
+  }
+  if (f.objectType && f.objectType !== "agent_run") conds.push(sql`false`);
+  if (f.objectId) conds.push(sql`a.id::text = ${f.objectId}`);
+  return sql`select
+      ${logId} as id,
       'auto'::text as type,
       'agent_run'::text as source,
       a.id::text as ref_id,
@@ -157,9 +268,7 @@ const AGENT_SOURCE = sql`select
       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
+      ${level} as level,
-      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,
@@ -169,23 +278,37 @@ const AGENT_SOURCE = sql`select
       '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,
+        'spawnReason', a.spawn_reason, 'model', a.model) as extra
-    false as routine,
+    from agent_run a
-    concat_ws(' ', a.goal, a.summary, a.id::text, left(a.error::text, 500)) as search
+    where ${andAll(conds)}
-  from agent_run a where a.trigger = 'event' and a.parent_run_id is null`;
+    order by a.created_at desc, ${logId} desc
+    limit ${n}`;
+};
 
-const AUDIT_SOURCE = sql`select
+const auditSource: SourceBuilder = (f, n) => {
-    'audit_event:' || e.id::text as id,
+  const logId = sql`('audit_event:' || e.id::text)`;
-    (case when e.action like 'admin.%' or e.action like 'api_key.%' then 'auth' else 'audit' end) as type,
+  const type = sql`(case when e.action like 'admin.%' or e.action like 'api_key.%' then 'auth' else 'audit' end)`;
+  const level = sql`(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)`;
+  const conds = [...commonConds(f, sql`e.at`, logId), ...levelIn(level, f)];
+  if (f.types) conds.push(sql`${type} = any(${[...f.types]}::text[])`);
+  if (f.like) {
+    conds.push(sql`concat_ws(' ', e.action, e.target_type, e.target_id, e.actor_id, left(e.payload::text, 2000))
+      ilike ${f.like}`);
+  }
+  if (f.objectType) conds.push(sql`e.target_type = ${f.objectType}`);
+  if (f.objectId) conds.push(sql`e.target_id = ${f.objectId}`);
+  return sql`select
+      ${logId} as id,
+      ${type} 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
+      ${level} as level,
-      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')
@@ -195,10 +318,33 @@ const AUDIT_SOURCE = sql`select
       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,
+      jsonb_build_object('action', e.action, 'traceId', e.trace_id, 'payload', e.payload) as extra
-    false as routine,
+    from audit_event e
-    concat_ws(' ', e.action, e.target_type, e.target_id, e.actor_id, left(e.payload::text, 2000)) as search
+    where ${andAll(conds)}
-  from audit_event e`;
+    order by e.at desc, ${logId} desc
+    limit ${n}`;
+};
+
+/**
+ * 关键词命中的业务对象("kind:id"),口径与引擎运行的标题一致(标题 = 业务对象名);每类最多 KEYWORD_OBJECT_CAP 个。
+ * 这些表都不大,先查出 id 集合,引擎来源再按业务引用 ∈ 集合筛选,不用对每条运行算标题。
+ */
+async function keywordObjects(db: Kysely<DB>, like: string): Promise<string[]> {
+  const cap = KEYWORD_OBJECT_CAP;
+  const r = await sql<{ k: string }>`
+    (select 'schedule:' || j.id as k from scheduled_job j where j.name ilike ${like} limit ${cap})
+    union all (select 'agent:' || a.id from agent_run a where a.goal ilike ${like} limit ${cap})
+    union all (select 'download:' || d.id from download_task d where d.name ilike ${like} limit ${cap})
+    union all (select 'organization:' || r.id from organization_record r left join media_entity e on e.id = r.media_entity_id
+      where coalesce(e.title, r.target_path, r.source_path) ilike ${like} limit ${cap})
+    union all (select 'subscription:' || s.id from subscription s left join media_entity e on e.id = s.media_entity_id
+      where coalesce(e.title, s.title_query) ilike ${like} limit ${cap})
+    union all (select 'feed:' || f.id from feed_source f where f.name ilike ${like} limit ${cap})
+    union all (select 'site:' || s.id from site s where s.name ilike ${like} limit ${cap})
+    union all (select 'decision:' || d.id from decision d where d.title ilike ${like} limit ${cap})
+    union all (select 'media:' || e.id from media_entity e where e.title ilike ${like} limit ${cap})`.execute(db);
+  return r.rows.map((x) => x.k);
+}
 
 type Row = {
   id: string;
@@ -234,6 +380,7 @@ export interface LogQueries {
 
 export function logQueries(db: Kysely<DB>, opts: { engineSchema?: string } = {}): LogQueries {
   const engineSchema = opts.engineSchema ?? "openworkflow";
+  const engine = engineSource(engineSchema);
   let engineReady = false;
   const engineExists = async () => {
     if (engineReady) return true;
@@ -247,35 +394,39 @@ export function logQueries(db: Kysely<DB>, opts: { engineSchema?: string } = {})
   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)]);
+      const withEngine = want("engine_run") && (await engineExists());
-      if (want("schedule_run")) sources.push(["schedule_run", SCHEDULE_SOURCE]);
+      const like = q.q ? `%${escapeLike(q.q)}%` : null;
-      if (want("agent_run")) sources.push(["agent_run", AGENT_SOURCE]);
+      const f: Filter = {
-      if (want("audit_event")) sources.push(["audit_event", AUDIT_SOURCE]);
+        types,
-      if (!sources.length) return { items: [], nextCursor: null };
+        levels: q.level?.length ? q.level : null,
-
+        from: q.from ? new Date(q.from) : null,
-      const conds: RawBuilder<unknown>[] = [];
+        to: q.to ? new Date(q.to) : null,
-      if (types) conds.push(sql`s.type = any(${[...types]}::text[])`);
+        like,
-      if (q.level?.length) conds.push(sql`s.level = any(${q.level}::text[])`);
+        matched: like && withEngine ? await keywordObjects(db, like) : [],
-      if (q.from) conds.push(sql`s.at >= ${new Date(q.from)}`);
+        objectType: q.objectType || null,
-      if (q.to) conds.push(sql`s.at < ${new Date(q.to)}`);
+        objectId: q.objectId || null,
-      if (q.q) conds.push(sql`s.search ilike ${`%${escapeLike(q.q)}%`}`);
+        hideRoutine: q.system !== "1" && !q.objectId,
-      if (q.objectType) conds.push(sql`s.object_type = ${q.objectType}`);
+        cursor: decodeCursor(q.cursor),
-      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(
+      const parts: RawBuilder<unknown>[] = [];
-        ([, src]) => sql`(select s.* from (${src}) s ${where} order by s.at desc, s.id desc limit ${n})`,
+      if (withEngine) parts.push(engine(f, n));
-      );
+      if (want("schedule_run")) parts.push(scheduleSource(f, n));
-      const rows = await sql<Row>`select u.*, (extract(epoch from u.at) * 1000000)::bigint as cursor_us
+      if (want("agent_run")) parts.push(agentSource(f, n));
-        from (${sql.join(parts, sql` union all `)}) u
+      if (want("audit_event")) parts.push(auditSource(f, n));
-        order by u.at desc, u.id desc limit ${n}`.execute(db);
+      if (!parts.length) return { items: [], nextCursor: null };
+
+      // 表达式很长,估算代价会触发 JIT,编译一次就要上秒;这条查询只取几百行,关掉 JIT
+      const rows = await db.transaction().execute(async (trx) => {
+        await sql`set local jit = off`.execute(trx);
+        return sql<Row>`select u.*, (extract(epoch from u.at) * 1000000)::bigint as cursor_us
+          from (${sql.join(
+            parts.map((p) => sql`(${p})`),
+            sql` union all `,
+          )}) u
+          order by u.at desc, u.id desc limit ${n}`.execute(trx);
+      });
       const items = rows.rows.slice(0, q.limit);
       const last = items[items.length - 1];
       return {

apps/server/test/logs/legacy-query.ts

@@ -0,0 +1,232 @@
+import type { LogSource, LogType } from "@umt/contracts";
+import { type DB, type Kysely, sql } from "@umt/db";
+import type { RawBuilder } from "kysely";
+import { BUSINESS_EXPR } from "../../src/modules/auth-settings/engine-business.ts";
+import {
+  BUSINESS_OBJECT_TYPE,
+  decodeCursor,
+  encodeCursor,
+  type LogQueries,
+  toEntry,
+} from "../../src/modules/logs/query.ts";
+
+// 改写前的日志查询(先构造带派生列的子查询、外层再筛选),只作对照口径:新查询在同一份数据上的结果必须与它一致。
+// 在大数据量上极慢,只用于小数据集。
+
+type Row = Parameters<typeof toEntry>[0];
+
+const SOURCE_TYPES: Record<LogSource, LogType[]> = {
+  engine_run: ["task"],
+  schedule_run: ["auto"],
+  agent_run: ["auto"],
+  audit_event: ["auth", "audit"],
+};
+
+const escapeLike = (s: string) => s.replace(/[\\%_]/g, (c) => `\\${c}`);
+
+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`;
+
+export function legacyLogQueries(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})`,
+      );
+      // 对照只关心结果;关掉 JIT 省掉每条查询上秒的编译
+      const rows = await db.transaction().execute(async (trx) => {
+        await sql`set local jit = off`.execute(trx);
+        return sql<Row & { cursor_us: string }>`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(trx);
+      });
+      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,
+      };
+    },
+  };
+}

apps/server/test/logs/logs-legacy-parity.test.ts

@@ -0,0 +1,117 @@
+import type { LogPage } from "@umt/contracts";
+import { sql } from "@umt/db";
+import { afterAll, beforeAll, describe, expect, it } from "vitest";
+import { type LogQueryInput, logQueries } from "../../src/modules/logs/query.ts";
+import { createHarness, type Harness, pgAvailable } from "../auth-settings/harness.ts";
+import { legacyLogQueries } from "./legacy-query.ts";
+import { seedLogVolume } from "./volume-seed.ts";
+
+// 改写后的日志查询与改写前(先算派生列再筛选)在同一份小数据上逐条一致:筛选、关键词、对象、时间、游标分页。
+
+const hasPg = await pgAvailable();
+
+describe.skipIf(!hasPg)("日志查询改写前后口径一致", () => {
+  let h: Harness;
+  let feedId = "";
+  let downloadId = "";
+  let agentId = "";
+
+  beforeAll(async () => {
+    h = await createHarness({ engine: true });
+    await seedLogVolume(h.db, h.engineSchema, 400);
+    const wr = sql.raw(`"${h.engineSchema}".workflow_runs`);
+    feedId = (await sql<{ id: string }>`select id::text from feed_source where name like 'my%'`.execute(h.db)).rows[0]!
+      .id;
+    downloadId = (await sql<{ id: string }>`select id::text from download_task order by name limit 1`.execute(h.db))
+      .rows[0]!.id;
+    agentId = (
+      await sql<{ id: string }>`select id::text from agent_run where status = 'failed' order by goal limit 1`.execute(
+        h.db,
+      )
+    ).rows[0]!.id;
+    // 没有业务 id 的运行靠业务任务行(task.engine_run_id)关联对象:一条指向失败的智能体会话、一条指向下载任务
+    await sql`insert into task (kind, idempotency_key, engine_run_id, subject_type, subject_id)
+      values ('agent', 'parity-1', 'vol-000060', 'agent_run', ${agentId}),
+        ('download', 'parity-2', 'vol-000061', 'download_task', ${downloadId})`.execute(h.db);
+    // 失败的智能体会话对应的引擎运行(完成)按智能体结果算失败;定时任务的手动重跑不算定时触发
+    await sql`update ${wr} set input = jsonb_build_object('input', jsonb_build_object('agentRunId', ${agentId}::text))
+      where id = 'vol-000077'`.execute(h.db);
+    await sql`insert into tasks_run_meta (run_id, task_name, task_version, idempotency_key)
+      values ('vol-000089', 'purgeTrash', 1, 'manual:1')`.execute(h.db);
+    // 审计里指向下载任务的一条(按对象筛选时与任务一起列出)
+    await sql`insert into audit_event (actor_type, actor_id, action, target_type, target_id, payload)
+      values ('agent', 'run-1', 'capability.denied', 'download_task', ${downloadId}, '{"reason":"超出授权"}'::jsonb)`.execute(
+      h.db,
+    );
+  }, 120_000);
+  afterAll(async () => {
+    await h?.close();
+  });
+
+  const at = (hoursAgo: number) => new Date(Date.now() - hoursAgo * 3600_000).toISOString();
+  const filters = (): [string, Omit<LogQueryInput, "limit">][] => [
+    ["全部", {}],
+    ["含例行", { system: "1" }],
+    ["error", { level: ["error"] }],
+    ["warn", { level: ["warn"] }],
+    ["success", { level: ["success"] }],
+    ["running + info", { level: ["running", "info"] }],
+    ["error 含例行", { level: ["error"], system: "1" }],
+    ["task", { type: ["task"] }],
+    ["task 含例行", { type: ["task"], system: "1" }],
+    ["auto", { type: ["auto"], system: "1" }],
+    ["auth", { type: ["auth"] }],
+    ["audit", { type: ["audit"] }],
+    ["task + auth + warn", { type: ["task", "auth"], level: ["warn"] }],
+    ["近 6 小时", { from: at(6), system: "1" }],
+    ["6 到 2 小时前", { from: at(6), to: at(2) }],
+    ["q 业务标题", { q: "feeds", system: "1" }],
+    ["q 下载名", { q: "下载 1" }],
+    ["q 智能体目标", { q: "调查 1", system: "1" }],
+    ["q 运行 id", { q: "VOL-00012", system: "1" }],
+    ["q 工作流名", { q: "ingest" }],
+    ["q 错误", { q: "失败原因 9", system: "1" }],
+    ["q 审计负载", { q: "10.1.0.5" }],
+    ["q 定时任务", { q: "超时", system: "1" }],
+    ["q 例行定时", { q: "清理回收站", system: "1" }],
+    ["q 字面 %", { q: "%" }],
+    ["q 不命中", { q: "zz-nomatch", system: "1" }],
+    ["对象 feed", { objectType: "feed_source", objectId: feedId }],
+    ["对象 下载(只给 id)", { objectId: downloadId }],
+    ["对象 下载 + 类型", { objectType: "download_task", objectId: downloadId }],
+    ["对象 定时任务", { objectType: "scheduled_job", objectId: "trash.purge" }],
+    ["对象 智能体", { objectType: "agent_run", objectId: agentId }],
+    ["对象类型不存在", { objectType: "nope" }],
+  ];
+
+  it("各种筛选一次取全:逐条一致", async () => {
+    const next = logQueries(h.db, { engineSchema: h.engineSchema });
+    const old = legacyLogQueries(h.db, { engineSchema: h.engineSchema });
+    const empty: string[] = [];
+    for (const [name, f] of filters()) {
+      const a: LogPage = await next.list({ ...f, limit: 200 });
+      const b: LogPage = await old.list({ ...f, limit: 200 });
+      expect(a, name).toEqual(b);
+      if (!a.items.length) empty.push(name);
+    }
+    // 对照有意义:除了故意不命中的,筛选都有结果
+    expect(empty).toEqual(["q 字面 %", "q 不命中", "对象类型不存在"]);
+  }, 120_000);
+
+  it("游标分页:每一页与改写前一致", async () => {
+    const next = logQueries(h.db, { engineSchema: h.engineSchema });
+    const old = legacyLogQueries(h.db, { engineSchema: h.engineSchema });
+    for (const f of [{ system: "1" as const }, { level: ["error" as const] }, { q: "下载", system: "1" as const }]) {
+      let cursor: string | null = null;
+      let pages = 0;
+      do {
+        const a: LogPage = await next.list({ ...f, limit: 37, cursor });
+        const b: LogPage = await old.list({ ...f, limit: 37, cursor });
+        expect(a).toEqual(b);
+        cursor = a.nextCursor;
+        pages++;
+      } while (cursor && pages < 40);
+      expect(pages).toBeGreaterThan(1);
+    }
+  }, 120_000);
+});

apps/server/test/logs/logs-volume.test.ts

@@ -0,0 +1,105 @@
+import { LogPage } from "@umt/contracts";
+import { sql } from "@umt/db";
+import { afterAll, beforeAll, describe, expect, it } from "vitest";
+import { createLogStreamPoller, createLogsModule } from "../../src/modules/logs/index.ts";
+import { logQueries } from "../../src/modules/logs/query.ts";
+import { bearer, createHarness, type Harness, pgAvailable } from "../auth-settings/harness.ts";
+import { DEV_KEY, json } from "../helpers.ts";
+import { seedLogVolume, VOLUME_RUNS } from "./volume-seed.ts";
+
+// 日志接口在真实数据量上的耗时(约 30,000 条引擎运行 + 定时任务运行 + 审计)。
+// 改写前这几种查询在部署库上超时(> 60s);每个来源先按原始列筛选取前 n 条再算标题后,应在 1 秒内(阈值放宽到 1.5s 防抖动)。
+
+const hasPg = await pgAvailable();
+const LIMIT_MS = 1500;
+
+describe.skipIf(!hasPg)("日志查询耗时(约 3 万条引擎运行)", () => {
+  let h: Harness;
+  let admin: ReturnType<typeof bearer>;
+
+  beforeAll(async () => {
+    h = await createHarness({ engine: true });
+    const { createApp } = await import("../../src/http/app.ts");
+    const { createAuthSettingsModule } = await import("../../src/modules/auth-settings/index.ts");
+    h.app = createApp({
+      config: h.ctx.config,
+      logger: h.ctx.logger,
+      bus: h.bus,
+      db: h.db,
+      role: "all",
+      version: "0.0.0-test",
+      modules: [
+        createAuthSettingsModule({ setupTokenOnStart: false, engineSchema: h.engineSchema }),
+        createLogsModule({ engineSchema: h.engineSchema }),
+      ],
+      ctx: h.ctx,
+    }) as Harness["app"];
+    admin = bearer(h.app, DEV_KEY);
+    await seedLogVolume(h.db, h.engineSchema);
+    const n = await sql<{
+      n: number;
+    }>`select count(*)::int as n from ${sql.raw(`"${h.engineSchema}".workflow_runs`)}`.execute(h.db);
+    expect(n.rows[0]!.n).toBe(VOLUME_RUNS);
+    // 预热连接
+    expect((await admin("GET", "/logs?limit=1")).status).toBe(200);
+  }, 120_000);
+  afterAll(async () => {
+    await h?.close();
+  });
+
+  const since24h = () => encodeURIComponent(new Date(Date.now() - 24 * 3600_000).toISOString());
+  const cases: [string, () => string, (items: LogPage["items"]) => void][] = [
+    [
+      "level=error&since=24h",
+      () => `?level=error&from=${since24h()}&limit=50`,
+      (items) => {
+        expect(items.length).toBeGreaterThan(0);
+        expect(items.every((e) => e.level === "error")).toBe(true);
+      },
+    ],
+    [
+      "system=1&q=feeds",
+      () => "?system=1&q=feeds&limit=50",
+      (items) => {
+        // 业务标题(RSS 源「my feeds 汇总」)命中的任务
+        expect(items.some((e) => e.source === "engine_run" && e.title === "my feeds 汇总")).toBe(true);
+      },
+    ],
+    [
+      "type=task&level=error",
+      () => "?type=task&level=error&limit=50",
+      (items) => {
+        expect(items).toHaveLength(50);
+        expect(items.every((e) => e.type === "task" && e.level === "error")).toBe(true);
+      },
+    ],
+    ["type=task", () => "?type=task&limit=50", (items) => expect(items).toHaveLength(50)],
+    ["默认", () => "?limit=50", (items) => expect(items).toHaveLength(50)],
+    // 关键词一条都不命中:引擎来源要扫完全部运行,是最坏情况
+    ["system=1&q=<不命中>", () => "?system=1&q=zz-nomatch&limit=50", (items) => expect(items).toEqual([])],
+  ];
+
+  for (const [name, query, check] of cases) {
+    it(`${name} < ${LIMIT_MS}ms`, async () => {
+      const t = performance.now();
+      const res = await admin("GET", `/logs${query()}`);
+      const ms = performance.now() - t;
+      expect(res.status).toBe(200);
+      const items = LogPage.parse((await json(res)).data).items;
+      console.log(`[logs-volume] ${name}: ${ms.toFixed(0)}ms(${items.length} 条)`);
+      check(items);
+      expect(ms).toBeLessThan(LIMIT_MS);
+    });
+  }
+
+  it(`流的一轮轮询 < ${LIMIT_MS}ms`, async () => {
+    const poller = createLogStreamPoller(logQueries(h.db, { engineSchema: h.engineSchema }), { system: "1" });
+    for (let i = 0; i < 2; i++) {
+      const t = performance.now();
+      await poller.poll();
+      const ms = performance.now() - t;
+      console.log(`[logs-volume] 流轮询 #${i + 1}: ${ms.toFixed(0)}ms`);
+      expect(ms).toBeLessThan(LIMIT_MS);
+    }
+  });
+});

apps/server/test/logs/volume-seed.ts

@@ -0,0 +1,87 @@
+import { type DB, type Kysely, sql } from "@umt/db";
+
+// 日志性能测试的数据量:约 30,000 条引擎运行(多种工作流 / 状态 / 业务对象,约 28 天)
+// + 定时任务运行 + 智能体运行 + 审计。分布参照部署库(大部分是 RSS / 站点快照的成功运行,失败约 1%)。
+
+export const VOLUME_RUNS = 30_000;
+
+export async function seedLogVolume(db: Kysely<DB>, engineSchema: string, runs = VOLUME_RUNS): Promise<void> {
+  const wr = sql.raw(`"${engineSchema}".workflow_runs`);
+  await sql`insert into site (key, name, adapter, base_url)
+    select 'vs' || g, '站点 ' || g, 'nexusphp', 'https://s' || g || '.example' from generate_series(1, 20) g`.execute(
+    db,
+  );
+  await sql`insert into feed_source (name, url_display)
+    select case when g = 7 then 'my feeds 汇总' else 'RSS 源 ' || g end, 'https://f' || g || '.example'
+    from generate_series(1, 30) g`.execute(db);
+  await sql`insert into media_entity (kind, title) select 'movie', '电影 ' || g from generate_series(1, 500) g`.execute(
+    db,
+  );
+  await sql`insert into subscription (media_kind, media_entity_id, title_query)
+    select 'movie', e.id, null from (select id from media_entity order by title limit 200) e`.execute(db);
+  const inst = await sql<{ id: string }>`insert into downloader_instance (name, base_url)
+    values ('qb-volume', 'http://qb') returning id`.execute(db);
+  await sql`insert into download_task (instance_id, torrent_hash, name)
+    select ${inst.rows[0]!.id}::uuid, md5('vol' || g) || left(md5('h' || g), 8), '下载 ' || g || ' 2160p'
+    from generate_series(1, 2000) g`.execute(db);
+  await sql`insert into agent_run (trigger, goal, status, created_at, started_at, finished_at)
+    select (array['event', 'schedule', 'user'])[1 + g % 3], '调查 ' || g,
+      (array['succeeded', 'succeeded', 'succeeded', 'failed', 'budget_exceeded', 'cancelled'])[1 + g % 6],
+      now() - g * interval '20 minutes', now() - g * interval '20 minutes', now() - g * interval '20 minutes' + interval '1 minute'
+    from generate_series(1, 1500) g`.execute(db);
+  await sql`insert into scheduled_job (id, name, board_group, kind, enabled, default_enabled, schedule, default_schedule)
+    values ('vol.feeds', 'RSS 抓取(量)', 'collect', 'task', true, true, '{"type":"interval","everyMs":60000}'::jsonb,
+      '{"type":"interval","everyMs":60000}'::jsonb),
+      ('trash.purge', '清理回收站', 'maintain', 'task', true, true, '{"type":"interval","everyMs":3600000}'::jsonb,
+      '{"type":"interval","everyMs":3600000}'::jsonb),
+      ('vol.check', '巡检(量)', 'agent', 'agent_check', true, true, '{"type":"interval","everyMs":86400000}'::jsonb,
+      '{"type":"interval","everyMs":86400000}'::jsonb)
+    on conflict (id) do nothing`.execute(db);
+  await sql`insert into scheduled_job_run (job_id, trigger, status, error, started_at, finished_at, duration_ms)
+    select (array['vol.feeds', 'trash.purge', 'vol.check'])[1 + g % 3], 'schedule',
+      (case when g % 50 = 0 then 'failed' when g % 7 = 0 then 'skipped' else 'ok' end),
+      (case when g % 50 = 0 then '站点超时 ' || g end),
+      now() - g * interval '37 minutes', now() - g * interval '37 minutes', 500
+    from generate_series(1, 1100) g`.execute(db);
+  await sql`insert into audit_event (at, actor_type, actor_id, action, target_type, target_id, payload)
+    select now() - g * interval '53 minutes', 'user', 'admin',
+      (array['admin.login_success', 'settings.update', 'admin.login_failed', 'download.cancel'])[1 + g % 4],
+      'settings', 'k' || g, jsonb_build_object('ip', '10.1.0.' || (g % 200))
+    from generate_series(1, 800) g`.execute(db);
+
+  // 引擎运行:按 g % 100 分桶决定工作流与业务对象,g 越大越早(每条间隔 80 秒)
+  await sql`with
+      feeds as (select array_agg(id::text order by name) a from feed_source),
+      sites as (select array_agg(id::text order by key) a from site where key like 'vs%'),
+      dls as (select array_agg(id::text order by name) a from download_task where instance_id = ${inst.rows[0]!.id}::uuid),
+      agents as (select array_agg(id::text order by goal) a from agent_run where goal like '调查 %'),
+      subs as (select array_agg(id::text order by id) a from subscription),
+      src as (
+        select g, g % 100 as b, now() - g * interval '80 seconds' as at,
+          (case when g <= 5 then 'running' when g % 97 = 0 then 'failed' when g % 193 = 0 then 'canceled'
+            else 'completed' end) as st
+        from generate_series(1, ${runs}) g)
+    insert into ${wr} (namespace_id, id, workflow_name, version, status, config, context, input, error, attempts,
+      available_at, started_at, finished_at, created_at, updated_at)
+    select 'default', 'vol-' || lpad(g::text, 6, '0'),
+      (case when b < 40 then 'ingestFeed' when b < 55 then 'snapshotSite' when b < 65 then 'refreshSiteAccount'
+        when b < 70 then 'submitDownload' when b < 75 then 'verifyAndImport' when b < 80 then 'runAgent'
+        when b < 88 then 'searchSubscription' when b < 93 then 'purgeTrash' when b < 98 then 'scanOrganizeSource'
+        else 'evaluateAcquisition' end),
+      'v1', st, '{}'::jsonb, null,
+      jsonb_build_object('input', (case
+        when b < 40 then jsonb_build_object('feedSourceId', feeds.a[1 + g % 30])
+        when b < 55 then jsonb_build_object('siteId', sites.a[1 + g % 20])
+        when b < 65 then '{}'::jsonb
+        when b < 75 then jsonb_build_object('downloadTaskId', dls.a[1 + g % 2000])
+        when b < 80 then jsonb_build_object('agentRunId', agents.a[1 + g % 1500])
+        when b < 88 then jsonb_build_object('subscriptionId', subs.a[1 + g % 200])
+        when b < 98 then '{}'::jsonb
+        else jsonb_build_object('acquisitionPlanId', md5('plan' || g)::uuid::text) end)),
+      (case when st = 'failed' then jsonb_build_object('message', '失败原因 ' || g) end),
+      1, at, at, (case when st = 'running' then null else at + interval '3 seconds' end), at, at
+    from src, feeds, sites, dls, agents, subs`.execute(db);
+  await sql`analyze ${wr}`.execute(db);
+  await sql`analyze site, feed_source, media_entity, subscription, download_task, agent_run, scheduled_job,
+    scheduled_job_run, audit_event`.execute(db);
+}

apps/web/scripts/smoke.mjs

@@ -275,7 +275,9 @@ async function main() {
     const t0 = Date.now();
     try {
       await page.goto("about:blank");
-      await page.goto(`${BASE}/console/audit?mock=1`, { waitUntil: "networkidle", timeout: 30_000 });
+      // 起始页用不跳转的日志页(/console/audit 已并进日志页,会跳到 /console/logs?type=audit 且丢掉 ?mock=1)
+      await page.goto(`${BASE}/console/logs?type=audit&mock=1`, { waitUntil: "networkidle", timeout: 30_000 });
+      const startPath = new URL(page.url()).pathname;
       const launcher = page.locator("[data-agent-launcher]:visible").last();
       await launcher.click();
       await page.getByRole("button", { name: "历史会话" }).click();
@@ -288,7 +290,7 @@ async function main() {
       // 等历史弹层退场后再按 Escape,避免按键被仍在退场的 Popover 消耗。
       await list.waitFor({ state: "hidden", timeout: 5_000 });
       await page.getByRole("button", { name: "新对话" }).and(page.locator(":enabled")).waitFor({ timeout: 10_000 });
-      if (new URL(page.url()).pathname !== "/console/audit") problems.push("点历史会话跳走了页面");
+      if (new URL(page.url()).pathname !== startPath) problems.push("点历史会话跳走了页面");
       await page.keyboard.press("Escape");
       await page.locator('[data-slot="sheet-content"]').waitFor({ state: "hidden", timeout: 5_000 });
       await launcher.click();

packages/db/migrations/0088_log_audit_at_index.ts

@@ -0,0 +1,17 @@
+import type { Kysely } from "kysely";
+import { execDdl } from "../src/migration-helpers.ts";
+
+// 日志接口:审计来源按时间倒序取前 n 条,原有索引都以 action / actor / target 开头,只能整表排序;
+// 补一个按时间的索引(审计会随登录、设置修改一直增长)。
+// 引擎表 openworkflow.workflow_runs 不在这里加索引:引擎自带 (namespace_id, created_at desc) 与
+// (namespace_id, status, created_at desc),日志查询已能沿用。
+
+// biome-ignore lint/suspicious/noExplicitAny: 迁移不依赖应用表类型
+export async function up(db: Kysely<any>): Promise<void> {
+  await execDdl(db, `create index if not exists audit_event_at_idx on audit_event (at desc);`);
+}
+
+// biome-ignore lint/suspicious/noExplicitAny: 同上
+export async function down(db: Kysely<any>): Promise<void> {
+  await execDdl(db, `drop index if exists audit_event_at_idx;`);
+}