diff --git a/apps/x/apps/renderer/src/components/bg-tasks-view.tsx b/apps/x/apps/renderer/src/components/bg-tasks-view.tsx index 45e23169..c0c3610f 100644 --- a/apps/x/apps/renderer/src/components/bg-tasks-view.tsx +++ b/apps/x/apps/renderer/src/components/bg-tasks-view.tsx @@ -15,7 +15,8 @@ import { useBackgroundTaskAgentStatus } from '@/hooks/use-bg-task-agent-status' import { formatRelativeTime } from '@/lib/relative-time' import { toast } from '@/lib/toast' import type { ConversationItem } from '@/lib/chat-conversation' -import { fetchAgentRunTranscript, type AgentRunTranscript } from '@/lib/agent-transcript' +import { fetchAgentRunTranscript } from '@/lib/agent-transcript' +import { useAgentRunTranscript } from '@/hooks/use-agent-run-transcript' import { CompactConversation } from '@/components/compact-conversation' import { RichMarkdownViewer } from '@/components/rich-markdown-viewer' import { HtmlFileViewer } from '@/components/html-file-viewer' @@ -1116,29 +1117,9 @@ function RunTranscriptView({ isInFlight: boolean onBack: () => void }) { - const [transcript, setTranscript] = useState(null) - const [loading, setLoading] = useState(true) - const [error, setError] = useState(null) - - useEffect(() => { - let cancelled = false - setLoading(true) - setError(null) - void (async () => { - try { - const t = await fetchAgentRunTranscript(runId) - if (cancelled) return - setTranscript(t) - } catch (err) { - if (cancelled) return - setError(err instanceof Error ? err.message : String(err)) - setTranscript(null) - } finally { - if (!cancelled) setLoading(false) - } - })() - return () => { cancelled = true } - }, [runId]) + // Live via the turns:events spine: an in-flight run's transcript streams + // in as the agent works; settled runs render from one snapshot fetch. + const { transcript, loading, error } = useAgentRunTranscript(runId) const summary = transcript ?? undefined const items: ConversationItem[] = transcript?.items ?? [] diff --git a/apps/x/apps/renderer/src/components/live-note-sidebar.tsx b/apps/x/apps/renderer/src/components/live-note-sidebar.tsx index 18654729..f863e6cb 100644 --- a/apps/x/apps/renderer/src/components/live-note-sidebar.tsx +++ b/apps/x/apps/renderer/src/components/live-note-sidebar.tsx @@ -13,7 +13,7 @@ import { import { LiveNoteSchema, type LiveNote, type Triggers } from '@x/shared/dist/live-note.js' import { useLiveNoteAgentStatus } from '@/hooks/use-live-note-agent-status' import { formatRelativeTime } from '@/lib/relative-time' -import { fetchAgentRunTranscript, type AgentRunTranscript } from '@/lib/agent-transcript' +import { useAgentRunTranscript } from '@/hooks/use-agent-run-transcript' import { CompactConversation } from '@/components/compact-conversation' export type OpenLiveNotePanelDetail = { @@ -659,37 +659,14 @@ function SectionRegion({ label, children }: { label?: string; children: React.Re } function LastRunTab({ live }: { live: LiveNote }) { - const [transcript, setTranscript] = useState(null) - const [loadingRun, setLoadingRun] = useState(false) - const [fetchError, setFetchError] = useState(null) - const runId = live.lastRunId ?? null - - useEffect(() => { - if (!runId) { - setTranscript(null) - setFetchError(null) - setLoadingRun(false) - return - } - let cancelled = false - setLoadingRun(true) - setFetchError(null) - void (async () => { - try { - const t = await fetchAgentRunTranscript(runId) - if (cancelled) return - setTranscript(t) - } catch (err) { - if (cancelled) return - setFetchError(err instanceof Error ? err.message : String(err)) - setTranscript(null) - } finally { - if (!cancelled) setLoadingRun(false) - } - })() - return () => { cancelled = true } - }, [runId]) + // Live via the turns:events spine: an in-flight run's transcript streams + // in as the agent works; settled runs render from one snapshot fetch. + const { + transcript, + loading: loadingRun, + error: fetchError, + } = useAgentRunTranscript(runId) if (!runId) { return ( diff --git a/apps/x/apps/renderer/src/hooks/use-agent-run-transcript.ts b/apps/x/apps/renderer/src/hooks/use-agent-run-transcript.ts new file mode 100644 index 00000000..eaa2b6ca --- /dev/null +++ b/apps/x/apps/renderer/src/hooks/use-agent-run-transcript.ts @@ -0,0 +1,61 @@ +import { useEffect, useMemo, useState } from 'react' +import { + runToTranscript, + turnStateToTranscript, + type AgentRunTranscript, +} from '@/lib/agent-transcript' +import { useTurn } from '@/hooks/use-turn' + +export interface UseAgentRunTranscriptResult { + transcript: AgentRunTranscript | null + loading: boolean + error: string | null +} + +// Live transcript of a headless agent run (background task, live note) by +// run id. Run ids are turn ids since the runs→turns migration, so useTurn +// keeps the transcript live via the turns:events spine; ids whose turn +// snapshot definitively fails (pre-migration legacy files) fall back to one +// runs:fetch. Callers open this for runs that already exist, so a failed +// snapshot means "not a turn", not "not created yet" — maxRetries 0 makes +// the legacy fallback immediate. +export function useAgentRunTranscript( + runId: string | undefined | null, +): UseAgentRunTranscriptResult { + const id = runId ?? undefined + const { state, error: turnError, snapshotFailed } = useTurn(id, { maxRetries: 0 }) + const [legacy, setLegacy] = useState(null) + const [legacyError, setLegacyError] = useState(null) + + useEffect(() => { + setLegacy(null) + setLegacyError(null) + if (!id || !snapshotFailed) { + return + } + let alive = true + window.ipc + .invoke('runs:fetch', { runId: id }) + .then((run) => { + if (alive) setLegacy(runToTranscript(run)) + }) + .catch((err: unknown) => { + if (alive) setLegacyError(err instanceof Error ? err.message : String(err)) + }) + return () => { + alive = false + } + }, [id, snapshotFailed]) + + const transcript = useMemo(() => { + if (id && state) { + return turnStateToTranscript(id, state) + } + return legacy + }, [id, state, legacy]) + + const error = transcript ? null : (turnError ?? legacyError) + const loading = !!id && !transcript && !error + + return { transcript, loading, error } +} diff --git a/apps/x/apps/renderer/src/hooks/use-turn.ts b/apps/x/apps/renderer/src/hooks/use-turn.ts index e619fcc2..da1c5ed4 100644 --- a/apps/x/apps/renderer/src/hooks/use-turn.ts +++ b/apps/x/apps/renderer/src/hooks/use-turn.ts @@ -1,4 +1,4 @@ -import { useEffect, useState } from 'react' +import { useEffect, useRef, useState } from 'react' import type { TurnState } from '@x/shared/src/turns.js' import { subscribeTurnFeed } from '@/lib/turn-feed' import { followTurn } from '@/lib/turn-follower' @@ -6,6 +6,10 @@ import { followTurn } from '@/lib/turn-follower' export interface UseTurnResult { state: TurnState | null error: string | null + // Snapshot retries exhausted with no state — the id may not be a turn at + // all (e.g. a pre-migration legacy run id). Clears if a later feed event + // recovers the turn. + snapshotFailed: boolean } // Live view of one turn by id: snapshot via sessions:getTurn, then durable @@ -14,20 +18,25 @@ export interface UseTurnResult { // sub-agents. export function useTurn( turnId: string | undefined, - opts?: { enabled?: boolean }, + opts?: { enabled?: boolean; maxRetries?: number }, ): UseTurnResult { const enabled = opts?.enabled ?? true + const maxRetries = opts?.maxRetries const [state, setState] = useState(null) const [error, setError] = useState(null) + const [snapshotFailed, setSnapshotFailed] = useState(false) + const lastTurnId = useRef(undefined) useEffect(() => { - if (!turnId) { + // A different turn means the previous turn's state is stale; a mere + // enabled toggle keeps the last rendered state while hidden. + if (turnId !== lastTurnId.current) { + lastTurnId.current = turnId setState(null) setError(null) - return + setSnapshotFailed(false) } - if (!enabled) { - // Keep the last rendered state while hidden; re-enabling refetches. + if (!turnId || !enabled) { return } return followTurn(turnId, { @@ -36,10 +45,13 @@ export function useTurn( onState: (next) => { setState(next) setError(null) + setSnapshotFailed(false) }, onError: (message) => setError(message), + onSnapshotFailed: () => setSnapshotFailed(true), + ...(maxRetries === undefined ? {} : { maxRetries }), }) - }, [turnId, enabled]) + }, [turnId, enabled, maxRetries]) - return { state, error } + return { state, error, snapshotFailed } } diff --git a/apps/x/apps/renderer/src/lib/turn-follower.test.ts b/apps/x/apps/renderer/src/lib/turn-follower.test.ts index b17adbdd..b8dae0dd 100644 --- a/apps/x/apps/renderer/src/lib/turn-follower.test.ts +++ b/apps/x/apps/renderer/src/lib/turn-follower.test.ts @@ -42,6 +42,7 @@ function harness(fetches: Array) { let unsubscribed = false const states: TurnState[] = [] const errors: string[] = [] + const snapshotFailures: string[] = [] const deps: TurnFollowerDeps = { fetchTurn, subscribe: (fn) => { @@ -53,6 +54,7 @@ function harness(fetches: Array) { }, onState: (state) => states.push(state), onError: (message) => errors.push(message), + onSnapshotFailed: (message) => snapshotFailures.push(message), maxRetries: 0, } return { @@ -60,6 +62,7 @@ function harness(fetches: Array) { fetchTurn, states, errors, + snapshotFailures, emit: (event: TurnBusEvent) => listener?.(event), get unsubscribed() { return unsubscribed @@ -155,6 +158,18 @@ describe('followTurn', () => { detach() }) + it('reports snapshot failure once retries are exhausted', async () => { + const h = harness([new Error('turn not found: legacy run id')]) + const detach = followTurn(TURN, h.deps) + await flush() + + // maxRetries 0: the first failure is definitive (legacy-run fallback + // hinges on this signal firing). + expect(h.snapshotFailures).toEqual(['turn not found: legacy run id']) + expect(h.states).toHaveLength(0) + detach() + }) + it('stops delivering after detach and unsubscribes from the feed', async () => { const full = log() const h = harness([full.slice(0, 2)]) diff --git a/apps/x/apps/renderer/src/lib/turn-follower.ts b/apps/x/apps/renderer/src/lib/turn-follower.ts index 7c289263..e0a2878c 100644 --- a/apps/x/apps/renderer/src/lib/turn-follower.ts +++ b/apps/x/apps/renderer/src/lib/turn-follower.ts @@ -23,6 +23,10 @@ export interface TurnFollowerDeps { subscribe: (listener: (event: TurnBusEvent) => void) => () => void onState: (state: TurnState) => void onError: (message: string) => void + // Fired when snapshot retries are exhausted (the id may not be a turn at + // all — e.g. a pre-migration legacy run). A later feed event still retries + // and can recover. + onSnapshotFailed?: (message: string) => void retryDelayMs?: number maxRetries?: number } @@ -89,7 +93,7 @@ export function followTurn(turnId: string, deps: TurnFollowerDeps): () => void { } } publish() - } catch { + } catch (err) { if (!alive) return // Turn file may not exist yet (subscription raced creation) or the // fetch failed transiently. @@ -97,6 +101,8 @@ export function followTurn(turnId: string, deps: TurnFollowerDeps): () => void { if (retries < maxRetries) { retries += 1 retryTimer = setTimeout(() => void fetchSnapshot(), retryDelayMs) + } else { + deps.onSnapshotFailed?.(err instanceof Error ? err.message : String(err)) } } finally { fetching = false diff --git a/apps/x/packages/core/docs/session-design.md b/apps/x/packages/core/docs/session-design.md index 2694f9e9..f2ba6ccc 100644 --- a/apps/x/packages/core/docs/session-design.md +++ b/apps/x/packages/core/docs/session-design.md @@ -376,7 +376,10 @@ which governs mutation of live logs, not their removal. ## 11. Headless standalone turns A helper covers the non-session callers (background tasks, live notes, -knowledge pipelines, scheduled agents): +knowledge pipelines, scheduled agents). Implemented as +`HeadlessAgentRunner` in `agents/headless.ts` (start/run handle with +turn id, reduced state, and final assistant text); the shape below is +the contract it fulfils: ```ts function runHeadlessTurn(input: { diff --git a/apps/x/packages/core/src/agents/headless.ts b/apps/x/packages/core/src/agents/headless.ts index 0f17eae4..0372d977 100644 --- a/apps/x/packages/core/src/agents/headless.ts +++ b/apps/x/packages/core/src/agents/headless.ts @@ -163,6 +163,18 @@ export class HeadlessAgentRunner implements IHeadlessAgentRunner { const execution = this.turnRuntime.advanceTurn(turnId, undefined, { signal: options.signal, }); + // Drain the execution stream: live delivery rides the turn event bus, + // and an unconsumed HotStream would buffer every durable event in + // memory until the turn settles. + void (async () => { + try { + for await (const event of execution.events) { + void event; + } + } catch { + // Infrastructure failures surface through the outcome. + } + })(); const done = execution.outcome.then(async (outcome) => { const state = reduceTurn( (await this.turnRuntime.getTurn(turnId)).events, diff --git a/apps/x/packages/core/src/application/assistant/skills/web-search/skill.ts b/apps/x/packages/core/src/application/assistant/skills/web-search/skill.ts deleted file mode 100644 index d3d0e6e1..00000000 --- a/apps/x/packages/core/src/application/assistant/skills/web-search/skill.ts +++ /dev/null @@ -1,52 +0,0 @@ -export const skill = String.raw` -# Web Search Skill - -You have access to two search tools for finding information on the internet. Choose the right one based on the user's intent. - -## Tools - -### web-search (Brave Search) -Quick, general-purpose web search. Returns titles, URLs, and short descriptions. - -**Best for:** -- Quick lookups for things that change ("current price of Bitcoin", "weather in SF") -- Current events and breaking news -- Finding a specific website or page -- Simple questions with direct answers -- Checking a fact or date - -### research-search (Exa Search) -Deep, research-oriented search. Returns full article text, highlights, and metadata (author, published date). - -**Best for:** -- Exploring a topic in depth ("what are the latest advances in CRISPR") -- Finding articles, blog posts, papers, and quality sources -- Discovering companies, people, or organizations -- Research where you need rich context, not just links -- When the user says "research", "find articles about", "look into", "deep dive" - -**Category filter:** Use the category parameter when the user's intent clearly maps to one: company, research paper, news, tweet, personal site, financial report, people. - -## How Many Searches to Do - -**CRITICAL: Always start with exactly ONE search call.** Pick the single best tool (\`web-search\` or \`research-search\`) and make one request. Wait for the result before deciding if more searches are needed. - -**NEVER call multiple search tools simultaneously.** No parallel web-search + research-search. No firing off two web-searches at once. Always sequential: one search at a time. - -Only make a follow-up search if: -- The first search returned truly uninformative or irrelevant results -- The query has clearly distinct sub-topics that the first search couldn't cover (e.g., "compare X and Y" after getting results for X only) -- The user explicitly asks you to dig deeper - -One good search is almost always enough. Default to one and stop. - -## Choosing Between the Two - -If both tools are attached, prefer: -- \`web-search\` when the user wants a quick answer or specific link -- \`research-search\` when the user wants to learn, explore, or gather sources - -If only one is attached, use whichever is available. -`; - -export default skill; diff --git a/apps/x/packages/core/src/sessions/headless.ts b/apps/x/packages/core/src/sessions/headless.ts deleted file mode 100644 index 071d7fd4..00000000 --- a/apps/x/packages/core/src/sessions/headless.ts +++ /dev/null @@ -1,41 +0,0 @@ -import type { z } from "zod"; -import type { UserMessage } from "@x/shared/dist/message.js"; -import type { - ConversationMessage, - RequestedAgent, -} from "@x/shared/dist/turns.js"; -import type { ITurnRuntime, TurnOutcome } from "../turns/api.js"; - -// Standalone turns for non-session callers (background tasks, live notes, -// knowledge pipelines, scheduled agents): sessionId null, automatic -// permissions, no human. Never appears in the session index; callers keep -// the turnId if they need history. -export async function runHeadlessTurn( - turnRuntime: ITurnRuntime, - input: { - agent: z.infer; - context?: Array>; - input: z.infer; - maxModelCalls?: number; - signal?: AbortSignal; - }, -): Promise<{ turnId: string; outcome: TurnOutcome }> { - const turnId = await turnRuntime.createTurn({ - agent: input.agent, - sessionId: null, - context: input.context ?? [], - input: input.input, - config: { - autoPermission: true, - humanAvailable: false, - ...(input.maxModelCalls === undefined - ? {} - : { maxModelCalls: input.maxModelCalls }), - }, - }); - const execution = turnRuntime.advanceTurn(turnId, undefined, { - signal: input.signal, - }); - const outcome = await execution.outcome; - return { turnId, outcome }; -} diff --git a/apps/x/packages/core/src/sessions/index.ts b/apps/x/packages/core/src/sessions/index.ts index df906d01..7e7e994c 100644 --- a/apps/x/packages/core/src/sessions/index.ts +++ b/apps/x/packages/core/src/sessions/index.ts @@ -1,7 +1,6 @@ export * from "./api.js"; export * from "./bus.js"; export * from "./fs-repo.js"; -export * from "./headless.js"; export * from "./in-memory-session-repo.js"; export * from "./repo.js"; export * from "./session-index.js"; diff --git a/apps/x/packages/core/src/sessions/sessions.test.ts b/apps/x/packages/core/src/sessions/sessions.test.ts index 93fc1a40..9bca5360 100644 --- a/apps/x/packages/core/src/sessions/sessions.test.ts +++ b/apps/x/packages/core/src/sessions/sessions.test.ts @@ -17,7 +17,6 @@ import type { import { TurnInputError } from "../turns/api.js"; import { HotStream } from "../turns/stream.js"; import { TurnNotSettledError } from "./api.js"; -import { runHeadlessTurn } from "./headless.js"; import { InMemorySessionRepo } from "./in-memory-session-repo.js"; import type { ISessionRepo } from "./repo.js"; import { SessionsImpl } from "./sessions.js"; @@ -823,26 +822,6 @@ describe("startup scan (13.6)", () => { }); }); -describe("headless standalone turns (13.8)", () => { - it("runs a standalone turn with sessionId null, auto permission, no human", async () => { - const fake = new FakeTurnRuntime(); - const { turnId, outcome } = await runHeadlessTurn(fake, { - agent: { agentId: "background-task-agent" }, - input: user("summarize"), - maxModelCalls: 3, - }); - expect(fake.createTurnInputs[0]).toEqual({ - agent: { agentId: "background-task-agent" }, - sessionId: null, - context: [], - input: user("summarize"), - config: { autoPermission: true, humanAvailable: false, maxModelCalls: 3 }, - }); - expect(outcome.status).toBe("completed"); - expect(fake.advanceCalls).toEqual([{ turnId, input: undefined }]); - }); -}); - describe("active-skill carry-forward", () => { const skillTool = { toolId: "builtin:file-writeText",