fix(core): bound consumed job results

packages/core/src/job.ts

@@ -36,6 +36,7 @@ export type Status = Background["status"]
 
 const decodeBackground = Schema.decodeUnknownResult(Background)
 const backgroundPrefix = "job.background/"
+const COMPLETED_LIMIT = 25
 
 export type Info = {
   id: string
@@ -58,6 +59,7 @@ type Active = {
   token: object
   blockingSessions: Map<SessionSchema.ID, number>
   isBackgrounded: boolean
+  consumed: boolean
   recovery?: Recovery
 }
 
@@ -70,6 +72,7 @@ type FinishResult = {
   info?: Info
   done?: Deferred.Deferred<Info>
   scope?: Scope.Closeable
+  token?: object
 }
 
 type BackgroundResult = {
@@ -82,11 +85,12 @@ type StartResult = { info: Info } | { info: Info; scope: Scope.Closeable; token:
 type BlockWait = {
   done: Deferred.Deferred<Info>
   backgrounded: Deferred.Deferred<Info>
+  token: object
 }
 
 type BlockStart =
   | { type: "missing" }
-  | { type: "finished"; info: Info }
+  | { type: "finished"; info: Info; token: object }
   | { type: "backgrounded"; info: Info }
   | { type: "wait"; wait: BlockWait }
 
@@ -164,6 +168,9 @@ function decrementSession(input: Map<SessionSchema.ID, number>, sessionID: Sessi
 /**
  * Makes one scoped, process-local registry. Explicitly recoverable background
  * work also owns a durable notification marker until its notification is admitted.
+ * Unconsumed results survive the start-to-wait handoff. Foreground block/cancel
+ * and non-recoverable wait results enter a 25-entry consumed history. Recoverable
+ * wait results stay available for background registration and acknowledgment.
  */
 export const make = Effect.gen(function* () {
   const kv = yield* KV.Service
@@ -172,6 +179,19 @@ export const make = Effect.gen(function* () {
     scope: yield* Scope.Scope,
   }
 
+  const consume = (id: string, token: object) =>
+    SynchronizedRef.update(state.jobs, (jobs) => {
+      const job = jobs.get(id)
+      if (!job || job.token !== token || job.info.status === "running" || job.consumed) return jobs
+      const next = new Map(jobs)
+      // Order history by first consumption, not by start time or subsequent reads.
+      next.delete(id)
+      next.set(id, { ...job, consumed: true })
+      const completed = [...next].filter(([, job]) => job.consumed && !job.info.notificationID)
+      for (const [id] of completed.slice(0, -COMPLETED_LIMIT)) next.delete(id)
+      return next
+    })
+
   const persistBackground = Effect.fnUntraced(function* (job: Active) {
     if (!job.recovery || !job.info.notificationID) return
     yield* kv.set(`${backgroundPrefix}${job.info.notificationID}`, {
@@ -258,6 +278,7 @@ export const make = Effect.gen(function* () {
               token,
               blockingSessions: new Map<SessionSchema.ID, number>(),
               isBackgrounded: false,
+              consumed: false,
               recovery: input.recovery,
             }
             return [{ info: snapshot(job), scope, token }, new Map(jobs).set(id, job)]
@@ -278,12 +299,19 @@ export const make = Effect.gen(function* () {
   const wait: Interface["wait"] = Effect.fn("Job.wait")(function* (input) {
     const job = (yield* SynchronizedRef.get(state.jobs)).get(input.id)
     if (!job) return { timedOut: false }
-    if (job.info.status !== "running") return { info: snapshot(job), timedOut: false }
-    if (input.timeout === undefined) return { info: yield* Deferred.await(job.done), timedOut: false }
-    if (input.timeout <= 0) return { info: snapshot(job), timedOut: true }
-    const info = yield* Deferred.await(job.done).pipe(Effect.timeoutOption(input.timeout))
-    if (info._tag === "Some") return { info: info.value, timedOut: false }
-    return { info: snapshot(job), timedOut: true }
+    return yield* Effect.gen(function* () {
+      if (job.info.status !== "running") return { info: snapshot(job), timedOut: false }
+      if (input.timeout === undefined) return { info: yield* Deferred.await(job.done), timedOut: false }
+      if (input.timeout <= 0) return { info: snapshot(job), timedOut: true }
+      const info = yield* Deferred.await(job.done).pipe(Effect.timeoutOption(input.timeout))
+      if (info._tag === "Some") return { info: info.value, timedOut: false }
+      return { info: snapshot(job), timedOut: true }
+    }).pipe(
+      // Recoverable wait -> background is a supported handoff, even after failure.
+      Effect.tap((result) =>
+        result.info.status === "running" || job.recovery ? Effect.void : consume(input.id, job.token),
+      ),
+    )
   })
 
   const removeBlock = Effect.fnUntraced(function* (input: BlockInput) {
@@ -301,10 +329,10 @@ export const make = Effect.gen(function* () {
     const result = yield* SynchronizedRef.modify(state.jobs, (jobs): readonly [BlockStart, Map<string, Active>] => {
       const job = jobs.get(input.id)
       if (!job) return [{ type: "missing" }, jobs]
-      if (job.info.status !== "running") return [{ type: "finished", info: snapshot(job) }, jobs]
+      if (job.info.status !== "running") return [{ type: "finished", info: snapshot(job), token: job.token }, jobs]
       if (job.isBackgrounded) return [{ type: "backgrounded", info: snapshot(job) }, jobs]
       return [
-        { type: "wait", wait: { done: job.done, backgrounded: job.backgrounded } },
+        { type: "wait", wait: { done: job.done, backgrounded: job.backgrounded, token: job.token } },
         new Map(jobs).set(input.id, {
           ...job,
           blockingSessions: incrementSession(job.blockingSessions, input.sessionID),
@@ -312,12 +340,18 @@ export const make = Effect.gen(function* () {
       ]
     })
     if (result.type === "missing") return undefined
-    if (result.type === "finished") return { type: "finished", info: result.info }
+    if (result.type === "finished") {
+      yield* consume(input.id, result.token)
+      return { type: "finished", info: result.info }
+    }
     if (result.type === "backgrounded") return { type: "backgrounded", info: result.info }
     return yield* Effect.raceFirst(
       Deferred.await(result.wait.done).pipe(Effect.map((info) => ({ type: "finished" as const, info }))),
       Deferred.await(result.wait.backgrounded).pipe(Effect.map((info) => ({ type: "backgrounded" as const, info }))),
-    ).pipe(Effect.ensuring(removeBlock(input)))
+    ).pipe(
+      Effect.tap((outcome) => (outcome.type === "finished" ? consume(input.id, result.wait.token) : Effect.void)),
+      Effect.ensuring(removeBlock(input)),
+    )
   })
 
   const markBackground = Effect.fnUntraced(function* (job: Active) {
@@ -381,7 +415,7 @@ export const make = Effect.gen(function* () {
       Effect.fnUntraced(function* (jobs): Effect.fn.Return<readonly [FinishResult, Map<string, Active>]> {
         const job = jobs.get(id)
         if (!job) return [{}, jobs]
-        if (job.info.status !== "running") return [{ info: snapshot(job) }, jobs]
+        if (job.info.status !== "running") return [{ info: snapshot(job), token: job.token }, jobs]
         const next = {
           ...job,
           blockingSessions: new Map<SessionSchema.ID, number>(),
@@ -392,11 +426,15 @@ export const make = Effect.gen(function* () {
           },
         }
         yield* persistBackground(next)
-        return [{ info: snapshot(next), done: job.done, scope: job.scope }, new Map(jobs).set(id, next)]
+        return [
+          { info: snapshot(next), done: job.done, scope: job.scope, token: job.token },
+          new Map(jobs).set(id, next),
+        ]
       }),
     )
     if (result.info && result.done) yield* Deferred.succeed(result.done, result.info)
     if (result.scope) yield* Scope.close(result.scope, Exit.void)
+    if (result.token) yield* consume(id, result.token)
     return result.info
   })
 
@@ -412,7 +450,16 @@ export const make = Effect.gen(function* () {
   }).pipe(Effect.withSpan("Job.pendingBackground"))
 
   const completeBackground: Interface["completeBackground"] = Effect.fn("Job.completeBackground")((notificationID) =>
-    kv.remove(`${backgroundPrefix}${notificationID}`),
+    SynchronizedRef.updateEffect(state.jobs, (jobs) =>
+      Effect.gen(function* () {
+        yield* kv.remove(`${backgroundPrefix}${notificationID}`)
+        const entry = [...jobs].find(([, job]) => job.info.notificationID === notificationID)
+        if (!entry || entry[1].info.status === "running") return jobs
+        const next = new Map(jobs)
+        next.delete(entry[0])
+        return next
+      }),
+    ),
   )
 
   return Service.of({

packages/core/test/job.test.ts

@@ -9,7 +9,191 @@ import { testEffect } from "./lib/effect"
 
 const it = testEffect(AppNodeBuilder.build(LayerNode.group([Job.node, KV.node])))
 
+const finishJobs = Effect.fn(function* (count: number) {
+  const jobs = yield* Job.Service
+  const ids: string[] = []
+  for (let index = 0; index < count; index++) {
+    const job = yield* jobs.start({ type: "test", run: Effect.succeed(`output-${index}`) })
+    expect((yield* jobs.wait({ id: job.id })).info?.output).toBe(`output-${index}`)
+    ids.push(job.id)
+  }
+  return ids
+})
+
 describe("Job", () => {
+  it.live("bounds consumed terminal results instead of retaining every completed job", () =>
+    Effect.gen(function* () {
+      const jobs = yield* Job.Service
+      const ids = yield* finishJobs(100)
+      const retained = yield* Effect.forEach(ids, jobs.get)
+      expect(retained.filter((info) => info !== undefined).map((info) => info.id)).toEqual(ids.slice(-25))
+      expect(yield* jobs.wait({ id: ids[0] })).toEqual({ timedOut: false })
+    }),
+  )
+
+  it.live("preserves running and unconsumed results until a caller receives them", () =>
+    Effect.gen(function* () {
+      const jobs = yield* Job.Service
+      const running = yield* jobs.start({ type: "test", run: Effect.never })
+      const unread = yield* jobs.start({ type: "test", run: Effect.succeed("not received yet") })
+      expect(yield* jobs.wait({ id: running.id, timeout: 0 })).toMatchObject({ timedOut: true })
+      yield* finishJobs(100)
+      expect(yield* jobs.get(running.id)).toMatchObject({ status: "running" })
+      expect(yield* jobs.get(unread.id)).toMatchObject({ status: "completed", output: "not received yet" })
+      expect(yield* jobs.block({ id: unread.id, sessionID: SessionSchema.ID.make("ses_late_waiter") })).toMatchObject({
+        type: "finished",
+        info: { output: "not received yet" },
+      })
+      yield* finishJobs(25)
+      expect(yield* jobs.get(unread.id)).toBeUndefined()
+      yield* jobs.cancel(running.id)
+    }),
+  )
+
+  it.live("bounds results received by foreground blocking callers", () =>
+    Effect.gen(function* () {
+      const jobs = yield* Job.Service
+      const sessionID = SessionSchema.ID.make("ses_foreground_churn")
+      const ids: string[] = []
+      for (let index = 0; index < 100; index++) {
+        const latch = yield* Deferred.make<void>()
+        const job = yield* jobs.start({ type: "shell", run: Deferred.await(latch).pipe(Effect.as("done")) })
+        const waiter = yield* jobs
+          .block({ id: job.id, sessionID })
+          .pipe(Effect.forkIn(yield* Scope.Scope, { startImmediately: true }))
+        yield* Deferred.succeed(latch, undefined)
+        expect(yield* Fiber.join(waiter)).toMatchObject({ type: "finished", info: { output: "done" } })
+        ids.push(job.id)
+      }
+      const retained = yield* Effect.forEach(ids, jobs.get)
+      expect(retained.filter((info) => info !== undefined).map((info) => info.id)).toEqual(ids.slice(-25))
+    }),
+  )
+
+  it.live("evicts consumed errors and cancellations along with successful results", () =>
+    Effect.gen(function* () {
+      const jobs = yield* Job.Service
+      const failed = yield* jobs.start({ type: "test", run: Effect.fail(new Error("failed")) })
+      const cancelled = yield* jobs.start({ type: "test", run: Effect.never })
+      yield* jobs.cancel(cancelled.id)
+      expect((yield* jobs.wait({ id: failed.id })).info?.status).toBe("error")
+      expect((yield* jobs.wait({ id: cancelled.id })).info?.status).toBe("cancelled")
+      yield* finishJobs(25)
+      expect(yield* jobs.get(failed.id)).toBeUndefined()
+      expect(yield* jobs.get(cancelled.id)).toBeUndefined()
+    }),
+  )
+
+  it.live("bounds explicitly cancelled jobs even without a subsequent wait", () =>
+    Effect.gen(function* () {
+      const jobs = yield* Job.Service
+      const ids: string[] = []
+      for (let index = 0; index < 50; index++) {
+        const job = yield* jobs.start({ type: "test", run: Effect.never })
+        expect((yield* jobs.cancel(job.id))?.status).toBe("cancelled")
+        ids.push(job.id)
+      }
+      const retained = yield* Effect.forEach(ids, jobs.get)
+      expect(retained.filter((info) => info !== undefined).map((info) => info.id)).toEqual(ids.slice(-25))
+    }),
+  )
+
+  it.live("preserves the recoverable wait-to-background handoff across consumed history eviction", () =>
+    Effect.gen(function* () {
+      const jobs = yield* Job.Service
+      const job = yield* jobs.start({
+        type: "shell",
+        recovery: {
+          kind: "shell",
+          sessionID: SessionSchema.ID.make("ses_late_background"),
+          shellID: "sh_late_background",
+          command: "exit 1",
+        },
+        run: Effect.fail(new Error("immediate failure")),
+      })
+      expect((yield* jobs.wait({ id: job.id })).info?.error).toBe("immediate failure")
+      yield* finishJobs(100)
+      const background = yield* jobs.background(job.id)
+      expect(background?.notificationID).toStartWith("msg_")
+      expect((yield* jobs.pendingBackground).find((item) => item.id === job.id)).toMatchObject({
+        status: "error",
+        error: "immediate failure",
+      })
+    }),
+  )
+
+  it.live("keeps previously registered observers' results valid after cache eviction", () =>
+    Effect.gen(function* () {
+      const jobs = yield* Job.Service
+      const scope = yield* Scope.Scope
+      const latch = yield* Deferred.make<void>()
+      const job = yield* jobs.start({ type: "test", run: Deferred.await(latch).pipe(Effect.as("shared output")) })
+      const waiters = yield* Effect.forEach([0, 1, 2], () =>
+        jobs.wait({ id: job.id }).pipe(Effect.forkIn(scope, { startImmediately: true })),
+      )
+      yield* Deferred.succeed(latch, undefined)
+      yield* finishJobs(100)
+      expect(yield* jobs.get(job.id)).toBeUndefined()
+      for (const waiter of waiters) {
+        expect(yield* Fiber.join(waiter)).toMatchObject({ info: { status: "completed", output: "shared output" } })
+      }
+    }),
+  )
+
+  it.live("protects pending background notifications from churn and releases them on acknowledgment", () =>
+    Effect.gen(function* () {
+      const jobs = yield* Job.Service
+      const job = yield* jobs.start({
+        type: "shell",
+        recovery: {
+          kind: "shell",
+          sessionID: SessionSchema.ID.make("ses_pending_notification"),
+          shellID: "sh_pending_notification",
+          command: "echo done",
+        },
+        run: Effect.succeed("background output"),
+      })
+      const background = yield* jobs.background(job.id)
+      if (!background?.notificationID) return yield* Effect.die("background marker missing")
+      yield* jobs.wait({ id: job.id })
+      yield* finishJobs(100)
+      expect(yield* jobs.get(job.id)).toMatchObject({ status: "completed", output: "background output" })
+      expect((yield* jobs.pendingBackground).find((item) => item.id === job.id)).toMatchObject({
+        output: "background output",
+      })
+      yield* jobs.completeBackground(background.notificationID)
+      yield* jobs.completeBackground(background.notificationID)
+      expect(yield* jobs.get(job.id)).toBeUndefined()
+      expect((yield* jobs.pendingBackground).find((item) => item.id === job.id)).toBeUndefined()
+    }),
+  )
+
+  it.live("does not remove a newer generation when an older notification is acknowledged", () =>
+    Effect.gen(function* () {
+      const jobs = yield* Job.Service
+      const id = "job_reused_notification"
+      yield* jobs.start({
+        id,
+        type: "subagent",
+        recovery: {
+          kind: "subagent",
+          parentSessionID: SessionSchema.ID.make("ses_generation_parent"),
+          childSessionID: SessionSchema.ID.make("ses_generation_child"),
+          agent: "explore",
+          description: "first generation",
+        },
+        run: Effect.succeed("old output"),
+      })
+      const background = yield* jobs.background(id)
+      if (!background?.notificationID) return yield* Effect.die("background marker missing")
+      yield* jobs.wait({ id })
+      yield* jobs.start({ id, type: "subagent", run: Effect.succeed("new output") })
+      yield* jobs.completeBackground(background.notificationID)
+      yield* finishJobs(100)
+      expect((yield* jobs.wait({ id })).info?.output).toBe("new output")
+    }),
+  )
+
   it.live("tracks process-local work through explicit observation", () =>
     Effect.gen(function* () {
       const jobs = yield* Job.Service