Rhinox13/chatapi
0
1from __future__ import annotations
2
3import uuid
4from dataclasses import dataclass
5from typing import Any, Callable
6
7from ..core import AppDependencies
8from ..repositories import build_title
9from .automation_rules import AutomationRuleEngine
10from .ntfy import notify_new_message
11from .output_controller import TurnOutputController
12from .payload_anthropic import build_anthropic_message_response
13from .payload_chat_completions import build_chat_completion_response
14from .payload_openai import (
15 build_openai_error,
16 build_openai_response,
17 estimate_usage,
18)
19from .pending import PendingTurn
20from .turn_request_tools import (
21 extract_tool_names,
22 extract_tool_schemas,
23 find_response_message_metadata,
24)
25from .turn_protocols import (
26 build_message_debug_metadata,
27 extract_chatbox_comparable_request_messages,
28 extract_context_text,
29 extract_request_messages,
30 normalize_chatbox_history_content,
31 normalize_message_text,
32 request_input_payload,
33 resolve_conversation_for_request,
34)
35
36
37def _normalize_reasoning_stream_mode(value: Any) -> str:
38 mode = str(value or "").strip().lower().replace("-", "_")
39 if mode == "summery":
40 mode = "summary"
41 elif mode == "reasoning":
42 mode = "reasoning_text"
43 if mode in {"summary", "reasoning_text"}:
44 return mode
45 return ""
46
47
48def _normalize_request_format(value: Any) -> str:
49 request_format = str(value or "").strip().lower().replace("-", "_")
50 if request_format in {"responses", "chat_completions", "anthropic_messages"}:
51 return request_format
52 return ""
53
54
55def _conversation_request_format(conversation: Any) -> str:
56 metadata = dict(getattr(conversation, "metadata", {}) or {})
57 return _normalize_request_format(metadata.get("request_format"))
58
59
60@dataclass(frozen=True)
61class PreparedTurn:
62 pending: PendingTurn
63 conversation_id: str
64
65
66class TurnCoordinator:
67 def __init__(
68 self,
69 deps: AppDependencies,
70 *,
71 extensions: dict[str, Any],
72 logger: Any,
73 publish_sync: Callable[[str, str | None], None] | None = None,
74 ):
75 self._deps = deps
76 self._extensions = extensions
77 self._logger = logger
78 self._publish_sync = publish_sync
79 self._output_controller = TurnOutputController(
80 store=deps.store,
81 pending_turns=deps.pending_turns,
82 publish_sync=publish_sync,
83 )
84 self._automation_rules = AutomationRuleEngine(
85 user_store=deps.user_store,
86 output_controller=self._output_controller,
87 )
88
89 @property
90 def auth(self):
91 return self._deps.auth
92
93 @property
94 def pending_turns(self):
95 return self._deps.pending_turns
96
97 @property
98 def store(self):
99 return self._deps.store
100
101 @property
102 def user_store(self):
103 return self._deps.user_store
104
105 @property
106 def settings(self):
107 return self._deps.settings
108
109 @property
110 def message_rate_limiter(self):
111 return self._deps.message_rate_limiter
112
113 def get_stream_heartbeat_settings(self, owner_id: str) -> dict[str, Any]:
114 return self._automation_rules.get_heartbeat_rule_settings(owner_id)
115
116 def update_stream_heartbeat_settings(
117 self,
118 owner_id: str,
119 *,
120 heartbeat_text: str,
121 heartbeat_interval_seconds: float,
122 ) -> dict[str, Any]:
123 return self._automation_rules.update_heartbeat_rule_settings(
124 owner_id,
125 heartbeat_text=heartbeat_text,
126 interval_seconds=heartbeat_interval_seconds,
127 )
128
129 def get_automation_rules(self, owner_id: str) -> list[dict[str, Any]]:
130 return self._automation_rules.load_rule_payloads(owner_id)
131
132 def update_automation_rules(self, owner_id: str, rules: list[dict[str, Any]]) -> list[dict[str, Any]]:
133 return self._automation_rules.save_rule_payloads(owner_id, rules)
134
135 def build_abort_error(self, message_text: str) -> tuple[dict[str, Any], int]:
136 return build_openai_error(
137 message_text or "request aborted",
138 code="request_aborted",
139 status=400,
140 )
141
142 def build_not_found_error(self, message: str, *, code: str = "not_found", status: int = 404):
143 return build_openai_error(message, code=code, status=status)
144
145 def _pending_limits(self) -> dict[str, Any]:
146 return self._deps.system_config_store.get_pending_limits()
147
148 def _mark_aborted_pending_turns(self, pending_turns: list[PendingTurn]) -> None:
149 for pending in pending_turns:
150 conversation = self.store.get_conversation(pending.conversation_id, pending.owner_id)
151 self.store.update_conversation(
152 pending.conversation_id,
153 pending.owner_id,
154 metadata={
155 **(conversation.metadata if conversation else {}),
156 "realtime_status": "aborted",
157 "realtime_draft_text": "",
158 },
159 )
160 self._notify(pending.owner_id, pending.conversation_id)
161
162 def enforce_pending_limits(self, owner_id: str) -> dict[str, Any]:
163 limits = self._pending_limits()
164 abort_message = str(limits.get("abort_message") or "本次回复等待超过限制，已自动结束，请重新发送。")
165 aborted = self.pending_turns.abort_expired(
166 max_age_seconds=float(limits.get("max_age_seconds") or 0),
167 error_message=abort_message,
168 )
169 aborted.extend(
170 self.pending_turns.abort_owner_over_limit(
171 owner_id=owner_id,
172 max_active=int(limits.get("max_per_user") or 10),
173 error_message=abort_message,
174 )
175 )
176 if aborted:
177 self._mark_aborted_pending_turns(aborted)
178 return limits
179
180 def _resolve_reasoning_stream_mode(
181 self,
182 data: dict[str, Any],
183 *,
184 request_format: str,
185 conversation: Any | None,
186 ) -> str:
187 if request_format != "responses":
188 return ""
189
190 requested_mode = _normalize_reasoning_stream_mode(
191 data.get("reasoning_stream_mode")
192 or data.get("responses_reasoning_stream_mode")
193 or data.get("reasoning_mode")
194 )
195 return requested_mode
196
197 def _resolve_conversation_protocol(
198 self,
199 data: dict[str, Any],
200 *,
201 request_format: str,
202 conversation: Any | None,
203 ) -> str:
204 if conversation is None:
205 return request_format
206 metadata = dict(conversation.metadata or {})
207 locked_format = _normalize_request_format(metadata.get("request_format"))
208 if locked_format and locked_format != request_format:
209 raise ValueError("conversation protocol is already locked")
210 if request_format == "responses":
211 return "responses"
212 if locked_format:
213 return locked_format
214 return request_format
215
216 def prepare_pending_turn(self, data: dict[str, Any], request_format: str):
217 if not isinstance(data, dict):
218 return build_openai_error("request body must be a JSON object")
219
220 owner = self.auth.owner_id()
221 normalized_data = self._deps.image_store.normalize_request_data(data, owner_id=owner)
222 context_text = extract_context_text(normalized_data, request_format)
223 if not context_text:
224 return build_openai_error("input is required")
225
226 model = str(normalized_data.get("model") or "mock-gpt-4.1-mini")
227 rate_limit = self.user_store.get_effective_messages_per_minute_limit(owner, 0)
228 if not self.message_rate_limiter.allow(owner, rate_limit):
229 return build_openai_error(
230 f"rate limit exceeded: max {rate_limit} messages per minute",
231 code="rate_limit_exceeded",
232 status=429,
233 )
234
235 resolved_conversation, conversation_error = resolve_conversation_for_request(
236 self.store,
237 normalized_data,
238 owner,
239 request_format,
240 )
241 if conversation_error is not None:
242 message, status = conversation_error
243 error_code = "conflict" if status == 409 else "not_found"
244 return build_openai_error(message, code=error_code, status=status)
245 if (
246 resolved_conversation is not None
247 and resolved_conversation.source in {"history", "tool_call_id"}
248 and _conversation_request_format(resolved_conversation.conversation)
249 and _conversation_request_format(resolved_conversation.conversation) != request_format
250 ):
251 resolved_conversation = None
252 conversation = (
253 resolved_conversation.conversation
254 if resolved_conversation is not None
255 else None
256 )
257 if conversation is None:
258 conversation = self.store.create_conversation(owner, title=build_title(context_text))
259
260 existing_pending = self.pending_turns.get_by_conversation(conversation.id)
261 if existing_pending is not None:
262 if resolved_conversation is not None and resolved_conversation.source == "history":
263 conversation = self.store.create_conversation(owner, title=build_title(context_text))
264 else:
265 return build_openai_error(
266 "conversation is waiting for a reply",
267 code="conflict",
268 status=409,
269 )
270
271 try:
272 request_format = self._resolve_conversation_protocol(
273 normalized_data,
274 request_format=request_format,
275 conversation=conversation,
276 )
277 reasoning_stream_mode = self._resolve_reasoning_stream_mode(
278 normalized_data,
279 request_format=request_format,
280 conversation=conversation,
281 )
282 except ValueError as error:
283 return build_openai_error(str(error), code="conflict", status=409)
284
285 conversation_metadata = {
286 **conversation.metadata,
287 "request_format": request_format,
288 }
289 conversation = self.store.update_conversation(
290 conversation.id,
291 owner,
292 metadata=conversation_metadata,
293 )
294
295 extracted_messages = extract_request_messages(
296 self.store,
297 normalized_data,
298 conversation_id=conversation.id,
299 owner=owner,
300 request_format=request_format,
301 )
302 comparable_messages = extract_chatbox_comparable_request_messages(
303 self.store,
304 normalized_data,
305 owner=owner,
306 request_format=request_format,
307 )
308 history_prefix_length = self.store.get_request_history_prefix_length(
309 conversation.id,
310 owner,
311 comparable_messages,
312 normalize_stored_content=normalize_chatbox_history_content,
313 )
314 if history_prefix_length > 0 and len(extracted_messages) >= history_prefix_length:
315 extracted_messages = extracted_messages[history_prefix_length:]
316 has_tool_results = any(
317 str(message_payload.get("metadata", {}).get("response_mode", "")).strip()
318 == "tool_result"
319 for message_payload in extracted_messages
320 if isinstance(message_payload, dict)
321 )
322 if has_tool_results:
323 extracted_messages = [
324 message_payload
325 for message_payload in extracted_messages
326 if str(message_payload.get("role") or "").strip() != "user"
327 ]
328 updated_conversation = self.store.update_conversation(
329 conversation.id,
330 owner,
331 title=conversation.title
332 if conversation.title not in {"新会话", "New conversation", ""}
333 else build_title(context_text),
334 last_user_text=context_text[:1000],
335 )
336 pending_limits = self.enforce_pending_limits(owner)
337 pending = self.pending_turns.register(
338 conversation_id=conversation.id,
339 owner_id=owner,
340 model=model,
341 input_text=context_text,
342 request_format=request_format,
343 reasoning_stream_mode=reasoning_stream_mode,
344 max_age_seconds=float(pending_limits.get("max_age_seconds") or 0),
345 auto_abort_message=str(pending_limits.get("abort_message") or ""),
346 max_output_chars=int(
347 300
348 if pending_limits.get("max_output_chars") is None
349 else pending_limits.get("max_output_chars")
350 ),
351 output_limit_abort_message=str(pending_limits.get("output_limit_abort_message") or ""),
352 available_tool_names=extract_tool_names(normalized_data),
353 available_tool_schemas=extract_tool_schemas(normalized_data),
354 )
355 try:
356 request_debug_metadata = build_message_debug_metadata(
357 auth=self.auth,
358 request_format=request_format,
359 request_data=normalized_data,
360 input_text=context_text,
361 input_payload=request_input_payload(normalized_data, request_format),
362 request_id=pending.request_id,
363 resolved_model=model,
364 )
365 if extracted_messages:
366 for index, message_payload in enumerate(extracted_messages):
367 metadata = dict(message_payload.get("metadata") or {})
368 if index == len(extracted_messages) - 1:
369 metadata = {**metadata, **request_debug_metadata}
370 self.store.add_message(
371 conversation.id,
372 str(message_payload.get("role") or "user"),
373 str(message_payload.get("content") or ""),
374 metadata=metadata,
375 )
376 else:
377 self.store.add_message(
378 conversation.id,
379 "user",
380 context_text,
381 metadata={
382 "turn": "user",
383 "status": "pending",
384 "source": request_format,
385 **request_debug_metadata,
386 },
387 )
388 notify_new_message(
389 self._deps.system_config_store,
390 self.user_store,
391 owner,
392 conversation_title=updated_conversation.title or build_title(context_text),
393 message_text=context_text,
394 logger=self._logger,
395 )
396 self.store.update_conversation(
397 conversation.id,
398 owner,
399 metadata={
400 **updated_conversation.metadata,
401 "realtime_status": "waiting",
402 "realtime_draft_text": "",
403 "request_format": request_format,
404 },
405 )
406 self._notify(owner, conversation.id)
407 self._automation_rules.start_for_pending(pending)
408 except Exception:
409 self.pending_turns.discard(
410 conversation_id=conversation.id,
411 owner_id=owner,
412 )
413 raise
414 return PreparedTurn(pending=pending, conversation_id=updated_conversation.id)
415
416 def finalize_pending_turn(self, pending: PendingTurn) -> dict[str, Any]:
417 updated_conversation = self.store.get_conversation(pending.conversation_id, pending.owner_id)
418 if updated_conversation is None:
419 raise ValueError("conversation not found")
420 usage = estimate_usage(pending.input_text, pending.assistant_text)
421 try:
422 messages = self.store.get_messages(pending.conversation_id, pending.owner_id)
423 except ValueError:
424 messages = []
425 message_metadata = find_response_message_metadata(messages, pending.response_id)
426 tool_name = str(message_metadata.get("tool_name", "")).strip()
427 tool_call_id = str(message_metadata.get("tool_call_id", "")).strip()
428 arguments = str(message_metadata.get("arguments", "")).strip()
429
430 if pending.request_format == "chat_completions":
431 return build_chat_completion_response(
432 response_id=pending.response_id,
433 model=pending.model,
434 assistant_text=pending.assistant_text,
435 usage=usage,
436 response_mode=pending.response_mode,
437 tool_name=tool_name,
438 tool_call_id=tool_call_id,
439 arguments=arguments,
440 )
441 if pending.request_format == "anthropic_messages":
442 return build_anthropic_message_response(
443 response_id=pending.response_id,
444 model=pending.model,
445 assistant_text=pending.assistant_text,
446 usage=usage,
447 response_mode=pending.response_mode,
448 tool_name=tool_name,
449 tool_call_id=tool_call_id,
450 arguments=arguments,
451 )
452 payload = build_openai_response(
453 response_id=pending.response_id,
454 model=pending.model,
455 conversation_id=updated_conversation.id,
456 assistant_text=pending.assistant_text,
457 usage=usage,
458 output_items=pending.response_output_items or None,
459 output_text=pending.response_output_text,
460 )
461 payload["conversation"] = updated_conversation.to_dict()
462 payload["input_text"] = pending.input_text
463 return payload
464
465 def complete_manual_output(self, data: dict[str, Any]):
466 if not isinstance(data, dict):
467 return {"error": "request body must be a JSON object"}, 400
468 mode = str(data.get("mode", "assistant_message")).strip() or "assistant_message"
469 if mode not in {"assistant_message", "thinking", "tool_call"}:
470 return {"error": "unsupported mode"}, 400
471 text = str(data.get("text", "")).strip()
472 tool_name = str(data.get("tool_name", "")).strip()
473 tool_call_id = str(data.get("tool_call_id", "")).strip()
474 reasoning_stream_mode = str(data.get("reasoning_stream_mode", "")).strip()
475 if mode == "tool_call":
476 if not tool_name:
477 return {"error": "tool_name is required"}, 400
478 if not text:
479 return {"error": "tool arguments are required"}, 400
480 if not tool_call_id:
481 tool_call_id = f"call_{uuid.uuid4().hex[:24]}"
482 conversation_id = str(data.get("conversation_id", "")).strip()
483 owner = self.auth.owner_id()
484 if not conversation_id:
485 return {"error": "conversation_id is required"}, 400
486
487 pending = self.pending_turns.get_by_conversation(conversation_id)
488 if pending is None:
489 return {"error": "conversation is not waiting for a reply"}, 409
490 if pending.owner_id != owner:
491 return {"error": "conversation not found"}, 404
492
493 try:
494 if mode == "tool_call":
495 pending, assistant_metadata = self._output_controller.complete_tool_call(
496 conversation_id=conversation_id,
497 owner_id=owner,
498 tool_name=tool_name,
499 arguments=text,
500 provider="human",
501 model=str(data.get("model") or pending.model or "mock-gpt-4.1-mini"),
502 tool_call_id=tool_call_id,
503 reasoning_stream_mode=reasoning_stream_mode,
504 )
505 assistant_text = pending.assistant_text
506 else:
507 pending, assistant_metadata = self._output_controller.complete_assistant_message(
508 conversation_id=conversation_id,
509 owner_id=owner,
510 provider="human",
511 model=str(data.get("model") or pending.model or "mock-gpt-4.1-mini"),
512 reasoning_stream_mode=reasoning_stream_mode,
513 )
514 assistant_text = pending.assistant_text
515 except ValueError as error:
516 return {"error": str(error)}, 409
517
518 conversation = self.store.get_conversation(conversation_id, owner)
519 return {
520 "ok": True,
521 "conversation": conversation.to_dict() if conversation else None,
522 "message": {
523 "role": "assistant" if mode == "assistant_message" else "tool_call",
524 "content": assistant_text,
525 "response_id": pending.response_id,
526 "metadata": assistant_metadata,
527 },
528 }
529
530 def add_manual_output_delta(self, data: dict[str, Any]):
531 if not isinstance(data, dict):
532 return {"error": "request body must be a JSON object"}, 400
533 text = normalize_message_text(str(data.get("text", "")).strip())
534 if not text:
535 return {"error": "text is required"}, 400
536 conversation_id = str(data.get("conversation_id", "")).strip()
537 reasoning_stream_mode = str(data.get("reasoning_stream_mode", "")).strip()
538 chunk_kind = "thinking" if str(data.get("kind", "")).strip() == "thinking" else "answer"
539 owner = self.auth.owner_id()
540 if not conversation_id:
541 return {"error": "conversation_id is required"}, 400
542
543 try:
544 pending = self._output_controller.add_text_delta(
545 conversation_id=conversation_id,
546 owner_id=owner,
547 text=text,
548 reasoning_stream_mode=reasoning_stream_mode,
549 kind=chunk_kind,
550 )
551 except ValueError as error:
552 return {"error": str(error)}, 409
553
554 return {
555 "ok": True,
556 "conversation_id": pending.conversation_id,
557 "request_id": pending.request_id,
558 "draft_text": pending.draft_text,
559 "draft_length": len(pending.draft_text),
560 }
561
562 def _notify(self, owner_id: str, conversation_id: str) -> None:
563 if self._publish_sync is None:
564 return
565 self._publish_sync(owner_id, conversation_id)
566