From 29f74afbf602bb22b62fda6efacfc727cb49ceee Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=90=89=E9=91=AB=E5=8D=87?= Date: Wed, 13 May 2026 20:58:23 +0800 Subject: [PATCH] fix(scheduler): give every scheduled run its own throwaway thread All scheduled tasks for a user shared one scheduler thread, so each run appended its messages to the same LangGraph checkpoint. Over time the history blew past the model's input window and every task on that account started failing with context-too-long. _execute_task now mints a fresh thread_id per run and threads it through ensure_thread_meta / snapshot / start_run / extract_*. The per-task scheduler_thread_id column is left in place for backwards compatibility but no longer drives the LLM call. Co-Authored-By: Claude Opus 4.7 (1M context) --- .../deerflow/runtime/scheduler/service.py | 23 +++++++++++-------- 1 file changed, 14 insertions(+), 9 deletions(-) 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)