claude-bot/EJClaw
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Refactor paired execution flow boundaries

Browse Source
This commit is contained in:
ejclaw
2026-04-07 00:45:17 +09:00
parent 0363021218
commit cbb6bc97c7
15 changed files with 1345 additions and 1116 deletions
Download Patch File Download Diff File Expand all files Collapse all files

48
runners/agent-runner/src/verification.ts
View File

@@ -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<typeof createHash>,
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,

8
shared/verification-snapshot.d.ts vendored Normal file
View File

@@ -0,0 +1,8 @@
export declare const VERIFICATION_SNAPSHOT_EXCLUDE_NAMES: ReadonlySet<string>;
export declare function isVerificationSnapshotExcludedName(
name: string,
): boolean;
export declare function computeVerificationSnapshotId(repoDir: string): string;
export declare function resolveVerificationResponsesDir(
hostIpcDir: string,
): string;

50
shared/verification-snapshot.js Normal file
View File

@@ -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');
}

71
src/agent-attempt.ts Normal file
View File

@@ -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<typeof getStructuredAgentOutput>;
evaluation: EvaluateStreamedOutputResult;
}
export function createInitialStreamedOutputState(): StreamedOutputState {
return {
sawOutput: false,
sawVisibleOutput: false,
sawSuccessNullResultWithoutOutput: false,
};
}
export function createEvaluatedOutputHandler(args: {
agentType: EvaluateStreamedOutputOptions['agentType'];
provider: string;
evaluationOptions?: Omit<EvaluateStreamedOutputOptions, 'agentType' | 'provider'>;
onEvaluatedOutput?: (
output: EvaluatedAgentOutput,
) => Promise<void> | void;
}): {
handleOutput: (output: AgentOutput) => Promise<void>;
getState: () => StreamedOutputState;
markVisibleOutput: () => void;
} {
let state = createInitialStreamedOutputState();
return {
async handleOutput(output: AgentOutput): Promise<void> {
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,
};
},
};
}

32
src/message-agent-executor.test.ts
View File

@@ -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',
}),

91
src/message-agent-executor.ts
View File

@@ -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,6 +286,7 @@ export async function runAgentForGroup(
const systemPrompt =
readArbiterPrompt(process.cwd()) || 'You are an arbiter.';
try {
const references = await collectMoaReferences({
config: moaConfig,
systemPrompt,
@@ -307,6 +304,15 @@ export async function runAgentForGroup(
'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;
@@ -411,18 +417,25 @@ export async function runAgentForGroup(
retryAfterMs?: number;
};
}> => {
let streamedState: StreamedOutputState = {
sawOutput: false,
sawVisibleOutput: false,
sawSuccessNullResultWithoutOutput: false,
};
const attemptSessionId = currentSessionId;
const wrappedOnOutput = onOutput
? async (output: AgentOutput) => {
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';
const outputText = getAgentOutputText(output);
const structuredOutput = getStructuredAgentOutput(output);
if (outputPhase !== 'final') {
log.info(
{
@@ -431,7 +444,7 @@ export async function runAgentForGroup(
outputStatus: output.status,
visibility: structuredOutput?.visibility ?? null,
preview:
typeof outputText === 'string' && outputText.length > 0
outputText && outputText.length > 0
? outputText.slice(0, 160)
: null,
errorPreview:
@@ -470,19 +483,8 @@ export async function runAgentForGroup(
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 (typeof outputText === 'string' && outputText.length > 0) {
if (outputText && outputText.length > 0) {
pairedExecutionSummary = outputText.slice(0, 500);
pairedFullOutput = outputText;
} else if (
@@ -491,11 +493,7 @@ export async function runAgentForGroup(
) {
pairedExecutionSummary = output.error.slice(0, 500);
}
if (
evaluation.newTrigger &&
typeof outputText === 'string' &&
output.status === 'success'
) {
if (evaluation.newTrigger && outputText && output.status === 'success') {
log.warn(
{
reason: evaluation.newTrigger.reason,
@@ -521,10 +519,7 @@ export async function runAgentForGroup(
if (evaluation.suppressedAuthError) {
log.warn(
{
resultPreview:
typeof outputText === 'string'
? outputText.slice(0, 120)
: undefined,
resultPreview: outputText ? outputText.slice(0, 120) : undefined,
},
'Suppressed Claude 401 auth error from chat output',
);
@@ -534,8 +529,7 @@ export async function runAgentForGroup(
if (evaluation.suppressedRetryableSessionFailure) {
log.warn(
{
resultPreview:
typeof outputText === 'string'
resultPreview: outputText
? outputText.slice(0, 160)
: output.error?.slice(0, 160),
},
@@ -547,15 +541,18 @@ export async function runAgentForGroup(
if (!evaluation.shouldForwardOutput) {
return;
}
if (typeof outputText === 'string' && outputText.length > 0) {
streamedState = {
...evaluation.state,
sawVisibleOutput: true,
};
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,

366
src/message-runtime-flow.ts Normal file
View File

@@ -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<typeof buildReviewerPendingPrompt>[0]['recentHumanMessages'];
labeledRecentMessages: Parameters<typeof buildArbiterPromptForTask>[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<PendingPairedTurn, null>;
chatJid: string;
group: RegisteredGroup;
runId: string;
log: typeof logger;
saveState: () => void;
lastAgentTimestamps: Record<string, string>;
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<boolean> {
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<string, string>;
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<boolean> {
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'
);
}

363
src/message-runtime.ts
View File

@@ -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<boolean> => {
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<boolean> => {
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;

81
src/paired-execution-context-arbiter.ts Normal file
View File

@@ -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;
}
}

201
src/paired-execution-context-owner.ts Normal file
View File

@@ -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',
);
}
}

139
src/paired-execution-context-reviewer.ts Normal file
View File

@@ -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;
}
}

122
src/paired-execution-context-shared.ts Normal file
View File

@@ -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(/<internal>[\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(/<internal>[\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<string, unknown>;
}): 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);
}

542
src/paired-execution-context.ts
View File

@@ -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 <internal>...</internal> tags — these are internal reasoning, not the verdict
const cleaned = summary.replace(/<internal>[\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(/<internal>[\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<string, unknown>;
}): 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;

81
src/task-scheduler.ts
View File

@@ -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,

46
src/verification.ts
View File

@@ -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<typeof createHash>,
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(