(self, prompt, btw: bool = False)
| 1133 | self._close_stream() |
| 1134 | |
| 1135 | def _run_turn(self, prompt, btw: bool = False) -> None: # prompt: str | list[ContentBlock] |
| 1136 | # btw=True → a "side question" (the original's /btw): run with full context |
| 1137 | # but DON'T persist the Q&A, so the main conversation isn't interrupted. |
| 1138 | from src.query.agent_loop_compat import run_query_as_agent_loop |
| 1139 | |
| 1140 | if self.init_error is not None: |
| 1141 | self._emit(_result_message( |
| 1142 | self.session_id, subtype="error", num_turns=0, |
| 1143 | result=self.init_error, is_error=True, error=self.init_error, |
| 1144 | )) |
| 1145 | return |
| 1146 | |
| 1147 | abort = AbortController() |
| 1148 | with self._lock: |
| 1149 | self._current_abort = abort |
| 1150 | # Wire the per-turn controller into the tool context so an interrupt |
| 1151 | # tears down an in-flight tool (Bash supervisor, etc.), not just the |
| 1152 | # model stream. A fresh controller per turn avoids a prior turn's |
| 1153 | # abort pre-cancelling the next one. |
| 1154 | if self.tool_context is not None: |
| 1155 | self.tool_context.abort_controller = abort |
| 1156 | |
| 1157 | # Snapshot history for a side-question turn so we can restore it after |
| 1158 | # (drops the ephemeral Q + A on every exit path via the finally below). |
| 1159 | _btw_snapshot = list(self.session.conversation.messages) if btw else None |
| 1160 | self.session.conversation.add_user_message(prompt) |
| 1161 | start = time.monotonic() |
| 1162 | |
| 1163 | def on_text_chunk(chunk: str) -> None: |
| 1164 | self._emit({ |
| 1165 | "type": "stream_event", |
| 1166 | "session_id": self.session_id, |
| 1167 | "event": { |
| 1168 | "type": "content_block_delta", |
| 1169 | "delta": {"type": "text_delta", "text": chunk}, |
| 1170 | }, |
| 1171 | }) |
| 1172 | |
| 1173 | def on_thinking_chunk(chunk: str) -> None: |
| 1174 | # Live reasoning deltas → a separate thinking delta the TUI renders |
| 1175 | # in its streaming thinking view (the original's live thinking, §3). |
| 1176 | self._emit({ |
| 1177 | "type": "stream_event", |
| 1178 | "session_id": self.session_id, |
| 1179 | "event": { |
| 1180 | "type": "content_block_delta", |
| 1181 | "delta": {"type": "thinking_delta", "thinking": chunk}, |
| 1182 | }, |
| 1183 | }) |
| 1184 | |
| 1185 | def on_message(message: Any) -> None: |
| 1186 | # Persist into the session conversation so the next turn pairs |
| 1187 | # tool_use ↔ tool_result, then ship the SDK envelope to the client. |
| 1188 | try: |
| 1189 | self.session.conversation.add_message(message.role, message.content) |
| 1190 | except Exception: # noqa: BLE001 |
| 1191 | logger.exception("[agent-server] persist failed") |
| 1192 | env = _sdk_envelope(message, self.session_id) |
no test coverage detected