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) <noreply@anthropic.com>

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)