LiuMangMang/agent-workflow
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: harden executor session handling

Browse Source
This commit is contained in:
liujing
2026-07-16 23:00:48 +08:00
parent 35e8fa3717
commit 3566a73a03
5 changed files with 207 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/workflow-engine.ts
+10 -3
View File
@@ -326,6 +326,7 @@ export async function startWorkflow(opts: StartWorkflowOpts): Promise<{ runId: s
runDir: dir,
cycleIndex,
config: config.executorConfig,
timeoutSeconds: state.timeoutSeconds,
signal: ac.signal,
});
} catch (err) {
@@ -338,8 +339,8 @@ export async function startWorkflow(opts: StartWorkflowOpts): Promise<{ runId: s
persistExecutorAttempt(dir, cycleRecord, execResult);
if (execResult.sessionHandle) {
state.sessionHandle = execResult.sessionHandle;
if (execResult.sessionHandle.kind === "agy" && execResult.sessionHandle.degraded) {
state.agyDegraded = true;
if (execResult.sessionHandle.kind === "agy") {
state.agyDegraded = execResult.sessionHandle.degraded;
}
}
@@ -725,6 +726,7 @@ async function resumeExecutorCycle(
runDir: dir,
cycleIndex,
config: state.executorConfig ?? {},
timeoutSeconds: state.timeoutSeconds,
signal: ac.signal,
});
} catch (err) {
@@ -741,7 +743,12 @@ async function resumeExecutorCycle(
cleanup();
persistExecutorAttempt(dir, cycleRecord, execResult);
if (execResult.sessionHandle) state.sessionHandle = execResult.sessionHandle;
if (execResult.sessionHandle) {
state.sessionHandle = execResult.sessionHandle;
if (execResult.sessionHandle.kind === "agy") {
state.agyDegraded = execResult.sessionHandle.degraded;
}
}
appendEvent(dir, {
kind: "executor_completed",
src/workflow-executor.test.ts
+94 -8
View File
@@ -55,7 +55,7 @@ function writeFakeCli(dir: string, opts: {
const stdout = opts.stdout ?? "";
const script = `#!/bin/bash
echo "$@" >> "${argsFile}"
printf '%s\\0' "$@" >> "${argsFile}"
echo '${stdout.replace(/'/g, "'\\''")}'
exit ${exitCode}
`;
@@ -224,11 +224,20 @@ describe("ClaudeAdapter", () => {
describe("AgyAdapter", () => {
let tmpDir: string;
let previousHome: string | undefined;
beforeEach(() => { tmpDir = makeTmpDir(); });
afterEach(() => { cleanupDir(tmpDir); });
beforeEach(() => {
tmpDir = makeTmpDir();
previousHome = process.env.HOME;
process.env.HOME = tmpDir;
});
afterEach(() => {
if (previousHome === undefined) delete process.env.HOME;
else process.env.HOME = previousHome;
cleanupDir(tmpDir);
});
it("start uses --print and --mode accept-edits", async () => {
it("start passes the complete prompt as the --print value and enables headless tools", async () => {
const argsFile = path.join(tmpDir, "args.txt");
const fakeBin = writeFakeCli(tmpDir, {
argsFile,
@@ -242,11 +251,19 @@ describe("AgyAdapter", () => {
runDir: tmpDir,
cycleIndex: 1,
config: { binary: fakeBin },
timeoutSeconds: 1800,
});
const capturedArgs = fs.existsSync(argsFile) ? fs.readFileSync(argsFile, "utf8") : "";
expect(capturedArgs).toContain("--print");
expect(capturedArgs).toContain("--mode accept-edits");
const capturedArgs = fs.readFileSync(argsFile, "utf8").split("\0").filter(Boolean);
const printIndex = capturedArgs.indexOf("--print");
expect(capturedArgs).toContain("--dangerously-skip-permissions");
expect(capturedArgs).toContain("--mode");
expect(capturedArgs[capturedArgs.indexOf("--mode") + 1]).toBe("accept-edits");
expect(capturedArgs).toContain("--print-timeout");
expect(capturedArgs[capturedArgs.indexOf("--print-timeout") + 1]).toBe("1800s");
expect(capturedArgs).not.toContain("--");
expect(printIndex).toBe(capturedArgs.length - 2);
expect(capturedArgs[printIndex + 1]).toContain("# Implementation Task: Test plan");
if (result.sessionHandle?.kind === "agy") {
expect(result.sessionHandle.conversationId).toBe("agy-conv-id-456");
@@ -285,11 +302,15 @@ describe("AgyAdapter", () => {
runDir: tmpDir,
cycleIndex: 2,
config: { binary: fakeBin },
timeoutSeconds: 900,
});
const capturedArgs = fs.existsSync(argsFile) ? fs.readFileSync(argsFile, "utf8") : "";
const capturedArgs = fs.readFileSync(argsFile, "utf8").split("\0").filter(Boolean);
expect(capturedArgs).toContain("--conversation");
expect(capturedArgs).toContain("known-conv-id");
expect(capturedArgs).toContain("--dangerously-skip-permissions");
expect(capturedArgs[capturedArgs.indexOf("--print-timeout") + 1]).toBe("900s");
expect(capturedArgs[capturedArgs.indexOf("--print") + 1]).toContain("# Fix Request: Test plan");
});
it("degraded resume uses --continue flag", async () => {
@@ -310,6 +331,71 @@ describe("AgyAdapter", () => {
const capturedArgs = fs.existsSync(argsFile) ? fs.readFileSync(argsFile, "utf8") : "";
expect(capturedArgs).toContain("--continue");
});
it("captures the conversation ID from the Agy log", async () => {
const fakeBin = path.join(tmpDir, "fake-agy.sh");
fs.writeFileSync(fakeBin, `#!/bin/bash
while [[ $# -gt 0 ]]; do
if [[ "$1" == "--log-file" ]]; then
shift
echo 'Print mode: conversation=12345678-1234-1234-1234-123456789abc, sending message' > "$1"
fi
shift
done
echo 'implemented'
`, { mode: 0o755 });
const result = await createExecutorAdapter("agy").start({
repoRoot: tmpDir,
plan: makeValidPlan(),
runDir: tmpDir,
cycleIndex: 1,
config: { binary: fakeBin },
});
expect(result.sessionHandle).toEqual({
kind: "agy",
conversationId: "12345678-1234-1234-1234-123456789abc",
degraded: false,
});
});
it("does not reuse a workspace conversation that predates start", async () => {
const cacheDir = path.join(tmpDir, ".gemini", "antigravity-cli", "cache");
fs.mkdirSync(cacheDir, { recursive: true });
fs.writeFileSync(path.join(cacheDir, "last_conversations.json"), JSON.stringify({
[tmpDir]: "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa",
}));
const fakeBin = writeFakeCli(tmpDir, { stdout: "implemented" });
const result = await createExecutorAdapter("agy").start({
repoRoot: tmpDir,
plan: makeValidPlan(),
runDir: tmpDir,
cycleIndex: 1,
config: { binary: fakeBin },
});
expect(result.sessionHandle).toEqual({ kind: "agy", degraded: true });
});
it("treats headless permission denial as failure even with exit code zero", async () => {
const fakeBin = writeFakeCli(tmpDir, {
stdout: 'jetski: no output produced — a tool required the "command" permission that headless mode cannot prompt for, so it was auto-denied.',
});
const result = await createExecutorAdapter("agy").start({
repoRoot: tmpDir,
plan: makeValidPlan(),
runDir: tmpDir,
cycleIndex: 1,
config: { binary: fakeBin },
});
expect(result.exitCode).toBe(0);
expect(result.failed).toBe(true);
expect(result.failureReason).toContain("no output produced");
});
});
// ---------------------------------------------------------------------------
src/workflow-executor.ts
+92 -17
View File
@@ -516,6 +516,59 @@ function extractAgyConversationId(output: string): string | undefined {
return undefined;
}
function extractAgyConversationIdFromLog(logFile: string): string | undefined {
try {
const log = fs.readFileSync(logFile, "utf8");
const matches = [
...log.matchAll(/Print mode: conversation=([0-9a-f-]{36})\b/gi),
...log.matchAll(/Created conversation ([0-9a-f-]{36})\b/gi),
];
return matches.at(-1)?.[1];
} catch {
return undefined;
}
}
function agyConversationCacheFile(): string {
const homeDir = process.env.HOME || os.homedir();
return path.join(homeDir, ".gemini", "antigravity-cli", "cache", "last_conversations.json");
}
function getAgyWorkspaceConversationId(repoRoot: string): string | undefined {
try {
const parsed = JSON.parse(fs.readFileSync(agyConversationCacheFile(), "utf8")) as Record<string, unknown>;
const id = parsed[repoRoot];
return typeof id === "string" && /^[0-9a-f-]{36}$/i.test(id) ? id : undefined;
} catch {
return undefined;
}
}
function extractAgyFailureReason(stdout: string, stderr: string): string | undefined {
const output = `${stdout}\n${stderr}`.trim();
if (!output) return "Agy exited without producing output.";
const diagnostic = output
.split("\n")
.map((line) => line.trim())
.find((line) =>
/no output produced/i.test(line)
|| /headless mode cannot prompt/i.test(line)
|| /permission .*auto-denied/i.test(line)
|| /tool required .* permission/i.test(line));
return diagnostic;
}
function agyCommonArgs(config: ExecutorConfig, timeoutSeconds?: number): string[] {
return [
"--dangerously-skip-permissions",
"--mode", "accept-edits",
...(timeoutSeconds ? ["--print-timeout", `${timeoutSeconds}s`] : []),
...(config.agent ? ["--agent", config.agent] : []),
...(config.model ? ["--model", config.model] : []),
];
}
class AgyAdapter implements ExecutorAdapter {
readonly kind: ExecutorKind = "agy";
@@ -534,11 +587,13 @@ class AgyAdapter implements ExecutorAdapter {
async start(opts: ExecutorStartOpts): Promise<ExecutorResult> {
const bin = resolveExecutable(opts.config, "agy");
const prompt = buildStartPrompt(opts);
const logFile = path.join(opts.runDir, `cycle-${opts.cycleIndex}-agy.log`);
const previousWorkspaceConversationId = getAgyWorkspaceConversationId(opts.repoRoot);
const args: string[] = [
...(opts.config.agent ? ["--agent", opts.config.agent] : []),
...(opts.config.model ? ["--model", opts.config.model] : []),
"--print", "--mode", "accept-edits", "--", prompt,
...agyCommonArgs(opts.config, opts.timeoutSeconds),
"--log-file", logFile,
"--print", prompt,
];
const startMs = Date.now();
@@ -550,7 +605,12 @@ class AgyAdapter implements ExecutorAdapter {
});
const durationMs = Date.now() - startMs;
const conversationId = extractAgyConversationId(child.stdout);
const conversationId = extractAgyConversationId(child.stdout)
?? extractAgyConversationIdFromLog(logFile)
?? (() => {
const currentId = getAgyWorkspaceConversationId(opts.repoRoot);
return currentId !== previousWorkspaceConversationId ? currentId : undefined;
})();
const handle: AgySessionHandle = {
kind: "agy",
...(conversationId ? { conversationId } : {}),
@@ -558,13 +618,15 @@ class AgyAdapter implements ExecutorAdapter {
};
const report = child.stdout + (child.stderr ? `\n[stderr]\n${child.stderr}` : "");
const failureReason = child.error?.message
?? extractAgyFailureReason(child.stdout, child.stderr);
return {
exitCode: child.status,
report,
durationMs,
sessionHandle: handle,
failed: child.status !== 0 || !!child.error,
...(child.error ? { failureReason: child.error.message } : {}),
failed: child.status !== 0 || !!failureReason,
...(failureReason ? { failureReason } : {}),
};
}
@@ -572,22 +634,25 @@ class AgyAdapter implements ExecutorAdapter {
const handle = opts.handle as AgySessionHandle;
const bin = resolveExecutable(opts.config, "agy");
const prompt = buildFixPrompt(opts);
const logFile = path.join(opts.runDir, `cycle-${opts.cycleIndex}-agy.log`);
const conversationId = handle.conversationId
?? getAgyWorkspaceConversationId(opts.repoRoot);
let args: string[];
if (handle.conversationId) {
if (conversationId) {
args = [
"--conversation", handle.conversationId,
...(opts.config.agent ? ["--agent", opts.config.agent] : []),
...(opts.config.model ? ["--model", opts.config.model] : []),
"--print", "--mode", "accept-edits", "--", prompt,
...agyCommonArgs(opts.config, opts.timeoutSeconds),
"--conversation", conversationId,
"--log-file", logFile,
"--print", prompt,
];
} else {
// Degraded mode: use --continue (best effort, may pick wrong session in concurrent setups)
args = [
...agyCommonArgs(opts.config, opts.timeoutSeconds),
"--continue",
...(opts.config.agent ? ["--agent", opts.config.agent] : []),
...(opts.config.model ? ["--model", opts.config.model] : []),
"--print", "--mode", "accept-edits", "--", prompt,
"--log-file", logFile,
"--print", prompt,
];
}
@@ -601,13 +666,23 @@ class AgyAdapter implements ExecutorAdapter {
const durationMs = Date.now() - startMs;
const report = child.stdout + (child.stderr ? `\n[stderr]\n${child.stderr}` : "");
const capturedConversationId = conversationId
?? extractAgyConversationId(child.stdout)
?? extractAgyConversationIdFromLog(logFile);
const updatedHandle: AgySessionHandle = {
kind: "agy",
...(capturedConversationId ? { conversationId: capturedConversationId } : {}),
degraded: !capturedConversationId,
};
const failureReason = child.error?.message
?? extractAgyFailureReason(child.stdout, child.stderr);
return {
exitCode: child.status,
report,
durationMs,
sessionHandle: handle,
failed: child.status !== 0 || !!child.error,
...(child.error ? { failureReason: child.error.message } : {}),
sessionHandle: updatedHandle,
failed: child.status !== 0 || !!failureReason,
...(failureReason ? { failureReason } : {}),
};
}
src/workflow-types.ts
+4
View File
@@ -334,6 +334,8 @@ export interface ExecutorStartOpts {
runDir: string;
cycleIndex: number;
config: ExecutorConfig;
/** Host timeout, forwarded to executors that enforce their own deadline. */
timeoutSeconds?: number;
signal?: AbortSignal;
}
@@ -346,5 +348,7 @@ export interface ExecutorResumeOpts {
runDir: string;
cycleIndex: number;
config: ExecutorConfig;
/** Host timeout, forwarded to executors that enforce their own deadline. */
timeoutSeconds?: number;
signal?: AbortSignal;
}