diff --git a/packages/agent-runtime/src/tools/handlers/tool/check-background-agent.ts b/packages/agent-runtime/src/tools/handlers/tool/check-background-agent.ts
index f181366639..0651b21b6f 100644
--- a/packages/agent-runtime/src/tools/handlers/tool/check-background-agent.ts
+++ b/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
diff --git a/sdk/src/__tests__/check-job.test.ts b/sdk/src/__tests__/check-job.test.ts
index 58e5c2daa8..30b8f93dc6 100644
--- a/sdk/src/__tests__/check-job.test.ts
+++ b/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,11 +249,11 @@ 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)
@@ -261,8 +261,7 @@ describe('checkJob', () => {
     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)
   })
 
diff --git a/sdk/src/__tests__/list-jobs.test.ts b/sdk/src/__tests__/list-jobs.test.ts
index bf2ab310a4..1a2be12c8d 100644
--- a/sdk/src/__tests__/list-jobs.test.ts
+++ b/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
diff --git a/sdk/src/__tests__/run-terminal-command.test.ts b/sdk/src/__tests__/run-terminal-command.test.ts
index 1256799d6f..92099e57ba 100644
--- a/sdk/src/__tests__/run-terminal-command.test.ts
+++ b/sdk/src/__tests__/run-terminal-command.test.ts
@@ -2,11 +2,17 @@ import { spawnSync } from 'child_process'
 import fs from 'fs'
 import os from 'os'
 import path from 'path'
-import { describe, expect, it } from 'bun:test'
+import { afterEach, describe, expect, it, spyOn } from 'bun:test'
 
+import { jobRegistry } from '@codebuff/common/util/job-registry'
 import { getOwnedTempRoots } from '@codebuff/common/util/project-path-containment'
 
-import { getBackgroundJob, killBackgroundJob } from '../tools/background-jobs'
+import {
+  __clearJobsForTest,
+  getBackgroundJob,
+  killBackgroundJob,
+  startBackgroundJob,
+} from '../tools/background-jobs'
 import {
   findWindowsBash,
   runTerminalCommand,
@@ -83,55 +89,67 @@ describe('runTerminalCommand cwd containment', () => {
       parentRunId: 'parent-1',
       parentAgentId: 'agent-1',
     }
-    const result = await runTerminalCommand({
-      command: 'sleep 30',
-      process_type: 'BACKGROUND',
-      cwd: process.cwd(),
-      projectRoot: process.cwd(),
-      timeout_seconds: 5,
-      signal: controller.signal,
-      owner,
-    })
-    const value = result[0].value as { jobId?: string; detached?: boolean }
-    expect(value.detached).toBe(false)
-    expect(value.jobId).toBeDefined()
-
-    const runningJob = getBackgroundJob(value.jobId!)
-    expect(runningJob?.owner).toEqual(owner)
-    expect(
-      JSON.parse(fs.readFileSync(runningJob!.metadataFile, 'utf8')).owner,
-    ).toEqual(owner)
-
-    controller.abort()
-    await new Promise((resolve) => setTimeout(resolve, 20))
-    const job = getBackgroundJob(value.jobId!)
-    // An abort-initiated kill is an intentional stop, recorded as 'stopped'
-    // (distinct from an error/non-zero natural exit).
-    expect(job?.status).toBe('stopped')
-    killBackgroundJob(value.jobId!, 'SIGKILL')
+    let value: { jobId?: string; detached?: boolean } | undefined
+    try {
+      const result = await runTerminalCommand({
+        command: 'sleep 30',
+        process_type: 'BACKGROUND',
+        cwd: process.cwd(),
+        projectRoot: process.cwd(),
+        timeout_seconds: 5,
+        signal: controller.signal,
+        owner,
+      })
+      value = result[0].value as { jobId?: string; detached?: boolean }
+      expect(value.detached).toBe(false)
+      expect(value.jobId).toBeDefined()
+
+      const runningJob = getBackgroundJob(value.jobId!)
+      expect(runningJob?.owner).toEqual(owner)
+      expect(
+        JSON.parse(fs.readFileSync(runningJob!.metadataFile, 'utf8')).owner,
+      ).toEqual(owner)
+
+      controller.abort()
+      await new Promise((resolve) => setTimeout(resolve, 20))
+      const job = getBackgroundJob(value.jobId!)
+      // An abort-initiated kill is an intentional stop, recorded as 'stopped'
+      // (distinct from an error/non-zero natural exit).
+      expect(job?.status).toBe('stopped')
+    } finally {
+      if (value?.jobId !== undefined) {
+        killBackgroundJob(value.jobId, 'SIGKILL')
+      }
+    }
   })
 
   it('terminates a background job that exceeds the bounded log quota', async () => {
-    const result = await runTerminalCommand({
-      command: 'yes x | head -c 12000000; sleep 30',
-      process_type: 'BACKGROUND',
-      cwd: process.cwd(),
-      projectRoot: process.cwd(),
-      timeout_seconds: 5,
-    })
-    const value = result[0].value as { jobId?: string }
-    expect(value.jobId).toBeDefined()
-
-    const deadline = Date.now() + 5_000
-    let job = getBackgroundJob(value.jobId!)
-    while (job?.status === 'running' && Date.now() < deadline) {
-      await new Promise((resolve) => setTimeout(resolve, 25))
-      job = getBackgroundJob(value.jobId!)
-    }
+    let value: { jobId?: string } | undefined
+    try {
+      const result = await runTerminalCommand({
+        command: 'yes x | head -c 12000000; sleep 30',
+        process_type: 'BACKGROUND',
+        cwd: process.cwd(),
+        projectRoot: process.cwd(),
+        timeout_seconds: 5,
+      })
+      value = result[0].value as { jobId?: string }
+      expect(value.jobId).toBeDefined()
 
-    expect(job?.status).toBe('error')
-    expect(fs.statSync(job!.logFile).size).toBeLessThanOrEqual(10 * 1024 * 1024)
-    killBackgroundJob(value.jobId!, 'SIGKILL')
+      const deadline = Date.now() + 5_000
+      let job = getBackgroundJob(value.jobId!)
+      while (job?.status === 'running' && Date.now() < deadline) {
+        await new Promise((resolve) => setTimeout(resolve, 25))
+        job = getBackgroundJob(value.jobId!)
+      }
+
+      expect(job?.status).toBe('error')
+      expect(fs.statSync(job!.logFile).size).toBeLessThanOrEqual(10 * 1024 * 1024)
+    } finally {
+      if (value?.jobId !== undefined) {
+        killBackgroundJob(value.jobId, 'SIGKILL')
+      }
+    }
   })
 
   it('accepts the project root itself as cwd', async () => {
@@ -292,28 +310,23 @@ describe('runTerminalCommand cwd containment', () => {
   })
 })
 
-describe('runTerminalCommand SYNC dirty-delta touchedPaths', () => {
-  function initTempGitRepo(): string {
-    const projectRoot = fs.mkdtempSync(
-      path.join(os.tmpdir(), 'terminal-dirty-'),
-    )
-    const run = (args: string[]) =>
-      spawnSync('git', args, {
-        cwd: projectRoot,
-        encoding: 'utf8',
-      })
-    expect(run(['init']).status).toBe(0)
-    run(['config', 'user.email', 'test@example.com'])
-    run(['config', 'user.name', 'Test'])
-    // Optional initial commit keeps porcelain stable across git versions.
-    fs.writeFileSync(path.join(projectRoot, 'README'), 'seed\n')
-    run(['add', 'README'])
-    run(['commit', '-m', 'seed'])
-    return projectRoot
-  }
+function initTempGitRepo(prefix: string): string {
+  const projectRoot = fs.mkdtempSync(path.join(os.tmpdir(), prefix))
+  const run = (args: string[]) =>
+    spawnSync('git', args, { cwd: projectRoot, encoding: 'utf8' })
+  expect(run(['init']).status).toBe(0)
+  run(['config', 'user.email', 'test@example.com'])
+  run(['config', 'user.name', 'Test'])
+  // Optional initial commit keeps porcelain stable across git versions.
+  fs.writeFileSync(path.join(projectRoot, 'README'), 'seed\n')
+  run(['add', 'README'])
+  run(['commit', '-m', 'seed'])
+  return projectRoot
+}
 
+describe('runTerminalCommand SYNC dirty-delta touchedPaths', () => {
   it('reports newly created project files in touchedPaths', async () => {
-    const projectRoot = initTempGitRepo()
+    const projectRoot = initTempGitRepo('terminal-dirty-')
     try {
       // Pre-existing dirt must not appear in the delta.
       fs.writeFileSync(path.join(projectRoot, 'already-dirty.txt'), 'old\n')
@@ -340,7 +353,7 @@ describe('runTerminalCommand SYNC dirty-delta touchedPaths', () => {
   })
 
   it('attributes paths relative to projectRoot when cwd is a subdirectory', async () => {
-    const projectRoot = initTempGitRepo()
+    const projectRoot = initTempGitRepo('terminal-dirty-')
     try {
       const sub = path.join(projectRoot, 'pkg')
       fs.mkdirSync(sub)
@@ -390,26 +403,13 @@ describe('runTerminalCommand SYNC dirty-delta touchedPaths', () => {
 })
 
 describe('runTerminalCommand BACKGROUND dirty snapshot at start', () => {
-  function initTempGitRepo(): string {
-    const projectRoot = fs.mkdtempSync(
-      path.join(os.tmpdir(), 'terminal-bg-dirty-'),
-    )
-    const run = (args: string[]) =>
-      spawnSync('git', args, {
-        cwd: projectRoot,
-        encoding: 'utf8',
-      })
-    expect(run(['init']).status).toBe(0)
-    run(['config', 'user.email', 'test@example.com'])
-    run(['config', 'user.name', 'Test'])
-    fs.writeFileSync(path.join(projectRoot, 'README'), 'seed\n')
-    run(['add', 'README'])
-    run(['commit', '-m', 'seed'])
-    return projectRoot
-  }
-
   it('stores pre-start dirty snapshot on the job without emitting touchedPaths', async () => {
-    const projectRoot = initTempGitRepo()
+    const projectRoot = initTempGitRepo('terminal-bg-dirty-')
+    let value: {
+      jobId?: string
+      touchedPaths?: string[]
+      backgroundProcessStatus?: string
+    } | undefined
     try {
       fs.writeFileSync(path.join(projectRoot, 'already-dirty.txt'), 'old\n')
 
@@ -420,7 +420,7 @@ describe('runTerminalCommand BACKGROUND dirty snapshot at start', () => {
         projectRoot,
         timeout_seconds: 5,
       })
-      const value = result[0].value as {
+      value = result[0].value as {
         jobId?: string
         touchedPaths?: string[]
         backgroundProcessStatus?: string
@@ -436,9 +436,10 @@ describe('runTerminalCommand BACKGROUND dirty snapshot at start', () => {
       expect(job?.dirtyBeforePaths).toContain('already-dirty.txt')
       expect(job?.dirtyBeforePaths).not.toContain('created-by-bg.txt')
       expect(job?.settlementTouchedPaths).toBeUndefined()
-
-      killBackgroundJob(value.jobId!, 'SIGKILL')
     } finally {
+      if (value?.jobId !== undefined) {
+        killBackgroundJob(value.jobId, 'SIGKILL')
+      }
       fs.rmSync(projectRoot, { recursive: true, force: true })
     }
   })
@@ -447,6 +448,7 @@ describe('runTerminalCommand BACKGROUND dirty snapshot at start', () => {
     const projectRoot = fs.mkdtempSync(
       path.join(os.tmpdir(), 'terminal-bg-nongit-'),
     )
+    let value: { jobId?: string; touchedPaths?: string[] } | undefined
     try {
       const result = await runTerminalCommand({
         command: 'sleep 30',
@@ -455,7 +457,7 @@ describe('runTerminalCommand BACKGROUND dirty snapshot at start', () => {
         projectRoot,
         timeout_seconds: 5,
       })
-      const value = result[0].value as {
+      value = result[0].value as {
         jobId?: string
         touchedPaths?: string[]
       }
@@ -465,10 +467,69 @@ describe('runTerminalCommand BACKGROUND dirty snapshot at start', () => {
       const job = getBackgroundJob(value.jobId!)
       expect(job?.projectRoot).toBe(projectRoot)
       expect(job?.dirtyBeforePaths).toBeUndefined()
-
-      killBackgroundJob(value.jobId!, 'SIGKILL')
     } finally {
+      if (value?.jobId !== undefined) {
+        killBackgroundJob(value.jobId, 'SIGKILL')
+      }
       fs.rmSync(projectRoot, { recursive: true, force: true })
     }
   })
 })
+
+describe('startBackgroundJob failed-spawn log file preservation', () => {
+  const FORCED_ID = 'job-test-forced-eexist'
+  const FOREIGN_CONTENT = 'foreign log data that must survive\n'
+  let foreignLogFile: string | undefined
+  let createSpy: { mockRestore(): void } | undefined
+
+  afterEach(() => {
+    createSpy?.mockRestore()
+    createSpy = undefined
+    if (foreignLogFile !== undefined) {
+      fs.rmSync(foreignLogFile, { force: true })
+      foreignLogFile = undefined
+    }
+    __clearJobsForTest()
+  })
+
+  it('preserves a pre-existing foreign log when creation hits EEXIST', () => {
+    foreignLogFile = path.join(os.tmpdir(), `openbuff-${FORCED_ID}.log`)
+    fs.writeFileSync(foreignLogFile, FOREIGN_CONTENT)
+
+    // Force the registry to allocate exactly the id whose log file already
+    // exists, so safeCreateJobLogFile's O_EXCL create fails with EEXIST and
+    // the catch block runs before spawn is ever reached.
+    const realCreate = jobRegistry.create.bind(jobRegistry)
+    const spy = spyOn(jobRegistry, 'create').mockImplementation((opts) => {
+      realCreate({ ...opts, jobId: FORCED_ID })
+      return jobRegistry.get(FORCED_ID)!
+    })
+    createSpy = spy
+
+    try {
+      expect(() =>
+        startBackgroundJob({
+          command: 'sleep 30',
+          shell: 'sh',
+          shellArgs: ['-c'],
+          cwd: process.cwd(),
+          env: { ...process.env },
+          owner: {
+            clientSessionId: 'session-eexist',
+            rootRunId: 'root-eexist',
+            parentRunId: 'parent-eexist',
+            parentAgentId: 'agent-eexist',
+          },
+        }),
+      ).toThrow()
+    } finally {
+      spy.mockRestore()
+      createSpy = undefined
+    }
+
+    // The failed-spawn cleanup must not delete a pre-existing/foreign log
+    // file this spawn never created (logFileCreatedByThisSpawn guard).
+    expect(fs.existsSync(foreignLogFile)).toBe(true)
+    expect(fs.readFileSync(foreignLogFile, 'utf8')).toBe(FOREIGN_CONTENT)
+  })
+})
diff --git a/sdk/src/tools/background-jobs.ts b/sdk/src/tools/background-jobs.ts
index 757026f1ce..34c448eda6 100644
--- a/sdk/src/tools/background-jobs.ts
+++ b/sdk/src/tools/background-jobs.ts
@@ -122,14 +122,6 @@ export interface BackgroundJob {
    * before a group-kill (pid reuse guard); undefined on non-Linux hosts.
    */
   childProcessStartTime?: string
-  /**
-   * Registry-side id backing this job when it differs from `jobId`. The
-   * registry allocates its own ids, so a cross-session-recovered job (whose
-   * `jobId` comes from the on-disk metadata file name) is re-emitted into
-   * the registry under a fresh id recorded here. Jobs spawned by this
-   * process use the registry-issued id directly and leave this undefined.
-   */
-  registryJobId?: string
   /**
    * Project root used for the pre-start dirty snapshot (BACKGROUND start).
    * In-memory only — not written to recovery metadata. Recovered jobs omit it.
@@ -288,6 +280,40 @@ export function safeOpenJobLogForRead(
   }
 }
 
+/**
+ * Truncate a background-job log file to at most `maxBytes` by dropping the
+ * HEAD (oldest bytes) and keeping the TAIL (newest bytes). The newest output
+ * is the most diagnostically useful for a job terminated due to log quota —
+ * errors, build failures, and recent status messages live there. No-op when
+ * the file is already within quota. Opens the file safely (O_NOFOLLOW) and
+ * rewrites the tail in place to avoid leaving a sparse/holey file behind
+ * (truncating a live append-only fd below its write offset is undefined per
+ * POSIX).
+ */
+function truncateLogToTail(logFile: string, maxBytes: number): void {
+  let fd: number | undefined
+  try {
+    fd = fs.openSync(logFile, fs.constants.O_RDWR | O_NOFOLLOW_FLAG)
+    const size = fs.fstatSync(fd).size
+    if (size <= maxBytes) return
+    const keepStart = size - maxBytes
+    const buf = Buffer.alloc(maxBytes)
+    const bytesRead = fs.readSync(fd, buf, 0, maxBytes, keepStart)
+    fs.ftruncateSync(fd, 0)
+    fs.writeSync(fd, buf.subarray(0, bytesRead), 0, bytesRead, 0)
+  } catch {
+    // best-effort truncation; the exit path owns final cleanup
+  } finally {
+    if (fd !== undefined) {
+      try {
+        fs.closeSync(fd)
+      } catch {
+        // already closed
+      }
+    }
+  }
+}
+
 function safeReadJobMetadataFile(metadataFile: string): string | undefined {
   let fd: number | undefined
   try {
@@ -438,11 +464,6 @@ function writeBackgroundJobMetadata(job: BackgroundJob): void {
   }
 }
 
-/** Registry-side id backing `job` (recovered jobs are remapped; see above). */
-function registryJobIdFor(job: BackgroundJob): string {
-  return job.registryJobId ?? job.jobId
-}
-
 /**
  * Fold a terminal lifecycle transition into the registry and mirror it onto
  * the adapter object + the write-only disk projection. All terminal paths
@@ -466,7 +487,7 @@ function settleBackgroundJob(
   if (job.settledAt === undefined) {
     job.settledAt = Date.now()
   }
-  jobRegistry.emit(registryJobIdFor(job), {
+  jobRegistry.emit(job.jobId, {
     type: 'lifecycle',
     state: status,
     exitCode,
@@ -497,7 +518,7 @@ export function pruneSettledJobs(now: number = Date.now()): void {
 /** Mirror a chunk of streamed log bytes into the registry as an output event. */
 function emitJobOutput(job: BackgroundJob, data: string): void {
   if (data.length === 0) return
-  jobRegistry.emit(registryJobIdFor(job), { type: 'output', data })
+  jobRegistry.emit(job.jobId, { type: 'output', data })
 }
 
 /**
@@ -601,11 +622,17 @@ export function startBackgroundJob(params: {
   // safeCreateJobLogFile/safeWriteJobMetadata also use O_EXCL + O_NOFOLLOW so
   // pre-created regular files and TOCTOU symlink swaps are rejected at open().
   let outFd: number | undefined
+  // True only once safeCreateJobLogFile has returned, i.e. this spawn
+  // exclusively created logFile via O_CREAT|O_EXCL. Throws from
+  // rejectIfSymlink or O_EXCL EEXIST leave it false: the path then holds a
+  // foreign/pre-existing file this process must never delete.
+  let logFileCreatedByThisSpawn = false
   let child: ChildProcess
   try {
     rejectIfSymlink(logFile)
     rejectIfSymlink(metadataFile)
     outFd = safeCreateJobLogFile(logFile)
+    logFileCreatedByThisSpawn = true
     child = spawn(shell, [...shellArgs, command], {
       cwd,
       env,
@@ -623,13 +650,15 @@ export function startBackgroundJob(params: {
         // already closed
       }
     }
-    // The log file was created exclusively for this spawn (O_EXCL); on a failed
-    // spawn nobody else can own it, so remove it best-effort rather than leaving
-    // an empty temp file behind until the 24h+ orphan sweep.
-    try {
-      fs.unlinkSync(logFile)
-    } catch {
-      // best-effort; the orphan sweep will clean it up if removal fails
+    // Remove the log file ONLY when this spawn created it (safeCreateJobLogFile
+    // succeeded). If the throw came from rejectIfSymlink or from O_EXCL EEXIST,
+    // the path belongs to a foreign/pre-existing file this process did NOT
+    // create, and deleting it would clobber another owner's file in the shared
+    // temp dir. removeFileIfPresent re-lstats immediately before unlinking so
+    // a TOCTOU symlink swap fails closed; a failed removal still leaves the
+    // empty file for the 24h+ orphan sweep.
+    if (logFileCreatedByThisSpawn) {
+      removeFileIfPresent(logFile)
     }
     // Fold the failed spawn into the registry so the freshly-created id does
     // not linger as a queued job (queued only transitions via running).
@@ -697,11 +726,12 @@ export function startBackgroundJob(params: {
       // Keep trimming while the process is unwinding so a chatty child cannot
       // regrow a sparse/oversized log between SIGTERM and exit.
       //
-      // Deliberate tradeoff: truncation keeps the HEAD (oldest bytes) and drops
-      // the newest output. The job is being terminated for exceeding its log
-      // quota, and the startup/context head is the most diagnostically useful
-      // part; losing the tail is a known, accepted cost.
-      fs.truncateSync(logFile, MAX_BACKGROUND_LOG_BYTES)
+      // Deliberate tradeoff: truncation keeps the TAIL (newest bytes) and drops
+      // the oldest output. The job is being terminated for exceeding its log
+      // quota, and the newest output is the most diagnostically useful — errors,
+      // build failures, and recent status messages live in the tail; losing the
+      // head is a known, accepted cost.
+      truncateLogToTail(logFile, MAX_BACKGROUND_LOG_BYTES)
     } catch {
       // The process exit/error handlers own final cleanup.
     }
@@ -732,11 +762,11 @@ export function startBackgroundJob(params: {
           ? 'completed'
           : 'error'
     if (quotaExceeded) {
-      // See the log-quota monitor above: truncation keeps the HEAD (oldest
-      // bytes) and drops the newest output. Known, accepted tradeoff for a
+      // See the log-quota monitor above: truncation keeps the TAIL (newest
+      // bytes) and drops the oldest output. Known, accepted tradeoff for a
       // job terminated for exceeding its log quota.
       try {
-        fs.truncateSync(logFile, MAX_BACKGROUND_LOG_BYTES)
+        truncateLogToTail(logFile, MAX_BACKGROUND_LOG_BYTES)
       } catch {
         // best-effort truncation; the exit path owns final cleanup
       }
@@ -803,31 +833,13 @@ function resolveRestampedOwner(
 }
 
 /**
- * Resolve a process adapter for a *registry* job id (list_jobs iteration).
- *
- * The adapter Map is keyed by the user-facing jobId. Live spawns use the same
- * id for both Map key and registry. Recovered / `__registerJobForTest` jobs
- * remap: Map key = disk/user jobId, `registryJobId` = fresh registry id. A
- * direct `jobs.get(registryId)` therefore misses remapped adapters — reverse-
- * scan by `registryJobId ?? jobId` so list_jobs can read `lastCheckCursor` /
- * lineCarry and expose the user-facing `adapter.jobId` that check_job/kill_job
- * resolve via `getBackgroundJob`.
+ * Read-only adapter lookup by jobId (no recovery, no ownership changes).
+ * list_jobs uses this to read `lastCheckCursor` / `lineCarry` from the
+ * adapter without triggering recovery or ownership restamping.
  */
-export function getBackgroundJobForRegistryId(
-  registryJobId: string,
-): BackgroundJob | undefined {
+export function getBackgroundJobAdapter(jobId: string): BackgroundJob | undefined {
   pruneSettledJobs()
-  const direct = jobs.get(registryJobId)
-  if (direct !== undefined) {
-    const backing = direct.registryJobId ?? direct.jobId
-    if (backing === registryJobId) return direct
-  }
-  for (const job of jobs.values()) {
-    if ((job.registryJobId ?? job.jobId) === registryJobId) {
-      return job
-    }
-  }
-  return undefined
+  return jobs.get(jobId)
 }
 
 export function getBackgroundJob(
@@ -849,13 +861,12 @@ export function getBackgroundJob(
     // this, the cached-job early return skips the re-stamp and the trusted
     // caller is locked out of its own job by assertOwned. Gate on the
     // REGISTRY record's current owner being unknown (SEC-2).
-    const registryJobId = existing.registryJobId ?? existing.jobId
-    const registryJob = jobRegistry.get(registryJobId)
+    const registryJob = jobRegistry.get(jobId)
     const upgraded = registryJob
       ? resolveRestampedOwner(registryJob.owner, opts?.restampOwner)
       : undefined
     if (upgraded) {
-      jobRegistry.restampOwner(registryJobId, upgraded)
+      jobRegistry.restampOwner(jobId, upgraded)
       existing.owner = upgraded
     }
     return existing
@@ -874,34 +885,44 @@ export function getBackgroundJob(
       return undefined
     }
 
-    jobs.set(jobId, recovered)
     // Re-emit the recovered job into the registry, which is the live source
-    // of truth for state/ownership. The registry allocates its own ids, so
-    // the disk-derived jobId is remapped onto a fresh registry id stored on
-    // the adapter object. Cross-session re-attach: a supplied restampOwner
-    // UPGRADES ownership only when the disk metadata carries a placeholder /
-    // missing owner. A job already stamped with a real owner keeps it, so a
-    // re-attaching run can never launder ownership of another session's job
-    // (assertOwned then refuses it as foreign) — this mirrors the cached-job
-    // branch above (SEC-2). Same-process callers pass no opts, so the
-    // original owner is preserved either way.
+    // of truth for state/ownership. The disk-derived jobId is passed as the
+    // explicit registry id so the registry record and adapter Map share one
+    // key. Cross-session re-attach: a supplied restampOwner UPGRADES ownership
+    // only when the disk metadata carries a placeholder / missing owner. A job
+    // already stamped with a real owner keeps it, so a re-attaching run can
+    // never launder ownership of another session's job (assertOwned then
+    // refuses it as foreign) — this mirrors the cached-job branch above
+    // (SEC-2). Same-process callers pass no opts, so the original owner is
+    // preserved either way.
     const owner =
       resolveRestampedOwner(recovered.owner, opts?.restampOwner) ??
       recovered.owner ??
       UNKNOWN_JOB_OWNER
     recovered.owner = owner
-    const registryJobId = jobRegistry.create({
+    // Pass the disk-derived jobId as the explicit registry id so the
+    // registry record and adapter Map share one key (single-id invariant).
+    // A collision means a live job already owns this id — a cross-session
+    // recovery must NOT silently remap onto a fresh id (that would desync the
+    // adapter from the on-disk log/metadata file names and make the caller's
+    // requested id unresolvable), so refuse the recovery and leave the id
+    // pointing at the live record. getBackgroundJob reports not-found.
+    if (jobRegistry.get(recovered.jobId) !== undefined) {
+      return undefined
+    }
+    jobRegistry.create({
       kind: 'process',
       label: recovered.command,
       owner,
-    }).jobId
-    recovered.registryJobId = registryJobId
-    jobRegistry.start(registryJobId)
+      jobId: recovered.jobId,
+    })
+    jobs.set(recovered.jobId, recovered)
+    jobRegistry.start(recovered.jobId)
     if (recovered.status !== 'running') {
       // A settled recovered job is folded straight into its terminal state so
       // the re-attaching run can serve its final output/exit code from the
       // registry.
-      jobRegistry.emit(registryJobId, {
+      jobRegistry.emit(recovered.jobId, {
         type: 'lifecycle',
         state: recovered.status,
         exitCode: recovered.exitCode,
@@ -1204,18 +1225,26 @@ export function readNewJobOutput(job: BackgroundJob): string {
 
 /** Test-only: register a job backed by an existing log file (no real process). */
 export function __registerJobForTest(job: BackgroundJob): void {
-  jobs.set(job.jobId, job)
-  // Mirror the job into the registry (under a fresh registry id, like a
-  // recovered job) so registry-backed lookups see the same state.
-  const registryJobId = jobRegistry.create({
+  // Pass the adapter's jobId as the explicit registry id so the registry
+  // record and adapter Map share one key (single-id invariant). A collision
+  // means the test registered two adapters under one id — that is a test bug
+  // and must fail loudly instead of silently remapping onto a fresh id (which
+  // would break every assertion made on the original jobId).
+  if (jobRegistry.get(job.jobId) !== undefined) {
+    throw new Error(
+      `__registerJobForTest: job id '${job.jobId}' is already registered`,
+    )
+  }
+  jobRegistry.create({
     kind: 'process',
     label: job.command,
     owner: job.owner ?? UNKNOWN_JOB_OWNER,
-  }).jobId
-  job.registryJobId = registryJobId
-  jobRegistry.start(registryJobId)
+    jobId: job.jobId,
+  })
+  jobs.set(job.jobId, job)
+  jobRegistry.start(job.jobId)
   if (job.status !== 'running') {
-    jobRegistry.emit(registryJobId, {
+    jobRegistry.emit(job.jobId, {
       type: 'lifecycle',
       state: job.status,
       exitCode: job.exitCode,
@@ -1224,9 +1253,8 @@ export function __registerJobForTest(job: BackgroundJob): void {
 }
 
 /**
- * Test-only: set/create a Map adapter keyed by registry jobId (production id shape).
- * Used only by tests so list_jobs can observe a same-id lastCheckCursor without
- * `__registerJobForTest`'s dual-id remapping.
+ * Test-only: set/create a Map adapter keyed by jobId (production id shape).
+ * Used only by tests so list_jobs can observe a same-id lastCheckCursor.
  */
 export function __setLastCheckCursorForTest(
   jobId: string,
@@ -1242,7 +1270,6 @@ export function __setLastCheckCursorForTest(
   // registry row under the same jobId as production live spawns).
   jobs.set(jobId, {
     jobId,
-    registryJobId: jobId,
     command: 'test',
     child: {} as ChildProcess,
     logFile: path.join(os.tmpdir(), `openbuff-${jobId}.log`),
diff --git a/sdk/src/tools/check-job.ts b/sdk/src/tools/check-job.ts
index 1ae8e55767..1d3e5c1d6d 100644
--- a/sdk/src/tools/check-job.ts
+++ b/sdk/src/tools/check-job.ts
@@ -1,4 +1,5 @@
 import {
+  flushJobLineCarry,
   getBackgroundJob,
   killBackgroundJob,
   peekJobLineCarry,
@@ -26,11 +27,6 @@ const POLL_INTERVAL_MS = 200
 export const CHECK_JOB_POLL_ACCUMULATION_CAP = CHECK_JOB_OUTPUT_LIMIT * 2
 const COLLECTED_TAIL_KEEP = Math.floor(CHECK_JOB_OUTPUT_LIMIT / 4)
 
-/** Registry-side id backing this adapter job (recovered jobs are remapped). */
-function registryIdFor(job: { jobId: string; registryJobId?: string }): string {
-  return job.registryJobId ?? job.jobId
-}
-
 /** True when a job can no longer produce output. */
 function jobSettled(job: { status: string }): boolean {
   return job.status !== 'running'
@@ -113,6 +109,134 @@ export function boundEventsToOutputTail(
   return { events: result, truncated }
 }
 
+/**
+ * Resolve the one-shot settlement dirty delta the first time a settled
+ * observation is returned. Stores `[]` when snapshot/git is missing so
+ * re-polls stay idempotent and never re-attribute post-settle dirt.
+ * Returns paths to emit only on the resolving observation (omit on
+ * subsequent polls and while still running).
+ */
+async function resolveSettlementTouchedPaths(job: {
+  settlementTouchedPaths?: string[]
+  dirtyBeforePaths?: string[]
+  projectRoot?: string
+}): Promise<string[] | undefined> {
+  if (job.settlementTouchedPaths !== undefined) {
+    // Already resolved on a prior settled check_job — do not re-emit.
+    return undefined
+  }
+  if (job.dirtyBeforePaths !== undefined && job.projectRoot !== undefined) {
+    const dirtyAfter = await listDirtyPaths(job.projectRoot)
+    const touched =
+      dirtyAfter !== null
+        ? dirtyDelta(new Set(job.dirtyBeforePaths), dirtyAfter)
+        : []
+    job.settlementTouchedPaths = touched
+    return touched
+  }
+  // Soft-fail: recovered jobs / no git snapshot — lock out recompute.
+  job.settlementTouchedPaths = []
+  return undefined
+}
+
+/**
+ * Resolve the follow deadline for a check_job call. The deadline is computed
+ * at ENTRY, before any registry calls: both getBackgroundJob (on recovery)
+ * and assertOwned invoke sweep() → Date.now(), so computing the deadline
+ * afterward would base it on a later clock read and could push the follow
+ * window out indefinitely (a mocked or fast-advancing clock would then never
+ * satisfy `Date.now() >= deadline`).
+ */
+function resolveCheckJobWaitBounds(params: {
+  timeoutSeconds?: number
+}): { timeoutMs: number; deadline: number } {
+  const timeoutMs = Math.max(0, (params.timeoutSeconds ?? 0) * 1000)
+  return { timeoutMs, deadline: Date.now() + timeoutMs }
+}
+
+/**
+ * Build the structured poll/follow result from the raw follow state. Computes
+ * the common output fields (state, events, nextCursor, truncated, dropped,
+ * matched, exitCode) shared by poll and follow modes. Advances the per-adapter
+ * consumer cursor (lastCheckCursor) as a side effect. The presentation-level
+ * output cap (boundEventsToOutputTail) is applied here; the caller folds the
+ * returned `timedOut` into its kill/return logic.
+ */
+function buildCheckJobPollResult(params: {
+  jobId: string
+  job: {
+    status: JobState
+    exitCode?: number | null
+    logFile: string
+    lastCheckCursor?: number
+  }
+  waitFor?: string
+  matched: boolean
+  entryCursor: number
+  newEvents: JobEvent[]
+  cursor: number
+  truncated: boolean
+}): {
+  jobId: string
+  state: JobState
+  events: JobEvent[]
+  nextCursor: number
+  truncated: boolean
+  dropped: number
+  exitCode?: number
+  matched?: boolean
+  logFile: string
+} {
+  const {
+    jobId,
+    job,
+    waitFor,
+    matched,
+    entryCursor,
+    newEvents,
+    cursor,
+    truncated: truncatedSoFar,
+  } = params
+  const registryJob = jobRegistry.get(jobId)
+  const state = registryJob?.state ?? job.status
+  const exitCode = registryJob?.exitCode ?? job.exitCode ?? undefined
+  // Re-snapshot from the entry cursor so the returned `events` cover the
+  // FULL window [entryCursor, nextCursor) — every `output` event drained
+  // during this follow, not just the final iteration's batch. This keeps
+  // `events` consistent with `matched` (computed over the accumulated
+  // `collected` window) and with `nextCursor`, so the caller receives all
+  // output it would otherwise never be able to refetch. Bounded by the
+  // registry's per-job event/byte ring buffer.
+  const finalSnapshot = jobRegistry.snapshot(jobId, entryCursor)
+  const rawEvents = finalSnapshot?.events ?? newEvents
+  const boundedResult = boundEventsToOutputTail(
+    rawEvents,
+    CHECK_JOB_OUTPUT_LIMIT,
+  )
+  const events = boundedResult.events
+  const nextCursor = finalSnapshot?.nextCursor ?? cursor
+  // Advance the per-adapter consumer cursor so the next check_job does not
+  // re-serve these events. Mirrored-by-live-drainer is not the same as
+  // consumed-by-check_job; only this advance marks consumption.
+  job.lastCheckCursor = nextCursor
+  const finalDropped = finalSnapshot?.dropped ?? 0
+  const finalTruncated =
+    truncatedSoFar ||
+    (finalSnapshot?.truncated ?? false) ||
+    boundedResult.truncated
+  return {
+    jobId,
+    state,
+    events,
+    nextCursor,
+    truncated: finalTruncated,
+    dropped: finalDropped,
+    ...(exitCode !== undefined && exitCode !== null ? { exitCode } : {}),
+    ...(waitFor ? { matched } : {}),
+    logFile: job.logFile,
+  }
+}
+
 /**
  * Join (poll) or wait (follow) on a background job started by
  * run_terminal_command.
@@ -150,7 +274,6 @@ export async function checkJob(params: {
   owner: BackgroundJobOwner
 }): Promise<CodebuffToolOutput<'check_job'>> {
   const { jobId, wait_for: waitFor, owner } = params
-  const timeoutMs = Math.max(0, (params.timeout_seconds ?? 0) * 1000)
   // Observation must be non-destructive by default. Callers can explicitly
   // request termination when a follow timeout represents a hard deadline.
   const killOnTimeout = params.kill_on_timeout ?? false
@@ -159,7 +282,9 @@ export async function checkJob(params: {
   // so computing the deadline afterward would base it on a later clock read and
   // could push the follow window out indefinitely (a mocked or fast-advancing
   // clock would then never satisfy `Date.now() >= deadline`).
-  const deadline = Date.now() + timeoutMs
+  const { timeoutMs, deadline } = resolveCheckJobWaitBounds({
+    timeoutSeconds: params.timeout_seconds,
+  })
 
   // Cross-session recovery re-stamps the registry record with THIS trusted
   // owner (never a model-supplied one).
@@ -179,8 +304,7 @@ export async function checkJob(params: {
   // Ownership gate: the registry is the source of truth for who owns this
   // job. A foreign job is refused with the SAME generic not_found error as
   // an unknown id so the caller cannot probe for other sessions' jobs.
-  const registryJobId = registryIdFor(job)
-  const ownership = jobRegistry.assertOwned(registryJobId, owner)
+  const ownership = jobRegistry.assertOwned(jobId, owner)
   if (!ownership.ok) {
     return [
       {
@@ -227,7 +351,7 @@ export async function checkJob(params: {
     // Recovered and test-registered jobs have no interval, so they still drain
     // here.
     if (!job.hasLiveDrainer) readNewJobOutput(job)
-    const snapshot = jobRegistry.snapshot(registryJobId, cursor)
+    const snapshot = jobRegistry.snapshot(jobId, cursor)
     const newEvents = snapshot?.events ?? []
     if (snapshot) {
       cursor = snapshot.nextCursor
@@ -244,27 +368,36 @@ export async function checkJob(params: {
     // yet emitted as a per-line registry `output` event — is still matchable;
     // the carry is NOT folded into `collected` (that must stay bounded and
     // event-derived, or it would double-count once the line is later emitted).
-    // For a live job (hasLiveDrainer) this `wait_for` window includes the
-    // pending lineCarry, so checkJob can return matched:true while the
-    // corresponding needle has NOT yet appeared in the returned events/
-    // outputText for that same response — the event is emitted on the next
-    // drain (≤250ms). This transient inconsistency is intentional and bounded
-    // by MAX_LINE_BYTES, not a bug.
-    if (
-      waitFor &&
-      !matched &&
-      (collected + chunk + peekJobLineCarry(job)).includes(waitFor)
-    ) {
-      matched = true
+    // When the needle is found in the pending partial line (carry), force-emit
+    // the carry as a registry output event so the returned events are consistent
+    // with matched: true. Without this, a needle in an unterminated partial line
+    // could be reported as matched while being absent from the returned events/
+    // outputText. The re-snapshot from entryCursor picks up the flushed event,
+    // so it appears in the returned events. The cursor variable in the follow
+    // loop is not updated by the flush, but this is fine because (a) the
+    // re-snapshot uses entryCursor not cursor, and (b) matched is now true so
+    // the loop exits.
+    if (waitFor && !matched) {
+      const matchWindow = collected + chunk + peekJobLineCarry(job)
+      if (matchWindow.includes(waitFor)) {
+        matched = true
+        // If the needle depends on the carry (not fully present in the
+        // already-emitted collected + chunk), force-emit the carry so the
+        // returned events are consistent with matched: true. This covers both
+        // the case where the needle is entirely in the carry and where it spans
+        // the boundary between chunk and carry. Without this, a needle in an
+        // unterminated partial line could be reported as matched while being
+        // absent from the returned events/outputText.
+        if (!(collected + chunk).includes(waitFor)) {
+          flushJobLineCarry(job)
+        }
+      }
     }
     // Bound the match window so a chatty long-running job can't grow
     // `collected` without limit (OOM) across many poll iterations.
     collected = appendBoundedCollected(collected, chunk)
     const finished = jobSettled(job)
     if (matched || finished || Date.now() >= deadline) {
-      const registryJob = jobRegistry.get(registryJobId)
-      const state = registryJob?.state ?? job.status
-      const exitCode = registryJob?.exitCode ?? job.exitCode ?? undefined
       // The follow-timeout fired (deadline reached, pattern NOT matched, job
       // NOT finished, and still running) and only in follow mode (timeoutMs > 0).
       // Poll mode (timeoutMs === 0) must never kill even though its deadline
@@ -272,95 +405,29 @@ export async function checkJob(params: {
       // there — but guard with timeoutMs > 0 to be explicit and safe.
       const timedOut =
         timeoutMs > 0 && !matched && !finished && Date.now() >= deadline
-      // Re-snapshot from the entry cursor so the returned `events` cover the
-      // FULL window [entryCursor, nextCursor) — every `output` event drained
-      // during this follow, not just the final iteration's batch. This keeps
-      // `events` consistent with `matched` (computed over the accumulated
-      // `collected` window) and with `nextCursor`, so the caller receives all
-      // output it would otherwise never be able to refetch. Bounded by the
-      // registry's per-job event/byte ring buffer.
-      const finalSnapshot = jobRegistry.snapshot(registryJobId, entryCursor)
-      const rawEvents = finalSnapshot?.events ?? newEvents
-      const boundedResult = boundEventsToOutputTail(
-        rawEvents,
-        CHECK_JOB_OUTPUT_LIMIT,
-      )
-      const events = boundedResult.events
-      const nextCursor = finalSnapshot?.nextCursor ?? cursor
-      // Advance the per-adapter consumer cursor so the next check_job does not
-      // re-serve these events. Mirrored-by-live-drainer is not the same as
-      // consumed-by-check_job; only this advance marks consumption.
-      job.lastCheckCursor = nextCursor
-      const finalDropped = finalSnapshot?.dropped ?? 0
-      const finalTruncated =
-        truncated ||
-        (finalSnapshot?.truncated ?? false) ||
-        boundedResult.truncated
-      const baseValue: {
-        jobId: string
-        state: JobState
-        events: JobEvent[]
-        nextCursor: number
-        truncated: boolean
-        dropped: number
-        exitCode?: number
-        matched?: boolean
-        logFile: string
-      } = {
+      const baseValue = buildCheckJobPollResult({
         jobId,
-        state,
-        events,
-        nextCursor,
-        truncated: finalTruncated,
-        dropped: finalDropped,
-        ...(exitCode !== undefined && exitCode !== null ? { exitCode } : {}),
-        ...(waitFor ? { matched } : {}),
-        logFile: job.logFile,
-      }
-
-      /**
-       * Resolve the one-shot settlement dirty delta the first time a settled
-       * observation is returned. Stores `[]` when snapshot/git is missing so
-       * re-polls stay idempotent and never re-attribute post-settle dirt.
-       * Returns paths to emit only on the resolving observation (omit on
-       * subsequent polls and while still running).
-       */
-      const resolveSettlementTouchedPaths = async (): Promise<
-        string[] | undefined
-      > => {
-        if (job.settlementTouchedPaths !== undefined) {
-          // Already resolved on a prior settled check_job — do not re-emit.
-          return undefined
-        }
-        if (
-          job.dirtyBeforePaths !== undefined &&
-          job.projectRoot !== undefined
-        ) {
-          const dirtyAfter = await listDirtyPaths(job.projectRoot)
-          const touched =
-            dirtyAfter !== null
-              ? dirtyDelta(new Set(job.dirtyBeforePaths), dirtyAfter)
-              : []
-          job.settlementTouchedPaths = touched
-          return touched
-        }
-        // Soft-fail: recovered jobs / no git snapshot — lock out recompute.
-        job.settlementTouchedPaths = []
-        return undefined
-      }
+        job,
+        waitFor,
+        matched,
+        entryCursor,
+        newEvents,
+        cursor,
+        truncated,
+      })
 
       if (timedOut && job.status === 'running' && killOnTimeout) {
         const killResult = killBackgroundJob(jobId, 'SIGTERM')
         if ('killed' in killResult) {
           // killBackgroundJob updates the in-memory job status; prefer the
           // fresh kill-result state/exitCode over the stale local snapshot.
-          const postKillJob = jobRegistry.get(registryJobId)
+          const postKillJob = jobRegistry.get(jobId)
           const postKillState = postKillJob?.state ?? killResult.status
           const postKillExitCode =
             postKillJob?.exitCode ?? killResult.exitCode ?? undefined
           // Kill settles the job; credit dirty delta on this first settled
           // observation (same one-shot path as natural finish).
-          const killTouched = await resolveSettlementTouchedPaths()
+          const killTouched = await resolveSettlementTouchedPaths(job)
           const killValue = {
             ...baseValue,
             state: postKillState,
@@ -396,7 +463,7 @@ export async function checkJob(params: {
       }
 
       const settlementTouched = finished
-        ? await resolveSettlementTouchedPaths()
+        ? await resolveSettlementTouchedPaths(job)
         : undefined
       const resultValue =
         settlementTouched !== undefined
@@ -429,7 +496,7 @@ export async function checkJob(params: {
     // before ever reaching this line, so wait() never blocks a poll. No
     // predicate is passed and the return value is ignored.
     const remaining = deadline - Date.now()
-    await jobRegistry.wait(registryJobId, {
+    await jobRegistry.wait(jobId, {
       timeoutMs: Math.min(POLL_INTERVAL_MS, remaining),
       cursor,
     })
diff --git a/sdk/src/tools/kill-job.ts b/sdk/src/tools/kill-job.ts
index 36c7f2cae8..5c623a591e 100644
--- a/sdk/src/tools/kill-job.ts
+++ b/sdk/src/tools/kill-job.ts
@@ -5,11 +5,6 @@ import { getBackgroundJob, killBackgroundJob } from './background-jobs'
 import type { BackgroundJobOwner } from './background-jobs'
 import type { CodebuffToolOutput } from '../../../common/src/tools/list'
 
-/** Registry-side id backing this adapter job (recovered jobs are remapped). */
-function registryIdFor(job: { jobId: string; registryJobId?: string }): string {
-  return job.registryJobId ?? job.jobId
-}
-
 export async function killJob(params: {
   jobId: string
   signal?: 'SIGTERM' | 'SIGKILL'
@@ -38,7 +33,7 @@ export async function killJob(params: {
   // Ownership gate BEFORE any kill path: a foreign job is refused with the
   // same generic not_found error as an unknown id (no existence leak), and
   // terminateProcessTree is never reached for another session's job.
-  const ownership = jobRegistry.assertOwned(registryIdFor(job), params.owner)
+  const ownership = jobRegistry.assertOwned(job.jobId, params.owner)
   if (!ownership.ok) {
     return [
       {
diff --git a/sdk/src/tools/list-jobs.ts b/sdk/src/tools/list-jobs.ts
index 8017b75e2a..cfece28db0 100644
--- a/sdk/src/tools/list-jobs.ts
+++ b/sdk/src/tools/list-jobs.ts
@@ -13,11 +13,10 @@ import {
 } from '@codebuff/common/util/job-registry'
 
 import {
-  getBackgroundJobForRegistryId,
+  getBackgroundJobAdapter,
   peekJobLineCarry,
+  type BackgroundJobOwner,
 } from './background-jobs'
-
-import type { BackgroundJobOwner } from './background-jobs'
 import type { CodebuffToolOutput } from '../../../common/src/tools/list'
 
 /** Last ≤10 non-empty output lines from buffered events (terminal peek only). */
@@ -40,7 +39,10 @@ export async function listJobs(params: {
    */
   owner: BackgroundJobOwner
 }): Promise<CodebuffToolOutput<'list_jobs'>> {
-  const owner = {
+  // Narrow the trusted owner to the registry's ownership key: list_jobs is
+  // scoped by (clientSessionId, rootRunId) only; parentRunId/parentAgentId are
+  // diagnostic and deliberately excluded from the ownership filter.
+  const scopeOwner = {
     clientSessionId: params.owner.clientSessionId,
     rootRunId: params.owner.rootRunId,
   }
@@ -54,19 +56,14 @@ export async function listJobs(params: {
   // the same fallback the row build below uses, keeping order/tie-break
   // parity), so adapter reverse-resolution, snapshot, pending, and tail work
   // runs only for the ≤LIST_JOBS_MAX_ROWS rows actually emitted.
-  const candidates = jobRegistry.list(owner).map((entry) => ({
+  const candidates = jobRegistry.list(scopeOwner).map((entry) => ({
     entry,
     status: entry.state,
     startedAt: entry.startedAt ?? entry.createdAt,
   }))
   const selected = selectListJobsRows(candidates)
   const rows: ListJobsViewRow[] = selected.rows.map(({ entry }) => {
-    // Resolve process adapters by *registry* id (not a direct Map get on the
-    // registry id alone). Live spawns share one id; recovered /
-    // `__registerJobForTest` remaps (Map key = user/disk jobId,
-    // registryJobId = fresh registry id). Reverse-scan so lastCheckCursor and
-    // lineCarry are visible, and emit the user-facing adapter.jobId so
-    // rediscovered ids work with check_job/kill_job.
+    // Resolve process adapters by jobId (Map key = registry id = jobId).
     //
     // `pending` line buckets count only registry events with
     // `payload.type === 'output'` relative to the process adapter's
@@ -77,7 +74,7 @@ export async function listJobs(params: {
     // ring truncation at the snapshot cursor for any kind.
     //
     // The lineCarry +1 counting rationale lives on `countPendingOutputLines`.
-    const adapter = getBackgroundJobForRegistryId(entry.jobId)
+    const adapter = getBackgroundJobAdapter(entry.jobId)
     const cursor = adapter?.lastCheckCursor ?? 0
     const snap = jobRegistry.snapshot(entry.jobId, cursor)
     const lineCarry =
@@ -89,10 +86,10 @@ export async function listJobs(params: {
       }),
     )
     const gap = snap?.truncated ?? false
-    // Prefer user-facing adapter.jobId when remapped; agents/no-adapter keep
-    // the registry id (the only id they have).
     const row: ListJobsViewRow = {
-      jobId: adapter?.jobId ?? entry.jobId,
+      // Single-id invariant: the adapter Map key, the registry id, and the
+      // user-facing jobId are the same string, so emit entry.jobId directly.
+      jobId: entry.jobId,
       kind: entry.kind,
       command: entry.label,
       status: entry.state,
diff --git a/sdk/src/tools/read-logs.ts b/sdk/src/tools/read-logs.ts
index 1aff08163b..f048d831f4 100644
--- a/sdk/src/tools/read-logs.ts
+++ b/sdk/src/tools/read-logs.ts
@@ -38,11 +38,6 @@ const MAX_TAIL_SCAN_CHARS_MULTIPLE = 8
  */
 const BACKGROUND_JOB_FILE_PATTERN = /^openbuff-(.+)\.(log|json)$/
 
-/** Registry-side id backing this adapter job (recovered jobs are remapped). */
-function registryIdFor(job: { jobId: string; registryJobId?: string }): string {
-  return job.registryJobId ?? job.jobId
-}
-
 type ReadLogsParams = {
   cwd: string
   path?: string
@@ -90,7 +85,7 @@ export async function readLogs(
 
     // Ownership gate before serving the log tail: a foreign job is refused
     // with the same generic not_found error as an unknown id.
-    const ownership = jobRegistry.assertOwned(registryIdFor(job), params.owner)
+    const ownership = jobRegistry.assertOwned(job.jobId, params.owner)
     if (!ownership.ok) {
       return [
         {
@@ -205,7 +200,7 @@ export async function readLogs(
   if (jobFileMatch) {
     const jobId = jobFileMatch[1]
     const job = getBackgroundJob(jobId)
-    if (job && !jobRegistry.assertOwned(registryIdFor(job), params.owner).ok) {
+    if (job && !jobRegistry.assertOwned(job.jobId, params.owner).ok) {
       return [
         {
           type: 'json',