diff --git a/crates/hts/src/backends/postgres/code_system.rs b/crates/hts/src/backends/postgres/code_system.rs
index 0a54e2e21..30599ff92 100644
--- a/crates/hts/src/backends/postgres/code_system.rs
+++ b/crates/hts/src/backends/postgres/code_system.rs
@@ -268,42 +268,81 @@ impl CodeSystemOperations for PostgresTerminologyBackend {
 /// Resolve a code system by URL, optional version, and optional date.
 ///
 /// Returns `(id, name_or_url, version)`.
+///
+/// Mirrors the SQLite implementation: an unspecified version defaults to the
+/// most recent (textual COALESCE-DESC), an explicit version with `.x` segments
+/// (or a bare numeric prefix like `"1"`) matches the highest version that
+/// shares the literal segments, and an exact version requires an exact match.
 async fn resolve_code_system(
     client: &tokio_postgres::Client,
     url: &str,
     version: Option<&str>,
     date: Option<&str>,
 ) -> Result<(String, String, Option<String>), HtsError> {
-    let rows = if let Some(ver) = version {
-        client
-            .query(
-                "SELECT id, COALESCE(name, url), version
-                 FROM code_systems
-                 WHERE url = $1 AND version = $2
-                   AND ($3::text IS NULL OR (resource_json->>'date') <= $3)",
-                &[&url, &ver, &date],
-            )
-            .await
-            .map_err(|e| HtsError::StorageError(e.to_string()))?
-    } else {
-        client
-            .query(
-                "SELECT id, COALESCE(name, url), version
-                 FROM code_systems
-                 WHERE url = $1
-                   AND ($2::text IS NULL OR (resource_json->>'date') <= $2)",
-                &[&url, &date],
-            )
-            .await
-            .map_err(|e| HtsError::StorageError(e.to_string()))?
-    };
+    let rows = client
+        .query(
+            "SELECT id, COALESCE(name, url), version
+             FROM code_systems
+             WHERE url = $1
+               AND ($2::text IS NULL OR (resource_json->>'date') <= $2)
+             ORDER BY COALESCE(version, '') DESC",
+            &[&url, &date],
+        )
+        .await
+        .map_err(|e| HtsError::StorageError(e.to_string()))?;
 
-    let row = rows
+    if rows.is_empty() {
+        return Err(HtsError::NotFound(format!("CodeSystem not found: {url}")));
+    }
+    let candidates: Vec<(String, String, Option<String>)> = rows
         .into_iter()
-        .next()
-        .ok_or_else(|| HtsError::NotFound(format!("CodeSystem not found: {url}")))?;
+        .map(|r| (r.get(0), r.get(1), r.get(2)))
+        .collect();
+
+    match version {
+        Some(ver) if ver.contains(".x") || ver == "x" || is_short_version(ver) => {
+            select_best_version_match(&candidates, ver).ok_or_else(|| {
+                HtsError::NotFound(format!("CodeSystem not found: {url} (version {ver})"))
+            })
+        }
+        Some(ver) => candidates
+            .into_iter()
+            .find(|(_, _, v)| v.as_deref() == Some(ver))
+            .ok_or_else(|| {
+                HtsError::NotFound(format!("CodeSystem not found: {url} (version {ver})"))
+            }),
+        None => Ok(candidates.into_iter().next().expect("non-empty checked")),
+    }
+}
 
-    Ok((row.get(0), row.get(1), row.get(2)))
+fn is_short_version(ver: &str) -> bool {
+    !ver.contains('.') && ver.chars().all(|c| c.is_ascii_digit())
+}
+
+fn select_best_version_match(
+    candidates: &[(String, String, Option<String>)],
+    pattern: &str,
+) -> Option<(String, String, Option<String>)> {
+    let pattern_segments: Vec<&str> = pattern.split('.').collect();
+    candidates
+        .iter()
+        .filter(|(_, _, v)| match v {
+            Some(actual) => version_matches(actual, &pattern_segments),
+            None => false,
+        })
+        .max_by(|a, b| a.2.cmp(&b.2))
+        .cloned()
+}
+
+fn version_matches(actual: &str, pattern_segments: &[&str]) -> bool {
+    let actual_segments: Vec<&str> = actual.split('.').collect();
+    if pattern_segments.len() > actual_segments.len() {
+        return false;
+    }
+    pattern_segments
+        .iter()
+        .zip(actual_segments.iter())
+        .all(|(p, a)| *p == "x" || *p == *a)
 }
 
 /// Look up a concept row by `(system_id, code)`.
diff --git a/crates/hts/src/backends/postgres/concept_map.rs b/crates/hts/src/backends/postgres/concept_map.rs
index 347c49bd7..a940f2ef4 100644
--- a/crates/hts/src/backends/postgres/concept_map.rs
+++ b/crates/hts/src/backends/postgres/concept_map.rs
@@ -100,7 +100,11 @@ impl ConceptMapOperations for PostgresTerminologyBackend {
 
         for (system_url, codes) in &by_system {
             let id_rows = client
-                .query("SELECT id FROM code_systems WHERE url = $1", &[system_url])
+                .query(
+                    "SELECT id FROM code_systems WHERE url = $1 \
+                     ORDER BY COALESCE(version, '') DESC LIMIT 1",
+                    &[system_url],
+                )
                 .await
                 .map_err(|e| HtsError::StorageError(format!("DB error: {e}")))?;
 
diff --git a/crates/hts/src/backends/postgres/mod.rs b/crates/hts/src/backends/postgres/mod.rs
index c67fbf1fa..98ae97ed4 100644
--- a/crates/hts/src/backends/postgres/mod.rs
+++ b/crates/hts/src/backends/postgres/mod.rs
@@ -113,14 +113,48 @@ impl TerminologyMetadata for PostgresTerminologyBackend {
         tokio::task::block_in_place(|| {
             tokio::runtime::Handle::current().block_on(async move {
                 let client = pool.get().await.ok()?;
-                let sql = match resource_type.as_str() {
-                    "CodeSystem" => "SELECT url FROM code_systems WHERE id = $1",
-                    "ValueSet" => "SELECT url FROM value_sets WHERE id = $1",
-                    "ConceptMap" => "SELECT url FROM concept_maps WHERE id = $1",
-                    _ => return None,
-                };
-                let rows = client.query(sql, &[&id]).await.ok()?;
-                rows.into_iter().next().map(|r| r.get::<_, String>(0))
+                match resource_type.as_str() {
+                    "CodeSystem" => {
+                        // Storage id may be the synthetic `<id>|<version>` form,
+                        // so first try a direct hit, then fall back to matching
+                        // the FHIR resource id captured in `resource_json` and
+                        // pick the latest version.
+                        if let Ok(rows) = client
+                            .query("SELECT url FROM code_systems WHERE id = $1", &[&id])
+                            .await
+                        {
+                            if let Some(row) = rows.into_iter().next() {
+                                return Some(row.get::<_, String>(0));
+                            }
+                        }
+                        let rows = client
+                            .query(
+                                "SELECT url FROM code_systems \
+                                 WHERE (resource_json->>'id') = $1 \
+                                 ORDER BY COALESCE(version, '') DESC \
+                                 LIMIT 1",
+                                &[&id],
+                            )
+                            .await
+                            .ok()?;
+                        rows.into_iter().next().map(|r| r.get::<_, String>(0))
+                    }
+                    "ValueSet" => {
+                        let rows = client
+                            .query("SELECT url FROM value_sets WHERE id = $1", &[&id])
+                            .await
+                            .ok()?;
+                        rows.into_iter().next().map(|r| r.get::<_, String>(0))
+                    }
+                    "ConceptMap" => {
+                        let rows = client
+                            .query("SELECT url FROM concept_maps WHERE id = $1", &[&id])
+                            .await
+                            .ok()?;
+                        rows.into_iter().next().map(|r| r.get::<_, String>(0))
+                    }
+                    _ => None,
+                }
             })
         })
     }
@@ -255,13 +289,17 @@ async fn write_code_system(
     let resource_json = Some(cs.resource_json.clone());
     let now = utc_now();
 
-    // Two-step upsert so we handle conflicts on *either* unique constraint
-    // (the `url` index and the PK on `id`). `ON CONFLICT (url) DO UPDATE`
-    // alone does not catch a PK collision that would happen when two
-    // concurrent importers use the same `cs.id` — the INSERT can fail on the
-    // PK arbiter index before the URL arbiter is consulted. `ON CONFLICT DO
-    // NOTHING` (no target) swallows any unique-constraint conflict, after
-    // which a plain UPDATE by URL refreshes the row and returns its id.
+    // Synthetic storage id encodes the version so multiple versions of the
+    // same canonical URL coexist even when they share the same FHIR `id`
+    // (tx-ecosystem `version/codesystem-version-1.json` + `-2.json` both ship
+    // `"id":"version"`). See `crate::import::fhir_bundle::storage_id_for`.
+    let storage_id = crate::import::fhir_bundle::storage_id_for(&cs.id, cs.version.as_deref());
+
+    // Upsert keyed on (url, version). The composite UNIQUE index
+    // `idx_code_systems_url_version` is the conflict arbiter; the legacy
+    // `ON CONFLICT (url)` is no longer applicable now that two rows can share
+    // a URL. `ON CONFLICT DO NOTHING` keeps a prior row in place; the UPDATE
+    // below then refreshes its mutable columns by (url, version).
     client
         .execute(
             "INSERT INTO code_systems
@@ -269,7 +307,7 @@ async fn write_code_system(
              VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $9)
              ON CONFLICT DO NOTHING",
             &[
-                &cs.id,
+                &storage_id,
                 &cs.url,
                 &cs.version,
                 &cs.name,
@@ -286,17 +324,15 @@ async fn write_code_system(
     let cs_rows = client
         .query(
             "UPDATE code_systems SET
-               version       = $1,
-               name          = $2,
-               title         = $3,
-               status        = $4,
-               content       = $5,
-               resource_json = $6,
-               updated_at    = $7
-             WHERE url = $8
+               name          = $1,
+               title         = $2,
+               status        = $3,
+               content       = $4,
+               resource_json = $5,
+               updated_at    = $6
+             WHERE url = $7 AND COALESCE(version, '') = COALESCE($8, '')
              RETURNING id",
             &[
-                &cs.version,
                 &cs.name,
                 &cs.title,
                 &cs.status,
@@ -304,6 +340,7 @@ async fn write_code_system(
                 &resource_json,
                 &now,
                 &cs.url,
+                &cs.version,
             ],
         )
         .await
diff --git a/crates/hts/src/backends/postgres/schema.rs b/crates/hts/src/backends/postgres/schema.rs
index 683d42567..098bac525 100644
--- a/crates/hts/src/backends/postgres/schema.rs
+++ b/crates/hts/src/backends/postgres/schema.rs
@@ -15,9 +15,14 @@ pub const SCHEMA: &str = "
 CREATE EXTENSION IF NOT EXISTS pg_trgm;
 
 -- ── Code Systems ──────────────────────────────────────────────────────────────
+-- Multi-version: a canonical URL may have multiple rows so long as each row has
+-- a distinct `version`. Uniqueness is enforced via the composite expression
+-- index `idx_code_systems_url_version` (the column-level UNIQUE on `url` was
+-- dropped). Rows with NULL version are coalesced to the empty string by the
+-- index so two un-versioned imports of the same URL still collide.
 CREATE TABLE IF NOT EXISTS code_systems (
     id           TEXT PRIMARY KEY,
-    url          TEXT NOT NULL UNIQUE,
+    url          TEXT NOT NULL,
     version      TEXT,
     name         TEXT,
     title        TEXT,
@@ -27,6 +32,29 @@ CREATE TABLE IF NOT EXISTS code_systems (
     created_at   TEXT NOT NULL,
     updated_at   TEXT NOT NULL
 );
+CREATE UNIQUE INDEX IF NOT EXISTS idx_code_systems_url_version
+    ON code_systems(url, COALESCE(version, ''));
+CREATE INDEX IF NOT EXISTS idx_code_systems_url ON code_systems(url);
+
+-- Legacy installs created code_systems with `url TEXT NOT NULL UNIQUE`, which
+-- bakes uniqueness into a hidden constraint that blocks multi-version imports.
+-- Drop both the standalone constraint name and the auto-named one so subsequent
+-- (url, version) duplicates can coexist; both forms are silently ignored when
+-- absent (legacy index is unnamed across PG versions).
+DO $$
+DECLARE
+    cons_name text;
+BEGIN
+    FOR cons_name IN
+        SELECT conname FROM pg_constraint c
+        JOIN pg_class t ON t.oid = c.conrelid
+        WHERE t.relname = 'code_systems'
+          AND c.contype = 'u'
+          AND pg_get_constraintdef(c.oid) = 'UNIQUE (url)'
+    LOOP
+        EXECUTE format('ALTER TABLE code_systems DROP CONSTRAINT %I', cons_name);
+    END LOOP;
+END $$;
 
 -- ── Concepts ───────────────────────────────────────────────────────────────────
 CREATE TABLE IF NOT EXISTS concepts (
diff --git a/crates/hts/src/backends/postgres/value_set.rs b/crates/hts/src/backends/postgres/value_set.rs
index 62552bb12..cc37749ce 100644
--- a/crates/hts/src/backends/postgres/value_set.rs
+++ b/crates/hts/src/backends/postgres/value_set.rs
@@ -337,17 +337,14 @@ async fn compute_expansion(
             Some(s) if !s.is_empty() => s,
             _ => continue,
         };
+        let inc_version = inc["version"].as_str();
 
-        let rows = client
-            .query("SELECT id FROM code_systems WHERE url = $1", &[&system_url])
-            .await
-            .map_err(|e| HtsError::StorageError(e.to_string()))?;
-
-        let system_id: String = match rows.into_iter().next() {
-            Some(r) => r.get(0),
+        let system_id = match resolve_compose_system_id(client, system_url, inc_version).await? {
+            Some(id) => id,
             None => {
                 tracing::warn!(
                     system_url,
+                    inc_version,
                     "Skipping unknown code system in ValueSet compose"
                 );
                 continue;
@@ -422,6 +419,77 @@ async fn compute_expansion(
     Ok(included)
 }
 
+/// Resolve the storage id of the `code_systems` row matching the (url,
+/// optional version) pair declared on a `compose.include[]` entry.
+///
+/// Mirrors the SQLite helper: `1.x.x`-style patterns match the highest
+/// version sharing the literal segments, an exact version requires a literal
+/// match, and `None` falls back to the latest revision.
+async fn resolve_compose_system_id(
+    client: &tokio_postgres::Client,
+    url: &str,
+    version: Option<&str>,
+) -> Result<Option<String>, HtsError> {
+    let rows = client
+        .query(
+            "SELECT id, version FROM code_systems \
+             WHERE url = $1 \
+             ORDER BY COALESCE(version, '') DESC",
+            &[&url],
+        )
+        .await
+        .map_err(|e| HtsError::StorageError(e.to_string()))?;
+
+    let candidates: Vec<(String, Option<String>)> = rows
+        .into_iter()
+        .map(|r| (r.get::<_, String>(0), r.get::<_, Option<String>>(1)))
+        .collect();
+    if candidates.is_empty() {
+        return Ok(None);
+    }
+
+    let chosen = match version {
+        Some(v) if v.contains(".x") || v == "x" || compose_short_version(v) => {
+            compose_select_version(&candidates, v)
+        }
+        Some(v) => candidates
+            .into_iter()
+            .find(|(_, ver)| ver.as_deref() == Some(v)),
+        None => candidates.into_iter().next(),
+    };
+    Ok(chosen.map(|(id, _)| id))
+}
+
+fn compose_short_version(ver: &str) -> bool {
+    !ver.contains('.') && ver.chars().all(|c| c.is_ascii_digit())
+}
+
+fn compose_select_version(
+    candidates: &[(String, Option<String>)],
+    pattern: &str,
+) -> Option<(String, Option<String>)> {
+    let segments: Vec<&str> = pattern.split('.').collect();
+    candidates
+        .iter()
+        .filter(|(_, v)| match v {
+            Some(actual) => compose_version_matches(actual, &segments),
+            None => false,
+        })
+        .max_by(|a, b| a.1.cmp(&b.1))
+        .cloned()
+}
+
+fn compose_version_matches(actual: &str, pattern_segments: &[&str]) -> bool {
+    let actual_segments: Vec<&str> = actual.split('.').collect();
+    if pattern_segments.len() > actual_segments.len() {
+        return false;
+    }
+    pattern_segments
+        .iter()
+        .zip(actual_segments.iter())
+        .all(|(p, a)| *p == "x" || *p == *a)
+}
+
 /// Find the canonical URL of a CodeSystem whose `valueSet` property equals `vs_url`.
 async fn find_cs_for_implicit_vs(
     client: &tokio_postgres::Client,
@@ -468,7 +536,11 @@ async fn build_hierarchical_expansion(
     let mut system_id_map: HashMap<String, String> = HashMap::new();
     for sys_url in &system_urls {
         let rows = client
-            .query("SELECT id FROM code_systems WHERE url = $1", &[sys_url])
+            .query(
+                "SELECT id FROM code_systems WHERE url = $1 \
+                 ORDER BY COALESCE(version, '') DESC LIMIT 1",
+                &[sys_url],
+            )
             .await
             .map_err(|e| HtsError::StorageError(e.to_string()))?;
         if let Some(row) = rows.into_iter().next() {
diff --git a/crates/hts/src/backends/sqlite/code_system.rs b/crates/hts/src/backends/sqlite/code_system.rs
index d965623e0..a03f74a4a 100644
--- a/crates/hts/src/backends/sqlite/code_system.rs
+++ b/crates/hts/src/backends/sqlite/code_system.rs
@@ -369,52 +369,88 @@ fn check_ancestor(
 ///
 /// Returns `(id, name_or_url, version)`.
 ///
-/// When `date` is provided, only code systems whose `$.date` (from `resource_json`)
-/// is ≤ the requested date are matched, enabling point-in-time evaluation.
+/// Version-matching rules (mirroring tx.fhir.org behaviour exercised by the
+/// tx-ecosystem `version/` test suite):
+///
+/// * `Some("1.x.x")` / `Some("1.x")` / `Some("1")` — partial match.  Each `x`
+///   segment acts as a wildcard, so `1.x.x` matches `1.0.0`, `1.2.0`, etc.
+///   The highest matching version wins.
+/// * `Some("1.0.0")` — exact match required.
+/// * `None` — no version pinning; the row with the highest `version` (sorted
+///   descending as text) wins so callers default to the most recent revision.
+///
+/// When `date` is provided, only code systems whose `$.date` (from
+/// `resource_json`) is ≤ the requested date are considered.
 fn resolve_code_system(
     conn: &rusqlite::Connection,
     url: &str,
     version: Option<&str>,
     date: Option<&str>,
 ) -> Result<(String, String, Option<String>), HtsError> {
-    let result = if let Some(ver) = version {
-        conn.query_row(
-            "SELECT id, COALESCE(name, url), version \
-             FROM code_systems \
-             WHERE url = ?1 AND version = ?2 \
-               AND (?3 IS NULL OR json_extract(resource_json, '$.date') <= ?3)",
-            rusqlite::params![url, ver, date],
-            |row| {
-                Ok((
-                    row.get::<_, String>(0)?,
-                    row.get::<_, String>(1)?,
-                    row.get::<_, Option<String>>(2)?,
-                ))
-            },
-        )
-    } else {
-        conn.query_row(
+    let candidates = fetch_versions(conn, url, date)?;
+    if candidates.is_empty() {
+        return Err(HtsError::NotFound(format!("CodeSystem not found: {url}")));
+    }
+
+    let chosen = match version {
+        Some(ver)
+            if ver.contains(".x") || ver == "x" || super::code_system_version_is_short(ver) =>
+        {
+            // Project to (id, version) for the shared matcher then re-attach name.
+            let id_ver: Vec<(String, Option<String>)> = candidates
+                .iter()
+                .map(|(id, _, v)| (id.clone(), v.clone()))
+                .collect();
+            let (matched_id, _) = super::code_system_select_version_match(&id_ver, ver)
+                .ok_or_else(|| {
+                    HtsError::NotFound(format!("CodeSystem not found: {url} (version {ver})"))
+                })?;
+            candidates
+                .into_iter()
+                .find(|(id, _, _)| id == &matched_id)
+                .expect("matched id was sourced from candidates")
+        }
+        Some(ver) => candidates
+            .iter()
+            .find(|(_, _, v)| v.as_deref() == Some(ver))
+            .cloned()
+            .ok_or_else(|| {
+                HtsError::NotFound(format!("CodeSystem not found: {url} (version {ver})"))
+            })?,
+        None => candidates.into_iter().next().expect("non-empty checked"),
+    };
+    Ok(chosen)
+}
+
+/// Fetch every (id, name, version) row for `url`, sorted with the highest
+/// version first so `None`-version requests default to the newest revision.
+fn fetch_versions(
+    conn: &rusqlite::Connection,
+    url: &str,
+    date: Option<&str>,
+) -> Result<Vec<(String, String, Option<String>)>, HtsError> {
+    let mut stmt = conn
+        .prepare(
             "SELECT id, COALESCE(name, url), version \
              FROM code_systems \
              WHERE url = ?1 \
-               AND (?2 IS NULL OR json_extract(resource_json, '$.date') <= ?2)",
-            rusqlite::params![url, date],
-            |row| {
-                Ok((
-                    row.get::<_, String>(0)?,
-                    row.get::<_, String>(1)?,
-                    row.get::<_, Option<String>>(2)?,
-                ))
-            },
+               AND (?2 IS NULL OR json_extract(resource_json, '$.date') <= ?2) \
+             ORDER BY COALESCE(version, '') DESC",
         )
-    };
+        .map_err(|e| HtsError::StorageError(e.to_string()))?;
 
-    result.map_err(|e| match e {
-        rusqlite::Error::QueryReturnedNoRows => {
-            HtsError::NotFound(format!("CodeSystem not found: {url}"))
-        }
-        other => HtsError::StorageError(other.to_string()),
-    })
+    let rows = stmt
+        .query_map(rusqlite::params![url, date], |row| {
+            Ok((
+                row.get::<_, String>(0)?,
+                row.get::<_, String>(1)?,
+                row.get::<_, Option<String>>(2)?,
+            ))
+        })
+        .map_err(|e| HtsError::StorageError(e.to_string()))?;
+
+    rows.collect::<Result<Vec<_>, _>>()
+        .map_err(|e| HtsError::StorageError(e.to_string()))
 }
 
 /// Look up a concept row by `(system_id, code)`.
@@ -1155,6 +1191,133 @@ mod tests {
         assert_eq!(resp.display, Some("Term (English default)".into()));
     }
 
+    // ── Multi-version resolution ──────────────────────────────────────────────
+
+    /// Insert two versions of the same canonical URL with different concept
+    /// displays so we can assert which version got picked.
+    fn seed_two_versions(b: &SqliteTerminologyBackend) {
+        let conn = b.pool().get().unwrap();
+        conn.execute_batch(
+            "INSERT INTO code_systems
+                 (id, url, version, name, status, content, created_at, updated_at)
+             VALUES ('mv|1.0.0', 'http://example.org/mv', '1.0.0', 'MV',
+                     'active', 'complete', '2024-01-01', '2024-01-01'),
+                    ('mv|1.2.0', 'http://example.org/mv', '1.2.0', 'MV',
+                     'active', 'complete', '2024-01-02', '2024-01-02');
+
+             INSERT INTO concepts (id, system_id, code, display)
+             VALUES (300, 'mv|1.0.0', 'code1', 'Display 1 (1.0)'),
+                    (301, 'mv|1.2.0', 'code1', 'Display 1 (1.2)');",
+        )
+        .unwrap();
+    }
+
+    #[tokio::test]
+    async fn lookup_without_version_picks_latest() {
+        let b = backend();
+        seed_two_versions(&b);
+
+        let resp = b
+            .lookup(
+                &ctx(),
+                LookupRequest {
+                    system: "http://example.org/mv".into(),
+                    code: "code1".into(),
+                    ..Default::default()
+                },
+            )
+            .await
+            .unwrap();
+
+        assert_eq!(resp.version.as_deref(), Some("1.2.0"));
+        assert_eq!(resp.display.as_deref(), Some("Display 1 (1.2)"));
+    }
+
+    #[tokio::test]
+    async fn lookup_with_exact_version_targets_that_row() {
+        let b = backend();
+        seed_two_versions(&b);
+
+        let resp = b
+            .lookup(
+                &ctx(),
+                LookupRequest {
+                    system: "http://example.org/mv".into(),
+                    code: "code1".into(),
+                    version: Some("1.0.0".into()),
+                    ..Default::default()
+                },
+            )
+            .await
+            .unwrap();
+
+        assert_eq!(resp.version.as_deref(), Some("1.0.0"));
+        assert_eq!(resp.display.as_deref(), Some("Display 1 (1.0)"));
+    }
+
+    #[tokio::test]
+    async fn lookup_with_partial_wildcard_picks_highest_match() {
+        let b = backend();
+        seed_two_versions(&b);
+
+        // `1.x.x` matches both 1.0.0 and 1.2.0; the higher one wins.
+        let resp = b
+            .lookup(
+                &ctx(),
+                LookupRequest {
+                    system: "http://example.org/mv".into(),
+                    code: "code1".into(),
+                    version: Some("1.x.x".into()),
+                    ..Default::default()
+                },
+            )
+            .await
+            .unwrap();
+
+        assert_eq!(resp.version.as_deref(), Some("1.2.0"));
+    }
+
+    #[tokio::test]
+    async fn lookup_with_short_version_prefix_matches_any_in_family() {
+        let b = backend();
+        seed_two_versions(&b);
+
+        // Bare numeric prefix `1` should match any 1.x.x version.
+        let resp = b
+            .lookup(
+                &ctx(),
+                LookupRequest {
+                    system: "http://example.org/mv".into(),
+                    code: "code1".into(),
+                    version: Some("1".into()),
+                    ..Default::default()
+                },
+            )
+            .await
+            .unwrap();
+        assert_eq!(resp.version.as_deref(), Some("1.2.0"));
+    }
+
+    #[tokio::test]
+    async fn lookup_with_unknown_version_returns_not_found() {
+        let b = backend();
+        seed_two_versions(&b);
+
+        let err = b
+            .lookup(
+                &ctx(),
+                LookupRequest {
+                    system: "http://example.org/mv".into(),
+                    code: "code1".into(),
+                    version: Some("9.9.9".into()),
+                    ..Default::default()
+                },
+            )
+            .await
+            .unwrap_err();
+        assert!(matches!(err, HtsError::NotFound(_)));
+    }
+
     #[tokio::test]
     async fn lookup_without_display_language_returns_all_designations() {
         let b = backend();
diff --git a/crates/hts/src/backends/sqlite/concept_map.rs b/crates/hts/src/backends/sqlite/concept_map.rs
index b4d35492e..2ff85b84e 100644
--- a/crates/hts/src/backends/sqlite/concept_map.rs
+++ b/crates/hts/src/backends/sqlite/concept_map.rs
@@ -298,10 +298,12 @@ fn closure_sync(conn: &Connection, req: &ClosureRequest) -> Result<ClosureRespon
     let mut groups: Vec<serde_json::Value> = Vec::new();
 
     for (system_url, codes) in &by_system {
-        // Look up the internal code_system.id.
+        // Look up the internal code_system.id, picking the latest version when
+        // a URL has multiple stored revisions.
         let system_id_opt: Option<String> = conn
             .query_row(
-                "SELECT id FROM code_systems WHERE url = ?1",
+                "SELECT id FROM code_systems WHERE url = ?1 \
+                 ORDER BY COALESCE(version, '') DESC LIMIT 1",
                 rusqlite::params![system_url],
                 |row| row.get(0),
             )
diff --git a/crates/hts/src/backends/sqlite/mod.rs b/crates/hts/src/backends/sqlite/mod.rs
index c553c69c8..e433dcd3b 100644
--- a/crates/hts/src/backends/sqlite/mod.rs
+++ b/crates/hts/src/backends/sqlite/mod.rs
@@ -65,7 +65,7 @@ impl SqliteTerminologyBackend {
 
         // Bootstrap: apply pragmas + schema on a single connection.
         {
-            let conn = pool.get().map_err(|e| {
+            let mut conn = pool.get().map_err(|e| {
                 HtsError::StorageError(format!("Failed to acquire connection for init: {e}"))
             })?;
 
@@ -79,6 +79,14 @@ impl SqliteTerminologyBackend {
             schema::migrate_search_columns(&conn).map_err(|e| {
                 HtsError::StorageError(format!("Failed to apply search column migration: {e}"))
             })?;
+            // Drop legacy column-level UNIQUE on code_systems.url so multi-version
+            // CodeSystems can share a canonical URL. Idempotent — no-op when the
+            // table was already created without that constraint.
+            schema::migrate_code_systems_drop_url_unique(&mut conn).map_err(|e| {
+                HtsError::StorageError(format!(
+                    "Failed to drop legacy code_systems.url UNIQUE: {e}"
+                ))
+            })?;
         }
 
         info!(db_path, "SQLite terminology backend initialized");
@@ -104,7 +112,7 @@ impl SqliteTerminologyBackend {
             .map_err(|e| HtsError::StorageError(format!("Failed to create in-memory pool: {e}")))?;
 
         {
-            let conn = pool.get().map_err(|e| {
+            let mut conn = pool.get().map_err(|e| {
                 HtsError::StorageError(format!("Failed to acquire in-memory connection: {e}"))
             })?;
             conn.execute_batch("PRAGMA foreign_keys=ON;").map_err(|e| {
@@ -118,6 +126,11 @@ impl SqliteTerminologyBackend {
                     "Failed to apply in-memory search column migration: {e}"
                 ))
             })?;
+            schema::migrate_code_systems_drop_url_unique(&mut conn).map_err(|e| {
+                HtsError::StorageError(format!(
+                    "Failed to drop legacy code_systems.url UNIQUE: {e}"
+                ))
+            })?;
         }
 
         Ok(Self { pool })
@@ -161,19 +174,88 @@ impl TerminologyMetadata for SqliteTerminologyBackend {
 
     /// Look up the canonical URL for a ValueSet or ConceptMap by its FHIR `id`.
     ///
-    /// Queries the HTS normalized table for the resource type.  Returns `None`
-    /// when the ID is unknown.
+    /// CodeSystem rows can share a FHIR id across versions (the synthetic
+    /// storage id encodes the version), so for CodeSystem we also try matching
+    /// `resource_json.id` and pick the latest version when several rows match.
+    ///
+    /// Returns `None` when the ID is unknown.
     fn resource_url_by_id(&self, resource_type: &str, id: &str) -> Option<String> {
         let conn = self.pool.get().ok()?;
-        let sql = match resource_type {
-            "CodeSystem" => "SELECT url FROM code_systems WHERE id = ?1",
-            "ValueSet" => "SELECT url FROM value_sets WHERE id = ?1",
-            "ConceptMap" => "SELECT url FROM concept_maps WHERE id = ?1",
-            _ => return None,
-        };
-        conn.query_row(sql, rusqlite::params![id], |row| row.get::<_, String>(0))
-            .ok()
+        match resource_type {
+            "CodeSystem" => {
+                if let Ok(url) = conn.query_row(
+                    "SELECT url FROM code_systems WHERE id = ?1",
+                    rusqlite::params![id],
+                    |row| row.get::<_, String>(0),
+                ) {
+                    return Some(url);
+                }
+                conn.query_row(
+                    "SELECT url FROM code_systems \
+                     WHERE json_extract(resource_json, '$.id') = ?1 \
+                     ORDER BY COALESCE(version, '') DESC \
+                     LIMIT 1",
+                    rusqlite::params![id],
+                    |row| row.get::<_, String>(0),
+                )
+                .ok()
+            }
+            "ValueSet" => conn
+                .query_row(
+                    "SELECT url FROM value_sets WHERE id = ?1",
+                    rusqlite::params![id],
+                    |row| row.get::<_, String>(0),
+                )
+                .ok(),
+            "ConceptMap" => conn
+                .query_row(
+                    "SELECT url FROM concept_maps WHERE id = ?1",
+                    rusqlite::params![id],
+                    |row| row.get::<_, String>(0),
+                )
+                .ok(),
+            _ => None,
+        }
+    }
+}
+
+// ── Multi-version helpers (shared by code_system.rs + value_set.rs) ───────────
+
+/// `true` for versions like `"1"` or `"2"` that should match any version
+/// starting with that segment.
+pub(super) fn code_system_version_is_short(ver: &str) -> bool {
+    !ver.contains('.') && ver.chars().all(|c| c.is_ascii_digit())
+}
+
+/// Pick the highest-version row that matches `pattern`.
+///
+/// Each `x` segment in the pattern is a wildcard.  Bare numeric prefixes
+/// (e.g. `"1"`) match any version starting with that segment.  Returns
+/// `None` when no candidate matches.
+pub(super) fn code_system_select_version_match(
+    candidates: &[(String, Option<String>)],
+    pattern: &str,
+) -> Option<(String, Option<String>)> {
+    let segments: Vec<&str> = pattern.split('.').collect();
+    candidates
+        .iter()
+        .filter(|(_, v)| match v {
+            Some(actual) => code_system_version_matches(actual, &segments),
+            None => false,
+        })
+        .max_by(|a, b| a.1.cmp(&b.1))
+        .cloned()
+}
+
+fn code_system_version_matches(actual: &str, pattern_segments: &[&str]) -> bool {
+    let actual_segments: Vec<&str> = actual.split('.').collect();
+    if pattern_segments.len() > actual_segments.len() {
+        return false;
     }
+    pattern_segments
+        .iter()
+        .zip(actual_segments.iter())
+        .all(|(p, a)| *p == "x" || *p == *a)
 }
 
 // ── BundleImportBackend ────────────────────────────────────────────────────────
diff --git a/crates/hts/src/backends/sqlite/schema.rs b/crates/hts/src/backends/sqlite/schema.rs
index 33e915a68..edbd6ed2a 100644
--- a/crates/hts/src/backends/sqlite/schema.rs
+++ b/crates/hts/src/backends/sqlite/schema.rs
@@ -17,11 +17,20 @@
 ///
 /// All statements use `CREATE TABLE IF NOT EXISTS` / `CREATE INDEX IF NOT EXISTS`
 /// so this can be applied safely on every startup without error.
+///
+/// # Multi-version code systems
+///
+/// `code_systems` allows multiple rows with the same `url` provided each row
+/// has a distinct `version`. Uniqueness is enforced via the composite index
+/// `idx_code_systems_url_version` (the column-level `UNIQUE` constraint on
+/// `url` was dropped to make multi-version coexistence possible). Rows whose
+/// `version` is `NULL` are coalesced to the empty string by the index so two
+/// imports of the same URL with no version still collide.
 pub const SCHEMA: &str = "
 -- ── Code Systems ──────────────────────────────────────────────────────────────
 CREATE TABLE IF NOT EXISTS code_systems (
     id          TEXT PRIMARY KEY,
-    url         TEXT NOT NULL UNIQUE,
+    url         TEXT NOT NULL,
     version     TEXT,
     name        TEXT,
     status      TEXT NOT NULL DEFAULT 'active',
@@ -29,6 +38,9 @@ CREATE TABLE IF NOT EXISTS code_systems (
     created_at  TEXT NOT NULL,
     updated_at  TEXT NOT NULL
 );
+CREATE UNIQUE INDEX IF NOT EXISTS idx_code_systems_url_version
+    ON code_systems(url, COALESCE(version, ''));
+CREATE INDEX IF NOT EXISTS idx_code_systems_url ON code_systems(url);
 
 -- ── Concepts ───────────────────────────────────────────────────────────────────
 CREATE TABLE IF NOT EXISTS concepts (
@@ -151,6 +163,59 @@ pub fn migrate_search_columns(conn: &rusqlite::Connection) -> rusqlite::Result<(
     Ok(())
 }
 
+/// Drop the legacy `UNIQUE` constraint on `code_systems.url` so multiple rows
+/// with the same canonical URL but different `version` values can coexist.
+///
+/// The original DDL declared `url TEXT NOT NULL UNIQUE`, baking the constraint
+/// into an internal `sqlite_autoindex_*` index that cannot be dropped directly.
+/// We detect that legacy index by inspecting `sqlite_master.sql` and, when
+/// present, rebuild the table without the column-level `UNIQUE` so the new
+/// composite `(url, version)` index in [`SCHEMA`] becomes the sole uniqueness
+/// guarantee. Idempotent: a no-op once the rebuild has run.
+pub fn migrate_code_systems_drop_url_unique(
+    conn: &mut rusqlite::Connection,
+) -> rusqlite::Result<()> {
+    let needs_rebuild: bool = conn
+        .query_row(
+            "SELECT 1 FROM sqlite_master \
+             WHERE type='table' AND name='code_systems' \
+               AND sql LIKE '%url%TEXT%NOT NULL%UNIQUE%'",
+            [],
+            |_| Ok(true),
+        )
+        .unwrap_or(false);
+
+    if !needs_rebuild {
+        return Ok(());
+    }
+
+    let tx = conn.transaction()?;
+    tx.execute_batch(
+        "CREATE TABLE code_systems_new (
+             id            TEXT PRIMARY KEY,
+             url           TEXT NOT NULL,
+             version       TEXT,
+             name          TEXT,
+             status        TEXT NOT NULL DEFAULT 'active',
+             content       TEXT NOT NULL DEFAULT 'complete',
+             created_at    TEXT NOT NULL,
+             updated_at    TEXT NOT NULL,
+             title         TEXT,
+             resource_json TEXT
+         );
+         INSERT INTO code_systems_new
+             (id, url, version, name, status, content, created_at, updated_at, title, resource_json)
+         SELECT id, url, version, name, status, content, created_at, updated_at, title, resource_json
+           FROM code_systems;
+         DROP TABLE code_systems;
+         ALTER TABLE code_systems_new RENAME TO code_systems;
+         CREATE UNIQUE INDEX IF NOT EXISTS idx_code_systems_url_version
+             ON code_systems(url, COALESCE(version, ''));
+         CREATE INDEX IF NOT EXISTS idx_code_systems_url ON code_systems(url);",
+    )?;
+    tx.commit()
+}
+
 // ── Tests ─────────────────────────────────────────────────────────────────────
 
 #[cfg(test)]
diff --git a/crates/hts/src/backends/sqlite/value_set.rs b/crates/hts/src/backends/sqlite/value_set.rs
index ad2827243..85b34f552 100644
--- a/crates/hts/src/backends/sqlite/value_set.rs
+++ b/crates/hts/src/backends/sqlite/value_set.rs
@@ -428,22 +428,18 @@ fn compute_expansion(
             Some(s) if !s.is_empty() => s,
             _ => continue,
         };
+        let inc_version = inc["version"].as_str();
 
-        // Resolve code system id from the `code_systems` table.
-        let system_id: Option<String> = conn
-            .query_row(
-                "SELECT id FROM code_systems WHERE url = ?1",
-                [system_url],
-                |row| row.get(0),
-            )
-            .optional()
-            .map_err(|e| HtsError::StorageError(e.to_string()))?;
-
-        let system_id = match system_id {
+        // Resolve the code_systems row honouring an optional version pin.
+        // When the include declares a version we look up that specific (url,
+        // version) row; otherwise we pick the highest-versioned row so a
+        // version-less include still expands the latest revision.
+        let system_id = match resolve_compose_system_id(conn, system_url, inc_version)? {
             Some(id) => id,
             None => {
                 tracing::warn!(
                     system_url,
+                    inc_version,
                     "Skipping unknown code system in ValueSet compose"
                 );
                 continue;
@@ -628,6 +624,52 @@ fn apply_compose_filters(
     Ok(result)
 }
 
+/// Look up the storage id of a code_systems row given a canonical URL and an
+/// optional version constraint from a `compose.include[]` entry.
+///
+/// Mirrors the version-resolution rules used by `$lookup` /
+/// `$validate-code` / `$subsumes`: an exact version requires a literal match,
+/// `1.x.x` / `1.x` / bare `1` patterns match the highest version that shares
+/// the literal segments, and `None` falls back to the most recent revision.
+///
+/// Returns `Ok(None)` when no row matches so callers can skip the include
+/// rather than abort the whole expansion.
+fn resolve_compose_system_id(
+    conn: &Connection,
+    url: &str,
+    version: Option<&str>,
+) -> Result<Option<String>, HtsError> {
+    let mut stmt = conn
+        .prepare(
+            "SELECT id, version FROM code_systems \
+             WHERE url = ?1 \
+             ORDER BY COALESCE(version, '') DESC",
+        )
+        .map_err(|e| HtsError::StorageError(e.to_string()))?;
+
+    let rows: Vec<(String, Option<String>)> = stmt
+        .query_map(rusqlite::params![url], |row| {
+            Ok((row.get::<_, String>(0)?, row.get::<_, Option<String>>(1)?))
+        })
+        .map_err(|e| HtsError::StorageError(e.to_string()))?
+        .collect::<Result<_, _>>()
+        .map_err(|e| HtsError::StorageError(e.to_string()))?;
+
+    if rows.is_empty() {
+        return Ok(None);
+    }
+
+    let chosen = match version {
+        Some(v) if v.contains(".x") || v == "x" || super::code_system_version_is_short(v) => {
+            super::code_system_select_version_match(&rows, v)
+        }
+        Some(v) => rows.into_iter().find(|(_, ver)| ver.as_deref() == Some(v)),
+        None => rows.into_iter().next(),
+    };
+
+    Ok(chosen.map(|(id, _)| id))
+}
+
 /// Find the canonical URL of a CodeSystem whose `valueSet` property equals `vs_url`.
 ///
 /// When a CodeSystem carries `"valueSet": "http://..."` it implicitly defines a
@@ -685,13 +727,16 @@ fn build_hierarchical_expansion(
         .map(|c| (c.system.clone(), c.code.clone()))
         .collect();
 
-    // For each unique system URL, look up the system_id from code_systems.
+    // For each unique system URL, pick the latest-versioned id so the
+    // hierarchy edges we walk reflect the most recent revision when the
+    // expansion combines codes from multiple versions of the same URL.
     let system_urls: HashSet<String> = flat.iter().map(|c| c.system.clone()).collect();
     let mut system_id_map: HashMap<String, String> = HashMap::new();
     for sys_url in &system_urls {
         if let Some(id) = conn
             .query_row(
-                "SELECT id FROM code_systems WHERE url = ?1",
+                "SELECT id FROM code_systems WHERE url = ?1 \
+                 ORDER BY COALESCE(version, '') DESC LIMIT 1",
                 [sys_url],
                 |row| row.get::<_, String>(0),
             )
@@ -1427,6 +1472,177 @@ mod tests {
         }"#
     }
 
+    /// `compose.include[].version` must select the matching code_systems row,
+    /// not just the latest one.
+    ///
+    /// The bundle imports two CodeSystems sharing
+    /// `http://example.org/cs-mv` with versions `1.0.0` (codes A, B) and
+    /// `2.0.0` (codes C, D), plus three ValueSets that pin different versions
+    /// in their compose includes. Each $expand should return only the codes
+    /// belonging to the selected version.
+    fn bundle_with_mv_compose() -> &'static str {
+        r#"{
+          "resourceType": "Bundle",
+          "type": "collection",
+          "entry": [
+            {
+              "resource": {
+                "resourceType": "CodeSystem",
+                "id": "mv",
+                "url": "http://example.org/cs-mv",
+                "version": "1.0.0",
+                "status": "active",
+                "content": "complete",
+                "concept": [
+                  { "code": "A", "display": "A v1" },
+                  { "code": "B", "display": "B v1" }
+                ]
+              }
+            },
+            {
+              "resource": {
+                "resourceType": "CodeSystem",
+                "id": "mv",
+                "url": "http://example.org/cs-mv",
+                "version": "2.0.0",
+                "status": "active",
+                "content": "complete",
+                "concept": [
+                  { "code": "C", "display": "C v2" },
+                  { "code": "D", "display": "D v2" }
+                ]
+              }
+            },
+            {
+              "resource": {
+                "resourceType": "ValueSet",
+                "id": "vs-pin-v1",
+                "url": "http://example.org/vs-pin-v1",
+                "status": "active",
+                "compose": {
+                  "include": [{
+                    "system": "http://example.org/cs-mv",
+                    "version": "1.0.0"
+                  }]
+                }
+              }
+            },
+            {
+              "resource": {
+                "resourceType": "ValueSet",
+                "id": "vs-pin-v2",
+                "url": "http://example.org/vs-pin-v2",
+                "status": "active",
+                "compose": {
+                  "include": [{
+                    "system": "http://example.org/cs-mv",
+                    "version": "2.0.0"
+                  }]
+                }
+              }
+            },
+            {
+              "resource": {
+                "resourceType": "ValueSet",
+                "id": "vs-mixed",
+                "url": "http://example.org/vs-mixed",
+                "status": "active",
+                "compose": {
+                  "include": [
+                    {
+                      "system": "http://example.org/cs-mv",
+                      "version": "1.0.0",
+                      "concept": [{ "code": "A" }]
+                    },
+                    {
+                      "system": "http://example.org/cs-mv",
+                      "version": "2.0.0",
+                      "concept": [{ "code": "C" }]
+                    }
+                  ]
+                }
+              }
+            }
+          ]
+        }"#
+    }
+
+    #[tokio::test]
+    async fn expand_compose_version_pin_selects_v1_codes() {
+        let b = backend();
+        b.import_bundle(&ctx(), bundle_with_mv_compose().as_bytes())
+            .await
+            .unwrap();
+
+        let resp = b
+            .expand(
+                &ctx(),
+                ExpandRequest {
+                    url: Some("http://example.org/vs-pin-v1".into()),
+                    ..Default::default()
+                },
+            )
+            .await
+            .unwrap();
+
+        let codes: Vec<&str> = resp.contains.iter().map(|c| c.code.as_str()).collect();
+        assert!(codes.contains(&"A"), "v1.0.0 codes only: {codes:?}");
+        assert!(codes.contains(&"B"));
+        assert!(!codes.contains(&"C"), "v2.0.0 codes must not leak in");
+        assert!(!codes.contains(&"D"));
+    }
+
+    #[tokio::test]
+    async fn expand_compose_version_pin_selects_v2_codes() {
+        let b = backend();
+        b.import_bundle(&ctx(), bundle_with_mv_compose().as_bytes())
+            .await
+            .unwrap();
+
+        let resp = b
+            .expand(
+                &ctx(),
+                ExpandRequest {
+                    url: Some("http://example.org/vs-pin-v2".into()),
+                    ..Default::default()
+                },
+            )
+            .await
+            .unwrap();
+
+        let codes: Vec<&str> = resp.contains.iter().map(|c| c.code.as_str()).collect();
+        assert!(codes.contains(&"C"));
+        assert!(codes.contains(&"D"));
+        assert!(!codes.contains(&"A"));
+        assert!(!codes.contains(&"B"));
+    }
+
+    /// Mirrors `tx-ecosystem/tests/version/valueset-version-mixed.json`:
+    /// each include clause pulls a single code from its own pinned version.
+    #[tokio::test]
+    async fn expand_compose_mixed_versions_combines_codes_per_version() {
+        let b = backend();
+        b.import_bundle(&ctx(), bundle_with_mv_compose().as_bytes())
+            .await
+            .unwrap();
+
+        let resp = b
+            .expand(
+                &ctx(),
+                ExpandRequest {
+                    url: Some("http://example.org/vs-mixed".into()),
+                    ..Default::default()
+                },
+            )
+            .await
+            .unwrap();
+
+        let codes: Vec<&str> = resp.contains.iter().map(|c| c.code.as_str()).collect();
+        assert!(codes.contains(&"A"), "v1 code A pulled from version 1.0.0");
+        assert!(codes.contains(&"C"), "v2 code C pulled from version 2.0.0");
+        assert_eq!(resp.total, Some(2), "exactly two codes: {codes:?}");
+    }
+
     #[tokio::test]
     async fn expand_implicit_vs_returns_all_cs_codes() {
         let b = backend();
diff --git a/crates/hts/src/import/fhir_bundle.rs b/crates/hts/src/import/fhir_bundle.rs
index fbacb2d86..64176bf87 100644
--- a/crates/hts/src/import/fhir_bundle.rs
+++ b/crates/hts/src/import/fhir_bundle.rs
@@ -104,16 +104,24 @@ fn write_code_system(
     let resource_json = serde_json::to_string(&cs.resource_json).ok();
     let now = utc_now();
 
-    // Non-destructive upsert: if a row with the same `url` already exists (e.g.
-    // from a prior chunk of a large CodeSystem), keep it and its concepts
-    // intact. Re-inserts with a different `id` are ignored rather than firing
-    // the `ON DELETE CASCADE` on the `concepts.system_id` FK.
+    // Synthetic storage id: `<fhir-id>|<version>` (or `<fhir-id>` when version
+    // is absent). This guarantees distinct rows per (url, version) even when
+    // the upstream resource ships the same FHIR `id` for multiple versions
+    // (e.g. tx-ecosystem `version/codesystem-version-1.json` + `-2.json` both
+    // declare `"id":"version"`). The pipe character is reserved in canonical
+    // URLs so it cannot collide with a legitimate FHIR id.
+    let storage_id = storage_id_for(&cs.id, cs.version.as_deref());
+
+    // Upsert keyed on (url, version): a re-import of the same version updates
+    // the existing row rather than creating a new one or wiping sibling
+    // versions. The composite UNIQUE index on (url, COALESCE(version,''))
+    // guarantees each (url, version) maps to at most one storage row.
     conn.execute(
         "INSERT OR IGNORE INTO code_systems
          (id, url, version, name, title, status, content, resource_json, created_at, updated_at)
          VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?9)",
         rusqlite::params![
-            cs.id,
+            storage_id,
             cs.url,
             cs.version,
             cs.name,
@@ -128,16 +136,14 @@ fn write_code_system(
 
     conn.execute(
         "UPDATE code_systems SET
-           version       = ?1,
-           name          = ?2,
-           title         = ?3,
-           status        = ?4,
-           content       = ?5,
-           resource_json = ?6,
-           updated_at    = ?7
-         WHERE url = ?8",
+           name          = ?1,
+           title         = ?2,
+           status        = ?3,
+           content       = ?4,
+           resource_json = ?5,
+           updated_at    = ?6
+         WHERE url = ?7 AND COALESCE(version, '') = COALESCE(?8, '')",
         rusqlite::params![
-            cs.version,
             cs.name,
             cs.title,
             cs.status,
@@ -145,16 +151,19 @@ fn write_code_system(
             resource_json,
             now,
             cs.url,
+            cs.version,
         ],
     )
     .map_err(|e| HtsError::StorageError(e.to_string()))?;
 
-    // Concepts reference the authoritative `id` resolved by URL, which may
-    // differ from `cs.id` if a prior chunk created the row.
+    // Concepts reference the authoritative `id` resolved by (url, version),
+    // which is the `storage_id` we just upserted. Re-fetch via the index so
+    // a prior import that used a different synthesised id still wins.
     let system_id: String = conn
         .query_row(
-            "SELECT id FROM code_systems WHERE url = ?1",
-            rusqlite::params![cs.url],
+            "SELECT id FROM code_systems \
+             WHERE url = ?1 AND COALESCE(version, '') = COALESCE(?2, '')",
+            rusqlite::params![cs.url, cs.version],
             |row| row.get(0),
         )
         .map_err(|e| HtsError::StorageError(format!("Failed to resolve CodeSystem id: {e}")))?;
@@ -350,12 +359,32 @@ fn write_concept_map(
 
 /// Look up a CodeSystem's canonical URL by its FHIR resource `id`.
 ///
+/// Falls back to matching the original FHIR id stored inside `resource_json`
+/// when the synthetic storage id (`<id>|<version>`) doesn't directly match —
+/// this is what CRUD callers see in URL paths like `/CodeSystem/version`.
+/// When several versions share the same FHIR id we return the latest version
+/// (sorted descending as text) so the caller has a defined target.
+///
 /// Returns `Ok(None)` when no code system with that `id` exists.
 #[cfg(feature = "sqlite")]
 pub(crate) fn get_code_system_url(conn: &Connection, id: &str) -> Result<Option<String>, HtsError> {
     use rusqlite::OptionalExtension;
+    if let Some(url) = conn
+        .query_row(
+            "SELECT url FROM code_systems WHERE id = ?1",
+            rusqlite::params![id],
+            |row| row.get::<_, String>(0),
+        )
+        .optional()
+        .map_err(|e| HtsError::StorageError(e.to_string()))?
+    {
+        return Ok(Some(url));
+    }
     conn.query_row(
-        "SELECT url FROM code_systems WHERE id = ?1",
+        "SELECT url FROM code_systems \
+         WHERE json_extract(resource_json, '$.id') = ?1 \
+         ORDER BY COALESCE(version, '') DESC \
+         LIMIT 1",
         rusqlite::params![id],
         |row| row.get::<_, String>(0),
     )
@@ -382,10 +411,15 @@ pub(crate) fn invalidate_expansion_cache_for_system(
 }
 
 /// Delete a CodeSystem and all its normalized data by its FHIR resource `id`.
+///
+/// Multi-version: matches both the synthetic storage id (`<id>|<version>`)
+/// and the original FHIR id captured in `resource_json.id`, so a CRUD DELETE
+/// `/CodeSystem/version` removes every stored version of that resource.
 #[cfg(feature = "sqlite")]
 pub(crate) fn delete_code_system(conn: &Connection, id: &str) -> Result<(), HtsError> {
     conn.execute(
-        "DELETE FROM code_systems WHERE id = ?1",
+        "DELETE FROM code_systems \
+         WHERE id = ?1 OR json_extract(resource_json, '$.id') = ?1",
         rusqlite::params![id],
     )
     .map_err(|e| HtsError::StorageError(e.to_string()))?;
@@ -420,6 +454,21 @@ fn utc_now() -> String {
     chrono::Utc::now().to_rfc3339()
 }
 
+/// Build a multi-version-safe storage id for a CodeSystem.
+///
+/// The HTS schema permits multiple `code_systems` rows that share a canonical
+/// `url` provided each row has a distinct `version`. Tx-ecosystem fixtures
+/// frequently ship the same FHIR `id` (e.g. `"version"`) for every version of
+/// a CodeSystem, so a 1:1 use of `id` would collide on the PK. Suffixing the
+/// version makes the storage id deterministic per (url, version) without
+/// forcing callers to thread the URL through.
+pub(crate) fn storage_id_for(fhir_id: &str, version: Option<&str>) -> String {
+    match version {
+        Some(v) if !v.is_empty() => format!("{fhir_id}|{v}"),
+        _ => fhir_id.to_owned(),
+    }
+}
+
 // ── Tests ──────────────────────────────────────────────────────────────────────
 
 #[cfg(all(test, feature = "sqlite"))]
@@ -557,6 +606,89 @@ mod tests {
         assert!(result.is_err());
     }
 
+    /// Two CodeSystems sharing a canonical URL but declaring distinct
+    /// `version` values (and the same FHIR `id`) must coexist.
+    ///
+    /// Mirrors `tx-ecosystem/tests/version/codesystem-version-{1,2}.json`,
+    /// which both ship `"id":"version"` + the same `url`. The legacy
+    /// `UNIQUE(url)` constraint dropped one of them; the new composite
+    /// `(url, version)` index lets both survive.
+    #[tokio::test]
+    async fn import_two_versions_same_url_keeps_both() {
+        let b = backend();
+        let ctx = ctx();
+
+        let bundle = r#"{
+          "resourceType": "Bundle",
+          "type": "collection",
+          "entry": [
+            {
+              "resource": {
+                "resourceType": "CodeSystem",
+                "id": "version",
+                "url": "http://example.org/cs/multi",
+                "version": "1.0.0",
+                "status": "active",
+                "content": "complete",
+                "concept": [{ "code": "code1", "display": "Display 1 (1.0)" }]
+              }
+            },
+            {
+              "resource": {
+                "resourceType": "CodeSystem",
+                "id": "version",
+                "url": "http://example.org/cs/multi",
+                "version": "1.2.0",
+                "status": "active",
+                "content": "complete",
+                "concept": [
+                  { "code": "code1", "display": "Display 1 (1.2)" },
+                  { "code": "code3", "display": "Display 3 (1.2)" }
+                ]
+              }
+            }
+          ]
+        }"#;
+
+        let stats = b.import_bundle(&ctx, bundle.as_bytes()).await.unwrap();
+        assert_eq!(stats.code_systems, 2);
+        assert!(
+            stats.errors.is_empty(),
+            "no errors expected, got: {:?}",
+            stats.errors
+        );
+
+        let conn = b.pool().get().unwrap();
+        let row_count: i64 = conn
+            .query_row(
+                "SELECT COUNT(*) FROM code_systems WHERE url = 'http://example.org/cs/multi'",
+                [],
+                |r| r.get(0),
+            )
+            .unwrap();
+        assert_eq!(row_count, 2, "both versions must coexist");
+
+        // Each version owns its own concept set.
+        let v1_concepts: i64 = conn
+            .query_row(
+                "SELECT COUNT(*) FROM concepts c JOIN code_systems s ON c.system_id = s.id \
+                 WHERE s.url = 'http://example.org/cs/multi' AND s.version = '1.0.0'",
+                [],
+                |r| r.get(0),
+            )
+            .unwrap();
+        let v2_concepts: i64 = conn
+            .query_row(
+                "SELECT COUNT(*) FROM concepts c JOIN code_systems s ON c.system_id = s.id \
+                 WHERE s.url = 'http://example.org/cs/multi' AND s.version = '1.2.0'",
+                [],
+                |r| r.get(0),
+            )
+            .unwrap();
+        assert_eq!(v1_concepts, 1);
+        assert_eq!(v2_concepts, 2);
+    }
+
     #[tokio::test]
     async fn hierarchy_materialized_from_nesting() {
         let b = backend();
@@ -565,14 +697,22 @@ mod tests {
             .await
             .unwrap();
 
+        // Multi-version storage_id is opaque, so resolve it via URL first.
         let conn = b.pool().get().unwrap();
-        let count: i64 = conn
+        let system_id: String = conn
             .query_row(
-                "SELECT COUNT(*) FROM concept_hierarchy WHERE system_id='cs-test'",
+                "SELECT id FROM code_systems WHERE url = 'http://example.org/cs'",
                 [],
                 |r| r.get(0),
             )
             .unwrap();
+        let count: i64 = conn
+            .query_row(
+                "SELECT COUNT(*) FROM concept_hierarchy WHERE system_id = ?1",
+                [&system_id],
+                |r| r.get(0),
+            )
+            .unwrap();
         // A→B and B→C
         assert_eq!(count, 2, "Two hierarchy edges should be materialized");
     }
diff --git a/crates/hts/src/operations/crud.rs b/crates/hts/src/operations/crud.rs
index 33384ffce..9c03d3120 100644
--- a/crates/hts/src/operations/crud.rs
+++ b/crates/hts/src/operations/crud.rs
@@ -959,21 +959,25 @@ mod tests {
             .unwrap();
         assert_eq!(resp.status(), StatusCode::CREATED);
 
-        // Verify HTS normalized tables were populated.
+        // Verify HTS normalized tables were populated. The synthetic storage
+        // id (`<fhir-id>|<version>`) is opaque, so we look up by URL.
         let conn = hts_pool.get().unwrap();
-        let count: i64 = conn
+        let storage_id: String = conn
             .query_row(
-                "SELECT COUNT(*) FROM code_systems WHERE id = 'cs-index'",
+                "SELECT id FROM code_systems WHERE url = 'http://example.org/cs/cs-index'",
                 [],
                 |r| r.get(0),
             )
             .unwrap();
-        assert_eq!(count, 1, "code_systems row should be created");
+        assert!(
+            storage_id.starts_with("cs-index"),
+            "storage id should be derived from FHIR id, got {storage_id}"
+        );
 
         let concept_count: i64 = conn
             .query_row(
-                "SELECT COUNT(*) FROM concepts WHERE system_id = 'cs-index'",
-                [],
+                "SELECT COUNT(*) FROM concepts WHERE system_id = ?1",
+                [&storage_id],
                 |r| r.get(0),
             )
             .unwrap();
@@ -1028,11 +1032,12 @@ mod tests {
             .unwrap();
         assert_eq!(del.status(), StatusCode::NO_CONTENT);
 
-        // Verify normalized rows are gone.
+        // Verify normalized rows are gone. Match by URL to avoid coupling to
+        // the synthetic storage-id format.
         let conn = hts_pool.get().unwrap();
         let cs_count: i64 = conn
             .query_row(
-                "SELECT COUNT(*) FROM code_systems WHERE id = 'cs-del'",
+                "SELECT COUNT(*) FROM code_systems WHERE url = 'http://example.org/cs/cs-del'",
                 [],
                 |r| r.get(0),
             )
@@ -1041,7 +1046,7 @@ mod tests {
 
         let concept_count: i64 = conn
             .query_row(
-                "SELECT COUNT(*) FROM concepts WHERE system_id = 'cs-del'",
+                "SELECT COUNT(*) FROM concepts WHERE system_id LIKE 'cs-del%'",
                 [],
                 |r| r.get(0),
             )