fix(supervisor): serialize cancellation and terminal job state

docs/supervisor.md

@@ -1,6 +1,6 @@
 # Supervisor (jobs + shells)
 
-> **Status: unsupported experimental.** These surfaces are source-available but outside the supported v3 product contract. They are hidden from default `hack --help` (see `hack help --all`) and print a warning when invoked.
+> **Status: unsupported experimental.** These surfaces are source-available but outside the supported product contract. They are hidden from default `hack --help` (see `hack help --all`) and print a warning when invoked.
 
 The supervisor is the execution engine behind remote workflows. It can run commands as jobs,
 stream logs/events, and host PTY-backed shells. The CLI exposes it locally via `hack x supervisor`
@@ -10,6 +10,16 @@ This page documents a beta-adjacent execution surface.
 Use [Beta workflows](beta.md) for the guided remote path and [Extensions & reference](reference.md)
 for the rest of the command and API material.
 
+Cancellation is coordinated with the job runner. A successful cancel response waits
+for the runner's terminal metadata and event, not process exit or log drain; it does not
+race a second writer against process exit. Once terminal persistence has started,
+a later cancellation returns `not_running` and preserves the completed/failed outcome.
+Repeated requests accepted before finalization share one cancellation outcome and event.
+If terminal persistence fails, cancellation and the run promise report that failure;
+the service logs it without replacing the claimed outcome with `failed`. Metadata or
+event history can be incomplete, so a persistence error is not a successful acknowledgement.
+No automatic retry appends another terminal event after a partial write.
+
 ## Local usage
 
 ```bash

src/control-plane/extensions/supervisor/runner.ts

@@ -10,6 +10,8 @@ export type JobRunResult = {
 
 export type JobSpawnListener = (opts: {
   readonly proc: SpawnedProcess;
+  /** Request cancellation before terminal status persistence begins. */
+  readonly cancel: () => Promise<boolean>;
 }) => void;
 
 type SpawnedProcess = ReturnType<typeof Bun.spawn>;
@@ -23,6 +25,7 @@ type SpawnedProcess = ReturnType<typeof Bun.spawn>;
  * @param opts.cwd - Optional working directory for the process.
  * @param opts.env - Optional environment overrides.
  * @param opts.onSpawn - Optional hook with the spawned process handle.
+ * @param opts.onTerminalClaim - Observe the chosen outcome before persistence yields.
  * @returns Final job status and exit code.
  */
 export async function runJob(opts: {
@@ -32,6 +35,7 @@ export async function runJob(opts: {
   readonly cwd?: string;
   readonly env?: Record<string, string>;
   readonly onSpawn?: JobSpawnListener;
+  readonly onTerminalClaim?: (opts: { readonly status: JobStatus }) => void;
 }): Promise<JobRunResult> {
   const meta = await opts.jobStore.readJobMeta({ jobId: opts.jobId });
   if (!meta) {
@@ -79,7 +83,48 @@ export async function runJob(opts: {
     type: "job.started",
     payload: { pid: proc.pid },
   });
-  opts.onSpawn?.({ proc });
+
+  let terminalStatus: JobStatus | undefined;
+  let terminalWrite: Promise<JobStatus> | undefined;
+  const finish = (input: {
+    readonly status: JobStatus;
+    readonly exitCode?: number;
+  }): Promise<JobStatus> => {
+    if (terminalWrite) {
+      return terminalWrite;
+    }
+    // Claim the outcome before storage yields; both paths share one writer.
+    terminalStatus = input.status;
+    opts.onTerminalClaim?.({ status: input.status });
+    terminalWrite = (async () => {
+      await opts.jobStore.updateJobStatus({
+        jobId: opts.jobId,
+        status: input.status,
+      });
+      await opts.jobStore.appendEvent({
+        jobId: opts.jobId,
+        type: `job.${input.status}`,
+        ...(input.exitCode === undefined
+          ? {}
+          : { payload: { exitCode: input.exitCode } }),
+      });
+      return input.status;
+    })();
+    return terminalWrite;
+  };
+  opts.onSpawn?.({
+    proc,
+    cancel: async () => {
+      if (terminalStatus && terminalStatus !== "cancelled") {
+        return false;
+      }
+      if (!terminalStatus) {
+        proc.kill();
+      }
+      await finish({ status: "cancelled" });
+      return true;
+    },
+  });
 
   const paths = opts.jobStore.getJobPaths({ jobId: opts.jobId });
   const stdoutTask = pipeStreamToFiles({
@@ -94,19 +139,10 @@ export async function runJob(opts: {
   const exitCode = await proc.exited;
   await Promise.all([stdoutTask, stderrTask]);
 
-  const metaAfter = await opts.jobStore.readJobMeta({ jobId: opts.jobId });
-  if (metaAfter?.status === "cancelled") {
-    return { jobId: opts.jobId, status: "cancelled", exitCode };
-  }
-
-  const status: JobStatus = exitCode === 0 ? "completed" : "failed";
-  await opts.jobStore.updateJobStatus({ jobId: opts.jobId, status });
-  await opts.jobStore.appendEvent({
-    jobId: opts.jobId,
-    type: status === "completed" ? "job.completed" : "job.failed",
-    payload: { exitCode },
+  const status = await finish({
+    status: exitCode === 0 ? "completed" : "failed",
+    exitCode,
   });
-
   return { jobId: opts.jobId, status, exitCode };
 }
 

src/control-plane/extensions/supervisor/service.ts

@@ -40,7 +40,7 @@ export type SupervisorService = {
     readonly env?: Record<string, string>;
   }) => Promise<CreateJobResult>;
   /**
-   * Attempt to cancel a running job.
+   * Request cancellation and wait for its terminal metadata and event.
    *
    * @param opts.projectDir - Project .hack directory.
    * @param opts.jobId - Job id to cancel.
@@ -76,15 +76,18 @@ export type SupervisorService = {
  * Create a supervisor service for managing jobs and their metadata.
  *
  * @param opts.logger - Optional logger override.
+ * @param opts.createStore - Optional job-store factory.
  * @returns Supervisor service helpers.
  */
 export function createSupervisorService(opts?: {
   readonly logger?: Logger;
+  readonly createStore?: typeof createJobStore;
 }): SupervisorService {
   const logger = opts?.logger ?? baseLogger;
+  const createStore = opts?.createStore ?? createJobStore;
   const runningJobs = new Map<
     string,
-    { readonly proc: ReturnType<typeof Bun.spawn> }
+    { readonly cancel: () => Promise<boolean> }
   >();
 
   const createJob = async (input: {
@@ -96,7 +99,7 @@ export function createSupervisorService(opts?: {
     readonly cwd?: string;
     readonly env?: Record<string, string>;
   }): Promise<CreateJobResult> => {
-    const store = await createJobStore({ projectDir: input.projectDir });
+    const store = await createStore({ projectDir: input.projectDir });
     const jobId = randomUUID();
     const meta = await store.createJob({
       jobId,
@@ -106,17 +109,24 @@ export function createSupervisorService(opts?: {
       projectName: input.projectName,
     });
 
+    let terminalClaimed = false;
     const run = runJob({
       jobStore: store,
       jobId,
       command: input.command,
       cwd: input.cwd,
       env: input.env,
-      onSpawn: ({ proc }) => {
-        runningJobs.set(jobId, { proc });
+      onTerminalClaim: () => {
+        terminalClaimed = true;
+      },
+      onSpawn: ({ cancel }) => {
+        runningJobs.set(jobId, { cancel });
       },
     })
       .catch(async (error) => {
+        if (terminalClaimed) {
+          throw error;
+        }
         logger.error({ message: `Job failed: ${formatError(error)}` });
         await store.updateJobStatus({ jobId, status: "failed" });
         await store.appendEvent({
@@ -131,6 +141,13 @@ export function createSupervisorService(opts?: {
         runningJobs.delete(jobId);
       });
 
+    // Background callers need not await run; observing errors here keeps the
+    // returned promise rejected without creating an unhandled daemon rejection.
+    void run.catch((error) => {
+      logger.error({
+        message: `Job persistence failed: ${formatError(error)}`,
+      });
+    });
     return { jobId, meta, run };
   };
 
@@ -141,7 +158,7 @@ export function createSupervisorService(opts?: {
     readonly projectDir: string;
     readonly jobId: string;
   }): Promise<CancelJobResult> => {
-    const store = await createJobStore({ projectDir });
+    const store = await createStore({ projectDir });
     const meta = await store.readJobMeta({ jobId });
     if (!meta) {
       return { ok: false, status: "not_found" };
@@ -152,10 +169,9 @@ export function createSupervisorService(opts?: {
       return { ok: false, status: "not_running" };
     }
 
-    running.proc.kill();
-    await store.updateJobStatus({ jobId, status: "cancelled" });
-    await store.appendEvent({ jobId, type: "job.cancelled" });
-
+    if (!(await running.cancel())) {
+      return { ok: false, status: "not_running" };
+    }
     return { ok: true, status: "cancelled" };
   };
 
@@ -166,7 +182,7 @@ export function createSupervisorService(opts?: {
     readonly projectDir: string;
     readonly jobId: string;
   }): Promise<JobMeta | null> => {
-    const store = await createJobStore({ projectDir });
+    const store = await createStore({ projectDir });
     return await store.readJobMeta({ jobId });
   };
 
@@ -175,7 +191,7 @@ export function createSupervisorService(opts?: {
   }: {
     readonly projectDir: string;
   }): Promise<readonly JobMeta[]> => {
-    const store = await createJobStore({ projectDir });
+    const store = await createStore({ projectDir });
     const entries = await safeReadDir(store.jobsRoot);
     const metas = await Promise.all(
       entries.map((jobId) => store.readJobMeta({ jobId }))