FROM feat/720-schedules-task-iq TO development (#721)

* feat: US-001 - Add DISTRIBUTED_WORKERS constant to src/constants

Consolidate DISTRIBUTED_WORKERS into src/constants/__init__.py and update
llm.py to import from there instead of defining it inline.

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
Signed-off-by: ryaneggz <kre8mymedia@gmail.com>

* feat: US-002 - Dispatch scheduled jobs to TaskIQ when distributed mode enabled

scheduled_llm_invoke() now checks DISTRIBUTED_WORKERS and dispatches to
run_agent_stream.kiq() when true, preserving the in-process fallback.

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
Signed-off-by: ryaneggz <kre8mymedia@gmail.com>

* feat: US-003 - Simplify schedule create route

Remove LLMController indirection from create_job route. Call
schedule_service.create_job() directly, removing unused imports.

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
Signed-off-by: ryaneggz <kre8mymedia@gmail.com>

* feat: US-004 - Add unit tests for TaskIQ dispatch path

Tests verify dispatch to TaskIQ when DISTRIBUTED_WORKERS=true, in-process
fallback when false, and thread_id extraction/generation logic.

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
Signed-off-by: ryaneggz <kre8mymedia@gmail.com>

* docs: update prd.json and progress.txt - all stories complete

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
Signed-off-by: ryaneggz <kre8mymedia@gmail.com>

* Finished ralph loop

* Move the input files

---------

Signed-off-by: ryaneggz <kre8mymedia@gmail.com>
Co-authored-by: Claude Opus 4.5 <noreply@anthropic.com>

Author: ryaneggz

backend/src/constants/__init__.py

@@ -145,6 +145,9 @@ def values(cls) -> list[str]:
 # GridSite
 MICROSOFT_TEAMS_WEBHOOK_URL = os.getenv("MICROSOFT_TEAMS_WEBHOOK_URL")
 
+# Distributed Workers
+DISTRIBUTED_WORKERS = os.getenv("DISTRIBUTED_WORKERS", "false").lower() == "true"
+
 # Thread Search
 # Number of recent messages to store per thread snapshot for semantic search
 THREAD_SNAPSHOT_MESSAGE_COUNT = 20

backend/src/routes/v0/llm.py

@@ -1,5 +1,4 @@
-import os
-from typing import Any, List
+from typing import Any
 from uuid import uuid4
 from fastapi.responses import JSONResponse, StreamingResponse
 from fastapi import (
@@ -20,7 +19,7 @@
 from src.controllers.llm import LLMController
 from src.services.llm import llm_service
 from src.services.prompt.optimize import PromptOptimizer, PromptOptimizerRequest
-from src.constants import GROQ_API_KEY
+from src.constants import DISTRIBUTED_WORKERS, GROQ_API_KEY
 from src.schemas.models import ProtectedUser
 from src.utils.auth import get_optional_user, get_optional_user_from_token
 from src.utils.logger import logger
@@ -34,9 +33,6 @@
 from src.constants.llm import DEFAULT_CHAT_MODEL, get_all_models, get_free_models
 from src.repos.user_settings_repo import UserSettingsRepo
 
-# Distributed workers mode - when true, tasks are enqueued to TaskIQ workers
-DISTRIBUTED_WORKERS = os.getenv("DISTRIBUTED_WORKERS", "false").lower() == "true"
-
 llm_router = APIRouter(tags=["LLM"], prefix="/llm")
 
 # TIME_LIMIT = "1/minute"

backend/src/routes/v0/schedule.py

@@ -1,11 +1,7 @@
 from fastapi import APIRouter, Depends, Response, Body
 from fastapi.responses import JSONResponse
 from fastapi_cache.decorator import cache
-from langgraph.store.base import BaseStore
 
-from src.controllers.llm import LLMController
-from src.agents import init_config
-from src.services.db import get_store
 from src.services.schedule import schedule_service
 from src.schemas.models import ProtectedUser
 from src.utils.auth import verify_credentials
@@ -84,11 +80,9 @@ async def get_job(
 async def create_job(
     job: ScheduleCreate = Body(openapi_examples=Examples.SCHEDULE_CREATE_EXAMPLES),
     user: ProtectedUser = Depends(verify_credentials),
-    store: BaseStore = Depends(get_store),
 ):
-    config = init_config(job.task, user.id)
-    llm_controller = LLMController(user=user, store=store, config=config)
-    schedule = await llm_controller.llm_task(job)
+    schedule_service.user_id = user.id
+    schedule = schedule_service.create_job(job)
     return JSONResponse(
         status_code=201,
         content={

backend/src/services/project.py

@@ -1,4 +1,4 @@
-from langgraph.store.base import BaseStore, SearchOp
+from langgraph.store.base import BaseStore
 from src.services.db import get_store_in_memory
 from src.repos.project_repo import Project, ProjectRepo
 from src.schemas.entities import SearchFilter

backend/src/services/schedule.py

@@ -54,6 +54,26 @@ async def scheduled_llm_invoke(task_dict: dict, user_id: str, title: str = None)
     This must be a module-level function (not a method) so APScheduler can pickle it.
     """
     from uuid import uuid4
+    from src.constants import DISTRIBUTED_WORKERS
+
+    # Distributed mode: dispatch to TaskIQ worker
+    if DISTRIBUTED_WORKERS:
+        from src.workers.tasks import run_agent_stream
+
+        metadata = task_dict.get("metadata") or {}
+        thread_id = metadata.get("thread_id") or str(uuid4())
+        logger.info(
+            f"🚀 Dispatching scheduled job '{title}' to TaskIQ worker"
+            f" (thread_id={thread_id})"
+        )
+        await run_agent_stream.kiq(
+            task_dict=task_dict,
+            user_id=user_id,
+            thread_id=thread_id,
+        )
+        return
+
+    # In-process mode: execute directly
     from src.schemas.entities import LLMRequest
     from src.agents import construct_agent, Orchestra, init_config
     from src.services.db import get_checkpoint_db, get_store_db
@@ -65,7 +85,7 @@ async def scheduled_llm_invoke(task_dict: dict, user_id: str, title: str = None)
     params = LLMRequest(**task_dict)
     params.metadata.user_id = user_id
     params.metadata.thread_id = params.metadata.thread_id or str(uuid4())
-    logger.info(f"✓ Successfully reconstructed LLMRequest")
+    logger.info("✓ Successfully reconstructed LLMRequest")
 
     # Initialize config and get files and todos
     config = init_config(params, user_id)
@@ -104,7 +124,7 @@ async def scheduled_llm_invoke(task_dict: dict, user_id: str, title: str = None)
             response = await agent.invoke(
                 params.input, config=config, context=ctx_schema
             )
-            logger.info(f"✓ LLM invocation completed successfully")
+            logger.info("✓ LLM invocation completed successfully")
 
             files_map = {**files_map, **response.get("files", {})}
             todos_list = [*todos_list, *response.get("todos", [])]

backend/src/services/tool.py

@@ -105,7 +105,9 @@ async def invoke_default_tool(
             configurable: dict[str, Any] = {"user_id": self.user_id}
             if config is not None:
                 # Merge caller config but exclude user_id to prevent override
-                config_without_user_id = {k: v for k, v in config.items() if k != "user_id"}
+                config_without_user_id = {
+                    k: v for k, v in config.items() if k != "user_id"
+                }
                 configurable.update(config_without_user_id)
             runnable_config = {"configurable": configurable}
         return await tool.ainvoke(

backend/src/tools/base/reasoning.py

@@ -1,4 +1,4 @@
-from pydantic import BaseModel, Field, field_validator
+from pydantic import BaseModel, Field
 from langchain_core.tools import tool
 
 

backend/src/tools/finance/news.py

@@ -46,7 +46,7 @@ def get_financial_news(ticker: str) -> tuple[str, dict]:
     )
 
     # Convert to markdown (clean, no index)
-    markdown = f"## Financial News\n" + f"```csv\n{df.to_csv(index=False)}\n```"
+    markdown = "## Financial News\n" + f"```csv\n{df.to_csv(index=False)}\n```"
     return markdown
 
     # TODO: Eval if this is cleaner later on. Get fidgety as table but looks good.

backend/src/tools/ms_teams/__init__.py

@@ -48,7 +48,7 @@ async def webhook_teams(text: str, runtime: ToolRuntime) -> str:
                 json={"text": text},
             )
             response.raise_for_status()
-            return f"Message sent to Microsoft Teams channel"
+            return "Message sent to Microsoft Teams channel"
         except Exception as e:
             raise ToolException(
                 f"Error sending message to Microsoft Teams channel: {e}"

backend/src/tools/search.py

@@ -6,9 +6,6 @@
 
 import httpx
 from charset_normalizer import from_bytes
-from langchain.tools import ToolRuntime
-from langchain_core.messages import ToolMessage
-from langgraph.types import Command
 from markdownify import markdownify as md
 from langchain_core.tools import tool
 from langchain_core.tools import ToolException