diff --git a/.github/ci/suites.v1.json b/.github/ci/suites.v1.json
index 1dadd748df..60f1f761c8 100644
--- a/.github/ci/suites.v1.json
+++ b/.github/ci/suites.v1.json
@@ -410,7 +410,12 @@
         "tests/test_skill_install_source.py",
         "tests/test_skills_hub_streaming.py",
         "tests/test_skills_hub_streaming_faults.py",
-        "src/opensquilla/application/skill_source.py"
+        "src/opensquilla/application/skill_source.py",
+        "tests/test_engine/test_skill_install_settlement.py",
+        "tests/test_gateway/test_skill_install_status.py",
+        "tests/test_skills/test_hub_install_operations.py",
+        "tests/test_skills/test_staging_io_worker.py",
+        "src/opensquilla/engine/agent.py"
       ]
     },
     "managed-toolchain": {
diff --git a/.github/scripts/plan_ci.py b/.github/scripts/plan_ci.py
index 47e24cb394..764e582a05 100644
--- a/.github/scripts/plan_ci.py
+++ b/.github/scripts/plan_ci.py
@@ -168,6 +168,10 @@
         "tests/test_skills_hub_lockfile_contract.py",
         "tests/test_skills_hub_doctor.py",
         "tests/test_skills_hash_consumers.py",
+        "tests/test_engine/test_skill_install_settlement.py",
+        "tests/test_gateway/test_skill_install_status.py",
+        "tests/test_skills/test_hub_install_operations.py",
+        "tests/test_skills/test_staging_io_worker.py",
         "tests/test_skill_install_source.py",
         "tests/test_skills_hub_streaming.py",
         "tests/test_skills_hub_streaming_faults.py",
@@ -203,6 +207,7 @@
         "src/opensquilla/cli/skills_meta_cmd.py",
         "src/opensquilla/application/skill_catalog.py",
         "src/opensquilla/application/skill_management.py",
+        "src/opensquilla/engine/agent.py",
         "src/opensquilla/application/skill_source.py",
         "src/opensquilla/application/skill_proposal_review.py",
         "src/opensquilla/gateway/app.py",
diff --git a/.github/scripts/windows_test_assignments.json b/.github/scripts/windows_test_assignments.json
index 73f1844ec9..32bbe5ff17 100644
--- a/.github/scripts/windows_test_assignments.json
+++ b/.github/scripts/windows_test_assignments.json
@@ -283,6 +283,7 @@
       "tests/test_skills/test_paper_source_readiness_gate.py",
       "tests/test_skills/test_short_drama_review_normalizer.py",
       "tests/test_skills/test_skill_disable_toggle.py",
+      "tests/test_skills/test_staging_io_worker.py",
       "tests/test_skills/test_subtitle_burner.py",
       "tests/test_skills/test_title_card_image.py",
       "tests/test_skills/test_toolchain_runtime_integration.py",
@@ -721,6 +722,7 @@
       "tests/test_gateway/test_shutdown_order.py",
       "tests/test_gateway/test_skill_catalog_adapter.py",
       "tests/test_gateway/test_skill_catalog_application.py",
+      "tests/test_gateway/test_skill_install_status.py",
       "tests/test_gateway/test_skill_management_adapter.py",
       "tests/test_gateway/test_skill_management_application.py",
       "tests/test_gateway/test_skill_management_service_injection.py",
@@ -1364,6 +1366,7 @@
       "tests/test_engine/test_runtime_error_persistence.py",
       "tests/test_engine/test_runtime_meta_invoke_surfacing.py",
       "tests/test_engine/test_runtime_tool_run_budget.py",
+      "tests/test_engine/test_skill_install_settlement.py",
       "tests/test_engine/test_spawn_depth_unified.py",
       "tests/test_engine/test_stream_repetition_guard.py",
       "tests/test_engine/test_stream_wrappers.py",
@@ -1468,6 +1471,7 @@
       "tests/test_skills/test_clarify_skip_summary.py",
       "tests/test_skills/test_creator_trigger_no_collision.py",
       "tests/test_skills/test_history_explorer.py",
+      "tests/test_skills/test_hub_install_operations.py",
       "tests/test_skills/test_meta_auto_trigger_enabled.py",
       "tests/test_skills/test_meta_on_failure_substitute.py",
       "tests/test_skills/test_meta_orchestrator_offload.py",
diff --git a/.github/scripts/windows_test_durations.json b/.github/scripts/windows_test_durations.json
index 6549e76966..6afa33bd89 100644
--- a/.github/scripts/windows_test_durations.json
+++ b/.github/scripts/windows_test_durations.json
@@ -1591,6 +1591,10 @@
     "tests/test_telemetry_server/test_client_pipeline.py": 0.01,
     "tests/test_scripts/test_gateway_ux.py": 0.01,
     "tests/test_skills_hub_streaming.py": 0.01,
-    "tests/test_skills_hub_streaming_faults.py": 0.01
+    "tests/test_skills_hub_streaming_faults.py": 0.01,
+    "tests/test_engine/test_skill_install_settlement.py": 0.01,
+    "tests/test_gateway/test_skill_install_status.py": 0.01,
+    "tests/test_skills/test_hub_install_operations.py": 0.01,
+    "tests/test_skills/test_staging_io_worker.py": 0.01
   }
 }
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index abc3837ec5..fa5c96beb6 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -2057,6 +2057,10 @@ jobs:
             tests/test_skills_hub_doctor.py \
             tests/test_skills_hash_consumers.py \
             tests/test_skill_install_source.py \
+            tests/test_engine/test_skill_install_settlement.py \
+            tests/test_gateway/test_skill_install_status.py \
+            tests/test_skills/test_hub_install_operations.py \
+            tests/test_skills/test_staging_io_worker.py \
             tests/test_skills_hub_streaming.py \
             tests/test_skills_hub_streaming_faults.py \
             tests/test_skills/test_hub_management_service.py \
diff --git a/contracts/gateway/v4/compatibility-manifest.generated.json b/contracts/gateway/v4/compatibility-manifest.generated.json
index d1a933dcae..6198b9609a 100644
--- a/contracts/gateway/v4/compatibility-manifest.generated.json
+++ b/contracts/gateway/v4/compatibility-manifest.generated.json
@@ -1321,6 +1321,12 @@
       "schema": "skills/skills-install-cancel.schema.json",
       "schemaSha256": "e22bfcec3766dab93bb4db02d25847fceb3b31446c7fe1807b6ea8fdc0cac666"
     },
+    {
+      "lifecycle": "stable",
+      "name": "skills.install.status",
+      "schema": "skills/skills-install-status.schema.json",
+      "schemaSha256": "8ed97d3207777b114f8efb90bbe9750d0964e694b2fcbcdc95f8cd12c658d57a"
+    },
     {
       "lifecycle": "stable",
       "name": "skills.list",
@@ -1457,10 +1463,10 @@
   "protocol": "opensquilla-websocket-json",
   "source": {
     "eventFamilyCount": 10,
-    "generatorSha256": "0e0b513d7844d926e5f2c6065e64e4b523a1f8df2023317327ee6e05bfa29038",
-    "methodCount": 214,
-    "schemaCount": 224,
-    "schemaTreeSha256": "7479e55aa84897a526ea6a8c8b5dca3f995ffb7fdde4dc57bc84123e21cdf8fc"
+    "generatorSha256": "f781c6d8e31b336c2d2e342784935b3d327c9947cb35e56de8e44a0a83bffae6",
+    "methodCount": 215,
+    "schemaCount": 225,
+    "schemaTreeSha256": "182cd4e1f6016ec223a6bdcf9e76cd6e734764b581e35a4f6c9efc21aeb173a6"
   },
   "wireVersion": 4
 }
diff --git a/contracts/gateway/v4/production-targets.json b/contracts/gateway/v4/production-targets.json
index 1988355c95..eae75a17a7 100644
--- a/contracts/gateway/v4/production-targets.json
+++ b/contracts/gateway/v4/production-targets.json
@@ -1254,6 +1254,13 @@
         "result"
       ]
     },
+    {
+      "kind": "method",
+      "wireName": "skills.install.status",
+      "roles": [
+        "result"
+      ]
+    },
     {
       "kind": "method",
       "wireName": "skills.list",
diff --git a/contracts/gateway/v4/skills/skills-install-status.schema.json b/contracts/gateway/v4/skills/skills-install-status.schema.json
new file mode 100644
index 0000000000..9ac4bfacf1
--- /dev/null
+++ b/contracts/gateway/v4/skills/skills-install-status.schema.json
@@ -0,0 +1,247 @@
+{
+  "$schema": "https://json-schema.org/draft/2020-12/schema",
+  "$id": "https://opensquilla.dev/contracts/gateway/v4/skills/skills-install-status.schema.json",
+  "title": "OpenSquilla skills.install.status Contract",
+  "type": "object",
+  "additionalProperties": false,
+  "properties": {
+    "request": {
+      "$ref": "#/$defs/Request"
+    },
+    "response": {
+      "$ref": "#/$defs/Response"
+    }
+  },
+  "x-opensquilla-wire": {
+    "protocol": "opensquilla-websocket-json",
+    "version": 4,
+    "compatibility": "exact-json-tree"
+  },
+  "x-opensquilla-codegen": {
+    "python": {
+      "tool": "datamodel-code-generator",
+      "version": "0.75.1",
+      "target": "pydantic_v2.BaseModel"
+    },
+    "typescript": {
+      "tool": "json-schema-to-typescript",
+      "version": "15.0.4"
+    },
+    "runtimeValidation": {
+      "tool": "ajv",
+      "version": "8.17.1",
+      "mode": "standalone-adapter-only"
+    }
+  },
+  "x-opensquilla-method": {
+    "name": "skills.install.status",
+    "kind": "query",
+    "scope": "operator.admin",
+    "guestAllowed": false,
+    "idempotency": "read-only",
+    "timeout": {
+      "policy": "caller"
+    },
+    "capability": {
+      "kind": "method-availability",
+      "name": "skills.install.status",
+      "optional": true
+    },
+    "request": "#/$defs/Request",
+    "params": "#/$defs/Params",
+    "response": "#/$defs/Response",
+    "result": "#/$defs/Result",
+    "errors": [
+      {
+        "code": "INVALID_REQUEST"
+      },
+      {
+        "code": "UNAUTHORIZED"
+      },
+      {
+        "code": "UNSUPPORTED"
+      },
+      {
+        "code": "INTERNAL_ERROR"
+      }
+    ]
+  },
+  "$defs": {
+    "Params": {
+      "type": "object",
+      "additionalProperties": true,
+      "properties": {
+        "operationId": {
+          "type": "string"
+        },
+        "operation_id": {
+          "type": "string"
+        }
+      },
+      "anyOf": [
+        {
+          "required": [
+            "operationId"
+          ]
+        },
+        {
+          "required": [
+            "operation_id"
+          ]
+        }
+      ]
+    },
+    "Request": {
+      "type": "object",
+      "additionalProperties": false,
+      "required": [
+        "type",
+        "id",
+        "method",
+        "params"
+      ],
+      "properties": {
+        "type": {
+          "const": "req"
+        },
+        "id": {
+          "type": "string"
+        },
+        "method": {
+          "const": "skills.install.status"
+        },
+        "params": {
+          "$ref": "#/$defs/Params"
+        }
+      }
+    },
+    "Result": {
+      "type": "object",
+      "additionalProperties": true,
+      "required": [
+        "operationId",
+        "state",
+        "phase",
+        "terminal",
+        "scope"
+      ],
+      "properties": {
+        "operationId": {
+          "type": "string"
+        },
+        "scope": {
+          "type": "string"
+        },
+        "phase": {
+          "type": "string"
+        },
+        "state": {
+          "enum": [
+            "unknown",
+            "running",
+            "succeeded",
+            "failed",
+            "cancelled",
+            "recovery_required"
+          ]
+        },
+        "terminal": {
+          "type": "boolean"
+        },
+        "progress": {
+          "type": "object",
+          "additionalProperties": true
+        },
+        "result": {
+          "type": "object",
+          "required": [
+            "success"
+          ],
+          "properties": {
+            "success": {
+              "type": "boolean"
+            }
+          },
+          "additionalProperties": true
+        }
+      }
+    },
+    "RpcError": {
+      "type": "object",
+      "additionalProperties": true,
+      "required": [
+        "code",
+        "message"
+      ],
+      "properties": {
+        "code": {
+          "type": "string"
+        },
+        "message": {
+          "type": "string"
+        }
+      }
+    },
+    "Ok": {
+      "type": "object",
+      "additionalProperties": false,
+      "required": [
+        "type",
+        "id",
+        "ok",
+        "payload"
+      ],
+      "properties": {
+        "type": {
+          "const": "res"
+        },
+        "id": {
+          "type": "string"
+        },
+        "ok": {
+          "const": true
+        },
+        "payload": {
+          "$ref": "#/$defs/Result"
+        }
+      }
+    },
+    "Err": {
+      "type": "object",
+      "additionalProperties": false,
+      "required": [
+        "type",
+        "id",
+        "ok",
+        "error"
+      ],
+      "properties": {
+        "type": {
+          "const": "res"
+        },
+        "id": {
+          "type": "string"
+        },
+        "ok": {
+          "const": false
+        },
+        "payload": {
+          "type": "null"
+        },
+        "error": {
+          "$ref": "#/$defs/RpcError"
+        }
+      }
+    },
+    "Response": {
+      "oneOf": [
+        {
+          "$ref": "#/$defs/Ok"
+        },
+        {
+          "$ref": "#/$defs/Err"
+        }
+      ]
+    }
+  }
+}
diff --git a/desktop/electron/scripts/test-profile-consolidation-flow.mjs b/desktop/electron/scripts/test-profile-consolidation-flow.mjs
index c481e818f2..238b72abd8 100644
--- a/desktop/electron/scripts/test-profile-consolidation-flow.mjs
+++ b/desktop/electron/scripts/test-profile-consolidation-flow.mjs
@@ -797,8 +797,8 @@ try {
     invalidCredentialStableCode: skippedCredentialEvent.stableCode,
   }, null, 2))
 } catch (error) {
-  const desktopLog = await readFile(join(userData, 'logs', 'desktop.log'), 'utf8').catch(() => '')
-  const gatewayLog = await readFile(join(userData, 'logs', 'gateway.log'), 'utf8').catch(() => '')
+  const desktopLog = await readFile(join(activeAppUserData, 'logs', 'desktop.log'), 'utf8').catch(() => '')
+  const gatewayLog = await readFile(join(activeAppUserData, 'logs', 'gateway.log'), 'utf8').catch(() => '')
   console.error(JSON.stringify({
     phase: 'run-error-before-cleanup',
     error: String(error?.stack || error),
diff --git a/opensquilla-webui/src/adapters/gateway/skillCatalogV4.ts b/opensquilla-webui/src/adapters/gateway/skillCatalogV4.ts
index 534e47017b..123b5d50f4 100644
--- a/opensquilla-webui/src/adapters/gateway/skillCatalogV4.ts
+++ b/opensquilla-webui/src/adapters/gateway/skillCatalogV4.ts
@@ -46,6 +46,10 @@ import {
   type Result as SkillsInstallCancelResult,
 } from '@/contracts/generated/v4/skillsInstallCancel'
 import { validateResult as validateSkillsInstallCancelResult } from '@/contracts/generated/v4/skillsInstallCancelValidators.mjs'
+import { SKILLS_INSTALL_STATUS_METHOD, type Result as SkillsInstallStatusResult } from '@/contracts/generated/v4/skillsInstallStatus'
+import { validateResult as validateSkillsInstallStatusResult } from '@/contracts/generated/v4/skillsInstallStatusValidators.mjs'
+import type { SkillInstallStatus } from '@/modules/skillCatalog'
+
 import {
   SKILLS_DEPS_INSTALL_METHOD,
   type Params as SkillsDepsInstallParams,
@@ -348,6 +352,16 @@ export function createV4SkillCatalog(rpc: RpcTransport): SkillCatalog {
       if (!validateSkillsInstallResult(result)) throw invalid(SKILLS_INSTALL_METHOD)
       return result as unknown as SkillInstallResult
     },
+    supportsInstallStatus() {
+      return rpc.supports(SKILLS_INSTALL_STATUS_METHOD)
+    },
+    async installStatus(operationId, options) {
+      const result = await rpc.request<SkillsInstallStatusResult>(
+        SKILLS_INSTALL_STATUS_METHOD, { operationId }, callOptions(options?.signal),
+      )
+      if (!validateSkillsInstallStatusResult(result)) throw invalid(SKILLS_INSTALL_STATUS_METHOD)
+      return result as unknown as SkillInstallStatus
+    },
     supportsInstallCancellation() {
       return rpc.supports(SKILLS_INSTALL_CANCEL_METHOD)
     },
diff --git a/opensquilla-webui/src/components/skills/SkillsAddDrawer.vue b/opensquilla-webui/src/components/skills/SkillsAddDrawer.vue
index f4c7b406e9..971aac6bb0 100644
--- a/opensquilla-webui/src/components/skills/SkillsAddDrawer.vue
+++ b/opensquilla-webui/src/components/skills/SkillsAddDrawer.vue
@@ -144,7 +144,7 @@
                   :data-status="item.status"
                 >
                   <span class="sk-add-queue-item__icon" aria-hidden="true">
-                    <span v-if="item.status === 'installing' || item.status === 'cancelling'" class="sk-spinner" aria-hidden="true" />
+                    <span v-if="item.status === 'installing' || item.status === 'waiting' || item.status === 'cancelling'" class="sk-spinner" aria-hidden="true" />
                     <Icon v-else-if="item.status === 'installed' || item.status === 'unchanged'" name="check" :size="18" />
                     <Icon v-else-if="item.status === 'failed' || item.status === 'unknown'" name="info" :size="18" />
                     <Icon v-else name="clock" :size="18" />
@@ -157,6 +157,7 @@
                         : t(`cronSkills.registry.queueStatus.${item.status}`) }}</span>
                     </div>
                     <code :title="item.identifier">{{ item.identifier }}</code>
+                    <p v-if="item.progress && (item.status === 'waiting' || item.status === 'installing')" aria-live="polite">{{ item.progress }}</p>
                     <p v-if="item.error" class="sk-add-queue-item__error">{{ item.error }}</p>
                     <p v-if="item.status === 'deferred'" class="sk-add-queue-item__note">
                       {{ t('cronSkills.registry.rateLimitDeferred') }}
@@ -462,12 +463,12 @@ function settleActivityExpansion(source: SkillInstallSource) {
   if (props.runningSource === source
     || items.some(item => item.status === 'queued'
       || item.status === 'installing'
-      || item.status === 'cancelling')) {
+      || item.status === 'waiting' || item.status === 'cancelling')) {
     activityExpanded.value[source] = true
     return
   }
   activityExpanded.value[source] = items.some(item =>
-    item.status === 'failed' || item.status === 'unknown')
+    item.status === 'failed' || item.status === 'unknown' || item.status === 'selection_required')
 }
 
 const currentActivity = computed(() => props.activities[sourceMode.value])
@@ -483,7 +484,7 @@ function activityPhase(source: SkillInstallSource) {
   if (props.runningSource !== source) return 'terminal'
   return activity.items.some(item => item.status === 'queued'
     || item.status === 'installing'
-    || item.status === 'cancelling')
+    || item.status === 'waiting' || item.status === 'cancelling')
     ? 'installing'
     : 'refreshing'
 }
@@ -509,10 +510,11 @@ const completedCount = computed(() => currentItems.value.filter(item =>
     || item.status === 'unchanged'
     || item.status === 'failed'
     || item.status === 'unknown'
-    || item.status === 'cancelled').length)
+    || item.status === 'cancelled'
+    || item.status === 'selection_required').length)
 const currentIndex = computed(() => {
   const installing = currentItems.value.findIndex(item =>
-    item.status === 'installing' || item.status === 'cancelling')
+    item.status === 'installing' || item.status === 'waiting' || item.status === 'cancelling')
   return installing >= 0
     ? installing + 1
     : Math.min(completedCount.value + 1, currentItems.value.length)
@@ -575,7 +577,7 @@ const installAnnouncement = computed(() => {
 
 function sourceAttentionCount(source: SkillInstallSource): number {
   return props.activities[source].items.filter(item =>
-    item.status === 'failed' || item.status === 'unknown').length
+    item.status === 'failed' || item.status === 'unknown' || item.status === 'selection_required').length
 }
 
 function sourceAttentionLabel(source: SkillInstallSource): string {
diff --git a/opensquilla-webui/src/composables/skills/skillInstallReceipts.test.ts b/opensquilla-webui/src/composables/skills/skillInstallReceipts.test.ts
new file mode 100644
index 0000000000..3ca4cd1cce
--- /dev/null
+++ b/opensquilla-webui/src/composables/skills/skillInstallReceipts.test.ts
@@ -0,0 +1,140 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
+import { effectScope } from 'vue'
+import { createSkillInstallReceipts } from './skillInstallReceipts'
+import { useSkillRegistry } from './useSkillRegistry'
+import type { SkillCatalog } from '@/modules/skillCatalog'
+
+vi.mock('@/composables/useToasts', () => ({ useToasts: () => ({ pushToast: vi.fn() }) }))
+const SCOPE = 'a'.repeat(64)
+const ID = '00000000-0000-4000-8000-000000000001'
+const rows = new Map<string, string>()
+
+beforeEach(() => {
+  rows.clear()
+  vi.useFakeTimers()
+  vi.stubGlobal('sessionStorage', {
+    getItem: (key: string) => rows.get(key) ?? null,
+    setItem: (key: string, value: string) => rows.set(key, value),
+  })
+})
+afterEach(() => { vi.useRealTimers(); vi.unstubAllGlobals() })
+
+function catalog() {
+  return {
+    supportsInstallStatus: () => true,
+    supportsInstallCancellation: () => true,
+    install: vi.fn(async () => { throw new Error('RPC timed out') }),
+    installStatus: vi.fn(async (operationId: string) => ({
+      operationId, scope: SCOPE, state: 'unknown', phase: 'unknown', terminal: true,
+    })),
+  } as unknown as SkillCatalog
+}
+
+describe('durable Skill receipts', () => {
+  it('waits after timeout and refreshes exactly once without another install', async () => {
+    const service = catalog()
+    let reads = 0
+    vi.mocked(service.installStatus!).mockImplementation(async operationId => {
+      if (operationId.startsWith('00000000-0000-0000')) {
+        return { operationId, scope: SCOPE, state: 'unknown', phase: 'unknown', terminal: true }
+      }
+      reads++
+      return reads === 1
+        ? { operationId, scope: SCOPE, state: 'running', phase: 'downloading', terminal: false }
+        : { operationId, scope: SCOPE, state: 'succeeded', phase: 'complete', terminal: true,
+          result: { success: true, installed: true, effectiveFrom: 'next_turn' } }
+    })
+    const load = vi.fn(async () => true)
+    const scope = effectScope()
+    const registry = scope.run(() => useSkillRegistry(service, load))!
+    registry.githubUrl.value = 'https://github.com/acme/demo'
+    const run = registry.installGithub()
+    await vi.advanceTimersByTimeAsync(0)
+    expect(registry.installActivities.value.github.items[0]?.status).toBe('waiting')
+    await vi.advanceTimersByTimeAsync(2000)
+    expect(load).not.toHaveBeenCalled()
+    await vi.advanceTimersByTimeAsync(5000)
+    await run
+    expect(registry.installActivities.value.github.items[0]?.status).toBe('installed')
+    expect(service.install).toHaveBeenCalledTimes(1)
+    expect(load).toHaveBeenCalledTimes(1)
+    scope.stop()
+  })
+
+  it('recovers an outstanding ID after remount without replaying installation', async () => {
+    const service = catalog()
+    const controller = new AbortController()
+    const receipts = createSkillInstallReceipts(service, controller.signal)
+    await receipts.remember({ operationId: ID, source: 'github',
+      identifier: 'https://github.com/acme/demo', displayName: 'demo' })
+    vi.mocked(service.installStatus!).mockImplementation(async operationId => ({
+      operationId, scope: SCOPE, state: 'succeeded', phase: 'complete', terminal: true,
+      result: { success: true, name: 'demo' },
+    }))
+    const load = vi.fn(async () => true)
+    const scope = effectScope()
+    const registry = scope.run(() => useSkillRegistry(service, load))!
+    await vi.advanceTimersByTimeAsync(2000)
+    expect(registry.installActivities.value.github.items[0]?.status).toBe('installed')
+    expect(service.install).not.toHaveBeenCalled()
+    expect(load).toHaveBeenCalledTimes(1)
+    expect(await receipts.pending()).toEqual([])
+    scope.stop()
+  })
+
+  it('does not restore IDs from a different profile or caller', async () => {
+    const service = catalog()
+    const receipts = createSkillInstallReceipts(service, new AbortController().signal)
+    await receipts.remember({ operationId: ID, source: 'github',
+      identifier: 'https://github.com/acme/demo', displayName: 'demo' })
+    vi.mocked(service.installStatus!).mockResolvedValue({
+      operationId: ID, scope: 'b'.repeat(64), state: 'unknown', phase: 'unknown', terminal: true,
+    })
+    expect(await receipts.pending()).toEqual([])
+    expect(service.install).not.toHaveBeenCalled()
+  })
+
+  it('retains every recovered result from the same source after remount', async () => {
+    const service = catalog()
+    const receipts = createSkillInstallReceipts(service, new AbortController().signal)
+    const secondId = '00000000-0000-4000-8000-000000000002'
+    await receipts.remember({ operationId: ID, source: 'github',
+      identifier: 'https://github.com/acme/first', displayName: 'first' })
+    await receipts.remember({ operationId: secondId, source: 'github',
+      identifier: 'https://github.com/acme/second', displayName: 'second' })
+    vi.mocked(service.installStatus!).mockImplementation(async operationId => (
+      operationId === ID
+        ? { operationId, scope: SCOPE, state: 'recovery_required', phase: 'complete', terminal: true,
+          result: { success: false, recoveryRequired: true, message: 'Recovery is required' } }
+        : { operationId, scope: SCOPE, state: 'succeeded', phase: 'complete', terminal: true,
+          result: { success: true, name: 'second' } }
+    ))
+    const scope = effectScope()
+    try {
+      const registry = scope.run(() => useSkillRegistry(service, vi.fn(async () => true)))!
+      await vi.advanceTimersByTimeAsync(4000)
+      expect(registry.installActivities.value.github.items.map(item => item.displayName))
+        .toEqual(['first', 'second'])
+      expect(registry.installActivities.value.github.items.map(item => item.status))
+        .toEqual(['unknown', 'installed'])
+      expect(registry.installActivities.value.github.items[0]?.error).toBe('Recovery is required')
+      expect(service.install).not.toHaveBeenCalled()
+      expect((await receipts.pending()).map(row => row.operationId)).toEqual([ID])
+    } finally {
+      scope.stop()
+    }
+  })
+})
+
+
+it('stops waiting when status access is revoked without replaying installation', async () => {
+  const service = catalog()
+  vi.mocked(service.installStatus!).mockRejectedValue(Object.assign(new Error('access denied'), { code: 'UNAUTHORIZED' }))
+  const receipts = createSkillInstallReceipts(service, new AbortController().signal)
+  const waiting = receipts.wait(ID, vi.fn())
+  const rejected = expect(waiting).rejects.toThrow('access denied')
+  await vi.advanceTimersByTimeAsync(2000)
+  await rejected
+  expect(service.installStatus).toHaveBeenCalledTimes(1)
+  expect(service.install).not.toHaveBeenCalled()
+})
diff --git a/opensquilla-webui/src/composables/skills/skillInstallReceipts.ts b/opensquilla-webui/src/composables/skills/skillInstallReceipts.ts
new file mode 100644
index 0000000000..a814fce669
--- /dev/null
+++ b/opensquilla-webui/src/composables/skills/skillInstallReceipts.ts
@@ -0,0 +1,87 @@
+import type { SkillCatalog, SkillInstallResult, SkillInstallStatus } from '@/modules/skillCatalog'
+
+const STORAGE_KEY = 'opensquilla.skillInstall.pending.v1'
+const UUID = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i
+
+export interface PendingSkillInstall {
+  scope: string
+  operationId: string
+  identifier: string
+  source: string
+  displayName: string
+}
+
+function readPending(): PendingSkillInstall[] {
+  try {
+    const value = JSON.parse(globalThis.sessionStorage?.getItem(STORAGE_KEY) || '[]')
+    if (!Array.isArray(value)) return []
+    return value.slice(-100).filter((row): row is PendingSkillInstall =>
+      row && typeof row.scope === 'string' && /^[a-f0-9]{64}$/.test(row.scope)
+      && typeof row.operationId === 'string' && UUID.test(row.operationId)
+      && typeof row.identifier === 'string' && row.identifier.length <= 4096
+      && typeof row.source === 'string' && row.source.length <= 128
+      && typeof row.displayName === 'string' && row.displayName.length <= 512)
+  } catch { return [] }
+}
+
+function save(rows: PendingSkillInstall[]) {
+  try { globalThis.sessionStorage?.setItem(STORAGE_KEY, JSON.stringify(rows.slice(-100))) } catch { /* Storage may be unavailable. */ }
+}
+
+function pause(delay: number, signal: AbortSignal): Promise<void> {
+  return new Promise((resolve, reject) => {
+    if (signal.aborted) { reject(new Error('Install receipt wait ended')); return }
+    const abort = () => { clearTimeout(timer); reject(new Error('Install receipt wait ended')) }
+    const timer = setTimeout(() => { signal.removeEventListener('abort', abort); resolve() }, delay)
+    signal.addEventListener('abort', abort, { once: true })
+  })
+}
+
+export function createSkillInstallReceipts(catalog: SkillCatalog, signal: AbortSignal) {
+  const supported = () => Boolean(catalog.supportsInstallStatus?.() && catalog.installStatus)
+  const scope = async () => {
+    if (!supported()) return ''
+    return (await catalog.installStatus!('00000000-0000-0000-0000-000000000000', { signal })).scope
+  }
+  const forget = (operationId: string) => save(readPending().filter(row => row.operationId !== operationId))
+  return {
+    supported,
+    forget,
+    async remember(row: Omit<PendingSkillInstall, 'scope'>) {
+      if (!supported()) return
+      const current = await scope()
+      if (!current) return
+      save([...readPending().filter(item => item.operationId !== row.operationId), { ...row, scope: current }])
+    },
+    async pending() {
+      const current = await scope()
+      return current ? readPending().filter(row => row.scope === current) : []
+    },
+    async wait(operationId: string, onStatus: (status: SkillInstallStatus) => void): Promise<SkillInstallResult> {
+      const expected = readPending().find(row => row.operationId === operationId)?.scope
+      let attempt = 0
+      while (!signal.aborted && supported()) {
+        await pause([2000, 5000, 10000][Math.min(attempt++, 2)]!, signal)
+        let status: SkillInstallStatus
+        try {
+          status = await catalog.installStatus!(operationId, { signal })
+        } catch (error) {
+          if (signal.aborted) break
+          const failure = error as { code?: unknown; data?: { code?: unknown } } | null
+          const code = String(failure?.code ?? failure?.data?.code ?? '').toUpperCase()
+          if (['METHOD_NOT_FOUND', 'UNSUPPORTED', 'UNAUTHORIZED', 'FORBIDDEN',
+            'OWNER_REQUIRED', 'PERMISSION_DENIED', 'INVALID_PARAMS'].includes(code)) throw error
+          continue
+        }
+        if (expected && status.scope !== expected) throw new Error('Install receipt belongs to another profile or caller')
+        onStatus(status)
+        if (status.terminal) {
+          if (!status.result) throw new Error('Install result is unknown; installation was not repeated')
+          if (status.state !== 'recovery_required') forget(operationId)
+          return status.result
+        }
+      }
+      throw new Error('Install receipt wait ended; installation was not repeated')
+    },
+  }
+}
diff --git a/opensquilla-webui/src/composables/skills/useSkillRegistry.ts b/opensquilla-webui/src/composables/skills/useSkillRegistry.ts
index c629dbcb37..4ceb66d157 100644
--- a/opensquilla-webui/src/composables/skills/useSkillRegistry.ts
+++ b/opensquilla-webui/src/composables/skills/useSkillRegistry.ts
@@ -1,4 +1,5 @@
-import { computed, ref, type ComputedRef, type Ref } from 'vue'
+import { computed, ref, watch, getCurrentScope, onScopeDispose, type ComputedRef, type Ref } from 'vue'
+import { createSkillInstallReceipts } from './skillInstallReceipts'
 import i18n from '@/i18n'
 import type { SkillCatalog, SkillInstallResult } from '@/modules/skillCatalog'
 import { useToasts } from '@/composables/useToasts'
@@ -17,6 +18,7 @@ export type InstallResult = SkillInstallResult
 export type SkillInstallQueueStatus =
   | 'queued'
   | 'installing'
+  | 'waiting'
   | 'cancelling'
   | 'cancelled'
   | 'installed'
@@ -34,6 +36,7 @@ export interface SkillInstallQueueItem {
   status: SkillInstallQueueStatus
   result?: InstallResult
   error?: string
+  progress?: string
 }
 
 export function skillInstallRequiresRiskAcknowledgement(
@@ -246,6 +249,9 @@ export function useSkillRegistry(
 ): SkillRegistry {
   const { pushToast } = useToasts()
   const t = i18n.global.t
+  const receiptAbort = new AbortController()
+  if (getCurrentScope()) onScopeDispose(() => receiptAbort.abort())
+  const receipts = createSkillInstallReceipts(catalog, receiptAbort.signal)
   const registryQuery = ref('')
   const githubUrl = ref('')
   const registryResults = ref<RegistryResult[]>([])
@@ -274,6 +280,55 @@ export function useSkillRegistry(
   let searchRequestId = 0
   let cancelRequestedOperationId = ''
 
+  function showInstallProgress(item: SkillInstallQueueItem, phase: string) {
+    item.status = 'waiting'
+    if (['resolving', 'downloading', 'scanning', 'committing'].includes(phase)) {
+      item.progress = t(`cronSkills.registry.installPhase.${phase}`)
+    }
+  }
+
+  async function restoreInstallReceipts() {
+    if (!receipts.supported() || runningSource.value) return
+    let pending
+    try { pending = await receipts.pending() } catch { return }
+    if (!pending.length || !mutationGate.acquire('install_queue')) return
+    try {
+      for (const row of pending) {
+        const source = activitySource(row.source)
+        const item = requestToQueueItem(row)
+        item.status = 'waiting'
+        const activity = installActivities.value[source]
+        activity.items.push(item)
+        activity.phase = 'installing'
+        const active = activity.items[activity.items.length - 1]!
+        runningSource.value = source
+        activeInstallOperation.value = { id: row.operationId, itemId: active.id, source }
+        try {
+          const result = await receipts.wait(row.operationId, status => showInstallProgress(active, status.phase))
+          active.result = result
+          active.status = result.recoveryRequired ? 'unknown' : result.cancelled ? 'cancelled' : result.success
+            ? (result.unchanged ? 'unchanged' : 'installed')
+            : skillInstallCandidates(result).length ? 'selection_required' : 'failed'
+          active.error = result.success || result.cancelled ? '' : result.message || ''
+          await refreshCatalogAfterBatch(source, [active])
+        } catch (error) {
+          active.status = 'unknown'
+          active.error = (error as Error).message
+        }
+        installActivities.value[source].phase = 'terminal'
+      }
+    } finally {
+      runningSource.value = null
+      activeInstallOperation.value = null
+      cancellingSource.value = null
+      mutationGate.release('install_queue')
+    }
+  }
+
+  watch(() => receipts.supported(), supported => {
+    if (supported) void restoreInstallReceipts()
+  }, { immediate: true })
+
   async function searchRegistry() {
     const query = registryQuery.value.trim()
     const requestId = ++searchRequestId
@@ -379,7 +434,7 @@ export function useSkillRegistry(
       item.result = undefined
       installingId.value = item.id
       let rateLimited = false
-      const operationId = installCancellationSupported.value
+      const operationId = installCancellationSupported.value || receipts.supported()
         ? createInstallOperationId()
         : ''
       if (operationId) {
@@ -390,6 +445,10 @@ export function useSkillRegistry(
         }
       }
       try {
+        if (operationId && receipts.supported()) {
+          await receipts.remember({ operationId, identifier: item.identifier,
+            source: item.source, displayName: item.displayName })
+        }
         const res = await catalog.install({
           identifier: item.identifier,
           source: item.source,
@@ -397,10 +456,15 @@ export function useSkillRegistry(
           ...(riskConfirmation
             ? { riskConfirmation }
             : {}),
+        }).catch(async error => {
+          if (!operationId || !receipts.supported()) throw error
+          item.status = 'waiting'
+          return receipts.wait(operationId, status => showInstallProgress(item, status.phase))
         })
+        if (operationId && !res.recoveryRequired) receipts.forget(operationId)
         item.result = res
         item.displayName = res.name || item.displayName
-        item.status = res.cancelled
+        item.status = res.recoveryRequired ? 'unknown' : res.cancelled
           ? 'cancelled'
           : res.success
             ? (res.unchanged ? 'unchanged' : 'installed')
@@ -437,7 +501,7 @@ export function useSkillRegistry(
     for (const item of installActivities.value[source].items) {
       if (item.status !== 'queued'
         && item.status !== 'installing'
-        && item.status !== 'cancelling') continue
+        && item.status !== 'cancelling' && item.status !== 'waiting') continue
       item.status = 'unknown'
       item.error ||= t('cronSkills.registry.installResultUnknown')
     }
diff --git a/opensquilla-webui/src/contracts/generated/v4/skillsInstallStatus.ts b/opensquilla-webui/src/contracts/generated/v4/skillsInstallStatus.ts
new file mode 100644
index 0000000000..35a2abe256
--- /dev/null
+++ b/opensquilla-webui/src/contracts/generated/v4/skillsInstallStatus.ts
@@ -0,0 +1,95 @@
+// @generated by scripts/contracts/generate_gateway_contracts.py; do not edit.
+// source-sha256: c436f674f065deef98d0ef54d8a5ca738fa1a010c56281ecb1f780745bf9290b
+
+/**
+ * This interface was referenced by `OpenSquillaSkillsInstallStatusContract`'s JSON-Schema
+ * via the `definition` "Params".
+ */
+export type Params =
+  | {
+      operationId: string;
+      operation_id?: string;
+      [k: string]: unknown;
+    }
+  | {
+      operationId?: string;
+      operation_id: string;
+      [k: string]: unknown;
+    };
+/**
+ * This interface was referenced by `OpenSquillaSkillsInstallStatusContract`'s JSON-Schema
+ * via the `definition` "Response".
+ */
+export type Response = Ok | Err;
+
+export interface OpenSquillaSkillsInstallStatusContract {
+  request?: Request;
+  response?: Response;
+}
+/**
+ * This interface was referenced by `OpenSquillaSkillsInstallStatusContract`'s JSON-Schema
+ * via the `definition` "Request".
+ */
+export interface Request {
+  type: "req";
+  id: string;
+  method: "skills.install.status";
+  params: Params;
+}
+/**
+ * This interface was referenced by `OpenSquillaSkillsInstallStatusContract`'s JSON-Schema
+ * via the `definition` "Ok".
+ */
+export interface Ok {
+  type: "res";
+  id: string;
+  ok: true;
+  payload: Result;
+}
+/**
+ * This interface was referenced by `OpenSquillaSkillsInstallStatusContract`'s JSON-Schema
+ * via the `definition` "Result".
+ */
+export interface Result {
+  operationId: string;
+  scope: string;
+  phase: string;
+  state: "unknown" | "running" | "succeeded" | "failed" | "cancelled" | "recovery_required";
+  terminal: boolean;
+  progress?: {
+    [k: string]: unknown;
+  };
+  result?: {
+    success: boolean;
+    [k: string]: unknown;
+  };
+  [k: string]: unknown;
+}
+/**
+ * This interface was referenced by `OpenSquillaSkillsInstallStatusContract`'s JSON-Schema
+ * via the `definition` "Err".
+ */
+export interface Err {
+  type: "res";
+  id: string;
+  ok: false;
+  payload?: null;
+  error: RpcError;
+}
+/**
+ * This interface was referenced by `OpenSquillaSkillsInstallStatusContract`'s JSON-Schema
+ * via the `definition` "RpcError".
+ */
+export interface RpcError {
+  code: string;
+  message: string;
+  [k: string]: unknown;
+}
+
+export const SKILLS_INSTALL_STATUS_CONTRACT_KIND = "query" as const
+export const SKILLS_INSTALL_STATUS_METHOD = "skills.install.status" as const
+export const SKILLS_INSTALL_STATUS_SCOPE = "operator.admin" as const
+export const SKILLS_INSTALL_STATUS_IDEMPOTENCY = "read-only" as const
+export const SKILLS_INSTALL_STATUS_TIMEOUT = {"policy":"caller"} as const
+export const SKILLS_INSTALL_STATUS_CAPABILITY = {"kind":"method-availability","name":"skills.install.status","optional":true} as const
+export const SKILLS_INSTALL_STATUS_ERRORS = [{"code":"INVALID_REQUEST"},{"code":"UNAUTHORIZED"},{"code":"UNSUPPORTED"},{"code":"INTERNAL_ERROR"}] as const
diff --git a/opensquilla-webui/src/contracts/generated/v4/skillsInstallStatusValidators.d.mts b/opensquilla-webui/src/contracts/generated/v4/skillsInstallStatusValidators.d.mts
new file mode 100644
index 0000000000..a74839688f
--- /dev/null
+++ b/opensquilla-webui/src/contracts/generated/v4/skillsInstallStatusValidators.d.mts
@@ -0,0 +1,9 @@
+// @generated by scripts/contracts/generate_gateway_contracts.py; do not edit.
+// source-sha256: c436f674f065deef98d0ef54d8a5ca738fa1a010c56281ecb1f780745bf9290b
+
+export interface ContractValidator {
+  (value: unknown): boolean
+  errors?: readonly unknown[] | null
+}
+
+export const validateResult: ContractValidator
diff --git a/opensquilla-webui/src/contracts/generated/v4/skillsInstallStatusValidators.mjs b/opensquilla-webui/src/contracts/generated/v4/skillsInstallStatusValidators.mjs
new file mode 100644
index 0000000000..a29bd27215
--- /dev/null
+++ b/opensquilla-webui/src/contracts/generated/v4/skillsInstallStatusValidators.mjs
@@ -0,0 +1,4 @@
+// @generated by scripts/contracts/generate_gateway_contracts.py; do not edit.
+// source-sha256: c436f674f065deef98d0ef54d8a5ca738fa1a010c56281ecb1f780745bf9290b
+
+"use strict";export const validateResult = validate20;const schema31 = {"$id":"urn:opensquilla:contract:v4:Result","$schema":"https://json-schema.org/draft/2020-12/schema","$ref":"https://opensquilla.dev/contracts/gateway/v4/skills/skills-install-status.schema.json#/$defs/Result"};const schema37 = {"type":"object","additionalProperties":true,"required":["operationId","state","phase","terminal","scope"],"properties":{"operationId":{"type":"string"},"scope":{"type":"string"},"phase":{"type":"string"},"state":{"enum":["unknown","running","succeeded","failed","cancelled","recovery_required"]},"terminal":{"type":"boolean"},"progress":{"type":"object","additionalProperties":true},"result":{"type":"object","required":["success"],"properties":{"success":{"type":"boolean"}},"additionalProperties":true}}};function validate20(data, {instancePath="", parentData, parentDataProperty, rootData=data, dynamicAnchors={}}={}){/*# sourceURL="urn:opensquilla:contract:v4:Result" */;let vErrors = null;let errors = 0;const evaluated0 = validate20.evaluated;if(evaluated0.dynamicProps){evaluated0.props = undefined;}if(evaluated0.dynamicItems){evaluated0.items = undefined;}if(data && typeof data == "object" && !Array.isArray(data)){if(data.operationId === undefined){const err0 = {instancePath,schemaPath:"https://opensquilla.dev/contracts/gateway/v4/skills/skills-install-status.schema.json#/$defs/Result/required",keyword:"required",params:{missingProperty: "operationId"},message:"must have required property '"+"operationId"+"'"};if(vErrors === null){vErrors = [err0];}else {vErrors.push(err0);}errors++;}if(data.state === undefined){const err1 = {instancePath,schemaPath:"https://opensquilla.dev/contracts/gateway/v4/skills/skills-install-status.schema.json#/$defs/Result/required",keyword:"required",params:{missingProperty: "state"},message:"must have required property '"+"state"+"'"};if(vErrors === null){vErrors = [err1];}else {vErrors.push(err1);}errors++;}if(data.phase === undefined){const err2 = {instancePath,schemaPath:"https://opensquilla.dev/contracts/gateway/v4/skills/skills-install-status.schema.json#/$defs/Result/required",keyword:"required",params:{missingProperty: "phase"},message:"must have required property '"+"phase"+"'"};if(vErrors === null){vErrors = [err2];}else {vErrors.push(err2);}errors++;}if(data.terminal === undefined){const err3 = {instancePath,schemaPath:"https://opensquilla.dev/contracts/gateway/v4/skills/skills-install-status.schema.json#/$defs/Result/required",keyword:"required",params:{missingProperty: "terminal"},message:"must have required property '"+"terminal"+"'"};if(vErrors === null){vErrors = [err3];}else {vErrors.push(err3);}errors++;}if(data.scope === undefined){const err4 = {instancePath,schemaPath:"https://opensquilla.dev/contracts/gateway/v4/skills/skills-install-status.schema.json#/$defs/Result/required",keyword:"required",params:{missingProperty: "scope"},message:"must have required property '"+"scope"+"'"};if(vErrors === null){vErrors = [err4];}else {vErrors.push(err4);}errors++;}if(data.operationId !== undefined){if(typeof data.operationId !== "string"){const err5 = {instancePath:instancePath+"/operationId",schemaPath:"https://opensquilla.dev/contracts/gateway/v4/skills/skills-install-status.schema.json#/$defs/Result/properties/operationId/type",keyword:"type",params:{type: "string"},message:"must be string"};if(vErrors === null){vErrors = [err5];}else {vErrors.push(err5);}errors++;}}if(data.scope !== undefined){if(typeof data.scope !== "string"){const err6 = {instancePath:instancePath+"/scope",schemaPath:"https://opensquilla.dev/contracts/gateway/v4/skills/skills-install-status.schema.json#/$defs/Result/properties/scope/type",keyword:"type",params:{type: "string"},message:"must be string"};if(vErrors === null){vErrors = [err6];}else {vErrors.push(err6);}errors++;}}if(data.phase !== undefined){if(typeof data.phase !== "string"){const err7 = {instancePath:instancePath+"/phase",schemaPath:"https://opensquilla.dev/contracts/gateway/v4/skills/skills-install-status.schema.json#/$defs/Result/properties/phase/type",keyword:"type",params:{type: "string"},message:"must be string"};if(vErrors === null){vErrors = [err7];}else {vErrors.push(err7);}errors++;}}if(data.state !== undefined){let data3 = data.state;if(!((((((data3 === "unknown") || (data3 === "running")) || (data3 === "succeeded")) || (data3 === "failed")) || (data3 === "cancelled")) || (data3 === "recovery_required"))){const err8 = {instancePath:instancePath+"/state",schemaPath:"https://opensquilla.dev/contracts/gateway/v4/skills/skills-install-status.schema.json#/$defs/Result/properties/state/enum",keyword:"enum",params:{allowedValues: schema37.properties.state.enum},message:"must be equal to one of the allowed values"};if(vErrors === null){vErrors = [err8];}else {vErrors.push(err8);}errors++;}}if(data.terminal !== undefined){if(typeof data.terminal !== "boolean"){const err9 = {instancePath:instancePath+"/terminal",schemaPath:"https://opensquilla.dev/contracts/gateway/v4/skills/skills-install-status.schema.json#/$defs/Result/properties/terminal/type",keyword:"type",params:{type: "boolean"},message:"must be boolean"};if(vErrors === null){vErrors = [err9];}else {vErrors.push(err9);}errors++;}}if(data.progress !== undefined){let data5 = data.progress;if(data5 && typeof data5 == "object" && !Array.isArray(data5)){}else {const err10 = {instancePath:instancePath+"/progress",schemaPath:"https://opensquilla.dev/contracts/gateway/v4/skills/skills-install-status.schema.json#/$defs/Result/properties/progress/type",keyword:"type",params:{type: "object"},message:"must be object"};if(vErrors === null){vErrors = [err10];}else {vErrors.push(err10);}errors++;}}if(data.result !== undefined){let data6 = data.result;if(data6 && typeof data6 == "object" && !Array.isArray(data6)){if(data6.success === undefined){const err11 = {instancePath:instancePath+"/result",schemaPath:"https://opensquilla.dev/contracts/gateway/v4/skills/skills-install-status.schema.json#/$defs/Result/properties/result/required",keyword:"required",params:{missingProperty: "success"},message:"must have required property '"+"success"+"'"};if(vErrors === null){vErrors = [err11];}else {vErrors.push(err11);}errors++;}if(data6.success !== undefined){if(typeof data6.success !== "boolean"){const err12 = {instancePath:instancePath+"/result/success",schemaPath:"https://opensquilla.dev/contracts/gateway/v4/skills/skills-install-status.schema.json#/$defs/Result/properties/result/properties/success/type",keyword:"type",params:{type: "boolean"},message:"must be boolean"};if(vErrors === null){vErrors = [err12];}else {vErrors.push(err12);}errors++;}}}else {const err13 = {instancePath:instancePath+"/result",schemaPath:"https://opensquilla.dev/contracts/gateway/v4/skills/skills-install-status.schema.json#/$defs/Result/properties/result/type",keyword:"type",params:{type: "object"},message:"must be object"};if(vErrors === null){vErrors = [err13];}else {vErrors.push(err13);}errors++;}}}else {const err14 = {instancePath,schemaPath:"https://opensquilla.dev/contracts/gateway/v4/skills/skills-install-status.schema.json#/$defs/Result/type",keyword:"type",params:{type: "object"},message:"must be object"};if(vErrors === null){vErrors = [err14];}else {vErrors.push(err14);}errors++;}validate20.errors = vErrors;return errors === 0;}validate20.evaluated = {"props":true,"dynamicProps":false,"dynamicItems":false};
diff --git a/opensquilla-webui/src/locales/de.json b/opensquilla-webui/src/locales/de.json
index f0d33f5e8e..321d7c7800 100644
--- a/opensquilla-webui/src/locales/de.json
+++ b/opensquilla-webui/src/locales/de.json
@@ -3060,7 +3060,14 @@
       "expandActivity": "Installationsaktivität erweitern",
       "collapseActivity": "Installationsaktivität einklappen",
       "selectSkillDirectory": "Skill-Verzeichnis auswählen",
+      "installPhase": {
+        "resolving": "Skill-Quelle ermitteln",
+        "downloading": "Skill-Dateien herunterladen",
+        "scanning": "Skill-Inhalt prüfen",
+        "committing": "Installation speichern"
+      },
       "queueStatus": {
+        "waiting": "Installationsergebnis abwarten",
         "selection_required": "Verzeichnis auswählen",
         "queued": "Wartend",
         "installing": "Wird installiert",
diff --git a/opensquilla-webui/src/locales/en.json b/opensquilla-webui/src/locales/en.json
index 974c3b1b5a..041fd782fd 100644
--- a/opensquilla-webui/src/locales/en.json
+++ b/opensquilla-webui/src/locales/en.json
@@ -3143,7 +3143,14 @@
       "expandActivity": "Expand install activity",
       "collapseActivity": "Collapse install activity",
       "selectSkillDirectory": "Select a Skill directory",
+      "installPhase": {
+        "resolving": "Resolving Skill source",
+        "downloading": "Downloading Skill files",
+        "scanning": "Scanning Skill content",
+        "committing": "Saving installation"
+      },
       "queueStatus": {
+        "waiting": "Waiting for installation result",
         "selection_required": "Choose directory",
         "queued": "Queued",
         "installing": "Installing",
diff --git a/opensquilla-webui/src/locales/es.json b/opensquilla-webui/src/locales/es.json
index 3e4b41bbe5..1dcaf66c40 100644
--- a/opensquilla-webui/src/locales/es.json
+++ b/opensquilla-webui/src/locales/es.json
@@ -3060,7 +3060,14 @@
       "expandActivity": "Expandir actividad de instalación",
       "collapseActivity": "Contraer actividad de instalación",
       "selectSkillDirectory": "Selecciona un directorio de habilidades",
+      "installPhase": {
+        "resolving": "Resolviendo la fuente",
+        "downloading": "Descargando archivos",
+        "scanning": "Analizando el contenido",
+        "committing": "Guardando la instalación"
+      },
       "queueStatus": {
+        "waiting": "Esperando el resultado de instalación",
         "selection_required": "Elegir directorio",
         "queued": "En cola",
         "installing": "Instalando",
diff --git a/opensquilla-webui/src/locales/fr.json b/opensquilla-webui/src/locales/fr.json
index 290fa24839..2aaa545148 100644
--- a/opensquilla-webui/src/locales/fr.json
+++ b/opensquilla-webui/src/locales/fr.json
@@ -3060,7 +3060,14 @@
       "expandActivity": "Développer l’activité d’installation",
       "collapseActivity": "Réduire l’activité d’installation",
       "selectSkillDirectory": "Sélectionnez un dossier de compétence",
+      "installPhase": {
+        "resolving": "Résolution de la source",
+        "downloading": "Téléchargement des fichiers",
+        "scanning": "Analyse du contenu",
+        "committing": "Enregistrement de l’installation"
+      },
       "queueStatus": {
+        "waiting": "En attente du résultat de l’installation",
         "selection_required": "Choisir un dossier",
         "queued": "En attente",
         "installing": "Installation",
diff --git a/opensquilla-webui/src/locales/ja.json b/opensquilla-webui/src/locales/ja.json
index db720b872d..1fd45c136b 100644
--- a/opensquilla-webui/src/locales/ja.json
+++ b/opensquilla-webui/src/locales/ja.json
@@ -3060,7 +3060,14 @@
       "expandActivity": "インストール状況を展開",
       "collapseActivity": "インストール状況を折りたたむ",
       "selectSkillDirectory": "スキルのディレクトリを選択",
+      "installPhase": {
+        "resolving": "スキルの取得元を確認中",
+        "downloading": "スキルファイルをダウンロード中",
+        "scanning": "スキルの内容を確認中",
+        "committing": "インストール結果を保存中"
+      },
       "queueStatus": {
+        "waiting": "インストール結果を待っています",
         "selection_required": "ディレクトリの選択待ち",
         "queued": "待機中",
         "installing": "インストール中",
diff --git a/opensquilla-webui/src/locales/zh-Hans.json b/opensquilla-webui/src/locales/zh-Hans.json
index abf7aeeb05..235cf1bae7 100644
--- a/opensquilla-webui/src/locales/zh-Hans.json
+++ b/opensquilla-webui/src/locales/zh-Hans.json
@@ -3142,7 +3142,14 @@
       "expandActivity": "展开安装活动",
       "collapseActivity": "收起安装活动",
       "selectSkillDirectory": "请选择技能目录",
+      "installPhase": {
+        "resolving": "解析技能来源",
+        "downloading": "下载技能文件",
+        "scanning": "扫描技能内容",
+        "committing": "保存安装结果"
+      },
       "queueStatus": {
+        "waiting": "等待安装结果",
         "selection_required": "待选择目录",
         "queued": "等待中",
         "installing": "安装中",
diff --git a/opensquilla-webui/src/modules/skillCatalog.ts b/opensquilla-webui/src/modules/skillCatalog.ts
index 386954a37b..260356abd2 100644
--- a/opensquilla-webui/src/modules/skillCatalog.ts
+++ b/opensquilla-webui/src/modules/skillCatalog.ts
@@ -13,6 +13,7 @@ import type {
 export interface SkillInstallResult {
   readonly success: boolean
   readonly cancelled?: boolean
+  readonly recoveryRequired?: boolean
   readonly unchanged?: boolean
   readonly name?: string
   readonly message?: string
@@ -80,7 +81,19 @@ export interface ProposalSettingsUpdate {
   readonly auto_enable_max_risk?: 'low' | 'medium' | 'high'
 }
 
+export interface SkillInstallStatus {
+  readonly operationId: string
+  readonly scope: string
+  readonly state: 'unknown' | 'running' | 'succeeded' | 'failed' | 'cancelled' | 'recovery_required'
+  readonly phase: string
+  readonly terminal: boolean
+  readonly progress?: Readonly<Record<string, unknown>>
+  readonly result?: SkillInstallResult
+}
+
 export interface SkillCatalog {
+  supportsInstallStatus?(): boolean
+  installStatus?(operationId: string, options?: { readonly signal?: AbortSignal }): Promise<SkillInstallStatus>
   list(options?: { readonly signal?: AbortSignal }): Promise<readonly Skill[]>
   detail(skill: Pick<Skill, 'name' | 'kind' | 'instance_id' | 'install_id' | 'active' | 'lifecycle'>, options?: {
     readonly signal?: AbortSignal
diff --git a/scripts/contracts/tests/production_targets.test.mjs b/scripts/contracts/tests/production_targets.test.mjs
index 2a5f097630..c021c29f42 100644
--- a/scripts/contracts/tests/production_targets.test.mjs
+++ b/scripts/contracts/tests/production_targets.test.mjs
@@ -24,7 +24,8 @@ test('production cannot import or re-export the verification compiler', () => {
 test('production references exactly match the reviewed target policy', () => {
   const result = evaluateProductionTargets()
   assert.deepEqual(result.failures, [])
-  assert.equal(result.targets.length, 216)
+  assert.equal(result.targets.length, 217)
+  assert.ok(result.targets.includes('method:skills.install.status:result'))
   assert.deepEqual(result.targets.filter(target => (
     target.startsWith('method:telemetry.product_active.record:')
   )), ['method:telemetry.product_active.record:result'])
diff --git a/scripts/contracts/verify_gateway_validator_profiles.mjs b/scripts/contracts/verify_gateway_validator_profiles.mjs
index d75e507736..0d0393b4b1 100644
--- a/scripts/contracts/verify_gateway_validator_profiles.mjs
+++ b/scripts/contracts/verify_gateway_validator_profiles.mjs
@@ -107,8 +107,8 @@ export async function verifyProfiles({ baselineRoot, verificationRoot } = {}) {
       result.roles++
     }
   }
-  assert.equal(result.roles, 866)
-  assert.equal(result.comparedRoles, baselineRoot ? 841 : selected.size)
+  assert.equal(result.roles, 870)
+  assert.equal(result.comparedRoles, baselineRoot ? 869 : selected.size)
   assert.deepEqual(result.rolesWithoutPositiveSeed, [], 'each role requires a positive seed')
   if (baselineRoot) assert.deepEqual(result.supplementalRoles, [
     'method:sessions.list:params',
diff --git a/src/opensquilla/contracts/generated/v4/gateway_contract_registry.py b/src/opensquilla/contracts/generated/v4/gateway_contract_registry.py
index c5bbce5586..384d839359 100644
--- a/src/opensquilla/contracts/generated/v4/gateway_contract_registry.py
+++ b/src/opensquilla/contracts/generated/v4/gateway_contract_registry.py
@@ -1,6 +1,6 @@
 # @generated by scripts/contracts/generate_gateway_contracts.py; do not edit.
-# sources-sha256: 40f348cb685bc2449ebd75cad5eb3aacde0e0e3e8498939001c0eeed10dccc1b
-# generator-sha256: 0e0b513d7844d926e5f2c6065e64e4b523a1f8df2023317327ee6e05bfa29038
+# sources-sha256: 1cededaac1f0a13abe3645e0c1dd33166102286170afc04fc2f4d9ad8494e80c
+# generator-sha256: f781c6d8e31b336c2d2e342784935b3d327c9947cb35e56de8e44a0a83bffae6
 # ruff: noqa
 
 from dataclasses import dataclass
@@ -787,6 +787,10 @@
 from opensquilla.contracts.generated.v4.skills_install_cancel import Request as _skills_install_cancel_request_model
 from opensquilla.contracts.generated.v4.skills_install_cancel import Response as _skills_install_cancel_response_model
 from opensquilla.contracts.generated.v4.skills_install_cancel import Result as _skills_install_cancel_result_model
+from opensquilla.contracts.generated.v4.skills_install_status import Params as _skills_install_status_params_model
+from opensquilla.contracts.generated.v4.skills_install_status import Request as _skills_install_status_request_model
+from opensquilla.contracts.generated.v4.skills_install_status import Response as _skills_install_status_response_model
+from opensquilla.contracts.generated.v4.skills_install_status import Result as _skills_install_status_result_model
 from opensquilla.contracts.generated.v4.skills_list import Params as _skills_list_params_model
 from opensquilla.contracts.generated.v4.skills_list import Request as _skills_list_request_model
 from opensquilla.contracts.generated.v4.skills_list import Response as _skills_list_response_model
@@ -3736,6 +3740,22 @@ class GatewayEventContract:
         response_model=_skills_install_cancel_response_model,
         result_model=_skills_install_cancel_result_model,
     ),
+    'skills.install.status': GatewayMethodContract(
+        name='skills.install.status',
+        kind='query',
+        scope='operator.admin',
+        guest_allowed=False,
+        idempotency='read-only',
+        timeout={'policy': 'caller'},
+        capability={'kind': 'method-availability', 'name': 'skills.install.status', 'optional': True},
+        errors=({'code': 'INVALID_REQUEST'}, {'code': 'UNAUTHORIZED'}, {'code': 'UNSUPPORTED'}, {'code': 'INTERNAL_ERROR'}),
+        protocol='opensquilla-websocket-json',
+        wire_version=4,
+        request_model=_skills_install_status_request_model,
+        params_model=_skills_install_status_params_model,
+        response_model=_skills_install_status_response_model,
+        result_model=_skills_install_status_result_model,
+    ),
     'skills.install': GatewayMethodContract(
         name='skills.install',
         kind='command',
diff --git a/src/opensquilla/contracts/generated/v4/skills_install_status.py b/src/opensquilla/contracts/generated/v4/skills_install_status.py
new file mode 100644
index 0000000000..9ebf5e7d32
--- /dev/null
+++ b/src/opensquilla/contracts/generated/v4/skills_install_status.py
@@ -0,0 +1,113 @@
+# @generated by scripts/contracts/generate_gateway_contracts.py; do not edit.
+# source-sha256: c436f674f065deef98d0ef54d8a5ca738fa1a010c56281ecb1f780745bf9290b
+# ruff: noqa
+
+# generated by datamodel-codegen:
+#   filename:  skills-install-status.schema.json
+
+from __future__ import annotations
+
+from enum import Enum
+from typing import Any, Literal
+
+from pydantic import BaseModel, ConfigDict, Field, RootModel, StrictBool, StrictStr
+
+
+class Params1(BaseModel):
+    model_config = ConfigDict(
+        extra='allow',
+    )
+    operationId: StrictStr
+    operation_id: StrictStr | None = None
+
+
+class Params2(BaseModel):
+    model_config = ConfigDict(
+        extra='allow',
+    )
+    operationId: StrictStr | None = None
+    operation_id: StrictStr
+
+
+class Params(RootModel[Params1 | Params2]):
+    root: Params1 | Params2
+
+
+class Request(BaseModel):
+    model_config = ConfigDict(
+        extra='forbid',
+    )
+    type: Literal['req']
+    id: StrictStr
+    method: Literal['skills.install.status']
+    params: Params
+
+
+class State(Enum):
+    unknown = 'unknown'
+    running = 'running'
+    succeeded = 'succeeded'
+    failed = 'failed'
+    cancelled = 'cancelled'
+    recovery_required = 'recovery_required'
+
+
+class Result1(BaseModel):
+    model_config = ConfigDict(
+        extra='allow',
+    )
+    success: StrictBool
+
+
+class Result(BaseModel):
+    model_config = ConfigDict(
+        extra='allow',
+    )
+    operationId: StrictStr
+    scope: StrictStr
+    phase: StrictStr
+    state: State
+    terminal: StrictBool
+    progress: dict[str, Any] = None  # type: ignore[assignment]
+    result: Result1 = None  # type: ignore[assignment]
+
+
+class RpcError(BaseModel):
+    model_config = ConfigDict(
+        extra='allow',
+    )
+    code: StrictStr
+    message: StrictStr
+
+
+class Ok(BaseModel):
+    model_config = ConfigDict(
+        extra='forbid',
+    )
+    type: Literal['res']
+    id: StrictStr
+    ok: Literal[True]
+    payload: Result
+
+
+class Err(BaseModel):
+    model_config = ConfigDict(
+        extra='forbid',
+    )
+    type: Literal['res']
+    id: StrictStr
+    ok: Literal[False]
+    payload: None = Field(None)
+    error: RpcError
+
+
+class Response(RootModel[Ok | Err]):
+    root: Ok | Err
+
+
+class OpensquillaSkillsInstallStatusContract(BaseModel):
+    model_config = ConfigDict(
+        extra='forbid',
+    )
+    request: Request | None = None
+    response: Response | None = None
diff --git a/src/opensquilla/contracts/generated/v4/skills_install_status_metadata.py b/src/opensquilla/contracts/generated/v4/skills_install_status_metadata.py
new file mode 100644
index 0000000000..088286f42a
--- /dev/null
+++ b/src/opensquilla/contracts/generated/v4/skills_install_status_metadata.py
@@ -0,0 +1,13 @@
+# @generated by scripts/contracts/generate_gateway_contracts.py; do not edit.
+# source-sha256: c436f674f065deef98d0ef54d8a5ca738fa1a010c56281ecb1f780745bf9290b
+# ruff: noqa
+
+from typing import Final
+
+SKILLS_INSTALL_STATUS_CONTRACT_KIND: Final = 'query'
+SKILLS_INSTALL_STATUS_METHOD: Final = 'skills.install.status'
+SKILLS_INSTALL_STATUS_SCOPE: Final = 'operator.admin'
+SKILLS_INSTALL_STATUS_IDEMPOTENCY: Final = 'read-only'
+SKILLS_INSTALL_STATUS_TIMEOUT: Final = {'policy': 'caller'}
+SKILLS_INSTALL_STATUS_CAPABILITY: Final = {'kind': 'method-availability', 'name': 'skills.install.status', 'optional': True}
+SKILLS_INSTALL_STATUS_ERRORS: Final = [{'code': 'INVALID_REQUEST'}, {'code': 'UNAUTHORIZED'}, {'code': 'UNSUPPORTED'}, {'code': 'INTERNAL_ERROR'}]
diff --git a/src/opensquilla/engine/agent.py b/src/opensquilla/engine/agent.py
index ac23be5fc2..9d470b9d94 100644
--- a/src/opensquilla/engine/agent.py
+++ b/src/opensquilla/engine/agent.py
@@ -11761,6 +11761,16 @@ async def _run_one(tc: ToolCall) -> ToolResult:
                                         timed_out=True,
                                     ),
                                 )
+                                if (
+                                    tc.tool_name == "skill_install_community"
+                                    and cancellation_policy == "must_settle"
+                                    and execution_task.done()
+                                    and not execution_task.cancelled()
+                                    and execution_task.exception() is None
+                                ):
+                                    # The installer settles cancellation against its durable
+                                    # commit receipt before returning. Preserve that result.
+                                    res = execution_task.result()
                         except asyncio.CancelledError:
                             if (
                                 execution_task is not None
diff --git a/src/opensquilla/gateway/adapters/skill_management_contract.py b/src/opensquilla/gateway/adapters/skill_management_contract.py
index 6812782e45..61eedac18e 100644
--- a/src/opensquilla/gateway/adapters/skill_management_contract.py
+++ b/src/opensquilla/gateway/adapters/skill_management_contract.py
@@ -19,6 +19,7 @@
     "skills.reload",
     "skills.install",
     "skills.install.cancel",
+    "skills.install.status",
     "skills.deps.install",
     "skills.uninstall",
 )
diff --git a/src/opensquilla/gateway/rpc_skills.py b/src/opensquilla/gateway/rpc_skills.py
index f48ea59ec9..605550cf98 100644
--- a/src/opensquilla/gateway/rpc_skills.py
+++ b/src/opensquilla/gateway/rpc_skills.py
@@ -1320,6 +1320,16 @@ async def _install_skill(
 ) -> SkillMutationResult:
     """Install a skill from a Community source."""
     operation_id = command.operation_id
+    installer = _management_service(ctx)
+    if operation_id and isinstance(installer, SkillManagementService):
+        result = await installer.install_operations.run(
+            _install_owner(ctx), operation_id,
+            {"identifier": command.identifier, "source": command.source, "force": command.force,
+             "replaceSource": command.replace_source,
+             "riskConfirmation": command.risk_confirmation},
+            lambda: _run_skill_install(command, ctx),
+        )
+        return cast(SkillMutationResult, result)
     operation_key = (
         _skill_install_operation_key(ctx, operation_id) if operation_id else None
     )
@@ -1427,7 +1437,12 @@ async def _cancel_skill_install(
     command: CancelSkillInstall,
     ctx: RpcContext,
 ) -> SkillCancelResult:
-    """Cancel one install owned by this connection and await its cleanup."""
+    """Cancel an owned install and wait for its commit or rollback result."""
+    installer = _management_service(ctx)
+    if isinstance(installer, SkillManagementService):
+        return cast(SkillCancelResult, await installer.install_operations.cancel(
+            _install_owner(ctx), command.operation_id,
+        ))
     operation_key = _skill_install_operation_key(ctx, command.operation_id)
     active_installs = _active_skill_installs(ctx)
     task = active_installs.get(operation_key)
@@ -1799,6 +1814,34 @@ async def _handle_skills_install(
     return await _skill_management(ctx).install(params)
 
 
+def _install_owner(ctx: RpcContext) -> str:
+    principal = ctx.principal
+    if principal.token_public_id:
+        return f"token:{principal.token_public_id}"
+    if principal.is_owner:
+        return "local-owner"
+    raise ValueError("Stable caller identity is required for Skill installation operations")
+
+
+async def _handle_skills_install_status(
+    params: dict[str, Any] | None, ctx: RpcContext,
+) -> dict[str, Any]:
+    if not isinstance(params, dict):
+        raise ValueError("params.operationId is required")
+    operation_id = GatewaySkillManagementAdapter._operation_id(params, required=True)
+    installer = _management_service(ctx)
+    if not isinstance(installer, SkillManagementService):
+        return {"operationId": operation_id, "state": "unknown", "phase": "unknown",
+                "terminal": True, "scope": ""}
+    owner = _install_owner(ctx)
+    receipt = installer.install_operations.store.read(owner, operation_id)
+    import hashlib
+    scope = hashlib.sha256(
+        f"{installer.journal_path}:{owner}".encode(),
+    ).hexdigest()
+    return {**receipt, "scope": scope}
+
+
 async def _handle_skills_install_cancel(
     params: dict[str, Any] | None,
     ctx: RpcContext,
@@ -1824,6 +1867,7 @@ async def _handle_skills_uninstall(
     ("skills.reload", _handle_skills_reload),
     ("skills.install", _handle_skills_install),
     ("skills.install.cancel", _handle_skills_install_cancel),
+    ("skills.install.status", _handle_skills_install_status),
     ("skills.deps.install", _handle_skills_deps_install),
     ("skills.uninstall", _handle_skills_uninstall),
 ):
diff --git a/src/opensquilla/gateway/scopes.py b/src/opensquilla/gateway/scopes.py
index b94380e9aa..ba1197b5f3 100644
--- a/src/opensquilla/gateway/scopes.py
+++ b/src/opensquilla/gateway/scopes.py
@@ -380,6 +380,7 @@
     "agents.files.set": ADMIN_SCOPE,
     "skills.install": ADMIN_SCOPE,
     "skills.install.cancel": ADMIN_SCOPE,
+    "skills.install.status": ADMIN_SCOPE,
     "skills.update": ADMIN_SCOPE,
     "skills.uninstall": ADMIN_SCOPE,
     "skills.reload": ADMIN_SCOPE,
diff --git a/src/opensquilla/skills/hub/clawhub.py b/src/opensquilla/skills/hub/clawhub.py
index b12bf7de7d..ed4499e645 100644
--- a/src/opensquilla/skills/hub/clawhub.py
+++ b/src/opensquilla/skills/hub/clawhub.py
@@ -25,6 +25,7 @@
     SkillDiagnostic,
 )
 from opensquilla.skills.hub.github import GitHubSource
+from opensquilla.skills.hub.operations import report_install_progress
 from opensquilla.skills.hub.source import (
     SkillBundle,
     SkillMeta,
@@ -788,6 +789,7 @@ async def _fetch_into(
                                 with archive_path.open("wb") as output:
                                     async for chunk in response.aiter_bytes(CHUNK_SIZE):
                                         size += len(chunk)
+                                        report_install_progress("downloading", downloadedBytes=size)
                                         if exceeds_limit(
                                             size, DEFAULT_ARCHIVE_LIMITS.max_archive_bytes
                                         ):
diff --git a/src/opensquilla/skills/hub/github.py b/src/opensquilla/skills/hub/github.py
index 068ec209d9..0643f40336 100644
--- a/src/opensquilla/skills/hub/github.py
+++ b/src/opensquilla/skills/hub/github.py
@@ -28,6 +28,7 @@
     DiagnosticSeverity,
     SkillDiagnostic,
 )
+from opensquilla.skills.hub.operations import report_install_progress
 from opensquilla.skills.hub.source import (
     SkillBundle,
     SkillMeta,
@@ -1009,9 +1010,11 @@ async def fetch_resolved_into(
                 pending = iter(selected)
                 actual_total = 0
                 budget = _DownloadBudget(DEFAULT_ARCHIVE_LIMITS.max_expanded_bytes)
+                completed_files = 0
+                report_install_progress("downloading", totalFiles=len(selected), completedFiles=0)
 
                 async def worker() -> None:
-                    nonlocal actual_total
+                    nonlocal actual_total, completed_files
                     for path, rel_path, _declared_size, file_mode in pending:
                         raw_url = (
                             f"https://raw.githubusercontent.com/{ref.repo_full}/"
@@ -1022,6 +1025,11 @@ async def worker() -> None:
                             client, raw_url, target, self._headers(), budget=budget,
                         )
                         actual_total += target.stat().st_size
+                        completed_files += 1
+                        report_install_progress(
+                            "downloading", completedFiles=completed_files,
+                            totalFiles=len(selected), downloadedBytes=actual_total,
+                        )
                         if exceeds_limit(actual_total, DEFAULT_ARCHIVE_LIMITS.max_expanded_bytes):
                             raise ValueError("Skill exceeds configured expanded-size limit")
                         if file_mode:
diff --git a/src/opensquilla/skills/hub/management.py b/src/opensquilla/skills/hub/management.py
index 05d34e0bc6..fdf530f262 100644
--- a/src/opensquilla/skills/hub/management.py
+++ b/src/opensquilla/skills/hub/management.py
@@ -48,6 +48,13 @@
     compute_sha256,
     compute_tree_sha256,
 )
+from opensquilla.skills.hub.operations import (
+    InstallOperations,
+    InstallOperationStore,
+    current_install_operation,
+    operation_database,
+    report_install_progress,
+)
 from opensquilla.skills.hub.router import SourceRouter
 from opensquilla.skills.hub.scanner import ScanResult, scan_skill_tree
 from opensquilla.skills.hub.source import (
@@ -65,6 +72,7 @@
     fsync_directory,
     fsync_staging_tree,
     guard_retained_recovery_journal,
+    managed_root_identity,
     path_is_occupied,
     recover_pending_skill_transaction,
     remove_transaction_journal,
@@ -459,6 +467,10 @@ def to_dict(self) -> dict[str, Any]:
             "rollbackPerformed": self.rollback_performed,
             "catalogGeneration": self.catalog_generation,
             "effectiveFrom": self.effective_from,
+            **({"recoveryRequired": True} if any(
+                item.blocking and item.code == "RECOVERY_REQUIRED"
+                for item in self.diagnostics
+            ) else {}),
         }
 
     def as_dict(self) -> dict[str, Any]:
@@ -1175,12 +1187,23 @@ def __init__(
         self._journal_path = journal_path or default_journal_path(managed_dir)
         self._mutation_lock = mutation_lock or mutation_lock_for(managed_dir)
         self._offline = offline or loader is None
+        self._install_operations: InstallOperations | None = None
         self._recovery_required_diagnostics: tuple[SkillDiagnostic, ...] = ()
         bind_lockfile = getattr(loader, "bind_managed_lockfile", None)
         if callable(bind_lockfile):
             bind_lockfile(lockfile_path)
         self._observe_recovery(list(startup_recovery_diagnostics))
 
+    @property
+    def install_operations(self) -> InstallOperations:
+        if self._install_operations is None:
+            self._install_operations = InstallOperations(InstallOperationStore(
+                operation_database(self._journal_path),
+                root_id=managed_root_identity(self._managed_dir),
+            ))
+        self._install_operations.store.recover_orphans(self._journal_path)
+        return self._install_operations
+
     @property
     def managed_dir(self) -> Path:
         return self._managed_dir
@@ -2031,6 +2054,7 @@ async def _install_or_update(
             )
             return self._recovery_required_result(recovery_name)
 
+        report_install_progress("downloading")
         transaction_id = uuid.uuid4().hex
         transaction_root = staging_root(self._managed_dir) / transaction_id
         raw_candidate = transaction_root / "_candidate"
@@ -2178,6 +2202,7 @@ def cleanup_pre_journal_reservation() -> None:
                 resolution,
                 identifier,
             )
+            report_install_progress("scanning")
             scan_result = await run_staging_worker(scan_skill_tree, candidate_dir)
             risk_confirmation_details: dict[str, Any] = {}
             risk_acknowledged = False
@@ -2644,6 +2669,7 @@ def reject_unconfirmed_risk() -> None:
                     rollback=rollback,
                     lockfile_path=self._lockfile_path,
                 )
+                report_install_progress("committing")
                 fsync_staging_tree(candidate_dir)
                 fsync_directory(rollback.parent)
                 fsync_directory(rollback.parent.parent)
@@ -2805,11 +2831,18 @@ def reject_unconfirmed_risk() -> None:
                     effective_from="next_turn" if self._loader is not None else "next_start",
                 )
                 if journal is not None:
+                    if journal.install_operation_id:
+                        journal.install_receipt = success_result.to_dict()
                     journal.advance("committed", self._journal_path)
                 durably_committed = True
                 if publication_barrier is not None:
                     publication_barrier.commit()
                 try:
+                    operation = current_install_operation()
+                    if operation is not None:
+                        operation.store.finish(
+                            operation.owner, operation.id, success_result.to_dict(),
+                        )
                     if journal is not None:
                         validate_transaction_journal_paths(
                             journal,
@@ -2944,9 +2977,7 @@ def reject_unconfirmed_risk() -> None:
                     drifted = True
             if journal is None or not path_is_occupied(self._journal_path):
                 cleanup_pre_journal_reservation()
-            if not isinstance(exc, Exception):
-                raise
-            return self._failure(
+            failure_result = self._failure(
                 name=name,
                 message=str(exc) or type(exc).__name__,
                 diagnostics=diagnostics,
@@ -2965,6 +2996,16 @@ def reject_unconfirmed_risk() -> None:
                 install_id=old_entry.install_id if old_entry is not None else "",
                 compatibility=_entry_compatibility(old_entry),
             )
+            if not isinstance(exc, Exception):
+                operation = current_install_operation()
+                if isinstance(exc, asyncio.CancelledError) and operation is not None:
+                    payload = failure_result.to_dict()
+                    payload["cancelled"] = not payload.get("recoveryRequired", False)
+                    if payload["cancelled"]:
+                        payload["message"] = "Skill installation cancelled"
+                    operation.store.finish(operation.owner, operation.id, payload)
+                raise
+            return failure_result
         finally:
             publication_stack.close()
 
diff --git a/src/opensquilla/skills/hub/operations.py b/src/opensquilla/skills/hub/operations.py
new file mode 100644
index 0000000000..e38dc3fff1
--- /dev/null
+++ b/src/opensquilla/skills/hub/operations.py
@@ -0,0 +1,381 @@
+"""Durable install operations independent of transport connection lifetimes."""
+
+from __future__ import annotations
+
+import asyncio
+import contextvars
+import hashlib
+import json
+import sqlite3
+import time
+import uuid
+from collections.abc import Awaitable, Callable, Iterator, Mapping
+from contextlib import contextmanager
+from pathlib import Path
+from typing import Any
+
+_PROCESS_ID = uuid.uuid4().hex
+_CURRENT: contextvars.ContextVar[InstallOperation | None] = contextvars.ContextVar(
+    "skill_install_operation",
+    default=None,
+)
+
+
+def operation_database(journal_path: Path) -> Path:
+    return journal_path.parent / "install-operations.sqlite3"
+
+
+def _interrupted_result(phase: str, result: dict[str, Any] | None) -> tuple[str, dict[str, Any]]:
+    if phase == "rolled_back" and result and result.get("rollbackPerformed") is True:
+        return "failed", result
+    return "recovery_required", {
+        "success": False,
+        "message": "The interrupted install has no proven terminal receipt.",
+        "recoveryRequired": True,
+    }
+
+
+class InstallOperationStore:
+    """Store receipts in the profile's existing Skill transaction state directory."""
+
+    def __init__(self, path: Path, *, root_id: str, process_id: str = _PROCESS_ID) -> None:
+        self.path = path
+        self.root_id = root_id
+        self.process_id = process_id
+        path.parent.mkdir(parents=True, exist_ok=True)
+        if path.is_symlink() or (getattr(path, "is_junction", lambda: False)()):
+            raise ValueError("Skill operation database must not be a link")
+        with self.connect() as db:
+            db.execute("""CREATE TABLE IF NOT EXISTS skill_installs (
+                root TEXT NOT NULL, owner TEXT NOT NULL, id TEXT NOT NULL,
+                signature TEXT NOT NULL, process TEXT NOT NULL, state TEXT NOT NULL,
+                phase TEXT NOT NULL, progress TEXT NOT NULL, result TEXT,
+                created REAL NOT NULL, updated REAL NOT NULL,
+                PRIMARY KEY(root, owner, id))""")
+            # Retire IDs independently of expiring receipt contents. An old ID
+            # must never become a fresh mutation merely because its receipt aged out.
+            db.execute("""CREATE TABLE IF NOT EXISTS skill_install_keys (
+                root TEXT NOT NULL, owner TEXT NOT NULL, id TEXT NOT NULL,
+                PRIMARY KEY(root, owner, id))""")
+            db.execute(
+                "INSERT OR IGNORE INTO skill_install_keys "
+                "SELECT root, owner, id FROM skill_installs"
+            )
+
+    @contextmanager
+    def connect(self) -> Iterator[sqlite3.Connection]:
+        db = sqlite3.connect(self.path, timeout=5)
+        db.row_factory = sqlite3.Row
+        try:
+            with db:
+                yield db
+        finally:
+            db.close()
+
+    def create(self, owner: str, operation_id: str, signature: str) -> bool:
+        uuid.UUID(operation_id)
+        now = time.time()
+        with self.connect() as db:
+            claimed = db.execute(
+                "INSERT OR IGNORE INTO skill_install_keys VALUES (?, ?, ?)",
+                (self.root_id, owner, operation_id),
+            ).rowcount == 1
+            if not claimed and db.execute(
+                "SELECT 1 FROM skill_installs WHERE root=? AND owner=? AND id=?",
+                (self.root_id, owner, operation_id),
+            ).fetchone() is None:
+                raise ValueError(
+                    "operationId receipt expired; query status without replaying install"
+                )
+            cursor = db.execute(
+                "INSERT OR IGNORE INTO skill_installs VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
+                (
+                    self.root_id,
+                    owner,
+                    operation_id,
+                    signature,
+                    self.process_id,
+                    "running",
+                    "resolving",
+                    "{}",
+                    None,
+                    now,
+                    now,
+                ),
+            )
+            created = cursor.rowcount == 1
+            if not created:
+                row = db.execute(
+                    "SELECT signature, state FROM skill_installs WHERE root=? AND owner=? AND id=?",
+                    (self.root_id, owner, operation_id),
+                ).fetchone()
+                if row["signature"] != signature and not (
+                    row["state"] == "cancelled" and not row["signature"]
+                ):
+                    raise ValueError("operationId is already associated with another install")
+        self.prune()
+        return created
+
+    def read(self, owner: str, operation_id: str) -> dict[str, Any]:
+        uuid.UUID(operation_id)
+        with self.connect() as db:
+            row = db.execute(
+                "SELECT * FROM skill_installs WHERE root=? AND owner=? AND id=?",
+                (self.root_id, owner, operation_id),
+            ).fetchone()
+        if row is None:
+            return {
+                "operationId": operation_id,
+                "state": "unknown",
+                "phase": "unknown",
+                "terminal": True,
+            }
+        state = row["state"]
+        result = json.loads(row["result"]) if row["result"] else None
+        if state == "running":
+            if row["process"] != self.process_id:
+                state, result = _interrupted_result(row["phase"], result)
+            else:
+                # A rollback checkpoint proves the disk state after a crash,
+                # but the current worker still owns catalog/cancellation settlement.
+                result = None
+        payload = {
+            "operationId": operation_id,
+            "state": state,
+            "phase": row["phase"],
+            "terminal": state != "running",
+            "progress": json.loads(row["progress"]),
+        }
+        if result is not None:
+            payload["result"] = result
+        return payload
+
+    def progress(self, owner: str, operation_id: str, phase: str, progress: dict[str, Any]) -> None:
+        with self.connect() as db:
+            db.execute(
+                "UPDATE skill_installs SET phase=?, progress=?, updated=? "
+                "WHERE root=? AND owner=? AND id=? AND state='running'",
+                (phase, json.dumps(progress), time.time(), self.root_id, owner, operation_id),
+            )
+
+    def checkpoint_rollback(self, owner: str, operation_id: str) -> None:
+        """Persist disk/lock restoration before its transaction journal is removed."""
+        result = {
+            "success": False,
+            "message": "Interrupted installation was rolled back.",
+            "rollbackPerformed": True,
+        }
+        with self.connect() as db:
+            updated = db.execute(
+                "UPDATE skill_installs SET phase='rolled_back', result=?, updated=? "
+                "WHERE root=? AND owner=? AND id=? AND process=? AND state='running'",
+                (json.dumps(result), time.time(), self.root_id, owner, operation_id,
+                 self.process_id),
+            ).rowcount
+            if updated != 1:
+                raise ValueError("Install rollback checkpoint has no running operation")
+
+    def finish(
+        self, owner: str, operation_id: str, result: dict[str, Any], *,
+        state: str = "", update_details: bool = False,
+    ) -> None:
+        if not state:
+            state = (
+                "succeeded"
+                if result.get("success")
+                else "recovery_required"
+                if result.get("recoveryRequired")
+                else "cancelled"
+                if result.get("cancelled")
+                else "failed"
+            )
+        with self.connect() as db:
+            now = time.time()
+            db.execute(
+                "INSERT OR IGNORE INTO skill_install_keys VALUES (?, ?, ?)",
+                (self.root_id, owner, operation_id),
+            )
+            db.execute(
+                "INSERT OR IGNORE INTO skill_installs VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
+                (
+                    self.root_id,
+                    owner,
+                    operation_id,
+                    "",
+                    self.process_id,
+                    state,
+                    "complete",
+                    "{}",
+                    json.dumps(result),
+                    now,
+                    now,
+                ),
+            )
+            db.execute(
+                "UPDATE skill_installs SET state=?, phase='complete', result=?, updated=? "
+                "WHERE root=? AND owner=? AND id=? AND (state IN ('running', 'recovery_required') "
+                "OR (? AND process=? AND state=?))",
+                (state, json.dumps(result), time.time(), self.root_id, owner, operation_id,
+                 update_details, self.process_id, state),
+            )
+        self.prune()
+
+    def recover_orphans(self, journal_path: Path) -> None:
+        if journal_path.exists():
+            return
+        with self.connect() as db:
+            rows = db.execute(
+                "SELECT owner, id, phase, result FROM skill_installs "
+                "WHERE root=? AND state='running' AND process!=?",
+                (self.root_id, self.process_id),
+            ).fetchall()
+            for row in rows:
+                state, result = _interrupted_result(
+                    row["phase"], json.loads(row["result"]) if row["result"] else None,
+                )
+                db.execute(
+                    "UPDATE skill_installs SET state=?, phase='complete', result=?, updated=? "
+                    "WHERE root=? AND owner=? AND id=? AND state='running' AND process!=?",
+                    (state, json.dumps(result), time.time(), self.root_id,
+                     row["owner"], row["id"], self.process_id),
+                )
+        self.prune()
+
+    def prune(self) -> None:
+        with self.connect() as db:
+            db.execute(
+                "DELETE FROM skill_installs WHERE root=? AND state!='running' AND updated<?",
+                (self.root_id, time.time() - 7 * 86400),
+            )
+            db.execute(
+                "DELETE FROM skill_installs WHERE rowid IN (SELECT rowid FROM skill_installs "
+                "WHERE root=? AND state!='running' ORDER BY updated DESC, rowid DESC "
+                "LIMIT -1 OFFSET 1000)",
+                (self.root_id,),
+            )
+
+
+class InstallOperation:
+    def __init__(self, store: InstallOperationStore, owner: str, operation_id: str) -> None:
+        self.store = store
+        self.owner = owner
+        self.id = operation_id
+        self.last_progress = 0.0
+        self.phase = ""
+        self.counts: dict[str, Any] = {}
+
+    def progress(self, phase: str, **counts: Any) -> None:
+        self.counts.update(counts)
+        now = time.monotonic()
+        if phase != self.phase or now - self.last_progress >= 0.25:
+            self.store.progress(self.owner, self.id, phase, self.counts)
+            self.phase = phase
+            self.last_progress = now
+
+
+def report_install_progress(phase: str, **counts: Any) -> None:
+    operation = _CURRENT.get()
+    if operation is not None:
+        operation.progress(phase, **counts)
+
+
+def current_install_operation() -> InstallOperation | None:
+    return _CURRENT.get()
+
+
+class InstallOperations:
+    def __init__(self, store: InstallOperationStore) -> None:
+        self.store = store
+        self.tasks: dict[tuple[str, str], asyncio.Task[dict[str, Any]]] = {}
+
+    async def run(
+        self,
+        owner: str,
+        operation_id: str,
+        request: dict[str, Any],
+        install: Callable[[], Awaitable[Mapping[str, Any]]],
+    ) -> dict[str, Any]:
+        signature = hashlib.sha256(json.dumps(request, sort_keys=True).encode()).hexdigest()
+        key = (owner, operation_id)
+        if self.store.create(owner, operation_id, signature):
+
+            async def execute() -> dict[str, Any]:
+                context = InstallOperation(self.store, owner, operation_id)
+                token = _CURRENT.set(context)
+                try:
+                    result = dict(await install())
+                except asyncio.CancelledError:
+                    receipt = self.store.read(owner, operation_id)
+                    result = receipt.get("result") or {
+                        "success": False,
+                        "cancelled": True,
+                        "message": "Skill installation cancelled",
+                    }
+                except Exception as exc:
+                    receipt = self.store.read(owner, operation_id)
+                    result = receipt.get("result") or {"success": False, "message": str(exc)}
+                finally:
+                    _CURRENT.reset(token)
+                # The worker may add cleanup diagnostics after the durable
+                # commit. Refresh its details without changing a proven outcome;
+                # restart recovery must not replace them with the older journal.
+                self.store.finish(owner, operation_id, result, update_details=True)
+                return dict(self.store.read(owner, operation_id)["result"])
+
+            task = asyncio.create_task(execute(), name=f"skill-install-{operation_id}")
+            self.tasks[key] = task
+            task.add_done_callback(lambda done: self.tasks.pop(key, None))
+        existing_task = self.tasks.get(key)
+        if existing_task is not None:
+            try:
+                return await asyncio.shield(existing_task)
+            except asyncio.CancelledError:
+                if not existing_task.cancelled():
+                    raise
+                cancelled = {
+                    "success": False,
+                    "cancelled": True,
+                    "message": "Skill installation cancelled",
+                }
+                self.store.finish(owner, operation_id, cancelled)
+                return cancelled
+        receipt = self.store.read(owner, operation_id)
+        if "result" in receipt:
+            return dict(receipt["result"])
+        raise ValueError("operationId is already running; query install status")
+
+    async def cancel(self, owner: str, operation_id: str) -> dict[str, Any]:
+        if self.store.read(owner, operation_id)["state"] == "unknown":
+            self.store.create(owner, operation_id, "")
+            self.store.finish(
+                owner,
+                operation_id,
+                {
+                    "success": False,
+                    "cancelled": True,
+                    "message": "Skill installation cancelled",
+                },
+            )
+        task = self.tasks.get((owner, operation_id))
+        if task is not None:
+            task.cancel()
+            try:
+                await asyncio.shield(task)
+            except asyncio.CancelledError:
+                if not task.cancelled():
+                    raise
+                self.store.finish(
+                    owner,
+                    operation_id,
+                    {
+                        "success": False,
+                        "cancelled": True,
+                        "message": "Skill installation cancelled",
+                    },
+                )
+        receipt = self.store.read(owner, operation_id)
+        result = receipt.get("result") or {
+            "success": False,
+            "message": "Install operation is unknown",
+        }
+        return {**result, "cancelled": bool(result.get("cancelled")), "pending": False}
diff --git a/src/opensquilla/skills/hub/transaction.py b/src/opensquilla/skills/hub/transaction.py
index dfea209de6..8469baae09 100644
--- a/src/opensquilla/skills/hub/transaction.py
+++ b/src/opensquilla/skills/hub/transaction.py
@@ -322,6 +322,9 @@ class SkillTransactionJournal:
     old_lock_b64: str
     old_target_exists: bool = False
     old_target_sha256: str = ""
+    install_operation_id: str = ""
+    install_operation_owner: str = ""
+    install_receipt: dict[str, Any] | None = None
 
     @classmethod
     def prepare(
@@ -342,7 +345,11 @@ def prepare(
         old_target_sha256 = (
             compute_tree_sha256(target) if old_target_exists else ""
         )
+        from opensquilla.skills.hub.operations import current_install_operation
+        install_operation = current_install_operation()
         journal = cls(
+            install_operation_id=install_operation.id if install_operation else "",
+            install_operation_owner=install_operation.owner if install_operation else "",
             operation=operation,
             phase="prepared",
             managed_root=str(managed_dir.resolve(strict=False)),
@@ -380,6 +387,11 @@ def to_dict(self) -> dict[str, Any]:
             "old_lock_b64": self.old_lock_b64,
             "old_target_exists": self.old_target_exists,
             "old_target_sha256": self.old_target_sha256,
+            **({
+                "install_operation_id": self.install_operation_id,
+                "install_operation_owner": self.install_operation_owner,
+                "install_receipt": self.install_receipt,
+            } if self.install_operation_id else {}),
         }
 
     def write(self, path: Path) -> None:
@@ -442,6 +454,20 @@ def load(cls, path: Path) -> SkillTransactionJournal | None:
                 raise ValueError(
                     f"Skill transaction journal field {field!r} must be a boolean"
                 )
+        operation_id = data.get("install_operation_id", "")
+        operation_owner = data.get("install_operation_owner", "")
+        receipt = data.get("install_receipt")
+        if operation_id:
+            import uuid
+            uuid.UUID(operation_id)
+            if (not isinstance(operation_owner, str) or not operation_owner
+                    or len(operation_owner) > 512):
+                raise ValueError("Invalid install operation owner in journal")
+            if receipt is not None and (not isinstance(receipt, dict)
+                                        or type(receipt.get("success")) is not bool):
+                raise ValueError("Invalid install receipt in journal")
+        elif operation_owner or receipt is not None:
+            raise ValueError("Install receipt has no operation identity")
         phase = data["phase"]
         if phase not in _VALID_PHASES:
             raise ValueError(f"invalid Skill transaction phase: {phase}")
@@ -459,6 +485,9 @@ def load(cls, path: Path) -> SkillTransactionJournal | None:
             old_lock_b64=data["old_lock_b64"],
             old_target_exists=data["old_target_exists"],
             old_target_sha256=data["old_target_sha256"],
+            install_operation_id=operation_id,
+            install_operation_owner=operation_owner,
+            install_receipt=receipt,
         )
 
 
@@ -999,6 +1028,45 @@ def recover_pending_skill_transaction(
                 )
             )
             return recovered
+        from opensquilla.skills.hub.operations import (
+            InstallOperationStore,
+            current_install_operation,
+            operation_database,
+        )
+
+        current_operation = current_install_operation()
+        settles_in_caller = bool(
+            current_operation is not None
+            and current_operation.id == journal.install_operation_id
+            and current_operation.owner == journal.install_operation_owner
+            and current_operation.store.root_id == journal.managed_root_id
+        )
+        # A running installer still has to restore the live catalog and report
+        # why it rolled back. Only orphan recovery owns its terminal receipt.
+        if settles_in_caller and journal.phase != "committed":
+            assert current_operation is not None
+            current_operation.store.checkpoint_rollback(
+                current_operation.owner, current_operation.id,
+            )
+        if journal.install_operation_id and not settles_in_caller:
+
+            store = InstallOperationStore(
+                operation_database(selected_journal), root_id=journal.managed_root_id,
+            )
+            if journal.phase == "committed" and journal.install_receipt is not None:
+                store.finish(journal.install_operation_owner, journal.install_operation_id,
+                             journal.install_receipt)
+            elif journal.phase != "committed":
+                store.finish(journal.install_operation_owner, journal.install_operation_id, {
+                    "success": False, "message": "Interrupted installation was rolled back.",
+                    "rollbackPerformed": True,
+                })
+            else:
+                store.finish(journal.install_operation_owner, journal.install_operation_id, {
+                    "success": False,
+                    "message": "Committed transaction has no recoverable receipt.",
+                    "recoveryRequired": True,
+                }, state="recovery_required")
         remove_transaction_journal(selected_journal)
         return recovered
     except Exception as exc:
diff --git a/src/opensquilla/tools/builtin/skill_tools.py b/src/opensquilla/tools/builtin/skill_tools.py
index d7b758b8f3..1e0654b071 100644
--- a/src/opensquilla/tools/builtin/skill_tools.py
+++ b/src/opensquilla/tools/builtin/skill_tools.py
@@ -637,6 +637,8 @@ async def skill_search_community(
         },
         required=["identifier"],
         owner_only=True,
+        execution_timeout_seconds=600,
+        cancellation_policy="must_settle",
     )
     async def skill_install_community(
         identifier: str,
@@ -672,13 +674,44 @@ async def skill_install_community(
             }
             installer = build_default_skill_installer(**builder_kwargs)
         if isinstance(installer, SkillManagementService):
-            result = await installer.install(
-                clean_identifier,
-                source_id,
-                force=force,
-                replace_source=replace_source,
-                risk_confirmation=clean_risk_confirmation,
-            )
+            import asyncio
+            import hashlib
+            import uuid
+
+            context = current_tool_context.get()
+            caller = f"{getattr(context, 'agent_id', '')}:{getattr(context, 'session_key', '')}"
+            owner = "agent:" + hashlib.sha256(caller.encode()).hexdigest()
+            operation_id = str(uuid.uuid4())
+
+            async def install_operation() -> dict[str, Any]:
+                installed = await installer.install(
+                    clean_identifier, source_id, force=force,
+                    replace_source=replace_source, risk_confirmation=clean_risk_confirmation,
+                )
+                return installed.to_dict()
+
+            try:
+                payload = await installer.install_operations.run(
+                    owner, operation_id,
+                    {"identifier": clean_identifier, "source": source_id, "force": force,
+                     "replaceSource": replace_source, "riskConfirmation": clean_risk_confirmation},
+                    install_operation,
+                )
+            except asyncio.CancelledError:
+                payload = await installer.install_operations.cancel(owner, operation_id)
+            payload = dict(payload)
+            payload.update({
+                "operationId": operation_id, "identifier": clean_identifier, "source": source_id,
+                "status": "installed" if payload.get("success") else "cancelled"
+                if payload.get("cancelled") else "failed",
+            })
+            if payload.get("success"):
+                payload["message"] = (
+                    f"{payload.get('message', '')} "
+                    "The current turn keeps its pinned Skill catalog; "
+                    "the installed catalog becomes observable from the next turn."
+                )
+            return json.dumps(payload)
         else:
             install = installer.install
             if replace_source and not supports_keyword_argument(install, "replace_source"):
@@ -724,7 +757,7 @@ async def skill_install_community(
                     result = exc.result
 
         serializer = getattr(result, "to_dict", None)
-        payload: dict[str, Any] = dict(serializer()) if callable(serializer) else {}
+        payload = dict(serializer()) if callable(serializer) else {}
         payload.update(
             {
                 "status": "installed" if result.success else "failed",
diff --git a/tests/contracts/test_gateway_contract_runner.py b/tests/contracts/test_gateway_contract_runner.py
index 8c76aec5d5..7bc921c48e 100644
--- a/tests/contracts/test_gateway_contract_runner.py
+++ b/tests/contracts/test_gateway_contract_runner.py
@@ -493,14 +493,15 @@ def test_compatibility_manifest_is_schema_derived_and_deterministic() -> None:
     assert manifest["protocol"] == runner.GATEWAY_PROTOCOL
     assert manifest["wireVersion"] == 4
     assert manifest["source"] == {
-        "schemaCount": 224,
-        "methodCount": 214,
+        "schemaCount": 225,
+        "methodCount": 215,
         "eventFamilyCount": 10,
         "schemaTreeSha256": runner._schema_tree_digest(specs),
         "generatorSha256": runner._generator_digest(),
     }
+    assert any(entry["name"] == "skills.install.status" for entry in manifest["methods"])
     assert Counter(entry["lifecycle"] for entry in manifest["methods"]) == {
-        "stable": 211,
+        "stable": 212,
         "legacy": 3,
     }
     assert [
diff --git a/tests/contracts/test_gateway_validator_profiles.py b/tests/contracts/test_gateway_validator_profiles.py
index 90ffa1f34b..722a085dd5 100644
--- a/tests/contracts/test_gateway_validator_profiles.py
+++ b/tests/contracts/test_gateway_validator_profiles.py
@@ -22,16 +22,17 @@ def test_production_targets_preserve_every_approved_validator_role() -> None:
     specs = runner.discover_contracts()
     targets = runner.load_production_targets(specs)
 
-    assert len(targets) == 196
+    assert len(targets) == 197
     assert Counter(role for roles in targets.values() for role in roles) == {
-        "result": 186,
+        "result": 187,
         "params": 20,
         "payload": 9,
         "frame": 1,
     }
-    assert sum(len(spec.targets) for spec in specs) == 866
+    assert sum(len(spec.targets) for spec in specs) == 870
     assert targets[("method", "sessions.executionLog.read")] == ("params", "result")
     assert targets[("method", "sessions.list")] == ("result",)
+    assert targets[("method", "skills.install.status")] == ("result",)
     assert targets[("method", "meta.list")] == ("result",)
     assert targets[("method", "meta.inspect")] == ("result",)
     assert targets[("method", "telemetry.product_active.record")] == ("result",)
diff --git a/tests/contracts/test_r7_contract_freeze.py b/tests/contracts/test_r7_contract_freeze.py
index 843a4215c2..5e0fe7ca75 100644
--- a/tests/contracts/test_r7_contract_freeze.py
+++ b/tests/contracts/test_r7_contract_freeze.py
@@ -234,9 +234,9 @@ def _specs_by_wire_name():
 def test_contract_inventory_freezes_all_webui_reachable_wire_names() -> None:
     specs = discover_contracts()
 
-    assert len(specs) == 224
+    assert len(specs) == 225
     assert Counter(spec.contract_type for spec in specs) == {
-        "method": 214,
+        "method": 215,
         "event": 10,
     }
     assert EXPECTED_METHOD_METADATA.keys() <= {spec.wire_name for spec in specs}
diff --git a/tests/test_ci/test_rpc_architecture_contracts.py b/tests/test_ci/test_rpc_architecture_contracts.py
index 447819fa90..677420271b 100644
--- a/tests/test_ci/test_rpc_architecture_contracts.py
+++ b/tests/test_ci/test_rpc_architecture_contracts.py
@@ -123,8 +123,8 @@
 # The retired 20-method RPC surface is absent; exact inventory remains pinned
 # so this does not authorize unrelated wire growth.
 # Primary-provider transitions add resetRecommended and upsertAndActivate.
-RUNTIME_RPC_METHOD_BASELINE = 295
-RUNTIME_RPC_METHOD_DIGEST = "34c09211505ec1711fef3a294f254fdbc0ce1bf587d8cf1fc42b87e5943745b5"
+RUNTIME_RPC_METHOD_BASELINE = 296
+RUNTIME_RPC_METHOD_DIGEST = "094f1e13c86c667dda71cb240c47e32a0d8f8db710e0ac913c926443d3de641a"
 STATIC_RPC_DECORATOR_BASELINE = 77
 
 # Physical lines in the sessions/runtime slice remain tracked for the final
diff --git a/tests/test_ci/test_upgrade_baselines.py b/tests/test_ci/test_upgrade_baselines.py
index 2ee3a59ac5..b5af95b9a8 100644
--- a/tests/test_ci/test_upgrade_baselines.py
+++ b/tests/test_ci/test_upgrade_baselines.py
@@ -244,38 +244,47 @@ def test_packaged_recovery_preserves_original_failure_after_cleanup(
         encoding="utf-8",
     )
     # This is a synthetic error/cleanup contract, not a recovery-latency check.
-    # Allow Windows CI headroom for Node startup and captured-pipe completion.
-    try:
-        result = subprocess.run(
-            [
-                node,
-                str(tmp_path / "test-packaged-session-recovery.mjs"),
-                "--executable",
-                str(tmp_path / "synthetic.exe"),
-                "--user-data-dir",
-                str(tmp_path / "profile"),
-                "--session-key",
-                "synthetic-main",
-                "--switch-session-key",
-                "synthetic-peer",
-                "--label",
-                "synthetic",
-            ],
-            capture_output=True,
-            text=True,
-            check=False,
-            timeout=30 if os.name == "nt" else 10,
-        )
-    except subprocess.TimeoutExpired as error:
-        error.add_note(f"Captured stdout: {error.stdout!r}\nCaptured stderr: {error.stderr!r}")
-        raise
+    # File-backed output avoids Windows pipe-reader threads and preserves partial
+    # diagnostics even when the child has not exited by the original deadline.
+    stdout_path = tmp_path / "node-stdout.log"
+    stderr_path = tmp_path / "node-stderr.log"
+    with stdout_path.open("wb") as stdout_file, stderr_path.open("wb") as stderr_file:
+        try:
+            result = subprocess.run(
+                [
+                    node,
+                    str(tmp_path / "test-packaged-session-recovery.mjs"),
+                    "--executable",
+                    str(tmp_path / "synthetic.exe"),
+                    "--user-data-dir",
+                    str(tmp_path / "profile"),
+                    "--session-key",
+                    "synthetic-main",
+                    "--switch-session-key",
+                    "synthetic-peer",
+                    "--label",
+                    "synthetic",
+                ],
+                stdout=stdout_file,
+                stderr=stderr_file,
+                check=False,
+                timeout=30 if os.name == "nt" else 10,
+            )
+        except subprocess.TimeoutExpired as error:
+            error.add_note(
+                f"Captured stdout: {stdout_path.read_text(encoding='utf-8', errors='replace')!r}\n"
+                f"Captured stderr: {stderr_path.read_text(encoding='utf-8', errors='replace')!r}"
+            )
+            raise
+    stdout = stdout_path.read_text(encoding="utf-8")
+    stderr = stderr_path.read_text(encoding="utf-8")
     assert result.returncode != 0
-    assert "packaged_session_recovery_failed_before_cleanup" in result.stderr
-    assert "synthetic-cleanup-ran" in result.stderr
-    assert "Error: synthetic recovery fault" in result.stderr
-    assert not result.stdout
+    assert "packaged_session_recovery_failed_before_cleanup" in stderr
+    assert "synthetic-cleanup-ran" in stderr
+    assert "Error: synthetic recovery fault" in stderr
+    assert not stdout
     if cleanup_fails:
-        assert result.stderr.rindex("Error: synthetic recovery fault") > result.stderr.rindex(
+        assert stderr.rindex("Error: synthetic recovery fault") > stderr.rindex(
             "Error: synthetic cleanup fault"
         )
 
diff --git a/tests/test_ci/test_workflows.py b/tests/test_ci/test_workflows.py
index 3d299c448d..d65e59dc6c 100644
--- a/tests/test_ci/test_workflows.py
+++ b/tests/test_ci/test_workflows.py
@@ -617,6 +617,10 @@ def test_skill_hub_contract_is_integrated_into_canonical_ci() -> None:
         "tests/test_skills_hub_lockfile_contract.py",
         "tests/test_skills_hub_doctor.py",
         "tests/test_skills_hash_consumers.py",
+        "tests/test_engine/test_skill_install_settlement.py",
+        "tests/test_gateway/test_skill_install_status.py",
+        "tests/test_skills/test_hub_install_operations.py",
+        "tests/test_skills/test_staging_io_worker.py",
         "tests/test_skill_install_source.py",
         "tests/test_skills_hub_streaming.py",
         "tests/test_skills_hub_streaming_faults.py",
diff --git a/tests/test_engine/test_skill_install_settlement.py b/tests/test_engine/test_skill_install_settlement.py
new file mode 100644
index 0000000000..de261c78ed
--- /dev/null
+++ b/tests/test_engine/test_skill_install_settlement.py
@@ -0,0 +1,61 @@
+"""A settled install result must survive the Agent tool deadline."""
+
+from __future__ import annotations
+
+import asyncio
+from collections.abc import AsyncIterator
+from typing import Any
+
+import pytest
+
+from opensquilla.engine import Agent, AgentConfig, ToolResult
+from opensquilla.engine.types import ToolCall, ToolResultEvent
+from opensquilla.provider import DoneEvent as ProviderDone
+from opensquilla.provider import ToolDefinition, ToolInputSchema
+from opensquilla.provider import ToolUseEndEvent as ProviderToolUseEnd
+from opensquilla.provider import ToolUseStartEvent as ProviderToolUseStart
+
+
+class InstallProvider:
+    provider_name = "fake"
+
+    async def chat(self, messages: Any, **kwargs: Any) -> AsyncIterator[Any]:
+        yield ProviderToolUseStart(tool_use_id="install-1", tool_name="skill_install_community")
+        yield ProviderToolUseEnd(
+            tool_use_id="install-1", tool_name="skill_install_community",
+            arguments={"identifier": "demo"},
+        )
+        yield ProviderDone(stop_reason="tool_use", input_tokens=1, output_tokens=1)
+
+
+@pytest.mark.asyncio
+async def test_install_timeout_preserves_settled_commit_receipt() -> None:
+    settled = asyncio.Event()
+
+    async def handler(call: ToolCall) -> ToolResult:
+        try:
+            await asyncio.Event().wait()
+        except asyncio.CancelledError:
+            settled.set()
+            return ToolResult(
+                tool_use_id=call.tool_use_id, tool_name=call.tool_name,
+                content='{"success":true,"installed":true,"effectiveFrom":"next_turn"}',
+            )
+
+    agent = Agent(
+        provider=InstallProvider(),
+        config=AgentConfig(max_iterations=1),
+        tool_definitions=[ToolDefinition(
+            name="skill_install_community", description="Install",
+            input_schema=ToolInputSchema(properties={}, required=[]),
+            execution_timeout_seconds=0.01,
+            cancellation_policy="must_settle",
+        )],
+        tool_handler=handler,
+    )
+    events = [event async for event in agent.run_turn("install demo")]
+    result = next(event for event in events if isinstance(event, ToolResultEvent))
+    assert settled.is_set()
+    assert not result.is_error
+    assert '"success":true' in result.result
+    assert "timed out" not in result.result
diff --git a/tests/test_gateway/test_chat_view_static.py b/tests/test_gateway/test_chat_view_static.py
index a676130877..c6a23bc210 100644
--- a/tests/test_gateway/test_chat_view_static.py
+++ b/tests/test_gateway/test_chat_view_static.py
@@ -1,5 +1,6 @@
 """Focused Vue chat wiring contracts retained after the vanilla UI removal."""
 
+import re
 from pathlib import Path
 
 CHAT_VIEW = Path("opensquilla-webui/src/views/ChatView.vue")
@@ -34,7 +35,10 @@ def test_chat_view_wires_middle_edit_branch_fork_id() -> None:
     send_end = view.index("\n})", send_start)
     assert "pendingForkBeforeMessageId," in view[send_start:send_end]
 
-    assert "watch(sessionKey, () => {\n  pendingForkBeforeMessageId.value = null" in view
+    assert re.search(
+        r"watch\(sessionKey, \(\) => \{[^}]*\n  pendingForkBeforeMessageId\.value = null",
+        view,
+    )
 
     assert "pendingForkBeforeMessageId: Ref<string | null>" in send
     assert "params.forkBeforeMessageId = forkBeforeMessageId" in send
diff --git a/tests/test_gateway/test_sandbox_runtime_contract_registration.py b/tests/test_gateway/test_sandbox_runtime_contract_registration.py
index 24c26365b6..fe6c3f9ea0 100644
--- a/tests/test_gateway/test_sandbox_runtime_contract_registration.py
+++ b/tests/test_gateway/test_sandbox_runtime_contract_registration.py
@@ -159,7 +159,7 @@ def register(
 def test_production_registry_uses_contract_wrappers_without_surface_drift() -> None:
     registry = get_dispatcher()
 
-    assert len(registry.list_methods()) == 295
+    assert len(registry.list_methods()) == 296
     assert registry.get_entry("telemetry.product_active.record") is not None
     assert registry.get_entry("sessions.executionLog.read") is not None
     assert tuple(PRODUCTION_HANDLER_NAMES) == SANDBOX_RUNTIME_CONTRACT_METHODS
diff --git a/tests/test_gateway/test_skill_install_status.py b/tests/test_gateway/test_skill_install_status.py
new file mode 100644
index 0000000000..1b02da5873
--- /dev/null
+++ b/tests/test_gateway/test_skill_install_status.py
@@ -0,0 +1,85 @@
+from __future__ import annotations
+
+import asyncio
+from dataclasses import replace
+from uuid import uuid4
+
+import pytest
+
+from opensquilla.gateway.auth import Principal
+from opensquilla.gateway.rpc import RpcContext, get_dispatcher
+from opensquilla.skills.hub.management import InstallResult, SkillManagementService
+from opensquilla.skills.hub.router import SourceRouter
+from opensquilla.skills.loader import SkillLoader
+
+
+@pytest.mark.asyncio
+async def test_status_and_cancel_follow_identity_across_connections(tmp_path) -> None:
+    entered, release = asyncio.Event(), asyncio.Event()
+
+    class Service(SkillManagementService):
+        async def install(self, *args, **kwargs):
+            entered.set()
+            await release.wait()
+            return InstallResult(success=True, installed=True, name="demo")
+
+    service = Service(
+        router=SourceRouter([]),
+        managed_dir=tmp_path / "managed",
+        lockfile_path=tmp_path / "lock.json",
+        journal_path=tmp_path / "state/journal.json",
+    )
+    loader = SkillLoader(managed_dir=tmp_path / "managed", snapshot_path=tmp_path / "snapshot.json")
+    loader.load_all()
+    principal = Principal(
+        role="operator",
+        scopes=frozenset({"operator.admin"}),
+        is_owner=True,
+        authenticated=True,
+        token_public_id="synthetic-owner",
+    )
+    first = RpcContext(
+        conn_id="first", principal=principal, skill_loader=loader, skill_management_service=service
+    )
+    second = replace(first, conn_id="second")
+    other = replace(second, principal=replace(principal, token_public_id="different-owner"))
+    reader = replace(second, principal=replace(principal, scopes=frozenset({"operator.read"})))
+    operation_id = str(uuid4())
+    dispatcher = get_dispatcher()
+    wait = asyncio.create_task(
+        dispatcher.dispatch(
+            "install",
+            "skills.install",
+            {
+                "identifier": "demo",
+                "operationId": operation_id,
+            },
+            first,
+        )
+    )
+    await entered.wait()
+    wait.cancel()
+    with pytest.raises(asyncio.CancelledError):
+        await wait
+    result = await dispatcher.dispatch(
+        "status", "skills.install.status", {"operationId": operation_id}, second
+    )
+    assert result.ok and result.payload["state"] == "running"
+    denied = await dispatcher.dispatch(
+        "status", "skills.install.status", {"operationId": operation_id}, reader
+    )
+    assert not denied.ok and denied.error.code == "UNAUTHORIZED"
+    unknown = await dispatcher.dispatch(
+        "status", "skills.install.status", {"operationId": operation_id}, other
+    )
+    assert unknown.ok and unknown.payload["state"] == "unknown"
+    release.set()
+    await asyncio.gather(*service.install_operations.tasks.values())
+    result = await dispatcher.dispatch(
+        "status", "skills.install.status", {"operationId": operation_id}, second
+    )
+    assert result.ok and result.payload["state"] == "succeeded"
+    cancelled = await dispatcher.dispatch(
+        "cancel", "skills.install.cancel", {"operationId": operation_id}, second
+    )
+    assert cancelled.ok and cancelled.payload["success"] and not cancelled.payload["cancelled"]
diff --git a/tests/test_gateway/test_turn_admission_composition.py b/tests/test_gateway/test_turn_admission_composition.py
index 845a8b320a..8ecda5c1df 100644
--- a/tests/test_gateway/test_turn_admission_composition.py
+++ b/tests/test_gateway/test_turn_admission_composition.py
@@ -148,7 +148,7 @@ def test_fresh_rpc_import_preserves_contract_entries_without_an_import_cycle() -
 assert rpc_chat._turn_admission_adapter_factory is rpc_sessions.build_gateway_turn_admission_adapter
 registry = get_dispatcher()
 # Includes the product-activity and session execution-log methods.
-assert len(registry.list_methods()) == 295
+assert len(registry.list_methods()) == 296
 assert registry.get_entry("telemetry.product_active.record") is not None
 assert registry.get_entry("sessions.executionLog.read") is not None
 for method in methods:
diff --git a/tests/test_skills/test_hub_install_operations.py b/tests/test_skills/test_hub_install_operations.py
new file mode 100644
index 0000000000..d3783bd6a9
--- /dev/null
+++ b/tests/test_skills/test_hub_install_operations.py
@@ -0,0 +1,445 @@
+from __future__ import annotations
+
+import asyncio
+import subprocess
+import sys
+import textwrap
+import threading
+import uuid
+from pathlib import Path
+
+import pytest
+
+from opensquilla.skills.hub import management as hub_management
+from opensquilla.skills.hub import transaction as hub_transaction
+from opensquilla.skills.hub.lockfile import Lockfile
+from opensquilla.skills.hub.operations import (
+    InstallOperations,
+    InstallOperationStore,
+    current_install_operation,
+    operation_database,
+)
+from opensquilla.skills.hub.transaction import (
+    SkillTransactionJournal,
+    managed_root_identity,
+    recover_pending_skill_transaction,
+    rollback_root,
+    staging_root,
+)
+from opensquilla.skills.loader import SkillLoader
+from tests.test_skills.test_hub_management_service import FakeImmutableSource, _service
+
+
+@pytest.mark.asyncio
+async def test_disconnected_waiter_does_not_cancel_or_replay_install(tmp_path: Path) -> None:
+    store = InstallOperationStore(tmp_path / "operations.db", root_id="root")
+    operations = InstallOperations(store)
+    started = asyncio.Event()
+    release = asyncio.Event()
+    calls = 0
+    operation_id = str(uuid.uuid4())
+
+    async def install():
+        nonlocal calls
+        calls += 1
+        started.set()
+        await release.wait()
+        return {"success": True, "installed": True, "effectiveFrom": "next_turn"}
+
+    waiter = asyncio.create_task(
+        operations.run("owner", operation_id, {"identifier": "demo"}, install)
+    )
+    await started.wait()
+    waiter.cancel()
+    with pytest.raises(asyncio.CancelledError):
+        await waiter
+    assert store.read("owner", operation_id)["state"] == "running"
+    assert store.read("another-owner", operation_id)["state"] == "unknown"
+    release.set()
+    result = await operations.run("owner", operation_id, {"identifier": "demo"}, install)
+    assert result["success"]
+    assert (await operations.run("owner", operation_id, {"identifier": "demo"}, install)) == result
+    assert calls == 1
+    assert store.read("owner", operation_id)["state"] == "succeeded"
+    with pytest.raises(ValueError, match="another install"):
+        await operations.run("owner", operation_id, {"identifier": "different"}, install)
+
+
+@pytest.mark.asyncio
+async def test_cancel_after_durable_commit_returns_success(tmp_path: Path) -> None:
+    store = InstallOperationStore(tmp_path / "operations.db", root_id="root")
+    operations = InstallOperations(store)
+    committed = asyncio.Event()
+    operation_id = str(uuid.uuid4())
+
+    async def install():
+        context = current_install_operation()
+        context.store.finish(context.owner, context.id, {"success": True, "installed": True})
+        committed.set()
+        await asyncio.Event().wait()
+
+    waiter = asyncio.create_task(operations.run("owner", operation_id, {}, install))
+    await committed.wait()
+    result = await operations.cancel("owner", operation_id)
+    assert result["success"]
+    assert not result["cancelled"]
+    assert (await waiter)["success"]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("cancelled", [False, True], ids=["failure", "cancellation"])
+@pytest.mark.parametrize("rollback_fails", [False, True], ids=["rolled-back", "needs-recovery"])
+async def test_postflight_settlement_preserves_result_and_recovery_state(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, cancelled: bool, rollback_fails: bool,
+) -> None:
+    managed = tmp_path / "managed"
+    loader = SkillLoader(managed_dir=managed, snapshot_path=tmp_path / "snapshot.json")
+    loader.reload(force=True, reason="test.install-operation-baseline")
+    source = FakeImmutableSource({
+        "SKILL.md": "---\nname: demo\ndescription: Synthetic install fixture.\n---\n# Demo\n",
+    })
+    service = _service(tmp_path, source, loader=loader)
+    operations = service.install_operations
+    operation_id = str(uuid.uuid4())
+    loop = asyncio.get_running_loop()
+    entered = asyncio.Event()
+    release = threading.Event()
+    finished = threading.Event()
+
+    def fail_postflight(*args, **kwargs):
+        loop.call_soon_threadsafe(entered.set)
+        try:
+            if not release.wait(timeout=5):
+                raise TimeoutError("test did not release postflight")
+            raise RuntimeError("synthetic postflight rejection")
+        finally:
+            finished.set()
+
+    monkeypatch.setattr(loader, "reload_verified", fail_postflight)
+    real_restore = hub_transaction._restore_lockfile
+    if rollback_fails:
+        def fail_rollback(*args, **kwargs):
+            raise OSError("synthetic rollback failure")
+
+        monkeypatch.setattr(hub_transaction, "_restore_lockfile", fail_rollback)
+
+    async def install():
+        return (await service.install("demo", "fake")).to_dict()
+
+    request = {"identifier": "demo", "source": "fake"}
+    waiter = asyncio.create_task(operations.run("owner", operation_id, request, install))
+    cancellation = None
+    try:
+        await asyncio.wait_for(entered.wait(), timeout=5)
+        assert (managed / "demo" / "SKILL.md").exists()
+        assert operations.store.read("owner", operation_id)["state"] == "running"
+        if cancelled:
+            cancellation = asyncio.create_task(operations.cancel("owner", operation_id))
+            await asyncio.sleep(0)
+            assert not cancellation.done()
+            assert not finished.is_set()
+    finally:
+        release.set()
+        result = await asyncio.wait_for(waiter, timeout=5)
+        cancel_result = await cancellation if cancellation else None
+
+    receipt = operations.store.read("owner", operation_id)
+    expected_state = (
+        "recovery_required" if rollback_fails else "cancelled" if cancelled else "failed"
+    )
+    assert receipt["state"] == expected_state
+    assert receipt["result"] == result
+    assert result["success"] is False
+    assert bool(result.get("cancelled")) is (cancelled and not rollback_fails)
+    assert bool(result.get("recoveryRequired")) is rollback_fails
+    assert result["rollbackPerformed"] is not rollback_fails
+    assert result["diagnostics"]
+    if cancel_result is not None:
+        assert cancel_result == {
+            **result, "cancelled": bool(result.get("cancelled")), "pending": False,
+        }
+    assert await operations.run("owner", operation_id, request, install) == result
+    assert source.resolve_calls == 1
+    assert finished.is_set()
+    assert loader.get_by_name("demo") is None
+    assert (managed / "demo").exists() is rollback_fails
+    assert (Lockfile.load(service.lockfile_path).get("demo") is not None) is rollback_fails
+    assert service.journal_path.exists() is rollback_fails
+    if rollback_fails:
+        monkeypatch.setattr(hub_transaction, "_restore_lockfile", real_restore)
+        recovered = recover_pending_skill_transaction(
+            managed_dir=managed, lockfile_path=service.lockfile_path,
+            journal_path=service.journal_path,
+        )
+        assert not any(item.blocking for item in recovered)
+        assert operations.store.read("owner", operation_id)["state"] == "failed"
+        assert not (managed / "demo").exists()
+        assert not service.journal_path.exists()
+
+
+@pytest.mark.asyncio
+async def test_committed_cleanup_warning_is_preserved_in_all_receipts(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    managed = tmp_path / "managed"
+    loader = SkillLoader(managed_dir=managed, snapshot_path=tmp_path / "snapshot.json")
+    source = FakeImmutableSource({
+        "SKILL.md": "---\nname: demo\ndescription: Synthetic install fixture.\n---\n# Demo\n",
+    })
+    service = _service(tmp_path, source, loader=loader)
+    operations = service.install_operations
+    operation_id = str(uuid.uuid4())
+    real_validate = hub_management.validate_transaction_journal_paths
+
+    def fail_cleanup(journal, **kwargs):
+        if journal.phase == "committed":
+            raise OSError("synthetic committed cleanup failure")
+        return real_validate(journal, **kwargs)
+
+    monkeypatch.setattr(hub_management, "validate_transaction_journal_paths", fail_cleanup)
+
+    async def install():
+        return (await service.install("demo", "fake")).to_dict()
+
+    result = await operations.run("owner", operation_id, {}, install)
+    assert result["success"] is True
+    assert any(item["code"] == "TRANSACTION_CLEANUP_PENDING" for item in result["diagnostics"])
+    assert operations.store.read("owner", operation_id)["result"] == result
+    assert await operations.run("owner", operation_id, {}, install) == result
+    assert (await operations.cancel("owner", operation_id))["success"] is True
+    assert (managed / "demo" / "SKILL.md").exists()
+    assert loader.get_by_name("demo") is not None
+    recovered = recover_pending_skill_transaction(
+        managed_dir=managed, lockfile_path=service.lockfile_path, journal_path=service.journal_path,
+    )
+    assert not any(item.blocking for item in recovered)
+    assert operations.store.read("owner", operation_id)["result"] == result
+
+
+@pytest.mark.parametrize("crash_point", ["before", "after"])
+def test_restart_retains_rollback_proof_around_journal_removal(
+    tmp_path: Path, crash_point: str,
+) -> None:
+    operation_id = str(uuid.uuid4())
+    worker = textwrap.dedent("""\
+        import asyncio
+        import os
+        import sys
+        from pathlib import Path
+
+        from opensquilla.skills.hub import transaction
+        from tests.test_skills.test_hub_management_service import FakeImmutableSource, _service
+
+        root = Path(sys.argv[1])
+        operation_id = sys.argv[2]
+        crash_point = sys.argv[3]
+        source = FakeImmutableSource({
+            "SKILL.md": "---\\nname: demo\\ndescription: Crash fixture.\\n---\\n# Demo\\n",
+        })
+        service = _service(root, source)
+
+        async def reject_postflight(**kwargs):
+            raise RuntimeError("synthetic postflight rejection")
+
+        service._reload_and_verify = reject_postflight
+        remove_journal = transaction.remove_transaction_journal
+
+        def crash_at_removal(path):
+            status = service.install_operations.store.read("owner", operation_id)
+            assert status["state"] == "running" and not status["terminal"]
+            assert "result" not in status
+            if crash_point == "after":
+                remove_journal(path)
+            os._exit(73)
+
+        transaction.remove_transaction_journal = crash_at_removal
+
+        async def install():
+            return (await service.install("demo", "fake")).to_dict()
+
+        asyncio.run(service.install_operations.run("owner", operation_id, {}, install))
+        raise AssertionError("worker missed its crash point")
+    """)
+    crashed = subprocess.run(
+        [sys.executable, "-c", worker, str(tmp_path), operation_id, crash_point],
+        capture_output=True, text=True, check=False, timeout=20,
+    )
+    assert crashed.returncode == 73, crashed.stderr
+    managed = tmp_path / "managed"
+    journal_path = tmp_path / "transaction.json"
+    lockfile_path = tmp_path / "skills-lock.json"
+    assert journal_path.exists() is (crash_point == "before")
+    assert not (managed / "demo").exists()
+    assert Lockfile.load(lockfile_path).get("demo") is None
+    restarted = InstallOperationStore(
+        operation_database(journal_path), root_id=managed_root_identity(managed),
+    )
+    receipt = restarted.read("owner", operation_id)
+    assert receipt["state"] == "failed"
+    assert receipt["result"]["rollbackPerformed"] is True
+    recovered = recover_pending_skill_transaction(
+        managed_dir=managed, lockfile_path=lockfile_path, journal_path=journal_path,
+    )
+    assert not any(item.blocking for item in recovered)
+    restarted.recover_orphans(journal_path)
+    assert restarted.read("owner", operation_id)["state"] == "failed"
+    assert restarted.read("owner", operation_id)["result"]["rollbackPerformed"] is True
+    assert not journal_path.exists()
+
+
+@pytest.mark.asyncio
+async def test_rollback_checkpoint_failure_keeps_journal_for_restart(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    source = FakeImmutableSource({
+        "SKILL.md": "---\nname: demo\ndescription: Crash fixture.\n---\n# Demo\n",
+    })
+    service = _service(tmp_path, source)
+    operations = service.install_operations
+    operation_id = str(uuid.uuid4())
+
+    async def fail_postflight(**kwargs):
+        raise RuntimeError("synthetic postflight rejection")
+
+    def fail_checkpoint(*args):
+        raise OSError("synthetic checkpoint failure")
+
+    monkeypatch.setattr(service, "_reload_and_verify", fail_postflight)
+    monkeypatch.setattr(operations.store, "checkpoint_rollback", fail_checkpoint)
+
+    async def install():
+        return (await service.install("demo", "fake")).to_dict()
+
+    result = await operations.run("owner", operation_id, {}, install)
+    assert result["recoveryRequired"] is True
+    assert operations.store.read("owner", operation_id)["state"] == "recovery_required"
+    assert service.journal_path.exists()
+    assert not (service.managed_dir / "demo").exists()
+    recovered = recover_pending_skill_transaction(
+        managed_dir=service.managed_dir, lockfile_path=service.lockfile_path,
+        journal_path=service.journal_path,
+    )
+    assert not any(item.blocking for item in recovered)
+    assert operations.store.read("owner", operation_id)["state"] == "failed"
+    assert not service.journal_path.exists()
+
+
+@pytest.mark.parametrize(
+    "phase", ["prepared", "old_moved", "new_moved", "lock_written", "committed"]
+)
+def test_restart_reconciles_operation_after_transaction_recovery(
+    tmp_path: Path, phase: str
+) -> None:
+    managed = tmp_path / "managed"
+    journal_path = tmp_path / "state" / "transaction.json"
+    lockfile = tmp_path / "lock.json"
+    operation_id = str(uuid.uuid4())
+    root_id = managed_root_identity(managed)
+    store = InstallOperationStore(
+        operation_database(journal_path), root_id=root_id, process_id="old"
+    )
+    store.create("owner", operation_id, "signature")
+    staging = staging_root(managed) / "tx" / "demo"
+    rollback = rollback_root(managed) / "tx" / "demo"
+    staging.mkdir(parents=True)
+    (staging / "SKILL.md").write_text("synthetic fixture")
+    journal = SkillTransactionJournal.prepare(
+        operation="install",
+        managed_dir=managed,
+        name="demo",
+        target=managed / "demo",
+        staging=staging,
+        rollback=rollback,
+        lockfile_path=lockfile,
+    )
+    journal.install_operation_id = operation_id
+    journal.install_operation_owner = "owner"
+    if phase == "committed":
+        journal.install_receipt = {"success": True, "installed": True, "name": "demo"}
+    journal.phase = phase
+    journal.write(journal_path)
+    result = recover_pending_skill_transaction(
+        managed_dir=managed,
+        lockfile_path=lockfile,
+        journal_path=journal_path,
+    )
+    assert not any(item.blocking for item in result)
+    restarted = InstallOperationStore(operation_database(journal_path), root_id=root_id)
+    receipt = restarted.read("owner", operation_id)
+    assert receipt["state"] == ("succeeded" if phase == "committed" else "failed")
+    assert receipt["result"]["success"] == (phase == "committed")
+    assert not journal_path.exists()
+
+
+def test_restart_without_proof_requires_recovery_and_retention_keeps_active(tmp_path: Path) -> None:
+    path = tmp_path / "operations.db"
+    old = InstallOperationStore(path, root_id="root", process_id="old")
+    unknown_id = str(uuid.uuid4())
+    old.create("owner", unknown_id, "signature")
+    current = InstallOperationStore(path, root_id="root")
+    assert current.read("owner", unknown_id)["state"] == "recovery_required"
+    active_id = str(uuid.uuid4())
+    current.create("owner", active_id, "signature")
+    with current.connect() as db:
+        db.execute("UPDATE skill_installs SET updated=0 WHERE id=?", (active_id,))
+        for i in range(1005):
+            db.execute(
+                "INSERT INTO skill_installs VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
+                (
+                    "root",
+                    "owner",
+                    str(uuid.uuid4()),
+                    "sig",
+                    "old",
+                    "failed",
+                    "complete",
+                    "{}",
+                    '{"success":false}',
+                    0,
+                    i + 9999999999,
+                ),
+            )
+    current.prune()
+    assert current.read("owner", active_id)["state"] == "running"
+    with current.connect() as db:
+        assert (
+            db.execute("SELECT count(*) FROM skill_installs WHERE state!='running'").fetchone()[0]
+            == 1000
+        )
+
+
+@pytest.mark.asyncio
+async def test_cancel_before_install_reserves_a_terminal_receipt(tmp_path: Path) -> None:
+    store = InstallOperationStore(tmp_path / "operations.db", root_id="root")
+    operations = InstallOperations(store)
+    operation_id = str(uuid.uuid4())
+    result = await operations.cancel("owner", operation_id)
+    assert result["cancelled"]
+
+    async def forbidden():
+        raise AssertionError("cancelled operation must never begin installation")
+
+    result = await operations.run("owner", operation_id, {"identifier": "demo"}, forbidden)
+    assert result["cancelled"]
+
+
+@pytest.mark.asyncio
+async def test_expired_receipt_never_replays_the_same_operation_id(tmp_path: Path) -> None:
+    store = InstallOperationStore(tmp_path / "operations.db", root_id="root")
+    operations = InstallOperations(store)
+    operation_id = str(uuid.uuid4())
+    calls = 0
+
+    async def install():
+        nonlocal calls
+        calls += 1
+        return {"success": True}
+
+    await operations.run("owner", operation_id, {"identifier": "demo"}, install)
+    with store.connect() as db:
+        db.execute("UPDATE skill_installs SET updated=0 WHERE id=?", (operation_id,))
+    store.prune()
+    assert store.read("owner", operation_id)["state"] == "unknown"
+    with pytest.raises(ValueError, match="receipt expired"):
+        await operations.run("owner", operation_id, {"identifier": "demo"}, install)
+    assert calls == 1
diff --git a/tests/test_skills/test_staging_io_worker.py b/tests/test_skills/test_staging_io_worker.py
new file mode 100644
index 0000000000..981dae94ae
--- /dev/null
+++ b/tests/test_skills/test_staging_io_worker.py
@@ -0,0 +1,100 @@
+from __future__ import annotations
+
+import asyncio
+import threading
+import time
+import uuid
+from pathlib import Path
+
+import pytest
+
+from opensquilla.skills.hub.management import SkillManagementService
+from opensquilla.skills.hub.router import SourceRouter
+from opensquilla.skills.hub.scanner import ScanResult
+from opensquilla.skills.io_worker import check_staging_cancelled, run_staging_worker
+from tests.test_skills.test_hub_management_service import FakeImmutableSource
+
+
+@pytest.mark.asyncio
+async def test_scan_keeps_status_responsive_and_cancel_waits_for_cleanup(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    loop = asyncio.get_running_loop()
+    entered = asyncio.Event()
+    exited = threading.Event()
+    main_thread = threading.get_ident()
+
+    def scanning(directory: Path) -> ScanResult:
+        assert threading.get_ident() != main_thread
+        loop.call_soon_threadsafe(entered.set)
+        end = time.monotonic() + 5
+        try:
+            while time.monotonic() < end:
+                check_staging_cancelled()
+                time.sleep(0.002)
+            raise AssertionError("scan cancellation did not reach its worker")
+        finally:
+            exited.set()
+
+    monkeypatch.setattr("opensquilla.skills.hub.management.scan_skill_tree", scanning)
+    managed = tmp_path / "skills"
+    service = SkillManagementService(
+        router=SourceRouter([FakeImmutableSource({
+            "SKILL.md": "---\nname: demo\ndescription: A synthetic sample.\n---\n# Demo\n",
+        })]), managed_dir=managed,
+        lockfile_path=tmp_path / "lock.json", journal_path=tmp_path / "state" / "journal.json",
+    )
+    operations = service.install_operations
+    operation_id = str(uuid.uuid4())
+
+    async def install():
+        return (await service.install("demo", "fake")).to_dict()
+
+    task = asyncio.create_task(operations.run("owner", operation_id, {}, install))
+    await asyncio.wait_for(entered.wait(), timeout=3)
+    status = operations.store.read("owner", operation_id)
+    assert status["state"] == "running"
+    assert status["phase"] == "scanning"
+    assert not exited.is_set()
+    result = await operations.cancel("owner", operation_id)
+    assert result["cancelled"]
+    assert (await task)["cancelled"]
+    assert exited.is_set()
+    assert not (managed / "demo").exists()
+    assert not list((managed / ".opensquilla-staging").glob("*"))
+
+
+@pytest.mark.asyncio
+async def test_repeated_cancel_joins_worker_before_returning() -> None:
+    loop = asyncio.get_running_loop()
+    entered = asyncio.Event()
+    cleanup_started = asyncio.Event()
+    release_cleanup = threading.Event()
+    exited = threading.Event()
+
+    def working() -> None:
+        loop.call_soon_threadsafe(entered.set)
+        try:
+            for _ in range(2500):
+                check_staging_cancelled()
+                time.sleep(0.002)
+        finally:
+            loop.call_soon_threadsafe(cleanup_started.set)
+            release_cleanup.wait(timeout=3)
+            exited.set()
+
+    task = asyncio.create_task(run_staging_worker(working))
+    await entered.wait()
+    task.cancel("original cancellation")
+    await cleanup_started.wait()
+    task.cancel("repeated cancellation")
+    await asyncio.sleep(0)
+    assert not task.done()
+    assert not exited.is_set()
+    release_cleanup.set()
+    with pytest.raises(asyncio.CancelledError, match="original cancellation"):
+        await task
+    assert exited.is_set()
+    # Cancellation is scoped to that worker, never to subsequent ordinary reads.
+    check_staging_cancelled()
+    assert await run_staging_worker(lambda: "next operation") == "next operation"