Rhinox13/chatapi

1from __future__ import annotations
2
3import uuid
4from typing import Any, Callable
5
6from ..repositories import ConversationStore
7from .pending import PendingTurnRegistry
8from .thinking import compose_thinking_text
9from .turn_protocols import build_protocol_response_id, normalize_message_text
10
11
12def _normalize_reasoning_stream_mode(value: str) -> str:
13    mode = str(value or "").strip().lower().replace("-", "_")
14    if mode == "summery":
15        mode = "summary"
16    elif mode == "reasoning":
17        mode = "reasoning_text"
18    if mode in {"summary", "reasoning_text"}:
19        return mode
20    return ""
21
22
23class TurnOutputController:
24    def __init__(
25        self,
26        *,
27        store: ConversationStore,
28        pending_turns: PendingTurnRegistry,
29        publish_sync: Callable[[str, str | None], None] | None = None,
30    ):
31        self._store = store
32        self._pending_turns = pending_turns
33        self._publish_sync = publish_sync
34
35    def add_text_delta(
36        self,
37        *,
38        conversation_id: str,
39        owner_id: str,
40        text: str,
41        reasoning_stream_mode: str = "",
42        kind: str = "answer",
43    ):
44        pending = self._pending_turns.add_draft(
45            conversation_id=conversation_id,
46            owner_id=owner_id,
47            chunk=normalize_message_text(text),
48            kind=kind,
49        )
50        if pending.aborted:
51            self._mark_aborted_conversation(pending)
52            raise ValueError(pending.abort_message or "request aborted")
53        self._apply_reasoning_stream_mode(
54            pending,
55            conversation_id=conversation_id,
56            owner_id=owner_id,
57            reasoning_stream_mode=reasoning_stream_mode,
58        )
59        conversation = self._store.get_conversation(conversation_id, owner_id)
60        if conversation is not None:
61            self._store.update_conversation(
62                conversation_id,
63                owner_id,
64                metadata={
65                    **conversation.metadata,
66                    "realtime_status": "waiting",
67                    "realtime_draft_text": pending.draft_text,
68                    **(
69                        {"request_format": pending.request_format}
70                        if pending.request_format
71                        else {}
72                    ),
73                    **(
74                        {"reasoning_stream_mode": pending.reasoning_stream_mode}
75                        if pending.request_format == "responses" and pending.reasoning_stream_mode
76                        else {}
77                    ),
78                },
79            )
80            self._notify(owner_id, conversation_id)
81        return pending
82
83    def complete_assistant_message(
84        self,
85        *,
86        conversation_id: str,
87        owner_id: str,
88        provider: str,
89        model: str | None = None,
90        reasoning_stream_mode: str = "",
91    ):
92        pending = self._require_pending(conversation_id, owner_id)
93        self._apply_reasoning_stream_mode(
94            pending,
95            conversation_id=conversation_id,
96            owner_id=owner_id,
97            reasoning_stream_mode=reasoning_stream_mode,
98        )
99        assistant_text = (
100            compose_thinking_text(pending.draft_segments)
101            if pending.request_format != "responses" and pending.draft_segments
102            else pending.draft_answer_text or pending.draft_text
103        )
104        if not assistant_text.strip():
105            raise ValueError("assistant message text is required")
106        response_id = build_protocol_response_id(pending.request_format, pending.request_id)
107        assistant_metadata = {
108            "provider": provider,
109            "model": str(model or pending.model or "mock-gpt-4.1-mini"),
110            "response_mode": "assistant_message",
111        }
112        updated_conversation = self._store.record_assistant_reply(
113            conversation_id,
114            owner_id,
115            pending.input_text,
116            assistant_text,
117            response_id=response_id,
118            assistant_metadata=assistant_metadata,
119        )
120        self._store.update_conversation(
121            conversation_id,
122            owner_id,
123            metadata={
124                **updated_conversation.metadata,
125                "realtime_status": "closed",
126                "realtime_draft_text": "",
127                "request_format": pending.request_format,
128                **(
129                    {"reasoning_stream_mode": pending.reasoning_stream_mode}
130                    if pending.request_format == "responses" and pending.reasoning_stream_mode
131                    else {}
132                ),
133            },
134        )
135        self._notify(owner_id, conversation_id)
136        resolved = self._pending_turns.resolve(
137            conversation_id=conversation_id,
138            owner_id=owner_id,
139            assistant_text=assistant_text,
140            response_id=response_id,
141            response_mode="assistant_message",
142            response_output_items=[],
143            response_output_text=assistant_text,
144        )
145        return resolved, assistant_metadata
146
147    def complete_tool_call(
148        self,
149        *,
150        conversation_id: str,
151        owner_id: str,
152        tool_name: str,
153        arguments: str,
154        provider: str,
155        model: str | None = None,
156        tool_call_id: str | None = None,
157        reasoning_stream_mode: str = "",
158    ):
159        pending = self._require_pending(conversation_id, owner_id)
160        self._apply_reasoning_stream_mode(
161            pending,
162            conversation_id=conversation_id,
163            owner_id=owner_id,
164            reasoning_stream_mode=reasoning_stream_mode,
165        )
166        call_id = tool_call_id or f"call_{uuid.uuid4().hex[:24]}"
167        assistant_text = f"{tool_name}({arguments})"
168        aborted = self._pending_turns.abort_if_output_would_exceed(
169            request_id=pending.request_id,
170            extra_text=assistant_text,
171        )
172        if aborted is not None:
173            self._mark_aborted_conversation(aborted)
174            raise ValueError(aborted.abort_message or "request aborted")
175        response_id = build_protocol_response_id(pending.request_format, pending.request_id)
176        output_items = [
177            {
178                "id": f"fc_{uuid.uuid4().hex[:24]}",
179                "type": "function_call",
180                "status": "completed",
181                "call_id": call_id,
182                "name": tool_name,
183                "arguments": arguments,
184            }
185        ]
186        assistant_metadata = {
187            "provider": provider,
188            "model": str(model or pending.model or "mock-gpt-4.1-mini"),
189            "response_mode": "tool_call",
190            "tool_name": tool_name,
191            "tool_call_id": call_id,
192            "arguments": arguments,
193        }
194        updated_conversation = self._store.record_assistant_reply(
195            conversation_id,
196            owner_id,
197            pending.input_text,
198            assistant_text,
199            response_id=response_id,
200            assistant_metadata=assistant_metadata,
201        )
202        self._store.update_conversation(
203            conversation_id,
204            owner_id,
205            metadata={
206                **updated_conversation.metadata,
207                "realtime_status": "closed",
208                "realtime_draft_text": "",
209                "request_format": pending.request_format,
210                **(
211                    {"reasoning_stream_mode": pending.reasoning_stream_mode}
212                    if pending.request_format == "responses" and pending.reasoning_stream_mode
213                    else {}
214                ),
215            },
216        )
217        self._notify(owner_id, conversation_id)
218        resolved = self._pending_turns.resolve(
219            conversation_id=conversation_id,
220            owner_id=owner_id,
221            assistant_text=assistant_text,
222            response_id=response_id,
223            response_mode="tool_call",
224            response_output_items=output_items,
225            response_output_text="",
226        )
227        return resolved, assistant_metadata
228
229    def abort(
230        self,
231        *,
232        conversation_id: str,
233        owner_id: str,
234        error_message: str,
235    ):
236        pending = self._pending_turns.abort(
237            conversation_id=conversation_id,
238            owner_id=owner_id,
239            error_message=error_message,
240        )
241        self._mark_aborted_conversation(pending)
242        return pending
243
244    def _mark_aborted_conversation(self, pending):
245        conversation = self._store.get_conversation(pending.conversation_id, pending.owner_id)
246        if conversation is None:
247            return
248        self._store.update_conversation(
249            pending.conversation_id,
250            pending.owner_id,
251            metadata={
252                **conversation.metadata,
253                "realtime_status": "aborted",
254                "realtime_draft_text": "",
255            },
256        )
257        self._notify(pending.owner_id, pending.conversation_id)
258
259    def _require_pending(self, conversation_id: str, owner_id: str):
260        pending = self._pending_turns.get_by_conversation(conversation_id)
261        if pending is None or pending.owner_id != owner_id:
262            raise ValueError("conversation is not waiting for a reply")
263        return pending
264
265    def _apply_reasoning_stream_mode(
266        self,
267        pending,
268        *,
269        conversation_id: str,
270        owner_id: str,
271        reasoning_stream_mode: str,
272    ) -> None:
273        if pending.request_format != "responses":
274            return
275
276        requested_mode = _normalize_reasoning_stream_mode(reasoning_stream_mode)
277        if requested_mode:
278            pending.reasoning_stream_mode = requested_mode
279
280    def _notify(self, owner_id: str, conversation_id: str) -> None:
281        if self._publish_sync is None:
282            return
283        self._publish_sync(owner_id, conversation_id)
284