diff --git a/app/src/components/intelligence/IntelligenceTeamsTab.test.tsx b/app/src/components/intelligence/IntelligenceTeamsTab.test.tsx index e92293980..89508ea86 100644 --- a/app/src/components/intelligence/IntelligenceTeamsTab.test.tsx +++ b/app/src/components/intelligence/IntelligenceTeamsTab.test.tsx @@ -5,13 +5,21 @@ import { type AgentTeam, agentTeamApi, type TeamView } from '../../services/api/ import IntelligenceTeamsTab from './IntelligenceTeamsTab'; vi.mock('../../services/api/agentTeamApi', () => ({ - agentTeamApi: { list: vi.fn(), get: vi.fn(), listMessages: vi.fn() }, + agentTeamApi: { + list: vi.fn(), + get: vi.fn(), + listMessages: vi.fn(), + messageMember: vi.fn(), + startMember: vi.fn(), + }, })); vi.mock('../../lib/i18n/I18nContext', () => ({ useT: () => ({ t: (k: string) => k }) })); const mockList = vi.mocked(agentTeamApi.list); const mockGet = vi.mocked(agentTeamApi.get); const mockMessages = vi.mocked(agentTeamApi.listMessages); +const mockMessageMember = vi.mocked(agentTeamApi.messageMember); +const mockStartMember = vi.mocked(agentTeamApi.startMember); function team(id: string, summary: string): AgentTeam { return { @@ -54,6 +62,25 @@ function view(t: AgentTeam): TeamView { }; } +// A view whose member is idle, so TeamHeader renders the per-member start +// control (hidden for `active` members). +function viewWithIdleMember(t: AgentTeam): TeamView { + const v = view(t); + return { + ...v, + members: [ + { + id: 'm2', + teamId: t.id, + name: 'scout', + memberStatus: 'idle', + createdAt: '2026-01-01T00:00:00Z', + updatedAt: '2026-01-01T00:00:00Z', + }, + ], + }; +} + beforeEach(() => { vi.clearAllMocks(); mockMessages.mockResolvedValue([]); @@ -188,4 +215,117 @@ describe('IntelligenceTeamsTab', () => { fireEvent.click(screen.getByText('intelligence.teams.refresh')); await waitFor(() => expect(mockList).toHaveBeenCalledTimes(2)); }); + + // ── interactive surfaces (handleSend / handleStartMember) ────────────────── + + it('sends a lead message to the whole team and refreshes the board', async () => { + const t = team('team-1', 'Ship onboarding'); + mockList.mockResolvedValue([t]); + mockGet.mockResolvedValue(view(t)); + mockMessageMember.mockResolvedValue({ + runId: 'team-1', + sequence: 1, + eventType: 'team_message', + payload: { from: 'lead', to: null, content: 'kick it off', visibility: 'team' }, + timestamp: '2026-01-01T00:00:00Z', + }); + render(); + await screen.findByText('Audit flow'); + + fireEvent.change(screen.getByPlaceholderText('intelligence.teams.composer.placeholder'), { + target: { value: 'kick it off' }, + }); + fireEvent.click(screen.getByLabelText('intelligence.teams.composer.send')); + + await waitFor(() => + expect(mockMessageMember).toHaveBeenCalledWith({ + teamId: 'team-1', + toMemberId: undefined, + content: 'kick it off', + }) + ); + // Refetched detail after the send (initial select + post-send). + await waitFor(() => expect(mockGet).toHaveBeenCalledTimes(2)); + }); + + it('surfaces a send failure as an action notice', async () => { + const t = team('team-1', 'Ship onboarding'); + mockList.mockResolvedValue([t]); + mockGet.mockResolvedValue(view(t)); + mockMessageMember.mockRejectedValue(new Error('send boom')); + render(); + await screen.findByText('Audit flow'); + + const input = screen.getByPlaceholderText( + 'intelligence.teams.composer.placeholder' + ) as HTMLInputElement; + fireEvent.change(input, { target: { value: 'hello' } }); + fireEvent.click(screen.getByLabelText('intelligence.teams.composer.send')); + + expect(await screen.findByText('send boom')).toBeInTheDocument(); + // A failed send must reject so the composer keeps the unsent draft for retry. + expect(input.value).toBe('hello'); + }); + + it('starts a member run and clears any prior notice on success', async () => { + const t = team('team-1', 'Ship onboarding'); + mockList.mockResolvedValue([t]); + mockGet.mockResolvedValue(viewWithIdleMember(t)); + mockStartMember.mockResolvedValue({ + kind: 'started', + runId: 'run-1', + task: viewWithIdleMember(t).tasks[0], + }); + render(); + await screen.findByText('Audit flow'); + + fireEvent.click(screen.getByLabelText('intelligence.teams.member.start scout')); + + await waitFor(() => + expect(mockStartMember).toHaveBeenCalledWith({ teamId: 'team-1', memberId: 'm2' }) + ); + // `started` shows no notice. + expect(screen.queryByText('intelligence.teams.action.alreadyActive')).not.toBeInTheDocument(); + }); + + it('names the unmet dependencies when a start is blocked', async () => { + const t = team('team-1', 'Ship onboarding'); + mockList.mockResolvedValue([t]); + mockGet.mockResolvedValue(viewWithIdleMember(t)); + mockStartMember.mockResolvedValue({ kind: 'blocked', unmet: ['task-a', 'task-b'] }); + render(); + await screen.findByText('Audit flow'); + + fireEvent.click(screen.getByLabelText('intelligence.teams.member.start scout')); + + expect( + await screen.findByText('intelligence.teams.action.blocked: task-a, task-b') + ).toBeInTheDocument(); + }); + + it('surfaces an already-active outcome as a notice', async () => { + const t = team('team-1', 'Ship onboarding'); + mockList.mockResolvedValue([t]); + mockGet.mockResolvedValue(viewWithIdleMember(t)); + mockStartMember.mockResolvedValue({ kind: 'alreadyActive' }); + render(); + await screen.findByText('Audit flow'); + + fireEvent.click(screen.getByLabelText('intelligence.teams.member.start scout')); + + expect(await screen.findByText('intelligence.teams.action.alreadyActive')).toBeInTheDocument(); + }); + + it('surfaces a start failure as an action notice', async () => { + const t = team('team-1', 'Ship onboarding'); + mockList.mockResolvedValue([t]); + mockGet.mockResolvedValue(viewWithIdleMember(t)); + mockStartMember.mockRejectedValue(new Error('start boom')); + render(); + await screen.findByText('Audit flow'); + + fireEvent.click(screen.getByLabelText('intelligence.teams.member.start scout')); + + expect(await screen.findByText('start boom')).toBeInTheDocument(); + }); }); diff --git a/app/src/components/intelligence/IntelligenceTeamsTab.tsx b/app/src/components/intelligence/IntelligenceTeamsTab.tsx index db1671304..deffea284 100644 --- a/app/src/components/intelligence/IntelligenceTeamsTab.tsx +++ b/app/src/components/intelligence/IntelligenceTeamsTab.tsx @@ -1,12 +1,16 @@ /** - * IntelligenceTeamsTab — the agent-team coordination surface (#3374 PR2). + * IntelligenceTeamsTab — the agent-team coordination surface (#3374). * - * Read-only window onto the durable team ledger (PR1, #3546). It lists the - * teams an agent has spawned, and — for a selected team — renders treatment "A": - * a {@link TeamHeader} identity strip, the {@link TeamTaskBoard} of owned tasks, - * and an always-visible {@link TeamActivityRail} of teammate messages. The - * board/rail sit in a grid that collapses the rail under the board on narrow - * widths. + * A window onto the durable team ledger (PR1, #3546). It lists the teams an + * agent has spawned, and — for a selected team — renders a {@link TeamHeader} + * identity strip, the {@link TeamTaskBoard} of owned tasks, and an + * always-visible {@link TeamActivityRail} of teammate messages. The board/rail + * sit in a grid that collapses the rail under the board on narrow widths. + * + * PR4 makes it interactive on an active team: the lead/user can message a named + * teammate (composer in the rail) and start a teammate live on its next + * claimable task (the play affordance on each member chip), both wired through + * {@link agentTeamApi}. A non-`started` outcome surfaces as a transient notice. * * Most users will see the EMPTY state first: a team only exists once an agent * spawns coordinated sub-agents, so "no teams yet" is the common first view and @@ -49,6 +53,11 @@ export default function IntelligenceTeamsTab() { const [loading, setLoading] = useState(true); const [error, setError] = useState(null); const [refreshing, setRefreshing] = useState(false); + const [sending, setSending] = useState(false); + const [startingMemberId, setStartingMemberId] = useState(null); + // Transient feedback for a member-start outcome (blocked / already claimed / + // nothing claimable) or a send/start error — distinct from the fatal `error`. + const [actionNotice, setActionNotice] = useState(null); const mountedRef = useRef(true); // Mirrors `view` for reads inside the poll interval without making the poll // effect depend on `view` (which would rebuild the interval every tick). @@ -107,6 +116,18 @@ export default function IntelligenceTeamsTab() { setView(null); viewRef.current = null; setMessages([]); + // Reset transient per-team feedback so a notice from the team we just left + // doesn't leak onto the next team opened. + setActionNotice(null); + setStartingMemberId(null); + }, []); + + // Select a team, clearing any transient notice from a previously viewed team + // so stale feedback doesn't carry across navigation. + const selectTeam = useCallback((teamId: string) => { + setActionNotice(null); + setStartingMemberId(null); + setSelectedId(teamId); }, []); // Mount fetch of the team list (mirrors IntelligenceAgentWorkTab's 0ms @@ -157,6 +178,65 @@ export default function IntelligenceTeamsTab() { } }, [selectedId, fetchDetail, fetchTeams]); + // Send a message as the lead (fromMemberId omitted) to a teammate or the + // whole team, then refresh so the new message appears in the rail. + const handleSend = useCallback( + async (toMemberId: string | null, content: string) => { + if (!selectedId) return; + setSending(true); + setActionNotice(null); + try { + await agentTeamApi.messageMember({ + teamId: selectedId, + toMemberId: toMemberId ?? undefined, + content, + }); + await fetchDetail(selectedId); + } catch (err) { + if (mountedRef.current) { + setActionNotice(err instanceof Error ? err.message : String(err)); + } + // Re-throw so the composer's onSend rejects and keeps the unsent draft. + // Swallowing here would resolve handleSend as success and TeamActivityRail + // would clear the draft text on a failed send. + throw err; + } finally { + if (mountedRef.current) setSending(false); + } + }, + [selectedId, fetchDetail] + ); + + // Start a live run for a member (auto-picks the member's next claimable task). + // A non-`started` outcome is surfaced as a transient notice; either way we + // refresh so the board/roster reflect the new state. + const handleStartMember = useCallback( + async (memberId: string) => { + if (!selectedId) return; + setStartingMemberId(memberId); + setActionNotice(null); + try { + const outcome = await agentTeamApi.startMember({ teamId: selectedId, memberId }); + if (outcome.kind === 'blocked') { + // Name the unmet dependency ids so the user knows what to finish + // first (t() does not interpolate — compose at the call site). + const base = t('intelligence.teams.action.blocked'); + setActionNotice(outcome.unmet.length ? `${base}: ${outcome.unmet.join(', ')}` : base); + } else if (outcome.kind !== 'started') { + setActionNotice(t(`intelligence.teams.action.${outcome.kind}`)); + } + await fetchDetail(selectedId); + } catch (err) { + if (mountedRef.current) { + setActionNotice(err instanceof Error ? err.message : String(err)); + } + } finally { + if (mountedRef.current) setStartingMemberId(null); + } + }, + [selectedId, fetchDetail, t] + ); + if (loading) { return (
@@ -215,11 +295,30 @@ export default function IntelligenceTeamsTab() { void refresh()} t={t} />
- + void handleStartMember(memberId) : undefined + } + startingMemberId={startingMemberId} + /> + + {actionNotice && ( +
+ {actionNotice} +
+ )}
- +
); @@ -239,7 +338,7 @@ export default function IntelligenceTeamsTab() {
  • + + + )} ); } diff --git a/app/src/components/intelligence/TeamHeader.test.tsx b/app/src/components/intelligence/TeamHeader.test.tsx new file mode 100644 index 000000000..58e95ccc8 --- /dev/null +++ b/app/src/components/intelligence/TeamHeader.test.tsx @@ -0,0 +1,69 @@ +import { fireEvent, render, screen } from '@testing-library/react'; +import { describe, expect, it, vi } from 'vitest'; + +import type { AgentTeam, AgentTeamMember } from '../../services/api/agentTeamApi'; +import { TeamHeader } from './TeamHeader'; + +vi.mock('../../lib/i18n/I18nContext', () => ({ useT: () => ({ t: (k: string) => k }) })); + +const team: AgentTeam = { + id: 'team-1', + leadAgentId: 'lead', + status: 'active', + summary: 'ship it', + createdAt: '2026-01-01T00:00:00Z', + updatedAt: '2026-01-01T00:00:00Z', +}; + +function member( + id: string, + name: string, + status: AgentTeamMember['memberStatus'] +): AgentTeamMember { + return { + id, + teamId: 'team-1', + name, + memberStatus: status, + createdAt: '2026-01-01T00:00:00Z', + updatedAt: '2026-01-01T00:00:00Z', + }; +} + +describe('TeamHeader', () => { + it('renders no start affordance without onStartMember', () => { + render(); + expect(screen.queryByLabelText(/intelligence.teams.member.start/)).not.toBeInTheDocument(); + }); + + it('starts an idle member and omits the button on an active member', () => { + const onStart = vi.fn(); + render( + + ); + // Only the idle member exposes a start button. + const buttons = screen.getAllByLabelText(/intelligence.teams.member.start/); + expect(buttons).toHaveLength(1); + fireEvent.click(buttons[0]); + expect(onStart).toHaveBeenCalledWith('m1'); + }); + + it('disables the start button for the member being dispatched', () => { + const onStart = vi.fn(); + render( + + ); + expect(screen.getByLabelText(/intelligence.teams.member.start/)).toBeDisabled(); + }); +}); diff --git a/app/src/components/intelligence/TeamHeader.tsx b/app/src/components/intelligence/TeamHeader.tsx index e435b9f7a..0ed7fba65 100644 --- a/app/src/components/intelligence/TeamHeader.tsx +++ b/app/src/components/intelligence/TeamHeader.tsx @@ -9,6 +9,8 @@ * (≈2–5 members) a header strip is the right density — a full sidebar would be * heavy furniture for three chips. */ +import { LuLoaderCircle, LuPlay } from 'react-icons/lu'; + import { useT } from '../../lib/i18n/I18nContext'; import type { AgentTeam, @@ -36,9 +38,19 @@ interface TeamHeaderProps { team: AgentTeam; members: AgentTeamMember[]; taskCount: number; + /** When provided, renders a "Start" affordance on each non-active member. */ + onStartMember?: (memberId: string) => void; + /** Id of the member whose live run is currently being dispatched. */ + startingMemberId?: string | null; } -export function TeamHeader({ team, members, taskCount }: TeamHeaderProps) { +export function TeamHeader({ + team, + members, + taskCount, + onStartMember, + startingMemberId, +}: TeamHeaderProps) { const { t } = useT(); const title = team.summary?.trim() || team.leadAgentId; @@ -74,6 +86,21 @@ export function TeamHeader({ team, members, taskCount }: TeamHeaderProps) { + {onStartMember && member.memberStatus !== 'active' && ( + + )} ))} diff --git a/app/src/lib/i18n/ar.ts b/app/src/lib/i18n/ar.ts index 9b71e5c23..970c3cfa7 100644 --- a/app/src/lib/i18n/ar.ts +++ b/app/src/lib/i18n/ar.ts @@ -3018,6 +3018,16 @@ const messages: TranslationMap = { 'intelligence.teams.activity.title': 'نشاط الفريق', 'intelligence.teams.activity.empty': 'لا توجد رسائل بعد', 'intelligence.teams.activity.toTeam': 'الفريق', + 'intelligence.teams.member.start': 'ابدأ', + 'intelligence.teams.composer.placeholder': 'رسالة إلى زميل…', + 'intelligence.teams.composer.send': 'إرسال', + 'intelligence.teams.composer.recipient': 'المستلم', + 'intelligence.teams.composer.toTeam': 'الفريق بأكمله', + 'intelligence.teams.action.blocked': 'في انتظار التبعيات', + 'intelligence.teams.action.alreadyClaimed': 'تم أخذها من قبل زميل آخر', + 'intelligence.teams.action.alreadyActive': 'الزميل قيد التشغيل بالفعل', + 'intelligence.teams.action.noClaimableTask': 'لا توجد مهمة جاهزة لهذا الزميل', + 'intelligence.teams.action.unknownTask': 'المهمة غير موجودة', 'intelligence.refine.objectiveDefault': 'حوّل المهمة المصدر إلى مهمة وكيل جاهزة للتنفيذ: {title}', 'intelligence.refine.sourceLine': 'المصدر: {url}', 'intelligence.refine.sourceIntake': 'المصدر: استقبال مصدر المهام', diff --git a/app/src/lib/i18n/bn.ts b/app/src/lib/i18n/bn.ts index a745ab5c2..16161ec8f 100644 --- a/app/src/lib/i18n/bn.ts +++ b/app/src/lib/i18n/bn.ts @@ -3080,6 +3080,16 @@ const messages: TranslationMap = { 'intelligence.teams.activity.title': 'টিম কার্যকলাপ', 'intelligence.teams.activity.empty': 'এখনো কোনো বার্তা নেই', 'intelligence.teams.activity.toTeam': 'টিম', + 'intelligence.teams.member.start': 'শুরু', + 'intelligence.teams.composer.placeholder': 'সতীর্থকে বার্তা…', + 'intelligence.teams.composer.send': 'পাঠান', + 'intelligence.teams.composer.recipient': 'প্রাপক', + 'intelligence.teams.composer.toTeam': 'পুরো দল', + 'intelligence.teams.action.blocked': 'নির্ভরতার অপেক্ষায়', + 'intelligence.teams.action.alreadyClaimed': 'অন্য সতীর্থ ইতিমধ্যে নিয়েছে', + 'intelligence.teams.action.alreadyActive': 'সহকর্মী ইতিমধ্যে সক্রিয়', + 'intelligence.teams.action.noClaimableTask': 'এই সতীর্থের জন্য কোনো কাজ প্রস্তুত নেই', + 'intelligence.teams.action.unknownTask': 'কাজ পাওয়া যায়নি', 'intelligence.refine.objectiveDefault': 'উৎস টাস্কটিকে বাস্তবায়নের জন্য প্রস্তুত একটি এজেন্ট টাস্কে রূপান্তর করুন: {title}', 'intelligence.refine.sourceLine': 'উৎস: {url}', diff --git a/app/src/lib/i18n/de.ts b/app/src/lib/i18n/de.ts index cba6bbabe..a738b9701 100644 --- a/app/src/lib/i18n/de.ts +++ b/app/src/lib/i18n/de.ts @@ -3153,6 +3153,16 @@ const messages: TranslationMap = { 'intelligence.teams.activity.title': 'Teamaktivität', 'intelligence.teams.activity.empty': 'Noch keine Nachrichten', 'intelligence.teams.activity.toTeam': 'Team', + 'intelligence.teams.member.start': 'Starten', + 'intelligence.teams.composer.placeholder': 'Einem Teammitglied schreiben…', + 'intelligence.teams.composer.send': 'Senden', + 'intelligence.teams.composer.recipient': 'Empfänger', + 'intelligence.teams.composer.toTeam': 'Gesamtes Team', + 'intelligence.teams.action.blocked': 'Wartet auf Abhängigkeiten', + 'intelligence.teams.action.alreadyClaimed': 'Bereits von einem anderen Teammitglied übernommen', + 'intelligence.teams.action.alreadyActive': 'Teammitglied läuft bereits', + 'intelligence.teams.action.noClaimableTask': 'Keine Aufgabe für dieses Teammitglied bereit', + 'intelligence.teams.action.unknownTask': 'Aufgabe nicht gefunden', 'intelligence.refine.objectiveDefault': 'Verwandle die Quellaufgabe in eine umsetzungsbereite Agentenaufgabe: {title}', 'intelligence.refine.sourceLine': 'Quelle: {url}', diff --git a/app/src/lib/i18n/en.ts b/app/src/lib/i18n/en.ts index d90c199e0..c133760a8 100644 --- a/app/src/lib/i18n/en.ts +++ b/app/src/lib/i18n/en.ts @@ -3670,6 +3670,16 @@ const en: TranslationMap = { 'intelligence.teams.activity.title': 'Team activity', 'intelligence.teams.activity.empty': 'No messages yet', 'intelligence.teams.activity.toTeam': 'team', + 'intelligence.teams.member.start': 'Start', + 'intelligence.teams.composer.placeholder': 'Message a teammate…', + 'intelligence.teams.composer.send': 'Send', + 'intelligence.teams.composer.recipient': 'Recipient', + 'intelligence.teams.composer.toTeam': 'Whole team', + 'intelligence.teams.action.blocked': 'Waiting on dependencies', + 'intelligence.teams.action.alreadyClaimed': 'Already claimed by another teammate', + 'intelligence.teams.action.alreadyActive': 'Teammate is already running', + 'intelligence.teams.action.noClaimableTask': 'No task ready for this teammate', + 'intelligence.teams.action.unknownTask': 'Task not found', 'intelligence.refine.objectiveDefault': 'Turn the source task into an implementation-ready agent task: {title}', 'intelligence.refine.sourceLine': 'Source: {url}', diff --git a/app/src/lib/i18n/es.ts b/app/src/lib/i18n/es.ts index 7958e6ce1..21daa7c57 100644 --- a/app/src/lib/i18n/es.ts +++ b/app/src/lib/i18n/es.ts @@ -3136,6 +3136,16 @@ const messages: TranslationMap = { 'intelligence.teams.activity.title': 'Actividad del equipo', 'intelligence.teams.activity.empty': 'Aún no hay mensajes', 'intelligence.teams.activity.toTeam': 'equipo', + 'intelligence.teams.member.start': 'Iniciar', + 'intelligence.teams.composer.placeholder': 'Mensaje a un compañero…', + 'intelligence.teams.composer.send': 'Enviar', + 'intelligence.teams.composer.recipient': 'Destinatario', + 'intelligence.teams.composer.toTeam': 'Todo el equipo', + 'intelligence.teams.action.blocked': 'Esperando dependencias', + 'intelligence.teams.action.alreadyClaimed': 'Ya reclamada por otro compañero', + 'intelligence.teams.action.alreadyActive': 'El compañero ya está en ejecución', + 'intelligence.teams.action.noClaimableTask': 'No hay tarea lista para este compañero', + 'intelligence.teams.action.unknownTask': 'Tarea no encontrada', 'intelligence.refine.objectiveDefault': 'Convierte la tarea de origen en una tarea de agente lista para implementar: {title}', 'intelligence.refine.sourceLine': 'Origen: {url}', diff --git a/app/src/lib/i18n/fr.ts b/app/src/lib/i18n/fr.ts index 5ecdfb5c2..3af36809d 100644 --- a/app/src/lib/i18n/fr.ts +++ b/app/src/lib/i18n/fr.ts @@ -3151,6 +3151,16 @@ const messages: TranslationMap = { 'intelligence.teams.activity.title': "Activité de l'équipe", 'intelligence.teams.activity.empty': "Aucun message pour l'instant", 'intelligence.teams.activity.toTeam': 'équipe', + 'intelligence.teams.member.start': 'Démarrer', + 'intelligence.teams.composer.placeholder': 'Message à un coéquipier…', + 'intelligence.teams.composer.send': 'Envoyer', + 'intelligence.teams.composer.recipient': 'Destinataire', + 'intelligence.teams.composer.toTeam': "Toute l'équipe", + 'intelligence.teams.action.blocked': 'En attente de dépendances', + 'intelligence.teams.action.alreadyClaimed': 'Déjà prise par un autre coéquipier', + 'intelligence.teams.action.alreadyActive': 'Le coéquipier est déjà en cours d’exécution', + 'intelligence.teams.action.noClaimableTask': 'Aucune tâche prête pour ce coéquipier', + 'intelligence.teams.action.unknownTask': 'Tâche introuvable', 'intelligence.refine.sourceIntake': 'Source : réception des sources de tâches', 'intelligence.refine.repositoryLine': 'Dépôt : {repo}', 'intelligence.refine.externalTaskLine': 'Tâche externe : {externalId}', diff --git a/app/src/lib/i18n/hi.ts b/app/src/lib/i18n/hi.ts index 65473aa9a..a71ec8d30 100644 --- a/app/src/lib/i18n/hi.ts +++ b/app/src/lib/i18n/hi.ts @@ -3081,6 +3081,16 @@ const messages: TranslationMap = { 'intelligence.teams.activity.title': 'टीम गतिविधि', 'intelligence.teams.activity.empty': 'अभी कोई संदेश नहीं', 'intelligence.teams.activity.toTeam': 'टीम', + 'intelligence.teams.member.start': 'शुरू करें', + 'intelligence.teams.composer.placeholder': 'साथी को संदेश…', + 'intelligence.teams.composer.send': 'भेजें', + 'intelligence.teams.composer.recipient': 'प्राप्तकर्ता', + 'intelligence.teams.composer.toTeam': 'पूरी टीम', + 'intelligence.teams.action.blocked': 'निर्भरताओं की प्रतीक्षा', + 'intelligence.teams.action.alreadyClaimed': 'किसी अन्य साथी ने पहले ही ले लिया', + 'intelligence.teams.action.alreadyActive': 'साथी पहले से चल रहा है', + 'intelligence.teams.action.noClaimableTask': 'इस साथी के लिए कोई कार्य तैयार नहीं', + 'intelligence.teams.action.unknownTask': 'कार्य नहीं मिला', 'intelligence.refine.objectiveDefault': 'स्रोत कार्य को कार्यान्वयन के लिए तैयार एजेंट कार्य में बदलें: {title}', 'intelligence.refine.sourceLine': 'स्रोत: {url}', diff --git a/app/src/lib/i18n/id.ts b/app/src/lib/i18n/id.ts index 21c5d8aec..914e5a507 100644 --- a/app/src/lib/i18n/id.ts +++ b/app/src/lib/i18n/id.ts @@ -3085,6 +3085,16 @@ const messages: TranslationMap = { 'intelligence.teams.activity.title': 'Aktivitas tim', 'intelligence.teams.activity.empty': 'Belum ada pesan', 'intelligence.teams.activity.toTeam': 'tim', + 'intelligence.teams.member.start': 'Mulai', + 'intelligence.teams.composer.placeholder': 'Pesan ke rekan tim…', + 'intelligence.teams.composer.send': 'Kirim', + 'intelligence.teams.composer.recipient': 'Penerima', + 'intelligence.teams.composer.toTeam': 'Seluruh tim', + 'intelligence.teams.action.blocked': 'Menunggu dependensi', + 'intelligence.teams.action.alreadyClaimed': 'Sudah diambil rekan tim lain', + 'intelligence.teams.action.alreadyActive': 'Rekan tim sudah berjalan', + 'intelligence.teams.action.noClaimableTask': 'Tidak ada tugas siap untuk rekan ini', + 'intelligence.teams.action.unknownTask': 'Tugas tidak ditemukan', 'intelligence.refine.objectiveDefault': 'Ubah tugas sumber menjadi tugas agen yang siap diimplementasikan: {title}', 'intelligence.refine.sourceLine': 'Sumber: {url}', diff --git a/app/src/lib/i18n/it.ts b/app/src/lib/i18n/it.ts index c85557062..39b10afbc 100644 --- a/app/src/lib/i18n/it.ts +++ b/app/src/lib/i18n/it.ts @@ -3128,6 +3128,16 @@ const messages: TranslationMap = { 'intelligence.teams.activity.title': 'Attività del team', 'intelligence.teams.activity.empty': 'Nessun messaggio', 'intelligence.teams.activity.toTeam': 'team', + 'intelligence.teams.member.start': 'Avvia', + 'intelligence.teams.composer.placeholder': 'Messaggio a un compagno…', + 'intelligence.teams.composer.send': 'Invia', + 'intelligence.teams.composer.recipient': 'Destinatario', + 'intelligence.teams.composer.toTeam': 'Tutto il team', + 'intelligence.teams.action.blocked': 'In attesa di dipendenze', + 'intelligence.teams.action.alreadyClaimed': 'Già presa da un altro compagno', + 'intelligence.teams.action.alreadyActive': 'Il compagno è già in esecuzione', + 'intelligence.teams.action.noClaimableTask': 'Nessuna attività pronta per questo compagno', + 'intelligence.teams.action.unknownTask': 'Attività non trovata', 'intelligence.refine.objectiveDefault': "Trasforma l'attività di origine in un'attività dell'agente pronta per l'implementazione: {title}", 'intelligence.refine.sourceLine': 'Origine: {url}', diff --git a/app/src/lib/i18n/ko.ts b/app/src/lib/i18n/ko.ts index 28152a9fe..00daf7049 100644 --- a/app/src/lib/i18n/ko.ts +++ b/app/src/lib/i18n/ko.ts @@ -3057,6 +3057,16 @@ const messages: TranslationMap = { 'intelligence.teams.activity.title': '팀 활동', 'intelligence.teams.activity.empty': '아직 메시지가 없습니다', 'intelligence.teams.activity.toTeam': '팀', + 'intelligence.teams.member.start': '시작', + 'intelligence.teams.composer.placeholder': '메시지를 입력하세요…', + 'intelligence.teams.composer.send': '보내기', + 'intelligence.teams.composer.recipient': '받는 사람', + 'intelligence.teams.composer.toTeam': '전체 팀', + 'intelligence.teams.action.blocked': '종속성 대기 중', + 'intelligence.teams.action.alreadyClaimed': '다른 팀원이 이미 맡고 있습니다.', + 'intelligence.teams.action.alreadyActive': '팀원이 이미 실행 중입니다', + 'intelligence.teams.action.noClaimableTask': '이 팀원이 맡을 작업이 없습니다.', + 'intelligence.teams.action.unknownTask': '작업을 찾을 수 없음', 'intelligence.refine.objectiveDefault': '소스 작업을 구현 준비가 된 에이전트 작업으로 전환하세요: {title}', 'intelligence.refine.sourceLine': '소스: {url}', diff --git a/app/src/lib/i18n/pl.ts b/app/src/lib/i18n/pl.ts index 308171e53..ba9f738fc 100644 --- a/app/src/lib/i18n/pl.ts +++ b/app/src/lib/i18n/pl.ts @@ -3119,6 +3119,16 @@ const messages: TranslationMap = { 'intelligence.teams.activity.title': 'Aktywność zespołu', 'intelligence.teams.activity.empty': 'Brak wiadomości', 'intelligence.teams.activity.toTeam': 'zespół', + 'intelligence.teams.member.start': 'Uruchom', + 'intelligence.teams.composer.placeholder': 'Wiadomość do członka zespołu…', + 'intelligence.teams.composer.send': 'Wyślij', + 'intelligence.teams.composer.recipient': 'Odbiorca', + 'intelligence.teams.composer.toTeam': 'Cały zespół', + 'intelligence.teams.action.blocked': 'Oczekiwanie na zależności', + 'intelligence.teams.action.alreadyClaimed': 'Już przejęte przez innego członka zespołu', + 'intelligence.teams.action.alreadyActive': 'Członek zespołu już działa', + 'intelligence.teams.action.noClaimableTask': 'Brak gotowego zadania dla tego członka zespołu', + 'intelligence.teams.action.unknownTask': 'Nie znaleziono zadania', 'intelligence.refine.objectiveDefault': 'Przekształć zadanie źródłowe w gotowe do wdrożenia zadanie agenta: {title}', 'intelligence.refine.sourceLine': 'Źródło: {url}', diff --git a/app/src/lib/i18n/pt.ts b/app/src/lib/i18n/pt.ts index 534b51749..403bd5abb 100644 --- a/app/src/lib/i18n/pt.ts +++ b/app/src/lib/i18n/pt.ts @@ -3132,6 +3132,16 @@ const messages: TranslationMap = { 'intelligence.teams.activity.title': 'Atividade da equipe', 'intelligence.teams.activity.empty': 'Nenhuma mensagem ainda', 'intelligence.teams.activity.toTeam': 'equipe', + 'intelligence.teams.member.start': 'Iniciar', + 'intelligence.teams.composer.placeholder': 'Mensagem para um colega…', + 'intelligence.teams.composer.send': 'Enviar', + 'intelligence.teams.composer.recipient': 'Destinatário', + 'intelligence.teams.composer.toTeam': 'Toda a equipe', + 'intelligence.teams.action.blocked': 'Aguardando dependências', + 'intelligence.teams.action.alreadyClaimed': 'Já reivindicada por outro colega', + 'intelligence.teams.action.alreadyActive': 'O colega já está em execução', + 'intelligence.teams.action.noClaimableTask': 'Nenhuma tarefa pronta para este colega', + 'intelligence.teams.action.unknownTask': 'Tarefa não encontrada', 'intelligence.refine.objectiveDefault': 'Transforme a tarefa de origem em uma tarefa de agente pronta para implementação: {title}', 'intelligence.refine.sourceLine': 'Origem: {url}', diff --git a/app/src/lib/i18n/ru.ts b/app/src/lib/i18n/ru.ts index 3d96682c2..7a108e048 100644 --- a/app/src/lib/i18n/ru.ts +++ b/app/src/lib/i18n/ru.ts @@ -3107,6 +3107,16 @@ const messages: TranslationMap = { 'intelligence.teams.activity.title': 'Активность команды', 'intelligence.teams.activity.empty': 'Сообщений пока нет', 'intelligence.teams.activity.toTeam': 'команда', + 'intelligence.teams.member.start': 'Запустить', + 'intelligence.teams.composer.placeholder': 'Сообщение участнику…', + 'intelligence.teams.composer.send': 'Отправить', + 'intelligence.teams.composer.recipient': 'Получатель', + 'intelligence.teams.composer.toTeam': 'Вся команда', + 'intelligence.teams.action.blocked': 'Ожидание зависимостей', + 'intelligence.teams.action.alreadyClaimed': 'Уже взято другим участником', + 'intelligence.teams.action.alreadyActive': 'Участник уже выполняется', + 'intelligence.teams.action.noClaimableTask': 'Нет готовой задачи для этого участника', + 'intelligence.teams.action.unknownTask': 'Задача не найдена', 'intelligence.refine.objectiveDefault': 'Превратите исходную задачу в готовую к реализации задачу агента: {title}', 'intelligence.refine.sourceLine': 'Источник: {url}', diff --git a/app/src/lib/i18n/zh-CN.ts b/app/src/lib/i18n/zh-CN.ts index ec4d2ef78..302345968 100644 --- a/app/src/lib/i18n/zh-CN.ts +++ b/app/src/lib/i18n/zh-CN.ts @@ -2933,6 +2933,16 @@ const messages: TranslationMap = { 'intelligence.teams.activity.title': '团队动态', 'intelligence.teams.activity.empty': '暂无消息', 'intelligence.teams.activity.toTeam': '团队', + 'intelligence.teams.member.start': '启动', + 'intelligence.teams.composer.placeholder': '给队友发消息…', + 'intelligence.teams.composer.send': '发送', + 'intelligence.teams.composer.recipient': '接收者', + 'intelligence.teams.composer.toTeam': '整个团队', + 'intelligence.teams.action.blocked': '正在等待依赖', + 'intelligence.teams.action.alreadyClaimed': '已被其他队友认领', + 'intelligence.teams.action.alreadyActive': '队友已在运行中', + 'intelligence.teams.action.noClaimableTask': '没有可供该队友处理的任务', + 'intelligence.teams.action.unknownTask': '未找到任务', 'intelligence.refine.objectiveDefault': '将来源任务转化为可立即实施的智能体任务:{title}', 'intelligence.refine.sourceLine': '来源:{url}', 'intelligence.refine.sourceIntake': '来源:任务来源接收', diff --git a/app/src/services/api/agentTeamApi.test.ts b/app/src/services/api/agentTeamApi.test.ts index ab9eca54b..e73e97b9c 100644 --- a/app/src/services/api/agentTeamApi.test.ts +++ b/app/src/services/api/agentTeamApi.test.ts @@ -198,3 +198,98 @@ describe('agentTeamApi.shutdownMember', () => { expect(mockCall).not.toHaveBeenCalled(); }); }); + +describe('agentTeamApi.messageMember', () => { + it('narrows the { message } event and omits absent optional fields', async () => { + mockCall.mockResolvedValueOnce({ + message: { + runId: 'team-1', + sequence: 5, + eventType: 'team_message', + payload: { from: 'lead', to: 'm1', content: 'go', visibility: 'team' }, + timestamp: '2026-01-01T00:00:00Z', + }, + }); + const msg = await agentTeamApi.messageMember({ + teamId: 'team-1', + toMemberId: 'm1', + content: 'go', + }); + expect(msg.payload).toEqual({ from: 'lead', to: 'm1', content: 'go', visibility: 'team' }); + // fromMemberId omitted → lead origin; no empty keys forwarded. + expect(mockCall).toHaveBeenCalledWith({ + method: 'openhuman.agent_team_message_member', + params: { teamId: 'team-1', content: 'go', toMemberId: 'm1' }, + }); + }); + + it('forwards fromMemberId for teammate-to-teammate messages', async () => { + mockCall.mockResolvedValueOnce({ + message: { + runId: 'team-1', + sequence: 6, + eventType: 'team_message', + payload: { from: 'm1', to: 'm2', content: 'hi', visibility: 'team' }, + timestamp: '2026-01-01T00:00:00Z', + }, + }); + await agentTeamApi.messageMember({ + teamId: 'team-1', + fromMemberId: 'm1', + toMemberId: 'm2', + content: 'hi', + }); + expect(mockCall).toHaveBeenCalledWith({ + method: 'openhuman.agent_team_message_member', + params: { teamId: 'team-1', content: 'hi', fromMemberId: 'm1', toMemberId: 'm2' }, + }); + }); + + it('throws on empty content or teamId, without calling core', async () => { + await expect(agentTeamApi.messageMember({ teamId: '', content: 'x' })).rejects.toThrow( + 'teamId is required' + ); + await expect(agentTeamApi.messageMember({ teamId: 'team-1', content: ' ' })).rejects.toThrow( + 'content is required' + ); + expect(mockCall).not.toHaveBeenCalled(); + }); +}); + +describe('agentTeamApi.startMember', () => { + it('unwraps { result } and forwards an explicit taskId', async () => { + mockCall.mockResolvedValueOnce({ + result: { kind: 'started', runId: 'teamrun-1', task: { id: 'task-1' } }, + }); + const outcome = await agentTeamApi.startMember({ + teamId: 'team-1', + memberId: 'm1', + taskId: 'task-1', + }); + expect(outcome).toEqual({ kind: 'started', runId: 'teamrun-1', task: { id: 'task-1' } }); + expect(mockCall).toHaveBeenCalledWith({ + method: 'openhuman.agent_team_start_member', + params: { teamId: 'team-1', memberId: 'm1', taskId: 'task-1' }, + }); + }); + + it('omits taskId when auto-picking and surfaces a blocked outcome', async () => { + mockCall.mockResolvedValueOnce({ result: { kind: 'blocked', unmet: ['task-a'] } }); + const outcome = await agentTeamApi.startMember({ teamId: 'team-1', memberId: 'm1' }); + expect(outcome).toEqual({ kind: 'blocked', unmet: ['task-a'] }); + expect(mockCall).toHaveBeenCalledWith({ + method: 'openhuman.agent_team_start_member', + params: { teamId: 'team-1', memberId: 'm1' }, + }); + }); + + it('throws when teamId or memberId is empty, without calling core', async () => { + await expect(agentTeamApi.startMember({ teamId: '', memberId: 'm1' })).rejects.toThrow( + 'required' + ); + await expect(agentTeamApi.startMember({ teamId: 'team-1', memberId: '' })).rejects.toThrow( + 'required' + ); + expect(mockCall).not.toHaveBeenCalled(); + }); +}); diff --git a/app/src/services/api/agentTeamApi.ts b/app/src/services/api/agentTeamApi.ts index 8ebaf657d..3e48ffc0f 100644 --- a/app/src/services/api/agentTeamApi.ts +++ b/app/src/services/api/agentTeamApi.ts @@ -5,9 +5,11 @@ * ledger (PR1, #3546): the read paths `agent_team_list`, `agent_team_get`, and * `agent_team_list_messages`, plus two lifecycle writes added with quality-gated * completion — `agent_team_complete_task` and `agent_team_shutdown_member`. - * Creating teams, assigning and claiming tasks, and posting messages stay the - * agents' job (driven over the same controllers from the run loop), so those - * write methods are deliberately absent here. + * PR4 adds the two user-driven writes the Teams UI needs: `messageMember` + * (address a named teammate / broadcast) and `startMember` (spawn a live worker + * on a task). Creating teams, assigning, and claiming tasks stay the agents' + * job (driven over the same controllers from the run loop), so those write + * methods are deliberately absent here. * * The Rust controllers serialize their row types with * `#[serde(rename_all = "camelCase")]`, so the wire payload is already camelCase @@ -106,6 +108,19 @@ export interface MemberShutdown { releasedTaskIds: string[]; } +/** + * Outcome of starting a live run for a member. Mirrors Rust `StartMemberOutcome` + * (internally-tagged on `kind`): `started` carries the dispatched `runId` and the + * claimed task; the rest explain why no worker was spawned. + */ +export type StartMemberOutcome = + | { kind: 'started'; runId: string; task: AgentTeamTask } + | { kind: 'blocked'; unmet: string[] } + | { kind: 'alreadyClaimed' } + | { kind: 'alreadyActive' } + | { kind: 'noClaimableTask' } + | { kind: 'unknownTask' }; + /** Parsed payload of a `team_message` run event. Mirrors the Rust `json!` body. */ export interface TeamMessagePayload { from: string; @@ -258,4 +273,64 @@ export const agentTeamApi = { log('shutdownMember released=%d', response.result.releasedTaskIds.length); return response.result; }, + + /** + * Send a message to a named teammate (or broadcast when `toMemberId` is + * omitted). `fromMemberId` is omitted for a lead/user-originated message — the + * core stores it with `from = "lead"`. Returns the appended message event. + */ + messageMember: async (params: { + teamId: string; + toMemberId?: string; + fromMemberId?: string; + content: string; + visibility?: string; + }): Promise => { + const { teamId, toMemberId, fromMemberId, content, visibility } = params; + if (!teamId) throw new Error('agentTeamApi.messageMember: teamId is required'); + if (!content.trim()) throw new Error('agentTeamApi.messageMember: content is required'); + log('messageMember teamId=%s to=%o from=%o', teamId, toMemberId, fromMemberId); + const response = await callCoreRpc<{ message: RawRunEvent }>({ + method: 'openhuman.agent_team_message_member', + params: { + teamId, + content, + ...(fromMemberId ? { fromMemberId } : {}), + ...(toMemberId ? { toMemberId } : {}), + ...(visibility ? { visibility } : {}), + }, + }); + const event = response.message; + return { + runId: event.runId, + sequence: event.sequence, + eventType: event.eventType, + payload: readMessagePayload(event.payload), + timestamp: event.timestamp, + }; + }, + + /** + * Start a live run for a member: the core claims a task (the explicit `taskId`, + * else the member's next claimable one) and spawns a worker that runs it to + * completion. Returns a {@link StartMemberOutcome} — `started` (worker + * dispatched) or a reason no work began. + */ + startMember: async (params: { + teamId: string; + memberId: string; + taskId?: string; + }): Promise => { + const { teamId, memberId, taskId } = params; + if (!teamId || !memberId) { + throw new Error('agentTeamApi.startMember: teamId and memberId are required'); + } + log('startMember teamId=%s memberId=%s taskId=%o', teamId, memberId, taskId); + const response = await callCoreRpc<{ result: StartMemberOutcome }>({ + method: 'openhuman.agent_team_start_member', + params: { teamId, memberId, ...(taskId ? { taskId } : {}) }, + }); + log('startMember kind=%s', response.result.kind); + return response.result; + }, }; diff --git a/app/test/e2e/specs/agent-teams-live.spec.ts b/app/test/e2e/specs/agent-teams-live.spec.ts new file mode 100644 index 000000000..7435a6bd7 --- /dev/null +++ b/app/test/e2e/specs/agent-teams-live.spec.ts @@ -0,0 +1,143 @@ +import { waitForApp, waitForAppReady } from '../helpers/app-helpers'; +import { callOpenhumanRpc, expectRpcOk } from '../helpers/core-rpc'; +import { triggerAuthDeepLinkBypass } from '../helpers/deep-link-helpers'; +import { + clickText, + textExists, + waitForText, + waitForWebView, + waitForWindowVisible, +} from '../helpers/element-helpers'; +import { supportsExecuteScript } from '../helpers/platform'; +import { completeOnboardingIfVisible, navigateViaHash } from '../helpers/shared-flows'; +import { startMockServer, stopMockServer } from '../mock-server'; + +/** + * Agent-teams live coordination surface spec (#3374 PR4). + * + * Goal: prove the Intelligence → Teams tab renders a seeded team's header, + * task board and teammate-message timeline, and exposes the PR4 interactive + * affordances — the message composer and the per-member live-start control. + * The team + task + lead message are seeded over the core JSON-RPC the app + * already talks to (driver-agnostic HTTP). The end-to-end *completion* of a + * live teammate run is asserted headlessly in the Rust JSON-RPC e2e + * (`json_rpc_agent_team_live_member_run_roundtrip`); this spec focuses on the + * UI surface. + * + * Mac2 skipped — the Intelligence pane is not mapped to Appium helpers + * (mirrors `insights-dashboard.spec.ts`); runs under tauri-driver on Linux CI. + */ +function stepLog(message: string, context?: unknown): void { + const stamp = new Date().toISOString(); + if (context === undefined) { + console.log(`[AgentTeamsLiveE2E][${stamp}] ${message}`); + return; + } + console.log(`[AgentTeamsLiveE2E][${stamp}] ${message}`, JSON.stringify(context, null, 2)); +} + +interface SeededTeam { + teamId: string; + memberId: string; + taskTitle: string; + summary: string; + messageBody: string; +} + +interface TeamCreateResult { + team: { id: string }; + members: Array<{ id: string; name: string }>; +} +interface TaskAssignResult { + task: { id: string }; +} + +describe('Agent teams live coordination surface', () => { + let seeded: SeededTeam | null = null; + + before(async function beforeSuite() { + this.timeout(90_000); + if (!supportsExecuteScript()) { + stepLog('Skipping suite on Mac2 — Intelligence pane not mapped'); + this.skip(); + } + + stepLog('starting mock server'); + await startMockServer(); + stepLog('waiting for app'); + await waitForApp(); + stepLog('triggering auth bypass deep link'); + await triggerAuthDeepLinkBypass('e2e-agent-teams-live'); + await waitForWindowVisible(25_000); + await waitForWebView(15_000); + await waitForAppReady(15_000); + await completeOnboardingIfVisible('[AgentTeamsLiveE2E]'); + + // Seed a team + member + task + a lead-originated message over the core RPC. + const summary = `E2E team ${Date.now()}`; + const taskTitle = 'Draft the launch note'; + const messageBody = 'kick off the draft when ready'; + + stepLog('seeding team'); + const created = await callOpenhumanRpc('openhuman.agent_team_create', { + leadAgentId: 'lead', + summary, + members: [{ name: 'scout', agentId: 'researcher' }], + }); + expectRpcOk('agent_team_create', created); + const teamId = created.result!.team.id; + const memberId = created.result!.members[0].id; + + stepLog('seeding task'); + const assigned = await callOpenhumanRpc('openhuman.agent_team_assign_task', { + teamId, + title: taskTitle, + ownerMemberId: memberId, + dependsOn: [], + }); + expectRpcOk('agent_team_assign_task', assigned); + + stepLog('seeding lead message'); + const messaged = await callOpenhumanRpc('openhuman.agent_team_message_member', { + teamId, + toMemberId: memberId, + content: messageBody, + }); + expectRpcOk('agent_team_message_member', messaged); + + seeded = { teamId, memberId, taskTitle, summary, messageBody }; + stepLog('seeded', seeded); + }); + + after(async () => { + stepLog('stopping mock server'); + await stopMockServer(); + }); + + it('renders the seeded team header and task board on the Teams tab', async () => { + if (!seeded) throw new Error('seed missing'); + stepLog('navigating to /intelligence'); + await navigateViaHash('/intelligence'); + await waitForWebView(15_000); + + stepLog('opening Teams tab'); + await clickText('Teams', 10_000); + + // Team summary (list or auto-selected header) + the owned task title. + await waitForText(seeded.summary, 15_000); + await waitForText(seeded.taskTitle, 15_000); + stepLog('team header + task rendered'); + }); + + it('shows the lead message in the activity timeline', async () => { + if (!seeded) throw new Error('seed missing'); + expect(await textExists(seeded.messageBody)).toBe(true); + }); + + it('exposes the PR4 composer affordance', async () => { + // The composer footer placeholder proves the interactive send surface is + // mounted on the active team (read-only PR2 had no input). + await waitForText('Message a teammate', 10_000); + stepLog('composer affordance present'); + }); +}); diff --git a/src/openhuman/agent_orchestration/agent_teams/mod.rs b/src/openhuman/agent_orchestration/agent_teams/mod.rs index 5ef5fca43..0babba9ae 100644 --- a/src/openhuman/agent_orchestration/agent_teams/mod.rs +++ b/src/openhuman/agent_orchestration/agent_teams/mod.rs @@ -8,18 +8,22 @@ //! never in the main chat context — so a coordination session can be listed, //! inspected, and resumed. //! -//! Scope (this module today): the durable model + 10 read/write controllers -//! (`create`, `list`, `get`, `assign_task`, `claim_task`, `message_member`, -//! `list_messages`, `complete_task`, `shutdown_member`, `close`), the atomic +//! Scope: the durable model + 11 read/write controllers (`create`, `list`, +//! `get`, `assign_task`, `claim_task`, `message_member`, `list_messages`, +//! `complete_task`, `shutdown_member`, `close`, `start_member`), the atomic //! compare-and-swap claim primitive, dependency validation (self / unknown / -//! cycle), and quality-gated task completion (dependencies done, claimant owns -//! the task, evidence present when required). Live agent execution (spawning -//! workers, driving the run loop) and the message-send UI are a follow-up PR. +//! cycle), quality-gated task completion (dependencies done, claimant owns the +//! task, evidence present when required), and — as of #3374 PR4 — **live +//! teammate execution**: `start_member` (in [`runtime`]) claims a task and +//! spawns a real worker sub-agent that runs to completion, capturing its output +//! as the task's evidence, with pending lead/teammate messages delivered into +//! the worker prompt at spawn. //! //! Namespace note: `agent_team` is distinct from the existing `team` domain, //! which manages backend org/team membership. pub mod ops; +pub mod runtime; mod schemas; pub mod types; @@ -27,8 +31,9 @@ pub use ops::{ assign_task, claim_task, close_team, complete_task, create_team, get_team, list_messages, list_teams, message_member, shutdown_member, NewMember, }; +pub use runtime::start_member_run; pub use schemas::{ all_controller_schemas as all_agent_team_controller_schemas, all_registered_controllers as all_agent_team_registered_controllers, }; -pub use types::{MemberShutdown, TeamError, TeamView}; +pub use types::{MemberShutdown, StartMemberOutcome, TeamError, TeamView}; diff --git a/src/openhuman/agent_orchestration/agent_teams/ops.rs b/src/openhuman/agent_orchestration/agent_teams/ops.rs index 526e225a2..43faab641 100644 --- a/src/openhuman/agent_orchestration/agent_teams/ops.rs +++ b/src/openhuman/agent_orchestration/agent_teams/ops.rs @@ -192,28 +192,47 @@ pub fn claim_task( run_ledger::claim_agent_team_task(config, team_id, task_id, member_id, claim_token) } +/// Sentinel `from` value for a message that originates from the team lead / the +/// human user rather than a teammate member. The UI send-composer uses this so a +/// person can address a named teammate without being a member row themselves. +pub const LEAD_SENDER: &str = "lead"; + /// Send a message from one member to another (or broadcast). /// -/// Persisted as a run-ledger event keyed by `run_id = team_id`, so the messaging -/// stream reuses the durable event log with no new table. +/// `from_member_id = None` marks a lead/user-originated message (stored with +/// `from = "lead"`); `Some(id)` is a teammate-to-teammate message and must +/// reference a real member. Persisted as a run-ledger event keyed by +/// `run_id = team_id`, so the messaging stream reuses the durable event log with +/// no new table. pub fn message_member( config: &Config, team_id: &str, - from_member_id: &str, + from_member_id: Option<&str>, to_member_id: Option<&str>, content: &str, visibility: Option<&str>, ) -> Result { log::debug!( - "{LOG_PREFIX} message_member.entry team={team_id} from={from_member_id} to={:?}", + "{LOG_PREFIX} message_member.entry team={team_id} from={:?} to={:?}", + from_member_id, to_member_id ); + // Reject unknown teams up front. A lead-origin broadcast (`from = None`, + // `to = None`) skips both member checks below, so without this guard an + // unknown `team_id` would still append an orphan `team_message` event to a + // non-existent team's run ledger. + if run_ledger::get_agent_team(config, team_id)?.is_none() { + return Err(anyhow!("unknown team: {team_id}")); + } + let members = run_ledger::list_agent_team_members(config, team_id)?; - if !members.iter().any(|m| m.id == from_member_id) { - return Err(anyhow!(TeamError::UnknownMember { - member_id: from_member_id.to_string(), - })); + if let Some(from) = from_member_id { + if !members.iter().any(|m| m.id == from) { + return Err(anyhow!(TeamError::UnknownMember { + member_id: from.to_string(), + })); + } } if let Some(to) = to_member_id { if !members.iter().any(|m| m.id == to) { @@ -223,13 +242,14 @@ pub fn message_member( } } + let from_value = from_member_id.unwrap_or(LEAD_SENDER); let event = run_ledger::append_run_event( config, RunEventAppend { run_id: team_id.to_string(), event_type: TEAM_MESSAGE_EVENT.to_string(), payload: json!({ - "from": from_member_id, + "from": from_value, "to": to_member_id, "content": content, "visibility": visibility.unwrap_or("team"), @@ -571,8 +591,8 @@ mod tests { let alice = view.members[0].id.clone(); let bob = view.members[1].id.clone(); - message_member(&config, &team_id, &alice, Some(&bob), "first", None).unwrap(); - message_member(&config, &team_id, &bob, Some(&alice), "second", None).unwrap(); + message_member(&config, &team_id, Some(&alice), Some(&bob), "first", None).unwrap(); + message_member(&config, &team_id, Some(&bob), Some(&alice), "second", None).unwrap(); let messages = list_messages(&config, &team_id, None).unwrap(); assert_eq!(messages.len(), 2); @@ -582,6 +602,31 @@ mod tests { assert_eq!(messages[1].payload["content"], "second"); } + #[test] + fn message_member_lead_origin_and_unknown_from() { + let dir = TempDir::new().unwrap(); + let config = test_config(&dir); + let (team_id, alice) = solo_team(&config, "alice"); + + // Lead/user-originated message: from = None → stored as "lead", no + // member validation on the sender. + message_member(&config, &team_id, None, Some(&alice), "from the lead", None).unwrap(); + let messages = list_messages(&config, &team_id, None).unwrap(); + assert_eq!(messages.len(), 1); + assert_eq!(messages[0].payload["from"], LEAD_SENDER); + assert_eq!(messages[0].payload["to"], alice); + + // A non-None sender that is not a member is still rejected. + let err = + message_member(&config, &team_id, Some("ghost"), Some(&alice), "x", None).unwrap_err(); + assert_eq!( + team_err(err), + TeamError::UnknownMember { + member_id: "ghost".into() + } + ); + } + /// Create a single-member team and return `(team_id, member_id)`. fn solo_team(config: &Config, name: &str) -> (String, String) { let view = create_team( diff --git a/src/openhuman/agent_orchestration/agent_teams/runtime.rs b/src/openhuman/agent_orchestration/agent_teams/runtime.rs new file mode 100644 index 000000000..03c54cda4 --- /dev/null +++ b/src/openhuman/agent_orchestration/agent_teams/runtime.rs @@ -0,0 +1,476 @@ +//! Live execution runtime for agent-team members (#3374 PR4). +//! +//! PR1–PR3 shipped the durable team model, race-safe claiming, quality-gated +//! completion, and the read-only board. They were entirely *store-only*: a +//! claim flipped a row, nothing ran. This module makes a teammate actually +//! **execute** — [`start_member_run`] atomically claims a task for a member, +//! marks the member `active`, and spawns a background worker that drives a real +//! sub-agent to completion, captures its output as the task's completion +//! evidence, runs it through the existing quality gate, and returns the member +//! to `idle`. +//! +//! ## Root parent context +//! +//! Like the workflow engine (#3375), the worker is spawned from a controller +//! background task with no enclosing agent turn, so it builds a root +//! [`ParentExecutionContext`] (shared [`build_root_parent`]) and runs inside +//! [`with_parent_context`] — every nested `spawn_agent` then resolves a real +//! provider / tools / memory. +//! +//! ## Message delivery boundary +//! +//! Pending lead/teammate messages addressed to the member are injected into the +//! worker's prompt **at spawn** (a well-defined boundary), and are always +//! visible in the team timeline. Mid-turn injection into an already-running +//! harness loop is intentionally **not** supported — the orchestration layer has +//! no live inbox (`AgentOrchestrationSession::message_agent` records metadata +//! only). Boundary delivery satisfies the issue's "see the message in the team +//! timeline or worker thread" criterion via both paths; a live inbox would be a +//! separate orchestration-layer change. + +use anyhow::{anyhow, Result}; +use serde_json::json; + +use crate::openhuman::agent::harness::fork_context::with_parent_context; +use crate::openhuman::agent_orchestration::parent_context::build_root_parent; +use crate::openhuman::agent_orchestration::{ + AgentOrchestrationSession, AgentStatus, SpawnAgentRequest, WaitAgentOptions, +}; +use crate::openhuman::config::Config; +use crate::openhuman::session_db::run_ledger::{ + self, AgentTeamMemberStatus, AgentTeamTask, AgentTeamTaskStatus, ClaimOutcome, RunEvent, + RunEventAppend, RunEventListRequest, +}; + +use super::types::{StartMemberOutcome, TeamError}; + +const LOG_TARGET: &str = "agent_team_runtime"; +/// Fallback worker archetype when a member carries no explicit `agent_id`. +const DEFAULT_TEAMMATE_AGENT_ID: &str = "researcher"; +const TEAM_MESSAGE_EVENT: &str = "team_message"; +/// Per-member delivery watermark event: payload `{ memberId, upToSeq }`. Stored +/// on the same run-event log as the messages, so no schema change is needed. +const MESSAGE_DELIVERED_EVENT: &str = "team_message_delivered"; +/// Event recorded when a worker run ends without completing its task. +const MEMBER_FAILED_EVENT: &str = "team_member_failed"; +/// Page size when draining the full run-event log for a team. `list_recent_run_events` +/// returns `sequence ASC` from the cursor and caps `limit` at 1000, so a team whose +/// event count exceeds one page MUST be paged or later events (watermarks + messages) +/// are silently dropped. +const EVENT_PAGE_SIZE: u32 = 1000; +/// Cap on how much worker output is captured as evidence (UTF-8 safe). +const EVIDENCE_MAX_CHARS: usize = 280; + +/// Start a live run for a team member. **Non-blocking.** +/// +/// Resolves a target task (the explicit `task_id`, else the member's next +/// claimable ready task), atomically claims it, marks the member `active` with +/// the new `run_id`, and `tokio::spawn`s the worker loop — then returns +/// immediately. The UI observes progress by polling `agent_team_get`. +/// +/// Returns a non-`Started` [`StartMemberOutcome`] (no side effects) when no work +/// could be dispatched: the task is already claimed, blocked on dependencies, +/// unknown, or the member has nothing claimable. An unknown member surfaces as +/// [`TeamError::UnknownMember`]. +pub async fn start_member_run( + config: &Config, + team_id: &str, + member_id: &str, + task_id: Option<&str>, + model_override: Option, +) -> Result { + log::debug!( + target: LOG_TARGET, + "[agent_team_runtime] start.entry team={team_id} member={member_id} task={task_id:?}" + ); + + let member = run_ledger::get_agent_team_member(config, member_id)? + .filter(|m| m.team_id == team_id) + .ok_or_else(|| { + anyhow!(TeamError::UnknownMember { + member_id: member_id.to_string(), + }) + })?; + + // Reject a start on an already-active member before any claim or state + // mutation. The UI hides the control for active members, but this entry + // point is reachable directly over RPC; without this guard two near- + // simultaneous calls would each claim a task and the second + // `mark_agent_team_member_running` would clobber the first's task/run + // pointer, leaving two workers for one member. + if member.member_status == AgentTeamMemberStatus::Active { + log::debug!( + target: LOG_TARGET, + "[agent_team_runtime] start.reject_active team={team_id} member={member_id}" + ); + return Ok(StartMemberOutcome::AlreadyActive); + } + + // Resolve the target task: an explicit id, or the member's next claimable + // ready task (unowned or owned-by-this-member, dependencies all done). + let tasks = run_ledger::list_agent_team_tasks(config, team_id)?; + let target = match task_id { + Some(tid) => match tasks.iter().find(|t| t.id == tid) { + Some(t) => t.clone(), + None => return Ok(StartMemberOutcome::UnknownTask), + }, + None => match pick_claimable(&tasks, member_id) { + Some(t) => t.clone(), + None => return Ok(StartMemberOutcome::NoClaimableTask), + }, + }; + + // The team-run id doubles as the claim token (CAS guard) and the member's + // worker/run pointer surfaced to the UI. + let run_id = format!("teamrun-{}", uuid::Uuid::new_v4().simple()); + let claimed = + match run_ledger::claim_agent_team_task(config, team_id, &target.id, member_id, &run_id)? { + ClaimOutcome::Claimed(task) => *task, + ClaimOutcome::AlreadyClaimed => return Ok(StartMemberOutcome::AlreadyClaimed), + ClaimOutcome::Blocked { unmet } => return Ok(StartMemberOutcome::Blocked { unmet }), + ClaimOutcome::UnknownTask => return Ok(StartMemberOutcome::UnknownTask), + }; + + // Mark active synchronously so the polling UI reflects the running member + // before the (async) worker even starts. + run_ledger::mark_agent_team_member_running( + config, + team_id, + member_id, + &claimed.id, + &run_id, + &run_id, + )?; + + let agent_id = member + .agent_id + .clone() + .unwrap_or_else(|| DEFAULT_TEAMMATE_AGENT_ID.to_string()); + let cfg = config.clone(); + let team = team_id.to_string(); + let mem = member_id.to_string(); + let task_for_loop = claimed.clone(); + let rid = run_id.clone(); + tokio::spawn(async move { + run_member_loop( + &cfg, + &team, + &mem, + &agent_id, + task_for_loop, + &rid, + model_override, + ) + .await; + }); + + log::debug!( + target: LOG_TARGET, + "[agent_team_runtime] start.spawned team={team_id} member={member_id} task={} run={run_id}", + claimed.id + ); + Ok(StartMemberOutcome::Started { + run_id, + task: Box::new(claimed), + }) +} + +/// Build the root parent context, then drive the member's worker inside it. +/// Engine-internal failures (config/agent build, spawn, wait) release the task +/// and idle the member so the work is reclaimable; they are recorded, not +/// propagated (there is no caller on the spawned task). +async fn run_member_loop( + config: &Config, + team_id: &str, + member_id: &str, + agent_id: &str, + task: AgentTeamTask, + run_id: &str, + model_override: Option, +) { + let outcome = match build_root_parent(config, "agent_team_runtime", "team", "teamrun").await { + Ok(parent) => { + with_parent_context(parent, async { + drive_member( + config, + team_id, + member_id, + agent_id, + &task, + run_id, + model_override, + ) + .await + }) + .await + } + Err(err) => Err(err), + }; + + if let Err(err) = outcome { + log::error!( + target: LOG_TARGET, + "[agent_team_runtime] loop.failed team={team_id} member={member_id} task={} err={err}", + task.id + ); + let _ = run_ledger::release_agent_team_task(config, team_id, &task.id); + let _ = run_ledger::mark_agent_team_member_idle(config, team_id, member_id); + record_failure_event(config, team_id, member_id, &task.id, &err.to_string()); + } +} + +/// Spawn the worker sub-agent for `task`, wait for it, and reconcile team state. +/// +/// Returns `Err` only for engine-internal failures (spawn/wait), which the +/// caller turns into a release + idle. A worker that runs but ends non-completed +/// is handled here (release + idle + failure event) and returns `Ok`. +async fn drive_member( + config: &Config, + team_id: &str, + member_id: &str, + agent_id: &str, + task: &AgentTeamTask, + run_id: &str, + model_override: Option, +) -> Result<()> { + let delivered = deliver_pending_messages(config, team_id, member_id)?; + let prompt = build_member_prompt(task, &delivered); + + let session = AgentOrchestrationSession::new(format!("team-{team_id}-{member_id}")); + let resp = session + .spawn_agent(SpawnAgentRequest { + agent_id: agent_id.to_string(), + prompt, + model: model_override, + metadata: [ + ("teamId".to_string(), team_id.to_string()), + ("memberId".to_string(), member_id.to_string()), + ("taskId".to_string(), task.id.clone()), + ("teamRunId".to_string(), run_id.to_string()), + ] + .into_iter() + .collect(), + ..Default::default() + }) + .await + .map_err(|e| anyhow!("spawn teammate worker failed: {e}"))?; + + let wait = session + .wait_agents(WaitAgentOptions { + orchestration_ids: vec![resp.orchestration_id.clone()], + timeout_ms: None, + }) + .await + .map_err(|e| anyhow!("wait teammate worker failed: {e}"))?; + + let snapshot = wait + .agents + .into_iter() + .find(|a| a.orchestration_id == resp.orchestration_id) + .ok_or_else(|| anyhow!("worker snapshot missing after wait"))?; + + match snapshot.status { + AgentStatus::Completed => { + let output = snapshot.result_summary.unwrap_or_default(); + let evidence = if output.trim().is_empty() { + Vec::new() + } else { + vec![format!( + "run:{run_id} — {}", + truncate_chars(output.trim(), EVIDENCE_MAX_CHARS) + )] + }; + // The teammate's own output is the completion evidence, so the gate + // does not additionally require evidence; it still enforces the + // dependency / claimant invariants. + let outcome = run_ledger::complete_agent_team_task( + config, team_id, &task.id, member_id, &evidence, false, + )?; + log::debug!( + target: LOG_TARGET, + "[agent_team_runtime] drive.completed team={team_id} member={member_id} task={} outcome={outcome:?}", + task.id + ); + run_ledger::mark_agent_team_member_idle(config, team_id, member_id)?; + Ok(()) + } + AgentStatus::Failed | AgentStatus::Cancelled | AgentStatus::Closed => { + let reason = snapshot + .error + .unwrap_or_else(|| "worker ended without completing".to_string()); + log::warn!( + target: LOG_TARGET, + "[agent_team_runtime] drive.worker_failed team={team_id} member={member_id} task={} reason={reason}", + task.id + ); + run_ledger::release_agent_team_task(config, team_id, &task.id)?; + run_ledger::mark_agent_team_member_idle(config, team_id, member_id)?; + record_failure_event(config, team_id, member_id, &task.id, &reason); + Ok(()) + } + other => { + // `wait_agents` with no timeout only returns on terminal status, so + // this is purely defensive. + log::warn!( + target: LOG_TARGET, + "[agent_team_runtime] drive.non_terminal team={team_id} member={member_id} task={} status={other:?}", + task.id + ); + run_ledger::release_agent_team_task(config, team_id, &task.id)?; + run_ledger::mark_agent_team_member_idle(config, team_id, member_id)?; + Ok(()) + } + } +} + +/// Pick the member's next claimable task: the first (by order) task that is +/// `todo`/`ready`, unclaimed, owned by no-one or by this member, and whose +/// dependencies are all `done`. +fn pick_claimable<'a>(tasks: &'a [AgentTeamTask], member_id: &str) -> Option<&'a AgentTeamTask> { + let done: std::collections::HashSet<&str> = tasks + .iter() + .filter(|t| t.status == AgentTeamTaskStatus::Done) + .map(|t| t.id.as_str()) + .collect(); + tasks.iter().find(|t| { + matches!( + t.status, + AgentTeamTaskStatus::Todo | AgentTeamTaskStatus::Ready + ) && t.claimed_by_member_id.is_none() + && t.owner_member_id + .as_deref() + .map(|o| o == member_id) + .unwrap_or(true) + && t.depends_on.iter().all(|d| done.contains(d.as_str())) + }) +} + +/// Compose the worker prompt from the task + any pending messages addressed to +/// the member. +fn build_member_prompt(task: &AgentTeamTask, messages: &[String]) -> String { + let mut prompt = format!("You are a teammate on an agent team. Task: {}", task.title); + if let Some(obj) = task.objective.as_deref() { + if !obj.trim().is_empty() { + prompt.push_str("\n\nObjective:\n"); + prompt.push_str(obj.trim()); + } + } + if !messages.is_empty() { + prompt.push_str("\n\nMessages from your lead / teammates:\n"); + for msg in messages { + prompt.push_str("- "); + prompt.push_str(msg); + prompt.push('\n'); + } + } + prompt.push_str("\n\nComplete the task and report what you did."); + prompt +} + +/// Drain the entire `sequence ASC` run-event log for a team by paging past the +/// per-query cap. A single `list_recent_run_events` call returns at most 1000 +/// rows from the cursor, so message delivery MUST page or a team that exceeds +/// one page would lose every watermark and message beyond it. +fn drain_run_events(config: &Config, team_id: &str) -> Result> { + let mut events: Vec = Vec::new(); + let mut after: Option = None; + loop { + let response = run_ledger::list_recent_run_events( + config, + &RunEventListRequest { + run_id: team_id.to_string(), + after_sequence: after, + limit: Some(EVENT_PAGE_SIZE), + }, + )?; + let exhausted = (response.count as u32) < EVENT_PAGE_SIZE; + after = response.events.last().map(|e| e.sequence); + events.extend(response.events); + if exhausted || after.is_none() { + break; + } + } + Ok(events) +} + +/// Read undelivered messages addressed to `member_id` (direct or broadcast), +/// advance the per-member delivery watermark, and return their contents. Uses +/// only the existing run-event log — no schema change. +fn deliver_pending_messages( + config: &Config, + team_id: &str, + member_id: &str, +) -> Result> { + let events = drain_run_events(config, team_id)?; + + let watermark = events + .iter() + .filter(|e| e.event_type == MESSAGE_DELIVERED_EVENT) + .filter(|e| e.payload.get("memberId").and_then(|v| v.as_str()) == Some(member_id)) + .filter_map(|e| e.payload.get("upToSeq").and_then(|v| v.as_i64())) + .max() + .unwrap_or(0); + + let mut max_seq = watermark; + let mut contents = Vec::new(); + for event in &events { + if event.event_type != TEAM_MESSAGE_EVENT || (event.sequence as i64) <= watermark { + continue; + } + let to = event.payload.get("to").and_then(|v| v.as_str()); + // Direct (to == member) or broadcast (to absent/null). + if to.is_none() || to == Some(member_id) { + if let Some(content) = event.payload.get("content").and_then(|v| v.as_str()) { + contents.push(content.to_string()); + } + max_seq = max_seq.max(event.sequence as i64); + } + } + + if !contents.is_empty() { + run_ledger::append_run_event( + config, + RunEventAppend { + run_id: team_id.to_string(), + event_type: MESSAGE_DELIVERED_EVENT.to_string(), + payload: json!({ "memberId": member_id, "upToSeq": max_seq }), + }, + )?; + } + Ok(contents) +} + +fn record_failure_event( + config: &Config, + team_id: &str, + member_id: &str, + task_id: &str, + reason: &str, +) { + let _ = run_ledger::append_run_event( + config, + RunEventAppend { + run_id: team_id.to_string(), + event_type: MEMBER_FAILED_EVENT.to_string(), + payload: json!({ + "memberId": member_id, + "taskId": task_id, + "reason": truncate_chars(reason, EVIDENCE_MAX_CHARS), + }), + }, + ); +} + +/// UTF-8-safe truncation by character count (never splits a codepoint). +fn truncate_chars(s: &str, max: usize) -> String { + if s.chars().count() <= max { + return s.to_string(); + } + let mut out: String = s.chars().take(max).collect(); + out.push('…'); + out +} + +#[cfg(test)] +#[path = "runtime_tests.rs"] +mod runtime_tests; diff --git a/src/openhuman/agent_orchestration/agent_teams/runtime_tests.rs b/src/openhuman/agent_orchestration/agent_teams/runtime_tests.rs new file mode 100644 index 000000000..dd5867592 --- /dev/null +++ b/src/openhuman/agent_orchestration/agent_teams/runtime_tests.rs @@ -0,0 +1,644 @@ +//! Live-runtime unit tests (#3374 PR4). +//! +//! The worker-driving core ([`super::drive_member`]) is exercised directly with +//! a mock `Provider` installed via [`with_parent_context`] (mirroring +//! `workflow_runs::engine_tests`), so a teammate runs deterministically without +//! touching the network. [`super::start_member_run`]'s pre-spawn outcome routing +//! (blocked / already-claimed / no-claimable / unknown) is tested through the +//! real entry point; its `Started` path (which spawns a loop building a real +//! `Agent` from config) is covered by the JSON-RPC e2e over the live core stack. + +use std::collections::HashSet; +use std::sync::Arc; + +use async_trait::async_trait; +use serde_json::json; + +use super::*; +use crate::openhuman::agent::harness::definition::AgentDefinitionRegistry; +use crate::openhuman::agent::harness::fork_context::{with_parent_context, ParentExecutionContext}; +use crate::openhuman::config::{AgentConfig, Config}; +use crate::openhuman::context::prompt::ToolCallFormat; +use crate::openhuman::inference::provider::traits::ProviderCapabilities; +use crate::openhuman::inference::provider::{ChatRequest, ChatResponse, Provider}; +use crate::openhuman::memory::{Memory, MemoryCategory, MemoryEntry, NamespaceSummary, RecallOpts}; +use crate::openhuman::session_db::run_ledger::{ + self, AgentTeamMemberStatus, AgentTeamMemberUpsert, AgentTeamStatus, AgentTeamTaskStatus, + AgentTeamTaskUpsert, AgentTeamUpsert, +}; +use crate::openhuman::tools::{Tool, ToolSpec}; + +// ── Mocks (mirror workflow_runs::engine_tests) ────────────────────────────── + +#[derive(Default)] +struct NoopMemory; + +#[async_trait] +impl Memory for NoopMemory { + async fn store( + &self, + _ns: &str, + _key: &str, + _content: &str, + _cat: MemoryCategory, + _sid: Option<&str>, + ) -> anyhow::Result<()> { + Ok(()) + } + async fn recall( + &self, + _q: &str, + _l: usize, + _o: RecallOpts<'_>, + ) -> anyhow::Result> { + Ok(Vec::new()) + } + async fn get(&self, _ns: &str, _key: &str) -> anyhow::Result> { + Ok(None) + } + async fn list( + &self, + _ns: Option<&str>, + _cat: Option<&MemoryCategory>, + _sid: Option<&str>, + ) -> anyhow::Result> { + Ok(Vec::new()) + } + async fn forget(&self, _ns: &str, _key: &str) -> anyhow::Result { + Ok(false) + } + async fn namespace_summaries(&self) -> anyhow::Result> { + Ok(Vec::new()) + } + async fn count(&self) -> anyhow::Result { + Ok(0) + } + async fn health_check(&self) -> bool { + true + } + fn name(&self) -> &str { + "noop" + } +} + +fn text_response(text: impl Into) -> ChatResponse { + ChatResponse { + text: Some(text.into()), + tool_calls: Vec::new(), + usage: None, + reasoning_content: None, + } +} + +/// Mock provider that answers every child with a fixed completion, or fails. +#[derive(Clone)] +struct CannedProvider { + output: String, + fail: bool, +} + +#[async_trait] +impl Provider for CannedProvider { + fn capabilities(&self) -> ProviderCapabilities { + ProviderCapabilities { + native_tool_calling: true, + vision: false, + } + } + async fn chat_with_system( + &self, + _s: Option<&str>, + _m: &str, + _model: &str, + _t: f64, + ) -> anyhow::Result { + Ok("ok".to_string()) + } + async fn chat( + &self, + _request: ChatRequest<'_>, + _model: &str, + _temperature: f64, + ) -> anyhow::Result { + if self.fail { + return Err(anyhow::anyhow!("mock provider forced failure")); + } + Ok(text_response(self.output.clone())) + } +} + +fn mock_parent(provider: Arc) -> ParentExecutionContext { + ParentExecutionContext { + agent_definition_id: "agent_team_runtime".to_string(), + allowed_subagent_ids: HashSet::new(), + provider, + all_tools: Arc::new(Vec::>::new()), + all_tool_specs: Arc::new(Vec::::new()), + model_name: "test-model".to_string(), + temperature: 0.0, + workspace_dir: std::env::temp_dir(), + memory: Arc::new(NoopMemory), + agent_config: AgentConfig::default(), + workflows: Arc::new(Vec::new()), + memory_context: Arc::new(None), + session_id: "team-runtime-test".to_string(), + channel: "test".to_string(), + connected_integrations: Vec::new(), + tool_call_format: ToolCallFormat::PFormat, + session_key: "0_agent_team_runtime".to_string(), + session_parent_prefix: None, + on_progress: None, + run_queue: None, + } +} + +fn test_config() -> (tempfile::TempDir, Config) { + let dir = tempfile::tempdir().expect("tempdir"); + let config = Config { + workspace_dir: dir.path().to_path_buf(), + action_dir: dir.path().join("actions"), + ..Config::default() + }; + (dir, config) +} + +fn seed_team(config: &Config, team_id: &str) { + run_ledger::upsert_agent_team( + config, + AgentTeamUpsert { + id: team_id.into(), + parent_thread_id: None, + lead_agent_id: "lead".into(), + status: AgentTeamStatus::Active, + summary: None, + created_at: None, + closed_at: None, + }, + ) + .unwrap(); +} + +fn seed_member(config: &Config, team_id: &str, member_id: &str, agent_id: Option<&str>) { + run_ledger::upsert_agent_team_member( + config, + AgentTeamMemberUpsert { + id: member_id.into(), + team_id: team_id.into(), + name: member_id.into(), + agent_id: agent_id.map(str::to_string), + member_status: AgentTeamMemberStatus::Idle, + current_task_id: None, + worker_thread_id: None, + run_id: None, + created_at: None, + }, + ) + .unwrap(); +} + +fn seed_task( + config: &Config, + team_id: &str, + task_id: &str, + status: AgentTeamTaskStatus, + owner: Option<&str>, + depends_on: Vec, +) { + run_ledger::upsert_agent_team_task( + config, + AgentTeamTaskUpsert { + id: task_id.into(), + team_id: team_id.into(), + title: format!("task {task_id}"), + objective: Some(format!("do {task_id}")), + status, + owner_member_id: owner.map(str::to_string), + depends_on, + gate_status: None, + gate_reason: None, + evidence: vec![], + source_run_id: None, + order_index: 0, + created_at: None, + }, + ) + .unwrap(); +} + +// ── drive_member (the live-exec core) ─────────────────────────────────────── + +#[tokio::test] +async fn drive_member_completes_task_with_worker_output_as_evidence() { + AgentDefinitionRegistry::init_global_builtins().unwrap(); + let (_dir, config) = test_config(); + seed_team(&config, "team-1"); + seed_member(&config, "team-1", "m1", Some("code_executor")); + seed_task( + &config, + "team-1", + "task-a", + AgentTeamTaskStatus::Todo, + None, + vec![], + ); + // Claim + mark running, mirroring what start_member_run does pre-spawn. + run_ledger::claim_agent_team_task(&config, "team-1", "task-a", "m1", "teamrun-x").unwrap(); + run_ledger::mark_agent_team_member_running( + &config, + "team-1", + "m1", + "task-a", + "teamrun-x", + "teamrun-x", + ) + .unwrap(); + let task = run_ledger::get_agent_team_task(&config, "task-a") + .unwrap() + .unwrap(); + + let provider = Arc::new(CannedProvider { + output: "did the thing".into(), + fail: false, + }); + with_parent_context(mock_parent(provider), async { + drive_member( + &config, + "team-1", + "m1", + "code_executor", + &task, + "teamrun-x", + Some("test-model".into()), + ) + .await + }) + .await + .expect("drive_member ok"); + + let done = run_ledger::get_agent_team_task(&config, "task-a") + .unwrap() + .unwrap(); + assert_eq!(done.status, AgentTeamTaskStatus::Done); + assert_eq!(done.gate_status, "passed"); + assert_eq!(done.evidence.len(), 1, "worker output captured as evidence"); + assert!(done.evidence[0].contains("teamrun-x")); + + let member = run_ledger::get_agent_team_member(&config, "m1") + .unwrap() + .unwrap(); + assert_eq!(member.member_status, AgentTeamMemberStatus::Idle); + assert_eq!(member.current_task_id, None); +} + +#[tokio::test] +async fn drive_member_releases_task_when_worker_fails() { + AgentDefinitionRegistry::init_global_builtins().unwrap(); + let (_dir, config) = test_config(); + seed_team(&config, "team-1"); + seed_member(&config, "team-1", "m1", Some("code_executor")); + seed_task( + &config, + "team-1", + "task-a", + AgentTeamTaskStatus::Todo, + None, + vec![], + ); + run_ledger::claim_agent_team_task(&config, "team-1", "task-a", "m1", "teamrun-x").unwrap(); + run_ledger::mark_agent_team_member_running( + &config, + "team-1", + "m1", + "task-a", + "teamrun-x", + "teamrun-x", + ) + .unwrap(); + let task = run_ledger::get_agent_team_task(&config, "task-a") + .unwrap() + .unwrap(); + + let provider = Arc::new(CannedProvider { + output: String::new(), + fail: true, + }); + with_parent_context(mock_parent(provider), async { + drive_member( + &config, + "team-1", + "m1", + "code_executor", + &task, + "teamrun-x", + Some("test-model".into()), + ) + .await + }) + .await + .expect("drive_member handles worker failure without erroring"); + + // Task released back to todo, claim cleared → reclaimable. + let released = run_ledger::get_agent_team_task(&config, "task-a") + .unwrap() + .unwrap(); + assert_eq!(released.status, AgentTeamTaskStatus::Todo); + assert_eq!(released.claimed_by_member_id, None); + + let member = run_ledger::get_agent_team_member(&config, "m1") + .unwrap() + .unwrap(); + assert_eq!(member.member_status, AgentTeamMemberStatus::Idle); +} + +// ── start_member_run pre-spawn outcome routing ────────────────────────────── + +#[tokio::test] +async fn start_member_run_blocks_on_unmet_dependency() { + let (_dir, config) = test_config(); + seed_team(&config, "team-1"); + seed_member(&config, "team-1", "m1", Some("code_executor")); + seed_task( + &config, + "team-1", + "task-a", + AgentTeamTaskStatus::Todo, + None, + vec![], + ); + seed_task( + &config, + "team-1", + "task-b", + AgentTeamTaskStatus::Todo, + None, + vec!["task-a".into()], + ); + + // Explicit task-b: its dep task-a is not done → Blocked, no spawn. + let outcome = start_member_run(&config, "team-1", "m1", Some("task-b"), None) + .await + .unwrap(); + match outcome { + StartMemberOutcome::Blocked { unmet } => assert_eq!(unmet, vec!["task-a".to_string()]), + other => panic!("expected Blocked, got {other:?}"), + } +} + +#[tokio::test] +async fn start_member_run_reports_already_claimed() { + let (_dir, config) = test_config(); + seed_team(&config, "team-1"); + seed_member(&config, "team-1", "m1", Some("code_executor")); + seed_member(&config, "team-1", "m2", Some("code_executor")); + seed_task( + &config, + "team-1", + "task-a", + AgentTeamTaskStatus::Todo, + None, + vec![], + ); + // m2 already holds task-a. + run_ledger::claim_agent_team_task(&config, "team-1", "task-a", "m2", "tok").unwrap(); + + let outcome = start_member_run(&config, "team-1", "m1", Some("task-a"), None) + .await + .unwrap(); + assert_eq!(outcome, StartMemberOutcome::AlreadyClaimed); +} + +#[tokio::test] +async fn start_member_run_no_claimable_and_unknown_task() { + let (_dir, config) = test_config(); + seed_team(&config, "team-1"); + seed_member(&config, "team-1", "m1", Some("code_executor")); + // No tasks at all → nothing claimable. + let none = start_member_run(&config, "team-1", "m1", None, None) + .await + .unwrap(); + assert_eq!(none, StartMemberOutcome::NoClaimableTask); + + // Explicit unknown task id → UnknownTask. + let unknown = start_member_run(&config, "team-1", "m1", Some("ghost"), None) + .await + .unwrap(); + assert_eq!(unknown, StartMemberOutcome::UnknownTask); + + // Unknown member → typed error. + let err = start_member_run(&config, "team-1", "ghost", None, None) + .await + .unwrap_err(); + assert_eq!( + err.downcast::().unwrap(), + TeamError::UnknownMember { + member_id: "ghost".into() + } + ); +} + +#[tokio::test] +async fn start_member_run_rejects_already_active_member_without_side_effects() { + let (_dir, config) = test_config(); + seed_team(&config, "team-1"); + // A member already mid-run (active), plus a fresh claimable task. + run_ledger::upsert_agent_team_member( + &config, + AgentTeamMemberUpsert { + id: "m1".into(), + team_id: "team-1".into(), + name: "m1".into(), + agent_id: Some("researcher".into()), + member_status: AgentTeamMemberStatus::Active, + current_task_id: Some("t-running".into()), + worker_thread_id: Some("run-running".into()), + run_id: Some("run-running".into()), + created_at: None, + }, + ) + .unwrap(); + seed_task( + &config, + "team-1", + "t-free", + AgentTeamTaskStatus::Todo, + None, + vec![], + ); + + let outcome = start_member_run(&config, "team-1", "m1", None, None) + .await + .unwrap(); + assert_eq!(outcome, StartMemberOutcome::AlreadyActive); + + // No claim happened — the free task is untouched and the member still points + // at its original run (no clobbered pointer). + let task = run_ledger::get_agent_team_task(&config, "t-free") + .unwrap() + .expect("task exists"); + assert_eq!(task.status, AgentTeamTaskStatus::Todo); + assert!(task.claimed_by_member_id.is_none()); + let member = run_ledger::get_agent_team_member(&config, "m1") + .unwrap() + .expect("member exists"); + assert_eq!(member.current_task_id.as_deref(), Some("t-running")); + assert_eq!(member.run_id.as_deref(), Some("run-running")); +} + +// ── helpers ───────────────────────────────────────────────────────────────── + +#[test] +fn pick_claimable_respects_deps_ownership_and_claim() { + let (_dir, config) = test_config(); + let _ = &config; + let tasks = vec![ + // done dep + AgentTeamTask { + id: "a".into(), + team_id: "t".into(), + title: "a".into(), + objective: None, + status: AgentTeamTaskStatus::Done, + owner_member_id: None, + claimed_by_member_id: Some("m1".into()), + claim_token: Some("x".into()), + depends_on: vec![], + gate_status: "passed".into(), + gate_reason: None, + evidence: vec![], + source_run_id: None, + order_index: 0, + created_at: chrono::Utc::now(), + updated_at: chrono::Utc::now(), + }, + // owned by m2 → not claimable by m1 + AgentTeamTask { + id: "b".into(), + team_id: "t".into(), + title: "b".into(), + objective: None, + status: AgentTeamTaskStatus::Todo, + owner_member_id: Some("m2".into()), + claimed_by_member_id: None, + claim_token: None, + depends_on: vec![], + gate_status: "pending".into(), + gate_reason: None, + evidence: vec![], + source_run_id: None, + order_index: 1, + created_at: chrono::Utc::now(), + updated_at: chrono::Utc::now(), + }, + // ready, unowned, dep a is done → claimable + AgentTeamTask { + id: "c".into(), + team_id: "t".into(), + title: "c".into(), + objective: None, + status: AgentTeamTaskStatus::Todo, + owner_member_id: None, + claimed_by_member_id: None, + claim_token: None, + depends_on: vec!["a".into()], + gate_status: "pending".into(), + gate_reason: None, + evidence: vec![], + source_run_id: None, + order_index: 2, + created_at: chrono::Utc::now(), + updated_at: chrono::Utc::now(), + }, + ]; + let picked = pick_claimable(&tasks, "m1").expect("c is claimable"); + assert_eq!(picked.id, "c"); +} + +#[test] +fn deliver_pending_messages_injects_then_watermarks() { + let (_dir, config) = test_config(); + seed_team(&config, "team-1"); + seed_member(&config, "team-1", "m1", None); + // Direct to m1, a broadcast, and one addressed elsewhere. + super::super::ops::message_member(&config, "team-1", None, Some("m1"), "hello m1", None) + .unwrap(); + super::super::ops::message_member(&config, "team-1", None, None, "broadcast", None).unwrap(); + seed_member(&config, "team-1", "m2", None); + super::super::ops::message_member(&config, "team-1", None, Some("m2"), "for m2", None).unwrap(); + + let first = deliver_pending_messages(&config, "team-1", "m1").unwrap(); + assert_eq!(first, vec!["hello m1".to_string(), "broadcast".to_string()]); + + // Second call: watermark advanced → nothing new. + let second = deliver_pending_messages(&config, "team-1", "m1").unwrap(); + assert!(second.is_empty(), "watermark should suppress redelivery"); +} + +#[test] +fn deliver_pending_messages_pages_past_first_event_page() { + // Regression: a single `list_recent_run_events` call returns at most one + // page (1000) from `sequence ASC`, but the pre-fix delivery read one + // unbounded page (capped at 100). A team with more events than the cap + // would drop every message AND watermark beyond it. Seed well past one page + // of filler events, then deliver a message landing at a sequence > the cap. + let (_dir, config) = test_config(); + seed_team(&config, "team-1"); + seed_member(&config, "team-1", "m1", None); + + // Push the sequence far past the old 100-row cap with unrelated events. + for i in 0..150 { + run_ledger::append_run_event( + &config, + run_ledger::RunEventAppend { + run_id: "team-1".into(), + event_type: "noise".into(), + payload: json!({ "i": i }), + }, + ) + .unwrap(); + } + + // This message is appended at sequence > 150 — unreachable on the first page. + super::super::ops::message_member(&config, "team-1", None, Some("m1"), "late note", None) + .unwrap(); + + let delivered = deliver_pending_messages(&config, "team-1", "m1").unwrap(); + assert_eq!( + delivered, + vec!["late note".to_string()], + "message beyond the first event page must still be delivered" + ); + + // Watermark (itself recorded beyond the cap) must also be read back so the + // redelivery guard holds for long-lived teams. + let again = deliver_pending_messages(&config, "team-1", "m1").unwrap(); + assert!( + again.is_empty(), + "watermark past the first page must suppress redelivery" + ); +} + +#[test] +fn build_member_prompt_includes_objective_and_messages() { + let task = AgentTeamTask { + id: "t1".into(), + team_id: "team".into(), + title: "Ship the widget".into(), + objective: Some("wire it up".into()), + status: AgentTeamTaskStatus::InProgress, + owner_member_id: None, + claimed_by_member_id: Some("m1".into()), + claim_token: Some("tok".into()), + depends_on: vec![], + gate_status: "pending".into(), + gate_reason: None, + evidence: vec![], + source_run_id: None, + order_index: 0, + created_at: chrono::Utc::now(), + updated_at: chrono::Utc::now(), + }; + let prompt = build_member_prompt(&task, &["go now".to_string()]); + assert!(prompt.contains("Ship the widget")); + assert!(prompt.contains("wire it up")); + assert!(prompt.contains("go now")); +} diff --git a/src/openhuman/agent_orchestration/agent_teams/schemas.rs b/src/openhuman/agent_orchestration/agent_teams/schemas.rs index 412f8ddcb..f6acf4f50 100644 --- a/src/openhuman/agent_orchestration/agent_teams/schemas.rs +++ b/src/openhuman/agent_orchestration/agent_teams/schemas.rs @@ -4,7 +4,8 @@ //! Surface: create a team with members, list/get teams, assign dependency-aware //! tasks, atomically claim a task, exchange + list teammate messages, complete a //! claimed task behind a quality gate, shut a member down (releasing its tasks), -//! and close a team. Live agent execution is a follow-up. +//! close a team, and — via `start_member` — spawn a live worker that claims and +//! runs a task to completion (see [`super::runtime`]). use serde_json::{Map, Value}; @@ -15,6 +16,7 @@ use crate::openhuman::session_db::run_ledger::AgentTeamListRequest; use crate::rpc::RpcOutcome; use super::ops::{self, NewMember}; +use super::runtime; /// Controller schemas exposed by the agent-teams module. pub fn all_controller_schemas() -> Vec { @@ -29,6 +31,7 @@ pub fn all_controller_schemas() -> Vec { schema_for("agent_team_complete_task"), schema_for("agent_team_shutdown_member"), schema_for("agent_team_close"), + schema_for("agent_team_start_member"), ] } @@ -75,6 +78,10 @@ pub fn all_registered_controllers() -> Vec { schema: schema_for("agent_team_close"), handler: handle_close, }, + RegisteredController { + schema: schema_for("agent_team_start_member"), + handler: handle_start_member, + }, ] } @@ -151,7 +158,10 @@ fn schema_for(function: &str) -> ControllerSchema { description: "Send a message from one member to another (or broadcast to the team).", inputs: vec![ required_str("teamId", "Team id."), - required_str("fromMemberId", "Sender member id."), + optional_str( + "fromMemberId", + "Sender member id; omit for a lead/user-originated message.", + ), optional_str("toMemberId", "Recipient member id (omit to broadcast)."), required_str("content", "Message content."), optional_str("visibility", "Message visibility (default team)."), @@ -217,6 +227,27 @@ fn schema_for(function: &str) -> ControllerSchema { ], outputs: vec![json_output("team", "The closed AgentTeam.")], }, + "agent_team_start_member" => ControllerSchema { + namespace: "agent_team", + function: "start_member", + description: "Spawn a live worker for a member: claims a task and runs a real sub-agent to completion.", + inputs: vec![ + required_str("teamId", "Team id."), + required_str("memberId", "Member to run."), + optional_str( + "taskId", + "Specific task id; omit to auto-pick the member's next claimable task.", + ), + optional_str( + "modelOverride", + "Test-only: pin the spawned worker to this model (production omits).", + ), + ], + outputs: vec![json_output( + "result", + "StartMemberOutcome: started | blocked | alreadyClaimed | noClaimableTask | unknownTask.", + )], + }, other => unreachable!("unknown agent_team schema: {other}"), } } @@ -337,14 +368,16 @@ fn handle_message_member(params: Map) -> ControllerFuture { log::warn!(target: "agent_team_rpc", "[agent_team_rpc][{cid}] message_member.config_failed err={err}"); })?; let team_id = require_str(¶ms, "teamId")?; - let from = require_str(¶ms, "fromMemberId")?; + // `fromMemberId` is optional: omitted ⇒ lead/user-originated message + // (stored as `from = "lead"`); present ⇒ teammate-to-teammate. + let from = opt_str(¶ms, "fromMemberId"); let to = opt_str(¶ms, "toMemberId"); let content = require_str(¶ms, "content")?; let visibility = opt_str(¶ms, "visibility"); let event = ops::message_member( &config, &team_id, - &from, + from.as_deref(), to.as_deref(), &content, visibility.as_deref(), @@ -436,6 +469,32 @@ fn handle_close(params: Map) -> ControllerFuture { }) } +fn handle_start_member(params: Map) -> ControllerFuture { + Box::pin(async move { + let cid = new_correlation_id(); + log::debug!(target: "agent_team_rpc", "[agent_team_rpc][{cid}] start_member.entry"); + let config = config_rpc::load_config_with_timeout().await.inspect_err(|err| { + log::warn!(target: "agent_team_rpc", "[agent_team_rpc][{cid}] start_member.config_failed err={err}"); + })?; + let team_id = require_str(¶ms, "teamId")?; + let member_id = require_str(¶ms, "memberId")?; + let task_id = opt_str(¶ms, "taskId"); + // Test-only seam: pin the spawned worker to a model (e.g. the e2e mock). + let model_override = opt_str(¶ms, "modelOverride"); + let outcome = runtime::start_member_run( + &config, + &team_id, + &member_id, + task_id.as_deref(), + model_override, + ) + .await + .map_err(|e| log_err(&cid, "start_member", e))?; + log::debug!(target: "agent_team_rpc", "[agent_team_rpc][{cid}] start_member.success team={team_id} member={member_id}"); + to_json(serde_json::json!({ "result": outcome })) + }) +} + fn parse_members(params: &Map) -> Result, String> { let raw = match params.get("members") { Some(Value::Array(items)) => items, @@ -570,7 +629,7 @@ mod tests { let schemas = all_controller_schemas(); let registered = all_registered_controllers(); assert_eq!(schemas.len(), registered.len()); - assert_eq!(schemas.len(), 10); + assert_eq!(schemas.len(), 11); assert!(schemas.iter().all(|s| s.namespace == "agent_team")); assert_eq!(schema_for("agent_team_claim_task").function, "claim_task"); assert_eq!( diff --git a/src/openhuman/agent_orchestration/agent_teams/types.rs b/src/openhuman/agent_orchestration/agent_teams/types.rs index 5b697c795..aea711a67 100644 --- a/src/openhuman/agent_orchestration/agent_teams/types.rs +++ b/src/openhuman/agent_orchestration/agent_teams/types.rs @@ -27,6 +27,37 @@ pub struct MemberShutdown { pub released_task_ids: Vec, } +/// Outcome of starting a live run for a team member ([`ops::start_member`]). +/// +/// `Started` means a task was atomically claimed for the member and a worker +/// agent was dispatched on a background task (the member flips to `active` with +/// the returned `run_id`); the run drives to completion asynchronously and the +/// UI observes progress by polling `get`. The non-`Started` variants mean no +/// worker was spawned — the caller can surface the reason without any side +/// effect on member or task state. +#[derive(Debug, Clone, PartialEq, Serialize)] +#[serde(rename_all = "camelCase", tag = "kind")] +pub enum StartMemberOutcome { + /// A task was claimed and a worker dispatched. `run_id` is the orchestration + /// id of the spawned agent; `task` is the freshly-claimed task row. + Started { + run_id: String, + task: Box, + }, + /// The target task's dependencies are not all `done`; `unmet` lists them. + Blocked { unmet: Vec }, + /// The target task is already claimed by another member. + AlreadyClaimed, + /// The member is already `active` on a run — starting again would spawn a + /// second concurrent worker and clobber its task/run pointer. Rejected before + /// any claim or state mutation. + AlreadyActive, + /// No explicit task was given and the member has no claimable ready task. + NoClaimableTask, + /// An explicit `task_id` was given but no such task exists in the team. + UnknownTask, +} + /// A validation / coordination problem surfaced by the agent-team ops. #[derive(Debug, Clone, PartialEq, Eq, Serialize)] #[serde(rename_all = "camelCase", tag = "kind", content = "detail")] diff --git a/src/openhuman/agent_orchestration/mod.rs b/src/openhuman/agent_orchestration/mod.rs index 233b41f9a..f928db6d4 100644 --- a/src/openhuman/agent_orchestration/mod.rs +++ b/src/openhuman/agent_orchestration/mod.rs @@ -8,6 +8,7 @@ pub mod agent_teams; pub mod command_center; mod ops; +pub(crate) mod parent_context; pub mod running_subagents; pub mod tools; pub mod types; diff --git a/src/openhuman/agent_orchestration/parent_context/mod.rs b/src/openhuman/agent_orchestration/parent_context/mod.rs new file mode 100644 index 000000000..cb261d35d --- /dev/null +++ b/src/openhuman/agent_orchestration/parent_context/mod.rs @@ -0,0 +1,108 @@ +//! Shared root [`ParentExecutionContext`] builder for controller-spawned +//! orchestration tasks (#3374 PR4, extracted from the #3375 workflow engine). +//! +//! Both the workflow-run engine ([`workflow_runs::engine`]) and the agent-team +//! runtime ([`agent_teams::runtime`]) need to spawn real sub-agents from a +//! background task that has **no** enclosing agent turn on the stack. Those +//! spawns read their parent execution context from a task-local +//! ([`current_parent`]) that is only set inside an agent turn — so a naive spawn +//! fails with `NoParentContext`. +//! +//! The fix (proven in `triage::escalation::dispatch_target_agent`) is to build a +//! *root* [`ParentExecutionContext`] from a config-built [`Agent`] and run the +//! whole loop inside [`with_parent_context`]. Every nested `spawn_agent` then +//! resolves `current_parent()` to this root, inheriting a real provider, tool +//! registry, memory, and model — the same construction path `agent_chat` uses. +//! +//! This was originally inlined in the workflow engine; PR4 lifts it here so the +//! team runtime reuses the exact same construction (and the single +//! registry-initialisation defense) rather than carrying a second copy of the +//! ~20-field context literal that could drift. + +use std::collections::HashSet; +use std::sync::Arc; + +use anyhow::{Context, Result}; + +use crate::openhuman::agent::harness::fork_context::ParentExecutionContext; +use crate::openhuman::agent::Agent; +use crate::openhuman::config::Config; + +const LOG_TARGET: &str = "agent_orchestration::parent_context"; + +/// Build a root [`ParentExecutionContext`] from a config-built [`Agent`]. +/// +/// Mirrors `triage::escalation::dispatch_target_agent` — the proven path for +/// running sub-agents without an enclosing agent turn. The caller supplies the +/// identity fields that distinguish one orchestration surface from another: +/// +/// - `agent_definition_id` — labels the root parent (e.g. `"workflow_engine"`, +/// `"agent_team_runtime"`); surfaced in spawn metadata / logs. +/// - `channel` — the logical channel the spawned work belongs to (e.g. +/// `"workflow"`, `"team"`). +/// - `session_prefix` — the `session_id` is `"{session_prefix}-{uuid}"`, keeping +/// each surface's sessions namespaced and greppable. +/// +/// Every other field is inherited verbatim from the config-built agent, so the +/// spawned children behave exactly like a normal sub-agent dispatch. +pub(crate) async fn build_root_parent( + config: &Config, + agent_definition_id: &str, + channel: &str, + session_prefix: &str, +) -> Result { + // Sub-agent spawns resolve their definition through the global + // agent-definition registry, so it MUST be initialised before any spawn. + // The full runtime boot (`bootstrap_core_runtime`) does this, but these + // engines can also be reached from contexts that only built the HTTP router + // (e.g. the JSON-RPC e2e harness) — so init defensively here. `OnceLock` + // makes this idempotent: a no-op when the registry is already loaded. + if crate::openhuman::agent::harness::AgentDefinitionRegistry::global().is_none() { + if let Err(err) = crate::openhuman::agent::harness::AgentDefinitionRegistry::init_global( + &config.workspace_dir, + ) { + // A concurrent init may have won the race and populated the registry, + // in which case the `AlreadyInitialized`-style error is benign. But if + // the registry is *still* `None`, init genuinely failed — fail fast + // here rather than letting every downstream `spawn_agent` fail later + // with `NoParentContext` after orchestration state has advanced. + if crate::openhuman::agent::harness::AgentDefinitionRegistry::global().is_none() { + return Err(err) + .context("initialize AgentDefinitionRegistry for orchestration root parent"); + } + log::debug!( + target: LOG_TARGET, + "[parent_context] registry_init_raced err={err}" + ); + } + } + + let mut agent = Agent::from_config(config) + .context("build Agent from config for orchestration root parent")?; + + let integrations = crate::openhuman::composio::fetch_connected_integrations(config).await; + agent.set_connected_integrations(integrations); + + Ok(ParentExecutionContext { + agent_definition_id: agent_definition_id.to_string(), + allowed_subagent_ids: HashSet::new(), + provider: agent.provider_arc(), + all_tools: agent.tools_arc(), + all_tool_specs: agent.tool_specs_arc(), + model_name: agent.model_name().to_string(), + temperature: agent.temperature(), + workspace_dir: agent.workspace_dir().to_path_buf(), + memory: agent.memory_arc(), + agent_config: agent.agent_config().clone(), + workflows: Arc::new(agent.workflows().to_vec()), + memory_context: Arc::new(None), + session_id: format!("{session_prefix}-{}", uuid::Uuid::new_v4()), + channel: channel.to_string(), + connected_integrations: agent.connected_integrations().to_vec(), + tool_call_format: crate::openhuman::context::prompt::ToolCallFormat::PFormat, + session_key: agent.session_key().to_string(), + session_parent_prefix: agent.session_parent_prefix().map(str::to_string), + on_progress: None, + run_queue: None, + }) +} diff --git a/src/openhuman/agent_orchestration/workflow_runs/engine.rs b/src/openhuman/agent_orchestration/workflow_runs/engine.rs index 20acba586..30dec184e 100644 --- a/src/openhuman/agent_orchestration/workflow_runs/engine.rs +++ b/src/openhuman/agent_orchestration/workflow_runs/engine.rs @@ -27,15 +27,15 @@ //! the root, inheriting a real provider, tool registry, memory, and model — the //! same construction path `agent_chat` uses. -use std::collections::{HashMap, HashSet}; +use std::collections::HashMap; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, Mutex, OnceLock}; use anyhow::{anyhow, Context, Result}; use serde_json::{json, Value}; -use crate::openhuman::agent::harness::fork_context::{with_parent_context, ParentExecutionContext}; -use crate::openhuman::agent::Agent; +use crate::openhuman::agent::harness::fork_context::with_parent_context; +use crate::openhuman::agent_orchestration::parent_context::build_root_parent; use crate::openhuman::config::Config; use crate::openhuman::session_db::run_ledger::{ get_workflow_run, upsert_workflow_run, WorkflowRun, WorkflowRunStatus, WorkflowRunUpsert, @@ -287,7 +287,7 @@ pub async fn resume_workflow_run(config: &Config, id: &str) -> Result { with_parent_context(parent, async { drive_phases(config, run_id, &definition, &cancel).await @@ -327,58 +327,6 @@ async fn run_engine_loop(config: &Config, run_id: &str, definition: WorkflowDefi clear_cancel_flag(run_id); } -/// Build a root [`ParentExecutionContext`] from a config-built [`Agent`]. -/// -/// Mirrors `triage::escalation::dispatch_target_agent` — the proven path for -/// running sub-agents without an enclosing agent turn. -async fn build_root_parent(config: &Config) -> Result { - // The engine fans out builtin sub-agents (planner / researcher / critic / - // summarizer), so the agent-definition registry MUST be initialised before - // any spawn. The full runtime boot (`bootstrap_core_runtime`) does this, but - // the engine can also be reached from contexts that only built the HTTP - // router (e.g. JSON-RPC e2e harness) — so init defensively here. `OnceLock` - // makes this idempotent: a no-op when the registry is already loaded. - if crate::openhuman::agent::harness::AgentDefinitionRegistry::global().is_none() { - if let Err(err) = crate::openhuman::agent::harness::AgentDefinitionRegistry::init_global( - &config.workspace_dir, - ) { - log::warn!( - target: LOG_TARGET, - "[workflow_run_engine] root_parent.registry_init_failed err={err}" - ); - } - } - - let mut agent = Agent::from_config(config) - .context("build Agent from config for workflow engine root parent")?; - - let integrations = crate::openhuman::composio::fetch_connected_integrations(config).await; - agent.set_connected_integrations(integrations); - - Ok(ParentExecutionContext { - agent_definition_id: "workflow_engine".to_string(), - allowed_subagent_ids: HashSet::new(), - provider: agent.provider_arc(), - all_tools: agent.tools_arc(), - all_tool_specs: agent.tool_specs_arc(), - model_name: agent.model_name().to_string(), - temperature: agent.temperature(), - workspace_dir: agent.workspace_dir().to_path_buf(), - memory: agent.memory_arc(), - agent_config: agent.agent_config().clone(), - workflows: Arc::new(agent.workflows().to_vec()), - memory_context: Arc::new(None), - session_id: format!("workflow-{}", uuid::Uuid::new_v4()), - channel: "workflow".to_string(), - connected_integrations: agent.connected_integrations().to_vec(), - tool_call_format: crate::openhuman::context::prompt::ToolCallFormat::PFormat, - session_key: agent.session_key().to_string(), - session_parent_prefix: agent.session_parent_prefix().map(str::to_string), - on_progress: None, - run_queue: None, - }) -} - /// Topologically walk the phase DAG, executing each phase whose dependencies /// are all `completed`. Persists after every phase. Returns `Ok(())` once every /// phase is completed, the run is interrupted, or a phase fails (the terminal diff --git a/src/openhuman/channels/runtime/test_support.rs b/src/openhuman/channels/runtime/test_support.rs index 57653366c..41733292c 100644 --- a/src/openhuman/channels/runtime/test_support.rs +++ b/src/openhuman/channels/runtime/test_support.rs @@ -283,16 +283,36 @@ fn memory_entry(input: TestMemoryEntry) -> MemoryEntry { } } +/// Shared serialization guard for any test code that mutates the +/// process-global native agent-turn handler (`AGENT_RUN_TURN_METHOD`). +/// +/// The dispatch harness registers a *mock* `AGENT_RUN_TURN_METHOD` handler, +/// while `start_channels` registers the *real* one (latest-wins on the global +/// registry). Both can run concurrently inside the same test binary, so the +/// real handler can clobber the harness's mock mid-run — producing flaky +/// assertions (e.g. `handler_had_progress` going false because the real +/// handler never feeds the harness progress channel). Every test path that +/// touches that global slot must hold this guard for the whole run. +fn agent_handler_lock() -> &'static tokio::sync::Mutex<()> { + static HARNESS_GUARD: std::sync::OnceLock> = std::sync::OnceLock::new(); + HARNESS_GUARD.get_or_init(|| tokio::sync::Mutex::new(())) +} + +/// Acquire the shared agent-handler guard. Hold the returned guard across any +/// call that re-registers `AGENT_RUN_TURN_METHOD` (the harness, or +/// `start_channels`) so concurrent registrations cannot race in the same +/// process. +pub async fn lock_agent_handler() -> tokio::sync::MutexGuard<'static, ()> { + agent_handler_lock().lock().await +} + pub async fn run_dispatch_harness(options: DispatchHarnessOptions) -> DispatchHarnessObservation { // `init_global` + `register_native_global` mutate process-global state, so - // concurrent harness runs in the same process can overwrite each other's - // handlers mid-run and produce flaky assertions. Serialize the whole run - // (handler registration through observation capture) behind a single lock. - static HARNESS_GUARD: std::sync::OnceLock> = std::sync::OnceLock::new(); - let _harness_guard = HARNESS_GUARD - .get_or_init(|| tokio::sync::Mutex::new(())) - .lock() - .await; + // concurrent harness runs (and concurrent `start_channels` calls) in the + // same process can overwrite each other's handlers mid-run and produce + // flaky assertions. Serialize the whole run (handler registration through + // observation capture) behind the shared agent-handler lock. + let _harness_guard = lock_agent_handler().await; init_global(DEFAULT_CAPACITY); let _ = diff --git a/src/openhuman/session_db/run_ledger/mod.rs b/src/openhuman/session_db/run_ledger/mod.rs index d15f451cc..9dc05487f 100644 --- a/src/openhuman/session_db/run_ledger/mod.rs +++ b/src/openhuman/session_db/run_ledger/mod.rs @@ -13,9 +13,10 @@ pub use ops::{ append_run_event, claim_agent_team_task, complete_agent_team_task, get_agent_run, get_agent_team, get_agent_team_member, get_agent_team_task, get_workflow_run, list_agent_runs, list_agent_team_members, list_agent_team_tasks, list_agent_teams, list_recent_run_events, - list_workflow_runs, shutdown_agent_team_member, transition_agent_run_status, upsert_agent_run, - upsert_agent_team, upsert_agent_team_member, upsert_agent_team_task, upsert_run_telemetry, - upsert_workflow_run, + list_workflow_runs, mark_agent_team_member_idle, mark_agent_team_member_running, + release_agent_team_task, shutdown_agent_team_member, transition_agent_run_status, + upsert_agent_run, upsert_agent_team, upsert_agent_team_member, upsert_agent_team_task, + upsert_run_telemetry, upsert_workflow_run, }; pub use types::{ AgentRun, AgentRunKind, AgentRunListRequest, AgentRunListResponse, AgentRunStatus, diff --git a/src/openhuman/session_db/run_ledger/ops.rs b/src/openhuman/session_db/run_ledger/ops.rs index eac8c80c9..f4d6f3274 100644 --- a/src/openhuman/session_db/run_ledger/ops.rs +++ b/src/openhuman/session_db/run_ledger/ops.rs @@ -1139,6 +1139,104 @@ pub fn shutdown_agent_team_member( Ok(result) } +/// Mark a member as actively running a task: status → `active`, with the +/// current task id and the worker/run identifiers of the spawned agent. Used by +/// the live runtime right after it claims a task and dispatches a worker. +/// Returns the updated member, or `None` if the member is not in the team. +pub fn mark_agent_team_member_running( + config: &Config, + team_id: &str, + member_id: &str, + task_id: &str, + worker_thread_id: &str, + run_id: &str, +) -> Result> { + log::debug!( + "{LOG_PREFIX} mark_agent_team_member_running.entry team={team_id} member={member_id} task={task_id} run={run_id}" + ); + crate::openhuman::session_db::store::with_connection(config, |conn| { + init_run_ledger_schema(conn)?; + let now = Utc::now(); + let changed = conn + .execute( + "UPDATE agent_team_members + SET member_status = 'active', current_task_id = ?1, + worker_thread_id = ?2, run_id = ?3, updated_at = ?4 + WHERE id = ?5 AND team_id = ?6", + params![ + task_id, + worker_thread_id, + run_id, + now.to_rfc3339(), + member_id, + team_id + ], + ) + .context("mark agent team member running")?; + if changed == 0 { + return Ok(None); + } + get_agent_team_member_inner(conn, member_id) + }) +} + +/// Mark a member idle: status → `idle`, clearing `current_task_id`. The +/// `worker_thread_id` / `run_id` are intentionally retained as a pointer to the +/// member's last run for history. Returns the updated member, or `None` if the +/// member is not in the team. Used when a worker run finishes (completed, +/// gate-failed, or failed) so the member is free to pick up new work. +pub fn mark_agent_team_member_idle( + config: &Config, + team_id: &str, + member_id: &str, +) -> Result> { + log::debug!("{LOG_PREFIX} mark_agent_team_member_idle.entry team={team_id} member={member_id}"); + crate::openhuman::session_db::store::with_connection(config, |conn| { + init_run_ledger_schema(conn)?; + let now = Utc::now(); + let changed = conn + .execute( + "UPDATE agent_team_members + SET member_status = 'idle', current_task_id = NULL, updated_at = ?1 + WHERE id = ?2 AND team_id = ?3", + params![now.to_rfc3339(), member_id, team_id], + ) + .context("mark agent team member idle")?; + if changed == 0 { + return Ok(None); + } + get_agent_team_member_inner(conn, member_id) + }) +} + +/// Release a single `in_progress` task back to `todo`, clearing its claim and +/// resetting the quality gate. Returns `true` if a row was actually released +/// (the task existed, belonged to the team, and was `in_progress`). Used by the +/// live runtime when a worker run fails or is aborted, so the task is free for +/// another teammate — the per-task analogue of the bulk release in +/// `shutdown_agent_team_member`. +pub fn release_agent_team_task(config: &Config, team_id: &str, task_id: &str) -> Result { + log::debug!("{LOG_PREFIX} release_agent_team_task.entry team={team_id} task={task_id}"); + crate::openhuman::session_db::store::with_connection(config, |conn| { + init_run_ledger_schema(conn)?; + let now = Utc::now(); + let changed = conn + .execute( + "UPDATE agent_team_tasks + SET status = 'todo', claimed_by_member_id = NULL, claim_token = NULL, + gate_status = 'pending', gate_reason = NULL, updated_at = ?1 + WHERE id = ?2 AND team_id = ?3 AND status = 'in_progress'", + params![now.to_rfc3339(), task_id, team_id], + ) + .context("release agent team task")?; + log::debug!( + "{LOG_PREFIX} release_agent_team_task.exit team={team_id} task={task_id} released={}", + changed > 0 + ); + Ok(changed > 0) + }) +} + fn get_agent_team_inner(conn: &Connection, id: &str) -> Result> { let mut stmt = conn.prepare( "SELECT id, parent_thread_id, lead_agent_id, status, summary, @@ -1572,6 +1670,85 @@ mod tests { assert_eq!(outcome, ClaimOutcome::UnknownTask); } + fn seed_member(config: &Config, team_id: &str, member_id: &str) { + upsert_agent_team_member( + config, + AgentTeamMemberUpsert { + id: member_id.into(), + team_id: team_id.into(), + name: member_id.into(), + agent_id: None, + member_status: AgentTeamMemberStatus::Pending, + current_task_id: None, + worker_thread_id: None, + run_id: None, + created_at: None, + }, + ) + .unwrap(); + } + + #[test] + fn mark_member_running_then_idle_keeps_run_pointer() { + let dir = TempDir::new().unwrap(); + let config = test_config(&dir); + seed_team(&config, "team-1"); + seed_member(&config, "team-1", "m1"); + seed_task(&config, "team-1", "task-a", vec![]); + claim_agent_team_task(&config, "team-1", "task-a", "m1", "tok-1").unwrap(); + + let running = + mark_agent_team_member_running(&config, "team-1", "m1", "task-a", "worker-x", "run-x") + .unwrap() + .expect("member updated"); + assert_eq!(running.member_status, AgentTeamMemberStatus::Active); + assert_eq!(running.current_task_id.as_deref(), Some("task-a")); + assert_eq!(running.worker_thread_id.as_deref(), Some("worker-x")); + assert_eq!(running.run_id.as_deref(), Some("run-x")); + + let idle = mark_agent_team_member_idle(&config, "team-1", "m1") + .unwrap() + .expect("member updated"); + assert_eq!(idle.member_status, AgentTeamMemberStatus::Idle); + assert_eq!(idle.current_task_id, None); + // worker/run pointer retained as last-run history. + assert_eq!(idle.worker_thread_id.as_deref(), Some("worker-x")); + assert_eq!(idle.run_id.as_deref(), Some("run-x")); + + // Unknown member → None, no-op. + assert!( + mark_agent_team_member_running(&config, "team-1", "ghost", "task-a", "w", "r") + .unwrap() + .is_none() + ); + assert!(mark_agent_team_member_idle(&config, "team-1", "ghost") + .unwrap() + .is_none()); + } + + #[test] + fn release_task_frees_in_progress_only() { + let dir = TempDir::new().unwrap(); + let config = test_config(&dir); + seed_team(&config, "team-1"); + seed_member(&config, "team-1", "m1"); + seed_task(&config, "team-1", "task-a", vec![]); + claim_agent_team_task(&config, "team-1", "task-a", "m1", "tok-1").unwrap(); + + // In progress → released back to todo, claim cleared, gate reset. + assert!(release_agent_team_task(&config, "team-1", "task-a").unwrap()); + let task = get_agent_team_task(&config, "task-a").unwrap().unwrap(); + assert_eq!(task.status, AgentTeamTaskStatus::Todo); + assert_eq!(task.claimed_by_member_id, None); + assert_eq!(task.claim_token, None); + assert_eq!(task.gate_status, "pending"); + + // Already todo (not in_progress) → no-op, returns false. + assert!(!release_agent_team_task(&config, "team-1", "task-a").unwrap()); + // Unknown task → false. + assert!(!release_agent_team_task(&config, "team-1", "ghost").unwrap()); + } + #[test] fn claim_blocked_until_dependency_done() { let dir = TempDir::new().unwrap(); diff --git a/tests/channels_web_startup_raw_coverage_e2e.rs b/tests/channels_web_startup_raw_coverage_e2e.rs index 55682b95f..d4526a27c 100644 --- a/tests/channels_web_startup_raw_coverage_e2e.rs +++ b/tests/channels_web_startup_raw_coverage_e2e.rs @@ -8,7 +8,7 @@ use std::time::Duration; use openhuman_core::openhuman::channels::start_channels; use openhuman_core::openhuman::channels::test_support::{ - run_dispatch_harness, DispatchHarnessOptions, TestMemoryEntry, + lock_agent_handler, run_dispatch_harness, DispatchHarnessOptions, TestMemoryEntry, }; use openhuman_core::openhuman::channels::web::{ all_web_channel_controller_schemas, all_web_channel_registered_controllers, channel_web_cancel, @@ -226,6 +226,14 @@ async fn web_chat_cancel_aborts_in_flight_thread_without_real_provider() { #[tokio::test] async fn startup_no_channels_initializes_runtime_and_exits_cleanly() { + // `start_channels` calls `register_agent_handlers()`, which re-registers the + // real `AGENT_RUN_TURN_METHOD` handler on the process-global native registry + // (latest-wins). The dispatch harness installs a *mock* handler on that same + // slot. Without this shared guard, the two race inside the test binary and + // the real handler can clobber the mock mid-run, flaking + // `dispatch_harness_covers_streaming_history_timeout_and_memory_paths` on + // `handler_had_progress`. Hold the guard across the whole startup call. + let _agent_handler_guard = lock_agent_handler().await; let (_tmp, config) = isolated_config(); timeout(Duration::from_secs(20), start_channels(config)) .await diff --git a/tests/json_rpc_e2e.rs b/tests/json_rpc_e2e.rs index ec1bdb8f0..50e9f4fba 100644 --- a/tests/json_rpc_e2e.rs +++ b/tests/json_rpc_e2e.rs @@ -3230,7 +3230,7 @@ async fn json_rpc_agent_team_coordination_roundtrip() { "summary": "ship feature", "members": [ { "name": "alice", "agentId": "researcher" }, - { "name": "bob", "agentId": "code_executor" } + { "name": "bob", "agentId": "researcher" } ] }), ) @@ -12320,3 +12320,268 @@ async fn json_rpc_workflow_run_engine_executes_builtin_to_completion_inner() { rpc_join.abort(); mock_join.abort(); } + +#[test] +fn json_rpc_agent_team_live_member_run_roundtrip() { + run_json_rpc_e2e_on_agent_stack( + "json_rpc_agent_team_live_member_run_roundtrip", + json_rpc_agent_team_live_member_run_roundtrip_inner, + ); +} + +/// End-to-end proof that a teammate actually *runs* (#3374 PR4): the lead creates +/// two teammate tasks (B depends on A), messages a named teammate, then starts +/// each member live. Each member's worker drives a real sub-agent against the +/// mock backend (pinned via `modelOverride`), completes its task through the +/// quality gate with the worker output as evidence, and returns to idle — +/// surfaced by polling `agent_team_get`. +async fn json_rpc_agent_team_live_member_run_roundtrip_inner() { + let _env_lock = json_rpc_e2e_env_lock(); + let tmp = tempdir().expect("tempdir"); + let home = tmp.path(); + let openhuman_home = home.join(".openhuman"); + + let _home_guard = EnvVarGuard::set_to_path("HOME", home); + let _workspace_guard = EnvVarGuard::unset("OPENHUMAN_WORKSPACE"); + let _backend_url_guard = EnvVarGuard::unset("BACKEND_URL"); + let _vite_backend_guard = EnvVarGuard::unset("VITE_BACKEND_URL"); + + let (mock_addr, mock_join) = serve_on_ephemeral(mock_upstream_router()).await; + let mock_origin = format!("http://{}", mock_addr); + + write_min_config(&openhuman_home, &mock_origin); + let user_scoped_dir = openhuman_home.join("users").join("e2e-user"); + write_min_config(&user_scoped_dir, &mock_origin); + + let (rpc_addr, rpc_join) = serve_on_ephemeral(build_core_http_router(false)).await; + let rpc_base = format!("http://{}", rpc_addr); + + tokio::time::sleep(Duration::from_millis(100)).await; + + // Authenticate so the spawned workers' provider has a backend session. + let store = post_json_rpc( + &rpc_base, + 38_000_1, + "openhuman.auth_store_session", + json!({ "token": "e2e-test-jwt", "user_id": "e2e-user" }), + ) + .await; + assert_no_jsonrpc_error(&store, "store_session"); + + // Lead creates a team with two named teammates. + let created = post_json_rpc( + &rpc_base, + 38_000_2, + "openhuman.agent_team_create", + json!({ + "leadAgentId": "lead", + "summary": "live run e2e", + "members": [ + { "name": "alice", "agentId": "researcher" }, + { "name": "bob", "agentId": "researcher" } + ] + }), + ) + .await; + let created_body = assert_no_jsonrpc_error(&created, "agent_team_create"); + let team_id = created_body + .get("team") + .and_then(|t| t.get("id")) + .and_then(Value::as_str) + .expect("team id") + .to_string(); + let members = created_body + .get("members") + .and_then(Value::as_array) + .expect("members"); + let alice_id = members[0] + .get("id") + .and_then(Value::as_str) + .unwrap() + .to_string(); + let bob_id = members[1] + .get("id") + .and_then(Value::as_str) + .unwrap() + .to_string(); + + // Task A (no deps) owned by alice; Task B depends on A, owned by bob. + let assign_a = post_json_rpc( + &rpc_base, + 38_000_3, + "openhuman.agent_team_assign_task", + json!({ "teamId": team_id, "title": "Task A", "ownerMemberId": alice_id, "dependsOn": [] }), + ) + .await; + let task_a_id = assert_no_jsonrpc_error(&assign_a, "assign A") + .get("task") + .and_then(|t| t.get("id")) + .and_then(Value::as_str) + .unwrap() + .to_string(); + let assign_b = post_json_rpc( + &rpc_base, + 38_000_4, + "openhuman.agent_team_assign_task", + json!({ "teamId": team_id, "title": "Task B", "ownerMemberId": bob_id, "dependsOn": [task_a_id.clone()] }), + ) + .await; + let task_b_id = assert_no_jsonrpc_error(&assign_b, "assign B") + .get("task") + .and_then(|t| t.get("id")) + .and_then(Value::as_str) + .unwrap() + .to_string(); + + // Lead messages a named teammate (fromMemberId omitted → lead origin). + let msg = post_json_rpc( + &rpc_base, + 38_000_5, + "openhuman.agent_team_message_member", + json!({ "teamId": team_id, "toMemberId": alice_id, "content": "please start Task A" }), + ) + .await; + assert_no_jsonrpc_error(&msg, "message_member"); + + // Start alice live on Task A → kind "started". + let start_a = post_json_rpc( + &rpc_base, + 38_000_6, + "openhuman.agent_team_start_member", + json!({ "teamId": team_id, "memberId": alice_id, "taskId": task_a_id, "modelOverride": "e2e-mock-model" }), + ) + .await; + assert_eq!( + assert_no_jsonrpc_error(&start_a, "start alice") + .get("result") + .and_then(|r| r.get("kind")) + .and_then(Value::as_str), + Some("started"), + "alice start should dispatch a worker: {start_a}" + ); + + // Poll until Task A is done. + assert!( + poll_team_task_status(&rpc_base, &team_id, &task_a_id, "done").await, + "Task A must reach done" + ); + + // With A done, start bob live on Task B. + let start_b = post_json_rpc( + &rpc_base, + 38_000_7, + "openhuman.agent_team_start_member", + json!({ "teamId": team_id, "memberId": bob_id, "taskId": task_b_id, "modelOverride": "e2e-mock-model" }), + ) + .await; + assert_eq!( + assert_no_jsonrpc_error(&start_b, "start bob") + .get("result") + .and_then(|r| r.get("kind")) + .and_then(Value::as_str), + Some("started"), + "bob start should dispatch a worker: {start_b}" + ); + assert!( + poll_team_task_status(&rpc_base, &team_id, &task_b_id, "done").await, + "Task B must reach done" + ); + + // Final state: both tasks done with evidence, both members idle, and the + // lead message is in the team timeline. + let got = post_json_rpc( + &rpc_base, + 38_000_8, + "openhuman.agent_team_get", + json!({ "teamId": team_id }), + ) + .await; + let team_view = assert_no_jsonrpc_error(&got, "agent_team_get") + .get("team") + .cloned() + .expect("team view"); + let tasks = team_view + .get("tasks") + .and_then(Value::as_array) + .expect("tasks"); + for task in tasks { + assert_eq!( + task.get("status").and_then(Value::as_str), + Some("done"), + "every task done: {task}" + ); + assert!( + task.get("evidence") + .and_then(Value::as_array) + .map(|e| !e.is_empty()) + .unwrap_or(false), + "completed task carries worker-output evidence: {task}" + ); + } + let view_members = team_view + .get("members") + .and_then(Value::as_array) + .expect("members"); + for m in view_members { + assert_eq!( + m.get("memberStatus").and_then(Value::as_str), + Some("idle"), + "members return to idle: {m}" + ); + } + + let messages = post_json_rpc( + &rpc_base, + 38_000_9, + "openhuman.agent_team_list_messages", + json!({ "teamId": team_id }), + ) + .await; + let msgs = assert_no_jsonrpc_error(&messages, "list_messages") + .get("messages") + .and_then(Value::as_array) + .expect("messages"); + assert!( + msgs.iter().any(|m| m + .get("payload") + .and_then(|p| p.get("from")) + .and_then(Value::as_str) + == Some("lead")), + "lead message present in timeline: {msgs:?}" + ); + + rpc_join.abort(); + mock_join.abort(); +} + +/// Poll `agent_team_get` until the named task reaches `want` status (or time out). +async fn poll_team_task_status(rpc_base: &str, team_id: &str, task_id: &str, want: &str) -> bool { + for attempt in 0..160 { + tokio::time::sleep(Duration::from_millis(250)).await; + let got = post_json_rpc( + rpc_base, + 38_100_000 + attempt, + "openhuman.agent_team_get", + json!({ "teamId": team_id }), + ) + .await; + let view = assert_no_jsonrpc_error(&got, "agent_team_get poll"); + // `agent_team_get` answers `{ team: TeamView }`, and TeamView itself nests + // `{ team, members, tasks }`. + let tasks = view + .get("team") + .and_then(|tv| tv.get("tasks")) + .and_then(Value::as_array); + if let Some(tasks) = tasks { + if let Some(task) = tasks + .iter() + .find(|t| t.get("id").and_then(Value::as_str) == Some(task_id)) + { + if task.get("status").and_then(Value::as_str) == Some(want) { + return true; + } + } + } + } + false +}