refactor(shell,workflow): produce into the job record instead of ctx.activities

bash and pwsh background runs declare record: true and pump their observed
streams straight into the job through observeProcessRecord — the open/end
pairing, outcome mapping, and registry-absent degradation branches are gone,
and the pump's final drain folds into hooks.done so the record holds its
last bytes before settlement closes it. activityPollMs becomes recordPollMs.

tool-workflow drops its foreground activity mirror deliberately: a
foreground run's live phase/log narration had the mirror as its only
carrier, and the transcript keeps the recorded run/member lifecycle events.
Live narration returns with backgrounded runs when workflow gains
run_in_background.

packages/shell/tool-bash/README.md

@@ -55,7 +55,7 @@ The tool executes `bash -c <command>` and returns the combined output. Commands
 
 ### Running long commands in the background
 
-Passing `run_in_background: true` returns a job id immediately and no timeout applies; the command keeps running while the agent works on something else. The agent reads its output with `job_output` (non-blocking unless `wait: true`), lists jobs with `job_list`, and stops it with `job_kill`; a finished job notifies the owning agent in-session. Background support needs the generic job runtime (`dsh-jobs-local`) and its control tools (`dsh-tool-jobs`) mounted. When the optional `ctx.activities` registry is loaded, the committed background run is also mirrored into it best-effort: a `bash` activity correlated to the call and job pumps the handle's non-consuming `observed` readers at `activityPollMs` (default 150), so the Web client streams live output without touching the job's cursor; any observation failure is logged and swallowed.
+Passing `run_in_background: true` returns a job id immediately and no timeout applies; the command keeps running while the agent works on something else. The agent reads its output with `job_output` (non-blocking unless `wait: true`), lists jobs with `job_list`, and stops it with `job_kill`; a finished job notifies the owning agent in-session. Background support needs the generic job runtime (`dsh-jobs-local`) and its control tools (`dsh-tool-jobs`) mounted. The background job declares an observation record: the run's non-consuming `observed` readers are pumped into it at `recordPollMs` (default 150), so the Web client streams live output without touching the job's consuming cursor; a pump failure is logged and swallowed while the job runs on.
 
 ### Sandboxed execution and escalation
 

packages/shell/tool-bash/package.json

@@ -37,7 +37,6 @@
     "@deepseek-ai/dsh-invariants": "workspace:^",
     "@deepseek-ai/dsh-jobs": "workspace:^",
     "@deepseek-ai/dsh-llm": "workspace:^",
-    "@deepseek-ai/dsh-activity": "workspace:^",
     "@deepseek-ai/dsh-sandbox": "workspace:^",
     "@deepseek-ai/dsh-sandbox-policy": "workspace:^",
     "@deepseek-ai/dsh-shell": "workspace:^",
@@ -59,7 +58,6 @@
     "@deepseek-ai/dsh-jobs": "workspace:^",
     "@deepseek-ai/dsh-jobs-local": "workspace:^",
     "@deepseek-ai/dsh-llm": "workspace:^",
-    "@deepseek-ai/dsh-activity": "workspace:^",
     "@deepseek-ai/dsh-sandbox": "workspace:^",
     "@deepseek-ai/dsh-sandbox-policy": "workspace:^",
     "@deepseek-ai/dsh-session": "workspace:^",

packages/shell/tool-bash/src/index.ts

@@ -22,10 +22,9 @@ import type { SandboxExecutionPolicy, SandboxMode } from '@deepseek-ai/dsh-sandb
 import { ESCALATION_TARGETS, approveEscalation, canonicalPath, validateEscalationArgs } from '@deepseek-ai/dsh-sandbox'
 import type { SandboxPolicyService } from '@deepseek-ai/dsh-sandbox-policy'
 import { DSH_ENV_PREFIX } from '@deepseek-ai/dsh-shell'
-import type { ShellProcess, ShellRunResult } from '@deepseek-ai/dsh-shell'
-import type {} from '@deepseek-ai/dsh-activity'
+import type { ShellRunResult } from '@deepseek-ai/dsh-shell'
 import { processOutcome } from './background.ts'
-import { observeBackgroundActivity } from './observe.ts'
+import { observeProcessRecord } from './observe.ts'
 import { parseExitStatus, renderProcessRead, renderResult } from './render.ts'
 
 export const name = 'tool-bash'
@@ -35,14 +34,14 @@ export const inject = ['tools', 'shell', 'systemPrompt', 'shellEnv']
 export interface Config {
   /** Expose `run_in_background` (default true); disabled calls are also rejected. */
   enableRunInBackground?: boolean
-  /** Poll cadence for mirroring background output into `ctx.activities`, in milliseconds (default 150). */
-  activityPollMs?: number
+  /** Poll cadence for copying background output into the job record, in milliseconds (default 150). */
+  recordPollMs?: number
 }
 
 /** Runtime configuration schema for the bash tool plugin. */
 export const Config: z<Config> = z.object({
   enableRunInBackground: z.boolean().default(true),
-  activityPollMs: z.number()
+  recordPollMs: z.number()
     .step(1)
     .min(1)
     .max(Number.MAX_SAFE_INTEGER)
@@ -197,7 +196,7 @@ const BACKGROUND_OUTPUT_PROPERTIES = {
 
 export function apply(ctx: Context, config: Config = {}): void {
   const backgroundEnabled = config.enableRunInBackground ?? true
-  const activityPollMs = config.activityPollMs ?? 150
+  const recordPollMs = config.recordPollMs ?? 150
   const defaultMode = ctx.shell.sandboxMode
   const escalationModes: readonly SandboxMode[] = defaultMode === undefined ? [] : ESCALATION_TARGETS
   const sandboxPolicy: SandboxPolicyService | undefined = defaultMode === undefined ? undefined : ctx.get('sandboxPolicy')
@@ -370,32 +369,23 @@ export function apply(ctx: Context, config: Config = {}): void {
           error.name = 'AbortError'
           throw error
         }
-        // Task preflight finishes before the starter can spawn a process.
-        let started: ShellProcess | undefined
         const id = jobs.start({
           kind: 'bash',
           label: args.command,
           ...exec.agent ? { owner: exec.agent } : {},
-          run: () => {
+          record: true,
+          run: (job) => {
             const proc = ctx.shell.start(ctx.shell.resolve(request))
-            started = proc
+            // The pump's final drain is folded into `done` so the record holds
+            // its last bytes before settlement trims and closes it.
+            const observed = observeProcessRecord(ctx, job, proc, recordPollMs)
             return {
               cancel: () => void proc.kill(),
-              done: proc.done.then(() => processOutcome(proc)),
+              done: observed.then(() => proc.done).then(() => processOutcome(proc)),
               readOutput: () => renderProcessRead(proc.readOutput(), proc.sandbox, escalationModes),
             }
           },
         })
-        // The starter ran synchronously inside the committed start; mirror the
-        // spawned process into the optional observation registry.
-        if (started !== undefined) {
-          observeBackgroundActivity(ctx, started, {
-            kind: 'bash',
-            label: args.command,
-            ...exec.agent ? { owner: exec.agent } : {},
-            correlation: { callId: exec.callId, jobId: id },
-          }, activityPollMs, processOutcome)
-        }
         return { kind: 'background' as const, jobId: id }
       }
       const result = await ctx.shell.run(ctx.shell.resolve({

packages/shell/tool-bash/src/observe.ts

@@ -1,82 +1,44 @@
 /**
- * Best-effort process-observation tap for background bash runs: mirrors the
- * spawned process's captured streams into `ctx.activities` so non-consuming
- * observers (the Web client) see live output without touching the job's
- * consuming `readOutput` cursor. Absent registry or absent backend offset
- * readers degrade to no observation; the job path is unaffected either way.
+ * Record tap for background bash runs: copies the spawned process's captured
+ * streams into the job's observation record, so non-consuming observers (the
+ * Web client) see live output without touching the job's consuming
+ * `readOutput` cursor. Observation is strictly best-effort: absent backend
+ * offset readers degrade to no observation, and a pump failure is logged and
+ * swallowed — the job path is unaffected either way.
  *
  * @module @deepseek-ai/dsh-tool-bash/observe
  */
 
 import type { Context } from '@deepseek-ai/cordis'
-import type { Agent } from '@deepseek-ai/dsh-agent'
 import type { ShellProcess } from '@deepseek-ai/dsh-shell'
-import { pumpActivityOutput } from '@deepseek-ai/dsh-activity'
-import type { ActivityCorrelation, ActivityKind, ActivityPumpSource } from '@deepseek-ai/dsh-activity'
-
-/** Identity of one observed background run. */
-export interface ObserveBackgroundSpec {
-  /** Process kind — the producing tool's registered {@link ActivityKind}. */
-  kind: ActivityKind
-  /** One-line label (the command). */
-  label: string
-  /** Owning live agent, when the call has one. */
-  owner?: Agent
-  /** Links to the starting tool call and the registered job. */
-  correlation: ActivityCorrelation
-}
+import { pumpJobOutput } from '@deepseek-ai/dsh-jobs'
+import type { JobPumpSource, RunningJob } from '@deepseek-ai/dsh-jobs'
 
 /**
- * Open a process beside a committed background job and pump the handle's
- * non-consuming offset readers into it until the run settles, then record the
- * mapped outcome. Observation is strictly best-effort: without a
- * `ctx.activities` registry this is a no-op, without backend offset readers
- * the process still carries status and settlement, and ANY observation
- * failure — including a rejected `open()` — is logged and swallowed, because
- * the background job is already committed and running.
- * @param ctx - plugin context used for the optional registry and warning logs.
+ * Pump the process's non-consuming offset readers into the job record until
+ * the run settles, then drain once more. The returned promise never rejects:
+ * a pump failure is logged and the record simply stops advancing, while the
+ * process and its job continue untouched.
+ * @param ctx - plugin context used for the warning log.
+ * @param job - the running job's producer face receiving the copied chunks.
  * @param proc - the started background process handle.
- * @param spec - process identity and correlation for the registry row.
  * @param pollMs - producer-configured pump cadence in milliseconds.
- * @param outcome - maps the settled handle to the process outcome.
+ * @returns resolves after the final drain; fold into the job's `done` chain so
+ *   the record holds its final bytes before settlement closes it.
  */
-export function observeBackgroundActivity(
+export function observeProcessRecord(
   ctx: Context,
+  job: RunningJob,
   proc: ShellProcess,
-  spec: ObserveBackgroundSpec,
   pollMs: number,
-  outcome: (proc: ShellProcess) => { status: 'completed' | 'killed'; detail: string },
-): void {
-  try {
-    const processes = ctx.get('activities')
-    if (processes === undefined) return
-    const handle = processes.open({
-      kind: spec.kind,
-      label: spec.label,
-      ...spec.owner !== undefined ? { owner: spec.owner } : {},
-      correlation: spec.correlation,
+): Promise<void> {
+  const sources: JobPumpSource[] = []
+  const stdout = proc.observed?.stdout
+  if (stdout !== undefined) sources.push({ channel: 'stdout', read: from => stdout.readFrom(from) })
+  const stderr = proc.observed?.stderr
+  if (stderr !== undefined) sources.push({ channel: 'stderr', read: from => stderr.readFrom(from) })
+  return pumpJobOutput(job, sources, { pollMs, done: proc.done })
+    .catch((error: unknown) => {
+      ctx.logger.warn(`record observation pump for ${job.id} failed: ${String(error)}`)
     })
-    const sources: ActivityPumpSource[] = []
-    const stdout = proc.observed?.stdout
-    if (stdout !== undefined) sources.push({ channel: 'stdout', read: from => stdout.readFrom(from) })
-    const stderr = proc.observed?.stderr
-    if (stderr !== undefined) sources.push({ channel: 'stderr', read: from => stderr.readFrom(from) })
-    void pumpActivityOutput(handle, sources, { pollMs, done: proc.done })
-      .catch((error: unknown) => {
-        ctx.logger.warn(`activity observation pump for ${handle.id} failed: ${String(error)}`)
-      })
-      .then(async () => {
-        // The pump can fail while the process is still running; the outcome
-        // mapping requires a settled handle, and `end` is first-wins, so an
-        // early mapping would freeze a wrong terminal state onto the row.
-        // `proc.done` never rejects.
-        await proc.done
-        handle.end(outcome(proc))
-      })
-      .catch((error: unknown) => {
-        ctx.logger.warn(`activity settlement mapping for ${handle.id} failed: ${String(error)}`)
-      })
-  } catch (error: unknown) {
-    ctx.logger.warn(`activity observation unavailable for this run: ${String(error)}`)
-  }
 }

packages/shell/tool-bash/tests/observe.spec.ts

@@ -10,18 +10,19 @@ import ToolRuntime from '@deepseek-ai/dsh-tools'
 import AgentRegistry from '@deepseek-ai/dsh-agent'
 import type { Agent } from '@deepseek-ai/dsh-agent'
 import LocalJobRegistry from '@deepseek-ai/dsh-jobs-local'
+import type { JobId, RunningJob } from '@deepseek-ai/dsh-jobs'
 import * as ToolTasks from '@deepseek-ai/dsh-tool-jobs'
-import LocalActivityRegistry from '@deepseek-ai/dsh-activity-local'
-import type { ActivitySnapshot } from '@deepseek-ai/dsh-activity'
+import type { ShellProcess } from '@deepseek-ai/dsh-shell'
 import { LocalBashExecutor } from '@deepseek-ai/dsh-bash-local'
 import LocalSubprocessRuntime from '@deepseek-ai/dsh-subprocess-local'
 import * as ToolBash from '@deepseek-ai/dsh-tool-bash'
+import { observeProcessRecord } from '../src/observe.ts'
 import * as BashEnvPlugin from '@deepseek-ai/dsh-shell-env'
 
 const testToolSignal = new AbortController().signal
 const spillDir = mkdtempSync(join(tmpdir(), 'dsh-tool-bash-observe-spec-'))
 
-/** Job harness plus the observation registry, with a fast pump for tests. */
+/** Job harness with a fast record pump for tests. */
 async function setup() {
   const ctx = new Context()
   await ctx.plugin(SystemPrompt)
@@ -29,12 +30,11 @@ async function setup() {
   await ctx.plugin(AgentRegistry)
   await ctx.plugin(LocalJobRegistry)
   await ctx.plugin(ToolTasks)
-  await ctx.plugin(LocalActivityRegistry)
   await ctx.plugin(LocalSubprocessRuntime)
   ;(ctx.subprocess as LocalSubprocessRuntime).internals = { spillDir }
   await ctx.plugin(BashEnvPlugin)
   await ctx.plugin(LocalBashExecutor, { timeoutMs: 10_000, graceMs: 200 })
-  await ctx.plugin(ToolBash, { activityPollMs: 25 })
+  await ctx.plugin(ToolBash, { recordPollMs: 25 })
   return ctx
 }
 
@@ -62,12 +62,12 @@ async function until<T>(read: () => T | undefined, timeoutMs = 5_000): Promise<T
   throw new Error('condition not reached before timeout')
 }
 
-function retainedText(ctx: Context, id: ActivitySnapshot['id']): string {
-  return ctx.activities.read(id, 0).chunks.map(chunk => chunk.text).join('')
+function retainedText(ctx: Context, id: JobId, caller?: Agent): string {
+  return ctx.jobs.readRecord(id, 0, caller).chunks.map(chunk => chunk.text).join('')
 }
 
-describe('background bash observation', () => {
-  it('mirrors a background run into ctx.activities with correlation, live output, and settlement', async () => {
+describe('background bash record', () => {
+  it('streams a background run into the job record with live output and settlement', async () => {
     const ctx = await setup()
     const ack = await call(ctx, {
       command: 'printf "line-1\\n"; sleep 0.4; printf "line-2\\n"',
@@ -77,21 +77,16 @@ describe('background bash observation', () => {
     expect(text(ack)).toContain('started background job')
     const job = ctx.jobs.list()[0]
     expect(job).toBeDefined()
+    expect(job!.outputTotal).toBeDefined()
 
-    const snapshot = await until(() =>
-      ctx.activities.list().find(row => row.correlation?.jobId === job!.id))
-    expect(snapshot.kind).toBe('bash')
-    expect(snapshot.label).toContain('printf')
-    expect(String(snapshot.correlation?.callId)).toMatch(/^observe-call-/)
+    // Live output appears in the record while the command is still running.
+    await until(() => retainedText(ctx, job!.id).includes('line-1') ? true : undefined)
+    expect(ctx.jobs.get(job!.id).status).toBe('running')
 
-    // Live output appears while the command is still running.
-    await until(() => retainedText(ctx, snapshot.id).includes('line-1') ? true : undefined)
-    expect(ctx.activities.get(snapshot.id).status).toBe('running')
-
-    // Settlement carries the mapped bash outcome and the trailing output.
-    await until(() => ctx.activities.get(snapshot.id).status === 'completed' ? true : undefined)
-    expect(ctx.activities.get(snapshot.id).detail).toBe('exit code: 0')
-    expect(retainedText(ctx, snapshot.id)).toContain('line-2')
+    // Settlement ends the record with the job; the trailing bytes are drained first.
+    await until(() => ctx.jobs.get(job!.id).status === 'completed' ? true : undefined)
+    expect(ctx.jobs.get(job!.id).detail).toBe('exit code: 0')
+    expect(retainedText(ctx, job!.id)).toContain('line-2')
 
     // The model-facing consuming cursor still delivers everything: observation stole nothing.
     const collected = await until(() => {
@@ -101,15 +96,13 @@ describe('background bash observation', () => {
     expect(collected).toContain('line-2')
   })
 
-  it('a killed background job settles its mirrored process as killed', async () => {
+  it('a killed background job ends its record with the killed settlement', async () => {
     const ctx = await setup()
     await call(ctx, { command: 'sleep 60', description: 'test command', run_in_background: true })
     const job = ctx.jobs.list()[0]
-    const snapshot = await until(() =>
-      ctx.activities.list().find(row => row.correlation?.jobId === job!.id))
     ctx.jobs.kill(job!.id, undefined, 'test cleanup')
-    await until(() => ctx.activities.get(snapshot.id).status === 'killed' ? true : undefined)
-    expect(ctx.activities.get(snapshot.id).detail).toMatch(/signal|killed before exit/)
+    await until(() => ctx.jobs.get(job!.id).status === 'killed' ? true : undefined)
+    expect(ctx.jobs.get(job!.id).detail).toMatch(/signal|killed before exit/)
   })
 
   it('labels stderr chunks with their channel', async () => {
@@ -120,153 +113,53 @@ describe('background bash observation', () => {
       run_in_background: true,
     })
     const job = ctx.jobs.list()[0]
-    const snapshot = await until(() =>
-      ctx.activities.list().find(row => row.correlation?.jobId === job!.id))
-    await until(() => ctx.activities.get(snapshot.id).status === 'completed' ? true : undefined)
-    const chunks = ctx.activities.read(snapshot.id, 0).chunks
+    await until(() => ctx.jobs.get(job!.id).status === 'completed' ? true : undefined)
+    const chunks = ctx.jobs.readRecord(job!.id, 0).chunks
     expect(chunks.find(chunk => chunk.text.includes('out-line'))?.channel).toBe('stdout')
     expect(chunks.find(chunk => chunk.text.includes('err-line'))?.channel).toBe('stderr')
   })
+})
 
-  it('a registry whose handle rejects appends degrades to a warning and still settles', async () => {
-    const ctx = new Context()
-    await ctx.plugin(SystemPrompt)
-    await ctx.plugin(ToolRuntime)
-    await ctx.plugin(AgentRegistry)
-    await ctx.plugin(LocalJobRegistry)
-    await ctx.plugin(ToolTasks)
-    const ended: unknown[] = []
-    await ctx.plugin({
-      name: 'appending-broken-activities-probe',
-      apply(child: Context) {
-        child.provide('activities', {
-          open: () => ({
-            id: 'bash-1',
-            append() { throw new Error('append boom') },
-            updateDetail() {},
-            end(outcome: unknown) { ended.push(outcome) },
-          }),
-        })
+describe('observeProcessRecord', () => {
+  /** A RunningJob face recording appends for pump assertions. */
+  function recordingJob(): { job: RunningJob; appends: string[] } {
+    const appends: string[] = []
+    return {
+      job: {
+        id: 'bash-1' as JobId,
+        append(chunk) { appends.push(chunk) },
+        updateDetail() {},
       },
-    })
-    await ctx.plugin(LocalSubprocessRuntime)
-    ;(ctx.subprocess as LocalSubprocessRuntime).internals = { spillDir }
-    await ctx.plugin(BashEnvPlugin)
-    await ctx.plugin(LocalBashExecutor, { timeoutMs: 10_000, graceMs: 200 })
-    const warn = vi.fn()
-    ctx.logger.warn = warn as never
-    await ctx.plugin(ToolBash, { activityPollMs: 25 })
-    await call(ctx, { command: 'echo tolerated', description: 'test command', run_in_background: true })
-    await until(() => ended.length > 0 ? true : undefined)
-    expect(warn).toHaveBeenCalledWith(expect.stringContaining('activity observation pump'))
-    expect(ended[0]).toMatchObject({ status: 'completed' })
+      appends,
+    }
+  }
+
+  it('degrades to no observation when the backend exposes no offset readers', async () => {
+    const ctx = new Context()
+    const { job, appends } = recordingJob()
+    const proc = { done: Promise.resolve() } as unknown as ShellProcess
+    await observeProcessRecord(ctx, job, proc, 25)
+    expect(appends).toEqual([])
   })
 
-  it('a pump failure waits for process settlement before mapping the outcome', async () => {
+  it('contains a pump failure with a warning; the job path is unaffected', async () => {
     const ctx = new Context()
-    await ctx.plugin(SystemPrompt)
-    await ctx.plugin(ToolRuntime)
-    await ctx.plugin(AgentRegistry)
-    await ctx.plugin(LocalJobRegistry)
-    await ctx.plugin(ToolTasks)
-    const ended: unknown[] = []
-    await ctx.plugin({
-      name: 'early-failing-activities-probe',
-      apply(child: Context) {
-        child.provide('activities', {
-          open: () => ({
-            id: 'bash-1',
-            append() { throw new Error('append boom') },
-            updateDetail() {},
-            end(outcome: unknown) { ended.push(outcome) },
-          }),
-        })
-      },
-    })
-    await ctx.plugin(LocalSubprocessRuntime)
-    ;(ctx.subprocess as LocalSubprocessRuntime).internals = { spillDir }
-    await ctx.plugin(BashEnvPlugin)
-    await ctx.plugin(LocalBashExecutor, { timeoutMs: 10_000, graceMs: 200 })
     const warn = vi.fn()
     ctx.logger.warn = warn as never
-    await ctx.plugin(ToolBash, { activityPollMs: 25 })
-    // One early line makes the first pump poll append (and fail) while the
-    // process still has most of its sleep ahead.
-    await call(ctx, { command: 'printf "early\\n"; sleep 0.7', description: 'test command', run_in_background: true })
-    await until(() => warn.mock.calls.some(args => String(args[0]).includes('activity observation pump')) ? true : undefined)
-    // The pump already failed; the outcome must wait for real settlement
-    // rather than freezing a fake terminal state onto the running process.
-    expect(ended).toHaveLength(0)
-    await until(() => ended.length > 0 ? true : undefined)
-    expect(ended[0]).toMatchObject({ status: 'completed', detail: 'exit code: 0' })
-  })
-
-  it('a throwing end() is swallowed by the settlement-mapping catch', async () => {
-    const ctx = new Context()
-    await ctx.plugin(SystemPrompt)
-    await ctx.plugin(ToolRuntime)
-    await ctx.plugin(AgentRegistry)
-    await ctx.plugin(LocalJobRegistry)
-    await ctx.plugin(ToolTasks)
-    await ctx.plugin({
-      name: 'end-throwing-activities-probe',
-      apply(child: Context) {
-        child.provide('activities', {
-          open: () => ({
-            id: 'bash-1',
-            append() {},
-            updateDetail() {},
-            end() { throw new Error('end boom') },
-          }),
-        })
+    const { job } = recordingJob()
+    const proc = {
+      done: new Promise<void>((resolve) => { setTimeout(resolve, 10) }),
+      observed: {
+        stdout: { readFrom() { throw new Error('reader boom') } },
       },
-    })
-    await ctx.plugin(LocalSubprocessRuntime)
-    ;(ctx.subprocess as LocalSubprocessRuntime).internals = { spillDir }
-    await ctx.plugin(BashEnvPlugin)
-    await ctx.plugin(LocalBashExecutor, { timeoutMs: 10_000, graceMs: 200 })
-    const warn = vi.fn()
-    ctx.logger.warn = warn as never
-    await ctx.plugin(ToolBash, { activityPollMs: 25 })
-    await call(ctx, { command: 'echo settled', description: 'test command', run_in_background: true })
-    await until(() => warn.mock.calls.some(args => String(args[0]).includes('activity settlement mapping')) ? true : undefined)
-  })
-
-  it('a throwing observation registry never breaks the background call', async () => {
-    const ctx = new Context()
-    await ctx.plugin(SystemPrompt)
-    await ctx.plugin(ToolRuntime)
-    await ctx.plugin(AgentRegistry)
-    await ctx.plugin(LocalJobRegistry)
-    await ctx.plugin(ToolTasks)
-    await ctx.plugin({
-      name: 'broken-activities-probe',
-      apply(child: Context) {
-        child.provide('activities', { open() { throw new Error('observation boom') } })
-      },
-    })
-    await ctx.plugin(LocalSubprocessRuntime)
-    ;(ctx.subprocess as LocalSubprocessRuntime).internals = { spillDir }
-    await ctx.plugin(BashEnvPlugin)
-    await ctx.plugin(LocalBashExecutor, { timeoutMs: 10_000, graceMs: 200 })
-    const warn = vi.fn()
-    ctx.logger.warn = warn as never
-    await ctx.plugin(ToolBash, { activityPollMs: 25 })
-    const ack = await call(ctx, { command: 'echo resilient', description: 'test command', run_in_background: true })
-    expect(text(ack)).toContain('started background job')
-    expect(warn).toHaveBeenCalledWith(expect.stringContaining('activity observation unavailable'))
-    // The job itself still completes and reads normally.
-    const job = ctx.jobs.list()[0]
-    const read = await until(() => {
-      const value = ctx.jobs.read(job!.id)
-      return value.text.includes('resilient') ? value : undefined
-    })
-    expect(read.snapshot.status).toBe('completed')
+    } as unknown as ShellProcess
+    await observeProcessRecord(ctx, job, proc, 5)
+    expect(warn).toHaveBeenCalledWith(expect.stringContaining('record observation pump for bash-1 failed'))
   })
 })
 
-describe('owned background observation', () => {
-  it('mirrors an owned background run under the owning session', async () => {
+describe('owned background record', () => {
+  it('fences an owned background run record under the owning session', async () => {
     const ctx = await setup()
     const owner = {
       id: SessionId('observe-owner'),
@@ -284,8 +177,9 @@ describe('owned background observation', () => {
     })
     const job = ctx.jobs.list(owner)[0]
     expect(job).toBeDefined()
-    const row = await until(() => ctx.activities.list(owner).find(item => item.correlation?.jobId === job!.id))
-    expect(row.ownerSession).toBe(owner.id)
+    expect(job!.ownerSession).toBe(owner.id)
     await until(() => ctx.jobs.get(job!.id, owner).status === 'completed' ? true : undefined)
+    expect(() => retainedText(ctx, job!.id)).toThrow(/belongs to another session/)
+    await until(() => retainedText(ctx, job!.id, owner).includes('owned') ? true : undefined)
   })
 })

packages/shell/tool-bash/tsconfig.json

@@ -49,9 +49,6 @@
     },
     {
       "path": "../../sandbox/sandbox-policy"
-    },
-    {
-      "path": "../../activity/activity"
     }
   ]
 }

packages/shell/tool-pwsh/README.md

@@ -9,7 +9,7 @@ English | [中文](README.zh.md)
 
 ## Summary
 
-`dsh-tool-pwsh` gives the agent a `pwsh` tool that runs PowerShell commands through the mounted shell executor — the Windows counterpart of `dsh-tool-bash`, mirroring it call-for-call. Each call runs in a fresh pwsh process, so no state survives; `run_in_background` turns long-running commands into background jobs. Commands are PowerShell-dialect: native `C:\...` paths and `$env:NAME` variables, with no dialect translation. Every call runs with the managed `DSH_*` environment, and under a sandboxing executor the tool teaches and enforces the Windows-specific language-mode and named-pipe contracts. Mount it with a PowerShell executor such as `dsh-pwsh-local` and the `dsh-shell-env` plugin. When the optional `ctx.activities` registry is loaded, a committed background run is also mirrored into it best-effort as a `pwsh` activity pumped at `activityPollMs` (default 150), exactly like `dsh-tool-bash`.
+`dsh-tool-pwsh` gives the agent a `pwsh` tool that runs PowerShell commands through the mounted shell executor — the Windows counterpart of `dsh-tool-bash`, mirroring it call-for-call. Each call runs in a fresh pwsh process, so no state survives; `run_in_background` turns long-running commands into background jobs. Commands are PowerShell-dialect: native `C:\...` paths and `$env:NAME` variables, with no dialect translation. Every call runs with the managed `DSH_*` environment, and under a sandboxing executor the tool teaches and enforces the Windows-specific language-mode and named-pipe contracts. Mount it with a PowerShell executor such as `dsh-pwsh-local` and the `dsh-shell-env` plugin. A background run's job declares an observation record pumped at `recordPollMs` (default 150), exactly like `dsh-tool-bash`.
 
 ## Table of Contents
 

packages/shell/tool-pwsh/package.json

@@ -37,7 +37,6 @@
     "@deepseek-ai/dsh-invariants": "workspace:^",
     "@deepseek-ai/dsh-jobs": "workspace:^",
     "@deepseek-ai/dsh-llm": "workspace:^",
-    "@deepseek-ai/dsh-activity": "workspace:^",
     "@deepseek-ai/dsh-sandbox": "workspace:^",
     "@deepseek-ai/dsh-sandbox-policy": "workspace:^",
     "@deepseek-ai/dsh-shell": "workspace:^",
@@ -59,7 +58,6 @@
     "@deepseek-ai/dsh-jobs-local": "workspace:^",
     "@deepseek-ai/dsh-llm": "workspace:^",
     "@deepseek-ai/dsh-loader-smoke": "workspace:^",
-    "@deepseek-ai/dsh-activity": "workspace:^",
     "@deepseek-ai/dsh-pwsh-local": "workspace:^",
     "@deepseek-ai/dsh-sandbox": "workspace:^",
     "@deepseek-ai/dsh-sandbox-policy": "workspace:^",

packages/shell/tool-pwsh/src/index.ts

@@ -32,11 +32,10 @@ import type {} from '@deepseek-ai/dsh-user-approval'
 import type { SandboxExecutionPolicy, SandboxMode } from '@deepseek-ai/dsh-sandbox'
 import { ESCALATION_TARGETS, approveEscalation, validateEscalationArgs } from '@deepseek-ai/dsh-sandbox'
 import type { SandboxPolicyService } from '@deepseek-ai/dsh-sandbox-policy'
-import type { ShellProcess, ShellRunResult } from '@deepseek-ai/dsh-shell'
+import type { ShellRunResult } from '@deepseek-ai/dsh-shell'
 import { parseExitStatus } from '@deepseek-ai/dsh-shell'
-import type {} from '@deepseek-ai/dsh-activity'
 import { processOutcome } from './background.ts'
-import { observeBackgroundActivity } from './observe.ts'
+import { observeProcessRecord } from './observe.ts'
 import { renderPwshProcessRead, renderPwshResult } from './render.ts'
 import type { RenderablePwshResult } from './render.ts'
 
@@ -46,12 +45,6 @@ declare module '@deepseek-ai/dsh-jobs' {
   }
 }
 
-declare module '@deepseek-ai/dsh-activity' {
-  interface ActivityKindMap {
-    pwsh: 'pwsh'
-  }
-}
-
 export const name = 'tool-pwsh'
 export const inject = ['tools', 'shell', 'systemPrompt', 'shellEnv']
 
@@ -60,14 +53,14 @@ export const inject = ['tools', 'shell', 'systemPrompt', 'shellEnv']
 export interface Config {
   /** Expose `run_in_background` (default true); disabled calls are also rejected. */
   enableRunInBackground?: boolean
-  /** Poll cadence for mirroring background output into `ctx.activities`, in milliseconds (default 150). */
-  activityPollMs?: number
+  /** Poll cadence for copying background output into the job record, in milliseconds (default 150). */
+  recordPollMs?: number
 }
 
 /** Runtime configuration schema for the pwsh tool plugin. */
 export const Config: z<Config> = z.object({
   enableRunInBackground: z.boolean().default(true),
-  activityPollMs: z.number()
+  recordPollMs: z.number()
     .step(1)
     .min(1)
     .max(Number.MAX_SAFE_INTEGER)
@@ -210,7 +203,7 @@ const BACKGROUND_OUTPUT_PROPERTIES = {
 /* jscpd:ignore-start -- deliberate mirror of dsh-tool-bash's apply() preamble (pwsh-tool-and-executor Agent Note). */
 export function apply(ctx: Context, config: Config = {}): void {
   const backgroundEnabled = config.enableRunInBackground ?? true
-  const activityPollMs = config.activityPollMs ?? 150
+  const recordPollMs = config.recordPollMs ?? 150
   const defaultMode = ctx.shell.sandboxMode
   const escalationModes: readonly SandboxMode[] = defaultMode === undefined ? [] : ESCALATION_TARGETS
   const sandboxPolicy: SandboxPolicyService | undefined = defaultMode === undefined ? undefined : ctx.get('sandboxPolicy')
@@ -394,32 +387,23 @@ export function apply(ctx: Context, config: Config = {}): void {
           error.name = 'AbortError'
           throw error
         }
-        // Task preflight finishes before the starter can spawn a process.
-        let started: ShellProcess | undefined
         const id = jobs.start({
           kind: 'pwsh',
           label: args.command,
           ...exec.agent ? { owner: exec.agent } : {},
-          run: () => {
+          record: true,
+          run: (job) => {
             const proc = ctx.shell.start(ctx.shell.resolve(request))
-            started = proc
+            // The pump's final drain is folded into `done` so the record holds
+            // its last bytes before settlement trims and closes it.
+            const observed = observeProcessRecord(ctx, job, proc, recordPollMs)
             return {
               cancel: () => void proc.kill(),
-              done: proc.done.then(() => processOutcome(proc)),
+              done: observed.then(() => proc.done).then(() => processOutcome(proc)),
               readOutput: () => renderPwshProcessRead(proc.readOutput(), proc.sandbox, escalationModes),
             }
           },
         })
-        // The starter ran synchronously inside the committed start; mirror the
-        // spawned process into the optional observation registry.
-        if (started !== undefined) {
-          observeBackgroundActivity(ctx, started, {
-            kind: 'pwsh',
-            label: args.command,
-            ...exec.agent ? { owner: exec.agent } : {},
-            correlation: { callId: exec.callId, jobId: id },
-          }, activityPollMs, processOutcome)
-        }
         return { kind: 'background' as const, jobId: id }
       }
       const result = await ctx.shell.run(ctx.shell.resolve({

packages/shell/tool-pwsh/src/observe.ts

@@ -1,84 +1,46 @@
 /**
- * Best-effort process-observation tap for background pwsh runs: mirrors the
- * spawned process's captured streams into `ctx.activities` so non-consuming
- * observers (the Web client) see live output without touching the job's
- * consuming `readOutput` cursor. Absent registry or absent backend offset
- * readers degrade to no observation; the job path is unaffected either way.
+ * Record tap for background pwsh runs: copies the spawned process's captured
+ * streams into the job's observation record, so non-consuming observers (the
+ * Web client) see live output without touching the job's consuming
+ * `readOutput` cursor. Observation is strictly best-effort: absent backend
+ * offset readers degrade to no observation, and a pump failure is logged and
+ * swallowed — the job path is unaffected either way.
  *
  * @module @deepseek-ai/dsh-tool-pwsh/observe
  */
 
 /* jscpd:ignore-start -- mirrors tool-bash's observe tap by design, like render/background. */
 import type { Context } from '@deepseek-ai/cordis'
-import type { Agent } from '@deepseek-ai/dsh-agent'
 import type { ShellProcess } from '@deepseek-ai/dsh-shell'
-import { pumpActivityOutput } from '@deepseek-ai/dsh-activity'
-import type { ActivityCorrelation, ActivityKind, ActivityPumpSource } from '@deepseek-ai/dsh-activity'
-
-/** Identity of one observed background run. */
-export interface ObserveBackgroundSpec {
-  /** Process kind — the producing tool's registered {@link ActivityKind}. */
-  kind: ActivityKind
-  /** One-line label (the command). */
-  label: string
-  /** Owning live agent, when the call has one. */
-  owner?: Agent
-  /** Links to the starting tool call and the registered job. */
-  correlation: ActivityCorrelation
-}
+import { pumpJobOutput } from '@deepseek-ai/dsh-jobs'
+import type { JobPumpSource, RunningJob } from '@deepseek-ai/dsh-jobs'
 
 /**
- * Open a process beside a committed background job and pump the handle's
- * non-consuming offset readers into it until the run settles, then record the
- * mapped outcome. Observation is strictly best-effort: without a
- * `ctx.activities` registry this is a no-op, without backend offset readers
- * the process still carries status and settlement, and ANY observation
- * failure — including a rejected `open()` — is logged and swallowed, because
- * the background job is already committed and running.
- * @param ctx - plugin context used for the optional registry and warning logs.
+ * Pump the process's non-consuming offset readers into the job record until
+ * the run settles, then drain once more. The returned promise never rejects:
+ * a pump failure is logged and the record simply stops advancing, while the
+ * process and its job continue untouched.
+ * @param ctx - plugin context used for the warning log.
+ * @param job - the running job's producer face receiving the copied chunks.
  * @param proc - the started background process handle.
- * @param spec - process identity and correlation for the registry row.
  * @param pollMs - producer-configured pump cadence in milliseconds.
- * @param outcome - maps the settled handle to the process outcome.
+ * @returns resolves after the final drain; fold into the job's `done` chain so
+ *   the record holds its final bytes before settlement closes it.
  */
-export function observeBackgroundActivity(
+export function observeProcessRecord(
   ctx: Context,
+  job: RunningJob,
   proc: ShellProcess,
-  spec: ObserveBackgroundSpec,
   pollMs: number,
-  outcome: (proc: ShellProcess) => { status: 'completed' | 'killed'; detail: string },
-): void {
-  try {
-    const processes = ctx.get('activities')
-    if (processes === undefined) return
-    const handle = processes.open({
-      kind: spec.kind,
-      label: spec.label,
-      ...spec.owner !== undefined ? { owner: spec.owner } : {},
-      correlation: spec.correlation,
+): Promise<void> {
+  const sources: JobPumpSource[] = []
+  const stdout = proc.observed?.stdout
+  if (stdout !== undefined) sources.push({ channel: 'stdout', read: from => stdout.readFrom(from) })
+  const stderr = proc.observed?.stderr
+  if (stderr !== undefined) sources.push({ channel: 'stderr', read: from => stderr.readFrom(from) })
+  return pumpJobOutput(job, sources, { pollMs, done: proc.done })
+    .catch((error: unknown) => {
+      ctx.logger.warn(`record observation pump for ${job.id} failed: ${String(error)}`)
     })
-    const sources: ActivityPumpSource[] = []
-    const stdout = proc.observed?.stdout
-    if (stdout !== undefined) sources.push({ channel: 'stdout', read: from => stdout.readFrom(from) })
-    const stderr = proc.observed?.stderr
-    if (stderr !== undefined) sources.push({ channel: 'stderr', read: from => stderr.readFrom(from) })
-    void pumpActivityOutput(handle, sources, { pollMs, done: proc.done })
-      .catch((error: unknown) => {
-        ctx.logger.warn(`activity observation pump for ${handle.id} failed: ${String(error)}`)
-      })
-      .then(async () => {
-        // The pump can fail while the process is still running; the outcome
-        // mapping requires a settled handle, and `end` is first-wins, so an
-        // early mapping would freeze a wrong terminal state onto the row.
-        // `proc.done` never rejects.
-        await proc.done
-        handle.end(outcome(proc))
-      })
-      .catch((error: unknown) => {
-        ctx.logger.warn(`activity settlement mapping for ${handle.id} failed: ${String(error)}`)
-      })
-  } catch (error: unknown) {
-    ctx.logger.warn(`activity observation unavailable for this run: ${String(error)}`)
-  }
 }
 /* jscpd:ignore-end */

packages/shell/tool-pwsh/tests/observe.spec.ts

@@ -8,7 +8,6 @@ import AgentRegistry from '@deepseek-ai/dsh-agent'
 import type { Agent } from '@deepseek-ai/dsh-agent'
 import LocalJobRegistry from '@deepseek-ai/dsh-jobs-local'
 import * as ToolTasks from '@deepseek-ai/dsh-tool-jobs'
-import LocalActivityRegistry from '@deepseek-ai/dsh-activity-local'
 import { ShellExecutor } from '@deepseek-ai/dsh-shell'
 import type { ShellExecRequest, ShellExecSpec, ShellProcess, ShellRunResult } from '@deepseek-ai/dsh-shell'
 import * as ToolPwsh from '@deepseek-ai/dsh-tool-pwsh'
@@ -81,10 +80,9 @@ async function setup() {
   await ctx.plugin(AgentRegistry)
   await ctx.plugin(LocalJobRegistry)
   await ctx.plugin(ToolTasks)
-  await ctx.plugin(LocalActivityRegistry)
   await ctx.plugin(BashEnvPlugin)
   await ctx.plugin(FakePwsh)
-  await ctx.plugin(ToolPwsh, { activityPollMs: 5 })
+  await ctx.plugin(ToolPwsh, { recordPollMs: 5 })
   return { ctx, pwsh: ctx.shell as FakePwsh }
 }
 
@@ -108,8 +106,8 @@ async function until<T>(read: () => T | undefined, timeoutMs = 5_000): Promise<T
   throw new Error('condition not reached before timeout')
 }
 
-describe('background pwsh observation', () => {
-  it('mirrors observed streams with channels and settles with the mapped outcome', async () => {
+describe('background pwsh record', () => {
+  it('streams observed channels into the job record and settles with the mapped outcome', async () => {
     const { ctx, pwsh } = await setup()
     const stdout = { text: '' }
     const stderr = { text: '' }
@@ -118,29 +116,32 @@ describe('background pwsh observation', () => {
 
     await call(ctx, { command: 'Get-Progress', description: 'test command', run_in_background: true })
     const job = ctx.jobs.list()[0]
-    const row = await until(() => ctx.activities.list().find(item => item.correlation?.jobId === job!.id))
-    expect(row.kind).toBe('pwsh')
+    expect(job).toBeDefined()
+    expect(job!.kind).toBe('pwsh')
+    expect(job!.outputTotal).toBe(0)
 
     stdout.text = 'progress-line\n'
     stderr.text = 'warn-line\n'
     await until(() => {
-      const chunks = ctx.activities.read(row.id, 0).chunks
+      const chunks = ctx.jobs.readRecord(job!.id, 0).chunks
       return chunks.some(chunk => chunk.text.includes('progress-line'))
         && chunks.some(chunk => chunk.text.includes('warn-line'))
         ? true
         : undefined
     })
-    const chunks = ctx.activities.read(row.id, 0).chunks
+    const chunks = ctx.jobs.readRecord(job!.id, 0).chunks
     expect(chunks.find(chunk => chunk.text.includes('progress-line'))?.channel).toBe('stdout')
     expect(chunks.find(chunk => chunk.text.includes('warn-line'))?.channel).toBe('stderr')
 
     scripted.finish()
-    await until(() => ctx.activities.get(row.id).status === 'completed' ? true : undefined)
-    expect(ctx.activities.get(row.id).detail).toBe('exit code: 0')
+    await until(() => ctx.jobs.get(job!.id).status === 'completed' ? true : undefined)
+    expect(ctx.jobs.get(job!.id).detail).toBe('exit code: 0')
   })
 
-  it('a pump failure waits for process settlement before mapping the outcome', async () => {
+  it('a pump failure warns and never fakes a terminal state onto the running job', async () => {
     const { ctx, pwsh } = await setup()
+    const warn = vi.fn()
+    ctx.logger.warn = warn as never
     let resolveDone: () => void = () => {}
     const done = new Promise<void>((resolve) => { resolveDone = resolve })
     const proc: ShellProcess = {
@@ -157,70 +158,36 @@ describe('background pwsh observation', () => {
     }
     pwsh.backgroundHandler = () => proc
     await call(ctx, { command: 'Get-Broken', description: 'test command', run_in_background: true })
-    const row = await until(() => ctx.activities.list()[0])
-    // The pump rejected on its first poll; the still-running process must not
-    // be mapped to a terminal state until it actually settles.
+    const job = ctx.jobs.list()[0]
+    await until(() => warn.mock.calls.some(args => String(args[0]).includes('record observation pump')) ? true : undefined)
+    // The pump already failed; the job must wait for real settlement rather
+    // than freezing a fake terminal state onto the running process.
     await new Promise(resolve => setTimeout(resolve, 50))
-    expect(ctx.activities.list()[0]?.status).toBe('running')
+    expect(ctx.jobs.get(job!.id).status).toBe('running')
     proc.status = 'completed'
     proc.exitCode = 0
     resolveDone()
-    await until(() => ctx.activities.list()[0]?.status !== 'running' ? true : undefined)
-    expect(ctx.activities.list()[0]).toMatchObject({ status: 'completed', detail: 'exit code: 0' })
-    expect(String(row.id)).toContain('pwsh')
+    await until(() => ctx.jobs.get(job!.id).status !== 'running' ? true : undefined)
+    expect(ctx.jobs.get(job!.id)).toMatchObject({ status: 'completed', detail: 'exit code: 0' })
   })
 
-  it('a throwing end() is swallowed by the settlement-mapping catch', async () => {
-    const ctx = new Context()
-    await ctx.plugin(SystemPrompt)
-    await ctx.plugin(ToolRuntime)
-    await ctx.plugin(AgentRegistry)
-    await ctx.plugin(LocalJobRegistry)
-    await ctx.plugin(ToolTasks)
-    await ctx.plugin({
-      name: 'end-throwing-activities-probe',
-      apply(child: Context) {
-        child.provide('activities', {
-          open: () => ({
-            id: 'pwsh-1',
-            append() {},
-            updateDetail() {},
-            end() { throw new Error('end boom') },
-          }),
-        })
-      },
-    })
-    await ctx.plugin(BashEnvPlugin)
-    await ctx.plugin(FakePwsh)
-    await ctx.plugin(ToolPwsh, { activityPollMs: 5 })
-    const pwsh = ctx.shell as FakePwsh
-    const warn = vi.fn()
-    ctx.logger.warn = warn as never
-    const handle = observableProcess({ stdout: { text: '' }, stderr: { text: '' } })
-    pwsh.backgroundHandler = () => handle.proc
-    await call(ctx, { command: 'Get-Doom', description: 'test command', run_in_background: true })
-    handle.finish()
-    await until(() => warn.mock.calls.some(args => String(args[0]).includes('activity settlement mapping')) ? true : undefined)
-  })
-
-  it('a backend without observed readers still yields a status-only process', async () => {
+  it('a backend without observed readers still yields a status-only record', async () => {
     const { ctx, pwsh } = await setup()
     const scripted = observableProcess()
     pwsh.backgroundHandler = () => scripted.proc
 
     await call(ctx, { command: 'Start-Job', description: 'test command', run_in_background: true })
     const job = ctx.jobs.list()[0]
-    const row = await until(() => ctx.activities.list().find(item => item.correlation?.jobId === job!.id))
-    expect(ctx.activities.read(row.id, 0).chunks).toEqual([])
+    expect(ctx.jobs.readRecord(job!.id, 0).chunks).toEqual([])
 
     scripted.finish()
-    await until(() => ctx.activities.get(row.id).status === 'completed' ? true : undefined)
-    expect(ctx.activities.get(row.id).outputTotal).toBe(0)
+    await until(() => ctx.jobs.get(job!.id).status === 'completed' ? true : undefined)
+    expect(ctx.jobs.get(job!.id).outputTotal).toBe(0)
   })
 })
 
-describe('owned and degraded observation (pwsh)', () => {
-  it('mirrors an owned background run under the owning session', async () => {
+describe('owned background record (pwsh)', () => {
+  it('keeps an owned background run record under the owning session', async () => {
     const { ctx, pwsh } = await setup()
     const owner = {
       id: SessionId('pwsh-observe-owner'),
@@ -240,40 +207,10 @@ describe('owned and degraded observation (pwsh)', () => {
     })
     const job = ctx.jobs.list(owner)[0]
     expect(job).toBeDefined()
-    const row = await until(() => ctx.activities.list(owner).find(item => item.correlation?.jobId === job!.id))
-    expect(row.ownerSession).toBe(owner.id)
+    expect(job!.ownerSession).toBe(owner.id)
+    expect(() => ctx.jobs.readRecord(job!.id, 0)).toThrow(/belongs to another session/)
+    expect(ctx.jobs.readRecord(job!.id, 0, owner).chunks).toEqual([])
     scripted.finish()
     await until(() => ctx.jobs.get(job!.id, owner).status === 'completed' ? true : undefined)
   })
-
-  it('swallows a throwing observation registry and keeps the job alive', async () => {
-    // The FakePwsh composition with a synchronously throwing open(): the
-    // failure lands in the outer observation catch and the job runs on.
-    const ctx = new Context()
-    await ctx.plugin(SystemPrompt)
-    await ctx.plugin(ToolRuntime)
-    await ctx.plugin(AgentRegistry)
-    await ctx.plugin(LocalJobRegistry)
-    await ctx.plugin(ToolTasks)
-    await ctx.plugin({
-      name: 'broken-activities-probe',
-      apply(child: Context) {
-        child.provide('activities', { open() { throw new Error('observation boom') } })
-      },
-    })
-    await ctx.plugin(BashEnvPlugin)
-    await ctx.plugin(FakePwsh)
-    const warn = vi.fn()
-    ctx.logger.warn = warn as never
-    await ctx.plugin(ToolPwsh, { activityPollMs: 5 })
-    const pwsh = ctx.shell as FakePwsh
-    const scripted = observableProcess()
-    pwsh.backgroundHandler = () => scripted.proc
-    await call(ctx, { command: 'Get-Slow', description: 'test command', run_in_background: true })
-    expect(warn).toHaveBeenCalledWith(expect.stringContaining('activity observation unavailable'))
-    const job = ctx.jobs.list()[0]
-    expect(job?.status).toBe('running')
-    scripted.finish()
-    await until(() => ctx.jobs.get(job!.id).status === 'completed' ? true : undefined)
-  })
 })

packages/shell/tool-pwsh/tsconfig.json

@@ -52,9 +52,6 @@
     },
     {
       "path": "../../runtime-diagnostics/invariants"
-    },
-    {
-      "path": "../../activity/activity"
     }
   ]
 }

packages/workflow/tool-workflow/README.md

@@ -68,7 +68,7 @@ The consumer owns the model-facing schema, the `tool:<toolName>` system-prompt g
 
 For a root transport execution (`exec.parent` absent), the tool projects the run into the calling Agent's Session with four log-only events: run-start after `start()` returns, member starts and endings filtered by `run.id`, then run-end only after the result is available and disposal reaches quiescence. Nested transport calls execute normally but write no record. The first failed Session append disables later recording for that run with one warning, leaving either no record or a legal continuous prefix without changing the tool result or cleanup. The package invariant rejects duplicate starts, unpaired members, terminal events with open members, and updates after run-end on both cold load and live append, while accepting missing terminal suffixes.
 
-Beside the durable record, each recorded top-level run is mirrored best-effort into the optional `ctx.activities` registry as a `workflow` activity: the otherwise-unconsumed `workflow/phase` and `workflow/log` events plus member lifecycle lines stream into it, so the Web client shows the run's live narration the session log deliberately does not carry per line. Without the registry, or on any mirror failure, the run is unaffected.
+The engine's `workflow/phase` and `workflow/log` events have no per-line durable or observer surface from this tool: the session log deliberately records run and member lifecycle only, and the Web transcript derives from those records.
 
 ### Render intent
 
@@ -80,7 +80,6 @@ Decided up front per the [render-intent Agent Note](../../../.agents/notes/imple
 |---|---|
 | [`src/index.ts`](src/index.ts) | Plugin entry: tool registration, run lifecycle, recorder wiring |
 | [`src/types.ts`](src/types.ts) | The four log-only record event payloads and their `SessionEventMap` declaration |
-| [`src/activity.ts`](src/activity.ts) | Best-effort activity mirror for top-level runs |
 | [`src/invariant.ts`](src/invariant.ts) | Invariant companion: durable workflow-record protocol validation |
 
 </details>

packages/workflow/tool-workflow/package.json

@@ -38,7 +38,6 @@
   "license": "MIT",
   "peerDependencies": {
     "@deepseek-ai/cordis": "workspace:^",
-    "@deepseek-ai/dsh-activity": "workspace:^",
     "@deepseek-ai/dsh-agent": "workspace:^",
     "@deepseek-ai/dsh-invariants": "workspace:^",
     "@deepseek-ai/dsh-llm": "workspace:^",
@@ -52,8 +51,6 @@
   },
   "devDependencies": {
     "@deepseek-ai/cordis": "workspace:^",
-    "@deepseek-ai/dsh-activity": "workspace:^",
-    "@deepseek-ai/dsh-activity-local": "workspace:^",
     "@deepseek-ai/dsh-agent": "workspace:^",
     "@deepseek-ai/dsh-invariants": "workspace:^",
     "@deepseek-ai/dsh-llm": "workspace:^",

packages/workflow/tool-workflow/src/activity.ts

@@ -1,119 +0,0 @@
-/**
- * Best-effort activity mirror for top-level workflow runs: opens one
- * `workflow` activity per run and streams the engine's `workflow/phase`,
- * `workflow/log`, and member lifecycle events into it as text lines, so the
- * Web client shows live progress the session log deliberately does not carry
- * per line. Without a `ctx.activities` registry every operation is a no-op,
- * and ANY mirror failure is logged and swallowed — the run is unaffected.
- *
- * @module @deepseek-ai/dsh-tool-workflow/activity
- */
-
-import type { Context } from '@deepseek-ai/cordis'
-import type { Agent } from '@deepseek-ai/dsh-agent'
-import type { ActivityHandle } from '@deepseek-ai/dsh-activity'
-import type { WorkflowRun, WorkflowRunId, WorkflowStopReason } from '@deepseek-ai/dsh-workflow'
-
-declare module '@deepseek-ai/dsh-activity' {
-  interface ActivityKindMap {
-    workflow: 'workflow'
-  }
-}
-
-/** Activity taps for the three points the tool already tracks a run at. */
-export interface WorkflowActivityMirror {
-  /**
-   * Open the run's activity. Call once per recorded top-level run.
-   * @param run - the started run (id and meta name).
-   * @param owner - the calling agent, when the call has one.
-   */
-  start(run: WorkflowRun, owner: Agent | undefined): void
-  /**
-   * End the run's activity with the mapped stop reason.
-   * @param runId - the settled run.
-   * @param stopReason - how the run stopped.
-   */
-  finish(runId: WorkflowRunId, stopReason: WorkflowStopReason): void
-  /**
-   * Drop tracking for a run that ends without a settled reason. First-wins
-   * with {@link finish}: after a normal finish this is inert.
-   * @param runId - the abandoned run.
-   */
-  abandon(runId: WorkflowRunId): void
-}
-
-/** Map a run's stop reason onto the activity outcome vocabulary. */
-function outcomeOf(stopReason: WorkflowStopReason): { status: 'completed' | 'failed' | 'killed'; detail: string } {
-  switch (stopReason) {
-    case 'completed': return { status: 'completed', detail: 'completed' }
-    case 'cancelled': return { status: 'killed', detail: 'cancelled' }
-    case 'error': return { status: 'failed', detail: 'error' }
-  }
-}
-
-/**
- * Create the run-to-activity mirror and subscribe the engine's live progress
- * events for the runs it tracks.
- * @param ctx - plugin context used for the optional registry and warning logs.
- * @returns the mirror taps the tool wires beside its durable recorder.
- */
-export function createWorkflowActivityMirror(ctx: Context): WorkflowActivityMirror {
-  const active = new Map<WorkflowRunId, ActivityHandle>()
-  const contained = (operation: string, act: () => void): void => {
-    try {
-      act()
-    } catch (error: unknown) {
-      ctx.logger.warn(`workflow activity mirror ${operation} failed: ${String(error)}`)
-    }
-  }
-
-  ctx.on('workflow/phase', (info, title) => {
-    const handle = active.get(info.id)
-    if (handle === undefined) return
-    contained('phase', () => {
-      handle.updateDetail(title)
-      handle.append(`▸ ${title}\n`)
-    })
-  })
-  ctx.on('workflow/log', (info, message) => {
-    const handle = active.get(info.id)
-    if (handle === undefined) return
-    contained('log', () => { handle.append(`${message}\n`) })
-  })
-  ctx.on('workflow/agent-start', (info, agent) => {
-    const handle = active.get(info.id)
-    if (handle === undefined) return
-    contained('agent-start', () => { handle.append(`agent #${agent.seq} ${agent.label} started\n`) })
-  })
-  ctx.on('workflow/agent-end', (info, agent) => {
-    const handle = active.get(info.id)
-    if (handle === undefined) return
-    contained('agent-end', () => { handle.append(`agent #${agent.seq} ${agent.outcome}\n`) })
-  })
-
-  return {
-    start(run, owner) {
-      contained('open', () => {
-        const activities = ctx.get('activities')
-        if (activities === undefined) return
-        active.set(run.id, activities.open({
-          kind: 'workflow',
-          label: run.meta.name,
-          ...owner !== undefined ? { owner } : {},
-        }))
-      })
-    },
-    finish(runId, stopReason) {
-      const handle = active.get(runId)
-      active.delete(runId)
-      if (handle === undefined) return
-      contained('finish', () => { handle.end(outcomeOf(stopReason)) })
-    },
-    abandon(runId) {
-      const handle = active.get(runId)
-      active.delete(runId)
-      if (handle === undefined) return
-      contained('abandon', () => { handle.end({ status: 'killed', detail: 'run abandoned' }) })
-    },
-  }
-}

packages/workflow/tool-workflow/src/index.ts

@@ -24,7 +24,6 @@ import type {
   ToolWorkflowAgentEndData, ToolWorkflowAgentStartData,
   ToolWorkflowRunEndData, ToolWorkflowRunStartData,
 } from './types.ts'
-import { createWorkflowActivityMirror } from './activity.ts'
 
 export const name = 'tool-workflow'
 export const inject = ['tools', 'workflowEngine', 'systemPrompt']
@@ -207,7 +206,6 @@ export function apply(ctx: Context, config: Config): void {
   // fields; the assertion records that resolution, not a hidden fallback.
   const { toolName, maxResultChars } = config as ResolvedConfig
   const recorder = createWorkflowRecorder(ctx)
-  const mirror = createWorkflowActivityMirror(ctx)
   // Usage policy ships with the tool (the master convention: tool guidance
   // lives in tool plugins as prompt sections, not in the deployment persona).
   ctx.systemPrompt.section({
@@ -294,7 +292,6 @@ export function apply(ctx: Context, config: Config): void {
       // worker messages, after start() returns and this run record is active.
       if (recordsRun) {
         recorder.start(parent.session, run)
-        mirror.start(run, parent)
       }
 
       // Bridge the tool's abort signal to the run: if the parent step is aborted while the
@@ -327,12 +324,10 @@ export function apply(ctx: Context, config: Config): void {
             /* v8 ignore next -- WorkflowRun.result never rejects by contract, so result is assigned before finally. */
             if (result === undefined) throw new Error('workflow run settled without a result')
             recorder.finish(run.id, result.stopReason)
-            mirror.finish(run.id, result.stopReason)
           }
         } finally {
           if (recordsRun) {
             recorder.abandon(run.id)
-            mirror.abandon(run.id)
           }
         }
       }

packages/workflow/tool-workflow/tests/activity-mirror.spec.ts

@@ -1,129 +0,0 @@
-import { describe, expect, it, vi } from 'vitest'
-import { Context } from '@deepseek-ai/cordis'
-import AgentRegistry from '@deepseek-ai/dsh-agent'
-import type { Agent } from '@deepseek-ai/dsh-agent'
-import LocalActivityRegistry from '@deepseek-ai/dsh-activity-local'
-import { SessionId } from '@deepseek-ai/dsh-session'
-import type { WorkflowRun, WorkflowRunId, WorkflowRunInfo } from '@deepseek-ai/dsh-workflow'
-import { createWorkflowActivityMirror } from '../src/activity.ts'
-
-function runStub(id = 'run-1'): { run: WorkflowRun; info: WorkflowRunInfo } {
-  const info = { id: id as WorkflowRunId, meta: { name: 'docs-sweep' } } as WorkflowRunInfo
-  return {
-    run: { id: id as WorkflowRunId, meta: { name: 'docs-sweep' } } as WorkflowRun,
-    info,
-  }
-}
-
-function retainedText(ctx: Context): string {
-  const row = ctx.activities.list()[0]
-  if (row === undefined) throw new Error('no mirrored activity')
-  return ctx.activities.read(row.id, 0).chunks.map(chunk => chunk.text).join('')
-}
-
-describe('workflow activity mirror', () => {
-  it('opens one workflow activity per run and streams phase, log, and member lines', async () => {
-    const ctx = new Context()
-    await ctx.plugin(LocalActivityRegistry)
-    const mirror = createWorkflowActivityMirror(ctx)
-    const { run, info } = runStub()
-    mirror.start(run, undefined)
-
-    const row = ctx.activities.list()[0]
-    expect(row?.kind).toBe('workflow')
-    expect(row?.label).toBe('docs-sweep')
-
-    ctx.emit('workflow/phase', info, 'Scan')
-    ctx.emit('workflow/log', info, '3 files queued')
-    ctx.emit('workflow/agent-start', info, { seq: 1, label: 'reader', childId: 'c1' } as never)
-    ctx.emit('workflow/agent-end', info, { seq: 1, label: 'reader', childId: 'c1', outcome: 'completed' } as never)
-    expect(ctx.activities.get(row!.id).detail).toBe('Scan')
-    expect(retainedText(ctx)).toBe('▸ Scan\n3 files queued\nagent #1 reader started\nagent #1 completed\n')
-
-    mirror.finish(run.id, 'completed')
-    const settled = ctx.activities.get(row!.id)
-    expect(settled.status).toBe('completed')
-    // The always-run abandon after a normal finish is inert.
-    mirror.abandon(run.id)
-    expect(ctx.activities.get(row!.id).status).toBe('completed')
-  })
-
-
-  it('opens an owned run under the owning agent session', async () => {
-    const ctx = new Context()
-    await ctx.plugin(AgentRegistry)
-    await ctx.plugin(LocalActivityRegistry)
-    const owner = {
-      id: SessionId('workflow-owner'),
-      session: { id: SessionId('workflow-owner') },
-      status: 'idle',
-      ctx,
-    } as unknown as Agent
-    ctx.agents.register(owner)
-    const mirror = createWorkflowActivityMirror(ctx)
-    const { run } = runStub('run-owned')
-    mirror.start(run, owner)
-    // The owned row is fenced to its owner's view.
-    const row = ctx.activities.list(owner)[0]
-    expect(row?.ownerSession).toBe(owner.id)
-    mirror.finish(run.id, 'completed')
-  })
-  it('maps cancelled and error stops, and ends an abandoned run as killed', async () => {
-    const ctx = new Context()
-    await ctx.plugin(LocalActivityRegistry)
-    const mirror = createWorkflowActivityMirror(ctx)
-
-    const cancelled = runStub('run-c')
-    mirror.start(cancelled.run, undefined)
-    mirror.finish(cancelled.run.id, 'cancelled')
-
-    const failed = runStub('run-e')
-    mirror.start(failed.run, undefined)
-    mirror.finish(failed.run.id, 'error')
-
-    const orphaned = runStub('run-o')
-    mirror.start(orphaned.run, undefined)
-    mirror.abandon(orphaned.run.id)
-
-    const byLabelOrder = ctx.activities.list()
-    expect(byLabelOrder.map(row => [row.status, row.detail])).toEqual([
-      ['killed', 'cancelled'],
-      ['failed', 'error'],
-      ['killed', 'run abandoned'],
-    ])
-  })
-
-  it('ignores events for untracked runs and stays inert without the registry', async () => {
-    const bare = new Context()
-    const inert = createWorkflowActivityMirror(bare)
-    const { run, info } = runStub()
-    // No registry: every tap is a no-op rather than a throw.
-    inert.start(run, undefined)
-    bare.emit('workflow/log', info, 'dropped')
-    inert.finish(run.id, 'completed')
-    inert.abandon(run.id)
-
-    const ctx = new Context()
-    await ctx.plugin(LocalActivityRegistry)
-    createWorkflowActivityMirror(ctx)
-    // Events for a run the mirror never opened change nothing.
-    ctx.emit('workflow/phase', info, 'ghost')
-    expect(ctx.activities.list()).toEqual([])
-  })
-
-  it('contains a throwing registry so the run path never breaks', async () => {
-    const ctx = new Context()
-    const warn = vi.fn()
-    ctx.logger.warn = warn as never
-    await ctx.plugin({
-      name: 'broken-activities-probe',
-      apply(child: Context) {
-        child.provide('activities', { open() { throw new Error('mirror boom') } })
-      },
-    })
-    const mirror = createWorkflowActivityMirror(ctx)
-    const { run } = runStub()
-    expect(() => { mirror.start(run, undefined) }).not.toThrow()
-    expect(warn).toHaveBeenCalledWith(expect.stringContaining('workflow activity mirror open failed'))
-  })
-})

packages/workflow/tool-workflow/tsconfig.json

@@ -37,9 +37,6 @@
     },
     {
       "path": "../../runtime-diagnostics/invariants"
-    },
-    {
-      "path": "../../activity/activity"
     }
   ]
 }