diff --git a/packages/jobs/jobs/package.json b/packages/jobs/jobs/package.json
index 34b32d40e7..dfaba10771 100644
--- a/packages/jobs/jobs/package.json
+++ b/packages/jobs/jobs/package.json
@@ -26,6 +26,10 @@
       "types": "./lib/types/brand.d.ts",
       "default": "./lib/types/brand.js"
     },
+    "./view": {
+      "types": "./lib/types/view.d.ts",
+      "default": "./lib/types/view.js"
+    },
     "./src/*": "./src/*",
     "./package.json": "./package.json"
   },
diff --git a/packages/jobs/jobs/src/index.ts b/packages/jobs/jobs/src/index.ts
index 4b6d6d9d43..a1c60cbeef 100644
--- a/packages/jobs/jobs/src/index.ts
+++ b/packages/jobs/jobs/src/index.ts
@@ -1,45 +1,43 @@
 /**
  * The background-job Service Definition (`ctx.jobs`). It owns the contract for
- * job ids, session-scoped access, lifecycle state, completion listeners,
- * owner cleanup, and the optional per-job observation record — an append-only
- * bounded output stream any number of independent observers read at absolute
- * byte offsets, invisible to the model-facing consuming cursor — while
- * producers retain their execution resources. The process-local registry
- * lives in `@deepseek-ai/dsh-jobs-local`.
+ * job ids, session-scoped access, lifecycle state, the per-job output ring —
+ * one bounded stream that the model consumes through a registry-kept cursor
+ * and that any number of observers read at absolute byte offsets — and the
+ * event stream announcing every commit, while producers retain their
+ * execution resources. The process-local registry lives in
+ * `@deepseek-ai/dsh-jobs-local`.
  * @module @deepseek-ai/dsh-jobs
  */
 
 import { Context, Service } from '@deepseek-ai/cordis'
-import type { Agent } from '@deepseek-ai/dsh-agent'
-import type {
-  JobDoneListener, JobId, JobOutputListener, JobRead, JobRecordRead, JobSnapshot, JobStart,
-  JobsChangedListener,
-} from './types.ts'
+import type { SessionId } from '@deepseek-ai/dsh-session'
+import type { JobEvents, JobId, JobSpec, VisibleJobs } from './types.ts'
 
 export { JobId } from './types.ts'
 export type {
   JobAppendOptions,
   JobChannel,
-  JobDoneListener,
+  JobChunk,
+  JobEvent,
+  JobEventFilter,
+  JobEventListener,
+  JobEvents,
+  JobHandle,
   JobHooks,
+  JobKillOptions,
   JobKind,
   JobKindMap,
   JobOutcome,
-  JobOutputListener,
+  JobOutputRead,
+  JobOutputSource,
   JobRead,
-  JobRecordChunk,
-  JobRecordRead,
-  JobSnapshot,
-  JobStart,
+  JobSettleCause,
+  JobSourceRead,
+  JobSpec,
   JobStatus,
-  JobsChangedListener,
-  PlainJobStart,
-  RecordingJob,
-  RecordingJobStart,
-  RunningJob,
+  JobView,
+  VisibleJobs,
 } from './types.ts'
-export { pumpJobOutput } from './pump.ts'
-export type { JobPumpOptions, JobPumpRead, JobPumpSource } from './pump.ts'
 
 declare module '@deepseek-ai/cordis' {
   interface Context {
@@ -48,7 +46,7 @@ declare module '@deepseek-ai/cordis' {
 }
 
 /**
- * Abstract background job registry. Subclass, implement the abstract methods,
+ * Abstract background job registry. Subclass, implement the abstract members,
  * and load the subclass as a plugin — it registers as `ctx.jobs` (one
  * implementation per context; loading a second throws, which is cordis'
  * standard duplicate-service behavior).
@@ -56,32 +54,30 @@ declare module '@deepseek-ai/cordis' {
  * Implementations must honor these semantics:
  * - Registrations outlive producer and controller fibers. Owner and
  *   service disposal cancel live work and await compliant producers; a
- *   throwing teardown cancel force-fails only the record. Teardown
- *   cancellation also marks the record reported, because a record its owner
- *   is being destroyed for has no reader left.
+ *   throwing teardown cancel force-fails only the record. Such settlements
+ *   announce `cause: 'teardown'`, because a job whose owner is being destroyed
+ *   has no reader left.
  * - Owned-job access is fenced by the owner's session id. Ids are
  *   predictable, so authorization — not secrecy — is the boundary.
- * - Settlement is first-wins: one terminal record, released waiters, and one
- *   round of contained listener notification, even against a late producer
- *   outcome. Completion is announced last, after the record is committed and
- *   every other observer of the settlement has seen it, because a reporter
- *   may open a model turn synchronously.
+ * - Settlement is first-wins: one terminal record, released waiters, then one
+ *   round of contained event delivery, even against a late producer outcome.
+ *   The `settled` event follows every released waiter, so a consumer that
+ *   claims a settlement while waiting always claims before the event.
  * - {@link start} refuses work while no attached job controller serves the
  *   spec's owner, so a producer cannot start work that owner cannot collect
  *   or stop. One registry serves every composition in the process, so this
- *   question — and completion-listener delivery — is owner-relative rather
- *   than process-wide: registrations made from an unscoped context serve
- *   every owner, and registrations made under an agent composition's scope
- *   serve exactly the agents composed under it.
- * - Record observation never consumes. Any number of {@link readRecord}
- *   readers hold their own absolute byte offsets; a record read changes no
- *   cursor and no notice state, so the model-facing surfaces (the consuming
- *   {@link read}, completion notices) are unaffected.
- * - Record retention is bounded. Appends past the live cap drop the oldest
+ *   question — and event delivery under `{ owners: 'scope' }` — is
+ *   owner-relative rather than process-wide: registrations made from an
+ *   unscoped context serve every owner, and registrations made under an agent
+ *   composition's scope serve exactly the agents composed under it.
+ * - Every job owns one output ring. Pull sources named by the spec are pumped
+ *   by the registry and drained once more before settlement; pushed appends
+ *   land whole. The model's consuming cursor and observers' absolute offsets
+ *   read the same bytes and never disturb each other.
+ * - Ring retention is bounded. Appends past the live cap drop the oldest
  *   retained bytes; a reader below the retained window gets a lossy read,
- *   never an error. Settlement — the producer outcome, a kill, or teardown —
- *   trims retention to the settled cap and ends the stream; the record has no
- *   separate lifecycle.
+ *   never an error. Settlement trims retention to the settled cap and ends
+ *   the stream; the ring has no separate lifecycle.
  */
 export abstract class JobRegistry extends Service {
   constructor(ctx: Context) {
@@ -94,125 +90,28 @@ export abstract class JobRegistry extends Service {
     super(ctx, 'jobs')
   }
 
+  /** Lifecycle and output events, filtered per subscription. */
+  abstract readonly events: JobEvents
+
   /**
    * Preflight access, validation, owner cleanup, and implementation-owned
    * admission before starting and atomically registering work. Any preflight
    * rejection leaves no job id or execution resource. A throwing starter
    * leaves nothing registered; after it returns, registration cannot fail.
-   * Settlement records the outcome, notifies listeners, and releases waiters.
-   * @param spec - job identity, owner, and synchronous starter.
+   * @param spec - job identity, owner, output sources, and synchronous starter.
    * @returns the registry-issued `<kind>-N` id.
    */
-  abstract start(spec: JobStart): JobId
+  abstract start(spec: JobSpec): JobId
 
   /**
-   * List caller-owned and unowned jobs in registration order without exposing
-   * another session's labels.
-   * @param caller - reading agent; a non-agent caller sees only unowned jobs.
-   * @returns fresh snapshots.
+   * Bind the caller's identity once and return its operations. A caller sees
+   * its own jobs and every unowned job; a caller-less view sees unowned jobs
+   * only. The view resolves visibility on every call, so it stays valid as
+   * jobs come and go.
+   * @param caller - the reading session, or undefined for an anonymous caller.
+   * @returns the operations available to that caller.
    */
-  abstract list(caller?: Agent): JobSnapshot[]
-
-  /**
-   * Return a non-consuming snapshot without changing its read cursor or notice
-   * state. Throws for an unknown or foreign job.
-   * @param id - job to look up.
-   * @param caller - reading agent checked against the owner.
-   * @returns a fresh snapshot.
-   */
-  abstract get(id: JobId, caller?: Agent): JobSnapshot
-
-  /**
-   * Read the next stream delta, or the idempotent final output after settlement.
-   * A terminal read marks the job reported. Throws for an unknown or foreign
-   * job.
-   * @param id - job to read.
-   * @param caller - reading agent checked against the owner.
-   * @returns output text and the post-read snapshot.
-   */
-  abstract read(id: JobId, caller?: Agent): JobRead
-
-  /**
-   * Request cancellation, then mark the job stopping and reported. A producer
-   * throw propagates without changing job state. Throws for an unknown or
-   * foreign job.
-   * @param id - job to cancel.
-   * @param caller - killing agent checked against the owner.
-   * @param reason - logged reason forwarded to the producer.
-   * @returns `requested` for live work, otherwise `already-finished`.
-   */
-  abstract kill(id: JobId, caller?: Agent, reason?: string): 'requested' | 'already-finished'
-
-  /**
-   * Wait for settlement or timeout without cancelling the job. Caller abort
-   * rejects only while the job is live; after settlement the terminal
-   * snapshot wins so a notice suppressed for this waiter is still delivered.
-   * Throws for invalid, unknown, or foreign input.
-   * @param id - job to wait for.
-   * @param timeoutMs - positive finite wait bound in milliseconds.
-   * @param caller - waiting agent checked against the owner.
-   * @param signal - optional cancellation of the wait itself.
-   * @returns snapshot at settlement or timeout.
-   */
-  abstract wait(id: JobId, timeoutMs: number, caller?: Agent, signal?: AbortSignal): Promise<JobSnapshot>
-
-  /**
-   * Register an effect-scoped completion listener. It receives the settlements
-   * of the owners its registering context's scope covers; each listener is
-   * contained; returned promises are observed but not awaited. No listener runs
-   * after service disposal.
-   * @param listener - receives each terminal snapshot and its exact owner.
-   * @returns disposer that unregisters the listener.
-   */
-  abstract onJobDone(listener: JobDoneListener): () => void
-
-  /**
-   * Register an effect-scoped observer of visible-set changes. It fires after
-   * every commit that changes what {@link list} returns for that owner —
-   * registration, every stopping transition (including the one teardown
-   * performs before it awaits a slow producer), settlement, owner-disposal
-   * removal, and the emptying that service disposal commits — so an observer
-   * re-reads rather than accumulating deltas.
-   *
-   * Delivery is owner-relative on the same terms as {@link onJobDone}: an
-   * observer registered from an unscoped context — a host composition's own
-   * carrier — sees every owner, while one registered under an agent
-   * composition's scope sees exactly the agents composed under it.
-   *
-   * This is not a superset of {@link onJobDone}: that one delivers the terminal
-   * record under first-wins semantics a job controller couples to notice
-   * delivery, while this one carries no delivery meaning and marks nothing
-   * reported. Listeners are contained and never awaited.
-   * @param listener - receives the owner whose visible set changed, or
-   *   `undefined` when an unowned job changed and every caller's set did.
-   * @returns disposer that unregisters the listener.
-   */
-  abstract onJobsChanged(listener: JobsChangedListener): () => void
-
-  /**
-   * Read retained record output from an absolute byte offset without consuming
-   * it. Resume by passing a previous read's `next`; a foreign offset inside a
-   * retained chunk returns that whole chunk (its `at` may precede `from`).
-   * Never marks the job reported. Throws for an unknown or foreign job, a job
-   * without a {@link JobStart.record} declaration, or a negative or
-   * non-integer offset.
-   * @param id - job to read.
-   * @param from - absolute byte offset to read from (0 for the retained head).
-   * @param caller - reading agent checked against the owner.
-   * @returns retained chunks overlapping `[from, total)`, the resume offset, and the lossy flag.
-   */
-  abstract readRecord(id: JobId, from: number, caller?: Agent): JobRecordRead
-
-  /**
-   * Register an effect-scoped observer of record advancement — one signal per
-   * committed append and one at settlement, carrying only the job id. A
-   * consumer schedules a {@link readRecord} from its own cursor; the registry
-   * never pushes payloads. Delivery is owner-relative on the same terms as
-   * {@link onJobsChanged}. Listeners are contained and never awaited.
-   * @param listener - receives the id whose record advanced.
-   * @returns disposer that unregisters the listener.
-   */
-  abstract onOutput(listener: JobOutputListener): () => void
+  abstract visibleTo(caller?: SessionId): VisibleJobs
 
   /**
    * Attach an effect-scoped controller that can read and stop jobs. It serves the
diff --git a/packages/jobs/jobs/src/invariant.ts b/packages/jobs/jobs/src/invariant.ts
index 36aca11fd8..84e3a22fe2 100644
--- a/packages/jobs/jobs/src/invariant.ts
+++ b/packages/jobs/jobs/src/invariant.ts
@@ -1,9 +1,8 @@
-/** Package-owned background-job snapshot invariants. @module @deepseek-ai/dsh-jobs/invariant */
+/** Package-owned background-job projection invariants. @module @deepseek-ai/dsh-jobs/invariant */
 
 import type { Context } from '@deepseek-ai/cordis'
-import type { Agent } from '@deepseek-ai/dsh-agent'
 import type { InvariantFailure, InvariantInstaller } from '@deepseek-ai/dsh-invariants'
-import type { JobSnapshot } from './types.ts'
+import type { JobView } from './types.ts'
 
 const PACKAGE_NAME = '@deepseek-ai/dsh-jobs'
 const TERMINAL_STATUSES = new Set(['completed', 'killed', 'failed'])
@@ -13,48 +12,44 @@ export const name = 'jobs-invariant'
 /** Service required before the companion can reserve package ownership. */
 export const inject = ['invariants']
 
-/** Validate the cross-field relationships in one registry snapshot. */
-function validateSnapshot(snapshot: JobSnapshot, owner: Agent | undefined, fail: InvariantFailure): void {
-  const id = String(snapshot.id)
-  const prefix = `${snapshot.kind}-`
+/** Validate the cross-field relationships in one registry projection. */
+function validateView(job: JobView, fail: InvariantFailure): void {
+  const id = String(job.id)
+  const prefix = `${job.kind}-`
   const ordinal = Number(id.slice(prefix.length))
-  if (snapshot.kind.length === 0 || !id.startsWith(prefix)
+  if (job.kind.length === 0 || !id.startsWith(prefix)
     || !Number.isSafeInteger(ordinal) || ordinal < 1) {
-    fail(`job snapshot id ${JSON.stringify(id)} must be ${JSON.stringify(prefix)} followed by a positive ordinal`)
+    fail(`job view id ${JSON.stringify(id)} must be ${JSON.stringify(prefix)} followed by a positive ordinal`)
   }
-  if (snapshot.label.length === 0) fail(`job ${JSON.stringify(id)} label must be non-empty`)
-  if (!Number.isSafeInteger(snapshot.startedAt) || snapshot.startedAt < 0) {
+  if (job.label.length === 0) fail(`job ${JSON.stringify(id)} label must be non-empty`)
+  if (!Number.isSafeInteger(job.startedAt) || job.startedAt < 0) {
     fail(`job ${JSON.stringify(id)} startedAt must be a non-negative epoch integer`)
   }
 
-  const terminal = TERMINAL_STATUSES.has(snapshot.status)
-  if (terminal !== (snapshot.finishedAt !== undefined)) {
+  const terminal = TERMINAL_STATUSES.has(job.status)
+  if (terminal !== (job.finishedAt !== undefined)) {
     fail(`job ${JSON.stringify(id)} finishedAt must be present exactly for a terminal status`)
   }
-  if (snapshot.finishedAt !== undefined
-    && (!Number.isSafeInteger(snapshot.finishedAt) || snapshot.finishedAt < snapshot.startedAt)) {
+  if (job.finishedAt !== undefined
+    && (!Number.isSafeInteger(job.finishedAt) || job.finishedAt < job.startedAt)) {
     fail(`job ${JSON.stringify(id)} finishedAt must be an epoch integer no earlier than startedAt`)
   }
-
-  const expectedOwner = owner?.id
-  if (snapshot.ownerSession !== expectedOwner) {
-    fail(`job ${JSON.stringify(id)} ownerSession does not match its completion owner`)
+  if (terminal && job.progress !== undefined) {
+    fail(`job ${JSON.stringify(id)} progress must be cleared once settled`)
   }
 
-  const { outputTotal, outputEarliest } = snapshot
-  if ((outputTotal !== undefined) !== (outputEarliest !== undefined)) {
-    fail(`job ${JSON.stringify(id)} record offsets must be present together`)
-  } else if (outputTotal !== undefined && outputEarliest !== undefined
-    && (!Number.isSafeInteger(outputTotal) || !Number.isSafeInteger(outputEarliest)
-      || outputEarliest < 0 || outputEarliest > outputTotal)) {
-    fail(`job ${JSON.stringify(id)} record offsets must satisfy 0 <= outputEarliest <= outputTotal`)
+  const { total, earliest } = job.output
+  if (!Number.isSafeInteger(total) || !Number.isSafeInteger(earliest) || earliest < 0 || earliest > total) {
+    fail(`job ${JSON.stringify(id)} output offsets must satisfy 0 <= earliest <= total`)
   }
 }
 
-/** Install checks over current unowned records and every terminal snapshot. */
+/** Install checks over the current unowned projections and every settlement. */
 const install: InvariantInstaller = Object.assign((ctx: Context, fail: InvariantFailure) => {
-  for (const snapshot of ctx.jobs.list()) validateSnapshot(snapshot, undefined, fail)
-  ctx.jobs.onJobDone((snapshot, owner) => { validateSnapshot(snapshot, owner, fail) })
+  for (const job of ctx.jobs.visibleTo().list()) validateView(job, fail)
+  ctx.jobs.events.subscribe({ owners: 'all' }, (event) => {
+    if (event.type === 'settled') validateView(event.job, fail)
+  })
 }, { inject: ['jobs'] })
 
 /**
diff --git a/packages/jobs/jobs/src/pump.ts b/packages/jobs/jobs/src/pump.ts
deleted file mode 100644
index a3c1956019..0000000000
--- a/packages/jobs/jobs/src/pump.ts
+++ /dev/null
@@ -1,122 +0,0 @@
-/**
- * Shared poll pump for producers whose output substrate is a non-consuming
- * offset reader (the subprocess `readFrom(fromByte)` family). Pure utility —
- * no cordis, no timers retained past settlement.
- * @module
- */
-
-import type { JobChannel, RecordingJob } from './types.ts'
-
-/** One non-consuming offset read from a producer-owned stream. */
-export interface JobPumpRead {
-  /** Text captured since the requested offset (the whole retained tail when lossy). */
-  text: string
-  /** Whole-stream offset to resume from on the next read. */
-  nextOffset: number
-  /** True when the requested offset slid out of the source's retained window. */
-  lossy: boolean
-}
-
-/**
- * One producer-owned stream the pump copies into a job record.
- *
- * The source is a pull-by-offset snapshot reader, not a Node.js `Readable`,
- * because the substrate it reads (the subprocess `readFrom` family) is a
- * retained ring that several readers share: the model-facing consuming
- * cursor, the foreground result collector, and this record pump each hold
- * their own absolute offset and read without disturbing the others. A
- * `Readable` is single-consumer and consuming, so wrapping the ring in one
- * would need a tee per reader and could not express `lossy` — the reader's
- * offset having slid out of the retained window — which the record turns
- * into a visible `gapBefore` instead of a silent splice.
- */
-export interface JobPumpSource {
-  /** Stream label attached to every chunk this source yields. */
-  channel?: JobChannel
-  /**
-   * Read everything captured since `fromByte` without consuming it.
-   * @param fromByte - whole-stream offset to resume from (a prior read's `nextOffset`; 0 first).
-   * @returns the delta text, the next offset, and the lossy flag.
-   */
-  read(fromByte: number): JobPumpRead
-}
-
-/** Cadence and settlement for one {@link pumpJobOutput} run. */
-export interface JobPumpOptions {
-  /** Poll interval in milliseconds; a positive finite number. */
-  pollMs: number
-  /**
-   * Settles when the producer's work finished and its sources hold their
-   * final bytes. The pump drains once more after settlement and returns;
-   * a rejection is treated as settlement.
-   */
-  done: Promise<unknown>
-}
-
-/**
- * Copy producer streams into a job record at a bounded cadence: drain every
- * source, sleep `pollMs` or until `done` settles, repeat, then drain one
- * final time. A lossy source read appends its surviving tail with
- * `gapBefore`, so the discontinuity stays visible to observers. Fold the
- * returned promise into the producer's `JobHooks.done` chain so the final
- * drain lands before settlement trims and closes the record.
- *
- * Each round drains the sources in array order, so bytes two sources
- * produced inside one poll window land in that order, not in the order they
- * were written; the record is a best-effort live view whose cross-source
- * reordering is bounded by `pollMs`, not a terminal transcript.
- *
- * The wait holds constant resources however long the job runs: one
- * subscription on `done` for the whole run and one pending timer at a time.
- * @param job - the running job's producer face receiving the copied chunks.
- * @param sources - producer streams, each pumped at its own offset.
- * @param options - poll cadence and the producer's settlement promise.
- * @returns resolves after the final post-settlement drain.
- */
-export async function pumpJobOutput(
-  job: RecordingJob,
-  sources: readonly JobPumpSource[],
-  options: JobPumpOptions,
-): Promise<void> {
-  if (!Number.isFinite(options.pollMs) || options.pollMs <= 0) {
-    throw new Error(`invalid pump pollMs: expected a positive finite number of milliseconds, got ${JSON.stringify(options.pollMs)}`)
-  }
-  const states = sources.map(source => ({ source, cursor: 0 }))
-  const drain = (): void => {
-    for (const state of states) {
-      const { source } = state
-      const read = source.read(state.cursor)
-      state.cursor = read.nextOffset
-      if (read.text.length === 0) continue
-      if (source.channel === undefined && !read.lossy) {
-        job.append(read.text)
-      } else {
-        job.append(read.text, {
-          ...source.channel !== undefined ? { channel: source.channel } : {},
-          ...read.lossy ? { gapBefore: true as const } : {},
-        })
-      }
-    }
-  }
-  let settled = false
-  let timer: ReturnType<typeof setTimeout> | undefined
-  let wake: (() => void) | undefined
-  const finish = (): void => {
-    settled = true
-    // Clear the pending poll so a finished pump holds no timer for up to one
-    // interval, and release the current wait so the final drain runs at once.
-    if (timer !== undefined) clearTimeout(timer)
-    wake?.()
-  }
-  void options.done.then(finish, finish)
-  while (!settled) {
-    drain()
-    await new Promise<void>((resolve) => {
-      wake = resolve
-      timer = setTimeout(resolve, options.pollMs)
-    })
-    wake = undefined
-    timer = undefined
-  }
-  drain()
-}
diff --git a/packages/jobs/jobs/src/types.ts b/packages/jobs/jobs/src/types.ts
index 3ec3f87357..0f46ff1715 100644
--- a/packages/jobs/jobs/src/types.ts
+++ b/packages/jobs/jobs/src/types.ts
@@ -1,154 +1,98 @@
 /**
- * Types shared by job producers, the registry, and controllers. The
- * service implementation lives in `./index.ts`.
+ * Types shared by job producers, the registry, and its consumers. The
+ * client-safe projection vocabulary lives in `./view.ts`; the service
+ * implementation lives in `./index.ts`.
  * @module @deepseek-ai/dsh-jobs/types
  */
 
-import type { Agent } from '@deepseek-ai/dsh-agent'
 import type { SessionId } from '@deepseek-ai/dsh-session'
 import type { JobId } from './brand.ts'
+import type { JobChannel, JobChunk, JobKind, JobView } from './view.ts'
 
 export { JobId } from './brand.ts'
-
-/**
- * Task lifecycle: `running`, optionally `stopping`, then exactly one terminal
- * status. Producer-specific facts belong in {@link JobSnapshot.detail}.
- */
-export type JobStatus = 'running' | 'stopping' | 'completed' | 'killed' | 'failed'
-
-/**
- * Producer-defined job kinds. Plugins extend this map by declaration merging;
- * the registry treats every value as an opaque id namespace.
- */
-export interface JobKindMap {
-  bash: 'bash'
-  subagent: 'subagent'
-}
-
-/** The merge-extensible union of registered producer kind names. */
-export type JobKind = JobKindMap[keyof JobKindMap]
+export type { JobChannel, JobChunk, JobKind, JobKindMap, JobStatus, JobView } from './view.ts'
 
 /** Terminal result supplied by a producer through {@link JobHooks.done}. */
 export interface JobOutcome {
   /** How the job ended: finished (`completed`), cancelled (`killed`), or broke (`failed`). */
   status: 'completed' | 'killed' | 'failed'
-  /** Kind-specific detail rendered into status lines ('exit code: 3', 'max-tokens'). */
+  /**
+   * Terminal reason rendered into status lines (`exit code: 3`, `max-tokens`).
+   * When the job settles `killed` after a {@link VisibleJobs.kill} with a
+   * reason, the registry appends that reason.
+   */
   detail?: string
-  /** Final output for jobs without `readOutput`; stream jobs leave it unset. */
-  output?: string
+  /**
+   * Return value for jobs whose result is a value rather than a stream (a
+   * workflow's rendered result, a subagent's report). The output ring carries
+   * the stream; this is handed out once by the model's next {@link VisibleJobs.read}.
+   */
+  result?: string
+}
+
+/** One incremental read from a {@link JobOutputSource}. */
+export interface JobSourceRead {
+  /** Text captured since the requested offset (the whole retained tail when lossy). */
+  text: string
+  /** Whole-stream offset to resume from on the next read. */
+  nextOffset: number
+  /** True when the requested offset slid out of the source's retained window. */
+  lossy: boolean
 }
 
 /**
- * Record-stream label. Producers without distinct streams omit it; consumers
- * treat an unrecognized label like an absent one.
+ * A pull source the registry pumps into the job's output ring — the subprocess
+ * `readFrom` family. The registry owns the cadence and drains every source one
+ * last time before settlement closes the ring, so a producer folds nothing
+ * into its `done`.
  */
-export type JobChannel = 'stdout' | 'stderr'
+export interface JobOutputSource {
+  /** Stream label attached to every chunk this source yields. */
+  channel?: JobChannel
+  /**
+   * Read everything captured since `fromByte` without consuming it.
+   * @param fromByte - whole-stream offset to resume from (a prior read's `nextOffset`; 0 first).
+   * @returns the delta text, the next offset, and the lossy flag.
+   */
+  read(fromByte: number): JobSourceRead
+}
 
-/** Options for one {@link RunningJob.append}. */
+/** Options for one {@link JobHandle.append}. */
 export interface JobAppendOptions {
   /** Stream label for the chunk; omitted for producers without distinct streams. */
   channel?: JobChannel
   /**
    * Marks that producer-side bytes between the previous chunk and this one
-   * were lost (a source window slid before the producer could copy it), so an
-   * observer can render the discontinuity instead of a silent splice.
+   * were lost, so an observer can render the discontinuity instead of a
+   * silent splice.
    */
   gapBefore?: true
 }
 
-/** Fields shared by every producer declaration passed to {@link JobRegistry.start}. */
-interface JobStartBase {
-  /** Producer kind — also the id prefix (`bash`, `subagent`, …). */
-  kind: JobKind
-  /** One-line model-facing label (the command; the delegation description). */
-  label: string
-  /**
-   * Optional UTF-8 byte cap for each complete model-facing completion notice or
-   * output read, including controller status metadata. Independent of record
-   * retention: it bounds the consuming model surface, never
-   * {@link JobRegistry.readRecord}.
-   */
-  outputLimitBytes?: number
-  /**
-   * Owning live agent. Access is fenced by its session id, and agent disposal
-   * cancels and awaits the job. The instance must be the one currently
-   * registered under its agent id. Omitting the owner creates an unowned job,
-   * open to any caller until service disposal.
-   */
-  owner?: Agent
-}
-
 /**
- * A start with the model-facing surfaces only. Its producer face carries no
- * `append`, so output the registry would have nowhere to keep is
- * unrepresentable rather than dropped.
+ * Producer face of one registered job, handed to {@link JobSpec.run} and
+ * valid for the job's whole life. All methods are synchronous. Writes staged
+ * inside the starter call are retained and become visible with the
+ * registration commit; after settlement — the producer's own outcome, a kill,
+ * or a registry-forced teardown end — writes log and drop instead of
+ * throwing, so a producer's trailing flush cannot break its own teardown path.
  */
-export interface PlainJobStart extends JobStartBase {
-  record?: undefined
-  /**
-   * Start the work after preflight and synchronously return its hooks. Called
-   * once with the job's producer face; a throw leaves nothing registered (the
-   * spent ordinal is skipped), and the producer must clean up any partially
-   * started resources.
-   * @param job - the issued id plus the live-detail writer.
-   */
-  run(job: RunningJob): JobHooks
-}
-
-/**
- * A start that declares an observable output record beside the model-facing
- * surfaces: {@link RecordingJob.append} retains chunks in a bounded ring that
- * any number of observers read at absolute byte offsets through
- * {@link JobRegistry.readRecord}, and snapshots carry
- * {@link JobSnapshot.outputTotal} and {@link JobSnapshot.outputEarliest}.
- */
-export interface RecordingJobStart extends JobStartBase {
-  record: true
-  /**
-   * Start the work after preflight and synchronously return its hooks. Called
-   * once with the job's producer face; a throw leaves nothing registered (the
-   * spent ordinal is skipped), and the producer must clean up any partially
-   * started resources.
-   * @param job - the issued id plus the record append and live-detail writers.
-   */
-  run(job: RecordingJob): JobHooks
-}
-
-/**
- * Producer declaration passed to {@link JobRegistry.start}, discriminated by
- * `record`. The runtime preflights access and cleanup before invoking `run`;
- * the producer owns execution resources while the runtime owns identity and
- * lifecycle state.
- */
-export type JobStart = PlainJobStart | RecordingJobStart
-
-/**
- * Producer face of one registered job, handed to `run` and valid for the
- * job's whole life. All methods are synchronous. Writes staged inside the
- * starter call are retained and become visible with the registration commit;
- * after settlement — the producer's own outcome, a kill, or a registry-forced
- * teardown end — writes log and drop instead of throwing, so a producer's
- * trailing flush cannot break its own teardown path.
- */
-export interface RunningJob {
+export interface JobHandle {
   /** The registry-issued id (`<kind>-N`). */
   readonly id: JobId
   /**
-   * Replace the snapshot's status detail with a live progress line (`3/10`).
-   * @param detail - the new detail line.
-   */
-  updateDetail(detail: string): void
-}
-
-/** Producer face of a {@link RecordingJobStart}: {@link RunningJob} plus the record append. */
-export interface RecordingJob extends RunningJob {
-  /**
-   * Append one record chunk. Offsets advance by the chunk's UTF-8 byte
-   * length; an empty chunk is dropped without waking observers.
+   * Append one chunk to the output ring. Offsets advance by the chunk's UTF-8
+   * byte length; an empty chunk is dropped without waking observers.
    * @param text - the chunk text, exactly as produced.
    * @param options - stream label and gap marker.
    */
   append(text: string, options?: JobAppendOptions): void
+  /**
+   * Replace the live progress line (`3/10`, the current phase). Settlement
+   * clears it; the terminal reason travels in {@link JobOutcome.detail}.
+   * @param line - the new progress line.
+   */
+  updateProgress(line: string): void
 }
 
 /** Hooks through which the runtime controls and observes producer work. */
@@ -162,104 +106,82 @@ export interface JobHooks {
    * Resolves after the producer releases its resources, not merely when work
    * finishes. Must not reject; the runtime converts a rejection to `failed`.
    * If teardown cancellation throws, the runtime may force-fail only the
-   * registry record without claiming that the work stopped. Settlement also
-   * ends the record: fold any record-filling pump into this promise so the
-   * final drain lands before the registry trims and closes the stream.
+   * registry record without claiming that the work stopped.
    */
   done: Promise<JobOutcome>
-  /**
-   * Consume output produced since the previous call. The producer formats
-   * truncation and spill notices. Absence marks a final-output-only job; each
-   * job has one consuming cursor. Independent of the record: this is the
-   * model-facing projection, {@link JobRegistry.readRecord} the observer one.
-   */
-  readOutput?(): string
 }
 
 /**
- * A read-only projection of one job, safe to hand to listeners and tools —
- * a fresh object per call, never live registry state.
+ * Producer declaration passed to {@link JobRegistry.start}. The runtime
+ * preflights access and cleanup before invoking {@link run}; the producer owns
+ * execution resources while the runtime owns identity, lifecycle state, and
+ * the output ring.
  */
-export interface JobSnapshot {
-  /** The registry-issued id (`<kind>-N`). */
-  id: JobId
-  /** The producer kind the job was registered with. */
+export interface JobSpec {
+  /** Producer kind — also the id prefix (`bash`, `subagent`, …). */
   kind: JobKind
-  /** The producer-supplied one-line label. */
+  /** One-line model-facing label (the command; the delegation description). */
   label: string
-  /** Producer-owned cap for complete model-facing notices and output reads. */
+  /**
+   * Owning session. Access is fenced by it, and the owner's live Agent must be
+   * the one currently registered under that id: its disposal cancels and
+   * awaits the job. Omitting the owner creates an unowned job, open to any
+   * caller until service disposal.
+   */
+  owner?: SessionId
+  /**
+   * Optional UTF-8 byte cap for each complete model-facing completion notice or
+   * output read, including controller status metadata. Independent of ring
+   * retention: it bounds the consuming model surface, never observers.
+   */
   outputLimitBytes?: number
   /**
-   * Owner session id used for authorization and correlation; absent for
-   * unowned jobs. Completion listeners receive the exact {@link Agent}
-   * separately through {@link JobDoneListener}.
+   * Pull sources the registry pumps into the ring at its own cadence.
+   * Producers that narrate their own progress use {@link JobHandle.append}
+   * instead; a job may use both.
    */
-  ownerSession?: SessionId
-  /** Current lifecycle state. */
-  status: JobStatus
+  output?: readonly JobOutputSource[]
   /**
-   * Kind-specific status detail: live progress while the producer updates it
-   * through {@link RunningJob.updateDetail}, the terminal detail once settled.
+   * Start the work after preflight and synchronously return its hooks. Called
+   * once with the job's producer face; a throw leaves nothing registered (the
+   * spent ordinal is skipped), and the producer must clean up any partially
+   * started resources.
+   * @param job - the issued id plus the ring append and progress writers.
    */
-  detail?: string
-  /** Epoch ms when the job was registered. */
-  startedAt: number
-  /** Epoch ms when the job settled; absent while `running`/`stopping`. */
-  finishedAt?: number
-  /**
-   * True when a kill, read, wait, or teardown cancel has reported or committed
-   * to report the terminal state. Completion reporters suppress redundant
-   * notices when set. Teardown claims it because the owner or service being
-   * destroyed leaves no reader: a reporter that opens a turn on notice would
-   * otherwise spend a model request per teardown layer.
-   */
-  reported: boolean
-  /**
-   * Total UTF-8 bytes ever appended to the record — the offset the next chunk
-   * starts at. Present exactly when the job declared {@link JobStart.record}.
-   */
-  outputTotal?: number
-  /**
-   * Offset of the oldest retained record byte. Greater than zero exactly when
-   * retention dropped the head; a reader starting below it gets a lossy read.
-   * Present exactly when the job declared {@link JobStart.record}.
-   */
-  outputEarliest?: number
+  run(job: JobHandle): JobHooks
 }
 
-/** Output and post-read state returned by {@link JobRegistry.read}. */
+/** Options for {@link VisibleJobs.kill}. */
+export interface JobKillOptions {
+  /**
+   * Cancellation reason: forwarded verbatim to the producer's cancel hook, and
+   * merged into the terminal `detail` when the job settles `killed`, so both
+   * the model and observers can see why the work stopped.
+   */
+  reason?: string
+}
+
+/** Output and post-read state returned by the consuming {@link VisibleJobs.read}. */
 export interface JobRead {
-  /**
-   * Stream kinds: the consuming delta since the previous read. Final-output
-   * kinds: empty while live, the terminal {@link JobOutcome.output} (or
-   * empty) once settled — idempotent, never consumed.
-   */
-  text: string
+  /** Ring chunks appended since the model cursor, in offset order; every channel included. */
+  chunks: readonly JobChunk[]
+  /** True when the cursor fell below the oldest retained byte, so bytes are missing before `chunks`. */
+  lossy: boolean
+  /** The producer's {@link JobOutcome.result}, handed out by the first read after settlement only. */
+  result?: string
   /** The job's state at read time. */
-  snapshot: JobSnapshot
+  job: JobView
 }
 
-/** One retained record chunk returned by {@link JobRegistry.readRecord}. */
-export interface JobRecordChunk {
-  /** Absolute offset of the chunk's first byte. */
-  at: number
-  /** Chunk text exactly as appended (possibly tail-trimmed by retention). */
-  text: string
-  /** Stream label, when the producer supplied one. */
-  channel?: JobChannel
-  /** Producer-reported loss immediately before this chunk. */
-  gapBefore?: true
-}
-
-/** Result of one non-consuming {@link JobRegistry.readRecord}. */
-export interface JobRecordRead {
+/** Result of one non-consuming {@link VisibleJobs.readAt}. */
+export interface JobOutputRead {
   /** Retained chunks overlapping `[from, total)`, in offset order. */
-  chunks: readonly JobRecordChunk[]
+  chunks: readonly JobChunk[]
   /**
-   * Offset to resume from — the record's current `outputTotal`. Always a
-   * chunk boundary: appends land whole and trimming only advances chunk
-   * starts, and consumers concatenate `chunks` under that assumption, so a
-   * provider serving partial chunks would silently duplicate text.
+   * Offset to resume from — the ring's current `total`. Always a chunk
+   * boundary: appends land whole and trimming only advances chunk starts, and
+   * consumers concatenate `chunks` under that assumption, so a provider
+   * serving partial chunks would silently duplicate text.
    */
   next: number
   /** True when `from` fell below the oldest retained byte, so bytes are missing before `chunks`. */
@@ -267,28 +189,118 @@ export interface JobRecordRead {
 }
 
 /**
- * Completion callback with the exact owner supplied at start, or `undefined`
- * for an unowned job. Returned promises are observed but not awaited.
+ * The operations one caller may perform, bound by {@link JobRegistry.visibleTo}.
+ * A caller sees its own jobs and every unowned job; every method throws for
+ * an id outside that set, without distinguishing unknown from foreign.
  */
-export type JobDoneListener = (
-  snapshot: JobSnapshot,
-  owner: Agent | undefined,
-) => void | PromiseLike<void>
+export interface VisibleJobs {
+  /**
+   * List the visible jobs in registration order.
+   * @returns fresh projections.
+   */
+  list(): JobView[]
+  /**
+   * Project one job without touching its cursor.
+   * @param id - job to look up.
+   * @returns a fresh projection.
+   */
+  get(id: JobId): JobView
+  /**
+   * Consume the ring from the model cursor and advance it to the current
+   * total. After settlement the first read also carries the producer's
+   * result; later reads return only what was appended since.
+   * @param id - job to read.
+   * @returns the chunks since the cursor, the lossy flag, the result once, and the post-read projection.
+   */
+  read(id: JobId): JobRead
+  /**
+   * Read retained ring output from an absolute byte offset without consuming
+   * it. Resume by passing a previous read's `next`; an offset inside a
+   * retained chunk returns that whole chunk (its `at` may precede `from`).
+   * Never moves the model cursor. Throws for a negative or non-integer offset.
+   * @param id - job to read.
+   * @param from - absolute byte offset to read from (0 for the retained head).
+   * @returns retained chunks overlapping `[from, total)`, the resume offset, and the lossy flag.
+   */
+  readAt(id: JobId, from: number): JobOutputRead
+  /**
+   * Request cancellation, then mark the job stopping. A recorded
+   * {@link JobKillOptions.reason} merges into the terminal `detail` when the
+   * job settles `killed`. A producer throw propagates without changing job
+   * state.
+   * @param id - job to cancel.
+   * @param options - cancellation reason.
+   * @returns `requested` for live work, otherwise `already-finished`.
+   */
+  kill(id: JobId, options?: JobKillOptions): 'requested' | 'already-finished'
+  /**
+   * Wait for settlement or timeout without cancelling the job. Caller abort
+   * rejects only while the job is live; after settlement the terminal
+   * projection wins. Throws for an invalid timeout.
+   * @param id - job to wait for.
+   * @param timeoutMs - positive finite wait bound in milliseconds.
+   * @param signal - optional cancellation of the wait itself.
+   * @returns projection at settlement or timeout.
+   */
+  wait(id: JobId, timeoutMs: number, signal?: AbortSignal): Promise<JobView>
+}
 
 /**
- * Observation callback for a change to what one owner's {@link JobRegistry.list}
- * would return. It is owner-granular rather than job-granular because the
- * change may be a removal, which no per-job record can express, and because
- * its consumers re-read the whole visible set anyway.
- *
- * An `undefined` owner means an unowned job changed, so every caller's visible
- * set changed with it.
+ * Why a job settled: its producer finished (`producer`), a
+ * {@link VisibleJobs.kill} ran first (`kill`), or owner or service teardown
+ * cancelled it (`teardown`) — a settlement whose owner has no reader left.
  */
-export type JobsChangedListener = (owner: Agent | undefined) => void
+export type JobSettleCause = 'producer' | 'kill' | 'teardown'
 
 /**
- * Observer of record advancement for one job: new appended output, or the
- * settlement that ends the stream. Carries no payload — a consumer schedules a
- * {@link JobRegistry.readRecord} from its own cursor.
+ * One lifecycle or output event. Lifecycle events carry the job's projection
+ * after the commit they announce; `output` carries only the id and the new
+ * total, so an observer schedules a {@link VisibleJobs.readAt} from its own
+ * cursor and the registry never pushes payloads.
  */
-export type JobOutputListener = (id: JobId) => void
+export type JobEvent =
+  | {
+    /** Registration commit, progress line change, stopping transition, or removal from the visible set. */
+    readonly type: 'registered' | 'progress' | 'stopping' | 'removed'
+    readonly job: JobView
+  }
+  | {
+    readonly type: 'settled'
+    readonly job: JobView
+    readonly cause: JobSettleCause
+  }
+  | {
+    readonly type: 'output'
+    readonly id: JobId
+    /** Owning session, absent for an unowned job. */
+    readonly owner?: SessionId
+    /** The ring's total after the append (or at settlement, which ends the stream). */
+    readonly total: number
+  }
+
+/**
+ * Who a subscription hears about. `{ owner }` delivers that session's jobs
+ * plus every unowned job (the set that session can see). `{ owners: 'scope' }`
+ * delivers the owners composed under the subscribing context — one registry
+ * serves every composition in the process, and a mount under one preset must
+ * not hear another preset's agents. `{ owners: 'all' }` delivers everything.
+ */
+export type JobEventFilter =
+  | { readonly owner: SessionId }
+  | { readonly owners: 'all' | 'scope' }
+
+/** Event callback; contained by the registry, never awaited. */
+export type JobEventListener = (event: JobEvent) => void
+
+/** The registry's event stream. */
+export interface JobEvents {
+  /**
+   * Register an effect-scoped listener. Events are dispatched synchronously
+   * after the commit they announce; a `settled` event follows the release of
+   * every waiter. No listener runs after service disposal.
+   * @param filter - which owners' events to deliver.
+   * @param listener - receives each matching event.
+   * @returns disposer that unregisters the listener.
+   */
+  subscribe(filter: JobEventFilter, listener: JobEventListener): () => void
+}
diff --git a/packages/jobs/jobs/src/view.ts b/packages/jobs/jobs/src/view.ts
new file mode 100644
index 0000000000..033fcfbe76
--- /dev/null
+++ b/packages/jobs/jobs/src/view.ts
@@ -0,0 +1,81 @@
+/**
+ * Client-safe job vocabulary: the read-only projection of one job and the
+ * chunks its output ring hands out. This leaf reaches no Host package, so
+ * browser programs and Remote wire types import it without pulling the
+ * registry's Host declaration merges.
+ * @module @deepseek-ai/dsh-jobs/view
+ */
+
+import type { SessionId } from '@deepseek-ai/dsh-session/types'
+import type { JobId } from './brand.ts'
+
+/**
+ * Job lifecycle: `running`, optionally `stopping`, then exactly one terminal
+ * status. Producer-specific facts belong in {@link JobView.progress} and
+ * {@link JobView.detail}.
+ */
+export type JobStatus = 'running' | 'stopping' | 'completed' | 'killed' | 'failed'
+
+/**
+ * Producer-defined job kinds. Plugins extend this map by declaration merging;
+ * the registry treats every value as an opaque id namespace.
+ */
+export interface JobKindMap {
+  bash: 'bash'
+  subagent: 'subagent'
+}
+
+/** The merge-extensible union of registered producer kind names. */
+export type JobKind = JobKindMap[keyof JobKindMap]
+
+/**
+ * Stream label of one output chunk. `stdout` and `stderr` reach the model's
+ * consuming read; `log` marks producer narration (a workflow's phase lines)
+ * that only observers see. Consumers treat an unrecognized label like an
+ * absent one.
+ */
+export type JobChannel = 'stdout' | 'stderr' | 'log'
+
+/** One chunk of a job's output ring: absolute offset, text, and its provenance. */
+export interface JobChunk {
+  /** Absolute offset of the chunk's first byte; offsets never move once assigned. */
+  readonly at: number
+  /** Chunk text exactly as appended (possibly tail-trimmed by retention). */
+  readonly text: string
+  /** Stream label, when the producer supplied one. */
+  readonly channel?: JobChannel
+  /** Bytes immediately before this chunk were lost, at the producer or to retention. */
+  readonly gapBefore?: true
+}
+
+/**
+ * Read-only projection of one job — a fresh object per call, never live
+ * registry state. The model tools, the browser roster, and the observation
+ * stream all consume this one shape.
+ */
+export interface JobView {
+  /** The registry-issued id (`<kind>-N`). */
+  readonly id: JobId
+  /** The producer kind the job was registered with. */
+  readonly kind: JobKind
+  /** The producer-supplied one-line label. */
+  readonly label: string
+  /** Owning session; absent for an unowned job, which every caller can see. */
+  readonly owner?: SessionId
+  /** Current lifecycle state. */
+  readonly status: JobStatus
+  /** The producer's live progress line (`3/10`, the current phase); cleared at settlement. */
+  readonly progress?: string
+  /** Terminal reason (`exit code: 3`); a recorded kill reason is merged in. */
+  readonly detail?: string
+  /** Epoch ms when the job was registered. */
+  readonly startedAt: number
+  /** Epoch ms when the job settled; absent while live. */
+  readonly finishedAt?: number
+  /**
+   * The output ring's absolute coordinates. `total` is the offset the next
+   * chunk starts at (0 while nothing was written); `earliest` is the oldest
+   * retained byte, greater than zero exactly when retention dropped the head.
+   */
+  readonly output: { readonly total: number; readonly earliest: number }
+}
diff --git a/packages/jobs/jobs/tests/invariant.spec.ts b/packages/jobs/jobs/tests/invariant.spec.ts
index fa42eb6c74..12a6d48bee 100644
--- a/packages/jobs/jobs/tests/invariant.spec.ts
+++ b/packages/jobs/jobs/tests/invariant.spec.ts
@@ -1,48 +1,49 @@
 import { describe, expect, it } from 'vitest'
 import { Context } from '@deepseek-ai/cordis'
-import type { Agent } from '@deepseek-ai/dsh-agent'
 import { SessionId } from '@deepseek-ai/dsh-session'
 import JobRegistry, { JobId } from '@deepseek-ai/dsh-jobs'
-import type { JobDoneListener, JobSnapshot } from '@deepseek-ai/dsh-jobs'
+import type { JobEventListener, JobView } from '@deepseek-ai/dsh-jobs'
 import * as JobsInvariant from '@deepseek-ai/dsh-jobs/invariant'
 import InvariantRegistry from '@deepseek-ai/dsh-invariants'
 
-const BASE: JobSnapshot = {
+const BASE: JobView = {
   id: JobId('bash-1'),
   kind: 'bash',
   label: 'compile',
   status: 'completed',
   startedAt: 10,
   finishedAt: 20,
-  reported: false,
+  output: { total: 0, earliest: 0 },
 }
 
-const RUNNING: JobSnapshot = {
+const RUNNING: JobView = {
   id: JobId('bash-1'),
   kind: 'bash',
   label: 'compile',
   status: 'running',
   startedAt: 10,
-  reported: false,
+  output: { total: 0, earliest: 0 },
 }
 
-const TERMINAL_WITHOUT_FINISH: JobSnapshot = {
+const TERMINAL_WITHOUT_FINISH: JobView = {
   id: JobId('bash-1'),
   kind: 'bash',
   label: 'compile',
   status: 'completed',
   startedAt: 10,
-  reported: false,
+  output: { total: 0, earliest: 0 },
 }
 
-async function setup(seed: JobSnapshot[] = []): Promise<(snapshot: unknown, owner?: Agent) => void> {
+async function setup(seed: JobView[] = []): Promise<(job: unknown) => void> {
   const ctx = new Context()
-  let listener: JobDoneListener | undefined
+  let listener: JobEventListener | undefined
   const probe = {
-    list: () => seed,
-    onJobDone(value: JobDoneListener) {
-      listener = value
-      return () => { listener = undefined }
+    visibleTo: () => ({ list: () => seed }),
+    events: {
+      subscribe(_filter: unknown, value: JobEventListener) {
+        listener = value
+        return () => { listener = undefined }
+      },
     },
   } as unknown as JobRegistry
   await ctx.plugin(InvariantRegistry)
@@ -51,44 +52,54 @@ async function setup(seed: JobSnapshot[] = []): Promise<(snapshot: unknown, owne
     apply(child: Context) { child.provide('jobs', probe) },
   })
   await ctx.plugin(JobsInvariant)
-  if (listener === undefined) throw new Error('job invariant did not subscribe to terminal snapshots')
-  return (snapshot, owner) => { listener!(snapshot as JobSnapshot, owner) }
+  if (listener === undefined) throw new Error('job invariant did not subscribe to settlements')
+  return (job) => { listener!({ type: 'settled', job: job as JobView, cause: 'producer' }) }
 }
 
 describe('job-registry invariants', () => {
-  it('accepts coherent current and terminal snapshots', async () => {
+  it('accepts coherent current and terminal projections', async () => {
     const notify = await setup([RUNNING])
     expect(() => { notify(BASE) }).not.toThrow()
-    const owner = { id: SessionId('owner') } as Agent
-    expect(() => { notify({ ...BASE, id: JobId('subagent-2'), kind: 'subagent', ownerSession: owner.id }, owner) })
+    expect(() => { notify({ ...BASE, id: JobId('subagent-2'), kind: 'subagent', owner: SessionId('owner') }) })
       .not.toThrow()
-    expect(() => { notify({ ...BASE, outputTotal: 8, outputEarliest: 4 }) }).not.toThrow()
+    expect(() => { notify({ ...BASE, output: { total: 8, earliest: 4 } }) }).not.toThrow()
+  })
+
+  it('ignores non-settlement events', async () => {
+    const ctx = new Context()
+    let listener: JobEventListener | undefined
+    const probe = {
+      visibleTo: () => ({ list: () => [] }),
+      events: { subscribe(_filter: unknown, value: JobEventListener) { listener = value; return () => {} } },
+    } as unknown as JobRegistry
+    await ctx.plugin(InvariantRegistry)
+    await ctx.plugin({ name: 'job-invariant-probe', apply(child: Context) { child.provide('jobs', probe) } })
+    await ctx.plugin(JobsInvariant)
+    expect(() => { listener!({ type: 'registered', job: { ...BASE, label: '' } }) }).not.toThrow()
   })
 
   it.each([
-    [{ ...BASE, id: JobId('-1'), kind: '' }, undefined, /positive ordinal/],
-    [{ ...BASE, id: JobId('other-1') }, undefined, /must be "bash-" followed by a positive ordinal/],
-    [{ ...BASE, id: JobId('bash-x') }, undefined, /positive ordinal/],
-    [{ ...BASE, id: JobId('bash-0') }, undefined, /positive ordinal/],
-    [{ ...BASE, startedAt: -1 }, undefined, /startedAt must be a non-negative epoch integer/],
-    [{ ...BASE, startedAt: 0.5 }, undefined, /startedAt must be a non-negative epoch integer/],
-    [{ ...BASE, status: 'running' }, undefined, /finishedAt must be present exactly for a terminal status/],
-    [TERMINAL_WITHOUT_FINISH, undefined, /finishedAt must be present exactly for a terminal status/],
-    [{ ...BASE, finishedAt: 9 }, undefined, /no earlier than startedAt/],
-    [{ ...BASE, finishedAt: 20.5 }, undefined, /no earlier than startedAt/],
-    [{ ...BASE, ownerSession: SessionId('recorded') }, { id: SessionId('actual') } as Agent, /does not match its completion owner/],
-    [{ ...BASE, outputTotal: 8 }, undefined, /record offsets must be present together/],
-    [{ ...BASE, outputEarliest: 0 }, undefined, /record offsets must be present together/],
-    [{ ...BASE, outputTotal: 8, outputEarliest: 9 }, undefined, /0 <= outputEarliest <= outputTotal/],
-    [{ ...BASE, outputTotal: 8.5, outputEarliest: 0 }, undefined, /0 <= outputEarliest <= outputTotal/],
-    [{ ...BASE, outputTotal: 8, outputEarliest: 0.5 }, undefined, /0 <= outputEarliest <= outputTotal/],
-    [{ ...BASE, outputTotal: 8, outputEarliest: -1 }, undefined, /0 <= outputEarliest <= outputTotal/],
-  ] as const)('rejects an incoherent registry snapshot', async (snapshot, owner, message) => {
+    [{ ...BASE, id: JobId('-1'), kind: '' }, /positive ordinal/],
+    [{ ...BASE, id: JobId('other-1') }, /must be "bash-" followed by a positive ordinal/],
+    [{ ...BASE, id: JobId('bash-x') }, /positive ordinal/],
+    [{ ...BASE, id: JobId('bash-0') }, /positive ordinal/],
+    [{ ...BASE, startedAt: -1 }, /startedAt must be a non-negative epoch integer/],
+    [{ ...BASE, startedAt: 0.5 }, /startedAt must be a non-negative epoch integer/],
+    [{ ...BASE, status: 'running' }, /finishedAt must be present exactly for a terminal status/],
+    [TERMINAL_WITHOUT_FINISH, /finishedAt must be present exactly for a terminal status/],
+    [{ ...BASE, finishedAt: 9 }, /no earlier than startedAt/],
+    [{ ...BASE, finishedAt: 20.5 }, /no earlier than startedAt/],
+    [{ ...BASE, progress: '3/10' }, /progress must be cleared once settled/],
+    [{ ...BASE, output: { total: 8, earliest: 9 } }, /0 <= earliest <= total/],
+    [{ ...BASE, output: { total: 8.5, earliest: 0 } }, /0 <= earliest <= total/],
+    [{ ...BASE, output: { total: 8, earliest: 0.5 } }, /0 <= earliest <= total/],
+    [{ ...BASE, output: { total: 8, earliest: -1 } }, /0 <= earliest <= total/],
+  ] as const)('rejects an incoherent registry projection', async (job, message) => {
     const notify = await setup()
-    expect(() => { notify(snapshot, owner) }).toThrow(message)
+    expect(() => { notify(job) }).toThrow(message)
   })
 
-  it('rejects an incoherent record already present at installation', async () => {
+  it('rejects an incoherent projection already present at installation', async () => {
     await expect(setup([{ ...BASE, label: '' }])).rejects.toThrow(/label must be non-empty/)
   })
 })
diff --git a/packages/jobs/jobs/tests/service.spec.ts b/packages/jobs/jobs/tests/service.spec.ts
index 331bf6596b..f3b456acbc 100644
--- a/packages/jobs/jobs/tests/service.spec.ts
+++ b/packages/jobs/jobs/tests/service.spec.ts
@@ -1,69 +1,60 @@
-import { describe, expect, it, vi } from 'vitest'
+import { describe, expect, it } from 'vitest'
 import { Context } from '@deepseek-ai/cordis'
-import type { Agent } from '@deepseek-ai/dsh-agent'
-import { JobId, JobRegistry, pumpJobOutput } from '@deepseek-ai/dsh-jobs'
+import { SessionId } from '@deepseek-ai/dsh-session'
+import { JobId, JobRegistry } from '@deepseek-ai/dsh-jobs'
 import type {
-  JobAppendOptions, JobDoneListener, JobHooks, JobOutputListener, JobPumpSource, JobRead, JobRecordRead,
-  JobSnapshot, JobStart, JobsChangedListener, PlainJobStart, RecordingJob, RecordingJobStart,
+  JobEvent, JobEventFilter, JobEventListener, JobHooks, JobOutputSource, JobRead, JobSpec, JobView, VisibleJobs,
 } from '@deepseek-ai/dsh-jobs'
 
+/** Compile-time probe: does `Shape` carry a `Key` member? */
+type HasKey<Shape, Key extends string> = Key extends keyof Shape ? true : false
+
 /**
- * Minimal concrete registry: one canned record. The Service Definition owns the contract
- * only (ids, snapshots, authorization-shaped signatures); the registry
+ * Minimal concrete registry: one canned row. The Service Definition owns the
+ * contract only (ids, the caller-bound view, the event stream); the registry
  * behavior suite lives with `@deepseek-ai/dsh-jobs-local`.
  */
 class StubJobRegistry extends JobRegistry {
-  snapshotOf(id: JobId): JobSnapshot {
+  readonly subscriptions: { filter: JobEventFilter; listener: JobEventListener }[] = []
+  readonly started: JobSpec[] = []
+
+  readonly events = {
+    subscribe: (filter: JobEventFilter, listener: JobEventListener): (() => void) => {
+      const entry = { filter, listener }
+      this.subscriptions.push(entry)
+      return () => { this.subscriptions.splice(this.subscriptions.indexOf(entry), 1) }
+    },
+  }
+
+  private view(id: JobId, owner: SessionId | undefined): JobView {
     return {
       id,
       kind: 'bash',
       label: 'sleep 60',
+      ...owner !== undefined ? { owner } : {},
       status: 'running',
       startedAt: 0,
-      reported: false,
+      output: { total: 0, earliest: 0 },
     }
   }
 
-  start(spec: JobStart): JobId {
+  start(spec: JobSpec): JobId {
+    this.started.push(spec)
     const id = JobId(`${spec.kind}-1`)
-    spec.run({ id, append() {}, updateDetail() {} })
+    spec.run({ id, append() {}, updateProgress() {} })
     return id
   }
 
-  list(): JobSnapshot[] {
-    return [this.snapshotOf(JobId('bash-1'))]
-  }
-
-  get(id: JobId): JobSnapshot {
-    return this.snapshotOf(id)
-  }
-
-  read(id: JobId): JobRead {
-    return { text: '', snapshot: this.snapshotOf(id) }
-  }
-
-  kill(): 'requested' | 'already-finished' {
-    return 'requested'
-  }
-
-  wait(id: JobId, _timeoutMs: number, _caller?: Agent, _signal?: AbortSignal): Promise<JobSnapshot> {
-    return Promise.resolve(this.snapshotOf(id))
-  }
-
-  onJobDone(_listener: JobDoneListener): () => void {
-    return () => {}
-  }
-
-  onJobsChanged(_listener: JobsChangedListener): () => void {
-    return () => {}
-  }
-
-  readRecord(_id: JobId, from: number): JobRecordRead {
-    return { chunks: [], next: from, lossy: false }
-  }
-
-  onOutput(_listener: JobOutputListener): () => void {
-    return () => {}
+  visibleTo(caller?: SessionId): VisibleJobs {
+    const view = (id: JobId): JobView => this.view(id, caller)
+    return {
+      list: () => [view(JobId('bash-1'))],
+      get: id => view(id),
+      read: (id): JobRead => ({ chunks: [], lossy: false, job: view(id) }),
+      readAt: (_id, from) => ({ chunks: [], next: from, lossy: false }),
+      kill: () => 'requested',
+      wait: id => Promise.resolve(view(id)),
+    }
   }
 
   attachController(_name: string): () => void {
@@ -71,63 +62,58 @@ class StubJobRegistry extends JobRegistry {
   }
 }
 
-/** Compile-time probe: does the producer face `Face` carry `append`? */
-type HasAppend<Face> = 'append' extends keyof Face ? true : false
-
-describe('JobStart producer faces', () => {
-  it('hands append only to a start that declares a record', () => {
-    const hooks: JobHooks = { cancel() {}, done: new Promise(() => {}) }
-    // The literal's `record` discriminant selects the face `run` receives.
-    const plain: JobStart = {
-      kind: 'bash',
-      label: 'no record',
-      run: (job) => {
-        job.updateDetail('works for every job')
-        return hooks
-      },
-    }
-    const recording: JobStart = {
-      kind: 'bash',
-      label: 'record',
-      record: true,
-      run: (job) => {
-        job.append('lands in the record')
-        return hooks
-      },
-    }
-    // Negative proof: a plain start's face has no `append` at the type level.
-    const plainFace: HasAppend<Parameters<PlainJobStart['run']>[0]> = false
-    const recordingFace: HasAppend<Parameters<RecordingJobStart['run']>[0]> = true
-    expect(plainFace).toBe(false)
-    expect(recordingFace).toBe(true)
-    expect(plain.record).toBeUndefined()
-    expect(recording.record).toBe(true)
-  })
-})
-
 describe('JobRegistry seam', () => {
   it('a concrete subclass registers as ctx.jobs and serves the abstract API', async () => {
     const ctx = new Context()
     await ctx.plugin(StubJobRegistry)
+    const caller = SessionId('session-1')
+    const hooks: JobHooks = { cancel() {}, done: new Promise(() => {}) }
 
     const detachController = ctx.jobs.attachController('seam-test')
-    const id = ctx.jobs.start({ kind: 'bash', label: 'sleep 60', run: () => ({ cancel() {}, done: new Promise(() => {}) }) })
+    const id = ctx.jobs.start({ kind: 'bash', label: 'sleep 60', owner: caller, run: () => hooks })
     expect(id).toBe('bash-1')
-    expect(ctx.jobs.list()).toHaveLength(1)
-    expect(ctx.jobs.get(id).status).toBe('running')
-    expect(ctx.jobs.read(id).text).toBe('')
-    expect(ctx.jobs.kill(id)).toBe('requested')
-    await expect(ctx.jobs.wait(id, 5)).resolves.toMatchObject({ id })
-    expect(ctx.jobs.readRecord(id, 0)).toEqual({ chunks: [], next: 0, lossy: false })
-    const detachListener = ctx.jobs.onJobDone(() => {})
-    detachListener()
-    const detachChanges = ctx.jobs.onJobsChanged(() => {})
-    detachChanges()
-    const detachOutput = ctx.jobs.onOutput(() => {})
-    detachOutput()
+
+    const jobs = ctx.jobs.visibleTo(caller)
+    expect(jobs.list()).toHaveLength(1)
+    expect(jobs.get(id).status).toBe('running')
+    expect(jobs.get(id).owner).toBe(caller)
+    expect(jobs.read(id)).toEqual({ chunks: [], lossy: false, job: expect.objectContaining({ id }) })
+    expect(jobs.readAt(id, 7)).toEqual({ chunks: [], next: 7, lossy: false })
+    expect(jobs.kill(id, { reason: 'seam test' })).toBe('requested')
+    await expect(jobs.wait(id, 5)).resolves.toMatchObject({ id })
+    // A caller-less view is the unowned-only view.
+    expect(ctx.jobs.visibleTo().get(id).owner).toBeUndefined()
+
+    const events: JobEvent[] = []
+    const unsubscribe = ctx.jobs.events.subscribe({ owners: 'all' }, (event) => { events.push(event) })
+    unsubscribe()
     detachController()
   })
 
+  it('types every start the same way: one handle, pull sources optional', () => {
+    const hooks: JobHooks = { cancel() {}, done: new Promise(() => {}) }
+    const source: JobOutputSource = {
+      channel: 'stdout',
+      read: fromByte => ({ text: '', nextOffset: fromByte, lossy: false }),
+    }
+    const spec: JobSpec = {
+      kind: 'bash',
+      label: 'x',
+      output: [source],
+      run: (job) => {
+        job.append('lands in the ring', { channel: 'log' })
+        job.updateProgress('1/2')
+        return hooks
+      },
+    }
+    expect(spec.output).toHaveLength(1)
+    // Bookkeeping never reaches the public view, and the record declaration is gone.
+    const viewHasReported: HasKey<JobView, 'reported'> = false
+    const specHasRecord: HasKey<JobSpec, 'record'> = false
+    const hooksHaveReadOutput: HasKey<JobHooks, 'readOutput'> = false
+    expect([viewHasReported, specHasRecord, hooksHaveReadOutput]).toEqual([false, false, false])
+  })
+
   it('loading a second implementation throws (one jobs service per context — cordis standard)', async () => {
     const ctx = new Context()
     await ctx.plugin(StubJobRegistry)
@@ -141,119 +127,3 @@ describe('JobRegistry seam', () => {
       .rejects.toThrow(/abstract job registry seam; load an implementation such as @deepseek-ai\/dsh-jobs-local/)
   })
 })
-
-/** Record every append so pump behavior is observable without a registry. */
-function recordingJob(): { job: RecordingJob; appends: { text: string; options?: JobAppendOptions }[] } {
-  const appends: { text: string; options?: JobAppendOptions }[] = []
-  const job: RecordingJob = {
-    id: JobId('bash-1'),
-    append(text, options) { appends.push({ text, ...options !== undefined ? { options } : {} }) },
-    updateDetail() {},
-  }
-  return { job, appends }
-}
-
-/** A scripted source: each read() shifts the next scripted result. */
-function scriptedSource(
-  reads: { text: string; lossy?: boolean }[],
-  channel?: 'stdout' | 'stderr',
-): { source: JobPumpSource; offsets: number[] } {
-  const offsets: number[] = []
-  let offset = 0
-  const source: JobPumpSource = {
-    ...channel !== undefined ? { channel } : {},
-    read(fromByte) {
-      offsets.push(fromByte)
-      const next = reads.shift() ?? { text: '' }
-      offset += Buffer.byteLength(next.text, 'utf8')
-      return { text: next.text, nextOffset: offset, lossy: next.lossy ?? false }
-    },
-  }
-  return { source, offsets }
-}
-
-describe('pumpJobOutput', () => {
-  it('rejects a non-positive poll interval before touching any source', async () => {
-    const { job } = recordingJob()
-    await expect(pumpJobOutput(job, [], { pollMs: 0, done: Promise.resolve() }))
-      .rejects.toThrow(/invalid pump pollMs/)
-  })
-
-  it('copies labeled deltas at the poll cadence and resumes each source at its own offset', async () => {
-    vi.useFakeTimers()
-    try {
-      const { job, appends } = recordingJob()
-      let settle!: () => void
-      const done = new Promise<void>((resolve) => { settle = resolve })
-      const out = scriptedSource([{ text: 'a' }, { text: 'bc' }], 'stdout')
-      const err = scriptedSource([{ text: '' }, { text: 'E' }], 'stderr')
-      const pump = pumpJobOutput(job, [out.source, err.source], { pollMs: 50, done })
-
-      await vi.advanceTimersByTimeAsync(50)
-      settle()
-      await pump
-      expect(appends).toEqual([
-        { text: 'a', options: { channel: 'stdout' } },
-        { text: 'bc', options: { channel: 'stdout' } },
-        { text: 'E', options: { channel: 'stderr' } },
-      ])
-      // Every read resumed from the previous nextOffset, never from a shared cursor.
-      expect(out.offsets).toEqual([0, 1, 3])
-      expect(err.offsets).toEqual([0, 0, 1])
-      // Settlement cleared the raced poll timer instead of leaving it pending.
-      expect(vi.getTimerCount()).toBe(0)
-    } finally {
-      vi.useRealTimers()
-    }
-  })
-
-  it('subscribes to done once and holds one poll timer however long the job runs', async () => {
-    vi.useFakeTimers()
-    try {
-      const { job } = recordingJob()
-      // A thenable stands in for the producer's done promise, and whatever the
-      // pump derives from it counts its own reactions too: a pending promise
-      // retains every reaction until it settles, so a day-long job must not
-      // gain one per poll round.
-      let subscriptions = 0
-      let settle!: () => void
-      const settled = new Promise<void>((resolve) => { settle = resolve })
-      const counting = (promise: Promise<unknown>): Promise<unknown> => ({
-        then(onFulfilled?: (value: unknown) => unknown, onRejected?: (reason: unknown) => unknown) {
-          subscriptions += 1
-          return counting(promise.then(onFulfilled, onRejected))
-        },
-      }) as unknown as Promise<unknown>
-      const done = counting(settled)
-      const { source, offsets } = scriptedSource([])
-      const pump = pumpJobOutput(job, [source], { pollMs: 50, done })
-
-      const rounds = 10_000
-      await vi.advanceTimersByTimeAsync(50 * rounds)
-      expect(offsets).toHaveLength(rounds + 1)
-      expect(subscriptions).toBeLessThanOrEqual(2)
-      expect(vi.getTimerCount()).toBe(1)
-
-      settle()
-      await pump
-      expect(offsets).toHaveLength(rounds + 2)
-      expect(vi.getTimerCount()).toBe(0)
-    } finally {
-      vi.useRealTimers()
-    }
-  })
-
-  it('marks a lossy source read as a gap so observers see the discontinuity', async () => {
-    const { job, appends } = recordingJob()
-    const { source } = scriptedSource([{ text: 'tail', lossy: true }])
-    await pumpJobOutput(job, [source], { pollMs: 1, done: Promise.resolve() })
-    expect(appends).toEqual([{ text: 'tail', options: { gapBefore: true } }])
-  })
-
-  it('treats a rejected done as settlement and still drains the final bytes', async () => {
-    const { job, appends } = recordingJob()
-    const { source } = scriptedSource([{ text: 'last' }])
-    await pumpJobOutput(job, [source], { pollMs: 1, done: Promise.reject(new Error('producer broke')) })
-    expect(appends).toEqual([{ text: 'last' }])
-  })
-})
diff --git a/tsconfig.base.json b/tsconfig.base.json
index bfdee3acfb..97ea6bcf0c 100644
--- a/tsconfig.base.json
+++ b/tsconfig.base.json
@@ -102,6 +102,7 @@
       "@deepseek-ai/dsh-commands/brand": ["./packages/interaction/commands/src/brand.ts"],
       "@deepseek-ai/dsh-commands/types": ["./packages/interaction/commands/src/types.ts"],
       "@deepseek-ai/dsh-jobs/brand": ["./packages/jobs/jobs/src/brand.ts"],
+      "@deepseek-ai/dsh-jobs/view": ["./packages/jobs/jobs/src/view.ts"],
       "@deepseek-ai/dsh-credentials/types": ["./packages/credentials/credentials/src/types.ts"],
       "@deepseek-ai/dsh-authorization/types": ["./packages/credentials/authorization/src/types.ts"],
       "@deepseek-ai/dsh-settings/types": ["./packages/settings/settings/src/types.ts"],