History

Address PR review comments for terminal integration

Julius Marminge committed 9 months ago

Commit 3bfda731 parent 0133d958 files · +374 −147Browse files

Unified · Split

8 files
  1. apps/server/src/git.test.ts +15 −0
  2. apps/server/src/git.ts +61 −4
  3. apps/server/src/terminalManager.test.ts +32 −10
  4. apps/server/src/terminalManager.ts +223 −109
  5. apps/server/src/wsServer.test.ts +16 −0
  6. apps/server/src/wsServer.ts +5 −2
  7. apps/web/src/components/ChatView.tsx +0 −1
  8. apps/web/src/components/ThreadTerminalDrawer.tsx +22 −21
apps/server/src/git.test.ts+15−0
@@ -53,6 +53,21 @@ async function initRepoWithCommit(cwd: string): Promise<void> { · 52 unmodified lines
53// ── Tests ──53// ── Tests ──
54 54
55describe("git integration", () => {55describe("git integration", () => {
56 describe("runTerminalCommand", () => {
57 it("caps captured output when maxOutputBytes is exceeded", async () => {
58 const result = await runTerminalCommand({
59 command: `node -e "process.stdout.write('x'.repeat(2000))"`,
60 cwd: process.cwd(),
61 timeoutMs: 10_000,
62 maxOutputBytes: 128,
63 });
64
65 expect(result.code).toBe(0);
66 expect(result.stdout.length).toBeLessThanOrEqual(128);
67 expect(result.stderr).toContain("output truncated");
68 });
69 });
70
56 // ── initGitRepo ──71 // ── initGitRepo ──
57 72
58 describe("initGitRepo", () => {73 describe("initGitRepo", () => {
apps/server/src/git.ts+61−4
@@ -18,6 +18,7 @@ export interface TerminalCommandInput { · 17 unmodified lines
18 command: string;18 command: string;
19 cwd: string;19 cwd: string;
20 timeoutMs?: number;20 timeoutMs?: number;
21 maxOutputBytes?: number;
21}22}
22 23
23export interface TerminalCommandResult {24export interface TerminalCommandResult {
@@ -28,6 +29,36 @@ export interface TerminalCommandResult { · 4 unmodified lines
28 timedOut: boolean;29 timedOut: boolean;
29}30}
30 31
32const DEFAULT_MAX_OUTPUT_BYTES = 1_000_000;
33
34function appendChunkWithinLimit(
35 target: string,
36 currentBytes: number,
37 chunk: Buffer,
38 maxBytes: number,
39): {
40 next: string;
41 nextBytes: number;
42 truncated: boolean;
43} {
44 const remaining = maxBytes - currentBytes;
45 if (remaining <= 0) {
46 return { next: target, nextBytes: currentBytes, truncated: true };
47 }
48 if (chunk.length <= remaining) {
49 return {
50 next: `${target}${chunk.toString()}`,
51 nextBytes: currentBytes + chunk.length,
52 truncated: false,
53 };
54 }
55 return {
56 next: `${target}${chunk.subarray(0, remaining).toString()}`,
57 nextBytes: currentBytes + remaining,
58 truncated: true,
59 };
60}
61
31/** Spawn git directly with an argv array — no shell, no quoting needed. */62/** Spawn git directly with an argv array — no shell, no quoting needed. */
32function runGit(args: string[], cwd: string, timeoutMs = 30_000): Promise<TerminalCommandResult> {63function runGit(args: string[], cwd: string, timeoutMs = 30_000): Promise<TerminalCommandResult> {
33 return new Promise((resolve, reject) => {64 return new Promise((resolve, reject) => {
@@ -37,9 +68,13 @@ function runGit(args: string[], cwd: string, timeoutMs = 30_000): Promise<Termin · 3 unmodified lines
37 stdio: ["ignore", "pipe", "pipe"],68 stdio: ["ignore", "pipe", "pipe"],
38 });69 });
39 70
71 const maxOutputBytes = DEFAULT_MAX_OUTPUT_BYTES;
40 let stdout = "";72 let stdout = "";
41 let stderr = "";73 let stderr = "";
74 let stdoutBytes = 0;
75 let stderrBytes = 0;
42 let timedOut = false;76 let timedOut = false;
77 let outputTruncated = false;
43 78
44 const timeout = setTimeout(() => {79 const timeout = setTimeout(() => {
45 timedOut = true;80 timedOut = true;
@@ -50,10 +85,16 @@ function runGit(args: string[], cwd: string, timeoutMs = 30_000): Promise<Termin · 4 unmodified lines
50 }, timeoutMs);85 }, timeoutMs);
51 86
52 child.stdout?.on("data", (chunk: Buffer) => {87 child.stdout?.on("data", (chunk: Buffer) => {
53 stdout += chunk.toString();88 const appended = appendChunkWithinLimit(stdout, stdoutBytes, chunk, maxOutputBytes);
89 stdout = appended.next;
90 stdoutBytes = appended.nextBytes;
91 outputTruncated = outputTruncated || appended.truncated;
54 });92 });
55 child.stderr?.on("data", (chunk: Buffer) => {93 child.stderr?.on("data", (chunk: Buffer) => {
56 stderr += chunk.toString();94 const appended = appendChunkWithinLimit(stderr, stderrBytes, chunk, maxOutputBytes);
95 stderr = appended.next;
96 stderrBytes = appended.nextBytes;
97 outputTruncated = outputTruncated || appended.truncated;
57 });98 });
58 child.on("error", (error) => {99 child.on("error", (error) => {
59 clearTimeout(timeout);100 clearTimeout(timeout);
@@ -61,6 +102,9 @@ function runGit(args: string[], cwd: string, timeoutMs = 30_000): Promise<Termin · 1 unmodified line
61 });102 });
62 child.on("close", (code, signal) => {103 child.on("close", (code, signal) => {
63 clearTimeout(timeout);104 clearTimeout(timeout);
105 if (outputTruncated) {
106 stderr = `${stderr}\n[output truncated at ${maxOutputBytes} bytes]`;
107 }
64 resolve({ stdout, stderr, code: code ?? null, signal: signal ?? null, timedOut });108 resolve({ stdout, stderr, code: code ?? null, signal: signal ?? null, timedOut });
65 });109 });
66 });110 });
@@ -69,6 +113,7 @@ function runGit(args: string[], cwd: string, timeoutMs = 30_000): Promise<Termin · 2 unmodified lines
69export async function runTerminalCommand(113export async function runTerminalCommand(
70 input: TerminalCommandInput,114 input: TerminalCommandInput,
71): Promise<TerminalCommandResult> {115): Promise<TerminalCommandResult> {
116 const maxOutputBytes = input.maxOutputBytes ?? DEFAULT_MAX_OUTPUT_BYTES;
72 const shellPath =117 const shellPath =
73 process.platform === "win32"118 process.platform === "win32"
74 ? (process.env.ComSpec ?? "cmd.exe")119 ? (process.env.ComSpec ?? "cmd.exe")
@@ -86,7 +131,10 @@ export async function runTerminalCommand( · 11 unmodified lines
86 131
87 let stdout = "";132 let stdout = "";
88 let stderr = "";133 let stderr = "";
134 let stdoutBytes = 0;
135 let stderrBytes = 0;
89 let timedOut = false;136 let timedOut = false;
137 let outputTruncated = false;
90 138
91 const timeout = setTimeout(() => {139 const timeout = setTimeout(() => {
92 timedOut = true;140 timedOut = true;
@@ -99,11 +147,17 @@ export async function runTerminalCommand( · 6 unmodified lines
99 }, input.timeoutMs ?? 30_000);147 }, input.timeoutMs ?? 30_000);
100 148
101 child.stdout?.on("data", (chunk: Buffer) => {149 child.stdout?.on("data", (chunk: Buffer) => {
102 stdout += chunk.toString();150 const appended = appendChunkWithinLimit(stdout, stdoutBytes, chunk, maxOutputBytes);
151 stdout = appended.next;
152 stdoutBytes = appended.nextBytes;
153 outputTruncated = outputTruncated || appended.truncated;
103 });154 });
104 155
105 child.stderr?.on("data", (chunk: Buffer) => {156 child.stderr?.on("data", (chunk: Buffer) => {
106 stderr += chunk.toString();157 const appended = appendChunkWithinLimit(stderr, stderrBytes, chunk, maxOutputBytes);
158 stderr = appended.next;
159 stderrBytes = appended.nextBytes;
160 outputTruncated = outputTruncated || appended.truncated;
107 });161 });
108 162
109 child.on("error", (error) => {163 child.on("error", (error) => {
@@ -113,6 +167,9 @@ export async function runTerminalCommand( · 3 unmodified lines
113 167
114 child.on("close", (code, signal) => {168 child.on("close", (code, signal) => {
115 clearTimeout(timeout);169 clearTimeout(timeout);
170 if (outputTruncated) {
171 stderr = `${stderr}\n[output truncated at ${maxOutputBytes} bytes]`;
172 }
116 resolve({173 resolve({
117 stdout,174 stdout,
118 stderr,175 stderr,
apps/server/src/terminalManager.test.ts+32−10
@@ -100,6 +100,14 @@ function openInput(overrides: Partial<TerminalOpenInput> = {}): TerminalOpenInpu · 99 unmodified lines
100 };100 };
101}101}
102 102
103function historyLogName(threadId: string): string {
104 return `terminal_${Buffer.from(threadId, "utf8").toString("base64url")}.log`;
105}
106
107function historyLogPath(logsDir: string, threadId = "thread-1"): string {
108 return path.join(logsDir, historyLogName(threadId));
109}
110
103describe("TerminalManager", () => {111describe("TerminalManager", () => {
104 const tempDirs: string[] = [];112 const tempDirs: string[] = [];
105 113
@@ -124,11 +132,12 @@ describe("TerminalManager", () => { · 18 unmodified lines
124 132
125 it("spawns lazily and reuses running terminal per thread", async () => {133 it("spawns lazily and reuses running terminal per thread", async () => {
126 const { manager, ptyAdapter } = makeManager();134 const { manager, ptyAdapter } = makeManager();
127 const first = await manager.open(openInput());135 const [first, second] = await Promise.all([manager.open(openInput()), manager.open(openInput())]);
128 const second = await manager.open(openInput());136 const third = await manager.open(openInput());
129 137
130 expect(first.threadId).toBe("thread-1");138 expect(first.threadId).toBe("thread-1");
131 expect(second.threadId).toBe("thread-1");139 expect(second.threadId).toBe("thread-1");
140 expect(third.threadId).toBe("thread-1");
132 expect(ptyAdapter.spawnInputs).toHaveLength(1);141 expect(ptyAdapter.spawnInputs).toHaveLength(1);
133 142
134 manager.dispose();143 manager.dispose();
@@ -162,9 +171,9 @@ describe("TerminalManager", () => { · 27 unmodified lines
162 if (!process) return;171 if (!process) return;
163 172
164 process.emitData("hello\n");173 process.emitData("hello\n");
165 await waitFor(() => fs.existsSync(path.join(logsDir, "thread-1.log")));174 await waitFor(() => fs.existsSync(historyLogPath(logsDir)));
166 await manager.clear({ threadId: "thread-1" });175 await manager.clear({ threadId: "thread-1" });
167 await waitFor(() => fs.readFileSync(path.join(logsDir, "thread-1.log"), "utf8") === "");176 await waitFor(() => fs.readFileSync(historyLogPath(logsDir), "utf8") === "");
168 177
169 expect(events.some((event) => event.type === "cleared")).toBe(true);178 expect(events.some((event) => event.type === "cleared")).toBe(true);
170 179
@@ -178,13 +187,13 @@ describe("TerminalManager", () => { · 7 unmodified lines
178 expect(firstProcess).toBeDefined();187 expect(firstProcess).toBeDefined();
179 if (!firstProcess) return;188 if (!firstProcess) return;
180 firstProcess.emitData("before restart\n");189 firstProcess.emitData("before restart\n");
181 await waitFor(() => fs.existsSync(path.join(logsDir, "thread-1.log")));190 await waitFor(() => fs.existsSync(historyLogPath(logsDir)));
182 191
183 const snapshot = await manager.restart(openInput());192 const snapshot = await manager.restart(openInput());
184 expect(snapshot.history).toBe("");193 expect(snapshot.history).toBe("");
185 expect(snapshot.status).toBe("running");194 expect(snapshot.status).toBe("running");
186 expect(ptyAdapter.spawnInputs).toHaveLength(2);195 expect(ptyAdapter.spawnInputs).toHaveLength(2);
187 await waitFor(() => fs.readFileSync(path.join(logsDir, "thread-1.log"), "utf8") === "");196 await waitFor(() => fs.readFileSync(historyLogPath(logsDir), "utf8") === "");
188 197
189 manager.dispose();198 manager.dispose();
190 });199 });
@@ -200,7 +209,7 @@ describe("TerminalManager", () => { · 9 unmodified lines
200 expect(process).toBeDefined();209 expect(process).toBeDefined();
201 if (!process) return;210 if (!process) return;
202 process.emitData("old data\n");211 process.emitData("old data\n");
203 await waitFor(() => fs.existsSync(path.join(logsDir, "thread-1.log")));212 await waitFor(() => fs.existsSync(historyLogPath(logsDir)));
204 process.emitExit({ exitCode: 0, signal: 0 });213 process.emitExit({ exitCode: 0, signal: 0 });
205 214
206 await waitFor(() => events.some((event) => event.type === "exited"));215 await waitFor(() => events.some((event) => event.type === "exited"));
@@ -208,7 +217,7 @@ describe("TerminalManager", () => { · 1 unmodified line
208 217
209 expect(reopened.history).toBe("");218 expect(reopened.history).toBe("");
210 expect(ptyAdapter.spawnInputs).toHaveLength(2);219 expect(ptyAdapter.spawnInputs).toHaveLength(2);
211 expect(fs.readFileSync(path.join(logsDir, "thread-1.log"), "utf8")).toBe("");220 expect(fs.readFileSync(historyLogPath(logsDir), "utf8")).toBe("");
212 221
213 manager.dispose();222 manager.dispose();
214 });223 });
@@ -237,10 +246,23 @@ describe("TerminalManager", () => { · 22 unmodified lines
237 expect(process).toBeDefined();246 expect(process).toBeDefined();
238 if (!process) return;247 if (!process) return;
239 process.emitData("bye\n");248 process.emitData("bye\n");
240 await waitFor(() => fs.existsSync(path.join(logsDir, "thread-1.log")));249 await waitFor(() => fs.existsSync(historyLogPath(logsDir)));
241 250
242 await manager.close({ threadId: "thread-1", deleteHistory: true });251 await manager.close({ threadId: "thread-1", deleteHistory: true });
243 expect(fs.existsSync(path.join(logsDir, "thread-1.log"))).toBe(false);252 expect(fs.existsSync(historyLogPath(logsDir))).toBe(false);
253
254 manager.dispose();
255 });
256
257 it("loads existing legacy transcript filenames and keeps new naming for writes", async () => {
258 const { manager, logsDir } = makeManager();
259 const legacyPath = path.join(logsDir, "thread-1.log");
260 fs.writeFileSync(legacyPath, "legacy-line\n", "utf8");
261
262 const snapshot = await manager.open(openInput());
263
264 expect(snapshot.history).toBe("legacy-line\n");
265 expect(fs.existsSync(legacyPath)).toBe(true);
244 266
245 manager.dispose();267 manager.dispose();
246 });268 });
apps/server/src/terminalManager.ts+223−109
@@ -21,6 +21,7 @@ import { createLogger } from "./logger"; · 20 unmodified lines
21import { NodePtyAdapter, type PtyAdapter, type PtyExitEvent, type PtyProcess } from "./ptyAdapter";21import { NodePtyAdapter, type PtyAdapter, type PtyExitEvent, type PtyProcess } from "./ptyAdapter";
22 22
23const DEFAULT_HISTORY_LINE_LIMIT = 5_000;23const DEFAULT_HISTORY_LINE_LIMIT = 5_000;
24const DEFAULT_PERSIST_DEBOUNCE_MS = 40;
24 25
25export interface TerminalManagerEvents {26export interface TerminalManagerEvents {
26 event: [event: TerminalEvent];27 event: [event: TerminalEvent];
@@ -68,10 +69,14 @@ function capHistory(history: string, maxLines: number): string { · 41 unmodified lines
68 return hasTrailingNewline ? `${capped}\n` : capped;69 return hasTrailingNewline ? `${capped}\n` : capped;
69}70}
70 71
71function toSafeThreadId(threadId: string): string {72function legacySafeThreadId(threadId: string): string {
72 return threadId.replace(/[^a-zA-Z0-9._-]/g, "_");73 return threadId.replace(/[^a-zA-Z0-9._-]/g, "_");
73}74}
74 75
76function toSafeThreadId(threadId: string): string {
77 return `terminal_${Buffer.from(threadId, "utf8").toString("base64url")}`;
78}
79
75export class TerminalManager extends EventEmitter<TerminalManagerEvents> {80export class TerminalManager extends EventEmitter<TerminalManagerEvents> {
76 private readonly sessions = new Map<string, TerminalSessionState>();81 private readonly sessions = new Map<string, TerminalSessionState>();
77 private readonly logsDir: string;82 private readonly logsDir: string;
@@ -79,6 +84,10 @@ export class TerminalManager extends EventEmitter<TerminalManagerEvents> { · 1 unmodified line
79 private readonly ptyAdapter: PtyAdapter;84 private readonly ptyAdapter: PtyAdapter;
80 private readonly shellResolver: () => string;85 private readonly shellResolver: () => string;
81 private readonly persistQueues = new Map<string, Promise<void>>();86 private readonly persistQueues = new Map<string, Promise<void>>();
87 private readonly persistTimers = new Map<string, ReturnType<typeof setTimeout>>();
88 private readonly pendingPersistHistory = new Map<string, string>();
89 private readonly threadLocks = new Map<string, Promise<void>>();
90 private readonly persistDebounceMs: number;
82 private readonly logger = createLogger("terminal");91 private readonly logger = createLogger("terminal");
83 92
84 constructor(options: TerminalManagerOptions = {}) {93 constructor(options: TerminalManagerOptions = {}) {
@@ -87,60 +96,63 @@ export class TerminalManager extends EventEmitter<TerminalManagerEvents> { · 2 unmodified lines
87 this.historyLineLimit = options.historyLineLimit ?? DEFAULT_HISTORY_LINE_LIMIT;96 this.historyLineLimit = options.historyLineLimit ?? DEFAULT_HISTORY_LINE_LIMIT;
88 this.ptyAdapter = options.ptyAdapter ?? new NodePtyAdapter();97 this.ptyAdapter = options.ptyAdapter ?? new NodePtyAdapter();
89 this.shellResolver = options.shellResolver ?? defaultShellResolver;98 this.shellResolver = options.shellResolver ?? defaultShellResolver;
99 this.persistDebounceMs = DEFAULT_PERSIST_DEBOUNCE_MS;
90 fs.mkdirSync(this.logsDir, { recursive: true });100 fs.mkdirSync(this.logsDir, { recursive: true });
91 }101 }
92 102
93 async open(raw: TerminalOpenInput): Promise<TerminalSessionSnapshot> {103 async open(raw: TerminalOpenInput): Promise<TerminalSessionSnapshot> {
94 const input = terminalOpenInputSchema.parse(raw);104 const input = terminalOpenInputSchema.parse(raw);
95 await this.assertValidCwd(input.cwd);105 return this.runWithThreadLock(input.threadId, async () => {
96 await this.flushPersistQueue(input.threadId);106 await this.assertValidCwd(input.cwd);
97 107
98 const existing = this.sessions.get(input.threadId);108 const existing = this.sessions.get(input.threadId);
99 if (!existing) {109 if (!existing) {
100 const history = await this.readHistory(input.threadId);110 await this.flushPersistQueue(input.threadId);
101 const session: TerminalSessionState = {111 const history = await this.readHistory(input.threadId);
102 threadId: input.threadId,112 const session: TerminalSessionState = {
103 cwd: input.cwd,113 threadId: input.threadId,
104 status: "starting",114 cwd: input.cwd,
105 pid: null,115 status: "starting",
106 history,116 pid: null,
107 exitCode: null,117 history,
108 exitSignal: null,118 exitCode: null,
109 updatedAt: new Date().toISOString(),119 exitSignal: null,
110 cols: input.cols,120 updatedAt: new Date().toISOString(),
111 rows: input.rows,121 cols: input.cols,
112 process: null,122 rows: input.rows,
113 unsubscribeData: null,123 process: null,
114 unsubscribeExit: null,124 unsubscribeData: null,
115 };125 unsubscribeExit: null,
116 this.sessions.set(input.threadId, session);126 };
117 this.startSession(session, input, "started");127 this.sessions.set(input.threadId, session);
118 return this.snapshot(session);128 this.startSession(session, input, "started");
119 }129 return this.snapshot(session);
130 }
120 131
121 if (existing.cwd !== input.cwd) {132 if (existing.cwd !== input.cwd) {
122 this.stopProcess(existing);133 this.stopProcess(existing);
123 existing.cwd = input.cwd;134 existing.cwd = input.cwd;
124 existing.history = "";135 existing.history = "";
125 await this.persistHistory(existing.threadId, existing.history);136 await this.persistHistory(existing.threadId, existing.history);
126 } else if (existing.status === "exited" || existing.status === "error") {137 } else if (existing.status === "exited" || existing.status === "error") {
127 existing.history = "";138 existing.history = "";
128 await this.persistHistory(existing.threadId, existing.history);139 await this.persistHistory(existing.threadId, existing.history);
129 }140 }
130 141
131 if (!existing.process) {142 if (!existing.process) {
132 this.startSession(existing, input, "started");143 this.startSession(existing, input, "started");
133 return this.snapshot(existing);144 return this.snapshot(existing);
134 }145 }
135 146
136 if (existing.cols !== input.cols || existing.rows !== input.rows) {147 if (existing.cols !== input.cols || existing.rows !== input.rows) {
137 existing.cols = input.cols;148 existing.cols = input.cols;
138 existing.rows = input.rows;149 existing.rows = input.rows;
139 existing.process.resize(input.cols, input.rows);150 existing.process.resize(input.cols, input.rows);
140 existing.updatedAt = new Date().toISOString();151 existing.updatedAt = new Date().toISOString();
141 }152 }
142 153
143 return this.snapshot(existing);154 return this.snapshot(existing);
155 });
144 }156 }
145 157
146 async write(raw: TerminalWriteInput): Promise<void> {158 async write(raw: TerminalWriteInput): Promise<void> {
@@ -166,61 +178,67 @@ export class TerminalManager extends EventEmitter<TerminalManagerEvents> { · 19 unmodified lines
166 178
167 async clear(raw: TerminalThreadInput): Promise<void> {179 async clear(raw: TerminalThreadInput): Promise<void> {
168 const input = terminalThreadInputSchema.parse(raw);180 const input = terminalThreadInputSchema.parse(raw);
169 const session = this.requireSession(input.threadId);181 await this.runWithThreadLock(input.threadId, async () => {
170 session.history = "";182 const session = this.requireSession(input.threadId);
171 session.updatedAt = new Date().toISOString();183 session.history = "";
172 await this.persistHistory(input.threadId, session.history);184 session.updatedAt = new Date().toISOString();
173 this.emitEvent({185 await this.persistHistory(input.threadId, session.history);
174 type: "cleared",186 this.emitEvent({
175 threadId: input.threadId,187 type: "cleared",
176 createdAt: new Date().toISOString(),188 threadId: input.threadId,
189 createdAt: new Date().toISOString(),
190 });
177 });191 });
178 }192 }
179 193
180 async restart(raw: TerminalOpenInput): Promise<TerminalSessionSnapshot> {194 async restart(raw: TerminalOpenInput): Promise<TerminalSessionSnapshot> {
181 const input = terminalOpenInputSchema.parse(raw);195 const input = terminalOpenInputSchema.parse(raw);
182 await this.assertValidCwd(input.cwd);196 return this.runWithThreadLock(input.threadId, async () => {
183 197 await this.assertValidCwd(input.cwd);
184 let session = this.sessions.get(input.threadId);198
185 if (!session) {199 let session = this.sessions.get(input.threadId);
186 session = {200 if (!session) {
187 threadId: input.threadId,201 session = {
188 cwd: input.cwd,202 threadId: input.threadId,
189 status: "starting",203 cwd: input.cwd,
190 pid: null,204 status: "starting",
191 history: "",205 pid: null,
192 exitCode: null,206 history: "",
193 exitSignal: null,207 exitCode: null,
194 updatedAt: new Date().toISOString(),208 exitSignal: null,
195 cols: input.cols,209 updatedAt: new Date().toISOString(),
196 rows: input.rows,210 cols: input.cols,
197 process: null,211 rows: input.rows,
198 unsubscribeData: null,212 process: null,
199 unsubscribeExit: null,213 unsubscribeData: null,
200 };214 unsubscribeExit: null,
201 this.sessions.set(input.threadId, session);215 };
202 } else {216 this.sessions.set(input.threadId, session);
203 this.stopProcess(session);217 } else {
204 session.cwd = input.cwd;218 this.stopProcess(session);
205 }219 session.cwd = input.cwd;
220 }
206 221
207 session.history = "";222 session.history = "";
208 await this.persistHistory(input.threadId, session.history);223 await this.persistHistory(input.threadId, session.history);
209 this.startSession(session, input, "restarted");224 this.startSession(session, input, "restarted");
210 return this.snapshot(session);225 return this.snapshot(session);
226 });
211 }227 }
212 228
213 async close(raw: TerminalCloseInput): Promise<void> {229 async close(raw: TerminalCloseInput): Promise<void> {
214 const input = terminalCloseInputSchema.parse(raw);230 const input = terminalCloseInputSchema.parse(raw);
215 await this.flushPersistQueue(input.threadId);231 await this.runWithThreadLock(input.threadId, async () => {
216 const session = this.sessions.get(input.threadId);232 const session = this.sessions.get(input.threadId);
217 if (session) {233 if (session) {
218 this.stopProcess(session);234 this.stopProcess(session);
219 this.sessions.delete(input.threadId);235 this.sessions.delete(input.threadId);
220 }236 }
221 if (input.deleteHistory) {237 await this.flushPersistQueue(input.threadId);
222 await this.deleteHistory(input.threadId);238 if (input.deleteHistory) {
223 }239 await this.deleteHistory(input.threadId);
240 }
241 });
224 }242 }
225 243
226 dispose(): void {244 dispose(): void {
@@ -228,6 +246,12 @@ export class TerminalManager extends EventEmitter<TerminalManagerEvents> { · 1 unmodified line
228 this.stopProcess(session);246 this.stopProcess(session);
229 }247 }
230 this.sessions.clear();248 this.sessions.clear();
249 for (const timer of this.persistTimers.values()) {
250 clearTimeout(timer);
251 }
252 this.persistTimers.clear();
253 this.pendingPersistHistory.clear();
254 this.threadLocks.clear();
231 this.persistQueues.clear();255 this.persistQueues.clear();
232 }256 }
233 257
@@ -247,8 +271,9 @@ export class TerminalManager extends EventEmitter<TerminalManagerEvents> { · 13 unmodified lines
247 session.updatedAt = new Date().toISOString();271 session.updatedAt = new Date().toISOString();
248 272
249 const shell = this.shellResolver();273 const shell = this.shellResolver();
274 let ptyProcess: PtyProcess | null = null;
250 try {275 try {
251 const ptyProcess = this.ptyAdapter.spawn({276 ptyProcess = this.ptyAdapter.spawn({
252 shell,277 shell,
253 cwd: session.cwd,278 cwd: session.cwd,
254 cols: session.cols,279 cols: session.cols,
@@ -272,6 +297,13 @@ export class TerminalManager extends EventEmitter<TerminalManagerEvents> { · 17 unmodified lines
272 snapshot: this.snapshot(session),297 snapshot: this.snapshot(session),
273 });298 });
274 } catch (error) {299 } catch (error) {
300 if (ptyProcess) {
301 try {
302 ptyProcess.kill();
303 } catch {
304 // Ignore kill errors during failed startup cleanup.
305 }
306 }
275 session.status = "error";307 session.status = "error";
276 session.pid = null;308 session.pid = null;
277 session.process = null;309 session.process = null;
@@ -346,31 +378,60 @@ export class TerminalManager extends EventEmitter<TerminalManagerEvents> { · 68 unmodified lines
346 }378 }
347 379
348 private queuePersist(threadId: string, history: string): void {380 private queuePersist(threadId: string, history: string): void {
349 const task = async () => {381 this.pendingPersistHistory.set(threadId, history);
350 await fs.promises.writeFile(this.historyPath(threadId), history, "utf8");382 this.schedulePersist(threadId);
351 };
352 const previous = this.persistQueues.get(threadId) ?? Promise.resolve();
353 const next = previous.catch(() => undefined).then(task);
354 this.persistQueues.set(threadId, next);
355 void next.finally(() => {
356 if (this.persistQueues.get(threadId) === next) {
357 this.persistQueues.delete(threadId);
358 }
359 });
360 }383 }
361 384
362 private async persistHistory(threadId: string, history: string): Promise<void> {385 private async persistHistory(threadId: string, history: string): Promise<void> {
386 this.clearPersistTimer(threadId);
387 this.pendingPersistHistory.delete(threadId);
388 await this.enqueuePersistWrite(threadId, history);
389 }
390
391 private enqueuePersistWrite(threadId: string, history: string): Promise<void> {
363 const task = async () => {392 const task = async () => {
364 await fs.promises.writeFile(this.historyPath(threadId), history, "utf8");393 await fs.promises.writeFile(this.historyPath(threadId), history, "utf8");
365 };394 };
366 const previous = this.persistQueues.get(threadId) ?? Promise.resolve();395 const previous = this.persistQueues.get(threadId) ?? Promise.resolve();
367 const next = previous.catch(() => undefined).then(task);396 const next = previous
397 .catch(() => undefined)
398 .then(task)
399 .catch((error) => {
400 this.logger.warn("failed to persist terminal history", {
401 threadId,
402 error: error instanceof Error ? error.message : String(error),
403 });
404 });
368 this.persistQueues.set(threadId, next);405 this.persistQueues.set(threadId, next);
369 await next.finally(() => {406 const finalized = next.finally(() => {
370 if (this.persistQueues.get(threadId) === next) {407 if (this.persistQueues.get(threadId) === next) {
371 this.persistQueues.delete(threadId);408 this.persistQueues.delete(threadId);
372 }409 }
410 if (this.pendingPersistHistory.has(threadId) && !this.persistTimers.has(threadId)) {
411 this.schedulePersist(threadId);
412 }
373 });413 });
414 void finalized.catch(() => undefined);
415 return finalized;
416 }
417
418 private schedulePersist(threadId: string): void {
419 if (this.persistTimers.has(threadId)) return;
420 const timer = setTimeout(() => {
421 this.persistTimers.delete(threadId);
422 const pendingHistory = this.pendingPersistHistory.get(threadId);
423 if (pendingHistory === undefined) return;
424 this.pendingPersistHistory.delete(threadId);
425 void this.enqueuePersistWrite(threadId, pendingHistory);
426 }, this.persistDebounceMs);
427 this.persistTimers.set(threadId, timer);
428 }
429
430 private clearPersistTimer(threadId: string): void {
431 const timer = this.persistTimers.get(threadId);
432 if (!timer) return;
433 clearTimeout(timer);
434 this.persistTimers.delete(threadId);
374 }435 }
375 436
376 private async readHistory(threadId: string): Promise<string> {437 private async readHistory(threadId: string): Promise<string> {
@@ -381,6 +442,19 @@ export class TerminalManager extends EventEmitter<TerminalManagerEvents> { · 4 unmodified lines
381 await fs.promises.writeFile(this.historyPath(threadId), capped, "utf8");442 await fs.promises.writeFile(this.historyPath(threadId), capped, "utf8");
382 }443 }
383 return capped;444 return capped;
445 } catch (error) {
446 if ((error as NodeJS.ErrnoException).code !== "ENOENT") {
447 throw error;
448 }
449 }
450
451 try {
452 const raw = await fs.promises.readFile(this.legacyHistoryPath(threadId), "utf8");
453 const capped = capHistory(raw, this.historyLineLimit);
454 if (capped !== raw) {
455 await fs.promises.writeFile(this.legacyHistoryPath(threadId), capped, "utf8");
456 }
457 return capped;
384 } catch (error) {458 } catch (error) {
385 if ((error as NodeJS.ErrnoException).code === "ENOENT") {459 if ((error as NodeJS.ErrnoException).code === "ENOENT") {
386 return "";460 return "";
@@ -391,7 +465,10 @@ export class TerminalManager extends EventEmitter<TerminalManagerEvents> { · 4 unmodified lines
391 465
392 private async deleteHistory(threadId: string): Promise<void> {466 private async deleteHistory(threadId: string): Promise<void> {
393 try {467 try {
394 await fs.promises.rm(this.historyPath(threadId), { force: true });468 await Promise.all([
469 fs.promises.rm(this.historyPath(threadId), { force: true }),
470 fs.promises.rm(this.legacyHistoryPath(threadId), { force: true }),
471 ]);
395 } catch (error) {472 } catch (error) {
396 this.logger.warn("failed to delete terminal history", {473 this.logger.warn("failed to delete terminal history", {
397 threadId,474 threadId,
@@ -401,9 +478,21 @@ export class TerminalManager extends EventEmitter<TerminalManagerEvents> { · 3 unmodified lines
401 }478 }
402 479
403 private async flushPersistQueue(threadId: string): Promise<void> {480 private async flushPersistQueue(threadId: string): Promise<void> {
404 const pending = this.persistQueues.get(threadId);481 this.clearPersistTimer(threadId);
405 if (!pending) return;482
406 await pending.catch(() => undefined);483 while (true) {
484 const pendingHistory = this.pendingPersistHistory.get(threadId);
485 if (pendingHistory !== undefined) {
486 this.pendingPersistHistory.delete(threadId);
487 await this.enqueuePersistWrite(threadId, pendingHistory);
488 }
489
490 const pending = this.persistQueues.get(threadId);
491 if (!pending) {
492 return;
493 }
494 await pending.catch(() => undefined);
495 }
407 }496 }
408 497
409 private async assertValidCwd(cwd: string): Promise<void> {498 private async assertValidCwd(cwd: string): Promise<void> {
@@ -449,4 +538,29 @@ export class TerminalManager extends EventEmitter<TerminalManagerEvents> { · 39 unmodified lines
449 private historyPath(threadId: string): string {538 private historyPath(threadId: string): string {
450 return path.join(this.logsDir, `${toSafeThreadId(threadId)}.log`);539 return path.join(this.logsDir, `${toSafeThreadId(threadId)}.log`);
451 }540 }
541
542 private legacyHistoryPath(threadId: string): string {
543 return path.join(this.logsDir, `${legacySafeThreadId(threadId)}.log`);
544 }
545
546 private async runWithThreadLock<T>(
547 threadId: string,
548 task: () => Promise<T>,
549 ): Promise<T> {
550 const previous = this.threadLocks.get(threadId) ?? Promise.resolve();
551 let release: () => void = () => {};
552 const current = new Promise<void>((resolve) => {
553 release = resolve;
554 });
555 this.threadLocks.set(threadId, current);
556 await previous.catch(() => undefined);
557 try {
558 return await task();
559 } finally {
560 release();
561 if (this.threadLocks.get(threadId) === current) {
562 this.threadLocks.delete(threadId);
563 }
564 }
565 }
452}566}
apps/server/src/wsServer.test.ts+16−0
@@ -382,6 +382,22 @@ describe("WebSocket Server", () => { · 381 unmodified lines
382 expect((push.data as TerminalEvent).type).toBe("output");382 expect((push.data as TerminalEvent).type).toBe("output");
383 });383 });
384 384
385 it("detaches terminal event listener on stop for injected manager", async () => {
386 const terminalManager = new MockTerminalManager();
387 server = createTestServer({
388 cwd: "/test",
389 terminalManager: terminalManager as unknown as TerminalManager,
390 });
391 await server.start();
392
393 expect(terminalManager.listenerCount("event")).toBe(1);
394
395 await server.stop();
396 server = null;
397
398 expect(terminalManager.listenerCount("event")).toBe(0);
399 });
400
385 it("returns validation errors for invalid terminal open params", async () => {401 it("returns validation errors for invalid terminal open params", async () => {
386 server = createTestServer({ cwd: "/test" });402 server = createTestServer({ cwd: "/test" });
387 await server.start();403 await server.start();
apps/server/src/wsServer.ts+5−2
@@ -9,6 +9,7 @@ import { · 8 unmodified lines
9 EDITORS,9 EDITORS,
10 WS_CHANNELS,10 WS_CHANNELS,
11 WS_METHODS,11 WS_METHODS,
12 type TerminalEvent,
12 type WsPush,13 type WsPush,
13 type WsRequest,14 type WsRequest,
14 type WsResponse,15 type WsResponse,
@@ -112,7 +113,7 @@ export function createServer(options: ServerOptions) { · 97 unmodified lines
112 logOutgoingPush(push, recipients);113 logOutgoingPush(push, recipients);
113 });114 });
114 115
115 terminalManager.on("event", (event) => {116 const onTerminalEvent = (event: TerminalEvent) => {
116 const push: WsPush = {117 const push: WsPush = {
117 type: "push",118 type: "push",
118 channel: WS_CHANNELS.terminalEvent,119 channel: WS_CHANNELS.terminalEvent,
@@ -127,7 +128,8 @@ export function createServer(options: ServerOptions) { · 8 unmodified lines
127 }128 }
128 }129 }
129 logOutgoingPush(push, recipients);130 logOutgoingPush(push, recipients);
130 });131 };
132 terminalManager.on("event", onTerminalEvent);
131 133
132 // HTTP server — serves static files or redirects to Vite dev server134 // HTTP server — serves static files or redirects to Vite dev server
133 const httpServer = http.createServer((req, res) => {135 const httpServer = http.createServer((req, res) => {
@@ -401,6 +403,7 @@ export function createServer(options: ServerOptions) { · 267 unmodified lines
401 }403 }
402 404
403 async function stop(): Promise<void> {405 async function stop(): Promise<void> {
406 terminalManager.off("event", onTerminalEvent);
404 providerManager.stopAll();407 providerManager.stopAll();
405 providerManager.dispose();408 providerManager.dispose();
406 terminalManager.dispose();409 terminalManager.dispose();
apps/web/src/components/ChatView.tsx+0−1
@@ -1207,7 +1207,6 @@ export default function ChatView() { · 1206 unmodified lines
1207 </form>1207 </form>
1208 </div>1208 </div>
1209 1209
1210 <BranchToolbar envMode={envMode} onEnvModeChange={setEnvMode} envLocked={envLocked} />
1211 <BranchToolbar envMode={envMode} onEnvModeChange={setEnvMode} envLocked={envLocked} />1210 <BranchToolbar envMode={envMode} onEnvModeChange={setEnvMode} envLocked={envLocked} />
1212 1211
1213 {activeThread.terminalOpen && api && activeProject && (1212 {activeThread.terminalOpen && api && activeProject && (
apps/web/src/components/ThreadTerminalDrawer.tsx+22−21
@@ -152,6 +152,20 @@ export default function ThreadTerminalDrawer({ · 151 unmodified lines
152 [],152 [],
153 );153 );
154 154
155 const fitAndResizeTerminal = useCallback(() => {
156 const activeTerminal = terminalRef.current;
157 const activeFitAddon = fitAddonRef.current;
158 if (!activeTerminal || !activeFitAddon) return;
159 activeFitAddon.fit();
160 void api.terminal
161 .resize({
162 threadId,
163 cols: activeTerminal.cols,
164 rows: activeTerminal.rows,
165 })
166 .catch(() => undefined);
167 }, [api, threadId]);
168
155 useEffect(() => {169 useEffect(() => {
156 const clampedHeight = clampDrawerHeight(height);170 const clampedHeight = clampDrawerHeight(height);
157 setDrawerHeight(clampedHeight);171 setDrawerHeight(clampedHeight);
@@ -201,8 +215,11 @@ export default function ThreadTerminalDrawer({ · 43 unmodified lines
201 useEffect(() => {215 useEffect(() => {
202 const onWindowResize = () => {216 const onWindowResize = () => {
203 const clampedHeight = clampDrawerHeight(drawerHeightRef.current);217 const clampedHeight = clampDrawerHeight(drawerHeightRef.current);
204 if (clampedHeight !== drawerHeightRef.current) {218 const changed = clampedHeight !== drawerHeightRef.current;
219 if (changed) {
205 setDrawerHeight(clampedHeight);220 setDrawerHeight(clampedHeight);
221 } else {
222 fitAndResizeTerminal();
206 }223 }
207 if (!resizeStateRef.current) {224 if (!resizeStateRef.current) {
208 syncHeight(clampedHeight);225 syncHeight(clampedHeight);
@@ -212,7 +229,7 @@ export default function ThreadTerminalDrawer({ · 3 unmodified lines
212 return () => {229 return () => {
213 window.removeEventListener("resize", onWindowResize);230 window.removeEventListener("resize", onWindowResize);
214 };231 };
215 }, [syncHeight]);232 }, [fitAndResizeTerminal, syncHeight]);
216 233
217 useEffect(() => {234 useEffect(() => {
218 return () => {235 return () => {
@@ -264,20 +281,6 @@ export default function ThreadTerminalDrawer({ · 45 unmodified lines
264 return false;281 return false;
265 });282 });
266 283
267 const resizeAndSync = () => {
268 const activeTerminal = terminalRef.current;
269 const activeFitAddon = fitAddonRef.current;
270 if (!activeTerminal || !activeFitAddon) return;
271 activeFitAddon.fit();
272 void api.terminal
273 .resize({
274 threadId,
275 cols: activeTerminal.cols,
276 rows: activeTerminal.rows,
277 })
278 .catch(() => undefined);
279 };
280
281 const inputDisposable = terminal.onData((data) => {284 const inputDisposable = terminal.onData((data) => {
282 void api.terminal285 void api.terminal
283 .write({ threadId, data })286 .write({ threadId, data })
@@ -363,14 +366,12 @@ export default function ThreadTerminalDrawer({ · 79 unmodified lines
363 }366 }
364 });367 });
365 368
366 window.addEventListener("resize", resizeAndSync);369 const fitTimer = window.setTimeout(fitAndResizeTerminal, 30);
367 const fitTimer = window.setTimeout(resizeAndSync, 30);
368 void openTerminal();370 void openTerminal();
369 371
370 return () => {372 return () => {
371 disposed = true;373 disposed = true;
372 window.clearTimeout(fitTimer);374 window.clearTimeout(fitTimer);
373 window.removeEventListener("resize", resizeAndSync);
374 unsubscribe();375 unsubscribe();
375 inputDisposable.dispose();376 inputDisposable.dispose();
376 themeObserver.disconnect();377 themeObserver.disconnect();
@@ -378,7 +379,7 @@ export default function ThreadTerminalDrawer({ · 1 unmodified line
378 fitAddonRef.current = null;379 fitAddonRef.current = null;
379 terminal.dispose();380 terminal.dispose();
380 };381 };
381 }, [api, cwd, threadId]);382 }, [api, cwd, fitAndResizeTerminal, threadId]);
382 383
383 useEffect(() => {384 useEffect(() => {
384 const terminal = terminalRef.current;385 const terminal = terminalRef.current;
@@ -408,7 +409,7 @@ export default function ThreadTerminalDrawer({ · 23 unmodified lines
408 return () => {409 return () => {
409 window.cancelAnimationFrame(frame);410 window.cancelAnimationFrame(frame);
410 };411 };
411 }, [api, drawerHeight, threadId]);412 }, [api, cwd, drawerHeight, threadId]);
412 413
413 return (414 return (
414 <aside415 <aside

Keyboard shortcuts

General

Show keyboard shortcuts
?
Focus the current search
/

Issues and pull requests

Open filters
f

Issues, pull requests and Inbox

Select next item
j↓
Select previous item
k↑
Open selected item
Enter
Dismiss filters or clear selection
Esc

Repository code

Go to file
t