diff --git a/interface/src/api/client-typed.ts b/interface/src/api/client-typed.ts index 0f4427559..ae9149e99 100644 --- a/interface/src/api/client-typed.ts +++ b/interface/src/api/client-typed.ts @@ -1,4 +1,5 @@ import createClient from "openapi-fetch"; +import { getAuthHeaders } from "./client"; import type { paths } from "./schema"; let baseUrl = ""; @@ -14,11 +15,6 @@ function getClient() { }); } -function getAuthHeaders(): Record { - const token = localStorage.getItem("spacebot_auth_token"); - return token ? { Authorization: `Bearer ${token}` } : {}; -} - // Re-export the typed client for direct use export { getClient }; export type { paths }; diff --git a/interface/src/api/client.ts b/interface/src/api/client.ts index 0edd17848..3e8f9780d 100644 --- a/interface/src/api/client.ts +++ b/interface/src/api/client.ts @@ -24,6 +24,40 @@ export function getApiBase(): string { return BASE_PATH + "/api"; } +/** Storage key holding the token configured as `api.auth_token`. */ +export const AUTH_TOKEN_KEY = "spacebot_auth_token"; + +/** + * The bearer header the API expects when `api.auth_token` is configured. + * + * Empty when no token is stored, which is the common case: the daemon leaves + * the token unset and its auth middleware passes every request through. + */ +export function getAuthHeaders(): Record { + const token = localStorage.getItem(AUTH_TOKEN_KEY); + return token ? {Authorization: `Bearer ${token}`} : {}; +} + +/** + * `fetch` for API requests, carrying the bearer token when one is configured. + * + * Every request to `/api` must go through this. The server rejects an + * unauthenticated request with 401 for every path except health, so a bare + * `fetch` silently breaks the whole dashboard the moment a token is set. + * + * Requests the browser issues without headers cannot use this — `EventSource` + * and any URL handed to an `` or a download — so those remain + * unauthenticated and are the reason `api.auth_token` is not yet fully + * supported end to end. + */ +export function apiFetch(url: string, init?: RequestInit): Promise { + const headers = new Headers(init?.headers); + for (const [name, value] of Object.entries(getAuthHeaders())) { + if (!headers.has(name)) headers.set(name, value); + } + return fetch(url, {...init, headers}); +} + import type * as Types from "./types"; // Re-export commonly used types from schema for backward compatibility @@ -418,7 +452,7 @@ export interface TimelineCheckpoint { // Note: TimelineItem is re-exported from types.ts as a union type async function fetchJson(path: string): Promise { - const response = await fetch(`${getApiBase()}${path}`); + const response = await apiFetch(`${getApiBase()}${path}`); if (!response.ok) { throw new Error(`API error: ${response.status}`); } @@ -1189,6 +1223,38 @@ export interface TaskCommentListResponse { next_cursor?: number | null; } +/** Mirrors TaskAttemptOutcome on the server. */ +export type TaskAttemptOutcome = + | "succeeded" + | "partial" + | "blocked" + | "failed" + | "cancelled" + | "timed_out" + | "interrupted"; + +/** One worker run attempted against a task. */ +export interface TaskAttempt { + id: string; + task_id: string; + worker_id: string; + attempt: number; + author_type: TaskAuthorKind; + author_id?: string | null; + agent_id?: string | null; + channel_id?: string | null; + started_at: string; + /** Absent while the run is still live. */ + outcome?: TaskAttemptOutcome | null; + outcome_summary?: string | null; + ended_at?: string | null; +} + +export interface TaskAttemptListResponse { + attempts: TaskAttempt[]; + summary?: string | null; +} + export interface TaskCommentResponse { comment: TaskComment; } @@ -1297,7 +1363,7 @@ async function taskRequest( init?: Omit & { body?: unknown }, ): Promise { const { body, ...rest } = init ?? {}; - const response = await fetch(`${getApiBase()}${path}`, { + const response = await apiFetch(`${getApiBase()}${path}`, { ...rest, headers: body === undefined @@ -1867,7 +1933,7 @@ export const api = { channels: () => fetchJson("/channels"), deleteChannel: async (agentId: string, channelId: string) => { const params = new URLSearchParams({ agent_id: agentId, channel_id: channelId }); - const response = await fetch(`${getApiBase()}/channels?${params}`, { method: "DELETE" }); + const response = await apiFetch(`${getApiBase()}/channels?${params}`, { method: "DELETE" }); if (!response.ok) throw new Error(`API error: ${response.status}`); return response.json() as Promise<{ success: boolean }>; }, @@ -1880,7 +1946,7 @@ export const api = { inspectPrompt: (channelId: string) => fetchJson(`/channels/prompt/inspect?channel_id=${encodeURIComponent(channelId)}`), setPromptCapture: async (channelId: string, enabled: boolean) => { - const response = await fetch(`${getApiBase()}/channels/prompt/capture`, { + const response = await apiFetch(`${getApiBase()}/channels/prompt/capture`, { method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify({ channel_id: channelId, enabled }), @@ -1965,7 +2031,7 @@ export const api = { return fetchJson(`/cortex-chat/messages?${search}`); }, cortexChatSend: (agentId: string, threadId: string, message: string, channelId?: string) => - fetch(`${getApiBase()}/cortex-chat/send`, { + apiFetch(`${getApiBase()}/cortex-chat/send`, { method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify({ @@ -1980,7 +2046,7 @@ export const api = { `/cortex-chat/threads?agent_id=${encodeURIComponent(agentId)}`, ), cortexChatDeleteThread: async (agentId: string, threadId: string) => { - const response = await fetch(`${getApiBase()}/cortex-chat/thread`, { + const response = await apiFetch(`${getApiBase()}/cortex-chat/thread`, { method: "DELETE", headers: { "Content-Type": "application/json" }, body: JSON.stringify({ agent_id: agentId, thread_id: threadId }), @@ -1992,7 +2058,7 @@ export const api = { agentIdentity: (agentId: string) => fetchJson<{ soul: string | null; identity: string | null; role: string | null }>(`/agents/identity?agent_id=${encodeURIComponent(agentId)}`), updateIdentity: async (request: { agent_id: string; soul?: string | null; identity?: string | null; role?: string | null }) => { - const response = await fetch(`${getApiBase()}/agents/identity`, { + const response = await apiFetch(`${getApiBase()}/agents/identity`, { method: "PUT", headers: { "Content-Type": "application/json" }, body: JSON.stringify(request), @@ -2003,7 +2069,7 @@ export const api = { return response.json() as Promise<{ soul: string | null; identity: string | null; role: string | null }>; }, createAgent: async (agentId: string, displayName?: string, role?: string) => { - const response = await fetch(`${getApiBase()}/agents`, { + const response = await apiFetch(`${getApiBase()}/agents`, { method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify({ agent_id: agentId, display_name: displayName || undefined, role: role || undefined }), @@ -2015,7 +2081,7 @@ export const api = { }, updateAgent: async (agentId: string, update: { display_name?: string; role?: string; gradient_start?: string; gradient_end?: string }) => { - const response = await fetch(`${getApiBase()}/agents`, { + const response = await apiFetch(`${getApiBase()}/agents`, { method: "PUT", headers: { "Content-Type": "application/json" }, body: JSON.stringify({ agent_id: agentId, ...update }), @@ -2028,7 +2094,7 @@ export const api = { deleteAgent: async (agentId: string) => { const params = new URLSearchParams({ agent_id: agentId }); - const response = await fetch(`${getApiBase()}/agents?${params}`, { + const response = await apiFetch(`${getApiBase()}/agents?${params}`, { method: "DELETE", }); if (!response.ok) { @@ -2043,7 +2109,7 @@ export const api = { /** Upload an avatar image for an agent. */ uploadAvatar: async (agentId: string, file: File) => { const params = new URLSearchParams({ agent_id: agentId }); - const response = await fetch(`${getApiBase()}/agents/avatar?${params}`, { + const response = await apiFetch(`${getApiBase()}/agents/avatar?${params}`, { method: "POST", headers: { "Content-Type": file.type }, body: file, @@ -2057,7 +2123,7 @@ export const api = { /** Delete the avatar for an agent. */ deleteAvatar: async (agentId: string) => { const params = new URLSearchParams({ agent_id: agentId }); - const response = await fetch(`${getApiBase()}/agents/avatar?${params}`, { + const response = await apiFetch(`${getApiBase()}/agents/avatar?${params}`, { method: "DELETE", }); if (!response.ok) { @@ -2069,7 +2135,7 @@ export const api = { agentConfig: (agentId: string) => fetchJson(`/agents/config?agent_id=${encodeURIComponent(agentId)}`), updateAgentConfig: async (request: AgentConfigUpdateRequest) => { - const response = await fetch(`${getApiBase()}/agents/config`, { + const response = await apiFetch(`${getApiBase()}/agents/config`, { method: "PUT", headers: { "Content-Type": "application/json" }, body: JSON.stringify(request), @@ -2092,7 +2158,7 @@ export const api = { }, createCronJob: async (agentId: string, request: CreateCronRequest) => { - const response = await fetch(`${getApiBase()}/agents/cron`, { + const response = await apiFetch(`${getApiBase()}/agents/cron`, { method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify({ ...request, agent_id: agentId }), @@ -2105,7 +2171,7 @@ export const api = { deleteCronJob: async (agentId: string, cronId: string) => { const search = new URLSearchParams({ agent_id: agentId, cron_id: cronId }); - const response = await fetch(`${getApiBase()}/agents/cron?${search}`, { + const response = await apiFetch(`${getApiBase()}/agents/cron?${search}`, { method: "DELETE", }); if (!response.ok) { @@ -2115,7 +2181,7 @@ export const api = { }, toggleCronJob: async (agentId: string, cronId: string, enabled: boolean) => { - const response = await fetch(`${getApiBase()}/agents/cron/toggle`, { + const response = await apiFetch(`${getApiBase()}/agents/cron/toggle`, { method: "PUT", headers: { "Content-Type": "application/json" }, body: JSON.stringify({ agent_id: agentId, cron_id: cronId, enabled }), @@ -2127,7 +2193,7 @@ export const api = { }, triggerCronJob: async (agentId: string, cronId: string) => { - const response = await fetch(`${getApiBase()}/agents/cron/trigger`, { + const response = await apiFetch(`${getApiBase()}/agents/cron/trigger`, { method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify({ agent_id: agentId, cron_id: cronId }), @@ -2145,7 +2211,7 @@ export const api = { autonomyFleet: () => fetchJson("/agents/autonomy/fleet"), updateAutonomyCeiling: async (ceiling: AutonomyLevel) => { - const response = await fetch(`${getApiBase()}/agents/autonomy/ceiling`, { + const response = await apiFetch(`${getApiBase()}/agents/autonomy/ceiling`, { method: "PUT", headers: { "Content-Type": "application/json" }, body: JSON.stringify({ ceiling }), @@ -2157,7 +2223,7 @@ export const api = { }, clearHomeChannel: async (agentId: string) => { - const response = await fetch( + const response = await apiFetch( `${getApiBase()}/agents/autonomy/home?agent_id=${encodeURIComponent(agentId)}`, { method: "DELETE" }, ); @@ -2180,7 +2246,7 @@ export const api = { fetchJson(`/agents/wakes?agent_id=${encodeURIComponent(agentId)}`), updateWake: async (agentId: string, wakeId: string, patch: WakeUpdate) => { - const response = await fetch( + const response = await apiFetch( `${getApiBase()}/agents/wakes/${encodeURIComponent(wakeId)}?agent_id=${encodeURIComponent(agentId)}`, { method: "PUT", @@ -2195,7 +2261,7 @@ export const api = { }, fireWake: async (agentId: string, wakeId: string) => { - const response = await fetch( + const response = await apiFetch( `${getApiBase()}/agents/wakes/${encodeURIComponent(wakeId)}/fire?agent_id=${encodeURIComponent(agentId)}`, { method: "POST" }, ); @@ -2206,7 +2272,7 @@ export const api = { }, cancelProcess: async (channelId: string, processType: "worker" | "branch", processId: string) => { - const response = await fetch(`${getApiBase()}/channels/cancel-process`, { + const response = await apiFetch(`${getApiBase()}/channels/cancel-process`, { method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify({ channel_id: channelId, process_type: processType, process_id: processId }), @@ -2220,7 +2286,7 @@ export const api = { // Provider management providers: () => fetchJson("/providers"), updateProvider: async (provider: string, apiKey: string, model: string, baseUrl?: string, apiVersion?: string, deployment?: string) => { - const response = await fetch(`${getApiBase()}/providers`, { + const response = await apiFetch(`${getApiBase()}/providers`, { method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify({ provider, api_key: apiKey, model, base_url: baseUrl, api_version: apiVersion, deployment }), @@ -2231,7 +2297,7 @@ export const api = { return response.json() as Promise; }, testProviderModel: async (provider: string, apiKey: string, model: string, baseUrl?: string, apiVersion?: string, deployment?: string) => { - const response = await fetch(`${getApiBase()}/providers/test-model`, { + const response = await apiFetch(`${getApiBase()}/providers/test-model`, { method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify({ provider, api_key: apiKey, model, base_url: baseUrl, api_version: apiVersion, deployment }), @@ -2242,7 +2308,7 @@ export const api = { return response.json() as Promise; }, getProviderConfig: async (provider: string, options?: { signal?: AbortSignal }) => { - const response = await fetch(`${getApiBase()}/providers/${provider}/config`, { + const response = await apiFetch(`${getApiBase()}/providers/${provider}/config`, { method: "GET", signal: options?.signal, }); @@ -2258,7 +2324,7 @@ export const api = { }>; }, providerDefaultModels: async () => { - const response = await fetch(`${getApiBase()}/providers/default-models`); + const response = await apiFetch(`${getApiBase()}/providers/default-models`); if (!response.ok) { throw new Error(`API error: ${response.status}`); } @@ -2268,7 +2334,7 @@ export const api = { }>; }, startOpenAiOAuthBrowser: async (params?: {model?: string}) => { - const response = await fetch(`${getApiBase()}/providers/openai/browser-oauth/start`, { + const response = await apiFetch(`${getApiBase()}/providers/openai/browser-oauth/start`, { method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify({ @@ -2281,7 +2347,7 @@ export const api = { return response.json() as Promise; }, openAiOAuthBrowserStatus: async (state: string) => { - const response = await fetch( + const response = await apiFetch( `${getApiBase()}/providers/openai/browser-oauth/status?state=${encodeURIComponent(state)}`, ); if (!response.ok) { @@ -2290,7 +2356,7 @@ export const api = { return response.json() as Promise; }, removeProvider: async (provider: string) => { - const response = await fetch(`${getApiBase()}/providers/${encodeURIComponent(provider)}`, { + const response = await apiFetch(`${getApiBase()}/providers/${encodeURIComponent(provider)}`, { method: "DELETE", }); if (!response.ok) { @@ -2308,7 +2374,7 @@ export const api = { return fetchJson(`/models${query}`); }, refreshModels: async () => { - const response = await fetch(`${getApiBase()}/models/refresh`, { + const response = await apiFetch(`${getApiBase()}/models/refresh`, { method: "POST", }); if (!response.ok) { @@ -2326,7 +2392,7 @@ export const api = { for (const file of files) { formData.append("files", file); } - const response = await fetch( + const response = await apiFetch( `${getApiBase()}/agents/ingest/files?agent_id=${encodeURIComponent(agentId)}`, { method: "POST", body: formData }, ); @@ -2338,7 +2404,7 @@ export const api = { deleteIngestFile: async (agentId: string, contentHash: string) => { const params = new URLSearchParams({ agent_id: agentId, content_hash: contentHash }); - const response = await fetch(`${getApiBase()}/agents/ingest/files?${params}`, { + const response = await apiFetch(`${getApiBase()}/agents/ingest/files?${params}`, { method: "DELETE", }); if (!response.ok) { @@ -2358,7 +2424,7 @@ export const api = { }, createBinding: async (request: CreateBindingRequest) => { - const response = await fetch(`${getApiBase()}/bindings`, { + const response = await apiFetch(`${getApiBase()}/bindings`, { method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify(request), @@ -2370,7 +2436,7 @@ export const api = { }, updateBinding: async (request: UpdateBindingRequest) => { - const response = await fetch(`${getApiBase()}/bindings`, { + const response = await apiFetch(`${getApiBase()}/bindings`, { method: "PUT", headers: { "Content-Type": "application/json" }, body: JSON.stringify(request), @@ -2382,7 +2448,7 @@ export const api = { }, deleteBinding: async (request: DeleteBindingRequest) => { - const response = await fetch(`${getApiBase()}/bindings`, { + const response = await apiFetch(`${getApiBase()}/bindings`, { method: "DELETE", headers: { "Content-Type": "application/json" }, body: JSON.stringify(request), @@ -2399,7 +2465,7 @@ export const api = { enabled, adapter: adapter ?? null, }; - const response = await fetch(`${getApiBase()}/messaging/toggle`, { + const response = await apiFetch(`${getApiBase()}/messaging/toggle`, { method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify(body), @@ -2415,7 +2481,7 @@ export const api = { platform, adapter: adapter ?? null, }; - const response = await fetch(`${getApiBase()}/messaging/disconnect`, { + const response = await apiFetch(`${getApiBase()}/messaging/disconnect`, { method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify(body), @@ -2427,7 +2493,7 @@ export const api = { }, createMessagingInstance: async (request: Types.CreateMessagingInstanceRequest) => { - const response = await fetch(`${getApiBase()}/messaging/instances`, { + const response = await apiFetch(`${getApiBase()}/messaging/instances`, { method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify(request), @@ -2439,7 +2505,7 @@ export const api = { }, deleteMessagingInstance: async (request: Types.DeleteMessagingInstanceRequest) => { - const response = await fetch(`${getApiBase()}/messaging/instances`, { + const response = await apiFetch(`${getApiBase()}/messaging/instances`, { method: "DELETE", headers: { "Content-Type": "application/json" }, body: JSON.stringify(request), @@ -2454,7 +2520,7 @@ export const api = { globalSettings: () => fetchJson("/settings"), updateGlobalSettings: async (settings: Types.GlobalSettingsUpdate) => { - const response = await fetch(`${getApiBase()}/settings`, { + const response = await apiFetch(`${getApiBase()}/settings`, { method: "PUT", headers: { "Content-Type": "application/json" }, body: JSON.stringify(settings), @@ -2468,7 +2534,7 @@ export const api = { // Raw config API rawConfig: () => fetchJson("/settings/raw"), updateRawConfig: async (content: string) => { - const response = await fetch(`${getApiBase()}/settings/raw`, { + const response = await apiFetch(`${getApiBase()}/settings/raw`, { method: "PUT", headers: { "Content-Type": "application/json" }, body: JSON.stringify({ content }), @@ -2488,21 +2554,21 @@ export const api = { // Update API updateCheck: () => fetchJson("/update-check"), updateCheckNow: async () => { - const response = await fetch(`${getApiBase()}/update-check`, { method: "POST" }); + const response = await apiFetch(`${getApiBase()}/update-check`, { method: "POST" }); if (!response.ok) { throw new Error(`API error: ${response.status}`); } return response.json() as Promise; }, updateApply: async () => { - const response = await fetch(`${getApiBase()}/update-apply`, { method: "POST" }); + const response = await apiFetch(`${getApiBase()}/update-apply`, { method: "POST" }); if (!response.ok) { throw new Error(`API error: ${response.status}`); } return response.json() as Promise; }, restart: async () => { - const response = await fetch(`${getApiBase()}/restart`, { method: "POST" }); + const response = await apiFetch(`${getApiBase()}/restart`, { method: "POST" }); // 503 carries a typed RestartResponse ({ status: "unavailable" }) so the // UI can show its unavailable-state message instead of a generic error. if (!response.ok && response.status !== 503) { @@ -2516,7 +2582,7 @@ export const api = { fetchJson(`/agents/skills?agent_id=${encodeURIComponent(agentId)}`), installSkill: async (request: InstallSkillRequest) => { - const response = await fetch(`${getApiBase()}/agents/skills/install`, { + const response = await apiFetch(`${getApiBase()}/agents/skills/install`, { method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify(request), @@ -2528,7 +2594,7 @@ export const api = { }, removeSkill: async (request: RemoveSkillRequest) => { - const response = await fetch(`${getApiBase()}/agents/skills/remove`, { + const response = await apiFetch(`${getApiBase()}/agents/skills/remove`, { method: "DELETE", headers: { "Content-Type": "application/json" }, body: JSON.stringify(request), @@ -2549,7 +2615,7 @@ export const api = { for (const file of files) { form.append("file", file); } - const response = await fetch( + const response = await apiFetch( `${getApiBase()}/agents/skills/upload?agent_id=${encodeURIComponent(agentId)}`, { method: "POST", body: form }, ); @@ -2581,7 +2647,7 @@ export const api = { agentLinks: (agentId: string) => fetchJson(`/agents/${encodeURIComponent(agentId)}/links`), createLink: async (request: CreateLinkRequest): Promise<{ link: AgentLinkResponse }> => { - const response = await fetch(`${getApiBase()}/links`, { + const response = await apiFetch(`${getApiBase()}/links`, { method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify(request), @@ -2592,7 +2658,7 @@ export const api = { return response.json() as Promise<{ link: AgentLinkResponse }>; }, updateLink: async (from: string, to: string, request: UpdateLinkRequest): Promise<{ link: AgentLinkResponse }> => { - const response = await fetch( + const response = await apiFetch( `${getApiBase()}/links/${encodeURIComponent(from)}/${encodeURIComponent(to)}`, { method: "PUT", @@ -2606,7 +2672,7 @@ export const api = { return response.json() as Promise<{ link: AgentLinkResponse }>; }, deleteLink: async (from: string, to: string): Promise => { - const response = await fetch( + const response = await apiFetch( `${getApiBase()}/links/${encodeURIComponent(from)}/${encodeURIComponent(to)}`, { method: "DELETE" }, ); @@ -2618,7 +2684,7 @@ export const api = { // Agent Groups API groups: () => fetchJson<{ groups: TopologyGroup[] }>("/links/groups"), createGroup: async (request: CreateGroupRequest): Promise<{ group: TopologyGroup }> => { - const response = await fetch(`${getApiBase()}/links/groups`, { + const response = await apiFetch(`${getApiBase()}/links/groups`, { method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify(request), @@ -2629,7 +2695,7 @@ export const api = { return response.json() as Promise<{ group: TopologyGroup }>; }, updateGroup: async (name: string, request: UpdateGroupRequest): Promise<{ group: TopologyGroup }> => { - const response = await fetch( + const response = await apiFetch( `${getApiBase()}/links/groups/${encodeURIComponent(name)}`, { method: "PUT", @@ -2643,7 +2709,7 @@ export const api = { return response.json() as Promise<{ group: TopologyGroup }>; }, deleteGroup: async (name: string): Promise => { - const response = await fetch( + const response = await apiFetch( `${getApiBase()}/links/groups/${encodeURIComponent(name)}`, { method: "DELETE" }, ); @@ -2655,7 +2721,7 @@ export const api = { // Humans API humans: () => fetchJson<{ humans: TopologyHuman[] }>("/links/humans"), createHuman: async (request: CreateHumanRequest): Promise<{ human: TopologyHuman }> => { - const response = await fetch(`${getApiBase()}/links/humans`, { + const response = await apiFetch(`${getApiBase()}/links/humans`, { method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify(request), @@ -2666,7 +2732,7 @@ export const api = { return response.json() as Promise<{ human: TopologyHuman }>; }, updateHuman: async (id: string, request: UpdateHumanRequest): Promise<{ human: TopologyHuman }> => { - const response = await fetch( + const response = await apiFetch( `${getApiBase()}/links/humans/${encodeURIComponent(id)}`, { method: "PUT", @@ -2680,7 +2746,7 @@ export const api = { return response.json() as Promise<{ human: TopologyHuman }>; }, deleteHuman: async (id: string): Promise => { - const response = await fetch( + const response = await apiFetch( `${getApiBase()}/links/humans/${encodeURIComponent(id)}`, { method: "DELETE" }, ); @@ -2693,7 +2759,7 @@ export const api = { uploadAttachment: (agentId: string, channelId: string, file: File) => { const form = new FormData(); form.append("file", file, file.name); - return fetch( + return apiFetch( `${getApiBase()}/agents/${encodeURIComponent(agentId)}/channels/${encodeURIComponent(channelId)}/attachments/upload`, { method: "POST", body: form }, ); @@ -2718,7 +2784,7 @@ export const api = { // Portal API (renamed from webchat) portalSend: (agentId: string, sessionId: string, message: string, senderName?: string, attachmentIds?: string[]) => - fetch(`${getApiBase()}/portal/send`, { + apiFetch(`${getApiBase()}/portal/send`, { method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify({ @@ -2731,7 +2797,7 @@ export const api = { }), portalHistory: (agentId: string, sessionId: string, limit = 100) => - fetch(`${getApiBase()}/portal/history?agent_id=${encodeURIComponent(agentId)}&session_id=${encodeURIComponent(sessionId)}&limit=${limit}`), + apiFetch(`${getApiBase()}/portal/history?agent_id=${encodeURIComponent(agentId)}&session_id=${encodeURIComponent(sessionId)}&limit=${limit}`), listPortalConversations: ( agentId: string, @@ -2747,7 +2813,7 @@ export const api = { title?: string, settings?: Types.ConversationSettings, ): Promise => { - const response = await fetch(`${getApiBase()}/portal/conversations`, { + const response = await apiFetch(`${getApiBase()}/portal/conversations`, { method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify({ agent_id: agentId, title, settings }), @@ -2763,7 +2829,7 @@ export const api = { archived?: boolean, settings?: Types.ConversationSettings, ): Promise => { - const response = await fetch( + const response = await apiFetch( `${getApiBase()}/portal/conversations/${encodeURIComponent(sessionId)}`, { method: "PUT", @@ -2779,7 +2845,7 @@ export const api = { agentId: string, sessionId: string, ): Promise<{ success: boolean }> => { - const response = await fetch( + const response = await apiFetch( `${getApiBase()}/portal/conversations/${encodeURIComponent(sessionId)}?agent_id=${encodeURIComponent(agentId)}`, { method: "DELETE" }, ); @@ -2797,7 +2863,7 @@ export const api = { ), updateChannelSettings: (channelId: string, agentId: string, settings: Types.ConversationSettings) => - fetch(`${getApiBase()}/channels/${encodeURIComponent(channelId)}/settings`, { + apiFetch(`${getApiBase()}/channels/${encodeURIComponent(channelId)}/settings`, { method: "PUT", headers: { "Content-Type": "application/json" }, body: JSON.stringify({ agent_id: agentId, settings }), @@ -2837,6 +2903,8 @@ export const api = { query ? `/tasks/${taskNumber}/comments?${query}` : `/tasks/${taskNumber}/comments`, ); }, + listTaskAttempts: (taskNumber: number): Promise => + taskRequest(`/tasks/${taskNumber}/attempts`), createTaskComment: ( taskNumber: number, request: CreateTaskCommentRequest, @@ -2870,7 +2938,7 @@ export const api = { body: { source: "portal", ...request }, }), deleteTask: async (taskNumber: number): Promise => { - const response = await fetch(`${getApiBase()}/tasks/${taskNumber}`, { + const response = await apiFetch(`${getApiBase()}/tasks/${taskNumber}`, { method: "DELETE", }); if (!response.ok) throw new Error(`API error: ${response.status}`); @@ -2905,7 +2973,7 @@ export const api = { secretsStatus: () => fetchJson("/secrets/status"), listSecrets: () => fetchJson("/secrets"), putSecret: async (name: string, value: string, category?: SecretCategory): Promise => { - const response = await fetch(`${getApiBase()}/secrets/${encodeURIComponent(name)}`, { + const response = await apiFetch(`${getApiBase()}/secrets/${encodeURIComponent(name)}`, { method: "PUT", headers: { "Content-Type": "application/json" }, body: JSON.stringify({ value, category }), @@ -2917,7 +2985,7 @@ export const api = { return response.json() as Promise; }, deleteSecret: async (name: string): Promise => { - const response = await fetch(`${getApiBase()}/secrets/${encodeURIComponent(name)}`, { + const response = await apiFetch(`${getApiBase()}/secrets/${encodeURIComponent(name)}`, { method: "DELETE", }); if (!response.ok) { @@ -2927,7 +2995,7 @@ export const api = { return response.json() as Promise; }, enableEncryption: async (): Promise => { - const response = await fetch(`${getApiBase()}/secrets/encrypt`, { method: "POST" }); + const response = await apiFetch(`${getApiBase()}/secrets/encrypt`, { method: "POST" }); if (!response.ok) { const body = await response.json().catch(() => ({})); throw new Error(body.error || `API error: ${response.status}`); @@ -2935,7 +3003,7 @@ export const api = { return response.json() as Promise; }, unlockSecrets: async (masterKey: string): Promise => { - const response = await fetch(`${getApiBase()}/secrets/unlock`, { + const response = await apiFetch(`${getApiBase()}/secrets/unlock`, { method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify({ master_key: masterKey }), @@ -2947,7 +3015,7 @@ export const api = { return response.json() as Promise; }, lockSecrets: async (): Promise<{ state: string; message: string }> => { - const response = await fetch(`${getApiBase()}/secrets/lock`, { method: "POST" }); + const response = await apiFetch(`${getApiBase()}/secrets/lock`, { method: "POST" }); if (!response.ok) { const body = await response.json().catch(() => ({})); throw new Error(body.error || `API error: ${response.status}`); @@ -2955,7 +3023,7 @@ export const api = { return response.json() as Promise<{ state: string; message: string }>; }, rotateKey: async (): Promise<{ master_key: string; message: string }> => { - const response = await fetch(`${getApiBase()}/secrets/rotate`, { method: "POST" }); + const response = await apiFetch(`${getApiBase()}/secrets/rotate`, { method: "POST" }); if (!response.ok) { const body = await response.json().catch(() => ({})); throw new Error(body.error || `API error: ${response.status}`); @@ -2963,7 +3031,7 @@ export const api = { return response.json() as Promise<{ master_key: string; message: string }>; }, migrateSecrets: async (): Promise => { - const response = await fetch(`${getApiBase()}/secrets/migrate`, { method: "POST" }); + const response = await apiFetch(`${getApiBase()}/secrets/migrate`, { method: "POST" }); if (!response.ok) { const body = await response.json().catch(() => ({})); throw new Error(body.error || `API error: ${response.status}`); @@ -2985,7 +3053,7 @@ export const api = { ), createProject: async (request: CreateProjectRequest): Promise => { - const response = await fetch(`${getApiBase()}/agents/projects`, { + const response = await apiFetch(`${getApiBase()}/agents/projects`, { method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify(request), @@ -2995,7 +3063,7 @@ export const api = { }, updateProject: async (projectId: string, request: UpdateProjectRequest): Promise => { - const response = await fetch(`${getApiBase()}/agents/projects/${encodeURIComponent(projectId)}`, { + const response = await apiFetch(`${getApiBase()}/agents/projects/${encodeURIComponent(projectId)}`, { method: "PUT", headers: { "Content-Type": "application/json" }, body: JSON.stringify(request), @@ -3005,7 +3073,7 @@ export const api = { }, deleteProject: async (projectId: string): Promise => { - const response = await fetch( + const response = await apiFetch( `${getApiBase()}/agents/projects/${encodeURIComponent(projectId)}`, { method: "DELETE" }, ); @@ -3014,7 +3082,7 @@ export const api = { }, scanProject: async (projectId: string): Promise => { - const response = await fetch( + const response = await apiFetch( `${getApiBase()}/agents/projects/${encodeURIComponent(projectId)}/scan`, { method: "POST" }, ); @@ -3023,7 +3091,7 @@ export const api = { }, reorderProjects: async (ids: string[]): Promise => { - const response = await fetch(`${getApiBase()}/agents/projects/reorder`, { + const response = await apiFetch(`${getApiBase()}/agents/projects/reorder`, { method: "PUT", headers: {"Content-Type": "application/json"}, body: JSON.stringify({ids}), @@ -3037,7 +3105,7 @@ export const api = { ), createProjectRepo: async (projectId: string, request: CreateRepoRequest): Promise<{ repo: ProjectRepo }> => { - const response = await fetch(`${getApiBase()}/agents/projects/${encodeURIComponent(projectId)}/repos`, { + const response = await apiFetch(`${getApiBase()}/agents/projects/${encodeURIComponent(projectId)}/repos`, { method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify(request), @@ -3047,7 +3115,7 @@ export const api = { }, deleteProjectRepo: async (projectId: string, repoId: string): Promise => { - const response = await fetch( + const response = await apiFetch( `${getApiBase()}/agents/projects/${encodeURIComponent(projectId)}/repos/${encodeURIComponent(repoId)}`, { method: "DELETE" }, ); @@ -3056,7 +3124,7 @@ export const api = { }, createProjectWorktree: async (projectId: string, request: CreateWorktreeRequest): Promise<{ worktree: ProjectWorktree }> => { - const response = await fetch(`${getApiBase()}/agents/projects/${encodeURIComponent(projectId)}/worktrees`, { + const response = await apiFetch(`${getApiBase()}/agents/projects/${encodeURIComponent(projectId)}/worktrees`, { method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify(request), @@ -3066,7 +3134,7 @@ export const api = { }, deleteProjectWorktree: async (projectId: string, worktreeId: string): Promise => { - const response = await fetch( + const response = await apiFetch( `${getApiBase()}/agents/projects/${encodeURIComponent(projectId)}/worktrees/${encodeURIComponent(worktreeId)}`, { method: "DELETE" }, ); @@ -3102,38 +3170,38 @@ export const api = { if (params?.limit !== undefined) query.set("limit", String(params.limit)); if (params?.offset !== undefined) query.set("offset", String(params.offset)); const qs = query.toString(); - const response = await fetch(`${getApiBase()}/notifications${qs ? `?${qs}` : ""}`); + const response = await apiFetch(`${getApiBase()}/notifications${qs ? `?${qs}` : ""}`); if (!response.ok) throw new Error(`API error: ${response.status}`); return response.json() as Promise; }, getUnreadCount: async (): Promise => { - const response = await fetch(`${getApiBase()}/notifications/unread_count`); + const response = await apiFetch(`${getApiBase()}/notifications/unread_count`); if (!response.ok) throw new Error(`API error: ${response.status}`); return response.json() as Promise; }, markNotificationRead: async (id: string): Promise => { - const response = await fetch(`${getApiBase()}/notifications/${encodeURIComponent(id)}/read`, { + const response = await apiFetch(`${getApiBase()}/notifications/${encodeURIComponent(id)}/read`, { method: "POST", }); if (!response.ok && response.status !== 404) throw new Error(`API error: ${response.status}`); }, dismissNotification: async (id: string): Promise => { - const response = await fetch(`${getApiBase()}/notifications/${encodeURIComponent(id)}/dismiss`, { + const response = await apiFetch(`${getApiBase()}/notifications/${encodeURIComponent(id)}/dismiss`, { method: "POST", }); if (!response.ok && response.status !== 404) throw new Error(`API error: ${response.status}`); }, markAllNotificationsRead: async (): Promise => { - const response = await fetch(`${getApiBase()}/notifications/read_all`, { method: "POST" }); + const response = await apiFetch(`${getApiBase()}/notifications/read_all`, { method: "POST" }); if (!response.ok) throw new Error(`API error: ${response.status}`); }, dismissReadNotifications: async (): Promise => { - const response = await fetch(`${getApiBase()}/notifications/dismiss_read`, { method: "POST" }); + const response = await apiFetch(`${getApiBase()}/notifications/dismiss_read`, { method: "POST" }); if (!response.ok) throw new Error(`API error: ${response.status}`); }, @@ -3159,7 +3227,7 @@ export const api = { }, createWikiPage: async (request: CreateWikiPageRequest): Promise => { - const response = await fetch(`${getApiBase()}/wiki`, { + const response = await apiFetch(`${getApiBase()}/wiki`, { method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify(request), @@ -3169,7 +3237,7 @@ export const api = { }, editWikiPage: async (slug: string, request: EditWikiPageRequest): Promise => { - const response = await fetch(`${getApiBase()}/wiki/${encodeURIComponent(slug)}/edit`, { + const response = await apiFetch(`${getApiBase()}/wiki/${encodeURIComponent(slug)}/edit`, { method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify(request), @@ -3182,7 +3250,7 @@ export const api = { fetchJson(`/wiki/${encodeURIComponent(slug)}/history?limit=${limit}`), restoreWikiVersion: async (slug: string, version: number): Promise => { - const response = await fetch(`${getApiBase()}/wiki/${encodeURIComponent(slug)}/restore`, { + const response = await apiFetch(`${getApiBase()}/wiki/${encodeURIComponent(slug)}/restore`, { method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify({ version }), @@ -3192,7 +3260,7 @@ export const api = { }, archiveWikiPage: async (slug: string): Promise<{ success: boolean; message: string }> => { - const response = await fetch(`${getApiBase()}/wiki/${encodeURIComponent(slug)}`, { + const response = await apiFetch(`${getApiBase()}/wiki/${encodeURIComponent(slug)}`, { method: "DELETE", }); if (!response.ok) throw new Error(`API error: ${response.status}`); diff --git a/interface/src/components/TaskAttempts.tsx b/interface/src/components/TaskAttempts.tsx new file mode 100644 index 000000000..4f0e355cb --- /dev/null +++ b/interface/src/components/TaskAttempts.tsx @@ -0,0 +1,230 @@ +import {useEffect, useRef, useState} from "react"; +import {useQuery, useQueryClient} from "@tanstack/react-query"; +import {FontAwesomeIcon} from "@fortawesome/react-fontawesome"; +import { + faCheck, + faChevronDown, + faChevronRight, + faCircleHalfStroke, + faHourglassEnd, + faPlug, + faSpinner, + faStop, + faTriangleExclamation, + faXmark, +} from "@fortawesome/free-solid-svg-icons"; +import {Badge} from "@spacedrive/primitives"; +import {api, type TaskAttempt, type TaskAttemptOutcome} from "@/api/client"; +import {useLiveContext} from "@/hooks/useLiveContext"; + +type BadgeVariant = "info" | "success" | "warning" | "error" | "default"; + +const OUTCOME_LABEL: Record = { + succeeded: "Succeeded", + partial: "Partial", + blocked: "Blocked", + failed: "Failed", + cancelled: "Cancelled", + timed_out: "Timed out", + interrupted: "Interrupted", +}; + +const OUTCOME_ICON: Record = { + succeeded: faCheck, + partial: faCircleHalfStroke, + blocked: faTriangleExclamation, + failed: faXmark, + cancelled: faStop, + timed_out: faHourglassEnd, + interrupted: faPlug, +}; + +const OUTCOME_VARIANT: Record = { + succeeded: "success", + partial: "info", + blocked: "warning", + failed: "error", + cancelled: "default", + timed_out: "warning", + interrupted: "default", +}; + +function formatTimestamp(value: string): string { + const parsed = new Date(value); + return Number.isNaN(parsed.getTime()) ? value : parsed.toLocaleString(); +} + +/** Wall-clock duration, or how long a live run has been going. */ +function formatDuration(startedAt: string, endedAt?: string | null): string | null { + const start = new Date(startedAt).getTime(); + const end = endedAt ? new Date(endedAt).getTime() : Date.now(); + if (Number.isNaN(start) || Number.isNaN(end) || end < start) return null; + + const seconds = Math.round((end - start) / 1000); + if (seconds < 60) return `${seconds}s`; + const minutes = Math.floor(seconds / 60); + if (minutes < 60) return `${minutes}m ${seconds % 60}s`; + return `${Math.floor(minutes / 60)}h ${minutes % 60}m`; +} + +/** + * The run's own output, fetched only when asked for. + * + * The attempt row records how the run ended; the worker holds what it actually + * produced, and that is often long enough to bury everything else. + */ +function AttemptOutput({agentId, workerId}: {agentId: string; workerId: string}) { + const [expanded, setExpanded] = useState(false); + + const {data, isLoading, error} = useQuery({ + queryKey: ["worker-detail", agentId, workerId], + queryFn: () => api.workerDetail(agentId, workerId), + enabled: expanded, + staleTime: 60_000, + }); + + return ( +
+ + + {expanded && ( +
+ {isLoading ? ( + Loading worker output… + ) : error ? ( + + Worker run is no longer available. + + ) : ( + 
+							{data?.result?.trim() || "This worker recorded no output."}
+
+ )} +
+ )} +
+ ); +} + +function AttemptRow({attempt, agentId}: {attempt: TaskAttempt; agentId?: string}) { + const live = !attempt.ended_at; + const outcome = attempt.outcome ?? null; + const duration = formatDuration(attempt.started_at, attempt.ended_at); + + return ( +
  • +
    + #{attempt.attempt} + + {live ? ( + + + Running + + ) : outcome ? ( + + + {OUTCOME_LABEL[outcome]} + + ) : ( + + Ended without an outcome + + )} + + + {attempt.worker_id.slice(0, 8)} + + + {formatTimestamp(attempt.started_at)} + {duration ? ` · ${duration}` : ""} + + {attempt.channel_id && ( + via {attempt.channel_id} + )} +
    + + {attempt.outcome_summary && ( +

    + {attempt.outcome_summary} +

    + )} + + {agentId && } +
    • + ); +} + +/** + * Every worker run attempted against this task. + * + * The task row names only the run executing now, so without this a task that + * failed twice before succeeding looks identical to one that worked first time. + */ +export function TaskAttempts({ + taskNumber, + agentId, +}: { + taskNumber: number; + agentId?: string; +}) { + const queryClient = useQueryClient(); + const {workerEventVersion} = useLiveContext(); + const queryKey = ["task-attempts", taskNumber]; + + // A run starting or finishing arrives over SSE. + const previousVersion = useRef(workerEventVersion); + useEffect(() => { + if (workerEventVersion !== previousVersion.current) { + previousVersion.current = workerEventVersion; + void queryClient.invalidateQueries({queryKey}); + } + }, [workerEventVersion, queryClient, taskNumber]); + + const {data, isLoading, error} = useQuery({ + queryKey, + queryFn: () => api.listTaskAttempts(taskNumber), + }); + + const attempts = data?.attempts ?? []; + + return ( +
    +

    + Runs{attempts.length > 0 ? ` (${attempts.length})` : ""} +

    + + {isLoading ? ( +

    Loading runs…

    + ) : error ? ( +

    Failed to load the run history.

    + ) : attempts.length === 0 ? ( +

    + Not worked yet. Every worker run against this task is recorded here. +

    + ) : ( + <> + {data?.summary && ( +

    {data.summary}

    + )} +
      + {attempts.map((attempt) => ( + + ))} +
    + + )} +
    + ); +} diff --git a/interface/src/components/portal/PortalTimeline.tsx b/interface/src/components/portal/PortalTimeline.tsx index 9612bd291..9fe42ecb2 100644 --- a/interface/src/components/portal/PortalTimeline.tsx +++ b/interface/src/components/portal/PortalTimeline.tsx @@ -340,28 +340,30 @@ export function PortalTimeline({ refetchInterval: 2000, }); - const conversationWorkers = (workersQuery.data?.workers ?? []).filter( - (w) => w.channel_id === conversationId, + // The workers query is a page of the agent's most recent workers, not this + // conversation's full set, so it cannot decide which rows exist. It only + // enriches the rows the timeline already carries; `renderTimelineItem` + // falls back to `synthesizeWorker` for any worker outside the page. + const conversationWorkers = useMemo( + () => + (workersQuery.data?.workers ?? []).filter( + (worker) => worker.channel_id === conversationId, + ), + [workersQuery.data, conversationId], ); - const workerIds = new Set(conversationWorkers.map((w) => w.id)); - - const visibleItems = timeline.filter((item) => { - if (item.type !== "worker_run") return true; - return workerIds.has(item.id); - }); const rows: TimelineRow[] = useMemo(() => { - const list: TimelineRow[] = visibleItems.map((item) => ({ + const list: TimelineRow[] = timeline.map((item) => ({ kind: "item", item, })); - if (conversationCreatedAt && visibleItems.length > 0) { + if (conversationCreatedAt && timeline.length > 0) { list.unshift({kind: "conversation_start", createdAt: conversationCreatedAt}); } if (isTyping) list.push({kind: "typing"}); list.push({kind: "spacer"}); return list; - }, [conversationCreatedAt, visibleItems, isTyping]); + }, [conversationCreatedAt, timeline, isTyping]); useEffect(() => { if (sendCount === 0) return; diff --git a/interface/src/routes/AgentTasks.tsx b/interface/src/routes/AgentTasks.tsx index 24bab8bbf..ed2002fc3 100644 --- a/interface/src/routes/AgentTasks.tsx +++ b/interface/src/routes/AgentTasks.tsx @@ -23,6 +23,7 @@ import { taskListTitle, TaskMetadataBadges, } from "@/components/TaskUtils"; +import {TaskAttempts} from "@/components/TaskAttempts"; import {TaskComments} from "@/components/TaskComments"; import {TaskHistory} from "@/components/TaskHistory"; @@ -240,6 +241,10 @@ export function AgentTasks({agentId}: {agentId: string}) { + (r.ok ? r.json() : null)) diff --git a/interface/src/routes/ChannelDetail.tsx b/interface/src/routes/ChannelDetail.tsx index 89d26f981..1df3c0327 100644 --- a/interface/src/routes/ChannelDetail.tsx +++ b/interface/src/routes/ChannelDetail.tsx @@ -418,6 +418,13 @@ export function ChannelDetail({ // paint, and the timeline keeps growing after the first render, so hold the // bottom across a few frames each time. Once the reader scrolls up, their // position is left alone. + // + // The channel counts as opened on the first pin, not after the frame loop + // finishes: rowCount changes on nearly every commit while history streams, + // and the cleanup cancels the pending frame each time, so a loop that only + // records itself at the end never gets there. Leaving it unrecorded holds + // `opening` true, which skips the distance check below and drags the reader + // back to the bottom on every update. useEffect(() => { if (rowCount === 0) return; const opening = openedChannelRef.current !== channelId; @@ -427,11 +434,10 @@ export function ChannelDetail({ let attempts = 0; const pinToEnd = () => { chatRef.current?.scrollToEnd({behavior: "auto"}); + openedChannelRef.current = channelId; attempts += 1; if (attempts < 12) { frame = requestAnimationFrame(pinToEnd); - } else { - openedChannelRef.current = channelId; } }; frame = requestAnimationFrame(pinToEnd); diff --git a/interface/src/routes/GlobalTasks.tsx b/interface/src/routes/GlobalTasks.tsx index dca36789e..00e14ed51 100644 --- a/interface/src/routes/GlobalTasks.tsx +++ b/interface/src/routes/GlobalTasks.tsx @@ -29,6 +29,7 @@ import { taskListTitle, TaskMetadataBadges, } from "@/components/TaskUtils"; +import {TaskAttempts} from "@/components/TaskAttempts"; import {TaskComments} from "@/components/TaskComments"; import {TaskHistory} from "@/components/TaskHistory"; @@ -328,6 +329,12 @@ export function GlobalTasks() { + = visible.iter().map(|task| task.task_number).collect(); + let attempts = deps + .task_store + .prior_attempt_summaries(&numbers) + .await + .unwrap_or_else(|error| { + tracing::warn!(%error, "failed to load task attempt history for the board"); + std::collections::HashMap::new() + }); + output.push_str(&format!("### {label}\n")); for task in visible { - output.push_str(&render_task_line(&task, &deps.agent_id)); + output.push_str(&render_task_line( + &task, + &deps.agent_id, + attempts.get(&task.task_number).map(String::as_str), + )); } output.push('\n'); } @@ -711,7 +728,7 @@ async fn render_task_state( Ok((output, any)) } -fn render_task_line(task: &Task, agent_id: &str) -> String { +fn render_task_line(task: &Task, agent_id: &str, prior_attempts: Option<&str>) -> String { let ownership = match task.assigned_agent_id.as_deref() { Some(assigned) if assigned == agent_id => String::new(), Some(assigned) => format!(" (assigned to {assigned})"), @@ -751,6 +768,10 @@ fn render_task_line(task: &Task, agent_id: &str) -> String { if let Some(parent) = task.stack_parent() { line.push_str(&format!(" [stacks on #{parent}]")); } + // What has already been tried, so a run does not repeat failed work. + if let Some(attempts) = prior_attempts { + line.push_str(&format!(" [{attempts}]")); + } line.push('\n'); line } diff --git a/src/agent/channel_dispatch.rs b/src/agent/channel_dispatch.rs index db34d1fd6..3e8a58382 100644 --- a/src/agent/channel_dispatch.rs +++ b/src/agent/channel_dispatch.rs @@ -129,6 +129,8 @@ fn classify_worker_completion( } } +/// How a run ended, as a task's attempt history records it. +/// fn completion_flags(kind: WorkerCompletionKind) -> (bool, bool) { let notify = true; let success = matches!( @@ -988,6 +990,7 @@ async fn spawn_worker_inner( None, None, secrets_store, + Some(state.deps.task_store.clone()), "builtin", worker.run().instrument(worker_span), ); @@ -1222,6 +1225,7 @@ async fn spawn_opencode_worker_inner( Some(opencode_cancellation), Some(directory_claim), oc_secrets_store, + Some(state.deps.task_store.clone()), "opencode", async move { let result = worker.run().await.map_err(SpacebotError::from); @@ -1294,6 +1298,8 @@ pub(crate) fn spawn_worker_task( >, opencode_directory_claim: Option, secrets_store: Option>, + // Present when the run should be recorded against a task's history. + task_store: Option>, #[cfg_attr(not(feature = "metrics"), allow(unused_variables))] worker_type: &'static str, future: F, ) -> WorkerTaskControl @@ -1381,6 +1387,7 @@ where }; let (notify, _success) = completion_flags(kind); let outcome_kind = outcome_kind(kind); + #[cfg(feature = "metrics")] { let metrics = crate::telemetry::Metrics::global(); @@ -1412,6 +1419,31 @@ where terminal_owner, ) .await; + + // Close this run in the task's attempt history, using the outcome the + // commit settled on: a completion racing a cancel or a timeout lands on + // a different terminal kind than the raw classification, and the board + // has to agree with the durable worker record. A commit that produced + // nothing still closes the attempt with what was classified here, so a + // failure to commit cannot leave the task blocked by an open run. + // Keyed by worker id, so a run never bound to a task matches nothing. + if let Some(task_store) = &task_store { + let (resolved, summary_source) = match &commit { + Ok(Some((terminal, _))) => (terminal.outcome_kind, terminal.result.as_str()), + _ => (outcome_kind, result_text.as_str()), + }; + if let Err(error) = task_store + .finish_task_attempt( + &worker_id.to_string(), + resolved.into(), + Some(summary_source), + ) + .await + { + tracing::warn!(%error, %worker_id, "failed to record the task attempt outcome"); + } + } + let (terminal, newly_committed) = match commit { Ok(Some(commit)) => commit, Ok(None) => { @@ -1746,6 +1778,7 @@ pub async fn resume_idle_worker_into_state( Some(opencode_cancellation), Some(directory_claim), oc_secrets_store, + Some(state.deps.task_store.clone()), "opencode", async move { let result = worker.run().await.map_err(SpacebotError::from)?; @@ -1876,6 +1909,7 @@ pub async fn resume_idle_worker_into_state( None, None, secrets_store, + Some(state.deps.task_store.clone()), "builtin", worker.run().instrument(worker_span), ); @@ -1927,8 +1961,14 @@ fn expand_tilde(path: &str) -> std::path::PathBuf { #[cfg(test)] mod tests { - use super::{WorkerCompletionError, WorkerOutcome, map_worker_completion, spawn_worker_task}; - use crate::conversation::ProcessRunLogger; + use super::{ + WorkerCompletionError, WorkerOutcome, commit_worker_outcome, map_worker_completion, + spawn_worker_task, + }; + use crate::conversation::{ + ProcessRunLogger, WorkerLifecycle, WorkerOutcomeKind, WorkerTerminalOwner, + }; + use crate::tasks::TaskAttemptOutcome; use crate::{ProcessEvent, WorkerId}; use std::sync::Arc; use std::time::Duration; @@ -1965,6 +2005,85 @@ mod tests { logger } + /// A cancel arriving while the worker is already completing commits as + /// partial. The attempt has to record what was committed: recording the raw + /// classification would put `cancelled` on the board against a worker record + /// that says `partial`. + #[tokio::test] + async fn a_cancel_racing_a_completion_records_what_was_committed() { + let worker_id = Uuid::new_v4(); + let logger = setup_worker(worker_id, "test:race-cancel").await; + let lifecycle = logger + .read_worker_lifecycle(worker_id) + .await + .unwrap() + .unwrap(); + logger + .claim_worker_completion(worker_id, lifecycle) + .await + .unwrap(); + + let (terminal, committed) = commit_worker_outcome( + &logger, + worker_id, + WorkerOutcomeKind::Cancelled, + "cancelled while finishing", + None, + WorkerTerminalOwner::Cancel, + ) + .await + .unwrap() + .unwrap(); + + assert!(committed); + assert_eq!(terminal.outcome_kind, WorkerOutcomeKind::Partial); + assert_eq!( + TaskAttemptOutcome::from(terminal.outcome_kind), + TaskAttemptOutcome::Partial + ); + assert_ne!( + TaskAttemptOutcome::from(WorkerOutcomeKind::Cancelled), + TaskAttemptOutcome::from(terminal.outcome_kind), + "the raw classification is what the attempt used to record" + ); + } + + /// The same disagreement in the other direction: a timeout landing on a + /// worker already cancelling, with nothing to show for the run, commits as + /// cancelled rather than timed out. + #[tokio::test] + async fn a_timeout_racing_a_cancel_records_what_was_committed() { + let worker_id = Uuid::new_v4(); + let logger = setup_worker(worker_id, "test:race-timeout").await; + let lifecycle = logger + .read_worker_lifecycle(worker_id) + .await + .unwrap() + .unwrap(); + logger + .transition_worker(worker_id, lifecycle, WorkerLifecycle::Cancelling) + .await + .unwrap(); + + let (terminal, _) = commit_worker_outcome( + &logger, + worker_id, + WorkerOutcomeKind::TimedOut, + "timed out", + None, + WorkerTerminalOwner::Timeout, + ) + .await + .unwrap() + .unwrap(); + + assert_eq!(terminal.outcome_kind, WorkerOutcomeKind::Cancelled); + assert_eq!( + TaskAttemptOutcome::from(terminal.outcome_kind), + TaskAttemptOutcome::Cancelled + ); + } + #[test] fn cancelled_errors_are_classified_as_cancelled_results() { let (text, notify, success) = @@ -2048,6 +2167,7 @@ mod tests { None, None, None, + None, "builtin", async { Err::( @@ -2102,6 +2222,7 @@ mod tests { None, None, None, + None, "builtin", async move { started_tx.send(()).expect("test receiver remains active"); @@ -2147,6 +2268,7 @@ mod tests { None, None, None, + None, "builtin", async { Ok::(WorkerOutcome::Success { @@ -2194,6 +2316,7 @@ mod tests { None, None, None, + None, "builtin", async { Ok::(WorkerOutcome::Success { diff --git a/src/api/server.rs b/src/api/server.rs index 1f59860b3..89c82f231 100644 --- a/src/api/server.rs +++ b/src/api/server.rs @@ -163,6 +163,7 @@ pub fn api_router() -> OpenApiRouter> { tasks::list_task_comments, tasks::create_task_comment )) + .routes(routes!(tasks::list_task_attempts)) .routes(routes!(tasks::list_task_revisions)) .routes(routes!(tasks::diff_task_revisions)) .routes(routes!(tasks::get_task_revision)) diff --git a/src/api/tasks.rs b/src/api/tasks.rs index d835f0801..c3181d7ef 100644 --- a/src/api/tasks.rs +++ b/src/api/tasks.rs @@ -351,6 +351,14 @@ pub struct TaskCommentResponse { pub comment: crate::tasks::TaskComment, } +#[derive(Serialize, Deserialize, utoipa::ToSchema)] +pub struct TaskAttemptListResponse { + /// Worker runs attempted against this task, newest first. + pub attempts: Vec, + /// One line summarising what has been tried, as prompt context renders it. + pub summary: Option, +} + #[derive(Serialize, Deserialize, utoipa::ToSchema)] pub struct TaskHistoryResponse { pub revisions: Vec, @@ -752,6 +760,7 @@ pub(super) async fn update_task( repo_id: request.repo_id, worktree_mode: request.worktree_mode, worktree_id: request.worktree_id, + goal_id: None, required_skills: request.required_skills, context, }, @@ -1050,6 +1059,41 @@ pub(super) async fn list_task_comments( })) } +/// `GET /tasks/{number}/attempts` — the worker runs attempted against a task. +/// +/// `tasks.worker_id` names only the run executing right now; this is the +/// history that says what has already been tried and how it ended. +#[utoipa::path( + get, + path = "/tasks/{number}/attempts", + params(("number" = i64, Path, description = "Task number")), + responses( + (status = 200, body = TaskAttemptListResponse), + (status = 404, description = "Task not found", body = TaskErrorBody), + (status = 503, description = "Task store not initialized", body = TaskErrorBody), + ), + tag = "tasks", +)] +pub(super) async fn list_task_attempts( + State(state): State>, + Path(number): Path, +) -> Result, TaskApiError> { + let store = task_store(&state)?; + + // Distinguish "never attempted" from "no such task". + store + .get_by_number(number) + .await? + .ok_or_else(|| TaskApiError::not_found(number))?; + + let attempts = store + .list_task_attempts(number, crate::tasks::MAX_ATTEMPT_PAGE) + .await?; + let summary = crate::tasks::render_prior_attempts(&attempts); + + Ok(Json(TaskAttemptListResponse { attempts, summary })) +} + /// `POST /tasks/{number}/comments` — append a comment to a task. #[utoipa::path( post, diff --git a/src/llm/history_repair.rs b/src/llm/history_repair.rs index 41e2b5ff4..e060547bd 100644 --- a/src/llm/history_repair.rs +++ b/src/llm/history_repair.rs @@ -11,23 +11,56 @@ //! itself past a stranded result, which keeps the surrounding turn intact and //! is the better place to solve it. This pass is the guarantee underneath them: //! whatever assembled the history, what leaves for the provider pairs. +//! +//! An unpairable result is rewritten as delimited plain text rather than +//! discarded. The content is often the most expensive thing in the history — +//! the output of a long shell command or a file read — and it stays useful to +//! the model as prose once it can no longer be a protocol message. The +//! delimiters mark it as historical data so it cannot read as an instruction. -use rig::message::{AssistantContent, Message, UserContent}; +use rig::message::{AssistantContent, Message, ToolResult, ToolResultContent, UserContent}; use rig::one_or_many::OneOrMany; -use std::collections::HashSet; +use std::collections::{HashMap, HashSet}; + +/// How much of a historical tool result survives as plain text. +const MAX_UNTRUSTED_RESULT_CHARS: usize = 1_024; + +/// What a repair pass changed, for logging and metrics. +#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)] +pub struct ToolHistoryRepair { + /// Results whose call is absent from the request entirely. + pub orphan_results: usize, + /// Results that appear at or before the call they claim. + pub stale_results: usize, + /// Second and later results claiming a call already answered. + pub duplicate_results: usize, + /// Calls left with no result, removed only when a provider rejected them. + pub unanswered_calls: usize, +} + +impl ToolHistoryRepair { + pub fn changed(&self) -> bool { + self.total() > 0 + } -/// Every identifier a provider might pair a result against. + pub fn total(&self) -> usize { + self.orphan_results + self.stale_results + self.duplicate_results + self.unanswered_calls + } +} + +/// Every identifier a provider might pair against, mapped to the position of +/// the call that carries it. /// /// The converters send `call_id` when it is present and non-empty and fall back /// to `id`, and the two halves of a pair do not always carry the same field — /// a result can hold a `call_id` where its call holds only an `id`. Collecting /// both from the call side and accepting either from the result side keeps the -/// match as permissive as the wire format allows, so a repair only ever removes +/// match as permissive as the wire format allows, so a repair only ever rewrites /// a result that no call in the request can claim under any pairing rule. -fn collect_call_identifiers(history: &OneOrMany) -> HashSet { - let mut identifiers = HashSet::new(); +fn call_positions(history: &[Message]) -> HashMap { + let mut positions = HashMap::new(); - for message in history.iter() { + for (index, message) in history.iter().enumerate() { let Message::Assistant { content, .. } = message else { continue; }; @@ -36,83 +69,295 @@ fn collect_call_identifiers(history: &OneOrMany) -> HashSet { continue; }; if !call.id.is_empty() { - identifiers.insert(call.id.clone()); + positions.entry(call.id.clone()).or_insert(index); } if let Some(call_id) = call.call_id.as_deref().filter(|id| !id.is_empty()) { - identifiers.insert(call_id.to_string()); + positions.entry(call_id.to_string()).or_insert(index); } } } - identifiers + positions +} + +/// The identifier a pair is keyed on, preferring the one providers send. +fn call_key(call: &rig::message::ToolCall) -> &str { + call.call_id + .as_deref() + .filter(|id| !id.is_empty()) + .unwrap_or(&call.id) +} + +fn result_key(result: &ToolResult) -> &str { + result + .call_id + .as_deref() + .filter(|id| !id.is_empty()) + .unwrap_or(&result.id) +} + +/// Where the call this result claims sits, under either identifier. +fn claiming_call(result: &ToolResult, positions: &HashMap) -> Option { + positions + .get(&result.id) + .or_else(|| { + result + .call_id + .as_deref() + .and_then(|call_id| positions.get(call_id)) + }) + .copied() +} + +/// Why a result cannot stay a protocol message. +#[derive(Clone, Copy, PartialEq, Eq)] +enum Unpairable { + Orphan, + Stale, + Duplicate, +} + +impl Unpairable { + fn reason(self) -> &'static str { + match self { + Self::Orphan => "no matching tool call in this request", + Self::Stale => "result recorded before the call it answers", + Self::Duplicate => "call already answered by an earlier result", + } + } +} + +fn classify( + result: &ToolResult, + index: usize, + positions: &HashMap, + answered: &mut HashSet, +) -> Option { + let Some(call_index) = claiming_call(result, positions) else { + return Some(Unpairable::Orphan); + }; + if index <= call_index { + return Some(Unpairable::Stale); + } + if !answered.insert(result_key(result).to_string()) { + return Some(Unpairable::Duplicate); + } + None } -/// Whether some tool call in the request claims this result. -fn is_claimed(result: &rig::message::ToolResult, identifiers: &HashSet) -> bool { - identifiers.contains(&result.id) - || result - .call_id - .as_deref() - .is_some_and(|call_id| identifiers.contains(call_id)) +fn bounded_result_text(result: &ToolResult) -> String { + let mut text = String::new(); + for item in result.content.iter() { + if !text.is_empty() { + text.push('\n'); + } + match item { + ToolResultContent::Text(value) => text.push_str(&value.text), + ToolResultContent::Image(_) => text.push_str("[historical image result omitted]"), + } + // A string shorter in bytes than the limit cannot exceed it in + // characters, so the count only runs once there is enough text to + // matter — a tool result can carry many items and be very long. + if text.len() >= MAX_UNTRUSTED_RESULT_CHARS + && text.chars().count() >= MAX_UNTRUSTED_RESULT_CHARS + { + break; + } + } + + match text.char_indices().nth(MAX_UNTRUSTED_RESULT_CHARS) { + Some((cut, _)) => { + let mut bounded = text[..cut].to_string(); + bounded.push_str("…[truncated]"); + bounded + } + None => text, + } } -/// Drop tool results that no call in `history` claims. +/// Rewrite a result as delimited historical data. /// -/// Returns the repaired history and the number of results dropped, or `None` -/// when every result is already paired — the common case, which allocates -/// nothing beyond the identifier set. +/// The delimiters are what make this safe to keep: the content came from a +/// tool, so it is not trusted input, and without a frame around it a shell +/// transcript can read as instructions once it is plain text. +fn historical_note(result: &ToolResult, verdict: Unpairable) -> UserContent { + UserContent::text(format!( + "[BEGIN UNTRUSTED HISTORICAL TOOL OUTPUT — {}; call id: {}]\n{}\n[END UNTRUSTED HISTORICAL TOOL OUTPUT]", + verdict.reason(), + result_key(result), + bounded_result_text(result) + )) +} + +/// Rewrite tool results this request cannot pair as delimited plain text. /// -/// A user message reduced to nothing is dropped along with its results. Text -/// and images in the same message survive, so a turn that mixes a prompt with a -/// stranded result keeps the prompt. -pub fn repair_orphaned_tool_results(history: &OneOrMany) -> Option<(Vec, usize)> { - let identifiers = collect_call_identifiers(history); +/// Returns the repaired history and what changed, or `None` when every result +/// already pairs — the common case, which allocates nothing beyond the +/// identifier map. +pub fn repair_orphaned_tool_results( + history: &OneOrMany, +) -> Option<(Vec, ToolHistoryRepair)> { + let positions = call_positions(history.iter().cloned().collect::>().as_slice()); - let orphaned = history - .iter() - .filter_map(|message| match message { - Message::User { content } => Some(content), - _ => None, - }) - .flat_map(|content| content.iter()) - .filter(|item| match item { - UserContent::ToolResult(result) => !is_claimed(result, &identifiers), - _ => false, - }) - .count(); + let mut report = ToolHistoryRepair::default(); + let mut answered = HashSet::new(); + let mut verdicts: Vec>> = Vec::with_capacity(history.len()); - if orphaned == 0 { + for (index, message) in history.iter().enumerate() { + let Message::User { content } = message else { + verdicts.push(Vec::new()); + continue; + }; + let mut row = Vec::new(); + for item in content.iter() { + let verdict = match item { + UserContent::ToolResult(result) => { + classify(result, index, &positions, &mut answered) + } + _ => None, + }; + match verdict { + Some(Unpairable::Orphan) => report.orphan_results += 1, + Some(Unpairable::Stale) => report.stale_results += 1, + Some(Unpairable::Duplicate) => report.duplicate_results += 1, + None => {} + } + row.push(verdict); + } + verdicts.push(row); + } + + if !report.changed() { return None; } let mut repaired = Vec::with_capacity(history.len()); - for message in history.iter() { + for (message, row) in history.iter().zip(verdicts) { let Message::User { content } = message else { repaired.push(message.clone()); continue; }; - let kept: Vec = content + let rewritten: Vec = content .iter() + .zip(row) + .map(|(item, verdict)| match (item, verdict) { + (UserContent::ToolResult(result), Some(verdict)) => { + historical_note(result, verdict) + } + (item, _) => item.clone(), + }) + .collect(); + + if let Ok(content) = OneOrMany::many(rewritten) { + repaired.push(Message::User { content }); + } + } + + Some((repaired, report)) +} + +/// Remove assistant tool calls that nothing in the history answers. +/// +/// Anthropic rejects a `tool_use` with no following `tool_result`, which the +/// result-side pass cannot fix because there is no result to rewrite. A call +/// still awaiting its result is the normal shape mid-loop, so this only runs +/// after a provider has already rejected the request, and never touches the +/// final assistant message. +pub fn drop_unanswered_tool_calls(history: &mut Vec) -> ToolHistoryRepair { + let mut answered: HashSet = HashSet::new(); + for message in history.iter() { + let Message::User { content } = message else { + continue; + }; + for item in content.iter() { + if let UserContent::ToolResult(result) = item { + answered.insert(result.id.clone()); + if let Some(call_id) = result.call_id.clone() { + answered.insert(call_id); + } + } + } + } + + let last_assistant = history + .iter() + .rposition(|message| matches!(message, Message::Assistant { .. })); + + let mut report = ToolHistoryRepair::default(); + let mut rebuilt = Vec::with_capacity(history.len()); + + for (index, message) in history.drain(..).enumerate() { + let Message::Assistant { id, content } = message else { + rebuilt.push(message); + continue; + }; + + if Some(index) == last_assistant { + rebuilt.push(Message::Assistant { id, content }); + continue; + } + + let kept: Vec = content + .into_iter() .filter(|item| match item { - UserContent::ToolResult(result) => is_claimed(result, &identifiers), + AssistantContent::ToolCall(call) => { + let paired = answered.contains(call_key(call)) + || answered.contains(&call.id) + || call + .call_id + .as_deref() + .is_some_and(|id| answered.contains(id)); + if !paired { + report.unanswered_calls += 1; + } + paired + } _ => true, }) - .cloned() .collect(); if let Ok(content) = OneOrMany::many(kept) { - repaired.push(Message::User { content }); + rebuilt.push(Message::Assistant { id, content }); } } - Some((repaired, orphaned)) + *history = rebuilt; + report +} + +/// Report the first pairing violation a provider would reject, if any. +/// +/// Used for observability after a cut rather than as a gate: a call still +/// awaiting its result is valid mid-loop, so an unanswered trailing call is +/// deliberately not a violation here. +pub fn validate_tool_history(history: &[Message]) -> Result<(), String> { + let positions = call_positions(history); + let mut answered = HashSet::new(); + + for (index, message) in history.iter().enumerate() { + let Message::User { content } = message else { + continue; + }; + for item in content.iter() { + let UserContent::ToolResult(result) = item else { + continue; + }; + match classify(result, index, &positions, &mut answered) { + Some(verdict) => { + return Err(format!("{}: {}", result_key(result), verdict.reason())); + } + None => continue, + } + } + } + + Ok(()) } #[cfg(test)] mod tests { use super::*; - use rig::message::{ToolResult, ToolResultContent}; fn tool_call(id: &str, call_id: Option<&str>) -> Message { Message::Assistant { @@ -131,10 +376,14 @@ mod tests { } fn tool_result(id: &str, call_id: Option<&str>) -> UserContent { + tool_result_with(id, call_id, "ok") + } + + fn tool_result_with(id: &str, call_id: Option<&str>, body: &str) -> UserContent { UserContent::ToolResult(ToolResult { id: id.to_string(), call_id: call_id.map(str::to_string), - content: OneOrMany::one(ToolResultContent::text("ok")), + content: OneOrMany::one(ToolResultContent::text(body)), }) } @@ -144,6 +393,20 @@ mod tests { } } + fn text_of(message: &Message) -> String { + let Message::User { content } = message else { + return String::new(); + }; + content + .iter() + .filter_map(|item| match item { + UserContent::Text(text) => Some(text.text.clone()), + _ => None, + }) + .collect::>() + .join("\n") + } + #[test] fn paired_history_is_left_alone() { let history = OneOrMany::many(vec![ @@ -153,22 +416,64 @@ mod tests { .expect("non-empty"); assert!(repair_orphaned_tool_results(&history).is_none()); + assert!(validate_tool_history(&history.iter().cloned().collect::>()).is_ok()); } /// A cut that lands between a call and its result leaves the result at the - /// front of the history with nothing to pair against. + /// front of the history with nothing to pair against. The output it carried + /// is preserved as prose rather than thrown away. #[test] - fn stranded_result_is_dropped_with_its_message() { + fn stranded_result_becomes_untrusted_text() { let history = OneOrMany::many(vec![ - results(vec![tool_result("call_gone", None)]), + results(vec![tool_result_with( + "call_gone", + None, + "total 48\ndrwxr-xr-x", + )]), Message::from("carry on"), ]) .expect("non-empty"); - let (repaired, dropped) = repair_orphaned_tool_results(&history).expect("repair"); - assert_eq!(dropped, 1); - assert_eq!(repaired.len(), 1); - assert!(matches!(repaired[0], Message::User { .. })); + let (repaired, report) = repair_orphaned_tool_results(&history).expect("repair"); + assert_eq!(report.orphan_results, 1); + assert_eq!(repaired.len(), 2); + + let note = text_of(&repaired[0]); + assert!(note.contains("UNTRUSTED HISTORICAL TOOL OUTPUT")); + assert!(note.contains("no matching tool call")); + assert!(note.contains("call_gone")); + assert!(note.contains("drwxr-xr-x"), "the output itself survives"); + + // Nothing pairs any more, so the request is now valid. + assert!(validate_tool_history(&repaired).is_ok()); + } + + /// The shape that took down a live worker: a fork's compaction cut removed + /// the assistant turn holding the first `read_skill` call while its result + /// stayed at the head of the retained history. + #[test] + fn a_forked_worker_history_cut_mid_turn_is_repaired() { + let history = OneOrMany::many(vec![ + results(vec![tool_result_with( + "call_HPJ4d0Mb42LJt6JzCqYcwRsq", + None, + "# Skill: instance-debugging", + )]), + tool_call("fc_next", Some("call_next")), + results(vec![tool_result("call_next", None)]), + ]) + .expect("non-empty"); + + assert!( + validate_tool_history(&history.iter().cloned().collect::>()).is_err(), + "the history a provider rejected must fail validation" + ); + + let (repaired, report) = repair_orphaned_tool_results(&history).expect("repair"); + assert_eq!(report.orphan_results, 1); + assert_eq!(report.total(), 1, "the intact pair is untouched"); + assert!(validate_tool_history(&repaired).is_ok()); + assert!(text_of(&repaired[0]).contains("instance-debugging")); } /// Providers pair on `call_id` when the call carries one, so a result @@ -198,9 +503,10 @@ mod tests { } /// One parallel call batch, one result of which lost its call: the batch's - /// surviving results stay, and only the stranded one goes. + /// surviving results stay protocol messages and only the stranded one is + /// rewritten. #[test] - fn only_the_unclaimed_result_of_a_batch_is_dropped() { + fn only_the_unclaimed_result_of_a_batch_is_rewritten() { let history = OneOrMany::many(vec![ tool_call("fc_kept", Some("call_kept")), results(vec![ @@ -210,30 +516,47 @@ mod tests { ]) .expect("non-empty"); - let (repaired, dropped) = repair_orphaned_tool_results(&history).expect("repair"); - assert_eq!(dropped, 1); - assert_eq!(repaired.len(), 2); + let (repaired, report) = repair_orphaned_tool_results(&history).expect("repair"); + assert_eq!(report.orphan_results, 1); + let Message::User { content } = &repaired[1] else { panic!("expected the result message to survive"); }; - assert_eq!(content.iter().count(), 1); + let kinds: Vec = content + .iter() + .map(|item| matches!(item, UserContent::ToolResult(_))) + .collect(); + assert_eq!(kinds, vec![true, false], "one stays a result, one is prose"); } - /// A history that is nothing but orphans repairs to no messages at all. - /// `OneOrMany` cannot represent that, so the caller has to turn it into an - /// error rather than send a request a provider will reject. + /// A second result for a call already answered is rejected by providers as + /// firmly as an orphan. #[test] - fn an_orphan_only_history_repairs_to_nothing() { + fn a_duplicate_result_is_rewritten() { let history = OneOrMany::many(vec![ - results(vec![tool_result("call_gone", None)]), - results(vec![tool_result("call_also_gone", None)]), + tool_call("fc_1", Some("call_1")), + results(vec![tool_result("call_1", None)]), + results(vec![tool_result("call_1", None)]), + ]) + .expect("non-empty"); + + let (repaired, report) = repair_orphaned_tool_results(&history).expect("repair"); + assert_eq!(report.duplicate_results, 1); + assert_eq!(report.orphan_results, 0); + assert!(validate_tool_history(&repaired).is_ok()); + } + + /// A result placed at or before its call cannot be paired in order. + #[test] + fn a_stale_result_is_rewritten() { + let history = OneOrMany::many(vec![ + results(vec![tool_result("call_1", None)]), + tool_call("fc_1", Some("call_1")), ]) .expect("non-empty"); - let (repaired, dropped) = repair_orphaned_tool_results(&history).expect("repair"); - assert_eq!(dropped, 2); - assert!(repaired.is_empty()); - assert!(OneOrMany::many(repaired).is_err()); + let (_, report) = repair_orphaned_tool_results(&history).expect("repair"); + assert_eq!(report.stale_results, 1); } /// A turn that mixes a stranded result with real prompt text keeps the text. @@ -245,15 +568,69 @@ mod tests { ])]) .expect("non-empty"); - let (repaired, dropped) = repair_orphaned_tool_results(&history).expect("repair"); - assert_eq!(dropped, 1); - assert_eq!(repaired.len(), 1); - let Message::User { content } = &repaired[0] else { - panic!("expected a user message"); - }; - assert!(matches!( - content.iter().next(), - Some(UserContent::Text(text)) if text.text == "what did you find?" - )); + let (repaired, report) = repair_orphaned_tool_results(&history).expect("repair"); + assert_eq!(report.orphan_results, 1); + assert!(text_of(&repaired[0]).contains("what did you find?")); + } + + /// Every message is preserved, so a history of nothing but orphans still + /// produces a sendable request instead of an empty one. + #[test] + fn an_orphan_only_history_still_produces_messages() { + let history = OneOrMany::many(vec![ + results(vec![tool_result("call_gone", None)]), + results(vec![tool_result("call_also_gone", None)]), + ]) + .expect("non-empty"); + + let (repaired, report) = repair_orphaned_tool_results(&history).expect("repair"); + assert_eq!(report.orphan_results, 2); + assert_eq!(repaired.len(), 2); + assert!(OneOrMany::many(repaired).is_ok()); + } + + /// Long output is bounded so a repair cannot blow the context it was + /// trimmed to fit. + #[test] + fn a_long_result_is_truncated_in_the_note() { + let body = "x".repeat(MAX_UNTRUSTED_RESULT_CHARS * 3); + let history = OneOrMany::many(vec![results(vec![tool_result_with("gone", None, &body)])]) + .expect("non-empty"); + + let (repaired, _) = repair_orphaned_tool_results(&history).expect("repair"); + let note = text_of(&repaired[0]); + assert!(note.contains("…[truncated]")); + assert!(note.chars().count() < MAX_UNTRUSTED_RESULT_CHARS + 300); + } + + /// An unanswered call mid-history is what Anthropic rejects; the trailing + /// one is a loop still in flight and must survive. + #[test] + fn only_a_non_trailing_unanswered_call_is_dropped() { + let mut history = vec![ + tool_call("fc_dead", Some("call_dead")), + Message::from("unrelated turn"), + tool_call("fc_live", Some("call_live")), + ]; + + let report = drop_unanswered_tool_calls(&mut history); + + assert_eq!(report.unanswered_calls, 1); + assert_eq!(history.len(), 2, "the emptied assistant message goes too"); + assert!(matches!(history[1], Message::Assistant { .. })); + } + + #[test] + fn answered_calls_are_never_dropped() { + let mut history = vec![ + tool_call("fc_1", Some("call_1")), + results(vec![tool_result("call_1", None)]), + Message::from("later"), + ]; + + let report = drop_unanswered_tool_calls(&mut history); + + assert_eq!(report.unanswered_calls, 0); + assert_eq!(history.len(), 3); } } diff --git a/src/llm/model.rs b/src/llm/model.rs index ab53ca61e..c78c77548 100644 --- a/src/llm/model.rs +++ b/src/llm/model.rs @@ -151,48 +151,94 @@ impl SpacebotModel { } } - /// Drop tool results this request cannot pair before it reaches a provider. + /// Rewrite tool results this request cannot pair before it reaches a + /// provider. /// /// A stranded result is rejected at the API boundary, so the model never /// runs and a retry of the same history fails the same way. Repairing here /// covers every caller regardless of which trim produced the history, and - /// the warning names what went so a cut that keeps stranding results is + /// the warning names what changed so a cut that keeps stranding results is /// still visible rather than silently absorbed. - /// - /// A history that repairs to nothing has no request left to send: the - /// provider requires at least one message, so this reports the empty - /// history rather than spending a call that is certain to be rejected. fn repair_request_history( &self, request: &mut CompletionRequest, ) -> Result<(), CompletionError> { - let Some((repaired, dropped)) = + let Some((repaired, report)) = crate::llm::history_repair::repair_orphaned_tool_results(&request.chat_history) else { return Ok(()); }; let Ok(chat_history) = OneOrMany::many(repaired) else { - tracing::error!( - model = %self.full_model_name, - dropped, - "request history is entirely unpaired tool results" - ); return Err(CompletionError::RequestError( - format!("request history is {dropped} unpaired tool results and nothing else") - .into(), + "request history repaired to no messages".into(), )); }; tracing::warn!( model = %self.full_model_name, - dropped, - "dropped tool results with no matching call from request history" + orphan_results = report.orphan_results, + stale_results = report.stale_results, + duplicate_results = report.duplicate_results, + "rewrote unpairable tool results as historical text" ); + #[cfg(feature = "metrics")] + crate::telemetry::Metrics::global() + .tool_history_recovery_total + .with_label_values(&[self.agent_id.as_deref().unwrap_or("unknown"), "repaired"]) + .inc(); + request.chat_history = chat_history; Ok(()) } + /// Repair a history a provider has already rejected, for one retry. + /// + /// The pre-send pass pairs every result, so a mismatch that survives it is + /// the other half of the protocol: an assistant tool call nothing answers, + /// which Anthropic rejects and which no result-side repair can reach. + /// Returns `false` when nothing changed, so an identical request is never + /// sent twice. + fn escalate_tool_history_repair(&self, request: &mut CompletionRequest) -> bool { + let mut history: Vec = + request.chat_history.iter().cloned().collect(); + let report = crate::llm::history_repair::drop_unanswered_tool_calls(&mut history); + + if !report.changed() { + return false; + } + + let Ok(chat_history) = OneOrMany::many(history) else { + return false; + }; + + tracing::warn!( + model = %self.full_model_name, + unanswered_calls = report.unanswered_calls, + "dropped unanswered tool calls after a provider rejected the history" + ); + request.chat_history = chat_history; + true + } + + /// Record how a tool-history retry ended, for both request paths. + fn record_tool_history_recovery(&self, succeeded: bool) { + #[cfg(feature = "metrics")] + crate::telemetry::Metrics::global() + .tool_history_recovery_total + .with_label_values(&[ + self.agent_id.as_deref().unwrap_or("unknown"), + if succeeded { + "retry_success" + } else { + "terminal_failure" + }, + ]) + .inc(); + #[cfg(not(feature = "metrics"))] + let _ = succeeded; + } + /// Direct call to the provider (no fallback logic). async fn attempt_completion( &self, @@ -386,6 +432,139 @@ impl SpacebotModel { was_rate_limit, )) } + + /// Run a prepared request through routing, retries and the fallback chain. + /// + /// Borrows the request: `attempt_with_retries` clones per attempt, so only + /// the unrouted path below needs a copy of its own. + async fn dispatch_completion( + &self, + request: &CompletionRequest, + ) -> Result, CompletionError> { + let Some(routing) = &self.routing else { + // No routing config — just call the model directly, no fallback/retry + return self.attempt_completion(request.clone()).await; + }; + + let cooldown = routing.rate_limit_cooldown_secs; + let mut fallbacks: Vec = routing.get_fallbacks(&self.full_model_name).to_vec(); + // Set when the configured model id was rejected outright and the + // fallbacks were derived from the provider's default routing table. + let mut provider_recovery = false; + let mut last_error: Option = None; + + // Try the primary model (with retries) unless it's in rate-limit cooldown + // and we have fallbacks to try instead. + let primary_rate_limited = self + .llm_manager + .is_rate_limited(&self.full_model_name, cooldown) + .await; + + let skip_primary = primary_rate_limited && !fallbacks.is_empty(); + + if skip_primary { + tracing::debug!( + model = %self.full_model_name, + "primary model in rate-limit cooldown, skipping to fallbacks" + ); + } else { + match self + .attempt_with_retries(&self.full_model_name, request) + .await + { + Ok(response) => return Ok(response), + Err((error, was_rate_limit)) => { + if was_rate_limit { + self.llm_manager + .record_rate_limit(&self.full_model_name) + .await; + } + // A rejected model id (stale default, typo, no access) never + // recovers on its own — try the provider's default models + // when no explicit chain is configured. + if fallbacks.is_empty() && routing::is_model_not_found_error(&error.to_string()) + { + fallbacks = routing::default_model_candidates(&self.provider) + .into_iter() + .filter(|candidate| candidate != &self.full_model_name) + .collect(); + provider_recovery = !fallbacks.is_empty(); + if provider_recovery { + tracing::warn!( + model = %self.full_model_name, + candidates = ?fallbacks, + "provider rejected configured model, trying its default models" + ); + } + } + if fallbacks.is_empty() { + // No fallbacks — this is the final error + return Err(error); + } + if !provider_recovery { + tracing::warn!( + model = %self.full_model_name, + "primary model exhausted retries, trying fallbacks" + ); + } + last_error = Some(error); + } + } + } + + // Try fallback chain, each with their own retry loop + for (index, fallback_name) in fallbacks.iter().take(MAX_FALLBACK_ATTEMPTS).enumerate() { + if self + .llm_manager + .is_rate_limited(fallback_name, cooldown) + .await + { + tracing::debug!( + fallback = %fallback_name, + "fallback model in cooldown, skipping" + ); + continue; + } + + match self.attempt_with_retries(fallback_name, request).await { + Ok(response) => { + tracing::info!( + original = %self.full_model_name, + fallback = %fallback_name, + attempt = index + 1, + "fallback model succeeded" + ); + return Ok(response); + } + Err((error, was_rate_limit)) => { + if was_rate_limit { + self.llm_manager.record_rate_limit(fallback_name).await; + } + tracing::warn!( + fallback = %fallback_name, + "fallback model exhausted retries, continuing chain" + ); + last_error = Some(error); + } + } + } + + let final_error = last_error.unwrap_or_else(|| { + CompletionError::ProviderError("all models in fallback chain failed".into()) + }); + if provider_recovery && routing::is_model_not_found_error(&final_error.to_string()) { + return Err(CompletionError::ProviderError(format!( + "provider '{}' rejected the configured model '{}' and every default \ + candidate ({}). Spacebot's built-in model ids for this provider appear \ + to be stale — pick a working model in Settings → Model Routing and \ + please report this as a bug.", + self.provider, + self.full_model_name, + fallbacks.join(", ") + ))); + } + Err(final_error) + } } impl CompletionModel for SpacebotModel { @@ -430,133 +609,19 @@ impl CompletionModel for SpacebotModel { self.repair_request_history(&mut request)?; - let result = async move { - let Some(routing) = &self.routing else { - // No routing config — just call the model directly, no fallback/retry - return self.attempt_completion(request).await; - }; - - let cooldown = routing.rate_limit_cooldown_secs; - let mut fallbacks: Vec = routing.get_fallbacks(&self.full_model_name).to_vec(); - // Set when the configured model id was rejected outright and the - // fallbacks were derived from the provider's default routing table. - let mut provider_recovery = false; - let mut last_error: Option = None; - - // Try the primary model (with retries) unless it's in rate-limit cooldown - // and we have fallbacks to try instead. - let primary_rate_limited = self - .llm_manager - .is_rate_limited(&self.full_model_name, cooldown) - .await; - - let skip_primary = primary_rate_limited && !fallbacks.is_empty(); - - if skip_primary { - tracing::debug!( - model = %self.full_model_name, - "primary model in rate-limit cooldown, skipping to fallbacks" - ); - } else { - match self - .attempt_with_retries(&self.full_model_name, &request) - .await - { - Ok(response) => return Ok(response), - Err((error, was_rate_limit)) => { - if was_rate_limit { - self.llm_manager - .record_rate_limit(&self.full_model_name) - .await; - } - // A rejected model id (stale default, typo, no access) never - // recovers on its own — try the provider's default models - // when no explicit chain is configured. - if fallbacks.is_empty() - && routing::is_model_not_found_error(&error.to_string()) - { - fallbacks = routing::default_model_candidates(&self.provider) - .into_iter() - .filter(|candidate| candidate != &self.full_model_name) - .collect(); - provider_recovery = !fallbacks.is_empty(); - if provider_recovery { - tracing::warn!( - model = %self.full_model_name, - candidates = ?fallbacks, - "provider rejected configured model, trying its default models" - ); - } - } - if fallbacks.is_empty() { - // No fallbacks — this is the final error - return Err(error); - } - if !provider_recovery { - tracing::warn!( - model = %self.full_model_name, - "primary model exhausted retries, trying fallbacks" - ); - } - last_error = Some(error); - } - } - } - - // Try fallback chain, each with their own retry loop - for (index, fallback_name) in fallbacks.iter().take(MAX_FALLBACK_ATTEMPTS).enumerate() { - if self - .llm_manager - .is_rate_limited(fallback_name, cooldown) - .await - { - tracing::debug!( - fallback = %fallback_name, - "fallback model in cooldown, skipping" - ); - continue; - } + let mut result = self.dispatch_completion(&request).await; - match self.attempt_with_retries(fallback_name, &request).await { - Ok(response) => { - tracing::info!( - original = %self.full_model_name, - fallback = %fallback_name, - attempt = index + 1, - "fallback model succeeded" - ); - return Ok(response); - } - Err((error, was_rate_limit)) => { - if was_rate_limit { - self.llm_manager.record_rate_limit(fallback_name).await; - } - tracing::warn!( - fallback = %fallback_name, - "fallback model exhausted retries, continuing chain" - ); - last_error = Some(error); - } - } - } - - let final_error = last_error.unwrap_or_else(|| { - CompletionError::ProviderError("all models in fallback chain failed".into()) - }); - if provider_recovery && routing::is_model_not_found_error(&final_error.to_string()) { - return Err(CompletionError::ProviderError(format!( - "provider '{}' rejected the configured model '{}' and every default \ - candidate ({}). Spacebot's built-in model ids for this provider appear \ - to be stale — pick a working model in Settings → Model Routing and \ - please report this as a bug.", - self.provider, - self.full_model_name, - fallbacks.join(", ") - ))); - } - Err(final_error) + // A mismatch that survives the pre-send repair is the other half of the + // protocol: an assistant call nothing answers, which no result-side + // repair can reach. Retry once, and only when the history changed, so + // an identical request is never sent twice. + if let Err(ref error) = result + && routing::is_tool_history_mismatch_error(&error.to_string()) + && self.escalate_tool_history_repair(&mut request) + { + result = self.dispatch_completion(&request).await; + self.record_tool_history_recovery(result.is_ok()); } - .await; #[cfg(feature = "metrics")] { @@ -698,6 +763,30 @@ impl CompletionModel for SpacebotModel { ) -> Result, CompletionError> { self.repair_request_history(&mut request)?; + let mut result = self.dispatch_stream(request.clone()).await; + + // The channel agent streams, so it needs the same escalation as the + // non-streaming path. A provider rejects an assistant call nothing + // answers while opening the stream, before any token is yielded, so the + // repaired history can still be sent once more. + if let Err(ref error) = result + && routing::is_tool_history_mismatch_error(&error.to_string()) + && self.escalate_tool_history_repair(&mut request) + { + result = self.dispatch_stream(request).await; + self.record_tool_history_recovery(result.is_ok()); + } + + result + } +} + +impl SpacebotModel { + /// Open a stream against whichever provider the current model belongs to. + async fn dispatch_stream( + &self, + request: CompletionRequest, + ) -> Result, CompletionError> { let provider_config = self.provider_config_for_current_model().await?; match provider_config.api_type { diff --git a/src/llm/routing.rs b/src/llm/routing.rs index c0bf70b44..a4646585e 100644 --- a/src/llm/routing.rs +++ b/src/llm/routing.rs @@ -196,6 +196,26 @@ pub fn default_model_candidates(provider: &str) -> Vec { candidates } +/// Whether a provider rejected the request because the tool calls and results +/// in the submitted history do not form a valid protocol sequence. +/// +/// These 400s are deterministic for an unchanged request: the rejection lands +/// before the model runs, so a retry of the same history fails identically. +/// Recovery is only possible after the history itself changes. +pub fn is_tool_history_mismatch_error(error_message: &str) -> bool { + let lower = error_message.to_lowercase(); + lower.contains("no tool call found for function call output") + || lower.contains("unexpected tool_use_id") + || (lower.contains("tool_call_id") && lower.contains("did not have a response message")) + || (lower.contains("tool_use") + && lower.contains("without") + && lower.contains("tool_result")) + || (lower.contains("tool result") + && lower.contains("without") + && lower.contains("tool call")) + || (lower.contains("function call output") && lower.contains("call_id")) +} + /// Whether a completion error indicates context window overflow. /// /// Providers return 400 with various phrasings when the request exceeds @@ -605,6 +625,29 @@ mod tests { assert!(!is_retriable_error("parse error")); } + #[test] + fn is_tool_history_mismatch_error_detects_provider_400s() { + // The rejection that took down a live worker. + assert!(is_tool_history_mismatch_error( + "OpenAI ChatGPT Responses API error (400 Bad Request): No tool call found for \ + function call output with call_id call_HPJ4d0Mb42LJt6JzCqYcwRsq" + )); + // Anthropic phrasing for the mirror-image failure. + assert!(is_tool_history_mismatch_error( + "messages.4: `tool_use` ids were found without `tool_result` blocks immediately after" + )); + assert!(is_tool_history_mismatch_error( + "Invalid parameter: messages with role 'tool' must be a response to a preceding \ + message with 'tool_calls'. tool_call_id call_9 did not have a response message." + )); + + // Other 400s must not route into history repair. + assert!(!is_tool_history_mismatch_error("400 Bad Request")); + assert!(!is_tool_history_mismatch_error( + "context length exceeded: 210000 tokens" + )); + } + #[test] fn is_model_not_found_error_detection() { // OpenAI phrasing diff --git a/src/main.rs b/src/main.rs index 456232743..3f74595d4 100644 --- a/src/main.rs +++ b/src/main.rs @@ -952,6 +952,47 @@ async fn run( .await .context("failed to migrate legacy projects to instance database")?; + // Tasks executed before the worktree binding was recorded have a + // `task-` worktree on disk that nothing points at. Reconnect them + // by name so a retry reuses the worktree instead of rediscovering it. + { + let mut candidates = Vec::new(); + match global_project_store.list_projects(None).await { + Ok(projects) => { + for project in projects { + match global_project_store.list_worktrees(&project.id).await { + Ok(worktrees) => { + candidates.extend(worktrees.into_iter().map(|w| (w.name, w.id))) + } + Err(error) => { + tracing::warn!( + project_id = %project.id, + %error, + "failed to list worktrees for task binding backfill" + ); + } + } + } + } + Err(error) => { + tracing::warn!(%error, "failed to list projects for task binding backfill"); + } + } + + match global_task_store + .backfill_worktree_bindings(&candidates) + .await + { + Ok(bound) if bound > 0 => { + tracing::info!(tasks = bound, "bound tasks to their existing worktrees"); + } + Ok(_) => {} + Err(error) => { + tracing::warn!(%error, "failed to backfill task worktree bindings"); + } + } + } + // Start HTTP API server if enabled let mut api_state = spacebot::api::ApiState::new_with_provider_sender( provider_tx, @@ -1200,6 +1241,79 @@ async fn run( let mut deferred_injections: HashMap> = HashMap::new(); + // Workers run in-process, so any attempt still open belongs to a run that + // died with the previous process. Close them, or the task-scoped spawn + // guard would see a live run forever and that task could never be worked + // again. + // + // A run can reach a terminal state and still leave its attempt open: the + // worker record lives in the agent database and the attempt in the instance + // one, so nothing spans both writes. Where the worker did commit an outcome + // the attempt is closed with it, and only the runs nothing decided are swept + // as interrupted. This runs after the agents are open because recovering an + // outcome means reading the agent database that holds it. + if agents_initialized { + let live = match global_task_store.live_attempts().await { + Ok(live) => live, + Err(error) => { + tracing::warn!(%error, "failed to read live task attempts"); + Vec::new() + } + }; + for attempt in live { + let Some(agent) = attempt + .agent_id + .as_deref() + .and_then(|id| agents.get(&spacebot::AgentId::from(id))) + else { + continue; + }; + let Ok(worker_id) = attempt.worker_id.parse() else { + continue; + }; + let run_logger = spacebot::conversation::ProcessRunLogger::new(agent.db.sqlite.clone()); + let terminal = match run_logger.read_worker_terminal(worker_id).await { + Ok(Some(terminal)) => terminal, + Ok(None) => continue, + Err(error) => { + tracing::warn!(%error, %worker_id, "failed to read a worker terminal outcome"); + continue; + } + }; + match global_task_store + .finish_task_attempt( + &attempt.worker_id, + terminal.outcome_kind.into(), + Some(&terminal.result), + ) + .await + { + Ok(true) => tracing::info!( + %worker_id, + outcome = terminal.outcome_kind.as_str(), + "recovered a committed outcome for an attempt left open" + ), + Ok(false) => {} + Err(error) => { + tracing::warn!(%error, %worker_id, "failed to recover a task attempt outcome"); + } + } + } + + match global_task_store.reconcile_interrupted_attempts().await { + Ok(closed) if closed > 0 => { + tracing::info!( + attempts = closed, + "closed task attempts interrupted by an exit" + ); + } + Ok(_) => {} + Err(error) => { + tracing::warn!(%error, "failed to reconcile interrupted task attempts"); + } + } + } + // Resume idle interactive workers that survived the restart. // For each idle worker, pre-create the channel if needed and spawn // the resumed worker into its state so follow-ups route correctly. diff --git a/src/tasks.rs b/src/tasks.rs index 3cedb58a8..4d76ff6e1 100644 --- a/src/tasks.rs +++ b/src/tasks.rs @@ -4,6 +4,7 @@ pub mod comments; pub mod migration; pub mod revisions; pub mod store; +pub mod worker_runs; pub use comments::{ CreateTaskCommentInput, MAX_COMMENT_BODY_BYTES, MAX_COMMENT_PAGE, MIN_COMMENT_BODY_CHARS, @@ -14,6 +15,10 @@ pub use revisions::{ TaskMutationContext, TaskMutationSource, TaskRevision, TaskRevisionDependency, TaskRevisionDiff, TaskRevisionSnapshot, TaskRevisionSummary, }; +pub use worker_runs::{ + MAX_ATTEMPT_PAGE, StartTaskAttempt, TaskAttempt, TaskAttemptOutcome, render_prior_attempts, +}; + pub use store::{ CreateTaskInput, ExecutionDefaults, ExecutionPlan, Patch, Task, TaskDependencyEdge, TaskDependencyKind, TaskListFilter, TaskPriority, TaskStatus, TaskStore, TaskSubtask, diff --git a/src/tasks/revisions.rs b/src/tasks/revisions.rs index 7f8d17f38..dffb32ab9 100644 --- a/src/tasks/revisions.rs +++ b/src/tasks/revisions.rs @@ -76,6 +76,13 @@ impl std::fmt::Display for TaskAuthorKind { } } +impl Default for TaskAuthorKind { + /// Unattributed writes are Spacebot's own. + fn default() -> Self { + Self::System + } +} + /// Which surface a task mutation arrived through. Recorded per revision so /// history reads as a sequence of decisions with their origin intact. #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, utoipa::ToSchema)] @@ -949,6 +956,197 @@ mod tests { assert_eq!(restored.task.metadata, serde_json::json!({})); } + /// The backfill reconnects a task to the `task-` worktree that was + /// provisioned for it before the binding was recorded. + #[tokio::test] + async fn worktree_backfill_binds_by_the_conventional_name() { + let (store, number) = store_with_task().await; + let candidates = vec![ + (format!("task-{number}"), "wt-for-this-task".to_string()), + ("task-9999".to_string(), "wt-for-another-task".to_string()), + ]; + + let bound = store + .backfill_worktree_bindings(&candidates) + .await + .expect("backfill should succeed"); + + assert_eq!(bound, 1); + let task = store + .get_by_number(number) + .await + .expect("load should succeed") + .expect("task should exist"); + assert_eq!(task.worktree_id.as_deref(), Some("wt-for-this-task")); + + // The revision records that the binding was inferred rather than + // observed, which is what keeps the two kinds distinguishable. + let history = store + .list_revisions(number, 10) + .await + .expect("history should load"); + let latest = history.first().expect("a revision was appended"); + assert!( + latest + .edit_summary + .as_deref() + .is_some_and(|summary| summary.contains("inferred")), + "the backfill revision should say the binding was inferred" + ); + } + + /// Running it twice must not append a second revision, and a task that + /// already has a binding is never touched. + #[tokio::test] + async fn worktree_backfill_is_idempotent_and_skips_bound_tasks() { + let (store, number) = store_with_task().await; + let candidates = vec![(format!("task-{number}"), "wt-1".to_string())]; + + assert_eq!( + store + .backfill_worktree_bindings(&candidates) + .await + .expect("first pass"), + 1 + ); + let after_first = store + .list_revisions(number, 10) + .await + .expect("history should load") + .len(); + + assert_eq!( + store + .backfill_worktree_bindings(&candidates) + .await + .expect("second pass"), + 0, + "a bound task must not be revisited" + ); + assert_eq!( + store + .list_revisions(number, 10) + .await + .expect("history should load") + .len(), + after_first, + "the second pass must not append a revision" + ); + + // A rename must not steal the binding the task already holds. + let renamed = vec![(format!("task-{number}"), "wt-2".to_string())]; + assert_eq!( + store + .backfill_worktree_bindings(&renamed) + .await + .expect("third pass"), + 0 + ); + let task = store + .get_by_number(number) + .await + .expect("load should succeed") + .expect("task should exist"); + assert_eq!(task.worktree_id.as_deref(), Some("wt-1")); + } + + /// A task with no matching worktree is left alone rather than guessed at. + #[tokio::test] + async fn worktree_backfill_ignores_tasks_with_no_matching_worktree() { + let (store, number) = store_with_task().await; + + let bound = store + .backfill_worktree_bindings(&[("task-4242".to_string(), "wt-x".to_string())]) + .await + .expect("backfill should succeed"); + + assert_eq!(bound, 0); + let task = store + .get_by_number(number) + .await + .expect("load should succeed") + .expect("task should exist"); + assert_eq!(task.worktree_id, None); + } + + /// `goal_id` is in the snapshot and `changes` diffs it, so a restore that + /// left the current goal in place would report success while producing a + /// task that does not match the revision it claims to have restored. + #[tokio::test] + async fn restore_reinstates_the_goal_it_recorded() { + let (store, number) = store_with_task().await; + + store + .update_with_status_transition( + number, + UpdateTaskInput { + goal_id: Some(Some("goal-original".to_string())), + context: user_context("Attach the original goal"), + ..Default::default() + }, + ) + .await + .expect("update should succeed"); + + let moved = store + .update_with_status_transition( + number, + UpdateTaskInput { + goal_id: Some(Some("goal-moved".to_string())), + context: user_context("Move to another goal"), + ..Default::default() + }, + ) + .await + .expect("update should succeed") + .expect("task should exist"); + assert_eq!(moved.task.goal_id.as_deref(), Some("goal-moved")); + + let restored = store + .restore_revision(number, 2, user_context("Back to the original goal")) + .await + .expect("restore should succeed"); + + assert_eq!(restored.task.goal_id.as_deref(), Some("goal-original")); + + // The revision the restore appended must match what it restored, or a + // diff against it still reports a goal change. + let diff = store + .diff_revisions(number, 2, None) + .await + .expect("diff should compute"); + assert!( + !diff.changes.iter().any(|change| change.field == "goal_id"), + "the restored revision should agree with revision 2 on goal_id" + ); + } + + /// A restore back to a revision that predates any goal must clear it, + /// matching how the other patch fields behave. + #[tokio::test] + async fn restore_clears_a_goal_the_target_revision_did_not_have() { + let (store, number) = store_with_task().await; + + store + .update_with_status_transition( + number, + UpdateTaskInput { + goal_id: Some(Some("goal-1".to_string())), + context: user_context("Attach a goal"), + ..Default::default() + }, + ) + .await + .expect("update should succeed"); + + let restored = store + .restore_revision(number, 1, user_context("Back to the start")) + .await + .expect("restore should succeed"); + + assert_eq!(restored.task.goal_id, None); + } + #[tokio::test] async fn diff_reports_only_the_fields_that_changed() { let (store, number) = store_with_task().await; diff --git a/src/tasks/store.rs b/src/tasks/store.rs index 696ff35a0..def340087 100644 --- a/src/tasks/store.rs +++ b/src/tasks/store.rs @@ -488,6 +488,7 @@ pub struct UpdateTaskInput { pub repo_id: Patch, pub worktree_mode: Patch, pub worktree_id: Patch, + pub goal_id: Patch, pub required_skills: Option>, /// Attribution and optimistic-concurrency expectations for this mutation. pub context: TaskMutationContext, @@ -846,6 +847,69 @@ impl TaskStore { Ok(Some(task)) } + /// Bind tasks to the worktree already carrying their conventional name. + /// + /// A task provisioned before the binding was recorded has a `task-` + /// worktree on disk and no `worktree_id`, so nothing connects the two and a + /// retry rediscovers the worktree by name every time. `candidates` is + /// `(worktree_name, worktree_id)`; the caller supplies them so this does + /// not reach into the project tables. + /// + /// The binding is inferred from the naming convention rather than observed + /// when the worktree was provisioned, and the revision it writes says so — + /// that is what keeps an inferred binding distinguishable from a real one. + /// Idempotent: a task that already has a binding is never revisited. + pub async fn backfill_worktree_bindings( + &self, + candidates: &[(String, String)], + ) -> Result { + if candidates.is_empty() { + return Ok(0); + } + + let unbound: Vec = sqlx::query_scalar( + "SELECT task_number FROM tasks WHERE worktree_id IS NULL ORDER BY task_number", + ) + .fetch_all(self.pool()) + .await + .context("failed to list tasks without a worktree binding")?; + + let mut bound = 0usize; + for task_number in unbound { + let name = format!("task-{task_number}"); + let Some((_, worktree_id)) = candidates.iter().find(|(known, _)| *known == name) else { + continue; + }; + + let context = TaskMutationContext::new( + crate::tasks::TaskAuthorKind::System, + Some("migration".to_string()), + crate::tasks::TaskMutationSource::Migration, + ) + .with_summary(Some(format!( + "Bound to the existing {name} worktree by name; inferred from the naming \ + convention, not observed when the worktree was provisioned" + ))); + + let updated = self + .update( + task_number, + UpdateTaskInput { + worktree_id: Some(Some(worktree_id.clone())), + context, + ..Default::default() + }, + ) + .await?; + + if updated.is_some() { + bound += 1; + } + } + + Ok(bound) + } + pub async fn update(&self, task_number: i64, input: UpdateTaskInput) -> Result> { Ok(self .update_with_status_transition(task_number, input) @@ -1056,6 +1120,7 @@ impl TaskStore { repo_id: Some(snapshot.repo_id), worktree_mode: Some(snapshot.worktree_mode), worktree_id: Some(snapshot.worktree_id), + goal_id: Some(snapshot.goal_id), required_skills: Some(snapshot.required_skills), context, ..Default::default() @@ -1310,6 +1375,7 @@ impl TaskStore { let next_repo_id = patch(input.repo_id, current.repo_id); let next_worktree_mode = patch(input.worktree_mode, current.worktree_mode); let next_worktree_id = patch(input.worktree_id, current.worktree_id); + let next_goal_id = patch(input.goal_id, current.goal_id); let next_required_skills = input.required_skills.unwrap_or(current.required_skills); let required_skills_json = serde_json::to_string(&next_required_skills) .context("failed to serialize required skills")?; @@ -1318,7 +1384,7 @@ impl TaskStore { "UPDATE tasks SET title = ?, description = ?, status = ?, priority = ?, \ assigned_agent_id = ?, subtasks = ?, metadata = ?, \ worker_type = ?, project_id = ?, repo_id = ?, worktree_mode = ?, \ - worktree_id = ?, required_skills = ?, ", + worktree_id = ?, goal_id = ?, required_skills = ?, ", ); if clear_worker { @@ -1358,6 +1424,7 @@ impl TaskStore { .bind(&next_repo_id) .bind(next_worktree_mode.map(TaskWorktreeMode::as_str)) .bind(&next_worktree_id) + .bind(&next_goal_id) .bind(&required_skills_json); if !clear_worker { @@ -1381,7 +1448,7 @@ impl TaskStore { task_from_row(updated) } - /// Delete a task with its comments and revisions. + /// Delete a task with its comments, revisions and run history. /// /// The child rows are removed explicitly rather than through the foreign /// key, which only cascades when `PRAGMA foreign_keys` is on. @@ -1406,7 +1473,12 @@ impl TaskStore { return Ok(false); }; - for table in ["task_comments", "task_revisions", "task_dependencies"] { + for table in [ + "task_comments", + "task_revisions", + "task_dependencies", + "task_worker_runs", + ] { sqlx::query(&format!("DELETE FROM {table} WHERE task_id = ?")) .bind(&task_id) .execute(&mut *tx) @@ -1757,6 +1829,36 @@ pub(crate) async fn setup_test_store() -> TaskStore { .await .expect("task_revisions should be created"); + sqlx::query( + "CREATE TABLE task_worker_runs ( + id TEXT PRIMARY KEY, + task_id TEXT NOT NULL REFERENCES tasks(id) ON DELETE CASCADE, + worker_id TEXT NOT NULL, + attempt INTEGER NOT NULL, + author_type TEXT NOT NULL DEFAULT 'system', + author_id TEXT, + agent_id TEXT, + channel_id TEXT, + started_at TIMESTAMP NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), + outcome_kind TEXT, + outcome_summary TEXT, + ended_at TIMESTAMP, + UNIQUE (task_id, worker_id), + UNIQUE (task_id, attempt) + )", + ) + .execute(&pool) + .await + .expect("task_worker_runs should be created"); + + sqlx::query( + "CREATE UNIQUE INDEX task_worker_runs_live ON task_worker_runs(task_id) \ + WHERE ended_at IS NULL", + ) + .execute(&pool) + .await + .expect("live attempt index should be created"); + sqlx::query("INSERT INTO task_number_seq (id, next_number) VALUES (1, 1)") .execute(&pool) .await diff --git a/src/tasks/worker_runs.rs b/src/tasks/worker_runs.rs new file mode 100644 index 000000000..042cb939e --- /dev/null +++ b/src/tasks/worker_runs.rs @@ -0,0 +1,1056 @@ +//! Every worker run attempted against a task. +//! +//! `tasks.worker_id` names the run executing right now and is overwritten by +//! the next spawn, so a task retried three times remembers only the last one. +//! That is enough to route a reply and not enough to decide anything: an +//! autonomous loop picking work off the board has to know what has already been +//! tried and how it ended before it spawns again, or it repeats failed work +//! forever. +//! +//! The reference to the worker is a bare id. Tasks live in the instance +//! database and `worker_runs` lives in the per-agent database, so the link +//! crosses a database boundary and no foreign key can enforce it. A run whose +//! worker row has been pruned still records that the attempt happened. + +use crate::error::Result; +use crate::tasks::revisions::TaskAuthorKind; +use crate::tasks::store::TaskStore; + +use anyhow::Context as _; +use serde::{Deserialize, Serialize}; +use sqlx::{Row as _, sqlite::SqliteRow}; + +/// Hard ceiling on rows returned by a single attempt-history call. +pub const MAX_ATTEMPT_PAGE: i64 = 100; + +/// How much of a run's result is kept on the attempt. +/// +/// The full result lives on the worker record; this is the line the board and +/// the prompt context read, and a worker can return a great deal of text. +const MAX_ATTEMPT_SUMMARY_CHARS: usize = 280; + +/// How a worker run ended, from the task's point of view. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, utoipa::ToSchema)] +#[serde(rename_all = "snake_case")] +pub enum TaskAttemptOutcome { + Succeeded, + /// Reached its budget with real work delivered but the task unfinished. + Partial, + /// Stopped waiting on something outside its control. + Blocked, + Failed, + Cancelled, + TimedOut, + /// The process died before the run reached a terminal state. Distinct from + /// a failure: nothing was decided about the work itself. + Interrupted, +} + +impl TaskAttemptOutcome { + pub fn as_str(self) -> &'static str { + match self { + Self::Succeeded => "succeeded", + Self::Partial => "partial", + Self::Blocked => "blocked", + Self::Failed => "failed", + Self::Cancelled => "cancelled", + Self::TimedOut => "timed_out", + Self::Interrupted => "interrupted", + } + } + + pub fn parse(value: &str) -> Option { + match value { + "succeeded" => Some(Self::Succeeded), + "partial" => Some(Self::Partial), + "blocked" => Some(Self::Blocked), + "failed" => Some(Self::Failed), + "cancelled" => Some(Self::Cancelled), + "timed_out" => Some(Self::TimedOut), + "interrupted" => Some(Self::Interrupted), + _ => None, + } + } + + /// Whether this outcome means the work was actually delivered. + pub fn is_success(self) -> bool { + matches!(self, Self::Succeeded) + } +} + +/// The worker's committed terminal kind is what an attempt records. +/// +/// There is no `Interrupted` on the worker side: that outcome describes a run +/// with no terminal record at all, which is the one case this conversion cannot +/// be reached from. +impl From for TaskAttemptOutcome { + fn from(kind: crate::conversation::WorkerOutcomeKind) -> Self { + use crate::conversation::WorkerOutcomeKind as Kind; + match kind { + Kind::Succeeded => Self::Succeeded, + Kind::Partial => Self::Partial, + Kind::Blocked => Self::Blocked, + Kind::Failed => Self::Failed, + Kind::Cancelled => Self::Cancelled, + Kind::TimedOut => Self::TimedOut, + } + } +} + +impl std::fmt::Display for TaskAttemptOutcome { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "{}", self.as_str()) + } +} + +/// One worker run recorded against a task. +#[derive(Debug, Clone, Serialize, Deserialize, utoipa::ToSchema)] +pub struct TaskAttempt { + pub id: String, + pub task_id: String, + pub worker_id: String, + /// 1 for the first run on this task. + pub attempt: i64, + pub author_type: TaskAuthorKind, + pub author_id: Option, + pub agent_id: Option, + pub channel_id: Option, + pub started_at: String, + /// `None` while the run is still live. + pub outcome: Option, + pub outcome_summary: Option, + pub ended_at: Option, +} + +impl TaskAttempt { + /// Whether this run has not reached a terminal state. + pub fn is_live(&self) -> bool { + self.ended_at.is_none() + } +} + +/// What to record when a run starts. +#[derive(Debug, Clone, Default)] +pub struct StartTaskAttempt { + pub worker_id: String, + pub author_type: TaskAuthorKind, + pub author_id: Option, + pub agent_id: Option, + pub channel_id: Option, +} + +const ATTEMPT_COLUMNS: &str = "SELECT id, task_id, worker_id, attempt, author_type, author_id, \ + agent_id, channel_id, started_at, outcome_kind, outcome_summary, ended_at \ + FROM task_worker_runs"; + +fn attempt_from_row(row: &SqliteRow) -> Result { + let author_type: String = row.try_get("author_type").unwrap_or_default(); + let outcome_kind: Option = row.try_get("outcome_kind").ok().flatten(); + + Ok(TaskAttempt { + id: row.try_get("id").context("attempt row missing id")?, + task_id: row + .try_get("task_id") + .context("attempt row missing task_id")?, + worker_id: row + .try_get("worker_id") + .context("attempt row missing worker_id")?, + attempt: row + .try_get("attempt") + .context("attempt row missing attempt")?, + author_type: TaskAuthorKind::parse(&author_type).unwrap_or(TaskAuthorKind::System), + author_id: row.try_get("author_id").ok().flatten(), + agent_id: row.try_get("agent_id").ok().flatten(), + channel_id: row.try_get("channel_id").ok().flatten(), + started_at: row + .try_get("started_at") + .context("attempt row missing started_at")?, + outcome: outcome_kind.as_deref().and_then(TaskAttemptOutcome::parse), + outcome_summary: row.try_get("outcome_summary").ok().flatten(), + ended_at: row.try_get("ended_at").ok().flatten(), + }) +} + +impl TaskStore { + /// Record that a worker run has started against a task. + /// + /// The attempt number is allocated inside the transaction, so two spawns + /// racing on the same task cannot both claim the same ordinal. Re-recording + /// the same worker returns the existing row rather than a second attempt, + /// which makes a retried bind idempotent. + pub async fn start_task_attempt( + &self, + task_number: i64, + input: StartTaskAttempt, + ) -> Result> { + let mut tx = self + .pool() + .begin_with("BEGIN IMMEDIATE") + .await + .context("failed to open task attempt transaction")?; + + let task_id: Option = + sqlx::query_scalar("SELECT id FROM tasks WHERE task_number = ?") + .bind(task_number) + .fetch_optional(&mut *tx) + .await + .context("failed to resolve task for attempt")?; + + let Some(task_id) = task_id else { + tx.rollback() + .await + .context("failed to roll back task attempt transaction")?; + return Ok(None); + }; + + let existing = sqlx::query(&format!( + "{ATTEMPT_COLUMNS} WHERE task_id = ? AND worker_id = ?" + )) + .bind(&task_id) + .bind(&input.worker_id) + .fetch_optional(&mut *tx) + .await + .context("failed to check for an existing attempt")?; + + if let Some(row) = existing { + let attempt = attempt_from_row(&row)?; + tx.commit() + .await + .context("failed to commit task attempt transaction")?; + return Ok(Some(attempt)); + } + + let next_attempt: i64 = sqlx::query_scalar( + "SELECT COALESCE(MAX(attempt), 0) + 1 FROM task_worker_runs WHERE task_id = ?", + ) + .bind(&task_id) + .fetch_one(&mut *tx) + .await + .context("failed to allocate an attempt number")?; + + let id = uuid::Uuid::new_v4().to_string(); + sqlx::query( + "INSERT INTO task_worker_runs \ + (id, task_id, worker_id, attempt, author_type, author_id, agent_id, channel_id) \ + VALUES (?, ?, ?, ?, ?, ?, ?, ?)", + ) + .bind(&id) + .bind(&task_id) + .bind(&input.worker_id) + .bind(next_attempt) + .bind(input.author_type.as_str()) + .bind(&input.author_id) + .bind(&input.agent_id) + .bind(&input.channel_id) + .execute(&mut *tx) + .await + .map_err(|error| { + // The live-attempt index is what settles two spawns racing on the + // same task, so a unique violation here is a lost race rather than + // a storage fault, and the caller has to be able to tell them apart. + if matches!(&error, sqlx::Error::Database(db) if db.is_unique_violation()) { + anyhow::anyhow!( + "task #{task_number} already has a live attempt — another spawn claimed it first" + ) + } else { + anyhow::Error::new(error).context("failed to record task attempt") + } + })?; + + let row = sqlx::query(&format!("{ATTEMPT_COLUMNS} WHERE id = ?")) + .bind(&id) + .fetch_one(&mut *tx) + .await + .context("failed to reload the recorded attempt")?; + let attempt = attempt_from_row(&row)?; + + tx.commit() + .await + .context("failed to commit task attempt transaction")?; + Ok(Some(attempt)) + } + + /// Record how a run ended. + /// + /// Terminal state is written once: a second call for the same worker leaves + /// the first outcome in place, so a duplicated completion cannot rewrite + /// history. Returns whether this call was the one that closed the run. + pub async fn finish_task_attempt( + &self, + worker_id: &str, + outcome: TaskAttemptOutcome, + summary: Option<&str>, + ) -> Result { + let summary: Option = summary + .map(|text| text.chars().take(MAX_ATTEMPT_SUMMARY_CHARS).collect()) + .filter(|text: &String| !text.is_empty()); + + let affected = sqlx::query( + "UPDATE task_worker_runs \ + SET outcome_kind = ?, outcome_summary = ?, \ + ended_at = strftime('%Y-%m-%dT%H:%M:%SZ', 'now') \ + WHERE worker_id = ? AND ended_at IS NULL", + ) + .bind(outcome.as_str()) + .bind(summary) + .bind(worker_id) + .execute(self.pool()) + .await + .context("failed to record task attempt outcome")? + .rows_affected(); + + Ok(affected > 0) + } + + /// The runs attempted against a task, newest first. + pub async fn list_task_attempts( + &self, + task_number: i64, + limit: i64, + ) -> Result> { + let limit = limit.clamp(1, MAX_ATTEMPT_PAGE); + let rows = sqlx::query(&format!( + "{ATTEMPT_COLUMNS} WHERE task_id = (SELECT id FROM tasks WHERE task_number = ?) \ + ORDER BY attempt DESC LIMIT ?" + )) + .bind(task_number) + .bind(limit) + .fetch_all(self.pool()) + .await + .context("failed to list task attempts")?; + + rows.iter().map(attempt_from_row).collect() + } + + /// Every attempt still open, across all tasks. + /// + /// Read at startup to recover runs whose worker reached a terminal state + /// that the attempt never learned about, before the rest are swept. + pub async fn live_attempts(&self) -> Result> { + let rows = sqlx::query(&format!( + "{ATTEMPT_COLUMNS} WHERE ended_at IS NULL ORDER BY started_at" + )) + .fetch_all(self.pool()) + .await + .context("failed to list live task attempts")?; + + rows.iter().map(attempt_from_row).collect() + } + + /// Close attempts left live by a process that died. + /// + /// Workers run in-process, so every attempt still open at startup belongs + /// to a run that no longer exists. Without this the spawn guard would see a + /// live attempt forever and the task could never be worked again — a crash + /// mid-run would permanently take that task off the board. + /// + /// Recorded as interrupted rather than failed: the process died, which says + /// nothing about whether the work was going to succeed. Runs that did reach + /// a terminal state are closed with it beforehand, from `live_attempts`, so + /// this only reaches the ones nothing decided. + pub async fn reconcile_interrupted_attempts(&self) -> Result { + let affected = sqlx::query( + "UPDATE task_worker_runs \ + SET outcome_kind = ?, \ + outcome_summary = COALESCE(outcome_summary, ?), \ + ended_at = strftime('%Y-%m-%dT%H:%M:%SZ', 'now') \ + WHERE ended_at IS NULL", + ) + .bind(TaskAttemptOutcome::Interrupted.as_str()) + .bind("The process running this attempt exited before it finished.") + .execute(self.pool()) + .await + .context("failed to reconcile interrupted task attempts")? + .rows_affected(); + + Ok(affected as usize) + } + + /// Prior-attempt lines for a set of tasks, keyed by task number. + /// + /// One query for the whole board. Rendering prompt context must not issue a + /// query per task, and a task that has never been attempted is simply + /// absent from the map rather than carrying an empty entry. + pub async fn prior_attempt_summaries( + &self, + task_numbers: &[i64], + ) -> Result> { + if task_numbers.is_empty() { + return Ok(std::collections::HashMap::new()); + } + + let placeholders = std::iter::repeat_n("?", task_numbers.len()) + .collect::>() + .join(", "); + let sql = format!( + "SELECT t.task_number AS task_number, r.id, r.task_id, r.worker_id, r.attempt, \ + r.author_type, r.author_id, r.agent_id, r.channel_id, r.started_at, \ + r.outcome_kind, r.outcome_summary, r.ended_at \ + FROM task_worker_runs r JOIN tasks t ON t.id = r.task_id \ + WHERE t.task_number IN ({placeholders}) ORDER BY r.attempt DESC" + ); + + let mut query = sqlx::query(&sql); + for number in task_numbers { + query = query.bind(number); + } + let rows = query + .fetch_all(self.pool()) + .await + .context("failed to load attempt history for the board")?; + + let mut grouped: std::collections::HashMap> = + std::collections::HashMap::new(); + for row in &rows { + let number: i64 = row + .try_get("task_number") + .context("attempt row missing task_number")?; + grouped + .entry(number) + .or_default() + .push(attempt_from_row(row)?); + } + + Ok(grouped + .into_iter() + .filter_map(|(number, attempts)| { + render_prior_attempts(&attempts).map(|line| (number, line)) + }) + .collect()) + } + + /// The task a worker was spawned for, if it was spawned for one. + pub async fn task_number_for_worker(&self, worker_id: &str) -> Result> { + let number: Option = sqlx::query_scalar( + "SELECT t.task_number FROM task_worker_runs r \ + JOIN tasks t ON t.id = r.task_id \ + WHERE r.worker_id = ?", + ) + .bind(worker_id) + .fetch_optional(self.pool()) + .await + .context("failed to resolve the task for a worker")?; + + Ok(number) + } + + /// The run currently executing against a task, if any. + /// + /// This is what makes a spawn guard task-scoped rather than channel-scoped: + /// it sees a live run no matter which channel started it. + pub async fn live_task_attempt(&self, task_number: i64) -> Result> { + let row = sqlx::query(&format!( + "{ATTEMPT_COLUMNS} WHERE task_id = (SELECT id FROM tasks WHERE task_number = ?) \ + AND ended_at IS NULL ORDER BY attempt DESC LIMIT 1" + )) + .bind(task_number) + .fetch_optional(self.pool()) + .await + .context("failed to look for a live task attempt")?; + + row.as_ref().map(attempt_from_row).transpose() + } +} + +/// One line summarising what has already been tried on a task. +/// +/// Rendered into prompt context so a spawn decision is made knowing the +/// history. Bounded on purpose: a heavily retried task must not crowd out the +/// rest of the board. +pub fn render_prior_attempts(attempts: &[TaskAttempt]) -> Option { + let finished: Vec<&TaskAttempt> = attempts.iter().filter(|a| !a.is_live()).collect(); + let live = attempts.iter().find(|a| a.is_live()); + + if finished.is_empty() && live.is_none() { + return None; + } + + let mut parts = Vec::new(); + + if !finished.is_empty() { + let outcomes: Vec = finished + .iter() + .take(3) + .map(|attempt| { + let outcome = attempt + .outcome + .map(|o| o.to_string()) + .unwrap_or_else(|| "ended without an outcome".to_string()); + format!("#{} {}", attempt.attempt, outcome) + }) + .collect(); + + let plural = if finished.len() == 1 { + "attempt" + } else { + "attempts" + }; + parts.push(format!( + "{} prior {plural} ({})", + finished.len(), + outcomes.join(", ") + )); + } + + if let Some(live) = live { + parts.push(format!("attempt #{} is running now", live.attempt)); + } + + Some(parts.join("; ")) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::conversation::WorkerOutcomeKind; + use crate::tasks::store::{CreateTaskInput, setup_test_store}; + + fn task_input(title: &str) -> CreateTaskInput { + CreateTaskInput { + owner_agent_id: "main".to_string(), + title: title.to_string(), + ..Default::default() + } + } + + async fn store_with_task() -> (TaskStore, i64) { + let store = setup_test_store().await; + let task = store + .create(task_input("linkage")) + .await + .expect("task should be created"); + (store, task.task_number) + } + + fn start(worker_id: &str) -> StartTaskAttempt { + StartTaskAttempt { + worker_id: worker_id.to_string(), + author_type: TaskAuthorKind::Agent, + author_id: Some("main".to_string()), + agent_id: Some("main".to_string()), + channel_id: Some("telegram:1".to_string()), + } + } + + /// The record this whole module exists for: a task run three times keeps + /// all three, where `tasks.worker_id` would remember only the last. + #[tokio::test] + async fn every_attempt_is_kept_with_its_outcome() { + let (store, number) = store_with_task().await; + + for (worker, outcome) in [ + ("worker-a", TaskAttemptOutcome::Failed), + ("worker-b", TaskAttemptOutcome::TimedOut), + ("worker-c", TaskAttemptOutcome::Succeeded), + ] { + store + .start_task_attempt(number, start(worker)) + .await + .expect("start should succeed") + .expect("task exists"); + store + .finish_task_attempt(worker, outcome, Some("summary")) + .await + .expect("finish should succeed"); + } + + let attempts = store + .list_task_attempts(number, 10) + .await + .expect("history should load"); + + assert_eq!(attempts.len(), 3); + // Newest first. + assert_eq!(attempts[0].attempt, 3); + assert_eq!(attempts[0].worker_id, "worker-c"); + assert_eq!(attempts[0].outcome, Some(TaskAttemptOutcome::Succeeded)); + assert_eq!(attempts[2].attempt, 1); + assert_eq!(attempts[2].outcome, Some(TaskAttemptOutcome::Failed)); + assert!(attempts.iter().all(|a| !a.is_live())); + } + + /// The reverse lookup the API had no way to answer. + #[tokio::test] + async fn a_worker_resolves_back_to_its_task() { + let (store, number) = store_with_task().await; + store + .start_task_attempt(number, start("worker-1")) + .await + .expect("start should succeed") + .expect("task exists"); + + assert_eq!( + store + .task_number_for_worker("worker-1") + .await + .expect("lookup should succeed"), + Some(number) + ); + assert_eq!( + store + .task_number_for_worker("worker-unknown") + .await + .expect("lookup should succeed"), + None + ); + } + + /// A live run is visible regardless of which channel started it, which is + /// what lets a spawn guard be task-scoped rather than channel-scoped. + #[tokio::test] + async fn a_live_attempt_is_visible_until_it_ends() { + let (store, number) = store_with_task().await; + assert!( + store + .live_task_attempt(number) + .await + .expect("lookup should succeed") + .is_none() + ); + + let mut other_channel = start("worker-1"); + other_channel.channel_id = Some("discord:99".to_string()); + store + .start_task_attempt(number, other_channel) + .await + .expect("start should succeed") + .expect("task exists"); + + let live = store + .live_task_attempt(number) + .await + .expect("lookup should succeed") + .expect("a run is live"); + assert_eq!(live.worker_id, "worker-1"); + assert_eq!(live.channel_id.as_deref(), Some("discord:99")); + + store + .finish_task_attempt("worker-1", TaskAttemptOutcome::Succeeded, None) + .await + .expect("finish should succeed"); + assert!( + store + .live_task_attempt(number) + .await + .expect("lookup should succeed") + .is_none() + ); + } + + /// The spawn guard reads before it writes, so two channels can both find a + /// task free. Storage is what settles it: the second worker is refused, and + /// the task opens again once the first run ends. + #[tokio::test] + async fn only_one_run_can_be_live_on_a_task() { + let (store, number) = store_with_task().await; + + store + .start_task_attempt(number, start("worker-1")) + .await + .expect("start should succeed") + .expect("task exists"); + + let raced = store + .start_task_attempt(number, start("worker-2")) + .await + .expect_err("a second live run should be refused"); + assert!( + raced.to_string().contains("already has a live attempt"), + "unexpected error: {raced}" + ); + assert_eq!( + store + .list_task_attempts(number, 10) + .await + .expect("history should load") + .len(), + 1 + ); + + store + .finish_task_attempt("worker-1", TaskAttemptOutcome::Failed, None) + .await + .expect("finish should succeed"); + + let retry = store + .start_task_attempt(number, start("worker-2")) + .await + .expect("start should succeed") + .expect("task exists"); + assert_eq!(retry.attempt, 2); + } + + /// Attempts carry worker ids and outcome text, so they must not outlive the + /// task. Foreign-key enforcement is not guaranteed to be on, which is why + /// the delete is explicit. + #[tokio::test] + async fn deleting_a_task_deletes_its_attempts() { + let (store, number) = store_with_task().await; + store + .start_task_attempt(number, start("worker-1")) + .await + .expect("start should succeed") + .expect("task exists"); + store + .finish_task_attempt("worker-1", TaskAttemptOutcome::Succeeded, Some("done")) + .await + .expect("finish should succeed"); + + // The condition the explicit delete exists for: with enforcement off, + // the cascade on the foreign key does nothing. + sqlx::query("PRAGMA foreign_keys = OFF") + .execute(store.pool()) + .await + .expect("pragma should apply"); + + assert!(store.delete(number).await.expect("delete should succeed")); + + assert_eq!( + store + .task_number_for_worker("worker-1") + .await + .expect("lookup should succeed"), + None + ); + let remaining: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM task_worker_runs") + .fetch_one(store.pool()) + .await + .expect("count should succeed"); + assert_eq!(remaining, 0); + } + + /// Re-binding the same worker must not invent a second attempt, so a + /// retried bind after a transient failure stays idempotent. + #[tokio::test] + async fn re_recording_the_same_worker_reuses_its_attempt() { + let (store, number) = store_with_task().await; + + let first = store + .start_task_attempt(number, start("worker-1")) + .await + .expect("start should succeed") + .expect("task exists"); + let again = store + .start_task_attempt(number, start("worker-1")) + .await + .expect("start should succeed") + .expect("task exists"); + + assert_eq!(first.id, again.id); + assert_eq!(again.attempt, 1); + assert_eq!( + store + .list_task_attempts(number, 10) + .await + .expect("history should load") + .len(), + 1 + ); + } + + /// A duplicated completion must not rewrite how the run ended. + #[tokio::test] + async fn terminal_state_is_written_once() { + let (store, number) = store_with_task().await; + store + .start_task_attempt(number, start("worker-1")) + .await + .expect("start should succeed") + .expect("task exists"); + + assert!( + store + .finish_task_attempt("worker-1", TaskAttemptOutcome::Succeeded, Some("done")) + .await + .expect("finish should succeed") + ); + assert!( + !store + .finish_task_attempt("worker-1", TaskAttemptOutcome::Failed, Some("nope")) + .await + .expect("finish should succeed"), + "a second completion must not close the run again" + ); + + let attempts = store + .list_task_attempts(number, 10) + .await + .expect("history should load"); + assert_eq!(attempts[0].outcome, Some(TaskAttemptOutcome::Succeeded)); + assert_eq!(attempts[0].outcome_summary.as_deref(), Some("done")); + } + + /// The two writes that close a run land in different databases, so a worker + /// can commit its outcome and the attempt still be open. Startup recovers + /// what the worker committed before the sweep runs, or a run that succeeded + /// would be recorded as interrupted and an autonomous loop would retry work + /// that was already delivered. + #[tokio::test] + async fn a_committed_outcome_is_recovered_before_the_sweep() { + let (store, number) = store_with_task().await; + let other = store + .create(task_input("swept")) + .await + .expect("task should be created"); + store + .start_task_attempt(number, start("worker-committed")) + .await + .expect("start should succeed") + .expect("task exists"); + store + .start_task_attempt(other.task_number, start("worker-vanished")) + .await + .expect("start should succeed") + .expect("task exists"); + + let live = store + .live_attempts() + .await + .expect("live attempts should load"); + assert_eq!(live.len(), 2); + assert!(live.iter().all(|attempt| attempt.is_live())); + + // What the startup pass does for a run whose worker record has an + // outcome; the other worker left nothing behind. + store + .finish_task_attempt( + "worker-committed", + WorkerOutcomeKind::Succeeded.into(), + Some("shipped it"), + ) + .await + .expect("recovery should succeed"); + + let swept = store + .reconcile_interrupted_attempts() + .await + .expect("reconcile should succeed"); + assert_eq!(swept, 1, "only the undecided run is swept"); + + let recovered = store + .list_task_attempts(number, 10) + .await + .expect("history should load"); + assert_eq!(recovered[0].outcome, Some(TaskAttemptOutcome::Succeeded)); + assert_eq!(recovered[0].outcome_summary.as_deref(), Some("shipped it")); + + let interrupted = store + .list_task_attempts(other.task_number, 10) + .await + .expect("history should load"); + assert_eq!( + interrupted[0].outcome, + Some(TaskAttemptOutcome::Interrupted) + ); + } + + /// A worker can return a great deal of text and the board reads this line. + #[tokio::test] + async fn an_attempt_summary_is_bounded() { + let (store, number) = store_with_task().await; + store + .start_task_attempt(number, start("worker-1")) + .await + .expect("start should succeed") + .expect("task exists"); + store + .finish_task_attempt( + "worker-1", + TaskAttemptOutcome::Succeeded, + Some(&"x".repeat(5_000)), + ) + .await + .expect("finish should succeed"); + + let attempts = store + .list_task_attempts(number, 10) + .await + .expect("history should load"); + assert_eq!( + attempts[0] + .outcome_summary + .as_deref() + .map(|summary| summary.chars().count()), + Some(MAX_ATTEMPT_SUMMARY_CHARS) + ); + } + + /// A crash mid-run must not take the task off the board for good. + #[tokio::test] + async fn a_restart_closes_a_live_attempt_and_unblocks_the_task() { + let (store, number) = store_with_task().await; + store + .start_task_attempt(number, start("worker-1")) + .await + .expect("start should succeed") + .expect("task exists"); + assert!( + store + .live_task_attempt(number) + .await + .expect("lookup should succeed") + .is_some() + ); + + let closed = store + .reconcile_interrupted_attempts() + .await + .expect("reconcile should succeed"); + + assert_eq!(closed, 1); + assert!( + store + .live_task_attempt(number) + .await + .expect("lookup should succeed") + .is_none(), + "the task must be spawnable again after a restart" + ); + + let attempts = store + .list_task_attempts(number, 10) + .await + .expect("history should load"); + assert_eq!(attempts[0].outcome, Some(TaskAttemptOutcome::Interrupted)); + + // A run that already ended keeps the outcome it recorded. + assert_eq!( + store + .reconcile_interrupted_attempts() + .await + .expect("second reconcile"), + 0 + ); + } + + #[tokio::test] + async fn an_attempt_on_a_missing_task_is_not_recorded() { + let store = setup_test_store().await; + assert!( + store + .start_task_attempt(4242, start("worker-1")) + .await + .expect("start should succeed") + .is_none() + ); + } + + /// The board renders in one query, and a task never attempted is absent + /// rather than carrying an empty line. + #[tokio::test] + async fn board_summaries_cover_only_attempted_tasks() { + let store = setup_test_store().await; + let attempted = store + .create(task_input("attempted")) + .await + .expect("task should be created") + .task_number; + let untouched = store + .create(task_input("untouched")) + .await + .expect("task should be created") + .task_number; + + store + .start_task_attempt(attempted, start("worker-1")) + .await + .expect("start should succeed") + .expect("task exists"); + store + .finish_task_attempt("worker-1", TaskAttemptOutcome::Failed, None) + .await + .expect("finish should succeed"); + store + .start_task_attempt(attempted, start("worker-2")) + .await + .expect("start should succeed") + .expect("task exists"); + + let summaries = store + .prior_attempt_summaries(&[attempted, untouched]) + .await + .expect("summaries should load"); + + assert_eq!(summaries.len(), 1); + let line = summaries + .get(&attempted) + .expect("attempted task summarised"); + assert!(line.contains("1 prior attempt"), "{line}"); + assert!(line.contains("#1 failed"), "{line}"); + assert!(line.contains("attempt #2 is running now"), "{line}"); + assert!(!summaries.contains_key(&untouched)); + } + + #[tokio::test] + async fn board_summaries_are_empty_without_tasks() { + let store = setup_test_store().await; + assert!( + store + .prior_attempt_summaries(&[]) + .await + .expect("summaries should load") + .is_empty() + ); + } + + #[test] + fn prior_attempts_render_nothing_for_a_fresh_task() { + assert_eq!(render_prior_attempts(&[]), None); + } + + #[test] + fn prior_attempts_name_the_outcomes_and_the_live_run() { + let attempt = |n: i64, outcome: Option, ended: bool| TaskAttempt { + id: format!("id-{n}"), + task_id: "task-1".to_string(), + worker_id: format!("worker-{n}"), + attempt: n, + author_type: TaskAuthorKind::Agent, + author_id: None, + agent_id: None, + channel_id: None, + started_at: "2026-08-14T00:00:00Z".to_string(), + outcome, + outcome_summary: None, + ended_at: ended.then(|| "2026-08-14T01:00:00Z".to_string()), + }; + + let rendered = render_prior_attempts(&[ + attempt(3, None, false), + attempt(2, Some(TaskAttemptOutcome::TimedOut), true), + attempt(1, Some(TaskAttemptOutcome::Failed), true), + ]) + .expect("a task with history renders"); + + assert!(rendered.contains("2 prior attempts")); + assert!(rendered.contains("#2 timed_out")); + assert!(rendered.contains("#1 failed")); + assert!(rendered.contains("attempt #3 is running now")); + } + + /// A heavily retried task must not crowd the board out of the prompt. + #[test] + fn prior_attempts_are_bounded() { + let attempts: Vec = (1..=20) + .rev() + .map(|n| TaskAttempt { + id: format!("id-{n}"), + task_id: "task-1".to_string(), + worker_id: format!("worker-{n}"), + attempt: n, + author_type: TaskAuthorKind::Agent, + author_id: None, + agent_id: None, + channel_id: None, + started_at: "2026-08-14T00:00:00Z".to_string(), + outcome: Some(TaskAttemptOutcome::Failed), + outcome_summary: None, + ended_at: Some("2026-08-14T01:00:00Z".to_string()), + }) + .collect(); + + let rendered = render_prior_attempts(&attempts).expect("renders"); + assert!(rendered.contains("20 prior attempts")); + assert_eq!(rendered.matches('#').count(), 3, "only three are named"); + } +} diff --git a/src/telemetry/registry.rs b/src/telemetry/registry.rs index 5630d3f00..a652d880f 100644 --- a/src/telemetry/registry.rs +++ b/src/telemetry/registry.rs @@ -170,6 +170,10 @@ pub struct Metrics { /// Labels: agent_id, process_type. pub context_overflow_total: IntCounterVec, + /// Tool-history repairs applied to a request. + /// Labels: agent_id, outcome (repaired/retry_success/terminal_failure). + pub tool_history_recovery_total: IntCounterVec, + // -- Cost -- /// Worker cost tracking in USD. /// Labels: agent_id, worker_type. @@ -502,6 +506,15 @@ impl Metrics { ) .expect("hardcoded metric descriptor"); + let tool_history_recovery_total = IntCounterVec::new( + Opts::new( + "spacebot_tool_history_recovery_total", + "Tool-history repairs applied to a request", + ), + &["agent_id", "outcome"], + ) + .expect("hardcoded metric descriptor"); + // Cost (1) let worker_cost_dollars = CounterVec::new( Opts::new( @@ -652,6 +665,9 @@ impl Metrics { registry .register(Box::new(context_overflow_total.clone())) .expect("hardcoded metric"); + registry + .register(Box::new(tool_history_recovery_total.clone())) + .expect("hardcoded metric"); // New: Cost registry @@ -709,6 +725,7 @@ impl Metrics { http_request_duration_seconds, branches_spawned_total, context_overflow_total, + tool_history_recovery_total, worker_cost_dollars, cron_executions_total, cron_delivery_total, diff --git a/src/tools/spawn_worker.rs b/src/tools/spawn_worker.rs index 9b38d926d..6e2e8504d 100644 --- a/src/tools/spawn_worker.rs +++ b/src/tools/spawn_worker.rs @@ -100,6 +100,27 @@ impl SpawnWorkerTool { } } + // Refuse a second run on a task something is already working. The + // delegation check elsewhere is per-channel, so without this two + // channels can spawn on the same task without either noticing. + // An unreadable history cannot establish that the task is free, so a + // lookup failure blocks the spawn rather than falling through it. + match deps.task_store.live_task_attempt(number).await { + Ok(Some(live)) => { + return Err(SpawnWorkerError(format!( + "task #{number} is already being worked by worker {} (attempt #{}, started {}). \ + Wait for it, or cancel it before spawning again.", + live.worker_id, live.attempt, live.started_at + ))); + } + Ok(None) => {} + Err(error) => { + return Err(SpawnWorkerError(format!( + "failed to check whether task #{number} is already being worked: {error}" + ))); + } + } + let project = match &task.project_id { Some(project_id) => Some( deps.project_store @@ -174,13 +195,30 @@ impl SpawnWorkerTool { let worktree_name = format!("task-{number}"); // Reuse the worktree from an earlier spawn attempt instead of - // failing on the existing path. - let existing = deps - .project_store - .list_worktrees(&project.id) - .await - .ok() - .and_then(|worktrees| worktrees.into_iter().find(|w| w.name == worktree_name)); + // failing on the existing path. The binding the task carries is + // authoritative; the `task-` name is the compatibility + // key for tasks provisioned before that binding was recorded. + let bound = match plan.worktree_id.as_deref() { + Some(worktree_id) => deps + .project_store + .get_worktree(worktree_id) + .await + .ok() + .flatten(), + None => None, + }; + + let existing = match bound { + Some(worktree) => Some(worktree), + None => deps + .project_store + .list_worktrees(&project.id) + .await + .ok() + .and_then(|worktrees| { + worktrees.into_iter().find(|w| w.name == worktree_name) + }), + }; match existing { Some(worktree) => { @@ -656,6 +694,10 @@ impl SpawnWorkerTool { crate::tasks::UpdateTaskInput { worker_id: Some(worker_id.to_string()), status: status_change, + // Record the worktree this run resolved to, so a retry + // reuses it instead of rediscovering it by name and a + // task's working directory is visible on the board. + worktree_id: plan.worktree_id.clone().map(Some), ..Default::default() }, ) @@ -668,6 +710,53 @@ impl SpawnWorkerTool { "failed to bind spawned worker to task" ); } + + // The pointer above names only the run executing now. This is the + // history: what has been tried on this task and how it ended. + // + // Unlike the binding above this one is not fire-and-forget. The + // live-attempt index rejects a second open run on the same task, so + // a failure here means another spawn claimed the task between the + // guard and this insert. An unrecorded worker is invisible to the + // guard and to the board, so it is stopped instead of left running. + if let Err(error) = self + .state + .deps + .task_store + .start_task_attempt( + plan.task_number, + crate::tasks::StartTaskAttempt { + worker_id: worker_id.to_string(), + author_type: crate::tasks::TaskAuthorKind::Agent, + author_id: Some(self.state.deps.agent_id.to_string()), + agent_id: Some(self.state.deps.agent_id.to_string()), + channel_id: Some(self.state.channel_id.to_string()), + }, + ) + .await + { + tracing::warn!( + %error, + task_number = plan.task_number, + %worker_id, + "failed to record the task attempt" + ); + if let Err(cancel_error) = self + .state + .cancel_worker_with_reason(worker_id, "task attempt could not be recorded") + .await + { + tracing::warn!( + %cancel_error, + %worker_id, + "failed to cancel a worker with no recorded attempt" + ); + } + return Err(SpawnWorkerError(format!( + "task #{} could not record this attempt, so worker {worker_id} was cancelled: {error}", + plan.task_number + ))); + } } // Link the worker to project/worktree if specified (fire-and-forget update). @@ -1025,6 +1114,7 @@ impl Tool for DetachedSpawnWorkerTool { None, None, secrets_store, + Some(self.deps.task_store.clone()), "builtin", worker.run().instrument(worker_span), ); diff --git a/src/tools/task_update.rs b/src/tools/task_update.rs index 3276759b9..1c80c94ef 100644 --- a/src/tools/task_update.rs +++ b/src/tools/task_update.rs @@ -297,6 +297,7 @@ impl Tool for TaskUpdateTool { repo_id: args.repo_id.map(Some), worktree_mode: worktree_mode.map(Some), worktree_id: args.worktree_id.map(Some), + goal_id: None, required_skills: args.required_skills, context: crate::tasks::TaskMutationContext::new(author_type, Some(author_id), source) .with_summary(args.edit_summary)