mirror of
https://github.com/tinyhumansai/openhuman.git
synced 2026-07-27 21:08:00 +00:00
feat: proactive agent task pipeline — dispatcher, approval, executors, write-back (#2974)
Co-authored-by: Claude <noreply@anthropic.com> Co-authored-by: Steven Enamakel <enamakel@tinyhumans.ai>
This commit is contained in:
sanil-23
co-authored by
Claude
Steven Enamakel
Claude
Steven Enamakel
parent
1ecc42d6b4
commit
886c4a6342
@@ -552,6 +552,9 @@ const ar5: TranslationMap = {
|
||||
'skills.uninstall.confirmTitle': 'إلغاء تثبيت {name}؟',
|
||||
'conversations.taskKanban.blocked': 'محظور',
|
||||
'conversations.taskKanban.done': 'مكتمل',
|
||||
'conversations.taskKanban.awaitingApproval': 'Awaiting approval',
|
||||
'conversations.taskKanban.ready': 'Ready',
|
||||
'conversations.taskKanban.rejected': 'Rejected',
|
||||
'conversations.taskKanban.inProgress': 'قيد التنفيذ',
|
||||
'intelligence.memoryChunk.detail.copiedHint': 'تم النسخ',
|
||||
'settings.composio.notYetRouted': 'لم يتم توجيهه بعد',
|
||||
|
||||
@@ -559,6 +559,9 @@ const bn5: TranslationMap = {
|
||||
'skills.uninstall.confirmTitle': '{name} আনইনস্টল করবেন?',
|
||||
'conversations.taskKanban.blocked': 'ব্লকড',
|
||||
'conversations.taskKanban.done': 'সম্পন্ন',
|
||||
'conversations.taskKanban.awaitingApproval': 'Awaiting approval',
|
||||
'conversations.taskKanban.ready': 'Ready',
|
||||
'conversations.taskKanban.rejected': 'Rejected',
|
||||
'conversations.taskKanban.inProgress': 'চলমান',
|
||||
'intelligence.memoryChunk.detail.copiedHint': 'কপি হয়েছে',
|
||||
'settings.composio.notYetRouted': 'এখনও রুট করা হয়নি',
|
||||
|
||||
@@ -582,6 +582,9 @@ const de5: TranslationMap = {
|
||||
'skills.uninstall.confirmTitle': '{name} deinstallieren?',
|
||||
'conversations.taskKanban.blocked': 'Blockiert',
|
||||
'conversations.taskKanban.done': 'Fertig',
|
||||
'conversations.taskKanban.awaitingApproval': 'Awaiting approval',
|
||||
'conversations.taskKanban.ready': 'Ready',
|
||||
'conversations.taskKanban.rejected': 'Rejected',
|
||||
'conversations.taskKanban.inProgress': 'In Bearbeitung',
|
||||
'intelligence.memoryChunk.detail.copiedHint': 'kopiert',
|
||||
'settings.composio.notYetRouted': 'noch nicht geroutet',
|
||||
|
||||
@@ -604,6 +604,9 @@ const en5: TranslationMap = {
|
||||
'skills.uninstall.confirmTitle': 'Uninstall {name}?',
|
||||
'conversations.taskKanban.blocked': 'Blocked',
|
||||
'conversations.taskKanban.done': 'Done',
|
||||
'conversations.taskKanban.awaitingApproval': 'Awaiting approval',
|
||||
'conversations.taskKanban.ready': 'Ready',
|
||||
'conversations.taskKanban.rejected': 'Rejected',
|
||||
'conversations.taskKanban.inProgress': 'In progress',
|
||||
'intelligence.memoryChunk.detail.copiedHint': 'copied',
|
||||
'settings.composio.notYetRouted': 'not yet routed',
|
||||
|
||||
@@ -564,6 +564,9 @@ const es5: TranslationMap = {
|
||||
'skills.uninstall.confirmTitle': '¿Desinstalar {name}?',
|
||||
'conversations.taskKanban.blocked': 'Bloqueado',
|
||||
'conversations.taskKanban.done': 'Completado',
|
||||
'conversations.taskKanban.awaitingApproval': 'Awaiting approval',
|
||||
'conversations.taskKanban.ready': 'Ready',
|
||||
'conversations.taskKanban.rejected': 'Rejected',
|
||||
'conversations.taskKanban.inProgress': 'En progreso',
|
||||
'intelligence.memoryChunk.detail.copiedHint': 'copiado',
|
||||
'settings.composio.notYetRouted': 'aún sin enrutar',
|
||||
|
||||
@@ -568,6 +568,9 @@ const fr5: TranslationMap = {
|
||||
'skills.uninstall.confirmTitle': 'Désinstaller {name} ?',
|
||||
'conversations.taskKanban.blocked': 'Bloqué',
|
||||
'conversations.taskKanban.done': 'Terminé',
|
||||
'conversations.taskKanban.awaitingApproval': 'Awaiting approval',
|
||||
'conversations.taskKanban.ready': 'Ready',
|
||||
'conversations.taskKanban.rejected': 'Rejected',
|
||||
'conversations.taskKanban.inProgress': 'En cours',
|
||||
'intelligence.memoryChunk.detail.copiedHint': 'copié',
|
||||
'settings.composio.notYetRouted': 'pas encore routé',
|
||||
|
||||
@@ -561,6 +561,9 @@ const hi5: TranslationMap = {
|
||||
'skills.uninstall.confirmTitle': '{name} अनइंस्टॉल करें?',
|
||||
'conversations.taskKanban.blocked': 'अवरुद्ध',
|
||||
'conversations.taskKanban.done': 'पूर्ण',
|
||||
'conversations.taskKanban.awaitingApproval': 'Awaiting approval',
|
||||
'conversations.taskKanban.ready': 'Ready',
|
||||
'conversations.taskKanban.rejected': 'Rejected',
|
||||
'conversations.taskKanban.inProgress': 'प्रगति पर',
|
||||
'intelligence.memoryChunk.detail.copiedHint': 'कॉपी हो गया',
|
||||
'settings.composio.notYetRouted': 'अभी तक रूट नहीं हुआ',
|
||||
|
||||
@@ -561,6 +561,9 @@ const id5: TranslationMap = {
|
||||
'skills.uninstall.confirmTitle': 'Copot {name}?',
|
||||
'conversations.taskKanban.blocked': 'Terhambat',
|
||||
'conversations.taskKanban.done': 'Selesai',
|
||||
'conversations.taskKanban.awaitingApproval': 'Awaiting approval',
|
||||
'conversations.taskKanban.ready': 'Ready',
|
||||
'conversations.taskKanban.rejected': 'Rejected',
|
||||
'conversations.taskKanban.inProgress': 'Sedang berjalan',
|
||||
'intelligence.memoryChunk.detail.copiedHint': 'disalin',
|
||||
'settings.composio.notYetRouted': 'belum dirutekan',
|
||||
|
||||
@@ -565,6 +565,9 @@ const it5: TranslationMap = {
|
||||
'skills.uninstall.confirmTitle': 'Disinstallare {name}?',
|
||||
'conversations.taskKanban.blocked': 'Bloccato',
|
||||
'conversations.taskKanban.done': 'Fatto',
|
||||
'conversations.taskKanban.awaitingApproval': 'Awaiting approval',
|
||||
'conversations.taskKanban.ready': 'Ready',
|
||||
'conversations.taskKanban.rejected': 'Rejected',
|
||||
'conversations.taskKanban.inProgress': 'In corso',
|
||||
'intelligence.memoryChunk.detail.copiedHint': 'copiato',
|
||||
'settings.composio.notYetRouted': 'non ancora instradato',
|
||||
|
||||
@@ -436,6 +436,9 @@ const ko5: TranslationMap = {
|
||||
'skills.uninstall.confirmTitle': '{name}을(를) 제거하시겠습니까?',
|
||||
'conversations.taskKanban.blocked': '차단됨',
|
||||
'conversations.taskKanban.done': '완료',
|
||||
'conversations.taskKanban.awaitingApproval': 'Awaiting approval',
|
||||
'conversations.taskKanban.ready': 'Ready',
|
||||
'conversations.taskKanban.rejected': 'Rejected',
|
||||
'conversations.taskKanban.inProgress': '진행 중',
|
||||
'intelligence.memoryChunk.detail.copiedHint': '복사됨',
|
||||
'settings.composio.notYetRouted': '아직 라우팅되지 않음',
|
||||
|
||||
@@ -613,6 +613,9 @@ const pl5: TranslationMap = {
|
||||
'skills.uninstall.confirmTitle': 'Odinstalować {name}?',
|
||||
'conversations.taskKanban.blocked': 'Zablokowane',
|
||||
'conversations.taskKanban.done': 'Zrobione',
|
||||
'conversations.taskKanban.awaitingApproval': 'Awaiting approval',
|
||||
'conversations.taskKanban.ready': 'Ready',
|
||||
'conversations.taskKanban.rejected': 'Rejected',
|
||||
'conversations.taskKanban.inProgress': 'W toku',
|
||||
'intelligence.memoryChunk.detail.copiedHint': 'skopiowano',
|
||||
'settings.composio.notYetRouted': 'jeszcze nie trasowane',
|
||||
|
||||
@@ -565,6 +565,9 @@ const pt5: TranslationMap = {
|
||||
'skills.uninstall.confirmTitle': 'Desinstalar {name}?',
|
||||
'conversations.taskKanban.blocked': 'Bloqueado',
|
||||
'conversations.taskKanban.done': 'Concluído',
|
||||
'conversations.taskKanban.awaitingApproval': 'Awaiting approval',
|
||||
'conversations.taskKanban.ready': 'Ready',
|
||||
'conversations.taskKanban.rejected': 'Rejected',
|
||||
'conversations.taskKanban.inProgress': 'Em andamento',
|
||||
'intelligence.memoryChunk.detail.copiedHint': 'copiado',
|
||||
'settings.composio.notYetRouted': 'ainda não roteado',
|
||||
|
||||
@@ -561,6 +561,9 @@ const ru5: TranslationMap = {
|
||||
'skills.uninstall.confirmTitle': 'Удалить {name}?',
|
||||
'conversations.taskKanban.blocked': 'Заблокировано',
|
||||
'conversations.taskKanban.done': 'Готово',
|
||||
'conversations.taskKanban.awaitingApproval': 'Awaiting approval',
|
||||
'conversations.taskKanban.ready': 'Ready',
|
||||
'conversations.taskKanban.rejected': 'Rejected',
|
||||
'conversations.taskKanban.inProgress': 'В работе',
|
||||
'intelligence.memoryChunk.detail.copiedHint': 'скопировано',
|
||||
'settings.composio.notYetRouted': 'пока не маршрутизируется',
|
||||
|
||||
@@ -533,6 +533,9 @@ const zhCN5: TranslationMap = {
|
||||
'skills.uninstall.confirmTitle': '卸载 {name}？',
|
||||
'conversations.taskKanban.blocked': '已阻塞',
|
||||
'conversations.taskKanban.done': '已完成',
|
||||
'conversations.taskKanban.awaitingApproval': 'Awaiting approval',
|
||||
'conversations.taskKanban.ready': 'Ready',
|
||||
'conversations.taskKanban.rejected': 'Rejected',
|
||||
'conversations.taskKanban.inProgress': '进行中',
|
||||
'intelligence.memoryChunk.detail.copiedHint': '已复制',
|
||||
'settings.composio.notYetRouted': '尚未路由',
|
||||
|
||||
@@ -4025,6 +4025,9 @@ const en: TranslationMap = {
|
||||
'skills.uninstall.confirmTitle': 'Uninstall {name}?',
|
||||
'conversations.taskKanban.blocked': 'Blocked',
|
||||
'conversations.taskKanban.done': 'Done',
|
||||
'conversations.taskKanban.awaitingApproval': 'Awaiting approval',
|
||||
'conversations.taskKanban.ready': 'Ready',
|
||||
'conversations.taskKanban.rejected': 'Rejected',
|
||||
'conversations.taskKanban.inProgress': 'In progress',
|
||||
'intelligence.memoryChunk.detail.copiedHint': 'copied',
|
||||
'settings.composio.notYetRouted': 'not yet routed',
|
||||
|
||||
@@ -85,6 +85,7 @@ import {
|
||||
getComposerBlockedSendFeedback,
|
||||
handleComposerSlashCommand,
|
||||
} from './conversations/composerSendDecision';
|
||||
import { runDecidePlan } from './conversations/taskPlanActions';
|
||||
import {
|
||||
type AgentBubblePosition,
|
||||
buildAcceptedInlineCompletion,
|
||||
@@ -1627,6 +1628,16 @@ const Conversations = ({
|
||||
onUpdateCard={(card, nextCard) => {
|
||||
void handleUpdateTaskCard(card, nextCard);
|
||||
}}
|
||||
onDecidePlan={(card, approve) => {
|
||||
void runDecidePlan({
|
||||
threadId: selectedThreadId,
|
||||
card,
|
||||
approve,
|
||||
dispatch,
|
||||
notify: setSendAdvisory,
|
||||
t,
|
||||
});
|
||||
}}
|
||||
/>
|
||||
)}
|
||||
{visibleMessages.map(msg => (
|
||||
|
||||
@@ -0,0 +1,78 @@
|
||||
import { fireEvent, render, screen } from '@testing-library/react';
|
||||
import { describe, expect, it, vi } from 'vitest';
|
||||
|
||||
import type { TaskBoard, TaskBoardCard } from '../../../types/turnState';
|
||||
import { TaskKanbanBoard } from './TaskKanbanBoard';
|
||||
|
||||
// Echo i18n keys so we can query by the stable key strings.
|
||||
vi.mock('../../../lib/i18n/I18nContext', () => ({ useT: () => ({ t: (key: string) => key }) }));
|
||||
|
||||
function card(partial: Partial<TaskBoardCard>): TaskBoardCard {
|
||||
return {
|
||||
id: 'c1',
|
||||
title: 'Do thing',
|
||||
status: 'todo',
|
||||
order: 0,
|
||||
updatedAt: '',
|
||||
...partial,
|
||||
} as TaskBoardCard;
|
||||
}
|
||||
|
||||
function board(cards: TaskBoardCard[]): TaskBoard {
|
||||
return { threadId: 't1', cards, updatedAt: '' };
|
||||
}
|
||||
|
||||
describe('TaskKanbanBoard approval surface', () => {
|
||||
it('renders Approve/Reject on an awaiting_approval card and calls onDecidePlan', () => {
|
||||
const onDecidePlan = vi.fn();
|
||||
render(
|
||||
<TaskKanbanBoard
|
||||
board={board([card({ id: 'a', status: 'awaiting_approval', title: 'Needs approval' })])}
|
||||
onDecidePlan={onDecidePlan}
|
||||
/>
|
||||
);
|
||||
|
||||
fireEvent.click(screen.getByTitle('chat.approval.approve'));
|
||||
expect(onDecidePlan).toHaveBeenCalledWith(expect.objectContaining({ id: 'a' }), true);
|
||||
|
||||
fireEvent.click(screen.getByTitle('chat.approval.deny'));
|
||||
expect(onDecidePlan).toHaveBeenCalledWith(expect.objectContaining({ id: 'a' }), false);
|
||||
});
|
||||
|
||||
it('buckets ready→todo and rejected→blocked columns so the cards still render', () => {
|
||||
render(
|
||||
<TaskKanbanBoard
|
||||
board={board([
|
||||
card({ id: 'r', status: 'ready', title: 'Ready card' }),
|
||||
card({ id: 'x', status: 'rejected', title: 'Rejected card' }),
|
||||
])}
|
||||
/>
|
||||
);
|
||||
|
||||
expect(screen.getByText('Ready card')).toBeInTheDocument();
|
||||
expect(screen.getByText('Rejected card')).toBeInTheDocument();
|
||||
// An approval-flow card without onDecidePlan shows no approve/reject controls.
|
||||
expect(screen.queryByTitle('chat.approval.approve')).toBeNull();
|
||||
});
|
||||
|
||||
it('edit dialog status select has a matching option for approval-flow statuses', () => {
|
||||
// Regression: the dialog <select> must carry an <option> for every status,
|
||||
// not just the four column statuses — otherwise an awaiting_approval card
|
||||
// renders a controlled select with no matching option (React warns and the
|
||||
// value silently shows as the first option, hiding the real status).
|
||||
render(
|
||||
<TaskKanbanBoard
|
||||
board={board([card({ id: 'a', status: 'awaiting_approval', title: 'Needs approval' })])}
|
||||
onUpdateCard={vi.fn()}
|
||||
/>
|
||||
);
|
||||
|
||||
fireEvent.click(screen.getByText('conversations.taskKanban.briefButton'));
|
||||
|
||||
// The status select shows the awaiting_approval label as its selected
|
||||
// value, proving a matching option exists (no fallback to 'todo').
|
||||
expect(
|
||||
screen.getByDisplayValue('conversations.taskKanban.awaitingApproval')
|
||||
).toBeInTheDocument();
|
||||
});
|
||||
});
|
||||
@@ -24,11 +24,52 @@ const COLUMN_DEFS: ColumnDef[] = [
|
||||
|
||||
const STATUS_INDEX = new Map(COLUMN_DEFS.map((column, index) => [column.status, index]));
|
||||
|
||||
/** Label key for *every* status, including the approval-flow statuses that
|
||||
* don't own a kanban column. Drives the edit dialog's status `<select>` so a
|
||||
* card whose status is `awaiting_approval`/`ready`/`rejected` renders a
|
||||
* matching option instead of a controlled-select value with no option (which
|
||||
* React warns about and which renders as the first option, hiding the real
|
||||
* status from the user). */
|
||||
const STATUS_LABEL_KEYS: Record<TaskBoardCardStatus, string> = {
|
||||
todo: 'conversations.taskKanban.todo',
|
||||
awaiting_approval: 'conversations.taskKanban.awaitingApproval',
|
||||
ready: 'conversations.taskKanban.ready',
|
||||
in_progress: 'conversations.taskKanban.inProgress',
|
||||
blocked: 'conversations.taskKanban.blocked',
|
||||
done: 'conversations.taskKanban.done',
|
||||
rejected: 'conversations.taskKanban.rejected',
|
||||
};
|
||||
|
||||
const ALL_STATUSES = Object.keys(STATUS_LABEL_KEYS) as TaskBoardCardStatus[];
|
||||
|
||||
/** Whether a status owns a kanban column (vs the approval-flow statuses that
|
||||
* are bucketed into an existing column). */
|
||||
function isColumnStatus(status: TaskBoardCardStatus): boolean {
|
||||
return STATUS_INDEX.has(status);
|
||||
}
|
||||
|
||||
/** Map a card status to the column it renders under. The approval-flow
|
||||
* statuses don't get their own columns: pre-execution ones sit in `todo`,
|
||||
* `rejected` sits with `blocked`. */
|
||||
function columnFor(status: TaskBoardCardStatus): TaskBoardCardStatus {
|
||||
switch (status) {
|
||||
case 'awaiting_approval':
|
||||
case 'ready':
|
||||
return 'todo';
|
||||
case 'rejected':
|
||||
return 'blocked';
|
||||
default:
|
||||
return status;
|
||||
}
|
||||
}
|
||||
|
||||
interface TaskKanbanBoardProps {
|
||||
board: TaskBoard;
|
||||
disabled?: boolean;
|
||||
onMove?: (card: TaskBoardCard, status: TaskBoardCardStatus) => void;
|
||||
onUpdateCard?: (card: TaskBoardCard, nextCard: TaskBoardCard) => void;
|
||||
/** Approve/reject a card awaiting plan approval. */
|
||||
onDecidePlan?: (card: TaskBoardCard, approve: boolean) => void;
|
||||
}
|
||||
|
||||
export function TaskKanbanBoard({
|
||||
@@ -36,6 +77,7 @@ export function TaskKanbanBoard({
|
||||
disabled = false,
|
||||
onMove,
|
||||
onUpdateCard,
|
||||
onDecidePlan,
|
||||
}: TaskKanbanBoardProps) {
|
||||
const { t } = useT();
|
||||
const [selectedCardId, setSelectedCardId] = useState<string | null>(null);
|
||||
@@ -55,7 +97,7 @@ export function TaskKanbanBoard({
|
||||
);
|
||||
|
||||
for (const card of [...board.cards].sort((a, b) => a.order - b.order)) {
|
||||
cardsByStatus[card.status]?.push(card);
|
||||
cardsByStatus[columnFor(card.status)]?.push(card);
|
||||
}
|
||||
|
||||
const moveCard = (card: TaskBoardCard, direction: -1 | 1) => {
|
||||
@@ -97,7 +139,26 @@ export function TaskKanbanBoard({
|
||||
<p className="min-w-0 flex-1 break-words text-xs font-medium leading-snug text-stone-800 dark:text-neutral-100">
|
||||
{card.title}
|
||||
</p>
|
||||
{onMove && (
|
||||
{card.status === 'awaiting_approval' && onDecidePlan ? (
|
||||
<div className="flex flex-shrink-0 items-center gap-1">
|
||||
<button
|
||||
type="button"
|
||||
title={t('chat.approval.approve')}
|
||||
disabled={disabled}
|
||||
onClick={() => onDecidePlan(card, true)}
|
||||
className="rounded-md bg-ocean-600 px-1.5 py-0.5 text-[10px] font-medium text-white transition-colors hover:bg-ocean-700 disabled:opacity-40">
|
||||
{t('chat.approval.approve')}
|
||||
</button>
|
||||
<button
|
||||
type="button"
|
||||
title={t('chat.approval.deny')}
|
||||
disabled={disabled}
|
||||
onClick={() => onDecidePlan(card, false)}
|
||||
className="rounded-md border border-stone-200 px-1.5 py-0.5 text-[10px] font-medium text-stone-600 transition-colors hover:bg-stone-100 disabled:opacity-40 dark:border-neutral-700 dark:text-neutral-300 dark:hover:bg-neutral-800">
|
||||
{t('chat.approval.deny')}
|
||||
</button>
|
||||
</div>
|
||||
) : onMove && isColumnStatus(card.status) ? (
|
||||
<div className="flex flex-shrink-0 items-center gap-0.5">
|
||||
<button
|
||||
type="button"
|
||||
@@ -118,7 +179,7 @@ export function TaskKanbanBoard({
|
||||
<LuArrowRight className="h-3 w-3" />
|
||||
</button>
|
||||
</div>
|
||||
)}
|
||||
) : null}
|
||||
</div>
|
||||
<div className="mt-2 flex flex-wrap gap-1.5">
|
||||
{card.assignedAgent && (
|
||||
@@ -286,9 +347,9 @@ function TaskBriefDialog({
|
||||
value={status}
|
||||
onChange={e => setStatus(e.target.value as TaskBoardCardStatus)}
|
||||
className="w-full rounded-md border border-stone-200 bg-white px-2 py-1.5 text-sm text-stone-900 dark:border-neutral-700 dark:bg-neutral-950 dark:text-neutral-50">
|
||||
{COLUMN_DEFS.map(column => (
|
||||
<option key={column.status} value={column.status}>
|
||||
{t(column.labelKey)}
|
||||
{ALL_STATUSES.map(s => (
|
||||
<option key={s} value={s}>
|
||||
{t(STATUS_LABEL_KEYS[s])}
|
||||
</option>
|
||||
))}
|
||||
</select>
|
||||
|
||||
@@ -0,0 +1,65 @@
|
||||
import { beforeEach, describe, expect, it, vi } from 'vitest';
|
||||
|
||||
import type { TaskBoardCard } from '../../types/turnState';
|
||||
import { runDecidePlan } from './taskPlanActions';
|
||||
|
||||
const mockDecidePlan = vi.fn();
|
||||
vi.mock('../../services/api/threadApi', () => ({
|
||||
threadApi: { decidePlan: (...args: unknown[]) => mockDecidePlan(...args) },
|
||||
}));
|
||||
|
||||
function card(): TaskBoardCard {
|
||||
return { id: 'c1', title: 'T', status: 'awaiting_approval', order: 0, updatedAt: '' };
|
||||
}
|
||||
|
||||
const t = (key: string) => key;
|
||||
|
||||
describe('runDecidePlan', () => {
|
||||
beforeEach(() => mockDecidePlan.mockReset());
|
||||
|
||||
it('is a no-op when there is no active thread', async () => {
|
||||
const dispatch = vi.fn();
|
||||
const notify = vi.fn();
|
||||
|
||||
await runDecidePlan({ threadId: null, card: card(), approve: true, dispatch, notify, t });
|
||||
|
||||
expect(mockDecidePlan).not.toHaveBeenCalled();
|
||||
expect(dispatch).not.toHaveBeenCalled();
|
||||
expect(notify).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('dispatches the refreshed board on success', async () => {
|
||||
mockDecidePlan.mockResolvedValueOnce({ threadId: 't1', cards: [], updatedAt: '' });
|
||||
const dispatch = vi.fn();
|
||||
const notify = vi.fn();
|
||||
|
||||
await runDecidePlan({ threadId: 't1', card: card(), approve: true, dispatch, notify, t });
|
||||
|
||||
expect(mockDecidePlan).toHaveBeenCalledWith('t1', 'c1', true);
|
||||
expect(dispatch).toHaveBeenCalledTimes(1);
|
||||
expect(notify).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('does not dispatch when the RPC returns null', async () => {
|
||||
mockDecidePlan.mockResolvedValueOnce(null);
|
||||
const dispatch = vi.fn();
|
||||
const notify = vi.fn();
|
||||
|
||||
await runDecidePlan({ threadId: 't1', card: card(), approve: false, dispatch, notify, t });
|
||||
|
||||
expect(mockDecidePlan).toHaveBeenCalledWith('t1', 'c1', false);
|
||||
expect(dispatch).not.toHaveBeenCalled();
|
||||
expect(notify).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('notifies (without throwing or dispatching) on RPC failure', async () => {
|
||||
mockDecidePlan.mockRejectedValueOnce(new Error('boom'));
|
||||
const dispatch = vi.fn();
|
||||
const notify = vi.fn();
|
||||
|
||||
await runDecidePlan({ threadId: 't1', card: card(), approve: true, dispatch, notify, t });
|
||||
|
||||
expect(notify).toHaveBeenCalledWith('conversations.taskKanban.updateFailed');
|
||||
expect(dispatch).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,51 @@
|
||||
/**
|
||||
* Task-plan approval action, extracted from `Conversations` so the whole
|
||||
* guard → decide → persist → refresh → error-advisory flow is unit-testable
|
||||
* without mounting the conversations page.
|
||||
*/
|
||||
import debugFactory from 'debug';
|
||||
|
||||
import { threadApi } from '../../services/api/threadApi';
|
||||
import { setTaskBoardForThread } from '../../store/chatRuntimeSlice';
|
||||
import type { TaskBoardCard } from '../../types/turnState';
|
||||
|
||||
const debug = debugFactory('conversations:taskPlan');
|
||||
|
||||
export interface RunDecidePlanArgs {
|
||||
/** Active thread; a no-op when null (nothing to decide against). */
|
||||
threadId: string | null;
|
||||
card: TaskBoardCard;
|
||||
approve: boolean;
|
||||
/** Redux dispatch (kept loosely typed so callers don't need the store type). */
|
||||
dispatch: (action: unknown) => void;
|
||||
/** Surface a user-facing advisory on failure. */
|
||||
notify: (message: string) => void;
|
||||
/** Translator for the advisory message. */
|
||||
t: (key: string) => string;
|
||||
}
|
||||
|
||||
/**
|
||||
* Approve or reject a card's plan via the core RPC, then optimistically refresh
|
||||
* the thread's board from the returned snapshot. A failure is logged and
|
||||
* surfaced via `notify` (never thrown) so a failed decision degrades
|
||||
* gracefully. No-op when `threadId` is null.
|
||||
*/
|
||||
export async function runDecidePlan({
|
||||
threadId,
|
||||
card,
|
||||
approve,
|
||||
dispatch,
|
||||
notify,
|
||||
t,
|
||||
}: RunDecidePlanArgs): Promise<void> {
|
||||
if (!threadId) return;
|
||||
try {
|
||||
const saved = await threadApi.decidePlan(threadId, card.id, approve);
|
||||
if (saved) {
|
||||
dispatch(setTaskBoardForThread({ threadId, board: saved }));
|
||||
}
|
||||
} catch (error) {
|
||||
debug('decidePlan failed: %o', error);
|
||||
notify(t('conversations.taskKanban.updateFailed'));
|
||||
}
|
||||
}
|
||||
@@ -140,4 +140,33 @@ describe('threadApi', () => {
|
||||
});
|
||||
expect(result).toEqual(thread);
|
||||
});
|
||||
|
||||
it('approves a plan via the todos_decide_plan RPC and rebuilds the board', async () => {
|
||||
mockCallCoreRpc.mockResolvedValueOnce({
|
||||
data: { threadId: 'thread-1', cards: [{ id: 'card-1', title: 'T', status: 'ready' }] },
|
||||
});
|
||||
|
||||
const { threadApi } = await import('./threadApi');
|
||||
const board = await threadApi.decidePlan('thread-1', 'card-1', true);
|
||||
|
||||
expect(mockCallCoreRpc).toHaveBeenCalledWith({
|
||||
method: 'openhuman.todos_decide_plan',
|
||||
params: { thread_id: 'thread-1', id: 'card-1', approve: true },
|
||||
});
|
||||
expect(board?.threadId).toBe('thread-1');
|
||||
expect(board?.cards[0].status).toBe('ready');
|
||||
});
|
||||
|
||||
it('returns null from decidePlan when the snapshot has no cards', async () => {
|
||||
mockCallCoreRpc.mockResolvedValueOnce({ data: {} });
|
||||
|
||||
const { threadApi } = await import('./threadApi');
|
||||
const board = await threadApi.decidePlan('thread-1', 'card-1', false);
|
||||
|
||||
expect(mockCallCoreRpc).toHaveBeenCalledWith({
|
||||
method: 'openhuman.todos_decide_plan',
|
||||
params: { thread_id: 'thread-1', id: 'card-1', approve: false },
|
||||
});
|
||||
expect(board).toBeNull();
|
||||
});
|
||||
});
|
||||
|
||||
@@ -161,6 +161,32 @@ export const threadApi = {
|
||||
return data?.taskBoard ?? null;
|
||||
},
|
||||
|
||||
/**
|
||||
* Approve or reject a task-board card that is awaiting plan approval
|
||||
* (`openhuman.todos_decide_plan`). Approve → the card becomes runnable
|
||||
* (`ready`); reject → `rejected`. Returns the updated board (rebuilt from
|
||||
* the returned todos snapshot) or null.
|
||||
*/
|
||||
decidePlan: async (
|
||||
threadId: string,
|
||||
cardId: string,
|
||||
approve: boolean
|
||||
): Promise<TaskBoard | null> => {
|
||||
const response = await callCoreRpc<{
|
||||
data?: { threadId?: string | null; cards?: TaskBoardCard[] };
|
||||
}>({
|
||||
method: 'openhuman.todos_decide_plan',
|
||||
params: { thread_id: threadId, id: cardId, approve },
|
||||
});
|
||||
const data = unwrapEnvelope(response);
|
||||
if (!data?.cards) return null;
|
||||
return {
|
||||
threadId: data.threadId ?? threadId,
|
||||
cards: data.cards,
|
||||
updatedAt: new Date().toISOString(),
|
||||
};
|
||||
},
|
||||
|
||||
updateLabels: async (threadId: string, labels: string[]): Promise<Thread> => {
|
||||
const response = await callCoreRpc<Envelope<Thread>>({
|
||||
method: 'openhuman.threads_update_labels',
|
||||
|
||||
@@ -11,7 +11,14 @@ export type PersistedTurnPhase = 'thinking' | 'tool_use' | 'subagent';
|
||||
|
||||
export type PersistedToolStatus = 'running' | 'success' | 'error';
|
||||
|
||||
export type TaskBoardCardStatus = 'todo' | 'in_progress' | 'blocked' | 'done';
|
||||
export type TaskBoardCardStatus =
|
||||
| 'todo'
|
||||
| 'awaiting_approval'
|
||||
| 'ready'
|
||||
| 'in_progress'
|
||||
| 'blocked'
|
||||
| 'done'
|
||||
| 'rejected';
|
||||
export type TaskApprovalMode = 'required' | 'not_required';
|
||||
|
||||
export interface TaskBoardCard {
|
||||
@@ -27,6 +34,10 @@ export interface TaskBoardCard {
|
||||
evidence?: string[];
|
||||
notes?: string | null;
|
||||
blocker?: string | null;
|
||||
/** Provider/source identifiers for a card ingested from a task source
|
||||
* (`{provider, source_id, external_id, url, repo?, urgency}`); absent on
|
||||
* agent/UI-authored cards. */
|
||||
sourceMetadata?: Record<string, unknown> | null;
|
||||
order: number;
|
||||
updatedAt: string;
|
||||
}
|
||||
|
||||
@@ -56,6 +56,9 @@ export interface TaskSource {
|
||||
intervalSecs: number;
|
||||
target: TaskSourceTarget;
|
||||
maxTasksPerFetch: number;
|
||||
/** Static executor routing (G7): personality/skill/agent handle every card
|
||||
* from this source is pre-assigned to. */
|
||||
assignedExecutor?: string;
|
||||
createdAt: string;
|
||||
lastFetchAt?: string;
|
||||
lastStatus?: string;
|
||||
@@ -100,6 +103,8 @@ export interface TaskSourcePatch {
|
||||
target?: TaskSourceTarget;
|
||||
maxTasksPerFetch?: number;
|
||||
connectionId?: string;
|
||||
/** Executor routing (G7): personality/skill/agent handle to pre-assign. */
|
||||
assignedExecutor?: string;
|
||||
}
|
||||
|
||||
export interface TaskSourceAddParams {
|
||||
@@ -110,6 +115,7 @@ export interface TaskSourceAddParams {
|
||||
interval_secs?: number;
|
||||
target?: TaskSourceTarget;
|
||||
max_tasks_per_fetch?: number;
|
||||
assigned_executor?: string;
|
||||
}
|
||||
|
||||
function ensureTauri(): void {
|
||||
|
||||
@@ -656,6 +656,18 @@ pub enum DomainEvent {
|
||||
provider: String,
|
||||
error: String,
|
||||
},
|
||||
/// A task-board card needs human plan approval before the dispatcher will
|
||||
/// execute it (emitted when `autonomy.require_task_plan_approval` is on and
|
||||
/// the dispatcher parks a `todo` card at `awaiting_approval`).
|
||||
///
|
||||
/// Surfacing: the parked card is persisted with status `awaiting_approval`,
|
||||
/// so the kanban board renders it with inline Approve/Reject on the next
|
||||
/// board fetch/refresh — that is the current (poll-based) surface and the
|
||||
/// reason this telemetry event has no dedicated subscriber yet. A realtime
|
||||
/// socket bridge (à la `ApprovalRequested` → `approval_request`) is a
|
||||
/// deliberate follow-up; emitting the event now lets that bridge attach
|
||||
/// without a schema change.
|
||||
TaskPlanAwaitingApproval { card_id: String, thread_id: String },
|
||||
}
|
||||
|
||||
impl DomainEvent {
|
||||
@@ -747,6 +759,8 @@ impl DomainEvent {
|
||||
| Self::TaskSourceTaskIngested { .. }
|
||||
| Self::TaskSourceFetchFailed { .. } => "task_sources",
|
||||
|
||||
Self::TaskPlanAwaitingApproval { .. } => "agent",
|
||||
|
||||
Self::ApprovalRequested { .. } | Self::ApprovalDecided { .. } => "approval",
|
||||
|
||||
Self::McpServerInstalled { .. }
|
||||
@@ -837,6 +851,7 @@ impl DomainEvent {
|
||||
Self::TaskSourceFetched { .. } => "TaskSourceFetched",
|
||||
Self::TaskSourceTaskIngested { .. } => "TaskSourceTaskIngested",
|
||||
Self::TaskSourceFetchFailed { .. } => "TaskSourceFetchFailed",
|
||||
Self::TaskPlanAwaitingApproval { .. } => "TaskPlanAwaitingApproval",
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -1846,6 +1846,9 @@ fn register_domain_subscribers(
|
||||
// Task-sources proactive ingestion: connection-created hook + poll.
|
||||
crate::openhuman::task_sources::bus::register_task_sources_subscriber();
|
||||
crate::openhuman::task_sources::start_periodic_poll();
|
||||
// Board poller: dispatch the highest-urgency `todo` card on the
|
||||
// task-sources board (catch-all for cards without a proactive trigger).
|
||||
crate::openhuman::agent::task_dispatcher::start_board_poller();
|
||||
// Seed memory_sources with active Composio connections so the
|
||||
// user sees their connected integrations as memory sources by
|
||||
// default. Best-effort: failure is logged but does not block startup.
|
||||
|
||||
@@ -42,6 +42,7 @@ pub mod prompts;
|
||||
mod schemas;
|
||||
pub mod stop_hooks;
|
||||
pub mod task_board;
|
||||
pub mod task_dispatcher;
|
||||
pub mod tool_policy;
|
||||
pub mod tools;
|
||||
pub mod tree_loader;
|
||||
|
||||
@@ -18,18 +18,31 @@ const TASK_BOARD_EXTENSION: &str = "json";
|
||||
#[serde(rename_all = "snake_case")]
|
||||
pub enum TaskCardStatus {
|
||||
Todo,
|
||||
/// Plan approval required and pending — the dispatcher parked the card here
|
||||
/// and emitted `TaskPlanAwaitingApproval`; it will not run until a human
|
||||
/// approves (→ `Ready`) or rejects (→ `Rejected`).
|
||||
AwaitingApproval,
|
||||
/// Approved for execution — the dispatcher runs `Ready` cards without a
|
||||
/// further approval check (distinguishes "approved" from the initial
|
||||
/// `Todo`, which the approval gate would otherwise re-park).
|
||||
Ready,
|
||||
InProgress,
|
||||
Blocked,
|
||||
Done,
|
||||
/// Plan approval was denied; the card is not executed.
|
||||
Rejected,
|
||||
}
|
||||
|
||||
impl TaskCardStatus {
|
||||
pub fn as_str(&self) -> &'static str {
|
||||
match self {
|
||||
Self::Todo => "todo",
|
||||
Self::AwaitingApproval => "awaiting_approval",
|
||||
Self::Ready => "ready",
|
||||
Self::InProgress => "in_progress",
|
||||
Self::Blocked => "blocked",
|
||||
Self::Done => "done",
|
||||
Self::Rejected => "rejected",
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -74,6 +87,12 @@ pub struct TaskBoardCard {
|
||||
pub notes: Option<String>,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub blocker: Option<String>,
|
||||
/// Provider/source identifiers for a card ingested from a task source
|
||||
/// (`{provider, source_id, external_id, url, repo?, urgency}`). Set by
|
||||
/// the `task_sources` route; consumed downstream for prioritisation and
|
||||
/// external write-back. `None` for agent/UI-authored cards.
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub source_metadata: Option<serde_json::Value>,
|
||||
#[serde(default)]
|
||||
pub order: u32,
|
||||
#[serde(default)]
|
||||
@@ -420,6 +439,7 @@ mod tests {
|
||||
evidence: vec![" cargo test ".into()],
|
||||
notes: Some(" note ".into()),
|
||||
blocker: None,
|
||||
source_metadata: None,
|
||||
order: 99,
|
||||
updated_at: String::new(),
|
||||
},
|
||||
@@ -436,6 +456,7 @@ mod tests {
|
||||
evidence: Vec::new(),
|
||||
notes: Some("waiting on user".into()),
|
||||
blocker: None,
|
||||
source_metadata: None,
|
||||
order: 99,
|
||||
updated_at: String::new(),
|
||||
},
|
||||
@@ -506,6 +527,7 @@ mod tests {
|
||||
evidence: Vec::new(),
|
||||
notes: None,
|
||||
blocker: None,
|
||||
source_metadata: None,
|
||||
order: 99,
|
||||
updated_at: String::new(),
|
||||
},
|
||||
@@ -522,6 +544,7 @@ mod tests {
|
||||
evidence: Vec::new(),
|
||||
notes: None,
|
||||
blocker: None,
|
||||
source_metadata: None,
|
||||
order: 99,
|
||||
updated_at: String::new(),
|
||||
},
|
||||
|
||||
@@ -0,0 +1,814 @@
|
||||
//! Deterministic task-card dispatcher.
|
||||
//!
|
||||
//! Turns a [`TaskBoardCard`] into work: it **claims** the card via a
|
||||
//! compare-and-set (re-load the board and transition only a `Todo`/`Ready`
|
||||
//! card to `in_progress`, so a stale/concurrent re-dispatch of the same card
|
||||
//! is rejected), runs a single **autonomous agent turn** toward the card's
|
||||
//! objective, and **writes the outcome back** to the board (`done` + evidence
|
||||
//! on success, `blocked` + reason on failure).
|
||||
//!
|
||||
//! This is the one executor both dispatch paths converge on:
|
||||
//! - the **board poller** (cards that arrived without a proactive trigger), and
|
||||
//! - the **proactive triage** arm (`agent::triage::apply_decision`), once it has
|
||||
//! decided to act on a task-board card.
|
||||
//!
|
||||
//! The runner mirrors `skills::spawn_skill_run_background`: build the
|
||||
//! `orchestrator` agent fresh inside a detached task, cap tool iterations, and
|
||||
//! run `agent.run_single` under `with_autonomous_iter_cap`. PR-4 generalises the
|
||||
//! executor from the default agent to a resolved personality/skill; this module
|
||||
//! keeps the default-agent path so the pipeline runs end-to-end first.
|
||||
|
||||
use std::path::Path;
|
||||
use std::sync::OnceLock;
|
||||
use std::time::Duration;
|
||||
|
||||
use crate::openhuman::agent::harness::definition::{AgentDefinitionRegistry, PromptSource};
|
||||
use crate::openhuman::agent::harness::session::Agent;
|
||||
use crate::openhuman::agent::harness::subagent_runner::with_autonomous_iter_cap;
|
||||
use crate::openhuman::agent::personality_paths::PersonalityContext;
|
||||
use crate::openhuman::agent::task_board::{TaskBoardCard, TaskCardStatus};
|
||||
use crate::openhuman::config::Config;
|
||||
use crate::openhuman::todos::ops::{self, BoardLocation, CardPatch};
|
||||
|
||||
/// Max chars of a personality SOUL.md / MEMORY.md or skill guideline block
|
||||
/// folded into the agent's system-prompt suffix.
|
||||
const EXECUTOR_PREAMBLE_MAX_CHARS: usize = 800;
|
||||
|
||||
/// Tool-iteration ceiling for an autonomous task run. Matches the skill-run
|
||||
/// cap — a task brief is the same shape of bounded autonomous work.
|
||||
const TASK_RUN_MAX_ITERATIONS: usize = 200;
|
||||
|
||||
/// Max chars of the agent's final output retained as board `evidence`.
|
||||
const EVIDENCE_MAX_CHARS: usize = 2_000;
|
||||
|
||||
/// Render a card into the goal prompt handed to the autonomous run.
|
||||
///
|
||||
/// The card's `content`/title is the display form; the prompt leads with the
|
||||
/// clean `objective`, then any `plan` steps and `acceptance_criteria`, and a
|
||||
/// pointer to the originating source so the agent can pull related context from
|
||||
/// memory via its `memory_recall` tool (the GitHub/Notion/… activity for this
|
||||
/// item is ingested into the summary tree by the memory-sources domain).
|
||||
pub fn build_task_prompt(card: &TaskBoardCard) -> String {
|
||||
let mut lines: Vec<String> = Vec::new();
|
||||
|
||||
let objective = card
|
||||
.objective
|
||||
.as_deref()
|
||||
.map(str::trim)
|
||||
.filter(|s| !s.is_empty())
|
||||
.unwrap_or_else(|| card.title.trim());
|
||||
lines.push(format!(
|
||||
"You are autonomously executing one task to completion. Objective:\n{objective}"
|
||||
));
|
||||
|
||||
if !card.plan.is_empty() {
|
||||
lines.push("\nPlan:".to_string());
|
||||
for (i, step) in card.plan.iter().enumerate() {
|
||||
lines.push(format!("{}. {}", i + 1, step.trim()));
|
||||
}
|
||||
}
|
||||
|
||||
if !card.acceptance_criteria.is_empty() {
|
||||
lines.push("\nAcceptance criteria (the task is done only when all hold):".to_string());
|
||||
for c in &card.acceptance_criteria {
|
||||
lines.push(format!("- {}", c.trim()));
|
||||
}
|
||||
}
|
||||
|
||||
if let Some(meta) = &card.source_metadata {
|
||||
let provider = meta.get("provider").and_then(|v| v.as_str());
|
||||
let repo = meta.get("repo").and_then(|v| v.as_str());
|
||||
let external_id = meta.get("external_id").and_then(|v| v.as_str());
|
||||
let url = meta.get("url").and_then(|v| v.as_str());
|
||||
let mut origin = String::new();
|
||||
if let Some(p) = provider {
|
||||
origin.push_str(p);
|
||||
}
|
||||
if let Some(r) = repo {
|
||||
origin.push_str(&format!(" {r}"));
|
||||
}
|
||||
if let Some(id) = external_id {
|
||||
origin.push_str(&format!("#{id}"));
|
||||
}
|
||||
// Gate on a known provider so the origin string is always meaningful
|
||||
// (an id-only card would render "#123" with a leading space).
|
||||
if provider.is_some() {
|
||||
lines.push(format!(
|
||||
"\nThis task originates from {}. Its activity has been ingested into memory — use \
|
||||
your memory_recall tool to pull related context (prior discussion, linked items) \
|
||||
before and while you work.",
|
||||
origin.trim()
|
||||
));
|
||||
}
|
||||
if let Some(u) = url {
|
||||
lines.push(format!("Source link: {u}"));
|
||||
}
|
||||
// G9b — agent-driven external write-back. When the upstream item is
|
||||
// addressable (provider + id), instruct the agent to close the loop on
|
||||
// the source itself via its integration tools. Runs under the
|
||||
// connection's existing write scope (no extra approval gate); if it
|
||||
// can't, it reports that instead of failing.
|
||||
if provider.is_some() && external_id.is_some() {
|
||||
lines.push(format!(
|
||||
"\nWhen the task is complete, record the outcome on the upstream source ({}): use \
|
||||
your integration tools to add a comment summarising the resolution and, if the \
|
||||
work fully addresses it, close/resolve the item. If you lack the permission or \
|
||||
connection to do so, say so in your final summary instead of guessing.",
|
||||
origin.trim()
|
||||
));
|
||||
}
|
||||
}
|
||||
|
||||
lines.push(
|
||||
"\nWork the task to completion. Do not pick up unrelated work. When finished, your final \
|
||||
message should summarise what you did and the evidence (commits, PRs, results)."
|
||||
.to_string(),
|
||||
);
|
||||
|
||||
lines.join("\n")
|
||||
}
|
||||
|
||||
/// Outcome of a dispatch attempt.
|
||||
#[derive(Debug)]
|
||||
pub enum DispatchOutcome {
|
||||
/// The card was claimed and a detached autonomous run was spawned.
|
||||
Running { run_id: String },
|
||||
/// Plan approval is required; the card was parked at `awaiting_approval`
|
||||
/// and a `TaskPlanAwaitingApproval` event was emitted. No run was spawned.
|
||||
AwaitingApproval,
|
||||
}
|
||||
|
||||
/// Dispatch one card: gate on plan approval, claim it, run an autonomous turn,
|
||||
/// write the result back.
|
||||
///
|
||||
/// Returns `Ok(Running)` once the card is claimed and the detached run is
|
||||
/// spawned, `Ok(AwaitingApproval)` if the card was parked for human approval,
|
||||
/// or `Err` *without* spawning when the card is no longer claimable — its
|
||||
/// freshly-loaded status isn't `Todo`/`Ready` (already running/done, or another
|
||||
/// dispatcher won the claim). Benign: the poller retries next tick.
|
||||
pub async fn dispatch_card(
|
||||
location: BoardLocation,
|
||||
card: TaskBoardCard,
|
||||
) -> Result<DispatchOutcome, String> {
|
||||
let card_id = card.id.clone();
|
||||
|
||||
let config = Config::load_or_init()
|
||||
.await
|
||||
.map_err(|e| format!("load config: {e:#}"))?;
|
||||
|
||||
// Claim CAS: re-load the card's CURRENT status before acting. The passed-in
|
||||
// `card` may be stale — the poller and the proactive triage arm can target
|
||||
// the same card, and a re-triggered card may already be running or done.
|
||||
// Only `Todo`/`Ready` are claimable. This is what actually dedupes a
|
||||
// double-dispatch: `enforce_single_in_progress` alone does NOT, because
|
||||
// re-flipping an already-`InProgress` card to `InProgress` keeps the count
|
||||
// at one (→ `Ok`). Thread boards aren't lock-serialised, so a narrow TOCTOU
|
||||
// window between this read and the write remains; a fully race-free claim
|
||||
// needs a per-thread-locked compare-and-set `ops` primitive (follow-up).
|
||||
// Re-load the full current card (not just its status): the passed-in `card`
|
||||
// can be stale (edited between selection and claim), so the run must use the
|
||||
// fresh objective / plan / assigned_agent, not the snapshot the caller held.
|
||||
let fresh_card = ops::list(&location)
|
||||
.map_err(|e| format!("[task_dispatcher] reload before claim failed for {card_id}: {e}"))?
|
||||
.cards
|
||||
.into_iter()
|
||||
.find(|c| c.id == card_id)
|
||||
.ok_or_else(|| format!("[task_dispatcher] card {card_id} not found on board; skipping"))?;
|
||||
if !matches!(
|
||||
fresh_card.status,
|
||||
TaskCardStatus::Todo | TaskCardStatus::Ready
|
||||
) {
|
||||
return Err(format!(
|
||||
"[task_dispatcher] card {card_id} not claimable (status: {}); skipping",
|
||||
fresh_card.status.as_str()
|
||||
));
|
||||
}
|
||||
|
||||
// Plan-approval gate: when required, a `todo` card is parked for human
|
||||
// approval before it can run. `Ready` (already approved) bypasses.
|
||||
if config.autonomy.require_task_plan_approval && fresh_card.status == TaskCardStatus::Todo {
|
||||
ops::update_status(&location, &card_id, TaskCardStatus::AwaitingApproval).map_err(|e| {
|
||||
format!("[task_dispatcher] park-for-approval failed for {card_id}: {e}")
|
||||
})?;
|
||||
if let Some(thread_id) = location.thread_id() {
|
||||
crate::core::event_bus::publish_global(
|
||||
crate::core::event_bus::DomainEvent::TaskPlanAwaitingApproval {
|
||||
card_id: card_id.clone(),
|
||||
thread_id: thread_id.to_string(),
|
||||
},
|
||||
);
|
||||
}
|
||||
tracing::info!(card_id = %card_id, "[task_dispatcher] parked card awaiting plan approval");
|
||||
return Ok(DispatchOutcome::AwaitingApproval);
|
||||
}
|
||||
|
||||
let prompt = build_task_prompt(&fresh_card);
|
||||
|
||||
// Claim: Todo|Ready→InProgress.
|
||||
ops::update_status(&location, &card_id, TaskCardStatus::InProgress)
|
||||
.map_err(|e| format!("[task_dispatcher] claim failed for card {card_id}: {e}"))?;
|
||||
|
||||
let run_id = uuid::Uuid::new_v4().to_string();
|
||||
|
||||
// Resolve which executor runs this card: default agent, a personality, or
|
||||
// a skill — one autonomous-run interface, three presets (G4 + G3).
|
||||
let executor = resolve_executor(&config.workspace_dir, fresh_card.assigned_agent.as_deref());
|
||||
tracing::info!(
|
||||
card_id = %card_id,
|
||||
run_id = %run_id,
|
||||
executor = %executor.label,
|
||||
agent_id = %executor.agent_id,
|
||||
prompt_chars = prompt.chars().count(),
|
||||
"[task_dispatcher] card claimed (→in_progress), spawning autonomous run"
|
||||
);
|
||||
|
||||
let run_id_for_return = run_id.clone();
|
||||
let location_for_run = location.clone();
|
||||
tokio::spawn(async move {
|
||||
let outcome = run_autonomous(config, &executor, &prompt, &run_id).await;
|
||||
write_back(&location_for_run, &card_id, &run_id, outcome);
|
||||
});
|
||||
|
||||
Ok(DispatchOutcome::Running {
|
||||
run_id: run_id_for_return,
|
||||
})
|
||||
}
|
||||
|
||||
/// A resolved executor: which built-in agent definition to build, an optional
|
||||
/// system-prompt suffix carrying a personality identity or skill guidelines,
|
||||
/// and a label for logs/telemetry.
|
||||
#[derive(Debug, Clone, PartialEq)]
|
||||
struct ResolvedExecutor {
|
||||
agent_id: String,
|
||||
prompt_suffix: Option<String>,
|
||||
label: String,
|
||||
}
|
||||
|
||||
impl ResolvedExecutor {
|
||||
fn default_agent() -> Self {
|
||||
Self {
|
||||
agent_id: "orchestrator".to_string(),
|
||||
prompt_suffix: None,
|
||||
label: "default".to_string(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Map a card's `assigned_agent` handle to one of three executor presets:
|
||||
/// **personality** (scoped SOUL/MEMORY folded into the prompt suffix, run as
|
||||
/// that profile's agent), **skill** (orchestrator seeded with the skill's
|
||||
/// `SKILL.md` guidelines), or **built-in agent**. An unset or unresolved handle
|
||||
/// degrades to the default `orchestrator` — "use the personality if valid,
|
||||
/// otherwise the default agent."
|
||||
fn resolve_executor(workspace_dir: &Path, assigned: Option<&str>) -> ResolvedExecutor {
|
||||
let Some(handle) = assigned.map(str::trim).filter(|s| !s.is_empty()) else {
|
||||
return ResolvedExecutor::default_agent();
|
||||
};
|
||||
if handle == "orchestrator" {
|
||||
return ResolvedExecutor::default_agent();
|
||||
}
|
||||
|
||||
// 1) Personality (#2895): a user-defined profile with scoped identity.
|
||||
if let Ok(state) = crate::openhuman::agent::profiles::load_profiles(workspace_dir) {
|
||||
if let Some(profile) = state.profiles.iter().find(|p| p.id == handle) {
|
||||
let ctx = PersonalityContext::from_profile(workspace_dir, profile.clone());
|
||||
let mut preamble = format!(
|
||||
"You are acting as the personality `{}` (\"{}\"). {}",
|
||||
profile.id, profile.name, profile.description
|
||||
);
|
||||
if let Some(soul) = &ctx.soul_md_override {
|
||||
preamble.push_str("\n\n[Personality SOUL.md]\n");
|
||||
preamble.push_str(&truncate_chars(soul, EXECUTOR_PREAMBLE_MAX_CHARS));
|
||||
}
|
||||
if let Some(mem) = &ctx.memory_md_override {
|
||||
preamble.push_str("\n\n[Personality MEMORY.md]\n");
|
||||
preamble.push_str(&truncate_chars(mem, EXECUTOR_PREAMBLE_MAX_CHARS));
|
||||
}
|
||||
return ResolvedExecutor {
|
||||
agent_id: profile.agent_id.clone(),
|
||||
prompt_suffix: Some(preamble),
|
||||
label: format!("personality:{handle}"),
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
// 2) Skill (#2824): the same autonomous run, seeded with SKILL.md.
|
||||
if let Some(skill) = crate::openhuman::skills::registry::get_skill(workspace_dir, handle) {
|
||||
let guidelines = match &skill.definition.system_prompt {
|
||||
PromptSource::Inline(s) => truncate_chars(s, EXECUTOR_PREAMBLE_MAX_CHARS),
|
||||
_ => String::new(),
|
||||
};
|
||||
let suffix = format!(
|
||||
"You are executing this task as the skill `{handle}`. Follow these skill \
|
||||
guidelines exactly:\n\n{guidelines}"
|
||||
);
|
||||
return ResolvedExecutor {
|
||||
agent_id: "orchestrator".to_string(),
|
||||
prompt_suffix: Some(suffix),
|
||||
label: format!("skill:{handle}"),
|
||||
};
|
||||
}
|
||||
|
||||
// 3) Built-in agent definition.
|
||||
if AgentDefinitionRegistry::global()
|
||||
.and_then(|r| r.get(handle))
|
||||
.is_some()
|
||||
{
|
||||
return ResolvedExecutor {
|
||||
agent_id: handle.to_string(),
|
||||
prompt_suffix: None,
|
||||
label: format!("agent:{handle}"),
|
||||
};
|
||||
}
|
||||
|
||||
// 4) Unresolved → degrade to the default agent (don't fail the card).
|
||||
tracing::warn!(
|
||||
handle = %handle,
|
||||
"[task_dispatcher] assigned executor did not resolve to a personality/skill/agent; \
|
||||
using default orchestrator"
|
||||
);
|
||||
ResolvedExecutor {
|
||||
label: "default-fallback".to_string(),
|
||||
..ResolvedExecutor::default_agent()
|
||||
}
|
||||
}
|
||||
|
||||
/// Run the resolved executor as a single autonomous turn using the
|
||||
/// already-loaded config. The executor's prompt suffix (personality identity or
|
||||
/// skill guidelines) rides in the system prompt; the card goal is the turn input.
|
||||
///
|
||||
/// SECURITY / threat model (prompt injection): the card objective/content and
|
||||
/// `source_metadata` derive from external, attacker-influenceable text (e.g. a
|
||||
/// GitHub issue body anyone in a watched repo can file), and this background
|
||||
/// run is gate-free at the per-tool level (background turns auto-allow, like
|
||||
/// skill runs) while `build_task_prompt` may instruct it to write back to the
|
||||
/// upstream item. The interactive checkpoint is therefore the up-front
|
||||
/// **plan-approval gate** (`require_task_plan_approval`), which a human reviews
|
||||
/// before the run starts — not per-action egress/write approval. Egress is
|
||||
/// widened to `*` only when the operator set no explicit allow-list (matching
|
||||
/// skill runs, since real task work needs broad reach: git, package registries,
|
||||
/// provider APIs). Tightening egress to the source provider's domains for
|
||||
/// source-ingested runs is a considered follow-up (it would break general task
|
||||
/// work, so it needs to key off provenance) — tracked for a later PR.
|
||||
async fn run_autonomous(
|
||||
mut config: Config,
|
||||
executor: &ResolvedExecutor,
|
||||
prompt: &str,
|
||||
run_id: &str,
|
||||
) -> Result<String, String> {
|
||||
config.agent.max_tool_iterations = TASK_RUN_MAX_ITERATIONS;
|
||||
// Match skill-run egress handling: only widen to the permissive default
|
||||
// when the operator hasn't configured an explicit allow-list. See the
|
||||
// threat-model note above on why `*` is the default here.
|
||||
if config.http_request.allowed_domains.is_empty() {
|
||||
config.http_request.allowed_domains = vec!["*".to_string()];
|
||||
}
|
||||
|
||||
let mut agent = Agent::from_config_for_agent_with_profile(
|
||||
&config,
|
||||
&executor.agent_id,
|
||||
None,
|
||||
executor.prompt_suffix.clone(),
|
||||
)
|
||||
.map_err(|e| format!("build agent: {e:#}"))?;
|
||||
agent.set_event_context(run_id.to_string(), "task");
|
||||
agent.set_agent_definition_name(format!(
|
||||
"task-{}-{}",
|
||||
executor.label,
|
||||
run_id.get(..8).unwrap_or(run_id)
|
||||
));
|
||||
|
||||
with_autonomous_iter_cap(TASK_RUN_MAX_ITERATIONS, agent.run_single(prompt))
|
||||
.await
|
||||
.map_err(|e| format!("{e:#}"))
|
||||
}
|
||||
|
||||
/// Deterministic board write-back: the dispatcher owns the card lifecycle.
|
||||
/// Success → `done` + evidence; failure → `blocked` + blocker reason. An
|
||||
/// external write failure here is logged, never propagated — the run already
|
||||
/// happened.
|
||||
fn write_back(
|
||||
location: &BoardLocation,
|
||||
card_id: &str,
|
||||
run_id: &str,
|
||||
outcome: Result<String, String>,
|
||||
) {
|
||||
let patch = match &outcome {
|
||||
Ok(output) => {
|
||||
tracing::info!(
|
||||
card_id = %card_id,
|
||||
run_id = %run_id,
|
||||
output_chars = output.chars().count(),
|
||||
"[task_dispatcher] run complete → done"
|
||||
);
|
||||
CardPatch {
|
||||
status: Some(TaskCardStatus::Done),
|
||||
evidence: Some(vec![truncate_chars(output.trim(), EVIDENCE_MAX_CHARS)]),
|
||||
..Default::default()
|
||||
}
|
||||
}
|
||||
Err(err) => {
|
||||
tracing::warn!(
|
||||
card_id = %card_id,
|
||||
run_id = %run_id,
|
||||
error = %err,
|
||||
"[task_dispatcher] run failed → blocked"
|
||||
);
|
||||
CardPatch {
|
||||
status: Some(TaskCardStatus::Blocked),
|
||||
blocker: Some(truncate_chars(err, EVIDENCE_MAX_CHARS)),
|
||||
..Default::default()
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
if let Err(e) = ops::edit(location, card_id, patch) {
|
||||
tracing::error!(
|
||||
card_id = %card_id,
|
||||
run_id = %run_id,
|
||||
error = %e,
|
||||
"[task_dispatcher] board write-back failed (run outcome lost from board)"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
fn truncate_chars(s: &str, max: usize) -> String {
|
||||
if s.chars().count() <= max {
|
||||
return s.to_string();
|
||||
}
|
||||
let mut out: String = s.chars().take(max.saturating_sub(1)).collect();
|
||||
out.push('…');
|
||||
out
|
||||
}
|
||||
|
||||
// ── Board poller ──────────────────────────────────────────────────────────
|
||||
|
||||
/// How often the poller wakes to look for a dispatchable card.
|
||||
const POLLER_TICK_SECONDS: u64 = 60;
|
||||
|
||||
static POLLER_STARTED: OnceLock<()> = OnceLock::new();
|
||||
|
||||
/// Spawn the board poller. Idempotent — only the first call installs the loop.
|
||||
///
|
||||
/// Each tick it scans the `task-sources` board and dispatches the
|
||||
/// highest-urgency `todo` card via [`dispatch_card`], gated by background-AI
|
||||
/// capacity (`scheduler_gate`). This is the catch-all for cards that arrive
|
||||
/// without a proactive trigger (`TodoOnly` sources, manual cards, or proactive
|
||||
/// turns the gate skipped). Cards that *did* get a proactive trigger are
|
||||
/// dispatched by the triage arm; the claim-based lock makes firing both safe.
|
||||
pub fn start_board_poller() {
|
||||
if POLLER_STARTED.set(()).is_err() {
|
||||
tracing::debug!("[task_dispatcher:poller] already running, skipping start");
|
||||
return;
|
||||
}
|
||||
tokio::spawn(async move {
|
||||
tracing::info!(
|
||||
tick_seconds = POLLER_TICK_SECONDS,
|
||||
"[task_dispatcher:poller] starting"
|
||||
);
|
||||
let mut ticker = tokio::time::interval(Duration::from_secs(POLLER_TICK_SECONDS));
|
||||
ticker.tick().await; // skip the immediate fire so startup isn't slammed
|
||||
loop {
|
||||
ticker.tick().await;
|
||||
if let Err(e) = poll_once().await {
|
||||
tracing::warn!(error = %e, "[task_dispatcher:poller] tick failed (continuing)");
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
/// One poller tick: dispatch the highest-urgency `todo` card on the
|
||||
/// task-sources board, if any and if capacity allows. `pub(crate)` so tests can
|
||||
/// drive a tick without the real interval.
|
||||
pub(crate) async fn poll_once() -> Result<(), String> {
|
||||
// Gate on background-AI capacity (autonomy / power / pause). Dropping the
|
||||
// permit immediately is fine: this is a "may background work start now"
|
||||
// check; the run itself is detached.
|
||||
let Some(_permit) = crate::openhuman::scheduler_gate::wait_for_capacity().await else {
|
||||
tracing::debug!("[task_dispatcher:poller] scheduler gate denied capacity; idle tick");
|
||||
return Ok(());
|
||||
};
|
||||
|
||||
let config = Config::load_or_init()
|
||||
.await
|
||||
.map_err(|e| format!("load config: {e:#}"))?;
|
||||
if !config.task_sources.enabled {
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
let location = BoardLocation::Thread {
|
||||
workspace_dir: config.workspace_dir.clone(),
|
||||
thread_id: crate::openhuman::task_sources::TASK_SOURCES_THREAD_ID.to_string(),
|
||||
};
|
||||
let snapshot = ops::list(&location)?;
|
||||
|
||||
// `enforce_single_in_progress` caps the board at one running card, so if
|
||||
// one is already in progress there's nothing for this tick to claim.
|
||||
if snapshot
|
||||
.cards
|
||||
.iter()
|
||||
.any(|c| c.status == TaskCardStatus::InProgress)
|
||||
{
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
let Some(card) = pick_next_todo(&snapshot.cards) else {
|
||||
return Ok(());
|
||||
};
|
||||
|
||||
tracing::info!(
|
||||
card_id = %card.id,
|
||||
urgency = card_urgency(&card),
|
||||
"[task_dispatcher:poller] dispatching highest-urgency todo card"
|
||||
);
|
||||
dispatch_card(location, card).await.map(|_| ())
|
||||
}
|
||||
|
||||
/// Highest-urgency dispatchable card (`todo` or approved `ready`; urgency from
|
||||
/// `source_metadata.urgency`, default 0.0; ties broken toward the lower board
|
||||
/// `order`). Returns a clone. `dispatch_card` then either runs a `ready` card
|
||||
/// or parks a `todo` one for approval, per the autonomy setting.
|
||||
fn pick_next_todo(cards: &[TaskBoardCard]) -> Option<TaskBoardCard> {
|
||||
cards
|
||||
.iter()
|
||||
.filter(|c| matches!(c.status, TaskCardStatus::Todo | TaskCardStatus::Ready))
|
||||
.max_by(|a, b| {
|
||||
card_urgency(a)
|
||||
.partial_cmp(&card_urgency(b))
|
||||
.unwrap_or(std::cmp::Ordering::Equal)
|
||||
// On equal urgency, prefer the lower `order` (earlier card):
|
||||
// reversing the order comparison makes it the "greater" pick.
|
||||
.then(b.order.cmp(&a.order))
|
||||
})
|
||||
.cloned()
|
||||
}
|
||||
|
||||
fn card_urgency(card: &TaskBoardCard) -> f64 {
|
||||
card.source_metadata
|
||||
.as_ref()
|
||||
.and_then(|m| m.get("urgency"))
|
||||
.and_then(serde_json::Value::as_f64)
|
||||
.unwrap_or(0.0)
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use serde_json::json;
|
||||
|
||||
fn card(objective: Option<&str>) -> TaskBoardCard {
|
||||
TaskBoardCard {
|
||||
id: "task-1".into(),
|
||||
title: "[GitHub] Fix login bug".into(),
|
||||
status: TaskCardStatus::Todo,
|
||||
objective: objective.map(str::to_string),
|
||||
plan: vec![],
|
||||
assigned_agent: None,
|
||||
allowed_tools: vec![],
|
||||
approval_mode: None,
|
||||
acceptance_criteria: vec![],
|
||||
evidence: vec![],
|
||||
notes: None,
|
||||
blocker: None,
|
||||
source_metadata: None,
|
||||
order: 0,
|
||||
updated_at: String::new(),
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn prompt_uses_objective_then_falls_back_to_title() {
|
||||
let p = build_task_prompt(&card(Some("Fix the login bug")));
|
||||
assert!(p.contains("Fix the login bug"));
|
||||
assert!(!p.contains("[GitHub]"));
|
||||
|
||||
let p2 = build_task_prompt(&card(None));
|
||||
assert!(p2.contains("[GitHub] Fix login bug"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn prompt_includes_plan_and_acceptance_criteria() {
|
||||
let mut c = card(Some("Do it"));
|
||||
c.plan = vec!["step one".into(), "step two".into()];
|
||||
c.acceptance_criteria = vec!["tests pass".into()];
|
||||
let p = build_task_prompt(&c);
|
||||
assert!(p.contains("Plan:"));
|
||||
assert!(p.contains("1. step one"));
|
||||
assert!(p.contains("2. step two"));
|
||||
assert!(p.contains("Acceptance criteria"));
|
||||
assert!(p.contains("- tests pass"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn prompt_points_at_source_and_memory_when_metadata_present() {
|
||||
let mut c = card(Some("Resolve issue"));
|
||||
c.source_metadata = Some(json!({
|
||||
"provider": "github",
|
||||
"repo": "octo/repo",
|
||||
"external_id": "123",
|
||||
"url": "https://github.com/octo/repo/issues/123",
|
||||
}));
|
||||
let p = build_task_prompt(&c);
|
||||
assert!(p.contains("github octo/repo#123"));
|
||||
assert!(p.contains("memory_recall"));
|
||||
assert!(p.contains("https://github.com/octo/repo/issues/123"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn prompt_omits_source_block_without_metadata() {
|
||||
let p = build_task_prompt(&card(Some("Do it")));
|
||||
assert!(!p.contains("memory_recall"));
|
||||
assert!(!p.contains("record the outcome on the upstream source"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn prompt_includes_external_writeback_when_addressable() {
|
||||
let mut c = card(Some("Resolve issue"));
|
||||
c.source_metadata = Some(json!({
|
||||
"provider": "github",
|
||||
"repo": "octo/repo",
|
||||
"external_id": "123",
|
||||
}));
|
||||
let p = build_task_prompt(&c);
|
||||
assert!(p.contains("record the outcome on the upstream source"));
|
||||
assert!(p.contains("close/resolve the item"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn prompt_omits_writeback_when_not_addressable() {
|
||||
// Urgency-only metadata (no provider/external_id) can't address an
|
||||
// upstream item, so no write-back instruction.
|
||||
let mut c = card(Some("Do it"));
|
||||
c.source_metadata = Some(json!({ "urgency": 0.5 }));
|
||||
let p = build_task_prompt(&c);
|
||||
assert!(!p.contains("record the outcome on the upstream source"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn truncate_caps_long_strings() {
|
||||
let s = "x".repeat(5_000);
|
||||
let out = truncate_chars(&s, EVIDENCE_MAX_CHARS);
|
||||
assert!(out.chars().count() <= EVIDENCE_MAX_CHARS);
|
||||
assert!(out.ends_with('…'));
|
||||
}
|
||||
|
||||
fn card_with(
|
||||
id: &str,
|
||||
status: TaskCardStatus,
|
||||
urgency: Option<f64>,
|
||||
order: u32,
|
||||
) -> TaskBoardCard {
|
||||
let mut c = card(Some("obj"));
|
||||
c.id = id.into();
|
||||
c.status = status;
|
||||
c.order = order;
|
||||
c.source_metadata = urgency.map(|u| json!({ "urgency": u }));
|
||||
c
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn poller_picks_highest_urgency_todo_skipping_other_statuses() {
|
||||
let cards = vec![
|
||||
card_with("a", TaskCardStatus::Todo, Some(0.3), 0),
|
||||
card_with("b", TaskCardStatus::Done, Some(0.99), 1),
|
||||
card_with("c", TaskCardStatus::Todo, Some(0.8), 2),
|
||||
card_with("d", TaskCardStatus::Todo, None, 3),
|
||||
];
|
||||
let picked = pick_next_todo(&cards).expect("a todo card is available");
|
||||
assert_eq!(
|
||||
picked.id, "c",
|
||||
"highest-urgency todo wins, done card ignored"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn poller_breaks_urgency_ties_toward_lower_order() {
|
||||
let cards = vec![
|
||||
card_with("late", TaskCardStatus::Todo, Some(0.5), 5),
|
||||
card_with("early", TaskCardStatus::Todo, Some(0.5), 2),
|
||||
];
|
||||
assert_eq!(pick_next_todo(&cards).unwrap().id, "early");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn poller_returns_none_when_no_todo_cards() {
|
||||
let cards = vec![card_with("a", TaskCardStatus::Done, Some(0.9), 0)];
|
||||
assert!(pick_next_todo(&cards).is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn poller_dispatches_ready_cards_and_skips_approval_states() {
|
||||
// Approved `ready` cards are dispatchable; `awaiting_approval` and
|
||||
// `rejected` are not.
|
||||
let cards = vec![
|
||||
card_with("await", TaskCardStatus::AwaitingApproval, Some(0.99), 0),
|
||||
card_with("rej", TaskCardStatus::Rejected, Some(0.95), 1),
|
||||
card_with("ready", TaskCardStatus::Ready, Some(0.5), 2),
|
||||
];
|
||||
assert_eq!(pick_next_todo(&cards).unwrap().id, "ready");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn poller_prefers_higher_urgency_across_todo_and_ready() {
|
||||
let cards = vec![
|
||||
card_with("ready-low", TaskCardStatus::Ready, Some(0.3), 0),
|
||||
card_with("todo-high", TaskCardStatus::Todo, Some(0.9), 1),
|
||||
];
|
||||
assert_eq!(pick_next_todo(&cards).unwrap().id, "todo-high");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn resolver_defaults_to_orchestrator_for_unset_or_orchestrator_handle() {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
for handle in [None, Some(""), Some(" "), Some("orchestrator")] {
|
||||
let r = resolve_executor(dir.path(), handle);
|
||||
assert_eq!(r.agent_id, "orchestrator");
|
||||
assert_eq!(r.label, "default");
|
||||
assert!(r.prompt_suffix.is_none());
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn resolver_uses_personality_branch_for_builtin_profile() {
|
||||
// `load_profiles` returns built-in profiles for any empty workspace, so
|
||||
// the personality branch is reachable with no fixture file. "research"
|
||||
// is a built-in profile backed by the "researcher" agent.
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let r = resolve_executor(dir.path(), Some("research"));
|
||||
assert_eq!(r.label, "personality:research");
|
||||
assert_eq!(r.agent_id, "researcher");
|
||||
let suffix = r.prompt_suffix.expect("personality preamble present");
|
||||
assert!(suffix.contains("acting as the personality `research`"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn resolver_degrades_to_default_for_unresolved_handle() {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let r = resolve_executor(dir.path(), Some("no-such-executor-xyz"));
|
||||
assert_eq!(r.agent_id, "orchestrator");
|
||||
assert_eq!(r.label, "default-fallback");
|
||||
assert!(r.prompt_suffix.is_none());
|
||||
}
|
||||
|
||||
fn board_loc(dir: &std::path::Path) -> BoardLocation {
|
||||
BoardLocation::Thread {
|
||||
workspace_dir: dir.to_path_buf(),
|
||||
thread_id: "t1".to_string(),
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn write_back_marks_done_with_evidence_on_success() {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let loc = board_loc(dir.path());
|
||||
let id = ops::add(&loc, "do the thing", CardPatch::default())
|
||||
.unwrap()
|
||||
.cards[0]
|
||||
.id
|
||||
.clone();
|
||||
ops::update_status(&loc, &id, TaskCardStatus::InProgress).unwrap();
|
||||
|
||||
write_back(
|
||||
&loc,
|
||||
&id,
|
||||
"run-1",
|
||||
Ok("completed: opened PR #5".to_string()),
|
||||
);
|
||||
|
||||
let card = ops::list(&loc)
|
||||
.unwrap()
|
||||
.cards
|
||||
.into_iter()
|
||||
.find(|c| c.id == id)
|
||||
.unwrap();
|
||||
assert_eq!(card.status, TaskCardStatus::Done);
|
||||
assert!(card.evidence.iter().any(|e| e.contains("opened PR #5")));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn write_back_marks_blocked_with_reason_on_failure() {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let loc = board_loc(dir.path());
|
||||
let id = ops::add(&loc, "do the thing", CardPatch::default())
|
||||
.unwrap()
|
||||
.cards[0]
|
||||
.id
|
||||
.clone();
|
||||
ops::update_status(&loc, &id, TaskCardStatus::InProgress).unwrap();
|
||||
|
||||
write_back(&loc, &id, "run-1", Err("agent build failed".to_string()));
|
||||
|
||||
let card = ops::list(&loc)
|
||||
.unwrap()
|
||||
.cards
|
||||
.into_iter()
|
||||
.find(|c| c.id == id)
|
||||
.unwrap();
|
||||
assert_eq!(card.status, TaskCardStatus::Blocked);
|
||||
assert!(card
|
||||
.blocker
|
||||
.as_deref()
|
||||
.unwrap_or_default()
|
||||
.contains("agent build failed"));
|
||||
}
|
||||
}
|
||||
@@ -255,6 +255,7 @@ fn patch_from_args(args: &serde_json::Value) -> anyhow::Result<CardPatch> {
|
||||
evidence: optional_string_array(args, "evidence")?,
|
||||
notes: optional_string(args, "notes"),
|
||||
blocker: optional_string(args, "blocker"),
|
||||
source_metadata: None,
|
||||
})
|
||||
}
|
||||
|
||||
|
||||
@@ -10,6 +10,18 @@
|
||||
use chrono::{DateTime, Utc};
|
||||
use serde_json::Value;
|
||||
|
||||
use crate::openhuman::todos::ops::BoardLocation;
|
||||
|
||||
/// Links a trigger to the task-board card it concerns, so the triage
|
||||
/// `apply_decision` arm can hand the card to the deterministic dispatcher
|
||||
/// (claim + autonomous run + write-back) instead of the one-shot triage
|
||||
/// sub-agent. `None` for triggers with no board card (composio/webhook/cron).
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct TaskCardLink {
|
||||
pub card_id: String,
|
||||
pub location: BoardLocation,
|
||||
}
|
||||
|
||||
/// Where the trigger came from, plus source-specific identifiers the
|
||||
/// triage prompt wants to surface (toolkit/trigger slug, cron job id,
|
||||
/// webhook tunnel id, etc.).
|
||||
@@ -84,6 +96,10 @@ pub struct TriggerEnvelope {
|
||||
/// pipeline can report a meaningful `latency_ms` when it
|
||||
/// publishes [`crate::core::event_bus::DomainEvent::TriggerEvaluated`].
|
||||
pub received_at: DateTime<Utc>,
|
||||
|
||||
/// Set when this trigger corresponds to a task-board card, so the
|
||||
/// triage escalation arm routes it through the deterministic dispatcher.
|
||||
pub card_link: Option<TaskCardLink>,
|
||||
}
|
||||
|
||||
impl TriggerEnvelope {
|
||||
@@ -118,6 +134,7 @@ impl TriggerEnvelope {
|
||||
display_label: format!("composio/{toolkit}/{trigger}"),
|
||||
payload,
|
||||
received_at: Utc::now(),
|
||||
card_link: None,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -136,6 +153,7 @@ impl TriggerEnvelope {
|
||||
display_label: format!("webhook/{method}/{path}"),
|
||||
payload,
|
||||
received_at: Utc::now(),
|
||||
card_link: None,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -153,6 +171,7 @@ impl TriggerEnvelope {
|
||||
display_label: format!("cron/{job_name}"),
|
||||
payload: serde_json::json!({ "output": output }),
|
||||
received_at: Utc::now(),
|
||||
card_link: None,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -171,8 +190,17 @@ impl TriggerEnvelope {
|
||||
display_label: format!("external/{caller_id}"),
|
||||
payload,
|
||||
received_at: Utc::now(),
|
||||
card_link: None,
|
||||
}
|
||||
}
|
||||
|
||||
/// Attach a task-board card link so the triage escalation arm dispatches
|
||||
/// the card deterministically (claim + autonomous run + write-back).
|
||||
#[must_use]
|
||||
pub fn with_task_card(mut self, card_id: String, location: BoardLocation) -> Self {
|
||||
self.card_link = Some(TaskCardLink { card_id, location });
|
||||
self
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
@@ -202,6 +230,28 @@ mod tests {
|
||||
assert_eq!(env.payload["from"], "a@b.com");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn with_task_card_attaches_card_link() {
|
||||
let env = TriggerEnvelope::from_external(
|
||||
"task_sources:ts-1",
|
||||
"external task ingested",
|
||||
json!({}),
|
||||
);
|
||||
assert!(env.card_link.is_none(), "no link by default");
|
||||
|
||||
let location = BoardLocation::Thread {
|
||||
workspace_dir: std::path::PathBuf::from("/tmp/ws"),
|
||||
thread_id: "task-sources".to_string(),
|
||||
};
|
||||
let linked = env.with_task_card("card-1".to_string(), location);
|
||||
let link = linked.card_link.expect("card_link attached");
|
||||
assert_eq!(link.card_id, "card-1");
|
||||
match link.location {
|
||||
BoardLocation::Thread { thread_id, .. } => assert_eq!(thread_id, "task-sources"),
|
||||
_ => panic!("expected Thread board location"),
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn composio_envelope_falls_back_to_metadata_id_when_uuid_missing() {
|
||||
let env = TriggerEnvelope::from_composio(
|
||||
|
||||
@@ -27,7 +27,7 @@ use crate::openhuman::agent::Agent;
|
||||
use crate::openhuman::config::Config;
|
||||
|
||||
use super::decision::TriageAction;
|
||||
use super::envelope::TriggerEnvelope;
|
||||
use super::envelope::{TaskCardLink, TriggerEnvelope};
|
||||
use super::evaluator::TriageRun;
|
||||
use super::events;
|
||||
|
||||
@@ -57,6 +57,13 @@ pub async fn apply_decision(run: TriageRun, envelope: &TriggerEnvelope) -> anyho
|
||||
reason = %run.decision.reason,
|
||||
"[triage::escalation] DROP — no downstream work"
|
||||
);
|
||||
// A dropped trigger that carries a board card (proactive
|
||||
// task-source ingest) must be terminally gated, or the board
|
||||
// poller — which dispatches any `Todo`/`Ready` card regardless of
|
||||
// the triage verdict — would re-run it on the next tick, silently
|
||||
// breaking the noise-gating contract documented on
|
||||
// `SourceTarget::AgentTodoProactive`.
|
||||
gate_linked_card_terminal(envelope, "drop");
|
||||
}
|
||||
TriageAction::Acknowledge => {
|
||||
tracing::info!(
|
||||
@@ -65,6 +72,9 @@ pub async fn apply_decision(run: TriageRun, envelope: &TriggerEnvelope) -> anyho
|
||||
reason = %run.decision.reason,
|
||||
"[triage::escalation] ACKNOWLEDGE — logged (memory-write is a future addition)"
|
||||
);
|
||||
// Acknowledge means "seen, no autonomous action needed" — same as
|
||||
// drop, the linked card must not be picked up by the board poller.
|
||||
gate_linked_card_terminal(envelope, "acknowledge");
|
||||
}
|
||||
TriageAction::React | TriageAction::Escalate => {
|
||||
let target = run
|
||||
@@ -85,6 +95,44 @@ pub async fn apply_decision(run: TriageRun, envelope: &TriggerEnvelope) -> anyho
|
||||
"[triage::escalation] dispatching sub-agent"
|
||||
);
|
||||
|
||||
// ── Unified task-board path ───────────────────────
|
||||
// A trigger linked to a board card is handed to the deterministic
|
||||
// dispatcher (claim → autonomous run → write-back). The claim
|
||||
// (todo→in_progress) deduplicates against the board poller, so
|
||||
// firing both is safe. Non-card triggers (composio/webhook/cron)
|
||||
// fall through to the one-shot triage sub-agent below.
|
||||
if let Some(link) = &envelope.card_link {
|
||||
use crate::openhuman::agent::task_dispatcher::DispatchOutcome;
|
||||
match dispatch_linked_card(link).await {
|
||||
Ok(DispatchOutcome::Running { run_id }) => {
|
||||
tracing::info!(
|
||||
card_id = %link.card_id,
|
||||
run_id = %run_id,
|
||||
"[triage::escalation] task-card dispatched to deterministic runner"
|
||||
);
|
||||
events::publish_escalated(envelope, "task_dispatcher");
|
||||
}
|
||||
Ok(DispatchOutcome::AwaitingApproval) => {
|
||||
// Parked for plan approval (autonomy gate). Not an
|
||||
// escalation yet — the approval flow resumes it.
|
||||
tracing::info!(
|
||||
card_id = %link.card_id,
|
||||
"[triage::escalation] task-card parked awaiting plan approval"
|
||||
);
|
||||
}
|
||||
Err(reason) => {
|
||||
// A failed claim (another card already in progress, or
|
||||
// the card vanished) is benign — the poller retries.
|
||||
tracing::info!(
|
||||
card_id = %link.card_id,
|
||||
reason = %reason,
|
||||
"[triage::escalation] task-card dispatch skipped (claim failed?)"
|
||||
);
|
||||
}
|
||||
}
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
// ── External-effect approval gate (#1339) ─────────
|
||||
// React / Escalate fire a sub-agent that may call
|
||||
// external-effect tools on the user's behalf. Catching
|
||||
@@ -268,6 +316,77 @@ async fn dispatch_target_agent(agent_id: &str, prompt: &str) -> anyhow::Result<S
|
||||
Ok(outcome.output)
|
||||
}
|
||||
|
||||
/// Load the linked card from its board and hand it to the deterministic task
|
||||
/// dispatcher (claim → autonomous run → write-back). Errors (card not found,
|
||||
/// or claim rejected because another card is already in progress) propagate to
|
||||
/// the caller, which treats them as benign skips.
|
||||
async fn dispatch_linked_card(
|
||||
link: &TaskCardLink,
|
||||
) -> Result<crate::openhuman::agent::task_dispatcher::DispatchOutcome, String> {
|
||||
let snapshot = crate::openhuman::todos::ops::list(&link.location)?;
|
||||
let card = snapshot
|
||||
.cards
|
||||
.into_iter()
|
||||
.find(|c| c.id == link.card_id)
|
||||
.ok_or_else(|| format!("card `{}` not found on board", link.card_id))?;
|
||||
crate::openhuman::agent::task_dispatcher::dispatch_card(link.location.clone(), card).await
|
||||
}
|
||||
|
||||
/// Terminally gate a card-linked trigger that triage decided to `drop` /
|
||||
/// `acknowledge`, so the board poller (which dispatches any pending
|
||||
/// `Todo`/`Ready` card) won't re-run it. Only a still-pending card is gated;
|
||||
/// if it already advanced (the poller claimed it, or it's already terminal)
|
||||
/// it is left untouched. Best-effort: a missing card or write failure is
|
||||
/// logged, never propagated — the trigger was already evaluated.
|
||||
fn gate_linked_card_terminal(envelope: &TriggerEnvelope, decision: &str) {
|
||||
use crate::openhuman::agent::task_board::TaskCardStatus;
|
||||
use crate::openhuman::todos::ops;
|
||||
|
||||
let Some(link) = &envelope.card_link else {
|
||||
return;
|
||||
};
|
||||
|
||||
let current = match ops::list(&link.location) {
|
||||
Ok(snapshot) => snapshot
|
||||
.cards
|
||||
.into_iter()
|
||||
.find(|c| c.id == link.card_id)
|
||||
.map(|c| c.status),
|
||||
Err(e) => {
|
||||
tracing::warn!(
|
||||
card_id = %link.card_id,
|
||||
error = %e,
|
||||
"[triage::escalation] reload before gating linked card failed"
|
||||
);
|
||||
return;
|
||||
}
|
||||
};
|
||||
|
||||
match current {
|
||||
Some(TaskCardStatus::Todo | TaskCardStatus::Ready | TaskCardStatus::AwaitingApproval) => {
|
||||
match ops::update_status(&link.location, &link.card_id, TaskCardStatus::Rejected) {
|
||||
Ok(_) => tracing::info!(
|
||||
card_id = %link.card_id,
|
||||
decision = %decision,
|
||||
"[triage::escalation] gated task-card → rejected (poller will skip)"
|
||||
),
|
||||
Err(e) => tracing::warn!(
|
||||
card_id = %link.card_id,
|
||||
decision = %decision,
|
||||
error = %e,
|
||||
"[triage::escalation] failed to gate task-card (poller may re-dispatch)"
|
||||
),
|
||||
}
|
||||
}
|
||||
other => tracing::debug!(
|
||||
card_id = %link.card_id,
|
||||
decision = %decision,
|
||||
status = ?other,
|
||||
"[triage::escalation] linked task-card not pending; no gating needed"
|
||||
),
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
@@ -426,6 +545,75 @@ mod tests {
|
||||
)));
|
||||
}
|
||||
|
||||
fn seed_task_card() -> (
|
||||
tempfile::TempDir,
|
||||
crate::openhuman::todos::ops::BoardLocation,
|
||||
String,
|
||||
) {
|
||||
use crate::openhuman::todos::ops::{self, BoardLocation, CardPatch};
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let location = BoardLocation::Thread {
|
||||
workspace_dir: dir.path().to_path_buf(),
|
||||
thread_id: "task-sources".to_string(),
|
||||
};
|
||||
let card_id = ops::add(&location, "ingested issue", CardPatch::default())
|
||||
.unwrap()
|
||||
.cards[0]
|
||||
.id
|
||||
.clone();
|
||||
(dir, location, card_id)
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn apply_decision_drop_gates_linked_card_to_rejected() {
|
||||
use crate::openhuman::agent::task_board::TaskCardStatus;
|
||||
use crate::openhuman::todos::ops;
|
||||
|
||||
let _events_guard = test_events_guard().await;
|
||||
let _ = init_global(32);
|
||||
let (_dir, location, card_id) = seed_task_card();
|
||||
|
||||
let envelope = envelope("esc-drop-card").with_task_card(card_id.clone(), location.clone());
|
||||
apply_decision(run(TriageAction::Drop), &envelope)
|
||||
.await
|
||||
.expect("drop should not fail");
|
||||
|
||||
let status = ops::list(&location)
|
||||
.unwrap()
|
||||
.cards
|
||||
.into_iter()
|
||||
.find(|c| c.id == card_id)
|
||||
.map(|c| c.status);
|
||||
assert_eq!(
|
||||
status,
|
||||
Some(TaskCardStatus::Rejected),
|
||||
"a dropped card-linked trigger must be gated terminally so the board poller skips it"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn apply_decision_acknowledge_gates_linked_card_to_rejected() {
|
||||
use crate::openhuman::agent::task_board::TaskCardStatus;
|
||||
use crate::openhuman::todos::ops;
|
||||
|
||||
let _events_guard = test_events_guard().await;
|
||||
let _ = init_global(32);
|
||||
let (_dir, location, card_id) = seed_task_card();
|
||||
|
||||
let envelope = envelope("esc-ack-card").with_task_card(card_id.clone(), location.clone());
|
||||
apply_decision(run(TriageAction::Acknowledge), &envelope)
|
||||
.await
|
||||
.expect("acknowledge should not fail");
|
||||
|
||||
let status = ops::list(&location)
|
||||
.unwrap()
|
||||
.cards
|
||||
.into_iter()
|
||||
.find(|c| c.id == card_id)
|
||||
.map(|c| c.status);
|
||||
assert_eq!(status, Some(TaskCardStatus::Rejected));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn apply_decision_react_failure_publishes_failed_event() {
|
||||
let _events_guard = test_events_guard().await;
|
||||
|
||||
@@ -69,6 +69,9 @@ pub async fn start_channels(mut config: Config) -> Result<()> {
|
||||
// configured external sources onto the agent's todo board.
|
||||
crate::openhuman::task_sources::bus::register_task_sources_subscriber();
|
||||
crate::openhuman::task_sources::start_periodic_poll();
|
||||
// Board poller: dispatch the highest-urgency `todo` card on the
|
||||
// task-sources board (catch-all for cards without a proactive trigger).
|
||||
crate::openhuman::agent::task_dispatcher::start_board_poller();
|
||||
// Native request handlers. Re-registering is safe (latest wins) so
|
||||
// this is idempotent even if `bootstrap_core_runtime` also runs.
|
||||
// Must happen before `run_message_dispatch_loop` begins, because
|
||||
|
||||
@@ -104,6 +104,7 @@ pub async fn handle_ingest(params: Map<String, Value>) -> Result<Value, String>
|
||||
"raw": req.raw_payload,
|
||||
}),
|
||||
received_at: ingest_started_at,
|
||||
card_link: None,
|
||||
};
|
||||
|
||||
match run_triage(&envelope).await {
|
||||
|
||||
@@ -44,6 +44,7 @@ pub async fn add(
|
||||
interval_secs: Option<u64>,
|
||||
target: Option<SourceTarget>,
|
||||
max_tasks_per_fetch: Option<u32>,
|
||||
assigned_executor: Option<String>,
|
||||
) -> Result<RpcOutcome<TaskSource>, String> {
|
||||
let defaults = &config.task_sources;
|
||||
let interval_secs = interval_secs.unwrap_or(defaults.default_interval_secs);
|
||||
@@ -66,9 +67,25 @@ pub async fn add(
|
||||
)
|
||||
.map_err(|e| e.to_string())?;
|
||||
|
||||
// Apply the optional static executor routing (G7) as a follow-up patch so
|
||||
// `add_source`'s signature (and its many callers) stays unchanged.
|
||||
let source = match assigned_executor.filter(|s| !s.trim().is_empty()) {
|
||||
Some(executor) => store::update_source(
|
||||
config,
|
||||
&source.id,
|
||||
TaskSourcePatch {
|
||||
assigned_executor: Some(executor),
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.map_err(|e| e.to_string())?,
|
||||
None => source,
|
||||
};
|
||||
|
||||
tracing::info!(
|
||||
source_id = %source.id,
|
||||
provider = %source.provider.as_str(),
|
||||
assigned_executor = ?source.assigned_executor,
|
||||
"[task_sources:ops] add created source"
|
||||
);
|
||||
Ok(RpcOutcome::new(source, vec![]))
|
||||
|
||||
@@ -165,6 +165,7 @@ mod tests {
|
||||
interval_secs,
|
||||
target: SourceTarget::TodoOnly,
|
||||
max_tasks_per_fetch: 25,
|
||||
assigned_executor: None,
|
||||
created_at: Utc::now(),
|
||||
last_fetch_at: None,
|
||||
last_status: None,
|
||||
|
||||
@@ -19,7 +19,7 @@ use crate::openhuman::todos::ops::{
|
||||
};
|
||||
use crate::openhuman::{scheduler_gate, todos};
|
||||
|
||||
use super::types::{EnrichedTask, SourceTarget, TaskSource};
|
||||
use super::types::{EnrichedTask, FilterSpec, SourceTarget, TaskSource};
|
||||
|
||||
/// Stable thread id whose board collects every ingested task.
|
||||
pub const TASK_SOURCES_THREAD_ID: &str = "task-sources";
|
||||
@@ -44,7 +44,7 @@ pub async fn route_enriched(
|
||||
Ok(card_id)
|
||||
}
|
||||
SourceTarget::AgentTodoProactive => {
|
||||
dispatch_triage(source, enriched).await?;
|
||||
dispatch_triage(config, source, enriched, &card_id).await?;
|
||||
Ok(card_id)
|
||||
}
|
||||
}
|
||||
@@ -112,11 +112,36 @@ fn add_card(
|
||||
Some(notes_parts.join("\n"))
|
||||
};
|
||||
|
||||
// Objective: the bare upstream title (the card `content`/title is the
|
||||
// `[provider] title` display form; the objective is the clean goal the
|
||||
// executing agent works toward).
|
||||
let objective = {
|
||||
let t = task.title.trim();
|
||||
(!t.is_empty()).then(|| t.to_string())
|
||||
};
|
||||
|
||||
// Stamp the source identifiers the downstream dispatcher / write-back
|
||||
// needs (provider + repo + issue id + url) plus the enrichment urgency
|
||||
// used for prioritisation. This is the only writer of `source_metadata`.
|
||||
let source_metadata = build_source_metadata(source, enriched);
|
||||
|
||||
// G7: pre-assign the card to the source's configured executor so the
|
||||
// dispatcher runs it deterministically (no LLM router). Unset → unassigned.
|
||||
let assigned_agent = source
|
||||
.assigned_executor
|
||||
.as_deref()
|
||||
.map(str::trim)
|
||||
.filter(|s| !s.is_empty())
|
||||
.map(str::to_string);
|
||||
|
||||
let snapshot = todo_add(
|
||||
&location,
|
||||
&content,
|
||||
CardPatch {
|
||||
notes,
|
||||
objective,
|
||||
assigned_agent,
|
||||
source_metadata: Some(source_metadata),
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
@@ -139,10 +164,43 @@ fn add_card(
|
||||
Ok(new_card_id)
|
||||
}
|
||||
|
||||
/// Build the card's `source_metadata` from the originating source + task:
|
||||
/// the provider/repo/issue identifiers a later dispatcher or external
|
||||
/// write-back needs to address the upstream item, plus the enrichment
|
||||
/// urgency used to prioritise pickup. Repo is only present for GitHub
|
||||
/// sources (the other providers don't carry a repo concept).
|
||||
fn build_source_metadata(source: &TaskSource, enriched: &EnrichedTask) -> serde_json::Value {
|
||||
let task = &enriched.task;
|
||||
let mut meta = json!({
|
||||
"provider": task.provider,
|
||||
"source_id": source.id,
|
||||
"external_id": task.external_id,
|
||||
"urgency": enriched.urgency,
|
||||
});
|
||||
if let Some(url) = task.url.as_deref().map(str::trim).filter(|s| !s.is_empty()) {
|
||||
meta["url"] = json!(url);
|
||||
}
|
||||
if let FilterSpec::Github {
|
||||
repo: Some(repo), ..
|
||||
} = &source.filter
|
||||
{
|
||||
let repo = repo.trim();
|
||||
if !repo.is_empty() {
|
||||
meta["repo"] = json!(repo);
|
||||
}
|
||||
}
|
||||
meta
|
||||
}
|
||||
|
||||
/// Dispatch a triage turn for a proactive task, gated by scheduler
|
||||
/// capacity. Card creation already happened; a gated-off or deferred
|
||||
/// turn is non-fatal — the task still sits on the board.
|
||||
async fn dispatch_triage(source: &TaskSource, enriched: &EnrichedTask) -> Result<(), String> {
|
||||
async fn dispatch_triage(
|
||||
config: &Config,
|
||||
source: &TaskSource,
|
||||
enriched: &EnrichedTask,
|
||||
card_id: &str,
|
||||
) -> Result<(), String> {
|
||||
// Respect background-AI throttling. When the gate denies capacity
|
||||
// (Off / paused), we keep the card but skip the proactive turn.
|
||||
let Some(_permit) = scheduler_gate::wait_for_capacity().await else {
|
||||
@@ -164,11 +222,19 @@ async fn dispatch_triage(source: &TaskSource, enriched: &EnrichedTask) -> Result
|
||||
"sourceId": source.id,
|
||||
});
|
||||
|
||||
// Link the envelope to the board card so triage's escalation arm routes
|
||||
// it through the deterministic dispatcher (claim → autonomous run →
|
||||
// write-back) instead of the one-shot triage sub-agent.
|
||||
let location = BoardLocation::Thread {
|
||||
workspace_dir: config.workspace_dir.clone(),
|
||||
thread_id: TASK_SOURCES_THREAD_ID.to_string(),
|
||||
};
|
||||
let envelope = TriggerEnvelope::from_external(
|
||||
&format!("task_sources:{}", source.id),
|
||||
"external task ingested",
|
||||
payload,
|
||||
);
|
||||
)
|
||||
.with_task_card(card_id.to_string(), location);
|
||||
|
||||
let outcome = run_triage(&envelope)
|
||||
.await
|
||||
@@ -228,6 +294,9 @@ pub fn board_cards(
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::openhuman::task_sources::types::ProviderSlug;
|
||||
use crate::openhuman::task_sources::NormalizedTask;
|
||||
use chrono::Utc;
|
||||
|
||||
#[test]
|
||||
fn provider_label_titlecases_known_and_unknown() {
|
||||
@@ -236,4 +305,146 @@ mod tests {
|
||||
assert_eq!(provider_label("asana"), "Asana");
|
||||
assert_eq!(provider_label(""), "");
|
||||
}
|
||||
|
||||
fn github_source(repo: Option<&str>) -> TaskSource {
|
||||
TaskSource {
|
||||
id: "ts-1".into(),
|
||||
provider: ProviderSlug::Github,
|
||||
connection_id: None,
|
||||
name: None,
|
||||
enabled: true,
|
||||
filter: FilterSpec::Github {
|
||||
repo: repo.map(str::to_string),
|
||||
labels: vec![],
|
||||
assignee_is_me: true,
|
||||
state: None,
|
||||
extra: json!({}),
|
||||
},
|
||||
interval_secs: 1800,
|
||||
target: SourceTarget::AgentTodoProactive,
|
||||
max_tasks_per_fetch: 25,
|
||||
assigned_executor: None,
|
||||
created_at: Utc::now(),
|
||||
last_fetch_at: None,
|
||||
last_status: None,
|
||||
}
|
||||
}
|
||||
|
||||
fn enriched(external_id: &str, url: Option<&str>, urgency: f32) -> EnrichedTask {
|
||||
let task = NormalizedTask {
|
||||
external_id: external_id.into(),
|
||||
provider: "github".into(),
|
||||
title: "Fix the bug".into(),
|
||||
url: url.map(str::to_string),
|
||||
..Default::default()
|
||||
};
|
||||
EnrichedTask {
|
||||
task,
|
||||
summary: "Fix the bug".into(),
|
||||
urgency,
|
||||
linked_people: vec![],
|
||||
linked_memory_ids: vec![],
|
||||
agent_prompt: "do it".into(),
|
||||
enriched_at: Utc::now(),
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn source_metadata_carries_github_repo_and_identifiers() {
|
||||
let src = github_source(Some("octo/repo"));
|
||||
let e = enriched("123", Some("https://github.com/octo/repo/issues/123"), 0.7);
|
||||
let meta = build_source_metadata(&src, &e);
|
||||
assert_eq!(meta["provider"], json!("github"));
|
||||
assert_eq!(meta["source_id"], json!("ts-1"));
|
||||
assert_eq!(meta["external_id"], json!("123"));
|
||||
assert_eq!(meta["repo"], json!("octo/repo"));
|
||||
assert_eq!(
|
||||
meta["url"],
|
||||
json!("https://github.com/octo/repo/issues/123")
|
||||
);
|
||||
let urgency = meta["urgency"].as_f64().expect("urgency is a number");
|
||||
assert!((urgency - 0.7).abs() < 1e-6, "urgency was {urgency}");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn source_metadata_omits_absent_repo_and_url() {
|
||||
let src = github_source(None);
|
||||
let e = enriched("9", None, 0.4);
|
||||
let meta = build_source_metadata(&src, &e);
|
||||
assert!(meta.get("repo").is_none());
|
||||
assert!(meta.get("url").is_none());
|
||||
assert_eq!(meta["external_id"], json!("9"));
|
||||
let urgency = meta["urgency"].as_f64().expect("urgency is a number");
|
||||
assert!((urgency - 0.4).abs() < 1e-6, "urgency was {urgency}");
|
||||
}
|
||||
|
||||
fn temp_config() -> (tempfile::TempDir, Config) {
|
||||
let tmp = tempfile::tempdir().unwrap();
|
||||
let config = Config {
|
||||
workspace_dir: tmp.path().join("workspace"),
|
||||
config_path: tmp.path().join("config.toml"),
|
||||
..Config::default()
|
||||
};
|
||||
std::fs::create_dir_all(&config.workspace_dir).unwrap();
|
||||
(tmp, config)
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn add_card_stamps_objective_assigned_agent_and_metadata() {
|
||||
let (_tmp, config) = temp_config();
|
||||
let mut src = github_source(Some("octo/repo"));
|
||||
// Whitespace around the executor must be trimmed into assigned_agent.
|
||||
src.assigned_executor = Some(" agent-x ".into());
|
||||
let e = enriched("123", Some("https://github.com/octo/repo/issues/123"), 0.7);
|
||||
|
||||
add_card(&config, &src, &e, None).expect("add_card succeeds");
|
||||
|
||||
let cards = board_cards(&config).expect("board_cards");
|
||||
assert_eq!(cards.len(), 1);
|
||||
let card = &cards[0];
|
||||
// Display title is the `[provider] title` form; objective is the bare title.
|
||||
assert_eq!(card.title, "[GitHub] Fix the bug");
|
||||
assert_eq!(card.objective.as_deref(), Some("Fix the bug"));
|
||||
assert_eq!(card.assigned_agent.as_deref(), Some("agent-x"));
|
||||
let meta = card
|
||||
.source_metadata
|
||||
.as_ref()
|
||||
.expect("source_metadata present");
|
||||
assert_eq!(meta["external_id"], json!("123"));
|
||||
assert_eq!(meta["repo"], json!("octo/repo"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn add_card_drops_whitespace_only_assigned_executor() {
|
||||
let (_tmp, config) = temp_config();
|
||||
let mut src = github_source(None);
|
||||
src.assigned_executor = Some(" ".into());
|
||||
let e = enriched("9", None, 0.4);
|
||||
|
||||
add_card(&config, &src, &e, None).expect("add_card succeeds");
|
||||
|
||||
let cards = board_cards(&config).expect("board_cards");
|
||||
assert_eq!(cards.len(), 1);
|
||||
assert!(
|
||||
cards[0].assigned_agent.is_none(),
|
||||
"whitespace-only executor should not assign the card"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn source_metadata_has_no_repo_for_non_github_provider() {
|
||||
let mut src = github_source(Some("octo/repo"));
|
||||
// A non-GitHub filter carries no repo concept.
|
||||
src.provider = ProviderSlug::Linear;
|
||||
src.filter = FilterSpec::Linear {
|
||||
team_id: None,
|
||||
assignee_is_me: true,
|
||||
state: None,
|
||||
extra: json!({}),
|
||||
};
|
||||
let e = enriched("LIN-5", None, 0.5);
|
||||
let meta = build_source_metadata(&src, &e);
|
||||
assert!(meta.get("repo").is_none());
|
||||
assert_eq!(meta["source_id"], json!("ts-1"));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -169,6 +169,13 @@ pub fn schemas(function: &str) -> ControllerSchema {
|
||||
comment: "Per-fetch task cap (u32 range); defaults from config.",
|
||||
required: false,
|
||||
},
|
||||
FieldSchema {
|
||||
name: "assigned_executor",
|
||||
ty: TypeSchema::Option(Box::new(TypeSchema::String)),
|
||||
comment: "Optional executor handle (personality/skill/agent id) every \
|
||||
card from this source is pre-assigned to.",
|
||||
required: false,
|
||||
},
|
||||
],
|
||||
outputs: vec![FieldSchema {
|
||||
name: "source",
|
||||
@@ -369,6 +376,7 @@ fn handle_add(params: Map<String, Value>) -> ControllerFuture {
|
||||
let interval_secs = read_optional::<u64>(¶ms, "interval_secs")?;
|
||||
let target = read_optional::<SourceTarget>(¶ms, "target")?;
|
||||
let max = read_optional_u32(¶ms, "max_tasks_per_fetch")?;
|
||||
let assigned_executor = read_optional::<String>(¶ms, "assigned_executor")?;
|
||||
to_json(
|
||||
ops::add(
|
||||
&config,
|
||||
@@ -379,6 +387,7 @@ fn handle_add(params: Map<String, Value>) -> ControllerFuture {
|
||||
interval_secs,
|
||||
target,
|
||||
max,
|
||||
assigned_executor,
|
||||
)
|
||||
.await?,
|
||||
)
|
||||
|
||||
@@ -26,20 +26,26 @@ use super::types::{
|
||||
};
|
||||
|
||||
/// Compute an edit-aware content hash for a task. Two fetches of the
|
||||
/// same `external_id` whose title/body/status/updated_at differ produce
|
||||
/// same `external_id` whose title/body/status/updated_at/url differ produce
|
||||
/// different hashes, so an *edited* upstream item re-ingests.
|
||||
///
|
||||
/// `url` is part of the canonical form because it is load-bearing downstream
|
||||
/// (it lands in the card's `source_metadata`/notes and drives external
|
||||
/// write-back); a provider that edits the URL without advancing `updated_at`
|
||||
/// would otherwise leave a stale link on the board.
|
||||
///
|
||||
/// Uses SHA-256 over a canonical field-delimited representation so the
|
||||
/// digest is stable across Rust/toolchain versions (the dedup key is
|
||||
/// persisted on disk; a non-deterministic hasher would force spurious
|
||||
/// re-ingests after a toolchain bump).
|
||||
pub fn content_hash(task: &NormalizedTask) -> String {
|
||||
let canonical = format!(
|
||||
"{}\u{1f}{}\u{1f}{}\u{1f}{}",
|
||||
"{}\u{1f}{}\u{1f}{}\u{1f}{}\u{1f}{}",
|
||||
task.title,
|
||||
task.body.as_deref().unwrap_or(""),
|
||||
task.status.as_deref().unwrap_or(""),
|
||||
task.updated_at.as_deref().unwrap_or(""),
|
||||
task.url.as_deref().unwrap_or(""),
|
||||
);
|
||||
let digest = Sha256::digest(canonical.as_bytes());
|
||||
format!("{digest:x}")
|
||||
@@ -166,6 +172,9 @@ pub fn update_source(config: &Config, id: &str, patch: TaskSourcePatch) -> Resul
|
||||
if let Some(connection_id) = patch.connection_id {
|
||||
source.connection_id = Some(connection_id).filter(|s| !s.trim().is_empty());
|
||||
}
|
||||
if let Some(assigned_executor) = patch.assigned_executor {
|
||||
source.assigned_executor = Some(assigned_executor).filter(|s| !s.trim().is_empty());
|
||||
}
|
||||
|
||||
let filter_json = serde_json::to_string(&source.filter).context("serialize filter")?;
|
||||
let target_json = serde_json::to_string(&source.target).context("serialize target")?;
|
||||
@@ -176,8 +185,9 @@ pub fn update_source(config: &Config, id: &str, patch: TaskSourcePatch) -> Resul
|
||||
conn.execute(
|
||||
"UPDATE task_sources
|
||||
SET provider = ?1, connection_id = ?2, name = ?3, enabled = ?4, filter = ?5,
|
||||
interval_secs = ?6, target = ?7, max_tasks_per_fetch = ?8
|
||||
WHERE id = ?9",
|
||||
interval_secs = ?6, target = ?7, max_tasks_per_fetch = ?8,
|
||||
assigned_executor = ?9
|
||||
WHERE id = ?10",
|
||||
params![
|
||||
source.provider.as_str(),
|
||||
source.connection_id,
|
||||
@@ -187,6 +197,7 @@ pub fn update_source(config: &Config, id: &str, patch: TaskSourcePatch) -> Resul
|
||||
interval_i64,
|
||||
target_json,
|
||||
i64::from(source.max_tasks_per_fetch),
|
||||
source.assigned_executor,
|
||||
id,
|
||||
],
|
||||
)
|
||||
@@ -339,7 +350,8 @@ pub fn clear_all(config: &Config) -> Result<usize> {
|
||||
}
|
||||
|
||||
const SELECT_SOURCE_COLUMNS: &str = "SELECT id, provider, connection_id, name, enabled, filter, \
|
||||
interval_secs, target, max_tasks_per_fetch, created_at, last_fetch_at, last_status \
|
||||
interval_secs, target, max_tasks_per_fetch, created_at, last_fetch_at, last_status, \
|
||||
assigned_executor \
|
||||
FROM task_sources";
|
||||
|
||||
fn map_source_row(row: &rusqlite::Row<'_>) -> rusqlite::Result<TaskSource> {
|
||||
@@ -369,6 +381,7 @@ fn map_source_row(row: &rusqlite::Row<'_>) -> rusqlite::Result<TaskSource> {
|
||||
target,
|
||||
max_tasks_per_fetch: u32::try_from(row.get::<_, i64>(8)?)
|
||||
.map_err(|_| sql_conv("invalid max_tasks_per_fetch in task_sources DB"))?,
|
||||
assigned_executor: row.get(12)?,
|
||||
created_at: parse_rfc3339(&created_at_raw).map_err(sql_conv)?,
|
||||
last_fetch_at: match last_fetch_raw {
|
||||
Some(raw) => Some(parse_rfc3339(&raw).map_err(sql_conv)?),
|
||||
@@ -421,7 +434,8 @@ fn with_connection<T>(config: &Config, f: impl FnOnce(&Connection) -> Result<T>)
|
||||
max_tasks_per_fetch INTEGER NOT NULL,
|
||||
created_at TEXT NOT NULL,
|
||||
last_fetch_at TEXT,
|
||||
last_status TEXT
|
||||
last_status TEXT,
|
||||
assigned_executor TEXT
|
||||
);
|
||||
CREATE TABLE IF NOT EXISTS ingested_tasks (
|
||||
source_id TEXT NOT NULL,
|
||||
@@ -441,6 +455,8 @@ fn with_connection<T>(config: &Config, f: impl FnOnce(&Connection) -> Result<T>)
|
||||
// Additive migration: add card_id to existing databases that pre-date
|
||||
// this column. Tolerate "duplicate column" in case of a concurrent open.
|
||||
add_column_if_missing(&conn, "ingested_tasks", "card_id", "TEXT")?;
|
||||
// G7: static executor routing on a source.
|
||||
add_column_if_missing(&conn, "task_sources", "assigned_executor", "TEXT")?;
|
||||
|
||||
f(&conn)
|
||||
}
|
||||
|
||||
@@ -215,6 +215,68 @@ fn dedup_detects_seen_and_edited_tasks() {
|
||||
assert_eq!(listed[0].external_id, "42");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn add_with_assigned_executor_persists_and_filters_blank() {
|
||||
use crate::openhuman::task_sources::ops;
|
||||
|
||||
let tmp = TempDir::new().unwrap();
|
||||
let config = test_config(&tmp);
|
||||
|
||||
// Some(non-empty) → persisted via the follow-up update_source patch
|
||||
// (exercises both ops::add's assigned-executor branch and store's
|
||||
// update_source patch arm). The store layer preserves the value verbatim;
|
||||
// route::add_card is what trims it when stamping a card's assigned_agent.
|
||||
let out = ops::add(
|
||||
&config,
|
||||
ProviderSlug::Github,
|
||||
None,
|
||||
None,
|
||||
github_filter(),
|
||||
Some(1800),
|
||||
Some(SourceTarget::TodoOnly),
|
||||
Some(25),
|
||||
Some("my-skill".into()),
|
||||
)
|
||||
.await
|
||||
.expect("add with executor");
|
||||
assert_eq!(out.value.assigned_executor.as_deref(), Some("my-skill"));
|
||||
|
||||
// Re-read from disk to confirm persistence (not just the returned value).
|
||||
let fetched = get_source(&config, &out.value.id).unwrap();
|
||||
assert_eq!(fetched.assigned_executor.as_deref(), Some("my-skill"));
|
||||
|
||||
// Whitespace-only executor is filtered to None before the patch runs.
|
||||
let blank = ops::add(
|
||||
&config,
|
||||
ProviderSlug::Github,
|
||||
None,
|
||||
None,
|
||||
github_filter(),
|
||||
Some(1800),
|
||||
Some(SourceTarget::TodoOnly),
|
||||
Some(25),
|
||||
Some(" ".into()),
|
||||
)
|
||||
.await
|
||||
.expect("add with blank executor");
|
||||
assert_eq!(blank.value.assigned_executor, None);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn content_hash_changes_when_only_url_changes() {
|
||||
// `url` is load-bearing downstream (source_metadata / external write-back),
|
||||
// so a URL-only upstream edit must produce a different hash and re-ingest —
|
||||
// even if `updated_at` didn't advance (coarse-`updated_at` providers).
|
||||
let base = sample_task("7", "Same title", "2025-01-01T00:00:00Z");
|
||||
let mut moved = base.clone();
|
||||
moved.url = Some("https://example.com/issues/7".into());
|
||||
assert_ne!(
|
||||
content_hash(&base),
|
||||
content_hash(&moved),
|
||||
"a URL-only change must re-ingest"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn list_ingested_orders_newest_first() {
|
||||
let tmp = TempDir::new().unwrap();
|
||||
|
||||
@@ -165,6 +165,12 @@ pub struct TaskSource {
|
||||
pub interval_secs: u64,
|
||||
pub target: SourceTarget,
|
||||
pub max_tasks_per_fetch: u32,
|
||||
/// Static executor routing (G7): a personality / skill / agent handle that
|
||||
/// every card from this source is pre-assigned to, so the dispatcher runs
|
||||
/// it deterministically without the LLM router. `None` leaves cards
|
||||
/// unassigned (router / poller decides).
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub assigned_executor: Option<String>,
|
||||
pub created_at: DateTime<Utc>,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub last_fetch_at: Option<DateTime<Utc>>,
|
||||
@@ -191,6 +197,8 @@ pub struct TaskSourcePatch {
|
||||
pub max_tasks_per_fetch: Option<u32>,
|
||||
#[serde(default)]
|
||||
pub connection_id: Option<String>,
|
||||
#[serde(default)]
|
||||
pub assigned_executor: Option<String>,
|
||||
}
|
||||
|
||||
/// An enriched, agent-ready task produced by [`super::enrich`].
|
||||
@@ -318,6 +326,7 @@ mod tests {
|
||||
interval_secs: 1800,
|
||||
target: SourceTarget::TodoOnly,
|
||||
max_tasks_per_fetch: 25,
|
||||
assigned_executor: None,
|
||||
created_at: DateTime::parse_from_rfc3339("2025-01-01T00:00:00Z")
|
||||
.unwrap()
|
||||
.with_timezone(&Utc),
|
||||
|
||||
@@ -164,6 +164,7 @@ fn task_board_update_is_stored_and_flushed() {
|
||||
evidence: Vec::new(),
|
||||
notes: None,
|
||||
blocker: None,
|
||||
source_metadata: None,
|
||||
order: 0,
|
||||
updated_at: "2026-05-15T00:00:00Z".into(),
|
||||
}],
|
||||
|
||||
+128
-2
@@ -36,11 +36,14 @@ fn maybe_scratch_lock(location: &BoardLocation) -> Option<MutexGuard<'static, ()
|
||||
pub fn parse_status(raw: &str) -> Result<TaskCardStatus, String> {
|
||||
match raw.trim().to_ascii_lowercase().as_str() {
|
||||
"todo" | "pending" => Ok(TaskCardStatus::Todo),
|
||||
"awaiting_approval" | "awaiting-approval" => Ok(TaskCardStatus::AwaitingApproval),
|
||||
"ready" | "approved" => Ok(TaskCardStatus::Ready),
|
||||
"in_progress" | "in-progress" | "inprogress" | "started" => Ok(TaskCardStatus::InProgress),
|
||||
"blocked" => Ok(TaskCardStatus::Blocked),
|
||||
"done" | "completed" | "complete" => Ok(TaskCardStatus::Done),
|
||||
"rejected" | "denied" => Ok(TaskCardStatus::Rejected),
|
||||
other => Err(format!(
|
||||
"invalid status '{other}' (expected todo|in_progress|blocked|done)"
|
||||
"invalid status '{other}' (expected todo|awaiting_approval|ready|in_progress|blocked|done|rejected)"
|
||||
)),
|
||||
}
|
||||
}
|
||||
@@ -68,6 +71,9 @@ pub struct CardPatch {
|
||||
pub evidence: Option<Vec<String>>,
|
||||
pub notes: Option<String>,
|
||||
pub blocker: Option<String>,
|
||||
/// Provider/source identifiers for a task-source-ingested card. `Some`
|
||||
/// sets the card's `source_metadata`; `None` leaves it untouched.
|
||||
pub source_metadata: Option<serde_json::Value>,
|
||||
}
|
||||
|
||||
/// Where to load/save the working set of cards.
|
||||
@@ -158,10 +164,12 @@ pub fn render_markdown(cards: &[TaskBoardCard]) -> String {
|
||||
let mut out = String::new();
|
||||
for card in cards {
|
||||
let marker = match card.status {
|
||||
TaskCardStatus::Todo => "[ ]",
|
||||
TaskCardStatus::Todo | TaskCardStatus::Ready => "[ ]",
|
||||
TaskCardStatus::AwaitingApproval => "[?]",
|
||||
TaskCardStatus::InProgress => "[~]",
|
||||
TaskCardStatus::Blocked => "[!]",
|
||||
TaskCardStatus::Done => "[x]",
|
||||
TaskCardStatus::Rejected => "[-]",
|
||||
};
|
||||
out.push_str("- ");
|
||||
out.push_str(marker);
|
||||
@@ -261,6 +269,7 @@ pub fn add(
|
||||
evidence: patch.evidence.unwrap_or_default(),
|
||||
notes: patch.notes.and_then(non_empty),
|
||||
blocker: patch.blocker.and_then(non_empty),
|
||||
source_metadata: patch.source_metadata,
|
||||
order: cards.len() as u32,
|
||||
updated_at: Utc::now().to_rfc3339(),
|
||||
};
|
||||
@@ -322,6 +331,9 @@ pub fn edit(location: &BoardLocation, id: &str, patch: CardPatch) -> Result<Todo
|
||||
if let Some(blocker) = patch.blocker {
|
||||
card.blocker = non_empty(blocker);
|
||||
}
|
||||
if let Some(source_metadata) = patch.source_metadata {
|
||||
card.source_metadata = Some(source_metadata);
|
||||
}
|
||||
card.updated_at = Utc::now().to_rfc3339();
|
||||
enforce_single_in_progress(&cards)?;
|
||||
let cards = save_cards(location, cards)?;
|
||||
@@ -345,6 +357,34 @@ pub fn update_status(
|
||||
)
|
||||
}
|
||||
|
||||
/// Resolve a plan-approval decision: approve (→`Ready`, so the dispatcher runs
|
||||
/// it) or reject (→`Rejected`). Errors unless the card is currently
|
||||
/// `AwaitingApproval`, so a stale/duplicate decision can't resurrect a card
|
||||
/// that already moved on.
|
||||
pub fn decide_plan(
|
||||
location: &BoardLocation,
|
||||
id: &str,
|
||||
approve: bool,
|
||||
) -> Result<TodosSnapshot, String> {
|
||||
let cards = load_cards(location)?;
|
||||
let current = cards
|
||||
.iter()
|
||||
.find(|c| c.id == id)
|
||||
.ok_or_else(|| format!("todo id '{id}' not found"))?;
|
||||
if current.status != TaskCardStatus::AwaitingApproval {
|
||||
return Err(format!(
|
||||
"card '{id}' is not awaiting approval (status: {})",
|
||||
current.status.as_str()
|
||||
));
|
||||
}
|
||||
let new_status = if approve {
|
||||
TaskCardStatus::Ready
|
||||
} else {
|
||||
TaskCardStatus::Rejected
|
||||
};
|
||||
update_status(location, id, new_status)
|
||||
}
|
||||
|
||||
/// Remove a card by id. Errors if `id` is unknown.
|
||||
pub fn remove(location: &BoardLocation, id: &str) -> Result<TodosSnapshot, String> {
|
||||
tracing::debug!(
|
||||
@@ -478,6 +518,13 @@ mod tests {
|
||||
);
|
||||
assert_eq!(parse_status("blocked").unwrap(), TaskCardStatus::Blocked);
|
||||
assert_eq!(parse_status("done").unwrap(), TaskCardStatus::Done);
|
||||
assert_eq!(
|
||||
parse_status("awaiting_approval").unwrap(),
|
||||
TaskCardStatus::AwaitingApproval
|
||||
);
|
||||
assert_eq!(parse_status("ready").unwrap(), TaskCardStatus::Ready);
|
||||
assert_eq!(parse_status("approved").unwrap(), TaskCardStatus::Ready);
|
||||
assert_eq!(parse_status("rejected").unwrap(), TaskCardStatus::Rejected);
|
||||
assert!(parse_status("nope").is_err());
|
||||
}
|
||||
|
||||
@@ -533,6 +580,83 @@ mod tests {
|
||||
assert_eq!(snap.cards[0].title, "Refined plan");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn source_metadata_round_trips_through_add_and_edit() {
|
||||
let dir = tempdir().unwrap();
|
||||
let loc = thread_loc(dir.path(), "t1");
|
||||
let added = add(
|
||||
&loc,
|
||||
"ingested task",
|
||||
CardPatch {
|
||||
source_metadata: Some(serde_json::json!({
|
||||
"provider": "github",
|
||||
"external_id": "7",
|
||||
})),
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.unwrap();
|
||||
let id = added.cards[0].id.clone();
|
||||
assert_eq!(
|
||||
added.cards[0].source_metadata.as_ref().unwrap()["external_id"],
|
||||
serde_json::json!("7")
|
||||
);
|
||||
|
||||
// A subsequent edit with `Some(..)` replaces the stamped metadata.
|
||||
let snap = edit(
|
||||
&loc,
|
||||
&id,
|
||||
CardPatch {
|
||||
source_metadata: Some(serde_json::json!({
|
||||
"provider": "github",
|
||||
"external_id": "8",
|
||||
})),
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.unwrap();
|
||||
assert_eq!(
|
||||
snap.cards[0].source_metadata.as_ref().unwrap()["external_id"],
|
||||
serde_json::json!("8")
|
||||
);
|
||||
|
||||
// An edit that leaves `source_metadata: None` preserves the value.
|
||||
let snap2 = edit(
|
||||
&loc,
|
||||
&id,
|
||||
CardPatch {
|
||||
notes: Some("touch".into()),
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.unwrap();
|
||||
assert_eq!(
|
||||
snap2.cards[0].source_metadata.as_ref().unwrap()["external_id"],
|
||||
serde_json::json!("8")
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn decide_plan_approves_and_rejects_only_when_awaiting() {
|
||||
let dir = tempdir().unwrap();
|
||||
let loc = thread_loc(dir.path(), "t1");
|
||||
let added = add(&loc, "task", CardPatch::default()).unwrap();
|
||||
let id = added.cards[0].id.clone();
|
||||
|
||||
// A todo card isn't awaiting approval yet → decision rejected.
|
||||
assert!(decide_plan(&loc, &id, true).is_err());
|
||||
|
||||
// Park it, then approve → Ready.
|
||||
update_status(&loc, &id, TaskCardStatus::AwaitingApproval).unwrap();
|
||||
let approved = decide_plan(&loc, &id, true).unwrap();
|
||||
assert_eq!(approved.cards[0].status, TaskCardStatus::Ready);
|
||||
|
||||
// Re-park, then reject → Rejected.
|
||||
update_status(&loc, &id, TaskCardStatus::AwaitingApproval).unwrap();
|
||||
let rejected = decide_plan(&loc, &id, false).unwrap();
|
||||
assert_eq!(rejected.cards[0].status, TaskCardStatus::Rejected);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn edit_can_clear_approval_mode() {
|
||||
let dir = tempdir().unwrap();
|
||||
@@ -609,6 +733,7 @@ mod tests {
|
||||
evidence: Vec::new(),
|
||||
notes: None,
|
||||
blocker: None,
|
||||
source_metadata: None,
|
||||
order: 0,
|
||||
updated_at: String::new(),
|
||||
},
|
||||
@@ -625,6 +750,7 @@ mod tests {
|
||||
evidence: Vec::new(),
|
||||
notes: None,
|
||||
blocker: None,
|
||||
source_metadata: None,
|
||||
order: 1,
|
||||
updated_at: String::new(),
|
||||
},
|
||||
|
||||
@@ -19,6 +19,7 @@ pub fn all_controller_schemas() -> Vec<ControllerSchema> {
|
||||
schemas("add"),
|
||||
schemas("edit"),
|
||||
schemas("update_status"),
|
||||
schemas("decide_plan"),
|
||||
schemas("remove"),
|
||||
schemas("replace"),
|
||||
schemas("clear"),
|
||||
@@ -43,6 +44,10 @@ pub fn all_registered_controllers() -> Vec<RegisteredController> {
|
||||
schema: schemas("update_status"),
|
||||
handler: handle_update_status,
|
||||
},
|
||||
RegisteredController {
|
||||
schema: schemas("decide_plan"),
|
||||
handler: handle_decide_plan,
|
||||
},
|
||||
RegisteredController {
|
||||
schema: schemas("remove"),
|
||||
handler: handle_remove,
|
||||
@@ -137,7 +142,27 @@ pub fn schemas(function: &str) -> ControllerSchema {
|
||||
inputs: vec![
|
||||
thread_id_input(),
|
||||
required_string("id", "Card identifier."),
|
||||
required_string("status", "New status (todo|in_progress|blocked|done)."),
|
||||
required_string(
|
||||
"status",
|
||||
"New status (todo|awaiting_approval|ready|in_progress|blocked|done|rejected).",
|
||||
),
|
||||
],
|
||||
outputs: vec![snapshot_output()],
|
||||
},
|
||||
"decide_plan" => ControllerSchema {
|
||||
namespace: "todos",
|
||||
function: "decide_plan",
|
||||
description: "Approve or reject a card awaiting plan approval \
|
||||
(approve → ready/runnable; reject → rejected).",
|
||||
inputs: vec![
|
||||
thread_id_input(),
|
||||
required_string("id", "Card identifier."),
|
||||
FieldSchema {
|
||||
name: "approve",
|
||||
ty: TypeSchema::Bool,
|
||||
comment: "true to approve (card becomes runnable), false to reject.",
|
||||
required: true,
|
||||
},
|
||||
],
|
||||
outputs: vec![snapshot_output()],
|
||||
},
|
||||
@@ -260,6 +285,13 @@ struct RemoveParams {
|
||||
id: String,
|
||||
}
|
||||
|
||||
#[derive(Debug, Deserialize)]
|
||||
struct DecidePlanParams {
|
||||
thread_id: String,
|
||||
id: String,
|
||||
approve: bool,
|
||||
}
|
||||
|
||||
#[derive(Debug, Deserialize)]
|
||||
struct ReplaceParams {
|
||||
thread_id: String,
|
||||
@@ -291,6 +323,7 @@ fn handle_add(params: Map<String, Value>) -> ControllerFuture {
|
||||
evidence: p.evidence,
|
||||
notes: p.notes,
|
||||
blocker: p.blocker,
|
||||
source_metadata: None,
|
||||
};
|
||||
tracing::debug!(thread_id = %p.thread_id, "[rpc][todos] add entry");
|
||||
snapshot_to_json(ops::add(&loc, &p.content, patch)?)
|
||||
@@ -314,6 +347,7 @@ fn handle_edit(params: Map<String, Value>) -> ControllerFuture {
|
||||
evidence: p.evidence,
|
||||
notes: p.notes,
|
||||
blocker: p.blocker,
|
||||
source_metadata: None,
|
||||
};
|
||||
tracing::debug!(thread_id = %p.thread_id, id = %p.id, "[rpc][todos] edit entry");
|
||||
snapshot_to_json(ops::edit(&loc, &p.id, patch)?)
|
||||
@@ -335,6 +369,20 @@ fn handle_update_status(params: Map<String, Value>) -> ControllerFuture {
|
||||
})
|
||||
}
|
||||
|
||||
fn handle_decide_plan(params: Map<String, Value>) -> ControllerFuture {
|
||||
Box::pin(async move {
|
||||
let p = parse::<DecidePlanParams>(params)?;
|
||||
let loc = thread_location(&p.thread_id).await?;
|
||||
tracing::debug!(
|
||||
thread_id = %p.thread_id,
|
||||
id = %p.id,
|
||||
approve = p.approve,
|
||||
"[rpc][todos] decide_plan entry"
|
||||
);
|
||||
snapshot_to_json(ops::decide_plan(&loc, &p.id, p.approve)?)
|
||||
})
|
||||
}
|
||||
|
||||
fn parse_approval_mode(raw: Option<String>) -> Result<Option<TaskApprovalMode>, String> {
|
||||
let Some(raw) = raw else {
|
||||
return Ok(None);
|
||||
|
||||
Reference in New Issue
Block a user