Merge branch 'worktree/process-plugin-streaming-72f9b3' into worktree/job-kill-delivery

# Conflicts:
#	apps/web/tests/background-job-list.e2e.ts
#	apps/web/tests/live-job-stream.e2e.ts
#	docs/config-catalog.i18n.yaml
#	docs/config-catalog.md
#	docs/config-catalog.zh.md
#	docs/event-producer-consumer.i18n.yaml
#	docs/event-producer-consumer.md
#	packages/client/ui-activity/README.md
#	packages/client/ui-activity/README.zh.md
#	packages/client/ui-activity/src/client/index.ts
#	packages/client/ui-jobs/README.i18n.yaml
#	packages/client/ui-jobs/src/client/JobListAction.tsx
#	packages/client/ui-jobs/tests/browser-plugin.client.spec.ts
#	packages/client/ui-jobs/tests/job-list-action.client.spec.tsx
#	packages/jobs/jobs-local/src/index.ts
#	packages/jobs/jobs/README.i18n.yaml
#	packages/jobs/jobs/README.md
#	packages/jobs/jobs/README.zh.md
#	packages/jobs/jobs/src/index.ts
#	packages/shell/tool-bash/src/index.ts
#	packages/shell/tool-bash/tests/observe.spec.ts
#	packages/shell/tool-pwsh/src/index.ts
#	packages/shell/tool-pwsh/tests/observe.spec.ts
#	packages/workflow/tool-workflow/tests/activity-mirror.spec.ts
#	pnpm-lock.yaml

packages/api/session-controller/src/client/index.ts

@@ -3,6 +3,7 @@
 import type { Context } from '@deepseek-ai/cordis'
 import type {} from '@deepseek-ai/dsh-agent/types'
 import { createSessionControlStream } from './transport.ts'
+import { ClientJobOutput, ClientJobOutputModel } from './job-output.ts'
 import { ClientSessions } from './sessions/service.ts'
 import type { SessionRemotes } from './sessions/remotes.ts'
 import type {} from '../remote-events.ts'
@@ -23,6 +24,8 @@ export type {
 } from './transport.ts'
 export { createScope, scopeOf } from './scope.ts'
 export type { AgentContext, AgentScopeHandle } from './scope.ts'
+export { ClientJobOutput, ClientJobOutputModel } from './job-output.ts'
+export type { IJobOutput, JobOutputSnapshot, JobOutputSource, ObservedJob } from './job-output.ts'
 export { SessionCreateError, SessionForkError } from './sessions/service.ts'
 export type { SessionBinding, SessionListState, SessionSummary } from './sessions/service.ts'
 export type {
@@ -89,6 +92,14 @@ export const inject = [
 export function apply(ctx: Context): void {
   const remotes = ctx.remote as unknown as SessionRemotes
   const sessions = new ClientSessions(ctx, remotes)
+  // Resolve the session namespace to a concrete value while this plugin's
+  // context is current: observation (re)opens run on caller stacks (a React
+  // event, a carrier retry) whose dynamic context has not declared
+  // `remote.session`, and a property access there fails the inject check.
+  new ClientJobOutput(ctx, {
+    $stream: options => remotes.$stream(options),
+    session: remotes.session,
+  }, new ClientJobOutputModel())
   ctx.remote.$on('api-session/added', (summary) => { sessions.handleSessionAdded(summary) })
   ctx.remote.$on('api-session/removed', (sessionId) => { sessions.handleSessionRemoved(sessionId) })
   ctx.remote.$on('api-session/status', (sessionId, running) => {

packages/api/session-controller/src/client/job-output.ts

@@ -0,0 +1,348 @@
+/**
+ * React-free client job-record observation: per-job accumulated output views
+ * over the `session.observeJob` stream, reference-counted per job so
+ * overlapping panels share one stream. The roster itself rides the session
+ * control stream (`jobsBySession`); this module carries only the output.
+ * @module @deepseek-ai/dsh-api-session-controller/client/job-output
+ */
+
+import { Service, type Context } from '@deepseek-ai/cordis'
+import { RemoteStreamCarrierError, type ClientRemote } from '@deepseek-ai/dsh-api-gateway/client'
+import { notifySubscribers } from '@deepseek-ai/dsh-client-store'
+import type { JobId } from '@deepseek-ai/dsh-jobs/brand'
+import type { SessionId } from '@deepseek-ai/dsh-session/types'
+import type { SessionJob, SessionObserveJobFrame } from '../types.ts'
+import type { SessionRemote } from './transport.ts'
+
+/** Bounded per-job render tail, in UTF-16 code units. */
+const RENDER_TAIL_LIMIT = 128 * 1024
+
+/** One observed job's live view state. */
+export interface ObservedJob {
+  readonly jobId: JobId
+  /** Accumulated output tail, bounded to the render limit. */
+  readonly text: string
+  /** True when bytes before {@link text} were dropped (eviction, resume gap, or the render bound). */
+  readonly gapBefore: boolean
+  /** Latest status delivered on the observation stream. */
+  readonly status: SessionJob['status']
+  readonly detail?: string
+  /** True while the observation stream is open and the job has not settled. */
+  readonly streaming: boolean
+  /** Terminal observation failure, when the stream ended abnormally. */
+  readonly error?: string
+}
+
+/** Immutable client job-output state. */
+export interface JobOutputSnapshot {
+  /** Live observation state keyed by job id. */
+  readonly observed: Readonly<Record<string, ObservedJob>>
+}
+
+/** Mutable observation bookkeeping behind one {@link ObservedJob} view. */
+interface ObservedState {
+  view: ObservedJob
+  /** Resume offset for the next generation's `from`. */
+  cursor: number | undefined
+}
+
+/**
+ * Owns per-job observation state. Pure data plus subscriptions — transport
+ * wiring stays in the client service, UI stays in slot components.
+ */
+export class ClientJobOutputModel {
+  private readonly observedStates = new Map<string, ObservedState>()
+  private readonly listeners = new Set<() => void>()
+  private snapshotCache: JobOutputSnapshot = { observed: {} }
+  private snapshotDirty = false
+
+  /**
+   * Read the identity-stable current snapshot.
+   * @returns the cached snapshot, rebuilt only after a change.
+   */
+  getSnapshot(): JobOutputSnapshot {
+    if (this.snapshotDirty) {
+      const observed: Record<string, ObservedJob> = {}
+      for (const [id, state] of this.observedStates) observed[id] = state.view
+      this.snapshotCache = { observed }
+      this.snapshotDirty = false
+    }
+    return this.snapshotCache
+  }
+
+  /**
+   * Subscribe to snapshot changes.
+   * @param listener - invalidation callback.
+   * @returns unsubscribe function.
+   */
+  subscribe(listener: () => void): () => void {
+    this.listeners.add(listener)
+    return () => { this.listeners.delete(listener) }
+  }
+
+  /**
+   * The resume offset for one job's next observation generation.
+   * @param id - observed job.
+   * @returns the last accepted `next`, or undefined for a fresh observation.
+   */
+  cursorOf(id: JobId): number | undefined {
+    return this.observedStates.get(String(id))?.cursor
+  }
+
+  /**
+   * Install or reset observation state when a generation's anchor arrives.
+   * @param id - observed job.
+   * @param frame - the generation's `opened` anchor.
+   */
+  observeOpened(id: JobId, frame: Extract<SessionObserveJobFrame, { type: 'opened' }>): void {
+    const existing = this.observedStates.get(String(id))
+    // A fresh view anchored past offset zero starts after an evicted head
+    // (fresh observations anchor at the registry's earliest retained byte), so
+    // it owes the same gap mark a live observer earned from lossy reads. A
+    // resume that already accumulated text keeps its recorded gap state.
+    const freshPastHead = (existing === undefined || existing.view.text === '') && frame.from > 0
+    const view: ObservedJob = {
+      jobId: id,
+      text: existing?.view.text ?? '',
+      gapBefore: (existing?.view.gapBefore ?? false) || frame.from < frame.earliest || freshPastHead,
+      status: frame.status,
+      ...frame.detail !== undefined ? { detail: frame.detail } : {},
+      streaming: true,
+    }
+    this.observedStates.set(String(id), { view, cursor: frame.from })
+    this.changed()
+  }
+
+  /**
+   * Append one output frame's chunks to the bounded render tail.
+   * @param id - observed job.
+   * @param frame - a coalesced `output` frame.
+   */
+  observeOutput(id: JobId, frame: Extract<SessionObserveJobFrame, { type: 'output' }>): void {
+    const state = this.observedStates.get(String(id))
+    /* v8 ignore next -- frames arrive only between opened and stop for a tracked id. */
+    if (state === undefined) return
+    let text = state.view.text + frame.chunks.map(chunk => chunk.text).join('')
+    let gapBefore = state.view.gapBefore || frame.lossy === true
+      || frame.chunks.some(chunk => chunk.gapBefore === true)
+    if (text.length > RENDER_TAIL_LIMIT) {
+      let cut = text.length - RENDER_TAIL_LIMIT
+      // Never split a surrogate pair at the render bound.
+      const unit = text.charCodeAt(cut)
+      if (unit >= 0xDC00 && unit <= 0xDFFF) cut += 1
+      text = text.slice(cut)
+      gapBefore = true
+    }
+    state.view = { ...state.view, text, gapBefore }
+    state.cursor = frame.next
+    this.changed()
+  }
+
+  /**
+   * Record the terminal status delivered at the end of the observation stream.
+   * @param id - observed job.
+   * @param frame - the terminal `status` frame.
+   */
+  observeStatus(id: JobId, frame: Extract<SessionObserveJobFrame, { type: 'status' }>): void {
+    const state = this.observedStates.get(String(id))
+    /* v8 ignore next -- frames arrive only between opened and stop for a tracked id. */
+    if (state === undefined) return
+    state.view = {
+      ...state.view,
+      status: frame.status,
+      ...frame.detail !== undefined ? { detail: frame.detail } : {},
+      streaming: false,
+    }
+    this.changed()
+  }
+
+  /**
+   * Record a terminal observation failure.
+   * @param id - observed job.
+   * @param error - the stream's terminal failure.
+   */
+  observeFailed(id: JobId, error: unknown): void {
+    const state = this.observedStates.get(String(id))
+    if (state === undefined) return
+    state.view = { ...state.view, streaming: false, error: String(error) }
+    this.changed()
+  }
+
+  /**
+   * Drop observation state after the last observer stops.
+   * @param id - the no-longer-observed job.
+   */
+  observeStopped(id: JobId): void {
+    if (!this.observedStates.delete(String(id))) return
+    this.changed()
+  }
+
+  private changed(): void {
+    this.snapshotDirty = true
+    notifySubscribers(this.listeners, 'job output')
+  }
+}
+
+/** Bare observable source for the client job-output snapshot. */
+export interface JobOutputSource {
+  /** Read the identity-stable current snapshot. */
+  getSnapshot(): JobOutputSnapshot
+  /**
+   * Subscribe to snapshot changes.
+   * @param listener - invalidation callback.
+   * @returns unsubscribe function.
+   */
+  subscribe(listener: () => void): () => void
+}
+
+/** The job-output client service face. */
+export interface IJobOutput {
+  /** Per-job observation state. */
+  readonly state: JobOutputSource
+  /**
+   * Start observing one job's live record; reference-counted, so two viewers
+   * of the same job share one stream.
+   * @param sessionId - owning session used for the fenced read; undefined for an unowned job.
+   * @param id - job to observe.
+   * @returns stop function releasing this observer's reference.
+   */
+  observe(sessionId: SessionId | undefined, id: JobId): () => void
+}
+
+/** One reference-counted observation stream. */
+interface ObservationEntry {
+  refs: number
+  stopped: boolean
+  dispose: () => Promise<void>
+}
+
+/** Remote face the observation runner drives. */
+type JobObserveRemote = Pick<ClientRemote, '$stream'> & { readonly session: Pick<SessionRemote, 'observeJob'> }
+
+declare module '@deepseek-ai/cordis' {
+  interface Context {
+    /** React-free client job-record observation control. */
+    jobOutput: IJobOutput
+  }
+}
+
+/** Owns the bare job-output snapshot and per-job observation streams. */
+export class ClientJobOutput extends Service implements IJobOutput {
+  readonly state: JobOutputSource
+  private readonly entries = new Map<string, ObservationEntry>()
+
+  /**
+   * @param ctx - client root Context.
+   * @param remote - generated session namespace plus the Gateway stream factory.
+   * @param model - shared client job-output model.
+   */
+  constructor(
+    ctx: Context,
+    private readonly remote: JobObserveRemote,
+    private readonly model: ClientJobOutputModel,
+  ) {
+    super(ctx, 'jobOutput')
+    this.state = model
+    ctx.effect(() => () => {
+      const open = [...this.entries.values()]
+      this.entries.clear()
+      for (const entry of open) {
+        entry.stopped = true
+        void entry.dispose()
+      }
+    }, 'session-controller.client.observations')
+  }
+
+  observe(sessionId: SessionId | undefined, id: JobId): () => void {
+    const key = String(id)
+    const existing = this.entries.get(key)
+    if (existing !== undefined && !existing.stopped) {
+      existing.refs += 1
+      return this.releaser(key, existing)
+    }
+    const entry = this.startObservation(sessionId, id)
+    this.entries.set(key, entry)
+    return this.releaser(key, entry)
+  }
+
+  /**
+   * Release closures bind the exact entry they were minted for, never the
+   * map's current occupant: a later `observe()` on the same id may have
+   * replaced a stopped entry, and decrementing or disposing through the key
+   * alone would tear down that newer stream's references.
+   */
+  private releaser(key: string, entry: ObservationEntry): () => void {
+    let released = false
+    return () => {
+      if (released) return
+      released = true
+      entry.refs -= 1
+      if (entry.refs > 0) return
+      if (this.entries.get(key) === entry) this.entries.delete(key)
+      entry.stopped = true
+      void entry.dispose().then(() => {
+        // Clear the view only while no successor observation holds the key:
+        // a re-expand inside the dispose round-trip already re-anchored the
+        // model, and a stale clear would blank its panel for good.
+        if (this.entries.has(key)) return
+        // The key is the stringified branded id this releaser was minted for.
+        this.model.observeStopped(key as JobId)
+      })
+    }
+  }
+
+  private startObservation(sessionId: SessionId | undefined, id: JobId): ObservationEntry {
+    const name = `job observation ${String(id)}`
+    const stream = this.remote.$stream<SessionObserveJobFrame>({
+      name,
+      open: (signal) => {
+        const from = this.model.cursorOf(id)
+        return this.remote.session.observeJob(
+          {
+            jobId: id,
+            ...sessionId !== undefined ? { sessionId } : {},
+            ...from !== undefined ? { from } : {},
+          },
+          signal,
+        )
+      },
+      // A premature end after the anchor is retryable (a Host reload closes the
+      // generation); resuming from the cursor loses nothing. An end before the
+      // anchor is terminal.
+      ended: accepted => accepted
+        ? new RemoteStreamCarrierError(`${name} ended before settlement`)
+        : new Error(`${name} ended before its anchor`),
+    })
+    const entry: ObservationEntry = {
+      refs: 1,
+      stopped: false,
+      dispose: () => stream.dispose(),
+    }
+    void (async () => {
+      try {
+        for await (const item of stream) {
+          const frame = item.value
+          if (frame.type === 'opened') {
+            this.model.observeOpened(id, frame)
+            item.accept()
+            continue
+          }
+          if (frame.type === 'output') {
+            this.model.observeOutput(id, frame)
+            continue
+          }
+          // Terminal status: leave the loop before the generation end is
+          // classified, then close the stream for good.
+          this.model.observeStatus(id, frame)
+          break
+        }
+      } catch (error) {
+        if (!entry.stopped) this.model.observeFailed(id, error)
+      } finally {
+        entry.stopped = true
+        void entry.dispose()
+      }
+    })()
+    return entry
+  }
+}

packages/api/session-controller/src/control.ts

@@ -212,5 +212,6 @@ function jobView(job: JobSnapshot): SessionJob {
     ...(job.detail === undefined ? {} : { detail: job.detail }),
     startedAt: job.startedAt,
     ...(job.finishedAt === undefined ? {} : { finishedAt: job.finishedAt }),
+    ...(job.outputTotal === undefined ? {} : { outputTotal: job.outputTotal }),
   }
 }

packages/api/session-controller/src/index.ts

@@ -2,6 +2,7 @@
 
 import { Context } from '@deepseek-ai/cordis'
 import z from '@deepseek-ai/schemastery'
+import type {} from '@deepseek-ai/dsh-jobs'
 import { errorChain } from '@deepseek-ai/dsh-llm'
 import { canOpenNativePath, openNativePath } from '@deepseek-ai/dsh-native-command'
 import type { SessionEvent, SessionHeader, SessionId } from '@deepseek-ai/dsh-session'
@@ -15,6 +16,7 @@ import {
 import { SessionCommandController } from './commands.ts'
 import { SessionControlController } from './control.ts'
 import { SessionHistoryController } from './history.ts'
+import { observeJobRecord } from './observe-job.ts'
 import { SessionFileReferences } from './file-references.ts'
 import { ApiSessionList, DEFAULT_COLD_BLANK_PROBE_MAX_BYTES } from './list.ts'
 import { buildModelCatalog } from './catalog.ts'
@@ -37,6 +39,8 @@ import type {
   SessionKillJobValue,
   SessionListRequest,
   SessionListValue,
+  SessionObserveJobFrame,
+  SessionObserveJobRequest,
   SessionOpenWorkspacePathRequest,
   SessionOpenWorkspacePathValue,
   SessionPage,
@@ -65,12 +69,22 @@ declare module '@deepseek-ai/cordis' {
   }
 }
 
+/** Default coalescing window between job-observation reads, in milliseconds. */
+const DEFAULT_OBSERVE_FLUSH_MS = 100
+
+/** Default soft byte budget per job-observation output frame. */
+const DEFAULT_OBSERVE_MAX_FRAME_BYTES = 64 * 1024
+
 /** Session Controller deployment policy. */
 export interface Config {
   /** Maximum cold Session artifact size eligible for one full projection observation. */
   readonly coldBlankProbeMaxBytes?: number
   /** Override platform desktop-opener detection. */
   readonly nativeOpen?: boolean
+  /** Coalescing window after new record output before a job-observation read, in milliseconds (default 100). */
+  readonly observeFlushMs?: number
+  /** Soft byte budget per job-observation output frame (default 65536); one larger chunk ships whole. */
+  readonly observeMaxFrameBytes?: number
 }
 
 /** Host integrations replaceable by direct unit tests. */
@@ -98,6 +112,8 @@ export class SessionController extends TypertRemoteService {
   static Config: z<Config> = z.object({
     coldBlankProbeMaxBytes: z.natural().default(DEFAULT_COLD_BLANK_PROBE_MAX_BYTES),
     nativeOpen: z.boolean(),
+    observeFlushMs: z.natural().min(1).default(DEFAULT_OBSERVE_FLUSH_MS),
+    observeMaxFrameBytes: z.natural().min(1).default(DEFAULT_OBSERVE_MAX_FRAME_BYTES),
   })
 
   private readonly agents: ApiSessionAgentController
@@ -108,6 +124,8 @@ export class SessionController extends TypertRemoteService {
   private readonly openPath: (path: string, signal: AbortSignal) => Promise<void>
   private readonly canOpenPath: () => boolean
   private readonly promotions = new Set<Promise<void>>()
+  private readonly observeFlushMs: number
+  private readonly observeMaxFrameBytes: number
 
   /**
    * @param ctx - Host context containing the Session capability assembly.
@@ -132,6 +150,8 @@ export class SessionController extends TypertRemoteService {
     this.openPath = internals.openPath ?? openNativePath
     this.canOpenPath = internals.canOpenPath
       ?? (() => config.nativeOpen ?? (internals.openPath !== undefined || canOpenNativePath()))
+    this.observeFlushMs = config.observeFlushMs ?? DEFAULT_OBSERVE_FLUSH_MS
+    this.observeMaxFrameBytes = config.observeMaxFrameBytes ?? DEFAULT_OBSERVE_MAX_FRAME_BYTES
     ctx.plugin(SessionFileReferences)
     ctx.plugin(SessionSkillCatalog)
 
@@ -399,6 +419,30 @@ export class SessionController extends TypertRemoteService {
     return this.controlState.control(signal)
   }
 
+  /**
+   * Stream one job's retained record output from an absolute byte offset,
+   * then its terminal status once settled and drained. Non-consuming: the
+   * model-facing cursor and notice state never observe these reads. The
+   * request's session resolves the fenced-read caller; the registry rejects a
+   * job the session does not own, a job without a record, or an unknown job.
+   * @param request - target job, owning session, and optional resume offset.
+   * @param signal - cancellation owned by the Remote stream carrier.
+   * @returns anchor, coalesced output frames, and the terminal status.
+   */
+  @Remote({ mode: 'stream' })
+  observeJob(request: SessionObserveJobRequest, signal: AbortSignal): AsyncIterable<SessionObserveJobFrame> {
+    const registry = this.ctx.get('jobs')
+    if (registry === undefined) {
+      throw new Error('job observation unavailable: no job registry is loaded (load @deepseek-ai/dsh-jobs-local)')
+    }
+    const caller = request.sessionId === undefined ? undefined : this.ctx.agents.get(request.sessionId)
+    return observeJobRecord(registry, request, {
+      flushMs: this.observeFlushMs,
+      maxFrameBytes: this.observeMaxFrameBytes,
+      caller,
+    }, signal)
+  }
+
 }
 
 export { buildModelCatalog }

packages/api/session-controller/src/observe-job.ts

@@ -0,0 +1,191 @@
+/** Per-job record observation generations: anchor, coalesced output, terminal status. */
+
+import type { Agent } from '@deepseek-ai/dsh-agent'
+import type { JobId } from '@deepseek-ai/dsh-jobs/brand'
+import type { JobRegistry, JobSnapshot } from '@deepseek-ai/dsh-jobs'
+import type { SessionJobWireChunk, SessionObserveJobFrame, SessionObserveJobRequest } from './types.ts'
+
+/** Cadence and framing bounds for one observation generation. */
+export interface ObserveJobOptions {
+  /** Coalescing window after a wake before reading, in milliseconds. */
+  readonly flushMs: number
+  /**
+   * Soft byte budget per `output` frame. Accumulated chunks flush once the
+   * budget is met; one chunk larger than the budget ships whole (the
+   * registry's retention cap is the hard bound on any single chunk).
+   */
+  readonly maxFrameBytes: number
+  /** The live owning agent for fenced reads, resolved from the request's session. */
+  readonly caller: Agent | undefined
+}
+
+function isTerminal(status: JobSnapshot['status']): boolean {
+  return status !== 'running' && status !== 'stopping'
+}
+
+/** A record job's total offset; the anchor check proved the declaration and it never leaves. */
+function recordTotal(snapshot: JobSnapshot): number {
+  /* v8 ignore next -- a declared record never loses its offsets. */
+  return snapshot.outputTotal ?? 0
+}
+
+/** Wake-flag waiter: a wake between waits is never lost. */
+class OutputWaiter {
+  private dirty = false
+  private resolve: (() => void) | undefined
+
+  wake(): void {
+    this.dirty = true
+    this.resolve?.()
+  }
+
+  /**
+   * Resolve on the next wake, immediately when one already arrived, or on abort.
+   * @param signal - generation cancellation.
+   */
+  wait(signal: AbortSignal): Promise<void> {
+    if (this.dirty || signal.aborted) {
+      this.dirty = false
+      return Promise.resolve()
+    }
+    return new Promise((resolve) => {
+      const finish = (): void => {
+        signal.removeEventListener('abort', finish)
+        /* v8 ignore next -- one wait owns the sole installed resolver. */
+        if (this.resolve === finish) this.resolve = undefined
+        this.dirty = false
+        resolve()
+      }
+      this.resolve = finish
+      signal.addEventListener('abort', finish, { once: true })
+    })
+  }
+}
+
+function sleep(ms: number, signal: AbortSignal): Promise<void> {
+  if (signal.aborted) return Promise.resolve()
+  return new Promise((resolve) => {
+    const timer = setTimeout(done, ms)
+    function done(): void {
+      clearTimeout(timer)
+      signal.removeEventListener('abort', done)
+      resolve()
+    }
+    signal.addEventListener('abort', done, { once: true })
+  })
+}
+
+/**
+ * Stream one job's retained record output from an absolute offset: one
+ * `opened` anchor, coalesced `output` frames as the record advances, then one
+ * terminal `status` after the settled job is drained, after which the
+ * generation closes normally. Reads are non-consuming — the model-facing
+ * cursor and notice state never observe them; reconnecting callers resume by
+ * passing the last frame's `next` as `from`. Throws for a job that declared
+ * no record.
+ * @param registry - the live job registry.
+ * @param request - target job and optional resume offset.
+ * @param options - cadence, framing bounds, and the fenced-read caller.
+ * @param signal - generation cancellation owned by the Remote stream carrier.
+ * @returns the observation frame sequence for one generation.
+ */
+export async function* observeJobRecord(
+  registry: JobRegistry,
+  request: SessionObserveJobRequest,
+  options: ObserveJobOptions,
+  signal: AbortSignal,
+): AsyncIterable<SessionObserveJobFrame> {
+  if (request.from !== undefined && (!Number.isSafeInteger(request.from) || request.from < 0)) {
+    throw new Error(`invalid observe offset: expected a non-negative safe integer, got ${JSON.stringify(request.from)}`)
+  }
+  signal.throwIfAborted()
+  // The brand is nominal typing only; the wire boundary stamps it here rather
+  // than value-importing the optional registry package's constructor.
+  const id = String(request.jobId) as JobId
+  const waiter = new OutputWaiter()
+  // Subscribe before the first read so an append between the anchor read and
+  // the wait cannot be missed.
+  const unsubscribe = registry.onOutput((changed) => {
+    if (changed === id) waiter.wake()
+  })
+  try {
+    const caller = options.caller
+    let snapshot = registry.get(id, caller)
+    if (snapshot.outputTotal === undefined || snapshot.outputEarliest === undefined) {
+      throw new Error(`job ${id} declared no output record`)
+    }
+    let cursor = request.from ?? snapshot.outputEarliest
+    yield {
+      type: 'opened',
+      jobId: id,
+      from: cursor,
+      earliest: snapshot.outputEarliest,
+      total: snapshot.outputTotal,
+      status: snapshot.status,
+      ...snapshot.detail !== undefined ? { detail: snapshot.detail } : {},
+    }
+    while (!signal.aborted) {
+      const read = registry.readRecord(id, cursor, caller)
+      if (read.chunks.length > 0) {
+        yield* outputFrames(read.chunks, read.next, read.lossy, options.maxFrameBytes)
+      }
+      cursor = read.next
+      snapshot = registry.get(id, caller)
+      if (isTerminal(snapshot.status) && cursor >= recordTotal(snapshot)) {
+        yield {
+          type: 'status',
+          status: snapshot.status,
+          ...snapshot.detail !== undefined ? { detail: snapshot.detail } : {},
+          // Settlement always stamps finishedAt; the guard only discharges the optional field type.
+          /* v8 ignore next */
+          ...snapshot.finishedAt !== undefined ? { finishedAt: snapshot.finishedAt } : {},
+        }
+        return
+      }
+      await waiter.wait(signal)
+      // Let a burst accumulate so producer chatter becomes bounded frames.
+      await sleep(options.flushMs, signal)
+    }
+  } finally {
+    unsubscribe()
+  }
+}
+
+/** Split one read into frames along the soft per-frame byte budget. */
+function* outputFrames(
+  chunks: readonly SessionJobWireChunk[],
+  next: number,
+  lossy: boolean,
+  maxFrameBytes: number,
+): Iterable<SessionObserveJobFrame> {
+  let batch: SessionJobWireChunk[] = []
+  let batchBytes = 0
+  let flaggedLossy = lossy
+  for (const chunk of chunks) {
+    batch.push(chunk)
+    batchBytes += Buffer.byteLength(chunk.text, 'utf8')
+    if (batchBytes >= maxFrameBytes) {
+      const last = batch[batch.length - 1]
+      /* v8 ignore start -- a non-empty batch always has a last chunk; the arm only discharges noUncheckedIndexedAccess. */
+      const end = last === undefined ? next : last.at + Buffer.byteLength(last.text, 'utf8')
+      /* v8 ignore stop */
+      yield {
+        type: 'output',
+        chunks: batch,
+        next: end,
+        ...flaggedLossy ? { lossy: true as const } : {},
+      }
+      flaggedLossy = false
+      batch = []
+      batchBytes = 0
+    }
+  }
+  if (batch.length > 0 || flaggedLossy) {
+    yield {
+      type: 'output',
+      chunks: batch,
+      next,
+      ...flaggedLossy ? { lossy: true as const } : {},
+    }
+  }
+}

packages/api/session-controller/src/types.ts

@@ -482,6 +482,12 @@ export interface SessionJob {
   readonly detail?: string
   readonly startedAt: number
   readonly finishedAt?: number
+  /**
+   * Total UTF-8 bytes the job's observation record holds so far. Present
+   * exactly when the job declared a record — the row is observable through
+   * `session.observeJob`.
+   */
+  readonly outputTotal?: number
 }
 
 /** Complete live control baseline emitted once per control stream generation. */
@@ -506,6 +512,63 @@ export type SessionControlFrame =
   | { readonly type: 'jobs'; readonly sessionId: SessionId; readonly jobs: readonly SessionJob[] }
   | ({ readonly type: 'projection' } & SessionProjectionUpdate)
 
+/** Opens one observation stream over a job's retained record output. */
+export interface SessionObserveJobRequest {
+  /** Owning session used for the fenced read; omitted for an unowned job. */
+  readonly sessionId?: SessionId
+  readonly jobId: JobId
+  /**
+   * Absolute byte offset to resume from — a previous frame's `next`. Omitted
+   * starts at the oldest retained byte.
+   */
+  readonly from?: number
+}
+
+/** One retained record chunk on the wire. `channel` widens to string like `kind`. */
+export interface SessionJobWireChunk {
+  /** Absolute offset of the chunk's first byte. */
+  readonly at: number
+  readonly text: string
+  readonly channel?: string
+  /** Bytes immediately before this chunk were lost at the producer or to retention. */
+  readonly gapBefore?: true
+}
+
+/**
+ * Job observation stream frames: one `opened` anchor, then coalesced `output`
+ * batches, then — once the job has settled and its retained record is
+ * drained — one terminal `status`, after which the stream closes normally.
+ * Status rides the same stream as output so settlement can never race a
+ * still-open output channel.
+ */
+export type SessionObserveJobFrame =
+  | {
+    readonly type: 'opened'
+    readonly jobId: JobId
+    /** Offset the first `output` frame continues from. */
+    readonly from: number
+    /** Oldest retained byte at open time; `from < earliest` means the head is gone. */
+    readonly earliest: number
+    /** Total bytes produced at open time. */
+    readonly total: number
+    readonly status: SessionJob['status']
+    readonly detail?: string
+  }
+  | {
+    readonly type: 'output'
+    readonly chunks: readonly SessionJobWireChunk[]
+    /** Offset to resume from after this frame. */
+    readonly next: number
+    /** Bytes between the requested offset and `chunks` were already evicted. */
+    readonly lossy?: true
+  }
+  | {
+    readonly type: 'status'
+    readonly status: SessionJob['status']
+    readonly detail?: string
+    readonly finishedAt?: number
+  }
+
 declare module '@deepseek-ai/cordis' {
   interface Events {
     /**