diff --git a/README.md b/README.md
index ac678b5..30cd864 100644
--- a/README.md
+++ b/README.md
@@ -118,9 +118,10 @@ ORCH-027은 새 `tnorm_legacycore*`/`tnorm_legacycompute*` schema에 실제 pre-
 
 결과는 `.integration-data/trading-normalization/<runId>/evidence.json`의 `versionPin`에
 시나리오별 실행 상태와 비민감 Run/Job/Revision/ACK ID로 남긴다. 재시작 전 로그는 같은 실행
-디렉터리의 `before-migration-logs/`에 보존한다. ORCH-028은 Core의 admission/dispatch 전용
-중지 기능이 없으므로 `NOT_RUN_MISSING_CONTROL`로 기록한다. 이 항목이 있는 실행의
-스크립트 성공은 **실행 가능한 V1 pin 검증**의 성공이며 공동 전환 전체의 PASS가 아니다.
+디렉터리의 `before-migration-logs/`에 보존한다. ORCH-028은 두 Core 인스턴스의 admission/create
+중지, claim/send 경쟁, 재시작, 불명확 create 대조, terminal·삭제·취소·기존 ACK 재시도와
+격리 canary의 V1/V1 HTTP 검증 후 재개를 확인한다. 필요한 Core V10/control이 없는 조합은
+PASS가 아니다. 실행별 Core·Compute schema를 새로 만들며 실패 시 중지·보류 상태를 보존한다.
 V2 Core publication은 이번 범위에서 제외한다.
 정확한 assertion, 공동 전환에 필요한 후속 기능과 V2 선행 구현 상태는
 [결과 버전 pin E2E 문서](docs/TRADING_NORMALIZATION_RESULT_VERSION_PIN_E2E.md)를 참조한다.
diff --git a/docs/TRADING_NORMALIZATION_JOINT_CUTOVER_E2E.md b/docs/TRADING_NORMALIZATION_JOINT_CUTOVER_E2E.md
new file mode 100644
index 0000000..6f72edf
--- /dev/null
+++ b/docs/TRADING_NORMALIZATION_JOINT_CUTOVER_E2E.md
@@ -0,0 +1,112 @@
+# Trading Normalization Joint Cutover E2E
+
+Context ADR-063/065/066/070, `TRADING_VERSIONING.md`의 공동 전환 및
+`TR-ACC-IMPORT-ORCH-028`을 실제 HTTP/PostgreSQL 경로로 검증한다.
+Core 구현과 운영 권한·절차는 [Core PR #81](https://github.com/refinvest-dev/core-api/pull/81)의
+`operations/NORMALIZATION_CUTOVER.md`를 따른다. Context·Compute business code·Web은 변경하지 않는다.
+
+## Execution
+
+```powershell
+$env:PYTHONWARNINGS = 'ignore::DeprecationWarning'
+.\scripts\test-trading-normalization-result-version-pin.ps1 -CorePort 8680 -ComputePort 8681
+```
+
+Windows PowerShell에서 출력 리디렉션 시 jsonschema deprecation warning이 native error로 승격될 수
+있어 위 실행은 그 경고만 억제한다. 그 외 검사 오류는 실패로 종료한다. 기존 sibling checkout/
+환경 파일/timeout parameter를 그대로 받으며 Web 경로도 증적에 전달한다. 지정 포트 외에 두 번째
+Core용 CorePort+10, canary용 CorePort+20/ComputePort+20/+21을 사용한다.
+
+첫 검사는 `verify-contract-sync.ps1 -IncludeWeb`이다. loopback의 `refinvest_integration`만 허용하며
+외부 credential/production/vendor 데이터 없이 기존 비민감 합성 Binance CSV를 사용한다.
+실행마다 `tnorm_core<runId>`와 `tnorm_<runId>` schema를 만들고 실제 Core V1~V10 및 Compute
+migration을 적용한다. Core와 Compute source root를 복제하거나 서비스 business code를 재구현하지 않는다.
+
+두 Core는 동일한 Core schema와 DB 제어를 읽으며 Snowflake node는 0/1이다. canary는 node 2와
+별도 Core·Compute schema를 사용한다. 이전 실패의 paused/unknown dispatch를 자동으로 재개하지 않는다.
+실제 Core 실행 JAR을 한 번 빌드해 모든 instance/restart에 사용하고 JAR SHA-256을 증적과
+process identity에 기록한다. artifact가 바뀐 기존 process는 재사용하지 않는다.
+Windows worker의 fatal `10014 Bad address`만 registry별 최대 3회 같은 schema로 재기동하며
+실패 로그와 recovery event를 보존한다. 일반 오류나 create 요청은 이 경로에서 재전송하지 않는다.
+성공·실패 schema, 암호화 artifact와 로그는 조사용으로 보존하며 로컬 원본은 커밋하지 않는다.
+각 독립 정상 publication fixture는 새 Book을 사용한다. 동일 CSV를 이미 발행된 Book에 반복 import하여
+source overlap/conflict 시나리오와 혼합하지 않는다.
+
+## Assertion mapping
+
+`evidence.json → versionPin → scenarios → TR-ACC-IMPORT-ORCH-028 → assertions`에서 다음 항목을
+추적한다. 시작은 모두 NOT_RUN이고 실제 assertion을 완료한 항목만 PASS로 바꾼다. 실패한 run은
+FAIL이며 이후 미실행 항목은 NOT_RUN으로 남는다. 필수 control/조합이 없으면 실행 성공으로 처리하지 않는다.
+
+| Item | Assertion and evidence |
+|---|---|
+| 01-validation-and-retry-503 | 유효한 validation과 실제 retryable Session의 retry를 두 Core에 전송하여 503, TEMPORARY_DEPENDENCY_FAILURE, retryable=true, message/traceId 검사 |
+| 02-rejected-state-unchanged | 각 요청 전후 Session 전체 값, Run·dispatch·public idempotency record 전체 값과 각 table row 수 불변; 새 public key claim 0 |
+| 03-zero-create-while-paused | PAUSE commit 후 원래 fleet 프록시에서 create HTTP 요청 0; 별도 schema canary traffic은 분리하여 기록 |
+| 04-two-core-instances | 동일 DB/schema의 실제 Core process 두 개에서 같은 validation/retry 거절, scheduler와 DB fence 적용 |
+| 05-admission-and-send-races | 실제 create를 프록시에서 잠시 멈추고 다른 Core의 admission은 Session row lock에서 대기; PAUSE가 DB advisory lock에서 기다리는 것을 확인한 뒤 두 작업 해제; PAUSE commit 뒤 새 send 없음 |
+| 06-restart-and-unsent-preserved | 해당 합성 dispatch의 next_attempt_at만 연기해 미전송 준비; Core 재시작 후 paused/epoch와 dispatch 전체 값·attempt 0·key/pin 보존, 503 지속 |
+| 07-terminal-deletion-cancellation-recovery | 실제 PENDING Job 취소→CANCELLED/공개 실패, Session DELETE→410, COMPLETED Job terminal polling→RESULT_READY/staging, 새 Revision/PERSISTED 0, 기존 publication ACK 실패→중지 중 재시도 완료 |
+| 07 recovery | 실제 Compute가 접수한 응답을 유실시켜 UNKNOWN 보류; 기존 Compute idempotency key hash와 실제 HTTP status의 Job/dispatch/Run/token/generation/input/VersionSet/pin 대조 후 같은 Job으로 bind; create 1회, unresolved 상태 VERIFY 거절 |
+| 08-failed-http-keeps-paused | 별도 canary의 실제 V1 Job descriptor에 한정된 pin fault→Core RESULT_VERSION_PIN_MISMATCH·발행 차단; 원래 fleet paused/미검증 유지, RESUME 거절 |
+| 09-verified-resume-same-public-keys | 새 canary Book의 public upload/validation→실제 V1/V1 create·accepted/status/terminal/payload/replay와 독립 V1 hash·publication/ACK 확인; 그 증적 digest로 VERIFY/RESUME 뒤 중지 때 거절된 동일 public key의 validation/retry 202 |
+| 10-no-duplicate-run-job-revision-ack | 재개 후 각 Run의 Job/dispatch/Revision/ACK 하나, 같은 public key 반복 요청으로 추가 생성 없음; explicit retry는 의도한 이전 실패 Run과 새 Run 두 개로 lineage 유지 |
+
+미전송 준비의 DB scheduling delay와 scoped 프록시의 capacity/응답 유실/descriptor fault는 Integration
+test fault다. Session/Run/Job/result를 합성 INSERT하여 service 실행을 대신하지 않는다. cancellation,
+삭제, retryable failure, publication과 ACK는 실제 HTTP와 서비스 transaction으로 생성한다.
+canary의 정상 create는 실제 Core dispatcher가 만든 새 logical request이며 무필드 요청을 허용하거나
+알 수 없는 create에 새 identity를 발급하지 않는다. canary schema의 독립 HTTP 결과를 확인한 뒤에만
+원래 gate를 재개한다.
+
+## State and operational limits
+
+정본의 중지 경계는 PAUSE transaction commit이다. 이미 permit을 잡은 admission/send는 이 commit
+전에 끝날 수 있고 명령은 그 완료를 기다린다. create 응답 유실 또는 crash의 미결 요청은 자동
+재전송하지 않고 보존·대조한다. 중지 중 terminal 성공을 staging까지 처리하되 새 publication과
+PERSISTED outbox는 생성하지 않는다. 기존 commit의 ACK 재시도와 삭제 stale 처리는 계속된다.
+
+테스트는 배포된 두 candidate process의 동작을 검증한다. 실제 production fleet의 이전 binary,
+egress queue, 권한 부여와 배포 인벤토리는 검증하지 않으므로 production 배포 승인을 뜻하지 않는다.
+운영자는 gate 미지원 이전 instance와 무필드 producer 0을 runbook에 따라 별도로 확인해야 한다.
+V2 신규 선택 활성화와 V2 publication E2E는 후속 작업이다.
+
+## Migration runner
+
+ORCH-027의 기존 terminal Job 복제 schema에서 Compute 0017 atomic 실행을 별도 진단하고 항상
+rollback한 뒤 기존 psql statement runner를 실행한다. atomic 실행의 SQLSTATE 55006은 FAIL 및
+knownDeploymentLimitation으로 기록하며 해당 배포 방식의 PASS로 합산하지 않는다. 빈 canary schema의
+atomic migration 성공은 기존 terminal Job의 atomic backfill 성공을 뜻하지 않는다. Core V10/공동
+gate는 이 Compute SQL 문제에 의존하지 않지만, 실제 배포가 atomic runner를 사용하면 Compute
+소유 후속 해결·검증까지 전환을 보류해야 한다.
+
+## Evidence
+
+각 run의 `.integration-data/trading-normalization/<runId>/evidence.json`에는 다섯 저장소의 정확한 commit,
+시나리오별 PASS/FAIL/NOT_RUN, 비민감 Session/Run/Job/Revision/ACK ID와 row count를 남긴다.
+증적은 서명 URI/token/CSV 본문을 저장하지 않는다. Core 후보 commit의 검증과 main 조합의 검증은
+별도로 표시한다. 최종 실행 결과와 제한은 검증 완료 후 아래에 추가한다.
+
+
+## 2026-10-28 Results
+
+검증된 조합은 Core feature branch와 Context/Compute/Web main이다. 전체 main 조합 또는 production
+전환 승인으로 해석하지 않는다. 정확한 commit과 시나리오·ID·count는
+[비민감 compatibility evidence](evidence/NORMALIZATION_JOINT_CUTOVER_2026-10-28.json)에 기록했다.
+
+| Scope | Result |
+|---|---|
+| OpenAPI·Core 문서·Web snapshot 및 V1/V2 fixed vector | PASS |
+| Core 전체 Gradle test·ArchUnit·실제 Flyway V1~V10 | PASS |
+| Core PostgreSQL store 24개 | PASS, skipped 0 |
+| Integration unit 58개 / process identity 23 assertions | PASS / PASS |
+| V1 publication·독립 integrity·version pin / ORCH-024~027 | PASS / PASS |
+| ORCH-028 항목 01~10, 두 Core 및 격리 canary | PASS |
+| 기존 terminal Job의 Compute 0017 atomic migration | FAIL, SQLSTATE 55006; atomic 배포 BLOCKED |
+| 현재 전체 main의 ORCH-028 / production fleet·운영 role | NOT_RUN / NOT_RUN |
+| V2 신규 Run 선택 활성화 | NOT_RUN, 범위 제외 |
+
+이전 실패 run은 성공으로 재분류하지 않았다. 원본 DB/schema/log를 보존하고 위 JSON에 실패 원인과
+run ID를 분리했다. 기존 psql statement runner의 ORCH-027 PASS는 atomic runner의 배포 통과를 뜻하지
+않는다. 이전 Core instance·무필드 producer 0, production 권한 부여 및 atomic runner 해결은 운영
+배포 선행 조건이며 Core/Integration PR은 저장소 정책에 따라 사람의 검토를 기다린다.
diff --git a/docs/TRADING_NORMALIZATION_RESULT_VERSION_PIN_E2E.md b/docs/TRADING_NORMALIZATION_RESULT_VERSION_PIN_E2E.md
index 9e194ae..7f078e8 100644
--- a/docs/TRADING_NORMALIZATION_RESULT_VERSION_PIN_E2E.md
+++ b/docs/TRADING_NORMALIZATION_RESULT_VERSION_PIN_E2E.md
@@ -11,8 +11,9 @@
 
 기존 runtime/publication script와 같은 sibling 경로 parameter를 받는다.
 `verify-contract-sync.ps1 -IncludeWeb`을 프로세스·DB 변경 전에 실행한다.
-loopback PostgreSQL의 `refinvest_integration`만 허용하며 Core는 `public`, Compute는 실행별
-`tnorm_*` schema를 사용한다. V1 migration 복구 검증은 별도의 새 Core/Compute schema에서 수행한다.
+loopback PostgreSQL의 `refinvest_integration`만 허용하며 Core·Compute 모두 실행별
+`tnorm_*` schema를 사용한다. 두 번째 Core는 같은 Core schema를, canary는 별도 schema를 사용한다.
+V1 migration 복구 검증도 별도의 새 Core/Compute schema에서 수행한다.
 비민감 합성 Binance fixture만 업로드하고 정상 JWT·CSRF와 실제 HTTP/DB를 사용한다.
 
 기존 publication 시나리오의 transport 검증, 독립 hash 계산, 9개 section digest,
@@ -33,7 +34,7 @@ ACK가 먼저 성공할 수 있는 기존 테스트의 타이밍 경쟁을 피
 | ORCH-026 descriptor | 실제 Compute V1 성공 결과의 descriptor만 V2/V2로 주입; Core `PROCESSING_FAILED`/Run·dispatch `TERMINAL_FAILURE`, `RESULT_VERSION_PIN_MISMATCH`/`WAIT_FOR_RECOVERY`; Revision·ACK 0, 기존 latest 보존 |
 | ORCH-026 payload | descriptor pin은 V1/V1로 유지하고 payload pin만 V2/V2로 변경; descriptor 길이·transport SHA-256은 변경 bytes와 일치시켜 pin 검증에 도달; 같은 차단 결과와 ACK 미수신 확인 |
 | ORCH-027 | 새 pre-pin schema에 이번 합성 publication만 복제; pin 없는 저장 Run/dispatch/Job, 이전 logical key·fingerprint로 시작; 실제 Core V9·Compute 0017 SQL 적용; 저장 pin·request/idempotency fingerprint 일치; 양 API 재시작 후 V1 status/terminal/replay 및 public validation replay; 같은 Run·Job·Revision·ACK와 기존 hash 보존 |
-| ORCH-028 | **미실행: 필요한 Core/운영 제어 부재. 공동 전환 PASS가 아님** |
+| ORCH-028 | 두 Core의 503·row 불변·create 0, admission/send와 PAUSE commit 경쟁, 재시작·미전송 보존, terminal·삭제·취소·복구·기존 ACK, 격리 HTTP 실패 시 중지 유지·성공 후 같은 key 재개와 중복 0 |
 
 ORCH-026의 fault는 새 합성 Session ID에 한정하며 status/Job pin은 실제 V1/V1을 유지한다.
 원본 Job/Run pin을 수정하거나 서비스 코드에 test flag를 넣지 않는다. payload fault는 pin gate의
@@ -41,7 +42,7 @@ ORCH-026의 fault는 새 합성 Session ID에 한정하며 status/Job pin은 실
 
 ORCH-027은 **저장 완료된 V1 publication**의 합성 pre-pin 복구 검증이다. 실행 중인 원본 schema를
 downgrade하지 않고, 새 schema에 실제 pre-pin migration을 적용한 뒤 fixture를 채운다.
-미결 create·pending 작업의 전체 fleet drain과 운영 중지/재개는 ORCH-028의 미검증 범위로 남긴다.
+미결 create·pending 작업의 중지·대조·재개는 별도 ORCH-028 assertion으로 검증한다.
 
 Compute 0017은 기존 `initialize-compute-schema.ps1`과 같은 `psql -f`의 statement별
 autocommit 경계로 실행한다. 저장된 terminal Job이 있을 때 0017 전체를 한 transaction으로
@@ -51,7 +52,9 @@ SQLSTATE `55006`(`ObjectInUse`)으로 실패하는 것을 별도로 재현했다
 비활성화하지 않았다. 원자적 migration runner를 사용하는 배포는 Compute 소유 SQL/runner 후속 검증이
 필요하며, 이 E2E의 migration PASS를 해당 배포 방식의 PASS로 해석하지 않는다.
 
-## 공동 전환 후속 작업
+## 이전 main의 미구현 범위 (2026-10-27 기록)
+
+아래는 당시 기록이다. 현재 제어와 검증은 다음 절을 따른다.
 
 Core main `0ee1053449280237dffb2fdb26124bd5a974fcc5` 기준:
 
@@ -69,6 +72,11 @@ Core main `0ee1053449280237dffb2fdb26124bd5a974fcc5` 기준:
 - 이후 ORCH-028 E2E에는 validation·retry 각각 503과 상태/row 불변, 새 dispatch/create 0,
   기존 terminal·삭제·취소 처리 지속, 실패 시 중지 유지, 성공 뒤 같은 public key 재사용을 추가해야 한다.
 
+## Joint cutover implementation
+
+현재 Core 제어와 ORCH-028의 항목별 assertion, 실행 절차·배포 제한은
+[공동 전환 E2E 문서](TRADING_NORMALIZATION_JOINT_CUTOVER_E2E.md)를 따른다.
+
 ## V2 선행 구현 상태
 
 Compute main `0701c4ac07ac89281fa6f1b263bde294b0db6452`에는 V1/V2 쌍과 V2 producer·보충 근거
diff --git a/docs/evidence/NORMALIZATION_JOINT_CUTOVER_2026-10-28.json b/docs/evidence/NORMALIZATION_JOINT_CUTOVER_2026-10-28.json
new file mode 100644
index 0000000..f5146a7
--- /dev/null
+++ b/docs/evidence/NORMALIZATION_JOINT_CUTOVER_2026-10-28.json
@@ -0,0 +1,323 @@
+{
+  "scenario": "TR-ACC-IMPORT-ORCH-028",
+  "result": "PASS",
+  "combination": "candidate Core branch; not all-main",
+  "date": "2026-10-28",
+  "runId": "f54dc0d95d484b4db627d6a8534a6da3",
+  "sourceEvidence": ".integration-data/trading-normalization/f54dc0d95d484b4db627d6a8534a6da3",
+  "commits": {
+    "context": "06fdd057100917444aad36ff997e483f0f4f588b",
+    "core": "c956391a5e9e2f1af9a552f081b5281c1b4efcbe",
+    "compute": "0701c4ac07ac89281fa6f1b263bde294b0db6452",
+    "integration": "0017e37b17f696c730c2f638dfadec3fcbccbec9",
+    "web": "2ef516694dd45f79eba6ecfbb99f0a424f9aa4dc"
+  },
+  "coreArtifactSha256": "5ef1a475e8b753a03dc1809530d1a9d6d84f053de6525e8509e1fd1f738ed605",
+  "durationSeconds": 553.86,
+  "preflight": {
+    "contractSyncIncludeWeb": "PASS",
+    "coreContextDocSync": "PASS",
+    "V1V2FixedVectors": "PASS"
+  },
+  "coreTests": {
+    "fullGradleTest": "PASS",
+    "postgresStoreTests": 24,
+    "postgresSkipped": 0,
+    "flywayV1ThroughV10": "PASS",
+    "ArchUnit": "PASS"
+  },
+  "integrationChecks": {
+    "unittest": {
+      "result": "PASS",
+      "tests": 58
+    },
+    "processIdentity": {
+      "result": "PASS",
+      "assertions": 23
+    }
+  },
+  "scenarios": {
+    "V1_PIN_PUBLICATION": {
+      "ackCommitVisible": true,
+      "ackId": "7dea238a-e7e0-4729-b71e-ede20192d782",
+      "ackPayloadId": "6f07aef2-18aa-4a46-a4dc-065ab493ad9a",
+      "jobId": "093e523c-a3c9-430e-b3cc-b32ab3a7b3e2",
+      "pair": [
+        "TRADING_NORMALIZATION_RESULT_V1",
+        "TRADING_NORMALIZATION_RESULT_HASH_V1"
+      ],
+      "result": "PASS",
+      "revisionId": "1f1c6e00-30dc-4cee-9c50-06fbd17ee967",
+      "runId": "387b2037-06a9-4a9c-8e7d-faf07c4b2b8a"
+    },
+    "TR-ACC-IMPORT-ORCH-024": {
+      "exactReplay": 202,
+      "jobAndGrantsUnchanged": true,
+      "jobId": "093e523c-a3c9-430e-b3cc-b32ab3a7b3e2",
+      "otherPair": 409,
+      "result": "PASS"
+    },
+    "TR-ACC-IMPORT-ORCH-025": {
+      "cases": [
+        {
+          "case": "missing_both",
+          "code": "REQUEST_INVALID",
+          "status": 422
+        },
+        {
+          "case": "missing_schema",
+          "code": "REQUEST_INVALID",
+          "status": 422
+        },
+        {
+          "case": "missing_hash",
+          "code": "REQUEST_INVALID",
+          "status": 422
+        },
+        {
+          "case": "cross_v1_v2",
+          "code": "RESULT_VERSION_PAIR_INVALID",
+          "status": 422
+        },
+        {
+          "case": "cross_v2_v1",
+          "code": "RESULT_VERSION_PAIR_INVALID",
+          "status": 422
+        },
+        {
+          "case": "unsupported_schema",
+          "code": "RESULT_SCHEMA_UNSUPPORTED",
+          "status": 422
+        },
+        {
+          "case": "unsupported_hash",
+          "code": "RESULT_SCHEMA_UNSUPPORTED",
+          "status": 422
+        },
+        {
+          "case": "unsupported_both",
+          "code": "RESULT_SCHEMA_UNSUPPORTED",
+          "status": 422
+        }
+      ],
+      "jobsAndIdempotencyUnchanged": true,
+      "result": "PASS"
+    },
+    "TR-ACC-IMPORT-ORCH-026": {
+      "cases": [
+        "descriptor",
+        "payload"
+      ],
+      "result": "PASS"
+    },
+    "TR-ACC-IMPORT-ORCH-027": {
+      "atomicRunner": {
+        "knownDeploymentLimitation": true,
+        "result": "FAIL",
+        "rolledBackDiagnostic": true,
+        "sqlstate": "55006"
+      },
+      "beforeRestartLogDirectory": ".integration-data/trading-normalization/f54dc0d95d484b4db627d6a8534a6da3/before-migration-logs",
+      "computeMigration": "0017",
+      "computeMigrationRunner": "existing psql -f autocommit boundary",
+      "computeSchema": "tnorm_legacycomputef6c9fdbe7713",
+      "coreMigration": "V9",
+      "coreSchema": "tnorm_legacycoref6c9fdbe7713",
+      "explicitReplay": true,
+      "jobId": "093e523c-a3c9-430e-b3cc-b32ab3a7b3e2",
+      "migrationLogPath": ".integration-data/trading-normalization/f54dc0d95d484b4db627d6a8534a6da3/compute-pin-migration.err.log",
+      "result": "PASS",
+      "revisionAndAckUnchanged": true,
+      "revisionId": "1f1c6e00-30dc-4cee-9c50-06fbd17ee967",
+      "runId": "387b2037-06a9-4a9c-8e7d-faf07c4b2b8a",
+      "serviceRestart": true
+    },
+    "TR-ACC-IMPORT-ORCH-028": {
+      "assertions": {
+        "01-validation-and-retry-503": {
+          "result": "PASS"
+        },
+        "02-rejected-state-unchanged": {
+          "result": "PASS",
+          "rowCountsAndValuesUnchanged": true
+        },
+        "03-zero-create-while-paused": {
+          "canarySchemasExcluded": true,
+          "newCreateHttpRequests": 0,
+          "result": "PASS"
+        },
+        "04-two-core-instances": {
+          "coreInstances": 2,
+          "nodeIds": [
+            0,
+            1
+          ],
+          "result": "PASS"
+        },
+        "05-admission-and-send-races": {
+          "pauseWaitedForClaimSendReceiptAndAdmission": true,
+          "result": "PASS"
+        },
+        "06-restart-and-unsent-preserved": {
+          "epoch": 1,
+          "result": "PASS",
+          "unsentDispatchId": "76aff009-e19f-48fa-9617-7a91ee1b19ae"
+        },
+        "07-terminal-deletion-cancellation-recovery": {
+          "cancellation": true,
+          "deletion": true,
+          "existingAckRetry": true,
+          "newPersistedAcks": 0,
+          "result": "PASS",
+          "sameJobReconciled": "856828fc-7b68-496a-a75c-592fe1771e62",
+          "terminalPolling": true
+        },
+        "08-failed-http-keeps-paused": {
+          "httpFault": "RESULT_VERSION_PIN_MISMATCH",
+          "originalFleetPaused": true,
+          "result": "PASS"
+        },
+        "09-verified-resume-same-public-keys": {
+          "result": "PASS",
+          "samePublicKeysAccepted": true,
+          "verificationRef": "sha256/ba4894f233eae44433668fa7ec14f8daaaf66c583e712d368b421bbd5670e6bc"
+        },
+        "10-no-duplicate-run-job-revision-ack": {
+          "result": "PASS",
+          "sessions": [
+            {
+              "acks": 1,
+              "dispatches": 1,
+              "jobId": "1f0b35a0-56f2-4a5c-879e-ff1f34bdbedb",
+              "jobs": 1,
+              "revisionId": "85ca6bac-f737-4566-8e0f-e16cc1820a39",
+              "revisions": 1,
+              "runId": "004caefb-2d4c-4e87-a0be-df3823754bdd",
+              "runs": 1,
+              "sessionId": "373694570862477312"
+            },
+            {
+              "acks": 1,
+              "dispatches": 1,
+              "jobId": "03b16e92-e924-4a4a-916e-c9df7c5bb163",
+              "jobs": 1,
+              "revisionId": "3836f122-2a39-439f-a558-eb469585f36b",
+              "revisions": 1,
+              "runId": "77935ccc-edd3-438f-9e31-8ef87301e2e6",
+              "runs": 2,
+              "sessionId": "373694120079654912"
+            },
+            {
+              "acks": 1,
+              "dispatches": 1,
+              "jobId": "ac259e73-c150-4e56-baca-4007f549447e",
+              "jobs": 1,
+              "revisionId": "500e0bd5-fe11-4df2-89b1-ba43772ceac4",
+              "revisions": 1,
+              "runId": "7f0e0548-c55e-4e52-b219-99c17e942da1",
+              "runs": 1,
+              "sessionId": "373694571915247616"
+            },
+            {
+              "acks": 1,
+              "dispatches": 1,
+              "jobId": "856828fc-7b68-496a-a75c-592fe1771e62",
+              "jobs": 1,
+              "revisionId": "f4e56bf2-a00e-4379-91aa-cd9e21f51bc4",
+              "revisions": 1,
+              "runId": "7b75629b-5c61-4ddd-88d8-047b3bfa1f5d",
+              "runs": 1,
+              "sessionId": "373694573286785024"
+            },
+            {
+              "acks": 1,
+              "dispatches": 1,
+              "jobId": "cb4af1c9-8dc9-4e08-bf8c-feea85bcf352",
+              "jobs": 1,
+              "revisionId": "fc53d384-f916-401c-a14c-4233148c8e48",
+              "revisions": 1,
+              "runId": "4d96e424-146a-49a2-a323-f5a501a6f92c",
+              "runs": 1,
+              "sessionId": "373694594249916416"
+            }
+          ]
+        }
+      },
+      "canary": {
+        "commits": {
+          "compute": "0701c4ac07ac89281fa6f1b263bde294b0db6452",
+          "context": "06fdd057100917444aad36ff997e483f0f4f588b",
+          "core": "c956391a5e9e2f1af9a552f081b5281c1b4efcbe"
+        },
+        "computeSchema": "tnorm_canarycomputee32bff1996ef",
+        "coreSchema": "tnorm_canarycoree32bff1996ef",
+        "createAcceptedStatusTerminalReplay": "PASS",
+        "independentV1Integrity": "PASS",
+        "jobId": "e8214b46-e6d9-4321-b019-70c7942d5686",
+        "revisionId": "a548e106-2403-4766-ac9b-18de64f827fe",
+        "runId": "2b467235-b6b3-4d21-b2f6-05eb8a2b48cf"
+      },
+      "epoch": 1,
+      "implementationCoreCommit": "c956391a5e9e2f1af9a552f081b5281c1b4efcbe",
+      "mainCombination": false,
+      "result": "PASS",
+      "uncertainCreateReplayed": false
+    }
+  },
+  "atomicCompute0017": {
+    "knownDeploymentLimitation": true,
+    "result": "FAIL",
+    "rolledBackDiagnostic": true,
+    "sqlstate": "55006"
+  },
+  "deployment": {
+    "productionFleetInventory": "NOT_RUN",
+    "productionRoleGrants": "NOT_RUN",
+    "allMainORCH028": "NOT_RUN",
+    "atomicRunnerWithTerminalJobs": "BLOCKED",
+    "V2NewRunSelection": "NOT_RUN"
+  },
+  "workerRecoveryEvents": [
+    {
+      "attempt": 1,
+      "phase": "pin fault payload",
+      "preservedLogs": ".integration-data/trading-normalization/f54dc0d95d484b4db627d6a8534a6da3/worker-recoveries",
+      "reason": "WINDOWS_SOCKET_10014",
+      "registry": "trading-normalization-f54dc0d95d48"
+    },
+    {
+      "attempt": 2,
+      "phase": "existing ACK publication",
+      "preservedLogs": ".integration-data/trading-normalization/f54dc0d95d484b4db627d6a8534a6da3/worker-recoveries",
+      "reason": "WINDOWS_SOCKET_10014",
+      "registry": "trading-normalization-f54dc0d95d48"
+    }
+  ],
+  "failedAttempts": [
+    {
+      "run": "preflight-before-run-id",
+      "result": "FAIL",
+      "reason": "PowerShell promoted jsonschema deprecation warning; no service/DB mutation"
+    },
+    {
+      "runId": "224136b5285742948f6d2d6bc8b70e5b",
+      "result": "FAIL",
+      "reason": "harness referenced non-contract canRetry field; corrected to failure.retryable"
+    },
+    {
+      "runId": "a2d768d2b9b74dee8ecc80d39717d2d1",
+      "result": "FAIL",
+      "reason": "repeated source in a published Book caused RESULT_ITEM_CONFLICT; independent fixtures now use fresh Books"
+    },
+    {
+      "runId": "05761782dab3408d9e432d32e3edd35d",
+      "result": "FAIL",
+      "reason": "existing ACK retry budget exhausted during preparation; due-time fault now preserves pending ACK"
+    },
+    {
+      "runId": "aa5b00ec53fe4d5fa7dc6c873abea3a8",
+      "result": "FAIL",
+      "reason": "Compute worker fatal Windows socket 10014; owned-worker bounded recovery added"
+    }
+  ]
+}
diff --git a/scripts/IntegrationEnvironment.psm1 b/scripts/IntegrationEnvironment.psm1
index 263b118..9c1d9a5 100644
--- a/scripts/IntegrationEnvironment.psm1
+++ b/scripts/IntegrationEnvironment.psm1
@@ -221,7 +221,8 @@ function New-CoreIntegrationIdentity {
     param(
         [hashtable]$Values, [string]$EnvironmentFile, [string]$CoreApiPath,
         [string]$Profile = 'default', [string]$RegistryId = 'default',
-        [string]$DatabaseSchema = 'public', [string]$ComputeBaseUrl, [switch]$Observability
+        [string]$DatabaseSchema = 'public', [string]$ComputeBaseUrl, [switch]$Observability,
+        [string]$CoreJarPath
     )
     $corePort = if ($Values['INTEGRATION_CORE_PORT']) { [int]$Values['INTEGRATION_CORE_PORT'] } else { 8080 }
     $computePort = if ($Values['INTEGRATION_COMPUTE_PORT']) { [int]$Values['INTEGRATION_COMPUTE_PORT'] } else { 8000 }
@@ -239,6 +240,10 @@ function New-CoreIntegrationIdentity {
         observability = [bool]$Observability
         metricsEndpoint = if ($Observability) { "http://127.0.0.1:$corePort/actuator/prometheus" } else { $null }
     }
+    if ($CoreJarPath) {
+        $configuration['launchMode'] = 'executable-jar'
+        $configuration['artifactSha256'] = (Get-FileHash -LiteralPath $CoreJarPath -Algorithm SHA256).Hash.ToLowerInvariant()
+    }
     New-IntegrationServiceIdentity -ServiceRole 'core-api' -Profile $Profile -RegistryId $RegistryId `
         -RepositoryPath $CoreApiPath -Configuration $configuration
 }
diff --git a/scripts/initialize-core-trading-schema.ps1 b/scripts/initialize-core-trading-schema.ps1
index 72d2260..4c0f747 100644
--- a/scripts/initialize-core-trading-schema.ps1
+++ b/scripts/initialize-core-trading-schema.ps1
@@ -1,6 +1,7 @@
 [CmdletBinding()]
 param([Parameter(Mandatory)][string]$EnvironmentFile,
-      [Parameter(Mandatory)][string]$CorePath)
+      [Parameter(Mandatory)][string]$CorePath,
+      [string]$DatabaseSchema = 'public')
 $ErrorActionPreference = 'Stop'
 Import-Module (Join-Path $PSScriptRoot 'IntegrationEnvironment.psm1') -Force
 $values = Get-IntegrationEnvironment $EnvironmentFile
@@ -15,8 +16,15 @@ $args = @('-h',$hostName,'-p',$values.INTEGRATION_POSTGRES_PORT,'-U',$values.POS
           '-d',$values.POSTGRES_DB,'-v','ON_ERROR_STOP=1')
 $target = (& $psql @args -tA -c 'SELECT current_database();').Trim()
 if ($LASTEXITCODE -ne 0 -or $target -cne 'refinvest_integration') { throw 'Core migration target differs.' }
+if ($DatabaseSchema -notmatch '^(public|tnorm_[a-z0-9]+)$') { throw 'Invalid Core integration schema.' }
+& $psql @args -c "CREATE SCHEMA IF NOT EXISTS $DatabaseSchema;" | Out-Null
+if ($LASTEXITCODE -ne 0) { throw 'Core integration schema creation failed.' }
+$previousPGOptions = $env:PGOPTIONS
+try {
+$env:PGOPTIONS = "-c search_path=$DatabaseSchema"
 $migrationRoot = Join-Path $CorePath 'tradingreview\adapter\persistence\src\main\resources\db\migration'
-$migrations = @(Get-ChildItem -LiteralPath $migrationRoot -Filter 'V*__*.sql' | Sort-Object Name)
+$migrations = @(Get-ChildItem -LiteralPath $migrationRoot -Filter 'V*__*.sql' |
+    Sort-Object { [int]($_.Name -replace '^V([0-9]+)__.*$', '$1') })
 if ($migrations.Count -lt 4) { throw 'Core trading migration set is incomplete.' }
 & $psql @args -c 'CREATE TABLE IF NOT EXISTS integration_core_trading_migrations (filename TEXT PRIMARY KEY, applied_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP);' | Out-Null
 if ($LASTEXITCODE -ne 0) { throw 'Core migration tracking is unavailable.' }
@@ -39,3 +47,7 @@ foreach ($migration in $migrations) {
     if ($LASTEXITCODE -ne 0) { throw "Could not record Core migration $name." }
 }
 Write-Host 'Core trading migrations are ready in the local integration database.'
+} finally {
+    if ($null -eq $previousPGOptions) { Remove-Item Env:PGOPTIONS -ErrorAction SilentlyContinue }
+    else { $env:PGOPTIONS = $previousPGOptions }
+}
diff --git a/scripts/seed-core-member.ps1 b/scripts/seed-core-member.ps1
index 3b92aee..57bbe22 100644
--- a/scripts/seed-core-member.ps1
+++ b/scripts/seed-core-member.ps1
@@ -2,6 +2,7 @@
 param(
     [string]$EnvironmentFile = (Join-Path $PSScriptRoot "..\.env.local"),
     [string]$MemberId,
+    [string]$DatabaseSchema = 'public',
     [ValidateSet('FREE', 'PRO')]
     [string]$Tier = 'FREE'
 )
@@ -14,6 +15,10 @@ if ($memberId -notmatch '^\d+$') { throw 'Integration member ID must be a positi
 $env:INTEGRATION_MEMBER_ID = $memberId
 $sql = "INSERT INTO members (id, role, created_at) VALUES ($memberId, 'MEMBER', CURRENT_TIMESTAMP) ON CONFLICT (id) DO NOTHING; INSERT INTO subscriptions (member_id, tier) VALUES ($memberId, '$Tier') ON CONFLICT (member_id) DO UPDATE SET tier = EXCLUDED.tier;"
 $psql = Get-IntegrationPsql
+if ($DatabaseSchema -notmatch '^(public|tnorm_[a-z0-9]+)$' -or $values.POSTGRES_DB -cne 'refinvest_integration') {
+    throw 'Member seed requires a dedicated integration schema/database.'
+}
+$sql = "SET search_path TO $DatabaseSchema; " + $sql
 
 if ($psql) {
     $env:PGPASSWORD = $env:POSTGRES_PASSWORD
diff --git a/scripts/start-core-api.ps1 b/scripts/start-core-api.ps1
index 08fdc23..59a8159 100644
--- a/scripts/start-core-api.ps1
+++ b/scripts/start-core-api.ps1
@@ -4,6 +4,7 @@ param(
     [string]$CoreApiPath = (Join-Path $PSScriptRoot '..\..\api\refinvest'),
     [string]$ComputeBaseUrl,
     [string]$DatabaseSchema,
+    [string]$JarPath,
     [string]$Profile = 'default',
     [string]$RegistryId = 'default',
     [switch]$Observability
@@ -16,7 +17,7 @@ $values = Get-IntegrationEnvironment $EnvironmentFile
 $schemaName = if ([string]::IsNullOrWhiteSpace($DatabaseSchema)) { 'public' } else { $DatabaseSchema }
 $identity = New-CoreIntegrationIdentity -Values $values -EnvironmentFile $EnvironmentFile -CoreApiPath $CoreApiPath `
     -Profile $Profile -RegistryId $RegistryId -DatabaseSchema $schemaName `
-    -ComputeBaseUrl $ComputeBaseUrl -Observability:$Observability
+    -ComputeBaseUrl $ComputeBaseUrl -Observability:$Observability -CoreJarPath $JarPath
 $port = if ($values['INTEGRATION_CORE_PORT']) { [int]$values['INTEGRATION_CORE_PORT'] } else { 8080 }
 $allowed = Assert-IntegrationServiceStartAllowed -RootPath (Join-Path $PSScriptRoot '..') `
     -Name 'core-api' -ExpectedIdentity $identity -RegistryId $RegistryId -Port $port
@@ -67,10 +68,16 @@ $startParameters = @{
     RedirectStandardError = Join-Path $logDir "$RegistryId-core-api.err.log"
     PassThru = $true
 }
+if ($JarPath) {
+    if (-not (Test-Path -LiteralPath $JarPath -PathType Leaf) -or
+        [IO.Path]::GetExtension($JarPath) -cne '.jar') { throw 'Core executable JAR is unavailable.' }
+    $startParameters.FilePath = (Get-Command java -ErrorAction Stop).Source
+    $startParameters.ArgumentList = @('-jar', ('"' + (Resolve-Path -LiteralPath $JarPath).Path + '"'), "--server.port=$port")
+}
 $runningOnWindows = [System.Environment]::OSVersion.Platform -eq [System.PlatformID]::Win32NT
 if ($runningOnWindows) {
     $startParameters.WindowStyle = 'Hidden'
-} else {
+} elseif (-not $JarPath) {
     $startParameters.FilePath = (Get-Command bash -ErrorAction Stop).Source
     $startParameters.ArgumentList = @($gradleWrapper) + $gradleArguments
 }
diff --git a/scripts/start-trading-normalization-service.ps1 b/scripts/start-trading-normalization-service.ps1
index 73e40a9..285914d 100644
--- a/scripts/start-trading-normalization-service.ps1
+++ b/scripts/start-trading-normalization-service.ps1
@@ -16,6 +16,7 @@ if ($Service -eq 'core-api') {
     $uri = "http://127.0.0.1:$($values.INTEGRATION_CORE_PORT)/actuator/health"
     $arguments = @{ EnvironmentFile=$EnvironmentFile; CoreApiPath=$CorePath; ComputeBaseUrl=$ComputeBaseUrl
                     RegistryId=$RegistryId; Profile=$RegistryId }
+    if ($env:TRADING_RUNTIME_CORE_JAR) { $arguments.JarPath = $env:TRADING_RUNTIME_CORE_JAR }
     if ($DatabaseSchema) { $arguments.DatabaseSchema = $DatabaseSchema }
     $startScript = 'start-core-api.ps1'
 } else {
diff --git a/scripts/test-integration-process-identity.ps1 b/scripts/test-integration-process-identity.ps1
index 8d3739b..ad41678 100644
--- a/scripts/test-integration-process-identity.ps1
+++ b/scripts/test-integration-process-identity.ps1
@@ -119,6 +119,16 @@ try {
     foreach($secret in @('user-sentinel','password-sentinel','api-key-sentinel','jwt-sentinel','POSTGRES_PASSWORD','REFINVEST_INTERNAL_API_KEY')) {
         Assert-True ($serialized -notmatch [regex]::Escape($secret)) "metadata excludes $secret"
     }
+    $testJar = Join-Path $testRoot 'synthetic-core.jar'
+    [IO.File]::WriteAllBytes($testJar, [byte[]]@(1,2,3))
+    $jarIdentity = New-CoreIntegrationIdentity -Values $secretValues -EnvironmentFile (Join-Path $testRoot '.env.local') `
+        -CoreApiPath $integrationRoot -CoreJarPath $testJar -RegistryId 'jar-identity'
+    Register-IntegrationProcess -RootPath $testRoot -Name core-api -Process $current -Identity $jarIdentity -RegistryId 'jar-identity'
+    [IO.File]::WriteAllBytes($testJar, [byte[]]@(4,5,6))
+    $changedJarIdentity = New-CoreIntegrationIdentity -Values $secretValues -EnvironmentFile (Join-Path $testRoot '.env.local') `
+        -CoreApiPath $integrationRoot -CoreJarPath $testJar -RegistryId 'jar-identity'
+    Assert-ThrowsLike { Assert-IntegrationServiceStartAllowed -RootPath $testRoot -Name core-api `
+        -ExpectedIdentity $changedJarIdentity -RegistryId 'jar-identity' } '*MISMATCH*' 'changed executable artifact cannot reuse a process'
     "PASS: $passed process/configuration identity assertions"
 } finally {
     if ([IO.Path]::GetFullPath($testRoot).StartsWith([IO.Path]::GetFullPath((Join-Path $integrationRoot '.integration-data\tests')), [StringComparison]::OrdinalIgnoreCase)) {
diff --git a/scripts/test-trading-normalization-runtime-contract.ps1 b/scripts/test-trading-normalization-runtime-contract.ps1
index e930b43..b00dd8d 100644
--- a/scripts/test-trading-normalization-runtime-contract.ps1
+++ b/scripts/test-trading-normalization-runtime-contract.ps1
@@ -58,6 +58,7 @@ Write-Host "Database target verified: $dbHost`:$($values.INTEGRATION_POSTGRES_PO
 $runId = [guid]::NewGuid().ToString('N')
 $registry = "trading-normalization-$($runId.Substring(0,12))"
 $computeSchema = "tnorm_$($runId.Substring(0,12))"
+$coreSchema = if ($ResultVersionPin) { "tnorm_core$($runId.Substring(0,12))" } else { 'public' }
 $runDir = Join-Path $TestRoot ".integration-data\trading-normalization\$runId"
 New-Item -ItemType Directory -Force -Path $runDir | Out-Null
 $activeEnvironmentFile = $EnvironmentFile
@@ -154,7 +155,7 @@ try {
             $ErrorActionPreference = 'Continue'
             $appliedCoreMigrations = @(& $psql -h $dbHost -p $values.INTEGRATION_POSTGRES_PORT `
                 -U $values.POSTGRES_USER -d $DatabaseName -tA -v ON_ERROR_STOP=1 `
-                -c 'SELECT filename FROM integration_core_trading_migrations ORDER BY filename;' 2>$null)
+                -c "SELECT filename FROM $coreSchema.integration_core_trading_migrations ORDER BY filename;" 2>$null)
         } finally { $ErrorActionPreference = $previousErrorAction }
         if ($LASTEXITCODE -eq 0) { break }
         Start-Sleep -Milliseconds 250
@@ -162,7 +163,7 @@ try {
     if ($LASTEXITCODE -ne 0 -or
         (($expectedCoreMigrations -join ',') -cne (($appliedCoreMigrations | Sort-Object) -join ','))) {
         & (Join-Path $PSScriptRoot 'initialize-core-trading-schema.ps1') -EnvironmentFile $activeEnvironmentFile `
-            -CorePath $CorePath
+            -CorePath $CorePath -DatabaseSchema $coreSchema
         if ($LASTEXITCODE -ne 0) { throw 'Core trading migration failed.' }
     } else { Write-Host 'Core trading migrations are already applied in the integration database.' }
     if ($ResultVersionPin) {
@@ -175,6 +176,23 @@ try {
     }
     if ($LASTEXITCODE -ne 0) { throw 'Compute migration failed.' }
     Remove-Item Env:PGOPTIONS -ErrorAction SilentlyContinue
+    if ($ResultVersionPin) {
+        $env:GRADLE_USER_HOME = Join-Path $TestRoot '.integration-data\gradle'
+        $buildLog = Join-Path $runDir 'core-bootjar.log'
+        $previousDirectory = Get-Location
+        $previousErrorAction = $ErrorActionPreference
+        try {
+            Set-Location -LiteralPath $CorePath
+            $ErrorActionPreference = 'Continue'
+            & .\gradlew.bat :app:bootJar --console=plain --no-daemon *> $buildLog
+            $jarBuildExit = $LASTEXITCODE
+        } finally { Set-Location -LiteralPath $previousDirectory; $ErrorActionPreference = $previousErrorAction }
+        if ($jarBuildExit -ne 0) { throw 'Core executable JAR build failed; see core-bootjar.log.' }
+        $jars = @(Get-ChildItem -LiteralPath (Join-Path $CorePath 'app/build/libs') -Filter '*.jar' |
+            Where-Object { $_.Name -notmatch '-plain\.jar$' })
+        if ($jars.Count -ne 1) { throw 'Expected one Core executable JAR.' }
+        $env:TRADING_RUNTIME_CORE_JAR = $jars[0].FullName
+    }
     $scenario = 'startup'
     $started = $true
     if ($ResultVersionPin) {
@@ -183,7 +201,7 @@ try {
             -RegistryId $registry -EvidenceDirectory $runDir
         & (Join-Path $PSScriptRoot 'start-trading-normalization-service.ps1') -EnvironmentFile $activeEnvironmentFile `
             -CorePath $CorePath -RegistryId $registry -ComputeBaseUrl "http://127.0.0.1:$fixturePort" `
-            -EvidenceDirectory $runDir
+            -EvidenceDirectory $runDir -DatabaseSchema $coreSchema
     } else {
         & (Join-Path $PSScriptRoot 'start-compute-api.ps1') -EnvironmentFile $activeEnvironmentFile `
             -ComputeApiPath $ComputePath -Schema $computeSchema -Profile $registry -RegistryId $registry
@@ -217,7 +235,7 @@ try {
         }
     }
     $scenario = 'runtime'
-    $token = & (Join-Path $PSScriptRoot 'seed-core-member.ps1') -EnvironmentFile $activeEnvironmentFile -Tier PRO |
+    $token = & (Join-Path $PSScriptRoot 'seed-core-member.ps1') -EnvironmentFile $activeEnvironmentFile -Tier PRO -DatabaseSchema $coreSchema |
         Select-Object -Last 1
     if (-not $token) { throw 'Integration member token could not be minted.' }
     $env:TRADING_RUNTIME_ACCESS_TOKEN = $token
@@ -228,6 +246,7 @@ try {
     $env:TRADING_RUNTIME_DB_PASSWORD = $values.POSTGRES_PASSWORD
     $env:TRADING_RUNTIME_PSQL = $psql
     $env:TRADING_RUNTIME_COMPUTE_SCHEMA = $computeSchema
+    $env:TRADING_RUNTIME_CORE_SCHEMA = $coreSchema
     $env:TRADING_RUNTIME_API_KEY = $values.REFINVEST_INTERNAL_API_KEY
     $env:TRADING_RUNTIME_VERSION_SET = $versionJson
     $runtimeArguments = @()
@@ -238,11 +257,11 @@ try {
         --core-url "http://127.0.0.1:$CorePort" --compute-url "http://127.0.0.1:$ComputePort" `
         --fixture-root $fixtureRoot --context-path $ContextPath --run-dir $runDir `
         --fixture-port $fixturePort --test-root $TestRoot --environment-file $activeEnvironmentFile `
-        --core-path $CorePath --compute-path $ComputePath --compute-schema $computeSchema --registry $registry `
+        --core-path $CorePath --compute-path $ComputePath --web-path $WebPath --compute-schema $computeSchema --registry $registry `
         --timeout $TimeoutSeconds @runtimeArguments
     if ($LASTEXITCODE -ne 0) { throw 'Runtime contract scenario failed. See evidence and logs.' }
     if ($ResultVersionPin) {
-        Write-Host 'V1 result pin executable scenarios PASS; ORCH-028 NOT RUN (missing admission/dispatch pause control).'
+        Write-Host 'V1 result pin and ORCH-024~028 runtime assertions PASS.'
     }
     Write-Host "Trading Normalization runtime contract PASS. Evidence: $runDir"
 } catch {
@@ -251,7 +270,7 @@ try {
     Write-Host "Core log: $(Join-Path $TestRoot ".integration-data\logs\$registry-core-api.out.log")"
     Write-Host "Compute API log: $(Join-Path $TestRoot ".integration-data\logs\$registry-compute-api.out.log")"
     Write-Host "Compute worker log: $(Join-Path $TestRoot ".integration-data\logs\$registry-compute-worker.out.log")"
-    Write-Host "DB diagnosis: psql against verified $DatabaseName; inspect public.trading_normalization_runs and $computeSchema.trading_review_jobs by IDs in evidence."
+    Write-Host "DB diagnosis: psql against verified $DatabaseName; inspect $coreSchema.trading_normalization_runs and $computeSchema.trading_review_jobs by IDs in evidence."
     exit 1
 } finally {
     if ($started) {
@@ -261,4 +280,5 @@ try {
     Remove-Item Env:TRADING_RUNTIME_ACCESS_TOKEN -ErrorAction SilentlyContinue
     Remove-Item Env:TRADING_RUNTIME_DB_PASSWORD -ErrorAction SilentlyContinue
     Remove-Item Env:TRADING_RUNTIME_API_KEY -ErrorAction SilentlyContinue
+    Remove-Item Env:TRADING_RUNTIME_CORE_JAR -ErrorAction SilentlyContinue
 }
diff --git a/scripts/test_trading_normalization_runtime_contract.py b/scripts/test_trading_normalization_runtime_contract.py
index 61d08c4..a65e66e 100644
--- a/scripts/test_trading_normalization_runtime_contract.py
+++ b/scripts/test_trading_normalization_runtime_contract.py
@@ -81,6 +81,8 @@ def request(client: httpx.Client, method: str, url: str, expected: int, **kwargs
 
 
 def db_connection(schema: str) -> psycopg.Connection:
+    if schema == "public":
+        schema = os.environ.get("TRADING_RUNTIME_CORE_SCHEMA", "public")
     name = os.environ["TRADING_RUNTIME_DB_NAME"]
     require(name == "refinvest_integration", "UNSAFE_DATABASE_NAME")
     require(schema == "public" or
@@ -490,6 +492,7 @@ def main() -> int:
     parser.add_argument("--environment-file", type=Path, required=True)
     parser.add_argument("--core-path", type=Path, required=True)
     parser.add_argument("--compute-path", type=Path, required=True)
+    parser.add_argument("--web-path", type=Path)
     parser.add_argument("--compute-schema", required=True)
     parser.add_argument("--registry", required=True)
     parser.add_argument("--timeout", type=int, default=180)
@@ -508,7 +511,13 @@ def main() -> int:
         "computeCommit": git_revision(args.compute_path),
         "contractSync": "PASS", "fixedVector": "PASS", "scenarios": {},
         "computeSchema": os.environ["TRADING_RUNTIME_COMPUTE_SCHEMA"],
+        "coreSchema": os.environ.get("TRADING_RUNTIME_CORE_SCHEMA", "public"),
+        "integrationCommit": git_revision(args.test_root),
+        "webCommit": git_revision(args.web_path or args.test_root.parent / "web/web"),
     }
+    if os.environ.get("TRADING_RUNTIME_CORE_JAR"):
+        with Path(os.environ["TRADING_RUNTIME_CORE_JAR"]).open("rb") as jar:
+            evidence["coreArtifactSha256"] = hashlib.file_digest(jar, "sha256").hexdigest()
     if args.result_integrity:
         evidence["resultFixedVector"] = "PASS"
         evidence["runtimeIdentityGate"] = os.environ.get("TRADING_RUNTIME_IDENTITY_GATE", "NOT_RUN")
@@ -519,8 +528,7 @@ def main() -> int:
             "TR-ACC-IMPORT-ORCH-026", "TR-ACC-IMPORT-ORCH-026-descriptor", "TR-ACC-IMPORT-ORCH-026-payload",
             "TR-ACC-IMPORT-ORCH-027")}, "V2CorePublication": "OUT_OF_SCOPE"}
         evidence["versionPin"]["scenarios"]["TR-ACC-IMPORT-ORCH-028"] = {
-            "result": "NOT_RUN_MISSING_CONTROL",
-            "reason": "Core admission/dispatch-only pause boundary unavailable"}
+            "result": "NOT_RUN"}
     vector_paths = {
         "context": args.context_path / "fixtures/trading-normalization/input-identity-v1.json",
         "core": args.core_path / "tradingreview/application/src/test/resources/trading-normalization/input-identity-v1.json",
@@ -629,7 +637,13 @@ def main() -> int:
             require(set(original["inputIdentity"]) == set(identity), "SERIALIZER_IDENTITY_SHAPE_DRIFT")
             require("artifactAccessGrants" not in original["inputIdentity"] and
                     len(original["artifactAccessGrants"]) == 2, "GRANTS_IN_LOGICAL_IDENTITY")
-            status = poll(lambda: request(compute, "GET",
+            terminal_poll = poll
+            if args.result_version_pin:
+                from trading_normalization_cutover import worker_poll, WORKER_RECOVERIES
+                evidence["workerRecoveryEvents"] = WORKER_RECOVERIES
+                terminal_poll = lambda fetch, accept, timeout, label: worker_poll(
+                    sys.modules[__name__], args, fetch, accept, timeout, label)
+            status = terminal_poll(lambda: request(compute, "GET",
                 f"{args.compute_url}/trading-review/normalization-jobs/{job_id}", 200),
                 lambda s: s["status"] in {"COMPLETED", "FAILED", "CANCELLED"},
                 args.timeout, "Compute terminal")
diff --git a/scripts/trading_normalization_cutover.py b/scripts/trading_normalization_cutover.py
new file mode 100644
index 0000000..d4fe92a
--- /dev/null
+++ b/scripts/trading_normalization_cutover.py
@@ -0,0 +1,455 @@
+"""ORCH-028 against real Core instances, Compute HTTP and integration PostgreSQL.
+
+Only synthetic-session transport faults and scheduling delays are injected. No
+application implementation or result is replaced by a local business-code copy.
+"""
+from __future__ import annotations
+
+import copy
+import hashlib
+import json
+import os
+from pathlib import Path
+import shutil
+import subprocess
+import threading
+import time
+from types import SimpleNamespace
+import uuid
+from concurrent.futures import ThreadPoolExecutor
+
+import httpx
+import psycopg
+from psycopg import sql
+
+from trading_normalization_version_pin import PinProxy, V1, pair
+
+
+ITEMS = (
+    "01-validation-and-retry-503", "02-rejected-state-unchanged", "03-zero-create-while-paused",
+    "04-two-core-instances", "05-admission-and-send-races", "06-restart-and-unsent-preserved",
+    "07-terminal-deletion-cancellation-recovery", "08-failed-http-keeps-paused",
+    "09-verified-resume-same-public-keys", "10-no-duplicate-run-job-revision-ack",
+)
+WORKER_RECOVERIES = []
+
+
+def worker_poll(rt, args, fetch, accept, timeout, label, environment=None):
+    """Recover only the known host socket failure, retaining the failed worker log.
+
+    This starts the actual worker against the same durable schema; it never
+    changes a Job, lease, request identity, terminal result or retry budget.
+    """
+    def guarded_fetch():
+        stderr = args.test_root / ".integration-data/logs" / (args.registry + "-compute-worker.err.log")
+        failure = stderr.read_text(encoding="utf-8", errors="replace") if stderr.exists() else ""
+        if "Traceback" in failure and ("10014" in failure or "Bad address" in failure):
+            count = sum(e["registry"] == args.registry for e in WORKER_RECOVERIES)
+            rt.require(count < 3, "WORKER_SOCKET_RECOVERY_LIMIT")
+            archive = args.run_dir / "worker-recoveries"
+            archive.mkdir(exist_ok=True)
+            for suffix in ("out", "err"):
+                source = stderr.with_name(args.registry + f"-compute-worker.{suffix}.log")
+                if source.exists():
+                    shutil.copy2(source, archive / (args.registry + f"-{count+1}.{suffix}.log"))
+            service(rt, args, "stop-integration.ps1", "-RegistryId", args.registry, "-Service", "compute-worker")
+            service(rt, args, "start-compute-worker.ps1", "-EnvironmentFile", args.environment_file,
+                "-ComputeApiPath", args.compute_path, "-Schema", args.compute_schema, "-RegistryId", args.registry,
+                "-Profile", args.registry, "-TradingReview", environment=environment)
+            WORKER_RECOVERIES.append({"registry": args.registry, "reason": "WINDOWS_SOCKET_10014",
+                "attempt": count + 1, "phase": label, "preservedLogs": str(archive)})
+            (args.run_dir / "worker-recoveries.json").write_text(json.dumps(WORKER_RECOVERIES, indent=2), encoding="utf-8")
+        return fetch()
+    return rt.poll(guarded_fetch, accept, timeout, label)
+
+
+def service(rt, args, script, *arguments, environment=None):
+    result = subprocess.run(["powershell.exe", "-NoProfile", "-NonInteractive", "-File",
+        str(args.test_root / "scripts" / script), *map(str, arguments)], env=environment,
+        stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False)
+    rt.require(result.returncode == 0, "CUTOVER_PROCESS_CONTROL_FAILED")
+
+
+def environment_file(args, name, core_port, compute_port):
+    destination = args.run_dir / (name + ".env")
+    overrides = {"INTEGRATION_CORE_PORT": str(core_port), "INTEGRATION_COMPUTE_PORT": str(compute_port)}
+    lines = []
+    for line in args.environment_file.read_text(encoding="utf-8-sig").splitlines():
+        key = line.split("=", 1)[0].strip()
+        lines.append(key + "=" + overrides.pop(key) if key in overrides else line)
+    lines.extend(k + "=" + v for k, v in overrides.items())
+    destination.write_text("\n".join(lines) + "\n", encoding="utf-8")
+    return destination
+
+
+def new_session(rt, args, core, book, artifacts):
+    # Re-importing identical source into a published Book exercises overlap/conflict
+    # semantics. Each independent cutover fixture instead owns a fresh Book.
+    book = rt.request(core, "POST", f"{args.core_url}/trading-accounts/{book['tradingAccountId']}/books", 201,
+        headers={"Idempotency-Key": "cutover-book-" + uuid.uuid4().hex},
+        json={"displayName": "Synthetic cutover", "productFamily": "LINEAR_PERPETUAL",
+            "positionMode": "ONE_WAY", "settlementAsset": "USDT", "reviewTimezone": "UTC"})
+    session = rt.request(core, "POST", f"{args.core_url}/trading-books/{book['id']}/imports", 201,
+        headers={"Idempotency-Key": "cutover-import-" + uuid.uuid4().hex},
+        json={"sourceTimezone": "Asia/Seoul", "reviewTimezone": "UTC", "declaredPositionMode": "ONE_WAY"})
+    uploaded = {}
+    for role, artifact in artifacts.items():
+        uploaded[role] = rt.upload(core, args.core_url, session["id"], role,
+            args.fixture_root / artifact["path"], artifact["sha256"])
+    return session["id"], uploaded
+
+
+def validate(rt, args, core, sid, key=None, endpoint="validation", expected=202):
+    return rt.request(core, "POST", f"{args.core_url}/trading-import-sessions/{sid}/{endpoint}", expected,
+        headers={"Idempotency-Key": key or "cutover-validation-" + uuid.uuid4().hex}, json={})
+
+
+def state(rt, args, core, sid):
+    return rt.request(core, "GET", f"{args.core_url}/trading-import-sessions/{sid}", 200)
+
+
+def dispatch(rt, db, sid):
+    return rt.row(db, """select d.*,n.compute_job_id,n.attempt_token::text,n.input_fingerprint,
+        n.deletion_generation,n.target_version_set_json,n.import_session_id from
+        trading_normalization_dispatches d join trading_normalization_runs n on n.id=d.run_id
+        where n.import_session_id=%s order by d.created_at desc limit 1""", (int(sid),))
+
+
+def control(rt, db, command, epoch, reference):
+    return rt.row(db, "select trading_normalization_transition_control(%s,%s,%s) as epoch",
+        (command, epoch, reference))["epoch"]
+
+
+def current(rt, db):
+    return rt.row(db, "select paused,epoch,verified_evidence from trading_normalization_transition where singleton", ())
+
+
+def snapshot(rt, db, sid, keys):
+    return rt.row(db, """select
+        (select to_jsonb(s) from trading_import_sessions s where id=%s) as session,
+        (select jsonb_agg(to_jsonb(n) order by id) from trading_normalization_runs n where import_session_id=%s) as runs,
+        (select jsonb_agg(to_jsonb(d) order by d.id) from trading_normalization_dispatches d
+            join trading_normalization_runs n on n.id=d.run_id where n.import_session_id=%s) as dispatches,
+        (select jsonb_agg(to_jsonb(i) order by i.request_key) from trading_idempotency i where request_key=any(%s)) as idempotency,
+        (select count(*) from trading_import_sessions) as session_count,
+        (select count(*) from trading_normalization_runs) as run_count,
+        (select count(*) from trading_normalization_dispatches) as dispatch_count,
+        (select count(*) from trading_idempotency) as idempotency_count""", (int(sid), int(sid), int(sid), keys))
+
+
+def expect_control_rejected(rt, db, command, epoch, reference):
+    try:
+        control(rt, db, command, epoch, reference)
+    except psycopg.Error:
+        rt.require(current(rt, db)["paused"], "FAILED_VERIFICATION_OPENED_GATE")
+    else:
+        raise rt.ScenarioFailure("UNVERIFIED_CONTROL_ACCEPTED")
+
+
+def isolated_canary(rt, args, core, db, artifacts, checks):
+    """Run the candidate pair in separate schemas while the original fleet stays paused."""
+    suffix = uuid.uuid4().hex[:12]
+    core_schema, compute_schema = "tnorm_canarycore" + suffix, "tnorm_canarycompute" + suffix
+    core_port = int(args.core_url.rsplit(":", 1)[1]) + 20
+    compute_port = int(args.compute_url.rsplit(":", 1)[1]) + 20
+    canary_args = SimpleNamespace(**vars(args))
+    canary_args.core_url, canary_args.compute_url = f"http://127.0.0.1:{core_port}", f"http://127.0.0.1:{compute_port}"
+    canary_args.fixture_port = compute_port + 1
+    canary_args.compute_schema = compute_schema
+    registry = args.registry + "-canary"
+    env_file = environment_file(args, "canary", core_port, compute_port)
+    canary_args.registry, canary_args.environment_file = registry, env_file
+    owner = rt.row(db, "select owner_member_id from trading_books order by id desc limit 1", ())["owner_member_id"]
+    for name in (core_schema, compute_schema):
+        db.execute(sql.SQL("CREATE SCHEMA {}").format(sql.Identifier(name)))
+    with rt.db_connection(core_schema) as canary_db, rt.db_connection(compute_schema) as canary_compute_db:
+        for table, condition in (("members", "id"), ("subscriptions", "member_id")):
+            source_schema = os.environ.get("TRADING_RUNTIME_CORE_SCHEMA", "public")
+            canary_db.execute(sql.SQL("CREATE TABLE {} (LIKE {}.{} INCLUDING ALL)").format(
+                sql.Identifier(table), sql.Identifier(source_schema), sql.Identifier(table)))
+            canary_db.execute(sql.SQL("INSERT INTO {} SELECT * FROM {}.{} WHERE {}=%s").format(
+                sql.Identifier(table), sql.Identifier(source_schema), sql.Identifier(table), sql.Identifier(condition)), (owner,))
+        migration_root = args.core_path / "tradingreview/adapter/persistence/src/main/resources/db/migration"
+        for path in sorted(migration_root.glob("V*__*.sql"), key=lambda p: int(p.name.split("__")[0][1:])):
+            with canary_db.transaction():
+                canary_db.execute(path.read_text(encoding="utf-8"), prepare=False)
+        for path in sorted((args.compute_path / "migrations").glob("*.sql")):
+            with canary_compute_db.transaction():
+                canary_compute_db.execute(path.read_text(encoding="utf-8"), prepare=False)
+        environment = dict(os.environ)
+        config = json.loads(environment["SPRING_APPLICATION_JSON"])
+        config.update({"refinvest.trading-review.grant-base-url": canary_args.core_url,
+                       "refinvest.id.node-id": 2, "refinvest.compute.result-delay": "PT1S",
+                       "refinvest.compute.ack-delay": "PT1S"})
+        environment["SPRING_APPLICATION_JSON"] = json.dumps(config)
+        environment["REFINVEST_TRADING_REVIEW_ARTIFACT_HOSTS"] += f",127.0.0.1:{core_port}"
+        canary_proxy = PinProxy(rt, canary_args)
+        canary_proxy.core_schema, canary_proxy.reject_acks = core_schema, False
+        try:
+            service(rt, args, "start-trading-normalization-service.ps1", "-EnvironmentFile", env_file,
+                "-Service", "compute-api", "-ComputePath", args.compute_path, "-DatabaseSchema", compute_schema,
+                "-RegistryId", registry, "-EvidenceDirectory", args.run_dir, environment=environment)
+            service(rt, args, "start-trading-normalization-service.ps1", "-EnvironmentFile", env_file,
+                "-CorePath", args.core_path, "-DatabaseSchema", core_schema, "-RegistryId", registry,
+                "-EvidenceDirectory", args.run_dir, "-ComputeBaseUrl", f"http://127.0.0.1:{canary_args.fixture_port}", environment=environment)
+            service(rt, args, "start-compute-worker.ps1", "-EnvironmentFile", env_file,
+                "-ComputeApiPath", args.compute_path, "-Schema", compute_schema,
+                "-RegistryId", registry, "-Profile", registry, "-TradingReview", environment=environment)
+            account = rt.request(core, "POST", canary_args.core_url + "/trading-accounts", 201,
+                headers={"Idempotency-Key": "canary-account-" + suffix}, json={"venue": "BINANCE", "displayName": "Synthetic canary"})
+            book = rt.request(core, "POST", canary_args.core_url + f"/trading-accounts/{account['id']}/books", 201,
+                headers={"Idempotency-Key": "canary-book-" + suffix}, json={"displayName": "Synthetic canary",
+                    "productFamily": "LINEAR_PERPETUAL", "positionMode": "ONE_WAY", "settlementAsset": "USDT", "reviewTimezone": "UTC"})
+            # The negative canary is a real accepted V1 Job with a scoped transport fault.
+            sid, _ = new_session(rt, canary_args, core, book, artifacts)
+            canary_proxy.mode = ("descriptor", str(sid))
+            validate(rt, canary_args, core, sid)
+            failed = worker_poll(rt, canary_args, lambda: state(rt, canary_args, core, sid),
+                lambda s: s["status"] in {"PROCESSING_FAILED", "ACCEPTED"}, args.timeout, "negative canary", environment)
+            rt.require(failed["status"] == "PROCESSING_FAILED" and failed["failure"]["code"] == "RESULT_VERSION_PIN_MISMATCH",
+                "NEGATIVE_CANARY_NOT_REJECTED")
+            rt.require(current(rt, db)["paused"] and current(rt, db)["verified_evidence"] is None, "FAILED_CANARY_RESUMED")
+            expect_control_rejected(rt, db, "RESUME", current(rt, db)["epoch"], "failed-canary")
+            checks[ITEMS[7]] = {"result": "PASS", "httpFault": "RESULT_VERSION_PIN_MISMATCH", "originalFleetPaused": True}
+            sid, uploaded = new_session(rt, canary_args, core, book, artifacts)
+            key = "canary-validation-" + suffix
+            validate(rt, canary_args, core, sid, key)
+            item = rt.poll(lambda: dispatch(rt, canary_db, sid), lambda d: d and d["compute_job_id"], args.timeout, "canary create")
+            job_id, run_id = item["compute_job_id"], str(item["run_id"])
+            sent, accepted = canary_proxy.creates[run_id]
+            rt.require(pair(sent) == pair(accepted) == V1, "CANARY_CREATE_PAIR_DRIFT")
+            with httpx.Client(timeout=20, headers={"X-Internal-Api-Key": os.environ["TRADING_RUNTIME_API_KEY"]}) as compute:
+                terminal = worker_poll(rt, canary_args, lambda: rt.request(compute, "GET", canary_args.compute_url + f"/trading-review/normalization-jobs/{job_id}", 200),
+                    lambda s: s["status"] in {"COMPLETED", "FAILED"}, args.timeout, "canary terminal", environment)
+                rt.require(terminal["status"] == "COMPLETED" and pair(terminal) == V1, "CANARY_STATUS_FAILED")
+                descriptor = rt.request(compute, "GET", canary_args.compute_url + f"/trading-review/normalization-jobs/{job_id}/result", 200)
+                response = compute.get(canary_args.compute_url + "/trading-review/terminal-payloads/" + descriptor["payloadId"])
+                payload = rt.verify_transport(response.content, descriptor, response.headers.get("Content-Type", ""), response.headers.get("Content-Encoding", ""))
+                rt.verify(payload, descriptor, {r: a["id"] for r, a in uploaded.items()}, {r: a["sha256"] for r, a in artifacts.items()})
+                replay = rt.request(compute, "POST", canary_args.compute_url + "/trading-review/normalization-jobs", 202,
+                    headers={"Idempotency-Key": item["compute_key"]}, json=sent)
+                rt.require(replay == accepted and pair(descriptor) == pair(payload) == V1, "CANARY_REPLAY_DRIFT")
+            published = rt.poll(lambda: state(rt, canary_args, core, sid), lambda s: s["status"] == "ACCEPTED", args.timeout, "canary publication")
+            rt.poll(lambda: rt.row(canary_db, "select status from trading_normalization_ack_dispatches where run_id=%s::uuid", (run_id,)),
+                lambda a: a and a["status"] == "ACKNOWLEDGED", args.timeout, "canary ACK")
+            validate(rt, canary_args, core, sid, key)
+            rt.require(not canary_proxy.errors, "CANARY_PROXY_ERRORS")
+            return {"coreSchema": core_schema, "computeSchema": compute_schema, "runId": run_id, "jobId": job_id,
+                "revisionId": published["createdLedgerRevisionId"], "createAcceptedStatusTerminalReplay": "PASS",
+                "independentV1Integrity": "PASS", "commits": {"context": rt.git_revision(args.context_path),
+                    "core": rt.git_revision(args.core_path), "compute": rt.git_revision(args.compute_path)}}
+        finally:
+            for role in ("core-api", "compute-worker", "compute-api"):
+                service(rt, args, "stop-integration.ps1", "-RegistryId", registry, "-Service", role)
+            canary_proxy.close()
+
+
+def verify_cutover(rt, args, evidence, core, compute, db, compute_db, proxy, book, artifacts):
+    checks = {name: {"result": "NOT_RUN"} for name in ITEMS}
+    result = evidence["versionPin"]["cutover"] = {"assertions": checks, "result": "RUNNING"}
+    evidence["workerRecoveryEvents"] = WORKER_RECOVERIES
+    def checkpoint():
+        (args.run_dir / "evidence.json").write_text(json.dumps(evidence, indent=2, default=str) + "\n", encoding="utf-8")
+    checkpoint()
+    start = current(rt, db)
+    rt.require(not start["paused"], "CUTOVER_ALREADY_PAUSED_REQUIRES_RECOVERY")
+    secondary_registry = args.registry + "-second"
+    secondary_port = int(args.core_url.rsplit(":", 1)[1]) + 10
+    secondary_args = SimpleNamespace(**vars(args))
+    secondary_args.core_url = f"http://127.0.0.1:{secondary_port}"
+    secondary_env_file = environment_file(args, "second-core", secondary_port, int(args.compute_url.rsplit(":", 1)[1]))
+    secondary_environment = dict(os.environ)
+    config = json.loads(secondary_environment["SPRING_APPLICATION_JSON"])
+    config["refinvest.id.node-id"] = 1
+    secondary_environment["SPRING_APPLICATION_JSON"] = json.dumps(config)
+    executor = ThreadPoolExecutor(max_workers=3)
+    admission_lock = None
+    try:
+        # Make a genuinely retryable Session through the actual dispatcher retry budget.
+        retry_sid, _ = new_session(rt, args, core, book, artifacts)
+        proxy.reject_create_sessions.add(str(retry_sid))
+        validate(rt, args, core, retry_sid)
+        failed = rt.poll(lambda: state(rt, args, core, retry_sid),
+            lambda s: s["status"] == "PROCESSING_FAILED", args.timeout, "retryable Session")
+        rt.require(failed["failure"]["retryable"], "SESSION_NOT_RETRYABLE")
+        proxy.reject_create_sessions.remove(str(retry_sid))
+        ack_sid, _ = new_session(rt, args, core, book, artifacts)
+        proxy.reject_acks = True
+        validate(rt, args, core, ack_sid)
+        ack_public = worker_poll(rt, args, lambda: state(rt, args, core, ack_sid),
+            lambda s: s["status"] in {"ACCEPTED", "PROCESSING_FAILED", "REJECTED"}, args.timeout, "existing ACK publication")
+        rt.require(ack_public["status"] == "ACCEPTED", "EXISTING_ACK_PUBLICATION_FAILED")
+        ack_run = str(dispatch(rt, db, ack_sid)["run_id"])
+        rt.poll(lambda: rt.row(db, "select status,attempt_count from trading_normalization_ack_dispatches where run_id=%s::uuid", (ack_run,)),
+            lambda a: a and a["status"] == "PENDING" and a["attempt_count"] > 0, args.timeout, "existing ACK failure")
+        # Keep the real, already-failed ACK pending across process startup without
+        # exhausting its immutable retry budget. Only the synthetic due time changes.
+        db.execute("""update trading_normalization_ack_dispatches set next_attempt_at=clock_timestamp()+interval '20 minutes'
+            where run_id=%s::uuid and status='PENDING'""", (ack_run,))
+        # Leave queued Jobs pending to test actual cancellation and terminal drain after PAUSE.
+        service(rt, args, "stop-integration.ps1", "-RegistryId", args.registry, "-Service", "compute-worker")
+        service(rt, args, "start-trading-normalization-service.ps1", "-EnvironmentFile", secondary_env_file,
+            "-CorePath", args.core_path, "-RegistryId", secondary_registry, "-EvidenceDirectory", args.run_dir,
+            "-DatabaseSchema", os.environ.get("TRADING_RUNTIME_CORE_SCHEMA", "public"),
+            "-ComputeBaseUrl", f"http://127.0.0.1:{args.fixture_port}", environment=secondary_environment)
+        received_sid, _ = new_session(rt, args, core, book, artifacts)
+        unsent_sid, _ = new_session(rt, args, core, book, artifacts)
+        validate(rt, args, core, unsent_sid)
+        # Test-only scheduling delay, confined to this synthetic dispatch. No identity is edited.
+        db.execute("""update trading_normalization_dispatches set next_attempt_at=clock_timestamp()+interval '20 minutes'
+            where run_id=(select active_normalization_run_id from trading_import_sessions where id=%s)
+            and create_started_at is null and status='PENDING'""", (int(unsent_sid),))
+        unsent_before = dispatch(rt, db, unsent_sid)
+        rt.require(unsent_before["attempt_count"] == 0 and unsent_before["create_started_at"] is None, "UNSENT_FIXTURE_ALREADY_CLAIMED")
+        unknown_sid, _ = new_session(rt, args, core, book, artifacts)
+        proxy.drop_response_sessions.add(str(unknown_sid))
+        validate(rt, args, core, unknown_sid)
+        unknown = rt.poll(lambda: dispatch(rt, db, unknown_sid),
+            lambda d: d and d["last_failure_code"] == "CREATE_OUTCOME_UNKNOWN", args.timeout, "ambiguous create")
+        rt.require(unknown["compute_job_id"] is None, "UNKNOWN_CREATE_NOT_HELD")
+        cancel_sid, _ = new_session(rt, args, core, book, artifacts)
+        validate(rt, args, core, cancel_sid)
+        cancel = rt.poll(lambda: dispatch(rt, db, cancel_sid), lambda d: d and d["compute_job_id"], args.timeout, "cancellable Job")
+        send_sid, _ = new_session(rt, args, core, book, artifacts)
+        admission_sid, _ = new_session(rt, args, core, book, artifacts)
+        proxy.block_create_session = str(send_sid)
+        validate(rt, args, core, send_sid)
+        rt.require(proxy.block_entered.wait(15), "SEND_RACE_NOT_ENTERED")
+        admission_lock = rt.db_connection("public")
+        admission_lock.autocommit = False
+        admission_lock.execute("select id from trading_import_sessions where id=%s for update", (int(admission_sid),))
+        admitted = executor.submit(validate, rt, secondary_args, core, admission_sid)
+        rt.poll(lambda: rt.row(db, "select count(*) as n from pg_stat_activity where datname=current_database() and wait_event_type='Lock'", ()),
+            lambda s: s["n"] >= 1, 5, "admission row lock")
+        def pause():
+            with rt.db_connection("public") as operator:
+                return control(rt, operator, "PAUSE", start["epoch"], "orch028-cutover")
+        stopping = executor.submit(pause)
+        rt.poll(lambda: rt.row(db, "select count(*) as n from pg_stat_activity where datname=current_database() and wait_event='advisory'", ()),
+            lambda s: s["n"] >= 1, 5, "stop waits for send and admission")
+        rt.require(not stopping.done(), "PAUSE_OVERRAN_INFLIGHT_PERMITS")
+        admission_lock.commit()
+        admission_lock.close()
+        admission_lock = None
+        proxy.block_release.set()
+        admitted.result(timeout=15)
+        epoch = stopping.result(timeout=30)
+        pause_at = time.monotonic()
+        rt.require(current(rt, db)["paused"], "STOP_NOT_DURABLE")
+        proxy.reject_acks = False
+        db.execute("update trading_normalization_ack_dispatches set next_attempt_at=clock_timestamp() where run_id=%s::uuid and status='PENDING'", (ack_run,))
+        rt.poll(lambda: rt.row(db, "select status from trading_normalization_ack_dispatches where run_id=%s::uuid", (ack_run,)),
+            lambda a: a and a["status"] == "ACKNOWLEDGED", args.timeout, "existing ACK retry while paused")
+        checks[ITEMS[4]] = {"result": "PASS", "pauseWaitedForClaimSendReceiptAndAdmission": True}
+        checkpoint()
+        validation_key, retry_key = "paused-validation-" + uuid.uuid4().hex, "paused-retry-" + uuid.uuid4().hex
+        before = {sid: snapshot(rt, db, sid, [validation_key, retry_key]) for sid in (received_sid, retry_sid)}
+        for base in (args, secondary_args):
+            for sid, endpoint, key in ((received_sid, "validation", validation_key), (retry_sid, "retry", retry_key)):
+                failure = validate(rt, base, core, sid, key, endpoint, 503)
+                rt.require(failure["code"] == "TEMPORARY_DEPENDENCY_FAILURE" and failure["retryable"] is True and
+                    bool(failure["message"]) and bool(failure["traceId"]), "PAUSED_FAILURE_CONTRACT_DRIFT")
+                rt.require(snapshot(rt, db, sid, [validation_key, retry_key]) == before[sid], "REJECTED_REQUEST_CHANGED_STATE")
+        checks[ITEMS[0]], checks[ITEMS[1]], checks[ITEMS[3]] = ({"result": "PASS"}, {"result": "PASS", "rowCountsAndValuesUnchanged": True},
+            {"result": "PASS", "coreInstances": 2, "nodeIds": [0, 1]})
+        checkpoint()
+        expect_control_rejected(rt, db, "VERIFY", epoch, "unresolved-create")
+        # Match real HTTP status and the existing Compute idempotency row before attaching the Job.
+        sent, accepted = proxy.creates[str(unknown["run_id"])]
+        job_id = accepted["jobId"]
+        status = rt.request(compute, "GET", args.compute_url + f"/trading-review/normalization-jobs/{job_id}", 200)
+        identity = rt.row(compute_db, "select job_id::text from trading_review_job_idempotency where idempotency_key_hash=%s",
+            (hashlib.sha256(unknown["compute_key"].encode()).hexdigest(),))
+        rt.require(identity and identity["job_id"] == job_id and status["coreLogicalJobId"] == str(unknown["id"]) and
+            status["coreNormalizationRunId"] == str(unknown["run_id"]) and status["attemptToken"] == unknown["attempt_token"] and
+            status["deletionGeneration"] == unknown["deletion_generation"] and status["expectedInputSetHash"] == unknown["input_fingerprint"] and
+            pair(status) == V1 and status["targetVersionSet"] == unknown["target_version_set_json"], "UNKNOWN_CREATE_IDENTITY_DRIFT")
+        db.execute("select trading_normalization_reconcile_create(%s,%s,%s,%s)",
+            (unknown["id"], job_id, epoch, "orch028-http-idempotency-match"))
+        rt.require(sum(a["sessionId"] == str(unknown_sid) for a in proxy.create_attempts) == 1, "UNKNOWN_CREATE_RETRANSMITTED")
+        cancellation = rt.request(compute, "POST", args.compute_url + f"/trading-review/normalization-jobs/{cancel['compute_job_id']}/cancellation", 202,
+            headers={"Idempotency-Key": "cutover-cancel-" + uuid.uuid4().hex})
+        rt.require(cancellation["status"] == "CANCELLED", "PAUSED_CANCELLATION_FAILED")
+        rt.request(core, "DELETE", args.core_url + f"/trading-import-sessions/{admission_sid}", 202,
+            headers={"Idempotency-Key": "cutover-delete-" + uuid.uuid4().hex})
+        rt.require(core.get(args.core_url + f"/trading-import-sessions/{admission_sid}").status_code == 410, "PAUSED_DELETION_ACCESSIBLE")
+        service(rt, args, "start-compute-worker.ps1", "-EnvironmentFile", args.environment_file,
+            "-ComputeApiPath", args.compute_path, "-Schema", args.compute_schema, "-RegistryId", args.registry,
+            "-Profile", args.registry, "-TradingReview")
+        worker_poll(rt, args, lambda: dispatch(rt, db, send_sid), lambda d: d and d["status"] == "RESULT_READY", args.timeout, "paused terminal polling")
+        rt.poll(lambda: state(rt, args, core, cancel_sid), lambda s: s["status"] == "PROCESSING_FAILED", args.timeout, "paused cancellation terminal")
+        for sid in (send_sid, unknown_sid):
+            rt.require(rt.row(db, """select count(*) as n from trading_normalization_ack_dispatches where
+                core_import_session_id=%s and disposition='PERSISTED'""", (int(sid),))["n"] == 0, "NEW_PERSISTED_DURING_STOP")
+        checks[ITEMS[6]] = {"result": "PASS", "terminalPolling": True, "cancellation": True,
+            "deletion": True, "sameJobReconciled": job_id, "existingAckRetry": True, "newPersistedAcks": 0}
+        checkpoint()
+        service(rt, args, "stop-integration.ps1", "-RegistryId", args.registry, "-Service", "core-api")
+        service(rt, args, "start-trading-normalization-service.ps1", "-EnvironmentFile", args.environment_file,
+            "-CorePath", args.core_path, "-RegistryId", args.registry, "-EvidenceDirectory", args.run_dir,
+            "-DatabaseSchema", os.environ.get("TRADING_RUNTIME_CORE_SCHEMA", "public"),
+            "-ComputeBaseUrl", f"http://127.0.0.1:{args.fixture_port}")
+        validate(rt, args, core, received_sid, validation_key, expected=503)
+        rt.require(dispatch(rt, db, unsent_sid) == unsent_before and current(rt, db)["paused"], "RESTART_LOST_STOP_OR_DISPATCH")
+        checks[ITEMS[5]] = {"result": "PASS", "unsentDispatchId": str(unsent_before["id"]), "epoch": epoch}
+        checkpoint()
+        # Stop the second instance before starting a separate canary fleet.
+        service(rt, args, "stop-integration.ps1", "-RegistryId", secondary_registry, "-Service", "core-api")
+        canary = isolated_canary(rt, args, core, db, artifacts, checks)
+        rt.poll(lambda: dispatch(rt, db, send_sid),
+            lambda d: d and d["last_failure_code"] == "NORMALIZATION_TRANSITION_PAUSED", args.timeout, "publication paused after staging")
+        rt.require(rt.row(db, """select (select count(*) from trading_ledger_revisions where import_session_id=%s) as revisions,
+            (select count(*) from trading_normalization_ack_dispatches where core_import_session_id=%s) as acks""",
+            (int(send_sid), int(send_sid))) == {"revisions": 0, "acks": 0}, "NEW_PUBLICATION_DURING_STOP")
+        rt.require(not [a for a in proxy.create_attempts if a["at"] > pause_at], "CREATE_SENT_WHILE_PAUSED")
+        checks[ITEMS[2]] = {"result": "PASS", "newCreateHttpRequests": 0, "canarySchemasExcluded": True}
+        reference = "sha256/" + hashlib.sha256(json.dumps(canary, sort_keys=True).encode()).hexdigest()
+        (args.run_dir / "canary-verification.json").write_text(json.dumps(canary, sort_keys=True), encoding="utf-8")
+        control(rt, db, "VERIFY", epoch, reference)
+        control(rt, db, "RESUME", epoch, reference)
+        result["canary"] = canary
+        # Remove the synthetic scheduling delay only after verification and RESUME.
+        db.execute("update trading_normalization_dispatches set next_attempt_at=clock_timestamp() where id=%s", (unsent_before["id"],))
+        validate(rt, args, core, received_sid, validation_key)
+        validate(rt, args, core, retry_sid, retry_key, "retry")
+        checks[ITEMS[8]] = {"result": "PASS", "samePublicKeysAccepted": True, "verificationRef": reference}
+        checkpoint()
+        resumed = []
+        for sid, key, endpoint in ((received_sid, validation_key, "validation"), (retry_sid, retry_key, "retry"),
+                (unsent_sid, None, None), (unknown_sid, None, None), (send_sid, None, None)):
+            published = worker_poll(rt, args, lambda: state(rt, args, core, sid), lambda s: s["status"] == "ACCEPTED", args.timeout, "resumed publication")
+            item = dispatch(rt, db, sid)
+            run_id = str(item["run_id"])
+            rt.poll(lambda: rt.row(db, "select status from trading_normalization_ack_dispatches where run_id=%s::uuid", (run_id,)),
+                lambda a: a and a["status"] == "ACKNOWLEDGED", args.timeout, "resumed ACK")
+            if key:
+                validate(rt, args, core, sid, key, endpoint)
+                validate(rt, args, core, sid, key, endpoint)
+            counts = rt.row(db, """select (select count(*) from trading_normalization_runs where import_session_id=%s) as runs,
+                (select count(*) from trading_normalization_dispatches where run_id=%s::uuid) as dispatches,
+                (select count(*) from trading_ledger_revisions where run_id=%s::uuid) as revisions,
+                (select count(*) from trading_normalization_ack_dispatches where run_id=%s::uuid) as acks""", (int(sid), run_id, run_id, run_id))
+            jobs = rt.row(compute_db, "select count(*) as n from trading_review_jobs where core_normalization_run_id=%s", (run_id,))["n"]
+            rt.require(counts == {"runs": 2 if sid == retry_sid else 1, "dispatches": 1, "revisions": 1, "acks": 1} and jobs == 1,
+                "RESUME_DUPLICATED_WORK")
+            resumed.append({"sessionId": sid, "runId": run_id, "jobId": item["compute_job_id"],
+                "revisionId": published["createdLedgerRevisionId"], **counts, "jobs": jobs})
+        checks[ITEMS[9]] = {"result": "PASS", "sessions": resumed}
+        rt.require(sum(a["sessionId"] == str(unknown_sid) for a in proxy.create_attempts) == 1, "RECOVERED_CREATE_RETRANSMITTED")
+        rt.require(all(v["result"] == "PASS" for v in checks.values()), "ORCH028_INCOMPLETE")
+        result.update(result="PASS", epoch=epoch, mainCombination=False,
+            implementationCoreCommit=rt.git_revision(args.core_path), uncertainCreateReplayed=False)
+    finally:
+        proxy.block_release.set()
+        if admission_lock is not None:
+            admission_lock.rollback()
+            admission_lock.close()
+        executor.shutdown(wait=True)
+        service(rt, args, "stop-integration.ps1", "-RegistryId", secondary_registry, "-Service", "core-api")
+        # A failed scenario deliberately leaves the original database gate paused.
+        if result["result"] != "PASS":
+            if "epoch" in locals() and not current(rt, db)["paused"]:
+                control(rt, db, "PAUSE", epoch, "orch028-failure-after-resume")
+            result["result"] = "FAIL"
+            result["pausedAtFailure"] = current(rt, db)["paused"]
+        (args.run_dir / "evidence.json").write_text(json.dumps(evidence, indent=2, default=str) + "\n", encoding="utf-8")
diff --git a/scripts/trading_normalization_version_pin.py b/scripts/trading_normalization_version_pin.py
index eab5dae..6469d02 100644
--- a/scripts/trading_normalization_version_pin.py
+++ b/scripts/trading_normalization_version_pin.py
@@ -12,10 +12,12 @@
 import os
 import subprocess
 import threading
+import time
 import uuid
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 
 import httpx
+import psycopg
 from psycopg import sql
 
 
@@ -56,6 +58,12 @@ def __init__(self, runtime, args):
         self.reject_acks = True
         self.injected_ack_failures = 0
         self.core_schema = "public"
+        self.create_attempts = []
+        self.reject_create_sessions = set()
+        self.drop_response_sessions = set()
+        self.block_create_session = None
+        self.block_entered = threading.Event()
+        self.block_release = threading.Event()
         proxy = self
 
         class Handler(BaseHTTPRequestHandler):
@@ -83,6 +91,25 @@ def forward(self):
                     headers = {k: v for k, v in self.headers.items()
                                if k.lower() not in {"host", "connection", "content-length", "accept-encoding", "transfer-encoding"}}
                     headers["Accept-Encoding"] = "identity"
+                    is_create = self.command == "POST" and self.path.endswith("/normalization-jobs")
+                    if is_create:
+                        sent = json.loads(body)
+                        sid = str(sent["coreImportSessionId"])
+                        proxy.create_attempts.append({"sessionId": sid,
+                            "runId": sent["coreNormalizationRunId"], "at": time.monotonic()})
+                        if sid in proxy.reject_create_sessions:
+                            fault_body = encoded({"code": "TRADING_REVIEW_CAPACITY_EXCEEDED",
+                                "message": "Synthetic capacity fault", "traceId": uuid.uuid4().hex, "retryable": True})
+                            self.send_response(503)
+                            self.send_header("Content-Type", "application/json")
+                            self.send_header("Retry-After", "1")
+                            self.send_header("Content-Length", str(len(fault_body)))
+                            self.end_headers()
+                            self.wfile.write(fault_body)
+                            return
+                        if sid == proxy.block_create_session:
+                            proxy.block_entered.set()
+                            runtime.require(proxy.block_release.wait(12), "CREATE_BARRIER_TIMEOUT")
                     if self.command == "POST" and self.path.endswith("/acknowledgement"):
                         ack = json.loads(body)
                         # A separate connection must see the committed Revision before HTTP ACK.
@@ -117,6 +144,10 @@ def forward(self):
                                 "runId": sent["coreNormalizationRunId"], "payloadId": None,
                                 "descriptorReads": 0, "payloadReads": 0}
                             proxy.mode = None
+                        if str(sent["coreImportSessionId"]) in proxy.drop_response_sessions:
+                            # Real Compute accepted; the caller receives no HTTP response.
+                            self.close_connection = True
+                            return
                     if response.status_code == 200 and self.path.endswith("/result"):
                         job_id = self.path.split("/")[-2]
                         fault = proxy.faults.get(job_id)
@@ -165,6 +196,7 @@ def log_message(self, *args):
 
     def evidence(self):
         return {"ackRequests": self.acks, "errors": self.errors,
+                "createAttempts": self.create_attempts,
                 "upstreamUnavailableRequests": self.transport_failures,
                 "injectedAckFailures": self.injected_ack_failures,
                 "creates": {run: {"jobId": accepted["jobId"], "requestPair": pair(sent),
@@ -191,6 +223,7 @@ def verify_version_pin(rt, args, evidence, core, compute, db, compute_db, proxy,
                        session_id, run_id, job_id, dispatch, original, status, descriptor,
                        payload, artifacts, validation_key):
     checks = evidence["versionPin"]
+    from trading_normalization_cutover import worker_poll
 
     def begin(name):
         checks["activeScenario"] = name
@@ -286,7 +319,7 @@ def stored_job():
         proxy.mode = (mode, str(sid))
         rt.request(core, "POST", f"{args.core_url}/trading-import-sessions/{sid}/validation", 202,
                    headers={"Idempotency-Key": "pin-validate-" + uuid.uuid4().hex}, json={})
-        failed = rt.poll(lambda: rt.request(core, "GET", f"{args.core_url}/trading-import-sessions/{sid}", 200),
+        failed = worker_poll(rt, args, lambda: rt.request(core, "GET", f"{args.core_url}/trading-import-sessions/{sid}", 200),
             lambda s: s["status"] in {"PROCESSING_FAILED", "ACCEPTED", "REJECTED"}, args.timeout, "pin fault " + mode)
         rt.require(failed["status"] == "PROCESSING_FAILED" and
                    failed["failure"]["code"] == "RESULT_VERSION_PIN_MISMATCH" and
@@ -311,6 +344,10 @@ def stored_job():
 
     checks["proxy"] = proxy.evidence()
     rt.require(not proxy.errors, "PIN_PROXY_ERRORS")
+    begin("TR-ACC-IMPORT-ORCH-028")
+    from trading_normalization_cutover import verify_cutover
+    verify_cutover(rt, args, evidence, core, compute, db, compute_db, proxy, book, artifacts)
+    passed("TR-ACC-IMPORT-ORCH-028", **checks.pop("cutover"))
     begin("TR-ACC-IMPORT-ORCH-027")
     migration_rehearsal(rt, args, checks, core, compute, db, compute_db, proxy,
                         book, account, session_id, run_id, job_id, sent, dispatch, validation_key)
@@ -376,7 +413,8 @@ def copy_rows(target, source, table, condition, params=(), changes=None):
         migrate(legacy_compute, [p for p in compute_migrations if int(p.name[:4]) < 17])
         owner = rt.row(db, "select owner_member_id from trading_books where id=%s", (int(book["id"]),))["owner_member_id"]
         copy_rows(legacy_core, db, "members", "id=%s", (owner,))
-        legacy_core.execute("CREATE TABLE subscriptions (LIKE public.subscriptions INCLUDING ALL)")
+        legacy_core.execute(sql.SQL("CREATE TABLE subscriptions (LIKE {}.subscriptions INCLUDING ALL)").format(
+            sql.Identifier(os.environ.get("TRADING_RUNTIME_CORE_SCHEMA", "public"))))
         copy_rows(legacy_core, db, "subscriptions", "member_id=%s", (owner,))
         copy_rows(legacy_core, db, "trading_accounts", "id=%s", (int(account["id"]),))
         copy_rows(legacy_core, db, "trading_books", "id=%s", (int(book["id"]),),
@@ -421,7 +459,20 @@ def copy_rows(target, source, table, condition, params=(), changes=None):
         revision_before = rt.row(legacy_core, "select * from trading_ledger_revisions where run_id=%s::uuid", (run_id,))
         job_before = rt.row(legacy_compute, "select * from trading_review_jobs where job_id=%s::uuid", (job_id,))
         ack_before = rt.row(legacy_core, "select * from trading_normalization_ack_dispatches where run_id=%s::uuid", (run_id,))
-        migrate(legacy_core, [p for p in core_migrations if int(p.name.split("__")[0][1:]) == 9])
+        # Diagnose the separate atomic-runner limitation against a real terminal Job.
+        # Always roll this attempt back, then use the existing statement runner below.
+        legacy_compute.execute("BEGIN")
+        try:
+            legacy_compute.execute((args.compute_path / "migrations/0017_pin_trading_normalization_result_pair.sql")
+                .read_text(encoding="utf-8"), prepare=False)
+            checks["migration"]["atomicRunner"] = {"result": "PASS", "rolledBackDiagnostic": True}
+        except psycopg.Error as error:
+            rt.require(error.sqlstate == "55006", "UNEXPECTED_ATOMIC_MIGRATION_FAILURE")
+            checks["migration"]["atomicRunner"] = {"result": "FAIL", "sqlstate": error.sqlstate,
+                "knownDeploymentLimitation": True, "rolledBackDiagnostic": True}
+        finally:
+            legacy_compute.execute("ROLLBACK")
+        migrate(legacy_core, [p for p in core_migrations if int(p.name.split("__")[0][1:]) >= 9])
         migrate(legacy_compute, [p for p in compute_migrations if int(p.name[:4]) == 17])
         for table, conn, field, identifier in (
             ("trading_normalization_runs", legacy_core, "id", run_id),