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>(new Map()); /** 每次轮询观测到的翻转 */ private readonly _pulses = signal([]); readonly pulses = this._pulses.asReadonly(); /** 上一轮观测(agentId → 是否在忙 / 时刻 / 任务描述) */ private marks = new Map(); /** 最近一条消息到达时刻(agentId → 毫秒),用于「正在回复」判定 */ private readonly lastMsgAt = new Map(); /** 会话里最后一条消息(含内容),用于识别新消息到达 */ 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} 分钟`; }