Route IPC send_message by paired room role
This commit is contained in:
@@ -669,6 +669,7 @@ async function runQuery(
|
|||||||
EJCLAW_IS_MAIN: containerInput.isMain ? '1' : '0',
|
EJCLAW_IS_MAIN: containerInput.isMain ? '1' : '0',
|
||||||
EJCLAW_AGENT_TYPE:
|
EJCLAW_AGENT_TYPE:
|
||||||
process.env.EJCLAW_AGENT_TYPE || 'claude-code',
|
process.env.EJCLAW_AGENT_TYPE || 'claude-code',
|
||||||
|
EJCLAW_ROOM_ROLE: containerInput.roomRoleContext?.role || '',
|
||||||
...(process.env.EJCLAW_IPC_DIR && {
|
...(process.env.EJCLAW_IPC_DIR && {
|
||||||
EJCLAW_IPC_DIR: process.env.EJCLAW_IPC_DIR,
|
EJCLAW_IPC_DIR: process.env.EJCLAW_IPC_DIR,
|
||||||
}),
|
}),
|
||||||
|
|||||||
@@ -30,6 +30,7 @@ import {
|
|||||||
waitForVerificationResponse,
|
waitForVerificationResponse,
|
||||||
} from './verification.js';
|
} from './verification.js';
|
||||||
import { resolveIpcDirectories } from './ipc-paths.js';
|
import { resolveIpcDirectories } from './ipc-paths.js';
|
||||||
|
import { buildSendMessageIpcPayload } from './ipc-message.js';
|
||||||
|
|
||||||
const { ipcDir: IPC_DIR, hostIpcDir: HOST_IPC_DIR } =
|
const { ipcDir: IPC_DIR, hostIpcDir: HOST_IPC_DIR } =
|
||||||
resolveIpcDirectories(process.env);
|
resolveIpcDirectories(process.env);
|
||||||
@@ -76,14 +77,13 @@ server.tool(
|
|||||||
sender: z.string().optional().describe('Your role/identity name (e.g. "Researcher"). When set, messages appear from a dedicated bot in Telegram.'),
|
sender: z.string().optional().describe('Your role/identity name (e.g. "Researcher"). When set, messages appear from a dedicated bot in Telegram.'),
|
||||||
},
|
},
|
||||||
async (args) => {
|
async (args) => {
|
||||||
const data: Record<string, string | undefined> = {
|
const data = buildSendMessageIpcPayload({
|
||||||
type: 'message',
|
|
||||||
chatJid,
|
chatJid,
|
||||||
text: args.text,
|
text: args.text,
|
||||||
sender: args.sender || undefined,
|
sender: args.sender || undefined,
|
||||||
|
senderRole: process.env.EJCLAW_ROOM_ROLE || undefined,
|
||||||
groupFolder,
|
groupFolder,
|
||||||
timestamp: new Date().toISOString(),
|
});
|
||||||
};
|
|
||||||
|
|
||||||
writeIpcFile(MESSAGES_DIR, data);
|
writeIpcFile(MESSAGES_DIR, data);
|
||||||
|
|
||||||
|
|||||||
22
runners/agent-runner/src/ipc-message.ts
Normal file
22
runners/agent-runner/src/ipc-message.ts
Normal file
@@ -0,0 +1,22 @@
|
|||||||
|
export interface SendMessageIpcPayloadInput {
|
||||||
|
chatJid: string;
|
||||||
|
text: string;
|
||||||
|
sender?: string;
|
||||||
|
senderRole?: string;
|
||||||
|
groupFolder: string;
|
||||||
|
timestamp?: string;
|
||||||
|
}
|
||||||
|
|
||||||
|
export function buildSendMessageIpcPayload(
|
||||||
|
input: SendMessageIpcPayloadInput,
|
||||||
|
): Record<string, string | undefined> {
|
||||||
|
return {
|
||||||
|
type: 'message',
|
||||||
|
chatJid: input.chatJid,
|
||||||
|
text: input.text,
|
||||||
|
sender: input.sender || undefined,
|
||||||
|
senderRole: input.senderRole || undefined,
|
||||||
|
groupFolder: input.groupFolder,
|
||||||
|
timestamp: input.timestamp || new Date().toISOString(),
|
||||||
|
};
|
||||||
|
}
|
||||||
38
runners/agent-runner/test/ipc-message.test.ts
Normal file
38
runners/agent-runner/test/ipc-message.test.ts
Normal file
@@ -0,0 +1,38 @@
|
|||||||
|
import { describe, expect, it } from 'vitest';
|
||||||
|
|
||||||
|
import { buildSendMessageIpcPayload } from '../src/ipc-message.js';
|
||||||
|
|
||||||
|
describe('agent runner IPC message payload', () => {
|
||||||
|
it('includes senderRole when provided', () => {
|
||||||
|
expect(
|
||||||
|
buildSendMessageIpcPayload({
|
||||||
|
chatJid: 'dc:123',
|
||||||
|
text: 'hello',
|
||||||
|
sender: 'Reviewer',
|
||||||
|
senderRole: 'reviewer',
|
||||||
|
groupFolder: 'discord-review',
|
||||||
|
timestamp: '2026-04-04T13:45:00.000Z',
|
||||||
|
}),
|
||||||
|
).toEqual({
|
||||||
|
type: 'message',
|
||||||
|
chatJid: 'dc:123',
|
||||||
|
text: 'hello',
|
||||||
|
sender: 'Reviewer',
|
||||||
|
senderRole: 'reviewer',
|
||||||
|
groupFolder: 'discord-review',
|
||||||
|
timestamp: '2026-04-04T13:45:00.000Z',
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|
||||||
|
it('omits empty senderRole values', () => {
|
||||||
|
expect(
|
||||||
|
buildSendMessageIpcPayload({
|
||||||
|
chatJid: 'dc:123',
|
||||||
|
text: 'hello',
|
||||||
|
senderRole: '',
|
||||||
|
groupFolder: 'discord-review',
|
||||||
|
timestamp: '2026-04-04T13:45:00.000Z',
|
||||||
|
}).senderRole,
|
||||||
|
).toBeUndefined();
|
||||||
|
});
|
||||||
|
});
|
||||||
@@ -46,6 +46,7 @@ import { startIpcWatcher } from './ipc.js';
|
|||||||
import {
|
import {
|
||||||
findChannel,
|
findChannel,
|
||||||
findChannelByName,
|
findChannelByName,
|
||||||
|
findChannelForDeliveryRole,
|
||||||
formatOutbound,
|
formatOutbound,
|
||||||
normalizeMessageForDedupe,
|
normalizeMessageForDedupe,
|
||||||
} from './router.js';
|
} from './router.js';
|
||||||
@@ -471,8 +472,8 @@ async function main(): Promise<void> {
|
|||||||
editFormattedTrackedChannelMessage(channels, jid, messageId, rawText),
|
editFormattedTrackedChannelMessage(channels, jid, messageId, rawText),
|
||||||
});
|
});
|
||||||
startIpcWatcher({
|
startIpcWatcher({
|
||||||
sendMessage: (jid, text) => {
|
sendMessage: (jid, text, senderRole) => {
|
||||||
const channel = findChannel(channels, jid);
|
const channel = findChannelForDeliveryRole(channels, jid, senderRole);
|
||||||
if (!channel) throw new Error(`No channel for JID: ${jid}`);
|
if (!channel) throw new Error(`No channel for JID: ${jid}`);
|
||||||
return channel.sendMessage(jid, text);
|
return channel.sendMessage(jid, text);
|
||||||
},
|
},
|
||||||
|
|||||||
@@ -13,7 +13,7 @@ import {
|
|||||||
getStoredRoomSettings,
|
getStoredRoomSettings,
|
||||||
getTaskById,
|
getTaskById,
|
||||||
} from './db.js';
|
} from './db.js';
|
||||||
import { processTaskIpc, IpcDeps } from './ipc.js';
|
import { forwardAuthorizedIpcMessage, processTaskIpc, IpcDeps } from './ipc.js';
|
||||||
import { DEFAULT_WATCH_CI_MAX_DURATION_MS } from './task-watch-status.js';
|
import { DEFAULT_WATCH_CI_MAX_DURATION_MS } from './task-watch-status.js';
|
||||||
import { RegisteredGroup } from './types.js';
|
import { RegisteredGroup } from './types.js';
|
||||||
|
|
||||||
@@ -536,6 +536,49 @@ describe('IPC message authorization', () => {
|
|||||||
isMessageAuthorized('whatsapp_main', true, 'unknown@g.us', groups),
|
isMessageAuthorized('whatsapp_main', true, 'unknown@g.us', groups),
|
||||||
).toBe(true);
|
).toBe(true);
|
||||||
});
|
});
|
||||||
|
|
||||||
|
it('forwards senderRole through authorized IPC messages', async () => {
|
||||||
|
const sendMessage = vi.fn(async () => {});
|
||||||
|
|
||||||
|
await forwardAuthorizedIpcMessage(
|
||||||
|
{
|
||||||
|
type: 'message',
|
||||||
|
chatJid: 'other@g.us',
|
||||||
|
text: 'review text',
|
||||||
|
senderRole: 'reviewer',
|
||||||
|
},
|
||||||
|
'other-group',
|
||||||
|
false,
|
||||||
|
groups,
|
||||||
|
sendMessage,
|
||||||
|
);
|
||||||
|
|
||||||
|
expect(sendMessage).toHaveBeenCalledWith(
|
||||||
|
'other@g.us',
|
||||||
|
'review text',
|
||||||
|
'reviewer',
|
||||||
|
);
|
||||||
|
});
|
||||||
|
|
||||||
|
it('does not forward unauthorized IPC messages even when senderRole exists', async () => {
|
||||||
|
const sendMessage = vi.fn(async () => {});
|
||||||
|
|
||||||
|
const result = await forwardAuthorizedIpcMessage(
|
||||||
|
{
|
||||||
|
type: 'message',
|
||||||
|
chatJid: 'third@g.us',
|
||||||
|
text: 'review text',
|
||||||
|
senderRole: 'reviewer',
|
||||||
|
},
|
||||||
|
'other-group',
|
||||||
|
false,
|
||||||
|
groups,
|
||||||
|
sendMessage,
|
||||||
|
);
|
||||||
|
|
||||||
|
expect(result.outcome).toBe('blocked');
|
||||||
|
expect(sendMessage).not.toHaveBeenCalled();
|
||||||
|
});
|
||||||
});
|
});
|
||||||
|
|
||||||
// --- schedule_task with cron and interval types ---
|
// --- schedule_task with cron and interval types ---
|
||||||
|
|||||||
119
src/ipc.ts
119
src/ipc.ts
@@ -34,7 +34,11 @@ import {
|
|||||||
import { AgentType, RegisteredGroup, RoomMode } from './types.js';
|
import { AgentType, RegisteredGroup, RoomMode } from './types.js';
|
||||||
|
|
||||||
export interface IpcDeps {
|
export interface IpcDeps {
|
||||||
sendMessage: (jid: string, text: string) => Promise<void>;
|
sendMessage: (
|
||||||
|
jid: string,
|
||||||
|
text: string,
|
||||||
|
senderRole?: string,
|
||||||
|
) => Promise<void>;
|
||||||
nudgeScheduler?: () => void;
|
nudgeScheduler?: () => void;
|
||||||
registeredGroups: () => Record<string, RegisteredGroup>;
|
registeredGroups: () => Record<string, RegisteredGroup>;
|
||||||
assignRoom: (jid: string, room: AssignRoomInput) => void;
|
assignRoom: (jid: string, room: AssignRoomInput) => void;
|
||||||
@@ -47,6 +51,53 @@ export interface IpcDeps {
|
|||||||
) => void;
|
) => void;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
export interface IpcMessagePayload {
|
||||||
|
type?: string;
|
||||||
|
chatJid?: string;
|
||||||
|
text?: string;
|
||||||
|
senderRole?: string;
|
||||||
|
}
|
||||||
|
|
||||||
|
export interface IpcMessageForwardResult {
|
||||||
|
outcome: 'ignored' | 'sent' | 'blocked';
|
||||||
|
chatJid?: string;
|
||||||
|
targetGroup?: string | null;
|
||||||
|
isMainOverride?: boolean;
|
||||||
|
}
|
||||||
|
|
||||||
|
export async function forwardAuthorizedIpcMessage(
|
||||||
|
msg: IpcMessagePayload,
|
||||||
|
sourceGroup: string,
|
||||||
|
isMain: boolean,
|
||||||
|
registeredGroups: Record<string, RegisteredGroup>,
|
||||||
|
sendMessage: IpcDeps['sendMessage'],
|
||||||
|
): Promise<IpcMessageForwardResult> {
|
||||||
|
if (!(msg.type === 'message' && msg.chatJid && msg.text)) {
|
||||||
|
return { outcome: 'ignored' };
|
||||||
|
}
|
||||||
|
|
||||||
|
const targetGroup = registeredGroups[msg.chatJid];
|
||||||
|
const isMainOverride = isMain === true;
|
||||||
|
if (
|
||||||
|
!(isMainOverride || (targetGroup && targetGroup.folder === sourceGroup))
|
||||||
|
) {
|
||||||
|
return {
|
||||||
|
outcome: 'blocked',
|
||||||
|
chatJid: msg.chatJid,
|
||||||
|
targetGroup: targetGroup?.folder ?? null,
|
||||||
|
isMainOverride,
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
|
await sendMessage(msg.chatJid, msg.text, msg.senderRole);
|
||||||
|
return {
|
||||||
|
outcome: 'sent',
|
||||||
|
chatJid: msg.chatJid,
|
||||||
|
targetGroup: targetGroup?.folder ?? null,
|
||||||
|
isMainOverride,
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
let ipcWatcherRunning = false;
|
let ipcWatcherRunning = false;
|
||||||
const IPC_PROCESSING_DIRNAME = '.processing';
|
const IPC_PROCESSING_DIRNAME = '.processing';
|
||||||
|
|
||||||
@@ -186,42 +237,36 @@ export function startIpcWatcher(deps: IpcDeps): void {
|
|||||||
const data = readJsonFile(claimedPath);
|
const data = readJsonFile(claimedPath);
|
||||||
if (!data || typeof data !== 'object')
|
if (!data || typeof data !== 'object')
|
||||||
throw new Error('Invalid JSON');
|
throw new Error('Invalid JSON');
|
||||||
const msg = data as {
|
const msg = data as IpcMessagePayload;
|
||||||
type?: string;
|
const forwardResult = await forwardAuthorizedIpcMessage(
|
||||||
chatJid?: string;
|
msg,
|
||||||
text?: string;
|
sourceGroup,
|
||||||
};
|
isMain,
|
||||||
if (msg.type === 'message' && msg.chatJid && msg.text) {
|
registeredGroups,
|
||||||
// Authorization: verify this group can send to this chatJid
|
deps.sendMessage,
|
||||||
const targetGroup = registeredGroups[msg.chatJid];
|
);
|
||||||
const isMainOverride = isMain === true;
|
if (forwardResult.outcome === 'sent') {
|
||||||
if (
|
logger.info(
|
||||||
isMainOverride ||
|
{
|
||||||
(targetGroup && targetGroup.folder === sourceGroup)
|
transition: 'ipc:auth:allow',
|
||||||
) {
|
chatJid: forwardResult.chatJid,
|
||||||
await deps.sendMessage(msg.chatJid, msg.text);
|
sourceGroup,
|
||||||
logger.info(
|
targetGroup: forwardResult.targetGroup ?? null,
|
||||||
{
|
isMainOverride: forwardResult.isMainOverride,
|
||||||
transition: 'ipc:auth:allow',
|
},
|
||||||
chatJid: msg.chatJid,
|
'IPC message sent',
|
||||||
sourceGroup,
|
);
|
||||||
targetGroup: targetGroup?.folder ?? null,
|
} else if (forwardResult.outcome === 'blocked') {
|
||||||
isMainOverride,
|
logger.warn(
|
||||||
},
|
{
|
||||||
'IPC message sent',
|
transition: 'ipc:auth:deny',
|
||||||
);
|
chatJid: forwardResult.chatJid,
|
||||||
} else {
|
sourceGroup,
|
||||||
logger.warn(
|
targetGroup: forwardResult.targetGroup ?? null,
|
||||||
{
|
isMainOverride: forwardResult.isMainOverride,
|
||||||
transition: 'ipc:auth:deny',
|
},
|
||||||
chatJid: msg.chatJid,
|
'Unauthorized IPC message attempt blocked',
|
||||||
sourceGroup,
|
);
|
||||||
targetGroup: targetGroup?.folder ?? null,
|
|
||||||
isMainOverride,
|
|
||||||
},
|
|
||||||
'Unauthorized IPC message attempt blocked',
|
|
||||||
);
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
fs.unlinkSync(claimedPath);
|
fs.unlinkSync(claimedPath);
|
||||||
} catch (err) {
|
} catch (err) {
|
||||||
|
|||||||
54
src/router.test.ts
Normal file
54
src/router.test.ts
Normal file
@@ -0,0 +1,54 @@
|
|||||||
|
import { describe, expect, it } from 'vitest';
|
||||||
|
|
||||||
|
import { findChannelForDeliveryRole } from './router.js';
|
||||||
|
import { type Channel } from './types.js';
|
||||||
|
|
||||||
|
function createChannel(name: string, ownedJids: string[]): Channel {
|
||||||
|
return {
|
||||||
|
name,
|
||||||
|
connect: async () => {},
|
||||||
|
sendMessage: async () => {},
|
||||||
|
isConnected: () => true,
|
||||||
|
ownsJid: (jid) => ownedJids.includes(jid),
|
||||||
|
disconnect: async () => {},
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
|
describe('findChannelForDeliveryRole', () => {
|
||||||
|
const jid = 'dc:123';
|
||||||
|
|
||||||
|
it('prefers the reviewer channel for reviewer IPC messages', () => {
|
||||||
|
const ownerChannel = createChannel('discord-main', [jid]);
|
||||||
|
const reviewerChannel = createChannel('discord-review', [jid]);
|
||||||
|
|
||||||
|
expect(
|
||||||
|
findChannelForDeliveryRole(
|
||||||
|
[ownerChannel, reviewerChannel],
|
||||||
|
jid,
|
||||||
|
'reviewer',
|
||||||
|
),
|
||||||
|
).toBe(reviewerChannel);
|
||||||
|
});
|
||||||
|
|
||||||
|
it('prefers the arbiter channel for arbiter IPC messages', () => {
|
||||||
|
const ownerChannel = createChannel('discord-main', [jid]);
|
||||||
|
const reviewerChannel = createChannel('discord-review', [jid]);
|
||||||
|
const arbiterChannel = createChannel('discord-arbiter', [jid]);
|
||||||
|
|
||||||
|
expect(
|
||||||
|
findChannelForDeliveryRole(
|
||||||
|
[ownerChannel, reviewerChannel, arbiterChannel],
|
||||||
|
jid,
|
||||||
|
'arbiter',
|
||||||
|
),
|
||||||
|
).toBe(arbiterChannel);
|
||||||
|
});
|
||||||
|
|
||||||
|
it('falls back to the first JID-owning channel when no role channel exists', () => {
|
||||||
|
const ownerChannel = createChannel('discord-main', [jid]);
|
||||||
|
|
||||||
|
expect(findChannelForDeliveryRole([ownerChannel], jid, 'reviewer')).toBe(
|
||||||
|
ownerChannel,
|
||||||
|
);
|
||||||
|
});
|
||||||
|
});
|
||||||
@@ -93,6 +93,26 @@ export function findChannel(
|
|||||||
return channels.find((c) => c.ownsJid(jid));
|
return channels.find((c) => c.ownsJid(jid));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
export function findChannelForDeliveryRole(
|
||||||
|
channels: Channel[],
|
||||||
|
jid: string,
|
||||||
|
senderRole?: string,
|
||||||
|
): Channel | undefined {
|
||||||
|
if (senderRole === 'reviewer') {
|
||||||
|
return (
|
||||||
|
findChannelByName(channels, 'discord-review') ||
|
||||||
|
findChannel(channels, jid)
|
||||||
|
);
|
||||||
|
}
|
||||||
|
if (senderRole === 'arbiter') {
|
||||||
|
return (
|
||||||
|
findChannelByName(channels, 'discord-arbiter') ||
|
||||||
|
findChannel(channels, jid)
|
||||||
|
);
|
||||||
|
}
|
||||||
|
return findChannel(channels, jid);
|
||||||
|
}
|
||||||
|
|
||||||
export function findChannelByName(
|
export function findChannelByName(
|
||||||
channels: Channel[],
|
channels: Channel[],
|
||||||
name: string,
|
name: string,
|
||||||
|
|||||||
Reference in New Issue
Block a user