diff --git a/src/agent-runner.ts b/src/agent-runner.ts index 2518347..e4a1cba 100644 --- a/src/agent-runner.ts +++ b/src/agent-runner.ts @@ -394,7 +394,12 @@ export async function runAgentProcess( stdout += chunk.slice(0, remaining); stdoutTruncated = true; logger.warn( - { group: group.name, chatJid: input.chatJid, runId: input.runId, size: stdout.length }, + { + group: group.name, + chatJid: input.chatJid, + runId: input.runId, + size: stdout.length, + }, 'Agent stdout truncated due to size limit', ); } else { @@ -424,7 +429,12 @@ export async function runAgentProcess( outputChain = outputChain.then(() => onOutput(parsed)); } catch (err) { logger.warn( - { group: group.name, chatJid: input.chatJid, runId: input.runId, error: err }, + { + group: group.name, + chatJid: input.chatJid, + runId: input.runId, + error: err, + }, 'Failed to parse streamed output chunk', ); } @@ -440,7 +450,12 @@ export async function runAgentProcess( const killOnTimeout = () => { timedOut = true; logger.error( - { group: group.name, chatJid: input.chatJid, runId: input.runId, processName }, + { + group: group.name, + chatJid: input.chatJid, + runId: input.runId, + processName, + }, 'Agent timeout, sending SIGTERM', ); proc.kill('SIGTERM'); @@ -640,7 +655,12 @@ export async function runAgentProcess( resolve(output); } catch (err) { logger.error( - { group: group.name, chatJid: input.chatJid, runId: input.runId, error: err }, + { + group: group.name, + chatJid: input.chatJid, + runId: input.runId, + error: err, + }, 'Failed to parse agent output', ); resolve({ @@ -654,7 +674,13 @@ export async function runAgentProcess( proc.on('error', (err) => { clearTimeout(timeout); logger.error( - { group: group.name, chatJid: input.chatJid, runId: input.runId, processName, error: err }, + { + group: group.name, + chatJid: input.chatJid, + runId: input.runId, + processName, + error: err, + }, 'Agent spawn error', ); resolve({ diff --git a/src/config.ts b/src/config.ts index 7a2659b..f1104ff 100644 --- a/src/config.ts +++ b/src/config.ts @@ -3,7 +3,11 @@ import path from 'path'; import { readEnvFile } from './env.js'; -const envConfig = readEnvFile(['ASSISTANT_NAME', 'ASSISTANT_HAS_OWN_NUMBER']); +const envConfig = readEnvFile([ + 'ASSISTANT_NAME', + 'ASSISTANT_HAS_OWN_NUMBER', + 'SESSION_COMMAND_USER_IDS', +]); export const ASSISTANT_NAME = process.env.ASSISTANT_NAME || envConfig.ASSISTANT_NAME || 'Andy'; @@ -15,6 +19,14 @@ export const SERVICE_AGENT_TYPE: 'claude-code' | 'codex' = export const ASSISTANT_HAS_OWN_NUMBER = (process.env.ASSISTANT_HAS_OWN_NUMBER || envConfig.ASSISTANT_HAS_OWN_NUMBER) === 'true'; +export const SESSION_COMMAND_USER_IDS = new Set( + (process.env.SESSION_COMMAND_USER_IDS || + envConfig.SESSION_COMMAND_USER_IDS || + '') + .split(',') + .map((id) => id.trim()) + .filter(Boolean), +); export const POLL_INTERVAL = 2000; export const SCHEDULER_POLL_INTERVAL = 60000; @@ -72,3 +84,7 @@ export const USAGE_UPDATE_INTERVAL = 300000; // 5 minutes // Uses system timezone by default export const TIMEZONE = process.env.TZ || Intl.DateTimeFormat().resolvedOptions().timeZone; + +export function isSessionCommandSenderAllowed(senderId: string): boolean { + return SESSION_COMMAND_USER_IDS.has(senderId); +} diff --git a/src/group-queue.ts b/src/group-queue.ts index 46c05e6..63f9ba7 100644 --- a/src/group-queue.ts +++ b/src/group-queue.ts @@ -245,7 +245,12 @@ export class GroupQueue { return true; } catch (err) { logger.warn( - { groupJid, runId: state.currentRunId, groupFolder: state.groupFolder, err }, + { + groupJid, + runId: state.currentRunId, + groupFolder: state.groupFolder, + err, + }, 'Failed to queue follow-up message for active agent', ); return false; @@ -311,7 +316,10 @@ export class GroupQueue { let outcome: 'success' | 'retry_scheduled' | 'error' = 'success'; try { if (this.processMessagesFn) { - const success = await this.processMessagesFn(groupJid, { runId, reason }); + const success = await this.processMessagesFn(groupJid, { + runId, + reason, + }); if (success) { state.retryCount = 0; } else { @@ -321,7 +329,10 @@ export class GroupQueue { } } catch (err) { outcome = 'error'; - logger.error({ groupJid, runId, err }, 'Error processing messages for group'); + logger.error( + { groupJid, runId, err }, + 'Error processing messages for group', + ); this.scheduleRetry(groupJid, state, runId); } finally { const durationMs = state.startedAt ? Date.now() - state.startedAt : null; diff --git a/src/ipc.ts b/src/ipc.ts index 8f04592..72a31c2 100644 --- a/src/ipc.ts +++ b/src/ipc.ts @@ -400,11 +400,7 @@ export async function processTaskIpc( await deps.syncGroups(true); // Write updated snapshot immediately const availableGroups = deps.getAvailableGroups(); - deps.writeGroupsSnapshot( - sourceGroup, - true, - availableGroups, - ); + deps.writeGroupsSnapshot(sourceGroup, true, availableGroups); } else { logger.warn( { sourceGroup }, diff --git a/src/message-runtime.ts b/src/message-runtime.ts index 1834c20..2eafdf4 100644 --- a/src/message-runtime.ts +++ b/src/message-runtime.ts @@ -12,12 +12,10 @@ import { getMessagesSince, getNewMessages, } from './db.js'; +import { isSessionCommandSenderAllowed } from './config.js'; import { GroupQueue, GroupRunContext } from './group-queue.js'; import { findChannel, formatMessages } from './router.js'; -import { - isTriggerAllowed, - loadSenderAllowlist, -} from './sender-allowlist.js'; +import { isTriggerAllowed, loadSenderAllowlist } from './sender-allowlist.js'; import { extractSessionCommand, handleSessionCommand, @@ -154,13 +152,19 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { const registeredGroups = deps.getRegisteredGroups(); const group = registeredGroups[chatJid]; if (!group) { - logger.warn({ chatJid, runId, reason }, 'Registered group missing for queued run'); + logger.warn( + { chatJid, runId, reason }, + 'Registered group missing for queued run', + ); return true; } const channel = findChannel(deps.channels, chatJid); if (!channel) { - logger.warn({ chatJid, runId, reason }, 'No channel owns JID, skipping messages'); + logger.warn( + { chatJid, runId, reason }, + 'No channel owns JID, skipping messages', + ); return true; } @@ -175,7 +179,13 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { if (missedMessages.length === 0) { logger.info( - { chatJid, group: group.name, groupFolder: group.folder, runId, reason }, + { + chatJid, + group: group.name, + groupFolder: group.folder, + runId, + reason, + }, 'No pending messages for queued run', ); return true; @@ -218,6 +228,7 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { deps.saveState(); }, formatMessages, + isAdminSender: (msg) => isSessionCommandSenderAllowed(msg.sender), canSenderInteract: (msg) => { const hasTrigger = deps.triggerPattern.test(msg.content.trim()); const requiresTrigger = @@ -227,11 +238,7 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { !requiresTrigger || (hasTrigger && (msg.is_from_me || - isTriggerAllowed( - chatJid, - msg.sender, - loadSenderAllowlist(), - ))) + isTriggerAllowed(chatJid, msg.sender, loadSenderAllowlist()))) ); }, }, @@ -293,40 +300,48 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { await channel.setTyping?.(chatJid, true); - const output = await runAgent(group, prompt, chatJid, runId, async (result) => { - if (result.result) { - const raw = - typeof result.result === 'string' - ? result.result - : JSON.stringify(result.result); - const text = raw.replace(/[\s\S]*?<\/internal>/g, '').trim(); - logger.info( - { - chatJid, - group: group.name, - groupFolder: group.folder, - runId, - resultStatus: result.status, - }, - `Agent output: ${raw.slice(0, 200)}`, - ); - if (text) { - await channel.sendMessage(chatJid, text); - outputSentToUser = true; + const output = await runAgent( + group, + prompt, + chatJid, + runId, + async (result) => { + if (result.result) { + const raw = + typeof result.result === 'string' + ? result.result + : JSON.stringify(result.result); + const text = raw + .replace(/[\s\S]*?<\/internal>/g, '') + .trim(); + logger.info( + { + chatJid, + group: group.name, + groupFolder: group.folder, + runId, + resultStatus: result.status, + }, + `Agent output: ${raw.slice(0, 200)}`, + ); + if (text) { + await channel.sendMessage(chatJid, text); + outputSentToUser = true; + } } - } - await channel.setTyping?.(chatJid, false); - resetIdleTimer(); + await channel.setTyping?.(chatJid, false); + resetIdleTimer(); - if (result.status === 'success') { - deps.queue.notifyIdle(chatJid, runId); - } + if (result.status === 'success') { + deps.queue.notifyIdle(chatJid, runId); + } - if (result.status === 'error') { - hadError = true; - } - }); + if (result.status === 'error') { + hadError = true; + } + }, + ); await channel.setTyping?.(chatJid, false); @@ -408,7 +423,10 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { const channel = findChannel(deps.channels, chatJid); if (!channel) { - logger.warn({ chatJid }, 'No channel owns JID, skipping messages'); + logger.warn( + { chatJid }, + 'No channel owns JID, skipping messages', + ); continue; } @@ -420,8 +438,7 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { const lastHuman = getLastHumanMessageTimestamp(chatJid); if ( !lastHuman || - Date.now() - new Date(lastHuman).getTime() > - 12 * 60 * 60 * 1000 + Date.now() - new Date(lastHuman).getTime() > 12 * 60 * 60 * 1000 ) { logger.info( { chatJid, lastHuman }, @@ -433,7 +450,8 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { const loopCmdMsg = groupMessages.find( (msg) => - extractSessionCommand(msg.content, deps.triggerPattern) !== null, + extractSessionCommand(msg.content, deps.triggerPattern) !== + null, ); if (loopCmdMsg) { @@ -441,9 +459,12 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { isSessionCommandAllowed( isMainGroup, loopCmdMsg.is_from_me === true, + isSessionCommandSenderAllowed(loopCmdMsg.sender), ) ) { - deps.queue.closeStdin(chatJid); + deps.queue.closeStdin(chatJid, { + reason: 'session-command-detected', + }); } deps.queue.enqueueMessageCheck(chatJid); continue; diff --git a/src/session-commands.test.ts b/src/session-commands.test.ts index 66e4c06..8f5896e 100644 --- a/src/session-commands.test.ts +++ b/src/session-commands.test.ts @@ -47,19 +47,23 @@ describe('extractSessionCommand', () => { describe('isSessionCommandAllowed', () => { it('allows main group regardless of sender', () => { - expect(isSessionCommandAllowed(true, false)).toBe(true); + expect(isSessionCommandAllowed(true, false, false)).toBe(true); }); it('allows trusted/admin sender (is_from_me) in non-main group', () => { - expect(isSessionCommandAllowed(false, true)).toBe(true); + expect(isSessionCommandAllowed(false, true, false)).toBe(true); + }); + + it('allows configured admin sender in non-main group', () => { + expect(isSessionCommandAllowed(false, false, true)).toBe(true); }); it('denies untrusted sender in non-main group', () => { - expect(isSessionCommandAllowed(false, false)).toBe(false); + expect(isSessionCommandAllowed(false, false, false)).toBe(false); }); it('allows trusted sender in main group', () => { - expect(isSessionCommandAllowed(true, true)).toBe(true); + expect(isSessionCommandAllowed(true, true, false)).toBe(true); }); }); @@ -89,6 +93,7 @@ function makeDeps( clearSession: vi.fn(), advanceCursor: vi.fn(), formatMessages: vi.fn().mockReturnValue(''), + isAdminSender: vi.fn().mockReturnValue(false), canSenderInteract: vi.fn().mockReturnValue(true), ...overrides, }; @@ -228,6 +233,25 @@ describe('handleSessionCommand', () => { ); }); + it('allows configured admin sender in non-main group', async () => { + const deps = makeDeps({ + isAdminSender: vi.fn().mockReturnValue(true), + }); + const result = await handleSessionCommand({ + missedMessages: [makeMsg('/clear', { is_from_me: false, sender: 'discord-user-1' })], + isMainGroup: false, + groupName: 'test', + triggerPattern: trigger, + timezone: 'UTC', + deps, + }); + expect(result).toEqual({ handled: true, success: true }); + expect(deps.clearSession).toHaveBeenCalledTimes(1); + expect(deps.sendMessage).toHaveBeenCalledWith( + 'Current session cleared. The next message will start a new conversation.', + ); + }); + it('reports failure when command-stage runAgent returns error without streamed status', async () => { // runAgent resolves 'error' but callback never gets status: 'error' const deps = makeDeps({ diff --git a/src/session-commands.ts b/src/session-commands.ts index 20fdf53..170c54b 100644 --- a/src/session-commands.ts +++ b/src/session-commands.ts @@ -18,13 +18,14 @@ export function extractSessionCommand( /** * Check if a session command sender is authorized. - * Allowed: main group (any sender), or trusted/admin sender (is_from_me) in any group. + * Allowed: main group (any sender), or trusted/admin sender in any group. */ export function isSessionCommandAllowed( isMainGroup: boolean, isFromMe: boolean, + isAdminSender: boolean, ): boolean { - return isMainGroup || isFromMe; + return isMainGroup || isFromMe || isAdminSender; } /** Minimal agent result interface — matches the subset of AgentOutput used here. */ @@ -45,6 +46,7 @@ export interface SessionCommandDeps { clearSession: () => void; advanceCursor: (timestamp: string) => void; formatMessages: (msgs: NewMessage[], timezone: string) => string; + isAdminSender: (msg: NewMessage) => boolean; /** Whether the denied sender would normally be allowed to interact (for denial messages). */ canSenderInteract: (msg: NewMessage) => boolean; } @@ -89,7 +91,13 @@ export async function handleSessionCommand(opts: { if (!command || !cmdMsg) return { handled: false }; - if (!isSessionCommandAllowed(isMainGroup, cmdMsg.is_from_me === true)) { + if ( + !isSessionCommandAllowed( + isMainGroup, + cmdMsg.is_from_me === true, + deps.isAdminSender(cmdMsg), + ) + ) { // DENIED: send denial if the sender would normally be allowed to interact, // then silently consume the command by advancing the cursor past it. // Trade-off: other messages in the same batch are also consumed (cursor is