Merge pull request #4149 from deepseek-harness/worktree/projection-cache-drain-20260914

test: wait for cache writes and workflow child starts

packages/session/session-projection-cache/tests/cache.spec.ts

@@ -148,6 +148,18 @@ const mark = (session: Session, marks: string[]): SessionEvent =>
 const endTurn = (session: Session): SessionEvent =>
   session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
 
+/** Resolve after this Session's next durable cache replacement. */
+function whenWritten(ctx: Context, id: SessionId): Promise<void> {
+  return new Promise((resolve) => {
+    const dispose = ctx.on('domain/changed', (change) => {
+      if (change.domain !== projectionCacheDomainSpec.name
+        || change.table !== 'sessions' || change.key !== id || change.operation !== 'put') return
+      dispose()
+      resolve()
+    })
+  })
+}
+
 /** The stored record for one session id (undefined = absent or unreadable). */
 async function storedRecord(root: string, id: Session['id']): Promise<CheckpointRecord | undefined> {
   try {
@@ -189,18 +201,20 @@ afterEach(async () => {
 describe('SessionProjectionCache write policy', () => {
   it('writes a durable checkpoint at turn/end (mandatory point)', async () => {
     const { ctx, root } = await harness()
-    const session = ctx.sessions.create(SessionId('turn-end'))
+    // The interval cannot substitute for the mandatory turn/end trigger.
+    vi.useFakeTimers({ toFake: ['setTimeout', 'clearTimeout'] })
+    const id = SessionId('turn-end')
+    const created = whenWritten(ctx, id)
+    const session = ctx.sessions.create(id)
     mark(session, ['a'])
-    // Creation already wrote the init cut; the mark is throttled, so the
-    // stored row is still the creation-time cut (no marks folded).
-    await vi.waitFor(async () => {
-      expect((await storedRows(root, session.id))?.['cache-test/marks']?.seq).toBe(-1)
-    }, { timeout: 5_000 })
+    // The mark is throttled, so the creation cut has no marks folded.
+    await created
+    expect((await storedRows(root, session.id))?.['cache-test/marks']?.seq).toBe(-1)
+    const written = whenWritten(ctx, id)
     const end = endTurn(session)
-    await vi.waitFor(async () => {
-      expect((await storedRows(root, session.id))?.['cache-test/marks'])
-        .toEqual({ ver: 1, seq: end.seq, val: { marks: ['a'] } })
-    }, { timeout: 5_000 })
+    await written
+    expect((await storedRows(root, session.id))?.['cache-test/marks'])
+      .toEqual({ ver: 1, seq: end.seq, val: { marks: ['a'] } })
   })
 
   it('writes a checkpoint at session creation, capturing the seed-derived cut', async () => {
@@ -208,44 +222,50 @@ describe('SessionProjectionCache write policy', () => {
     // A forked child seeded with its ancestor's title-like event: no
     // conversation follows, yet the creation write must capture the fold so
     // a crash or a live-held fork still lists the derived value.
-    const session = ctx.sessions.create(SessionId('seeded'), {
+    const id = SessionId('seeded')
+    const created = whenWritten(ctx, id)
+    const session = ctx.sessions.create(id, {
       seed: [{ type: 'cache-test/mark', seq: 0, time: 1, data: { marks: ['seed'] } }] as SessionEvent[],
     })
-    await vi.waitFor(async () => {
-      expect((await storedRows(root, session.id))?.['cache-test/marks']?.val)
-        .toEqual({ marks: ['seed'] })
-    }, { timeout: 5_000 })
+    await created
+    expect((await storedRows(root, session.id))?.['cache-test/marks']?.val)
+      .toEqual({ marks: ['seed'] })
   })
 
   it('writes at session disposal (detach, the live-to-cold moment)', async () => {
     const { ctx, root } = await harness()
+    vi.useFakeTimers({ toFake: ['setTimeout', 'clearTimeout'] })
+    const id = SessionId('detach')
+    const created = whenWritten(ctx, id)
     // Sessions dispose with their owning fiber: create in a child plugin.
     let session: Session | undefined
     const owner = await ctx.plugin(Object.assign((inner: Context) => {
-      session = inner.sessions.create(SessionId('detach'))
+      session = inner.sessions.create(id)
     }, { inject: ['sessions'] }))
     if (session === undefined) throw new Error('session was not created')
+    await created
     mark(session, ['live'])
+    const written = whenWritten(ctx, id)
     await owner.dispose()
-    const detached = session
-    await vi.waitFor(async () => {
-      expect((await storedRows(root, detached.id))?.['cache-test/marks']?.val).toEqual({ marks: ['live'] })
-    }, { timeout: 5_000 })
+    await written
+    expect((await storedRows(root, id))?.['cache-test/marks']?.val).toEqual({ marks: ['live'] })
   })
 
   it('flushes when the in-turn event count reaches the configured threshold', async () => {
     const { ctx, root } = await harness({ config: { writeEveryEvents: 3, writeIntervalMs: 60_000 } })
-    const session = ctx.sessions.create(SessionId('count'))
+    vi.useFakeTimers({ toFake: ['setTimeout', 'clearTimeout'] })
+    const id = SessionId('count')
+    const created = whenWritten(ctx, id)
+    const session = ctx.sessions.create(id)
     mark(session, ['1'])
     mark(session, ['2'])
-    await vi.waitFor(async () => {
-      expect((await storedRows(root, session.id))?.['cache-test/marks'])
-        .toEqual({ ver: 1, seq: -1, val: null }) // still the creation cut
-    }, { timeout: 5_000 })
+    await created
+    expect((await storedRows(root, id))?.['cache-test/marks'])
+      .toEqual({ ver: 1, seq: -1, val: null }) // still the creation cut
+    const written = whenWritten(ctx, id)
     mark(session, ['3'])
-    await vi.waitFor(async () => {
-      expect((await storedRows(root, session.id))?.['cache-test/marks']?.val).toEqual({ marks: ['3'] })
-    }, { timeout: 5_000 })
+    await written
+    expect((await storedRows(root, id))?.['cache-test/marks']?.val).toEqual({ marks: ['3'] })
   })
 
   it('flushes on the configured interval when the count threshold is not reached', async () => {
@@ -320,16 +340,15 @@ describe('SessionProjectionCache write policy', () => {
     await vi.waitFor(() => {
       expect(warn).toHaveBeenCalledWith(expect.stringContaining('turn/end write for "fail-soft" failed'))
     }, { timeout: 5_000 })
-    await vi.waitFor(async () => {
-      expect(await storedRows(root, session.id)).toBeUndefined()
-    }, { timeout: 5_000 })
+    expect(await storedRows(root, session.id)).toBeUndefined()
     // Self-heal: once the blocker clears, the next mandatory point writes.
     await rm(recordPath(root, session.id), { recursive: true })
+    vi.useFakeTimers({ toFake: ['setTimeout', 'clearTimeout'] })
     mark(session, ['y'])
+    const written = whenWritten(ctx, session.id)
     endTurn(session)
-    await vi.waitFor(async () => {
-      expect((await storedRows(root, session.id))?.['cache-test/marks']?.val).toEqual({ marks: ['y'] })
-    }, { timeout: 5_000 })
+    await written
+    expect((await storedRows(root, session.id))?.['cache-test/marks']?.val).toEqual({ marks: ['y'] })
   })
 })
 
@@ -665,14 +684,6 @@ describe('SessionProjectionCache cold-read seeding', () => {
     })
     const { cache, ctx } = await harness({ root })
     const apply = vi.fn((_state: number, _event: SessionEvent) => 1)
-    const whenWritten = (id: SessionId): Promise<void> => new Promise((resolve) => {
-      const dispose = ctx.on('domain/changed', (change) => {
-        if (change.domain !== projectionCacheDomainSpec.name
-          || change.table !== 'sessions' || change.key !== id || change.operation !== 'put') return
-        dispose()
-        resolve()
-      })
-    })
     ctx.sessionProjections.register({
       key: 'cache-test/count',
       stateSchema: z.number().int().nonnegative(),
@@ -684,7 +695,7 @@ describe('SessionProjectionCache cold-read seeding', () => {
     const events = Array.from({ length: 5 }, (_, seq) => ({
       type: 'cache-test/mark', seq: SessionSeq(seq), time: seq, data: { marks: [`m${seq}`] },
     })) as SessionEvent[]
-    const refreshed = whenWritten(meta.id)
+    const refreshed = whenWritten(ctx, meta.id)
     const snapshot = cache.coldSnapshot(meta, SessionLogOffset(0), events)
     // The full log was traversed, but the fold applied only seqs 3 and 4.
     expect(apply).toHaveBeenCalledTimes(2)
@@ -698,7 +709,7 @@ describe('SessionProjectionCache cold-read seeding', () => {
     // No cached row yet: the first cold read folds from init over the full
     // log and creates the cache row (the `?? {}` seed path).
     const fresh = headerOf(SessionId('cold-fresh'), 10)
-    const created = whenWritten(fresh.id)
+    const created = whenWritten(ctx, fresh.id)
     cache.coldSnapshot(fresh, SessionLogOffset(0), events)
     expect(apply).toHaveBeenCalledTimes(7) // 2 tail + 5 full
     await created

packages/workflow/workflow-ptc/tests/workflow-ptc.spec.ts

@@ -582,12 +582,15 @@ describe('dsh-workflow-ptc', { timeout: 120_000 }, () => {
 
     it('cancel() aborts in-flight children and settles after their cleanup', async () => {
       const { ctx, parent, provider } = await setup({ manual: true })
+      const starts: number[] = []
+      ctx.on('workflow/agent-start', (_info, agent) => { starts.push(agent.seq) })
       const ends: unknown[] = []
       ctx.on('workflow/agent-end', (_info, agent) => { ends.push(agent) })
       const runEnds: WorkflowResultInfo[] = []
       ctx.on('workflow/end', (_info, result) => { runEnds.push(result) })
       const handle = ctx.workflowEngine.start({ ...scripted("return await agent('long job')"), parent })
-      await waitFor(() => { expect(provider.runs.length).toBe(1) })
+      // Cancellation emits child ends only for starts already observed by the host.
+      await waitFor(() => { expect(starts).toHaveLength(1) })
       handle.cancel('user stopped it')
       const result = await handle.result
       expect(result.stopReason).toBe('cancelled')
@@ -851,7 +854,9 @@ describe('dsh-workflow-ptc', { timeout: 120_000 }, () => {
     })
 
     it('cancellation emits exactly one agent-end per observed start before workflow/end', async () => {
-      const { ctx, parent, provider } = await setup({ manual: true })
+      const { ctx, parent } = await setup({ manual: true })
+      const starts: number[] = []
+      ctx.on('workflow/agent-start', (_info, agent) => { starts.push(agent.seq) })
       const ends: { seq: number; outcome: string }[] = []
       const order: string[] = []
       ctx.on('workflow/agent-end', (_info, agent) => {
@@ -863,7 +868,7 @@ describe('dsh-workflow-ptc', { timeout: 120_000 }, () => {
         ...scripted("await parallel([() => agent('a'), () => agent('b')])\nreturn 'unreachable'"),
         parent,
       })
-      await waitFor(() => { expect(provider.runs.length).toBe(2) })
+      await waitFor(() => { expect(starts).toHaveLength(2) })
       handle.cancel('user stop')
       const result = await handle.result
       expect(result.stopReason).toBe('cancelled')