diff --git a/acp_adapter/tools.py b/acp_adapter/tools.py index e5aefe9e52..8dda3affa3 100644 --- a/acp_adapter/tools.py +++ b/acp_adapter/tools.py @@ -104,9 +104,8 @@ def _plural(count: int, word: str, suffix: str = "s") -> str: def _failure(data: Args, prefix: str) -> Optional[str]: """Structured tool-level failure text (``success: false`` or ``error`` set).""" - if data.get("success") is False or data.get("error"): - return f"{prefix}: {data.get('error', 'unknown error')}" - return None + failed = data.get("success") is False or data.get("error") + return f"{prefix}: {data.get('error', 'unknown error')}" if failed else None def _structured(text_fallback: bool = False): @@ -115,10 +114,9 @@ def _structured(text_fallback: bool = False): def deco(fn: Callable[[str, Args, Args], Optional[str]]) -> _Formatter: def wrapper(tool_name: str, result: Optional[str], args: Optional[Args]) -> Optional[str]: - data = _json_loads_maybe(result) - if not isinstance(data, dict): - return _nonempty(result) if text_fallback else None - return fn(tool_name, data, args or {}) + if isinstance(data := _json_loads_maybe(result), dict): + return fn(tool_name, data, args or {}) + return _nonempty(result) if text_fallback else None return wrapper @@ -161,9 +159,7 @@ def _json_loads_maybe(value: Optional[str]) -> Any: def _truncate_text(text: str, limit: int = 5000) -> str: - if len(text) <= limit: - return text - return text[: max(0, limit - 100)] + f"\n... ({len(text)} chars total, truncated)" + return text if len(text) <= limit else text[: max(0, limit - 100)] + f"\n... ({len(text)} chars total, truncated)" def _fenced_text(text: str, language: str = "") -> str: @@ -186,10 +182,8 @@ def _tool_result_failed(result: Optional[str], tool_name: str | None = None) -> data = _json_loads_maybe(result) if not isinstance(data, dict): return False - if any(data.get(key) is False for key in ("success", "ok")): - return True exit_code = data.get("exit_code", data.get("returncode")) - if isinstance(exit_code, int) and exit_code != 0: + if any(data.get(key) is False for key in ("success", "ok")) or (isinstance(exit_code, int) and exit_code != 0): return True # Polished tools report failures as {"error": ...} without a success flag; # generic/plugin payloads stay conservative so diagnostics aren't marked failed. @@ -212,11 +206,9 @@ def _title_web_extract(args: Args) -> str: def _title_delegate(args: Args) -> str: - tasks = args.get("tasks") - if isinstance(tasks, list) and tasks: + if isinstance(tasks := args.get("tasks"), list) and tasks: return f"delegate batch ({len(tasks)} tasks)" - goal = args.get("goal", "") - return f"delegate: {_clip(goal, 60)}" if goal else "delegate task" + return f"delegate: {_clip(goal, 60)}" if (goal := args.get("goal", "")) else "delegate task" def _title_execute_code(args: Args) -> str: @@ -272,34 +264,27 @@ def build_tool_title(tool_name: str, args: Args) -> str: def _format_todo_result(tool_name: str, data: Args, args: Args) -> Optional[str]: if not isinstance(data.get("todos"), list): return None - summary = data.get("summary") if isinstance(data.get("summary"), dict) else {} icon = {"completed": "✅", "in_progress": "🔄", "pending": "⏳", "cancelled": "✗"} todos = [t for t in data["todos"] if isinstance(t, dict)] by_id = {str(t.get("id") or ""): t for t in todos} def _depth(item: Args) -> int: - depth, seen, node = 0, set(), item - while node is not None: - parent = str(node.get("parent") or "") - if not parent or parent not in by_id or parent in seen: - break + depth, seen = 0, set() + while (parent := str(item.get("parent") or "")) and parent in by_id and parent not in seen: seen.add(parent) depth += 1 - node = by_id.get(parent) + item = by_id[parent] return min(depth, 4) lines = ["**Todo list**", ""] for item in todos: - content = str(item.get("content") or item.get("id") or "").strip() - if content: + if content := str(item.get("content") or item.get("id") or "").strip(): lines.append(f"{' ' * _depth(item)}- {icon.get(str(item.get('status') or 'pending'), '•')} {content}") - if summary: + if isinstance(summary := data.get("summary"), dict) and summary: cancelled = summary.get("cancelled", 0) - lines.extend([ - "", - f"**Progress:** {summary.get('completed', 0)} completed, {summary.get('in_progress', 0)} in progress, " - f"{summary.get('pending', 0)} pending" + (f", {cancelled} cancelled" if cancelled else ""), - ]) + progress = f"{summary.get('completed', 0)} completed, {summary.get('in_progress', 0)} in progress, " + progress += f"{summary.get('pending', 0)} pending" + (f", {cancelled} cancelled" if cancelled else "") + lines += ["", f"**Progress:** {progress}"] return "\n".join(lines) @@ -307,16 +292,12 @@ def _format_todo_result(tool_name: str, data: Args, args: Args) -> Optional[str] def _format_read_file_result(tool_name: str, data: Args, a: Args) -> Optional[str]: if data.get("error") and not data.get("content"): return f"Read failed: {data.get('error')}" - content = data.get("content") - if not isinstance(content, str): + if not isinstance(content := data.get("content"), str): return None - range_bits = [f"from line {a['offset']}"] if a.get("offset") else [] - if a.get("limit"): - range_bits.append(f"limit {a['limit']}") + range_bits = [f"{label} {a[key]}" for key, label in (("offset", "from line"), ("limit", "limit")) if a.get(key)] header = f"Read {str(a.get('path') or data.get('path') or 'file').strip()}" header += f" ({', '.join(range_bits)})" if range_bits else "" - if data.get("total_lines") is not None: - header += f" — {data.get('total_lines')} total lines" + header += f" — {data.get('total_lines')} total lines" if data.get("total_lines") is not None else "" # read_file output is `|`-line-numbered; raw Markdown would let Zed parse # pipes as tables, so fence the payload to keep lines literal. return _truncate_text(f"{header}\n\n{_fenced_text(content)}") @@ -342,9 +323,8 @@ def _format_search_files_result(tool_name: str, data: Args, args: Args) -> Optio continue path = str(_first(match, "path", "file", "filename", default="?")) line = match.get("line") or match.get("line_number") - content = str(match.get("content") or match.get("text") or "").strip() lines.append(f"- {path}:{line}" if line else f"- {path}") - if content: + if content := str(match.get("content") or match.get("text") or "").strip(): lines.append(f" {_truncate_text(' '.join(content.split()), 300)}") if bool(data.get("truncated")) or len(matches) > shown: lines.extend(["", "Results truncated. Narrow the search, add file_glob, or use offset to page."]) @@ -370,34 +350,21 @@ def _format_execute_code_result(tool_name: str, data: Args, args: Args) -> Optio return _truncate_text("\n".join(parts)) -def _extract_markdown_headings(content: str, limit: int = 8) -> list[str]: - headings: list[str] = [] - for line in content.splitlines(): - heading = line.strip().lstrip("#").strip() if line.strip().startswith("#") else "" - if heading: - headings.append(heading) - if len(headings) >= limit: - break - return headings - - @_structured() def _format_skill_view_result(tool_name: str, data: Args, args: Args) -> Optional[str]: if data.get("success") is False: return f"Skill view failed: {data.get('error', 'unknown error')}" content = str(data.get("content") or "") - linked = data.get("linked_files") if isinstance(data.get("linked_files"), dict) else None lines = ["**Skill loaded**", "", f"- **Name:** `{data.get('name') or 'skill'}`", f"- **File:** `{data.get('file') or data.get('path') or 'SKILL.md'}`"] - description = str(data.get("description") or "").strip() - if description: + if description := str(data.get("description") or "").strip(): lines.append(f"- **Description:** {description}") if content: lines.append(f"- **Content:** {len(content):,} chars loaded into agent context") - if linked: + if isinstance(linked := data.get("linked_files"), dict) and linked: lines.append(f"- **Linked files:** {sum(len(v) for v in linked.values() if isinstance(v, list))}") - headings = _extract_markdown_headings(content) - if headings: + stripped = (line.strip() for line in content.splitlines()) + if headings := [h for h in (s.lstrip("#").strip() for s in stripped if s.startswith("#")) if h][:8]: lines.extend(["", "**Sections**", *(f"- {heading}" for heading in headings)]) lines.extend(["", "_Full skill content is available to the agent but hidden here to keep ACP readable._"]) return "\n".join(lines) @@ -412,14 +379,11 @@ def _format_skill_manage_result(tool_name: str, data: Args, a: Args) -> Optional lines = [f"**{status}**", "", f"- **Action:** `{action}`", f"- **Skill:** `{name}`"] if action != "delete": lines.append(f"- **File:** `{file_path}`") - message = str(data.get("message") or data.get("error") or "").strip() - if message: + if message := str(data.get("message") or data.get("error") or "").strip(): lines.append(f"- **Result:** {message}") - replacements = data.get("replacements") or data.get("replacement_count") - if replacements is not None: + if (replacements := data.get("replacements") or data.get("replacement_count")) is not None: lines.append(f"- **Replacements:** {replacements}") - path = str(data.get("path") or "").strip() - if path: + if path := str(data.get("path") or "").strip(): lines.append(f"- **Path:** `{path}`") return "\n".join(lines) @@ -434,9 +398,8 @@ def _format_web_search_result(tool_name: str, data: Args, args: Args) -> Optiona if not isinstance(item, dict): continue url = str(item.get("url") or "").strip() - desc = str(item.get("description") or "").strip() lines.append(f"• {str(item.get('title') or item.get('url') or 'result').strip()}" + (f" — {url}" if url else "")) - if desc: + if desc := str(item.get("description") or "").strip(): lines.append(f" {desc}") return _truncate_text("\n".join(lines)) @@ -446,8 +409,7 @@ def _format_web_extract_result(tool_name: str, data: Args, args: Args) -> Option """Return only web_extract errors for ACP; success stays compact via title.""" if data.get("success") is False and data.get("error"): return f"Web extract failed: {data.get('error')}" - results = data.get("results") - if not isinstance(results, list): + if not isinstance(results := data.get("results"), list): return None failures: list[str] = [] for item in results[:10]: @@ -456,12 +418,9 @@ def _format_web_extract_result(tool_name: str, data: Args, args: Args) -> Option continue url = str(item.get("url") or "").strip() title = str(item.get("title") or url or "Untitled").strip() - failures.append( - f"- {title}" + (f" — {url}" if url and url != title else "") + f"\n Error: {_truncate_text(error, limit=500)}" - ) - if not failures: - return None - return "\n".join([f"Web extract failed for {_plural(len(failures), 'URL')}", *failures]) + where = f" — {url}" if url and url != title else "" + failures.append(f"- {title}{where}\n Error: {_truncate_text(error, limit=500)}") + return "\n".join([f"Web extract failed for {_plural(len(failures), 'URL')}", *failures]) if failures else None @_structured(text_fallback=True) @@ -469,22 +428,18 @@ def _format_process_result(tool_name: str, data: Args, args: Args) -> Optional[s if data.get("success") is False and data.get("error"): return f"Process error: {data.get('error')}" action = _arg(args, "action", default="process") - processes = data.get("processes") - if isinstance(processes, list): + if isinstance(processes := data.get("processes"), list): lines = [f"Processes: {len(processes)}"] for proc in processes[:20]: if not isinstance(proc, dict): lines.append(f"- {proc}") continue - cmd = str(proc.get("command") or "").strip() bits = [str(proc.get("status") or ("exited" if proc.get("exited") else "running"))] - bits += [f"pid {proc['pid']}"] if proc.get("pid") is not None else [] - bits += [f"exit {proc['exit_code']}"] if proc.get("exit_code") is not None else [] + bits += [f"{lbl} {proc[k]}" for k, lbl in (("pid", "pid"), ("exit_code", "exit")) if proc.get(k) is not None] + cmd = str(proc.get("command") or "").strip() sid = _first(proc, "session_id", "id", default="?") lines.append(f"- `{sid}` — {', '.join(bits)}" + (f" — {cmd[:120]}" if cmd else "")) - if len(processes) > 20: - lines.append(f"... {len(processes) - 20} more process(es)") - return "\n".join(lines) + return "\n".join(lines + _more(processes, 20, " process(es)")) status = str(data.get("status") or data.get("state") or action).strip() sid = str(data.get("session_id") or args.get("session_id") or "").strip() @@ -494,10 +449,9 @@ def _format_process_result(tool_name: str, data: Args, args: Args) -> Optional[s lines.append(f"- **{label}:** {data.get(key)}") output = _first(data, "output", "new_output", "log", "stdout", default=None) error = _first(data, "error", "stderr", default=None) - if output: - lines.extend(["", "Output:", _truncate_text(str(output), limit=5000)]) - if error: - lines.extend(["", "Error:", _truncate_text(str(error), limit=2000)]) + for label, value, limit in (("Output:", output, 5000), ("Error:", error, 2000)): + if value: + lines.extend(["", label, _truncate_text(str(value), limit=limit)]) if data.get("message") and not output and not error: lines.append(str(data.get("message"))) return _truncate_text("\n".join(lines), limit=7000) @@ -505,11 +459,8 @@ def _format_process_result(tool_name: str, data: Args, args: Args) -> Optional[s @_structured() def _format_delegate_result(tool_name: str, data: Args, args: Args) -> Optional[str]: - results = data.get("results") - if data.get("error") and not isinstance(results, list): - return f"Delegation failed: {data.get('error')}" - if not isinstance(results, list): - return None + if not isinstance(results := data.get("results"), list): + return f"Delegation failed: {data.get('error')}" if data.get("error") else None total = data.get("total_duration_seconds") lines = [f"Delegation results: {_plural(len(results), 'task')}" + (f" in {total}s" if total is not None else "")] icon = {"completed": "✅", "failed": "✗", "error": "✗", "timeout": "⏱", "interrupted": "⚠"} @@ -523,14 +474,11 @@ def _format_delegate_result(tool_name: str, data: Args, args: Args) -> Optional[ bits += [f"role={item['_child_role']}"] if item.get("_child_role") else [] bits += [f"{item['duration_seconds']}s"] if item.get("duration_seconds") is not None else [] lines.extend(["", header + (" (" + ", ".join(bits) + ")" if bits else "")]) - summary, error = str(item.get("summary") or "").strip(), str(item.get("error") or "").strip() - if summary: - lines.append(_truncate_text(summary, limit=1200)) - if error: - lines.append("Error: " + _truncate_text(error, limit=800)) - trace = item.get("tool_trace") - names = [str(t.get("tool") or "?") for t in trace if isinstance(t, dict)] if isinstance(trace, list) else [] - if names: + for key, prefix, limit in (("summary", "", 1200), ("error", "Error: ", 800)): + if value := str(item.get(key) or "").strip(): + lines.append(prefix + _truncate_text(value, limit=limit)) + trace = item.get("tool_trace") if isinstance(item.get("tool_trace"), list) else [] + if names := [str(t.get("tool") or "?") for t in trace if isinstance(t, dict)]: lines.append("Tools: " + ", ".join(names[:12]) + (f" (+{len(names)-12})" if len(names) > 12 else "")) return _truncate_text("\n".join(lines), limit=8000) @@ -539,13 +487,10 @@ def _format_delegate_result(tool_name: str, data: Args, args: Args) -> Optional[ def _format_session_search_result(tool_name: str, data: Args, args: Args) -> Optional[str]: if data.get("success") is False: return f"Session search failed: {data.get('error', 'unknown error')}" - results = data.get("results") - if not isinstance(results, list): + if not isinstance(results := data.get("results"), list): return None - if (data.get("mode") or "search") == "recent": - lines = ["Recent sessions"] - else: - lines = ["Session search results" + _fmt(data.get("query"), " for `{}`", "")] + recent = (data.get("mode") or "search") == "recent" + lines = ["Recent sessions" if recent else "Session search results" + _fmt(data.get("query"), " for `{}`", "")] if not results: lines.append(str(data.get("message") or "No matching sessions found.")) return "\n".join(lines) @@ -557,8 +502,7 @@ def _format_session_search_result(tool_name: str, data: Args, args: Args) -> Opt count = item.get("message_count") meta = ", ".join(str(x) for x in [when, str(item.get("source") or "").strip(), f"{count} msgs" if count is not None else ""] if x) lines.append(f"- **{title}** (`{item.get('session_id') or '?'}`)" + (f" — {meta}" if meta else "")) - summary = str(item.get("summary") or item.get("preview") or "").strip() - if summary: + if summary := str(item.get("summary") or item.get("preview") or "").strip(): lines.append(" " + _truncate_text(" ".join(summary.split()), limit=500)) return _truncate_text("\n".join(lines), limit=7000) @@ -569,8 +513,7 @@ def _format_memory_result(tool_name: str, data: Args, args: Args) -> Optional[st target = str(data.get("target") or args.get("target") or "memory") if data.get("success") is False: lines = [f"✗ Memory {action} failed ({target})", str(data.get("error") or "unknown error")] - matches = data.get("matches") - if isinstance(matches, list) and matches: + if isinstance(matches := data.get("matches"), list) and matches: lines.extend(["Matches:", *(f"- {_truncate_text(str(m), 160)}" for m in matches[:5])]) return "\n".join(lines) lines = [f"✅ Memory {action} saved ({target})"] @@ -581,8 +524,7 @@ def _format_memory_result(tool_name: str, data: Args, args: Args) -> Optional[st if data.get("usage"): lines.append(f"Usage: {data.get('usage')}") # Never dump all memory entries into the ACP UI; only preview the new value. - preview = _arg(args, "content", "old_text") - if preview: + if preview := _arg(args, "content", "old_text"): lines.append("Preview: " + _truncate_text(preview, limit=300)) return "\n".join(lines) @@ -592,28 +534,22 @@ def _format_edit_result(tool_name: str, result: Optional[str], args: Optional[Ar path = str((args or {}).get("path") or "file").strip() done = f"✅ {tool_name} completed" + (f" for `{path}`" if path else "") if not isinstance(data, dict): - text = _nonempty(result) - return _truncate_text(text, limit=3000) if text else done - failed = _failure(data, f"{tool_name} failed for {path}") - if failed: + return _truncate_text(text, limit=3000) if (text := _nonempty(result)) else done + if failed := _failure(data, f"{tool_name} failed for {path}"): return failed lines = [done] - message = str(data.get("message") or "").strip() - if message: + if message := str(data.get("message") or "").strip(): lines.append(message) - replacements = data.get("replacements") or data.get("replacement_count") - if replacements is not None: + if (replacements := data.get("replacements") or data.get("replacement_count")) is not None: lines.append(f"Replacements: {replacements}") - files = data.get("files_modified") - if files and isinstance(files, list): + if isinstance(files := data.get("files_modified"), list) and files: lines.append("Files: " + ", ".join(f"`{f}`" for f in files[:8])) return "\n".join(lines) @_structured(text_fallback=True) def _format_browser_result(tool_name: str, data: Args, args: Args) -> Optional[str]: - failed = _failure(data, f"{tool_name} failed") - if failed: + if failed := _failure(data, f"{tool_name} failed"): return failed images = (data.get("images") or data.get("data")) if tool_name == "browser_get_images" else None if isinstance(images, list): @@ -624,19 +560,17 @@ def _format_browser_result(tool_name: str, data: Args, args: Args) -> Optional[s lines.append(f"- {str(img.get('alt') or '').strip() or 'image'}" + (f" — {url}" if url else "")) return _truncate_text("\n".join(lines), limit=5000) title = str(_first(data, "title", "url", "status", default=tool_name)) - text = str(_first(data, "text", "content", "snapshot", "analysis", "message")).strip() lines = [title] if data.get("url") and data.get("url") != title: lines.append(str(data.get("url"))) - if text: + if text := str(_first(data, "text", "content", "snapshot", "analysis", "message")).strip(): lines.extend(["", _truncate_text(text, limit=5000)]) return _truncate_text("\n".join(lines), limit=7000) @_structured(text_fallback=True) def _format_media_or_cron_result(tool_name: str, data: Args, args: Args) -> Optional[str]: - failed = _failure(data, f"{tool_name} failed") - if failed: + if failed := _failure(data, f"{tool_name} failed"): return failed keys = ("file_path", "path", "url", "image_url", "job_id", "id", "status", "message", "next_run") return "\n".join([f"✅ {tool_name} completed", *(f"- **{k}:** {data.get(k)}" for k in keys if data.get(k))]) @@ -652,9 +586,8 @@ def _format_structured_value(key: str, value: Any, *, indent: int = 0, max_depth return f"{bullet}{label} {text}" if label else f"{bullet}{text}" def _child(child_key: str, child_value: Any, extra_indent: int) -> List[str]: - return _format_structured_value( - child_key, child_value, indent=indent + extra_indent, max_depth=max_depth - 1, max_items=max_items, - ) + return _format_structured_value(child_key, child_value, indent=indent + extra_indent, max_depth=max_depth - 1, + max_items=max_items) if value in _EMPTYISH: return [] @@ -720,12 +653,9 @@ def _format_generic_structured_result(tool_name: str, result: Optional[str], *, lines.extend(_format_structured_value("", item, indent=0, max_depth=2, max_items=6)) else: lines.append(f"- {_truncate_text(str(item), limit=240)}") - if len(data) > 12: - lines.append(f"... {len(data) - 12} more items") - return _truncate_text("\n".join(lines), limit=5000) + return _truncate_text("\n".join(lines + _more(data, 12, " items")), limit=5000) - failed = _failure(data, f"{tool_name} failed") - if failed: + if failed := _failure(data, f"{tool_name} failed"): return failed lines = [f"✅ {tool_name} completed" if data.get("success") is True else f"{tool_name} result"] seen = {key for key in _PRIORITY_KEYS if data.get(key) not in _EMPTYISH} @@ -737,8 +667,7 @@ def _format_generic_structured_result(tool_name: str, result: Optional[str], *, if len(lines) >= 40: lines.append("- ... more fields truncated") break - content = data.get("content") - if isinstance(content, str) and content.strip(): + if isinstance(content := data.get("content"), str) and content.strip(): lines.extend(["", _truncate_text(content.strip(), limit=1500)]) return _truncate_text("\n".join(lines), limit=7000) @@ -762,33 +691,21 @@ _COMPLETION_FORMATTERS: Dict[str, _Formatter] = { } -def _build_polished_completion_content(tool_name: str, result: Optional[str], function_args: Optional[Args]) -> Optional[List[Any]]: - formatter = _COMPLETION_FORMATTERS.get(tool_name) - if formatter is not None: - text = formatter(tool_name, result, function_args) - else: - text = _format_generic_structured_result(tool_name, result, fallback_to_text=tool_name in _POLISHED_TOOLS) - return [_text(text)] if text else None - - def _parse_unified_diff_content(diff_text: str) -> List[Any]: """Convert unified diff text into ACP diff content blocks (one per ``---``/``+++`` pair).""" content: List[Any] = [] - if not diff_text: - return content state: Dict[str, Any] = {"old": None, "new": None, "old_lines": [], "new_lines": []} def _flush() -> None: old_path, new_path = state["old"], state["new"] - if old_path is not None or new_path is not None: - path = new_path if new_path and new_path != "/dev/null" else old_path - if path and path != "/dev/null": - path = str(path).strip() - content.append(acp.tool_diff_content( - path=path[2:] if path.startswith(("a/", "b/")) else path, - old_text="\n".join(state["old_lines"]) if state["old_lines"] else None, - new_text="\n".join(state["new_lines"]), - )) + path = new_path if new_path and new_path != "/dev/null" else old_path + if path and path != "/dev/null": + path = str(path).strip() + content.append(acp.tool_diff_content( + path=path[2:] if path.startswith(("a/", "b/")) else path, + old_text="\n".join(state["old_lines"]) if state["old_lines"] else None, + new_text="\n".join(state["new_lines"]), + )) state.update(old=None, new=None, old_lines=[], new_lines=[]) for line in diff_text.splitlines(): @@ -825,21 +742,27 @@ def _build_tool_complete_content( return diff_content except Exception: pass - return _build_polished_completion_content(tool_name, result, function_args) or [_text(_truncate_text(result or ""))] + if (formatter := _COMPLETION_FORMATTERS.get(tool_name)) is not None: + text = formatter(tool_name, result, function_args) + else: + text = _format_generic_structured_result(tool_name, result, fallback_to_text=tool_name in _POLISHED_TOOLS) + return [_text(text)] if text else [_text(_truncate_text(result or ""))] # --- ToolCallStart / ToolCallProgress events --------------------------------- +def _more(items: list, shown: int, unit: str = "") -> List[str]: + """``["... N more"]`` trailer when ``items`` overflowed the ``shown`` cap, else ``[]``.""" + return [f"... {len(items) - shown} more{unit}"] if len(items) > shown else [] + + def _start_todo(args: Args) -> str: - items = args.get("todos") - if not isinstance(items, list): + if not isinstance(items := args.get("todos"), list): return "Reading todo list" lines = ["Updating todo list", ""] lines.extend(f"- {i.get('status', 'pending')}: {i.get('content', i.get('id', ''))}" for i in items[:8] if isinstance(i, dict)) - if len(items) > 8: - lines.append(f"... {len(items) - 8} more") - return "\n".join(lines) + return "\n".join(lines + _more(items, 8)) def _start_skill_manage(args: Args) -> Any: @@ -867,16 +790,13 @@ def _start_execute_code(args: Args) -> str: def _start_delegate(args: Args) -> str: - tasks = args.get("tasks") - if not (isinstance(tasks, list) and tasks): + if not (isinstance(tasks := args.get("tasks"), list) and tasks): return "Delegating task" + _fmt(_truncate_text(_arg(args, "goal"), limit=800), ":\n{}", "") lines = [f"Delegating {len(tasks)} tasks", ""] for i, task in enumerate(tasks[:8], 1): if isinstance(task, dict): lines.append(f"{i}. " + _truncate_text(_arg(task, "goal"), limit=160) + _fmt(_arg(task, "role"), " ({})", "")) - if len(tasks) > 8: - lines.append(f"... {len(tasks) - 8} more") - return "\n".join(lines) + return "\n".join(lines + _more(tasks, 8)) def _preview(label: str, value: str, limit: int) -> str: @@ -934,8 +854,7 @@ def _build_tool_start(tool_call_id: str, tool_name: str, arguments: Args, *, edi if tool_name in ("patch", "write_file") and edit_diff is not None: content = [acp.tool_diff_content(path=edit_diff.path, old_text=edit_diff.old_text, new_text=edit_diff.new_text)] elif tool_name in _START_CONTENT_BUILDERS: - builder = _START_CONTENT_BUILDERS[tool_name] - built = builder(arguments) if builder is not None else None + built = builder(arguments) if (builder := _START_CONTENT_BUILDERS[tool_name]) is not None else None content = None if built is None else [_text(built) if isinstance(built, str) else built] elif tool_name in _POLISHED_TOOLS: content = [_text(_truncate_text(_args_json(arguments), limit=1200))] @@ -970,7 +889,6 @@ def build_tool_complete( def extract_locations(arguments: Args) -> List[ToolCallLocation]: """Extract file-system locations from tool arguments.""" - path = arguments.get("path") - if not path: + if not (path := arguments.get("path")): return [] return [ToolCallLocation(path=path, line=arguments.get("offset") or arguments.get("line"))]