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

feat: auto-suspend tasks on quota errors, add Claude progress tracking

Browse Source 
- Auto-suspend tasks after 3 consecutive quota/auth errors with
  retry-after date parsing and Discord notification
- Add tool_progress and tool_use_summary streaming from Claude Agent SDK
- Prefix progress messages with invisible marker so bots ignore them
- Fix misleading provider label in codex service logs
This commit is contained in:
Eyejoker
2026-03-23 17:03:17 +09:00
parent 42d721a9ec
commit 77e79505eb
8 changed files with 320 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files

71
runners/agent-runner/src/index.ts
View File

@@ -32,6 +32,7 @@ interface ContainerInput {
interface ContainerOutput {
status: 'success' | 'error';
phase?: 'progress' | 'final';
result: string | null;
newSessionId?: string;
error?: string;
@@ -59,6 +60,11 @@ interface SDKUserMessage {
session_id: string;
}
interface AssistantContentBlock {
type?: string;
text?: string;
}
// Paths configurable via env vars.
const GROUP_DIR = process.env.EJCLAW_GROUP_DIR || '/workspace/group';
const IPC_DIR = process.env.EJCLAW_IPC_DIR || '/workspace/ipc';
@@ -174,6 +180,25 @@ function log(message: string): void {
console.error(`[agent-runner] ${message}`);
}
function extractAssistantText(message: unknown): string | null {
const assistant = message as {
message?: {
content?: AssistantContentBlock[];
};
};
const blocks = assistant.message?.content;
if (!Array.isArray(blocks)) return null;
const text = blocks
.filter((block) => block?.type === 'text' && typeof block.text === 'string')
.map((block) => block.text!.trim())
.filter(Boolean)
.join('\n\n')
.trim();
return text || null;
}
function getSessionSummary(sessionId: string, transcriptPath: string): string | null {
const projectDir = path.dirname(transcriptPath);
const indexPath = path.join(projectDir, 'sessions-index.json');
@@ -525,6 +550,32 @@ async function runQuery(
log(`Task notification: task=${tn.task_id} status=${tn.status} summary=${tn.summary}`);
}
if (message.type === 'tool_progress') {
const tp = message as {
tool_name: string;
elapsed_time_seconds: number;
};
const label = `${tp.tool_name} (${Math.round(tp.elapsed_time_seconds)}s)`;
log(`Tool progress: ${label}`);
writeOutput({
status: 'success',
phase: 'progress',
result: label,
newSessionId,
});
}
if (message.type === 'tool_use_summary') {
const ts = message as { summary: string };
log(`Tool use summary: ${ts.summary.slice(0, 200)}`);
writeOutput({
status: 'success',
phase: 'progress',
result: ts.summary,
newSessionId,
});
}
if (message.type === 'result') {
resultCount++;
const textResult = 'result' in message ? (message as { result?: string }).result : null;
@@ -569,6 +620,26 @@ async function runQuery(
log('Terminal result observed, ending query stream');
break;
}
if (message.type === 'assistant') {
const stopReason = (message as { stop_reason?: string }).stop_reason;
const textResult = extractAssistantText(message);
if (stopReason === 'end_turn' && textResult) {
resultCount++;
log(
`Terminal assistant turn observed without result event (${textResult.length} chars), ending query stream`,
);
writeOutput({
status: 'success',
result: textResult,
newSessionId,
});
terminalResultObserved = true;
ipcPolling = false;
stream.end();
break;
}
}
}
ipcPolling = false;

41
src/db.ts
View File

@@ -214,6 +214,14 @@ function createSchema(database: Database.Database): void {
/* column already exists */
}
try {
database.exec(
`ALTER TABLE scheduled_tasks ADD COLUMN suspended_until TEXT`,
);
} catch {
/* column already exists */
}
database.exec(`
UPDATE scheduled_tasks
SET agent_type = COALESCE(
@@ -928,7 +936,12 @@ export function updateTask(
updates: Partial<
Pick<
ScheduledTask,
'prompt' | 'schedule_type' | 'schedule_value' | 'next_run' | 'status'
| 'prompt'
| 'schedule_type'
| 'schedule_value'
| 'next_run'
| 'status'
| 'suspended_until'
>
>,
): void {
@@ -955,6 +968,10 @@ export function updateTask(
fields.push('status = ?');
values.push(updates.status);
}
if (updates.suspended_until !== undefined) {
fields.push('suspended_until = ?');
values.push(updates.suspended_until);
}
if (fields.length === 0) return;
@@ -1005,10 +1022,11 @@ export function getDueTasks(
`
SELECT * FROM scheduled_tasks
WHERE status = 'active' AND agent_type = ? AND next_run IS NOT NULL AND next_run <= ?
AND (suspended_until IS NULL OR suspended_until <= ?)
ORDER BY next_run
`,
)
.all(agentType, now) as ScheduledTask[];
.all(agentType, now, now) as ScheduledTask[];
}
export function updateTaskAfterRun(
@@ -1042,6 +1060,25 @@ export function logTaskRun(log: TaskRunLog): void {
);
}
export function getRecentConsecutiveErrors(
taskId: string,
limit: number = 5,
): string[] {
const rows = db
.prepare(
`SELECT status, error FROM task_run_logs
WHERE task_id = ? ORDER BY run_at DESC LIMIT ?`,
)
.all(taskId, limit) as Array<{ status: string; error: string | null }>;
const errors: string[] = [];
for (const row of rows) {
if (row.status !== 'error' || !row.error) break;
errors.push(row.error);
}
return errors;
}
// --- Router state accessors ---
export function getRouterState(key: string): string | undefined {

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

@@ -167,17 +167,19 @@ export async function runAgentForGroup(
);
}
const agentType = group.agentType || 'claude-code';
const providerLabel = canFallback ? provider : agentType;
logger.info(
{
chatJid,
group: group.name,
groupFolder: group.folder,
runId,
provider,
provider: providerLabel,
canFallback,
groupHasOverride,
},
`Using provider: ${provider}`,
`Using provider: ${providerLabel}`,
);
try {

59
src/message-runtime.test.ts
View File

@@ -1,5 +1,10 @@
import { beforeEach, describe, expect, it, vi } from 'vitest';
import { TASK_STATUS_MESSAGE_PREFIX } from './task-watch-status.js';
/** Prefix helper for progress message assertions */
const P = (text: string) => `${TASK_STATUS_MESSAGE_PREFIX}${text}`;
vi.mock('./agent-runner.js', () => ({
runAgentProcess: vi.fn(),
writeGroupsSnapshot: vi.fn(),
@@ -686,17 +691,17 @@ describe('createMessageRuntime', () => {
expect(result).toBe(true);
expect(channel.sendAndTrack).toHaveBeenCalledWith(
chatJid,
'CI 상태 확인 중입니다.\n\n0초',
P('CI 상태 확인 중입니다.\n\n0초'),
);
expect(channel.editMessage).toHaveBeenCalledWith(
chatJid,
'progress-1',
'CI 상태 확인 중입니다.\n\n10초',
P('CI 상태 확인 중입니다.\n\n10초'),
);
expect(channel.editMessage).toHaveBeenCalledWith(
chatJid,
'progress-1',
'CI 상태 확인 중입니다.\n\n10초',
P('CI 상태 확인 중입니다.\n\n10초'),
);
expect(channel.sendMessage).toHaveBeenCalledTimes(1);
expect(channel.sendMessage).toHaveBeenCalledWith(
@@ -779,11 +784,11 @@ describe('createMessageRuntime', () => {
expect(result).toBe(true);
expect(channel.sendAndTrack).toHaveBeenCalledWith(
chatJid,
'진행 중입니다.\n\n0초',
P('진행 중입니다.\n\n0초'),
);
expect(channel.sendMessage).toHaveBeenCalledWith(
chatJid,
'진행 중입니다.\n\n0초',
P('진행 중입니다.\n\n0초'),
);
});
@@ -851,11 +856,11 @@ describe('createMessageRuntime', () => {
expect(result).toBe(true);
expect(channel.sendAndTrack).toHaveBeenCalledWith(
chatJid,
'진행 중입니다.\n\n0초',
P('진행 중입니다.\n\n0초'),
);
expect(channel.sendMessage).toHaveBeenCalledWith(
chatJid,
'진행 중입니다.\n\n0초',
P('진행 중입니다.\n\n0초'),
);
});
@@ -949,17 +954,17 @@ describe('createMessageRuntime', () => {
expect(channel.sendAndTrack).toHaveBeenCalledTimes(1);
expect(channel.sendAndTrack).toHaveBeenCalledWith(
chatJid,
'첫 번째 진행상황입니다.\n\n0초',
P('첫 번째 진행상황입니다.\n\n0초'),
);
expect(channel.editMessage).toHaveBeenCalledWith(
chatJid,
'progress-1',
'첫 번째 진행상황입니다.\n\n10초',
P('첫 번째 진행상황입니다.\n\n10초'),
);
expect(channel.editMessage).not.toHaveBeenCalledWith(
chatJid,
'progress-1',
'늦게 도착한 진행상황입니다.\n\n0초',
P('늦게 도착한 진행상황입니다.\n\n0초'),
);
expect(channel.editMessage).not.toHaveBeenCalledWith(
chatJid,
@@ -1049,17 +1054,17 @@ describe('createMessageRuntime', () => {
expect(result).toBe(true);
expect(channel.sendAndTrack).toHaveBeenCalledWith(
chatJid,
'오래 걸리는 작업입니다.\n\n0초',
P('오래 걸리는 작업입니다.\n\n0초'),
);
expect(channel.editMessage).toHaveBeenCalledWith(
chatJid,
'progress-1',
'오래 걸리는 작업입니다.\n\n1시간 0초',
P('오래 걸리는 작업입니다.\n\n1시간 0초'),
);
expect(channel.editMessage).toHaveBeenCalledWith(
chatJid,
'progress-1',
'오래 걸리는 작업입니다.\n\n1시간 10초',
P('오래 걸리는 작업입니다.\n\n1시간 10초'),
);
expect(channel.sendMessage).toHaveBeenCalledWith(
chatJid,
@@ -1142,17 +1147,17 @@ describe('createMessageRuntime', () => {
expect(result).toBe(true);
expect(channel.sendAndTrack).toHaveBeenCalledWith(
chatJid,
'테스트를 돌리는 중입니다.\n\n0초',
P('테스트를 돌리는 중입니다.\n\n0초'),
);
expect(channel.editMessage).toHaveBeenCalledWith(
chatJid,
'progress-1',
'테스트를 돌리는 중입니다.\n\n10초',
P('테스트를 돌리는 중입니다.\n\n10초'),
);
expect(channel.editMessage).toHaveBeenCalledWith(
chatJid,
'progress-1',
'테스트를 돌리는 중입니다.\n\n10초',
P('테스트를 돌리는 중입니다.\n\n10초'),
);
expect(channel.sendMessage).toHaveBeenCalledTimes(1);
expect(channel.sendMessage).toHaveBeenCalledWith(
@@ -1313,29 +1318,29 @@ describe('createMessageRuntime', () => {
expect(channel.sendAndTrack).toHaveBeenNthCalledWith(
1,
chatJid,
'첫 번째 진행상황입니다.\n\n0초',
P('첫 번째 진행상황입니다.\n\n0초'),
);
expect(channel.sendAndTrack).toHaveBeenNthCalledWith(
2,
chatJid,
'두 번째 진행상황입니다.\n\n0초',
P('두 번째 진행상황입니다.\n\n0초'),
);
expect(channel.editMessage).toHaveBeenNthCalledWith(
1,
chatJid,
'progress-1',
'첫 번째 진행상황입니다.\n\n10초',
P('첫 번째 진행상황입니다.\n\n10초'),
);
expect(channel.editMessage).toHaveBeenNthCalledWith(
2,
chatJid,
'progress-2',
'두 번째 진행상황입니다.\n\n10초',
P('두 번째 진행상황입니다.\n\n10초'),
);
expect(channel.editMessage).toHaveBeenCalledWith(
chatJid,
'progress-2',
'두 번째 진행상황입니다.\n\n10초',
P('두 번째 진행상황입니다.\n\n10초'),
);
expect(channel.sendMessage).toHaveBeenCalledTimes(1);
expect(channel.sendMessage).toHaveBeenCalledWith(
@@ -1425,17 +1430,17 @@ describe('createMessageRuntime', () => {
expect(result).toBe(true);
expect(channel.sendAndTrack).toHaveBeenCalledWith(
chatJid,
'검증 중입니다.\n\n0초',
P('검증 중입니다.\n\n0초'),
);
expect(channel.editMessage).toHaveBeenCalledWith(
chatJid,
'progress-1',
'커밋은 정상 들어갔고 pre-commit도 통과했습니다.\n\n10초',
P('커밋은 정상 들어갔고 pre-commit도 통과했습니다.\n\n10초'),
);
expect(channel.editMessage).toHaveBeenCalledWith(
chatJid,
'progress-1',
'커밋은 정상 들어갔고 pre-commit도 통과했습니다.\n\n10초',
P('커밋은 정상 들어갔고 pre-commit도 통과했습니다.\n\n10초'),
);
expect(channel.sendMessage).toHaveBeenCalledTimes(1);
expect(channel.sendMessage).toHaveBeenCalledWith(
@@ -1531,7 +1536,7 @@ describe('createMessageRuntime', () => {
expect(channel.sendAndTrack).toHaveBeenCalledTimes(1);
expect(channel.sendAndTrack).toHaveBeenCalledWith(
chatJid,
'진행 중입니다.\n\n0초',
P('진행 중입니다.\n\n0초'),
);
// 같은 progress 메시지는 유지하고, final은 별도 메시지로 보낸다.
expect(channel.editMessage).toHaveBeenCalledWith(
@@ -1542,7 +1547,7 @@ describe('createMessageRuntime', () => {
expect(channel.editMessage).toHaveBeenLastCalledWith(
chatJid,
'progress-1',
'아직 진행 중.\n\n10초',
P('아직 진행 중.\n\n10초'),
);
expect(channel.sendMessage).toHaveBeenCalledTimes(1);
expect(channel.sendMessage).toHaveBeenCalledWith(
@@ -1764,7 +1769,7 @@ describe('createMessageRuntime', () => {
expect(result).toBe(true);
expect(channel.sendAndTrack).toHaveBeenCalledWith(
chatJid,
'중간 진행상황입니다.\n\n0초',
P('중간 진행상황입니다.\n\n0초'),
);
expect(channel.sendMessage).toHaveBeenCalledTimes(1);
expect(channel.sendMessage).toHaveBeenCalledWith(

3
src/message-turn-controller.ts
View File

@@ -2,6 +2,7 @@ import { type AgentOutput } from './agent-runner.js';
import { logger } from './logger.js';
import { formatOutbound } from './router.js';
import { shouldResetSessionOnAgentFailure } from './session-recovery.js';
import { TASK_STATUS_MESSAGE_PREFIX } from './task-watch-status.js';
import { type Channel, type RegisteredGroup } from './types.js';
export type VisiblePhase = 'silent' | 'progress' | 'final';
@@ -211,7 +212,7 @@ export class MessageTurnController {
if (minutes > 0) elapsedParts.push(`${minutes}분`);
elapsedParts.push(`${seconds}초`);
return `${text}\n\n${elapsedParts.join(' ')}`;
return `${TASK_STATUS_MESSAGE_PREFIX}${text}\n\n${elapsedParts.join(' ')}`;
}
private clearProgressTicker(): void {

28
src/task-scheduler.ts
View File

@@ -29,6 +29,11 @@ import {
} from './group-folder.js';
import { logger } from './logger.js';
import { createTaskStatusTracker } from './task-status-tracker.js';
import {
evaluateTaskSuspension,
formatSuspensionNotice,
suspendTask,
} from './task-suspension.js';
import {
getTaskQueueJid,
getTaskRuntimeTaskId,
@@ -305,8 +310,31 @@ async function runTask(
return;
}
// Clear suspension on success
if (!error && currentTask.suspended_until) {
updateTask(task.id, { suspended_until: null });
}
// Check for repeated quota/auth errors → auto-suspend
let suspended = false;
if (error) {
const suspension = evaluateTaskSuspension(currentTask, error);
if (suspension.suspended && suspension.suspendedUntil) {
suspended = true;
suspendTask(task.id, suspension.suspendedUntil);
const notice = formatSuspensionNotice(
currentTask,
suspension.suspendedUntil,
suspension.reason || error.slice(0, 200),
);
await deps.sendMessage(task.chat_jid, notice);
}
}
if (error && !suspended) {
await statusTracker.update('retrying', nextRun);
} else if (suspended) {
// Don't update status tracker — task is suspended, not retrying
} else if (nextRun) {
await statusTracker.update('waiting', nextRun);
} else {

143
src/task-suspension.ts Normal file
View File

@@ -0,0 +1,143 @@
import { TIMEZONE } from './config.js';
import { getRecentConsecutiveErrors, updateTask } from './db.js';
import { logger } from './logger.js';
import type { ScheduledTask } from './types.js';
const CONSECUTIVE_FAILURES_THRESHOLD = 3;
const DEFAULT_BACKOFF_STEPS_MS = [
3_600_000, // 1 hour
14_400_000, // 4 hours
43_200_000, // 12 hours
];
// Patterns that indicate a quota / auth / billing error (not transient).
const QUOTA_PATTERNS = [
/usage limit/i,
/rate limit/i,
/quota exceeded/i,
/billing/i,
/insufficient.*(credit|fund|balance)/i,
/exceeded.*plan/i,
/purchase more credits/i,
];
export interface SuspensionResult {
suspended: boolean;
suspendedUntil: string | null;
reason: string | null;
}
/**
* Detect if a quota/auth error message contains a human-readable
* retry-after date like "try again at Mar 26th, 2026 9:00 AM".
*/
export function parseRetryAfterDate(error: string): Date | null {
const match = error.match(
/try again (?:at|after)\s+(.+?\d{4}[\s,]+\d{1,2}:\d{2}\s*(?:AM|PM)?)/i,
);
if (!match) return null;
const raw = match[1]
.replace(/(\d+)(?:st|nd|rd|th)/g, '$1')
.replace(/,\s*/g, ', ');
const parsed = new Date(raw);
if (Number.isNaN(parsed.getTime())) return null;
// Sanity: must be in the future and within 30 days
const now = Date.now();
if (parsed.getTime() <= now || parsed.getTime() > now + 30 * 86_400_000) {
return null;
}
return parsed;
}
function isQuotaError(error: string): boolean {
return QUOTA_PATTERNS.some((p) => p.test(error));
}
function computeBackoffMs(consecutiveCount: number): number {
const idx = Math.min(
consecutiveCount - CONSECUTIVE_FAILURES_THRESHOLD,
DEFAULT_BACKOFF_STEPS_MS.length - 1,
);
return DEFAULT_BACKOFF_STEPS_MS[Math.max(0, idx)];
}
/**
* Check recent run history and decide whether to suspend a task.
* Returns suspension info if the task should be suspended.
*/
export function evaluateTaskSuspension(
task: ScheduledTask,
latestError: string,
): SuspensionResult {
const noSuspension: SuspensionResult = {
suspended: false,
suspendedUntil: null,
reason: null,
};
if (!isQuotaError(latestError)) return noSuspension;
const recentErrors = getRecentConsecutiveErrors(
task.id,
CONSECUTIVE_FAILURES_THRESHOLD,
);
// +1 because the current error hasn't been logged yet
const totalConsecutive = recentErrors.length + 1;
if (totalConsecutive < CONSECUTIVE_FAILURES_THRESHOLD) return noSuspension;
// Try to parse an explicit retry-after date from the error
const retryDate = parseRetryAfterDate(latestError);
const suspendedUntil = retryDate
? retryDate.toISOString()
: new Date(Date.now() + computeBackoffMs(totalConsecutive)).toISOString();
return {
suspended: true,
suspendedUntil,
reason: latestError.slice(0, 200),
};
}
/**
* Apply suspension to a task in the DB.
*/
export function suspendTask(
taskId: string,
suspendedUntil: string,
): void {
updateTask(taskId, { suspended_until: suspendedUntil });
logger.info(
{ taskId, suspendedUntil },
'Task suspended due to repeated quota/auth errors',
);
}
/**
* Format the suspension notification for Discord.
*/
export function formatSuspensionNotice(
task: ScheduledTask,
suspendedUntil: string,
reason: string,
): string {
const resumeLabel = new Intl.DateTimeFormat('ko-KR', {
month: 'numeric',
day: 'numeric',
hour: '2-digit',
minute: '2-digit',
hour12: false,
timeZone: TIMEZONE,
}).format(new Date(suspendedUntil));
const lines = [
`⏸ 태스크 일시 중단`,
`- 사유: ${reason}`,
`- 자동 재개: ${resumeLabel}`,
`- 태스크 ID: \`${task.id}\``,
];
return lines.join('\n');
}

1
src/types.ts
View File

@@ -48,6 +48,7 @@ export interface ScheduledTask {
last_run: string | null;
last_result: string | null;
status: 'active' | 'paused' | 'completed';
suspended_until?: string | null;
created_at: string;
}