Feat/task memory evidence pipeline

packages/agent-runtime/src/tools/handlers/tool/check-background-agent.ts

@@ -17,7 +17,11 @@ import type {
   CodebuffToolOutput,
 } from '@codebuff/common/tools/list'
 import type { AgentState } from '@codebuff/common/types/session-state'
-import type { JobEvent } from '@codebuff/common/util/job-registry'
+import type {
+  JobEvent,
+  JobSnapshot,
+  WaitJobResult,
+} from '@codebuff/common/util/job-registry'
 
 type ToolName = 'check_background_agent'
 
@@ -89,6 +93,54 @@ function eventToSearchString(event: JobEvent): string {
   return chunkType
 }
 
+/**
+ * Build the structured poll/follow result from the raw wait result and the
+ * wait_for predicate. Computes the common output fields (state, events,
+ * nextCursor, truncated, dropped, matched, timedOut) shared by poll and
+ * follow modes. Advances the per-consumer cursor for cursorless polls as a
+ * side effect. Agent-specific fields (result/error/cancelled) are resolved
+ * separately by the caller from the core/view.
+ */
+function buildAgentPollResult(params: {
+  /** Poll mode yields a JobSnapshot; follow mode yields a WaitJobResult. */
+  result: WaitJobResult | JobSnapshot | undefined
+  predicate: ((event: JobEvent) => boolean) | undefined
+  fallbackState: string
+  jobId: string
+  consumerId: string
+  cursorOmitted: boolean
+}): {
+  state: string
+  events: JobEvent[]
+  nextCursor: number
+  truncated: boolean
+  dropped: number
+  matched: boolean | undefined
+  timedOut: boolean
+} {
+  const { result, predicate, fallbackState, jobId, consumerId, cursorOmitted } =
+    params
+  const events = result?.events ?? []
+  // Falls back to 0 rather than echoing the caller's cursor: the owned job
+  // always yields a result here, and reporting a cursor the core did not
+  // confirm could pin a consumer past every future event.
+  const nextCursor = result?.nextCursor ?? 0
+  // Only a cursorless poll owns the stored position, and it advances only as
+  // far as the core confirmed: a follow-mode timeout that returned no events
+  // confirms the cursor it started from, so events that land later are still
+  // delivered to this consumer.
+  if (cursorOmitted) {
+    advanceBackgroundAgentConsumerCursor(jobId, consumerId, nextCursor)
+  }
+  const state = result?.state ?? fallbackState
+  const dropped = result?.dropped ?? 0
+  const truncated =
+    result && 'truncated' in result ? result.truncated : dropped > 0
+  const matched = predicate ? events.some(predicate) : undefined
+  const timedOut = result && 'timedOut' in result ? result.timedOut : false
+  return { state, events, nextCursor, truncated, dropped, matched, timedOut }
+}
+
 /**
  * The single not_found message shape, used for an id the unified core no
  * longer knows about (never allocated, or reclaimed by the settled-job TTL
@@ -223,24 +275,15 @@ export const handleCheckBackgroundAgent = (async ({
       })
     : (snapshotBackgroundAgentJob(jobId, effectiveCursor) ?? undefined)
 
-  const events = result?.events ?? []
-  // Falls back to 0 rather than echoing the caller's cursor: the owned job
-  // always yields a result here, and reporting a cursor the core did not
-  // confirm could pin a consumer past every future event.
-  const nextCursor = result?.nextCursor ?? 0
-  // Only a cursorless poll owns the stored position, and it advances only as
-  // far as the core confirmed: a follow-mode timeout that returned no events
-  // confirms the cursor it started from, so events that land later are still
-  // delivered to this consumer.
-  if (cursorOmitted) {
-    advanceBackgroundAgentConsumerCursor(jobId, consumerId, nextCursor)
-  }
-  const state = result?.state ?? owned.job.state
-  const dropped = result?.dropped ?? 0
-  const truncated =
-    result && 'truncated' in result ? result.truncated : dropped > 0
-  const matched = predicate ? events.some(predicate) : undefined
-  const timedOut = result && 'timedOut' in result ? result.timedOut : false
+  const { state, events, nextCursor, truncated, dropped, matched, timedOut } =
+    buildAgentPollResult({
+      result,
+      predicate,
+      fallbackState: owned.job.state,
+      jobId,
+      consumerId,
+      cursorOmitted,
+    })
 
   // Both the settled error and the settled result are folded into the core
   // lifecycle, so they are resolved from the core first and fall back to the

sdk/src/__tests__/check-job.test.ts

@@ -22,6 +22,7 @@ import {
   appendBoundedCollected,
   checkJob,
 } from '../tools/check-job'
+import { listJobs } from '../tools/list-jobs'
 import {
   SETTLED_JOB_TTL_MS,
   jobRegistry,
@@ -146,8 +147,7 @@ describe('readNewJobOutput', () => {
     }
 
     expect(peekJobLineCarry(job)).toBe('')
-    const registryId = job.registryJobId ?? job.jobId
-    const snapshot = jobRegistry.snapshot(registryId, 0)
+    const snapshot = jobRegistry.snapshot(job.jobId, 0)
     const outputEvents = (snapshot?.events ?? []).filter(
       (event) => event.payload.type === 'output',
     )
@@ -249,20 +249,19 @@ describe('checkJob', () => {
     expect(job.readOffset).toBe(offsetAfterLiveDrain)
   })
 
-  test('wait_for matches only via peekJobLineCarry for an unterminated partial line', async () => {
-    // Documented live-drainer match-vs-events lag: a needle drained into
-    // lineCarry (no trailing newline yet) is matchable via peekJobLineCarry
-    // even though no complete per-line registry `output` event has been
-    // emitted for it yet. matched can be true while events/outputText lag.
+  test('wait_for matches via peekJobLineCarry for an unterminated partial line and force-emits the carry', async () => {
+    // A needle drained into lineCarry (no trailing newline yet) is matchable
+    // via peekJobLineCarry. When the match is found in the carry, checkJob
+    // force-emits the carry as a registry output event so the returned events
+    // are consistent with matched: true — the needle appears in outputText.
     const job = makeJob()
     const partial = 'Ready > Listening on :3000'
     fs.appendFileSync(job.logFile, partial)
 
     expect(readNewJobOutput(job)).toBe(partial)
     expect(peekJobLineCarry(job)).toContain('Listening on')
 
-    const registryId = job.registryJobId ?? job.jobId
-    const preMatchSnapshot = jobRegistry.snapshot(registryId, 0)
+    const preMatchSnapshot = jobRegistry.snapshot(job.jobId, 0)
     const preMatchOutput = (preMatchSnapshot?.events ?? [])
       .filter((event) => event.payload.type === 'output')
       .map((event) =>
@@ -281,9 +280,62 @@ describe('checkJob', () => {
     expect(result.matched).toBe(true)
     expect(result.state).toBe('running')
     expect(result.timedOut).toBeUndefined()
-    // Carry still holds the unterminated needle; events need not include it.
-    expect(peekJobLineCarry(job)).toContain('Listening on')
-    expect(outputText(result)).not.toContain('Listening on')
+    // Carry was force-emitted as a registry output event, so the needle is
+    // now present in the returned events/outputText and the carry is cleared.
+    expect(peekJobLineCarry(job)).toBe('')
+    expect(outputText(result)).toContain('Listening on')
+  })
+
+  test('wait_for force-emits the carry when the needle spans the chunk/carry boundary', async () => {
+    // A needle that spans the boundary between chunk (already emitted to the
+    // registry) and carry (not yet emitted) must still be fully present in the
+    // returned events. The carry must be force-emitted so the returned events
+    // are consistent with matched: true.
+    const job = makeJob()
+    // Write a complete line followed by a partial line (no newline at end).
+    // After draining, the complete line is in the registry (chunk) and the
+    // partial line is in the carry.
+    fs.appendFileSync(job.logFile, 'foo bar\nbaz')
+    expect(readNewJobOutput(job)).toBe('foo bar\nbaz')
+    expect(peekJobLineCarry(job)).toBe('baz')
+
+    const result = value(
+      await checkJob({
+        jobId: job.jobId,
+        wait_for: 'bar\nbaz',
+        owner: TRUSTED_OWNER,
+      }),
+    )
+    expect(result.matched).toBe(true)
+    expect(result.state).toBe('running')
+    // The carry was force-emitted, so the needle is fully present in the
+    // returned events.
+    expect(peekJobLineCarry(job)).toBe('')
+    expect(outputText(result)).toContain('bar\nbaz')
+  })
+
+  test('wait_for does not force-emit the carry when the needle is fully in the chunk', async () => {
+    // When the needle is fully present in chunk (already emitted to the
+    // registry), the carry must NOT be force-emitted. This keeps the original
+    // path unchanged for the common case and verifies the new boundary check
+    // only flushes the carry when the needle depends on it.
+    const job = makeJob()
+    fs.appendFileSync(job.logFile, 'prefix\nListening on :3000\nsuffix')
+    expect(readNewJobOutput(job)).toBe('prefix\nListening on :3000\nsuffix')
+    expect(peekJobLineCarry(job)).toBe('suffix')
+
+    const result = value(
+      await checkJob({
+        jobId: job.jobId,
+        wait_for: 'Listening on :3000',
+        owner: TRUSTED_OWNER,
+      }),
+    )
+    expect(result.matched).toBe(true)
+    expect(result.state).toBe('running')
+    // Carry must be untouched because the needle was fully in the chunk.
+    expect(peekJobLineCarry(job)).toBe('suffix')
+    expect(outputText(result)).toContain('Listening on :3000')
   })
 
   test('follow mode returns matched=true once the pattern is present', async () => {
@@ -970,12 +1022,10 @@ describe('checkJob', () => {
     // The unified jobRegistry is now the source of truth for live
     // state/ownership (the pending-background-jobs mirror is the legacy
     // store M4 removes). A recovered job is re-emitted into the registry
-    // under a fresh registry id stored on the adapter object, carrying the
-    // preserved owner.
+    // under the disk-derived jobId (passed as explicit registry id), carrying
+    // the preserved owner.
     const recovered = getBackgroundJob(jobId)
-    const registryJob = recovered?.registryJobId
-      ? jobRegistry.get(recovered.registryJobId)
-      : undefined
+    const registryJob = jobRegistry.get(jobId)
     expect(registryJob?.owner).toEqual(owner)
   })
 

sdk/src/__tests__/list-jobs.test.ts

@@ -301,17 +301,17 @@ describe('listJobs', () => {
     expect(entry!.tail).toEqual(emitted.slice(-10))
   })
 
-  test('dual-id remapped process job: list_jobs exposes user jobId; pending uses adapter cursor; listed id works with check_job', async () => {
-    // Recovered / __registerJobForTest jobs remap: adapter Map key = user/
-    // disk jobId, registryJobId = fresh registry id. list_jobs must reverse-
-    // resolve the adapter, emit the user-facing id, and honor lastCheckCursor
-    // so rediscovery works with check_job/kill_job.
-    const userJobId = 'job-dual-id-user'
+  test('registered process job: list_jobs exposes jobId; pending uses adapter cursor; listed id works with check_job', async () => {
+    // After the id collapse, __registerJobForTest passes the adapter's jobId
+    // as the explicit registry id, so the registry record and adapter Map
+    // share one key. list_jobs must resolve the adapter, emit the id, and
+    // honor lastCheckCursor so rediscovery works with check_job/kill_job.
+    const userJobId = 'job-registered-user'
     const logFile = path.join(os.tmpdir(), `openbuff-${userJobId}.log`)
     fs.writeFileSync(logFile, '')
     const adapter: BackgroundJob = {
       jobId: userJobId,
-      command: 'dual-id-cmd',
+      command: 'registered-cmd',
       child: { pid: 4242 } as BackgroundJob['child'],
       logFile,
       metadataFile: path.join(os.tmpdir(), `openbuff-${userJobId}.json`),
@@ -322,16 +322,15 @@ describe('listJobs', () => {
       owner,
     }
     __registerJobForTest(adapter)
-    const registryJobId = adapter.registryJobId
-    expect(registryJobId).toBeDefined()
-    expect(registryJobId).not.toBe(userJobId)
+    // After collapse: registry id === adapter.jobId (no remapping).
+    expect(adapter.jobId).toBe(userJobId)
 
-    // Emit on the *registry* id (where process output is mirrored).
-    jobRegistry.emit(registryJobId!, { type: 'output', data: 'line-1\n' })
-    jobRegistry.emit(registryJobId!, { type: 'output', data: 'line-2\n' })
-    jobRegistry.emit(registryJobId!, { type: 'output', data: 'line-3\n' })
+    // Emit on the jobId (where process output is mirrored).
+    jobRegistry.emit(userJobId, { type: 'output', data: 'line-1\n' })
+    jobRegistry.emit(userJobId, { type: 'output', data: 'line-2\n' })
+    jobRegistry.emit(userJobId, { type: 'output', data: 'line-3\n' })
 
-    // Advance lastCheckCursor via check_job on the user-facing id.
+    // Advance lastCheckCursor via check_job on the id.
     const checkResult = (await checkJob({ jobId: userJobId, owner }))[0]
       .value as {
       jobId: string
@@ -348,19 +347,18 @@ describe('listJobs', () => {
       gap: boolean
       command: string
     }>
-    // Must expose user-facing id, never the internal remapped registry id.
+    // Must expose the id.
     expect(listed.map((j) => j.jobId)).toContain(userJobId)
-    expect(listed.map((j) => j.jobId)).not.toContain(registryJobId)
 
     const entry = listed.find((j) => j.jobId === userJobId)
-    expect(entry?.command).toBe('dual-id-cmd')
+    expect(entry?.command).toBe('registered-cmd')
     // Cursor advanced by check_job → pending none at adapter cursor; no gap.
     expect(entry?.pending).toBe('none')
     expect(entry?.gap).toBe(false)
 
-    // New output after the listed cursor should re-bucket pending, still under
-    // the user-facing id, and check_job must still resolve that listed id.
-    jobRegistry.emit(registryJobId!, { type: 'output', data: 'line-4\n' })
+    // New output after the listed cursor should re-bucket pending, and
+    // check_job must still resolve that listed id.
+    jobRegistry.emit(userJobId, { type: 'output', data: 'line-4\n' })
     const afterMore = (
       value(await listJobs({ owner })).jobs as Array<{
         jobId: string