feat: stream persisted history and show workspace kind indicators

This commit is contained in:
Mohamed Boudra
2026-03-21 00:12:44 +07:00
parent 48924626df
commit 862cc43db7
6 changed files with 440 additions and 66 deletions
@@ -25,7 +25,7 @@ import { navigateToWorkspace } from '@/hooks/use-workspace-navigation'
import { StyleSheet, UnistylesRuntime, useUnistyles } from 'react-native-unistyles'
import { type GestureType } from 'react-native-gesture-handler'
import * as Clipboard from 'expo-clipboard'
import { Archive, ChevronDown, ChevronRight, Copy, MoreVertical, Plus } from 'lucide-react-native'
import { Archive, Check, ChevronDown, ChevronRight, CircleHelp, Copy, FolderGit2, Monitor, MoreVertical, Plus } from 'lucide-react-native'
import { NestableScrollContainer } from 'react-native-draggable-flatlist'
import { DraggableList, type DraggableRenderItemInfo } from './draggable-list'
import type { DraggableListDragHandleProps } from './draggable-list.types'
@@ -160,18 +160,26 @@ function resolveStatusDotColor(input: {
function WorkspaceStatusIndicator({
bucket,
workspaceKind,
loading = false,
}: {
bucket: SidebarWorkspaceEntry['statusBucket']
workspaceKind: SidebarWorkspaceEntry['workspaceKind']
loading?: boolean
}) {
const { theme } = useUnistyles()
const color = resolveStatusDotColor({ theme, bucket })
const shouldShowSyncedLoader = shouldRenderSyncedStatusLoader({ bucket })
const shouldRenderIdlePlaceholder = !loading && !shouldShowSyncedLoader && bucket === 'done'
const isIdle = !loading && !shouldShowSyncedLoader && bucket === 'done'
if (shouldRenderIdlePlaceholder) {
return null
if (isIdle) {
const KindIcon = workspaceKind === 'local_checkout' ? Monitor : workspaceKind === 'worktree' ? FolderGit2 : null
if (!KindIcon) return null
return (
<View style={styles.workspaceStatusDot}>
<KindIcon size={14} color={theme.colors.foregroundMuted} />
</View>
)
}
return (
@@ -180,6 +188,10 @@ function WorkspaceStatusIndicator({
<ActivityIndicator size={8} color={theme.colors.foregroundMuted} />
) : shouldShowSyncedLoader ? (
<SyncedLoader size={11} color={theme.colors.palette.amber[500]} />
) : bucket === 'needs_input' ? (
<CircleHelp size={14} color={theme.colors.palette.amber[500]} />
) : bucket === 'attention' ? (
<Check size={14} color={theme.colors.palette.green[500]} />
) : (
<View style={[styles.workspaceStatusDotFill, { backgroundColor: color }]} />
)}
@@ -213,7 +225,7 @@ function ProjectLeadingVisual({
{showChevron && chevron !== null ? (
<ProjectInlineChevron chevron={chevron} />
) : shouldShowWorkspaceStatus && activeWorkspace ? (
<WorkspaceStatusIndicator bucket={activeWorkspace.statusBucket} loading={isArchiving} />
<WorkspaceStatusIndicator bucket={activeWorkspace.statusBucket} workspaceKind={activeWorkspace.workspaceKind} loading={isArchiving} />
) : iconDataUri ? (
<Image source={{ uri: iconDataUri }} style={styles.projectIcon} />
) : (
@@ -686,6 +698,7 @@ function WorkspaceRowInner({
>
<WorkspaceStatusIndicator
bucket={workspace.statusBucket}
workspaceKind={workspace.workspaceKind}
loading={isArchiving || isCreating}
/>
<Text
@@ -1903,7 +1916,7 @@ const styles = StyleSheet.create((theme) => ({
workspaceRowLeft: {
flexDirection: 'row',
alignItems: 'center',
gap: theme.spacing[1],
gap: theme.spacing[2],
flex: 1,
minWidth: 0,
},
@@ -1938,7 +1951,7 @@ const styles = StyleSheet.create((theme) => ({
position: 'relative',
},
workspaceStatusDot: {
width: 11,
width: 14,
height: 16,
borderRadius: theme.borderRadius.full,
flexShrink: 0,
@@ -147,6 +147,45 @@ describe("convertClaudeHistoryEntry", () => {
]);
});
test("maps queue-operation task notifications to synthetic tool calls", () => {
const entry = {
type: "queue-operation",
operation: "enqueue",
uuid: "task-note-queue-1",
content: [
"<task-notification>",
"<task-id>bg-queue-1</task-id>",
"<status>completed</status>",
"<summary>Background task completed</summary>",
"<output-file>/tmp/bg-queue-1.txt</output-file>",
"</task-notification>",
].join("\n"),
};
expect(convertClaudeHistoryEntry(entry, () => [])).toEqual([
{
type: "tool_call",
callId: "task_notification_task-note-queue-1",
name: "task_notification",
status: "completed",
error: null,
detail: {
type: "plain_text",
label: "Background task completed",
icon: "wrench",
text: entry.content,
},
metadata: {
synthetic: true,
source: "claude_task_notification",
taskId: "bg-queue-1",
status: "completed",
outputFile: "/tmp/bg-queue-1.txt",
},
},
]);
});
test("passes assistant content blocks through to the mapper", () => {
const entry = {
type: "assistant",
@@ -29,6 +29,7 @@ import {
mapClaudeRunningToolCall,
} from "./claude/tool-call-mapper.js";
import {
coerceTaskNotificationHistoryRecordToSystemMessage,
isTaskNotificationUserContent,
mapTaskNotificationSystemRecordToToolCall,
mapTaskNotificationUserContentToToolCall,
@@ -1495,6 +1496,9 @@ class ClaudeAgentSession implements AgentSession {
private readonly timelineAssembler = new TimelineAssembler();
private persistedHistory: AgentTimelineItem[] = [];
private historyPending = false;
private historyOffsetSessionId: string | null = null;
private historyReadOffsetBytes = 0;
private historyLineFragment = "";
private turnState: TurnState = "idle";
private preReplayMetadataSeen = false;
private pendingAutonomousWakeReservations = 0;
@@ -3074,6 +3078,9 @@ class ClaudeAgentSession implements AgentSession {
this.persistence = null;
this.persistedHistory = [];
this.historyPending = false;
this.historyOffsetSessionId = null;
this.historyReadOffsetBytes = 0;
this.historyLineFragment = "";
this.cachedRuntimeInfo = null;
this.queryRestartNeeded = false;
return true;
@@ -3998,46 +4005,163 @@ class ClaudeAgentSession implements AgentSession {
}
}
private waitForLiveHistoryPoll(): Promise<void> {
return new Promise((resolve) => setTimeout(resolve, 250));
private async waitForLiveHistoryPoll(): Promise<void> {
await new Promise((resolve) => setTimeout(resolve, 250));
if (this.claudeSessionId) {
this.loadPersistedHistory(this.claudeSessionId, { dispatchLive: true });
}
}
private loadPersistedHistory(sessionId: string) {
private loadPersistedHistory(
sessionId: string,
options?: { dispatchLive?: boolean }
) {
try {
const historyPath = this.resolveHistoryPath(sessionId);
if (!historyPath || !fs.existsSync(historyPath)) {
return;
}
const content = fs.readFileSync(historyPath, "utf8");
const timeline: AgentTimelineItem[] = [];
for (const line of content.split(/\n+/)) {
const trimmed = line.trim();
if (!trimmed) continue;
try {
const entry = JSON.parse(trimmed);
if (entry.isSidechain) {
continue;
}
if (entry.type === "user" && typeof entry.uuid === "string") {
this.rememberUserMessageId(entry.uuid);
}
const items = this.convertHistoryEntry(entry);
if (items.length > 0) {
timeline.push(...items);
}
} catch (error) {
// ignore malformed history line
}
if (this.historyOffsetSessionId !== sessionId) {
this.historyOffsetSessionId = sessionId;
this.historyReadOffsetBytes = 0;
this.historyLineFragment = "";
}
if (timeline.length > 0) {
this.persistedHistory = timeline;
this.historyPending = true;
const content = fs.readFileSync(historyPath);
if (content.byteLength < this.historyReadOffsetBytes) {
this.historyReadOffsetBytes = 0;
this.historyLineFragment = "";
}
if (content.byteLength === this.historyReadOffsetBytes) {
return;
}
const unreadChunk = content
.subarray(this.historyReadOffsetBytes)
.toString("utf8");
this.historyReadOffsetBytes = content.byteLength;
this.ingestPersistedHistoryChunk(unreadChunk, {
dispatchLive: options?.dispatchLive ?? false,
});
} catch (error) {
// ignore history load failures
}
}
private ingestPersistedHistoryChunk(
chunk: string,
options: { dispatchLive: boolean }
): void {
if (!chunk) {
return;
}
const combined = `${this.historyLineFragment}${chunk}`;
this.historyLineFragment = "";
const lines = combined.split(/\r?\n/);
const trailing = lines.pop() ?? "";
const timeline: AgentTimelineItem[] = [];
for (const line of lines) {
this.ingestPersistedHistoryLine(line, {
dispatchLive: options.dispatchLive,
timeline,
});
}
if (trailing.trim().length > 0) {
const handled = this.ingestPersistedHistoryLine(trailing, {
dispatchLive: options.dispatchLive,
timeline,
});
if (!handled) {
this.historyLineFragment = trailing;
}
}
if (!options.dispatchLive && timeline.length > 0) {
this.persistedHistory = [...this.persistedHistory, ...timeline];
this.historyPending = true;
}
}
private ingestPersistedHistoryLine(
line: string,
options: {
dispatchLive: boolean;
timeline: AgentTimelineItem[];
}
): boolean {
const trimmed = line.trim();
if (!trimmed) {
return true;
}
let entry: Record<string, unknown>;
try {
entry = JSON.parse(trimmed) as Record<string, unknown>;
} catch {
return false;
}
if (entry.isSidechain) {
return true;
}
if (entry.type === "user" && typeof entry.uuid === "string") {
this.rememberUserMessageId(entry.uuid);
}
if (options.dispatchLive) {
this.dispatchPersistedHistoryEntry(entry);
return true;
}
const items = this.convertHistoryEntry(entry);
if (items.length > 0) {
options.timeline.push(...items);
}
return true;
}
private dispatchPersistedHistoryEntry(entry: Record<string, unknown>): void {
const liveMessage = this.normalizePersistedHistoryEntryToLiveMessage(entry);
if (liveMessage) {
this.routeSdkMessageFromPump(liveMessage);
return;
}
const items = this.convertHistoryEntry(entry);
for (const item of items) {
this.pushEvent({
type: "timeline",
item,
provider: "claude",
});
}
}
private normalizePersistedHistoryEntryToLiveMessage(
entry: Record<string, unknown>
): SDKMessage | null {
const taskNotificationMessage =
coerceTaskNotificationHistoryRecordToSystemMessage(entry);
if (taskNotificationMessage) {
return taskNotificationMessage as unknown as SDKMessage;
}
const type = readTrimmedString(entry.type);
switch (type) {
case "assistant":
case "result":
case "stream_event":
case "system":
case "tool_progress":
case "user":
return entry as unknown as SDKMessage;
default:
return null;
}
}
private resolveHistoryPath(sessionId: string): string | null {
const cwd = this.config.cwd;
if (!cwd) return null;
@@ -4624,13 +4748,9 @@ export function convertClaudeHistoryEntry(
}];
}
if (entry.type === "system") {
const taskNotificationItem = mapTaskNotificationSystemRecordToToolCall(
entry
);
if (taskNotificationItem) {
return [taskNotificationItem];
}
const taskNotificationItem = mapTaskNotificationSystemRecordToToolCall(entry);
if (taskNotificationItem) {
return [taskNotificationItem];
}
if (entry.isCompactSummary) {
@@ -1,4 +1,4 @@
import { mkdtempSync, mkdirSync, rmSync, writeFileSync } from "node:fs";
import { appendFileSync, mkdtempSync, mkdirSync, rmSync, writeFileSync } from "node:fs";
import os from "node:os";
import path from "node:path";
import { afterEach, beforeEach, describe, expect, test, vi } from "vitest";
@@ -22,6 +22,8 @@ vi.mock("@anthropic-ai/claude-agent-sdk", () => ({
const LIVE_REPLY_MARKER = "LIVE_ONLY_REPLY_MARKER";
const HISTORY_USER_MARKER = "HISTORY_ONLY_USER_MARKER";
const HISTORY_ASSISTANT_MARKER = "HISTORY_ONLY_ASSISTANT_MARKER";
const APPENDED_TASK_NOTIFICATION_MARKER = "Appended background task completed";
const APPENDED_ASSISTANT_MARKER = "APPENDED_BACKGROUND_ASSISTANT_MARKER";
function buildSdkQueryMock() {
const events = [
@@ -70,6 +72,19 @@ function buildSdkQueryMock() {
};
}
function buildIdleSdkQueryMock() {
return {
next: vi.fn(async () => ({ done: true, value: undefined })),
interrupt: vi.fn(async () => undefined),
return: vi.fn(async () => undefined),
setPermissionMode: vi.fn(async () => undefined),
setModel: vi.fn(async () => undefined),
supportedModels: vi.fn(async () => [{ value: "opus", displayName: "Opus" }]),
supportedCommands: vi.fn(async () => []),
rewindFiles: vi.fn(async () => ({ canRewind: true })),
};
}
function collectTimelineText(events: AgentStreamEvent[]): string {
const chunks: string[] = [];
for (const event of events) {
@@ -86,6 +101,28 @@ function collectTimelineText(events: AgentStreamEvent[]): string {
return chunks.join("\n");
}
async function readNextEvent(
iterator: AsyncIterator<AgentStreamEvent>,
timeoutMs: number
): Promise<AgentStreamEvent> {
const outcome = await Promise.race([
iterator.next().then((result) => ({ kind: "result" as const, result })),
new Promise<{ kind: "timeout" }>((resolve) => {
setTimeout(() => resolve({ kind: "timeout" }), timeoutMs);
}),
]);
if (outcome.kind === "timeout") {
throw new Error("Timed out waiting for live event");
}
if (outcome.result.done) {
throw new Error("Live event stream ended before appended transcript arrived");
}
return outcome.result.value;
}
describe("ClaudeAgentSession history replay regression", () => {
let tempRoot: string;
let cwd: string;
@@ -215,6 +252,117 @@ describe("ClaudeAgentSession history replay regression", () => {
expect(timelineText).toContain(HISTORY_ASSISTANT_MARKER);
});
test("emits appended transcript lines through streamLiveEvents after history was primed", async () => {
sdkMocks.query.mockImplementation(() => {
const mock = buildIdleSdkQueryMock();
sdkMocks.lastQuery = mock;
return mock;
});
const logger = createTestLogger();
const client = new ClaudeAgentClient({ logger });
const handle: AgentPersistenceHandle = {
provider: "claude",
sessionId: "history-session",
nativeHandle: "history-session",
metadata: {
provider: "claude",
cwd,
},
};
const sanitized = cwd.replace(/[\\/\.]/g, "-").replace(/_/g, "-");
const historyPath = path.join(
configDir,
"projects",
sanitized,
"history-session.jsonl"
);
const session = await client.resumeSession(handle, { cwd });
try {
for await (const _event of session.streamHistory()) {
// Prime existing persisted history the same way agent-manager does.
}
const liveEvents = session.streamLiveEvents();
const iterator = liveEvents[Symbol.asyncIterator]();
appendFileSync(
historyPath,
`\n${JSON.stringify({
type: "queue-operation",
operation: "enqueue",
uuid: "appended-task-note-1",
content: [
"<task-notification>",
"<task-id>appended-bg-1</task-id>",
"<status>completed</status>",
`<summary>${APPENDED_TASK_NOTIFICATION_MARKER}</summary>`,
"<output-file>/tmp/appended-bg-1.txt</output-file>",
"</task-notification>",
].join("\n"),
})}\n${JSON.stringify({
type: "assistant",
sessionId: "history-session",
cwd,
message: {
role: "assistant",
content: APPENDED_ASSISTANT_MARKER,
},
})}`,
"utf8"
);
const appendedEvents: AgentStreamEvent[] = [];
for (let attempt = 0; attempt < 4; attempt += 1) {
appendedEvents.push(await readNextEvent(iterator, 1_500));
const sawTaskNotification = appendedEvents.some(
(event): event is Extract<AgentStreamEvent, { type: "timeline" }> =>
event.type === "timeline" &&
event.item.type === "tool_call" &&
event.item.name === "task_notification"
);
const sawAssistant = appendedEvents.some(
(event): event is Extract<AgentStreamEvent, { type: "timeline" }> =>
event.type === "timeline" &&
event.item.type === "assistant_message" &&
event.item.text.includes(APPENDED_ASSISTANT_MARKER)
);
if (sawTaskNotification && sawAssistant) {
break;
}
}
const timelineText = collectTimelineText(appendedEvents);
const taskNotificationEvent = appendedEvents.find(
(event): event is Extract<AgentStreamEvent, { type: "timeline" }> =>
event.type === "timeline" &&
event.item.type === "tool_call" &&
event.item.name === "task_notification"
);
const turnStartedEvent = appendedEvents.find(
(event): event is Extract<AgentStreamEvent, { type: "turn_started" }> =>
event.type === "turn_started"
);
expect(taskNotificationEvent).toBeTruthy();
expect(turnStartedEvent).toBeTruthy();
expect(timelineText).toContain(APPENDED_ASSISTANT_MARKER);
expect(taskNotificationEvent?.item.metadata).toMatchObject({
taskId: "appended-bg-1",
status: "completed",
outputFile: "/tmp/appended-bg-1.txt",
});
expect(taskNotificationEvent?.item.detail).toMatchObject({
type: "plain_text",
label: APPENDED_TASK_NOTIFICATION_MARKER,
});
} finally {
await session.close();
}
});
test("listCommands includes rewind command", async () => {
const logger = createTestLogger();
const client = new ClaudeAgentClient({ logger });
@@ -32,9 +32,10 @@ const TaskNotificationUserContentSchema = z.union([
z.array(TaskNotificationUserContentBlockSchema),
]);
const TaskNotificationSystemRecordSchema = z
const TaskNotificationHistoryRecordSchema = z
.object({
subtype: z.literal("task_notification"),
type: z.string().optional(),
subtype: z.string().optional(),
uuid: z.string().optional(),
message_id: z.string().optional(),
task_id: z.string().optional(),
@@ -57,13 +58,24 @@ export type MapTaskNotificationUserContentToToolCallInput = z.infer<
>;
type TaskNotificationUserContent = z.infer<typeof TaskNotificationUserContentSchema>;
type TaskNotificationSystemRecord = z.infer<typeof TaskNotificationSystemRecordSchema>;
type TaskNotificationHistoryRecord = z.infer<typeof TaskNotificationHistoryRecordSchema>;
type TaskNotificationLifecycle =
| { status: "completed"; error: null }
| { status: "failed"; error: unknown }
| { status: "canceled"; error: null };
export type TaskNotificationSystemMessageLike = {
type: "system";
subtype: "task_notification";
uuid?: string;
task_id?: string;
status?: "completed" | "failed" | "stopped";
summary?: string;
output_file?: string;
content?: string;
};
type ReadTaskNotificationTagInput = {
text: string;
tagName: string;
@@ -157,12 +169,18 @@ function parseTaskNotificationFromUserContent(
function parseTaskNotificationFromSystemRecord(
record: unknown
): TaskNotificationEnvelope | null {
const parsedRecord = TaskNotificationSystemRecordSchema.safeParse(record);
const parsedRecord = TaskNotificationHistoryRecordSchema.safeParse(record);
if (!parsedRecord.success) {
return null;
}
const systemRecord: TaskNotificationSystemRecord = parsedRecord.data;
const systemRecord: TaskNotificationHistoryRecord = parsedRecord.data;
const isSystemTaskNotification =
systemRecord.type === "system" && systemRecord.subtype === "task_notification";
const isQueueOperation = systemRecord.type === "queue-operation";
if (!isSystemTaskNotification && !isQueueOperation) {
return null;
}
const rawText = toNonEmptyString(systemRecord.content);
return TaskNotificationEnvelopeSchema.parse({
@@ -330,3 +348,31 @@ export function mapTaskNotificationSystemRecordToToolCall(
}
return toTaskNotificationToolCall(parsed);
}
export function coerceTaskNotificationHistoryRecordToSystemMessage(
record: unknown
): TaskNotificationSystemMessageLike | null {
const parsed = parseTaskNotificationFromSystemRecord(record);
if (!parsed) {
return null;
}
const normalizedStatus = parsed.status?.toLowerCase() ?? null;
const status =
normalizedStatus === "failed" || normalizedStatus === "error"
? "failed"
: normalizedStatus === "canceled" || normalizedStatus === "cancelled"
? "stopped"
: "completed";
return {
type: "system",
subtype: "task_notification",
...(parsed.messageId ? { uuid: parsed.messageId } : {}),
...(parsed.taskId ? { task_id: parsed.taskId } : {}),
status,
...(parsed.summary ? { summary: parsed.summary } : {}),
...(parsed.outputFile ? { output_file: parsed.outputFile } : {}),
...(parsed.rawText ? { content: parsed.rawText } : {}),
};
}
+28 -20
View File
@@ -1109,36 +1109,44 @@ export async function getCheckoutShortstat(
const configured = await getConfiguredBaseRefForCwd(cwd, context);
const localBaseRef = configured.baseRef ?? (await resolveBaseRef(cwd));
if (!localBaseRef) {
return null;
}
const currentBranch = await getCurrentBranch(cwd);
if (currentBranch === localBaseRef) {
return null;
}
const comparisonBaseRef = await resolveBestComparisonBaseRef(
cwd,
normalizeLocalBranchRefName(localBaseRef)
);
let diffTarget: string;
let mergeBase: string;
try {
const { stdout } = await execAsync(`git merge-base HEAD ${comparisonBaseRef}`, {
if (currentBranch && localBaseRef && currentBranch !== localBaseRef) {
// Feature branch: diff against the merge-base with the base branch
const comparisonBaseRef = await resolveBestComparisonBaseRef(
cwd,
env: READ_ONLY_GIT_ENV,
});
mergeBase = stdout.trim();
if (!mergeBase) {
normalizeLocalBranchRefName(localBaseRef)
);
try {
const { stdout } = await execAsync(`git merge-base HEAD ${comparisonBaseRef}`, {
cwd,
env: READ_ONLY_GIT_ENV,
});
const mergeBase = stdout.trim();
if (!mergeBase) {
return null;
}
diffTarget = mergeBase;
} catch {
return null;
}
} catch {
} else if (currentBranch) {
// On the base branch (or no base ref configured): diff against remote tracking branch
const hasOrigin = await doesGitRefExist(cwd, `refs/remotes/origin/${currentBranch}`);
if (!hasOrigin) {
return null;
}
diffTarget = `origin/${currentBranch}`;
} else {
return null;
}
try {
const { stdout } = await execAsync(`git diff --shortstat ${mergeBase} HEAD`, {
// Omit HEAD so the diff includes uncommitted (staged + unstaged) changes
const { stdout } = await execAsync(`git diff --shortstat ${diffTarget}`, {
cwd,
env: READ_ONLY_GIT_ENV,
});