diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/runtime/scheduler/service.py b/offline-backend-20260512/backend/packages/harness/deerflow/runtime/scheduler/service.py index 1c200f34..7f011ce3 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/runtime/scheduler/service.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/runtime/scheduler/service.py @@ -459,12 +459,17 @@ class ScheduledTaskService: self._running_task_ids.discard(task["task_id"]) async def _execute_task(self, task: dict[str, Any], *, scheduled_for: datetime, manual: bool) -> None: + # Each run uses its own throwaway thread so message history cannot accumulate + # across runs and blow past the model's input window. The per-user + # task["scheduler_thread_id"] is no longer used to drive the LLM call — + # we keep it on the task row only for backwards compatibility. + runtime_thread_id = str(uuid.uuid4()) run_row = await self.store.create_run_once( { "id": str(uuid.uuid4()), "task_id": task["task_id"], "user_id": task["user_id"], - "scheduler_thread_id": task["scheduler_thread_id"], + "scheduler_thread_id": runtime_thread_id, "scheduled_for": scheduled_for, "status": "running", "started_at": datetime.now(UTC), @@ -478,16 +483,16 @@ class ScheduledTaskService: await self.store.update_task(task["task_id"], task["user_id"], {"next_run_at": next_run_at, "last_status": "running"}) try: - await self._ensure_thread_meta(task["user_id"], task["scheduler_thread_id"]) + await self._ensure_thread_meta(task["user_id"], runtime_thread_id) async with self._agent_run_lock: - output_snapshot = self._snapshot_result_files(task["scheduler_thread_id"], task["user_id"]) - record = await self._start_agent_run(task, scheduled_for=scheduled_for) + output_snapshot = self._snapshot_result_files(runtime_thread_id, task["user_id"]) + record = await self._start_agent_run(task, scheduled_for=scheduled_for, runtime_thread_id=runtime_thread_id) await self.store.update_run(run_row["id"], {"agent_run_id": record.run_id}) if record.task is not None: await record.task - result_text = await self._extract_latest_ai_text(task["scheduler_thread_id"]) + result_text = await self._extract_latest_ai_text(runtime_thread_id) result_files = await self._extract_result_files( - task["scheduler_thread_id"], + runtime_thread_id, task["user_id"], run_id=run_row["id"], output_snapshot=output_snapshot, @@ -542,7 +547,7 @@ class ScheduledTaskService: finally: reset_current_user(token) - async def _start_agent_run(self, task: dict[str, Any], *, scheduled_for: datetime): + async def _start_agent_run(self, task: dict[str, Any], *, scheduled_for: datetime, runtime_thread_id: str): from app.gateway.routers.thread_runs import RunCreateRequest from app.gateway.services import start_run @@ -555,7 +560,7 @@ class ScheduledTaskService: execution_context.update( { "is_scheduled_run": True, - "thread_id": task["scheduler_thread_id"], + "thread_id": runtime_thread_id, "thinking_enabled": execution_context.get("thinking_enabled", True), "is_plan_mode": execution_context.get("is_plan_mode", True), } @@ -572,7 +577,7 @@ class ScheduledTaskService: context=execution_context, multitask_strategy="enqueue" if False else "reject", ) - return await start_run(body, task["scheduler_thread_id"], request) # type: ignore[arg-type] + return await start_run(body, runtime_thread_id, request) # type: ignore[arg-type] finally: reset_current_user(token)