diff --git a/CHANGELOG.md b/CHANGELOG.md index de02a62..78daed4 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/README.md b/README.md index 9987851..d6db1a8 100644 --- a/README.md +++ b/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 diff --git a/src/pine_assistant/chat.py b/src/pine_assistant/chat.py index f3ae4cb..8d6b3a9 100644 --- a/src/pine_assistant/chat.py +++ b/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,6 +153,7 @@ 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( @@ -148,7 +161,9 @@ async def chat( 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 diff --git a/src/pine_assistant/client.py b/src/pine_assistant/client.py index a923f2c..1ba59b0 100644 --- a/src/pine_assistant/client.py +++ b/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 diff --git a/tests/protocol/test_flow.py b/tests/protocol/test_flow.py index 1cbb7c4..fce1c1e 100644 --- a/tests/protocol/test_flow.py +++ b/tests/protocol/test_flow.py @@ -10,8 +10,8 @@ import pytest -from pine_assistant import AsyncPineAI -from pine_assistant.chat import FULL_REBUILD_REVISION +from pine_assistant import SUPPORTED_EVENTS, AsyncPineAI +from pine_assistant.chat import CONTENT_EVENTS, FULL_REBUILD_REVISION from tests.protocol.fake import SESSION_ID, FakeAsyncClient, envelope, load_fixture OTHER_SESSION = "1900000000000000999" @@ -102,7 +102,7 @@ def responder(_request): envelope("session:an_event_from_the_future", unknown_payload, event_id="unknown-1"), load_fixture("text"), - envelope("session:input_state", {"content": "waiting_input"}, + envelope("session:state", {"content": "task_finished"}, event_id="final-1", role="system"), ] @@ -125,7 +125,7 @@ def responder(_request): envelope("session:text", {"content": "elsewhere"}, session_id=OTHER_SESSION, event_id="other-1"), envelope("session:text", {"content": "here"}, event_id="here-1"), - envelope("session:input_state", {"content": "waiting_input"}, + envelope("session:state", {"content": "task_finished"}, event_id="final-2", role="system"), ] @@ -147,8 +147,8 @@ async def test_duplicate_events_are_suppressed(client): def responder(_request): return [duplicate, duplicate, - envelope("session:input_state", {"content": "waiting_input"}, - event_id="final-3", role="system")] + envelope("session:state", {"content": "task_finished"}, + event_id="final-3", role="system")] fake.responders["session:message"] = responder events = [e async for e in pine.chat(SESSION_ID, "hello")] @@ -167,7 +167,7 @@ def responder(_request): envelope("session:text", {"content": "a text"}, event_id=shared_id), envelope("session:update_title", {"content": "a title"}, event_id=shared_id, role="system"), - envelope("session:input_state", {"content": "waiting_input"}, + envelope("session:state", {"content": "task_finished"}, event_id="final-4", role="system"), ] @@ -182,20 +182,20 @@ def responder(_request): # -- Turn termination ----------------------------------------------------- -async def test_turn_ends_when_input_reopens_after_a_response(client): +async def test_turn_ends_on_a_settled_session_state(client): pine, fake = client def responder(_request): return [ load_fixture("text"), - envelope("session:input_state", {"content": "waiting_input"}, + envelope("session:state", {"content": "task_finished"}, event_id="final-5", role="system"), ] fake.responders["session:message"] = responder events = [e async for e in pine.chat(SESSION_ID, "hello")] - assert events[-1].type == "session:input_state" + assert events[-1].type == "session:state" async def test_turn_ends_on_a_terminal_session_state(client): @@ -224,7 +224,7 @@ def responder(_request): envelope("session:payment", {"status": "pending"}, event_id="oos-1"), envelope("session:reward", {"charge_type": "percentage"}, event_id="oos-2"), load_fixture("text"), - envelope("session:input_state", {"content": "waiting_input"}, + envelope("session:state", {"content": "task_finished"}, event_id="final-6", role="system"), ] @@ -232,7 +232,7 @@ def responder(_request): events = [e async for e in pine.chat(SESSION_ID, "hello")] types = [e.type for e in events] - assert types.index("session:text") < types.index("session:input_state") + assert types.index("session:text") < types.index("session:state") assert "session:payment" in types @@ -250,3 +250,114 @@ async def test_emit_event_sends_anything_enveloped(client): assert len(sent) == 1 assert sent[0]["payload"]["data"] == {"list": [{"id": "p1"}]} assert sent[0]["payload"]["message_id"] == "m1" + + +# -- When a turn ends ----------------------------------------------------- + + +@pytest.fixture +async def impatient(client): + """A client whose waits are measured in fractions of a second.""" + pine, fake = client + pine._chat._idle_timeout_s = 1.0 + pine._chat._response_idle_timeout_s = 0.1 + + async def _in_progress(_sid): + return {"state": "chat"} + pine._chat._check_session_state = _in_progress + return pine, fake + + +async def _collect(pine, budget=4.0, **kwargs): + async def drain(): + return [e async for e in pine.chat(SESSION_ID, "hello", **kwargs)] + try: + return await asyncio.wait_for(drain(), timeout=budget) + except asyncio.TimeoutError: + return None + + +async def test_a_turn_ends_when_the_agent_stops_speaking(impatient): + pine, fake = impatient + fake.responders["session:message"] = lambda _r: [ + envelope("session:text_part", {"content": "half a "}, event_id="t1"), + envelope("session:text_part", {"content": "sentence"}, event_id="t2"), + ] + + events = await _collect(pine) + + assert events is not None, "the turn never ended" + assert [e.type for e in events] == ["session:text_part", "session:text_part"] + + +async def test_a_running_tool_is_not_mistaken_for_a_finished_turn(impatient): + """An agent that says what it is about to do, then does it, goes quiet + while the work runs. Reading that silence as the end of the turn cuts the + caller off from the result.""" + pine, fake = impatient + fake.responders["session:message"] = lambda _r: [ + envelope("session:text", {"content": "Placing the call now."}, event_id="s1"), + envelope("session:tool_status", + {"tool_name": "phone_call", "status": "in_progress"}, event_id="w1"), + ] + + events = await _collect(pine, budget=1.5) + + assert events is None, "the turn ended while the tool was still running" + + +async def test_a_turn_not_ending_on_content_waits(impatient): + """The boundary of the rule, stated rather than left to be discovered. + + A turn is over when the agent has spoken and gone quiet. When the last + thing to arrive is not the agent speaking, the silence that follows is + read as work still running, and the turn waits. Every plain turn observed + against a live server ended on content, but nothing guarantees it — + `turn_timeout` is what bounds the case where it does not. + """ + pine, fake = impatient + trailing_reasoning = [ + envelope("session:text", {"content": "Done."}, event_id="s2"), + envelope("session:llm_thinking", {"type": "turn_end", "final": True}, event_id="k1"), + ] + + fake.responders["session:message"] = lambda _r: trailing_reasoning + assert await _collect(pine, budget=2.0) is None + + fake.responders["session:message"] = lambda _r: trailing_reasoning + events = await _collect(pine, budget=2.0, turn_timeout=0.4) + assert events is not None + assert [e.type for e in events] == ["session:text", "session:llm_thinking"] + + +async def test_turn_timeout_bounds_a_turn_that_would_not_end(impatient): + pine, fake = impatient + fake.responders["session:message"] = lambda _r: [ + envelope("session:llm_thinking", {"type": "placeholder"}, event_id="p1"), + ] + + events = await _collect(pine, budget=4.0, turn_timeout=0.5) + + assert events is not None, "turn_timeout did not bound the turn" + assert [e.type for e in events] == ["session:llm_thinking"], "events seen before the deadline are kept" + + +async def test_content_events_are_all_supported(): + """No timing may hinge on an event outside the supported surface.""" + assert {e.value for e in CONTENT_EVENTS} <= SUPPORTED_EVENTS + + +async def test_a_session_stopped_on_its_balance_ends_the_turn(impatient): + """Credits run out mid-task. Nothing further arrives until the balance is + restored, which is not something the session can be told.""" + pine, fake = impatient + fake.responders["session:message"] = lambda _r: [ + envelope("session:text", {"content": "Working on it."}, event_id="b1"), + envelope("session:state", {"content": "credits_exhausted"}, + event_id="b2", role="system"), + ] + + events = await _collect(pine, budget=4.0) + + assert events is not None, "the turn never ended" + assert events[-1].data["content"] == "credits_exhausted" diff --git a/tests/test_chat_engine.py b/tests/test_chat_engine.py index f816d95..2dd111d 100644 --- a/tests/test_chat_engine.py +++ b/tests/test_chat_engine.py @@ -5,12 +5,10 @@ """ from pine_assistant.chat import ( - SUBSTANTIVE_EVENTS, ChatEvent, Deduplicator, event_from_envelope, ) -from pine_assistant.models.events import SUPPORTED_EVENTS def _event(event_id, event_type="session:text"): @@ -53,8 +51,3 @@ def test_tolerates_a_missing_payload_and_metadata(self): event = event_from_envelope("session:text", {}, "s1") assert event.data is None assert event.event_id is None - - -def test_turn_control_uses_only_supported_events(): - """A turn must not begin or end on an event we do not maintain.""" - assert {e.value for e in SUBSTANTIVE_EVENTS} <= SUPPORTED_EVENTS