feat(feishu): persistent session watches and completion pushes

- /watch resolves the target READ-ONLY: sessions are validated against
  registered workspace listings (current or any other) without binding the
  conversation or switching workspaces. /unwatch and /watchlist round out
  the command set; the watchlist card supports button and number-reply
  unwatching.
- Watches persist in the state store (sessionId + title + chatId + lastSeq)
  and resume at runtime start: the bridge starts the global event-mux
  watcher in its constructor when the harness supports it.
- Completion pushes fire on turn/end, deduped by sessionId + turn id, with
  the seq watermark persisted per watch. After a mux reconnect the bridge
  replays each watched session's recent history (session.history) through
  the normal handler, so missed turn/end events are compensated without
  duplicates.
- Bound-session push targets are persisted (chatTargets) instead of living
  only in memory, and refresh on every accepted message.
- Watch failures map to safe user-facing messages.
This commit is contained in:
liuwenbo00 2026-08-21 10:10:36 +08:00 • committed by xmanrui
parent 0579c24b13
commit beb269dc0e
6 changed files with 746 additions and 132 deletions

View file

@ -2404,3 +2404,142 @@ test('repair monitor reports expiry without claiming that the callback was fixed
.find((text) => text.includes('授权链接已过期'));
assert.doesNotMatch(terminal, /修复完成/);
});
// ── Watches: read-only tracking, persistence, compensation, dedup ─────────
import { StateStore } from '../../../src/channels/feishu/state-store.mjs';
function watchHarness({ sessionsByWorkspace = { 'C:/work': [] }, current = 'C:/work', history = [] } = {}) {
const listeners = [];
return {
ensureRunning: async () => true,
currentWorkspace: () => current,
listWorkspaces: async () => Object.keys(sessionsByWorkspace),
listWorkspaceSessions: async (workspace) => ({ workspace, sessions: sessionsByWorkspace[workspace] ?? [] }),
bindWorkspaceSession: async (_key, sessionId) => ({ sessionId, title: `Title ${sessionId}` }),
switchWorkspace: async (path) => path,
rpc: async (method, params) => (method === 'session.history' ? { events: history } : null),
watchHarnessEvents: ({ onSessionEvent, onReconnect }) => {
listeners.push({ onSessionEvent, onReconnect });
return Promise.resolve();
},
_listeners: listeners,
};
}
async function watchStoreFixture(seedSessions = []) {
const store = new StateStore(join(tmpdir(), `dsh-im-watch-test-${Math.random().toString(36).slice(2)}.json`));
await store.load();
for (const [key, sessionId] of seedSessions) await store.setSession(key, sessionId);
return { store, state: store };
}
test('/watch resolves read-only: no binding, no workspace switch', async () => {
const { state } = await watchStoreFixture([['p2p:ou_owner', 'bound-session']]);
let bindCalls = 0;
let switchCalls = 0;
const harness = watchHarness({
sessionsByWorkspace: { 'C:/work': [{ sessionId: 'target-session', title: 'Target' }] },
});
harness.bindWorkspaceSession = async () => { bindCalls += 1; throw new Error('must not bind'); };
harness.switchWorkspace = async () => { switchCalls += 1; throw new Error('must not switch'); };
const sent = [];
const bridge = new FeishuHarnessBridge({
client: textClient(async ({ text }) => sent.push(text)),
channel: {},
harness,
state,
status: bridgeStatus(),
allowedSenderOpenIds: new Set(['ou_owner']),
});
await bridge.accept(event('watch-1', '/watch 1', { senderOpenId: 'ou_owner' }));
await bridge.waitForIdle();
assert.match(sent.at(-1), /已关注会话「Target」/);
assert.equal(bindCalls, 0, 'watch must not bind the conversation');
assert.equal(switchCalls, 0, 'watch must not switch workspaces');
assert.equal(state.sessionFor('p2p:ou_owner'), 'bound-session', 'existing binding unchanged');
const entry = state.watchEntry('p2p:ou_owner', 'target-session');
assert.ok(entry, 'watch entry persisted');
assert.equal(entry.chatId, 'oc_chat');
});
test('/watch finds a session in another workspace without switching', async () => {
const { state } = await watchStoreFixture();
const harness = watchHarness({
current: 'C:/work',
sessionsByWorkspace: {
'C:/work': [],
'D:/other': [{ sessionId: 'other-session', title: 'Other Session' }],
},
});
const sent = [];
const bridge = new FeishuHarnessBridge({
client: textClient(async ({ text }) => sent.push(text)),
channel: {},
harness,
state,
status: bridgeStatus(),
allowedSenderOpenIds: new Set(['ou_owner']),
});
await bridge.accept(event('watch-x', '/watch other-session', { senderOpenId: 'ou_owner' }));
await bridge.waitForIdle();
assert.match(sent.at(-1), /已关注会话「Other Session」/);
assert.equal(state.sessionFor('p2p:ou_owner'), null, 'cross-workspace watch must not bind');
assert.ok(state.watchEntry('p2p:ou_owner', 'other-session'));
});
test('persisted watches resume the event watcher at runtime start', async () => {
const { state } = await watchStoreFixture();
await state.setWatch('p2p:ou_owner', { sessionId: 'kept-session', title: 'Kept', chatId: 'oc_chat', lastSeq: 3 });
const harness = watchHarness();
const bridge = new FeishuHarnessBridge({
client: textClient(async () => {}),
channel: {},
harness,
state,
status: bridgeStatus(),
allowedSenderOpenIds: new Set(['ou_owner']),
});
await new Promise((resolve) => setTimeout(resolve, 10));
assert.equal(harness._listeners.length, 1, 'watcher must restart from persisted state');
assert.ok(bridge);
});
test('reconnect compensation replays missed turn/end and dedups duplicates', async () => {
const { state } = await watchStoreFixture();
await state.setWatch('p2p:ou_owner', { sessionId: 'watched-session', title: 'Watched', chatId: 'oc_chat', lastSeq: null });
const harness = watchHarness({
sessionsByWorkspace: { 'C:/work': [{ sessionId: 'watched-session', title: 'Watched' }] },
history: [
{ type: 'turn/end', seq: 10, data: { turn: 't1', reason: { kind: 'completed' } } },
{ type: 'turn/end', seq: 11, data: { turn: 't2', reason: { kind: 'completed' } } },
],
});
const cards = [];
const bridge = new FeishuHarnessBridge({
client: cardClient(async ({ msgType, content }) => {
if (msgType === 'interactive') cards.push(content);
}),
channel: {},
harness,
state,
status: bridgeStatus(),
allowedSenderOpenIds: new Set(['ou_owner']),
});
await new Promise((resolve) => setTimeout(resolve, 10));
assert.equal(harness._listeners.length, 1);
// Live event: t1 arrives before any replay.
harness._listeners[0].onSessionEvent({ sessionId: 'watched-session', event: { type: 'turn/end', seq: 10, data: { turn: 't1', reason: { kind: 'completed' } } } });
await bridge.waitForIdle();
assert.equal(cards.length, 1, 'one completion for the live event');
// Reconnect: history replays t1 (dedup) and t2 (new).
await harness._listeners[0].onReconnect();
await new Promise((resolve) => setTimeout(resolve, 20));
assert.equal(cards.length, 2, 'exactly one new completion from compensation (t1 deduped)');
assert.match(JSON.stringify(cards[1]), /watched-session/);
});