claude-bot/EJClaw
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: rename container references to agent across codebase

Browse Source 
- container-runner.ts → agent-runner.ts
- ContainerInput/Output → AgentInput/Output
- ContainerConfig → AgentConfig
- runContainerAgent → runAgentProcess
- CONTAINER_TIMEOUT → AGENT_TIMEOUT (with env fallback)
- MAX_CONCURRENT_CONTAINERS → MAX_CONCURRENT_AGENTS
- containerName → processName, isTaskContainer → isTaskProcess
- DB column container_config kept as-is (backwards compat)
This commit is contained in:
Eyejoker
2026-03-13 16:18:36 +09:00
parent 35277cea0a
commit bd7f4408ae
12 changed files with 144 additions and 133 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
src/container-runner.test.ts → src/agent-runner.test.ts
View File

@@ -2,14 +2,14 @@ import { describe, it, expect, beforeEach, vi, afterEach } from 'vitest';
import { EventEmitter } from 'events'; import { EventEmitter } from 'events';
import { PassThrough } from 'stream'; import { PassThrough } from 'stream';
// Sentinel markers must match container-runner.ts // Sentinel markers must match agent-runner.ts
const OUTPUT_START_MARKER = '---NANOCLAW_OUTPUT_START---'; const OUTPUT_START_MARKER = '---NANOCLAW_OUTPUT_START---';
const OUTPUT_END_MARKER = '---NANOCLAW_OUTPUT_END---'; const OUTPUT_END_MARKER = '---NANOCLAW_OUTPUT_END---';
// Mock config // Mock config
vi.mock('./config.js', () => ({ vi.mock('./config.js', () => ({
CONTAINER_MAX_OUTPUT_SIZE: 10485760, AGENT_MAX_OUTPUT_SIZE: 10485760,
CONTAINER_TIMEOUT: 1800000, // 30min AGENT_TIMEOUT: 1800000, // 30min
DATA_DIR: '/tmp/nanoclaw-test-data', DATA_DIR: '/tmp/nanoclaw-test-data',
GROUPS_DIR: '/tmp/nanoclaw-test-groups', GROUPS_DIR: '/tmp/nanoclaw-test-groups',
IDLE_TIMEOUT: 1800000, // 30min IDLE_TIMEOUT: 1800000, // 30min
@@ -89,7 +89,7 @@ vi.mock('child_process', async () => {
}; };
}); });
import { runContainerAgent, ContainerOutput } from './container-runner.js'; import { runAgentProcess, AgentOutput } from './agent-runner.js';
import type { RegisteredGroup } from './types.js'; import type { RegisteredGroup } from './types.js';
const testGroup: RegisteredGroup = { const testGroup: RegisteredGroup = {
@@ -108,13 +108,13 @@ const testInput = {
function emitOutputMarker( function emitOutputMarker(
proc: ReturnType<typeof createFakeProcess>, proc: ReturnType<typeof createFakeProcess>,
output: ContainerOutput, output: AgentOutput,
) { ) {
const json = JSON.stringify(output); const json = JSON.stringify(output);
proc.stdout.push(`${OUTPUT_START_MARKER}\n${json}\n${OUTPUT_END_MARKER}\n`); proc.stdout.push(`${OUTPUT_START_MARKER}\n${json}\n${OUTPUT_END_MARKER}\n`);
} }
describe('container-runner timeout behavior', () => { describe('agent-runner timeout behavior', () => {
beforeEach(() => { beforeEach(() => {
vi.useFakeTimers(); vi.useFakeTimers();
fakeProc = createFakeProcess(); fakeProc = createFakeProcess();
@@ -126,7 +126,7 @@ describe('container-runner timeout behavior', () => {
it('timeout after output resolves as success', async () => { it('timeout after output resolves as success', async () => {
const onOutput = vi.fn(async () => {}); const onOutput = vi.fn(async () => {});
const resultPromise = runContainerAgent( const resultPromise = runAgentProcess(
testGroup, testGroup,
testInput, testInput,
() => {}, () => {},
@@ -146,7 +146,7 @@ describe('container-runner timeout behavior', () => {
// Fire the hard timeout (IDLE_TIMEOUT + 30s = 1830000ms) // Fire the hard timeout (IDLE_TIMEOUT + 30s = 1830000ms)
await vi.advanceTimersByTimeAsync(1830000); await vi.advanceTimersByTimeAsync(1830000);
// Emit close event (as if container was stopped by the timeout) // Emit close event (as if agent was stopped by the timeout)
fakeProc.emit('close', 137); fakeProc.emit('close', 137);
// Let the promise resolve // Let the promise resolve
@@ -162,7 +162,7 @@ describe('container-runner timeout behavior', () => {
it('timeout with no output resolves as error', async () => { it('timeout with no output resolves as error', async () => {
const onOutput = vi.fn(async () => {}); const onOutput = vi.fn(async () => {});
const resultPromise = runContainerAgent( const resultPromise = runAgentProcess(
testGroup, testGroup,
testInput, testInput,
() => {}, () => {},
@@ -185,7 +185,7 @@ describe('container-runner timeout behavior', () => {
it('normal exit after output resolves as success', async () => { it('normal exit after output resolves as success', async () => {
const onOutput = vi.fn(async () => {}); const onOutput = vi.fn(async () => {});
const resultPromise = runContainerAgent( const resultPromise = runAgentProcess(
testGroup, testGroup,
testInput, testInput,
() => {}, () => {},

44
src/container-runner.ts → src/agent-runner.ts
View File

@@ -8,8 +8,8 @@ import os from 'os';
import path from 'path'; import path from 'path';
import { import {
CONTAINER_MAX_OUTPUT_SIZE, AGENT_MAX_OUTPUT_SIZE,
CONTAINER_TIMEOUT, AGENT_TIMEOUT,
DATA_DIR, DATA_DIR,
GROUPS_DIR, GROUPS_DIR,
IDLE_TIMEOUT, IDLE_TIMEOUT,
@@ -24,7 +24,7 @@ import { RegisteredGroup } from './types.js';
const OUTPUT_START_MARKER = '---NANOCLAW_OUTPUT_START---'; const OUTPUT_START_MARKER = '---NANOCLAW_OUTPUT_START---';
const OUTPUT_END_MARKER = '---NANOCLAW_OUTPUT_END---'; const OUTPUT_END_MARKER = '---NANOCLAW_OUTPUT_END---';
export interface ContainerInput { export interface AgentInput {
prompt: string; prompt: string;
sessionId?: string; sessionId?: string;
groupFolder: string; groupFolder: string;
@@ -35,7 +35,7 @@ export interface ContainerInput {
agentType?: 'claude-code' | 'codex'; agentType?: 'claude-code' | 'codex';
} }
export interface ContainerOutput { export interface AgentOutput {
status: 'success' | 'error'; status: 'success' | 'error';
result: string | null; result: string | null;
newSessionId?: string; newSessionId?: string;
@@ -137,8 +137,8 @@ function prepareGroupEnvironment(
// Additional mount directories (validated) // Additional mount directories (validated)
const extraDirs: string[] = []; const extraDirs: string[] = [];
if (group.containerConfig?.additionalMounts) { if (group.agentConfig?.additionalMounts) {
for (const mount of group.containerConfig.additionalMounts) { for (const mount of group.agentConfig.additionalMounts) {
if (fs.existsSync(mount.hostPath)) { if (fs.existsSync(mount.hostPath)) {
extraDirs.push(mount.hostPath); extraDirs.push(mount.hostPath);
} }
@@ -212,12 +212,12 @@ function prepareGroupEnvironment(
// Codex model/effort configuration (per-group overrides global) // Codex model/effort configuration (per-group overrides global)
const codexModel = const codexModel =
group.containerConfig?.codexModel || group.agentConfig?.codexModel ||
envVars.CODEX_MODEL || envVars.CODEX_MODEL ||
process.env.CODEX_MODEL; process.env.CODEX_MODEL;
if (codexModel) env.CODEX_MODEL = codexModel; if (codexModel) env.CODEX_MODEL = codexModel;
const codexEffort = const codexEffort =
group.containerConfig?.codexEffort || group.agentConfig?.codexEffort ||
envVars.CODEX_EFFORT || envVars.CODEX_EFFORT ||
process.env.CODEX_EFFORT; process.env.CODEX_EFFORT;
if (codexEffort) env.CODEX_EFFORT = codexEffort; if (codexEffort) env.CODEX_EFFORT = codexEffort;
@@ -321,23 +321,23 @@ NANOCLAW_IS_MAIN = ${JSON.stringify(isMain ? '1' : '0')}
const val = envVars[key as keyof typeof envVars] || process.env[key]; const val = envVars[key as keyof typeof envVars] || process.env[key];
if (val) env[key] = val; if (val) env[key] = val;
} }
if (group.containerConfig?.claudeModel) { if (group.agentConfig?.claudeModel) {
env.CLAUDE_MODEL = group.containerConfig.claudeModel; env.CLAUDE_MODEL = group.agentConfig.claudeModel;
} }
if (group.containerConfig?.claudeEffort) { if (group.agentConfig?.claudeEffort) {
env.CLAUDE_EFFORT = group.containerConfig.claudeEffort; env.CLAUDE_EFFORT = group.agentConfig.claudeEffort;
} }
} }
return { env, groupDir, runnerDir }; return { env, groupDir, runnerDir };
} }
export async function runContainerAgent( export async function runAgentProcess(
group: RegisteredGroup, group: RegisteredGroup,
input: ContainerInput, input: AgentInput,
onProcess: (proc: ChildProcess, containerName: string) => void, onProcess: (proc: ChildProcess, processName: string) => void,
onOutput?: (output: ContainerOutput) => Promise<void>, onOutput?: (output: AgentOutput) => Promise<void>,
): Promise<ContainerOutput> { ): Promise<AgentOutput> {
const startTime = Date.now(); const startTime = Date.now();
const { env, groupDir, runnerDir } = prepareGroupEnvironment( const { env, groupDir, runnerDir } = prepareGroupEnvironment(
group, group,
@@ -400,7 +400,7 @@ export async function runContainerAgent(
const chunk = data.toString(); const chunk = data.toString();
if (!stdoutTruncated) { if (!stdoutTruncated) {
const remaining = CONTAINER_MAX_OUTPUT_SIZE - stdout.length; const remaining = AGENT_MAX_OUTPUT_SIZE - stdout.length;
if (chunk.length > remaining) { if (chunk.length > remaining) {
stdout += chunk.slice(0, remaining); stdout += chunk.slice(0, remaining);
stdoutTruncated = true; stdoutTruncated = true;
@@ -426,7 +426,7 @@ export async function runContainerAgent(
parseBuffer = parseBuffer.slice(endIdx + OUTPUT_END_MARKER.length); parseBuffer = parseBuffer.slice(endIdx + OUTPUT_END_MARKER.length);
try { try {
const parsed: ContainerOutput = JSON.parse(jsonStr); const parsed: AgentOutput = JSON.parse(jsonStr);
if (parsed.newSessionId) { if (parsed.newSessionId) {
newSessionId = parsed.newSessionId; newSessionId = parsed.newSessionId;
} }
@@ -450,7 +450,7 @@ export async function runContainerAgent(
if (line) logger.debug({ agent: group.folder }, line); if (line) logger.debug({ agent: group.folder }, line);
} }
if (stderrTruncated) return; if (stderrTruncated) return;
const remaining = CONTAINER_MAX_OUTPUT_SIZE - stderr.length; const remaining = AGENT_MAX_OUTPUT_SIZE - stderr.length;
if (chunk.length > remaining) { if (chunk.length > remaining) {
stderr += chunk.slice(0, remaining); stderr += chunk.slice(0, remaining);
stderrTruncated = true; stderrTruncated = true;
@@ -461,7 +461,7 @@ export async function runContainerAgent(
let timedOut = false; let timedOut = false;
let hadStreamingOutput = false; let hadStreamingOutput = false;
const configTimeout = group.containerConfig?.timeout || CONTAINER_TIMEOUT; const configTimeout = group.agentConfig?.timeout || AGENT_TIMEOUT;
const timeoutMs = Math.max(configTimeout, IDLE_TIMEOUT + 30_000); const timeoutMs = Math.max(configTimeout, IDLE_TIMEOUT + 30_000);
const killOnTimeout = () => { const killOnTimeout = () => {
@@ -596,7 +596,7 @@ export async function runContainerAgent(
const lines = stdout.trim().split('\n'); const lines = stdout.trim().split('\n');
jsonLine = lines[lines.length - 1]; jsonLine = lines[lines.length - 1];
} }
const output: ContainerOutput = JSON.parse(jsonLine); const output: AgentOutput = JSON.parse(jsonLine);
logger.info( logger.info(
{ group: group.name, duration, status: output.status }, { group: group.name, duration, status: output.status },
'Agent completed', 'Agent completed',

14
src/config.ts
View File

@@ -32,19 +32,19 @@ export const DATA_DIR = path.resolve(
process.env.NANOCLAW_DATA_DIR || path.join(PROJECT_ROOT, 'data'), process.env.NANOCLAW_DATA_DIR || path.join(PROJECT_ROOT, 'data'),
); );
export const CONTAINER_TIMEOUT = parseInt( export const AGENT_TIMEOUT = parseInt(
process.env.CONTAINER_TIMEOUT || '1800000', process.env.AGENT_TIMEOUT || process.env.CONTAINER_TIMEOUT || '1800000',
10, 10,
); );
export const CONTAINER_MAX_OUTPUT_SIZE = parseInt( export const AGENT_MAX_OUTPUT_SIZE = parseInt(
process.env.CONTAINER_MAX_OUTPUT_SIZE || '10485760', process.env.AGENT_MAX_OUTPUT_SIZE || process.env.CONTAINER_MAX_OUTPUT_SIZE || '10485760',
10, 10,
); // 10MB default ); // 10MB default
export const IPC_POLL_INTERVAL = 1000; export const IPC_POLL_INTERVAL = 1000;
export const IDLE_TIMEOUT = parseInt(process.env.IDLE_TIMEOUT || '1800000', 10); // 30min default — how long to keep container alive after last result export const IDLE_TIMEOUT = parseInt(process.env.IDLE_TIMEOUT || '1800000', 10); // 30min default — how long to keep agent alive after last result
export const MAX_CONCURRENT_CONTAINERS = Math.max( export const MAX_CONCURRENT_AGENTS = Math.max(
1, 1,
parseInt(process.env.MAX_CONCURRENT_CONTAINERS || '5', 10) || 5, parseInt(process.env.MAX_CONCURRENT_AGENTS || process.env.MAX_CONCURRENT_CONTAINERS || '5', 10) || 5,
); );
function escapeRegex(str: string): string { function escapeRegex(str: string): string {

6
src/db.ts
View File

@@ -600,7 +600,7 @@ export function getRegisteredGroup(
folder: row.folder, folder: row.folder,
trigger: row.trigger_pattern, trigger: row.trigger_pattern,
added_at: row.added_at, added_at: row.added_at,
containerConfig: row.container_config agentConfig: row.container_config
? JSON.parse(row.container_config) ? JSON.parse(row.container_config)
: undefined, : undefined,
requiresTrigger: requiresTrigger:
@@ -624,7 +624,7 @@ export function setRegisteredGroup(jid: string, group: RegisteredGroup): void {
group.folder, group.folder,
group.trigger, group.trigger,
group.added_at, group.added_at,
group.containerConfig ? JSON.stringify(group.containerConfig) : null, group.agentConfig ? JSON.stringify(group.agentConfig) : null,
group.requiresTrigger === undefined ? 1 : group.requiresTrigger ? 1 : 0, group.requiresTrigger === undefined ? 1 : group.requiresTrigger ? 1 : 0,
group.isMain ? 1 : 0, group.isMain ? 1 : 0,
group.agentType || 'claude-code', group.agentType || 'claude-code',
@@ -659,7 +659,7 @@ export function getAllRegisteredGroups(): Record<string, RegisteredGroup> {
folder: row.folder, folder: row.folder,
trigger: row.trigger_pattern, trigger: row.trigger_pattern,
added_at: row.added_at, added_at: row.added_at,
containerConfig: row.container_config agentConfig: row.container_config
? JSON.parse(row.container_config) ? JSON.parse(row.container_config)
: undefined, : undefined,
requiresTrigger: requiresTrigger:

36
src/group-queue.test.ts
View File

@@ -5,7 +5,7 @@ import { GroupQueue } from './group-queue.js';
// Mock config to control concurrency limit // Mock config to control concurrency limit
vi.mock('./config.js', () => ({ vi.mock('./config.js', () => ({
DATA_DIR: '/tmp/nanoclaw-test-data', DATA_DIR: '/tmp/nanoclaw-test-data',
MAX_CONCURRENT_CONTAINERS: 2, MAX_CONCURRENT_AGENTS: 2,
})); }));
// Mock fs operations used by sendMessage/closeStdin // Mock fs operations used by sendMessage/closeStdin
@@ -36,7 +36,7 @@ describe('GroupQueue', () => {
// --- Single group at a time --- // --- Single group at a time ---
it('only runs one container per group at a time', async () => { it('only runs one agent per group at a time', async () => {
let concurrentCount = 0; let concurrentCount = 0;
let maxConcurrent = 0; let maxConcurrent = 0;
@@ -87,7 +87,7 @@ describe('GroupQueue', () => {
// Let promises settle // Let promises settle
await vi.advanceTimersByTimeAsync(10); await vi.advanceTimersByTimeAsync(10);
// Only 2 should be active (MAX_CONCURRENT_CONTAINERS = 2) // Only 2 should be active (MAX_CONCURRENT_AGENTS = 2)
expect(maxActive).toBe(2); expect(maxActive).toBe(2);
expect(activeCount).toBe(2); expect(activeCount).toBe(2);
@@ -280,7 +280,7 @@ describe('GroupQueue', () => {
// --- Idle preemption --- // --- Idle preemption ---
it('does NOT preempt active container when not idle', async () => { it('does NOT preempt active agent when not idle', async () => {
const fs = await import('fs'); const fs = await import('fs');
let resolveProcess: () => void; let resolveProcess: () => void;
@@ -301,15 +301,15 @@ describe('GroupQueue', () => {
queue.registerProcess( queue.registerProcess(
'group1@g.us', 'group1@g.us',
{} as any, {} as any,
'container-1', 'agent-1',
'test-group', 'test-group',
); );
// Enqueue a task while container is active but NOT idle // Enqueue a task while agent is active but NOT idle
const taskFn = vi.fn(async () => {}); const taskFn = vi.fn(async () => {});
queue.enqueueTask('group1@g.us', 'task-1', taskFn); queue.enqueueTask('group1@g.us', 'task-1', taskFn);
// _close should NOT have been written (container is working, not idle) // _close should NOT have been written (agent is working, not idle)
const writeFileSync = vi.mocked(fs.default.writeFileSync); const writeFileSync = vi.mocked(fs.default.writeFileSync);
const closeWrites = writeFileSync.mock.calls.filter( const closeWrites = writeFileSync.mock.calls.filter(
(call) => typeof call[0] === 'string' && call[0].endsWith('_close'), (call) => typeof call[0] === 'string' && call[0].endsWith('_close'),
@@ -320,7 +320,7 @@ describe('GroupQueue', () => {
await vi.advanceTimersByTimeAsync(10); await vi.advanceTimersByTimeAsync(10);
}); });
it('preempts idle container when task is enqueued', async () => { it('preempts idle agent when task is enqueued', async () => {
const fs = await import('fs'); const fs = await import('fs');
let resolveProcess: () => void; let resolveProcess: () => void;
@@ -341,7 +341,7 @@ describe('GroupQueue', () => {
queue.registerProcess( queue.registerProcess(
'group1@g.us', 'group1@g.us',
{} as any, {} as any,
'container-1', 'agent-1',
'test-group', 'test-group',
); );
queue.notifyIdle('group1@g.us'); queue.notifyIdle('group1@g.us');
@@ -353,7 +353,7 @@ describe('GroupQueue', () => {
const taskFn = vi.fn(async () => {}); const taskFn = vi.fn(async () => {});
queue.enqueueTask('group1@g.us', 'task-1', taskFn); queue.enqueueTask('group1@g.us', 'task-1', taskFn);
// _close SHOULD have been written (container is idle) // _close SHOULD have been written (agent is idle)
const closeWrites = writeFileSync.mock.calls.filter( const closeWrites = writeFileSync.mock.calls.filter(
(call) => typeof call[0] === 'string' && call[0].endsWith('_close'), (call) => typeof call[0] === 'string' && call[0].endsWith('_close'),
); );
@@ -380,11 +380,11 @@ describe('GroupQueue', () => {
queue.registerProcess( queue.registerProcess(
'group1@g.us', 'group1@g.us',
{} as any, {} as any,
'container-1', 'agent-1',
'test-group', 'test-group',
); );
// Container becomes idle // Agent becomes idle
queue.notifyIdle('group1@g.us'); queue.notifyIdle('group1@g.us');
// A new user message arrives — resets idleWaiting // A new user message arrives — resets idleWaiting
@@ -406,7 +406,7 @@ describe('GroupQueue', () => {
await vi.advanceTimersByTimeAsync(10); await vi.advanceTimersByTimeAsync(10);
}); });
it('sendMessage returns false for task containers so user messages queue up', async () => { it('sendMessage returns false for task agents so user messages queue up', async () => {
let resolveTask: () => void; let resolveTask: () => void;
const taskFn = vi.fn(async () => { const taskFn = vi.fn(async () => {
@@ -415,17 +415,17 @@ describe('GroupQueue', () => {
}); });
}); });
// Start a task (sets isTaskContainer = true) // Start a task (sets isTaskProcess = true)
queue.enqueueTask('group1@g.us', 'task-1', taskFn); queue.enqueueTask('group1@g.us', 'task-1', taskFn);
await vi.advanceTimersByTimeAsync(10); await vi.advanceTimersByTimeAsync(10);
queue.registerProcess( queue.registerProcess(
'group1@g.us', 'group1@g.us',
{} as any, {} as any,
'container-1', 'agent-1',
'test-group', 'test-group',
); );
// sendMessage should return false — user messages must not go to task containers // sendMessage should return false — user messages must not go to task agents
const result = queue.sendMessage('group1@g.us', 'hello'); const result = queue.sendMessage('group1@g.us', 'hello');
expect(result).toBe(false); expect(result).toBe(false);
@@ -454,7 +454,7 @@ describe('GroupQueue', () => {
queue.registerProcess( queue.registerProcess(
'group1@g.us', 'group1@g.us',
{} as any, {} as any,
'container-1', 'agent-1',
'test-group', 'test-group',
); );
@@ -469,7 +469,7 @@ describe('GroupQueue', () => {
); );
expect(closeWrites).toHaveLength(0); expect(closeWrites).toHaveLength(0);
// Now container becomes idle — should preempt because task is pending // Now agent becomes idle — should preempt because task is pending
writeFileSync.mockClear(); writeFileSync.mockClear();
queue.notifyIdle('group1@g.us'); queue.notifyIdle('group1@g.us');

66
src/group-queue.ts
View File

@@ -2,7 +2,7 @@ import { ChildProcess } from 'child_process';
import fs from 'fs'; import fs from 'fs';
import path from 'path'; import path from 'path';
import { DATA_DIR, MAX_CONCURRENT_CONTAINERS } from './config.js'; import { DATA_DIR, MAX_CONCURRENT_AGENTS } from './config.js';
import { logger } from './logger.js'; import { logger } from './logger.js';
interface QueuedTask { interface QueuedTask {
@@ -17,12 +17,12 @@ const BASE_RETRY_MS = 5000;
interface GroupState { interface GroupState {
active: boolean; active: boolean;
idleWaiting: boolean; idleWaiting: boolean;
isTaskContainer: boolean; isTaskProcess: boolean;
runningTaskId: string | null; runningTaskId: string | null;
pendingMessages: boolean; pendingMessages: boolean;
pendingTasks: QueuedTask[]; pendingTasks: QueuedTask[];
process: ChildProcess | null; process: ChildProcess | null;
containerName: string | null; processName: string | null;
groupFolder: string | null; groupFolder: string | null;
retryCount: number; retryCount: number;
} }
@@ -41,12 +41,12 @@ export class GroupQueue {
state = { state = {
active: false, active: false,
idleWaiting: false, idleWaiting: false,
isTaskContainer: false, isTaskProcess: false,
runningTaskId: null, runningTaskId: null,
pendingMessages: false, pendingMessages: false,
pendingTasks: [], pendingTasks: [],
process: null, process: null,
containerName: null, processName: null,
groupFolder: null, groupFolder: null,
retryCount: 0, retryCount: 0,
}; };
@@ -64,18 +64,18 @@ export class GroupQueue {
const state = this.getGroup(groupJid); const state = this.getGroup(groupJid);
// Pre-set groupFolder so sendMessage can pipe IPC while container starts // Pre-set groupFolder so sendMessage can pipe IPC while agent process starts
if (groupFolder && !state.groupFolder) { if (groupFolder && !state.groupFolder) {
state.groupFolder = groupFolder; state.groupFolder = groupFolder;
} }
if (state.active) { if (state.active) {
state.pendingMessages = true; state.pendingMessages = true;
logger.debug({ groupJid }, 'Container active, message queued'); logger.debug({ groupJid }, 'Agent active, message queued');
return; return;
} }
if (this.activeCount >= MAX_CONCURRENT_CONTAINERS) { if (this.activeCount >= MAX_CONCURRENT_AGENTS) {
state.pendingMessages = true; state.pendingMessages = true;
if (!this.waitingGroups.includes(groupJid)) { if (!this.waitingGroups.includes(groupJid)) {
this.waitingGroups.push(groupJid); this.waitingGroups.push(groupJid);
@@ -112,11 +112,11 @@ export class GroupQueue {
if (state.idleWaiting) { if (state.idleWaiting) {
this.closeStdin(groupJid); this.closeStdin(groupJid);
} }
logger.debug({ groupJid, taskId }, 'Container active, task queued'); logger.debug({ groupJid, taskId }, 'Agent active, task queued');
return; return;
} }
if (this.activeCount >= MAX_CONCURRENT_CONTAINERS) { if (this.activeCount >= MAX_CONCURRENT_AGENTS) {
state.pendingTasks.push({ id: taskId, groupJid, fn }); state.pendingTasks.push({ id: taskId, groupJid, fn });
if (!this.waitingGroups.includes(groupJid)) { if (!this.waitingGroups.includes(groupJid)) {
this.waitingGroups.push(groupJid); this.waitingGroups.push(groupJid);
@@ -137,18 +137,18 @@ export class GroupQueue {
registerProcess( registerProcess(
groupJid: string, groupJid: string,
proc: ChildProcess, proc: ChildProcess,
containerName: string, processName: string,
groupFolder?: string, groupFolder?: string,
): void { ): void {
const state = this.getGroup(groupJid); const state = this.getGroup(groupJid);
state.process = proc; state.process = proc;
state.containerName = containerName; state.processName = processName;
if (groupFolder) state.groupFolder = groupFolder; if (groupFolder) state.groupFolder = groupFolder;
} }
/** /**
* Mark the container as idle-waiting (finished work, waiting for IPC input). * Mark the agent process as idle-waiting (finished work, waiting for IPC input).
* If tasks are pending, preempt the idle container immediately. * If tasks are pending, preempt the idle agent process immediately.
*/ */
notifyIdle(groupJid: string): void { notifyIdle(groupJid: string): void {
const state = this.getGroup(groupJid); const state = this.getGroup(groupJid);
@@ -159,12 +159,12 @@ export class GroupQueue {
} }
/** /**
* Send a follow-up message to the active container via IPC file. * Send a follow-up message to the active agent process via IPC file.
* Returns true if the message was written, false if no active container. * Returns true if the message was written, false if no active agent process.
*/ */
sendMessage(groupJid: string, text: string): boolean { sendMessage(groupJid: string, text: string): boolean {
const state = this.getGroup(groupJid); const state = this.getGroup(groupJid);
if (!state.active || !state.groupFolder || state.isTaskContainer) if (!state.active || !state.groupFolder || state.isTaskProcess)
return false; return false;
state.idleWaiting = false; // Agent is about to receive work, no longer idle state.idleWaiting = false; // Agent is about to receive work, no longer idle
@@ -183,7 +183,7 @@ export class GroupQueue {
} }
/** /**
* Signal the active container to wind down by writing a close sentinel. * Signal the active agent process to wind down by writing a close sentinel.
*/ */
closeStdin(groupJid: string): void { closeStdin(groupJid: string): void {
const state = this.getGroup(groupJid); const state = this.getGroup(groupJid);
@@ -205,13 +205,13 @@ export class GroupQueue {
const state = this.getGroup(groupJid); const state = this.getGroup(groupJid);
state.active = true; state.active = true;
state.idleWaiting = false; state.idleWaiting = false;
state.isTaskContainer = false; state.isTaskProcess = false;
state.pendingMessages = false; state.pendingMessages = false;
this.activeCount++; this.activeCount++;
logger.debug( logger.debug(
{ groupJid, reason, activeCount: this.activeCount }, { groupJid, reason, activeCount: this.activeCount },
'Starting container for group', 'Starting agent process for group',
); );
try { try {
@@ -229,7 +229,7 @@ export class GroupQueue {
} finally { } finally {
state.active = false; state.active = false;
state.process = null; state.process = null;
state.containerName = null; state.processName = null;
state.groupFolder = null; state.groupFolder = null;
this.activeCount--; this.activeCount--;
this.drainGroup(groupJid); this.drainGroup(groupJid);
@@ -240,7 +240,7 @@ export class GroupQueue {
const state = this.getGroup(groupJid); const state = this.getGroup(groupJid);
state.active = true; state.active = true;
state.idleWaiting = false; state.idleWaiting = false;
state.isTaskContainer = true; state.isTaskProcess = true;
state.runningTaskId = task.id; state.runningTaskId = task.id;
this.activeCount++; this.activeCount++;
@@ -255,10 +255,10 @@ export class GroupQueue {
logger.error({ groupJid, taskId: task.id, err }, 'Error running task'); logger.error({ groupJid, taskId: task.id, err }, 'Error running task');
} finally { } finally {
state.active = false; state.active = false;
state.isTaskContainer = false; state.isTaskProcess = false;
state.runningTaskId = null; state.runningTaskId = null;
state.process = null; state.process = null;
state.containerName = null; state.processName = null;
state.groupFolder = null; state.groupFolder = null;
this.activeCount--; this.activeCount--;
this.drainGroup(groupJid); this.drainGroup(groupJid);
@@ -323,7 +323,7 @@ export class GroupQueue {
private drainWaiting(): void { private drainWaiting(): void {
while ( while (
this.waitingGroups.length > 0 && this.waitingGroups.length > 0 &&
this.activeCount < MAX_CONCURRENT_CONTAINERS this.activeCount < MAX_CONCURRENT_AGENTS
) { ) {
const nextJid = this.waitingGroups.shift()!; const nextJid = this.waitingGroups.shift()!;
const state = this.getGroup(nextJid); const state = this.getGroup(nextJid);
@@ -352,19 +352,19 @@ export class GroupQueue {
async shutdown(_gracePeriodMs: number): Promise<void> { async shutdown(_gracePeriodMs: number): Promise<void> {
this.shuttingDown = true; this.shuttingDown = true;
// Count active containers but don't kill them — they'll finish on their own // Count active agent processes but don't kill them — they'll finish on their own
// via idle timeout or container timeout. The --rm flag cleans them up on exit. // via idle timeout or agent timeout.
// This prevents WhatsApp reconnection restarts from killing working agents. // This prevents reconnection restarts from killing working agents.
const activeContainers: string[] = []; const activeProcesses: string[] = [];
for (const [jid, state] of this.groups) { for (const [jid, state] of this.groups) {
if (state.process && !state.process.killed && state.containerName) { if (state.process && !state.process.killed && state.processName) {
activeContainers.push(state.containerName); activeProcesses.push(state.processName);
} }
} }
logger.info( logger.info(
{ activeCount: this.activeCount, detachedContainers: activeContainers }, { activeCount: this.activeCount, detachedProcesses: activeProcesses },
'GroupQueue shutting down (containers detached, not killed)', 'GroupQueue shutting down (agent processes detached, not killed)',
); );
} }
} }

40
src/index.ts
View File

@@ -14,11 +14,11 @@ import {
getRegisteredChannelNames, getRegisteredChannelNames,
} from './channels/registry.js'; } from './channels/registry.js';
import { import {
ContainerOutput, AgentOutput,
runContainerAgent, runAgentProcess,
writeGroupsSnapshot, writeGroupsSnapshot,
writeTasksSnapshot, writeTasksSnapshot,
} from './container-runner.js'; } from './agent-runner.js';
import { import {
getAllChats, getAllChats,
getAllRegisteredGroups, getAllRegisteredGroups,
@@ -121,7 +121,7 @@ function registerGroup(jid: string, group: RegisteredGroup): void {
* Get available groups list for the agent. * Get available groups list for the agent.
* Returns groups ordered by most recent activity. * Returns groups ordered by most recent activity.
*/ */
export function getAvailableGroups(): import('./container-runner.js').AvailableGroup[] { export function getAvailableGroups(): import('./agent-runner.js').AvailableGroup[] {
const chats = getAllChats(); const chats = getAllChats();
const registeredJids = new Set(Object.keys(registeredGroups)); const registeredJids = new Set(Object.keys(registeredGroups));
@@ -237,7 +237,7 @@ async function processGroupMessages(chatJid: string): Promise<boolean> {
idleTimer = setTimeout(() => { idleTimer = setTimeout(() => {
logger.debug( logger.debug(
{ group: group.name }, { group: group.name },
'Idle timeout, closing container stdin', 'Idle timeout, closing agent stdin',
); );
queue.closeStdin(chatJid); queue.closeStdin(chatJid);
}, IDLE_TIMEOUT); }, IDLE_TIMEOUT);
@@ -307,12 +307,12 @@ async function runAgent(
group: RegisteredGroup, group: RegisteredGroup,
prompt: string, prompt: string,
chatJid: string, chatJid: string,
onOutput?: (output: ContainerOutput) => Promise<void>, onOutput?: (output: AgentOutput) => Promise<void>,
): Promise<'success' | 'error'> { ): Promise<'success' | 'error'> {
const isMain = group.isMain === true; const isMain = group.isMain === true;
const sessionId = sessions[group.folder]; const sessionId = sessions[group.folder];
// Update tasks snapshot for container to read (filtered by group) // Update tasks snapshot for agent to read (filtered by group)
const tasks = getAllTasks(); const tasks = getAllTasks();
writeTasksSnapshot( writeTasksSnapshot(
group.folder, group.folder,
@@ -339,7 +339,7 @@ async function runAgent(
// Wrap onOutput to track session ID from streamed results // Wrap onOutput to track session ID from streamed results
const wrappedOnOutput = onOutput const wrappedOnOutput = onOutput
? async (output: ContainerOutput) => { ? async (output: AgentOutput) => {
if (output.newSessionId) { if (output.newSessionId) {
sessions[group.folder] = output.newSessionId; sessions[group.folder] = output.newSessionId;
setSession(group.folder, output.newSessionId); setSession(group.folder, output.newSessionId);
@@ -349,7 +349,7 @@ async function runAgent(
: undefined; : undefined;
try { try {
const output = await runContainerAgent( const output = await runAgentProcess(
group, group,
{ {
prompt, prompt,
@@ -359,8 +359,8 @@ async function runAgent(
isMain, isMain,
assistantName: ASSISTANT_NAME, assistantName: ASSISTANT_NAME,
}, },
(proc, containerName) => (proc, processName) =>
queue.registerProcess(chatJid, proc, containerName, group.folder), queue.registerProcess(chatJid, proc, processName, group.folder),
wrappedOnOutput, wrappedOnOutput,
); );
@@ -372,7 +372,7 @@ async function runAgent(
if (output.status === 'error') { if (output.status === 'error') {
logger.error( logger.error(
{ group: group.name, error: output.error }, { group: group.name, error: output.error },
'Container agent error', 'Agent process error',
); );
return 'error'; return 'error';
} }
@@ -461,9 +461,9 @@ async function startMessageLoop(): Promise<void> {
); );
if (loopCmdMsg) { if (loopCmdMsg) {
// Only close active container if the sender is authorized — otherwise an // Only close active agent if the sender is authorized — otherwise an
// untrusted user could kill in-flight work by sending /compact (DoS). // untrusted user could kill in-flight work by sending /compact (DoS).
// closeStdin no-ops internally when no container is active. // closeStdin no-ops internally when no agent is active.
if ( if (
isSessionCommandAllowed( isSessionCommandAllowed(
isMainGroup, isMainGroup,
@@ -473,7 +473,7 @@ async function startMessageLoop(): Promise<void> {
queue.closeStdin(chatJid); queue.closeStdin(chatJid);
} }
// Enqueue so processGroupMessages handles auth + cursor advancement. // Enqueue so processGroupMessages handles auth + cursor advancement.
// Don't pipe via IPC — slash commands need a fresh container with // Don't pipe via IPC — slash commands need a fresh agent with
// string prompt (not MessageStream) for SDK recognition. // string prompt (not MessageStream) for SDK recognition.
queue.enqueueMessageCheck(chatJid); queue.enqueueMessageCheck(chatJid);
continue; continue;
@@ -510,19 +510,19 @@ async function startMessageLoop(): Promise<void> {
if (queue.sendMessage(chatJid, formatted)) { if (queue.sendMessage(chatJid, formatted)) {
logger.debug( logger.debug(
{ chatJid, count: messagesToSend.length }, { chatJid, count: messagesToSend.length },
'Piped messages to active container', 'Piped messages to active agent',
); );
lastAgentTimestamp[chatJid] = lastAgentTimestamp[chatJid] =
messagesToSend[messagesToSend.length - 1].timestamp; messagesToSend[messagesToSend.length - 1].timestamp;
saveState(); saveState();
// Show typing indicator while the container processes the piped message // Show typing indicator while the agent processes the piped message
channel channel
.setTyping?.(chatJid, true) .setTyping?.(chatJid, true)
?.catch((err) => ?.catch((err) =>
logger.warn({ chatJid, err }, 'Failed to set typing indicator'), logger.warn({ chatJid, err }, 'Failed to set typing indicator'),
); );
} else { } else {
// No active container — enqueue for a new one // No active agent — enqueue for a new one
queue.enqueueMessageCheck(chatJid, group.folder); queue.enqueueMessageCheck(chatJid, group.folder);
} }
} }
@@ -628,8 +628,8 @@ async function main(): Promise<void> {
registeredGroups: () => registeredGroups, registeredGroups: () => registeredGroups,
getSessions: () => sessions, getSessions: () => sessions,
queue, queue,
onProcess: (groupJid, proc, containerName, groupFolder) => onProcess: (groupJid, proc, processName, groupFolder) =>
queue.registerProcess(groupJid, proc, containerName, groupFolder), queue.registerProcess(groupJid, proc, processName, groupFolder),
sendMessage: async (jid, rawText) => { sendMessage: async (jid, rawText) => {
const channel = findChannel(channels, jid); const channel = findChannel(channels, jid);
if (!channel) { if (!channel) {

6
src/ipc.ts
View File

@@ -4,7 +4,7 @@ import path from 'path';
import { CronExpressionParser } from 'cron-parser'; import { CronExpressionParser } from 'cron-parser';
import { DATA_DIR, IPC_POLL_INTERVAL, TIMEZONE } from './config.js'; import { DATA_DIR, IPC_POLL_INTERVAL, TIMEZONE } from './config.js';
import { AvailableGroup } from './container-runner.js'; import { AvailableGroup } from './agent-runner.js';
import { createTask, deleteTask, getTaskById, updateTask } from './db.js'; import { createTask, deleteTask, getTaskById, updateTask } from './db.js';
import { isValidGroupFolder } from './group-folder.js'; import { isValidGroupFolder } from './group-folder.js';
import { logger } from './logger.js'; import { logger } from './logger.js';
@@ -170,7 +170,7 @@ export async function processTaskIpc(
folder?: string; folder?: string;
trigger?: string; trigger?: string;
requiresTrigger?: boolean; requiresTrigger?: boolean;
containerConfig?: RegisteredGroup['containerConfig']; agentConfig?: RegisteredGroup['agentConfig'];
}, },
sourceGroup: string, // Verified identity from IPC directory sourceGroup: string, // Verified identity from IPC directory
isMain: boolean, // Verified from directory path isMain: boolean, // Verified from directory path
@@ -438,7 +438,7 @@ export async function processTaskIpc(
folder: data.folder, folder: data.folder,
trigger: data.trigger, trigger: data.trigger,
added_at: new Date().toISOString(), added_at: new Date().toISOString(),
containerConfig: data.containerConfig, agentConfig: data.agentConfig,
requiresTrigger: data.requiresTrigger, requiresTrigger: data.requiresTrigger,
}); });
} else { } else {

2
src/session-commands.ts
View File

@@ -26,7 +26,7 @@ export function isSessionCommandAllowed(
return isMainGroup || isFromMe; return isMainGroup || isFromMe;
} }
/** Minimal agent result interface — matches the subset of ContainerOutput used here. */ /** Minimal agent result interface — matches the subset of AgentOutput used here. */
export interface AgentResult { export interface AgentResult {
status: 'success' | 'error'; status: 'success' | 'error';
result?: string | object | null; result?: string | object | null;

22
src/task-scheduler.ts
View File

@@ -4,10 +4,10 @@ import fs from 'fs';
import { ASSISTANT_NAME, SCHEDULER_POLL_INTERVAL, TIMEZONE } from './config.js'; import { ASSISTANT_NAME, SCHEDULER_POLL_INTERVAL, TIMEZONE } from './config.js';
import { import {
ContainerOutput, AgentOutput,
runContainerAgent, runAgentProcess,
writeTasksSnapshot, writeTasksSnapshot,
} from './container-runner.js'; } from './agent-runner.js';
import { import {
getAllTasks, getAllTasks,
getDueTasks, getDueTasks,
@@ -69,7 +69,7 @@ export interface SchedulerDependencies {
onProcess: ( onProcess: (
groupJid: string, groupJid: string,
proc: ChildProcess, proc: ChildProcess,
containerName: string, processName: string,
groupFolder: string, groupFolder: string,
) => void; ) => void;
sendMessage: (jid: string, text: string) => Promise<void>; sendMessage: (jid: string, text: string) => Promise<void>;
@@ -129,7 +129,7 @@ async function runTask(
return; return;
} }
// Update tasks snapshot for container to read (filtered by group) // Update tasks snapshot for agent to read (filtered by group)
const isMain = group.isMain === true; const isMain = group.isMain === true;
const tasks = getAllTasks(); const tasks = getAllTasks();
writeTasksSnapshot( writeTasksSnapshot(
@@ -154,7 +154,7 @@ async function runTask(
const sessionId = const sessionId =
task.context_mode === 'group' ? sessions[task.group_folder] : undefined; task.context_mode === 'group' ? sessions[task.group_folder] : undefined;
// After the task produces a result, close the container promptly. // After the task produces a result, close the agent promptly.
// Tasks are single-turn — no need to wait IDLE_TIMEOUT (30 min) for the // Tasks are single-turn — no need to wait IDLE_TIMEOUT (30 min) for the
// query loop to time out. A short delay handles any final MCP calls. // query loop to time out. A short delay handles any final MCP calls.
const TASK_CLOSE_DELAY_MS = 10000; const TASK_CLOSE_DELAY_MS = 10000;
@@ -163,13 +163,13 @@ async function runTask(
const scheduleClose = () => { const scheduleClose = () => {
if (closeTimer) return; // already scheduled if (closeTimer) return; // already scheduled
closeTimer = setTimeout(() => { closeTimer = setTimeout(() => {
logger.debug({ taskId: task.id }, 'Closing task container after result'); logger.debug({ taskId: task.id }, 'Closing task agent after result');
deps.queue.closeStdin(task.chat_jid); deps.queue.closeStdin(task.chat_jid);
}, TASK_CLOSE_DELAY_MS); }, TASK_CLOSE_DELAY_MS);
}; };
try { try {
const output = await runContainerAgent( const output = await runAgentProcess(
group, group,
{ {
prompt: task.prompt, prompt: task.prompt,
@@ -180,9 +180,9 @@ async function runTask(
isScheduledTask: true, isScheduledTask: true,
assistantName: ASSISTANT_NAME, assistantName: ASSISTANT_NAME,
}, },
(proc, containerName) => (proc, processName) =>
deps.onProcess(task.chat_jid, proc, containerName, task.group_folder), deps.onProcess(task.chat_jid, proc, processName, task.group_folder),
async (streamedOutput: ContainerOutput) => { async (streamedOutput: AgentOutput) => {
if (streamedOutput.result) { if (streamedOutput.result) {
result = streamedOutput.result; result = streamedOutput.result;
// Forward result to user (sendMessage handles formatting) // Forward result to user (sendMessage handles formatting)

17
src/token-refresh.ts
View File

@@ -81,17 +81,28 @@ function syncToSessionDirs(credsPath: string): void {
const groups = fs.readdirSync(sessionsDir); const groups = fs.readdirSync(sessionsDir);
let synced = 0; let synced = 0;
for (const group of groups) { for (const group of groups) {
const dest = path.join(sessionsDir, group, '.claude', '.credentials.json'); const dest = path.join(
sessionsDir,
group,
'.claude',
'.credentials.json',
);
if (fs.existsSync(path.dirname(dest))) { if (fs.existsSync(path.dirname(dest))) {
fs.copyFileSync(credsPath, dest); fs.copyFileSync(credsPath, dest);
synced++; synced++;
} }
} }
if (synced > 0) { if (synced > 0) {
logger.info({ count: synced }, 'Synced credentials to session directories'); logger.info(
{ count: synced },
'Synced credentials to session directories',
);
} }
} catch (err) { } catch (err) {
logger.warn({ err: err instanceof Error ? err.message : String(err) }, 'Failed to sync credentials to sessions'); logger.warn(
{ err: err instanceof Error ? err.message : String(err) },
'Failed to sync credentials to sessions',
);
} }
} }

4
src/types.ts
View File

@@ -4,7 +4,7 @@ export interface AdditionalMount {
readonly?: boolean; // Default: true for safety readonly?: boolean; // Default: true for safety
} }
export interface ContainerConfig { export interface AgentConfig {
additionalMounts?: AdditionalMount[]; additionalMounts?: AdditionalMount[];
timeout?: number; // Default: 300000 (5 minutes) timeout?: number; // Default: 300000 (5 minutes)
// Per-group model/effort overrides (take precedence over global env vars) // Per-group model/effort overrides (take precedence over global env vars)
@@ -21,7 +21,7 @@ export interface RegisteredGroup {
folder: string; folder: string;
trigger: string; trigger: string;
added_at: string; added_at: string;
containerConfig?: ContainerConfig; agentConfig?: AgentConfig;
requiresTrigger?: boolean; // Default: true for groups, false for solo chats requiresTrigger?: boolean; // Default: true for groups, false for solo chats
isMain?: boolean; // True for the main control group (no trigger, elevated privileges) isMain?: boolean; // True for the main control group (no trigger, elevated privileges)
agentType?: AgentType; agentType?: AgentType;