feat: add bounded SSH relay SFTP file destination

Co-authored-by: Orca <help@stably.ai>
2026-10-27 16:02:42 +00:00 · 2026-07-16 10:06:52 -04:00
parent 705bca2dba
commit 72d482201e
6 changed files with 604 additions and 5 deletions
@@ -285,6 +285,7 @@ jobs:
            src/main/ssh/ssh-relay-runtime-source-tree.test.ts \
            src/main/ssh/ssh-relay-runtime-source-scan.test.ts \
            src/main/ssh/ssh-relay-runtime-source-stream.test.ts \
+            src/main/ssh/ssh-relay-runtime-sftp-file-destination.test.ts \
            src/main/ssh/ssh-relay-compiled-manifest-trust.test.ts \
            src/main/ssh/ssh-relay-official-manifest.test.ts \
            src/main/ssh/ssh-relay-manifest-accepted-keys.test.ts \
@@ -840,6 +841,7 @@ jobs:
            src/main/ssh/ssh-relay-runtime-source-tree.test.ts `
            src/main/ssh/ssh-relay-runtime-source-scan.test.ts `
            src/main/ssh/ssh-relay-runtime-source-stream.test.ts `
+            src/main/ssh/ssh-relay-runtime-sftp-file-destination.test.ts `
            src/main/ssh/ssh-relay-compiled-manifest-trust.test.ts `
            src/main/ssh/ssh-relay-official-manifest.test.ts `
            src/main/ssh/ssh-relay-manifest-accepted-keys.test.ts `
@@ -143,6 +143,7 @@ describe('SSH relay runtime artifact workflow', () => {
      'ssh-relay-runtime-source-tree.test.ts',
      'ssh-relay-runtime-source-scan.test.ts',
      'ssh-relay-runtime-source-stream.test.ts',
+      'ssh-relay-runtime-sftp-file-destination.test.ts',
      'ssh-relay-compiled-manifest-trust.test.ts',
      'ssh-relay-official-manifest.test.ts',
      'ssh-relay-manifest-accepted-keys.test.ts',
@@ -9,9 +9,21 @@ keeps commands, hashes, runner identities, timings, and failure details.
 A checked box means the work has evidence in the detailed ledger. Design approval alone does not
 complete a box.

-Active checkpoint: **Milestone 6 / Work Package 5 bounded runtime transfer — bounded client source
-streaming is exact-head green; the smallest disconnected SFTP adapter audit is next, 2026-07-15,
-Codex implementation owner.**
+Active checkpoint: **Milestone 6 / Work Package 5 bounded runtime transfer — disconnected SFTP
+per-file destination is locally green; exact-head native proof is next, 2026-07-15, Codex
+implementation owner.** `E-M6-SFTP-FILE-DESTINATION-AUDIT-001` fixes exclusive `wx` open, callback-bounded
+positional writes, explicit POSIX mode repair/handle-stat proof, and joined close-before-unlink
+cleanup. It deliberately borrows but does not acquire/end an SFTP session and does not create
+directories, clean a whole staging tree, or publish/launch anything. Session-wide cancellation,
+full-size live SFTP, product/default wiring, and SignPath remain later gates.
+`E-M6-SFTP-FILE-DESTINATION-LOCAL-RED-001` records the expected missing-module failure for the
+12-case purpose contract; the 7/7 native workflow oracle already pins it once in both command
+families. Protected resolvers remain untouched and no production importer exists.
+`E-M6-SFTP-FILE-DESTINATION-LOCAL-001` passes the hardened 13-case purpose suite, 103 focused cases
+with one declared skip, 579 broad relay cases with five declared skips, 280/280 workflow cases,
+typecheck, full lint, targeted format, max-lines, diff, protected-resolver, and no-product-import
+gates. It proves the exact per-file lifecycle only; SFTP session/tree orchestration, live/full-size
+remote proof, and every product/default path remain open.
 `E-M6-SOURCE-STREAM-CI-001` closes exact commit `df39c287d`: all six primary native clients pass the
 27-case source-stream suite (Windows has only the declared POSIX-mode skip), exact 85–125 MB trees
 stream in 103.663–530.937 ms with 0–5,492,736 incremental RSS bytes, both Linux supplements and the
@@ -8,8 +8,8 @@ work; keep exact commands, runner identities, hashes, metrics, and residual gaps

 Date created: 2026-07-14<br>
 Last updated: 2026-07-15<br>
-Current phase: Milestone 6 / Work Package 5 bounded runtime transfer — **In progress — 2026-07-15, Codex implementation owner: bounded client source streaming is exact-head green; the smallest disconnected SFTP adapter audit is next**. Keep every capability disconnected from every production/default SSH path. Real Apple/SignPath rehearsals remain an explicit late gate and are not on this package's critical path.<br>
-Session checkpoint: **BOUNDED SOURCE STREAM EXACT-HEAD GREEN / SFTP ADAPTER AUDIT NEXT — 2026-07-15, Codex implementation owner** — `E-M6-SOURCE-STREAM-CI-001` closes all-six native purpose and exact full-size client-stream proof. The first adjacent computer-use Windows attempt exposed only two unrelated five-second launcher timeouts under `E-M6-SOURCE-STREAM-ADJACENT-CI-RED-001`; its same-SHA failed-job rerun is fully green without changing timeouts or code. Audit the smallest SFTP destination/session boundary before writing its RED tests. Do not add a product caller, settings/fallback wiring, tuple enablement, publication, or default behavior. Merging to `main` remains prohibited, SignPath is deferred, and legacy remains the production default.<br>
+Current phase: Milestone 6 / Work Package 5 bounded runtime transfer — **In progress — 2026-07-15, Codex implementation owner: disconnected SFTP per-file destination is locally green; exact-head native proof is next**. Keep every capability disconnected from every production/default SSH path. Real Apple/SignPath rehearsals remain an explicit late gate and are not on this package's critical path.<br>
+Session checkpoint: **SFTP PER-FILE DESTINATION LOCAL GREEN — 2026-07-15, Codex implementation owner** — `E-M6-SFTP-FILE-DESTINATION-LOCAL-001` passes the hardened 13-case purpose contract, focused/broad/workflow tests, and all static/isolation gates. It proves exclusive remote open, exact callback-bounded offsets, in-flight cancellation joining, explicit POSIX mode repair/handle-stat proof, Windows mode skip, close-before-unlink cleanup, and idempotent aggregation. Commit/push this disconnected package and require exact-head all-six native/adjacent CI before advancing to SFTP session/tree orchestration. The adapter borrows but does not acquire/end an SFTP session and does not create directories or publish/launch a tree; session-wide cancellation and whole-staging cleanup remain later gates. Do not add a product caller, settings/fallback wiring, tuple enablement, publication, or default behavior. Merging to `main` remains prohibited, SignPath is deferred, and legacy remains the production default.<br>
 Primary design: [SSH relay GitHub Release plan](./2026-07-14-ssh-relay-github-release-plan.html)<br>
 Motivating issues: [#8450](https://github.com/stablyai/orca/issues/8450), [#1693](https://github.com/stablyai/orca/issues/1693)

@@ -16738,6 +16738,91 @@ runtime source stream destination is incomplete`. The contract step failed after
  install/launch step, product/Beta/fallback/tuple/publication/default path, and SignPath remain
  open. Legacy remains the sole production/default path and every tuple remains disabled.

+### E-M6-SFTP-FILE-DESTINATION-AUDIT-001 — exclusive bounded SFTP file lifecycle boundary
+
+- Date/owner: 2026-07-15, Codex implementation owner.
+- Exact input boundary: accept only a caller-owned SFTP file primitive set, one already validated
+  exact remote staging-file path, the signed `0o644` or `0o755` mode, an explicit POSIX-mode flag,
+  and the same required caller-owned `AbortSignal` used by the source streamer. Do not derive any
+  remote path from a client-native local path and do not acquire or end the SFTP session here.
+- Open/write contract: check cancellation before open; use only exclusive `wx` open; once the handle
+  exists this adapter owns that new file. Each `write(chunk)` issues exactly one positional SFTP
+  write and resolves only from its callback, making that callback the source stream's reusable
+  64-KiB buffer-lifetime boundary. Writes are sequential, offsets are exact, empty/concurrent/late
+  writes are rejected, and no chunk/tree concatenation or base64 representation exists.
+- Completion contract: for POSIX remotes, apply the signed mode through the owned handle, re-read
+  handle attributes, and require the exact permission bits before closing. Windows/non-POSIX
+  remotes skip POSIX mode operations explicitly. A close is successful only after the remote handle
+  callback settles and a final signal check passes.
+- Failure/cancellation contract: an open failure never unlinks because no ownership was obtained.
+  Any later write, mode, verification, close, or cancellation failure leaves the destination
+  abortable; abort attempts owned-handle close before unlink, joins cleanup failures, and is
+  idempotent. Cancellation during an in-flight SFTP callback remains joined until that callback is
+  forced to settle by the later session owner, preventing reuse of a possibly retained source
+  buffer.
+- Required purpose RED: missing destination module; exclusive open and exact offsets; deferred
+  callback/buffer lifetime; pre-open and post-open cancellation; write/mode/stat/close failures;
+  POSIX exact-mode verification; explicit Windows mode skip; no deletion on failed exclusive open;
+  close-before-unlink order; joined/idempotent cleanup; illegal concurrent/empty/late operations;
+  and path-free adapter-authored diagnostics. Add the suite exactly once to both native artifact
+  command families before implementation.
+- Explicit residual boundary: this is a per-file primitive, not an SFTP transfer. SFTP session
+  acquisition/end/error ownership, bounded tree concurrency, remote directory creation, remote-path
+  policy, session-wide timeout/cancellation, full-staging cleanup, real server behavior, full-size
+  transfer, mode behavior across servers, and later remote tree verification/install/publish/launch
+  all remain open. No production importer, setting, fallback, tuple, release, or default path may be
+  added by this package; legacy remains the sole production path.
+
+### E-M6-SFTP-FILE-DESTINATION-LOCAL-RED-001 — destination lifecycle is absent
+
+- Date/owner/environment: 2026-07-15, Codex implementation owner; local macOS arm64, Node 26.0.0,
+  pnpm 10.24.0, atop exact checkpoint commit `705bca2db` plus the purpose/workflow/checklist RED diff.
+- Purpose command: `pnpm exec vitest run --config config/vitest.config.ts --maxWorkers=1
+src/main/ssh/ssh-relay-runtime-sftp-file-destination.test.ts` — expected RED, exit 1: one failed
+  suite, zero collected tests, because `./ssh-relay-runtime-sftp-file-destination` does not exist.
+  The 260-line source contract declares 12 cases including two parameterized failure cells.
+- Native workflow oracle: `pnpm exec vitest run --config config/vitest.config.ts --maxWorkers=1
+config/scripts/ssh-relay-runtime-workflow.test.mjs` — PASS, 7/7 in 356 ms. The workflow source and
+  oracle already require the purpose suite exactly once in both POSIX and PowerShell native command
+  families, so implementation cannot become locally green while native coverage remains omitted.
+- Static precondition: `git diff --check` passes and the test remains below the max-lines ratchet at
+  260 lines. Protected resolver files remain untouched. No production importer, SFTP session owner,
+  remote directory/tree orchestrator, product caller, setting, fallback, tuple, publication, or
+  default behavior exists in the RED diff.
+
+### E-M6-SFTP-FILE-DESTINATION-LOCAL-001 — per-file SFTP lifecycle is locally green
+
+- Date/owner/environment: 2026-07-15, Codex implementation owner; local macOS arm64, Node 26.0.0,
+  pnpm 10.24.0, atop exact checkpoint commit `705bca2db` plus the implementation/checklist diff.
+- Implementation: the 210-line `ssh-relay-runtime-sftp-file-destination.ts` accepts only the audited
+  primitive boundary. It opens with `wx` and signed mode, awaits one SFTP positional-write callback
+  per source chunk before advancing the exact offset, checks the same signal around every callback,
+  repairs and verifies POSIX permission bits before close, and skips those operations only from the
+  explicit non-POSIX flag. Adapter-authored state/input errors contain no remote path.
+- Purpose command: `/usr/bin/time -l pnpm exec vitest run --config config/vitest.config.ts
+--maxWorkers=1 src/main/ssh/ssh-relay-runtime-sftp-file-destination.test.ts` — PASS, 13/13 in
+  431 ms Vitest / 2.36 seconds wall, 133,005,312-byte maximum RSS, zero swaps. The additional
+  hardening case proves cancellation during a pending write cannot settle or begin cleanup until the
+  SFTP callback releases the borrowed source buffer; it then closes before unlinking.
+- Focused command: the destination, source stream/scan/tree, acquisition/cache population and
+  resolution, integration, and workflow-oracle ten-file command — PASS, 103 tests with one declared
+  platform skip in 8.28 seconds.
+- Broad commands: `src/main/ssh/ssh-relay-*.test.ts` — PASS, 579 tests with five declared platform/
+  full-size skips in 27.25 seconds; `config/scripts/ssh-relay-runtime-*.test.mjs` — PASS, 280/280 in
+  20.82 seconds. The first quoted-glob invocations matched no files and were discarded as command
+  syntax mistakes; the unquoted exact checklist commands above are the credited results.
+- Static/isolation commands: `pnpm typecheck`, targeted `pnpm exec oxlint`, `pnpm lint`, targeted
+  `pnpm exec oxfmt --check`, `pnpm run check:max-lines-ratchet`, `git diff --check`, the protected-
+  resolver diff gate, and the no-non-test-importer `rg` gate pass. Full lint includes switch
+  exhaustiveness, styled-scrollbar, 41 reliability, bundled-skill, max-lines, and localization gates;
+  only pre-existing unrelated warnings remain. The module/test are 210/289 lines with no bypass.
+- Residual: this is callback-contract proof with mocked primitives, not a live SFTP server, session,
+  SSH connection, remote directory/staging tree, bounded multi-file transfer, session-wide abort/
+  timeout, whole-tree cleanup, or full-size remote measurement. It has no non-test importer. Exact-
+  head all-six Node 24 and adjacent Actions evidence is required before auditing the session/tree
+  adapter. Product/Beta/fallback/tuple/publication/default behavior and SignPath remain absent;
+  legacy remains the sole production path.
+
 ## Accepted Gaps

 No product gap is accepted merely because it appears in this list. Each entry requires explicit
@@ -0,0 +1,289 @@
+import { describe, expect, it, vi } from 'vitest'
+
+import {
+  openSshRelayRuntimeSftpFileDestination,
+  type SshRelayRuntimeSftpFileOperations
+} from './ssh-relay-runtime-sftp-file-destination'
+
+type Callback = (error?: Error) => void
+
+function createOperations(options: { mode?: number } = {}): {
+  operations: SshRelayRuntimeSftpFileOperations
+  events: string[]
+  handle: Buffer
+} {
+  const handle = Buffer.from('owned-handle')
+  const events: string[] = []
+  const operations = {
+    open: vi.fn(
+      (
+        _path: string,
+        _flags: string,
+        _attributes: { mode: number },
+        callback: (error: Error | undefined, value: Buffer) => void
+      ) => {
+        events.push('open')
+        callback(undefined, handle)
+      }
+    ),
+    write: vi.fn(
+      (
+        _handle: Buffer,
+        _buffer: Buffer,
+        _offset: number,
+        _length: number,
+        _position: number,
+        callback: Callback
+      ) => {
+        events.push('write')
+        callback()
+      }
+    ),
+    fchmod: vi.fn((_handle: Buffer, _mode: number, callback: Callback) => {
+      events.push('fchmod')
+      callback()
+    }),
+    fstat: vi.fn(
+      (_handle: Buffer, callback: (error: Error | undefined, value: { mode: number }) => void) => {
+        events.push('fstat')
+        callback(undefined, { mode: options.mode ?? 0o100755 })
+      }
+    ),
+    close: vi.fn((_handle: Buffer, callback: Callback) => {
+      events.push('close')
+      callback()
+    }),
+    unlink: vi.fn((_path: string, callback: Callback) => {
+      events.push('unlink')
+      callback()
+    })
+  } as unknown as SshRelayRuntimeSftpFileOperations
+  return { operations, events, handle }
+}
+
+function openDestination(
+  operations: SshRelayRuntimeSftpFileOperations,
+  options: { signal?: AbortSignal; enforcePosixMode?: boolean; remotePath?: string } = {}
+) {
+  return openSshRelayRuntimeSftpFileDestination({
+    operations,
+    remotePath: options.remotePath ?? '/owned-staging/bin/node',
+    mode: 0o755,
+    enforcePosixMode: options.enforcePosixMode ?? true,
+    signal: options.signal ?? new AbortController().signal
+  })
+}
+
+describe('SSH relay runtime SFTP file destination', () => {
+  it('opens exclusively and awaits each exact positional write callback', async () => {
+    const { operations, handle } = createOperations()
+    let firstCallback: Callback | undefined
+    vi.mocked(operations.write).mockImplementationOnce(
+      (_handle, _buffer, _offset, _length, _position, callback) => {
+        firstCallback = callback
+      }
+    )
+
+    const destination = await openDestination(operations)
+    const firstChunk = Buffer.from('abc')
+    let firstSettled = false
+    const firstWrite = destination.write(firstChunk).then(() => {
+      firstSettled = true
+    })
+    await Promise.resolve()
+    expect(firstSettled).toBe(false)
+    expect(operations.open).toHaveBeenCalledWith(
+      '/owned-staging/bin/node',
+      'wx',
+      { mode: 0o755 },
+      expect.any(Function)
+    )
+    expect(operations.write).toHaveBeenNthCalledWith(
+      1,
+      handle,
+      firstChunk,
+      0,
+      3,
+      0,
+      expect.any(Function)
+    )
+
+    firstCallback?.()
+    await firstWrite
+    const secondChunk = Buffer.from('de')
+    await destination.write(secondChunk)
+    expect(operations.write).toHaveBeenNthCalledWith(
+      2,
+      handle,
+      secondChunk,
+      0,
+      2,
+      3,
+      expect.any(Function)
+    )
+  })
+
+  it('repairs and verifies exact POSIX mode before closing the owned handle', async () => {
+    const { operations, events, handle } = createOperations()
+    const destination = await openDestination(operations)
+    await destination.close()
+
+    expect(operations.fchmod).toHaveBeenCalledWith(handle, 0o755, expect.any(Function))
+    expect(operations.fstat).toHaveBeenCalledWith(handle, expect.any(Function))
+    expect(events).toEqual(['open', 'fchmod', 'fstat', 'close'])
+  })
+
+  it('skips POSIX mode operations only when the caller explicitly disables them', async () => {
+    const { operations, events } = createOperations()
+    const destination = await openDestination(operations, { enforcePosixMode: false })
+    await destination.close()
+
+    expect(events).toEqual(['open', 'close'])
+    expect(operations.fchmod).not.toHaveBeenCalled()
+    expect(operations.fstat).not.toHaveBeenCalled()
+  })
+
+  it('rejects a mode mismatch and removes the owned incomplete file on abort', async () => {
+    const { operations, events } = createOperations({ mode: 0o100644 })
+    const destination = await openDestination(operations)
+    await expect(destination.close()).rejects.toThrow(/mode/i)
+    await destination.abort(new Error('mode mismatch'))
+
+    expect(events).toEqual(['open', 'fchmod', 'fstat', 'close', 'unlink'])
+  })
+
+  it.each(['fchmod', 'fstat'] as const)('fails closed when %s fails', async (operation) => {
+    const { operations } = createOperations()
+    vi.mocked(operations[operation]).mockImplementationOnce((...args: unknown[]) => {
+      const callback = args.at(-1) as Callback
+      callback(new Error(`${operation} failed`))
+    })
+    const destination = await openDestination(operations)
+
+    await expect(destination.close()).rejects.toThrow(`${operation} failed`)
+    await destination.abort(new Error(`${operation} failed`))
+    expect(operations.close).toHaveBeenCalledOnce()
+    expect(operations.unlink).toHaveBeenCalledOnce()
+  })
+
+  it('does not unlink a pre-existing file when exclusive open fails', async () => {
+    const { operations } = createOperations()
+    vi.mocked(operations.open).mockImplementationOnce((...args: unknown[]) => {
+      const callback = args.at(-1) as (error: Error) => void
+      callback(new Error('already exists'))
+    })
+
+    await expect(openDestination(operations)).rejects.toThrow('already exists')
+    expect(operations.close).not.toHaveBeenCalled()
+    expect(operations.unlink).not.toHaveBeenCalled()
+  })
+
+  it('rejects pre-open cancellation without creating a remote file', async () => {
+    const { operations } = createOperations()
+    const controller = new AbortController()
+    controller.abort(new Error('cancelled before open'))
+
+    await expect(openDestination(operations, { signal: controller.signal })).rejects.toThrow(
+      'cancelled before open'
+    )
+    expect(operations.open).not.toHaveBeenCalled()
+  })
+
+  it('cleans an exclusively opened file when cancellation wins the open callback', async () => {
+    const { operations, events, handle } = createOperations()
+    const controller = new AbortController()
+    vi.mocked(operations.open).mockImplementationOnce((...args: unknown[]) => {
+      const callback = args.at(-1) as (error: undefined, value: Buffer) => void
+      controller.abort(new Error('cancelled after open'))
+      callback(undefined, handle)
+    })
+
+    await expect(openDestination(operations, { signal: controller.signal })).rejects.toThrow(
+      'cancelled after open'
+    )
+    expect(events).toEqual(['close', 'unlink'])
+  })
+
+  it('joins cancellation to the in-flight write callback before cleanup', async () => {
+    const { operations, events } = createOperations()
+    const controller = new AbortController()
+    let writeCallback: Callback | undefined
+    vi.mocked(operations.write).mockImplementationOnce(
+      (_handle, _buffer, _offset, _length, _position, callback) => {
+        events.push('write')
+        writeCallback = callback
+      }
+    )
+    const destination = await openDestination(operations, { signal: controller.signal })
+    let settled = false
+    const write = destination.write(Buffer.from('bytes')).finally(() => {
+      settled = true
+    })
+
+    controller.abort(new Error('cancelled during write'))
+    await Promise.resolve()
+    expect(settled).toBe(false)
+    expect(operations.close).not.toHaveBeenCalled()
+    writeCallback?.()
+    await expect(write).rejects.toThrow('cancelled during write')
+    await destination.abort(controller.signal.reason)
+    expect(events).toEqual(['open', 'write', 'close', 'unlink'])
+  })
+
+  it('propagates a write failure and then closes before unlinking', async () => {
+    const { operations, events } = createOperations()
+    vi.mocked(operations.write).mockImplementationOnce((...args: unknown[]) => {
+      const callback = args.at(-1) as Callback
+      events.push('write')
+      callback(new Error('remote write failed'))
+    })
+    const destination = await openDestination(operations)
+
+    await expect(destination.write(Buffer.from('bytes'))).rejects.toThrow('remote write failed')
+    await destination.abort(new Error('remote write failed'))
+    expect(events).toEqual(['open', 'write', 'close', 'unlink'])
+  })
+
+  it('joins close and unlink cleanup failures and makes abort idempotent', async () => {
+    const { operations } = createOperations()
+    vi.mocked(operations.close).mockImplementation((_handle, callback) =>
+      callback(new Error('close cleanup failed'))
+    )
+    vi.mocked(operations.unlink).mockImplementation((_path, callback) =>
+      callback(new Error('unlink cleanup failed'))
+    )
+    const destination = await openDestination(operations)
+
+    const firstAbort = destination.abort(new Error('primary'))
+    const secondAbort = destination.abort(new Error('secondary'))
+    expect(firstAbort).toBe(secondAbort)
+    await expect(firstAbort).rejects.toMatchObject({
+      errors: [
+        expect.objectContaining({ message: 'close cleanup failed' }),
+        expect.objectContaining({ message: 'unlink cleanup failed' })
+      ]
+    })
+    expect(operations.close).toHaveBeenCalledOnce()
+    expect(operations.unlink).toHaveBeenCalledOnce()
+  })
+
+  it('rejects empty, concurrent, and late writes with path-free diagnostics', async () => {
+    const { operations } = createOperations()
+    let writeCallback: Callback | undefined
+    vi.mocked(operations.write).mockImplementationOnce(
+      (_handle, _buffer, _offset, _length, _position, callback) => {
+        writeCallback = callback
+      }
+    )
+    const secretPath = '/home/private-user/staging/node'
+    const destination = await openDestination(operations, { remotePath: secretPath })
+
+    await expect(destination.write(Buffer.alloc(0))).rejects.not.toThrow(secretPath)
+    const pending = destination.write(Buffer.from('first'))
+    await expect(destination.write(Buffer.from('second'))).rejects.toThrow(/concurrent/i)
+    writeCallback?.()
+    await pending
+    await destination.close()
+    await expect(destination.write(Buffer.from('late'))).rejects.toThrow(/closed/i)
+  })
+})
@@ -0,0 +1,210 @@
+import type { SshRelayRuntimeSourceDestination } from './ssh-relay-runtime-source-stream'
+
+type SftpCallback = (error?: Error) => void
+
+export type SshRelayRuntimeSftpFileOperations = Readonly<{
+  open: (
+    path: string,
+    flags: 'wx',
+    attributes: { mode: number },
+    callback: (error: Error | undefined, handle: Buffer) => void
+  ) => void
+  write: (
+    handle: Buffer,
+    buffer: Buffer,
+    offset: number,
+    length: number,
+    position: number,
+    callback: SftpCallback
+  ) => void
+  fchmod: (handle: Buffer, mode: number, callback: SftpCallback) => void
+  fstat: (
+    handle: Buffer,
+    callback: (error: Error | undefined, attributes: { mode: number }) => void
+  ) => void
+  close: (handle: Buffer, callback: SftpCallback) => void
+  unlink: (path: string, callback: SftpCallback) => void
+}>
+
+export type OpenSshRelayRuntimeSftpFileDestinationOptions = Readonly<{
+  operations: SshRelayRuntimeSftpFileOperations
+  remotePath: string
+  mode: 0o644 | 0o755
+  enforcePosixMode: boolean
+  signal: AbortSignal
+}>
+
+type DestinationState = 'open' | 'closed' | 'complete' | 'aborting' | 'aborted'
+
+function waitForSftpCallback(register: (callback: SftpCallback) => void): Promise<void> {
+  return new Promise((resolve, reject) => {
+    register((error) => (error ? reject(error) : resolve()))
+  })
+}
+
+function waitForSftpValue<T>(
+  register: (callback: (error: Error | undefined, value: T) => void) => void
+): Promise<T> {
+  return new Promise((resolve, reject) => {
+    register((error, value) => (error ? reject(error) : resolve(value)))
+  })
+}
+
+function cleanupFailure(failures: readonly unknown[]): unknown | undefined {
+  if (failures.length === 0) {
+    return undefined
+  }
+  return failures.length === 1
+    ? failures[0]
+    : new AggregateError(failures, 'SSH relay runtime SFTP file cleanup failed')
+}
+
+function validateOptions(options: OpenSshRelayRuntimeSftpFileDestinationOptions): void {
+  if (!options.operations || typeof options.remotePath !== 'string' || options.remotePath === '') {
+    throw new Error('SSH relay runtime SFTP file destination input is invalid')
+  }
+  if (options.remotePath.includes('\0')) {
+    throw new Error('SSH relay runtime SFTP file destination path is invalid')
+  }
+  if (options.mode !== 0o644 && options.mode !== 0o755) {
+    throw new Error('SSH relay runtime SFTP file destination mode is invalid')
+  }
+}
+
+function createDestination(
+  options: OpenSshRelayRuntimeSftpFileDestinationOptions,
+  handle: Buffer
+): SshRelayRuntimeSourceDestination {
+  const { operations, remotePath, mode, enforcePosixMode, signal } = options
+  let state: DestinationState = 'open'
+  let position = 0
+  let activeOperation: Promise<void> | undefined
+  let abortPromise: Promise<void> | undefined
+
+  const assertWritable = (): void => {
+    if (state !== 'open') {
+      throw new Error('SSH relay runtime SFTP file destination is closed')
+    }
+    if (activeOperation) {
+      throw new Error('SSH relay runtime SFTP file destination has a concurrent operation')
+    }
+  }
+
+  const runExclusive = (operation: () => Promise<void>): Promise<void> => {
+    const pending = operation()
+    activeOperation = pending
+    void pending
+      .finally(() => {
+        if (activeOperation === pending) {
+          activeOperation = undefined
+        }
+      })
+      .catch(() => {})
+    return pending
+  }
+
+  const write = (chunk: Buffer): Promise<void> => {
+    try {
+      assertWritable()
+    } catch (error) {
+      return Promise.reject(error)
+    }
+    if (!Buffer.isBuffer(chunk) || chunk.length === 0) {
+      return Promise.reject(new Error('SSH relay runtime SFTP file destination chunk is empty'))
+    }
+    return runExclusive(async () => {
+      signal.throwIfAborted()
+      // Why: the callback is the remote-consumption boundary; resolving earlier could let the
+      // source worker reuse its one buffer while ssh2 still retains the chunk view.
+      await waitForSftpCallback((callback) =>
+        operations.write(handle, chunk, 0, chunk.length, position, callback)
+      )
+      position += chunk.length
+      signal.throwIfAborted()
+    })
+  }
+
+  const close = (): Promise<void> => {
+    try {
+      assertWritable()
+    } catch (error) {
+      return Promise.reject(error)
+    }
+    return runExclusive(async () => {
+      signal.throwIfAborted()
+      if (enforcePosixMode) {
+        await waitForSftpCallback((callback) => operations.fchmod(handle, mode, callback))
+        signal.throwIfAborted()
+        const attributes = await waitForSftpValue<{ mode: number }>((callback) =>
+          operations.fstat(handle, callback)
+        )
+        if (!Number.isInteger(attributes.mode) || (attributes.mode & 0o777) !== mode) {
+          throw new Error('SSH relay runtime SFTP file mode verification failed')
+        }
+      }
+      signal.throwIfAborted()
+      await waitForSftpCallback((callback) => operations.close(handle, callback))
+      state = 'closed'
+      signal.throwIfAborted()
+      state = 'complete'
+    })
+  }
+
+  const performAbort = async (): Promise<void> => {
+    if (state === 'complete' || state === 'aborted') {
+      return
+    }
+    const pending = activeOperation
+    if (pending) {
+      await pending.catch(() => {})
+    }
+    const handleNeedsClose = state !== 'closed'
+    state = 'aborting'
+    const failures: unknown[] = []
+    if (handleNeedsClose) {
+      await waitForSftpCallback((callback) => operations.close(handle, callback)).catch((error) =>
+        failures.push(error)
+      )
+    }
+    await waitForSftpCallback((callback) => operations.unlink(remotePath, callback)).catch(
+      (error) => failures.push(error)
+    )
+    state = 'aborted'
+    const failure = cleanupFailure(failures)
+    if (failure !== undefined) {
+      throw failure
+    }
+  }
+
+  const abort = (_reason: unknown): Promise<void> => {
+    abortPromise ??= performAbort()
+    return abortPromise
+  }
+
+  return Object.freeze({ write, close, abort })
+}
+
+export async function openSshRelayRuntimeSftpFileDestination(
+  options: OpenSshRelayRuntimeSftpFileDestinationOptions
+): Promise<SshRelayRuntimeSourceDestination> {
+  validateOptions(options)
+  options.signal.throwIfAborted()
+  const handle = await waitForSftpValue<Buffer>((callback) =>
+    options.operations.open(options.remotePath, 'wx', { mode: options.mode }, callback)
+  )
+  const destination = createDestination(options, handle)
+  try {
+    options.signal.throwIfAborted()
+    return destination
+  } catch (error) {
+    try {
+      await destination.abort(error)
+    } catch (cleanupError) {
+      throw new AggregateError(
+        [error, cleanupError],
+        'SSH relay runtime SFTP file open cancellation cleanup failed'
+      )
+    }
+    throw error
+  }
+}