From 490966593a08387206bd9e6f3cf03b3d2a65fa92 Mon Sep 17 00:00:00 2001 From: imtahirabatool Date: Thu, 29 Oct 2026 12:32:47 +0500 Subject: [PATCH] stop() kills the agent's process tree and ends the stream with task.stopped --- README.md | 4 +- runner/package.json | 3 +- runner/pnpm-lock.yaml | 9 +++++ runner/src/process/kill-tree.ts | 23 +++++++++-- runner/src/runner.ts | 60 +++++++++++++++++++++------- runner/src/session-store.ts | 1 + runner/test/helpers/replay-agent.mjs | 24 +++++++---- runner/test/runner.test.ts | 60 ++++++++++++++++++++++++++-- 8 files changed, 153 insertions(+), 31 deletions(-) diff --git a/README.md b/README.md index 0b1286f..dfc212e 100644 --- a/README.md +++ b/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 | diff --git a/runner/package.json b/runner/package.json index 864baec..07ab696 100644 --- a/runner/package.json +++ b/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", diff --git a/runner/pnpm-lock.yaml b/runner/pnpm-lock.yaml index cae035b..b3c7c49 100644 --- a/runner/pnpm-lock.yaml +++ b/runner/pnpm-lock.yaml @@ -14,6 +14,9 @@ importers: execa: specifier: ^10.2.0 version: 10.2.0 + tree-kill: + specifier: ^1.2.2 + version: 1.2.2 devDependencies: '@types/node': specifier: ^26.6.5 @@ -680,6 +683,10 @@ packages: resolution: {integrity: sha512-u8KszXvGfU68hVcZpRHKG28T0krMuv2G5nDhiHaMLen/gIuFEgIJhaJuO69qjnXg5paSrbPMFfx3brNuN8eVSg==} engines: {node: '>=14.0.0'} + tree-kill@1.2.2: + resolution: {integrity: sha512-L0Orpi8qGpRG//Nd+H90vFB+3iHnue1zSSGmNOOCh1GLJ7rUKVwV2HvijphGQS2UmhUZewS9VgvxYIdgr+fG1A==} + hasBin: true + tsx@4.23.15: resolution: {integrity: sha512-Yiex1Ovn8z2xPpOWckIiysV1SSyRMY9BkLF++q0yKiDxCqRhosKfMg3janKkiLBwZ5c/YryloKwGZcrEmtwxKw==} engines: {node: '>=18.0.0'} @@ -1263,6 +1270,8 @@ snapshots: tinyspy@4.0.6: {} + tree-kill@1.2.2: {} + tsx@4.23.15: dependencies: esbuild: 0.28.2 diff --git a/runner/src/process/kill-tree.ts b/runner/src/process/kill-tree.ts index 7a2c709..05c26cd 100644 --- a/runner/src/process/kill-tree.ts +++ b/runner/src/process/kill-tree.ts @@ -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 { - 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((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, graceMs = 3000): Promise { + await kill(pid, "SIGTERM"); + let timer: NodeJS.Timeout | undefined; + const timedOut = await Promise.race([ + exited.then(() => false), + new Promise((resolve) => (timer = setTimeout(() => resolve(true), graceMs))), + ]); + clearTimeout(timer); + if (timedOut) { + await kill(pid, "SIGKILL"); + await exited; + } } diff --git a/runner/src/runner.ts b/runner/src/runner.ts index 025082c..e15eca7 100644 --- a/runner/src/runner.ts +++ b/runner/src/runner.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>(); - async function* start(o: StartOptions): AsyncIterable { - 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 = {}) => { - 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 { + 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 { + 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)"); }, diff --git a/runner/src/session-store.ts b/runner/src/session-store.ts index 4a2041f..9df57f3 100644 --- a/runner/src/session-store.ts +++ b/runner/src/session-store.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 }; } diff --git a/runner/test/helpers/replay-agent.mjs b/runner/test/helpers/replay-agent.mjs index cf86c56..4be703b 100644 --- a/runner/test/helpers/replay-agent.mjs +++ b/runner/test/helpers/replay-agent.mjs @@ -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 [exitCode] [edit] +// node replay-agent.mjs [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); diff --git a/runner/test/runner.test.ts b/runner/test/runner.test.ts index a41189a..424d3f0 100644 --- a/runner/test/runner.test.ts +++ b/runner/test/runner.test.ts @@ -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 | 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/); + }); +});