feat(phase8 #80 D8.6 chunk-2): wire UIMessageStore.write live + drop snapshot_assembler / agent_artifact

aperag/db/repositories/agent_runtime.py

@@ -18,7 +18,6 @@
 
 from aperag.db.repositories.base import AsyncRepositoryProtocol
 from aperag.domains.agent_runtime.db.models import (
-    AgentArtifact,
     AgentTimelineEvent,
     AgentTurn,
 )
@@ -182,45 +181,3 @@ async def _query(session):
             return result.scalars().all()
 
         return await self._execute_query(_query)
-
-    async def create_agent_artifact(
-        self,
-        *,
-        turn_id: str,
-        artifact_type,
-        summary: Optional[str],
-        payload: dict,
-        storage_ref: Optional[str] = None,
-    ) -> AgentArtifact:
-        async def _operation(session):
-            instance = AgentArtifact(
-                turn_id=turn_id,
-                artifact_type=artifact_type,
-                summary=summary,
-                payload=payload,
-                storage_ref=storage_ref,
-            )
-            session.add(instance)
-            await session.flush()
-            await session.refresh(instance)
-            return instance
-
-        return await self.execute_with_transaction(_operation)
-
-    async def query_agent_artifact(self, artifact_id: str) -> Optional[AgentArtifact]:
-        async def _query(session):
-            stmt = select(AgentArtifact).where(AgentArtifact.id == artifact_id)
-            result = await session.execute(stmt)
-            return result.scalars().first()
-
-        return await self._execute_query(_query)
-
-    async def query_agent_artifacts_by_turn(self, turn_id: str) -> list[AgentArtifact]:
-        async def _query(session):
-            stmt = (
-                select(AgentArtifact).where(AgentArtifact.turn_id == turn_id).order_by(AgentArtifact.gmt_created.asc())
-            )
-            result = await session.execute(stmt)
-            return result.scalars().all()
-
-        return await self._execute_query(_query)

aperag/domains/agent_runtime/api/routes.py

@@ -42,7 +42,6 @@
 from aperag.domains.agent_runtime.db.models import AgentEventActor, AgentTurnStatus
 from aperag.domains.agent_runtime.runtime import agent_runtime_manager as runtime_manager
 from aperag.domains.agent_runtime.schemas import (
-    AgentArtifactEnvelope,
     CancelTurnResponse,
     CreateTurnRequest,
     CreateTurnResponse,
@@ -154,13 +153,6 @@ async def cancel_turn_view(
     return CancelTurnResponse(turn_id=turn_id, status=AgentTurnStatus.CANCELLED)
 
 
-@router.get("/agent/artifacts/{artifact_id}")
-async def get_artifact_view(
-    artifact_id: str, user: AuthenticatedUser = Depends(required_user)
-) -> AgentArtifactEnvelope:
-    return await runtime_manager.artifact_service.get_artifact_for_user(str(user.id), artifact_id)
-
-
 # -- D8.3 (#75) consent + elicitation HTTP surfaces ---------------
 
 

aperag/domains/agent_runtime/db/models.py

@@ -14,18 +14,21 @@
 
 """Agent-runtime-domain SQLAlchemy models.
 
-Owns ``AgentTurn`` + ``AgentTimelineEvent`` + ``AgentArtifact`` —
-the three entities the agent runtime is responsible for — plus
-``AgentTurnStatus`` / ``AgentEventActor`` / ``AgentArtifactType``
-enums. Moved here from ``aperag.db.models`` in Phase 5 step 5-S2b;
-the legacy aggregate module retains a re-export shim until Phase 6
-cleanup.
-
-All three tables key on ``turn_id`` (events + artifacts) or
-``chat_id`` (turn); these are opaque string FKs, so no Python-level
-cross-domain import is required. ``chat_id`` is logically owned by
-the ``conversation`` domain but the relationship is traversed at
-query time via the turn service, not via ORM relationship() bindings.
+Owns ``AgentTurn`` + ``AgentTimelineEvent`` + ``AgentMessage`` plus
+``AgentTurnStatus`` / ``AgentEventActor`` enums. Moved here from
+``aperag.db.models`` in Phase 5 step 5-S2b; the legacy aggregate
+module retains a re-export shim until Phase 6 cleanup.
+
+Phase 8 D8.6 (#80) chunk-2 hard-cut removed the legacy
+``AgentArtifact`` table and the ``AgentArtifactType`` enum — at-rest
+answer/citation persistence is now canonical via ``AgentMessage``
+(D8.2 #74) populated by the runtime emit path.
+
+Tables key on ``turn_id`` (events + messages) or ``chat_id`` (turn);
+these are opaque string FKs, so no Python-level cross-domain import
+is required. ``chat_id`` is logically owned by the ``conversation``
+domain but the relationship is traversed at query time via the turn
+service, not via ORM relationship() bindings.
 """
 
 from __future__ import annotations
@@ -83,14 +86,6 @@ class AgentEventActor(str, Enum):
     SYSTEM = "system"
 
 
-class AgentArtifactType(str, Enum):
-    ANSWER = "answer"
-    REFERENCE_BUNDLE = "reference_bundle"
-    TOOL_RESULT_SUMMARY = "tool_result_summary"
-    SEARCH_RESULT_SUMMARY = "search_result_summary"
-    ERROR_SUMMARY = "error_summary"
-
-
 class AgentTurn(Base):
     __tablename__ = "agent_turn"
     __table_args__ = (
@@ -110,8 +105,6 @@ class AgentTurn(Base):
     model_profile = Column(JSON, default=lambda: {}, nullable=False)
     error_code = Column(String(128), nullable=True)
     error_message = Column(Text, nullable=True)
-    answer_artifact_id = Column(String(24), nullable=True, index=True)
-    reference_bundle_artifact_id = Column(String(24), nullable=True, index=True)
     timeline_cursor = Column(Integer, default=0, nullable=False)
     gmt_created = Column(DateTime(timezone=True), default=utc_now, nullable=False)
     gmt_started = Column(DateTime(timezone=True), nullable=True)
@@ -138,20 +131,6 @@ class AgentTimelineEvent(Base):
     gmt_created = Column(DateTime(timezone=True), default=utc_now, nullable=False)
 
 
-class AgentArtifact(Base):
-    __tablename__ = "agent_artifact"
-    __table_args__ = (Index("idx_agent_artifact_turn_type", "turn_id", "artifact_type"),)
-
-    id = Column(String(24), primary_key=True, default=lambda: "art" + _random_id())
-    turn_id = Column(String(24), nullable=False, index=True)
-    artifact_type = Column(_enum_column(AgentArtifactType), nullable=False, index=True)
-    summary = Column(Text, nullable=True)
-    payload = Column(JSON, default=lambda: {}, nullable=False)
-    storage_ref = Column(Text, nullable=True)
-    gmt_created = Column(DateTime(timezone=True), default=utc_now, nullable=False)
-    gmt_updated = Column(DateTime(timezone=True), default=utc_now, nullable=False)
-
-
 class AgentMessage(Base):
     """At-rest UIMessage envelope (Phase 8 D8.2 first-cut, D8.5-BE refined).
 
@@ -170,10 +149,11 @@ class AgentMessage(Base):
     semantics independent of runtime origin per Weston msg=94dac98a /
     architect canonical lock msg=e01e9b4b.
 
-    The legacy ``AgentArtifact`` / ``AgentTimelineEvent`` tables are
-    retained alongside this table during D8.x rollout — they will be
-    dropped in D8.6 (#80) once D8.4 FE renderer is consuming
-    ``AgentMessage`` exclusively.
+    Phase 8 D8.6 (#80) chunk-2 dropped the legacy ``AgentArtifact``
+    table and column FKs from ``AgentTurn``; ``AgentMessage`` is now
+    the single authoritative at-rest store for assistant turn
+    contents. ``AgentTimelineEvent`` remains pending chunk-3 cleanup
+    (replay/reload semantic change is reviewed separately).
     """
 
     __tablename__ = "agent_message"

aperag/domains/agent_runtime/runtime.py

@@ -26,8 +26,9 @@
 from pydantic_ai.models.openai import OpenAIChatModel
 from pydantic_ai.providers.openai import OpenAIProvider
 
+from aperag.config import get_async_session
 from aperag.db.ops import async_db_ops
-from aperag.domains.agent_runtime.db.models import AgentArtifactType, AgentEventActor, AgentTurnStatus
+from aperag.domains.agent_runtime.db.models import AgentEventActor, AgentTurnStatus
 from aperag.domains.agent_runtime.ports import PromptTemplateOps
 from aperag.domains.agent_runtime.schemas import (
     AgentMessage,
@@ -36,13 +37,22 @@
     VisibleAgentState,
 )
 from aperag.domains.agent_runtime.services import (
-    ArtifactService,
     EventService,
     HistoryWriter,
     TurnService,
     _parse_bot_config,
 )
-from aperag.domains.agent_runtime.storage import DEFAULT_AGENT_TURN_LEASE_TTL_SECONDS
+from aperag.domains.agent_runtime.storage import DEFAULT_AGENT_TURN_LEASE_TTL_SECONDS, AgentRuntimeRedisStore
+from aperag.domains.agent_runtime.uimessage import (
+    CitationData,
+    DataCitationPart,
+    SourceUrlPart,
+    TextPart,
+    UIMessage,
+    UIMessagePart,
+    UrlCitationLocation,
+)
+from aperag.domains.agent_runtime.uimessage_store import UIMessageDbOps, UIMessageStore
 from aperag.domains.knowledge_base.schemas import Collection as KBCollectionSchema
 from aperag.domains.model_platform.schemas import ModelCapability
 from aperag.exceptions import ResourceNotFoundException, ValidationException
@@ -82,6 +92,43 @@ def _get_prompt_template_ops() -> PromptTemplateOps:
 DEFAULT_AGENT_TURN_LEASE_RENEW_INTERVAL_SECONDS = int(os.getenv("APERAG_AGENT_TURN_LEASE_RENEW_INTERVAL_SECONDS", "30"))
 
 
+def _compose_assistant_parts(
+    *,
+    turn_id: str,
+    answer_text: str,
+    references: list[ReferenceBundleItem],
+) -> list[UIMessagePart]:
+    """Project the runtime's accumulated answer + references into a
+    canonical at-rest ``UIMessagePart`` list for ``UIMessageStore.write``.
+
+    Order: ``TextPart`` (answer) first, then ``SourceUrlPart`` /
+    ``DataCitationPart`` pairs from each reference. Items without a
+    URI surface as ``DataCitationPart`` only — same shape the FE
+    renderer expects from the live SSE stream (D8 §2 byte-equal).
+    """
+
+    parts: list[UIMessagePart] = []
+    if answer_text:
+        parts.append(TextPart(text=answer_text))
+    for index, ref in enumerate(references):
+        metadata = ref.metadata if isinstance(ref.metadata, dict) else {}
+        url = ref.uri or metadata.get("url")
+        title = ref.title or metadata.get("title")
+        snippet = ref.snippet or ""
+        source_id = ref.source_id or f"{turn_id}-ref-{index}"
+        if url:
+            parts.append(SourceUrlPart(source_id=str(source_id), url=str(url), title=title))
+        parts.append(
+            DataCitationPart(
+                data=CitationData(
+                    cited_text=str(snippet),
+                    location=UrlCitationLocation(url=str(url) if url else "", title=title),
+                )
+            )
+        )
+    return parts
+
+
 @dataclass
 class ResolvedAgentRequest:
     agent_message: AgentMessage
@@ -181,12 +228,15 @@ async def cancel_turn(self, turn_id: str) -> None:
 class AgentRuntimeTaskManager:
     def __init__(self):
         self.tasks: dict[str, asyncio.Task] = {}
-        self.turn_service = TurnService()
+        self.uimessage_store = UIMessageStore(
+            db_ops=UIMessageDbOps(session_factory=get_async_session),
+            redis_store=AgentRuntimeRedisStore(),
+        )
+        self.turn_service = TurnService(uimessage_store=self.uimessage_store)
         self.event_service = EventService()
-        self.artifact_service = ArtifactService()
-        self.history_writer = HistoryWriter()
+        self.history_writer = HistoryWriter(uimessage_store=self.uimessage_store)
         self.runtime: AgentRuntime = PydanticAIRuntime(
-            self.turn_service, self.event_service, self.artifact_service, self.history_writer
+            self.turn_service, self.event_service, self.history_writer, self.uimessage_store
         )
 
     async def claim_turn(self, turn_id: str) -> str | None:
@@ -245,13 +295,13 @@ def __init__(
         self,
         turn_service: TurnService,
         event_service: EventService,
-        artifact_service: ArtifactService,
         history_writer: HistoryWriter,
+        uimessage_store: UIMessageStore,
     ):
         self.turn_service = turn_service
         self.event_service = event_service
-        self.artifact_service = artifact_service
         self.history_writer = history_writer
+        self.uimessage_store = uimessage_store
 
     async def cancel_turn(self, turn_id: str) -> None:
         await self.turn_service.redis_store.mark_cancelled(turn_id)
@@ -434,38 +484,19 @@ async def emit(
 
             lease_guard.ensure_owned()
             answer_text = "".join(text_chunks).strip()
-            answer_artifact = await self.artifact_service.create_artifact(
+
+            await self.uimessage_store.write(
                 turn_id=turn.id,
-                artifact_type=AgentArtifactType.ANSWER,
-                summary=answer_text[:200] if answer_text else "",
-                payload={"text": answer_text, "query": request.query},
+                chat_id=chat.id,
+                message=UIMessage(
+                    id=f"msg-{turn.id}",
+                    role="assistant",
+                    parts=_compose_assistant_parts(
+                        turn_id=turn.id, answer_text=answer_text, references=reference_items
+                    ),
+                ),
             )
-            reference_artifact = None
-            if reference_items:
-                reference_artifact = await self.artifact_service.create_artifact(
-                    turn_id=turn.id,
-                    artifact_type=AgentArtifactType.REFERENCE_BUNDLE,
-                    summary=f"{len(reference_items)} references",
-                    payload={"items": [item.model_dump() for item in reference_items]},
-                )
-                await emit(
-                    "artifact.created",
-                    actor=AgentEventActor.SYSTEM,
-                    label="reference_bundle",
-                    status="ready",
-                    data={
-                        "artifact_id": reference_artifact.artifact_id,
-                        "artifact_type": reference_artifact.artifact_type,
-                    },
-                )
 
-            await emit(
-                "artifact.created",
-                actor=AgentEventActor.SYSTEM,
-                label="answer",
-                status="ready",
-                data={"artifact_id": answer_artifact.artifact_id, "artifact_type": answer_artifact.artifact_type},
-            )
             await emit(
                 "agent.state.changed",
                 actor=AgentEventActor.AGENT,
@@ -478,12 +509,7 @@ async def emit(
                 label=VisibleAgentState.COMPLETED.value,
                 status="completed",
             )
-            await self.turn_service.mark_completed(
-                turn.id,
-                answer_artifact_id=answer_artifact.artifact_id,
-                reference_bundle_artifact_id=reference_artifact.artifact_id if reference_artifact else None,
-                sequence=sequence,
-            )
+            await self.turn_service.mark_completed(turn.id, sequence=sequence)
             await self.history_writer.commit_completed_turn(
                 turn=turn,
                 request=request,
@@ -522,19 +548,6 @@ async def emit(
                 )
                 return
             logger.exception("Agent Runtime V3 turn failed: %s", turn.id)
-            error_artifact = await self.artifact_service.create_artifact(
-                turn_id=turn.id,
-                artifact_type=AgentArtifactType.ERROR_SUMMARY,
-                summary=str(exc),
-                payload={"message": str(exc), "type": exc.__class__.__name__},
-            )
-            await emit(
-                "artifact.created",
-                actor=AgentEventActor.SYSTEM,
-                label="error_summary",
-                status="ready",
-                data={"artifact_id": error_artifact.artifact_id, "artifact_type": error_artifact.artifact_type},
-            )
             await emit(
                 "agent.state.changed",
                 actor=AgentEventActor.AGENT,