diff --git a/packages/plugin-sdk/src/contracts.ts b/packages/plugin-sdk/src/contracts.ts index 11bf5451e5..57c2c40da4 100644 --- a/packages/plugin-sdk/src/contracts.ts +++ b/packages/plugin-sdk/src/contracts.ts @@ -12487,8 +12487,24 @@ export const contracts: { HistoryView: { description: 'Tool output is available through source reads in both views. Conversation\nwindows carry tool metadata only, keeping large results out of chat paging.', - type: 'string', - enum: ['text', 'conversation'], + oneOf: [ + { + type: 'string', + enum: ['text', 'conversation'], + }, + { + description: + 'Visible chat rows. Completed work can be read separately at the same fence.', + type: 'string', + const: 'chat', + }, + { + description: + 'All conversation rows belonging to the live history Run at the anchor.', + type: 'string', + const: 'turn', + }, + ], }, }, }, @@ -12631,6 +12647,16 @@ export const contracts: { }, ], }, + foldedWork: { + anyOf: [ + { + $ref: '#/$defs/HistoryFoldedWork', + }, + { + type: 'null', + }, + ], + }, field: { $ref: '#/$defs/HistoryField', }, @@ -13034,48 +13060,54 @@ export const contracts: { additionalProperties: false, required: ['inputTokens', 'readTokens'], }, - HistoryField: { - description: 'A readable field in a conversation window, in presentation order.', - type: 'string', - enum: ['thinking', 'body'], - }, - Role: { - description: - 'Canonical message author/kind; the field identifies body versus thinking.\nTool body JSON contributes string values, never its keys.', - type: 'string', - enum: [ - 'system', - 'context', - 'tool', - 'compaction', - 'failure', - 'user', - 'assistant', - 'tool_call', - 'tool_result', - ], - }, - HistoryOrder: { - description: - 'Lexicographic tree order: original marker sequences followed by the leaf\nsequence/ordinal. This is presentation order, never a fabricated Root sequence.', + HistoryFoldedWork: { + description: 'Completed activity available on demand at the same history fence.', type: 'object', properties: { - path: { + thoughts: { + type: 'integer', + format: 'uint64', + minimum: 0, + }, + tools: { + type: 'integer', + format: 'uint64', + minimum: 0, + }, + messages: { + type: 'integer', + format: 'uint64', + minimum: 0, + }, + imageCount: { + type: 'integer', + format: 'uint64', + minimum: 0, + }, + images: { type: 'array', items: { - type: 'integer', - format: 'uint64', - minimum: 0, + $ref: '#/$defs/HistoryWorkImage', }, }, - itemIndex: { - type: 'integer', - format: 'uint64', + }, + additionalProperties: false, + required: ['thoughts', 'tools', 'messages', 'imageCount', 'images'], + }, + HistoryWorkImage: { + type: 'object', + properties: { + position: { + $ref: '#/$defs/HistoryRowAddress', + }, + image: { + type: ['integer', 'null'], + format: 'uint32', minimum: 0, }, }, additionalProperties: false, - required: ['path', 'itemIndex'], + required: ['position'], }, HistoryRowAddress: { description: @@ -13176,6 +13208,49 @@ export const contracts: { type: 'string', format: 'uuid', }, + HistoryField: { + description: 'A readable field in a conversation window, in presentation order.', + type: 'string', + enum: ['thinking', 'body'], + }, + Role: { + description: + 'Canonical message author/kind; the field identifies body versus thinking.\nTool body JSON contributes string values, never its keys.', + type: 'string', + enum: [ + 'system', + 'context', + 'tool', + 'compaction', + 'failure', + 'user', + 'assistant', + 'tool_call', + 'tool_result', + ], + }, + HistoryOrder: { + description: + 'Lexicographic tree order: original marker sequences followed by the leaf\nsequence/ordinal. This is presentation order, never a fabricated Root sequence.', + type: 'object', + properties: { + path: { + type: 'array', + items: { + type: 'integer', + format: 'uint64', + minimum: 0, + }, + }, + itemIndex: { + type: 'integer', + format: 'uint64', + minimum: 0, + }, + }, + additionalProperties: false, + required: ['path', 'itemIndex'], + }, HistoryTool: { description: 'Canonical references at the requested fence, never execution authority.\nBoth references are present only for a unique same-Turn call/result pair\nwith matching origin and parentage. An unpaired row retains its own reference.', @@ -20553,8 +20628,24 @@ export const contracts: { HistoryView: { description: 'Tool output is available through source reads in both views. Conversation\nwindows carry tool metadata only, keeping large results out of chat paging.', - type: 'string', - enum: ['text', 'conversation'], + oneOf: [ + { + type: 'string', + enum: ['text', 'conversation'], + }, + { + description: + 'Visible chat rows. Completed work can be read separately at the same fence.', + type: 'string', + const: 'chat', + }, + { + description: + 'All conversation rows belonging to the live history Run at the anchor.', + type: 'string', + const: 'turn', + }, + ], }, }, }, @@ -20697,6 +20788,16 @@ export const contracts: { }, ], }, + foldedWork: { + anyOf: [ + { + $ref: '#/$defs/HistoryFoldedWork', + }, + { + type: 'null', + }, + ], + }, field: { $ref: '#/$defs/HistoryField', }, @@ -21100,48 +21201,54 @@ export const contracts: { additionalProperties: false, required: ['inputTokens', 'readTokens'], }, - HistoryField: { - description: 'A readable field in a conversation window, in presentation order.', - type: 'string', - enum: ['thinking', 'body'], - }, - Role: { - description: - 'Canonical message author/kind; the field identifies body versus thinking.\nTool body JSON contributes string values, never its keys.', - type: 'string', - enum: [ - 'system', - 'context', - 'tool', - 'compaction', - 'failure', - 'user', - 'assistant', - 'tool_call', - 'tool_result', - ], - }, - HistoryOrder: { - description: - 'Lexicographic tree order: original marker sequences followed by the leaf\nsequence/ordinal. This is presentation order, never a fabricated Root sequence.', + HistoryFoldedWork: { + description: 'Completed activity available on demand at the same history fence.', type: 'object', properties: { - path: { + thoughts: { + type: 'integer', + format: 'uint64', + minimum: 0, + }, + tools: { + type: 'integer', + format: 'uint64', + minimum: 0, + }, + messages: { + type: 'integer', + format: 'uint64', + minimum: 0, + }, + imageCount: { + type: 'integer', + format: 'uint64', + minimum: 0, + }, + images: { type: 'array', items: { - type: 'integer', - format: 'uint64', - minimum: 0, + $ref: '#/$defs/HistoryWorkImage', }, }, - itemIndex: { - type: 'integer', - format: 'uint64', + }, + additionalProperties: false, + required: ['thoughts', 'tools', 'messages', 'imageCount', 'images'], + }, + HistoryWorkImage: { + type: 'object', + properties: { + position: { + $ref: '#/$defs/HistoryRowAddress', + }, + image: { + type: ['integer', 'null'], + format: 'uint32', minimum: 0, }, }, additionalProperties: false, - required: ['path', 'itemIndex'], + required: ['position'], }, HistoryRowAddress: { description: @@ -21242,6 +21349,49 @@ export const contracts: { type: 'string', format: 'uuid', }, + HistoryField: { + description: 'A readable field in a conversation window, in presentation order.', + type: 'string', + enum: ['thinking', 'body'], + }, + Role: { + description: + 'Canonical message author/kind; the field identifies body versus thinking.\nTool body JSON contributes string values, never its keys.', + type: 'string', + enum: [ + 'system', + 'context', + 'tool', + 'compaction', + 'failure', + 'user', + 'assistant', + 'tool_call', + 'tool_result', + ], + }, + HistoryOrder: { + description: + 'Lexicographic tree order: original marker sequences followed by the leaf\nsequence/ordinal. This is presentation order, never a fabricated Root sequence.', + type: 'object', + properties: { + path: { + type: 'array', + items: { + type: 'integer', + format: 'uint64', + minimum: 0, + }, + }, + itemIndex: { + type: 'integer', + format: 'uint64', + minimum: 0, + }, + }, + additionalProperties: false, + required: ['path', 'itemIndex'], + }, HistoryTool: { description: 'Canonical references at the requested fence, never execution authority.\nBoth references are present only for a unique same-Turn call/result pair\nwith matching origin and parentage. An unpaired row retains its own reference.', diff --git a/packages/plugin-sdk/src/generated/methods.ts b/packages/plugin-sdk/src/generated/methods.ts index dd0a37201e..f38fd78d20 100644 --- a/packages/plugin-sdk/src/generated/methods.ts +++ b/packages/plugin-sdk/src/generated/methods.ts @@ -533,7 +533,10 @@ export type AgentLoadedHistoryCheckpointHistoryInput_HistoryField = 'thinking' | * Tool output is available through source reads in both views. Conversation * windows carry tool metadata only, keeping large results out of chat paging. */ -export type AgentLoadedHistoryCheckpointHistoryInput_HistoryView = 'text' | 'conversation'; +export type AgentLoadedHistoryCheckpointHistoryInput_HistoryView = + | ('text' | 'conversation') + | 'chat' + | 'turn'; /** * An original row within the actual read capability supplied by the caller. * Marker paths are relative to that capability; none of these fields grants access. diff --git a/packages/plugin-sdk/src/generated/runtime.ts b/packages/plugin-sdk/src/generated/runtime.ts index a8b4520431..92576a0c65 100644 --- a/packages/plugin-sdk/src/generated/runtime.ts +++ b/packages/plugin-sdk/src/generated/runtime.ts @@ -4284,7 +4284,7 @@ export type HistoryField = 'thinking' | 'body'; * Tool output is available through source reads in both views. Conversation * windows carry tool metadata only, keeping large results out of chat paging. */ -export type HistoryView = 'text' | 'conversation'; +export type HistoryView = ('text' | 'conversation') | 'chat' | 'turn'; /** * Trusted producer attribution, preserved independently of editable content. */ @@ -8120,6 +8120,7 @@ export interface AgentHistoryChunk { */ attachments?: AgentHistoryAttachment[]; timing?: HistoryTurnTiming | null; + foldedWork?: HistoryFoldedWork | null; field: HistoryField; messageId?: string | null; turnId?: string | null; @@ -8173,6 +8174,20 @@ export interface PromptCacheUsage { inputTokens: number; readTokens: number; } +/** + * Completed activity available on demand at the same history fence. + */ +export interface HistoryFoldedWork { + thoughts: number; + tools: number; + messages: number; + imageCount: number; + images: HistoryWorkImage[]; +} +export interface HistoryWorkImage { + position: HistoryRowAddress; + image?: number | null; +} /** * Canonical references at the requested fence, never execution authority. * Both references are present only for a unique same-Turn call/result pair diff --git a/plugins/agent/api/src/contract.rs b/plugins/agent/api/src/contract.rs index 03a0ab6157..164f22268b 100644 --- a/plugins/agent/api/src/contract.rs +++ b/plugins/agent/api/src/contract.rs @@ -1413,6 +1413,8 @@ pub struct HistoryChunk { pub attachments: Vec, #[serde(skip_serializing_if = "Option::is_none")] pub timing: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub folded_work: Option, pub field: maka_agent_ports::session::history::HistoryField, pub message_id: Option, pub turn_id: Option, @@ -1670,6 +1672,7 @@ impl From for HistoryPage { }) .collect(), timing: chunk.timing, + folded_work: chunk.folded_work, field: chunk.field, message_id: chunk.message_id, turn_id: chunk.turn_id, diff --git a/plugins/agent/conversations/ui/Conversation.tsx b/plugins/agent/conversations/ui/Conversation.tsx index 6805b64280..2301087d75 100644 --- a/plugins/agent/conversations/ui/Conversation.tsx +++ b/plugins/agent/conversations/ui/Conversation.tsx @@ -85,6 +85,7 @@ import { } from '../../../../products/maka/workbench/app/ui'; import { acceptedSend, + activityMessage, sameHistoryRow, compareHistoryOrder, historyTurnKey, @@ -247,7 +248,9 @@ type HistoryNavigation = type HistoryWindow = Pick & { through: number; identity: object; + view: HistoryListView; }; +type HistoryListView = 'chat' | 'conversation'; export type Control = (method: string, input: any) => Promise; export type ConversationHandle = Omit< @@ -545,6 +548,14 @@ export function Conversation(props: ConversationProps) { !historyFollowPaused() && innerReader() !== null && innerReader() === readerAtEnd(); const metrics = signal(null); const turnStates = new Map(); + const workPages = new Map< + string, + { + through: number; + after: AgentHistoryPage['after']; + task?: Promise; + } + >(); const groupStates = new Map(); const rowStates = new Map(); const turnState = (id: string) => { @@ -1095,6 +1106,7 @@ export function Conversation(props: ConversationProps) { if ( turn && (entry.kind === 'process' || + (row?.role === 'compaction' && turn.foldedWork) || row?.role === 'assistant' || // Recovered failures sit under the summary that folds them. (row?.role === 'failure' && @@ -1408,7 +1420,15 @@ export function Conversation(props: ConversationProps) { reading.set('trajectory'); }); if (!trajectorySideBySide()) pendingTrajectoryAnchor = key; - jumpTrajectoryEntry(key); + const window = historyWindow(); + const position = currentRow(rowId)?.position; + if (window?.view === 'chat' && position) { + void resetHistory({ kind: 'row', position, field: 'body' }, window.through) + .then((opened) => { + if (opened && reading() === 'trajectory') jumpTrajectoryEntry(key); + }) + .catch(report); + } else jumpTrajectoryEntry(key); }; const resumeHistoryFollowing = (animated = false) => { @@ -1553,13 +1573,21 @@ export function Conversation(props: ConversationProps) { } if (!current.missingAttachments.size) attachmentMetadataFailed.set(false); }; - const history = async (through: number, position: HistoryPosition) => { + const chatHistoryView = (): HistoryListView => + platform.idiom !== 'terminal' && reading() === 'chat' && botView() !== 'bot' + ? 'chat' + : 'conversation'; + const history = async ( + through: number, + position: HistoryPosition, + view: HistoryListView | 'turn' = 'conversation', + ) => { const page = await host.query('agent', 'history', { sessionId, through, position, includeThinking: true, - view: 'conversation', + view, }); if (page.through !== through) throw new UiError('History changed its requested fence.'); return { ...page, through: page.through }; @@ -1572,6 +1600,7 @@ export function Conversation(props: ConversationProps) { through: number, verify?: (page: AgentHistoryPage) => void, current: () => boolean = () => true, + view: HistoryListView = chatHistoryView(), ): Promise => { if (!current()) return Promise.resolve(false); const identity = {}; @@ -1603,7 +1632,7 @@ export function Conversation(props: ConversationProps) { let location = position; let page: Awaited>; for (;;) { - page = await history(through, location); + page = await history(through, location, view); if (!canPublish()) return false; verify?.(page); if (page.chunks.length) break; @@ -1615,7 +1644,7 @@ export function Conversation(props: ConversationProps) { if (!canPublish()) return false; if (position.kind === 'tail') { const tail = await readHistoryTail(page, async (cursor) => { - const older = await history(through, { kind: 'before', cursor }); + const older = await history(through, { kind: 'before', cursor }, view); return canPublish() ? older : null; }); if (tail === null || !canPublish()) return false; @@ -1623,8 +1652,10 @@ export function Conversation(props: ConversationProps) { } batch(() => { historyRows.set(new HistoryRows()); + workPages.clear(); historyWindow.set({ identity, + view, through: page.through, before: page.before, after: page.after, @@ -1639,6 +1670,15 @@ export function Conversation(props: ConversationProps) { historyFollowPaused.set(position.kind !== 'tail'); innerReader.set(null); }); + for (const [key, state] of turnStates) { + if ( + state.expanded() === true && + historyRows() + .byTurn.get(key) + ?.some((row) => row.foldedWork) + ) + loadTurnWork(key); + } if (page.after === null && latestKnown > through) void follow().catch(report); return true; })() @@ -1657,6 +1697,60 @@ export function Conversation(props: ConversationProps) { windowTask = task; return task; }; + const loadTurnWork = (key: string) => { + const owner = historyWindow(); + const first = historyRows() + .byTurn.get(key) + ?.find((row) => row.foldedWork && row.position); + if (!owner || !first?.position) return; + const existing = workPages.get(key); + if (existing && (existing.task || existing.after === null)) return; + const state = existing ?? { through: first.metadataThrough ?? owner.through, after: null }; + const position: HistoryPosition = state.after + ? { kind: 'after', cursor: state.after } + : { kind: 'row', position: first.position, field: 'body' }; + workPages.set(key, state); + const load = async () => { + let cursor = position; + for (;;) { + const page = await history(state.through, cursor, 'turn'); + if (!alive || workPages.get(key) !== state) return; + state.after = page.after; + const current = historyRows(); + const changed = mergeHistoryPage( + page.chunks, + (key) => current.byKey.get(key), + false, + page.through, + ); + if (changed.length) { + current.put(changed); + historyRows.set(current); + } + if ( + page.after === null || + page.chunks.some( + (chunk) => + chunk.role !== 'compaction' && + (chunk.role !== 'user' || activityMessage(chunk.origin)) && + !chunk.finalOutput && + (chunk.field !== 'thinking' || chunk.text.trim()), + ) + ) + return; + cursor = { kind: 'after', cursor: page.after }; + } + }; + state.task = load() + .catch((reason) => { + if (workPages.get(key) !== state) return; + if (state.after === null) workPages.delete(key); + report(reason); + }) + .finally(() => { + state.task = undefined; + }); + }; const hydrateAttachmentMetadata = () => { for (const rowKey of historyRows().missingAttachments) { const row = historyRows().byKey.get(rowKey)!; @@ -2126,7 +2220,7 @@ export function Conversation(props: ConversationProps) { const task = (async () => { let position: HistoryPosition = { kind: direction, cursor: edge }; for (;;) { - const page = await history(owner.through, position); + const page = await history(owner.through, position, owner.view); if (!ownsWindow(owner.identity)) return false; const current = historyWindow()!; if (page.chunks.length || page[direction] === null) { @@ -2305,6 +2399,7 @@ export function Conversation(props: ConversationProps) { throw new UiError('The original message could not be verified.'); }, () => navigation() === target && current(), + 'conversation', ); if (!opened || !alive || navigation() !== target || !current()) throw new UiError('History navigation was cancelled.'); @@ -3187,7 +3282,31 @@ export function Conversation(props: ConversationProps) { innerReader.set(null); reading.set(mode); }); - if (historyFollowPaused() && anchor !== null) scrollToTurn(anchor); + const window = historyWindow(); + const desired = chatHistoryView(); + const atTail = atHistoryTail() && !historyFollowPaused(); + const rowsAtAnchor = anchor === null ? undefined : historyRows().byTurn.get(anchor); + const row = + rowsAtAnchor?.find((row) => row.role === 'user' && !activityMessage(row.origin)) ?? + rowsAtAnchor?.find((row) => row.finalOutput) ?? + rowsAtAnchor?.[0]; + if (window && window.view !== desired) { + const position: HistoryPosition = + !atTail && row?.position + ? { + kind: 'row', + position: row.position, + field: row.field === 'thinking' ? 'thinking' : 'body', + } + : { kind: 'tail' }; + void resetHistory(position, window.through, undefined, () => reading() === mode, desired) + .then((opened) => { + if (!opened || reading() !== mode) return; + if (position.kind === 'tail') resumeHistoryFollowing(); + else if (anchor !== null) scrollToTurn(anchor); + }) + .catch(report); + } else if (historyFollowPaused() && anchor !== null) scrollToTurn(anchor); else if (atHistoryTail()) resumeHistoryFollowing(); }; const sendRendered = ({ @@ -3413,7 +3532,12 @@ export function Conversation(props: ConversationProps) { }; // The loaded window already reaches the latest row and follows new ones: // scroll it in place, keeping the rows above, without waiting on the Host. - if (!reload && atHistoryTail() && navigation() === null) { + if ( + !reload && + atHistoryTail() && + navigation() === null && + historyWindow()?.view === chatHistoryView() + ) { props.focusComposer(); showLatest(true); return; @@ -3807,12 +3931,19 @@ export function Conversation(props: ConversationProps) { const first = projectedTurn(initial.turnKey); if (!first) return ; const turn = () => projectedTurn(initial.turnKey) ?? first; - const images = computed(() => - turnImages( - projectedSection(initial.turnKey)?.rowsById.values() ?? [], - initial.turnKey, - ), - ); + const images = computed(() => { + const folded = turn().foldedWork; + return folded + ? folded.images.map(({ position, image }) => ({ + key: `${historyRowKey(position)}/${image ?? 'content'}`, + position, + image: image ?? undefined, + })) + : turnImages( + projectedSection(initial.turnKey)?.rowsById.values() ?? [], + initial.turnKey, + ); + }); const groups = computed(() => (projectedSection(initial.turnKey)?.entries ?? []).filter( (item): item is ProcessGroup => @@ -3831,11 +3962,13 @@ export function Conversation(props: ConversationProps) { sessionId, through: () => latestKnown, refs: images, + total: () => turn().foldedWork?.imageCount, preview: props.previewImage, }} toggle={(open) => { if (open) { historyFollowPaused.set(true); + if (turn().foldedWork) loadTurnWork(initial.turnKey); // Replace the source reading anchor before the folded rows return. list?.scrollToOffset({ offset: historyOffset, animated: false }); } @@ -3897,6 +4030,14 @@ export function Conversation(props: ConversationProps) { if (historyFollowPaused() && latestActivity()?.key === initial.group.key) resumeHistoryAtEnd(historyOffset); }} + nearActivityEnd={() => { + const projected = projectedSection(initial.group.turnKey); + const last = projected?.entries.findLast( + (entry) => entry.kind === 'process' && entry.turnKey === initial.group.turnKey, + ); + if (last?.key === initial.group.key && turn().foldedWork) + loadTurnWork(initial.group.turnKey); + }} settled={(atEnd) => { if (atEnd) readerAtEnd.set(initial.group.key); else if (readerAtEnd() === initial.group.key) readerAtEnd.set(null); diff --git a/plugins/agent/native/src/owner/bot.rs b/plugins/agent/native/src/owner/bot.rs index 78aef656c3..89398e2fae 100644 --- a/plugins/agent/native/src/owner/bot.rs +++ b/plugins/agent/native/src/owner/bot.rs @@ -891,6 +891,10 @@ impl Owner { }) .await } + pub(crate) async fn has_bot_history(&self, session: &str) -> Result { + self.bot.is_or_was(session).await + } + pub async fn bot_outbox(&self, input: &BotOutboxRead) -> Result { self.bot.load().await.map_err(|error| error.to_string())?; let session_id = self.bot.session(); diff --git a/plugins/agent/native/src/service/history.rs b/plugins/agent/native/src/service/history.rs index f41b71e8aa..4b7d14d4aa 100644 --- a/plugins/agent/native/src/service/history.rs +++ b/plugins/agent/native/src/service/history.rs @@ -91,7 +91,7 @@ async fn cut(history: &SessionHistory, through: Option) -> Result Result { wire::Read { session_id: input.session_id.clone(), @@ -102,6 +102,9 @@ pub(super) async fn read( .validate() .map_err(|e| e.to_string())?; let history = open(owner, scope, &input.session_id).await?; + if input.view == wire::HistoryView::Chat && owner.has_bot_history(&input.session_id).await? { + input.view = wire::HistoryView::Conversation; + } let cut = cut(&history, input.through).await?; let page = window(owner, scope, &history, &cut, &input).await?; authorize(owner, scope, &input.session_id).await?; @@ -285,6 +288,7 @@ async fn tree_window( view: wire::HistoryView, stop: &tokio_util::sync::CancellationToken, ) -> Result { + use maka_agent_state::session_history::source_tree::RunRows; use wire::{Position, SourcePosition}; const PAGE_CHUNKS: usize = 8; let mut reverse = matches!(&position, Position::Tail | Position::Before { .. }); @@ -317,8 +321,22 @@ async fn tree_window( _ => None, }; let mut cache = source::Cache::default(); + cache.chat = view == wire::HistoryView::Chat; + let mut turn = None; + if view == wire::HistoryView::Turn && anchor.is_none() { + return Err("agent_history_turn_requires_anchor".into()); + } if let Some(anchor) = &anchor { - let selected = source::selected(tree, &anchor.position, &mut cache, stop).await?; + let selected = source::unpaired(tree, &anchor.position, None, &mut cache, stop).await?; + if view == wire::HistoryView::Turn { + use maka_agent_state::session_history::source_tree::SourceItem; + let (wire::HistoryRowAddress::Live { .. }, SourceItem::Canonical(item)) = + (&anchor.position, &selected.item) + else { + return Err("agent_history_turn_requires_live_run".into()); + }; + turn = Some((item.recorded.run, item.admission_sequence)); + } if !matches!(&position, Position::Event { .. }) && (selected.row.chunk.field != anchor.field || (!include_thinking && anchor.field == wire::HistoryField::Thinking)) @@ -331,6 +349,7 @@ async fn tree_window( reverse = true; } } + let scoped = turn.map(|(run, _)| RunRows::Only { run, start: None }); let rows = source::window( tree, anchor.as_ref().map(|cursor| cursor.position.clone()), @@ -339,6 +358,10 @@ async fn tree_window( // One additional source row covers the inclusive boundary cursor. PAGE_CHUNKS + 1, &mut cache, + turn.map(|(run, admission)| RunRows::Only { + run, + start: matches!(position, Position::Row { .. }).then_some(admission), + }), stop, ) .await?; @@ -446,6 +469,7 @@ async fn tree_window( include_thinking, 2, &mut cache, + scoped, stop, ) .await? @@ -463,6 +487,7 @@ async fn tree_window( include_thinking, 2, &mut cache, + scoped, stop, ) .await? diff --git a/plugins/agent/native/src/service/history/imported.rs b/plugins/agent/native/src/service/history/imported.rs index 8656545870..e01c660894 100644 --- a/plugins/agent/native/src/service/history/imported.rs +++ b/plugins/agent/native/src/service/history/imported.rs @@ -289,6 +289,7 @@ pub(super) async fn row( origin: None, compaction: None, timing: None, + folded_work: None, field: wire::HistoryField::Body, imported_turn: Some(message.turn_ordinal), message_id: message.source_message_id.clone(), diff --git a/plugins/agent/native/src/service/history/projection.rs b/plugins/agent/native/src/service/history/projection.rs index 01981d4ff5..098f012dcc 100644 --- a/plugins/agent/native/src/service/history/projection.rs +++ b/plugins/agent/native/src/service/history/projection.rs @@ -105,6 +105,7 @@ fn base( origin: None, compaction: None, timing: None, + folded_work: None, field: HistoryField::Body, imported_turn: None, message_id: Some(item.recorded.id.to_string()), @@ -224,6 +225,8 @@ pub(super) async fn project( address: &wire::HistoryRowAddress, views: &mut std::collections::BTreeMap, executor: Option<&maka_agent_state::session_history::ExecutorTranscript>, + final_attempt: Option, + terminated: bool, stop: &tokio_util::sync::CancellationToken, ) -> Result, String> { let turn = source.spec.turn_id.as_str(); @@ -333,31 +336,45 @@ pub(super) async fn project( }); row.chunk.message_id = Some(item.recorded.id.to_string()); if let BackendSpec::Model { binding, .. } = &source.spec.backend { - if let std::collections::btree_map::Entry::Vacant(e) = - views.entry(item.recorded.run) + let reported_model_id = if let Some(step) = item.scope.step + && matches!(reader, data::DataSource::Live { .. }) { - e.insert( - reader - .view_at(item.recorded.run) - .await - .map_err(|error| error.to_string())?, - ); - } - let original = views[&item.recorded.run] - .attempts - .get(attempt) - .ok_or("agent_history_original_attempt_missing")?; - let reported_model_id = match original.closed.as_ref() { - Some(closed) - if closed.outcome == maka_agent_facts::accounting::Outcome::Success => + reader + .reported_model(item.recorded.run, step, *attempt) + .await + .map_err(|error| error.to_string())? + } else { + if let std::collections::btree_map::Entry::Vacant(e) = + views.entry(item.recorded.run) { - let protocol: maka_agent_facts::context::ModelClosureProtocol = - serde::Deserialize::deserialize(&closed.protocol).map_err(|error| { - format!("agent_history_model_closure_invalid: {error}") - })?; - protocol.model + e.insert( + reader + .view_at(item.recorded.run) + .await + .map_err(|error| error.to_string())?, + ); } - _ => None, + let original = views[&item.recorded.run] + .attempts + .get(attempt) + .ok_or("agent_history_original_attempt_missing")?; + original + .closed + .as_ref() + .filter(|closed| { + closed.outcome == maka_agent_facts::accounting::Outcome::Success + }) + .map(|closed| { + serde::Deserialize::deserialize(&closed.protocol) + .map( + |protocol: maka_agent_facts::context::ModelClosureProtocol| { + protocol.model + }, + ) + .map_err(|error| error.to_string()) + }) + .transpose()? + .flatten() }; row.chunk.model = Some(wire::HistoryModel::Native { requested_model_id: binding.model.clone(), @@ -461,26 +478,37 @@ pub(super) async fn project( false } }; - if let std::collections::btree_map::Entry::Vacant(e) = views.entry(item.recorded.run) { - let view = reader - .view_at(item.recorded.run) - .await - .map_err(|e| e.to_string())?; - e.insert(view); - } - let view = &views[&item.recorded.run]; - row.chunk.final_output = view.terminal == Some(Terminal::Completed) - && view.current_attempt == Some(*attempt) - && final_output_part; + row.chunk.final_output = final_attempt == Some(*attempt) && final_output_part; if let ModelPart::ToolCall { call } = part { let key = ToolKey { attempt: *attempt, call: call.id.clone(), }; - if let Some(state) = view.tools.values().find(|tool| tool.key == key) + let projected = reader + .projected_tool(item.recorded.run, key.clone()) + .await + .map_err(|error| error.to_string())?; + let state = if let Some(state) = projected.as_ref() { + Some(state) + } else { + if let std::collections::btree_map::Entry::Vacant(e) = + views.entry(item.recorded.run) + { + e.insert( + reader + .view_at(item.recorded.run) + .await + .map_err(|error| error.to_string())?, + ); + } + views[&item.recorded.run] + .tools + .get(&maka_agent_state::fold::tool_key(&key)) + }; + if let Some(state) = state && let Some(call) = row.chunk.tool.as_mut().and_then(|tool| tool.call.as_mut()) { - call.pending = view.terminal.is_none() && state.accepted.is_none(); + call.pending = !terminated && state.accepted.is_none(); call.dispatched = state.admitted; if let Some(display) = state.decision.as_ref().map(ToolDecision::display) { apply_display(call, display); @@ -811,20 +839,36 @@ pub(super) async fn project( result.push(row); } Fact::Cell { event } => { - if let std::collections::btree_map::Entry::Vacant(e) = views.entry(item.recorded.run) { - e.insert( - reader - .view_at(item.recorded.run) - .await - .map_err(|error| error.to_string())?, - ); - } + let cell = if visible(&item.recorded.fact) { + if let Some(projected) = reader + .projected_cell(item.recorded.run, event.cell()) + .await + .map_err(|error| error.to_string())? + { + Some(projected) + } else { + if let std::collections::btree_map::Entry::Vacant(e) = + views.entry(item.recorded.run) + { + e.insert( + reader + .view_at(item.recorded.run) + .await + .map_err(|error| error.to_string())?, + ); + } + views[&item.recorded.run].cells.get(&event.cell()).cloned() + } + } else { + None + }; result.push(cells::project( item, turn, address, event, - &views[&item.recorded.run], + cell.as_ref(), + terminated, )?); } Fact::ContextCompacted { baseline } => { diff --git a/plugins/agent/native/src/service/history/projection/cells.rs b/plugins/agent/native/src/service/history/projection/cells.rs index 4cef801a04..d4144971c1 100644 --- a/plugins/agent/native/src/service/history/projection/cells.rs +++ b/plugins/agent/native/src/service/history/projection/cells.rs @@ -18,9 +18,8 @@ */ use super::*; -use maka_agent_facts::cells::{CellEvent, CellOwner, CellStatus}; +use maka_agent_facts::cells::{CellEvent, CellOwner, CellStatus, CellView}; use maka_agent_facts::tool_call::ToolSettlement; -use maka_agent_state::fold::RunView; use maka_plugin_core::operation::{OperationOutcome, Receipt}; pub(super) fn project( @@ -28,7 +27,8 @@ pub(super) fn project( turn: &str, address: &wire::HistoryRowAddress, event: &CellEvent, - view: &RunView, + cell: Option<&CellView>, + terminated: bool, ) -> Result { let mut row = base(item, turn, 0, Role::ToolResult, address)?; let receipt = if let CellEvent::ChildObserved { @@ -73,10 +73,10 @@ pub(super) fn project( return Ok(row); } }; - let cell = view - .cells - .get(&event.cell()) - .ok_or("agent_history_cell_missing")?; + if !row.visible { + return Ok(row); + } + let cell = cell.ok_or("agent_history_cell_missing")?; let child = cell .children .get(operation) @@ -110,7 +110,7 @@ pub(super) fn project( )); } call.dispatched = child.admitted; - call.pending = view.terminal.is_none() + call.pending = !terminated && matches!(cell.status, CellStatus::Running) && child.local_result.is_none() && child.provider_result.is_none() diff --git a/plugins/agent/native/src/service/history/source.rs b/plugins/agent/native/src/service/history/source.rs index 03de487539..9f8537270c 100644 --- a/plugins/agent/native/src/service/history/source.rs +++ b/plugins/agent/native/src/service/history/source.rs @@ -29,7 +29,7 @@ use wire::{SourceField, SourceFormat, SourcePosition, SourceSegment}; use maka_agent_state::session_history::{ snapshot::Snapshot, - source_tree::{SourceItem, SourceTree}, + source_tree::{CompletedWork, RunRows, SourceItem, SourceTree}, }; pub(super) struct Selected { @@ -46,9 +46,13 @@ pub(super) struct Cache { )>, source: Option, timing: Option, + final_attempt: Option, executor: Option, views: std::collections::BTreeMap, tools: std::collections::BTreeMap>, + completed: std::collections::BTreeMap>, + folded_work: std::collections::BTreeMap, + pub chat: bool, } pub(super) fn snapshot_id( address: &wire::HistoryRowAddress, @@ -59,7 +63,39 @@ pub(super) fn snapshot_id( | wire::HistoryRowAddress::Imported { snapshot, .. } => Some(*snapshot), } } -async fn unpaired( +async fn completed_work( + tree: &SourceTree, + address: &wire::HistoryRowAddress, + run: RunId, + cache: &mut Cache, +) -> Result, String> { + if !cache.chat || !matches!(address, wire::HistoryRowAddress::Live { .. }) { + return Ok(None); + } + if let std::collections::btree_map::Entry::Vacant(entry) = cache.completed.entry(run) { + let source = tree + .source_for_path(address.path(), None) + .await + .map_err(|error| error.to_string())?; + let original = source + .run_source(run) + .await + .map_err(|error| error.to_string())?; + entry.insert( + if matches!(original.spec.backend, BackendSpec::Model { .. }) { + source + .completed_work(run, &original.spec.turn_id) + .await + .map_err(|error| error.to_string())? + } else { + None + }, + ); + } + Ok(cache.completed[&run].clone()) +} + +pub(super) async fn unpaired( tree: &SourceTree, address: &wire::HistoryRowAddress, order: Option, @@ -91,6 +127,7 @@ async fn unpaired( }; let mut row = match &item { SourceItem::Canonical(item) => { + let completed = completed_work(tree, address, item.recorded.run, cache).await?; let key = (snapshot_id(address), item.recorded.run); if cache.key.as_ref() != Some(&key) { let source = data @@ -119,6 +156,20 @@ async fn unpaired( cache.views.insert(source.run, view); cache.executor = Some(transcript); } + cache.final_attempt = if let Some(work) = completed.as_ref() { + Some(work.attempt) + } else if timing + .finished + .as_ref() + .is_some_and(|finished| finished.outcome == wire::TurnOutcome::Completed) + && matches!(&source.spec.backend, BackendSpec::Model { .. }) + { + data.final_attempt(source.run) + .await + .map_err(|error| error.to_string())? + } else { + None + }; cache.timing = Some(timing); cache.source = Some(source); cache.key = Some(key); @@ -130,6 +181,11 @@ async fn unpaired( address, &mut cache.views, cache.executor.as_ref(), + cache.final_attempt, + cache + .timing + .as_ref() + .is_some_and(|timing| timing.finished.is_some()), stop, ) .await? @@ -137,6 +193,49 @@ async fn unpaired( .find(|row| row.chunk.position == *address) .ok_or("agent_history_source_missing")?; row.chunk.timing = cache.timing.clone(); + if let Some(work) = completed { + let run = item.recorded.run; + if let std::collections::btree_map::Entry::Vacant(entry) = + cache.folded_work.entry(run) + { + let mut images = Vec::new(); + for position in work.image_sources { + let SourceItem::Canonical(image) = tree + .item(&position) + .await + .map_err(|error| error.to_string())? + else { + continue; + }; + if let Some((content, count, _)) = + stream::media_metadata_from(&data, &image, stop).await? + { + if content { + images.push(wire::WorkImage { + position: position.clone(), + image: None, + }); + } + images.extend((count.saturating_sub(4)..count).map(|index| { + wire::WorkImage { + position: position.clone(), + image: Some(index), + } + })); + } + } + let retained = images.len().saturating_sub(4); + images.drain(..retained); + entry.insert(wire::FoldedWork { + thoughts: work.thoughts, + messages: work.messages, + tools: work.tools, + image_count: work.image_count, + images, + }); + } + row.chunk.folded_work = Some(cache.folded_work[&run].clone()); + } row } SourceItem::Imported { @@ -282,26 +381,54 @@ pub(super) async fn window( include_thinking: bool, limit: usize, cache: &mut Cache, + turn: Option, stop: &tokio_util::sync::CancellationToken, ) -> Result, String> { let mut bound = anchor; let mut exclude_anchor = false; let mut rows = Vec::new(); - loop { + let mut folded = None; + let mut first = true; + 'pages: loop { + let filter = match turn { + Some(RunRows::Only { run, start }) => Some(RunRows::Only { + run, + start: if first { start } else { None }, + }), + Some(selection) => Some(selection), + None => folded.map(|(run, attempt)| RunRows::Folded { run, attempt }), + }; let positions = tree - .positions(bound.clone(), reverse, include_thinking, 32) + .positions_filtered(bound.clone(), reverse, include_thinking, 32, filter) .await .map_err(|error| error.to_string())?; + first = false; let full = positions.len() == 32; let last = positions.last().map(|row| row.address.clone()); for qualified in positions { if exclude_anchor && Some(&qualified.address) == bound.as_ref() { continue; } + let work = match qualified.run { + Some(run) => completed_work(tree, &qualified.address, run, cache) + .await? + .map(|work| (run, work.attempt)), + None => None, + }; + if work.is_some() && work != folded { + // Re-seek in the index before loading any of this Run's activity. + bound = Some(qualified.address); + exclude_anchor = false; + folded = work; + continue 'pages; + } let row = unpaired(tree, &qualified.address, Some(qualified.order), cache, stop).await?; if !row.row.visible || (!include_thinking && row.row.chunk.field == wire::HistoryField::Thinking) + || (work.is_some() + && row.row.chunk.role == wire::Role::Assistant + && !row.row.chunk.final_output) { continue; } @@ -433,7 +560,7 @@ pub(super) async fn display( stop: &tokio_util::sync::CancellationToken, ) -> Result<(SourceFormat, u64, u64, String), String> { let row = &selected.row; - if view == wire::HistoryView::Conversation && row.chunk.role == wire::Role::Compaction { + if view != wire::HistoryView::Text && row.chunk.role == wire::Role::Compaction { return excerpt_ref(&Field::Text(TextBody::default()), None, position, limit); } let field = if view == wire::HistoryView::Text diff --git a/plugins/agent/ports/src/session/history.rs b/plugins/agent/ports/src/session/history.rs index 33f774719e..833a5c09d8 100644 --- a/plugins/agent/ports/src/session/history.rs +++ b/plugins/agent/ports/src/session/history.rs @@ -129,6 +129,10 @@ pub enum HistoryView { #[default] Text, Conversation, + /// Visible chat rows. Completed work can be read separately at the same fence. + Chat, + /// All conversation rows belonging to the live history Run at the anchor. + Turn, } /// A readable field in a conversation window, in presentation order. @@ -475,6 +479,8 @@ pub struct Chunk { pub compaction: Option, #[serde(default, skip_serializing_if = "Option::is_none")] pub timing: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub folded_work: Option, pub field: HistoryField, pub message_id: Option, pub run_id: Option, @@ -491,6 +497,27 @@ pub struct Chunk { pub attachments: Vec, } +/// Completed activity available on demand at the same history fence. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +#[schemars(rename = "HistoryFoldedWork")] +pub struct FoldedWork { + pub thoughts: u64, + pub tools: u64, + pub messages: u64, + pub image_count: u64, + pub images: Vec, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +#[schemars(rename = "HistoryWorkImage")] +pub struct WorkImage { + pub position: HistoryRowAddress, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub image: Option, +} + /// Duration of a successfully adopted context checkpoint, in recorded milliseconds. #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] #[serde(rename_all = "camelCase", deny_unknown_fields)] diff --git a/plugins/agent/state/src/session_history/source_tree.rs b/plugins/agent/state/src/session_history/source_tree.rs index 74545e9258..565dd0a1e3 100644 --- a/plugins/agent/state/src/session_history/source_tree.rs +++ b/plugins/agent/state/src/session_history/source_tree.rs @@ -41,8 +41,27 @@ pub enum HistoryDataSource { }, Snapshot(Snapshot), } +/// An optional constraint on a canonical Run's contribution to a history window. +#[derive(Clone, Copy)] +pub enum RunRows { + /// Fetch only rows of the named Run within the anchor's source. + Only { run: RunId, start: Option }, + /// Retain public input, output from the final attempt and checkpoint markers. + Folded { run: RunId, attempt: ModelAttemptId }, +} +#[derive(Clone)] +pub struct CompletedWork { + pub attempt: ModelAttemptId, + pub thoughts: u64, + pub messages: u64, + pub tools: u64, + pub image_sources: Vec, + pub image_count: u64, +} + #[derive(Clone, Debug)] pub struct QualifiedRow { + pub run: Option, pub address: HistoryRowAddress, pub order: HistoryOrder, } @@ -217,7 +236,14 @@ impl SourceTree { let section = &self.segments[index]; let row = section .source - .positions(Some(coordinate), false, true, section.after_admission, 1) + .positions( + Some(coordinate), + false, + true, + section.after_admission, + 1, + None, + ) .await? .into_iter() .next() @@ -352,6 +378,7 @@ impl SourceTree { include_internal, section.after_admission, 1, + None, ) .await? } @@ -388,6 +415,18 @@ impl SourceTree { reverse: bool, include_internal: bool, limit: u32, + ) -> Result, StoreError> { + self.positions_filtered(anchor, reverse, include_internal, limit, None) + .await + } + + pub async fn positions_filtered( + &self, + anchor: Option, + reverse: bool, + include_internal: bool, + limit: u32, + run_rows: Option, ) -> Result, StoreError> { if limit == 0 { return Ok(vec![]); @@ -397,12 +436,22 @@ impl SourceTree { let (path, snapshot, coordinate) = location(address); let index = self.section(path, snapshot)?; self.validate_address(address).await?; - (index, Some(coordinate)) + ( + index, + match run_rows { + Some(RunRows::Only { + start: Some(sequence), + .. + }) => Some((sequence, 0)), + _ => Some(coordinate), + }, + ) } else if reverse { (self.segments.len() - 1, None) } else { (0, None) }; + let anchored_section = section; let mut output = Vec::new(); loop { let current = &self.segments[section]; @@ -419,6 +468,7 @@ impl SourceTree { include_internal, current.after_admission, batch, + (section == anchored_section).then_some(run_rows).flatten(), ) .await?; let full = rows.len() == batch as usize; @@ -443,6 +493,9 @@ impl SourceTree { bound = Some(last); exclude_anchor = true; } + if matches!(run_rows, Some(RunRows::Only { .. })) { + break; + } bound = None; if reverse { if section == 0 { diff --git a/plugins/agent/state/src/session_history/source_tree/address.rs b/plugins/agent/state/src/session_history/source_tree/address.rs index 7da4f644ec..6647d7f056 100644 --- a/plugins/agent/state/src/session_history/source_tree/address.rs +++ b/plugins/agent/state/src/session_history/source_tree/address.rs @@ -114,6 +114,10 @@ pub(in crate::session_history) fn qualify( } }; Ok(QualifiedRow { + run: match row { + SnapshotPosition::Canonical { run, .. } => Some(*run), + SnapshotPosition::Imported { .. } => None, + }, address, order: HistoryOrder { path: order, diff --git a/plugins/agent/state/src/session_history/source_tree/data_source.rs b/plugins/agent/state/src/session_history/source_tree/data_source.rs index 3e9fe2d326..27ced092b2 100644 --- a/plugins/agent/state/src/session_history/source_tree/data_source.rs +++ b/plugins/agent/state/src/session_history/source_tree/data_source.rs @@ -67,6 +67,109 @@ impl HistoryDataSource { Self::Snapshot(source) => source.terminated(run).await, } } + pub async fn final_attempt(&self, run: RunId) -> Result, StoreError> { + match self { + Self::Live { history, cut } => { + history + .data + .query(queries::FinalAttemptAt( + history.namespace.clone(), + cut.clone(), + run, + )) + .await + } + Self::Snapshot(source) => { + let view = source.view_at(run).await?; + Ok(view + .current_attempt + .filter(|_| view.terminal == Some(Terminal::Completed))) + } + } + } + pub async fn completed_work( + &self, + run: RunId, + turn: &str, + ) -> Result, StoreError> { + match self { + Self::Live { history, cut } => { + history + .data + .query(queries::CompletedWorkAt { + namespace: history.namespace.clone(), + cut: cut.clone(), + run, + turn: turn.to_owned(), + }) + .await + } + Self::Snapshot(_) => Ok(None), + } + } + pub async fn reported_model( + &self, + run: RunId, + step: StepId, + attempt: ModelAttemptId, + ) -> Result, StoreError> { + match self { + Self::Live { history, cut } => { + history + .data + .query(queries::ReportedModelAt( + history.namespace.clone(), + cut.clone(), + run, + step, + attempt, + )) + .await + } + Self::Snapshot(_) => Err(invalid("Native model metadata query requires live history")), + } + } + pub async fn projected_tool( + &self, + run: RunId, + tool: ToolKey, + ) -> Result, StoreError> { + match self { + Self::Live { history, cut } => { + history + .data + .query(queries::ProjectedToolAt( + history.namespace.clone(), + cut.clone(), + run, + tool, + )) + .await + } + Self::Snapshot(_) => Ok(None), + } + } + + pub async fn projected_cell( + &self, + run: RunId, + cell: maka_agent_facts::cells::CellId, + ) -> Result, StoreError> { + match self { + Self::Live { history, cut } => { + history + .data + .query(queries::ProjectedCellAt( + history.namespace.clone(), + cut.clone(), + run, + cell, + )) + .await + } + Self::Snapshot(_) => Ok(None), + } + } pub async fn request(&self, run: RunId, step: StepId) -> Result { match self { Self::Live { history, cut } => { @@ -192,6 +295,7 @@ impl HistoryDataSource { include_internal: bool, min_admission: u64, limit: u32, + run_rows: Option, ) -> Result, StoreError> { match self { Self::Snapshot(source) => { @@ -210,6 +314,7 @@ impl HistoryDataSource { include_internal, min_admission, limit, + run_rows, )) .await } diff --git a/plugins/agent/state/src/session_history/source_tree/queries.rs b/plugins/agent/state/src/session_history/source_tree/queries.rs index e5425c04c8..897dd7e6b9 100644 --- a/plugins/agent/state/src/session_history/source_tree/queries.rs +++ b/plugins/agent/state/src/session_history/source_tree/queries.rs @@ -39,6 +39,7 @@ pub(super) struct LivePositions( pub(super) bool, pub(super) u64, pub(super) u32, + pub(super) Option, ); impl Query for LivePositions { type Output = Vec; @@ -56,7 +57,28 @@ impl Query for LivePositions { (">=", "ASC") }; let visible = if self.4 { "" } else { " AND d.internal=0" }; - let mut statement=c.prepare_cached(&format!("SELECT d.run,d.sequence,h.admission_sequence,h.fact_id,d.ordinal,d.internal FROM agent_recall_messages d JOIN agent_session_history h ON h.owner=d.owner AND h.session=d.session AND h.sequence=d.sequence WHERE d.owner=? AND d.session=? AND d.sequence<=? AND h.admission_sequence>? AND (d.sequence,d.ordinal){comparison}(?,?){visible} ORDER BY d.sequence {order},d.ordinal {order} LIMIT ?"))?; + let (constraint, run, attempt) = match self.7 { + None => ("?8 IS NULL AND ?9 IS NULL", None, None), + Some(RunRows::Only { run, .. }) => ("d.run=?8 AND ?9 IS NULL", Some(run), None), + Some(RunRows::Folded { run, attempt }) => ( + r#"(d.run<>?8 OR ( + d.role='"compaction"' + OR (d.role='"user"' AND coalesce(json_extract(d.origin_json,'$.kind'),'') + NOT IN ('session','session_completion','automation','bot_worker')) + OR (d.role='"assistant"' AND d.internal=0 AND h.attempt=?9) + ))"#, + Some(run), + Some(attempt), + ), + }; + let mut statement = c.prepare_cached(&format!( + "SELECT d.run,d.sequence,h.admission_sequence,h.fact_id,d.ordinal,d.internal + FROM agent_recall_messages d + JOIN agent_session_history h ON h.owner=d.owner AND h.session=d.session AND h.sequence=d.sequence + WHERE d.owner=?1 AND d.session=?2 AND d.sequence<=?3 AND h.admission_sequence>?4 + AND (d.sequence,d.ordinal){comparison}(?5,?6){visible} AND {constraint} + ORDER BY d.sequence {order},d.ordinal {order} LIMIT ?7" + ))?; let rows = statement .query_map( params![ @@ -66,7 +88,9 @@ impl Query for LivePositions { sql::bind_u64(self.5)?, sql::bind_u64(bound.0)?, sql::bind_u64(bound.1)?, - i64::from(self.6) + i64::from(self.6), + run.map(|run| run.to_string()), + attempt.map(|attempt| attempt.to_string()), ], |r| { Ok(( @@ -194,3 +218,231 @@ impl Query for RunAt { Ok(source) } } + +pub(super) struct FinalAttemptAt( + pub(super) HistoryNamespace, + pub(super) HistoryCut, + pub(super) RunId, +); +impl Query for FinalAttemptAt { + type Output = Option; + fn execute(self, c: &Connection) -> Result { + let attempt: Option = c + .query_row( + r#"SELECT r.current_attempt + FROM agent_runs r + JOIN agent_session_history h + ON h.owner=r.owner AND h.session=r.session AND h.run=r.run + AND h.step IS NULL AND h.kind='run_terminated' + WHERE r.run=?1 AND r.owner=?2 AND r.session=?3 AND h.sequence<=?4 + AND r.terminal='"completed"' AND r.current_attempt IS NOT NULL"#, + params![ + self.2.to_string(), + serde_json::to_string(&self.0.owner)?, + self.0.session, + sql::bind_u64(self.1.through_sequence)? + ], + |row| row.get(0), + ) + .optional()?; + attempt.map(sql::identity).transpose() + } +} + +/// Read a settled turn's summary from its indexed history coordinates, without +/// decoding the tool results, receipts, or the Run's accumulated execution state. +pub(super) struct CompletedWorkAt { + pub(super) namespace: HistoryNamespace, + pub(super) cut: HistoryCut, + pub(super) run: RunId, + pub(super) turn: String, +} + +impl Query for CompletedWorkAt { + type Output = Option; + + fn execute(self, c: &Connection) -> Result { + sql::validate_cut(c, &self.namespace.owner, &self.namespace.session, &self.cut)?; + let owner = serde_json::to_string(&self.namespace.owner)?; + let run = self.run.to_string(); + let through = sql::bind_u64(self.cut.through_sequence)?; + let Some(attempt) = + FinalAttemptAt(self.namespace.clone(), self.cut.clone(), self.run).execute(c)? + else { + return Ok(None); + }; + let (thoughts, messages, tools, earlier_output): (u64, u64, u64, u64) = + c.query_row( + r#"WITH visible AS ( + SELECT d.sequence, d.role, d.internal, d.body, + coalesce(json_extract(d.origin_json,'$.kind'),'') IN + ('session','session_completion','automation','bot_worker') AS activity, + h.tool_use_id, h.tool_start, h.attempt + FROM agent_recall_messages d INDEXED BY agent_recall_turn + JOIN agent_session_history h + ON h.owner=d.owner AND h.session=d.session AND h.sequence=d.sequence + WHERE d.owner=?1 AND d.session=?2 AND d.turn=?3 AND d.sequence<=?4 AND d.run=?5 + ) + SELECT count(CASE WHEN internal AND trim(body)<>'' THEN 1 END), + count(CASE WHEN role='"user"' AND activity THEN 1 END), + count(DISTINCT CASE WHEN tool_start THEN tool_use_id END), + count(CASE WHEN role='"assistant"' AND NOT internal AND attempt<>?6 THEN 1 END) + FROM visible + HAVING count(CASE WHEN role='"assistant"' AND NOT internal AND attempt=?6 THEN 1 END)>0"#, + params![owner,self.namespace.session,self.turn,through,run,attempt.to_string()], + |row| Ok((sql::read_u64(row, 0)?, sql::read_u64(row, 1)?, sql::read_u64(row, 2)?, sql::read_u64(row, 3)?)), + ).optional()?.unwrap_or_default(); + if earlier_output + thoughts + tools + messages == 0 { + return Ok(None); + } + let image_count = c.query_row( + "SELECT count(DISTINCT b.artifact) + FROM agent_artifact_run_bindings b + JOIN agent_artifacts a ON a.session=b.session AND a.artifact=b.artifact + JOIN agent_session_history h ON h.fact_id=b.fact + WHERE b.run=?1 AND b.session=?2 AND h.sequence<=?3 AND b.origin IN ('tool','cell') + AND json_extract(a.record_json,'$.artifact.mimeType') LIKE 'image/%'", + params![run, self.namespace.session, through], + |row| sql::read_u64(row, 0), + )?; + let mut images = c.prepare_cached( + "SELECT DISTINCT h.sequence,h.fact_id + FROM agent_artifact_run_bindings b + JOIN agent_artifacts a ON a.session=b.session AND a.artifact=b.artifact + JOIN agent_session_history h ON h.fact_id=b.fact + WHERE b.run=?1 AND b.session=?2 AND h.sequence<=?3 AND b.origin IN ('tool','cell') + AND json_extract(a.record_json,'$.artifact.mimeType') LIKE 'image/%' + ORDER BY h.sequence DESC LIMIT 4", + )?; + let image_sources = images + .query_map(params![run, self.namespace.session, through], |row| { + Ok((sql::read_u64(row, 0)?, row.get::<_, String>(1)?)) + })? + .collect::, _>>()? + .into_iter() + .rev() + .map(|(sequence, fact)| { + Ok(HistoryRowAddress::Live { + sequence, + fact: sql::identity(fact)?, + item_index: 0, + }) + }) + .collect::, StoreError>>()?; + Ok(Some(CompletedWork { + attempt, + thoughts, + messages, + tools, + image_sources, + image_count, + })) + } +} + +pub(super) struct ReportedModelAt( + pub(super) HistoryNamespace, + pub(super) HistoryCut, + pub(super) RunId, + pub(super) StepId, + pub(super) ModelAttemptId, +); +impl Query for ReportedModelAt { + type Output = Option; + fn execute(self, c: &Connection) -> Result { + sql::validate_cut(c, &self.0.owner, &self.0.session, &self.1)?; + let row: Option<(u32, Vec, String)> = c + .query_row( + "SELECT f.schema_version,b.payload,b.content_sha256 + FROM agent_session_history h + JOIN agent_facts f ON f.sequence=h.sequence + JOIN contents b ON b.id=f.body_content_id + WHERE h.owner=?1 AND h.session=?2 AND h.run=?3 AND h.step=?4 + AND h.kind='model_closed' AND h.attempt=?5 AND h.sequence<=?6", + params![ + serde_json::to_string(&self.0.owner)?, + self.0.session, + self.2.to_string(), + self.3.to_string(), + self.4.to_string(), + sql::bind_u64(self.1.through_sequence)? + ], + |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)), + ) + .optional()?; + let Some((version, bytes, digest)) = row else { + return Ok(None); + }; + sql::check_bytes(&bytes, &digest)?; + let Fact::ModelClosed { + outcome, protocol, .. + } = sql::decode_record(version, &bytes)?.fact + else { + return Err(invalid("Model closure index names another fact")); + }; + if outcome != maka_agent_facts::accounting::Outcome::Success { + return Ok(None); + } + let protocol: maka_agent_facts::context::ModelClosureProtocol = + serde::Deserialize::deserialize(&protocol)?; + Ok(protocol.model) + } +} + +pub(super) struct ProjectedToolAt( + pub(super) HistoryNamespace, + pub(super) HistoryCut, + pub(super) RunId, + pub(super) ToolKey, +); +impl Query for ProjectedToolAt { + type Output = Option; + fn execute(self, c: &Connection) -> Result { + let row: Option = c + .query_row( + "SELECT t.body FROM agent_tools t JOIN agent_runs r ON r.run=t.run + JOIN agent_facts f ON f.run=r.run AND f.revision=r.revision + WHERE t.run=?1 AND t.key=?2 AND r.owner=?3 AND r.session=?4 AND f.sequence<=?5", + params![ + self.2.to_string(), + crate::fold::tool_key(&self.3), + serde_json::to_string(&self.0.owner)?, + self.0.session, + sql::bind_u64(self.1.through_sequence)? + ], + |row| row.get(0), + ) + .optional()?; + row.map(|body| serde_json::from_str(&body).map_err(StoreError::from)) + .transpose() + } +} + +pub(super) struct ProjectedCellAt( + pub(super) HistoryNamespace, + pub(super) HistoryCut, + pub(super) RunId, + pub(super) maka_agent_facts::cells::CellId, +); +impl Query for ProjectedCellAt { + type Output = Option; + fn execute(self, c: &Connection) -> Result { + let row: Option = c + .query_row( + "SELECT t.body FROM agent_cells t JOIN agent_runs r ON r.run=t.run + JOIN agent_facts f ON f.run=r.run AND f.revision=r.revision + WHERE t.run=?1 AND t.cell=?2 AND r.owner=?3 AND r.session=?4 AND f.sequence<=?5", + params![ + self.2.to_string(), + self.3.to_string(), + serde_json::to_string(&self.0.owner)?, + self.0.session, + sql::bind_u64(self.1.through_sequence)? + ], + |row| row.get(0), + ) + .optional()?; + row.map(|body| serde_json::from_str(&body).map_err(StoreError::from)) + .transpose() + } +} diff --git a/products/maka/terminal-shell/tests/real_host/desktop_new_session.rs b/products/maka/terminal-shell/tests/real_host/desktop_new_session.rs index 8ecb31bec9..aaad181a0c 100644 --- a/products/maka/terminal-shell/tests/real_host/desktop_new_session.rs +++ b/products/maka/terminal-shell/tests/real_host/desktop_new_session.rs @@ -2142,7 +2142,47 @@ async fn desktop_folds_work_when_responses_starts_the_final_answer() { assert!(desktop.controls(thought).is_empty()); desktop.press("Turn work details").await; desktop.control(thought).await; + + // Reopen to exercise cold Chat history. Live previews already knew the + // thought, so expanding before reconnect alone cannot verify lazy paging. + let session = desktop + .sessions() + .await + .into_iter() + .find(|session| session["id"] != fixture.session) + .unwrap()["id"] + .as_str() + .unwrap() + .to_owned(); + desktop.close().await; + let client = fixture.client().await; + let page = client + .query( + "agent", + "history", + json!({ + "sessionId": session, + "position": {"kind": "tail"}, + "includeThinking": true, + "view": "chat", + }), + ) + .await + .unwrap(); + let rows = page["chunks"].as_array().unwrap(); + assert!(rows.iter().any(|row| row["foldedWork"]["thoughts"] == 1)); + assert!(!rows.iter().any(|row| row["field"] == "thinking")); + assert!(rows.iter().any(|row| row["text"] == "The log is clean.")); + let original = std::mem::replace(&mut fixture.session, session); + let desktop = Desktop::open(&fixture, &fixture.address).await; + desktop.control("Copy message").await; + assert!(desktop.controls(thought).is_empty()); + desktop.artifact("reopened-history-collapsed"); + desktop.press("Turn work details").await; + desktop.control(thought).await; + desktop.artifact("reopened-history-expanded"); desktop.close().await; + fixture.session = original; fixture.finish().await; } diff --git a/products/maka/workbench/app/ToolImage.tsx b/products/maka/workbench/app/ToolImage.tsx index 9b15411e71..c3f21bb04d 100644 --- a/products/maka/workbench/app/ToolImage.tsx +++ b/products/maka/workbench/app/ToolImage.tsx @@ -107,12 +107,13 @@ export function ToolImageStrip(props: { sessionId: string; through: () => number; images: () => ToolImageRef[]; + total?: () => number | undefined; preview(image: ImagePreview): void; }) { // Terminal cells are too coarse for thumbnails; expanded tools still show each image. if (platform.idiom === 'terminal') return ; const shown = () => props.images().slice(-stripLimit); - const hidden = () => props.images().length - shown().length; + const hidden = () => (props.total?.() ?? props.images().length) - shown().length; return ( 0}> diff --git a/products/maka/workbench/app/TurnPanel.tsx b/products/maka/workbench/app/TurnPanel.tsx index 90c007d96e..72d5980b55 100644 --- a/products/maka/workbench/app/TurnPanel.tsx +++ b/products/maka/workbench/app/TurnPanel.tsx @@ -98,6 +98,14 @@ export function activityCounts( const value = row(id); return value?.role === 'user' && activityMessage(value.origin); }).length; + return activityCountLabel({ thoughts, tools, messages }); +} + +function activityCountLabel({ + thoughts, + tools, + messages, +}: Pick, 'thoughts' | 'tools' | 'messages'>) { return [ thoughts ? tr(thoughts === 1 ? '{count} thought' : '{count} thoughts', { count: thoughts }) @@ -126,6 +134,7 @@ export function TurnSummary(props: { sessionId: string; through: () => number; refs: () => ToolImageRef[]; + total?: () => number | undefined; preview(image: ImagePreview): void; }; }) { @@ -165,7 +174,9 @@ export function TurnSummary(props: { : ''; const counts = platform.idiom !== 'terminal' && props.canFold() - ? activityCounts(props.groups(), props.row) + ? turn.foldedWork + ? activityCountLabel(turn.foldedWork) + : activityCounts(props.groups(), props.row) : ''; const usage = turn.finalOutputIds.length === 0 ? turn.timing?.executorUsage : null; return ( @@ -194,6 +205,7 @@ export function TurnSummary(props: { sessionId={props.images!.sessionId} through={props.images!.through} images={props.images!.refs} + total={props.images!.total} preview={props.images!.preview} /> @@ -286,6 +298,7 @@ function ActivityLog(props: { resumed(): void; /** Whether the activity list is scrolled to its end, following or not. */ settled?(atEnd: boolean): void; + nearActivityEnd?(): void; reveal(): void; }) { const state = props.state; @@ -421,6 +434,7 @@ function ActivityLog(props: { nearEnd = distance <= activityFollowThreshold; atEnd = distance <= 1; props.settled?.(atEnd); + if (distance < event.size.height) props.nearActivityEnd?.(); // Reflow and restoring a saved offset also report scroll geometry. if (movement === 0 || scrollInput === 'idle') return; if (scrollInput === 'pending') scrollInput = 'idle'; @@ -578,6 +592,7 @@ type TurnPanelProps = { reading(away: boolean): void; resumed(): void; settled?(atEnd: boolean): void; + nearActivityEnd?(): void; }; export function TurnPanel(props: TurnPanelProps) { diff --git a/products/maka/workbench/app/history.ts b/products/maka/workbench/app/history.ts index 1665115209..2a491d9779 100644 --- a/products/maka/workbench/app/history.ts +++ b/products/maka/workbench/app/history.ts @@ -148,6 +148,7 @@ export type Row = { finalOutput?: boolean; finalAnswer?: boolean; compaction?: HistoryCompaction; + foldedWork?: AgentHistoryChunk['foldedWork']; metadataThrough?: number | null; runId?: RunId; stepId?: StepId; @@ -312,6 +313,7 @@ export function historyMetadata( finalOutput: chunk.finalOutput, finalAnswer: chunk.finalAnswer, compaction: chunk.compaction ?? undefined, + foldedWork: chunk.foldedWork ?? previous?.foldedWork, tool: chunk.tool, metadataThrough: through, }; diff --git a/products/maka/workbench/app/processGroups.ts b/products/maka/workbench/app/processGroups.ts index e5e1db90b2..2162f2397b 100644 --- a/products/maka/workbench/app/processGroups.ts +++ b/products/maka/workbench/app/processGroups.ts @@ -40,6 +40,7 @@ export type ProcessRow = Pick< | 'turnId' | 'importedTurn' | 'timing' + | 'foldedWork' | 'finalOutput' | 'finalAnswer' | 'origin' @@ -79,6 +80,7 @@ export type ProcessEntry = ProcessGroup | MessageEntry; export type TurnPresentation = { turnKey: string; timing: Row['timing'] | null; + foldedWork?: Row['foldedWork']; active: boolean; finalOutputIds: string[]; /** Trailing assistant text until more work arrives or history identifies the final output. */ @@ -103,6 +105,7 @@ export function turnWorkFold( terminal: boolean, ): 'unavailable' | 'open' | 'closed' { const outcome = turnOutcome(turn); + if (turn.foldedWork) return 'closed'; if (!turn.worked || turn.interleavedInput || (outcome !== null && outcome !== 'completed')) return 'unavailable'; if (turn.finalAnswerStarted) return 'closed'; @@ -293,6 +296,10 @@ export function projectProcessGroups(input: { const rowTurn = historyTurnKey(row); if (rowTurn === undefined) continue; const presentation = turn(rowTurn); + if (row.foldedWork) { + presentation.foldedWork = row.foldedWork; + presentation.worked = true; + } const through = row.metadataThrough ?? null; const previousThrough = timingThrough.get(rowTurn); if (row.timing && (through === null || previousThrough == null || through >= previousThrough)) {