mirror of
https://github.com/tinyhumansai/openhuman.git
synced 2026-07-27 20:46:24 +00:00
feat(task-board): autonomous task runs as live task sessions (View work, cancel, degeneration guards) (#3380)
Co-authored-by: Claude <noreply@anthropic.com>
This commit is contained in:
@@ -28,6 +28,7 @@ import { useNavigate } from 'react-router-dom';
|
||||
|
||||
import { useT } from '../../lib/i18n/I18nContext';
|
||||
import { TaskKanbanBoard } from '../../pages/conversations/components/TaskKanbanBoard';
|
||||
import { isTaskThread } from '../../pages/conversations/utils/threadFilter';
|
||||
import type { AgentDefinitionDisplay } from '../../services/api/agentLibraryApi';
|
||||
import { threadApi } from '../../services/api/threadApi';
|
||||
import {
|
||||
@@ -379,7 +380,11 @@ export default function IntelligenceTasksTab() {
|
||||
card.id,
|
||||
'in_progress'
|
||||
);
|
||||
if (mountedRef.current) setPersonalBoard(startedBoard);
|
||||
// Link the card to its session thread so the board offers "View session".
|
||||
const linkedBoard = await todosApi
|
||||
.setSessionThread(USER_TASKS_THREAD_ID, card.id, thread.id)
|
||||
.catch(() => startedBoard);
|
||||
if (mountedRef.current) setPersonalBoard(linkedBoard);
|
||||
|
||||
dispatch(setSelectedThread(thread.id));
|
||||
dispatch(setToolTimelineForThread({ threadId: thread.id, entries: [] }));
|
||||
@@ -459,12 +464,36 @@ export default function IntelligenceTasksTab() {
|
||||
const now = new Date().toISOString();
|
||||
setActionError(null);
|
||||
try {
|
||||
const sourceExternalId = readSourceMetadata(sourceCard.sourceMetadata).externalId;
|
||||
// Idempotent approve: if this source item was already promoted to
|
||||
// user-tasks (picked up / running), don't add a second card — just
|
||||
// retire the inbox card. Stops the duplicate when an edited item is
|
||||
// re-offered for approval.
|
||||
const alreadyPromoted = sourceExternalId
|
||||
? personalBoard?.cards.some(
|
||||
c => readSourceMetadata(c.sourceMetadata).externalId === sourceExternalId
|
||||
)
|
||||
: false;
|
||||
if (alreadyPromoted) {
|
||||
const sourceSaved = await todosApi.updateStatus(
|
||||
TASK_SOURCES_THREAD_ID,
|
||||
sourceCard.id,
|
||||
'done'
|
||||
);
|
||||
if (mountedRef.current) {
|
||||
setTaskSourcesBoard(sourceSaved);
|
||||
setRefiningCard(null);
|
||||
}
|
||||
return;
|
||||
}
|
||||
const added = await todosApi.add({
|
||||
threadId: USER_TASKS_THREAD_ID,
|
||||
content: draft.title,
|
||||
status: 'todo',
|
||||
objective: draft.objective,
|
||||
notes: draft.notes,
|
||||
// Stamp the source link so the inbox can detect it's now picked up.
|
||||
sourceMetadata: sourceCard.sourceMetadata,
|
||||
});
|
||||
const created =
|
||||
added.cards.find(card => card.title === draft.title && card.updatedAt >= now) ??
|
||||
@@ -504,7 +533,7 @@ export default function IntelligenceTasksTab() {
|
||||
if (mountedRef.current) setActionError(t('intelligence.tasks.sourcePlan.createFailed'));
|
||||
}
|
||||
},
|
||||
[t]
|
||||
[t, personalBoard]
|
||||
);
|
||||
|
||||
// ── derived agent board list (read-only) ─────────────────────────────
|
||||
@@ -512,16 +541,36 @@ export default function IntelligenceTasksTab() {
|
||||
const threadMap = new Map(threads.map(th => [th.id, th]));
|
||||
const allThreadIds = new Set([...Object.keys(liveBoards), ...Object.keys(persistedBoards)]);
|
||||
|
||||
// Hide task-source items already picked up (promoted to the user-tasks board)
|
||||
// so an already-running task isn't re-offered for approval in the inbox.
|
||||
const pickedUpExternalIds = new Set(
|
||||
(personalBoard?.cards ?? [])
|
||||
.map(c => readSourceMetadata(c.sourceMetadata).externalId)
|
||||
.filter((id): id is string => Boolean(id))
|
||||
);
|
||||
const visibleTaskSourcesBoard: TaskBoard | null = taskSourcesBoard
|
||||
? {
|
||||
...taskSourcesBoard,
|
||||
cards: taskSourcesBoard.cards.filter(
|
||||
c => !pickedUpExternalIds.has(readSourceMetadata(c.sourceMetadata).externalId ?? '')
|
||||
),
|
||||
}
|
||||
: null;
|
||||
|
||||
const boardEntries: ThreadTaskBoard[] = [];
|
||||
for (const threadId of allThreadIds) {
|
||||
if (threadId === USER_TASKS_THREAD_ID) continue; // personal board rendered separately
|
||||
if (threadId === TASK_SOURCES_THREAD_ID) continue; // task sources rendered separately
|
||||
const thread = threadMap.get(threadId);
|
||||
// Skip task SESSION threads (autonomous `task-*` runs + manual task-labelled
|
||||
// threads): their cards already appear on the user-tasks / task-sources
|
||||
// boards and the session is reachable via the card's "View work" — rendering
|
||||
// their live board here just duplicates those cards as redundant tables.
|
||||
if (threadId.startsWith('task-') || (thread && isTaskThread(thread))) continue;
|
||||
const liveBoard = liveBoards[threadId];
|
||||
const persistedBoard = persistedBoards[threadId];
|
||||
const board = liveBoard ?? persistedBoard;
|
||||
if (!board || board.cards.length === 0) continue;
|
||||
|
||||
const thread = threadMap.get(threadId);
|
||||
const title =
|
||||
thread?.title && thread.title.trim().length > 0
|
||||
? thread.title
|
||||
@@ -575,6 +624,24 @@ export default function IntelligenceTasksTab() {
|
||||
onUpdateCard={handleUpdatePersonal}
|
||||
onDeleteCard={handleDeletePersonal}
|
||||
onWorkTask={handleWorkPersonal}
|
||||
onViewSession={card => {
|
||||
if (!card.sessionThreadId) return;
|
||||
const tid = card.sessionThreadId;
|
||||
// Open the exact session — mirror the manual "Work" path's
|
||||
// thread-open sequence so /chat lands on this thread, not just
|
||||
// the Conversations page.
|
||||
// Navigation only — do NOT mark the thread active. activeThreadId
|
||||
// tracks a true in-flight turn; a completed session never emits the
|
||||
// done/error lifecycle that would clear it, so forcing it active
|
||||
// would wedge the composer until then.
|
||||
dispatch(setSelectedThread(tid));
|
||||
void dispatch(loadThreads());
|
||||
void dispatch(loadThreadMessages(tid));
|
||||
// Pass the thread as an explicit open-intent so Conversations'
|
||||
// mount-resume honors it (its default resume only considers
|
||||
// General-tab threads and would otherwise drop this task session).
|
||||
navigate('/chat', { state: { openThreadId: tid } });
|
||||
}}
|
||||
workingCardId={workingCardId}
|
||||
/>
|
||||
) : (
|
||||
@@ -594,7 +661,7 @@ export default function IntelligenceTasksTab() {
|
||||
{taskSourcesBoard && (
|
||||
<section className="space-y-2">
|
||||
<TaskSourceTaskList
|
||||
board={taskSourcesBoard}
|
||||
board={visibleTaskSourcesBoard ?? taskSourcesBoard}
|
||||
disabled={loading}
|
||||
onWorkOnTask={setRefiningCard}
|
||||
/>
|
||||
@@ -773,6 +840,7 @@ function TaskSourceTaskList({
|
||||
onWorkOnTask: (card: TaskBoardCard) => void;
|
||||
}) {
|
||||
const { t } = useT();
|
||||
const navigate = useNavigate();
|
||||
const sortedCards = useMemo(
|
||||
() => [...board.cards].sort((a, b) => a.order - b.order),
|
||||
[board.cards]
|
||||
@@ -789,11 +857,12 @@ function TaskSourceTaskList({
|
||||
{t('intelligence.tasks.sourceList.subtitle')}
|
||||
</p>
|
||||
</div>
|
||||
<a
|
||||
href="#/settings/task-sources"
|
||||
<button
|
||||
type="button"
|
||||
onClick={() => navigate('/settings/task-sources')}
|
||||
className="text-xs font-medium text-ocean-600 hover:text-ocean-700 dark:text-ocean-300 dark:hover:text-ocean-200">
|
||||
{t('conversations.taskKanban.sources.manage')}
|
||||
</a>
|
||||
</button>
|
||||
</div>
|
||||
|
||||
{sortedCards.length === 0 ? (
|
||||
|
||||
@@ -25,7 +25,9 @@ const hoisted = vi.hoisted(() => ({
|
||||
todosAdd: vi.fn(),
|
||||
todosEdit: vi.fn(),
|
||||
todosUpdateStatus: vi.fn(),
|
||||
todosSetSessionThread: vi.fn(),
|
||||
todosRemove: vi.fn(),
|
||||
navigate: vi.fn(),
|
||||
selectorResult: {
|
||||
chatRuntime: { taskBoardByThread: {} as Record<string, unknown> },
|
||||
thread: { threads: [] as unknown[] },
|
||||
@@ -57,6 +59,7 @@ vi.mock('../../../services/api/todosApi', () => ({
|
||||
add: hoisted.todosAdd,
|
||||
edit: hoisted.todosEdit,
|
||||
updateStatus: hoisted.todosUpdateStatus,
|
||||
setSessionThread: hoisted.todosSetSessionThread,
|
||||
remove: hoisted.todosRemove,
|
||||
},
|
||||
}));
|
||||
@@ -69,7 +72,7 @@ vi.mock('../../../store/hooks', () => ({
|
||||
|
||||
vi.mock('react-router-dom', async () => {
|
||||
const actual = await vi.importActual<typeof import('react-router-dom')>('react-router-dom');
|
||||
return { ...actual, useNavigate: () => vi.fn() };
|
||||
return { ...actual, useNavigate: () => hoisted.navigate };
|
||||
});
|
||||
|
||||
// Stub the composer so we can drive its `onCreated` callback without
|
||||
@@ -109,12 +112,17 @@ vi.mock('../../../pages/conversations/components/TaskKanbanBoard', () => ({
|
||||
onMove,
|
||||
onDeleteCard,
|
||||
onWorkTask,
|
||||
onViewSession,
|
||||
}: {
|
||||
board: { threadId: string; cards: { id: string; title: string; status: string }[] };
|
||||
board: {
|
||||
threadId: string;
|
||||
cards: { id: string; title: string; status: string; sessionThreadId?: string }[];
|
||||
};
|
||||
headerTitleKey?: string;
|
||||
onMove?: (card: unknown, status: string) => void;
|
||||
onDeleteCard?: (card: unknown) => void;
|
||||
onWorkTask?: (card: unknown) => void;
|
||||
onViewSession?: (card: unknown) => void;
|
||||
}) => (
|
||||
<div data-testid="kanban-stub">
|
||||
<span>{board.threadId}</span>
|
||||
@@ -137,6 +145,11 @@ vi.mock('../../../pages/conversations/components/TaskKanbanBoard', () => ({
|
||||
stub-work-task
|
||||
</button>
|
||||
)}
|
||||
{onViewSession && (
|
||||
<button type="button" onClick={() => onViewSession(board.cards[0])}>
|
||||
stub-view-session
|
||||
</button>
|
||||
)}
|
||||
</div>
|
||||
),
|
||||
}));
|
||||
@@ -178,7 +191,9 @@ describe('IntelligenceTasksTab', () => {
|
||||
hoisted.todosAdd.mockReset();
|
||||
hoisted.todosEdit.mockReset();
|
||||
hoisted.todosUpdateStatus.mockReset();
|
||||
hoisted.todosSetSessionThread.mockReset();
|
||||
hoisted.todosRemove.mockReset();
|
||||
hoisted.navigate.mockReset();
|
||||
hoisted.selectorResult.chatRuntime.taskBoardByThread = {};
|
||||
hoisted.selectorResult.thread.threads = [];
|
||||
hoisted.selectorResult.agentProfiles.activeProfileId = 'agent-profile-1';
|
||||
@@ -214,6 +229,11 @@ describe('IntelligenceTasksTab', () => {
|
||||
createdAt: '2026-01-01T00:00:00Z',
|
||||
});
|
||||
hoisted.chatSend.mockResolvedValue(undefined);
|
||||
// The "Work" flow links the card to its session thread after starting it
|
||||
// (`todosApi.setSessionThread(...).catch(...)`), so resolve it to a board
|
||||
// by default — otherwise the awaited `.catch()` chain stalls the handler
|
||||
// before it reaches chatSend.
|
||||
hoisted.todosSetSessionThread.mockResolvedValue(makeBoard('user-tasks', []));
|
||||
hoisted.listAgentDefinitions.mockResolvedValue([
|
||||
{
|
||||
id: 'researcher',
|
||||
@@ -276,6 +296,10 @@ describe('IntelligenceTasksTab', () => {
|
||||
});
|
||||
expect(screen.getByText('No source tasks waiting.')).toBeInTheDocument();
|
||||
expect(hoisted.todosList).toHaveBeenCalledWith('task-sources');
|
||||
|
||||
// "Manage sources" jumps to the dedicated settings page.
|
||||
fireEvent.click(screen.getByText('Manage sources'));
|
||||
expect(hoisted.navigate).toHaveBeenCalledWith('/settings/task-sources');
|
||||
});
|
||||
|
||||
test('refines a source task and approves it into the personal agent board', async () => {
|
||||
@@ -368,6 +392,74 @@ describe('IntelligenceTasksTab', () => {
|
||||
expect(hoisted.todosUpdateStatus).toHaveBeenCalledWith('task-sources', 'source-1', 'done');
|
||||
});
|
||||
|
||||
test('work flow still completes when linking the session thread fails', async () => {
|
||||
hoisted.todosList.mockImplementation((threadId: string) =>
|
||||
Promise.resolve(
|
||||
threadId === 'user-tasks'
|
||||
? {
|
||||
threadId,
|
||||
cards: [
|
||||
{
|
||||
id: 'personal-1',
|
||||
title: 'Linkless task',
|
||||
status: 'todo',
|
||||
order: 0,
|
||||
updatedAt: '2026-01-01T00:00:00Z',
|
||||
},
|
||||
],
|
||||
updatedAt: '2026-01-01T00:00:00Z',
|
||||
}
|
||||
: makeBoard(threadId, [])
|
||||
)
|
||||
);
|
||||
// The session-thread link rejects; the work flow must fall back and still
|
||||
// dispatch the agent turn.
|
||||
hoisted.todosSetSessionThread.mockRejectedValue(new Error('link offline'));
|
||||
|
||||
vi.resetModules();
|
||||
const Tab = await importTab();
|
||||
renderTab(Tab);
|
||||
|
||||
await waitFor(() => expect(screen.getByText('Linkless task')).toBeInTheDocument());
|
||||
fireEvent.click(screen.getByText('stub-work-task'));
|
||||
|
||||
await waitFor(() => expect(hoisted.chatSend).toHaveBeenCalledTimes(1));
|
||||
});
|
||||
|
||||
test('View work on a personal card opens its exact session thread', async () => {
|
||||
hoisted.todosList.mockImplementation((threadId: string) =>
|
||||
Promise.resolve(
|
||||
threadId === 'user-tasks'
|
||||
? {
|
||||
threadId,
|
||||
cards: [
|
||||
{
|
||||
id: 'personal-1',
|
||||
title: 'Worked card',
|
||||
status: 'in_progress',
|
||||
sessionThreadId: 'task-session-99',
|
||||
order: 0,
|
||||
updatedAt: '2026-01-01T00:00:00Z',
|
||||
},
|
||||
],
|
||||
updatedAt: '2026-01-01T00:00:00Z',
|
||||
}
|
||||
: makeBoard(threadId, [])
|
||||
)
|
||||
);
|
||||
|
||||
vi.resetModules();
|
||||
const Tab = await importTab();
|
||||
renderTab(Tab);
|
||||
|
||||
await waitFor(() => expect(screen.getByText('Worked card')).toBeInTheDocument());
|
||||
fireEvent.click(screen.getByText('stub-view-session'));
|
||||
|
||||
expect(hoisted.navigate).toHaveBeenCalledWith('/chat', {
|
||||
state: { openThreadId: 'task-session-99' },
|
||||
});
|
||||
});
|
||||
|
||||
test('renders persisted agent boards from the turn-state list', async () => {
|
||||
hoisted.listTurnStates.mockResolvedValue([
|
||||
{ threadId: 'thread-x', taskBoard: makeBoard('thread-x', ['Write docs', 'Fix bug']) },
|
||||
@@ -491,14 +583,18 @@ describe('IntelligenceTasksTab', () => {
|
||||
'personal-1',
|
||||
'in_progress'
|
||||
);
|
||||
expect(hoisted.chatSend).toHaveBeenCalledWith(
|
||||
expect.objectContaining({
|
||||
threadId: 'thread-agent-task',
|
||||
message: expect.stringContaining('Acceptance criteria:'),
|
||||
model: 'reasoning-v1',
|
||||
profileId: 'agent-profile-1',
|
||||
locale: 'en',
|
||||
})
|
||||
// chatSend is the last call in the work flow, after an extra `await`
|
||||
// (session-thread link), so wait for it rather than asserting synchronously.
|
||||
await waitFor(() =>
|
||||
expect(hoisted.chatSend).toHaveBeenCalledWith(
|
||||
expect.objectContaining({
|
||||
threadId: 'thread-agent-task',
|
||||
message: expect.stringContaining('Acceptance criteria:'),
|
||||
model: 'reasoning-v1',
|
||||
profileId: 'agent-profile-1',
|
||||
locale: 'en',
|
||||
})
|
||||
)
|
||||
);
|
||||
});
|
||||
|
||||
|
||||
@@ -45,6 +45,7 @@ export type SettingsRoute =
|
||||
| 'webhooks-triggers'
|
||||
| 'composio-triggers'
|
||||
| 'composio-routing'
|
||||
| 'task-sources'
|
||||
| 'mcp-server'
|
||||
| 'dev-workflow'
|
||||
| 'sandbox-settings'
|
||||
@@ -115,6 +116,7 @@ export const useSettingsNavigation = (): SettingsNavigationHook => {
|
||||
if (path.includes('/settings/webhooks-triggers')) return 'webhooks-triggers';
|
||||
if (path.includes('/settings/composio-triggers')) return 'composio-triggers';
|
||||
if (path.includes('/settings/composio-routing')) return 'composio-routing';
|
||||
if (path.includes('/settings/task-sources')) return 'task-sources';
|
||||
if (path.includes('/settings/intelligence')) return 'intelligence';
|
||||
if (path.includes('/settings/crypto')) return 'crypto';
|
||||
if (path.includes('/settings/recovery-phrase')) return 'recovery-phrase';
|
||||
|
||||
@@ -2503,6 +2503,7 @@ const messages: TranslationMap = {
|
||||
'conversations.taskKanban.saveChanges': 'التغييرات',
|
||||
'conversations.taskKanban.deleteCard': 'Delete',
|
||||
'conversations.taskKanban.workTask': 'العمل على المهمة',
|
||||
'conversations.taskKanban.viewWork': 'عرض العمل',
|
||||
'conversations.taskKanban.startingTask': 'جارٍ البدء…',
|
||||
'conversations.taskKanban.updateFailed': 'ولم يتمكن من استكمال المهمة؛ ولم يتم توفير التغييرات.',
|
||||
'conversations.taskKanban.sourcesButton': 'المصادر',
|
||||
|
||||
@@ -2551,6 +2551,7 @@ const messages: TranslationMap = {
|
||||
'conversations.taskKanban.saveChanges': 'পরিবর্তন সংরক্ষণ করা হবে',
|
||||
'conversations.taskKanban.deleteCard': 'Delete',
|
||||
'conversations.taskKanban.workTask': 'কাজ শুরু করুন',
|
||||
'conversations.taskKanban.viewWork': 'কাজ দেখুন',
|
||||
'conversations.taskKanban.startingTask': 'শুরু হচ্ছে...',
|
||||
'conversations.taskKanban.updateFailed': 'কাজটি সংরক্ষণ করা যায়নি।',
|
||||
'conversations.taskKanban.sourcesButton': 'উৎস',
|
||||
|
||||
@@ -2615,6 +2615,7 @@ const messages: TranslationMap = {
|
||||
'conversations.taskKanban.saveChanges': 'Änderungen speichern',
|
||||
'conversations.taskKanban.deleteCard': 'Löschen',
|
||||
'conversations.taskKanban.workTask': 'Aufgabe bearbeiten',
|
||||
'conversations.taskKanban.viewWork': 'Arbeit anzeigen',
|
||||
'conversations.taskKanban.startingTask': 'Startet…',
|
||||
'conversations.taskKanban.updateFailed':
|
||||
'Aufgabe konnte nicht aktualisiert werden; Änderungen wurden nicht gespeichert.',
|
||||
|
||||
@@ -2960,6 +2960,7 @@ const en: TranslationMap = {
|
||||
'conversations.taskKanban.saveChanges': 'Save changes',
|
||||
'conversations.taskKanban.deleteCard': 'Delete',
|
||||
'conversations.taskKanban.workTask': 'Work task',
|
||||
'conversations.taskKanban.viewWork': 'View work',
|
||||
'conversations.taskKanban.startingTask': 'Starting…',
|
||||
'conversations.taskKanban.updateFailed': 'Could not update task; changes were not saved.',
|
||||
'conversations.taskKanban.sourcesButton': 'Sources',
|
||||
|
||||
@@ -2599,6 +2599,7 @@ const messages: TranslationMap = {
|
||||
'conversations.taskKanban.saveChanges': 'Guardar cambios',
|
||||
'conversations.taskKanban.deleteCard': 'Borrar',
|
||||
'conversations.taskKanban.workTask': 'Trabajar tarea',
|
||||
'conversations.taskKanban.viewWork': 'Ver trabajo',
|
||||
'conversations.taskKanban.startingTask': 'Iniciando…',
|
||||
'conversations.taskKanban.updateFailed':
|
||||
'No se pudo actualizar la tarea; los cambios no se guardaron.',
|
||||
|
||||
@@ -2607,6 +2607,7 @@ const messages: TranslationMap = {
|
||||
'conversations.taskKanban.saveChanges': 'Enregistrer les modifications',
|
||||
'conversations.taskKanban.deleteCard': 'Supprimer',
|
||||
'conversations.taskKanban.workTask': 'Travailler la tâche',
|
||||
'conversations.taskKanban.viewWork': 'Voir le travail',
|
||||
'conversations.taskKanban.startingTask': 'Démarrage…',
|
||||
'conversations.taskKanban.updateFailed':
|
||||
"Impossible de mettre à jour la tâche; les modifications n'ont pas été enregistrées.",
|
||||
|
||||
@@ -2555,6 +2555,7 @@ const messages: TranslationMap = {
|
||||
'conversations.taskKanban.saveChanges': 'परिवर्तन सहेजें',
|
||||
'conversations.taskKanban.deleteCard': 'Delete',
|
||||
'conversations.taskKanban.workTask': 'कार्य शुरू करें',
|
||||
'conversations.taskKanban.viewWork': 'कार्य देखें',
|
||||
'conversations.taskKanban.startingTask': 'शुरू हो रहा है…',
|
||||
'conversations.taskKanban.updateFailed': 'कार्य अद्यतन नहीं कर सका; परिवर्तन बचाया नहीं गया।',
|
||||
'conversations.taskKanban.sourcesButton': 'स्रोत',
|
||||
|
||||
@@ -2557,6 +2557,7 @@ const messages: TranslationMap = {
|
||||
'conversations.taskKanban.saveChanges': 'Simpan perubahan',
|
||||
'conversations.taskKanban.deleteCard': 'Hapus',
|
||||
'conversations.taskKanban.workTask': 'Kerjakan tugas',
|
||||
'conversations.taskKanban.viewWork': 'Lihat pekerjaan',
|
||||
'conversations.taskKanban.startingTask': 'Memulai…',
|
||||
'conversations.taskKanban.updateFailed': 'Tak bisa memutakhirkan tugas; perubahan tak disimpan.',
|
||||
'conversations.taskKanban.sourcesButton': 'Sumber',
|
||||
|
||||
@@ -2589,6 +2589,7 @@ const messages: TranslationMap = {
|
||||
'conversations.taskKanban.saveChanges': 'Salva modifiche',
|
||||
'conversations.taskKanban.deleteCard': 'Elimina',
|
||||
'conversations.taskKanban.workTask': "Lavora sull'attività",
|
||||
'conversations.taskKanban.viewWork': 'Visualizza lavoro',
|
||||
'conversations.taskKanban.startingTask': 'Avvio…',
|
||||
'conversations.taskKanban.updateFailed':
|
||||
"Impossibile aggiornare l'attività; le modifiche non sono state salvate.",
|
||||
|
||||
@@ -2529,6 +2529,7 @@ const messages: TranslationMap = {
|
||||
'conversations.taskKanban.saveChanges': '변경 사항 저장',
|
||||
'conversations.taskKanban.deleteCard': '삭제',
|
||||
'conversations.taskKanban.workTask': '작업 시작',
|
||||
'conversations.taskKanban.viewWork': '작업 보기',
|
||||
'conversations.taskKanban.startingTask': '시작 중…',
|
||||
'conversations.taskKanban.updateFailed':
|
||||
'작업을 업데이트할 수 없어 변경 사항이 저장되지 않았습니다.',
|
||||
|
||||
@@ -2584,6 +2584,7 @@ const messages: TranslationMap = {
|
||||
'conversations.taskKanban.saveChanges': 'Zapisz zmiany',
|
||||
'conversations.taskKanban.deleteCard': 'Usuń',
|
||||
'conversations.taskKanban.workTask': 'Pracuj nad zadaniem',
|
||||
'conversations.taskKanban.viewWork': 'Zobacz pracę',
|
||||
'conversations.taskKanban.startingTask': 'Uruchamianie…',
|
||||
'conversations.taskKanban.updateFailed':
|
||||
'Nie udało się zaktualizować zadania; zmian nie zapisano.',
|
||||
|
||||
@@ -2596,6 +2596,7 @@ const messages: TranslationMap = {
|
||||
'conversations.taskKanban.saveChanges': 'Salvar alterações',
|
||||
'conversations.taskKanban.deleteCard': 'Excluir',
|
||||
'conversations.taskKanban.workTask': 'Trabalhar na tarefa',
|
||||
'conversations.taskKanban.viewWork': 'Ver trabalho',
|
||||
'conversations.taskKanban.startingTask': 'Iniciando…',
|
||||
'conversations.taskKanban.updateFailed':
|
||||
'Não foi possível atualizar a tarefa; as alterações não foram salvas.',
|
||||
|
||||
@@ -2572,6 +2572,7 @@ const messages: TranslationMap = {
|
||||
'conversations.taskKanban.saveChanges': 'Сохранить изменения',
|
||||
'conversations.taskKanban.deleteCard': 'Удалить',
|
||||
'conversations.taskKanban.workTask': 'Работать над задачей',
|
||||
'conversations.taskKanban.viewWork': 'Показать работу',
|
||||
'conversations.taskKanban.startingTask': 'Запуск…',
|
||||
'conversations.taskKanban.updateFailed': 'Не удалось обновить задачу; изменения не сохранились.',
|
||||
'conversations.taskKanban.sourcesButton': 'Источники',
|
||||
|
||||
@@ -2428,6 +2428,7 @@ const messages: TranslationMap = {
|
||||
'conversations.taskKanban.saveChanges': '保存更改',
|
||||
'conversations.taskKanban.deleteCard': '删除',
|
||||
'conversations.taskKanban.workTask': '处理任务',
|
||||
'conversations.taskKanban.viewWork': '查看工作',
|
||||
'conversations.taskKanban.startingTask': '正在启动…',
|
||||
'conversations.taskKanban.updateFailed': '无法更新任务;更改未保存。',
|
||||
'conversations.taskKanban.sourcesButton': '来源',
|
||||
|
||||
@@ -186,6 +186,7 @@ const Conversations = ({
|
||||
const { t } = useT();
|
||||
const dispatch = useAppDispatch();
|
||||
const navigate = useNavigate();
|
||||
const location = useLocation();
|
||||
const { threads, selectedThreadId, messages, isLoadingMessages, messagesError, activeThreadId } =
|
||||
useAppSelector(state => state.thread);
|
||||
|
||||
@@ -370,6 +371,27 @@ const Conversations = ({
|
||||
// Match the sidebar's default General filter here so initial/resume
|
||||
// selection can't auto-pick a thread hidden by the selected tab.
|
||||
const visibleThreads = data.threads.filter(t => isThreadVisibleInTab(t, GENERAL_TAB_VALUE));
|
||||
// An explicit "open this session" intent (e.g. View work from the Agent
|
||||
// Tasks board) wins over passive resume — and bypasses the General-tab
|
||||
// visibility filter so a task-labelled session thread can actually be
|
||||
// opened (the resume default below only considers General threads).
|
||||
const openThreadId = (location.state as { openThreadId?: string } | null)?.openThreadId;
|
||||
const openThread = openThreadId ? data.threads.find(t => t.id === openThreadId) : undefined;
|
||||
if (openThread) {
|
||||
// Switch the sidebar tab to the bucket that contains the opened
|
||||
// thread (e.g. Tasks for a task session) so it's visible/selected in
|
||||
// the list instead of hidden behind the default General tab.
|
||||
setSelectedLabel(
|
||||
isThreadVisibleInTab(openThread, TASKS_TAB_VALUE)
|
||||
? TASKS_TAB_VALUE
|
||||
: isThreadVisibleInTab(openThread, SUBCONSCIOUS_TAB_VALUE)
|
||||
? SUBCONSCIOUS_TAB_VALUE
|
||||
: GENERAL_TAB_VALUE
|
||||
);
|
||||
dispatch(setSelectedThread(openThread.id));
|
||||
void dispatch(loadThreadMessages(openThread.id));
|
||||
return;
|
||||
}
|
||||
if (visibleThreads.length > 0) {
|
||||
// Prefer the thread the user was last viewing (persisted across
|
||||
// reloads via redux-persist on the `thread` slice). Only fall
|
||||
@@ -423,7 +445,6 @@ const Conversations = ({
|
||||
});
|
||||
}, [dispatch]);
|
||||
|
||||
const location = useLocation();
|
||||
const { containerRef: messagesContainerRef, endRef: messagesEndRef } = useStickToBottom(
|
||||
messages,
|
||||
selectedThreadId,
|
||||
@@ -1588,6 +1609,15 @@ const Conversations = ({
|
||||
t,
|
||||
});
|
||||
}}
|
||||
onViewSession={card => {
|
||||
if (!card.sessionThreadId) return;
|
||||
// Navigation only — do NOT mark the thread active. activeThreadId
|
||||
// tracks a true in-flight turn (set on send, cleared on
|
||||
// done/error). A completed session never emits that lifecycle
|
||||
// event, so forcing it active would wedge the composer.
|
||||
dispatch(setSelectedThread(card.sessionThreadId));
|
||||
void dispatch(loadThreadMessages(card.sessionThreadId));
|
||||
}}
|
||||
/>
|
||||
)}
|
||||
{visibleMessages.map(msg => (
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
import { useCallback, useEffect, useState } from 'react';
|
||||
import { useSearchParams } from 'react-router-dom';
|
||||
|
||||
import { ConfirmationModal } from '../components/intelligence/ConfirmationModal';
|
||||
import IntelligenceSubconsciousTab from '../components/intelligence/IntelligenceSubconsciousTab';
|
||||
@@ -25,7 +26,28 @@ type IntelligenceTab = 'memory' | 'subconscious' | 'tasks' | 'workflows' | 'coun
|
||||
export default function Intelligence() {
|
||||
const { t } = useT();
|
||||
|
||||
const [activeTab, setActiveTab] = useState<IntelligenceTab>('memory');
|
||||
// Tab is URL-backed (`/intelligence?tab=…`) so navigating away — e.g. to
|
||||
// Settings → Task Sources from the Agent Tasks tab — and coming back via
|
||||
// browser-back restores the same tab instead of resetting to Memory.
|
||||
// `replace` so switching tabs doesn't stack history entries.
|
||||
const [searchParams, setSearchParams] = useSearchParams();
|
||||
const tabParam = searchParams.get('tab');
|
||||
const activeTab: IntelligenceTab =
|
||||
tabParam && ['memory', 'subconscious', 'tasks', 'workflows', 'council'].includes(tabParam)
|
||||
? (tabParam as IntelligenceTab)
|
||||
: 'memory';
|
||||
const setActiveTab = useCallback(
|
||||
(tab: IntelligenceTab) => {
|
||||
setSearchParams(
|
||||
prev => {
|
||||
prev.set('tab', tab);
|
||||
return prev;
|
||||
},
|
||||
{ replace: true }
|
||||
);
|
||||
},
|
||||
[setSearchParams]
|
||||
);
|
||||
|
||||
// The legacy header pills (system-status + Ingesting/Queued chips) were
|
||||
// sourced from `useConsciousItems` + `useMemoryIngestionStatus`. They are
|
||||
|
||||
@@ -19,6 +19,7 @@ import { CoreRpcError } from '../../services/coreRpcClient';
|
||||
import agentProfileReducer from '../../store/agentProfileSlice';
|
||||
import chatRuntimeReducer, {
|
||||
setInferenceStatusForThread,
|
||||
setTaskBoardForThread,
|
||||
setToolTimelineForThread,
|
||||
} from '../../store/chatRuntimeSlice';
|
||||
import socketReducer from '../../store/socketSlice';
|
||||
@@ -1621,3 +1622,94 @@ describe('Conversations — thread title editing', () => {
|
||||
expect(threadApi.updateTitle).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
|
||||
describe('Conversations — open-session resume (View work)', () => {
|
||||
beforeEach(() => {
|
||||
vi.clearAllMocks();
|
||||
mockGetThreadMessages.mockResolvedValue({ messages: [], count: 0 });
|
||||
});
|
||||
|
||||
it('honours location.state.openThreadId to open a task session on mount', async () => {
|
||||
// A task-labelled session thread, reachable only via an explicit
|
||||
// open-intent because it's hidden behind the default General tab.
|
||||
const taskThread = makeThread({
|
||||
id: 'task-open-1',
|
||||
title: 'Autonomous run',
|
||||
labels: ['tasks'],
|
||||
});
|
||||
mockGetThreads.mockResolvedValue({ threads: [taskThread], count: 1 });
|
||||
|
||||
const store = buildStore({ thread: emptyThreadState });
|
||||
const { default: Conversations } = await import('../Conversations');
|
||||
|
||||
await act(async () => {
|
||||
render(
|
||||
<Provider store={store}>
|
||||
<MemoryRouter
|
||||
initialEntries={[
|
||||
{ pathname: '/conversations', state: { openThreadId: 'task-open-1' } },
|
||||
]}>
|
||||
<Conversations />
|
||||
</MemoryRouter>
|
||||
</Provider>
|
||||
);
|
||||
});
|
||||
|
||||
// The open-intent selects the task session (bypassing the General-tab
|
||||
// filter) and loads its messages.
|
||||
await waitFor(() => expect(store.getState().thread.selectedThreadId).toBe('task-open-1'));
|
||||
await waitFor(() => expect(mockGetThreadMessages).toHaveBeenCalled());
|
||||
});
|
||||
|
||||
it("View work on a selected task board opens that card's session thread", async () => {
|
||||
const thread = makeThread({ id: 'board-thread', title: 'Board thread' });
|
||||
mockGetThreads.mockResolvedValue({ threads: [thread], count: 1 });
|
||||
|
||||
const store = buildStore({ thread: selectedThreadState(thread) });
|
||||
|
||||
const { default: Conversations } = await import('../Conversations');
|
||||
await act(async () => {
|
||||
render(
|
||||
<Provider store={store}>
|
||||
<MemoryRouter initialEntries={['/conversations']}>
|
||||
<Conversations />
|
||||
</MemoryRouter>
|
||||
</Provider>
|
||||
);
|
||||
});
|
||||
// Let the mount-resume effect settle, then seed the selected thread's task
|
||||
// board with a card that has a live session (seeding before mount gets
|
||||
// clobbered by turn-state hydration).
|
||||
await screen.findByPlaceholderText('How can I help you today?');
|
||||
const selectedId = store.getState().thread.selectedThreadId ?? 'board-thread';
|
||||
await act(async () => {
|
||||
store.dispatch(
|
||||
setTaskBoardForThread({
|
||||
threadId: selectedId,
|
||||
board: {
|
||||
threadId: selectedId,
|
||||
updatedAt: '',
|
||||
cards: [
|
||||
{
|
||||
id: 'tc1',
|
||||
title: 'Worked card',
|
||||
status: 'in_progress',
|
||||
order: 0,
|
||||
updatedAt: '',
|
||||
sessionThreadId: 'sess-99',
|
||||
},
|
||||
],
|
||||
},
|
||||
})
|
||||
);
|
||||
});
|
||||
|
||||
const viewBtn = await screen.findByTitle('View work');
|
||||
await act(async () => {
|
||||
fireEvent.click(viewBtn);
|
||||
});
|
||||
|
||||
// onViewSession navigates the chat view to the card's session thread.
|
||||
await waitFor(() => expect(store.getState().thread.selectedThreadId).toBe('sess-99'));
|
||||
});
|
||||
});
|
||||
|
||||
@@ -0,0 +1,90 @@
|
||||
import { fireEvent, render, screen } from '@testing-library/react';
|
||||
import { MemoryRouter } from 'react-router-dom';
|
||||
import { beforeEach, describe, expect, it, vi } from 'vitest';
|
||||
|
||||
// Stub the heavy tab content + chrome so the test exercises only the
|
||||
// URL-backed tab selection logic in Intelligence.tsx.
|
||||
vi.mock('../../lib/i18n/I18nContext', () => ({ useT: () => ({ t: (key: string) => key }) }));
|
||||
vi.mock('../../components/intelligence/MemorySection', () => ({
|
||||
default: () => <div data-testid="tab-memory" />,
|
||||
}));
|
||||
vi.mock('../../components/intelligence/IntelligenceSubconsciousTab', () => ({
|
||||
default: () => <div data-testid="tab-subconscious" />,
|
||||
}));
|
||||
vi.mock('../../components/intelligence/IntelligenceTasksTab', () => ({
|
||||
default: () => <div data-testid="tab-tasks" />,
|
||||
}));
|
||||
vi.mock('../../components/intelligence/ModelCouncilTab', () => ({
|
||||
default: () => <div data-testid="tab-council" />,
|
||||
}));
|
||||
vi.mock('../AgentWorkflows', () => ({ default: () => <div data-testid="tab-workflows" /> }));
|
||||
vi.mock('../../components/intelligence/Toast', () => ({ ToastContainer: () => null }));
|
||||
vi.mock('../../components/intelligence/ConfirmationModal', () => ({
|
||||
ConfirmationModal: () => null,
|
||||
}));
|
||||
vi.mock('../../components/PillTabBar', () => ({
|
||||
default: ({ selected, onChange }: { selected: string; onChange: (tab: string) => void }) => (
|
||||
<div data-testid="pilltabs">
|
||||
<span>selected:{selected}</span>
|
||||
{['memory', 'subconscious', 'tasks', 'workflows', 'council'].map(tab => (
|
||||
<button key={tab} type="button" onClick={() => onChange(tab)}>
|
||||
go-{tab}
|
||||
</button>
|
||||
))}
|
||||
</div>
|
||||
),
|
||||
}));
|
||||
vi.mock('../../hooks/useIntelligenceSocket', () => ({
|
||||
useIntelligenceSocket: () => ({ isConnected: true }),
|
||||
useIntelligenceSocketManager: () => ({}),
|
||||
}));
|
||||
vi.mock('../../hooks/useSubconscious', () => ({
|
||||
useSubconscious: () => ({
|
||||
status: 'idle',
|
||||
mode: 'manual',
|
||||
intervalMinutes: 30,
|
||||
triggering: false,
|
||||
settingMode: false,
|
||||
triggerTick: vi.fn(),
|
||||
setMode: vi.fn(),
|
||||
setIntervalMinutes: vi.fn(),
|
||||
}),
|
||||
}));
|
||||
|
||||
const Intelligence = (await import('../Intelligence')).default;
|
||||
|
||||
function renderAt(path: string) {
|
||||
return render(
|
||||
<MemoryRouter initialEntries={[path]}>
|
||||
<Intelligence />
|
||||
</MemoryRouter>
|
||||
);
|
||||
}
|
||||
|
||||
describe('Intelligence URL-backed tab', () => {
|
||||
beforeEach(() => vi.clearAllMocks());
|
||||
|
||||
it('defaults to the memory tab when no ?tab is present', () => {
|
||||
renderAt('/intelligence');
|
||||
expect(screen.getByTestId('tab-memory')).toBeInTheDocument();
|
||||
expect(screen.getByText('selected:memory')).toBeInTheDocument();
|
||||
});
|
||||
|
||||
it('honours ?tab=tasks from the URL', () => {
|
||||
renderAt('/intelligence?tab=tasks');
|
||||
expect(screen.getByTestId('tab-tasks')).toBeInTheDocument();
|
||||
expect(screen.getByText('selected:tasks')).toBeInTheDocument();
|
||||
});
|
||||
|
||||
it('falls back to memory for an unknown ?tab value', () => {
|
||||
renderAt('/intelligence?tab=bogus');
|
||||
expect(screen.getByTestId('tab-memory')).toBeInTheDocument();
|
||||
});
|
||||
|
||||
it('switching tabs updates the active tab via the URL', () => {
|
||||
renderAt('/intelligence');
|
||||
fireEvent.click(screen.getByText('go-council'));
|
||||
expect(screen.getByTestId('tab-council')).toBeInTheDocument();
|
||||
expect(screen.getByText('selected:council')).toBeInTheDocument();
|
||||
});
|
||||
});
|
||||
@@ -21,6 +21,14 @@ vi.mock('../../../utils/tauriCommands', () => ({
|
||||
openhumanTaskSourcesUpdate: vi.fn(),
|
||||
}));
|
||||
|
||||
// TaskSourceControls navigates to the settings page via useNavigate(); these
|
||||
// tests render the board without a <Router>, so capture navigation via a spy.
|
||||
const navigateSpy = vi.hoisted(() => vi.fn());
|
||||
vi.mock('react-router-dom', async () => {
|
||||
const actual = await vi.importActual<typeof import('react-router-dom')>('react-router-dom');
|
||||
return { ...actual, useNavigate: () => navigateSpy };
|
||||
});
|
||||
|
||||
function card(partial: Partial<TaskBoardCard>): TaskBoardCard {
|
||||
return {
|
||||
id: 'c1',
|
||||
@@ -182,5 +190,40 @@ describe('TaskKanbanBoard approval surface', () => {
|
||||
await waitFor(() =>
|
||||
expect(openhumanTaskSourcesUpdate).toHaveBeenCalledWith('src-1', { enabled: false })
|
||||
);
|
||||
|
||||
// "Manage sources" jumps to the settings page.
|
||||
fireEvent.click(screen.getByText('conversations.taskKanban.sources.manage'));
|
||||
expect(navigateSpy).toHaveBeenCalledWith('/settings/task-sources');
|
||||
});
|
||||
|
||||
it('shows a "View work" button on a card with a session thread and calls onViewSession', () => {
|
||||
const onViewSession = vi.fn();
|
||||
render(
|
||||
<TaskKanbanBoard
|
||||
board={board([
|
||||
card({
|
||||
id: 'c1',
|
||||
title: 'Worked card',
|
||||
status: 'in_progress',
|
||||
sessionThreadId: 'task-xyz',
|
||||
}),
|
||||
])}
|
||||
onViewSession={onViewSession}
|
||||
/>
|
||||
);
|
||||
|
||||
fireEvent.click(screen.getByTitle('conversations.taskKanban.viewWork'));
|
||||
expect(onViewSession).toHaveBeenCalledWith(expect.objectContaining({ id: 'c1' }));
|
||||
});
|
||||
|
||||
it('omits the "View work" button when the card has no session thread', () => {
|
||||
render(
|
||||
<TaskKanbanBoard
|
||||
board={board([card({ id: 'c2', title: 'Unworked card', status: 'todo' })])}
|
||||
onViewSession={vi.fn()}
|
||||
/>
|
||||
);
|
||||
|
||||
expect(screen.queryByTitle('conversations.taskKanban.viewWork')).toBeNull();
|
||||
});
|
||||
});
|
||||
|
||||
@@ -13,6 +13,7 @@ import {
|
||||
LuWrench,
|
||||
LuX,
|
||||
} from 'react-icons/lu';
|
||||
import { useNavigate } from 'react-router-dom';
|
||||
|
||||
import { useT } from '../../../lib/i18n/I18nContext';
|
||||
import type { TaskBoard, TaskBoardCard, TaskBoardCardStatus } from '../../../types/turnState';
|
||||
@@ -97,6 +98,9 @@ interface TaskKanbanBoardProps {
|
||||
onDecidePlan?: (card: TaskBoardCard, approve: boolean) => void;
|
||||
/** Start work on a card from a higher-level task board. */
|
||||
onWorkTask?: (card: TaskBoardCard) => void;
|
||||
/** Jump to the card's agent session in Conversations. Shown on any card that
|
||||
* carries a `sessionThreadId` (a run is live or has happened). */
|
||||
onViewSession?: (card: TaskBoardCard) => void;
|
||||
workingCardId?: string | null;
|
||||
}
|
||||
|
||||
@@ -110,6 +114,7 @@ export function TaskKanbanBoard({
|
||||
onDeleteCard,
|
||||
onDecidePlan,
|
||||
onWorkTask,
|
||||
onViewSession,
|
||||
workingCardId = null,
|
||||
}: TaskKanbanBoardProps) {
|
||||
const { t } = useT();
|
||||
@@ -195,6 +200,7 @@ export function TaskKanbanBoard({
|
||||
hasBriefActions={Boolean(onUpdateCard || onDeleteCard)}
|
||||
onDecidePlan={onDecidePlan}
|
||||
onWorkTask={onWorkTask}
|
||||
onViewSession={onViewSession}
|
||||
working={workingCardId === card.id}
|
||||
onOpenBrief={() => setSelectedCardId(card.id)}
|
||||
/>
|
||||
@@ -224,6 +230,7 @@ function TaskBoardArticle({
|
||||
hasBriefActions,
|
||||
onDecidePlan,
|
||||
onWorkTask,
|
||||
onViewSession,
|
||||
working,
|
||||
onOpenBrief,
|
||||
}: {
|
||||
@@ -234,6 +241,7 @@ function TaskBoardArticle({
|
||||
hasBriefActions: boolean;
|
||||
onDecidePlan?: (card: TaskBoardCard, approve: boolean) => void;
|
||||
onWorkTask?: (card: TaskBoardCard) => void;
|
||||
onViewSession?: (card: TaskBoardCard) => void;
|
||||
working: boolean;
|
||||
onOpenBrief: () => void;
|
||||
}) {
|
||||
@@ -246,7 +254,16 @@ function TaskBoardArticle({
|
||||
<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>
|
||||
{card.status === 'awaiting_approval' && onDecidePlan ? (
|
||||
{card.sessionThreadId && onViewSession ? (
|
||||
<button
|
||||
type="button"
|
||||
title={t('conversations.taskKanban.viewWork')}
|
||||
onClick={() => onViewSession(card)}
|
||||
className="inline-flex flex-shrink-0 items-center gap-1 rounded-md bg-ocean-50 px-1.5 py-0.5 text-[10px] font-medium text-ocean-700 transition-colors hover:bg-ocean-100 dark:bg-ocean-500/10 dark:text-ocean-200 dark:hover:bg-ocean-500/20">
|
||||
<LuExternalLink className="h-3 w-3 flex-none" />
|
||||
{t('conversations.taskKanban.viewWork')}
|
||||
</button>
|
||||
) : card.status === 'awaiting_approval' && onDecidePlan ? (
|
||||
<div className="flex flex-shrink-0 items-center gap-1">
|
||||
<button
|
||||
type="button"
|
||||
@@ -456,6 +473,7 @@ function formatFetchNotice(outcome: FetchOutcome, t: (key: string) => string): s
|
||||
|
||||
function TaskSourceControls({ disabled, compact }: { disabled: boolean; compact: boolean }) {
|
||||
const { t } = useT();
|
||||
const navigate = useNavigate();
|
||||
const [loading, setLoading] = useState(true);
|
||||
const [sources, setSources] = useState<TaskSource[]>([]);
|
||||
const [status, setStatus] = useState<TaskSourcesStatus | null>(null);
|
||||
@@ -543,11 +561,12 @@ function TaskSourceControls({ disabled, compact }: { disabled: boolean; compact:
|
||||
)}
|
||||
</div>
|
||||
<div className="flex items-center gap-2">
|
||||
<a
|
||||
href="#/settings/task-sources"
|
||||
<button
|
||||
type="button"
|
||||
onClick={() => navigate('/settings/task-sources')}
|
||||
className="text-[11px] font-medium text-ocean-600 hover:text-ocean-700 dark:text-ocean-300 dark:hover:text-ocean-200">
|
||||
{t('conversations.taskKanban.sources.manage')}
|
||||
</a>
|
||||
</button>
|
||||
<button
|
||||
type="button"
|
||||
aria-label={t('settings.taskSources.refresh')}
|
||||
|
||||
@@ -15,7 +15,7 @@ function isSubconsciousThread(thread: Thread): boolean {
|
||||
return hasAnyLabel(thread, [SUBCONSCIOUS_TAB_VALUE, ...LEGACY_SUBCONSCIOUS_LABELS]);
|
||||
}
|
||||
|
||||
function isTaskThread(thread: Thread): boolean {
|
||||
export function isTaskThread(thread: Thread): boolean {
|
||||
return Boolean(
|
||||
thread.parentThreadId || hasAnyLabel(thread, [TASKS_TAB_VALUE, ...LEGACY_TASK_LABELS])
|
||||
);
|
||||
|
||||
@@ -0,0 +1,44 @@
|
||||
import { beforeEach, describe, expect, it, vi } from 'vitest';
|
||||
|
||||
const mockCallCoreRpc = vi.fn();
|
||||
|
||||
vi.mock('../../coreRpcClient', () => ({
|
||||
callCoreRpc: (...args: unknown[]) => mockCallCoreRpc(...args),
|
||||
}));
|
||||
|
||||
const { todosApi } = await import('../todosApi');
|
||||
|
||||
describe('todosApi.setSessionThread', () => {
|
||||
beforeEach(() => {
|
||||
mockCallCoreRpc.mockReset();
|
||||
});
|
||||
|
||||
it('links a card to its session thread via the RPC and returns the board', async () => {
|
||||
mockCallCoreRpc.mockResolvedValueOnce({
|
||||
threadId: 'user-tasks',
|
||||
cards: [
|
||||
{ id: 'c1', title: 'T', status: 'todo', order: 0, updatedAt: '2026-01-01T00:00:00Z' },
|
||||
],
|
||||
});
|
||||
|
||||
const board = await todosApi.setSessionThread('user-tasks', 'c1', 'task-abc');
|
||||
|
||||
expect(mockCallCoreRpc).toHaveBeenCalledWith({
|
||||
method: 'openhuman.todos_set_session_thread',
|
||||
params: { thread_id: 'user-tasks', id: 'c1', sessionThreadId: 'task-abc' },
|
||||
});
|
||||
expect(board.threadId).toBe('user-tasks');
|
||||
expect(board.cards).toHaveLength(1);
|
||||
});
|
||||
|
||||
it('passes null through to clear the link', async () => {
|
||||
mockCallCoreRpc.mockResolvedValueOnce({ threadId: 'user-tasks', cards: [] });
|
||||
|
||||
await todosApi.setSessionThread('user-tasks', 'c1', null);
|
||||
|
||||
expect(mockCallCoreRpc).toHaveBeenCalledWith({
|
||||
method: 'openhuman.todos_set_session_thread',
|
||||
params: { thread_id: 'user-tasks', id: 'c1', sessionThreadId: null },
|
||||
});
|
||||
});
|
||||
});
|
||||
@@ -51,6 +51,9 @@ export interface AddTodoInput {
|
||||
notes?: string | null;
|
||||
assignedAgent?: string | null;
|
||||
approvalMode?: TaskApprovalMode | null;
|
||||
/** Originating task-source identifiers, stamped onto the promoted card so the
|
||||
* inbox can detect it was already picked up. */
|
||||
sourceMetadata?: Record<string, unknown> | null;
|
||||
}
|
||||
|
||||
/** Fields accepted when editing a card. Omitted fields are left unchanged. */
|
||||
@@ -124,6 +127,7 @@ export const todosApi = {
|
||||
notes: input.notes,
|
||||
assignedAgent: input.assignedAgent,
|
||||
approvalMode: input.approvalMode,
|
||||
sourceMetadata: input.sourceMetadata,
|
||||
}),
|
||||
});
|
||||
return snapshotToBoard(snap, input.threadId);
|
||||
@@ -167,6 +171,21 @@ export const todosApi = {
|
||||
return snapshotToBoard(snap, threadId);
|
||||
},
|
||||
|
||||
/** Link a card to its agent session's conversation thread (drives the board
|
||||
* "View session" jump). Pass `null` to clear the link. */
|
||||
setSessionThread: async (
|
||||
threadId: string,
|
||||
id: string,
|
||||
sessionThreadId: string | null
|
||||
): Promise<TaskBoard> => {
|
||||
log('setSessionThread threadId=%s id=%s sessionThreadId=%s', threadId, id, sessionThreadId);
|
||||
const snap = await callCoreRpc<TodosSnapshotWire>({
|
||||
method: 'openhuman.todos_set_session_thread',
|
||||
params: { thread_id: threadId, id, sessionThreadId },
|
||||
});
|
||||
return snapshotToBoard(snap, threadId);
|
||||
},
|
||||
|
||||
/** Remove a card by id. */
|
||||
remove: async (threadId: string, id: string): Promise<TaskBoard> => {
|
||||
log('remove threadId=%s id=%s', threadId, id);
|
||||
|
||||
@@ -34,6 +34,10 @@ export interface TaskBoardCard {
|
||||
evidence?: string[];
|
||||
notes?: string | null;
|
||||
blocker?: string | null;
|
||||
/** Conversation thread id of the card's live/last agent session, if any —
|
||||
* drives the "View session" jump into Conversations. Set by the autonomous
|
||||
* dispatcher and the manual "Work" path. */
|
||||
sessionThreadId?: 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. */
|
||||
|
||||
@@ -33,7 +33,7 @@ use crate::openhuman::inference::provider::{
|
||||
use super::super::parse::build_native_assistant_history;
|
||||
use super::super::run_queue::RunQueue;
|
||||
use super::super::token_budget::trim_chat_messages_to_budget;
|
||||
use super::super::tool_loop::{RepeatFailureGuard, STREAM_CHUNK_MIN_CHARS};
|
||||
use super::super::tool_loop::{RepeatFailureGuard, RepeatOutputGuard, STREAM_CHUNK_MIN_CHARS};
|
||||
use super::checkpoint::CheckpointStrategy;
|
||||
use super::parser::ResponseParser;
|
||||
use super::progress::ProgressReporter;
|
||||
@@ -111,6 +111,10 @@ pub(crate) async fn run_turn_engine(
|
||||
// Repeated-failure circuit breaker — halts with a root cause rather than
|
||||
// grinding to `max_iterations`.
|
||||
let mut failure_guard = RepeatFailureGuard::new();
|
||||
// No-progress narration breaker — trips when the model re-emits the same
|
||||
// response + tool call across iterations even when each call "succeeds"
|
||||
// (the gap left by the failure guard + per-generation frequency_penalty).
|
||||
let mut repeat_guard = RepeatOutputGuard::new();
|
||||
let mut halt_reason: Option<String> = None;
|
||||
for iteration in 0..max_iterations {
|
||||
progress
|
||||
@@ -501,6 +505,51 @@ pub(crate) async fn run_turn_engine(
|
||||
});
|
||||
}
|
||||
|
||||
// No-progress narration breaker: if this iteration's assistant output
|
||||
// (text + tool-call name/args) is byte-identical to the previous N in a
|
||||
// row, the run is stuck re-issuing the same step. Halt with a summary
|
||||
// rather than grinding to the iteration cap. Checked BEFORE executing
|
||||
// the (repeated) tool call so we don't burn another no-op iteration.
|
||||
{
|
||||
let mut sig = response_text.trim().to_string();
|
||||
for call in &tool_calls {
|
||||
sig.push('\u{1}');
|
||||
sig.push_str(&call.name);
|
||||
sig.push('\u{1}');
|
||||
sig.push_str(&call.arguments.to_string());
|
||||
}
|
||||
if let Some(reason) = repeat_guard.record(&sig) {
|
||||
tracing::warn!(
|
||||
iteration,
|
||||
"[agent_loop] repeat-output circuit breaker tripped — identical response+tool-call repeated; halting with no-progress summary"
|
||||
);
|
||||
history.push(ChatMessage::assistant(assistant_history_content.clone()));
|
||||
// Mirror the assistant turn to the observer like every other
|
||||
// assistant-append path, so transcript/mirroring isn't skipped
|
||||
// for the final repeated iteration on this early exit.
|
||||
observer
|
||||
.on_assistant(
|
||||
&display_text,
|
||||
&response_text,
|
||||
reasoning_content.as_deref(),
|
||||
&native_tool_calls,
|
||||
&tool_calls,
|
||||
iteration,
|
||||
false,
|
||||
)
|
||||
.await;
|
||||
observer.after_iteration(history, iteration);
|
||||
progress.turn_completed((iteration + 1) as u32).await;
|
||||
return Ok(TurnEngineOutcome {
|
||||
text: reason,
|
||||
iterations: (iteration + 1) as u32,
|
||||
cost: turn_cost,
|
||||
hit_cap: false,
|
||||
early_exit_tool: None,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
// Print any text the LLM produced alongside tool calls (unless silent)
|
||||
if !silent && !display_text.is_empty() {
|
||||
print!("{display_text}");
|
||||
|
||||
@@ -154,6 +154,68 @@ impl RepeatFailureGuard {
|
||||
}
|
||||
}
|
||||
|
||||
/// If the model emits the IDENTICAL assistant output (narrative text + the same
|
||||
/// tool-call name/args) this many times in a row, it's stuck in a no-progress
|
||||
/// narration loop — halt. Set low enough to bail early (the observed
|
||||
/// degeneration repeated ~195×) but above any legitimate short retry.
|
||||
pub(crate) const REPEAT_OUTPUT_THRESHOLD: u32 = 4;
|
||||
|
||||
/// Repeat-OUTPUT circuit breaker — distinct from [`RepeatFailureGuard`], which
|
||||
/// only counts tool *failures* and resets on every success.
|
||||
///
|
||||
/// This catches the degenerate case where each iteration re-emits the SAME
|
||||
/// narration + SAME tool call and the call nominally "succeeds" yet nothing
|
||||
/// advances (e.g. the model narrating "now let me create the files…" and
|
||||
/// re-issuing the same `run_code` forever). That loop is invisible to two
|
||||
/// things people reach for first:
|
||||
/// * `frequency_penalty` — per-generation only; each iteration is a fresh,
|
||||
/// individually non-repetitive generation, so it has nothing to penalise
|
||||
/// and no memory across turns.
|
||||
/// * [`RepeatFailureGuard`] — resets on success, so a repeated *successful*
|
||||
/// no-op never trips it.
|
||||
///
|
||||
/// Trips on `REPEAT_OUTPUT_THRESHOLD` consecutive identical signatures; a
|
||||
/// different signature (real progress) resets the run, so interleaved varied
|
||||
/// work never trips it.
|
||||
#[derive(Default)]
|
||||
pub(crate) struct RepeatOutputGuard {
|
||||
last_hash: Option<u64>,
|
||||
consecutive: u32,
|
||||
}
|
||||
|
||||
impl RepeatOutputGuard {
|
||||
pub(crate) fn new() -> Self {
|
||||
Self::default()
|
||||
}
|
||||
|
||||
/// Record one iteration's output signature (assistant text + tool-call
|
||||
/// name/args). Returns `Some(halt summary)` once the identical signature has
|
||||
/// repeated [`REPEAT_OUTPUT_THRESHOLD`] times back-to-back.
|
||||
pub(crate) fn record(&mut self, signature: &str) -> Option<String> {
|
||||
use std::hash::{Hash, Hasher};
|
||||
let mut hasher = std::collections::hash_map::DefaultHasher::new();
|
||||
signature.hash(&mut hasher);
|
||||
let h = hasher.finish();
|
||||
if self.last_hash == Some(h) {
|
||||
self.consecutive += 1;
|
||||
} else {
|
||||
self.last_hash = Some(h);
|
||||
self.consecutive = 1;
|
||||
}
|
||||
if self.consecutive >= REPEAT_OUTPUT_THRESHOLD {
|
||||
return Some(format!(
|
||||
"Stopping: the last {} iterations produced the IDENTICAL response and tool call \
|
||||
with no change — the run is stuck repeating the same step without making \
|
||||
progress. Re-issuing it will not help. Summarise what (if anything) was actually \
|
||||
accomplished and report that the task could not progress, or take a genuinely \
|
||||
different approach.",
|
||||
self.consecutive,
|
||||
));
|
||||
}
|
||||
None
|
||||
}
|
||||
}
|
||||
|
||||
/// Clamp the last-error text embedded in a circuit-breaker halt summary so a huge
|
||||
/// tool error (already capped at 1MB upstream) can't blow up the agent's result.
|
||||
pub(crate) fn truncate_for_halt(s: &str) -> String {
|
||||
|
||||
@@ -1349,3 +1349,39 @@ async fn auto_approved_external_effect_tool_runs_through_loop_without_parking()
|
||||
"auto-approved external-effect tool must execute (gate must not park it)"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn repeat_output_guard_trips_at_threshold() {
|
||||
let mut g = RepeatOutputGuard::new();
|
||||
// The first THRESHOLD-1 identical signatures must NOT trip.
|
||||
for _ in 1..REPEAT_OUTPUT_THRESHOLD {
|
||||
assert!(g.record("same-narration|run_code|{args}").is_none());
|
||||
}
|
||||
// The THRESHOLD-th identical signature trips with a no-progress summary.
|
||||
let halt = g
|
||||
.record("same-narration|run_code|{args}")
|
||||
.expect("identical streak at threshold must trip");
|
||||
assert!(
|
||||
halt.contains("IDENTICAL") || halt.contains("stuck") || halt.contains("progress"),
|
||||
"halt summary should explain the no-progress loop: {halt}"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn repeat_output_guard_resets_on_changed_signature() {
|
||||
let mut g = RepeatOutputGuard::new();
|
||||
assert!(g.record("a").is_none());
|
||||
assert!(g.record("a").is_none());
|
||||
// A different signature = real progress; the streak resets.
|
||||
assert!(g.record("b").is_none());
|
||||
// It then takes a FULL fresh streak of the new signature to trip — so
|
||||
// interleaved/varied work never trips it.
|
||||
let mut last = None;
|
||||
for _ in 1..REPEAT_OUTPUT_THRESHOLD {
|
||||
last = g.record("b");
|
||||
}
|
||||
assert!(
|
||||
last.is_some(),
|
||||
"a fresh THRESHOLD-long identical streak should trip after a reset"
|
||||
);
|
||||
}
|
||||
|
||||
@@ -43,6 +43,7 @@ mod schemas;
|
||||
pub mod stop_hooks;
|
||||
pub mod task_board;
|
||||
pub mod task_dispatcher;
|
||||
pub(crate) mod task_session;
|
||||
pub mod tool_policy;
|
||||
pub mod tools;
|
||||
pub mod tree_loader;
|
||||
|
||||
@@ -87,6 +87,12 @@ pub struct TaskBoardCard {
|
||||
pub notes: Option<String>,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub blocker: Option<String>,
|
||||
/// Conversation thread id of the card's live/last agent session, when one
|
||||
/// exists. Set by the autonomous dispatcher (`task_session`) and the manual
|
||||
/// "Work" path so the UI can offer a "View session" jump into Conversations.
|
||||
/// `None` for a card that has never been run.
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub session_thread_id: 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
|
||||
@@ -351,6 +357,11 @@ pub fn normalise_board(board: &mut TaskBoard) {
|
||||
.as_ref()
|
||||
.map(|s| s.trim().to_string())
|
||||
.filter(|s| !s.is_empty());
|
||||
card.session_thread_id = card
|
||||
.session_thread_id
|
||||
.as_ref()
|
||||
.map(|s| s.trim().to_string())
|
||||
.filter(|s| !s.is_empty());
|
||||
if card.status == TaskCardStatus::Blocked && card.blocker.is_none() {
|
||||
card.blocker = card.notes.clone();
|
||||
tracing::trace!(
|
||||
@@ -439,6 +450,7 @@ mod tests {
|
||||
evidence: vec![" cargo test ".into()],
|
||||
notes: Some(" note ".into()),
|
||||
blocker: None,
|
||||
session_thread_id: Some(" ".into()),
|
||||
source_metadata: None,
|
||||
order: 99,
|
||||
updated_at: String::new(),
|
||||
@@ -456,6 +468,7 @@ mod tests {
|
||||
evidence: Vec::new(),
|
||||
notes: Some("waiting on user".into()),
|
||||
blocker: None,
|
||||
session_thread_id: None,
|
||||
source_metadata: None,
|
||||
order: 99,
|
||||
updated_at: String::new(),
|
||||
@@ -479,6 +492,9 @@ mod tests {
|
||||
);
|
||||
assert_eq!(saved.cards[0].acceptance_criteria, vec!["tests pass"]);
|
||||
assert_eq!(saved.cards[0].evidence, vec!["cargo test"]);
|
||||
// Whitespace-only session_thread_id normalises to None so the board
|
||||
// never offers a blank "View session" jump target.
|
||||
assert_eq!(saved.cards[0].session_thread_id, None);
|
||||
assert_eq!(saved.cards[0].order, 0);
|
||||
assert!(saved.cards[0].id.starts_with("task-"));
|
||||
assert_eq!(saved.cards[1].blocker.as_deref(), Some("waiting on user"));
|
||||
@@ -527,6 +543,7 @@ mod tests {
|
||||
evidence: Vec::new(),
|
||||
notes: None,
|
||||
blocker: None,
|
||||
session_thread_id: None,
|
||||
source_metadata: None,
|
||||
order: 99,
|
||||
updated_at: String::new(),
|
||||
@@ -544,6 +561,7 @@ mod tests {
|
||||
evidence: Vec::new(),
|
||||
notes: None,
|
||||
blocker: None,
|
||||
session_thread_id: None,
|
||||
source_metadata: None,
|
||||
order: 99,
|
||||
updated_at: String::new(),
|
||||
|
||||
@@ -18,8 +18,9 @@
|
||||
//! 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::collections::HashMap;
|
||||
use std::path::Path;
|
||||
use std::sync::OnceLock;
|
||||
use std::sync::{Mutex, OnceLock};
|
||||
use std::time::Duration;
|
||||
|
||||
use crate::openhuman::agent::harness::definition::{AgentDefinitionRegistry, PromptSource};
|
||||
@@ -27,6 +28,7 @@ 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::{TaskApprovalMode, TaskBoardCard, TaskCardStatus};
|
||||
use crate::openhuman::agent::task_session;
|
||||
use crate::openhuman::config::Config;
|
||||
use crate::openhuman::todos::ops::{self, BoardLocation, CardPatch, USER_TASKS_THREAD_ID};
|
||||
use crate::openhuman::todos::runs::{self, RunLimits, RunOutcome};
|
||||
@@ -42,6 +44,85 @@ 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;
|
||||
|
||||
/// Handle to an in-flight autonomous run, keyed by its session `thread_id`.
|
||||
///
|
||||
/// Autonomous runs are detached `tokio` tasks, not web-channel turns, so they
|
||||
/// are invisible to the web channel's own in-flight registry — which is why the
|
||||
/// chat **Cancel** button (which calls `channel_web_cancel`) couldn't stop them.
|
||||
/// Registering the run's [`AbortHandle`](tokio::task::AbortHandle) here lets
|
||||
/// [`cancel_session`] abort it from that same cancel path.
|
||||
struct ActiveRun {
|
||||
abort: tokio::task::AbortHandle,
|
||||
hb_cancel: tokio::sync::watch::Sender<bool>,
|
||||
location: BoardLocation,
|
||||
card_id: String,
|
||||
run_id: String,
|
||||
}
|
||||
|
||||
static ACTIVE_RUNS: OnceLock<Mutex<HashMap<String, ActiveRun>>> = OnceLock::new();
|
||||
|
||||
fn active_runs() -> &'static Mutex<HashMap<String, ActiveRun>> {
|
||||
ACTIVE_RUNS.get_or_init(|| Mutex::new(HashMap::new()))
|
||||
}
|
||||
|
||||
fn register_active_run(thread_id: String, run: ActiveRun) {
|
||||
active_runs()
|
||||
.lock()
|
||||
.expect("active_runs mutex poisoned")
|
||||
.insert(thread_id, run);
|
||||
}
|
||||
|
||||
/// Remove and return the active-run entry for `thread_id`. The naturally
|
||||
/// completing run and a concurrent [`cancel_session`] race on this — whoever
|
||||
/// gets `Some` "owns" the terminal board write-back, so it happens exactly once.
|
||||
fn take_active_run(thread_id: &str) -> Option<ActiveRun> {
|
||||
active_runs()
|
||||
.lock()
|
||||
.expect("active_runs mutex poisoned")
|
||||
.remove(thread_id)
|
||||
}
|
||||
|
||||
/// Cancel the in-flight autonomous run streaming into session `thread_id`.
|
||||
///
|
||||
/// Aborts the detached run task, stops its heartbeat, marks the card `blocked`
|
||||
/// (user-cancelled) so it doesn't dangle `in_progress`, and emits the terminal
|
||||
/// chat event (broadcast as `"system"`) so the session UI stops "processing".
|
||||
/// Returns `true` if a run was found and cancelled. Wired into the web channel's
|
||||
/// `channel_web_cancel` as the fallback when the thread has no web-channel turn.
|
||||
pub async fn cancel_session(thread_id: &str) -> bool {
|
||||
let Some(run) = take_active_run(thread_id) else {
|
||||
return false;
|
||||
};
|
||||
run.abort.abort();
|
||||
let _ = run.hb_cancel.send(true);
|
||||
// The aborted task never reaches its own write-back — do it here so the
|
||||
// card lands in a terminal state instead of a stale `in_progress`.
|
||||
write_back(
|
||||
&run.location,
|
||||
&run.card_id,
|
||||
&run.run_id,
|
||||
Err("Cancelled by user".to_string()),
|
||||
);
|
||||
crate::openhuman::channels::providers::web::publish_web_channel_event(
|
||||
crate::core::socketio::WebChannelEvent {
|
||||
event: "chat_error".to_string(),
|
||||
client_id: "system".to_string(),
|
||||
thread_id: thread_id.to_string(),
|
||||
request_id: run.run_id.clone(),
|
||||
message: Some("Cancelled".to_string()),
|
||||
error_type: Some("cancelled".to_string()),
|
||||
..Default::default()
|
||||
},
|
||||
);
|
||||
tracing::info!(
|
||||
thread_id = %thread_id,
|
||||
card_id = %run.card_id,
|
||||
run_id = %run.run_id,
|
||||
"[task_dispatcher] cancelled autonomous run via chat cancel"
|
||||
);
|
||||
true
|
||||
}
|
||||
|
||||
/// 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
|
||||
@@ -270,14 +351,77 @@ pub async fn dispatch_card(
|
||||
let (hb_cancel_tx, hb_cancel_rx) = tokio::sync::watch::channel(false);
|
||||
runs::spawn_heartbeat_task(location.clone(), run_id.clone(), hb_cancel_rx);
|
||||
|
||||
// Materialise this autonomous run as a top-level task-session thread so it
|
||||
// surfaces in Conversations → Tasks like a manually-run todo. Best-effort:
|
||||
// `None` just means the run streams nowhere (headless), exactly as before.
|
||||
let session_thread_id = task_session::create_session_thread(
|
||||
config.workspace_dir.clone(),
|
||||
&fresh_card,
|
||||
&run_id,
|
||||
&prompt,
|
||||
);
|
||||
|
||||
// Stamp the session thread onto the card so the board UI can offer a
|
||||
// "View session" jump into Conversations. Best-effort: a failure here just
|
||||
// means the link is unavailable; the run proceeds regardless.
|
||||
if let Some(thread_id) = session_thread_id.as_deref() {
|
||||
if let Err(e) = ops::set_session_thread(&location, &card_id, Some(thread_id.to_string())) {
|
||||
tracing::warn!(
|
||||
card_id = %card_id,
|
||||
thread_id = %thread_id,
|
||||
error = %e,
|
||||
"[task_dispatcher] failed to stamp session thread on card (View session link unavailable)"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
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;
|
||||
let _ = hb_cancel_tx.send(true);
|
||||
write_back(&location_for_run, &card_id, &run_id, outcome);
|
||||
// Clones for the active-run registry (the originals move into the task).
|
||||
let reg_thread = session_thread_id.clone();
|
||||
let reg_location = location.clone();
|
||||
let reg_card_id = card_id.clone();
|
||||
let reg_run_id = run_id.clone();
|
||||
let hb_cancel_for_task = hb_cancel_tx.clone();
|
||||
let task_thread = session_thread_id.clone();
|
||||
// Gate the task on registration: a fast-finishing run could otherwise reach
|
||||
// its terminal `take_active_run` before `register_active_run` below has run,
|
||||
// see no entry, and skip `write_back` — leaving card/run state inconsistent.
|
||||
// The task parks on `start_rx` until we release it after registration.
|
||||
let (start_tx, start_rx) = tokio::sync::oneshot::channel::<()>();
|
||||
let join = tokio::spawn(async move {
|
||||
let _ = start_rx.await;
|
||||
let outcome = run_autonomous(config, &executor, &prompt, &run_id, session_thread_id).await;
|
||||
let _ = hb_cancel_for_task.send(true);
|
||||
// Race with a concurrent cancel: whoever removes the registry entry owns
|
||||
// the write-back, so it runs exactly once. No entry (no session thread,
|
||||
// or a cancel already took it) → we skip it.
|
||||
let still_ours = match &task_thread {
|
||||
Some(tid) => take_active_run(tid).is_some(),
|
||||
None => true,
|
||||
};
|
||||
if still_ours {
|
||||
write_back(&location_for_run, &card_id, &run_id, outcome);
|
||||
}
|
||||
});
|
||||
|
||||
// Register the run so the chat Cancel (web `channel_web_cancel` →
|
||||
// `cancel_session`) can abort it — task threads aren't web-channel turns.
|
||||
if let Some(tid) = reg_thread {
|
||||
register_active_run(
|
||||
tid,
|
||||
ActiveRun {
|
||||
abort: join.abort_handle(),
|
||||
hb_cancel: hb_cancel_tx,
|
||||
location: reg_location,
|
||||
card_id: reg_card_id,
|
||||
run_id: reg_run_id,
|
||||
},
|
||||
);
|
||||
}
|
||||
// Registration (if any) is in place — release the task to start running.
|
||||
let _ = start_tx.send(());
|
||||
|
||||
Ok(DispatchOutcome::Running {
|
||||
run_id: run_id_for_return,
|
||||
})
|
||||
@@ -404,6 +548,7 @@ async fn run_autonomous(
|
||||
executor: &ResolvedExecutor,
|
||||
prompt: &str,
|
||||
run_id: &str,
|
||||
session_thread_id: Option<String>,
|
||||
) -> Result<String, String> {
|
||||
config.agent.max_tool_iterations = TASK_RUN_MAX_ITERATIONS;
|
||||
// Match skill-run egress handling: only widen to the permissive default
|
||||
@@ -427,16 +572,85 @@ async fn run_autonomous(
|
||||
run_id.get(..8).unwrap_or(run_id)
|
||||
));
|
||||
|
||||
// Stream this autonomous run into its task-session thread exactly like a
|
||||
// chat turn: wire the agent's progress into the web-channel bridge with the
|
||||
// broadcast client id "system" — the same mechanism cron/welcome agents use.
|
||||
// The bridge (a) emits live text/tool socket events that any client viewing
|
||||
// the thread renders in real time (the frontend keys by thread_id), and
|
||||
// (b) persists a TurnStateMirror so the tool timeline replays when the
|
||||
// session is opened mid/after run. Best-effort — with no session thread the
|
||||
// run is headless, exactly as before this feature.
|
||||
let workspace_dir = config.workspace_dir.clone();
|
||||
if let Some(thread_id) = session_thread_id.as_deref() {
|
||||
let (progress_tx, progress_rx) = tokio::sync::mpsc::channel(64);
|
||||
agent.set_on_progress(Some(progress_tx));
|
||||
crate::openhuman::channels::providers::web::spawn_progress_bridge(
|
||||
progress_rx,
|
||||
"system".to_string(),
|
||||
thread_id.to_string(),
|
||||
run_id.to_string(),
|
||||
crate::openhuman::threads::turn_state::TurnStateStore::new(workspace_dir.clone()),
|
||||
);
|
||||
}
|
||||
|
||||
// Sub-agent task runs are internal to the agent harness — the user
|
||||
// already authorized the parent turn that dispatched this task. Label
|
||||
// as CLI so the approval gate doesn't fail closed on internal
|
||||
// sub-agent invocations.
|
||||
crate::openhuman::agent::turn_origin::with_origin(
|
||||
let run = crate::openhuman::agent::turn_origin::with_origin(
|
||||
crate::openhuman::agent::turn_origin::AgentTurnOrigin::Cli,
|
||||
with_autonomous_iter_cap(TASK_RUN_MAX_ITERATIONS, agent.run_single(prompt)),
|
||||
)
|
||||
.await
|
||||
.map_err(|e| format!("{e:#}"))
|
||||
);
|
||||
let result = match session_thread_id.as_deref() {
|
||||
Some(thread_id) => {
|
||||
crate::openhuman::inference::provider::thread_context::with_thread_id(
|
||||
thread_id.to_string(),
|
||||
run,
|
||||
)
|
||||
.await
|
||||
}
|
||||
None => run.await,
|
||||
}
|
||||
.map_err(|e| format!("{e:#}"));
|
||||
|
||||
// Emit the terminal chat event so a client viewing the session stops
|
||||
// "processing" and finalizes the assistant bubble — the SAME chat_done /
|
||||
// chat_error the web channel emits at the end of a normal turn. The
|
||||
// progress bridge only streams intermediate deltas; without this terminal
|
||||
// signal the live-streamed session spins forever. Broadcast as "system" so
|
||||
// any viewer of the thread receives it (frontend keys by thread_id).
|
||||
if let Some(thread_id) = session_thread_id.as_deref() {
|
||||
match &result {
|
||||
Ok(response) => {
|
||||
crate::openhuman::channels::providers::presentation::deliver_response(
|
||||
"system",
|
||||
thread_id,
|
||||
run_id,
|
||||
response,
|
||||
prompt,
|
||||
&[],
|
||||
)
|
||||
.await;
|
||||
}
|
||||
Err(err) => {
|
||||
crate::openhuman::channels::providers::web::publish_web_channel_event(
|
||||
crate::core::socketio::WebChannelEvent {
|
||||
event: "chat_error".to_string(),
|
||||
client_id: "system".to_string(),
|
||||
thread_id: thread_id.to_string(),
|
||||
request_id: run_id.to_string(),
|
||||
message: Some(err.clone()),
|
||||
error_type: Some("agent_error".to_string()),
|
||||
..Default::default()
|
||||
},
|
||||
);
|
||||
}
|
||||
}
|
||||
// Persist the final response as the closing assistant message so a
|
||||
// reopened session shows the outcome like a finished manual run.
|
||||
task_session::append_final(workspace_dir, thread_id, &result);
|
||||
}
|
||||
result
|
||||
}
|
||||
|
||||
/// Deterministic board write-back: the dispatcher owns the card lifecycle.
|
||||
@@ -747,6 +961,31 @@ mod tests {
|
||||
use super::*;
|
||||
use serde_json::json;
|
||||
|
||||
#[tokio::test]
|
||||
async fn active_run_registry_take_is_once() {
|
||||
// Race-safety: the completing run and a concurrent cancel both call
|
||||
// `take_active_run`; exactly one gets `Some` (and owns the write-back).
|
||||
let (tx, _rx) = tokio::sync::watch::channel(false);
|
||||
let handle = tokio::spawn(async { std::future::pending::<()>().await });
|
||||
let key = "task-cancel-registry-test";
|
||||
register_active_run(
|
||||
key.to_string(),
|
||||
ActiveRun {
|
||||
abort: handle.abort_handle(),
|
||||
hb_cancel: tx,
|
||||
location: BoardLocation::Scratch,
|
||||
card_id: "c1".to_string(),
|
||||
run_id: "r1".to_string(),
|
||||
},
|
||||
);
|
||||
assert!(take_active_run(key).is_some(), "first take owns the run");
|
||||
assert!(
|
||||
take_active_run(key).is_none(),
|
||||
"second take gets nothing — write-back happens exactly once"
|
||||
);
|
||||
handle.abort();
|
||||
}
|
||||
|
||||
fn card(objective: Option<&str>) -> TaskBoardCard {
|
||||
TaskBoardCard {
|
||||
id: "task-1".into(),
|
||||
@@ -761,6 +1000,7 @@ mod tests {
|
||||
evidence: vec![],
|
||||
notes: None,
|
||||
blocker: None,
|
||||
session_thread_id: None,
|
||||
source_metadata: None,
|
||||
order: 0,
|
||||
updated_at: String::new(),
|
||||
|
||||
@@ -0,0 +1,251 @@
|
||||
//! Materialises an autonomous task-board run as a top-level "task session"
|
||||
//! conversation thread, so background agent work shows up in the
|
||||
//! Conversations → Tasks tab exactly like a manually-launched todo.
|
||||
//!
|
||||
//! A manually-run todo (`handleWorkPersonal` in the app) creates a top-level
|
||||
//! thread labelled `tasks` and streams the agent turn into it. Autonomous card
|
||||
//! runs (`task_dispatcher`) previously ran headless with no thread, so they
|
||||
//! never appeared in chat. This module gives them the same surface:
|
||||
//!
|
||||
//! 1. [`create_session_thread`] — a **top-level** (`parent_thread_id: None`)
|
||||
//! thread stamped `labels: ["tasks"]`, seeded with the task prompt as the
|
||||
//! opening `user` message. The frontend `isTaskThread()` predicate then
|
||||
//! lists it in the Tasks tab next to manual runs (no UI change needed).
|
||||
//! 2. [`append_final`] — appends the run's final response (or failure reason)
|
||||
//! as the closing `assistant` message after the turn completes.
|
||||
//!
|
||||
//! The *live* streaming (text/tool deltas + tool timeline) is not done here:
|
||||
//! [`task_dispatcher::run_autonomous`](super::task_dispatcher) wires the agent's
|
||||
//! `on_progress` into the web-channel `spawn_progress_bridge` with the broadcast
|
||||
//! client id `"system"`, so any client viewing the thread sees the run stream in
|
||||
//! real time — the same mechanism cron/welcome agents use.
|
||||
//!
|
||||
//! All writes are best-effort: a thread-store failure is logged and the
|
||||
//! autonomous run still proceeds headless, exactly as it did before this
|
||||
//! feature. The session is a surface over the run, never a gate on it.
|
||||
|
||||
use std::path::PathBuf;
|
||||
|
||||
use serde_json::json;
|
||||
|
||||
use crate::openhuman::agent::task_board::TaskBoardCard;
|
||||
use crate::openhuman::memory_conversations::{
|
||||
self as conversations, ConversationMessage, CreateConversationThread,
|
||||
};
|
||||
|
||||
/// Label that lands a thread in the Conversations → Tasks tab. Mirrors the
|
||||
/// frontend `TASKS_TAB_VALUE`/`LEGACY_TASK_LABELS` predicate and the label
|
||||
/// `worker_thread::create_worker_thread` stamps on delegation sub-threads.
|
||||
const TASKS_LABEL: &str = "tasks";
|
||||
|
||||
/// Max chars of a session-thread title taken from the card title.
|
||||
const TITLE_MAX_CHARS: usize = 80;
|
||||
|
||||
/// Create a top-level task-session thread for an autonomous card run and seed
|
||||
/// it with the task prompt as the opening `user` message.
|
||||
///
|
||||
/// Returns the new thread id, or `None` if the store rejected the create
|
||||
/// (best-effort: the run still proceeds headless). The `run_id`/`card_id` are
|
||||
/// stamped into the seed message metadata so the session can be correlated back
|
||||
/// to its board card and liveness run.
|
||||
pub(crate) fn create_session_thread(
|
||||
workspace_dir: PathBuf,
|
||||
card: &TaskBoardCard,
|
||||
run_id: &str,
|
||||
prompt: &str,
|
||||
) -> Option<String> {
|
||||
let thread_id = format!("task-{}", uuid::Uuid::new_v4());
|
||||
let now = chrono::Utc::now().to_rfc3339();
|
||||
|
||||
if let Err(err) = conversations::ensure_thread(
|
||||
workspace_dir.clone(),
|
||||
CreateConversationThread {
|
||||
id: thread_id.clone(),
|
||||
title: session_title(card),
|
||||
created_at: now.clone(),
|
||||
parent_thread_id: None,
|
||||
labels: Some(vec![TASKS_LABEL.to_string()]),
|
||||
personality_id: None,
|
||||
},
|
||||
) {
|
||||
tracing::warn!(
|
||||
card_id = %card.id,
|
||||
run_id = %run_id,
|
||||
error = %err,
|
||||
"[task_session] failed to create session thread (run proceeds headless)"
|
||||
);
|
||||
return None;
|
||||
}
|
||||
|
||||
if let Err(err) = conversations::append_message(
|
||||
workspace_dir,
|
||||
&thread_id,
|
||||
ConversationMessage {
|
||||
id: format!("user:{}", uuid::Uuid::new_v4()),
|
||||
content: prompt.to_string(),
|
||||
message_type: "text".to_string(),
|
||||
extra_metadata: json!({
|
||||
"scope": "autonomous_task",
|
||||
"card_id": card.id,
|
||||
"run_id": run_id,
|
||||
}),
|
||||
sender: "user".to_string(),
|
||||
created_at: now,
|
||||
},
|
||||
) {
|
||||
tracing::warn!(
|
||||
thread_id = %thread_id,
|
||||
run_id = %run_id,
|
||||
error = %err,
|
||||
"[task_session] failed to seed task prompt (continuing)"
|
||||
);
|
||||
}
|
||||
|
||||
tracing::info!(
|
||||
card_id = %card.id,
|
||||
run_id = %run_id,
|
||||
thread_id = %thread_id,
|
||||
"[task_session] created top-level task session thread for autonomous run"
|
||||
);
|
||||
Some(thread_id)
|
||||
}
|
||||
|
||||
/// Human-readable title for the session thread — the card title, trimmed and
|
||||
/// clipped, with a generic fallback for an unnamed card.
|
||||
fn session_title(card: &TaskBoardCard) -> String {
|
||||
let trimmed = card.title.trim();
|
||||
if trimmed.is_empty() {
|
||||
return "Autonomous task".to_string();
|
||||
}
|
||||
trimmed.chars().take(TITLE_MAX_CHARS).collect()
|
||||
}
|
||||
|
||||
/// Append the run's final response (or failure reason) to the session thread as
|
||||
/// the closing `assistant` message, so a reopened session shows the outcome.
|
||||
/// No-op on an empty response. Best-effort: a store failure is logged only.
|
||||
pub(crate) fn append_final(
|
||||
workspace_dir: PathBuf,
|
||||
thread_id: &str,
|
||||
outcome: &Result<String, String>,
|
||||
) {
|
||||
let (content, success) = match outcome {
|
||||
Ok(text) => (text.trim().to_string(), true),
|
||||
Err(err) => (format!("Run failed: {err}"), false),
|
||||
};
|
||||
if content.is_empty() {
|
||||
return;
|
||||
}
|
||||
if let Err(err) = conversations::append_message(
|
||||
workspace_dir,
|
||||
thread_id,
|
||||
ConversationMessage {
|
||||
id: format!("assistant:{}", uuid::Uuid::new_v4()),
|
||||
content,
|
||||
message_type: "text".to_string(),
|
||||
extra_metadata: json!({ "scope": "autonomous_task_result", "success": success }),
|
||||
sender: "assistant".to_string(),
|
||||
created_at: chrono::Utc::now().to_rfc3339(),
|
||||
},
|
||||
) {
|
||||
tracing::debug!(
|
||||
thread_id = %thread_id,
|
||||
error = %err,
|
||||
"[task_session] failed to append final response"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::openhuman::agent::task_board::{TaskBoardCard, TaskCardStatus};
|
||||
|
||||
fn card(title: &str) -> TaskBoardCard {
|
||||
TaskBoardCard {
|
||||
id: "card-1".to_string(),
|
||||
title: title.to_string(),
|
||||
status: TaskCardStatus::InProgress,
|
||||
objective: None,
|
||||
plan: Vec::new(),
|
||||
assigned_agent: None,
|
||||
allowed_tools: Vec::new(),
|
||||
approval_mode: None,
|
||||
acceptance_criteria: Vec::new(),
|
||||
evidence: Vec::new(),
|
||||
notes: None,
|
||||
blocker: None,
|
||||
session_thread_id: None,
|
||||
source_metadata: None,
|
||||
order: 0,
|
||||
updated_at: String::new(),
|
||||
}
|
||||
}
|
||||
|
||||
fn temp_ws() -> PathBuf {
|
||||
let dir = std::env::temp_dir().join(format!("task-session-test-{}", uuid::Uuid::new_v4()));
|
||||
std::fs::create_dir_all(&dir).unwrap();
|
||||
dir
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn creates_top_level_tasks_thread_and_seeds_prompt() {
|
||||
let ws = temp_ws();
|
||||
let id = create_session_thread(
|
||||
ws.clone(),
|
||||
&card("Design the onboarding"),
|
||||
"run-1",
|
||||
"Do the thing",
|
||||
)
|
||||
.expect("thread created");
|
||||
|
||||
// Top-level (no parent) + labelled `tasks` so it lands in the Tasks tab.
|
||||
let threads = conversations::list_threads(ws.clone()).expect("list threads");
|
||||
let t = threads.iter().find(|t| t.id == id).expect("thread listed");
|
||||
assert!(
|
||||
t.parent_thread_id.is_none(),
|
||||
"session thread must be top-level"
|
||||
);
|
||||
assert!(
|
||||
t.labels.iter().any(|l| l == "tasks"),
|
||||
"must carry the tasks label"
|
||||
);
|
||||
|
||||
// Seed user message carries the prompt + correlation metadata.
|
||||
let msgs = conversations::get_messages(ws, &id).expect("messages");
|
||||
assert_eq!(msgs.len(), 1);
|
||||
assert_eq!(msgs[0].sender, "user");
|
||||
assert_eq!(msgs[0].content, "Do the thing");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn append_final_writes_assistant_outcome() {
|
||||
let ws = temp_ws();
|
||||
let id = create_session_thread(ws.clone(), &card("X"), "run-2", "prompt").expect("thread");
|
||||
append_final(ws.clone(), &id, &Ok("All done.".to_string()));
|
||||
|
||||
let msgs = conversations::get_messages(ws, &id).expect("messages");
|
||||
let last = msgs.last().expect("has messages");
|
||||
assert_eq!(last.sender, "assistant");
|
||||
assert_eq!(last.content, "All done.");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn append_final_skips_empty_response() {
|
||||
let ws = temp_ws();
|
||||
let id = create_session_thread(ws.clone(), &card("X"), "run-3", "prompt").expect("thread");
|
||||
append_final(ws.clone(), &id, &Ok(" ".to_string()));
|
||||
|
||||
let msgs = conversations::get_messages(ws, &id).expect("messages");
|
||||
assert_eq!(
|
||||
msgs.len(),
|
||||
1,
|
||||
"empty final response must not append a message"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn empty_title_falls_back_to_generic_label() {
|
||||
assert_eq!(session_title(&card(" ")), "Autonomous task");
|
||||
assert_eq!(session_title(&card("Real title")), "Real title");
|
||||
}
|
||||
}
|
||||
@@ -1229,7 +1229,7 @@ async fn run_chat_task(
|
||||
/// agent turn loop and translates them into [`WebChannelEvent`]s tagged
|
||||
/// with the correct client/thread/request IDs. The task runs until the
|
||||
/// sender is dropped (i.e. when the agent turn finishes).
|
||||
fn spawn_progress_bridge(
|
||||
pub(crate) fn spawn_progress_bridge(
|
||||
mut rx: tokio::sync::mpsc::Receiver<crate::openhuman::agent::progress::AgentProgress>,
|
||||
client_id: String,
|
||||
thread_id: String,
|
||||
@@ -2101,9 +2101,19 @@ pub async fn channel_web_cancel(
|
||||
) -> Result<RpcOutcome<Value>, String> {
|
||||
let cancelled_request_id = cancel_chat(client_id, thread_id).await?;
|
||||
|
||||
// No web-channel turn for this thread → it may be an autonomous task run
|
||||
// streaming into a task session. Those are detached dispatcher tasks (not in
|
||||
// IN_FLIGHT), so cancel them via the dispatcher's registry instead — this is
|
||||
// what makes the chat Cancel button work on task threads.
|
||||
let cancelled = if cancelled_request_id.is_some() {
|
||||
true
|
||||
} else {
|
||||
crate::openhuman::agent::task_dispatcher::cancel_session(thread_id.trim()).await
|
||||
};
|
||||
|
||||
Ok(RpcOutcome::single_log(
|
||||
json!({
|
||||
"cancelled": cancelled_request_id.is_some(),
|
||||
"cancelled": cancelled,
|
||||
"client_id": client_id.trim(),
|
||||
"thread_id": thread_id.trim(),
|
||||
"request_id": cancelled_request_id,
|
||||
|
||||
@@ -263,7 +263,14 @@ impl Provider for IterativeToolProvider {
|
||||
"Completed after {completed_iterations} tool iterations."
|
||||
))
|
||||
} else {
|
||||
Ok(tool_call_payload())
|
||||
// Prefix a per-iteration progress note so each turn's assistant
|
||||
// output is distinct. A healthy multi-step agent varies its
|
||||
// narration as it advances; only byte-identical repeats (the
|
||||
// degeneration signature) should trip the harness repeat guard.
|
||||
Ok(format!(
|
||||
"Progress update {completed_iterations}.\n{}",
|
||||
tool_call_payload()
|
||||
))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -54,6 +54,72 @@ use compatible_types::{
|
||||
/// unaffected.
|
||||
const CHAT_FREQUENCY_PENALTY: f64 = 0.3;
|
||||
|
||||
/// Consecutive identical substantial lines that trip the in-generation repeat
|
||||
/// cutoff. Autoregressive models can latch onto a line and emit it verbatim
|
||||
/// until the token cap (observed: 234× the same sentence in one response).
|
||||
/// `frequency_penalty` / stronger model tiers only lower the odds — they don't
|
||||
/// prevent it — so this is the deterministic, model-agnostic stop. Set well
|
||||
/// above any legitimate repetition.
|
||||
const STREAM_REPEAT_THRESHOLD: u32 = 6;
|
||||
/// Minimum trimmed length for a line to count toward [`STREAM_REPEAT_THRESHOLD`].
|
||||
/// Keeps short, legitimately-repeated lines (`}`, blank-ish code) from tripping
|
||||
/// it; degenerate spirals are long sentences well over this.
|
||||
const MIN_REPEAT_LINE_CHARS: usize = 16;
|
||||
|
||||
/// Detects in-generation repetition degeneration on the streaming path so the
|
||||
/// reader can abort the stream and truncate the blob. Trips after
|
||||
/// [`STREAM_REPEAT_THRESHOLD`] consecutive identical substantial lines; blank
|
||||
/// separator lines are ignored, so `"sentence\n\nsentence\n\n…"` still trips.
|
||||
#[derive(Default)]
|
||||
struct StreamRepeatDetector {
|
||||
current_line: String,
|
||||
last_line: Option<String>,
|
||||
consecutive: u32,
|
||||
}
|
||||
|
||||
impl StreamRepeatDetector {
|
||||
fn new() -> Self {
|
||||
Self::default()
|
||||
}
|
||||
|
||||
/// Feed one streamed text delta. Returns `true` once the same substantial
|
||||
/// line has repeated [`STREAM_REPEAT_THRESHOLD`] times back-to-back.
|
||||
fn observe(&mut self, delta: &str) -> bool {
|
||||
for ch in delta.chars() {
|
||||
if ch == '\n' {
|
||||
if self.finalize_line() {
|
||||
return true;
|
||||
}
|
||||
} else {
|
||||
self.current_line.push(ch);
|
||||
}
|
||||
}
|
||||
false
|
||||
}
|
||||
|
||||
fn finalize_line(&mut self) -> bool {
|
||||
let line = self.current_line.trim().to_string();
|
||||
self.current_line.clear();
|
||||
if line.is_empty() {
|
||||
// Blank separator between repeats — ignore, don't reset the run.
|
||||
return false;
|
||||
}
|
||||
if line.chars().count() < MIN_REPEAT_LINE_CHARS {
|
||||
// Short line — not a degenerate-sentence repeat; reset the run.
|
||||
self.last_line = Some(line);
|
||||
self.consecutive = 1;
|
||||
return false;
|
||||
}
|
||||
if self.last_line.as_deref() == Some(line.as_str()) {
|
||||
self.consecutive += 1;
|
||||
} else {
|
||||
self.last_line = Some(line);
|
||||
self.consecutive = 1;
|
||||
}
|
||||
self.consecutive >= STREAM_REPEAT_THRESHOLD
|
||||
}
|
||||
}
|
||||
|
||||
/// A provider that speaks the OpenAI-compatible chat completions API.
|
||||
/// Used by: Venice, Vercel AI Gateway, Cloudflare AI Gateway, Moonshot,
|
||||
/// Synthetic, `OpenCode` Zen, `Z.AI`, `GLM`, `MiniMax`, Bedrock, Qianfan, Groq, Mistral, `xAI`, etc.
|
||||
@@ -1177,8 +1243,10 @@ impl OpenAiCompatibleProvider {
|
||||
|
||||
let mut bytes_stream = response.bytes_stream();
|
||||
let mut buffer = String::new();
|
||||
let mut repeat_detector = StreamRepeatDetector::new();
|
||||
let mut degenerate_repeat = false;
|
||||
|
||||
while let Some(item) = bytes_stream.next().await {
|
||||
'stream: while let Some(item) = bytes_stream.next().await {
|
||||
let bytes = item?;
|
||||
buffer.push_str(&String::from_utf8_lossy(&bytes));
|
||||
|
||||
@@ -1231,6 +1299,20 @@ impl OpenAiCompatibleProvider {
|
||||
delta: content.clone(),
|
||||
})
|
||||
.await;
|
||||
// Deterministic in-generation repeat cutoff: a
|
||||
// model spiraling on one line can't be stopped by
|
||||
// prompt or penalty, so abort the stream once the
|
||||
// same substantial line repeats too many times.
|
||||
if repeat_detector.observe(content) {
|
||||
log::warn!(
|
||||
"[stream] {} degenerate repetition detected (≥{} identical lines) — aborting generation, truncating (text_chars={})",
|
||||
self.name,
|
||||
STREAM_REPEAT_THRESHOLD,
|
||||
text_accum.chars().count(),
|
||||
);
|
||||
degenerate_repeat = true;
|
||||
break 'stream;
|
||||
}
|
||||
}
|
||||
}
|
||||
// Reasoning / thinking delta.
|
||||
@@ -1368,6 +1450,14 @@ impl OpenAiCompatibleProvider {
|
||||
}
|
||||
}
|
||||
|
||||
if degenerate_repeat {
|
||||
// Mark the truncated output so downstream (and the user) see why it
|
||||
// was cut off rather than a silently shortened response.
|
||||
text_accum.push_str(
|
||||
"\n\n[Output stopped: detected repeated/looping generation (model degeneration).]",
|
||||
);
|
||||
}
|
||||
|
||||
let tool_call_count = tool_accum.len();
|
||||
log::info!(
|
||||
"[stream] {} aggregated text_chars={} thinking_chars={} tool_calls={}",
|
||||
|
||||
@@ -2846,3 +2846,41 @@ fn convert_messages_for_native_promotes_image_marker() {
|
||||
])
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn stream_repeat_detector_trips_at_threshold() {
|
||||
let mut d = StreamRepeatDetector::new();
|
||||
// The degenerate pattern: one substantial sentence emitted with blank
|
||||
// separators, over and over (exactly what we observed, 234×).
|
||||
let chunk =
|
||||
"Now I have a complete understanding. Let me also check the llm.rs extraction logic.\n\n";
|
||||
let mut tripped_at = 0;
|
||||
for i in 1..=(STREAM_REPEAT_THRESHOLD + 3) {
|
||||
if d.observe(chunk) {
|
||||
tripped_at = i;
|
||||
break;
|
||||
}
|
||||
}
|
||||
assert_eq!(
|
||||
tripped_at, STREAM_REPEAT_THRESHOLD,
|
||||
"should trip exactly at the threshold, ignoring blank separators"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn stream_repeat_detector_ignores_varied_and_short_lines() {
|
||||
let mut d = StreamRepeatDetector::new();
|
||||
// Distinct substantial lines never trip (real, progressing output).
|
||||
for i in 0..20 {
|
||||
assert!(
|
||||
!d.observe(&format!(
|
||||
"This is distinct analysis step number {i} of the task.\n"
|
||||
)),
|
||||
"varied lines must not trip"
|
||||
);
|
||||
}
|
||||
// Short identical lines (e.g. code braces) are below the min length → no trip.
|
||||
for _ in 0..20 {
|
||||
assert!(!d.observe("}\n"), "short repeated lines must not trip");
|
||||
}
|
||||
}
|
||||
|
||||
@@ -118,6 +118,16 @@ async fn run_inner(
|
||||
if store::is_ingested(config, &source.id, &task.external_id, &hash)
|
||||
.map_err(|e| format!("dedup check failed: {e}"))?
|
||||
{
|
||||
// Dedup is scoped to THIS source (`WHERE source_id AND external_id`),
|
||||
// so the same external_id under a different source would NOT hit
|
||||
// here — it dedups per-source, never cross-source.
|
||||
tracing::debug!(
|
||||
source_id = %source.id,
|
||||
provider = %source.provider.as_str(),
|
||||
external_id = %task.external_id,
|
||||
content_hash = %hash,
|
||||
"[task_sources:dedup] skip — already ingested under THIS source with same content_hash (per-source, unchanged)"
|
||||
);
|
||||
outcome.skipped_dupe += 1;
|
||||
continue;
|
||||
}
|
||||
@@ -127,6 +137,20 @@ async fn run_inner(
|
||||
let stale_card_id = store::get_card_id(config, &source.id, &task.external_id)
|
||||
.map_err(|e| format!("get_card_id failed: {e}"))?;
|
||||
|
||||
tracing::debug!(
|
||||
source_id = %source.id,
|
||||
provider = %source.provider.as_str(),
|
||||
external_id = %task.external_id,
|
||||
content_hash = %hash,
|
||||
edited = stale_card_id.is_some(),
|
||||
"[task_sources:dedup] route — not a dupe for this source ({})",
|
||||
if stale_card_id.is_some() {
|
||||
"content changed since last ingest → re-route, replace stale card"
|
||||
} else {
|
||||
"new external_id for this source"
|
||||
}
|
||||
);
|
||||
|
||||
let enriched = enrich::enrich_task(task);
|
||||
|
||||
// Route first; only mark ingested on success so a routing
|
||||
|
||||
@@ -164,6 +164,7 @@ fn task_board_update_is_stored_and_flushed() {
|
||||
evidence: Vec::new(),
|
||||
notes: None,
|
||||
blocker: None,
|
||||
session_thread_id: None,
|
||||
source_metadata: None,
|
||||
order: 0,
|
||||
updated_at: "2026-05-15T00:00:00Z".into(),
|
||||
|
||||
@@ -293,6 +293,7 @@ pub fn add(
|
||||
evidence: patch.evidence.unwrap_or_default(),
|
||||
notes: patch.notes.and_then(non_empty),
|
||||
blocker: patch.blocker.and_then(non_empty),
|
||||
session_thread_id: None,
|
||||
source_metadata: patch.source_metadata,
|
||||
order: cards.len() as u32,
|
||||
updated_at: Utc::now().to_rfc3339(),
|
||||
@@ -365,6 +366,30 @@ pub fn edit(location: &BoardLocation, id: &str, patch: CardPatch) -> Result<Todo
|
||||
Ok(into_snapshot(location, cards))
|
||||
}
|
||||
|
||||
/// Stamp (or clear) a card's `session_thread_id` — the conversation thread of
|
||||
/// its live/last agent run — so the UI can offer a "View session" jump into
|
||||
/// Conversations. Used by the autonomous dispatcher (`task_session`, direct
|
||||
/// call) and the manual "Work" path (via the `todos_set_session_thread` RPC).
|
||||
/// A blank id clears the link. Does NOT touch status or `enforce_single_in_progress`
|
||||
/// — this is pure session-link bookkeeping, orthogonal to the card lifecycle.
|
||||
pub fn set_session_thread(
|
||||
location: &BoardLocation,
|
||||
id: &str,
|
||||
session_thread_id: Option<String>,
|
||||
) -> Result<TodosSnapshot, String> {
|
||||
let _scratch_guard = maybe_scratch_lock(location);
|
||||
let mut cards = load_cards(location)?;
|
||||
let card = cards
|
||||
.iter_mut()
|
||||
.find(|c| c.id == id)
|
||||
.ok_or_else(|| format!("todo id '{id}' not found"))?;
|
||||
card.session_thread_id = session_thread_id.and_then(non_empty);
|
||||
card.updated_at = Utc::now().to_rfc3339();
|
||||
let cards = save_cards(location, cards)?;
|
||||
emit_progress(location, &cards);
|
||||
Ok(into_snapshot(location, cards))
|
||||
}
|
||||
|
||||
/// Update only the status of a card.
|
||||
pub fn update_status(
|
||||
location: &BoardLocation,
|
||||
@@ -637,6 +662,28 @@ mod tests {
|
||||
assert!(parse_status("nope").is_err());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn set_session_thread_links_then_clears() {
|
||||
let dir = tempdir().unwrap();
|
||||
let loc = thread_loc(dir.path(), "t1");
|
||||
let snap = add(&loc, "Do the thing", CardPatch::default()).unwrap();
|
||||
let card_id = snap.cards[0].id.clone();
|
||||
|
||||
// Link a session thread → exposed on the card for the UI "View session".
|
||||
let linked = set_session_thread(&loc, &card_id, Some("thread-xyz".into())).unwrap();
|
||||
assert_eq!(
|
||||
linked.cards[0].session_thread_id.as_deref(),
|
||||
Some("thread-xyz")
|
||||
);
|
||||
|
||||
// A blank id clears the link (non_empty trims to None).
|
||||
let cleared = set_session_thread(&loc, &card_id, Some(" ".into())).unwrap();
|
||||
assert!(cleared.cards[0].session_thread_id.is_none());
|
||||
|
||||
// Unknown card id is an error, not a silent no-op.
|
||||
assert!(set_session_thread(&loc, "missing", Some("t".into())).is_err());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn add_appends_and_returns_markdown() {
|
||||
let dir = tempdir().unwrap();
|
||||
@@ -842,6 +889,7 @@ mod tests {
|
||||
evidence: Vec::new(),
|
||||
notes: None,
|
||||
blocker: None,
|
||||
session_thread_id: None,
|
||||
source_metadata: None,
|
||||
order: 0,
|
||||
updated_at: String::new(),
|
||||
@@ -859,6 +907,7 @@ mod tests {
|
||||
evidence: Vec::new(),
|
||||
notes: None,
|
||||
blocker: None,
|
||||
session_thread_id: None,
|
||||
source_metadata: None,
|
||||
order: 1,
|
||||
updated_at: String::new(),
|
||||
|
||||
@@ -20,6 +20,7 @@ pub fn all_controller_schemas() -> Vec<ControllerSchema> {
|
||||
schemas("add"),
|
||||
schemas("edit"),
|
||||
schemas("update_status"),
|
||||
schemas("set_session_thread"),
|
||||
schemas("decide_plan"),
|
||||
schemas("remove"),
|
||||
schemas("replace"),
|
||||
@@ -48,6 +49,10 @@ pub fn all_registered_controllers() -> Vec<RegisteredController> {
|
||||
schema: schemas("update_status"),
|
||||
handler: handle_update_status,
|
||||
},
|
||||
RegisteredController {
|
||||
schema: schemas("set_session_thread"),
|
||||
handler: handle_set_session_thread,
|
||||
},
|
||||
RegisteredController {
|
||||
schema: schemas("decide_plan"),
|
||||
handler: handle_decide_plan,
|
||||
@@ -115,6 +120,13 @@ pub fn schemas(function: &str) -> ControllerSchema {
|
||||
string_array_input("evidence", "Verification output, links, files, or notes."),
|
||||
optional_string("notes", "Free-text notes."),
|
||||
optional_string("blocker", "Reason the card is blocked, if any."),
|
||||
FieldSchema {
|
||||
name: "sourceMetadata",
|
||||
ty: TypeSchema::Json,
|
||||
comment: "Originating task-source identifiers ({provider, external_id, …}) \
|
||||
stamped onto a card promoted from the task-sources inbox.",
|
||||
required: false,
|
||||
},
|
||||
],
|
||||
outputs: vec![snapshot_output()],
|
||||
},
|
||||
@@ -165,6 +177,21 @@ pub fn schemas(function: &str) -> ControllerSchema {
|
||||
],
|
||||
outputs: vec![snapshot_output()],
|
||||
},
|
||||
"set_session_thread" => ControllerSchema {
|
||||
namespace: "todos",
|
||||
function: "set_session_thread",
|
||||
description: "Link a card to its agent session's conversation thread so the UI can \
|
||||
offer a \"View session\" jump. Empty/absent sessionThreadId clears the link.",
|
||||
inputs: vec![
|
||||
thread_id_input(),
|
||||
required_string("id", "Card identifier."),
|
||||
optional_string(
|
||||
"sessionThreadId",
|
||||
"Conversation thread id of the card's agent session; omit or empty to clear.",
|
||||
),
|
||||
],
|
||||
outputs: vec![snapshot_output()],
|
||||
},
|
||||
"decide_plan" => ControllerSchema {
|
||||
namespace: "todos",
|
||||
function: "decide_plan",
|
||||
@@ -299,6 +326,8 @@ struct ThreadIdParams {
|
||||
struct AddParams {
|
||||
thread_id: String,
|
||||
content: String,
|
||||
#[serde(default, alias = "sourceMetadata")]
|
||||
source_metadata: Option<Value>,
|
||||
#[serde(default)]
|
||||
status: Option<String>,
|
||||
#[serde(default)]
|
||||
@@ -365,6 +394,14 @@ struct RemoveParams {
|
||||
id: String,
|
||||
}
|
||||
|
||||
#[derive(Debug, Deserialize)]
|
||||
struct SetSessionThreadParams {
|
||||
thread_id: String,
|
||||
id: String,
|
||||
#[serde(default, alias = "sessionThreadId")]
|
||||
session_thread_id: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Deserialize)]
|
||||
struct DecidePlanParams {
|
||||
thread_id: String,
|
||||
@@ -403,7 +440,9 @@ fn handle_add(params: Map<String, Value>) -> ControllerFuture {
|
||||
evidence: p.evidence,
|
||||
notes: p.notes,
|
||||
blocker: p.blocker,
|
||||
source_metadata: None,
|
||||
// Carry the originating source identifiers onto the promoted card so
|
||||
// the inbox can tell an item was already picked up (dedup/hide).
|
||||
source_metadata: p.source_metadata,
|
||||
};
|
||||
tracing::debug!(thread_id = %p.thread_id, "[rpc][todos] add entry");
|
||||
snapshot_to_json(ops::add(&loc, &p.content, patch)?)
|
||||
@@ -449,6 +488,20 @@ fn handle_update_status(params: Map<String, Value>) -> ControllerFuture {
|
||||
})
|
||||
}
|
||||
|
||||
fn handle_set_session_thread(params: Map<String, Value>) -> ControllerFuture {
|
||||
Box::pin(async move {
|
||||
let p = parse::<SetSessionThreadParams>(params)?;
|
||||
let loc = thread_location(&p.thread_id).await?;
|
||||
tracing::debug!(
|
||||
thread_id = %p.thread_id,
|
||||
id = %p.id,
|
||||
session_thread_id = ?p.session_thread_id,
|
||||
"[rpc][todos] set_session_thread entry"
|
||||
);
|
||||
snapshot_to_json(ops::set_session_thread(&loc, &p.id, p.session_thread_id)?)
|
||||
})
|
||||
}
|
||||
|
||||
fn handle_decide_plan(params: Map<String, Value>) -> ControllerFuture {
|
||||
Box::pin(async move {
|
||||
let p = parse::<DecidePlanParams>(params)?;
|
||||
|
||||
@@ -159,6 +159,7 @@ async fn dispatcher_rejects_missing_and_stale_non_claimable_cards() {
|
||||
acceptance_criteria: vec![],
|
||||
evidence: vec![],
|
||||
notes: None,
|
||||
session_thread_id: None,
|
||||
blocker: None,
|
||||
source_metadata: None,
|
||||
order: 0,
|
||||
|
||||
@@ -1626,6 +1626,7 @@ fn agent_task_board_and_dispatcher_public_paths_cover_storage_and_prompt_shapes(
|
||||
acceptance_criteria: vec!["Focused tests pass".into()],
|
||||
evidence: vec![],
|
||||
notes: Some("Keep scope narrow".into()),
|
||||
session_thread_id: None,
|
||||
blocker: None,
|
||||
source_metadata: Some(json!({
|
||||
"provider": "github",
|
||||
@@ -1664,6 +1665,7 @@ fn agent_task_board_and_dispatcher_public_paths_cover_storage_and_prompt_shapes(
|
||||
let title_prompt = build_task_prompt(&TaskBoardCard {
|
||||
objective: Some(" ".into()),
|
||||
source_metadata: Some(json!({ "external_id": "123" })),
|
||||
session_thread_id: None,
|
||||
..loaded.cards[0].clone()
|
||||
});
|
||||
assert!(title_prompt.contains("Fallback title"));
|
||||
|
||||
@@ -3213,6 +3213,7 @@ fn turn_state_mirror_persists_progress_edges_from_public_events() {
|
||||
acceptance_criteria: Vec::new(),
|
||||
evidence: Vec::new(),
|
||||
notes: None,
|
||||
session_thread_id: None,
|
||||
blocker: None,
|
||||
source_metadata: None,
|
||||
order: 0,
|
||||
|
||||
@@ -581,6 +581,7 @@ async fn agent_task_board_store_normalizes_persists_and_surfaces_errors() {
|
||||
acceptance_criteria: vec![" tests pass ".to_string()],
|
||||
evidence: vec![" coverage measured ".to_string()],
|
||||
notes: Some(" waiting ".to_string()),
|
||||
session_thread_id: Some(" task-sess-owned ".to_string()),
|
||||
blocker: None,
|
||||
source_metadata: None,
|
||||
order: 99,
|
||||
@@ -598,6 +599,7 @@ async fn agent_task_board_store_normalizes_persists_and_surfaces_errors() {
|
||||
acceptance_criteria: Vec::new(),
|
||||
evidence: Vec::new(),
|
||||
notes: None,
|
||||
session_thread_id: None,
|
||||
blocker: None,
|
||||
source_metadata: None,
|
||||
order: 99,
|
||||
@@ -614,12 +616,22 @@ async fn agent_task_board_store_normalizes_persists_and_surfaces_errors() {
|
||||
assert_eq!(saved.cards[0].title, "Draft owned coverage");
|
||||
assert_eq!(saved.cards[0].plan, vec!["inspect", "test"]);
|
||||
assert_eq!(saved.cards[0].blocker.as_deref(), Some("waiting"));
|
||||
// A padded session_thread_id is trimmed (not dropped) on persist.
|
||||
assert_eq!(
|
||||
saved.cards[0].session_thread_id.as_deref(),
|
||||
Some("task-sess-owned")
|
||||
);
|
||||
assert_eq!(saved.cards[0].order, 0);
|
||||
|
||||
let loaded = board_for_thread(dir.path(), " thread-owned ")
|
||||
.expect("board_for_thread")
|
||||
.cards;
|
||||
assert_eq!(loaded[0].approval_mode, Some(TaskApprovalMode::Required));
|
||||
// …and the normalized value survives a reload from disk.
|
||||
assert_eq!(
|
||||
loaded[0].session_thread_id.as_deref(),
|
||||
Some("task-sess-owned")
|
||||
);
|
||||
|
||||
assert!(store.delete("thread-owned").expect("delete present"));
|
||||
assert!(!store.delete("thread-owned").expect("delete missing"));
|
||||
|
||||
Reference in New Issue
Block a user