feat(automations): add chat_message action

surfsense_backend/app/automations/actions/builtin/__init__.py

@@ -2,4 +2,4 @@
 
 from __future__ import annotations
 
-from . import agent_task  # noqa: F401
+from . import agent_task, chat_message  # noqa: F401

surfsense_backend/app/automations/actions/builtin/chat_message/__init__.py

@@ -0,0 +1,15 @@
+"""``chat_message`` action: post one turn into an existing chat thread.
+
+Imports ``definition`` for its side-effect (self-registration on the actions
+registry) and re-exports ``build_handler`` for direct consumers.
+"""
+
+from __future__ import annotations
+
+from .factory import build_handler
+from .params import ChatMessageActionParams
+
+__all__ = ["ChatMessageActionParams", "build_handler"]
+
+# Side-effect: register on the actions store.
+from . import definition  # noqa: F401

surfsense_backend/app/automations/actions/builtin/chat_message/definition.py

@@ -0,0 +1,18 @@
+"""``chat_message`` ``ActionDefinition`` registration."""
+
+from __future__ import annotations
+
+from ...store import register_action
+from ...types import ActionDefinition
+from .factory import build_handler
+from .params import ChatMessageActionParams
+
+CHAT_MESSAGE_ACTION = ActionDefinition(
+    type="chat_message",
+    name="Chat message",
+    description="Post a message into an existing chat thread and run one durable, persisted agent turn.",
+    params_model=ChatMessageActionParams,
+    build_handler=build_handler,
+)
+
+register_action(CHAT_MESSAGE_ACTION)

surfsense_backend/app/automations/actions/builtin/chat_message/factory.py

@@ -0,0 +1,23 @@
+"""Bind ``ActionContext`` to a callable that runs one ``chat_message`` step."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from ...types import ActionContext, ActionHandler
+from .invoke import run_chat_message
+from .params import ChatMessageActionParams
+
+
+def build_handler(ctx: ActionContext) -> ActionHandler:
+    """Return a handler closure that validates params and posts the turn."""
+
+    async def handle(params: dict[str, Any]) -> dict[str, Any]:
+        validated = ChatMessageActionParams.model_validate(params)
+        return await run_chat_message(
+            ctx=ctx,
+            thread_id=validated.thread_id,
+            message=validated.message,
+        )
+
+    return handle

surfsense_backend/app/automations/actions/builtin/chat_message/invoke.py

@@ -0,0 +1,59 @@
+"""Run one ``chat_message`` step by draining ``stream_new_chat``.
+
+Reuses the interactive chat turn: it persists the messages and advances the
+durable checkpointer for the thread, so a scheduled tick shares the same
+conversation memory as a user turn. The SSE frames have no client, so drain them.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from app.auth.context import AuthContext
+from app.db import ChatVisibility, User
+from app.services.chat_session_state_service import get_session_state
+from app.tasks.chat.streaming.flows.new_chat.orchestrator import stream_new_chat
+
+from ...types import ActionContext
+
+
+async def run_chat_message(
+    *,
+    ctx: ActionContext,
+    thread_id: int,
+    message: str,
+) -> dict[str, Any]:
+    """Post ``message`` into ``thread_id`` and run one durable agent turn.
+
+    Skips when a turn is already in flight on the thread, so a slow tick can't
+    overlap the next scheduled fire.
+    """
+    state = await get_session_state(ctx.session, thread_id)
+    if state is not None and state.ai_responding_to_user_id is not None:
+        return {"thread_id": thread_id, "frames": 0, "skipped": "in_flight"}
+
+    user_id = str(ctx.creator_user_id) if ctx.creator_user_id else None
+    auth_context: AuthContext | None = None
+    if ctx.creator_user_id:
+        user = await ctx.session.get(User, ctx.creator_user_id)
+        if user is not None:
+            auth_context = AuthContext.system(user, source="automation")
+
+    llm_config_id = ctx.chat_model_id if ctx.chat_model_id is not None else -1
+    request_id = f"automation:{ctx.run_id}:{ctx.step_id}"
+
+    frames = 0
+    async for _sse in stream_new_chat(
+        user_query=message,
+        workspace_id=ctx.workspace_id,
+        chat_id=thread_id,
+        user_id=user_id,
+        llm_config_id=llm_config_id,
+        needs_history_bootstrap=False,
+        thread_visibility=ChatVisibility.PRIVATE,
+        auth_context=auth_context,
+        request_id=request_id,
+    ):
+        frames += 1
+
+    return {"thread_id": thread_id, "frames": frames, "skipped": None}

surfsense_backend/app/automations/actions/builtin/chat_message/params.py

@@ -0,0 +1,21 @@
+"""``ChatMessageActionParams`` — params for the ``chat_message`` action type."""
+
+from __future__ import annotations
+
+from pydantic import BaseModel, ConfigDict, Field
+
+
+class ChatMessageActionParams(BaseModel):
+    """Post one turn into an existing chat thread from an automation step."""
+
+    model_config = ConfigDict(extra="forbid")
+
+    thread_id: int = Field(
+        ...,
+        description="NewChatThread id (the LangGraph thread) to post the turn into.",
+    )
+    message: str = Field(
+        ...,
+        min_length=1,
+        description="Message sent as the user turn; rendered at execute time.",
+    )