From cbb6bc97c772ccfc9ec32ce37886ed6bc30bd12f Mon Sep 17 00:00:00 2001 From: ejclaw Date: Tue, 7 Apr 2026 00:45:17 +0900 Subject: [PATCH] Refactor paired execution flow boundaries --- runners/agent-runner/src/verification.ts | 48 +- shared/verification-snapshot.d.ts | 8 + shared/verification-snapshot.js | 50 +++ src/agent-attempt.ts | 71 +++ src/message-agent-executor.test.ts | 32 +- src/message-agent-executor.ts | 311 +++++++------ src/message-runtime-flow.ts | 366 +++++++++++++++ src/message-runtime.ts | 363 +++------------ src/paired-execution-context-arbiter.ts | 81 ++++ src/paired-execution-context-owner.ts | 201 +++++++++ src/paired-execution-context-reviewer.ts | 139 ++++++ src/paired-execution-context-shared.ts | 122 +++++ src/paired-execution-context.ts | 542 +---------------------- src/task-scheduler.ts | 81 ++-- src/verification.ts | 46 +- 15 files changed, 1345 insertions(+), 1116 deletions(-) create mode 100644 shared/verification-snapshot.d.ts create mode 100644 shared/verification-snapshot.js create mode 100644 src/agent-attempt.ts create mode 100644 src/message-runtime-flow.ts create mode 100644 src/paired-execution-context-arbiter.ts create mode 100644 src/paired-execution-context-owner.ts create mode 100644 src/paired-execution-context-reviewer.ts create mode 100644 src/paired-execution-context-shared.ts diff --git a/runners/agent-runner/src/verification.ts b/runners/agent-runner/src/verification.ts index eebf243..b274942 100644 --- a/runners/agent-runner/src/verification.ts +++ b/runners/agent-runner/src/verification.ts @@ -1,6 +1,9 @@ -import { createHash } from 'crypto'; import fs from 'fs'; import path from 'path'; +export { + computeVerificationSnapshotId, + resolveVerificationResponsesDir, +} from '../../../shared/verification-snapshot.js'; export const VERIFICATION_PROFILES = [ 'test', @@ -24,8 +27,6 @@ export interface VerificationResponse { error?: string; } -const SNAPSHOT_EXCLUDE_NAMES = new Set(['.git', 'node_modules', '.env']); - export function isVerificationProfile( value: unknown, ): value is VerificationProfile { @@ -35,47 +36,6 @@ export function isVerificationProfile( ); } -function updateSnapshotHash( - hash: ReturnType, - repoDir: string, - currentPath: string, -): void { - const relPath = path.relative(repoDir, currentPath) || '.'; - const stat = fs.lstatSync(currentPath); - - if (stat.isDirectory()) { - if (relPath !== '.') { - hash.update(`dir\0${relPath}\0`); - } - for (const entry of fs.readdirSync(currentPath).sort()) { - if (SNAPSHOT_EXCLUDE_NAMES.has(entry)) continue; - updateSnapshotHash(hash, repoDir, path.join(currentPath, entry)); - } - return; - } - - if (stat.isSymbolicLink()) { - hash.update(`symlink\0${relPath}\0${fs.readlinkSync(currentPath)}\0`); - return; - } - - if (stat.isFile()) { - hash.update(`file\0${relPath}\0`); - hash.update(fs.readFileSync(currentPath)); - hash.update('\0'); - } -} - -export function computeVerificationSnapshotId(repoDir: string): string { - const hash = createHash('sha256'); - updateSnapshotHash(hash, repoDir, repoDir); - return `fs:${hash.digest('hex').slice(0, 24)}`; -} - -export function resolveVerificationResponsesDir(hostIpcDir: string): string { - return path.join(hostIpcDir, 'verification-responses'); -} - export async function waitForVerificationResponse( responseDir: string, requestId: string, diff --git a/shared/verification-snapshot.d.ts b/shared/verification-snapshot.d.ts new file mode 100644 index 0000000..0102eda --- /dev/null +++ b/shared/verification-snapshot.d.ts @@ -0,0 +1,8 @@ +export declare const VERIFICATION_SNAPSHOT_EXCLUDE_NAMES: ReadonlySet; +export declare function isVerificationSnapshotExcludedName( + name: string, +): boolean; +export declare function computeVerificationSnapshotId(repoDir: string): string; +export declare function resolveVerificationResponsesDir( + hostIpcDir: string, +): string; diff --git a/shared/verification-snapshot.js b/shared/verification-snapshot.js new file mode 100644 index 0000000..9e0a968 --- /dev/null +++ b/shared/verification-snapshot.js @@ -0,0 +1,50 @@ +import { createHash } from 'crypto'; +import fs from 'fs'; +import path from 'path'; + +export const VERIFICATION_SNAPSHOT_EXCLUDE_NAMES = new Set([ + '.git', + 'node_modules', + '.env', +]); + +export function isVerificationSnapshotExcludedName(name) { + return VERIFICATION_SNAPSHOT_EXCLUDE_NAMES.has(name); +} + +function updateVerificationSnapshotHash(hash, repoDir, currentPath) { + const relPath = path.relative(repoDir, currentPath) || '.'; + const stat = fs.lstatSync(currentPath); + + if (stat.isDirectory()) { + if (relPath !== '.') { + hash.update(`dir\0${relPath}\0`); + } + for (const entry of fs.readdirSync(currentPath).sort()) { + if (isVerificationSnapshotExcludedName(entry)) continue; + updateVerificationSnapshotHash(hash, repoDir, path.join(currentPath, entry)); + } + return; + } + + if (stat.isSymbolicLink()) { + hash.update(`symlink\0${relPath}\0${fs.readlinkSync(currentPath)}\0`); + return; + } + + if (stat.isFile()) { + hash.update(`file\0${relPath}\0`); + hash.update(fs.readFileSync(currentPath)); + hash.update('\0'); + } +} + +export function computeVerificationSnapshotId(repoDir) { + const hash = createHash('sha256'); + updateVerificationSnapshotHash(hash, repoDir, repoDir); + return `fs:${hash.digest('hex').slice(0, 24)}`; +} + +export function resolveVerificationResponsesDir(hostIpcDir) { + return path.join(hostIpcDir, 'verification-responses'); +} diff --git a/src/agent-attempt.ts b/src/agent-attempt.ts new file mode 100644 index 0000000..d22c39d --- /dev/null +++ b/src/agent-attempt.ts @@ -0,0 +1,71 @@ +import { + getAgentOutputText, + getStructuredAgentOutput, +} from './agent-output.js'; +import type { AgentOutput } from './agent-runner.js'; +import { + evaluateStreamedOutput, + type EvaluateStreamedOutputOptions, + type EvaluateStreamedOutputResult, + type StreamedOutputState, +} from './streamed-output-evaluator.js'; + +export interface EvaluatedAgentOutput { + output: AgentOutput; + outputText: string | null; + structuredOutput: ReturnType; + evaluation: EvaluateStreamedOutputResult; +} + +export function createInitialStreamedOutputState(): StreamedOutputState { + return { + sawOutput: false, + sawVisibleOutput: false, + sawSuccessNullResultWithoutOutput: false, + }; +} + +export function createEvaluatedOutputHandler(args: { + agentType: EvaluateStreamedOutputOptions['agentType']; + provider: string; + evaluationOptions?: Omit; + onEvaluatedOutput?: ( + output: EvaluatedAgentOutput, + ) => Promise | void; +}): { + handleOutput: (output: AgentOutput) => Promise; + getState: () => StreamedOutputState; + markVisibleOutput: () => void; +} { + let state = createInitialStreamedOutputState(); + + return { + async handleOutput(output: AgentOutput): Promise { + const rawOutputText = getAgentOutputText(output); + const outputText = + typeof rawOutputText === 'string' ? rawOutputText : null; + const structuredOutput = getStructuredAgentOutput(output); + const evaluation = evaluateStreamedOutput(output, state, { + agentType: args.agentType, + provider: args.provider, + ...args.evaluationOptions, + }); + state = evaluation.state; + await args.onEvaluatedOutput?.({ + output, + outputText, + structuredOutput, + evaluation, + }); + }, + getState(): StreamedOutputState { + return state; + }, + markVisibleOutput(): void { + state = { + ...state, + sawVisibleOutput: true, + }; + }, + }; +} diff --git a/src/message-agent-executor.test.ts b/src/message-agent-executor.test.ts index 0cdf23a..ad0443f 100644 --- a/src/message-agent-executor.test.ts +++ b/src/message-agent-executor.test.ts @@ -288,7 +288,7 @@ describe('runAgentForGroup room memory', () => { memoryBriefing: '## Shared Room Memory\n- remembered context', }), expect.any(Function), - undefined, + expect.any(Function), undefined, ); }); @@ -317,7 +317,7 @@ describe('runAgentForGroup room memory', () => { memoryBriefing: undefined, }), expect.any(Function), - undefined, + expect.any(Function), undefined, ); }); @@ -338,7 +338,7 @@ describe('runAgentForGroup room memory', () => { prompt: 'hello', }), expect.any(Function), - undefined, + expect.any(Function), undefined, ); }); @@ -365,7 +365,7 @@ describe('runAgentForGroup room memory', () => { }), }), expect.any(Function), - undefined, + expect.any(Function), undefined, ); }); @@ -398,7 +398,7 @@ describe('runAgentForGroup room memory', () => { prompt: 'hello', }), expect.any(Function), - undefined, + expect.any(Function), undefined, ); }); @@ -455,7 +455,7 @@ describe('runAgentForGroup room memory', () => { }), }), expect.any(Function), - undefined, + expect.any(Function), undefined, ); }); @@ -513,7 +513,7 @@ describe('runAgentForGroup room memory', () => { }), }), expect.any(Function), - undefined, + expect.any(Function), undefined, ); }); @@ -578,7 +578,7 @@ describe('runAgentForGroup room memory', () => { sessionId: undefined, }), expect.any(Function), - undefined, + expect.any(Function), undefined, ); }); @@ -635,7 +635,7 @@ describe('runAgentForGroup room memory', () => { }), expect.any(Object), expect.any(Function), - undefined, + expect.any(Function), {}, ); }); @@ -772,22 +772,22 @@ describe('runAgentForGroup room memory', () => { roomRoleContext: expect.any(Object), }), expect.any(Function), - undefined, + expect.any(Function), expect.objectContaining({ EJCLAW_WORK_DIR: '/tmp/paired/owner', EJCLAW_PAIRED_TASK_ID: 'paired-task-1', EJCLAW_PAIRED_ROLE: 'owner', }), ); - // Owner produced no visible output (mock doesn't go through streamed - // evaluator) → treated as interrupted, status is 'failed' to prevent - // auto-triggering the reviewer. + // The shared attempt harness now routes runner output through the streamed + // evaluator, so a visible public result is treated as a successful owner + // turn. expect( pairedExecutionContext.completePairedExecutionContext, ).toHaveBeenCalledWith({ taskId: 'paired-task-1', role: 'owner', - status: 'failed', + status: 'succeeded', summary: 'ok', }); }); @@ -1045,7 +1045,7 @@ describe('runAgentForGroup room memory', () => { sessionId: 'reviewer-session', }), expect.any(Function), - undefined, + expect.any(Function), undefined, ); }); @@ -1144,7 +1144,7 @@ describe('runAgentForGroup room memory', () => { sessionId: undefined, }), expect.any(Function), - undefined, + expect.any(Function), expect.objectContaining({ EJCLAW_UNSAFE_HOST_PAIRED_MODE: '1', }), diff --git a/src/message-agent-executor.ts b/src/message-agent-executor.ts index d54c678..07bdd9d 100644 --- a/src/message-agent-executor.ts +++ b/src/message-agent-executor.ts @@ -5,8 +5,8 @@ import { getErrorMessage } from './utils.js'; import { getAgentOutputText, - getStructuredAgentOutput, } from './agent-output.js'; +import { createEvaluatedOutputHandler } from './agent-attempt.js'; import { AgentOutput, runAgentProcess, @@ -63,10 +63,6 @@ import { activateCodexFailover, getEffectiveChannelLease, } from './service-routing.js'; -import { - evaluateStreamedOutput, - type StreamedOutputState, -} from './streamed-output-evaluator.js'; import { detectCodexRotationTrigger, getCodexAccountCount, @@ -290,21 +286,31 @@ export async function runAgentForGroup( const systemPrompt = readArbiterPrompt(process.cwd()) || 'You are an arbiter.'; - const references = await collectMoaReferences({ - config: moaConfig, - systemPrompt, - contextPrompt: prompt, - }); + try { + const references = await collectMoaReferences({ + config: moaConfig, + systemPrompt, + contextPrompt: prompt, + }); - const moaSection = formatMoaReferencesForPrompt(references); - if (moaSection) { - moaEnrichedPrompt = prompt + '\n' + moaSection; - log.info( + 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( { - successCount: references.filter((r) => !r.error).length, - totalCount: references.length, + err, + models: moaConfig.referenceModels.map((m) => m.name), }, - 'MoA: injected reference opinions into arbiter prompt', + 'MoA: failed to collect reference opinions; continuing without enrichment', ); } } @@ -411,151 +417,142 @@ export async function runAgentForGroup( retryAfterMs?: number; }; }> => { - let streamedState: StreamedOutputState = { - sawOutput: false, - sawVisibleOutput: false, - sawSuccessNullResultWithoutOutput: false, - }; const attemptSessionId = currentSessionId; + const streamedOutputHandler = createEvaluatedOutputHandler({ + agentType: isClaudeCodeAgent ? 'claude-code' : 'codex', + provider, + evaluationOptions: { + suppressClaudeAuthErrorOutput: provider === 'claude', + trackSuccessNullResult: true, + shortCircuitTriggeredErrors: + provider === 'claude' + ? canRetryClaudeCredentials + : getCodexAccountCount() > 1, + }, + onEvaluatedOutput: async ({ + output, + outputText, + structuredOutput, + evaluation, + }) => { + const outputPhase = output.phase ?? 'final'; + if (outputPhase !== 'final') { + log.info( + { + provider, + outputPhase, + outputStatus: output.status, + visibility: structuredOutput?.visibility ?? null, + preview: + outputText && outputText.length > 0 + ? outputText.slice(0, 160) + : null, + errorPreview: + typeof output.error === 'string' && output.error.length > 0 + ? output.error.slice(0, 160) + : null, + activeRole, + effectiveServiceId, + effectiveAgentType, + sessionFolder, + resumedSession: attemptSessionId ?? null, + streamedSessionId: output.newSessionId ?? null, + roomRoleServiceId: roomRoleContext?.serviceId ?? null, + roomRole: roomRoleContext?.role ?? null, + pairedTaskId: pairedExecutionContext?.task.id ?? null, + workspaceDir: + pairedExecutionContext?.workspace?.workspace_dir ?? + group.workDir ?? + null, + }, + 'Observed streamed agent activity', + ); + } + if ( + isClaudeCodeAgent && + provider === 'claude' && + shouldResetSessionOnAgentFailure(output) + ) { + resetSessionRequested = true; + } + if ( + output.newSessionId && + !resetSessionRequested && + shouldPersistSession + ) { + deps.persistSession(sessionFolder, output.newSessionId); + currentSessionId = output.newSessionId; + } - const wrappedOnOutput = onOutput - ? async (output: AgentOutput) => { - const outputPhase = output.phase ?? 'final'; - const outputText = getAgentOutputText(output); - const structuredOutput = getStructuredAgentOutput(output); - if (outputPhase !== 'final') { - log.info( - { - provider, - outputPhase, - outputStatus: output.status, - visibility: structuredOutput?.visibility ?? null, - preview: - typeof outputText === 'string' && outputText.length > 0 - ? outputText.slice(0, 160) - : null, - errorPreview: - typeof output.error === 'string' && output.error.length > 0 - ? output.error.slice(0, 160) - : null, - activeRole, - effectiveServiceId, - effectiveAgentType, - sessionFolder, - resumedSession: attemptSessionId ?? null, - streamedSessionId: output.newSessionId ?? null, - roomRoleServiceId: roomRoleContext?.serviceId ?? null, - roomRole: roomRoleContext?.role ?? null, - pairedTaskId: pairedExecutionContext?.task.id ?? null, - workspaceDir: - pairedExecutionContext?.workspace?.workspace_dir ?? - group.workDir ?? - null, - }, - 'Observed streamed agent activity', - ); - } - if ( - isClaudeCodeAgent && - provider === 'claude' && - shouldResetSessionOnAgentFailure(output) - ) { - resetSessionRequested = true; - } - if ( - output.newSessionId && - !resetSessionRequested && - shouldPersistSession - ) { - deps.persistSession(sessionFolder, output.newSessionId); - currentSessionId = output.newSessionId; - } - const evaluation = evaluateStreamedOutput(output, streamedState, { - agentType: isClaudeCodeAgent ? 'claude-code' : 'codex', - provider, - suppressClaudeAuthErrorOutput: provider === 'claude', - trackSuccessNullResult: true, - shortCircuitTriggeredErrors: - provider === 'claude' - ? canRetryClaudeCredentials - : getCodexAccountCount() > 1, - }); - streamedState = evaluation.state; + if (outputText && outputText.length > 0) { + pairedExecutionSummary = outputText.slice(0, 500); + pairedFullOutput = outputText; + } else if ( + typeof output.error === 'string' && + output.error.length > 0 + ) { + pairedExecutionSummary = output.error.slice(0, 500); + } + if (evaluation.newTrigger && outputText && output.status === 'success') { + log.warn( + { + reason: evaluation.newTrigger.reason, + resultPreview: outputText.slice(0, 120), + }, + 'Detected Claude rotation trigger in successful output', + ); + } else if ( + evaluation.newTrigger && + typeof output.error === 'string' + ) { + log.warn( + { + reason: evaluation.newTrigger.reason, + errorPreview: output.error.slice(0, 120), + }, + provider === 'claude' + ? 'Detected Claude rotation trigger in streamed error output' + : 'Detected Codex rotation trigger in streamed error output', + ); + } - if (typeof outputText === 'string' && outputText.length > 0) { - pairedExecutionSummary = outputText.slice(0, 500); - pairedFullOutput = outputText; - } else if ( - typeof output.error === 'string' && - output.error.length > 0 - ) { - pairedExecutionSummary = output.error.slice(0, 500); - } - if ( - evaluation.newTrigger && - typeof outputText === 'string' && - output.status === 'success' - ) { - log.warn( - { - reason: evaluation.newTrigger.reason, - resultPreview: outputText.slice(0, 120), - }, - 'Detected Claude rotation trigger in successful output', - ); - } else if ( - evaluation.newTrigger && - typeof output.error === 'string' - ) { - log.warn( - { - reason: evaluation.newTrigger.reason, - errorPreview: output.error.slice(0, 120), - }, - provider === 'claude' - ? 'Detected Claude rotation trigger in streamed error output' - : 'Detected Codex rotation trigger in streamed error output', - ); - } + if (evaluation.suppressedAuthError) { + log.warn( + { + resultPreview: outputText ? outputText.slice(0, 120) : undefined, + }, + 'Suppressed Claude 401 auth error from chat output', + ); + return; + } - if (evaluation.suppressedAuthError) { - log.warn( - { - resultPreview: - typeof outputText === 'string' - ? outputText.slice(0, 120) - : undefined, - }, - 'Suppressed Claude 401 auth error from chat output', - ); - return; - } + if (evaluation.suppressedRetryableSessionFailure) { + log.warn( + { + resultPreview: outputText + ? outputText.slice(0, 160) + : output.error?.slice(0, 160), + }, + 'Suppressed retryable Claude session failure from chat output', + ); + return; + } - if (evaluation.suppressedRetryableSessionFailure) { - log.warn( - { - resultPreview: - typeof outputText === 'string' - ? outputText.slice(0, 160) - : output.error?.slice(0, 160), - }, - 'Suppressed retryable Claude session failure from chat output', - ); - return; - } - - if (!evaluation.shouldForwardOutput) { - return; - } - if (typeof outputText === 'string' && outputText.length > 0) { - streamedState = { - ...evaluation.state, - sawVisibleOutput: true, - }; - } + if (!evaluation.shouldForwardOutput) { + return; + } + if (outputText && outputText.length > 0) { + streamedOutputHandler.markVisibleOutput(); + } + if (onOutput) { await onOutput(output); } - : undefined; + }, + }); + + const wrappedOnOutput = async (output: AgentOutput) => { + await streamedOutputHandler.handleOutput(output); + }; const providerLog = log.child({ provider, @@ -584,11 +581,12 @@ export async function runAgentForGroup( providerLog.info( { status: output.status, - sawOutput: streamedState.sawOutput, + sawOutput: streamedOutputHandler.getState().sawOutput, }, `Provider response completed (provider: ${provider})`, ); + const streamedState = streamedOutputHandler.getState(); return { output, sawOutput: streamedState.sawOutput, @@ -600,6 +598,7 @@ export async function runAgentForGroup( streamedTriggerReason: streamedState.streamedTriggerReason, }; } catch (error) { + const streamedState = streamedOutputHandler.getState(); return { error, sawOutput: streamedState.sawOutput, diff --git a/src/message-runtime-flow.ts b/src/message-runtime-flow.ts new file mode 100644 index 0000000..b808fc4 --- /dev/null +++ b/src/message-runtime-flow.ts @@ -0,0 +1,366 @@ +import { + getLastHumanMessageContent, + getPairedTurnOutputs, + getRecentChatMessages, +} from './db.js'; +import { logger } from './logger.js'; +import { + buildArbiterPromptForTask, + buildFinalizePendingPrompt, + buildOwnerPendingPrompt, + buildReviewerPendingPrompt, +} from './message-runtime-prompts.js'; +import { + advanceLastAgentCursor, + resolveCursorKey, + resolveNextTurnAction, +} from './message-runtime-rules.js'; +import { hasReviewerLease } from './service-routing.js'; +import type { + Channel, + PairedTask, + PairedRoomRole, + RegisteredGroup, +} from './types.js'; + +export type PendingPairedTurn = { + prompt: string; + channel: Channel | null; + cursor: string | number | null; + cursorKey?: string; + role?: 'reviewer' | 'arbiter'; +} | null; + +export type BotOnlyPairedFollowUpAction = + | { kind: 'none' } + | { + kind: 'inline-finalize'; + task: PairedTask; + cursor: string | number | null; + } + | { + kind: 'requeue-pending-turn'; + task: PairedTask; + cursor: string | number | null; + cursorKey: string; + nextRole: 'owner' | 'reviewer' | 'arbiter'; + }; + +export function buildPendingPairedTurn(args: { + chatJid: string; + timezone: string; + task: PairedTask; + rawMissedMessages: Array<{ seq?: number | null; timestamp?: string | null }>; + recentHumanMessages: Parameters[0]['recentHumanMessages']; + labeledRecentMessages: Parameters[0]['labeledRecentMessages']; + resolveChannel: (taskStatus?: string | null) => Channel | null; +}): PendingPairedTurn { + const { + chatJid, + timezone, + task, + rawMissedMessages, + recentHumanMessages, + labeledRecentMessages, + resolveChannel, + } = args; + const lastRaw = rawMissedMessages[rawMissedMessages.length - 1]; + const cursor = lastRaw?.seq ?? lastRaw?.timestamp ?? null; + const taskStatus = task.status; + const turnOutputs = getPairedTurnOutputs(task.id); + const lastTurnOutput = turnOutputs[turnOutputs.length - 1]; + const nextTurnAction = resolveNextTurnAction({ + taskStatus, + lastTurnOutputRole: lastTurnOutput?.role ?? null, + }); + const recentMessages = getRecentChatMessages(chatJid, 20); + const lastHumanMessage = getLastHumanMessageContent(chatJid); + + if (nextTurnAction.kind === 'reviewer-turn') { + return { + prompt: buildReviewerPendingPrompt({ + chatJid, + timezone, + turnOutputs, + recentHumanMessages, + lastHumanMessage, + }), + channel: resolveChannel(taskStatus), + cursor, + cursorKey: resolveCursorKey(chatJid, taskStatus), + role: 'reviewer', + }; + } + + if (nextTurnAction.kind === 'arbiter-turn') { + return { + prompt: buildArbiterPromptForTask({ + task, + chatJid, + timezone, + turnOutputs, + recentMessages, + labeledRecentMessages, + }), + channel: resolveChannel(taskStatus), + cursor, + cursorKey: resolveCursorKey(chatJid, taskStatus), + role: 'arbiter', + }; + } + + if (nextTurnAction.kind === 'finalize-owner-turn') { + return { + prompt: buildFinalizePendingPrompt({ turnOutputs }), + channel: resolveChannel(taskStatus), + cursor, + }; + } + + if (nextTurnAction.kind === 'owner-follow-up') { + return { + prompt: buildOwnerPendingPrompt({ + chatJid, + timezone, + turnOutputs, + recentHumanMessages, + lastHumanMessage, + }), + channel: resolveChannel(taskStatus), + cursor, + }; + } + + return null; +} + +export async function executePendingPairedTurn(args: { + pendingTurn: Exclude; + chatJid: string; + group: RegisteredGroup; + runId: string; + log: typeof logger; + saveState: () => void; + lastAgentTimestamps: Record; + executeTurn: (args: { + group: RegisteredGroup; + prompt: string; + chatJid: string; + runId: string; + channel: Channel; + startSeq: number | null; + endSeq: number | null; + deliveryRole?: PairedRoomRole; + }) => Promise<{ deliverySucceeded: boolean }>; + getFixedRoleChannelName: (role: 'reviewer' | 'arbiter') => string; +}): Promise { + const { + pendingTurn, + chatJid, + group, + runId, + log, + saveState, + lastAgentTimestamps, + executeTurn, + getFixedRoleChannelName, + } = args; + + if (!pendingTurn.channel) { + const missingRole = pendingTurn.role ?? 'reviewer'; + log.error( + { + role: missingRole, + requiredChannel: getFixedRoleChannelName(missingRole), + }, + 'Skipping paired turn because the dedicated Discord role channel is not configured', + ); + return false; + } + + if (pendingTurn.cursor != null) { + advanceLastAgentCursor( + lastAgentTimestamps, + saveState, + chatJid, + pendingTurn.cursor, + pendingTurn.cursorKey, + ); + } + + const { deliverySucceeded } = await executeTurn({ + group, + prompt: pendingTurn.prompt, + chatJid, + runId, + channel: pendingTurn.channel, + deliveryRole: pendingTurn.role, + startSeq: null, + endSeq: null, + }); + + return deliverySucceeded; +} + +export function resolveBotOnlyPairedFollowUpAction(args: { + chatJid: string; + task: PairedTask | null | undefined; + isBotOnlyPairedFollowUp: boolean; + pendingCursorSource: + | { seq?: number | null; timestamp?: string | null } + | undefined; +}): BotOnlyPairedFollowUpAction { + const { chatJid, task, isBotOnlyPairedFollowUp, pendingCursorSource } = args; + if (!task || !isBotOnlyPairedFollowUp) { + return { kind: 'none' }; + } + + const cursor = pendingCursorSource?.seq ?? pendingCursorSource?.timestamp ?? null; + const lastTurnOutput = getPairedTurnOutputs(task.id).at(-1); + const nextTurnAction = resolveNextTurnAction({ + taskStatus: task.status, + lastTurnOutputRole: lastTurnOutput?.role ?? null, + }); + + if (nextTurnAction.kind === 'finalize-owner-turn') { + return { + kind: 'inline-finalize', + task, + cursor, + }; + } + + if ( + nextTurnAction.kind === 'owner-follow-up' || + nextTurnAction.kind === 'reviewer-turn' || + nextTurnAction.kind === 'arbiter-turn' + ) { + return { + kind: 'requeue-pending-turn', + task, + cursor, + cursorKey: resolveCursorKey(chatJid, task.status), + nextRole: + nextTurnAction.kind === 'owner-follow-up' + ? 'owner' + : nextTurnAction.kind === 'arbiter-turn' + ? 'arbiter' + : 'reviewer', + }; + } + + return { kind: 'none' }; +} + +export async function executeBotOnlyPairedFollowUpAction(args: { + action: BotOnlyPairedFollowUpAction; + chatJid: string; + group: RegisteredGroup; + runId: string; + channel: Channel; + log: typeof logger; + saveState: () => void; + lastAgentTimestamps: Record; + executeTurn: (args: { + group: RegisteredGroup; + prompt: string; + chatJid: string; + runId: string; + channel: Channel; + startSeq: number | null; + endSeq: number | null; + }) => Promise<{ deliverySucceeded: boolean }>; + enqueueGroupMessageCheck: () => void; + closeStdin: () => void; +}): Promise { + const { + action, + chatJid, + group, + runId, + channel, + log, + saveState, + lastAgentTimestamps, + executeTurn, + enqueueGroupMessageCheck, + closeStdin, + } = args; + + if (action.kind === 'none') { + return false; + } + + if (action.cursor != null) { + advanceLastAgentCursor( + lastAgentTimestamps, + saveState, + chatJid, + action.cursor, + action.kind === 'requeue-pending-turn' ? action.cursorKey : undefined, + ); + } + + if (action.kind === 'inline-finalize') { + log.info( + { + chatJid, + group: group.name, + groupFolder: group.folder, + taskId: action.task.id, + taskStatus: action.task.status, + handoffMode: 'inline-finalize', + nextRole: 'owner', + cursor: action.cursor, + }, + 'Executing merge_ready finalize turn inline after bot-only reviewer follow-up', + ); + const { deliverySucceeded } = await executeTurn({ + group, + prompt: buildFinalizePendingPrompt({ + turnOutputs: getPairedTurnOutputs(action.task.id), + }), + chatJid, + runId, + channel, + startSeq: null, + endSeq: null, + }); + if (!deliverySucceeded) { + enqueueGroupMessageCheck(); + } + return true; + } + + closeStdin(); + enqueueGroupMessageCheck(); + log.info( + { + chatJid, + group: group.name, + groupFolder: group.folder, + taskId: action.task.id, + taskStatus: action.task.status, + handoffMode: 'requeue', + nextRole: action.nextRole, + cursor: action.cursor, + cursorKey: action.cursorKey, + }, + 'Queued fresh paired pending turn instead of piping bot-only follow-up into the active agent', + ); + return true; +} + +export function shouldSkipGenericFollowUpAfterDeliveryRetry(args: { + chatJid: string; + deliveryRole: PairedRoomRole; + pendingTask: PairedTask | null | undefined; +}): boolean { + return ( + hasReviewerLease(args.chatJid) && + args.deliveryRole === 'reviewer' && + resolveNextTurnAction({ + taskStatus: args.pendingTask?.status ?? null, + }).kind === 'finalize-owner-turn' + ); +} diff --git a/src/message-runtime.ts b/src/message-runtime.ts index ea9c19b..70b0d57 100644 --- a/src/message-runtime.ts +++ b/src/message-runtime.ts @@ -42,6 +42,13 @@ import { shouldSkipBotOnlyCollaboration, } from './message-runtime-rules.js'; import { runAgentForGroup } from './message-agent-executor.js'; +import { + buildPendingPairedTurn, + executeBotOnlyPairedFollowUpAction, + executePendingPairedTurn, + resolveBotOnlyPairedFollowUpAction, + shouldSkipGenericFollowUpAfterDeliveryRetry, +} from './message-runtime-flow.js'; import { MessageTurnController } from './message-turn-controller.js'; import { buildArbiterPromptForTask, @@ -208,157 +215,6 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { (message) => message.is_from_me === true || !!message.is_bot_message, ); - type BotOnlyPairedFollowUpAction = - | { kind: 'none' } - | { - kind: 'inline-finalize'; - task: PairedTask; - cursor: string | number | null; - } - | { - kind: 'requeue-pending-turn'; - task: PairedTask; - cursor: string | number | null; - cursorKey: string; - nextRole: 'owner' | 'reviewer' | 'arbiter'; - }; - - const resolveBotOnlyPairedFollowUpAction = (args: { - chatJid: string; - task: PairedTask | null | undefined; - isBotOnlyPairedFollowUp: boolean; - pendingCursorSource: NewMessage | undefined; - }): BotOnlyPairedFollowUpAction => { - const { chatJid, task, isBotOnlyPairedFollowUp, pendingCursorSource } = - args; - if (!task || !isBotOnlyPairedFollowUp) { - return { kind: 'none' }; - } - - const cursor = - pendingCursorSource?.seq ?? pendingCursorSource?.timestamp ?? null; - const lastTurnOutput = getPairedTurnOutputs(task.id).at(-1); - - const nextTurnAction = resolveNextTurnAction({ - taskStatus: task.status, - lastTurnOutputRole: lastTurnOutput?.role ?? null, - }); - - if (nextTurnAction.kind === 'finalize-owner-turn') { - return { - kind: 'inline-finalize', - task, - cursor, - }; - } - - if ( - nextTurnAction.kind === 'owner-follow-up' || - nextTurnAction.kind === 'reviewer-turn' || - nextTurnAction.kind === 'arbiter-turn' - ) { - return { - kind: 'requeue-pending-turn', - task, - cursor, - cursorKey: resolveCursorKey(chatJid, task.status), - nextRole: - nextTurnAction.kind === 'owner-follow-up' - ? 'owner' - : nextTurnAction.kind === 'arbiter-turn' - ? 'arbiter' - : 'reviewer', - }; - } - - return { kind: 'none' }; - }; - - const executeBotOnlyPairedFollowUpAction = async (args: { - action: BotOnlyPairedFollowUpAction; - chatJid: string; - group: RegisteredGroup; - runId: string; - channel: Channel; - log: typeof logger; - enqueueGroupMessageCheck: () => void; - }): Promise => { - const { - action, - chatJid, - group, - runId, - channel, - log, - enqueueGroupMessageCheck, - } = args; - - if (action.kind === 'none') { - return false; - } - - if (action.cursor != null) { - advanceLastAgentCursor( - deps.getLastAgentTimestamps(), - deps.saveState, - chatJid, - action.cursor, - action.kind === 'requeue-pending-turn' ? action.cursorKey : undefined, - ); - } - - if (action.kind === 'inline-finalize') { - log.info( - { - chatJid, - group: group.name, - groupFolder: group.folder, - taskId: action.task.id, - taskStatus: action.task.status, - handoffMode: 'inline-finalize', - nextRole: 'owner', - cursor: action.cursor, - }, - 'Executing merge_ready finalize turn inline after bot-only reviewer follow-up', - ); - const { deliverySucceeded } = await executeTurn({ - group, - prompt: buildFinalizePendingPrompt({ - turnOutputs: getPairedTurnOutputs(action.task.id), - }), - chatJid, - runId, - channel, - startSeq: null, - endSeq: null, - }); - if (!deliverySucceeded) { - enqueueGroupMessageCheck(); - } - return true; - } - - deps.queue.closeStdin(chatJid, { - reason: 'paired-pending-turn-follow-up', - }); - enqueueGroupMessageCheck(); - log.info( - { - chatJid, - group: group.name, - groupFolder: group.folder, - taskId: action.task.id, - taskStatus: action.task.status, - handoffMode: 'requeue', - nextRole: action.nextRole, - cursor: action.cursor, - cursorKey: action.cursorKey, - }, - 'Queued fresh paired pending turn instead of piping bot-only follow-up into the active agent', - ); - return true; - }; - const enqueueScopedGroupMessageCheck = ( chatJid: string, groupFolder: string, @@ -832,152 +688,22 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { return role === 'owner' ? channel : roleToChannel[role]; }; - const buildPendingPairedTurn = ( - task: PairedTask, - rawMissedMessages: NewMessage[], - ): { - prompt: string; - channel: Channel | null; - cursor: string | number | null; - cursorKey?: string; - role?: 'reviewer' | 'arbiter'; - } | null => { - const lastRaw = rawMissedMessages[rawMissedMessages.length - 1]; - const cursor = lastRaw?.seq ?? lastRaw?.timestamp ?? null; - const taskStatus = task.status; - const turnOutputs = getPairedTurnOutputs(task.id); - const lastTurnOutput = turnOutputs[turnOutputs.length - 1]; - const nextTurnAction = resolveNextTurnAction({ - taskStatus, - lastTurnOutputRole: lastTurnOutput?.role ?? null, - }); - const recentMessages = getRecentChatMessages(chatJid, 20); - const recentHumanMessages = recentMessages.filter( - (message) => !message.is_bot_message, - ); - const lastHumanMessage = getLastHumanMessageContent(chatJid); - - if (nextTurnAction.kind === 'reviewer-turn') { - return { - prompt: buildReviewerPendingPrompt({ - chatJid, - timezone: deps.timezone, - turnOutputs, - recentHumanMessages, - lastHumanMessage, - }), - channel: resolveChannel(taskStatus), - cursor, - cursorKey: resolveCursorKey(chatJid, taskStatus), - role: 'reviewer', - }; - } - - if (nextTurnAction.kind === 'arbiter-turn') { - return { - prompt: buildArbiterPromptForTask({ - task, - chatJid, - timezone: deps.timezone, - turnOutputs, - recentMessages, - labeledRecentMessages: labelPairedSenders(chatJid, recentMessages), - }), - channel: resolveChannel(taskStatus), - cursor, - cursorKey: resolveCursorKey(chatJid, taskStatus), - role: 'arbiter', - }; - } - - if (nextTurnAction.kind === 'finalize-owner-turn') { - return { - prompt: buildFinalizePendingPrompt({ turnOutputs }), - channel: resolveChannel(taskStatus), - cursor, - }; - } - - if (nextTurnAction.kind === 'owner-follow-up') { - return { - prompt: buildOwnerPendingPrompt({ - chatJid, - timezone: deps.timezone, - turnOutputs, - recentHumanMessages, - lastHumanMessage, - }), - channel: resolveChannel(taskStatus), - cursor, - }; - } - - return null; - }; - - const executePendingPairedTurn = async (args: { - prompt: string; - channel: Channel | null; - cursor: string | number | null; - cursorKey?: string; - role?: 'reviewer' | 'arbiter'; - }): Promise => { - if (!args.channel) { - const missingRole = args.role ?? 'reviewer'; - log.error( - { - role: missingRole, - requiredChannel: getFixedRoleChannelName(missingRole), - }, - 'Skipping paired turn because the dedicated Discord role channel is not configured', - ); - return false; - } - - if (args.cursor != null) { - advanceLastAgentCursor( - deps.getLastAgentTimestamps(), - deps.saveState, - chatJid, - args.cursor, - args.cursorKey, - ); - } - - const { deliverySucceeded } = await executeTurn({ - group, - prompt: args.prompt, - chatJid, - runId, - channel: args.channel, - deliveryRole: args.role, - startSeq: null, - endSeq: null, - }); - - return deliverySucceeded; - }; - const enqueueGroupMessageCheck = (): void => { enqueueScopedGroupMessageCheck(chatJid, group.folder); }; - const shouldSkipGenericFollowUpAfterDeliveryRetry = (args: { - deliveryRole: PairedRoomRole; - pendingTask: PairedTask | null | undefined; - }): boolean => - hasReviewerLease(chatJid) && - args.deliveryRole === 'reviewer' && - resolveNextTurnAction({ - taskStatus: args.pendingTask?.status ?? null, - }).kind === 'finalize-owner-turn'; - const enqueueGenericFollowUpAfterDeliveryRetry = (args: { deliveryRole: PairedRoomRole; pendingTask: PairedTask | null | undefined; workItemId: string | number; }): void => { - if (shouldSkipGenericFollowUpAfterDeliveryRetry(args)) { + if ( + shouldSkipGenericFollowUpAfterDeliveryRetry({ + chatJid, + deliveryRole: args.deliveryRole, + pendingTask: args.pendingTask, + }) + ) { log.info( { workItemId: args.workItemId, @@ -1074,10 +800,33 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { ? getLatestOpenPairedTaskForChat(chatJid) : null; const pendingTurn = pendingReviewTask - ? buildPendingPairedTurn(pendingReviewTask, rawMissedMessages) + ? buildPendingPairedTurn({ + chatJid, + timezone: deps.timezone, + task: pendingReviewTask, + rawMissedMessages, + recentHumanMessages: getRecentChatMessages(chatJid, 20).filter( + (message) => !message.is_bot_message, + ), + labeledRecentMessages: labelPairedSenders( + chatJid, + getRecentChatMessages(chatJid, 20), + ), + resolveChannel, + }) : null; if (pendingTurn) { - return executePendingPairedTurn(pendingTurn); + return executePendingPairedTurn({ + pendingTurn, + chatJid, + group, + runId, + log, + saveState: deps.saveState, + lastAgentTimestamps: deps.getLastAgentTimestamps(), + executeTurn, + getFixedRoleChannelName, + }); } const lastIgnored = rawMissedMessages[rawMissedMessages.length - 1]; @@ -1097,10 +846,33 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { : null; const botOnlyPendingTurn = botOnlyPendingTask && isBotOnlyPairedRoomTurn(chatJid, missedMessages) - ? buildPendingPairedTurn(botOnlyPendingTask, rawMissedMessages) + ? buildPendingPairedTurn({ + chatJid, + timezone: deps.timezone, + task: botOnlyPendingTask, + rawMissedMessages, + recentHumanMessages: getRecentChatMessages(chatJid, 20).filter( + (message) => !message.is_bot_message, + ), + labeledRecentMessages: labelPairedSenders( + chatJid, + getRecentChatMessages(chatJid, 20), + ), + resolveChannel, + }) : null; if (botOnlyPendingTurn) { - return executePendingPairedTurn(botOnlyPendingTurn); + return executePendingPairedTurn({ + pendingTurn: botOnlyPendingTurn, + chatJid, + group, + runId, + log, + saveState: deps.saveState, + lastAgentTimestamps: deps.getLastAgentTimestamps(), + executeTurn, + getFixedRoleChannelName, + }); } if (shouldSkipBotOnlyCollaboration(chatJid, missedMessages)) { @@ -1483,8 +1255,15 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { runId: `loop-merge-ready-${Date.now().toString(36)}`, channel, log: logger, + saveState: deps.saveState, + lastAgentTimestamps: deps.getLastAgentTimestamps(), + executeTurn, enqueueGroupMessageCheck: () => enqueueScopedGroupMessageCheck(chatJid, group.folder), + closeStdin: () => + deps.queue.closeStdin(chatJid, { + reason: 'paired-pending-turn-follow-up', + }), }) ) { continue; diff --git a/src/paired-execution-context-arbiter.ts b/src/paired-execution-context-arbiter.ts new file mode 100644 index 0000000..de46e5b --- /dev/null +++ b/src/paired-execution-context-arbiter.ts @@ -0,0 +1,81 @@ +import { ARBITER_DEADLOCK_THRESHOLD } from './config.js'; +import { updatePairedTask } from './db.js'; +import { logger } from './logger.js'; +import { classifyArbiterVerdict } from './paired-execution-context-shared.js'; +import type { PairedTask } from './types.js'; + +export function handleFailedArbiterExecution(args: { + task: PairedTask; + taskId: string; +}): void { + const { task, taskId } = args; + const now = new Date().toISOString(); + const fallbackStatus = + task.status === 'in_arbitration' || task.status === 'arbiter_requested' + ? 'arbiter_requested' + : task.status; + if (fallbackStatus !== task.status) { + updatePairedTask(taskId, { status: fallbackStatus, updated_at: now }); + logger.warn( + { + taskId, + role: 'arbiter', + previousStatus: task.status, + nextStatus: fallbackStatus, + }, + 'Preserved arbiter task in arbitration-requested state after failed execution', + ); + } +} + +export function handleArbiterCompletion(args: { + taskId: string; + summary?: string | null; +}): void { + const { taskId, summary } = args; + const now = new Date().toISOString(); + const arbiterVerdict = classifyArbiterVerdict(summary); + + logger.info( + { taskId, arbiterVerdict, summary: summary?.slice(0, 200) }, + 'Arbiter verdict rendered', + ); + + switch (arbiterVerdict) { + case 'proceed': + case 'revise': + case 'reset': + updatePairedTask(taskId, { + status: 'active', + round_trip_count: Math.max(0, ARBITER_DEADLOCK_THRESHOLD - 1), + arbiter_verdict: arbiterVerdict, + updated_at: now, + }); + logger.info( + { taskId, arbiterVerdict }, + 'Arbiter resolved deadlock — resuming ping-pong', + ); + return; + case 'escalate': + updatePairedTask(taskId, { + status: 'completed', + arbiter_verdict: 'escalate', + completion_reason: 'arbiter_escalated', + updated_at: now, + }); + logger.info({ taskId }, 'Arbiter escalated to user — task completed'); + return; + default: + updatePairedTask(taskId, { + status: 'active', + round_trip_count: Math.max(0, ARBITER_DEADLOCK_THRESHOLD - 1), + arbiter_verdict: 'unknown', + updated_at: now, + }); + logger.warn( + { taskId, summary: summary?.slice(0, 200) }, + 'Arbiter verdict unrecognized — falling back to proceed', + ); + return; + } +} diff --git a/src/paired-execution-context-owner.ts b/src/paired-execution-context-owner.ts new file mode 100644 index 0000000..186b3f4 --- /dev/null +++ b/src/paired-execution-context-owner.ts @@ -0,0 +1,201 @@ +import { + ARBITER_DEADLOCK_THRESHOLD, + PAIRED_MAX_ROUND_TRIPS, +} from './config.js'; +import { + getPairedWorkspace, + hasActiveCiWatcherForChat, + updatePairedTask, +} from './db.js'; +import { logger } from './logger.js'; +import { markPairedTaskReviewReady } from './paired-workspace-manager.js'; +import { + classifyVerdict, + hasCodeChangesSinceRef, + requestArbiterOrEscalate, + resolveCanonicalSourceRef, +} from './paired-execution-context-shared.js'; +import type { PairedTask } from './types.js'; + +export function handleFailedOwnerExecution(args: { + task: PairedTask; + taskId: string; +}): void { + const { task, taskId } = args; + if (task.status !== 'active') { + const now = new Date().toISOString(); + updatePairedTask(taskId, { status: 'active', updated_at: now }); + logger.info( + { taskId, role: 'owner', previousStatus: task.status }, + 'Reset task to active after failed execution', + ); + } +} + +function handleOwnerFinalizeCompletion(args: { + task: PairedTask; + taskId: string; + summary?: string | null; + now: string; +}): boolean { + const { task, taskId, summary, now } = args; + const ownerVerdict = classifyVerdict(summary); + + if (ownerVerdict === 'blocked' || ownerVerdict === 'needs_context') { + requestArbiterOrEscalate({ + taskId, + now, + arbiterLogMessage: 'Owner blocked during finalize — requesting arbiter', + escalateLogMessage: 'Owner blocked during finalize — escalating to user', + logContext: { + taskId, + ownerVerdict, + summary: summary?.slice(0, 100), + }, + }); + return true; + } + + if (ownerVerdict === 'done_with_concerns') { + if (task.round_trip_count >= ARBITER_DEADLOCK_THRESHOLD) { + requestArbiterOrEscalate({ + taskId, + now, + arbiterLogMessage: 'Owner finalize loop detected — requesting arbiter', + escalateLogMessage: 'Owner finalize loop detected — escalating to user', + logContext: { + taskId, + ownerVerdict, + roundTrips: task.round_trip_count, + }, + }); + return true; + } + updatePairedTask(taskId, { status: 'active', updated_at: now }); + logger.info( + { + taskId, + ownerVerdict, + summary: summary?.slice(0, 100), + }, + 'Owner raised concerns during finalize — task set back to active', + ); + return false; + } + + const workspace = getPairedWorkspace(task.id, 'owner'); + const hasNewChanges = workspace?.workspace_dir + ? hasCodeChangesSinceRef(workspace.workspace_dir, task.source_ref) + : null; + + if (hasNewChanges === true) { + if (task.round_trip_count >= ARBITER_DEADLOCK_THRESHOLD) { + requestArbiterOrEscalate({ + taskId, + now, + arbiterLogMessage: + 'Owner finalize DONE loop detected — requesting arbiter', + escalateLogMessage: + 'Owner finalize DONE loop detected — escalating to user', + logContext: { + taskId, + roundTrips: task.round_trip_count, + hasNewChanges, + }, + }); + return true; + } + logger.info( + { + taskId, + sourceRef: task.source_ref, + hasNewChanges, + }, + 'Owner made changes after reviewer approval — re-triggering review', + ); + return false; + } + + updatePairedTask(taskId, { + status: 'completed', + completion_reason: 'done', + updated_at: now, + }); + logger.info( + { taskId, hasNewChanges, summary: summary?.slice(0, 100) }, + 'Owner finalized after reviewer approval — task completed', + ); + return true; +} + +export function handleOwnerCompletion(args: { + task: PairedTask; + taskId: string; + summary?: string | null; +}): void { + const { task, taskId, summary } = args; + const now = new Date().toISOString(); + + if (task.status === 'merge_ready') { + const shouldStop = handleOwnerFinalizeCompletion({ + task, + taskId, + summary, + now, + }); + if (shouldStop) { + return; + } + } + + if (hasActiveCiWatcherForChat(task.chat_jid)) { + logger.info( + { taskId, chatJid: task.chat_jid }, + 'Active CI watcher found, deferring auto-review until watcher completes', + ); + return; + } + + if (task.status !== 'merge_ready') { + const ownerVerdict = classifyVerdict(summary); + if (ownerVerdict === 'blocked' || ownerVerdict === 'needs_context') { + requestArbiterOrEscalate({ + taskId, + now, + arbiterLogMessage: 'Owner blocked/needs_context — requesting arbiter', + escalateLogMessage: 'Owner blocked/needs_context — escalating to user', + logContext: { + taskId, + ownerVerdict, + summary: summary?.slice(0, 100), + }, + }); + return; + } + } + + if (task.round_trip_count >= PAIRED_MAX_ROUND_TRIPS) { + logger.info( + { + taskId, + roundTrips: task.round_trip_count, + max: PAIRED_MAX_ROUND_TRIPS, + }, + 'Round trip limit reached, skipping auto-review', + ); + return; + } + + const result = markPairedTaskReviewReady(taskId); + if (result) { + updatePairedTask(taskId, { + round_trip_count: task.round_trip_count + 1, + review_requested_at: now, + updated_at: now, + }); + logger.info( + { taskId, roundTrip: task.round_trip_count + 1 }, + 'Auto-triggered reviewer after owner completion', + ); + } +} diff --git a/src/paired-execution-context-reviewer.ts b/src/paired-execution-context-reviewer.ts new file mode 100644 index 0000000..2ebfe19 --- /dev/null +++ b/src/paired-execution-context-reviewer.ts @@ -0,0 +1,139 @@ +import { ARBITER_DEADLOCK_THRESHOLD } from './config.js'; +import { getPairedWorkspace, updatePairedTask } from './db.js'; +import { logger } from './logger.js'; +import { + classifyVerdict, + requestArbiterOrEscalate, + resolveCanonicalSourceRef, +} from './paired-execution-context-shared.js'; +import type { PairedTask } from './types.js'; + +export function handleFailedReviewerExecution(args: { + task: PairedTask; + taskId: string; + summary?: string | null; +}): void { + const { task, taskId, summary } = args; + const now = new Date().toISOString(); + + if (summary) { + const verdict = classifyVerdict(summary); + if ( + verdict === 'done' || + verdict === 'blocked' || + verdict === 'needs_context' + ) { + const ownerWs = + verdict === 'done' ? getPairedWorkspace(taskId, 'owner') : null; + const approvedSourceRef = + verdict === 'done' && ownerWs?.workspace_dir + ? resolveCanonicalSourceRef(ownerWs.workspace_dir) + : task.source_ref; + updatePairedTask(taskId, { + status: verdict === 'done' ? 'merge_ready' : 'completed', + ...(verdict === 'done' ? { source_ref: approvedSourceRef } : {}), + ...(verdict !== 'done' ? { completion_reason: 'escalated' } : {}), + updated_at: now, + }); + logger.info( + { + taskId, + verdict, + approvedSourceRef, + summary: summary.slice(0, 100), + }, + 'Reviewer verdict detected from failed execution — stopping ping-pong', + ); + return; + } + } + + const fallbackStatus = + task.status === 'in_review' || task.status === 'review_ready' + ? 'review_ready' + : task.status; + if (fallbackStatus !== task.status) { + updatePairedTask(taskId, { status: fallbackStatus, updated_at: now }); + logger.warn( + { + taskId, + role: 'reviewer', + previousStatus: task.status, + nextStatus: fallbackStatus, + }, + 'Preserved reviewer task in review-ready state after failed execution', + ); + } +} + +export function handleReviewerCompletion(args: { + task: PairedTask; + taskId: string; + summary?: string | null; +}): void { + const { task, taskId, summary } = args; + const now = new Date().toISOString(); + const verdict = classifyVerdict(summary); + + switch (verdict) { + case 'done': { + const ownerWs = getPairedWorkspace(taskId, 'owner'); + const approvedSourceRef = ownerWs?.workspace_dir + ? resolveCanonicalSourceRef(ownerWs.workspace_dir) + : task.source_ref; + updatePairedTask(taskId, { + status: 'merge_ready', + source_ref: approvedSourceRef, + updated_at: now, + }); + logger.info( + { + taskId, + verdict, + approvedSourceRef, + summary: summary?.slice(0, 100), + }, + 'Reviewer approved — owner gets final turn to finalize', + ); + return; + } + + case 'blocked': + case 'needs_context': + requestArbiterOrEscalate({ + taskId, + now, + arbiterLogMessage: + 'Reviewer blocked/needs_context — requesting arbiter before escalating', + escalateLogMessage: 'Reviewer escalated to user — ping-pong stopped', + logContext: { taskId, verdict, summary: summary?.slice(0, 100) }, + }); + return; + + case 'done_with_concerns': + case 'continue': + default: + if (task.round_trip_count >= ARBITER_DEADLOCK_THRESHOLD) { + requestArbiterOrEscalate({ + taskId, + now, + arbiterLogMessage: + 'Deadlock detected — requesting arbiter intervention', + escalateLogMessage: + 'Stopped ping-pong — escalating to user (arbiter not configured)', + logContext: { + taskId, + verdict, + roundTrips: task.round_trip_count, + }, + }); + return; + } + updatePairedTask(taskId, { status: 'active', updated_at: now }); + logger.info( + { taskId, verdict }, + 'Reviewer has feedback, task set back to active for owner', + ); + return; + } +} diff --git a/src/paired-execution-context-shared.ts b/src/paired-execution-context-shared.ts new file mode 100644 index 0000000..349e2c1 --- /dev/null +++ b/src/paired-execution-context-shared.ts @@ -0,0 +1,122 @@ +import { execFileSync } from 'child_process'; + +import { isArbiterEnabled } from './config.js'; +import { updatePairedTask } from './db.js'; +import { logger } from './logger.js'; + +export type Verdict = + | 'done' + | 'done_with_concerns' + | 'blocked' + | 'needs_context' + | 'continue'; + +export function classifyVerdict(summary: string | null | undefined): Verdict { + if (!summary) return 'continue'; + const cleaned = summary.replace(/[\s\S]*?<\/internal>/g, '').trim(); + if (!cleaned) return 'continue'; + const firstLine = cleaned.split('\n')[0].trim(); + if (/^\*{0,2}BLOCKED\*{0,2}\b/i.test(firstLine)) return 'blocked'; + if (/^\*{0,2}NEEDS_CONTEXT\*{0,2}\b/i.test(firstLine)) + return 'needs_context'; + if (/^\*{0,2}DONE_WITH_CONCERNS\*{0,2}\b/i.test(firstLine)) + return 'done_with_concerns'; + if (/^\*{0,2}DONE\*{0,2}\b/i.test(firstLine)) return 'done'; + if (/^\*{0,2}Approved\.?\*{0,2}/i.test(firstLine)) return 'done'; + if (/^\*{0,2}LGTM\*{0,2}/i.test(firstLine)) return 'done'; + return 'continue'; +} + +export type ArbiterVerdictResult = + | 'proceed' + | 'revise' + | 'reset' + | 'escalate' + | 'unknown'; + +export function classifyArbiterVerdict( + summary: string | null | undefined, +): ArbiterVerdictResult { + if (!summary) return 'unknown'; + const cleaned = summary.replace(/[\s\S]*?<\/internal>/g, '').trim(); + if (!cleaned) return 'unknown'; + const firstLine = cleaned.split('\n')[0].trim(); + const verdictMatch = firstLine.match( + /\*{0,2}(?:VERDICT\s*[:—-]\s*)?(PROCEED|REVISE|RESET|ESCALATE)\*{0,2}/i, + ); + if (verdictMatch) { + return verdictMatch[1].toLowerCase() as ArbiterVerdictResult; + } + return 'unknown'; +} + +export function resolveCanonicalSourceRef(workDir: string): string { + const treeHash = resolveCanonicalTreeHash(workDir); + return treeHash || 'HEAD'; +} + +function resolveCanonicalTreeHash(workDir: string): string | null { + try { + const treeHash = execFileSync('git', ['rev-parse', 'HEAD^{tree}'], { + cwd: workDir, + encoding: 'utf-8', + stdio: ['ignore', 'pipe', 'pipe'], + }).trim(); + return treeHash || null; + } catch { + return null; + } +} + +export function hasCodeChangesSinceRef( + workDir: string, + sourceRef: string | null | undefined, +): boolean | null { + if (!sourceRef) return null; + try { + execFileSync('git', ['diff', '--quiet', sourceRef, 'HEAD'], { + cwd: workDir, + stdio: ['ignore', 'pipe', 'pipe'], + }); + return false; + } catch (error) { + const exitCode = + typeof error === 'object' && + error !== null && + 'status' in error && + typeof (error as { status?: unknown }).status === 'number' + ? (error as { status: number }).status + : null; + if (exitCode === 1) { + return true; + } + return null; + } +} + +export function requestArbiterOrEscalate(args: { + taskId: string; + now: string; + arbiterLogMessage: string; + escalateLogMessage: string; + logContext?: Record; +}): void { + const { taskId, now, arbiterLogMessage, escalateLogMessage, logContext } = + args; + if (isArbiterEnabled()) { + updatePairedTask(taskId, { + status: 'arbiter_requested', + arbiter_requested_at: now, + updated_at: now, + }); + logger.info(logContext ?? { taskId }, arbiterLogMessage); + return; + } + + updatePairedTask(taskId, { + status: 'completed', + completion_reason: 'escalated', + updated_at: now, + }); + logger.info(logContext ?? { taskId }, escalateLogMessage); +} diff --git a/src/paired-execution-context.ts b/src/paired-execution-context.ts index 639b499..38d8094 100644 --- a/src/paired-execution-context.ts +++ b/src/paired-execution-context.ts @@ -1,4 +1,3 @@ -import { execFileSync } from 'child_process'; import crypto from 'crypto'; import fs from 'fs'; import path from 'path'; @@ -12,7 +11,6 @@ import { DATA_DIR, PAIRED_MAX_ROUND_TRIPS, REVIEWER_AGENT_TYPE, - isArbiterEnabled, } from './config.js'; import { createPairedTask, @@ -25,6 +23,22 @@ import { upsertPairedProject, } from './db.js'; import { logger } from './logger.js'; +import { + handleArbiterCompletion, + handleFailedArbiterExecution, +} from './paired-execution-context-arbiter.js'; +import { + handleFailedOwnerExecution, + handleOwnerCompletion, +} from './paired-execution-context-owner.js'; +import { + handleFailedReviewerExecution, + handleReviewerCompletion, +} from './paired-execution-context-reviewer.js'; +import { + resolveCanonicalSourceRef, + requestArbiterOrEscalate, +} from './paired-execution-context-shared.js'; import { markPairedTaskReviewReady, prepareReviewerWorkspaceForExecution, @@ -40,137 +54,6 @@ import type { } from './types.js'; import { resolveRoleAgentPlan } from './role-agent-plan.js'; -// --------------------------------------------------------------------------- -// Reviewer verdict detection -// --------------------------------------------------------------------------- - -type Verdict = - | 'done' - | 'done_with_concerns' - | 'blocked' - | 'needs_context' - | 'continue'; - -function classifyVerdict(summary: string | null | undefined): Verdict { - if (!summary) return 'continue'; - // Strip ... tags — these are internal reasoning, not the verdict - const cleaned = summary.replace(/[\s\S]*?<\/internal>/g, '').trim(); - if (!cleaned) return 'continue'; - const firstLine = cleaned.split('\n')[0].trim(); - // Match verdict markers at the start of the first visible line - if (/^\*{0,2}BLOCKED\*{0,2}\b/i.test(firstLine)) return 'blocked'; - if (/^\*{0,2}NEEDS_CONTEXT\*{0,2}\b/i.test(firstLine)) return 'needs_context'; - if (/^\*{0,2}DONE_WITH_CONCERNS\*{0,2}\b/i.test(firstLine)) - return 'done_with_concerns'; - if (/^\*{0,2}DONE\*{0,2}\b/i.test(firstLine)) return 'done'; - if (/^\*{0,2}Approved\.?\*{0,2}/i.test(firstLine)) return 'done'; - if (/^\*{0,2}LGTM\*{0,2}/i.test(firstLine)) return 'done'; - return 'continue'; -} - -// --------------------------------------------------------------------------- -// Arbiter verdict detection -// --------------------------------------------------------------------------- - -type ArbiterVerdictResult = - | 'proceed' - | 'revise' - | 'reset' - | 'escalate' - | 'unknown'; - -function classifyArbiterVerdict( - summary: string | null | undefined, -): ArbiterVerdictResult { - if (!summary) return 'unknown'; - const cleaned = summary.replace(/[\s\S]*?<\/internal>/g, '').trim(); - if (!cleaned) return 'unknown'; - const firstLine = cleaned.split('\n')[0].trim(); - // Match verdict keywords with optional prefix like "VERDICT:" and markdown bold - const verdictMatch = firstLine.match( - /\*{0,2}(?:VERDICT\s*[:—-]\s*)?(PROCEED|REVISE|RESET|ESCALATE)\*{0,2}/i, - ); - if (verdictMatch) { - return verdictMatch[1].toLowerCase() as ArbiterVerdictResult; - } - return 'unknown'; -} - -// --------------------------------------------------------------------------- -// Helpers -// --------------------------------------------------------------------------- - -function resolveCanonicalSourceRef(workDir: string): string { - const treeHash = resolveCanonicalTreeHash(workDir); - return treeHash || 'HEAD'; -} - -function resolveCanonicalTreeHash(workDir: string): string | null { - try { - const treeHash = execFileSync('git', ['rev-parse', 'HEAD^{tree}'], { - cwd: workDir, - encoding: 'utf-8', - stdio: ['ignore', 'pipe', 'pipe'], - }).trim(); - return treeHash || null; - } catch { - return null; - } -} - -function hasCodeChangesSinceRef( - workDir: string, - sourceRef: string | null | undefined, -): boolean | null { - if (!sourceRef) return null; - try { - execFileSync('git', ['diff', '--quiet', sourceRef, 'HEAD'], { - cwd: workDir, - stdio: ['ignore', 'pipe', 'pipe'], - }); - return false; - } catch (error) { - const exitCode = - typeof error === 'object' && - error !== null && - 'status' in error && - typeof (error as { status?: unknown }).status === 'number' - ? (error as { status: number }).status - : null; - if (exitCode === 1) { - return true; - } - return null; - } -} - -function requestArbiterOrEscalate(args: { - taskId: string; - now: string; - arbiterLogMessage: string; - escalateLogMessage: string; - logContext?: Record; -}): void { - const { taskId, now, arbiterLogMessage, escalateLogMessage, logContext } = - args; - if (isArbiterEnabled()) { - updatePairedTask(taskId, { - status: 'arbiter_requested', - arbiter_requested_at: now, - updated_at: now, - }); - logger.info(logContext ?? { taskId }, arbiterLogMessage); - return; - } - - updatePairedTask(taskId, { - status: 'completed', - completion_reason: 'escalated', - updated_at: now, - }); - logger.info(logContext ?? { taskId }, escalateLogMessage); -} - function ensurePairedProject( group: RegisteredGroup, chatJid: string, @@ -444,399 +327,6 @@ export function preparePairedExecutionContext(args: { }; } -// --------------------------------------------------------------------------- -// completePairedExecutionContext -// --------------------------------------------------------------------------- - -function handleFailedReviewerExecution(args: { - task: PairedTask; - taskId: string; - summary?: string | null; -}): void { - const { task, taskId, summary } = args; - const now = new Date().toISOString(); - - if (summary) { - const verdict = classifyVerdict(summary); - if ( - verdict === 'done' || - verdict === 'blocked' || - verdict === 'needs_context' - ) { - const ownerWs = - verdict === 'done' ? getPairedWorkspace(taskId, 'owner') : null; - const approvedSourceRef = - verdict === 'done' && ownerWs?.workspace_dir - ? resolveCanonicalSourceRef(ownerWs.workspace_dir) - : task.source_ref; - updatePairedTask(taskId, { - status: verdict === 'done' ? 'merge_ready' : 'completed', - ...(verdict === 'done' ? { source_ref: approvedSourceRef } : {}), - ...(verdict !== 'done' ? { completion_reason: 'escalated' } : {}), - updated_at: now, - }); - logger.info( - { - taskId, - verdict, - approvedSourceRef, - summary: summary.slice(0, 100), - }, - 'Reviewer verdict detected from failed execution — stopping ping-pong', - ); - return; - } - } - - const fallbackStatus = - task.status === 'in_review' || task.status === 'review_ready' - ? 'review_ready' - : task.status; - if (fallbackStatus !== task.status) { - updatePairedTask(taskId, { status: fallbackStatus, updated_at: now }); - logger.warn( - { - taskId, - role: 'reviewer', - previousStatus: task.status, - nextStatus: fallbackStatus, - }, - 'Preserved reviewer task in review-ready state after failed execution', - ); - } -} - -function handleFailedArbiterExecution(args: { - task: PairedTask; - taskId: string; -}): void { - const { task, taskId } = args; - const now = new Date().toISOString(); - const fallbackStatus = - task.status === 'in_arbitration' || task.status === 'arbiter_requested' - ? 'arbiter_requested' - : task.status; - if (fallbackStatus !== task.status) { - updatePairedTask(taskId, { status: fallbackStatus, updated_at: now }); - logger.warn( - { - taskId, - role: 'arbiter', - previousStatus: task.status, - nextStatus: fallbackStatus, - }, - 'Preserved arbiter task in arbitration-requested state after failed execution', - ); - } -} - -function handleFailedOwnerExecution(args: { - task: PairedTask; - taskId: string; -}): void { - const { task, taskId } = args; - if (task.status !== 'active') { - const now = new Date().toISOString(); - updatePairedTask(taskId, { status: 'active', updated_at: now }); - logger.info( - { taskId, role: 'owner', previousStatus: task.status }, - 'Reset task to active after failed execution', - ); - } -} - -function handleOwnerFinalizeCompletion(args: { - task: PairedTask; - taskId: string; - summary?: string | null; - now: string; -}): boolean { - const { task, taskId, summary, now } = args; - const ownerVerdict = classifyVerdict(summary); - - if (ownerVerdict === 'blocked' || ownerVerdict === 'needs_context') { - requestArbiterOrEscalate({ - taskId, - now, - arbiterLogMessage: 'Owner blocked during finalize — requesting arbiter', - escalateLogMessage: 'Owner blocked during finalize — escalating to user', - logContext: { - taskId, - ownerVerdict, - summary: summary?.slice(0, 100), - }, - }); - return true; - } - - if (ownerVerdict === 'done_with_concerns') { - if (task.round_trip_count >= ARBITER_DEADLOCK_THRESHOLD) { - requestArbiterOrEscalate({ - taskId, - now, - arbiterLogMessage: 'Owner finalize loop detected — requesting arbiter', - escalateLogMessage: 'Owner finalize loop detected — escalating to user', - logContext: { - taskId, - ownerVerdict, - roundTrips: task.round_trip_count, - }, - }); - return true; - } - updatePairedTask(taskId, { status: 'active', updated_at: now }); - logger.info( - { - taskId, - ownerVerdict, - summary: summary?.slice(0, 100), - }, - 'Owner raised concerns during finalize — task set back to active', - ); - return false; - } - - const workspace = getPairedWorkspace(task.id, 'owner'); - const hasNewChanges = workspace?.workspace_dir - ? hasCodeChangesSinceRef(workspace.workspace_dir, task.source_ref) - : null; - - if (hasNewChanges === true) { - if (task.round_trip_count >= ARBITER_DEADLOCK_THRESHOLD) { - requestArbiterOrEscalate({ - taskId, - now, - arbiterLogMessage: - 'Owner finalize DONE loop detected — requesting arbiter', - escalateLogMessage: - 'Owner finalize DONE loop detected — escalating to user', - logContext: { - taskId, - roundTrips: task.round_trip_count, - hasNewChanges, - }, - }); - return true; - } - logger.info( - { - taskId, - sourceRef: task.source_ref, - hasNewChanges, - }, - 'Owner made changes after reviewer approval — re-triggering review', - ); - return false; - } - - updatePairedTask(taskId, { - status: 'completed', - completion_reason: 'done', - updated_at: now, - }); - logger.info( - { taskId, hasNewChanges, summary: summary?.slice(0, 100) }, - 'Owner finalized after reviewer approval — task completed', - ); - return true; -} - -function handleOwnerCompletion(args: { - task: PairedTask; - taskId: string; - summary?: string | null; -}): void { - const { task, taskId, summary } = args; - const now = new Date().toISOString(); - - if (task.status === 'merge_ready') { - const shouldStop = handleOwnerFinalizeCompletion({ - task, - taskId, - summary, - now, - }); - if (shouldStop) { - return; - } - } - - if (hasActiveCiWatcherForChat(task.chat_jid)) { - logger.info( - { taskId, chatJid: task.chat_jid }, - 'Active CI watcher found, deferring auto-review until watcher completes', - ); - return; - } - - if (task.status !== 'merge_ready') { - const ownerVerdict = classifyVerdict(summary); - if (ownerVerdict === 'blocked' || ownerVerdict === 'needs_context') { - requestArbiterOrEscalate({ - taskId, - now, - arbiterLogMessage: 'Owner blocked/needs_context — requesting arbiter', - escalateLogMessage: 'Owner blocked/needs_context — escalating to user', - logContext: { - taskId, - ownerVerdict, - summary: summary?.slice(0, 100), - }, - }); - return; - } - } - - if (task.round_trip_count >= PAIRED_MAX_ROUND_TRIPS) { - logger.info( - { - taskId, - roundTrips: task.round_trip_count, - max: PAIRED_MAX_ROUND_TRIPS, - }, - 'Round trip limit reached, skipping auto-review', - ); - return; - } - - const result = markPairedTaskReviewReady(taskId); - if (result) { - updatePairedTask(taskId, { - round_trip_count: task.round_trip_count + 1, - review_requested_at: now, - updated_at: now, - }); - logger.info( - { taskId, roundTrip: task.round_trip_count + 1 }, - 'Auto-triggered reviewer after owner completion', - ); - } -} - -function handleReviewerCompletion(args: { - task: PairedTask; - taskId: string; - summary?: string | null; -}): void { - const { task, taskId, summary } = args; - const now = new Date().toISOString(); - const verdict = classifyVerdict(summary); - - switch (verdict) { - case 'done': { - const ownerWs = getPairedWorkspace(taskId, 'owner'); - const approvedSourceRef = ownerWs?.workspace_dir - ? resolveCanonicalSourceRef(ownerWs.workspace_dir) - : task.source_ref; - updatePairedTask(taskId, { - status: 'merge_ready', - source_ref: approvedSourceRef, - updated_at: now, - }); - logger.info( - { - taskId, - verdict, - approvedSourceRef, - summary: summary?.slice(0, 100), - }, - 'Reviewer approved — owner gets final turn to finalize', - ); - return; - } - - case 'blocked': - case 'needs_context': - requestArbiterOrEscalate({ - taskId, - now, - arbiterLogMessage: - 'Reviewer blocked/needs_context — requesting arbiter before escalating', - escalateLogMessage: 'Reviewer escalated to user — ping-pong stopped', - logContext: { taskId, verdict, summary: summary?.slice(0, 100) }, - }); - return; - - case 'done_with_concerns': - case 'continue': - default: - if (task.round_trip_count >= ARBITER_DEADLOCK_THRESHOLD) { - requestArbiterOrEscalate({ - taskId, - now, - arbiterLogMessage: - 'Deadlock detected — requesting arbiter intervention', - escalateLogMessage: - 'Stopped ping-pong — escalating to user (arbiter not configured)', - logContext: { - taskId, - verdict, - roundTrips: task.round_trip_count, - }, - }); - return; - } - updatePairedTask(taskId, { status: 'active', updated_at: now }); - logger.info( - { taskId, verdict }, - 'Reviewer has feedback, task set back to active for owner', - ); - return; - } -} - -function handleArbiterCompletion(args: { - taskId: string; - summary?: string | null; -}): void { - const { taskId, summary } = args; - const now = new Date().toISOString(); - const arbiterVerdict = classifyArbiterVerdict(summary); - - logger.info( - { taskId, arbiterVerdict, summary: summary?.slice(0, 200) }, - 'Arbiter verdict rendered', - ); - - switch (arbiterVerdict) { - case 'proceed': - case 'revise': - case 'reset': - updatePairedTask(taskId, { - status: 'active', - round_trip_count: Math.max(0, ARBITER_DEADLOCK_THRESHOLD - 1), - arbiter_verdict: arbiterVerdict, - updated_at: now, - }); - logger.info( - { taskId, arbiterVerdict }, - 'Arbiter resolved deadlock — resuming ping-pong', - ); - return; - case 'escalate': - updatePairedTask(taskId, { - status: 'completed', - arbiter_verdict: 'escalate', - completion_reason: 'arbiter_escalated', - updated_at: now, - }); - logger.info({ taskId }, 'Arbiter escalated to user — task completed'); - return; - default: - updatePairedTask(taskId, { - status: 'active', - round_trip_count: Math.max(0, ARBITER_DEADLOCK_THRESHOLD - 1), - arbiter_verdict: 'unknown', - updated_at: now, - }); - logger.warn( - { taskId, summary: summary?.slice(0, 200) }, - 'Arbiter verdict unrecognized — falling back to proceed', - ); - return; - } -} - export function completePairedExecutionContext(args: { taskId: string; role: PairedRoomRole; diff --git a/src/task-scheduler.ts b/src/task-scheduler.ts index 0154c7f..2e0507c 100644 --- a/src/task-scheduler.ts +++ b/src/task-scheduler.ts @@ -2,6 +2,7 @@ import { ChildProcess } from 'child_process'; import { CronExpressionParser } from 'cron-parser'; import fs from 'fs'; import { getAgentOutputText } from './agent-output.js'; +import { createEvaluatedOutputHandler } from './agent-attempt.js'; import { getErrorMessage } from './utils.js'; import { ASSISTANT_NAME, SCHEDULER_POLL_INTERVAL, TIMEZONE } from './config.js'; @@ -53,10 +54,6 @@ import { formatSuspensionNotice, suspendTask, } from './task-suspension.js'; -import { - evaluateStreamedOutput, - type StreamedOutputState, -} from './streamed-output-evaluator.js'; import { extractWatchCiTarget, getTaskQueueJid, @@ -339,58 +336,31 @@ async function runTask( attemptResult: string | null; attemptError: string | null; }> => { - let streamedState: StreamedOutputState = { - sawOutput: false, - sawVisibleOutput: false, - sawSuccessNullResultWithoutOutput: false, - }; let attemptResult: string | null = null; let attemptError: string | null = null; - - const output = await runAgentProcess( - context.group, - { - prompt: task.prompt, - sessionId: context.sessionId, - groupFolder: task.group_folder, - chatJid: task.chat_jid, - isMain: context.isMain, - isScheduledTask: true, - runtimeTaskId: context.runtimeTaskId, - useTaskScopedSession: context.useTaskScopedSession, - assistantName: ASSISTANT_NAME, + const streamedOutputHandler = createEvaluatedOutputHandler({ + agentType: isClaudeAgent ? 'claude-code' : 'codex', + provider, + evaluationOptions: { + shortCircuitTriggeredErrors: true, }, - (proc, processName) => - deps.onProcess( - context.queueJid, - proc, - processName, - context.runtimeIpcDir, - ), - async (streamedOutput: AgentOutput) => { + onEvaluatedOutput: async ({ + output: streamedOutput, + outputText, + evaluation, + }) => { if (streamedOutput.phase === 'progress') { return; } - const evaluation = evaluateStreamedOutput( - streamedOutput, - streamedState, - { - agentType: isClaudeAgent ? 'claude-code' : 'codex', - provider, - shortCircuitTriggeredErrors: true, - }, - ); - streamedState = evaluation.state; - if ( evaluation.newTrigger && - typeof streamedOutput.result === 'string' && + outputText && streamedOutput.status === 'success' ) { log.warn( { reason: evaluation.newTrigger.reason, - resultPreview: streamedOutput.result.slice(0, 120), + resultPreview: outputText.slice(0, 120), }, 'Detected Claude rotation trigger during scheduled task output', ); @@ -416,7 +386,6 @@ async function runTask( return; } - const outputText = getAgentOutputText(streamedOutput); if (outputText) { attemptResult = outputText; // Paired-room scheduler output must use the reviewer bot slot. @@ -427,6 +396,29 @@ async function runTask( attemptError = streamedOutput.error || 'Unknown error'; } }, + }); + + const output = await runAgentProcess( + context.group, + { + prompt: task.prompt, + sessionId: context.sessionId, + groupFolder: task.group_folder, + chatJid: task.chat_jid, + isMain: context.isMain, + isScheduledTask: true, + runtimeTaskId: context.runtimeTaskId, + useTaskScopedSession: context.useTaskScopedSession, + assistantName: ASSISTANT_NAME, + }, + (proc, processName) => + deps.onProcess( + context.queueJid, + proc, + processName, + context.runtimeIpcDir, + ), + streamedOutputHandler.handleOutput, undefined, ); @@ -439,6 +431,7 @@ async function runTask( } } + const streamedState = streamedOutputHandler.getState(); return { output, sawOutput: streamedState.sawOutput, diff --git a/src/verification.ts b/src/verification.ts index e6bd918..af2efa0 100644 --- a/src/verification.ts +++ b/src/verification.ts @@ -1,5 +1,4 @@ import { execFile, execFileSync } from 'child_process'; -import { createHash } from 'crypto'; import fs from 'fs'; import os from 'os'; import path from 'path'; @@ -18,6 +17,11 @@ import { detectPnpmStorePath, hasInstalledNodeModules, } from './workspace-package-manager.js'; +import { + computeVerificationSnapshotId, + isVerificationSnapshotExcludedName, + resolveVerificationResponsesDir, +} from '../shared/verification-snapshot.js'; export const VERIFICATION_PROFILES = ['test', 'typecheck', 'build'] as const; @@ -58,7 +62,6 @@ interface VerificationCommandSpec { } const PRIMARY_PROJECT_MOUNT = '/workspace/project'; -const SNAPSHOT_EXCLUDE_NAMES = new Set(['.git', 'node_modules', '.env']); const MAX_OUTPUT_CHARS = 24_000; const COMMAND_TIMEOUT_MS = 20 * 60 * 1000; const COMMAND_MAX_BUFFER = 20 * 1024 * 1024; @@ -92,47 +95,14 @@ export function buildVerificationCommand( } function shouldExcludePath(name: string): boolean { - return SNAPSHOT_EXCLUDE_NAMES.has(name); -} - -function updateSnapshotHash( - hash: ReturnType, - repoDir: string, - currentPath: string, -): void { - const relPath = path.relative(repoDir, currentPath) || '.'; - const stat = fs.lstatSync(currentPath); - - if (stat.isDirectory()) { - if (relPath !== '.') { - hash.update(`dir\0${relPath}\0`); - } - for (const entry of fs.readdirSync(currentPath).sort()) { - if (shouldExcludePath(entry)) continue; - updateSnapshotHash(hash, repoDir, path.join(currentPath, entry)); - } - return; - } - - if (stat.isSymbolicLink()) { - hash.update(`symlink\0${relPath}\0${fs.readlinkSync(currentPath)}\0`); - return; - } - - if (stat.isFile()) { - hash.update(`file\0${relPath}\0`); - hash.update(fs.readFileSync(currentPath)); - hash.update('\0'); - } + return isVerificationSnapshotExcludedName(name); } export function computeVerificationSnapshot( repoDir: string, ): VerificationSnapshot { - const hash = createHash('sha256'); - updateSnapshotHash(hash, repoDir, repoDir); return { - snapshotId: `fs:${hash.digest('hex').slice(0, 24)}`, + snapshotId: computeVerificationSnapshotId(repoDir), }; } @@ -403,7 +373,7 @@ export async function runVerificationRequest( } export function resolveVerificationResponseDir(groupFolder: string): string { - return path.join(resolveGroupIpcPath(groupFolder), 'verification-responses'); + return resolveVerificationResponsesDir(resolveGroupIpcPath(groupFolder)); } export function writeVerificationResponse(