fix: make managed SSH sessions fast and resilient
GoodBuddy Agents / Validate Agent source (push) Canceled after 0s
GoodBuddy Agents / Build Linux arm64 Agent (push) Canceled after 0s
GoodBuddy Agents / Build Linux x64 Agent (push) Canceled after 0s
Cross-platform packages / Validate source (push) Canceled after 0s
Cross-platform packages / Authorize desktop packaging (push) Canceled after 0s
Cross-platform packages / linux arm64 (push) Canceled after 0s
Cross-platform packages / macos arm64 (push) Canceled after 0s
Cross-platform packages / windows arm64 (push) Canceled after 0s
Cross-platform packages / linux x64 (push) Canceled after 0s
Cross-platform packages / macos x64 (push) Canceled after 0s
Cross-platform packages / windows x64 (push) Canceled after 0s
Cross-platform packages / Publish GitHub and OSS release (push) Canceled after 0s

Managed SSH project selection repeatedly activated the remote Host and could
leave idle ACP bindings rejected as stale-controller after a controller
generation change. Host environment updates also coupled project state to
component installation evidence and presented a crowded control surface.

Project selection is now local-only, one Host connection serves compatible
project and conversation callers, idle bindings are safely replaced only
after proven controller takeover, and active prompts keep exact resume/replay.
Remote environment setup now uses explicit signed compound package
preparation with bounded transfer, rollback, recovery, and clearer settings.

Managed Hosts must be updated to separately distributed Agent 0.11.6 before
the daemon-side controller recovery fix takes effect. Desktop packages do not
embed the Agent or remote Runtime, and existing project data is preserved.

Release note: 托管 SSH 项目切换不再重复连接,同一 Host 可顺畅承载多个项目和对话;同时修复重连后的 stale-controller,并优化远程环境安装、更新与状态布局。
This commit is contained in:
mesalogo
2026-08-29 10:58:19 +08:00
parent eb947213c8
commit 459cc21e3d
117 changed files with 19876 additions and 9664 deletions
+242 -4
View File
@@ -1,4 +1,8 @@
import { EventEmitter } from 'node:events'
import { createHash } from 'node:crypto'
import { mkdtemp, rm, writeFile } from 'node:fs/promises'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { describe, expect, it, vi } from 'vitest'
import { BoundedStagedSftp } from './bounded-sftp'
@@ -153,11 +157,28 @@ function createSftp(root = '/safe/staging') {
open: vi.fn(
(
path: string,
_mode: string,
callback: (error?: Error, handle?: Buffer) => void
mode: string,
attributesOrCallback:
| { mode: number }
| ((error?: Error, handle?: Buffer) => void),
possibleCallback?: (
error?: Error,
handle?: Buffer
) => void
) => {
const entry = entries.get(path)
if (!entry) {
const callback =
typeof attributesOrCallback === 'function'
? attributesOrCallback
: possibleCallback!
let entry = entries.get(path)
if (mode === 'wx') {
if (entry) {
callback(new Error('destination exists'))
return
}
entry = { type: 'file', contents: Buffer.alloc(0) }
entries.set(path, entry)
} else if (!entry) {
callback(missingError())
return
}
@@ -166,6 +187,41 @@ function createSftp(root = '/safe/staging') {
callback(undefined, handle)
}
),
write: vi.fn(
(
handle: Buffer,
buffer: Buffer,
offset: number,
length: number,
position: number,
callback: (
error: Error | undefined,
bytesWritten: number
) => void
) => {
const entry = handles.get(handle.toString())
if (!entry) {
callback(missingError(), 0)
return
}
const requiredSize = position + length
if (
!entry.contents ||
entry.contents.byteLength < requiredSize
) {
const grown = Buffer.alloc(requiredSize)
entry.contents?.copy(grown)
entry.contents = grown
}
buffer.copy(
entry.contents,
position,
offset,
offset + length
)
callback(undefined, length)
}
),
fstat: vi.fn(
(
handle: Buffer,
@@ -221,6 +277,188 @@ function createSftp(root = '/safe/staging') {
}
describe('bounded staged SFTP', () => {
it('streams local files in bounded chunks and verifies both identities', async () => {
const directory = await mkdtemp(
join(tmpdir(), 'goodbuddy-sftp-upload-')
)
try {
const sourcePath = join(directory, 'package.gbagent')
const contents = Buffer.alloc(3 * 64 * 1024 + 17, 0x5a)
await writeFile(sourcePath, contents)
const { sftp, entries } = createSftp()
const staged = new BoundedStagedSftp(
sftp as never,
'/safe/staging',
{
maximumFileBytes: contents.byteLength,
maximumTotalBytes: contents.byteLength
}
)
await staged.uploadFile('package.gbagent', sourcePath, {
size: contents.byteLength,
sha256: createHash('sha256')
.update(contents)
.digest('hex')
})
expect(
entries.get('/safe/staging/package.gbagent')?.contents
).toEqual(contents)
expect(sftp.write).toHaveBeenCalledTimes(4)
expect(
Math.max(
...sftp.write.mock.calls.map(
([, , , length]) => length
)
)
).toBe(64 * 1024)
expect(
sftp.write.mock.calls.every(
([, buffer]) => buffer.byteLength <= 64 * 1024
)
).toBe(true)
expect(sftp.open).toHaveBeenCalledWith(
'/safe/staging/package.gbagent',
'wx',
{ mode: 0o600 },
expect.any(Function)
)
expect(sftp.fstat).toHaveBeenCalledTimes(2)
} finally {
await rm(directory, { recursive: true, force: true })
}
})
it('rejects local upload size and hash mismatches', async () => {
const directory = await mkdtemp(
join(tmpdir(), 'goodbuddy-sftp-upload-')
)
try {
const sourcePath = join(directory, 'package.gbagent')
const contents = Buffer.alloc(64 * 1024 + 1, 0x41)
await writeFile(sourcePath, contents)
const sizeHarness = createSftp()
const sizeSftp = new BoundedStagedSftp(
sizeHarness.sftp as never,
'/safe/staging',
{
maximumFileBytes: contents.byteLength,
maximumTotalBytes: contents.byteLength
}
)
await expect(
sizeSftp.uploadFile('size', sourcePath, {
size: contents.byteLength - 1,
sha256: createHash('sha256')
.update(contents)
.digest('hex')
})
).rejects.toThrow('预期的普通文件')
expect(sizeHarness.sftp.open).not.toHaveBeenCalled()
const hashHarness = createSftp()
const hashSftp = new BoundedStagedSftp(
hashHarness.sftp as never,
'/safe/staging',
{
maximumFileBytes: contents.byteLength,
maximumTotalBytes: contents.byteLength
}
)
await expect(
hashSftp.uploadFile('hash', sourcePath, {
size: contents.byteLength,
sha256: '0'.repeat(64)
})
).rejects.toThrow('SHA-256 校验失败')
expect(hashHarness.sftp.close).toHaveBeenCalledOnce()
} finally {
await rm(directory, { recursive: true, force: true })
}
})
it('cancels a streaming upload and rejects unsafe boundaries', async () => {
const directory = await mkdtemp(
join(tmpdir(), 'goodbuddy-sftp-upload-')
)
try {
const sourcePath = join(directory, 'package.gbagent')
const contents = Buffer.alloc(2 * 64 * 1024, 0x42)
await writeFile(sourcePath, contents)
const hash = createHash('sha256').update(contents).digest('hex')
const cancellationHarness = createSftp()
cancellationHarness.sftp.write.mockImplementation(
() => undefined
)
const cancellable = new BoundedStagedSftp(
cancellationHarness.sftp as never,
'/safe/staging',
{
maximumFileBytes: contents.byteLength,
maximumTotalBytes: contents.byteLength
}
)
const controller = new AbortController()
const upload = cancellable.uploadFile(
'package.gbagent',
sourcePath,
{ size: contents.byteLength, sha256: hash },
controller.signal
)
await vi.waitFor(() =>
expect(
cancellationHarness.sftp.write
).toHaveBeenCalledOnce()
)
controller.abort()
await expect(upload).rejects.toMatchObject({
name: 'AbortError'
})
expect(
cancellationHarness.sftp.end
).toHaveBeenCalledOnce()
const boundaryHarness = createSftp()
const bounded = new BoundedStagedSftp(
boundaryHarness.sftp as never,
'/safe/staging',
{
maximumFileBytes: contents.byteLength - 1,
maximumTotalBytes: contents.byteLength
}
)
await expect(
bounded.uploadFile('too-large', sourcePath, {
size: contents.byteLength,
sha256: hash
})
).rejects.toThrow('大小超过安全限制')
boundaryHarness.entries.set('/safe/staging/link', {
type: 'symbolic-link'
})
const ancestors = new BoundedStagedSftp(
boundaryHarness.sftp as never,
'/safe/staging',
{
maximumFileBytes: contents.byteLength,
maximumTotalBytes: contents.byteLength
}
)
await expect(
ancestors.uploadFile('link/package', sourcePath, {
size: contents.byteLength,
sha256: hash
})
).rejects.toThrow('符号链接')
expect(boundaryHarness.sftp.open).not.toHaveBeenCalled()
} finally {
await rm(directory, { recursive: true, force: true })
}
})
it('keeps writes inside staging and supports bounded metadata and reads', async () => {
const { sftp } = createSftp(
'/home/builder/.goodbuddy/staging/random'
+174 -1
View File
@@ -1,3 +1,6 @@
import { createHash } from 'node:crypto'
import { constants } from 'node:fs'
import { lstat as localLstat, open as localOpen } from 'node:fs/promises'
import type { SFTPWrapper, Stats } from 'ssh2'
const DEFAULT_OPERATION_TIMEOUT_MS = 30_000
@@ -5,6 +8,7 @@ const MAX_OPERATION_TIMEOUT_MS = 60_000
const MINIMUM_TRANSFER_BYTES_PER_SECOND = 64 * 1024
const MAX_TRANSFER_TIMEOUT_MS = 60 * 60_000
const DEFAULT_MAX_FILE_BYTES = 256 * 1024 * 1024
const MAX_MAXIMUM_FILE_BYTES = 512 * 1024 * 1024
const DEFAULT_MAX_TOTAL_BYTES = 1024 * 1024 * 1024
const DEFAULT_MAX_OPERATIONS = 2_048
const MAX_REMOTE_PATH_BYTES = 4_096
@@ -37,6 +41,11 @@ export type SftpEntryMetadata = {
export type AllowedSftpMode = (typeof ALLOWED_MODES)[number]
export type SftpUploadIdentity = {
size: number
sha256: string
}
export type StagedSftp = {
readonly stagingDirectory: string
mkdir(relativePath: string, signal?: AbortSignal): Promise<void>
@@ -45,6 +54,12 @@ export type StagedSftp = {
contents: Buffer,
signal?: AbortSignal
): Promise<void>
uploadFile(
relativePath: string,
sourcePath: string,
identity: SftpUploadIdentity,
signal?: AbortSignal
): Promise<void>
readFile(
relativePath: string,
signal?: AbortSignal
@@ -100,6 +115,7 @@ type SftpLike = Pick<
| 'stat'
| 'open'
| 'read'
| 'write'
| 'close'
| 'fstat'
| 'end'
@@ -246,7 +262,7 @@ export class BoundedStagedSftp implements StagedSftp {
this.maximumFileBytes = validatePositiveLimit(
limits.maximumFileBytes,
DEFAULT_MAX_FILE_BYTES,
DEFAULT_MAX_FILE_BYTES
MAX_MAXIMUM_FILE_BYTES
)
this.maximumTotalBytes = validatePositiveLimit(
limits.maximumTotalBytes,
@@ -316,6 +332,146 @@ export class BoundedStagedSftp implements StagedSftp {
)
}
async uploadFile(
relativePath: string,
sourcePath: string,
identity: SftpUploadIdentity,
signal?: AbortSignal
): Promise<void> {
const validated = validateRelativePath(relativePath)
const remotePath = this.resolveValidated(validated)
if (
!Number.isSafeInteger(identity.size) ||
identity.size < 0 ||
identity.size > this.maximumFileBytes
) {
throw new Error('SFTP 文件大小超过安全限制')
}
if (!/^[a-f0-9]{64}$/u.test(identity.sha256)) {
throw new Error('SFTP 文件 SHA-256 无效')
}
return this.enqueue(
signal,
async () => {
if (
this.totalBytes + identity.size >
this.maximumTotalBytes
) {
throw new Error('SFTP 传输总量超过安全限制')
}
await this.assertSafeAncestors(validated)
const pathMetadata = await localLstat(sourcePath)
if (
!pathMetadata.isFile() ||
pathMetadata.isSymbolicLink() ||
pathMetadata.size !== identity.size
) {
throw new Error('SFTP 本地上传源不是预期的普通文件')
}
const source = await localOpen(
sourcePath,
constants.O_RDONLY | (constants.O_NOFOLLOW ?? 0)
)
let remoteHandle: Buffer | undefined
const closeSourceOnAbort = (): void => {
void source.close().catch(() => undefined)
}
signal?.addEventListener('abort', closeSourceOnAbort, {
once: true
})
try {
const openedMetadata = await source.stat()
if (
!openedMetadata.isFile() ||
openedMetadata.size !== identity.size ||
openedMetadata.dev !== pathMetadata.dev ||
openedMetadata.ino !== pathMetadata.ino
) {
throw new Error(
'SFTP 本地上传源在打开期间发生变化'
)
}
remoteHandle = await this.call<Buffer>((done) => {
this.sftp.open(
remotePath,
'wx',
{ mode: 0o600 },
done
)
})
const initialRemoteMetadata = metadataFromStats(
await this.call<Stats>((done) => {
this.sftp.fstat(remoteHandle!, done)
})
)
this.assertType(initialRemoteMetadata, 'file')
if (initialRemoteMetadata.size !== 0) {
throw new Error('SFTP 上传目标初始大小无效')
}
const hash = createHash('sha256')
const buffer = Buffer.allocUnsafe(READ_CHUNK_BYTES)
let position = 0
while (true) {
signal?.throwIfAborted()
const { bytesRead } = await source.read(
buffer,
0,
buffer.byteLength,
position
)
if (bytesRead === 0) {
break
}
position += bytesRead
if (
position > identity.size ||
this.totalBytes + bytesRead >
this.maximumTotalBytes
) {
throw new Error('SFTP 文件大小超过安全限制')
}
const chunk = buffer.subarray(0, bytesRead)
hash.update(chunk)
await this.writeHandle(
remoteHandle,
chunk,
position - bytesRead
)
this.totalBytes += bytesRead
}
if (position !== identity.size) {
throw new Error('SFTP 本地上传源大小与预期不一致')
}
if (hash.digest('hex') !== identity.sha256) {
throw new Error('SFTP 本地上传源 SHA-256 校验失败')
}
const remoteMetadata = metadataFromStats(
await this.call<Stats>((done) => {
this.sftp.fstat(remoteHandle!, done)
})
)
this.assertType(remoteMetadata, 'file')
if (remoteMetadata.size !== identity.size) {
throw new Error('SFTP 上传目标大小与预期不一致')
}
} finally {
signal?.removeEventListener(
'abort',
closeSourceOnAbort
)
await Promise.allSettled([
source.close(),
remoteHandle
? this.closeHandle(remoteHandle)
: Promise.resolve()
])
}
},
this.transferTimeoutMs(identity.size)
)
}
async readFile(
relativePath: string,
signal?: AbortSignal
@@ -713,6 +869,23 @@ export class BoundedStagedSftp implements StagedSftp {
: contents.subarray(0, actualSize)
}
private async writeHandle(
handle: Buffer,
contents: Buffer,
position: number
): Promise<void> {
await this.call<void>((done) => {
this.sftp.write(
handle,
contents,
0,
contents.byteLength,
position,
done
)
})
}
private async closeHandle(handle: Buffer): Promise<void> {
try {
await this.call<void>((done) => {
File diff suppressed because it is too large Load Diff
File diff suppressed because it is too large Load Diff
+51
View File
@@ -0,0 +1,51 @@
import type {
CurrentSshConnectionTarget,
SshConnectionTarget
} from './ssh-host-store'
export function toCurrentSshConnectionTarget(
target: SshConnectionTarget
): CurrentSshConnectionTarget {
return {
hostId: target.host.id,
hostRevision: target.hostRevision,
hostKeyGeneration: target.hostKeyGeneration,
username: target.host.username
}
}
export function assertValidSshConnectionTarget(
requestedHostId: string,
target: SshConnectionTarget,
message = 'SSH 主机连接目标无效'
): void {
if (
target.host.id !== requestedHostId ||
target.hostRevision < 1 ||
target.hostKeyGeneration < 1 ||
!target.host.hostKey ||
target.host.hostKey.generation !== target.hostKeyGeneration
) {
throw new Error(message)
}
}
export function assertSshLeaseMatchesTarget(
lease: {
identity: {
hostId: string
hostRevision: number
hostKeyGeneration: number
}
},
target: SshConnectionTarget,
message: string
): void {
if (
lease.identity.hostId !== target.host.id ||
lease.identity.hostRevision !== target.hostRevision ||
lease.identity.hostKeyGeneration !== target.hostKeyGeneration
) {
throw new Error(message)
}
}
@@ -55,6 +55,10 @@ function createLease(
openAgentAttach: vi.fn(),
runAgentDoctor: vi.fn(),
runAgentBootstrapProbe: vi.fn(),
probeRemotePackageBootstrap: vi.fn(),
prepareRemotePackageBootstrap: vi.fn(),
commitRemotePackageBootstrap: vi.fn(),
cleanupRemotePackageBootstrap: vi.fn(),
runAgentLifecycleAction: vi.fn(),
runAgentRuntimeAction: vi.fn(),
openStagedSftp: vi.fn(),
+17 -47
View File
@@ -2,14 +2,17 @@ import type {
SshHostDirectoryBrowseResult
} from '../../shared/ssh-host-contracts'
import type {
CurrentSshConnectionTarget,
SshConnectionTarget,
SshHostStore
} from './ssh-host-store'
import type {
SshConnectionLease,
SshConnectionPool
} from './ssh-connection-pool'
import {
assertSshLeaseMatchesTarget,
assertValidSshConnectionTarget,
toCurrentSshConnectionTarget
} from './ssh-connection-target'
export type SshHostDirectoryBrowserOptions = {
sshHosts: Pick<
@@ -19,47 +22,6 @@ export type SshHostDirectoryBrowserOptions = {
sshPool: Pick<SshConnectionPool, 'acquire'>
}
function currentTarget(
target: SshConnectionTarget
): CurrentSshConnectionTarget {
return {
hostId: target.host.id,
hostRevision: target.hostRevision,
hostKeyGeneration: target.hostKeyGeneration,
username: target.host.username
}
}
function assertTarget(
requestedHostId: string,
target: SshConnectionTarget
): void {
if (
target.host.id !== requestedHostId ||
target.hostRevision < 1 ||
target.hostKeyGeneration < 1 ||
!target.host.hostKey ||
target.host.hostKey.generation !==
target.hostKeyGeneration
) {
throw new Error('SSH 主机连接目标无效')
}
}
function assertLeaseIdentity(
target: SshConnectionTarget,
lease: SshConnectionLease
): void {
if (
lease.identity.hostId !== target.host.id ||
lease.identity.hostRevision !== target.hostRevision ||
lease.identity.hostKeyGeneration !==
target.hostKeyGeneration
) {
throw new Error('SSH 目录浏览连接身份不匹配')
}
}
export class SshHostDirectoryBrowser {
readonly #sshHosts: SshHostDirectoryBrowserOptions['sshHosts']
readonly #sshPool: SshHostDirectoryBrowserOptions['sshPool']
@@ -78,18 +40,26 @@ export class SshHostDirectoryBrowser {
const target =
await this.#sshHosts.resolveConnectionTarget(hostId)
signal?.throwIfAborted()
assertTarget(hostId, target)
const expected = currentTarget(target)
assertValidSshConnectionTarget(hostId, target)
const expected = toCurrentSshConnectionTarget(target)
this.#sshHosts.assertConnectionTargetCurrent(expected)
let lease: SshConnectionLease | undefined
try {
lease = await this.#sshPool.acquire(target, signal)
signal?.throwIfAborted()
assertLeaseIdentity(target, lease)
assertSshLeaseMatchesTarget(
lease,
target,
'SSH 目录浏览连接身份不匹配'
)
const listing = await lease.listDirectories(path, signal)
signal?.throwIfAborted()
assertLeaseIdentity(target, lease)
assertSshLeaseMatchesTarget(
lease,
target,
'SSH 目录浏览连接身份不匹配'
)
this.#sshHosts.assertConnectionTargetCurrent(expected)
return listing
} finally {
+273 -22
View File
@@ -2,8 +2,12 @@ import { describe, expect, it, vi } from 'vitest'
import type { StagedSftp } from './bounded-sftp'
import type {
SshConnectionLease,
SshConnectionPoolTarget
SshConnectionPoolTarget,
SshRemotePackageBootstrapLease
} from './ssh-connection-pool'
import type {
SshRemotePackageBootstrapProbeResult
} from './ssh-remote-package-bootstrap'
import {
SshHostRemoteEnvironmentInspector
} from './ssh-host-remote-environment'
@@ -57,11 +61,37 @@ function missing(): Error & { code: number } {
return Object.assign(new Error('missing'), { code: 2 })
}
function harness(registries: {
agent?: unknown
runtime?: unknown
} = {}) {
type HarnessOptions = {
registries?: {
agent?: unknown
runtime?: unknown
}
candidate?: {
source: 'github' | 'mirror'
size: number
sha256: string
urls: readonly string[]
} | null
candidateError?: Error
candidateFailure?: {
reason: 'package-unavailable' | 'probe-failed'
source: 'github' | 'mirror' | null
packageSize: number | null
}
bootstrapResult?: SshRemotePackageBootstrapProbeResult
bootstrapError?: Error
bootstrapIdentity?: {
hostId: string
hostRevision: number
hostKeyGeneration: number
authenticationIdentity: string
}
}
function harness(options: HarnessOptions = {}) {
const registries = options.registries ?? {}
const release = vi.fn()
const bootstrapRelease = vi.fn()
const close = vi.fn()
const readFile = vi.fn(async (path: string) => {
const value =
@@ -96,25 +126,66 @@ function harness(registries: {
openStagedSftp: vi.fn(async () => sftp),
release
} as unknown as SshConnectionLease
const probe = vi.fn(async () => {
if (options.bootstrapError) {
throw options.bootstrapError
}
return options.bootstrapResult ?? { available: true as const }
})
const prepare = vi.fn()
const commit = vi.fn()
const bootstrapLease = {
identity: options.bootstrapIdentity ?? lease.identity,
isUsable: vi.fn(() => true),
probe,
prepare,
commit,
cleanup: vi.fn(),
release: bootstrapRelease
} as unknown as SshRemotePackageBootstrapLease
const assertConnectionTargetCurrent = vi.fn()
const acquireRemotePackageBootstrap = vi.fn(
async () => bootstrapLease
)
const loadRemoteEnvironmentCatalog = vi.fn(async () => {
const candidate =
options.candidateError
? null
: options.candidate === undefined
? {
source: 'mirror' as const,
size: 12_345,
sha256: 'a'.repeat(64),
urls: [
'https://goodbuddy.oss-cn-beijing.aliyuncs.com/agent.zip'
]
}
: options.candidate
return {
expected: {
agent: { version: '0.11.1' },
runtimes: [{
runtimeId: 'opencode' as const,
provider: 'opencode' as const,
version: '1.18.9'
}]
},
candidate,
candidateFailure: options.candidateFailure
}
})
const inspector = new SshHostRemoteEnvironmentInspector({
sshHosts: {
resolveConnectionTarget: vi.fn(async () => currentTarget),
assertConnectionTargetCurrent
},
sshPool: {
acquire: vi.fn(async () => lease)
acquire: vi.fn(async () => lease),
acquireRemotePackageBootstrap
},
agentRuntimeLockPath: 'unused',
remoteRuntimeLockPath: 'unused',
loadExpectedCatalog: async () => ({
agent: { version: '0.11.1' },
runtimes: [{
runtimeId: 'opencode',
provider: 'opencode',
version: '1.18.9'
}]
}),
loadRemoteEnvironmentCatalog,
now: () => new Date('2030-01-01T00:00:00.000Z')
})
return {
@@ -122,13 +193,19 @@ function harness(registries: {
lease,
close,
release,
bootstrapRelease,
probe,
prepare,
commit,
acquireRemotePackageBootstrap,
loadRemoteEnvironmentCatalog,
assertConnectionTargetCurrent
}
}
describe('SshHostRemoteEnvironmentInspector', () => {
it('reports installed Agent and Runtime versions against the desktop catalog', async () => {
const value = harness({
const value = harness({ registries: {
agent: {
formatVersion: 1,
minimumTrustedReleaseSequence: 1,
@@ -153,7 +230,7 @@ describe('SshHostRemoteEnvironmentInspector', () => {
signingKeyId: 'test-key'
}]
}
})
} })
await expect(value.inspector.inspect(hostId)).resolves.toEqual({
hostId,
@@ -182,20 +259,26 @@ describe('SshHostRemoteEnvironmentInspector', () => {
version: '1.18.9',
architecture: 'x64'
}
}]
}],
remoteDownload: {
available: true,
source: 'mirror',
packageSize: 12_345
}
})
expect(value.close).toHaveBeenCalledOnce()
expect(value.release).toHaveBeenCalledOnce()
expect(value.assertConnectionTargetCurrent).toHaveBeenCalledTimes(2)
expect(value.bootstrapRelease).toHaveBeenCalledOnce()
expect(value.assertConnectionTargetCurrent).toHaveBeenCalledTimes(5)
})
it('reports an older Agent and a missing Runtime without installing either', async () => {
const value = harness({
const value = harness({ registries: {
agent: {
formatVersion: 1,
current: agentEntry('0.11.0', 'b')
}
})
} })
await expect(value.inspector.inspect(hostId)).resolves.toMatchObject({
agent: {
@@ -210,15 +293,17 @@ describe('SshHostRemoteEnvironmentInspector', () => {
})
expect(value.lease.runAgentBootstrapProbe).toHaveBeenCalledOnce()
expect(value.lease.openStagedSftp).toHaveBeenCalledOnce()
expect(value.prepare).not.toHaveBeenCalled()
expect(value.commit).not.toHaveBeenCalled()
})
it('rejects corrupt remote metadata and still closes every lease', async () => {
const value = harness({
const value = harness({ registries: {
agent: {
formatVersion: 1,
current: { command: 'untrusted' }
}
})
} })
await expect(value.inspector.inspect(hostId)).rejects.toThrow(
'元数据已损坏'
@@ -226,4 +311,170 @@ describe('SshHostRemoteEnvironmentInspector', () => {
expect(value.close).toHaveBeenCalledOnce()
expect(value.release).toHaveBeenCalledOnce()
})
it.each([
['missing-curl', 'missing-tools'],
['missing-sha256sum', 'missing-tools'],
['missing-unzip', 'missing-tools'],
['bootstrap-tools-unavailable', 'missing-tools'],
['managed-path-unavailable', 'home-unwritable'],
['insufficient-disk-space', 'insufficient-disk-space'],
['download-unavailable', 'source-unreachable']
] as const)(
'maps direct probe reason %s to %s',
async (probeReason, reason) => {
const value = harness({
bootstrapResult: {
available: false,
reason: probeReason
}
})
await expect(value.inspector.inspect(hostId)).resolves.toMatchObject({
agent: { state: 'not-installed' },
runtimes: [{ state: 'not-installed' }],
remoteDownload: {
available: false,
source: 'mirror',
packageSize: 12_345,
reason
}
})
expect(value.bootstrapRelease).toHaveBeenCalledOnce()
}
)
it('sanitizes catalog failures without hiding version status', async () => {
const value = harness({
candidateError: new Error(
'private catalog URL and credential details'
)
})
await expect(value.inspector.inspect(hostId)).resolves.toMatchObject({
agent: { state: 'not-installed' },
runtimes: [{ state: 'not-installed' }],
remoteDownload: {
available: false,
source: null,
packageSize: null,
reason: 'package-unavailable'
}
})
expect(value.acquireRemotePackageBootstrap).not.toHaveBeenCalled()
})
it('probes and reports an available signed format-v1 candidate', async () => {
const value = harness({
candidate: {
source: 'github',
size: 99,
sha256: 'b'.repeat(64),
urls: ['https://github.com/example/package.zip']
}
})
await expect(value.inspector.inspect(hostId)).resolves.toMatchObject({
remoteDownload: {
available: true,
source: 'github',
packageSize: 99
}
})
expect(value.acquireRemotePackageBootstrap).toHaveBeenCalledOnce()
expect(value.probe).toHaveBeenCalledWith(
{
operationId: expect.any(String),
urls: ['https://github.com/example/package.zip'],
size: 99,
sha256: 'b'.repeat(64)
},
{ signal: undefined }
)
})
it('reports a missing compatible package without source metadata', async () => {
const value = harness({ candidate: null })
await expect(value.inspector.inspect(hostId)).resolves.toMatchObject({
remoteDownload: {
available: false,
source: null,
packageSize: null,
reason: 'package-unavailable'
}
})
})
it('preserves a catalog candidate probe failure and its source metadata', async () => {
const value = harness({
candidate: null,
candidateFailure: {
reason: 'probe-failed',
source: 'mirror',
packageSize: 12_345
}
})
await expect(value.inspector.inspect(hostId)).resolves.toMatchObject({
remoteDownload: {
available: false,
source: 'mirror',
packageSize: 12_345,
reason: 'probe-failed'
}
})
expect(value.acquireRemotePackageBootstrap).not.toHaveBeenCalled()
})
it('reports direct bootstrap probe failures without hiding version status', async () => {
const value = harness({
bootstrapError: new Error('sensitive SSH diagnostic')
})
await expect(value.inspector.inspect(hostId)).resolves.toMatchObject({
agent: { state: 'not-installed' },
remoteDownload: {
available: false,
source: 'mirror',
packageSize: 12_345,
reason: 'probe-failed'
}
})
expect(value.bootstrapRelease).toHaveBeenCalledOnce()
})
it('rejects a changed Host identity and releases both leases', async () => {
const value = harness({
bootstrapIdentity: {
hostId,
hostRevision: 4,
hostKeyGeneration: 2,
authenticationIdentity: 'b'.repeat(64)
}
})
await expect(value.inspector.inspect(hostId)).rejects.toThrow(
'连接身份不匹配'
)
expect(value.bootstrapRelease).toHaveBeenCalledOnce()
expect(value.release).toHaveBeenCalledOnce()
})
it('honors cancellation while loading the remote environment catalog', async () => {
const controller = new AbortController()
const value = harness()
value.loadRemoteEnvironmentCatalog.mockImplementationOnce(
async () => {
controller.abort()
throw controller.signal.reason
}
)
await expect(
value.inspector.inspect(hostId, controller.signal)
).rejects.toMatchObject({ name: 'AbortError' })
expect(value.acquireRemotePackageBootstrap).not.toHaveBeenCalled()
expect(value.release).toHaveBeenCalledOnce()
})
})
+202 -56
View File
@@ -1,3 +1,4 @@
import { randomUUID } from 'node:crypto'
import { readFile } from 'node:fs/promises'
import {
agentRuntimeLockSchema,
@@ -23,10 +24,19 @@ import type {
SshConnectionTarget,
SshHostStore
} from './ssh-host-store'
import {
assertSshLeaseMatchesTarget,
assertValidSshConnectionTarget,
toCurrentSshConnectionTarget
} from './ssh-connection-target'
import type {
SshConnectionLease,
SshRemotePackageBootstrapLease,
SshConnectionPool
} from './ssh-connection-pool'
import type {
SshRemotePackageBootstrapUnavailableReason
} from './ssh-remote-package-bootstrap'
const AGENT_REGISTRY_PATH = '.goodbuddy/agent/registry.json'
const RUNTIME_REGISTRY_PATH = '.goodbuddy/runtimes/registry.json'
@@ -43,38 +53,63 @@ type ExpectedCatalog = {
}>
}
type RemoteInstallCandidate = {
source: 'github' | 'mirror'
size: number
sha256: string
urls: readonly string[]
}
type RemoteEnvironmentCatalog = {
expected: ExpectedCatalog
candidate: RemoteInstallCandidate | null
candidateFailure?: {
reason: 'package-unavailable' | 'probe-failed'
source: 'github' | 'mirror' | null
packageSize: number | null
}
}
export type SshHostRemoteEnvironmentInspectorOptions = {
sshHosts: Pick<
SshHostStore,
'resolveConnectionTarget' | 'assertConnectionTargetCurrent'
>
sshPool: Pick<SshConnectionPool, 'acquire'>
sshPool: Pick<
SshConnectionPool,
'acquire' | 'acquireRemotePackageBootstrap'
>
agentRuntimeLockPath: string
remoteRuntimeLockPath: string
loadExpectedCatalog?: (
architecture: 'x64' | 'arm64'
) => Promise<ExpectedCatalog>
loadRemoteEnvironmentCatalog?: (
architecture: 'x64' | 'arm64',
options: { signal?: AbortSignal }
) => Promise<RemoteEnvironmentCatalog>
now?: () => Date
}
export class SshHostRemoteEnvironmentInspector {
readonly #sshHosts: SshHostRemoteEnvironmentInspectorOptions['sshHosts']
readonly #sshPool: SshHostRemoteEnvironmentInspectorOptions['sshPool']
readonly #loadExpectedCatalog: (
architecture: 'x64' | 'arm64'
) => Promise<ExpectedCatalog>
readonly #loadRemoteEnvironmentCatalog: NonNullable<
SshHostRemoteEnvironmentInspectorOptions[
'loadRemoteEnvironmentCatalog'
]
>
readonly #now: () => Date
constructor(options: SshHostRemoteEnvironmentInspectorOptions) {
this.#sshHosts = options.sshHosts
this.#sshPool = options.sshPool
this.#loadExpectedCatalog =
options.loadExpectedCatalog ??
(() =>
loadExpectedCatalog(
this.#loadRemoteEnvironmentCatalog =
options.loadRemoteEnvironmentCatalog ??
(async () => ({
expected: await loadExpectedCatalog(
options.agentRuntimeLockPath,
options.remoteRuntimeLockPath
))
),
candidate: null
}))
this.#now = options.now ?? (() => new Date())
}
@@ -86,15 +121,19 @@ export class SshHostRemoteEnvironmentInspector {
const target =
await this.#sshHosts.resolveConnectionTarget(hostId)
signal?.throwIfAborted()
assertTarget(hostId, target)
const expectedTarget = currentTarget(target)
assertValidSshConnectionTarget(hostId, target)
const expectedTarget = toCurrentSshConnectionTarget(target)
this.#sshHosts.assertConnectionTargetCurrent(expectedTarget)
let lease: SshConnectionLease | undefined
let sftp: StagedSftp | undefined
try {
lease = await this.#sshPool.acquire(target, signal)
assertLeaseIdentity(target, lease)
assertSshLeaseMatchesTarget(
lease,
target,
'SSH 远端运行环境连接身份不匹配'
)
const probe = await lease.runAgentBootstrapProbe(signal)
signal?.throwIfAborted()
if (!probe.ready) {
@@ -114,7 +153,10 @@ export class SshHostRemoteEnvironmentInspector {
)
const [catalog, agentRegistry, runtimeRegistry] =
await Promise.all([
this.#expected(probe.architecture),
this.#loadRemoteEnvironmentCatalog(
probe.architecture,
{ signal }
),
readOptionalRegistry(
sftp,
AGENT_REGISTRY_PATH,
@@ -129,18 +171,36 @@ export class SshHostRemoteEnvironmentInspector {
)
])
signal?.throwIfAborted()
assertLeaseIdentity(target, lease)
assertSshLeaseMatchesTarget(
lease,
target,
'SSH 远端运行环境连接身份不匹配'
)
this.#sshHosts.assertConnectionTargetCurrent(expectedTarget)
const remoteDownload = await this.#remoteDownload(
target,
expectedTarget,
catalog.candidate,
catalog.candidateFailure,
signal
)
signal?.throwIfAborted()
assertSshLeaseMatchesTarget(
lease,
target,
'SSH 远端运行环境连接身份不匹配'
)
this.#sshHosts.assertConnectionTargetCurrent(expectedTarget)
return sshHostRemoteEnvironmentSchema.parse({
hostId,
checkedAt: this.#now().toISOString(),
architecture: probe.architecture,
agent: agentStatus(
catalog.agent,
catalog.expected.agent,
agentRegistry?.current,
probe.architecture
),
runtimes: catalog.runtimes.map((runtime) =>
runtimes: catalog.expected.runtimes.map((runtime) =>
runtimeStatus(
runtime,
runtimeRegistry?.current.find(
@@ -150,7 +210,8 @@ export class SshHostRemoteEnvironmentInspector {
),
probe.architecture
)
)
),
remoteDownload
})
} finally {
sftp?.close()
@@ -158,49 +219,134 @@ export class SshHostRemoteEnvironmentInspector {
}
}
#expected(
architecture: 'x64' | 'arm64'
): Promise<ExpectedCatalog> {
return this.#loadExpectedCatalog(architecture)
async #remoteDownload(
target: SshConnectionTarget,
expectedTarget: CurrentSshConnectionTarget,
candidate: RemoteInstallCandidate | null,
candidateFailure:
| RemoteEnvironmentCatalog['candidateFailure']
| undefined,
signal?: AbortSignal
): Promise<SshHostRemoteEnvironment['remoteDownload']> {
if (!candidate) {
return unavailableRemoteDownload(
candidateFailure?.reason ?? 'package-unavailable',
candidateFailure
? {
source: candidateFailure.source,
packageSize: candidateFailure.packageSize
}
: null
)
}
const exposure = {
source: candidate.source,
packageSize: candidate.size
}
let bootstrapLease:
| SshRemotePackageBootstrapLease
| undefined
try {
this.#sshHosts.assertConnectionTargetCurrent(expectedTarget)
bootstrapLease =
await this.#sshPool.acquireRemotePackageBootstrap(
target,
signal
)
assertSshLeaseMatchesTarget(
bootstrapLease,
target,
'SSH 远端运行环境连接身份不匹配'
)
const result = await bootstrapLease.probe(
{
operationId: randomUUID(),
urls: candidate.urls,
size: candidate.size,
sha256: candidate.sha256
},
{ signal }
)
signal?.throwIfAborted()
assertSshLeaseMatchesTarget(
bootstrapLease,
target,
'SSH 远端运行环境连接身份不匹配'
)
this.#sshHosts.assertConnectionTargetCurrent(expectedTarget)
return result.available
? {
available: true,
...exposure
}
: unavailableRemoteDownload(
mapRemoteUnavailableReason(result.reason),
exposure
)
} catch {
signal?.throwIfAborted()
if (bootstrapLease) {
assertSshLeaseMatchesTarget(
bootstrapLease,
target,
'SSH 远端运行环境连接身份不匹配'
)
this.#sshHosts.assertConnectionTargetCurrent(
expectedTarget
)
}
return unavailableRemoteDownload(
'probe-failed',
exposure
)
} finally {
bootstrapLease?.release()
}
}
}
function currentTarget(
target: SshConnectionTarget
): CurrentSshConnectionTarget {
type RemoteDownloadUnavailable = Extract<
SshHostRemoteEnvironment['remoteDownload'],
{ available: false }
>
function unavailableRemoteDownload(
reason: RemoteDownloadUnavailable['reason'],
exposure: {
source: 'github' | 'mirror' | null
packageSize: number | null
} | null
): SshHostRemoteEnvironment['remoteDownload'] {
return {
hostId: target.host.id,
hostRevision: target.hostRevision,
hostKeyGeneration: target.hostKeyGeneration,
username: target.host.username
available: false,
source: exposure?.source ?? null,
packageSize: exposure?.packageSize ?? null,
reason
}
}
function assertTarget(
requestedHostId: string,
target: SshConnectionTarget
): void {
if (
target.host.id !== requestedHostId ||
target.hostRevision < 1 ||
target.hostKeyGeneration < 1 ||
!target.host.hostKey ||
target.host.hostKey.generation !== target.hostKeyGeneration
) {
throw new Error('SSH 主机连接目标无效')
}
}
function assertLeaseIdentity(
target: SshConnectionTarget,
lease: SshConnectionLease
): void {
if (
lease.identity.hostId !== target.host.id ||
lease.identity.hostRevision !== target.hostRevision ||
lease.identity.hostKeyGeneration !== target.hostKeyGeneration
) {
throw new Error('SSH 远端运行环境连接身份不匹配')
function mapRemoteUnavailableReason(
reason: SshRemotePackageBootstrapUnavailableReason
): Extract<
SshHostRemoteEnvironment['remoteDownload'],
{ available: false }
>['reason'] {
switch (reason) {
case 'missing-curl':
case 'missing-sha256sum':
case 'missing-unzip':
case 'bootstrap-tools-unavailable':
return 'missing-tools'
case 'managed-path-unavailable':
return 'home-unwritable'
case 'insufficient-disk-space':
return 'insufficient-disk-space'
case 'download-unavailable':
return 'source-unreachable'
default:
throw new Error(
`未知的远端下载安装探测结果:${String(reason)}`
)
}
}
+10
View File
@@ -650,6 +650,16 @@ describe('SshHostService', () => {
expect(onHostEdited.mock.calls[0]).toEqual([host.id])
expect(onHostRemoved).not.toHaveBeenCalled()
await expect(
harness.service.remove(host.id, async () => {
throw new Error('project cleanup failed')
})
).rejects.toThrow('project cleanup failed')
await expect(harness.service.getSnapshot()).resolves.toMatchObject({
hosts: [expect.objectContaining({ id: host.id })]
})
expect(onHostRemoved).not.toHaveBeenCalled()
await harness.service.remove(host.id)
expect(onHostRemoved).toHaveBeenCalledWith(host.id)
+13 -2
View File
@@ -73,14 +73,25 @@ export class SshHostService {
return this.store.getSnapshot()
}
async remove(hostId: string): Promise<void> {
await this.store.remove(hostId)
remove(hostId: string): Promise<void>
remove<T>(
hostId: string,
commitLocalDeletion: () => T
): Promise<T>
async remove<T>(
hostId: string,
commitLocalDeletion?: () => T
): Promise<T | void> {
const result = commitLocalDeletion
? await this.store.remove(hostId, commitLocalDeletion)
: await this.store.remove(hostId)
for (const [candidateId, pending] of this.pendingHostKeys) {
if (pending.hostId === hostId) {
this.pendingHostKeys.delete(candidateId)
}
}
this.lifecycleHooks.onHostRemoved?.(hostId)
return result
}
async inspectDraftHostKey(
+26
View File
@@ -133,6 +133,32 @@ describe('SshHostStore', () => {
})
})
it('restores a Host when its related local deletion fails', async () => {
const { filePath, store } = await createHarness()
const created = await store.commitValidated({
input: passwordHostInput(),
hostKey: keyA
})
await expect(
store.remove(created.id, async () => {
throw new Error('project cleanup failed')
})
).rejects.toThrow('project cleanup failed')
await expect(store.resolveHost(created.id)).resolves.toMatchObject({
id: created.id,
password: 'private password'
})
expect(
(
JSON.parse(await readFile(filePath, 'utf8')) as {
hosts: Array<{ id: string }>
}
).hosts
).toEqual([expect.objectContaining({ id: created.id })])
})
it('resolves one atomic connection target snapshot with credentials and generations', async () => {
const { store } = await createHarness()
const created = await store.commitValidated({
+31 -7
View File
@@ -366,19 +366,43 @@ export class SshHostStore {
})
}
remove(hostId: string): Promise<void> {
remove(hostId: string): Promise<void>
remove<T>(hostId: string, commitLocalDeletion: () => T): Promise<T>
remove<T>(
hostId: string,
commitLocalDeletion?: () => T
): Promise<T | void> {
return this.queue(async () => {
const settings = cloneSettings(await this.load())
const index = settings.hosts.findIndex(
const previousSettings = cloneSettings(await this.load())
const nextSettings = cloneSettings(previousSettings)
const index = nextSettings.hosts.findIndex(
(host) => host.id === hostId
)
if (index === -1) {
throw new Error('SSH 主机不存在')
}
settings.hosts.splice(index, 1)
await this.persist(settings)
this.settings = settings
this.credentialSourceCache.delete(hostId)
nextSettings.hosts.splice(index, 1)
await this.persist(nextSettings)
try {
const result = await commitLocalDeletion?.()
this.settings = nextSettings
this.credentialSourceCache.delete(hostId)
return result
} catch (error) {
try {
await this.persist(previousSettings)
} catch (rollbackError) {
this.settings = undefined
this.settingsLoad = undefined
this.credentialSourceCache.clear()
throw new AggregateError(
[error, rollbackError],
'SSH 主机删除失败,且主机设置恢复失败',
{ cause: rollbackError }
)
}
throw error
}
})
}
File diff suppressed because it is too large Load Diff
File diff suppressed because it is too large Load Diff