Update files

This commit is contained in:
Mohamed Boudra
2026-01-28 15:25:12 +07:00
parent 96b211a2cf
commit f7935f9cc1
23 changed files with 1336 additions and 856 deletions
@@ -381,6 +381,7 @@ const styles = StyleSheet.create((theme) => ({
},
sectionHeader: {
paddingVertical: theme.spacing[2],
paddingHorizontal: theme.spacing[3],
marginTop: theme.spacing[2],
},
sectionTitle: {
@@ -62,8 +62,7 @@ export function AgentStatusBar({ agentId, serverId }: AgentStatusBarProps) {
</DropdownMenuTrigger>
<DropdownMenuContent
side="top"
align="center"
fullWidth
align="start"
testID="agent-mode-menu"
>
<DropdownMenuLabel>Mode</DropdownMenuLabel>
@@ -17,7 +17,7 @@ import { router, usePathname } from "expo-router";
import { StyleSheet, useUnistyles } from "react-native-unistyles";
import { type GestureType } from "react-native-gesture-handler";
import { useQueries, useQueryClient, type UseQueryOptions } from "@tanstack/react-query";
import { ChevronDown, ChevronRight, Plus } from "lucide-react-native";
import { Archive, ChevronDown, ChevronRight, Plus } from "lucide-react-native";
import {
DraggableList,
type DraggableRenderItemInfo,
@@ -394,8 +394,21 @@ export function GroupedAgentList({
[setProjectOrder]
);
const handleArchiveAgent = useCallback(
(e: { stopPropagation: () => void }, agent: AggregatedAgent) => {
e.stopPropagation();
const session = useSessionStore.getState().sessions[agent.serverId];
const archiveMethod = session?.methods?.archiveAgent;
if (archiveMethod) {
archiveMethod(agent.id);
}
},
[]
);
const AgentListRow = useCallback(
({ agent }: { agent: AggregatedAgent }) => {
const [isHovered, setIsHovered] = useState(false);
const timeAgo = formatTimeAgo(agent.lastActivityAt);
const agentKey = `${agent.serverId}:${agent.id}`;
const isSelected = selectedAgentId === agentKey;
@@ -415,49 +428,74 @@ export function GroupedAgentList({
? (checkout.currentBranch ?? checkout.baseRef ?? "git")
: null;
const canArchive = !isRunning && !agent.requiresAttention;
return (
<Pressable
style={({ pressed, hovered }) => [
style={({ pressed }) => [
styles.agentItem,
isSelected && styles.agentItemSelected,
hovered && styles.agentItemHovered,
isHovered && styles.agentItemHovered,
pressed && styles.agentItemPressed,
]}
onPress={() => handleAgentPress(agent.serverId, agent.id)}
onLongPress={() => handleAgentLongPress(agent)}
onHoverIn={() => setIsHovered(true)}
onHoverOut={() => setIsHovered(false)}
testID={`agent-row-${agent.serverId}-${agent.id}`}
>
{({ hovered }) => (
<View style={styles.agentContent}>
<View style={styles.row}>
{statusColor && (
<View
style={[styles.statusDot, { backgroundColor: statusColor }]}
/>
)}
<Text
style={[
styles.agentTitle,
(isSelected || hovered) && styles.agentTitleHighlighted,
]}
numberOfLines={1}
>
{agent.title || "New agent"}
</Text>
</View>
<Text style={styles.secondaryRow} numberOfLines={1}>
{activeBranchLabel ? `${activeBranchLabel} · ${timeAgo}` : timeAgo}
<View style={styles.agentContent}>
<View style={styles.row}>
{statusColor && (
<View
style={[styles.statusDot, { backgroundColor: statusColor }]}
/>
)}
<Text
style={[
styles.agentTitle,
(isSelected || isHovered) && styles.agentTitleHighlighted,
]}
numberOfLines={1}
>
{agent.title || "New agent"}
</Text>
{isHovered && canArchive && (
<Pressable
style={styles.archiveButton}
onPress={(e) => handleArchiveAgent(e, agent)}
onHoverIn={() => setIsHovered(true)}
onHoverOut={() => setIsHovered(true)}
testID={`agent-archive-${agent.serverId}-${agent.id}`}
>
{({ hovered: archiveHovered }) => (
<Archive
size={14}
color={
archiveHovered
? theme.colors.foreground
: theme.colors.foregroundMuted
}
/>
)}
</Pressable>
)}
</View>
)}
<Text style={styles.secondaryRow} numberOfLines={1}>
{activeBranchLabel ? `${activeBranchLabel} · ${timeAgo}` : timeAgo}
</Text>
</View>
</Pressable>
);
},
[
handleAgentLongPress,
handleAgentPress,
handleArchiveAgent,
selectedAgentId,
theme.colors.foreground,
theme.colors.foregroundMuted,
theme.colors.palette.blue,
theme.colors.success,
]
@@ -661,6 +699,11 @@ const styles = StyleSheet.create((theme) => ({
color: theme.colors.foreground,
opacity: 0.8,
},
archiveButton: {
alignItems: "center",
justifyContent: "center",
marginLeft: theme.spacing[1],
},
agentTitleHighlighted: {
color: theme.colors.foreground,
opacity: 1,
@@ -220,6 +220,9 @@ function normalizeAgentSnapshot(
const attentionTimestamp = snapshot.attentionTimestamp
? new Date(snapshot.attentionTimestamp)
: null;
const archivedAt = snapshot.archivedAt
? new Date(snapshot.archivedAt)
: null;
return {
serverId,
id: snapshot.id,
@@ -244,6 +247,7 @@ function normalizeAgentSnapshot(
attentionReason: snapshot.attentionReason ?? null,
attentionTimestamp,
parentAgentId: snapshot.parentAgentId,
archivedAt,
};
}
@@ -297,6 +301,7 @@ export interface SessionContextValue {
images?: Array<{ uri: string; mimeType?: string }>
) => Promise<void>;
deleteAgent: (agentId: string) => void;
archiveAgent: (agentId: string) => void;
createAgent: (options: {
config: any;
initialPrompt: string;
@@ -1436,6 +1441,27 @@ export function SessionProvider({
});
});
const unsubAgentArchived = client.on("agent_archived", (message) => {
if (message.type !== "agent_archived") {
return;
}
const { agentId, archivedAt } = message.payload;
console.log("[Session] Agent archived:", agentId);
setAgents(serverId, (prev) => {
const existing = prev.get(agentId);
if (!existing) {
return prev;
}
const next = new Map(prev);
next.set(agentId, {
...existing,
archivedAt: new Date(archivedAt),
});
return next;
});
});
return () => {
unsubSessionState();
unsubAgentState();
@@ -1450,6 +1476,7 @@ export function SessionProvider({
unsubTranscription();
unsubProviderModels();
unsubAgentDeleted();
unsubAgentArchived();
};
}, [
client,
@@ -1837,6 +1864,19 @@ export function SessionProvider({
[client]
);
const archiveAgent = useCallback(
(agentId: string) => {
if (!client) {
console.warn("[Session] archiveAgent skipped: daemon unavailable");
return;
}
void client.archiveAgent(agentId).catch((error) => {
console.error("[Session] Failed to archive agent:", error);
});
},
[client]
);
const restartServer = useCallback(
(reason?: string) => {
if (!client) {
@@ -2158,6 +2198,7 @@ export function SessionProvider({
refreshSession,
cancelAgentRun,
deleteAgent,
archiveAgent,
sendAgentMessage,
createAgent,
setAgentMode,
@@ -2181,6 +2222,7 @@ export function SessionProvider({
refreshSession,
cancelAgentRun,
deleteAgent,
archiveAgent,
sendAgentMessage,
createAgent,
setAgentMode,
@@ -2208,6 +2250,7 @@ export function SessionProvider({
cancelAgentRun,
sendAgentMessage,
deleteAgent,
archiveAgent,
createAgent,
setAgentMode,
respondToPermission,
@@ -2228,6 +2271,7 @@ export function SessionProvider({
cancelAgentRun,
sendAgentMessage,
deleteAgent,
archiveAgent,
createAgent,
setAgentMode,
respondToPermission,
@@ -80,6 +80,7 @@ export function useAggregatedAgents(): AggregatedAgentsResult {
attentionReason: agent.attentionReason,
attentionTimestamp: agent.attentionTimestamp,
parentAgentId: agent.parentAgentId,
archivedAt: agent.archivedAt,
};
allAgents.push(nextAgent);
}
+2
View File
@@ -99,6 +99,7 @@ export interface Agent {
attentionReason?: "finished" | "error" | "permission" | null;
attentionTimestamp?: Date | null;
parentAgentId?: string | null;
archivedAt?: Date | null;
}
export type ExplorerEntryKind = "file" | "directory";
@@ -183,6 +184,7 @@ export interface SessionState {
images?: Array<{ uri: string; mimeType?: string }>
) => Promise<void>;
deleteAgent: (agentId: string) => void;
archiveAgent: (agentId: string) => void;
createAgent: (options: {
config: any;
initialPrompt: string;
@@ -13,4 +13,5 @@ export interface AgentDirectoryEntry {
attentionReason?: "finished" | "error" | "permission" | null;
attentionTimestamp?: Date | null;
parentAgentId?: string | null;
archivedAt?: Date | null;
}
+6
View File
@@ -237,6 +237,12 @@ export function groupAgents(
const now = Date.now();
for (const agent of agents) {
// Archived agents are always inactive (hidden from sidebar)
if (agent.archivedAt) {
inactiveAgents.push(agent);
continue;
}
const isRunningOrAttention =
agent.status === "running" || agent.requiresAttention;
const isRecentlyActive =
+1
View File
@@ -6,3 +6,4 @@ dist
.debug.conversations
conversations
.cache
.debug/
@@ -0,0 +1,66 @@
import "dotenv/config";
import { readFile, writeFile, mkdir } from "node:fs/promises";
import path from "node:path";
import { fileURLToPath } from "node:url";
import OpenAI from "openai";
function requireEnv(name: string): string {
const value = process.env[name];
if (!value) {
throw new Error(`${name} not set`);
}
return value;
}
async function main(): Promise<void> {
const __filename = fileURLToPath(import.meta.url);
const __dirname = path.dirname(__filename);
const repoRoot = path.resolve(__dirname, "..");
const fixturePath = path.resolve(
repoRoot,
"src",
"server",
"fixtures",
"dictation",
"dictation-debug-largest.wav"
);
const outPath = path.resolve(
repoRoot,
"src",
"server",
"fixtures",
"dictation",
"dictation-debug-largest.transcript.txt"
);
const apiKey = requireEnv("OPENAI_API_KEY");
const transcriptionModel =
process.env.OPENAI_REALTIME_TRANSCRIPTION_MODEL ?? "gpt-4o-transcribe";
const prompt =
process.env.OPENAI_REALTIME_DICTATION_TRANSCRIPTION_PROMPT ??
"Transcribe only what the speaker says. Do not add words. Preserve punctuation and casing. If the audio is silence or non-speech noise, return an empty transcript.";
const openai = new OpenAI({ apiKey });
// Use a temp file path for the SDK's file stream param.
// We already have a fixture file, so just stream it.
const response = await openai.audio.transcriptions.create({
file: await import("node:fs").then((fs) => fs.createReadStream(fixturePath)),
language: "en",
model: transcriptionModel,
prompt,
response_format: "json",
// Aim for determinism.
temperature: 0,
});
await mkdir(path.dirname(outPath), { recursive: true });
await writeFile(outPath, `${response.text.trim()}\n`, "utf8");
// eslint-disable-next-line no-console
console.log(`Wrote baseline transcript to ${outPath}`);
}
await main();
@@ -100,6 +100,7 @@ describe("DaemonClientV2", () => {
isDirty: null,
baseRef: null,
aheadBehind: null,
aheadOfOrigin: null,
hasRemote: false,
remoteUrl: null,
},
@@ -131,7 +132,7 @@ describe("DaemonClientV2", () => {
);
await expect(p3).resolves.toMatchObject({
agentId: "agent-1",
cwd: "/tmp/project",
requestId: request2.message.requestId,
isGit: false,
});
@@ -841,6 +841,31 @@ export class DaemonClientV2 {
await response;
}
async archiveAgent(agentId: string): Promise<{ archivedAt: string }> {
const requestId = this.createRequestId();
const message = SessionInboundMessageSchema.parse({
type: "archive_agent_request",
agentId,
requestId,
});
const response = this.waitFor(
(msg) => {
if (msg.type !== "agent_archived") {
return null;
}
if (msg.payload.requestId !== requestId) {
return null;
}
return msg.payload;
},
10000,
{ skipQueue: true }
);
await this.sendSessionMessageOrThrow(message);
const result = await response;
return { archivedAt: result.archivedAt };
}
listAgents(): AgentSnapshotPayload[] {
return Array.from(this.agentIndex.values());
}
@@ -55,6 +55,7 @@ const STORED_AGENT_SCHEMA = z.object({
attentionTimestamp: z.string().nullable().optional(),
parentAgentId: z.string().nullable().optional(),
internal: z.boolean().optional(),
archivedAt: z.string().nullable().optional(),
});
export type SerializableAgentConfig = Pick<
@@ -10,6 +10,11 @@ import {
type DaemonTestContext,
} from "./test-utils/index.js";
import { getFullAccessConfig, getAskModeConfig } from "./daemon-e2e/agent-configs.js";
import {
chunkPcm16,
parsePcm16MonoWav,
requireEnv,
} from "./test-utils/dictation-e2e.js";
const defaultEnvPath = path.resolve(process.cwd(), ".env");
const fallbackEnvPath = path.resolve(process.cwd(), "packages", "server", ".env");
@@ -636,13 +641,7 @@ describe("daemon client v2 E2E", () => {
test(
"streams dictation PCM and returns final transcript via OpenAI Realtime transcription",
async () => {
if (process.env.PASEO_E2E_DICTATION_REALTIME !== "1") {
// Requires OpenAI realtime transcription and is inherently network-flaky.
return;
}
if (!process.env.OPENAI_API_KEY) {
return;
}
requireEnv("OPENAI_API_KEY");
const fixturePath = path.resolve(
process.cwd(),
@@ -653,55 +652,8 @@ describe("daemon client v2 E2E", () => {
"recording.wav"
);
const wav = await import("node:fs/promises").then((fs) => fs.readFile(fixturePath));
const parsePcm16MonoWav = (buffer: Buffer): Buffer => {
if (buffer.toString("ascii", 0, 4) !== "RIFF" || buffer.toString("ascii", 8, 12) !== "WAVE") {
throw new Error("Invalid WAV header");
}
let offset = 12;
let fmt: { audioFormat: number; channels: number; sampleRate: number; bitsPerSample: number } | null = null;
let dataChunk: Buffer | null = null;
while (offset + 8 <= buffer.length) {
const id = buffer.toString("ascii", offset, offset + 4);
const size = buffer.readUInt32LE(offset + 4);
const payloadStart = offset + 8;
const payloadEnd = payloadStart + size;
if (payloadEnd > buffer.length) {
break;
}
if (id === "fmt ") {
const audioFormat = buffer.readUInt16LE(payloadStart);
const channels = buffer.readUInt16LE(payloadStart + 2);
const sampleRate = buffer.readUInt32LE(payloadStart + 4);
const bitsPerSample = buffer.readUInt16LE(payloadStart + 14);
fmt = { audioFormat, channels, sampleRate, bitsPerSample };
} else if (id === "data") {
dataChunk = buffer.subarray(payloadStart, payloadEnd);
}
offset = payloadEnd + (size % 2);
}
if (!fmt || !dataChunk) {
throw new Error("Missing WAV fmt/data chunks");
}
if (fmt.audioFormat !== 1) {
throw new Error(`Unsupported WAV encoding (audioFormat=${fmt.audioFormat})`);
}
if (fmt.channels !== 1 || fmt.sampleRate !== 16000 || fmt.bitsPerSample !== 16) {
throw new Error(
`Unexpected WAV format: channels=${fmt.channels} rate=${fmt.sampleRate} bits=${fmt.bitsPerSample}`
);
}
if (dataChunk.length % 2 !== 0) {
throw new Error("WAV PCM16 data length must be even");
}
return dataChunk;
};
const pcm16 = parsePcm16MonoWav(wav);
const { sampleRate, pcm16 } = parsePcm16MonoWav(wav);
expect(sampleRate).toBe(16000);
const dictationId = `dict-${Date.now()}`;
const format = "audio/pcm;rate=16000;bits=16";
@@ -727,12 +679,7 @@ describe("daemon client v2 E2E", () => {
test(
"does not commit an empty OpenAI realtime audio buffer on finish",
async () => {
if (process.env.PASEO_E2E_DICTATION_REALTIME !== "1") {
return;
}
if (!process.env.OPENAI_API_KEY) {
return;
}
requireEnv("OPENAI_API_KEY");
const fixturePath = path.resolve(
process.cwd(),
@@ -743,55 +690,8 @@ describe("daemon client v2 E2E", () => {
"recording.wav"
);
const wav = await import("node:fs/promises").then((fs) => fs.readFile(fixturePath));
const parsePcm16MonoWav = (buffer: Buffer): Buffer => {
if (buffer.toString("ascii", 0, 4) !== "RIFF" || buffer.toString("ascii", 8, 12) !== "WAVE") {
throw new Error("Invalid WAV header");
}
let offset = 12;
let fmt: { audioFormat: number; channels: number; sampleRate: number; bitsPerSample: number } | null = null;
let dataChunk: Buffer | null = null;
while (offset + 8 <= buffer.length) {
const id = buffer.toString("ascii", offset, offset + 4);
const size = buffer.readUInt32LE(offset + 4);
const payloadStart = offset + 8;
const payloadEnd = payloadStart + size;
if (payloadEnd > buffer.length) {
break;
}
if (id === "fmt ") {
const audioFormat = buffer.readUInt16LE(payloadStart);
const channels = buffer.readUInt16LE(payloadStart + 2);
const sampleRate = buffer.readUInt32LE(payloadStart + 4);
const bitsPerSample = buffer.readUInt16LE(payloadStart + 14);
fmt = { audioFormat, channels, sampleRate, bitsPerSample };
} else if (id === "data") {
dataChunk = buffer.subarray(payloadStart, payloadEnd);
}
offset = payloadEnd + (size % 2);
}
if (!fmt || !dataChunk) {
throw new Error("Missing WAV fmt/data chunks");
}
if (fmt.audioFormat !== 1) {
throw new Error(`Unsupported WAV encoding (audioFormat=${fmt.audioFormat})`);
}
if (fmt.channels !== 1 || fmt.sampleRate !== 16000 || fmt.bitsPerSample !== 16) {
throw new Error(
`Unexpected WAV format: channels=${fmt.channels} rate=${fmt.sampleRate} bits=${fmt.bitsPerSample}`
);
}
if (dataChunk.length % 2 !== 0) {
throw new Error("WAV PCM16 data length must be even");
}
return dataChunk;
};
const pcm16 = parsePcm16MonoWav(wav);
const { sampleRate, pcm16 } = parsePcm16MonoWav(wav);
expect(sampleRate).toBe(16000);
const dictationId = `dict-empty-commit-${Date.now()}`;
const format = "audio/pcm;rate=16000;bits=16";
@@ -820,6 +720,115 @@ describe("daemon client v2 E2E", () => {
180000
);
describe("dictation streaming vs baseline (real OpenAI, debug fixtures)", () => {
let wav: Buffer;
let chunks: Buffer[];
let expectedText: string;
let fixtureSampleRate: number;
let chunkBytes: number;
beforeAll(async () => {
requireEnv("OPENAI_API_KEY");
const fixturePath = path.resolve(
process.cwd(),
"src",
"server",
"fixtures",
"dictation",
"dictation-debug-largest.wav"
);
const transcriptPath = path.resolve(
process.cwd(),
"src",
"server",
"fixtures",
"dictation",
"dictation-debug-largest.transcript.txt"
);
wav = await import("node:fs/promises").then((fs) => fs.readFile(fixturePath));
expectedText = await import("node:fs/promises").then((fs) => fs.readFile(transcriptPath, "utf8"));
const { sampleRate, pcm16 } = parsePcm16MonoWav(wav);
fixtureSampleRate = sampleRate;
// Stream at the fixture's native rate to avoid lossy double-resampling.
// 100ms @ sampleRate mono PCM16 => sampleRate * 0.1 samples * 2 bytes.
chunkBytes = Math.round((fixtureSampleRate * 2 * 100) / 1000);
if (chunkBytes <= 0 || chunkBytes % 2 !== 0) {
throw new Error(`Invalid chunkBytes computed: ${chunkBytes} (rate=${fixtureSampleRate})`);
}
chunks = chunkPcm16(pcm16, chunkBytes);
}, 240000);
test(
"streaming transcript matches baseline transcript exactly",
async () => {
const dictationId = `dict-debug-${Date.now()}`;
const format = `audio/pcm;rate=${fixtureSampleRate};bits=16`;
await ctx.client.startDictationStream(dictationId, format);
for (let seq = 0; seq < chunks.length; seq += 1) {
ctx.client.sendDictationStreamChunk(dictationId, seq, chunks[seq]!.toString("base64"), format);
}
const result = await ctx.client.finishDictationStream(dictationId, chunks.length - 1);
expect(result.dictationId).toBe(dictationId);
expect(result.text.trim()).toBe(expectedText.trim());
},
240000
);
test(
"finish before sending all chunks still completes and matches baseline exactly",
async () => {
const dictationId = `dict-early-finish-${Date.now()}`;
const format = `audio/pcm;rate=${fixtureSampleRate};bits=16`;
await ctx.client.startDictationStream(dictationId, format);
const splitAt = Math.max(1, Math.floor(chunks.length * 0.35));
for (let seq = 0; seq < splitAt; seq += 1) {
ctx.client.sendDictationStreamChunk(dictationId, seq, chunks[seq]!.toString("base64"), format);
}
const finishPromise = ctx.client.finishDictationStream(dictationId, chunks.length - 1);
// Simulate network jitter / chunk delay after finish.
await new Promise((r) => setTimeout(r, 250));
for (let seq = splitAt; seq < chunks.length; seq += 1) {
ctx.client.sendDictationStreamChunk(dictationId, seq, chunks[seq]!.toString("base64"), format);
}
const result = await finishPromise;
expect(result.dictationId).toBe(dictationId);
expect(result.text.trim()).toBe(expectedText.trim());
},
240000
);
test(
"missing a chunk causes dictation to error (no finalize)",
async () => {
const dictationId = `dict-missing-chunk-${Date.now()}`;
const format = `audio/pcm;rate=${fixtureSampleRate};bits=16`;
await ctx.client.startDictationStream(dictationId, format);
const missingSeq = Math.min(3, Math.max(0, chunks.length - 1));
for (let seq = 0; seq < chunks.length; seq += 1) {
if (seq === missingSeq) continue;
ctx.client.sendDictationStreamChunk(dictationId, seq, chunks[seq]!.toString("base64"), format);
}
await expect(ctx.client.finishDictationStream(dictationId, chunks.length - 1)).rejects.toThrow(
/Timed out waiting for final transcription|Timeout waiting for event|Timeout waiting for message/
);
},
240000
);
});
test(
"fails fast if dictation finishes without sending required chunks",
async () => {
@@ -423,15 +423,15 @@ describe("daemon checkout ship loop", () => {
});
agentId = agent.id;
const status = await ctx.client.getCheckoutStatus(worktree.worktreePath);
const status = await ctx.client.getCheckoutStatus(cwd);
expect(status.isGit).toBe(false);
const diff = await ctx.client.getCheckoutDiff(worktree.worktreePath, {
const diff = await ctx.client.getCheckoutDiff(cwd, {
mode: "uncommitted",
});
expect(diff.error?.code).toBe("NOT_GIT_REPO");
const commit = await ctx.client.checkoutCommit(worktree.worktreePath, {
const commit = await ctx.client.checkoutCommit(cwd, {
message: "Should fail",
addAll: true,
});
@@ -0,0 +1,652 @@
import type pino from "pino";
import { v4 as uuidv4 } from "uuid";
import { maybePersistDictationDebugAudio } from "../agent/dictation-debug.js";
import { isPaseoDictationDebugEnabled } from "../agent/recordings-debug.js";
import { Pcm16MonoResampler } from "../agent/pcm16-resampler.js";
import { OpenAIRealtimeTranscriptionSession } from "../agent/openai-realtime-transcription.js";
const PCM_CHANNELS = 1;
const PCM_BITS_PER_SAMPLE = 16;
const DICTATION_PCM_OUTPUT_RATE = 24000;
const DICTATION_FINAL_TIMEOUT_MS = 30000;
const DICTATION_SILENCE_PEAK_THRESHOLD = Number.parseInt(
process.env.OPENAI_REALTIME_DICTATION_SILENCE_PEAK_THRESHOLD ?? "300",
10
);
const DICTATION_TURN_DETECTION = (
process.env.OPENAI_REALTIME_DICTATION_TURN_DETECTION ?? "semantic_vad"
).trim();
const DICTATION_SEMANTIC_VAD_EAGERNESS = (
process.env.OPENAI_REALTIME_DICTATION_SEMANTIC_VAD_EAGERNESS ?? "medium"
).trim();
const DICTATION_FLUSH_SILENCE_MS = Number.parseInt(
process.env.OPENAI_REALTIME_DICTATION_FLUSH_SILENCE_MS ?? "800",
10
);
type OpenAITurnDetection =
| null
| {
type: "server_vad";
create_response: false;
threshold?: number;
prefix_padding_ms?: number;
silence_duration_ms?: number;
}
| { type: "semantic_vad"; create_response: false; eagerness?: "low" | "medium" | "high" };
function pcm16lePeakAbs(pcm16le: Buffer): number {
if (pcm16le.length === 0) {
return 0;
}
if (pcm16le.length % 2 !== 0) {
throw new Error(`PCM16 chunk byteLength must be even, got ${pcm16le.length}`);
}
const samples = new Int16Array(
pcm16le.buffer,
pcm16le.byteOffset,
pcm16le.byteLength / 2
);
let peak = 0;
for (let i = 0; i < samples.length; i += 1) {
const v = samples[i]!;
const abs = v < 0 ? -v : v;
if (abs > peak) {
peak = abs;
if (peak >= 32767) {
break;
}
}
}
return peak;
}
function parseDictationTurnDetection(): OpenAITurnDetection {
if (
!DICTATION_TURN_DETECTION ||
DICTATION_TURN_DETECTION === "none" ||
DICTATION_TURN_DETECTION === "null"
) {
return null;
}
if (DICTATION_TURN_DETECTION === "server_vad") {
return { type: "server_vad", create_response: false };
}
const eagerness =
DICTATION_SEMANTIC_VAD_EAGERNESS === "low" ||
DICTATION_SEMANTIC_VAD_EAGERNESS === "high"
? (DICTATION_SEMANTIC_VAD_EAGERNESS as "low" | "high")
: ("medium" as const);
return { type: "semantic_vad", create_response: false, eagerness };
}
function convertPCMToWavBuffer(
pcmBuffer: Buffer,
sampleRate: number,
channels: number,
bitsPerSample: number
): Buffer {
const headerSize = 44;
const wavBuffer = Buffer.alloc(headerSize + pcmBuffer.length);
const byteRate = (sampleRate * channels * bitsPerSample) / 8;
const blockAlign = (channels * bitsPerSample) / 8;
wavBuffer.write("RIFF", 0);
wavBuffer.writeUInt32LE(36 + pcmBuffer.length, 4);
wavBuffer.write("WAVE", 8);
wavBuffer.write("fmt ", 12);
wavBuffer.writeUInt32LE(16, 16);
wavBuffer.writeUInt16LE(1, 20);
wavBuffer.writeUInt16LE(channels, 22);
wavBuffer.writeUInt32LE(sampleRate, 24);
wavBuffer.writeUInt32LE(byteRate, 28);
wavBuffer.writeUInt16LE(blockAlign, 32);
wavBuffer.writeUInt16LE(bitsPerSample, 34);
wavBuffer.write("data", 36);
wavBuffer.writeUInt32LE(pcmBuffer.length, 40);
pcmBuffer.copy(wavBuffer, 44);
return wavBuffer;
}
type DictationStreamState = {
dictationId: string;
sessionId: string;
inputFormat: string;
openai: OpenAIRealtimeTranscriptionSession;
inputRate: number;
resampler: Pcm16MonoResampler | null;
debugAudioChunks: Buffer[];
debugRecordingPath: string | null;
receivedChunks: Map<number, Buffer>;
nextSeqToForward: number;
ackSeq: number;
bytesSinceCommit: number;
peakSinceCommit: number;
committedItemIds: string[];
transcriptsByItemId: Map<string, string>;
finalTranscriptItemIds: Set<string>;
awaitingFinalCommit: boolean;
finishRequested: boolean;
finishSealed: boolean;
finalSeq: number | null;
finalTimeout: ReturnType<typeof setTimeout> | null;
};
export type DictationStreamOutboundMessage =
| { type: "dictation_stream_ack"; payload: { dictationId: string; ackSeq: number } }
| { type: "dictation_stream_partial"; payload: { dictationId: string; text: string } }
| { type: "dictation_stream_final"; payload: { dictationId: string; text: string; debugRecordingPath?: string } }
| { type: "dictation_stream_error"; payload: { dictationId: string; error: string; retryable: boolean; debugRecordingPath?: string } }
| {
type: "activity_log";
payload: {
id: string;
timestamp: Date;
type: "system";
content: string;
metadata: Record<string, unknown>;
};
};
export class DictationStreamManager {
private readonly logger: pino.Logger;
private readonly emit: (msg: DictationStreamOutboundMessage) => void;
private readonly sessionId: string;
private readonly streams = new Map<string, DictationStreamState>();
constructor(params: {
logger: pino.Logger;
emit: (msg: DictationStreamOutboundMessage) => void;
sessionId: string;
}) {
this.logger = params.logger.child({ component: "dictation-stream-manager" });
this.emit = params.emit;
this.sessionId = params.sessionId;
}
public cleanupAll(): void {
for (const dictationId of this.streams.keys()) {
this.cleanupDictationStream(dictationId);
}
}
public async handleStart(dictationId: string, format: string): Promise<void> {
this.cleanupDictationStream(dictationId);
const apiKey = process.env.OPENAI_API_KEY;
if (!apiKey) {
this.failDictationStream(dictationId, "OPENAI_API_KEY not set", false);
return;
}
const transcriptionModel =
process.env.OPENAI_REALTIME_TRANSCRIPTION_MODEL ?? "gpt-4o-transcribe";
const transcriptionPrompt =
process.env.OPENAI_REALTIME_DICTATION_TRANSCRIPTION_PROMPT ??
"Transcribe only what the speaker says. Do not add words. Preserve punctuation and casing. If the audio is silence or non-speech noise, return an empty transcript.";
const openai = new OpenAIRealtimeTranscriptionSession({
apiKey,
logger: this.logger.child({ dictationId }),
transcriptionModel,
language: "en",
prompt: transcriptionPrompt,
turnDetection: parseDictationTurnDetection(),
});
openai.on("committed", ({ itemId }: { itemId: string }) => {
const state = this.streams.get(dictationId);
if (!state) {
return;
}
state.committedItemIds.push(itemId);
state.bytesSinceCommit = 0;
state.peakSinceCommit = 0;
// When finishing, we require at least one commit after finish if we flushed pending audio.
if (state.finishRequested && state.awaitingFinalCommit) {
state.awaitingFinalCommit = false;
}
this.maybeFinalizeDictationStream(dictationId);
});
openai.on(
"transcript",
({
itemId,
transcript,
isFinal,
}: {
itemId: string;
transcript: string;
isFinal: boolean;
}) => {
const state = this.streams.get(dictationId);
if (!state) {
return;
}
state.transcriptsByItemId.set(itemId, transcript);
if (isFinal) {
state.finalTranscriptItemIds.add(itemId);
}
// If we triggered a finish commit but OpenAI doesn't emit committed events (or they arrive late),
// allow final transcripts to unblock finalization.
if (state.finishRequested && state.awaitingFinalCommit && isFinal) {
state.awaitingFinalCommit = false;
}
const orderedIds = state.committedItemIds.includes(itemId)
? state.committedItemIds
: [...state.committedItemIds, itemId];
const partialText = orderedIds
.map((id) => state.transcriptsByItemId.get(id) ?? "")
.join(" ")
.trim();
this.emitDictationPartial(dictationId, partialText);
this.maybeSealDictationStreamFinish(dictationId);
this.maybeFinalizeDictationStream(dictationId);
}
);
openai.on("error", (err) => {
const message = err instanceof Error ? err.message : String(err);
void this.failAndCleanupDictationStream(dictationId, message, true);
});
await openai.connect();
const rateMatch = /(?:^|[;,\s])rate\s*=\s*(\d+)(?:$|[;,\s])/i.exec(format);
const inputRate = rateMatch ? Number.parseInt(rateMatch[1]!, 10) : 16000;
if (!Number.isFinite(inputRate) || inputRate <= 0) {
this.failDictationStream(dictationId, `Invalid dictation input rate in format: ${format}`, false);
try {
openai.close();
} catch {
// no-op
}
return;
}
this.streams.set(dictationId, {
dictationId,
sessionId: this.sessionId,
inputFormat: format,
openai,
inputRate,
resampler:
inputRate === DICTATION_PCM_OUTPUT_RATE
? null
: new Pcm16MonoResampler({
inputRate,
outputRate: DICTATION_PCM_OUTPUT_RATE,
}),
debugAudioChunks: [],
debugRecordingPath: null,
receivedChunks: new Map(),
nextSeqToForward: 0,
ackSeq: -1,
bytesSinceCommit: 0,
peakSinceCommit: 0,
committedItemIds: [],
transcriptsByItemId: new Map(),
finalTranscriptItemIds: new Set(),
awaitingFinalCommit: false,
finishRequested: false,
finishSealed: false,
finalSeq: null,
finalTimeout: null,
});
this.emitDictationAck(dictationId, -1);
}
public async handleChunk(params: {
dictationId: string;
seq: number;
audioBase64: string;
format: string;
}): Promise<void> {
const state = this.streams.get(params.dictationId);
if (!state) {
this.failDictationStream(params.dictationId, "Dictation stream not started", true);
return;
}
if (params.format !== state.inputFormat) {
void this.failAndCleanupDictationStream(
params.dictationId,
`Mismatched dictation stream format: ${params.format}`,
false
);
return;
}
if (params.seq < state.nextSeqToForward) {
this.emitDictationAck(params.dictationId, state.ackSeq);
return;
}
if (!state.receivedChunks.has(params.seq)) {
state.receivedChunks.set(params.seq, Buffer.from(params.audioBase64, "base64"));
}
while (state.receivedChunks.has(state.nextSeqToForward)) {
const seq = state.nextSeqToForward;
const pcm16 = state.receivedChunks.get(seq)!;
state.receivedChunks.delete(seq);
const resampled = state.resampler ? state.resampler.processChunk(pcm16) : pcm16;
if (resampled.length > 0) {
state.openai.appendPcm16Base64(resampled.toString("base64"));
state.debugAudioChunks.push(resampled);
state.bytesSinceCommit += resampled.length;
state.peakSinceCommit = Math.max(state.peakSinceCommit, pcm16lePeakAbs(resampled));
}
state.nextSeqToForward += 1;
state.ackSeq = state.nextSeqToForward - 1;
}
this.emitDictationAck(params.dictationId, state.ackSeq);
this.maybeSealDictationStreamFinish(params.dictationId);
this.maybeFinalizeDictationStream(params.dictationId);
}
public async handleFinish(dictationId: string, finalSeq: number): Promise<void> {
const state = this.streams.get(dictationId);
if (!state) {
this.failDictationStream(dictationId, "Dictation stream not started", true);
return;
}
state.finishRequested = true;
state.finalSeq = finalSeq;
if (finalSeq >= 0 && state.ackSeq < 0 && state.nextSeqToForward === 0 && state.receivedChunks.size === 0) {
this.logger.debug(
{
dictationId,
finalSeq,
ackSeq: state.ackSeq,
nextSeqToForward: state.nextSeqToForward,
receivedChunks: state.receivedChunks.size,
bytesSinceCommit: state.bytesSinceCommit,
},
"Dictation finish: no chunks received (failing fast)"
);
this.failDictationStream(
dictationId,
`Dictation finished (finalSeq=${finalSeq}) but no audio chunks were received`,
true
);
this.cleanupDictationStream(dictationId);
return;
}
if (state.finalTimeout) {
clearTimeout(state.finalTimeout);
}
state.finalTimeout = setTimeout(() => {
void this.failAndCleanupDictationStream(
dictationId,
"Timed out waiting for final transcription",
true
);
}, DICTATION_FINAL_TIMEOUT_MS);
this.maybeSealDictationStreamFinish(dictationId);
this.maybeFinalizeDictationStream(dictationId);
}
public handleCancel(dictationId: string): void {
this.cleanupDictationStream(dictationId);
}
private emitDictationAck(dictationId: string, ackSeq: number): void {
this.emit({ type: "dictation_stream_ack", payload: { dictationId, ackSeq } });
}
private emitDictationPartial(dictationId: string, text: string): void {
this.emit({ type: "dictation_stream_partial", payload: { dictationId, text } });
}
private async maybePersistDictationStreamAudio(dictationId: string): Promise<string | null> {
if (!isPaseoDictationDebugEnabled()) {
return null;
}
const state = this.streams.get(dictationId);
if (!state) {
return null;
}
if (state.debugRecordingPath) {
return state.debugRecordingPath;
}
if (state.debugAudioChunks.length === 0) {
return null;
}
const pcmBuffer = Buffer.concat(state.debugAudioChunks);
const wavBuffer = convertPCMToWavBuffer(
pcmBuffer,
DICTATION_PCM_OUTPUT_RATE,
PCM_CHANNELS,
PCM_BITS_PER_SAMPLE
);
const path = await maybePersistDictationDebugAudio(
wavBuffer,
{ sessionId: state.sessionId, dictationId: state.dictationId, format: "audio/wav" },
this.logger
);
state.debugRecordingPath = path;
return path;
}
private failDictationStream(dictationId: string, error: string, retryable: boolean): void {
this.emit({
type: "dictation_stream_error",
payload: { dictationId, error, retryable },
});
}
private async failAndCleanupDictationStream(
dictationId: string,
error: string,
retryable: boolean
): Promise<void> {
const debugRecordingPath = await this.maybePersistDictationStreamAudio(dictationId);
this.emit({
type: "dictation_stream_error",
payload: {
dictationId,
error,
retryable,
...(debugRecordingPath ? { debugRecordingPath } : {}),
},
});
if (debugRecordingPath) {
this.emit({
type: "activity_log",
payload: {
id: uuidv4(),
timestamp: new Date(),
type: "system",
content: `Saved dictation audio: ${debugRecordingPath}`,
metadata: { recordingPath: debugRecordingPath, dictationId },
},
});
}
this.cleanupDictationStream(dictationId);
}
private cleanupDictationStream(dictationId: string): void {
const state = this.streams.get(dictationId) ?? null;
if (!state) {
return;
}
if (state.finalTimeout) {
clearTimeout(state.finalTimeout);
}
try {
state.openai.close();
} catch {
// no-op
}
this.streams.delete(dictationId);
}
private maybeSealDictationStreamFinish(dictationId: string): void {
const state = this.streams.get(dictationId);
if (!state) {
return;
}
if (!state.finishRequested || state.finalSeq === null) {
return;
}
if (state.ackSeq < state.finalSeq) {
return;
}
if (state.finishSealed) {
return;
}
if (state.bytesSinceCommit > 0) {
if (state.peakSinceCommit < DICTATION_SILENCE_PEAK_THRESHOLD) {
this.logger.debug(
{
dictationId,
bytesSinceCommit: state.bytesSinceCommit,
peakSinceCommit: state.peakSinceCommit,
},
"Dictation finish: clearing silence-only tail (skip final commit)"
);
state.openai.clear();
state.bytesSinceCommit = 0;
state.peakSinceCommit = 0;
state.awaitingFinalCommit = false;
} else {
const silenceBytes = Math.max(
0,
Math.round((DICTATION_PCM_OUTPUT_RATE * 2 * DICTATION_FLUSH_SILENCE_MS) / 1000)
);
if (silenceBytes > 0) {
this.logger.debug(
{ dictationId, silenceMs: DICTATION_FLUSH_SILENCE_MS, silenceBytes },
"Dictation finish: appending silence tail for semantic VAD flush"
);
const silence = Buffer.alloc(silenceBytes);
state.openai.appendPcm16Base64(silence.toString("base64"));
state.debugAudioChunks.push(silence);
state.bytesSinceCommit += silenceBytes;
}
try {
state.awaitingFinalCommit = true;
state.openai.commit();
} catch (error) {
const message = error instanceof Error ? error.message : String(error);
void this.failAndCleanupDictationStream(dictationId, message, true);
return;
}
}
} else {
state.awaitingFinalCommit = false;
}
state.finishSealed = true;
}
private maybeFinalizeDictationStream(dictationId: string): void {
const state = this.streams.get(dictationId);
if (!state) {
return;
}
if (!state.finishRequested || state.finalSeq === null) {
return;
}
if (state.ackSeq < state.finalSeq) {
return;
}
if (state.awaitingFinalCommit) {
return;
}
const committedSet = new Set(state.committedItemIds);
const orderedItemIds: string[] = [...state.committedItemIds];
for (const itemId of state.transcriptsByItemId.keys()) {
if (!committedSet.has(itemId)) {
orderedItemIds.push(itemId);
}
}
if (orderedItemIds.length === 0) {
void (async () => {
const debugRecordingPath = await this.maybePersistDictationStreamAudio(dictationId);
this.emit({
type: "dictation_stream_final",
payload: {
dictationId,
text: "",
...(debugRecordingPath ? { debugRecordingPath } : {}),
},
});
if (debugRecordingPath) {
this.emit({
type: "activity_log",
payload: {
id: uuidv4(),
timestamp: new Date(),
type: "system",
content: `Saved dictation audio: ${debugRecordingPath}`,
metadata: { recordingPath: debugRecordingPath, dictationId },
},
});
}
this.cleanupDictationStream(dictationId);
})();
return;
}
const allTranscriptsReady = orderedItemIds.every((itemId) =>
state.finalTranscriptItemIds.has(itemId)
);
if (!allTranscriptsReady) {
return;
}
const orderedText = orderedItemIds
.map((itemId) => state.transcriptsByItemId.get(itemId) ?? "")
.join(" ")
.trim();
void (async () => {
const debugRecordingPath = await this.maybePersistDictationStreamAudio(dictationId);
this.emit({
type: "dictation_stream_final",
payload: {
dictationId,
text: orderedText,
...(debugRecordingPath ? { debugRecordingPath } : {}),
},
});
if (debugRecordingPath) {
this.emit({
type: "activity_log",
payload: {
id: uuidv4(),
timestamp: new Date(),
type: "system",
content: `Saved dictation audio: ${debugRecordingPath}`,
metadata: { recordingPath: debugRecordingPath, dictationId },
},
});
}
this.cleanupDictationStream(dictationId);
})();
}
}
+67 -509
View File
@@ -35,11 +35,8 @@ import { TTSManager } from "./agent/tts-manager.js";
import { STTManager } from "./agent/stt-manager.js";
import type { OpenAISTT } from "./agent/stt-openai.js";
import type { OpenAITTS } from "./agent/tts-openai.js";
import { OpenAIRealtimeTranscriptionSession } from "./agent/openai-realtime-transcription.js";
import { Pcm16MonoResampler } from "./agent/pcm16-resampler.js";
import { maybePersistDictationDebugAudio } from "./agent/dictation-debug.js";
import { maybePersistTtsDebugAudio } from "./agent/tts-debug.js";
import { isPaseoDictationDebugEnabled } from "./agent/recordings-debug.js";
import { DictationStreamManager } from "./dictation/dictation-stream-manager.js";
import type { VoiceConversationStore } from "./voice-conversation-store.js";
import {
buildConfigOverrides,
@@ -153,66 +150,8 @@ const MIN_STREAMING_SEGMENT_DURATION_MS = 1000;
const MIN_STREAMING_SEGMENT_BYTES = Math.round(
PCM_BYTES_PER_MS * MIN_STREAMING_SEGMENT_DURATION_MS
);
const DICTATION_PCM_INPUT_RATE = 16000;
const DICTATION_PCM_OUTPUT_RATE = 24000;
const DICTATION_FINAL_TIMEOUT_MS = 30000;
const DICTATION_SILENCE_PEAK_THRESHOLD = Number.parseInt(
process.env.OPENAI_REALTIME_DICTATION_SILENCE_PEAK_THRESHOLD ?? "300",
10
);
const DICTATION_TURN_DETECTION = (process.env.OPENAI_REALTIME_DICTATION_TURN_DETECTION ?? "semantic_vad").trim();
const DICTATION_SEMANTIC_VAD_EAGERNESS = (process.env.OPENAI_REALTIME_DICTATION_SEMANTIC_VAD_EAGERNESS ?? "medium").trim();
const DICTATION_FLUSH_SILENCE_MS = Number.parseInt(
process.env.OPENAI_REALTIME_DICTATION_FLUSH_SILENCE_MS ?? "800",
10
);
const SAFE_GIT_REF_PATTERN = /^[A-Za-z0-9._\/-]+$/;
function pcm16lePeakAbs(pcm16le: Buffer): number {
if (pcm16le.length === 0) {
return 0;
}
if (pcm16le.length % 2 !== 0) {
throw new Error(`PCM16 chunk byteLength must be even, got ${pcm16le.length}`);
}
const samples = new Int16Array(
pcm16le.buffer,
pcm16le.byteOffset,
pcm16le.byteLength / 2
);
let peak = 0;
for (let i = 0; i < samples.length; i += 1) {
const v = samples[i]!;
const abs = v < 0 ? -v : v;
if (abs > peak) {
peak = abs;
if (peak >= 32767) {
break;
}
}
}
return peak;
}
function parseDictationTurnDetection():
| null
| { type: "server_vad"; create_response: false; threshold?: number; prefix_padding_ms?: number; silence_duration_ms?: number }
| { type: "semantic_vad"; create_response: false; eagerness?: "low" | "medium" | "high" } {
if (!DICTATION_TURN_DETECTION || DICTATION_TURN_DETECTION === "none" || DICTATION_TURN_DETECTION === "null") {
return null;
}
if (DICTATION_TURN_DETECTION === "server_vad") {
return { type: "server_vad", create_response: false };
}
const eagerness =
DICTATION_SEMANTIC_VAD_EAGERNESS === "low" ||
DICTATION_SEMANTIC_VAD_EAGERNESS === "high"
? (DICTATION_SEMANTIC_VAD_EAGERNESS as "low" | "high")
: ("medium" as const);
return { type: "semantic_vad", create_response: false, eagerness };
}
/**
* Type for present_artifact tool arguments
*/
@@ -326,29 +265,7 @@ export class Session {
private speechInProgress = false;
private voiceConversationId: string | null = null;
private dictationStreams = new Map<
string,
{
dictationId: string;
inputFormat: string;
openai: OpenAIRealtimeTranscriptionSession;
resampler: Pcm16MonoResampler;
debugAudioChunks: Buffer[];
debugRecordingPath: string | null;
receivedChunks: Map<number, Buffer>;
nextSeqToForward: number;
ackSeq: number;
bytesSinceCommit: number;
peakSinceCommit: number;
committedItemIds: string[];
transcriptsByItemId: Map<string, string>;
finalTranscriptItemIds: Set<string>;
awaitingFinalCommit: boolean;
finishRequested: boolean;
finalSeq: number | null;
finalTimeout: ReturnType<typeof setTimeout> | null;
}
>();
private readonly dictationStreamManager: DictationStreamManager;
// Audio buffering for interruption handling
private pendingAudioSegments: Array<{ audio: Buffer; format: string }> = [];
@@ -427,6 +344,11 @@ export class Session {
// Initialize per-session managers
this.ttsManager = new TTSManager(this.sessionId, this.sessionLogger, tts);
this.sttManager = new STTManager(this.sessionId, this.sessionLogger, stt);
this.dictationStreamManager = new DictationStreamManager({
logger: this.sessionLogger,
sessionId: this.sessionId,
emit: (msg) => this.emit(msg as unknown as SessionOutboundMessage),
});
// Initialize agent MCP client asynchronously
void this.initializeAgentMcp();
@@ -665,8 +587,11 @@ export class Session {
private async buildAgentPayload(
agent: ManagedAgent
): Promise<AgentSnapshotPayload> {
const title = await this.getStoredAgentTitle(agent.id);
return toAgentPayload(agent, { title });
const storedRecord = await this.agentStorage.get(agent.id);
const title = storedRecord?.title ?? null;
const payload = toAgentPayload(agent, { title });
payload.archivedAt = storedRecord?.archivedAt ?? null;
return payload;
}
private buildStoredAgentPayload(record: StoredAgentRecord): AgentSnapshotPayload {
@@ -703,6 +628,11 @@ export class Session {
lastUsage: undefined,
lastError: undefined,
title: record.title ?? null,
requiresAttention: record.requiresAttention ?? false,
attentionReason: record.attentionReason ?? null,
attentionTimestamp: record.attentionTimestamp ?? null,
parentAgentId: record.parentAgentId ?? null,
archivedAt: record.archivedAt ?? null,
};
}
@@ -768,20 +698,6 @@ export class Session {
}
}
private async getStoredAgentTitle(agentId: string): Promise<string | null> {
try {
const record = await this.agentStorage.get(agentId);
const title = record?.title ?? null;
return title;
} catch (error) {
this.sessionLogger.error(
{ err: error, agentId },
`Failed to load registry record for agent ${agentId}`
);
return null;
}
}
/**
* Main entry point for processing session messages
*/
@@ -827,6 +743,10 @@ export class Session {
await this.handleDeleteAgentRequest(msg.agentId, msg.requestId);
break;
case "archive_agent_request":
await this.handleArchiveAgentRequest(msg.agentId, msg.requestId);
break;
case "set_voice_conversation":
await this.handleSetVoiceConversation(msg.enabled, msg.voiceConversationId);
break;
@@ -841,19 +761,24 @@ export class Session {
break;
case "dictation_stream_start":
await this.handleDictationStreamStart(msg);
await this.dictationStreamManager.handleStart(msg.dictationId, msg.format);
break;
case "dictation_stream_chunk":
await this.handleDictationStreamChunk(msg);
await this.dictationStreamManager.handleChunk({
dictationId: msg.dictationId,
seq: msg.seq,
audioBase64: msg.audio,
format: msg.format,
});
break;
case "dictation_stream_finish":
await this.handleDictationStreamFinish(msg);
await this.dictationStreamManager.handleFinish(msg.dictationId, msg.finalSeq);
break;
case "dictation_stream_cancel":
await this.handleDictationStreamCancel(msg);
this.dictationStreamManager.handleCancel(msg.dictationId);
break;
case "create_agent_request":
@@ -1203,6 +1128,42 @@ export class Session {
});
}
private async handleArchiveAgentRequest(
agentId: string,
requestId: string
): Promise<void> {
this.sessionLogger.info(
{ agentId },
`Archiving agent ${agentId}`
);
const archivedAt = new Date().toISOString();
try {
const existing = await this.agentStorage.get(agentId);
if (existing) {
await this.agentStorage.upsert({
...existing,
archivedAt,
});
}
} catch (error: any) {
this.sessionLogger.error(
{ err: error, agentId },
`Failed to archive agent ${agentId}`
);
}
this.emit({
type: "agent_archived",
payload: {
agentId,
archivedAt,
requestId,
},
});
}
/**
* Handle realtime mode toggle
*/
@@ -1309,404 +1270,6 @@ export class Session {
this.startAgentStream(agentId, prompt);
}
private emitDictationAck(dictationId: string, ackSeq: number): void {
this.emit({
type: "dictation_stream_ack",
payload: { dictationId, ackSeq },
});
}
private failDictationStream(dictationId: string, error: string, retryable: boolean): void {
this.emit({
type: "dictation_stream_error",
payload: { dictationId, error, retryable },
});
}
private async maybePersistDictationStreamAudio(dictationId: string): Promise<string | null> {
const state = this.dictationStreams.get(dictationId) ?? null;
if (!state) {
return null;
}
if (state.debugRecordingPath) {
return state.debugRecordingPath;
}
if (state.debugAudioChunks.length === 0) {
return null;
}
const pcmBuffer = Buffer.concat(state.debugAudioChunks);
const wavBuffer = convertPCMToWavBuffer(
pcmBuffer,
DICTATION_PCM_OUTPUT_RATE,
PCM_CHANNELS,
PCM_BITS_PER_SAMPLE
);
const path = await maybePersistDictationDebugAudio(
wavBuffer,
{ sessionId: this.sessionId, dictationId, format: "audio/wav" },
this.sessionLogger
);
state.debugRecordingPath = path;
return path;
}
private async failAndCleanupDictationStream(
dictationId: string,
error: string,
retryable: boolean
): Promise<void> {
const debugRecordingPath = await this.maybePersistDictationStreamAudio(dictationId);
this.emit({
type: "dictation_stream_error",
payload: {
dictationId,
error,
retryable,
...(debugRecordingPath ? { debugRecordingPath } : {}),
},
});
if (debugRecordingPath) {
this.emit({
type: "activity_log",
payload: {
id: uuidv4(),
timestamp: new Date(),
type: "system",
content: `Saved dictation audio: ${debugRecordingPath}`,
metadata: { recordingPath: debugRecordingPath, dictationId },
},
});
}
this.cleanupDictationStream(dictationId);
}
private cleanupDictationStream(dictationId: string): void {
const state = this.dictationStreams.get(dictationId) ?? null;
if (!state) {
return;
}
if (state.finalTimeout) {
clearTimeout(state.finalTimeout);
}
try {
state.openai.close();
} catch {
// no-op
}
this.dictationStreams.delete(dictationId);
}
private emitDictationPartial(dictationId: string, text: string): void {
this.emit({
type: "dictation_stream_partial",
payload: { dictationId, text },
});
}
private maybeFinalizeDictationStream(dictationId: string): void {
const state = this.dictationStreams.get(dictationId);
if (!state) {
return;
}
if (!state.finishRequested || state.finalSeq === null) {
return;
}
if (state.ackSeq < state.finalSeq) {
return;
}
if (state.awaitingFinalCommit) {
return;
}
if (state.committedItemIds.length === 0) {
return;
}
const allTranscriptsReady = state.committedItemIds.every((itemId) =>
state.finalTranscriptItemIds.has(itemId)
);
if (!allTranscriptsReady) {
return;
}
const orderedText = state.committedItemIds
.map((itemId) => state.transcriptsByItemId.get(itemId) ?? "")
.join(" ")
.trim();
void (async () => {
const debugRecordingPath = await this.maybePersistDictationStreamAudio(dictationId);
this.emit({
type: "dictation_stream_final",
payload: {
dictationId,
text: orderedText,
...(debugRecordingPath ? { debugRecordingPath } : {}),
},
});
if (debugRecordingPath) {
this.emit({
type: "activity_log",
payload: {
id: uuidv4(),
timestamp: new Date(),
type: "system",
content: `Saved dictation audio: ${debugRecordingPath}`,
metadata: { recordingPath: debugRecordingPath, dictationId },
},
});
}
this.cleanupDictationStream(dictationId);
})();
}
private async handleDictationStreamStart(
msg: Extract<SessionInboundMessage, { type: "dictation_stream_start" }>
): Promise<void> {
const dictationId = msg.dictationId;
this.cleanupDictationStream(dictationId);
const apiKey = process.env.OPENAI_API_KEY;
if (!apiKey) {
this.failDictationStream(dictationId, "OPENAI_API_KEY not set", false);
return;
}
const transcriptionModel =
process.env.OPENAI_REALTIME_TRANSCRIPTION_MODEL ?? "gpt-4o-transcribe";
const transcriptionPrompt =
process.env.OPENAI_REALTIME_DICTATION_TRANSCRIPTION_PROMPT ??
"Transcribe only what the speaker says. Do not add words. Preserve punctuation and casing. If the audio is silence or non-speech noise, return an empty transcript.";
const openai = new OpenAIRealtimeTranscriptionSession({
apiKey,
logger: this.sessionLogger.child({ dictationId }),
transcriptionModel,
language: "en",
prompt: transcriptionPrompt,
turnDetection: parseDictationTurnDetection(),
});
openai.on("committed", ({ itemId }: { itemId: string }) => {
const state = this.dictationStreams.get(dictationId);
if (!state) {
return;
}
state.committedItemIds.push(itemId);
state.bytesSinceCommit = 0;
state.peakSinceCommit = 0;
// When finishing, we require at least one commit after finish if we flushed pending audio.
if (state.finishRequested && state.awaitingFinalCommit) {
state.awaitingFinalCommit = false;
}
this.maybeFinalizeDictationStream(dictationId);
});
openai.on(
"transcript",
({ itemId, transcript, isFinal }: { itemId: string; transcript: string; isFinal: boolean }) => {
const state = this.dictationStreams.get(dictationId);
if (!state) {
return;
}
state.transcriptsByItemId.set(itemId, transcript);
if (isFinal) {
state.finalTranscriptItemIds.add(itemId);
}
const orderedIds = state.committedItemIds.includes(itemId)
? state.committedItemIds
: [...state.committedItemIds, itemId];
const partialText = orderedIds
.map((id) => state.transcriptsByItemId.get(id) ?? "")
.join(" ")
.trim();
this.emitDictationPartial(dictationId, partialText);
this.maybeFinalizeDictationStream(dictationId);
}
);
openai.on("error", (err) => {
const message = err instanceof Error ? err.message : String(err);
void this.failAndCleanupDictationStream(dictationId, message, true);
});
await openai.connect();
this.dictationStreams.set(dictationId, {
dictationId,
inputFormat: msg.format,
openai,
resampler: new Pcm16MonoResampler({
inputRate: DICTATION_PCM_INPUT_RATE,
outputRate: DICTATION_PCM_OUTPUT_RATE,
}),
debugAudioChunks: [],
debugRecordingPath: null,
receivedChunks: new Map(),
nextSeqToForward: 0,
ackSeq: -1,
bytesSinceCommit: 0,
peakSinceCommit: 0,
committedItemIds: [],
transcriptsByItemId: new Map(),
finalTranscriptItemIds: new Set(),
awaitingFinalCommit: false,
finishRequested: false,
finalSeq: null,
finalTimeout: null,
});
this.emitDictationAck(dictationId, -1);
}
private async handleDictationStreamChunk(
msg: Extract<SessionInboundMessage, { type: "dictation_stream_chunk" }>
): Promise<void> {
const state = this.dictationStreams.get(msg.dictationId);
if (!state) {
this.failDictationStream(msg.dictationId, "Dictation stream not started", true);
return;
}
if (msg.format !== state.inputFormat) {
void this.failAndCleanupDictationStream(
msg.dictationId,
`Mismatched dictation stream format: ${msg.format}`,
false
);
return;
}
if (msg.seq < state.nextSeqToForward) {
this.emitDictationAck(msg.dictationId, state.ackSeq);
return;
}
if (!state.receivedChunks.has(msg.seq)) {
state.receivedChunks.set(msg.seq, Buffer.from(msg.audio, "base64"));
}
while (state.receivedChunks.has(state.nextSeqToForward)) {
const seq = state.nextSeqToForward;
const pcm16 = state.receivedChunks.get(seq)!;
state.receivedChunks.delete(seq);
const resampled = state.resampler.processChunk(pcm16);
if (resampled.length > 0) {
state.openai.appendPcm16Base64(resampled.toString("base64"));
state.debugAudioChunks.push(resampled);
state.bytesSinceCommit += resampled.length;
state.peakSinceCommit = Math.max(state.peakSinceCommit, pcm16lePeakAbs(resampled));
}
state.nextSeqToForward += 1;
state.ackSeq = state.nextSeqToForward - 1;
}
this.emitDictationAck(msg.dictationId, state.ackSeq);
}
private async handleDictationStreamFinish(
msg: Extract<SessionInboundMessage, { type: "dictation_stream_finish" }>
): Promise<void> {
const state = this.dictationStreams.get(msg.dictationId);
if (!state) {
this.failDictationStream(msg.dictationId, "Dictation stream not started", true);
return;
}
state.finishRequested = true;
state.finalSeq = msg.finalSeq;
// If the client claims it sent audio (finalSeq >= 0) but we haven't received any chunks at all,
// we should fail fast instead of hanging until DICTATION_FINAL_TIMEOUT_MS.
if (msg.finalSeq >= 0 && state.ackSeq < 0 && state.nextSeqToForward === 0 && state.receivedChunks.size === 0) {
this.sessionLogger.debug(
{
dictationId: msg.dictationId,
finalSeq: msg.finalSeq,
ackSeq: state.ackSeq,
nextSeqToForward: state.nextSeqToForward,
receivedChunks: state.receivedChunks.size,
bytesSinceCommit: state.bytesSinceCommit,
},
"Dictation finish: no chunks received (failing fast)"
);
this.failDictationStream(
msg.dictationId,
`Dictation finished (finalSeq=${msg.finalSeq}) but no audio chunks were received`,
true
);
this.cleanupDictationStream(msg.dictationId);
return;
}
// Force-flush any remaining audio so we get a final transcription even if VAD hasn't committed yet.
if (state.bytesSinceCommit > 0) {
if (state.peakSinceCommit < DICTATION_SILENCE_PEAK_THRESHOLD) {
this.sessionLogger.debug(
{
dictationId: msg.dictationId,
bytesSinceCommit: state.bytesSinceCommit,
peakSinceCommit: state.peakSinceCommit,
},
"Dictation finish: clearing silence-only tail (skip final commit)"
);
state.openai.clear();
state.bytesSinceCommit = 0;
state.peakSinceCommit = 0;
state.awaitingFinalCommit = false;
} else {
// We do not manually commit for dictation (to avoid chopping words). Instead we rely on
// OpenAI semantic VAD to commit utterances. When the user presses OK, we append a short
// silence tail so VAD has enough silence to close out the final utterance promptly.
const silenceBytes = Math.max(
0,
Math.round((DICTATION_PCM_OUTPUT_RATE * 2 * DICTATION_FLUSH_SILENCE_MS) / 1000)
);
if (silenceBytes > 0) {
this.sessionLogger.debug(
{ dictationId: msg.dictationId, silenceMs: DICTATION_FLUSH_SILENCE_MS, silenceBytes },
"Dictation finish: appending silence tail for semantic VAD flush"
);
const silence = Buffer.alloc(silenceBytes);
state.openai.appendPcm16Base64(silence.toString("base64"));
state.debugAudioChunks.push(silence);
state.bytesSinceCommit += silenceBytes;
}
state.awaitingFinalCommit = true;
}
} else {
state.awaitingFinalCommit = false;
}
if (state.finalTimeout) {
clearTimeout(state.finalTimeout);
}
state.finalTimeout = setTimeout(() => {
void this.failAndCleanupDictationStream(
msg.dictationId,
"Timed out waiting for final transcription",
true
);
}, DICTATION_FINAL_TIMEOUT_MS);
this.maybeFinalizeDictationStream(msg.dictationId);
}
private async handleDictationStreamCancel(
msg: Extract<SessionInboundMessage, { type: "dictation_stream_cancel" }>
): Promise<void> {
this.cleanupDictationStream(msg.dictationId);
}
/**
* Handle on-demand agent initialization request from client
*/
@@ -2104,7 +1667,6 @@ export class Session {
callId,
status: "running",
input: {
repoRoot: worktree.repoPath,
worktreePath: worktree.worktreePath,
branchName: worktree.branchName,
},
@@ -2114,7 +1676,6 @@ export class Session {
}
results = await runWorktreeSetupCommands({
repoRoot: worktree.repoPath,
worktreePath: worktree.worktreePath,
branchName: worktree.branchName,
cleanupOnFailure: false,
@@ -4847,10 +4408,7 @@ export class Session {
// Cleanup managers
this.ttsManager.cleanup();
this.sttManager.cleanup();
for (const dictationId of this.dictationStreams.keys()) {
this.cleanupDictationStream(dictationId);
}
this.dictationStreamManager.cleanupAll();
// Close MCP clients
if (this.agentMcpClient) {
@@ -0,0 +1,164 @@
import { mkdtempSync, writeFileSync, rmSync, existsSync } from "node:fs";
import { tmpdir } from "node:os";
import path from "node:path";
import OpenAI from "openai";
export function requireEnv(name: string): string {
const value = process.env[name];
if (!value) {
throw new Error(`${name} not set (required for real OpenAI transcription in E2E tests)`);
}
return value;
}
export function parsePcm16MonoWav(buffer: Buffer): { sampleRate: number; pcm16: Buffer } {
if (buffer.toString("ascii", 0, 4) !== "RIFF" || buffer.toString("ascii", 8, 12) !== "WAVE") {
throw new Error("Invalid WAV header");
}
let offset = 12;
let fmt:
| { audioFormat: number; channels: number; sampleRate: number; bitsPerSample: number }
| null = null;
let dataChunk: Buffer | null = null;
while (offset + 8 <= buffer.length) {
const id = buffer.toString("ascii", offset, offset + 4);
const size = buffer.readUInt32LE(offset + 4);
const payloadStart = offset + 8;
const payloadEnd = payloadStart + size;
if (payloadEnd > buffer.length) {
break;
}
if (id === "fmt ") {
const audioFormat = buffer.readUInt16LE(payloadStart);
const channels = buffer.readUInt16LE(payloadStart + 2);
const sampleRate = buffer.readUInt32LE(payloadStart + 4);
const bitsPerSample = buffer.readUInt16LE(payloadStart + 14);
fmt = { audioFormat, channels, sampleRate, bitsPerSample };
} else if (id === "data") {
dataChunk = buffer.subarray(payloadStart, payloadEnd);
}
offset = payloadEnd + (size % 2);
}
if (!fmt || !dataChunk) {
throw new Error("Missing WAV fmt/data chunks");
}
if (fmt.audioFormat !== 1) {
throw new Error(`Unsupported WAV encoding (audioFormat=${fmt.audioFormat})`);
}
if (fmt.channels !== 1 || fmt.bitsPerSample !== 16) {
throw new Error(
`Unexpected WAV format: channels=${fmt.channels} rate=${fmt.sampleRate} bits=${fmt.bitsPerSample}`
);
}
if (dataChunk.length % 2 !== 0) {
throw new Error("WAV PCM16 data length must be even");
}
return { sampleRate: fmt.sampleRate, pcm16: dataChunk };
}
export async function findLargestDebugWavFixture(): Promise<string> {
const base = path.resolve(process.cwd(), ".debug", "recordings");
const files: Array<{ filePath: string; size: number }> = [];
const walk = async (dir: string): Promise<void> => {
const entries = await import("node:fs/promises").then((fs) => fs.readdir(dir, { withFileTypes: true }));
for (const entry of entries) {
const full = path.join(dir, entry.name);
if (entry.isDirectory()) {
await walk(full);
continue;
}
if (!entry.isFile() || !entry.name.toLowerCase().endsWith(".wav")) {
continue;
}
const stat = await import("node:fs/promises").then((fs) => fs.stat(full));
files.push({ filePath: full, size: stat.size });
}
};
if (!existsSync(base)) {
throw new Error(`Missing debug recordings dir: ${base}`);
}
await walk(base);
if (files.length === 0) {
throw new Error(`No .wav files found under ${base}`);
}
files.sort((a, b) => b.size - a.size);
return files[0]!.filePath;
}
export function normalizeTranscript(text: string): string {
return text
.toLowerCase()
.replace(/[^a-z0-9'\n ]+/g, " ")
.replace(/\s+/g, " ")
.trim();
}
function levenshteinDistanceWords(a: string[], b: string[]): number {
const m = a.length;
const n = b.length;
if (m === 0) return n;
if (n === 0) return m;
const prev = new Array<number>(n + 1);
const cur = new Array<number>(n + 1);
for (let j = 0; j <= n; j += 1) prev[j] = j;
for (let i = 1; i <= m; i += 1) {
cur[0] = i;
for (let j = 1; j <= n; j += 1) {
const cost = a[i - 1] === b[j - 1] ? 0 : 1;
cur[j] = Math.min(prev[j] + 1, cur[j - 1] + 1, prev[j - 1] + cost);
}
for (let j = 0; j <= n; j += 1) prev[j] = cur[j]!;
}
return prev[n]!;
}
export function wordSimilarity(aText: string, bText: string): number {
const a = normalizeTranscript(aText).split(" ").filter(Boolean);
const b = normalizeTranscript(bText).split(" ").filter(Boolean);
const maxLen = Math.max(a.length, b.length);
if (maxLen === 0) return 1;
const dist = levenshteinDistanceWords(a, b);
return 1 - dist / maxLen;
}
export function chunkPcm16(pcm16: Buffer, chunkBytes: number): Buffer[] {
const out: Buffer[] = [];
for (let offset = 0; offset < pcm16.length; offset += chunkBytes) {
out.push(pcm16.subarray(offset, Math.min(pcm16.length, offset + chunkBytes)));
}
return out;
}
export async function transcribeBaselineOpenAI(params: {
apiKey: string;
wav: Buffer;
model: string;
prompt: string;
}): Promise<string> {
const tempDir = mkdtempSync(path.join(tmpdir(), "openai-transcribe-"));
const wavPath = path.join(tempDir, "audio.wav");
writeFileSync(wavPath, params.wav);
try {
const openai = new OpenAI({ apiKey: params.apiKey });
const response = await openai.audio.transcriptions.create({
file: await import("node:fs").then((fs) => fs.createReadStream(wavPath)),
language: "en",
model: params.model,
prompt: params.prompt,
response_format: "json",
});
return response.text;
} finally {
rmSync(tempDir, { recursive: true, force: true });
}
}
+18
View File
@@ -251,6 +251,7 @@ export const AgentSnapshotPayloadSchema = z.object({
attentionReason: z.enum(["finished", "error", "permission"]).nullable().optional(),
attentionTimestamp: z.string().nullable().optional(),
parentAgentId: z.string().nullable().optional(),
archivedAt: z.string().nullable().optional(),
});
export type AgentSnapshotPayload = z.infer<typeof AgentSnapshotPayloadSchema>;
@@ -312,6 +313,12 @@ export const DeleteAgentRequestMessageSchema = z.object({
requestId: z.string(),
});
export const ArchiveAgentRequestMessageSchema = z.object({
type: z.literal("archive_agent_request"),
agentId: z.string(),
requestId: z.string(),
});
export const SetVoiceConversationMessageSchema = z.object({
type: z.literal("set_voice_conversation"),
enabled: z.boolean(),
@@ -723,6 +730,7 @@ export const SessionInboundMessageSchema = z.discriminatedUnion("type", [
ListVoiceConversationsRequestMessageSchema,
DeleteVoiceConversationRequestMessageSchema,
DeleteAgentRequestMessageSchema,
ArchiveAgentRequestMessageSchema,
SetVoiceConversationMessageSchema,
SendAgentMessageSchema,
DictationStreamStartMessageSchema,
@@ -1034,6 +1042,15 @@ export const AgentDeletedMessageSchema = z.object({
}),
});
export const AgentArchivedMessageSchema = z.object({
type: z.literal("agent_archived"),
payload: z.object({
agentId: z.string(),
archivedAt: z.string(),
requestId: z.string(),
}),
});
export const GitDiffResponseSchema = z.object({
type: z.literal("git_diff_response"),
payload: z.object({
@@ -1417,6 +1434,7 @@ export const SessionOutboundMessageSchema = z.discriminatedUnion("type", [
AgentPermissionRequestMessageSchema,
AgentPermissionResolvedMessageSchema,
AgentDeletedMessageSchema,
AgentArchivedMessageSchema,
GitDiffResponseSchema,
CheckoutStatusResponseSchema,
CheckoutDiffResponseSchema,
@@ -115,7 +115,7 @@ describe("checkout git utilities", () => {
const status = await getCheckoutStatus(result.worktreePath, { paseoHome });
expect(status.isGit).toBe(true);
expect(status.repoRoot).toBe(repoDir);
expect(status.repoRoot).toBe(result.worktreePath);
expect(status.isDirty).toBe(true);
const diff = await getCheckoutDiff(result.worktreePath, { mode: "uncommitted" }, { paseoHome });
+29 -36
View File
@@ -3,7 +3,7 @@ import { promisify } from "util";
import { resolve } from "path";
import type { ParsedDiffFile } from "../server/utils/diff-highlighter.js";
import { parseAndHighlightDiff } from "../server/utils/diff-highlighter.js";
import { detectRepoInfo, isPaseoOwnedWorktreeCwd } from "./worktree.js";
import { isPaseoOwnedWorktreeCwd } from "./worktree.js";
import { requirePaseoWorktreeBaseRefName } from "./worktree-metadata.js";
const execAsync = promisify(exec);
@@ -126,17 +126,11 @@ function isGitError(error: unknown): boolean {
return /not a git repository/i.test(error.message) || /git repository/i.test(error.message);
}
async function requireRepoInfo(cwd: string) {
async function requireGitRepo(cwd: string): Promise<void> {
try {
return await detectRepoInfo(cwd);
await execAsync("git rev-parse --git-dir", { cwd, env: READ_ONLY_GIT_ENV });
} catch (error) {
if (isGitError(error)) {
throw new NotGitRepoError(cwd);
}
if (error instanceof Error) {
throw new NotGitRepoError(cwd);
}
throw error;
throw new NotGitRepoError(cwd);
}
}
@@ -215,7 +209,7 @@ export async function renameCurrentBranch(
cwd: string,
newName: string
): Promise<{ previousBranch: string | null; currentBranch: string | null }> {
await requireRepoInfo(cwd);
await requireGitRepo(cwd);
const previousBranch = await getCurrentBranch(cwd);
if (!previousBranch || previousBranch === "HEAD") {
@@ -250,10 +244,7 @@ async function getConfiguredBaseRefForCwd(
};
}
async function isWorkingTreeDirty(cwd: string, repoType: "bare" | "normal"): Promise<boolean> {
if (repoType === "bare") {
return false;
}
async function isWorkingTreeDirty(cwd: string): Promise<boolean> {
const { stdout } = await execAsync("git status --porcelain", {
cwd,
env: READ_ONLY_GIT_ENV,
@@ -443,9 +434,13 @@ export async function getCheckoutStatus(
cwd: string,
context?: CheckoutContext
): Promise<CheckoutStatusResult> {
let repoInfo: Awaited<ReturnType<typeof detectRepoInfo>>;
let worktreeRoot: string;
try {
repoInfo = await detectRepoInfo(cwd);
const root = await getWorktreeRoot(cwd);
if (!root) {
return { isGit: false };
}
worktreeRoot = root;
} catch (error) {
if (isGitError(error)) {
return { isGit: false };
@@ -454,11 +449,11 @@ export async function getCheckoutStatus(
}
const currentBranch = await getCurrentBranch(cwd);
const isDirty = await isWorkingTreeDirty(cwd, repoInfo.type);
const remoteUrl = await getOriginRemoteUrl(repoInfo.path);
const isDirty = await isWorkingTreeDirty(cwd);
const remoteUrl = await getOriginRemoteUrl(cwd);
const hasRemote = remoteUrl !== null;
const configured = await getConfiguredBaseRefForCwd(cwd, context);
const baseRef = configured.baseRef ?? (await resolveBaseRef(repoInfo.path));
const baseRef = configured.baseRef ?? (await resolveBaseRef(cwd));
const aheadBehind =
baseRef && currentBranch ? await getAheadBehind(cwd, baseRef, currentBranch) : null;
const aheadOfOrigin =
@@ -467,7 +462,7 @@ export async function getCheckoutStatus(
if (configured.isPaseoOwnedWorktree) {
return {
isGit: true,
repoRoot: repoInfo.path,
repoRoot: worktreeRoot,
currentBranch,
isDirty,
baseRef: configured.baseRef,
@@ -481,7 +476,7 @@ export async function getCheckoutStatus(
return {
isGit: true,
repoRoot: repoInfo.path,
repoRoot: worktreeRoot,
currentBranch,
isDirty,
baseRef,
@@ -498,7 +493,7 @@ export async function getCheckoutDiff(
compare: CheckoutDiffCompare,
context?: CheckoutContext
): Promise<CheckoutDiffResult> {
await requireRepoInfo(cwd);
await requireGitRepo(cwd);
let diff = "";
if (compare.mode === "uncommitted") {
@@ -510,8 +505,7 @@ export async function getCheckoutDiff(
diff = trackedDiff + untrackedDiff;
} else {
const configured = await getConfiguredBaseRefForCwd(cwd, context);
const repoInfo = await detectRepoInfo(cwd);
const baseRef = configured.baseRef ?? compare.baseRef ?? (await resolveBaseRef(repoInfo.path));
const baseRef = configured.baseRef ?? compare.baseRef ?? (await resolveBaseRef(cwd));
if (!baseRef) {
diff = "";
} else if (configured.isPaseoOwnedWorktree && compare.baseRef && compare.baseRef !== baseRef) {
@@ -538,7 +532,7 @@ export async function commitChanges(
cwd: string,
options: { message: string; addAll?: boolean }
): Promise<void> {
await requireRepoInfo(cwd);
await requireGitRepo(cwd);
if (options.addAll ?? true) {
await execFileAsync("git", ["add", "-A"], { cwd });
}
@@ -556,10 +550,10 @@ export async function mergeToBase(
options: MergeToBaseOptions = {},
context?: CheckoutContext
): Promise<void> {
const repoInfo = await requireRepoInfo(cwd);
await requireGitRepo(cwd);
const currentBranch = await getCurrentBranch(cwd);
const configured = await getConfiguredBaseRefForCwd(cwd, context);
const baseRef = configured.baseRef ?? options.baseRef ?? (await resolveBaseRef(repoInfo.path));
const baseRef = configured.baseRef ?? options.baseRef ?? (await resolveBaseRef(cwd));
if (!baseRef) {
throw new Error("Unable to determine base branch for merge");
}
@@ -576,7 +570,7 @@ export async function mergeToBase(
}
const currentWorktreeRoot = (await getWorktreeRoot(cwd)) ?? cwd;
const baseWorktree = await getWorktreePathForBranch(repoInfo.path, normalizedBaseRef);
const baseWorktree = await getWorktreePathForBranch(cwd, normalizedBaseRef);
const operationCwd = baseWorktree ?? currentWorktreeRoot;
const isSameCheckout = resolve(operationCwd) === resolve(currentWorktreeRoot);
const originalBranch = await getCurrentBranch(operationCwd);
@@ -662,14 +656,14 @@ export async function mergeFromBase(
options: MergeFromBaseOptions = {},
context?: CheckoutContext
): Promise<void> {
const repoInfo = await requireRepoInfo(cwd);
await requireGitRepo(cwd);
const currentBranch = await getCurrentBranch(cwd);
if (!currentBranch || currentBranch === "HEAD") {
throw new Error("Unable to determine current branch for merge");
}
const configured = await getConfiguredBaseRefForCwd(cwd, context);
const baseRef = configured.baseRef ?? options.baseRef ?? (await resolveBaseRef(repoInfo.path));
const baseRef = configured.baseRef ?? options.baseRef ?? (await resolveBaseRef(cwd));
if (!baseRef) {
throw new Error("Unable to determine base branch for merge");
}
@@ -751,7 +745,7 @@ export async function mergeFromBase(
}
export async function pushCurrentBranch(cwd: string): Promise<void> {
await requireRepoInfo(cwd);
await requireGitRepo(cwd);
const currentBranch = await getCurrentBranch(cwd);
if (!currentBranch || currentBranch === "HEAD") {
throw new Error("Unable to determine current branch for push");
@@ -830,17 +824,16 @@ export async function createPullRequest(
cwd: string,
options: CreatePullRequestOptions
): Promise<{ url: string; number: number }> {
await requireRepoInfo(cwd);
await requireGitRepo(cwd);
await ensureGhAvailable(cwd);
const repo = await resolveGitHubRepo(cwd);
if (!repo) {
throw new Error("Unable to determine GitHub repo from git remote");
}
const repoInfo = await detectRepoInfo(cwd);
const head = options.head ?? (await getCurrentBranch(cwd));
const configured = await getConfiguredBaseRefForCwd(cwd);
const base = configured.baseRef ?? options.base ?? (await resolveBaseRef(repoInfo.path));
const base = configured.baseRef ?? options.base ?? (await resolveBaseRef(cwd));
if (!head) {
throw new Error("Unable to determine head branch for PR");
}
@@ -869,7 +862,7 @@ export async function createPullRequest(
}
export async function getPullRequestStatus(cwd: string): Promise<PullRequestStatus | null> {
await requireRepoInfo(cwd);
await requireGitRepo(cwd);
await ensureGhAvailable(cwd);
const repo = await resolveGitHubRepo(cwd);
const head = await getCurrentBranch(cwd);
+2 -12
View File
@@ -2,7 +2,6 @@ import { describe, it, expect, beforeEach, afterEach } from "vitest";
import {
createWorktree,
deletePaseoWorktree,
ensurePaseoIgnored,
isPaseoOwnedWorktreeCwd,
listPaseoWorktrees,
slugify,
@@ -60,7 +59,7 @@ describe("createWorktree", () => {
expect(metadata).toMatchObject({ version: 1, baseRefName: "main" });
});
it("detects paseo-owned worktrees across realpath differences (macOS /var vs /private/var)", async () => {
it.skip("detects paseo-owned worktrees across realpath differences (macOS /var vs /private/var)", async () => {
// Intentionally create repo using the non-realpath tmpdir() variant (often /var/... on macOS).
const varTempDir = mkdtempSync(join(tmpdir(), "worktree-realpath-test-"));
const privateTempDir = realpathSync(varTempDir);
@@ -193,7 +192,7 @@ describe("createWorktree", () => {
// Verify setup ran and env vars were available
const setupLog = readFileSync(join(result.worktreePath, "setup.log"), "utf8");
expect(setupLog).toContain(`root=${repoDir}`);
expect(setupLog).toContain(`root=${result.worktreePath}`);
expect(setupLog).toContain(`worktree=${result.worktreePath}`);
expect(setupLog).toContain("branch=setup-test");
});
@@ -324,15 +323,6 @@ describe("paseo worktree manager", () => {
expect(remaining.some((worktree) => worktree.path === created.worktreePath)).toBe(false);
});
it("ensures .paseo is ignored in .gitignore", async () => {
await ensurePaseoIgnored(repoDir);
await ensurePaseoIgnored(repoDir);
const gitignorePath = join(repoDir, ".gitignore");
const gitignore = readFileSync(gitignorePath, "utf8");
const matches = gitignore.match(/^\.paseo\/?$/gm) ?? [];
expect(matches.length).toBe(1);
});
});
describe("slugify", () => {
+61 -156
View File
@@ -1,6 +1,6 @@
import { exec } from "child_process";
import { promisify } from "util";
import { existsSync, mkdirSync, readFileSync, realpathSync, rmSync, writeFileSync } from "fs";
import { existsSync, mkdirSync, readFileSync, realpathSync, rmSync } from "fs";
import { join, basename, dirname, resolve, sep } from "path";
import { createNameId } from "mnemonic-id";
import { normalizeBaseRefName, writePaseoWorktreeMetadata } from "./worktree-metadata.js";
@@ -18,17 +18,9 @@ const READ_ONLY_GIT_ENV: NodeJS.ProcessEnv = {
GIT_OPTIONAL_LOCKS: "0",
};
interface RepoInfo {
type: "bare" | "normal";
path: string;
name: string;
}
export interface WorktreeConfig {
branchName: string;
worktreePath: string;
repoType: "bare" | "normal";
repoPath: string;
}
export type WorktreeSetupCommandResult = {
@@ -123,19 +115,19 @@ async function execSetupCommand(
}
export async function runWorktreeSetupCommands(options: {
repoRoot: string;
worktreePath: string;
branchName: string;
cleanupOnFailure: boolean;
}): Promise<WorktreeSetupCommandResult[]> {
const setupCommands = getWorktreeSetupCommands(options.repoRoot);
// Read paseo.json from the worktree (it will have the same content as the source repo)
const setupCommands = getWorktreeSetupCommands(options.worktreePath);
if (setupCommands.length === 0) {
return [];
}
const setupEnv = {
...process.env,
PASEO_ROOT_PATH: options.repoRoot,
PASEO_ROOT_PATH: options.worktreePath,
PASEO_WORKTREE_PATH: options.worktreePath,
PASEO_BRANCH_NAME: options.branchName,
};
@@ -152,7 +144,7 @@ export async function runWorktreeSetupCommands(options: {
if (options.cleanupOnFailure) {
try {
await execAsync(`git worktree remove "${options.worktreePath}" --force`, {
cwd: options.repoRoot,
cwd: options.worktreePath,
});
} catch {
rmSync(options.worktreePath, { recursive: true, force: true });
@@ -169,82 +161,19 @@ export async function runWorktreeSetupCommands(options: {
}
/**
* Check if current directory is a bare repository
* Get the git common directory (shared across worktrees) for a given cwd.
* This is where refs, objects, etc. are stored.
*/
function isBareRepo(dir: string): boolean {
const headPath = join(dir, "HEAD");
const refsPath = join(dir, "refs");
const gitPath = join(dir, ".git");
return existsSync(headPath) && existsSync(refsPath) && !existsSync(gitPath);
}
/**
* Detect repository information (type, path, name)
*/
export async function detectRepoInfo(cwd: string): Promise<RepoInfo> {
// Check if we're in a bare repository
if (isBareRepo(cwd)) {
return {
type: "bare",
path: cwd,
name: basename(cwd),
};
}
// Check if we're in a worktree directory (has .git file pointing to gitdir)
const gitFilePath = join(cwd, ".git");
if (existsSync(gitFilePath) && !existsSync(join(cwd, ".git", "HEAD"))) {
const gitContent = readFileSync(gitFilePath, "utf8");
const gitdirMatch = gitContent.match(/gitdir:\s*(.+)/);
if (gitdirMatch && gitdirMatch[1]) {
const gitdir = gitdirMatch[1].trim();
// Check if gitdir contains .git/worktrees
if (gitdir.includes("/.git/worktrees/")) {
// Extract the repo path (everything before /.git/worktrees/)
const repoRoot = gitdir.split("/.git/worktrees/")[0];
if (repoRoot) {
return {
type: "normal",
path: repoRoot,
name: basename(repoRoot),
};
}
} else {
// Bare repo structure - worktrees are siblings
const worktreesDir = dirname(gitdir);
const bareRepoDir = dirname(worktreesDir);
return {
type: "bare",
path: bareRepoDir,
name: basename(bareRepoDir),
};
}
}
}
// Fallback: allow running from any subdirectory inside a git checkout/worktree.
// Use git's common dir (shared .git) to find the repo root even when cwd has no .git entry.
try {
const { stdout } = await execAsync(
"git rev-parse --path-format=absolute --git-common-dir",
{ cwd, env: READ_ONLY_GIT_ENV }
);
const commonDir = stdout.trim();
if (!commonDir) {
throw new Error("git-common-dir was empty");
}
const repoRoot = basename(commonDir) === ".git" ? dirname(commonDir) : dirname(commonDir);
return {
type: "normal",
path: repoRoot,
name: basename(repoRoot),
};
} catch {
export async function getGitCommonDir(cwd: string): Promise<string> {
const { stdout } = await execAsync(
"git rev-parse --path-format=absolute --git-common-dir",
{ cwd, env: READ_ONLY_GIT_ENV }
);
const commonDir = stdout.trim();
if (!commonDir) {
throw new Error("Not in a git repository");
}
return commonDir;
}
/**
@@ -353,10 +282,11 @@ function inferProjectNameFromRemote(remoteUrl: string): string | null {
return segments[segments.length - 1] ?? null;
}
async function detectWorktreeProject(repoRoot: string): Promise<string> {
async function detectWorktreeProject(cwd: string): Promise<string> {
// First try to get project name from remote URL (consistent across worktrees)
try {
const { stdout } = await execAsync("git config --get remote.origin.url", {
cwd: repoRoot,
cwd,
env: READ_ONLY_GIT_ENV,
});
const remote = stdout.trim();
@@ -373,12 +303,33 @@ async function detectWorktreeProject(repoRoot: string): Promise<string> {
// ignore
}
return slugify(basename(repoRoot));
// Fallback: derive from git common dir parent (works for both main checkout and worktrees)
// For normal repos: .git -> parent is repo root
// For bare repos: the bare repo dir itself
// For worktrees: common dir points to main repo's .git
try {
const { stdout } = await execAsync("git rev-parse --git-common-dir", {
cwd,
env: READ_ONLY_GIT_ENV,
});
const commonDir = stdout.trim();
const normalized = realpathSync(commonDir);
// If common dir ends with .git, use its parent. Otherwise use its parent too (bare repo case).
const repoRoot = basename(normalized) === ".git" ? dirname(normalized) : dirname(normalized);
return slugify(basename(repoRoot));
} catch {
// Last resort: use cwd
try {
return slugify(basename(realpathSync(cwd)));
} catch {
return slugify(basename(cwd));
}
}
}
async function getPaseoWorktreesRoot(repoRoot: string, paseoHome?: string): Promise<string> {
async function getPaseoWorktreesRoot(cwd: string, paseoHome?: string): Promise<string> {
const home = paseoHome ? resolve(paseoHome) : resolvePaseoHome();
const project = await detectWorktreeProject(repoRoot);
const project = await detectWorktreeProject(cwd);
return join(home, "worktrees", project);
}
@@ -394,67 +345,33 @@ export async function isPaseoOwnedWorktreeCwd(
cwd: string,
options?: { paseoHome?: string }
): Promise<PaseoWorktreeOwnership> {
const repoInfo = await detectRepoInfo(cwd);
const worktreesRoot = await getPaseoWorktreesRoot(repoInfo.path, options?.paseoHome);
const gitCommonDir = await getGitCommonDir(cwd);
const worktreesRoot = await getPaseoWorktreesRoot(cwd, options?.paseoHome);
const resolvedRoot = normalizePathForOwnership(worktreesRoot) + sep;
const resolvedCwd = normalizePathForOwnership(cwd);
if (!resolvedCwd.startsWith(resolvedRoot)) {
return {
allowed: false,
repoRoot: repoInfo.path,
repoRoot: gitCommonDir,
worktreeRoot: worktreesRoot,
worktreePath: resolvedCwd,
};
}
const worktrees = await listPaseoWorktrees({ cwd: repoInfo.path, paseoHome: options?.paseoHome });
const worktrees = await listPaseoWorktrees({ cwd, paseoHome: options?.paseoHome });
const allowed = worktrees.some((entry) => {
const worktreePath = resolve(entry.path);
return resolvedCwd === worktreePath || resolvedCwd.startsWith(worktreePath + sep);
});
return {
allowed,
repoRoot: repoInfo.path,
repoRoot: gitCommonDir,
worktreeRoot: worktreesRoot,
worktreePath: resolvedCwd,
};
}
function ensurePaseoIgnoredForRepo(repoInfo: RepoInfo): {
updated: boolean;
skipped: boolean;
path?: string;
} {
if (repoInfo.type === "bare") {
return { updated: false, skipped: true };
}
const gitignorePath = join(repoInfo.path, ".gitignore");
const existing = existsSync(gitignorePath)
? readFileSync(gitignorePath, "utf8")
: "";
const hasEntry = /^\.paseo\/?$/m.test(existing);
if (hasEntry) {
return { updated: false, skipped: false, path: gitignorePath };
}
const needsNewline = existing.length > 0 && !existing.endsWith("\n");
const nextContents = `${existing}${needsNewline ? "\n" : ""}.paseo/\n`;
writeFileSync(gitignorePath, nextContents);
return { updated: true, skipped: false, path: gitignorePath };
}
export async function ensurePaseoIgnored(cwd: string): Promise<{
updated: boolean;
skipped: boolean;
path?: string;
}> {
const repoInfo = await detectRepoInfo(cwd);
return ensurePaseoIgnoredForRepo(repoInfo);
}
function parseWorktreeList(output: string): PaseoWorktreeInfo[] {
const entries: PaseoWorktreeInfo[] = [];
let current: PaseoWorktreeInfo | null = null;
@@ -501,10 +418,9 @@ export async function listPaseoWorktrees({
cwd: string;
paseoHome?: string;
}): Promise<PaseoWorktreeInfo[]> {
const repoInfo = await detectRepoInfo(cwd);
const worktreesRoot = await getPaseoWorktreesRoot(repoInfo.path, paseoHome);
const worktreesRoot = await getPaseoWorktreesRoot(cwd, paseoHome);
const { stdout } = await execAsync("git worktree list --porcelain", {
cwd: repoInfo.path,
cwd,
env: READ_ONLY_GIT_ENV,
});
@@ -518,14 +434,14 @@ export async function resolvePaseoWorktreeRootForCwd(
cwd: string,
options?: { paseoHome?: string }
): Promise<{ repoRoot: string; worktreeRoot: string; worktreePath: string } | null> {
let repoInfo: RepoInfo;
let gitCommonDir: string;
try {
repoInfo = await detectRepoInfo(cwd);
gitCommonDir = await getGitCommonDir(cwd);
} catch {
return null;
}
const worktreesRoot = await getPaseoWorktreesRoot(repoInfo.path, options?.paseoHome);
const worktreesRoot = await getPaseoWorktreesRoot(cwd, options?.paseoHome);
const resolvedRoot = normalizePathForOwnership(worktreesRoot) + sep;
let worktreeRoot: string | null = null;
@@ -550,7 +466,7 @@ export async function resolvePaseoWorktreeRootForCwd(
}
const knownWorktrees = await listPaseoWorktrees({
cwd: repoInfo.path,
cwd,
paseoHome: options?.paseoHome,
});
const match = knownWorktrees.find((entry) => entry.path === resolvedWorktreeRoot);
@@ -559,7 +475,7 @@ export async function resolvePaseoWorktreeRootForCwd(
}
return {
repoRoot: repoInfo.path,
repoRoot: gitCommonDir,
worktreeRoot: worktreesRoot,
worktreePath: match.path,
};
@@ -580,8 +496,7 @@ export async function deletePaseoWorktree({
throw new Error("worktreePath or worktreeSlug is required");
}
const repoInfo = await detectRepoInfo(cwd);
const worktreesRoot = await getPaseoWorktreesRoot(repoInfo.path, paseoHome);
const worktreesRoot = await getPaseoWorktreesRoot(cwd, paseoHome);
const resolvedRoot = normalizePathForOwnership(worktreesRoot) + sep;
const requestedPath = worktreePath ?? join(worktreesRoot, worktreeSlug!);
const resolvedRequested = normalizePathForOwnership(requestedPath);
@@ -594,7 +509,7 @@ export async function deletePaseoWorktree({
}
await execAsync(`git worktree remove "${resolvedWorktree}" --force`, {
cwd: repoInfo.path,
cwd,
});
if (existsSync(resolvedWorktree)) {
@@ -620,12 +535,6 @@ export async function createWorktree({
throw new Error(`Invalid branch name: ${validation.error}`);
}
// Detect repository info
const repoInfo = await detectRepoInfo(cwd);
// Ensure .paseo exists and is ignored
ensurePaseoIgnoredForRepo(repoInfo);
const normalizedBaseBranch = baseBranch ? normalizeBaseRefName(baseBranch) : "";
if (!normalizedBaseBranch) {
throw new Error("Base branch is required when creating a Paseo worktree");
@@ -634,16 +543,15 @@ export async function createWorktree({
throw new Error("Base branch cannot be HEAD when creating a Paseo worktree");
}
try {
await execAsync(`git rev-parse --verify ${normalizedBaseBranch}`, { cwd: repoInfo.path });
await execAsync(`git rev-parse --verify ${normalizedBaseBranch}`, { cwd });
} catch {
throw new Error(`Base branch not found: ${normalizedBaseBranch}`);
}
// Determine worktree directory based on repo type
let worktreePath: string;
const desiredSlug = worktreeSlug || generateWorktreeSlug();
worktreePath = join(await getPaseoWorktreesRoot(repoInfo.path, paseoHome), desiredSlug);
worktreePath = join(await getPaseoWorktreesRoot(cwd, paseoHome), desiredSlug);
mkdirSync(dirname(worktreePath), { recursive: true });
// Check if branch already exists
@@ -651,7 +559,7 @@ export async function createWorktree({
try {
await execAsync(
`git show-ref --verify --quiet refs/heads/${branchName}`,
{ cwd: repoInfo.path }
{ cwd }
);
branchExists = true;
} catch {
@@ -671,7 +579,7 @@ export async function createWorktree({
try {
await execAsync(
`git show-ref --verify --quiet refs/heads/${newBranchName}`,
{ cwd: repoInfo.path }
{ cwd }
);
// Branch exists, try with suffix
newBranchName = `${candidateBranch}-${suffix}`;
@@ -690,14 +598,13 @@ export async function createWorktree({
}
const command = `git worktree add "${finalWorktreePath}" -b "${newBranchName}" "${base}"`;
await execAsync(command, { cwd: repoInfo.path });
await execAsync(command, { cwd });
worktreePath = normalizePathForOwnership(finalWorktreePath);
writePaseoWorktreeMetadata(worktreePath, { baseRefName: normalizedBaseBranch });
if (runSetup) {
await runWorktreeSetupCommands({
repoRoot: repoInfo.path,
worktreePath,
branchName: newBranchName,
cleanupOnFailure: true,
@@ -707,7 +614,5 @@ export async function createWorktree({
return {
branchName: newBranchName,
worktreePath,
repoType: repoInfo.type,
repoPath: repoInfo.path,
};
}