asset listing filters

tests/test_asset_registry.py

@@ -279,6 +279,80 @@ def test_asset_registry_applies_persisted_metadata_schema_assignments() -> None:
     assert service.list_metadata_schema_assignments()[0].policy_ref == "local://metadata-policy/policy-note"
 
 
+def test_asset_registry_filters_assets_by_standard_metadata_and_records() -> None:
+    repo = InMemoryAssetRegistryRepository()
+    service = AssetRegistryService(repo)
+    context = operation_context()
+
+    service.create_asset(
+        "Architecture ADR",
+        Classification(
+            asset_type="document",
+            sensitivity=Sensitivity.PUBLIC,
+            topics=("architecture", "adr"),
+            owner="Platform Knowledge",
+            review_state="approved",
+        ),
+        context,
+        asset_id="asset-adr",
+        metadata_records=[
+            MetadataRecord("status", "accepted", confirmed=True),
+            MetadataRecord("tags", ["governance", "markdown"], confirmed=True),
+        ],
+    )
+    service.create_asset(
+        "Internal Risk Note",
+        Classification(
+            asset_type="note",
+            sensitivity=Sensitivity.CONFIDENTIAL,
+            topics=("risk",),
+            owner="Security",
+            review_state="draft",
+        ),
+        context,
+        asset_id="asset-risk",
+        metadata_records=[MetadataRecord("status", "accepted", confirmed=False)],
+    )
+    service.create_asset(
+        "Architecture Brief",
+        Classification(
+            asset_type="document",
+            sensitivity=Sensitivity.INTERNAL,
+            topics=("architecture",),
+            owner="Platform Knowledge",
+            review_state="draft",
+        ),
+        context,
+        asset_id="asset-brief",
+        metadata_records=[MetadataRecord("status", "draft", confirmed=True)],
+    )
+
+    assert [asset.id for asset in service.list_assets(sensitivity=Sensitivity.PUBLIC)] == ["asset-adr"]
+    assert [asset.id for asset in service.list_assets(owner="Security")] == ["asset-risk"]
+    assert [asset.id for asset in service.list_assets(topic="architecture")] == [
+        "asset-adr",
+        "asset-brief",
+    ]
+    assert [asset.id for asset in service.list_assets(review_state="draft")] == [
+        "asset-brief",
+        "asset-risk",
+    ]
+    assert [asset.id for asset in service.list_assets(metadata_filters={"status": "accepted"})] == [
+        "asset-adr",
+        "asset-risk",
+    ]
+    assert [
+        asset.id
+        for asset in service.list_assets(
+            metadata_filters={"status": "accepted"},
+            confirmed_metadata_only=True,
+        )
+    ] == ["asset-adr"]
+    assert [asset.id for asset in service.list_assets(metadata_filters={"tags": "markdown"})] == [
+        "asset-adr"
+    ]
+
+
 def test_sqlite_asset_registry_survives_reinstantiation(tmp_path: Path) -> None:
     db_path = tmp_path / "registry.sqlite"
     repo = SQLiteAssetRegistryRepository(db_path)
@@ -421,6 +495,55 @@ def test_sqlite_registry_persists_metadata_schemas_and_assignments(tmp_path: Pat
     assert reloaded_repo.get_metadata_schema_assignment("assignment-review-documents").schema_id == "schema-review-v1"
 
 
+def test_sqlite_registry_filters_assets_after_reload(tmp_path: Path) -> None:
+    db_path = tmp_path / "registry.sqlite"
+    repo = SQLiteAssetRegistryRepository(db_path)
+    service = AssetRegistryService(repo)
+    context = operation_context()
+
+    service.create_asset(
+        "Public Guide",
+        Classification(
+            asset_type="guide",
+            sensitivity=Sensitivity.PUBLIC,
+            topics=("markdown", "proxy"),
+            owner="Docs",
+            review_state="approved",
+        ),
+        context,
+        asset_id="asset-guide",
+        metadata_records=[MetadataRecord("channel", "public", confirmed=True)],
+    )
+    service.create_asset(
+        "Internal Guide",
+        Classification(
+            asset_type="guide",
+            sensitivity=Sensitivity.INTERNAL,
+            topics=("markdown",),
+            owner="Docs",
+            review_state="draft",
+        ),
+        context,
+        asset_id="asset-internal-guide",
+        metadata_records=[MetadataRecord("channel", "public", confirmed=False)],
+    )
+
+    reloaded = AssetRegistryService(SQLiteAssetRegistryRepository(db_path))
+
+    assert [asset.id for asset in reloaded.list_assets(asset_type="guide", owner="Docs")] == [
+        "asset-internal-guide",
+        "asset-guide",
+    ]
+    assert [asset.id for asset in reloaded.list_assets(topic="proxy")] == ["asset-guide"]
+    assert [
+        asset.id
+        for asset in reloaded.list_assets(
+            metadata_filters={"channel": "public"},
+            confirmed_metadata_only=True,
+        )
+    ] == ["asset-guide"]
+
+
 def test_sqlite_registry_enforces_representation_asset_reference(tmp_path: Path) -> None:
     repo = SQLiteAssetRegistryRepository(tmp_path / "registry.sqlite")
     representation = AssetRepresentation.from_content(