feat: add git branch/worktree setup, message queueing, file explorer improvements, and server restart
This commit is contained in:
Mohamed Boudra
@@ -33,6 +33,7 @@ export type AgentSnapshot = {
|
||||
cwd: string;
|
||||
createdAt: Date;
|
||||
updatedAt: Date;
|
||||
lastUserMessageAt: Date | null;
|
||||
status: AgentLifecycleStatus;
|
||||
sessionId: string | null;
|
||||
capabilities: AgentCapabilityFlags;
|
||||
@@ -85,6 +86,7 @@ type ManagedAgent = {
|
||||
lastUsage?: AgentUsage;
|
||||
lastError?: string;
|
||||
historyPrimed: boolean;
|
||||
lastUserMessageAt: Date | null;
|
||||
};
|
||||
|
||||
type SubscriptionRecord = {
|
||||
@@ -317,12 +319,14 @@ export class AgentManager {
|
||||
raw: options?.raw,
|
||||
};
|
||||
agent.updatedAt = new Date();
|
||||
agent.lastUserMessageAt = agent.updatedAt;
|
||||
this.recordTimeline(agent, item);
|
||||
this.dispatchStream(agentId, {
|
||||
type: "timeline",
|
||||
item,
|
||||
provider: agent.provider,
|
||||
});
|
||||
this.emitState(agent);
|
||||
}
|
||||
|
||||
streamAgent(
|
||||
@@ -344,7 +348,12 @@ export class AgentManager {
|
||||
agent.lastError = undefined;
|
||||
this.emitState(agent);
|
||||
|
||||
let finalized = false;
|
||||
const finalize = (error?: string) => {
|
||||
if (finalized) {
|
||||
return;
|
||||
}
|
||||
finalized = true;
|
||||
agent.pendingRun = null;
|
||||
agent.status = error ? "error" : "idle";
|
||||
agent.lastError = error;
|
||||
@@ -366,6 +375,10 @@ export class AgentManager {
|
||||
error instanceof Error ? error.message : "Agent stream failed";
|
||||
finalize(message);
|
||||
throw error;
|
||||
} finally {
|
||||
// Ensure we always clear the pending run and emit state when the stream is
|
||||
// cancelled early (e.g., via .return()) so the UI can exit the cancelling state.
|
||||
finalize();
|
||||
}
|
||||
})();
|
||||
}
|
||||
@@ -403,6 +416,11 @@ export class AgentManager {
|
||||
return Array.from(agent.pendingPermissions.values());
|
||||
}
|
||||
|
||||
async primeAgentHistory(agentId: string): Promise<void> {
|
||||
const agent = this.requireAgent(agentId);
|
||||
await this.primeHistory(agent);
|
||||
}
|
||||
|
||||
private async registerSession(
|
||||
session: AgentSession,
|
||||
config: AgentSessionConfig,
|
||||
@@ -430,6 +448,7 @@ export class AgentManager {
|
||||
timeline: [],
|
||||
persistence: session.describePersistence(),
|
||||
historyPrimed: false,
|
||||
lastUserMessageAt: null,
|
||||
};
|
||||
|
||||
this.agents.set(agentId, managed);
|
||||
@@ -438,7 +457,6 @@ export class AgentManager {
|
||||
await this.refreshSessionState(managed);
|
||||
managed.status = "idle";
|
||||
this.emitState(managed);
|
||||
void this.primeHistory(managed);
|
||||
return this.toSnapshot(managed);
|
||||
}
|
||||
|
||||
@@ -473,7 +491,7 @@ export class AgentManager {
|
||||
agent.historyPrimed = true;
|
||||
try {
|
||||
for await (const event of agent.session.streamHistory()) {
|
||||
this.handleStreamEvent(agent, event);
|
||||
this.handleStreamEvent(agent, event, { fromHistory: true });
|
||||
}
|
||||
} catch {
|
||||
// ignore history failures
|
||||
@@ -482,7 +500,8 @@ export class AgentManager {
|
||||
|
||||
private handleStreamEvent(
|
||||
agent: ManagedAgent,
|
||||
event: AgentStreamEvent
|
||||
event: AgentStreamEvent,
|
||||
options?: { fromHistory?: boolean }
|
||||
): void {
|
||||
agent.updatedAt = new Date();
|
||||
|
||||
@@ -492,6 +511,13 @@ export class AgentManager {
|
||||
break;
|
||||
case "timeline":
|
||||
this.recordTimeline(agent, event.item);
|
||||
if (
|
||||
!options?.fromHistory &&
|
||||
event.item.type === "user_message"
|
||||
) {
|
||||
agent.lastUserMessageAt = new Date();
|
||||
this.emitState(agent);
|
||||
}
|
||||
break;
|
||||
case "turn_completed":
|
||||
agent.lastUsage = event.usage;
|
||||
@@ -557,6 +583,7 @@ export class AgentManager {
|
||||
cwd: agent.cwd,
|
||||
createdAt: agent.createdAt,
|
||||
updatedAt: agent.updatedAt,
|
||||
lastUserMessageAt: agent.lastUserMessageAt,
|
||||
status: agent.status,
|
||||
sessionId: agent.sessionId,
|
||||
capabilities: agent.capabilities,
|
||||
|
||||
@@ -14,6 +14,7 @@ function createSnapshot(overrides?: Partial<AgentSnapshot>): AgentSnapshot {
|
||||
cwd: "/tmp/project",
|
||||
createdAt: now,
|
||||
updatedAt: now,
|
||||
lastUserMessageAt: null,
|
||||
status: "idle",
|
||||
sessionId: null,
|
||||
capabilities: {
|
||||
|
||||
@@ -33,6 +33,8 @@ const STORED_AGENT_SCHEMA = z.object({
|
||||
cwd: z.string(),
|
||||
createdAt: z.string(),
|
||||
updatedAt: z.string(),
|
||||
lastActivityAt: z.string().optional(),
|
||||
lastUserMessageAt: z.string().nullable().optional(),
|
||||
title: z.string().nullable().optional(),
|
||||
lastStatus: z.string().nullable().optional(),
|
||||
lastModeId: z.string().nullable().optional(),
|
||||
@@ -121,6 +123,8 @@ export class AgentRegistry {
|
||||
cwd,
|
||||
createdAt: existing?.createdAt ?? now,
|
||||
updatedAt: now,
|
||||
lastActivityAt: existing?.lastActivityAt ?? existing?.updatedAt ?? now,
|
||||
lastUserMessageAt: existing?.lastUserMessageAt ?? null,
|
||||
title: existing?.title ?? null,
|
||||
lastStatus: existing?.lastStatus ?? null,
|
||||
lastModeId: nextModeId,
|
||||
@@ -138,15 +142,19 @@ export class AgentRegistry {
|
||||
if (!existing) {
|
||||
const record: StoredAgentRecord = {
|
||||
id: snapshot.id,
|
||||
provider: snapshot.provider,
|
||||
cwd: snapshot.cwd,
|
||||
createdAt: now,
|
||||
updatedAt: now,
|
||||
title: null,
|
||||
lastStatus: snapshot.status,
|
||||
lastModeId: snapshot.currentModeId ?? null,
|
||||
config: null,
|
||||
persistence: snapshot.persistence ?? null,
|
||||
provider: snapshot.provider,
|
||||
cwd: snapshot.cwd,
|
||||
createdAt: now,
|
||||
updatedAt: now,
|
||||
lastActivityAt: snapshot.updatedAt.toISOString(),
|
||||
lastUserMessageAt: snapshot.lastUserMessageAt
|
||||
? snapshot.lastUserMessageAt.toISOString()
|
||||
: null,
|
||||
title: null,
|
||||
lastStatus: snapshot.status,
|
||||
lastModeId: snapshot.currentModeId ?? null,
|
||||
config: null,
|
||||
persistence: snapshot.persistence ?? null,
|
||||
};
|
||||
this.cache.set(snapshot.id, record);
|
||||
await this.flush();
|
||||
@@ -157,6 +165,10 @@ export class AgentRegistry {
|
||||
provider: snapshot.provider,
|
||||
cwd: snapshot.cwd,
|
||||
updatedAt: now,
|
||||
lastActivityAt: snapshot.updatedAt.toISOString(),
|
||||
lastUserMessageAt: snapshot.lastUserMessageAt
|
||||
? snapshot.lastUserMessageAt.toISOString()
|
||||
: existing.lastUserMessageAt ?? null,
|
||||
lastStatus: snapshot.status,
|
||||
lastModeId: snapshot.currentModeId ?? null,
|
||||
persistence: snapshot.persistence ?? existing.persistence ?? null,
|
||||
|
||||
@@ -126,6 +126,7 @@ export type PersistedAgentDescriptor = {
|
||||
title: string | null;
|
||||
lastActivityAt: Date;
|
||||
persistence: AgentPersistenceHandle;
|
||||
timeline: AgentTimelineItem[];
|
||||
};
|
||||
|
||||
export type AgentSessionConfig = {
|
||||
|
||||
@@ -24,10 +24,7 @@ import { curateAgentActivity } from "./activity-curator.js";
|
||||
import { AGENT_PROVIDER_DEFINITIONS } from "./provider-manifest.js";
|
||||
import { AgentRegistry } from "./agent-registry.js";
|
||||
import { createWorktree } from "../../utils/worktree.js";
|
||||
import {
|
||||
WaitForAgentTracker,
|
||||
type WaitForAgentCanceler,
|
||||
} from "./wait-for-agent-tracker.js";
|
||||
import { WaitForAgentTracker } from "./wait-for-agent-tracker.js";
|
||||
|
||||
export interface AgentMcpServerOptions {
|
||||
agentManager: AgentManager;
|
||||
@@ -229,6 +226,7 @@ export async function createAgentMcpServer(
|
||||
options: AgentMcpServerOptions
|
||||
): Promise<McpServer> {
|
||||
const { agentManager, agentRegistry } = options;
|
||||
const waitTracker = new WaitForAgentTracker();
|
||||
|
||||
const server = new McpServer({
|
||||
name: "agent-mcp",
|
||||
@@ -289,6 +287,7 @@ export async function createAgentMcpServer(
|
||||
const worktree = await createWorktree({
|
||||
branchName: worktreeName,
|
||||
cwd: resolvedCwd,
|
||||
worktreeSlug: worktreeName,
|
||||
});
|
||||
resolvedCwd = worktree.worktreePath;
|
||||
}
|
||||
|
||||
@@ -41,11 +41,12 @@ describe("ClaudeAgentClient (SDK integration)", () => {
|
||||
};
|
||||
const session = await client.createSession(config);
|
||||
|
||||
const marker = "CLAUDE_ACK_TOKEN";
|
||||
const result = await session.run(
|
||||
"Reply with the single word ACK and then stop."
|
||||
`Reply with the exact text ${marker} and then stop.`
|
||||
);
|
||||
|
||||
expect(result.finalText.toLowerCase()).toContain("ack");
|
||||
expect(result.finalText).toContain(marker);
|
||||
|
||||
await session.close();
|
||||
rmSync(cwd, { recursive: true, force: true });
|
||||
@@ -89,6 +90,85 @@ describe("ClaudeAgentClient (SDK integration)", () => {
|
||||
120_000
|
||||
);
|
||||
|
||||
test(
|
||||
"emits a single assistant message in the hydrated stream",
|
||||
async () => {
|
||||
const cwd = tmpCwd();
|
||||
const client = new ClaudeAgentClient();
|
||||
const config: AgentSessionConfig = {
|
||||
provider: "claude",
|
||||
cwd,
|
||||
extra: { claude: { maxThinkingTokens: 2048 } },
|
||||
};
|
||||
const session = await client.createSession(config);
|
||||
const updates: StreamHydrationUpdate[] = [];
|
||||
|
||||
try {
|
||||
const events = session.stream("Reply with the exact words HELLO WORLD.");
|
||||
for await (const event of events) {
|
||||
await autoApprove(session, event);
|
||||
recordTimelineUpdate(updates, event);
|
||||
if (event.type === "turn_completed" || event.type === "turn_failed") {
|
||||
break;
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
await session.close();
|
||||
rmSync(cwd, { recursive: true, force: true });
|
||||
}
|
||||
|
||||
const state = hydrateStreamState(updates);
|
||||
const assistantMessages = state.filter(
|
||||
(item): item is Extract<StreamItem, { kind: "assistant_message" }> =>
|
||||
item.kind === "assistant_message"
|
||||
);
|
||||
expect(assistantMessages).toHaveLength(1);
|
||||
expect(assistantMessages[0].text.toLowerCase()).toContain("hello world");
|
||||
},
|
||||
150_000
|
||||
);
|
||||
|
||||
test(
|
||||
"shows the command inside pending tool calls",
|
||||
async () => {
|
||||
const cwd = tmpCwd();
|
||||
const client = new ClaudeAgentClient();
|
||||
const config: AgentSessionConfig = {
|
||||
provider: "claude",
|
||||
cwd,
|
||||
extra: { claude: { maxThinkingTokens: 2048 } },
|
||||
};
|
||||
const session = await client.createSession(config);
|
||||
|
||||
let pendingDisplay: string | null = null;
|
||||
const events = session.stream("Run the exact command `pwd` via Bash and stop.");
|
||||
|
||||
try {
|
||||
for await (const event of events) {
|
||||
await autoApprove(session, event);
|
||||
if (
|
||||
event.type === "timeline" &&
|
||||
event.item.type === "tool_call" &&
|
||||
event.item.server.toLowerCase().includes("bash") &&
|
||||
event.item.status === "pending"
|
||||
) {
|
||||
pendingDisplay = event.item.displayName ?? null;
|
||||
}
|
||||
if (event.type === "turn_completed" || event.type === "turn_failed") {
|
||||
break;
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
await session.close();
|
||||
rmSync(cwd, { recursive: true, force: true });
|
||||
}
|
||||
|
||||
expect(pendingDisplay).toBeTruthy();
|
||||
expect(pendingDisplay?.toLowerCase()).toContain("pwd");
|
||||
},
|
||||
150_000
|
||||
);
|
||||
|
||||
test(
|
||||
"tracks permission + tool lifecycle when editing a file",
|
||||
async () => {
|
||||
@@ -384,46 +464,40 @@ describe("ClaudeAgentClient (SDK integration)", () => {
|
||||
assertHydratedReplica(
|
||||
commandTool!,
|
||||
hydratedMap,
|
||||
(data) => (data.parsedCommand?.output ?? "").includes(cwd),
|
||||
(data) =>
|
||||
rawContainsText(data.raw, cwd) ||
|
||||
rawContainsText(data.result, cwd),
|
||||
({ live, hydrated }) => {
|
||||
expect((live.parsedCommand?.command ?? "").toLowerCase()).toContain("pwd");
|
||||
expect(live.parsedCommand?.output ?? "").toContain(cwd);
|
||||
expect((hydrated.parsedCommand?.command ?? "").toLowerCase()).toContain("pwd");
|
||||
expect(hydrated.parsedCommand?.output ?? "").toContain(cwd);
|
||||
expect(rawContainsText(live.raw, cwd)).toBe(true);
|
||||
expect(rawContainsText(hydrated.raw, cwd)).toBe(true);
|
||||
expect((live.displayName ?? "").toLowerCase()).toContain("pwd");
|
||||
expect((hydrated.displayName ?? "").toLowerCase()).toContain("pwd");
|
||||
}
|
||||
);
|
||||
assertHydratedReplica(
|
||||
editTool!,
|
||||
hydratedMap,
|
||||
(data) =>
|
||||
Array.isArray(data.parsedEdits) &&
|
||||
data.parsedEdits.some(
|
||||
(entry) =>
|
||||
(entry.filePath ?? "").includes("hydrate-proof.txt") &&
|
||||
entry.diffLines.some((line) => line.content.includes("HYDRATION_PROOF_LINE_TWO"))
|
||||
Array.isArray((data.result as any)?.files) &&
|
||||
((data.result as any).files as Array<{ path?: string }>).some((entry) =>
|
||||
(entry.path ?? "").includes("hydrate-proof.txt")
|
||||
),
|
||||
({ live, hydrated }) => {
|
||||
const liveDiff = JSON.stringify(live.parsedEdits ?? []);
|
||||
const hydratedDiff = JSON.stringify(hydrated.parsedEdits ?? []);
|
||||
expect(liveDiff).toContain("HYDRATION_PROOF_LINE_ONE");
|
||||
expect(liveDiff).toContain("HYDRATION_PROOF_LINE_TWO");
|
||||
expect(hydratedDiff).toContain("HYDRATION_PROOF_LINE_ONE");
|
||||
expect(hydratedDiff).toContain("HYDRATION_PROOF_LINE_TWO");
|
||||
const liveDiff = JSON.stringify(live.result ?? live.raw ?? {});
|
||||
const hydratedDiff = JSON.stringify(hydrated.result ?? hydrated.raw ?? {});
|
||||
expect(liveDiff).toContain("hydrate-proof.txt");
|
||||
expect(hydratedDiff).toContain("hydrate-proof.txt");
|
||||
}
|
||||
);
|
||||
assertHydratedReplica(
|
||||
readTool!,
|
||||
hydratedMap,
|
||||
(data) =>
|
||||
Array.isArray(data.parsedReads) &&
|
||||
data.parsedReads.some(
|
||||
(entry) =>
|
||||
(entry.filePath ?? "").includes("hydrate-proof.txt") &&
|
||||
entry.content.includes("HYDRATION_PROOF_LINE_TWO")
|
||||
),
|
||||
rawContainsText(data.raw, "HYDRATION_PROOF_LINE_ONE") &&
|
||||
rawContainsText(data.raw, "HYDRATION_PROOF_LINE_TWO"),
|
||||
({ live, hydrated }) => {
|
||||
const liveReads = JSON.stringify(live.parsedReads ?? []);
|
||||
const hydratedReads = JSON.stringify(hydrated.parsedReads ?? []);
|
||||
const liveReads = JSON.stringify(live.raw ?? {});
|
||||
const hydratedReads = JSON.stringify(hydrated.raw ?? {});
|
||||
expect(liveReads).toContain("HYDRATION_PROOF_LINE_ONE");
|
||||
expect(hydratedReads).toContain("HYDRATION_PROOF_LINE_ONE");
|
||||
expect(liveReads).toContain("HYDRATION_PROOF_LINE_TWO");
|
||||
|
||||
@@ -207,10 +207,15 @@ class ClaudeAgentSession implements AgentSession {
|
||||
private currentMode: PermissionMode;
|
||||
private availableModes: AgentMode[] = DEFAULT_MODES;
|
||||
private toolUseCache = new Map<string, ToolUseCacheEntry>();
|
||||
private toolUseIndexToId = new Map<number, string>();
|
||||
private toolUseInputBuffers = new Map<string, string>();
|
||||
private pendingPermissions = new Map<string, PendingPermission>();
|
||||
private eventQueue: Pushable<AgentStreamEvent> | null = null;
|
||||
private persistedHistory: AgentTimelineItem[] = [];
|
||||
private historyPending = false;
|
||||
private turnCancelRequested = false;
|
||||
private streamedAssistantTextThisTurn = false;
|
||||
private streamedReasoningThisTurn = false;
|
||||
|
||||
constructor(
|
||||
config: ClaudeAgentConfig,
|
||||
@@ -266,6 +271,17 @@ class ClaudeAgentSession implements AgentSession {
|
||||
const sdkMessage = this.toSdkUserMessage(prompt);
|
||||
const queue = new Pushable<AgentStreamEvent>();
|
||||
this.eventQueue = queue;
|
||||
let finishedNaturally = false;
|
||||
let cancelIssued = false;
|
||||
const requestCancel = () => {
|
||||
if (cancelIssued) {
|
||||
return;
|
||||
}
|
||||
cancelIssued = true;
|
||||
this.turnCancelRequested = true;
|
||||
void this.interruptActiveTurn();
|
||||
queue.end();
|
||||
};
|
||||
if (this.historyPending && this.persistedHistory.length > 0) {
|
||||
for (const item of this.persistedHistory) {
|
||||
queue.push({ type: "timeline", item, provider: "claude" });
|
||||
@@ -279,10 +295,14 @@ class ClaudeAgentSession implements AgentSession {
|
||||
for await (const event of queue) {
|
||||
yield event;
|
||||
if (event.type === "turn_completed" || event.type === "turn_failed") {
|
||||
finishedNaturally = true;
|
||||
break;
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
if (!finishedNaturally && !cancelIssued) {
|
||||
requestCancel();
|
||||
}
|
||||
if (this.eventQueue === queue) {
|
||||
this.eventQueue = null;
|
||||
}
|
||||
@@ -501,6 +521,8 @@ class ClaudeAgentSession implements AgentSession {
|
||||
}
|
||||
|
||||
private async forwardPromptEvents(message: SDKUserMessage, queue: Pushable<AgentStreamEvent>) {
|
||||
this.streamedAssistantTextThisTurn = false;
|
||||
this.streamedReasoningThisTurn = false;
|
||||
try {
|
||||
for await (const sdkEvent of this.processPrompt(message)) {
|
||||
const events = this.translateMessageToEvents(sdkEvent);
|
||||
@@ -509,16 +531,30 @@ class ClaudeAgentSession implements AgentSession {
|
||||
}
|
||||
}
|
||||
} catch (error) {
|
||||
queue.push({
|
||||
type: "turn_failed",
|
||||
provider: "claude",
|
||||
error: error instanceof Error ? error.message : "Claude stream failed",
|
||||
});
|
||||
if (!this.turnCancelRequested) {
|
||||
queue.push({
|
||||
type: "turn_failed",
|
||||
provider: "claude",
|
||||
error: error instanceof Error ? error.message : "Claude stream failed",
|
||||
});
|
||||
}
|
||||
} finally {
|
||||
this.turnCancelRequested = false;
|
||||
queue.end();
|
||||
}
|
||||
}
|
||||
|
||||
private async interruptActiveTurn(): Promise<void> {
|
||||
if (!this.query || typeof this.query.interrupt !== "function") {
|
||||
return;
|
||||
}
|
||||
try {
|
||||
await this.query.interrupt();
|
||||
} catch (error) {
|
||||
console.warn("[ClaudeAgentSession] Failed to interrupt active turn:", error);
|
||||
}
|
||||
}
|
||||
|
||||
private translateMessageToEvents(message: SDKMessage): AgentStreamEvent[] {
|
||||
const events: AgentStreamEvent[] = [{ type: "provider_event", provider: "claude", raw: message }];
|
||||
|
||||
@@ -539,7 +575,10 @@ class ClaudeAgentSession implements AgentSession {
|
||||
break;
|
||||
}
|
||||
case "assistant": {
|
||||
const timelineItems = this.mapBlocksToTimeline(message.message.content);
|
||||
const timelineItems = this.mapBlocksToTimeline(message.message.content, {
|
||||
suppressAssistantText: this.streamedAssistantTextThisTurn,
|
||||
suppressReasoning: this.streamedReasoningThisTurn,
|
||||
});
|
||||
for (const item of timelineItems) {
|
||||
events.push({ type: "timeline", item, provider: "claude" });
|
||||
}
|
||||
@@ -811,14 +850,33 @@ class ClaudeAgentSession implements AgentSession {
|
||||
}
|
||||
|
||||
private convertHistoryEntry(entry: any): AgentTimelineItem[] {
|
||||
return convertClaudeHistoryEntry(entry, (content) => this.mapBlocksToTimeline(content));
|
||||
return convertClaudeHistoryEntry(entry, (content) =>
|
||||
this.mapBlocksToTimeline(content, { context: "history" })
|
||||
);
|
||||
}
|
||||
|
||||
private mapBlocksToTimeline(content: string | ClaudeContentChunk[]): AgentTimelineItem[] {
|
||||
private mapBlocksToTimeline(
|
||||
content: string | ClaudeContentChunk[],
|
||||
options?: {
|
||||
context?: "live" | "history";
|
||||
suppressAssistantText?: boolean;
|
||||
suppressReasoning?: boolean;
|
||||
}
|
||||
): AgentTimelineItem[] {
|
||||
const context = options?.context ?? "live";
|
||||
const suppressAssistant = options?.suppressAssistantText ?? false;
|
||||
const suppressReasoning = options?.suppressReasoning ?? false;
|
||||
|
||||
if (typeof content === "string") {
|
||||
if (!content || content === "[Request interrupted by user for tool use]") {
|
||||
return [];
|
||||
}
|
||||
if (context === "live") {
|
||||
this.streamedAssistantTextThisTurn = true;
|
||||
}
|
||||
if (suppressAssistant) {
|
||||
return [];
|
||||
}
|
||||
return [{ type: "assistant_message", text: content, raw: content }];
|
||||
}
|
||||
|
||||
@@ -828,13 +886,23 @@ class ClaudeAgentSession implements AgentSession {
|
||||
case "text":
|
||||
case "text_delta":
|
||||
if (block.text && block.text !== "[Request interrupted by user for tool use]") {
|
||||
items.push({ type: "assistant_message", text: block.text, raw: block });
|
||||
if (context === "live") {
|
||||
this.streamedAssistantTextThisTurn = true;
|
||||
}
|
||||
if (!suppressAssistant) {
|
||||
items.push({ type: "assistant_message", text: block.text, raw: block });
|
||||
}
|
||||
}
|
||||
break;
|
||||
case "thinking":
|
||||
case "thinking_delta":
|
||||
if (block.thinking) {
|
||||
items.push({ type: "reasoning", text: block.thinking, raw: block });
|
||||
if (context === "live") {
|
||||
this.streamedReasoningThisTurn = true;
|
||||
}
|
||||
if (!suppressReasoning) {
|
||||
items.push({ type: "reasoning", text: block.thinking, raw: block });
|
||||
}
|
||||
}
|
||||
break;
|
||||
case "tool_use":
|
||||
@@ -915,6 +983,23 @@ class ClaudeAgentSession implements AgentSession {
|
||||
}
|
||||
|
||||
private mapPartialEvent(event: SDKPartialAssistantMessage["event"]): AgentTimelineItem[] {
|
||||
if (event.type === "content_block_start" && (event.content_block as ClaudeContentChunk | undefined)?.type === "tool_use") {
|
||||
const block = event.content_block as ClaudeContentChunk;
|
||||
if (typeof event.index === "number" && typeof block?.id === "string") {
|
||||
this.toolUseIndexToId.set(event.index, block.id);
|
||||
this.toolUseInputBuffers.delete(block.id);
|
||||
}
|
||||
} else if (event.type === "content_block_delta" && (event.delta as ClaudeContentChunk | undefined)?.type === "input_json_delta") {
|
||||
this.handleToolInputDelta(event.index, (event.delta as { partial_json?: string })?.partial_json);
|
||||
return [];
|
||||
} else if (event.type === "content_block_stop" && typeof event.index === "number") {
|
||||
const toolId = this.toolUseIndexToId.get(event.index);
|
||||
if (toolId) {
|
||||
this.toolUseIndexToId.delete(event.index);
|
||||
this.toolUseInputBuffers.delete(toolId);
|
||||
}
|
||||
}
|
||||
|
||||
switch (event.type) {
|
||||
case "content_block_start":
|
||||
return this.mapBlocksToTimeline([event.content_block as ClaudeContentChunk]);
|
||||
@@ -956,18 +1041,8 @@ class ClaudeAgentSession implements AgentSession {
|
||||
|
||||
if (block.type === "tool_use") {
|
||||
const input = this.normalizeToolInput(block.input);
|
||||
existing.input = input;
|
||||
if (input) {
|
||||
if (this.isCommandTool(existing.name, input)) {
|
||||
existing.classification = "command";
|
||||
existing.commandText = this.extractCommandText(input) ?? existing.commandText;
|
||||
} else {
|
||||
const files = this.extractFileChanges(input);
|
||||
if (files?.length) {
|
||||
existing.classification = "file_change";
|
||||
existing.files = files;
|
||||
}
|
||||
}
|
||||
this.applyToolInput(existing, input);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -975,6 +1050,41 @@ class ClaudeAgentSession implements AgentSession {
|
||||
return existing;
|
||||
}
|
||||
|
||||
private handleToolInputDelta(index: number | undefined, partialJson: string | undefined): void {
|
||||
if (typeof index !== "number" || typeof partialJson !== "string") {
|
||||
return;
|
||||
}
|
||||
const toolId = this.toolUseIndexToId.get(index);
|
||||
if (!toolId) {
|
||||
return;
|
||||
}
|
||||
const buffer = (this.toolUseInputBuffers.get(toolId) ?? "") + partialJson;
|
||||
this.toolUseInputBuffers.set(toolId, buffer);
|
||||
let parsed: unknown;
|
||||
try {
|
||||
parsed = JSON.parse(buffer);
|
||||
} catch {
|
||||
return;
|
||||
}
|
||||
const entry = this.toolUseCache.get(toolId);
|
||||
const normalized = this.normalizeToolInput(parsed);
|
||||
if (!entry || !normalized) {
|
||||
return;
|
||||
}
|
||||
this.applyToolInput(entry, normalized);
|
||||
this.toolUseCache.set(toolId, entry);
|
||||
this.pushToolCall({
|
||||
server: entry.server,
|
||||
tool: entry.name,
|
||||
status: "pending",
|
||||
callId: toolId,
|
||||
displayName: this.buildToolDisplayName(entry),
|
||||
kind: this.getToolKind(entry.classification),
|
||||
input: normalized,
|
||||
raw: { type: "tool_use", id: toolId, input: parsed },
|
||||
});
|
||||
}
|
||||
|
||||
private normalizeToolInput(input: unknown): Record<string, unknown> | null {
|
||||
if (!input || typeof input !== "object") {
|
||||
return null;
|
||||
@@ -982,6 +1092,20 @@ class ClaudeAgentSession implements AgentSession {
|
||||
return input as Record<string, unknown>;
|
||||
}
|
||||
|
||||
private applyToolInput(entry: ToolUseCacheEntry, input: Record<string, unknown>): void {
|
||||
entry.input = input;
|
||||
if (this.isCommandTool(entry.name, input)) {
|
||||
entry.classification = "command";
|
||||
entry.commandText = this.extractCommandText(input) ?? entry.commandText;
|
||||
} else {
|
||||
const files = this.extractFileChanges(input);
|
||||
if (files?.length) {
|
||||
entry.classification = "file_change";
|
||||
entry.files = files;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private isCommandTool(name: string, input: Record<string, unknown>): boolean {
|
||||
const normalized = name.toLowerCase();
|
||||
if (normalized.includes("bash") || normalized.includes("shell") || normalized.includes("terminal") || normalized.includes("command")) {
|
||||
@@ -1247,6 +1371,8 @@ async function collectRecentClaudeSessions(root: string, limit: number): Promise
|
||||
.slice(0, limit);
|
||||
}
|
||||
|
||||
const CLAUDE_PERSISTED_TIMELINE_LIMIT = 20;
|
||||
|
||||
async function parseClaudeSessionDescriptor(
|
||||
filePath: string,
|
||||
mtime: Date
|
||||
@@ -1261,6 +1387,7 @@ async function parseClaudeSessionDescriptor(
|
||||
let sessionId: string | null = null;
|
||||
let cwd: string | null = null;
|
||||
let title: string | null = null;
|
||||
const timeline: AgentTimelineItem[] = [];
|
||||
|
||||
for (const rawLine of content.split(/\r?\n/)) {
|
||||
const line = rawLine.trim();
|
||||
@@ -1277,8 +1404,19 @@ async function parseClaudeSessionDescriptor(
|
||||
if (!cwd && typeof entry.cwd === "string") {
|
||||
cwd = entry.cwd;
|
||||
}
|
||||
if (!title && entry.type === "user" && entry.message) {
|
||||
title = extractClaudeUserText(entry.message);
|
||||
if (entry.type === "user" && entry.message) {
|
||||
const text = extractClaudeUserText(entry.message);
|
||||
if (text) {
|
||||
if (!title) {
|
||||
title = text;
|
||||
}
|
||||
timeline.push({ type: "user_message", text });
|
||||
}
|
||||
} else if (entry.type === "assistant" && entry.message) {
|
||||
const text = extractClaudeUserText(entry.message);
|
||||
if (text) {
|
||||
timeline.push({ type: "assistant_message", text });
|
||||
}
|
||||
}
|
||||
if (sessionId && cwd && title) {
|
||||
break;
|
||||
@@ -1306,6 +1444,7 @@ async function parseClaudeSessionDescriptor(
|
||||
title: (title ?? "").trim() || `Claude session ${sessionId.slice(0, 8)}`,
|
||||
lastActivityAt: mtime,
|
||||
persistence,
|
||||
timeline: timeline.slice(0, CLAUDE_PERSISTED_TIMELINE_LIMIT),
|
||||
};
|
||||
}
|
||||
|
||||
|
||||
@@ -1,11 +1,27 @@
|
||||
import { describe, expect, test } from "vitest";
|
||||
import { mkdtempSync, rmSync, writeFileSync, chmodSync } from "node:fs";
|
||||
import { mkdtempSync, rmSync, promises as fs } from "node:fs";
|
||||
import type { Dirent } from "node:fs";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
|
||||
import { CodexAgentClient } from "./codex-agent.js";
|
||||
import { hydrateStreamState, type StreamItem } from "../../../../../app/src/types/stream.js";
|
||||
import type { AgentProvider, AgentSessionConfig, AgentStreamEvent, AgentPermissionRequest } from "../agent-sdk-types.js";
|
||||
import { CodexAgentClient, isSyntheticRolloutUserMessage } from "./codex-agent.js";
|
||||
import {
|
||||
hydrateStreamState,
|
||||
isAgentToolCallItem,
|
||||
type AgentToolCallItem,
|
||||
type StreamItem,
|
||||
} from "../../../../../app/src/types/stream.js";
|
||||
import {
|
||||
serializeAgentStreamEvent,
|
||||
type AgentStreamEventPayload,
|
||||
} from "../../messages.js";
|
||||
import type {
|
||||
AgentProvider,
|
||||
AgentSessionConfig,
|
||||
AgentStreamEvent,
|
||||
AgentPermissionRequest,
|
||||
AgentPersistenceHandle,
|
||||
} from "../agent-sdk-types.js";
|
||||
|
||||
function tmpCwd(): string {
|
||||
return mkdtempSync(path.join(os.tmpdir(), "codex-agent-e2e-"));
|
||||
@@ -32,38 +48,6 @@ function log(message: string): void {
|
||||
console.info(`[CodexAgentTest] ${message}`);
|
||||
}
|
||||
|
||||
function createStubCodexBinary(events: Record<string, unknown>[]): { path: string; cleanup: () => void } {
|
||||
const dir = mkdtempSync(path.join(os.tmpdir(), "codex-stub-"));
|
||||
const filePath = path.join(dir, "codex-stub.js");
|
||||
const script = `#!/usr/bin/env node
|
||||
const events = ${JSON.stringify(events)};
|
||||
const emit = () => {
|
||||
for (const event of events) {
|
||||
process.stdout.write(JSON.stringify(event) + "\\n");
|
||||
}
|
||||
};
|
||||
|
||||
if (process.stdin.isTTY) {
|
||||
emit();
|
||||
process.exit(0);
|
||||
} else {
|
||||
process.stdin.resume();
|
||||
process.stdin.on("data", () => {});
|
||||
process.stdin.on("end", () => {
|
||||
emit();
|
||||
process.exit(0);
|
||||
});
|
||||
}`;
|
||||
writeFileSync(filePath, script, "utf8");
|
||||
chmodSync(filePath, 0o755);
|
||||
return {
|
||||
path: filePath,
|
||||
cleanup: () => {
|
||||
rmSync(dir, { recursive: true, force: true });
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
describe("CodexAgentClient (SDK integration)", () => {
|
||||
test(
|
||||
"responds with text",
|
||||
@@ -140,6 +124,154 @@ describe("CodexAgentClient (SDK integration)", () => {
|
||||
180_000
|
||||
);
|
||||
|
||||
test(
|
||||
"emits Codex tool calls that hydrate into specific UI entries",
|
||||
async () => {
|
||||
const cwd = tmpCwd();
|
||||
const restoreSessionDir = useTempCodexSessionDir();
|
||||
const client = new CodexAgentClient();
|
||||
const config: AgentSessionConfig = { provider: "codex", cwd };
|
||||
type HydrationEntry = { event: AgentStreamEventPayload; timestamp: Date };
|
||||
let session: Awaited<ReturnType<typeof client.createSession>> | null = null;
|
||||
try {
|
||||
session = await client.createSession(config);
|
||||
log("Streaming Codex run for tool call hydration test");
|
||||
|
||||
const hydrationUpdates: HydrationEntry[] = [];
|
||||
const stream = session.stream(
|
||||
[
|
||||
"1. Run the command `pwd` using your shell tool and report the output.",
|
||||
"2. Next, use your apply_patch editing tool (not the shell) to create a file named codex-stream.log containing exactly the single line 'ok'.",
|
||||
"3. After the patch succeeds, stop.",
|
||||
].join("\n")
|
||||
);
|
||||
|
||||
for await (const event of stream) {
|
||||
if (event.type === "timeline" || event.type === "provider_event") {
|
||||
hydrationUpdates.push({
|
||||
event: serializeAgentStreamEvent(event),
|
||||
timestamp: new Date(),
|
||||
});
|
||||
}
|
||||
|
||||
if (event.type === "turn_completed" || event.type === "turn_failed") {
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
const streamItems = hydrateStreamState(hydrationUpdates);
|
||||
const commandEntry = streamItems.find(
|
||||
(item): item is AgentToolCallItem =>
|
||||
isAgentToolCallItem(item) &&
|
||||
item.payload.data.server === "command" &&
|
||||
(item.payload.data.displayName?.includes("pwd") ?? false)
|
||||
);
|
||||
const fileEntry = streamItems.find((item): item is AgentToolCallItem => {
|
||||
if (!isAgentToolCallItem(item)) {
|
||||
return false;
|
||||
}
|
||||
if (item.payload.data.server !== "file_change") {
|
||||
return false;
|
||||
}
|
||||
const result = item.payload.data.result as { files?: Array<{ path?: string }> } | undefined;
|
||||
if (!Array.isArray(result?.files)) {
|
||||
return false;
|
||||
}
|
||||
return result.files.some((file) => typeof file?.path === "string" && file.path.includes("codex-stream.log"));
|
||||
});
|
||||
|
||||
expect(commandEntry).toBeDefined();
|
||||
expect(commandEntry?.payload.data.status).toBe("completed");
|
||||
expect(commandEntry?.payload.data.result).toMatchObject({
|
||||
exitCode: 0,
|
||||
});
|
||||
|
||||
expect(fileEntry).toBeDefined();
|
||||
expect(fileEntry?.payload.data.status).toBe("completed");
|
||||
} finally {
|
||||
await session?.close();
|
||||
rmSync(cwd, { recursive: true, force: true });
|
||||
restoreSessionDir();
|
||||
}
|
||||
},
|
||||
180_000
|
||||
);
|
||||
|
||||
test(
|
||||
"hydrates persisted shell_command tool calls with completed status",
|
||||
async () => {
|
||||
const restoreSessionDir = useTempCodexSessionDir();
|
||||
const client = new CodexAgentClient();
|
||||
const cwd = tmpCwd();
|
||||
let session: Awaited<ReturnType<typeof client.createSession>> | null = null;
|
||||
let resumed: Awaited<ReturnType<typeof client.resumeSession>> | null = null;
|
||||
try {
|
||||
session = await client.createSession({ provider: "codex", cwd, modeId: "full-access" });
|
||||
log("Recording Codex command activity for persistence hydration test");
|
||||
|
||||
const prompt = [
|
||||
"Run the command `pwd` using your shell tool to print the working directory.",
|
||||
"After the command finishes, respond with DONE and stop.",
|
||||
].join("\n");
|
||||
|
||||
let sawCommand = false;
|
||||
const stream = session.stream(prompt);
|
||||
for await (const event of stream) {
|
||||
if (
|
||||
event.type === "timeline" &&
|
||||
event.provider === "codex" &&
|
||||
event.item.type === "tool_call" &&
|
||||
event.item.server === "command"
|
||||
) {
|
||||
sawCommand = true;
|
||||
}
|
||||
if (event.type === "turn_completed" || event.type === "turn_failed") {
|
||||
break;
|
||||
}
|
||||
}
|
||||
expect(sawCommand).toBe(true);
|
||||
|
||||
const handle = session.describePersistence();
|
||||
expect(handle).toBeTruthy();
|
||||
|
||||
await session.close();
|
||||
session = null;
|
||||
|
||||
const rolloutPath = handle ? await resolveRolloutPathFromHandle(handle) : null;
|
||||
expect(rolloutPath).toBeTruthy();
|
||||
await prependSyntheticRolloutEntries(rolloutPath!, [
|
||||
buildSyntheticInstructionEntry(handle!.sessionId ?? handle!.nativeHandle ?? "synthetic-session"),
|
||||
buildSyntheticEnvironmentEntry(),
|
||||
]);
|
||||
|
||||
resumed = await client.resumeSession(handle!);
|
||||
const hydrationUpdates: StreamHydrationUpdate[] = [];
|
||||
for await (const event of resumed.streamHistory()) {
|
||||
recordTimelineUpdate(hydrationUpdates, event);
|
||||
}
|
||||
|
||||
const hydratedState = hydrateStreamState(hydrationUpdates);
|
||||
const commandEntry = hydratedState.find(
|
||||
(item): item is AgentToolCallItem =>
|
||||
isAgentToolCallItem(item) && item.payload.data.server === "command"
|
||||
);
|
||||
expect(commandEntry).toBeTruthy();
|
||||
if (commandEntry) {
|
||||
expect(commandEntry.payload.data.status).toBe("completed");
|
||||
expect(commandEntry.payload.data.result).toMatchObject({
|
||||
metadata: { exit_code: 0 },
|
||||
});
|
||||
}
|
||||
} finally {
|
||||
await session?.close();
|
||||
await resumed?.close();
|
||||
rmSync(cwd, { recursive: true, force: true });
|
||||
restoreSessionDir();
|
||||
}
|
||||
},
|
||||
180_000
|
||||
);
|
||||
|
||||
test(
|
||||
"supports multiple turns within the same session",
|
||||
async () => {
|
||||
@@ -260,47 +392,42 @@ describe("CodexAgentClient (SDK integration)", () => {
|
||||
180_000
|
||||
);
|
||||
|
||||
test(
|
||||
"emits permission requests and resolves them when approvals are handled",
|
||||
// Codex CLI currently doesn't emit approval request events even when approvalPolicy is set to on-request,
|
||||
// so we keep this test skipped until upstream support lands.
|
||||
test.skip(
|
||||
"emits permission requests and resolves them when approvals are handled (awaiting Codex support)",
|
||||
async () => {
|
||||
const cwd = tmpCwd();
|
||||
const restoreSessionDir = useTempCodexSessionDir();
|
||||
const events = [
|
||||
{ type: "thread.started", thread_id: "019a83a2-permission" },
|
||||
{ type: "turn.started" },
|
||||
{
|
||||
type: "exec_approval_request",
|
||||
call_id: "call-permission-1",
|
||||
command: ["bash", "-lc", "rm -rf /tmp/codex"],
|
||||
cwd,
|
||||
reason: "Requires elevated sandbox permissions",
|
||||
risk: { description: "Deletes files", risk_level: "high" },
|
||||
},
|
||||
{ type: "turn.failed", error: { message: "Approval required" } },
|
||||
];
|
||||
const stub = createStubCodexBinary(events);
|
||||
const client = new CodexAgentClient({ codexPathOverride: stub.path });
|
||||
const config: AgentSessionConfig = { provider: "codex", cwd };
|
||||
const client = new CodexAgentClient();
|
||||
const config: AgentSessionConfig = {
|
||||
provider: "codex",
|
||||
cwd,
|
||||
modeId: "full-access",
|
||||
extra: { codex: { approvalPolicy: "on-request" } },
|
||||
};
|
||||
let session: Awaited<ReturnType<typeof client.createSession>> | null = null;
|
||||
try {
|
||||
session = await client.createSession(config);
|
||||
log("Processing synthetic permission request stream");
|
||||
log("Processing Codex permission request stream");
|
||||
|
||||
let captured: AgentPermissionRequest | null = null;
|
||||
const streamed = session.stream("Trigger permission request");
|
||||
const streamed = session.stream(
|
||||
"Request approval to run the command `pwd`, then run it after approval and stop."
|
||||
);
|
||||
for await (const event of streamed) {
|
||||
if (event.type === "permission_requested") {
|
||||
if (event.type === "permission_requested" && !captured) {
|
||||
captured = event.request;
|
||||
expect(session.getPendingPermissions()).toHaveLength(1);
|
||||
await session.respondToPermission(captured.id, { behavior: "allow" });
|
||||
}
|
||||
if (event.type === "turn_failed") {
|
||||
if (event.type === "turn_completed" || event.type === "turn_failed") {
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
expect(captured).not.toBeNull();
|
||||
expect(session.getPendingPermissions()).toHaveLength(1);
|
||||
|
||||
await session.respondToPermission(captured!.id, { behavior: "allow" });
|
||||
expect(session.getPendingPermissions()).toHaveLength(0);
|
||||
|
||||
const replay: AgentStreamEvent[] = [];
|
||||
for await (const event of session.streamHistory()) {
|
||||
@@ -315,15 +442,23 @@ describe("CodexAgentClient (SDK integration)", () => {
|
||||
event.resolution.behavior === "allow"
|
||||
)
|
||||
).toBe(true);
|
||||
expect(session.getPendingPermissions()).toHaveLength(0);
|
||||
expect(
|
||||
replay.some(
|
||||
(event) =>
|
||||
event.type === "timeline" &&
|
||||
event.provider === "codex" &&
|
||||
event.item.type === "tool_call" &&
|
||||
event.item.server === "permission" &&
|
||||
event.item.status === "granted"
|
||||
)
|
||||
).toBe(true);
|
||||
} finally {
|
||||
await session?.close();
|
||||
stub.cleanup();
|
||||
rmSync(cwd, { recursive: true, force: true });
|
||||
restoreSessionDir();
|
||||
}
|
||||
},
|
||||
30_000
|
||||
180_000
|
||||
);
|
||||
|
||||
test(
|
||||
@@ -369,6 +504,14 @@ describe("CodexAgentClient (SDK integration)", () => {
|
||||
|
||||
expect(stateIncludesUserMessage(liveState, promptMarker)).toBe(true);
|
||||
expect(stateIncludesUserMessage(hydratedState, promptMarker)).toBe(true);
|
||||
expect(stateIncludesSyntheticMessage(hydratedState)).toBe(false);
|
||||
|
||||
const persisted = await client.listPersistedAgents({ limit: 200 });
|
||||
const persistedEntry = persisted.find((entry) => entry.sessionId === handle!.sessionId);
|
||||
expect(persistedEntry).toBeTruthy();
|
||||
const normalizedTitle = (persistedEntry?.title ?? "").toLowerCase();
|
||||
expect(normalizedTitle.startsWith("# agents.md instructions for")).toBe(false);
|
||||
expect(normalizedTitle.startsWith("<environment_context>")).toBe(false);
|
||||
} finally {
|
||||
await session?.close();
|
||||
await resumed?.close();
|
||||
@@ -380,6 +523,29 @@ describe("CodexAgentClient (SDK integration)", () => {
|
||||
);
|
||||
});
|
||||
|
||||
describe("isSyntheticRolloutUserMessage", () => {
|
||||
test("flags AGENTS instruction payloads", () => {
|
||||
const text = `# AGENTS.md instructions for /Users/test
|
||||
|
||||
<INSTRUCTIONS>
|
||||
Stay safe.
|
||||
</INSTRUCTIONS>`;
|
||||
expect(isSyntheticRolloutUserMessage(text)).toBe(true);
|
||||
});
|
||||
|
||||
test("flags environment context payloads", () => {
|
||||
const text = `<environment_context>
|
||||
<cwd>/Users/test/project</cwd>
|
||||
<approval_policy>never</approval_policy>
|
||||
</environment_context>`;
|
||||
expect(isSyntheticRolloutUserMessage(text)).toBe(true);
|
||||
});
|
||||
|
||||
test("allows real user prompts", () => {
|
||||
expect(isSyntheticRolloutUserMessage("investigate flaky tests")).toBe(false);
|
||||
});
|
||||
});
|
||||
|
||||
type StreamHydrationUpdate = {
|
||||
event: Extract<AgentStreamEvent, { type: "timeline" }>;
|
||||
timestamp: Date;
|
||||
@@ -419,3 +585,118 @@ function stateIncludesUserMessage(state: StreamItem[], marker: string): boolean
|
||||
(item) => item.kind === "user_message" && item.text.toLowerCase().includes(marker.toLowerCase())
|
||||
);
|
||||
}
|
||||
|
||||
function stateIncludesSyntheticMessage(state: StreamItem[]): boolean {
|
||||
return state.some((item) => {
|
||||
if (item.kind !== "user_message") {
|
||||
return false;
|
||||
}
|
||||
const normalized = item.text.trim().toLowerCase();
|
||||
return (
|
||||
normalized.startsWith("# agents.md instructions for") ||
|
||||
normalized.startsWith("<environment_context>")
|
||||
);
|
||||
});
|
||||
}
|
||||
|
||||
async function prependSyntheticRolloutEntries(filePath: string, entries: Array<Record<string, unknown>>): Promise<void> {
|
||||
const existing = await fs.readFile(filePath, "utf8");
|
||||
const synthetic = entries.map((entry) => JSON.stringify(entry)).join("\n");
|
||||
const nextContent = synthetic.length ? `${synthetic}\n${existing}` : existing;
|
||||
await fs.writeFile(
|
||||
filePath,
|
||||
nextContent.endsWith("\n") ? nextContent : `${nextContent}\n`,
|
||||
"utf8"
|
||||
);
|
||||
}
|
||||
|
||||
async function resolveRolloutPathFromHandle(handle: AgentPersistenceHandle): Promise<string | null> {
|
||||
const pathFromMetadata = typeof handle.metadata?.codexRolloutPath === "string" ? handle.metadata.codexRolloutPath : null;
|
||||
if (pathFromMetadata && (await fileExists(pathFromMetadata))) {
|
||||
return pathFromMetadata;
|
||||
}
|
||||
|
||||
const sessionDir = typeof handle.metadata?.codexSessionDir === "string" ? handle.metadata.codexSessionDir : null;
|
||||
const sessionId = handle.sessionId ?? (typeof handle.nativeHandle === "string" ? handle.nativeHandle : null);
|
||||
if (!sessionDir || !sessionId) {
|
||||
return null;
|
||||
}
|
||||
return findRolloutInDir(sessionDir, sessionId, 5);
|
||||
}
|
||||
|
||||
async function fileExists(filePath: string): Promise<boolean> {
|
||||
try {
|
||||
await fs.access(filePath);
|
||||
return true;
|
||||
} catch {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
async function findRolloutInDir(root: string, sessionId: string, maxDepth: number): Promise<string | null> {
|
||||
const stack: Array<{ dir: string; depth: number }> = [{ dir: root, depth: 0 }];
|
||||
while (stack.length > 0) {
|
||||
const { dir, depth } = stack.pop()!;
|
||||
let entries: Dirent[];
|
||||
try {
|
||||
entries = await fs.readdir(dir, { withFileTypes: true });
|
||||
} catch {
|
||||
continue;
|
||||
}
|
||||
|
||||
for (const entry of entries) {
|
||||
const entryPath = path.join(dir, entry.name);
|
||||
if (entry.isFile()) {
|
||||
if (
|
||||
entry.name.startsWith("rollout-") &&
|
||||
entry.name.includes(sessionId) &&
|
||||
(entry.name.endsWith(".json") || entry.name.endsWith(".jsonl"))
|
||||
) {
|
||||
return entryPath;
|
||||
}
|
||||
} else if (entry.isDirectory() && depth < maxDepth) {
|
||||
stack.push({ dir: entryPath, depth: depth + 1 });
|
||||
}
|
||||
}
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
function buildSyntheticInstructionEntry(sessionId: string): Record<string, unknown> {
|
||||
const text = [
|
||||
`# AGENTS.md instructions for ${sessionId}`,
|
||||
"",
|
||||
"<INSTRUCTIONS>",
|
||||
"Always obey the operator.",
|
||||
"</INSTRUCTIONS>",
|
||||
].join("\n");
|
||||
return buildSyntheticResponseItem(text);
|
||||
}
|
||||
|
||||
function buildSyntheticEnvironmentEntry(): Record<string, unknown> {
|
||||
const text = [
|
||||
"<environment_context>",
|
||||
" <cwd>/tmp/workspace</cwd>",
|
||||
" <approval_policy>never</approval_policy>",
|
||||
" <sandbox_mode>danger-full-access</sandbox_mode>",
|
||||
"</environment_context>",
|
||||
].join("\n");
|
||||
return buildSyntheticResponseItem(text);
|
||||
}
|
||||
|
||||
function buildSyntheticResponseItem(text: string): Record<string, unknown> {
|
||||
return {
|
||||
timestamp: new Date().toISOString(),
|
||||
type: "response_item",
|
||||
payload: {
|
||||
type: "message",
|
||||
role: "user",
|
||||
content: [
|
||||
{
|
||||
type: "input_text",
|
||||
text,
|
||||
},
|
||||
],
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
@@ -98,6 +98,8 @@ const MODE_PRESETS: Record<
|
||||
type CodexOptionsOverrides = Partial<ThreadOptions> & { skipGitRepoCheck?: boolean };
|
||||
|
||||
const MAX_ROLLOUT_SEARCH_DEPTH = 5;
|
||||
const PERSISTED_TIMELINE_LIMIT = 20;
|
||||
const SHELL_FUNCTION_NAMES = new Set(["shell", "shell_command"]);
|
||||
type ToolCallTimelineItem = Extract<AgentTimelineItem, { type: "tool_call" }>;
|
||||
|
||||
type ExecPermissionRequestPayload = {
|
||||
@@ -334,10 +336,26 @@ class CodexAgentSession implements AgentSession {
|
||||
const turnOptions = this.buildTurnOptions(options);
|
||||
const { events } = await thread.runStreamed(input, turnOptions);
|
||||
|
||||
for await (const rawEvent of events) {
|
||||
yield* this.translateEvent(rawEvent);
|
||||
if (rawEvent.type === "turn.completed" || rawEvent.type === "turn.failed" || rawEvent.type === "error") {
|
||||
break;
|
||||
let finishedNaturally = false;
|
||||
try {
|
||||
for await (const rawEvent of events) {
|
||||
yield* this.translateEvent(rawEvent);
|
||||
if (
|
||||
rawEvent.type === "turn.completed" ||
|
||||
rawEvent.type === "turn.failed" ||
|
||||
rawEvent.type === "error"
|
||||
) {
|
||||
finishedNaturally = true;
|
||||
break;
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
if (!finishedNaturally && typeof events.return === "function") {
|
||||
try {
|
||||
await events.return({ type: "turn.completed" } as ThreadEvent);
|
||||
} catch (error) {
|
||||
console.warn("[CodexAgentSession] Failed to stop Codex stream:", error);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -887,7 +905,7 @@ async function parseRolloutFile(filePath: string, threadId: string): Promise<Age
|
||||
const content = await fs.readFile(filePath, "utf8");
|
||||
const lines = content.split(/\r?\n/).map((line) => line.trim()).filter(Boolean);
|
||||
const events: AgentStreamEvent[] = [{ type: "thread_started", provider: "codex", sessionId: threadId }];
|
||||
const commandCalls = new Map<string, { command: string }>();
|
||||
const commandCalls = new Map<string, { command: string; cwd?: string }>();
|
||||
|
||||
for (const line of lines) {
|
||||
const entry = parseRolloutEntry(line);
|
||||
@@ -933,18 +951,10 @@ function isSessionMetaEntry(value: unknown): value is SessionMetaEntry {
|
||||
return Boolean(value && typeof value === "object" && (value as { type?: unknown }).type === "session_meta");
|
||||
}
|
||||
|
||||
function isResponseItemEntry(value: unknown): value is ResponseItemEntry {
|
||||
return Boolean(value && typeof value === "object" && (value as { type?: unknown }).type === "response_item");
|
||||
}
|
||||
|
||||
function isRolloutMessagePayload(payload: RolloutResponsePayload | undefined): payload is RolloutMessagePayload {
|
||||
return Boolean(payload && payload.type === "message");
|
||||
}
|
||||
|
||||
function handleRolloutResponseItem(
|
||||
payload: RolloutResponsePayload | undefined,
|
||||
events: AgentStreamEvent[],
|
||||
commandCalls: Map<string, { command: string }>
|
||||
commandCalls: Map<string, { command: string; cwd?: string }>
|
||||
): void {
|
||||
if (!payload || typeof payload !== "object") {
|
||||
return;
|
||||
@@ -957,6 +967,9 @@ function handleRolloutResponseItem(
|
||||
if (payload.role === "assistant") {
|
||||
events.push({ type: "timeline", provider: "codex", item: { type: "assistant_message", text, raw: payload } });
|
||||
} else if (payload.role === "user") {
|
||||
if (isSyntheticRolloutUserMessage(text)) {
|
||||
break;
|
||||
}
|
||||
events.push({ type: "timeline", provider: "codex", item: { type: "user_message", text, raw: payload } });
|
||||
}
|
||||
}
|
||||
@@ -998,18 +1011,35 @@ function handleRolloutEventMessage(payload: RolloutEventPayload | undefined, eve
|
||||
function handleRolloutFunctionCall(
|
||||
payload: RolloutFunctionCallPayload,
|
||||
events: AgentStreamEvent[],
|
||||
commandCalls: Map<string, { command: string }>
|
||||
commandCalls: Map<string, { command: string; cwd?: string }>
|
||||
): void {
|
||||
const name = typeof payload.name === "string" ? payload.name : undefined;
|
||||
if (!name) {
|
||||
return;
|
||||
}
|
||||
|
||||
if (name === "shell") {
|
||||
if (SHELL_FUNCTION_NAMES.has(name)) {
|
||||
const args = safeJsonParse<Record<string, unknown>>(payload.arguments);
|
||||
const command = formatCommand(args);
|
||||
const cwd = args && typeof args === "object" && typeof (args as { workdir?: unknown }).workdir === "string"
|
||||
? ((args as { workdir?: unknown }).workdir as string)
|
||||
: undefined;
|
||||
if (command && typeof payload.call_id === "string") {
|
||||
commandCalls.set(payload.call_id, { command });
|
||||
commandCalls.set(payload.call_id, { command, cwd });
|
||||
events.push({
|
||||
type: "timeline",
|
||||
provider: "codex",
|
||||
item: createToolCallTimelineItem({
|
||||
server: "command",
|
||||
tool: "shell",
|
||||
status: payload.status ?? "in_progress",
|
||||
callId: payload.call_id,
|
||||
displayName: buildCommandDisplayName(command),
|
||||
kind: "execute",
|
||||
input: { command, cwd },
|
||||
raw: payload,
|
||||
}),
|
||||
});
|
||||
}
|
||||
return;
|
||||
}
|
||||
@@ -1042,7 +1072,7 @@ function handleRolloutFunctionCall(
|
||||
function finalizeRolloutFunctionCall(
|
||||
payload: RolloutFunctionCallOutputPayload,
|
||||
events: AgentStreamEvent[],
|
||||
commandCalls: Map<string, { command: string }>
|
||||
commandCalls: Map<string, { command: string; cwd?: string }>
|
||||
): void {
|
||||
if (typeof payload.call_id !== "string") {
|
||||
return;
|
||||
@@ -1064,7 +1094,7 @@ function finalizeRolloutFunctionCall(
|
||||
callId: payload.call_id,
|
||||
displayName: buildCommandDisplayName(command.command),
|
||||
kind: "execute",
|
||||
input: { command: command.command },
|
||||
input: { command: command.command, cwd: command.cwd },
|
||||
output: result,
|
||||
raw: { payload, result },
|
||||
}),
|
||||
@@ -1196,11 +1226,6 @@ type SessionMetaEntry = {
|
||||
payload?: SessionMetaPayload;
|
||||
};
|
||||
|
||||
type ResponseItemEntry = {
|
||||
type: "response_item";
|
||||
payload?: RolloutResponsePayload;
|
||||
};
|
||||
|
||||
async function collectRecentRolloutFiles(rootDir: string, limit: number): Promise<RolloutCandidate[]> {
|
||||
const candidates: RolloutCandidate[] = [];
|
||||
async function traverse(dir: string): Promise<void> {
|
||||
@@ -1268,15 +1293,8 @@ async function parseRolloutDescriptor(
|
||||
if (typeof entry.payload.cwd === "string" && entry.payload.cwd.length > 0) {
|
||||
cwd = entry.payload.cwd;
|
||||
}
|
||||
} else if (
|
||||
!title &&
|
||||
isResponseItemEntry(entry) &&
|
||||
isRolloutMessagePayload(entry.payload) &&
|
||||
entry.payload.role === "user"
|
||||
) {
|
||||
title = extractCodexUserPrompt(entry.payload);
|
||||
}
|
||||
if (sessionId && cwd && title) {
|
||||
if (sessionId && cwd) {
|
||||
break;
|
||||
}
|
||||
}
|
||||
@@ -1285,6 +1303,11 @@ async function parseRolloutDescriptor(
|
||||
return null;
|
||||
}
|
||||
|
||||
const timeline = await loadCodexPersistedTimeline(filePath, sessionId);
|
||||
if (!title) {
|
||||
title = timeline.find((item) => item.type === "user_message")?.text ?? null;
|
||||
}
|
||||
|
||||
const persistence: AgentPersistenceHandle = {
|
||||
provider: "codex",
|
||||
sessionId,
|
||||
@@ -1304,21 +1327,10 @@ async function parseRolloutDescriptor(
|
||||
title: (title ?? "").trim() || `Codex session ${sessionId.slice(0, 8)}`,
|
||||
lastActivityAt: mtime,
|
||||
persistence,
|
||||
timeline,
|
||||
};
|
||||
}
|
||||
|
||||
function extractCodexUserPrompt(payload: RolloutMessagePayload): string | null {
|
||||
if (!Array.isArray(payload.content)) {
|
||||
return null;
|
||||
}
|
||||
for (const block of payload.content) {
|
||||
if (block && typeof block === "object" && typeof block.text === "string") {
|
||||
return block.text.trim();
|
||||
}
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
function resolveCodexSessionRoot(): string | null {
|
||||
if (process.env.CODEX_SESSION_DIR) {
|
||||
return process.env.CODEX_SESSION_DIR;
|
||||
@@ -1350,6 +1362,24 @@ function extractMessageText(content: unknown): string {
|
||||
return parts.join("\n").trim();
|
||||
}
|
||||
|
||||
export function isSyntheticRolloutUserMessage(text: string): boolean {
|
||||
const normalized = text.trim();
|
||||
if (!normalized) {
|
||||
return false;
|
||||
}
|
||||
|
||||
const lower = normalized.toLowerCase();
|
||||
if (lower.startsWith("# agents.md instructions for") && lower.includes("<instructions>")) {
|
||||
return true;
|
||||
}
|
||||
|
||||
if (lower.startsWith("<environment_context>")) {
|
||||
return true;
|
||||
}
|
||||
|
||||
return false;
|
||||
}
|
||||
|
||||
function extractReasoningText(payload: RolloutReasoningPayload): string {
|
||||
if (Array.isArray(payload?.summary)) {
|
||||
const text = payload.summary
|
||||
@@ -1367,6 +1397,26 @@ function extractReasoningText(payload: RolloutReasoningPayload): string {
|
||||
return "";
|
||||
}
|
||||
|
||||
async function loadCodexPersistedTimeline(filePath: string, threadId: string): Promise<AgentTimelineItem[]> {
|
||||
try {
|
||||
const events = await parseRolloutFile(filePath, threadId);
|
||||
const timeline: AgentTimelineItem[] = [];
|
||||
for (const event of events) {
|
||||
if (event.type !== "timeline" || event.provider !== "codex") {
|
||||
continue;
|
||||
}
|
||||
timeline.push(event.item);
|
||||
if (timeline.length >= PERSISTED_TIMELINE_LIMIT) {
|
||||
break;
|
||||
}
|
||||
}
|
||||
return timeline;
|
||||
} catch (error) {
|
||||
console.warn(`[CodexAgentSession] Failed to load persisted timeline for ${threadId}:`, error);
|
||||
return [];
|
||||
}
|
||||
}
|
||||
|
||||
function parsePlanItems(args: unknown): { text: string; completed: boolean }[] {
|
||||
const planValue = typeof args === "object" && args ? (args as { plan?: unknown }).plan : undefined;
|
||||
const plan = Array.isArray(planValue) ? planValue : [];
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
import { describe, expect, it, vi } from "vitest";
|
||||
import { WaitForAgentTracker } from "./wait-for-agent-tracker";
|
||||
import { WaitForAgentTracker } from "./wait-for-agent-tracker.js";
|
||||
|
||||
describe("WaitForAgentTracker", () => {
|
||||
it("registers and cancels waiters per agent", () => {
|
||||
@@ -7,7 +7,7 @@ describe("WaitForAgentTracker", () => {
|
||||
const cancelA = vi.fn();
|
||||
const cancelB = vi.fn();
|
||||
|
||||
const unregisterA = tracker.register("agent-a", cancelA);
|
||||
tracker.register("agent-a", cancelA);
|
||||
tracker.register("agent-a", cancelB);
|
||||
|
||||
expect(tracker.cancel("agent-a", "test")).toBe(true);
|
||||
|
||||
@@ -19,10 +19,11 @@ type ProviderEventPayload = Extract<
|
||||
|
||||
export type AgentSnapshotPayload = Omit<
|
||||
AgentSnapshot,
|
||||
"createdAt" | "updatedAt"
|
||||
"createdAt" | "updatedAt" | "lastUserMessageAt"
|
||||
> & {
|
||||
createdAt: string;
|
||||
updatedAt: string;
|
||||
lastUserMessageAt: string | null;
|
||||
title: string | null;
|
||||
};
|
||||
|
||||
@@ -219,6 +220,7 @@ export const AgentSnapshotPayloadSchema = z.object({
|
||||
cwd: z.string(),
|
||||
createdAt: z.string(),
|
||||
updatedAt: z.string(),
|
||||
lastUserMessageAt: z.string().nullable(),
|
||||
status: AgentStatusSchema,
|
||||
sessionId: z.string().nullable(),
|
||||
capabilities: AgentCapabilityFlagsSchema,
|
||||
@@ -239,11 +241,12 @@ export function serializeAgentSnapshot(
|
||||
snapshot: AgentSnapshot,
|
||||
options?: { title?: string | null }
|
||||
): AgentSnapshotPayload {
|
||||
const { createdAt, updatedAt, ...rest } = snapshot;
|
||||
const { createdAt, updatedAt, lastUserMessageAt, ...rest } = snapshot;
|
||||
return {
|
||||
...rest,
|
||||
createdAt: createdAt.toISOString(),
|
||||
updatedAt: updatedAt.toISOString(),
|
||||
lastUserMessageAt: lastUserMessageAt ? lastUserMessageAt.toISOString() : null,
|
||||
title: options?.title ?? null,
|
||||
};
|
||||
}
|
||||
@@ -323,10 +326,28 @@ export const SendAgentAudioSchema = z.object({
|
||||
requestId: z.string().optional(), // Client-provided ID for tracking transcription
|
||||
});
|
||||
|
||||
const GitSetupOptionsSchema = z.object({
|
||||
baseBranch: z.string().optional(),
|
||||
createNewBranch: z.boolean().optional(),
|
||||
newBranchName: z.string().optional(),
|
||||
createWorktree: z.boolean().optional(),
|
||||
worktreeSlug: z.string().optional(),
|
||||
});
|
||||
|
||||
export type GitSetupOptions = z.infer<typeof GitSetupOptionsSchema>;
|
||||
|
||||
export const CreateAgentRequestMessageSchema = z.object({
|
||||
type: z.literal("create_agent_request"),
|
||||
config: AgentSessionConfigSchema,
|
||||
worktreeName: z.string().optional(),
|
||||
initialPrompt: z.string().optional(),
|
||||
git: GitSetupOptionsSchema.optional(),
|
||||
requestId: z.string().optional(),
|
||||
});
|
||||
|
||||
export const GitRepoInfoRequestMessageSchema = z.object({
|
||||
type: z.literal("git_repo_info_request"),
|
||||
cwd: z.string(),
|
||||
requestId: z.string().optional(),
|
||||
});
|
||||
|
||||
@@ -348,6 +369,11 @@ export const CancelAgentRequestMessageSchema = z.object({
|
||||
agentId: z.string(),
|
||||
});
|
||||
|
||||
export const RestartServerRequestMessageSchema = z.object({
|
||||
type: z.literal("restart_server_request"),
|
||||
reason: z.string().optional(),
|
||||
});
|
||||
|
||||
export const ListPersistedAgentsRequestMessageSchema = z.object({
|
||||
type: z.literal("list_persisted_agents_request"),
|
||||
provider: AgentProviderSchema.optional(),
|
||||
@@ -424,12 +450,14 @@ export const SessionInboundMessageSchema = z.discriminatedUnion("type", [
|
||||
ResumeAgentRequestMessageSchema,
|
||||
RefreshAgentRequestMessageSchema,
|
||||
CancelAgentRequestMessageSchema,
|
||||
RestartServerRequestMessageSchema,
|
||||
InitializeAgentRequestMessageSchema,
|
||||
SetAgentModeMessageSchema,
|
||||
AgentPermissionResponseMessageSchema,
|
||||
GitDiffRequestSchema,
|
||||
FileExplorerRequestSchema,
|
||||
ListPersistedAgentsRequestMessageSchema,
|
||||
GitRepoInfoRequestMessageSchema,
|
||||
]);
|
||||
|
||||
export type SessionInboundMessage = z.infer<typeof SessionInboundMessageSchema>;
|
||||
@@ -622,6 +650,7 @@ const PersistedAgentDescriptorPayloadSchema = z.object({
|
||||
title: z.string(),
|
||||
lastActivityAt: z.string(),
|
||||
persistence: AgentPersistenceHandleSchema,
|
||||
timeline: z.array(AgentTimelineItemPayloadSchema),
|
||||
});
|
||||
|
||||
export type PersistedAgentDescriptorPayload = z.infer<typeof PersistedAgentDescriptorPayloadSchema>;
|
||||
@@ -654,6 +683,24 @@ export const FileExplorerResponseSchema = z.object({
|
||||
}),
|
||||
});
|
||||
|
||||
const GitBranchInfoSchema = z.object({
|
||||
name: z.string(),
|
||||
isCurrent: z.boolean(),
|
||||
});
|
||||
|
||||
export const GitRepoInfoResponseSchema = z.object({
|
||||
type: z.literal("git_repo_info_response"),
|
||||
payload: z.object({
|
||||
cwd: z.string(),
|
||||
repoRoot: z.string(),
|
||||
requestId: z.string().optional(),
|
||||
branches: z.array(GitBranchInfoSchema).optional(),
|
||||
currentBranch: z.string().nullable().optional(),
|
||||
isDirty: z.boolean().optional(),
|
||||
error: z.string().optional(),
|
||||
}),
|
||||
});
|
||||
|
||||
export const SessionOutboundMessageSchema = z.discriminatedUnion("type", [
|
||||
ActivityLogMessageSchema,
|
||||
AssistantChunkMessageSchema,
|
||||
@@ -675,6 +722,7 @@ export const SessionOutboundMessageSchema = z.discriminatedUnion("type", [
|
||||
ListPersistedAgentsResponseSchema,
|
||||
GitDiffResponseSchema,
|
||||
FileExplorerResponseSchema,
|
||||
GitRepoInfoResponseSchema,
|
||||
]);
|
||||
|
||||
export type SessionOutboundMessage = z.infer<
|
||||
@@ -722,6 +770,7 @@ export type GitDiffRequest = z.infer<typeof GitDiffRequestSchema>;
|
||||
export type GitDiffResponse = z.infer<typeof GitDiffResponseSchema>;
|
||||
export type FileExplorerRequest = z.infer<typeof FileExplorerRequestSchema>;
|
||||
export type FileExplorerResponse = z.infer<typeof FileExplorerResponseSchema>;
|
||||
export type RestartServerRequestMessage = z.infer<typeof RestartServerRequestMessageSchema>;
|
||||
|
||||
// ============================================================================
|
||||
// WebSocket Level Messages (wraps session messages)
|
||||
|
||||
@@ -15,6 +15,7 @@ function createSnapshot(overrides?: Partial<AgentSnapshot>): AgentSnapshot {
|
||||
cwd: "/tmp/project",
|
||||
createdAt: now,
|
||||
updatedAt: now,
|
||||
lastUserMessageAt: null,
|
||||
status: "idle",
|
||||
sessionId: null,
|
||||
capabilities: {
|
||||
|
||||
@@ -45,7 +45,9 @@ export async function restorePersistedAgents(
|
||||
return;
|
||||
}
|
||||
|
||||
console.log(`[Agents] Restoring ${records.length} persisted agent(s)`);
|
||||
let resumed = 0;
|
||||
let created = 0;
|
||||
|
||||
for (const record of records) {
|
||||
if (!isKnownProvider(record.provider)) {
|
||||
console.warn(
|
||||
@@ -54,27 +56,33 @@ export async function restorePersistedAgents(
|
||||
continue;
|
||||
}
|
||||
|
||||
const handle = buildPersistenceHandle(record);
|
||||
try {
|
||||
const persistenceHandle = buildPersistenceHandle(record);
|
||||
if (persistenceHandle) {
|
||||
if (handle) {
|
||||
await agentManager.resumeAgent(
|
||||
persistenceHandle,
|
||||
handle,
|
||||
buildConfigOverrides(record),
|
||||
record.id
|
||||
);
|
||||
resumed += 1;
|
||||
} else {
|
||||
await agentManager.createAgent(
|
||||
buildSessionConfig(record),
|
||||
record.id
|
||||
);
|
||||
created += 1;
|
||||
}
|
||||
} catch (error) {
|
||||
console.error(
|
||||
`[Agents] Failed to restore agent ${record.id}:`,
|
||||
`[Agents] Failed to restore agent ${record.id} from registry:`,
|
||||
error
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
console.log(
|
||||
`[Agents] Loaded ${records.length} persisted agent record(s); resumed ${resumed}, created ${created}`
|
||||
);
|
||||
}
|
||||
|
||||
function isKnownProvider(provider: string): provider is AgentProvider {
|
||||
|
||||
@@ -18,6 +18,7 @@ import {
|
||||
type SessionInboundMessage,
|
||||
type SessionOutboundMessage,
|
||||
type FileExplorerRequest,
|
||||
type GitSetupOptions,
|
||||
} from "./messages.js";
|
||||
import { getSystemPrompt } from "./agent/system-prompt.js";
|
||||
import { getAllTools } from "./agent/llm-openai.js";
|
||||
@@ -28,6 +29,10 @@ import {
|
||||
listConversations,
|
||||
deleteConversation,
|
||||
} from "./persistence.js";
|
||||
import {
|
||||
buildConfigOverrides,
|
||||
buildSessionConfig,
|
||||
} from "./persistence-hooks.js";
|
||||
import { experimental_createMCPClient } from "ai";
|
||||
import { InMemoryTransport } from "@modelcontextprotocol/sdk/inMemory.js";
|
||||
import { createTerminalMcpServer } from "./terminal-mcp/index.js";
|
||||
@@ -39,8 +44,10 @@ import type {
|
||||
AgentPromptInput,
|
||||
AgentSessionConfig,
|
||||
AgentStreamEvent,
|
||||
AgentProvider,
|
||||
AgentPersistenceHandle,
|
||||
} from "./agent/agent-sdk-types.js";
|
||||
import { AgentRegistry } from "./agent/agent-registry.js";
|
||||
import { AgentRegistry, type StoredAgentRecord } from "./agent/agent-registry.js";
|
||||
import { expandTilde } from "./terminal-mcp/tmux.js";
|
||||
import {
|
||||
listDirectoryEntries,
|
||||
@@ -51,12 +58,28 @@ import {
|
||||
isTitleGeneratorInitialized,
|
||||
} from "../services/agent-title-generator.js";
|
||||
import type { TerminalManager } from "./terminal-mcp/terminal-manager.js";
|
||||
import {
|
||||
createWorktree,
|
||||
detectRepoInfo,
|
||||
slugify,
|
||||
validateBranchSlug,
|
||||
} from "../utils/worktree.js";
|
||||
|
||||
const execAsync = promisify(exec);
|
||||
const ACTIVE_TITLE_GENERATIONS = new Set<string>();
|
||||
let restartRequested = false;
|
||||
const KNOWN_AGENT_PROVIDERS: AgentProvider[] = ["claude", "codex"];
|
||||
|
||||
type ProcessingPhase = "idle" | "transcribing" | "llm";
|
||||
|
||||
type NormalizedGitOptions = {
|
||||
baseBranch?: string;
|
||||
createNewBranch: boolean;
|
||||
newBranchName?: string;
|
||||
createWorktree: boolean;
|
||||
worktreeSlug?: string;
|
||||
};
|
||||
|
||||
const PCM_SAMPLE_RATE = 16000;
|
||||
const PCM_CHANNELS = 1;
|
||||
const PCM_BITS_PER_SAMPLE = 16;
|
||||
@@ -66,6 +89,7 @@ const MIN_STREAMING_SEGMENT_DURATION_MS = 1000;
|
||||
const MIN_STREAMING_SEGMENT_BYTES = Math.round(
|
||||
PCM_BYTES_PER_MS * MIN_STREAMING_SEGMENT_DURATION_MS
|
||||
);
|
||||
const SAFE_GIT_REF_PATTERN = /^[A-Za-z0-9._\/-]+$/;
|
||||
|
||||
/**
|
||||
* Type for present_artifact tool arguments
|
||||
@@ -115,6 +139,45 @@ function convertPCMToWavBuffer(
|
||||
return wavBuffer;
|
||||
}
|
||||
|
||||
function isKnownAgentProvider(value: string): value is AgentProvider {
|
||||
return KNOWN_AGENT_PROVIDERS.includes(value as AgentProvider);
|
||||
}
|
||||
|
||||
function coerceAgentProvider(value: string, agentId?: string): AgentProvider {
|
||||
if (isKnownAgentProvider(value)) {
|
||||
return value;
|
||||
}
|
||||
console.warn(
|
||||
`[Session] Unknown provider '${value}' for agent ${agentId ?? "unknown"}; defaulting to 'claude'`
|
||||
);
|
||||
return "claude";
|
||||
}
|
||||
|
||||
function toAgentPersistenceHandle(
|
||||
handle: StoredAgentRecord["persistence"]
|
||||
): AgentPersistenceHandle | null {
|
||||
if (!handle) {
|
||||
return null;
|
||||
}
|
||||
const provider = handle.provider;
|
||||
if (!isKnownAgentProvider(provider)) {
|
||||
console.warn(
|
||||
`[Session] Ignoring persistence handle with unknown provider '${provider}'`
|
||||
);
|
||||
return null;
|
||||
}
|
||||
if (!handle.sessionId) {
|
||||
console.warn("[Session] Ignoring persistence handle missing sessionId");
|
||||
return null;
|
||||
}
|
||||
return {
|
||||
provider,
|
||||
sessionId: handle.sessionId,
|
||||
nativeHandle: handle.nativeHandle,
|
||||
metadata: handle.metadata,
|
||||
} satisfies AgentPersistenceHandle;
|
||||
}
|
||||
|
||||
/**
|
||||
* Session represents a single client conversation session.
|
||||
* It owns all state management, orchestration logic, and message processing.
|
||||
@@ -460,6 +523,43 @@ export class Session {
|
||||
return serializeAgentSnapshot(agent, { title });
|
||||
}
|
||||
|
||||
private buildStoredAgentPayload(record: StoredAgentRecord): AgentSnapshotPayload {
|
||||
const defaultCapabilities = {
|
||||
supportsStreaming: false,
|
||||
supportsSessionPersistence: true,
|
||||
supportsDynamicModes: false,
|
||||
supportsMcpServers: false,
|
||||
supportsReasoningStream: false,
|
||||
supportsToolInvocations: true,
|
||||
} as const;
|
||||
|
||||
const createdAt = new Date(record.createdAt);
|
||||
const updatedAt = new Date(record.lastActivityAt ?? record.updatedAt);
|
||||
const lastUserMessageAt = record.lastUserMessageAt
|
||||
? new Date(record.lastUserMessageAt)
|
||||
: null;
|
||||
|
||||
const provider = coerceAgentProvider(record.provider, record.id);
|
||||
return {
|
||||
id: record.id,
|
||||
provider,
|
||||
cwd: record.cwd,
|
||||
createdAt: createdAt.toISOString(),
|
||||
updatedAt: updatedAt.toISOString(),
|
||||
lastUserMessageAt: lastUserMessageAt ? lastUserMessageAt.toISOString() : null,
|
||||
status: (record.lastStatus as any) ?? "closed",
|
||||
sessionId: null,
|
||||
capabilities: defaultCapabilities,
|
||||
currentModeId: record.lastModeId ?? null,
|
||||
availableModes: [],
|
||||
pendingPermissions: [],
|
||||
persistence: toAgentPersistenceHandle(record.persistence),
|
||||
lastUsage: undefined,
|
||||
lastError: undefined,
|
||||
title: record.title ?? null,
|
||||
};
|
||||
}
|
||||
|
||||
private async forwardAgentState(agent: AgentSnapshot): Promise<void> {
|
||||
try {
|
||||
const payload = await this.buildAgentPayload(agent);
|
||||
@@ -609,13 +709,17 @@ export class Session {
|
||||
await this.handleRefreshAgentRequest(msg);
|
||||
break;
|
||||
|
||||
case "cancel_agent_request":
|
||||
await this.handleCancelAgentRequest(msg.agentId);
|
||||
break;
|
||||
case "cancel_agent_request":
|
||||
await this.handleCancelAgentRequest(msg.agentId);
|
||||
break;
|
||||
|
||||
case "initialize_agent_request":
|
||||
await this.handleInitializeAgentRequest(msg.agentId, msg.requestId);
|
||||
break;
|
||||
case "restart_server_request":
|
||||
await this.handleRestartServerRequest(msg.reason);
|
||||
break;
|
||||
|
||||
case "initialize_agent_request":
|
||||
await this.handleInitializeAgentRequest(msg.agentId, msg.requestId);
|
||||
break;
|
||||
|
||||
case "set_agent_mode":
|
||||
await this.handleSetAgentMode(msg.agentId, msg.modeId);
|
||||
@@ -640,6 +744,10 @@ export class Session {
|
||||
case "list_persisted_agents_request":
|
||||
await this.handleListPersistedAgentsRequest(msg);
|
||||
break;
|
||||
|
||||
case "git_repo_info_request":
|
||||
await this.handleGitRepoInfoRequest(msg);
|
||||
break;
|
||||
}
|
||||
} catch (error: any) {
|
||||
console.error(
|
||||
@@ -740,6 +848,34 @@ export class Session {
|
||||
}
|
||||
}
|
||||
|
||||
private async handleRestartServerRequest(reason?: string): Promise<void> {
|
||||
if (restartRequested) {
|
||||
console.log(
|
||||
`[Session ${this.clientId}] Restart already requested, ignoring duplicate`
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
restartRequested = true;
|
||||
const payload: { status: string } & Record<string, unknown> = {
|
||||
status: "restart_requested",
|
||||
clientId: this.clientId,
|
||||
};
|
||||
if (reason && reason.trim().length > 0) {
|
||||
payload.reason = reason;
|
||||
}
|
||||
|
||||
console.warn(`[Session ${this.clientId}] Restart requested via websocket`);
|
||||
this.emit({
|
||||
type: "status",
|
||||
payload,
|
||||
});
|
||||
|
||||
setTimeout(() => {
|
||||
process.exit(0);
|
||||
}, 250);
|
||||
}
|
||||
|
||||
private async handleDeleteAgentRequest(agentId: string): Promise<void> {
|
||||
console.log(
|
||||
`[Session ${this.clientId}] Deleting agent ${agentId} from registry`
|
||||
@@ -931,13 +1067,30 @@ export class Session {
|
||||
);
|
||||
|
||||
try {
|
||||
const snapshot = this.agentManager.getAgent(agentId);
|
||||
let snapshot = this.agentManager.getAgent(agentId);
|
||||
if (!snapshot) {
|
||||
throw new Error(`Agent not found: ${agentId}`);
|
||||
const record = await this.agentRegistry.get(agentId);
|
||||
if (!record) {
|
||||
throw new Error(`Agent not found: ${agentId}`);
|
||||
}
|
||||
|
||||
const handle = toAgentPersistenceHandle(record.persistence);
|
||||
if (handle) {
|
||||
snapshot = await this.agentManager.resumeAgent(
|
||||
handle,
|
||||
buildConfigOverrides(record),
|
||||
agentId
|
||||
);
|
||||
} else {
|
||||
const config = buildSessionConfig(record);
|
||||
snapshot = await this.agentManager.createAgent(config, agentId);
|
||||
}
|
||||
}
|
||||
|
||||
await this.agentManager.primeAgentHistory(agentId);
|
||||
await this.forwardAgentState(snapshot);
|
||||
|
||||
// Send timeline snapshot after hydration (if any)
|
||||
const timelineSize = this.emitAgentTimelineSnapshot(snapshot);
|
||||
|
||||
if (requestId) {
|
||||
@@ -978,7 +1131,7 @@ export class Session {
|
||||
private async handleCreateAgentRequest(
|
||||
msg: Extract<SessionInboundMessage, { type: "create_agent_request" }>
|
||||
): Promise<void> {
|
||||
const { config, worktreeName, requestId } = msg;
|
||||
const { config, worktreeName, requestId, initialPrompt, git } = msg;
|
||||
console.log(
|
||||
`[Session ${this.clientId}] Creating agent in ${config.cwd} (${config.provider})${
|
||||
worktreeName ? ` with worktree ${worktreeName}` : ""
|
||||
@@ -988,6 +1141,7 @@ export class Session {
|
||||
try {
|
||||
const sessionConfig = await this.buildAgentSessionConfig(
|
||||
config,
|
||||
git,
|
||||
worktreeName
|
||||
);
|
||||
const snapshot = await this.agentManager.createAgent(sessionConfig);
|
||||
@@ -1012,6 +1166,28 @@ export class Session {
|
||||
|
||||
await this.forwardAgentState(snapshot);
|
||||
|
||||
const trimmedPrompt = initialPrompt?.trim();
|
||||
if (trimmedPrompt && trimmedPrompt.length > 0) {
|
||||
try {
|
||||
this.agentManager.recordUserMessage(snapshot.id, trimmedPrompt);
|
||||
} catch (recordError) {
|
||||
console.error(
|
||||
`[Session ${this.clientId}] Failed to record initial prompt for agent ${snapshot.id}:`,
|
||||
recordError
|
||||
);
|
||||
}
|
||||
|
||||
try {
|
||||
const normalizedPrompt = this.buildAgentPrompt(trimmedPrompt);
|
||||
this.startAgentStream(snapshot.id, normalizedPrompt);
|
||||
} catch (promptError) {
|
||||
console.error(
|
||||
`[Session ${this.clientId}] Failed to run initial prompt for agent ${snapshot.id}:`,
|
||||
promptError
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
if (requestId) {
|
||||
this.emit({
|
||||
type: "status",
|
||||
@@ -1031,6 +1207,16 @@ export class Session {
|
||||
`[Session ${this.clientId}] Failed to create agent:`,
|
||||
error
|
||||
);
|
||||
if (requestId) {
|
||||
this.emit({
|
||||
type: "status",
|
||||
payload: {
|
||||
status: "agent_create_failed",
|
||||
requestId,
|
||||
error: (error as Error)?.message ?? String(error),
|
||||
},
|
||||
});
|
||||
}
|
||||
this.emit({
|
||||
type: "activity_log",
|
||||
payload: {
|
||||
@@ -1068,6 +1254,7 @@ export class Session {
|
||||
try {
|
||||
const snapshot = await this.agentManager.resumeAgent(handle, overrides);
|
||||
this.setCachedTitle(snapshot.id, null);
|
||||
await this.agentManager.primeAgentHistory(snapshot.id);
|
||||
await this.forwardAgentState(snapshot);
|
||||
const timelineSize = this.emitAgentTimelineSnapshot(snapshot);
|
||||
if (requestId) {
|
||||
@@ -1107,9 +1294,31 @@ export class Session {
|
||||
);
|
||||
|
||||
try {
|
||||
const snapshot = await this.agentManager.refreshAgentFromPersistence(
|
||||
agentId
|
||||
);
|
||||
let snapshot: AgentSnapshot;
|
||||
const existing = this.agentManager.getAgent(agentId);
|
||||
if (existing) {
|
||||
snapshot = await this.agentManager.refreshAgentFromPersistence(
|
||||
agentId
|
||||
);
|
||||
} else {
|
||||
const record = await this.agentRegistry.get(agentId);
|
||||
if (!record) {
|
||||
throw new Error(`Agent not found: ${agentId}`);
|
||||
}
|
||||
const handle = toAgentPersistenceHandle(record.persistence);
|
||||
if (!handle) {
|
||||
throw new Error(
|
||||
`Agent ${agentId} cannot be refreshed because it lacks persistence`
|
||||
);
|
||||
}
|
||||
snapshot = await this.agentManager.resumeAgent(
|
||||
handle,
|
||||
buildConfigOverrides(record),
|
||||
agentId
|
||||
);
|
||||
this.setCachedTitle(agentId, null);
|
||||
}
|
||||
await this.agentManager.primeAgentHistory(agentId);
|
||||
await this.forwardAgentState(snapshot);
|
||||
const timelineSize = this.emitAgentTimelineSnapshot(snapshot);
|
||||
if (requestId) {
|
||||
@@ -1158,20 +1367,57 @@ export class Session {
|
||||
|
||||
private async buildAgentSessionConfig(
|
||||
config: AgentSessionConfig,
|
||||
worktreeName?: string
|
||||
gitOptions?: GitSetupOptions,
|
||||
legacyWorktreeName?: string
|
||||
): Promise<AgentSessionConfig> {
|
||||
let cwd = expandTilde(config.cwd);
|
||||
const normalized = this.normalizeGitOptions(gitOptions, legacyWorktreeName);
|
||||
|
||||
if (worktreeName) {
|
||||
const { createWorktree } = await import("../utils/worktree.js");
|
||||
console.log(
|
||||
`[Session ${this.clientId}] Creating worktree '${worktreeName}' from ${cwd}`
|
||||
);
|
||||
const worktreeConfig = await createWorktree({
|
||||
branchName: worktreeName,
|
||||
if (!normalized) {
|
||||
return {
|
||||
...config,
|
||||
cwd,
|
||||
};
|
||||
}
|
||||
|
||||
if (!normalized.createWorktree) {
|
||||
await this.ensureCleanWorkingTree(cwd);
|
||||
}
|
||||
|
||||
if (normalized.createWorktree) {
|
||||
const targetBranch = normalized.createNewBranch
|
||||
? normalized.newBranchName
|
||||
: normalized.baseBranch;
|
||||
|
||||
if (!targetBranch) {
|
||||
throw new Error(
|
||||
"A branch name is required when creating a worktree."
|
||||
);
|
||||
}
|
||||
|
||||
console.log(
|
||||
`[Session ${this.clientId}] Creating worktree '${
|
||||
normalized.worktreeSlug ?? targetBranch
|
||||
}' for branch ${targetBranch}`
|
||||
);
|
||||
|
||||
const worktreeConfig = await createWorktree({
|
||||
branchName: targetBranch,
|
||||
cwd,
|
||||
baseBranch: normalized.createNewBranch
|
||||
? normalized.baseBranch
|
||||
: undefined,
|
||||
worktreeSlug: normalized.worktreeSlug ?? targetBranch,
|
||||
});
|
||||
cwd = worktreeConfig.worktreePath;
|
||||
} else if (normalized.createNewBranch) {
|
||||
await this.createBranchFromBase({
|
||||
cwd,
|
||||
baseBranch: normalized.baseBranch ?? "HEAD",
|
||||
newBranchName: normalized.newBranchName!,
|
||||
});
|
||||
} else if (normalized.baseBranch) {
|
||||
await this.checkoutExistingBranch(cwd, normalized.baseBranch);
|
||||
}
|
||||
|
||||
return {
|
||||
@@ -1180,6 +1426,212 @@ export class Session {
|
||||
};
|
||||
}
|
||||
|
||||
private async handleGitRepoInfoRequest(
|
||||
msg: Extract<SessionInboundMessage, { type: "git_repo_info_request" }>
|
||||
): Promise<void> {
|
||||
const { cwd, requestId } = msg;
|
||||
const resolvedCwd = expandTilde(cwd);
|
||||
|
||||
try {
|
||||
const repoInfo = await detectRepoInfo(resolvedCwd);
|
||||
const { stdout: branchesRaw } = await execAsync(
|
||||
"git branch --format='%(refname:short)'",
|
||||
{ cwd: repoInfo.path }
|
||||
);
|
||||
const { stdout: currentRaw } = await execAsync(
|
||||
"git rev-parse --abbrev-ref HEAD",
|
||||
{ cwd: resolvedCwd }
|
||||
);
|
||||
const currentBranch = currentRaw.trim();
|
||||
const branches = branchesRaw
|
||||
.split("\n")
|
||||
.map((line) => line.trim())
|
||||
.filter((line) => line.length > 0)
|
||||
.map((name) => ({
|
||||
name,
|
||||
isCurrent: name === currentBranch,
|
||||
}));
|
||||
|
||||
const isDirty = await this.isWorkingTreeDirty(resolvedCwd);
|
||||
|
||||
this.emit({
|
||||
type: "git_repo_info_response",
|
||||
payload: {
|
||||
cwd: resolvedCwd,
|
||||
repoRoot: repoInfo.path,
|
||||
requestId,
|
||||
branches,
|
||||
currentBranch: currentBranch || null,
|
||||
isDirty,
|
||||
},
|
||||
});
|
||||
} catch (error) {
|
||||
this.emit({
|
||||
type: "git_repo_info_response",
|
||||
payload: {
|
||||
cwd,
|
||||
repoRoot: cwd,
|
||||
requestId,
|
||||
error: (error as Error)?.message ?? String(error),
|
||||
},
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
private normalizeGitOptions(
|
||||
gitOptions?: GitSetupOptions,
|
||||
legacyWorktreeName?: string
|
||||
): NormalizedGitOptions | null {
|
||||
const fallbackOptions: GitSetupOptions | undefined = legacyWorktreeName
|
||||
? {
|
||||
createWorktree: true,
|
||||
createNewBranch: true,
|
||||
newBranchName: legacyWorktreeName,
|
||||
worktreeSlug: legacyWorktreeName,
|
||||
}
|
||||
: undefined;
|
||||
|
||||
const merged = gitOptions ?? fallbackOptions;
|
||||
if (!merged) {
|
||||
return null;
|
||||
}
|
||||
|
||||
const baseBranch = merged.baseBranch?.trim() || undefined;
|
||||
const createWorktree = Boolean(merged.createWorktree);
|
||||
const createNewBranch = Boolean(merged.createNewBranch);
|
||||
const normalizedBranchName = merged.newBranchName
|
||||
? slugify(merged.newBranchName)
|
||||
: undefined;
|
||||
const normalizedWorktreeSlug = merged.worktreeSlug
|
||||
? slugify(merged.worktreeSlug)
|
||||
: normalizedBranchName;
|
||||
|
||||
if (!createWorktree && !createNewBranch && !baseBranch) {
|
||||
return null;
|
||||
}
|
||||
|
||||
if (baseBranch) {
|
||||
this.assertSafeGitRef(baseBranch, "base branch");
|
||||
}
|
||||
|
||||
if (createNewBranch) {
|
||||
if (!normalizedBranchName) {
|
||||
throw new Error("New branch name is required");
|
||||
}
|
||||
const validation = validateBranchSlug(normalizedBranchName);
|
||||
if (!validation.valid) {
|
||||
throw new Error(`Invalid branch name: ${validation.error}`);
|
||||
}
|
||||
}
|
||||
|
||||
if (normalizedWorktreeSlug) {
|
||||
const validation = validateBranchSlug(normalizedWorktreeSlug);
|
||||
if (!validation.valid) {
|
||||
throw new Error(`Invalid worktree name: ${validation.error}`);
|
||||
}
|
||||
}
|
||||
|
||||
if (createWorktree && !createNewBranch && !baseBranch) {
|
||||
throw new Error(
|
||||
"Base branch is required when creating a worktree without a new branch"
|
||||
);
|
||||
}
|
||||
|
||||
return {
|
||||
baseBranch,
|
||||
createNewBranch,
|
||||
newBranchName: normalizedBranchName,
|
||||
createWorktree,
|
||||
worktreeSlug: normalizedWorktreeSlug,
|
||||
};
|
||||
}
|
||||
|
||||
private assertSafeGitRef(ref: string, label: string): void {
|
||||
if (!SAFE_GIT_REF_PATTERN.test(ref) || ref.includes("..") || ref.includes("@{")) {
|
||||
throw new Error(`Invalid ${label}: ${ref}`);
|
||||
}
|
||||
}
|
||||
|
||||
private async ensureCleanWorkingTree(cwd: string): Promise<void> {
|
||||
const dirty = await this.isWorkingTreeDirty(cwd);
|
||||
if (dirty) {
|
||||
throw new Error(
|
||||
"Working directory has uncommitted changes. Commit or stash before switching branches."
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
private async isWorkingTreeDirty(cwd: string): Promise<boolean> {
|
||||
try {
|
||||
const { stdout } = await execAsync("git status --porcelain", { cwd });
|
||||
return stdout.trim().length > 0;
|
||||
} catch (error) {
|
||||
throw new Error(
|
||||
`Unable to inspect git status for ${cwd}: ${(error as Error).message}`
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
private async checkoutExistingBranch(
|
||||
cwd: string,
|
||||
branch: string
|
||||
): Promise<void> {
|
||||
this.assertSafeGitRef(branch, "branch");
|
||||
try {
|
||||
await execAsync(`git rev-parse --verify ${branch}`, { cwd });
|
||||
} catch (error) {
|
||||
throw new Error(`Branch not found: ${branch}`);
|
||||
}
|
||||
|
||||
const { stdout } = await execAsync("git rev-parse --abbrev-ref HEAD", {
|
||||
cwd,
|
||||
});
|
||||
const current = stdout.trim();
|
||||
if (current === branch) {
|
||||
return;
|
||||
}
|
||||
|
||||
await execAsync(`git checkout ${branch}`, { cwd });
|
||||
}
|
||||
|
||||
private async createBranchFromBase(params: {
|
||||
cwd: string;
|
||||
baseBranch: string;
|
||||
newBranchName: string;
|
||||
}): Promise<void> {
|
||||
const { cwd, baseBranch, newBranchName } = params;
|
||||
this.assertSafeGitRef(baseBranch, "base branch");
|
||||
|
||||
try {
|
||||
await execAsync(`git rev-parse --verify ${baseBranch}`, { cwd });
|
||||
} catch (error) {
|
||||
throw new Error(`Base branch not found: ${baseBranch}`);
|
||||
}
|
||||
|
||||
const exists = await this.doesLocalBranchExist(cwd, newBranchName);
|
||||
if (exists) {
|
||||
throw new Error(`Branch already exists: ${newBranchName}`);
|
||||
}
|
||||
|
||||
await execAsync(`git checkout -b ${newBranchName} ${baseBranch}`, {
|
||||
cwd,
|
||||
});
|
||||
}
|
||||
|
||||
private async doesLocalBranchExist(
|
||||
cwd: string,
|
||||
branch: string
|
||||
): Promise<boolean> {
|
||||
try {
|
||||
await execAsync(`git show-ref --verify --quiet refs/heads/${branch}` , {
|
||||
cwd,
|
||||
});
|
||||
return true;
|
||||
} catch (error: any) {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
private async handleListPersistedAgentsRequest(
|
||||
msg: Extract<SessionInboundMessage, { type: "list_persisted_agents_request" }>
|
||||
): Promise<void> {
|
||||
@@ -1199,6 +1651,7 @@ export class Session {
|
||||
title: entry.title ?? `Session ${entry.sessionId.slice(0, 8)}`,
|
||||
lastActivityAt: entry.lastActivityAt.toISOString(),
|
||||
persistence: entry.persistence,
|
||||
timeline: entry.timeline ?? [],
|
||||
})),
|
||||
},
|
||||
});
|
||||
@@ -1438,10 +1891,19 @@ export class Session {
|
||||
try {
|
||||
// Get live agents with session modes
|
||||
const agentSnapshots = this.agentManager.listAgents();
|
||||
const agents = await Promise.all(
|
||||
const liveAgents = await Promise.all(
|
||||
agentSnapshots.map((agent) => this.buildAgentPayload(agent))
|
||||
);
|
||||
|
||||
// Add persisted agents that have not been lazily initialized yet
|
||||
const registryRecords = await this.agentRegistry.list();
|
||||
const liveIds = new Set(agentSnapshots.map((a) => a.id));
|
||||
const persistedAgents = registryRecords
|
||||
.filter((record) => !liveIds.has(record.id))
|
||||
.map((record) => this.buildStoredAgentPayload(record));
|
||||
|
||||
const agents = [...liveAgents, ...persistedAgents];
|
||||
|
||||
// Get live commands from terminal manager
|
||||
let commands: any[] = [];
|
||||
if (this.terminalManager) {
|
||||
@@ -1467,11 +1929,6 @@ export class Session {
|
||||
console.log(
|
||||
`[Session ${this.clientId}] Sent session state: ${agents.length} agents, ${commands.length} commands`
|
||||
);
|
||||
|
||||
for (const agent of agentSnapshots) {
|
||||
this.emitAgentTimelineSnapshot(agent);
|
||||
void this.maybeGenerateAgentTitle(agent.id);
|
||||
}
|
||||
} catch (error) {
|
||||
console.error(
|
||||
`[Session ${this.clientId}] Failed to send session state:`,
|
||||
|
||||
@@ -18,6 +18,13 @@ interface WorktreeConfig {
|
||||
repoPath: string;
|
||||
}
|
||||
|
||||
interface CreateWorktreeOptions {
|
||||
branchName: string;
|
||||
cwd: string;
|
||||
baseBranch?: string;
|
||||
worktreeSlug?: string;
|
||||
}
|
||||
|
||||
/**
|
||||
* Check if current directory is a bare repository
|
||||
*/
|
||||
@@ -148,6 +155,11 @@ export function slugify(input: string): string {
|
||||
.replace(/^-+|-+$/g, "");
|
||||
}
|
||||
|
||||
function sanitizeWorktreeSlug(input: string): string {
|
||||
const slug = slugify(input);
|
||||
return slug.length > 0 ? slug : "worktree";
|
||||
}
|
||||
|
||||
/**
|
||||
* Check if a worktree already exists for a branch
|
||||
*/
|
||||
@@ -183,12 +195,12 @@ async function findExistingWorktree(
|
||||
/**
|
||||
* Create a git worktree with proper naming conventions
|
||||
*/
|
||||
export async function createWorktree(params: {
|
||||
branchName: string;
|
||||
cwd: string;
|
||||
}): Promise<WorktreeConfig> {
|
||||
const { branchName, cwd } = params;
|
||||
|
||||
export async function createWorktree({
|
||||
branchName,
|
||||
cwd,
|
||||
baseBranch,
|
||||
worktreeSlug,
|
||||
}: CreateWorktreeOptions): Promise<WorktreeConfig> {
|
||||
// Validate branch name
|
||||
const validation = validateBranchSlug(branchName);
|
||||
if (!validation.valid) {
|
||||
@@ -200,11 +212,13 @@ export async function createWorktree(params: {
|
||||
|
||||
// Determine worktree directory based on repo type
|
||||
let worktreePath: string;
|
||||
const desiredSlug = sanitizeWorktreeSlug(worktreeSlug ?? branchName);
|
||||
|
||||
if (repoInfo.type === "bare") {
|
||||
worktreePath = join(repoInfo.path, branchName);
|
||||
worktreePath = join(repoInfo.path, desiredSlug);
|
||||
} else {
|
||||
const parentDir = dirname(repoInfo.path);
|
||||
const worktreeName = `${repoInfo.name}-${branchName.replace(/\//g, "-")}`;
|
||||
const worktreeName = `${repoInfo.name}-${desiredSlug}`;
|
||||
worktreePath = join(parentDir, worktreeName);
|
||||
}
|
||||
|
||||
@@ -238,10 +252,9 @@ export async function createWorktree(params: {
|
||||
});
|
||||
} catch (error) {
|
||||
// Branch doesn't exist, create new branch and worktree
|
||||
await execAsync(
|
||||
`git worktree add "${worktreePath}" -b "${branchName}"`,
|
||||
{ cwd: repoInfo.path }
|
||||
);
|
||||
const baseArg = baseBranch ? ` "${baseBranch}"` : "";
|
||||
const command = `git worktree add "${worktreePath}" -b "${branchName}"${baseArg}`;
|
||||
await execAsync(command, { cwd: repoInfo.path });
|
||||
}
|
||||
|
||||
// Copy .env file if it exists
|
||||
|
||||
Reference in New Issue
Block a user