diff --git a/.gitignore b/.gitignore index 5d89d6f..60ecf65 100644 --- a/.gitignore +++ b/.gitignore @@ -11,6 +11,7 @@ store-*/ data/ data-*/ logs/ +.ejclaw-reviewer-runtime/ # Groups - only track base structure and specific CLAUDE.md files groups-*/ diff --git a/runners/agent-runner/package.json b/runners/agent-runner/package.json index e183357..6715457 100644 --- a/runners/agent-runner/package.json +++ b/runners/agent-runner/package.json @@ -2,7 +2,7 @@ "name": "ejclaw-agent-runner", "version": "1.0.0", "type": "module", - "description": "Container-side agent runner for EJClaw", + "description": "Agent runner subprocess for EJClaw", "main": "dist/index.js", "scripts": { "build": "tsc", diff --git a/runners/agent-runner/src/index.ts b/runners/agent-runner/src/index.ts index 4478502..949762e 100644 --- a/runners/agent-runner/src/index.ts +++ b/runners/agent-runner/src/index.ts @@ -34,7 +34,7 @@ import { } from './reviewer-runtime.js'; import { selectCompactMemoriesFromSummary } from './memory-selection.js'; -interface ContainerInput { +interface RunnerInput { prompt: string; sessionId?: string; groupFolder: string; @@ -47,7 +47,7 @@ interface ContainerInput { } /** Mirrors AgentOutput in src/agent-runner.ts (separate package, can't import directly). */ -interface ContainerOutput { +interface RunnerOutput { status: 'success' | 'error'; phase?: 'progress' | 'final' | 'tool-activity' | 'intermediate'; agentId?: string; @@ -199,7 +199,7 @@ async function readStdin(): Promise { const OUTPUT_START_MARKER = '---EJCLAW_OUTPUT_START---'; const OUTPUT_END_MARKER = '---EJCLAW_OUTPUT_END---'; -function writeOutput(output: ContainerOutput): void { +function writeOutput(output: RunnerOutput): void { console.log(OUTPUT_START_MARKER); console.log(JSON.stringify(output)); console.log(OUTPUT_END_MARKER); @@ -207,7 +207,7 @@ function writeOutput(output: ContainerOutput): void { function normalizeStructuredOutput(result: string | null): { result: string | null; - output?: ContainerOutput['output']; + output?: RunnerOutput['output']; } { if (typeof result !== 'string' || result.length === 0) { return { result }; @@ -545,7 +545,7 @@ async function runQuery( prompt: string, sessionId: string | undefined, mcpServerPath: string, - containerInput: ContainerInput, + runnerInput: RunnerInput, sdkEnv: Record, reviewerRuntime: boolean, claudeReadonlyReviewerRuntime: boolean, @@ -698,12 +698,12 @@ async function runQuery( command: 'node', args: [mcpServerPath], env: { - EJCLAW_CHAT_JID: containerInput.chatJid, - EJCLAW_GROUP_FOLDER: containerInput.groupFolder, - EJCLAW_IS_MAIN: containerInput.isMain ? '1' : '0', + EJCLAW_CHAT_JID: runnerInput.chatJid, + EJCLAW_GROUP_FOLDER: runnerInput.groupFolder, + EJCLAW_IS_MAIN: runnerInput.isMain ? '1' : '0', EJCLAW_AGENT_TYPE: process.env.EJCLAW_AGENT_TYPE || 'claude-code', - EJCLAW_ROOM_ROLE: containerInput.roomRoleContext?.role || '', + EJCLAW_ROOM_ROLE: runnerInput.roomRoleContext?.role || '', ...(process.env.EJCLAW_IPC_DIR && { EJCLAW_IPC_DIR: process.env.EJCLAW_IPC_DIR, }), @@ -714,7 +714,7 @@ async function runQuery( }, }, hooks: { - PreCompact: [{ hooks: [createPreCompactHook(containerInput.assistantName)] }], + PreCompact: [{ hooks: [createPreCompactHook(runnerInput.assistantName)] }], PreToolUse: [ { matcher: 'Bash', @@ -958,14 +958,14 @@ async function runQuery( } async function main(): Promise { - let containerInput: ContainerInput; + let runnerInput: RunnerInput; try { const stdinData = await readStdin(); - containerInput = JSON.parse(stdinData); + runnerInput = JSON.parse(stdinData); // Delete the temp file the entrypoint wrote — it contains secrets try { fs.unlinkSync('/tmp/input.json'); } catch { /* may not exist */ } - log(`Received input for group: ${containerInput.groupFolder}`); + log(`Received input for group: ${runnerInput.groupFolder}`); } catch (err) { writeOutput({ status: 'error', @@ -978,14 +978,14 @@ async function main(): Promise { // Build SDK env: merge secrets into process.env for the SDK only. // Secrets never touch process.env itself, so Bash subprocesses can't see them. const sdkEnv: Record = { ...process.env }; - for (const [key, value] of Object.entries(containerInput.secrets || {})) { + for (const [key, value] of Object.entries(runnerInput.secrets || {})) { sdkEnv[key] = value; } const reviewerRuntime = process.env.EJCLAW_REVIEWER_RUNTIME === '1' || - isReviewerRuntime(containerInput.roomRoleContext); + isReviewerRuntime(runnerInput.roomRoleContext); const claudeReadonlyReviewerRuntime = - isClaudeReadonlyReviewerRuntime(containerInput.roomRoleContext); + isClaudeReadonlyReviewerRuntime(runnerInput.roomRoleContext); const claudeReadonlySandboxMode = claudeReadonlyReviewerRuntime ? getClaudeReadonlySandboxMode() : null; @@ -1002,18 +1002,18 @@ async function main(): Promise { const __dirname = path.dirname(fileURLToPath(import.meta.url)); const mcpServerPath = path.join(__dirname, 'ipc-mcp-stdio.js'); - let sessionId = containerInput.sessionId; + let sessionId = runnerInput.sessionId; fs.mkdirSync(IPC_INPUT_DIR, { recursive: true }); - // Clean up stale _close sentinel from previous container runs + // Clean up stale _close sentinel from previous runner sessions try { fs.unlinkSync(IPC_INPUT_CLOSE_SENTINEL); } catch { /* ignore */ } // Effective working directory (WORK_DIR overrides GROUP_DIR) const mainEffectiveCwd = WORK_DIR || GROUP_DIR; // Build initial prompt (drain any pending IPC messages too) - let prompt = containerInput.prompt; - if (containerInput.isScheduledTask) { + let prompt = runnerInput.prompt; + if (runnerInput.isScheduledTask) { prompt = `[SCHEDULED TASK - The following message was sent automatically and is not coming directly from the user or group.]\n\n${prompt}`; } const pending = drainIpcInput(); @@ -1024,10 +1024,10 @@ async function main(): Promise { // --- Slash command handling --- // Check BEFORE prepending room role header so /compact isn't masked. const KNOWN_SESSION_COMMANDS = new Set(['/compact']); - const isSessionSlashCommand = KNOWN_SESSION_COMMANDS.has(containerInput.prompt.trim()); + const isSessionSlashCommand = KNOWN_SESSION_COMMANDS.has(runnerInput.prompt.trim()); if (!isSessionSlashCommand) { - prompt = prependRoomRoleHeader(prompt, containerInput.roomRoleContext); + prompt = prependRoomRoleHeader(prompt, runnerInput.roomRoleContext); } const trimmedPrompt = prompt.trim(); @@ -1052,7 +1052,7 @@ async function main(): Promise { settingSources: ['project', 'user'] as const, abortController: agentAbortController, hooks: { - PreCompact: [{ hooks: [createPreCompactHook(containerInput.assistantName)] }], + PreCompact: [{ hooks: [createPreCompactHook(runnerInput.assistantName)] }], }, }, })) { @@ -1133,7 +1133,7 @@ async function main(): Promise { prompt, sessionId, mcpServerPath, - containerInput, + runnerInput, guardedSdkEnv, reviewerRuntime, claudeReadonlyReviewerRuntime, diff --git a/runners/codex-runner/src/index.ts b/runners/codex-runner/src/index.ts index 7bb37ec..bfbdbf0 100644 --- a/runners/codex-runner/src/index.ts +++ b/runners/codex-runner/src/index.ts @@ -4,7 +4,7 @@ * App-server only runtime. * * Input protocol: - * Stdin: Full ContainerInput JSON (read until EOF) + * Stdin: Full RunnerInput JSON (read until EOF) * IPC: Follow-up messages as JSON files in $EJCLAW_IPC_DIR/input/ * Sentinel: _close — signals session end * @@ -31,7 +31,7 @@ import { // ── Types ────────────────────────────────────────────────────────── -interface ContainerInput { +interface RunnerInput { prompt: string; sessionId?: string; groupFolder: string; @@ -43,7 +43,7 @@ interface ContainerInput { roomRoleContext?: RoomRoleContext; } -interface ContainerOutput { +interface RunnerOutput { status: 'success' | 'error'; result: string | null; output?: { @@ -77,7 +77,7 @@ let closeRequested = false; // ── Helpers ──────────────────────────────────────────────────────── -function writeOutput(output: ContainerOutput): void { +function writeOutput(output: RunnerOutput): void { console.log(OUTPUT_START_MARKER); console.log(JSON.stringify(output)); console.log(OUTPUT_END_MARKER); @@ -85,7 +85,7 @@ function writeOutput(output: ContainerOutput): void { function normalizeStructuredOutput(result: string | null): { result: string | null; - output?: ContainerOutput['output']; + output?: RunnerOutput['output']; } { if (typeof result !== 'string' || result.length === 0) { return { result }; @@ -400,12 +400,12 @@ async function runAppServerCompact( } async function runAppServerSession( - containerInput: ContainerInput, + runnerInput: RunnerInput, prompt: string, ): Promise { const reviewerRuntime = process.env.EJCLAW_REVIEWER_RUNTIME === '1' || - isReviewerRuntime(containerInput.roomRoleContext); + isReviewerRuntime(runnerInput.roomRoleContext); const readonlyRuntime = reviewerRuntime || process.env.EJCLAW_ARBITER_RUNTIME === '1'; const clientEnv = buildReviewerGitGuardEnv(process.env, reviewerRuntime); @@ -421,17 +421,17 @@ async function runAppServerSession( let threadId: string | undefined; try { try { - threadId = await client.startOrResumeThread(containerInput.sessionId, { + threadId = await client.startOrResumeThread(runnerInput.sessionId, { cwd: EFFECTIVE_CWD, model: CODEX_MODEL || undefined, }); log( - containerInput.sessionId + runnerInput.sessionId ? `App-server thread resumed (${threadId})` : `App-server thread started (${threadId})`, ); } catch (err) { - if (!containerInput.sessionId) throw err; + if (!runnerInput.sessionId) throw err; log( `App-server resume failed, retrying with new thread: ${ err instanceof Error ? err.message : String(err) @@ -483,17 +483,17 @@ async function runAppServerSession( // ── Main ────────────────────────────────────────────────────────── async function main(): Promise { - let containerInput: ContainerInput; + let runnerInput: RunnerInput; try { const stdinData = await readStdin(); - containerInput = JSON.parse(stdinData); + runnerInput = JSON.parse(stdinData); try { fs.unlinkSync('/tmp/input.json'); } catch { /* may not exist */ } - log(`Received input for group: ${containerInput.groupFolder}`); + log(`Received input for group: ${runnerInput.groupFolder}`); } catch (err) { writeOutput({ status: 'error', @@ -513,9 +513,9 @@ async function main(): Promise { /* ignore */ } - const rawPrompt = containerInput.prompt; + const rawPrompt = runnerInput.prompt; let prompt = rawPrompt; - if (containerInput.isScheduledTask) { + if (runnerInput.isScheduledTask) { prompt = `[SCHEDULED TASK]\n\n${prompt}`; } const pending = drainIpcInput(); @@ -526,12 +526,12 @@ async function main(): Promise { // so /compact is not masked by the header prefix. const isSessionCommand = rawPrompt.trim() === '/compact'; if (!isSessionCommand) { - prompt = prependRoomRoleHeader(prompt, containerInput.roomRoleContext); + prompt = prependRoomRoleHeader(prompt, runnerInput.roomRoleContext); } try { log('Runtime selected: app-server'); - await runAppServerSession(containerInput, prompt); + await runAppServerSession(runnerInput, prompt); } catch (err) { const errorMessage = err instanceof Error ? err.message : String(err); log(`Runner error: ${errorMessage}`); diff --git a/src/agent-runner-environment.test.ts b/src/agent-runner-environment.test.ts index 386cfdd..ee8af84 100644 --- a/src/agent-runner-environment.test.ts +++ b/src/agent-runner-environment.test.ts @@ -79,7 +79,7 @@ vi.mock('os', async () => { }); import { - prepareContainerSessionEnvironment, + prepareReadonlySessionEnvironment, prepareGroupEnvironment, } from './agent-runner-environment.js'; import * as config from './config.js'; @@ -341,12 +341,12 @@ describe('prepareGroupEnvironment codex auth handling', () => { }); }); -describe('prepareContainerSessionEnvironment codex compatibility', () => { +describe('prepareReadonlySessionEnvironment codex compatibility', () => { let tempRoot: string; let previousCwd: string; beforeEach(() => { - tempRoot = fs.mkdtempSync(path.join('/tmp', 'ejclaw-container-env-')); + tempRoot = fs.mkdtempSync(path.join('/tmp', 'ejclaw-readonly-env-')); previousCwd = process.cwd(); process.chdir(tempRoot); @@ -369,7 +369,7 @@ describe('prepareContainerSessionEnvironment codex compatibility', () => { fs.rmSync(tempRoot, { recursive: true, force: true }); }); - it('writes matching AGENTS.md and copies host codex auth/config into the container session', () => { + it('writes matching AGENTS.md and copies host codex auth/config into the role-scoped session', () => { vi.mocked(serviceRouting.hasReviewerLease).mockReturnValue(true); const promptsDir = path.join(tempRoot, 'prompts'); @@ -404,8 +404,8 @@ args = ["other.js"] `, ); - const sessionDir = path.join(tempRoot, 'container-reviewer-session'); - prepareContainerSessionEnvironment({ + const sessionDir = path.join(tempRoot, 'readonly-reviewer-session'); + prepareReadonlySessionEnvironment({ sessionDir, chatJid: 'dc:test', isMain: false, diff --git a/src/agent-runner-environment.ts b/src/agent-runner-environment.ts index 9b5c1f9..4137ed0 100644 --- a/src/agent-runner-environment.ts +++ b/src/agent-runner-environment.ts @@ -487,7 +487,7 @@ export function prepareGroupEnvironment( : undefined; // Owner CLAUDE.md: platform rules + owner paired room rules. // Reviewer paired room rules are NOT included — those belong to the - // container reviewer only (via prepareContainerSessionEnvironment). + // Read-only reviewer/arbiter session only (via prepareReadonlySessionEnvironment). const sessionClaudeMd = [ ownerCommonPlatformPrompt, claudePlatformPrompt, @@ -558,14 +558,15 @@ export function prepareGroupEnvironment( } /** - * Prepare the Claude session directory for a container-based reviewer. + * Prepare a role-scoped session directory for a host-based read-only reviewer + * or arbiter run. * * Writes CLAUDE.md (platform + paired room prompts + global memory + briefing), * syncs skills, and ensures settings.json exist — the same steps that - * `prepareGroupEnvironment` does for host-mode agents, but targeted at an - * externally provided session directory (the one mounted into the container). + * `prepareGroupEnvironment` does for regular host-mode agents, but targeted at + * an externally provided session directory. */ -export function prepareContainerSessionEnvironment(args: { +export function prepareReadonlySessionEnvironment(args: { sessionDir: string; chatJid: string; isMain: boolean; @@ -665,7 +666,7 @@ export function prepareContainerSessionEnvironment(args: { hasGlobalMemory: !!globalClaudeMemory, hasMemoryBriefing: !!memoryBriefing, }, - 'Container session CLAUDE.md written', + 'Readonly session CLAUDE.md written', ); } else if (fs.existsSync(sessionClaudeMdPath)) { fs.unlinkSync(sessionClaudeMdPath); diff --git a/src/agent-runner.test.ts b/src/agent-runner.test.ts index 7b8ad05..c2ba3ad 100644 --- a/src/agent-runner.test.ts +++ b/src/agent-runner.test.ts @@ -318,9 +318,9 @@ describe('agent-runner timeout behavior', () => { it('uses host IPC and workdir in unsafe host paired mode reviewer sessions', async () => { vi.useRealTimers(); fakeProc = createFakeProcess(); - const prepareContainerSessionEnvironmentSpy = vi.spyOn( + const prepareReadonlySessionEnvironmentSpy = vi.spyOn( agentRunnerEnvironment, - 'prepareContainerSessionEnvironment', + 'prepareReadonlySessionEnvironment', ); vi.mocked(fs.existsSync).mockImplementation((p: fs.PathLike) => { @@ -356,7 +356,7 @@ describe('agent-runner timeout behavior', () => { fakeProc.emit('close', 0); const result = await resultPromise; expect(result.status).toBe('success'); - expect(prepareContainerSessionEnvironmentSpy).toHaveBeenCalledWith( + expect(prepareReadonlySessionEnvironmentSpy).toHaveBeenCalledWith( expect.objectContaining({ sessionDir: '/tmp/host-reviewer-session', ipcDir: '/tmp/ejclaw-test-data/ipc/test-group', diff --git a/src/agent-runner.ts b/src/agent-runner.ts index e275282..51aa9bb 100644 --- a/src/agent-runner.ts +++ b/src/agent-runner.ts @@ -13,7 +13,7 @@ import { IDLE_TIMEOUT, } from './config.js'; import { - prepareContainerSessionEnvironment, + prepareReadonlySessionEnvironment, prepareGroupEnvironment, } from './agent-runner-environment.js'; import { getEnv } from './env.js'; @@ -99,7 +99,7 @@ export async function runAgentProcess( (input.roomRoleContext?.role === 'reviewer' || input.roomRoleContext?.role === 'arbiter') ) { - prepareContainerSessionEnvironment({ + prepareReadonlySessionEnvironment({ sessionDir: envOverrides.CLAUDE_CONFIG_DIR, chatJid: input.chatJid, isMain: input.isMain, diff --git a/src/message-agent-executor.test.ts b/src/message-agent-executor.test.ts index 240e139..4522e53 100644 --- a/src/message-agent-executor.test.ts +++ b/src/message-agent-executor.test.ts @@ -1237,6 +1237,83 @@ describe('runAgentForGroup room memory', () => { expect(result).toBe('success'); expect(deps.queue.enqueueMessageCheck).not.toHaveBeenCalled(); }); + + it('re-enqueues the group when a reviewer fails and the task remains review_ready', async () => { + const group = { ...makeGroup(), folder: 'test-group' }; + const deps = makeDeps(); + + vi.mocked(serviceRouting.getEffectiveChannelLease).mockReturnValue({ + chat_jid: 'group@test', + owner_agent_type: 'claude-code', + reviewer_agent_type: 'claude-code', + arbiter_agent_type: null, + owner_service_id: 'claude', + reviewer_service_id: 'claude', + arbiter_service_id: null, + activated_at: null, + reason: null, + explicit: false, + }); + vi.mocked( + pairedExecutionContext.preparePairedExecutionContext, + ).mockReturnValue({ + task: { + id: 'paired-task-review-ready', + chat_jid: 'group@test', + group_folder: 'test-group', + owner_service_id: 'claude', + reviewer_service_id: 'claude', + title: null, + source_ref: 'HEAD', + plan_notes: null, + round_trip_count: 1, + review_requested_at: '2026-03-31T00:00:00.000Z', + status: 'in_review', + arbiter_verdict: null, + arbiter_requested_at: null, + completion_reason: null, + created_at: '2026-03-31T00:00:00.000Z', + updated_at: '2026-03-31T00:00:00.000Z', + }, + workspace: null, + envOverrides: {}, + }); + vi.mocked(db.getPairedTaskById).mockReturnValue({ + id: 'paired-task-review-ready', + chat_jid: 'group@test', + group_folder: 'test-group', + owner_service_id: 'claude', + reviewer_service_id: 'claude', + title: null, + source_ref: 'HEAD', + plan_notes: null, + round_trip_count: 1, + review_requested_at: '2026-03-31T00:00:00.000Z', + status: 'review_ready', + arbiter_verdict: null, + arbiter_requested_at: null, + completion_reason: null, + created_at: '2026-03-31T00:00:00.000Z', + updated_at: '2026-03-31T00:00:00.000Z', + }); + vi.mocked(agentRunner.runAgentProcess).mockResolvedValue({ + status: 'error', + result: null, + error: 'SDK crashed with exit code 1', + }); + + const result = await runAgentForGroup(deps, { + group, + prompt: 'please review', + chatJid: 'group@test', + runId: 'run-review-ready-requeue', + forcedRole: 'reviewer', + onOutput: async () => {}, + }); + + expect(result).toBe('error'); + expect(deps.queue.enqueueMessageCheck).toHaveBeenCalledWith('group@test'); + }); }); describe('runAgentForGroup Claude rotation', () => { diff --git a/src/message-agent-executor.ts b/src/message-agent-executor.ts index 37ebe3b..3d207f1 100644 --- a/src/message-agent-executor.ts +++ b/src/message-agent-executor.ts @@ -1165,6 +1165,28 @@ export async function runAgentForGroup( } else if (finishedCheck?.status !== 'completed') { deps.queue.enqueueMessageCheck(chatJid); } + } else if (pairedExecutionContext) { + const completedRole = roomRoleContext?.role ?? 'owner'; + const finishedCheck = getPairedTaskById(pairedExecutionContext.task.id); + const shouldRequeuePendingPairedTurn = + (completedRole === 'reviewer' || completedRole === 'arbiter') && + (finishedCheck?.status === 'review_ready' || + finishedCheck?.status === 'in_review' || + finishedCheck?.status === 'arbiter_requested' || + finishedCheck?.status === 'in_arbitration' || + finishedCheck?.status === 'merge_ready'); + if (shouldRequeuePendingPairedTurn) { + deps.queue.enqueueMessageCheck(chatJid); + log.info( + { + taskId: pairedExecutionContext.task.id, + role: completedRole, + pairedExecutionStatus, + taskStatus: finishedCheck?.status ?? null, + }, + 'Queued paired follow-up after failed reviewer/arbiter execution left a pending task state', + ); + } } } }