Stabilize streaming runtime completion

duck_core/api.py

@@ -207,6 +207,7 @@ def create_app() -> FastAPI:
             body.debug,
             history_messages=history,
             memory_records=memory_records,
+            skill_summary=await selected_skill_summary(body.message),
         )
         await conversations.add_message(
             conversation.conversation_id,
@@ -252,6 +253,21 @@ def create_app() -> FastAPI:
             if record.text
         ]
 
+    async def selected_skill_summary(user_request: str) -> str | None:
+        candidates = await skills.find_candidate_skills(user_request, limit=1)
+        if not candidates:
+            return None
+        skill = candidates[0].skill
+        parts = [
+            f"{skill.id}: {skill.title}",
+            skill.description,
+        ]
+        if skill.procedure:
+            parts.append("Procedure:\n" + skill.procedure.strip())
+        if skill.success_criteria:
+            parts.append("Success criteria:\n- " + "\n- ".join(skill.success_criteria))
+        return "\n\n".join(part for part in parts if part)
+
     @app.get("/v1/conversations")
     async def list_conversations() -> list[dict[str, Any]]:
         return [conversation.model_dump() for conversation in await conversations.list()]
@@ -327,7 +343,10 @@ def create_app() -> FastAPI:
             generation_stats = GenerationStats()
             try:
                 messages = await runtime.context_builder.build_async_messages(
-                    task, history, memory_records
+                    task,
+                    history,
+                    memory_records,
+                    skill_summary=await selected_skill_summary(body.message),
                 )
                 yield runtime_status(
                     task.task_id,
@@ -454,6 +473,7 @@ def create_app() -> FastAPI:
                         "generation_stats": generation_stats.summary(),
                     },
                 )
+                asyncio.create_task(runtime.complete_postprocessing(task.task_id, content))
                 yield sse(
                     "done",
                     {
@@ -562,7 +582,10 @@ def create_app() -> FastAPI:
                 tool_observation = await runtime._run_approved_or_denied_action(
                     task_id, approval.normalized_action, approval.decision
                 )
-                messages = await runtime.context_builder.build_async_messages(task)
+                messages = await runtime.context_builder.build_async_messages(
+                    task,
+                    skill_summary=await selected_skill_summary(task.user_message),
+                )
                 tool_observations = [tool_observation]
                 yield runtime_status(
                     task_id,
@@ -682,6 +705,7 @@ def create_app() -> FastAPI:
                         "generation_stats": generation_stats.summary(),
                     },
                 )
+                asyncio.create_task(runtime.complete_postprocessing(task_id, content))
                 yield sse(
                     "done",
                     {
@@ -755,7 +779,10 @@ def create_app() -> FastAPI:
                     approval.decision,
                     password=body.password,
                 )
-                messages = await runtime.context_builder.build_async_messages(task)
+                messages = await runtime.context_builder.build_async_messages(
+                    task,
+                    skill_summary=await selected_skill_summary(task.user_message),
+                )
                 tool_observations = [tool_observation]
                 yield runtime_status(
                     task_id,
@@ -849,6 +876,7 @@ def create_app() -> FastAPI:
                         "generation_stats": generation_stats.summary(),
                     },
                 )
+                asyncio.create_task(runtime.complete_postprocessing(task_id, content))
                 yield sse(
                     "done",
                     {

duck_core/runtime_loop.py

@@ -61,6 +61,7 @@ class RuntimeLoop:
         debug: bool = False,
         history_messages: list[dict[str, str]] | None = None,
         memory_records: list[dict[str, str]] | None = None,
+        skill_summary: str | None = None,
         reflect: bool = True,
     ) -> ChatResult:
         task = await self.task_store.create_task(message, workspace, debug)
@@ -71,7 +72,7 @@ class RuntimeLoop:
         )
         try:
             messages = await self.context_builder.build_async_messages(
-                task, history_messages, memory_records
+                task, history_messages, memory_records, skill_summary=skill_summary
             )
             tool_observations = await self._run_action_loop(task.task_id, messages, workspace)
             if any(observation.get("requires_approval") for observation in tool_observations):
@@ -130,9 +131,7 @@ class RuntimeLoop:
                     "reasoning_content": response.reasoning_content,
                 },
             )
-            await self._run_memory_policy(task.task_id, response.content)
-            if reflect:
-                await self._run_reflection(task.task_id)
+            await self.complete_postprocessing(task.task_id, response.content, reflect=reflect)
             return ChatResult(
                 task_id=task.task_id,
                 status="completed",
@@ -151,6 +150,13 @@ class RuntimeLoop:
                 reasoning_content=None,
             )
 
+    async def complete_postprocessing(
+        self, task_id: str, final_response: str, reflect: bool = True
+    ) -> None:
+        await self._run_memory_policy(task_id, final_response)
+        if reflect:
+            await self._run_reflection(task_id)
+
     async def continue_after_approval(self, task_id: str, approval_id: str) -> ChatResult:
         if self.approval_service is None:
             return ChatResult(