mirror of
https://github.com/rowboatlabs/rowboat.git
synced 2026-07-12 21:02:17 +02:00
refactor(x): finish turn-spine adoption for headless surfaces
Background-task and live-note transcript views move from fetch-once to the shared useAgentRunTranscript hook (useTurn over the turns:events spine, with a one-shot legacy runs:fetch fallback for pre-migration run ids), so an in-flight run's transcript now streams live. The headless runner drains its execution stream — delivery rides the turn event bus, and an unconsumed HotStream buffered every durable event until settle. Deletes the caller-less runHeadlessTurn duplicate (HeadlessAgentRunner is the implementation; session-design.md now says so) and the orphaned web-search skill file that was never registered. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
parent
248bb0d1a1
commit
f916cb1d70
12 changed files with 132 additions and 180 deletions
|
|
@ -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<AgentRunTranscript | null>(null)
|
||||
const [loading, setLoading] = useState(true)
|
||||
const [error, setError] = useState<string | null>(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 ?? []
|
||||
|
|
|
|||
|
|
@ -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<AgentRunTranscript | null>(null)
|
||||
const [loadingRun, setLoadingRun] = useState(false)
|
||||
const [fetchError, setFetchError] = useState<string | null>(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 (
|
||||
|
|
|
|||
61
apps/x/apps/renderer/src/hooks/use-agent-run-transcript.ts
Normal file
61
apps/x/apps/renderer/src/hooks/use-agent-run-transcript.ts
Normal file
|
|
@ -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<AgentRunTranscript | null>(null)
|
||||
const [legacyError, setLegacyError] = useState<string | null>(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 }
|
||||
}
|
||||
|
|
@ -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<TurnState | null>(null)
|
||||
const [error, setError] = useState<string | null>(null)
|
||||
const [snapshotFailed, setSnapshotFailed] = useState(false)
|
||||
const lastTurnId = useRef<string | undefined>(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 }
|
||||
}
|
||||
|
|
|
|||
|
|
@ -42,6 +42,7 @@ function harness(fetches: Array<TEvent[] | Error>) {
|
|||
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<TEvent[] | Error>) {
|
|||
},
|
||||
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<TEvent[] | Error>) {
|
|||
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)])
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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: {
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
|
|
@ -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<typeof RequestedAgent>;
|
||||
context?: Array<z.infer<typeof ConversationMessage>>;
|
||||
input: z.infer<typeof UserMessage>;
|
||||
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 };
|
||||
}
|
||||
|
|
@ -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";
|
||||
|
|
|
|||
|
|
@ -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",
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue