fmode-app/src/app/core/task-pulse.service.ts

import { computed, effect, inject, Injectable, signal } from '@angular/core';
import { Agent } from './api/models';
import { LiveService } from './live.service';
import { QUIET_MS } from './turn-aggregator';
import { isReporterSelfTalk, HEARTBEAT_STALE_SEC, resolveAgentState, sessionRunning } from './agent-state';

/**
 * 服务端数据「新鲜」窗口（毫秒）。
 *
 * 🔴 2026-11-10 用户裁决：**与全站统一为外层心跳阈值**（80 分钟）。
 *
 * 原值是 90 秒。后果：容器一轮对话结束 90 秒后，顶栏就开始显示
 * 「状态未同步 · 上次上报 8 分钟前」——而侧栏、通讯录、新建弹窗
 * 走的是 `HEARTBEAT_STALE_SEC`（80 分钟），判成「空闲」。
 * 同一个容器，顶栏说「未同步」、别处说「在线」，这就是用户反馈的
 * 「外层和内层不一致」。
 *
 * 容器上报是**小时级**的（用户裁决：一小时同步一次，带 0~15 分钟抖动）。
 * 对话结束后不需要逼它"证明还活着"——只要没超过外层兜底阈值，就展示正常态。
 */
const FRESH_MS = HEARTBEAT_STALE_SEC * 1000;

/**
 * 「还在回复」的静默窗口：最近一次消息进来后这么久内，都算正在输出。
 *
 * 🔴 必须与 `turn-aggregator.QUIET_MS`（回合卡片的 live 判定）**同一口径**
 * （2026-11-08 用户反馈「只闪一下『正在回复』就消失」）。
 *
 * 原先这里是 20 秒、回合那边是 3 秒：顶栏说「正在回复…」，
 * 正文区的「正在思考…」却早已消失 —— 两处对同一件事给出相反结论，
 * 用户不知道该信哪个。现在两处直接取同一个常量，想改一起改。
 */
const REPLY_QUIET_MS = QUIET_MS;

/** 超过这么久的事件视为「历史重放」，不据此判断"正在回复" */
const STALE_MSG_MS = 60_000;

export type PulseKind = 'started' | 'finished';

export interface TaskPulse {
  agentId: string;
  kind: PulseKind;
  /** 观测到翻转的时刻（本地，毫秒） */
  at: number;
  /** 开始时的任务描述（来自上一轮 currentTask） */
  task?: string;
  /** 完成时的用时（由两次观测的时刻差估算） */
  durationMs?: number;
}

interface Mark {
  running: boolean;
  at: number;
  task?: string;
}

/**
 * 运行态脉冲 —— 「当前任务」不要等很久才出现。
 *
 * ## 解决什么（用户实测反馈）
 *
 * 容器的上报间隔是分钟级（Hermes turn 开始/结束才写库，再等下一轮上报），
 * 所以「发了消息 → 右侧还是空闲」会持续很久，用户以为没发出去。
 *
 * ## 做法：**三分职责，谁都不越界**
 *
 * | 信号来源 | 能确定的事 | 谁负责 |
 * |---|---|---|
 * | 用户在 App 里发消息 / 传附件 | 任务**已经开始** | 前端**立即**置 running |
 * | 容器上报 `status=idle` | 任务**已经结束** | 只有容器能说 |
 * | 服务端快照 | 上一轮观测到的任务描述 | 轮询对比得出「变化」 |
 *
 * 🔴 **前端绝不自己翻回 idle** —— 那是编造「完成」。
 *    下一条用户消息出现前的一切「忙」，都以容器的 `idle` 为准。
 *
 * ## 界面该怎么描述
 *
 * 只描述**已观测到的事实**：
 *
 * ```
 * 刚刚开始 · 正在处理你的消息      ← 用户刚发消息
 * 正在执行 · 处理企业微信用户对话  ← 容器上报的任务描述
 * 已完成 · 用时 3 分 20 秒        ← 观测到 running → idle
 * 空闲                            ← 容器说闲 AND 快照新鲜
 * 状态未同步 · 等待容器上报        ← 快照超过 90s 没变（不假装在线空闲）
 * ```
 *
 * 不写「先梳理需求 → 再执行」这类**计划**：容器没报计划，写了就是编。
 */
@Injectable({ providedIn: 'root' })
export class TaskPulseService {
  private readonly live = inject(LiveService);

  /** 本地认为「刚开始」的会话（agentId → 发出消息的时刻） */
  private readonly _localStart = signal<ReadonlyMap<string, number>>(new Map());
  /** 每次轮询观测到的翻转 */
  private readonly _pulses = signal<readonly TaskPulse[]>([]);

  readonly pulses = this._pulses.asReadonly();

  /** 上一轮观测（agentId → 是否在忙 / 时刻 / 任务描述） */
  private marks = new Map<string, Mark>();

  /** 最近一条消息到达时刻（agentId → 毫秒），用于「正在回复」判定 */
  private readonly lastMsgAt = new Map<string, number>();
  /** 会话里最后一条消息（含内容），用于识别新消息到达 */
  private lastSeenMsgId = '';
  /** 首次观测需要先建立基线，不能把「第一次看到 idle」当成「刚完成」 */
  private primed = false;

  constructor() {
    effect(() => {
      // 每轮轮询都要观测一次（数据未变时 agents 引用不变，靠 tick 推进）
      this.live.tick();
      const agents = this.live.agents();
      if (agents.length === 0) return;
      this.observe(agents);
    });
  }

  /** 用户刚发出消息：本地立刻进入 running（**仅前端展示，不改服务端**） */
  noteUserAction(agentId: string): void {
    const next = new Map(this._localStart());
    next.set(agentId, Date.now());
    this._localStart.set(next);
    // 本地推测也记进 marks 的「开始时刻」，便于算用时
    const prev = this.marks.get(agentId);
    if (prev && !prev.running) {
      this.push({ agentId, kind: 'started', at: Date.now(), task: prev.task });
    }
    this.marks.set(agentId, { running: true, at: Date.now(), task: prev?.task });
  }

  /**
   * 记录「有新消息到达」（由 ChatService / 聊天页调用）。
   *
   * 这是比容器上报**更新鲜**的信号：消息是实时的，运行态是分钟级的。
   */
  noteMessage(
    agentId: string,
    msgId: string,
    opts?: { live?: boolean; fromAgent?: boolean; ts?: number },
  ): void {
    if (!agentId || msgId === this.lastSeenMsgId) return;
    this.lastSeenMsgId = msgId;
    /*
     * 🔴 只有「**实时到达的、对方发的**」消息才算「正在回复」。
     *
     * 上一版只看「尾部 id 变了」，导致**刷新页面时重新灌入历史**也被当成新消息
     * → 一打开就显示「正在回复」，20 秒后才好（用户实测）。
     *
     * 历史重读与实时事件的区分由调用方给（`live`）；默认按历史处理，
     * 宁可少报「正在回复」，也不能在没有任何回复时假装它在回。
     */
    if (opts?.live !== true || opts?.fromAgent !== true) return;
    /*
     * 🔴 还要看**消息自身的时间戳**：matrix-js-sdk 在初始同步时会把已同步的历史
     *    事件经由 onMessage 回放一遍 —— 它走的是"实时"回调，但消息其实很旧。
     *    只看「回调是不是实时的」会把刷新时的历史重放当成新消息（用户实测：
     *    一打开就显示「正在回复」）。
     */
    const ts = opts.ts ?? Date.now();
    if (Date.now() - ts > STALE_MSG_MS) return;
    this.lastMsgAt.set(agentId, Date.now());
  }

  /** 最近一次消息活动的时刻（没有则返回 0，视作很久以前） */
  private lastActivity(agentId: string): number {
    return this.lastMsgAt.get(agentId) ?? 0;
  }

  /** 当前会话是否在「本地推测的忙碌」中（容器明确说 idle 时自动解除） */
  localStartAt = (agentId: string): number | undefined => this._localStart().get(agentId);

  /**
   * 是否在忙（供 UI 判定）。
   *
   * `localStart` 只在下一次上报到来之前有效：容器已在处理时它自然会变成 running，
   * 容器报 idle 时**以容器为准**，本地推测立刻作废。
   */
  isBusy(agent: Agent | undefined, now = Date.now()): boolean {
    if (!agent) return false;
    if (agent.status === 'running') return true;
    if (agent.status === 'idle' || agent.status === 'offline' || agent.status === 'stopped') {
      return false;
    }
    const local = this._localStart().get(agent.id);
    return local !== undefined && now - local < REPLY_QUIET_MS;
  }

  /** 右侧「最近动态」：只列本会话的翻转（用户真正关心的是它在干什么） */
  pulsesFor(agentId: string | null): readonly TaskPulse[] {
    const all = this._pulses();
    return (agentId ? all.filter((p) => p.agentId === agentId) : all).slice(0, 6);
  }

  /** 「当前任务」的一句话文案（唯一实现，列表/头部/右栏都用它） */
  describe(
    agent: Agent | undefined,
    now = Date.now(),
    sessionId?: string | null,
    roomId?: string | null,
    myUserId?: string | null,
  ): TaskLine {
    // 依赖这些信号，让调用方在数据更新后重算
    this._localStart();
    this.live.tick();
    return this.describeOf(agent, now, sessionId, roomId, myUserId);
  }

  private describeOf(
    agent: Agent | undefined,
    now: number,
    sessionId?: string | null,
    roomId?: string | null,
    myUserId?: string | null,
  ): TaskLine {
    if (!agent) return { text: '', tone: 'neutral' };
    const local = this._localStart().get(agent.id);
    /*
     * 🔴 不可用优先，且判据与全站同一套（2026-11-08）。
     *
     * 原先这里只判 `status`，不看心跳。后果：45 天没心跳、`status=running`
     * 的僵尸容器在这里显示「状态未同步」或甚至任务名，而移动端会话列表
     * 已经把它判成红色「离线」—— 同一个容器两处结论相反。
     * 现在直接复用 `resolveAgentState`，先处理不可用分支。
     */
    const resolved = resolveAgentState({ agent, now });
    if (resolved.state === 'error') {
      return { text: '运行异常', tone: 'danger' };
    }
    if (resolved.state === 'offline') {
      return { text: '已离线', tone: 'danger' };
    }
    /*
     * 🔴 2026-11-09 调整顺序（C3）：
     *
     * 原顺序是「status=running → 心跳未同步 → 正在回复」，有两个问题：
     *   ① `local && isBusy`（用户刚发消息）排在 `status=running` 之后，
     *      容器报 running 时永远走不到 → 用户发完消息看不到「刚刚开始」；
     *   ② 心跳过期（>90s）检查排在「最近有消息进来」之前 → 容器正在回话时，
     *      顶栏却显示「状态未同步」，与正文的「正在输出…」直接打架。
     *
     * 正确优先级：**用户动作 / 实时消息流 > 容器自报 > 心跳新鲜度**。
     * 理由：消息流是实时的，容器上报是分钟级的；越实时的信号越应该优先。
     */
    if (local && this.isBusy(agent, now)) {
      return {
        text: '刚刚开始 · 正在处理你的消息',
        tone: 'info',
        active: true,
        since: local,
      };
    }
    /*
     * 🔴 会话页：**只看本会话的进度**（2026-11-11 用户裁决）。
     *
     * 现象：打开会话 A，页头显示「已完成 · 用时 10 分 16 秒」，而这个数字来自
     * 容器**全局**的 `status` / `currentTask` —— 容器同一时刻可能正在跑会话 B，
     * 用户看到的却是别人的进度（实测截图）。
     *
     * 容器在 `pre_llm_call`/`post_llm_call` 里按会话维护 `extra.currentSessions`
     * （`{sessionKey,status,turnStartAt}`），App 侧 `sessionRunning()` 已经能按
     * `sessionId` 查出本会话状态。所以：
     *
     *   · 本会话 running         → 显示任务（容器全局任务名只在**本会话确实在跑**时采信）
     *   · 本会话明确不在跑       → 不再引用全局 `status`/`lastFinish`，直接「在线」
     *   · 容器没上报会话级状态   → 返回 undefined，回落到原有全局口径（老容器兜底）
     *
     * 用户原话：「右上角仅显示当前会话的进度，如果当前会话没有就显示在线状态」。
     */
    const sessionAware = sessionId !== undefined && sessionId !== null;
    /*
     * 容器**有**会话级上报，我才敢拿它下结论。
     * `currentSessions` 为空 = 老容器从没报过 → `mine` 只能是 undefined，
     * 此时回落到下面的全局口径，行为与改造前完全一致。
     */
    const hasSessionReport = (agent.currentSessions?.length ?? 0) > 0;
    const mine = sessionAware ? sessionRunning(agent.currentSessions, sessionId, roomId, myUserId) : undefined;
    if (sessionAware && hasSessionReport && mine !== true) {
      /*
       * 本会话没在跑 —— 两种情况都归到这里：
       *   · `mine === false`：容器明确报了本会话 idle
       *   · `mine === undefined`：容器报了别的会话、**没报本会话**
       *     （`sessionRunning` 约定 undefined = 不知道；但在「页面已指定会话」
       *     的语义下，容器报了一堆会话却唯独没有我这条，就是**我没有在跑**）
       *
       * 但上报是分钟级的，刚发完消息时容器可能还没更新 —— 用「最近有消息进来」
       * 兜住，避免正文还在流式输出就显示「在线」。
       */
      const quietMs = now - this.lastActivity(agent.id);
      if (quietMs < REPLY_QUIET_MS) {
        return { text: '正在回复…', tone: 'info', active: true };
      }
      return { text: '在线', tone: 'success' };
    }
    /*
     * 容器已经回 idle，但可能**还在往外发消息**（用户实测：还在回复就显示「已完成」）。
     *
     * 容器的运行态是分钟级上报的，turn 结束与最后一条消息之间可能差几十秒；
     * 而消息流是实时的。所以「最近有消息进来」比「上报说 idle」更新鲜，
     * 必须以它为准 —— 否则用户看着还在输出，状态却说完成了。
     */
    const quietMs = now - this.lastActivity(agent.id);
    if (quietMs < REPLY_QUIET_MS) {
      return { text: '正在回复…', tone: 'info', active: true };
    }
    /*
     * `mine === true` 表示**本会话确实在跑**，此时才采信容器的任务名；
     * 没上报会话级状态（老容器）仍按全局 `status` 判断，行为不变。
     */
    if ((mine === true || (mine === undefined && agent.status === 'running')) && !isReporterSelfTalk(agent)) {
      const task = agent.currentTask?.trim() || agent.currentAction?.trim();
      return { text: task || '正在处理', tone: 'success', active: true };
    }
    // 容器说 idle，但快照可能早就停了 → 分清「真空闲」和「状态没同步」
    const beat = agent.heartbeatAt ? Date.parse(agent.heartbeatAt) : NaN;
    if (Number.isFinite(beat) && now - beat > FRESH_MS) {
      return { text: `状态未同步 · 上次上报 ${agoText(now - beat)}`, tone: 'warning' };
    }

    const last = this.lastFinish(agent.id);
    if (last?.durationMs != null && now - last.at < 10 * 60_000) {
      return { text: `已完成 · 用时 ${fmtDur(last.durationMs)}`, tone: 'success' };
    }
    return { text: '空闲', tone: 'neutral' };
  }

  private lastFinish(agentId: string): TaskPulse | undefined {
    return this._pulses().find((p) => p.agentId === agentId && p.kind === 'finished');
  }

  private observe(agents: Agent[]): void {
    const now = Date.now();
    const pulses: TaskPulse[] = [];
    for (const a of agents) {
      const running = a.status === 'running';
      const prev = this.marks.get(a.id);
      if (prev) {
        if (!prev.running && running) {
          pulses.push({ agentId: a.id, kind: 'started', at: now, task: a.currentTask });
        } else if (prev.running && !running) {
          pulses.push({
            agentId: a.id,
            kind: 'finished',
            at: now,
            task: prev.task,
            durationMs: Math.max(0, now - prev.at),
          });
        }
      }
      /*
       * 本地「我刚发出消息」标记的清理：不能只挂在「running → 非 running」
       * 这条翻转上（2026-11-09）。
       *
       * 原因：若容器**从头到尾没进过 running**（比如它自报一直是 idle，
       * 或者上报间隔很长错过了整个翻转），`prev.running` 一直是 false，
       * 那段清理代码永远走不到，`localStart` 就会一直挂到进程结束。
       * 容器明确报非 running 时直接清掉即可，不需要等翻转。
       */
      if (!running) {
        const cur = this._localStart();
        if (cur.has(a.id)) {
          const next = new Map(cur);
          next.delete(a.id);
          this._localStart.set(next);
        }
      }
      this.marks.set(a.id, { running, at: running ? prev?.at ?? now : now, task: a.currentTask });
    }
    if (!this.primed) {
      this.primed = true;
      return; // 首次只建基线，不报翻转
    }
    if (pulses.length) this.push(...pulses);
  }

  private push(...items: TaskPulse[]): void {
    this._pulses.update((old) => [...items.slice().reverse(), ...old].slice(0, 40));
  }
}

export interface TaskLine {
  text: string;
  tone: 'success' | 'warning' | 'info' | 'danger' | 'neutral';
  /** 是否处于「正在做」状态（UI 加动效） */
  active?: boolean;
  /** 开始时刻（用来在前端递增显示已用时） */
  since?: number;
}

function agoText(ms: number): string {
  const sec = Math.max(0, Math.round(ms / 1000));
  if (sec < 60) return `${sec} 秒前`;
  const min = Math.round(sec / 60);
  if (min < 60) return `${min} 分钟前`;
  return `${Math.round(min / 60)} 小时前`;
}

function fmtDur(ms: number): string {
  const sec = Math.round(ms / 1000);
  if (sec < 60) return `${sec} 秒`;
  const m = Math.floor(sec / 60);
  const s = sec % 60;
  return s ? `${m} 分 ${s} 秒` : `${m} 分钟`;
}