diff --git a/backend/app/domains/products/diff.py b/backend/app/domains/products/diff.py
index 556e450..2779120 100644
--- a/backend/app/domains/products/diff.py
+++ b/backend/app/domains/products/diff.py
@@ -21,10 +21,11 @@ from __future__ import annotations
 
 import hashlib
 import json
-from dataclasses import dataclass, field
+from dataclasses import dataclass, field, replace
 from decimal import Decimal
 
-from .models import CLEAR_DEFAULTS, CanonicalProduct, CanonicalVariant
+from . import hosted
+from .models import CLEAR_DEFAULTS, CanonicalProduct, CanonicalVariant, RowError
 
 
 @dataclass
@@ -119,6 +120,7 @@ def compute_diff(catalog: dict[str, CatalogProduct], products: list[CanonicalPro
     summary = {"adds": 0, "updates": 0, "unchanged": 0, "errors": 0}
     for product in products:
         current = catalog.get(product.handle)
+        _resolve_hosted_images(product, current)
         if product.errors:
             product_plan = ProductPlan(kind="error", canonical=product, catalog=current)
             detail: dict = {"errors": [e.as_json() for e in product.errors]}
@@ -145,6 +147,31 @@ def compute_diff(catalog: dict[str, CatalogProduct], products: list[CanonicalPro
     return DiffResult(records=records, summary=summary, fingerprint=fingerprint, plan=plan)
 
 
+def _resolve_hosted_images(product: CanonicalProduct, current: CatalogProduct | None) -> None:
+    """Recognize re-imported hosted image URLs (INV-12). A hosted URL is resolved
+    to this product's existing image with that id — re-keyed onto that image's
+    source_url so the by_src match treats it as the existing image (never an add,
+    never a fetch). A hosted URL whose id is not one of this product's images
+    (cross-storefront, deleted, or a brand-new product) is a row error."""
+    current_by_id = {i.id: i for i in current.images} if current else {}
+    resolved: list = []
+    for image in product.images:
+        hosted_id = hosted.parse_image_id(image.source_url)
+        if hosted_id is None:
+            resolved.append(image)
+            continue
+        match = current_by_id.get(hosted_id)
+        if match is None:
+            product.errors.append(
+                RowError(image.line_number, "Image Src",
+                         f"image '{image.source_url}' refers to an unknown hosted id")
+            )
+            resolved.append(image)
+            continue
+        resolved.append(replace(image, source_url=match.source_url))
+    product.images = resolved
+
+
 def _add_plan(product: CanonicalProduct) -> ProductPlan:
     return ProductPlan(
         kind="add",
diff --git a/backend/tests/test_products_diff.py b/backend/tests/test_products_diff.py
index c7ccc86..2287741 100644
--- a/backend/tests/test_products_diff.py
+++ b/backend/tests/test_products_diff.py
@@ -131,6 +131,56 @@ def test_blank_position_cell_updates_to_file_order():
     assert {"field": "position", "before": 2, "after": 1} in ventry["changes"]
 
 
+def _catalog_lamp_with_image():
+    # A well-formed single-variant product carrying one fetched image id=55.
+    fields = {
+        "title": "Lamp", "description_html": None, "vendor": "Acme",
+        "product_type": "standalone", "google_product_category": None,
+        "tags": ["home"], "status": "active", "published": True,
+    }
+    return {
+        "lamp": CatalogProduct(
+            id=1, handle="lamp", title="Lamp",
+            option_names=(None, None, None), fields=fields,
+            variants=[CatalogVariant(id=10, options=(None, None, None), position=1,
+                                     fields={"sku": "SKU-L", "barcode": None, "price": Decimal("30.00"),
+                                             "cost": None, "weight": None, "weight_unit": None,
+                                             "volume": None, "volume_unit": None, "tax_id_1": None,
+                                             "tax_id_2": None, "inventory_tracker": None,
+                                             "inventory_qty": 5, "variant_image": None})],
+            images=[CatalogImage(id=55, source_url="https://m.example/a.png", position=1,
+                                 alt_text=None, status="fetched")],
+        )
+    }
+
+
+LAMP_HEADER = "Handle,Title,Vendor,Tags,Status,Variant SKU,Variant Price,Variant Inventory Qty,Image Src,Image Position"
+LAMP_ROW_PREFIX = "lamp,Lamp,Acme,home,active,SKU-L,30.00,5,"
+
+
+def test_hosted_url_for_existing_image_is_unchanged_not_add():
+    # Catalog product "lamp" has a fetched image id=55; canonical re-imports it
+    # as the hosted URL -> must be 'unchanged', never 'add'/'update'.
+    diff = compute_diff(
+        _catalog_lamp_with_image(),
+        _canon(LAMP_HEADER, LAMP_ROW_PREFIX + "/api/products/images/55/detail,1"),
+    )
+    kinds = {r["handle"]: r["kind"] for r in diff.records}
+    assert kinds["lamp"] == "unchanged"
+
+
+def test_hosted_url_for_unknown_id_is_row_error():
+    # Catalog "lamp" has NO images; canonical references /images/999/detail -> error.
+    catalog = _catalog_lamp_with_image()
+    catalog["lamp"].images = []
+    diff = compute_diff(
+        catalog,
+        _canon(LAMP_HEADER, LAMP_ROW_PREFIX + "/api/products/images/999/detail,1"),
+    )
+    kinds = {r["handle"]: r["kind"] for r in diff.records}
+    assert kinds["lamp"] == "error"
+
+
 def test_error_product_classifies_error():
     diff = compute_diff({}, _canon("Handle,Title,Variant Price", "mug,Mug,nope"))
     [rec] = diff.records
diff --git a/backend/tests/test_products_serialize.py b/backend/tests/test_products_serialize.py
index 3dc56a4..a27685b 100644
--- a/backend/tests/test_products_serialize.py
+++ b/backend/tests/test_products_serialize.py
@@ -166,18 +166,23 @@ def _gen_catalog(seed: int) -> dict:
                 fields={"sku": f"SKU-{pid}", "variant_image": None}))
         images = []
         for ii in range(rng.randint(0, 3)):
+            # Mix fetched and non-fetched images: a fetched image exports its
+            # hosted /images/{id}/detail URL, which the diff pre-pass resolves
+            # back to the same id -> still a no-op (INV-12 over hosted images).
+            status = "fetched" if rng.random() < 0.5 else "pending"
             images.append(CatalogImage(
                 id=pid * 10 + ii, source_url=f"https://img/{handle}-{ii}.jpg",
-                position=ii + 1, alt_text=rng.choice([None, f"alt {ii}"])))
+                position=ii + 1, alt_text=rng.choice([None, f"alt {ii}"]),
+                status=status))
         catalog[handle] = CatalogProduct(
             id=pid, handle=handle, title=fields["title"], option_names=option_names,
             fields=fields, variants=variants, images=images)
     return catalog
 
 
-def _roundtrip_diff(catalog: dict) -> diff.DiffResult:
+def _roundtrip_diff(catalog: dict, base_url: str = "") -> diff.DiffResult:
     """export → bytes → import pipeline → diff against the same catalog."""
-    text = "".join(serialize.catalog_to_csv(catalog.values()))
+    text = "".join(serialize.catalog_to_csv(catalog.values(), base_url=base_url))
     parsed = codec.parse_csv(text.encode("utf-8"))
     products = validate.build_products(parsed)
     return diff.compute_diff(catalog, products)
@@ -186,7 +191,9 @@ def _roundtrip_diff(catalog: dict) -> diff.DiffResult:
 def test_inv12_roundtrip_is_noop_over_generated_catalogs():
     for seed in range(200):
         catalog = _gen_catalog(seed)
-        result = _roundtrip_diff(catalog)
+        # A fixed base_url so fetched images export an absolute hosted URL; the
+        # diff resolves it back by id -> the round-trip stays a no-op.
+        result = _roundtrip_diff(catalog, base_url="https://shop.test")
         assert result.summary["adds"] == 0, f"seed {seed}: {result.summary}"
         assert result.summary["updates"] == 0, f"seed {seed}: {result.summary}"
         assert result.summary["errors"] == 0, f"seed {seed}: {result.summary}"