feat: paired review system overhaul — unified service, configurable agents, silent output removal
- 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
This commit is contained in:
@@ -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)
|
||||
@@ -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)
|
||||
33
CLAUDE.md
33
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/` |
|
||||
|
||||
@@ -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) {
|
||||
// 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}`,
|
||||
|
||||
@@ -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');
|
||||
@@ -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());
|
||||
}
|
||||
|
||||
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;
|
||||
}
|
||||
|
||||
@@ -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();
|
||||
|
||||
@@ -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');
|
||||
});
|
||||
|
||||
@@ -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 =
|
||||
|
||||
12
src/db.ts
12
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,
|
||||
|
||||
34
src/index.ts
34
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,8 +322,7 @@ async function main(): Promise<void> {
|
||||
initTokenRotation();
|
||||
initCodexTokenRotation();
|
||||
|
||||
// Unified mode: start credential proxy for container isolation and clean up orphaned containers
|
||||
if (UNIFIED_MODE) {
|
||||
// Start credential proxy for container isolation and clean up orphaned containers
|
||||
startCredentialProxy(CREDENTIAL_PROXY_PORT).catch((err) =>
|
||||
logger.warn(
|
||||
{ err },
|
||||
@@ -340,18 +330,8 @@ async function main(): Promise<void> {
|
||||
),
|
||||
);
|
||||
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',
|
||||
);
|
||||
}
|
||||
|
||||
loadState();
|
||||
|
||||
@@ -456,8 +436,6 @@ async function main(): Promise<void> {
|
||||
}
|
||||
|
||||
// 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,
|
||||
@@ -471,12 +449,6 @@ async function main(): Promise<void> {
|
||||
editTrackedMessage: (jid, messageId, rawText) =>
|
||||
editFormattedTrackedChannelMessage(channels, jid, messageId, rawText),
|
||||
});
|
||||
} else {
|
||||
logger.info(
|
||||
{ serviceId: SERVICE_ID },
|
||||
'Skipping scheduler for review service',
|
||||
);
|
||||
}
|
||||
startIpcWatcher({
|
||||
sendMessage: (jid, text) => {
|
||||
const channel = findChannel(channels, jid);
|
||||
@@ -538,7 +510,6 @@ async function main(): Promise<void> {
|
||||
purgeOnStart: true,
|
||||
});
|
||||
|
||||
if (UNIFIED_MODE || isClaudeService()) {
|
||||
leaseRecoveryTimer = setInterval(() => {
|
||||
if (!hasAvailableClaudeToken()) {
|
||||
return;
|
||||
@@ -578,7 +549,6 @@ async function main(): Promise<void> {
|
||||
);
|
||||
}
|
||||
}, 5_000);
|
||||
}
|
||||
runtime.startMessageLoop().catch((err) => {
|
||||
logger.fatal({ err }, 'Message loop crashed unexpectedly');
|
||||
process.exit(1);
|
||||
|
||||
@@ -142,6 +142,7 @@ function makeDeps() {
|
||||
assistantName: 'Andy',
|
||||
queue: {
|
||||
registerProcess: vi.fn(),
|
||||
enqueueMessageCheck: vi.fn(),
|
||||
},
|
||||
getRegisteredGroups: () => ({}),
|
||||
getSessions: () => ({}),
|
||||
|
||||
@@ -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') {
|
||||
if (pairedExecutionContext && pairedExecutionStatus === 'succeeded') {
|
||||
deps.queue.enqueueMessageCheck(chatJid);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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<typeof filterProcessableMessages>[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));
|
||||
}
|
||||
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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);
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
@@ -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,18 +290,30 @@ 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();
|
||||
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 completed, task set back to active for owner',
|
||||
'Reviewer requested changes, task set back to active for owner',
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
@@ -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 {
|
||||
|
||||
Reference in New Issue
Block a user