diff --git a/backend/druks/contrib/chat/app.py b/backend/druks/contrib/chat/app.py index a1d94050..9d3731d1 100644 --- a/backend/druks/contrib/chat/app.py +++ b/backend/druks/contrib/chat/app.py @@ -1,4 +1,6 @@ +from druks.agents import Agent from druks.apps import App +from druks.contrib.chat.contracts import TurnOutput class Chat(App): @@ -8,3 +10,9 @@ class Chat(App): # Every table this app owns carries the ``chat_`` prefix, so the thread's # schema can never collide with core's or another app's. prefix_tables = True + + reply = Agent( + description="replies to the operator on one conversation turn", + prompt="chat/talk.md", + contract=TurnOutput, + ) diff --git a/backend/druks/contrib/chat/contracts.py b/backend/druks/contrib/chat/contracts.py new file mode 100644 index 00000000..c81e7414 --- /dev/null +++ b/backend/druks/contrib/chat/contracts.py @@ -0,0 +1,6 @@ +from druks.agents import AgentOutput + + +class TurnOutput(AgentOutput): + # What one chat turn returns: the assistant line to append. + text: str diff --git a/backend/druks/contrib/chat/models.py b/backend/druks/contrib/chat/models.py index d1cda2cb..52575e19 100644 --- a/backend/druks/contrib/chat/models.py +++ b/backend/druks/contrib/chat/models.py @@ -4,6 +4,7 @@ from sqlalchemy.orm import Mapped, mapped_column from druks.contrib.chat.enums import Autonomy, Role +from druks.contrib.chat.schemas import ConversationSummary from druks.db import Base, StoredSubject, db_session @@ -54,6 +55,20 @@ async def add_message(self, *, role: Role, body: str) -> "Message": async def list_messages(self) -> list["Message"]: return await Message.list_for_conversation(self.id) + def get_summary(self) -> ConversationSummary: + return ConversationSummary.model_validate(self) + + @classmethod + async def list_summaries(cls, account_id: str | None) -> list[ConversationSummary]: + """This account's threads. A missing caller is not a shared board — + conversations are per-account, so the list is empty.""" + if account_id: + return [ + conversation.get_summary() + for conversation in await cls.list_for_account(account_id) + ] + return [] + class Message(Base): __tablename__ = "chat_messages" diff --git a/backend/druks/contrib/chat/routes.py b/backend/druks/contrib/chat/routes.py new file mode 100644 index 00000000..4e177e26 --- /dev/null +++ b/backend/druks/contrib/chat/routes.py @@ -0,0 +1,22 @@ +from typing import Annotated + +from fastapi import APIRouter, Body, Depends, status + +from druks.accounts.dependencies import current_account +from druks.accounts.models import Account +from druks.contrib.chat.enums import Role +from druks.contrib.chat.models import Conversation +from druks.contrib.chat.workflows import Talk + +router = APIRouter(prefix="/conversations") + + +@router.post("", status_code=status.HTTP_201_CREATED, operation_id="create_conversation") +async def create_conversation( + body: Annotated[str, Body(embed=True)], + account: Account = Depends(current_account), +) -> dict[str, int]: + conversation = await Conversation.create(account_id=account.id, title="") + await conversation.add_message(role=Role.USER, body=body) + await Talk.dispatch(conversation=conversation) + return {"id": conversation.id} diff --git a/backend/druks/contrib/chat/schemas.py b/backend/druks/contrib/chat/schemas.py new file mode 100644 index 00000000..9b368d10 --- /dev/null +++ b/backend/druks/contrib/chat/schemas.py @@ -0,0 +1,12 @@ +from datetime import datetime + +from druks.contrib.chat.enums import Autonomy +from druks.workflows import SubjectSummary + + +class ConversationSummary(SubjectSummary): + # The conversation's domain header — title and autonomy are this app's; + # status and the timeline come from the platform's subject read-side. + title: str + autonomy: Autonomy + created_at: datetime diff --git a/backend/druks/contrib/chat/templates/talk.md b/backend/druks/contrib/chat/templates/talk.md new file mode 100644 index 00000000..4f7fd17b --- /dev/null +++ b/backend/druks/contrib/chat/templates/talk.md @@ -0,0 +1,7 @@ +Reply to the operator. + +Autonomy: {{ autonomy }} + +{% for message in messages %} +{{ message.role }}: {{ message.body }} +{% endfor %} diff --git a/backend/druks/contrib/chat/workflows.py b/backend/druks/contrib/chat/workflows.py new file mode 100644 index 00000000..b0088e89 --- /dev/null +++ b/backend/druks/contrib/chat/workflows.py @@ -0,0 +1,51 @@ +from datetime import timedelta + +from druks.contrib.chat.app import Chat +from druks.contrib.chat.enums import Role +from druks.contrib.chat.models import Conversation +from druks.workflows import Gate, Workflow, step + + +class ChatTurn(Gate): + """The operator's next line, or a stop. Not ``review()`` — those controls + would offer approve/request_changes on a chat turn.""" + + name = "chat_turn" + text: str + stop: bool = False + + +class Talk(Workflow): + """One conversation: agent reply, park, operator line, repeat until stop.""" + + subject = Conversation + steps_reuse_sandbox = True + sandbox_hold = timedelta(minutes=15) + + async def run_multistep(self) -> None: + while True: + conversation = await self.subject + messages = await conversation.list_messages() + result = await Chat.reply( + autonomy=conversation.autonomy, + messages=[{"role": message.role, "body": message.body} for message in messages], + ) + await self.record_message(Role.ASSISTANT, result.text) + reply = await ChatTurn.wait( + input_request={"presentation": "in_app", "label": "Chat turn"}, + hold_sandbox=self.sandbox_hold, + ) + if reply.stop: + return + await self.record_message(Role.USER, reply.text) + + @step + async def record_message(self, role: Role, body: str) -> None: + conversation = await self.subject + await conversation.add_message(role=role, body=body) + + @classmethod + async def dispatch(cls, *, conversation: Conversation) -> str: + # One Talk per conversation: start() already dedups live runs on the + # subject, so a later line answers ChatTurn instead of starting again. + return await cls.start(subject=conversation) diff --git a/backend/tests/chat/test_models.py b/backend/tests/chat/test_models.py index 641b6fcc..fd2aae29 100644 --- a/backend/tests/chat/test_models.py +++ b/backend/tests/chat/test_models.py @@ -50,3 +50,17 @@ async def test_messages_are_rows_and_empty_is_a_list(): assert listed[0].id == first.id assert listed[1].id == second.id assert all(isinstance(message, Message) for message in listed) + + +async def test_list_summaries_is_this_accounts_threads(): + owner = await Account.get_or_create("op@example.com") + other = await Account.get_or_create("dev@example.com") + mine = await Conversation.create(account_id=owner.id, title="mine") + await Conversation.create(account_id=other.id, title="theirs") + + listed = await Conversation.list_summaries(owner.id) + assert [summary.id for summary in listed] == [str(mine.id)] + assert listed[0].title == "mine" + assert listed[0].autonomy == Autonomy.PROPOSE + assert listed[0].label == mine.label + assert await Conversation.list_summaries(None) == [] diff --git a/backend/tests/chat/test_routes.py b/backend/tests/chat/test_routes.py new file mode 100644 index 00000000..f47b6cf3 --- /dev/null +++ b/backend/tests/chat/test_routes.py @@ -0,0 +1,49 @@ +from druks.accounts.models import Account +from druks.contrib.chat.enums import Role +from druks.contrib.chat.models import Conversation +from druks.contrib.chat.workflows import Talk + + +async def test_create_conversation_posts_body_and_starts_talk(druks_client, monkeypatch): + started = [] + + async def dispatch(*, conversation): + started.append(conversation.id) + return "run-1" + + monkeypatch.setattr(Talk, "dispatch", staticmethod(dispatch)) + + created = await druks_client.post( + "/api/chat/conversations", + json={"body": "hello"}, + ) + + assert created.status_code == 201 + conversation = await Conversation.get(created.json()["id"]) + assert conversation.title == "" + account = await Account.get_or_create("op@example.com") + assert conversation.account_id == account.id + assert started == [conversation.id] + messages = await conversation.list_messages() + assert [message.body for message in messages] == ["hello"] + assert [message.role for message in messages] == [Role.USER] + + +async def test_a_later_post_starts_another_conversation_not_a_new_talk_on_the_first( + druks_client, monkeypatch +): + started = [] + + async def dispatch(*, conversation): + started.append(conversation.id) + return "run" + + monkeypatch.setattr(Talk, "dispatch", staticmethod(dispatch)) + + first = await druks_client.post("/api/chat/conversations", json={"body": "hello"}) + second = await druks_client.post("/api/chat/conversations", json={"body": "again"}) + + assert first.status_code == 201 + assert second.status_code == 201 + assert first.json()["id"] != second.json()["id"] + assert started == [first.json()["id"], second.json()["id"]] diff --git a/backend/tests/chat/test_workflows.py b/backend/tests/chat/test_workflows.py new file mode 100644 index 00000000..4c25db4f --- /dev/null +++ b/backend/tests/chat/test_workflows.py @@ -0,0 +1,92 @@ +from datetime import timedelta +from unittest import mock + +from druks.accounts.models import Account +from druks.contrib.chat.app import Chat +from druks.contrib.chat.contracts import TurnOutput +from druks.contrib.chat.enums import Role +from druks.contrib.chat.models import Conversation +from druks.contrib.chat.workflows import ChatTurn, Talk +from druks.workflows import current_workflow + + +async def _run_talk(conversation: Conversation) -> None: + flow = Talk() + flow.subject = conversation + token = current_workflow.set(flow) + try: + await flow.run_multistep() + finally: + current_workflow.reset(token) + + +async def test_dispatch_starts_talk_for_the_conversation(monkeypatch): + conversation = Conversation(id=42) + start = mock.AsyncMock(return_value="run-1") + monkeypatch.setattr(Talk, "start", staticmethod(start)) + + run_id = await Talk.dispatch(conversation=conversation) + + assert run_id == "run-1" + start.assert_awaited_once_with(subject=conversation) + + +async def test_talk_appends_the_assistant_line_and_stops(druks_db, monkeypatch): + account = await Account.get_or_create("op@example.com") + conversation = await Conversation.create(account_id=account.id, title="") + await conversation.add_message(role=Role.USER, body="hello") + + reply = mock.AsyncMock(return_value=TurnOutput(text="hi")) + monkeypatch.setattr(Chat, "reply", staticmethod(reply)) + waits: list[dict] = [] + + async def wait(cls, **kwargs): + waits.append(kwargs) + return ChatTurn(text="", stop=True) + + monkeypatch.setattr(ChatTurn, "wait", classmethod(wait)) + monkeypatch.setattr(Talk, "record_message", Talk.record_message.__wrapped__) + + await _run_talk(conversation) + + reply.assert_awaited_once() + assert waits[0]["hold_sandbox"] == timedelta(minutes=15) + assert waits[0]["input_request"] == {"presentation": "in_app", "label": "Chat turn"} + messages = await conversation.list_messages() + assert [message.body for message in messages] == ["hello", "hi"] + assert [message.role for message in messages] == [Role.USER, Role.ASSISTANT] + + +async def test_talk_appends_the_operator_line_and_loops(druks_db, monkeypatch): + account = await Account.get_or_create("op@example.com") + conversation = await Conversation.create(account_id=account.id, title="") + await conversation.add_message(role=Role.USER, body="hello") + + outputs = iter([TurnOutput(text="hi"), TurnOutput(text="ok")]) + turns: list[dict] = [] + + async def reply(**kwargs): + turns.append(kwargs) + return next(outputs) + + monkeypatch.setattr(Chat, "reply", staticmethod(reply)) + answers = iter([ChatTurn(text="and then?", stop=False), ChatTurn(text="", stop=True)]) + + async def wait(cls, **kwargs): + return next(answers) + + monkeypatch.setattr(ChatTurn, "wait", classmethod(wait)) + monkeypatch.setattr(Talk, "record_message", Talk.record_message.__wrapped__) + + await _run_talk(conversation) + + messages = await conversation.list_messages() + assert [message.body for message in messages] == ["hello", "hi", "and then?", "ok"] + assert [message.role for message in messages] == [ + Role.USER, + Role.ASSISTANT, + Role.USER, + Role.ASSISTANT, + ] + assert turns[0]["autonomy"] == conversation.autonomy + assert turns[1]["messages"][-1] == {"role": Role.USER, "body": "and then?"} diff --git a/backend/tests/test_app_roster.py b/backend/tests/test_app_roster.py index 3ffdeb8e..e9038420 100644 --- a/backend/tests/test_app_roster.py +++ b/backend/tests/test_app_roster.py @@ -19,8 +19,7 @@ def test_roster_lists_installed_apps_with_subject_types(tmp_path: Path): assert chat["builtin"] is False assert chat["hasFrontend"] is False assert chat["icon"] == "message-square" - # No workflow yet — subjects follow from workflows, not from StoredSubject alone. - assert chat["subjectTypes"] == [] + assert chat["subjectTypes"] == ["conversation"] field_notes = roster["field_notes"] assert field_notes["subjectTypes"] == ["note", "repository"]