diff --git a/runners/agent-runner/src/index.ts b/runners/agent-runner/src/index.ts index 83eb598..26f1898 100644 --- a/runners/agent-runner/src/index.ts +++ b/runners/agent-runner/src/index.ts @@ -42,10 +42,13 @@ import { import { drainIpcInput, shouldClose } from './ipc-input.js'; import { MessageStream, + buildCompactionOutput, buildMultimodalContent, + compactBoundaryOutput, extractAssistantText, normalizeStructuredOutput, readStdin, + type RunnerCompaction, writeOutput, } from './output-protocol.js'; import { @@ -90,6 +93,30 @@ function log(message: string): void { console.error(`[agent-runner] ${message}`); } +function compactBoundaryFromMessage( + message: unknown, +): RunnerCompaction | undefined { + if ( + !( + typeof message === 'object' && + message !== null && + (message as { type?: string }).type === 'system' && + (message as { subtype?: string }).subtype === 'compact_boundary' + ) + ) { + return undefined; + } + const meta = ( + message as { + compact_metadata?: { trigger?: string; pre_tokens?: number }; + } + ).compact_metadata; + log( + `Compact boundary — trigger=${meta?.trigger || '?'} pre_tokens=${meta?.pre_tokens ?? '?'}`, + ); + return compactBoundaryOutput(meta?.trigger); +} + // 번들 CLI binary를 명시해야 SDK가 musl/glibc 잘못 탐색하는 걸 우회함. // (SDK 0.2.114의 W7() 헬퍼는 linux-x64-musl를 linux-x64보다 먼저 시도하므로 // glibc 호스트에서 musl 패키지가 빈 껍데기로 설치돼 있으면 실패한다.) @@ -127,6 +154,7 @@ async function runQuery( newSessionId?: string; closedDuringQuery: boolean; terminalResultObserved: boolean; + compaction?: RunnerCompaction; }> { const stream = new MessageStream((text) => buildMultimodalContent(text, log)); stream.push(prompt); @@ -172,6 +200,7 @@ async function runQuery( let resultCount = 0; let terminalResultObserved = false; let pendingProgressText: string | null = null; + let compaction: RunnerCompaction | undefined; const trackedAgentTasks = new TopLevelAgentTaskTracker(); // Discover additional directories @@ -360,19 +389,7 @@ async function runQuery( log(`Session initialized: ${newSessionId}`); } - if ( - message.type === 'system' && - (message as { subtype?: string }).subtype === 'compact_boundary' - ) { - const meta = ( - message as { - compact_metadata?: { trigger?: string; pre_tokens?: number }; - } - ).compact_metadata; - log( - `Compact boundary — trigger=${meta?.trigger || '?'} pre_tokens=${meta?.pre_tokens ?? '?'}`, - ); - } + compaction = compactBoundaryFromMessage(message) ?? compaction; if ( message.type === 'system' && @@ -522,6 +539,7 @@ async function runQuery( result: textResult || null, newSessionId, error: errorText || `Agent error: ${message.subtype}`, + ...buildCompactionOutput(compaction), }); } else { const normalized = normalizeStructuredOutput(textResult || null); @@ -529,6 +547,7 @@ async function runQuery( status: 'success', ...normalized, newSessionId, + ...buildCompactionOutput(compaction), }); } @@ -561,6 +580,7 @@ async function runQuery( status: 'success', ...normalizeStructuredOutput(textResult), newSessionId, + ...buildCompactionOutput(compaction), }); terminalResultObserved = true; ipcPolling = false; @@ -600,6 +620,7 @@ async function runQuery( status: 'success', ...normalizeStructuredOutput(pendingProgressText), newSessionId, + ...buildCompactionOutput(compaction), }); terminalResultObserved = true; resultCount++; @@ -609,7 +630,12 @@ async function runQuery( log( `Query done. Messages: ${messageCount}, results: ${resultCount}, closedDuringQuery: ${closedDuringQuery}`, ); - return { newSessionId, closedDuringQuery, terminalResultObserved }; + return { + newSessionId, + closedDuringQuery, + terminalResultObserved, + compaction, + }; } async function main(): Promise { @@ -698,6 +724,7 @@ async function main(): Promise { log(`Handling session command: ${trimmedPrompt}`); let slashSessionId: string | undefined; let compactBoundarySeen = false; + let slashCompaction: RunnerCompaction | undefined; let hadError = false; let resultEmitted = false; @@ -745,20 +772,10 @@ async function main(): Promise { log(`Session after slash command: ${slashSessionId}`); } - // Observe compact_boundary to confirm compaction completed - if ( - message.type === 'system' && - (message as { subtype?: string }).subtype === 'compact_boundary' - ) { + const observedCompaction = compactBoundaryFromMessage(message); + if (observedCompaction) { compactBoundarySeen = true; - const meta = ( - message as { - compact_metadata?: { trigger?: string; pre_tokens?: number }; - } - ).compact_metadata; - log( - `Compact boundary — trigger=${meta?.trigger || '?'} pre_tokens=${meta?.pre_tokens ?? '?'}`, - ); + slashCompaction = observedCompaction; } if (message.type === 'result') { @@ -775,12 +792,14 @@ async function main(): Promise { result: null, error: textResult || 'Session command failed.', newSessionId: slashSessionId, + ...buildCompactionOutput(slashCompaction), }); } else { writeOutput({ status: 'success', result: textResult || 'Conversation compacted.', newSessionId: slashSessionId, + ...buildCompactionOutput(slashCompaction), }); } resultEmitted = true; @@ -812,6 +831,7 @@ async function main(): Promise { ? 'Conversation compacted.' : 'Compaction requested but compact_boundary was not observed.', newSessionId: slashSessionId, + ...buildCompactionOutput(slashCompaction), }); } else if (!hadError) { // Emit session-only marker so host updates session tracking @@ -819,6 +839,7 @@ async function main(): Promise { status: 'success', result: null, newSessionId: slashSessionId, + ...buildCompactionOutput(slashCompaction), }); } return; @@ -843,7 +864,12 @@ async function main(): Promise { } if (!queryResult.closedDuringQuery && !queryResult.terminalResultObserved) { - writeOutput({ status: 'success', result: null, newSessionId: sessionId }); + writeOutput({ + status: 'success', + result: null, + newSessionId: sessionId, + ...buildCompactionOutput(queryResult.compaction), + }); } else if (queryResult.terminalResultObserved) { log('Terminal result already emitted, exiting single-turn runtime'); } else { diff --git a/runners/agent-runner/src/output-protocol.ts b/runners/agent-runner/src/output-protocol.ts index 3630137..faebaf3 100644 --- a/runners/agent-runner/src/output-protocol.ts +++ b/runners/agent-runner/src/output-protocol.ts @@ -18,6 +18,24 @@ export interface RunnerOutput { output?: RunnerStructuredOutput; newSessionId?: string; error?: string; + compaction?: { + completed: boolean; + trigger?: string | null; + }; +} + +export type RunnerCompaction = NonNullable; + +export function buildCompactionOutput( + compaction: RunnerCompaction | undefined, +): Pick { + return compaction ? { compaction } : {}; +} + +export function compactBoundaryOutput( + trigger: string | null | undefined, +): RunnerCompaction { + return { completed: true, trigger: trigger || null }; } type ContentBlock = diff --git a/runners/codex-runner/src/index.ts b/runners/codex-runner/src/index.ts index 1762fc5..2ecbe0c 100644 --- a/runners/codex-runner/src/index.ts +++ b/runners/codex-runner/src/index.ts @@ -63,6 +63,10 @@ interface RunnerOutput { phase?: 'progress' | 'final'; newSessionId?: string; error?: string; + compaction?: { + completed: boolean; + trigger?: string | null; + }; } // ── Constants ────────────────────────────────────────────────────── @@ -85,6 +89,12 @@ function writeOutput(output: RunnerOutput): void { writeProtocolOutput(output); } +function compactionOutput( + completed: boolean | undefined, +): Pick { + return completed ? { compaction: { completed: true, trigger: null } } : {}; +} + function normalizeStructuredOutput(result: string | null): { result: string | null; output?: RunnerOutput['output']; @@ -212,7 +222,11 @@ async function executeAppServerTurn( threadId: string, prompt: string, retryCount = 0, -): Promise<{ result: string | null; error?: string }> { +): Promise<{ + result: string | null; + error?: string; + compactionCompleted?: boolean; +}> { let lastProgressMessage: string | null = null; const activeTurn = await client.startTurn( threadId, @@ -282,10 +296,10 @@ async function executeAppServerTurn( try { const { state, result } = await activeTurn.wait(); if (state.status === 'completed') { - return { result }; + return { result, compactionCompleted: state.compactionCompleted }; } if (state.status === 'interrupted' && consumeCloseSentinel()) { - return { result }; + return { result, compactionCompleted: state.compactionCompleted }; } if (state.status === 'interrupted' && retryCount < 1) { log('Codex turn interrupted, retrying once...'); @@ -295,6 +309,7 @@ async function executeAppServerTurn( result, error: state.errorMessage || `Codex turn finished with status ${state.status}`, + compactionCompleted: state.compactionCompleted, }; } finally { polling = false; @@ -320,6 +335,7 @@ async function runAppServerCompact( result: null, newSessionId: threadId, error: state.errorMessage || 'Conversation compaction failed.', + ...compactionOutput(state.compactionCompleted), }); return; } @@ -330,6 +346,7 @@ async function runAppServerCompact( ? 'Conversation compacted.' : 'Compaction requested but contextCompaction was not observed.', newSessionId: threadId, + ...compactionOutput(state.compactionCompleted), }); } @@ -386,7 +403,7 @@ async function runAppServerSession( } log('Starting app-server turn...'); - const { result, error } = await executeAppServerTurn( + const { result, error, compactionCompleted } = await executeAppServerTurn( client, threadId, prompt, @@ -400,6 +417,7 @@ async function runAppServerSession( ...normalized, newSessionId: threadId, error, + ...compactionOutput(compactionCompleted), }); } else { const normalized = normalizeStructuredOutput(result || null); @@ -408,6 +426,7 @@ async function runAppServerSession( ...normalized, ...(result ? { phase: 'final' as const } : {}), newSessionId: threadId, + ...compactionOutput(compactionCompleted), }); } diff --git a/src/agent-runner.ts b/src/agent-runner.ts index f1b9a84..6a826da 100644 --- a/src/agent-runner.ts +++ b/src/agent-runner.ts @@ -50,6 +50,10 @@ export interface AgentOutput { agentDone?: boolean; newSessionId?: string; error?: string; + compaction?: { + completed: boolean; + trigger?: string | null; + }; } function readRoomSkillOverridesForRunner( diff --git a/src/compact-refresh.test.ts b/src/compact-refresh.test.ts new file mode 100644 index 0000000..66235ae --- /dev/null +++ b/src/compact-refresh.test.ts @@ -0,0 +1,104 @@ +import fs from 'fs'; +import { beforeEach, describe, expect, it, vi } from 'vitest'; + +vi.mock('./config.js', () => ({ + DATA_DIR: '/tmp/ejclaw-compact-refresh-test', +})); + +import { + clearCompactRefreshIfUnchanged, + markCompactRefreshNeeded, + maybeApplyCompactRefresh, + readCompactRefreshFlag, +} from './compact-refresh.js'; + +const refreshDir = '/tmp/ejclaw-compact-refresh-test/compact-refresh'; + +describe('compact-refresh', () => { + beforeEach(() => { + fs.rmSync(refreshDir, { recursive: true, force: true }); + }); + + it('marks and reads compact refresh flags by session folder', () => { + const flag = markCompactRefreshNeeded({ + sessionFolder: 'room:reviewer', + sessionId: 'session-1', + trigger: 'auto', + }); + + expect(readCompactRefreshFlag('room:reviewer')).toMatchObject({ + sessionFolder: 'room:reviewer', + sessionId: 'session-1', + trigger: 'auto', + compactedAt: flag.compactedAt, + }); + }); + + it('applies a one-shot compact refresh only for owner and reviewer roles', () => { + const flag = markCompactRefreshNeeded({ + sessionFolder: 'room', + sessionId: 'session-1', + }); + + const applied = maybeApplyCompactRefresh({ + sessionFolder: 'room', + sessionId: 'session-1', + role: 'owner', + prompt: 'continue task', + }); + + expect(applied?.flag).toEqual(flag); + expect(applied?.prompt).toContain('EJClaw compact refresh'); + expect(applied?.prompt).toContain('continue task'); + + clearCompactRefreshIfUnchanged({ + sessionFolder: 'room', + flag, + }); + + expect(readCompactRefreshFlag('room')).toBeNull(); + }); + + it('does not apply to arbiter, compact commands, fresh sessions, or stale sessions', () => { + markCompactRefreshNeeded({ + sessionFolder: 'room', + sessionId: 'session-1', + }); + + expect( + maybeApplyCompactRefresh({ + sessionFolder: 'room', + sessionId: 'session-1', + role: 'arbiter', + prompt: 'judge', + }), + ).toBeNull(); + + expect( + maybeApplyCompactRefresh({ + sessionFolder: 'room', + sessionId: 'session-1', + role: 'owner', + prompt: '/compact', + }), + ).toBeNull(); + + expect( + maybeApplyCompactRefresh({ + sessionFolder: 'room', + role: 'owner', + prompt: 'fresh', + }), + ).toBeNull(); + + expect( + maybeApplyCompactRefresh({ + sessionFolder: 'room', + sessionId: 'different-session', + role: 'owner', + prompt: 'stale', + }), + ).toBeNull(); + expect(readCompactRefreshFlag('room')).toBeNull(); + }); +}); diff --git a/src/compact-refresh.ts b/src/compact-refresh.ts new file mode 100644 index 0000000..2cf069d --- /dev/null +++ b/src/compact-refresh.ts @@ -0,0 +1,128 @@ +import fs from 'fs'; +import path from 'path'; + +import { DATA_DIR } from './config.js'; +import type { PairedRoomRole } from './types.js'; + +export interface CompactRefreshFlag { + sessionFolder: string; + sessionId: string; + compactedAt: string; + trigger?: string | null; +} + +export interface AppliedCompactRefresh { + flag: CompactRefreshFlag; + prompt: string; +} + +const COMPACT_REFRESH_DIR = path.join(DATA_DIR, 'compact-refresh'); + +const COMPACT_REFRESH_PROMPT = `[EJClaw compact refresh] +The previous run compacted this same SDK session. Continue following the current AGENTS.md/CLAUDE.md role rules exactly: required first-line status, paired-room role boundaries, output-only task context, concise Korean responses, and verification before completion. This is not new user scope. +[/EJClaw compact refresh]`; + +function getFlagPath(sessionFolder: string): string { + return path.join( + COMPACT_REFRESH_DIR, + `${encodeURIComponent(sessionFolder)}.json`, + ); +} + +function isRefreshableRole(role: PairedRoomRole): boolean { + return role === 'owner' || role === 'reviewer'; +} + +function isSessionCommand(prompt: string): boolean { + return prompt.trim() === '/compact'; +} + +export function readCompactRefreshFlag( + sessionFolder: string, +): CompactRefreshFlag | null { + const flagPath = getFlagPath(sessionFolder); + if (!fs.existsSync(flagPath)) return null; + + try { + const parsed = JSON.parse( + fs.readFileSync(flagPath, 'utf-8'), + ) as Partial; + if ( + parsed.sessionFolder !== sessionFolder || + typeof parsed.sessionId !== 'string' || + parsed.sessionId.length === 0 || + typeof parsed.compactedAt !== 'string' + ) { + fs.rmSync(flagPath, { force: true }); + return null; + } + return { + sessionFolder, + sessionId: parsed.sessionId, + compactedAt: parsed.compactedAt, + trigger: + typeof parsed.trigger === 'string' && parsed.trigger.length > 0 + ? parsed.trigger + : null, + }; + } catch { + fs.rmSync(flagPath, { force: true }); + return null; + } +} + +export function markCompactRefreshNeeded(args: { + sessionFolder: string; + sessionId: string; + trigger?: string | null; +}): CompactRefreshFlag { + const flag: CompactRefreshFlag = { + sessionFolder: args.sessionFolder, + sessionId: args.sessionId, + compactedAt: new Date().toISOString(), + trigger: args.trigger ?? null, + }; + fs.mkdirSync(COMPACT_REFRESH_DIR, { recursive: true }); + fs.writeFileSync( + getFlagPath(args.sessionFolder), + `${JSON.stringify(flag)}\n`, + ); + return flag; +} + +export function clearCompactRefreshIfUnchanged(args: { + sessionFolder: string; + flag: CompactRefreshFlag; +}): void { + const current = readCompactRefreshFlag(args.sessionFolder); + if ( + current && + current.sessionId === args.flag.sessionId && + current.compactedAt === args.flag.compactedAt + ) { + fs.rmSync(getFlagPath(args.sessionFolder), { force: true }); + } +} + +export function maybeApplyCompactRefresh(args: { + sessionFolder: string; + sessionId?: string; + role: PairedRoomRole; + prompt: string; +}): AppliedCompactRefresh | null { + if (!args.sessionId) return null; + if (!isRefreshableRole(args.role)) return null; + if (isSessionCommand(args.prompt)) return null; + + const flag = readCompactRefreshFlag(args.sessionFolder); + if (!flag) return null; + if (flag.sessionId !== args.sessionId) { + fs.rmSync(getFlagPath(args.sessionFolder), { force: true }); + return null; + } + + return { + flag, + prompt: `${COMPACT_REFRESH_PROMPT}\n\n${args.prompt}`, + }; +} diff --git a/src/message-agent-compact-refresh.test.ts b/src/message-agent-compact-refresh.test.ts new file mode 100644 index 0000000..95bdde8 --- /dev/null +++ b/src/message-agent-compact-refresh.test.ts @@ -0,0 +1,114 @@ +import fs from 'fs'; +import { beforeEach, describe, expect, it, vi } from 'vitest'; + +vi.mock('./config.js', () => ({ + DATA_DIR: '/tmp/ejclaw-message-compact-refresh-test', +})); + +vi.mock('./agent-runner.js', () => ({ + runAgentProcess: vi.fn(), +})); + +vi.mock('./codex-token-rotation.js', () => ({ + getCodexAccountCount: vi.fn(() => 1), +})); + +vi.mock('./session-recovery.js', () => ({ + shouldResetCodexSessionOnAgentFailure: vi.fn(() => false), + shouldResetSessionOnAgentFailure: vi.fn(() => false), + shouldRetryFreshCodexSessionOnAgentFailure: vi.fn(() => false), +})); + +import { runAgentProcess, type AgentOutput } from './agent-runner.js'; +import { + maybeApplyCompactRefresh, + readCompactRefreshFlag, +} from './compact-refresh.js'; +import { runMessageAgentAttempt } from './message-agent-executor-attempt-runner.js'; +import type { RegisteredGroup } from './types.js'; + +const dataDir = '/tmp/ejclaw-message-compact-refresh-test'; + +const group: RegisteredGroup = { + name: 'Room', + folder: 'room', + trigger: '@Andy', + added_at: new Date().toISOString(), + agentType: 'claude-code', +}; + +function makeLifecycle() { + return { + updateSummary: vi.fn(), + recordFinalOutputBeforeDelivery: vi.fn(() => true), + }; +} + +function makeLogger(): any { + const logger = { + info: vi.fn(), + warn: vi.fn(), + error: vi.fn(), + child: vi.fn(), + }; + logger.child.mockReturnValue(logger); + return logger; +} + +describe('message agent compact refresh', () => { + beforeEach(() => { + fs.rmSync(dataDir, { recursive: true, force: true }); + vi.resetAllMocks(); + }); + + it('marks compacted owner sessions so the next turn gets one minimal refresh', async () => { + const compactedOutput: AgentOutput = { + status: 'success', + result: 'done', + output: { visibility: 'public', text: 'done' }, + newSessionId: 'session-1', + compaction: { completed: true, trigger: 'auto' }, + }; + vi.mocked(runAgentProcess).mockResolvedValue(compactedOutput); + + await runMessageAgentAttempt({ + provider: 'claude', + currentSessionId: 'session-1', + isClaudeCodeAgent: true, + canRetryClaudeCredentials: false, + shouldPersistSession: true, + effectiveGroup: group, + agentInput: { + prompt: 'work', + sessionId: 'session-1', + groupFolder: 'room', + chatJid: 'room@test', + runId: 'run-1', + isMain: false, + assistantName: 'Andy', + }, + activeRole: 'owner', + effectiveServiceId: 'claude', + effectiveAgentType: 'claude-code', + sessionFolder: 'room', + onPersistSession: vi.fn(), + registerProcess: vi.fn(), + pairedExecutionLifecycle: makeLifecycle(), + log: makeLogger(), + }); + + expect(readCompactRefreshFlag('room')).toMatchObject({ + sessionId: 'session-1', + trigger: 'auto', + }); + + const applied = maybeApplyCompactRefresh({ + sessionFolder: 'room', + sessionId: 'session-1', + role: 'owner', + prompt: 'next work', + }); + expect(applied?.prompt).toContain('EJClaw compact refresh'); + expect(applied?.prompt).toContain('next work'); + }); +}); diff --git a/src/message-agent-executor-attempt-runner.ts b/src/message-agent-executor-attempt-runner.ts index 30d015d..fc2c792 100644 --- a/src/message-agent-executor-attempt-runner.ts +++ b/src/message-agent-executor-attempt-runner.ts @@ -3,6 +3,7 @@ import type { Logger } from 'pino'; import { createEvaluatedOutputHandler } from './agent-attempt.js'; import type { AttemptStreamedTrigger } from './agent-attempt-retry.js'; import { runAgentProcess, type AgentOutput } from './agent-runner.js'; +import { markCompactRefreshNeeded } from './compact-refresh.js'; import { getCodexAccountCount } from './codex-token-rotation.js'; import type { PreparedPairedExecutionContext } from './paired-execution-context.js'; import { @@ -34,6 +35,35 @@ interface AgentInput { roomRoleContext?: RoomRoleContext; } +function maybeMarkCompactRefreshForOutput(args: { + output: AgentOutput; + activeRole: string; + sessionFolder: string; +}): void { + if ( + (args.activeRole !== 'owner' && args.activeRole !== 'reviewer') || + args.output.compaction?.completed !== true || + !args.output.newSessionId + ) { + return; + } + markCompactRefreshNeeded({ + sessionFolder: args.sessionFolder, + sessionId: args.output.newSessionId, + trigger: args.output.compaction.trigger ?? null, + }); +} + +function createProviderLog( + log: Logger, + provider: 'claude' | 'codex', + agentType: AgentType, +): Logger { + const providerLog = log.child({ provider, agentType }); + providerLog.info('Using provider'); + return providerLog; +} + export async function runMessageAgentAttempt(args: { provider: 'claude' | 'codex'; currentSessionId: string | undefined; @@ -111,6 +141,7 @@ export async function runMessageAgentAttempt(args: { structuredOutput, evaluation, }) => { + maybeMarkCompactRefreshForOutput({ output, activeRole, sessionFolder }); const outputPhase = output.phase ?? 'final'; if (outputPhase !== 'final') { log.info( @@ -252,11 +283,7 @@ export async function runMessageAgentAttempt(args: { await streamedOutputHandler.handleOutput(output); }; - const providerLog = log.child({ - provider, - agentType: effectiveAgentType, - }); - providerLog.info('Using provider'); + const providerLog = createProviderLog(log, provider, effectiveAgentType); try { const output = await runAgentProcess( @@ -273,6 +300,7 @@ export async function runMessageAgentAttempt(args: { if (output.newSessionId && shouldPersistSession) { onPersistSession(output.newSessionId); } + maybeMarkCompactRefreshForOutput({ output, activeRole, sessionFolder }); providerLog.info( { diff --git a/src/message-agent-executor.ts b/src/message-agent-executor.ts index 1b6c26b..d230c07 100644 --- a/src/message-agent-executor.ts +++ b/src/message-agent-executor.ts @@ -1,5 +1,6 @@ import { AgentOutput } from './agent-runner.js'; import { GroupQueue } from './group-queue.js'; +import type { Logger } from 'pino'; import type { AgentTriggerReason } from './agent-error-detection.js'; import { getMoaConfig } from './config.js'; import { createPairedExecutionLifecycle } from './message-agent-executor-paired.js'; @@ -12,9 +13,82 @@ import { collectMoaReferences, formatMoaReferencesForPrompt } from './moa.js'; import { readArbiterPrompt } from './platform-prompts.js'; import { shouldRetryFreshSessionOnAgentFailure } from './session-recovery.js'; import type { AgentType, PairedRoomRole, RegisteredGroup } from './types.js'; +import { + clearCompactRefreshIfUnchanged, + maybeApplyCompactRefresh, + type AppliedCompactRefresh, +} from './compact-refresh.js'; // ── Main executor ───────────────────────────────────────────────── +function buildPromptWithCompactRefresh(args: { + prompt: string; + sessionFolder: string; + sessionId?: string; + role: PairedRoomRole; +}): { effectivePrompt: string; compactRefresh: AppliedCompactRefresh | null } { + const compactRefresh = maybeApplyCompactRefresh({ + sessionFolder: args.sessionFolder, + sessionId: args.sessionId, + role: args.role, + prompt: args.prompt, + }); + return { + effectivePrompt: compactRefresh?.prompt ?? args.prompt, + compactRefresh, + }; +} + +function clearAppliedCompactRefreshAfterSuccess(args: { + result: 'success' | 'error'; + sessionFolder: string; + compactRefresh: AppliedCompactRefresh | null; +}): void { + if (args.result !== 'success' || !args.compactRefresh) return; + clearCompactRefreshIfUnchanged({ + sessionFolder: args.sessionFolder, + flag: args.compactRefresh.flag, + }); +} + +async function enrichArbiterPromptWithMoa(args: { + prompt: string; + enabled: boolean; + log: Pick; +}): Promise { + const moaConfig = getMoaConfig(); + if (!args.enabled || !moaConfig.enabled) return args.prompt; + args.log.info( + { models: moaConfig.referenceModels.map((m) => m.name) }, + 'MoA: collecting reference opinions before arbiter', + ); + const systemPrompt = + readArbiterPrompt(process.cwd()) || 'You are an arbiter.'; + try { + const references = await collectMoaReferences({ + config: moaConfig, + systemPrompt, + contextPrompt: args.prompt, + }); + const moaSection = formatMoaReferencesForPrompt(references); + if (!moaSection) return args.prompt; + args.log.info( + { + successCount: references.filter((r) => !r.error).length, + totalCount: references.length, + }, + 'MoA: injected reference opinions into arbiter prompt', + ); + return args.prompt + '\n' + moaSection; + } catch (err) { + args.log.warn( + { err, models: moaConfig.referenceModels.map((m) => m.name) }, + 'MoA: failed to collect reference opinions; continuing without enrichment', + ); + return args.prompt; + } +} + export interface MessageAgentExecutorDeps { assistantName: string; queue: Pick & @@ -81,54 +155,18 @@ export async function runAgentForGroup( clearRoleSdkSessions(); } - // ── MoA prompt enrichment ───────────────────────────────────── - // When MoA is enabled and we're in arbiter mode, query external API - // models (Kimi, GLM, etc.) in parallel for their opinions, then inject - // those opinions into the arbiter's prompt. The SDK-based arbiter - // agent naturally aggregates all perspectives. - let moaEnrichedPrompt = prompt; - const moaConfig = getMoaConfig(); - if (arbiterMode && moaConfig.enabled && pairedExecutionContext) { - log.info( - { - models: moaConfig.referenceModels.map((m) => m.name), - }, - 'MoA: collecting reference opinions before arbiter', - ); + const moaEnrichedPrompt = await enrichArbiterPromptWithMoa({ + prompt, + enabled: arbiterMode && Boolean(pairedExecutionContext), + log, + }); - const systemPrompt = - readArbiterPrompt(process.cwd()) || 'You are an arbiter.'; - - try { - const references = await collectMoaReferences({ - config: moaConfig, - systemPrompt, - contextPrompt: prompt, - }); - - const moaSection = formatMoaReferencesForPrompt(references); - if (moaSection) { - moaEnrichedPrompt = prompt + '\n' + moaSection; - log.info( - { - successCount: references.filter((r) => !r.error).length, - totalCount: references.length, - }, - 'MoA: injected reference opinions into arbiter prompt', - ); - } - } catch (err) { - log.warn( - { - err, - models: moaConfig.referenceModels.map((m) => m.name), - }, - 'MoA: failed to collect reference opinions; continuing without enrichment', - ); - } - } - - const effectivePrompt = moaEnrichedPrompt; + const { effectivePrompt, compactRefresh } = buildPromptWithCompactRefresh({ + prompt: moaEnrichedPrompt, + sessionFolder, + sessionId: currentSessionId, + role: activeRole, + }); const pairedExecutionLifecycle = createPairedExecutionLifecycle({ pairedExecutionContext, pairedTurnIdentity: runtimePairedTurnIdentity, @@ -237,7 +275,7 @@ export async function runAgentForGroup( }); try { - return await executeMessageAgentAttemptLifecycle({ + const result = await executeMessageAgentAttemptLifecycle({ provider, runAttempt, isClaudeCodeAgent, @@ -260,6 +298,12 @@ export async function runAgentForGroup( }, log, }); + clearAppliedCompactRefreshAfterSuccess({ + result, + sessionFolder, + compactRefresh, + }); + return result; } finally { await pairedExecutionLifecycle.asyncFinalize(); }