fix(core): multi-turn task agent

Author: gusye1234

src/server/core/acontext_core/infra/async_mq.py

@@ -180,7 +180,7 @@ async def _process_message(self, config: ConsumerConfig, message: Message) -> No
                                 timeout=config.timeout,
                             )
                             _end_s = perf_counter()
-                            LOG.info(
+                            LOG.debug(
                                 f"Queue: {config.queue_name} processed in {_end_s - _start_s:.4f}s"
                             )
                     except ValidationError as e:

src/server/core/acontext_core/infra/db.py

@@ -1,5 +1,4 @@
-from ast import Or
-import os
+import traceback
 from typing import AsyncGenerator, Optional
 from contextlib import asynccontextmanager
 
@@ -141,6 +140,7 @@ async def get_session_context(self) -> AsyncGenerator[AsyncSession, None]:
             await session.commit()
         except Exception as e:
             logger.error(f"DB Session failed: {str(e)}. Rollback...")
+            logger.error(traceback.format_exc(), extra={"__internal__": True})
             await session.rollback()
             raise e
         finally:

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

@@ -80,6 +80,10 @@ async def task_agent_curd(
         use_tools = llm_return.tool_calls
         just_finish = False
         async with DB_CLIENT.get_session_context() as db_session:
+            r = await TD.fetch_current_tasks(db_session, session_id)
+            current_tasks, eil = r.unpack()
+            if eil:
+                return r
             use_ctx = TaskCtx(
                 db_session=db_session,
                 session_id=session_id,

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

@@ -26,9 +26,9 @@ async def _append_messages_to_task_handler(
         )
     actually_task_id = ctx.task_ids_index[task_order - 1]
     actually_message_ids = [
-        ctx.message_ids_index[i]
+        ctx.message_ids_index[i - 1]
         for i in message_order_indexes
-        if i < len(ctx.message_ids_index)
+        if i <= len(ctx.message_ids_index)
     ]
     if not actually_message_ids:
         return Result.resolve(
@@ -39,8 +39,13 @@ async def _append_messages_to_task_handler(
         actually_message_ids,
         actually_task_id,
     )
-    return r
-    pass
+    return (
+        Result.resolve(
+            f"Messages {message_order_indexes} appended to task {task_order}"
+        )
+        if r.ok()
+        else r
+    )
 
 
 _append_messages_to_task_tool = (

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

@@ -29,11 +29,10 @@ async def _append_messages_to_planning_section_handler(
         ctx.session_id,
         actually_message_ids,
     )
-    _, eil = r.unpack()
-    if eil:
-        return r
-    return Result.resolve(
-        f"Messages {message_order_indexes} appended to planning section"
+    return (
+        Result.resolve(f"Messages {message_order_indexes} appended to planning section")
+        if r.ok()
+        else r
     )
 
 

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

@@ -30,9 +30,13 @@ async def update_task_handler(
         ctx.db_session,
         actually_task_id,
         status=status,
-        patch_data={
-            "task_description": description,
-        },
+        patch_data=(
+            {
+                "task_description": description,
+            }
+            if description
+            else None
+        ),
     )
     t, eil = r.unpack()
     if eil:

src/server/core/acontext_core/schema/env.py

@@ -29,7 +29,7 @@ class CoreConfig(BaseModel):
     mq_consumer_handler_timeout: float = 96
     mq_default_message_ttl_seconds: int = 7 * 24 * 60 * 60
     mq_default_dlx_ttl_days: int = 7
-    mq_default_max_retries: int = 1
+    mq_default_max_retries: int = 3
     mq_default_retry_delay_unit_sec: float = 1.0
 
     # Database Configuration

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

@@ -102,7 +102,7 @@ class Message(CommonMixin):
         metadata={
             "db": Column(
                 UUID(as_uuid=True),
-                ForeignKey("tasks.id", ondelete="CASCADE"),
+                ForeignKey("tasks.id", ondelete="SET NULL"),
                 nullable=True,
             )
         },

src/server/core/acontext_core/schema/result.py

@@ -36,3 +36,8 @@ def unpack(self) -> tuple[Optional[T], Optional[Error]]:
         if self.error.status != Code.SUCCESS:
             return None, self.error
         return self.data, None
+
+    def ok(self) -> bool:
+        if self.error.status != Code.SUCCESS:
+            return False
+        return True

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

@@ -41,7 +41,6 @@ async def process_session_pending_message(session_id: asUUID):
             ]
 
         r = await AT.task_agent_curd(session_id, previous_messages_data, messages_data)
-
         async with DB_CLIENT.get_session_context() as session:
             await MD.update_message_status_to(
                 session, pending_message_ids, TaskStatus.SUCCESS