Skip to content

fix: wake exec waits on completion and buffered output (MEMMY-675) #618

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
origami-chen merged 2 commits into from
Oct 15, 2026
Merged

fix: wake exec waits on completion and buffered output (MEMMY-675) #618

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
27e0418
fix: wake exec waits on process completion and buffered output
yxyi123 Oct 12, 2026
e48fe20
Merge remote-tracking branch 'upstream/v1.2.3' into fix/memmy-675-exe…
yxyi123 Oct 12, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
103 changes: 82 additions & 21 deletions App/memmy-agent/src/core/agent-runtime/tools/exec-session.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,8 @@ export const DEFAULT_MAX_OUTPUT_CHARS = 10_000;
export const MAX_OUTPUT_CHARS = 50_000;
const STDIO_DRAIN_GRACE_MS = 1000;
const PROCESS_TERMINATE_GRACE_MS = 1000;
// Batch progress output without turning every stream chunk into a tool return.
const OUTPUT_COALESCE_MS = 1000;
export const EXEC_LIFECYCLE_ERROR =
"Error: Command process exited but descendants kept stdout/stderr open; the managed process tree was terminated.";

Expand Down Expand Up @@ -41,10 +43,6 @@ export type ExecSessionInfo = {
returncode: number | null;
};

function sleep(ms: number): Promise<void> {
return new Promise((resolve) => setTimeout(resolve, ms));
}

function processExited(child: ChildProcess): boolean {
return child.exitCode != null || child.signalCode != null;
}
Expand Down Expand Up @@ -169,18 +167,17 @@ export async function drainExitedProcessStreams({
streamsClosed,
forceCloseStreams,
options,
abortSignal,
}: {
child: ChildProcess;
closePromise: Promise<void>;
streamsClosed: () => boolean;
forceCloseStreams: () => void;
options?: ManagedProcessOptions;
abortSignal?: AbortSignal | null;
}): Promise<string | null> {
if (streamsClosed()) return null;
const closedDuringDrain = await Promise.race([
closePromise.then(() => true),
sleep(STDIO_DRAIN_GRACE_MS).then(() => streamsClosed()),
]);
const closedDuringDrain = await waitForDrain(closePromise, streamsClosed, abortSignal);
if (closedDuringDrain || streamsClosed()) return null;
await terminateManagedProcessTree(child, options);
if (!streamsClosed()) forceCloseStreams();
Expand All @@ -201,20 +198,27 @@ function throwIfAborted(signal?: AbortSignal | null): void {
if (signal?.aborted) throw createToolAbortError();
}

function abortableSleep(ms: number, signal?: AbortSignal | null): Promise<void> {
if (ms <= 0) return Promise.resolve();
function waitForDrain(
closePromise: Promise<void>,
streamsClosed: () => boolean,
signal?: AbortSignal | null,
): Promise<boolean> {
throwIfAborted(signal);
if (!signal) return sleep(ms);
return new Promise((resolve, reject) => {
const timer = setTimeout(() => {
signal.removeEventListener("abort", onAbort);
resolve();
}, ms);
const onAbort = () => {
let settled = false;
const finish = (closed: boolean, error?: Error) => {
if (settled) return;
settled = true;
clearTimeout(timer);
reject(createToolAbortError());
signal?.removeEventListener("abort", onAbort);
if (error) reject(error);
else resolve(closed);
};
signal.addEventListener("abort", onAbort, { once: true });
const onAbort = () => finish(false, createToolAbortError());
const timer = setTimeout(() => finish(streamsClosed()), STDIO_DRAIN_GRACE_MS);
signal?.addEventListener("abort", onAbort, { once: true });
void closePromise.then(() => finish(true));
if (signal?.aborted) onAbort();
});
}

Expand All @@ -241,6 +245,7 @@ export class ExecSession {
deadlineMs: number;
lastAccess = Date.now();
private chunks: string[] = [];
private outputStartedAt: number | null = null;
private timedOut = false;
private stderrStarted = false;
private readonly stdoutDecoder: CommandOutputDecoder;
Expand All @@ -251,6 +256,7 @@ export class ExecSession {
private lifecycleError: string | null = null;
private readonly closePromise: Promise<void>;
private resolveClose!: () => void;
private readonly outputWaiters = new Set<() => void>();

constructor({
command,
Expand Down Expand Up @@ -296,7 +302,7 @@ export class ExecSession {
args.push("-c", command);
this.process = spawnProcess(resolvedShell, args, { cwd, env, detached: true });
}
this.process.on("error", (error) => this.chunks.push(`Error executing command: ${error.message}`));
this.process.on("error", (error) => this.appendStdout(`Error executing command: ${error.message}`));
this.process.stdout.on("data", (chunk: Buffer) => {
this.stdoutDecoder.push(chunk);
this.appendStdout(this.stdoutDecoder.read());
Expand Down Expand Up @@ -348,7 +354,7 @@ export class ExecSession {
): Promise<SessionPoll> {
this.lastAccess = Date.now();
throwIfAborted(abortSignal);
if (yieldTimeMs > 0 && !this.isDone()) await abortableSleep(Math.min(yieldTimeMs, MAX_YIELD_MS), abortSignal);
if (yieldTimeMs > 0) await this.waitForActivity(Math.min(yieldTimeMs, MAX_YIELD_MS), abortSignal);
throwIfAborted(abortSignal);
if (!this.isDone() && Date.now() >= this.deadlineMs) {
this.timedOut = true;
Expand All @@ -360,11 +366,13 @@ export class ExecSession {
closePromise: this.closePromise,
streamsClosed: () => this.streamsClosed,
forceCloseStreams: () => this.forceCloseStreams(),
abortSignal,
});
}
throwIfAborted(abortSignal);
const raw = this.chunks.join("");
this.chunks = [];
this.outputStartedAt = null;
const [output, truncated] = truncateOutput(raw, maxOutputChars);
const elapsed = Math.max(0, (Date.now() - this.startedAt) / 1000);
const done = this.streamsClosed;
Expand All @@ -381,14 +389,66 @@ export class ExecSession {
};
}

private waitForActivity(ms: number, signal?: AbortSignal | null): Promise<void> {
throwIfAborted(signal);
if (this.isDone() || this.streamsClosed) return Promise.resolve();
return new Promise((resolve, reject) => {
let settled = false;
let outputTimer: ReturnType<typeof setTimeout> | undefined;
const finish = (error?: Error) => {
if (settled) return;
settled = true;
clearTimeout(waitTimer);
clearTimeout(outputTimer);
this.outputWaiters.delete(onOutput);
this.process.removeListener("exit", onDone);
this.process.removeListener("close", onDone);
signal?.removeEventListener("abort", onAbort);
if (error) reject(error);
else resolve();
};
const onDone = () => finish();
const onAbort = () => finish(createToolAbortError());
const onOutput = () => {
if (this.isDone() || this.streamsClosed) {
finish();
} else if (this.chunks.length > 0 && outputTimer === undefined) {
// Anchor at the first readable output, never debounce continuous logs.
const remaining = Math.max(0, (this.outputStartedAt ?? Date.now()) + OUTPUT_COALESCE_MS - Date.now());
if (remaining === 0) finish();
else outputTimer = setTimeout(onDone, remaining);
}
};
const waitMs = Math.max(0, Math.min(ms, this.deadlineMs - Date.now()));
const waitTimer = setTimeout(onDone, waitMs);
this.process.once("exit", onDone);
this.process.once("close", onDone);
this.outputWaiters.add(onOutput);
signal?.addEventListener("abort", onAbort, { once: true });
// Recheck after subscribing so completion, cancellation and buffered
// output cannot fall between checking state and installing listeners.
if (signal?.aborted) onAbort();
else onOutput();
});
}

private notifyOutputWaiters(): void {
for (const wake of this.outputWaiters) wake();
}

private appendStdout(text: string): void {
if (text) this.chunks.push(text);
if (!text) return;
this.chunks.push(text);
this.outputStartedAt ??= Date.now();
this.notifyOutputWaiters();
}

private appendStderr(text: string): void {
if (!text) return;
this.chunks.push(this.stderrStarted ? text : `STDERR:\n${text}`);
this.outputStartedAt ??= Date.now();
this.stderrStarted = true;
this.notifyOutputWaiters();
}

private finishStdoutOnce(): void {
Expand All @@ -412,6 +472,7 @@ export class ExecSession {
if (this.streamsClosed) return;
this.streamsClosed = true;
this.resolveClose();
this.notifyOutputWaiters();
}
}

Expand Down
Loading
Loading