DRU-382 - Park each chat turn on ChatTurn so one conversation has one live Talk.

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,
+    )

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

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"

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}

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

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 %}

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)

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) == []

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"]]

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?"}

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"]