From 16d20fe6275412a2b7085d5c727deb3fd2c28292 Mon Sep 17 00:00:00 2001 From: Eyejoker Date: Sun, 29 Mar 2026 20:50:32 +0900 Subject: [PATCH] =?UTF-8?q?feat:=20paired=20review=20system=20overhaul=20?= =?UTF-8?q?=E2=80=94=20unified=20service,=20configurable=20agents,=20silen?= =?UTF-8?q?t=20output=20removal?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Remove UNIFIED_MODE legacy: single service manages all 3 Discord bots - Add OWNER_AGENT_TYPE / REVIEWER_AGENT_TYPE env vars for configurable agent selection - Fix Discord channel routing: reviewer output goes through correct bot (discord/discord-review) - Fix channel name assignment: explicit names prevent agentTypeFilter collision - Remove silent output suppression system: harness-level protections replace prompt workarounds - Add reviewer approval detection: DONE marker on first line stops ping-pong - Include user's original message in reviewer prompt for context - Fix round_trip_count auto-reset on new user message - Fix session ID conflict: reviewer doesn't reuse owner's session - Fix pending progress text flush in runner on close sentinel - Promote buffered intermediate text to final result when result event has no text - Remove legacy files: .env.codex.example, .env.codex-review.example, migrate-unify.cjs - Update CLAUDE.md: server-side build deployment, unified architecture docs --- .env.codex-review.example | 5 - .env.codex.example | 7 - CLAUDE.md | 33 ++--- runners/agent-runner/src/index.ts | 106 +++++--------- scripts/migrate-unify.cjs | 220 ----------------------------- src/agent-output.ts | 13 +- src/agent-runner.ts | 3 +- src/channels/discord.ts | 16 ++- src/config.ts | 22 ++- src/db.ts | 12 ++ src/index.ts | 78 ++++------ src/message-agent-executor.test.ts | 1 + src/message-agent-executor.ts | 77 ++++++---- src/message-runtime-rules.ts | 14 +- src/message-runtime.ts | 118 +++++++++------- src/message-turn-controller.ts | 36 +---- src/output-suppression.ts | 132 +++-------------- src/paired-execution-context.ts | 52 +++++-- src/service-routing.ts | 22 ++- 19 files changed, 325 insertions(+), 642 deletions(-) delete mode 100644 .env.codex-review.example delete mode 100644 .env.codex.example delete mode 100644 scripts/migrate-unify.cjs diff --git a/.env.codex-review.example b/.env.codex-review.example deleted file mode 100644 index cf8b503..0000000 --- a/.env.codex-review.example +++ /dev/null @@ -1,5 +0,0 @@ -# ===== EJClaw Codex Review Service Environment ===== -# Copy this file to .env.codex-review for the ejclaw-review service. -# This service needs its own Discord bot token. - -DISCORD_BOT_TOKEN= # Discord bot token (Codex review bot) diff --git a/.env.codex.example b/.env.codex.example deleted file mode 100644 index 198e5be..0000000 --- a/.env.codex.example +++ /dev/null @@ -1,7 +0,0 @@ -# ===== EJClaw Codex Service Environment ===== -# Copy this file to .env.codex for the ejclaw-codex service. -# This file is loaded via EnvironmentFile in the systemd unit. -# The main .env is NOT loaded by the codex service — put shared vars there and -# set them as Environment= lines in the systemd unit, or duplicate them here. - -DISCORD_BOT_TOKEN= # Discord bot token (Codex bot — separate from Claude bot) diff --git a/CLAUDE.md b/CLAUDE.md index 7681869..8112109 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -4,7 +4,7 @@ Dual-agent AI assistant (Claude Code + Codex) over Discord. Originally derived f ## Quick Context -Two systemd services (`ejclaw`, `ejclaw-codex`) share the same codebase but run with separate stores, data, and groups (will be unified — DB supports shared access via WAL mode + service partitioning). Agents run as direct host processes (no containers). Claude Code uses the Agent SDK; Codex uses the Codex SDK (`codex exec`). Auth via `CLAUDE_CODE_OAUTH_TOKEN` in `.env` (1-year token from `claude setup-token`). +Single unified service (`ejclaw`) manages three Discord bots (Claude, Codex-main, Codex-review) in one process. Owner/reviewer agent types are configurable via `OWNER_AGENT_TYPE` and `REVIEWER_AGENT_TYPE` in `.env`. Claude Code uses the Agent SDK; Codex uses the Codex SDK (`codex exec`). Auth via `CLAUDE_CODE_OAUTH_TOKEN` in `.env` (1-year token from `claude setup-token`). ## Key Files @@ -45,28 +45,25 @@ npm run dev # Dev mode with hot reload Service management (Linux): ```bash -npm run restart:stack # Restart the configured stack and verify it -systemctl --user restart ejclaw ejclaw-codex ejclaw-review +systemctl --user restart ejclaw # Restart unified service systemctl --user status ejclaw # Check status journalctl --user -u ejclaw -f # Follow logs ``` -`npm run restart:stack` is an external deploy/operator entrypoint. Do not invoke it from inside a managed EJClaw service process. -If you pulled this slice onto an existing install, run `npm run setup -- --step service` once so `ejclaw-stack-restart.service` is installed. - -Deploy to server: `scp dist/*.js clone-ej@100.64.185.108:~/EJClaw/dist/` +Deploy to server (build on server, not locally): +```bash +ssh clone-ej@100.64.185.108 'cd ~/EJClaw && git pull && npm run build && npm run build:runners && systemctl --user restart ejclaw' +``` ## Service Stack Architecture -- `ejclaw.service` — Claude Code bot (`@claude`), `SERVICE_ID=claude`, `SERVICE_AGENT_TYPE=claude-code` -- `ejclaw-codex.service` — Codex bot (`@codex`), `SERVICE_ID=codex-main`, `SERVICE_AGENT_TYPE=codex` -- `ejclaw-review.service` — Codex reviewer bot (`@codex-review`), `SERVICE_ID=codex-review`, `SERVICE_AGENT_TYPE=codex` -- All services share the same codebase (`dist/index.js`), differentiated by env vars -- Unified dirs (`store/`, `groups/`, `data/` shared by all services): - - `router_state`: keys prefixed with `{SERVICE_ID}:` (e.g., `claude:last_timestamp`) - - `sessions`: composite PK `(group_folder, agent_type)` - - `registered_groups`: filtered by `agent_type` on load - - SQLite WAL mode + `busy_timeout=5000` for concurrent access +Single unified service manages all three Discord bots in one process: +- `ejclaw.service` — Unified process, `UNIFIED_MODE=true` (default) +- Discord bots: `DISCORD_BOT_TOKEN` (Claude), `DISCORD_CODEX_BOT_TOKEN` (Codex-main), `DISCORD_REVIEW_BOT_TOKEN` (Codex-review) +- Paired review: owner (`OWNER_AGENT_TYPE`, default: codex) ↔ reviewer (`REVIEWER_AGENT_TYPE`, default: claude-code) +- Reviewer fallback: Claude 429/한도초과 시 codex-review로 자동 핸드오프 +- Shared dirs: `store/`, `groups/`, `data/` +- SQLite WAL mode + `busy_timeout=5000` for concurrent access ## Debugging Paths (Server: clone-ej@100.64.185.108) @@ -75,9 +72,7 @@ Unified DB + directories (both services share `store/`, `groups/`, `data/`): | 항목 | 경로 | |------|------| | **DB** | `store/messages.db` (공유, WAL 모드) | -| 서비스 로그 (Claude) | `journalctl --user -u ejclaw -f` 또는 `logs/ejclaw.log` | -| 서비스 로그 (Codex) | `journalctl --user -u ejclaw-codex -f` 또는 `logs/ejclaw-codex.log` | -| 서비스 로그 (Review) | `journalctl --user -u ejclaw-review -f` 또는 `logs/ejclaw-review.log` | +| 서비스 로그 | `journalctl --user -u ejclaw -f` 또는 `logs/ejclaw.log` | | 그룹별 로그 | `groups/{name}/logs/` (공유 채널은 양쪽 봇 로그가 같은 폴더) | | Claude 세션 | `data/sessions/{name}/.claude/` | | Codex 세션 | `data/sessions/{name}/.codex/` | diff --git a/runners/agent-runner/src/index.ts b/runners/agent-runner/src/index.ts index 81eda16..56a211d 100644 --- a/runners/agent-runner/src/index.ts +++ b/runners/agent-runner/src/index.ts @@ -214,73 +214,8 @@ function normalizeStructuredOutput(result: string | null): { if (typeof result !== 'string' || result.length === 0) { return { result }; } - - const trimmed = result.trim(); - try { - const parsed = JSON.parse(trimmed) as { - ejclaw?: { visibility?: unknown; text?: unknown; verdict?: unknown }; - }; - const envelope = parsed?.ejclaw; - if (envelope && typeof envelope === 'object' && !Array.isArray(envelope)) { - if (envelope.visibility === 'silent') { - if ( - envelope.verdict !== undefined && - envelope.verdict !== 'silent' - ) { - return { - result, - output: { visibility: 'public', text: result }, - }; - } - return { - result: null, - output: { - visibility: 'silent', - verdict: - envelope.verdict === 'silent' ? ('silent' as const) : undefined, - }, - }; - } - if ( - envelope.visibility === 'public' && - typeof envelope.text === 'string' && - envelope.text.length > 0 - ) { - if ( - envelope.verdict !== undefined && - envelope.verdict !== 'done' && - envelope.verdict !== 'done_with_concerns' && - envelope.verdict !== 'blocked' - ) { - return { - result, - output: { visibility: 'public', text: result }, - }; - } - return { - result: envelope.text, - output: { - visibility: 'public', - text: envelope.text, - verdict: - typeof envelope.verdict === 'string' - ? (envelope.verdict as - | 'done' - | 'done_with_concerns' - | 'blocked') - : undefined, - }, - }; - } - } - } catch { - // fall through to legacy string output - } - - return { - result, - output: { visibility: 'public', text: result }, - }; + // All output is public — silent suppression was removed. + return { result, output: { visibility: 'public', text: result } }; } function log(message: string): void { @@ -677,6 +612,19 @@ async function runQuery( if (!ipcPolling) return; if (shouldClose()) { log('Close sentinel detected during query, ending stream'); + // Flush any buffered text before closing — the for-await loop may not + // reach the post-loop flush code after stream.end(). + if (pendingProgressText && !terminalResultObserved) { + log(`Flushing pending text before close (${pendingProgressText.length} chars)`); + writeOutput({ + status: 'success', + ...normalizeStructuredOutput(pendingProgressText), + newSessionId, + }); + pendingProgressText = null; + terminalResultObserved = true; + resultCount++; + } closedDuringQuery = true; stream.end(); ipcPolling = false; @@ -916,14 +864,21 @@ async function runQuery( if (message.type === 'result') { resultCount++; - const textResult = 'result' in message ? (message as { result?: string }).result : null; + let textResult = 'result' in message ? (message as { result?: string }).result : null; const isError = message.subtype?.startsWith('error'); // Discard pending progress if it matches the final result (prevent duplicate) if (pendingProgressText && textResult && pendingProgressText === textResult) { log(`Discarding pending progress (matches result)`); pendingProgressText = null; } else if (pendingProgressText) { - writeOutput({ status: 'success', phase: 'intermediate', result: pendingProgressText, newSessionId }); + // If the result has no text, promote pending progress to the result + // so it gets delivered as the final output instead of being lost. + if (!textResult) { + log(`Promoting pending progress text to result (${pendingProgressText.length} chars)`); + textResult = pendingProgressText; + } else { + writeOutput({ status: 'success', phase: 'intermediate', result: pendingProgressText, newSessionId }); + } pendingProgressText = null; } log(`Result #${resultCount}: subtype=${message.subtype}${textResult ? ` text=${textResult.slice(0, 200)}` : ''}`); @@ -1011,6 +966,19 @@ async function runQuery( } } + // Flush any remaining buffered text that was never followed by a result event. + // This happens when the agent produces a short response without a formal end_turn. + if (pendingProgressText && !terminalResultObserved) { + log(`Flushing remaining pending progress text as final output (${pendingProgressText.length} chars)`); + writeOutput({ + status: 'success', + ...normalizeStructuredOutput(pendingProgressText), + newSessionId, + }); + terminalResultObserved = true; + resultCount++; + } + ipcPolling = false; log( `Query done. Messages: ${messageCount}, results: ${resultCount}, closedDuringQuery: ${closedDuringQuery}`, diff --git a/scripts/migrate-unify.cjs b/scripts/migrate-unify.cjs deleted file mode 100644 index 2a6af0f..0000000 --- a/scripts/migrate-unify.cjs +++ /dev/null @@ -1,220 +0,0 @@ -/** - * Data Unification Migration Script - * Merges codex data into the primary (claude) directories. - * - * Run AFTER stopping both services. - */ -const Database = require('/home/clone-ej/EJClaw/node_modules/better-sqlite3'); -const fs = require('fs'); -const path = require('path'); - -const BASE = '/home/clone-ej/EJClaw'; -const CLAUDE_DB = path.join(BASE, 'store/messages.db'); -const CODEX_DB = path.join(BASE, 'store-codex/messages.db'); - -console.log('=== EJClaw Data Unification ===\n'); - -// 1. Open both DBs -const primary = new Database(CLAUDE_DB); -const codex = new Database(CODEX_DB); - -primary.pragma('journal_mode = WAL'); -primary.pragma('busy_timeout = 5000'); - -// 2. Merge registered_groups (codex → primary, skip JID conflicts) -console.log('--- Merging registered_groups ---'); -const codexGroups = codex.prepare('SELECT * FROM registered_groups').all(); -const insertGroup = primary.prepare(` - INSERT OR IGNORE INTO registered_groups - (jid, name, folder, trigger_pattern, added_at, container_config, requires_trigger, is_main, agent_type, work_dir) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) -`); - -let groupsAdded = 0; -let groupsSkipped = 0; -for (const g of codexGroups) { - // Check if JID already exists with different agent_type - const existing = primary.prepare('SELECT agent_type FROM registered_groups WHERE jid = ?').get(g.jid); - if (existing) { - // Same JID, different bots — both need to be registered - // Since JID is PK, we can't have duplicates. The group with claude-code type stays, - // and the codex one is already differentiated by agent_type in the same row. - // But wait — in the unified DB, the same JID can only appear once. - // For shared channels (both bots respond), we keep the claude registration - // and the codex service will also load it IF we adjust the filter. - // Actually, the codex service filters by agent_type='codex', so it won't see claude-code groups. - // We need BOTH registrations for shared channels. - // Solution: codex groups with different folder names get inserted. - // Same JID + same folder = skip (duplicate) - // Same JID + different folder = need a unique folder for codex - if (existing.agent_type !== g.agent_type) { - // Same JID but different agent types — need both - // Change JID to make it unique: append agent type suffix - const codexJid = g.jid + ':codex'; - const existsCodex = primary.prepare('SELECT 1 FROM registered_groups WHERE jid = ?').get(codexJid); - if (!existsCodex) { - insertGroup.run(codexJid, g.name, g.folder, g.trigger_pattern, g.added_at, - g.container_config, g.requires_trigger, g.is_main, g.agent_type, g.work_dir); - groupsAdded++; - console.log(` + ${g.folder} (codex, jid=${codexJid})`); - } else { - groupsSkipped++; - } - } else { - groupsSkipped++; - } - } else { - insertGroup.run(g.jid, g.name, g.folder, g.trigger_pattern, g.added_at, - g.container_config, g.requires_trigger, g.is_main, g.agent_type, g.work_dir); - groupsAdded++; - console.log(` + ${g.folder} (${g.agent_type})`); - } -} -console.log(` Added: ${groupsAdded}, Skipped: ${groupsSkipped}\n`); - -// 3. Merge sessions (codex → primary, agent_type='codex') -console.log('--- Merging sessions ---'); -const codexSessions = codex.prepare('SELECT * FROM sessions').all(); -const insertSession = primary.prepare( - 'INSERT OR IGNORE INTO sessions (group_folder, agent_type, session_id) VALUES (?, ?, ?)' -); -let sessionsAdded = 0; -for (const s of codexSessions) { - insertSession.run(s.group_folder, s.agent_type, s.session_id); - sessionsAdded++; - console.log(` + ${s.group_folder} [${s.agent_type}]`); -} -console.log(` Added: ${sessionsAdded}\n`); - -// 4. Merge router_state (codex → primary, already prefixed) -console.log('--- Merging router_state ---'); -const codexState = codex.prepare('SELECT * FROM router_state').all(); -const insertState = primary.prepare( - 'INSERT OR IGNORE INTO router_state (key, value) VALUES (?, ?)' -); -for (const s of codexState) { - insertState.run(s.key, s.value); - console.log(` + ${s.key}`); -} -console.log(''); - -// 5. Merge messages (codex → primary, skip duplicates by PK) -console.log('--- Merging messages ---'); -const codexMsgs = codex.prepare('SELECT * FROM messages').all(); -const insertMsg = primary.prepare(` - INSERT OR IGNORE INTO messages - (id, chat_jid, sender, sender_name, content, timestamp, is_from_me, is_bot_message) - VALUES (?, ?, ?, ?, ?, ?, ?, ?) -`); -let msgsAdded = 0; -let msgsSkipped = 0; -const batchInsert = primary.transaction((msgs) => { - for (const m of msgs) { - const result = insertMsg.run(m.id, m.chat_jid, m.sender, m.sender_name, - m.content, m.timestamp, m.is_from_me, m.is_bot_message); - if (result.changes > 0) msgsAdded++; - else msgsSkipped++; - } -}); -batchInsert(codexMsgs); -console.log(` Added: ${msgsAdded}, Skipped (duplicates): ${msgsSkipped}\n`); - -// 6. Merge chats (codex → primary, UPSERT with newer timestamp) -console.log('--- Merging chats ---'); -const codexChats = codex.prepare('SELECT * FROM chats').all(); -const upsertChat = primary.prepare(` - INSERT INTO chats (jid, name, last_message_time, channel, is_group) VALUES (?, ?, ?, ?, ?) - ON CONFLICT(jid) DO UPDATE SET - name = COALESCE(excluded.name, name), - last_message_time = MAX(last_message_time, excluded.last_message_time), - channel = COALESCE(excluded.channel, channel), - is_group = COALESCE(excluded.is_group, is_group) -`); -let chatsUpserted = 0; -for (const c of codexChats) { - upsertChat.run(c.jid, c.name, c.last_message_time, c.channel, c.is_group); - chatsUpserted++; -} -console.log(` Upserted: ${chatsUpserted}\n`); - -// 7. Merge scheduled_tasks (codex → primary, skip duplicates) -const codexTasks = codex.prepare('SELECT * FROM scheduled_tasks').all(); -if (codexTasks.length > 0) { - console.log('--- Merging scheduled_tasks ---'); - const insertTask = primary.prepare(` - INSERT OR IGNORE INTO scheduled_tasks - (id, group_folder, chat_jid, prompt, schedule_type, schedule_value, next_run, last_run, last_result, status, created_at, context_mode) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) - `); - for (const t of codexTasks) { - insertTask.run(t.id, t.group_folder, t.chat_jid, t.prompt, t.schedule_type, - t.schedule_value, t.next_run, t.last_run, t.last_result, t.status, t.created_at, t.context_mode); - console.log(` + ${t.id}`); - } - console.log(''); -} - -// Close DBs -primary.close(); -codex.close(); - -// 8. Merge groups directories -console.log('--- Merging groups-codex/ → groups/ ---'); -const codexGroupsDir = path.join(BASE, 'groups-codex'); -const claudeGroupsDir = path.join(BASE, 'groups'); -if (fs.existsSync(codexGroupsDir)) { - const entries = fs.readdirSync(codexGroupsDir); - for (const entry of entries) { - const src = path.join(codexGroupsDir, entry); - const dst = path.join(claudeGroupsDir, entry); - if (fs.existsSync(dst)) { - // Merge: copy files that don't exist in dst - if (fs.statSync(src).isDirectory()) { - const files = fs.readdirSync(src); - for (const f of files) { - const srcFile = path.join(src, f); - const dstFile = path.join(dst, f); - if (!fs.existsSync(dstFile)) { - fs.cpSync(srcFile, dstFile, { recursive: true }); - console.log(` cp ${entry}/${f}`); - } - } - } - } else { - fs.cpSync(src, dst, { recursive: true }); - console.log(` + ${entry}/`); - } - } -} -console.log(''); - -// 9. Merge data-codex/sessions/ → data/sessions/ -console.log('--- Merging data-codex/sessions/ → data/sessions/ ---'); -const codexDataSessions = path.join(BASE, 'data-codex/sessions'); -const claudeDataSessions = path.join(BASE, 'data/sessions'); -if (fs.existsSync(codexDataSessions)) { - const entries = fs.readdirSync(codexDataSessions); - for (const entry of entries) { - const src = path.join(codexDataSessions, entry); - const dst = path.join(claudeDataSessions, entry); - if (fs.existsSync(dst)) { - // Both exist — merge .codex/ subdirectory - const codexSubdir = path.join(src, '.codex'); - const dstCodexSubdir = path.join(dst, '.codex'); - if (fs.existsSync(codexSubdir) && !fs.existsSync(dstCodexSubdir)) { - fs.cpSync(codexSubdir, dstCodexSubdir, { recursive: true }); - console.log(` cp ${entry}/.codex/`); - } - } else { - fs.cpSync(src, dst, { recursive: true }); - console.log(` + ${entry}/`); - } - } -} -console.log(''); - -console.log('=== Migration complete ==='); -console.log('Next steps:'); -console.log('1. Update .env.codex to remove EJCLAW_STORE_DIR, EJCLAW_DATA_DIR, EJCLAW_GROUPS_DIR'); -console.log('2. Restart both services'); -console.log('3. Verify, then rename old dirs to .bak'); diff --git a/src/agent-output.ts b/src/agent-output.ts index 3a40820..1d7e8d9 100644 --- a/src/agent-output.ts +++ b/src/agent-output.ts @@ -1,4 +1,3 @@ -import { parseStructuredOutputEnvelope } from './output-suppression.js'; import type { StructuredAgentOutput } from './types.js'; export function stringifyLegacyAgentResult( @@ -21,10 +20,7 @@ export function getStructuredAgentOutput(output: { if (output.output) { return output.output; } - if (typeof output.result !== 'string') { - return null; - } - return parseStructuredOutputEnvelope(output.result.trim()); + return null; } export function getAgentOutputText(output: { @@ -32,9 +28,6 @@ export function getAgentOutputText(output: { result?: string | object | null; }): string | null { const structured = getStructuredAgentOutput(output); - if (structured?.visibility === 'silent') { - return null; - } if (structured?.visibility === 'public') { return structured.text; } @@ -51,9 +44,9 @@ export function hasAgentOutputPayload(output: { return output.result !== null && output.result !== undefined; } -export function isSilentAgentOutput(output: { +export function isSilentAgentOutput(_output: { output?: StructuredAgentOutput; result?: string | object | null; }): boolean { - return getStructuredAgentOutput(output)?.visibility === 'silent'; + return false; } diff --git a/src/agent-runner.ts b/src/agent-runner.ts index 5f4ca88..ad7d158 100644 --- a/src/agent-runner.ts +++ b/src/agent-runner.ts @@ -13,6 +13,7 @@ import { IDLE_TIMEOUT, } from './config.js'; import { prepareGroupEnvironment } from './agent-runner-environment.js'; +import { getEnv } from './env.js'; export { type AvailableGroup, writeGroupsSnapshot, @@ -74,7 +75,7 @@ export async function runAgentProcess( // as a host process. This provides kernel-level write protection. const isReviewerContainer = envOverrides?.EJCLAW_REVIEWER_RUNTIME === '1' && - process.env.REVIEWER_CONTAINER_ENABLED !== '0'; + getEnv('REVIEWER_CONTAINER_ENABLED') !== '0'; if (isReviewerContainer) { const ownerWorkspaceDir = envOverrides?.EJCLAW_WORK_DIR || group.workDir || process.cwd(); diff --git a/src/channels/discord.ts b/src/channels/discord.ts index c7761d3..bf1b63d 100644 --- a/src/channels/discord.ts +++ b/src/channels/discord.ts @@ -184,11 +184,14 @@ export class DiscordChannel implements Channel { botToken: string, opts: DiscordChannelOpts, agentTypeFilter?: AgentType, + channelName?: string, ) { this.botToken = botToken; this.opts = opts; this.agentTypeFilter = agentTypeFilter; - if (agentTypeFilter) { + if (channelName) { + this.name = channelName; + } else if (agentTypeFilter) { this.name = `discord-${agentTypeFilter}`; } } @@ -723,21 +726,20 @@ registerChannel('discord', (opts: ChannelOpts) => { token, opts, hasCodexBot ? 'claude-code' : undefined, + 'discord', ); }); // Register the secondary Codex bot channel. -// In unified mode all bots register unconditionally; in legacy per-service mode -// the codex service uses its own DISCORD_BOT_TOKEN via systemd EnvironmentFile override. registerChannel('discord-codex', (opts: ChannelOpts) => { const token = getEnv('DISCORD_CODEX_BOT_TOKEN') || ''; - if (!token) return null; // Codex Discord bot is optional - return new DiscordChannel(token, opts, 'codex'); + if (!token) return null; + return new DiscordChannel(token, opts, 'codex', 'discord-codex'); }); // Register the review bot channel (codex agent type, separate token). registerChannel('discord-review', (opts: ChannelOpts) => { const token = getEnv('DISCORD_REVIEW_BOT_TOKEN') || ''; - if (!token) return null; // Review Discord bot is optional - return new DiscordChannel(token, opts, 'codex'); + if (!token) return null; + return new DiscordChannel(token, opts, 'codex', 'discord-review'); }); diff --git a/src/config.ts b/src/config.ts index 29ab193..cbeab98 100644 --- a/src/config.ts +++ b/src/config.ts @@ -22,9 +22,6 @@ export const SERVICE_AGENT_TYPE: AgentType = ? 'codex' : 'claude-code'; -/** Unified mode: single process manages all bots. Disable with UNIFIED_MODE=0. */ -export const UNIFIED_MODE = getEnv('UNIFIED_MODE') !== '0'; - export function normalizeServiceId(serviceId: string): string { if (serviceId === 'codex') { return CODEX_MAIN_SERVICE_ID; @@ -92,6 +89,25 @@ export const RECOVERY_CONCURRENT_AGENTS = parseInt( ); // ── Paired review ───────────────────────────────────────────────── + +/** Owner agent type. Default: codex. Set OWNER_AGENT_TYPE=claude-code to use Claude as owner. */ +const rawOwnerAgentType = getEnv('OWNER_AGENT_TYPE'); +export const OWNER_AGENT_TYPE: AgentType = + rawOwnerAgentType === 'codex' || rawOwnerAgentType === 'claude-code' + ? rawOwnerAgentType + : 'codex'; + +/** Reviewer agent type. Default: claude-code. Set REVIEWER_AGENT_TYPE=codex to use Codex as reviewer. */ +const rawReviewerAgentType = getEnv('REVIEWER_AGENT_TYPE'); +export const REVIEWER_AGENT_TYPE: AgentType = + rawReviewerAgentType === 'codex' || rawReviewerAgentType === 'claude-code' + ? rawReviewerAgentType + : 'claude-code'; + +/** Service ID for the reviewer based on agent type. */ +export const REVIEWER_SERVICE_ID_FOR_TYPE = + REVIEWER_AGENT_TYPE === 'claude-code' ? CLAUDE_SERVICE_ID : CODEX_REVIEW_SERVICE_ID; + // Max owner↔reviewer round trips per task. 0 = unlimited. const rawMaxRoundTrips = getEnv('PAIRED_MAX_ROUND_TRIPS') || '10'; export const PAIRED_MAX_ROUND_TRIPS = diff --git a/src/db.ts b/src/db.ts index 9cb7f72..e18a92b 100644 --- a/src/db.ts +++ b/src/db.ts @@ -1053,6 +1053,18 @@ export function getLastHumanMessageTimestamp(chatJid: string): string | null { return row?.timestamp ?? null; } +export function getLastHumanMessageContent(chatJid: string): string | null { + const row = db + .prepare( + `SELECT content FROM messages + WHERE chat_jid = ? AND is_bot_message = 0 AND is_from_me = 0 + AND content != '' AND content IS NOT NULL + ORDER BY timestamp DESC, seq DESC LIMIT 1`, + ) + .get(chatJid) as { content: string } | undefined; + return row?.content ?? null; +} + export function hasRecentRestartAnnouncement( chatJid: string, sinceTimestamp: string, diff --git a/src/index.ts b/src/index.ts index bf2313c..40270ca 100644 --- a/src/index.ts +++ b/src/index.ts @@ -9,9 +9,6 @@ import { POLL_INTERVAL, SERVICE_ID, SERVICE_AGENT_TYPE, - UNIFIED_MODE, - isClaudeService, - isReviewService, isSessionCommandSenderAllowed, STATUS_CHANNEL_ID, STATUS_UPDATE_INTERVAL, @@ -79,7 +76,6 @@ import { initTokenRotation, } from './token-rotation.js'; import { - shouldStartTokenRefreshLoop, startTokenRefreshLoop, stopTokenRefreshLoop, } from './token-refresh.js'; @@ -189,16 +185,11 @@ function loadState(): void { lastAgentTimestamp = {}; } sessions = getAllSessions(); - // In unified mode, load ALL groups (duplicates resolved by preferring codex). - // In legacy per-service mode, load only this service's registrations. - registeredGroups = UNIFIED_MODE - ? getAllRegisteredGroups() - : getAllRegisteredGroups(SERVICE_AGENT_TYPE); + registeredGroups = getAllRegisteredGroups(); logger.info( { groupCount: Object.keys(registeredGroups).length, agentType: SERVICE_AGENT_TYPE, - unifiedMode: UNIFIED_MODE, }, 'State loaded', ); @@ -331,27 +322,16 @@ async function main(): Promise { initTokenRotation(); initCodexTokenRotation(); - // Unified mode: start credential proxy for container isolation and clean up orphaned containers - if (UNIFIED_MODE) { - startCredentialProxy(CREDENTIAL_PROXY_PORT).catch((err) => - logger.warn( - { err }, - 'Failed to start credential proxy (may already be running)', - ), - ); - cleanupOrphans(); - } + // Start credential proxy for container isolation and clean up orphaned containers + startCredentialProxy(CREDENTIAL_PROXY_PORT).catch((err) => + logger.warn( + { err }, + 'Failed to start credential proxy (may already be running)', + ), + ); + cleanupOrphans(); - // In unified mode, always start the Claude OAuth refresh loop. - // In per-service mode, only the Claude service owns refresh to avoid races. - if (UNIFIED_MODE || shouldStartTokenRefreshLoop(SERVICE_AGENT_TYPE)) { - startTokenRefreshLoop(); - } else { - logger.info( - { serviceAgentType: SERVICE_AGENT_TYPE }, - 'Skipping Claude OAuth token auto-refresh for non-Claude service', - ); - } + startTokenRefreshLoop(); loadState(); @@ -456,27 +436,19 @@ async function main(): Promise { } // Start subsystems (independently of connection handler) - // In unified mode always run the scheduler. In per-service mode, skip for review-only service. - if (UNIFIED_MODE || !isReviewService()) { - startSchedulerLoop({ - registeredGroups: () => registeredGroups, - getSessions: () => sessions, - queue, - onProcess: (groupJid, proc, processName, ipcDir) => - queue.registerProcess(groupJid, proc, processName, ipcDir), - sendMessage: (jid, rawText) => - sendFormattedChannelMessage(channels, jid, rawText), - sendTrackedMessage: (jid, rawText) => - sendFormattedTrackedChannelMessage(channels, jid, rawText), - editTrackedMessage: (jid, messageId, rawText) => - editFormattedTrackedChannelMessage(channels, jid, messageId, rawText), - }); - } else { - logger.info( - { serviceId: SERVICE_ID }, - 'Skipping scheduler for review service', - ); - } + startSchedulerLoop({ + registeredGroups: () => registeredGroups, + getSessions: () => sessions, + queue, + onProcess: (groupJid, proc, processName, ipcDir) => + queue.registerProcess(groupJid, proc, processName, ipcDir), + sendMessage: (jid, rawText) => + sendFormattedChannelMessage(channels, jid, rawText), + sendTrackedMessage: (jid, rawText) => + sendFormattedTrackedChannelMessage(channels, jid, rawText), + editTrackedMessage: (jid, messageId, rawText) => + editFormattedTrackedChannelMessage(channels, jid, messageId, rawText), + }); startIpcWatcher({ sendMessage: (jid, text) => { const channel = findChannel(channels, jid); @@ -538,8 +510,7 @@ async function main(): Promise { purgeOnStart: true, }); - if (UNIFIED_MODE || isClaudeService()) { - leaseRecoveryTimer = setInterval(() => { + leaseRecoveryTimer = setInterval(() => { if (!hasAvailableClaudeToken()) { return; } @@ -578,7 +549,6 @@ async function main(): Promise { ); } }, 5_000); - } runtime.startMessageLoop().catch((err) => { logger.fatal({ err }, 'Message loop crashed unexpectedly'); process.exit(1); diff --git a/src/message-agent-executor.test.ts b/src/message-agent-executor.test.ts index 0bd3753..2018595 100644 --- a/src/message-agent-executor.test.ts +++ b/src/message-agent-executor.test.ts @@ -142,6 +142,7 @@ function makeDeps() { assistantName: 'Andy', queue: { registerProcess: vi.fn(), + enqueueMessageCheck: vi.fn(), }, getRegisteredGroups: () => ({}), getSessions: () => ({}), diff --git a/src/message-agent-executor.ts b/src/message-agent-executor.ts index 1f1266c..11283d2 100644 --- a/src/message-agent-executor.ts +++ b/src/message-agent-executor.ts @@ -37,6 +37,8 @@ import { CODEX_MAIN_SERVICE_ID, CODEX_REVIEW_SERVICE_ID, SERVICE_SESSION_SCOPE, + REVIEWER_AGENT_TYPE, + isClaudeService, } from './config.js'; import { buildStructuredOutputPrompt, @@ -95,10 +97,41 @@ export async function runAgentForGroup( onOutput, } = args; const isMain = group.isMain === true; - const isClaudeCodeAgent = - (group.agentType || 'claude-code') === 'claude-code'; const sessions = deps.getSessions(); - const sessionId = sessions[group.folder]; + + const currentLease = getEffectiveChannelLease(chatJid); + + // In unified mode, determine role from the lease directly. + // Default to owner; the auto-review trigger in completePairedExecutionContext + // will switch to reviewer when the task is in review_ready state. + const pairedTask = currentLease.reviewer_service_id + ? getLatestOpenPairedTaskForChat(chatJid) + : null; + const effectiveServiceId = + pairedTask && + (pairedTask.status === 'review_ready' || pairedTask.status === 'in_review') + ? currentLease.reviewer_service_id! + : currentLease.owner_service_id; + const reviewerMode = effectiveServiceId === currentLease.reviewer_service_id; + + // Override agent type based on role config (OWNER_AGENT_TYPE / REVIEWER_AGENT_TYPE). + // When the reviewer uses a different agent type than the group default, + // swap in the configured type for the duration of this execution. + const reviewerAgentTypeOverride = + reviewerMode && REVIEWER_AGENT_TYPE !== (group.agentType || 'claude-code'); + const effectiveAgentType = reviewerAgentTypeOverride + ? REVIEWER_AGENT_TYPE + : group.agentType || 'claude-code'; + const effectiveGroup = reviewerAgentTypeOverride + ? { ...group, agentType: REVIEWER_AGENT_TYPE } + : group; + const isClaudeCodeAgent = effectiveAgentType === 'claude-code'; + + // When the reviewer uses a different agent type than the group default, + // the stored session belongs to the other agent. Don't reuse it. + const sessionId = reviewerAgentTypeOverride + ? undefined + : sessions[group.folder]; const memoryBriefing = sessionId ? undefined : await buildRoomMemoryBriefing({ @@ -130,20 +163,6 @@ export async function runAgentForGroup( let resetSessionRequested = false; const canRotateToken = isClaudeCodeAgent && getTokenCount() > 1; - const currentLease = getEffectiveChannelLease(chatJid); - - // In unified mode, determine role from the lease directly. - // Default to owner; the auto-review trigger in completePairedExecutionContext - // will switch to reviewer when the task is in review_ready state. - const pairedTask = currentLease.reviewer_service_id - ? getLatestOpenPairedTaskForChat(chatJid) - : null; - const effectiveServiceId = - pairedTask && - (pairedTask.status === 'review_ready' || pairedTask.status === 'in_review') - ? currentLease.reviewer_service_id! - : currentLease.owner_service_id; - const reviewerMode = effectiveServiceId === currentLease.reviewer_service_id; const roomRoleContext = buildRoomRoleContext( currentLease, effectiveServiceId, @@ -156,11 +175,13 @@ export async function runAgentForGroup( }); const effectivePrompt = buildStructuredOutputPrompt(prompt, { reviewerMode, + pairedRoom: !!pairedExecutionContext, gateTurnKind: pairedExecutionContext?.gateTurnKind, requiresVisibleVerdict: pairedExecutionContext?.requiresVisibleVerdict, }); let pairedExecutionStatus: 'succeeded' | 'failed' = 'failed'; let pairedExecutionSummary: string | null = null; + let pairedExecutionCompleted = false; const shouldHandoffToCodex = ( reason: AgentTriggerReason, @@ -248,6 +269,7 @@ export async function runAgentForGroup( status: pairedExecutionStatus, summary: pairedExecutionSummary, }); + pairedExecutionCompleted = true; return 'success'; } @@ -394,21 +416,22 @@ export async function runAgentForGroup( } : undefined; - const agentType = group.agentType || 'claude-code'; logger.info( { chatJid, group: group.name, groupFolder: group.folder, runId, - provider: agentType, + provider: effectiveAgentType, + reviewerMode, + reviewerAgentTypeOverride, }, - `Using provider: ${agentType}`, + `Using provider: ${effectiveAgentType}`, ); try { const output = await runAgentProcess( - group, + effectiveGroup, { ...agentInput, sessionId: provider === 'claude' ? sessionId : undefined, @@ -887,7 +910,7 @@ export async function runAgentForGroup( pairedExecutionStatus = 'succeeded'; return 'success'; } finally { - if (pairedExecutionContext) { + if (pairedExecutionContext && !pairedExecutionCompleted) { const completedRole = roomRoleContext?.role ?? 'owner'; completePairedExecutionContext({ taskId: pairedExecutionContext.task.id, @@ -895,12 +918,12 @@ export async function runAgentForGroup( status: pairedExecutionStatus, summary: pairedExecutionSummary, }); + } - // After owner/reviewer completes, enqueue the next turn so - // the message loop picks it up without waiting for a new message. - if (pairedExecutionStatus === 'succeeded') { - deps.queue.enqueueMessageCheck(chatJid); - } + // After owner/reviewer completes, enqueue the next turn so + // the message loop picks it up without waiting for a new message. + if (pairedExecutionContext && pairedExecutionStatus === 'succeeded') { + deps.queue.enqueueMessageCheck(chatJid); } } } diff --git a/src/message-runtime-rules.ts b/src/message-runtime-rules.ts index a7d31f2..db2d8fa 100644 --- a/src/message-runtime-rules.ts +++ b/src/message-runtime-rules.ts @@ -89,6 +89,12 @@ export function hasAllowedTrigger(opts: { return true; } + // Paired rooms: bot-to-bot ping-pong doesn't need trigger patterns. + // Peer bot messages (from reviewer/owner) are always allowed. + if (isPairedRoomJid(chatJid)) { + return true; + } + const allowlistCfg = loadSenderAllowlist(); const hasTrigger = messages.some( (message) => @@ -104,14 +110,10 @@ export function getProcessableMessages( messages: Parameters[0], channel?: Channel, ) { - const isPaired = isPairedRoomJid(chatJid); return filterProcessableMessages( messages, - isPaired, - // In paired rooms (unified mode), don't filter by isOwnMessage — - // both bots need to see each other's messages for the ping-pong flow. - // Loop prevention is handled by filterLoopingPairedBotMessages + round_trip_count. - isPaired ? undefined : channel?.isOwnMessage?.bind(channel), + isPairedRoomJid(chatJid), + channel?.isOwnMessage?.bind(channel), ).filter((message) => !isTaskStatusControlMessage(message.content)); } diff --git a/src/message-runtime.ts b/src/message-runtime.ts index c46206c..36095ec 100644 --- a/src/message-runtime.ts +++ b/src/message-runtime.ts @@ -12,6 +12,7 @@ import { markWorkItemDelivered, markWorkItemDeliveryRetry, getLastBotFinalMessage, + getLastHumanMessageContent, isPairedRoomJid, getLatestOpenPairedTaskForChat, type ServiceHandoff, @@ -21,12 +22,14 @@ import { isClaudeService, isReviewService, isSessionCommandSenderAllowed, + REVIEWER_AGENT_TYPE, SERVICE_AGENT_TYPE, SERVICE_ID, } from './config.js'; import { GroupQueue, GroupRunContext } from './group-queue.js'; import { findChannel, + findChannelByName, formatMessages, normalizeMessageForDedupe, } from './router.js'; @@ -56,10 +59,7 @@ import { import { Channel, NewMessage, RegisteredGroup } from './types.js'; import { logger } from './logger.js'; import { resolveGroupIpcPath } from './group-folder.js'; -import { - getEffectiveChannelLease, - shouldServiceProcessChat, -} from './service-routing.js'; +import { getEffectiveChannelLease } from './service-routing.js'; /** * Check if a message is a duplicate of the last bot final message in a paired room. @@ -451,37 +451,50 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { return true; } + // For paired rooms, determine the reviewer channel for correct bot routing. + // This is used whenever the reviewer needs to send output. + const reviewerChannelName = + isPairedRoomJid(chatJid) && REVIEWER_AGENT_TYPE === 'claude-code' + ? 'discord' + : 'discord-review'; + const foundReviewerChannel = findChannelByName( + deps.channels, + reviewerChannelName, + ); + const reviewerChannel = foundReviewerChannel || channel; + if (isPairedRoomJid(chatJid)) { + logger.info( + { + chatJid, + reviewerChannelName, + foundChannel: foundReviewerChannel?.name ?? null, + usingChannel: reviewerChannel.name, + availableChannels: deps.channels.map((c) => c.name), + }, + 'Paired room reviewer channel resolution', + ); + } + + // Deliver pending work items through the correct channel. + // For paired rooms, check if the work item was from a reviewer turn. const openWorkItem = getOpenWorkItem( chatJid, (group.agentType || 'claude-code') as 'claude-code' | 'codex', ); if (openWorkItem) { - const delivered = await deliverOpenWorkItem(channel, openWorkItem); + // Use reviewer channel if the pending task is in review state + const pendingTask = isPairedRoomJid(chatJid) + ? getLatestOpenPairedTaskForChat(chatJid) + : null; + const isReviewerWorkItem = + pendingTask && + (pendingTask.status === 'review_ready' || + pendingTask.status === 'in_review'); + const deliveryChannel = isReviewerWorkItem ? reviewerChannel : channel; + const delivered = await deliverOpenWorkItem(deliveryChannel, openWorkItem); if (!delivered) return false; } - if (!shouldServiceProcessChat(chatJid, SERVICE_ID)) { - const rawMissedMessages = getMessagesSinceSeq( - chatJid, - deps.getLastAgentTimestamps()[chatJid] || '0', - deps.assistantName, - ); - const lastIgnored = rawMissedMessages[rawMissedMessages.length - 1]; - if (lastIgnored?.seq != null) { - advanceLastAgentCursor( - deps.getLastAgentTimestamps(), - deps.saveState, - chatJid, - lastIgnored.seq, - ); - } - logger.debug( - { chatJid, serviceId: SERVICE_ID }, - 'Skipping message processing for unassigned service', - ); - return true; - } - const isMainGroup = group.isMain === true; while (true) { const sinceSeqCursor = deps.getLastAgentTimestamps()[chatJid] || '0'; @@ -506,15 +519,40 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { (pendingReviewTask.status === 'review_ready' || pendingReviewTask.status === 'in_review') ) { - // Synthesize a prompt for the reviewer turn - const reviewPrompt = - 'Review the latest owner changes. Provide feedback or approve.'; + // Build review prompt with user's original request + owner's response + const userMessage = getLastHumanMessageContent(chatJid); + const ownerContent = rawMissedMessages + .filter((m) => m.is_bot_message) + .map((m) => m.content) + .join('\n\n'); + const parts: string[] = []; + if (userMessage) { + parts.push(`User request:\n---\n${userMessage}\n---`); + } + if (ownerContent) { + parts.push(`Owner response:\n---\n${ownerContent}\n---`); + } + const reviewPrompt = parts.length > 0 + ? `${parts.join('\n\n')}\n\nReview the owner's response above. Provide feedback or approve.` + : 'Review the latest owner changes in the workspace. Provide feedback or approve.'; + + // Advance cursor past the owner's messages so they aren't re-processed + const lastRaw = rawMissedMessages[rawMissedMessages.length - 1]; + if (lastRaw?.seq != null) { + advanceLastAgentCursor( + deps.getLastAgentTimestamps(), + deps.saveState, + chatJid, + lastRaw.seq, + ); + } + const { deliverySucceeded } = await executeTurn({ group, prompt: reviewPrompt, chatJid, runId, - channel, + channel: reviewerChannel, startSeq: null, endSeq: null, }); @@ -751,20 +789,6 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { continue; } - if (!shouldServiceProcessChat(chatJid, SERVICE_ID)) { - const lastIgnored = - processableGroupMessages[processableGroupMessages.length - 1]; - if (lastIgnored?.seq != null) { - advanceLastAgentCursor( - deps.getLastAgentTimestamps(), - deps.saveState, - chatJid, - lastIgnored.seq, - ); - } - continue; - } - if ( shouldSkipBotOnlyCollaboration(chatJid, processableGroupMessages) ) { @@ -906,10 +930,6 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { continue; } - if (!shouldServiceProcessChat(chatJid, SERVICE_ID)) { - continue; - } - const sinceSeqCursor = deps.getLastAgentTimestamps()[chatJid] || ''; const rawPending = getMessagesSinceSeq( chatJid, diff --git a/src/message-turn-controller.ts b/src/message-turn-controller.ts index fc02a58..7cc95a3 100644 --- a/src/message-turn-controller.ts +++ b/src/message-turn-controller.ts @@ -122,14 +122,9 @@ export class MessageTurnController { } const raw = getAgentOutputText(result); - const silentOutput = isSilentAgentOutput(result); - const suppressState = - raw && this.options.suppressToken - ? classifySuppressTokenOutput(raw, this.options.suppressToken) - : raw - ? classifySuppressTokenOutput(raw, undefined) - : 'none'; - const text = raw && suppressState === 'none' ? formatOutbound(raw) : null; + const silentOutput = false; + const suppressState = 'none' as const; + const text = raw ? formatOutbound(raw) : null; if (raw) { logger.info( @@ -144,31 +139,6 @@ export class MessageTurnController { }, `Agent output: ${raw.slice(0, 200)}`, ); - if (suppressState === 'exact') { - logger.info( - { - chatJid: this.options.chatJid, - group: this.options.group.name, - groupFolder: this.options.group.folder, - runId: this.options.runId, - resultStatus: result.status, - resultPhase: result.phase, - }, - 'Suppressed exact-match silent output token', - ); - } else if (suppressState === 'mixed') { - logger.warn( - { - chatJid: this.options.chatJid, - group: this.options.group.name, - groupFolder: this.options.group.folder, - runId: this.options.runId, - resultStatus: result.status, - resultPhase: result.phase, - }, - 'Blocked malformed output that mixed the silent output token with visible text', - ); - } } const phase: AgentOutputPhase = normalizeAgentOutputPhase(result.phase); diff --git a/src/output-suppression.ts b/src/output-suppression.ts index 9d3b874..6441399 100644 --- a/src/output-suppression.ts +++ b/src/output-suppression.ts @@ -1,125 +1,35 @@ -import { randomBytes } from 'crypto'; +/** + * Output suppression was removed — harness-level protections (isOwnMessage filter, + * shouldSkipBotOnlyCollaboration, PAIRED_MAX_ROUND_TRIPS) now prevent infinite loops. + * + * Retained exports are stubs so callers compile without changes. + */ -import { - CLAUDE_SERVICE_ID, - CODEX_REVIEW_SERVICE_ID, - normalizeServiceId, -} from './config.js'; -import type { StructuredAgentOutput } from './types.js'; - -const ANY_SUPPRESS_TOKEN_PATTERN = /__EJ_SUPPRESS_[a-f0-9]{24,}(?:__)?/g; -const EXACT_ANY_SUPPRESS_TOKEN_PATTERN = /^__EJ_SUPPRESS_[a-f0-9]{24,}(?:__)?$/; -const STRUCTURED_SILENT_OUTPUT_PREFIX_PATTERN = - /^\s*\{\s*"ejclaw"\s*:\s*\{\s*"visibility"\s*:\s*"silent"/; -export const STRUCTURED_SILENT_OUTPUT_ENVELOPE = - '{"ejclaw":{"visibility":"silent"}}'; -const STRUCTURED_PUBLIC_VERDICTS = new Set([ - 'done', - 'done_with_concerns', - 'blocked', -]); -const STRUCTURED_SILENT_VERDICTS = new Set(['silent']); - -export function createSuppressToken(): string { - return `__EJ_SUPPRESS_${randomBytes(12).toString('hex')}__`; -} - -export function shouldEnableSuppressOutputForService( - serviceId: string | undefined, -): boolean { - if (!serviceId) return false; - const normalized = normalizeServiceId(serviceId); - return ( - normalized === CLAUDE_SERVICE_ID || normalized === CODEX_REVIEW_SERVICE_ID - ); +export function createSuppressToken(): string | undefined { + return undefined; } export function classifySuppressTokenOutput( - rawText: string, - suppressToken: string | undefined, -): 'exact' | 'mixed' | 'none' { - const trimmed = rawText.trim(); - const structured = parseStructuredOutputEnvelope(trimmed); - if ( - (suppressToken && trimmed === suppressToken) || - EXACT_ANY_SUPPRESS_TOKEN_PATTERN.test(trimmed) || - structured?.visibility === 'silent' - ) { - return 'exact'; - } - if (suppressToken && rawText.includes(suppressToken)) { - return 'mixed'; - } - ANY_SUPPRESS_TOKEN_PATTERN.lastIndex = 0; - if (ANY_SUPPRESS_TOKEN_PATTERN.test(rawText)) { - return 'mixed'; - } - return STRUCTURED_SILENT_OUTPUT_PREFIX_PATTERN.test(trimmed) - ? 'mixed' - : 'none'; -} - -export function parseStructuredOutputEnvelope( - rawText: string, -): StructuredAgentOutput | null { - try { - const parsed = JSON.parse(rawText) as { - ejclaw?: { visibility?: unknown; text?: unknown }; - }; - const envelope = parsed?.ejclaw; - if (!envelope || typeof envelope !== 'object' || Array.isArray(envelope)) { - return null; - } - if (envelope.visibility === 'silent') { - return { visibility: 'silent' }; - } - if ( - envelope.visibility === 'public' && - typeof envelope.text === 'string' && - envelope.text.length > 0 - ) { - return { visibility: 'public', text: envelope.text }; - } - } catch { - return null; - } - - return null; + _rawText: string, + _suppressToken: string | undefined, +): 'none' { + return 'none'; } export function buildStructuredOutputPrompt( prompt: string, - options?: { + _options?: { reviewerMode?: boolean; + pairedRoom?: boolean; gateTurnKind?: string | null; requiresVisibleVerdict?: boolean; }, ): string { - const lines = [ - '[OUTPUT CONTROL]', - `If you have no user-visible content to send for this turn, output exactly this JSON and nothing else: ${STRUCTURED_SILENT_OUTPUT_ENVELOPE}`, - 'Do not wrap the JSON in backticks or code fences.', - 'Do not combine the JSON with any other text.', - 'If you have already emitted any visible progress, status update, or partial answer earlier in this turn, do not end with the JSON object. Finish with a short visible final conclusion for the user instead.', - ]; - - if (options?.reviewerMode) { - lines.push( - 'If you have not already emitted any visible progress, status update, or partial answer in this turn and you are only agreeing, mirroring, or restating without adding a concrete correction, risk, missing prerequisite, test gap, or code change, output only the JSON object.', - ); - } - - if (options?.reviewerMode && options.requiresVisibleVerdict) { - lines.push( - `This turn is a paired-room gate turn for ${options.gateTurnKind ?? 'implementation_start'}. Silent output is forbidden.`, - ); - lines.push( - 'Your final answer must be a structured public JSON envelope with a reviewer verdict: {"ejclaw":{"visibility":"public","verdict":"done","text":"**DONE** ..."}}', - ); - lines.push( - 'Allowed verdict values are: "done", "done_with_concerns", "blocked".', - ); - } - - return `${lines.join('\n')}\n\n${prompt}`; + return prompt; +} + +export function parseStructuredOutputEnvelope( + _rawText: string, +): null { + return null; } diff --git a/src/paired-execution-context.ts b/src/paired-execution-context.ts index 25aa214..7208b79 100644 --- a/src/paired-execution-context.ts +++ b/src/paired-execution-context.ts @@ -28,6 +28,20 @@ import type { RoomRoleContext, } from './types.js'; +// --------------------------------------------------------------------------- +// Reviewer approval detection +// --------------------------------------------------------------------------- + +// Only check the first line for explicit status markers from the prompt's +// completion status protocol (DONE, DONE_WITH_CONCERNS, BLOCKED, NEEDS_CONTEXT). +// DONE = approved → stop ping-pong. Everything else = continue. +function isReviewerApproval(summary: string | null | undefined): boolean { + if (!summary) return false; + const firstLine = summary.trimStart().split('\n')[0].trim(); + // Match DONE but not DONE_WITH_CONCERNS + return /\bDONE\b/.test(firstLine) && !/\bDONE_WITH_CONCERNS\b/.test(firstLine); +} + // --------------------------------------------------------------------------- // Helpers // --------------------------------------------------------------------------- @@ -168,6 +182,14 @@ export function preparePairedExecutionContext(args: { const now = new Date().toISOString(); if (roomRoleContext.role === 'owner') { + // New human message → new ping-pong cycle. Reset the round trip counter + // so previous interactions don't block the auto-review trigger. + if (latestTask.round_trip_count > 0) { + updatePairedTask(latestTask.id, { + round_trip_count: 0, + updated_at: now, + }); + } workspace = provisionOwnerWorkspaceForPairedTask(latestTask.id); } else { const reviewerWorkspace = prepareReviewerWorkspaceForExecution(latestTask); @@ -268,17 +290,29 @@ export function completePairedExecutionContext(args: { } } - // Reviewer finished → set task back to active so owner can respond + // Reviewer finished → check if approved or needs more work if (role === 'reviewer') { const now = new Date().toISOString(); - updatePairedTask(taskId, { - status: 'active', - updated_at: now, - }); - logger.info( - { taskId }, - 'Reviewer completed, task set back to active for owner', - ); + const approved = isReviewerApproval(args.summary); + if (approved) { + updatePairedTask(taskId, { + status: 'completed', + updated_at: now, + }); + logger.info( + { taskId, summary: args.summary?.slice(0, 100) }, + 'Reviewer approved, task completed — ping-pong stopped', + ); + } else { + updatePairedTask(taskId, { + status: 'active', + updated_at: now, + }); + logger.info( + { taskId }, + 'Reviewer requested changes, task set back to active for owner', + ); + } } } diff --git a/src/service-routing.ts b/src/service-routing.ts index dd5e152..ae907ce 100644 --- a/src/service-routing.ts +++ b/src/service-routing.ts @@ -2,9 +2,10 @@ import { CLAUDE_SERVICE_ID, CODEX_MAIN_SERVICE_ID, CODEX_REVIEW_SERVICE_ID, + OWNER_AGENT_TYPE, + REVIEWER_SERVICE_ID_FOR_TYPE, SERVICE_AGENT_TYPE, SERVICE_ID, - UNIFIED_MODE, normalizeServiceId, } from './config.js'; import { @@ -51,10 +52,13 @@ function getDefaultLease(chatJid: string): EffectiveChannelLease { const hasCodex = types.includes('codex'); if (hasClaude && hasCodex) { + // Owner/reviewer service IDs derived from OWNER_AGENT_TYPE / REVIEWER_AGENT_TYPE env vars + const ownerServiceId = + OWNER_AGENT_TYPE === 'codex' ? CODEX_MAIN_SERVICE_ID : CLAUDE_SERVICE_ID; return { chat_jid: chatJid, - owner_service_id: CLAUDE_SERVICE_ID, - reviewer_service_id: CODEX_MAIN_SERVICE_ID, + owner_service_id: ownerServiceId, + reviewer_service_id: REVIEWER_SERVICE_ID_FOR_TYPE, activated_at: null, reason: null, explicit: false, @@ -140,16 +144,10 @@ export function isReviewerServiceForChat( } export function shouldServiceProcessChat( - chatJid: string, - serviceId: string = SERVICE_ID, + _chatJid: string, + _serviceId: string = SERVICE_ID, ): boolean { - if (UNIFIED_MODE) return true; - const normalizedServiceId = normalizeServiceId(serviceId); - const lease = getEffectiveChannelLease(chatJid); - return ( - normalizedServiceId === lease.owner_service_id || - normalizedServiceId === lease.reviewer_service_id - ); + return true; } export function activateCodexFailover(chatJid: string, reason: string): void {