stop() kills the agent's process tree and ends the stream with task.stopped

Author: imtahirabatool

README.md

@@ -73,7 +73,7 @@ Week 1 (Oct 19 to 23, 2026).
 | Worktrees: branch names, add, remove, prune | [runner/src/git/worktree.ts](runner/src/git/worktree.ts) | Done |
 | Diff since the base commit, commit on approve | [runner/src/git/diff.ts](runner/src/git/diff.ts) | Done |
 | `start`, `diff`, `approve`, `reject` | [runner/src/runner.ts](runner/src/runner.ts) | Done, tested with real runs |
-| `stop`: kill the whole process tree, keep the worktree | [runner/src/process/kill-tree.ts](runner/src/process/kill-tree.ts) | To do. Throws `not implemented` |
+| `stop`: kill the whole process tree, keep the worktree | [runner/src/process/kill-tree.ts](runner/src/process/kill-tree.ts) | Done. `task.stopped` comes out of the `start()` stream; `stop()` resolves once the session is `stopped` |
 | `resume`: same worktree, `seq` continues | [runner/src/runner.ts](runner/src/runner.ts) | To do. Throws `not implemented`. The `--resume` arguments exist |
 | Codex: `codex exec --json` to events | [runner/src/agents/codex.ts](runner/src/agents/codex.ts) | To do. `start()` with `agent: "codex"` fails cleanly for now |
 | Repo notes read before a run, summary appended after | [runner/src/memory/notes.ts](runner/src/memory/notes.ts) | To do. `repoKey()` is done |
@@ -211,7 +211,7 @@ The scripts in `runner/scripts/`:
 | File | Covers |
 | --- | --- |
 | [claude-parser.test.ts](runner/test/claude-parser.test.ts) | The whole recorded run replays in order; each mapping above; blocked commands; failed tests hidden behind exit code 0; error results; broken lines; `raw` on every event |
-| [runner.test.ts](runner/test/runner.test.ts) | A worktree on a new branch with your repo untouched; clean failures for a repo with no commits, an agent with no result line, a missing binary and an unsupported agent; diff with new files; approve commits on the branch and leaves `main` alone; reject deletes the branch; nothing to approve |
+| [runner.test.ts](runner/test/runner.test.ts) | A worktree on a new branch with your repo untouched; clean failures for a repo with no commits, an agent with no result line, a missing binary and an unsupported agent; diff with new files; approve commits on the branch and leaves `main` alone; reject deletes the branch; nothing to approve; stop kills the agent and its children, keeps the worktree and refuses a session that is not running |
 | [fake-runner.test.ts](runner/test/fake-runner.test.ts) | Event order and `seq`; approve only after finishing; stop then resume with `seq` continuing |
 | [worktree.test.ts](runner/test/worktree.test.ts) | Branch names |
 

runner/package.json

@@ -22,7 +22,8 @@
   },
   "dependencies": {
     "@codehands/contracts": "file:../../codehands-contracts",
-    "execa": "^10.2.0"
+    "execa": "^10.2.0",
+    "tree-kill": "^1.2.2"
   },
   "devDependencies": {
     "@types/node": "^26.6.5",

runner/pnpm-lock.yaml

@@ -1,7 +1,24 @@
 // Week 1 guide, section 7 and diagram 1e. Killing the parent does not kill its
 // children on Windows (claude -> npm test -> node), so kill the whole tree.
 
-// TODO(Wed): ask politely, wait about 3 s, then `tree-kill` (taskkill /T /F on Windows).
-export async function stopProcessTree(_pid: number, _graceMs = 3000): Promise<void> {
-  throw new Error("not implemented: stopProcessTree (Wed)");
+import treeKill from "tree-kill";
+
+// Errors are ignored: the most common one is "already gone", which is what we want.
+const kill = (pid: number, signal: NodeJS.Signals) =>
+  new Promise<void>((resolve) => treeKill(pid, signal, () => resolve()));
+
+// Ask politely, wait up to graceMs for `exited`, then force. On Windows tree-kill always
+// runs `taskkill /T /F`, because a console program can't be asked to close, so the first kill is final.
+export async function stopProcessTree(pid: number, exited: Promise<unknown>, graceMs = 3000): Promise<void> {
+  await kill(pid, "SIGTERM");
+  let timer: NodeJS.Timeout | undefined;
+  const timedOut = await Promise.race([
+    exited.then(() => false),
+    new Promise<boolean>((resolve) => (timer = setTimeout(() => resolve(true), graceMs))),
+  ]);
+  clearTimeout(timer);
+  if (timedOut) {
+    await kill(pid, "SIGKILL");
+    await exited;
+  }
 }

runner/src/process/kill-tree.ts

@@ -6,7 +6,8 @@ import { EventSequence } from "./events.js";
 import { commitAll, diffSince } from "./git/diff.js";
 import { addWorktree, baseCommit, branchName, pruneWorktrees, removeWorktree, worktreePath } from "./git/worktree.js";
 import { repoKey } from "./memory/notes.js";
-import { SessionStore } from "./session-store.js";
+import { stopProcessTree } from "./process/kill-tree.js";
+import { SessionStore, type SessionRecord } from "./session-store.js";
 
 // The real runner. Diagram 1c in the week 1 guide is the map for start().
 
@@ -20,18 +21,26 @@ const claudeCommand = (o: StartOptions) => ({ file: "claude", args: claudeArgs(o
 
 export function createRunner({ worktreeRoot, agentCommand = claudeCommand }: RunnerOptions = {}): Runner {
   const store = new SessionStore();
+  // The running agent of each session, so stop() can wait for it to exit.
+  const exits = new Map<string, Promise<unknown>>();
 
-  async function* start(o: StartOptions): AsyncIterable<RunnerEvent> {
-    const record = store.create(o);
-    const seq = new EventSequence(o.sessionId);
+  // Every event goes through emit, so lastSeq is always right for resume().
+  function emitters(record: SessionRecord, seq: EventSequence) {
     const emit = (event: RunnerEvent) => {
       record.lastSeq = seq.last;
       return event;
     };
     const fail = (reason: string, extra: Record<string, unknown> = {}) => {
-      store.transition(o.sessionId, "failed");
+      store.transition(record.options.sessionId, "failed");
       return emit(seq.next("task.failed", { reason, ...extra }));
     };
+    return { emit, fail };
+  }
+
+  async function* start(o: StartOptions): AsyncIterable<RunnerEvent> {
+    const record = store.create(o);
+    const seq = new EventSequence(o.sessionId);
+    const { fail } = emitters(record, seq);
 
     if (o.agent !== "claude-code") {
       yield fail(`agent ${o.agent} is not supported yet`);
@@ -50,17 +59,26 @@ export function createRunner({ worktreeRoot, agentCommand = claudeCommand }: Run
       return;
     }
     store.transition(o.sessionId, "running");
+    yield* runAgent(record, seq, agentCommand(o));
+  }
+
+  // Steps 2 to 5 of a run. start() and resume() share them, so a resumed run ends the same way.
+  async function* runAgent(record: SessionRecord, seq: EventSequence, { file, args }: { file: string; args: string[] }): AsyncIterable<RunnerEvent> {
+    const { sessionId } = record.options;
+    const { worktreePath, baseCommit } = worktreeOf(sessionId);
+    const { emit, fail } = emitters(record, seq);
+    record.stopRequested = false;
 
     // 2. Run the agent inside the copy. reject: false, so a crash or a missing binary is a result, not a throw.
-    const { file, args } = agentCommand(o);
     const child = execa(file, args, {
-      cwd: record.worktreePath,
+      cwd: worktreePath,
       stdin: "ignore",
       reject: false,
       buffer: { stdout: false, stderr: true },
       windowsHide: true,
     });
     record.pid = child.pid;
+    exits.set(sessionId, child);
 
     // 3. Every complete stdout line becomes events. readline never hands over half a line.
     //    The final event waits until the process is gone, so it can carry the diff.
@@ -76,6 +94,14 @@ export function createRunner({ worktreeRoot, agentCommand = claudeCommand }: Run
     // 4. The process is done. No result line means it crashed or never started.
     const result = await child;
     record.pid = undefined;
+    exits.delete(sessionId);
+
+    // stop() killed it. The worktree stays so the session can resume.
+    if (record.stopRequested) {
+      if (record.status === "running") store.transition(sessionId, "stopped");
+      yield emit(seq.next("task.stopped", {}));
+      return;
+    }
     if (!finished) {
       yield fail("the agent exited without a result", {
         exitCode: result.exitCode,
@@ -85,19 +111,19 @@ export function createRunner({ worktreeRoot, agentCommand = claudeCommand }: Run
       return;
     }
     if (finished.type === "task.failed") {
-      store.transition(o.sessionId, "failed");
+      store.transition(sessionId, "failed");
       yield emit(finished);
       return;
     }
 
     // 5. Store the change and put its summary on task.finished, so the phone shows it right away.
     try {
-      record.diff = await diffSince(record.worktreePath, record.baseCommit);
+      record.diff = await diffSince(worktreePath, baseCommit);
       finished.payload.diffStat = record.diff.stat;
     } catch (err) {
       finished.payload.diffError = (err as Error).message;
     }
-    store.transition(o.sessionId, "awaiting_approval");
+    store.transition(sessionId, "awaiting_approval");
     yield emit(finished);
   }
 
@@ -112,11 +138,17 @@ export function createRunner({ worktreeRoot, agentCommand = claudeCommand }: Run
 
   return {
     start,
-    // TODO(Wed): stopProcessTree(pid), transition to stopped, yield task.stopped. Keep the worktree.
-    async stop() {
-      throw new Error("not implemented: stop (Wed)");
+    // Kills the agent and everything it started. task.stopped comes out of the start() stream,
+    // and the session is stopped by the time this resolves. The worktree stays.
+    async stop(sessionId) {
+      const record = store.check(sessionId, "stopped");
+      const exited = exits.get(sessionId);
+      if (!record.pid || !exited) throw new Error(`session ${sessionId} has no agent process to stop`);
+      record.stopRequested = true;
+      await stopProcessTree(record.pid, exited);
+      if (record.status === "running") store.transition(sessionId, "stopped");
     },
-    // TODO(Thu am): claudeResumeArgs in the same worktree, seq continues from lastSeq.
+    // TODO: runAgent(record, new EventSequence(id, record.lastSeq), claudeResumeArgs(...)) in the same worktree.
     resume() {
       throw new Error("not implemented: resume (Thu)");
     },

runner/src/runner.ts

@@ -9,6 +9,7 @@ export interface SessionRecord {
   branch?: string;
   baseCommit?: string;
   pid?: number;
+  stopRequested?: boolean;  // set by stop(), so the agent's exit reads as task.stopped, not task.failed
   lastSeq: number;
   diff?: { patch: string; stat: string };
 }

runner/src/session-store.ts

@@ -1,12 +1,22 @@
 // Stand-in for `claude`: prints a recorded JSONL file line by line, then exits with the given code.
-// node replay-agent.mjs <file.jsonl> [exitCode] [edit]
+// node replay-agent.mjs <file.jsonl> [exitCode] [edit | hang] [pidFile]
 // With "edit" it also changes a.txt and adds new.txt in its working folder, like a real fix would.
+// With "hang" it prints the first 2 lines (task.started), starts a child that never exits
+// (like claude -> npm test), writes the child's pid to pidFile and never exits itself.
+import { spawn } from "node:child_process";
 import { readFileSync, writeFileSync } from "node:fs";
 
-const [file, exitCode = "0", mode] = process.argv.slice(2);
-if (mode === "edit") {
-  writeFileSync("a.txt", "hello\nworld\n");
-  writeFileSync("new.txt", "a new file\n");
+const [file, exitCode = "0", mode, pidFile] = process.argv.slice(2);
+if (mode === "hang") {
+  const child = spawn(process.execPath, ["-e", "setInterval(() => {}, 1000)"], { stdio: "ignore" });
+  writeFileSync(pidFile, String(child.pid));
+  process.stdout.write(readFileSync(file, "utf8").split("\n").slice(0, 2).join("\n") + "\n");
+  setInterval(() => {}, 1000);
+} else {
+  if (mode === "edit") {
+    writeFileSync("a.txt", "hello\nworld\n");
+    writeFileSync("new.txt", "a new file\n");
+  }
+  if (file) process.stdout.write(readFileSync(file, "utf8"));
+  process.exitCode = Number(exitCode);
 }
-if (file) process.stdout.write(readFileSync(file, "utf8"));
-process.exitCode = Number(exitCode);

runner/test/helpers/replay-agent.mjs

@@ -1,11 +1,11 @@
 import { execFileSync } from "node:child_process";
 import { randomUUID } from "node:crypto";
-import { existsSync, mkdtempSync, rmSync, writeFileSync } from "node:fs";
+import { existsSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs";
 import { tmpdir } from "node:os";
 import { join } from "node:path";
 import { fileURLToPath } from "node:url";
 import type { RunnerEvent, StartOptions } from "@codehands/contracts";
-import { afterAll, describe, expect, it } from "vitest";
+import { afterAll, describe, expect, it, vi } from "vitest";
 import { createRunner, type RunnerOptions } from "../src/index.js";
 
 // start() against a throwaway git repo. The agent is a script that replays spike-1.jsonl, so no API cost.
@@ -40,8 +40,17 @@ async function collect(o: StartOptions, runnerOptions: RunnerOptions = {}, runne
 
 const newRunner = (runnerOptions: RunnerOptions = {}) => createRunner({ worktreeRoot: join(tmp, "worktrees"), ...runnerOptions });
 
-const replayAgent = (file: string, exitCode = 0, mode = ""): RunnerOptions["agentCommand"] =>
-  () => ({ file: process.execPath, args: [replay, file, String(exitCode), mode] });
+const replayAgent = (file: string, exitCode = 0, mode = "", pidFile = ""): RunnerOptions["agentCommand"] =>
+  () => ({ file: process.execPath, args: [replay, file, String(exitCode), mode, pidFile] });
+
+const alive = (pid: number) => {
+  try {
+    process.kill(pid, 0);
+    return true;
+  } catch {
+    return false;
+  }
+};
 
 describe("runner.start", () => {
   it("makes a worktree on a new branch, runs the agent there and streams its events", async () => {
@@ -156,3 +165,46 @@ describe("runner diff, approve and reject", () => {
     await expect(runner.reject(o.sessionId)).rejects.toThrow(/no worktree/);
   });
 });
+
+describe("runner.stop", () => {
+  // A run whose agent hangs after task.started, with a child of its own, like claude -> npm test.
+  async function stoppedRun() {
+    const repo = makeRepo();
+    const o = options(repo);
+    const pidFile = join(tmp, `child-${o.sessionId}.pid`);
+    const runner = newRunner({ agentCommand: replayAgent(spike, 0, "hang", pidFile) });
+    const events: RunnerEvent[] = [];
+    let stopping: Promise<void> | undefined;
+    for await (const e of runner.start(o)) {
+      events.push(e);
+      if (e.type === "task.started") stopping = runner.stop(o.sessionId);
+    }
+    await stopping;
+    return { repo, o, runner, events, childPid: Number(readFileSync(pidFile, "utf8")) };
+  }
+
+  it("kills the agent and its children, and the stream ends with task.stopped", async () => {
+    const { events, childPid } = await stoppedRun();
+    expect(events.map((e) => e.type)).toEqual(["task.started", "task.stopped"]);
+    expect(events.map((e) => e.seq)).toEqual([1, 2]);
+    await vi.waitFor(() => expect(alive(childPid)).toBe(false), { timeout: 3000 });
+  });
+
+  it("keeps the worktree and the branch, so the session can resume", async () => {
+    const { repo, o } = await stoppedRun();
+    const short = o.sessionId.slice(0, 8);
+    expect(git(repo, "worktree", "list")).toContain(short);
+    expect(git(repo, "branch", "--list", `codehands/fix-the-failing-test-${short}`)).not.toBe("");
+  });
+
+  it("refuses to stop a session that is not running", async () => {
+    const { runner, o } = await stoppedRun();
+    await expect(runner.stop(o.sessionId)).rejects.toThrow(/illegal move/); // already stopped
+
+    const finished = options(makeRepo());
+    const other = newRunner({ agentCommand: replayAgent(spike) });
+    await collect(finished, {}, other);
+    await expect(other.stop(finished.sessionId)).rejects.toThrow(/illegal move/); // awaiting approval
+    await expect(other.stop(randomUUID())).rejects.toThrow(/unknown session/);
+  });
+});