diff --git a/surfsense_backend/app/automations/actions/builtin/__init__.py b/surfsense_backend/app/automations/actions/builtin/__init__.py index f3d21a044..19a93dbfc 100644 --- a/surfsense_backend/app/automations/actions/builtin/__init__.py +++ b/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 diff --git a/surfsense_backend/app/automations/actions/builtin/chat_message/__init__.py b/surfsense_backend/app/automations/actions/builtin/chat_message/__init__.py new file mode 100644 index 000000000..6ca297493 --- /dev/null +++ b/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 diff --git a/surfsense_backend/app/automations/actions/builtin/chat_message/definition.py b/surfsense_backend/app/automations/actions/builtin/chat_message/definition.py new file mode 100644 index 000000000..5652d28c5 --- /dev/null +++ b/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) diff --git a/surfsense_backend/app/automations/actions/builtin/chat_message/factory.py b/surfsense_backend/app/automations/actions/builtin/chat_message/factory.py new file mode 100644 index 000000000..494ea5ffc --- /dev/null +++ b/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 diff --git a/surfsense_backend/app/automations/actions/builtin/chat_message/invoke.py b/surfsense_backend/app/automations/actions/builtin/chat_message/invoke.py new file mode 100644 index 000000000..596ccd60c --- /dev/null +++ b/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} diff --git a/surfsense_backend/app/automations/actions/builtin/chat_message/params.py b/surfsense_backend/app/automations/actions/builtin/chat_message/params.py new file mode 100644 index 000000000..61768bc13 --- /dev/null +++ b/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.", + )