From 6db2854ba31f7f4f3dfd598c40ee09dfa2926584 Mon Sep 17 00:00:00 2001 From: Mateusz Date: Fri, 14 Aug 2026 14:09:06 +0200 Subject: [PATCH 1/3] fix(acp): stream in-progress tool cards and still-running heartbeats Emit a started tool summary on the first non-terminal ACP tool_call, keep the completion card for terminal status, and surface periodic still-running lines so long tools are not silent until they finish. Co-authored-by: Cursor --- src/connectors/acp_core/base_connector.py | 308 ++++++++++-------- src/connectors/acp_core/tool_markdown.py | 81 +++-- src/connectors/acp_core/types.py | 2 + .../acp_core/test_base_connector.py | 150 +++++++-- .../connectors/acp_core/test_tool_markdown.py | 29 +- 5 files changed, 365 insertions(+), 205 deletions(-) diff --git a/src/connectors/acp_core/base_connector.py b/src/connectors/acp_core/base_connector.py index 63d25f77d..f98a9cc74 100644 --- a/src/connectors/acp_core/base_connector.py +++ b/src/connectors/acp_core/base_connector.py @@ -27,6 +27,8 @@ extract_tool_name, extract_tool_output, format_acp_tool_completion_summary, + format_acp_tool_heartbeat_line, + format_acp_tool_started_summary, is_terminal_tool_status, iter_coalesced_acp_tool_session_dicts, payload_utf8_byte_length, @@ -78,6 +80,7 @@ DEFAULT_PROCESS_TIMEOUT = 300.0 DEFAULT_IDLE_TIMEOUT = 30.0 +DEFAULT_ACP_TOOL_HEARTBEAT_SECONDS = 30.0 MAX_RESPONSE_LINE_SIZE = 10 * 1024 * 1024 MAX_STDERR_TAIL_SIZE = 16 * 1024 ACP_UPDATE_METHOD = "session/update" @@ -130,8 +133,7 @@ def _canonical_chat_message_for_history_hash(message: ChatMessage) -> dict[str, def _hash_chat_messages_prefix_stable( - messages: Sequence[ChatMessage], - end_exclusive: int, + messages: Sequence[ChatMessage], end_exclusive: int ) -> str: """SHA-256 hex digest of the first ``end_exclusive`` messages (conversation prefix).""" @@ -171,10 +173,7 @@ def cancel(self) -> None: return connector = cast(Any, self._connector) task = loop.create_task( - connector._cancel_active_request( - self._runtime, - self._prompt_request_id, - ) + connector._cancel_active_request(self._runtime, self._prompt_request_id) ) task.add_done_callback(lambda t: t.exception() if not t.cancelled() else None) @@ -197,6 +196,7 @@ def __init__( self._model = "auto" self._process_timeout = DEFAULT_PROCESS_TIMEOUT self._idle_timeout = DEFAULT_IDLE_TIMEOUT + self._acp_tool_heartbeat_seconds = DEFAULT_ACP_TOOL_HEARTBEAT_SECONDS self._runtime_pool_lock = asyncio.Lock() self._runtimes: dict[tuple[str, str, str], RuntimeT] = {} @@ -426,8 +426,7 @@ async def _acquire_runtime( ) -> RuntimeT: project_dir = self._resolve_project_dir_for_request(request) requested_model = strip_vendor_prefix( - request.effective_model or self._model, - self.VENDOR_PREFIX, + request.effective_model or self._model, self.VENDOR_PREFIX ) client_session_id = self._resolve_client_session_id(request) runtime_key = self._build_runtime_key( @@ -463,9 +462,7 @@ def _resolve_project_dir_for_request( options = cast(dict[str, Any] | None, request.options) usable = first_usable_workspace_dir( - extra_dict, - options, - is_usable=is_usable_workspace_directory, + extra_dict, options, is_usable=is_usable_workspace_directory ) if usable is not None: return usable @@ -475,10 +472,7 @@ def _resolve_project_dir_for_request( if hint is not None: raise BackendError( message=f"Unusable ACP workspace directory: {hint}", - details={ - "code": ACP_MISSING_PROJECT_WORKSPACE_CODE, - "hint": hint, - }, + details={"code": ACP_MISSING_PROJECT_WORKSPACE_CODE, "hint": hint}, ) raise BackendError( message=( @@ -490,10 +484,7 @@ def _resolve_project_dir_for_request( hint = first_workspace_hint_str(extra_dict, options) if hint is not None and logger.isEnabledFor(logging.DEBUG): - logger.debug( - "Ignoring unusable ACP project_dir override: %s", - hint, - ) + logger.debug("Ignoring unusable ACP project_dir override: %s", hint) if self._default_project_dir is None: raise ConfigurationError( @@ -503,9 +494,7 @@ def _resolve_project_dir_for_request( return self._default_project_dir async def _reap_idle_runtime( - self, - runtime_key: tuple[str, str, str], - runtime: RuntimeT, + self, runtime_key: tuple[str, str, str], runtime: RuntimeT ) -> RuntimeT: """Drop idle subprocesses and swap in a fresh :class:`ACPProcessRuntime` slot. @@ -541,9 +530,7 @@ async def _reap_idle_runtime( current = self._runtimes.get(runtime_key) if current is runtime: replacement = self._create_runtime( - runtime.project_dir, - runtime.model, - runtime.client_session_id, + runtime.project_dir, runtime.model, runtime.client_session_id ) self._runtimes[runtime_key] = replacement return replacement @@ -655,9 +642,7 @@ async def _kill_all_runtimes(self) -> None: await self._kill_runtime(runtime) def _cleanup_runtime_state( - self, - runtime: RuntimeT, - process: subprocess.Popen[bytes] | None = None, + self, runtime: RuntimeT, process: subprocess.Popen[bytes] | None = None ) -> None: self._stop_stderr_drain(runtime) self._cleanup_process(process or runtime.process) @@ -748,9 +733,7 @@ def _read_all() -> None: stderr_bytes.append(bytes(stream.read())) reader = threading.Thread( - target=_read_all, - name=f"acp-stderr-fallback-{process.pid}", - daemon=True, + target=_read_all, name=f"acp-stderr-fallback-{process.pid}", daemon=True ) reader.start() while reader.is_alive(): @@ -808,8 +791,7 @@ def _write() -> None: try: await asyncio.wait_for( - asyncio.to_thread(_write), - timeout=self._process_timeout, + asyncio.to_thread(_write), timeout=self._process_timeout ) except asyncio.TimeoutError as exc: # A blocked pipe means the child is no longer making progress. Tear @@ -823,10 +805,7 @@ def _write() -> None: runtime.last_activity = time.monotonic() async def _send_jsonrpc_message( - self, - runtime: RuntimeT, - method: str, - params: dict[str, Any], + self, runtime: RuntimeT, method: str, params: dict[str, Any] ) -> int: message_id = self._get_next_message_id(runtime) payload = { @@ -849,8 +828,7 @@ async def _send_jsonrpc_result( self, runtime: RuntimeT, request_id: int, result: dict[str, Any] ) -> None: await self._write_json_line( - runtime, - {"jsonrpc": "2.0", "id": request_id, "result": result}, + runtime, {"jsonrpc": "2.0", "id": request_id, "result": result} ) async def _read_jsonrpc_message(self, runtime: RuntimeT) -> ACPNotification | None: @@ -897,9 +875,7 @@ def _read_limited() -> bytes: ) from exc async def _await_response( - self, - runtime: RuntimeT, - request_id: int, + self, runtime: RuntimeT, request_id: int ) -> ACPNotification: deadline = time.monotonic() + self._process_timeout while True: @@ -911,8 +887,7 @@ async def _await_response( ) response = await asyncio.wait_for( - self._read_jsonrpc_message(runtime), - timeout=remaining, + self._read_jsonrpc_message(runtime), timeout=remaining ) if response is None: continue @@ -1060,11 +1035,7 @@ def _acp_progress_reasoning_line( return None def _resolve_tool_stream_key( - self, - runtime: RuntimeT, - tc: dict[str, Any], - *, - for_new_invocation: bool, + self, runtime: RuntimeT, tc: dict[str, Any], *, for_new_invocation: bool ) -> str: ck = extract_tool_correlation_key(tc) if ck: @@ -1139,6 +1110,53 @@ def _acp_try_emit_tool_summary( acc.pending_terminal_summary = False return [AcpStreamPiece(content=text)] + def _acp_start_summary_pieces( + self, acc: AcpToolStreamAccum + ) -> list[AcpStreamPiece]: + if acc.start_emitted or acc.summary_emitted or not acc.started_wall_iso: + return [] + text = format_acp_tool_started_summary( + acc.tool_name, + input_payload=acc.last_input, + input_bytes=acc.last_input_bytes, + started_iso=acc.started_wall_iso, + ) + acc.start_emitted = True + acc.last_heartbeat_perf = ( + acc.started_perf if acc.started_perf > 0 else time.perf_counter() + ) + return [AcpStreamPiece(content=text)] + + def _acp_in_progress_heartbeat_pieces( + self, runtime: RuntimeT + ) -> list[AcpStreamPiece]: + interval = self._acp_tool_heartbeat_seconds + if interval <= 0: + return [] + now = time.perf_counter() + out: list[AcpStreamPiece] = [] + for acc in runtime.acp_tool_stream_accum.values(): + if acc.summary_emitted or not acc.start_emitted: + continue + last = acc.last_heartbeat_perf or acc.started_perf + if last > 0 and (now - last) < interval: + continue + started = acc.started_perf if acc.started_perf > 0 else now + elapsed = max(0.0, now - started) + out.append( + AcpStreamPiece( + content=format_acp_tool_heartbeat_line(acc.tool_name, elapsed) + ) + ) + acc.last_heartbeat_perf = now + return out + + def _acp_has_in_progress_tools(self, runtime: RuntimeT) -> bool: + return any( + acc.start_emitted and not acc.summary_emitted + for acc in runtime.acp_tool_stream_accum.values() + ) + def _acp_terminal_summary_pieces( self, acc: AcpToolStreamAccum, @@ -1203,6 +1221,8 @@ def _acp_pieces_for_tool_call( status_str, allow_defer=not batch_multi, ) + if not pieces and not is_terminal_tool_status(status_str): + pieces = self._acp_start_summary_pieces(acc) out.extend(pieces) return out @@ -1223,9 +1243,12 @@ def _acp_pieces_for_tool_call_update( self._acp_update_tool_sizes_from_merged(acc, merged) status_raw = merged.get("status") or merged.get("state") status_str = status_raw.strip() if isinstance(status_raw, str) else None - return self._acp_terminal_summary_pieces( + pieces = self._acp_terminal_summary_pieces( acc, merged, status_str, allow_defer=True ) + if pieces or is_terminal_tool_status(status_str): + return pieces + return self._acp_start_summary_pieces(acc) def _flush_incomplete_acp_tool_streams( self, runtime: RuntimeT @@ -1250,8 +1273,7 @@ def _session_update_to_stream_pieces( except Exception: if logger.isEnabledFor(logging.DEBUG): logger.debug( - "ACP session/update params could not be parsed", - exc_info=True, + "ACP session/update params could not be parsed", exc_info=True ) return [] upd = envelope.update @@ -1264,8 +1286,7 @@ def _session_update_to_stream_pieces( if kind == ACP_AGENT_MESSAGE_CHUNK: if text: return self._prepend_thinking_close_if_needed( - runtime, - [AcpStreamPiece(content=text)], + runtime, [AcpStreamPiece(content=text)] ) return [] if kind == ACP_AGENT_THOUGHT_CHUNK: @@ -1274,13 +1295,11 @@ def _session_update_to_stream_pieces( return [] if kind == "tool_call": return self._prepend_thinking_close_if_needed( - runtime, - self._acp_pieces_for_tool_call(runtime, upd), + runtime, self._acp_pieces_for_tool_call(runtime, upd) ) if kind == "tool_call_update": return self._prepend_thinking_close_if_needed( - runtime, - self._acp_pieces_for_tool_call_update(runtime, upd), + runtime, self._acp_pieces_for_tool_call_update(runtime, upd) ) progress = self._acp_progress_reasoning_line(kind, upd) @@ -1307,25 +1326,38 @@ def _session_update_to_stream_piece( ) async def _iter_acp_stream_pieces( - self, - runtime: RuntimeT, - prompt_request_id: int, - response_model: str, + self, runtime: RuntimeT, prompt_request_id: int, response_model: str ) -> AsyncGenerator[AcpStreamPiece, None]: runtime.acp_tool_stream_accum.clear() runtime.acp_anon_tool_seq = 0 runtime.acp_last_anon_stream_key = None runtime.acp_thinking_block_open = False + deadline = time.monotonic() + self._process_timeout + read_task: asyncio.Task[ACPNotification | None] | None = None try: while True: + remaining = deadline - time.monotonic() + if remaining <= 0: + raise asyncio.TimeoutError() + interval = self._acp_tool_heartbeat_seconds + slice_for_heartbeat = interval > 0 and self._acp_has_in_progress_tools( + runtime + ) + wait_timeout = ( + min(interval, remaining) if slice_for_heartbeat else remaining + ) + if runtime.cancellation_event is not None: - read_task = asyncio.create_task(self._read_jsonrpc_message(runtime)) + if read_task is None: + read_task = asyncio.create_task( + self._read_jsonrpc_message(runtime) + ) cancel_task = asyncio.create_task(runtime.cancellation_event.wait()) try: done, pending = await asyncio.wait( {read_task, cancel_task}, return_when=asyncio.FIRST_COMPLETED, - timeout=self._process_timeout, + timeout=wait_timeout, ) except asyncio.CancelledError: read_task.cancel() @@ -1334,26 +1366,56 @@ async def _iter_acp_stream_pieces( await read_task with contextlib.suppress(asyncio.CancelledError): await cancel_task + read_task = None raise + if cancel_task not in done: + cancel_task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await cancel_task if not done: - for t in pending: - t.cancel() - with contextlib.suppress(asyncio.CancelledError): - await t - raise asyncio.TimeoutError() + for piece in self._acp_in_progress_heartbeat_pieces(runtime): + if piece.content or piece.reasoning_content: + yield piece + continue for t in pending: + if t is read_task: + continue t.cancel() with contextlib.suppress(asyncio.CancelledError): await t if cancel_task in done: + if read_task is not None: + read_task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await read_task + read_task = None return response = read_task.result() + read_task = None else: - response = await asyncio.wait_for( - self._read_jsonrpc_message(runtime), - timeout=self._process_timeout, - ) + if read_task is None: + read_task = asyncio.create_task( + self._read_jsonrpc_message(runtime) + ) + try: + response = await asyncio.wait_for( + asyncio.shield(read_task), timeout=wait_timeout + ) + read_task = None + except asyncio.TimeoutError: + if time.monotonic() >= deadline: + pending_read = read_task + if pending_read is not None: + pending_read.cancel() + with contextlib.suppress(asyncio.CancelledError): + await pending_read + read_task = None + raise + for piece in self._acp_in_progress_heartbeat_pieces(runtime): + if piece.content or piece.reasoning_content: + yield piece + continue if response is None: continue @@ -1387,12 +1449,14 @@ async def _iter_acp_stream_pieces( message="Timeout waiting for ACP response", details={"timeout": self._process_timeout, "model": response_model}, ) from exc + finally: + if read_task is not None and not read_task.done(): + read_task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await read_task async def _iter_stream_pieces( - self, - runtime: RuntimeT, - request_id: int, - response_model: str, + self, runtime: RuntimeT, request_id: int, response_model: str ) -> AsyncGenerator[AcpStreamPiece, None]: """Dispatch to the protocol-specific stream-piece iterator. @@ -1452,9 +1516,7 @@ async def _collect_non_streaming_response( ], ) envelope = ResponseEnvelope( - content=response.model_dump(exclude_none=True), - headers={}, - status_code=200, + content=response.model_dump(exclude_none=True), headers={}, status_code=200 ) return self.ensure_usage_in_response( envelope, list(request.processed_messages), requested_model @@ -1475,13 +1537,7 @@ def _create_sse_chunk_from_piece( "object": "chat.completion.chunk", "created": int(time.time()), "model": model, - "choices": [ - { - "index": 0, - "delta": delta, - "finish_reason": None, - } - ], + "choices": [{"index": 0, "delta": delta, "finish_reason": None}], } return f"data: {json.dumps(payload)}\n\n" @@ -1493,13 +1549,7 @@ def _create_sse_done_chunk() -> str: def _is_terminal_finish_reason(value: Any) -> bool: """Return whether a finish reason terminates an OpenAI stream.""" - return value in { - "stop", - "length", - "content_filter", - "tool_calls", - "error", - } + return value in {"stop", "length", "content_filter", "tool_calls", "error"} @classmethod def _stream_chunk_is_terminal(cls, chunk: ProcessedResponse) -> bool: @@ -1567,10 +1617,7 @@ def _stream_chunk_is_terminal(cls, chunk: ProcessedResponse) -> bool: return False async def _stream_response( - self, - runtime: RuntimeT, - requested_model: str, - prompt_request_id: int, + self, runtime: RuntimeT, requested_model: str, prompt_request_id: int ) -> AsyncGenerator[ProcessedResponse, None]: chunk_id = str(uuid.uuid4()) async for piece in self._iter_stream_pieces( @@ -1587,9 +1634,7 @@ async def _stream_response( yield ProcessedResponse(content=self._create_sse_done_chunk()) async def _compute_history_and_user_message( - self, - runtime: RuntimeT, - messages: Sequence[ChatMessage], + self, runtime: RuntimeT, messages: Sequence[ChatMessage] ) -> tuple[str, HistoryState]: """Compute the user-message text and resulting history state for a turn. @@ -1663,9 +1708,7 @@ async def _compute_history_and_user_message( return user_message, new_history_state async def _prepare_turn_request_locked( - self, - runtime: RuntimeT, - request: ConnectorChatCompletionsRequest, + self, runtime: RuntimeT, request: ConnectorChatCompletionsRequest ) -> tuple[int, str]: """Build ``session/prompt`` text and JSON-RPC id under ``runtime.request_lock``. @@ -1691,8 +1734,7 @@ async def _prepare_turn_request_locked( raise BackendError(message="No user message found in request") requested_model = request.effective_model or add_vendor_prefix( - runtime.model, - self.VENDOR_PREFIX, + runtime.model, self.VENDOR_PREFIX ) prompt_params: dict[str, Any] = { "sessionId": runtime.session_id, @@ -1700,9 +1742,7 @@ async def _prepare_turn_request_locked( "messageId": str(uuid.uuid4()), } prompt_request_id = await self._send_jsonrpc_message( - runtime, - "session/prompt", - prompt_params, + runtime, "session/prompt", prompt_params ) runtime.history_state = new_history_state return prompt_request_id, requested_model @@ -1842,17 +1882,12 @@ async def _acquire_runtime_request_lock(self, runtime: RuntimeT) -> None: ) async def _wait_for_process_exit( - self, - process: subprocess.Popen[bytes], - timeout_s: float, + self, process: subprocess.Popen[bytes], timeout_s: float ) -> bool: if process.poll() is not None: return True try: - await asyncio.wait_for( - asyncio.to_thread(process.wait), - timeout=timeout_s, - ) + await asyncio.wait_for(asyncio.to_thread(process.wait), timeout=timeout_s) return True except asyncio.TimeoutError: return False @@ -1860,10 +1895,7 @@ async def _wait_for_process_exit( return process.poll() is not None async def _attempt_graceful_cancel( - self, - runtime: RuntimeT, - request_id: int, - total_timeout_s: float, + self, runtime: RuntimeT, request_id: int, total_timeout_s: float ) -> bool: process = runtime.process if process is None or process.poll() is not None: @@ -1890,8 +1922,7 @@ async def _attempt_graceful_cancel( continue exited = await self._wait_for_process_exit( - process, - timeout_s=min(remaining, 1.5), + process, timeout_s=min(remaining, 1.5) ) if exited: return True @@ -1901,8 +1932,7 @@ async def _attempt_graceful_cancel( process.stdin.close() remaining = deadline - time.monotonic() if remaining > 0 and await self._wait_for_process_exit( - process, - timeout_s=min(remaining, 3.0), + process, timeout_s=min(remaining, 3.0) ): return True @@ -1969,9 +1999,7 @@ async def _cancel_active_request( ) graceful_cancelled = await self._attempt_graceful_cancel( - runtime, - prompt_request_id, - ACP_GRACEFUL_CANCEL_TIMEOUT_SECONDS, + runtime, prompt_request_id, ACP_GRACEFUL_CANCEL_TIMEOUT_SECONDS ) if graceful_cancelled: @@ -2008,8 +2036,7 @@ async def _cancel_active_request( return True async def chat_completions( # type: ignore[override] - self, - request: ConnectorChatCompletionsRequest, + self, request: ConnectorChatCompletionsRequest ) -> ResponseEnvelope | StreamingResponseEnvelope: if ( request.cancellation_coordinator is not None @@ -2033,9 +2060,10 @@ async def chat_completions( # type: ignore[override] if bool(getattr(request.request, "stream", False)): await self._acquire_runtime_request_lock(runtime) try: - prompt_request_id, requested_model = ( - await self._prepare_turn_request_locked(runtime, request) - ) + ( + prompt_request_id, + requested_model, + ) = await self._prepare_turn_request_locked(runtime, request) except Exception: runtime.request_lock.release() raise @@ -2046,9 +2074,7 @@ async def chat_completions( # type: ignore[override] async def _cancel_streaming_request() -> None: await self._cancel_active_request( - runtime, - prompt_request_id, - expected_generation=request_generation, + runtime, prompt_request_id, expected_generation=request_generation ) stream_id: str | None = getattr(request.request, "session_id", None) @@ -2057,10 +2083,7 @@ async def _cancel_streaming_request() -> None: async def _stream_with_keepalive() -> AsyncIterator[ProcessedResponse]: inner = self._stream_response_with_lock( - runtime, - requested_model, - prompt_request_id, - request_generation, + runtime, requested_model, prompt_request_id, request_generation ) async for chunk in wrap_processed_stream_with_idle_keepalive( inner, @@ -2087,9 +2110,10 @@ async def _stream_with_keepalive() -> AsyncIterator[ProcessedResponse]: # acquire the lock against a half-torn-down child. await self._acquire_runtime_request_lock(runtime) try: - prompt_request_id, requested_model = ( - await self._prepare_turn_request_locked(runtime, request) - ) + ( + prompt_request_id, + requested_model, + ) = await self._prepare_turn_request_locked(runtime, request) cancellable_registered = False if ( request.cancellation_coordinator is not None diff --git a/src/connectors/acp_core/tool_markdown.py b/src/connectors/acp_core/tool_markdown.py index b8eea75bb..0263bdf4e 100644 --- a/src/connectors/acp_core/tool_markdown.py +++ b/src/connectors/acp_core/tool_markdown.py @@ -95,6 +95,48 @@ def _filtered_tool_input_fallback(tc: dict[str, Any]) -> Any | None: return fallback or None +def _render_tool_arguments(input_payload: Any) -> str: + if isinstance(input_payload, str): + try: + render_value = json.loads(input_payload) + except (json.JSONDecodeError, TypeError): + rendered_input = input_payload + else: + rendered_input = json.dumps( + render_value, ensure_ascii=False, separators=(",", ":"), default=str + ) + else: + try: + rendered_input = json.dumps( + input_payload, ensure_ascii=False, separators=(",", ":"), default=str + ) + except (TypeError, ValueError): + rendered_input = str(input_payload) + if len(rendered_input) > _MAX_TOOL_ARGUMENT_CHARS: + rendered_input = ( + rendered_input[:_MAX_TOOL_ARGUMENT_CHARS].rstrip() + "… [truncated]" + ) + return rendered_input + + +def format_acp_tool_started_summary( + tool_name: str, *, input_payload: Any | None, input_bytes: int, started_iso: str +) -> str: + """Compact fenced block when a tool starts (no Ended/Output).""" + lines = ["---", "```text", f"Tool: {tool_name}", "Status: started"] + if input_payload is not None: + lines.append(f"Arguments: {_render_tool_arguments(input_payload)}") + lines.extend( + [f"Input size: {input_bytes} bytes", f"Started: {started_iso}", "```", ""] + ) + return "\n".join(lines) + + +def format_acp_tool_heartbeat_line(tool_name: str, elapsed_s: float) -> str: + elapsed_display = int(elapsed_s) if elapsed_s >= 1 else 0 + return f"Tool still running: {tool_name} ({elapsed_display}s)\n" + + def format_acp_tool_completion_summary( tool_name: str, *, @@ -106,39 +148,9 @@ def format_acp_tool_completion_summary( elapsed_s: float, ) -> str: """Single compact fenced block after a tool finishes.""" - lines = [ - "---", - "```text", - f"Tool: {tool_name}", - ] + lines = ["---", "```text", f"Tool: {tool_name}"] if input_payload is not None: - if isinstance(input_payload, str): - try: - render_value = json.loads(input_payload) - except (json.JSONDecodeError, TypeError): - rendered_input = input_payload - else: - rendered_input = json.dumps( - render_value, - ensure_ascii=False, - separators=(",", ":"), - default=str, - ) - else: - try: - rendered_input = json.dumps( - input_payload, - ensure_ascii=False, - separators=(",", ":"), - default=str, - ) - except (TypeError, ValueError): - rendered_input = str(input_payload) - if len(rendered_input) > _MAX_TOOL_ARGUMENT_CHARS: - rendered_input = ( - rendered_input[:_MAX_TOOL_ARGUMENT_CHARS].rstrip() + "… [truncated]" - ) - lines.append(f"Arguments: {rendered_input}") + lines.append(f"Arguments: {_render_tool_arguments(input_payload)}") lines.extend( [ f"Input size: {input_bytes} bytes", @@ -159,10 +171,7 @@ def format_transcript_assistant_tool_record(name: str, arguments: Any) -> str: def format_transcript_tool_message_record( - *, - tool_call_id: str | None, - name: str | None, - content: Any, + *, tool_call_id: str | None, name: str | None, content: Any ) -> str: """History text for a ``role: tool`` message (output size only).""" label = (name or "").strip() or "tool" diff --git a/src/connectors/acp_core/types.py b/src/connectors/acp_core/types.py index 9e12b7858..ff8b78576 100644 --- a/src/connectors/acp_core/types.py +++ b/src/connectors/acp_core/types.py @@ -108,6 +108,8 @@ class AcpToolStreamAccum: last_input: Any | None = None last_input_bytes: int = 0 last_output_bytes: int = 0 + start_emitted: bool = False + last_heartbeat_perf: float = 0.0 @dataclass(frozen=True, slots=True) diff --git a/tests/unit/connectors/acp_core/test_base_connector.py b/tests/unit/connectors/acp_core/test_base_connector.py index 0173ce295..0f395f56a 100644 --- a/tests/unit/connectors/acp_core/test_base_connector.py +++ b/tests/unit/connectors/acp_core/test_base_connector.py @@ -82,9 +82,7 @@ def _make_request( ) request = CanonicalChatRequest( - model="dummy/model", - stream=stream, - messages=resolved_messages, + model="dummy/model", stream=stream, messages=resolved_messages ) context: ConnectorRequestContext | None = None if session_id is not None: @@ -143,6 +141,37 @@ async def test_acp_error_includes_structured_detail( } +@pytest.mark.asyncio +async def test_acp_quota_error_code_is_forwarded(connector: DummyAcpConnector) -> None: + runtime = connector._create_runtime(Path("/tmp/ws"), "dummy/model") + runtime.session_id = "dummy-session" + response = ACPNotification( + id=7, + error=ACPError( + code=-32003, + message="Model provider quota or rate limit exceeded", + data={"error": "RESOURCE_EXHAUSTED (code 429): Individual quota reached"}, + ), + ) + + with ( + patch.object( + connector, "_read_jsonrpc_message", AsyncMock(return_value=response) + ), + pytest.raises( + BackendError, + match=( + "ACP process error: Model provider quota or rate limit exceeded: " + "RESOURCE_EXHAUSTED" + ), + ) as exc_info, + ): + await connector._iter_acp_stream_pieces(runtime, 7, "dummy/model").__anext__() + + assert exc_info.value.details["code"] == -32003 + assert "RESOURCE_EXHAUSTED" in str(exc_info.value.details["data"]["error"]) + + @pytest.mark.asyncio async def test_windows_terminate_kills_tree_before_root_process( connector: DummyAcpConnector, @@ -235,10 +264,7 @@ def join(self, timeout: float | None = None) -> None: "src.connectors.acp_core.base_connector.subprocess.Popen", return_value=process, ), - patch( - "src.connectors.acp_core.base_connector.threading.Thread", - FakeThread, - ), + patch("src.connectors.acp_core.base_connector.threading.Thread", FakeThread), patch( "src.connectors.acp_core.base_connector.capture_acp_subprocess_identity", return_value=None, @@ -259,9 +285,7 @@ async def test_stale_stream_cancel_callback_does_not_touch_new_generation( runtime.active_request_generation = 2 cancelled = await connector._cancel_active_request( - runtime, - prompt_request_id=1, - expected_generation=1, + runtime, prompt_request_id=1, expected_generation=1 ) assert cancelled is False @@ -295,10 +319,7 @@ def join(self, *, timeout: float | None = None) -> None: await runtime.request_lock.acquire() with patch.object(connector, "_kill_runtime", AsyncMock()) as kill_mock: - cancelled = await connector._cancel_active_request( - runtime, - prompt_request_id=1, - ) + cancelled = await connector._cancel_active_request(runtime, prompt_request_id=1) assert cancelled is True kill_mock.assert_not_called() @@ -481,7 +502,10 @@ def test_session_update_flat_acp_tool_call_spec_shape( }, ) piece = connector._session_update_to_stream_piece(msg, runtime) - assert piece is None + assert piece is not None + assert piece.content is not None + assert "Status: started" in piece.content + assert "Tool: Reading configuration file" in piece.content flush = connector._flush_incomplete_acp_tool_streams(runtime) assert len(flush) == 1 assert flush[0].content is not None @@ -516,6 +540,7 @@ def test_session_update_tool_call_emits_summary_when_completed( assert "Tool: read_file" in joined assert 'Arguments: {"path":"/x"}' in joined assert "Input size:" in joined + assert "Status: started" not in joined def test_session_update_tool_call_update_emits_status_and_size_summary( @@ -554,7 +579,11 @@ def test_session_update_tool_call_update_emits_status_and_size_summary( ) first = connector._session_update_to_stream_pieces(call, runtime) second = connector._session_update_to_stream_pieces(upd, runtime) - assert first == [] + started = "".join(p.content or "" for p in first) + assert "Status: started" in started + assert "Tool: list_dir" in started + assert 'Arguments: {"path":"."}' in started + assert "Ended:" not in started joined = "".join(p.content or "" for p in second) assert "Status:" not in joined assert joined.startswith("---\n```text\nTool: list_dir") @@ -715,16 +744,15 @@ def test_session_update_tool_call_update_seen_empty_tail_returns_none( "sessionId": "s1", "update": { "sessionUpdate": "tool_call_update", - "toolCallUpdate": { - "toolCallId": "tc-1", - "name": "list_dir", - }, + "toolCallUpdate": {"toolCallId": "tc-1", "name": "list_dir"}, }, }, ) first = connector._session_update_to_stream_piece(call, runtime) second = connector._session_update_to_stream_piece(redundant, runtime) - assert first is None + assert first is not None + assert first.content is not None + assert "Status: started" in first.content assert second is None @@ -1027,8 +1055,7 @@ async def test_parallel_acquire_after_idle_reap_uses_same_pool_runtime( with ( patch( - "src.connectors.acp_core.base_connector.time.monotonic", - return_value=100.0, + "src.connectors.acp_core.base_connector.time.monotonic", return_value=100.0 ), patch.object(connector, "_terminate_process", AsyncMock()), ): @@ -1109,3 +1136,80 @@ async def test_requires_explicit_workspace_accepts_options_project_dir( req = replace(base, options={"project_dir": str(workspace)}) resolved = connector._resolve_project_dir_for_request(req) assert resolved == workspace.resolve() + + +def test_in_progress_heartbeat_emits_after_interval( + connector: DummyAcpConnector, +) -> None: + runtime = connector._create_runtime(Path("/tmp/ws"), "m") + connector._acp_tool_heartbeat_seconds = 0.01 + call = ACPNotification( + method="session/update", + params={ + "sessionId": "s1", + "update": { + "sessionUpdate": "tool_call", + "toolCall": { + "toolCallId": "tc-hb", + "name": "run_command", + "status": "in_progress", + "rawInput": {"CommandLine": "npm test"}, + }, + }, + }, + ) + start_pieces = connector._session_update_to_stream_pieces(call, runtime) + assert any("Status: started" in (p.content or "") for p in start_pieces) + acc = next(iter(runtime.acp_tool_stream_accum.values())) + acc.last_heartbeat_perf = acc.started_perf - 1.0 + acc.started_perf = acc.started_perf - 45.0 + heartbeats = connector._acp_in_progress_heartbeat_pieces(runtime) + assert len(heartbeats) == 1 + assert heartbeats[0].content is not None + assert heartbeats[0].content.startswith("Tool still running: run_command") + assert connector._acp_in_progress_heartbeat_pieces(runtime) == [] + + +@pytest.mark.asyncio +async def test_iter_acp_stream_emits_heartbeat_while_tool_in_progress( + connector: DummyAcpConnector, +) -> None: + runtime = connector._create_runtime(Path("/tmp/ws"), "m") + runtime.session_id = "dummy-session" + connector._acp_tool_heartbeat_seconds = 0.05 + connector._process_timeout = 2.0 + start = ACPNotification( + method="session/update", + params={ + "sessionId": "s1", + "update": { + "sessionUpdate": "tool_call", + "toolCallId": "tc-wait", + "title": "run_command", + "status": "in_progress", + }, + }, + ) + done = ACPNotification(id=7, result={"stopReason": "end_turn"}) + + async def delayed_messages() -> Any: + yield start + await asyncio.sleep(0.16) + yield done + + gen = delayed_messages() + + async def next_message(_runtime: Any) -> ACPNotification: + return await anext(gen) + + with patch.object(connector, "_read_jsonrpc_message", side_effect=next_message): + pieces = [ + piece + async for piece in connector._iter_acp_stream_pieces( + runtime, 7, "dummy/model" + ) + ] + + contents = [p.content or "" for p in pieces] + assert any("Status: started" in text for text in contents) + assert any("Tool still running: run_command" in text for text in contents) diff --git a/tests/unit/connectors/acp_core/test_tool_markdown.py b/tests/unit/connectors/acp_core/test_tool_markdown.py index 438c2ba9e..7366c1ae8 100644 --- a/tests/unit/connectors/acp_core/test_tool_markdown.py +++ b/tests/unit/connectors/acp_core/test_tool_markdown.py @@ -8,12 +8,36 @@ extract_tool_name, extract_tool_output, format_acp_tool_completion_summary, + format_acp_tool_heartbeat_line, + format_acp_tool_started_summary, is_terminal_tool_status, iter_coalesced_acp_tool_session_dicts, payload_utf8_byte_length, ) +def test_format_acp_tool_started_summary_shape() -> None: + text = format_acp_tool_started_summary( + "list_dir", + input_payload={"path": "src", "depth": 2}, + input_bytes=12, + started_iso="2026-01-01T00:00:00+00:00", + ) + assert text.startswith("---\n```text\nTool: list_dir") + assert "Status: started" in text + assert 'Arguments: {"path":"src","depth":2}' in text + assert "Started: 2026-01-01T00:00:00+00:00" in text + assert "Ended:" not in text + assert "Output size:" not in text + assert text.endswith("```\n") + + +def test_format_acp_tool_heartbeat_line() -> None: + assert format_acp_tool_heartbeat_line("run_command", 45.2) == ( + "Tool still running: run_command (45s)\n" + ) + + def test_extract_tool_fields() -> None: tc = { "toolCallId": "x1", @@ -112,10 +136,7 @@ def test_coalesce_tool_call_update_flattens_content_blocks() -> None: "sessionUpdate": "tool_call_update", "toolCallId": "call_001", "content": [ - { - "type": "content", - "content": {"type": "text", "text": "Found 3 files"}, - } + {"type": "content", "content": {"type": "text", "text": "Found 3 files"}} ], } merged = coalesce_acp_tool_call_update_session_dict(upd) From b0ffd95bbde8d520f4e5e9ee506d81c0079aa591 Mon Sep 17 00:00:00 2001 From: Mateusz Date: Fri, 14 Aug 2026 14:15:12 +0200 Subject: [PATCH 2/3] fix(acp): guard None read_task before result() Keep mypy and the in-progress wait loop aligned when cancellation wins the race. Co-authored-by: Cursor --- src/connectors/acp_core/base_connector.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/src/connectors/acp_core/base_connector.py b/src/connectors/acp_core/base_connector.py index f98a9cc74..25b214106 100644 --- a/src/connectors/acp_core/base_connector.py +++ b/src/connectors/acp_core/base_connector.py @@ -1391,6 +1391,8 @@ async def _iter_acp_stream_pieces( await read_task read_task = None return + if read_task is None: + continue response = read_task.result() read_task = None else: From 2b0603649a2f0d64a6a4aa197826964d7b644449 Mon Sep 17 00:00:00 2001 From: Mateusz Date: Fri, 14 Aug 2026 14:18:19 +0200 Subject: [PATCH 3/3] ci: diff architecture-check against PR commits only Working-tree vs origin/base picked up CRLF checkout noise and linted the whole tree, including unrelated nvidia scripts. Co-authored-by: Cursor --- .github/workflows/architecture-check.yml | 10 +++++++--- 1 file changed, 7 insertions(+), 3 deletions(-) diff --git a/.github/workflows/architecture-check.yml b/.github/workflows/architecture-check.yml index 567762c84..24b2dd549 100644 --- a/.github/workflows/architecture-check.yml +++ b/.github/workflows/architecture-check.yml @@ -12,6 +12,10 @@ jobs: steps: - uses: actions/checkout@v5 + with: + # Need merge-base history so PR diffs are commit-to-commit, not + # working-tree vs origin/base (CRLF checkout noise lists the whole repo). + fetch-depth: 0 - name: Set up Python uses: actions/setup-python@v6 @@ -26,9 +30,9 @@ jobs: - name: Run architectural linter on changed files run: | if [ "${{ github.event_name }}" == "pull_request" ]; then - # Get list of changed Python files in the PR - git fetch origin ${{ github.base_ref }} --depth=1 - CHANGED_FILES=$(git diff --name-only origin/${{ github.base_ref }} | grep "\.py$" || true) + # Get list of changed Python files in the PR (three-dot, commits only) + git fetch origin ${{ github.base_ref }} + CHANGED_FILES=$(git diff --name-only origin/${{ github.base_ref }}...HEAD | grep "\.py$" || true) else # Get list of changed Python files in the push git fetch origin ${{ github.event.before }} --depth=1