fix(core): append before update status

Author: gusye1234

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

@@ -73,6 +73,7 @@ async def build_task_ctx(
         project_id=project_id,
         session_id=session_id,
         task_ids_index=[t.id for t in current_tasks],
+        task_index=current_tasks,
         message_ids_index=[m.message_id for m in messages],
     )
     return use_ctx

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

@@ -65,14 +65,13 @@ def system_prompt(cls) -> str:
 
 ## Report your Thinking
 Use extremely brief wordings to report:
-1. Any new user requirement or planning? What are the tasks/steps?
-2. For each task, is it a task modification or creation situation?
-3. How existing tasks are related to current conversation? Do the existing tasks need to be updated?
+1. Any user requirement or planning?
+2. How existing tasks are related to current conversation? 
+3. Any new task is created?
 4. Messages are contributed to which task?
-5. Do New/Existing tasks' status need to be updated?
-6. Describe your actions.
-7. Confirm your will call every necessary tools in this response.
-8. Confirm your will call `finish` tool once every tools are called
+5. Which task's status/description need to be updated?
+6. Describe your tool-call actions to correctly manage the tasks.
+7. Confirm your will call `finish` tool after every tools are called
 """
 
     @classmethod

src/server/core/acontext_core/llm/tool/task_lib/append.py

@@ -6,15 +6,15 @@
 from ....schema.result import Result
 from ....schema.orm import Task
 from ....service.data import task as TD
-from ....env import LOG
+from ....schema.session.task import TaskStatus
 from .ctx import TaskCtx
 
 
 async def _append_messages_to_task_handler(
     ctx: TaskCtx,
     llm_arguments: dict,
 ) -> Result[str]:
-    task_order = llm_arguments.get("task_order", None)
+    task_order: int = llm_arguments.get("task_order", None)
     message_order_indexes = llm_arguments.get("message_ids", [])
     if not task_order:
         return Result.resolve(
@@ -25,6 +25,7 @@ async def _append_messages_to_task_handler(
             f"Task order {task_order} is out of range, appending failed."
         )
     actually_task_id = ctx.task_ids_index[task_order - 1]
+    actually_task = ctx.task_index[task_order - 1]
     actually_message_ids = [
         ctx.message_ids_index[i]
         for i in message_order_indexes
@@ -34,6 +35,10 @@ async def _append_messages_to_task_handler(
         return Result.resolve(
             f"No message ids to append, skip: {message_order_indexes}"
         )
+    if actually_task.task_status in (TaskStatus.SUCCESS, TaskStatus.FAILED):
+        return Result.resolve(
+            f"Task {task_order} is already {actually_task.task_status}, appending failed."
+        )
     r = await TD.append_messages_to_task(
         ctx.db_session,
         actually_message_ids,
@@ -56,6 +61,7 @@ async def _append_messages_to_task_handler(
                 "name": "append_messages_to_task",
                 "description": """Link current message ids to a task for tracking progress and context.
 Use this to associate conversation messages with relevant tasks.
+Make sure you append messages first(if any), then update the task status.
 If the task is marked as 'success' or 'failed', don't append messages to it.""",
                 "parameters": {
                     "type": "object",

src/server/core/acontext_core/llm/tool/task_lib/ctx.py

@@ -1,6 +1,7 @@
 from dataclasses import dataclass
 from ....infra.db import AsyncSession
 from ....schema.utils import asUUID
+from ....schema.session.task import TaskSchema
 
 
 @dataclass
@@ -9,4 +10,5 @@ class TaskCtx:
     project_id: asUUID
     session_id: asUUID
     task_ids_index: list[asUUID]
+    task_index: list[TaskSchema]
     message_ids_index: list[asUUID]

src/server/core/acontext_core/schema/session/message.py

@@ -1,3 +1,4 @@
+import json
 from pydantic import BaseModel
 from typing import List, Optional
 from ..orm import Part, ToolCallMeta
@@ -21,7 +22,13 @@ def pack_part_line(role: str, part: Part) -> str:
         return f"<{role}> {part.text}"
     elif part.type == "tool-call":
         tool_call_meta = ToolCallMeta(**part.meta)
-        return f"<{role}> USE TOOL {tool_call_meta.tool_name}, WITH PARAMS {tool_call_meta.arguments}"
+        tool_data = json.dumps(
+            {
+                "tool_name": tool_call_meta.tool_name,
+                "arguments": tool_call_meta.arguments,
+            }
+        )
+        return f"<{role}> {tool_data}"
     else:
         raise TypeError(f"Unknown message part type: {part.type}")
 

src/server/core/acontext_core/service/__init__.py

@@ -1 +1,2 @@
 from . import session_message
+from . import space_task

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

@@ -13,6 +13,7 @@
 async def process_space_task(
     project_config: ProjectConfig, space_id: asUUID, task: TaskSchema
 ):
+    print("SPACE!!!!", space_id, task)
     if task.task_status != TaskStatus.SUCCESS:
         LOG.info(f"Task {task.id} is not success, skipping")
         return
@@ -28,7 +29,7 @@ async def process_space_task(
             MessageBlob(message_id=m.id, role=m.role, parts=m.parts, task_id=m.task_id)
             for m in messages
         ]
-    print(messages_data)
+    print("\n".join([m.to_string() for m in messages_data]))
     # 2. call agent to digest raw messages to SOP
     ...
 

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

@@ -33,7 +33,9 @@ async def fetch_planning_task(
             task_description="",
             task_data=planning.task_data,
             space_digested=planning.space_digested,
-            raw_message_ids=[msg.id for msg in planning.messages],
+            raw_message_ids=[
+                msg.id for msg in sorted(planning.messages, key=lambda m: m.created_at)
+            ],
         )
     )
 
@@ -53,7 +55,9 @@ async def fetch_task(db_session: AsyncSession, task_id: asUUID) -> Result[TaskSc
             task_description=task.task_data.get("task_description", ""),
             task_data=task.task_data,
             space_digested=task.space_digested,
-            raw_message_ids=[msg.id for msg in task.messages],
+            raw_message_ids=[
+                msg.id for msg in sorted(task.messages, key=lambda m: m.created_at)
+            ],
         )
     )
 
@@ -81,11 +85,13 @@ async def fetch_current_tasks(
             task_description=t.task_data.get("task_description", ""),
             task_data=t.task_data,
             space_digested=t.space_digested,
-            raw_message_ids=[msg.id for msg in t.messages],
+            raw_message_ids=[
+                msg.id for msg in sorted(t.messages, key=lambda m: m.created_at)
+            ],
         )
         for t in tasks
     ]
-    return Result.resolve(tasks_d)  # Fixed: return tasks_d instead of tasks
+    return Result.resolve(tasks_d)
 
 
 async def update_task(