diff --git a/packages/shell/tool-bash/tests/observe.spec.ts b/packages/shell/tool-bash/tests/observe.spec.ts index 1a100cf833..714558227f 100644 --- a/packages/shell/tool-bash/tests/observe.spec.ts +++ b/packages/shell/tool-bash/tests/observe.spec.ts @@ -4,9 +4,11 @@ import { join } from 'node:path' import { describe, expect, it, vi } from 'vitest' import { Context } from '@deepseek-ai/cordis' import { ToolCallId } from '@deepseek-ai/dsh-llm' +import { SessionId } from '@deepseek-ai/dsh-session' import SystemPrompt from '@deepseek-ai/dsh-system-prompt' 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 * as ToolTasks from '@deepseek-ai/dsh-tool-jobs' import LocalActivityRegistry from '@deepseek-ai/dsh-activity-local' @@ -262,3 +264,28 @@ describe('background bash observation', () => { expect(read.snapshot.status).toBe('completed') }) }) + +describe('owned background observation', () => { + it('mirrors an owned background run under the owning session', async () => { + const ctx = await setup() + const owner = { + id: SessionId('observe-owner'), + session: { id: SessionId('observe-owner'), header: { cwd: process.cwd() } }, + status: 'idle', + ctx, + } as unknown as Agent + ctx.agents.register(owner) + await ctx.tools.execute({ + signal: testToolSignal, + callId: ToolCallId('observe-owned-1'), + name: 'bash', + arguments: { command: 'echo owned', description: 'test command', run_in_background: true }, + agent: owner, + }) + 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) + await until(() => ctx.jobs.get(job!.id, owner).status === 'completed' ? true : undefined) + }) +}) diff --git a/packages/shell/tool-pwsh/tests/observe.spec.ts b/packages/shell/tool-pwsh/tests/observe.spec.ts index 97f13b064e..c9024203fb 100644 --- a/packages/shell/tool-pwsh/tests/observe.spec.ts +++ b/packages/shell/tool-pwsh/tests/observe.spec.ts @@ -1,9 +1,11 @@ import { describe, expect, it, vi } from 'vitest' import { Context } from '@deepseek-ai/cordis' import { ToolCallId } from '@deepseek-ai/dsh-llm' +import { SessionId } from '@deepseek-ai/dsh-session' import SystemPrompt from '@deepseek-ai/dsh-system-prompt' 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 * as ToolTasks from '@deepseek-ai/dsh-tool-jobs' import LocalActivityRegistry from '@deepseek-ai/dsh-activity-local' @@ -216,3 +218,62 @@ describe('background pwsh observation', () => { expect(ctx.activities.get(row.id).outputTotal).toBe(0) }) }) + +describe('owned and degraded observation (pwsh)', () => { + it('mirrors an owned background run under the owning session', async () => { + const { ctx, pwsh } = await setup() + const owner = { + id: SessionId('pwsh-observe-owner'), + session: { id: SessionId('pwsh-observe-owner'), header: { cwd: process.cwd() } }, + status: 'idle', + ctx, + } as unknown as Agent + ctx.agents.register(owner) + const scripted = observableProcess() + pwsh.backgroundHandler = () => scripted.proc + await ctx.tools.execute({ + signal: testToolSignal, + callId: ToolCallId('pwsh-observe-owned'), + name: 'pwsh', + arguments: { command: 'Get-Slow', description: 'test command', run_in_background: true }, + agent: owner, + }) + 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) + 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) + }) +}) diff --git a/packages/workflow/tool-workflow/tests/activity-mirror.spec.ts b/packages/workflow/tool-workflow/tests/activity-mirror.spec.ts index 50788d453f..8f82cb2841 100644 --- a/packages/workflow/tool-workflow/tests/activity-mirror.spec.ts +++ b/packages/workflow/tool-workflow/tests/activity-mirror.spec.ts @@ -1,6 +1,9 @@ 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' @@ -45,6 +48,26 @@ describe('workflow activity mirror', () => { 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)