Merge pull request #5115 from DanielThomas/dannyt/sendUserMessage-agent-end

fix(coding-agent): drain follow-ups queued during agent_end

packages/coding-agent/docs/extensions.md

@@ -2557,6 +2557,7 @@ All examples in [examples/extensions/](../examples/extensions/).
 | `custom-compaction.ts` | Custom compaction summary | `on("session_before_compact")` |
 | `trigger-compact.ts` | Trigger compaction manually | `compact()` |
 | `git-checkpoint.ts` | Git stash on turns | `on("turn_start")`, `on("session_before_fork")`, `exec` |
+| `git-merge-and-resolve.ts` | Fetch, merge, and resolve conflicts | `on("agent_end")`, `exec`, `sendUserMessage` |
 | `auto-commit-on-exit.ts` | Commit on shutdown | `on("session_shutdown")`, `exec` |
 | **UI Components** |||
 | `status-line.ts` | Footer status indicator | `setStatus`, session events |

packages/coding-agent/examples/extensions/git-merge-and-resolve.ts

@@ -0,0 +1,115 @@
+/**
+ * Merge and Resolve
+ *
+ * Keeps the working branch up to date with its upstream tracking ref.
+ * After each agent turn, fetches and merges. Clean merges complete
+ * silently. When conflicts arise, the working tree is left dirty and
+ * the agent receives a follow-up message listing each conflict block
+ * with file, line range, and ours/theirs sections so it can resolve them.
+ * Also re-sends unresolved conflicts from a previous incomplete merge.
+ *
+ * Start pi with this extension:
+ *   pi -e ./examples/extensions/git-merge-and-resolve.ts
+ */
+import { createReadStream } from "node:fs";
+import { join } from "node:path";
+import { createInterface } from "node:readline";
+import type { ExtensionAPI } from "@earendil-works/pi-coding-agent";
+
+interface ConflictBlock {
+	file: string;
+	startLine: number;
+	separatorLine: number;
+	endLine: number;
+}
+
+/** Parse conflict markers from working tree files with unmerged paths. */
+async function findConflicts(pi: ExtensionAPI, cwd: string): Promise<ConflictBlock[]> {
+	const { stdout, code } = await pi.exec("git", ["diff", "--name-only", "--diff-filter=U"]);
+	if (code !== 0 || !stdout.trim()) return [];
+
+	const blocks: ConflictBlock[] = [];
+	for (const file of stdout.trim().split("\n")) {
+		try {
+			const rl = createInterface({ input: createReadStream(join(cwd, file), "utf-8") });
+			let lineNo = 0;
+			let blockStart: number | undefined;
+			let separatorLine: number | undefined;
+			for await (const line of rl) {
+				lineNo++;
+				if (line.startsWith("<<<<<<<")) {
+					blockStart = lineNo;
+					separatorLine = undefined;
+				} else if (line.startsWith("=======") && blockStart !== undefined) {
+					separatorLine = lineNo;
+				} else if (line.startsWith(">>>>>>>") && blockStart !== undefined && separatorLine !== undefined) {
+					blocks.push({ file, startLine: blockStart, separatorLine, endLine: lineNo });
+					blockStart = undefined;
+					separatorLine = undefined;
+				}
+			}
+		} catch {}
+	}
+	return blocks;
+}
+
+function formatRange(start: number, end: number): string {
+	if (start > end) return "empty";
+	if (start === end) return `${start}`;
+	return `${start}-${end}`;
+}
+
+function formatConflicts(ref: string, blocks: ConflictBlock[]): string {
+	const lines = [`Merged ${ref} with conflicts:`, ""];
+	for (const b of blocks) {
+		const ours = formatRange(b.startLine + 1, b.separatorLine - 1);
+		const theirs = formatRange(b.separatorLine + 1, b.endLine - 1);
+		lines.push(`  ${b.file}:${b.startLine}-${b.endLine} (ours ${ours}, theirs ${theirs})`);
+	}
+	lines.push("", "Resolve these conflicts.");
+	return lines.join("\n");
+}
+
+export default function (pi: ExtensionAPI) {
+	pi.on("agent_end", async (_event, ctx) => {
+		const { code: revParseCode } = await pi.exec("git", ["rev-parse", "--git-dir"]);
+		if (revParseCode !== 0) return;
+
+		let ref = "MERGE_HEAD";
+
+		// If not already in a merge, attempt one
+		const { code: mergeHeadCode } = await pi.exec("git", ["rev-parse", "MERGE_HEAD"]);
+		if (mergeHeadCode !== 0) {
+			// Only attempt a new merge if the working tree is clean
+			const { stdout: status } = await pi.exec("git", ["status", "--porcelain"]);
+			if (status.trim()) return;
+
+			const { stdout: upstream, code: upstreamCode } = await pi.exec("git", [
+				"rev-parse",
+				"--abbrev-ref",
+				"--symbolic-full-name",
+				"@{u}",
+			]);
+			if (upstreamCode !== 0) return;
+
+			ref = upstream.trim();
+			const remote = ref.split("/")[0];
+			ctx.ui.notify(`git-merge-and-resolve: fetching ${remote}, merging ${ref}`, "info");
+
+			const { code: fetchCode, stderr: fetchErr } = await pi.exec("git", ["fetch", remote]);
+			if (fetchCode !== 0) {
+				ctx.ui.notify(`git-merge-and-resolve: fetch failed: ${fetchErr.trim()}`, "warning");
+				return;
+			}
+
+			const { code: mergeCode } = await pi.exec("git", ["merge", "--no-ff", ref]);
+			if (mergeCode === 0) return;
+		}
+
+		// Either we just merged with conflicts, or we were already in an unfinished merge
+		const conflicts = await findConflicts(pi, ctx.cwd);
+		if (conflicts.length === 0) return;
+
+		pi.sendUserMessage(formatConflicts(ref, conflicts), { deliverAs: "followUp" });
+	});
+}

packages/coding-agent/src/core/agent-session.ts

@@ -947,7 +947,13 @@ export class AgentSession {
 			this._retryAttempt = 0;
 		}
 
-		return await this._checkCompaction(msg);
+		if (await this._checkCompaction(msg)) {
+			return true;
+		}
+
+		// The agent loop drains both queues before emitting agent_end. Any messages
+		// here were queued by agent_end extension handlers and need a continuation.
+		return this.agent.hasQueuedMessages();
 	}
 
 	/**

packages/coding-agent/test/git-merge-and-resolve-extension.test.ts

@@ -0,0 +1,206 @@
+import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from "node:fs";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { afterEach, describe, expect, it, vi } from "vitest";
+import mergeAndResolve from "../examples/extensions/git-merge-and-resolve.ts";
+import type { ExecResult, ExtensionAPI, ExtensionContext } from "../src/core/extensions/index.ts";
+
+type AgentEndHandler = (event: { type: "agent_end" }, ctx: ExtensionContext) => Promise<undefined>;
+
+const ok: ExecResult = { stdout: "", stderr: "", code: 0, killed: false };
+const fail: ExecResult = { stdout: "", stderr: "error", code: 1, killed: false };
+
+/** Standard exec results for a clean repo tracking origin/main, not in a merge. */
+function withUpstream(results: Map<string, ExecResult>): Map<string, ExecResult> {
+	results.set("git rev-parse --git-dir", ok);
+	results.set("git rev-parse MERGE_HEAD", fail);
+	results.set("git status --porcelain", ok);
+	results.set("git rev-parse --abbrev-ref --symbolic-full-name @{u}", { ...ok, stdout: "origin/main\n" });
+	results.set("git fetch origin", ok);
+	return results;
+}
+
+function setup(cwd: string, execResults: Map<string, ExecResult>) {
+	let handler: AgentEndHandler | undefined;
+	const sendUserMessage = vi.fn();
+
+	const exec = vi.fn<ExtensionAPI["exec"]>().mockImplementation(async (cmd, args) => {
+		const key = [cmd, ...args].join(" ");
+		return execResults.get(key) ?? fail;
+	});
+
+	const api = {
+		on: (event: string, h: AgentEndHandler) => {
+			if (event === "agent_end") handler = h;
+		},
+		exec,
+		sendUserMessage,
+	} as unknown as ExtensionAPI;
+
+	mergeAndResolve(api);
+
+	const ctx = { cwd, ui: { notify: vi.fn() } } as unknown as ExtensionContext;
+
+	async function trigger() {
+		await handler!({ type: "agent_end" }, ctx);
+	}
+
+	return { trigger, exec, sendUserMessage };
+}
+
+describe("git-merge-and-resolve example", () => {
+	let tempDir: string;
+
+	afterEach(() => {
+		if (tempDir) rmSync(tempDir, { recursive: true, force: true });
+	});
+
+	function createTempDir() {
+		tempDir = mkdtempSync(join(tmpdir(), "pi-merge-test-"));
+		return tempDir;
+	}
+
+	it("skips when not a git repository", async () => {
+		const cwd = createTempDir();
+		const results = new Map<string, ExecResult>();
+		results.set("git rev-parse --git-dir", fail);
+
+		const { trigger, exec, sendUserMessage } = setup(cwd, results);
+		await trigger();
+
+		expect(exec).toHaveBeenCalledTimes(1);
+		expect(sendUserMessage).not.toHaveBeenCalled();
+	});
+
+	it("skips when no upstream is configured", async () => {
+		const cwd = createTempDir();
+		const results = new Map<string, ExecResult>();
+		results.set("git rev-parse --git-dir", ok);
+		results.set("git rev-parse --abbrev-ref --symbolic-full-name @{u}", fail);
+
+		const { trigger, sendUserMessage } = setup(cwd, results);
+		await trigger();
+
+		expect(sendUserMessage).not.toHaveBeenCalled();
+	});
+
+	it("re-sends conflicts when in an unfinished merge", async () => {
+		const cwd = createTempDir();
+		const conflictContent = ["<<<<<<< HEAD", "ours", "=======", "theirs", ">>>>>>> origin/main"].join("\n");
+		writeFileSync(join(cwd, "file.ts"), conflictContent);
+
+		const results = new Map<string, ExecResult>();
+		results.set("git rev-parse --git-dir", ok);
+		results.set("git rev-parse MERGE_HEAD", ok);
+		results.set("git diff --name-only --diff-filter=U", { ...ok, stdout: "file.ts\n" });
+
+		const { trigger, exec, sendUserMessage } = setup(cwd, results);
+		await trigger();
+
+		// Should not attempt a new fetch/merge
+		expect(exec).not.toHaveBeenCalledWith("git", ["fetch", "origin"]);
+		expect(sendUserMessage).toHaveBeenCalledTimes(1);
+		const message = sendUserMessage.mock.calls[0][0] as string;
+		expect(message).toContain("file.ts:1-5");
+	});
+
+	it("skips when working tree is dirty and not in a merge", async () => {
+		const cwd = createTempDir();
+		const results = new Map<string, ExecResult>();
+		results.set("git rev-parse --git-dir", ok);
+		results.set("git rev-parse MERGE_HEAD", fail);
+		results.set("git status --porcelain", { ...ok, stdout: " M src/index.ts\n" });
+
+		const { trigger, exec, sendUserMessage } = setup(cwd, results);
+		await trigger();
+
+		expect(exec).not.toHaveBeenCalledWith("git", ["fetch", "origin"]);
+		expect(sendUserMessage).not.toHaveBeenCalled();
+	});
+
+	it("skips when fetch fails", async () => {
+		const cwd = createTempDir();
+		const results = withUpstream(new Map<string, ExecResult>());
+		results.set("git fetch origin", fail);
+
+		const { trigger, sendUserMessage } = setup(cwd, results);
+		await trigger();
+
+		expect(sendUserMessage).not.toHaveBeenCalled();
+	});
+
+	it("skips when merge is clean", async () => {
+		const cwd = createTempDir();
+		const results = withUpstream(new Map<string, ExecResult>());
+		results.set("git merge --no-ff origin/main", ok);
+
+		const { trigger, sendUserMessage } = setup(cwd, results);
+		await trigger();
+
+		expect(sendUserMessage).not.toHaveBeenCalled();
+	});
+
+	it("sends conflict report as a follow-up", async () => {
+		const cwd = createTempDir();
+		const conflictContent = [
+			"line 1",
+			"<<<<<<< HEAD",
+			"our change",
+			"=======",
+			"their change",
+			">>>>>>> origin/main",
+			"line 7",
+			"<<<<<<< HEAD",
+			"second conflict",
+			"=======",
+			"their second",
+			">>>>>>> origin/main",
+		].join("\n");
+
+		mkdirSync(join(cwd, "src"), { recursive: true });
+		writeFileSync(join(cwd, "src/index.ts"), conflictContent);
+
+		const results = withUpstream(new Map<string, ExecResult>());
+		results.set("git merge --no-ff origin/main", { ...fail, code: 1 });
+		results.set("git diff --name-only --diff-filter=U", { ...ok, stdout: "src/index.ts\n" });
+
+		const { trigger, sendUserMessage } = setup(cwd, results);
+		await trigger();
+
+		expect(sendUserMessage).toHaveBeenCalledTimes(1);
+		const [message, options] = sendUserMessage.mock.calls[0];
+		expect(message).toContain("src/index.ts:2-6 (ours 3, theirs 5)");
+		expect(message).toContain("src/index.ts:8-12 (ours 9, theirs 11)");
+		expect(options).toEqual({ deliverAs: "followUp" });
+	});
+
+	it("handles empty ours or theirs sections", async () => {
+		const cwd = createTempDir();
+		const conflictContent = ["<<<<<<< HEAD", "=======", "only theirs", ">>>>>>> origin/main"].join("\n");
+
+		writeFileSync(join(cwd, "empty-ours.ts"), conflictContent);
+
+		const results = withUpstream(new Map<string, ExecResult>());
+		results.set("git merge --no-ff origin/main", { ...fail, code: 1 });
+		results.set("git diff --name-only --diff-filter=U", { ...ok, stdout: "empty-ours.ts\n" });
+
+		const { trigger, sendUserMessage } = setup(cwd, results);
+		await trigger();
+
+		expect(sendUserMessage).toHaveBeenCalledTimes(1);
+		const message = sendUserMessage.mock.calls[0][0] as string;
+		expect(message).toContain("empty-ours.ts:1-4 (ours empty, theirs 3)");
+	});
+
+	it("skips message when merge fails but no conflict markers found", async () => {
+		const cwd = createTempDir();
+		const results = withUpstream(new Map<string, ExecResult>());
+		results.set("git merge --no-ff origin/main", { ...fail, code: 1 });
+		results.set("git diff --name-only --diff-filter=U", { ...ok, stdout: "" });
+
+		const { trigger, sendUserMessage } = setup(cwd, results);
+		await trigger();
+
+		expect(sendUserMessage).not.toHaveBeenCalled();
+	});
+});

packages/coding-agent/test/suite/agent-session-queue.test.ts

@@ -419,4 +419,27 @@ describe("AgentSession queue characterization", () => {
 			'Extension command "/testcmd" cannot be queued. Use prompt() or execute the command when not streaming.',
 		);
 	});
+
+	it("delivers follow-ups queued during agent_end", async () => {
+		let sent = false;
+		const harness = await createHarness({
+			extensionFactories: [
+				(pi: ExtensionAPI) => {
+					pi.on("agent_end", async () => {
+						if (sent) return;
+						sent = true;
+						pi.sendUserMessage("conflict report", { deliverAs: "followUp" });
+					});
+				},
+			],
+		});
+		harnesses.push(harness);
+
+		harness.setResponses([fauxAssistantMessage("reply"), fauxAssistantMessage("follow-up reply")]);
+
+		await harness.session.prompt("hello");
+		await harness.session.agent.waitForIdle();
+
+		expect(getUserTexts(harness)).toEqual(["hello", "conflict report"]);
+	});
 });