From 96637366aaba4846d3fb85c812613fb0f37aeb83 Mon Sep 17 00:00:00 2001 From: Timothy Jaeryang Baek Date: Thu, 23 Jul 2026 17:29:14 -0400 Subject: [PATCH 1/3] refac --- cptr/utils/ai.py | 30 ++- cptr/utils/chat_task.py | 494 +++++++++++++--------------------------- 2 files changed, 183 insertions(+), 341 deletions(-) diff --git a/cptr/utils/ai.py b/cptr/utils/ai.py index 60d80a75..5b822f1a 100644 --- a/cptr/utils/ai.py +++ b/cptr/utils/ai.py @@ -259,7 +259,10 @@ def _to_anthropic_messages(messages: list[dict]) -> list[dict]: # assistant with tool calls → Anthropic tool_use blocks blocks: list[dict] = [] if content: - blocks.append({"type": "text", "text": content}) + if isinstance(content, list): + blocks.extend(block for block in content if block.get("type") == "text") + else: + blocks.append({"type": "text", "text": content}) for tc in m["tool_calls"]: blocks.append( { @@ -422,8 +425,8 @@ def _to_openai_messages( Strict default OpenAI-compatible requests do not receive provider-specific reasoning fields. llama.cpp compatibility replays text reasoning as - reasoning_content. Other non-standard fields (fc_id in tool_calls) are - stripped. + reasoning_content and preserves assistant text before tool calls. Other + non-standard fields (fc_id in tool_calls) are stripped. """ result = [] if instructions: @@ -433,6 +436,8 @@ def _to_openai_messages( continue content = m.get("content", "") + if provider_type != "llama.cpp" and m.get("role") == "assistant" and m.get("tool_calls"): + content = "" if isinstance(content, list): formatted_content = [] for block in content: @@ -461,6 +466,7 @@ def _to_openai_messages( else: out = dict(m) out.pop("reasoning_items", None) + out["content"] = content if out.get("tool_calls"): out["tool_calls"] = [ {k: v for k, v in tc.items() if k != "fc_id"} for tc in out["tool_calls"] @@ -726,6 +732,24 @@ def _to_responses_input( m.get("reasoning_items"), provider_type=provider_type ): items.append(ri) + content = m.get("content", "") + if isinstance(content, list): + text = "".join( + block.get("text") or "" + for block in content + if isinstance(block, dict) + and block.get("type") in ("text", "output_text", "input_text") + ) + else: + text = content + if isinstance(text, str) and text.strip(): + items.append( + { + "type": "message", + "role": "assistant", + "content": [{"type": "output_text", "text": text}], + } + ) for tc in m["tool_calls"]: call_id = tc.get("id", "") # Skip function_calls that have no matching tool result diff --git a/cptr/utils/chat_task.py b/cptr/utils/chat_task.py index f27cf299..adc57bdd 100644 --- a/cptr/utils/chat_task.py +++ b/cptr/utils/chat_task.py @@ -785,6 +785,115 @@ async def generate_chat_title( # ── Message history ───────────────────────────────────────── +def _output_items_to_messages(output_items: list[dict], message_id: str | None = None) -> list[dict]: + """Convert ordered persisted output items into model-visible messages.""" + native_agent_call_ids = { + item["call_id"] + for item in output_items + if item.get("type") == "function_call" + and item.get("call_id") + and _is_native_agent_tool_item(item) + } + output_call_ids = { + item["call_id"] + for item in output_items + if item.get("type") == "function_call_output" + and item.get("call_id") not in native_agent_call_ids + } + messages: list[dict] = [] + pending_content: list[str] = [] + pending_reasoning_items: list[dict] = [] + pending_tool_calls: list[dict] = [] + call_names: dict[str, str] = {} + + def flush_pending() -> None: + nonlocal pending_content, pending_reasoning_items, pending_tool_calls + if not pending_content and not pending_reasoning_items and not pending_tool_calls: + return + assistant_msg: dict = { + "role": "assistant", + "content": "".join(pending_content), + } + if message_id: + assistant_msg["id"] = message_id + if pending_tool_calls: + assistant_msg["tool_calls"] = pending_tool_calls + if pending_reasoning_items: + assistant_msg["reasoning_items"] = pending_reasoning_items + messages.append(assistant_msg) + pending_content = [] + pending_reasoning_items = [] + pending_tool_calls = [] + + for item in output_items: + itype = item.get("type") + if itype == "message": + text = "".join( + block.get("text") or "" + for block in item.get("content") or [] + if isinstance(block, dict) and block.get("type") in ("text", "output_text") + ) + if text: + pending_content.append(text) + elif itype == "reasoning": + if ( + item.get("status") not in (None, "completed") + or str(item.get("id", "")).startswith("reasoning-") + or ( + not item.get("encrypted_content") + and not item.get("reasoning_details") + and _reasoning_text_len(item) <= 0 + ) + ): + continue + pending_reasoning_items.append(item) + elif itype == "function_call" and item.get("status") == "completed": + call_id = item.get("call_id") + if not call_id or call_id in native_agent_call_ids: + continue + if call_id not in output_call_ids: + logger.warning( + "[history] Skipping orphaned function_call %s (%s) — no matching output", + call_id, + item.get("name", "?"), + ) + continue + arguments = item.get("arguments", {}) + if not isinstance(arguments, str): + arguments = json.dumps(arguments) + tc = { + "id": call_id, + "type": "function", + "function": { + "name": item.get("name", ""), + "arguments": arguments, + }, + } + if item.get("fc_id"): + tc["fc_id"] = item["fc_id"] + call_names[call_id] = item.get("name", "") + pending_tool_calls.append(tc) + elif itype == "function_call_output" and item.get("call_id") not in native_agent_call_ids: + call_id = item.get("call_id") + if call_id not in call_names: + continue + flush_pending() + tool_msg = { + "role": "tool", + "tool_call_id": call_id, + "content": _tool_result_for_model( + call_names.get(call_id, ""), + item.get("output", ""), + ), + } + if message_id: + tool_msg["id"] = message_id + messages.append(tool_msg) + + flush_pending() + return messages + + async def _load_message_history(chat_id: str, message_id: str) -> tuple[list[dict], str | None]: """Load the ancestor chain from message_id to root as LLM messages. @@ -880,151 +989,14 @@ async def _load_message_history(chat_id: str, message_id: str) -> tuple[list[dic elif text_content != entry["content"]: entry["content"] = text_content - # Reconstruct tool calls from output items for the provider. - # - # Output items accumulate across agentic-loop iterations within a - # single assistant message. The Responses API (reasoning models) - # requires that each function_call is paired with the reasoning - # items from the *same* API response. To reconstruct the correct - # interleaved history we group output items into "turns": - # - # Turn = [reasoning…, function_call…, function_call_output…] - # - # Each turn produces one assistant message (with tool_calls + - # reasoning_items) followed by the corresponding tool-result - # messages. + # Reconstruct model-visible messages from the ordered output stream. + # `m.output` is the source of truth: assistant text/reasoning, + # function_call, function_call_output, then possibly more text. if m.output: - native_agent_call_ids = { - item["call_id"] - for item in m.output - if item.get("type") == "function_call" - and item.get("call_id") - and _is_native_agent_tool_item(item) - } - # ── Collect the set of call_ids that have outputs ── - # This is needed to filter out orphaned function_calls - # (e.g. from crashes, partial persistence, or data corruption). - output_call_ids = { - item["call_id"] - for item in m.output - if item.get("type") == "function_call_output" - and item.get("call_id") not in native_agent_call_ids - } - - # ── Group output items into per-iteration turns ── - turns: list[dict] = [] # each: {reasoning: [], calls: [], outputs: []} - current_turn: dict = {"reasoning": [], "calls": [], "outputs": []} - - for item in m.output: - itype = item.get("type") - if itype == "reasoning": - if ( - item.get("status") not in (None, "completed") - or str(item.get("id", "")).startswith("reasoning-") - or ( - not item.get("encrypted_content") - and not item.get("reasoning_details") - and _reasoning_text_len(item) <= 0 - ) - ): - continue - # A reasoning item after we already have outputs means - # a new API iteration started — flush the current turn. - if current_turn["outputs"]: - turns.append(current_turn) - current_turn = {"reasoning": [], "calls": [], "outputs": []} - current_turn["reasoning"].append(item) - elif itype == "function_call" and item.get("status") == "completed": - if item.get("call_id") in native_agent_call_ids: - continue - # Only include calls that have a matching output - if item["call_id"] not in output_call_ids: - logger.warning( - "[history] Skipping orphaned function_call %s (%s) — no matching output", - item.get("call_id", "?"), - item.get("name", "?"), - ) - continue - # A new function_call after outputs means - # a new iteration (model saw results and called again). - if current_turn["outputs"]: - turns.append(current_turn) - current_turn = {"reasoning": [], "calls": [], "outputs": []} - tc = { - "id": item["call_id"], - "type": "function", - "function": { - "name": item["name"], - "arguments": json.dumps(item.get("arguments", {})), - }, - } - # Preserve Responses API fc_ ID for round-tripping - if item.get("fc_id"): - tc["fc_id"] = item["fc_id"] - current_turn["calls"].append(tc) - elif ( - itype == "function_call_output" - and item.get("call_id") not in native_agent_call_ids - ): - current_turn["outputs"].append(item) - # Skip other types (message, artifact, pending calls, etc.) - - # Don't forget the last turn - if current_turn["reasoning"] or current_turn["calls"] or current_turn["outputs"]: - turns.append(current_turn) - - if not turns: - # No tool calls — keep entry as-is (plain text assistant) - pass - else: - for ti, turn in enumerate(turns): - # Filter calls to only those with matching outputs - turn_output_ids = {o["call_id"] for o in turn["outputs"]} - matched_calls = [tc for tc in turn["calls"] if tc["id"] in turn_output_ids] - call_names = { - tc["id"]: tc.get("function", {}).get("name", "") for tc in turn["calls"] - } - if turn["reasoning"] and not matched_calls and not turn["outputs"]: - if ti == 0: - entry["reasoning_items"] = turn["reasoning"] - else: - result.append(entry) - entry = { - "id": m.id, - "role": "assistant", - "content": "", - "reasoning_items": turn["reasoning"], - } - continue - if matched_calls: - if ti == 0: - # First turn: attach to the existing entry - entry["tool_calls"] = matched_calls - if turn["reasoning"]: - entry["reasoning_items"] = turn["reasoning"] - else: - # Subsequent turns: flush the pending entry (last tool - # result from previous turn) before creating a new one. - result.append(entry) - entry = { - "id": m.id, - "role": "assistant", - "content": "", - "tool_calls": matched_calls, - } - if turn["reasoning"]: - entry["reasoning_items"] = turn["reasoning"] - for out in turn["outputs"]: - result.append(entry) - entry = { - "id": m.id, - "role": "tool", - "tool_call_id": out["call_id"], - "content": _tool_result_for_model( - call_names.get(out["call_id"], ""), - out.get("output", ""), - ), - } + output_messages = _output_items_to_messages(m.output, m.id) + if output_messages: + result.extend(output_messages) + continue result.append(entry) @@ -1155,151 +1127,6 @@ def _tool_result_for_model(tool_name: str, result: str) -> str: } ) -def _append_tool_to_messages( - messages: list[dict], - event: dict, - result: str, - provider: str, - reasoning_items: list[dict] | None = None, -): - """Append a tool call + result to the message history for the next API call.""" - result = _tool_result_for_model(event["name"], result) - - # Check for image result before truncation (data URI is large but needed) - image = _parse_image_data_uri(result) - - if not image: - # Guard against oversized tool outputs (skip for images, handled above) - if len(result) > CHAT_TOOL_MAX_CHARS: - half = CHAT_TOOL_MAX_CHARS // 2 - result = result[:half] + "\n\n...(truncated)...\n\n" + result[-half:] - - # Add assistant message with tool_call - assistant_msg = { - "role": "assistant", - "content": "", - "tool_calls": [ - { - "id": event["call_id"], - "fc_id": event.get("id", ""), - "type": "function", - "function": { - "name": event["name"], - "arguments": json.dumps(event["arguments"]), - }, - } - ], - } - # Attach reasoning items for round-tripping (Responses API reasoning models) - if reasoning_items: - assistant_msg["reasoning_items"] = reasoning_items - messages.append(assistant_msg) - - if image: - # Structured multimodal content — provider converters handle the - # "image" block type appropriately for each API. - media_type, b64_data = image - path = event["arguments"].get("path", "image") - messages.append( - { - "role": "tool", - "tool_call_id": event["call_id"], - "content": [ - {"type": "text", "text": f"Image file: {path}"}, - { - "type": "image", - "media_type": media_type, - "base64": b64_data, - }, - ], - } - ) - else: - # Plain text tool result - messages.append( - { - "role": "tool", - "tool_call_id": event["call_id"], - "content": result, - } - ) - - -def _append_batch_to_messages( - messages: list[dict], - call_results: list[tuple[dict, str]], - provider: str, - reasoning_items: list[dict] | None = None, -): - """Append multiple tool calls + results as a single assistant message. - - Unlike _append_tool_to_messages (which creates one assistant message per - call), this batches all calls into one assistant message. This is - required for the Responses API with reasoning models, where all - function_calls from one API response share the same reasoning items. - """ - if not call_results: - return - - tool_calls = [] - for event, _ in call_results: - tool_calls.append( - { - "id": event["call_id"], - "fc_id": event.get("id", ""), - "type": "function", - "function": { - "name": event["name"], - "arguments": json.dumps(event["arguments"]), - }, - } - ) - - assistant_msg: dict = { - "role": "assistant", - "content": "", - "tool_calls": tool_calls, - } - if reasoning_items: - assistant_msg["reasoning_items"] = reasoning_items - messages.append(assistant_msg) - - for event, result in call_results: - result = _tool_result_for_model(event["name"], result) - - # Guard against oversized tool outputs - image = _parse_image_data_uri(result) - if not image: - if len(result) > CHAT_TOOL_MAX_CHARS: - half = CHAT_TOOL_MAX_CHARS // 2 - result = result[:half] + "\n\n...(truncated)...\n\n" + result[-half:] - - if image: - media_type, b64_data = image - path = event["arguments"].get("path", "image") - messages.append( - { - "role": "tool", - "tool_call_id": event["call_id"], - "content": [ - {"type": "text", "text": f"Image file: {path}"}, - { - "type": "image", - "media_type": media_type, - "base64": b64_data, - }, - ], - } - ) - else: - messages.append( - { - "role": "tool", - "tool_call_id": event["call_id"], - "content": result, - } - ) - def _scrub_incomplete_items(output_items: list[dict]) -> None: """Mark any in-progress function_call items as 'failed'. @@ -1951,6 +1778,7 @@ async def _finish_reasoning_item(): provider = connection["provider"] api_key = decrypt_key(connection.get("api_key", ""), _get_jwt_secret()) base_url = connection.get("base_url") or _default_base_url(provider) + api_type = connection.get("api_type", "chat_completions") chat_obj = await Chat.get_by_id(chat_id) chat_params = (chat_obj.meta or {}).get("params", {}) if chat_obj else {} @@ -1960,6 +1788,11 @@ async def _finish_reasoning_item(): except Exception: chat_models_config = {} builtin_tools = resolve_builtin_tools_config(chat_models_config, model, configured_model) + provider_type = ( + "llama.cpp" + if provider == "openai" and connection.get("provider_type") == "llama.cpp" + else "default" + ) messages, loaded_summary = await _load_message_history(chat_id, message_id) skill_settings = await get_skill_settings() skill_authoring_allowed = _has_prior_real_chat_content(messages, loaded_summary) @@ -2101,7 +1934,6 @@ async def _finish_reasoning_item(): drop_zone = messages[:split_idx] keep_zone = messages[split_idx:] - api_type = connection.get("api_type", "chat_completions") summary = await summarize_messages( drop_zone, loaded_summary, @@ -2158,8 +1990,8 @@ async def _finish_reasoning_item(): api_messages = messages if provider != "anthropic": image_blocks = [] - api_messages = [] - for m in messages: + normalized_messages = [] + for m in api_messages: if m.get("role") == "tool" and isinstance(m.get("content"), list): text_parts = [] for part in m["content"]: @@ -2167,9 +1999,10 @@ async def _finish_reasoning_item(): text_parts.append(part.get("text", "")) elif part.get("type") == "image": image_blocks.append(part) - api_messages.append({**m, "content": "\n".join(text_parts)}) + normalized_messages.append({**m, "content": "\n".join(text_parts)}) else: - api_messages.append(m) + normalized_messages.append(m) + api_messages = normalized_messages if image_blocks: api_messages.append( { @@ -2191,17 +2024,11 @@ async def _finish_reasoning_item(): instructions=system, tools=tools, ) - provider_type = ( - "llama.cpp" - if provider == "openai" and connection.get("provider_type") == "llama.cpp" - else "default" - ) - if provider == "anthropic": stream = stream_anthropic( form_data, base_url, api_key, request_params=request_params ) - elif connection.get("api_type") == "responses": + elif api_type == "responses": stream = stream_openai_responses( form_data, base_url, @@ -2243,7 +2070,9 @@ async def _finish_reasoning_item(): text_buffer += event["content"] await emit(delta=event["content"]) if provider_type == "llama.cpp" and usage_context_tokens(last_usage) <= 0: - estimated_context_tokens = estimated_prompt_tokens + estimate_tokens(content) + estimated_context_tokens = estimated_prompt_tokens + estimate_tokens( + content + ) if estimated_context_tokens - last_emitted_context_tokens >= 256: last_emitted_context_tokens = estimated_context_tokens await emit( @@ -2404,13 +2233,16 @@ async def _finish_reasoning_item(): await emit(output=item) await emit(output=result_item) _sync_state() - _append_batch_to_messages( - messages, - [(tc, error)], - provider, - reasoning_items=response_reasoning_items, + new_messages = _output_items_to_messages( + [ + *response_reasoning_items, + *([flushed_item] if flushed_item else []), + item, + result_item, + ] ) - new_messages_since += 2 + messages.extend(new_messages) + new_messages_since += len(new_messages) await _save_message( "invalid ask_user", content=content, output=output_items ) @@ -2536,7 +2368,7 @@ async def auto_answer(): # Execute non-delegate tools sequentially first # Collect results for batch message construction - sequential_results: list[tuple[dict, str]] = [] # (event, result) + sequential_result_items: list[dict] = [] for idx in other_indices: tc, item = call_items[idx] if tc["name"] == "create_artifact": @@ -2564,6 +2396,7 @@ async def auto_answer(): await emit(output=item) await emit(output=result_item) _sync_state() + sequential_result_items.append(result_item) artifact_item = build_artifact_item(tc["name"], tc["arguments"], result) if artifact_item: @@ -2580,21 +2413,8 @@ async def auto_answer(): output_items.append(file_item) await emit(output=file_item) _sync_state() - - sequential_results.append((tc, result)) - - # Build a single combined assistant message for all sequential calls - # with their shared reasoning items (required for reasoning model round-tripping) - if sequential_results: - _append_batch_to_messages( - messages, - sequential_results, - provider, - reasoning_items=response_reasoning_items, - ) - new_messages_since += 1 + len(sequential_results) - # Execute delegate_task calls concurrently, emit each as it completes + delegate_result_items: list[dict] = [] if delegate_indices: # Create tasks, mapping task → index inflight: dict[asyncio.Task, int] = {} @@ -2608,8 +2428,6 @@ async def auto_answer(): ) ) inflight[task] = idx - - delegate_results: list[tuple[dict, str]] = [] while inflight: done_set, _ = await asyncio.wait( inflight.keys(), return_when=asyncio.FIRST_COMPLETED @@ -2634,19 +2452,19 @@ async def auto_answer(): await emit(output=item) await emit(output=result_item) _sync_state() - delegate_results.append((tc, result)) - - # Build combined message for all delegate calls - if delegate_results: - # Only attach reasoning if sequential calls didn't consume it - ri = response_reasoning_items if not other_indices else None - _append_batch_to_messages( - messages, - delegate_results, - provider, - reasoning_items=ri, - ) - new_messages_since += 1 + len(delegate_results) + delegate_result_items.append(result_item) + + new_messages = _output_items_to_messages( + [ + *response_reasoning_items, + *([flushed_item] if flushed_item else []), + *(item for _, item in call_items if item.get("status") == "completed"), + *sequential_result_items, + *delegate_result_items, + ] + ) + messages.extend(new_messages) + new_messages_since += len(new_messages) # Persist after all tool calls await _save_message("tool calls complete", content=content, output=output_items) From e42f95f821a6977ce649e1aab4fa58a53c2bf07c Mon Sep 17 00:00:00 2001 From: Timothy Jaeryang Baek Date: Thu, 23 Jul 2026 17:32:38 -0400 Subject: [PATCH 2/3] refac --- cptr/utils/ai.py | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) diff --git a/cptr/utils/ai.py b/cptr/utils/ai.py index 5b822f1a..b75733cd 100644 --- a/cptr/utils/ai.py +++ b/cptr/utils/ai.py @@ -425,8 +425,8 @@ def _to_openai_messages( Strict default OpenAI-compatible requests do not receive provider-specific reasoning fields. llama.cpp compatibility replays text reasoning as - reasoning_content and preserves assistant text before tool calls. Other - non-standard fields (fc_id in tool_calls) are stripped. + reasoning_content. Other non-standard fields (fc_id in tool_calls) are + stripped. """ result = [] if instructions: @@ -436,8 +436,6 @@ def _to_openai_messages( continue content = m.get("content", "") - if provider_type != "llama.cpp" and m.get("role") == "assistant" and m.get("tool_calls"): - content = "" if isinstance(content, list): formatted_content = [] for block in content: From 7fb978de1c87b90a461b581747a87653482c5a52 Mon Sep 17 00:00:00 2001 From: Timothy Jaeryang Baek Date: Thu, 23 Jul 2026 17:32:48 -0400 Subject: [PATCH 3/3] refac --- CHANGELOG.md | 6 ++++++ pyproject.toml | 2 +- uv.lock | 2 +- 3 files changed, 8 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 7ada93d6..1b8ba394 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,12 @@ All notable changes to this project will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [0.9.14] - 2026-07-23 + +### Changed + +- 🔧 **Chat handling got a cleanup.** Computer now carries conversations forward more cleanly after some internal refactoring. + ## [0.9.13] - 2026-07-23 ### Fixed diff --git a/pyproject.toml b/pyproject.toml index 50d103c0..8e70247c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "cptr" -version = "0.9.13" +version = "0.9.14" description = "Your computer, from anywhere. Code, manage, and control your machine from the web." license = {file = "LICENSE"} readme = "README.md" diff --git a/uv.lock b/uv.lock index 6e9a9905..103af922 100644 --- a/uv.lock +++ b/uv.lock @@ -284,7 +284,7 @@ wheels = [ [[package]] name = "cptr" -version = "0.9.13" +version = "0.9.14" source = { editable = "." } dependencies = [ { name = "aiosqlite" },