diff --git a/src/db.test.ts b/src/db.test.ts index 0d687f5..0a15fc6 100644 --- a/src/db.test.ts +++ b/src/db.test.ts @@ -1242,10 +1242,53 @@ describe('paired task state', () => { expect(outputs.map((output) => output.turn_number)).toEqual([1, 2]); expect(outputs[0].role).toBe('owner'); expect(outputs[0].output_text).toHaveLength(50_000); + expect(outputs[0].verdict).toBe('continue'); expect(outputs[1].output_text).toBe('review turn'); + expect(outputs[1].verdict).toBe('continue'); expect(getLatestTurnNumber('paired-task-turn-output')).toBe(2); }); + it('stores the parsed visible verdict with paired turn outputs', () => { + createPairedTask({ + id: 'paired-task-turn-output-verdict', + chat_jid: 'dc:paired', + group_folder: 'paired-room', + owner_service_id: 'codex-main', + reviewer_service_id: 'codex-review', + title: null, + source_ref: null, + plan_notes: null, + round_trip_count: 0, + review_requested_at: null, + status: 'active', + arbiter_verdict: null, + arbiter_requested_at: null, + completion_reason: null, + created_at: '2026-03-28T00:00:00.000Z', + updated_at: '2026-03-28T00:00:00.000Z', + }); + + insertPairedTurnOutput( + 'paired-task-turn-output-verdict', + 1, + 'owner', + 'STEP_DONE\n1단계 완료', + ); + insertPairedTurnOutput( + 'paired-task-turn-output-verdict', + 2, + 'owner', + 'TASK_DONE\n요청 범위 전체 완료', + ); + + const outputs = getPairedTurnOutputs('paired-task-turn-output-verdict'); + + expect(outputs.map((output) => output.verdict)).toEqual([ + 'step_done', + 'task_done', + ]); + }); + it('preserves explicit created_at when inserting a paired turn output', () => { createPairedTask({ id: 'paired-task-turn-output-created-at', diff --git a/src/db/base-schema.ts b/src/db/base-schema.ts index 1fd66ba..d44e693 100644 --- a/src/db/base-schema.ts +++ b/src/db/base-schema.ts @@ -119,6 +119,10 @@ export function applyBaseSchema(database: Database): void { review_requested_at TEXT, round_trip_count INTEGER NOT NULL DEFAULT 0, owner_failure_count INTEGER NOT NULL DEFAULT 0, + owner_step_done_streak INTEGER NOT NULL DEFAULT 0, + finalize_step_done_count INTEGER NOT NULL DEFAULT 0, + task_done_then_user_reopen_count INTEGER NOT NULL DEFAULT 0, + empty_step_done_streak INTEGER NOT NULL DEFAULT 0, status TEXT NOT NULL DEFAULT 'active', arbiter_verdict TEXT, arbiter_requested_at TEXT, @@ -154,6 +158,7 @@ export function applyBaseSchema(database: Database): void { turn_number INTEGER NOT NULL, role TEXT NOT NULL, output_text TEXT NOT NULL, + verdict TEXT, created_at TEXT NOT NULL, UNIQUE(task_id, turn_number, role) ); diff --git a/src/db/bootstrap.test.ts b/src/db/bootstrap.test.ts index c375201..49dc122 100644 --- a/src/db/bootstrap.test.ts +++ b/src/db/bootstrap.test.ts @@ -36,6 +36,7 @@ function getExpectedSchemaMigrations(): Array<{ { version: 9, name: 'paired_workspace_project_schema_cleanup' }, { version: 10, name: 'paired_turn_provenance_upgrade' }, { version: 11, name: 'owner_failure_count' }, + { version: 12, name: 'paired_verdict_and_step_telemetry' }, ]; } diff --git a/src/db/migrations/012_paired-verdict-and-step-telemetry.ts b/src/db/migrations/012_paired-verdict-and-step-telemetry.ts new file mode 100644 index 0000000..d4f9e6e --- /dev/null +++ b/src/db/migrations/012_paired-verdict-and-step-telemetry.ts @@ -0,0 +1,75 @@ +import type { Database } from 'bun:sqlite'; + +import { tableHasColumn } from './helpers.js'; +import type { SchemaMigrationDefinition } from './types.js'; + +export const PAIRED_VERDICT_AND_STEP_TELEMETRY_MIGRATION: SchemaMigrationDefinition = + { + version: 12, + name: 'paired_verdict_and_step_telemetry', + apply(database: Database) { + if (!tableHasColumn(database, 'paired_turn_outputs', 'verdict')) { + database.exec(` + ALTER TABLE paired_turn_outputs + ADD COLUMN verdict TEXT + `); + } + + if (!tableHasColumn(database, 'paired_tasks', 'owner_step_done_streak')) { + database.exec(` + ALTER TABLE paired_tasks + ADD COLUMN owner_step_done_streak INTEGER NOT NULL DEFAULT 0 + `); + } + + if ( + !tableHasColumn(database, 'paired_tasks', 'finalize_step_done_count') + ) { + database.exec(` + ALTER TABLE paired_tasks + ADD COLUMN finalize_step_done_count INTEGER NOT NULL DEFAULT 0 + `); + } + + if ( + !tableHasColumn( + database, + 'paired_tasks', + 'task_done_then_user_reopen_count', + ) + ) { + database.exec(` + ALTER TABLE paired_tasks + ADD COLUMN task_done_then_user_reopen_count INTEGER NOT NULL DEFAULT 0 + `); + } + + if (!tableHasColumn(database, 'paired_tasks', 'empty_step_done_streak')) { + database.exec(` + ALTER TABLE paired_tasks + ADD COLUMN empty_step_done_streak INTEGER NOT NULL DEFAULT 0 + `); + } + + database.exec(` + UPDATE paired_tasks + SET owner_step_done_streak = 0 + WHERE owner_step_done_streak IS NULL + `); + database.exec(` + UPDATE paired_tasks + SET finalize_step_done_count = 0 + WHERE finalize_step_done_count IS NULL + `); + database.exec(` + UPDATE paired_tasks + SET task_done_then_user_reopen_count = 0 + WHERE task_done_then_user_reopen_count IS NULL + `); + database.exec(` + UPDATE paired_tasks + SET empty_step_done_streak = 0 + WHERE empty_step_done_streak IS NULL + `); + }, + }; diff --git a/src/db/migrations/index.ts b/src/db/migrations/index.ts index 85eaf42..d6b5e83 100644 --- a/src/db/migrations/index.ts +++ b/src/db/migrations/index.ts @@ -11,6 +11,7 @@ import { PAIRED_TASK_SCHEMA_CLEANUP_MIGRATION } from './008_paired-task-schema-c import { PAIRED_WORKSPACE_PROJECT_SCHEMA_CLEANUP_MIGRATION } from './009_paired-workspace-project-schema-cleanup.js'; import { PAIRED_TURN_PROVENANCE_UPGRADE_MIGRATION } from './010_paired-turn-provenance-upgrade.js'; import { OWNER_FAILURE_COUNT_MIGRATION } from './011_owner-failure-count.js'; +import { PAIRED_VERDICT_AND_STEP_TELEMETRY_MIGRATION } from './012_paired-verdict-and-step-telemetry.js'; import type { SchemaMigrationArgs, SchemaMigrationDefinition, @@ -30,6 +31,7 @@ const ORDERED_SCHEMA_MIGRATIONS: readonly SchemaMigrationDefinition[] = [ PAIRED_WORKSPACE_PROJECT_SCHEMA_CLEANUP_MIGRATION, PAIRED_TURN_PROVENANCE_UPGRADE_MIGRATION, OWNER_FAILURE_COUNT_MIGRATION, + PAIRED_VERDICT_AND_STEP_TELEMETRY_MIGRATION, ]; function ensureSchemaMigrationsTable(database: Database): void { diff --git a/src/db/paired-state.ts b/src/db/paired-state.ts index cb589bd..0ebbb20 100644 --- a/src/db/paired-state.ts +++ b/src/db/paired-state.ts @@ -51,6 +51,10 @@ export type PairedTaskUpdates = Partial< | 'review_requested_at' | 'round_trip_count' | 'owner_failure_count' + | 'owner_step_done_streak' + | 'finalize_step_done_count' + | 'task_done_then_user_reopen_count' + | 'empty_step_done_streak' | 'status' | 'arbiter_verdict' | 'arbiter_requested_at' @@ -173,6 +177,10 @@ export function createPairedTaskInDatabase( review_requested_at, round_trip_count, owner_failure_count, + owner_step_done_streak, + finalize_step_done_count, + task_done_then_user_reopen_count, + empty_step_done_streak, status, arbiter_verdict, arbiter_requested_at, @@ -180,7 +188,7 @@ export function createPairedTaskInDatabase( created_at, updated_at ) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) `, ) .run( @@ -198,6 +206,10 @@ export function createPairedTaskInDatabase( task.review_requested_at, task.round_trip_count, task.owner_failure_count ?? 0, + task.owner_step_done_streak ?? 0, + task.finalize_step_done_count ?? 0, + task.task_done_then_user_reopen_count ?? 0, + task.empty_step_done_streak ?? 0, task.status, task.arbiter_verdict, task.arbiter_requested_at, @@ -306,6 +318,22 @@ export function updatePairedTaskInDatabase( fields.push('owner_failure_count = ?'); values.push(updates.owner_failure_count); } + if (updates.owner_step_done_streak !== undefined) { + fields.push('owner_step_done_streak = ?'); + values.push(updates.owner_step_done_streak); + } + if (updates.finalize_step_done_count !== undefined) { + fields.push('finalize_step_done_count = ?'); + values.push(updates.finalize_step_done_count); + } + if (updates.task_done_then_user_reopen_count !== undefined) { + fields.push('task_done_then_user_reopen_count = ?'); + values.push(updates.task_done_then_user_reopen_count); + } + if (updates.empty_step_done_streak !== undefined) { + fields.push('empty_step_done_streak = ?'); + values.push(updates.empty_step_done_streak); + } if (updates.status !== undefined) { fields.push('status = ?'); values.push(updates.status); @@ -369,6 +397,22 @@ export function updatePairedTaskIfUnchangedInDatabase( fields.push('owner_failure_count = ?'); values.push(updates.owner_failure_count); } + if (updates.owner_step_done_streak !== undefined) { + fields.push('owner_step_done_streak = ?'); + values.push(updates.owner_step_done_streak); + } + if (updates.finalize_step_done_count !== undefined) { + fields.push('finalize_step_done_count = ?'); + values.push(updates.finalize_step_done_count); + } + if (updates.task_done_then_user_reopen_count !== undefined) { + fields.push('task_done_then_user_reopen_count = ?'); + values.push(updates.task_done_then_user_reopen_count); + } + if (updates.empty_step_done_streak !== undefined) { + fields.push('empty_step_done_streak = ?'); + values.push(updates.empty_step_done_streak); + } if (updates.status !== undefined) { fields.push('status = ?'); values.push(updates.status); diff --git a/src/db/paired-turn-outputs.ts b/src/db/paired-turn-outputs.ts index 9c89bb3..01eceda 100644 --- a/src/db/paired-turn-outputs.ts +++ b/src/db/paired-turn-outputs.ts @@ -1,6 +1,7 @@ import { Database } from 'bun:sqlite'; import { logger } from '../logger.js'; +import { parseVisibleVerdict } from '../paired-verdict.js'; import { PairedRoomRole, PairedTurnOutput } from '../types.js'; const MAX_TURN_OUTPUT_CHARS = 50_000; @@ -29,14 +30,15 @@ export function insertPairedTurnOutputInDatabase( database .prepare( `INSERT OR REPLACE INTO paired_turn_outputs - (task_id, turn_number, role, output_text, created_at) - VALUES (?, ?, ?, ?, ?)`, + (task_id, turn_number, role, output_text, verdict, created_at) + VALUES (?, ?, ?, ?, ?, ?)`, ) .run( taskId, turnNumber, role, outputText.slice(0, MAX_TURN_OUTPUT_CHARS), + parseVisibleVerdict(outputText), createdAt ?? new Date().toISOString(), ); } diff --git a/src/message-runtime-flow.ts b/src/message-runtime-flow.ts index 5d4f46d..9fa5aeb 100644 --- a/src/message-runtime-flow.ts +++ b/src/message-runtime-flow.ts @@ -24,12 +24,12 @@ import { } from './message-runtime-rules.js'; import type { ExecuteTurnFn } from './message-runtime-types.js'; import { buildPairedTurnIdentity } from './paired-turn-identity.js'; +import { resolveStoredVisibleVerdict } from './paired-verdict.js'; import { claimPairedTurnExecution, type ScheduledPairedFollowUpIntentKind, } from './paired-follow-up-scheduler.js'; import { hasReviewerLease } from './service-routing.js'; -import { parseVisibleVerdict } from './paired-execution-context-shared.js'; import type { Channel, NewMessage, @@ -128,9 +128,10 @@ export function buildPendingPairedTurn(args: { const nextTurnAction = resolveNextTurnAction({ taskStatus, lastTurnOutputRole: lastTurnOutput?.role ?? null, - lastTurnOutputVerdict: lastTurnOutput?.output_text - ? parseVisibleVerdict(lastTurnOutput.output_text) - : null, + lastTurnOutputVerdict: resolveStoredVisibleVerdict({ + verdict: lastTurnOutput?.verdict ?? null, + outputText: lastTurnOutput?.output_text ?? null, + }), }); const recentMessages = getRecentChatMessages(chatJid, 20); const lastHumanMessage = getLastHumanMessageContent(chatJid); diff --git a/src/message-runtime-follow-up.ts b/src/message-runtime-follow-up.ts index 97725eb..43fbf0d 100644 --- a/src/message-runtime-follow-up.ts +++ b/src/message-runtime-follow-up.ts @@ -1,8 +1,8 @@ import { getPairedTurnOutputs } from './db.js'; import { - parseVisibleVerdict, type VisibleVerdict, } from './paired-execution-context-shared.js'; +import { resolveStoredVisibleVerdict } from './paired-verdict.js'; import { matchesExpectedPairedFollowUpIntent, resolveFollowUpDispatch, @@ -55,9 +55,11 @@ export function resolveLatestPairedTurnOutputContext(args: { : null; return { role: latestOutput?.role ?? args.fallbackLastTurnOutputRole ?? null, - verdict: latestOutput?.output_text - ? parseVisibleVerdict(latestOutput.output_text) - : (args.fallbackLastTurnOutputVerdict ?? null), + verdict: + resolveStoredVisibleVerdict({ + verdict: latestOutput?.verdict ?? null, + outputText: latestOutput?.output_text ?? null, + }) ?? args.fallbackLastTurnOutputVerdict ?? null, }; } diff --git a/src/message-runtime-queue.ts b/src/message-runtime-queue.ts index 1ceaee8..4cacf38 100644 --- a/src/message-runtime-queue.ts +++ b/src/message-runtime-queue.ts @@ -11,7 +11,7 @@ import { isBotOnlyPairedRoomTurn, } from './message-runtime-flow.js'; import { buildPairedTurnIdentity } from './paired-turn-identity.js'; -import { parseVisibleVerdict } from './paired-execution-context-shared.js'; +import { resolveStoredVisibleVerdict } from './paired-verdict.js'; import { advanceLastAgentCursor, resolveActiveRole, @@ -185,9 +185,10 @@ export async function runQueuedGroupTurn(args: { ? (turnOutputs.at(-1)?.role ?? null) : null; const lastTurnOutputVerdict = currentTask - ? turnOutputs.at(-1)?.output_text - ? parseVisibleVerdict(turnOutputs.at(-1)?.output_text) - : null + ? resolveStoredVisibleVerdict({ + verdict: turnOutputs.at(-1)?.verdict ?? null, + outputText: turnOutputs.at(-1)?.output_text ?? null, + }) : null; const turnRole = currentTask ? hasHumanMsg diff --git a/src/message-runtime.test.ts b/src/message-runtime.test.ts index b5d82df..1af391c 100644 --- a/src/message-runtime.test.ts +++ b/src/message-runtime.test.ts @@ -842,6 +842,127 @@ describe('createMessageRuntime', () => { ); }); + it('suppresses a stale owner work item when a new human message arrives while the paired task is still active', async () => { + const chatJid = 'group@test'; + const group = makeGroup('codex'); + const channel = makeChannel(chatJid); + const enqueueMessageCheck = vi.fn(); + const activeTask = { + id: 'task-active-owner-follow-up', + chat_jid: chatJid, + group_folder: group.folder, + owner_service_id: 'claude', + reviewer_service_id: 'codex-main', + title: null, + source_ref: 'HEAD', + plan_notes: null, + review_requested_at: null, + round_trip_count: 0, + status: 'active', + arbiter_verdict: null, + arbiter_requested_at: null, + completion_reason: null, + created_at: '2026-03-30T00:00:00.000Z', + updated_at: '2026-03-30T00:00:05.000Z', + } as any; + + vi.mocked(serviceRouting.hasReviewerLease).mockReturnValue(true); + vi.mocked(db.getOpenWorkItem).mockReturnValue({ + id: 109, + group_folder: group.folder, + chat_jid: chatJid, + agent_type: 'codex', + service_id: 'claude', + delivery_role: 'owner', + status: 'delivery_retry', + start_seq: 1, + end_seq: 1, + result_payload: '이전 step 결과입니다.', + delivery_attempts: 1, + created_at: '2026-03-30T00:00:05.000Z', + updated_at: '2026-03-30T00:00:05.000Z', + delivered_at: null, + delivery_message_id: null, + last_error: 'discord send failed', + }); + vi.mocked(db.getMessagesSince).mockReturnValue([ + { + id: 'msg-2', + chat_jid: chatJid, + sender: 'user@test', + sender_name: 'User', + content: '이전 답 말고 이 방향으로 진행해줘', + timestamp: '2026-03-30T00:00:10.000Z', + seq: 2, + }, + ]); + vi.mocked(db.getLatestOpenPairedTaskForChat).mockReturnValue(activeTask); + vi.mocked(agentRunner.runAgentProcess).mockImplementation( + async (_group, input, _onProcess, onOutput) => { + expect(input.prompt).toContain('이 방향으로 진행해줘'); + expect(input.prompt).not.toContain('이전 step 결과입니다.'); + await onOutput?.({ + status: 'success', + phase: 'final', + result: 'STEP_DONE\n새 입력 기준으로 계속 진행', + output: { + visibility: 'public', + text: 'STEP_DONE\n새 입력 기준으로 계속 진행', + }, + } as any); + return { + status: 'success', + result: 'STEP_DONE\n새 입력 기준으로 계속 진행', + }; + }, + ); + + const runtime = createMessageRuntime({ + assistantName: 'Andy', + idleTimeout: 1_000, + pollInterval: 1_000, + timezone: 'UTC', + triggerPattern: /^@Andy\b/i, + channels: [channel], + queue: { + registerProcess: vi.fn(), + closeStdin: vi.fn(), + notifyIdle: vi.fn(), + enqueueMessageCheck, + } as any, + getRoomBindings: () => ({ [chatJid]: group }), + getSessions: () => ({}), + getLastTimestamp: () => '', + setLastTimestamp: vi.fn(), + getLastAgentTimestamps: () => ({}), + saveState: vi.fn(), + persistSession: vi.fn(), + clearSession: vi.fn(), + }); + + const result = await runtime.processGroupMessages(chatJid, { + runId: 'run-suppress-stale-active-owner-work-item', + reason: 'messages', + }); + + expect(result).toBe(true); + expect(db.markWorkItemDelivered).toHaveBeenCalledWith(109); + expect(channel.sendMessage).not.toHaveBeenCalledWith( + chatJid, + '이전 step 결과입니다.', + ); + expect(agentRunner.runAgentProcess).toHaveBeenCalledTimes(1); + expect(logger.info).toHaveBeenCalledWith( + expect.objectContaining({ + chatJid, + workItemId: 109, + taskId: 'task-active-owner-follow-up', + taskStatus: 'active', + }), + 'Suppressed stale owner delivery retry because a new human message arrived while the paired task was still active', + ); + }); + it('suppresses duplicate stale work item delivery and logs the suppression reason', async () => { const chatJid = 'group@test'; const group = makeGroup('codex'); diff --git a/src/message-runtime.ts b/src/message-runtime.ts index de06cbe..39dbf06 100644 --- a/src/message-runtime.ts +++ b/src/message-runtime.ts @@ -56,6 +56,7 @@ import { } from './types.js'; import { createScopedLogger, logger } from './logger.js'; import { hasReviewerLease } from './service-routing.js'; +import type { WorkItem } from './db/work-items.js'; export { resolveHandoffCursorKey, resolveHandoffRoleOverride, @@ -113,6 +114,54 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { deps.queue, ); + const getFreshHumanPreflightMessages = ( + chatJid: string, + channel: Channel, + ): NewMessage[] => { + const sinceSeqCursor = deps.getLastAgentTimestamps()[chatJid] || '0'; + const preflightRawMessages = getMessagesSinceSeq( + chatJid, + sinceSeqCursor, + deps.assistantName, + ); + const preflightMessages = filterLoopingPairedBotMessages( + chatJid, + getProcessableMessages(chatJid, preflightRawMessages, channel), + FAILURE_FINAL_TEXT, + ); + return preflightMessages.filter( + (message) => message.is_from_me !== true && !message.is_bot_message, + ); + }; + + const hasHumanMessageAfterWorkItem = ( + openWorkItem: WorkItem, + freshHumanMessages: NewMessage[], + ): boolean => { + if (freshHumanMessages.length === 0) { + return false; + } + + const workItemSeq = openWorkItem.end_seq ?? openWorkItem.start_seq ?? null; + const workItemUpdatedAt = Date.parse(openWorkItem.updated_at); + + return freshHumanMessages.some((message) => { + if (message.seq != null && workItemSeq != null) { + return message.seq > workItemSeq; + } + + const messageTimestamp = Date.parse(message.timestamp); + if ( + Number.isFinite(messageTimestamp) && + Number.isFinite(workItemUpdatedAt) + ) { + return messageTimestamp > workItemUpdatedAt; + } + + return true; + }); + }; + const scheduleQueuedPairedFollowUp = (args: { chatJid: string; runId: string; @@ -351,25 +400,12 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { ? getLatestOpenPairedTaskForChat(chatJid) : null; let openWorkItem = getOpenWorkItemForChat(chatJid, SERVICE_SESSION_SCOPE); - if ( - pendingTask?.status === 'merge_ready' && - openWorkItem?.delivery_role === 'owner' - ) { - const sinceSeqCursor = deps.getLastAgentTimestamps()[chatJid] || '0'; - const preflightRawMessages = getMessagesSinceSeq( - chatJid, - sinceSeqCursor, - deps.assistantName, - ); - const preflightMessages = filterLoopingPairedBotMessages( - chatJid, - getProcessableMessages(chatJid, preflightRawMessages, channel), - FAILURE_FINAL_TEXT, - ); - const hasFreshHumanMessage = preflightMessages.some( - (message) => message.is_from_me !== true && !message.is_bot_message, - ); - if (hasFreshHumanMessage) { + if (openWorkItem?.delivery_role === 'owner' && pendingTask) { + const freshHumanMessages = getFreshHumanPreflightMessages(chatJid, channel); + if ( + pendingTask.status === 'merge_ready' && + freshHumanMessages.length > 0 + ) { const resolvedTask = resolveOwnerTaskForHumanMessage({ group, chatJid, @@ -389,6 +425,23 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { ); openWorkItem = undefined; } + } else if ( + pendingTask.status === 'active' && + hasHumanMessageAfterWorkItem(openWorkItem, freshHumanMessages) + ) { + markWorkItemDelivered(openWorkItem.id); + log.info( + { + chatJid, + workItemId: openWorkItem.id, + taskId: pendingTask.id, + taskStatus: pendingTask.status, + workItemEndSeq: openWorkItem.end_seq ?? null, + freshHumanMessageCount: freshHumanMessages.length, + }, + 'Suppressed stale owner delivery retry because a new human message arrived while the paired task was still active', + ); + openWorkItem = undefined; } } const openWorkItemOutcome = await processOpenWorkItemDelivery({ diff --git a/src/paired-execution-context-owner.ts b/src/paired-execution-context-owner.ts index c59ae8b..547b4b9 100644 --- a/src/paired-execution-context-owner.ts +++ b/src/paired-execution-context-owner.ts @@ -22,6 +22,8 @@ import type { PairedTask } from './types.js'; type OwnerFinalizeOutcome = 'stop' | 're_review' | 'continue_owner'; const OWNER_FAILURE_ESCALATION_THRESHOLD = 2; +const EMPTY_STEP_DONE_THRESHOLD = 2; +const OWNER_STEP_DONE_LOOP_THRESHOLD = 3; export function handleFailedOwnerExecution(args: { task: PairedTask; @@ -101,6 +103,14 @@ function handleOwnerFinalizeCompletion(args: { const hasNewChanges = workspace?.workspace_dir ? hasCodeChangesSinceRef(workspace.workspace_dir, task.source_ref) : null; + const nextFinalizeStepDoneCount = + ownerVerdict === 'step_done' + ? (task.finalize_step_done_count ?? 0) + 1 + : task.finalize_step_done_count ?? 0; + const nextEmptyStepDoneStreak = + ownerVerdict === 'step_done' && hasNewChanges === false + ? (task.empty_step_done_streak ?? 0) + 1 + : 0; const signal = resolveOwnerCompletionSignal({ phase: 'finalize', visibleVerdict: ownerVerdict, @@ -138,6 +148,41 @@ function handleOwnerFinalizeCompletion(args: { }, patch: { owner_failure_count: 0, + owner_step_done_streak: 0, + finalize_step_done_count: nextFinalizeStepDoneCount, + empty_step_done_streak: nextEmptyStepDoneStreak, + }, + }); + return 'stop'; + } + + if ( + signal.kind === 'request_owner_continue' && + hasNewChanges === false && + nextEmptyStepDoneStreak >= EMPTY_STEP_DONE_THRESHOLD + ) { + requestArbiterOrEscalate({ + taskId, + currentStatus: task.status, + expectedUpdatedAt: task.updated_at, + now, + arbiterLogMessage: + 'Owner repeated STEP_DONE during finalize without code changes — requesting arbiter', + escalateLogMessage: + 'Owner repeated STEP_DONE during finalize without code changes — escalating to user', + logContext: { + taskId, + ownerVerdict, + hasNewChanges, + emptyStepDoneStreak: nextEmptyStepDoneStreak, + finalizeStepDoneCount: nextFinalizeStepDoneCount, + summary: summary?.slice(0, 100), + }, + patch: { + owner_failure_count: 0, + owner_step_done_streak: 0, + finalize_step_done_count: nextFinalizeStepDoneCount, + empty_step_done_streak: nextEmptyStepDoneStreak, }, }); return 'stop'; @@ -153,6 +198,9 @@ function handleOwnerFinalizeCompletion(args: { updatedAt: now, patch: { owner_failure_count: 0, + owner_step_done_streak: 0, + finalize_step_done_count: nextFinalizeStepDoneCount, + empty_step_done_streak: 0, }, }); } @@ -179,12 +227,17 @@ function handleOwnerFinalizeCompletion(args: { updatedAt: now, patch: { owner_failure_count: 0, + owner_step_done_streak: 0, + finalize_step_done_count: nextFinalizeStepDoneCount, + empty_step_done_streak: nextEmptyStepDoneStreak, }, }); logger.info( { taskId, ownerVerdict, + emptyStepDoneStreak: nextEmptyStepDoneStreak, + finalizeStepDoneCount: nextFinalizeStepDoneCount, summary: summary?.slice(0, 100), }, 'Owner marked finalize output as an intermediate step — task returned to active without re-review', @@ -201,6 +254,9 @@ function handleOwnerFinalizeCompletion(args: { patch: { completion_reason: 'done', owner_failure_count: 0, + owner_step_done_streak: 0, + finalize_step_done_count: nextFinalizeStepDoneCount, + empty_step_done_streak: 0, }, }); logger.info( @@ -242,6 +298,8 @@ function maybeAutoTriggerReviewerAfterOwnerCompletion(args: { patch: { round_trip_count: task.round_trip_count + 1, owner_failure_count: 0, + owner_step_done_streak: 0, + empty_step_done_streak: 0, }, }); if (hasActiveCiWatcherForChat(task.chat_jid)) { @@ -287,6 +345,16 @@ export function handleOwnerCompletion(args: { } const ownerVerdict = parseVisibleVerdict(summary); + const workspace = getPairedWorkspace(task.id, 'owner'); + const hasNewChanges = workspace?.workspace_dir + ? hasCodeChangesSinceRef(workspace.workspace_dir, task.source_ref) + : null; + const nextOwnerStepDoneStreak = + ownerVerdict === 'step_done' ? (task.owner_step_done_streak ?? 0) + 1 : 0; + const nextEmptyStepDoneStreak = + ownerVerdict === 'step_done' && hasNewChanges === false + ? (task.empty_step_done_streak ?? 0) + 1 + : 0; const signal = resolveOwnerCompletionSignal({ phase: 'normal', visibleVerdict: ownerVerdict, @@ -303,10 +371,46 @@ export function handleOwnerCompletion(args: { logContext: { taskId, ownerVerdict, + ownerStepDoneStreak: nextOwnerStepDoneStreak, + emptyStepDoneStreak: nextEmptyStepDoneStreak, summary: summary?.slice(0, 100), }, patch: { owner_failure_count: 0, + owner_step_done_streak: nextOwnerStepDoneStreak, + empty_step_done_streak: nextEmptyStepDoneStreak, + }, + }); + return; + } + + if ( + signal.kind === 'request_owner_continue' && + hasNewChanges === false && + nextOwnerStepDoneStreak >= OWNER_STEP_DONE_LOOP_THRESHOLD && + nextEmptyStepDoneStreak >= OWNER_STEP_DONE_LOOP_THRESHOLD + ) { + requestArbiterOrEscalate({ + taskId, + currentStatus: task.status, + expectedUpdatedAt: task.updated_at, + now, + arbiterLogMessage: + 'Owner repeated STEP_DONE in active mode without code changes — requesting arbiter', + escalateLogMessage: + 'Owner repeated STEP_DONE in active mode without code changes — escalating to user', + logContext: { + taskId, + ownerVerdict, + hasNewChanges, + ownerStepDoneStreak: nextOwnerStepDoneStreak, + emptyStepDoneStreak: nextEmptyStepDoneStreak, + summary: summary?.slice(0, 100), + }, + patch: { + owner_failure_count: 0, + owner_step_done_streak: nextOwnerStepDoneStreak, + empty_step_done_streak: nextEmptyStepDoneStreak, }, }); return; @@ -319,15 +423,38 @@ export function handleOwnerCompletion(args: { updatedAt: now, patch: { owner_failure_count: 0, + owner_step_done_streak: nextOwnerStepDoneStreak, + empty_step_done_streak: nextEmptyStepDoneStreak, }, }); logger.info( - { taskId, ownerVerdict, summary: summary?.slice(0, 100) }, + { + taskId, + ownerVerdict, + hasNewChanges, + ownerStepDoneStreak: nextOwnerStepDoneStreak, + emptyStepDoneStreak: nextEmptyStepDoneStreak, + summary: summary?.slice(0, 100), + }, 'Owner marked the current output as an intermediate completed step — keeping task active for follow-up', ); return; } + if ( + nextOwnerStepDoneStreak !== (task.owner_step_done_streak ?? 0) || + nextEmptyStepDoneStreak !== (task.empty_step_done_streak ?? 0) + ) { + applyPairedTaskPatch({ + taskId, + expectedUpdatedAt: task.updated_at, + updatedAt: now, + patch: { + owner_step_done_streak: nextOwnerStepDoneStreak, + empty_step_done_streak: nextEmptyStepDoneStreak, + }, + }); + } maybeAutoTriggerReviewerAfterOwnerCompletion({ task, taskId, diff --git a/src/paired-execution-context-shared.ts b/src/paired-execution-context-shared.ts index 8bf11fc..6e59957 100644 --- a/src/paired-execution-context-shared.ts +++ b/src/paired-execution-context-shared.ts @@ -3,17 +3,12 @@ import { execFileSync } from 'child_process'; import { isArbiterEnabled } from './config.js'; import { updatePairedTaskIfUnchanged } from './db.js'; import { logger } from './logger.js'; +import { + parseVisibleVerdict, + type VisibleVerdict, +} from './paired-verdict.js'; import type { PairedTaskStatus } from './types.js'; -export type VisibleVerdict = - | 'step_done' - | 'task_done' - | 'done' - | 'done_with_concerns' - | 'blocked' - | 'needs_context' - | 'continue'; - export type CompletionSignal = | { kind: 'request_reviewer'; resetStatusToActive: boolean } | { kind: 'request_owner_finalize' } @@ -23,24 +18,8 @@ export type CompletionSignal = | { kind: 'complete'; completionReason: 'done' | 'escalated' } | { kind: 'preserve_review_ready' }; -export function parseVisibleVerdict( - summary: string | null | undefined, -): VisibleVerdict { - 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}STEP_DONE\*{0,2}\b/i.test(firstLine)) return 'step_done'; - if (/^\*{0,2}TASK_DONE\*{0,2}\b/i.test(firstLine)) return 'task_done'; - 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 { parseVisibleVerdict }; +export type { VisibleVerdict }; export function resolveOwnerCompletionSignal(args: { phase: 'normal' | 'finalize'; @@ -270,6 +249,10 @@ export function transitionPairedTaskStatus(args: { review_requested_at?: string | null; round_trip_count?: number; owner_failure_count?: number; + owner_step_done_streak?: number; + finalize_step_done_count?: number; + task_done_then_user_reopen_count?: number; + empty_step_done_streak?: number; arbiter_verdict?: string | null; arbiter_requested_at?: string | null; completion_reason?: string | null; @@ -314,6 +297,10 @@ export function applyPairedTaskPatch(args: { review_requested_at?: string | null; round_trip_count?: number; owner_failure_count?: number; + owner_step_done_streak?: number; + finalize_step_done_count?: number; + task_done_then_user_reopen_count?: number; + empty_step_done_streak?: number; status?: PairedTaskStatus; arbiter_verdict?: string | null; arbiter_requested_at?: string | null; @@ -355,6 +342,10 @@ export function requestArbiterOrEscalate(args: { review_requested_at?: string | null; round_trip_count?: number; owner_failure_count?: number; + owner_step_done_streak?: number; + finalize_step_done_count?: number; + task_done_then_user_reopen_count?: number; + empty_step_done_streak?: number; arbiter_verdict?: string | null; arbiter_requested_at?: string | null; completion_reason?: string | null; diff --git a/src/paired-execution-context.test.ts b/src/paired-execution-context.test.ts index e3f29bc..c7a6df2 100644 --- a/src/paired-execution-context.test.ts +++ b/src/paired-execution-context.test.ts @@ -143,6 +143,10 @@ function buildPairedTask(overrides: Partial = {}): PairedTask { review_requested_at: null, round_trip_count: 0, owner_failure_count: 0, + owner_step_done_streak: 0, + finalize_step_done_count: 0, + task_done_then_user_reopen_count: 0, + empty_step_done_streak: 0, status: 'active', arbiter_verdict: null, arbiter_requested_at: null, @@ -293,6 +297,62 @@ describe('paired execution context', () => { expect(db.insertPairedTurnOutput).not.toHaveBeenCalled(); }); + it('records a quick reopen when a new owner task starts shortly after TASK_DONE completion', () => { + const previousTask = buildPairedTask({ + id: 'task-completed', + status: 'completed', + completion_reason: 'done', + updated_at: new Date(Date.now() - 60_000).toISOString(), + task_done_then_user_reopen_count: 0, + }); + vi.mocked(db.getLatestOpenPairedTaskForChat).mockReturnValue(undefined); + vi.mocked(db.getLatestPairedTaskForChat).mockReturnValue(previousTask); + + const result = resolveOwnerTaskForHumanMessage({ + group, + chatJid: 'dc:test', + roomRoleContext: ownerContext, + existingTask: null, + }); + + expect(result.task).not.toBeNull(); + expect(db.updatePairedTask).toHaveBeenCalledWith( + 'task-completed', + expect.objectContaining({ + task_done_then_user_reopen_count: 1, + }), + ); + }); + + it('resets active STEP_DONE loop counters when a new human message continues an active task', () => { + vi.mocked(db.getLatestOpenPairedTaskForChat).mockReturnValue( + buildPairedTask({ + status: 'active', + owner_failure_count: 1, + owner_step_done_streak: 2, + empty_step_done_streak: 2, + }), + ); + + preparePairedExecutionContext({ + group, + chatJid: 'dc:test', + runId: 'run-human-reset-step-done-loop', + roomRoleContext: ownerContext, + hasHumanMessage: true, + }); + + expect(db.updatePairedTask).toHaveBeenCalledWith( + 'task-1', + expect.objectContaining({ + round_trip_count: 0, + owner_failure_count: 0, + owner_step_done_streak: 0, + empty_step_done_streak: 0, + }), + ); + }); + it('uses room role context agent overrides when creating a paired task', () => { preparePairedExecutionContext({ group, @@ -761,6 +821,7 @@ describe('paired execution context', () => { 'task-1', expect.objectContaining({ owner_failure_count: 0, + owner_step_done_streak: 1, }), ); expect( @@ -774,6 +835,43 @@ describe('paired execution context', () => { ); }); + it('requests arbiter when active STEP_DONE repeats without code changes', () => { + vi.spyOn(config, 'isArbiterEnabled').mockReturnValue(true); + + const repoDir = createCanonicalRepoWithCommit('active step done loop'); + const sourceRef = resolveTreeRef(repoDir); + vi.mocked(db.getPairedTaskById).mockReturnValue( + buildPairedTask({ + status: 'active', + source_ref: sourceRef, + owner_step_done_streak: 2, + empty_step_done_streak: 2, + }), + ); + vi.mocked(db.getPairedWorkspace).mockImplementation((_taskId, role) => + role === 'owner' ? buildWorkspace('owner', repoDir) : undefined, + ); + + completePairedExecutionContext({ + taskId: 'task-1', + role: 'owner', + status: 'succeeded', + summary: 'STEP_DONE\n요약만 반복되고 코드 변경은 없음', + }); + + expect(db.updatePairedTask).toHaveBeenCalledWith( + 'task-1', + expect.objectContaining({ + status: 'arbiter_requested', + owner_step_done_streak: 3, + empty_step_done_streak: 3, + }), + ); + expect( + pairedWorkspaceManager.markPairedTaskReviewReady, + ).not.toHaveBeenCalled(); + }); + it('returns merge_ready owner finalize output to active when the owner reports STEP_DONE', () => { vi.mocked(db.getPairedTaskById).mockReturnValue( buildPairedTask({ @@ -794,6 +892,44 @@ describe('paired execution context', () => { expect.objectContaining({ status: 'active', owner_failure_count: 0, + finalize_step_done_count: 1, + }), + ); + expect( + pairedWorkspaceManager.markPairedTaskReviewReady, + ).not.toHaveBeenCalled(); + }); + + it('requests arbiter when finalize STEP_DONE repeats without code changes', () => { + vi.spyOn(config, 'isArbiterEnabled').mockReturnValue(true); + + const repoDir = createCanonicalRepoWithCommit('reviewed'); + const approvedSourceRef = resolveTreeRef(repoDir); + vi.mocked(db.getPairedTaskById).mockReturnValue( + buildPairedTask({ + status: 'merge_ready', + source_ref: approvedSourceRef, + empty_step_done_streak: 1, + finalize_step_done_count: 1, + }), + ); + vi.mocked(db.getPairedWorkspace).mockImplementation((_taskId, role) => + role === 'owner' ? buildWorkspace('owner', repoDir) : undefined, + ); + + completePairedExecutionContext({ + taskId: 'task-1', + role: 'owner', + status: 'succeeded', + summary: 'STEP_DONE\n아직 남았지만 코드 변경은 없음', + }); + + expect(db.updatePairedTask).toHaveBeenCalledWith( + 'task-1', + expect.objectContaining({ + status: 'arbiter_requested', + empty_step_done_streak: 2, + finalize_step_done_count: 2, }), ); expect( diff --git a/src/paired-execution-context.ts b/src/paired-execution-context.ts index 45038e3..3732eee 100644 --- a/src/paired-execution-context.ts +++ b/src/paired-execution-context.ts @@ -30,6 +30,7 @@ import { hasActiveCiWatcherForChat, insertPairedTurnOutput, releasePairedTaskExecutionLease, + updatePairedTask, upsertPairedProject, } from './db.js'; import { logger } from './logger.js'; @@ -72,6 +73,8 @@ import type { } from './types.js'; import { resolveRoleAgentPlan } from './role-agent-plan.js'; +const TASK_DONE_REOPEN_WINDOW_MS = 10 * 60_000; + function ensurePairedProject( group: RegisteredGroup, chatJid: string, @@ -144,6 +147,10 @@ function createActiveTaskForRoom(args: { review_requested_at: null, round_trip_count: 0, owner_failure_count: 0, + owner_step_done_streak: 0, + finalize_step_done_count: 0, + task_done_then_user_reopen_count: 0, + empty_step_done_streak: 0, status: 'active', arbiter_verdict: null, arbiter_requested_at: null, @@ -164,6 +171,39 @@ function createActiveTaskForRoom(args: { return task; } +function maybeRecordTaskDoneReopen(previousTask: PairedTask | null): void { + if ( + !previousTask || + previousTask.status !== 'completed' || + previousTask.completion_reason !== 'done' + ) { + return; + } + + const completedAt = Date.parse(previousTask.updated_at); + if (!Number.isFinite(completedAt)) { + return; + } + if (Date.now() - completedAt > TASK_DONE_REOPEN_WINDOW_MS) { + return; + } + + updatePairedTask(previousTask.id, { + task_done_then_user_reopen_count: + (previousTask.task_done_then_user_reopen_count ?? 0) + 1, + updated_at: new Date().toISOString(), + }); + logger.info( + { + taskId: previousTask.id, + chatJid: previousTask.chat_jid, + reopenCount: (previousTask.task_done_then_user_reopen_count ?? 0) + 1, + completionReason: previousTask.completion_reason, + }, + 'Recorded paired task reopen shortly after TASK_DONE completion', + ); +} + function cancelOutstandingFinalizeOwnerTurn(task: PairedTask): void { const turnIdentity = buildPairedTurnIdentity({ taskId: task.id, @@ -253,6 +293,7 @@ export function resolveOwnerTaskForHumanMessage(args: { args.existingTask ?? getLatestOpenPairedTaskForChat(args.chatJid) ?? null; if (!existing) { + maybeRecordTaskDoneReopen(getLatestPairedTaskForChat(args.chatJid) ?? null); return { task: canonicalWorkDir ? createActiveTaskForRoom({ @@ -432,7 +473,12 @@ export function preparePairedExecutionContext(args: { updatedAt: now, patch: { ...(hasHuman - ? { round_trip_count: 0, owner_failure_count: 0 } + ? { + round_trip_count: 0, + owner_failure_count: 0, + owner_step_done_streak: 0, + empty_step_done_streak: 0, + } : {}), }, }); @@ -443,7 +489,12 @@ export function preparePairedExecutionContext(args: { updatedAt: now, patch: { ...(hasHuman - ? { round_trip_count: 0, owner_failure_count: 0 } + ? { + round_trip_count: 0, + owner_failure_count: 0, + owner_step_done_streak: 0, + empty_step_done_streak: 0, + } : {}), }, }); diff --git a/src/paired-verdict.ts b/src/paired-verdict.ts new file mode 100644 index 0000000..57e375d --- /dev/null +++ b/src/paired-verdict.ts @@ -0,0 +1,41 @@ +export type VisibleVerdict = + | 'step_done' + | 'task_done' + | 'done' + | 'done_with_concerns' + | 'blocked' + | 'needs_context' + | 'continue'; + +export function parseVisibleVerdict( + summary: string | null | undefined, +): VisibleVerdict { + 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}STEP_DONE\*{0,2}\b/i.test(firstLine)) return 'step_done'; + if (/^\*{0,2}TASK_DONE\*{0,2}\b/i.test(firstLine)) return 'task_done'; + 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 function resolveStoredVisibleVerdict(args: { + verdict?: VisibleVerdict | null; + outputText?: string | null; +}): VisibleVerdict | null { + if (args.verdict) { + return args.verdict; + } + if (!args.outputText) { + return null; + } + return parseVisibleVerdict(args.outputText); +} diff --git a/src/types.ts b/src/types.ts index 49842fa..5c00ed3 100644 --- a/src/types.ts +++ b/src/types.ts @@ -1,3 +1,5 @@ +import type { VisibleVerdict } from './paired-verdict.js'; + export interface AgentConfig { timeout?: number; // Default: 300000 (5 minutes) // Per-group model/effort overrides (take precedence over global env vars) @@ -83,6 +85,10 @@ export interface PairedTask { review_requested_at: string | null; round_trip_count: number; owner_failure_count?: number | null; + owner_step_done_streak?: number | null; + finalize_step_done_count?: number | null; + task_done_then_user_reopen_count?: number | null; + empty_step_done_streak?: number | null; status: PairedTaskStatus; arbiter_verdict: string | null; arbiter_requested_at: string | null; @@ -97,6 +103,7 @@ export interface PairedTurnOutput { turn_number: number; role: PairedRoomRole; output_text: string; + verdict?: VisibleVerdict | null; created_at: string; }