feat(transcript): derived transcript view — append-only session log as source of truth (#5086)

This commit is contained in:
Steven Enamakel
2026-07-21 16:03:01 +03:00
committed by GitHub
parent 49a0d3b1d5
commit 00846fdfd0
35 changed files with 4698 additions and 207 deletions
@@ -103,7 +103,7 @@ import {
clearRuntimeForThread,
clearThreadSendPending,
enqueueFollowup,
fetchAndHydrateTurnHistory,
fetchAndHydrateDerivedTranscript,
fetchAndHydrateTurnState,
hydrateThreadUsage,
markSubagentCancelled,
@@ -776,8 +776,12 @@ const Conversations = ({
if (selectedThreadId) {
void dispatch(loadThreadMessages(selectedThreadId));
void dispatch(fetchAndHydrateTurnState(selectedThreadId));
// Per-turn history: each past answer's own process trail (Phase 5).
void dispatch(fetchAndHydrateTurnHistory(selectedThreadId));
// Per-turn history: each past answer's own process trail. Phase C derives
// this from the append-only transcript projection
// (`threads_transcript_get`), auto-falling back to the legacy
// `turn_state_history` hydration when the derived path is off, errors, or
// the thread has no persisted transcript (legacy thread).
void dispatch(fetchAndHydrateDerivedTranscript(selectedThreadId));
void threadApi
.getTaskBoard(selectedThreadId)
.then(board => {
@@ -0,0 +1,78 @@
import { render, screen } from '@testing-library/react';
import { Provider } from 'react-redux';
import { describe, expect, it } from 'vitest';
import { store } from '../../../store';
import type { DerivedDisplayItem } from '../../../types/derivedTranscript';
import { PastTurnInsights } from '../components/PastTurnInsights';
import { mapDisplayItems } from './mapDisplayItems';
function renderInStore(ui: React.ReactNode) {
return render(<Provider store={store}>{ui}</Provider>);
}
/** Newest-first page from chronological items (as the RPC returns). */
function newestFirst(chronological: DerivedDisplayItem[]): DerivedDisplayItem[] {
return [...chronological].reverse();
}
describe('derived transcript restore (mapper → PastTurnInsights)', () => {
it('renders a restored turn with reasoning, tool rows, and a sub-agent trail', () => {
// A settled turn's projected display items, exactly as `threads_transcript_get`
// returns them (newest-first). This is a NON-newest turn so the mapper keeps it.
const chronological: DerivedDisplayItem[] = [
{ kind: 'turnBoundary', requestId: 'req-1' },
{ kind: 'userMessage', content: 'research this', requestId: 'req-1' },
{ kind: 'reasoning', text: 'planning the research' },
{ kind: 'assistantMessage', content: 'searching now', interim: true, requestId: 'req-1' },
{
kind: 'toolCall',
callId: 'c1',
name: 'read_file',
args: { path: 'notes.md' },
result: 'ok',
status: 'success',
},
{
kind: 'subagent',
id: 'researcher',
items: [
{ kind: 'reasoning', text: 'child reasoning trail' },
{
kind: 'toolCall',
callId: 'child-1',
name: 'web_search',
args: { q: 'topic' },
result: 'hits',
status: 'success',
},
],
},
{ kind: 'assistantMessage', content: 'the final answer', requestId: 'req-1' },
// A later turn so req-1 is not the newest (which the mapper would skip).
{ kind: 'turnBoundary', requestId: 'req-2' },
{ kind: 'reasoning', text: 'newest turn thought' },
];
const { timelines, transcripts } = mapDisplayItems(newestFirst(chronological));
renderInStore(
<PastTurnInsights entries={timelines['req-1']} transcript={transcripts['req-1']} />
);
// Reasoning replays.
expect(screen.getByTestId('processing-thinking').textContent).toContain(
'planning the research'
);
// Interim narration replays (not the final answer — that renders from the message).
expect(screen.getByTestId('processing-transcript').textContent).toContain('searching now');
expect(screen.getByTestId('processing-transcript').textContent).not.toContain(
'the final answer'
);
// Tool rows render.
expect(screen.getAllByTestId('processing-tool-row').length).toBeGreaterThan(0);
// The sub-agent's own reasoning trail renders beneath.
const subagents = screen.getByTestId('past-turn-subagents');
expect(subagents.textContent).toContain('child reasoning trail');
});
});
@@ -0,0 +1,305 @@
import { describe, expect, it } from 'vitest';
import type { DerivedDisplayItem } from '../../../types/derivedTranscript';
import { mapDisplayItems } from './mapDisplayItems';
/**
* Build a newest-first page (as the RPC returns) from chronological items — the
* mapper is responsible for reversing back to display order.
*/
function newestFirst(chronological: DerivedDisplayItem[]): DerivedDisplayItem[] {
return [...chronological].reverse();
}
describe('mapDisplayItems', () => {
it('projects reasoning + interim narration + tool call for one turn, skipping the final answer', () => {
const chronological: DerivedDisplayItem[] = [
{ kind: 'turnBoundary', requestId: 'req-1' },
{ kind: 'userMessage', content: 'hello', requestId: 'req-1' },
{ kind: 'reasoning', text: 'let me think' },
{ kind: 'assistantMessage', content: 'looking it up', interim: true, requestId: 'req-1' },
{
kind: 'toolCall',
callId: 'call-a',
name: 'shell',
args: { cmd: 'ls' },
result: 'file.txt',
status: 'success',
},
{ kind: 'assistantMessage', content: 'here is the answer', requestId: 'req-1' },
];
const { timelines, transcripts, interrupted } = mapDisplayItems(newestFirst(chronological));
// The final (non-interim) answer and the user text are NOT emitted — they
// render from the thread message list.
expect(interrupted).toEqual([]);
expect(Object.keys(transcripts)).toEqual(['req-1']);
expect(transcripts['req-1']).toEqual([
{ kind: 'thinking', round: 0, seq: 0, text: 'let me think' },
{ kind: 'narration', round: 0, seq: 1, text: 'looking it up' },
{ kind: 'toolCall', round: 0, seq: 2, callId: 'call-a' },
]);
expect(timelines['req-1']).toEqual([
expect.objectContaining({
id: 'call-a',
name: 'shell',
seq: 2,
status: 'success',
argsBuffer: JSON.stringify({ cmd: 'ls' }),
result: 'file.txt',
}),
]);
});
it('preserves chronological (issue) order when reversing a newest-first page across turns', () => {
const chronological: DerivedDisplayItem[] = [
{ kind: 'turnBoundary', requestId: 'req-1' },
{ kind: 'reasoning', text: 'turn one thought' },
{ kind: 'turnBoundary', requestId: 'req-2' },
{ kind: 'reasoning', text: 'turn two thought' },
];
const { transcripts } = mapDisplayItems(newestFirst(chronological));
expect(Object.keys(transcripts).sort()).toEqual(['req-1', 'req-2']);
expect(transcripts['req-1']).toEqual([
{ kind: 'thinking', round: 0, seq: 0, text: 'turn one thought' },
]);
expect(transcripts['req-2']).toEqual([
{ kind: 'thinking', round: 0, seq: 0, text: 'turn two thought' },
]);
});
it('maps a running (unpaired) tool call to a settled cancelled row', () => {
const chronological: DerivedDisplayItem[] = [
{ kind: 'turnBoundary', requestId: 'req-1' },
{ kind: 'toolCall', callId: 'call-x', name: 'shell', status: 'running' },
];
const { timelines } = mapDisplayItems(newestFirst(chronological));
expect(timelines['req-1'][0]).toEqual(
expect.objectContaining({ id: 'call-x', status: 'cancelled' })
);
});
it('maps an error tool call to an error row', () => {
const chronological: DerivedDisplayItem[] = [
{ kind: 'turnBoundary', requestId: 'req-1' },
{ kind: 'toolCall', callId: 'call-e', name: 'shell', status: 'error', result: 'boom' },
];
const { timelines } = mapDisplayItems(newestFirst(chronological));
expect(timelines['req-1'][0]).toEqual(
expect.objectContaining({ id: 'call-e', status: 'error', result: 'boom' })
);
});
it('maps a failed tool call onto a ToolFailureExplanation for ToolFailureLines', () => {
const chronological: DerivedDisplayItem[] = [
{ kind: 'turnBoundary', requestId: 'req-1' },
{
kind: 'toolCall',
callId: 'call-e',
name: 'shell',
status: 'error',
result: 'boom',
failure: { detail: 'exit 1: command not found' },
},
];
const { timelines } = mapDisplayItems(newestFirst(chronological));
const row = timelines['req-1'][0];
expect(row.status).toBe('error');
expect(row.failure).toBeDefined();
// The wire detail becomes the `causePlain` the ToolFailureLines renderer
// shows for an unrecognised failure class.
expect(row.failure?.causePlain).toBe('exit 1: command not found');
expect(typeof row.failure?.class).toBe('string');
expect(typeof row.failure?.nextAction).toBe('string');
});
it('falls back to the tool result as failure cause when no detail was captured', () => {
const chronological: DerivedDisplayItem[] = [
{ kind: 'turnBoundary', requestId: 'req-1' },
{
kind: 'toolCall',
callId: 'call-e',
name: 'shell',
status: 'error',
result: 'raw error text',
failure: {},
},
];
const { timelines } = mapDisplayItems(newestFirst(chronological));
expect(timelines['req-1'][0].failure?.causePlain).toBe('raw error text');
});
it('derives displayName/detail for a tool row (parity with turn_state rows)', () => {
const chronological: DerivedDisplayItem[] = [
{ kind: 'turnBoundary', requestId: 'req-1' },
{
kind: 'toolCall',
callId: 'c1',
name: 'shell',
args: { command: 'ls -la' },
result: 'ok',
status: 'success',
},
];
const { timelines } = mapDisplayItems(newestFirst(chronological));
const row = timelines['req-1'][0];
expect(typeof row.displayName).toBe('string');
expect(row.displayName?.length ?? 0).toBeGreaterThan(0);
});
it('anchors a subagent to its own requestId, not the current turn cursor', () => {
// The subagent item is appended after both turns (as the projection emits
// it) but belongs to req-1 via its core-derived requestId.
const chronological: DerivedDisplayItem[] = [
{ kind: 'turnBoundary', requestId: 'req-1' },
{ kind: 'reasoning', text: 'turn one' },
{ kind: 'turnBoundary', requestId: 'req-2' },
{ kind: 'reasoning', text: 'turn two' },
{
kind: 'subagent',
id: 'coder',
requestId: 'req-1',
items: [{ kind: 'assistantMessage', content: 'sub done', iteration: 1 }],
},
];
const { timelines } = mapDisplayItems(newestFirst(chronological));
expect(timelines['req-1']?.some(e => e.name === 'subagent:coder')).toBe(true);
expect(timelines['req-2']?.some(e => e.name === 'subagent:coder')).toBeFalsy();
});
it('projects a subagent item into a timeline row carrying its activity + transcript', () => {
const chronological: DerivedDisplayItem[] = [
{ kind: 'turnBoundary', requestId: 'req-1' },
{
kind: 'subagent',
id: 'researcher',
items: [
{ kind: 'reasoning', text: 'child thinking' },
{ kind: 'assistantMessage', content: 'child answer', iteration: 1 },
{
kind: 'toolCall',
callId: 'child-call',
name: 'web_search',
args: { q: 'x' },
result: 'hits',
status: 'success',
},
],
},
];
const { timelines } = mapDisplayItems(newestFirst(chronological));
const row = timelines['req-1'][0];
expect(row.name).toBe('subagent:researcher');
expect(row.subagent).toBeDefined();
expect(row.subagent?.agentId).toBe('researcher');
expect(row.subagent?.toolCalls).toEqual([
expect.objectContaining({ callId: 'child-call', toolName: 'web_search', status: 'success' }),
]);
expect(row.subagent?.transcript).toEqual([
{ kind: 'thinking', text: 'child thinking' },
{ kind: 'text', iteration: 1, text: 'child answer' },
expect.objectContaining({ kind: 'tool', callId: 'child-call', toolName: 'web_search' }),
]);
});
it('collects interrupted partials by requestId and never emits them as trail items', () => {
const chronological: DerivedDisplayItem[] = [
{ kind: 'turnBoundary', requestId: 'req-1' },
{ kind: 'interruptedPartial', text: 'half an ans', thinking: 'mid thought' },
];
const { transcripts, timelines, interrupted } = mapDisplayItems(newestFirst(chronological));
expect(interrupted).toEqual([
{ requestId: 'req-1', content: 'half an ans', thinking: 'mid thought' },
]);
expect(transcripts['req-1']).toBeUndefined();
expect(timelines['req-1']).toBeUndefined();
});
it('drops compaction markers (no settled-turn renderer)', () => {
const chronological: DerivedDisplayItem[] = [
{ kind: 'turnBoundary', requestId: 'req-1' },
{ kind: 'compaction', replacedCount: 3, keptCount: 1 },
{ kind: 'reasoning', text: 'after compaction' },
];
const { transcripts } = mapDisplayItems(newestFirst(chronological));
expect(transcripts['req-1']).toEqual([
{ kind: 'thinking', round: 0, seq: 0, text: 'after compaction' },
]);
});
it('does not emit the final assistant or user text as trail items (dedupe vs thread messages)', () => {
const chronological: DerivedDisplayItem[] = [
{ kind: 'turnBoundary', requestId: 'req-1' },
{
kind: 'userMessage',
content: 'a question',
displayContent: 'a question',
requestId: 'req-1',
},
{ kind: 'assistantMessage', content: 'a final answer', requestId: 'req-1' },
];
const { transcripts, timelines } = mapDisplayItems(newestFirst(chronological));
expect(transcripts['req-1']).toBeUndefined();
expect(timelines['req-1']).toBeUndefined();
});
it('omits skipped request ids entirely', () => {
const chronological: DerivedDisplayItem[] = [
{ kind: 'turnBoundary', requestId: 'req-old' },
{ kind: 'reasoning', text: 'old thought' },
{ kind: 'turnBoundary', requestId: 'req-live' },
{ kind: 'reasoning', text: 'live thought' },
];
const { transcripts } = mapDisplayItems(newestFirst(chronological), {
skipRequestIds: new Set(['req-live']),
});
expect(Object.keys(transcripts)).toEqual(['req-old']);
expect(transcripts['req-live']).toBeUndefined();
});
it('carries the assistant iteration onto the turn round for its items', () => {
const chronological: DerivedDisplayItem[] = [
{ kind: 'turnBoundary', requestId: 'req-1' },
{
kind: 'assistantMessage',
content: 'step',
interim: true,
iteration: 2,
requestId: 'req-1',
},
{ kind: 'toolCall', callId: 'c1', name: 'shell', status: 'success' },
];
const { transcripts, timelines } = mapDisplayItems(newestFirst(chronological));
expect(transcripts['req-1'][0]).toEqual(
expect.objectContaining({ kind: 'narration', round: 2 })
);
expect(timelines['req-1'][0].round).toBe(2);
});
});
@@ -0,0 +1,371 @@
/**
* Phase C mapper: project the transcript-derived RPC's newest-first
* {@link DerivedDisplayItem}s onto the **existing** settled-turn renderer
* models, keyed by producing `requestId` — the exact shapes
* `fetchAndHydrateTurnHistory` produces from the legacy `turn_state_history`
* snapshot ring, so `PastTurnInsights` / `ProcessingTranscriptView` /
* `ToolTimelineBlock` / `SubagentActivityBlock` are reused unchanged.
*
* Division of labour (matches how `turnTimelinesByThread` / `PastTurnInsights`
* anchor today):
* - **Final assistant text and user text are NOT emitted here** — they stay
* rendered from the thread message list (`threads_messages_list`). This
* mapper only produces the *process trail* (reasoning, interim narration,
* tool cards, sub-agent trails) that renders above a past answer.
* - Items are grouped per `requestId` (from `turnBoundary` markers and each
* message's own `requestId`) so the caller can anchor a turn's trail to its
* first agent message by `requestId`.
*
* Live streaming is untouched: the caller skips the live/most-recent turn's
* `requestId` so derived data never fights socket-fed `chatRuntimeSlice` state.
*/
import debug from 'debug';
import type {
ProcessingTranscriptItem,
SubagentActivity,
SubagentToolCallEntry,
SubagentTranscriptItem,
ToolFailureExplanation,
ToolTimelineEntry,
ToolTimelineEntryStatus,
} from '../../../store/chatRuntimeSlice';
import type {
DerivedDisplayItem,
DerivedToolCall,
DerivedToolCallStatus,
DerivedToolFailure,
} from '../../../types/derivedTranscript';
import { formatTimelineEntry } from '../../../utils/toolTimelineFormatting';
const log = debug('conversations.derived.mapDisplayItems');
/** A partial assistant answer left behind by an interrupted turn. */
export interface DerivedInterruptedAnswer {
requestId: string;
content: string;
thinking: string;
}
/**
* Per-thread settled-turn process trails derived from the transcript
* projection, ready to feed `setTurnTimelinesForThread` (timelines +
* transcripts) and, for completeness, any interrupted partials found.
*/
export interface MappedTranscript {
/** `requestId -> tool timeline rows` for each settled turn. */
timelines: Record<string, ToolTimelineEntry[]>;
/** `requestId -> processing transcript` (narration / thinking / tool ptr). */
transcripts: Record<string, ProcessingTranscriptItem[]>;
/**
* Interrupted partials found in the derived data, in chronological order.
* Exposed for a future phase; the hydration thunk deliberately leaves the
* live/current-turn interrupted partial owned by the `turn_state` snapshot
* path so it never fights live state.
*/
interrupted: DerivedInterruptedAnswer[];
}
/** Options controlling {@link mapDisplayItems}. */
export interface MapDisplayItemsOptions {
/**
* Request ids to omit from the output entirely — the live/most-recent turn
* (rendered from socket-fed state / the turn_state snapshot) and any turn
* currently streaming. Their trails must not double-render against the live
* anchor.
*/
skipRequestIds?: ReadonlySet<string>;
}
/** Map the Rust `ToolCallStatus` onto the timeline status vocabulary. A settled
* turn whose tool row is still `running` had no result line paired (the turn
* was interrupted before completion) — settle it to `cancelled` (terminal,
* muted, non-pulsing), mirroring `settleOrphanedTimelineEntry`. */
function timelineStatusFromDerived(status: DerivedToolCallStatus): ToolTimelineEntryStatus {
switch (status) {
case 'success':
return 'success';
case 'error':
return 'error';
case 'running':
default:
return 'cancelled';
}
}
/** Same mapping for a sub-agent child tool call. */
function subagentToolStatus(status: DerivedToolCallStatus): ToolTimelineEntryStatus {
return timelineStatusFromDerived(status);
}
/**
* Expand the minimal wire {@link DerivedToolFailure} into the richer
* {@link ToolFailureExplanation} the `ToolFailureLines` renderer consumes,
* matching `turn_state`'s `PersistedToolFailure` shape. The projection only
* records that a call failed plus an optional short reason, so we synthesise an
* unlocalized `Unknown`/`Recoverable` explanation whose `causePlain` carries the
* captured detail (or the tool's error output) — `ToolFailureLines` falls back
* to `causePlain`/`nextAction` for unrecognised classes.
*/
function toFailureExplanation(
failure: DerivedToolFailure | undefined,
result: string | undefined
): ToolFailureExplanation | undefined {
if (!failure) return undefined;
const causePlain = failure.detail?.trim() || result?.trim() || 'The tool reported an error.';
return {
class: 'Unknown',
category: 'Recoverable',
recoverable: true,
causePlain,
nextAction: 'Review the tool output and try again.',
};
}
function stringifyArgs(args: unknown): string | undefined {
if (args === undefined || args === null) return undefined;
if (typeof args === 'string') return args;
try {
return JSON.stringify(args);
} catch {
return undefined;
}
}
/**
* Build a {@link SubagentActivity} from a `subagent` display item's nested
* items. The nested vocabulary (reasoning / assistantMessage / toolCall)
* projects onto the sub-agent transcript (`thinking` / `text` / `tool`) plus a
* flat `toolCalls` list — exactly what `SubagentActivityBlock` reads.
*/
function buildSubagentActivity(id: string, items: DerivedDisplayItem[]): SubagentActivity {
const toolCalls: SubagentToolCallEntry[] = [];
const transcript: SubagentTranscriptItem[] = [];
for (const item of items) {
switch (item.kind) {
case 'reasoning':
if (item.text.trim()) {
transcript.push({ kind: 'thinking', text: item.text });
}
break;
case 'assistantMessage':
// A sub-agent's own answer text is part of its inline trail (there is
// no separate message bubble for a delegated worker).
if (item.content.trim()) {
transcript.push({ kind: 'text', iteration: item.iteration, text: item.content });
}
break;
case 'toolCall': {
const status = subagentToolStatus(item.status);
toolCalls.push({
callId: item.callId,
toolName: item.name,
status,
args: item.args,
result: item.result,
});
transcript.push({
kind: 'tool',
callId: item.callId,
toolName: item.name,
status,
args: item.args,
result: item.result,
});
break;
}
// Nested sub-agents, turn boundaries, interrupted partials and compaction
// markers do not surface inside a sub-agent block — the projection nests
// deeper sub-agents as their own top-level items.
default:
break;
}
}
return { taskId: id, agentId: id, status: 'completed', toolCalls, transcript };
}
/** Mutable per-turn accumulator. */
interface TurnAccumulator {
entries: ToolTimelineEntry[];
transcript: ProcessingTranscriptItem[];
seq: number;
round: number;
}
function ensureTurn(turns: Map<string, TurnAccumulator>, requestId: string): TurnAccumulator {
let turn = turns.get(requestId);
if (!turn) {
turn = { entries: [], transcript: [], seq: 0, round: 0 };
turns.set(requestId, turn);
}
return turn;
}
/**
* Map a newest-first page of derived display items into per-`requestId`
* settled-turn trails. Returns empty maps when nothing anchors to a `requestId`
* (e.g. a legacy thread whose lines carry no `requestId`).
*/
export function mapDisplayItems(
items: DerivedDisplayItem[],
options: MapDisplayItemsOptions = {}
): MappedTranscript {
const skip = options.skipRequestIds ?? new Set<string>();
const turns = new Map<string, TurnAccumulator>();
const interrupted: DerivedInterruptedAnswer[] = [];
// The RPC returns items newest-first; walk chronologically so `seq` reflects
// issue order and turn boundaries advance forward.
const chronological = [...items].reverse();
let currentRequestId: string | undefined;
const skipped = new Set<string>();
for (const item of chronological) {
switch (item.kind) {
case 'turnBoundary':
currentRequestId = item.requestId;
break;
case 'userMessage':
// User text renders from the thread message list; only advance the
// turn cursor so following items anchor to this turn.
if (item.requestId) currentRequestId = item.requestId;
break;
case 'assistantMessage': {
if (item.requestId) currentRequestId = item.requestId;
if (item.iteration !== undefined && currentRequestId && !skip.has(currentRequestId)) {
ensureTurn(turns, currentRequestId).round = item.iteration;
}
// Final (non-interim) answer renders from the thread message; only
// interim narration belongs to the process trail.
if (!item.interim) break;
if (!currentRequestId || skip.has(currentRequestId)) {
if (currentRequestId) skipped.add(currentRequestId);
break;
}
const text = item.content.trim();
if (!text) break;
const turn = ensureTurn(turns, currentRequestId);
turn.transcript.push({
kind: 'narration',
round: turn.round,
seq: turn.seq++,
text: item.content,
});
break;
}
case 'reasoning': {
if (!currentRequestId || skip.has(currentRequestId)) {
if (currentRequestId) skipped.add(currentRequestId);
break;
}
if (!item.text.trim()) break;
const turn = ensureTurn(turns, currentRequestId);
turn.transcript.push({
kind: 'thinking',
round: turn.round,
seq: turn.seq++,
text: item.text,
});
break;
}
case 'toolCall': {
if (!currentRequestId || skip.has(currentRequestId)) {
if (currentRequestId) skipped.add(currentRequestId);
break;
}
const turn = ensureTurn(turns, currentRequestId);
pushToolCall(turn, item);
break;
}
case 'subagent': {
// Anchor to the turn the sub-agent was spawned in (core-derived
// `requestId`), not the current cursor — sub-agent items are appended
// after all root items, so the cursor is the last turn by then.
const anchorRequestId = item.requestId ?? currentRequestId;
if (!anchorRequestId || skip.has(anchorRequestId)) {
if (anchorRequestId) skipped.add(anchorRequestId);
break;
}
const turn = ensureTurn(turns, anchorRequestId);
const activity = buildSubagentActivity(item.id, item.items);
turn.entries.push({
id: `subagent:${item.id}`,
name: `subagent:${item.id}`,
round: turn.round,
seq: turn.seq++,
status: 'success',
subagent: activity,
});
break;
}
case 'interruptedPartial':
if (currentRequestId) {
interrupted.push({
requestId: currentRequestId,
content: item.text,
thinking: item.thinking ?? '',
});
}
break;
// Compaction markers have no settled-turn renderer — drop them.
case 'compaction':
default:
break;
}
}
const timelines: Record<string, ToolTimelineEntry[]> = {};
const transcripts: Record<string, ProcessingTranscriptItem[]> = {};
for (const [requestId, turn] of turns) {
if (turn.entries.length > 0) timelines[requestId] = turn.entries;
if (turn.transcript.length > 0) transcripts[requestId] = turn.transcript;
}
log(
'mapped turns=%d timelines=%d transcripts=%d interrupted=%d skipped=%d',
turns.size,
Object.keys(timelines).length,
Object.keys(transcripts).length,
interrupted.length,
skipped.size
);
return { timelines, transcripts, interrupted };
}
/** Push a tool-call display item as both a timeline entry and a transcript
* pointer sharing one `seq`, so the tool row orders consistently among the
* turn's narration / thinking. */
function pushToolCall(turn: TurnAccumulator, item: DerivedToolCall): void {
const seq = turn.seq++;
const entry: ToolTimelineEntry = {
id: item.callId,
name: item.name,
round: turn.round,
seq,
status: timelineStatusFromDerived(item.status),
argsBuffer: stringifyArgs(item.args),
result: item.result,
};
// A failed tool renders its "why / next" explanation via `ToolFailureLines`.
const failure = toFailureExplanation(item.failure, item.result);
if (failure) entry.failure = failure;
// Derive the human label + detail from tool name + args (the same TS
// formatter the live path runs), so settled rows carry `displayName`/`detail`
// at parity with `turn_state` rows instead of being unlabelled.
const formatted = formatTimelineEntry(entry);
entry.displayName = formatted.title;
if (formatted.detail !== undefined) entry.detail = formatted.detail;
turn.entries.push(entry);
turn.transcript.push({ kind: 'toolCall', round: turn.round, seq, callId: item.callId });
}
+10 -1
View File
@@ -39,6 +39,7 @@ import {
clearProcessingForThread,
clearStreamingAssistantForThread,
endInferenceTurn,
fetchAndHydrateDerivedTranscript,
markInferenceTurnStreaming,
parseToolFailure,
recordChatTurnUsage,
@@ -76,7 +77,7 @@ import {
setActiveThread,
setSelectedThread,
} from '../store/threadSlice';
import { IS_PROD } from '../utils/config';
import { DERIVED_TRANSCRIPT_ENABLED, IS_PROD } from '../utils/config';
const logChatRuntime = debug('openhuman:chat-runtime');
const USER_FACING_AGENT_ERROR_MESSAGE =
@@ -443,6 +444,14 @@ const ChatRuntimeProvider = ({ children }: { children: React.ReactNode }) => {
await flushQueuedFollowups(event.thread_id);
dispatch(endInferenceTurn({ threadId: event.thread_id }));
dispatch(clearThreadInferenceActive(event.thread_id));
// Live-turn seam: the turn just settled and its line was appended to the
// append-only transcript. Invalidate/refresh the thread's derived
// settled-turn trails so the next reopen is fresh. The just-finished turn
// is the newest, so the derived hydration skips it — this never fights the
// live anchor, and it does not touch any socket delta handler.
if (DERIVED_TRANSCRIPT_ENABLED) {
void dispatch(fetchAndHydrateDerivedTranscript(event.thread_id));
}
};
rtLog('subscribe_chat_events', { socket: socketStatus });
+35
View File
@@ -221,4 +221,39 @@ describe('threadApi', () => {
params: { runId: 'sub-1', afterSequence: 0 },
});
});
it('fetches the derived transcript page with pagination controls', async () => {
const pageData = {
threadId: 'thread-1',
items: [{ kind: 'turnBoundary', requestId: 'req-1' }],
total: 1,
hasMore: false,
hasTranscript: true,
};
mockCallCoreRpc.mockResolvedValueOnce({ data: pageData });
const { threadApi } = await import('./threadApi');
const result = await threadApi.getDerivedTranscript('thread-1', { cursor: '10', limit: 50 });
expect(mockCallCoreRpc).toHaveBeenLastCalledWith({
method: 'openhuman.threads_transcript_get',
params: { thread_id: 'thread-1', cursor: '10', limit: 50 },
});
expect(result).toEqual(pageData);
});
it('fetches the derived transcript with default (undefined) pagination when omitted', async () => {
mockCallCoreRpc.mockResolvedValueOnce({
data: { threadId: 'thread-1', items: [], total: 0, hasMore: false, hasTranscript: false },
});
const { threadApi } = await import('./threadApi');
const result = await threadApi.getDerivedTranscript('thread-1');
expect(mockCallCoreRpc).toHaveBeenLastCalledWith({
method: 'openhuman.threads_transcript_get',
params: { thread_id: 'thread-1', cursor: undefined, limit: undefined },
});
expect(result.hasTranscript).toBe(false);
});
});
+25
View File
@@ -1,5 +1,9 @@
import debug from 'debug';
import type {
DerivedTranscriptGetOptions,
DerivedTranscriptPage,
} from '../../types/derivedTranscript';
import type {
PurgeResultData,
Thread,
@@ -271,4 +275,25 @@ export const threadApi = {
});
return unwrapEnvelope(response);
},
/**
* Transcript-derived view (Phase B/C): project the thread's append-only
* `session_raw/*.jsonl` source of truth into typed display items for the
* settled-turn restore path. Newest-first paginated; `cursor` comes from a
* prior page's `nextCursor`, `limit` defaults to 50 (core-clamped to 500).
*
* A thread with no persisted transcript yet returns an empty page with
* `hasTranscript: false` (not an error) — the caller then falls back to the
* legacy `turn_state_history` hydration.
*/
getDerivedTranscript: async (
threadId: string,
options?: DerivedTranscriptGetOptions
): Promise<DerivedTranscriptPage> => {
const response = await callCoreRpc<Envelope<DerivedTranscriptPage>>({
method: 'openhuman.threads_transcript_get',
params: { thread_id: threadId, cursor: options?.cursor, limit: options?.limit },
});
return unwrapEnvelope(response);
},
};
@@ -0,0 +1,141 @@
import { configureStore } from '@reduxjs/toolkit';
import { beforeEach, describe, expect, it, vi } from 'vitest';
import type { DerivedDisplayItem, DerivedTranscriptPage } from '../../types/derivedTranscript';
import reducer, {
fetchAndHydrateDerivedTranscript,
setStreamingAssistantForThread,
} from '../chatRuntimeSlice';
const { mockThreadApi, flag } = vi.hoisted(() => ({
mockThreadApi: { getDerivedTranscript: vi.fn(), getTurnStateHistory: vi.fn() },
flag: { enabled: true },
}));
vi.mock('../../services/api/threadApi', () => ({ threadApi: mockThreadApi }));
vi.mock('../../utils/config', async importOriginal => {
const actual = await importOriginal<typeof import('../../utils/config')>();
return {
...actual,
get DERIVED_TRANSCRIPT_ENABLED() {
return flag.enabled;
},
};
});
function page(items: DerivedDisplayItem[], overrides: Partial<DerivedTranscriptPage> = {}) {
return {
threadId: 'thread-1',
items,
total: items.length,
hasMore: false,
hasTranscript: true,
...overrides,
} satisfies DerivedTranscriptPage;
}
/** Newest-first page from chronological items (as the RPC returns). */
function newestFirst(chronological: DerivedDisplayItem[]): DerivedDisplayItem[] {
return [...chronological].reverse();
}
beforeEach(() => {
flag.enabled = true;
mockThreadApi.getDerivedTranscript.mockReset();
mockThreadApi.getTurnStateHistory.mockReset();
});
describe('fetchAndHydrateDerivedTranscript', () => {
it('hydrates settled-turn trails from the projection, skipping the newest turn', async () => {
const store = configureStore({ reducer });
mockThreadApi.getDerivedTranscript.mockResolvedValueOnce(
page(
newestFirst([
{ kind: 'turnBoundary', requestId: 'req-old' },
{ kind: 'reasoning', text: 'old thought' },
{ kind: 'toolCall', callId: 'c1', name: 'shell', status: 'success' },
{ kind: 'turnBoundary', requestId: 'req-new' },
{ kind: 'reasoning', text: 'newest thought' },
])
)
);
await store.dispatch(fetchAndHydrateDerivedTranscript('thread-1'));
expect(mockThreadApi.getDerivedTranscript).toHaveBeenCalledWith('thread-1', { limit: 500 });
expect(mockThreadApi.getTurnStateHistory).not.toHaveBeenCalled();
const timelines = store.getState().turnTimelinesByThread['thread-1'];
const transcripts = store.getState().turnTranscriptsByThread['thread-1'];
// Newest turn (req-new) is skipped — rendered by the live anchor.
expect(Object.keys(transcripts)).toEqual(['req-old']);
expect(timelines['req-old']).toHaveLength(1);
expect(transcripts['req-new']).toBeUndefined();
expect(timelines['req-new']).toBeUndefined();
});
it('falls back to turn_state history when the flag is off', async () => {
flag.enabled = false;
const store = configureStore({ reducer });
mockThreadApi.getTurnStateHistory.mockResolvedValueOnce([]);
await store.dispatch(fetchAndHydrateDerivedTranscript('thread-1'));
expect(mockThreadApi.getDerivedTranscript).not.toHaveBeenCalled();
expect(mockThreadApi.getTurnStateHistory).toHaveBeenCalledWith('thread-1');
});
it('falls back to turn_state history when the RPC errors', async () => {
const store = configureStore({ reducer });
mockThreadApi.getDerivedTranscript.mockRejectedValueOnce(new Error('boom'));
mockThreadApi.getTurnStateHistory.mockResolvedValueOnce([]);
await expect(
store.dispatch(fetchAndHydrateDerivedTranscript('thread-1'))
).resolves.toBeDefined();
expect(mockThreadApi.getTurnStateHistory).toHaveBeenCalledWith('thread-1');
});
it('falls back to turn_state history when the thread has no persisted transcript (legacy)', async () => {
const store = configureStore({ reducer });
mockThreadApi.getDerivedTranscript.mockResolvedValueOnce(page([], { hasTranscript: false }));
mockThreadApi.getTurnStateHistory.mockResolvedValueOnce([]);
await store.dispatch(fetchAndHydrateDerivedTranscript('thread-1'));
expect(mockThreadApi.getTurnStateHistory).toHaveBeenCalledWith('thread-1');
// The legacy fallback ran with empty history — no derived trails installed.
expect(store.getState().turnTimelinesByThread['thread-1']).toEqual({});
});
it('skips a turn that is currently streaming (live-turn requestId skip)', async () => {
const store = configureStore({ reducer });
// Seed a live stream on a NON-newest turn (req-mid).
store.dispatch(
setStreamingAssistantForThread({
threadId: 'thread-1',
streaming: { requestId: 'req-mid', content: 'streaming...', thinking: '' },
})
);
mockThreadApi.getDerivedTranscript.mockResolvedValueOnce(
page(
newestFirst([
{ kind: 'turnBoundary', requestId: 'req-old' },
{ kind: 'reasoning', text: 'old thought' },
{ kind: 'turnBoundary', requestId: 'req-mid' },
{ kind: 'reasoning', text: 'mid thought' },
{ kind: 'turnBoundary', requestId: 'req-new' },
{ kind: 'reasoning', text: 'new thought' },
])
)
);
await store.dispatch(fetchAndHydrateDerivedTranscript('thread-1'));
const transcripts = store.getState().turnTranscriptsByThread['thread-1'];
// req-new skipped (newest), req-mid skipped (streaming), req-old kept.
expect(Object.keys(transcripts)).toEqual(['req-old']);
expect(transcripts['req-mid']).toBeUndefined();
expect(transcripts['req-new']).toBeUndefined();
});
});
+170
View File
@@ -1,7 +1,9 @@
import { createAsyncThunk, createSlice, type PayloadAction } from '@reduxjs/toolkit';
import debug from 'debug';
import { mapDisplayItems } from '../features/conversations/derived/mapDisplayItems';
import { threadApi } from '../services/api/threadApi';
import type { DerivedDisplayItem, DerivedTranscriptPage } from '../types/derivedTranscript';
import type { ThreadMessage } from '../types/thread';
import type {
AgentRun,
@@ -13,6 +15,7 @@ import type {
PersistedTurnState,
TaskBoard,
} from '../types/turnState';
import { DERIVED_TRANSCRIPT_ENABLED } from '../utils/config';
import {
formatTimelineEntry,
isKnownClientTool,
@@ -2403,4 +2406,171 @@ export const fetchAndHydrateTurnHistory = createAsyncThunk(
}
);
/**
* Initial derived-transcript page size. Sized generously (the core clamps to
* 500) so a reopened thread's visible turns all carry their process trail
* without a second round-trip. Older turns beyond this window load lazily via
* {@link loadOlderDerivedTranscript}.
*/
const DERIVED_TRANSCRIPT_INITIAL_LIMIT = 500;
const derivedLog = debug('chatRuntime.derivedTranscript');
/**
* Read the {@link ChatRuntimeState} out of an arbitrary redux root, tolerating
* both the app store (`state.chatRuntime`) and a bare test store whose root IS
* the slice state. Used only to read live-turn request ids for the skip set.
*/
function readChatRuntimeState(state: unknown): ChatRuntimeState | undefined {
if (!state || typeof state !== 'object') return undefined;
const root = state as Record<string, unknown>;
if ('chatRuntime' in root && root.chatRuntime && typeof root.chatRuntime === 'object') {
return root.chatRuntime as ChatRuntimeState;
}
if ('streamingAssistantByThread' in root) {
return root as unknown as ChatRuntimeState;
}
return undefined;
}
/**
* The request ids whose derived trail must NOT be hydrated: the newest turn
* (rendered as the live "agent insights" anchor from `toolTimelineByThread` /
* the socket stream, or the `turn_state` snapshot via
* {@link fetchAndHydrateTurnState}) and any turn currently streaming. Mirrors
* `fetchAndHydrateTurnHistory`'s `history.slice(1)` newest-turn skip.
*/
function liveRequestIdsToSkip(
state: unknown,
threadId: string,
items: DerivedDisplayItem[]
): Set<string> {
const skip = new Set<string>();
// Newest turn = first request id encountered walking newest-first.
for (const item of items) {
const rid =
item.kind === 'turnBoundary'
? item.requestId
: 'requestId' in item
? item.requestId
: undefined;
if (rid) {
skip.add(rid);
break;
}
}
const runtime = readChatRuntimeState(state);
const streamingRid = runtime?.streamingAssistantByThread[threadId]?.requestId;
if (streamingRid) skip.add(streamingRid);
for (const [rid, mappedThread] of Object.entries(runtime?.parallelRequestThreads ?? {})) {
if (mappedThread === threadId) skip.add(rid);
}
return skip;
}
/**
* Phase C settled-turn restore: hydrate past-turn process trails from the
* transcript-derived projection (`openhuman.threads_transcript_get`) instead of
* the legacy `turn_state_history` snapshot ring. Populates the SAME
* {@link ChatRuntimeState.turnTimelinesByThread} /
* {@link ChatRuntimeState.turnTranscriptsByThread} the legacy path did, so the
* renderers are reused unchanged. The live/most-recent turn is skipped so
* derived data never fights socket-fed live state.
*
* Automatic fallback to {@link fetchAndHydrateTurnHistory} when the flag is
* off, the RPC errors, or the thread has no persisted transcript (legacy
* thread). Failures never block navigation.
*/
export const fetchAndHydrateDerivedTranscript = createAsyncThunk(
'chatRuntime/fetchAndHydrateDerivedTranscript',
async (threadId: string, { dispatch, getState }) => {
if (!DERIVED_TRANSCRIPT_ENABLED) {
derivedLog('disabled thread=%s -> turn_state history', threadId);
await dispatch(fetchAndHydrateTurnHistory(threadId));
return null;
}
let page: DerivedTranscriptPage;
try {
page = await threadApi.getDerivedTranscript(threadId, {
limit: DERIVED_TRANSCRIPT_INITIAL_LIMIT,
});
} catch (error) {
derivedLog('rpc failed thread=%s err=%O -> turn_state history fallback', threadId, error);
await dispatch(fetchAndHydrateTurnHistory(threadId));
return null;
}
if (!page.hasTranscript) {
derivedLog('no transcript thread=%s -> turn_state history fallback (legacy)', threadId);
await dispatch(fetchAndHydrateTurnHistory(threadId));
return null;
}
const skipRequestIds = liveRequestIdsToSkip(getState(), threadId, page.items);
const { timelines, transcripts } = mapDisplayItems(page.items, { skipRequestIds });
derivedLog(
'hydrated thread=%s items=%d timelines=%d transcripts=%d skip=%d hasMore=%s',
threadId,
page.items.length,
Object.keys(timelines).length,
Object.keys(transcripts).length,
skipRequestIds.size,
page.hasMore
);
dispatch(setTurnTimelinesForThread({ threadId, timelines, transcripts }));
// TODO(pagination): when `page.hasMore`, an insights "load older" affordance
// should call `loadOlderDerivedTranscript` with `page.nextCursor`. No UI
// surfaces older past-turn trails yet, so the first (generous) page is all
// we hydrate today.
return { timelines, transcripts, nextCursor: page.nextCursor ?? null, hasMore: page.hasMore };
}
);
/**
* Load-older hook (Phase C, pagination): fetch the next (older) derived page
* for a thread by `cursor` and MERGE its trails into the already-hydrated
* {@link ChatRuntimeState.turnTimelinesByThread} / `turnTranscriptsByThread`
* (existing turns win — the newer page is authoritative). Wired but currently
* uncalled: no UI exposes a "load older insights" affordance yet.
*/
export const loadOlderDerivedTranscript = createAsyncThunk(
'chatRuntime/loadOlderDerivedTranscript',
async (arg: { threadId: string; cursor: string }, { dispatch, getState }) => {
if (!DERIVED_TRANSCRIPT_ENABLED) return null;
const { threadId, cursor } = arg;
let page: DerivedTranscriptPage;
try {
page = await threadApi.getDerivedTranscript(threadId, {
cursor,
limit: DERIVED_TRANSCRIPT_INITIAL_LIMIT,
});
} catch (error) {
derivedLog('load-older rpc failed thread=%s err=%O', threadId, error);
return null;
}
if (!page.hasTranscript) return null;
const skipRequestIds = liveRequestIdsToSkip(getState(), threadId, page.items);
const { timelines, transcripts } = mapDisplayItems(page.items, { skipRequestIds });
const runtime = readChatRuntimeState(getState());
const mergedTimelines = { ...timelines, ...(runtime?.turnTimelinesByThread[threadId] ?? {}) };
const mergedTranscripts = {
...transcripts,
...(runtime?.turnTranscriptsByThread[threadId] ?? {}),
};
derivedLog(
'load-older merged thread=%s added_timelines=%d added_transcripts=%d hasMore=%s',
threadId,
Object.keys(timelines).length,
Object.keys(transcripts).length,
page.hasMore
);
dispatch(
setTurnTimelinesForThread({
threadId,
timelines: mergedTimelines,
transcripts: mergedTranscripts,
})
);
return { nextCursor: page.nextCursor ?? null, hasMore: page.hasMore };
}
);
export default chatRuntimeSlice.reducer;
+1
View File
@@ -202,6 +202,7 @@ vi.mock('../utils/config', () => ({
E2E_RESTART_APP_AS_RELOAD: false,
DEV_FORCE_ONBOARDING: false,
CHAT_ATTACHMENTS_ENABLED: true,
DERIVED_TRANSCRIPT_ENABLED: true,
SKILLS_GITHUB_REPO: 'test/skills',
GA_MEASUREMENT_ID: undefined,
OPENPANEL_API_URL: 'https://panel.tinyhumans.ai/api',
+158
View File
@@ -0,0 +1,158 @@
/**
* Wire shape of the transcript-derived view RPC
* (`openhuman.threads_transcript_get`, Phase B — see
* `src/openhuman/threads/transcript_view/types.rs`).
*
* The Rust core projects the append-only `session_raw/*.jsonl` source of truth
* into typed **display items** in the frontend's chat vocabulary. Phase C maps
* these onto the existing settled-turn renderers (`PastTurnInsights` /
* `ProcessingTranscriptView` / `ToolTimelineBlock` / `SubagentActivityBlock`)
* via `features/conversations/derived/mapDisplayItems.ts`.
*
* Serde is camelCase on the wire, so every field here is camelCase and mirrors
* the Rust `DisplayItem` / `TranscriptPage` serialization exactly.
*/
/**
* Terminal state of a projected tool call. Mirrors the Rust `ToolCallStatus`
* (snake_case on the wire) and the live timeline's `ToolTimelineStatus`
* vocabulary so the settled projection and the live stream render identically.
*/
export type DerivedToolCallStatus = 'running' | 'success' | 'error';
/**
* One item in a projected transcript, in the frontend's display vocabulary.
* Discriminated by `kind` (camelCase discriminator from the Rust
* `#[serde(tag = "kind")]` enum).
*/
export type DerivedDisplayItem =
| DerivedUserMessage
| DerivedAssistantMessage
| DerivedReasoning
| DerivedToolCall
| DerivedSubagent
| DerivedTurnBoundary
| DerivedInterruptedPartial
| DerivedCompaction;
/**
* A user prompt. `content` is the raw persisted content (may carry an injected
* `Current Date & Time:` scaffolding line); `displayContent` is the sanitized
* version to show, present only when it differs from raw — prefer it.
*/
export interface DerivedUserMessage {
kind: 'userMessage';
content: string;
displayContent?: string;
requestId?: string;
}
/**
* An assistant answer. `interim: true` marks a non-terminal tool-calling step
* within a multi-iteration turn (narration between tool calls), **not** the
* final answer bubble — the final (non-interim) answer stays rendered from the
* thread message list.
*/
export interface DerivedAssistantMessage {
kind: 'assistantMessage';
content: string;
interim?: boolean;
requestId?: string;
model?: string;
iteration?: number;
}
/** The model's reasoning/thinking that preceded an assistant message. */
export interface DerivedReasoning {
kind: 'reasoning';
text: string;
}
/**
* Failure payload attached to an errored tool call. Minimal on the wire (the
* persisted transcript only records the failure plus an optional short reason);
* the mapper expands it into the richer `ToolFailureExplanation` shape the
* `ToolFailureLines` renderer consumes.
*/
export interface DerivedToolFailure {
/** Short, single-line reason for the failure, when the writer captured one. */
detail?: string;
}
/** A tool invocation with its paired result, when available. */
export interface DerivedToolCall {
kind: 'toolCall';
callId: string;
name: string;
args?: unknown;
result?: string;
status: DerivedToolCallStatus;
/** Present only when `status` is `'error'`. */
failure?: DerivedToolFailure;
}
/**
* A delegated sub-agent run, with its own nested projected items. `requestId`
* anchors the whole trail to the parent turn that spawned it (derived core-side
* from the sub-agent's spawn timestamp vs. the parent turns' timestamp ranges);
* absent for legacy/CLI transcripts with no `requestId`.
*/
export interface DerivedSubagent {
kind: 'subagent';
id: string;
requestId?: string;
items: DerivedDisplayItem[];
}
/** A turn boundary — emitted when the `requestId` changes between lines. */
export interface DerivedTurnBoundary {
kind: 'turnBoundary';
requestId: string;
}
/** A partial assistant answer captured when a turn was interrupted. */
export interface DerivedInterruptedPartial {
kind: 'interruptedPartial';
text: string;
thinking?: string;
}
/**
* A context-compaction marker: the reduced set replaced everything before it.
* Carried for completeness; the settled-turn renderers have no compaction
* concept, so the Phase C mapper currently drops it.
*/
export interface DerivedCompaction {
kind: 'compaction';
replacedCount: number;
keptCount: number;
ts?: string;
requestId?: string;
}
/**
* One newest-first page of a thread's projected transcript. `hasTranscript` is
* `false` when the thread has no persisted transcript yet (legacy thread /
* brand-new) — the caller then falls back to the turn_state hydration path.
*/
export interface DerivedTranscriptPage {
threadId: string;
/** Display items for this page, **newest-first**. */
items: DerivedDisplayItem[];
/** Total top-level items available for the thread. */
total: number;
/** Opaque cursor to pass back for the next (older) page; absent at the end. */
nextCursor?: string;
/** `true` when more (older) items remain beyond this page. */
hasMore: boolean;
/** `false` when the thread has no persisted transcript yet (empty page). */
hasTranscript: boolean;
}
/** Optional pagination controls for {@link DerivedTranscriptPage} fetches. */
export interface DerivedTranscriptGetOptions {
/** Opaque token from a prior page's `nextCursor`. */
cursor?: string;
/** Page size; core default 50, clamped to 500. */
limit?: number;
}
+16
View File
@@ -102,6 +102,22 @@ export const CHAT_ATTACHMENTS_ENABLED = import.meta.env.VITE_CHAT_ATTACHMENTS !=
export const SKILLS_GITHUB_REPO =
import.meta.env.VITE_SKILLS_GITHUB_REPO || 'tinyhumansai/openhuman-skills';
/**
* Transcript-derived restore path (Phase C, `docs/plans/transcript-derived-view.md`).
*
* When **on** (default), the settled-turn process trails on thread open are
* hydrated from the `openhuman.threads_transcript_get` projection of the
* append-only `session_raw/*.jsonl` source of truth, instead of the legacy
* `turn_state_history` snapshot ring. Live token streaming is untouched either
* way — in-flight turns still render from socket-fed `chatRuntimeSlice` state.
*
* Automatic fallback to the legacy `turn_state_history` hydration when the RPC
* errors or reports `hasTranscript: false` (legacy threads), so the old path
* stays fully working. Hard-disable the whole derived path for a build with
* `VITE_DERIVED_TRANSCRIPT=false`.
*/
export const DERIVED_TRANSCRIPT_ENABLED = import.meta.env.VITE_DERIVED_TRANSCRIPT !== 'false';
/** Google Analytics 4 Measurement ID. Leave blank to disable GA. */
export const GA_MEASUREMENT_ID = import.meta.env.VITE_GA_MEASUREMENT_ID as string | undefined;
+106
View File
@@ -0,0 +1,106 @@
# Transcript-Derived View — Raw Session Files as Source of Truth
Status: **draft — approved direction, phased implementation**
Branch: `feat/transcript-derived-view` (stacked on `fix/transcript-restore-fidelity`)
Companion: [`conversations-timeline-refactor.md`](conversations-timeline-refactor.md) (Phases 12, 45 landed; this plan supersedes its Phase 5 hydration story for settled turns)
## Goal
Stop maintaining chat state that must be *synced* with what is live. Derive the settled
transcript from the raw session files (`session_raw/*.jsonl`), and demote every other
store to a cache over that file. Live token streaming is untouched: in-flight turns
render from ephemeral socket-fed state exactly as today; the file is authoritative only
for **settled** turns.
This is the Codex rollout model (one JSONL replayed into both model context and UI,
with an explicit persistence policy) adapted to our layout, plus hermes-agent's
soft-compaction lesson (history is never destroyed, only superseded).
## Why derivation is unsafe today (must fix first)
1. **Destructive compaction.** `agent/harness/session/transcript.rs::write_transcript`
full-rewrites the file so context reduction deletes earlier turns from disk.
2. **Display data missing from the file**: interrupted partial answers, `request_id`
turn boundaries, narration items. They exist only in `turn_state` snapshots.
3. **Internal scaffolding in message content**: channel-context prefixes on user
messages, tool-policy preamble in system content — must be sanitized (or tagged) at
projection, never shown raw.
## Architecture
```
live turn (unchanged)
socket events ──► chatRuntimeSlice (ephemeral) ──► renderer
│ chat_done
▼ invalidate
settled turns:
session_raw/{root}.jsonl ─┐
session_raw/{root}__sub-*.jsonl ─┴─► core projection (threads.transcript_get)
│ mtime-keyed cache
typed display items ──► renderer (same components)
```
- **Model context** keeps reading the same file via the existing loader, now replaying
compaction records instead of trusting a rewritten file.
- **`turn_state`** shrinks to live-turn crash recovery (interrupted `streamingText`
until the interrupted line is appended to the file, then only the in-flight turn).
- The 20-turn retention cap stops being user-visible loss: history comes from the file.
## Phases
### Phase A — append-only transcript (Rust, prerequisite)
`transcript.rs` + harness call sites:
- Replace full-rewrite with **append-only** line writes. Context reduction appends a
`compaction` record `{ kind: "compaction", replacement_ids | replacement_history }`;
the model-context loader (`read_transcript` path) replays records to reconstruct the
post-compaction context; a new display reader returns *all* records.
- Stamp `request_id` on every line of a turn (turn boundary markers); keep `iteration`,
`ts`, `seq` alignment with the progress-bridge envelope.
- On turn abort/interrupt, append the partial assistant line flagged
`{ interrupted: true }` so the partial answer is in the file, not only in turn_state.
- Migration: existing files are valid append-only files with zero compaction records —
no migration needed. Old cores reading new files must skip unknown `kind` lines
(verify the `_extra` flatten tolerates this; add a version field to `_meta`).
- Tests: compaction round-trip (model context reduced, display history complete),
interrupted-partial append, request_id stamping, legacy-file read.
### Phase B — projection RPC (Rust)
New `threads.transcript_get(thread_id, {cursor?, limit?})` in the threads domain
(canonical module shape: ops/schemas):
- Resolve root transcript via `find_root_transcript_for_thread`; discover
`__sub-*.jsonl` children; project into typed display items:
`user_message | assistant_message | reasoning | tool_call {args, result, status} |
subagent {id, items} | turn_boundary {request_id} | interrupted_partial`.
- Sanitize scaffolding (channel-context prefix, tool-policy preamble) at projection;
tag rather than mutate where ambiguity exists.
- Cache: per-thread projection keyed on (file paths, mtimes, lengths); invalidated
implicitly by key change. No cache writes to disk — pure memory cache.
- Pagination newest-first with cursor; default window sized for one screen.
- Tests: JSON-RPC E2E (`tests/json_rpc_e2e.rs`) — write file, call RPC, assert items;
subagent merge; sanitization; cache-key invalidation.
### Phase C — frontend switch (TS)
- Thread-open restore path: replace `turn_state_history` hydration for settled turns
with `transcript_get`; map items onto the existing renderers
(`PastTurnInsights`/`ToolTimelineBlock`/`ProcessingTranscriptView`, bubbles).
- Live turn: untouched (socket → chatRuntimeSlice). On `chat_done`, drop the live
turn's ephemeral state and refetch/append the settled projection.
- Keep the turn_state hydration path as fallback behind a flag for one release.
- Tests: restore renders identical item sequence to live for a scripted turn;
interrupted partial visible; legacy thread (no request_id) fallback.
### Phase D — cleanup
- Demote/remove `turn_state` ring retention (keep live-turn snapshot only), drop the
`.md` mirror or mark derived-only, delete the fallback flag.
## Non-goals
- No change to the socket streaming protocol or delta handling.
- No SQLite consolidation (sessions.db stays an index; separate discussion).
- Phase 3 of the timeline refactor (dedup consolidation) remains its own track.
## Risks
- **Old-core / new-file compat** — unknown `kind` lines must be skipped, not fatal.
- **File growth** — append-only grows; mitigate later with Codex-style zstd of old
sessions; out of scope here.
- **Sanitization false positives** — prefer tagging + frontend hiding over deletion.
@@ -511,6 +511,7 @@ impl AgentBuilder {
// `subagents` declaration against the global registry.
agent_definition_id: agent_definition_name.clone(),
session_transcript_path: None,
persisted_transcript_messages: Vec::new(),
session_key: {
let unix_ts = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
@@ -1481,6 +1481,81 @@ fn seed_resume_from_thread_transcript_preserves_tool_calls_and_reasoning() {
);
}
/// Cold-boot resume over an **append-only** transcript that carries a
/// compaction record: the resumed model context must equal the REDUCED set the
/// compaction installed (byte-identical to what the old full-rewrite produced),
/// not the full pre-compaction history.
#[test]
fn seed_resume_replays_compaction_to_reduced_context() {
use super::transcript::{self, TranscriptMeta};
use crate::openhuman::inference::provider::ChatMessage;
let ws = tempfile::TempDir::new().expect("temp workspace");
let wsp = ws.path().to_path_buf();
let thread_id = "thr_compaction_resume";
let meta = TranscriptMeta {
agent_name: "orchestrator_thread-compact".to_string(),
agent_id: Some("orchestrator".to_string()),
agent_type: Some("root".to_string()),
dispatcher: "native".to_string(),
provider: None,
model: None,
created: "2026-01-01T00:00:00Z".to_string(),
updated: "2026-01-01T00:00:00Z".to_string(),
turn_count: 2,
input_tokens: 0,
output_tokens: 0,
cached_input_tokens: 0,
charged_amount_usd: 0.0,
thread_id: Some(thread_id.to_string()),
task_id: None,
};
let path = transcript::resolve_keyed_transcript_path(&wsp, "1700000000_orchestrator")
.expect("resolve transcript path");
// Turn 1: a full exchange. Turn 2: a context reduction (not a prefix) that
// must land as a compaction record.
let full = vec![
ChatMessage::system("system prompt"),
ChatMessage::user("q1"),
ChatMessage::assistant("a1"),
ChatMessage::user("q2"),
ChatMessage::assistant("a2"),
];
transcript::append_transcript_turn(&path, &[], &full, &meta, None, None)
.expect("append turn 1");
let reduced = vec![
ChatMessage::system("system prompt"),
ChatMessage::assistant("[summary] q1/q2"),
ChatMessage::user("q3"),
ChatMessage::assistant("a3"),
];
transcript::append_transcript_turn(&path, &full, &reduced, &meta, None, None)
.expect("append turn 2 (compaction)");
let mut agent = build_minimal_agent_with_definition_name(Some("some_other_agent_name"));
agent.workspace_dir = wsp.clone();
let loaded = agent.seed_resume_from_thread_transcript(thread_id);
assert!(
loaded,
"cold-boot resume must load the compacted transcript"
);
let cached = agent
.cached_transcript_messages
.as_ref()
.expect("cached transcript populated");
assert_eq!(
cached
.iter()
.map(|m| m.content.as_str())
.collect::<Vec<_>>(),
vec!["system prompt", "[summary] q1/q2", "q3", "a3"],
"resumed context must be the reduced set the compaction installed"
);
}
/// When no root transcript exists for the thread, the transcript resume is a
/// no-op returning `false` so the caller falls back to prose-pair seeding.
#[test]
File diff suppressed because it is too large Load Diff
@@ -976,3 +976,367 @@ fn read_thread_usage_summary_groups_subagents_by_archetype() {
assert_eq!(researcher.input_tokens, 500);
assert_eq!(researcher.runs, 1);
}
// ── Phase A: append-only + compaction + display + interrupted ─────────
/// A helper mirroring the in-process persist loop: track the previously
/// persisted logical set and feed each turn through `append_transcript_turn`.
struct AppendHarness {
path: std::path::PathBuf,
prev: Vec<ChatMessage>,
}
impl AppendHarness {
fn new(path: std::path::PathBuf) -> Self {
Self {
path,
prev: Vec::new(),
}
}
fn turn(
&mut self,
messages: &[ChatMessage],
meta: &TranscriptMeta,
usage: Option<&TurnUsage>,
request_id: Option<&str>,
) {
append_transcript_turn(&self.path, &self.prev, messages, meta, usage, request_id)
.expect("append turn");
self.prev = messages.to_vec();
}
}
fn roles(messages: &[ChatMessage]) -> Vec<&str> {
messages.iter().map(|m| m.role.as_str()).collect()
}
/// Pure extension across turns: the model-context read reflects the final
/// (growing) message set and never rewrites earlier lines.
#[test]
fn append_pure_extension_grows_context() {
let dir = TempDir::new().unwrap();
let path = dir.path().join("append.jsonl");
let meta = sample_meta();
let mut h = AppendHarness::new(path.clone());
let turn1 = vec![
ChatMessage::system("sys"),
ChatMessage::user("hi"),
ChatMessage::assistant("hello"),
];
h.turn(&turn1, &meta, None, None);
let mut turn2 = turn1.clone();
turn2.push(ChatMessage::user("again"));
turn2.push(ChatMessage::assistant("hello again"));
h.turn(&turn2, &meta, None, None);
let loaded = read_transcript(&path).unwrap();
assert_eq!(
roles(&loaded.messages),
vec!["system", "user", "assistant", "user", "assistant"]
);
assert_eq!(loaded.messages[4].content, "hello again");
}
/// Compaction round-trip: after a reduction, the model-context read returns the
/// REDUCED context, while the display read returns the FULL pre-compaction
/// history plus the compaction marker.
#[test]
fn compaction_round_trip_model_vs_display() {
let dir = TempDir::new().unwrap();
let path = dir.path().join("compact.jsonl");
let meta = sample_meta();
let mut h = AppendHarness::new(path.clone());
// Three growing turns.
let base = vec![
ChatMessage::system("sys"),
ChatMessage::user("q1"),
ChatMessage::assistant("a1"),
ChatMessage::user("q2"),
ChatMessage::assistant("a2"),
];
h.turn(&base, &meta, None, None);
// A reduction: the harness drops the earliest exchange and keeps a summary
// + the recent tail. This is NOT a prefix of `base`, so it must land as a
// compaction record.
let reduced = vec![
ChatMessage::system("sys"),
ChatMessage::assistant("[summary] earlier discussion about q1/q2"),
ChatMessage::user("q3"),
ChatMessage::assistant("a3"),
];
h.turn(&reduced, &meta, None, None);
// Model-context read == the reduced set only.
let model = read_transcript(&path).unwrap();
assert_eq!(
model
.messages
.iter()
.map(|m| m.content.as_str())
.collect::<Vec<_>>(),
vec![
"sys",
"[summary] earlier discussion about q1/q2",
"q3",
"a3"
],
"model context must reflect the reduced set after compaction"
);
// Display read == full history: the 5 pre-compaction messages, then a
// compaction marker carrying the 4-message replacement.
let display = read_transcript_display(&path).unwrap();
let pre: Vec<&str> = display
.records
.iter()
.take_while(|r| matches!(r, DisplayRecord::Message(_)))
.filter_map(|r| match r {
DisplayRecord::Message(m) => Some(m.message.content.as_str()),
_ => None,
})
.collect();
assert_eq!(pre, vec!["sys", "q1", "a1", "q2", "a2"]);
let marker = display
.records
.iter()
.find_map(|r| match r {
DisplayRecord::Compaction(c) => Some(c),
_ => None,
})
.expect("display must retain the compaction marker");
assert_eq!(marker.replacement.len(), 4);
assert_eq!(
marker.replacement[1].message.content,
"[summary] earlier discussion about q1/q2"
);
}
/// After a compaction, a subsequent pure extension appends normally and the
/// model-context read replays reset-then-extend to the correct final set.
#[test]
fn append_after_compaction_extends_reduced_set() {
let dir = TempDir::new().unwrap();
let path = dir.path().join("compact_then_extend.jsonl");
let meta = sample_meta();
let mut h = AppendHarness::new(path.clone());
h.turn(
&[
ChatMessage::system("sys"),
ChatMessage::user("q1"),
ChatMessage::assistant("a1"),
],
&meta,
None,
None,
);
let reduced = vec![
ChatMessage::system("sys"),
ChatMessage::assistant("[summary]"),
];
h.turn(&reduced, &meta, None, None);
let mut extended = reduced.clone();
extended.push(ChatMessage::user("q2"));
extended.push(ChatMessage::assistant("a2"));
h.turn(&extended, &meta, None, None);
let model = read_transcript(&path).unwrap();
assert_eq!(
model
.messages
.iter()
.map(|m| m.content.as_str())
.collect::<Vec<_>>(),
vec!["sys", "[summary]", "q2", "a2"]
);
}
/// request_id turn-boundary stamping round-trips into the display projection on
/// every appended line of a turn.
#[test]
fn request_id_stamped_on_every_line() {
let dir = TempDir::new().unwrap();
let path = dir.path().join("reqid.jsonl");
let meta = sample_meta();
let mut h = AppendHarness::new(path.clone());
let turn1 = vec![ChatMessage::system("sys"), ChatMessage::user("q1")];
h.turn(&turn1, &meta, None, Some("req-1"));
let mut turn2 = turn1.clone();
turn2.push(ChatMessage::assistant("a2"));
h.turn(&turn2, &meta, None, Some("req-2"));
let display = read_transcript_display(&path).unwrap();
let msgs: Vec<&DisplayMessage> = display
.records
.iter()
.filter_map(|r| match r {
DisplayRecord::Message(m) => Some(m),
_ => None,
})
.collect();
// turn1 wrote sys + user with req-1; turn2 appended only the assistant tail
// with req-2.
assert_eq!(msgs[0].request_id.as_deref(), Some("req-1"));
assert_eq!(msgs[1].request_id.as_deref(), Some("req-1"));
assert_eq!(msgs[2].request_id.as_deref(), Some("req-2"));
assert_eq!(msgs[2].message.content, "a2");
}
/// An interrupted partial is appended to the file, is visible in the display
/// read flagged `interrupted`, and is SKIPPED by the model-context read (a
/// resumed context never carries a truncated answer).
#[test]
fn interrupted_partial_display_only() {
let dir = TempDir::new().unwrap();
let path = dir.path().join("interrupted.jsonl");
let meta = sample_meta();
let mut h = AppendHarness::new(path.clone());
h.turn(
&[ChatMessage::system("sys"), ChatMessage::user("q1")],
&meta,
None,
Some("req-1"),
);
append_interrupted_partial(
&path,
"partial answer that was cut off",
Some("req-1"),
Some(3),
Some("thinking that was cut off"),
)
.expect("append interrupted");
// Model context: the partial is skipped.
let model = read_transcript(&path).unwrap();
assert_eq!(roles(&model.messages), vec!["system", "user"]);
assert!(
!model.messages.iter().any(|m| m.content.contains("cut off")),
"interrupted partial must NOT enter the model context"
);
// Display: the partial is present and flagged.
let display = read_transcript_display(&path).unwrap();
let partial = display
.records
.iter()
.find_map(|r| match r {
DisplayRecord::Message(m) if m.interrupted => Some(m),
_ => None,
})
.expect("display must include the interrupted partial");
assert_eq!(partial.message.content, "partial answer that was cut off");
assert_eq!(partial.request_id.as_deref(), Some("req-1"));
assert_eq!(partial.iteration, Some(3));
assert_eq!(
partial.reasoning_content.as_deref(),
Some("thinking that was cut off"),
"interrupted partial must carry its reasoning_content"
);
}
/// Empty partial content is a no-op — no line is written.
#[test]
fn interrupted_partial_empty_is_noop() {
let dir = TempDir::new().unwrap();
let path = dir.path().join("empty_interrupt.jsonl");
let meta = sample_meta();
let mut h = AppendHarness::new(path.clone());
h.turn(&[ChatMessage::user("q")], &meta, None, None);
append_interrupted_partial(&path, "", None, None, None).expect("noop");
let display = read_transcript_display(&path).unwrap();
assert!(display
.records
.iter()
.all(|r| matches!(r, DisplayRecord::Message(m) if !m.interrupted)));
}
/// A legacy file — one produced by the full-rewrite `write_transcript` with no
/// compaction records and no `version` — reads identically under both the
/// model-context and display readers.
#[test]
fn legacy_file_reads_identically() {
let dir = TempDir::new().unwrap();
let path = dir.path().join("legacy.jsonl");
let messages = sample_messages();
let meta = sample_meta();
// Full-rewrite writer == the legacy shape (append-only readers must tolerate
// it: zero compaction records, last `_meta` == the only `_meta`).
write_transcript(&path, &messages, &meta, None).unwrap();
let model = read_transcript(&path).unwrap();
assert_eq!(model.messages.len(), messages.len());
assert_eq!(roles(&model.messages), roles(&messages));
let display = read_transcript_display(&path).unwrap();
let display_roles: Vec<&str> = display
.records
.iter()
.filter_map(|r| match r {
DisplayRecord::Message(m) => Some(m.message.role.as_str()),
_ => None,
})
.collect();
assert_eq!(display_roles, roles(&messages));
}
/// A file carrying an unknown record kind (as a future core might write) is
/// skipped by the reader rather than crashing it.
#[test]
fn unknown_record_kind_is_skipped_not_fatal() {
let dir = TempDir::new().unwrap();
let path = dir.path().join("unknown_kind.jsonl");
let meta = sample_meta();
let mut h = AppendHarness::new(path.clone());
h.turn(
&[ChatMessage::system("sys"), ChatMessage::user("q1")],
&meta,
None,
None,
);
// Simulate a future kind by appending a foreign record line.
{
use std::io::Write;
let mut f = std::fs::OpenOptions::new()
.append(true)
.open(&path)
.unwrap();
writeln!(f, "{{\"kind\":\"future_thing\",\"payload\":42}}").unwrap();
}
// Append a normal turn after the unknown line to prove reading continues.
let mut msgs = vec![ChatMessage::system("sys"), ChatMessage::user("q1")];
msgs.push(ChatMessage::assistant("a1"));
h.prev = vec![ChatMessage::system("sys"), ChatMessage::user("q1")];
h.turn(&msgs, &meta, None, None);
let model = read_transcript(&path).unwrap();
// The unknown record is skipped; the real messages survive.
assert!(model.messages.iter().any(|m| m.content == "a1"));
assert!(!model
.messages
.iter()
.any(|m| m.content.contains("future_thing")));
}
/// The `_meta` version field is stamped by the append writer and absent (0) on
/// legacy files — but both remain readable.
#[test]
fn meta_version_stamped_and_optional() {
let dir = TempDir::new().unwrap();
let path = dir.path().join("version.jsonl");
let meta = sample_meta();
let mut h = AppendHarness::new(path.clone());
h.turn(&[ChatMessage::user("q")], &meta, None, None);
let raw = std::fs::read_to_string(&path).unwrap();
assert!(
raw.lines().next().unwrap().contains("\"version\":1"),
"append writer must stamp the schema version on the meta header"
);
}
@@ -332,6 +332,67 @@ fn tool_records_from_conversation(
records
}
/// Stamp each **failed** tool-result [`ChatMessage`] with its failure outcome
/// before persistence, so the derived transcript view can render an error tool
/// row instead of a false success.
///
/// The harness folds a tool result into a `role:"tool"` message whose native
/// content envelope (`{"tool_call_id":…,"content":…}`) has already dropped
/// `ToolResult::is_error`. The only structured per-call success signal is the
/// captured [`ToolCallOutcome`] side-channel; correlate by provider call id and
/// re-attach an additive failure marker (see
/// `transcript::attach_tool_failure_metadata`). Non-tool messages, tool messages
/// with no matching outcome, and successful calls are left untouched.
fn stamp_tool_failures(
messages: &mut [ChatMessage],
tool_outcomes: &[crate::openhuman::tinyagents::ToolCallOutcome],
) {
use crate::openhuman::agent::harness::session::transcript;
if tool_outcomes.is_empty() {
return;
}
for msg in messages.iter_mut() {
if msg.role != "tool" {
continue;
}
let Some(call_id) = parse_tool_call_id(&msg.content) else {
continue;
};
let Some(outcome) = tool_outcomes.iter().find(|o| o.call_id == call_id) else {
continue;
};
if outcome.success {
continue;
}
let detail = short_failure_detail(&outcome.content);
log::debug!(
"[transcript] stamping tool failure call_id={call_id} name={}",
outcome.name
);
transcript::attach_tool_failure_metadata(msg, detail.as_deref());
}
}
/// Extract the `tool_call_id` from a native tool-result content envelope
/// (`{"tool_call_id":…,"content":…}`). `None` for non-envelope content (XML /
/// P-Format dispatchers, which don't emit `role:"tool"` messages anyway).
fn parse_tool_call_id(content: &str) -> Option<String> {
let value: serde_json::Value = serde_json::from_str(content).ok()?;
value.get("tool_call_id")?.as_str().map(str::to_string)
}
/// Reduce a tool's error output to a short, single-line reason for display.
fn short_failure_detail(content: &str) -> Option<String> {
const MAX: usize = 160;
let line = content.lines().map(str::trim).find(|l| !l.is_empty())?;
let short: String = line.chars().take(MAX).collect();
if short.is_empty() {
None
} else {
Some(short)
}
}
/// Rewrite the **trailing** assistant `Chat` message in `history` to `text`,
/// keeping the persisted transcript and the next turn's KV-cache prefix
/// consistent with a repaired required-output reply (issue #4117). Only the last
@@ -1465,7 +1526,11 @@ impl Agent {
},
);
let persisted = self.tool_dispatcher.to_provider_messages(&self.history);
let mut persisted = self.tool_dispatcher.to_provider_messages(&self.history);
// Re-attach per-call failure outcomes (dropped when the engine folded
// each tool result into a `role:"tool"` message) so the derived
// transcript view renders failed tools as errors, not successes.
stamp_tool_failures(&mut persisted, &outcome.tool_outcomes);
// Carry the turn's provider (event channel) + effective model and usage
// into the persisted transcript meta. Passing `None` here dropped
// `provider`/`model` from every transcript (they are `TranscriptMeta`
@@ -546,8 +546,25 @@ impl Agent {
task_id: None,
};
match transcript::write_transcript(path, messages, &meta, turn_usage) {
// Append-only write (Phase A, transcript-derived view): diff this turn's
// logical messages against the previously-persisted set tracked in
// memory. A pure extension appends only the new tail; a context
// reduction appends a `compaction` record. The file is never rewritten,
// so pre-compaction history survives on disk for the display projection.
// `request_id` (web-chat only) stamps a turn boundary on each line.
let prev = std::mem::take(&mut self.persisted_transcript_messages);
let request_id = crate::openhuman::agent::turn_origin::current_request_id();
match transcript::append_transcript_turn(
path,
&prev,
messages,
&meta,
turn_usage,
request_id.as_deref(),
) {
Ok(()) => {
// Track the new persisted logical set for the next turn's diff.
self.persisted_transcript_messages = messages.to_vec();
// Best-effort, non-fatal dual-write into the TinyAgents store.
// Gated by the default-ON session dual-write flag
// (`OPENHUMAN_SESSION_DUAL_WRITE` is a kill switch). Only runs
@@ -556,8 +573,11 @@ impl Agent {
self.maybe_dual_write_session_store(path, messages, &meta, turn_usage);
}
Err(err) => {
// Restore the tracked state so a transient failure doesn't make
// the next turn mis-diff (and spuriously emit a compaction).
self.persisted_transcript_messages = prev;
log::warn!(
"[transcript] failed to write transcript {}: {err}",
"[transcript] failed to append transcript {}: {err}",
path.display()
);
}
+8 -1
View File
@@ -122,9 +122,16 @@ pub struct Agent {
/// [`AgentDefinitionRegistry`]: crate::openhuman::agent::harness::definition::AgentDefinitionRegistry
pub(super) agent_definition_id: String,
/// Resolved filesystem path for this session's transcript file.
/// Set on first write, reused for subsequent overwrites within the
/// Set on first write, reused for subsequent **appends** within the
/// same session.
pub(super) session_transcript_path: Option<PathBuf>,
/// The logical message set most recently persisted to
/// `session_transcript_path`, tracked in memory so the append-only writer
/// can diff each turn's messages against it (pure extension → append tail;
/// reduction → compaction record) without re-reading the growing file.
/// Empty until the first persist. Each process writes its own transcript
/// file, so this in-memory state is always aligned with the file it owns.
pub(super) persisted_transcript_messages: Vec<ChatMessage>,
/// Unique transcript key for this session, formatted as
/// `"{unix_ts}_{agent_id}"`. Generated once at agent-build time so
/// every transcript write in this session uses the same filename
+12
View File
@@ -134,6 +134,18 @@ pub fn current() -> Option<AgentTurnOrigin> {
AGENT_TURN_ORIGIN.try_with(|o| o.clone()).ok()
}
/// Read the ambient web-chat `request_id` for the current turn, when one was
/// scoped by an [`AgentTurnOrigin::WebChat`] entry point. `None` for every
/// other origin (channel / cron / CLI / sub-agent) and outside any scope —
/// those turns are not request-scoped, so their transcript lines carry no
/// turn-boundary marker.
pub fn current_request_id() -> Option<String> {
match current() {
Some(AgentTurnOrigin::WebChat { request_id, .. }) => request_id,
_ => None,
}
}
#[cfg(test)]
mod tests {
use super::*;
+1
View File
@@ -9,6 +9,7 @@ pub mod ops;
pub mod schemas;
pub mod title;
pub mod tools;
pub mod transcript_view;
pub mod turn_state;
pub mod welcome_migration;
+51
View File
@@ -693,6 +693,57 @@ pub struct ThreadTokenUsageRequest {
pub thread_id: String,
}
/// Request for [`transcript_get`]: the thread to project, plus newest-first
/// pagination controls. `cursor` is the opaque token from a prior page's
/// `nextCursor`; `limit` defaults to one screen.
#[derive(Debug, Clone, serde::Deserialize)]
pub struct TranscriptGetRequest {
pub thread_id: String,
#[serde(default)]
pub cursor: Option<String>,
#[serde(default)]
pub limit: Option<usize>,
}
/// Project a thread's settled transcript (derived from `session_raw/*.jsonl`)
/// into typed display items, newest-first paginated. Returns an empty page with
/// `hasTranscript: false` when the thread has no persisted transcript yet.
pub async fn transcript_get(
request: TranscriptGetRequest,
) -> Result<
RpcOutcome<ApiEnvelope<crate::openhuman::threads::transcript_view::TranscriptPage>>,
String,
> {
let dir = workspace_dir().await?;
let thread_id = request.thread_id.trim();
if thread_id.is_empty() {
return Err("thread_id is required".to_string());
}
let page = crate::openhuman::threads::transcript_view::get_page(
&dir,
thread_id,
request.cursor.as_deref(),
request.limit,
);
let counts = counts([
("items", page.items.len()),
("total", page.total),
("has_transcript", usize::from(page.has_transcript)),
]);
let pagination = Some(PaginationMeta {
limit: request
.limit
.unwrap_or(crate::openhuman::threads::transcript_view::DEFAULT_LIMIT),
offset: request
.cursor
.as_deref()
.and_then(|c| c.trim().parse::<usize>().ok())
.unwrap_or(0),
count: page.total,
});
Ok(envelope(page, Some(counts), pagination))
}
/// Aggregated token/cost usage for one thread, read back from its persisted
/// session transcripts. Seeds the UI footer when the user selects a thread so
/// the totals reflect prior turns instead of starting at zero.
+44
View File
@@ -39,6 +39,7 @@ pub fn all_controller_schemas() -> Vec<ControllerSchema> {
schemas("task_board_get"),
schemas("task_board_put"),
schemas("token_usage"),
schemas("transcript_get"),
]
}
@@ -120,6 +121,10 @@ pub fn all_registered_controllers() -> Vec<RegisteredController> {
schema: schemas("token_usage"),
handler: handle_token_usage,
},
RegisteredController {
schema: schemas("transcript_get"),
handler: handle_transcript_get,
},
]
}
@@ -527,6 +532,38 @@ pub fn schemas(function: &str) -> ControllerSchema {
required: true,
}],
},
"transcript_get" => ControllerSchema {
namespace: "threads",
function: "transcript_get",
description:
"Project a thread's settled transcript (derived from session_raw/*.jsonl) into typed display items, newest-first paginated.",
inputs: vec![
FieldSchema {
name: "thread_id",
ty: TypeSchema::String,
comment: "Thread identifier.",
required: true,
},
FieldSchema {
name: "cursor",
ty: TypeSchema::Option(Box::new(TypeSchema::String)),
comment: "Opaque pagination cursor from a prior page's nextCursor; absent starts at the newest item.",
required: false,
},
FieldSchema {
name: "limit",
ty: TypeSchema::Option(Box::new(TypeSchema::U64)),
comment: "Max items to return (default 50, capped at 500).",
required: false,
},
],
outputs: vec![FieldSchema {
name: "result",
ty: TypeSchema::Json,
comment: "Envelope with the newest-first page of display items, total, and nextCursor.",
required: true,
}],
},
_other => ControllerSchema {
namespace: "threads",
function: "unknown",
@@ -745,6 +782,13 @@ fn handle_token_usage(params: Map<String, Value>) -> ControllerFuture {
})
}
fn handle_transcript_get(params: Map<String, Value>) -> ControllerFuture {
Box::pin(async move {
let p = parse::<ops::TranscriptGetRequest>(params)?;
to_json(ops::transcript_get(p).await?)
})
}
// ── Helpers ──────────────────────────────────────────────────────────
fn parse<T: DeserializeOwned>(params: Map<String, Value>) -> Result<T, String> {
+1
View File
@@ -22,6 +22,7 @@ const ALL_FUNCTIONS: &[&str] = &[
"task_board_get",
"task_board_put",
"token_usage",
"transcript_get",
];
#[test]
@@ -0,0 +1,152 @@
//! In-memory, mtime-keyed projection cache for the transcript view.
//!
//! The settled transcript is derived from `session_raw/*.jsonl` on every
//! request. Re-projecting a long thread on each page fetch is wasteful, so we
//! memoize per-thread projections keyed on the backing files' `(path, mtime,
//! len)` signature. An append (new turn, interrupted partial, sub-agent file)
//! changes the signature and transparently invalidates the entry — there are
//! **no disk writes** and no explicit invalidation call. The cache is bounded
//! (LRU over a few dozen threads) so a long-lived core can't grow it without
//! limit.
use std::collections::HashMap;
use std::collections::VecDeque;
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex, OnceLock};
use std::time::SystemTime;
use super::project;
use super::types::ProjectedTranscript;
const LOG_PREFIX: &str = "[threads][transcript][cache]";
/// Number of distinct threads whose projections are retained. Beyond this the
/// least-recently-used entry is evicted.
const CACHE_CAPACITY: usize = 32;
/// Signature of one backing file — changes on any append/rewrite.
#[derive(Debug, Clone, PartialEq, Eq)]
struct FileSig {
path: PathBuf,
mtime: Option<SystemTime>,
len: u64,
}
fn file_sig(path: &Path) -> FileSig {
let (mtime, len) = match std::fs::metadata(path) {
Ok(meta) => (meta.modified().ok(), meta.len()),
Err(_) => (None, 0),
};
FileSig {
path: path.to_path_buf(),
mtime,
len,
}
}
struct CacheEntry {
signature: Vec<FileSig>,
projected: Arc<ProjectedTranscript>,
}
#[derive(Default)]
struct CacheInner {
entries: HashMap<String, CacheEntry>,
/// LRU order — front is least-recently-used.
order: VecDeque<String>,
}
/// Bounded per-thread projection cache.
#[derive(Default)]
pub struct TranscriptViewCache {
inner: Mutex<CacheInner>,
}
impl TranscriptViewCache {
/// Project `thread_id`'s transcript, serving a cached result when the
/// backing files are unchanged. `None` when the thread has no transcript.
pub fn get_or_project(
&self,
workspace_dir: &Path,
thread_id: &str,
) -> Option<Arc<ProjectedTranscript>> {
let (root_path, sub_paths) = project::resolve_files(workspace_dir, thread_id)?;
let signature: Vec<FileSig> = std::iter::once(file_sig(&root_path))
.chain(sub_paths.iter().map(|p| file_sig(p)))
.collect();
{
let mut inner = self.inner.lock().ok()?;
if let Some(entry) = inner.entries.get(thread_id) {
if entry.signature == signature {
let projected = entry.projected.clone();
touch(&mut inner, thread_id);
log::debug!(
"{LOG_PREFIX} hit thread={thread_id} items={}",
projected.items.len()
);
return Some(projected);
}
log::debug!("{LOG_PREFIX} miss (signature changed) thread={thread_id}");
} else {
log::debug!("{LOG_PREFIX} miss (cold) thread={thread_id}");
}
}
let projected = Arc::new(project::project_from_files(
thread_id, &root_path, &sub_paths,
));
let mut inner = self.inner.lock().ok()?;
inner.entries.insert(
thread_id.to_string(),
CacheEntry {
signature,
projected: projected.clone(),
},
);
touch(&mut inner, thread_id);
evict_if_needed(&mut inner);
log::debug!(
"{LOG_PREFIX} stored thread={thread_id} items={} cache_size={}",
projected.items.len(),
inner.entries.len()
);
Some(projected)
}
#[cfg(test)]
fn len(&self) -> usize {
self.inner.lock().unwrap().entries.len()
}
}
/// Move `thread_id` to the most-recently-used end of the LRU order.
fn touch(inner: &mut CacheInner, thread_id: &str) {
if let Some(pos) = inner.order.iter().position(|t| t == thread_id) {
inner.order.remove(pos);
}
inner.order.push_back(thread_id.to_string());
}
/// Evict least-recently-used entries until within [`CACHE_CAPACITY`].
fn evict_if_needed(inner: &mut CacheInner) {
while inner.entries.len() > CACHE_CAPACITY {
let Some(victim) = inner.order.pop_front() else {
break;
};
inner.entries.remove(&victim);
log::debug!("{LOG_PREFIX} evicted thread={victim}");
}
}
/// Process-wide cache singleton. The transcript view is read-only and derived,
/// so one shared cache across RPC calls is correct.
pub fn global() -> &'static TranscriptViewCache {
static CACHE: OnceLock<TranscriptViewCache> = OnceLock::new();
CACHE.get_or_init(TranscriptViewCache::default)
}
#[cfg(test)]
#[path = "cache_tests.rs"]
mod tests;
@@ -0,0 +1,74 @@
//! Cache hit/miss + invalidation tests for the transcript view cache.
use super::TranscriptViewCache;
use crate::openhuman::agent::harness::session::transcript;
use std::path::Path;
use tempfile::TempDir;
fn meta_line(thread_id: &str) -> String {
format!(
r#"{{"_meta":{{"version":1,"agent":"orchestrator","dispatcher":"native","created":"2026-07-21T00:00:00Z","updated":"2026-07-21T00:00:00Z","turn_count":1,"input_tokens":0,"output_tokens":0,"cached_input_tokens":0,"charged_amount_usd":0.0,"thread_id":"{thread_id}"}}}}"#
)
}
fn write_raw(workspace: &Path, stem: &str, thread_id: &str, body: &[&str]) -> std::path::PathBuf {
let path = transcript::resolve_keyed_transcript_path(workspace, stem).unwrap();
let mut buf = meta_line(thread_id);
buf.push('\n');
for line in body {
buf.push_str(line);
buf.push('\n');
}
std::fs::write(&path, buf).unwrap();
path
}
#[test]
fn recomputes_when_file_grows() {
let dir = TempDir::new().unwrap();
let path = write_raw(
dir.path(),
"100_orchestrator",
"thr_cache",
&[r#"{"role":"user","content":"first"}"#],
);
let cache = TranscriptViewCache::default();
let a = cache
.get_or_project(dir.path(), "thr_cache")
.expect("first");
assert_eq!(a.items.len(), 1);
// Second call, unchanged file → same cached Arc (hit).
let b = cache
.get_or_project(dir.path(), "thr_cache")
.expect("second");
assert!(
std::sync::Arc::ptr_eq(&a, &b),
"unchanged file must serve cached Arc"
);
// Append a line → file length changes → signature invalidates → recompute.
let mut appended = std::fs::read_to_string(&path).unwrap();
appended.push_str(r#"{"role":"assistant","content":"second"}"#);
appended.push('\n');
std::fs::write(&path, appended).unwrap();
let c = cache
.get_or_project(dir.path(), "thr_cache")
.expect("third");
assert!(!std::sync::Arc::ptr_eq(&a, &c), "grown file must recompute");
assert_eq!(
c.items.len(),
2,
"recomputed projection reflects the append"
);
}
#[test]
fn missing_thread_returns_none() {
let dir = TempDir::new().unwrap();
let cache = TranscriptViewCache::default();
assert!(cache.get_or_project(dir.path(), "ghost").is_none());
assert_eq!(cache.len(), 0);
}
@@ -0,0 +1,105 @@
//! Transcript-derived view: project the append-only `session_raw/*.jsonl`
//! source of truth into typed display items for the chat renderer, with
//! newest-first pagination over a bounded in-memory cache.
//!
//! Entry point: [`get_page`] (used by the `threads.transcript_get` RPC).
mod cache;
mod project;
pub mod types;
use std::path::Path;
use serde::Serialize;
pub use types::{DisplayItem, ProjectedTranscript, ToolCallStatus};
const LOG_PREFIX: &str = "[threads][transcript]";
/// Default page size — roughly one screen of chat items.
pub const DEFAULT_LIMIT: usize = 50;
/// Hard upper bound on a single page so a client can't request an unbounded
/// projection slice.
pub const MAX_LIMIT: usize = 500;
/// One newest-first page of a thread's projected transcript.
#[derive(Debug, Clone, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct TranscriptPage {
pub thread_id: String,
/// Display items for this page, **newest-first**.
pub items: Vec<DisplayItem>,
/// Total top-level items available for the thread.
pub total: usize,
/// Opaque cursor to pass back for the next (older) page; `null` at the end.
#[serde(skip_serializing_if = "Option::is_none")]
pub next_cursor: Option<String>,
/// `true` when more (older) items remain beyond this page.
pub has_more: bool,
/// `false` when the thread has no persisted transcript yet (empty page).
pub has_transcript: bool,
}
/// Project `thread_id`'s transcript and return one newest-first page.
///
/// `cursor` is the opaque token returned as `next_cursor` by a previous call
/// (an offset from the newest item); `None`/empty starts at the newest item.
/// `limit` defaults to [`DEFAULT_LIMIT`] and is clamped to [`MAX_LIMIT`].
pub fn get_page(
workspace_dir: &Path,
thread_id: &str,
cursor: Option<&str>,
limit: Option<usize>,
) -> TranscriptPage {
let limit = limit.unwrap_or(DEFAULT_LIMIT).clamp(1, MAX_LIMIT);
let offset = parse_cursor(cursor);
let Some(projected) = cache::global().get_or_project(workspace_dir, thread_id) else {
log::debug!("{LOG_PREFIX} get_page thread={thread_id}: no transcript");
return TranscriptPage {
thread_id: thread_id.to_string(),
items: Vec::new(),
total: 0,
next_cursor: None,
has_more: false,
has_transcript: false,
};
};
let total = projected.items.len();
let start = offset.min(total);
let end = (offset + limit).min(total);
// Newest-first: item `offset` is the newest, walking backwards from the end.
let items: Vec<DisplayItem> = (start..end)
.map(|i| projected.items[total - 1 - i].clone())
.collect();
let has_more = end < total;
let next_cursor = has_more.then(|| end.to_string());
log::debug!(
"{LOG_PREFIX} get_page thread={thread_id} total={total} offset={offset} returned={} has_more={has_more}",
items.len()
);
TranscriptPage {
thread_id: thread_id.to_string(),
items,
total,
next_cursor,
has_more,
has_transcript: true,
}
}
/// Parse the opaque cursor into a numeric offset (0 on absent/invalid).
fn parse_cursor(cursor: Option<&str>) -> usize {
cursor
.map(str::trim)
.filter(|c| !c.is_empty())
.and_then(|c| c.parse::<usize>().ok())
.unwrap_or(0)
}
#[cfg(test)]
#[path = "tests.rs"]
mod tests;
@@ -0,0 +1,499 @@
//! Project raw session-transcript records into typed display items.
//!
//! Turns the append-only log's [`DisplayRecord`]s (message lines, compaction
//! markers, interrupted partials) into the frontend's chat vocabulary
//! ([`DisplayItem`]), sanitizing injected scaffolding as it goes. Sub-agent
//! sibling files are discovered and nested one level deep.
use std::collections::VecDeque;
use std::fs;
use std::path::{Path, PathBuf};
use crate::openhuman::agent::harness::session::transcript::{
self, CompactionMarker, DisplayMessage, DisplayRecord,
};
use super::types::{DisplayItem, ProjectedTranscript, ToolCallFailure, ToolCallStatus};
const LOG_PREFIX: &str = "[threads][transcript]";
/// Max sub-agent nesting depth the projection descends. The plan calls for
/// one level of recursion; we allow a small bound so a delegated worker that
/// itself delegates still surfaces, without unbounded fan-out.
const MAX_SUBAGENT_DEPTH: usize = 3;
/// The scaffolding line injected onto every user message (see
/// `agent::prompts::current_datetime_line`). Stripped at projection so the UI
/// shows the user's actual words, not the per-turn time stamp.
const DATETIME_PREFIX: &str = "Current Date & Time:";
/// A legacy/alternate channel-context prefix. Kept for defensiveness; the
/// live injector currently only prepends [`DATETIME_PREFIX`].
const CHANNEL_CONTEXT_PREFIX: &str = "[Channel context]";
/// Resolve a thread's root transcript, discover its sub-agent siblings, and
/// project everything into display items. Returns `None` when the thread has
/// no root transcript yet (brand-new thread / first turn not persisted).
pub fn project_thread(workspace_dir: &Path, thread_id: &str) -> Option<ProjectedTranscript> {
let (root_path, sub_paths) = resolve_files(workspace_dir, thread_id)?;
Some(project_from_files(thread_id, &root_path, &sub_paths))
}
/// Resolve the on-disk file set backing a thread's transcript view: the root
/// transcript path plus every sub-agent sibling file. `None` when the thread
/// has no root transcript yet. Exposed so the cache can key on these paths
/// (and their mtimes/lengths) without re-projecting.
pub fn resolve_files(workspace_dir: &Path, thread_id: &str) -> Option<(PathBuf, Vec<PathBuf>)> {
let root_path = transcript::find_root_transcript_for_thread(workspace_dir, thread_id)?;
let root_stem = root_path.file_stem()?.to_str()?.to_string();
let sub_paths = discover_subagent_files(workspace_dir, &root_stem);
Some((root_path, sub_paths))
}
/// Project a thread from an already-resolved file set (root + sub-agent
/// siblings). Missing/unreadable files degrade to empty rather than failing.
pub fn project_from_files(
thread_id: &str,
root_path: &Path,
sub_paths: &[PathBuf],
) -> ProjectedTranscript {
let root_stem = root_path
.file_stem()
.and_then(|s| s.to_str())
.unwrap_or_default()
.to_string();
log::debug!(
"{LOG_PREFIX} projecting thread={thread_id} root={} subagent_files={}",
root_path.display(),
sub_paths.len()
);
// Read the root display records once: they feed both the top-level items
// and the per-turn timestamp ranges used to anchor sub-agent trails.
let (mut items, segments) = match transcript::read_transcript_display(root_path) {
Ok(d) => (project_records(&d.records), turn_segments(&d.records)),
Err(err) => {
log::warn!(
"{LOG_PREFIX} failed to read root transcript {}: {err}",
root_path.display()
);
(Vec::new(), Vec::new())
}
};
let subagents = build_subagent_items(sub_paths, &root_stem, 0, &segments);
log::debug!(
"{LOG_PREFIX} projected thread={thread_id} top_level_items={} subagents={}",
items.len(),
subagents.len()
);
items.extend(subagents);
ProjectedTranscript {
thread_id: thread_id.to_string(),
items,
}
}
/// Discover every sub-agent transcript file for `root_stem` under
/// `session_raw/`. Sub-agent stems are `{root_stem}__…`; results are sorted so
/// the timestamp-prefixed suffixes order by creation time.
fn discover_subagent_files(workspace_dir: &Path, root_stem: &str) -> Vec<PathBuf> {
let raw_dir = workspace_dir.join("session_raw");
let prefix = format!("{root_stem}__");
let Ok(entries) = fs::read_dir(&raw_dir) else {
return Vec::new();
};
let mut paths: Vec<PathBuf> = entries
.flatten()
.map(|e| e.path())
.filter(|p| p.extension().and_then(|s| s.to_str()) == Some("jsonl"))
.filter(|p| {
p.file_stem()
.and_then(|s| s.to_str())
.is_some_and(|stem| stem.starts_with(&prefix))
})
.collect();
paths.sort();
paths
}
/// Build nested [`DisplayItem::Subagent`] items for the direct children of
/// `parent_stem` among `all_sub_paths`, recursing one level per depth up to
/// [`MAX_SUBAGENT_DEPTH`]. Attachment is flat (ordered by file timestamp): the
/// transcript doesn't record a robust delegation-call → file link, so we nest
/// by stem lineage rather than guessing the parent tool call.
///
/// Each item is anchored to a parent turn via [`anchor_request_id`] so the
/// frontend can render the trail under the turn that spawned it rather than the
/// most recent turn. `segments` are the root turns' start timestamps.
fn build_subagent_items(
all_sub_paths: &[PathBuf],
parent_stem: &str,
depth: usize,
segments: &[(String, i64)],
) -> Vec<DisplayItem> {
if depth >= MAX_SUBAGENT_DEPTH {
return Vec::new();
}
let child_prefix = format!("{parent_stem}__");
let mut out = Vec::new();
for path in all_sub_paths {
let Some(stem) = path.file_stem().and_then(|s| s.to_str()) else {
continue;
};
let Some(rest) = stem.strip_prefix(&child_prefix) else {
continue;
};
// Direct child only — no further `__` in the remainder.
if rest.contains("__") {
continue;
}
let display = match transcript::read_transcript_display(path) {
Ok(d) => d,
Err(err) => {
log::warn!(
"{LOG_PREFIX} failed to read sub-agent transcript {}: {err}",
path.display()
);
continue;
}
};
let mut items = project_records(&display.records);
items.extend(build_subagent_items(
all_sub_paths,
stem,
depth + 1,
segments,
));
// Prefer the archetype id from meta; fall back to the stem suffix.
let id = if display.meta.agent_name.is_empty() {
rest.to_string()
} else {
display.meta.agent_name.clone()
};
// Anchor to the parent turn active at the sub-agent's spawn time.
let request_id = anchor_request_id(child_spawn_unix(rest), segments);
log::debug!("{LOG_PREFIX} subagent id={id} stem={rest} anchored request_id={request_id:?}");
out.push(DisplayItem::Subagent {
id,
request_id,
items,
});
}
out
}
/// The root turns' start timestamps as `(request_id, unix_seconds)` in file
/// order — one entry per turn boundary. Built from the first timestamped line
/// of each `request_id` run. Turns whose lines carry no `request_id` or no
/// parseable timestamp contribute nothing (legacy/CLI transcripts yield an
/// empty list, so sub-agents there stay unanchored).
fn turn_segments(records: &[DisplayRecord]) -> Vec<(String, i64)> {
let mut segments: Vec<(String, i64)> = Vec::new();
let mut last_request_id: Option<String> = None;
for record in records {
let DisplayRecord::Message(msg) = record else {
continue;
};
let (Some(rid), Some(ts)) = (msg.request_id.as_deref(), msg.ts.as_deref()) else {
continue;
};
if last_request_id.as_deref() == Some(rid) {
continue;
}
let Some(unix) = parse_rfc3339_unix(ts) else {
continue;
};
segments.push((rid.to_string(), unix));
last_request_id = Some(rid.to_string());
}
segments
}
/// Extract a sub-agent's spawn unix timestamp (seconds) from its file-stem
/// suffix. Stems are `{unix_ts}_{agent_id}`; the leading integer is the agent
/// build/spawn time (see the transcript module's stem docs). `None` for
/// non-numeric legacy stems.
fn child_spawn_unix(stem_suffix: &str) -> Option<i64> {
stem_suffix
.split('_')
.next()
.and_then(|s| s.parse::<i64>().ok())
}
/// Parse an RFC-3339 timestamp into unix seconds.
fn parse_rfc3339_unix(ts: &str) -> Option<i64> {
chrono::DateTime::parse_from_rfc3339(ts)
.ok()
.map(|dt| dt.timestamp())
}
/// Anchor a sub-agent to the parent turn that was active at its `child_unix`
/// spawn time: the last turn segment whose start is `<= child_unix`.
///
/// Fallbacks (documented heuristic, since sub-agent files carry no explicit
/// delegation-call back-link):
/// - No segments (legacy/CLI root): `None` — the item stays unanchored and the
/// frontend leaves it under the current turn cursor, as before.
/// - Unknown spawn time (non-numeric stem): the newest turn (best effort).
/// - Spawn time precedes every turn start: the first turn.
fn anchor_request_id(child_unix: Option<i64>, segments: &[(String, i64)]) -> Option<String> {
if segments.is_empty() {
return None;
}
let Some(child_unix) = child_unix else {
return segments.last().map(|(rid, _)| rid.clone());
};
let mut chosen = &segments[0];
for seg in segments {
if seg.1 <= child_unix {
chosen = seg;
}
}
Some(chosen.0.clone())
}
/// Project one file's display records into display items, in file order.
///
/// - System lines are dropped (they carry the tool-policy preamble and other
/// scaffolding that must never render as a chat item).
/// - `reasoning_content` on an assistant line becomes a [`DisplayItem::Reasoning`]
/// preceding its message.
/// - Assistant `tool_calls` register pending [`DisplayItem::ToolCall`]s; a later
/// `role:"tool"` line pairs to one by id, falling back to FIFO order.
/// - Interrupted partials and compaction markers pass through as their items.
/// - A [`DisplayItem::TurnBoundary`] is emitted whenever `request_id` changes.
pub fn project_records(records: &[DisplayRecord]) -> Vec<DisplayItem> {
let mut items: Vec<DisplayItem> = Vec::new();
// Pending tool calls awaiting a result line: (call_id, index into `items`).
let mut pending: VecDeque<(String, usize)> = VecDeque::new();
let mut last_request_id: Option<String> = None;
for record in records {
match record {
DisplayRecord::Message(msg) => {
maybe_emit_turn_boundary(msg, &mut last_request_id, &mut items);
project_message(msg, &mut items, &mut pending);
}
DisplayRecord::Compaction(marker) => {
project_compaction(marker, &mut items);
// A compaction supersedes prior context; drop stale pending
// pairings so a post-compaction result never binds to them.
pending.clear();
}
}
}
items
}
fn maybe_emit_turn_boundary(
msg: &DisplayMessage,
last_request_id: &mut Option<String>,
items: &mut Vec<DisplayItem>,
) {
let Some(rid) = msg.request_id.as_deref() else {
return;
};
if last_request_id.as_deref() != Some(rid) {
items.push(DisplayItem::TurnBoundary {
request_id: rid.to_string(),
});
*last_request_id = Some(rid.to_string());
}
}
fn project_message(
msg: &DisplayMessage,
items: &mut Vec<DisplayItem>,
pending: &mut VecDeque<(String, usize)>,
) {
// Interrupted partial: display-only, carries its own thinking.
if msg.interrupted {
items.push(DisplayItem::InterruptedPartial {
text: msg.message.content.clone(),
thinking: msg.reasoning_content.clone(),
});
return;
}
match msg.message.role.as_str() {
"system" => {
// Scaffolding (tool-policy preamble, etc.) — never a display item.
log::debug!("{LOG_PREFIX} sanitize: dropped system line from projection");
}
"user" => {
let raw = msg.message.content.clone();
let sanitized = sanitize_user_content(&raw);
if sanitized.is_some() {
log::debug!("{LOG_PREFIX} sanitize: stripped injected prefix from user message");
}
items.push(DisplayItem::UserMessage {
content: raw,
display_content: sanitized,
request_id: msg.request_id.clone(),
});
}
"assistant" => project_assistant(msg, items, pending),
"tool" => project_tool_result(msg, items, pending),
other => {
log::debug!("{LOG_PREFIX} projecting unknown role {other:?} as assistant message");
items.push(DisplayItem::AssistantMessage {
content: msg.message.content.clone(),
interim: false,
request_id: msg.request_id.clone(),
model: msg.turn_usage.as_ref().map(|tu| tu.model.clone()),
iteration: msg.iteration,
});
}
}
}
fn project_assistant(
msg: &DisplayMessage,
items: &mut Vec<DisplayItem>,
pending: &mut VecDeque<(String, usize)>,
) {
// Reasoning precedes the message it belongs to.
if let Some(reasoning) = msg.reasoning_content.as_deref() {
if !reasoning.trim().is_empty() {
items.push(DisplayItem::Reasoning {
text: reasoning.to_string(),
});
}
}
let tool_calls = msg
.turn_usage
.as_ref()
.map(|tu| tu.tool_calls.as_slice())
.unwrap_or_default();
let interim = !tool_calls.is_empty();
// The assistant's prose (if any) shows before its tool calls.
if !msg.message.content.trim().is_empty() {
items.push(DisplayItem::AssistantMessage {
content: msg.message.content.clone(),
interim,
request_id: msg.request_id.clone(),
model: msg.turn_usage.as_ref().map(|tu| tu.model.clone()),
iteration: msg.iteration,
});
}
for call in tool_calls {
let args = parse_tool_args(&call.arguments);
items.push(DisplayItem::ToolCall {
call_id: call.id.clone(),
name: call.name.clone(),
args,
result: None,
status: ToolCallStatus::Running,
failure: None,
});
pending.push_back((call.id.clone(), items.len() - 1));
}
}
fn project_tool_result(
msg: &DisplayMessage,
items: &mut Vec<DisplayItem>,
pending: &mut VecDeque<(String, usize)>,
) {
let result = msg.message.content.clone();
// A failed tool line (`ToolResult::is_error`, stamped at persistence) pairs
// to an error row with a failure payload instead of a false success.
let (status, failure) = if msg.failure {
(
ToolCallStatus::Error,
Some(ToolCallFailure {
detail: msg.failure_detail.clone(),
}),
)
} else {
(ToolCallStatus::Success, None)
};
// Pair by explicit call id first, else FIFO.
let idx = msg
.message
.id
.as_deref()
.and_then(|id| take_pending_by_id(pending, id))
.or_else(|| pending.pop_front().map(|(_, idx)| idx));
if let Some(idx) = idx {
if let Some(DisplayItem::ToolCall {
result: slot,
status: status_slot,
failure: failure_slot,
..
}) = items.get_mut(idx)
{
*slot = Some(result);
*status_slot = status;
*failure_slot = failure;
return;
}
}
// Orphan result (no matching assistant tool_call recorded) — surface it as
// a best-effort completed tool row so the output is not lost.
log::debug!("{LOG_PREFIX} tool result with no pending call — emitting orphan tool row");
items.push(DisplayItem::ToolCall {
call_id: msg.message.id.clone().unwrap_or_default(),
name: "tool".to_string(),
args: None,
result: Some(result),
status,
failure,
});
}
/// Remove and return the pending entry whose call id matches `id`, if any.
fn take_pending_by_id(pending: &mut VecDeque<(String, usize)>, id: &str) -> Option<usize> {
let pos = pending.iter().position(|(cid, _)| cid == id)?;
pending.remove(pos).map(|(_, idx)| idx)
}
fn project_compaction(marker: &CompactionMarker, items: &mut Vec<DisplayItem>) {
items.push(DisplayItem::Compaction {
replaced_count: 0,
kept_count: marker.replacement.len(),
ts: marker.ts.clone(),
request_id: marker.request_id.clone(),
});
}
/// Parse a tool call's raw argument string into JSON when possible; a
/// non-JSON string is wrapped so the frontend still receives structured args.
fn parse_tool_args(raw: &str) -> Option<serde_json::Value> {
let trimmed = raw.trim();
if trimmed.is_empty() {
return None;
}
match serde_json::from_str::<serde_json::Value>(trimmed) {
Ok(value) => Some(value),
Err(_) => Some(serde_json::Value::String(trimmed.to_string())),
}
}
/// Strip the injected scaffolding prefix from a user message, returning the
/// sanitized body only when a prefix was actually present (so the caller can
/// tag rather than mutate — the raw `content` is preserved alongside).
fn sanitize_user_content(content: &str) -> Option<String> {
let trimmed_start = content.trim_start();
if trimmed_start.starts_with(DATETIME_PREFIX)
|| trimmed_start.starts_with(CHANNEL_CONTEXT_PREFIX)
{
// The injector prepends the scaffolding line followed by a blank line,
// then the user's actual text. Strip the first paragraph.
if let Some(idx) = content.find("\n\n") {
let body = content[idx + 2..].to_string();
return Some(body);
}
// No body after the prefix — the whole message was scaffolding.
return Some(String::new());
}
None
}
@@ -0,0 +1,435 @@
//! Projection + pagination + sanitization tests for the transcript view.
use super::project::{project_records, project_thread};
use super::types::{DisplayItem, ToolCallStatus};
use super::{get_page, DEFAULT_LIMIT};
use crate::openhuman::agent::harness::session::transcript::{self, read_transcript_display};
use crate::openhuman::inference::provider::ChatMessage;
use std::path::{Path, PathBuf};
use tempfile::TempDir;
fn meta_line(thread_id: &str) -> String {
format!(
r#"{{"_meta":{{"version":1,"agent":"orchestrator","dispatcher":"native","created":"2026-07-21T00:00:00Z","updated":"2026-07-21T00:00:10Z","turn_count":1,"input_tokens":30,"output_tokens":13,"cached_input_tokens":0,"charged_amount_usd":0.003,"thread_id":"{thread_id}"}}}}"#
)
}
/// Write a raw JSONL transcript (meta header + given body lines) into
/// `session_raw/{stem}.jsonl` and return the path.
fn write_raw(workspace: &Path, stem: &str, thread_id: &str, body: &[&str]) -> PathBuf {
let path = transcript::resolve_keyed_transcript_path(workspace, stem).expect("resolve");
let mut buf = meta_line(thread_id);
buf.push('\n');
for line in body {
buf.push_str(line);
buf.push('\n');
}
std::fs::write(&path, buf).expect("write raw transcript");
path
}
/// A full turn: system scaffolding, a user prompt with the injected datetime
/// prefix, an assistant tool-calling step (reasoning + tool_calls), a tool
/// result, then the final assistant answer.
fn full_turn_body() -> Vec<&'static str> {
vec![
r#"{"role":"system","content":"[tool-policy preamble] you may use tools ..."}"#,
r#"{"role":"user","content":"Current Date & Time: 2026-07-21 09:00:00 UTC\n\nWhat's the weather in NYC?","request_id":"req-1"}"#,
r#"{"role":"assistant","content":"Let me check.","provider":"anthropic","model":"claude-x","usage":{"input":10,"output":5,"cached_input":0,"cost_usd":0.001},"ts":"2026-07-21T09:00:01Z","reasoning_content":"I should call the weather tool.","tool_calls":[{"id":"call-1","name":"get_weather","arguments":"{\"city\":\"NYC\"}"}],"iteration":1,"request_id":"req-1"}"#,
r#"{"role":"tool","content":"72F and sunny","id":"call-1","request_id":"req-1"}"#,
r#"{"role":"assistant","content":"It's 72F and sunny in NYC.","provider":"anthropic","model":"claude-x","usage":{"input":20,"output":8,"cached_input":0,"cost_usd":0.002},"ts":"2026-07-21T09:00:02Z","iteration":2,"request_id":"req-1"}"#,
]
}
#[test]
fn projects_turn_with_tools_reasoning_and_sanitization() {
let dir = TempDir::new().unwrap();
let path = write_raw(dir.path(), "100_orchestrator", "thr_w", &full_turn_body());
let display = read_transcript_display(&path).unwrap();
let items = project_records(&display.records);
// Expected order: turnBoundary, userMessage, reasoning, assistant(interim),
// toolCall(paired), assistant(final). System line dropped.
assert_eq!(items.len(), 6, "unexpected items: {items:#?}");
match &items[0] {
DisplayItem::TurnBoundary { request_id } => assert_eq!(request_id, "req-1"),
other => panic!("expected turnBoundary, got {other:?}"),
}
match &items[1] {
DisplayItem::UserMessage {
content,
display_content,
request_id,
} => {
assert!(content.starts_with("Current Date & Time:"), "raw kept");
assert_eq!(
display_content.as_deref(),
Some("What's the weather in NYC?"),
"datetime prefix stripped into displayContent"
);
assert_eq!(request_id.as_deref(), Some("req-1"));
}
other => panic!("expected userMessage, got {other:?}"),
}
match &items[2] {
DisplayItem::Reasoning { text } => assert_eq!(text, "I should call the weather tool."),
other => panic!("expected reasoning, got {other:?}"),
}
match &items[3] {
DisplayItem::AssistantMessage {
content, interim, ..
} => {
assert_eq!(content, "Let me check.");
assert!(*interim, "tool-calling assistant step is interim");
}
other => panic!("expected interim assistantMessage, got {other:?}"),
}
match &items[4] {
DisplayItem::ToolCall {
call_id,
name,
args,
result,
status,
failure,
} => {
assert_eq!(call_id, "call-1");
assert_eq!(name, "get_weather");
assert_eq!(
args.as_ref()
.and_then(|v| v.get("city"))
.and_then(|v| v.as_str()),
Some("NYC")
);
assert_eq!(result.as_deref(), Some("72F and sunny"), "paired by id");
assert_eq!(*status, ToolCallStatus::Success);
assert!(failure.is_none(), "successful tool carries no failure");
}
other => panic!("expected toolCall, got {other:?}"),
}
match &items[5] {
DisplayItem::AssistantMessage {
content, interim, ..
} => {
assert_eq!(content, "It's 72F and sunny in NYC.");
assert!(!*interim, "final answer is not interim");
}
other => panic!("expected final assistantMessage, got {other:?}"),
}
}
#[test]
fn projects_compaction_and_interrupted_partial() {
let dir = TempDir::new().unwrap();
let body = vec![
r#"{"role":"user","content":"hi","request_id":"req-1"}"#,
r#"{"kind":"compaction","replacement":[{"role":"user","content":"summary so far"}],"ts":"2026-07-21T09:05:00Z","request_id":"req-2"}"#,
r#"{"role":"assistant","content":"partial ans","interrupted":true,"reasoning_content":"mid-thought","iteration":3,"request_id":"req-2"}"#,
];
let path = write_raw(dir.path(), "200_orchestrator", "thr_c", &body);
let display = read_transcript_display(&path).unwrap();
let items = project_records(&display.records);
let has_compaction = items.iter().any(|i| {
matches!(
i,
DisplayItem::Compaction { kept_count, .. } if *kept_count == 1
)
});
assert!(has_compaction, "compaction projected: {items:#?}");
let partial = items
.iter()
.find_map(|i| match i {
DisplayItem::InterruptedPartial { text, thinking } => Some((text, thinking)),
_ => None,
})
.expect("interrupted partial projected");
assert_eq!(partial.0, "partial ans");
assert_eq!(partial.1.as_deref(), Some("mid-thought"));
}
#[test]
fn legacy_file_without_version_or_request_id_projects() {
let dir = TempDir::new().unwrap();
// Legacy meta: no `version`, messages carry no `request_id`.
let path = transcript::resolve_keyed_transcript_path(dir.path(), "300_orchestrator").unwrap();
let raw = concat!(
r#"{"_meta":{"agent":"orchestrator","dispatcher":"native","created":"2026-01-01T00:00:00Z","updated":"2026-01-01T00:00:00Z","turn_count":1,"input_tokens":0,"output_tokens":0,"cached_input_tokens":0,"charged_amount_usd":0.0,"thread_id":"thr_legacy"}}"#,
"\n",
r#"{"role":"user","content":"plain question"}"#,
"\n",
r#"{"role":"assistant","content":"plain answer"}"#,
"\n",
);
std::fs::write(&path, raw).unwrap();
let display = read_transcript_display(&path).unwrap();
let items = project_records(&display.records);
// No request_id → no turn boundary; user + assistant still project, and an
// un-prefixed user message keeps no displayContent (nothing to strip).
assert!(
!items
.iter()
.any(|i| matches!(i, DisplayItem::TurnBoundary { .. })),
"legacy lines have no request_id, so no boundary"
);
match items.first() {
Some(DisplayItem::UserMessage {
content,
display_content,
..
}) => {
assert_eq!(content, "plain question");
assert_eq!(
display_content.as_deref(),
None,
"no prefix, no displayContent"
);
}
other => panic!("expected userMessage first, got {other:?}"),
}
assert!(items.iter().any(
|i| matches!(i, DisplayItem::AssistantMessage { content, .. } if content == "plain answer")
));
}
#[test]
fn subagent_file_projects_as_nested_item() {
let dir = TempDir::new().unwrap();
let root_stem = "400_orchestrator";
write_raw(
dir.path(),
root_stem,
"thr_s",
&[r#"{"role":"user","content":"delegate please","request_id":"req-1"}"#],
);
// Sub-agent sibling shares the root stem with a `__` suffix.
write_raw(
dir.path(),
&format!("{root_stem}__100_coder"),
"thr_s",
&[
r#"{"role":"assistant","content":"sub work done","provider":"anthropic","model":"claude-x","usage":{"input":5,"output":3,"cached_input":0,"cost_usd":0.0},"ts":"2026-07-21T09:10:00Z","iteration":1}"#,
],
);
let projected = project_thread(dir.path(), "thr_s").expect("project thread");
let subagent = projected
.items
.iter()
.find_map(|i| match i {
DisplayItem::Subagent { id, items, .. } => Some((id, items)),
_ => None,
})
.expect("subagent item present");
assert_eq!(subagent.0, "orchestrator");
assert!(subagent.1.iter().any(
|i| matches!(i, DisplayItem::AssistantMessage { content, .. } if content == "sub work done")
));
}
#[test]
fn get_page_paginates_newest_first_with_cursor() {
let dir = TempDir::new().unwrap();
// Five plain user messages → five top-level items.
let body: Vec<String> = (0..5)
.map(|i| format!(r#"{{"role":"user","content":"msg-{i}"}}"#))
.collect();
let body_refs: Vec<&str> = body.iter().map(String::as_str).collect();
write_raw(dir.path(), "500_orchestrator", "thr_p", &body_refs);
// First page: newest first, limit 2 → msg-4, msg-3.
let page1 = get_page(dir.path(), "thr_p", None, Some(2));
assert_eq!(page1.total, 5);
assert!(page1.has_more);
assert_eq!(page1.items.len(), 2);
assert!(
matches!(&page1.items[0], DisplayItem::UserMessage { content, .. } if content == "msg-4")
);
assert!(
matches!(&page1.items[1], DisplayItem::UserMessage { content, .. } if content == "msg-3")
);
let cursor = page1.next_cursor.clone().expect("next cursor");
let page2 = get_page(dir.path(), "thr_p", Some(&cursor), Some(2));
assert!(
matches!(&page2.items[0], DisplayItem::UserMessage { content, .. } if content == "msg-2")
);
// Walk to the end.
let last = get_page(dir.path(), "thr_p", page2.next_cursor.as_deref(), Some(2));
assert!(!last.has_more, "final page exhausts the thread");
assert!(last.next_cursor.is_none());
}
#[test]
fn failed_tool_line_projects_error_status_with_failure_payload() {
let dir = TempDir::new().unwrap();
// Assistant issues a tool call; the paired tool result line carries the
// additive `failure` flag (stamped at persistence from `is_error`).
let body = vec![
r#"{"role":"assistant","content":"trying","provider":"anthropic","model":"m","usage":{"input":1,"output":1,"cached_input":0,"cost_usd":0.0},"ts":"2026-07-21T09:00:01Z","tool_calls":[{"id":"call-9","name":"shell","arguments":"{\"cmd\":\"boom\"}"}],"iteration":1,"request_id":"req-1"}"#,
r#"{"role":"tool","content":"error: command not found","id":"call-9","request_id":"req-1","failure":true,"failure_detail":"error: command not found"}"#,
];
let path = write_raw(dir.path(), "600_orchestrator", "thr_f", &body);
let display = read_transcript_display(&path).unwrap();
let items = project_records(&display.records);
let tool = items
.iter()
.find_map(|i| match i {
DisplayItem::ToolCall {
status, failure, ..
} => Some((status, failure)),
_ => None,
})
.expect("toolCall projected");
assert_eq!(*tool.0, ToolCallStatus::Error, "failed tool → error status");
let failure = tool.1.as_ref().expect("failure payload present");
assert_eq!(failure.detail.as_deref(), Some("error: command not found"));
}
#[test]
fn tool_failure_metadata_round_trips_write_to_display_line() {
// Full write path: a failed tool ChatMessage stamped with failure metadata
// must serialise the additive `failure` line field and read back as a failed
// display message — proving the harness → transcript → projection seam.
let dir = TempDir::new().unwrap();
let now = "2026-07-21T09:00:00Z".to_string();
let meta = transcript::TranscriptMeta {
agent_name: "orchestrator".into(),
agent_id: Some("orchestrator".into()),
agent_type: Some("root".into()),
dispatcher: "native".into(),
provider: Some("anthropic".into()),
model: Some("m".into()),
created: now.clone(),
updated: now,
turn_count: 1,
input_tokens: 0,
output_tokens: 0,
cached_input_tokens: 0,
charged_amount_usd: 0.0,
thread_id: Some("thr_rt".into()),
task_id: None,
};
let mut tool_msg = ChatMessage {
id: Some("call-1".into()),
role: "tool".into(),
content: r#"{"tool_call_id":"call-1","content":"boom"}"#.into(),
extra_metadata: None,
};
transcript::attach_tool_failure_metadata(&mut tool_msg, Some("boom: exit 1"));
let messages = vec![
ChatMessage {
id: None,
role: "user".into(),
content: "do it".into(),
extra_metadata: None,
},
tool_msg,
];
let path = transcript::resolve_keyed_transcript_path(dir.path(), "700_orchestrator").unwrap();
transcript::write_transcript(&path, &messages, &meta, None).unwrap();
let display = read_transcript_display(&path).unwrap();
let failed = display
.records
.iter()
.find_map(|r| match r {
transcript::DisplayRecord::Message(m) if m.message.role == "tool" => Some(m),
_ => None,
})
.expect("tool display message present");
assert!(
failed.failure,
"failure flag survived the write/read round trip"
);
assert_eq!(failed.failure_detail.as_deref(), Some("boom: exit 1"));
}
#[test]
fn subagent_anchors_to_parent_turn_by_spawn_timestamp() {
let dir = TempDir::new().unwrap();
let root_stem = "800_orchestrator";
let thread_id = "thr_anchor";
let t1 = chrono::DateTime::from_timestamp(1_000_000, 0)
.unwrap()
.to_rfc3339();
let t2 = chrono::DateTime::from_timestamp(2_000_000, 0)
.unwrap()
.to_rfc3339();
// Two turns: req-1 (assistant ts t1), req-2 (assistant ts t2).
let root_body = vec![
r#"{"role":"user","content":"one","request_id":"req-1"}"#.to_string(),
format!(
r#"{{"role":"assistant","content":"a1","provider":"anthropic","model":"m","usage":{{"input":1,"output":1,"cached_input":0,"cost_usd":0.0}},"ts":"{t1}","iteration":1,"request_id":"req-1"}}"#
),
r#"{"role":"user","content":"two","request_id":"req-2"}"#.to_string(),
format!(
r#"{{"role":"assistant","content":"a2","provider":"anthropic","model":"m","usage":{{"input":1,"output":1,"cached_input":0,"cost_usd":0.0}},"ts":"{t2}","iteration":1,"request_id":"req-2"}}"#
),
];
let root_refs: Vec<&str> = root_body.iter().map(String::as_str).collect();
write_raw(dir.path(), root_stem, thread_id, &root_refs);
// Sub-agent stems encode the spawn unix timestamp: coder spawned during
// turn 1 (1_000_050), planner during turn 2 (2_000_050).
write_raw(
dir.path(),
&format!("{root_stem}__1000050_coder"),
thread_id,
&[r#"{"role":"assistant","content":"coder work"}"#],
);
write_raw(
dir.path(),
&format!("{root_stem}__2000050_planner"),
thread_id,
&[r#"{"role":"assistant","content":"planner work"}"#],
);
let projected = project_thread(dir.path(), thread_id).expect("project thread");
// The seeded sub-agent files share the `orchestrator` meta agent name, so
// key the anchoring by each sub-agent's inner work content instead of `id`.
let mut anchors: Vec<(String, Option<String>)> = projected
.items
.iter()
.filter_map(|i| match i {
DisplayItem::Subagent {
request_id, items, ..
} => {
let marker = items.iter().find_map(|inner| match inner {
DisplayItem::AssistantMessage { content, .. } => Some(content.clone()),
_ => None,
})?;
Some((marker, request_id.clone()))
}
_ => None,
})
.collect();
anchors.sort();
assert_eq!(
anchors,
vec![
("coder work".to_string(), Some("req-1".to_string())),
("planner work".to_string(), Some("req-2".to_string())),
],
"each sub-agent anchors to the turn active at its spawn time"
);
}
#[test]
fn get_page_missing_thread_is_empty_not_error() {
let dir = TempDir::new().unwrap();
let page = get_page(dir.path(), "no_such_thread", None, Some(DEFAULT_LIMIT));
assert!(!page.has_transcript);
assert_eq!(page.total, 0);
assert!(page.items.is_empty());
}
@@ -0,0 +1,139 @@
//! Typed display items for the transcript projection RPC
//! (`threads.transcript_get`).
//!
//! These mirror the frontend's existing chat vocabulary (user/assistant
//! bubbles, reasoning drawer, tool timeline rows, sub-agent activity) so the
//! Phase C renderer can map them onto the same components. Serde is camelCase
//! on the wire — the frontend reads `displayContent`, `callId`, `requestId`,
//! etc.
use serde::Serialize;
/// Terminal state of a projected tool call. Mirrors the live timeline's
/// `ToolTimelineStatus` vocabulary (`running` / `success` / `error`) so the
/// settled projection and the live stream render identically.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum ToolCallStatus {
/// Call issued but no result line has been paired yet.
Running,
/// A result line was paired to the call.
Success,
/// A result line the projection identified as a **failure**: the persisted
/// tool line carried the additive `failure` flag (stamped at turn-loop
/// persistence from the tool's `ToolResult::is_error` outcome). Paired with
/// a [`ToolCallFailure`] payload on the item.
Error,
}
/// Failure payload attached to an errored [`DisplayItem::ToolCall`]. Minimal by
/// design: the persisted transcript only records that the call failed plus an
/// optional short reason. The frontend mapper expands this into its richer
/// `ToolFailureExplanation` shape (`class` / `category` / `causePlain` /
/// `nextAction`) for the `ToolFailureLines` renderer.
#[derive(Debug, Clone, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct ToolCallFailure {
/// Short, single-line reason for the failure, when the writer captured one.
#[serde(skip_serializing_if = "Option::is_none")]
pub detail: Option<String>,
}
/// One item in a projected transcript, in the frontend's display vocabulary.
///
/// `#[serde(tag = "kind")]` gives each variant a camelCase discriminator
/// (`userMessage`, `assistantMessage`, …) and every field is camelCase.
#[derive(Debug, Clone, PartialEq, Serialize)]
#[serde(
tag = "kind",
rename_all = "camelCase",
rename_all_fields = "camelCase"
)]
pub enum DisplayItem {
/// A user prompt. `content` is the raw persisted content (may carry the
/// injected `Current Date & Time:` scaffolding line); `displayContent` is
/// the sanitized version to show, present only when it differs from raw.
UserMessage {
content: String,
#[serde(skip_serializing_if = "Option::is_none")]
display_content: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
request_id: Option<String>,
},
/// An assistant answer. `interim: true` marks a non-terminal tool-calling
/// step within a multi-iteration turn (not the final answer bubble).
AssistantMessage {
content: String,
#[serde(default, skip_serializing_if = "is_false")]
interim: bool,
#[serde(skip_serializing_if = "Option::is_none")]
request_id: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
model: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
iteration: Option<u32>,
},
/// The model's reasoning/thinking that preceded an assistant message.
Reasoning { text: String },
/// A tool invocation with its paired result, when available.
ToolCall {
call_id: String,
name: String,
#[serde(skip_serializing_if = "Option::is_none")]
args: Option<serde_json::Value>,
#[serde(skip_serializing_if = "Option::is_none")]
result: Option<String>,
status: ToolCallStatus,
/// Present only when `status` is `Error` — the failure payload the
/// frontend expands for the `ToolFailureLines` renderer.
#[serde(skip_serializing_if = "Option::is_none")]
failure: Option<ToolCallFailure>,
},
/// A delegated sub-agent run, with its own nested projected items.
///
/// `request_id` anchors the whole sub-agent trail to the parent turn that
/// spawned it. Sub-agent transcripts are sibling files with no explicit
/// back-link to the delegating tool call, so the projection derives this by
/// matching the sub-agent's spawn timestamp (encoded in its file stem)
/// against the parent turns' timestamp ranges (see
/// `project::anchor_request_id`). Absent for legacy/CLI transcripts whose
/// lines carry no `request_id`.
Subagent {
id: String,
#[serde(skip_serializing_if = "Option::is_none")]
request_id: Option<String>,
items: Vec<DisplayItem>,
},
/// A turn boundary — emitted when the `request_id` changes between lines.
TurnBoundary { request_id: String },
/// A partial assistant answer captured when a turn was interrupted.
InterruptedPartial {
text: String,
#[serde(skip_serializing_if = "Option::is_none")]
thinking: Option<String>,
},
/// A context-compaction marker: the reduced set replaced everything before
/// it. Counts describe what the record superseded/installed.
Compaction {
replaced_count: usize,
kept_count: usize,
#[serde(skip_serializing_if = "Option::is_none")]
ts: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
request_id: Option<String>,
},
}
#[allow(clippy::trivially_copy_pass_by_ref)]
fn is_false(b: &bool) -> bool {
!*b
}
/// A projected transcript for one thread, before pagination. Chronological
/// (file) order; the RPC layer paginates newest-first.
#[derive(Debug, Clone)]
pub struct ProjectedTranscript {
pub thread_id: String,
/// All top-level display items in chronological order.
pub items: Vec<DisplayItem>,
}
@@ -558,6 +558,69 @@ impl TurnStateMirror {
self.state.active_subagent = None;
self.state.updated_at = chrono::Utc::now().to_rfc3339();
self.flush();
self.persist_interrupted_partial();
}
/// Append the partial streamed answer of an interrupted turn to the session
/// transcript so the derived display view (Phase B) can surface it even
/// after the live turn_state snapshot is gone. Display-only: the
/// model-context reader skips `interrupted:true` lines.
///
/// Guard: the root transcript file must already exist. An interrupted
/// **first** turn has no session file yet (the harness has not persisted a
/// turn), so there is nothing to append to — that case stays recoverable
/// from the turn_state snapshot alone, as today. We log and skip it.
fn persist_interrupted_partial(&self) {
let partial = self.state.streaming_text.trim();
if partial.is_empty() {
return;
}
let thread_id = self.state.thread_id.trim();
if thread_id.is_empty() {
return;
}
let workspace_dir = self.store.workspace_dir();
let Some(path) =
crate::openhuman::agent::harness::session::transcript::find_root_transcript_for_thread(
workspace_dir,
thread_id,
)
else {
log::debug!(
"{MIRROR_LOG_PREFIX} no root transcript for thread={thread_id} yet — leaving interrupted partial ({} chars) in turn_state snapshot only",
partial.len()
);
return;
};
let request_id = if self.state.request_id.is_empty() {
None
} else {
Some(self.state.request_id.as_str())
};
let thinking = self.state.thinking.trim();
let reasoning = if thinking.is_empty() {
None
} else {
Some(thinking)
};
match crate::openhuman::agent::harness::session::transcript::append_interrupted_partial(
&path,
partial,
request_id,
Some(self.state.iteration),
reasoning,
) {
Ok(()) => log::debug!(
"{MIRROR_LOG_PREFIX} appended interrupted partial ({} chars, thinking={} chars) for thread={thread_id} request_id={} to {}",
partial.len(),
thinking.len(),
self.state.request_id,
path.display()
),
Err(err) => log::warn!(
"{MIRROR_LOG_PREFIX} failed to append interrupted partial for thread={thread_id}: {err}"
),
}
}
fn flush(&mut self) {
@@ -674,3 +674,138 @@ fn subagent_transcript_persists_interleaved_prose_and_tools() {
"no snake_case fields on the wire"
);
}
// ── Interrupted-partial → session transcript wiring (Task 1) ──────────
use crate::openhuman::agent::harness::session::transcript::{
self, read_transcript, read_transcript_display, DisplayRecord, TranscriptMeta,
};
use crate::openhuman::inference::provider::ChatMessage;
fn seed_root_transcript(workspace: &std::path::Path, thread_id: &str) -> std::path::PathBuf {
let stem = "100_orchestrator".to_string();
let path = transcript::resolve_keyed_transcript_path(workspace, &stem).expect("resolve path");
let meta = TranscriptMeta {
agent_name: "orchestrator".into(),
agent_id: None,
agent_type: Some("root".into()),
dispatcher: "native".into(),
provider: None,
model: None,
created: "2026-07-21T00:00:00Z".into(),
updated: "2026-07-21T00:00:00Z".into(),
turn_count: 1,
input_tokens: 0,
output_tokens: 0,
cached_input_tokens: 0,
charged_amount_usd: 0.0,
thread_id: Some(thread_id.to_string()),
task_id: None,
};
transcript::write_transcript(&path, &[ChatMessage::user("hello there")], &meta, None)
.expect("seed transcript");
path
}
/// When a streaming turn is interrupted and a root transcript already exists,
/// `finish()` appends the partial streamed answer (display-only) to the file.
#[test]
fn finish_appends_interrupted_partial_to_existing_transcript() {
let dir = tempdir().expect("tempdir");
let thread_id = "thr_abc";
let path = seed_root_transcript(dir.path(), thread_id);
let store = TurnStateStore::new(dir.path().to_path_buf());
let mut m = TurnStateMirror::new(store, thread_id, "req-9");
m.observe(&AgentProgress::IterationStarted {
iteration: 2,
max_iterations: 25,
});
m.observe(&AgentProgress::ThinkingDelta {
delta: "hmm".into(),
iteration: 2,
});
m.observe(&AgentProgress::TextDelta {
delta: "half an ".into(),
iteration: 2,
});
m.observe(&AgentProgress::TextDelta {
delta: "answer".into(),
iteration: 2,
});
// No TurnCompleted — the bridge exits, marking the turn interrupted.
m.finish();
// Model context must NOT carry the partial.
let model = read_transcript(&path).expect("read model context");
assert!(
!model
.messages
.iter()
.any(|msg| msg.content.contains("half an answer")),
"interrupted partial must be excluded from the model context"
);
// Display projection carries the flagged partial with request_id + thinking.
let display = read_transcript_display(&path).expect("read display");
let partial = display
.records
.iter()
.find_map(|r| match r {
DisplayRecord::Message(msg) if msg.interrupted => Some(msg),
_ => None,
})
.expect("display must include the interrupted partial");
assert_eq!(partial.message.content, "half an answer");
assert_eq!(partial.request_id.as_deref(), Some("req-9"));
assert_eq!(partial.iteration, Some(2));
assert_eq!(partial.reasoning_content.as_deref(), Some("hmm"));
}
/// A completed turn never writes an interrupted partial.
#[test]
fn finish_after_completion_writes_no_partial() {
let dir = tempdir().expect("tempdir");
let thread_id = "thr_done";
let path = seed_root_transcript(dir.path(), thread_id);
let store = TurnStateStore::new(dir.path().to_path_buf());
let mut m = TurnStateMirror::new(store, thread_id, "req-done");
m.observe(&AgentProgress::TextDelta {
delta: "final answer".into(),
iteration: 1,
});
m.observe(&AgentProgress::TurnCompleted { iterations: 1 });
m.finish();
let display = read_transcript_display(&path).expect("read display");
assert!(
!display
.records
.iter()
.any(|r| matches!(r, DisplayRecord::Message(msg) if msg.interrupted)),
"a completed turn must not append an interrupted partial"
);
}
/// An interrupted FIRST turn (no root transcript file yet) is a no-op — the
/// partial stays in the turn_state snapshot only, and finish() does not panic.
#[test]
fn finish_first_turn_without_transcript_is_noop() {
let dir = tempdir().expect("tempdir");
let store = TurnStateStore::new(dir.path().to_path_buf());
let mut m = TurnStateMirror::new(store, "thr_new", "req-first");
m.observe(&AgentProgress::TextDelta {
delta: "orphan partial".into(),
iteration: 1,
});
// Must not panic even though no session_raw transcript exists.
m.finish();
// The snapshot itself still records the interrupted turn.
let listed = TurnStateStore::new(dir.path().to_path_buf())
.get("thr_new")
.expect("get")
.expect("snapshot present");
assert_eq!(listed.lifecycle, TurnLifecycle::Interrupted);
assert_eq!(listed.streaming_text, "orphan partial");
}
@@ -54,6 +54,13 @@ impl TurnStateStore {
Self { workspace_dir }
}
/// Workspace root this store persists under. Exposed so the mirror can
/// resolve sibling session transcripts (append the interrupted partial to
/// `session_raw/{root}.jsonl`) without re-plumbing the path.
pub fn workspace_dir(&self) -> &std::path::Path {
&self.workspace_dir
}
/// Atomically write the snapshot for `state.request_id` under
/// `state.thread_id`. On a `Completed` write, prune the thread's completed
/// turns to the newest [`COMPLETED_RETENTION`].
+218
View File
@@ -15398,3 +15398,221 @@ async fn json_rpc_agent_meetings_generate_summary_rejects_empty_meeting_id() {
rpc_join.abort();
}
/// Seed a raw append-only session transcript at
/// `{workspace}/session_raw/{stem}.jsonl` (meta header + body lines).
fn seed_raw_transcript(workspace: &Path, stem: &str, thread_id: &str, body: &[&str]) {
let raw_dir = workspace.join("session_raw");
std::fs::create_dir_all(&raw_dir).expect("create session_raw");
let meta = format!(
r#"{{"_meta":{{"version":1,"agent":"orchestrator","dispatcher":"native","created":"2026-07-21T00:00:00Z","updated":"2026-07-21T00:00:10Z","turn_count":1,"input_tokens":30,"output_tokens":13,"cached_input_tokens":0,"charged_amount_usd":0.003,"thread_id":"{thread_id}"}}}}"#
);
let mut buf = meta;
buf.push('\n');
for line in body {
buf.push_str(line);
buf.push('\n');
}
std::fs::write(raw_dir.join(format!("{stem}.jsonl")), buf).expect("write transcript");
}
#[tokio::test]
async fn json_rpc_threads_transcript_get_projects_and_paginates() {
let _env_lock = json_rpc_e2e_env_lock();
let tmp = tempdir().expect("tempdir");
// Config resolves the runtime workspace to `OPENHUMAN_WORKSPACE/workspace`
// (see resolve_config_dir_for_workspace), so seed transcripts there.
let workspace = tmp.path().join("workspace");
std::fs::create_dir_all(&workspace).expect("create workspace");
let workspace = workspace.as_path();
let _workspace_guard = EnvVarGuard::set_to_path("OPENHUMAN_WORKSPACE", tmp.path());
let _backend_url_guard = EnvVarGuard::unset("BACKEND_URL");
let _vite_backend_url_guard = EnvVarGuard::unset("VITE_BACKEND_URL");
let _api_url_guard = EnvVarGuard::unset("OPENHUMAN_API_URL");
let thread_id = "thr_transcript_e2e";
let root_stem = "1000_orchestrator";
// A full turn: system scaffolding (dropped), user with injected datetime
// prefix, assistant tool-calling step (reasoning + tool_calls), tool result,
// final assistant answer.
seed_raw_transcript(
workspace,
root_stem,
thread_id,
&[
r#"{"role":"system","content":"[tool-policy preamble] ..."}"#,
r#"{"role":"user","content":"Current Date & Time: 2026-07-21 09:00:00 UTC\n\nWeather in NYC?","request_id":"req-1"}"#,
r#"{"role":"assistant","content":"Checking.","provider":"anthropic","model":"claude-x","usage":{"input":10,"output":5,"cached_input":0,"cost_usd":0.001},"ts":"2026-07-21T09:00:01Z","reasoning_content":"call the tool","tool_calls":[{"id":"call-1","name":"get_weather","arguments":"{\"city\":\"NYC\"}"},{"id":"call-2","name":"get_traffic","arguments":"{\"city\":\"NYC\"}"}],"iteration":1,"request_id":"req-1"}"#,
r#"{"role":"tool","content":"72F sunny","id":"call-1","request_id":"req-1"}"#,
r#"{"role":"tool","content":"error: traffic service unavailable","id":"call-2","request_id":"req-1","failure":true,"failure_detail":"traffic service unavailable"}"#,
r#"{"role":"assistant","content":"72F and sunny.","provider":"anthropic","model":"claude-x","usage":{"input":20,"output":8,"cached_input":0,"cost_usd":0.002},"ts":"2026-07-21T09:00:02Z","iteration":2,"request_id":"req-1"}"#,
],
);
// A sub-agent sibling sharing the root stem.
seed_raw_transcript(
workspace,
&format!("{root_stem}__50_coder"),
thread_id,
&[
r#"{"role":"assistant","content":"sub work done","provider":"anthropic","model":"claude-x","usage":{"input":5,"output":3,"cached_input":0,"cost_usd":0.0},"ts":"2026-07-21T09:10:00Z","iteration":1}"#,
],
);
let (rpc_addr, rpc_join) = serve_on_ephemeral(build_core_http_router(false)).await;
let rpc_base = format!("http://{rpc_addr}");
// Full projection (default limit).
let full = post_json_rpc(
&rpc_base,
41_001,
"openhuman.threads_transcript_get",
json!({ "thread_id": thread_id }),
)
.await;
let full_result = assert_no_jsonrpc_error(&full, "threads_transcript_get");
let data = full_result.get("data").expect("data envelope");
assert_eq!(
data.get("threadId").and_then(Value::as_str),
Some(thread_id)
);
assert_eq!(
data.get("hasTranscript").and_then(Value::as_bool),
Some(true)
);
let items = data
.get("items")
.and_then(Value::as_array)
.expect("items array");
// 7 top-level from the root turn (turnBoundary, userMessage, reasoning,
// interim assistant, 2 toolCalls, final assistant) + 1 subagent = 8.
assert_eq!(
data.get("total").and_then(Value::as_u64),
Some(8),
"items: {items:#?}"
);
let kinds: Vec<&str> = items
.iter()
.filter_map(|i| i.get("kind").and_then(Value::as_str))
.collect();
assert!(kinds.contains(&"turnBoundary"));
assert!(kinds.contains(&"userMessage"));
assert!(kinds.contains(&"reasoning"));
assert!(kinds.contains(&"toolCall"));
assert!(kinds.contains(&"subagent"));
// Sanitization: the user message keeps raw content but exposes a stripped
// displayContent.
let user = items
.iter()
.find(|i| i.get("kind").and_then(Value::as_str) == Some("userMessage"))
.expect("userMessage present");
assert_eq!(
user.get("displayContent").and_then(Value::as_str),
Some("Weather in NYC?")
);
// Tool call paired with its result.
let tool = items
.iter()
.find(|i| i.get("callId").and_then(Value::as_str) == Some("call-1"))
.expect("toolCall call-1 present");
assert_eq!(
tool.get("result").and_then(Value::as_str),
Some("72F sunny")
);
assert_eq!(tool.get("status").and_then(Value::as_str), Some("success"));
assert!(
tool.get("failure").is_none(),
"successful tool has no failure"
);
// A failed tool result projects an error status + failure payload rather
// than a false success (Gap 1).
let failed_tool = items
.iter()
.find(|i| i.get("callId").and_then(Value::as_str) == Some("call-2"))
.expect("toolCall call-2 present");
assert_eq!(
failed_tool.get("status").and_then(Value::as_str),
Some("error")
);
assert_eq!(
failed_tool
.get("failure")
.and_then(|f| f.get("detail"))
.and_then(Value::as_str),
Some("traffic service unavailable")
);
// Pagination: newest-first, limit 2 → 2 items + a cursor.
let page1 = post_json_rpc(
&rpc_base,
41_002,
"openhuman.threads_transcript_get",
json!({ "thread_id": thread_id, "limit": 2 }),
)
.await;
let page1_data = assert_no_jsonrpc_error(&page1, "transcript_get page1")
.get("data")
.cloned()
.expect("data");
assert_eq!(
page1_data
.get("items")
.and_then(Value::as_array)
.map(Vec::len),
Some(2)
);
assert_eq!(
page1_data.get("hasMore").and_then(Value::as_bool),
Some(true)
);
let cursor = page1_data
.get("nextCursor")
.and_then(Value::as_str)
.expect("nextCursor present")
.to_string();
let page2 = post_json_rpc(
&rpc_base,
41_003,
"openhuman.threads_transcript_get",
json!({ "thread_id": thread_id, "limit": 2, "cursor": cursor }),
)
.await;
let page2_data = assert_no_jsonrpc_error(&page2, "transcript_get page2")
.get("data")
.cloned()
.expect("data");
assert_eq!(
page2_data
.get("items")
.and_then(Value::as_array)
.map(Vec::len),
Some(2),
"second page returns the next two items"
);
// Missing thread → empty page, not an error.
let missing = post_json_rpc(
&rpc_base,
41_004,
"openhuman.threads_transcript_get",
json!({ "thread_id": "no_such_thread" }),
)
.await;
let missing_data = assert_no_jsonrpc_error(&missing, "transcript_get missing")
.get("data")
.cloned()
.expect("data");
assert_eq!(
missing_data.get("hasTranscript").and_then(Value::as_bool),
Some(false)
);
assert_eq!(missing_data.get("total").and_then(Value::as_u64), Some(0));
rpc_join.abort();
}