fix(km): read back mixed enum labels safely

apps/api/src/repositories/knowledge_repository.py

@@ -337,15 +337,18 @@ class KnowledgeDBRepository:
             )
             result = await self.db.execute(stmt)
             entry_id: str = result.scalar_one()
-            # UPSERT 後用 id 重新 SELECT，確保取到完整的 ORM 物件
+            # Read through the enum-safe projection because production keeps
+            # both legacy uppercase labels and the lowercase published label.
             fetch_result = await self.db.execute(
-                select(KnowledgeEntryRecord).where(KnowledgeEntryRecord.id == entry_id)
+                select(*_knowledge_entry_columns()).where(
+                    KnowledgeEntryRecord.id == entry_id
+                )
             )
-            record = fetch_result.scalar_one()
+            record = fetch_result.mappings().one()
             logger.info(
                 "knowledge_entry_upserted",
-                entry_id=record.id,
+                entry_id=record["id"],
-                title=record.title,
+                title=record["title"],
                 path_type=data.path_type,
                 incident_id=data.related_incident_id,
             )
@@ -372,9 +375,19 @@ class KnowledgeDBRepository:
         )
         self.db.add(record)
         await self.db.flush()
-        await self.db.refresh(record)
+        entry_id = str(record.id)
-        logger.info("knowledge_entry_created", entry_id=record.id, title=record.title)
+        fetch_result = await self.db.execute(
-        return self._to_model(record)
+            select(*_knowledge_entry_columns()).where(
+                KnowledgeEntryRecord.id == entry_id
+            )
+        )
+        readback = fetch_result.mappings().one()
+        logger.info(
+            "knowledge_entry_created",
+            entry_id=readback["id"],
+            title=readback["title"],
+        )
+        return self._to_model(readback)
 
     async def get_by_id(self, entry_id: str) -> KnowledgeEntry | None:
         """根據 ID 取得知識條目（排除 archived）"""

apps/api/tests/test_knowledge_repository_read_model.py

@@ -7,7 +7,12 @@ import pytest
 from sqlalchemy.dialects import postgresql
 
 from src.db.models import KnowledgeEntryRecord
-from src.models.knowledge import EntrySource, EntryStatus, EntryType
+from src.models.knowledge import (
+    EntrySource,
+    EntryStatus,
+    EntryType,
+    KnowledgeEntryCreate,
+)
 from src.repositories.knowledge_repository import (
     _ASSET_TAXONOMY_KEYS,
     KnowledgeDBRepository,
@@ -277,3 +282,56 @@ def test_knowledge_enum_db_labels_match_deployed_postgresql_contract() -> None:
     assert str(_knowledge_enum_db_value(EntryStatus.PUBLISHED)) == (
         "'published'::entrystatus"
     )
+
+
+@pytest.mark.asyncio
+async def test_idempotent_create_reads_back_published_with_enum_safe_projection(
+) -> None:
+    class _CreatedId:
+        def scalar_one(self):
+            return "km-created"
+
+    class _CreateDb:
+        def __init__(self):
+            self.statements = []
+            self.results = [
+                _CreatedId(),
+                _MappingResult([
+                    _record(
+                        id="km-created",
+                        entry_type="ANTI_PATTERN",
+                        source="AI_EXTRACTED",
+                        status="published",
+                        related_incident_id="INC-TEST-ENUM",
+                        path_type="ansible_no_write_replay:test",
+                    ).__dict__
+                ]),
+            ]
+
+        async def execute(self, statement):
+            self.statements.append(statement)
+            return self.results.pop(0)
+
+    db = _CreateDb()
+    entry = await KnowledgeDBRepository(db=db).create(  # type: ignore[arg-type]
+        KnowledgeEntryCreate(
+            title="Enum-safe writeback",
+            content="Durable AI learning receipt",
+            entry_type=EntryType.ANTI_PATTERN,
+            category="AI automation",
+            source=EntrySource.AI_EXTRACTED,
+            status=EntryStatus.PUBLISHED,
+            related_incident_id="INC-TEST-ENUM",
+            path_type="ansible_no_write_replay:test",
+        )
+    )
+
+    assert entry.status is EntryStatus.PUBLISHED
+    readback_sql = str(
+        db.statements[1].compile(
+            dialect=postgresql.dialect(),
+            compile_kwargs={"literal_binds": True},
+        )
+    ).lower()
+    assert "cast(knowledge_entries.status as varchar)" in readback_sql
+    assert "cast(knowledge_entries.entry_type as varchar)" in readback_sql