mirror of
https://github.com/tinyhumansai/openhuman.git
synced 2026-07-30 23:14:37 +00:00
fix(flows): live-run reliability — reconcile orphaned/cancelled runs + detach agent-initiated run_flow (B41, B42) (#5135)
This commit is contained in:
Cyrus Gray
@@ -71,6 +71,11 @@ export const FLOW_RUN_STATUS_ACCENT: Record<FlowRunStatus, string> = {
|
||||
'border-coral-200 bg-coral-50 text-coral-700 dark:border-coral-500/30 dark:bg-coral-500/10 dark:text-coral-300',
|
||||
// Neutral treatment, matching `WorkflowRunDetail.tsx`'s `RUN_STATUS_ACCENT.cancelled`.
|
||||
cancelled: 'border-line bg-surface-muted text-content-secondary',
|
||||
// Interrupted (bug B42): a run reconciled after its future was dropped
|
||||
// mid-flight. Amber-leaning "worth a look" like `pending_approval`, but
|
||||
// settled — it carries an `error` reason banner.
|
||||
interrupted:
|
||||
'border-amber-200 bg-amber-50 text-amber-700 dark:border-amber-500/30 dark:bg-amber-500/10 dark:text-amber-300',
|
||||
};
|
||||
|
||||
/** Header status dot per run status — mirrors `PHASE_STATUS_DOT`. Exported, see above. */
|
||||
@@ -82,6 +87,8 @@ export const FLOW_RUN_STATUS_DOT: Record<FlowRunStatus, string> = {
|
||||
pending_approval: 'bg-amber-500 animate-pulse',
|
||||
failed: 'bg-coral-500',
|
||||
cancelled: 'bg-surface-strong',
|
||||
// Settled (no pulse) — reconciled after being dropped mid-flight (bug B42).
|
||||
interrupted: 'bg-amber-500',
|
||||
};
|
||||
|
||||
/** i18n key per run status. Exported, see above. */
|
||||
@@ -92,6 +99,7 @@ export const FLOW_RUN_STATUS_KEY: Record<FlowRunStatus, string> = {
|
||||
pending_approval: 'flowRuns.status.pending_approval',
|
||||
failed: 'flowRuns.status.failed',
|
||||
cancelled: 'flowRuns.status.cancelled',
|
||||
interrupted: 'flowRuns.status.interrupted',
|
||||
};
|
||||
|
||||
function formatTimestamp(value: string | null | undefined): string | null {
|
||||
|
||||
@@ -170,6 +170,30 @@ describe('useFlowRunPoller', () => {
|
||||
expect(getFlowRun).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it('stops polling once the run is interrupted', async () => {
|
||||
// Bug B42: an `interrupted` run (reconciled after being dropped mid-flight)
|
||||
// is terminal — the poller must not loop forever on it.
|
||||
getFlowRun.mockResolvedValue(
|
||||
makeRun({
|
||||
status: 'interrupted',
|
||||
error: 'Run interrupted before completion',
|
||||
finished_at: '2026-01-01T00:01:00Z',
|
||||
})
|
||||
);
|
||||
const { result } = renderHook(() => useFlowRunPoller('thread-1'));
|
||||
|
||||
await act(async () => {
|
||||
await vi.advanceTimersByTimeAsync(0);
|
||||
});
|
||||
expect(result.current.run?.status).toBe('interrupted');
|
||||
expect(getFlowRun).toHaveBeenCalledTimes(1);
|
||||
|
||||
await act(async () => {
|
||||
await vi.advanceTimersByTimeAsync(10_000);
|
||||
});
|
||||
expect(getFlowRun).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it('stops polling once the run fails', async () => {
|
||||
getFlowRun.mockResolvedValue(makeRun({ status: 'failed', error: 'boom' }));
|
||||
const { result } = renderHook(() => useFlowRunPoller('thread-1'));
|
||||
|
||||
@@ -26,8 +26,9 @@
|
||||
*
|
||||
* `pending_approval` is explicitly NOT terminal — a paused run still needs
|
||||
* live status so the drawer reflects an approval elsewhere resolving it.
|
||||
* `completed_with_warnings` (run honesty, PR2) and `cancelled` are terminal,
|
||||
* same as `completed`/`failed`.
|
||||
* `completed_with_warnings` (run honesty, PR2), `cancelled`, and `interrupted`
|
||||
* (bug B42 — reconciled after being dropped mid-flight) are terminal, same as
|
||||
* `completed`/`failed`.
|
||||
*/
|
||||
import debug from 'debug';
|
||||
import { useEffect, useRef, useState } from 'react';
|
||||
@@ -51,6 +52,10 @@ const TERMINAL = new Set<FlowRunStatus>([
|
||||
'completed_with_warnings',
|
||||
'failed',
|
||||
'cancelled',
|
||||
// Reconciled after its future was dropped mid-flight (bug B42) — a settled
|
||||
// terminal state. Without this the poller would loop forever on a run that
|
||||
// never leaves `interrupted`.
|
||||
'interrupted',
|
||||
]);
|
||||
|
||||
function isTerminal(run: FlowRun | null): boolean {
|
||||
|
||||
@@ -40,6 +40,9 @@ const TERMINAL_STATUSES = new Set<FlowRunStatus>([
|
||||
'completed_with_warnings',
|
||||
'failed',
|
||||
'cancelled',
|
||||
// Reconciled after its future was dropped mid-flight (bug B42) — settled, so
|
||||
// the list's active-run backstop poll can quiesce once every run is terminal.
|
||||
'interrupted',
|
||||
]);
|
||||
|
||||
/** Trailing debounce window for a burst of `flow:run_progress` events. */
|
||||
|
||||
@@ -4041,6 +4041,7 @@ const messages: TranslationMap = {
|
||||
'flowRuns.status.pending_approval': 'بانتظار الموافقة',
|
||||
'flowRuns.status.failed': 'فشل',
|
||||
'flowRuns.status.cancelled': 'ملغى',
|
||||
'flowRuns.status.interrupted': 'تمت المقاطعة',
|
||||
|
||||
'flows.page.title': 'سير العمل',
|
||||
'flows.page.description': 'أتمتة محفوظة يمكنك تفعيلها وتشغيلها ومتابعتها.',
|
||||
@@ -4066,6 +4067,7 @@ const messages: TranslationMap = {
|
||||
'flows.allRuns.status.pending_approval': 'بانتظار الموافقة',
|
||||
'flows.allRuns.status.failed': 'فشل',
|
||||
'flows.allRuns.status.cancelled': 'أُلغي',
|
||||
'flows.allRuns.status.interrupted': 'تمت المقاطعة',
|
||||
'flows.list.minutesAgo': 'منذ {count} دقيقة',
|
||||
'flows.list.hoursAgo': 'منذ {count} ساعة',
|
||||
'flows.list.daysAgo': 'منذ {count} يوم',
|
||||
|
||||
@@ -4136,6 +4136,7 @@ const messages: TranslationMap = {
|
||||
'flowRuns.status.pending_approval': 'অনুমোদনের অপেক্ষায়',
|
||||
'flowRuns.status.failed': 'ব্যর্থ',
|
||||
'flowRuns.status.cancelled': 'বাতিল করা হয়েছে',
|
||||
'flowRuns.status.interrupted': 'বিঘ্নিত',
|
||||
|
||||
'flows.page.title': 'ওয়ার্কফ্লো',
|
||||
'flows.page.description': 'সংরক্ষিত অটোমেশন যা আপনি সক্ষম, চালাতে এবং পর্যবেক্ষণ করতে পারেন।',
|
||||
@@ -4162,6 +4163,7 @@ const messages: TranslationMap = {
|
||||
'flows.allRuns.status.pending_approval': 'অনুমোদনের অপেক্ষায়',
|
||||
'flows.allRuns.status.failed': 'ব্যর্থ',
|
||||
'flows.allRuns.status.cancelled': 'বাতিল',
|
||||
'flows.allRuns.status.interrupted': 'বিঘ্নিত',
|
||||
'flows.list.minutesAgo': '{count} মিনিট আগে',
|
||||
'flows.list.hoursAgo': '{count} ঘণ্টা আগে',
|
||||
'flows.list.daysAgo': '{count} দিন আগে',
|
||||
|
||||
@@ -4251,6 +4251,7 @@ const messages: TranslationMap = {
|
||||
'flowRuns.status.pending_approval': 'Wartet auf Genehmigung',
|
||||
'flowRuns.status.failed': 'Fehlgeschlagen',
|
||||
'flowRuns.status.cancelled': 'Abgebrochen',
|
||||
'flowRuns.status.interrupted': 'Unterbrochen',
|
||||
|
||||
'flows.page.title': 'Workflows',
|
||||
'flows.page.description':
|
||||
@@ -4278,6 +4279,7 @@ const messages: TranslationMap = {
|
||||
'flows.allRuns.status.pending_approval': 'Warten auf Freigabe',
|
||||
'flows.allRuns.status.failed': 'Fehlgeschlagen',
|
||||
'flows.allRuns.status.cancelled': 'Abgebrochen',
|
||||
'flows.allRuns.status.interrupted': 'Unterbrochen',
|
||||
'flows.list.minutesAgo': 'vor {count} Min.',
|
||||
'flows.list.hoursAgo': 'vor {count} Std.',
|
||||
'flows.list.daysAgo': 'vor {count} Tagen',
|
||||
|
||||
@@ -4799,6 +4799,7 @@ const en: TranslationMap = {
|
||||
'flowRuns.status.pending_approval': 'Awaiting approval',
|
||||
'flowRuns.status.failed': 'Failed',
|
||||
'flowRuns.status.cancelled': 'Cancelled',
|
||||
'flowRuns.status.interrupted': 'Interrupted',
|
||||
|
||||
// ── Workflows list page + nav tab (B5a): the `flows::` domain's
|
||||
// discoverable hub at /flows. Distinct from the legacy SKILL.md
|
||||
@@ -4845,6 +4846,7 @@ const en: TranslationMap = {
|
||||
'flows.allRuns.status.pending_approval': 'Pending approval',
|
||||
'flows.allRuns.status.failed': 'Failed',
|
||||
'flows.allRuns.status.cancelled': 'Cancelled',
|
||||
'flows.allRuns.status.interrupted': 'Interrupted',
|
||||
'flows.list.minutesAgo': '{count}m ago',
|
||||
'flows.list.hoursAgo': '{count}h ago',
|
||||
'flows.list.daysAgo': '{count}d ago',
|
||||
|
||||
@@ -4208,6 +4208,7 @@ const messages: TranslationMap = {
|
||||
'flowRuns.status.pending_approval': 'Esperando aprobación',
|
||||
'flowRuns.status.failed': 'Fallido',
|
||||
'flowRuns.status.cancelled': 'Cancelado',
|
||||
'flowRuns.status.interrupted': 'Interrumpido',
|
||||
|
||||
'flows.page.title': 'Flujos de trabajo',
|
||||
'flows.page.description':
|
||||
@@ -4234,6 +4235,7 @@ const messages: TranslationMap = {
|
||||
'flows.allRuns.status.pending_approval': 'Pendiente de aprobación',
|
||||
'flows.allRuns.status.failed': 'Fallido',
|
||||
'flows.allRuns.status.cancelled': 'Cancelado',
|
||||
'flows.allRuns.status.interrupted': 'Interrumpido',
|
||||
'flows.list.minutesAgo': 'hace {count} min',
|
||||
'flows.list.hoursAgo': 'hace {count} h',
|
||||
'flows.list.daysAgo': 'hace {count} d',
|
||||
|
||||
@@ -4233,6 +4233,7 @@ const messages: TranslationMap = {
|
||||
'flowRuns.status.pending_approval': "En attente d'approbation",
|
||||
'flowRuns.status.failed': 'Échoué',
|
||||
'flowRuns.status.cancelled': 'Annulé',
|
||||
'flowRuns.status.interrupted': 'Interrompu',
|
||||
|
||||
'flows.page.title': 'Workflows',
|
||||
'flows.page.description':
|
||||
@@ -4261,6 +4262,7 @@ const messages: TranslationMap = {
|
||||
'flows.allRuns.status.pending_approval': 'En attente d’approbation',
|
||||
'flows.allRuns.status.failed': 'Échoué',
|
||||
'flows.allRuns.status.cancelled': 'Annulé',
|
||||
'flows.allRuns.status.interrupted': 'Interrompu',
|
||||
'flows.list.minutesAgo': 'il y a {count} min',
|
||||
'flows.list.hoursAgo': 'il y a {count} h',
|
||||
'flows.list.daysAgo': 'il y a {count} j',
|
||||
|
||||
@@ -4135,6 +4135,7 @@ const messages: TranslationMap = {
|
||||
'flowRuns.status.pending_approval': 'अनुमोदन की प्रतीक्षा में',
|
||||
'flowRuns.status.failed': 'विफल',
|
||||
'flowRuns.status.cancelled': 'रद्द किया गया',
|
||||
'flowRuns.status.interrupted': 'बाधित',
|
||||
|
||||
'flows.page.title': 'वर्कफ़्लो',
|
||||
'flows.page.description': 'सहेजे गए ऑटोमेशन जिन्हें आप सक्षम, चला और मॉनिटर कर सकते हैं।',
|
||||
@@ -4160,6 +4161,7 @@ const messages: TranslationMap = {
|
||||
'flows.allRuns.status.pending_approval': 'अनुमोदन लंबित',
|
||||
'flows.allRuns.status.failed': 'विफल',
|
||||
'flows.allRuns.status.cancelled': 'रद्द',
|
||||
'flows.allRuns.status.interrupted': 'बाधित',
|
||||
'flows.list.minutesAgo': '{count} मिनट पहले',
|
||||
'flows.list.hoursAgo': '{count} घंटे पहले',
|
||||
'flows.list.daysAgo': '{count} दिन पहले',
|
||||
|
||||
@@ -4151,6 +4151,7 @@ const messages: TranslationMap = {
|
||||
'flowRuns.status.pending_approval': 'Menunggu persetujuan',
|
||||
'flowRuns.status.failed': 'Gagal',
|
||||
'flowRuns.status.cancelled': 'Dibatalkan',
|
||||
'flowRuns.status.interrupted': 'Terhenti',
|
||||
|
||||
'flows.page.title': 'Alur Kerja',
|
||||
'flows.page.description': 'Otomatisasi tersimpan yang dapat Anda aktifkan, jalankan, dan pantau.',
|
||||
@@ -4177,6 +4178,7 @@ const messages: TranslationMap = {
|
||||
'flows.allRuns.status.pending_approval': 'Menunggu persetujuan',
|
||||
'flows.allRuns.status.failed': 'Gagal',
|
||||
'flows.allRuns.status.cancelled': 'Dibatalkan',
|
||||
'flows.allRuns.status.interrupted': 'Terhenti',
|
||||
'flows.list.minutesAgo': '{count} menit lalu',
|
||||
'flows.list.hoursAgo': '{count} jam lalu',
|
||||
'flows.list.daysAgo': '{count} hari lalu',
|
||||
|
||||
@@ -4206,6 +4206,7 @@ const messages: TranslationMap = {
|
||||
'flowRuns.status.pending_approval': 'In attesa di approvazione',
|
||||
'flowRuns.status.failed': 'Non riuscito',
|
||||
'flowRuns.status.cancelled': 'Annullato',
|
||||
'flowRuns.status.interrupted': 'Interrotto',
|
||||
|
||||
'flows.page.title': 'Flussi di lavoro',
|
||||
'flows.page.description': 'Automazioni salvate che puoi abilitare, eseguire e monitorare.',
|
||||
@@ -4231,6 +4232,7 @@ const messages: TranslationMap = {
|
||||
'flows.allRuns.status.pending_approval': 'In attesa di approvazione',
|
||||
'flows.allRuns.status.failed': 'Non riuscito',
|
||||
'flows.allRuns.status.cancelled': 'Annullato',
|
||||
'flows.allRuns.status.interrupted': 'Interrotto',
|
||||
'flows.list.minutesAgo': '{count} min fa',
|
||||
'flows.list.hoursAgo': '{count} h fa',
|
||||
'flows.list.daysAgo': '{count} g fa',
|
||||
|
||||
@@ -4094,6 +4094,7 @@ const messages: TranslationMap = {
|
||||
'flowRuns.status.pending_approval': '승인 대기 중',
|
||||
'flowRuns.status.failed': '실패',
|
||||
'flowRuns.status.cancelled': '취소됨',
|
||||
'flowRuns.status.interrupted': '중단됨',
|
||||
|
||||
'flows.page.title': '워크플로',
|
||||
'flows.page.description': '활성화, 실행, 모니터링할 수 있는 저장된 자동화입니다.',
|
||||
@@ -4118,6 +4119,7 @@ const messages: TranslationMap = {
|
||||
'flows.allRuns.status.pending_approval': '승인 대기 중',
|
||||
'flows.allRuns.status.failed': '실패',
|
||||
'flows.allRuns.status.cancelled': '취소됨',
|
||||
'flows.allRuns.status.interrupted': '중단됨',
|
||||
'flows.list.minutesAgo': '{count}분 전',
|
||||
'flows.list.hoursAgo': '{count}시간 전',
|
||||
'flows.list.daysAgo': '{count}일 전',
|
||||
|
||||
@@ -4190,6 +4190,7 @@ const messages: TranslationMap = {
|
||||
'flowRuns.status.pending_approval': 'Oczekuje na zatwierdzenie',
|
||||
'flowRuns.status.failed': 'Niepowodzenie',
|
||||
'flowRuns.status.cancelled': 'Anulowano',
|
||||
'flowRuns.status.interrupted': 'Przerwano',
|
||||
|
||||
'flows.page.title': 'Przepływy pracy',
|
||||
'flows.page.description':
|
||||
@@ -4217,6 +4218,7 @@ const messages: TranslationMap = {
|
||||
'flows.allRuns.status.pending_approval': 'Oczekuje na zatwierdzenie',
|
||||
'flows.allRuns.status.failed': 'Niepowodzenie',
|
||||
'flows.allRuns.status.cancelled': 'Anulowano',
|
||||
'flows.allRuns.status.interrupted': 'Przerwano',
|
||||
'flows.list.minutesAgo': '{count} min temu',
|
||||
'flows.list.hoursAgo': '{count} godz. temu',
|
||||
'flows.list.daysAgo': '{count} dni temu',
|
||||
|
||||
@@ -4198,6 +4198,7 @@ const messages: TranslationMap = {
|
||||
'flowRuns.status.pending_approval': 'Aguardando aprovação',
|
||||
'flowRuns.status.failed': 'Falhou',
|
||||
'flowRuns.status.cancelled': 'Cancelado',
|
||||
'flowRuns.status.interrupted': 'Interrompido',
|
||||
|
||||
'flows.page.title': 'Fluxos de trabalho',
|
||||
'flows.page.description': 'Automações salvas que você pode habilitar, executar e monitorar.',
|
||||
@@ -4224,6 +4225,7 @@ const messages: TranslationMap = {
|
||||
'flows.allRuns.status.pending_approval': 'Aguardando aprovação',
|
||||
'flows.allRuns.status.failed': 'Falhou',
|
||||
'flows.allRuns.status.cancelled': 'Cancelado',
|
||||
'flows.allRuns.status.interrupted': 'Interrompido',
|
||||
'flows.list.minutesAgo': 'há {count} min',
|
||||
'flows.list.hoursAgo': 'há {count} h',
|
||||
'flows.list.daysAgo': 'há {count} d',
|
||||
|
||||
@@ -4171,6 +4171,7 @@ const messages: TranslationMap = {
|
||||
'flowRuns.status.pending_approval': 'Ожидает подтверждения',
|
||||
'flowRuns.status.failed': 'Не удалось',
|
||||
'flowRuns.status.cancelled': 'Отменено',
|
||||
'flowRuns.status.interrupted': 'Прервано',
|
||||
|
||||
'flows.page.title': 'Рабочие процессы',
|
||||
'flows.page.description':
|
||||
@@ -4199,6 +4200,7 @@ const messages: TranslationMap = {
|
||||
'flows.allRuns.status.pending_approval': 'Ожидает одобрения',
|
||||
'flows.allRuns.status.failed': 'Ошибка',
|
||||
'flows.allRuns.status.cancelled': 'Отменено',
|
||||
'flows.allRuns.status.interrupted': 'Прервано',
|
||||
'flows.list.minutesAgo': '{count} мин назад',
|
||||
'flows.list.hoursAgo': '{count} ч назад',
|
||||
'flows.list.daysAgo': '{count} дн назад',
|
||||
|
||||
@@ -3920,6 +3920,7 @@ const messages: TranslationMap = {
|
||||
'flowRuns.status.pending_approval': '等待批准',
|
||||
'flowRuns.status.failed': '失败',
|
||||
'flowRuns.status.cancelled': '已取消',
|
||||
'flowRuns.status.interrupted': '已中断',
|
||||
|
||||
'flows.page.title': '工作流',
|
||||
'flows.page.description': '已保存的自动化流程，可启用、运行并监控。',
|
||||
@@ -3944,6 +3945,7 @@ const messages: TranslationMap = {
|
||||
'flows.allRuns.status.pending_approval': '等待批准',
|
||||
'flows.allRuns.status.failed': '失败',
|
||||
'flows.allRuns.status.cancelled': '已取消',
|
||||
'flows.allRuns.status.interrupted': '已中断',
|
||||
'flows.list.minutesAgo': '{count}分钟前',
|
||||
'flows.list.hoursAgo': '{count}小时前',
|
||||
'flows.list.daysAgo': '{count}天前',
|
||||
|
||||
@@ -37,6 +37,7 @@ const STATUS_CLASS: Record<FlowRunStatus, string> = {
|
||||
pending_approval: 'bg-amber-500/15 text-amber-700 dark:text-amber-300',
|
||||
failed: 'bg-coral-500/15 text-coral-700 dark:text-coral-300',
|
||||
cancelled: 'bg-content-faint/15 text-content-secondary',
|
||||
interrupted: 'bg-amber-500/15 text-amber-700 dark:text-amber-300',
|
||||
};
|
||||
|
||||
export default function WorkflowRunsPage() {
|
||||
|
||||
@@ -56,7 +56,12 @@ export type FlowRunStatus =
|
||||
| 'completed_with_warnings'
|
||||
| 'pending_approval'
|
||||
| 'failed'
|
||||
| 'cancelled';
|
||||
| 'cancelled'
|
||||
// A run whose future was dropped mid-flight (harness tool abort, chat turn
|
||||
// end, timeout, or an app restart), reconciled to a terminal state by the
|
||||
// core's `RunRowFinalizer` drop-guard or its boot-time orphan sweep (bug
|
||||
// B42). Carries a human `error` reason; rendered as a settled, non-active run.
|
||||
| 'interrupted';
|
||||
|
||||
/** One reconstructed step of a persisted `FlowRun` (`src/openhuman/flows/types.rs::FlowRunStep`). */
|
||||
export interface FlowRunStep {
|
||||
@@ -87,10 +92,22 @@ export interface FlowRun {
|
||||
thread_id: string;
|
||||
status: FlowRunStatus;
|
||||
started_at: string;
|
||||
/**
|
||||
* RFC3339 timestamp stamped when the run settled — set for every terminal
|
||||
* status, including `'interrupted'` (the drop-guard / boot sweep stamps it
|
||||
* exactly like a normal terminal write). `null`/absent only while the run is
|
||||
* still `'running'`.
|
||||
*/
|
||||
finished_at?: string | null;
|
||||
steps: FlowRunStep[];
|
||||
/** Node ids paused awaiting approval when `status === 'pending_approval'`. */
|
||||
pending_approvals: string[];
|
||||
/**
|
||||
* Human-readable failure reason. Set for `'failed'` runs and for
|
||||
* `'interrupted'` ones (where it carries the reconciliation reason — tool
|
||||
* abort / turn end / app restart), so the UI can surface *why* a run stopped
|
||||
* rather than showing a bare terminal state.
|
||||
*/
|
||||
error?: string | null;
|
||||
}
|
||||
|
||||
|
||||
@@ -627,6 +627,12 @@ impl CoreRuntime {
|
||||
if self.services.cron {
|
||||
services::spawn_cron_service();
|
||||
}
|
||||
// Flow-run boot reconciliation is selected by the flows *domain*, not by
|
||||
// a background service — runs can be started without cron in the
|
||||
// ServiceSet, so their orphans must be reconcilable without it too.
|
||||
if self.ctx.domains().flows {
|
||||
services::spawn_flows_boot_reconcile();
|
||||
}
|
||||
if self.services.channels {
|
||||
services::spawn_channels_service();
|
||||
}
|
||||
|
||||
@@ -104,6 +104,51 @@ pub fn spawn_update_scheduler() {
|
||||
});
|
||||
}
|
||||
|
||||
/// Boot-time flow-run reconciliation (bug B42): reconciles any `flow_runs` row
|
||||
/// left at `running` by a prior process (crash/SIGKILL/power loss — where the
|
||||
/// in-process `RunRowFinalizer` drop-guard never got to run) to a terminal
|
||||
/// `interrupted`, so the run-details sidebar never shows a perpetual blank
|
||||
/// spinner for a run nothing is executing.
|
||||
///
|
||||
/// Owned by the flows domain rather than piggybacked on cron bootstrap: runs
|
||||
/// can be started by the RPC "Run" control, the agent `run_flow` tool and the
|
||||
/// trigger bus, none of which need the cron *service* to be in the active
|
||||
/// [`ServiceSet`]. Gating this on cron would silently skip reconciliation on any
|
||||
/// cron-less selection (`headless_api()`, embedders), leaving prior-process
|
||||
/// orphans wedged forever. Selected by the `flows` **domain** flag instead, and
|
||||
/// safe at any point in boot — the sweep's own `PROCESS_RUN_FLOOR` guard means
|
||||
/// it can never touch a run this process started, so it carries no ordering
|
||||
/// requirement against the cron scheduler or any agent turn.
|
||||
pub fn spawn_flows_boot_reconcile() {
|
||||
#[cfg(feature = "flows")]
|
||||
{
|
||||
log::debug!("[flows] boot reconcile: scheduling orphaned-run sweep");
|
||||
tokio::spawn(async {
|
||||
log::debug!("[flows] boot reconcile: loading config");
|
||||
match crate::openhuman::config::Config::load_or_init().await {
|
||||
Ok(config) => {
|
||||
let swept =
|
||||
crate::openhuman::flows::ops::sweep_orphaned_running_runs_on_boot(&config)
|
||||
.await;
|
||||
// Logged unconditionally: a silent success and a task that
|
||||
// never ran are otherwise indistinguishable in a boot log.
|
||||
log::debug!("[flows] boot reconcile: completed; reconciled_runs={swept}");
|
||||
if swept > 0 {
|
||||
log::info!(
|
||||
"[flows] boot sweep reconciled {swept} orphaned running run(s) to 'interrupted'"
|
||||
);
|
||||
}
|
||||
}
|
||||
Err(err) => {
|
||||
log::warn!("[core] config load failed, skipping flows boot reconcile: {err}");
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
#[cfg(not(feature = "flows"))]
|
||||
log::debug!("[flows] flows feature disabled at compile time — no boot run reconciliation");
|
||||
}
|
||||
|
||||
/// Cron scheduler — polls `due_jobs()` every ~5s and executes them
|
||||
/// automatically. Gated by `config.cron.enabled`.
|
||||
pub fn spawn_cron_service() {
|
||||
|
||||
+438
-21
@@ -4,7 +4,7 @@
|
||||
//! `src/openhuman/cron/ops.rs`.
|
||||
|
||||
use std::collections::HashSet;
|
||||
use std::sync::Arc;
|
||||
use std::sync::{Arc, LazyLock};
|
||||
|
||||
use chrono::Utc;
|
||||
use serde_json::{json, Value};
|
||||
@@ -3705,6 +3705,133 @@ pub async fn flows_run(
|
||||
input: Value,
|
||||
trigger: FlowRunTrigger,
|
||||
) -> Result<RpcOutcome<Value>, String> {
|
||||
// Prep synchronously (validate + compile-check + mint the run id), insert
|
||||
// the initial `running` row, and announce it, then hand off to the shared
|
||||
// run body. Both the synchronous "Run" RPC path (this fn) and the detached
|
||||
// agent path ([`flows_run_detached`]) reuse `run_flow_body` so a single
|
||||
// [`RunRowFinalizer`] guards the row on every exit — bug B42.
|
||||
let prepared = prepare_flow_run(config, flow_id)?;
|
||||
let thread_id = prepared.thread_id.clone();
|
||||
let no_actionable_nodes = prepared.no_actionable_nodes;
|
||||
|
||||
// Register BEFORE the row exists, so a `flows_cancel_run` can never observe
|
||||
// a `running` row that no live run owns (see [`run_flow_body`]'s doc).
|
||||
let (cancel_token, run_guard) = run_registry::register(&thread_id);
|
||||
start_flow_run_row(config, &thread_id, flow_id);
|
||||
publish_flow_run_started(flow_id, &thread_id);
|
||||
|
||||
run_flow_body(
|
||||
Arc::new(config.clone()),
|
||||
prepared.flow,
|
||||
flow_id.to_string(),
|
||||
thread_id,
|
||||
input,
|
||||
trigger,
|
||||
no_actionable_nodes,
|
||||
cancel_token,
|
||||
run_guard,
|
||||
)
|
||||
.await
|
||||
}
|
||||
|
||||
/// Agent-initiated `run_flow` entry point (bug B41). Unlike [`flows_run`], this
|
||||
/// does NOT block on the engine: the tinyagents harness caps a single tool call
|
||||
/// at 120s, but any flow whose first real node is a live-research agent node
|
||||
/// (`web_search` + `web_fetch` + `parallel_research`) inherently runs longer
|
||||
/// than that, so a blocking `run_flow` tool call could *never* succeed for a
|
||||
/// realistic flow — it died at exactly 120s, orphaning the run row (bug B42).
|
||||
///
|
||||
/// Instead this validates + compile-checks the flow synchronously (so a broken
|
||||
/// flow still returns an immediate, actionable error to the agent), inserts the
|
||||
/// `running` row, publishes `FlowRunStarted`, then spawns [`run_flow_body`] on a
|
||||
/// background task and returns `{ run_id, status: "running", detached: true }`
|
||||
/// in well under 120s. The copilot already polls `get_flow_run(run_id)` (seen
|
||||
/// in live traces), so it observes the run settle to a terminal state on its
|
||||
/// own cadence. Mirrors how the UI "Run" control and the trigger bus
|
||||
/// (`flows::bus::spawn_run`) already fire runs fire-and-forget. Combined with
|
||||
/// B42's finalizer + boot sweep, a detached run ALWAYS settles to a terminal
|
||||
/// row even if the process dies mid-run.
|
||||
pub async fn flows_run_detached(
|
||||
config: &Config,
|
||||
flow_id: &str,
|
||||
input: Value,
|
||||
trigger: FlowRunTrigger,
|
||||
) -> Result<RpcOutcome<Value>, String> {
|
||||
let prepared = prepare_flow_run(config, flow_id)?;
|
||||
let thread_id = prepared.thread_id.clone();
|
||||
let no_actionable_nodes = prepared.no_actionable_nodes;
|
||||
|
||||
// Register BEFORE the `run_id` becomes observable to the agent. The spawned
|
||||
// task below may not be polled for some time, so registering inside it
|
||||
// would leave a window where a `flows_cancel_run` on the returned `run_id`
|
||||
// sees no in-flight run, settles the row `cancelled` + drops the
|
||||
// checkpoint, and the background run then executes the flow's real side
|
||||
// effects anyway and overwrites that terminal status. Registering here
|
||||
// means such a cancel always takes the signalled branch and this run's own
|
||||
// cancellation arm unwinds it. See [`run_flow_body`]'s doc.
|
||||
let (cancel_token, run_guard) = run_registry::register(&thread_id);
|
||||
start_flow_run_row(config, &thread_id, flow_id);
|
||||
publish_flow_run_started(flow_id, &thread_id);
|
||||
|
||||
tracing::info!(
|
||||
target: "flows",
|
||||
flow_id = %flow_id,
|
||||
run_id = %thread_id,
|
||||
"[flows] flows_run_detached: registered + spawning background run; returning run_id immediately"
|
||||
);
|
||||
|
||||
let config_arc = Arc::new(config.clone());
|
||||
let flow = prepared.flow;
|
||||
let flow_id_owned = flow_id.to_string();
|
||||
let body_thread_id = thread_id.clone();
|
||||
tokio::spawn(async move {
|
||||
if let Err(e) = run_flow_body(
|
||||
config_arc,
|
||||
flow,
|
||||
flow_id_owned,
|
||||
body_thread_id,
|
||||
input,
|
||||
trigger,
|
||||
no_actionable_nodes,
|
||||
cancel_token,
|
||||
run_guard,
|
||||
)
|
||||
.await
|
||||
{
|
||||
// The row is already reconciled by the body's terminal write /
|
||||
// finalizer — this only logs that the detached run ended in error.
|
||||
tracing::warn!(target: "flows", error = %e, "[flows] flows_run_detached: background run ended with error (row already reconciled)");
|
||||
}
|
||||
});
|
||||
|
||||
let result = json!({
|
||||
"run_id": thread_id,
|
||||
"flow_id": flow_id,
|
||||
"status": "running",
|
||||
"detached": true,
|
||||
});
|
||||
Ok(RpcOutcome::single_log(
|
||||
result,
|
||||
format!("flow run started (detached): {thread_id}"),
|
||||
))
|
||||
}
|
||||
|
||||
/// A validated, ready-to-execute flow run: the loaded [`Flow`], the freshly
|
||||
/// minted `thread_id` (== run id / checkpointer key), and whether the graph has
|
||||
/// no actionable nodes. Produced by [`prepare_flow_run`] and consumed by both
|
||||
/// `flows_run` entry points.
|
||||
struct PreparedFlowRun {
|
||||
flow: Flow,
|
||||
thread_id: String,
|
||||
no_actionable_nodes: bool,
|
||||
}
|
||||
|
||||
/// Synchronous prep shared by [`flows_run`] and [`flows_run_detached`]: loads
|
||||
/// the flow, warns on an actionless graph, rejects an engine-incompatible
|
||||
/// topology, compile-checks the graph so a broken flow fails fast *before* any
|
||||
/// `running` row is inserted, and mints the run's `thread_id`. Returns an error
|
||||
/// (never a wedged row) if the flow can't run at all.
|
||||
fn prepare_flow_run(config: &Config, flow_id: &str) -> Result<PreparedFlowRun, String> {
|
||||
let flow = store::get_flow(config, flow_id)
|
||||
.map_err(|e| e.to_string())?
|
||||
.ok_or_else(|| format!("flow '{flow_id}' not found"))?;
|
||||
@@ -3743,26 +3870,31 @@ pub async fn flows_run(
|
||||
);
|
||||
return Err(error);
|
||||
}
|
||||
let compiled = tinyflows::compiler::compile(&flow.graph).map_err(|e| e.to_string())?;
|
||||
// Compile-check up front so a structurally broken graph fails the caller
|
||||
// immediately, before a `running` row exists. `run_flow_body` recompiles
|
||||
// (cheap) to actually execute.
|
||||
tinyflows::compiler::compile(&flow.graph).map_err(|e| e.to_string())?;
|
||||
|
||||
let config_arc = Arc::new(config.clone());
|
||||
// Scope the state store per-flow so two flows never collide on a state key.
|
||||
let caps =
|
||||
crate::openhuman::tinyflows::build_capabilities(config_arc, format!("flow:{flow_id}"));
|
||||
let checkpointer =
|
||||
crate::openhuman::tinyflows::open_flow_checkpointer(config).map_err(|e| e.to_string())?;
|
||||
let thread_id = format!("flow:{flow_id}:{}", uuid::Uuid::new_v4());
|
||||
|
||||
tracing::debug!(
|
||||
target: "flows",
|
||||
flow_id = %flow_id,
|
||||
thread_id = %thread_id,
|
||||
require_approval = flow.require_approval,
|
||||
"[flows] flows_run: starting checkpointed run"
|
||||
"[flows] flows_run: prepared checkpointed run"
|
||||
);
|
||||
|
||||
start_flow_run_row(config, &thread_id, flow_id);
|
||||
Ok(PreparedFlowRun {
|
||||
flow,
|
||||
thread_id,
|
||||
no_actionable_nodes,
|
||||
})
|
||||
}
|
||||
|
||||
/// Announces a freshly-started run on the global event bus so the frontend run
|
||||
/// list flips to `running` immediately. Factored out of [`flows_run`] so both
|
||||
/// entry points publish identically.
|
||||
fn publish_flow_run_started(flow_id: &str, thread_id: &str) {
|
||||
tracing::debug!(
|
||||
target: "flows",
|
||||
flow_id = %flow_id,
|
||||
@@ -3771,13 +3903,172 @@ pub async fn flows_run(
|
||||
);
|
||||
crate::core::event_bus::publish_global(crate::core::event_bus::DomainEvent::FlowRunStarted {
|
||||
flow_id: flow_id.to_string(),
|
||||
run_id: thread_id.clone(),
|
||||
run_id: thread_id.to_string(),
|
||||
});
|
||||
}
|
||||
|
||||
// Register this run as in-flight (issue G4) so a concurrent
|
||||
// `flows_cancel_run` can signal it to abort. The guard deregisters on any
|
||||
// exit from this fn (including the early returns below).
|
||||
let (cancel_token, _run_guard) = run_registry::register(&thread_id);
|
||||
/// Human-readable reason stamped on a run row that the [`RunRowFinalizer`]
|
||||
/// drop-guard reconciles because its run future was dropped mid-flight (harness
|
||||
/// tool abort, chat turn end, runtime shutdown, panic) before any terminal
|
||||
/// write landed. Surfaced verbatim in the run-details sidebar (bug B42c) so a
|
||||
/// cancelled/timed-out run reads as interrupted rather than a blank spinner.
|
||||
const INTERRUPTED_DROP_REASON: &str =
|
||||
"Run interrupted before completion — it was cancelled, timed out, or the app shut down mid-run.";
|
||||
|
||||
/// Cancellation-safe finalizer for a live `flow_runs` row (bug B42).
|
||||
///
|
||||
/// While a run's engine future is awaiting, dropping that future — the harness
|
||||
/// 120s tool abort, a chat turn ending, tokio runtime shutdown, or a panic —
|
||||
/// would otherwise leave the row wedged at `status="running"`, `error=NULL`,
|
||||
/// `steps=[]` forever, which the run-details sidebar renders as a perpetual
|
||||
/// blank spinner. Held across the await, this guard writes a terminal
|
||||
/// `"interrupted"` status + human reason on `Drop` UNLESS it has been
|
||||
/// explicitly [`disarm`](Self::disarm)ed after a real terminal write. The
|
||||
/// `armed` flag is a single-task `Cell` (the guard never crosses tasks by
|
||||
/// reference), so the type stays `Send` for `tokio::spawn`.
|
||||
struct RunRowFinalizer {
|
||||
config: Arc<Config>,
|
||||
thread_id: String,
|
||||
flow_id: String,
|
||||
armed: std::cell::Cell<bool>,
|
||||
}
|
||||
|
||||
impl RunRowFinalizer {
|
||||
fn new(config: Arc<Config>, thread_id: &str, flow_id: &str) -> Self {
|
||||
Self {
|
||||
config,
|
||||
thread_id: thread_id.to_string(),
|
||||
flow_id: flow_id.to_string(),
|
||||
armed: std::cell::Cell::new(true),
|
||||
}
|
||||
}
|
||||
|
||||
/// Disarm the guard after a real terminal write (success/failure/cancel/
|
||||
/// pause) has already finalized the row, so `Drop` becomes a no-op.
|
||||
fn disarm(&self) {
|
||||
self.armed.set(false);
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for RunRowFinalizer {
|
||||
fn drop(&mut self) {
|
||||
if !self.armed.get() {
|
||||
return;
|
||||
}
|
||||
tracing::warn!(
|
||||
target: "flows",
|
||||
flow_id = %self.flow_id,
|
||||
thread_id = %self.thread_id,
|
||||
"[flows] RunRowFinalizer: run future dropped before settling — reconciling orphaned 'running' row to 'interrupted'"
|
||||
);
|
||||
// Preserve whatever steps the live observer already persisted.
|
||||
let observed = current_persisted_steps(&self.config, &self.thread_id);
|
||||
finish_flow_run_row(
|
||||
&self.config,
|
||||
&self.thread_id,
|
||||
&self.flow_id,
|
||||
"interrupted",
|
||||
&observed,
|
||||
&[],
|
||||
Some(INTERRUPTED_DROP_REASON),
|
||||
);
|
||||
// Keep the flow-definition summary in step with the row, exactly as the
|
||||
// success/failure/cancel arms and the boot sweep do — otherwise the
|
||||
// runs list keeps advertising the *previous* run's `last_status` /
|
||||
// `last_run_at` for a flow whose latest run was interrupted.
|
||||
// `record_run` is synchronous, so it is safe in `Drop`.
|
||||
if let Err(e) = store::record_run(&self.config, &self.flow_id, "interrupted") {
|
||||
tracing::warn!(
|
||||
target: "flows",
|
||||
flow_id = %self.flow_id,
|
||||
thread_id = %self.thread_id,
|
||||
error = %e,
|
||||
"[flows] RunRowFinalizer: failed to update flow summary for interrupted run"
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Executes an already-prepared, already-`running`-row-inserted flow run to a
|
||||
/// terminal state, finalizing the `flow_runs` row on every exit path.
|
||||
///
|
||||
/// Split out of [`flows_run`] (bugs B41/B42) so the synchronous and detached
|
||||
/// entry points share ONE run body — and so a single [`RunRowFinalizer`]
|
||||
/// reconciles the row to `"interrupted"` if this future is dropped mid-await
|
||||
/// before any terminal write lands. The caller MUST have already
|
||||
/// [`run_registry::register`]ed `thread_id` (handing the token + guard in
|
||||
/// here), inserted the initial `running` row ([`start_flow_run_row`]) and
|
||||
/// published `FlowRunStarted`.
|
||||
///
|
||||
/// **Registration is the caller's job on purpose.** It used to happen here, but
|
||||
/// on the detached path that left a window: `flows_run_detached` returned the
|
||||
/// `run_id` to the agent before the spawned task had registered, so a
|
||||
/// `flows_cancel_run` landing in that gap saw `is_in_flight == false`, took the
|
||||
/// "parked/stale" branch, wrote a terminal `cancelled` row and dropped the
|
||||
/// checkpoint — while this body then started and executed the flow's real
|
||||
/// side effects anyway, finally overwriting `cancelled` with its own terminal
|
||||
/// status. Registering before the `run_id` is observable makes the cancel
|
||||
/// always take the signalled branch instead. `_run_guard` is held for the whole
|
||||
/// body and deregisters on any exit, including the early returns below.
|
||||
async fn run_flow_body(
|
||||
config_arc: Arc<Config>,
|
||||
flow: Flow,
|
||||
flow_id: String,
|
||||
thread_id: String,
|
||||
input: Value,
|
||||
trigger: FlowRunTrigger,
|
||||
no_actionable_nodes: bool,
|
||||
cancel_token: tokio_util::sync::CancellationToken,
|
||||
_run_guard: run_registry::RunGuard,
|
||||
) -> Result<RpcOutcome<Value>, String> {
|
||||
let config: &Config = config_arc.as_ref();
|
||||
let flow_id: &str = flow_id.as_str();
|
||||
|
||||
// Recompile to execute — the entry point already compile-checked to fail
|
||||
// fast before the running row existed. A failure *now* (after the row was
|
||||
// inserted) must finalize the row as failed, never orphan it.
|
||||
let compiled = match tinyflows::compiler::compile(&flow.graph) {
|
||||
Ok(compiled) => compiled,
|
||||
Err(e) => {
|
||||
let msg = e.to_string();
|
||||
tracing::warn!(target: "flows", flow_id, error = %msg, "[flows] run_flow_body: compile failed after start row inserted");
|
||||
let observed = current_persisted_steps(config, &thread_id);
|
||||
finish_flow_run_row(
|
||||
config,
|
||||
&thread_id,
|
||||
flow_id,
|
||||
"failed",
|
||||
&observed,
|
||||
&[],
|
||||
Some(&msg),
|
||||
);
|
||||
return Err(msg);
|
||||
}
|
||||
};
|
||||
|
||||
// Scope the state store per-flow so two flows never collide on a state key.
|
||||
let caps = crate::openhuman::tinyflows::build_capabilities(
|
||||
config_arc.clone(),
|
||||
format!("flow:{flow_id}"),
|
||||
);
|
||||
let checkpointer = match crate::openhuman::tinyflows::open_flow_checkpointer(config) {
|
||||
Ok(checkpointer) => checkpointer,
|
||||
Err(e) => {
|
||||
let msg = e.to_string();
|
||||
tracing::warn!(target: "flows", flow_id, error = %msg, "[flows] run_flow_body: checkpointer open failed after start row inserted");
|
||||
let observed = current_persisted_steps(config, &thread_id);
|
||||
finish_flow_run_row(
|
||||
config,
|
||||
&thread_id,
|
||||
flow_id,
|
||||
"failed",
|
||||
&observed,
|
||||
&[],
|
||||
Some(&msg),
|
||||
);
|
||||
return Err(msg);
|
||||
}
|
||||
};
|
||||
|
||||
// Record a failed attempt so `last_run_at`/`last_status` reflect reality
|
||||
// (a stop-policy engine/capability failure or a timeout) rather than
|
||||
@@ -3845,6 +4136,11 @@ pub async fn flows_run(
|
||||
);
|
||||
let timed = tokio::time::timeout(std::time::Duration::from_secs(FLOW_RUN_TIMEOUT_SECS), run);
|
||||
tokio::pin!(timed);
|
||||
// B42 drop-guard: armed for the whole awaiting region below. If this future
|
||||
// is dropped before any terminal write (harness abort, turn end, runtime
|
||||
// shutdown, panic), its `Drop` reconciles the orphaned `running` row to
|
||||
// `interrupted`. Every settled path disarms it after its own terminal write.
|
||||
let finalizer = RunRowFinalizer::new(config_arc.clone(), &thread_id, flow_id);
|
||||
// Race the run against a cancellation signal (issue G4). `biased` checks the
|
||||
// cancel arm first so a `flows_cancel_run` that lands right as the run
|
||||
// settles still wins deterministically.
|
||||
@@ -3865,6 +4161,7 @@ pub async fn flows_run(
|
||||
&[],
|
||||
Some("run cancelled"),
|
||||
);
|
||||
finalizer.disarm();
|
||||
drop_checkpoint(config, &thread_id).await;
|
||||
return Ok(RpcOutcome::single_log(
|
||||
json!({
|
||||
@@ -3880,12 +4177,14 @@ pub async fn flows_run(
|
||||
Ok(Ok(journaled)) => journaled,
|
||||
Ok(Err(e)) => {
|
||||
record_failed(&e.to_string());
|
||||
finalizer.disarm();
|
||||
tracing::warn!(target: "flows", flow_id = %flow_id, error = %e, "[flows] flows_run: run failed");
|
||||
return Err(e.to_string());
|
||||
}
|
||||
Err(_elapsed) => {
|
||||
let msg = format!("flow run timed out after {FLOW_RUN_TIMEOUT_SECS}s");
|
||||
record_failed(&msg);
|
||||
finalizer.disarm();
|
||||
tracing::warn!(target: "flows", flow_id = %flow_id, timeout_secs = FLOW_RUN_TIMEOUT_SECS, "[flows] flows_run: run timed out");
|
||||
return Err(msg);
|
||||
}
|
||||
@@ -3895,7 +4194,10 @@ pub async fn flows_run(
|
||||
|
||||
let settled = settle_steps(config, &thread_id, &outcome.output);
|
||||
let (status, error) = finalize_terminal_status(&settled, &outcome.pending_approvals);
|
||||
store::record_run(config, flow_id, status).map_err(|e| e.to_string())?;
|
||||
// Finalize the run row (and disarm the drop-guard) BEFORE the flow-summary
|
||||
// write, so a `record_run` failure can never leave the row wedged at
|
||||
// `running` — the row's terminal state is the correctness-critical write;
|
||||
// the summary is best-effort observability (see `start_flow_run_row`).
|
||||
finish_flow_run_row(
|
||||
config,
|
||||
&thread_id,
|
||||
@@ -3905,6 +4207,10 @@ pub async fn flows_run(
|
||||
&outcome.pending_approvals,
|
||||
error.as_deref(),
|
||||
);
|
||||
finalizer.disarm();
|
||||
if let Err(e) = store::record_run(config, flow_id, status) {
|
||||
tracing::warn!(target: "flows", flow_id = %flow_id, status, error = %e, "[flows] flows_run: failed to record run summary (run row already finalized)");
|
||||
}
|
||||
export_run_to_langfuse(
|
||||
config,
|
||||
&flow.name,
|
||||
@@ -4296,6 +4602,90 @@ pub async fn sweep_expired_parked_runs(config: &Config) -> usize {
|
||||
swept.len()
|
||||
}
|
||||
|
||||
/// Boot-time orphan sweep (bug B42, part b): reconciles every `flow_runs` row
|
||||
/// still at `status = 'running'` that has **no live in-process run** to a
|
||||
/// terminal `"interrupted"`. A hard crash / SIGKILL / power loss leaves the
|
||||
/// [`RunRowFinalizer`] drop-guard no chance to run, so a `running` row from the
|
||||
/// prior process would otherwise stay wedged forever, rendering as a perpetual
|
||||
/// blank spinner in the run-details sidebar.
|
||||
///
|
||||
/// Two independent guards keep the sweep off a run that **this** process owns:
|
||||
///
|
||||
/// 1. **A boot floor.** Only rows whose `started_at` predates
|
||||
/// [`PROCESS_RUN_FLOOR`] are candidates at all, so a row this process
|
||||
/// inserted is provably out of scope regardless of registration timing —
|
||||
/// which is what the sweep is actually for: rows left by a *prior* process.
|
||||
/// Sweeping a live run would not merely mislabel it (its own terminal write
|
||||
/// would correct that) — it would `drop_checkpoint` it mid-run, and that is
|
||||
/// unrecoverable.
|
||||
/// 2. **The in-flight registry.** [`run_registry::is_in_flight`] gates each
|
||||
/// surviving candidate. Both run entry points now register **before**
|
||||
/// inserting the row, so within this process a `running` row is never
|
||||
/// unregistered; this guard covers clock skew and rows stamped by a
|
||||
/// differently-skewed process.
|
||||
///
|
||||
/// The two are deliberately redundant: either alone would be sufficient today,
|
||||
/// and neither depends on the other's ordering assumption holding.
|
||||
///
|
||||
/// Each swept run also updates the flow summary, announces a terminal
|
||||
/// `FlowRunFinished`, and drops its durable checkpoint (a `running` row is never
|
||||
/// resumable — only `pending_approval` is). Best-effort by construction: a store
|
||||
/// error is logged and the sweep returns what it managed.
|
||||
pub async fn sweep_orphaned_running_runs_on_boot(config: &Config) -> usize {
|
||||
let now_str = Utc::now().to_rfc3339();
|
||||
const REASON: &str =
|
||||
"Run interrupted by an app restart — no live run was executing this row after boot.";
|
||||
|
||||
let floor: &str = PROCESS_RUN_FLOOR.as_str();
|
||||
tracing::debug!(target: "flows", floor, "[flows] boot sweep: reconciling only runs started before this process");
|
||||
let candidates = match store::list_running_run_ids(config, floor) {
|
||||
Ok(candidates) => candidates,
|
||||
Err(e) => {
|
||||
tracing::warn!(target: "flows", error = %e, "[flows] boot sweep: failed to list running runs (skipping)");
|
||||
return 0;
|
||||
}
|
||||
};
|
||||
if candidates.is_empty() {
|
||||
return 0;
|
||||
}
|
||||
tracing::debug!(target: "flows", count = candidates.len(), "[flows] boot sweep: examining running rows for orphans");
|
||||
|
||||
let mut swept = 0usize;
|
||||
for (run_id, flow_id) in candidates {
|
||||
if run_registry::is_in_flight(&run_id) {
|
||||
tracing::debug!(target: "flows", run_id = %run_id, flow_id = %flow_id, "[flows] boot sweep: run is live in-process — leaving it running");
|
||||
continue;
|
||||
}
|
||||
match store::mark_run_interrupted(config, &run_id, &now_str, REASON) {
|
||||
Ok(true) => {
|
||||
swept += 1;
|
||||
if let Err(e) = store::record_run(config, &flow_id, "interrupted") {
|
||||
tracing::warn!(target: "flows", run_id = %run_id, flow_id = %flow_id, error = %e, "[flows] boot sweep: failed to update flow summary for reconciled run");
|
||||
}
|
||||
crate::core::event_bus::publish_global(
|
||||
crate::core::event_bus::DomainEvent::FlowRunFinished {
|
||||
flow_id: flow_id.clone(),
|
||||
run_id: run_id.clone(),
|
||||
status: "interrupted".to_string(),
|
||||
},
|
||||
);
|
||||
drop_checkpoint(config, &run_id).await;
|
||||
tracing::info!(target: "flows", run_id = %run_id, flow_id = %flow_id, "[flows] boot sweep: reconciled orphaned running run to 'interrupted'");
|
||||
}
|
||||
Ok(false) => {
|
||||
tracing::debug!(target: "flows", run_id = %run_id, "[flows] boot sweep: row changed status concurrently — skipped");
|
||||
}
|
||||
Err(e) => {
|
||||
tracing::warn!(target: "flows", run_id = %run_id, error = %e, "[flows] boot sweep: failed to reconcile running run");
|
||||
}
|
||||
}
|
||||
}
|
||||
if swept > 0 {
|
||||
tracing::info!(target: "flows", count = swept, "[flows] boot sweep reconciled orphaned running runs to 'interrupted'");
|
||||
}
|
||||
swept
|
||||
}
|
||||
|
||||
/// Cancels a flow run (issue G4), settling it to a terminal `"cancelled"`
|
||||
/// status and dropping its durable checkpoint so the aborted thread can never
|
||||
/// be resumed.
|
||||
@@ -4310,9 +4700,11 @@ pub async fn sweep_expired_parked_runs(config: &Config) -> usize {
|
||||
/// this settles the row terminally itself and drops the checkpoint.
|
||||
///
|
||||
/// A run that is already terminal (`completed` / `completed_with_warnings` /
|
||||
/// `failed` / `cancelled`) is a clear error, not a silent no-op — otherwise a
|
||||
/// settled warning run could be overwritten as `"cancelled"`, corrupting the
|
||||
/// run-honesty status it already recorded.
|
||||
/// `failed` / `cancelled` / `interrupted`) is a clear error, not a silent
|
||||
/// no-op — otherwise a settled warning run could be overwritten as
|
||||
/// `"cancelled"`, corrupting the run-honesty status it already recorded, and an
|
||||
/// already-`interrupted` run (reconciled by the drop-guard / boot sweep, bug
|
||||
/// B42) could be clobbered back to `"cancelled"`.
|
||||
pub async fn flows_cancel_run(config: &Config, run_id: &str) -> Result<RpcOutcome<Value>, String> {
|
||||
let run = store::get_flow_run(config, run_id)
|
||||
.map_err(|e| e.to_string())?
|
||||
@@ -4320,7 +4712,7 @@ pub async fn flows_cancel_run(config: &Config, run_id: &str) -> Result<RpcOutcom
|
||||
|
||||
if matches!(
|
||||
run.status.as_str(),
|
||||
"completed" | "completed_with_warnings" | "failed" | "cancelled"
|
||||
"completed" | "completed_with_warnings" | "failed" | "cancelled" | "interrupted"
|
||||
) {
|
||||
return Err(format!(
|
||||
"flow run '{run_id}' is already terminal (status: {}) — nothing to cancel",
|
||||
@@ -4402,10 +4794,35 @@ fn workflow_origin(flow_id: &str, require_approval: bool) -> AgentTurnOrigin {
|
||||
}
|
||||
}
|
||||
|
||||
/// RFC3339 instant at which THIS process first entered the flow-run lifecycle —
|
||||
/// the floor the boot orphan sweep (bug B42) uses to bound its candidate set.
|
||||
///
|
||||
/// Initialized on first touch by whichever comes first: [`start_flow_run_row`]
|
||||
/// (which forces it *before* stamping the row it is about to insert) or
|
||||
/// [`sweep_orphaned_running_runs_on_boot`]. Either ordering yields the same
|
||||
/// invariant — **every `flow_runs` row this process inserts has
|
||||
/// `started_at >= *PROCESS_RUN_FLOOR`** — so a sweep restricted to
|
||||
/// `started_at < *PROCESS_RUN_FLOOR` provably only ever sees rows left behind by
|
||||
/// a *prior* process.
|
||||
///
|
||||
/// The floor makes that guarantee structural rather than a consequence of
|
||||
/// registration ordering. `run_registry::is_in_flight` alone once left a window
|
||||
/// — the entry points used to insert the `running` row before `run_flow_body`
|
||||
/// registered, so a live run was briefly `running`-but-not-in-flight, and
|
||||
/// sweeping it there would `drop_checkpoint` it mid-run (unrecoverable, unlike
|
||||
/// the status, which the live run's own terminal write would fix). Registration
|
||||
/// has since moved ahead of the insert, closing that window at the source too;
|
||||
/// the floor stays because it holds regardless of what future callers do with
|
||||
/// that ordering.
|
||||
static PROCESS_RUN_FLOOR: LazyLock<String> = LazyLock::new(|| Utc::now().to_rfc3339());
|
||||
|
||||
/// Best-effort insert of the initial `"running"` `flow_runs` row. Logged,
|
||||
/// never fails the run — run-history persistence is an observability aid,
|
||||
/// not a correctness requirement of the run itself.
|
||||
fn start_flow_run_row(config: &Config, thread_id: &str, flow_id: &str) {
|
||||
// Anchor the boot-sweep floor BEFORE stamping this row, so this row's
|
||||
// `started_at` can never precede it. See [`PROCESS_RUN_FLOOR`].
|
||||
LazyLock::force(&PROCESS_RUN_FLOOR);
|
||||
let started_at = Utc::now().to_rfc3339();
|
||||
if let Err(e) = store::insert_flow_run(config, thread_id, flow_id, thread_id, &started_at) {
|
||||
tracing::warn!(target: "flows", flow_id, thread_id, error = %e, "[flows] failed to persist flow run start");
|
||||
|
||||
@@ -2608,6 +2608,49 @@ async fn flows_cancel_run_of_a_completed_with_warnings_run_errors() {
|
||||
assert_eq!(run_row.value.status, "completed_with_warnings");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn flows_cancel_run_of_an_interrupted_run_errors() {
|
||||
// An `interrupted` run (bug B42 — reconciled by the drop-guard / boot
|
||||
// sweep) is terminal: cancelling it must be a clear error, never fall
|
||||
// through to the not-in-flight path and clobber the row to `"cancelled"`,
|
||||
// discarding the interruption reason it already carries.
|
||||
let tmp = TempDir::new().unwrap();
|
||||
let config = test_config(&tmp);
|
||||
let created = flows_create(&config, "demo".to_string(), trigger_only_graph(), false)
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let run = flows_run(&config, &created.value.id, json!({}), FlowRunTrigger::Rpc)
|
||||
.await
|
||||
.unwrap();
|
||||
let thread_id = run.value["thread_id"].as_str().unwrap().to_string();
|
||||
|
||||
// Force the settled row to `interrupted` directly.
|
||||
store::finish_flow_run(
|
||||
&config,
|
||||
&thread_id,
|
||||
"interrupted",
|
||||
&chrono::Utc::now().to_rfc3339(),
|
||||
&[],
|
||||
&[],
|
||||
Some("interrupted mid-flight"),
|
||||
)
|
||||
.unwrap();
|
||||
|
||||
let err = flows_cancel_run(&config, &thread_id)
|
||||
.await
|
||||
.expect_err("cancelling an interrupted run must be a clear error");
|
||||
assert!(err.contains("already terminal"), "got: {err}");
|
||||
|
||||
// And the row must still read back as `interrupted`, not overwritten.
|
||||
let run_row = flows_get_run(&config, &thread_id).await.unwrap();
|
||||
assert_eq!(run_row.value.status, "interrupted");
|
||||
assert_eq!(
|
||||
run_row.value.error.as_deref(),
|
||||
Some("interrupted mid-flight")
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn flows_cancel_run_missing_run_errors() {
|
||||
let tmp = TempDir::new().unwrap();
|
||||
@@ -6318,3 +6361,230 @@ fn combine_trail_off_fallback_returns_fallback_alone_for_genuine_silence() {
|
||||
assert_eq!(combine_trail_off_fallback(&fallback, ""), fallback);
|
||||
assert_eq!(combine_trail_off_fallback(&fallback, " \n\n "), fallback);
|
||||
}
|
||||
|
||||
// ── Live-run reliability: drop-guard + boot sweep + detach (bugs B41/B42) ───
|
||||
|
||||
/// Seeds a real flow plus an already-inserted `running` `flow_runs` row, and
|
||||
/// returns `(config, flow_id, run_id)`. The `TempDir` is returned so the caller
|
||||
/// keeps the on-disk store alive for the duration of the test.
|
||||
fn seed_running_run(tmp: &TempDir) -> (Config, String, String) {
|
||||
let config = test_config(tmp);
|
||||
let flow = store::create_flow(
|
||||
&config,
|
||||
"reliability".to_string(),
|
||||
structurally_valid_graph(trigger_only_graph()),
|
||||
false,
|
||||
true,
|
||||
)
|
||||
.unwrap();
|
||||
let run_id = format!("flow:{}:{}", flow.id, uuid::Uuid::new_v4());
|
||||
// Stamped well before `PROCESS_RUN_FLOOR` so this row models what the boot
|
||||
// sweep actually targets: a `running` row left behind by a *prior* process.
|
||||
// Using `Utc::now()` here would make the sweep tests order-dependent — the
|
||||
// floor is a process-wide `LazyLock`, so a sibling test that ran a real
|
||||
// flow first would push it past a "now" seed and the row would (correctly)
|
||||
// fall out of the candidate set.
|
||||
store::insert_flow_run(
|
||||
&config,
|
||||
&run_id,
|
||||
&flow.id,
|
||||
&run_id,
|
||||
PRIOR_PROCESS_STARTED_AT,
|
||||
)
|
||||
.unwrap();
|
||||
(config, flow.id, run_id)
|
||||
}
|
||||
|
||||
/// A `started_at` that provably predates this process's `PROCESS_RUN_FLOOR`.
|
||||
const PRIOR_PROCESS_STARTED_AT: &str = "2020-01-01T00:00:00+00:00";
|
||||
|
||||
#[test]
|
||||
fn run_row_finalizer_reconciles_orphaned_running_row_to_interrupted_on_drop() {
|
||||
let tmp = TempDir::new().unwrap();
|
||||
let (config, flow_id, run_id) = seed_running_run(&tmp);
|
||||
|
||||
// Simulate the run future being dropped mid-await without any terminal
|
||||
// write: the guard is created armed and never disarmed, so its `Drop`
|
||||
// reconciles the row.
|
||||
{
|
||||
let _finalizer = RunRowFinalizer::new(Arc::new(config.clone()), &run_id, &flow_id);
|
||||
}
|
||||
|
||||
let row = store::get_flow_run(&config, &run_id).unwrap().unwrap();
|
||||
assert_eq!(
|
||||
row.status, "interrupted",
|
||||
"a dropped run must not stay 'running'"
|
||||
);
|
||||
assert_eq!(row.error.as_deref(), Some(INTERRUPTED_DROP_REASON));
|
||||
assert!(
|
||||
row.finished_at.is_some(),
|
||||
"an interrupted run must be stamped finished"
|
||||
);
|
||||
|
||||
// The flow-definition summary must track the row, like every other
|
||||
// terminal path — otherwise the runs list keeps advertising the previous
|
||||
// run's status for a flow whose latest run was interrupted.
|
||||
let flow = store::get_flow(&config, &flow_id).unwrap().unwrap();
|
||||
assert_eq!(
|
||||
flow.last_status.as_deref(),
|
||||
Some("interrupted"),
|
||||
"the drop-guard must update the flow summary, not just the run row"
|
||||
);
|
||||
assert!(
|
||||
flow.last_run_at.is_some(),
|
||||
"the drop-guard must stamp last_run_at"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn run_row_finalizer_disarm_leaves_a_settled_row_untouched() {
|
||||
let tmp = TempDir::new().unwrap();
|
||||
let (config, flow_id, run_id) = seed_running_run(&tmp);
|
||||
|
||||
// A run that settled normally disarms its guard after the real terminal
|
||||
// write; dropping the disarmed guard must be a no-op.
|
||||
{
|
||||
let finalizer = RunRowFinalizer::new(Arc::new(config.clone()), &run_id, &flow_id);
|
||||
finalizer.disarm();
|
||||
}
|
||||
|
||||
let row = store::get_flow_run(&config, &run_id).unwrap().unwrap();
|
||||
assert_eq!(
|
||||
row.status, "running",
|
||||
"a disarmed finalizer must not overwrite the row's real status"
|
||||
);
|
||||
assert!(row.error.is_none());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn boot_sweep_reconciles_orphaned_running_run_to_interrupted() {
|
||||
let tmp = TempDir::new().unwrap();
|
||||
let (config, _flow_id, run_id) = seed_running_run(&tmp);
|
||||
|
||||
// No in-process run owns this row (the registry is empty), so the boot
|
||||
// sweep must reconcile it.
|
||||
let swept = sweep_orphaned_running_runs_on_boot(&config).await;
|
||||
assert_eq!(swept, 1, "the orphaned running row must be swept");
|
||||
|
||||
let row = store::get_flow_run(&config, &run_id).unwrap().unwrap();
|
||||
assert_eq!(row.status, "interrupted");
|
||||
assert!(
|
||||
row.error
|
||||
.as_deref()
|
||||
.is_some_and(|e| e.contains("app restart")),
|
||||
"the reason must explain the boot reconciliation, got {:?}",
|
||||
row.error
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn boot_sweep_skips_a_run_that_is_live_in_flight() {
|
||||
let tmp = TempDir::new().unwrap();
|
||||
let (config, _flow_id, run_id) = seed_running_run(&tmp);
|
||||
|
||||
// Register the run as live in this process; the sweep must leave it alone.
|
||||
let (_token, _guard) = run_registry::register(&run_id);
|
||||
assert!(run_registry::is_in_flight(&run_id));
|
||||
|
||||
let swept = sweep_orphaned_running_runs_on_boot(&config).await;
|
||||
assert_eq!(swept, 0, "a live in-flight run must never be swept");
|
||||
|
||||
let row = store::get_flow_run(&config, &run_id).unwrap().unwrap();
|
||||
assert_eq!(row.status, "running", "the live run must stay running");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn boot_sweep_skips_a_run_started_after_the_process_floor() {
|
||||
let tmp = TempDir::new().unwrap();
|
||||
let (config, flow_id, _prior_run_id) = seed_running_run(&tmp);
|
||||
|
||||
// A row this process inserted, but NOT yet registered in the run registry —
|
||||
// exactly the TOCTOU window between `start_flow_run_row` and
|
||||
// `run_registry::register`. The `is_in_flight` guard does not cover it; the
|
||||
// `PROCESS_RUN_FLOOR` floor must. Sweeping it would flip a live run to
|
||||
// `interrupted` AND drop its durable checkpoint mid-run.
|
||||
let live_run_id = format!("flow:{flow_id}:{}", uuid::Uuid::new_v4());
|
||||
start_flow_run_row(&config, &live_run_id, &flow_id);
|
||||
assert!(
|
||||
!run_registry::is_in_flight(&live_run_id),
|
||||
"the row must be unregistered for this test to exercise the window"
|
||||
);
|
||||
|
||||
let swept = sweep_orphaned_running_runs_on_boot(&config).await;
|
||||
|
||||
let live = store::get_flow_run(&config, &live_run_id).unwrap().unwrap();
|
||||
assert_eq!(
|
||||
live.status, "running",
|
||||
"a run started by THIS process must never be swept, registered or not"
|
||||
);
|
||||
assert_eq!(
|
||||
swept, 1,
|
||||
"only the prior-process orphan may be reconciled, got {swept}"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn flows_run_detached_returns_running_run_id_and_inserts_row() {
|
||||
let tmp = TempDir::new().unwrap();
|
||||
let config = test_config(&tmp);
|
||||
let flow = store::create_flow(
|
||||
&config,
|
||||
"detached".to_string(),
|
||||
structurally_valid_graph(trigger_only_graph()),
|
||||
false,
|
||||
true,
|
||||
)
|
||||
.unwrap();
|
||||
|
||||
let outcome = flows_run_detached(&config, &flow.id, json!({}), FlowRunTrigger::Rpc)
|
||||
.await
|
||||
.expect("detached run must start");
|
||||
|
||||
assert_eq!(outcome.value["status"], json!("running"));
|
||||
assert_eq!(outcome.value["detached"], json!(true));
|
||||
let run_id = outcome.value["run_id"]
|
||||
.as_str()
|
||||
.expect("run_id must be a string")
|
||||
.to_string();
|
||||
assert!(
|
||||
run_id.starts_with(&format!("flow:{}:", flow.id)),
|
||||
"run_id: {run_id}"
|
||||
);
|
||||
|
||||
// The `running` row is inserted synchronously before the background task is
|
||||
// spawned, so the copilot's immediate `get_flow_run(run_id)` poll finds it.
|
||||
let row = store::get_flow_run(&config, &run_id)
|
||||
.unwrap()
|
||||
.expect("a run row must exist immediately after detaching");
|
||||
assert_eq!(row.flow_id, flow.id);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn flows_run_detached_registers_the_run_before_returning_its_id() {
|
||||
let tmp = TempDir::new().unwrap();
|
||||
let config = test_config(&tmp);
|
||||
let flow = store::create_flow(
|
||||
&config,
|
||||
"detached-cancel-race".to_string(),
|
||||
structurally_valid_graph(trigger_only_graph()),
|
||||
false,
|
||||
true,
|
||||
)
|
||||
.unwrap();
|
||||
|
||||
let outcome = flows_run_detached(&config, &flow.id, json!({}), FlowRunTrigger::Rpc)
|
||||
.await
|
||||
.expect("detached run must start");
|
||||
let run_id = outcome.value["run_id"].as_str().unwrap().to_string();
|
||||
|
||||
// The moment the agent can see this `run_id` it can be cancelled. If
|
||||
// registration happened inside the spawned task instead, this would be
|
||||
// false until the task was first polled — and `flows_cancel_run` would take
|
||||
// its "parked/stale" branch, writing a terminal `cancelled` row and
|
||||
// dropping the checkpoint while the background run went on to execute the
|
||||
// flow's real side effects and overwrite that status.
|
||||
assert!(
|
||||
run_registry::is_in_flight(&run_id),
|
||||
"a detached run must be registered before its run_id is returned"
|
||||
);
|
||||
}
|
||||
|
||||
@@ -64,6 +64,18 @@ pub(crate) fn cancel(run_id: &str) -> bool {
|
||||
}
|
||||
}
|
||||
|
||||
/// Returns `true` when `run_id` is currently registered as an in-flight run in
|
||||
/// THIS process. Used by the boot-time orphan sweep (bug B42) to distinguish a
|
||||
/// genuinely orphaned `running` row (left by a prior process — not in flight)
|
||||
/// from one a freshly-started run in this process legitimately owns, so the
|
||||
/// sweep never reconciles a live run out from under itself.
|
||||
pub(crate) fn is_in_flight(run_id: &str) -> bool {
|
||||
IN_FLIGHT
|
||||
.lock()
|
||||
.unwrap_or_else(|e| e.into_inner())
|
||||
.contains_key(run_id)
|
||||
}
|
||||
|
||||
/// RAII guard that removes a run's entry from the in-flight registry on drop.
|
||||
pub(crate) struct RunGuard {
|
||||
run_id: String,
|
||||
|
||||
@@ -819,6 +819,61 @@ pub fn expire_parked_runs(
|
||||
})
|
||||
}
|
||||
|
||||
/// Lists the `(id, flow_id)` of every run persisted at `status = 'running'`
|
||||
/// whose `started_at` is strictly **before** `started_before` (RFC3339). Used by
|
||||
/// the boot-time orphan sweep (bug B42): after a crash/restart no in-process
|
||||
/// task is executing these rows, so
|
||||
/// [`crate::openhuman::flows::ops::sweep_orphaned_running_runs_on_boot`]
|
||||
/// reconciles each one that isn't backed by a live in-flight run to a terminal
|
||||
/// `'interrupted'` via [`mark_run_interrupted`].
|
||||
///
|
||||
/// The `started_before` floor is what makes the sweep provably unable to touch
|
||||
/// a run **this** process started: the sweep passes the instant this process
|
||||
/// first entered the flow-run lifecycle, and every row this process inserts is
|
||||
/// stamped at or after that instant. Without it, the sweep's only guard is the
|
||||
/// in-flight registry, which a row briefly escapes between `start_flow_run_row`
|
||||
/// and `run_registry::register`. `started_at` is a fixed-shape UTC RFC3339
|
||||
/// string, so the lexicographic `<` matches chronological order (same
|
||||
/// comparison the parked-run TTL sweep already relies on).
|
||||
pub fn list_running_run_ids(
|
||||
config: &Config,
|
||||
started_before: &str,
|
||||
) -> Result<Vec<(String, String)>> {
|
||||
with_connection(config, |conn| {
|
||||
let mut stmt = conn.prepare(
|
||||
"SELECT id, flow_id FROM flow_runs WHERE status = 'running' AND started_at < ?1",
|
||||
)?;
|
||||
let rows: Vec<(String, String)> = stmt
|
||||
.query_map(params![started_before], |row| {
|
||||
Ok((row.get(0)?, row.get(1)?))
|
||||
})?
|
||||
.collect::<rusqlite::Result<_>>()?;
|
||||
Ok(rows)
|
||||
})
|
||||
}
|
||||
|
||||
/// Reconciles a single orphaned `'running'` run row to a terminal
|
||||
/// `'interrupted'` status stamped `now` (RFC3339) with `reason`, guarded by a
|
||||
/// `status = 'running'` predicate so a run that settled or was resumed
|
||||
/// concurrently is never clobbered. Returns `true` when a row was actually
|
||||
/// flipped (bug B42 — cancellation-safe finalizer + boot sweep). Best-effort by
|
||||
/// contract at the call site.
|
||||
pub fn mark_run_interrupted(config: &Config, id: &str, now: &str, reason: &str) -> Result<bool> {
|
||||
with_connection(config, |conn| {
|
||||
let changed = conn
|
||||
.execute(
|
||||
"UPDATE flow_runs SET status = 'interrupted', finished_at = ?1, error = ?2 \
|
||||
WHERE id = ?3 AND status = 'running'",
|
||||
params![now, reason, id],
|
||||
)
|
||||
.context("Failed to reconcile orphaned running flow run")?;
|
||||
if changed > 0 {
|
||||
tracing::info!(target: "flows", run_id = id, "[flows] reconciled orphaned 'running' flow run to 'interrupted'");
|
||||
}
|
||||
Ok(changed > 0)
|
||||
})
|
||||
}
|
||||
|
||||
/// Loads one flow run by id (== thread_id).
|
||||
pub fn get_flow_run(config: &Config, id: &str) -> Result<Option<FlowRun>> {
|
||||
with_connection(config, |conn| {
|
||||
|
||||
@@ -782,3 +782,149 @@ fn upsert_suggestions_empty_is_noop() {
|
||||
let config = test_config(&tmp);
|
||||
assert_eq!(upsert_suggestions(&config, &[]).unwrap(), 0);
|
||||
}
|
||||
|
||||
// ── Orphaned-running-run reconciliation (bug B42) ──────────────────────────
|
||||
|
||||
#[test]
|
||||
fn list_running_run_ids_returns_only_running_rows() {
|
||||
let tmp = TempDir::new().unwrap();
|
||||
let config = test_config(&tmp);
|
||||
let flow = create_flow(&config, "demo".to_string(), trigger_graph(), false, true).unwrap();
|
||||
|
||||
insert_flow_run(
|
||||
&config,
|
||||
"run-live-1",
|
||||
&flow.id,
|
||||
"run-live-1",
|
||||
"2026-01-01T00:00:00Z",
|
||||
)
|
||||
.unwrap();
|
||||
insert_flow_run(
|
||||
&config,
|
||||
"run-live-2",
|
||||
&flow.id,
|
||||
"run-live-2",
|
||||
"2026-01-01T00:00:01Z",
|
||||
)
|
||||
.unwrap();
|
||||
insert_flow_run(
|
||||
&config,
|
||||
"run-done",
|
||||
&flow.id,
|
||||
"run-done",
|
||||
"2026-01-01T00:00:02Z",
|
||||
)
|
||||
.unwrap();
|
||||
finish_flow_run(
|
||||
&config,
|
||||
"run-done",
|
||||
"completed",
|
||||
"2026-01-01T00:00:03Z",
|
||||
&[],
|
||||
&[],
|
||||
None,
|
||||
)
|
||||
.unwrap();
|
||||
|
||||
let mut running = list_running_run_ids(&config, "2099-01-01T00:00:00Z").unwrap();
|
||||
running.sort();
|
||||
assert_eq!(
|
||||
running,
|
||||
vec![
|
||||
("run-live-1".to_string(), flow.id.clone()),
|
||||
("run-live-2".to_string(), flow.id.clone()),
|
||||
],
|
||||
"only the two still-running rows must be listed, not the completed one"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn list_running_run_ids_excludes_rows_started_at_or_after_the_floor() {
|
||||
let tmp = TempDir::new().unwrap();
|
||||
let config = test_config(&tmp);
|
||||
let flow = create_flow(&config, "demo".to_string(), trigger_graph(), false, true).unwrap();
|
||||
|
||||
insert_flow_run(
|
||||
&config,
|
||||
"run-old",
|
||||
&flow.id,
|
||||
"run-old",
|
||||
"2026-01-01T00:00:00Z",
|
||||
)
|
||||
.unwrap();
|
||||
insert_flow_run(
|
||||
&config,
|
||||
"run-at",
|
||||
&flow.id,
|
||||
"run-at",
|
||||
"2026-01-01T00:00:05Z",
|
||||
)
|
||||
.unwrap();
|
||||
insert_flow_run(
|
||||
&config,
|
||||
"run-new",
|
||||
&flow.id,
|
||||
"run-new",
|
||||
"2026-01-01T00:00:09Z",
|
||||
)
|
||||
.unwrap();
|
||||
|
||||
// The floor is exclusive: a row stamped exactly at the boot floor was
|
||||
// inserted by THIS process (`start_flow_run_row` anchors the floor before
|
||||
// stamping), so it must fall outside the candidate set along with newer
|
||||
// rows — otherwise the sweep could interrupt a live run and drop its
|
||||
// checkpoint mid-flight.
|
||||
let running = list_running_run_ids(&config, "2026-01-01T00:00:05Z").unwrap();
|
||||
assert_eq!(
|
||||
running,
|
||||
vec![("run-old".to_string(), flow.id.clone())],
|
||||
"only rows strictly older than the floor are sweep candidates"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn mark_run_interrupted_reconciles_a_running_row_with_reason() {
|
||||
let tmp = TempDir::new().unwrap();
|
||||
let config = test_config(&tmp);
|
||||
let flow = create_flow(&config, "demo".to_string(), trigger_graph(), false, true).unwrap();
|
||||
insert_flow_run(&config, "run-x", &flow.id, "run-x", "2026-01-01T00:00:00Z").unwrap();
|
||||
|
||||
let flipped =
|
||||
mark_run_interrupted(&config, "run-x", "2026-01-01T00:05:00Z", "boom reason").unwrap();
|
||||
assert!(flipped, "a running row must be reconciled");
|
||||
|
||||
let row = get_flow_run(&config, "run-x").unwrap().unwrap();
|
||||
assert_eq!(row.status, "interrupted");
|
||||
assert_eq!(row.finished_at.as_deref(), Some("2026-01-01T00:05:00Z"));
|
||||
assert_eq!(row.error.as_deref(), Some("boom reason"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn mark_run_interrupted_is_a_noop_for_a_terminal_row() {
|
||||
let tmp = TempDir::new().unwrap();
|
||||
let config = test_config(&tmp);
|
||||
let flow = create_flow(&config, "demo".to_string(), trigger_graph(), false, true).unwrap();
|
||||
insert_flow_run(&config, "run-y", &flow.id, "run-y", "2026-01-01T00:00:00Z").unwrap();
|
||||
finish_flow_run(
|
||||
&config,
|
||||
"run-y",
|
||||
"completed",
|
||||
"2026-01-01T00:00:01Z",
|
||||
&[],
|
||||
&[],
|
||||
None,
|
||||
)
|
||||
.unwrap();
|
||||
|
||||
// The `status = 'running'` guard must protect an already-settled run.
|
||||
let flipped =
|
||||
mark_run_interrupted(&config, "run-y", "2026-01-01T00:05:00Z", "should not apply").unwrap();
|
||||
assert!(
|
||||
!flipped,
|
||||
"a completed run must never be clobbered to interrupted"
|
||||
);
|
||||
|
||||
let row = get_flow_run(&config, "run-y").unwrap().unwrap();
|
||||
assert_eq!(row.status, "completed");
|
||||
assert!(row.error.is_none());
|
||||
}
|
||||
|
||||
@@ -294,10 +294,18 @@ impl Tool for RunFlowTool {
|
||||
tracing::info!(
|
||||
target: "flows",
|
||||
%flow_id,
|
||||
"[flows] run_flow: agent-initiated test run starting"
|
||||
"[flows] run_flow: agent-initiated test run starting (detached)"
|
||||
);
|
||||
|
||||
match crate::openhuman::flows::ops::flows_run(
|
||||
// Detach (bug B41): a flow whose first real node is a live-research
|
||||
// agent node inherently runs longer than the tinyagents harness's 120s
|
||||
// per-tool-call cap, so a blocking `run_flow` could never succeed —
|
||||
// it died at 120s, orphaning the run row (bug B42). `flows_run_detached`
|
||||
// validates + compile-checks synchronously (so a broken flow still
|
||||
// returns an immediate, actionable error), fires the run on a background
|
||||
// task, and returns `{ run_id, status: "running" }` in well under 120s.
|
||||
// The copilot then polls `get_flow_run(run_id)` to observe completion.
|
||||
match crate::openhuman::flows::ops::flows_run_detached(
|
||||
&self.config,
|
||||
&flow_id,
|
||||
input,
|
||||
@@ -305,15 +313,30 @@ impl Tool for RunFlowTool {
|
||||
)
|
||||
.await
|
||||
{
|
||||
Ok(outcome) => Ok(ToolResult::success(serde_json::to_string_pretty(&json!({
|
||||
"type": "workflow_run_result",
|
||||
"flow_id": flow_id,
|
||||
"result": outcome.value,
|
||||
}))?)),
|
||||
Ok(outcome) => {
|
||||
let run_id = outcome
|
||||
.value
|
||||
.get("run_id")
|
||||
.and_then(Value::as_str)
|
||||
.unwrap_or_default()
|
||||
.to_string();
|
||||
Ok(ToolResult::success(serde_json::to_string_pretty(&json!({
|
||||
"type": "workflow_run_started",
|
||||
"flow_id": flow_id,
|
||||
"run_id": run_id,
|
||||
"status": "running",
|
||||
"detached": true,
|
||||
"note": "The run started in the background and is now 'running'. Poll \
|
||||
get_flow_run with this run_id to see it settle to a terminal \
|
||||
status (completed / failed / interrupted / pending_approval); \
|
||||
do not assume success from this response alone.",
|
||||
"result": outcome.value,
|
||||
}))?))
|
||||
}
|
||||
Err(e) => {
|
||||
tracing::debug!(target: "flows", %flow_id, error = %e, "[flows] run_flow: failed");
|
||||
tracing::debug!(target: "flows", %flow_id, error = %e, "[flows] run_flow: failed to start");
|
||||
Ok(ToolResult::error(format!(
|
||||
"Could not run flow '{flow_id}': {e}"
|
||||
"Could not start flow '{flow_id}': {e}"
|
||||
)))
|
||||
}
|
||||
}
|
||||
|
||||
@@ -328,17 +328,23 @@ pub struct FlowRun {
|
||||
/// `"running"` | `"completed"` | `"completed_with_warnings"` |
|
||||
/// `"pending_approval"` | `"failed"` | `"cancelled"` (issue G4 — a run
|
||||
/// cancelled via `flows_cancel_run`, or a parked `pending_approval` run
|
||||
/// swept by the TTL expiry). `"completed_with_warnings"` (run honesty,
|
||||
/// swept by the TTL expiry) | `"interrupted"` (bug B42 — a run whose future
|
||||
/// was dropped mid-flight, reconciled either by the in-process
|
||||
/// `RunRowFinalizer` drop-guard or the boot-time orphan sweep, so a
|
||||
/// cancelled/timed-out/crashed run always settles to a terminal row instead
|
||||
/// of wedging at `running`). `"completed_with_warnings"` (run honesty,
|
||||
/// PR2) is a terminal status like `"completed"`, but at least one settled
|
||||
/// [`FlowRunStep`] carries non-empty `diagnostics` (a `=`-binding that
|
||||
/// resolved to `null`) even though no step outright errored. All of
|
||||
/// `completed` / `completed_with_warnings` / `failed` / `cancelled` are
|
||||
/// terminal.
|
||||
/// `completed` / `completed_with_warnings` / `failed` / `cancelled` /
|
||||
/// `interrupted` are terminal.
|
||||
pub status: String,
|
||||
/// RFC3339 timestamp when the run started.
|
||||
pub started_at: String,
|
||||
/// RFC3339 timestamp when the run last settled (completed/paused/failed).
|
||||
/// `None` while a run row is still `"running"`.
|
||||
/// RFC3339 timestamp when the run last settled — stamped for every terminal
|
||||
/// status (completed/paused/failed/cancelled/`"interrupted"`; the B42
|
||||
/// drop-guard and boot sweep stamp it exactly like a normal terminal
|
||||
/// write). `None` only while a run row is still `"running"`.
|
||||
pub finished_at: Option<String>,
|
||||
/// Reconstructed per-node steps (see [`FlowRunStep`]).
|
||||
#[serde(default)]
|
||||
@@ -347,7 +353,11 @@ pub struct FlowRun {
|
||||
/// "pending_approval"`; empty otherwise.
|
||||
#[serde(default)]
|
||||
pub pending_approvals: Vec<String>,
|
||||
/// Error message when `status == "failed"`.
|
||||
/// Human-readable failure reason. Set when `status == "failed"`, and also
|
||||
/// when `status == "interrupted"` (bug B42) — there it carries the
|
||||
/// reconciliation reason (tool abort / chat turn end / app restart) so the
|
||||
/// run-details sidebar can explain *why* the run stopped instead of
|
||||
/// rendering a bare terminal state.
|
||||
#[serde(default)]
|
||||
pub error: Option<String>,
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user