feat(api): serve job-record observation from the session namespace

SessionJob rows carry outputTotal exactly for record-declaring jobs, and a
new session.observeJob stream serves one job's retained record from an
absolute offset (anchor, coalesced output frames, terminal status) with the
fenced read resolved from the request's session — the same authorization
shape as the jobs roster. The client half gains ctx.jobOutput: the
reference-counted per-job observation service and its accumulated view
model, ported from the activity controller this replaces.

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,7 @@ export const inject = [
 export function apply(ctx: Context): void {
   const remotes = ctx.remote as unknown as SessionRemotes
   const sessions = new ClientSessions(ctx, remotes)
+  new ClientJobOutput(ctx, remotes, 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'
@@ -35,6 +37,8 @@ import type {
   SessionForkValue,
   SessionListRequest,
   SessionListValue,
+  SessionObserveJobFrame,
+  SessionObserveJobRequest,
   SessionOpenWorkspacePathRequest,
   SessionOpenWorkspacePathValue,
   SessionPage,
@@ -63,12 +67,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. */
@@ -96,6 +110,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
@@ -106,6 +122,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.
@@ -130,6 +148,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)
 
@@ -386,6 +406,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,189 @@
+/** Per-job record observation generations: anchor, coalesced output, terminal status. */
+
+import type { Agent } from '@deepseek-ai/dsh-agent'
+import { JobId } from '@deepseek-ai/dsh-jobs'
+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()
+  const id = JobId(String(request.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

@@ -469,6 +469,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. */
@@ -493,6 +499,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 {
     /**

packages/api/session-controller/tests/control-jobs.host.spec.ts

@@ -162,6 +162,25 @@ describe('Session control jobs updates', () => {
     ])
   })
 
+  it('carries outputTotal exactly for record-declaring jobs', async () => {
+    const { ctx, agent, control } = await harness(true)
+    const abort = new AbortController()
+    const collected = collectJobs(control.control(abort.signal), 1, abort)
+    ctx.jobs.start({
+      kind: 'bash',
+      label: 'observable run',
+      owner: agent,
+      record: true,
+      run: (job) => {
+        job.append('hi')
+        return { cancel: () => {}, done: new Promise<never>(() => {}) }
+      },
+    })
+
+    const [frame] = await collected
+    expect(frame?.jobs[0]?.outputTotal).toBe(2)
+  })
+
   it('fans an unowned change out to every attached session', async () => {
     const { ctx, control } = await harness(true)
     const second = ctx.sessions.create()

packages/api/session-controller/tests/job-output.client.spec.ts

@@ -0,0 +1,413 @@
+import { describe, expect, it, vi } from 'vitest'
+import { Context } from '@deepseek-ai/cordis'
+import type { JobId } from '@deepseek-ai/dsh-jobs/brand'
+import type { SessionId } from '@deepseek-ai/dsh-session/types'
+import { ClientJobOutput, ClientJobOutputModel } from '../src/client/job-output.ts'
+import type { SessionObserveJobFrame } from '../src/types.ts'
+
+const ID = 'bash-1' as JobId
+
+function opened(model: ClientJobOutputModel, over: Partial<{ from: number; earliest: number; total: number }> = {}): void {
+  model.observeOpened(ID, {
+    type: 'opened',
+    jobId: ID,
+    from: 0,
+    earliest: 0,
+    total: 0,
+    status: 'running',
+    ...over,
+  })
+}
+
+describe('ClientJobOutputModel observation', () => {
+  it('accumulates output, advances the resume cursor, and settles on status', () => {
+    const model = new ClientJobOutputModel()
+    opened(model)
+    expect(model.cursorOf(ID)).toBe(0)
+    model.observeOutput(ID, { type: 'output', chunks: [{ at: 0, text: 'a' }, { at: 1, text: 'b' }], next: 2 })
+    model.observeOutput(ID, { type: 'output', chunks: [{ at: 2, text: 'c' }], next: 3 })
+    expect(model.cursorOf(ID)).toBe(3)
+    const view = model.getSnapshot().observed[String(ID)]
+    expect(view?.text).toBe('abc')
+    expect(view?.streaming).toBe(true)
+    model.observeStatus(ID, { type: 'status', status: 'completed', detail: 'exit code: 0' })
+    const settled = model.getSnapshot().observed[String(ID)]
+    expect(settled?.streaming).toBe(false)
+    expect(settled?.status).toBe('completed')
+    expect(settled?.detail).toBe('exit code: 0')
+  })
+
+  it('keeps snapshot identity stable between changes and notifies subscribers', () => {
+    const model = new ClientJobOutputModel()
+    let notified = 0
+    const unsubscribe = model.subscribe(() => { notified += 1 })
+    const before = model.getSnapshot()
+    expect(model.getSnapshot()).toBe(before)
+    opened(model)
+    expect(notified).toBe(1)
+    expect(model.getSnapshot()).not.toBe(before)
+    unsubscribe()
+    model.observeStopped(ID)
+    expect(notified).toBe(1)
+  })
+
+  it('marks gaps from lossy frames, gap chunks, and a resume behind the retained head', () => {
+    const model = new ClientJobOutputModel()
+    opened(model, { from: 4, earliest: 8, total: 10 })
+    expect(model.getSnapshot().observed[String(ID)]?.gapBefore).toBe(true)
+
+    const clean = new ClientJobOutputModel()
+    opened(clean)
+    clean.observeOutput(ID, { type: 'output', chunks: [{ at: 0, text: 'x', gapBefore: true }], next: 1 })
+    expect(clean.getSnapshot().observed[String(ID)]?.gapBefore).toBe(true)
+
+    const lossy = new ClientJobOutputModel()
+    opened(lossy)
+    lossy.observeOutput(ID, { type: 'output', chunks: [], next: 5, lossy: true })
+    expect(lossy.getSnapshot().observed[String(ID)]?.gapBefore).toBe(true)
+  })
+
+  it('marks a fresh observation anchored past the evicted head', () => {
+    // Fresh observations anchor at the registry's earliest retained byte, so
+    // from === earliest > 0 means the head was already discarded.
+    const fresh = new ClientJobOutputModel()
+    opened(fresh, { from: 60_240, earliest: 60_240, total: 321_328 })
+    expect(fresh.getSnapshot().observed[String(ID)]?.gapBefore).toBe(true)
+
+    // A retry that accumulated no text yet earns the mark the same way.
+    const retried = new ClientJobOutputModel()
+    opened(retried)
+    expect(retried.getSnapshot().observed[String(ID)]?.gapBefore).toBe(false)
+    opened(retried, { from: 6, earliest: 6, total: 6 })
+    expect(retried.getSnapshot().observed[String(ID)]?.gapBefore).toBe(true)
+  })
+
+  it('bounds the render tail without splitting a surrogate pair', () => {
+    const model = new ClientJobOutputModel()
+    opened(model)
+    const emoji = '😀'.repeat((64 * 1024) + 8)
+    model.observeOutput(ID, { type: 'output', chunks: [{ at: 0, text: emoji }], next: emoji.length * 2 })
+    const view = model.getSnapshot().observed[String(ID)]
+    expect(view?.gapBefore).toBe(true)
+    expect(view!.text.length).toBeLessThanOrEqual(128 * 1024)
+    // The bound landed between pairs: the surviving text still round-trips.
+    expect(/^(?:😀)+$/u.test(view!.text)).toBe(true)
+  })
+
+  it('carries the anchor detail and trims a plain-ASCII tail without a boundary shift', () => {
+    const model = new ClientJobOutputModel()
+    model.observeOpened(ID, {
+      type: 'opened', jobId: ID, from: 0, earliest: 0, total: 0, status: 'running', detail: 'exit soon',
+    })
+    expect(model.getSnapshot().observed[String(ID)]?.detail).toBe('exit soon')
+    const long = 'x'.repeat((128 * 1024) + 5)
+    model.observeOutput(ID, { type: 'output', chunks: [{ at: 0, text: long }], next: long.length })
+    const view = model.getSnapshot().observed[String(ID)]
+    expect(view?.text.length).toBe(128 * 1024)
+    expect(view?.gapBefore).toBe(true)
+  })
+
+  it('advances the cut past a low surrogate landing exactly on the bound', () => {
+    const model = new ClientJobOutputModel()
+    opened(model)
+    // 'z' + one emoji + odd ASCII tail puts a low surrogate exactly at the cut index.
+    const text = 'z😀' + 'a'.repeat((128 * 1024) - 1)
+    model.observeOutput(ID, { type: 'output', chunks: [{ at: 0, text }], next: text.length })
+    const view = model.getSnapshot().observed[String(ID)]
+    expect(view?.text.length).toBe((128 * 1024) - 1)
+    expect(view?.text.startsWith('a')).toBe(true)
+  })
+
+  it('preserves accumulated text across a reconnect anchor and clears on stop', () => {
+    const model = new ClientJobOutputModel()
+    opened(model)
+    model.observeOutput(ID, { type: 'output', chunks: [{ at: 0, text: 'kept' }], next: 4 })
+    opened(model, { from: 4, earliest: 0, total: 4 })
+    expect(model.getSnapshot().observed[String(ID)]?.text).toBe('kept')
+    model.observeStopped(ID)
+    expect(model.getSnapshot().observed[String(ID)]).toBeUndefined()
+    expect(model.cursorOf(ID)).toBeUndefined()
+    // A second stop is inert.
+    model.observeStopped(ID)
+  })
+
+  it('records a terminal stream failure on the live view', () => {
+    const model = new ClientJobOutputModel()
+    opened(model)
+    model.observeFailed(ID, new Error('carrier gone'))
+    const view = model.getSnapshot().observed[String(ID)]
+    expect(view?.streaming).toBe(false)
+    expect(view?.error).toContain('carrier gone')
+    // A failure for an untracked id is inert.
+    model.observeFailed('bash-9' as JobId, new Error('ignored'))
+  })
+})
+
+/** One scripted logical stream: frames are pushed by the test, never reopened. */
+class FakeStream {
+  disposed = false
+  /** Model a carrier whose disposal surfaces as an iterator throw. */
+  throwOnDispose = false
+  /** Hold the dispose promise open until {@link releaseDispose}. */
+  deferDispose = false
+  private disposeRelease: (() => void) | undefined
+  private disposePending: Promise<void> | undefined
+  private readonly frames: SessionObserveJobFrame[] = []
+  private waiter: (() => void) | undefined
+  private failure: Error | undefined
+  readonly accepts: number[] = []
+
+  push(frame: SessionObserveJobFrame): void {
+    this.frames.push(frame)
+    this.waiter?.()
+  }
+
+  poison(error: Error): void {
+    this.failure = error
+    this.waiter?.()
+  }
+
+  dispose(): Promise<void> {
+    if (this.throwOnDispose && this.failure === undefined) {
+      this.failure = new Error('carrier tore down')
+    } else {
+      this.disposed = true
+    }
+    this.waiter?.()
+    if (this.deferDispose) {
+      // Repeat disposals (releaser plus the consumer's finally) share one
+      // deferred promise, so releaseDispose resumes every waiter.
+      this.disposePending ??= new Promise((resolve) => { this.disposeRelease = resolve })
+      return this.disposePending
+    }
+    return Promise.resolve()
+  }
+
+  releaseDispose(): void {
+    this.disposeRelease?.()
+  }
+
+  async *[Symbol.asyncIterator]() {
+    let generation = 1
+    while (!this.disposed) {
+      if (this.failure !== undefined) throw this.failure
+      const frame = this.frames.shift()
+      if (frame === undefined) {
+        await new Promise<void>((resolve) => { this.waiter = resolve })
+        continue
+      }
+      yield {
+        generation,
+        value: frame,
+        signal: new AbortController().signal,
+        accept: () => { this.accepts.push(generation) },
+      }
+      generation = 1
+    }
+  }
+}
+
+function bench() {
+  const ctx = new Context()
+  const model = new ClientJobOutputModel()
+  const streams: { options: { open: (signal: AbortSignal) => unknown }; stream: FakeStream }[] = []
+  const observeCalls: unknown[] = []
+  const remote = {
+    $stream: (options: { open: (signal: AbortSignal) => unknown }) => {
+      const stream = new FakeStream()
+      streams.push({ options, stream })
+      return stream
+    },
+    session: {
+      observeJob: (request: unknown) => {
+        observeCalls.push(request)
+        return { [Symbol.asyncIterator]: async function* () { /* never yields */ } }
+      },
+    },
+  }
+  const output = new ClientJobOutput(ctx, remote as never, model)
+  return { ctx, model, output, streams, observeCalls }
+}
+
+const tick = () => new Promise<void>(resolve => setTimeout(resolve, 0))
+
+describe('ClientJobOutput observation streams', () => {
+  it('feeds anchor, output, and terminal frames into the model, then closes', async () => {
+    const { model, output, streams } = bench()
+    output.observe(undefined, ID)
+    const { stream } = streams[0]!
+    stream.push({ type: 'opened', jobId: ID, from: 0, earliest: 0, total: 0, status: 'running' })
+    stream.push({ type: 'output', chunks: [{ at: 0, text: 'hi' }], next: 2 })
+    await tick()
+    expect(model.getSnapshot().observed[String(ID)]?.text).toBe('hi')
+    expect(stream.accepts).toHaveLength(1)
+
+    stream.push({ type: 'status', status: 'completed' })
+    await tick()
+    expect(model.getSnapshot().observed[String(ID)]?.streaming).toBe(false)
+    expect(stream.disposed).toBe(true)
+  })
+
+  it('shares one stream across observers and stops after the last release', async () => {
+    const { model, output, streams } = bench()
+    const first = output.observe(undefined, ID)
+    const second = output.observe(undefined, ID)
+    expect(streams).toHaveLength(1)
+    streams[0]!.stream.push({ type: 'opened', jobId: ID, from: 0, earliest: 0, total: 0, status: 'running' })
+    await tick()
+
+    first()
+    first()
+    expect(streams[0]!.stream.disposed).toBe(false)
+    second()
+    await tick()
+    expect(streams[0]!.stream.disposed).toBe(true)
+    expect(model.getSnapshot().observed[String(ID)]).toBeUndefined()
+  })
+
+  it('a late release of a superseded observation neither tears down nor clears its successor', async () => {
+    const { model, output, streams } = bench()
+    const firstStop = output.observe(undefined, ID)
+    const first = streams[0]!.stream
+    first.push({ type: 'opened', jobId: ID, from: 0, earliest: 0, total: 0, status: 'running' })
+    first.push({ type: 'status', status: 'completed' })
+    await tick()
+    // A successor observation replaces the settled entry while the first
+    // observer still holds its releaser.
+    const secondStop = output.observe(undefined, ID)
+    const second = streams[1]!.stream
+    second.push({ type: 'opened', jobId: ID, from: 0, earliest: 0, total: 0, status: 'running' })
+    await tick()
+
+    firstStop()
+    await tick()
+    expect(second.disposed).toBe(false)
+    expect(model.getSnapshot().observed[String(ID)]).toBeDefined()
+
+    secondStop()
+    await tick()
+    expect(second.disposed).toBe(true)
+    expect(model.getSnapshot().observed[String(ID)]).toBeUndefined()
+  })
+
+  it('a re-observation inside the dispose round-trip keeps its fresh view', async () => {
+    const { model, output, streams } = bench()
+    const stop = output.observe(undefined, ID)
+    const first = streams[0]!.stream
+    first.deferDispose = true
+    first.push({ type: 'opened', jobId: ID, from: 0, earliest: 0, total: 0, status: 'running' })
+    await tick()
+
+    stop()
+    // Re-expand while the previous generation's dispose is still in flight.
+    output.observe(undefined, ID)
+    const second = streams[1]!.stream
+    second.push({ type: 'opened', jobId: ID, from: 0, earliest: 0, total: 0, status: 'running' })
+    await tick()
+
+    first.releaseDispose()
+    await tick()
+    // The stale post-dispose clear must not blank the successor's view.
+    expect(model.getSnapshot().observed[String(ID)]).toBeDefined()
+    second.push({ type: 'output', chunks: [{ at: 0, text: 'alive' }], next: 5 })
+    await tick()
+    expect(model.getSnapshot().observed[String(ID)]?.text).toBe('alive')
+  })
+
+  it('opens a fresh stream for a re-observed job and resumes from the model cursor', async () => {
+    const { output, streams, observeCalls, model } = bench()
+    const stop = output.observe('alice' as SessionId, ID)
+    streams[0]!.stream.push({ type: 'opened', jobId: ID, from: 0, earliest: 0, total: 0, status: 'running' })
+    streams[0]!.stream.push({ type: 'output', chunks: [{ at: 0, text: 'abc' }], next: 3 })
+    await tick()
+    // The generation opener reads the live cursor at open time and carries the
+    // fenced-read session.
+    streams[0]!.options.open(new AbortController().signal)
+    expect(observeCalls.at(-1)).toEqual({ jobId: ID, sessionId: 'alice', from: 3 })
+
+    stop()
+    await tick()
+    output.observe(undefined, ID)
+    expect(streams).toHaveLength(2)
+    streams[1]!.options.open(new AbortController().signal)
+    // Stopped observation dropped the cursor: the fresh stream starts unanchored.
+    expect(observeCalls.at(-1)).toEqual({ jobId: ID })
+    expect(model.cursorOf(ID)).toBeUndefined()
+  })
+
+  it('classifies a premature end as retryable only after the anchor', () => {
+    const { output, streams } = bench()
+    output.observe(undefined, ID)
+    const options = streams[0]!.options as unknown as { ended: (accepted: boolean) => Error }
+    expect(options.ended(true).name).toBe('RemoteStreamCarrierError')
+    expect(options.ended(false).name).toBe('Error')
+  })
+
+  it('records a consumer failure on the live view', async () => {
+    const { model, output, streams } = bench()
+    output.observe(undefined, ID)
+    const { stream } = streams[0]!
+    stream.push({ type: 'opened', jobId: ID, from: 0, earliest: 0, total: 0, status: 'running' })
+    await tick()
+    stream.poison(new Error('frame decode broke'))
+    await tick()
+    const view = model.getSnapshot().observed[String(ID)]
+    expect(view?.error).toContain('frame decode broke')
+    expect(view?.streaming).toBe(false)
+  })
+
+  it('treats a release after service disposal as inert and stays silent for post-stop failures', async () => {
+    const { ctx, model, output, streams } = bench()
+    const stop = output.observe(undefined, ID)
+    const { stream } = streams[0]!
+    stream.push({ type: 'opened', jobId: ID, from: 0, earliest: 0, total: 0, status: 'running' })
+    await tick()
+    // Poison wakes the consumer, but the synchronous stop lands first, so the
+    // failure arrives on an already-stopped entry and stays silent.
+    stream.poison(new Error('after stop'))
+    stop()
+    await tick()
+    expect(model.getSnapshot().observed[String(ID)]).toBeUndefined()
+
+    const second = output.observe(undefined, ID)
+    await ctx.fiber.dispose()
+    // The disposal path already removed every entry; a live releaser whose
+    // entry is gone is inert.
+    second()
+  })
+
+  it('stays silent when the carrier teardown itself throws after a stop', async () => {
+    const { model, output, streams } = bench()
+    const stop = output.observe(undefined, ID)
+    const { stream } = streams[0]!
+    stream.throwOnDispose = true
+    stream.push({ type: 'opened', jobId: ID, from: 0, earliest: 0, total: 0, status: 'running' })
+    await tick()
+    // The releaser marks the entry stopped, then its dispose surfaces as an
+    // iterator throw — which the consumer swallows for a stopped entry.
+    stop()
+    await tick()
+    expect(model.getSnapshot().observed[String(ID)]?.error).toBeUndefined()
+  })
+
+  it('records a terminal stream failure and service disposal closes live streams', async () => {
+    const { ctx, model, output, streams } = bench()
+    output.observe(undefined, ID)
+    const failing = streams[0]!.stream
+    failing.push({ type: 'opened', jobId: ID, from: 0, earliest: 0, total: 0, status: 'running' })
+    await tick()
+    const iterator = failing[Symbol.asyncIterator]()
+    void iterator
+    // Simulate a terminal consumer failure by disposing the underlying stream:
+    // the consumer loop ends without a status frame and cleans up.
+    const spy = vi.spyOn(model, 'observeFailed')
+    await failing.dispose()
+    await tick()
+
+    const other = 'bash-2' as JobId
+    output.observe(undefined, other)
+    await ctx.fiber.dispose()
+    expect(streams.every(entry => entry.stream.disposed)).toBe(true)
+    expect(spy).not.toHaveBeenCalled()
+  })
+})

packages/api/session-controller/tests/observe-job.host.spec.ts

@@ -0,0 +1,298 @@
+import { describe, expect, it } from 'vitest'
+import { Context } from '@deepseek-ai/cordis'
+import AgentRegistry, { Inbox } from '@deepseek-ai/dsh-agent'
+import type { Agent } from '@deepseek-ai/dsh-agent'
+import { JobId } from '@deepseek-ai/dsh-jobs'
+import type { JobOutcome, RunningJob } from '@deepseek-ai/dsh-jobs'
+import LocalJobRegistry from '@deepseek-ai/dsh-jobs-local'
+import SessionStore, { Session, SessionId } from '@deepseek-ai/dsh-session'
+import { observeJobRecord } from '../src/observe-job.ts'
+import type { ObserveJobOptions } from '../src/observe-job.ts'
+import type { SessionObserveJobFrame } from '../src/types.ts'
+import { createSessionTestController } from './test-remote.ts'
+
+async function harness() {
+  const ctx = new Context()
+  await ctx.plugin(AgentRegistry)
+  await ctx.plugin(LocalJobRegistry, { retainBytes: 64, settledRetainBytes: 64 })
+  ctx.jobs.attachController('observe-job-test')
+  return ctx
+}
+
+function registerAgent(ctx: Context, rawId: string): Agent {
+  const scopeFiber = ctx.plugin(() => {})
+  const id = SessionId(rawId)
+  const session = Session.create(id)
+  const agent = {
+    id,
+    session,
+    inbox: new Inbox(session, { inserted: () => {}, discarded: () => {}, claimed: () => {} }),
+    status: 'idle',
+    ctx: scopeFiber.ctx,
+  } as unknown as Agent
+  ctx.agents.register(agent)
+  return agent
+}
+
+/** Start one record job whose done settles on demand; expose the producer face. */
+function startRecordJob(ctx: Context, options: { label?: string; owner?: Agent; record?: boolean } = {}) {
+  let settle!: (outcome: JobOutcome) => void
+  let face!: RunningJob
+  const id = ctx.jobs.start({
+    kind: 'bash',
+    label: options.label ?? 'echo',
+    ...options.owner !== undefined ? { owner: options.owner } : {},
+    ...options.record === false ? {} : { record: true },
+    run: (job) => {
+      face = job
+      return {
+        cancel() {},
+        done: new Promise<JobOutcome>((resolve) => { settle = resolve }),
+      }
+    },
+  })
+  return {
+    id,
+    append: (text: string, opts?: Parameters<RunningJob['append']>[1]) => { face.append(text, opts) },
+    settle: async (outcome: JobOutcome) => {
+      settle(outcome)
+      await new Promise(resolve => setTimeout(resolve, 0))
+    },
+  }
+}
+
+/** Read frames until `count` arrive or the generator finishes, then abort. */
+async function collect<Frame>(
+  iterable: AsyncIterable<Frame>,
+  count: number,
+  abort: AbortController,
+): Promise<Frame[]> {
+  const frames: Frame[] = []
+  const iterator = iterable[Symbol.asyncIterator]()
+  try {
+    while (frames.length < count) {
+      const step = await iterator.next()
+      if (step.done) return frames
+      frames.push(step.value)
+    }
+  } finally {
+    abort.abort()
+    await iterator.next().catch(() => undefined)
+  }
+  return frames
+}
+
+const OBSERVE: Omit<ObserveJobOptions, 'caller'> = { flushMs: 5, maxFrameBytes: 1024 }
+
+describe('observeJobRecord', () => {
+  function observed(ctx: Context, request: { jobId: JobId; from?: number }, caller?: Agent) {
+    const abort = new AbortController()
+    const frames: SessionObserveJobFrame[] = []
+    const stream = observeJobRecord(ctx.jobs, request, { ...OBSERVE, caller }, abort.signal)
+    const done = (async () => {
+      for await (const frame of stream) frames.push(frame)
+    })()
+    return { abort, frames, done }
+  }
+
+  it('anchors, streams coalesced output, and closes with the terminal status', async () => {
+    const ctx = await harness()
+    const job = startRecordJob(ctx)
+    job.append('early ')
+    const { frames, done } = observed(ctx, { jobId: job.id })
+    await new Promise(resolve => setTimeout(resolve, 20))
+    job.append('live', { channel: 'stderr' })
+    await new Promise(resolve => setTimeout(resolve, 20))
+    await job.settle({ status: 'completed', detail: 'exit code: 0' })
+    await done
+
+    expect(frames[0]).toMatchObject({ type: 'opened', from: 0, earliest: 0, total: 6, status: 'running' })
+    const outputs = frames.filter(frame => frame.type === 'output')
+    const text = outputs.flatMap(frame => frame.chunks.map(chunk => chunk.text)).join('')
+    expect(text).toBe('early live')
+    expect(outputs.at(-1)?.next).toBe(10)
+    expect(outputs.flatMap(frame => frame.chunks).find(chunk => chunk.text === 'live')?.channel).toBe('stderr')
+    expect(frames.at(-1)).toMatchObject({ type: 'status', status: 'completed', detail: 'exit code: 0' })
+  })
+
+  it('resumes from an explicit offset and flags an evicted resume lossy', async () => {
+    const ctx = await harness()
+    const job = startRecordJob(ctx, { label: 'spam' })
+    job.append('a'.repeat(60))
+    job.append('b'.repeat(60))
+    await job.settle({ status: 'completed' })
+
+    const fresh = observed(ctx, { jobId: job.id, from: 0 })
+    await fresh.done
+    const output = fresh.frames.find(frame => frame.type === 'output')
+    expect(output?.lossy).toBe(true)
+    expect(output?.chunks.map(chunk => chunk.text).join('')).toBe('b'.repeat(60))
+
+    const resumed = observed(ctx, { jobId: job.id, from: 120 })
+    await resumed.done
+    expect(resumed.frames.some(frame => frame.type === 'output')).toBe(false)
+    expect(resumed.frames.at(-1)?.type).toBe('status')
+  })
+
+  it('splits oversized reads along the frame budget', async () => {
+    const ctx = await harness()
+    const job = startRecordJob(ctx, { label: 'wide' })
+    job.append('x'.repeat(30))
+    job.append('y'.repeat(30))
+    await job.settle({ status: 'completed' })
+    const abort = new AbortController()
+    const frames = await collect<SessionObserveJobFrame>(
+      observeJobRecord(ctx.jobs, { jobId: job.id }, {
+        flushMs: 5,
+        maxFrameBytes: 30,
+        caller: undefined,
+      }, abort.signal),
+      5,
+      abort,
+    )
+    const outputs = frames.filter(frame => frame.type === 'output')
+    expect(outputs).toHaveLength(2)
+    expect(outputs[0]?.next).toBe(30)
+    expect(outputs[1]?.next).toBe(60)
+  })
+
+  it('carries the lossy flag on the first frame of a split evicted read', async () => {
+    const ctx = await harness()
+    const job = startRecordJob(ctx, { label: 'evicted split' })
+    job.append('a'.repeat(40))
+    job.append('b'.repeat(40))
+    job.append('c'.repeat(40))
+    await job.settle({ status: 'completed' })
+    const abort = new AbortController()
+    const frames = await collect<SessionObserveJobFrame>(
+      observeJobRecord(ctx.jobs, { jobId: job.id, from: 0 }, {
+        flushMs: 5,
+        maxFrameBytes: 40,
+        caller: undefined,
+      }, abort.signal),
+      5,
+      abort,
+    )
+    const outputs = frames.filter(frame => frame.type === 'output')
+    expect(outputs[0]?.lossy).toBe(true)
+    expect(outputs[1]?.lossy).toBeUndefined()
+  })
+
+  it('reads owned records through the resolved caller and rejects foreign, bad, and unknown input', async () => {
+    const ctx = await harness()
+    const owner = registerAgent(ctx, 'alice')
+    const job = startRecordJob(ctx, { label: 'secret', owner })
+    job.append('classified')
+    await job.settle({ status: 'completed' })
+
+    const { frames, done } = observed(ctx, { jobId: job.id }, owner)
+    await done
+    expect(frames.some(frame => frame.type === 'output'
+      && frame.chunks.some(chunk => chunk.text === 'classified'))).toBe(true)
+
+    const foreign = new AbortController()
+    await expect(collect(
+      observeJobRecord(ctx.jobs, { jobId: job.id }, { ...OBSERVE, caller: undefined }, foreign.signal),
+      1,
+      foreign,
+    )).rejects.toThrow(/belongs to another session/)
+
+    const abort = new AbortController()
+    await expect(collect(
+      observeJobRecord(ctx.jobs, { jobId: job.id, from: -1 }, { ...OBSERVE, caller: owner }, abort.signal),
+      1,
+      abort,
+    )).rejects.toThrow(/invalid observe offset/)
+
+    const abortUnknown = new AbortController()
+    await expect(collect(
+      observeJobRecord(ctx.jobs, { jobId: JobId('bash-999') }, { ...OBSERVE, caller: undefined }, abortUnknown.signal),
+      1,
+      abortUnknown,
+    )).rejects.toThrow(/unknown job/)
+  })
+
+  it('refuses a job that declared no record', async () => {
+    const ctx = await harness()
+    const job = startRecordJob(ctx, { record: false })
+    const abort = new AbortController()
+    await expect(collect(
+      observeJobRecord(ctx.jobs, { jobId: job.id }, { ...OBSERVE, caller: undefined }, abort.signal),
+      1,
+      abort,
+    )).rejects.toThrow(/declared no output record/)
+    await job.settle({ status: 'completed' })
+  })
+
+  it('ignores output signals for other jobs while waiting', async () => {
+    const ctx = await harness()
+    const watched = startRecordJob(ctx, { label: 'watched' })
+    const noisy = startRecordJob(ctx, { label: 'noisy' })
+    const { abort, frames, done } = observed(ctx, { jobId: watched.id })
+    await new Promise(resolve => setTimeout(resolve, 15))
+    noisy.append('unrelated')
+    await new Promise(resolve => setTimeout(resolve, 15))
+    await watched.settle({ status: 'completed' })
+    await done
+    abort.abort()
+    expect(frames.filter(frame => frame.type === 'output')).toEqual([])
+    await noisy.settle({ status: 'completed' })
+  })
+
+  it('stops cleanly on abort while waiting for output', async () => {
+    const ctx = await harness()
+    const job = startRecordJob(ctx, { label: 'idle' })
+    const { abort, frames, done } = observed(ctx, { jobId: job.id })
+    await new Promise(resolve => setTimeout(resolve, 20))
+    abort.abort()
+    await done
+    expect(frames[0]?.type).toBe('opened')
+    expect(frames.at(-1)?.type).not.toBe('status')
+    await job.settle({ status: 'completed' })
+  })
+})
+
+describe('SessionController.observeJob', () => {
+  const defaults = {
+    defaultModelSelection: () => ({ provider: 'fixture', model: 'fixture-model' }),
+    cwd: '/tmp',
+  }
+
+  it('refuses observation without a job registry', async () => {
+    const ctx = new Context()
+    await ctx.plugin(SessionStore)
+    await ctx.plugin(AgentRegistry)
+    const controller = createSessionTestController(ctx, defaults)
+    expect(() => controller.observeJob({ jobId: JobId('bash-1') }, new AbortController().signal))
+      .toThrow(/job observation unavailable/)
+  })
+
+  it('resolves the fenced-read caller from the request session', async () => {
+    const ctx = new Context()
+    await ctx.plugin(SessionStore)
+    await ctx.plugin(AgentRegistry)
+    await ctx.plugin(LocalJobRegistry)
+    ctx.jobs.attachController('observe-job-test')
+    const controller = createSessionTestController(ctx, defaults)
+    const session = ctx.sessions.create(SessionId('observing-session'))
+    const owner = {
+      id: session.id,
+      session,
+      inbox: new Inbox(session, { inserted: () => {}, discarded: () => {}, claimed: () => {} }),
+      status: 'idle',
+      ctx,
+    } as unknown as Agent
+    ctx.agents.register(owner)
+    const job = startRecordJob(ctx, { label: 'owned run', owner })
+    job.append('hi')
+    await job.settle({ status: 'completed' })
+
+    const abort = new AbortController()
+    const frames = await collect<SessionObserveJobFrame>(
+      controller.observeJob({ sessionId: session.id, jobId: job.id }, abort.signal),
+      3,
+      abort,
+    )
+    expect(frames.map(frame => frame.type)).toEqual(['opened', 'output', 'status'])
+  })
+})

packages/api/session-controller/tests/test-remote.ts

@@ -34,6 +34,8 @@ import type {
   SessionFollowRequest,
   SessionListRequest,
   SessionListValue,
+  SessionObserveJobFrame,
+  SessionObserveJobRequest,
   SessionOpenWorkspacePathRequest,
   SessionOpenWorkspacePathValue,
   SessionPage,
@@ -71,6 +73,7 @@ export interface TestSessionRemote {
   page(request: SessionPageRequest, signal?: AbortSignal): Promise<RemoteResult<SessionPage>>
   follow(request: SessionFollowRequest, signal?: AbortSignal): AsyncIterable<SessionFollowFrame>
   control(signal?: AbortSignal): AsyncIterable<SessionControlFrame>
+  observeJob(request: SessionObserveJobRequest, signal?: AbortSignal): AsyncIterable<SessionObserveJobFrame>
 }
 
 /** Dependencies and policy supplied by a Session Controller unit harness. */
@@ -273,5 +276,6 @@ export function createSessionTestRemote(
     ),
     follow: (request, signal = new AbortController().signal) => direct.follow(request, signal),
     control: (signal = new AbortController().signal) => direct.control(signal),
+    observeJob: (request, signal = new AbortController().signal) => direct.observeJob(request, signal),
   }
 }

packages/api/session-controller/tsconfig.host.json

@@ -18,6 +18,7 @@
     "src/history.ts",
     "src/list.ts",
     "src/model-selection-projection.ts",
+    "src/observe-job.ts",
     "src/skill-catalog.ts"
   ],
   "references": [