feat(conductor): task state machine — TIMED_OUT state and retriable steps
- Add 'timed_out' to flow_runs/flow_steps CHECK constraints - Add retry_count and max_retries columns to flow_steps - Add timeout detection in advanceInner loop (configurable FLOW_STEP_TIMEOUT_MS) - Add retriable logic: re-dispatch on timeout if maxRetries > 0 and retryCount < maxRetries - Add isRetriable() + shouldRetry() pure decision functions - Add timed_out handling to reconcileResumeStep and reconcileRun - Add 'timed_out' to ws-frames enum, publishStep status type
This commit is contained in:
@@ -100,6 +100,9 @@ interface FlowStepRow {
|
||||
status: string;
|
||||
chat_id: string | null;
|
||||
output: string | null;
|
||||
updated_at: string | null;
|
||||
retry_count: number | null;
|
||||
max_retries: number | null;
|
||||
}
|
||||
|
||||
export function createFlowRunner(deps: Deps): FlowRunner {
|
||||
@@ -263,7 +266,8 @@ export function createFlowRunner(deps: Deps): FlowRunner {
|
||||
const dispatch: DispatchFn = (agent, task) => dispatchSubAgent(run.project_id, model, agent, task);
|
||||
|
||||
const rows = await sql<FlowStepRow[]>`
|
||||
SELECT step_id, kind, agent, status, chat_id, output FROM flow_steps WHERE run_id = ${runId}
|
||||
SELECT step_id, kind, agent, status, chat_id, output, updated_at, retry_count, max_retries
|
||||
FROM flow_steps WHERE run_id = ${runId}
|
||||
`;
|
||||
|
||||
// Re-derive the excluded set (band/when pre-skips) from the flow def + input —
|
||||
@@ -275,6 +279,7 @@ export function createFlowRunner(deps: Deps): FlowRunner {
|
||||
const done = new Set<string>();
|
||||
const skipped = new Set<string>();
|
||||
const inFlight = new Set<string>();
|
||||
const timedOut = new Set<string>();
|
||||
const results: Record<string, string> = {};
|
||||
for (const r of rows) {
|
||||
switch (r.status) {
|
||||
@@ -288,6 +293,9 @@ export function createFlowRunner(deps: Deps): FlowRunner {
|
||||
case 'running':
|
||||
inFlight.add(r.step_id);
|
||||
break;
|
||||
case 'timed_out':
|
||||
timedOut.add(r.step_id);
|
||||
break;
|
||||
case 'failed':
|
||||
// A failed worker makes the deterministic report untrustworthy — fail the
|
||||
// whole run (matches the Phase-1 CLI, which throws on a dispatch failure).
|
||||
@@ -300,10 +308,68 @@ export function createFlowRunner(deps: Deps): FlowRunner {
|
||||
}
|
||||
}
|
||||
|
||||
// ─── Timeout detection ───────────────────────────────────────────────────────
|
||||
// Check running steps. If a step has been 'running' longer than
|
||||
// FLOW_STEP_TIMEOUT_MS, mark it timed_out or re-dispatch if retriable.
|
||||
const timeoutMs = config.FLOW_STEP_TIMEOUT_MS;
|
||||
const nowDate = new Date();
|
||||
let detectedTimedOut = false;
|
||||
for (const r of rows) {
|
||||
if (r.status !== 'running') continue;
|
||||
if (!r.updated_at) continue;
|
||||
const elapsed = nowDate.getTime() - new Date(r.updated_at).getTime();
|
||||
if (elapsed <= timeoutMs) continue;
|
||||
|
||||
// Step has exceeded the timeout
|
||||
detectedTimedOut = true;
|
||||
const retryCount = r.retry_count ?? 0;
|
||||
const maxRetries = r.max_retries ?? 0;
|
||||
|
||||
if (maxRetries > 0 && retryCount < maxRetries) {
|
||||
// Retriable: re-dispatch the step with an incremented retry_count
|
||||
const step = flow.steps.find((s) => s.id === r.step_id);
|
||||
if (!step || step.kind !== 'agent') {
|
||||
// Non-agent steps can't be retried via dispatch
|
||||
inFlight.delete(r.step_id);
|
||||
await failRun(runId, flow, input, model,
|
||||
`step '${r.step_id}' timed out (non-retriable kind)`, r.step_id);
|
||||
return;
|
||||
}
|
||||
inFlight.delete(r.step_id);
|
||||
await sql`
|
||||
UPDATE flow_steps
|
||||
SET retry_count = ${retryCount + 1}, updated_at = clock_timestamp()
|
||||
WHERE run_id = ${runId} AND step_id = ${r.step_id} AND status = 'running'
|
||||
`;
|
||||
await dispatchAgentStep(runId, run.project_id, model, step, ctx);
|
||||
inFlight.add(r.step_id);
|
||||
log.warn({ runId, stepId: r.step_id, retry: retryCount + 1, maxRetries },
|
||||
'flow-runner: step timed out, retrying');
|
||||
} else {
|
||||
// Not retriable — mark as timed_out, fail the run
|
||||
inFlight.delete(r.step_id);
|
||||
await sql`
|
||||
UPDATE flow_steps SET status = 'timed_out', updated_at = clock_timestamp()
|
||||
WHERE run_id = ${runId} AND step_id = ${r.step_id} AND status = 'running'
|
||||
`;
|
||||
timedOut.add(r.step_id);
|
||||
publishStep(runId, r.step_id, 'timed_out');
|
||||
await failRun(runId, flow, input, model,
|
||||
`step '${r.step_id}' timed out`, r.step_id);
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
// If we modified any steps, re-query so the state sets reflect the latest DB.
|
||||
if (detectedTimedOut) {
|
||||
// Continue with the in-memory state we already adjusted above (inFlight/timedOut
|
||||
// were mutated directly). No re-query needed.
|
||||
}
|
||||
|
||||
// Drain ready skips + code steps (synchronous), re-evaluating after each batch,
|
||||
// then dispatch the full ready agent wave and wait for their terminal callbacks.
|
||||
for (;;) {
|
||||
const state: SchedulerState = { done, skipped, inFlight, excluded };
|
||||
const state: SchedulerState = { done, skipped, inFlight, excluded, timedOut };
|
||||
|
||||
if (isRunComplete(flow, state)) {
|
||||
await finishRun(runId, flow, input, results, model, dispatch);
|
||||
@@ -545,7 +611,7 @@ export function createFlowRunner(deps: Deps): FlowRunner {
|
||||
function publishStep(
|
||||
runId: string,
|
||||
stepId: string,
|
||||
status: 'running' | 'completed' | 'failed' | 'skipped' | 'cancelled' | 'blocked',
|
||||
status: 'running' | 'completed' | 'failed' | 'skipped' | 'cancelled' | 'blocked' | 'timed_out',
|
||||
extra?: { run_status?: 'running' | 'completed' | 'failed' | 'cancelled'; report?: string },
|
||||
): void {
|
||||
publishUser({
|
||||
@@ -683,6 +749,38 @@ export function createFlowRunner(deps: Deps): FlowRunner {
|
||||
log.info({ runId, stepId: step.step_id, taskId: task!.id }, 'flow-runner: step re-dispatched on resume');
|
||||
break;
|
||||
}
|
||||
|
||||
case 'retry': {
|
||||
// Like re-dispatch but increments retry_count and sets status to 'running'.
|
||||
if (!step.input) {
|
||||
await sql`
|
||||
UPDATE flow_steps
|
||||
SET status = 'failed', error = 'retry: no stored prompt',
|
||||
updated_at = clock_timestamp()
|
||||
WHERE run_id = ${runId} AND step_id = ${step.step_id}
|
||||
`;
|
||||
break;
|
||||
}
|
||||
const chatIdR = step.chat_id;
|
||||
const [chatR] = chatIdR
|
||||
? await sql<{ session_id: string }[]>`SELECT session_id FROM chats WHERE id = ${chatIdR}`
|
||||
: [];
|
||||
const sessionIdR = chatR?.session_id ?? null;
|
||||
const [taskR] = await sql<{ id: string }[]>`
|
||||
INSERT INTO tasks (project_id, input, agent, model, mode_id, session_id, chat_id)
|
||||
VALUES (${projectId}, ${step.input}, 'qwen', ${model}, 'plan', ${sessionIdR}, ${chatIdR})
|
||||
RETURNING id
|
||||
`;
|
||||
await sql`
|
||||
UPDATE flow_steps
|
||||
SET task_id = ${taskR!.id}, retry_count = retry_count + 1, status = 'running',
|
||||
updated_at = clock_timestamp()
|
||||
WHERE run_id = ${runId} AND step_id = ${step.step_id}
|
||||
`;
|
||||
log.info({ runId, stepId: step.step_id, taskId: taskR!.id },
|
||||
'flow-runner: step retried on resume');
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -697,7 +795,9 @@ export function createFlowRunner(deps: Deps): FlowRunner {
|
||||
status: string;
|
||||
chat_id: string | null;
|
||||
input: string | null;
|
||||
}[]>`SELECT step_id, task_id, status, chat_id, input FROM flow_steps WHERE run_id = ${run.id}`;
|
||||
retry_count: number | null;
|
||||
max_retries: number | null;
|
||||
}[]>`SELECT step_id, task_id, status, chat_id, input, retry_count, max_retries FROM flow_steps WHERE run_id = ${run.id}`;
|
||||
|
||||
// Load task states for all referenced tasks in one query.
|
||||
const taskIds = rows.map((r) => r.task_id).filter((id): id is string => id !== null);
|
||||
@@ -710,7 +810,13 @@ export function createFlowRunner(deps: Deps): FlowRunner {
|
||||
}
|
||||
|
||||
const decisions = reconcileRun(
|
||||
rows.map((r) => ({ stepId: r.step_id, taskId: r.task_id, status: r.status })),
|
||||
rows.map((r) => ({
|
||||
stepId: r.step_id,
|
||||
taskId: r.task_id,
|
||||
status: r.status,
|
||||
retryCount: r.retry_count ?? undefined,
|
||||
maxRetries: r.max_retries,
|
||||
})),
|
||||
taskStates,
|
||||
);
|
||||
|
||||
@@ -752,13 +858,13 @@ export function createFlowRunner(deps: Deps): FlowRunner {
|
||||
// Mark all non-terminal steps cancelled and collect in-flight task_ids.
|
||||
const steps = await sql<{ step_id: string; task_id: string | null; kind: string }[]>`
|
||||
SELECT step_id, task_id, kind FROM flow_steps
|
||||
WHERE run_id = ${runId} AND status NOT IN ('completed', 'failed', 'cancelled', 'skipped')
|
||||
WHERE run_id = ${runId} AND status NOT IN ('completed', 'failed', 'cancelled', 'skipped', 'timed_out')
|
||||
`;
|
||||
|
||||
if (steps.length > 0) {
|
||||
await sql`
|
||||
UPDATE flow_steps SET status = 'cancelled', updated_at = clock_timestamp()
|
||||
WHERE run_id = ${runId} AND status NOT IN ('completed', 'failed', 'cancelled', 'skipped')
|
||||
WHERE run_id = ${runId} AND status NOT IN ('completed', 'failed', 'cancelled', 'skipped', 'timed_out')
|
||||
`;
|
||||
for (const s of steps) {
|
||||
if (s.kind === 'agent') publishStep(runId, s.step_id, 'cancelled', { run_status: 'cancelled' });
|
||||
|
||||
Reference in New Issue
Block a user