Skip to content

Revert fresh-agent progressive hydration #470

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
danshapiro merged 1 commit into from
Jun 22, 2026
Merged

Revert fresh-agent progressive hydration #470

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1,397 changes: 0 additions & 1,397 deletions docs/superpowers/plans/2026-06-21-fresh-agent-progressive-hydration.md
View file
Open in desktop

This file was deleted.

54 changes: 4 additions & 50 deletions server/agent-api/router.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -84,8 +84,6 @@ function freshAgentErrorStatus(error: unknown): number {
}

const FRESH_AGENT_SEND_IDLE_TIMEOUT_MS = 600_000
const FRESH_AGENT_CAPTURE_TURN_LIMIT = 200
const FRESH_AGENT_CAPTURE_MAX_PAGES = 1000

async function waitForFreshAgentIdle(
runtimeManager: NonNullable<FreshAgentRuntimeManagerLike>,
Expand Down Expand Up @@ -321,8 +319,8 @@ function terminalInputFailureMessage(result: Exclude<TerminalInputResult, { stat
return 'Terminal is not running.'
}

function renderFreshAgentTranscript(input: any): string {
const turns = Array.isArray(input?.turns) ? input.turns : []
function renderFreshAgentTranscript(snapshot: any): string {
const turns = Array.isArray(snapshot?.turns) ? snapshot.turns : []
return turns.map((turn: any) => {
const role = typeof turn?.role === 'string' ? turn.role : 'turn'
const items = Array.isArray(turn?.items) ? turn.items : []
Expand All @@ -339,48 +337,6 @@ function renderFreshAgentTranscript(input: any): string {
}).join('\n\n')
}

async function captureFreshAgentTranscript(
runtimeManager: NonNullable<FreshAgentRuntimeManagerLike>,
input: FreshAgentThreadLocator,
): Promise<{ turns: any[] }> {
if (!runtimeManager.getTurnPage) {
throw new Error('fresh-agent transcript paging not available on this server')
}
const pages: any[][] = []
const seenCursors = new Set<string>()
let cursor: string | undefined
let revision: number | undefined

for (let pageIndex = 0; pageIndex < FRESH_AGENT_CAPTURE_MAX_PAGES; pageIndex += 1) {
const page = await runtimeManager.getTurnPage({
...input,
...(cursor ? { cursor } : {}),
...(revision !== undefined ? { revision } : {}),
limit: FRESH_AGENT_CAPTURE_TURN_LIMIT,
includeBodies: true,
priority: 'visible',
})
pages.push(Array.isArray(page?.turns) ? page.turns : [])
if (revision === undefined && typeof page?.revision === 'number' && Number.isFinite(page.revision)) {
revision = page.revision
}

const nextCursor = typeof page?.nextCursor === 'string' && page.nextCursor.length > 0
? page.nextCursor
: undefined
if (!nextCursor) {
return { turns: pages.reverse().flat() }
}
if (seenCursors.has(nextCursor)) {
throw new Error('fresh-agent transcript paging returned a repeated cursor')
}
seenCursors.add(nextCursor)
cursor = nextCursor
}

throw new Error(`fresh-agent transcript exceeded ${FRESH_AGENT_CAPTURE_MAX_PAGES} pages`)
}

function shouldWaitForCodexIdentity(payload: Record<string, unknown>): boolean {
return truthy(payload.waitForCodexIdentity)
}
Expand Down Expand Up @@ -506,7 +462,6 @@ type FreshAgentRuntimeManagerLike = {
send: (locator: FreshAgentSessionLocator, input: { text: string; settings?: any }) => Promise<{ sessionId?: string; submittedTurnId?: string; sessionRef?: { provider: string; sessionId: string } } | void>
attach: (locator: FreshAgentSessionLocator) => Promise<{ sessionId: string; sessionRef?: { provider: string; sessionId: string } }>
getSnapshot: (input: FreshAgentThreadLocator) => Promise<any>
getTurnPage?: (input: FreshAgentThreadLocator & { cursor?: string; revision?: number; limit?: number; includeBodies?: boolean; priority?: string }) => Promise<any>
}

const parseRegex = (raw: string) => {
Expand Down Expand Up @@ -946,15 +901,14 @@ export function createAgentApiRouter({
if (paneSnapshot?.kind === 'fresh-agent') {
const c = paneSnapshot.paneContent || {}
if (!freshAgentRuntimeManager) return res.status(503).json(fail('fresh-agent runtime not available on this server'))
if (!freshAgentRuntimeManager.getTurnPage) return res.status(503).json(fail('fresh-agent transcript paging not available on this server'))
try {
const transcript = await captureFreshAgentTranscript(freshAgentRuntimeManager, {
const snapshot = await freshAgentRuntimeManager.getSnapshot({
sessionType: c.sessionType,
provider: c.provider,
threadId: c.sessionId,
cwd: freshAgentPaneCwd(c),
})
return res.type('text/plain').send(renderFreshAgentTranscript(transcript))
return res.type('text/plain').send(renderFreshAgentTranscript(snapshot))
} catch (err: any) {
return res.status(agentRouteErrorStatus(err)).json(fail(err?.message || 'fresh-agent capture failed'))
}
Expand Down
2 changes: 1 addition & 1 deletion server/fresh-agent/adapters/claude/adapter.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -253,7 +253,7 @@ export function createClaudeFreshAgentAdapter(deps: ClaudeFreshAgentAdapterDeps)
sessionId: thread.threadId,
cursor: typeof query.cursor === 'string' ? query.cursor : undefined,
priority: typeof query.priority === 'string' ? query.priority as 'visible' | 'background' : undefined,
revision: typeof query.revision === 'number' ? query.revision : undefined,
revision: Number(query.revision),
limit: typeof query.limit === 'number' ? query.limit : undefined,
includeBodies: query.includeBodies === true,
})
Expand Down
109 changes: 53 additions & 56 deletions server/fresh-agent/adapters/codex/adapter.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -71,7 +71,6 @@ type CodexRuntimePort = {
cursor?: string
limit?: number
itemsView?: 'notLoaded' | 'summary' | 'full'
sortDirection?: 'asc' | 'desc'
}) => Promise<Record<string, any>>
readThreadTurn: (input: { threadId: string; turnId: string; revision?: number }) => Promise<Record<string, any>>
}
Expand Down Expand Up @@ -264,22 +263,7 @@ function isCodexIncludeTurnsUnavailable(error: unknown): boolean {
|| error.message.includes('not materialized yet')
}

function nonEmptyString(value: unknown): string | undefined {
return typeof value === 'string' && value.length > 0 ? value : undefined
}

function findActiveTurnId(rawSnapshot: Record<string, any>): string | undefined {
const thread = rawSnapshot.thread && typeof rawSnapshot.thread === 'object' && !Array.isArray(rawSnapshot.thread)
? rawSnapshot.thread as Record<string, unknown>
: {}
const status = thread.status && typeof thread.status === 'object' && !Array.isArray(thread.status)
? thread.status as Record<string, unknown>
: {}
const metadataTurnId = nonEmptyString(status.activeTurnId)
?? nonEmptyString(status.turnId)
?? nonEmptyString(thread.activeTurnId)
if (metadataTurnId) return metadataTurnId

const turns = Array.isArray(rawSnapshot.thread?.turns) ? rawSnapshot.thread.turns : []
for (let index = turns.length - 1; index >= 0; index -= 1) {
const turn = turns[index]
Expand Down Expand Up @@ -541,33 +525,18 @@ export function createCodexFreshAgentAdapter(deps: {
let providerCursor: string | null | undefined
let backwardsCursor: string | null | undefined

const finishPage = (nextCursor: string | null, pageBackwardsCursor: string | null) => {
const readingOrderTurns = turns.slice().reverse()
return FreshAgentTurnPageSchema.parse({
sessionType: 'freshcodex',
provider: 'codex',
threadId: input.threadId,
revision: input.revision,
nextCursor,
backwardsCursor: pageBackwardsCursor,
turns: readingOrderTurns.map((turn, index) => ({ ...turn, ordinal: index })),
bodies: Object.fromEntries(readingOrderTurns.map((turn) => [turn.turnId, turn])),
})
}

const appendTurnRows = (rawTurn: Record<string, unknown>, offset: number, providerCursorAfterTurn: string | null) => {
const normalized = normalizeSingleRawTurn({
threadId: input.threadId,
revision: input.revision,
rawTurn,
})
const newestFirstRows = normalized.turns.slice().reverse()
const availableRows = newestFirstRows.slice(offset)
const availableRows = normalized.turns.slice(offset)
const remainingSlots = limit - turns.length
const selectedRows = availableRows.slice(0, remainingSlots)
turns.push(...selectedRows)
const nextDisplayOffset = offset + selectedRows.length
if (nextDisplayOffset < newestFirstRows.length) {
if (nextDisplayOffset < normalized.turns.length) {
return createDisplayCursor({
threadId: input.threadId,
revision: input.revision,
Expand Down Expand Up @@ -600,7 +569,16 @@ export function createCodexFreshAgentAdapter(deps: {
if (cursor.rawTurn) {
const nextCursor = appendTurnRows(cursor.rawTurn, cursor.nextDisplayOffset, cursor.providerCursor)
if (turns.length >= limit || nextCursor || !cursor.providerCursor) {
return finishPage(nextCursor, null)
return FreshAgentTurnPageSchema.parse({
sessionType: 'freshcodex',
provider: 'codex',
threadId: input.threadId,
revision: input.revision,
nextCursor,
backwardsCursor: null,
turns: turns.map((turn, index) => ({ ...turn, ordinal: index })),
bodies: Object.fromEntries(turns.map((turn) => [turn.turnId, turn])),
})
}
}
}
Expand All @@ -611,7 +589,6 @@ export function createCodexFreshAgentAdapter(deps: {
...(providerCursor ? { cursor: providerCursor } : {}),
limit: 1,
itemsView: 'full',
sortDirection: 'desc',
})
const pageRevision = Number(rawPage.revision ?? input.revision)
if (pageRevision !== input.revision) {
Expand All @@ -628,12 +605,30 @@ export function createCodexFreshAgentAdapter(deps: {
}
const nextCursor = appendTurnRows(rawTurns[0], 0, providerCursor)
if (turns.length >= limit || nextCursor) {
return finishPage(nextCursor, backwardsCursor ?? null)
return FreshAgentTurnPageSchema.parse({
sessionType: 'freshcodex',
provider: 'codex',
threadId: input.threadId,
revision: input.revision,
nextCursor,
backwardsCursor: backwardsCursor ?? null,
turns: turns.map((turn, index) => ({ ...turn, ordinal: index })),
bodies: Object.fromEntries(turns.map((turn) => [turn.turnId, turn])),
})
}
if (!providerCursor) break
}

return finishPage(null, backwardsCursor ?? null)
return FreshAgentTurnPageSchema.parse({
sessionType: 'freshcodex',
provider: 'codex',
threadId: input.threadId,
revision: input.revision,
nextCursor: null,
backwardsCursor: backwardsCursor ?? null,
turns: turns.map((turn, index) => ({ ...turn, ordinal: index })),
bodies: Object.fromEntries(turns.map((turn) => [turn.turnId, turn])),
})
}

const findDisplayIndexEntry = (threadId: string, revision: number, displayTurnId: string): DisplayIndexEntry | undefined => {
Expand All @@ -653,7 +648,6 @@ export function createCodexFreshAgentAdapter(deps: {
...(cursor ? { cursor } : {}),
limit: 100,
itemsView: 'full',
sortDirection: 'desc',
})
const currentRevision = Number(rawPage.revision ?? revision)
normalizeRawPage({ threadId, revision: currentRevision, rawPage })
Expand Down Expand Up @@ -1031,23 +1025,38 @@ export function createCodexFreshAgentAdapter(deps: {
thread.threadId,
settingsFromLocator(thread) ?? settingsByThread.get(thread.threadId),
)
const rawSnapshot = await runtime.readThread({ threadId: thread.threadId, includeTurns: false })
let rawSnapshot: Record<string, any>
try {
rawSnapshot = await runtime.readThread({ threadId: thread.threadId, includeTurns: true })
} catch (error) {
if (!isCodexIncludeTurnsUnavailable(error)) {
throw error
}
rawSnapshot = await runtime.readThread({ threadId: thread.threadId, includeTurns: false })
}
const rawThreadTurns: unknown[] = Array.isArray(rawSnapshot.thread?.turns)
? rawSnapshot.thread.turns
: []
const activeTurnId = findActiveTurnId(rawSnapshot)
if (activeTurnId) {
activeTurnByThread.set(thread.threadId, activeTurnId)
} else if (normalizeCodexThreadStatus(rawSnapshot.thread?.status) !== 'running') {
activeTurnByThread.delete(thread.threadId)
}
const rawTurns = rawThreadTurns
.filter((turn): turn is Record<string, unknown> => !!turn && typeof turn === 'object' && !Array.isArray(turn))
const revisionNumber = Number(rawSnapshot.thread?.updatedAt ?? revision ?? 0)
if (!Number.isFinite(revisionNumber)) {
throw new FreshAgentUnprovableThreadRevisionError(0)
}
const turns = normalizeRawTurns({
threadId: thread.threadId,
revision: revisionNumber,
rawTurns,
})
return normalizeCodexThreadSnapshot({
threadId: thread.threadId,
revision: revisionNumber,
status: normalizeCodexThreadStatus(rawSnapshot.thread?.status),
transcript: {
turns: [],
turns,
},
rawSnapshot,
})
Expand All @@ -1058,22 +1067,10 @@ export function createCodexFreshAgentAdapter(deps: {
thread.threadId,
settingsFromLocator(thread) ?? settingsByThread.get(thread.threadId),
)
let revision = typeof query.revision === 'number' ? query.revision : undefined
if (revision == null && typeof query.cursor !== 'string') {
const metadata = await runtime.readThread({ threadId: thread.threadId, includeTurns: false })
const metadataRevision = Number(metadata.thread?.updatedAt)
if (!Number.isFinite(metadataRevision)) {
throw new FreshAgentUnprovableThreadRevisionError(0)
}
revision = metadataRevision
}
if (revision == null) {
throw new FreshAgentUnprovableThreadRevisionError(0)
}
return normalizeDisplayTurnPage({
runtime,
threadId: thread.threadId,
revision,
revision: Number(query.revision ?? 0),
cursor: typeof query.cursor === 'string' ? query.cursor : undefined,
limit: typeof query.limit === 'number' ? query.limit : undefined,
})
Expand Down
Loading
Loading