fix(chat): end a turn when the agent stops speaking, not once it has spoken

CHANGELOG.md

@@ -13,6 +13,9 @@ guarantee. What the SDK models is now that subset and nothing else.
 
 ### Added
 
+- `turn_timeout` on `chat()`, `listen()` and `chat_sync()` — a wall-clock bound
+  on one turn, keeping whatever arrived before it elapsed. It defaults to none,
+  which leaves a turn waiting indefinitely on a session that never closes it.
 - `is_supported_event()` and `SUPPORTED_EVENTS` — whether an event carries the
   guarantee.
 - `session:llm_thinking`, `session:tool_status` and `session:restriction`, with
@@ -31,6 +34,16 @@ guarantee. What the SDK models is now that subset and nothing else.
 
 ### Changed
 
+- A turn now ends when the agent has spoken and then gone quiet, rather than
+  when it has spoken at some point during the turn. The old rule set a flag on
+  the first reply and never cleared it, so the rest of the turn ran on the
+  two-second timeout — including a tool call, where silence means the work is
+  taking a while. `session:tool_status` is no longer counted as speech for this
+  purpose: it reports what the agent is doing, not what it says.
+- `session:state` values `credits_exhausted` and `task_paused` now end a turn.
+  Both stop on the account rather than on the agent, so nothing further arrives
+  from the session. `task_stale` is dropped from that set: the server has no
+  such state, and staleness is `is_stale` on the session object over REST.
 - `session:join` now carries `since_revision` "0", on first join and on
   reconnect. The incremental-synchronization fields in the response are ignored.
 - Events are deduplicated on the event identifier together with the message

README.md

@@ -25,7 +25,8 @@ session = await client.sessions.create()
 await client.join_session(session["id"])
 await client.rebuild(session["id"])          # load the session's messages
 
-async for event in client.chat(session["id"], "Negotiate my Comcast bill"):
+async for event in client.chat(session["id"], "Negotiate my Comcast bill",
+                               turn_timeout=120):
     print(event.type, event.data)
 
 await client.disconnect()
@@ -99,6 +100,31 @@ the agent has finished speaking, and the complete `session:text` is the durable
 record, read back from history. Assemble the parts by `message_id` rather than
 waiting for the complete message to arrive live.
 
+## When a turn ends
+
+`chat()` yields until the agent has spoken and then gone quiet — two seconds of
+silence following text, a form, or a document. Silence following anything else
+is read as work still running: an agent that says "placing the call now" and
+starts a call is not finished, and the wait stays long.
+
+A turn also ends when `session:state` settles — `task_finished`,
+`task_cancelled`, `credits_exhausted` or `task_paused`. The last two stop on the
+account rather than on the agent.
+
+Nothing else ends a turn, so a turn whose last event is neither content nor a
+settled state waits. Pass `turn_timeout` to bound it in wall-clock seconds;
+whatever arrived before the deadline is still yielded.
+
+```python
+async for event in client.chat(sid, "...", turn_timeout=120):
+    ...
+```
+
+Without `turn_timeout` a turn is waited on indefinitely. Note also that a task
+outlives the turn that started it: an outbound call reports through
+`session:tool_status` minutes after `chat()` has returned, which `subscribe()`
+is for.
+
 ## Everything else passes through
 
 The server emits many more events. The SDK delivers every one of them unchanged

src/pine_assistant/chat.py

@@ -7,13 +7,24 @@
 """
 
 import asyncio
+import time
 from collections.abc import AsyncGenerator, Callable, Coroutine
 from typing import Any
 
 from pine_assistant.models.events import C2SEvent, S2CEvent
 from pine_assistant.transport.socketio import SocketIOManager
 
-TERMINAL_STATES = {"task_finished", "task_cancelled", "task_stale"}
+# States in which nothing further arrives until something changes outside the
+# session. `task_stale` was in this set and is not a state the server has —
+# staleness is `is_stale` on the session object, read over REST.
+SETTLED_STATES = frozenset({
+    "task_finished",
+    "task_cancelled",
+    # The task stopped on the account rather than on the agent. It can resume,
+    # but not from anything a client sends into the session.
+    "credits_exhausted",
+    "task_paused",
+})
 DEFAULT_IDLE_TIMEOUT_S = 120.0
 DEFAULT_RESPONSE_IDLE_TIMEOUT_S = 2.0
 
@@ -22,15 +33,16 @@
 # rebuild is not.
 FULL_REBUILD_REVISION = "0"
 
-# An agent response, for the purpose of deciding a turn has begun. Scope events
-# only — a turn must not hinge on an event we do not maintain.
-SUBSTANTIVE_EVENTS = frozenset({
+# What the agent says, as opposed to what it does. A turn is over when the
+# agent has spoken and then gone quiet; while it is working, silence means the
+# work is taking a while. Scope events only — no timing may hinge on an event
+# we do not maintain.
+CONTENT_EVENTS = frozenset({
     S2CEvent.SESSION_TEXT,
     S2CEvent.SESSION_TEXT_PART,
     S2CEvent.SESSION_RICH_CONTENT,
     S2CEvent.SESSION_FORM_TO_USER,
     S2CEvent.SESSION_TASK_FINISHED,
-    S2CEvent.SESSION_TOOL_STATUS,
     S2CEvent.SESSION_RESTRICTION,
 })
 
@@ -141,14 +153,17 @@ async def chat(
         *,
         attachments: list[dict[str, Any]] | None = None,
         referenced_sessions: list[dict[str, str]] | None = None,
+        turn_timeout: float | None = None,
     ) -> AsyncGenerator[ChatEvent, None]:
         """Send a message and yield the events that follow."""
         self._sio.emit(
             C2SEvent.SESSION_MESSAGE,
             self._build_message_data(content, attachments, referenced_sessions),
             session_id,
         )
-        async for event in self._listen(session_id, _skip_state_precheck=True):
+        async for event in self._listen(
+            session_id, turn_timeout=turn_timeout, _skip_state_precheck=True,
+        ):
             yield event
 
     def send_message(
@@ -167,13 +182,19 @@ def send_message(
         )
 
     async def _listen(
-        self, session_id: str, *, _skip_state_precheck: bool = False,
+        self, session_id: str, *, turn_timeout: float | None = None,
+        _skip_state_precheck: bool = False,
     ) -> AsyncGenerator[ChatEvent, None]:
-        """Yield events for a session until the turn ends."""
+        """Yield events for a session until the turn ends.
+
+        `turn_timeout` bounds the whole call in wall-clock seconds. Without one
+        a turn ends only when the session says so, and a session that says
+        nothing is waited on indefinitely.
+        """
         if not _skip_state_precheck and self._check_session_state:
             try:
                 session = await self._check_session_state(session_id)
-                if session.get("state") in TERMINAL_STATES:
+                if session.get("state") in SETTLED_STATES:
                     yield ChatEvent(type=S2CEvent.SESSION_STATE, session_id=session_id,
                                     data={"content": session["state"]})
                     return
@@ -183,10 +204,14 @@ async def _listen(
         queue: asyncio.Queue[ChatEvent | None] = asyncio.Queue()
         dedup = Deduplicator()
         done = False
-        received_agent_response = False
+        # Whether the most recent event was the agent speaking, which is what
+        # makes a silence meaningful. Re-evaluated on every event: a single
+        # flag, set once and never cleared, put the whole rest of the turn on
+        # the short timeout — including a tool call, where silence is expected.
+        spoke_last = False
 
         def handler(event: str, raw: dict[str, Any]) -> None:
-            nonlocal done, received_agent_response
+            nonlocal done, spoke_last
             payload = raw.get("payload") or {}
             p_session_id = payload.get("session_id")
             if p_session_id and p_session_id != session_id:
@@ -197,28 +222,35 @@ def handler(event: str, raw: dict[str, Any]) -> None:
                 return
             queue.put_nowait(chat_event)
 
-            if event in SUBSTANTIVE_EVENTS:
-                received_agent_response = True
+            spoke_last = event in CONTENT_EVENTS
             data = payload.get("data")
             if (event == S2CEvent.SESSION_STATE and isinstance(data, dict)
-                    and data.get("content", "") in TERMINAL_STATES):
+                    and data.get("content", "") in SETTLED_STATES):
                 done = True
                 queue.put_nowait(None)
 
         remove_handler = self._sio.add_event_handler(handler)
+        deadline = None if turn_timeout is None else time.monotonic() + turn_timeout
 
         try:
             while not done:
-                timeout = self._response_idle_timeout_s if received_agent_response else self._idle_timeout_s
+                timeout = self._response_idle_timeout_s if spoke_last else self._idle_timeout_s
+                if deadline is not None:
+                    remaining = deadline - time.monotonic()
+                    if remaining <= 0:
+                        break
+                    timeout = min(timeout, remaining)
                 try:
                     evt = await asyncio.wait_for(queue.get(), timeout=timeout)
                 except asyncio.TimeoutError:
-                    if received_agent_response:
+                    if deadline is not None and time.monotonic() >= deadline:
+                        break
+                    if spoke_last:
                         break
                     if self._check_session_state:
                         try:
                             session = await self._check_session_state(session_id)
-                            if session.get("state") in TERMINAL_STATES:
+                            if session.get("state") in SETTLED_STATES:
                                 yield ChatEvent(type=S2CEvent.SESSION_STATE, session_id=session_id,
                                                 data={"content": session["state"]})
                                 break

src/pine_assistant/client.py

@@ -197,17 +197,23 @@ async def chat(
         *,
         attachments: list[dict[str, Any]] | None = None,
         referenced_sessions: list[dict[str, str]] | None = None,
+        turn_timeout: float | None = None,
     ) -> AsyncGenerator[ChatEvent, None]:
         """Send a message and yield the events that follow.
 
         Events the SDK does not recognise are yielded unchanged alongside the
         rest; ignore what you do not handle.
+
+        `turn_timeout` bounds the call in wall-clock seconds, keeping whatever
+        arrived before it elapsed. Without one, a turn the session never closes
+        is waited on indefinitely.
         """
         self._ensure_connected()
         async for event in self._chat.chat(  # type: ignore[union-attr]
             session_id, content,
             attachments=attachments,
             referenced_sessions=referenced_sessions,
+            turn_timeout=turn_timeout,
         ):
             yield event
 
@@ -227,10 +233,14 @@ def send_message(
             referenced_sessions=referenced_sessions,
         )
 
-    async def listen(self, session_id: str) -> AsyncGenerator[ChatEvent, None]:
+    async def listen(
+        self, session_id: str, turn_timeout: float | None = None,
+    ) -> AsyncGenerator[ChatEvent, None]:
         """Listen for events on a joined session without sending a message."""
         self._ensure_connected()
-        async for event in self._chat._listen(session_id):  # type: ignore[union-attr]
+        async for event in self._chat._listen(  # type: ignore[union-attr]
+            session_id, turn_timeout=turn_timeout,
+        ):
             yield event
 
     async def subscribe(self, session_id: str) -> AsyncGenerator[ChatEvent, None]:
@@ -354,6 +364,7 @@ def chat_sync(
         *,
         attachments: list[dict[str, Any]] | None = None,
         referenced_sessions: list[dict[str, str]] | None = None,
+        turn_timeout: float | None = None,
     ) -> list[ChatEvent]:
         """Send a message and return all events as a list (blocking)."""
         async def _collect() -> list[ChatEvent]:
@@ -362,6 +373,7 @@ async def _collect() -> list[ChatEvent]:
                 session_id, content,
                 attachments=attachments,
                 referenced_sessions=referenced_sessions,
+                turn_timeout=turn_timeout,
             ):
                 events.append(event)
             return events