diff --git a/CHANGELOG.md b/CHANGELOG.md
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,5 +1,17 @@
 # Changelog for clickhouse-migrate-test-support
 
+## 0.2.1.0 — 2026-11-14
+
+### Other Changes
+
+- Use Apple Container with exact digest/platform inspection, owned bind mounts
+  and inspected replica addresses.
+- Cover denied/own-user process observations, callback failure/cancellation,
+  exact interrupted receipts and active-work refusal under explicit external
+  authority.
+- Qualify bounded read-only inspection and the direct Linux arm64 26.8.28.14
+  LTS image.
+
 ## 0.2.0.0 — 2026-11-08
 
 ### New Features
diff --git a/clickhouse-migrate-test-support.cabal b/clickhouse-migrate-test-support.cabal
--- a/clickhouse-migrate-test-support.cabal
+++ b/clickhouse-migrate-test-support.cabal
@@ -1,6 +1,6 @@
 cabal-version: 3.12
 name: clickhouse-migrate-test-support
-version: 0.2.0.0
+version: 0.2.1.0
 synopsis: Isolated server fixtures for ClickHouse migration tests
 description:
   Isolated server fixtures for ClickHouse migration tests.
diff --git a/test/RepairBoundaries.hs b/test/RepairBoundaries.hs
--- a/test/RepairBoundaries.hs
+++ b/test/RepairBoundaries.hs
@@ -11,6 +11,7 @@
 import ClickHouseMigrate.TestSupport.Server
 import Control.Monad (forM_, when)
 import Data.Aeson (Value)
+import Data.IORef (modifyIORef', newIORef, readIORef)
 import Data.Text qualified as T
 import Data.Text.IO qualified as TIO
 import RunnerIntegration (assert, field, fixture, fromJson, planFor, provider, right)
@@ -82,7 +83,10 @@
     newPosition <- maybe (fail "Missing new position") pure currentPosition
     (newAttempt, newSteps) <- case current of [R.MigrationSummary _ _ (Just value) values] -> pure (value, values); _ -> fail "Missing new active attempt"
     assert "Authorization did not change attempt" (newAttempt /= attempt)
-    freshEvidence <- right (Repair.recoveryEvidence audit quiet receipt)
+    -- Exercise the explicit external policy across the real lost-authorization
+    -- receipts, with all original position/step/receipt checks still in force.
+    processObservations <- newIORef []
+    freshEvidence <- right (Repair.externalDrainRecoveryEvidence audit quiet receipt (\value -> modifyIORef' processObservations (<> [value])))
     fresh <- case action of
       "accept" -> do
         step <- selected newSteps
@@ -104,6 +108,8 @@
         pure ()
       else pure ()
     _ <- I.verify I.Observe bound (I.verifyOptions True) plan >>= right
+    observations <- readIORef processObservations
+    assert "Receipt recovery lost its explicit diagnostic distinction" (not (null observations) && all (== Repair.NoVisibleRecordedQueries) observations)
     assert "Recovery replayed completed steps or executed acceptance SQL" . (== 2) =<< count proxy
     putStrLn ("repair authorization " ++ T.unpack action ++ ": lost response=" ++ show lost ++ "; ordinary restart blocked; fresh request succeeds without replay")
   -- Acceptance and finalization have their own dependent append acknowledgements.
diff --git a/test/RepairIntegration.hs b/test/RepairIntegration.hs
--- a/test/RepairIntegration.hs
+++ b/test/RepairIntegration.hs
@@ -1,6 +1,7 @@
 module RepairIntegration (repairEarlyCases) where
 
 import ClickHouseMigrate
+import ClickHouseMigrate.CLI qualified as CLI
 import ClickHouseMigrate.Event qualified as Event
 import ClickHouseMigrate.Inspection qualified as Inspect
 import ClickHouseMigrate.Ledger qualified as Ledger
@@ -9,7 +10,8 @@
 import ClickHouseMigrate.Runner qualified as Run
 import ClickHouseMigrate.TestSupport.FaultProxy
 import ClickHouseMigrate.TestSupport.Server
-import Control.Monad (forM_, when)
+import Control.Exception (AsyncException (ThreadKilled), SomeAsyncException, SomeException, fromException, throwIO, try)
+import Control.Monad (forM_, void, when)
 import Data.Aeson (Value)
 import Data.IORef (modifyIORef', newIORef, readIORef)
 import Data.List.NonEmpty (NonEmpty (..))
@@ -61,6 +63,8 @@
 
 repairEarlyCases :: Server -> IO ()
 repairEarlyCases server = do
+  externalDrainCase server
+  externalOperationCases server
   oversizedRecovery server
   access <- provider
   audit <- right (Ledger.auditContext "runner fixture" "explicit first use" ["owned server; parent serializes whole job"])
@@ -197,3 +201,113 @@
   _ <- Inspect.verify Inspect.Observe bound (Inspect.verifyOptions True) plan >>= right
   assert "accept/finalize submitted application SQL" . (== 0) =<< applicationCount proxy
   putStrLn "recovery size: oversized acceptance preserves (3,3) events, position and attempt; combined audits reject authorization; replacement against original position accepts and finalizes without SQL"
+
+-- First real embedding producer/consumer: separate app-scoped credentials may
+-- not inspect processes, while the parent owns every earlier completed request.
+externalDrainCase :: Server -> IO ()
+externalDrainCase server = do
+  let label = "repair_external_drain"
+      username = "repair_external_user"
+  ledger <- newLedgerId
+  key <- right (targetKey ("runner-tests/" <> label))
+  access <- provider
+  audit <- right (Ledger.auditContext "owned external-drain fixture" "all predecessor calls joined under parent exclusion" ["owned fixture: no runtime, schedule or predecessor remains"])
+  quiet <- right (Ledger.attestQuiescence key audit)
+  plan <- planWithTail label
+  owner <- fixture server label (serverEndpoint server) FirstUse
+  let stop (Event.RunEvent phase _ _ _ step _ _) = when (phase == Event.BeforeSql && step == Just "insert") (fail "controlled before-SQL stop")
+  initial <- Run.up (Run.mutationContext owner access stop Event.redactedDiagnostics) (Run.upOptions audit (Just (Ledger.authorizeInitialization key ledger audit))) plan
+  report <- case initial of Left (R.RunFailure (R.CallbackFailed _) acknowledged _) -> pure acknowledged; _ -> fail "External fixture did not retain an uncertain intent"
+  (position, attempt, steps) <- checkpoint report
+  step <- selected steps
+  config <- scopedRepairFixture server label username ledger False
+  let context = Run.mutationContext config access Event.quietEvents Event.redactedDiagnostics
+  strict <- right (Repair.recoveryEvidence audit quiet []) >>= right . Repair.retryStepRequest position attempt step
+  rejected <- Repair.repair context plan strict
+  assert "Strict repair did not refuse denied process metadata" (case rejected of Left (R.RunFailure (R.LedgerRequestFailed _ _ _) _ _) -> True; _ -> False)
+  -- The observer is part of pre-authorization policy and cannot fail open.
+  throwing <- right (Repair.externalDrainRecoveryEvidence audit quiet [] (const (fail "SQL_SECRET_MARKER"))) >>= right . Repair.retryStepRequest position attempt step
+  failed <- Repair.repair context plan throwing
+  assert "External observer failure did not stop authorization" (case failed of Left (R.RunFailure (R.CallbackFailed (R.EventCallback "RepairProcessObservation")) _ _) -> not ("SQL_SECRET_MARKER" `Text.isInfixOf` Text.pack (show failed)); _ -> False)
+  cancelling <- right (Repair.externalDrainRecoveryEvidence audit quiet [] (const (throwIO ThreadKilled))) >>= right . Repair.retryStepRequest position attempt step
+  cancelled <- try @SomeException (Repair.repair context plan cancelling)
+  assert
+    "Observer cancellation became an ordinary repair result"
+    ( case cancelled of
+        Left exception -> case fromException exception :: Maybe SomeAsyncException of
+          Just _ -> True
+          _ -> False
+        Right _ -> False
+    )
+  observations <- newIORef []
+  let sink observation = modifyIORef' observations (<> [observation])
+      environment = CLI.commandEnvironment (Just config) (Just access) Event.quietEvents Event.redactedDiagnostics CLI.AdministrativeCommands
+      input = CLI.AuditInput "owned embedding" "independent predecessor drain retained" ["owned fixture: all calls joined; no other submitter exists"]
+      command = CLI.RepairCommand CLI.RetryStep (CLI.RepairInput key position attempt (Just step) input True [])
+  result <- CLI.handleCommandWithExternalDrain sink environment plan command
+  retried <- case result of CLI.CommandResult _ (Right (CLI.RunReport value)) -> pure value; _ -> fail (Text.unpack (Encoding.decodeUtf8 (CLI.renderJson result)))
+  assert "Denied observation was erased into an empty result" . (== [Repair.ProcessObservationUnavailable]) =<< readIORef observations
+  assert "External retry did not execute the exact selected step" . (== 1) =<< scalar server ("SELECT toUInt32(count()) AS value FROM default." <> label)
+  staleEvidence <- right (Repair.externalDrainRecoveryEvidence audit quiet [] sink)
+  staleRequest <- right (Repair.retryStepRequest position attempt step staleEvidence)
+  stale <- Repair.repair context plan staleRequest
+  assert "External authority bypassed a stale position" (case stale of Left (R.RunFailure (R.StalePosition _ _) _ _) -> True; _ -> False)
+  assert "Stale repair reached optional diagnostics" . (== 1) . length =<< readIORef observations
+  (after, nextAttempt, _) <- checkpoint retried
+  resumed <- right (Repair.resumeRequest after nextAttempt staleEvidence) >>= Repair.repair context plan >>= right
+  assert "External resume failed the exact pending suffix" (case resumed of R.RunReport _ _ _ _ _ [R.MigrationSummary _ R.Applied _ _, R.MigrationSummary _ R.Pending _ _] _ _ -> True; _ -> False)
+  assert "External resume replayed the selected insert" . (== 1) =<< scalar server ("SELECT toUInt32(count()) AS value FROM default." <> label)
+  putStrLn "external repair: actual denied metadata; strict refusal; callback failure/cancellation; CLI retry and SDK resume; stale-position refusal"
+
+scopedRepairFixture :: Server -> Text.Text -> Text.Text -> LedgerId -> Bool -> IO TargetConfig
+scopedRepairFixture server label username ledger ownProcesses = do
+  key <- right (targetKey ("runner-tests/" <> label))
+  void (queryServer server ("CREATE USER " <> username <> " IDENTIFIED WITH no_password"))
+  void (queryServer server ("GRANT SELECT,INSERT,ALTER ON default." <> label <> " TO " <> username))
+  void (queryServer server ("GRANT SELECT,INSERT ON " <> label <> "_history.* TO " <> username))
+  forM_ ["databases", "tables", "columns", "storage_policies", "disks"] $ \table ->
+    void (queryServer server ("GRANT SELECT ON system." <> table <> " TO " <> username))
+  rows <- queryServer server "SELECT toString(serverUUID()) AS id" >>= fromJson :: IO [Value]
+  identity <- case rows of [row] -> field "id" row; _ -> fail "Owned server identity missing"
+  config <- targetConfig key <$> right (directEndpoint (serverEndpoint server)) <*> right (serverId identity) <*> right (ledgerNamespace (label <> "_history")) <*> pure (ExistingLedger ledger) <*> right (basicCredentials username "") <*> pure defaultHttpLimits
+  when ownProcesses (void (queryServer server ("GRANT SELECT ON system.processes TO " <> username)))
+  pure config
+
+externalOperationCases :: Server -> IO ()
+externalOperationCases server = forM_ [(visible, action) | visible <- [False, True], action <- ["retry", "accept", "finalize"], visible || action /= "retry"] $ \(visible, action) -> do
+  let suffix = if visible then "own" else "denied"
+      label = "repair_external_" <> action <> "_" <> suffix
+      username = label <> "_user"
+  ledger <- newLedgerId
+  key <- right (targetKey ("runner-tests/" <> label))
+  access <- provider
+  audit <- right (Ledger.auditContext "scoped recovery fixture" "all previous native requests joined; parent retains target exclusion" ["owned serialized fixture: no unjoined client or schedule"])
+  quiet <- right (Ledger.attestQuiescence key audit)
+  plan <- planWithTail label
+  owner <- fixture server label (serverEndpoint server) FirstUse
+  let stop (Event.RunEvent phase _ _ _ step _ _) = when ((action == "finalize" && phase == Event.BeforeMigrationComplete) || (action /= "finalize" && phase == Event.BeforeSql && step == Just "insert")) (fail "controlled recovery boundary")
+  initial <- Run.up (Run.mutationContext owner access stop Event.redactedDiagnostics) (Run.upOptions audit (Just (Ledger.authorizeInitialization key ledger audit))) plan
+  report <- case initial of Left (R.RunFailure (R.CallbackFailed _) value _) -> pure value; _ -> fail "Scoped recovery interruption missing"
+  (position, attempt, steps) <- checkpoint report
+  when (action == "accept") (void (queryServer server ("INSERT INTO default." <> label <> " VALUES(7,'SQL_SECRET_MARKER')")))
+  config <- scopedRepairFixture server label username ledger visible
+  observations <- newIORef []
+  evidence <- right (Repair.externalDrainRecoveryEvidence audit quiet [] (\observation -> modifyIORef' observations (<> [observation])))
+  request <-
+    if action == "finalize"
+      then right (Repair.finalizeRequest position attempt evidence)
+      else do
+        step <- selected steps
+        right (if action == "accept" then Repair.acceptStepRequest position attempt step evidence else Repair.retryStepRequest position attempt step evidence)
+  let context = Run.mutationContext config access Event.quietEvents Event.redactedDiagnostics
+      expectedObservation = if visible then Repair.NoVisibleRecordedQueries else Repair.ProcessObservationUnavailable
+  repaired <- Repair.repair context plan request >>= right
+  assert "Scoped observation lost its visibility distinction" . (== [expectedObservation]) =<< readIORef observations
+  assert "Acceptance/finalization replayed or omitted actual effects" . (== 1) =<< scalar server ("SELECT toUInt32(count()) AS value FROM default." <> label)
+  when (action /= "finalize") $ do
+    assert "Selected recovery ran the remaining suffix" . (== 0) =<< scalar server ("SELECT toUInt32(count()) AS value FROM system.columns WHERE database='default' AND table='" <> label <> "' AND name='later'")
+    (after, next, _) <- checkpoint repaired
+    _ <- right (Repair.resumeRequest after next evidence) >>= Repair.repair context plan >>= right
+    assert "Scoped resume did not execute the pending suffix" . (== 1) =<< scalar server ("SELECT toUInt32(count()) AS value FROM system.columns WHERE database='default' AND table='" <> label <> "' AND name='later'")
+  assert "Scoped recovery started another migration" . (== 1) =<< scalar server ("SELECT toUInt32(count()) AS value FROM default." <> label)
+  putStrLn ("external repair: " ++ Text.unpack action ++ "; process visibility=" ++ Text.unpack suffix ++ "; exact step and pending suffix preserved")
diff --git a/test/RunnerIntegration.hs b/test/RunnerIntegration.hs
--- a/test/RunnerIntegration.hs
+++ b/test/RunnerIntegration.hs
@@ -134,6 +134,24 @@
     kinds <- mapM (field "stage") received :: IO [Maybe Text.Text]
     assert "Completed migration replayed" (length (filter (== Just "application-sql") kinds) == 2)
     _ <- Inspect.verify Inspect.Observe bound (Inspect.verifyOptions True) plan >>= right
+    -- A constrained reader rejects the legacy unlimited-result requests.
+    -- Explicit bounded observation succeeds; overflow cannot return a prefix.
+    let reader = "bounded_history_reader"
+        namespace = label <> "_history"
+    _ <- queryServer server ("CREATE USER " <> reader <> " IDENTIFIED BY 'bounded-test' SETTINGS readonly=2 CONST, max_result_rows=10000 MIN 1 MAX 10000, max_result_bytes=1048576 MIN 1 MAX 1048576")
+    _ <- queryServer server ("GRANT SELECT ON " <> namespace <> ".* TO " <> reader)
+    _ <- queryServer server ("GRANT SELECT ON system.disks TO " <> reader)
+    _ <- queryServer server ("GRANT SELECT ON system.storage_policies TO " <> reader)
+    sid <- case configuredTopology bound of DirectServer value -> pure value; _ -> fail "expected direct fixture"
+    readerConfig <- targetConfig (configuredTargetKey bound) <$> right (directEndpoint (serverEndpoint server)) <*> pure sid <*> right (ledgerNamespace namespace) <*> pure (ExistingLedger persisted) <*> right (basicCredentials reader "bounded-test") <*> pure defaultHttpLimits
+    legacy <- Inspect.verify Inspect.Observe readerConfig (Inspect.verifyOptions True) plan
+    assert "unbounded request unexpectedly admitted by constrained role" (case legacy of Left _ -> True; _ -> False)
+    bounded <- maybe (fail "invalid bounds") pure (Inspect.boundedObservation 10000 1048576)
+    _ <- Inspect.verify bounded readerConfig (Inspect.verifyOptions True) plan >>= right
+    tiny <- maybe (fail "invalid tiny bound") pure (Inspect.boundedObservation 1 1048576)
+    overflow <- Inspect.verify tiny readerConfig (Inspect.verifyOptions True) plan
+    assert "bounded ledger overflow returned false success" (case overflow of Left _ -> True; _ -> False)
+    putStrLn "bounded readonly inspection: constrained role accepted; overflowing history refuses without prefix success"
     changed <- planFor label 8
     blocked <- Run.up (Run.mutationContext bound access Event.quietEvents Event.redactedDiagnostics) (Run.upOptions audit Nothing) changed
     assert "Modified applied SQL accepted" (case blocked of Left (R.RunFailure (R.IncompatibleHistory _) _ _) -> True; _ -> False)
diff --git a/test/SurvivingWork.hs b/test/SurvivingWork.hs
--- a/test/SurvivingWork.hs
+++ b/test/SurvivingWork.hs
@@ -15,7 +15,7 @@
 import Control.Exception (SomeAsyncException, finally, fromException)
 import Control.Monad (forM_, void)
 import Data.Aeson (Value)
-import Data.IORef (atomicModifyIORef', newIORef)
+import Data.IORef (atomicModifyIORef', modifyIORef', newIORef, readIORef)
 import Data.List.NonEmpty (NonEmpty (..))
 import Data.Set qualified as Set
 import Data.Text qualified as T
@@ -122,6 +122,12 @@
             unsafe <- right (Repair.acceptStepRequest expected attempt selected falseEvidence)
             rejected <- Repair.repair (Run.mutationContext bound broken E.quietEvents E.redactedDiagnostics) plan unsafe
             assert "Recovery accepted a positively active predecessor" (case rejected of Left (R.RunFailure (R.PriorQueriesActive queries) _ _) -> query `elem` queries; _ -> False)
+            observations <- newIORef []
+            externalEvidence <- right (Repair.externalDrainRecoveryEvidence audit falseQuiet [] (\value -> modifyIORef' observations (<> [value])))
+            externalRequest <- right (Repair.acceptStepRequest expected attempt selected externalEvidence)
+            externallyRejected <- Repair.repair (Run.mutationContext bound broken E.quietEvents E.redactedDiagnostics) plan externalRequest
+            assert "External policy accepted a positively active predecessor" (case externallyRejected of Left (R.RunFailure (R.PriorQueriesActive queries) _ _) -> query `elem` queries; _ -> False)
+            assert "Active work was not retained as a positive diagnostic" . any (\case Repair.RecordedQueriesActive queries -> query `elem` queries; _ -> False) =<< readIORef observations
             stopClient
             assert "Server mutation did not survive client shutdown" =<< activeMutation
             blocked <- hostLocalExclusiveAccess root 1000000 Nothing >>= right
diff --git a/test/data/clickhouse-matrix.json b/test/data/clickhouse-matrix.json
--- a/test/data/clickhouse-matrix.json
+++ b/test/data/clickhouse-matrix.json
@@ -1,6 +1,6 @@
 {
   "matrixVersion": 1,
-  "checkedAt": "2026-11-07",
+  "checkedAt": "2026-11-14",
   "policy": "Only exact qualified patches and declared topology groups; no prereleases or managed-cloud certification",
   "entries": [
     {
@@ -153,6 +153,48 @@
         ],
         "evidence": "docs/release-evidence/2026-11-07-replicated.md"
       }
+    },
+    {
+      "id": "lts-linux-arm64",
+      "version": "26.8.28.14",
+      "tag": "v26.8.28.14-lts",
+      "commit": "3694fe07040369cef78c937311cbaac5f0555484",
+      "platform": "aarch64-linux",
+      "architecture": "arm64",
+      "image": "docker.io/clickhouse/clickhouse-server@sha256:597add6eeadf0b3c562b58c9a63d9712b9a0cde4e49b68e64a697dbf2b363b25",
+      "indexDigest": "sha256:63e02621435cde1146ab0fde66057091ee75b312d12917c0109a6a0c99bc7ce1",
+      "status": "native-qualified",
+      "groups": [
+        "transport",
+        "completion",
+        "ledger",
+        "exclusive",
+        "runner",
+        "repair",
+        "boundaries",
+        "survivors",
+        "adoption",
+        "consumer",
+        "cli",
+        "example",
+        "remediation"
+      ],
+      "scope": "Owned direct Atomic Linux arm64 server; no replicated, managed-cloud or Linux-client qualification",
+      "qualification": {
+        "date": "2026-11-14",
+        "cohort": "f4bc4e6d32decfaaf1a19e1033a22f3e28fc167c",
+        "compiler": "ghc-9.12.4",
+        "clientHost": "macOS arm64",
+        "requiredGroups": 13,
+        "artifactVariants": [
+          "working-tree Cabal binaries",
+          "seven extracted sibling source distributions",
+          "Nix installed suite, driver and example"
+        ],
+        "evidence": "docs/release-evidence/2026-11-14-lts-linux-arm64.md",
+        "scope": "Direct Atomic local synchronous effects on exact Linux arm64 OCI artifact; no replicated, managed-cloud or Linux-client certification"
+      },
+      "tagObject": "cb8b2f96f32feb4203ab9edbf4f03b274bbb6952"
     }
   ],
   "unqualifiedReleases": [
diff --git a/test/scripts/native_fixture.py b/test/scripts/native_fixture.py
--- a/test/scripts/native_fixture.py
+++ b/test/scripts/native_fixture.py
@@ -58,6 +58,18 @@
     return binary
 
 
+def verify_apple_image(entry):
+    """Require the exact local matrix digest and Linux architecture; never a tag."""
+    image = json.loads(subprocess.check_output(
+        ['container', 'image', 'inspect', entry['image']], text=True))[0]
+    digest = entry['image'].split('@', 1)[1]
+    if not any(v.get('digest') == digest and
+               v.get('platform', {}).get('architecture') == entry['architecture'] and
+               v.get('platform', {}).get('os') == 'linux'
+               for v in image.get('variants', [])):
+        raise RuntimeError('Apple Container candidate digest/platform mismatch; pre-pull the exact matrix digest')
+
+
 class NativeServer:
     def __init__(self, entry, deployment=None):
         self.entry = entry
@@ -65,18 +77,10 @@
         self.port = port()
         self.tcp_port = port()
         self.container = None
-        self.docker = None
+        self.runtime = None
         if 'image' in entry:
-            context = os.environ.get('CLICKHOUSE_MIGRATE_DOCKER_CONTEXT')
-            if not context:
-                raise RuntimeError('OCI candidate requires an explicit isolated Docker context')
-            self.docker = ['docker', '--context', context]
-            endpoint = json.loads(subprocess.check_output(self.docker + ['context', 'inspect', context], text=True))[0]['Endpoints']['docker']['Host']
-            if not endpoint.startswith('unix://'):
-                raise RuntimeError('Owned OCI fixture requires a local Unix Docker socket')
-            image = json.loads(subprocess.check_output(self.docker + ['image', 'inspect', entry['image']], text=True))[0]
-            if image['Architecture'] != entry['architecture'] or image['Os'] != 'linux':
-                raise RuntimeError('OCI candidate platform mismatch; pre-pull the exact matrix digest')
+            self.runtime = ['container']
+            verify_apple_image(entry)
             self.binary = None
         else:
             self.binary = binary_for(entry)
@@ -92,13 +96,13 @@
 <profile>default</profile><quota>default</quota><access_management>1</access_management></default></users>
 <quotas><default><interval><duration>3600</duration><queries>0</queries><errors>0</errors>
 <result_rows>0</result_rows><read_rows>0</read_rows><execution_time>0</execution_time></interval></default></quotas></clickhouse>''')
-        root = '/fixture' if self.docker else escape(str(self.root))
-        if self.docker:
+        root = '/fixture' if self.runtime else escape(str(self.root))
+        if self.runtime:
             users.write_text(users.read_text().replace('<ip>127.0.0.1</ip>', '<ip>::/0</ip>'))
-        users_path = '/fixture/users.xml' if self.docker else str(users)
-        listener = '0.0.0.0' if self.docker else '127.0.0.1'
-        http_port = 8123 if self.docker else self.port
-        tcp_port = 9000 if self.docker else self.tcp_port
+        users_path = '/fixture/users.xml' if self.runtime else str(users)
+        listener = '0.0.0.0' if self.runtime else '127.0.0.1'
+        http_port = 8123 if self.runtime else self.port
+        tcp_port = 9000 if self.runtime else self.tcp_port
         (self.root / 'config.xml').write_text(f'''<clickhouse>
 <logger><level>warning</level><log>{root}/server.log</log><errorlog>{root}/error.log</errorlog></logger>
 <path>{root}/data/</path><tmp_path>{root}/tmp/</tmp_path><user_files_path>{root}/user_files/</user_files_path>
@@ -118,47 +122,48 @@
             config = self.root / 'config.xml'
             config.write_text(config.read_text().replace('</clickhouse>', extra + '</clickhouse>'))
         try:
-            if self.docker:
+            if self.runtime:
                 name = deployment['name'] if deployment else 'clickhouse-migrate-' + uuid.uuid4().hex
-                subprocess.run(self.docker + ['create', '--name', name, '--label', 'clickhouse-migrate.fixture=owned',
+                self.container = name
+                subprocess.run(self.runtime + ['run', '--detach', '--name', name,
+                    '--label', 'clickhouse-migrate.fixture=owned',
                     *(['--network', deployment['network']] if deployment else []),
                     '--user', str(os.getuid()), '--memory', '2g', '--cpus', '2',
-                    '-p', '127.0.0.1::8123', '-p', '127.0.0.1::9000',
+                    '--platform', 'linux/' + entry['architecture'],
+                    '--volume', str(self.root) + ':/fixture',
+                    '-p', f'127.0.0.1:{self.port}:8123',
+                    '-p', f'127.0.0.1:{self.tcp_port}:9000',
                     '--entrypoint', '/bin/sh', entry['image'], '-c', 'exec sleep 2147483647'],
-                    check=True, capture_output=True, timeout=20)
-                self.container = name
-                subprocess.run(self.docker + ['cp', str(self.root) + '/.', name + ':/fixture'], check=True, capture_output=True, timeout=20)
-                # Docker allocates ports on its own host, not the client OS.
-                # Keep the namespace alive while restarting only ClickHouse so
-                # these exact bindings remain reserved for the fixture lifetime.
-                subprocess.run(self.docker + ['start', name], check=True, capture_output=True, timeout=20)
-                self.port = int(subprocess.check_output(self.docker + ['port', name, '8123/tcp'], text=True, timeout=5).strip().rsplit(':', 1)[1])
-                self.tcp_port = int(subprocess.check_output(self.docker + ['port', name, '9000/tcp'], text=True, timeout=5).strip().rsplit(':', 1)[1])
-            self.start()
-            deadline = time.monotonic() + 30
-            while time.monotonic() < deadline:
-                if self.process.poll() is not None:
-                    raise RuntimeError(f'Owned server exited before readiness: {self.process.returncode}')
-                try:
-                    with urllib.request.urlopen(self.endpoint + '/ping', timeout=.2) as response:
-                        if response.read() == b'Ok.\n':
-                            break
-                except OSError:
-                    pass
-                time.sleep(.05)
-            else:
-                raise RuntimeError('Owned server readiness deadline expired')
-            identity = self.query('SELECT version() AS version, toString(serverUUID()) AS server_id')[0]
-            version = identity['version']
-            if identity['server_id'] != self.retained_uuid():
-                raise RuntimeError('Endpoint is not the owned server data directory')
-            if version != entry['version']:
-                raise RuntimeError('Candidate server version mismatch')
+                    check=True, capture_output=True, timeout=60)
+            if not (deployment and deployment.get('defer_start')):
+                self.ready()
         except BaseException:
             print(f"Owned server failure evidence: {self.root}", file=sys.stderr, flush=True)
             self.close(False)
             raise
 
+    def ready(self):
+        self.start()
+        deadline = time.monotonic() + 30
+        while time.monotonic() < deadline:
+            if self.process.poll() is not None:
+                raise RuntimeError(f'Owned server exited before readiness: {self.process.returncode}')
+            try:
+                with urllib.request.urlopen(self.endpoint + '/ping', timeout=.2) as response:
+                    if response.read() == b'Ok.\n':
+                        break
+            except OSError:
+                pass
+            time.sleep(.05)
+        else:
+            raise RuntimeError('Owned server readiness deadline expired')
+        identity = self.query('SELECT version() AS version, toString(serverUUID()) AS server_id')[0]
+        version = identity['version']
+        if identity['server_id'] != self.retained_uuid():
+            raise RuntimeError('Endpoint is not the owned server data directory')
+        if version != self.entry['version']:
+            raise RuntimeError('Candidate server version mismatch')
+
     @property
     def endpoint(self):
         return f'http://127.0.0.1:{self.port}'
@@ -169,25 +174,25 @@
             return [json.loads(line) for line in response.read().splitlines() if line.strip()]
 
     def retained_uuid(self):
-        if self.docker:
-            return subprocess.check_output(self.docker + ['exec', self.container, 'cat', '/fixture/data/uuid'], text=True, timeout=5).strip()
+        if self.runtime:
+            return subprocess.check_output(self.runtime + ['exec', self.container, 'cat', '/fixture/data/uuid'], text=True, timeout=5).strip()
         return (self.root / 'data/uuid').read_text().strip()
 
     def start(self):
-        command = self.docker + ['exec', self.container, '/bin/sh', '-c', 'echo $$ > /fixture/owned-server.pid; exec clickhouse server --config-file=/fixture/config.xml'] if self.docker else [str(self.binary), 'server', '--config-file', str(self.root / 'config.xml')]
+        command = self.runtime + ['exec', self.container, '/bin/sh', '-c', 'echo $$ > /fixture/owned-server.pid; exec clickhouse server --config-file=/fixture/config.xml'] if self.runtime else [str(self.binary), 'server', '--config-file', str(self.root / 'config.xml')]
         self.process = subprocess.Popen(command, stdout=self.log, stderr=subprocess.STDOUT, start_new_session=True)
 
     def stop(self, sig):
         if self.process is not None and self.process.poll() is None:
-            if self.docker:
-                subprocess.run(self.docker + ['exec', self.container, '/bin/sh', '-c', 'kill -s "$1" "$(cat /fixture/owned-server.pid)"', 'sh', signal.Signals(sig).name.removeprefix('SIG')], check=True, capture_output=True, timeout=10)
+            if self.runtime:
+                subprocess.run(self.runtime + ['exec', self.container, '/bin/sh', '-c', 'kill -s "$1" "$(cat /fixture/owned-server.pid)"', 'sh', signal.Signals(sig).name.removeprefix('SIG')], check=True, capture_output=True, timeout=10)
             else:
                 os.killpg(self.process.pid, sig)
             try:
                 self.process.wait(timeout=10)
             except subprocess.TimeoutExpired:
-                if self.docker:
-                    subprocess.run(self.docker + ['exec', self.container, '/bin/sh', '-c', 'kill -KILL "$(cat /fixture/owned-server.pid)"'], check=True, capture_output=True, timeout=10)
+                if self.runtime:
+                    subprocess.run(self.runtime + ['exec', self.container, '/bin/sh', '-c', 'kill -KILL "$(cat /fixture/owned-server.pid)"'], check=True, capture_output=True, timeout=10)
                 else:
                     os.killpg(self.process.pid, signal.SIGKILL)
                 self.process.wait(timeout=5)
@@ -216,13 +221,8 @@
             self.stop(signal.SIGTERM)
         finally:
             if self.container:
-                # Copy only this fixture's private data/logs before removing its
-                # recorded container. No ambient container or host volume is used.
-                if not success:
-                    evidence = self.root / 'container-evidence'
-                    evidence.mkdir()
-                    subprocess.run(self.docker + ['cp', self.container + ':/fixture/.', str(evidence)], capture_output=True, timeout=30, check=True)
-                subprocess.run(self.docker + ['rm', '--force', self.container], capture_output=True, timeout=15, check=True)
+                # The private bind mount retains failure evidence on the host.
+                subprocess.run(self.runtime + ['delete', '--force', self.container], capture_output=True, timeout=30, check=True)
             self.log.close()
         if success:
             shutil.rmtree(self.root)
diff --git a/test/scripts/replicated_fixture.py b/test/scripts/replicated_fixture.py
--- a/test/scripts/replicated_fixture.py
+++ b/test/scripts/replicated_fixture.py
@@ -1,5 +1,5 @@
 #!/usr/bin/env python3
-"""Owned Keeper plus two replicas; no shared server or Docker network."""
+"""Owned Keeper plus two replicas; no shared server or container network."""
 import base64
 import http.client
 import json
@@ -17,7 +17,7 @@
 import urllib.parse
 import uuid
 
-from native_fixture import NativeServer, PASSWORD
+from native_fixture import NativeServer, PASSWORD, port, verify_apple_image
 
 
 class RoundRobin:
@@ -70,23 +70,15 @@
             raise RuntimeError('Replicated fixture requires an OCI candidate')
         self.entry = entry
         self.root = Path(tempfile.mkdtemp(prefix='clickhouse-migrate-replicated-', dir='/tmp'))
-        context = os.environ.get('CLICKHOUSE_MIGRATE_DOCKER_CONTEXT')
-        if not context:
-            raise RuntimeError('Replicated fixture requires an explicit isolated Docker context')
-        self.docker = ['docker', '--context', context]
-        endpoint = json.loads(subprocess.check_output(self.docker + ['context', 'inspect', context], text=True))[0]['Endpoints']['docker']['Host']
-        if not endpoint.startswith('unix://'):
-            raise RuntimeError('Owned replicated fixture requires a local Unix Docker socket')
-        image = json.loads(subprocess.check_output(self.docker + ['image', 'inspect', entry['image']], text=True))[0]
-        if image['Architecture'] != entry['architecture'] or image['Os'] != 'linux':
-            raise RuntimeError('Replicated OCI candidate platform mismatch; pre-pull the exact matrix digest')
+        self.runtime = ['container']
+        verify_apple_image(entry)
         self.network = 'clickhouse-migrate-' + uuid.uuid4().hex
         self.keeper = None
         self.replicas = []
         self.proxy = None
         self.closed = False
         try:
-            subprocess.run(self.docker + ['network', 'create', '--label', 'clickhouse-migrate.fixture=owned', self.network], check=True, capture_output=True)
+            subprocess.run(self.runtime + ['network', 'create', '--label', 'clickhouse-migrate.fixture=owned', self.network], check=True, capture_output=True)
             keeper_name = self.network + '-keeper'
             self.keeper = keeper_name
             keeper_config = self.root / 'keeper.xml'
@@ -96,12 +88,15 @@
 <coordination_settings><operation_timeout_ms>10000</operation_timeout_ms><session_timeout_ms>10000</session_timeout_ms></coordination_settings>
 <raft_configuration><server><id>1</id><hostname>localhost</hostname><port>9234</port></server></raft_configuration>
 </keeper_server></clickhouse>''')
-            subprocess.run(self.docker + ['create', '--name', keeper_name, '--network', self.network, '--label', 'clickhouse-migrate.fixture=owned', '--user', str(os.getuid()), '-p', '127.0.0.1::9181', '--memory', '512m', '--entrypoint', '/bin/sh', entry['image'], '-c', 'exec sleep 2147483647'], check=True, capture_output=True)
-            subprocess.run(self.docker + ['cp', str(self.root) + '/.', keeper_name + ':/fixture'], check=True, capture_output=True)
-            subprocess.run(self.docker + ['start', keeper_name], check=True, capture_output=True)
+            keeper_port = port()
+            subprocess.run(self.runtime + ['run', '--detach', '--name', keeper_name,
+                '--network', self.network, '--label', 'clickhouse-migrate.fixture=owned',
+                '--user', str(os.getuid()), '-p', f'127.0.0.1:{keeper_port}:9181',
+                '--memory', '512m', '--platform', 'linux/' + entry['architecture'],
+                '--volume', str(self.root) + ':/fixture', '--entrypoint', '/bin/sh',
+                entry['image'], '-c', 'exec sleep 2147483647'], check=True, capture_output=True, timeout=60)
             self.keeper_log = (self.root / 'keeper.log').open('w')
-            self.keeper_process = subprocess.Popen(self.docker + ['exec', keeper_name, 'clickhouse', 'keeper', '--config-file=/fixture/keeper.xml'], stdout=self.keeper_log, stderr=subprocess.STDOUT)
-            keeper_port = int(subprocess.check_output(self.docker + ['port', keeper_name, '9181/tcp'], text=True).strip().rsplit(':', 1)[1])
+            self.keeper_process = subprocess.Popen(self.runtime + ['exec', keeper_name, 'clickhouse', 'keeper', '--config-file=/fixture/keeper.xml'], stdout=self.keeper_log, stderr=subprocess.STDOUT)
             deadline = time.monotonic() + 20
             while time.monotonic() < deadline:
                 if self.keeper_process.poll() is not None:
@@ -117,11 +112,20 @@
             else:
                 raise RuntimeError('Keeper readiness deadline expired')
             names = [self.network + '-r1', self.network + '-r2']
-            cluster = '<remote_servers><fixture><shard>' + ''.join('<replica><host>' + name + '</host><port>9000</port><user>default</user><password>' + PASSWORD + '</password></replica>' for name in names) + '</shard></fixture></remote_servers>'
-            for i, name in enumerate(names):
-                extra = cluster + '<zookeeper><node><host>' + keeper_name + '</host><port>9181</port></node></zookeeper><macros><cluster>fixture</cluster><shard>1</shard><replica>r' + str(i + 1) + '</replica></macros><interserver_http_port>9009</interserver_http_port><interserver_http_host>' + name + '</interserver_http_host><distributed_ddl><path>/clickhouse/task_queue/ddl</path></distributed_ddl>'
-                self.replicas.append(NativeServer(entry, dict(name=name, network=self.network, config=extra)))
-                (self.root / 'replicas.json').write_text(json.dumps([dict(replica='r' + str(n + 1), root=str(replica.root), endpoint=replica.endpoint) for n, replica in enumerate(self.replicas)], indent=2) + '\n')
+            for name in names:
+                self.replicas.append(NativeServer(entry, dict(name=name, network=self.network, config='', defer_start=True)))
+            def address(name):
+                inspected = json.loads(subprocess.check_output(['container', 'inspect', name], text=True))[0]
+                return inspected['status']['networks'][0]['ipv4Address'].split('/')[0]
+            addresses = [address(name) for name in names]
+            keeper_address = address(keeper_name)
+            cluster = '<remote_servers><fixture><shard>' + ''.join('<replica><host>' + ip + '</host><port>9000</port><user>default</user><password>' + PASSWORD + '</password></replica>' for ip in addresses) + '</shard></fixture></remote_servers>'
+            for i, replica in enumerate(self.replicas):
+                extra = cluster + '<zookeeper><node><host>' + keeper_address + '</host><port>9181</port></node></zookeeper><macros><cluster>fixture</cluster><shard>1</shard><replica>r' + str(i + 1) + '</replica></macros><interserver_http_port>9009</interserver_http_port><interserver_http_host>' + addresses[i] + '</interserver_http_host><distributed_ddl><path>/clickhouse/task_queue/ddl</path></distributed_ddl>'
+                config = replica.root / 'config.xml'
+                config.write_text(config.read_text().replace('</clickhouse>', extra + '</clickhouse>'))
+                replica.ready()
+                (self.root / 'replicas.json').write_text(json.dumps([dict(replica='r' + str(n + 1), root=str(r.root), endpoint=r.endpoint) for n, r in enumerate(self.replicas)], indent=2) + '\n')
             # Administrative bootstrap exactly mirrors consumer infrastructure.
             for database in ['fixture_app', 'fixture_ledger']:
                 result = self.request(0, "CREATE DATABASE " + database + " ON CLUSTER fixture ENGINE=Replicated('/clickhouse/databases/{uuid}', '{shard}', '{replica}')", user='default')
@@ -182,13 +186,13 @@
         for replica in reversed(self.replicas):
             replica.close(success)
         if self.keeper:
-            logs = subprocess.run(self.docker + ['logs', self.keeper], capture_output=True, text=True)
+            logs = subprocess.run(self.runtime + ['logs', self.keeper], capture_output=True, text=True)
             (self.root / 'keeper-container.log').write_text(logs.stdout + logs.stderr)
-            subprocess.run(self.docker + ['rm', '-f', self.keeper], capture_output=True, timeout=15)
+            subprocess.run(self.runtime + ['delete', '--force', self.keeper], capture_output=True, timeout=15)
             if hasattr(self, 'keeper_process'):
                 self.keeper_process.wait(timeout=10)
                 self.keeper_log.close()
-        subprocess.run(self.docker + ['network', 'rm', self.network], capture_output=True, timeout=15)
+        subprocess.run(self.runtime + ['network', 'delete', self.network], capture_output=True, timeout=15)
         if success:
             shutil.rmtree(self.root)
         else: