Add configurable group session scope (default user_in_chat).

Ops bots now isolate per-speaker Harness sessions in groups, with a
settings/RPC toggle to restore shared chat sessions when needed.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
oliver 2026-09-05 15:39:28 +08:00
parent d10a941fc8
commit 7f1b9bd480
30 changed files with 798 additions and 282 deletions

View file

@ -530,6 +530,7 @@ export class DiscordRuntime {
#state;
#contextEnhancement;
#accessPolicy;
#groupSessionScope;
#logger;
#replyTimeoutMs;
#connectTimeoutMs;
@ -562,6 +563,7 @@ export class DiscordRuntime {
state,
contextEnhancement,
accessPolicy,
groupSessionScope,
logger = console,
replyTimeoutMs = 600_000,
connectTimeoutMs = 20_000,
@ -579,6 +581,7 @@ export class DiscordRuntime {
this.#state = state;
this.#contextEnhancement = contextEnhancement;
this.#accessPolicy = accessPolicy;
this.#groupSessionScope = groupSessionScope;
this.#logger = logger;
this.#replyTimeoutMs = replyTimeoutMs;
this.#connectTimeoutMs = connectTimeoutMs;
@ -659,6 +662,7 @@ export class DiscordRuntime {
state: this.#state,
contextEnhancement: this.#contextEnhancement,
accessPolicy: this.#accessPolicy,
groupSessionScope: this.#groupSessionScope,
status: this.#status,
logger: this.#logger,
replyTimeoutMs: this.#replyTimeoutMs,

View file

@ -17,6 +17,11 @@ import {
normalizeAccessPolicy,
validateAccessPolicy,
} from './access-policy.mjs';
import {
DEFAULT_GROUP_SESSION_SCOPE,
normalizeGroupSessionScope,
validateGroupSessionScope,
} from './session-scope.mjs';
import { CONNECTION_TEST_STATE_IDENTITY } from './connection-test.mjs';
import {
DEFAULT_CONTEXT_ENHANCEMENT_CONFIG,
@ -179,6 +184,19 @@ function normalizeDocument(value) {
}
}
}
let groupSessionScopes = {};
if (value.groupSessionScopes !== undefined) {
if (!value.groupSessionScopes || typeof value.groupSessionScopes !== 'object'
|| Array.isArray(value.groupSessionScopes)) return null;
for (const [botId, scope] of Object.entries(value.groupSessionScopes)) {
if (!/^[A-Za-z0-9_-]{1,128}$/.test(botId)) return null;
try {
groupSessionScopes[botId] = validateGroupSessionScope(scope);
} catch {
return null;
}
}
}
const contextEnhancement = Object.create(null);
// Enhancement damage is isolated from the existing workspace/preset document.
if (value.contextEnhancement && typeof value.contextEnhancement === 'object'
@ -200,6 +218,7 @@ function normalizeDocument(value) {
version,
workspaces,
agentPresets,
groupSessionScopes,
contextEnhancement,
deliveryTargets,
accessPolicies,
@ -210,12 +229,14 @@ function storedDocument({
version,
workspaces,
agentPresets,
groupSessionScopes,
contextEnhancement,
deliveryTargets,
accessPolicies,
}) {
const document = { version, workspaces };
if (Object.keys(agentPresets).length > 0) document.agentPresets = agentPresets;
if (Object.keys(groupSessionScopes).length > 0) document.groupSessionScopes = groupSessionScopes;
if (Object.keys(contextEnhancement).length > 0) {
document.contextEnhancement = contextEnhancement;
}
@ -267,6 +288,7 @@ export class BotWorkspaceStore {
#version = 1;
#workspaces = {};
#agentPresets = {};
#groupSessionScopes = {};
#contextEnhancement = {};
#deliveryTargets = Object.create(null);
#accessPolicies = Object.create(null);
@ -293,6 +315,7 @@ export class BotWorkspaceStore {
this.#version = normalized.version;
this.#workspaces = normalized.workspaces;
this.#agentPresets = normalized.agentPresets;
this.#groupSessionScopes = normalized.groupSessionScopes;
this.#contextEnhancement = normalized.contextEnhancement;
this.#deliveryTargets = normalized.deliveryTargets;
this.#accessPolicies = normalized.accessPolicies;
@ -301,6 +324,7 @@ export class BotWorkspaceStore {
this.#version = 1;
this.#workspaces = {};
this.#agentPresets = {};
this.#groupSessionScopes = {};
this.#contextEnhancement = {};
this.#deliveryTargets = Object.create(null);
this.#accessPolicies = Object.create(null);
@ -335,6 +359,14 @@ export class BotWorkspaceStore {
return this.#agentPresets[botIdOf(botId)] ?? null;
}
groupSessionScopeFor(botId) {
const id = botIdOf(botId);
if (this.has(id) && Object.hasOwn(this.#groupSessionScopes, id)) {
return normalizeGroupSessionScope(this.#groupSessionScopes[id]);
}
return DEFAULT_GROUP_SESSION_SCOPE;
}
contextEnhancementFor(botId) {
const id = botIdOf(botId);
return this.has(id) && Object.hasOwn(this.#contextEnhancement, id)
@ -565,6 +597,38 @@ export class BotWorkspaceStore {
});
}
async setGroupSessionScope(botId, value, { incarnation } = {}) {
const id = botIdOf(botId);
if (!this.has(id)
|| (incarnation !== undefined && incarnation !== this.incarnationFor(id))) {
const error = new Error('找不到要修改的机器人。');
error.code = 'workspace-bot-not-found';
throw error;
}
const scope = validateGroupSessionScope(value);
return this.#enqueue(id, async () => {
if (!this.has(id)
|| (incarnation !== undefined && incarnation !== this.incarnationFor(id))) {
const error = new Error('找不到要修改的机器人。');
error.code = 'workspace-bot-not-found';
throw error;
}
const previous = Object.hasOwn(this.#groupSessionScopes, id)
? this.#groupSessionScopes[id]
: undefined;
if (previous === scope) return scope;
this.#groupSessionScopes[id] = scope;
try {
await this.#persist();
} catch (error) {
if (previous === undefined) delete this.#groupSessionScopes[id];
else this.#groupSessionScopes[id] = previous;
throw error;
}
return scope;
});
}
async setContextEnhancement(botId, value, { incarnation } = {}) {
const id = botIdOf(botId);
const expectedIncarnation = incarnation === undefined ? this.incarnationFor(id) : incarnation;
@ -764,6 +828,7 @@ export class BotWorkspaceStore {
const candidates = new Set([
...Object.keys(this.#workspaces),
...Object.keys(this.#agentPresets),
...Object.keys(this.#groupSessionScopes),
...Object.keys(this.#contextEnhancement),
...Object.keys(this.#deliveryTargets),
...Object.keys(this.#accessPolicies),
@ -783,6 +848,7 @@ export class BotWorkspaceStore {
...bot,
workspace: this.workspaceFor(bot.botId),
agentPreset: this.agentPresetFor(bot.botId),
groupSessionScope: this.groupSessionScopeFor(bot.botId),
contextEnhancement: this.contextEnhancementFor(bot.botId),
accessPolicy: this.accessPolicyFor(bot.botId),
}
@ -814,13 +880,15 @@ export class BotWorkspaceStore {
async #retireCurrentIncarnation(id) {
const hadWorkspace = Object.hasOwn(this.#workspaces, id);
const hadPreset = Object.hasOwn(this.#agentPresets, id);
const hadGroupSessionScope = Object.hasOwn(this.#groupSessionScopes, id);
const hadContextEnhancement = Object.hasOwn(this.#contextEnhancement, id);
const hadDeliveryTargets = Object.hasOwn(this.#deliveryTargets, id);
const hadAccessPolicy = Object.hasOwn(this.#accessPolicies, id);
const needsCleanup = hadWorkspace || hadPreset || hadContextEnhancement
const needsCleanup = hadWorkspace || hadPreset || hadGroupSessionScope || hadContextEnhancement
|| hadDeliveryTargets || hadAccessPolicy || this.#dirtyRemovals.has(id);
delete this.#workspaces[id];
delete this.#agentPresets[id];
delete this.#groupSessionScopes[id];
delete this.#contextEnhancement[id];
delete this.#deliveryTargets[id];
delete this.#accessPolicies[id];
@ -863,6 +931,7 @@ export class BotWorkspaceStore {
version,
workspaces: this.#workspaces,
agentPresets: this.#agentPresets,
groupSessionScopes: this.#groupSessionScopes,
contextEnhancement,
deliveryTargets,
accessPolicies,
@ -873,6 +942,7 @@ export class BotWorkspaceStore {
async #persistCurrentDocument() {
if (Object.keys(this.#workspaces).length > 0
|| Object.keys(this.#agentPresets).length > 0
|| Object.keys(this.#groupSessionScopes).length > 0
|| Object.keys(this.#contextEnhancement).length > 0
|| Object.keys(this.#deliveryTargets).length > 0
|| Object.keys(this.#accessPolicies).length > 0) {
@ -1427,6 +1497,30 @@ export function createWorkspaceAwareController(controller, { workspaces, stateFo
return result;
});
};
const updateGroupSessionScope = (botId, value, projectStatus) => {
const incarnation = workspaces.incarnationFor(botId);
const scope = validateGroupSessionScope(value);
return withBotTransition(botId, async () => {
const snapshot = await controller.status();
if (!snapshot?.bots?.some((bot) => bot?.botId === botId)) {
const error = new Error('找不到要修改的机器人。');
error.code = 'workspace-bot-not-found';
throw error;
}
const catalog = await resolveAgentPresetCatalog(agentPresetCatalog);
const decorated = workspaces.decorateStatus(snapshot);
const updated = {
...decorated,
bots: decorated.bots.map((bot) => bot?.botId === botId
? { ...bot, groupSessionScope: scope } : bot),
...(catalog ? { agentPresetCatalog: catalog } : {}),
};
const result = projectStatus ? await projectStatus(updated) : updated;
await workspaces.setGroupSessionScope(botId, scope, { incarnation });
return result;
});
};
const deleteWithWorkspace = (botId, invokeDelete) => withBotTransition(botId, async () => {
// Fence the old runtime without changing the durable mapping. A crash
// before the controller removes its config therefore keeps the bot's
@ -1468,6 +1562,7 @@ export function createWorkspaceAwareController(controller, { workspaces, stateFo
if (property === 'updateAgentPreset') return updateAgentPreset;
if (property === 'updateContextEnhancement') return updateContextEnhancement;
if (property === 'updateAccessPolicy') return updateAccessPolicy;
if (property === 'updateGroupSessionScope') return updateGroupSessionScope;
const value = Reflect.get(target, property, target);
if (typeof value !== 'function') return value;
if (property === 'deleteBot') {

View file

@ -0,0 +1,61 @@
/**
* Group conversation → Harness Session key policy (oclaw-aligned).
*
* - `chat`: one Session per group (`group:<conversationId>`) — upstream dsh-im default
* - `user_in_chat`: one Session per speaker in a group (`group:<conversationId>:user:<senderId>`)
*/
export const GROUP_SESSION_SCOPES = Object.freeze(['chat', 'user_in_chat']);
/** Ops-fork default: isolate concurrent operators in the same WhatsApp/ops group. */
export const DEFAULT_GROUP_SESSION_SCOPE = 'user_in_chat';
/**
* @param {unknown} value
* @returns {'chat' | 'user_in_chat'}
*/
export function normalizeGroupSessionScope(value) {
const scope = String(value ?? '').trim().toLowerCase();
if (scope === 'chat' || scope === 'shared' || scope === 'shared_chat') return 'chat';
if (scope === 'user_in_chat' || scope === 'user' || scope === 'per_user' || scope === 'member') {
return 'user_in_chat';
}
return DEFAULT_GROUP_SESSION_SCOPE;
}
/**
* @param {unknown} value
* @returns {'chat' | 'user_in_chat'}
*/
export function validateGroupSessionScope(value) {
const scope = String(value ?? '').trim().toLowerCase();
if (scope === 'chat' || scope === 'user_in_chat') return scope;
const error = new TypeError(`Invalid group session scope: ${String(value)}`);
error.code = 'invalid-group-session-scope';
throw error;
}
/**
* Build the durable conversation key used for session binding and queues.
* @param {{ kind: string, conversationId: string, senderId: string, groupSessionScope?: string }} input
* @returns {string}
*/
export function conversationKeyFor({
kind,
conversationId,
senderId,
groupSessionScope = DEFAULT_GROUP_SESSION_SCOPE,
}) {
const conversation = String(conversationId ?? '').trim();
const sender = String(senderId ?? '').trim();
if (!conversation) throw new TypeError('conversationId is required');
if (kind === 'group') {
const scope = normalizeGroupSessionScope(groupSessionScope);
if (scope === 'user_in_chat') {
if (!sender) throw new TypeError('senderId is required for user_in_chat group sessions');
return `group:${conversation}:user:${sender}`;
}
return `group:${conversation}`;
}
return `direct:${conversation}`;
}

View file

@ -62,6 +62,11 @@ import {
setLastMessageFailure,
} from './message-failure.mjs';
import { beginStatusReaction } from './status-reaction.mjs';
import {
conversationKeyFor,
DEFAULT_GROUP_SESSION_SCOPE,
normalizeGroupSessionScope,
} from './session-scope.mjs';
const INTERACTION_RESOLVED_TEXT = '这个问题已在其他客户端处理,无需再次回答。';
const FILE_ONLY_COMPLETION_TEXT = '任务已完成。';
@ -137,6 +142,8 @@ export class TextHarnessBridge {
#state;
#contextEnhancement;
#accessPolicy;
/** @type {string | { getScope?: () => string }} */
#groupSessionScope;
#status;
#logger;
#replyTimeoutMs;
@ -158,6 +165,7 @@ export class TextHarnessBridge {
state,
contextEnhancement,
accessPolicy,
groupSessionScope = DEFAULT_GROUP_SESSION_SCOPE,
status = createTextBridgeStatus(),
logger = console,
replyTimeoutMs = 600_000,
@ -172,6 +180,7 @@ export class TextHarnessBridge {
this.#state = state;
this.#contextEnhancement = contextEnhancement;
this.#accessPolicy = accessPolicy;
this.#groupSessionScope = groupSessionScope;
this.#status = status;
this.#logger = logger;
this.#replyTimeoutMs = replyTimeoutMs;
@ -186,6 +195,15 @@ export class TextHarnessBridge {
return structuredClone(this.#status);
}
#resolveGroupSessionScope() {
const configured = this.#groupSessionScope;
if (configured && typeof configured === 'object'
&& typeof configured.getScope === 'function') {
return normalizeGroupSessionScope(configured.getScope());
}
return normalizeGroupSessionScope(configured);
}
accept(message, { contextSnapshot, accessDecision } = {}) {
if (this.#signal?.aborted) return Promise.resolve();
const conversationId = cleanText(message?.conversationId);
@ -262,7 +280,12 @@ export class TextHarnessBridge {
);
}
const key = `${normalized.kind}:${normalized.conversationId}`;
const key = conversationKeyFor({
kind: normalized.kind,
conversationId: normalized.conversationId,
senderId,
groupSessionScope: this.#resolveGroupSessionScope(),
});
const pending = this.#pendingInteractions.get(key);
const text = cleanText(normalized.content);
const batchCommand = isBatchInputCommand(text);

View file

@ -407,6 +407,7 @@ export class SlackRuntime {
#state;
#contextEnhancement;
#accessPolicy;
#groupSessionScope;
#logger;
#replyTimeoutMs;
#connectTimeoutMs;
@ -432,6 +433,7 @@ export class SlackRuntime {
state,
contextEnhancement,
accessPolicy,
groupSessionScope,
logger = console,
replyTimeoutMs = 600_000,
connectTimeoutMs = 20_000,
@ -449,6 +451,7 @@ export class SlackRuntime {
this.#state = state;
this.#contextEnhancement = contextEnhancement;
this.#accessPolicy = accessPolicy;
this.#groupSessionScope = groupSessionScope;
this.#logger = logger;
this.#replyTimeoutMs = replyTimeoutMs;
this.#connectTimeoutMs = connectTimeoutMs;
@ -527,6 +530,7 @@ export class SlackRuntime {
state: this.#state,
contextEnhancement: this.#contextEnhancement,
accessPolicy: this.#accessPolicy,
groupSessionScope: this.#groupSessionScope,
status: this.#status,
logger: this.#logger,
replyTimeoutMs: this.#replyTimeoutMs,

View file

@ -768,6 +768,7 @@ export class TelegramRuntime {
#state;
#contextEnhancement;
#accessPolicy;
#groupSessionScope;
#logger;
#replyTimeoutMs;
#createApi;
@ -787,6 +788,7 @@ export class TelegramRuntime {
state,
contextEnhancement,
accessPolicy,
groupSessionScope,
logger = console,
replyTimeoutMs = 600_000,
createApi = (options) => new TelegramApi(options),
@ -801,6 +803,7 @@ export class TelegramRuntime {
this.#state = state;
this.#contextEnhancement = contextEnhancement;
this.#accessPolicy = accessPolicy;
this.#groupSessionScope = groupSessionScope;
this.#logger = logger;
this.#replyTimeoutMs = replyTimeoutMs;
this.#createApi = createApi;
@ -902,6 +905,7 @@ export class TelegramRuntime {
state: this.#state,
contextEnhancement: this.#contextEnhancement,
accessPolicy: this.#accessPolicy,
groupSessionScope: this.#groupSessionScope,
status: this.#status,
logger: this.#logger,
replyTimeoutMs: this.#replyTimeoutMs,

View file

@ -625,6 +625,7 @@ export class WhatsappRuntime {
#state;
#contextEnhancement;
#accessPolicy;
#groupSessionScope;
#logger;
#replyTimeoutMs;
#connectTimeoutMs;
@ -644,6 +645,7 @@ export class WhatsappRuntime {
state,
contextEnhancement,
accessPolicy,
groupSessionScope,
logger = console,
replyTimeoutMs = 600_000,
connectTimeoutMs = 30_000,
@ -659,6 +661,7 @@ export class WhatsappRuntime {
this.#state = state;
this.#contextEnhancement = contextEnhancement;
this.#accessPolicy = accessPolicy;
this.#groupSessionScope = groupSessionScope;
this.#logger = logger;
this.#replyTimeoutMs = replyTimeoutMs;
this.#connectTimeoutMs = connectTimeoutMs;
@ -756,6 +759,7 @@ export class WhatsappRuntime {
} : {}),
equals: whatsappAccessPolicyIdsEqual,
} : undefined,
groupSessionScope: this.#groupSessionScope,
status: this.#status,
logger: this.#logger,
replyTimeoutMs: this.#replyTimeoutMs,