From 4f9b32f58ce37268fe1fad1a095e10dde752fe98 Mon Sep 17 00:00:00 2001 From: Jinwoo-H Date: Wed, 12 Aug 2026 12:46:11 -0400 Subject: [PATCH] feat(skills): add bounded operation observability --- ...-skill-sharing-implementation-checklist.md | 18 +- docs/agent-skill-sharing-installation-plan.md | 12 + src/main/skills/skill-bundle-creation.ts | 36 ++- .../skill-bundle-install-request-service.ts | 35 ++- ...skill-bundle-observability-summary.test.ts | 30 +++ .../skill-bundle-observability-summary.ts | 52 +++++ .../skills/skill-bundle-ssh-relay-service.ts | 9 + .../skills/skill-client-mediated-transfer.ts | 28 ++- .../skills/skill-cloud-grant-installation.ts | 13 ++ src/main/skills/skill-cloud-service.ts | 67 ++++-- src/main/skills/skill-install-service.ts | 3 +- src/main/skills/skill-install-transaction.ts | 9 +- .../skill-operation-observability.test.ts | 201 +++++++++++++++- .../skills/skill-operation-observability.ts | 220 +++++++++++++++++- src/main/skills/skill-package-creation.ts | 33 ++- src/main/skills/skill-package-download.ts | 21 +- .../skills/skill-placement-reconciliation.ts | 53 ++++- .../skills/skill-remote-install-service.ts | 21 +- src/main/skills/skill-ssh-package-transfer.ts | 26 ++- src/main/skills/skill-ssh-relay-service.ts | 9 + ...kill-transaction-recovery-observability.ts | 18 ++ .../skill-transaction-startup-recovery.ts | 28 ++- 22 files changed, 877 insertions(+), 65 deletions(-) create mode 100644 src/main/skills/skill-bundle-observability-summary.test.ts create mode 100644 src/main/skills/skill-bundle-observability-summary.ts create mode 100644 src/main/skills/skill-transaction-recovery-observability.ts diff --git a/docs/agent-skill-sharing-implementation-checklist.md b/docs/agent-skill-sharing-implementation-checklist.md index 2e2ccc7585a..801c478e8e0 100644 --- a/docs/agent-skill-sharing-implementation-checklist.md +++ b/docs/agent-skill-sharing-implementation-checklist.md @@ -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 diff --git a/docs/agent-skill-sharing-installation-plan.md b/docs/agent-skill-sharing-installation-plan.md index 078366d6d02..e368d72307e 100644 --- a/docs/agent-skill-sharing-installation-plan.md +++ b/docs/agent-skill-sharing-installation-plan.md @@ -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 diff --git a/src/main/skills/skill-bundle-creation.ts b/src/main/skills/skill-bundle-creation.ts index a3e7d85e879..a10391676af 100644 --- a/src/main/skills/skill-bundle-creation.ts +++ b/src/main/skills/skill-bundle-creation.ts @@ -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 { + 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 + } +} diff --git a/src/main/skills/skill-bundle-install-request-service.ts b/src/main/skills/skill-bundle-install-request-service.ts index d40e897886e..183a9a39ae3 100644 --- a/src/main/skills/skill-bundle-install-request-service.ts +++ b/src/main/skills/skill-bundle-install-request-service.ts @@ -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 } @@ -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 { - 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 { + 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 } } diff --git a/src/main/skills/skill-bundle-observability-summary.test.ts b/src/main/skills/skill-bundle-observability-summary.test.ts new file mode 100644 index 00000000000..dd6b0271526 --- /dev/null +++ b/src/main/skills/skill-bundle-observability-summary.test.ts @@ -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 + ) + }) +}) diff --git a/src/main/skills/skill-bundle-observability-summary.ts b/src/main/skills/skill-bundle-observability-summary.ts new file mode 100644 index 00000000000..e9469766171 --- /dev/null +++ b/src/main/skills/skill-bundle-observability-summary.ts @@ -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 + errorCategories: Map +} + +function increment(attributes: Record, key: string): void { + attributes[key] = (attributes[key] ?? 0) + 1 +} + +function recordErrorCategory(categories: Map, 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 = { + skillCount: result.skills.length, + placementCount: 0, + conflictCount: 0 + } + const errorCategories = new Map() + 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 } +} diff --git a/src/main/skills/skill-bundle-ssh-relay-service.ts b/src/main/skills/skill-bundle-ssh-relay-service.ts index 0fcbb713f15..644a58bd0b5 100644 --- a/src/main/skills/skill-bundle-ssh-relay-service.ts +++ b/src/main/skills/skill-bundle-ssh-relay-service.ts @@ -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({ diff --git a/src/main/skills/skill-client-mediated-transfer.ts b/src/main/skills/skill-client-mediated-transfer.ts index 83061c65eaa..536f007a347 100644 --- a/src/main/skills/skill-client-mediated-transfer.ts +++ b/src/main/skills/skill-client-mediated-transfer.ts @@ -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( 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 }> { +} + +async function transferSkillPackageToRuntimeUnobserved( + input: SkillClientTransferInput +): Promise<{ uploadId: string; cleanup(): Promise }> { 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 }> { + 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 + } +} diff --git a/src/main/skills/skill-cloud-grant-installation.ts b/src/main/skills/skill-cloud-grant-installation.ts index f0a886cd897..e7d74218c6e 100644 --- a/src/main/skills/skill-cloud-grant-installation.ts +++ b/src/main/skills/skill-cloud-grant-installation.ts @@ -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 { diff --git a/src/main/skills/skill-cloud-service.ts b/src/main/skills/skill-cloud-service.ts index 7467f032a29..d675b6624e7 100644 --- a/src/main/skills/skill-cloud-service.ts +++ b/src/main/skills/skill-cloud-service.ts @@ -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 }) diff --git a/src/main/skills/skill-install-service.ts b/src/main/skills/skill-install-service.ts index 04c3c32ef34..eacf1cefbf0 100644 --- a/src/main/skills/skill-install-service.ts +++ b/src/main/skills/skill-install-service.ts @@ -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) diff --git a/src/main/skills/skill-install-transaction.ts b/src/main/skills/skill-install-transaction.ts index 11a8426cd53..1410a6d585f 100644 --- a/src/main/skills/skill-install-transaction.ts +++ b/src/main/skills/skill-install-transaction.ts @@ -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) { diff --git a/src/main/skills/skill-operation-observability.test.ts b/src/main/skills/skill-operation-observability.test.ts index f113832d12d..5cd1d5408bd 100644 --- a/src/main/skills/skill-operation-observability.test.ts +++ b/src/main/skills/skill-operation-observability.test.ts @@ -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) + }) }) diff --git a/src/main/skills/skill-operation-observability.ts b/src/main/skills/skill-operation-observability.ts index b2ad7f58348..c0cddfda155 100644 --- a/src/main/skills/skill-operation-observability.ts +++ b/src/main/skills/skill-operation-observability.ts @@ -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) } } } diff --git a/src/main/skills/skill-package-creation.ts b/src/main/skills/skill-package-creation.ts index 1042c04fff5..694313f4812 100644 --- a/src/main/skills/skill-package-creation.ts +++ b/src/main/skills/skill-package-creation.ts @@ -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 { + 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 + } +} diff --git a/src/main/skills/skill-package-download.ts b/src/main/skills/skill-package-download.ts index 3e2dca32ffd..86e124ca7bc 100644 --- a/src/main/skills/skill-package-download.ts +++ b/src/main/skills/skill-package-download.ts @@ -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 { input.fetcher ??= fetch @@ -211,3 +212,21 @@ export async function downloadSkillPackageGrant( throw error } } + +export async function downloadSkillPackageGrant( + input: SkillPackageDownloadInput +): Promise { + 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 + } +} diff --git a/src/main/skills/skill-placement-reconciliation.ts b/src/main/skills/skill-placement-reconciliation.ts index 8ec11d6e653..92be675fefa 100644 --- a/src/main/skills/skill-placement-reconciliation.ts +++ b/src/main/skills/skill-placement-reconciliation.ts @@ -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 { 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 { 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 { 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 } } diff --git a/src/main/skills/skill-remote-install-service.ts b/src/main/skills/skill-remote-install-service.ts index c16bf4a1702..6b719adf3c6 100644 --- a/src/main/skills/skill-remote-install-service.ts +++ b/src/main/skills/skill-remote-install-service.ts @@ -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 { - 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({ diff --git a/src/main/skills/skill-ssh-package-transfer.ts b/src/main/skills/skill-ssh-package-transfer.ts index 3d5ea2b6bd1..8c8603072b2 100644 --- a/src/main/skills/skill-ssh-package-transfer.ts +++ b/src/main/skills/skill-ssh-package-transfer.ts @@ -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 { @@ -132,3 +133,26 @@ export async function transferSkillPackageToSshHost( } } } + +export async function transferSkillPackageToSshHost( + client: SkillSshRelayClient, + input: SkillSshTransferInput +): Promise { + 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 + } +} diff --git a/src/main/skills/skill-ssh-relay-service.ts b/src/main/skills/skill-ssh-relay-service.ts index 3c936739ab1..91360eec830 100644 --- a/src/main/skills/skill-ssh-relay-service.ts +++ b/src/main/skills/skill-ssh-relay-service.ts @@ -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({ diff --git a/src/main/skills/skill-transaction-recovery-observability.ts b/src/main/skills/skill-transaction-recovery-observability.ts new file mode 100644 index 00000000000..ec8d6cf90e4 --- /dev/null +++ b/src/main/skills/skill-transaction-recovery-observability.ts @@ -0,0 +1,18 @@ +import { startSkillPhaseOperation } from './skill-operation-observability' + +export async function settleObservedSkillTransactionRecovery(input: { + rollback: boolean + recover(): Promise +}): Promise { + 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) + } +} diff --git a/src/main/skills/skill-transaction-startup-recovery.ts b/src/main/skills/skill-transaction-startup-recovery.ts index 728de370995..012cecb5e7c 100644 --- a/src/main/skills/skill-transaction-startup-recovery.ts +++ b/src/main/skills/skill-transaction-startup-recovery.ts @@ -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 { 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 { + 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 + } +}