mirror of
https://github.com/stablyai/orca.git
synced 2026-10-22 00:03:47 +00:00
feat(skills): add bounded operation observability
This commit is contained in:
@@ -1201,11 +1201,11 @@ physical SSH macOS and supported Windows remain open.
|
||||
|
||||
### Metrics and dashboards
|
||||
|
||||
- [ ] Record bounded package byte/file counts and package, upload, finalization, download, transfer,
|
||||
- [x] Record bounded package byte/file counts and package, upload, finalization, download, transfer,
|
||||
install, placement, and recovery durations.
|
||||
- [ ] Record outcomes by error category, OS, destination kind, transport, conflict type, and
|
||||
- [x] Record outcomes by error category, OS, destination kind, transport, conflict type, and
|
||||
placement topology.
|
||||
- [ ] Record junction, alias, copy-fallback, rollback, recovery, capability-absence, and orphan
|
||||
- [x] Record junction, alias, copy-fallback, rollback, recovery, capability-absence, and orphan
|
||||
reconciliation counts.
|
||||
- [x] Add dashboards for grant/finalize/share rates, authorization and rate limits, finalization
|
||||
saturation, archive rejection, and digest mismatch.
|
||||
@@ -1242,6 +1242,18 @@ replaced only the inert descriptor-propagation timer. A manual run emitted only
|
||||
namespace measurements with zero overdue objects; the post-apply targeted plan is zero-diff, SQL
|
||||
remained `NEVER`/`STOPPED`, all Relay MIG targets remained zero, and production remained disabled.
|
||||
|
||||
Desktop operations now emit local diagnostic spans for package creation, direct upload,
|
||||
finalization, grant download, client-mediated runtime/SSH transfer, single and bundle install,
|
||||
provider placement, transaction rollback settlement, and startup recovery. Package bytes/files,
|
||||
OS, destination kind, transport, conflicts, placement topology/mechanism, copy fallback,
|
||||
capability absence, recovered transactions, stale extraction/lock cleanup, and failure categories
|
||||
use fixed or bounded values; bundle error-category cardinality is capped at 32. The implementation
|
||||
adds no product telemetry event, RPC field, persisted schema, or remote opcode. The skill-sharing
|
||||
release suite passes 423 tests with 29 expected platform skips, the focused affected matrix passes
|
||||
98 tests with one expected skip, typecheck and the full lint pipeline pass, and the real-process
|
||||
macOS recovery matrix passes all 17 crash boundaries. Physical Windows/WSL reruns remain separate
|
||||
platform gates below.
|
||||
|
||||
### Logging and privacy
|
||||
|
||||
- [x] Log package/version IDs, phases, destination labels, placement outcomes, and bounded error
|
||||
|
||||
@@ -91,6 +91,18 @@ SSH now carries only schema-validated structured skill failures through optional
|
||||
when it is absent. Invalid gzip input has a stable archive category, and deterministic `EACCES`
|
||||
and `ENOSPC` injection proves failed updates preserve the previous installed version.
|
||||
|
||||
Desktop skill operations now have privacy-bounded local diagnostic spans for package creation,
|
||||
Cloud upload and finalization, grant download, client-mediated runtime and SSH transfer, single
|
||||
and bundle installation, provider placement, transaction rollback settlement, and startup
|
||||
recovery. They record aggregate bytes/files, phase duration, OS, destination and transport,
|
||||
conflict/topology/mechanism outcomes, copy fallback, capability absence, recovery/orphan counts,
|
||||
and bounded error categories without paths, skill names, contents, share URLs, grants, host IDs,
|
||||
or credentials. Bundle summaries are aggregate and cap error-category cardinality at 32. This is
|
||||
not product telemetry and changes no RPC, persisted state, or remote opcode, so old clients and
|
||||
hosts remain compatible. The release suite passes 423 tests, full typecheck/lint pass, and the
|
||||
real-process macOS crash-recovery matrix passes all 17 boundaries; physical platform reruns remain
|
||||
tracked in the checklist.
|
||||
|
||||
The dedicated staging bucket, IAM, database, principal, enabled secret version, metrics, dashboard,
|
||||
and alerts exist in `onorca-cloud-staging`. The complete skill-infrastructure target is zero-diff.
|
||||
After `#314`, the targeted Auth/API plan also reports zero intended changes. Known shared Cloud SQL
|
||||
|
||||
@@ -22,6 +22,7 @@ import { renameSkillPathWithWindowsRetry } from './skill-filesystem-retry'
|
||||
import { extractSkillBundleArchive } from './skill-bundle-extraction'
|
||||
import { observeSkillPackage, type ObservedSkillPackage } from './skill-package-identity'
|
||||
import { writeSkillTarGzip, type SkillTarWriteEntry } from './skill-package-tar'
|
||||
import { startSkillPhaseOperation } from './skill-operation-observability'
|
||||
|
||||
export type SkillBundleSource = { id?: string; sourceDirectory: string }
|
||||
|
||||
@@ -110,7 +111,7 @@ async function stageSkill(input: {
|
||||
})
|
||||
}
|
||||
|
||||
export async function createSkillBundleArchive(
|
||||
async function createSkillBundleArchiveUnobserved(
|
||||
input: {
|
||||
sources: readonly SkillBundleSource[]
|
||||
archivePath: string
|
||||
@@ -223,3 +224,36 @@ export async function createSkillBundleArchive(
|
||||
await rm(temporaryArchive, { force: true }).catch(() => undefined)
|
||||
}
|
||||
}
|
||||
|
||||
export async function createSkillBundleArchive(
|
||||
input: {
|
||||
sources: readonly SkillBundleSource[]
|
||||
archivePath: string
|
||||
packageId: string
|
||||
versionId: string
|
||||
bundleName: string
|
||||
description?: string
|
||||
createdAt?: string
|
||||
},
|
||||
dependencies: SkillBundleCreationDependencies = {}
|
||||
): Promise<CreatedSkillBundle> {
|
||||
const operation = startSkillPhaseOperation({
|
||||
phase: 'package',
|
||||
packageKind: 'bundle',
|
||||
skillCount: input.sources.length
|
||||
})
|
||||
try {
|
||||
const created = await createSkillBundleArchiveUnobserved(input, dependencies)
|
||||
const files = created.manifest.skills.flatMap((skill) => skill.files)
|
||||
operation.complete({
|
||||
status: 'complete',
|
||||
fileCount: files.length,
|
||||
totalBytes: files.reduce((total, file) => total + file.size, 0),
|
||||
compressedBytes: created.compressedBytes
|
||||
})
|
||||
return created
|
||||
} catch (error) {
|
||||
operation.fail(error)
|
||||
throw error
|
||||
}
|
||||
}
|
||||
|
||||
@@ -12,6 +12,7 @@ import { installSkillBundle } from './skill-bundle-install-service'
|
||||
import { downloadSkillPackageGrant } from './skill-package-download'
|
||||
import { detectSkillProvidersInWsl } from './skill-wsl-provider-detection'
|
||||
import { createWslSkillInstallFilesystem } from './skill-wsl-install-filesystem'
|
||||
import { startSkillBundleInstallOperation } from './skill-operation-observability'
|
||||
|
||||
type StagedSkillBundle = { archivePath: string; cleanup(): Promise<void> }
|
||||
|
||||
@@ -60,17 +61,17 @@ async function resolveIngress(
|
||||
return { archivePath: request.ingress.path, cleanup: async () => undefined }
|
||||
}
|
||||
|
||||
export async function executeSkillBundleInstallRequest(
|
||||
input: unknown,
|
||||
async function executeParsedSkillBundleInstallRequest(
|
||||
request: SkillBundleInstallRequest,
|
||||
dependencies: SkillBundleInstallRequestDependencies
|
||||
): Promise<SkillBundleInstallResult> {
|
||||
const request = SkillBundleInstallRequestSchema.parse(input)
|
||||
const destination = await resolveSkillInstallDestination(
|
||||
request.destination,
|
||||
dependencies.authority
|
||||
)
|
||||
const ingress = await resolveIngress(request, dependencies)
|
||||
let ingress: StagedSkillBundle | null = null
|
||||
try {
|
||||
const destination = await resolveSkillInstallDestination(
|
||||
request.destination,
|
||||
dependencies.authority
|
||||
)
|
||||
ingress = await resolveIngress(request, dependencies)
|
||||
const detectedProviders = destination.wslDistro
|
||||
? await detectSkillProvidersInWsl(destination.wslDistro)
|
||||
: await dependencies.detectProviders()
|
||||
@@ -105,6 +106,22 @@ export async function executeSkillBundleInstallRequest(
|
||||
onProgress: dependencies.onProgress
|
||||
})
|
||||
} finally {
|
||||
await ingress.cleanup()
|
||||
await ingress?.cleanup()
|
||||
}
|
||||
}
|
||||
|
||||
export async function executeSkillBundleInstallRequest(
|
||||
input: unknown,
|
||||
dependencies: SkillBundleInstallRequestDependencies
|
||||
): Promise<SkillBundleInstallResult> {
|
||||
const request = SkillBundleInstallRequestSchema.parse(input)
|
||||
const operation = startSkillBundleInstallOperation(request)
|
||||
try {
|
||||
const result = await executeParsedSkillBundleInstallRequest(request, dependencies)
|
||||
operation.complete(result)
|
||||
return result
|
||||
} catch (error) {
|
||||
operation.fail(error)
|
||||
throw error
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,30 @@
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import type { SkillBundleInstallResult } from '../../shared/skill-bundle-install-contract'
|
||||
import { summarizeSkillBundleObservation } from './skill-bundle-observability-summary'
|
||||
|
||||
describe('skill bundle observability summary', () => {
|
||||
it('caps distinct error categories while preserving the aggregate count', () => {
|
||||
const result: SkillBundleInstallResult = {
|
||||
operationId: 'operation',
|
||||
packageId: 'package',
|
||||
versionId: 'version',
|
||||
bundleDigest: 'a'.repeat(64),
|
||||
status: 'failed',
|
||||
skills: Array.from({ length: 40 }, (_, index) => ({
|
||||
skillId: `skill-${index}`,
|
||||
name: `skill-${index}`,
|
||||
digest: 'b'.repeat(64),
|
||||
status: 'failed' as const,
|
||||
placements: [],
|
||||
errorCategory: `skill-error-${index}`
|
||||
}))
|
||||
}
|
||||
|
||||
const summary = summarizeSkillBundleObservation(result)
|
||||
|
||||
expect(summary.errorCategories.size).toBe(32)
|
||||
expect([...summary.errorCategories.values()].reduce((total, count) => total + count, 0)).toBe(
|
||||
40
|
||||
)
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,52 @@
|
||||
import type { SkillBundleInstallResult } from '../../shared/skill-bundle-install-contract'
|
||||
|
||||
const MAX_ERROR_CATEGORIES = 32
|
||||
const SAFE_ERROR_CATEGORY = /^[a-z0-9][a-z0-9._-]{0,63}$/i
|
||||
|
||||
export type SkillBundleObservationSummary = {
|
||||
attributes: Record<string, number>
|
||||
errorCategories: Map<string, number>
|
||||
}
|
||||
|
||||
function increment(attributes: Record<string, number>, key: string): void {
|
||||
attributes[key] = (attributes[key] ?? 0) + 1
|
||||
}
|
||||
|
||||
function recordErrorCategory(categories: Map<string, number>, value: string): void {
|
||||
const category = SAFE_ERROR_CATEGORY.test(value) ? value : 'other'
|
||||
const key =
|
||||
categories.has(category) || (category !== 'other' && categories.size < MAX_ERROR_CATEGORIES - 1)
|
||||
? category
|
||||
: 'other'
|
||||
categories.set(key, (categories.get(key) ?? 0) + 1)
|
||||
}
|
||||
|
||||
export function summarizeSkillBundleObservation(
|
||||
result: SkillBundleInstallResult
|
||||
): SkillBundleObservationSummary {
|
||||
const attributes: Record<string, number> = {
|
||||
skillCount: result.skills.length,
|
||||
placementCount: 0,
|
||||
conflictCount: 0
|
||||
}
|
||||
const errorCategories = new Map<string, number>()
|
||||
for (const skill of result.skills) {
|
||||
increment(attributes, `${skill.status}SkillCount`)
|
||||
if (skill.conflict) {
|
||||
increment(attributes, 'conflictCount')
|
||||
increment(attributes, `${skill.conflict.kind}ConflictCount`)
|
||||
}
|
||||
if (skill.errorCategory) {
|
||||
recordErrorCategory(errorCategories, skill.errorCategory)
|
||||
}
|
||||
for (const placement of skill.placements) {
|
||||
increment(attributes, 'placementCount')
|
||||
increment(attributes, `${placement.topology}PlacementCount`)
|
||||
increment(attributes, `${placement.status}PlacementCount`)
|
||||
if (placement.errorCategory) {
|
||||
recordErrorCategory(errorCategories, placement.errorCategory)
|
||||
}
|
||||
}
|
||||
}
|
||||
return { attributes, errorCategories }
|
||||
}
|
||||
@@ -27,6 +27,7 @@ import {
|
||||
import { transferSkillPackageToSshHost } from './skill-ssh-package-transfer'
|
||||
import { retrySkillTransferRpc } from './skill-transfer-rpc-retry'
|
||||
import { startSkillInstallProgressPolling } from './skill-install-progress-polling'
|
||||
import { recordSkillCapabilityAbsence } from './skill-operation-observability'
|
||||
|
||||
export async function installSkillBundleOnSshHost(input: {
|
||||
provider: IPtyProvider
|
||||
@@ -41,6 +42,10 @@ export async function installSkillBundleOnSshHost(input: {
|
||||
const client = requireSkillSshRelayClient(input.provider)
|
||||
const supported = await skillSshRelayCapabilities(client)
|
||||
if (!supported.includes(SKILL_BUNDLE_INSTALL_CAPABILITY)) {
|
||||
recordSkillCapabilityAbsence({
|
||||
capability: SKILL_BUNDLE_INSTALL_CAPABILITY,
|
||||
destination: 'global-ssh'
|
||||
})
|
||||
throw new Error('skill-bundle-ssh-update-required')
|
||||
}
|
||||
const stopProgress =
|
||||
@@ -84,6 +89,10 @@ export async function installSkillBundleOnSshHost(input: {
|
||||
}
|
||||
}
|
||||
if (!supported.includes(SKILL_UPLOAD_CAPABILITY)) {
|
||||
recordSkillCapabilityAbsence({
|
||||
capability: SKILL_UPLOAD_CAPABILITY,
|
||||
destination: 'global-ssh'
|
||||
})
|
||||
throw new Error('skill-bundle-ssh-download-unavailable')
|
||||
}
|
||||
return await retrySkillTransferRpc({
|
||||
|
||||
@@ -9,6 +9,7 @@ import {
|
||||
} from '../../shared/skill-upload-session-contract'
|
||||
import { callRuntimeEnvironment } from '../ipc/runtime-environment-transport-routing'
|
||||
import { downloadSkillPackageGrant } from './skill-package-download'
|
||||
import { startSkillPhaseOperation } from './skill-operation-observability'
|
||||
import { retrySkillTransferRpc, throwIfSkillTransferCancelled } from './skill-transfer-rpc-retry'
|
||||
|
||||
const REMOTE_TRANSFER_TIMEOUT_MS = 5 * 60_000
|
||||
@@ -48,7 +49,7 @@ async function remoteCall<T>(
|
||||
return response.result
|
||||
}
|
||||
|
||||
export async function transferSkillPackageToRuntime(input: {
|
||||
type SkillClientTransferInput = {
|
||||
userDataPath: string
|
||||
environmentId: string
|
||||
transferId: string
|
||||
@@ -56,7 +57,11 @@ export async function transferSkillPackageToRuntime(input: {
|
||||
grant: { url: string; expiresAt: string }
|
||||
requireHttps: boolean
|
||||
signal?: AbortSignal
|
||||
}): Promise<{ uploadId: string; cleanup(): Promise<void> }> {
|
||||
}
|
||||
|
||||
async function transferSkillPackageToRuntimeUnobserved(
|
||||
input: SkillClientTransferInput
|
||||
): Promise<{ uploadId: string; cleanup(): Promise<void> }> {
|
||||
const downloaded = await downloadSkillPackageGrant({
|
||||
url: input.grant.url,
|
||||
expiresAt: input.grant.expiresAt,
|
||||
@@ -165,3 +170,22 @@ export async function transferSkillPackageToRuntime(input: {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
export async function transferSkillPackageToRuntime(
|
||||
input: SkillClientTransferInput
|
||||
): Promise<{ uploadId: string; cleanup(): Promise<void> }> {
|
||||
const operation = startSkillPhaseOperation({
|
||||
phase: 'transfer',
|
||||
transport: 'runtime-rpc',
|
||||
destination: 'remote-runtime',
|
||||
compressedBytes: input.package.compressedBytes
|
||||
})
|
||||
try {
|
||||
const transfer = await transferSkillPackageToRuntimeUnobserved(input)
|
||||
operation.complete({ status: 'complete', compressedBytes: input.package.compressedBytes })
|
||||
return transfer
|
||||
} catch (error) {
|
||||
operation.fail(error)
|
||||
throw error
|
||||
}
|
||||
}
|
||||
|
||||
@@ -17,6 +17,7 @@ import type { SkillCloudDownloadGrant } from '../../shared/skill-cloud-contract'
|
||||
import { getRuntimeEnvironmentStatus } from '../ipc/runtime-environment-transport-routing'
|
||||
import type { OrcaRuntimeService } from '../runtime/orca-runtime'
|
||||
import { skillInstallFailureFromError } from './skill-install-operation-error'
|
||||
import { recordSkillCapabilityAbsence } from './skill-operation-observability'
|
||||
import {
|
||||
installSkillBundleOnRemoteRuntime,
|
||||
installSkillOnRemoteRuntime
|
||||
@@ -104,6 +105,12 @@ export async function installSkillBundleCloudGrant(
|
||||
status.ok !== true ||
|
||||
status.result.capabilities?.includes(SKILL_BUNDLE_INSTALL_CAPABILITY) !== true
|
||||
) {
|
||||
if (status.ok === true) {
|
||||
recordSkillCapabilityAbsence({
|
||||
capability: SKILL_BUNDLE_INSTALL_CAPABILITY,
|
||||
destination: 'remote-runtime'
|
||||
})
|
||||
}
|
||||
return { status: 'unsupported' as const, message: SKILL_INSTALL_UPDATE_REQUIRED_MESSAGE }
|
||||
}
|
||||
return {
|
||||
@@ -160,6 +167,12 @@ export async function installSkillCloudGrant(
|
||||
status.ok !== true ||
|
||||
status.result.capabilities?.includes(SKILL_INSTALL_CAPABILITY) !== true
|
||||
) {
|
||||
if (status.ok === true) {
|
||||
recordSkillCapabilityAbsence({
|
||||
capability: SKILL_INSTALL_CAPABILITY,
|
||||
destination: 'remote-runtime'
|
||||
})
|
||||
}
|
||||
return { status: 'unsupported' as const, message: SKILL_INSTALL_UPDATE_REQUIRED_MESSAGE }
|
||||
}
|
||||
return {
|
||||
|
||||
@@ -15,6 +15,7 @@ import { resolveArtifactCloudApiUrl } from '../artifacts/artifact-cloud-config'
|
||||
import { runSkillCloudOperation } from './skill-cloud-auth'
|
||||
import { uploadSkillPackageToSignedPolicy } from './skill-cloud-direct-upload'
|
||||
import { skillCloudRequest } from './skill-cloud-request'
|
||||
import { startSkillPhaseOperation } from './skill-operation-observability'
|
||||
|
||||
const ID_PATTERN = /^[A-Za-z0-9_-]{1,128}$/
|
||||
|
||||
@@ -62,34 +63,58 @@ export class SkillCloudService {
|
||||
idempotencyKey: randomUUID(),
|
||||
signal: request.signal
|
||||
})
|
||||
await uploadSkillPackageToSignedPolicy({
|
||||
policy: upload.upload.policy,
|
||||
archivePath: request.archivePath,
|
||||
expectedBytes: request.compressedBytes,
|
||||
signal: request.signal,
|
||||
onProgress: (bytesSent) =>
|
||||
request.onProgress?.({
|
||||
phase: 'uploading',
|
||||
bytesSent,
|
||||
totalBytes: request.compressedBytes
|
||||
})
|
||||
const uploadOperation = startSkillPhaseOperation({
|
||||
phase: 'upload',
|
||||
compressedBytes: request.compressedBytes
|
||||
})
|
||||
try {
|
||||
await uploadSkillPackageToSignedPolicy({
|
||||
policy: upload.upload.policy,
|
||||
archivePath: request.archivePath,
|
||||
expectedBytes: request.compressedBytes,
|
||||
signal: request.signal,
|
||||
onProgress: (bytesSent) =>
|
||||
request.onProgress?.({
|
||||
phase: 'uploading',
|
||||
bytesSent,
|
||||
totalBytes: request.compressedBytes
|
||||
})
|
||||
})
|
||||
uploadOperation.complete({
|
||||
status: 'complete',
|
||||
compressedBytes: request.compressedBytes
|
||||
})
|
||||
} catch (error) {
|
||||
uploadOperation.fail(error)
|
||||
throw error
|
||||
}
|
||||
request.onProgress?.({
|
||||
phase: 'finalizing',
|
||||
bytesSent: request.compressedBytes,
|
||||
totalBytes: request.compressedBytes
|
||||
})
|
||||
const finalized = await skillCloudRequest<{ version: SkillCloudVersion }>({
|
||||
apiUrl,
|
||||
authToken: token,
|
||||
path: `/v1/skill-packages/uploads/${id(upload.upload.id)}/finalize`,
|
||||
method: 'POST',
|
||||
body: { releaseNotes: request.releaseNotes },
|
||||
idempotencyKey: upload.upload.id,
|
||||
signal: request.signal
|
||||
const finalizationOperation = startSkillPhaseOperation({
|
||||
phase: 'finalization',
|
||||
compressedBytes: request.compressedBytes
|
||||
})
|
||||
if (finalized.version.packageId !== request.packageId) {
|
||||
throw new Error('skill-cloud-published-package-mismatch')
|
||||
let finalized: { version: SkillCloudVersion }
|
||||
try {
|
||||
finalized = await skillCloudRequest<{ version: SkillCloudVersion }>({
|
||||
apiUrl,
|
||||
authToken: token,
|
||||
path: `/v1/skill-packages/uploads/${id(upload.upload.id)}/finalize`,
|
||||
method: 'POST',
|
||||
body: { releaseNotes: request.releaseNotes },
|
||||
idempotencyKey: upload.upload.id,
|
||||
signal: request.signal
|
||||
})
|
||||
if (finalized.version.packageId !== request.packageId) {
|
||||
throw new Error('skill-cloud-published-package-mismatch')
|
||||
}
|
||||
finalizationOperation.complete({ status: 'complete' })
|
||||
} catch (error) {
|
||||
finalizationOperation.fail(error)
|
||||
throw error
|
||||
}
|
||||
return finalized.version
|
||||
})
|
||||
|
||||
@@ -170,7 +170,8 @@ async function completeSharedSkillInstall(input: {
|
||||
previousReceipt: input.previousReceipt,
|
||||
packageDigest: result.packageDigest,
|
||||
fileModes: input.manifest.files,
|
||||
filesystem: input.input.filesystem
|
||||
filesystem: input.input.filesystem,
|
||||
...(input.input.wslDistro ? { targetPlatform: 'linux' as const } : {})
|
||||
})
|
||||
if (placement) {
|
||||
placements.push(placement)
|
||||
|
||||
@@ -36,6 +36,7 @@ import {
|
||||
skillInstallFailureResult,
|
||||
skillInstallReplacementAllowed
|
||||
} from './skill-install-transaction-result'
|
||||
import { settleObservedSkillTransactionRecovery } from './skill-transaction-recovery-observability'
|
||||
|
||||
export { recoverSkillInstallTransaction } from './skill-install-recovery'
|
||||
export type {
|
||||
@@ -286,9 +287,11 @@ export async function installLocalExtractedSkillPackage(
|
||||
}
|
||||
} catch (error) {
|
||||
if (journal) {
|
||||
await recoverSkillInstallTransaction(input.stateDirectory, canonicalPath, filesystem).catch(
|
||||
() => undefined
|
||||
)
|
||||
await settleObservedSkillTransactionRecovery({
|
||||
rollback: journal.phase === 'backup-created',
|
||||
recover: () =>
|
||||
recoverSkillInstallTransaction(input.stateDirectory, canonicalPath, filesystem)
|
||||
})
|
||||
}
|
||||
const result = skillInstallFailureResult(input, extracted.manifest, canonicalPath, error)
|
||||
if (result) {
|
||||
|
||||
@@ -1,11 +1,25 @@
|
||||
import { mkdtempSync, rmSync, writeFileSync } from 'node:fs'
|
||||
import { mkdirSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from 'node:fs'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
import { afterEach, beforeEach, describe, expect, it } from 'vitest'
|
||||
import type {
|
||||
SkillBundleInstallRequest,
|
||||
SkillBundleInstallResult
|
||||
} from '../../shared/skill-bundle-install-contract'
|
||||
import type { SkillInstallRequest, SkillInstallResult } from '../../shared/skill-install-contract'
|
||||
import { SKILL_PACKAGE_CONTENT_TYPE } from '../../shared/skill-package-manifest'
|
||||
import { collectBundle } from '../observability/bundle'
|
||||
import { _resetTracerForTests, setActiveSink, type TracerSink } from '../observability/tracer'
|
||||
import { startSkillInstallOperation } from './skill-operation-observability'
|
||||
import {
|
||||
recordSkillCapabilityAbsence,
|
||||
startSkillBundleInstallOperation,
|
||||
startSkillInstallOperation,
|
||||
startSkillPhaseOperation
|
||||
} from './skill-operation-observability'
|
||||
import { createSkillPackageArchive } from './skill-package-creation'
|
||||
import { downloadSkillPackageGrant } from './skill-package-download'
|
||||
import { reconcileSkillProviderPlacement } from './skill-placement-reconciliation'
|
||||
import { settleObservedSkillTransactionRecovery } from './skill-transaction-recovery-observability'
|
||||
|
||||
type CapturingSink = TracerSink & { records: unknown[] }
|
||||
|
||||
@@ -60,6 +74,57 @@ function result(): SkillInstallResult {
|
||||
}
|
||||
}
|
||||
|
||||
function bundleRequest(): SkillBundleInstallRequest {
|
||||
return {
|
||||
operationId: 'bundle-operation-private',
|
||||
package: {
|
||||
packageId: 'bundle-package-123',
|
||||
versionId: 'bundle-version-456',
|
||||
bundleDigest: 'c'.repeat(64),
|
||||
archiveSha256: 'd'.repeat(64),
|
||||
compressedBytes: 4321
|
||||
},
|
||||
selectedSkillIds: ['selected-private-skill'],
|
||||
ingress: { kind: 'local-file', path: PRIVATE_VALUES.localPath },
|
||||
destination: {
|
||||
scope: 'global',
|
||||
executionTarget: { kind: 'ssh', connectionId: PRIVATE_VALUES.connectionId }
|
||||
},
|
||||
conflictDecisions: [
|
||||
{ skillId: 'selected-private-skill', resolution: 'replace-and-discard-local' }
|
||||
]
|
||||
}
|
||||
}
|
||||
|
||||
function bundleResult(): SkillBundleInstallResult {
|
||||
return {
|
||||
operationId: 'bundle-operation-private',
|
||||
packageId: 'bundle-package-123',
|
||||
versionId: 'bundle-version-456',
|
||||
bundleDigest: 'c'.repeat(64),
|
||||
status: 'partial',
|
||||
skills: [
|
||||
{
|
||||
skillId: 'selected-private-skill',
|
||||
name: PRIVATE_VALUES.skillName,
|
||||
digest: 'e'.repeat(64),
|
||||
status: 'kept-local',
|
||||
canonicalPath: PRIVATE_VALUES.canonicalPath,
|
||||
placements: [
|
||||
{
|
||||
provider: 'codex',
|
||||
path: PRIVATE_VALUES.providerPath,
|
||||
topology: 'provider-alias',
|
||||
status: 'skipped'
|
||||
}
|
||||
],
|
||||
conflict: { kind: 'modified', existingDigest: 'f'.repeat(64) },
|
||||
errorCategory: PRIVATE_VALUES.manifest
|
||||
}
|
||||
]
|
||||
}
|
||||
}
|
||||
|
||||
let sink: CapturingSink
|
||||
let directory: string
|
||||
|
||||
@@ -124,4 +189,136 @@ describe('skill operation observability', () => {
|
||||
expect(bundle.payload).not.toContain(value)
|
||||
}
|
||||
})
|
||||
|
||||
it('records bundle counts and conflicts without skill identities or paths', () => {
|
||||
const operation = startSkillBundleInstallOperation(bundleRequest())
|
||||
operation.complete(bundleResult())
|
||||
|
||||
const serialized = JSON.stringify(sink.records)
|
||||
expect(serialized).toContain('bundle-package-123')
|
||||
expect(serialized).toContain('selectedSkillCount')
|
||||
expect(serialized).toContain('modified')
|
||||
expect(serialized).toContain('provider-alias')
|
||||
expect(serialized).not.toContain('selected-private-skill')
|
||||
for (const value of Object.values(PRIVATE_VALUES)) {
|
||||
expect(serialized).not.toContain(value)
|
||||
}
|
||||
})
|
||||
|
||||
it('bounds phase counts and records only fixed capability labels', () => {
|
||||
const recovery = startSkillPhaseOperation({ phase: 'recovery', destination: 'startup' })
|
||||
recovery.complete({
|
||||
status: 'partial',
|
||||
scannedCount: Number.MAX_SAFE_INTEGER,
|
||||
recoveredCount: 4,
|
||||
failureCount: 1,
|
||||
orphanCount: 3,
|
||||
truncated: true
|
||||
})
|
||||
const upload = startSkillPhaseOperation({
|
||||
phase: 'upload',
|
||||
compressedBytes: 40 * 1024 * 1024
|
||||
})
|
||||
upload.complete({ status: 'complete' })
|
||||
recordSkillCapabilityAbsence({
|
||||
capability: 'skills.install.bundle.v1',
|
||||
destination: 'remote-runtime'
|
||||
})
|
||||
|
||||
const serialized = JSON.stringify(sink.records)
|
||||
expect(serialized).toContain('1000000')
|
||||
expect(serialized).toContain('41943040')
|
||||
expect(serialized).toContain('orphanCount')
|
||||
expect(serialized).toContain('skills.install.bundle.v1')
|
||||
expect(serialized).toContain('remote-runtime')
|
||||
})
|
||||
|
||||
it('reduces phase failures to a bounded category', () => {
|
||||
const download = startSkillPhaseOperation({
|
||||
phase: 'download',
|
||||
transport: 'download-grant',
|
||||
compressedBytes: 1234
|
||||
})
|
||||
download.fail(new Error(Object.values(PRIVATE_VALUES).join(' ')))
|
||||
|
||||
const serialized = JSON.stringify(sink.records)
|
||||
expect(serialized).toContain('skill-operation-unknown')
|
||||
for (const value of Object.values(PRIVATE_VALUES)) {
|
||||
expect(serialized).not.toContain(value)
|
||||
}
|
||||
})
|
||||
|
||||
it('measures real package, download, and placement phases without local paths', async () => {
|
||||
const source = join(directory, 'private-source')
|
||||
const archivePath = join(directory, 'private-package.tar.gz')
|
||||
mkdirSync(source)
|
||||
writeFileSync(
|
||||
join(source, 'SKILL.md'),
|
||||
'---\nname: observed-skill\ndescription: Private test\n---\n\nPrivate instructions\n'
|
||||
)
|
||||
const created = await createSkillPackageArchive({
|
||||
sourceDirectory: source,
|
||||
archivePath,
|
||||
packageId: 'observed-package',
|
||||
versionId: 'observed-version'
|
||||
})
|
||||
const archive = readFileSync(archivePath)
|
||||
const fetcher: typeof fetch = async () =>
|
||||
new Response(Uint8Array.from(archive), {
|
||||
headers: {
|
||||
'content-type': SKILL_PACKAGE_CONTENT_TYPE,
|
||||
'content-length': String(archive.length)
|
||||
}
|
||||
})
|
||||
const downloaded = await downloadSkillPackageGrant({
|
||||
url: 'https://storage.test/package',
|
||||
expiresAt: '2030-01-01T00:00:00Z',
|
||||
expectedArchiveSha256: created.archiveSha256,
|
||||
expectedCompressedBytes: created.compressedBytes,
|
||||
temporaryRoot: join(directory, 'private-downloads'),
|
||||
allowedOrigins: ['https://storage.test'],
|
||||
requireHttps: true,
|
||||
fetcher
|
||||
})
|
||||
await downloaded.cleanup()
|
||||
|
||||
const providerRoot = join(directory, 'private-provider')
|
||||
const placement = await reconcileSkillProviderPlacement({
|
||||
canonicalPath: source,
|
||||
skillName: 'observed-skill',
|
||||
destination: { provider: 'claude', readsCanonicalRoot: false, rootPath: providerRoot },
|
||||
previousReceipt: null,
|
||||
packageDigest: created.manifest.packageDigest
|
||||
})
|
||||
|
||||
expect(placement?.status).toBe('installed')
|
||||
const serialized = JSON.stringify(sink.records)
|
||||
expect(serialized).toContain('skill.package')
|
||||
expect(serialized).toContain('skill.download')
|
||||
expect(serialized).toContain('skill.placement')
|
||||
expect(serialized).toContain('fileCount')
|
||||
expect(serialized).not.toContain(source)
|
||||
expect(serialized).not.toContain(archivePath)
|
||||
expect(serialized).not.toContain(providerRoot)
|
||||
expect(serialized).not.toContain('observed-skill')
|
||||
expect(serialized).not.toContain('Private instructions')
|
||||
})
|
||||
|
||||
it('records transaction rollback settlement without exposing recovery errors', async () => {
|
||||
await settleObservedSkillTransactionRecovery({
|
||||
rollback: true,
|
||||
recover: async () => undefined
|
||||
})
|
||||
await settleObservedSkillTransactionRecovery({
|
||||
rollback: false,
|
||||
recover: async () => {
|
||||
throw new Error(`${PRIVATE_VALUES.canonicalPath} recovery failed`)
|
||||
}
|
||||
})
|
||||
|
||||
const serialized = JSON.stringify(sink.records)
|
||||
expect(serialized).toContain('rollbackCount')
|
||||
expect(serialized).toContain('skill-operation-unknown')
|
||||
expect(serialized).not.toContain(PRIVATE_VALUES.canonicalPath)
|
||||
})
|
||||
})
|
||||
|
||||
@@ -4,27 +4,167 @@ import type {
|
||||
SkillInstallResult,
|
||||
SkillPlacementResult
|
||||
} from '../../shared/skill-install-contract'
|
||||
import type {
|
||||
SkillBundleInstallRequest,
|
||||
SkillBundleInstallResult
|
||||
} from '../../shared/skill-bundle-install-contract'
|
||||
import { startSpan, type ActiveSpan } from '../observability/tracer'
|
||||
import { summarizeSkillBundleObservation } from './skill-bundle-observability-summary'
|
||||
import { skillInstallFailureFromError } from './skill-install-operation-error'
|
||||
|
||||
const SAFE_LABEL = /^[a-z0-9][a-z0-9._-]{0,63}$/i
|
||||
const MAX_OBSERVED_COUNT = 1_000_000
|
||||
const MAX_OBSERVED_BYTES = 64 * 1024 * 1024
|
||||
|
||||
type SkillOperationPhase =
|
||||
| 'package'
|
||||
| 'upload'
|
||||
| 'finalization'
|
||||
| 'download'
|
||||
| 'transfer'
|
||||
| 'placement'
|
||||
| 'recovery'
|
||||
|
||||
type SkillPhaseStart = {
|
||||
phase: SkillOperationPhase
|
||||
platform?: 'darwin' | 'linux' | 'win32' | 'other'
|
||||
packageKind?: 'single' | 'bundle'
|
||||
transport?: 'download-grant' | 'staged-upload' | 'local-file' | 'runtime-rpc' | 'ssh-relay'
|
||||
destination?: 'startup' | 'transaction' | 'provider-placement' | 'remote-runtime' | 'global-ssh'
|
||||
compressedBytes?: number
|
||||
skillCount?: number
|
||||
provider?: 'codex' | 'claude'
|
||||
}
|
||||
|
||||
type SkillPhaseOutcome = {
|
||||
status?: 'complete' | 'partial' | 'installed' | 'unchanged' | 'removed' | 'skipped' | 'failed'
|
||||
errorCategory?: string
|
||||
fileCount?: number
|
||||
totalBytes?: number
|
||||
compressedBytes?: number
|
||||
chunkCount?: number
|
||||
scannedCount?: number
|
||||
recoveredCount?: number
|
||||
failureCount?: number
|
||||
orphanCount?: number
|
||||
rollbackCount?: number
|
||||
topology?: SkillPlacementResult['topology']
|
||||
aliasMechanism?: 'symlink' | 'junction' | 'filesystem' | 'none'
|
||||
copyFallbackCount?: number
|
||||
truncated?: boolean
|
||||
}
|
||||
|
||||
export type SkillPhaseOperationSpan = {
|
||||
complete(outcome?: SkillPhaseOutcome): void
|
||||
fail(error: unknown): void
|
||||
}
|
||||
|
||||
export type SkillInstallOperationSpan = {
|
||||
complete(result: SkillInstallResult): void
|
||||
fail(error: unknown): void
|
||||
}
|
||||
|
||||
export type SkillBundleInstallOperationSpan = {
|
||||
complete(result: SkillBundleInstallResult): void
|
||||
fail(error: unknown): void
|
||||
}
|
||||
|
||||
function safeLabel(value: string | undefined, fallback: string): string {
|
||||
return value && SAFE_LABEL.test(value) ? value : fallback
|
||||
}
|
||||
|
||||
function destinationLabel(destination: SkillInstallDestination): string {
|
||||
function boundedNumber(value: number, maximum = MAX_OBSERVED_COUNT): number {
|
||||
return Number.isFinite(value) ? Math.min(Math.max(Math.trunc(value), 0), maximum) : 0
|
||||
}
|
||||
|
||||
function platformLabel(destination?: SkillInstallDestination): string {
|
||||
if (destination?.scope === 'global' && destination.executionTarget?.kind === 'wsl') {
|
||||
return 'linux'
|
||||
}
|
||||
return ['darwin', 'linux', 'win32'].includes(process.platform) ? process.platform : 'other'
|
||||
}
|
||||
|
||||
export function skillInstallDestinationLabel(destination: SkillInstallDestination): string {
|
||||
if (destination.scope === 'workspace') {
|
||||
return 'workspace'
|
||||
}
|
||||
return `global-${destination.executionTarget?.kind ?? 'host'}`
|
||||
}
|
||||
|
||||
function failureLabel(
|
||||
error: unknown,
|
||||
fallback = 'skill-operation-unknown'
|
||||
): { category: 'cancelled' | 'failed'; code: string } {
|
||||
const failure = skillInstallFailureFromError(error)
|
||||
const message = error instanceof Error ? error.message : undefined
|
||||
return {
|
||||
category: failure?.category === 'cancelled' ? 'cancelled' : 'failed',
|
||||
code: safeLabel(failure?.code ?? message, fallback)
|
||||
}
|
||||
}
|
||||
|
||||
function setOutcome(span: ActiveSpan, outcome: SkillPhaseOutcome): void {
|
||||
for (const [key, value] of Object.entries(outcome)) {
|
||||
span.setAttribute(
|
||||
key,
|
||||
key === 'errorCategory' && typeof value === 'string'
|
||||
? safeLabel(value, 'skill-operation-unknown')
|
||||
: typeof value === 'number'
|
||||
? boundedNumber(value, key.endsWith('Bytes') ? MAX_OBSERVED_BYTES : MAX_OBSERVED_COUNT)
|
||||
: value
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
export function startSkillPhaseOperation(input: SkillPhaseStart): SkillPhaseOperationSpan {
|
||||
const span = startSpan(`skill.${input.phase}`, {
|
||||
attributes: {
|
||||
phase: input.phase,
|
||||
platform: input.platform ?? platformLabel(),
|
||||
...(input.packageKind ? { packageKind: input.packageKind } : {}),
|
||||
...(input.transport ? { transport: input.transport } : {}),
|
||||
...(input.destination ? { destination: safeLabel(input.destination, 'unknown') } : {}),
|
||||
...(input.compressedBytes === undefined
|
||||
? {}
|
||||
: { compressedBytes: boundedNumber(input.compressedBytes, MAX_OBSERVED_BYTES) }),
|
||||
...(input.skillCount === undefined ? {} : { skillCount: boundedNumber(input.skillCount) }),
|
||||
...(input.provider ? { provider: safeLabel(input.provider, 'unknown') } : {})
|
||||
}
|
||||
})
|
||||
return {
|
||||
complete(outcome = {}) {
|
||||
setOutcome(span, outcome)
|
||||
span.end()
|
||||
},
|
||||
fail(error) {
|
||||
const failure = failureLabel(error)
|
||||
span.setAttribute('status', failure.category)
|
||||
span.setAttribute('errorCategory', failure.code)
|
||||
if (failure.category === 'cancelled') {
|
||||
span.interrupt(failure.code)
|
||||
} else {
|
||||
span.fail(failure.code)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
export function recordSkillCapabilityAbsence(input: {
|
||||
capability: 'skills.install.v1' | 'skills.install.bundle.v1' | 'skills.upload.v1'
|
||||
destination: 'remote-runtime' | 'global-ssh'
|
||||
}): void {
|
||||
const span = startSpan('skill.capability', {
|
||||
attributes: {
|
||||
phase: 'capability',
|
||||
platform: platformLabel(),
|
||||
capability: safeLabel(input.capability, 'unknown'),
|
||||
destination: safeLabel(input.destination, 'unknown'),
|
||||
status: 'absent'
|
||||
}
|
||||
})
|
||||
span.end()
|
||||
}
|
||||
|
||||
function recordPlacement(span: ActiveSpan, placement: SkillPlacementResult): void {
|
||||
span.addEvent('skill.placement', {
|
||||
provider: safeLabel(placement.provider, 'unknown'),
|
||||
@@ -39,10 +179,11 @@ export function startSkillInstallOperation(
|
||||
): SkillInstallOperationSpan {
|
||||
const span = startSpan('skill.install', {
|
||||
attributes: {
|
||||
packageId: request.package.packageId,
|
||||
versionId: request.package.versionId,
|
||||
packageId: safeLabel(request.package.packageId, 'unknown'),
|
||||
versionId: safeLabel(request.package.versionId, 'unknown'),
|
||||
phase: 'install',
|
||||
destination: destinationLabel(request.destination),
|
||||
platform: platformLabel(request.destination),
|
||||
destination: skillInstallDestinationLabel(request.destination),
|
||||
transport: request.ingress.kind,
|
||||
compressedBytes: request.package.compressedBytes
|
||||
}
|
||||
@@ -51,6 +192,15 @@ export function startSkillInstallOperation(
|
||||
complete(result) {
|
||||
span.setAttribute('status', result.status)
|
||||
span.setAttribute('placementCount', result.placements.length)
|
||||
span.setAttribute('conflictType', result.conflict?.kind ?? 'none')
|
||||
span.setAttribute(
|
||||
'aliasPlacementCount',
|
||||
result.placements.filter((placement) => placement.topology === 'provider-alias').length
|
||||
)
|
||||
span.setAttribute(
|
||||
'copyPlacementCount',
|
||||
result.placements.filter((placement) => placement.topology === 'independent-copy').length
|
||||
)
|
||||
for (const placement of result.placements) {
|
||||
recordPlacement(span, placement)
|
||||
}
|
||||
@@ -63,13 +213,63 @@ export function startSkillInstallOperation(
|
||||
}
|
||||
},
|
||||
fail(error) {
|
||||
const failure = skillInstallFailureFromError(error)
|
||||
const code = safeLabel(failure?.code, 'skill-install-unknown')
|
||||
span.setAttribute('status', failure?.category === 'cancelled' ? 'cancelled' : 'failed')
|
||||
if (failure?.category === 'cancelled') {
|
||||
span.interrupt(code)
|
||||
const failure = failureLabel(error, 'skill-install-unknown')
|
||||
span.setAttribute('status', failure.category)
|
||||
span.setAttribute('errorCategory', failure.code)
|
||||
if (failure.category === 'cancelled') {
|
||||
span.interrupt(failure.code)
|
||||
} else {
|
||||
span.fail(code)
|
||||
span.fail(failure.code)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
export function startSkillBundleInstallOperation(
|
||||
request: SkillBundleInstallRequest
|
||||
): SkillBundleInstallOperationSpan {
|
||||
const span = startSpan('skill.install', {
|
||||
attributes: {
|
||||
packageId: safeLabel(request.package.packageId, 'unknown'),
|
||||
versionId: safeLabel(request.package.versionId, 'unknown'),
|
||||
packageKind: 'bundle',
|
||||
phase: 'install',
|
||||
platform: platformLabel(request.destination),
|
||||
destination: skillInstallDestinationLabel(request.destination),
|
||||
transport: request.ingress.kind,
|
||||
compressedBytes: request.package.compressedBytes,
|
||||
selectedSkillCount: boundedNumber(request.selectedSkillIds.length)
|
||||
}
|
||||
})
|
||||
return {
|
||||
complete(result) {
|
||||
span.setAttribute('status', result.status)
|
||||
const summary = summarizeSkillBundleObservation(result)
|
||||
for (const [key, value] of Object.entries(summary.attributes)) {
|
||||
span.setAttribute(key, boundedNumber(value))
|
||||
}
|
||||
for (const [category, count] of summary.errorCategories) {
|
||||
span.addEvent('skill.error-category', {
|
||||
category: safeLabel(category, 'skill-bundle-install-unknown'),
|
||||
count: boundedNumber(count)
|
||||
})
|
||||
}
|
||||
if (result.status === 'cancelled') {
|
||||
span.interrupt('skill-bundle-install-cancelled')
|
||||
} else if (result.status === 'failed') {
|
||||
span.fail('skill-bundle-install-failed')
|
||||
} else {
|
||||
span.end()
|
||||
}
|
||||
},
|
||||
fail(error) {
|
||||
const failure = failureLabel(error, 'skill-bundle-install-unknown')
|
||||
span.setAttribute('status', failure.category)
|
||||
span.setAttribute('errorCategory', failure.code)
|
||||
if (failure.category === 'cancelled') {
|
||||
span.interrupt(failure.code)
|
||||
} else {
|
||||
span.fail(failure.code)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -10,6 +10,7 @@ import {
|
||||
} from '../../shared/skill-package-manifest'
|
||||
import { summarizeSkillMarkdown } from '../../shared/skill-metadata'
|
||||
import { renameSkillPathWithWindowsRetry } from './skill-filesystem-retry'
|
||||
import { startSkillPhaseOperation } from './skill-operation-observability'
|
||||
import { observeSkillPackage, type ObservedSkillPackage } from './skill-package-identity'
|
||||
import { extractSkillPackageArchive } from './skill-package-extraction'
|
||||
import { writeSkillTarGzip, type SkillTarWriteEntry } from './skill-package-tar'
|
||||
@@ -70,7 +71,7 @@ function packageManifest(input: {
|
||||
})
|
||||
}
|
||||
|
||||
export async function createSkillPackageArchive(
|
||||
async function createSkillPackageArchiveUnobserved(
|
||||
input: {
|
||||
sourceDirectory: string
|
||||
archivePath: string
|
||||
@@ -145,3 +146,33 @@ export async function createSkillPackageArchive(
|
||||
await rm(temporaryArchive, { force: true }).catch(() => undefined)
|
||||
}
|
||||
}
|
||||
|
||||
export async function createSkillPackageArchive(
|
||||
input: {
|
||||
sourceDirectory: string
|
||||
archivePath: string
|
||||
packageId: string
|
||||
versionId: string
|
||||
createdAt?: string
|
||||
},
|
||||
dependencies: SkillPackageCreationDependencies = {}
|
||||
): Promise<CreatedSkillPackage> {
|
||||
const operation = startSkillPhaseOperation({
|
||||
phase: 'package',
|
||||
packageKind: 'single',
|
||||
skillCount: 1
|
||||
})
|
||||
try {
|
||||
const created = await createSkillPackageArchiveUnobserved(input, dependencies)
|
||||
operation.complete({
|
||||
status: 'complete',
|
||||
fileCount: created.manifest.files.length,
|
||||
totalBytes: created.manifest.files.reduce((total, file) => total + file.size, 0),
|
||||
compressedBytes: created.compressedBytes
|
||||
})
|
||||
return created
|
||||
} catch (error) {
|
||||
operation.fail(error)
|
||||
throw error
|
||||
}
|
||||
}
|
||||
|
||||
@@ -7,6 +7,7 @@ import {
|
||||
} from '../../shared/skill-package-manifest'
|
||||
import { SKILL_INSTALL_CANCELLED_FAILURE } from '../../shared/skill-install-failure'
|
||||
import { SkillInstallOperationError } from './skill-install-operation-error'
|
||||
import { startSkillPhaseOperation } from './skill-operation-observability'
|
||||
|
||||
const REDIRECT_STATUSES = new Set([301, 302, 303, 307, 308])
|
||||
const MAX_REDIRECTS = 3
|
||||
@@ -117,7 +118,7 @@ function hashesEqual(actual: string, expected: string): boolean {
|
||||
return timingSafeEqual(Buffer.from(actual, 'hex'), Buffer.from(expected, 'hex'))
|
||||
}
|
||||
|
||||
export async function downloadSkillPackageGrant(
|
||||
async function downloadSkillPackageGrantUnobserved(
|
||||
input: SkillPackageDownloadInput
|
||||
): Promise<SkillPackageDownloadResult> {
|
||||
input.fetcher ??= fetch
|
||||
@@ -211,3 +212,21 @@ export async function downloadSkillPackageGrant(
|
||||
throw error
|
||||
}
|
||||
}
|
||||
|
||||
export async function downloadSkillPackageGrant(
|
||||
input: SkillPackageDownloadInput
|
||||
): Promise<SkillPackageDownloadResult> {
|
||||
const operation = startSkillPhaseOperation({
|
||||
phase: 'download',
|
||||
transport: 'download-grant',
|
||||
compressedBytes: input.expectedCompressedBytes
|
||||
})
|
||||
try {
|
||||
const downloaded = await downloadSkillPackageGrantUnobserved(input)
|
||||
operation.complete({ status: 'complete', compressedBytes: downloaded.compressedBytes })
|
||||
return downloaded
|
||||
} catch (error) {
|
||||
operation.fail(error)
|
||||
throw error
|
||||
}
|
||||
}
|
||||
|
||||
@@ -12,6 +12,9 @@ import {
|
||||
createSkillPlacementCopyAtMissingDestination,
|
||||
replaceOwnedSkillPlacementCopy
|
||||
} from './skill-placement-copy'
|
||||
import { startSkillPhaseOperation } from './skill-operation-observability'
|
||||
|
||||
type PlacementObservation = { copyFallback: boolean }
|
||||
|
||||
function normalizedPath(path: string): string {
|
||||
const normalized = resolve(path)
|
||||
@@ -70,6 +73,7 @@ async function reconcileExistingPlacement(input: {
|
||||
packageDigest: string
|
||||
filesystem: SkillInstallFilesystem
|
||||
fileModes?: readonly SkillInstalledFileMode[]
|
||||
observation: PlacementObservation
|
||||
}): Promise<SkillPlacementResult> {
|
||||
const previous = previousPlacement(
|
||||
input.previousReceipt,
|
||||
@@ -114,7 +118,8 @@ async function reconcileExistingPlacement(input: {
|
||||
destinationPath: input.destinationPath,
|
||||
destination: input.destination,
|
||||
filesystem: input.filesystem,
|
||||
fileModes: input.fileModes
|
||||
fileModes: input.fileModes,
|
||||
observation: input.observation
|
||||
})
|
||||
}
|
||||
return {
|
||||
@@ -174,6 +179,7 @@ async function createMissingPlacement(input: {
|
||||
destination: SkillProviderDestination
|
||||
filesystem: SkillInstallFilesystem
|
||||
fileModes?: readonly SkillInstalledFileMode[]
|
||||
observation: PlacementObservation
|
||||
}): Promise<SkillPlacementResult> {
|
||||
try {
|
||||
await createProviderAlias(input.canonicalPath, input.destinationPath, input.filesystem)
|
||||
@@ -184,6 +190,7 @@ async function createMissingPlacement(input: {
|
||||
status: 'installed'
|
||||
}
|
||||
} catch {
|
||||
input.observation.copyFallback = true
|
||||
try {
|
||||
await createSkillPlacementCopyAtMissingDestination(
|
||||
input.canonicalPath,
|
||||
@@ -217,23 +224,59 @@ export async function reconcileSkillProviderPlacement(input: {
|
||||
packageDigest: string
|
||||
filesystem?: SkillInstallFilesystem
|
||||
fileModes?: readonly SkillInstalledFileMode[]
|
||||
targetPlatform?: 'darwin' | 'linux' | 'win32' | 'other'
|
||||
}): Promise<SkillPlacementResult | null> {
|
||||
if (input.destination.readsCanonicalRoot) {
|
||||
return null
|
||||
}
|
||||
const destinationPath = join(input.destination.rootPath, input.skillName)
|
||||
const filesystem = input.filesystem ?? nativeSkillInstallFilesystem
|
||||
const observation: PlacementObservation = { copyFallback: false }
|
||||
const operation = startSkillPhaseOperation({
|
||||
phase: 'placement',
|
||||
platform: input.targetPlatform,
|
||||
destination: 'provider-placement',
|
||||
provider: input.destination.provider
|
||||
})
|
||||
try {
|
||||
return (await pathExists(destinationPath))
|
||||
? await reconcileExistingPlacement({ ...input, destinationPath, filesystem })
|
||||
: await createMissingPlacement({ ...input, destinationPath, filesystem })
|
||||
const result = (await pathExists(destinationPath))
|
||||
? await reconcileExistingPlacement({
|
||||
...input,
|
||||
destinationPath,
|
||||
filesystem,
|
||||
observation
|
||||
})
|
||||
: await createMissingPlacement({ ...input, destinationPath, filesystem, observation })
|
||||
operation.complete({
|
||||
status: result.status,
|
||||
errorCategory: result.errorCategory ?? 'none',
|
||||
topology: result.topology,
|
||||
aliasMechanism:
|
||||
result.topology !== 'provider-alias'
|
||||
? 'none'
|
||||
: filesystem.createAlias
|
||||
? 'filesystem'
|
||||
: process.platform === 'win32'
|
||||
? 'junction'
|
||||
: 'symlink',
|
||||
copyFallbackCount: observation.copyFallback ? 1 : 0
|
||||
})
|
||||
return result
|
||||
} catch {
|
||||
return {
|
||||
const result: SkillPlacementResult = {
|
||||
provider: input.destination.provider,
|
||||
path: destinationPath,
|
||||
topology: 'independent-copy',
|
||||
status: 'failed',
|
||||
...placementFailure('skill-placement-reconciliation-failed', true)
|
||||
}
|
||||
operation.complete({
|
||||
status: result.status,
|
||||
errorCategory: result.errorCategory,
|
||||
topology: result.topology,
|
||||
aliasMechanism: 'none',
|
||||
copyFallbackCount: observation.copyFallback ? 1 : 0
|
||||
})
|
||||
return result
|
||||
}
|
||||
}
|
||||
|
||||
@@ -21,6 +21,7 @@ import { callRuntimeEnvironment } from '../ipc/runtime-environment-transport-rou
|
||||
import { transferSkillPackageToRuntime } from './skill-client-mediated-transfer'
|
||||
import { SkillInstallFailureSchema } from '../../shared/skill-install-failure'
|
||||
import { SkillInstallOperationError } from './skill-install-operation-error'
|
||||
import { recordSkillCapabilityAbsence } from './skill-operation-observability'
|
||||
import { startSkillInstallProgressPolling } from './skill-install-progress-polling'
|
||||
import { retrySkillTransferRpc } from './skill-transfer-rpc-retry'
|
||||
import {
|
||||
@@ -135,6 +136,10 @@ export async function installSkillOnRemoteRuntime(input: {
|
||||
return SkillInstallResultSchema.parse(direct.result)
|
||||
}
|
||||
if (!input.capabilities.includes(SKILL_UPLOAD_CAPABILITY)) {
|
||||
recordSkillCapabilityAbsence({
|
||||
capability: SKILL_UPLOAD_CAPABILITY,
|
||||
destination: 'remote-runtime'
|
||||
})
|
||||
throw new Error('skill-install-remote-download-unavailable')
|
||||
}
|
||||
|
||||
@@ -181,10 +186,14 @@ export async function installSkillBundleOnRemoteRuntime(input: {
|
||||
signal?: AbortSignal
|
||||
onProgress?: (progress: SkillBundleInstallProgress) => void
|
||||
}): Promise<SkillBundleInstallResult> {
|
||||
if (
|
||||
input.request.ingress.kind !== 'download-grant' ||
|
||||
!input.capabilities.includes(SKILL_BUNDLE_INSTALL_CAPABILITY)
|
||||
) {
|
||||
if (input.request.ingress.kind !== 'download-grant') {
|
||||
throw new Error('skill-bundle-remote-ingress-invalid')
|
||||
}
|
||||
if (!input.capabilities.includes(SKILL_BUNDLE_INSTALL_CAPABILITY)) {
|
||||
recordSkillCapabilityAbsence({
|
||||
capability: SKILL_BUNDLE_INSTALL_CAPABILITY,
|
||||
destination: 'remote-runtime'
|
||||
})
|
||||
throw new Error('skill-bundle-remote-ingress-invalid')
|
||||
}
|
||||
const stopProgress =
|
||||
@@ -214,6 +223,10 @@ export async function installSkillBundleOnRemoteRuntime(input: {
|
||||
return SkillBundleInstallResultSchema.parse(direct.result)
|
||||
}
|
||||
if (!input.capabilities.includes(SKILL_UPLOAD_CAPABILITY)) {
|
||||
recordSkillCapabilityAbsence({
|
||||
capability: SKILL_UPLOAD_CAPABILITY,
|
||||
destination: 'remote-runtime'
|
||||
})
|
||||
throw new Error('skill-bundle-remote-download-unavailable')
|
||||
}
|
||||
return retrySkillTransferRpc({
|
||||
|
||||
@@ -13,6 +13,7 @@ import {
|
||||
SkillUploadBeginResultSchema
|
||||
} from '../../shared/skill-upload-session-contract'
|
||||
import { downloadSkillPackageGrant } from './skill-package-download'
|
||||
import { startSkillPhaseOperation } from './skill-operation-observability'
|
||||
import {
|
||||
SKILL_SSH_REQUEST_TIMEOUT_MS,
|
||||
retryableSkillSshTransportError,
|
||||
@@ -40,7 +41,7 @@ function allowedOrigins(requireHttps: boolean): string[] {
|
||||
return [...new Set(origins)]
|
||||
}
|
||||
|
||||
export async function transferSkillPackageToSshHost(
|
||||
async function transferSkillPackageToSshHostUnobserved(
|
||||
client: SkillSshRelayClient,
|
||||
input: SkillSshTransferInput
|
||||
): Promise<string> {
|
||||
@@ -132,3 +133,26 @@ export async function transferSkillPackageToSshHost(
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
export async function transferSkillPackageToSshHost(
|
||||
client: SkillSshRelayClient,
|
||||
input: SkillSshTransferInput
|
||||
): Promise<string> {
|
||||
const operation = startSkillPhaseOperation({
|
||||
phase: 'transfer',
|
||||
transport: 'ssh-relay',
|
||||
destination: 'global-ssh',
|
||||
compressedBytes: input.request.package.compressedBytes
|
||||
})
|
||||
try {
|
||||
const uploadId = await transferSkillPackageToSshHostUnobserved(client, input)
|
||||
operation.complete({
|
||||
status: 'complete',
|
||||
compressedBytes: input.request.package.compressedBytes
|
||||
})
|
||||
return uploadId
|
||||
} catch (error) {
|
||||
operation.fail(error)
|
||||
throw error
|
||||
}
|
||||
}
|
||||
|
||||
@@ -23,6 +23,7 @@ import {
|
||||
type SkillSshWorkspaceAuthority
|
||||
} from '../../shared/skill-ssh-relay-contract'
|
||||
import type { IPtyProvider } from '../providers/pty-provider-contract'
|
||||
import { recordSkillCapabilityAbsence } from './skill-operation-observability'
|
||||
import { retrySkillTransferRpc } from './skill-transfer-rpc-retry'
|
||||
import { transferSkillPackageToSshHost } from './skill-ssh-package-transfer'
|
||||
import {
|
||||
@@ -51,6 +52,10 @@ export async function installSkillOnSshHost(input: {
|
||||
const client = requireSkillSshRelayClient(input.provider)
|
||||
const supported = await skillSshRelayCapabilities(client)
|
||||
if (!supported.includes(SKILL_INSTALL_CAPABILITY)) {
|
||||
recordSkillCapabilityAbsence({
|
||||
capability: SKILL_INSTALL_CAPABILITY,
|
||||
destination: 'global-ssh'
|
||||
})
|
||||
throw new Error('skill-install-ssh-update-required')
|
||||
}
|
||||
try {
|
||||
@@ -75,6 +80,10 @@ export async function installSkillOnSshHost(input: {
|
||||
}
|
||||
}
|
||||
if (!supported.includes(SKILL_UPLOAD_CAPABILITY)) {
|
||||
recordSkillCapabilityAbsence({
|
||||
capability: SKILL_UPLOAD_CAPABILITY,
|
||||
destination: 'global-ssh'
|
||||
})
|
||||
throw new Error('skill-install-ssh-download-unavailable')
|
||||
}
|
||||
return retrySkillTransferRpc({
|
||||
|
||||
@@ -0,0 +1,18 @@
|
||||
import { startSkillPhaseOperation } from './skill-operation-observability'
|
||||
|
||||
export async function settleObservedSkillTransactionRecovery(input: {
|
||||
rollback: boolean
|
||||
recover(): Promise<void>
|
||||
}): Promise<void> {
|
||||
const operation = startSkillPhaseOperation({ phase: 'recovery', destination: 'transaction' })
|
||||
try {
|
||||
await input.recover()
|
||||
operation.complete({
|
||||
status: 'complete',
|
||||
recoveredCount: 1,
|
||||
rollbackCount: input.rollback ? 1 : 0
|
||||
})
|
||||
} catch (error) {
|
||||
operation.fail(error)
|
||||
}
|
||||
}
|
||||
@@ -16,6 +16,7 @@ import {
|
||||
readSkillRemovalRecoveryJournal,
|
||||
recoverSkillRemovalTransaction
|
||||
} from './skill-remove-recovery'
|
||||
import { startSkillPhaseOperation } from './skill-operation-observability'
|
||||
import { WslSkillInstallFilesystem } from './skill-wsl-install-filesystem'
|
||||
|
||||
const MAX_PENDING_TRANSACTION_JOURNALS = 64
|
||||
@@ -31,6 +32,8 @@ type PendingTransaction = {
|
||||
export type SkillTransactionStartupRecoveryReport = {
|
||||
scanned: number
|
||||
recovered: number
|
||||
orphanedExtractionsRecovered: number
|
||||
orphanedLocksReclaimed: number
|
||||
failures: { journalKey: string; code: string }[]
|
||||
truncated: boolean
|
||||
}
|
||||
@@ -119,7 +122,7 @@ function pendingTransactions(
|
||||
return [...pending.values()]
|
||||
}
|
||||
|
||||
export async function recoverPendingSkillTransactions(
|
||||
async function recoverPendingSkillTransactionsUnobserved(
|
||||
stateDirectory: string
|
||||
): Promise<SkillTransactionStartupRecoveryReport> {
|
||||
const [installs, removals, extractions, locks] = await Promise.all([
|
||||
@@ -131,6 +134,8 @@ export async function recoverPendingSkillTransactions(
|
||||
const report: SkillTransactionStartupRecoveryReport = {
|
||||
scanned: installs.candidates.length + removals.candidates.length + extractions.scanned,
|
||||
recovered: extractions.recovered,
|
||||
orphanedExtractionsRecovered: extractions.recovered,
|
||||
orphanedLocksReclaimed: locks.reclaimed,
|
||||
failures: [...installs.failures, ...removals.failures, ...extractions.failures],
|
||||
truncated: installs.truncated || removals.truncated || extractions.truncated || locks.truncated
|
||||
}
|
||||
@@ -179,3 +184,24 @@ export async function recoverPendingSkillTransactions(
|
||||
}
|
||||
return report
|
||||
}
|
||||
|
||||
export async function recoverPendingSkillTransactions(
|
||||
stateDirectory: string
|
||||
): Promise<SkillTransactionStartupRecoveryReport> {
|
||||
const operation = startSkillPhaseOperation({ phase: 'recovery', destination: 'startup' })
|
||||
try {
|
||||
const report = await recoverPendingSkillTransactionsUnobserved(stateDirectory)
|
||||
operation.complete({
|
||||
status: report.failures.length || report.truncated ? 'partial' : 'complete',
|
||||
scannedCount: report.scanned,
|
||||
recoveredCount: report.recovered,
|
||||
failureCount: report.failures.length,
|
||||
orphanCount: report.orphanedExtractionsRecovered + report.orphanedLocksReclaimed,
|
||||
truncated: report.truncated
|
||||
})
|
||||
return report
|
||||
} catch (error) {
|
||||
operation.fail(error)
|
||||
throw error
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user