feat(core): add task id for previous messages

Author: gusye1234

src/server/core/acontext_core/llm/agent/task.py

@@ -24,8 +24,28 @@ def pack_task_section(tasks: List[TaskSchema]) -> str:
     return section
 
 
-def pack_previous_messages_section(messages: list[MessageBlob]) -> str:
-    return "\n".join([m.to_string() for m in messages])
+def pack_previous_messages_section(
+    planning_task: TaskSchema | None,
+    tasks: list[TaskSchema],
+    messages: list[MessageBlob],
+) -> str:
+    task_ids = [m.task_id for m in messages]
+    mappings = {t.id: t for t in tasks}
+    task_descs = []
+    for ti in task_ids:
+        if ti is None:
+            task_descs.append("(no task linked)")
+            continue
+        elif ti in mappings:
+            task_descs.append(f"(append to task_{mappings[ti].task_order})")
+        elif planning_task is not None and ti == planning_task.id:
+            task_descs.append("(append to planning_section)")
+        else:
+            LOG.warning(f"Unknown task id: {ti}")
+            task_descs.append("(no task linked)")
+    return "\n---\n".join(
+        [f"{td}\n{m.to_string()}" for td, m in zip(task_descs, messages)]
+    )
 
 
 def pack_current_message_with_ids(messages: list[MessageBlob]) -> str:
@@ -64,8 +84,15 @@ async def task_agent_curd(
         if eil:
             return r
 
+        r = await TD.fetch_planning_task(db_session, session_id)
+        planning_section, eil = r.unpack()
+        if eil:
+            return r
+
     task_section = pack_task_section(tasks)
-    previous_messages_section = pack_previous_messages_section(previous_messages)
+    previous_messages_section = pack_previous_messages_section(
+        planning_section, tasks, previous_messages
+    )
     current_messages_section = pack_current_message_with_ids(messages)
 
     LOG.info(f"Task Section: {task_section}")

src/server/core/acontext_core/service/controller/message.py

@@ -48,11 +48,15 @@ async def process_session_pending_message(
             if eil:
                 return
             messages_data = [
-                MessageBlob(message_id=m.id, role=m.role, parts=m.parts)
+                MessageBlob(
+                    message_id=m.id, role=m.role, parts=m.parts, task_id=m.task_id
+                )
                 for m in messages
             ]
             previous_messages_data = [
-                MessageBlob(message_id=m.id, role=m.role, parts=m.parts)
+                MessageBlob(
+                    message_id=m.id, role=m.role, parts=m.parts, task_id=m.task_id
+                )
                 for m in previous_messages
             ]
 

src/server/core/acontext_core/service/data/task.py

@@ -11,6 +11,32 @@
 from ...schema.session.task import TaskSchema
 
 
+async def fetch_planning_task(
+    db_session: AsyncSession, session_id: asUUID
+) -> Result[TaskSchema | None]:
+    query = (
+        select(Task)
+        .where(Task.session_id == session_id)
+        .options(selectinload(Task.messages))
+        .where(Task.is_planning_task == True)
+    )
+    result = await db_session.execute(query)
+    planning = result.scalars().first()
+    if planning is None:
+        return Result.resolve(None)
+    return Result.resolve(
+        TaskSchema(
+            id=planning.id,
+            session_id=planning.session_id,
+            task_order=planning.task_order,
+            task_status=planning.task_status,
+            task_description="",
+            task_data=planning.task_data,
+            raw_message_ids=[msg.id for msg in planning.messages],
+        )
+    )
+
+
 async def fetch_current_tasks(
     db_session: AsyncSession, session_id: asUUID, status: str = None
 ) -> Result[List[TaskSchema]]:
@@ -40,29 +66,6 @@ async def fetch_current_tasks(
     return Result.resolve(tasks_d)  # Fixed: return tasks_d instead of tasks
 
 
-async def fetch_planning_section(
-    db_session: AsyncSession, session_id: asUUID
-) -> Result[TaskSchema]:
-    query = (
-        select(Task)
-        .where(Task.session_id == session_id)
-        .where(Task.is_planning_task == True)
-        .options(selectinload(Task.messages))
-    )
-    result = await db_session.execute(query)
-    task = result.scalars().first()
-    return Resolver.resolve(
-        TaskSchema(
-            id=task.id,
-            session_id=task.session_id,
-            task_order=task.task_order,
-            task_status=task.task_status,
-            task_data=task.task_data,
-            raw_message_ids=[msg.id for msg in task.messages],
-        )
-    )
-
-
 async def update_task(
     db_session: AsyncSession,
     task_id: asUUID,