refactor(tasks): 后台任务改读宿主客户端 jobs 服务，删除自建三条路由

- 新增 src/client/jobs-client.ts：结构探测 ctx.jobs（watchRows / observe /
  state / kill），引用计数的按会话 roster 订阅、只观测展开的那一行、
  跨树去重（无主任务归到承载它的被观测会话，树外 owner 丢弃）。
- 抽屉与输出浮窗改读该服务：行来自推送 roster（状态 / 时长 / owner / 进度），
  输出面换 TerminalBlock 显示宿主为观察者流的保留输出（gapBefore → 截断提示、
  error → 失败提示），kill 走 jobs.kill(owner, jobId)，两击确认不变。
- 自动激活（use-host-feeds）改骑同一条推送 roster；detectNewJob 增加
  「观测起点时钟」参数，避免「页面打开时已在跑的任务」被当成新工作。
- 删除 src/jobs-routes.ts（+ tests/jobs-routes.spec.ts）、index.ts 的三条路由、
  api.jobsList/jobOutput/jobKill 与 JobOutputResult、context-types 的
  SidebarJobsService（换成 SidebarClientJobsService/SidebarJobsSnapshot/
  SidebarObservedJob 镜像，SidebarJobView 补 owner/progress）、
  页面里的 jobs.list 轮询与 collectTreeJobs。
- 三个 spec 改用假 ctx.jobs 服务（jobs-view 14 例 / 自动激活 12 例 /
  页面 20 例）：观测只在展开行、切换与关闭释放观测、watch 隐藏时不建立、
  卸载时全部释放、无服务时不渲染任务面。

验证：typecheck / lint 通过；pnpm test 1240 passed / 9 skipped；pnpm build + pack 通过。

(cherry picked from commit 7d9b82e73190218f05910433ed520cb90e4d4ab2)

src/client/JobsDrawer.tsx

@@ -1,24 +1,26 @@
 /**
  * The background-jobs bottom drawer of the Tasks page and its output popover.
  *
- * Drawer: the tree's jobs (owner-labeled, fed by the `session/jobs` push
- * mirror) collapse into a bottom bar that AUTO-COLLAPSES once the tree has
- * many agents — the manual toggle always wins afterwards.
+ * Drawer: the tree's jobs (owner-labeled, fed by the HOST's client jobs
+ * service — see jobs-client.ts) collapse into a bottom bar that AUTO-COLLAPSES
+ * once the tree has many agents — the manual toggle always wins afterwards.
  *
  * Popover: clicking a row opens a DRAGGABLE, portalled output card (see
  * AnchoredPopover) instead of docking a pane inside the page. It shows the
- * output the MODEL has read (event replay — never the model's job_output
- * cursor), with a copy action, a follow-latest switch, the two-click kill,
- * and a terminal-style tail while the job runs.
+ * output the host streams for an OBSERVER (never the model's consuming
+ * cursor), with a copy action, a follow-latest switch, the two-click kill, and
+ * a terminal-style tail while the job runs.
  *
- * Every control is a host primitive (Button / Tag / Switch / StateDot).
+ * Every control is a host primitive (Button / Tag / Switch / StateDot /
+ * TerminalBlock).
  */
 import { useCallback, useEffect, useMemo, useRef, useState, type ReactNode } from 'react'
 import {
   Button, IconChevronUpOutlineRegular, IconCopyOutlineRegular, IconStopFillRegular, StateDot, Switch, Tag,
-  type TagTone,
+  TerminalBlock, type TagTone,
 } from '@deepseek-ai/dsh-client-ui-primitives'
-import type { SidebarJobView } from '../context-types.ts'
+import type { SidebarClientJobsService, SidebarJobView, SidebarObservedJob } from '../context-types.ts'
+import { terminalBlockLabels } from './block-labels.ts'
 import {
   formatJobDuration,
   isJobLive,
@@ -26,12 +28,9 @@ import {
   jobStatusLabel,
   type TreeJob,
 } from './subagent-jobs.ts'
-import { api, type JobOutputResult } from './api.ts'
 import { t } from './locales.ts'
 import css from './tasks-graph.module.css'
 
-/** Refresh cadence of an open job-output popover while its job runs. */
-const JOB_POLL_MS = 2000
 /** How long the kill button stays armed before it needs re-confirming. */
 const JOB_KILL_ARM_MS = 3000
 /** The agent count at which the drawer starts collapsed. */
@@ -47,6 +46,10 @@ function jobTone(job: SidebarJobView): TagTone {
 
 export interface JobsDrawerProps {
   rows: readonly TreeJob[]
+  /** The host client jobs service; absent → no kill control is offered. */
+  jobs: SidebarClientJobsService | undefined
+  /** The observed output of the job whose panel is open (host stream). */
+  observed: SidebarObservedJob | undefined
   /** Agent count of the tree (root included) — the auto-collapse signal. */
   agentCount: number
   /** Open the output popover of one row (anchor = the row's main button). */
@@ -56,7 +59,7 @@ export interface JobsDrawerProps {
 }
 
 export function JobsDrawer(props: JobsDrawerProps): ReactNode {
-  const { rows, agentCount, onOpenOutput, openJobId } = props
+  const { rows, agentCount, jobs, onOpenOutput, openJobId } = props
   const autoOpen = agentCount < JOBS_DRAWER_COLLAPSE_AT
   /** Manual override; undefined = follow the auto rule. */
   const [manualOpen, setManualOpen] = useState<boolean | undefined>(undefined)
@@ -90,17 +93,18 @@ export function JobsDrawer(props: JobsDrawerProps): ReactNode {
   }, [liveCount])
 
   const kill = useCallback(async (row: TreeJob): Promise<void> => {
+    if (jobs === undefined) return
     setKillingId(row.job.id)
     setKillErrorId(undefined)
     try {
-      await api.jobKill({ sessionId: row.ownerSessionId }, row.job.id)
+      await jobs.kill(row.ownerSessionId, row.job.id)
     } catch {
       setKillErrorId(row.job.id)
     } finally {
       setKillingId(undefined)
       setArmedId(undefined)
     }
-  }, [])
+  }, [jobs])
 
   if (rows.length === 0) return null
 
@@ -205,46 +209,24 @@ export function JobsDrawer(props: JobsDrawerProps): ReactNode {
  * keeps its two-click confirm.
  */
 export function JobOutputPopoverContent(props: {
-  ownerSessionId: string
+  /** The host client jobs service (absent → the panel only reports that). */
+  jobs: SidebarClientJobsService | undefined
   job: SidebarJobView
-  active: boolean
+  /** The owner session the roster carried the job under (kill + observation). */
+  ownerSessionId: string
+  /** The host's streamed observation of THIS job (undefined until it arrives). */
+  observed: SidebarObservedJob | undefined
 }): ReactNode {
-  const { ownerSessionId, job, active } = props
-  const [state, setState] = useState<'loading' | JobOutputResult | 'error'>('loading')
+  const { jobs, job, ownerSessionId, observed } = props
   const [follow, setFollow] = useState(true)
   const [copied, setCopied] = useState(false)
   const [armed, setArmed] = useState(false)
   const [killing, setKilling] = useState(false)
   const [killFailed, setKillFailed] = useState(false)
-  const controllerRef = useRef<AbortController | undefined>(undefined)
   const preRef = useRef<HTMLPreElement>(null)
   const live = isJobLive(job)
-
-  const load = useCallback(async (): Promise<void> => {
-    controllerRef.current?.abort()
-    const controller = new AbortController()
-    controllerRef.current = controller
-    try {
-      const result = await api.jobOutput({ sessionId: ownerSessionId }, job.id, controller.signal)
-      setState(result)
-    } catch {
-      // A newer pull aborted this one, or the wire failed: keep the last
-      // known output; only a popover that never loaded anything shows an error.
-      setState(current => (current === 'loading' ? 'error' : current))
-    }
-  }, [ownerSessionId, job.id])
-
-  useEffect(() => {
-    void load()
-    if (!active || !live) return
-    const timer = window.setInterval(() => { void load() }, JOB_POLL_MS)
-    return () => { window.clearInterval(timer) }
-    // isJobLive reads only job.status; the job object churns on every catalog
-    // refresh and must not restart the poll interval.
-    // eslint-disable-next-line react-hooks/exhaustive-deps
-  }, [load, active, job.status])
-
-  useEffect(() => () => { controllerRef.current?.abort() }, [])
+  const text = observed?.text ?? ''
+  const labels = useMemo(() => terminalBlockLabels(), [])
 
   // The armed kill disarms itself, like the drawer's button.
   useEffect(() => {
@@ -253,26 +235,22 @@ export function JobOutputPopoverContent(props: {
     return () => { window.clearTimeout(timer) }
   }, [armed])
 
-  // Terminal-tail behavior: each refresh pins the view to the newest output
+  // Terminal-tail behavior: every frame pins the view to the newest output
   // while the reader keeps the follow switch on.
   useEffect(() => {
-    if (!live || !follow || typeof state !== 'object' || state.text.length === 0) return
+    if (!live || !follow || text.length === 0) return
     const pre = preRef.current
     if (pre !== null) pre.scrollTop = pre.scrollHeight
-    // Same as the poll effect: only the status transition matters.
-    // eslint-disable-next-line react-hooks/exhaustive-deps
-  }, [state, job.status, follow])
+  }, [text, live, follow])
 
-  const text = typeof state === 'object' ? state.text : ''
-
-  /** Two-click kill from the popover (then re-read so the row settles). */
+  /** Two-click kill from the popover (the roster settles the row afterwards). */
   const kill = async (): Promise<void> => {
+    if (jobs === undefined) return
     setKilling(true)
     setKillFailed(false)
     try {
-      await api.jobKill({ sessionId: ownerSessionId }, job.id)
+      await jobs.kill(ownerSessionId, job.id)
       setArmed(false)
-      await load()
     } catch {
       setKillFailed(true)
     } finally {
@@ -311,16 +289,26 @@ export function JobOutputPopoverContent(props: {
         {job.detail !== undefined && job.detail !== '' && <span>{job.detail}</span>}
         <span className={css.popHint}>{t('jobDragHint')}</span>
       </div>
-      {state === 'loading' && <div className={css.popHint}>{t('loading')}</div>}
-      {state === 'error' && <div className={css.popError}>{t('jobOutputError')}</div>}
-      {typeof state === 'object' && (
+      {observed === undefined && <div className={css.popHint}>{t('loading')}</div>}
+      {observed !== undefined && (
         <>
-          {state.text.length > 0
-            ? <pre ref={preRef} className={`${css.popPre} ${css.jobPopPre}`} data-popover-no-drag>{state.text}</pre>
-            : state.read
-              ? <div className={css.popHint}>{t('jobNoOutput')}</div>
-              : <div className={css.popHint}>{t('jobNotReadYet')}</div>}
-          {state.truncated && <div className={css.popHint}>{t('jobOutputTruncated')}</div>}
+          {observed.gapBefore && <div className={css.popHint}>{t('jobOutputTruncated')}</div>}
+          {observed.error !== undefined && <div className={css.popError}>{t('jobOutputError')}</div>}
+          {observed.text.length > 0
+            ? (
+              <div ref={preRef as unknown as React.RefObject<HTMLDivElement>} data-popover-no-drag>
+                <TerminalBlock
+                  command={job.label}
+                  output={observed.text}
+                  running={observed.streaming}
+                  runStateDot={false}
+                  maxLines={Number.POSITIVE_INFINITY}
+                  copyText={job.label}
+                  labels={labels}
+                />
+              </div>
+            )
+            : <div className={css.popHint}>{t('jobNoOutput')}</div>}
         </>
       )}
       <div className={css.jobPopActions} data-popover-no-drag>

src/client/SubagentView.tsx

@@ -34,8 +34,9 @@ import {
 } from '@deepseek-ai/dsh-client-ui-primitives'
 import type {
   Context,
+  SidebarClientJobsService,
+  SidebarObservedJob,
   SidebarChildLiveView,
-  SidebarJobView,
   SidebarSessionList,
   SidebarSubagentAddress,
 } from '../context-types.ts'
@@ -46,9 +47,12 @@ import {
 } from './subagent-detect.ts'
 import { subagentCatalogs } from './subagent-catalog.ts'
 import { treeSessionIds } from './subagent-lineage.ts'
-import { collectTreeJobs, orderJobs } from './subagent-jobs.ts'
+import { orderJobs, type TreeJob } from './subagent-jobs.ts'
 import { api, type TeamsViewResult } from './api.ts'
 import { usePolling } from './use-polling.ts'
+import {
+  clientJobs, collectRows, useJobObservation, useJobWatchers, useJobsSnapshot,
+} from './jobs-client.ts'
 import { t } from './locales.ts'
 import { buildTasksModel, type TasksAgentNode, type TasksWorkflowNode } from './tasks-model.ts'
 import { TasksGraph } from './TasksGraph.tsx'
@@ -66,8 +70,6 @@ import legacy from './SubagentView.module.css'
 const POLL_MS = 3000
 /** Poll cadence of the workflow-run and team views while the page is visible. */
 const TELEMETRY_POLL_MS = 5000
-/** Poll cadence of the background-job lists (one read per tree session). */
-const JOBS_POLL_MS = 3000
 
 /**
  * The workspace face's "show this conversation" verb. DSH moved child and
@@ -156,43 +158,34 @@ function useTeamView(
 }
 
 /**
- * The background-job lists of the whole tree, keyed by OWNER session.
- *
- * DSH 0.1.7 dropped the client session snapshot's `jobsBySession` push mirror,
- * so the registry is read through the plugin's own `jobs.list` route — and the
- * registry's access fence admits a job to its OWNER session only, so the whole
- * tree is one read per tree session, fanned out on each tick. A host without
- * the jobs service (503) or one dropped read degrades to an empty set for THAT
- * session; the next tick retries.
+ * The tree's background jobs, read through the HOST's client jobs service
+ * (`ctx.jobs` — see jobs-client.ts). Watching is per tree session (the host
+ * shares one stream between watchers) and only while the page is on screen;
+ * a deployment without the service renders no jobs surface at all.
  */
 function useTreeJobs(
+  ctx: Context,
   byId: SidebarSessionList['byId'],
   rootId: string | undefined,
   active: boolean,
-): Readonly<Record<string, readonly SidebarJobView[]>> {
+): {
+  rows: TreeJob[]
+  jobs: SidebarClientJobsService | undefined
+  /** The observed output per job id (the panel picks its own entry). */
+  observed: Readonly<Record<string, SidebarObservedJob | undefined>>
+} {
+  const jobs = useMemo(() => clientJobs(ctx), [ctx])
+  const snapshot = useJobsSnapshot(jobs)
   const treeIds = useMemo(
     () => (rootId === undefined ? [] : [...treeSessionIds(byId, rootId)]),
     [byId, rootId],
   )
-  const [jobsBySession, setJobsBySession] = useState<Readonly<Record<string, readonly SidebarJobView[]>>>({})
-  useEffect(() => { setJobsBySession({}) }, [rootId])
-  const poll = useCallback(async (signal: AbortSignal): Promise<void> => {
-    const entries = await Promise.all(treeIds.map(async (sessionId): Promise<[string, readonly SidebarJobView[]]> => {
-      try {
-        const result = await api.jobsList(sessionId, signal)
-        return [sessionId, result.jobs]
-      } catch {
-        return [sessionId, []]
-      }
-    }))
-    if (!signal.aborted) setJobsBySession(Object.fromEntries(entries))
-  }, [treeIds])
-  usePolling(active && treeIds.length > 0, poll, {
-    intervalMs: JOBS_POLL_MS,
-    mode: 'self-scheduling',
-    immediate: true,
-  })
-  return jobsBySession
+  useJobWatchers(jobs, treeIds, active)
+  const rows = useMemo(
+    () => (rootId === undefined ? [] : orderJobs(collectRows(snapshot, byId, new Set(treeIds)))),
+    [snapshot, byId, rootId, treeIds],
+  )
+  return { rows, jobs, observed: snapshot.observed }
 }
 
 /** The open popover of the page (one at a time, anchored). */
@@ -380,12 +373,15 @@ export function SubagentView(props: {
   )
   const agentCount = totals.count + 1
 
-  /** The tree's ordered job rows (owner-labeled, read per tree session). */
-  const jobsBySession = useTreeJobs(byId, rootId, active)
-  const jobRows = useMemo(
-    () => orderJobs(collectTreeJobs(byId, jobsBySession, rootId)),
-    [byId, jobsBySession, rootId],
-  )
+  /** The tree's ordered job rows (owner-labeled, off the host's push roster). */
+  const treeJobs = useTreeJobs(ctx, byId, rootId, active)
+  const jobRows = treeJobs.rows
+  // Only the open panel's job is observed: the host streams output per observer.
+  const openJob = popover?.kind === 'job'
+    ? jobRows.find(row => row.job.id === popover.jobId)
+    : undefined
+  useJobObservation(treeJobs.jobs, openJob?.ownerSessionId, openJob?.job.id)
+  const observedJob = openJob === undefined ? undefined : treeJobs.observed[openJob.job.id]
 
   /** Catalogs that failed to load (surfaced as one banner in both modes). */
   const failedParents = useMemo(
@@ -456,7 +452,14 @@ export function SubagentView(props: {
     if (popover.kind === 'job') {
       const row = jobRows.find(candidate => candidate.job.id === popover.jobId)
       if (row === undefined) return null
-      return <JobOutputPopoverContent ownerSessionId={row.ownerSessionId} job={row.job} active={active} />
+      return (
+        <JobOutputPopoverContent
+          jobs={treeJobs.jobs}
+          ownerSessionId={row.ownerSessionId}
+          job={row.job}
+          observed={treeJobs.observed[row.job.id]}
+        />
+      )
     }
     const node = model.find(candidate => candidate.id === popover.nodeId)
     if (node === undefined) return null
@@ -587,6 +590,8 @@ export function SubagentView(props: {
       <JobsDrawer
         rows={jobRows}
         agentCount={agentCount}
+        jobs={treeJobs.jobs}
+        observed={observedJob}
         openJobId={popover?.kind === 'job' ? popover.jobId : undefined}
         onOpenOutput={(row, anchor) => {
           setPopover(popover?.kind === 'job' && popover.jobId === row.job.id

src/client/api.ts

@@ -11,7 +11,6 @@ import type { SidechatLiveEvent, SidechatLogEvent, SidechatThreadInfo } from '..
 import type {
   SidebarCreateTeamTaskRequest,
   SidebarChildLiveView,
-  SidebarJobView,
   SidebarSessionEvent,
   SidebarTeamMemberView,
   SidebarTeamTaskMutationResult,
@@ -104,21 +103,6 @@ export interface FsTextResult { kind: 'text'; content: string; truncated: boolea
  *  `head` carries the first bytes (base64) for viewer detect sniffing. */
 export interface FsBinaryResult { kind: 'binary'; size: number; truncated: boolean; head: string }
 
-/**
- * One jobs.output response: the output the MODEL has read so far for the
- * job (replayed from the owner session's event log — the model's
- * job_output cursor is never touched, so the pane can never steal the
- * agent's bytes). `read` is false until the model actually called
- * job_output for the job.
- */
-export interface JobOutputResult {
-  text: string
-  /** True when the host capped the text at its output limit. */
-  truncated: boolean
-  /** Whether the model has read the job at least once. */
-  read: boolean
-}
-
 /** The `subagents.live` response: one row per tree child (and the root). */
 export type SubagentLiveResult = { live: Record<string, SidebarChildLiveView> }
 
@@ -348,30 +332,6 @@ export const api = {
   /** Cherry-pick one commit onto the current branch. */
   gitCherryPick: (scope: SessionScope, hash: string, worktree?: string) =>
     call<{ ok: true }>('git.cherry-pick', gitPayload(scope, worktree, { hash })),
-  /**
-   * The output the model has read so far for one background job (replayed
-   * from the owner session's event log — never the model's job_output
-   * cursor). The scope MUST be the job's OWNER session.
-   */
-  jobOutput: (scope: SessionScope, id: string, signal?: AbortSignal) =>
-    call<JobOutputResult>('jobs.output', scopePayload(scope, { id }), signal),
-  /**
-   * The background-job list of one session (the Tasks page's jobs section and
-   * the job auto-open trigger). DSH 0.1.7 dropped the client session
-   * snapshot's jobs mirror, so the registry is read through the plugin's own
-   * `jobs.list` route. The registry's access fence admits a job to its OWNER
-   * session (and to unowned jobs) only, so a caller that needs the whole tree
-   * asks once per tree session. A host without the jobs service answers 503:
-   * the rejection is the caller's to degrade from (an empty section).
-   */
-  jobsList: (sessionId: string, signal?: AbortSignal) =>
-    call<{ jobs: SidebarJobView[] }>('jobs.list', { sessionId }, signal),
-  /** Request cancellation of one background job (live jobs flip to stopping). */
-  jobKill: (scope: SessionScope, id: string, reason?: string) =>
-    call<{ ok: true; outcome: 'requested' | 'already-finished' }>('jobs.kill', scopePayload(scope, {
-      id,
-      ...(reason !== undefined ? { reason } : {}),
-    })),
   /**
    * One batch live-preview fetch for the whole Subagent tree. The payload is
    * the already-resolved topology ROOT (not a session scope); the host

src/client/block-labels.ts

@@ -0,0 +1,69 @@
+/**
+ * Display copy for the host primitives the sidebar renders OUTSIDE the main
+ * conversation: `TerminalBlock` (a captured command's output surface) is
+ * cordis-free and takes every label as a prop, so the owning render site must
+ * supply them.
+ *
+ * One builder, two consumers (the Side Chat tool card and the Tasks page's
+ * background-job output panel): the two surfaces must read identically, and a
+ * new required label on the host side has exactly one place to land.
+ */
+import { t } from './locales.ts'
+
+/** Labels for the sidebar's markdown / diff / read Block chrome. */
+export interface BlockLabels {
+  copy: string
+  copied: string
+  collapse: string
+  collapseAria: string
+  expand: (hidden: number) => string
+  expandAria: (hidden: number) => string
+}
+
+/**
+ * The sidebar's Blocks reuse the Side Chat dictionary's chrome family (copy
+ * pair + the collapse/expand family that is common to every Block kind).
+ */
+export function blockLabels(): BlockLabels {
+  return {
+    copy: t('copy'),
+    copied: t('copied'),
+    collapse: t('sideChatBlockCollapse'),
+    collapseAria: t('sideChatBlockCollapseAria'),
+    expand: (hidden: number) => t('sideChatBlockExpand', { hidden }),
+    expandAria: (hidden: number) => t('sideChatBlockExpandAria', { hidden }),
+  }
+}
+
+/** The `TerminalBlockLabels` shape (structural mirror; the host owns the type). */
+export interface TerminalLabels {
+  signal: (signal: string) => string
+  exitCode: (exitCode: number) => string
+  noExitCode: string
+  running: string
+  failed: string
+  done: string
+  copy: string
+  copied: string
+  noOutput: string
+  collapseAria: string
+  collapse: string
+  expandAria: (hidden: number) => string
+  expand: (hidden: number) => string
+}
+
+/** Labels for one `TerminalBlock` (a captured command plus its output). */
+export function terminalBlockLabels(): TerminalLabels {
+  return {
+    ...blockLabels(),
+    signal: (signal: string) => t('sideChatBlockSignal', { signal }),
+    exitCode: (exitCode: number) => t('sideChatBlockExitCode', { code: exitCode }),
+    // DSH 0.1.6-alpha.2 added this pill text: a settle the view cannot
+    // name (killed by an unknown signal, or never started).
+    noExitCode: t('sideChatBlockNoExitCode'),
+    running: t('sideChatBlockRunning'),
+    failed: t('sideChatBlockFailed'),
+    done: t('sideChatBlockDone'),
+    noOutput: t('sideChatBlockNoOutput'),
+  }
+}

src/client/jobs-client.ts

@@ -0,0 +1,125 @@
+/**
+ * The Tasks page's read of the HOST's client jobs service (`ctx.jobs`).
+ *
+ * The plugin used to carry its own three-route jobs transport: a per-session
+ * `jobs.list` fan-out every 3s, a `job_output` trace replayed from the owner
+ * session's event log (the registry's only read is the model's CONSUMING
+ * cursor, which a human pane must never move), and a kill passthrough. The
+ * host service replaces all three — and this module is the whole data layer of
+ * that swap, kept apart from the DRAWER so the auto-open trigger in
+ * `sidebar/use-host-feeds.ts` reads the same feed the page renders.
+ *
+ * Structural detection only (the plugin never imports the host package): a
+ * deployment without the service sees `undefined` and renders no jobs surface.
+ */
+import { useCallback, useEffect, useMemo, useSyncExternalStore } from 'react'
+import type { SidebarSessionList } from '../context-types.ts'
+import type {
+  Context,
+  SidebarClientJobsService,
+  SidebarJobsSnapshot,
+} from '../context-types.ts'
+import { type TreeJob } from './subagent-jobs.ts'
+
+/** The empty snapshot a deployment without the jobs service reads. */
+const EMPTY_JOBS: SidebarJobsSnapshot = { rows: {}, observed: {} }
+
+/**
+ * The client jobs service, when this deployment mounts the host jobs UI.
+ * @param ctx - the client context.
+ */
+export function clientJobs(ctx: Context): SidebarClientJobsService | undefined {
+  const service = ctx.get('jobs') as SidebarClientJobsService | undefined
+  if (service === undefined) return undefined
+  if (typeof service.watchRows !== 'function' || typeof service.observe !== 'function') return undefined
+  if (typeof service.kill !== 'function' || service.state === undefined) return undefined
+  return service
+}
+
+/**
+ * Subscribe to the jobs snapshot (roster + observations).
+ * @param jobs - the client jobs service (absent → an empty snapshot forever).
+ */
+export function useJobsSnapshot(jobs: SidebarClientJobsService | undefined): SidebarJobsSnapshot {
+  const subscribe = useMemo(
+    () => (callback: () => void) => (jobs === undefined ? () => {} : jobs.state.subscribe(callback)),
+    [jobs],
+  )
+  const read = useCallback(() => (jobs === undefined ? EMPTY_JOBS : jobs.state.getSnapshot()), [jobs])
+  return useSyncExternalStore(subscribe, read)
+}
+
+/**
+ * Watch (and release) one roster per session, while the page is on screen. The
+ * host shares a single stream between watchers of one session, so this is
+ * cheap; the key is the joined id list so a re-render with the same tree does
+ * not re-subscribe.
+ * @param jobs - the client jobs service.
+ * @param sessionIds - the tree's sessions (the root included).
+ * @param active - whether the page is visible at all.
+ */
+export function useJobWatchers(
+  jobs: SidebarClientJobsService | undefined,
+  sessionIds: readonly string[],
+  active: boolean,
+): void {
+  const key = sessionIds.join(',')
+  useEffect(() => {
+    if (jobs === undefined || !active || key === '') return
+    const releases = key.split(',').map(sessionId => jobs.watchRows(sessionId))
+    return () => {
+      for (const release of releases) release()
+    }
+  }, [jobs, key, active])
+}
+
+/**
+ * Observe ONE job's output while its panel is open; the release runs on close.
+ * @param jobs - the client jobs service.
+ * @param ownerSessionId - the session the roster carried the job under.
+ * @param jobId - the observed job (undefined closes the observation).
+ */
+export function useJobObservation(
+  jobs: SidebarClientJobsService | undefined,
+  ownerSessionId: string | undefined,
+  jobId: string | undefined,
+): void {
+  useEffect(() => {
+    if (jobs === undefined || jobId === undefined) return
+    return jobs.observe(ownerSessionId, jobId)
+  }, [jobs, ownerSessionId, jobId])
+}
+
+/**
+ * One job's rows as the drawer renders them, deduped across the tree.
+ *
+ * An UNOWNED job is visible to every watched session, so the union of rosters
+ * repeats it: rows are keyed by job id and attributed to the watched session
+ * whose roster carried them (which is also the session the observation and the
+ * kill go through). Jobs owned by a session outside the tree are dropped.
+ *
+ * @param snapshot - the client service snapshot.
+ * @param byId - the session mirror (owner labels).
+ * @param treeIds - the tree's session ids (root included).
+ */
+export function collectRows(
+  snapshot: SidebarJobsSnapshot,
+  byId: SidebarSessionList['byId'],
+  treeIds: ReadonlySet<string>,
+): TreeJob[] {
+  const rows: TreeJob[] = []
+  const seen = new Set<string>()
+  for (const [watched, jobs] of Object.entries(snapshot.rows)) {
+    for (const job of jobs) {
+      if (seen.has(job.id)) continue
+      if (job.owner !== undefined && !treeIds.has(job.owner)) continue
+      seen.add(job.id)
+      rows.push({
+        ownerSessionId: watched,
+        ownerTitle: byId[job.owner ?? watched]?.displayTitle ?? job.owner ?? watched,
+        job,
+      })
+    }
+  }
+  return rows
+}

src/client/sidebar/use-host-feeds.ts

@@ -5,30 +5,20 @@
  * jump-back). All of it reacts to the host's live feeds for the CURRENT
  * session; the sidebar shell only consumes the returned jump-back ref.
  */
-import { useCallback, useEffect, useRef } from 'react'
+import { useEffect, useRef } from 'react'
 import type { Context, SidebarJobView, SidebarSessionList } from '../../context-types.ts'
 import type { SidebarStore } from '../state.ts'
 import { isNarrowWidth } from '../breakpoints.ts'
 import { detectNewDirectSubagent } from '../subagent-detect.ts'
 import { detectNewJob } from '../subagent-jobs.ts'
-import { api } from '../api.ts'
+import { clientJobs, useJobsSnapshot } from '../jobs-client.ts'
 import { mountedSessionId } from '../native/surface.ts'
-import { usePolling } from '../use-polling.ts'
 import { t } from '../locales.ts'
 
 /** How many consecutive reconnect failures stop the agent-opens push loop
  * (the loop restarts on session switch). */
 const FAILURE_LIMIT = 3
 
-/**
- * Background-job poll cadence (ms) for the job auto-open trigger. DSH 0.1.7
- * removed the client session snapshot's jobs mirror, so the trigger reads the
- * plugin's `jobs.list` route on a timer instead of reacting to a push. One
- * in-process registry read per tick bounds the auto-open latency at this
- * value.
- */
-const JOB_POLL_MS = 2_000
-
 /**
  * Subagent auto-open debounce (ms). The host delivers a new child's origin
  * and its title in SEPARATE frames: a Side Chat thread's first visible
@@ -235,42 +225,44 @@ export function useHostFeeds(feeds: {
    * (0 → N only), ANY new job id triggers: the agent may start several jobs in
    * one session, and each should surface.
    *
-   * The list is POLLED: DSH 0.1.7 stopped mirroring background jobs into the
-   * client session snapshot, so the registry (fenced per OWNER session) is
-   * read through the plugin's `jobs.list` route. A fresh page load never
-   * triggers — the FIRST successful read only arms the baseline — and a host
-   * without the jobs service (503) leaves the baseline unarmed, so the first
-   * list that does arrive is never mistaken for new work.
+   * The roster is PUSHED by the host's own client jobs service: one
+   * reference-counted `watchRows` stream for the current session replaces the
+   * 3s `jobs.list` poll this used to run. A fresh page load never triggers —
+   * the FIRST frame only arms the baseline — and a deployment without the
+   * jobs service leaves the baseline unarmed, so the first roster that does
+   * arrive is never mistaken for new work.
+   *
+   * `detectNewJob` also compares against the moment the watch STARTED: the
+   * first frame of a session that already had running jobs must not surface
+   * them as new work.
    */
+  const jobsService = clientJobs(ctx)
+  const jobsSnapshot = useJobsSnapshot(jobsService)
   const jobBaselineRef = useRef<readonly SidebarJobView[] | undefined>(undefined)
-  // A session switch voids the previous session's baseline BEFORE the poller
-  // below restarts on the new id (effects run in declaration order).
-  useEffect(() => { jobBaselineRef.current = undefined }, [sessionId])
-  const pollJobs = useCallback(async (signal: AbortSignal): Promise<void> => {
-    if (sessionId === undefined) return
-    let jobs: readonly SidebarJobView[]
-    try {
-      const result = await api.jobsList(sessionId, signal)
-      jobs = result.jobs
-    } catch {
-      // No jobs service, or a dropped read: keep the last baseline and let the
-      // next tick retry (never treat an unreadable list as an empty one).
-      return
+  const jobWatchStartRef = useRef<number | undefined>(undefined)
+  useEffect(() => {
+    if (jobsService === undefined || sessionId === undefined) return
+    const release = jobsService.watchRows(sessionId)
+    // A session switch voids the previous session's baseline AND its clock.
+    jobWatchStartRef.current = Date.now()
+    jobBaselineRef.current = undefined
+    return () => {
+      release()
+      jobBaselineRef.current = undefined
+      jobWatchStartRef.current = undefined
     }
-    if (signal.aborted) return
+  }, [jobsService, sessionId])
+  useEffect(() => {
+    if (jobsService === undefined || sessionId === undefined) return
+    const rows = jobsSnapshot.rows[sessionId] ?? []
     const prev = jobBaselineRef.current
-    jobBaselineRef.current = jobs
+    jobBaselineRef.current = rows
     if (prev === undefined) return
-    if (!detectNewJob(prev, jobs)) return
+    if (!detectNewJob(prev, rows, jobWatchStartRef.current)) return
     if (!store.getPrefs().autoOpenJobs) return
     if (ctx.get('betterSidebar')?.isTabEnabled('subagent') === false) return
     activateTasksPage(ctx, sessionId, { background: true })
-  }, [sessionId, store, ctx])
-  usePolling(sessionId !== undefined, pollJobs, {
-    intervalMs: JOB_POLL_MS,
-    mode: 'self-scheduling',
-    immediate: true,
-  })
+  }, [jobsService, jobsSnapshot, sessionId, store, ctx])
 
   /**
    * Topology jump-back: clicking a subagent node on the Subagent page calls

src/client/subagent-jobs.ts

@@ -9,7 +9,6 @@
  * status mapping mirror the official ui-jobs header list.
  */
 import type {
-  SidebarSessionList,
   SidebarJobStatus,
   SidebarJobView,
 } from '../context-types.ts'
@@ -35,42 +34,31 @@ export { treeSessionIds }
 
 /**
  * Whether a NEW background job appeared for the current session between two
- * consecutive reads of its job list (a job id the previous list lacked).
- * Unlike the subagent auto-open (0 → N only), ANY new job id triggers: the
- * agent may start several jobs over a session, and each new one should
- * surface the Tasks page containing the background-jobs section. A fresh page
- * load never triggers — the caller builds its baseline from the first list it
- * reads instead of comparing against one (see the poller in
- * sidebar/use-host-feeds.ts).
+ * roster frames (a job id the previous frame lacked). Unlike the subagent
+ * auto-open (0 → N only), ANY new job id triggers: the agent may start
+ * several jobs over a session, and each new one should surface the Tasks page
+ * that contains the background-jobs section.
+ *
+ * `since` (epoch ms) is the moment the caller STARTED watching the roster, and
+ * it is what keeps a page load honest now that the roster arrives as a push
+ * stream: the host's client jobs model drops a session's key when it sees no
+ * jobs, so an empty first frame is indistinguishable from "not delivered yet".
+ * A job that already existed when the watcher opened therefore looks brand new
+ * to an id-set diff — comparing its start time against `since` filters it out,
+ * while a job the agent starts afterwards still triggers.
+ *
+ * @param prev - the previous frame's rows.
+ * @param next - the current frame's rows.
+ * @param since - epoch ms the watch began; omitted skips the age test.
+ * @returns whether genuinely new work appeared.
  */
 export function detectNewJob(
   prev: readonly SidebarJobView[],
   next: readonly SidebarJobView[],
+  since?: number,
 ): boolean {
   const prevIds = new Set(prev.map(job => job.id))
-  return next.some(job => !prevIds.has(job.id))
-}
-
-/**
- * Collect the background jobs of the whole current tree, owner-labeled. The
- * map is keyed by the OWNER session (the fence's unit of access): sessions the
- * caller could not read, or that returned nothing, contribute nothing — an
- * absent entry is an empty set, never a dropped row of another tree.
- */
-export function collectTreeJobs(
-  byId: SidebarSessionList['byId'],
-  jobsBySession: Readonly<Record<string, readonly SidebarJobView[]>> | undefined,
-  rootId: string | undefined,
-): TreeJob[] {
-  const rows: TreeJob[] = []
-  if (jobsBySession === undefined) return rows
-  for (const sessionId of treeSessionIds(byId, rootId)) {
-    const jobs = jobsBySession[sessionId]
-    if (jobs === undefined || jobs.length === 0) continue
-    const ownerTitle = byId[sessionId]?.displayTitle ?? sessionId
-    for (const job of jobs) rows.push({ ownerSessionId: sessionId, ownerTitle, job })
-  }
-  return rows
+  return next.some(job => !prevIds.has(job.id) && (since === undefined || job.startedAt >= since))
 }
 
 /**

src/context-types.ts

@@ -210,8 +210,16 @@ export interface SidebarJobView {
   kind: string
   /** Producer-supplied one-line label: the command, or the delegation description. */
   label: string
+  /**
+   * Owning session; absent for an unowned job, which every caller can see.
+   * The client service's roster is keyed by WATCHED session, so this is what
+   * tells a row whether it belongs to the tree on screen.
+   */
+  owner?: string
   /** Current lifecycle state. */
   status: SidebarJobStatus
+  /** The producer's live progress line ('3/10'), cleared at settlement. */
+  progress?: string
   /** Kind-specific status detail ('exit code: 3'), present once supplied. */
   detail?: string
   /** Epoch ms when the job was registered. */
@@ -230,11 +238,51 @@ export interface SidebarJobView {
  * reads `list` through the plugin's own `jobs.list` route (the client session
  * snapshot stopped mirroring background jobs in the same release).
  */
-export interface SidebarJobsService {
-  /** Caller-owned and unowned jobs in registration order (the Tasks page list). */
-  list(caller?: string): SidebarJobView[]
-  /** Request cancellation; throws for an unknown or foreign job. */
-  kill(id: string, caller?: string, reason?: string): 'requested' | 'already-finished'
+/**
+ * One job's retained output as the HOST's client service observes it
+ * (mirror of `@deepseek-ai/dsh-api-job-controller/client`'s observed entry).
+ * `gapBefore` marks output the observer missed between frames; `error` carries
+ * a producer-side failure; `streaming` is true while the job still writes.
+ */
+export interface SidebarObservedJob {
+  text: string
+  gapBefore: boolean
+  streaming: boolean
+  error?: string
+}
+
+/** The host client jobs service's snapshot (roster by session + observations). */
+export interface SidebarJobsSnapshot {
+  /** Whole-set roster per WATCHED session: empty arrays are how a session ends. */
+  rows: Record<string, SidebarJobView[]>
+  /** Retained output per observed job id. */
+  observed: Record<string, SidebarObservedJob | undefined>
+}
+
+/**
+ * The host's CLIENT jobs service (`ctx.jobs`, mounted by the web profile's
+ * `@deepseek-ai/dsh-api-job-controller/client`), structurally mirrored: the
+ * plugin never imports the host package.
+ *
+ * It replaces the plugin's own three-route jobs transport: `watchRows` keeps
+ * one session's roster current as a push stream, `observe` streams a job's
+ * retained output WITHOUT moving the model's consuming cursor, and `kill` is
+ * the same registry admission the old route forwarded.
+ */
+export interface SidebarClientJobsService {
+  state: {
+    getSnapshot(): SidebarJobsSnapshot
+    subscribe(listener: () => void): () => void
+  }
+  /** Watch one session's roster; returns the release function (ref-counted). */
+  watchRows(sessionId: string): () => void
+  /**
+   * Observe one job's output; returns the release function. `sessionId` may be
+   * undefined for a job whose owner the roster has not resolved yet.
+   */
+  observe(sessionId: string | undefined, jobId: string): () => void
+  /** Request cancellation of one job (owned by `sessionId`). */
+  kill(sessionId: string, jobId: string): Promise<void>
 }
 
 /** The host agent registry face (structural mirror of the runtime `ctx.agents`). */
@@ -660,7 +708,7 @@ export interface SidebarContextShape {
   /** The client module system (rc.8+ chunk-loader externals). */
   modules: { import(specifier: string): Promise<unknown> }
   /** The host background-job registry (optional; routes degrade to 503). */
-  jobs: SidebarJobsService
+  jobs: SidebarClientJobsService
   /** The host live-agent registry (optional; side chat thread agents). */
   agents: SidebarAgentsService
   /**

src/index.ts

@@ -41,7 +41,6 @@ import { launchExternal } from './open-external.ts'
 import * as git from './git.ts'
 import { SettingsConflictError } from '@deepseek-ai/dsh-settings'
 import { AgentOpenRegistry, registerOpenTool, type AgentOpenRequest } from './agent-opens.ts'
-import { buildJobsApi, type SidebarJobsRoutes } from './jobs-routes.ts'
 import { buildSubagentLiveApi, type SidebarSubagentLiveRoutes } from './subagent-live-route.ts'
 import { buildTeamsApi, type SidebarTeamsRoutes } from './team-routes.ts'
 import { buildWorkflowsApi, type SidebarWorkflowRoutes } from './workflow-routes.ts'
@@ -264,12 +263,6 @@ function buildApi(
     const requested = typeof record?.worktree === 'string' && record.worktree !== '' ? record.worktree : undefined
     return { sessionId: base.sessionId, cwd: await git.resolveWorktree(base.cwd, requested) }
   }
-  // Background jobs: the LIST rides the harness's `session/jobs` push
-  // mirror, so these routes only replay output the model has read (from the
-  // session's own event log — no DSH source is touched, the model's
-  // job_output cursor is never consumed) and kill (the registry's stock
-  // API). A deployment without the jobs registry downgrades kill to a 503.
-  const jobsApi: SidebarJobsRoutes = buildJobsApi(ctx, resolved.readLimit)
   // Subagent live previews: one batch request instead of N per-child
   // `subagents.history` calls. The route degrades to a 503 when the host
   // subagent runtime is absent (the page has no topology to show anyway).
@@ -484,16 +477,6 @@ function buildApi(
       const window = filtered.length > CHANGES_EVENTS_CAP ? filtered.slice(filtered.length - CHANGES_EVENTS_CAP) : filtered
       return { events: window, lastSeq: window.at(-1)?.seq ?? afterSeq }
     },
-    // Background jobs: list the caller's own jobs, read one job's output (a
-    // REPLAY of what the model has read so far, from the owner session's
-    // event log — the model's job_output cursor is never touched, so the
-    // human pane can never steal the agent's bytes), and kill one job. The
-    // list route exists because DSH 0.1.7 deleted the session/jobs push
-    // mirror the Tasks page used to read. Kill is fenced to the owning
-    // session by the jobs registry.
-    'jobs.list': (payload) => jobsApi.list(payload),
-    'jobs.output': (payload) => jobsApi.output(payload),
-    'jobs.kill': (payload) => jobsApi.kill(payload),
     // Subagent live previews: one batch request per refresh; the route folds
     // the newest text/tool activity of every running child in the tree.
     'subagents.live': (payload) => subagentLiveApi.live(payload),

src/jobs-routes.ts

@@ -1,296 +0,0 @@
-/**
- * Background-job routes of the /sidebar JSON API ('jobs.output' /
- * 'jobs.kill'). The job LIST needs no route: it arrives through the
- * harness's `session/jobs` push mirror (`jobsBySession` in the sessions
- * list feed). The routes:
- *
- * - 'jobs.output' — REPLAYS the output the MODEL has read so far for one
- *   job. The source is the owner session's own event log: `tool/call` rows
- *   of `job_output` name the job via `arguments.job_id`, and the paired
- *   `tool/result` rows carry the finalized content the model received.
- *   Because the session store's in-memory log can lag the live append feed
- *   after a host restart (the store session stays frozen at its
- *   rehydration boundary), the plugin ALSO mirrors job_output events from
- *   the live `session/event` feed and merges both sources (deduped by seq).
- *   This touches NO DSH source: the model's `job_output` cursor is never
- *   consumed, and the pane stays empty until the agent reads the job.
- * - 'jobs.kill' — the registry's stock `kill` (a pristine DSH API), fenced
- *   by the owning session id (the 0.1.7 registry compares `SessionId`, the
- *   0.1.6 one compared a live Agent). Absent registry → 503, mirroring the
- *   settings routes' optional-service downgrade.
- */
-import type { Context, SidebarJobsService, SidebarJobView, SidebarSessionEvent } from './context-types.ts'
-import { requireString, SidebarError } from './wire.ts'
-
-/** The background-job routes of the sidebar API. */
-export interface SidebarJobsRoutes {
-  /**
-   * The caller-visible jobs of one session, read straight from the registry.
-   *
-   * DSH 0.1.7 stopped mirroring background jobs into the client session list
-   * (`jobsBySession` is gone with no replacement), so the Tasks page reads
-   * them here instead. This is also the more authoritative source: the mirror
-   * was last-wins over push frames, while the registry is the state itself.
-   */
-  list(payload: unknown): { jobs: SidebarJobView[] }
-  /** The output the model has read so far for one job (event replay, capped). */
-  output(payload: unknown): { text: string; truncated: boolean; read: boolean }
-  /** Request cancellation of one job (live jobs flip to stopping). */
-  kill(payload: unknown): { ok: true; outcome: 'requested' | 'already-finished' }
-}
-
-/** The 'tool/result' message envelope inside a session event's data. */
-interface ToolResultMessageLike {
-  source?: { kind?: unknown; callId?: unknown }
-  content?: unknown
-  /** The 0.1.7 first-class tool message lifts the error flag onto the message. */
-  isError?: unknown
-}
-
-/** One 'tool-result' content block (the inner blocks carry the text). */
-interface ToolResultBlockLike {
-  type?: unknown
-  content?: unknown
-  isError?: unknown
-}
-
-/**
- * The result blocks and error flag of one tool/result message, read under BOTH
- * logged shapes: 0.1.6 wrapped the result in a single `type: 'tool-result'`
- * content block on a user-role message (the text nested inside it, `isError`
- * on the wrapper), 0.1.7's first-class tool-role message carries the blocks at
- * the message's own top level with `isError` lifted onto the message.
- * Historical logs keep the old shape forever, so both are read. Undefined when
- * the message carries no block array.
- */
-function resultOf(message: ToolResultMessageLike): { blocks: readonly unknown[]; isError: boolean } | undefined {
-  if (!Array.isArray(message.content)) return undefined
-  for (const block of message.content) {
-    if (block === null || typeof block !== 'object') continue
-    const wrapper = block as ToolResultBlockLike
-    if (wrapper.type !== 'tool-result') continue
-    return {
-      blocks: Array.isArray(wrapper.content) ? wrapper.content as readonly unknown[] : [],
-      isError: wrapper.isError === true,
-    }
-  }
-  return { blocks: message.content as readonly unknown[], isError: message.isError === true }
-}
-
-/**
- * Extract the plain text of a finalized tool result: its text blocks, joined
- * with newlines. Error results and non-text blocks contribute nothing.
- */
-function resultText(message: ToolResultMessageLike): string | undefined {
-  const blocks = resultOf(message)?.blocks
-  if (blocks === undefined) return undefined
-  const parts: string[] = []
-  for (const item of blocks) {
-    if (item === null || typeof item !== 'object') continue
-    const textItem = item as { type?: unknown; text?: unknown }
-    if (textItem.type === 'text' && typeof textItem.text === 'string') {
-      parts.push(textItem.text)
-    }
-  }
-  return parts.length > 0 ? parts.join('\n') : undefined
-}
-
-/** Whether a tool/result is an error result (0.1.7's message-level flag, else
- *  the 0.1.6 wrapper block's flag). */
-function resultIsError(message: ToolResultMessageLike): boolean {
-  return resultOf(message)?.isError === true
-}
-
-/** Whether a job_output result carries no new output — the controller's
- *  model-facing "(no new output)" body, noise for the human pane. */
-function isNoNewOutput(text: string): boolean {
-  return text.startsWith('(no new output)')
-}
-
-/** One compact job_output trace (a tool/call or its paired tool/result). */
-interface JobOutputTrace {
-  seq: number
-  kind: 'call' | 'result'
-  /** The tool call identity pairing the two rows. */
-  callId: string
-  /** tool/call: the job id parsed from the model arguments. */
-  jobId?: string
-  /** tool/result: the finalized text the model received. */
-  text?: string
-  /** tool/result: whether the result was an error (read counts, text skipped). */
-  isError?: boolean
-}
-
-/** Extract the job_output trace of one raw session event (undefined = unrelated). */
-function traceOf(event: SidebarSessionEvent): JobOutputTrace | undefined {
-  if (event.type === 'tool/call') {
-    const data = event.data as { name?: unknown; callId?: unknown; arguments?: unknown }
-    if (data.name !== 'job_output' || typeof data.callId !== 'string') return undefined
-    let jobId: string | undefined
-    try {
-      const args = JSON.parse(typeof data.arguments === 'string' ? data.arguments : '') as { job_id?: unknown }
-      if (typeof args.job_id === 'string') jobId = args.job_id
-    } catch {
-      // Malformed model arguments: not a job_output pair.
-    }
-    if (jobId === undefined) return undefined
-    return { seq: event.seq, kind: 'call', callId: data.callId, jobId }
-  }
-  if (event.type === 'tool/result') {
-    const message = (event.data as { message?: unknown }).message as ToolResultMessageLike | undefined
-    if (message === undefined) return undefined
-    const callId = message.source?.callId
-    if (typeof callId !== 'string') return undefined
-    return {
-      seq: event.seq,
-      kind: 'result',
-      callId,
-      text: resultText(message),
-      isError: resultIsError(message),
-    }
-  }
-  return undefined
-}
-
-/** Per-session cap of mirrored live traces (a bounded, lossy ring). */
-const MIRROR_MAX_ENTRIES = 200
-
-/**
- * The live job_output mirror: subscribes to the session append feed and
- * caches the job_output traces the session store's own log can lag behind
- * (after a host restart the store session stays frozen at its rehydration
- * boundary, so `session.events` misses everything appended since — the very
- * reads the pane exists to show). Zero DSH writes: the api-proxy pushes the
- * same feed to browsers.
- */
-function createJobOutputMirror(ctx: Context): { entries(sessionId: string): readonly JobOutputTrace[] } {
-  const perSession = new Map<string, JobOutputTrace[]>()
-  // tool/call identities per session, so unrelated tool/result rows are
-  // never cached (only job_output results pair with a cached call).
-  const callIds = new Map<string, Set<string>>()
-  if (typeof ctx.on !== 'function') {
-    // Test doubles without the event API degrade to seed-only replay.
-    return { entries: () => [] }
-  }
-  const dispose = ctx.on('session/event', (session, event) => {
-    const sessionId = (session as { id?: unknown } | null)?.id
-    if (typeof sessionId !== 'string') return
-    if (event.type === 'tool/call') {
-      const trace = traceOf(event)
-      if (trace?.kind !== 'call') return
-      let ids = callIds.get(sessionId)
-      if (ids === undefined) callIds.set(sessionId, ids = new Set())
-      ids.add(trace.callId)
-      push(sessionId, trace)
-    } else if (event.type === 'tool/result') {
-      const trace = traceOf(event)
-      if (trace?.kind !== 'result') return
-      if (!callIds.get(sessionId)?.has(trace.callId)) return
-      push(sessionId, trace)
-    }
-  })
-  ctx.effect(() => dispose, 'dsh-better-sidebar: job-output event mirror')
-
-  const push = (sessionId: string, trace: JobOutputTrace): void => {
-    let list = perSession.get(sessionId)
-    if (list === undefined) perSession.set(sessionId, list = [])
-    list.push(trace)
-    if (list.length > MIRROR_MAX_ENTRIES) {
-      const removed = list.splice(0, list.length - MIRROR_MAX_ENTRIES)
-      const ids = callIds.get(sessionId)
-      if (ids !== undefined) {
-        for (const entry of removed) {
-          if (entry.kind === 'call') ids.delete(entry.callId)
-        }
-        if (ids.size === 0) callIds.delete(sessionId)
-      }
-    }
-  }
-
-  return { entries: (sessionId) => perSession.get(sessionId) ?? [] }
-}
-
-/**
- * Build the jobs routes bound to the plugin context. `list` reads the
- * registry's own projection, `output` merges the owner session's event log
- * with the live job_output mirror, and `kill` cancels through the registry.
- * Every route that needs the registry degrades to a 503 when the deployment
- * lacks it.
- * @param ctx - host plugin context.
- * @param outputLimit - response cap for one output replay in bytes; longer
- *   texts are sliced and flagged `truncated` (mirrors the fs.read cap).
- */
-export function buildJobsApi(ctx: Context, outputLimit: number): SidebarJobsRoutes {
-  const jobs = ctx.get('jobs')
-  const mirror = createJobOutputMirror(ctx)
-  /** Registry refusals become a 404 job-error; unknown and foreign ids are indistinguishable. */
-  const registryError = (error: unknown): SidebarError =>
-    new SidebarError('job-error', error instanceof Error ? error.message : String(error), 404)
-  /** The registry, or the 503 every registry-backed route returns without it. */
-  const requireJobs = (): SidebarJobsService => {
-    if (jobs === undefined) {
-      throw new SidebarError('job-error', 'the background-job registry is not mounted in this deployment', 503)
-    }
-    return jobs
-  }
-  return {
-    list(payload) {
-      const sessionId = requireString(payload, 'sessionId')
-      try {
-        return { jobs: requireJobs().list(sessionId) }
-      } catch (error) {
-        if (error instanceof SidebarError) throw error
-        throw registryError(error)
-      }
-    },
-    output(payload) {
-      const sessionId = requireString(payload, 'sessionId')
-      const id = requireString(payload, 'id')
-      // Merge the store's event log (durable seed + whatever it received)
-      // with the live mirror, deduped by seq — a trace never double-counts.
-      const bySeq = new Map<number, JobOutputTrace>()
-      for (const event of ctx.sessions.get(sessionId)?.snapshotEvents() ?? []) {
-        const trace = traceOf(event)
-        if (trace !== undefined) bySeq.set(trace.seq, trace)
-      }
-      for (const trace of mirror.entries(sessionId)) bySeq.set(trace.seq, trace)
-      // Pair calls with results in seq order: the model's reads, oldest first.
-      const jobOf = new Map<string, string>()
-      const parts: string[] = []
-      let read = false
-      for (const trace of [...bySeq.values()].sort((left, right) => left.seq - right.seq)) {
-        if (trace.kind === 'call') {
-          if (trace.jobId !== undefined) jobOf.set(trace.callId, trace.jobId)
-        } else if (jobOf.get(trace.callId) === id) {
-          read = true
-          if (trace.isError !== true && trace.text !== undefined && !isNoNewOutput(trace.text)) {
-            parts.push(trace.text)
-          }
-        }
-      }
-      const text = parts.join('\n')
-      return {
-        text: text.length > outputLimit ? text.slice(0, outputLimit) : text,
-        truncated: text.length > outputLimit,
-        read,
-      }
-    },
-    kill(payload) {
-      if (jobs === undefined) {
-        throw new SidebarError('job-error', 'the background-job registry is not mounted in this deployment', 503)
-      }
-      const sessionId = requireString(payload, 'sessionId')
-      const id = requireString(payload, 'id')
-      const record = payload as { reason?: unknown } | null
-      const reason = typeof record?.reason === 'string' && record.reason !== ''
-        ? record.reason
-        : 'user requested via sidebar'
-      try {
-        // The 0.1.7 registry fences by SessionId, not by live Agent.
-        return { ok: true, outcome: jobs.kill(id, sessionId, reason) }
-      } catch (error) {
-        throw registryError(error)
-      }
-    },
-  }
-}

tests/jobs-routes.spec.ts

@@ -1,273 +0,0 @@
-/**
- * Host route tests for the background-job API ('jobs.output' / 'jobs.kill').
- * 'jobs.output' REPLAYS the output the model has read so far from the owner
- * session's event log (tool/call + tool/result pairs of job_output calls) —
- * the model's cursor is never touched, so nothing is stolen and unread jobs
- * report `read: false`. 'jobs.kill' uses the registry's stock kill, fenced
- * by the owning session, with a 503 when the registry is absent.
- */
-import { describe, expect, it, vi } from 'vitest'
-import { createToolResultMessage, ToolCallId } from '@deepseek-ai/dsh-llm'
-import { buildJobsApi } from '../src/jobs-routes.ts'
-import { SidebarError } from '../src/wire.ts'
-import type { Context, SidebarSessionEvent } from '../src/context-types.ts'
-
-/** A context whose `get` serves only the jobs face, with a session store. */
-function ctxWith(sessions: unknown, jobs: unknown): Context {
-  return {
-    sessions,
-    get: (key: string) => (key === 'jobs' ? jobs : undefined),
-  } as unknown as Context
-}
-
-/** A context that additionally captures the session/event listener (live mirror). */
-function ctxWithFeed(sessions: unknown): {
-  ctx: Context
-  emit: (session: unknown, event: SidebarSessionEvent) => void
-} {
-  let listener: ((session: unknown, event: SidebarSessionEvent) => void) | undefined
-  const base = ctxWith(sessions, undefined) as unknown as {
-    on: (event: string, fn: (session: unknown, event: SidebarSessionEvent) => void) => () => void
-    effect: (fn: () => void | (() => void)) => void
-  }
-  base.on = (_event: string, fn) => {
-    listener = fn
-    return () => { if (listener === fn) listener = undefined }
-  }
-  // The vendored cordis runs the registration effect immediately.
-  base.effect = (fn) => { fn() }
-  return {
-    ctx: base as unknown as Context,
-    emit: (session, event) => { listener?.(session, event) },
-  }
-}
-
-/** One job_output tool/call event. */
-function jobOutputCall(seq: number, callId: string, jobId: string): SidebarSessionEvent {
-  return { type: 'tool/call', seq, time: seq, data: { name: 'job_output', callId, arguments: JSON.stringify({ job_id: jobId }) } }
-}
-
-/** One tool/result event carrying the finalized text the model received —
- *  the 0.1.7 first-class tool-role shape (content + isError on the message). */
-function jobOutputResult(seq: number, callId: string, text: string, over: { isError?: boolean } = {}): SidebarSessionEvent {
-  return {
-    type: 'tool/result',
-    seq,
-    time: seq,
-    data: {
-      message: {
-        role: 'tool',
-        source: { kind: 'tool', callId },
-        toolCallId: callId,
-        isError: over.isError === true,
-        content: [{ type: 'text', text }],
-      },
-    },
-  }
-}
-
-/** The retired 0.1.6 shape: role-'user' message wrapping the result in one
- *  `tool-result` block. Historical logs still carry it. */
-function legacyJobOutputResult(seq: number, callId: string, text: string, over: { isError?: boolean } = {}): SidebarSessionEvent {
-  return {
-    type: 'tool/result',
-    seq,
-    time: seq,
-    data: {
-      message: {
-        role: 'user',
-        source: { kind: 'tool', callId },
-        content: [{
-          type: 'tool-result',
-          toolCallId: callId,
-          isError: over.isError === true,
-          content: [{ type: 'text', text }],
-        }],
-      },
-    },
-  }
-}
-
-/** The owner session with the given event log. */
-function session(events: SidebarSessionEvent[]): unknown {
-  return { header: { cwd: '/p' }, snapshotEvents: () => events }
-}
-
-describe('jobs.output route (event replay)', () => {
-  it('concatenates the job_output results the model read for the job, oldest first', () => {
-    const events = [
-      jobOutputCall(0, 'c1', 'bash-1'),
-      jobOutputResult(1, 'c1', 'line1\nline2\n[status: running]'),
-      jobOutputCall(2, 'c2', 'bash-1'),
-      jobOutputResult(3, 'c2', 'line3\n[status: completed, exit code: 0]'),
-      // Another job's reads are ignored.
-      jobOutputCall(4, 'c3', 'bash-2'),
-      jobOutputResult(5, 'c3', 'other job output'),
-    ]
-    const api = buildJobsApi(ctxWith({ get: () => session(events) }, undefined), 512 * 1024)
-    expect(api.output({ sessionId: 's1', id: 'bash-1' })).toEqual({
-      text: 'line1\nline2\n[status: running]\nline3\n[status: completed, exit code: 0]',
-      truncated: false,
-      read: true,
-    })
-  })
-
-  it('skips the controller\'s "(no new output)" deltas and error results', () => {
-    const events = [
-      jobOutputCall(0, 'c1', 'bash-1'),
-      jobOutputResult(1, 'c1', 'real output\n[status: running]'),
-      jobOutputCall(2, 'c2', 'bash-1'),
-      jobOutputResult(3, 'c2', '(no new output)\n[status: running]'),
-      jobOutputCall(4, 'c3', 'bash-1'),
-      jobOutputResult(5, 'c3', 'boom', { isError: true }),
-    ]
-    const api = buildJobsApi(ctxWith({ get: () => session(events) }, undefined), 512 * 1024)
-    expect(api.output({ sessionId: 's1', id: 'bash-1' })).toEqual({
-      text: 'real output\n[status: running]',
-      truncated: false,
-      read: true,
-    })
-  })
-
-  it('reads the tool/result message dsh-llm actually produces (producer round-trip)', () => {
-    const message = createToolResultMessage({
-      callId: ToolCallId('c1'),
-      content: [{ type: 'text', text: 'produced output\n[status: running]' }],
-      isError: false,
-    })
-    const events: SidebarSessionEvent[] = [
-      jobOutputCall(0, 'c1', 'bash-1'),
-      { type: 'tool/result', seq: 1, time: 1, data: { message } },
-    ]
-    const api = buildJobsApi(ctxWith({ get: () => session(events) }, undefined), 512 * 1024)
-    expect(api.output({ sessionId: 's1', id: 'bash-1' })).toEqual({
-      text: 'produced output\n[status: running]',
-      truncated: false,
-      read: true,
-    })
-  })
-
-  it('reads a LEGACY 0.1.6-era tool-result wrapper (historical logs keep that shape)', () => {
-    // The first legacy read is an error (its wrapper flag must still be
-    // honored), the second supplies the text.
-    const events = [
-      jobOutputCall(0, 'c1', 'bash-1'),
-      legacyJobOutputResult(1, 'c1', 'legacy boom', { isError: true }),
-      jobOutputCall(2, 'c2', 'bash-1'),
-      legacyJobOutputResult(3, 'c2', 'legacy line\n[status: running]'),
-    ]
-    const api = buildJobsApi(ctxWith({ get: () => session(events) }, undefined), 512 * 1024)
-    expect(api.output({ sessionId: 's1', id: 'bash-1' })).toEqual({
-      text: 'legacy line\n[status: running]',
-      truncated: false,
-      read: true,
-    })
-  })
-
-  it('reports read:false until the model reads the job (no registry call at all)', () => {
-    const events = [jobOutputCall(0, 'c1', 'bash-2')]
-    const jobs = { kill: vi.fn() }
-    const api = buildJobsApi(ctxWith({ get: () => session(events) }, jobs), 100)
-    expect(api.output({ sessionId: 's1', id: 'bash-1' })).toEqual({ text: '', truncated: false, read: false })
-    // The replay never touches the registry — the model's cursor is safe by construction.
-    expect(jobs.kill).not.toHaveBeenCalled()
-  })
-
-  it('mirrors live job_output events the store log misses (restart divergence)', () => {
-    // The store session is frozen at its rehydration boundary (no events);
-    // the read exists only on the live session/event feed.
-    const { ctx, emit } = ctxWithFeed({ get: () => session([]) })
-    const api = buildJobsApi(ctx, 512 * 1024)
-    expect(api.output({ sessionId: 's1', id: 'bash-1' })).toEqual({ text: '', truncated: false, read: false })
-
-    // The live feed delivers the job_output call and its result.
-    emit({ id: 's1' }, jobOutputCall(100, 'c-live', 'bash-1'))
-    emit({ id: 's1' }, jobOutputResult(101, 'c-live', 'live line\n[status: completed, exit code: 0]'))
-    expect(api.output({ sessionId: 's1', id: 'bash-1' })).toEqual({
-      text: 'live line\n[status: completed, exit code: 0]',
-      truncated: false,
-      read: true,
-    })
-
-    // Unrelated results are never cached (no job_output call paired them),
-    // and another session's feed does not leak into this one.
-    emit({ id: 's1' }, jobOutputResult(102, 'c-other', 'unpaired line'))
-    emit({ id: 's2' }, jobOutputCall(103, 'c-s2', 'bash-1'))
-    emit({ id: 's2' }, jobOutputResult(104, 'c-s2', 'other session line'))
-    expect(api.output({ sessionId: 's1', id: 'bash-1' })).toEqual({
-      text: 'live line\n[status: completed, exit code: 0]',
-      truncated: false,
-      read: true,
-    })
-    expect(api.output({ sessionId: 's2', id: 'bash-1' })).toEqual({
-      text: 'other session line',
-      truncated: false,
-      read: true,
-    })
-  })
-
-  it('merges store-log traces with live mirror traces without double-counting', () => {
-    // A seed read (seq 5) in the store log plus a live read (seq 106) that
-    // ALSO reached the store log would double-count — the seq dedupe keeps
-    // exactly one copy of each.
-    const events = [jobOutputCall(5, 'c-seed', 'bash-1'), jobOutputResult(6, 'c-seed', 'seed line')]
-    const { ctx, emit } = ctxWithFeed({ get: () => session(events) })
-    const api = buildJobsApi(ctx, 512 * 1024)
-    emit({ id: 's1' }, jobOutputCall(106, 'c-live', 'bash-1'))
-    emit({ id: 's1' }, jobOutputResult(107, 'c-live', 'live line'))
-    expect(api.output({ sessionId: 's1', id: 'bash-1' })).toEqual({
-      text: 'seed line\nlive line',
-      truncated: false,
-      read: true,
-    })
-  })
-
-  it('caps oversized replays with the truncated flag', () => {
-    const events = [jobOutputCall(0, 'c1', 'bash-1'), jobOutputResult(1, 'c1', 'x'.repeat(10_000))]
-    const api = buildJobsApi(ctxWith({ get: () => session(events) }, undefined), 100)
-    const value = api.output({ sessionId: 's1', id: 'bash-1' })
-    expect(value.text).toBe('x'.repeat(100))
-    expect(value.truncated).toBe(true)
-    expect(value.read).toBe(true)
-  })
-
-  it('rejects a missing sessionId or id as bad-request', () => {
-    const api = buildJobsApi(ctxWith({ get: () => undefined }, undefined), 100)
-    expect(() => api.output({ id: 'bash-1' })).toThrowError(expect.objectContaining<Partial<SidebarError>>({ code: 'bad-request' }))
-    expect(() => api.output({ sessionId: 's1' })).toThrowError(expect.objectContaining<Partial<SidebarError>>({ code: 'bad-request' }))
-  })
-})
-
-describe('jobs.kill route', () => {
-  it('kills with the forwarded reason and the owning session id', () => {
-    const jobs = { kill: vi.fn(() => 'requested' as const) }
-    const api = buildJobsApi(ctxWith({ get: () => undefined }, jobs), 100)
-    expect(api.kill({ sessionId: 's1', id: 'bash-1', reason: 'user pressed stop' }))
-      .toEqual({ ok: true, outcome: 'requested' })
-    // 0.1.7's registry fences by SessionId; the 0.1.6 Agent object is gone.
-    expect(jobs.kill).toHaveBeenCalledWith('bash-1', 's1', 'user pressed stop')
-  })
-
-  it('defaults the reason when none is supplied', () => {
-    const jobs = { kill: vi.fn(() => 'already-finished' as const) }
-    const api = buildJobsApi(ctxWith({ get: () => undefined }, jobs), 100)
-    expect(api.kill({ sessionId: 's1', id: 'bash-1' })).toEqual({ ok: true, outcome: 'already-finished' })
-    expect(jobs.kill).toHaveBeenCalledWith('bash-1', 's1', 'user requested via sidebar')
-  })
-
-  it('maps registry refusals to a 404 job-error', () => {
-    const jobs = { kill: vi.fn(() => { throw new Error('unknown job bash-9') }) }
-    const api = buildJobsApi(ctxWith({ get: () => undefined }, jobs), 100)
-    expect(() => api.kill({ sessionId: 's1', id: 'bash-9' })).toThrowError(
-      expect.objectContaining<Partial<SidebarError>>({ code: 'job-error', status: 404 }),
-    )
-  })
-
-  it('degrades to a 503 when the jobs registry is absent (output keeps working)', () => {
-    const api = buildJobsApi(ctxWith({ get: () => session([]) }, undefined), 100)
-    expect(() => api.kill({ sessionId: 's1', id: 'bash-1' })).toThrowError(
-      expect.objectContaining<Partial<SidebarError>>({ code: 'job-error', status: 503 }),
-    )
-    expect(api.output({ sessionId: 's1', id: 'bash-1' })).toEqual({ text: '', truncated: false, read: false })
-  })
-})

tests/sidebar-auto-activation.spec.tsx

@@ -17,8 +17,8 @@
  * (`ctx.sidebarRight.mounted`): the session-list snapshot never carried a
  * current-session field, so the park gate used to read a phantom one and was
  * permanently false. Background jobs are likewise no longer mirrored into that
- * snapshot — the trigger polls the plugin's `jobs.list` route, so a job is
- * delivered by mutating the stubbed registry and advancing the poll.
+ * snapshot — they arrive as whole-set rosters PUSHED by the host's client jobs
+ * service, so a job is delivered by publishing a frame.
  */
 // @vitest-environment jsdom
 import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
@@ -37,7 +37,13 @@ import {
   type SidebarSurface,
   type TabComponentProps,
 } from '../src/client/service.ts'
-import type { Context, SidebarJobView, SidebarSessionList } from '../src/context-types.ts'
+import type {
+  Context,
+  SidebarClientJobsService,
+  SidebarJobsSnapshot,
+  SidebarJobView,
+  SidebarSessionList,
+} from '../src/context-types.ts'
 
 class FakeWebSocket {
   onmessage: ((event: { data: unknown }) => void) | null = null
@@ -162,15 +168,58 @@ interface MountedSidebar {
   surface: NativeSurfaceSpy
   column: NativeColumnSpy
   sessionId: string
-  /** The stubbed `jobs.list` registry, keyed by OWNER session (mutable). */
-  jobs: Record<string, SidebarJobView[]>
+  /** The client jobs feed the auto-open trigger watches (push rosters). */
+  jobs: JobsFeed
   unmount: () => void
 }
 
+/**
+ * The host client jobs service double: `watchRows` opens a reference-counted
+ * roster stream and `publish` pushes a whole-set frame through the same
+ * subscribe/getSnapshot pair the trigger reads.
+ */
+interface JobsFeed {
+  service: SidebarClientJobsService
+  watched: string[]
+  released: string[]
+  publish(sessionId: string, jobs: readonly SidebarJobView[]): void
+}
+
+function makeJobsFeed(): JobsFeed {
+  let snapshot: SidebarJobsSnapshot = { rows: {}, observed: {} }
+  const listeners = new Set<() => void>()
+  const feed: JobsFeed = {
+    watched: [],
+    released: [],
+    service: {
+      state: {
+        getSnapshot: () => snapshot,
+        subscribe: (listener: () => void) => {
+          listeners.add(listener)
+          return () => { listeners.delete(listener) }
+        },
+      },
+      watchRows(sessionId: string) {
+        feed.watched.push(sessionId)
+        return () => { feed.released.push(sessionId) }
+      },
+      observe: () => () => {},
+      kill: async () => {},
+    },
+    publish(sessionId, jobs) {
+      snapshot = { rows: { ...snapshot.rows, ...(jobs.length === 0 ? {} : { [sessionId]: [...jobs] }) }, observed: {} }
+      if (jobs.length === 0) {
+        const { [sessionId]: _dropped, ...rest } = snapshot.rows
+        snapshot = { rows: rest, observed: {} }
+      }
+      for (const listener of [...listeners]) listener()
+    },
+  }
+  return feed
+}
+
 let sessionSeq = 0
 const mounted: MountedSidebar[] = []
-/** Owner sessions the stubbed `jobs.list` route was read for. */
-const jobListReads: string[] = []
 
 function setViewport(width: number): void {
   Object.defineProperty(window, 'innerWidth', { configurable: true, value: width })
@@ -190,21 +239,9 @@ function mountSidebar(
     },
   }
   const feed = makeSessionFeed(initial)
-  const jobs: Record<string, SidebarJobView[]> = {}
-  // The jobs feed is polled through the plugin's own route (0.1.7 removed the
-  // snapshot's jobs mirror): the registry below is what a test mutates.
-  vi.stubGlobal('fetch', async (url: string | URL | Request, init?: RequestInit) => {
-    const method = String(url).split('/').pop()
-    if (method !== 'jobs.list') throw new Error(`unexpected fetch ${String(url)}`)
-    const body = JSON.parse(String(init?.body)) as { sessionId?: string }
-    const owner = body.sessionId ?? ''
-    jobListReads.push(owner)
-    return {
-      ok: true,
-      status: 200,
-      json: async () => ({ ok: true, value: { jobs: jobs[owner] ?? [] } }),
-    } as unknown as Response
-  })
+  // Background jobs arrive through the host's own client jobs service
+  // (`ctx.jobs`): the feed below is what a test pushes rosters into.
+  const jobs = makeJobsFeed()
   const store = createSidebarStore()
   store.setPrefs({ ...store.getPrefs(), autoOpenSubagent: true, autoOpenJobs: true })
   store.setSession(sessionId)
@@ -221,7 +258,7 @@ function mountSidebar(
     betterSidebar: service,
     get: (name: string) => name === 'betterSidebar'
       ? service
-      : name === 'sidebarRight' ? column.face : undefined,
+      : name === 'sidebarRight' ? column.face : name === 'jobs' ? jobs.service : undefined,
   } as unknown as Context
   const container = document.createElement('div')
   document.body.append(container)
@@ -280,19 +317,20 @@ function flushSubagentDebounce(): void {
 }
 
 /**
- * Deliver one new job through the polled jobs route: let the mount-time
- * baseline read land, register the job, then advance one poll interval.
+ * Deliver one new job: the roster frame the host's client service pushes. The
+ * job's start stamp is AFTER the watch began, which is what makes it "new
+ * work" rather than a job that was already running when the page mounted.
  */
 async function publishJob(sidebar: MountedSidebar): Promise<void> {
   await flushFeeds()
-  sidebar.jobs[sessionIdOf(sidebar)] = [{
+  sidebar.jobs.publish(sessionIdOf(sidebar), [{
     id: 'bash-1',
     kind: 'bash',
     label: 'sleep 30',
     status: 'running',
-    startedAt: 1_000,
-  }]
-  await act(async () => { await vi.advanceTimersByTimeAsync(2_000) })
+    startedAt: Date.now() + 1,
+  }])
+  await flushFeeds()
 }
 
 /** The conversation the sidebar is bound to (its own store's session). */
@@ -394,32 +432,39 @@ describe('Sidebar background-activity auto-activation (#162)', () => {
   it('a page that loads with jobs already running never triggers; the next NEW job does', async () => {
     const sidebar = mountSidebar(1024)
     // The conversation is already running work when the sidebar mounts: the
-    // first successful read only ARMS the baseline, it is never "new work".
-    sidebar.jobs[sidebar.sessionId] = [{
+    // first frame only ARMS the baseline (and the watch clock filters a job
+    // whose start stamp predates it), so it is never "new work".
+    sidebar.jobs.publish(sidebar.sessionId, [{
       id: 'bash-0',
       kind: 'bash',
       label: 'already running',
       status: 'running',
-      startedAt: 500,
-    }]
+      startedAt: Date.now() - 5_000,
+    }])
     await flushFeeds()
-    await act(async () => { await vi.advanceTimersByTimeAsync(10_000) })
     expect(sidebar.surface.opens).toEqual([])
     // A second, genuinely new job id is what surfaces the Tasks page.
-    sidebar.jobs[sidebar.sessionId] = [
-      ...sidebar.jobs[sidebar.sessionId] ?? [],
-      { id: 'bash-1', kind: 'bash', label: 'sleep 30', status: 'running', startedAt: 1_000 },
-    ]
-    await act(async () => { await vi.advanceTimersByTimeAsync(2_000) })
+    sidebar.jobs.publish(sidebar.sessionId, [
+      {
+        id: 'bash-0',
+        kind: 'bash',
+        label: 'already running',
+        status: 'running',
+        startedAt: Date.now() - 5_000,
+      },
+      { id: 'bash-1', kind: 'bash', label: 'sleep 30', status: 'running', startedAt: Date.now() + 1 },
+    ])
+    await flushFeeds()
     expectNativeTasksOpen(sidebar, sidebar.sessionId)
   })
 
   it('a session switch re-arms the job baseline instead of firing on the new session\'s jobs', async () => {
     const sidebar = mountSidebar(1024)
-    sidebar.jobs['child'] = [{ id: 'bash-9', kind: 'bash', label: 'child work', status: 'running', startedAt: 10 }]
+    sidebar.jobs.publish('child', [
+      { id: 'bash-9', kind: 'bash', label: 'child work', status: 'running', startedAt: Date.now() + 1 },
+    ])
     switchToChild(sidebar)
     await flushFeeds()
-    await act(async () => { await vi.advanceTimersByTimeAsync(10_000) })
     // The child's pre-existing job belongs to the NEW baseline, not to "new
     // work in the session I am looking at" (the baseline resets on session).
     expect(sidebar.surface.opens).toEqual([])

tests/subagent-jobs-view.spec.tsx

@@ -2,12 +2,11 @@
  * Tasks page tests: the projection-backed topology (branch expansion inferred
  * from the child's OWN catalog, the `mode: 'unknown'` row, navigation through
  * the workspace face, catalog retry through the 0.1.7 refresh name) and the
- * background-jobs DRAWER — rows read through the plugin's `jobs.list` route
- * one session at a time (the registry's access fence admits a job to its OWNER
- * only), clicking a row peeks its output in an anchored popover (portaled to
- * document.body) through `jobs.output` with the owner scope, the kill button
- * needs a two-click confirm, settled rows offer no kill, and nothing polls
- * while the page is hidden.
+ * background-jobs DRAWER — rows pushed by the HOST's client jobs service
+ * (`ctx.jobs`, structurally faked here), output streamed for the one OPEN row
+ * through `observe` (never the model's consuming cursor), the kill button
+ * needing a two-click confirm and going through the service, and the watch
+ * released while the page is hidden.
  */
 // @vitest-environment jsdom
 import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
@@ -17,12 +16,56 @@ import { renderRoot } from './test-utils.ts'
 import { SubagentView } from '../src/client/SubagentView.tsx'
 import type {
   Context,
-  SidebarJobView,
+  SidebarClientJobsService,
+  SidebarJobsSnapshot,
   SidebarSessionList,
   SidebarSubagentAddress,
   SidebarSubagentCatalogEntry,
 } from '../src/context-types.ts'
 
+/**
+ * A structurally faithful double of the host's client jobs service: a
+ * subscribable snapshot (roster by watched session + observations), a
+ * reference-counted `watchRows`, an `observe` that records its calls, and a
+ * `kill` that records its arguments.
+ */
+function makeJobsService() {
+  let snapshot: SidebarJobsSnapshot = { rows: {}, observed: {} }
+  const listeners = new Set<() => void>()
+  const watches: string[] = []
+  const releases: string[] = []
+  const observations: Array<{ sessionId: string | undefined; jobId: string }> = []
+  const released: string[] = []
+  const kills: Array<{ sessionId: string; jobId: string }> = []
+  const service: SidebarClientJobsService = {
+    state: {
+      getSnapshot: () => snapshot,
+      subscribe: (listener: () => void) => {
+        listeners.add(listener)
+        return () => { listeners.delete(listener) }
+      },
+    },
+    watchRows: (sessionId: string) => {
+      watches.push(sessionId)
+      return () => { releases.push(sessionId) }
+    },
+    observe: (sessionId: string | undefined, jobId: string) => {
+      observations.push({ sessionId, jobId })
+      return () => { released.push(jobId) }
+    },
+    kill: async (sessionId: string, jobId: string) => {
+      kills.push({ sessionId, jobId })
+    },
+  }
+  const emit = (next: Partial<SidebarJobsSnapshot>): void => {
+    snapshot = { ...snapshot, ...next }
+    for (const listener of [...listeners]) listener()
+  }
+  return { service, emit, watches, releases, observations, released, kills }
+}
+
+type JobsFake = ReturnType<typeof makeJobsService>
+
 /** A subscribable sessions-list snapshot (mirror of the runtime list feed). */
 function makeStore(initial: SidebarSessionList) {
   let snapshot = initial
@@ -52,21 +95,21 @@ interface NavigationSpy {
  * workspace face: 0.1.6 moved main-view selection off `ISessions`, so a spy
  * left on `sessions.openSubagent` would never fire.
  */
-function makeCtx(store: Store, navigation?: NavigationSpy): Context {
+function makeCtx(store: Store, navigation?: NavigationSpy, jobs?: JobsFake): Context {
   return {
     sessions: { list: store },
-    get: (name: string) => (name === 'uiWorkspace' && navigation !== undefined
-      ? { openSession: (target: SidebarSubagentAddress | string) => { navigation.opened.push(target) } }
-      : undefined),
+    get: (name: string) => {
+      if (name === 'jobs') return jobs?.service
+      if (name === 'uiWorkspace' && navigation !== undefined) {
+        return { openSession: (target: SidebarSubagentAddress | string) => { navigation.opened.push(target) } }
+      }
+      return undefined
+    },
   } as unknown as Context
 }
 
-const outputCalls: Array<{ sessionId: string; id: string }> = []
-const killCalls: Array<{ sessionId: string; id: string }> = []
-/** Job lists the stubbed `jobs.list` route answers with, keyed by OWNER session. */
-let jobsByOwner: Record<string, SidebarJobView[]> = {}
-/** The owner sessions the page actually read (the fence's unit of access). */
-let jobListCalls: string[] = []
+/** The jobs service the page reads (rebuilt per test in beforeEach). */
+let jobsFake: JobsFake
 
 function jsonResponse(value: unknown): Response {
   return { ok: true, status: 200, json: async () => value } as unknown as Response
@@ -100,40 +143,21 @@ function baseSnapshot(): SidebarSessionList {
 }
 
 beforeEach(() => {
-  outputCalls.length = 0
-  killCalls.length = 0
-  jobListCalls = []
-  jobsByOwner = {
-    root: [{ id: 'bash-1', kind: 'bash', label: 'sleep 300', status: 'running', startedAt: 1_000 }],
-    child: [{ id: 'bash-2', kind: 'bash', label: 'echo hi', status: 'completed', startedAt: 2_000, finishedAt: 3_000 }],
-  }
-  vi.stubGlobal('fetch', async (url: string | URL | Request, init?: RequestInit) => {
+  jobsFake = makeJobsService()
+  jobsFake.emit({
+    rows: {
+      root: [{ id: 'bash-1', kind: 'bash', label: 'sleep 300', status: 'running', startedAt: 1_000 }],
+      child: [{
+        id: 'bash-2', kind: 'bash', label: 'echo hi', status: 'completed',
+        startedAt: 2_000, finishedAt: 3_000, owner: 'child',
+      }],
+    },
+  })
+  vi.stubGlobal('fetch', async (url: string | URL | Request) => {
     const method = String(url).split('/').pop()
-    const body = JSON.parse(String(init?.body)) as { sessionId?: string; id?: string; rootSessionId?: string }
     if (method === 'subagents.live') {
       return jsonResponse({ ok: true, value: { live: {} } })
     }
-    if (method === 'jobs.list') {
-      const owner = body.sessionId ?? ''
-      jobListCalls.push(owner)
-      return jsonResponse({ ok: true, value: { jobs: jobsByOwner[owner] ?? [] } })
-    }
-    if (method === 'jobs.output') {
-      outputCalls.push({ sessionId: body.sessionId ?? '', id: body.id ?? '' })
-      // bash-9 stands for a job the model never read (read:false).
-      return jsonResponse({
-        ok: true,
-        value: {
-          text: body.id === 'bash-9' ? '' : `output-of-${body.id ?? ''}`,
-          truncated: false,
-          read: body.id !== 'bash-9',
-        },
-      })
-    }
-    if (method === 'jobs.kill') {
-      killCalls.push({ sessionId: body.sessionId ?? '', id: body.id ?? '' })
-      return jsonResponse({ ok: true, value: { ok: true, outcome: 'requested' } })
-    }
     if (method === 'workflows.list') {
       return jsonResponse({ ok: true, value: { runs: [] } })
     }
@@ -151,20 +175,22 @@ afterEach(() => {
   for (const el of document.querySelectorAll('body > div')) el.remove()
 })
 
-/**
- * Flush the mount-time job reads: the poll fires immediately, and the request
- * → envelope → json → Promise.all chain settles over several microtask ticks.
- */
+/** Flush the mount-time requests (routes + the service snapshot). */
 async function flushJobs(): Promise<void> {
   for (let tick = 0; tick < 4; tick++) {
     await act(async () => { await Promise.resolve() })
   }
 }
 
-/** Render the page and wait for its first job read to land. */
-async function renderPage(store: Store, active = true, navigation?: NavigationSpy) {
+/** Render the page with the jobs service wired in. */
+async function renderPage(
+  store: Store,
+  active = true,
+  navigation?: NavigationSpy,
+  jobs: JobsFake = jobsFake,
+) {
   const rendered = renderRoot(
-    createElement(SubagentView, { sessionId: 'root', active, ctx: makeCtx(store, navigation) }),
+    createElement(SubagentView, { sessionId: 'root', active, ctx: makeCtx(store, navigation, jobs) }),
   )
   await flushJobs()
   return rendered
@@ -276,99 +302,155 @@ describe('SubagentView background jobs', () => {
     expect(text).toContain('echo hi')
     expect(text).toContain('主会话')
     expect(text).toContain('子代理')
-    // The fence admits a job to its owner only, so the tree costs one read per
-    // tree session (and never a read of a session outside the tree).
-    expect([...jobListCalls].sort()).toEqual(['child', 'root'])
+    // The host shares one roster stream per session: the page watches every
+    // session of the tree while it is on screen.
+    expect([...jobsFake.watches].sort()).toEqual(['child', 'root'])
     // Only the running row offers a kill button.
     expect(container.querySelectorAll('button[aria-label="终止"]')).toHaveLength(1)
     unmount()
   })
 
-  it('renders nothing job-related when the tree has no jobs', async () => {
-    jobsByOwner = {}
-    const store = makeStore({ byId: { root: { id: 'root', displayTitle: '主会话' } } })
-    const { container, unmount } = await renderPage(store)
+  it('renders nothing job-related on a deployment without the jobs service', async () => {
+    const store = makeStore(baseSnapshot())
+    const bare = makeCtx(store)
+    const { container, unmount } = renderRoot(
+      createElement(SubagentView, { sessionId: 'root', active: true, ctx: bare }),
+    )
+    await flushJobs()
     expect(container.textContent).not.toContain('后台任务')
     unmount()
   })
 
-  it('survives the job list emptying while mounted (hook-order regression #300)', async () => {
-    vi.useFakeTimers()
-    try {
-      const store = makeStore(baseSnapshot())
-      const { container, unmount } = await renderPage(store)
-      expect(container.textContent).toContain('sleep 300')
-      // Every job settles and drops: the section must vanish WITHOUT reordering
-      // hooks — a hook below the empty-state return would crash React with
-      // "Rendered fewer hooks than expected" (the minified #300 the sidebar
-      // boundary surfaces with a retry button).
-      jobsByOwner = {}
-      await act(async () => { await vi.advanceTimersByTimeAsync(5_000) })
-      expect(container.textContent).not.toContain('后台任务')
-      // And returning jobs must work too, with the same hook order.
-      jobsByOwner = {
-        root: [{ id: 'bash-1', kind: 'bash', label: 'sleep 300', status: 'running', startedAt: 1_000 }],
-      }
-      await act(async () => { await vi.advanceTimersByTimeAsync(5_000) })
-      expect(container.textContent).toContain('sleep 300')
-      unmount()
-    } finally {
-      vi.useRealTimers()
-    }
+  it('releases every watch when the page goes away', async () => {
+    const store = makeStore(baseSnapshot())
+    const { unmount } = await renderPage(store)
+    expect(jobsFake.releases).toEqual([])
+    unmount()
+    expect([...jobsFake.releases].sort()).toEqual(['child', 'root'])
   })
 
-  it('shows the selected job output in an anchored popover, dismissable', async () => {
+  it('watches nothing while the page is hidden', async () => {
+    const store = makeStore(baseSnapshot())
+    // A mounted-but-hidden page renders from the last snapshot it holds; what
+    // it must NOT do is hold a roster stream open.
+    const { unmount } = await renderPage(store, false)
+    expect(jobsFake.watches).toEqual([])
+    unmount()
+  })
+
+  it('survives the roster emptying while mounted (hook-order regression #300)', async () => {
     const store = makeStore(baseSnapshot())
     const { container, unmount } = await renderPage(store)
+    expect(container.textContent).toContain('sleep 300')
+    // Every job settles and drops: the drawer must vanish WITHOUT reordering
+    // hooks — a hook below the empty-state return would crash React with
+    // "Rendered fewer hooks than expected" (the minified #300 the sidebar
+    // boundary surfaces with a retry button).
+    await act(async () => { jobsFake.emit({ rows: {} }) })
+    expect(container.textContent).not.toContain('后台任务')
+    // And returning jobs must work too, with the same hook order.
+    await act(async () => {
+      jobsFake.emit({
+        rows: { root: [{ id: 'bash-1', kind: 'bash', label: 'sleep 300', status: 'running', startedAt: 1_000 }] },
+      })
+    })
+    expect(container.textContent).toContain('sleep 300')
+    unmount()
+  })
+
+  it('observes only the OPEN row and streams its output into the popover', async () => {
+    const store = makeStore(baseSnapshot())
+    const { container, unmount } = await renderPage(store)
+    // Nothing is observed until a row opens: the host streams per observer.
+    expect(jobsFake.observations).toEqual([])
     const row = container.querySelector('button[aria-label*="sleep 300"]') as HTMLButtonElement
     await act(async () => { row.click() })
-    // The peek request carries the OWNER session (the fence compares it).
-    expect(outputCalls).toEqual([{ sessionId: 'root', id: 'bash-1' }])
-    // The popover is portaled to document.body (outside the tab container).
-    expect(document.body.textContent).toContain('output-of-bash-1')
+    expect(jobsFake.observations).toEqual([{ sessionId: 'root', jobId: 'bash-1' }])
+    // The observed frames land in the portalled popover.
+    await act(async () => {
+      jobsFake.emit({
+        observed: { 'bash-1': { text: 'building…\ndone', gapBefore: false, streaming: true } },
+      })
+    })
+    expect(document.body.textContent).toContain('building…')
     expect(document.querySelectorAll('[role="dialog"]')).toHaveLength(1)
-    // Escape dismisses the popover.
+
+    // Switching rows releases the first observation and opens the second (its
+    // owner session rides along — the kill and the stream both need it).
+    const second = container.querySelector('button[aria-label*="echo hi"]') as HTMLButtonElement
+    await act(async () => { second.click() })
+    expect(jobsFake.released).toEqual(['bash-1'])
+    expect(jobsFake.observations).toEqual([
+      { sessionId: 'root', jobId: 'bash-1' },
+      { sessionId: 'child', jobId: 'bash-2' },
+    ])
+    expect(document.body.textContent).not.toContain('building…')
+
+    // Escape dismisses the popover AND releases the observation.
     await act(async () => {
       document.dispatchEvent(new KeyboardEvent('keydown', { key: 'Escape' }))
     })
-    expect(document.body.textContent).not.toContain('output-of-bash-1')
     expect(document.querySelectorAll('[role="dialog"]')).toHaveLength(0)
+    expect(jobsFake.released).toContain('bash-2')
     unmount()
   })
 
-  it('switches the single output popover between rows', async () => {
+  it('flags a gap and a producer error in the observed stream', async () => {
     const store = makeStore(baseSnapshot())
     const { container, unmount } = await renderPage(store)
-    const first = container.querySelector('button[aria-label*="sleep 300"]') as HTMLButtonElement
-    await act(async () => { first.click() })
-    expect(document.body.textContent).toContain('output-of-bash-1')
-    const second = container.querySelector('button[aria-label*="echo hi"]') as HTMLButtonElement
-    await act(async () => { second.click() })
-    // One popover, now fed by the second job (its owner session scopes the replay).
-    expect(outputCalls).toEqual([
-      { sessionId: 'root', id: 'bash-1' },
-      { sessionId: 'child', id: 'bash-2' },
-    ])
-    expect(document.querySelectorAll('[role="dialog"]')).toHaveLength(1)
-    expect(document.body.textContent).not.toContain('output-of-bash-1')
-    expect(document.body.textContent).toContain('output-of-bash-2')
+    await act(async () => {
+      (container.querySelector('button[aria-label*="sleep 300"]') as HTMLButtonElement).click()
+    })
+    await act(async () => {
+      jobsFake.emit({
+        observed: {
+          'bash-1': { text: 'tail only', gapBefore: true, streaming: false, error: 'writer died' },
+        },
+      })
+    })
+    const body = document.body.textContent ?? ''
+    // The dropped-frame notice and the producer error both surface (the
+    // plugin's own copy for each).
+    expect(body).toContain('输出过长，已截断显示')
+    expect(body).toContain('输出读取失败')
+    expect(body).toContain('tail only')
     unmount()
   })
 
-  it('explains when the model has not read the job yet', async () => {
-    jobsByOwner = {
-      root: [
-        ...jobsByOwner.root ?? [],
-        { id: 'bash-9', kind: 'bash', label: 'unread cmd', status: 'running', startedAt: 9_000 },
-      ],
-    }
+  it('kills a live job through the service only after the two-click confirm', async () => {
     const store = makeStore(baseSnapshot())
     const { container, unmount } = await renderPage(store)
-    const row = container.querySelector('button[aria-label*="unread cmd"]') as HTMLButtonElement
-    await act(async () => { row.click() })
-    // read:false → the popover explains the output awaits the model's job_output
-    // (never the model's cursor, so there is nothing to steal yet).
-    expect(document.body.textContent).toContain('等待模型读取该任务的输出')
+    const kill = container.querySelector('button[aria-label="终止"]') as HTMLButtonElement
+    await act(async () => { kill.click() })
+    // First click only arms the confirm — nothing reaches the registry.
+    expect(jobsFake.kills).toEqual([])
+    const confirm = container.querySelector('button[aria-label="再次点击确认终止"]')
+    expect(confirm).not.toBeNull()
+    await act(async () => { (confirm as HTMLButtonElement).click() })
+    expect(jobsFake.kills).toEqual([{ sessionId: 'root', jobId: 'bash-1' }])
+    unmount()
+  })
+
+  it('drops rows owned by a session outside the tree', async () => {
+    const store = makeStore(baseSnapshot())
+    // An unowned job is visible to every watcher, so it stays; a job owned by
+    // a session this tree does not contain must not appear.
+    await act(async () => {
+      jobsFake.emit({
+        rows: {
+          root: [
+            { id: 'bash-1', kind: 'bash', label: 'sleep 300', status: 'running', startedAt: 1_000 },
+            { id: 'bash-7', kind: 'bash', label: 'foreign cmd', status: 'running', startedAt: 1_100, owner: 'elsewhere' },
+            { id: 'bash-8', kind: 'bash', label: 'unowned cmd', status: 'running', startedAt: 1_200 },
+          ],
+        },
+      })
+    })
+    const { container, unmount } = await renderPage(store)
+    const text = container.textContent ?? ''
+    expect(text).toContain('sleep 300')
+    expect(text).toContain('unowned cmd')
+    expect(text).not.toContain('foreign cmd')
     unmount()
   })
 
@@ -381,47 +463,17 @@ describe('SubagentView background jobs', () => {
       startedAt: 1_000 + index,
       ...(index % 2 === 0 ? {} : { finishedAt: 2_000 + index, detail: 'exit code: 1' }),
     }))
-    jobsByOwner = { root: many }
     const store = makeStore({ byId: { root: { id: 'root', displayTitle: '主会话' } } })
+    await act(async () => { jobsFake.emit({ rows: { root: many } }) })
     const { container, unmount } = await renderPage(store)
-    expect(container.textContent).toContain('60 个后台任务 · 30 运行中')
+    // One agent: the drawer is open by default (the auto-collapse rule counts
+    // AGENTS, not jobs — see the 8-child case in tasks-page.spec.tsx).
+    expect(container.textContent).toContain('后台任务')
     expect(container.querySelectorAll('button[aria-label*="bulk cmd"]')).toHaveLength(60)
     // Clicking a row anywhere in the long list still feeds the single popover.
     const row = container.querySelector('button[aria-label*="bulk cmd 59"]') as HTMLButtonElement
     await act(async () => { row.click() })
-    expect(outputCalls).toEqual([{ sessionId: 'root', id: 'bash-69' }])
-    expect(document.body.textContent).toContain('output-of-bash-69')
+    expect(jobsFake.observations).toEqual([{ sessionId: 'root', jobId: 'bash-69' }])
     unmount()
   })
-
-  it('kills a live job only after the two-click confirm', async () => {
-    const store = makeStore(baseSnapshot())
-    const { container, unmount } = await renderPage(store)
-    const kill = container.querySelector('button[aria-label="终止"]') as HTMLButtonElement
-    await act(async () => { kill.click() })
-    // First click only arms the confirm — no request leaves the page.
-    expect(killCalls).toEqual([])
-    const confirm = container.querySelector('button[aria-label="再次点击确认终止"]')
-    expect(confirm).not.toBeNull()
-    await act(async () => { (confirm as HTMLButtonElement).click() })
-    expect(killCalls).toEqual([{ sessionId: 'root', id: 'bash-1' }])
-    unmount()
-  })
-
-  it('does not read jobs or poll output while the page is hidden', async () => {
-    vi.useFakeTimers()
-    try {
-      const store = makeStore(baseSnapshot())
-      const { container, unmount } = await renderPage(store, false)
-      // A hidden page issues no job read at all (it also has no rows to show).
-      expect(jobListCalls).toEqual([])
-      expect(container.textContent).not.toContain('后台任务')
-      await act(async () => { await vi.advanceTimersByTimeAsync(10_000) })
-      expect(jobListCalls).toEqual([])
-      expect(outputCalls).toEqual([])
-      unmount()
-    } finally {
-      vi.useRealTimers()
-    }
-  })
 })

tests/subagent-jobs.spec.ts

@@ -1,10 +1,10 @@
 /**
- * Pure-helper tests for the Subagent page's background-job section:
- * tree-membership collection, ordering, and status presentation mapping.
+ * Pure-helper tests for the Tasks page's background-job presentation:
+ * tree-membership, ordering, status mapping, and the auto-open trigger's
+ * "is this really new work?" rule.
  */
 import { describe, expect, it } from 'vitest'
 import {
-  collectTreeJobs,
   detectNewJob,
   formatJobDuration,
   isJobLive,
@@ -61,31 +61,6 @@ describe('treeSessionIds', () => {
   })
 })
 
-describe('collectTreeJobs', () => {
-  it('collects jobs of the whole tree with owner titles, ignoring outside sessions', () => {
-    const byId = {
-      root: summary('root'),
-      child: summary('child', { origin: 'subagent', parentId: 'root' }),
-    }
-    // The map is the client's per-owner read of `jobs.list` (the registry's
-    // fence admits a job to its owner session only), keyed by that owner.
-    const jobsByOwner = {
-      root: [job('bash-1')],
-      child: [job('bash-2', { status: 'completed', finishedAt: 2_000 })],
-      stranger: [job('bash-9')],
-    }
-    const rows = collectTreeJobs(byId, jobsByOwner, 'root')
-    expect(rows.map(row => [row.ownerSessionId, row.ownerTitle, row.job.id]))
-      .toEqual([['root', 'title-root', 'bash-1'], ['child', 'title-child', 'bash-2']])
-  })
-
-  it('returns an empty list for an absent read or empty sets', () => {
-    const byId = { root: summary('root') }
-    expect(collectTreeJobs(byId, undefined, 'root')).toEqual([])
-    expect(collectTreeJobs(byId, {}, 'root')).toEqual([])
-  })
-})
-
 describe('orderJobs', () => {
   it('puts live rows first in start order, then settled rows newest-first', () => {
     const row = (id: string, status: SidebarJobStatus, startedAt: number, finishedAt?: number) => ({
@@ -138,10 +113,11 @@ describe('status presentation helpers', () => {
 describe('detectNewJob', () => {
   /**
    * The baseline rule the auto-open trigger adds on top of this helper: the
-   * FIRST list a page reads only arms the baseline (the poller keeps
-   * `prev === undefined` until a read succeeds), so a conversation that is
-   * already running jobs when the page loads never pops the Tasks page. The
-   * helper itself is pure over two lists.
+   * FIRST frame a page observes only arms the baseline, so a conversation that
+   * is already running jobs when the page loads never pops the Tasks page. The
+   * helper itself is pure over two frames — plus the WATCH CLOCK (`since`),
+   * which is what keeps that promise now that the roster is a push stream
+   * whose empty frames are indistinguishable from "not delivered yet".
    */
   it('fires on EVERY job id the previous list lacked, and on nothing else', () => {
     // A new id appears (the first job, then another one alongside it).
@@ -156,8 +132,22 @@ describe('detectNewJob', () => {
     // are not new work.
     expect(detectNewJob([job('bash-1')], [job('bash-1')])).toBe(false)
     expect(detectNewJob([job('bash-1'), job('bash-2')], [job('bash-2')])).toBe(false)
-    // The first list of a fresh page is compared against itself by the caller
-    // (prev stays undefined until a read succeeds) — the helper's quiet case.
+    // The first frame of a fresh page is compared against itself by the caller
+    // (prev stays undefined until a frame lands) — the helper's quiet case.
     expect(detectNewJob([], [])).toBe(false)
   })
+
+  it('ignores work that started before the watcher opened', () => {
+    const watchStart = 10_000
+    // The page mounts while a job already runs: its id is new to the id-set
+    // diff, but its start time predates the watch.
+    expect(detectNewJob([], [job('bash-1', { startedAt: 500 })], watchStart)).toBe(false)
+    // Work the agent starts after the watcher opened still triggers.
+    expect(detectNewJob([], [job('bash-1', { startedAt: 10_000 })], watchStart)).toBe(true)
+    expect(detectNewJob([], [job('bash-1', { startedAt: 12_000 })], watchStart)).toBe(true)
+    // An already-seen id never triggers, however new its stamp.
+    expect(detectNewJob([job('bash-1')], [job('bash-1', { startedAt: 20_000 })], watchStart)).toBe(false)
+    // Omitting the clock keeps the old id-set behaviour.
+    expect(detectNewJob([], [job('bash-1', { startedAt: 0 })])).toBe(true)
+  })
 })

tests/subagent-live-polling.spec.tsx

@@ -46,6 +46,9 @@ function makeCtx(store: Store, historySpy: ReturnType<typeof vi.fn>): Context {
         subagents: { history: historySpy },
       },
     },
+    // No host jobs service in this harness: the page renders no jobs surface
+    // (and therefore issues no jobs request).
+    get: () => undefined,
   } as unknown as Context
 }
 
@@ -135,7 +138,6 @@ describe('SubagentView live polling', () => {
           },
         })
       }
-      if (method === 'jobs.list') return jsonResponse({ ok: true, value: { jobs: [] } })
       throw new Error(`unexpected fetch ${String(url)}`)
     })
 
@@ -171,7 +173,6 @@ describe('SubagentView live polling', () => {
         liveCalls.push(body.rootSessionId ?? '')
         return new Promise<Response>((resolve) => { resolveFirst = resolve })
       }
-      if (method === 'jobs.list') return jsonResponse({ ok: true, value: { jobs: [] } })
       throw new Error(`unexpected fetch ${String(url)}`)
     })
 
@@ -210,7 +211,6 @@ describe('SubagentView live polling', () => {
         liveCalls.push(body.rootSessionId ?? '')
         return new Promise<Response>((resolve) => { resolveFirst = resolve })
       }
-      if (method === 'jobs.list') return jsonResponse({ ok: true, value: { jobs: [] } })
       throw new Error(`unexpected fetch ${String(url)}`)
     })
 
@@ -254,7 +254,6 @@ describe('SubagentView live polling', () => {
           : {}
         return jsonResponse({ ok: true, value: { live } })
       }
-      if (method === 'jobs.list') return jsonResponse({ ok: true, value: { jobs: [] } })
       throw new Error(`unexpected fetch ${String(url)}`)
     })
 

tests/tasks-page.spec.tsx

@@ -12,7 +12,8 @@ import { renderRoot } from './test-utils.ts'
 import { SubagentView } from '../src/client/SubagentView.tsx'
 import type {
   Context,
-  SidebarJobView,
+  SidebarClientJobsService,
+  SidebarJobsSnapshot,
   SidebarProjectionSnapshot,
   SidebarSessionList,
   SidebarSessionSummary,
@@ -44,6 +45,41 @@ interface NavigationSpy {
   opened: Array<SidebarSubagentAddress | string>
 }
 
+/**
+ * A structurally faithful double of the host's client jobs service: a
+ * subscribable snapshot, a reference-counted `watchRows`, an `observe` that
+ * records its calls, and a `kill` that records its arguments.
+ */
+function makeJobsService() {
+  let snapshot: SidebarJobsSnapshot = { rows: {}, observed: {} }
+  const listeners = new Set<() => void>()
+  const observations: Array<{ sessionId: string | undefined; jobId: string }> = []
+  const kills: Array<{ sessionId: string; jobId: string }> = []
+  const service: SidebarClientJobsService = {
+    state: {
+      getSnapshot: () => snapshot,
+      subscribe: (listener: () => void) => {
+        listeners.add(listener)
+        return () => { listeners.delete(listener) }
+      },
+    },
+    watchRows: () => () => {},
+    observe: (sessionId: string | undefined, jobId: string) => {
+      observations.push({ sessionId, jobId })
+      return () => {}
+    },
+    kill: async (sessionId: string, jobId: string) => { kills.push({ sessionId, jobId }) },
+  }
+  const emit = (next: Partial<SidebarJobsSnapshot>): void => {
+    snapshot = { ...snapshot, ...next }
+    for (const listener of [...listeners]) listener()
+  }
+  return { service, emit, observations, kills }
+}
+
+/** The jobs service the page reads (rebuilt per test in beforeEach). */
+let jobsFake: ReturnType<typeof makeJobsService>
+
 /** The client context face SubagentView touches. */
 function makeCtx(store: Store, navigation?: NavigationSpy): Context {
   return {
@@ -51,9 +87,13 @@ function makeCtx(store: Store, navigation?: NavigationSpy): Context {
       list: store,
       refreshProjections: async () => {},
     },
-    get: (name: string) => (name === 'uiWorkspace' && navigation !== undefined
-      ? { openSession: (target: SidebarSubagentAddress | string) => { navigation.opened.push(target) } }
-      : undefined),
+    get: (name: string) => {
+      if (name === 'jobs') return jobsFake.service
+      if (name === 'uiWorkspace' && navigation !== undefined) {
+        return { openSession: (target: SidebarSubagentAddress | string) => { navigation.opened.push(target) } }
+      }
+      return undefined
+    },
   } as unknown as Context
 }
 
@@ -64,8 +104,6 @@ function jsonResponse(value: unknown): Response {
 let teamPayload: unknown = { available: false }
 const teamMutations: Array<{ method: string; body: Record<string, unknown> }> = []
 const fetchedMethods: string[] = []
-/** Job lists the stubbed `jobs.list` route answers with, keyed by OWNER session. */
-let jobsByOwner: Record<string, SidebarJobView[]> = {}
 /** The `subagents.live` payload the stub answers with (session id → live view). */
 let livePayload: Record<string, unknown> = {}
 /** The folded workflow runs the stub answers `workflows.list` with. */
@@ -77,9 +115,12 @@ beforeEach(() => {
   runsPayload = []
   teamMutations.length = 0
   fetchedMethods.length = 0
-  jobsByOwner = {
-    root: [{ id: 'bash-1', kind: 'bash', label: 'sleep 300', status: 'running', startedAt: 1_000 }],
-  }
+  jobsFake = makeJobsService()
+  jobsFake.emit({
+    rows: {
+      root: [{ id: 'bash-1', kind: 'bash', label: 'sleep 300', status: 'running', startedAt: 1_000 }],
+    },
+  })
   vi.stubGlobal('fetch', async (url: string | URL | Request, init?: RequestInit) => {
     const method = String(url).split('/').pop()
     fetchedMethods.push(method ?? '')
@@ -87,9 +128,6 @@ beforeEach(() => {
     if (method === 'subagents.live') return jsonResponse({ ok: true, value: { live: livePayload } })
     if (method === 'workflows.list') return jsonResponse({ ok: true, value: { runs: runsPayload } })
     if (method === 'teams.view') return jsonResponse({ ok: true, value: teamPayload })
-    if (method === 'jobs.list') {
-      return jsonResponse({ ok: true, value: { jobs: jobsByOwner[String(body.sessionId ?? '')] ?? [] } })
-    }
     if (method === 'teams.taskCreate' || method === 'teams.taskUpdate') {
       teamMutations.push({ method: method ?? '', body })
       return jsonResponse({