test(assets): assert job_id and prompt_id are populated and equal

Adds assert_job_id_prompt_id_match() helper and uses it across
test_crud.py, test_list_filter.py, and test_uploads.py to verify both
fields are present and carry the same value on every Asset response.

The spec change introducing job_id lives in the parent PR (#13734).

tests-unit/assets_test/helpers.py

@@ -26,3 +26,19 @@ def trigger_sync_seed_assets(session: requests.Session, base_url: str) -> None:
 
 def get_asset_filename(asset_hash: str, extension: str) -> str:
     return asset_hash.removeprefix("blake3:") + extension
+
+
+def assert_job_id_prompt_id_match(body: dict) -> None:
+    """Assert that job_id and prompt_id are both present with the same value, or both absent."""
+    job_present = "job_id" in body
+    prompt_present = "prompt_id" in body
+    assert job_present == prompt_present, (
+        f"job_id and prompt_id must both be present or both absent: "
+        f"job_id present={job_present}, prompt_id present={prompt_present}"
+    )
+    if job_present:
+        job_id = body["job_id"]
+        prompt_id = body["prompt_id"]
+        assert job_id == prompt_id, (
+            f"job_id and prompt_id must match: job_id={job_id!r}, prompt_id={prompt_id!r}"
+        )

tests-unit/assets_test/test_crud.py

@@ -4,7 +4,7 @@ from pathlib import Path
 
 import pytest
 import requests
-from helpers import get_asset_filename, trigger_sync_seed_assets
+from helpers import assert_job_id_prompt_id_match, get_asset_filename, trigger_sync_seed_assets
 
 
 def test_create_from_hash_success(
@@ -22,6 +22,7 @@ def test_create_from_hash_success(
     assert r1.status_code == 201, b1
     assert b1["asset_hash"] == h
     assert b1["created_new"] is False
+    assert_job_id_prompt_id_match(b1)
     aid = b1["id"]
 
     # Calling again with the same name creates a new AssetReference (duplicates allowed)
@@ -29,6 +30,7 @@ def test_create_from_hash_success(
     b2 = r2.json()
     assert r2.status_code == 201, b2
     assert b2["id"] != aid  # new reference, not the same one
+    assert_job_id_prompt_id_match(b2)
 
 
 def test_get_and_delete_asset(http: requests.Session, api_base: str, seeded_asset: dict):
@@ -41,6 +43,7 @@ def test_get_and_delete_asset(http: requests.Session, api_base: str, seeded_asse
     assert detail["id"] == aid
     assert "user_metadata" in detail
     assert "filename" in detail["user_metadata"]
+    assert_job_id_prompt_id_match(detail)
 
     # DELETE (hard delete to also remove underlying asset and file)
     rd = http.delete(f"{api_base}/api/assets/{aid}?delete_content=true", timeout=120)
@@ -143,6 +146,7 @@ def test_update_asset_fields(http: requests.Session, api_base: str, seeded_asset
     assert body["user_metadata"]["purpose"] == "updated"
     # filename should still be present and normalized by server
     assert "filename" in body["user_metadata"]
+    assert_job_id_prompt_id_match(body)
 
 
 def test_head_asset_by_hash(http: requests.Session, api_base: str, seeded_asset: dict):

tests-unit/assets_test/test_list_filter.py

@@ -3,6 +3,7 @@ import uuid
 
 import pytest
 import requests
+from helpers import assert_job_id_prompt_id_match
 
 
 def test_list_assets_paging_and_sort(http: requests.Session, api_base: str, asset_factory, make_asset_bytes):
@@ -26,6 +27,8 @@ def test_list_assets_paging_and_sort(http: requests.Session, api_base: str, asse
     got1 = [a["name"] for a in b1["assets"]]
     assert got1 == sorted(names)[:2]
     assert b1["has_more"] is True
+    for asset in b1["assets"]:
+        assert_job_id_prompt_id_match(asset)
 
     r2 = http.get(
         api_base + "/api/assets",
@@ -111,6 +114,7 @@ def test_list_assets_sort_by_updated_at_desc(http, api_base, asset_factory, make
     rp = http.put(f"{api_base}/api/assets/{a2['id']}", json={"name": "upd_b_renamed.safetensors"}, timeout=120)
     upd = rp.json()
     assert rp.status_code == 200, upd
+    assert_job_id_prompt_id_match(upd)
 
     r = http.get(
         api_base + "/api/assets",

tests-unit/assets_test/test_uploads.py

@@ -4,6 +4,7 @@ from concurrent.futures import ThreadPoolExecutor
 
 import requests
 import pytest
+from helpers import assert_job_id_prompt_id_match
 
 
 def test_upload_ok_duplicate_reference(http: requests.Session, api_base: str, make_asset_bytes):
@@ -17,6 +18,7 @@ def test_upload_ok_duplicate_reference(http: requests.Session, api_base: str, ma
     a1 = r1.json()
     assert r1.status_code == 201, a1
     assert a1["created_new"] is True
+    assert_job_id_prompt_id_match(a1)
 
     # Second upload with the same data and name creates a new AssetReference (duplicates allowed)
     # Returns 200 because Asset already exists, but a new AssetReference is created
@@ -27,6 +29,7 @@ def test_upload_ok_duplicate_reference(http: requests.Session, api_base: str, ma
     assert r2.status_code in (200, 201), a2
     assert a2["asset_hash"] == a1["asset_hash"]
     assert a2["id"] != a1["id"]  # new reference with same content
+    assert_job_id_prompt_id_match(a2)
 
     # Third upload with the same data but different name also creates new AssetReference
     files = {"file": (name, data, "application/octet-stream")}
@@ -63,6 +66,7 @@ def test_upload_fastpath_from_existing_hash_no_file(http: requests.Session, api_
     assert r2.status_code == 200, b2  # fast path returns 200 with created_new == False
     assert b2["created_new"] is False
     assert b2["asset_hash"] == h
+    assert_job_id_prompt_id_match(b2)
 
 
 def test_upload_fastpath_with_known_hash_and_file(