feat(core): add task sends

Author: gusye1234

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

@@ -58,7 +58,10 @@ def pack_current_message_with_ids(messages: list[MessageBlob]) -> str:
 
 
 async def build_task_ctx(
-    db_session: AsyncSession, session_id: asUUID, messages: list[MessageBlob]
+    db_session: AsyncSession,
+    project_id: asUUID,
+    session_id: asUUID,
+    messages: list[MessageBlob],
 ) -> TaskCtx:
     LOG.debug(f"Building task context for session {session_id}")
     r = await TD.fetch_current_tasks(db_session, session_id)
@@ -67,6 +70,7 @@ async def build_task_ctx(
         return r
     use_ctx = TaskCtx(
         db_session=db_session,
+        project_id=project_id,
         session_id=session_id,
         task_ids_index=[t.id for t in current_tasks],
         message_ids_index=[m.message_id for m in messages],
@@ -76,6 +80,7 @@ async def build_task_ctx(
 
 @track_process
 async def task_agent_curd(
+    project_id: asUUID,
     session_id: asUUID,
     previous_messages: List[MessageBlob],
     messages: List[MessageBlob],
@@ -130,7 +135,7 @@ async def task_agent_curd(
         use_tools = llm_return.tool_calls
         just_finish = False
         async with DB_CLIENT.get_session_context() as db_session:
-            USE_CTX = await build_task_ctx(db_session, session_id, messages)
+            USE_CTX = await build_task_ctx(db_session, project_id, session_id, messages)
             tool_response = []
             for tool_call in use_tools:
                 try:
@@ -155,7 +160,9 @@ async def task_agent_curd(
                     )
                     if tool_name not in NEED_UPDATE_CTX:
                         continue
-                    USE_CTX = await build_task_ctx(db_session, session_id, messages)
+                    USE_CTX = await build_task_ctx(
+                        db_session, project_id, session_id, messages
+                    )
                 except KeyError as e:
                     return Result.reject(f"Tool {tool_name} not found: {str(e)}")
                 except Exception as e:

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

@@ -6,6 +6,7 @@
 @dataclass
 class TaskCtx:
     db_session: AsyncSession
+    project_id: asUUID
     session_id: asUUID
     task_ids_index: list[asUUID]
     message_ids_index: list[asUUID]

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

@@ -1,15 +1,27 @@
+import asyncio
 from typing import Any
-from ....infra.db import AsyncSession
+from ....infra.async_mq import MQ_CLIENT
 from ..base import Tool, ToolPool
 from ....schema.llm import ToolSchema
 from ....schema.utils import asUUID
 from ....schema.result import Result
 from ....schema.orm import Task
+from ....schema.mq.space import NewTaskComplete
+from ....schema.session.task import TaskStatus
 from ....service.data import task as TD
 from ....env import LOG
+from ....service.constants import EX, RK
 from .ctx import TaskCtx
 
 
+async def send_complete_new_task(body: NewTaskComplete):
+    await MQ_CLIENT.publish(
+        exchange_name=EX.space_task,
+        routing_key=RK.space_task_new_complete,
+        body=body.model_dump_json(),
+    )
+
+
 async def update_task_handler(
     ctx: TaskCtx,
     llm_arguments: dict,
@@ -24,23 +36,33 @@ async def update_task_handler(
             f"Task order {task_order} is out of range, updating failed."
         )
     actually_task_id = ctx.task_ids_index[task_order - 1]
-    status = llm_arguments.get("task_status", None)
-    description = llm_arguments.get("task_description", None)
+    task_status = llm_arguments.get("task_status", None)
+    task_description = llm_arguments.get("task_description", None)
     r = await TD.update_task(
         ctx.db_session,
         actually_task_id,
-        status=status,
+        status=task_status,
         patch_data=(
             {
-                "task_description": description,
+                "task_description": task_description,
             }
-            if description
+            if task_description
             else None
         ),
     )
     t, eil = r.unpack()
     if eil:
         return r
+    if task_status is not None and task_status == TaskStatus.SUCCESS.value:
+        asyncio.create_task(
+            send_complete_new_task(
+                NewTaskComplete(
+                    project_id=ctx.project_id,
+                    session_id=ctx.session_id,
+                    task_id=actually_task_id,
+                )
+            )
+        )
     return Result.resolve(f"Task {t.task_order} updated")
 
 

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

@@ -18,6 +18,7 @@ class TaskSchema(BaseModel):
     task_description: str
     task_status: TaskStatus
     task_data: dict
+    space_digested: bool
     raw_message_ids: list[asUUID]
 
     def to_string(self) -> str:

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

@@ -4,7 +4,7 @@ class EX:
 
 
 class RK:
-    space_task_complete = "space.task.complete"
+    space_task_new_complete = "space.task.new.complete"
 
     session_message_insert = "session.message.insert"
     session_message_insert_retry = "session.message.insert.retry"

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

@@ -10,7 +10,7 @@
 
 
 async def process_session_pending_message(
-    project_config: ProjectConfig, session_id: asUUID
+    project_config: ProjectConfig, project_id: asUUID, session_id: asUUID
 ):
     pending_message_ids = None
     try:
@@ -60,7 +60,9 @@ async def process_session_pending_message(
                 for m in previous_messages
             ]
 
-        r = await AT.task_agent_curd(session_id, previous_messages_data, messages_data)
+        r = await AT.task_agent_curd(
+            project_id, session_id, previous_messages_data, messages_data
+        )
 
         after_status = TaskStatus.SUCCESS
         if not r.ok():

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

@@ -7,9 +7,30 @@
 from ...schema.result import ResultError
 from ...env import LOG, DEFAULT_CORE_CONFIG
 from ...schema.config import ProjectConfig
+from ...schema.session.task import TaskSchema
 
 
-async def process_space_pending_task(
-    project_config: ProjectConfig, space_id: asUUID, task_id: asUUID
+async def process_space_task(
+    project_config: ProjectConfig, space_id: asUUID, task: TaskSchema
 ):
-    pass
+    if task.task_status != TaskStatus.SUCCESS:
+        LOG.info(f"Task {task.id} is not success, skipping")
+        return
+
+    async with DB_CLIENT.get_session_context() as db_session:
+        # 1. fetch messages from task
+        msg_ids = task.raw_message_ids
+        r = await MD.fetch_messages_data_by_ids(db_session, msg_ids)
+        if not r.ok():
+            return
+        messages, _ = r.unpack()
+        messages_data = [
+            MessageBlob(message_id=m.id, role=m.role, parts=m.parts, task_id=m.task_id)
+            for m in messages
+        ]
+    print(messages_data)
+    # 2. call agent to digest raw messages to SOP
+    ...
+
+    # 3. Create block and trigger space_agent to save it
+    ...

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

@@ -32,6 +32,7 @@ async def fetch_planning_task(
             task_status=planning.task_status,
             task_description="",
             task_data=planning.task_data,
+            space_digested=planning.space_digested,
             raw_message_ids=[msg.id for msg in planning.messages],
         )
     )
@@ -51,6 +52,7 @@ async def fetch_task(db_session: AsyncSession, task_id: asUUID) -> Result[TaskSc
             task_status=task.task_status,
             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],
         )
     )
@@ -78,6 +80,7 @@ async def fetch_current_tasks(
             task_status=t.task_status,
             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],
         )
         for t in tasks

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

@@ -127,7 +127,9 @@ async def insert_new_message(body: InsertNewMessage, message: Message):
                 routing_key=RK.session_message_insert_retry,
                 body=body.model_dump_json(),
             )
-        await MC.process_session_pending_message(project_config, body.session_id)
+        await MC.process_session_pending_message(
+            project_config, body.project_id, body.session_id
+        )
     finally:
         await release_session_message_lock(str(body.session_id))
 
@@ -185,6 +187,8 @@ async def buffer_new_message(body: InsertNewMessage, message: Message):
         )
         return
     try:
-        await MC.process_session_pending_message(project_config, body.session_id)
+        await MC.process_session_pending_message(
+            project_config, body.project_id, body.session_id
+        )
     finally:
         await release_session_message_lock(str(body.session_id))

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

@@ -22,15 +22,14 @@
     mq_client=MQ_CLIENT,
     config=ConsumerConfigData(
         exchange_name=EX.space_task,
-        routing_key=RK.space_task_complete,
-        queue_name=RK.space_task_complete,
+        routing_key=RK.space_task_new_complete,
+        queue_name=RK.space_task_new_complete,
     ),
 )
-async def complete_new_task(body: NewTaskComplete, message: Message):
-    async with DB_CLIENT.session() as db_session:
+async def space_complete_new_task(body: NewTaskComplete, message: Message):
+    async with DB_CLIENT.get_session_context() as db_session:
         r = await SD.fetch_session(db_session, body.session_id)
         if not r.ok():
-            LOG.error(f"Session {body.session_id} not found, error: {r.error}")
             return
         session_data, _ = r.unpack()
         if session_data.space_id is None:
@@ -39,14 +38,10 @@ async def complete_new_task(body: NewTaskComplete, message: Message):
         SPACE_ID = session_data.space_id
         r = await TD.fetch_task(db_session, body.task_id)
         if not r.ok():
-            LOG.error(f"Task {body.task_id} not found, error: {r.error}")
             return
         TASK_DATA, _ = r.unpack()
         r = await TD.set_task_space_digested(db_session, body.task_id)
         if not r.ok():
-            LOG.error(
-                f"Failed to update task {body.task_id} space digested, error: {r.error}"
-            )
             return
         already_digested, _ = r.unpack()
         if already_digested:
@@ -58,4 +53,4 @@ async def complete_new_task(body: NewTaskComplete, message: Message):
         if eil:
             return
 
-    await STC.process_space_pending_task(project_config, SPACE_ID, TASK_DATA.id)
+    await STC.process_space_task(project_config, SPACE_ID, TASK_DATA)